fix(host): preserve subagent continuation ownership

packages/host/apiproxy/src/api-proxy.ts

@@ -504,10 +504,10 @@ function historyPage(
  * registry). An absent registry means the deployment has no projection seam:
  * the whole block is absent and clients treat every key as capability-absent.
  */
-function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | undefined {
+function projectionsFor(ctx: Context, session: Session): SessionProjectionsBlock | undefined {
   const registry = ctx.get('sessionProjections')
   if (registry === undefined) return undefined
-  return registry.snapshot(agent.session)
+  return registry.snapshot(session)
 }
 
 /**
@@ -639,6 +639,13 @@ async function catalogChild(
  */
 class SessionNotFound extends Error {}
 
+/** Session identity whose lifecycle belongs to subagent routing, not generic Host resume. */
+class SubagentSessionOwnership extends Error {
+  constructor(readonly sessionId: SessionId) {
+    super(`session "${sessionId}" is a subagent session; use subagent delivery`)
+  }
+}
+
 /** Requested identity already belongs to a session with another project cwd. */
 class SessionCwdConflict extends Error {
   constructor(
@@ -1008,28 +1015,70 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
     })
   }
 
+  /** Whether the session's own suffix carries the durable subagent discriminator. */
+  function hasSubagentDescriptor(session: Pick<Session, 'events' | 'header'>): boolean {
+    const ownStart = session.header.seedLength ?? 0
+    return session.events.slice(ownStart).some(event => event.type === 'subagent/descriptor')
+  }
+
   /**
-   * Gate the cold path on the store: an id absent from it, or naming a legacy
-   * log without a cwd (pre-release stance: not served, no compatibility), is
-   * not-found before any resume is attempted. With the gate passed, a later
-   * resume failure is genuinely internal. No persistence configured skips the
-   * gate — resume itself then fails loud with its own diagnostic.
+   * Generic Host interaction cannot claim a durably classified subagent or an
+   * Agent created through its live parent. The runtime-owner arm also covers
+   * descriptor-less child publication windows and older stored headers.
    */
