fix(parts): archive a job-held part once its job ends

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)).

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

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',

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