Merge remote-tracking branch 'origin/worktree-agent-handle' into worktree-bash-owner-token

packages/acp/tests/dispose.spec.ts

@@ -273,4 +273,47 @@ describe('acp bridge — disposal & HMR safety', () => {
     expect(harness.ctx.sessions.get('guard-a')).toBeUndefined() // detach still ran
     await harness.dispose()
   })
+
+  it('concurrent AgentHandle dispose() calls all await the SAME teardown (memoized)', async () => {
+    // The handle's dispose() must memoize: the underlying cordis effect disposer
+    // is single-shot, so a second dispose() while the first is mid-teardown would
+    // otherwise resolve IMMEDIATELY (effect epoch already cleared) — before the
+    // first call's await agent.done + final flush finished. Every caller must
+    // observe the same quiescence boundary.
+    const harness = await makeBridgeHarness({ storageDir, script: ['hang'] })
+    const handle = harness.ctx.agents.create({
+      agentId: 'conc-a', sessionId: 'conc-a', agentOptions: { model: 'mock' },
+    })
+    // Drive a turn that hangs in the model stream, so the loop is mid-turn when
+    // disposed — its exit runs a final session/flush we can gate to hold the
+    // teardown observably in-flight.
+    handle.agent.send([{ type: 'text', text: 'go' }])
+    await new Promise(r => setTimeout(r, 30))
+    expect(handle.agent.status).toBe('running')
+    let releaseFlush!: () => void
+    const flushGate = new Promise<void>((resolve) => { releaseFlush = resolve })
+    harness.ctx.on('session/flush', () => flushGate)
+
+    // First dispose enters teardown (aborts the hanging step) and blocks in the
+    // gated final flush.
+    const first = handle.dispose()
+    let firstSettled = false
+    void first.then(() => { firstSettled = true })
+    await new Promise(r => setTimeout(r, 20))
+    expect(firstSettled).toBe(false)
+
+    // Second dispose MUST await the same in-flight teardown, not resolve early.
+    const second = handle.dispose()
+    let secondSettled = false
+    void second.then(() => { secondSettled = true })
+    await new Promise(r => setTimeout(r, 20))
+    expect(secondSettled).toBe(false) // memoized: still pending with the first
+
+    // Release the flush; both resolve together and the session is gone.
+    releaseFlush()
+    await Promise.all([first, second])
+    expect(harness.ctx.agents.get('conc-a')).toBeUndefined()
+    expect(harness.ctx.sessions.get('conc-a')).toBeUndefined()
+    await harness.dispose()
+  })
 })

packages/agent-loop/src/index.ts

@@ -267,15 +267,25 @@ export class AgentLoop extends Service implements AgentFactory {
 
   /**
    * Build an {@link AgentHandle} for a PREPARED session + a fresh agent. The
-   * handle's `dispose()` just runs the composite effect's disposer (see
+   * handle's `dispose()` runs the composite effect's disposer (see
    * {@link start}) — which stops the loop, awaits its exit (final flush
    * captured), unregisters the agent, and detaches the session, in that order.
    * The same composite effect is what a fiber unload disposes, so both teardown
    * triggers honor the ordering identically.
+   *
+   * `dispose()` is MEMOIZED: the underlying cordis effect disposer is
+   * single-shot (a second call returns immediately because the effect's epoch is
+   * already cleared, NOT awaiting the in-flight teardown), so concurrent/repeated
+   * `dispose()` calls would otherwise resolve before the first call's
+   * `await agent.done` + final flush completed. Memoizing the promise makes every
+   * caller observe the SAME quiescence boundary, honoring the
+   * `AgentHandle.dispose(): Promise<void>` contract (mirrors the ACP `quiesce()`
+   * helper).
    */
   private startOwned(id: AgentId, options: AgentOptions, session: Session): AgentHandle {
     const { agent, disposeAgent } = this.start(id, options, session)
-    return { agent, dispose: disposeAgent }
+    let disposing: Promise<void> | undefined
+    return { agent, dispose: () => (disposing ??= disposeAgent()) }
   }
 }
 

packages/agent-loop/src/loop.ts

@@ -201,14 +201,22 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
 
     // Pre-step cancel (window 2): `setStatus('running')` emits `agent/status`
     // SYNCHRONOUSLY, so a `running` listener can `cancel()` in the gap between the
