Merge remote-tracking branch 'origin/master' into feat/profile-plugin-management

# Conflicts:
#	apps/cli/src/headless.ts
#	packages/host/apiproxy/README.i18n.yaml

packages/bundle/headless/package.json

@@ -33,6 +33,7 @@
     "schemastery": "^3.18.0"
   },
   "peerDependencies": {
+    "@deepseek-ai/dsh-agent": "^0.0.1",
     "@deepseek-ai/dsh-host-apiproxy": "^0.0.1",
     "@deepseek-ai/dsh-host-webserver": "^0.0.1",
     "@deepseek-ai/dsh-invariants": "^0.0.1",
@@ -40,6 +41,7 @@
     "cordis": "^4.0.0-rc.7"
   },
   "devDependencies": {
+    "@deepseek-ai/dsh-agent": "workspace:^",
     "@deepseek-ai/dsh-host-apiproxy": "workspace:^",
     "@deepseek-ai/dsh-host-webserver": "workspace:^",
     "@deepseek-ai/dsh-invariants": "workspace:^",

packages/bundle/headless/src/index.ts

@@ -2,19 +2,21 @@
  * @deepseek-ai/dsh-headless — the one-shot headless bundle: the bundle patch
  * (`cordis.patch.yml`) rides over dsh-base + dsh-web-app (the headless
  * session is web-observable while it runs — same composition), and this
- * runner plugin drives one task turn through the in-process API carrier
+ * runner plugin drives one task through the in-process API carrier
  * (InProcessApiClient over toFetchHandler(ctx.apiProxy), so the full wire
  * chain — serialization, zod, SSE framing — really runs), prints the final
- * assistant text, and exits (completed → 0, else 1). The task text arrives as
- * launcher-patched config (`dsh --profile headless "task"`).
+ * assistant text at agent quiescence, and exits (completed → 0, else 1). The
+ * task text arrives as launcher-patched config
+ * (`dsh --profile headless "task"`).
  * @module @deepseek-ai/dsh-headless
  */
 
 import type { Context } from 'cordis'
 import z from 'schemastery'
 import { InProcessApiClient, toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
-// Empty type import carries the httpServer Context merge for the port read below.
+// Empty type imports carry the httpServer and agent/status Context merges used below.
 import type {} from '@deepseek-ai/dsh-host-webserver'
+import type {} from '@deepseek-ai/dsh-agent'
 import type { MuxFrame } from '@deepseek-ai/dsh-host-apiproxy/api'
 import type { RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
 import type { SessionId } from '@deepseek-ai/dsh-session'
@@ -35,7 +37,7 @@ export const Config: z<Config> = z.object({
   task: z.string().required(),
 })
 
-/** Outcome of one headless turn: aggregated final text plus the turn-end reason kind. */
+/** Outcome of one headless run: aggregated final text plus the last turn-end reason kind. */
 interface TurnOutcome {
   text: string
   reason: string
@@ -73,46 +75,56 @@ async function unwrap<T>(response: RpcResponse<T>, io: HeadlessIo): Promise<T> {
 }
 
 /**
- * Consume mux frames until the task turn ends: anchor on the first turn/start
- * whose trigger kind is 'message' (startup-injected turns are skipped),
- * aggregate text from that turn's assistant/message events (last one wins),
- * finish on its turn/end.
+ * Consume mux frames until the agent reaches idle, per the one-shot CLI
+ * idle-to-idle contract: the stream opens immediately before the prompt, and
+ * its first observed turn/start begins the task. Text is the last committed
+ * assistant message of the whole interval (steering or injected work may run
+ * further turns before quiescence), and the outcome reason is the final
+ * turn/end's kind. Idleness is signalled out of band by the caller's
+ * `agent/status` subscription; the stream itself carries no status frame.
+ * @param frames - the mux stream opened before the prompt.
+ * @param sessionId - the headless session.
+ * @param idle - resolves when the agent reaches quiescence.
+ * @param io - process-facing effects for stream diagnostics.
+ * @returns the aggregated outcome.
  */
-async function consumeUntilTurnEnd(
-  frames: AsyncIterable<RpcRequest<MuxFrame>>, sessionId: SessionId, io: HeadlessIo,
+async function consumeUntilIdle(
+  frames: AsyncIterable<RpcRequest<MuxFrame>>,
+  sessionId: SessionId,
+  idle: Promise<void>,
+  io: HeadlessIo,
 ): Promise<TurnOutcome> {
-  let targetTurn: number | undefined
+  let started = false
   let text = ''
-  try {
-    for await (const frame of frames) {
-      const payload = frame.payload
-      if (payload.type === 'stream/error') {
-        io.stderr.write(`dsh: stream error: ${payload.error.message}\n`)
-        return { text, reason: 'error' }
-      }
-      if (payload.type !== 'session/event' || payload.sessionId !== sessionId) continue
-      const event = payload.event
-      if (targetTurn === undefined) {
-        if (event.type === 'turn/start' && event.data.trigger.kind === 'message') targetTurn = event.data.turn
-        continue
-      }
-      if (event.type === 'assistant/message' && event.data.turn === targetTurn) {
-        const joined = event.data.message.content.filter(block => block.type === 'text').map(block => block.text).join('')
-        if (joined !== '') text = joined
-      }
-      if (event.type === 'turn/end' && event.data.turn === targetTurn) {
-        return { text, reason: event.data.reason.kind }
+  let reason: string = 'error'
+  void (async () => {
+    try {
+      for await (const frame of frames) {
+        const payload = frame.payload
+        if (payload.type === 'stream/error') return
+        if (payload.type !== 'session/event' || payload.sessionId !== sessionId) continue
+        const event = payload.event
+        if (event.type === 'turn/start') {
+          started = true
+          continue
+        }
+        if (!started) continue
+        if (event.type === 'assistant/message') {
+          const joined = event.data.message.content.filter(block => block.type === 'text').map(block => block.text).join('')
+          if (joined !== '') text = joined
+        }
+        if (event.type === 'turn/end') reason = event.data.reason.kind
       }
+    } catch (error: unknown) {
+      io.stderr.write(`dsh: event stream failed: ${String(error)}\n`)
     }
-  } catch (error: unknown) {
-    io.stderr.write(`dsh: event stream failed: ${String(error)}\n`)
-  }
-  return { text, reason: 'error' }
+  })()
+  await idle
+  return { text, reason }
 }
 
 /**
- * Run one headless turn for the configured task and request exit
- * (completed → 0, else 1).
+ * Run one headless task to quiescence and request exit (completed → 0, else 1).
  * @param ctx - plugin context carrying apiProxy, httpServer, and the launcher's headlessIo.
  * @param config - validated {@link Config}.
  */
@@ -121,7 +133,7 @@ export function apply(ctx: Context, config: Config): void {
   if (io === undefined) {
     throw new Error('headless-runner: the launcher must provide ctx.headlessIo before the tree mounts')
   }
-  // Fire-and-forget by design: the turn outlives plugin activation, and every
+  // Fire-and-forget by design: the run outlives plugin activation, and every
   // failure path inside ends in io.exit, not a rejection.
   void (async () => {
     // The headless session is web-observable while it runs (same composition).
@@ -133,7 +145,12 @@ export function apply(ctx: Context, config: Config): void {
     // a move to a remote HTTP carrier unchanged.
     const abort = new AbortController()
     const frames = api.events.mux({}, abort.signal)
-    const done = consumeUntilTurnEnd(frames, created.sessionId, io)
+    const idle = new Promise<void>((resolve) => {
+      ctx.on('agent/status', (agent, status) => {
+        if (agent.id === created.sessionId && status === 'idle') resolve()
+      })
+    })
+    const done = consumeUntilIdle(frames, created.sessionId, idle, io)
     await unwrap(await api.sessions.prompt({
       sessionId: created.sessionId,
       mode: 'queue',

packages/bundle/headless/tests/headless.spec.ts

@@ -1,12 +1,13 @@
 /**
- * One-shot runner behavior over a scripted in-process API: turn anchoring on
- * the first message-triggered turn, last-text-wins aggregation, exit-code
- * mapping by turn-end reason, stream/error and RPC-error paths, and the
+ * One-shot runner behavior over a scripted in-process API: idle-to-idle
+ * aggregation (last text of the whole interval), exit-code mapping by the
+ * final turn-end reason, stream-error and RPC-error paths, and the
  * launcher-owned `ctx.headlessIo` requirement.
  */
 
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
+import type { Agent } from '@deepseek-ai/dsh-agent'
 import { apply, Config, type HeadlessIo } from '../src/index.ts'
 
 interface ScriptedEvent { type: string; seq?: number; time?: number; sessionId?: string; data: Record<string, unknown> }
@@ -46,7 +47,10 @@ function scriptedApi(events: ScriptedEvent[], options: { promptFails?: boolean }
   }
 }
 
-/** Mount the runner against a scripted API and wait for its exit request. */
+/**
+ * Mount the runner against a scripted API, emit the idle transition after the
+ * scripted frames drain, and wait for its exit request.
+ */
 async function run(events: ScriptedEvent[], options: { promptFails?: boolean } = {}): Promise<{ code: number; out: string; err: string }> {
   const ctx = new Context()
   let out = ''
@@ -62,6 +66,10 @@ async function run(events: ScriptedEvent[], options: { promptFails?: boolean } =
   ctx.provide('apiProxy', scriptedApi(events, options) as never)
   ctx.provide('httpServer', { port: 12345 } as never)
   apply(ctx, { task: 'do the thing' })
+  // Quiescence is out of band: give the scripted stream a beat to drain, then
+  // flip the agent idle exactly as the loop would.
+  await new Promise(resolve => setTimeout(resolve, 10))
+  ctx.emit('agent/status', { id: 'S1' } as Agent, 'idle')
   const code = await exited
   await ctx.fiber.dispose()
   return { code, out, err }
@@ -76,15 +84,15 @@ const text = (turn: number, value: string): ScriptedEvent => ({
 const end = (turn: number, reason: string): ScriptedEvent => ({ type: 'turn/end', data: { turn, reason: { kind: reason } } })
 
 describe('headless runner', () => {
-  it('anchors past startup turns, keeps the last text, prints, and exits 0 on completion', async () => {
+  it('aggregates to quiescence: last text wins across turns, final turn-end reason maps to exit 0', async () => {
     const { code, out, err } = await run([
       startupTurn,
-      end(0, 'completed'),
-      messageTurn,
-      // Off-session, non-text, and text-empty frames are skipped without affecting the aggregate.
+      // Off-session, non-text, and text-empty frames never affect the aggregate.
       { type: 'assistant/message', sessionId: 'OTHER', data: { turn: 1, message: { content: [{ type: 'text', text: 'other session' }] } } },
       { type: 'assistant/message', data: { turn: 1, message: { content: [{ type: 'tool_call', text: 'ignored' }] } } },
-      text(1, 'draft'),
+      text(0, 'draft'),
+      end(0, 'completed'),
+      messageTurn,
       text(1, 'final answer'),
       end(1, 'completed'),
     ])
@@ -93,24 +101,25 @@ describe('headless runner', () => {
     expect(err).toContain('observing at http://127.0.0.1:12345')
   })
 
-  it('exits 1 when the turn ends for any other reason', async () => {
+  it('exits 1 when the final turn ends for any other reason', async () => {
     const { code } = await run([messageTurn, end(1, 'aborted')])
     expect(code).toBe(1)
   })
 
-  it('reports a stream error and exits 1', async () => {
-    const { code, err } = await run([messageTurn, { type: 'stream/error', data: {} }])
+  it('exits 1 when no turn ever starts (idle without work)', async () => {
+    const { code, out } = await run([])
     expect(code).toBe(1)
-    expect(err).toContain('stream error')
+    expect(out).toBe('\n')
   })
 
-  it('prints an RPC business error and exits 1 without prompting further', async () => {
-    const { code, err } = await run([messageTurn, end(1, 'completed')], { promptFails: true })
+  it('keeps the error outcome after a stream error ends the frame consumer early', async () => {
+    const { code } = await run([messageTurn, { type: 'stream/error', data: {} }, end(1, 'completed')])
+    // The consumer stopped at the stream error; the completed turn-end after
+    // it is never observed, so the reason stays 'error'.
     expect(code).toBe(1)
-    expect(err).toContain('agent-busy')
   })
 
-  it('exits 1 through the stream-error path when the underlying carrier dies', async () => {
+  it('prints an RPC business error and exits 1 without waiting for idle', async () => {
     const ctx = new Context()
     let err = ''
     const exited = new Promise<number>((resolve) => {
@@ -120,36 +129,15 @@ describe('headless runner', () => {
         exit: resolve,
       } satisfies HeadlessIo)
     })
-    ctx.provide('apiProxy', {
-      sessions: {
-        create: (request: RpcShapedRequest) =>
-          Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value: { sessionId: 'S1' } } }),
-        prompt: (request: RpcShapedRequest) =>
-          Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value: { accepted: true } } }),
-      },
-      events: {
-        mux: async function* (): AsyncGenerator<never> {
-          throw new Error('carrier died')
-        },
-      },
-    } as never)
+    ctx.provide('apiProxy', scriptedApi([messageTurn, end(1, 'completed')], { promptFails: true }) as never)
     ctx.provide('httpServer', { port: 1 } as never)
     apply(ctx, { task: 't' })
     expect(await exited).toBe(1)
-    // The carrier converts its own failure into a stream/error frame.
-    expect(err).toContain('stream error')
-    expect(err).toContain('carrier died')
+    expect(err).toContain('agent-busy')
     await ctx.fiber.dispose()
   })
 
-  it('fails loud without the launcher-owned headlessIo seam', () => {
-    const ctx = new Context()
-    ctx.provide('apiProxy', scriptedApi([]) as never)
-    ctx.provide('httpServer', { port: 1 } as never)
-    expect(() => { apply(ctx, { task: 't' }) }).toThrow('must provide ctx.headlessIo')
-  })
-
-  it('exits 1 with the stream-failed diagnostic when the event channel cannot open at all', async () => {
+  it('reports the stream-failed diagnostic when the event channel dies, still settling at idle', async () => {
     const ctx = new Context()
     let err = ''
     const exited = new Promise<number>((resolve) => {
@@ -174,11 +162,20 @@ describe('headless runner', () => {
     } as never)
     ctx.provide('httpServer', { port: 1 } as never)
     apply(ctx, { task: 't' })
+    await new Promise(resolve => setTimeout(resolve, 10))
+    ctx.emit('agent/status', { id: 'S1' } as Agent, 'idle')
     expect(await exited).toBe(1)
     expect(err).toContain('event stream failed')
     await ctx.fiber.dispose()
   })
 
+  it('fails loud without the launcher-owned headlessIo seam', () => {
+    const ctx = new Context()
+    ctx.provide('apiProxy', scriptedApi([]) as never)
+    ctx.provide('httpServer', { port: 1 } as never)
+    expect(() => { apply(ctx, { task: 't' }) }).toThrow('must provide ctx.headlessIo')
+  })
+
   it('validates config: the task is required', () => {
     expect(() => new Config({ } as never)).toThrow()
     expect(new Config({ task: 'x' })).toEqual({ task: 'x' })

packages/bundle/headless/tsconfig.json

@@ -20,6 +20,9 @@
     {
       "path": "../../host/webserver"
     },
+    {
+      "path": "../../core/agent"
+    },
     {
       "path": "../../core/session"
     },