From 6b0111a68013a3e5abde53218df2aa2504fca27e Mon Sep 17 00:00:00 2001
From: Day Dreamer <87640074+HugoXK@users.noreply.github.com>
Date: Tue, 27 Oct 2026 22:17:51 +0000
Subject: [PATCH] fix(parts): archive a job-held part once its job ends

Fixes #1433
Owner: Dev2
Reviewer: Reviewer
Written-by: Rework

Replayed onto 55e1c27a09a (main). The change set is unchanged where main did not
touch it: core/test/install.test.js, core/test/part-archive.test.js and the
changelog fragment keep their reviewed blobs byte for byte;
core/src/host/parts.js and docs/subsystems/core-host.md were merged with main's
#1500 (steward retry) edits, and the two generated ds-bot patch mirrors were
rebuilt from the merged core.

Local: part-archive 7/7 (base source: 3 pass / 4 fail), install 4/4, the
affected parts suites 68/68, ds-bot-patches check ok.
---
 changelog.d/1433-job-part-archive.fixed.md    |   1 +
 core/src/host/parts.js                        |  95 ++++++++++++-
 core/test/install.test.js                     |   1 +
 core/test/part-archive.test.js                | 112 ++++++++++++++-
 docs/subsystems/core-host.md                  |   2 +-
 patches/ds-bot/0001-core-fixes.patch          |   8 ++
 patches/ds-bot/0014-flybot-branding.patch     | 132 ++++++++++++++++--
 .../ds-bot/files/test/part-archive.test.js    | 112 ++++++++++++++-
 8 files changed, 441 insertions(+), 22 deletions(-)
 create mode 100644 changelog.d/1433-job-part-archive.fixed.md