-  async function assertServable(sessionId: SessionId): Promise<void> {
+  function hasSubagentOwner(
+    session: Pick<Session, 'events' | 'header'>,
+    agent: Agent | undefined,
+  ): boolean {
+    if (session.header.origin === 'subagent' || hasSubagentDescriptor(session)) return true
+    const parentId = session.header.parentSession
+    if (parentId === undefined || agent === undefined) return false
+    const parent = ctx.agents.get(parentId)
+    return parent !== undefined && ctx.agents.isOwnedBy(agent.id, parent)
+  }
+
+  /** Stable generic-Host error for an identity reserved to subagent routing. */
+  function subagentOwnershipError(sessionId: SessionId): RpcError {
+    return {
+      code: 'agent-busy',
+      message: `session "${sessionId}" is owned by subagent routing`,
+      details: { reason: 'use subagent delivery for this child session' },
+    }
+  }
+
+  /** Inspect one cold served session without repairing, resuming, or publishing it. */
+  async function inspectServable(sessionId: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
     const persistence = ctx.get('sessionPersistence')
-    if (persistence === undefined) return
+    if (persistence === undefined) {
+      throw new Error('session persistence is not configured (load a dsh-session-persistence backend)')
+    }
     const meta = (await persistence.list()).find(m => m.id === sessionId)
     if (meta === undefined || meta.cwd === undefined) throw new SessionNotFound(`session "${sessionId}" not found`)
+    const inspected = await persistence.inspect(sessionId)
+    if (inspected.meta.cwd === undefined) throw new SessionNotFound(`session "${sessionId}" not found`)
+    return inspected
   }
 
   async function agentFor(sessionId: SessionId): Promise<{ agent: Agent } | { error: RpcError }> {
+    const attached = ctx.sessions.get(sessionId)
     const live = ctx.agents.get(sessionId)
+    if (attached !== undefined && hasSubagentOwner(attached, live)) {
+      return { error: subagentOwnershipError(sessionId) }
+    }
     if (live !== undefined) return { agent: live }
     let resume = resumes.get(sessionId)
     if (resume === undefined) {
       resume = (async () => {
         try {
-          await assertServable(sessionId)
+          const inspected = await inspectServable(sessionId)
+          if (hasSubagentOwner({ header: inspected.meta, events: inspected.events }, undefined)) {
+            throw new SubagentSessionOwnership(sessionId)
+          }
+          const publishedSession = ctx.sessions.get(sessionId)
+          const publishedAgent = ctx.agents.get(sessionId)
+          if (publishedSession !== undefined && hasSubagentOwner(publishedSession, publishedAgent)) {
+            throw new SubagentSessionOwnership(sessionId)
+          }
           const handle = await ctx.agents.resume({
             resumeSessionId: sessionId,
             agentOptions,
@@ -1048,27 +1097,77 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
       if (error instanceof SessionNotFound) {
         return { error: { code: 'session-not-found', message: error.message, details: { sessionId } } }
       }
+      if (error instanceof SubagentSessionOwnership) {
+        return { error: subagentOwnershipError(error.sessionId) }
+      }
       // The internal details slot is contractually {}; the reason rides the message.
       return { error: { code: 'internal', message: `resume failed for session "${sessionId}": ${String(error)}`, details: {} } }
     }
   }
 
+  type SessionReadState = {
+    id: SessionId
+    header: SessionHeader
+    events: SessionEvent[]
+  }
+
+  /** Read one stable session prefix without acquiring an Agent owner. */
+  async function readSessionState(sessionId: SessionId): Promise<SessionReadState> {
+    const attached = ctx.sessions.get(sessionId)
+    if (attached !== undefined) {
+      return {
+        id: attached.id,
+        header: attached.header,
+        events: [...attached.events],
+      }
+    }
+    const inspected = await inspectServable(sessionId)
+    return { id: inspected.meta.id, header: inspected.meta, events: inspected.events }
+  }
+
+  /** Read one transcript cut and optional projection baseline without acquiring an Agent owner. */
+  async function historyStateFor(
+    sessionId: SessionId,
+    includeProjections: boolean,
+  ): Promise<{ events: SessionEvent[]; projections?: SessionProjectionsBlock }> {
+    const attached = ctx.sessions.get(sessionId)
+    if (attached !== undefined) {
+      const events = [...attached.events]
+      const projections = includeProjections ? projectionsFor(ctx, attached) : undefined
+      return { events, ...projections === undefined ? {} : { projections } }
+    }
+    const inspected = await inspectServable(sessionId)
+    const projections = includeProjections ? detachedProjectionsFor(ctx, inspected.events) : undefined
+    return {
+      events: inspected.events,
+      ...projections === undefined ? {} : { projections },
+    }
+  }
+
   /** Resolve one requested identity to a live agent, creating or resuming it once. */
   async function ensureSession(sessionId: SessionId, cwd: string, checkPersistedIdentity: boolean): Promise<Agent> {
     let creation = sessionCreations.get(sessionId)
     if (creation === undefined) {
       creation = (async () => {
+        const attached = ctx.sessions.get(sessionId)
         const live = ctx.agents.get(sessionId)
+        if (attached !== undefined && hasSubagentOwner(attached, live)) {
+          throw new SubagentSessionOwnership(sessionId)
+        }
         if (live !== undefined) return live
 
         const persistence = checkPersistedIdentity ? ctx.get('sessionPersistence') : undefined
         const stored = persistence === undefined
           ? undefined
           : (await persistence.list()).find(header => header.id === sessionId)
-        if (stored !== undefined) {
+        if (persistence !== undefined && stored !== undefined) {
           if (stored.cwd !== cwd) {
             throw new SessionCwdConflict(sessionId, cwd, stored.cwd)
           }
+          const inspected = await persistence.inspect(sessionId)
+          if (hasSubagentOwner({ header: inspected.meta, events: inspected.events }, undefined)) {
+            throw new SubagentSessionOwnership(sessionId)
+          }
           return (await ctx.agents.resume({
             resumeSessionId: sessionId,
             agentOptions,
@@ -1091,7 +1190,14 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
         // Another Host entry path may have published the same identity while
         // this operation crossed an asynchronous persistence/filesystem step.
         const live = ctx.agents.get(sessionId)
-        if (live !== undefined) return live
+        if (live !== undefined) {
+          if (hasSubagentOwner(live.session, live)) throw new SubagentSessionOwnership(sessionId)
+          return live
+        }
+        const attached = ctx.sessions.get(sessionId)
+        if (attached !== undefined && hasSubagentOwner(attached, undefined)) {
+          throw new SubagentSessionOwnership(sessionId)
+        }
         throw error
       }).finally(() => {
         sessionCreations.delete(sessionId)
@@ -1099,6 +1205,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
       sessionCreations.set(sessionId, creation)
     }
     const agent = await creation
+    if (hasSubagentOwner(agent.session, agent)) throw new SubagentSessionOwnership(sessionId)
     if (agent.session.header.cwd !== cwd) {
       throw new SessionCwdConflict(sessionId, cwd, agent.session.header.cwd)
     }
@@ -1498,6 +1605,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
               },
             })
           }
+          if (error instanceof SubagentSessionOwnership) {
+            return err(request, subagentOwnershipError(error.sessionId))
+          }
           return err(request, {
             code: 'internal',
             message: `failed to create session "${sessionId}": ${String(error)}`,
@@ -1520,18 +1630,24 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
 
       async history(request) {
         const { sessionId, beforeSeq, maxMessages } = request.payload
-        const found = await agentFor(sessionId)
-        if ('error' in found) return err(request, found.error)
-        // Everything below the resume above is synchronous: the page slice,
-        // the seq read, and the projection walk see one un-torn session state.
-        const page = historyPage(ctx, found.agent.session.events, beforeSeq, maxMessages)
-        // Baseline rider: tail page only — loadOlder (beforeSeq present) is
-        // the one path that never needs a fresh projection baseline.
-        const projections = beforeSeq === undefined ? projectionsFor(ctx, found.agent) : undefined
+        let state: { events: SessionEvent[]; projections?: SessionProjectionsBlock }
+        try {
+          state = await historyStateFor(sessionId, beforeSeq === undefined)
+        } catch (error: unknown) {
+          if (error instanceof SessionNotFound) {
+            return err(request, { code: 'session-not-found', message: error.message, details: { sessionId } })
+          }
+          return err(request, {
+            code: 'internal',
+            message: `history unavailable for session "${sessionId}": ${String(error)}`,
+            details: {},
+          })
+        }
+        const page = historyPage(ctx, state.events, beforeSeq, maxMessages)
         return ok(request, {
           events: page.events,
           hasMore: page.hasMore,
-          ...projections === undefined ? {} : { projections },
+          ...state.projections === undefined ? {} : { projections: state.projections },
         })
       },
 
@@ -1606,9 +1722,19 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
 
       async fork(request) {
         const { sessionId, atSeq } = request.payload
-        const found = await agentFor(sessionId)
-        if ('error' in found) return err(request, found.error)
-        const source = found.agent.session
+        let source: SessionReadState
+        try {
+          source = await readSessionState(sessionId)
+        } catch (error: unknown) {
+          if (error instanceof SessionNotFound) {
+            return err(request, { code: 'session-not-found', message: error.message, details: { sessionId } })
+          }
+          return err(request, {
+            code: 'internal',
+            message: `fork source unavailable for session "${sessionId}": ${String(error)}`,
+            details: {},
+          })
+        }
         const events = source.events
         // An in-log anchor belongs to the turn containing it and must never
         // clip backward to an earlier completed turn. Omitted and past-end
@@ -1694,6 +1820,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
       updateQueue(request) {
         const { sessionId, itemId, action } = request.payload
         const agent = ctx.agents.get(sessionId)
+        if (agent !== undefined && hasSubagentOwner(agent.session, agent)) {
+          return Promise.resolve(err(request, subagentOwnershipError(sessionId)))
+        }
         if (agent === undefined || agent.updateInbox(itemId, action) === 'not-found') {
           return Promise.resolve(err(request, {
             code: 'queue-item-not-found',
@@ -1714,6 +1843,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
             details: { sessionId },
           }))
         }
+        if (hasSubagentOwner(agent.session, agent)) {
+          return Promise.resolve(err(request, subagentOwnershipError(sessionId)))
+        }
         agent.cancel({ kind: 'user' }, { keepInbox: true })
         return Promise.resolve(ok(request, { accepted: true as const }))
       },

packages/host/apiproxy/src/api/commands.ts

@@ -1,8 +1,8 @@
 /**
  * commands domain contract: the web catalog/dispatch face of the host command
- * registry (`ctx.commands`). Both methods address one session's agent via
- * `sessionId` — every served session has an Agent (Session+Agent are born
- * together), so there is no agent-less surface on this wire.
+ * registry (`ctx.commands`). Both methods address an ordinary session's Agent
+ * via `sessionId`, resuming it when cold. Session-backed subagents reject with
+ * `agent-busy` and retain their dedicated continuation owner.
  */
 
 import type { CommandId } from '@deepseek-ai/dsh-commands/brand'
@@ -27,7 +27,8 @@ export interface CommandDescriptor {
 export interface CommandsApi {
   /**
    * Lists the addressed agent's effective command catalog (name-sorted,
-   * globals plus its scoped shadows).
+   * globals plus its scoped shadows). Session-backed subagents reject with
+   * `agent-busy`.
    */
   list(request: RpcRequest<{ sessionId: SessionId }>): Promise<RpcResponse<{ commands: readonly CommandDescriptor[] }>>
 
@@ -42,6 +43,7 @@ export interface CommandsApi {
    * pairing id, letting the issuing client correlate this acknowledgment
    * with that flow node. The signal rides beside the request, never on the
    * wire: the fetch carrier's request signal cancels the running handler.
+   * Session-backed subagents reject with `agent-busy` before dispatch.
    */
   execute(request: RpcRequest<{ sessionId: SessionId; line: string }>, signal: AbortSignal):
   Promise<RpcResponse<{ matched: boolean; commandId?: CommandId }>>

packages/host/apiproxy/src/api/goals.ts

@@ -22,7 +22,11 @@ export interface GoalRef {
   readonly revision: number
 }
 
-/** Goal-domain unary methods (every mutation resolves the session's agent and applies one CAS-guarded verb). */
+/**
+ * Goal-domain unary methods. Every mutation resolves an ordinary session's
+ * Agent and applies one CAS-guarded verb; session-backed subagents reject with
+ * `agent-busy`.
+ */
 export interface GoalsApi {
   /** Create and arm a goal. */
   create(request: RpcRequest<{ sessionId: SessionId; objective: string; maxGoalRounds?: number }>):

packages/host/apiproxy/src/api/sessions.ts

@@ -219,17 +219,22 @@ export interface SessionsApi {
    * the client needs a fresh baseline already pulls the tail page, and
    * loadOlder (the only beforeSeq path) is the only path that never needs one.
    * A deployment without the registry serves histories without the block.
+   * Reading history uses an attached Session or persistence inspection and
+   * never resumes or publishes an Agent.
    */
   history(request: RpcRequest<{ sessionId: SessionId; beforeSeq?: number; maxMessages?: number }>):
   Promise<RpcResponse<{ events: HistoryEntry[]; hasMore: boolean; projections?: SessionProjectionsBlock }>>
 
-  /** Reads a fresh advisory model directory for this session. Provider lookups run independently. */
+  /**
+   * Reads a fresh advisory model directory for an ordinary session. Provider
+   * lookups run independently; subagents reject with `agent-busy`.
+   */
   models(request: RpcRequest<{ sessionId: SessionId }>): Promise<RpcResponse<SessionModels>>
 
   /**
    * Selects the complete target for this session. Exact model metadata
    * validates an optional reasoning effort, while catalog membership remains
-   * advisory.
+   * advisory. Session-backed subagents reject with `agent-busy`.
    */
   selectModel(request: RpcRequest<{
     sessionId: SessionId
@@ -245,6 +250,7 @@ export interface SessionsApi {
    * normalized accepted title and the title event's seq return so the caller
    * can settle its projection cell without waiting for the push frame. A
    * title that normalizes to empty fails with `title-invalid`.
+   * Session-backed subagents reject with `agent-busy`.
    */
   rename(request: RpcRequest<{ sessionId: SessionId; title: string }>):
   Promise<RpcResponse<{ title: string; seq: number }>>
@@ -265,23 +271,30 @@ export interface SessionsApi {
    * falls back to the source's last completed turn. An in-log anchor whose
    * turn is still open fails with `fork-unavailable` instead of clipping to
    * an earlier turn. The child inherits the source cwd, latest logged model
-   * target, workspace attachment, and `parentSessionId` lineage; the seed
-   * prefix carries the source title.
+   * target and `parentSessionId` lineage; the seed prefix carries the source
+   * title. Reading the source uses attached state or persistence inspection
+   * without acquiring an Agent. Workspace attachment follows the source
+   * directly when it belongs to one.
    */
   fork(request: RpcRequest<{ sessionId: SessionId; atSeq?: number }>):
   Promise<RpcResponse<{ sessionId: SessionId }>>
 
-  /** Sends a message. content is core's ContentBlock[] verbatim; mode maps 1:1 — queue→send, steer→steer. */
+  /** Sends a message to an ordinary session Agent. Session-backed subagents reject with `agent-busy` and use `subagent.prompt`. */
   prompt(request: RpcRequest<{ sessionId: SessionId; mode: 'queue' | 'steer'; content: ContentBlock[] }>):
   Promise<RpcResponse<{ accepted: true; command?: { kind: 'success'; text?: string } }>>
 
   /**
-   * Edits or removes one pending queued occurrence.
+   * Edits or removes one pending queued occurrence on an ordinary session.
+   * Session-backed subagents reject with `agent-busy`.
    */
   updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: InboxItemId; action: QueueAction }>):
   Promise<RpcResponse<{ accepted: true }>>
 
-  /** Stops the active turn, preserving pending inbox work that resumes in FIFO order after cancellation settles. */
+  /**
+   * Stops an ordinary session's active turn, preserving pending inbox work
+   * that resumes in FIFO order after cancellation settles. Session-backed
+   * subagents reject with `agent-busy`.
+   */
   cancel(request: RpcRequest<{ sessionId: SessionId }>): Promise<RpcResponse<{ accepted: true }>>
 
 }