Commit a part before its handoff turn can run

changelog.d/parts-commit-before-handoff.fixed.md

@@ -0,0 +1 @@
+A part switch now saves the new segment before the handoff turn starts, so an immediate failure can no longer start a second switch against an uncommitted part and lose the previous part's handoff note; a failed handoff still rolls back to the old part and keeps its input ([#751](https://github.com/ThinkFlowLab/FlyBot/issues/751)).

core/src/host/parts.js

@@ -785,35 +785,29 @@ export function install(rt) {
       .catch(error => rt.warn('ds-bot: permission for the fresh part of %s not applied: %s', bot.name, error))
     const segment = { sessionId: agent.id, endedAt: Date.now(), ...(tokens !== undefined ? { tokens } : {}), ...noted }
     const resume = job.resume || job.overflow
-    // Send the package to the fresh part BEFORE it becomes current: if that fails, the old part
-    // stays current and the stopped turn's input goes on there. A part that became current with no
-    // package and no transcript would lose the work it was meant to carry.
-    await rt.ensureCapabilities(freshId)
-    const fresh = await rt.resolveAgent(freshId)
-    const cwd = fresh.session.header?.cwd
-    const folder = cwd !== undefined && cwd !== session.header?.cwd ? cwd : undefined
-    // A fresh part that is not where it was made for would move on at every step.
-    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 })
-    try {
-      fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) }))
-      await ctx.sessions.flush(fresh.session)
-      freshParts.set(freshId, estimateTokens(text))
-    } catch (error) {
-      freshParts.delete(freshId)
-      await ctx.workspaceRegistry.archiveSession(freshId).catch(() => { /* best effort */ })
-      throw error
-    }
-    // The package is in the fresh part: point the Bot at it, atomically with the save. A failed
-    // save puts the old part back, so the finalizer hands the input on where it was.
+    // Commit the new part before its handoff turn can run: a handoff visible to a failure handler
+    // before the segment is saved would let a second switch lose the previous part's note. A failed
+    // handoff rolls the pointer back below, so the finalizer still hands the input on in the old part.
     const previousSession = bot.sessionId
     const previousSegments = bot.segments === undefined ? undefined : [...bot.segments]
     const previousRoomSessions = room === undefined ? undefined : { ...room.sessions }
     const previousRoomSegments = room?.segments?.[bot.id] === undefined ? undefined : [...room.segments[bot.id]]
     const previousQuestion = state.questions[agent.id]
+    const restore = () => {
+      if (room !== undefined) {
+        if (previousRoomSegments === undefined) delete room.segments[bot.id]
+        else room.segments[bot.id] = previousRoomSegments
+        if (previousRoomSessions === undefined) delete room.sessions[bot.id]
+        else room.sessions = previousRoomSessions
+      } else {
+        if (previousSegments === undefined) delete bot.segments
+        else bot.segments = previousSegments
+        bot.sessionId = previousSession
+        delete state.questions[freshId]
+        if (previousQuestion !== undefined) state.questions[agent.id] = previousQuestion
+      }
+      freshParts.delete(freshId)
+    }
     if (room !== undefined) {
       room.segments ??= {}
       ;(room.segments[bot.id] ??= []).push(segment)
@@ -832,19 +826,29 @@ export function install(rt) {
     try {
       await rt.save()
     } catch (error) {
-      if (room !== undefined) {
-        if (previousRoomSegments === undefined) delete room.segments[bot.id]
-        else room.segments[bot.id] = previousRoomSegments
-        if (previousRoomSessions === undefined) delete room.sessions[bot.id]
-        else room.sessions = previousRoomSessions
-      } else {
-        if (previousSegments === undefined) delete bot.segments
-        else bot.segments = previousSegments
-        bot.sessionId = previousSession
-        delete state.questions[freshId]
-        if (previousQuestion !== undefined) state.questions[agent.id] = previousQuestion
+      restore()
+      await ctx.workspaceRegistry.archiveSession(freshId).catch(() => { /* best effort */ })
+      throw error
+    }
+    let folder
+    try {
+      await rt.ensureCapabilities(freshId)
+      const fresh = await rt.resolveAgent(freshId)
+      const cwd = fresh.session.header?.cwd
+      folder = cwd !== undefined && cwd !== session.header?.cwd ? cwd : undefined
+      // A fresh part that is not where it was made for would move on at every step.
+      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)
       }
-      freshParts.delete(freshId)
+      const text = handoffPackage({ room, segment, lines: carried, number: number + 1, resume, overflow: job.overflow, failed: job.failed, bare: job.bare, folder })
+      fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) }))
+      await ctx.sessions.flush(fresh.session)
+      freshParts.set(freshId, estimateTokens(text))
+    } catch (error) {
+      // No package: the old part is current again, and the finalizer hands the input on there.
+      restore()
+      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
     }

docs/subsystems/core-host.md