diff --git a/changelog.d/1433-job-part-archive.fixed.md b/changelog.d/1433-job-part-archive.fixed.md
new file mode 100644
index 00000000..b1f5d892
--- /dev/null
+++ b/changelog.d/1433-job-part-archive.fixed.md
@@ -0,0 +1 @@
+A conversation part that a job still held when the Bot moved on to a fresh part is archived once that job ends, instead of staying active for the life of the instance. Nothing is force-stopped: a job, turn or subagent is left to finish, and the part is asked for once more, with no options, when its holder is reported ended — the job registry's own settlement, or the Session's next idle — keeping the id while a job, turn or subagent still holds it. A scheduled task keeps the existing stop-then-archive path ([#896](https://github.com/ThinkFlowLab/FlyBot/issues/896), [#1433](https://github.com/ThinkFlowLab/FlyBot/issues/1433)).
diff --git a/core/src/host/parts.js b/core/src/host/parts.js
index 542f6660..06dd0139 100644
--- a/core/src/host/parts.js
+++ b/core/src/host/parts.js
@@ -796,6 +796,21 @@ export function install(rt) {
   const noteFailures = new Map()
   // Parts the steward is deciding about at the end of a turn.
   const checking = new Set()
+  // Earlier parts whose handoff archive a holder that ends on its own refused: a job, a
+  // turn or a subagent, never a scheduled task. Such a holder is never stopped, so the id
+  // waits here until that end is reported — its job settling, or its Session going idle —
+  // and the archive is asked for once more, with no options (#1433). A refusal that names
+  // no kind at all is not remembered: nothing says what would end.
+  const pendingArchive = new Set()
+  // Ids whose one more attempt is in flight, so a job that settles just as its Session
+  // goes idle is asked for once, not twice.
+  const attemptingArchive = new Set()
+  // The most ids that may wait: a Set keeps insertion order, so the oldest leaves first.
+  // Nothing here retries on a timer — one attempt per id at a time, and only when a
+  // holder-end signal names it — so the table cannot grow by itself. The cap is the
+  // backstop for a host that never reports the end: such a part is dropped, never forced
+  // (#896).
+  const PENDING_ARCHIVE_MAX = 32
 
   const placeOf = (sessionId) => {
     const botId = rt.chatOwners.get(sessionId)
@@ -884,7 +899,14 @@ export function install(rt) {
           // refusal a scheduled task alone caused (#896 review): a job, turn or subagent
           // that holds the part is left to finish, exactly as before this fix.
           const kinds = Array.isArray(error?.activity) ? error.activity.map(entry => entry?.kind) : []
-          if (kinds.length === 0 || kinds.some(kind => kind !== 'schedule')) return
+          if (kinds.length === 0 || kinds.some(kind => kind !== 'schedule')) {
+            // A job, turn or subagent that holds the part is left to finish, and the part
+            // used to stay active for the life of the instance (#1433). Remember it for
+            // one more attempt once its holder ends; a schedule-only refusal takes the
+            // stop below.
+            if (kinds.length > 0) rememberPendingArchive(agent.id)
+            return
+          }
           // A task that could not be moved still holds the part, and leaving it active
           // forever is the leak #896 is about. Stop its activity, the way a deletion does
           // (roster.js:450): a finished part is never left running a task nobody sees.
@@ -1188,11 +1210,74 @@ export function install(rt) {
     if (end !== undefined && end.data?.reason?.kind !== 'error') rescued.delete(agent.id)
   }
 
+  function rememberPendingArchive(sessionId) {
+    pendingArchive.add(sessionId)
+    while (pendingArchive.size > PENDING_ARCHIVE_MAX) pendingArchive.delete(pendingArchive.values().next().value)
+    watchJobSettlements()
+  }
+
+  // A background job ends on its own schedule, and its Session is idle by then: the idle
+  // event came before this attempt and will not come again, and the part that was handed
+  // over never takes another turn (its input goes to the fresh part, `forward`). The job
+  // registry's own settlement is the event that names that moment (#1433), read through
+  // `ctx.get` — no inject asked, and `undefined` when the service is not there yet: a
+  // composition may provide the job registry after this plugin loads, and one without it
+  // has no job to wait for.
+  let jobUnwatch
+  function watchJobSettlements() {
+    if (jobUnwatch !== undefined) return
+    const events = ctx.get('jobs')?.events
+    if (events === undefined) return
+    jobUnwatch = events.subscribe({ owners: 'all' }, (event) => {
+      if (event?.type !== 'settled' || typeof event.job?.owner !== 'string') return
+      attemptPendingArchive(event.job.owner, 'its job')
+    })
+  }
+  ctx.effect(() => () => jobUnwatch?.(), 'ds-bot: archive a part once its job ends')
+
+  // One more archive attempt for a part whose holder has ended, with no options, so a job,
+  // a turn or a subagent is never stopped for it (#896). A refusal that still names one of
+  // those means the holder has not ended after all — the Session reported its idle while
+  // the job ran — so the id stays and the next signal tries again instead of the one
+  // chance being spent. It leaves when the attempt lands, when the refusal names no
+  // activity at all, or when a scheduled task alone names it — that one is answered by the
+  // handoff's own `stopActivity` retry, never by holding the id.
+  function attemptPendingArchive(sessionId, endedBy) {
+    if (!pendingArchive.has(sessionId) || attemptingArchive.has(sessionId)) return
+    attemptingArchive.add(sessionId)
+    void ctx.workspaceRegistry.archiveSession(sessionId)
+      .then(() => pendingArchive.delete(sessionId))
+      .catch(error => {
+        const kinds = Array.isArray(error?.activity) ? error.activity.map(entry => entry?.kind) : []
+        // A holder that ends on its own has not ended after all: keep the id and let the
+        // next signal try again, instead of spending the one chance.
+        if (kinds.some(kind => kind !== 'schedule')) {
+          rt.warn('ds-bot: earlier part %s still held by its %s after %s ended: %s', sessionId, kinds.join(', '), endedBy, error)
+          return
+        }
+        pendingArchive.delete(sessionId)
+        // A scheduled task alone is the one refusal the handoff answers by stopping the
+        // activity (#896): a task that could not be moved still holds the part, and leaving
+        // it active is the leak itself. The retry answers it the same way, and the id is
+        // spent either way — a stop that fails is not asked for again.
+        if (kinds.length > 0) {
+          void ctx.workspaceRegistry.archiveSession(sessionId, { stopActivity: true })
+            .catch(again => rt.warn('ds-bot: earlier part %s not archived after %s ended, not even stopped: %s', sessionId, endedBy, again))
+          return
+        }
+        rt.warn('ds-bot: earlier part %s not archived after %s ended: %s', sessionId, endedBy, error)
+      })
+      .finally(() => attemptingArchive.delete(sessionId))
+  }
+
   ctx.on('agent/status', ({ agent, status }) => {
-    if (status === 'idle' && rt.isBotSession(agent.id)) {
-      releaseRescued(agent)
-      void checkContext(agent).catch(error => rt.warn('ds-bot: context check for %s failed: %s', agent.id, error))
-    }
+    if (status !== 'idle') return
+    // Before the Bot check: the part that was handed over is no longer this Bot's current
+    // Session, and its holder ends in it all the same.
+    attemptPendingArchive(agent.id, 'its turn')
+    if (!rt.isBotSession(agent.id)) return
+    releaseRescued(agent)
+    void checkContext(agent).catch(error => rt.warn('ds-bot: context check for %s failed: %s', agent.id, error))
   })
 
   // The handoff package opens a part as a turn of its own, so it is the first message
diff --git a/core/test/install.test.js b/core/test/install.test.js
index 91fe0d46..5bb485f9 100644
--- a/core/test/install.test.js
+++ b/core/test/install.test.js
@@ -86,6 +86,7 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => {
     'on agent/pre-step (prepend)',
     'on agent/request-error (prepend)',
     'on agent/request-error (prepend)',
+    'effect ds-bot: archive a part once its job ends',
     'on agent/status',
     'on llm/stream',
     'on llm/stream',
diff --git a/core/test/part-archive.test.js b/core/test/part-archive.test.js
index 95b90a62..f45652fe 100644
--- a/core/test/part-archive.test.js
+++ b/core/test/part-archive.test.js
@@ -109,7 +109,18 @@ async function start({ schedule, activities = {} } = {}) {
   for (const dir of [team, project]) mkdirSync(dir, { recursive: true })
   writeFileSync(join(home, 'state.json'), JSON.stringify(TEAM))
 
-  const ctx = fakeContext({ services: schedule === undefined ? {} : { schedule } })
+  // DSH's job registry, as the host sees it: the background job a Session owns holds the
+  // Session open, and its settlement is the event #1433 waits for.
+  const jobListeners = []
+  const jobs = {
+    events: {
+      subscribe(filter, listener) {
+        jobListeners.push({ filter, listener })
+        return () => {}
+      },
+    },
+  }
+  const ctx = fakeContext({ services: { ...(schedule === undefined ? {} : { schedule }), jobs } })
   const created = []
   const cwds = new Map()
   const agents = new Map()
@@ -134,6 +145,7 @@ async function start({ schedule, activities = {} } = {}) {
   }
   const archived = []
   const stopped = []
+  const attempts = []
   const registry = new Map([['w1', { id: 'w1', path: team, title: 'DS Bot' }]])
   Object.assign(ctx, {
     llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) },
@@ -143,6 +155,7 @@ async function start({ schedule, activities = {} } = {}) {
       resolveByPath: async path => [...registry.values()].find(workspace => workspace.path === path),
       create: async path => ({ id: 'w-new', path }),
       async archiveSession(sessionId, options) {
+        attempts.push(sessionId)
         const kinds = [...(schedule?.armedOn(sessionId).length > 0 ? ['schedule'] : []), ...(activities[sessionId] ?? [])]
         if (kinds.length > 0 && options?.stopActivity !== true) throw activeSessionError(sessionId, kinds)
         if (options?.stopActivity === true) {
@@ -185,7 +198,11 @@ async function start({ schedule, activities = {} } = {}) {
   }
   // Loading once fills the Session-owner indexes.
   await tool('list_bots', {}, 'a')
-  return { ctx, created, agentOf, call, state, archived, stopped, activities, home, project, until, settle, stop }
+  // The job registry's settlement, for the Session that owned the job.
+  const settleJob = (sessionId) => {
+    for (const { listener } of jobListeners) listener({ type: 'settled', job: { id: 'job-1', owner: sessionId, status: 'succeeded' } })
+  }
+  return { ctx, created, agentOf, call, state, archived, stopped, attempts, activities, settleJob, home, project, until, settle, stop }
 }
 
 const settle = () => new Promise(resolve => setTimeout(resolve, 20))
@@ -237,9 +254,98 @@ test('#896: a part held by a job is never force-archived', async (t) => {
   assert.equal(answer.ok, true, answer.error?.message)
   await h.until(() => h.state().bots.b.sessionId === 's1')
   await drained()
-  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the earlier part stays active, as before this fix')
   assert.deepEqual(h.stopped, [], 'a stop request never reaches a job')
   assert.deepEqual(h.activities.b, ['job'], 'the job still runs')
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the part stays active while the job holds it')
+  // The job ends and its registry says so: that settlement is the retry #1433 adds, and it
+  // asks for no options — the part ends up archived without anything being stopped.
+  delete h.activities.b
+  h.settleJob('b')
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), 'b', 'the earlier part is archived once the job ended')
+  assert.deepEqual(h.stopped, [], 'a job is archived, never stopped')
+})
+
+// #1433: a part a job holds at handoff used to stay active for the life of the instance.
+// The Session is idle while the job runs, so its idle event has already been published by
+// the time the archive is refused, and the part that was handed over takes no further turn
+// (its input goes to the fresh part): the job's own settlement is the signal that names the
+// end. A refused retry keeps the id rather than spending the one chance.
+test('#1433: an idle before the job ended spends nothing, and the part is archived when it ends', async (t) => {
+  const schedule = scheduleService()
+  const h = await start({ schedule, activities: { b: ['job'] } })
+  t.after(h.stop)
+  const answer = await h.call('update-bot', { id: 'b', workDir: h.project })
+  assert.equal(answer.ok, true, answer.error?.message)
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the first attempt is refused while the job holds the part')
+  assert.deepEqual(h.stopped, [], 'the refusal stops nothing')
+  // The Session reports idle while the job still runs: this attempt is refused as well and
+  // nothing is forced, but the id is kept for the end that has not happened yet.
+  h.ctx.emit('agent/status', { agent: h.agentOf('b'), status: 'idle' })
+  await drained()
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'a refused retry archives nothing')
+  assert.deepEqual(h.stopped, [], 'a refused retry stops nothing')
+  assert.deepEqual(h.activities.b, ['job'], 'the job still runs')
+  assert.equal(h.attempts.filter(id => id === 'b').length, 2, 'the retry was actually attempted')
+  // The job ends: that settlement still finds the id, because the refusal above kept it.
+  delete h.activities.b
+  h.settleJob('b')
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), 'b', 'the part is archived once the job ended')
+  assert.deepEqual(h.stopped, [], 'nothing was ever stopped for it')
+})
+
+test('#1433: one idle after the job ended archives the part, and the id is dropped once it lands', async (t) => {
+  const schedule = scheduleService()
+  const h = await start({ schedule, activities: { b: ['job'] } })
+  t.after(h.stop)
+  const answer = await h.call('update-bot', { id: 'b', workDir: h.project })
+  assert.equal(answer.ok, true, answer.error?.message)
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  assert.deepEqual(h.stopped, [], 'the refusal stops nothing')
+  delete h.activities.b
+  h.ctx.emit('agent/status', { agent: h.agentOf('b'), status: 'idle' })
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), 'b', 'the earlier part is archived once the job ended')
+  assert.deepEqual(h.stopped, [], 'the retry asks for no options')
+  const landed = h.attempts.filter(id => id === 'b').length
+  assert.equal(landed, 2, 'one attempt at the handoff and one after the job ended')
+  // The id left the table when the attempt landed, so no later signal asks again — a
+  // long-lived instance collects no queue of them and retries nothing in a loop.
+  h.ctx.emit('agent/status', { agent: h.agentOf('b'), status: 'idle' })
+  h.settleJob('b')
+  await drained()
+  assert.equal(h.attempts.filter(id => id === 'b').length, landed, 'no second attempt after it landed')
+})
+
+// #1433: the retry can also meet a refusal a scheduled task alone caused — the handoff's
+// task move failed (`failPut`), so the task is still armed on the earlier part when the job
+// ends. That refusal is the one the handoff answers by stopping the activity (`#896`), and
+// the retry has to answer it the same way instead of dropping the id and leaving the part
+// active for the life of the instance.
+test('#1433: a scheduled task alone still holding the part takes the handoff stop on the retry', async (t) => {
+  const schedule = scheduleService({ failPut: true })
+  const h = await start({ schedule, activities: { b: ['job'] } })
+  t.after(h.stop)
+  schedule.arm('b')
+  const answer = await h.call('update-bot', { id: 'b', workDir: h.project })
+  assert.equal(answer.ok, true, answer.error?.message)
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  // The job holds the part at handoff, so neither the part nor its task is forced.
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the job holds the part at handoff, so nothing is forced')
+  assert.deepEqual(h.stopped, [], 'the handoff stops nothing while a job holds the part')
+  assert.equal(schedule.armedOn('b').length, 1, 'the failed move left the task on the earlier part')
+  // The job ends on its own; the task is still armed, so this refusal names the schedule alone.
+  delete h.activities.b
+  h.settleJob('b')
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), ['b', { stopActivity: true }], 'the retry stops the task and archives the part')
+  assert.deepEqual(schedule.armedOn('b'), [], 'no armed task is left on the finished part')
+  assert.deepEqual(h.stopped, ['b'], 'the one stop came from the schedule-only refusal, never for the job')
 })
 
 // The refusal the three cases above can meet when `TMPDIR` sits inside `DSH_HOME`, from a
diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md
index 0cdc8b4d..ce4409c8 100644
--- a/docs/subsystems/core-host.md
+++ b/docs/subsystems/core-host.md
@@ -409,7 +409,7 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi
 - **A checkpoint is judged by the prompt that is left** (`parts.js` `pressureLeft`): the token meter stays anchored on the provider usage from before the span was replaced, and its surface delta only subtracts a 4-characters-per-token estimate of the removed text. A denser report — the mock's 3 characters per token, or CJK on a real model — can therefore still read above the check line after the checkpoint has replaced the span, and the part would move on even though the steward said to condense in place. After a compaction, the part stays when the heuristic size of the remaining surface plus the tool schemas is under the check line. A surface that is still over it moves on, as before. The decision to cross the line in the first place still uses the anchored total (#917).
 - **The steward is asked again before a part moves** (`parts.js` `decide`, #1447): its one answer decides whether the part moves, and a rate limit on that request — a per-minute count on the same key the conversation uses, recovered in seconds — used to read as "no answer" and moved a part that was still working: a fresh Session, the latest messages re-sent, a note written through the same limited key, and a handoff whose reason said "the previous part grew long". A route reports a failed request as a terminal finish chunk, not as a thrown value, so `askSteward` carries that chunk's `failure` — its `code`, `status` and `providerRetryAfterMs`, not only its message — onto the Error it throws (`stewardFailure`), and the classification below reads exactly those. A failure that can pass (rate limit, timeout, 5xx, by `RATE_LIMIT`/`SERVER`/`TRANSPORT`/`TIMEOUT` or by a 429/5xx status) is now retried up to `STEWARD_ATTEMPTS` (3), waiting the `providerRetryAfterMs` the provider names (capped at `STEWARD_RETRY_MAX_MS`, 5 s; the raw adapters' `retryAfter` is seconds and still read) or a short backoff that grows by attempt; a success at any attempt is the answer. Below the hard line a failure that outlasts the attempts spends the round instead of the part: the part stays, and the next round asks again only after `STEWARD_COOLDOWN_MS` (60 s, then 300 s, counting failures in a row). Over the hard line, and for a failure that retrying cannot change, the part still moves on; `segment.steward` then records the failure and the handoff says `the context steward could not be asked whether to keep it (…)` rather than the default reason. A note's own retry (`NOTE_ATTEMPTS`, `NOTE_COOLDOWN_MS`) is a separate policy and unchanged. Covered by `core/test/steward-retry.test.js`.
 - **Making room before a step has its own deadline** (`parts.js` `condense`, #1285): the pre-step overflow compaction used to carry only the turn's own signal, so a summary request stuck in retries — a 429 storm retries each request five times with its backoff — parked the turn before its first step: no `step/start`, no `turn/end`, `load: 1` and the roster's `stalled` hint until a human called `stop_bot`. `COMPACT_DEADLINE_MS` (`config.compactDeadlineMs`, 300 000 ms) now bounds that call through `AbortSignal.any([signal, deadline])`. When the deadline is what ended the wait, `condense` warns `ds-bot: making room for %s ran out of %d ms before a step; the turn ends here and the next one tries again: …` and rethrows, so DSH's own `finally` appends `turn/end` and the turn fails normally. DSH claimed that step's input before the pre-step hook ran and does not hand it back for a turn that ends in an error, so before the throw the deadline branch queues the batch again (`keepClaimed`, `agent.inbox.append('next-turn', …)`, one ordinary queued message each, a copy the inbox already holds refused): nothing the turn was woken for is lost, and the next turn — which the loop starts straight away while input is pending — reads it and tries to make room again. A provider that stays down therefore fails one turn per deadline, back to back, each ending normally instead of parking at `load: 1`; no backoff is added here, the queue is the thing that carries the work forward. A turn the user aborted keeps the old path (warn, `false`, move on), and the `!signal?.aborted` half of the guard keeps a turn the user stopped in charge even when the deadline was already reached. The idle pass is unchanged on `NOTE_TIMEOUT_MS` (600 000 ms, one shared signal over both note attempts). The turn-ending waits stay where they were: steward 120 s, idle condense and the handoff note each at that same 600 s budget. Covered by `core/test/pre-step-compaction.test.js`.
-- **A part is committed before its handoff runs** (`parts.js` `moveOn`): the Bot's `sessionId` / `room.sessions`, the new segment and the `rescued` mark are saved first, so the turn the handoff starts can never see an uncommitted part — a failure handler that did would start a second switch against it and lose the previous part's note, and the rescue would not know it may only happen once. The handoff package is written next; if that fails, the pointer is rolled back (best effort saved, the fresh Session archived, the rescued mark dropped), so the old part is current and the finalizer hands the input on there. Two windows stay open, and both are written down rather than discovered later: a crash after that save and before the handoff package is flushed leaves the fresh part current with no package — the stopped turn's input still goes to it, so no work is lost, but the note is, and the rollback never runs; and the rollback's own save is best effort (`rt.save().catch(() => {})`), so a failing rollback leaves the file showing the fresh part while memory holds the old one, and the next start behaves like the first window. Neither is a reason to clear the pointer before the save: a note lost this way is a part without a handoff note, while clearing first loses the handoff package itself. The move is marked done only after the stopped turn's input and the old inbox messages are forwarded; a switch that never starts (no place, another switch, a read-only host) lets the step continue instead of rejecting its messages, which DSH has already taken out of the inbox. An armed scheduled task keeps the earlier Session active and DSH refuses to archive it, so the switch keeps the `moveSchedules` promise and waits for the move before archiving the earlier part; when the move failed and the refusal reports `schedule` as the only activity, the plain archive is retried with `{ stopActivity: true }` (the way a deletion does, `roster.js:450`), so a finished part is never left active with a task nobody sees; a refusal caused by a job, turn or subagent is not retried, because `stopActivity` would stop that too (#896). Covered by `core/test/work-dir.test.js`, `core/test/part-archive.test.js` and `core/test/bot-messages.test.js`.
+- **A part is committed before its handoff runs** (`parts.js` `moveOn`): the Bot's `sessionId` / `room.sessions`, the new segment and the `rescued` mark are saved first, so the turn the handoff starts can never see an uncommitted part — a failure handler that did would start a second switch against it and lose the previous part's note, and the rescue would not know it may only happen once. The handoff package is written next; if that fails, the pointer is rolled back (best effort saved, the fresh Session archived, the rescued mark dropped), so the old part is current and the finalizer hands the input on there. Two windows stay open, and both are written down rather than discovered later: a crash after that save and before the handoff package is flushed leaves the fresh part current with no package — the stopped turn's input still goes to it, so no work is lost, but the note is, and the rollback never runs; and the rollback's own save is best effort (`rt.save().catch(() => {})`), so a failing rollback leaves the file showing the fresh part while memory holds the old one, and the next start behaves like the first window. Neither is a reason to clear the pointer before the save: a note lost this way is a part without a handoff note, while clearing first loses the handoff package itself. The move is marked done only after the stopped turn's input and the old inbox messages are forwarded; a switch that never starts (no place, another switch, a read-only host) lets the step continue instead of rejecting its messages, which DSH has already taken out of the inbox. An armed scheduled task keeps the earlier Session active and DSH refuses to archive it, so the switch keeps the `moveSchedules` promise and waits for the move before archiving the earlier part; when the move failed and the refusal reports `schedule` as the only activity, the plain archive is retried with `{ stopActivity: true }` (the way a deletion does, `roster.js:450`), so a finished part is never left active with a task nobody sees; a refusal caused by a job, turn or subagent is never answered with `stopActivity`, because that would stop it too (#896): the id is remembered instead and the archive is asked for once more, with no options, when that holder is reported ended — the job registry's `settled` event for a job, the Session's own next `agent/status` idle otherwise — keeping the id while a refusal still names one of them, and holding at most 32, oldest dropped (#1433). Covered by `core/test/work-dir.test.js`, `core/test/part-archive.test.js` and `core/test/bot-messages.test.js`.
 
 ### Connectors: GitHub, the Composio gateway, and the Feishu sender (#311, #1263)
 
diff --git a/patches/ds-bot/0001-core-fixes.patch b/patches/ds-bot/0001-core-fixes.patch
index f0f6fe37..90ede286 100644
--- a/patches/ds-bot/0001-core-fixes.patch
+++ b/patches/ds-bot/0001-core-fixes.patch
@@ -8842,6 +8842,14 @@ diff --git a/test/install.test.js b/test/install.test.js
      'on approval/request (prepend)',
      'on agent/inbox/discarded',
      'on session/event',
+@@ -81,6 +81,7 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => {
+     'on agent/pre-step (prepend)',
+     'on agent/request-error (prepend)',
+     'on agent/request-error (prepend)',
++    'effect ds-bot: archive a part once its job ends',
+     'on agent/status',
+     'on llm/stream',
+     'on llm/stream',
 @@ -91,12 +91,23 @@ test('apply registers listeners, tools, and effects in a fixed order', (t) => {
      'on llm/stream',
      'effect ds-bot: load secrets',
diff --git a/patches/ds-bot/0014-flybot-branding.patch b/patches/ds-bot/0014-flybot-branding.patch
index 71889cac..d11cfaf9 100644
--- a/patches/ds-bot/0014-flybot-branding.patch
+++ b/patches/ds-bot/0014-flybot-branding.patch
@@ -1,5 +1,6 @@
 FlyBot patch 0014-flybot-branding on Fei2411/DeepSeek-Bot@2b10c70 (ds-bot/). Generated by scripts/ds-bot-patches.mjs; edits to upstream files only (files FlyBot adds are in files/).
 
+
 diff --git a/src/client/about.js b/src/client/about.js
 --- a/src/client/about.js
 +++ b/src/client/about.js
@@ -549,7 +550,9 @@ diff --git a/src/host/parts.js b/src/host/parts.js
    const clock = createClock(config.timeZone)
  
    // Lines for the model the Session last routed to; undefined while it has not
-@@ -502,41 +502,135 @@ export function install(rt) {
+@@ -500,43 +500,137 @@ export function install(rt) {
+       signal: AbortSignal.timeout(STEWARD_TIMEOUT_MS),
+     })) assembler.push(chunk)
      const finish = assembler.finish
 -    if (finish.kind === 'error' || finish.kind === 'aborted') throw new Error(finish.failure?.message ?? finish.kind)
 +    if (finish.kind === 'error' || finish.kind === 'aborted') throw stewardFailure(finish)
@@ -767,6 +770,28 @@ diff --git a/src/host/parts.js b/src/host/parts.js
    const switches = new Map()
    // Parts whose move to their working folder failed, and the time before which it is
    // not tried again.
+@@ -640,6 +640,21 @@ export function install(rt) {
+   const noteFailures = new Map()
+   // Parts the steward is deciding about at the end of a turn.
+   const checking = new Set()
++  // Earlier parts whose handoff archive a holder that ends on its own refused: a job, a
++  // turn or a subagent, never a scheduled task. Such a holder is never stopped, so the id
++  // waits here until that end is reported — its job settling, or its Session going idle —
++  // and the archive is asked for once more, with no options (#1433). A refusal that names
++  // no kind at all is not remembered: nothing says what would end.
++  const pendingArchive = new Set()
++  // Ids whose one more attempt is in flight, so a job that settles just as its Session
++  // goes idle is asked for once, not twice.
++  const attemptingArchive = new Set()
++  // The most ids that may wait: a Set keeps insertion order, so the oldest leaves first.
++  // Nothing here retries on a timer — one attempt per id at a time, and only when a
++  // holder-end signal names it — so the table cannot grow by itself. The cap is the
++  // backstop for a host that never reports the end: such a part is dropped, never forced
++  // (#896).
++  const PENDING_ARCHIVE_MAX = 32
+ 
+   const placeOf = (sessionId) => {
+     const botId = rt.chatOwners.get(sessionId)
 @@ -670,10 +681,10 @@ export function install(rt) {
  
    // `hard`: past the hard line or the window, the part moves on even without a note
@@ -797,7 +822,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js
    }
  
    async function runSwitch(agent, job) {
-@@ -716,8 +727,24 @@ export function install(rt) {
+@@ -716,8 +716,31 @@ export function install(rt) {
      }
      if (job.moved) {
        rt.renewMemory(agent.id)
@@ -813,7 +838,14 @@ diff --git a/src/host/parts.js b/src/host/parts.js
 +          // refusal a scheduled task alone caused (#896 review): a job, turn or subagent
 +          // that holds the part is left to finish, exactly as before this fix.
 +          const kinds = Array.isArray(error?.activity) ? error.activity.map(entry => entry?.kind) : []
-+          if (kinds.length === 0 || kinds.some(kind => kind !== 'schedule')) return
++          if (kinds.length === 0 || kinds.some(kind => kind !== 'schedule')) {
++            // A job, turn or subagent that holds the part is left to finish, and the part
++            // used to stay active for the life of the instance (#1433). Remember it for
++            // one more attempt once its holder ends; a schedule-only refusal takes the
++            // stop below.
++            if (kinds.length > 0) rememberPendingArchive(agent.id)
++            return
++          }
 +          // A task that could not be moved still holds the part, and leaving it active
 +          // forever is the leak #896 is about. Stop its activity, the way a deletion does
 +          // (roster.js:450): a finished part is never left running a task nobody sees.
@@ -873,7 +905,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js
      if (room !== undefined) {
        room.segments ??= {}
        ;(room.segments[bot.id] ??= []).push(segment)
-@@ -790,28 +801,50 @@ export function install(rt) {
+@@ -790,28 +790,50 @@ export function install(rt) {
          delete state.questions[agent.id]
        }
      }
@@ -892,6 +924,12 @@ diff --git a/src/host/parts.js b/src/host/parts.js
 -    if (intended !== undefined && cwd !== undefined && cwd !== intended) {
 -      folderRetry.set(freshId, Date.now() + FOLDER_RETRY_MS)
 -      rt.warn('ds-bot: the fresh part of %s did not start in %s; it works in %s', bot.name, intended, cwd)
+-    }
+-    const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder })
+-    freshParts.set(freshId, estimateTokens(text))
+-    fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) }))
+-    await ctx.sessions.flush(fresh.session)
+-    // Then what the old part held: the stopped turn's input, then whatever waited.
 +    try {
 +      await rt.save()
 +    } catch (error) {
@@ -920,12 +958,7 @@ diff --git a/src/host/parts.js b/src/host/parts.js
 +      await rt.save().catch(() => { /* the in-memory state is the old part; a failed rollback save is the worst case */ })
 +      await ctx.workspaceRegistry.archiveSession(freshId).catch(() => { /* best effort */ })
 +      throw error
-     }
--    const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder })
--    freshParts.set(freshId, estimateTokens(text))
--    fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) }))
--    await ctx.sessions.flush(fresh.session)
--    // Then what the old part held: the stopped turn's input, then whatever waited.
++    }
 +    // The earlier part may not be archived while this move still holds its tasks (#896), so
 +    // the switch keeps the promise and waits for it before archiving; the handoff goes on now.
 +    job.scheduleMove = rt.moveSchedules(agent.id, freshId)
