fix(headless): dsh run is a direct core front door

packages/bundle/headless/src/index.ts

@@ -1,36 +1,33 @@
 /**
- * @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 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 at agent quiescence, and exits (completed → 0, else 1). The
- * task text arrives as launcher-patched config (`dsh run "task"`).
+ * @deepseek-ai/dsh-headless — one-shot direct Agent driver. The bundle patch
+ * rides over dsh-base without Host, HTTP, or browser plugins; this runner
+ * creates one Agent through the core registry, drives the task to quiescence,
+ * flushes its Session, prints the final assistant text, and exits.
+ *
  * @module @deepseek-ai/dsh-headless
  */
 
+import { randomUUID } from 'node:crypto'
 import type { Context } from 'cordis'
 import z from 'schemastery'
-import { InProcessApiClient, toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
-// 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 { installModelSelection } from '@deepseek-ai/dsh-agent'
+import type { ModelSelectionRef } from '@deepseek-ai/dsh-agent'
+import type {} from '@deepseek-ai/dsh-agent-default-model'
+import { createUserMessage } from '@deepseek-ai/dsh-llm'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
 // Empty type import carries the loader Context merge for the settlement await.
 import type {} from '@cordisjs/plugin-loader'
-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'
 
 /** Stable Cordis plugin name. */
 export const name = 'headless-runner'
 
-/** Services required before the one-shot turn can start. */
-export const inject = ['apiProxy', 'httpServer']
+/** Core services required before the one-shot turn can start. */
+export const inject = ['agentDefaultModel', 'agents', 'sessions']
 
 /** Plugin config: the task, patched in by the launcher. */
 export interface Config {
-  /** The prompt text for the single turn. */
+  /** The prompt text for the single run. */
   task: string
 }
 
@@ -38,16 +35,15 @@ export const Config: z<Config> = z.object({
   task: z.string().required(),
 })
 
-/** Outcome of one headless run: aggregated final text plus the last turn-end reason kind. */
-interface TurnOutcome {
+/** Outcome of one owned run interval. */
+interface RunOutcome {
   text: string
   reason: string
 }
 
 /**
- * The process-facing effects of one run, injectable for tests: output
- * streams and the exit request (the launcher wires it to its bounded
- * shutdown controller).
+ * Process-facing effects of one run, injectable for tests. The launcher owns
+ * bounded tree shutdown and wires `exit()` to it.
  */
 export interface HeadlessIo {
   stdout: { write(chunk: string): unknown }
@@ -56,127 +52,91 @@ export interface HeadlessIo {
   exit(code: number): void
 }
 
-/** Host seam: the launcher provides the exit wiring before the tree mounts. */
 declare module 'cordis' {
   interface Context {
-    /** Process-facing effects for the one-shot headless runner. */
+    /** Process-facing effects provided before the headless tree mounts. */
     headlessIo?: HeadlessIo
   }
 }
 
-/** Unwrap an RpcResponse or fail loud: business errors print and exit 1. */
-async function unwrap<T>(response: RpcResponse<T>, io: HeadlessIo): Promise<T> {
-  if (response.result.ok) return response.result.value
-  const { code, message } = response.result.error
-  io.stderr.write(`dsh: ${code}: ${message}\n`)
-  io.exit(1)
-  // Exit is asynchronous (bounded tree disposal); park this turn forever so
-  // no further request rides a session that is already being torn down.
-  return new Promise<never>(() => {})
-}
-
-/**
- * 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 to the final session-event sequence when the agent reaches quiescence.
- * @param io - process-facing effects for stream diagnostics.
- * @returns the aggregated outcome.
- */
-async function consumeUntilIdle(
-  frames: AsyncIterable<RpcRequest<MuxFrame>>,
-  sessionId: SessionId,
-  idle: Promise<number>,
-  io: HeadlessIo,
-): Promise<TurnOutcome> {
+/** Aggregate the last assistant text and turn outcome in one owned interval. */
+function summarize(events: readonly SessionEvent[], firstSeq: number): RunOutcome {
   let started = false
   let text = ''
-  let reason: string = 'error'
-  let observedSeq = -1
-  let resolveProgress: (() => void) | undefined
-  const streamDone = (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
-        observedSeq = event.seq
-        resolveProgress?.()
-        resolveProgress = undefined
-        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`)
+  let reason = 'error'
+  for (const event of events) {
+    if (event.seq < firstSeq) continue
+    if (event.type === 'turn/start') {
+      started = true
+      continue
     }
-  })()
-  const streamEnded = streamDone.then(() => 'ended' as const)
-  const idleSeq = await idle
-  while (observedSeq < idleSeq) {
-    const progress = new Promise<'progress'>((resolve) => { resolveProgress = () => { resolve('progress') } })
-    if (await Promise.race([progress, streamEnded]) === 'ended') break
+    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
   }
   return { text, reason }
 }
 
+/** Report an unexpected direct-driver failure and request a failing exit. */
+function fail(io: HeadlessIo, error: unknown): void {
+  io.stderr.write(`dsh: ${error instanceof Error ? error.message : String(error)}\n`)
+  io.exit(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}.
+ * Run one task through a freshly created Agent and request process exit.
+ * @param ctx - plugin context carrying the Agent, default model, Session, and launcher IO services.
+ * @param task - one-shot task text.
+ * @param io - process-facing effects.
+ */
+async function run(ctx: Context, task: string, io: HeadlessIo): Promise<void> {
+  // Loader siblings mount concurrently. Await the complete application before
+  // creating an Agent so its scoped tools and adapters are not half-composed.
+  await ctx.get('loader')?.await()
+  const agents = ctx.get('agents')
+  const defaultModel = ctx.get('agentDefaultModel')
+  const sessions = ctx.get('sessions')
+  // Early process shutdown can dispose the tree while settlement is pending.
+  if (agents === undefined || defaultModel === undefined || sessions === undefined) return
+
+  const selection = defaultModel.currentSelection()
+  const { agent } = await agents.create({
+    sessionId: SessionId(`session-${randomUUID()}`),
+    meta: { cwd: process.cwd() },
+    agentOptions: { provider: selection.provider, model: selection.model },
+    setup: (agentCtx) => {
+      const selected: ModelSelectionRef = { current: selection, assembled: undefined }
+      installModelSelection(agentCtx, selected)
+    },
+  })
+  await agent.whenIdle()
+  const firstSeq = agent.session.seq
+  agent.followup(createUserMessage({
+    content: [{ type: 'text', text: task }],
+    source: { kind: 'user' },
+  }))
+  await agent.whenIdle()
+  await sessions.flush(agent.session)
+  const outcome = summarize(agent.session.events, firstSeq)
+  io.stdout.write(outcome.text + '\n')
+  io.exit(outcome.reason === 'completed' ? 0 : 1)
+}
+
+/**
+ * Mount the one-shot direct driver.
+ * @param ctx - plugin context carrying core services and the launcher-owned IO seam.
+ * @param config - validated task config.
  */
 export function apply(ctx: Context, config: Config): void {
   const io = ctx.headlessIo
   if (io === undefined) {
     throw new Error('headless-runner: the launcher must provide ctx.headlessIo before the tree mounts')
   }
-  // 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 Loader mounts sibling rows concurrently and this plugin's inject
-    // gate covers only apiProxy/httpServer; prompting before the agent loop,
-    // adapters, and tools settle would fail the turn on a half-mounted tree.
-    // The old launcher ran strictly after settled boot — preserve that.
-    // A tree disposed mid-settlement (early SIGTERM) has nothing to run.
-    await ctx.get('loader')?.await()
-    if (ctx.get('httpServer') === undefined) return
-    // The headless session is web-observable while it runs (same composition).
-    io.stderr.write(`dsh: observing at http://127.0.0.1:${String(ctx.httpServer.port)}\n`)
-    const api = new InProcessApiClient(toFetchHandler(ctx.apiProxy))
-    const created = await unwrap(await api.sessions.create({}), io)
-    // Open the stream before prompting so no frame is lost. The quiescence
-    // anchor below is an in-process ctx subscription, so a remote-carrier
-    // port of this runner must replace it with a wire-visible idle signal.
-    const abort = new AbortController()
-    const frames = api.events.mux({}, abort.signal)
-    const idle = new Promise<number>((resolve) => {
-      ctx.on('agent/status', ({ agent, status }) => {
-        if (agent.id === created.sessionId && status === 'idle') resolve(agent.session.seq - 1)
-      })
-    })
-    const done = consumeUntilIdle(frames, created.sessionId, idle, io)
-    await unwrap(await api.sessions.prompt({
-      sessionId: created.sessionId,
-      mode: 'queue',
-      content: [{ type: 'text', text: config.task }],
-    }), io)
-    const outcome = await done
-    io.stdout.write(outcome.text + '\n')
-    abort.abort()
-    io.exit(outcome.reason === 'completed' ? 0 : 1)
-  })()
+  void run(ctx, config.task, io).catch((error: unknown) => { fail(io, error) })
 }