fix(parts): wait for the task move before archiving the earlier part (#896)

changelog.d/896-part-archive-race.fixed.md

@@ -0,0 +1 @@
+A Bot chat whose earlier part had an armed scheduled task now archives that part when it hands off: the switch waits for the task move to finish before archiving, and if a task could not be moved it archives the finished part with that task stopped, instead of leaving it active forever with a warning only in the log; a part held by a running job or turn is left alone, as before. The task keeps running on the new part when it moves ([#896](https://github.com/ThinkFlowLab/FlyBot/issues/896)).

core/src/host/parts.js

@@ -725,8 +725,24 @@ export function install(rt) {
     }
     if (job.moved) {
       rt.renewMemory(agent.id)
+      // An armed scheduled task keeps its Session active, and DSH refuses to archive one
+      // (#896): the earlier part's tasks have to finish moving to the fresh part first.
+      await job.scheduleMove?.catch(error => rt.warn('ds-bot: tasks of earlier part %s not moved: %s', agent.id, error))
       await ctx.workspaceRegistry.archiveSession(agent.id)
-        .catch(error => rt.warn('ds-bot: earlier part %s not archived: %s', agent.id, error))
+        .catch(async (error) => {
+          rt.warn('ds-bot: earlier part %s not archived: %s', agent.id, error)
+          // `stopActivity` asks every `workspace/session-stop` provider to stop what it
+          // holds, not only the schedule service, so the retry may only be asked for a
+          // 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
+          // 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.
+          await ctx.workspaceRegistry.archiveSession(agent.id, { stopActivity: true })
+            .catch(again => rt.warn('ds-bot: earlier part %s not archived, not even stopped: %s', agent.id, again))
+        })
     }
   }
 
@@ -857,7 +873,10 @@ export function install(rt) {
       await ctx.workspaceRegistry.archiveSession(freshId).catch(() => { /* best effort */ })
       throw error
     }
-    void rt.moveSchedules(agent.id, freshId)
+    // 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)
+    job.scheduleMove.catch(error => rt.warn('ds-bot: tasks of %s not moved: %s', agent.id, error))
     // The old part's held input goes on here. Mark the move done only once it was handed on, so a
     // failure lets the finalizer retry it on the part that is now current.
     const held = [...agent.inbox.nextStep, ...agent.inbox.nextTurn]

core/test/part-archive.test.js

@@ -0,0 +1,213 @@
+import assert from 'node:assert/strict'
+import { mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { apply } from '../index.js'
+import { fakeContext } from './fake-ctx.js'
+
+// A Main Bot and a Bot, both idle in their own chat.
+const TEAM = {
+  version: 1,
+  mainBotIds: ['a'],
+  workspaceId: 'w1',
+  bots: {
+    a: { id: 'a', name: 'Chief', role: 'Boss', sessionId: 'a', createdAt: 1 },
+    b: { id: 'b', name: 'Writer', role: 'Write', sessionId: 'b', createdAt: 2 },
+  },
+  rooms: {},
+  agents: {},
+  exchanges: [],
+  exchangeTurns: {},
+  questions: {},
+  prefs: {},
+  revision: 0,
+}
+
+// DSH's schedule service, as the host sees it: tasks are stored per Session, and a
+// Session with an active task refuses to be archived unless its activity is stopped.
+function scheduleService({ listDelay = 0, failCreate = false } = {}) {
+  const tasks = new Map()
+  let next = 1
+  const armedOn = (sessionId) => [...tasks.entries()].filter(([, task]) => task.sessionId === sessionId).map(([id]) => id)
+  return {
+    tasks,
+    listDelay,
+    failCreate,
+    armedOn,
+    stop(sessionId) {
+      for (const id of armedOn(sessionId)) tasks.delete(id)
+    },
+    arm(sessionId, title = 'nightly check') {
+      const id = `schedule-${next++}`
+      tasks.set(id, { sessionId, record: { id, title, prompt: 'Wake up.', kind: 'at', scheduledAt: '2030-01-01T01:00:00.000Z' } })
+      return id
+    },
+    async list({ sessionId }) {
+      if (listDelay > 0) await new Promise(resolve => setTimeout(resolve, listDelay))
+      return [...tasks.values()].filter(task => task.sessionId === sessionId).map(task => task.record)
+    },
+    async create(sessionId, request) {
+      if (failCreate) throw new Error('create failed')
+      const id = `schedule-${next++}`
+      tasks.set(id, { sessionId, record: { id, title: request.title, prompt: request.prompt, kind: 'at', scheduledAt: request.at } })
+      return tasks.get(id).record
+    },
+    async delete({ sessionId, id }) {
+      if (tasks.get(id)?.sessionId !== sessionId) return { id, deleted: false, code: 'schedule_not_found' }
+      tasks.delete(id)
+      return { id, deleted: true }
+    },
+  }
+}
+
+// DSH's refusal, with the activity kinds it reports (`schedule`, `job`, …).
+function activeSessionError(sessionId, kinds) {
+  return Object.assign(new Error(`cannot archive session '${sessionId}': the session is active (${kinds.join(', ')})`), {
+    name: 'WorkspaceActiveSessionError',
+    sessionId,
+    activity: kinds.map(kind => ({ kind })),
+  })
+}
+
+// The team's data in a temporary home, and a project folder beside it. The registry
+// refuses an archive while a Session still has an active task or another activity,
+// exactly as DSH does.
+async function start({ schedule, activities = {} } = {}) {
+  const root = realpathSync.native(mkdtempSync(join(tmpdir(), 'ds-bot-part-archive-')))
+  const dsh = join(root, 'dsh')
+  const home = join(dsh, 'bot')
+  const team = join(home, 'workspace')
+  const project = join(root, 'project')
+  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 } })
+  const created = []
+  const cwds = new Map()
+  const agents = new Map()
+  const agentOf = (sessionId) => {
+    if (!agents.has(sessionId)) {
+      const calls = []
+      agents.set(sessionId, {
+        id: sessionId,
+        session: { id: sessionId, header: { cwd: cwds.get(sessionId) ?? team }, requestHeader: () => undefined, deriveMessages: () => [] },
+        calls,
+        status: 'idle',
+        followup(message) { calls.push(['followup', message]) },
+        steer(message) { calls.push(['steer', message]) },
+        inject(message) { calls.push(['inject', message]) },
+        cancel() {},
+        async whenIdle() {},
+        async runMaintenance(fn) { return fn(AbortSignal.timeout(60_000)) },
+        inbox: { nextStep: [], nextTurn: [], remove() {} },
+      })
+    }
+    return agents.get(sessionId)
+  }
+  const archived = []
+  const stopped = []
+  const registry = new Map([['w1', { id: 'w1', path: team, title: 'DS Bot' }]])
+  Object.assign(ctx, {
+    llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) },
+    workspaceRegistry: {
+      get: id => registry.get(id),
+      list: () => [...registry.values()],
+      resolveByPath: async path => [...registry.values()].find(workspace => workspace.path === path),
+      create: async path => ({ id: 'w-new', path }),
+      async archiveSession(sessionId, options) {
+        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) {
+          stopped.push(sessionId)
+          schedule?.stop(sessionId)
+          delete activities[sessionId]
+        }
+        archived.push(options === undefined ? sessionId : [sessionId, options])
+      },
+    },
+    sessionController: {
+      create: async (options) => {
+        const sessionId = `s${created.length + 1}`
+        created.push(options)
+        cwds.set(sessionId, options.cwd ?? registry.get(options.workspaceId)?.path)
+        return { sessionId }
+      },
+      resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }),
+      selectModel: async () => {},
+      page: async () => ({ records: [], hasMore: false }),
+    },
+    sessionTitle: { rename: async () => {} },
+    sessions: { flush: async () => {} },
+  })
+  let route
+  const register = ctx.connection.fetch.register
+  ctx.connection.fetch.register = (options) => {
+    route = options
+    return register(options)
+  }
+  apply(ctx, { home })
+  const call = async (endpoint, payload) => (await route.fetch(new Request('http://localhost/api/bot', {
+    method: 'POST', body: JSON.stringify({ endpoint, payload }),
+  }))).json()
+  const tool = (name, args, sessionId) => ctx.registeredTools.get(name).execute(args, { agent: agentOf(sessionId), name, callId: 'c1' })
+  const state = () => JSON.parse(readFileSync(join(home, 'state.json'), 'utf8'))
+  const stop = () => {
+    ctx.dispose()
+    rmSync(root, { recursive: true, force: true, maxRetries: 5 })
+  }
+  // Loading once fills the Session-owner indexes.
+  await tool('list_bots', {}, 'a')
+  return { ctx, created, agentOf, call, state, archived, stopped, activities, project, until, settle, stop }
+}
+
+const settle = () => new Promise(resolve => setTimeout(resolve, 20))
+const until = async (check) => {
+  const end = Date.now() + 2000
+  while (!check()) {
+    if (Date.now() > end) throw new Error('until: the condition did not hold within 2 s')
+    await settle()
+  }
+}
+// The part switch runs in the background; give its archive time to land (or not).
+const drained = async () => { for (let i = 0; i < 25; i += 1) await settle() }
+const archivedEntry = (archived, sessionId) => archived.find(entry => (Array.isArray(entry) ? entry[0] : entry) === sessionId)
+
+test('#896: a part whose task moved is archived once the move finished', async (t) => {
+  const schedule = scheduleService({ listDelay: 40 })
+  const h = await start({ schedule })
+  t.after(h.stop)
+  schedule.arm('b')
+  // A folder change moves an idle Bot's chat to a fresh part, the same switch a context
+  // check takes.
+  await h.call('update-bot', { id: 'b', workDir: h.project })
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  assert.deepEqual(archivedEntry(h.archived, 'b'), 'b', 'the earlier part is archived')
+  assert.deepEqual(schedule.armedOn('b'), [], 'the earlier part keeps no task')
+  assert.equal(schedule.armedOn('s1').length, 1, 'the task moved to the fresh part')
+})
+
+test('#896: a failed task move does not leave an armed earlier part active', async (t) => {
+  const schedule = scheduleService({ failCreate: true })
+  const h = await start({ schedule })
+  t.after(h.stop)
+  schedule.arm('b')
+  await h.call('update-bot', { id: 'b', workDir: h.project })
+  await h.until(() => h.state().bots.b.sessionId === 's1')
+  await drained()
+  assert.deepEqual(archivedEntry(h.archived, 'b'), ['b', { stopActivity: true }], 'the earlier part is archived with its activity stopped')
+  assert.deepEqual(schedule.armedOn('b'), [], 'no armed task is left on the finished part')
+})
+
+test('#896: a part held by a job is never force-archived', async (t) => {
+  const schedule = scheduleService()
+  const h = await start({ schedule, activities: { b: ['job'] } })
+  t.after(h.stop)
+  await h.call('update-bot', { id: 'b', workDir: h.project })
+  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')
+})