-    // check above and `runTurn`. `cancel()` already cleared the queued FIFO, so
-    // drop the turn before it starts (runTurn would otherwise throw on an empty
-    // queue) and transition back to idle — `running` was already emitted, so a
-    // real `idle` transition (which also settles waiters) balances the status.
+    // check above and `runTurn`. Mirror window 1: clear the marker, then
+    //   - if NOTHING new is queued, drop the about-to-run turn and transition
+    //     back to `idle` (`running` was already emitted, so a real idle
+    //     transition balances the status AND settles `whenIdle()` waiters);
+    //   - if a NEW prompt was queued AFTER the cancel (a `running` listener that
+    //     cancels then sends), the marker was for the cancelled work only — fall
+    //     through and run the new prompt's turn (status is already `running`), so
+    //     a `whenIdle()` waiter resolves on THAT turn's running→idle, not before
+    //     it runs. Settling here would resolve quiescence while the replacement
+    //     is still queued and unrun (the same early-resolve race window 1 fixes).
     if (handle.isCancelled()) {
       handle.clearCancel()
-      handle.setStatus('idle')
-      continue
+      if (!agent.inbox.hasQueued) {
+        handle.setStatus('idle')
+        continue
+      }
     }
 
     // Re-derive the turn number from the log each iteration (do NOT keep a local

packages/agent-loop/tests/cancel.spec.ts

@@ -252,6 +252,36 @@ describe('Agent.cancel()', () => {
     expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
   })
 