@@ -1001,6 +1034,85 @@ diff --git a/src/host/parts.js b/src/host/parts.js
      } finally {
        checking.delete(agent.id)
      }
+@@ -957,11 +957,74 @@ export function install(rt) {
+     if (end !== undefined && end.data?.reason?.kind !== 'error') rescued.delete(agent.id)
+   }
+ 
++  function rememberPendingArchive(sessionId) {
++    pendingArchive.add(sessionId)
++    while (pendingArchive.size > PENDING_ARCHIVE_MAX) pendingArchive.delete(pendingArchive.values().next().value)
++    watchJobSettlements()
++  }
++
++  // A background job ends on its own schedule, and its Session is idle by then: the idle
++  // event came before this attempt and will not come again, and the part that was handed
++  // over never takes another turn (its input goes to the fresh part, `forward`). The job
++  // registry's own settlement is the event that names that moment (#1433), read through
++  // `ctx.get` — no inject asked, and `undefined` when the service is not there yet: a
++  // composition may provide the job registry after this plugin loads, and one without it
++  // has no job to wait for.
++  let jobUnwatch
++  function watchJobSettlements() {
++    if (jobUnwatch !== undefined) return
++    const events = ctx.get('jobs')?.events
++    if (events === undefined) return
++    jobUnwatch = events.subscribe({ owners: 'all' }, (event) => {
++      if (event?.type !== 'settled' || typeof event.job?.owner !== 'string') return
++      attemptPendingArchive(event.job.owner, 'its job')
++    })
++  }
++  ctx.effect(() => () => jobUnwatch?.(), 'ds-bot: archive a part once its job ends')
++
++  // One more archive attempt for a part whose holder has ended, with no options, so a job,
++  // a turn or a subagent is never stopped for it (#896). A refusal that still names one of
++  // those means the holder has not ended after all — the Session reported its idle while
++  // the job ran — so the id stays and the next signal tries again instead of the one
++  // chance being spent. It leaves when the attempt lands, when the refusal names no
++  // activity at all, or when a scheduled task alone names it — that one is answered by the
++  // handoff's own `stopActivity` retry, never by holding the id.
++  function attemptPendingArchive(sessionId, endedBy) {
++    if (!pendingArchive.has(sessionId) || attemptingArchive.has(sessionId)) return
++    attemptingArchive.add(sessionId)
++    void ctx.workspaceRegistry.archiveSession(sessionId)
++      .then(() => pendingArchive.delete(sessionId))
++      .catch(error => {
++        const kinds = Array.isArray(error?.activity) ? error.activity.map(entry => entry?.kind) : []
++        // A holder that ends on its own has not ended after all: keep the id and let the
++        // next signal try again, instead of spending the one chance.
++        if (kinds.some(kind => kind !== 'schedule')) {
++          rt.warn('ds-bot: earlier part %s still held by its %s after %s ended: %s', sessionId, kinds.join(', '), endedBy, error)
++          return
++        }
++        pendingArchive.delete(sessionId)
++        // A scheduled task alone is the one refusal the handoff answers by stopping the
++        // activity (#896): a task that could not be moved still holds the part, and leaving
++        // it active is the leak itself. The retry answers it the same way, and the id is
++        // spent either way — a stop that fails is not asked for again.
++        if (kinds.length > 0) {
++          void ctx.workspaceRegistry.archiveSession(sessionId, { stopActivity: true })
++            .catch(again => rt.warn('ds-bot: earlier part %s not archived after %s ended, not even stopped: %s', sessionId, endedBy, again))
++          return
++        }
++        rt.warn('ds-bot: earlier part %s not archived after %s ended: %s', sessionId, endedBy, error)
++      })
++      .finally(() => attemptingArchive.delete(sessionId))
++  }
++
+   ctx.on('agent/status', ({ agent, status }) => {
+-    if (status === 'idle' && rt.isBotSession(agent.id)) {
+-      releaseRescued(agent)
+-      void checkContext(agent).catch(error => rt.warn('ds-bot: context check for %s failed: %s', agent.id, error))
+-    }
++    if (status !== 'idle') return
++    // Before the Bot check: the part that was handed over is no longer this Bot's current
++    // Session, and its holder ends in it all the same.
++    attemptPendingArchive(agent.id, 'its turn')
++    if (!rt.isBotSession(agent.id)) return
++    releaseRescued(agent)
++    void checkContext(agent).catch(error => rt.warn('ds-bot: context check for %s failed: %s', agent.id, error))
+   })
+ 
+   // The handoff package opens a part as a turn of its own, so it is the first message
 diff --git a/src/host/secrets.js b/src/host/secrets.js
 --- a/src/host/secrets.js
 +++ b/src/host/secrets.js