docs/subsystems/core-host.md

@@ -241,7 +241,7 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi
 - A friend's line in a Bot's own chat (#420): `deliver()` (`noteFriendLine`) stamps the receiving Bot with `friendLineAt` (published in the roster, so the save moves the revision) and keeps the senders seen per Bot in `state.remoteContacts[botId][handle]` (host only, capped at 200, dropped with the Bot). A sender's first line carries `source.remote.firstContact`. Group lines, and re-deliveries (`deferred`, `resumed`, `retried`, `resent`), stamp nothing. Covered by `core/test/flybot.test.js`.
 - **Pickup after a restart** (`resume.js`): every running Bot turn and every delivered-but-not-taken message is noted in `inflight.json`. At the next start a noted turn is picked up once only if a restart really cut it: the durable Session log's `turn/end` for that turn is `interrupted` (or an abort caused by `disposed`, or there is no end). A turn that finished while this plugin was not listening — `completed`, `blocked`, `max-tokens`, `forked`, or aborted by the user / hook / parent — is not repeated; a `turn/start` seen after the start means the noted turn ended too. Notes from a build before the record carried `turn` keep the old behaviour. A pickup is confirmed before its note is dropped: the notes read at start are seeded into the live `turns`/`queued` (so the file keeps everything still owed), a turn's note stays while its pickup runs and is dropped only once handed over (unless the resumed turn has written its own note first), and a queued message whose resend fails is kept for the next start, an exchange is noted once (a resend delivers the `exchangeId` the notes already hold, and noting it again left a copy behind after a turn took it up, so the start after that sent the message once more), and a turn taking up a message forgets every note of its exchange; a crash or a failed delivery during the pickup no longer loses the work — a pickup is at-least-once, not once: a crash between handing the turn over and the write that drops its note repeats that pickup once, because the `taken` set and the inbox that would tell the next start otherwise are process-local. That is the safe side of the trade, a repeated turn the user can see instead of work the user was told about going silently dead. Covered by `core/test/follow-through.test.js`.
 - A repeated Bot-to-Bot line becomes one turn (#641): `deliver()` keys the last 60 s of Bot-to-Bot text by sender + recipient Session + the text with its whitespace collapsed (`DEDUPE_MS`, `duplicateDelivery`), drops a repeat before it is recorded as an exchange or handed to the Session, and journals the drop as `delivery: dedupe <sender> -> <session>` in the host warning log. The key includes the Session, so the same line to another Session (a group copy, another Bot) is its own delivery; `deferred` / `resumed` / `retried` / `resent` carry on a message the Session already has and are never judged, and a friend's line, the user's own message and a group round are not Bot-to-Bot. Routing (parent and direct children) and the existing unread-queue refusal are unchanged, and the map is process-local and capped at 500 entries. Covered by `core/test/repeated-delivery.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. Covered by `core/test/work-dir.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 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`.
 
 ### Connectors: GitHub and the Composio gateway (#311)