diff --git a/changelog.d/resume-finished-turn.fixed.md b/changelog.d/resume-finished-turn.fixed.md
new file mode 100644
index 000000000..2bb00840a
--- /dev/null
+++ b/changelog.d/resume-finished-turn.fixed.md
@@ -0,0 +1 @@
+A turn that finished while the plugin was not listening (a plugin reload, or a crash right after DSH recorded the end) is no longer picked up as "restarted while you were working" and repeated: the durable Session log decides, and only a turn the restart really cut is resumed.
diff --git a/changelog.d/resume-note-kept.fixed.md b/changelog.d/resume-note-kept.fixed.md
new file mode 100644
index 000000000..2b4d32916
--- /dev/null
+++ b/changelog.d/resume-note-kept.fixed.md
@@ -0,0 +1 @@
+A restart's pickup no longer loses the work it was about to resume when a delivery fails or the app stops during the pickup: the inflight notes stay until each turn or queued message is handed over successfully ([#747](https://github.com/ThinkFlowLab/FlyBot/issues/747)).
diff --git a/core/src/host/resume.js b/core/src/host/resume.js
index 0293d8c5d..3fdc5bb23 100644
--- a/core/src/host/resume.js
+++ b/core/src/host/resume.js
@@ -89,26 +89,33 @@ export function install(rt) {
     else turns[id] = next
     persist()
   }
