refactor(core): fold initiator scope into agents

packages/core/agent/README.md

@@ -1,10 +1,10 @@
 # dsh-agent
 
-Agent interface, registry, and `agent/*` event vocabulary. Every plugin (UI, hooks, orchestrators) programs against the `Agent` handle defined here — it has zero loop dependency, so the loop is swappable.
+Agent interface, registry, process-local initiator scope, and `agent/*` event vocabulary. Every plugin (UI, hooks, orchestrators) programs against the `Agent` handle defined here — it has zero loop dependency, so the loop is swappable.
 
 ## Service: `AgentRegistry` (ctx key: `agents`)
 
-Tracks live agents so UI, hook, and orchestrator plugins can find them without importing the concrete loop package.
+Tracks live agents and carries the initiating Agent through asynchronous driver work without importing the concrete loop package.
 
 ### Public API
 
@@ -17,6 +17,17 @@ The scoped-registration surface: `Agent.ctx` is the agent's scope context (`dsh-
 - `ctx.agents.list(): Agent[]`
 - `ctx.agents.roots(): Agent[]` — live agents created without an owning agent context; a resumed lineage-bearing session can still be a runtime root.
 
+#### Initiating Agent scope
+
+`AgentLoop` runs each concrete driver's complete lifetime inside an initiator boundary. Concurrent drivers remain isolated, a child driver shadows its parent, and the parent returns after the child settles. Creation, persistence load, and unpublished setup remain outside the child's boundary, so setup initiated by a parent inherits the parent while `agentCtx.agent` identifies the child explicitly.
+
+- `ctx.agents.currentInitiator(): Agent | undefined` — read the inherited initiator without requiring one.
+- `ctx.agents.requireInitiator(): Agent` — read it or throw `no initiating agent is active`.
+- `ctx.agents.withInitiator(agent, operation)` — run with one exact Agent and preserve the operation's exact synchronous value or Promise.
+- `ctx.agents.withoutInitiator(operation)` — hide an inherited initiator for unrelated process-local work.
+
+The scope carries the `Agent` itself and is process-local. Ambient presence is neither liveness proof nor authorization; explicit Agent fields remain authoritative at service, worker, process, persistence, and wire boundaries. Teardown rejects new boundaries, lets injected dependents and returned-Promise boundaries drain, then disables the underlying `AsyncLocalStorage`; unreturned work remains owned by the subsystem that detached it. The [initiator-scope decision](../../../docs/rfc/implemented/architecture/2026-07-15-agent-initiator-scope.md) owns the detailed boundary and teardown contract.
+
 #### Factory seam (creation)
 
 Agent *creation* is provided by the plugin implementing `AgentFactory` (`dsh-agent-loop`), registered via `setFactory`. This keeps creation on the `dsh-agent` interface so consumers (UI, the ACP bridge) program against `ctx.agents` without depending on the concrete loop package. The registry canonicalizes an already traced Service to its concrete target and re-traces each call through the caller's context; this avoids nested Cordis shadows while passing an explicit caller-bound `ownerCtx` to plain factories.
@@ -72,6 +83,8 @@ The handle every plugin programs against:
 
 ## Known Limitations and Deferred Work
 
+- **Initiator scope is process-local** — workers, child processes, HTTP, durable queues, and restarts materialize any required identity explicitly.
+- **Ambient identity may outlive liveness** — consumers still check `agent.status`, cancellation, and the owning capability contract before lifecycle-sensitive work.
 - **Inter-agent channels beyond delegation** — shared state, streaming child output, and background/poll semantics remain outside the current synchronous `ctx.subagents` seam.
 - **`agent/session-start` cannot gate startup** — it remains a synchronous, veto-less notification; async composition that must finish before publication belongs in the factory's `setup(agentCtx)` transaction instead.
 - **No public step-only abort** — `cancel()` clears ALL pending work (queued + steering + in-flight); an abort that preserves queued prompts returns only with a named consumer ([stop-surface RFC](../../../docs/rfc/implemented/simplification/2026-06-20-public-agent-stop-surface.md)).

