Merge remote-tracking branch 'origin/master' into xtr/explicit-agent-context

# Conflicts:
#	packages/bundle/headless/tests/headless.spec.ts
#	packages/core/agent-loop/src/agent.ts
#	packages/subagent/subagent/tests/continuation.spec.ts

packages/api/session-controller/tests/commands-queue-attachment.host.spec.ts

@@ -1,6 +1,6 @@
 import { Context } from '@deepseek-ai/cordis'
-import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
-import type { Agent, ModelSelectionRef } from '@deepseek-ai/dsh-agent'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import type { Agent, Inbox, ModelSelectionRef } from '@deepseek-ai/dsh-agent'
 import { AttachmentError, AttachmentId } from '@deepseek-ai/dsh-attachment'
 import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
 import { createAssistantMessage, createUserMessage, MessageId } from '@deepseek-ai/dsh-llm'
@@ -13,6 +13,7 @@ import { subagentIdentityProjectionDefinition } from '@deepseek-ai/dsh-subagent/
 import { describe, expect, it, vi } from 'vitest'
 import { ApiSessionAgentController } from '../src/agent.ts'
 import { SessionCommandController } from '../src/commands.ts'
+import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit'
 import { installSessionReadTestServices, testSessionPersistence } from './test-remote.ts'
 
 async function commandHarness(
@@ -68,7 +69,7 @@ async function commandHarness(
       provider: 1,
     } as never)
   }
-  const inbox = new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
+  const inbox = createInboxStub()
   const steer = vi.fn((message: UserMessage) => { inbox.append('next-step', message) })
   const cancel = vi.fn()
   const agent = {
@@ -132,6 +133,14 @@ describe('Session queue commands', () => {
         }],
       },
     })), 'session/attachment-invalid')
+    for (const content of [[], [{ type: 'text' as const, text: ' \t\n' }]]) {
+      await expectFailure(Promise.resolve().then(() => controller.updateQueue({
+        sessionId: agent.id,
+        itemId: queued.id,
+        action: { kind: 'edit', content },
+      })), 'gateway/bad-request')
+    }
+    expect(inbox.nextTurn[0]?.content).toEqual([{ type: 'text', text: 'queued' }])
     await expectFailure(Promise.resolve().then(() => controller.updateQueue({
       sessionId: SessionId('missing'), itemId: queued.id, action: { kind: 'remove' },
     })), 'session/queue-item-not-found')

packages/api/session-controller/tests/commands-upload-file.host.spec.ts

@@ -1,5 +1,6 @@
 import { Context } from '@deepseek-ai/cordis'