+  // Every entry of that exchange: one exchange can be noted under two Sessions while a Session
+  // moves to a new part, and leaving one behind made the next start send the message again.
   const forgetQueued = (exchangeId) => {
+    const rest = {}
+    let dropped = false
     for (const [sessionId, list] of Object.entries(queued)) {
-      const at = list.findIndex(item => item.args.exchangeId === exchangeId)
-      if (at === -1) continue
-      const rest = list.toSpliced(at, 1)
-      queued = { ...queued }
-      if (rest.length === 0) delete queued[sessionId]
-      else queued[sessionId] = rest
-      persist()
-      return
+      const kept = list.filter(item => item.args.exchangeId !== exchangeId)
+      if (kept.length !== list.length) dropped = true
+      if (kept.length > 0) rest[sessionId] = kept
     }
+    if (!dropped) return
+    queued = rest
+    persist()
   }
 
   const deliver = rt.deliver
   rt.deliver = async (args) => {
     const noted = { ...args, exchangeId: args.exchangeId ?? randomUUID() }
     const sessionId = noted.sessionId ?? rt.chatOf(noted.toId)
-    // Noted before it is sent: a turn may take it up before the send returns.
-    queued = { ...queued, [sessionId]: [...(queued[sessionId] ?? []), { at: Date.now(), args: noted }].slice(-QUEUED_MAX) }
-    persist()
+    // Noted before it is sent: a turn may take it up before the send returns. A resend delivers an
+    // exchange the notes already hold, and noting it twice left the second copy on disk after the
+    // turn took it up; that copy was sent again by the next start.
+    const notedAlready = Object.values(queued).some(list => list.some(item => item.args.exchangeId === noted.exchangeId))
+    if (!notedAlready) {
+      queued = { ...queued, [sessionId]: [...(queued[sessionId] ?? []), { at: Date.now(), args: noted }].slice(-QUEUED_MAX) }
+      persist()
+    }
     try {
       return await deliver(noted)
     } catch (error) {
@@ -142,7 +149,7 @@ export function install(rt) {
       case 'turn/start': {
         const owner = rt.groupOwners.get(id)
         const botId = owner?.botId ?? rt.chatOwners.get(id)
-        update(id, () => ({ botId, ...(owner ? { roomId: owner.roomId } : {}), since: event.time ?? Date.now(), ...originNote(id) }))
+        update(id, () => ({ botId, ...(owner ? { roomId: owner.roomId } : {}), since: event.time ?? Date.now(), ...(event.data?.turn !== undefined ? { turn: event.data.turn } : {}), ...originNote(id) }))
         break
       }
       case 'user/message': {
@@ -159,7 +166,9 @@ export function install(rt) {
       case 'turn/end': {
         // DSH stops every turn as it shuts down; that one is picked up at the next start.
         const reason = event.data?.reason
-        if (reason?.kind === 'aborted' && reason.reason?.kind === 'disposed') break
+        // DSH stopping every turn as it shuts down, or the format layer recording a restart as
+        // `interrupted`: that one is picked up at the next start, so its note stays.
+        if ((reason?.kind === 'aborted' && reason.reason?.kind === 'disposed') || reason?.kind === 'interrupted') break
         update(id, () => undefined)
         break
       }
@@ -206,11 +215,50 @@ export function install(rt) {
       && current(item.args) !== undefined && taken?.has(item.args.exchangeId) !== true)
     if (fresh.length === 0) return
     const target = current(fresh[0].args)
+    // No Session to open yet: every item stays noted (seeded by resumeAll) for the next start.
     const agent = await rt.resolveAgent(target)
     const waiting = new Set([...(agent.inbox?.nextStep ?? []), ...(agent.inbox?.nextTurn ?? [])].map(message => message.source?.exchangeId))
-    for (const { args } of fresh) {
+    const failed = []
+    for (const item of fresh) {
+      const { args } = item
       if (waiting.has(args.exchangeId) || taken?.has(args.exchangeId) === true) continue
-      await rt.deliver({ ...args, sessionId: current(args), steer: false, resent: true })
+      try {
+        await rt.deliver({ ...args, sessionId: current(args), steer: false, resent: true })
+      } catch {
+        // The delivery wrapper dropped its note; put it back so the next start retries it.
+        failed.push(item)
+      }
+    }
+    if (failed.length > 0) {
+      queued = { ...queued, [sessionId]: [...(queued[sessionId] ?? []), ...failed].slice(-QUEUED_MAX) }
+      persist()
+      rt.warn('ds-bot: %d message(s) for %s were not sent again; kept for the next start', failed.length, sessionId)
+    }
+  }
+
+  // Did the noted turn really end while nobody was listening (a plugin reload, or a crash after
+  // DSH appended its end)? The durable Session log is the truth: a turn that finished ends
+  // `completed` (or `blocked` / `max-tokens` / `forked`, or is aborted by the user / hook / parent);
+  // only `interrupted`, an abort caused by `disposed` and a missing end describe work a restart
+  // cut. A note from a build before this change has no `turn`, so it keeps the old behaviour.
+  async function endedNormally(sessionId, record) {
+    if (!Number.isInteger(record.turn) || typeof ctx.sessionController?.page !== 'function') return false
+    try {
+      const agent = await rt.resolveAgent(sessionId)
+      const seq = Number(agent?.session?.seq)
+      const page = await ctx.sessionController.page({
+        address: { kind: 'session', sessionId },
+        ...(Number.isFinite(seq) && seq > 0 ? { throughSeq: seq - 1 } : {}),
+        maxMessages: 200,
+      }, AbortSignal.timeout(5000))
+      const end = (page?.records ?? []).map(entry => entry?.event)
+        .filter(event => event?.type === 'turn/end' && event.data?.turn === record.turn).at(-1)
+      const reason = end?.data?.reason
+      if (reason === undefined) return false
+      if (reason.kind === 'aborted') return ['user', 'parent', 'hook'].includes(reason.reason?.kind)
+      return ['completed', 'blocked', 'max-tokens', 'forked'].includes(reason.kind)
+    } catch {
+      return false
     }
   }
 
@@ -219,26 +267,41 @@ export function install(rt) {
     if (rt.readOnly() !== undefined) return
     const { turns: noted, queued: dropped } = interrupted
     interrupted = { turns: {}, queued: {} }
-    // Taken up now, so another restart does not pick them up twice.
-    persist()
+    // Keep everything still owed in the live notes until its pickup succeeds: clearing the file
+    // first meant one failed delivery (or a crash during the pickup) lost the note for good.
+    turns = { ...noted }
+    queued = { ...dropped }
     // Agents and inboxes are read before anything wakes their Sessions.
     const cut = []
     for (const [sessionId, record] of Object.entries(noted)) {
       const first = firstTurn.get(sessionId)
-      if (first === 'turn/end') continue
-      // Still running: noted again, as if it had started here.
-      if (first === undefined && (await rt.resolveAgent(sessionId).catch(() => undefined))?.status === 'running') update(sessionId, () => record)
-      else cut.push([sessionId, record])
+      // `turn/end` is the noted turn ending here; `turn/start` means another turn began, so the
+      // noted one must have ended too (a turn cannot start until the previous one ends).
+      if (first === 'turn/end' || first === 'turn/start') { update(sessionId, () => undefined); continue }
+      // Still running: the note stays, as if it had started here.
+      if (first === undefined && (await rt.resolveAgent(sessionId).catch(() => undefined))?.status === 'running') continue
+      // Finished while this plugin was not listening: do not repeat work that is already done.
+      if (await endedNormally(sessionId, record)) { update(sessionId, () => undefined); continue }
+      cut.push([sessionId, record])
     }
     firstTurn = undefined
     for (const [sessionId, list] of Object.entries(dropped)) {
-      if (!Array.isArray(list)) continue
+      if (!Array.isArray(list)) { delete queued[sessionId]; continue }
       await resendQueued(sessionId, list).catch(error => rt.warn('ds-bot: sending again what %s was sent failed: %s', sessionId, error))
     }
     taken = undefined
     for (const [sessionId, record] of cut) {
-      await resumeTurn(sessionId, record).catch(error => rt.warn('ds-bot: picking up the turn of %s failed: %s', sessionId, error))
+      // The note stays while the pickup runs: a failed delivery — or a crash in the middle of one —
+      // must not lose it. Only a pickup handed over drops it, and not when the resumed turn has
+      // written its own note meanwhile (that one describes the running turn, not this pickup).
+      try {
+        await resumeTurn(sessionId, record)
+        update(sessionId, current => (current === record ? undefined : current))
+      } catch (error) {
+        rt.warn('ds-bot: picking up the turn of %s failed: %s', sessionId, error)
+      }
     }
+    persist()
   }
 
   ctx.effect(() => {
diff --git a/core/test/follow-through.test.js b/core/test/follow-through.test.js
index 8704795ef..32285e083 100644
--- a/core/test/follow-through.test.js
+++ b/core/test/follow-through.test.js
@@ -3,7 +3,7 @@ import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'no
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { test } from 'node:test'
-import { apply } from '../index.js'
+import { createRuntime, provideFlyBot } from '../index.js'
 import { fakeContext } from './fake-ctx.js'
 import { stripWrittenCalls, writesToolCall } from '../src/host/turns.js'
 
@@ -27,9 +27,13 @@ const seed = () => ({
 // One DS Bot host on `home`; a second start on the same home is a restart.
 // Hosts by home: a home is removed only after its hosts stopped and their writes landed.
 const started = new Map()
+// The durable Session log per home (what DSH keeps across a plugin reload or a restart).
+const durable = new Map()
 
 async function start(home) {
   const ctx = fakeContext()
+  const log = durable.get(home) ?? []
+  durable.set(home, log)
   const agents = new Map()
   const agentOf = (sessionId) => {
     if (!agents.has(sessionId)) {
@@ -43,25 +47,37 @@ async function start(home) {
   Object.assign(ctx, {
     llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) },
     workspaceRegistry: { get: () => undefined, resolveByPath: async () => undefined, create: async path => ({ id: 'w1', path }), archiveSession: async () => {} },
-    sessionController: { create: async () => ({ sessionId: 's1' }), resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }), selectModel: async () => {} },
+    sessionController: {
+      create: async () => ({ sessionId: 's1' }),
+      resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }),
+      selectModel: async () => {},
+      // What the durable log holds for a Session (newest page; the test keeps all of it).
+      page: async ({ address } = {}) => ({ records: log.filter(entry => entry.id === address?.sessionId).map(entry => ({ event: entry.event })), 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 })
+  // The runtime itself, so a test can hold a write or a delivery open where the product has one.
+  const rt = createRuntime(ctx, { home })
+  provideFlyBot(ctx, rt.flybot)
   let seq = 0
   const emit = (id, type, data) => {
+    const event = { type, seq: ++seq, data }
+    log.push({ id, event })
     for (const listener of ctx.listeners.get('session/event') ?? []) {
-      if (listener.length >= 2) listener({ id }, { type, seq: ++seq, data })
+      if (listener.length >= 2) listener({ id }, event)
     }
   }
+  // Durable only: an event DSH wrote while no FlyBot host was listening.
+  const record = (id, type, data) => { log.push({ id, event: { type, seq: ++seq, data } }) }
   const tool = (name, args, sessionId, exec = {}) => ctx.registeredTools.get(name).execute(args, { agent: { id: sessionId }, ...exec })
   const askApproval = sessionId => ctx.listeners.get('approval/request')[0]({ agent: { session: { id: sessionId } }, toolName: 'pwsh', reason: 'escalate sandbox' }, () => 'from-dsh')
   const call = async (endpoint, payload) => (await route.fetch(new Request('http://localhost/api/bot', { method: 'POST', body: JSON.stringify({ endpoint, payload }) }))).json()
   await tool('list_bots', {}, 'a')
-  const host = { ctx, agentOf, emit, tool, call, askApproval, stop: () => ctx.dispose() }
+  const host = { ctx, rt, agentOf, emit, record, tool, call, askApproval, stop: () => ctx.dispose() }
   started.set(home, [...(started.get(home) ?? []), host])
   return host
 }
@@ -72,6 +88,7 @@ const freshHome = (t) => {
   t.after(async () => {
     await Promise.all((started.get(home) ?? []).map(host => host.stop()))
     started.delete(home)
+    durable.delete(home)
     rmSync(home, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 })
   })
   return home
@@ -81,6 +98,10 @@ const settle = async (check) => {
   while (!check() && Date.now() < until) await new Promise(resolve => setImmediate(resolve))
   assert.ok(check(), 'condition not reached')
 }
+// Writes to `inflight.json` land on a later tick; after a mock-timer tick and after `stop()` the file
+// can still hold the previous write. Wait for them rather than reading early (the note of a failed
+// pickup is only proved kept once the clearing write has landed and been replaced).
+const flushWrites = async (rounds = 20) => { for (let i = 0; i < rounds; i++) await new Promise(resolve => setImmediate(resolve)) }
 const text = message => message.content.map(block => block.text ?? '').join('\n')
 const notes = (home) => {
   try { return JSON.parse(readFileSync(join(home, 'inflight.json'), 'utf8')) } catch { return { turns: {}, queued: {} } }
@@ -195,6 +216,110 @@ test('when only DS Bot restarts, a turn still running or finished meanwhile is n
   assert.deepEqual([third.agentOf('b').calls[0].source.exchangeId, third.agentOf('a').calls.length], ['x1', 0])
 })
 
+test('a turn that finished while the plugin was not listening is not picked up', async (t) => {
+  t.mock.timers.enable({ apis: ['setTimeout'] })
+  const home = freshHome(t)
+  const first = await start(home)
+  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
+  await settle(() => inflight(home).a !== undefined)
+  // The turn ends while no FlyBot host is listening (a plugin reload, or a crash after DSH
+  // appended the end). Only the durable log knows; the note stays in inflight.json.
+  first.record('a', 'turn/end', { turn: 1, reason: { kind: 'completed' } })
+  await first.stop()
+  const second = await start(home)
+  t.mock.timers.tick(3000)
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(second.agentOf('a').calls.length, 0, 'the finished turn is not repeated')
+})
+
+test('a turn the restart truly cut (interrupted in the durable log) is still picked up', async (t) => {
+  t.mock.timers.enable({ apis: ['setTimeout'] })
+  const home = freshHome(t)
+  const first = await start(home)
+  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
+  await settle(() => inflight(home).a !== undefined)
+  first.record('a', 'turn/end', { turn: 1, reason: { kind: 'interrupted' } })
+  await first.stop()
+  const second = await start(home)
+  t.mock.timers.tick(3000)
+  await settle(() => second.agentOf('a').calls.length === 1)
+  assert.match(text(second.agentOf('a').calls[0]), /restarted while you were working/)
+})
+
+test('a pickup that fails keeps its note for the next start', async (t) => {
+  t.mock.timers.enable({ apis: ['setTimeout'] })
+  const home = freshHome(t)
+  const first = await start(home)
+  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
+  await settle(() => inflight(home).a !== undefined)
+  await first.stop()
+
+  // The second start cannot hand its pickup over. Hold it inside its delivery: a pickup that
+  // cleared its note first has an empty file at this moment, and a crash here loses the note.
+  const second = await start(home)
+  let release
+  const gate = new Promise(resolve => { release = resolve })
+  let entered = false
+  second.rt.deliver = async () => {
+    entered = true
+    await gate
+    throw new Error('the Session is not ready yet')
+  }
+  t.mock.timers.tick(3000)
+  await settle(() => entered)
+  await flushWrites()
+  assert.ok(inflight(home).a, 'the note is on disk while the pickup is in flight')
+  release()
+  await settle(() => inflight(home).a !== undefined)
+  await second.stop()
+  // `stop()` waits for the note writes, so this reads what is on disk: a note kept only in memory
+  // until the next change would be lost by a crash right after the failed pickup.
+  assert.ok(inflight(home).a, 'the note is on disk, not only in memory')
+
+  // The third start picks it up.
+  const third = await start(home)
+  t.mock.timers.tick(3000)
+  await settle(() => third.agentOf('a').calls.length === 1)
+  assert.match(text(third.agentOf('a').calls[0]), /restarted while you were working/)
+})
+
+test('a resumed turn that writes its own note keeps it', async (t) => {
+  t.mock.timers.enable({ apis: ['setTimeout'] })
+  const home = freshHome(t)
+  const first = await start(home)
+  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
+  await settle(() => inflight(home).a !== undefined)
+  await first.stop()
+
+  // The resumed turn starts before the pickup is handed over, so it notes the turn it runs now.
+  // Handing the pickup over must not clear that newer note.
+  const second = await start(home)
+  const deliver = second.rt.deliver
+  let delivered = false
+  second.rt.deliver = async args => {
+    if (!delivered) {
+      delivered = true
+      second.emit('a', 'turn/start', { turn: 7 })
+    }
+    return deliver(args)
+  }
+  t.mock.timers.tick(3000)
+  await settle(() => delivered)
+  await settle(() => second.agentOf('a').calls.length === 1)
+  await second.stop()
+  assert.equal(inflight(home).a?.turn, 7, 'the note the resumed turn wrote is the one that stays')
+})
+
+test('stop waits for the note it just wrote', async (t) => {
+  const home = freshHome(t)
+  const host = await start(home)
+  run(host, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
+  // No wait for the write: a stop that does not wait for it can leave a running turn unnoted, and
+  // the next start would not pick it up.
+  await host.stop()
+  assert.ok(inflight(home).a, 'the note of the running turn is on disk after stop')
+})
+
 test('a finished turn leaves nothing to pick up', async (t) => {
   t.mock.timers.enable({ apis: ['setTimeout'] })
   const home = freshHome(t)
@@ -246,6 +371,65 @@ test('a message a shutdown cancelled is sent again once; one still waiting, take
   assert.equal((await second.call('exchange', { a: 'a', b: 'b' })).value.length, 4)
 })
 
+test('a message sent again is forgotten everywhere once a turn takes it up', async (t) => {
+  const home = freshHome(t)
+  const first = await start(home)
+  await first.tool('message_bot', { to: 'Writer', message: 'Draft the plan', when: 'queue' }, 'a')
+  await settle(() => (notes(home).queued.b ?? []).length === 1)
+  const [plan] = first.agentOf('b').calls
+  const { exchangeId } = plan.source
+  await first.stop()
+
+  // The second start sends the noted message again, under the same exchange.
+  t.mock.timers.enable({ apis: ['setTimeout'] })
+  const second = await start(home)
+  t.mock.timers.tick(3000)
+  await settle(() => second.agentOf('b').calls.length === 1)
+  const [again] = second.agentOf('b').calls
+  assert.equal(again.source.exchangeId, exchangeId)
+  // The turn takes it up. Nothing of that exchange may stay noted: a second copy on disk made
+  // every later start send the message once more.
+  second.emit('b', 'user/message', { content: again.content, source: again.source })
+  const notedExchange = () => Object.values(notes(home).queued ?? {}).flat().filter(item => item?.args?.exchangeId === exchangeId)
+  await settle(() => notedExchange().length === 0)
+  await second.stop()
+  assert.equal(notedExchange().length, 0, 'the exchange a turn took up is still noted on disk')
+
+  // The third start has nothing to send.
+  const third = await start(home)
+  t.mock.timers.tick(3000)
+  await flushWrites()
+  assert.equal(third.agentOf('b').calls.length, 0, 'the message was sent a third time')
+})
+
+test('a live interrupted end does not drop the note of a turn still owed a pickup', async (t) => {
+  t.mock.timers.enable({ apis: ['setTimeout'] })
+  const home = freshHome(t)
+  const first = await start(home)
+  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
+  await settle(() => inflight(home).a !== undefined)
+  first.emit('a', 'turn/end', { turn: 1, reason: { kind: 'interrupted' } })
+  assert.notEqual(inflight(home).a, undefined, 'the note survives an interrupted end')
+  await first.stop()
+  const second = await start(home)
+  t.mock.timers.tick(3000)
+  await settle(() => second.agentOf('a').calls.length === 1)
+})
+
+test('a turn/start this process sees proves the noted turn ended, even when the log cannot', async (t) => {
+  t.mock.timers.enable({ apis: ['setTimeout'] })
+  const home = freshHome(t)
+  const first = await start(home)
+  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
+  await settle(() => inflight(home).a !== undefined)
+  await first.stop()                                  // the crash wrote no turn/end
+  const second = await start(home)
+  second.emit('a', 'turn/start', { turn: 2 })
+  t.mock.timers.tick(3000)
+  await new Promise(resolve => setImmediate(resolve))
+  assert.equal(second.agentOf('a').calls.length, 0)
+})
+
 const WRITTEN = 'Checking now.\n<tool_call>\n<function=pwsh>\n<parameter=command>\nGet-ChildItem E:\\\n</parameter>\n</function>\n</tool_call>'
 
 test('a tool call written as text gets one more turn on the same round slot', async (t) => {
diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md
index 9859774fb..7e68d2bcc 100644
--- a/docs/subsystems/core-host.md
+++ b/docs/subsystems/core-host.md
@@ -225,6 +225,7 @@ Work a member hands a Bot with `message_bot` from its group Session runs in a hi
 - A friend's Bot @-mentioned in a group never has its reply delivered back into that group (it lands in the friend's Bot's own chat), so the host posts one receipt line into the room per mention (`src/host/hooks.js` `mentionReceipt`, source role `mention`) and the client renders it as a muted event line (`src/client/cells.js`) (#80, #81). The role stays outside the room's round trigger (`groups.js` `opens`) and outside `RELAY_ROLES`, so the receipt opens no round and folds no turns. When that friend's Bot is already a current participant of the room, `sendUserMentions` skips the 1:1 `bot.send` entirely (#143): the human line in the room is the delivery, and Chief's grant is not consulted. A friend's line Bot Social attests they typed themselves (#227) arrives as `remoteChats.deliver({ human: true, ownerName })`: the message source carries `typedBy`, `turns.js` `claimTrigger` records the turn in `state.remoteTypedTurns[sessionId][turn]` (capped at 400 per Session, dropped with the part), `channel.js` publishes it as `roster.remoteTypedTurns` (narrowed per owner in `authz.js`), and the client draws that turn's reply as the person (`typedBy` in `src/client/text.js`, `PersonMessage` in `cells.js`). The flag is never derived from text. A failed 1:1 remote-chat send (`remote-chat.js` `forward` / `notSentLine`) is one short English line with a next step, never a raw `Bot Social send_as_member: …` preview. A local Bot's answer to a friend's Bot (`turns.js` `answer` → `flybotReplyRemote`, `role: 'reply'`, `via: 'auto-reply'`) that the send chain refuses is logged with its reason (#272): a `deny` resolves, it does not throw.
 - `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`.
+- **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 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`.
 
diff --git a/patches/ds-bot/0001-core-fixes.patch b/patches/ds-bot/0001-core-fixes.patch
index 6652cbbd5..dce955fe4 100644
--- a/patches/ds-bot/0001-core-fixes.patch
+++ b/patches/ds-bot/0001-core-fixes.patch
@@ -3227,7 +3227,7 @@ diff --git a/src/host/prompt.js b/src/host/prompt.js
 diff --git a/src/host/resume.js b/src/host/resume.js
 --- a/src/host/resume.js
 +++ b/src/host/resume.js
-@@ -9,6 +9,7 @@ import { readFileSync } from 'node:fs'
+@@ -9,6 +9,7 @@
  import { mkdir, writeFile } from 'node:fs/promises'
  import { join } from 'node:path'
  import { SOURCE_KIND } from './constants.js'
@@ -3235,7 +3235,7 @@ diff --git a/src/host/resume.js b/src/host/resume.js
  
  // Long enough for DSH to finish opening its workspaces after a start.
  const RESUME_DELAY_MS = 3000
-@@ -51,7 +51,7 @@ export function install(rt) {
+@@ -51,7 +52,7 @@
    // What the last run left; read before this run notes anything.
    let interrupted = readNotes(file)
  
@@ -3244,7 +3244,7 @@ diff --git a/src/host/resume.js b/src/host/resume.js
    let turns = {}
    /** @type {Record<string, {at:number, args:any}[]>} deliveries no turn has taken up, by Session */
    let queued = {}
-@@ -76,6 +76,10 @@ export function install(rt) {
+@@ -76,6 +77,10 @@
        await writeFile(file, snapshot, 'utf8')
      }).catch(error => rt.warn('ds-bot: noting running turns failed: %s', error))
    }
@@ -3255,12 +3255,57 @@ diff --git a/src/host/resume.js b/src/host/resume.js
    const update = (id, change) => {
      const next = change(turns[id])
      if (next === turns[id]) return
-@@ -137,10 +137,13 @@ export function install(rt) {
+@@ -84,26 +89,33 @@
+     else turns[id] = next
+     persist()
+   }
++  // Every entry of that exchange: one exchange can be noted under two Sessions while a Session
++  // moves to a new part, and leaving one behind made the next start send the message again.
+   const forgetQueued = (exchangeId) => {
++    const rest = {}
++    let dropped = false
+     for (const [sessionId, list] of Object.entries(queued)) {
+-      const at = list.findIndex(item => item.args.exchangeId === exchangeId)
+-      if (at === -1) continue
+-      const rest = list.toSpliced(at, 1)
+-      queued = { ...queued }
+-      if (rest.length === 0) delete queued[sessionId]
+-      else queued[sessionId] = rest
+-      persist()
+-      return
++      const kept = list.filter(item => item.args.exchangeId !== exchangeId)
++      if (kept.length !== list.length) dropped = true
++      if (kept.length > 0) rest[sessionId] = kept
+     }
++    if (!dropped) return
++    queued = rest
++    persist()
+   }
+ 
+   const deliver = rt.deliver
+   rt.deliver = async (args) => {
+     const noted = { ...args, exchangeId: args.exchangeId ?? randomUUID() }
+     const sessionId = noted.sessionId ?? rt.chatOf(noted.toId)
+-    // Noted before it is sent: a turn may take it up before the send returns.
+-    queued = { ...queued, [sessionId]: [...(queued[sessionId] ?? []), { at: Date.now(), args: noted }].slice(-QUEUED_MAX) }
+-    persist()
++    // Noted before it is sent: a turn may take it up before the send returns. A resend delivers an
++    // exchange the notes already hold, and noting it twice left the second copy on disk after the
++    // turn took it up; that copy was sent again by the next start.
++    const notedAlready = Object.values(queued).some(list => list.some(item => item.args.exchangeId === noted.exchangeId))
++    if (!notedAlready) {
++      queued = { ...queued, [sessionId]: [...(queued[sessionId] ?? []), { at: Date.now(), args: noted }].slice(-QUEUED_MAX) }
++      persist()
++    }
+     try {
+       return await deliver(noted)
+     } catch (error) {
+@@ -137,10 +149,13 @@
        case 'turn/start': {
          const owner = rt.groupOwners.get(id)
          const botId = owner?.botId ?? rt.chatOwners.get(id)
 -        update(id, () => ({ botId, ...(owner ? { roomId: owner.roomId } : {}), since: event.time ?? Date.now() }))
-+        update(id, () => ({ botId, ...(owner ? { roomId: owner.roomId } : {}), since: event.time ?? Date.now(), ...originNote(id) }))
++        update(id, () => ({ botId, ...(owner ? { roomId: owner.roomId } : {}), since: event.time ?? Date.now(), ...(event.data?.turn !== undefined ? { turn: event.data.turn } : {}), ...originNote(id) }))
          break
        }
        case 'user/message': {
@@ -3270,7 +3315,18 @@ diff --git a/src/host/resume.js b/src/host/resume.js
          if (event.data?.source?.kind !== SOURCE_KIND) break
          // turns.js, installed earlier, has already claimed the turn's trigger.
          const info = rt.live.get(id)?.info
-@@ -173,17 +173,17 @@ export function install(rt) {
+@@ -151,7 +166,9 @@
+       case 'turn/end': {
+         // DSH stops every turn as it shuts down; that one is picked up at the next start.
+         const reason = event.data?.reason
+-        if (reason?.kind === 'aborted' && reason.reason?.kind === 'disposed') break
++        // DSH stopping every turn as it shuts down, or the format layer recording a restart as
++        // `interrupted`: that one is picked up at the next start, so its note stays.
++        if ((reason?.kind === 'aborted' && reason.reason?.kind === 'disposed') || reason?.kind === 'interrupted') break
+         update(id, () => undefined)
+         break
+       }
+@@ -173,17 +190,17 @@
        // A turn already picked up once and cut off again stays stopped: a restart loop
        // must not run it forever.
        if (record.resumed) return
@@ -3291,6 +3347,110 @@ diff --git a/src/host/resume.js b/src/host/resume.js
    }
  
    // A hard stop keeps the inbox, which the Session runs once it wakes; only what DSH
+@@ -198,39 +215,93 @@
+       && current(item.args) !== undefined && taken?.has(item.args.exchangeId) !== true)
+     if (fresh.length === 0) return
+     const target = current(fresh[0].args)
++    // No Session to open yet: every item stays noted (seeded by resumeAll) for the next start.
+     const agent = await rt.resolveAgent(target)
+     const waiting = new Set([...(agent.inbox?.nextStep ?? []), ...(agent.inbox?.nextTurn ?? [])].map(message => message.source?.exchangeId))
+-    for (const { args } of fresh) {
++    const failed = []
++    for (const item of fresh) {
++      const { args } = item
+       if (waiting.has(args.exchangeId) || taken?.has(args.exchangeId) === true) continue
+-      await rt.deliver({ ...args, sessionId: current(args), steer: false, resent: true })
++      try {
++        await rt.deliver({ ...args, sessionId: current(args), steer: false, resent: true })
++      } catch {
++        // The delivery wrapper dropped its note; put it back so the next start retries it.
++        failed.push(item)
++      }
+     }
++    if (failed.length > 0) {
++      queued = { ...queued, [sessionId]: [...(queued[sessionId] ?? []), ...failed].slice(-QUEUED_MAX) }
++      persist()
++      rt.warn('ds-bot: %d message(s) for %s were not sent again; kept for the next start', failed.length, sessionId)
++    }
+   }
+ 
++  // Did the noted turn really end while nobody was listening (a plugin reload, or a crash after
++  // DSH appended its end)? The durable Session log is the truth: a turn that finished ends
++  // `completed` (or `blocked` / `max-tokens` / `forked`, or is aborted by the user / hook / parent);
++  // only `interrupted`, an abort caused by `disposed` and a missing end describe work a restart
++  // cut. A note from a build before this change has no `turn`, so it keeps the old behaviour.
++  async function endedNormally(sessionId, record) {
++    if (!Number.isInteger(record.turn) || typeof ctx.sessionController?.page !== 'function') return false
++    try {
++      const agent = await rt.resolveAgent(sessionId)
++      const seq = Number(agent?.session?.seq)
++      const page = await ctx.sessionController.page({
++        address: { kind: 'session', sessionId },
++        ...(Number.isFinite(seq) && seq > 0 ? { throughSeq: seq - 1 } : {}),
++        maxMessages: 200,
++      }, AbortSignal.timeout(5000))
++      const end = (page?.records ?? []).map(entry => entry?.event)
++        .filter(event => event?.type === 'turn/end' && event.data?.turn === record.turn).at(-1)
++      const reason = end?.data?.reason
++      if (reason === undefined) return false
++      if (reason.kind === 'aborted') return ['user', 'parent', 'hook'].includes(reason.reason?.kind)
++      return ['completed', 'blocked', 'max-tokens', 'forked'].includes(reason.kind)
++    } catch {
++      return false
++    }
++  }
++
+   async function resumeAll() {
+     await rt.load()
+     if (rt.readOnly() !== undefined) return
+     const { turns: noted, queued: dropped } = interrupted
+     interrupted = { turns: {}, queued: {} }
+-    // Taken up now, so another restart does not pick them up twice.
+-    persist()
++    // Keep everything still owed in the live notes until its pickup succeeds: clearing the file
++    // first meant one failed delivery (or a crash during the pickup) lost the note for good.
++    turns = { ...noted }
++    queued = { ...dropped }
+     // Agents and inboxes are read before anything wakes their Sessions.
+     const cut = []
+     for (const [sessionId, record] of Object.entries(noted)) {
+       const first = firstTurn.get(sessionId)
+-      if (first === 'turn/end') continue
+-      // Still running: noted again, as if it had started here.
+-      if (first === undefined && (await rt.resolveAgent(sessionId).catch(() => undefined))?.status === 'running') update(sessionId, () => record)
+-      else cut.push([sessionId, record])
++      // `turn/end` is the noted turn ending here; `turn/start` means another turn began, so the
++      // noted one must have ended too (a turn cannot start until the previous one ends).
++      if (first === 'turn/end' || first === 'turn/start') { update(sessionId, () => undefined); continue }
++      // Still running: the note stays, as if it had started here.
++      if (first === undefined && (await rt.resolveAgent(sessionId).catch(() => undefined))?.status === 'running') continue
++      // Finished while this plugin was not listening: do not repeat work that is already done.
++      if (await endedNormally(sessionId, record)) { update(sessionId, () => undefined); continue }
++      cut.push([sessionId, record])
+     }
+     firstTurn = undefined
+     for (const [sessionId, list] of Object.entries(dropped)) {
+-      if (!Array.isArray(list)) continue
++      if (!Array.isArray(list)) { delete queued[sessionId]; continue }
+       await resendQueued(sessionId, list).catch(error => rt.warn('ds-bot: sending again what %s was sent failed: %s', sessionId, error))
+     }
+     taken = undefined
+     for (const [sessionId, record] of cut) {
+-      await resumeTurn(sessionId, record).catch(error => rt.warn('ds-bot: picking up the turn of %s failed: %s', sessionId, error))
++      // The note stays while the pickup runs: a failed delivery — or a crash in the middle of one —
++      // must not lose it. Only a pickup handed over drops it, and not when the resumed turn has
++      // written its own note meanwhile (that one describes the running turn, not this pickup).
++      try {
++        await resumeTurn(sessionId, record)
++        update(sessionId, current => (current === record ? undefined : current))
++      } catch (error) {
++        rt.warn('ds-bot: picking up the turn of %s failed: %s', sessionId, error)
++      }
+     }
++    persist()
+   }
+ 
+   ctx.effect(() => {
 diff --git a/src/host/schedules.js b/src/host/schedules.js
 --- a/src/host/schedules.js
 +++ b/src/host/schedules.js
@@ -5245,6 +5405,270 @@ diff --git a/test/connectors.test.js b/test/connectors.test.js
 +  assert.equal(github.ok, false)
 +  assert.match(github.error.message, /no apps to pick/)
 +})
+diff --git a/test/follow-through.test.js b/test/follow-through.test.js
+--- a/test/follow-through.test.js
++++ b/test/follow-through.test.js
+@@ -3,7 +3,7 @@
+ import { tmpdir } from 'node:os'
+ import { join } from 'node:path'
+ import { test } from 'node:test'
+-import { apply } from '../index.js'
++import { createRuntime, provideFlyBot } from '../index.js'
+ import { fakeContext } from './fake-ctx.js'
+ import { stripWrittenCalls, writesToolCall } from '../src/host/turns.js'
+ 
+@@ -27,9 +27,13 @@
+ // One DS Bot host on `home`; a second start on the same home is a restart.
+ // Hosts by home: a home is removed only after its hosts stopped and their writes landed.
+ const started = new Map()
++// The durable Session log per home (what DSH keeps across a plugin reload or a restart).
++const durable = new Map()
+ 
+ async function start(home) {
+   const ctx = fakeContext()
++  const log = durable.get(home) ?? []
++  durable.set(home, log)
+   const agents = new Map()
+   const agentOf = (sessionId) => {
+     if (!agents.has(sessionId)) {
+@@ -43,25 +47,37 @@
+   Object.assign(ctx, {
+     llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) },
+     workspaceRegistry: { get: () => undefined, resolveByPath: async () => undefined, create: async path => ({ id: 'w1', path }), archiveSession: async () => {} },
+-    sessionController: { create: async () => ({ sessionId: 's1' }), resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }), selectModel: async () => {} },
++    sessionController: {
++      create: async () => ({ sessionId: 's1' }),
++      resolveAgent: async sessionId => ({ agent: agentOf(sessionId) }),
++      selectModel: async () => {},
++      // What the durable log holds for a Session (newest page; the test keeps all of it).
++      page: async ({ address } = {}) => ({ records: log.filter(entry => entry.id === address?.sessionId).map(entry => ({ event: entry.event })), 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 })
++  // The runtime itself, so a test can hold a write or a delivery open where the product has one.
++  const rt = createRuntime(ctx, { home })
++  provideFlyBot(ctx, rt.flybot)
+   let seq = 0
+   const emit = (id, type, data) => {
++    const event = { type, seq: ++seq, data }
++    log.push({ id, event })
+     for (const listener of ctx.listeners.get('session/event') ?? []) {
+-      if (listener.length >= 2) listener({ id }, { type, seq: ++seq, data })
++      if (listener.length >= 2) listener({ id }, event)
+     }
+   }
++  // Durable only: an event DSH wrote while no FlyBot host was listening.
++  const record = (id, type, data) => { log.push({ id, event: { type, seq: ++seq, data } }) }
+   const tool = (name, args, sessionId, exec = {}) => ctx.registeredTools.get(name).execute(args, { agent: { id: sessionId }, ...exec })
+   const askApproval = sessionId => ctx.listeners.get('approval/request')[0]({ agent: { session: { id: sessionId } }, toolName: 'pwsh', reason: 'escalate sandbox' }, () => 'from-dsh')
+   const call = async (endpoint, payload) => (await route.fetch(new Request('http://localhost/api/bot', { method: 'POST', body: JSON.stringify({ endpoint, payload }) }))).json()
+   await tool('list_bots', {}, 'a')
+-  const host = { ctx, agentOf, emit, tool, call, askApproval, stop: () => ctx.dispose() }
++  const host = { ctx, rt, agentOf, emit, record, tool, call, askApproval, stop: () => ctx.dispose() }
+   started.set(home, [...(started.get(home) ?? []), host])
+   return host
+ }
+@@ -72,6 +88,7 @@
+   t.after(async () => {
+     await Promise.all((started.get(home) ?? []).map(host => host.stop()))
+     started.delete(home)
++    durable.delete(home)
+     rmSync(home, { recursive: true, force: true, maxRetries: 5, retryDelay: 20 })
+   })
+   return home
+@@ -81,6 +98,10 @@
+   while (!check() && Date.now() < until) await new Promise(resolve => setImmediate(resolve))
+   assert.ok(check(), 'condition not reached')
+ }
++// Writes to `inflight.json` land on a later tick; after a mock-timer tick and after `stop()` the file
++// can still hold the previous write. Wait for them rather than reading early (the note of a failed
++// pickup is only proved kept once the clearing write has landed and been replaced).
++const flushWrites = async (rounds = 20) => { for (let i = 0; i < rounds; i++) await new Promise(resolve => setImmediate(resolve)) }
+ const text = message => message.content.map(block => block.text ?? '').join('\n')
+ const notes = (home) => {
+   try { return JSON.parse(readFileSync(join(home, 'inflight.json'), 'utf8')) } catch { return { turns: {}, queued: {} } }
+@@ -195,6 +216,110 @@
+   assert.deepEqual([third.agentOf('b').calls[0].source.exchangeId, third.agentOf('a').calls.length], ['x1', 0])
+ })
+ 
++test('a turn that finished while the plugin was not listening is not picked up', async (t) => {
++  t.mock.timers.enable({ apis: ['setTimeout'] })
++  const home = freshHome(t)
++  const first = await start(home)
++  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
++  await settle(() => inflight(home).a !== undefined)
++  // The turn ends while no FlyBot host is listening (a plugin reload, or a crash after DSH
++  // appended the end). Only the durable log knows; the note stays in inflight.json.
++  first.record('a', 'turn/end', { turn: 1, reason: { kind: 'completed' } })
++  await first.stop()
++  const second = await start(home)
++  t.mock.timers.tick(3000)
++  await new Promise(resolve => setImmediate(resolve))
++  assert.equal(second.agentOf('a').calls.length, 0, 'the finished turn is not repeated')
++})
++
++test('a turn the restart truly cut (interrupted in the durable log) is still picked up', async (t) => {
++  t.mock.timers.enable({ apis: ['setTimeout'] })
++  const home = freshHome(t)
++  const first = await start(home)
++  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
++  await settle(() => inflight(home).a !== undefined)
++  first.record('a', 'turn/end', { turn: 1, reason: { kind: 'interrupted' } })
++  await first.stop()
++  const second = await start(home)
++  t.mock.timers.tick(3000)
++  await settle(() => second.agentOf('a').calls.length === 1)
++  assert.match(text(second.agentOf('a').calls[0]), /restarted while you were working/)
++})
++
++test('a pickup that fails keeps its note for the next start', async (t) => {
++  t.mock.timers.enable({ apis: ['setTimeout'] })
++  const home = freshHome(t)
++  const first = await start(home)
++  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
++  await settle(() => inflight(home).a !== undefined)
++  await first.stop()
++
++  // The second start cannot hand its pickup over. Hold it inside its delivery: a pickup that
++  // cleared its note first has an empty file at this moment, and a crash here loses the note.
++  const second = await start(home)
++  let release
++  const gate = new Promise(resolve => { release = resolve })
++  let entered = false
++  second.rt.deliver = async () => {
++    entered = true
++    await gate
++    throw new Error('the Session is not ready yet')
++  }
++  t.mock.timers.tick(3000)
++  await settle(() => entered)
++  await flushWrites()
++  assert.ok(inflight(home).a, 'the note is on disk while the pickup is in flight')
++  release()
++  await settle(() => inflight(home).a !== undefined)
++  await second.stop()
++  // `stop()` waits for the note writes, so this reads what is on disk: a note kept only in memory
++  // until the next change would be lost by a crash right after the failed pickup.
++  assert.ok(inflight(home).a, 'the note is on disk, not only in memory')
++
++  // The third start picks it up.
++  const third = await start(home)
++  t.mock.timers.tick(3000)
++  await settle(() => third.agentOf('a').calls.length === 1)
++  assert.match(text(third.agentOf('a').calls[0]), /restarted while you were working/)
++})
++
++test('a resumed turn that writes its own note keeps it', async (t) => {
++  t.mock.timers.enable({ apis: ['setTimeout'] })
++  const home = freshHome(t)
++  const first = await start(home)
++  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
++  await settle(() => inflight(home).a !== undefined)
++  await first.stop()
++
++  // The resumed turn starts before the pickup is handed over, so it notes the turn it runs now.
++  // Handing the pickup over must not clear that newer note.
++  const second = await start(home)
++  const deliver = second.rt.deliver
++  let delivered = false
++  second.rt.deliver = async args => {
++    if (!delivered) {
++      delivered = true
++      second.emit('a', 'turn/start', { turn: 7 })
++    }
++    return deliver(args)
++  }
++  t.mock.timers.tick(3000)
++  await settle(() => delivered)
++  await settle(() => second.agentOf('a').calls.length === 1)
++  await second.stop()
++  assert.equal(inflight(home).a?.turn, 7, 'the note the resumed turn wrote is the one that stays')
++})
++
++test('stop waits for the note it just wrote', async (t) => {
++  const home = freshHome(t)
++  const host = await start(home)
++  run(host, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
++  // No wait for the write: a stop that does not wait for it can leave a running turn unnoted, and
++  // the next start would not pick it up.
++  await host.stop()
++  assert.ok(inflight(home).a, 'the note of the running turn is on disk after stop')
++})
++
+ test('a finished turn leaves nothing to pick up', async (t) => {
+   t.mock.timers.enable({ apis: ['setTimeout'] })
+   const home = freshHome(t)
+@@ -246,6 +371,65 @@
+   assert.equal((await second.call('exchange', { a: 'a', b: 'b' })).value.length, 4)
+ })
+ 
++test('a message sent again is forgotten everywhere once a turn takes it up', async (t) => {
++  const home = freshHome(t)
++  const first = await start(home)
++  await first.tool('message_bot', { to: 'Writer', message: 'Draft the plan', when: 'queue' }, 'a')
++  await settle(() => (notes(home).queued.b ?? []).length === 1)
++  const [plan] = first.agentOf('b').calls
++  const { exchangeId } = plan.source
++  await first.stop()
++
++  // The second start sends the noted message again, under the same exchange.
++  t.mock.timers.enable({ apis: ['setTimeout'] })
++  const second = await start(home)
++  t.mock.timers.tick(3000)
++  await settle(() => second.agentOf('b').calls.length === 1)
++  const [again] = second.agentOf('b').calls
++  assert.equal(again.source.exchangeId, exchangeId)
++  // The turn takes it up. Nothing of that exchange may stay noted: a second copy on disk made
++  // every later start send the message once more.
++  second.emit('b', 'user/message', { content: again.content, source: again.source })
++  const notedExchange = () => Object.values(notes(home).queued ?? {}).flat().filter(item => item?.args?.exchangeId === exchangeId)
++  await settle(() => notedExchange().length === 0)
++  await second.stop()
++  assert.equal(notedExchange().length, 0, 'the exchange a turn took up is still noted on disk')
++
++  // The third start has nothing to send.
++  const third = await start(home)
++  t.mock.timers.tick(3000)
++  await flushWrites()
++  assert.equal(third.agentOf('b').calls.length, 0, 'the message was sent a third time')
++})
++
++test('a live interrupted end does not drop the note of a turn still owed a pickup', async (t) => {
++  t.mock.timers.enable({ apis: ['setTimeout'] })
++  const home = freshHome(t)
++  const first = await start(home)
++  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
++  await settle(() => inflight(home).a !== undefined)
++  first.emit('a', 'turn/end', { turn: 1, reason: { kind: 'interrupted' } })
++  assert.notEqual(inflight(home).a, undefined, 'the note survives an interrupted end')
++  await first.stop()
++  const second = await start(home)
++  t.mock.timers.tick(3000)
++  await settle(() => second.agentOf('a').calls.length === 1)
++})
++
++test('a turn/start this process sees proves the noted turn ended, even when the log cannot', async (t) => {
++  t.mock.timers.enable({ apis: ['setTimeout'] })
++  const home = freshHome(t)
++  const first = await start(home)
++  run(first, 'a', 1, { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } })
++  await settle(() => inflight(home).a !== undefined)
++  await first.stop()                                  // the crash wrote no turn/end
++  const second = await start(home)
++  second.emit('a', 'turn/start', { turn: 2 })
++  t.mock.timers.tick(3000)
++  await new Promise(resolve => setImmediate(resolve))
++  assert.equal(second.agentOf('a').calls.length, 0)
++})
++
+ const WRITTEN = 'Checking now.\n<tool_call>\n<function=pwsh>\n<parameter=command>\nGet-ChildItem E:\\\n</parameter>\n</function>\n</tool_call>'
+ 
+ test('a tool call written as text gets one more turn on the same round slot', async (t) => {
 diff --git a/test/group-status.test.js b/test/group-status.test.js
 --- a/test/group-status.test.js
 +++ b/test/group-status.test.js
diff --git a/patches/ds-bot/topics.json b/patches/ds-bot/topics.json
index 6b95a9c3d..7f8777789 100644
--- a/patches/ds-bot/topics.json
+++ b/patches/ds-bot/topics.json
@@ -47,7 +47,8 @@
     "test/prompt.test.js",
     "src/client/browser-pane.js",
     "src/client/pane-width.js",
-    "src/client/panes.js"
+    "src/client/panes.js",
+    "test/follow-through.test.js"
   ],
   "0002-flybot-api-host": [
     "index.js",