packages/core/agent/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@deepseek-ai/dsh-agent",
-  "description": "Agent interface, registry, and event vocabulary for the DeepSeek Harness",
+  "description": "Agent interface, registry, initiator scope, and event vocabulary for the DeepSeek Harness",
   "version": "0.0.1",
   "private": true,
   "type": "module",

packages/core/agent/src/index.ts

@@ -1,11 +1,13 @@
 /**
- * Agent registry service. Tracks live agents so plugins can find them without
- * depending on the concrete loop package. Agent creation belongs to the loop.
+ * Agent service: live registry, factory delegation, and process-local
+ * initiator scope. Concrete creation and driving belong to the loop.
  *
  * @module @deepseek-ai/dsh-agent
  */
 
 import { Context, getTraceable, Service, symbols } from 'cordis'
+import { AsyncLocalStorage } from 'node:async_hooks'
+import { isPromise } from 'node:util/types'
 import { scopeTarget } from '@deepseek-ai/dsh-scope'
 import type { Scoped } from '@deepseek-ai/dsh-scope'
 import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
@@ -173,6 +175,8 @@ export interface AgentFactory {
 
 /** Thrown when create/resume is called before an agent factory is registered. */
 const NO_FACTORY_MESSAGE = 'no agent factory registered (load an agent-loop plugin)'
+const NO_INITIATOR_MESSAGE = 'no initiating agent is active'
+const DISPOSED_INITIATOR_MESSAGE = 'agent initiator scope is disposed'
 
 /** All mutable lifecycle state for one exact registry entry. */
 interface AgentEntry {
@@ -192,15 +196,19 @@ interface FactorySlot {
 }
 
 /**
- * Agent registry (`ctx.agents`): tracks live agents so UI, hook, and
- * orchestrator plugins can find them without depending on the concrete loop
- * package. Agent *creation* is provided by whichever plugin implements the
- * {@link AgentFactory} (`@deepseek-ai/dsh-agent-loop`), registered via
- * {@link setFactory}.
+ * Agent service (`ctx.agents`): tracks live agents and carries the initiating
+ * Agent through one process-local asynchronous driver chain. Agent *creation*
+ * is provided by whichever plugin implements the {@link AgentFactory}
+ * (`@deepseek-ai/dsh-agent-loop`), registered via {@link setFactory}.
  */
 export class AgentRegistry extends Service {
   private store = new Map<SessionId, AgentEntry>()
   private factory: FactorySlot | undefined
+  private readonly initiators = new AsyncLocalStorage<Agent | undefined>()
+  private initiatorState: 'active' | 'closing' | 'disposed' = 'active'
+  private activeInitiatorRuns = 0
+  private initiatorDrain: PromiseWithResolvers<void> | undefined
+  private initiatorDisposal: Promise<void> | undefined
 
   constructor(ctx: Context) {
     super(ctx, 'agents')
@@ -211,6 +219,54 @@ export class AgentRegistry extends Service {
     // accessor body never needs to resolve a scope itself. Effect-scoped:
     // unwinds with this service's fiber.
     ctx.accessor('agent', { get: () => undefined })
+    ctx.effect(function* (this: AgentRegistry) {
+      yield () => this.disposeInitiators()
+      yield () => { this.closeInitiators() }
+    }.bind(this), 'agents.initiatorLifecycle()')
+  }
+
+  /**
+   * Read the Agent that initiated the inherited asynchronous driver chain.
+   * @returns the inherited Agent, or `undefined` outside a driver and inside an explicit clearing boundary.
+   * @throws when this service instance has been disposed.
+   */
+  currentInitiator(): Agent | undefined {
+    this.assertInitiatorsReadable()
+    return this.initiators.getStore()
+  }
+
+  /**
+   * Read the initiating Agent and fail when no driver boundary is active.
+   * @returns the inherited Agent.
+   * @throws when no initiator is active or this service instance has been disposed.
+   */
+  requireInitiator(): Agent {
+    const agent = this.currentInitiator()
+    if (agent === undefined) throw new Error(NO_INITIATOR_MESSAGE)
+    return agent
+  }
+
+  /**
+   * Run an operation with one exact Agent as its process-local initiator. The
+   * exact synchronous value or Promise returned by the operation is preserved.
+   * @param agent - initiating Agent to inherit; presence is neither liveness proof nor authorization.
+   * @param operation - synchronous or asynchronous operation to invoke.
+   * @returns the exact value returned by `operation`.
+   * @throws when the initiator scope is closing/disposed, or when `operation` throws.
+   */
+  withInitiator<T>(agent: Agent, operation: () => T): T {
+    return this.runWithInitiator(agent, operation)
+  }
+
+  /**
+   * Run an operation inside a boundary that hides any inherited initiating
+   * Agent. The exact synchronous value or Promise is preserved.
+   * @param operation - synchronous or asynchronous operation to invoke without an initiator.
+   * @returns the exact value returned by `operation`.
+   * @throws when the initiator scope is closing/disposed, or when `operation` throws.
+   */
+  withoutInitiator<T>(operation: () => T): T {
+    return this.runWithInitiator(undefined, operation)
   }
 
   /**
@@ -471,6 +527,57 @@ export class AgentRegistry extends Service {
       .filter(entry => entry.owner === undefined)
       .map(entry => entry.agent)
   }
+
+  /** Reject new initiator boundaries while inherited continuations drain. */
+  private closeInitiators(): void {
+    if (this.initiatorState === 'active') this.initiatorState = 'closing'
+  }
+
+  /** Wait for returned-Promise boundaries, then invalidate retained references. */
+  private disposeInitiators(): Promise<void> {
+    return (this.initiatorDisposal ??= (async () => {
+      this.closeInitiators()
+      if (this.activeInitiatorRuns !== 0) {
+        this.initiatorDrain ??= Promise.withResolvers<void>()
+        await this.initiatorDrain.promise
+      }
+      this.initiatorState = 'disposed'
+      this.initiators.disable()
+    })())
+  }
+
+  /** Establish one tracked initiator or clearing boundary. */
+  private runWithInitiator<T>(agent: Agent | undefined, operation: () => T): T {
+    if (this.initiatorState !== 'active') throw new Error(DISPOSED_INITIATOR_MESSAGE)
+    this.activeInitiatorRuns += 1
+    let result: T
+    try {
+      result = this.initiators.run(agent, operation)
+    } catch (error: unknown) {
+      this.releaseInitiatorRun()
+      throw error
+    }
+    if (isPromise(result)) {
+      void result.then(
+        () => { this.releaseInitiatorRun() },
+        () => { this.releaseInitiatorRun() },
+      )
+    } else {
+      this.releaseInitiatorRun()
+    }
+    return result
+  }
+
+  private assertInitiatorsReadable(): void {
+    if (this.initiatorState === 'disposed') throw new Error(DISPOSED_INITIATOR_MESSAGE)
+  }
+
+  private releaseInitiatorRun(): void {
+    this.activeInitiatorRuns -= 1
+    if (this.activeInitiatorRuns !== 0) return
+    this.initiatorDrain?.resolve()
+    this.initiatorDrain = undefined
+  }
 }
 
 export default AgentRegistry

packages/core/agent/tests/agent-initiator.spec.ts

@@ -0,0 +1,163 @@
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import { runInNewContext } from 'node:vm'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import { SessionId } from '@deepseek-ai/dsh-session'
+
+function agent(id: string): Agent {
+  return { id: SessionId(id) } as Agent
+}
+
+async function harness(): Promise<{
+  ctx: Context
+  service: AgentRegistry
+  dispose: () => Promise<void>
+}> {
+  const ctx = new Context()
+  const fiber = await ctx.plugin(AgentRegistry)
+  return {
+    ctx,
+    service: ctx.agents,
+    dispose: fiber.dispose,
+  }
+}
+
+describe('AgentRegistry initiator scope', () => {
+  it('reports an absent initiator and requires an active boundary', async () => {
+    const { service, dispose } = await harness()
+    expect(service.currentInitiator()).toBeUndefined()
+    expect(() => service.requireInitiator()).toThrow('no initiating agent is active')
+    await dispose()
+  })
+
+  it('preserves exact synchronous and Promise return identities across await', async () => {
+    const { service, dispose } = await harness()
+    const initiator = agent('identity')
+    const value = { result: true }
+    expect(service.withInitiator(initiator, () => {
+      expect(service.requireInitiator()).toBe(initiator)
+      return value
+    })).toBe(value)
+
+    const promise = service.withInitiator(initiator, async () => {
+      expect(service.requireInitiator()).toBe(initiator)
+      await Promise.resolve()
+      expect(service.requireInitiator()).toBe(initiator)
+      return value
+    })
+    expect(service.withInitiator(initiator, () => promise)).toBe(promise)
+    await expect(promise).resolves.toBe(value)
+    expect(service.currentInitiator()).toBeUndefined()
+    await dispose()
+  })
+
+  it('isolates overlapping initiators', async () => {
+    const { service, dispose } = await harness()
+    const a = agent('a')
+    const b = agent('b')
+    const bothStarted = Promise.withResolvers<boolean>()
+    const release = Promise.withResolvers<boolean>()
+    let starts = 0
+    const run = (initiator: Agent): Promise<void> => service.withInitiator(initiator, async () => {
+      expect(service.requireInitiator()).toBe(initiator)
+      starts += 1
+      if (starts === 2) bothStarted.resolve(true)
+      await release.promise
+      expect(service.requireInitiator()).toBe(initiator)
+    })
+
+    const pending = [run(a), run(b)]
+    await bothStarted.promise
+    expect(service.currentInitiator()).toBeUndefined()
+    release.resolve(true)
+    await Promise.all(pending)
+    await dispose()
+  })
+
+  it('restores nested and explicitly cleared boundaries', async () => {
+    const { service, dispose } = await harness()
+    const parent = agent('parent')
+    const child = agent('child')
+
+    service.withInitiator(parent, () => {
+      expect(service.requireInitiator()).toBe(parent)
+      service.withInitiator(child, () => { expect(service.requireInitiator()).toBe(child) })
+      expect(service.requireInitiator()).toBe(parent)
+      service.withoutInitiator(() => {
+        expect(service.currentInitiator()).toBeUndefined()
+        expect(() => service.requireInitiator()).toThrow('no initiating agent is active')
+      })
+      expect(service.requireInitiator()).toBe(parent)
+    })
+    expect(service.currentInitiator()).toBeUndefined()
+    await dispose()
+  })
+
+  it('restores the parent after synchronous throws and rejected operations', async () => {
+    const { service, dispose } = await harness()
+    const parent = agent('parent')
+    const child = agent('child')
+    const syncError = new Error('sync failure')
+    const asyncError = new Error('async failure')
+
+    service.withInitiator(parent, () => {
+      expect(() => service.withInitiator(child, () => { throw syncError })).toThrow(syncError)
+      expect(service.requireInitiator()).toBe(parent)
+    })
+    await expect(service.withInitiator(child, async () => {
+      await Promise.resolve()
+      throw asyncError
+    })).rejects.toBe(asyncError)
+    expect(service.currentInitiator()).toBeUndefined()
+    await dispose()
+  })
+
+  it('stops new boundaries, drains active Promises, and invalidates retained references', async () => {
+    const { ctx, service, dispose } = await harness()
+    const initiator = agent('draining')
+    const release = Promise.withResolvers<boolean>()
+    const pending = service.withInitiator(initiator, async () => {
+      await release.promise
+      expect(service.requireInitiator()).toBe(initiator)
+    })
+    let disposed = false
+    const disposal = dispose().then(() => { disposed = true })
+    await Promise.resolve()
+
+    expect(() => service.withInitiator(initiator, () => 1)).toThrow('agent initiator scope is disposed')
+    expect(() => service.withoutInitiator(() => 1)).toThrow('agent initiator scope is disposed')
+    expect(disposed).toBe(false)
+    expect(ctx.get('agents')).toBeUndefined()
+    release.resolve(true)
+    await pending
+    await disposal
+    expect(() => service.currentInitiator()).toThrow('agent initiator scope is disposed')
+    expect(() => service.requireInitiator()).toThrow('agent initiator scope is disposed')
+  })
+
+  it('drains cross-realm Promise boundaries before disposal', async () => {
+    const { service, dispose } = await harness()
+    const initiator = agent('cross-realm')
+    const release = Promise.withResolvers<boolean>()
+    const operation = runInNewContext(
+      '(async () => { await release; inspect() })',
+      {
+        release: release.promise,
+        inspect: () => { expect(service.requireInitiator()).toBe(initiator) },
+      },
+    ) as () => Promise<void>
+    const pending = service.withInitiator(initiator, operation)
+    expect(pending).not.toBeInstanceOf(Promise)
+
+    let disposed = false
+    const disposal = dispose().then(() => { disposed = true })
+    await Promise.resolve()
+    expect(disposed).toBe(false)
+
+    release.resolve(true)
+    await pending
+    await disposal
+    expect(disposed).toBe(true)
+  })
+})

packages/core/agent/tests/gen-cordis-catalog.spec.ts

@@ -6,7 +6,7 @@ import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, describe, expect, it } from 'vitest'
-import { collectEvents, collectServices } from '../../../../scripts/gen-cordis-catalog.ts'
+import { collectEvents, collectServices, renderEvents, renderServices } from '../../../../scripts/gen-cordis-catalog.ts'
 
 /** Write a fixture package exposing one `interface Events` block and return the
  * scan root to hand `collectEvents`. */
@@ -58,6 +58,8 @@ describe('gen-cordis-catalog collectEvents', () => {
     ))
     expect(events).toHaveLength(1)
     expect(events[0]).toMatchObject({ name: 'fix/happened', scope: 'fix', mode: 'emit', doc: 'A thing happened.' })
+    expect(events[0]?.jsDoc).toBe('/**\n * A thing happened.\n * @param id - which thing.\n * @mode emit\n */')
+    expect(renderEvents(events)).toContain("```ts cordis-catalog\n/**\n * A thing happened.\n * @param id - which thing.\n * @mode emit\n */\n'fix/happened'(id: string): void\n```")
   })
 
   it('classifies a trailing-next signature as a waterfall', () => {
@@ -158,26 +160,11 @@ export class FixService {
     expect(services).toHaveLength(1)
     expect(services[0]).toMatchObject({ key: 'fix', type: 'FixService', abstract: false, doc: 'Fixture service.' })
     expect(services[0]?.methods).toHaveLength(3)
-  })
-
-  it('extracts an interface service as an abstract seam', () => {
-    const services = collectServices(makeService(`/** Fixture service interface. */
-export interface FixService {
-  /**
-   * Do the thing.
-   * @param id - which thing to do.
-   * @returns the outcome of doing it.
-   */
-  run(id: string): string
-}`))
-    expect(services).toHaveLength(1)
-    expect(services[0]).toMatchObject({
-      key: 'fix',
-      type: 'FixService',
-      abstract: true,
-      doc: 'Fixture service interface.',
+    expect(services[0]?.methods[0]).toEqual({
+      signature: 'run(id: string): string',
+      jsDoc: '/**\n * Do the thing.\n * @param id - which thing to do.\n * @returns the outcome of doing it.\n */',
     })
-    expect(services[0]?.methods).toEqual(['run(id: string): string'])
+    expect(renderServices(services)).toContain('```ts cordis-catalog\n/**\n * Do the thing.\n * @param id - which thing to do.\n * @returns the outcome of doing it.\n */\nrun(id: string): string\n\n/** Fire and forget (void needs no @returns). */\npoke(): void')
   })
 
   it('hard-errors on a public method with no JSDoc at all', () => {