fix(host): apply the remaining inline state writers after the adopt (#1048)

changelog.d/1048-mutate-writers.fixed.md

@@ -0,0 +1 @@
+A window that was read-only only because the other dsh had the team open now takes the free lock for the writes that were still inline, instead of dropping them or refusing with the read-only error: a friend's first line and the unread mark, a team exchange, and every DSH Agent change (add, adopt, archive, remove). The change the other host left survives, because the write adopts and re-resolves its records before it applies ([#1048](https://github.com/ThinkFlowLab/FlyBot/issues/1048)).

core/src/host/agents.js

@@ -77,23 +77,29 @@ export function install(rt) {
     const readOnly = rt.readOnly()
     if (readOnly !== undefined) throw new Error(readOnly)
   }
+  // A change made outside a request may start in a window that was read-only only because
+  // the other dsh had the team open: refresh first — it adopts what that host left and
+  // takes the free lock (#1027, #1032) — then refuse if the lock is still someone else's.
+  const beforeWrite = async () => {
+    if (rt.readOnly() !== undefined) await rt.refresh()
+    writeCheck()
+  }
   const record = (agentId) => {
     const agent = state.agents[agentId]
     if (agent === undefined) throw new Error('Unknown DSH Agent')
     return agent
   }
 
   async function createAgentSession(agentId, workspaceId) {
-    writeCheck()
-    const agent = record(agentId)
+    record(agentId)
+    await beforeWrite()
     // A known workspace id wins (the Agent's latest Session may live outside the
     // team workspace after a workspace switch); anything else gets the team's own.
     const target = typeof workspaceId === 'string' && ctx.workspaceRegistry.get(workspaceId)
       ? workspaceId
       : await rt.ensureWorkspace()
     const { sessionId } = await ctx.sessionController.create({ workspaceId: target })
-    agent.sessions.push(sessionId)
-    await rt.save()
+    await rt.mutate(() => { record(agentId).sessions.push(sessionId) })
     // agent/created fired before the session joined the index, so restrict again now.
     await restrictTeamTools(await rt.resolveAgent(sessionId))
     return sessionId
@@ -103,48 +109,54 @@ export function install(rt) {
   // picker or a fork) is adopted into the same Agent: it keeps dsh's surface and
   // the team tools stay hidden.
   async function adoptAgentSession(agentId, sessionId) {
-    writeCheck()
-    const agent = record(agentId)
-    if (agent.sessions.includes(sessionId)) return sessionId
+    record(agentId)
+    await beforeWrite()
+    if (record(agentId).sessions.includes(sessionId)) return sessionId
     await rt.resolveAgent(sessionId)
-    const owned = rt.selfOf(sessionId) !== undefined || rt.botOf(sessionId) !== undefined
-      || rt.roomOf(sessionId) !== undefined || agentOf(sessionId) !== undefined
-    if (owned) throw new Error('Session already belongs to the team')
-    agent.sessions.push(sessionId)
-    await rt.save()
+    // The ownership check reads the records, so it runs inside the change, after the adopt.
+    await rt.mutate(() => {
+      const owned = rt.selfOf(sessionId) !== undefined || rt.botOf(sessionId) !== undefined
+        || rt.roomOf(sessionId) !== undefined || agentOf(sessionId) !== undefined
+      if (owned) throw new Error('Session already belongs to the team')
+      const agent = record(agentId)
+      if (!agent.sessions.includes(sessionId)) agent.sessions.push(sessionId)
+    })
     await restrictTeamTools(await rt.resolveAgent(sessionId))
     return sessionId
   }
 
   async function addAgent() {
-    writeCheck()
-    if (Object.keys(state.agents).length > 0) throw new Error('The team already has a DSH Agent')
-    const agent = { id: randomUUID(), name: AGENT_NAME, sessions: [], createdAt: Date.now() }
-    state.agents[agent.id] = agent
+    const id = randomUUID()
+    await rt.mutate(() => {
+      writeCheck()
+      if (Object.keys(state.agents).length > 0) throw new Error('The team already has a DSH Agent')
+      state.agents[id] = { id, name: AGENT_NAME, sessions: [], createdAt: Date.now() }
+    })
     try {
-      await createAgentSession(agent.id)
+      await createAgentSession(id)
     } catch (error) {
       // The first Session may already be created and saved: archive it and
       // persist the rollback, or the Agent returns on restart with its team
       // tools never denied.
-      const createdSessions = [...agent.sessions]
-      delete state.agents[agent.id]
+      const createdSessions = [...record(id).sessions]
+      await rt.mutate(() => { delete state.agents[id] }).catch(() => { /* rollback is best effort; the original error is what the caller sees */ })
       for (const sessionId of createdSessions) await ctx.workspaceRegistry.archiveSession(sessionId).catch(() => { /* rollback is best effort; the original error is what the caller sees */ })
-      await rt.save().catch(() => { /* rollback is best effort; the original error is what the caller sees */ })
       throw error
     }
-    return agent
+    return record(id)
   }
 
   async function archiveAgentSession(agentId, sessionId) {
-    writeCheck()
-    const agent = record(agentId)
-    if (!agent.sessions.includes(sessionId)) throw new Error('Session does not belong to the DSH Agent')
+    record(agentId)
+    await beforeWrite()
+    if (!record(agentId).sessions.includes(sessionId)) throw new Error('Session does not belong to the DSH Agent')
     // Archive first: the Session leaves the Agent only once it is actually gone,
     // so a failed archive stays retryable and keeps its tool restriction.
     await ctx.workspaceRegistry.archiveSession(sessionId)
-    agent.sessions = agent.sessions.filter(id => id !== sessionId)
-    await rt.save()
+    await rt.mutate(() => {
+      const agent = record(agentId)
+      agent.sessions = agent.sessions.filter(id => id !== sessionId)
+    })
   }
 
   async function renameAgentSession(agentId, sessionId, title) {
@@ -158,9 +170,9 @@ export function install(rt) {
   }
 
   async function removeAgent(agentId) {
-    writeCheck()
-    const agent = record(agentId)
-    const sessions = [...agent.sessions]
+    record(agentId)
+    await beforeWrite()
+    const sessions = [...record(agentId).sessions]
     const kept = []
     let failure
     for (const sessionId of sessions) {
@@ -174,12 +186,10 @@ export function install(rt) {
     // What could not be archived stays with the Agent, so it keeps its tool
     // restriction and the removal can be retried from the same place.
     if (kept.length > 0) {
-      agent.sessions = kept
-      await rt.save()
+      await rt.mutate(() => { record(agentId).sessions = kept })
       throw failure
     }
-    delete state.agents[agentId]
-    await rt.save()
+    await rt.mutate(() => { delete state.agents[agentId] })
   }
 
   Object.assign(rt, { agentOf, addAgent, createAgentSession, adoptAgentSession, archiveAgentSession, renameAgentSession, removeAgent })

core/src/host/delivery.js

@@ -101,19 +101,25 @@ export function install(rt) {
   // line it has not shown yet arrived (its row turns unread), and the senders seen so far, kept apart
   // from the roster, tell the first line from a sender. A group line is the group's, not the Bot's.
   const CONTACTS_KEPT = 200
-  const noteFriendLine = (to, sessionId, roomId, handle) => {
+  const noteFriendLine = async (to, sessionId, roomId, handle) => {
     if (to === undefined || roomId || (sessionId !== undefined && sessionId !== rt.chatOf(to.id))) return false
     const key = String(handle)
-    const contacts = (state.remoteContacts ??= {})[to.id] ??= {}
-    const first = contacts[key] === undefined
     const now = Date.now()
-    if (first) {
-      contacts[key] = now
-      const keys = Object.keys(contacts)
-      if (keys.length > CONTACTS_KEPT) for (const old of keys.slice(0, keys.length - CONTACTS_KEPT)) delete contacts[old]
-    }
-    to.friendLineAt = now
-    void rt.save().catch(() => {})
+    let first = false
+    // A change made outside a request: `mutate` adopts what the other host left before the
+    // change, so the Bot record is re-resolved inside the callback, never used from before it.
+    await rt.mutate(() => {
+      const bot = state.bots[to.id]
+      if (bot === undefined) return
+      const contacts = (state.remoteContacts ??= {})[bot.id] ??= {}
+      first = contacts[key] === undefined
+      if (first) {
+        contacts[key] = now
+        const keys = Object.keys(contacts)
+        if (keys.length > CONTACTS_KEPT) for (const old of keys.slice(0, keys.length - CONTACTS_KEPT)) delete contacts[old]
+      }
+      bot.friendLineAt = now
+    })
     return first
   }
 
@@ -166,7 +172,7 @@ export function install(rt) {
       return exchangeId
     }
     const origin = originOf({ from, to, fromId, remote, given: givenOrigin })
-    const firstContact = remote?.handle && !deferred && !resumed && !retried && !resent ? noteFriendLine(to, sessionId, roomId, remote.handle) : false
+    const firstContact = remote?.handle && !deferred && !resumed && !retried && !resent ? await noteFriendLine(to, sessionId, roomId, remote.handle) : false
     // Session events reject `undefined` members, so optional fields are omitted instead.
     const source = {
       kind: SOURCE_KIND,
@@ -194,8 +200,8 @@ export function install(rt) {
       ...(remote?.handle && Array.isArray(attachments) && attachments.length ? { attachments: attachments.map(file => ({ id: String(file.id ?? ''), name: String(file.name), size: Number(file.size), mime: String(file.mime ?? '') })) } : {}),
     }
     if (EXCHANGE_ROLES.has(role) && !deferred && !resumed && !retried && !resent) {
-      state.exchanges.push({ id: exchangeId, from: fromId, to: toId, text, role, time: Date.now(), ...(replyTo ? { replyTo } : {}), ...(audience === 'user' ? { audience } : {}) })
-      void rt.save().catch(() => {})
+      const exchange = { id: exchangeId, from: fromId, to: toId, text, role, time: Date.now(), ...(replyTo ? { replyTo } : {}), ...(audience === 'user' ? { audience } : {}) }
+      await rt.mutate(() => { state.exchanges.push(exchange) })
     }
     await rt.ensureCapabilities(targetSessionId)
     const agent = await resolveAgent(targetSessionId)

core/test/mutate-writers.test.js

@@ -0,0 +1,104 @@
+import assert from 'node:assert/strict'
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
+import { hostname, tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { createRuntime } from '../index.js'
+import { fakeContext } from './fake-ctx.js'
+
+// #1048: the writers that still change `state` inline and then `void rt.save()` must go
+// through `rt.mutate(fn)` instead. A window that was read-only only because the other dsh
+// had the team open takes the free lock on its next write (#1027), and that write must not
+// replace what the other host left with this window's older copy (#1032): the change goes
+// *after* the adopt, with every record re-resolved inside the callback.
+
+const TEAM = {
+  version: 1,
+  mainBotIds: ['chief'],
+  bots: { chief: { id: 'chief', sessionId: 'chief', name: 'Chief', role: 'Lead', instructions: '', createdAt: 1 } },
+  rooms: {},
+  agents: {},
+  exchanges: [],
+  exchangeTurns: {},
+  questions: {},
+  revision: 7,
+}
+
+const tempHome = (t) => {
+  const home = mkdtempSync(join(tmpdir(), 'ds-bot-mutate-writers-'))
+  t.after(() => rmSync(home, { recursive: true, force: true, maxRetries: 5 }))
+  return home
+}
+
+// A runtime playing the reading host: another live dsh holds the team, so this window
+// loads the team and cannot change it. `takeOver` is what the other host's exit looks
+// like — it saves once more, then closes, without this window having read since.
+async function readingWindow(t) {
+  const home = tempHome(t)
+  writeFileSync(join(home, 'state.json'), JSON.stringify(TEAM))
+  writeFileSync(join(home, 'state.lock'), JSON.stringify({ pid: process.ppid, host: hostname(), token: 'other' }))
+  const ctx = fakeContext()
+  const agents = new Map()
+  let sessions = 0
+  const agentFor = async (id) => {
+    if (agents.has(id)) return agents.get(id)
+    const agent = {
+      id,
+      session: { id },
+      ctx: { tools: { restrict: () => {} }, effect: execute => execute() },
+      followup: () => {},
+    }
+    agents.set(id, agent)
+    return agent
+  }
+  Object.assign(ctx, {
+    workspaceRegistry: { get: () => undefined, resolveByPath: async () => undefined, create: async path => ({ id: 'w1', path }), archiveSession: async () => {} },
+    sessionController: {
+      create: async () => ({ sessionId: `new${++sessions}` }),
+      resolveAgent: async id => ({ agent: await agentFor(id) }),
+    },
+    sessionTitle: { rename: async () => {} },
+    sessions: { flush: async () => {} },
+  })
+  const rt = createRuntime(ctx, { home, github: false, diagnosticsFile: false })
+  t.after(() => ctx.dispose())
+  await rt.load()
+  assert.match(rt.readOnly(), /open in another dsh/, 'the rig starts on the reading side')
+
+  // The other host writes a change this window has not seen, then closes.
+  writeFileSync(join(home, 'state.json'), JSON.stringify({ ...TEAM, revision: 8, prefs: { theme: 'light' } }))
+  rmSync(join(home, 'state.lock'))
+  const saved = () => JSON.parse(readFileSync(join(home, 'state.json'), 'utf8'))
+  return { ctx, rt, home, saved, sessionsMade: () => sessions }
+}
+
+test('#1048 delivery: a friend line written from a window that was read-only keeps what the other host left', async (t) => {
+  const { rt, saved } = await readingWindow(t)
+
+  // A friend's first line in the Bot's own chat: noteFriendLine records the contact and
+  // marks the row unread, then saves — an inline writer today.
+  // `flybotChecked`: the approval that let the line in; this test is about the write below it.
+  await rt.deliver({ toId: 'chief', role: 'request', text: 'hi', remote: { handle: '@friend.home', name: 'Friend' }, flybotChecked: true })
+  // The exchange record is the second inline writer in the same request.
+  await rt.deliver({ fromId: 'chief', toId: 'chief', role: 'reply', text: 'hello back' })
+  // The writers save without being awaited, so settle the write chain before reading.
+  await rt.save()
+
+  const file = saved()
+  assert.equal(file.prefs.theme, 'light', 'the change the other host left survives')
+  assert.equal(typeof file.remoteContacts.chief['@friend.home'], 'number', 'the first contact is written')
+  assert.equal(typeof file.bots.chief.friendLineAt, 'number', 'the unread mark is written')
+  assert.ok(file.exchanges.some(exchange => exchange.role === 'reply'), 'the exchange is written')
+})
+
+test('#1048 agents: adding the DSH Agent from a window that was read-only keeps what the other host left', async (t) => {
+  const { rt, saved, sessionsMade } = await readingWindow(t)
+
+  const agent = await rt.addAgent()
+
+  const file = saved()
+  assert.equal(file.prefs.theme, 'light', 'the change the other host left survives')
+  assert.equal(sessionsMade(), 1, 'one Session is created for the new Agent')
+  assert.deepEqual(file.agents[agent.id].sessions, ['new1'], 'the Agent and its Session are written')
+  assert.equal(rt.readOnly(), undefined, 'the free lock is taken over')
+})