+  it('window 2: whenIdle() does NOT resolve early when a running listener cancels then queues replacement work', async () => {
+    // The window-1 early-resolve race has a window-2 twin: a synchronous
+    // agent/status('running') listener cancels the about-to-run turn AND queues a
+    // replacement. window 2 must NOT settle waiters (via setStatus('idle')) while
+    // the replacement is still queued-and-unrun — it must fall through and run it,
+    // so whenIdle() resolves on the replacement turn's running→idle, not before.
+    const adapter = new MockAdapter([textResponse('A reply'), textResponse('B reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    let replaced = false
+    const dispose = ctx.on('agent/status', (subject, status) => {
+      if (subject !== agent || status !== 'running' || replaced) return
+      replaced = true
+      agent.cancel('drop A')
+      send(agent, 'B')
+    })
+
+    send(agent, 'A')
+    const idle = agent.whenIdle()
+    await idle
+    dispose()
+
+    // whenIdle() resolved only AFTER B's turn ran: B's user message + a turn/end
+    // are in the log, and A was dropped.
+    expect(userTexts(agent)).toContain('B')
+    expect(userTexts(agent)).not.toContain('A')
+    expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)
+  })
+
   it('whenIdle() does NOT resolve early when a new prompt is queued during a pre-step cancel', async () => {
     // The subtle race: a whenIdle() waiter is registered for prompt A; cancel()
     // clears A; prompt B is queued BEFORE the loop resumes from the idle wait.

packages/session-persistence/src/coordinator.ts

@@ -2,21 +2,20 @@
  * The backend-agnostic write-path orchestration shared by every first-party
  * {@link SessionPersistence} backend.
  *
- * The two durable backends (`dsh-session-persistence-jsonl` over file bytes,
- * `dsh-session-persistence-sqlite` over `node:sqlite` rows) were byte-identical
- * — or same-algorithm — for ALL of their orchestration: the in-memory
- * bookkeeping (the per-id state, the write-behind buffers, the per-id
- * serialization chains, the per-session init promises), the `session/event` →
- * buffer → `session/flush` drain, lazy materialization, crash-tail repair on
- * load, the four `session/created` adoption cases (new / HMR-adopt / collision /
- * ownerless-claim), and dispose-time quiescence. Only the STORAGE primitives
- * differed (write bytes vs. INSERT rows). {@link PersistenceCoordinator} owns
- * the orchestration once; a backend supplies the storage primitives as a small
+ * Every durable backend needs the same orchestration: the in-memory bookkeeping
+ * (the per-id state, the write-behind buffers, the per-id serialization chains,
+ * the per-session init promises), the `session/event` → buffer → `session/flush`
+ * drain, lazy materialization, crash-tail repair on load, the four
+ * `session/created` adoption cases (new / HMR-adopt / collision /
+ * ownerless-claim), and dispose-time quiescence. Only the STORAGE primitives are
+ * backend-specific (file bytes for `dsh-session-persistence-jsonl`, `node:sqlite`
+ * rows for `dsh-session-persistence-sqlite`). {@link PersistenceCoordinator} owns
+ * the orchestration; a backend supplies the storage primitives as a small
  * {@link PersistenceBackend} hook object.
  *
- * The abstract {@link SessionPersistence} service's public API is unchanged: a
- * backend still IS a `SessionPersistence` (its six public methods delegate to a
- * coordinator it composes), so a third-party backend MAY implement the service
+ * The abstract {@link SessionPersistence} service's public API is independent of
+ * this: a backend IS a `SessionPersistence` (its six public methods delegate to
+ * a coordinator it composes), so a third-party backend MAY implement the service
  * directly without using the coordinator at all.
  *
  * See the write-coordinator RFC (docs/rfc/implemented/2026-06-18-shared-persistence-write-coordinator.md)
@@ -115,7 +114,19 @@ interface SessionState {
   meta: SessionHeader
   /** The next seq the backend expects to append (the stored log length). */
   cursor: number
-  /** Whether the session has been physically materialized. */
+  /**
+   * Whether the backend has physically written this session (a JSONL file /
+   * SQLite row exists). `create()` registers state LAZILY — cursor 0,
+   * materialized false, nothing on disk — so an empty session leaves no
+   * artifact and the FIRST `appendBatch` writes the header + its events in ONE
+   * transaction (the "a row exists ⇔ it has events" invariant `has`/`list`
+   * rely on; a separate up-front materialize could crash leaving a row with
+   * zero events). The flag is the only signal that distinguishes a session
+   * registered-but-never-written from one durably present, which two callers
+   * need: `has()` (lazy-but-unwritten is not yet durable) and the reclaim path
+   * (an abandoned id with no artifact AND no buffered events is free to reuse;
+   * a materialized one is a real collision).
+   */
   materialized: boolean
   /**
    * The live Session this state was bound to via `onCreated`, if any. State
@@ -450,9 +461,18 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       if (tracked.owner === session) return
       if (tracked.owner === undefined) {
         // Ownerless state from the public create()/load() API. The FIRST live
-        // session claims it — but ONLY if its seed reproduces the persisted
-        // prefix (else a fresh, unrelated session reusing the id would have its
-        // seq 0..cursor-1 events filtered as already-written and grafted on).
+        // session claims it — but ONLY if BOTH the cwd scope and the seed match.
+        // The cwd guard mirrors case-2's cwd-scoped loadLive(): a same-id
+        // ownerless artifact at a DIFFERENT cwd is a collision, not a claim
+        // (claiming it would append the live cwd's events under the stored
+        // header's cwd, the exact cross-cwd corruption the loadLive scope
+        // prevents). The seed guard then ensures the live events reproduce the
+        // persisted prefix (else a fresh, unrelated session reusing the id would
+        // have its seq 0..cursor-1 events filtered as already-written and
+        // grafted on).
+        if (tracked.meta.cwd !== session.header.cwd) {
+          throw new Error(`session "${id}" is already persisted at a different cwd (persisted: ${String(tracked.meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`)
+        }
         if (!await this.seedMatchesPersisted(id, seed, tracked.cursor)) {
           throw new Error(`session "${id}" is already persisted with ${tracked.cursor} event(s) that do not match this live session (id collision)`)
         }

packages/session-persistence/tests/coordinator-contract.ts

@@ -68,6 +68,7 @@ export interface CoordinatorFixture {
 
 /** A constant absolute cwd; jsonl keys directories off it, memory/sqlite ignore it. */
 const WORK = '/w'
+const OTHER = '/other'
 
 /** The per-session init map a backend exposes for white-box init awaits. */
 function inits(persistence: SessionPersistence): Map<Session, Promise<void>> {
@@ -535,6 +536,58 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       }
     })
 
+    it('a live session at a DIFFERENT cwd cannot claim cursor-0 ownerless state (cwd scope)', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      try {
+        // create() registers ownerless state at cwd /a (cursor 0 — claims would
+        // otherwise match trivially on the seed).
+        await ctx.sessionPersistence.create(meta('wrong-cwd-claim', OTHER))
+        // A live session reusing the id but at cwd WORK must NOT claim it — the
+        // cwd scope is the fence (without it, WORK events would append under the
+        // OTHER header). Rejected as a collision.
+        const live = ctx.sessions.create('wrong-cwd-claim', { seed: oneTurnLog(), meta: { cwd: WORK } })
+        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
+    it('a live session at a DIFFERENT cwd cannot claim loaded-prefix ownerless state (cwd scope)', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      try {
+        // Materialize + load at cwd OTHER (ownerless, cursor = 6).
+        await ctx.sessionPersistence.create(meta('wrong-cwd-load', OTHER))
+        await ctx.sessionPersistence.append(SessionId('wrong-cwd-load'), oneTurnLog())
+        const { events } = await ctx.sessionPersistence.load(SessionId('wrong-cwd-load'))
+        // A live session whose SEED matches the loaded prefix but whose cwd is
+        // WORK must still be rejected — the cwd guard runs before the seed check.
+        const live = ctx.sessions.create('wrong-cwd-load', { seed: events, meta: { cwd: WORK } })
+        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
+    it('a no-cwd ownerless state cannot be claimed by a live session WITH a cwd (cwd scope, undefined side)', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      try {
+        // Ownerless state created WITHOUT a cwd (the no-cwd bucket).
+        await ctx.sessionPersistence.create(meta('no-cwd-state'))
+        // A live session reusing the id but WITH cwd WORK is a cwd mismatch
+        // (undefined vs WORK) and must be rejected.
+        const live = ctx.sessions.create('no-cwd-state', { seed: oneTurnLog(), meta: { cwd: WORK } })
+        await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/)
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
     // --- append adopts a storage-only session (fresh instance, no prior create/load) ---
 
     it('append adopts a storage-only session (fresh instance) and continues the seq', async () => {

packages/session/src/index.ts

@@ -285,14 +285,18 @@ export class SessionStore extends Service {
    * {@link announce}, so a throwing `session/created` listener rolls the attach
    * back instead of leaking it.
    *
-   * The id was already validated by {@link prepare}, which runs in the SAME
-   * synchronous sequence as `enter` (a config/factory caller does
-   * `prepare()` → `ctx.effect(generator)`, and a synchronous generator effect
-   * iterates inline — no await between them), so no concurrent create can claim
-   * the id in the gap. `enter` therefore does not re-check; it is not a public
-   * reservation primitive.
+   * Re-checks the id for a duplicate: `prepare` and `enter` are public
+   * cross-package primitives and a caller may interleave arbitrary work (or
+   * another create) between them, so a stale prepared session must NOT overwrite
+   * a live store entry of the same id — its detach disposer would later delete
+   * the REAL session. The {@link create} convenience and the agent factory call
+   * the two back-to-back so they never trip this, but the public seam cannot
+   * assume that.
+   *
+   * @throws if a session with this id is already in the store.
    */
   enter(session: Session): () => void {
+    if (this.store.has(session.id)) throw new Error(`session "${session.id}" already exists`)
     session.onAppend = (event) => { this.ctx.emit('session/event', session, event) }
     this.store.set(session.id, session)
     return () => {

packages/session/tests/session.spec.ts

@@ -221,6 +221,40 @@ describe('SessionStore', () => {
     expect(forked.deriveMessages()).toEqual(a.deriveMessages())
   })
 
+  it('enter() rejects a stale prepared session whose id is already live (no overwrite)', async () => {
+    // prepare()/enter() are public cross-package primitives that a caller may
+    // separate with arbitrary work. A stale prepared session must NOT overwrite
+    // a live store entry of the same id — its detach disposer would later delete
+    // the REAL session, breaking the store-uniqueness invariant.
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const stale = ctx.sessions.prepare('racy')
+    const live = ctx.sessions.create('racy')
+    expect(() => ctx.sessions.enter(stale)).toThrow(/already exists/)
+    // The live session is intact and still the store entry.
+    expect(ctx.sessions.get('racy')).toBe(live)
+  })
+
+  it('prepare() + enter() + announce() register a session and emit session/created', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const created: Session[] = []
+    ctx.on('session/created', session => void created.push(session))
+
+    const session = ctx.sessions.prepare('lifecycle')
+    // prepare alone does NOT enter the store.
+    expect(ctx.sessions.get('lifecycle')).toBeUndefined()
+    const detach = ctx.sessions.enter(session)
+    expect(ctx.sessions.get('lifecycle')).toBe(session)
+    // enter does NOT announce.
+    expect(created).toEqual([])
+    ctx.sessions.announce(session)
+    expect(created).toEqual([session])
+    // The detach disposer removes the entry + stops notification.
+    detach()
+    expect(ctx.sessions.get('lifecycle')).toBeUndefined()
+  })
+
   it('synthesizes a minimal v1 header for a bare-created session', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)