@@ -224,7 +224,7 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi
 - `deliver()` (`src/host/delivery.js`) copies a cleaned, plugin-attested `remote.typedBy` onto the message `source.remote` for a friend's line to a local Bot; the client uses it to draw the person (#233 follow-up).
 - 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`.
 - 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 moves before it is committed** (`parts.js` `moveOn`): the handoff package is written to the fresh Session first, then the Bot's `sessionId` / `room.sessions` and the new segment are saved (a failed save restores the old state and archives the fresh Session), and the move is marked done only after the stopped turn's input and the old inbox messages are forwarded. A failure at any point leaves the old part current and hands that input on there; 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`.
+- **A part is committed before its handoff runs** (`parts.js` `moveOn`): the Bot's `sessionId` / `room.sessions` and the new segment are saved first, so a turn the handoff starts can never see an uncommitted part (a second switch would then lose the previous part's note). The handoff package is written next; if that fails, the pointer is rolled back — best effort saved, the fresh Session archived — so the old part is current and the finalizer hands the input on there. 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`.
 
 ### Connectors: GitHub and the Composio gateway (#311)
 

patches/ds-bot/0014-flybot-branding.patch

@@ -504,51 +504,20 @@ diff --git a/src/host/parts.js b/src/host/parts.js
    }
 
    async function runSwitch(agent, job) {
-@@ -775,6 +775,36 @@ export function install(rt) {
+@@ -775,6 +775,30 @@ export function install(rt) {
      await rt.applyPermission(bot, freshId)
        .catch(error => rt.warn('ds-bot: permission for the fresh part of %s not applied: %s', bot.name, error))
      const segment = { sessionId: agent.id, endedAt: Date.now(), ...(tokens !== undefined ? { tokens } : {}), ...noted }
 +    const resume = job.resume || job.overflow
-+    // Send the package to the fresh part BEFORE it becomes current: if that fails, the old part
-+    // stays current and the stopped turn's input goes on there. A part that became current with no
-+    // package and no transcript would lose the work it was meant to carry.
-+    await rt.ensureCapabilities(freshId)
-+    const fresh = await rt.resolveAgent(freshId)
-+    const cwd = fresh.session.header?.cwd
-+    const folder = cwd !== undefined && cwd !== session.header?.cwd ? cwd : undefined
-+    // A fresh part that is not where it was made for would move on at every step.
-+    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 })
-+    try {
-+      fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) }))
-+      await ctx.sessions.flush(fresh.session)
-+      freshParts.set(freshId, estimateTokens(text))
-+    } catch (error) {
-+      freshParts.delete(freshId)
-+      await ctx.workspaceRegistry.archiveSession(freshId).catch(() => { /* best effort */ })
-+      throw error
-+    }
-+    // The package is in the fresh part: point the Bot at it, atomically with the save. A failed
-+    // save puts the old part back, so the finalizer hands the input on where it was.
++    // Commit the new part before its handoff turn can run: a handoff visible to a failure handler
++    // before the segment is saved would let a second switch lose the previous part's note. A failed
++    // handoff rolls the pointer back below, so the finalizer still hands the input on in the old part.
 +    const previousSession = bot.sessionId
 +    const previousSegments = bot.segments === undefined ? undefined : [...bot.segments]
 +    const previousRoomSessions = room === undefined ? undefined : { ...room.sessions }
 +    const previousRoomSegments = room?.segments?.[bot.id] === undefined ? undefined : [...room.segments[bot.id]]
 +    const previousQuestion = state.questions[agent.id]
-     if (room !== undefined) {
-       room.segments ??= {}
-       ;(room.segments[bot.id] ??= []).push(segment)
-@@ -790,28 +790,35 @@ export function install(rt) {
-         delete state.questions[agent.id]
-       }
-     }
--    await rt.save()
-+    try {
-+      await rt.save()
-+    } catch (error) {
++    const restore = () => {
 +      if (room !== undefined) {
 +        if (previousRoomSegments === undefined) delete room.segments[bot.id]
 +        else room.segments[bot.id] = previousRoomSegments
@@ -562,6 +531,41 @@ diff --git a/src/host/parts.js b/src/host/parts.js
 +        if (previousQuestion !== undefined) state.questions[agent.id] = previousQuestion
 +      }
 +      freshParts.delete(freshId)
++    }
+     if (room !== undefined) {
+       room.segments ??= {}
+       ;(room.segments[bot.id] ??= []).push(segment)
+@@ -790,28 +790,45 @@ export function install(rt) {
+         delete state.questions[agent.id]
+       }
+     }
+-    await rt.save()
++    try {
++      await rt.save()
++    } catch (error) {
++      restore()
++      await ctx.workspaceRegistry.archiveSession(freshId).catch(() => { /* best effort */ })
++      throw error
++    }
++    let folder
++    try {
++      await rt.ensureCapabilities(freshId)
++      const fresh = await rt.resolveAgent(freshId)
++      const cwd = fresh.session.header?.cwd
++      folder = cwd !== undefined && cwd !== session.header?.cwd ? cwd : undefined
++      // A fresh part that is not where it was made for would move on at every step.
++      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 })
++      fresh.followup(createUserMessage({ content: [{ type: 'text', text }], source: handoffSource(job, resume) }))
++      await ctx.sessions.flush(fresh.session)
++      freshParts.set(freshId, estimateTokens(text))
++    } catch (error) {
++      // No package: the old part is current again, and the finalizer hands the input on there.
++      restore()
++      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
 +    }