-import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit'
 import type { Agent, ModelSelectionRef } from '@deepseek-ai/dsh-agent'
 import AttachmentStore, { AttachmentId } from '@deepseek-ai/dsh-attachment'
 import type {
@@ -38,7 +39,7 @@ async function uploadHarness(origin?: 'subagent'): Promise<{
   const session = ctx.sessions.create(SESSION, {
     meta: { cwd: '/workspace', ...(origin === undefined ? {} : { origin }) },
   })
-  const inbox = new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
+  const inbox = createInboxStub()
   const followup = vi.fn()
   const agent = {
     id: session.id,

packages/api/session-controller/tests/control-jobs.host.spec.ts

@@ -1,5 +1,5 @@
 import { Context } from '@deepseek-ai/cordis'
-import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { JobOutcome } from '@deepseek-ai/dsh-jobs'
 import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local'
@@ -9,6 +9,7 @@ import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
 import { describe, expect, it } from 'vitest'
 import { SessionControlController } from '../src/control.ts'
 import type { SessionControlFrame } from '../src/types.ts'
+import { unsupportedInbox } from '@deepseek-ai/dsh-agent-loop-testkit'
 
 type BaselineFrame = Extract<SessionControlFrame, { type: 'baseline' }>
 type JobFrame = Extract<SessionControlFrame, { type: 'jobs' }>
@@ -36,20 +37,28 @@ async function harness(withJobs: boolean): Promise<{
 }> {
   const ctx = new Context()
   await ctx.plugin(SessionStore)
-  await ctx.plugin(AgentRegistry)
   await ctx.plugin(SessionProjectionRegistry)
+  await ctx.plugin(AgentRegistry)
   if (withJobs) {
     await ctx.plugin(LocalJobRegistry)
     ctx.jobs.attachController('session-controller-test')
   }
   const session = ctx.sessions.create()
-  const agent = {
+  const agent: Agent = {
     id: session.id,
+    options: {},
     session,
-    inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }),
+    inbox: unsupportedInbox(),
     status: 'idle',
     ctx,
-  } as Agent
+    send: () => {},
+    followup: () => {},
+    steer: () => {},
+    inject: () => {},
+    cancel: () => {},
+    runMaintenance: task => task(new AbortController().signal),
+    whenIdle: () => Promise.resolve(),
+  }
   ctx.agents.register(agent)
   const control = new SessionControlController(ctx)
   await new Promise(resolve => setTimeout(resolve, 0))

packages/api/session-controller/tests/control-queue.host.spec.ts

@@ -1,11 +1,20 @@
 import { Context } from '@deepseek-ai/cordis'
-import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
-import type { Agent } from '@deepseek-ai/dsh-agent'
+import type { Agent, Inbox } from '@deepseek-ai/dsh-agent'
 import { createUserMessage } from '@deepseek-ai/dsh-llm'
-import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
-import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
-import { describe, expect, it } from 'vitest'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import { afterEach, describe, expect, it } from 'vitest'
 import { SessionControlController } from '../src/control.ts'
+import type { SessionControlFrame } from '../src/types.ts'
+import {
+  mountAgentLoopTestDependencies,
+  mountAgentLoopTestHarness,
+} from '@deepseek-ai/dsh-agent-loop-testkit'
+
+const ownedContexts = new Set<Context>()
+afterEach(async () => {
+  await Promise.all([...ownedContexts].map(ctx => ctx.fiber.dispose()))
+  ownedContexts.clear()
+})
 
 async function harness(): Promise<{
   ctx: Context
@@ -14,14 +23,11 @@ async function harness(): Promise<{
   inbox: Inbox
 }> {
   const ctx = new Context()
-  await ctx.plugin(SessionStore)
-  await ctx.plugin(AgentRegistry)
-  await ctx.plugin(SessionProjectionRegistry)
-  const session = ctx.sessions.create(SessionId('queue-session'))
-  const inbox = new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
-  const agent = { id: session.id, session, inbox, status: 'running', ctx } as Agent
-  ctx.agents.register(agent)
-  return { ctx, control: new SessionControlController(ctx), agent, inbox }
+  ownedContexts.add(ctx)
+  await mountAgentLoopTestDependencies(ctx)
+  const loop = await mountAgentLoopTestHarness(ctx)
+  const agent = await loop.create(SessionId('queue-session'))
+  return { ctx, control: new SessionControlController(ctx), agent, inbox: agent.inbox }
 }
 
 function message(text: string, source: 'user' | 'plugin' = 'user') {
@@ -32,6 +38,17 @@ function message(text: string, source: 'user' | 'plugin' = 'user') {
 }
 
 describe('Session control queue projection', () => {
+  /** Consume frames until the next queue replacement (inbox projection frames interleave). */
+  async function nextQueueFrame(
+    iterator: AsyncIterator<SessionControlFrame>,
+  ): Promise<Extract<SessionControlFrame, { type: 'queue' }>> {
+    for (;;) {
+      const next = await iterator.next()
+      if (next.done) throw new Error('stream ended before a queue frame')
+      if (next.value.type === 'queue') return next.value
+    }
+  }
+
   it('projects both pending lists in baselines and live replacement frames', async () => {
     const { control, inbox } = await harness()
     const queued = message('queued')
@@ -59,13 +76,41 @@ describe('Session control queue projection', () => {
 
     const replacement = message('replacement')
     inbox.append('next-turn', replacement)
-    const replaced = await iterator.next()
-    if (replaced.done || replaced.value.type !== 'queue') throw new Error('missing queue replacement')
-    expect(replaced.value.items.map(item => item.id)).toContain(replacement.id)
+    const replaced = await nextQueueFrame(iterator)
+    expect(replaced.items.map(item => item.id)).toContain(replacement.id)
     inbox.remove(steering.id)
-    const removed = await iterator.next()
-    if (removed.done || removed.value.type !== 'queue') throw new Error('missing queue replacement')
-    expect(removed.value.items.map(item => item.id)).not.toContain(steering.id)
+    const removed = await nextQueueFrame(iterator)
+    expect(removed.items.map(item => item.id)).not.toContain(steering.id)
+
+    abort.abort()
+    await iterator.next()
+  })
+
+  it('derives queue replacements from the completed projection regardless of registration order', async () => {
+    const ctx = new Context()
+    ownedContexts.add(ctx)
+    await mountAgentLoopTestDependencies(ctx)
+    const loop = await mountAgentLoopTestHarness(ctx)
+    const control = new SessionControlController(ctx)
+    const agent = await loop.create(SessionId('late-projection-queue'))
+    const { inbox } = agent
+    const abort = new AbortController()
+    const iterator = control.control(abort.signal)[Symbol.asyncIterator]()
+    await iterator.next()
+    const pending = message('late projection')
+
+    inbox.append('next-turn', pending)
+
+    await expect(iterator.next()).resolves.toMatchObject({
+      value: {
+        type: 'projection',
+        key: 'inbox',
+        value: { 'next-turn': [{ id: pending.id }], 'next-step': [] },
+      },
+    })
+    await expect(nextQueueFrame(iterator)).resolves.toMatchObject({
+      items: [{ id: pending.id, placement: 'queued' }],
+    })
 
     abort.abort()
     await iterator.next()
@@ -133,14 +178,19 @@ describe('Session control queue projection', () => {
     const { ctx, control, inbox } = await harness()
     const iterator = control.control(new AbortController().signal)[Symbol.asyncIterator]()
     await iterator.next()
-    inbox.append('next-turn', message('first'))
-    inbox.append('next-turn', message('second'))
+    const first = message('first')
+    const second = message('second')
+    inbox.append('next-turn', first)
+    inbox.append('next-turn', second)
 
-    const first = await iterator.next()
-    expect(first).toMatchObject({ done: false, value: { type: 'queue' } })
+    const queues: Extract<SessionControlFrame, { type: 'queue' }>[] = []
+    ownedContexts.delete(ctx)
     await ctx.fiber.dispose()
-    const second = await iterator.next()
-    expect(second).toMatchObject({ done: false, value: { type: 'queue' } })
-    await expect(iterator.next()).resolves.toMatchObject({ done: true })
+    for (;;) {
+      const next = await iterator.next()
+      if (next.done) break
+      if (next.value.type === 'queue') queues.push(next.value)
+    }
+    expect(queues.map(queue => queue.items.map(item => item.id))).toEqual([[first.id], [first.id, second.id]])
   })
 })

packages/api/session-controller/tests/session-cold.host.spec.ts

@@ -8,14 +8,16 @@ import { SESSION_FORMAT_VERSION, SessionLogOffset, SessionSeq } from '@deepseek-
 import { describe, expect, it, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import SessionStore from '@deepseek-ai/dsh-session'
-import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
 import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
 import { subagentIdentityProjectionDefinition } from '@deepseek-ai/dsh-subagent/src/projection.ts'
 import TypertRegistry from '@deepseek-ai/dsh-typert-registry'
 import { createUserMessage, MessageId } from '@deepseek-ai/dsh-llm'
 import { snapshotSubagentDescriptor } from '@deepseek-ai/dsh-subagent'
-import type { Agent } from '@deepseek-ai/dsh-agent'
-import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
+import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit'
+import type { Agent, Inbox } from '@deepseek-ai/dsh-agent'
+import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
+import AttachmentStore from '@deepseek-ai/dsh-attachment'
 import type { SessionPromptRequest, SessionRequestId } from '../src/types.ts'
 import {
   SessionPersistenceRevision,
@@ -42,8 +44,8 @@ function promptRequest(
   }
 }
 
-function inboxFor(session: Session): Inbox {
-  return new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
+function inboxFor(): Inbox {
+  return createInboxStub()
 }
 
 function header(id: string, createdAt: number, extra: Partial<SessionHeader> = {}): SessionHeader {
@@ -545,7 +547,7 @@ describe('subagent ownership fence', () => {
     })
     const followup = vi.fn()
     const agent = {
-      id: session.id, session, inbox: inboxFor(session), status: 'idle', ctx, followup,
+      id: session.id, session, inbox: inboxFor(), status: 'idle', ctx, followup,
     } as unknown as Agent
     ctx.agents.register(agent)
     const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
@@ -566,7 +568,7 @@ describe('subagent ownership fence', () => {
     const session = ctx.sessions.create(sid('session-browser-zone'), { meta: { cwd: '/proj' } })
     const followup = vi.fn()
     const agent = {
-      id: session.id, session, inbox: inboxFor(session), status: 'idle', ctx, followup,
+      id: session.id, session, inbox: inboxFor(), status: 'idle', ctx, followup,
     } as unknown as Agent
     ctx.agents.register(agent)
     const remote = createSessionTestRemote(ctx, {
@@ -677,6 +679,101 @@ describe('degenerate composition (no persistence, no factory)', () => {
 })
 
 describe('sessions.prompt synchronous rejection', () => {
+  it('rejects content without non-whitespace text or an attachment before delivery or Session events', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(AgentRegistry)
+    const session = ctx.sessions.create(sid('session-empty-prompt'))
+    const followup = vi.fn()
+    const steer = vi.fn()
+    ctx.agents.register({
+      id: session.id,
+      session,
+      inbox: inboxFor(),
+      status: 'idle',
+      ctx,
+      followup,
+      steer,
+    } as unknown as Agent)
+    const savedImage = {
+      attachmentId: 'accepted-image',
+      mediaType: 'image/png' as const,
+      bytes: 1,
+      width: 1,
+      height: 1,
+    }
+    const saveImages = vi.fn(() => Promise.resolve([savedImage]))
+    ctx.provide('attachments', Object.setPrototypeOf(
+      { saveImages },
+      AttachmentStore.prototype,
+    ) as never)
+    ctx.provide('llm', {
+      listProviders: () => [{ id: 'p', name: 'Provider' }],
+      resolveModelInfo: () => Promise.resolve({
+        provider: 'p', id: 'm', name: 'Model', inputModalities: ['text', 'image'],
+      }),
+    } as never)
+    const remote = createSessionTestRemote(ctx, {
+      defaultModelSelection: () => ({ provider: 'p', model: 'm' }),
+      cwd: '/tmp',
+    })
+    const initialEvents = session.snapshotEvents()
+    const rejectedContent: readonly SessionPromptRequest['content'][] = [
+      [],
+      [{ type: 'text', text: '' }],
+      [{ type: 'text', text: ' \t\n' }, { type: 'text', text: '' }],
+    ]
+
+    for (const [index, content] of rejectedContent.entries()) {
+      const response = await remote.prompt(promptRequest({
+        sessionId: session.id,
+        mode: index === 1 ? 'steer' : 'queue',
+        content,
+      }))
+      expect(response).toMatchObject({
+        ok: false,
+        error: {
+          code: 'gateway/bad-request',
+          message: 'prompt content must include non-whitespace text or an attachment',
+          details: {},
+        },
+      })
+    }
+    expect(followup).not.toHaveBeenCalled()
+    expect(steer).not.toHaveBeenCalled()
+    expect(session.snapshotEvents()).toEqual(initialEvents)
+
+    const queued = await remote.prompt(promptRequest({
+      sessionId: session.id,
+      mode: 'queue',
+      content: [{ type: 'text', text: ' queued ' }],
+    }))
+    const steered = await remote.prompt(promptRequest({
+      sessionId: session.id,
+      mode: 'steer',
+      content: [{ type: 'text', text: 'steered' }],
+    }))
+    const imageQueued = await remote.prompt(promptRequest({
+      sessionId: session.id,
+      mode: 'queue',
+      content: [{ type: 'image', mediaType: 'image/png', data: 'AQ==' }],
+    }))
+    expect(queued).toMatchObject({ ok: true, value: { accepted: true } })
+    expect(steered).toMatchObject({ ok: true, value: { accepted: true } })
+    expect(imageQueued).toMatchObject({ ok: true, value: { accepted: true } })
+    expect(followup).toHaveBeenCalledWith(expect.objectContaining({
+      content: [{ type: 'text', text: ' queued ' }],
+    }))
+    expect(steer).toHaveBeenCalledWith(expect.objectContaining({
+      content: [{ type: 'text', text: 'steered' }],
+    }))
+    expect(saveImages).toHaveBeenCalledOnce()
+    expect(followup).toHaveBeenCalledWith(expect.objectContaining({
+      content: [{ type: 'image', attachment: savedImage }],
+    }))
+    await ctx.fiber.dispose()
+  })
+
   it('maps a synchronous send throw (disposed/invalid input) to agent-busy with the reason attached', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -687,7 +784,7 @@ describe('sessions.prompt synchronous rejection', () => {
     ctx.agents.register({
       id: session.id,
       session,
-      inbox: inboxFor(session),
+      inbox: inboxFor(),
       status: 'idle',
       ctx,
       followup: () => { throw new Error('agent "session-throwing" lifecycle disposed') },

packages/api/session-controller/tests/session-projections.host.spec.ts

@@ -7,19 +7,18 @@
  * pushed through the control stream.
  */
 
-import { describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { mkdtemp, readFile, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { Context } from '@deepseek-ai/cordis'
 import { z } from 'zod'
-import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
 import { AttachmentStore } from '@deepseek-ai/dsh-attachment'
 import { agentPresetProjectionDefinition } from '@deepseek-ai/dsh-agent-presets'
-import type { Agent } from '@deepseek-ai/dsh-agent'
 import { createUserMessage } from '@deepseek-ai/dsh-llm'
-import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionSeq } from '@deepseek-ai/dsh-session'
-import type { Session } from '@deepseek-ai/dsh-session'
+import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
+import type { Session, SessionEvent, SessionHeader, UserMessage } from '@deepseek-ai/dsh-session'
 import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
 import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
 import SessionProjectionCache, { projectionCacheDomainSpec } from '@deepseek-ai/dsh-session-projection-cache'
@@ -27,8 +26,19 @@ import Storage from '@deepseek-ai/dsh-storage'
 import * as StorageDomain from '@deepseek-ai/dsh-storage-domain'
 import * as StorageJson from '@deepseek-ai/dsh-storage-json'
 import type { SessionControlFrame, SessionFollowFrame } from '@deepseek-ai/dsh-api-session-controller/types'
+import {
+  mountAgentLoopTestDependencies,
+  mountAgentLoopTestHarness,
+} from '@deepseek-ai/dsh-agent-loop-testkit'
 import { createSessionTestRemote, testSessionPersistence, type TestSessionRemote } from './test-remote.ts'
 
+const ownedContexts = new Set<Context>()
+afterEach(async () => {
+  await Promise.all([...ownedContexts].map(ctx => ctx.fiber.dispose()))
+  ownedContexts.clear()
+})
+let nextHarnessSession = 1
+
 declare module '@deepseek-ai/dsh-session-projection/types' {
   interface SessionProjectionStateMap {
     'test/last-user': LastUserState
@@ -108,15 +118,35 @@ const privatePromptUnit = () => ({
   stateVersion: 1,
 }) satisfies ProjectionDefinition<'test/private-prompt', string | null>
 
-async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: Session }> {
+async function harness(withRegistry: boolean): Promise<{
+  ctx: Context
+  session: Session
+  readonly claim: (target: 'next-turn' | 'next-step') => UserMessage[]
+}> {
   const ctx = new Context()
-  await ctx.plugin(SessionStore)
-  await ctx.plugin(AgentRegistry)
-  if (withRegistry) await ctx.plugin(SessionProjectionRegistry)
-  const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
-  // The gateway reads both the session and durable inbox baseline.
-  ctx.agents.register({ id: session.id, session, inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }), status: 'idle', ctx } as Agent)
-  return { ctx, session }
+  ownedContexts.add(ctx)
+  if (!withRegistry) {
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(AgentRegistry)
+    const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
+    return {
+      ctx,
+      session,
+      claim: () => { throw new Error('inbox is unavailable without the projection registry') },
+    }
+  }
+  await mountAgentLoopTestDependencies(ctx)
+  const loop = await mountAgentLoopTestHarness(ctx)
+  const agent = await loop.create(
+    SessionId(`session-projections-${String(nextHarnessSession++)}`),
+    {},
+    { cwd: '/workspace' },
+  )
+  return {
+    ctx,
+    session: agent.session,
+    claim: target => loop.claim(agent, target, 1),
+  }
 }
 
 /** Append `count` user messages so the log has paginable message boundaries. */
@@ -205,6 +235,74 @@ describe('session.history projections block', () => {
     expect(last?.event.seq).toBe(projections.asOfSeq)
   })
 
+  it('reconstructs a cold persisted queue without publishing or resuming an Agent', async () => {
+    const { ctx } = await harness(true)
+    const coldId = SessionId('cold-persisted-queue')
+    const meta: SessionHeader = { version: SESSION_FORMAT_VERSION, id: coldId, createdAt: 1, cwd: '/tmp', isSeeded: false }
+    const message = createUserMessage({
+      content: [{ type: 'text', text: 'survive process restart' }],
+      source: { kind: 'user' },
+    })
+    const events: SessionEvent[] = [{
+      type: 'agent/inbox/spliced',
+      seq: SessionSeq(0),
+      time: 2,
+      data: { target: 'next-turn', start: 0, inserted: [message] },
+    }]
+    ctx.provide('sessionPersistence', testSessionPersistence(ctx, {
+      list: () => Promise.resolve([meta]),
+      inspect: () => Promise.resolve({ meta, events, inheritedEventCount: SessionLogOffset(0) }),
+    }) as never)
+    const snapshot = await opening(remote(ctx), coldId)
+
+    expect(snapshot.projections.values.inbox).toEqual({
+      'next-turn': [message],
+      'next-step': [],
+    })
+    expect(ctx.agents.get(coldId)).toBeUndefined()
+    expect(ctx.sessions.get(coldId)).toBeUndefined()
+  })
+
+  it('removes claimed steering from the pending Inbox projection immediately', async () => {
+    const { ctx, session, claim } = await harness(true)
+    const proxy = remote(ctx)
+    const message = createUserMessage({
+      content: [{ type: 'text', text: 'apply this now' }],
+      source: { kind: 'user' },
+    })
+    const agent = ctx.agents.get(session.id)
+    if (agent === undefined) throw new Error('missing Agent')
+    agent.inbox.append('next-step', message)
+    claim('next-step')
+
+    const during = await opening(proxy, session.id)
+    expect(during.projections.values.inbox).toEqual({
+      'next-turn': [],
+      'next-step': [],
+    })
+
+    session.append('user/message', message, { surfaceOp: 'append' })
+    const settled = await opening(proxy, session.id)
+    expect(settled.projections.values.inbox).toEqual({
+      'next-turn': [],
+      'next-step': [],
+    })
+
+    const rejected = createUserMessage({
+      content: [{ type: 'text', text: 'reject this pre-step' }],
+      source: { kind: 'user' },
+    })
+    session.append('turn/start', { turn: 1 })
+    agent.inbox.append('next-step', rejected)
+    claim('next-step')
+    session.append('turn/end', { turn: 1, reason: { kind: 'blocked' } })
+    const closed = await opening(proxy, session.id)
+    expect(closed.projections.values.inbox).toEqual({
+      'next-turn': [],
+      'next-step': [],
+    })
+  })
+
   it('returns a complete current replacement cut on each follow generation', async () => {
     const { ctx, session } = await harness(true)
     ctx.sessionProjections.register(lastUserUnit())