diff --git a/patches/ds-bot/files/test/part-archive.test.js b/patches/ds-bot/files/test/part-archive.test.js
index 95b90a62..f45652fe 100644
--- a/patches/ds-bot/files/test/part-archive.test.js
+++ b/patches/ds-bot/files/test/part-archive.test.js
@@ -109,7 +109,18 @@ async function start({ schedule, activities = {} } = {}) {
   for (const dir of [team, project]) mkdirSync(dir, { recursive: true })
   writeFileSync(join(home, 'state.json'), JSON.stringify(TEAM))
 
-  const ctx = fakeContext({ services: schedule === undefined ? {} : { schedule } })
+  // DSH's job registry, as the host sees it: the background job a Session owns holds the
+  // Session open, and its settlement is the event #1433 waits for.
+  const jobListeners = []
+  const jobs = {
+    events: {
+      subscribe(filter, listener) {
+        jobListeners.push({ filter, listener })
+        return () => {}
+      },
+    },
+  }
+  const ctx = fakeContext({ services: { ...(schedule === undefined ? {} : { schedule }), jobs } })
   const created = []
   const cwds = new Map()
   const agents = new Map()
@@ -134,6 +145,7 @@ async function start({ schedule, activities = {} } = {}) {
   }
   const archived = []
   const stopped = []
+  const attempts = []
   const registry = new Map([['w1', { id: 'w1', path: team, title: 'DS Bot' }]])
   Object.assign(ctx, {
     llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) },
@@ -143,6 +155,7 @@ async function start({ schedule, activities = {} } = {}) {
       resolveByPath: async path => [...registry.values()].find(workspace => workspace.path === path),
       create: async path => ({ id: 'w-new', path }),
       async archiveSession(sessionId, options) {
+        attempts.push(sessionId)
         const kinds = [...(schedule?.armedOn(sessionId).length > 0 ? ['schedule'] : []), ...(activities[sessionId] ?? [])]
         if (kinds.length > 0 && options?.stopActivity !== true) throw activeSessionError(sessionId, kinds)
         if (options?.stopActivity === true) {
@@ -185,7 +198,11 @@ async function start({ schedule, activities = {} } = {}) {
   }
   // Loading once fills the Session-owner indexes.
   await tool('list_bots', {}, 'a')
-  return { ctx, created, agentOf, call, state, archived, stopped, activities, home, project, until, settle, stop }
+  // The job registry's settlement, for the Session that owned the job.
+  const settleJob = (sessionId) => {
+    for (const { listener } of jobListeners) listener({ type: 'settled', job: { id: 'job-1', owner: sessionId, status: 'succeeded' } })
+  }
+  return { ctx, created, agentOf, call, state, archived, stopped, attempts, activities, settleJob, home, project, until, settle, stop }
 }
 
 const settle = () => new Promise(resolve => setTimeout(resolve, 20))
@@ -237,9 +254,98 @@ test('#896: a part held by a job is never force-archived', async (t) => {
   assert.equal(answer.ok, true, answer.error?.message)
   await h.until(() => h.state().bots.b.sessionId === 's1')
   await drained()
-  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the earlier part stays active, as before this fix')
   assert.deepEqual(h.stopped, [], 'a stop request never reaches a job')
   assert.deepEqual(h.activities.b, ['job'], 'the job still runs')
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the part stays active while the job holds it')
+  // The job ends and its registry says so: that settlement is the retry #1433 adds, and it
+  // asks for no options — the part ends up archived without anything being stopped.
+  delete h.activities.b
+  h.settleJob('b')
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), 'b', 'the earlier part is archived once the job ended')
+  assert.deepEqual(h.stopped, [], 'a job is archived, never stopped')
+})
+
+// #1433: a part a job holds at handoff used to stay active for the life of the instance.
+// The Session is idle while the job runs, so its idle event has already been published by
+// the time the archive is refused, and the part that was handed over takes no further turn
+// (its input goes to the fresh part): the job's own settlement is the signal that names the
+// end. A refused retry keeps the id rather than spending the one chance.
+test('#1433: an idle before the job ended spends nothing, and the part is archived when it ends', async (t) => {
+  const schedule = scheduleService()
+  const h = await start({ schedule, activities: { b: ['job'] } })
+  t.after(h.stop)
+  const answer = await h.call('update-bot', { id: 'b', workDir: h.project })
+  assert.equal(answer.ok, true, answer.error?.message)
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the first attempt is refused while the job holds the part')
+  assert.deepEqual(h.stopped, [], 'the refusal stops nothing')
+  // The Session reports idle while the job still runs: this attempt is refused as well and
+  // nothing is forced, but the id is kept for the end that has not happened yet.
+  h.ctx.emit('agent/status', { agent: h.agentOf('b'), status: 'idle' })
+  await drained()
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'a refused retry archives nothing')
+  assert.deepEqual(h.stopped, [], 'a refused retry stops nothing')
+  assert.deepEqual(h.activities.b, ['job'], 'the job still runs')
+  assert.equal(h.attempts.filter(id => id === 'b').length, 2, 'the retry was actually attempted')
+  // The job ends: that settlement still finds the id, because the refusal above kept it.
+  delete h.activities.b
+  h.settleJob('b')
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), 'b', 'the part is archived once the job ended')
+  assert.deepEqual(h.stopped, [], 'nothing was ever stopped for it')
+})
+
+test('#1433: one idle after the job ended archives the part, and the id is dropped once it lands', async (t) => {
+  const schedule = scheduleService()
+  const h = await start({ schedule, activities: { b: ['job'] } })
+  t.after(h.stop)
+  const answer = await h.call('update-bot', { id: 'b', workDir: h.project })
+  assert.equal(answer.ok, true, answer.error?.message)
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  assert.deepEqual(h.stopped, [], 'the refusal stops nothing')
+  delete h.activities.b
+  h.ctx.emit('agent/status', { agent: h.agentOf('b'), status: 'idle' })
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), 'b', 'the earlier part is archived once the job ended')
+  assert.deepEqual(h.stopped, [], 'the retry asks for no options')
+  const landed = h.attempts.filter(id => id === 'b').length
+  assert.equal(landed, 2, 'one attempt at the handoff and one after the job ended')
+  // The id left the table when the attempt landed, so no later signal asks again — a
+  // long-lived instance collects no queue of them and retries nothing in a loop.
+  h.ctx.emit('agent/status', { agent: h.agentOf('b'), status: 'idle' })
+  h.settleJob('b')
+  await drained()
+  assert.equal(h.attempts.filter(id => id === 'b').length, landed, 'no second attempt after it landed')
+})
+
+// #1433: the retry can also meet a refusal a scheduled task alone caused — the handoff's
+// task move failed (`failPut`), so the task is still armed on the earlier part when the job
+// ends. That refusal is the one the handoff answers by stopping the activity (`#896`), and
+// the retry has to answer it the same way instead of dropping the id and leaving the part
+// active for the life of the instance.
+test('#1433: a scheduled task alone still holding the part takes the handoff stop on the retry', async (t) => {
+  const schedule = scheduleService({ failPut: true })
+  const h = await start({ schedule, activities: { b: ['job'] } })
+  t.after(h.stop)
+  schedule.arm('b')
+  const answer = await h.call('update-bot', { id: 'b', workDir: h.project })
+  assert.equal(answer.ok, true, answer.error?.message)
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  // The job holds the part at handoff, so neither the part nor its task is forced.
+  assert.equal(archivedEntry(h.archived, 'b'), undefined, 'the job holds the part at handoff, so nothing is forced')
+  assert.deepEqual(h.stopped, [], 'the handoff stops nothing while a job holds the part')
+  assert.equal(schedule.armedOn('b').length, 1, 'the failed move left the task on the earlier part')
+  // The job ends on its own; the task is still armed, so this refusal names the schedule alone.
+  delete h.activities.b
+  h.settleJob('b')
+  await h.until(() => archivedEntry(h.archived, 'b') !== undefined)
+  assert.deepEqual(archivedEntry(h.archived, 'b'), ['b', { stopActivity: true }], 'the retry stops the task and archives the part')
+  assert.deepEqual(schedule.armedOn('b'), [], 'no armed task is left on the finished part')
+  assert.deepEqual(h.stopped, ['b'], 'the one stop came from the schedule-only refusal, never for the job')
 })
 
 // The refusal the three cases above can meet when `TMPDIR` sits inside `DSH_HOME`, from a