feat(headless): add stdin task, --session-id, and --json run events

.agents/notes/implemented/feature/2026-09-09-headless-machine-readable-run-surface.i18n.yaml

@@ -0,0 +1,6 @@
+# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
+# side as of the last confirmed-consistent state. Both languages carry equal authority;
+# after editing either side, bring the other along and re-record with:
+#   pnpm run verify-translation-pairing --write .agents/notes/implemented/feature/2026-09-09-headless-machine-readable-run-surface.md
+2026-09-09-headless-machine-readable-run-surface.md: b188c0afee35b3cbb17c374f61194126d1e85be7
+2026-09-09-headless-machine-readable-run-surface.zh.md: 47cf843a612a174564f766265bb599e948b9527a

.agents/notes/implemented/feature/2026-09-09-headless-machine-readable-run-surface.md

@@ -0,0 +1,97 @@
+# Agent Note: Headless machine-readable run surface
+
+Status: implemented
+
+English | [中文](2026-09-09-headless-machine-readable-run-surface.zh.md)
+
+## Problem
+
+`dsh --profile headless` serves a human terminal: the task arrives only through argv, stdout carries one final assistant message, provider reasoning streams to stderr, and every run creates a fresh random session. [Headless is a direct core entry point](../../archived/architecture/2026-08-09-headless-direct-core-entry-point.md) owns that transport and completion contract; [headless reasoning progress](../../archived/feature/2026-08-21-headless-reasoning-progress.md) owns the stderr projection.
+
+A supervising process that drives one headless process per wake, such as an external agent runtime, needs three things that contract does not provide. It needs the task over a private pipe rather than argv, because a long prompt exceeds the argument limit and argv is visible to other processes. It needs a machine-readable stream that separates assistant text, reasoning, tool calls and results, turn boundaries, and usage, because scraping stderr yields only reasoning and the final stdout line yields no tool activity. It needs an exact session identity it can pass back on the next wake, because a fresh random session per process makes continuity impossible.
+
+## Decision
+
+The `dsh-headless` bundle owns an opt-in machine-readable run surface. The default invocation keeps the previous contract unchanged: one final assistant message on stdout, reasoning on stderr, exit 0 exactly when the terminal `turn/end` reason is `completed`.
+
+Three additions extend the app-owned command line that [Apps own their command lines](../../archived/architecture/2026-08-06-app-owned-command-line.md) established:
+
+- `--json` replaces the stdout payload with newline-delimited JSON run events. Reasoning becomes an event instead of stderr output, so stderr carries only `dsh:` diagnostics.
+- `--session-id <id>` selects the exact session identity: adopt the persisted session when it exists, otherwise create it. Without the flag the run mints `session-<uuid>` as before.
+- The task text also arrives on stdin when no positional task is present, or when the positional is `-`.
+
+A per-run `--model` override is deliberately out of scope; the composition default stays authoritative.
+
+The change is confined to `packages/bundle/headless`: `src/startup.ts`, `src/index.ts`, the new `src/json-stream.ts`, the package manifest and `tsconfig.json`, and its tests. No core session, persistence, session-controller, base composition, or launcher file changes.
+
+### Command-line contract
+
+```text
+dsh --profile headless [--json] [--session-id <id>] [<task>... | -]
+```
+
+Task resolution order: joined positionals, then `-`, then piped stdin. A terminal stdin with no positional task remains a usage error, so an interactive invocation cannot hang waiting for input.
+
+`--json` changes the stdout payload and the destination of the reasoning projection only. Exit status, shutdown ordering, session flush, and the durable session log are unchanged, so a supervisor classifies a run exactly as it does today.
+
+### Event stream
+
+`--json` writes one JSON object per line to stdout and nothing else. The vocabulary is a projection of the session event log, not the log itself.
+
+| `type` | Fields | Emitted |
+|---|---|---|
+| `session` | `sessionId`, `cwd` | first line, before any model output |
+| `status` | `phase` (`turn_start`, `step_start`, `step_end`, `turn_end`), `turn`, `step`, `usage`, `reason` | one per boundary |
+| `text` | `text` | assistant text delta |
+| `thinking` | `text` | reasoning delta |
+| `tool_call` | `callId`, `tool`, `input` | once per call |
+| `tool_result` | `callId`, `status`, `result` | once per call |
+| `error` | `message` | process-level failure outside a turn |
+| `final` | `text` | last line |
+
+Projection rules:
+
+- Assistant text and reasoning deltas coalesce until 100 ms or 512 bytes accumulate, so a streaming reply stays live without one line per token.
+- The assembled assistant message is not repeated after its deltas. `user/message` echoes and internal session events (title, model selection, projection, checkpoint, goal, subagent) are not projected.
+- Every projected string is bounded at 8 KiB; an event with a cut string carries `truncated: true`. `tool_result` carries no spill path, because the tool already appends its own truncation notice to the result text.
+- `usage` appears on `step_end`, matching the token accounting a provider reports per step.
+- Raw session events stay out of scope. A debug escape hatch can be added later without changing this vocabulary.
+
+### Session identity
+
+The runtime owns identity. A run without `--session-id` mints `session-<uuid>` and reports it in the first event. A supervisor persists that value and passes it back on the next wake.
+
+`--session-id <id>` is adopt-or-create: observe the persisted session, resume it when it exists, create it otherwise. Create-only would fail the second run, because the JSONL store rejects an existing log id ([session persistence](../../implemented/architecture/2026-06-14-session-persistence.md)).
+
+Adoption compares the persisted session's recorded cwd with the process cwd, since sessions are organized per project directory ([project session directories](../../implemented/architecture/2026-07-24-project-session-directories.md)). A mismatch exits 1 with a `dsh:` diagnostic instead of silently continuing a conversation rooted elsewhere. A session linked to a parent or subagent is rejected. Two live processes cannot write one id; the store's write lease already rejects the second writer. The runner reads the observation through the composed `sessionQuery` service and fails loudly when `--session-id` is requested without it.
+
+## Consequences
+
+What landed: `src/startup.ts` parses `--json` and `--session-id <id>`, treats an absent or `-` task as "read stdin", and raises the usage error only when stdin is a terminal. `src/index.ts` resolves the task, adopts or creates the exact session, and wires either the stderr reasoning projection or the new `src/json-stream.ts` projection. `cordis.patch.yml` forwards the two new settings.
+
+- Default mode is unchanged: a text-only run writes one final assistant line to stdout and nothing to stderr, and exit status still follows the terminal reason.
+- `--json` stdout parses line by line as JSON, starts with `session`, ends with `final`, and contains no plain text. Stderr carries no reasoning in this mode.
+- Two consecutive runs with the same `--session-id` share history. A run whose cwd differs from the persisted session exits 1 with a diagnostic.
+- A piped task with no positional task is honored, and an interactive invocation without a task still fails with the usage error.
+- Unit coverage lands in `packages/bundle/headless/tests/startup.spec.ts`, `tests/headless.spec.ts`, and `tests/json-stream.spec.ts`. The product headless profile expectation test in `apps/cli/tests/profiles/headless/tests/headless.expected.e2e.ts` covers both output modes end to end.
+
+Deferred and open:
+
+- The per-run `--model` override is unimplemented. A later change must respect the session-local selection precedence owned by the Session Controller rather than overriding a stored selection.
+- Cold start plus log replay grows with session length, so a long-lived conversation pays more per wake than a fresh one.
+- `--json` moves reasoning from stderr to stdout, so a log collector that watches stderr sees nothing on a reasoned run in that mode.
+- Bounded `tool_result` payloads hide full output from the supervisor; the 8 KiB cap is owned by `src/json-stream.ts` and should stay a single constant.
+
+## Alternatives considered
+
+**`--verbose` human text on stderr.** A supervisor parses stdout, so a stderr-only projection is invisible to it. Default-mode stderr reasoning already is the human verbose surface.
+
+**Dump raw session events.** They repeat the assembled message beside its deltas, echo `user/message`, and include internal events. Measured on one prompt, pi's delta stream produced 84 lines and 11.7 KB against opencode's 3 lines and 962 B, with roughly a quarter of pi's bytes spent repeating one message across `message_end`, `turn_end`, and `agent_end`.
+
+**A long-lived SDK process instead of one process per wake.** The SDK already speaks structured events and create-or-adopt identity, but it replaces the one-process-per-wake model the supervisor is built on. Measured cold start for the headless profile is about 0.45 s warm and 1.2 s cold, small against a real turn.
+
+**Let the supervisor mint the session id.** Identity belongs to the runtime that owns the log. The supervisor records what the first event reports.
+
+**Create-only `--session-id`.** The second wake would fail against the existing log, which is the opposite of the continuity the flag exists for.
+
+**Task from argv only.** Long prompts exceed `ARG_MAX` and expose the prompt in the process list.

.agents/notes/implemented/feature/2026-09-09-headless-machine-readable-run-surface.zh.md

@@ -0,0 +1,97 @@
+# Agent Note: Headless 的机器可读运行接口
+
+Status: implemented
+
+[English](2026-09-09-headless-machine-readable-run-surface.md) | 中文
+
+## 问题
+
+`dsh --profile headless` 面向的是人类终端：任务只能通过 argv 传入，stdout 只输出最终一条助手消息，provider 的推理过程流式写到 stderr，而且每次运行都新建一个随机会话。[Headless is a direct core entry point](../../archived/architecture/2026-08-09-headless-direct-core-entry-point.md) 拥有那套传输与完成契约；[headless reasoning progress](../../archived/feature/2026-08-21-headless-reasoning-progress.md) 拥有 stderr 投影。
+
+一个"每次唤醒起一个 headless 进程"的监督进程（例如外部 agent 运行时）需要三样该契约没有提供的东西。它需要通过私有管道而不是 argv 传入任务，因为长提示词会超出参数上限，而 argv 对其他进程可见。它需要一条机器可读的流，把助手文本、推理、工具调用与结果、轮次边界和用量区分开，因为抓取 stderr 只能拿到推理，而 stdout 最后一行拿不到任何工具活动。它需要一个精确的会话身份，以便在下一次唤醒时传回去，因为每次进程都新建随机会话意味着无法连续。
+
+## 决策
+
+`dsh-headless` bundle 拥有一个可选的机器可读运行接口。默认调用保持原有契约不变：stdout 输出一条最终助手消息，推理走 stderr，当且仅当终端 `turn/end` 原因为 `completed` 时退出码为 0。
+
+三项新增扩展 [Apps own their command lines](../../archived/architecture/2026-08-06-app-owned-command-line.md) 确立的 app 自有命令行：
+
+- `--json` 把 stdout 负载换成逐行 JSON 运行事件。推理变成一条事件而不再写 stderr，因此该模式下 stderr 只承载 `dsh:` 诊断。
+- `--session-id <id>` 选定精确的会话身份：已存在持久化会话就采用它，否则创建。不带该 flag 时，运行仍像以前一样生成 `session-<uuid>`。
+- 没有位置参数、或者位置参数为 `-` 时，任务文本改从 stdin 读取。
+
+每次运行的 `--model` 覆盖被明确排除在范围之外；组合默认模型仍然权威。
+
+改动范围限于 `packages/bundle/headless`：`src/startup.ts`、`src/index.ts`、新增的 `src/json-stream.ts`、包清单与 `tsconfig.json`，以及测试。不修改任何 core session、持久化、session-controller、base 组合或 launcher 文件。
+
+### 命令行契约
+
+```text
+dsh --profile headless [--json] [--session-id <id>] [<task>... | -]
+```
+
+任务解析顺序：拼接后的位置参数，其次是 `-`，其次是管道 stdin。终端 stdin 且无位置参数仍是用法错误，因此交互式调用不会挂起等待输入。
+
+`--json` 只改变 stdout 负载和推理投影的去向。退出码、关闭顺序、会话 flush 和持久化会话日志都不变，因此监督进程对一次运行的分类方式与现在完全一致。
+
+### 事件流
+
+`--json` 向 stdout 每行写一个 JSON 对象，不写其他内容。词汇表是会话事件日志的投影，而不是日志本身。
+
+| `type` | 字段 | 发出时机 |
+|---|---|---|
+| `session` | `sessionId`、`cwd` | 第一行，先于任何模型输出 |
+| `status` | `phase`（`turn_start`、`step_start`、`step_end`、`turn_end`）、`turn`、`step`、`usage`、`reason` | 每个边界一条 |
+| `text` | `text` | 助手文本增量 |
+| `thinking` | `text` | 推理增量 |
+| `tool_call` | `callId`、`tool`、`input` | 每次调用一条 |
+| `tool_result` | `callId`、`status`、`result` | 每次调用一条 |
+| `error` | `message` | 轮次之外的进程级失败 |
+| `final` | `text` | 最后一行 |
+
+投影规则：
+
+- 助手文本与推理增量按 100 ms 或 512 字节合并，因此流式回复保持实时，又不会每个 token 一行。
+- 增量发完之后不再重复整条助手消息。`user/message` 回显和内部会话事件（标题、模型选择、投影、检查点、目标、子 agent）都不投影。
+- 每个被投影的字符串都限制在 8 KiB；被截断的事件带 `truncated: true`。`tool_result` 不带 spill 路径，因为工具已经把自己的截断提示追加到结果文本里。
+- `usage` 出现在 `step_end` 上，对应 provider 每步上报的 token 计量。
+- 原始会话事件不在范围内。调试用的逃生口可以以后再加，不必改动这套词汇表。
+
+### 会话身份
+
+身份由运行时拥有。不带 `--session-id` 的运行生成 `session-<uuid>`，并在第一条事件里报告它。监督进程保存该值，并在下一次唤醒时传回。
+
+`--session-id <id>` 是采用或创建：先观察持久化会话，存在就 resume，不存在就 create。只创建会让第二次运行失败，因为 JSONL 存储拒绝已存在的日志 id（见 [session persistence](../../implemented/architecture/2026-06-14-session-persistence.zh.md)）。
+
+采用时会比较持久化会话记录的 cwd 与进程 cwd，因为会话按项目目录组织（见 [project session directories](../../implemented/architecture/2026-07-24-project-session-directories.zh.md)）。不一致时以 `dsh:` 诊断退出 1，而不是静默续接一个根目录在别处的会话。带父会话或子 agent 关联的会话被拒绝。两个存活进程不能写同一个 id；存储的写租约已经会拒绝第二个写入者。runner 通过已组合的 `sessionQuery` 服务读取观察结果，并在请求 `--session-id` 却没有该服务时显式失败。
+
+## 后果
+
+实际落地：`src/startup.ts` 解析 `--json` 与 `--session-id <id>`，把缺失或为 `-` 的任务视为"从 stdin 读取"，并且只在 stdin 是终端时抛出用法错误。`src/index.ts` 解析任务、采用或创建精确会话，并接上 stderr 推理投影或新的 `src/json-stream.ts` 投影。`cordis.patch.yml` 转发这两个新设置。
+
+- 默认模式不变：纯文本运行向 stdout 写一行最终助手消息、stderr 无输出，退出码仍跟随终端原因。
+- `--json` 的 stdout 逐行可解析为 JSON，以 `session` 开头、以 `final` 结尾，不含纯文本。该模式下 stderr 不承载推理。
+- 两次连续的相同 `--session-id` 运行共享历史。cwd 与持久化会话不一致的运行以诊断退出 1。
+- 无位置参数但 stdin 有管道输入时任务被采纳，而交互式无任务调用仍以用法错误失败。
+- 单元覆盖落在 `packages/bundle/headless/tests/startup.spec.ts`、`tests/headless.spec.ts` 与 `tests/json-stream.spec.ts`。`apps/cli/tests/profiles/headless/tests/headless.expected.e2e.ts` 的产品 headless profile 期望测试端到端覆盖两种输出模式。
+
+延期与未决：
+
+- 每次运行的 `--model` 覆盖尚未实现。后续改动必须尊重 Session Controller 拥有的会话局部选择优先级，而不是覆盖已保存的选择。
+- 冷启动加上日志重放会随会话变长而增长，因此长会话每次唤醒的代价高于新会话。
+- `--json` 把推理从 stderr 移到 stdout，因此只监听 stderr 的日志收集器在该模式的有推理运行上什么都看不到。
+- 有界的 `tool_result` 负载会让监督进程看不到完整输出；8 KiB 上限由 `src/json-stream.ts` 拥有，应保持单一常量。
+
+## 备选方案
+
+**`--verbose` 人类可读文本写到 stderr。** 监督进程解析的是 stdout，只落在 stderr 的投影对它不可见。默认模式的 stderr 推理已经是人类可读的 verbose 面。
+
+**直接倾倒原始会话事件。** 它们在增量之外重复整条已组装消息，回显 `user/message`，还夹带内部事件。在同一条提示词上实测，pi 的增量流产生 84 行、11.7 KB，而 opencode 是 3 行、962 B；pi 大约四分之一的字节花在把同一条消息在 `message_end`、`turn_end`、`agent_end` 里重复三遍。
+
+**用长驻 SDK 进程代替每次唤醒一个进程。** SDK 已经有结构化事件和采用或创建的身份语义，但它会替换掉监督进程所依赖的"一次唤醒一个进程"模型。实测 headless profile 的冷启动约为热态 0.45 s、冷态 1.2 s，相对一个真实轮次很小。
+
+**让监督进程生成会话 id。** 身份属于拥有日志的运行时。监督进程记录第一条事件报告的值即可。
+
+**只创建语义的 `--session-id`。** 第二次唤醒会撞上已存在的日志，与该 flag 存在的目的正好相反。
+
+**任务只走 argv。** 长提示词会超出 `ARG_MAX`，并且把提示词暴露在进程列表里。

apps/cli/tests/profiles/headless/tests/headless.expected.e2e.ts

@@ -251,6 +251,36 @@ describe('headless stream-json snapshots', () => {
     expect(result.stderr).toBe(await readFile(headlessReasoningExpected, 'utf8'))
   }, LOADER_SMOKE_TEST_TIMEOUT_MS)
 
+  it('projects the same run as JSON events with an exact session identity', async () => {
+    const task = 'Prove the machine-readable product headless profile path.'
+    const result = await runLoaderSmoke({
+      label: 'product headless profile json snapshot',
+      tempDirPrefix: 'headless-snapshot-profile-json-',
+      binScript: dshBinScript,
+      configPath: headlessOverlayPath,
+      binArgs: [
+        '--profile', 'headless', '--patch', headlessOverlayPath,
+        '--json', '--session-id', 'headless-json-session', task,
+      ],
+      tsconfigPath,
+      env: {
+        DSH_PERMISSION_MODE: 'danger-full-access',
+        DSH_TELEMETRY_DISABLED: '1',
+        NODE_OPTIONS: [process.env.NODE_OPTIONS, '--disable-warning=ExperimentalWarning'].filter(Boolean).join(' '),
+      },
+    })
+
+    const events = result.stdout.trim().split('\n').map(line => JSON.parse(line) as JsonObject)
+    expect(events[0]).toMatchObject({ type: 'session', sessionId: 'headless-json-session' })
+    expect(typeof events[0]?.cwd).toBe('string')
+    expect(events.at(-1)).toMatchObject({ type: 'final', text: 'CLI tool round trip complete: CLI_TOOL_ROUND_TRIP' })
+    expect(events.map(event => event.type)).toContain('thinking')
+    expect(events.map(event => event.type)).toContain('tool_call')
+    expect(events.map(event => event.type)).toContain('tool_result')
+    expect(events.map(event => event.type)).not.toContain('error')
+    expect(result.stderr).toBe('')
+  }, LOADER_SMOKE_TEST_TIMEOUT_MS)
+
   it('prints a terminal model failure through the product headless profile command', async () => {
     const result = await runLoaderSmoke({
       label: 'product headless profile model failure snapshot',

docs/config-catalog.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write docs/config-catalog.md
-config-catalog.md: 50626a7d85aa54f6d2e3cdb6140b3af8081d1461
-config-catalog.zh.md: d2ae35b91a4a83da882e8ec5af82a505aa5d5bd1
+config-catalog.md: c2f1b9a72980b78a687e407f62ef208c97042988
+config-catalog.zh.md: 311d52f570493db8e67eda46cb7e7afa2a133f6d

docs/config-catalog.md

@@ -817,14 +817,18 @@ Source: [`packages/goal/goal/src/index.ts:172`](../packages/goal/goal/src/index.
 Requires: `agentDefaultModel` · `agents` · `sessions`
 
 ```ts config-catalog
-/** Plugin config: the task resolved from this app's injected provider service. */
+/** Plugin config: the task and run options resolved from this app's injected provider service. */
 export interface Config {
-  /** The prompt text for the single run. */
-  task: string
+  /** The prompt text for the single run; absent when the task arrives on stdin. */
+  task?: string
+  /** Exact Session identity to adopt or create; absent for a fresh random identity. */
+  sessionId?: string
+  /** Whether stdout carries the machine-readable event stream instead of final text. */
+  json?: boolean
 }
 ```
 
-Source: [`packages/bundle/headless/src/index.ts:34`](../packages/bundle/headless/src/index.ts)
+Source: [`packages/bundle/headless/src/index.ts:40`](../packages/bundle/headless/src/index.ts)
 
 <a id="deepseek-aidsh-hooks-claude-code"></a>
 

docs/config-catalog.zh.md

@@ -819,14 +819,18 @@ export interface Config {
 需要：`agentDefaultModel` · `agents` · `sessions`
 
 ```ts config-catalog
-/** Plugin config: the task resolved from this app's injected provider service. */
+/** Plugin config: the task and run options resolved from this app's injected provider service. */
 export interface Config {
-  /** The prompt text for the single run. */
-  task: string
+  /** The prompt text for the single run; absent when the task arrives on stdin. */
+  task?: string
+  /** Exact Session identity to adopt or create; absent for a fresh random identity. */
+  sessionId?: string
+  /** Whether stdout carries the machine-readable event stream instead of final text. */
+  json?: boolean
 }
 ```
 
-来源：[`packages/bundle/headless/src/index.ts:34`](../packages/bundle/headless/src/index.ts)
+来源：[`packages/bundle/headless/src/index.ts:40`](../packages/bundle/headless/src/index.ts)
 
 <a id="deepseek-aidsh-hooks-claude-code"></a>
 

packages/bundle/headless/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/bundle/headless/README.md
-README.md: 98a9cb2294d8b05a40287c990b3ff53f755747cf
-README.zh.md: 0f4721d856357a85964a13433c3bbc701f8fb299
+README.md: bafe5fddeca801b06e3227cd7df5ed0401e7dd25
+README.zh.md: 1cd2badf6a7ecaf4a7ce8f6ac8f4ced41af51cf5

packages/bundle/headless/README.md

@@ -9,7 +9,7 @@ English | [中文](README.zh.md)
 
 ## Summary
 
-`dsh-headless` runs one dsh task from the command line and prints the final answer, then exits — no GUI, no server, no browser. Type `dsh --profile headless "run the tests"` and the agent works through the task with the same model, tools, and safety defaults as every other surface. It is ideal for scripts, CI, and one-off jobs: the process opens no ports and leaves nothing running behind. The exit code tells you the outcome — 0 when the task completed, 1 when it aborted or errored. The main boundary: one task per invocation, with no interactive follow-up.
+`dsh-headless` runs one dsh task from the command line and prints the final answer, then exits — no GUI, no server, no browser. Type `dsh --profile headless "run the tests"` and the agent works through the task with the same model, tools, and safety defaults as every other surface. It suits scripts, CI, and one-off jobs: it opens no ports and leaves nothing behind. It also offers a JSON event stream (`--json`) and a caller-chosen identity (`--session-id`). Exit code 0 means the task completed; 1 means it aborted or errored. The boundary: one task per invocation, with no interactive follow-up.
 
 ## Table of Contents
 
@@ -25,7 +25,7 @@ English | [中文](README.zh.md)
 <a id="use-this-package"></a>
 ## Use this package
 
-Run one task, get the final answer, and exit. The task is the command line itself, so the whole invocation is the smallest working example.
+Run one task, get the final answer, and exit. The task is the command-line argument, or stdin when you omit it; the whole invocation is the smallest working example.
 
 ### Running a one-shot task
 
@@ -33,21 +33,37 @@ Run one task, get the final answer, and exit. The task is the command line itsel
 dsh --profile headless "run the tests"
 ```
 
-The agent works through the task, streams each non-empty provider reasoning delta to stderr under a `dsh: reasoning:` heading, then prints the final answer on stdout and exits. Consecutive reasoning deltas stay in one section, and the runner closes that section before later output when the provider supplied no trailing newline. A successful run without reasoning keeps stderr empty; a failure exits 1 and prints `dsh: <code>: <message>` to stderr. A missing or blank task is rejected before anything runs. The task text is supplied through the single `task` setting:
+The agent works through the task, streams each non-empty provider reasoning delta to stderr under a `dsh: reasoning:` heading, then prints the final answer on stdout and exits. Consecutive reasoning deltas stay in one section, and the runner closes that section before later output when the provider supplied no trailing newline. A successful run without reasoning keeps stderr empty; a failure exits 1 and prints `dsh: <code>: <message>` to stderr. The task comes from the positional argument, or from stdin when the argument is omitted or is `-`; a blank argument or an empty pipe is rejected before anything runs.
+
+```sh
+git diff --stat | dsh --profile headless "summarize these changes"
+```
+
+The task and run options are supplied through three settings:
 
 | Field | Default | Meaning |
 |---|---|---|
-| `task` | required | The task text for the single run |
+| `task` | stdin | The task text; stdin supplies it when omitted or `-` |
+| `sessionId` | `session-<uuid>` | Exact Session identity to adopt or create |
+| `json` | `false` | Project the run as newline-delimited events on stdout |
 
 The generated [configuration catalog](../../../docs/config-catalog.md#deepseek-aidsh-headless) is the exhaustive source for every accepted field and its JSDoc.
 
+### Choosing the session identity
+
+Every invocation defaults to a fresh `session-<uuid>` identity. Pass `--session-id <id>` to name it yourself: the runner adopts the persisted Session with that id when one exists, and creates it otherwise. Adoption is scoped to the current working directory and refuses a Session owned by a subagent, so a supervisor cannot silently drive someone else's conversation; either mismatch fails before the task runs.
+
+### Machine-readable output
+
+`--json` replaces the final-text stdout line with a newline-delimited JSON event stream, while stderr keeps only the `dsh:` diagnostics. The stream opens with `session` (carrying the identity the run used) and closes with `final`, and carries `status`, `text`, `thinking`, `tool_call`, and `tool_result` events in between. Streamed text and thinking deltas are coalesced before they are written, and every string is capped at 8 KiB and flagged with `truncated`. A process-level failure outside a turn is reported as an `error` event on stdout in addition to the `dsh:` stderr line.
+
 ### When to use it
 
-Use headless for scripted or automated dsh runs — CI steps, batch jobs, quick answers from a terminal. Avoid it when you need a multi-turn interactive session or a GUI; the browser surface ([dsh-web-app](../web-app/README.md)) serves that. The process stays alive only for the run, opens no listening port, and exits on its own, so it fits pipelines that wait on the process.
+Use headless for scripted or automated dsh runs — CI steps, batch jobs, quick answers from a terminal. Avoid it when you need a multi-turn interactive session or a GUI; the browser surface ([dsh-web-app](../web-app/README.md)) serves that. The process stays alive only for the run, opens no listening port, and exits on its own, so it fits pipelines that wait on the process. When a supervisor needs progress rather than just the answer, `--json` gives it the event stream and `--session-id` lets a later invocation continue the same conversation.
 
 ### Help and task errors
 
-`dsh --profile headless --help` prints the command's help text and exits without running anything. A missing or whitespace-only task is a usage error: nothing runs and the process exits 1.
+`dsh --profile headless --help` prints the command's help text and exits without running anything. A missing or whitespace-only task is a usage error when stdin is a terminal: nothing runs and the process exits 1. When stdin is not a terminal the runner reads the task from it instead and rejects an empty result the same way.
 
 -----
 
@@ -61,25 +77,27 @@ The runner is a direct driver over the core API carrier: it creates one fresh Ag
 
 ### Run flow
 
-The runner awaits the complete application (`ctx.get('loader')?.await()`) so the composed tools and adapters are not half-mounted, reads the shared [`agentDefaultModel`](../../core/agent-default-model/README.md) selection, creates one fresh persisted Agent with that provider and model, and submits the task as an ordinary user message. It streams that Agent's non-empty reasoning deltas to stderr, waits for quiescence, then flushes the Session and folds the owned interval (`firstSeq` onward) into the last non-empty `assistant/message` text and final `turn/end` reason. It writes the final text to stdout and requests exit.
+The runner awaits the complete application (`ctx.get('loader')?.await()`) so the composed tools and adapters are not half-mounted, reads the shared [`agentDefaultModel`](../../core/agent-default-model/README.md) selection, resolves the task from config or stdin, then creates the exact Agent identity: a fresh `session-<uuid>` by default, or the id `--session-id` names, which it adopts through [`sessionQuery`](../../session-query/session-query/README.md) when a persisted log exists and creates otherwise. It submits the task as an ordinary user message. Without `--json` it streams that Agent's non-empty reasoning deltas to stderr; with `--json` it projects the run instead. It waits for quiescence, then flushes the Session and folds the owned interval (`firstSeq` onward) into the last non-empty `assistant/message` text and final `turn/end` reason. It writes the final text to stdout (or the `final` event) and requests exit.
 
 ### Patch surface over base
 
-The patch rides over `dsh-base`: it inherits the projection cache, sets the coding persona prefix and separate cwd suffix on the base `system-prompt` row, keeps the same temporary process-wide PTC mode opt-in (`DSH_TOOLS_MODE`) as the Web surface, disables the shared HMR row, inserts PTC mode's worker as a core execution capability, and mounts the startup provider and the runner. The cache checkpoints each persisted one-shot session for later consumers; its durability barrier flushes each covered log prefix before publishing the cache row and may split otherwise coalesced JSONL runs. The startup provider ([`src/startup.ts`](src/startup.ts)) injects `ctx.cmdlineArgs` ([`dsh-cmdline`](../../boot/cmdline/README.md)), reads the positional argument, prints the app's `--help`, and provides `headlessStartup`; the runner injects that service and reads its task from lazy config.
+The patch rides over `dsh-base`: it inherits the projection cache, sets the coding persona prefix and separate cwd suffix on the base `system-prompt` row, keeps the same temporary process-wide PTC mode opt-in (`DSH_TOOLS_MODE`) as the Web surface, disables the shared HMR row, inserts PTC mode's worker as a core execution capability, and mounts the startup provider and the runner. The cache checkpoints each persisted one-shot session for later consumers; its durability barrier flushes each covered log prefix before publishing the cache row and may split otherwise coalesced JSONL runs. The startup provider ([`src/startup.ts`](src/startup.ts)) injects `ctx.cmdlineArgs` ([`dsh-cmdline`](../../boot/cmdline/README.md)), reads the positional argument and the `--session-id`/`--json` options, prints the app's `--help`, and provides `headlessStartup`; the runner injects that service and reads its task and run options from lazy config.
 
 ### Exit mapping
 
-A completed final `turn/end` exits 0; any other outcome — aborted, error, or no turn in the owned interval — exits 1. An `error` reason also writes `dsh: <code>: <message>` to stderr. A direct driver failure (for example, Agent creation) writes `dsh: <message>` to stderr and exits 1.
+A completed final `turn/end` exits 0; any other outcome — aborted, error, or no turn in the owned interval — exits 1. An `error` reason also writes `dsh: <code>: <message>` to stderr. A direct driver failure (for example, Agent creation or an unusable `--session-id`) writes `dsh: <message>` to stderr and exits 1, and in `--json` mode also emits an `error` event.
 
 ### Source map
 
 | File | Role |
 |---|---|
-| [`src/index.ts`](src/index.ts) | The `headless-runner` plugin: run flow, output contract, exit mapping |
-| [`src/startup.ts`](src/startup.ts) | The `headless-startup` provider: task positional and `--help` |
+| [`src/index.ts`](src/index.ts) | The `headless-runner` plugin: run flow, session resolution, output contract, exit mapping |
+| [`src/startup.ts`](src/startup.ts) | The `headless-startup` provider: task positional, `--session-id`, `--json`, and `--help` |
+| [`src/json-stream.ts`](src/json-stream.ts) | The `--json` projection: event vocabulary, coalescing, string bounding |
 | [`cordis.patch.yml`](cordis.patch.yml) | The one-shot patch over `dsh-base` |
 | — | No runtime invariant companion is published; the runner's observable contract (provider reasoning on stderr, final text on stdout, exit code by turn-end reason) is process-level and owned by the launcher e2e; it registers nothing and holds no mutable relation to audit inside the tree. |
-| [`tests/headless.spec.ts`](tests/headless.spec.ts) | Run flow, aggregation, flush, and exit mapping |
+| [`tests/headless.spec.ts`](tests/headless.spec.ts) | Run flow, aggregation, flush, session adoption, and exit mapping |
+| [`tests/json-stream.spec.ts`](tests/json-stream.spec.ts) | Projection ordering, coalescing, bounding, and disposal |
 | [`tests/startup.spec.ts`](tests/startup.spec.ts) | Command-line parsing over a real Loader tree |
 
 ### Invariant ownership
@@ -121,9 +139,11 @@ These limits tell you when headless does not fit and what it needs from the `dsh
 
 - **One task per run** — after the task is answered the process exits; there is no interactive follow-up, so split multi-step work into separate runs.
 - **Runs through the `dsh` launcher** — starting the headless profile another way fails at startup, because only the launcher can request the process exit.
-- **No pre-token heartbeat** — stderr stays silent until the provider emits a non-empty reasoning delta; a delayed first token exposes no earlier progress signal.
-- **Reasoning enters stderr logs** — redirection and supervisors may retain substantially more and potentially sensitive model output; route stderr to a controlled sink when needed.
-- **Only reasoning and the final answer are printed** — a run without an assistant message prints an empty stdout line and exits 1; intermediate tool output is not printed.
+- **No pre-token heartbeat** — in default mode stderr stays silent until the provider emits a non-empty reasoning delta; a delayed first token exposes no earlier progress signal.
+- **Reasoning enters stderr logs** — in default mode, redirection and supervisors may retain substantially more and potentially sensitive model output; route stderr to a controlled sink when needed.
+- **Default stdout carries only the final answer** — a run without an assistant message prints an empty stdout line and exits 1; intermediate tool output is not printed unless you opt into `--json`.
+- **Adoption is cwd- and ownership-scoped** — `--session-id` refuses a Session recorded in another working directory or owned by a subagent, and requires the composed Session query service.
+- **The event stream is a projection, not the log** — `--json` coalesces deltas and caps strings at 8 KiB, so it is not a lossless copy of the Session log.
 
 <a id="dev-note"></a>
 ### Dev Note

packages/bundle/headless/README.zh.md

@@ -9,7 +9,7 @@ kind: "package-bundle"
 
 ## 概述
 
-`dsh-headless` 从命令行运行一个 dsh 任务并打印最终答案，然后退出——没有 GUI、没有服务器、没有浏览器。输入 `dsh --profile headless "run the tests"`，agent（智能体）会以与其他表层相同的模型、工具与安全默认值完成该任务。它非常适合脚本、CI 与一次性任务：进程不打开任何端口，也不会留下任何后台运行的东西。退出码告诉你结果——任务完成时为 0，中止或出错时为 1。主要边界：每次调用只运行一个任务，没有交互式后续。
+`dsh-headless` 从命令行运行一个 dsh 任务并打印最终答案，然后退出——没有 GUI、没有服务器、没有浏览器。输入 `dsh --profile headless "run the tests"`，agent（智能体）会以与其他表层相同的模型、工具与安全默认值完成该任务。它非常适合脚本、CI 与一次性任务：进程不打开任何端口，也不会留下任何后台运行的东西。监督进程还可以通过按行 JSON 事件流（`--json`）驱动它，并用调用方指定的标识（`--session-id`）固定这段对话。退出码告诉你结果——任务完成时为 0，中止或出错时为 1。主要边界：每次调用只运行一个任务，没有交互式后续。
 
 ## 目录
 
@@ -25,7 +25,7 @@ kind: "package-bundle"
 <a id="use-this-package"></a>
 ## 使用本包
 
-运行一个任务，获得最终答案，然后退出。任务就是命令行本身，因此整条命令就是最小的可运行示例。
+运行一个任务，获得最终答案，然后退出。任务就是命令行参数，省略时则来自 stdin；整条命令就是最小的可运行示例。
 
 ### 运行一次性任务
 
@@ -33,21 +33,37 @@ kind: "package-bundle"
 dsh --profile headless "run the tests"
 ```
 
-agent（智能体）会完成该任务，把提供方的每个非空推理增量流式写入 stderr 的 `dsh: reasoning:` 段，然后把最终答案写入 stdout 并退出。连续推理增量保持在同一段中；提供方未给尾换行时，runner 会在后续输出前结束该段。没有推理内容的成功运行保持 stderr 为空；失败时退出码为 1，并以 `dsh: <code>: <message>` 向 stderr 写入错误。缺失或空白任务会在任何内容运行前被拒绝。任务文本通过唯一的 `task` 设置提供：
+agent（智能体）会完成该任务，把提供方的每个非空推理增量流式写入 stderr 的 `dsh: reasoning:` 段，然后把最终答案写入 stdout 并退出。连续推理增量保持在同一段中；提供方未给尾换行时，runner 会在后续输出前结束该段。没有推理内容的成功运行保持 stderr 为空；失败时退出码为 1，并以 `dsh: <code>: <message>` 向 stderr 写入错误。任务来自位置参数，参数省略或为 `-` 时则来自 stdin；空白参数或空管道会在任何内容运行前被拒绝。
+
+```sh
+git diff --stat | dsh --profile headless "summarize these changes"
+```
+
+任务与运行选项通过三个设置提供：
 
 | 字段 | 默认值 | 含义 |
 |---|---|---|
-| `task` | 必填 | 单次运行的任务文本 |
+| `task` | stdin | 任务文本；省略或传 `-` 时由 stdin 提供 |
+| `sessionId` | `session-<uuid>` | 要沿用或创建的精确 Session 标识 |
+| `json` | `false` | 把本次运行投影为 stdout 上的按行 JSON 事件 |
 
 生成的[配置目录](../../../docs/config-catalog.zh.md#deepseek-aidsh-headless)是每个受支持字段及其 JSDoc 的穷尽式真源。
 
+### 选择 Session 标识
+
+每次调用默认使用全新的 `session-<uuid>` 标识。传入 `--session-id <id>` 可自行命名：该 id 对应的持久化 Session 存在时 runner 会沿用，否则创建。沿用被限定在当前工作目录内，并拒绝由 subagent 拥有的 Session，因此监督进程无法悄悄驱动他人的会话；任一不匹配都会在任务运行前失败。
+
+### 机器可读输出
+
+`--json` 用按行 JSON 事件流取代 stdout 的最终文本行，stderr 仅保留 `dsh:` 诊断信息。事件流以 `session`（携带本次运行使用的标识）开头、以 `final` 结尾，其间为 `status`、`text`、`thinking`、`tool_call` 与 `tool_result` 事件。流式的 text 与 thinking 增量在写出前会被合并，每个字符串上限为 8 KiB，超出时标记 `truncated`。轮次之外的进程级失败除了 stderr 的 `dsh:` 行外，还会在 stdout 上以 `error` 事件报告。
+
 ### 何时使用
 
-在脚本化或自动化的 dsh 运行中使用 headless——CI 步骤、批处理任务、从终端快速获取答案。当需要多轮交互会话或 GUI 时请避免它；浏览器表层（[dsh-web-app](../web-app/README.zh.md)）负责这类场景。进程只为本次运行而存活，不打开监听端口，并且自行退出，因此适合等待进程结束的流水线。
+在脚本化或自动化的 dsh 运行中使用 headless——CI 步骤、批处理任务、从终端快速获取答案。当需要多轮交互会话或 GUI 时请避免它；浏览器表层（[dsh-web-app](../web-app/README.zh.md)）负责这类场景。进程只为本次运行而存活，不打开监听端口，并且自行退出，因此适合等待进程结束的流水线。当监督进程需要进度而不只是答案时，`--json` 提供事件流，`--session-id` 则让后续调用继续同一段对话。
 
 ### 帮助与任务错误
 
-`dsh --profile headless --help` 打印该命令的帮助文本并直接退出，不运行任何内容。缺失或只有空白的任务属于用法错误：什么都不运行，进程退出 1。
+`dsh --profile headless --help` 打印该命令的帮助文本并直接退出，不运行任何内容。stdin 是终端时，缺失或只有空白的任务属于用法错误：什么都不运行，进程退出 1。stdin 不是终端时，runner 改从 stdin 读取任务，并以同样方式拒绝空结果。
 
 -----
 
@@ -61,25 +77,27 @@ runner 是核心 API 载体之上的直接驱动器：它通过注册表创建
 
 ### 运行流程
 
-runner 等待整个应用结算（`ctx.get('loader')?.await()`），确保已组合的工具与适配器不会半挂载，读取共享的 [`agentDefaultModel`](../../core/agent-default-model/README.zh.md) 选择，用该 provider 与模型创建一个全新的持久化 Agent（智能体），并把任务作为普通用户消息提交。它把该 Agent 的非空推理增量流式写入 stderr、等待完全停稳，然后 flush Session，并把所属区间（从 `firstSeq` 起）折叠为最后一条非空 `assistant/message` 文本与最终 `turn/end` 原因。最后，它把最终文本写入 stdout 并请求退出。
+runner 等待整个应用结算（`ctx.get('loader')?.await()`），确保已组合的工具与适配器不会半挂载，读取共享的 [`agentDefaultModel`](../../core/agent-default-model/README.zh.md) 选择，从配置或 stdin 解析任务，然后确定精确的 Agent（智能体）标识：默认是全新的 `session-<uuid>`，或 `--session-id` 指定的 id——存在持久化日志时通过 [`sessionQuery`](../../session-query/session-query/README.zh.md) 沿用，否则创建。它把任务作为普通用户消息提交。不带 `--json` 时，它把该 Agent 的非空推理增量流式写入 stderr；带 `--json` 时改为投影本次运行。它等待完全停稳，然后 flush Session，并把所属区间（从 `firstSeq` 起）折叠为最后一条非空 `assistant/message` 文本与最终 `turn/end` 原因。最后，它把最终文本写入 stdout（或 `final` 事件）并请求退出。
 
 ### 叠加在 base 之上的 patch 表层
 
-patch 叠加在 `dsh-base` 之上：继承投影缓存，在基础 `system-prompt` 行上设置编码 persona 前缀与独立的 cwd 后缀，保留与 Web 表层相同的临时进程级 PTC mode 开关（`DSH_TOOLS_MODE`），禁用共享的 HMR 行，把 PTC mode 的 worker 作为核心执行能力插入，并挂载启动提供方与 runner。缓存为每个已持久化的一次性会话写入检查点，供后续消费方使用；其持久性屏障会在发布缓存行前 flush 所覆盖的日志前缀，因此可能拆分原本会合并的 JSONL 行。启动提供方（[`src/startup.ts`](src/startup.ts)）注入 `ctx.cmdlineArgs`（[`dsh-cmdline`](../../boot/cmdline/README.zh.md)），读取位置参数、打印应用自己的 `--help`，并提供 `headlessStartup`；runner 注入该服务，再从惰性配置中读取任务。
+patch 叠加在 `dsh-base` 之上：继承投影缓存，在基础 `system-prompt` 行上设置编码 persona 前缀与独立的 cwd 后缀，保留与 Web 表层相同的临时进程级 PTC mode 开关（`DSH_TOOLS_MODE`），禁用共享的 HMR 行，把 PTC mode 的 worker 作为核心执行能力插入，并挂载启动提供方与 runner。缓存为每个已持久化的一次性会话写入检查点，供后续消费方使用；其持久性屏障会在发布缓存行前 flush 所覆盖的日志前缀，因此可能拆分原本会合并的 JSONL 行。启动提供方（[`src/startup.ts`](src/startup.ts)）注入 `ctx.cmdlineArgs`（[`dsh-cmdline`](../../boot/cmdline/README.zh.md)），读取位置参数与 `--session-id`/`--json` 选项、打印应用自己的 `--help`，并提供 `headlessStartup`；runner 注入该服务，再从惰性配置中读取任务与运行选项。
 
 ### 退出映射
 
-最终 `turn/end` 完成时退出码为 0；任何其他结果——aborted、error，或所属区间内没有轮次——退出码为 1。结束原因为 `error` 时还会向 stderr 写入 `dsh: <code>: <message>`。直接驱动器失败（例如 Agent 创建失败）向 stderr 写入 `dsh: <message>` 并退出 1。
+最终 `turn/end` 完成时退出码为 0；任何其他结果——aborted、error，或所属区间内没有轮次——退出码为 1。结束原因为 `error` 时还会向 stderr 写入 `dsh: <code>: <message>`。直接驱动器失败（例如 Agent 创建失败或不可用的 `--session-id`）向 stderr 写入 `dsh: <message>` 并退出 1，且在 `--json` 模式下额外发出一个 `error` 事件。
 
 ### 源码地图
 
 | 文件 | 职责 |
 |---|---|
-| [`src/index.ts`](src/index.ts) | `headless-runner` 插件：运行流程、输出约定、退出映射 |
-| [`src/startup.ts`](src/startup.ts) | `headless-startup` 提供方：任务位置参数与 `--help` |
+| [`src/index.ts`](src/index.ts) | `headless-runner` 插件：运行流程、Session 解析、输出约定、退出映射 |
+| [`src/startup.ts`](src/startup.ts) | `headless-startup` 提供方：任务位置参数、`--session-id`、`--json` 与 `--help` |
+| [`src/json-stream.ts`](src/json-stream.ts) | `--json` 投影：事件词汇、合并、字符串限长 |
 | [`cordis.patch.yml`](cordis.patch.yml) | 叠加在 `dsh-base` 之上的一次性 patch |
 | — | 不发布运行时不变式伴生入口；可观察的行为属于进程级组合，本包只持有静态 patch 列表。 |
-| [`tests/headless.spec.ts`](tests/headless.spec.ts) | 运行流程、汇总、flush 与退出映射 |
+| [`tests/headless.spec.ts`](tests/headless.spec.ts) | 运行流程、汇总、flush、Session 沿用与退出映射 |
+| [`tests/json-stream.spec.ts`](tests/json-stream.spec.ts) | 投影顺序、合并、限长与释放 |
 | [`tests/startup.spec.ts`](tests/startup.spec.ts) | 在真实 Loader 树上的命令行解析 |
 
 ### 不变式归属
@@ -121,9 +139,11 @@ runner 不向请求前缀添加任何内容；它只是把一条用户消息驱
 
 - **每次运行一个任务**——任务得到回答后进程即退出；没有交互式后续，因此多步工作请拆成多次运行。
 - **通过 `dsh` 启动器运行**——以其他方式启动 headless profile 会在启动时失败，因为只有启动器能请求进程退出。
-- **首个 token 前没有心跳**——提供方发出第一个非空推理增量前，stderr 保持静默；延迟首个 token 的提供方不会更早给出进度信号。
-- **推理进入 stderr 日志**——重定向与监督进程可能保留更多且可能敏感的模型输出；需要时应把 stderr 路由到受控位置。
-- **只打印推理和最终答案**——没有 assistant 消息的运行向 stdout 打印空行并以 1 退出；中间工具输出不会打印。
+- **首个 token 前没有心跳**——默认模式下，提供方发出第一个非空推理增量前 stderr 保持静默；延迟首个 token 的提供方不会更早给出进度信号。
+- **推理进入 stderr 日志**——默认模式下，重定向与监督进程可能保留更多且可能敏感的模型输出；需要时应把 stderr 路由到受控位置。
+- **默认 stdout 只承载最终答案**——没有 assistant 消息的运行向 stdout 打印空行并以 1 退出；中间工具输出不会打印，除非显式启用 `--json`。
+- **沿用受 cwd 与归属限制**——`--session-id` 会拒绝记录在其他工作目录或由 subagent 拥有的 Session，并要求已组合的 Session 查询服务。
+- **事件流是投影而非日志**——`--json` 会合并增量并把字符串限制在 8 KiB，因此它不是 Session 日志的无损副本。
 
 <a id="dev-note"></a>
 ### 开发备注

packages/bundle/headless/cordis.patch.yml

@@ -1,8 +1,9 @@
 # The dsh-headless bundle patch: one-shot task mode directly over dsh-base.
 # It mounts no Host, HTTP server, Web runtime, or browser plugin. An ordinary
 # provider plugin injects `cmdlineArgs`, parses the task positional
-# (`dsh --profile headless "<task>"`) and this app's --help, then the direct
-# driver creates an Agent through the core registry and prints its durable result.
+# (`dsh --profile headless "<task>"`), the `--session-id` and `--json` options,
+# and this app's --help, then the direct driver creates or adopts an Agent
+# through the core registry and prints its durable result.
 
 - id: system-prompt
   config:
@@ -23,9 +24,11 @@
     - id: headless-startup
       name: '@deepseek-ai/dsh-headless/startup'
 
-    # Reads its task from the ordinary headlessStartup provider.
+    # Reads its task and run options from the ordinary headlessStartup provider.
     - id: headless-runner
       name: '@deepseek-ai/dsh-headless'
       inject: [headlessStartup]
       config:
         task: !!js ctx.headlessStartup.task
+        sessionId: !!js ctx.headlessStartup.sessionId
+        json: !!js ctx.headlessStartup.json

packages/bundle/headless/package.json

@@ -52,7 +52,8 @@
     "@deepseek-ai/dsh-agent": "workspace:^",
     "@deepseek-ai/dsh-agent-default-model": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
-    "@deepseek-ai/dsh-session": "workspace:^"
+    "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-session-query": "workspace:^"
   },
   "devDependencies": {
     "@deepseek-ai/cordis": "workspace:^",
@@ -62,6 +63,7 @@
     "@deepseek-ai/dsh-agent-loop": "workspace:^",
     "@deepseek-ai/dsh-agent-loop-testkit": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
-    "@deepseek-ai/dsh-session": "workspace:^"
+    "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-session-query": "workspace:^"
   }
 }

packages/bundle/headless/src/index.ts

@@ -1,9 +1,11 @@
 /**
  * @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,
- * streams provider reasoning to stderr, flushes its Session, prints the final
- * assistant text to stdout, and exits.
+ * creates one Agent through the core registry (or adopts the exact Session a
+ * `--session-id` names), drives the task to quiescence, streams provider
+ * reasoning to stderr, flushes its Session, prints the final assistant text to
+ * stdout, and exits. With `--json` it projects the run as newline-delimited
+ * events instead of the final text.
  *
  * @module @deepseek-ai/dsh-headless
  */
@@ -19,10 +21,14 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm'
 import { assertNever } from '@deepseek-ai/dsh-util-values'
 import { SessionSeq } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
-// Empty type imports carry the loader Context merge for the settlement await
-// and the cmdline Context merge for the appExit host value.
+import { SessionQueryError } from '@deepseek-ai/dsh-session-query'
+// Empty type imports carry the loader Context merge for the settlement await,
+// the cmdline Context merge for the appExit host value, and the sessionQuery
+// Context merge for exact Session adoption.
 import type {} from '@deepseek-ai/cordis-plugin-loader'
 import type {} from '@deepseek-ai/dsh-cmdline'
+import type {} from '@deepseek-ai/dsh-session-query'
+import { projectJsonRun } from './json-stream.ts'
 
 /** Stable Cordis plugin name. */
 export const name = 'headless-runner'
@@ -30,14 +36,20 @@ export const name = 'headless-runner'
 /** Core services required before the one-shot turn can start. */
 export const inject = ['agentDefaultModel', 'agents', 'sessions']
 
-/** Plugin config: the task resolved from this app's injected provider service. */
+/** Plugin config: the task and run options resolved from this app's injected provider service. */
 export interface Config {
-  /** The prompt text for the single run. */
-  task: string
+  /** The prompt text for the single run; absent when the task arrives on stdin. */
+  task?: string
+  /** Exact Session identity to adopt or create; absent for a fresh random identity. */
+  sessionId?: string
+  /** Whether stdout carries the machine-readable event stream instead of final text. */
+  json?: boolean
 }
 
 export const Config: z<Config> = z.object({
-  task: z.string().required(),
+  task: z.string(),
+  sessionId: z.string(),
+  json: z.boolean(),
 })
 
 /** Outcome of one owned run interval. */
@@ -54,10 +66,19 @@ interface HeadlessIo {
   exit(code: number): void
 }
 
-/** The process streams the runner writes to; tests substitute captures. */
-export const internals: { stdout: HeadlessIo['stdout']; stderr: HeadlessIo['stderr'] } = {
+/** The process streams the runner reads and writes; tests substitute captures. */
+export const internals: {
+  stdout: HeadlessIo['stdout']
+  stderr: HeadlessIo['stderr']
+  readStdin: () => Promise<string>
+} = {
   stdout: process.stdout,
   stderr: process.stderr,
+  readStdin: async () => {
+    const chunks: Buffer[] = []
+    for await (const chunk of process.stdin as AsyncIterable<Buffer>) chunks.push(chunk)
+    return Buffer.concat(chunks).toString('utf8')
+  },
 }
 
 /** Aggregate the last assistant text and turn outcome in one owned interval. */
@@ -154,19 +175,67 @@ function streamReasoning(
   }
 }
 
+/**
+ * Resolve the Agent for one run: reuse a live identity, adopt the persisted
+ * Session with the requested id, or create that exact id when no log exists.
+ * @param ctx - plugin context carrying the Session query service.
+ * @param agents - the core Agent registry.
+ * @param sessionId - exact Session identity to adopt or create.
+ * @param agentOptions - provider/model pair for this run.
+ * @param setup - per-Agent scope setup installing the model selection.
+ * @returns the live, resumed, or freshly created Agent.
+ */
+async function resolveAgent(
+  ctx: Context,
+  agents: Context['agents'],
+  sessionId: SessionId,
+  agentOptions: { provider: string; model: string },
+  setup: (agentCtx: Context) => void,
+): Promise<Agent> {
+  const live = agents.get(sessionId)
+  if (live !== undefined) return live
+  const query = ctx.get('sessionQuery')
+  if (query === undefined) {
+    throw new Error('headless --session-id requires the sessionQuery service; dsh-base provides it')
+  }
+  try {
+    using observation = await query.observeSession(sessionId)
+    const header = observation.header
+    if (header.origin === 'subagent' || header.parentSession !== undefined) {
+      throw new Error(`session "${sessionId}" belongs to a subagent and cannot be driven directly`)
+    }
+    if (header.cwd !== process.cwd()) {
+      throw new Error(`session "${sessionId}" was recorded in "${header.cwd}", not "${process.cwd()}"`)
+    }
+    const { agent } = await agents.resume({ resumeSessionId: sessionId, agentOptions, setup })
+    return agent
+  } catch (error: unknown) {
+    if (!(error instanceof SessionQueryError) || error.code !== 'SESSION_QUERY_SESSION_NOT_FOUND') throw error
+  }
+  const { agent } = await agents.create({
+    sessionId,
+    meta: { cwd: process.cwd() },
+    agentOptions,
+    setup,
+  })
+  return agent
+}
+
 /** 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`)
+function fail(io: HeadlessIo, error: unknown, json: boolean): void {
+  const message = error instanceof Error ? error.message : String(error)
+  if (json) io.stdout.write(`${JSON.stringify({ type: 'error', message })}\n`)
+  io.stderr.write(`dsh: ${message}\n`)
   io.exit(1)
 }
 
 /**
- * Run one task through a freshly created Agent and request process exit.
+ * Run one task through one 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 config - task, optional exact Session identity, and output mode.
  * @param io - process-facing effects.
  */
-async function run(ctx: Context, task: string, io: HeadlessIo): Promise<void> {
+async function run(ctx: Context, config: Config, 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()
@@ -176,45 +245,59 @@ async function run(ctx: Context, task: string, io: HeadlessIo): Promise<void> {
   // Early process shutdown can dispose the tree while settlement is pending.
   if (agents === undefined || defaultModel === undefined || sessions === undefined) return
 
+  const task = config.task === undefined || config.task === '-'
+    ? await internals.readStdin()
+    : config.task
+  if (task.trim() === '') {
+    throw new Error('a task is required, for example: dsh --profile headless "run the tests"')
+  }
+
   const selection = defaultModel.currentSelection()
-  // This bundle composes no preset roster, so the model-facing rows sit in the
-  // host plane and the agent reads them from the global layer. A deployment
-  // that DOES configure one has to join it here first
-  // (@deepseek-ai/dsh-agent-presets README, "Composing a child agent").
-  const { agent } = await agents.create({
-    sessionId: brandString<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)
-    },
-  })
+  const agentOptions = { provider: selection.provider, model: selection.model }
+  const setup = (agentCtx: Context): void => {
+    const selected: ModelSelectionRef = { current: selection, assembled: undefined }
+    installModelSelection(agentCtx, selected)
+  }
+  const sessionId = brandString<SessionId>(config.sessionId ?? `session-${randomUUID()}`)
+  const agent = config.sessionId === undefined
+    ? (await agents.create({
+      sessionId,
+      meta: { cwd: process.cwd() },
+      agentOptions,
+      setup,
+    })).agent
+    : await resolveAgent(ctx, agents, sessionId, agentOptions, setup)
   await agent.whenIdle()
   const firstSeq = agent.session.seq
-  const stopReasoning = streamReasoning(ctx, agent, io.stderr)
+  const projection = config.json === true ? projectJsonRun(ctx, agent, io.stdout) : undefined
+  const stopReasoning = projection === undefined ? streamReasoning(ctx, agent, io.stderr) : undefined
   try {
-    agent.followup(createUserMessage({
-      content: [{ type: 'text', text: task }],
-      source: { kind: 'user' },
-    }))
-    await agent.whenIdle()
+    try {
+      agent.followup(createUserMessage({
+        content: [{ type: 'text', text: task }],
+        source: { kind: 'user' },
+      }))
+      await agent.whenIdle()
+    } finally {
+      stopReasoning?.()
+    }
+    await sessions.flush(agent.session)
+    const outcome = summarize(agent.session, firstSeq)
+    if (projection === undefined) io.stdout.write(outcome.text + '\n')
+    else projection.finish(outcome.text)
+    if (outcome.reason?.kind === 'error') {
+      io.stderr.write(`dsh: ${outcome.reason.error.code}: ${outcome.reason.error.message}\n`)
+    }
+    io.exit(outcome.reason?.kind === 'completed' ? 0 : 1)
   } finally {
-    stopReasoning()
-  }
-  await sessions.flush(agent.session)
-  const outcome = summarize(agent.session, firstSeq)
-  io.stdout.write(outcome.text + '\n')
-  if (outcome.reason?.kind === 'error') {
-    io.stderr.write(`dsh: ${outcome.reason.error.code}: ${outcome.reason.error.message}\n`)
+    projection?.dispose()
   }
-  io.exit(outcome.reason?.kind === 'completed' ? 0 : 1)
 }
 
 /**
  * Mount the one-shot direct driver.
  * @param ctx - plugin context carrying core services and the launcher-provided exit request.
- * @param config - validated task config.
+ * @param config - validated task and run options.
  */
 export function apply(ctx: Context, config: Config): void {
   // Read through the global service store, not the property proxy: appExit is
@@ -224,5 +307,5 @@ export function apply(ctx: Context, config: Config): void {
     throw new Error('headless-runner: the launcher must provide ctx.appExit before the tree mounts')
   }
   const io: HeadlessIo = { stdout: internals.stdout, stderr: internals.stderr, exit }
-  void run(ctx, config.task, io).catch((error: unknown) => { fail(io, error) })
+  void run(ctx, config, io).catch((error: unknown) => { fail(io, error, config.json === true) })
 }

packages/bundle/headless/src/json-stream.ts

@@ -0,0 +1,253 @@
+/**
+ * `--json` run projection: a bounded, ordered event stream derived from one
+ * Agent's durable Session events plus its live Assistant frames. The stream is
+ * a small vocabulary rather than a dump of the Session log, so a supervising
+ * process can consume it without filtering internal events or de-duplicating
+ * an assembled message against its own deltas.
+ * @module @deepseek-ai/dsh-headless/json-stream
+ */
+
+import type { Context } from '@deepseek-ai/cordis'
+import type { Agent, AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
+
+/** Default delay before a buffered streaming delta is flushed. */
+export const COALESCE_MS = 100
+
+/** Default buffered byte count that flushes streaming deltas immediately. */
+export const COALESCE_BYTES = 512
+
+/** Default per-string cap applied to every projected payload. */
+export const MAX_STRING_BYTES = 8 * 1024
+
+/** The stdout sink a projection writes newline-delimited events to. */
+export interface JsonSink {
+  /** Write one chunk of the event stream. */
+  write(chunk: string): unknown
+}
+
+/** Tunables for {@link projectJsonRun}; every field defaults. */
+export interface JsonProjectionOptions {
+  /** Working directory reported by the opening `session` event. */
+  cwd?: string
+  /** Coalescing delay for streaming deltas in milliseconds. */
+  coalesceMs?: number
+  /** Buffered delta bytes that flush immediately instead of waiting. */
+  coalesceBytes?: number
+  /** Per-string byte cap; longer strings are truncated and flagged. */
+  maxStringBytes?: number
+}
+
+/** The live handle of one `--json` projection. */
+export interface JsonProjection {
+  /** Flush buffered deltas and write the terminal `final` event. */
+  finish(text: string): void
+  /** Stop observing; buffered deltas that were never flushed are discarded. */
+  dispose(): void
+}
+
+/** Mutable truncation state threaded through one payload bound. */
+interface BoundState {
+  truncated: boolean
+}
+
+/** Truncate one UTF-8 string to `maxBytes`, dropping a split trailing character. */
+function truncateUtf8(text: string, maxBytes: number): string {
+  const buffer = Buffer.from(text, 'utf8').subarray(0, maxBytes)
+  const decoded = buffer.toString('utf8')
+  return decoded.endsWith('\uFFFD') ? decoded.slice(0, -1) : decoded
+}
+
+/** Recursively cap every string in one JSON-serializable value. */
+function boundValue(value: unknown, maxBytes: number, state: BoundState): unknown {
+  if (typeof value === 'string') {
+    if (Buffer.byteLength(value, 'utf8') <= maxBytes) return value
+    state.truncated = true
+    return truncateUtf8(value, maxBytes)
+  }
+  if (Array.isArray(value)) return value.map(item => boundValue(item, maxBytes, state))
+  if (value !== null && typeof value === 'object') {
+    const bounded: Record<string, unknown> = {}
+    for (const [key, item] of Object.entries(value)) bounded[key] = boundValue(item, maxBytes, state)
+    return bounded
+  }
+  return value
+}
+
+/** Bound one event payload and flag the event when any string was cut. */
+function boundEvent(event: Record<string, unknown>, maxBytes: number): Record<string, unknown> {
+  const state: BoundState = { truncated: false }
+  const bounded = boundValue(event, maxBytes, state) as Record<string, unknown>
+  if (state.truncated) bounded.truncated = true
+  return bounded
+}
+
+/** Parse raw tool-call arguments, keeping the unparsed string when it is not JSON. */
+function parseArguments(raw: string): unknown {
+  try {
+    return JSON.parse(raw) as unknown
+  } catch {
+    return raw
+  }
+}
+
+/** Join the text blocks of a tool result's model-facing content. */
+function resultText(blocks: readonly { type: string; text?: string }[]): string {
+  return blocks
+    .filter(block => block.type === 'text' && typeof block.text === 'string')
+    .map(block => block.text ?? '')
+    .join('')
+}
+
+/**
+ * Project one Agent's run as newline-delimited JSON on `sink`.
+ *
+ * The opening `session` event is written before the subscription starts, so a
+ * caller must invoke this before submitting the task. Deltas are coalesced but
+ * always flushed before any later event, which preserves stream order.
+ * @param ctx - plugin context carrying the live Session and Assistant feeds.
+ * @param agent - the exact Agent whose events belong to this invocation.
+ * @param sink - stdout sink receiving one JSON object per line.
+ * @param options - projection tunables.
+ * @returns the projection handle that finishes or disposes the stream.
+ */
+export function projectJsonRun(
+  ctx: Context,
+  agent: Agent,
+  sink: JsonSink,
+  options: JsonProjectionOptions = {},
+): JsonProjection {
+  const coalesceMs = options.coalesceMs ?? COALESCE_MS
+  const coalesceBytes = options.coalesceBytes ?? COALESCE_BYTES
+  const maxStringBytes = options.maxStringBytes ?? MAX_STRING_BYTES
+  let disposed = false
+  let timer: ReturnType<typeof setTimeout> | undefined
+  let bufferedBytes = 0
+  const deltas: { kind: 'text' | 'thinking'; text: string }[] = []
+  let stepUsage: SessionEvent<'assistant/message'>['data']['usage']
+
+  const write = (event: Record<string, unknown>): void => {
+    if (disposed) return
+    sink.write(`${JSON.stringify(boundEvent(event, maxStringBytes))}\n`)
+  }
+
+  const flush = (): void => {
+    if (timer !== undefined) {
+      clearTimeout(timer)
+      timer = undefined
+    }
+    bufferedBytes = 0
+    for (const delta of deltas.splice(0)) write({ type: delta.kind, text: delta.text })
+  }
+
+  const emit = (event: Record<string, unknown>): void => {
+    flush()
+    write(event)
+  }
+
+  const pushDelta = (kind: 'text' | 'thinking', text: string): void => {
+    const last = deltas[deltas.length - 1]
+    if (last !== undefined && last.kind === kind) last.text += text
+    else deltas.push({ kind, text })
+    bufferedBytes += Buffer.byteLength(text, 'utf8')
+    if (bufferedBytes >= coalesceBytes) {
+      flush()
+      return
+    }
+    timer ??= setTimeout(flush, coalesceMs)
+  }
+
+  const onSessionEvent = (session: unknown, event: SessionEvent): void => {
+    if (session !== agent.session) return
+    switch (event.type) {
+      case 'turn/start':
+        emit({ type: 'status', phase: 'turn_start', turn: event.data.turn })
+        return
+      case 'step/start':
+        emit({ type: 'status', phase: 'step_start', turn: event.data.turn, step: event.data.step })
+        return
+      case 'assistant/message':
+        stepUsage = event.data.usage
+        return
+      case 'step/end': {
+        const usage = stepUsage
+        stepUsage = undefined
+        emit({
+          type: 'status', phase: 'step_end', turn: event.data.turn, step: event.data.step,
+          ...usage === undefined ? {} : { usage },
+        })
+        return
+      }
+      case 'turn/end':
+        emit({ type: 'status', phase: 'turn_end', turn: event.data.turn, reason: event.data.reason })
+        return
+      case 'tool/call':
+        emit({
+          type: 'tool_call',
+          callId: event.data.callId,
+          tool: event.data.name,
+          input: parseArguments(event.data.arguments),
+        })
+        return
+      case 'tool/result': {
+        const block = event.data.message.content[0]
+        emit({
+          type: 'tool_result',
+          callId: block.toolCallId,
+          status: block.isError === true ? 'error' : 'completed',
+          result: resultText(block.content),
+        })
+        return
+      }
+      default:
+        return
+    }
+  }
+
+  const onFrame = (payload: { agent: Agent; frame: AssistantStreamFrame }): void => {
+    if (payload.agent !== agent) return
+    const frame = payload.frame
+    if (frame.type !== 'chunk') {
+      flush()
+      return
+    }
+    const chunk = frame.chunk
+    switch (chunk.type) {
+      case 'text-delta':
+        if (chunk.text !== '') pushDelta('text', chunk.text)
+        return
+      case 'reasoning-delta':
+        if (chunk.text !== '') pushDelta('thinking', chunk.text)
+        return
+      case 'block-start':
+      case 'block-end':
+      case 'tool-call-delta':
+      case 'usage':
+      case 'finish':
+        flush()
+        return
+      /* v8 ignore next -- closed-union exhaustiveness guard */
+      default:
+        return
+    }
+  }
+
+  write({ type: 'session', sessionId: agent.id, cwd: options.cwd ?? process.cwd() })
+  const stopSession = ctx.on('session/event', onSessionEvent)
+  const stopStream = ctx.on('agent/assistant-stream', onFrame)
+
+  return {
+    finish(text: string): void {
+      flush()
+      write({ type: 'final', text })
+    },
+    dispose(): void {
+      disposed = true
+      if (timer !== undefined) clearTimeout(timer)
+      timer = undefined
+      deltas.length = 0
+      stopSession()
+      stopStream()
+    },
+  }
+}

packages/bundle/headless/src/startup.ts

@@ -1,7 +1,8 @@
 /**
- * The one-shot app's command-line provider: it parses the task positional and
- * `--help`, then publishes {@link HEADLESS_STARTUP_SERVICE}. The runner is an
- * ordinary consumer whose lazy config waits for that service.
+ * The one-shot app's command-line provider: it parses the task positional,
+ * `--session-id`, `--json`, and `--help`, then publishes
+ * {@link HEADLESS_STARTUP_SERVICE}. The runner is an ordinary consumer whose
+ * lazy config waits for that service.
  * @module @deepseek-ai/dsh-headless/startup
  */
 
@@ -20,12 +21,21 @@ export const HEADLESS_STARTUP_SERVICE = 'headlessStartup'
 
 /** What the runner row reads from {@link HEADLESS_STARTUP_SERVICE}. */
 export interface HeadlessStartupValues {
-  /** The task text this invocation asked for. */
-  task: string
+  /** The task text this invocation asked for; absent when the runner reads stdin. */
+  task: string | undefined
+  /** Exact Session identity to adopt or create; absent for a fresh random identity. */
+  sessionId: string | undefined
+  /** Whether stdout carries the machine-readable event stream instead of final text. */
+  json: boolean
+}
+
+/** Process facts the provider reads; tests substitute them. */
+export const internals: { stdinIsTty: () => boolean } = {
+  stdinIsTty: () => process.stdin.isTTY,
 }
 
 /**
- * This app's command: the task positional, its description, and its help text.
+ * This app's command: the task positional, its options, and its help text.
  * @returns a fresh program, so one process can parse more than once (tests).
  */
 function headlessCommand(): Command {
@@ -33,25 +43,42 @@ function headlessCommand(): Command {
     .name('dsh --profile headless')
     .description('Answer one task, stream reasoning to stderr, print the final assistant message, and exit.')
     .helpOption('-h, --help', 'show this help')
-    .argument('[task...]', 'the task text; multiple words are joined by spaces')
+    .option('--json', 'write newline-delimited run events to stdout instead of the final message')
+    .option('--session-id <id>', 'adopt the persisted Session with this id, or create it when absent')
+    .argument('[task...]', 'the task text; multiple words are joined by spaces, and `-` reads stdin')
     .addHelpText('after', `
 Examples:
-  dsh --profile headless "run the tests"     answer one task and exit
+  dsh --profile headless "run the tests"          answer one task and exit
+  echo "run the tests" | dsh --profile headless   read the task from stdin
+  dsh --profile headless --json "run the tests"   emit machine-readable run events
+  dsh --profile headless --session-id session-… "continue"   adopt a Session
 `)
 }
 
 /**
  * Parse and provide the one-shot task as an ordinary Cordis service. The
- * command's action publishes the task; a missing or whitespace-only task is a
- * usage error, so on rejection (and on `--help`) nothing is provided.
+ * command's action publishes the task; a missing task on an interactive stdin
+ * is a usage error, so on rejection (and on `--help`) nothing is provided.
  * @param ctx - plugin context carrying the command line.
  */
 export function apply(ctx: Context): void {
   const program = headlessCommand()
   program.action(() => {
-    const task = program.args.join(' ')
-    if (task.trim() === '') program.error('error: a task is required, for example: dsh --profile headless "run the tests"')
-    ctx.provide(HEADLESS_STARTUP_SERVICE, { task } satisfies HeadlessStartupValues)
+    const joined = program.args.join(' ')
+    const task = joined.trim() === '' ? undefined : joined
+    if (task === undefined && internals.stdinIsTty()) {
+      program.error('error: a task is required, for example: dsh --profile headless "run the tests"')
+    }
+    const options = program.opts<{ json?: boolean; sessionId?: string }>()
+    const sessionId = options.sessionId?.trim()
+    if (options.sessionId !== undefined && sessionId === '') {
+      program.error('error: --session-id requires a non-empty session id')
+    }
+    ctx.provide(HEADLESS_STARTUP_SERVICE, {
+      task,
+      sessionId,
+      json: options.json === true,
+    } satisfies HeadlessStartupValues)
   })
   parseCmdline(ctx, program)
 }

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

@@ -1,14 +1,22 @@
-/** Direct one-shot Agent driving, durable aggregation, flushing, and exit mapping. */
+/** Direct one-shot Agent driving, exact Session adoption, machine-readable projection, and exit mapping. */
 
 import { afterEach, describe, expect, it } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
+import { brandString } from '@deepseek-ai/dsh-brand'
 import AgentRegistry from '@deepseek-ai/dsh-agent'
-import type { Agent, AgentHandle, AssistantStreamFrame, CreateAgentOptions } from '@deepseek-ai/dsh-agent'
+import type {
+  Agent,
+  AgentHandle,
+  AssistantStreamFrame,
+  CreateAgentOptions,
+  ResumeAgentOptions,
+} from '@deepseek-ai/dsh-agent'
 import AgentDefaultModelConfig from '@deepseek-ai/dsh-agent-default-model'
-import { LlmAttemptId, createAssistantMessage, type StreamChunk } from '@deepseek-ai/dsh-llm'
+import { LlmAttemptId, ToolCallId, createAssistantMessage, createToolResultMessage, type StreamChunk } from '@deepseek-ai/dsh-llm'
 import SessionStore from '@deepseek-ai/dsh-session'
 import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
-import type { Session, UserMessage } from '@deepseek-ai/dsh-session'
+import type { Session, SessionId, UserMessage } from '@deepseek-ai/dsh-session'
+import { SessionQueryError } from '@deepseek-ai/dsh-session-query'
 import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit'
 import { apply, Config, internals } from '../src/index.ts'
 
@@ -20,6 +28,22 @@ interface Script {
   afterPrompt(session: Session, message: UserMessage, agent: Agent): Promise<void> | void
 }
 
+/** Observation stub returned by the `--session-id` query path. */
+interface ObservationStub {
+  header: { cwd: string; origin?: string; parentSession?: string }
+  [Symbol.dispose](): void
+}
+
+/** Runner invocation options layered over the scripted Agent factory. */
+interface BenchOptions {
+  task?: string
+  useStdin?: boolean
+  readStdin?: () => Promise<string>
+  sessionId?: string
+  json?: boolean
+  observe?: () => Promise<ObservationStub>
+}
+
 const frameStates = new WeakMap<Agent, { attemptId: ReturnType<typeof LlmAttemptId>; revision: number; index: number }>()
 
 function startFrames(agent: Agent, turn = 1, step = 1): void {
@@ -74,7 +98,7 @@ function appendTurn(
 }
 
 /** Mount the real registries around a small scripted Agent factory. */
-async function bench(script: Script): Promise<{
+async function bench(script: Script, options: BenchOptions = {}): Promise<{
   ctx: Context
   output(): { out: string; err: string; order: string[] }
   run(): Promise<{ code: number; out: string; err: string; order: string[] }>
@@ -83,42 +107,60 @@ async function bench(script: Script): Promise<{
   let out = ''
   let err = ''
   const order: string[] = []
+
+  const mount = async (
+    ownerCtx: Context,
+    session: Session,
+    createOptions: CreateAgentOptions | ResumeAgentOptions,
+  ): Promise<Agent> => {
+    const inbox = createInboxStub()
+    let idle = Promise.resolve()
+    const agent: Agent = {
+      id: session.id,
+      options: createOptions.agentOptions ?? {},
+      session,
+      inbox,
+      status: 'idle',
+      ctx: ownerCtx,
+      cancel: () => {},
+      runMaintenance: () => Promise.reject(new Error('not used')),
+      send: () => {},
+      followup: (message: UserMessage) => {
+        agent.inbox.append('next-turn', message)
+        idle = Promise.resolve().then(() => script.afterPrompt(session, message, agent))
+      },
+      steer: () => {},
+      inject: () => {},
+      whenIdle: () => idle,
+    }
+    await createOptions.setup?.(ownerCtx, agent)
+    ctx.agents.register(agent)
+    return agent
+  }
+
   await ctx.plugin(SessionStore)
   await ctx.plugin(SessionProjectionRegistry)
   await ctx.plugin(AgentRegistry)
   await ctx.plugin(AgentDefaultModelConfig, { provider: 'test-provider', model: 'test-model' })
   ctx.agents.setFactory({
-    async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
-      const session = ctx.sessions.create(options.sessionId, {
-        ...options.meta === undefined ? {} : { meta: options.meta },
+    async createAgent(ownerCtx: Context, createOptions: CreateAgentOptions): Promise<AgentHandle> {
+      const session = ctx.sessions.create(createOptions.sessionId, {
+        ...createOptions.meta === undefined ? {} : { meta: createOptions.meta },
       })
-      const inbox = createInboxStub()
-      let idle = Promise.resolve()
-      const agent: Agent = {
-        id: session.id,
-        options: options.agentOptions ?? {},
-        session,
-        inbox,
-        status: 'idle',
-        ctx: ownerCtx,
-        cancel: () => {},
-        runMaintenance: () => Promise.reject(new Error('not used')),
-        send: () => {},
-        followup: (message: UserMessage) => {
-          agent.inbox.append('next-turn', message)
-          idle = Promise.resolve().then(() => script.afterPrompt(session, message, agent))
-        },
-        steer: () => {},
-        inject: () => {},
-        whenIdle: () => idle,
-      }
-      await options.setup?.(ownerCtx, agent)
       script.before?.(session)
-      ctx.agents.register(agent)
+      const agent = await mount(ownerCtx, session, createOptions)
+      return { agent, dispose: () => Promise.resolve() }
+    },
+    async resume(ownerCtx: Context, resumeOptions: ResumeAgentOptions): Promise<AgentHandle> {
+      const session = ctx.sessions.get(resumeOptions.resumeSessionId)
+      if (session === undefined) throw new Error(`no attached Session ${resumeOptions.resumeSessionId}`)
+      const agent = await mount(ownerCtx, session, resumeOptions)
       return { agent, dispose: () => Promise.resolve() }
     },
-    resume: () => Promise.reject(new Error('not used')),
   })
+  if (options.observe !== undefined) {
+    ctx.provide('sessionQuery', { observeSession: () => options.observe!() } as never)
+  }
   return {
     ctx,
     output: () => ({ out, err, order: [...order] }),
@@ -126,10 +168,15 @@ async function bench(script: Script): Promise<{
       ctx.on('session/flush', () => { order.push('flush') })
       internals.stdout = { write: (chunk: string) => { out += chunk; return true } }
       internals.stderr = { write: (chunk: string) => { err += chunk; return true } }
+      if (options.readStdin !== undefined) internals.readStdin = options.readStdin
       const exited = new Promise<number>((resolve) => {
         ctx.provide('appExit', (code: number) => { order.push('exit'); resolve(code) })
       })
-      apply(ctx, { task: 'do the thing' })
+      apply(ctx, {
+        ...options.useStdin === true ? {} : { task: options.task ?? 'do the thing' },
+        ...options.sessionId === undefined ? {} : { sessionId: options.sessionId },
+        ...options.json === undefined ? {} : { json: options.json },
+      })
       return { code: await exited, out, err, order }
     },
   }
@@ -375,6 +422,174 @@ describe('headless runner', () => {
     await test.ctx.fiber.dispose()
   })
 
+  it('reads the task from stdin when the invocation omits one', async () => {
+    const test = await bench({
+      afterPrompt(session, message) { appendTurn(session, 1, message, 'stdin answer', true) },
+    }, {
+      useStdin: true,
+      readStdin: () => Promise.resolve('task from stdin'),
+    })
+    expect(await test.run()).toMatchObject({ code: 0, out: 'stdin answer\n', err: '' })
+    await test.ctx.fiber.dispose()
+  })
+
+  it('rejects an empty stdin task', async () => {
+    const test = await bench({ afterPrompt: () => {} }, {
+      useStdin: true,
+      readStdin: () => Promise.resolve('   \n'),
+    })
+    expect(await test.run()).toMatchObject({
+      code: 1,
+      err: 'dsh: a task is required, for example: dsh --profile headless "run the tests"\n',
+    })
+    await test.ctx.fiber.dispose()
+  })
+
+  it('creates the exact requested Session when the query reports it missing', async () => {
+    const seen: string[] = []
+    const test = await bench({
+      afterPrompt(session, message) {
+        seen.push(session.id)
+        appendTurn(session, 1, message, 'created', true)
+      },
+    }, {
+      sessionId: 'session-exact',
+      observe: () => Promise.reject(new SessionQueryError('missing', 'SESSION_QUERY_SESSION_NOT_FOUND')),
+    })
+    expect(await test.run()).toMatchObject({ code: 0, out: 'created\n', err: '' })
+    expect(seen).toEqual(['session-exact'])
+    await test.ctx.fiber.dispose()
+  })
+
+  it('resumes the persisted Session when the query finds it', async () => {
+    const test = await bench({
+      afterPrompt(session, message) { appendTurn(session, 1, message, 'resumed answer', true) },
+    }, {
+      sessionId: 'session-exact',
+      observe: () => Promise.resolve({
+        header: { cwd: process.cwd(), origin: 'user' },
+        [Symbol.dispose]() {},
+      }),
+    })
+    const session = test.ctx.sessions.create(brandString<SessionId>('session-exact'), { meta: { cwd: process.cwd() } })
+    const history = {
+      role: 'user', content: [{ type: 'text', text: 'earlier' }], source: { kind: 'user' }, id: 'history',
+    } as UserMessage
+    appendTurn(session, 0, history, 'earlier answer', true)
+    const before = session.seq
+    expect(await test.run()).toMatchObject({ code: 0, out: 'resumed answer\n', err: '' })
+    expect(session.seq).toBeGreaterThan(before)
+    await test.ctx.fiber.dispose()
+  })
+
+  it('rejects a persisted Session recorded in another working directory', async () => {
+    const test = await bench({ afterPrompt: () => {} }, {
+      sessionId: 'session-exact',
+      observe: () => Promise.resolve({
+        header: { cwd: '/somewhere/else', origin: 'user' },
+        [Symbol.dispose]() {},
+      }),
+    })
+    const result = await test.run()
+    expect(result.code).toBe(1)
+    expect(result.err).toContain('was recorded in "/somewhere/else"')
+    await test.ctx.fiber.dispose()
+  })
+
+  it('rejects a persisted Session owned by a subagent', async () => {
+    const test = await bench({ afterPrompt: () => {} }, {
+      sessionId: 'session-exact',
+      observe: () => Promise.resolve({
+        header: { cwd: process.cwd(), origin: 'subagent' },
+        [Symbol.dispose]() {},
+      }),
+    })
+    const result = await test.run()
+    expect(result.code).toBe(1)
+    expect(result.err).toContain('belongs to a subagent')
+    await test.ctx.fiber.dispose()
+  })
+
+  it('requires the Session query service for an exact Session identity', async () => {
+    const test = await bench({ afterPrompt: () => {} }, { sessionId: 'session-exact' })
+    const result = await test.run()
+    expect(result.code).toBe(1)
+    expect(result.err).toContain('requires the sessionQuery service')
+    await test.ctx.fiber.dispose()
+  })
+
+  it('projects the run as ordered newline-delimited events in --json mode', async () => {
+    const test = await bench({
+      afterPrompt(session, message, agent) {
+        session.append('turn/start', { turn: 1 })
+        session.append('step/start', { turn: 1, step: 1 })
+        session.append('user/message', message, { surfaceOp: 'append' })
+        startFrames(agent)
+        emitChunk(agent, { type: 'reasoning-delta', index: 0, text: 'thinking hard' })
+        emitChunk(agent, { type: 'text-delta', index: 0, text: 'answer' })
+        session.append('assistant/message', {
+          stream: [],
+          turn: 1,
+          step: 1,
+          usage: { inputTokens: 3, outputTokens: 4 },
+          message: createAssistantMessage({
+            content: [
+              { type: 'text', text: 'answer' },
+              { type: 'tool-call', id: ToolCallId('call-1'), name: 'bash', arguments: '{"command":"ls"}' },
+            ],
+            source: { provider: 'test-provider', model: 'test-model' },
+          }),
+        }, { surfaceOp: 'append' })
+        session.append('tool/call', {
+          turn: 1, step: 1, callId: ToolCallId('call-1'), name: 'bash', arguments: '{"command":"ls"}',
+        })
+        session.append('tool/result', {
+          turn: 1,
+          step: 1,
+          message: createToolResultMessage({
+            callId: ToolCallId('call-1'),
+            content: [{ type: 'text', text: 'a.txt' }],
+            isError: false,
+          }),
+        }, { surfaceOp: 'append' })
+        session.append('step/end', { turn: 1, step: 1 })
+        session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+      },
+    }, { json: true })
+    const result = await test.run()
+    const events = result.out.trim().split('\n').map(line => JSON.parse(line) as Record<string, unknown>)
+    expect(events.map(event => event.type)).toEqual([
+      'session', 'status', 'status', 'thinking', 'text',
+      'tool_call', 'tool_result', 'status', 'status', 'final',
+    ])
+    expect(events[0]).toMatchObject({ type: 'session', cwd: process.cwd() })
+    expect(typeof events[0]?.sessionId).toBe('string')
+    expect(events[1]).toMatchObject({ type: 'status', phase: 'turn_start', turn: 1 })
+    expect(events[3]).toMatchObject({ type: 'thinking', text: 'thinking hard' })
+    expect(events[4]).toMatchObject({ type: 'text', text: 'answer' })
+    expect(events[5]).toMatchObject({ type: 'tool_call', callId: 'call-1', tool: 'bash', input: { command: 'ls' } })
+    expect(events[6]).toMatchObject({ type: 'tool_result', callId: 'call-1', status: 'completed', result: 'a.txt' })
+    expect(events[7]).toMatchObject({ type: 'status', phase: 'step_end', usage: { inputTokens: 3, outputTokens: 4 } })
+    expect(events[8]).toMatchObject({ type: 'status', phase: 'turn_end', reason: { kind: 'completed' } })
+    expect(events[9]).toMatchObject({ type: 'final', text: 'answer' })
+    expect(result.err).toBe('')
+    expect(result.code).toBe(0)
+    await test.ctx.fiber.dispose()
+  })
+
+  it('reports a direct failure as an error event in --json mode', async () => {
+    const test = await bench({ afterPrompt: () => {} }, {
+      useStdin: true,
+      readStdin: () => Promise.resolve(''),
+      json: true,
+    })
+    const result = await test.run()
+    expect(result.code).toBe(1)
+    expect(JSON.parse(result.out.trim())).toMatchObject({ type: 'error' })
+    expect(result.err).toContain('a task is required')
+    await test.ctx.fiber.dispose()
+  })
+
   it('reports a direct Agent creation failure', async () => {
     const ctx = new Context()
     let err = ''
@@ -442,8 +657,9 @@ describe('headless runner', () => {
     expect(() => { apply(ctx, { task: 't' }) }).toThrow('must provide ctx.appExit')
   })
 
-  it('validates config: the task is required', () => {
-    expect(() => new Config({} as never)).toThrow()
-    expect(new Config({ task: 'x' })).toEqual({ task: 'x' })
+  it('validates config: the task and run options are optional', () => {
+    expect(new Config({})).toEqual({})
+    expect(new Config({ task: 'x', sessionId: 'session-x', json: true }))
+      .toEqual({ task: 'x', sessionId: 'session-x', json: true })
   })
 })

packages/bundle/headless/tests/json-stream.spec.ts

@@ -0,0 +1,172 @@
+/** The `--json` run projection: ordering, coalescing, bounding, and disposal. */
+
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { Context } from '@deepseek-ai/cordis'
+import type { Agent, AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
+import { LlmAttemptId, type StreamChunk } from '@deepseek-ai/dsh-llm'
+import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import { projectJsonRun, type JsonProjectionOptions } from '../src/json-stream.ts'
+
+afterEach(() => { vi.useRealTimers() })
+
+interface ProjectionHarness {
+  readonly lines: string[]
+  readonly projection: ReturnType<typeof projectJsonRun>
+  readonly agent: Agent
+  readonly session: Session
+  readonly parsed: () => Record<string, unknown>[]
+  emitSession(event: SessionEvent): void
+  emitFrame(chunk: StreamChunk): void
+  emitRawSession(session: unknown, event: SessionEvent): void
+  emitRawFrame(payload: { agent: Agent; frame: AssistantStreamFrame }): void
+}
+
+/** Drive the projector through a minimal Context and Agent double. */
+function harness(options: JsonProjectionOptions = {}): ProjectionHarness {
+  const lines: string[] = []
+  const sessionListeners = new Set<(session: unknown, event: SessionEvent) => void>()
+  const frameListeners = new Set<(payload: { agent: Agent; frame: AssistantStreamFrame }) => void>()
+  const ctx = {
+    on(name: string, handler: unknown) {
+      const set = name === 'session/event' ? sessionListeners : frameListeners
+      set.add(handler as never)
+      return () => { set.delete(handler as never) }
+    },
+  } as unknown as Context
+  const session = {} as Session
+  const agent = { id: 'session-1', session } as unknown as Agent
+  const projection = projectJsonRun(ctx, agent, {
+    write: (chunk: string) => { lines.push(chunk); return true },
+  }, { cwd: '/', ...options })
+  const attemptId = LlmAttemptId('attempt')
+  let revision = 0
+  return {
+    lines,
+    projection,
+    agent,
+    session,
+    parsed: () => lines.map(line => JSON.parse(line) as Record<string, unknown>),
+    emitSession: (event) => { for (const listener of sessionListeners) listener(session, event) },
+    emitFrame: (chunk) => {
+      revision += 1
+      const frame: AssistantStreamFrame = {
+        type: 'chunk', attemptId, revision, index: 0, time: 0, chunk,
+      }
+      for (const listener of frameListeners) listener({ agent, frame })
+    },
+    emitRawSession: (rawSession, event) => { for (const listener of sessionListeners) listener(rawSession, event) },
+    emitRawFrame: (payload) => { for (const listener of frameListeners) listener(payload) },
+  }
+}
+
+describe('--json projection', () => {
+  it('opens with the session event before any observed event', () => {
+    const test = harness()
+    expect(test.parsed()).toEqual([{ type: 'session', sessionId: 'session-1', cwd: '/' }])
+  })
+
+  it('coalesces consecutive same-kind deltas and flushes them before a later event', () => {
+    const test = harness({ coalesceMs: 10_000 })
+    test.emitFrame({ type: 'text-delta', index: 0, text: 'an' })
+    test.emitFrame({ type: 'text-delta', index: 0, text: 'swer' })
+    test.emitFrame({ type: 'reasoning-delta', index: 0, text: 'think' })
+    test.emitFrame({ type: 'text-delta', index: 0, text: '!' })
+    test.emitSession({ type: 'tool/call', data: { turn: 1, step: 1, callId: 'c1', name: 'bash', arguments: '{"a":1}' } } as unknown as SessionEvent)
+    expect(test.parsed().map(event => event.type)).toEqual(['session', 'text', 'thinking', 'text', 'tool_call'])
+    expect(test.parsed()[1]).toEqual({ type: 'text', text: 'answer' })
+    expect(test.parsed()[3]).toEqual({ type: 'text', text: '!' })
+    expect(test.parsed()[4]).toMatchObject({ type: 'tool_call', callId: 'c1', tool: 'bash', input: { a: 1 } })
+  })
+
+  it('flushes a buffered delta when the byte cap is reached', () => {
+    const test = harness({ coalesceMs: 10_000, coalesceBytes: 4 })
+    test.emitFrame({ type: 'text-delta', index: 0, text: 'abc' })
+    expect(test.lines).toHaveLength(1)
+    test.emitFrame({ type: 'text-delta', index: 0, text: 'd' })
+    expect(test.parsed().map(event => event.type)).toEqual(['session', 'text'])
+    expect(test.parsed()[1]).toEqual({ type: 'text', text: 'abcd' })
+  })
+
+  it('flushes a buffered delta on the coalescing timer', () => {
+    vi.useFakeTimers()
+    const test = harness({ coalesceMs: 100 })
+    test.emitFrame({ type: 'text-delta', index: 0, text: 'later' })
+    expect(test.lines).toHaveLength(1)
+    vi.advanceTimersByTime(100)
+    expect(test.parsed().map(event => event.type)).toEqual(['session', 'text'])
+  })
+
+  it('bounds every long string and flags the event once', () => {
+    const test = harness({ coalesceMs: 10_000, maxStringBytes: 16 })
+    test.emitFrame({ type: 'text-delta', index: 0, text: 'abcdefghijklmnopqrst' })
+    test.emitSession({
+      type: 'tool/call',
+      data: { turn: 1, step: 1, callId: 'c1', name: 'bash', arguments: JSON.stringify({ command: 'x'.repeat(40) }) },
+    } as unknown as SessionEvent)
+    const events = test.parsed()
+    expect(events[1]).toEqual({ type: 'text', text: 'abcdefghijklmnop', truncated: true })
+    expect(events[2]).toMatchObject({ type: 'tool_call', truncated: true })
+    expect(events[2]?.input).toEqual({ command: 'x'.repeat(16) })
+  })
+
+  it('ignores events from other Sessions and Agents', () => {
+    const test = harness({ coalesceMs: 10_000 })
+    test.emitRawSession({}, { type: 'turn/start', data: { turn: 1 } } as unknown as SessionEvent)
+    test.emitRawFrame({
+      agent: { id: 'other', session: {} } as unknown as Agent,
+      frame: { type: 'chunk', attemptId: LlmAttemptId('other'), revision: 1, index: 0, time: 0, chunk: { type: 'text-delta', index: 0, text: 'foreign' } },
+    })
+    test.projection.finish('mine')
+    expect(test.parsed().map(event => event.type)).toEqual(['session', 'final'])
+  })
+
+  it('reports tool results as completed or errored and keeps the final event last', () => {
+    const test = harness({ coalesceMs: 10_000 })
+    test.emitSession({
+      type: 'tool/result',
+      data: {
+        turn: 1,
+        step: 1,
+        message: {
+          content: [{ type: 'tool-result', toolCallId: 'c1', content: [{ type: 'text', text: 'a.txt' }], isError: false }],
+        },
+      },
+    } as unknown as SessionEvent)
+    test.emitSession({
+      type: 'tool/result',
+      data: {
+        turn: 1,
+        step: 1,
+        message: {
+          content: [{ type: 'tool-result', toolCallId: 'c2', content: [{ type: 'text', text: 'boom' }], isError: true }],
+        },
+      },
+    } as unknown as SessionEvent)
+    test.projection.finish('done')
+    const events = test.parsed()
+    expect(events[1]).toEqual({ type: 'tool_result', callId: 'c1', status: 'completed', result: 'a.txt' })
+    expect(events[2]).toEqual({ type: 'tool_result', callId: 'c2', status: 'error', result: 'boom' })
+    expect(events.at(-1)).toEqual({ type: 'final', text: 'done' })
+  })
+
+  it('attaches step usage to the step_end status and drops buffered deltas on dispose', () => {
+    const test = harness({ coalesceMs: 10_000 })
+    test.emitSession({
+      type: 'assistant/message',
+      data: {
+        stream: [],
+        turn: 1,
+        step: 1,
+        usage: { inputTokens: 3, outputTokens: 4 },
+        message: { role: 'assistant', content: [], source: { kind: 'model', provider: 'p', model: 'm' } },
+      },
+    } as unknown as SessionEvent)
+    test.emitSession({ type: 'step/end', data: { turn: 1, step: 1 } } as unknown as SessionEvent)
+    expect(test.parsed()[1]).toEqual({ type: 'status', phase: 'step_end', turn: 1, step: 1, usage: { inputTokens: 3, outputTokens: 4 } })
+
+    test.emitFrame({ type: 'text-delta', index: 0, text: 'dropped' })
+    test.projection.dispose()
+    test.projection.finish('ignored')
+    expect(test.parsed().map(event => event.type)).toEqual(['session', 'status'])
+  })
+})

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

@@ -1,7 +1,7 @@
 /**
  * The one-shot app's ordinary command-line provider over a real Loader tree:
- * the task becomes injected runner config, while help and usage errors leave
- * the consumer pending.
+ * the task, exact Session identity, and output mode become injected runner
+ * config, while help and interactive usage errors leave the consumer pending.
  */
 
 import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
@@ -11,9 +11,14 @@ import { pathToFileURL } from 'node:url'
 import { Context } from '@deepseek-ai/cordis'
 import Loader from '@deepseek-ai/cordis-plugin-loader'
 import Include from '@deepseek-ai/cordis-plugin-include'
-import { internals, provideCmdline } from '@deepseek-ai/dsh-cmdline'
+import { internals as cmdlineInternals, provideCmdline } from '@deepseek-ai/dsh-cmdline'
 import { afterEach, describe, expect, it } from 'vitest'
-import { apply, HEADLESS_STARTUP_SERVICE, type HeadlessStartupValues } from '../src/startup.ts'
+import {
+  apply,
+  HEADLESS_STARTUP_SERVICE,
+  internals as startupInternals,
+  type HeadlessStartupValues,
+} from '../src/startup.ts'
 
 /** What one boot of the fixture tree observed. */
 interface Observed {
@@ -30,16 +35,21 @@ const tempDirs: string[] = []
 afterEach(async () => {
   for (const dispose of disposers.splice(0)) await dispose()
   for (const dir of tempDirs.splice(0)) rmSync(dir, { recursive: true, force: true })
-  internals.stdout = process.stdout
-  internals.stderr = process.stderr
+  cmdlineInternals.stdout = process.stdout
+  cmdlineInternals.stderr = process.stderr
+  startupInternals.stdinIsTty = () => process.stdin.isTTY
 })
 
 /**
  * Mount the real provider over a runner stand-in.
  * @param args - the invocation's inner arguments.
+ * @param options - process facts the provider reads.
  * @returns the resolved service value and observed runner/process effects.
  */
-async function bootStartup(args: string[]): Promise<{ task: HeadlessStartupValues | undefined; observed: Observed }> {
+async function bootStartup(
+  args: string[],
+  options: { stdinIsTty?: boolean } = {},
+): Promise<{ task: HeadlessStartupValues | undefined; observed: Observed }> {
   const dir = mkdtempSync(join(tmpdir(), 'dsh-headless-startup-'))
   tempDirs.push(dir)
   const observed: Observed = { exits: [], out: '' }
@@ -58,13 +68,16 @@ export const apply = ctx => globalThis.__headlessStartupApply(ctx)
     `  inject: [${HEADLESS_STARTUP_SERVICE}]`,
     '  config:',
     '    task: !!js ctx.headlessStartup.task',
+    '    sessionId: !!js ctx.headlessStartup.sessionId',
+    '    json: !!js ctx.headlessStartup.json',
     '- id: headless-startup',
     `  name: ${pathToFileURL(join(dir, 'startup.mjs')).href}`,
     '',
   ].join('\n'))
   const observing = { write: (chunk: string) => { observed.out += chunk; return true } }
-  internals.stdout = observing
-  internals.stderr = observing
+  cmdlineInternals.stdout = observing
+  cmdlineInternals.stderr = observing
+  startupInternals.stdinIsTty = () => options.stdinIsTty === true
   const globals = globalThis as unknown as {
     __headlessStartupApply: typeof apply
     __headlessStartupObserved: Observed
@@ -88,23 +101,48 @@ export const apply = ctx => globalThis.__headlessStartupApply(ctx)
 describe('headless command-line provider', () => {
   it('joins the task positional into the runner config', async () => {
     const { task, observed } = await bootStartup(['run', 'the', 'tests'])
-    expect(task).toEqual({ task: 'run the tests' })
-    expect(observed.runnerConfig).toEqual({ task: 'run the tests' })
+    expect(task).toEqual({ task: 'run the tests', sessionId: undefined, json: false })
+    expect(observed.runnerConfig).toMatchObject({ task: 'run the tests', json: false })
     expect(observed.exits).toEqual([])
   })
 
-  it.each([{ args: [] }, { args: ['   '] }])('rejects an invocation with no non-whitespace task ($args)', async ({ args }) => {
-    const { task, observed } = await bootStartup(args)
+  it('publishes the machine-readable output mode and the exact Session identity', async () => {
+    const { task, observed } = await bootStartup(['--json', '--session-id', 'session-exact', 'do', 'it'])
+    expect(task).toEqual({ task: 'do it', sessionId: 'session-exact', json: true })
+    expect(observed.runnerConfig).toMatchObject({ task: 'do it', sessionId: 'session-exact', json: true })
+  })
+
+  it('keeps the stdin marker as the task so the runner reads the pipe', async () => {
+    const { task } = await bootStartup(['-'], { stdinIsTty: false })
+    expect(task).toEqual({ task: '-', sessionId: undefined, json: false })
+  })
+
+  it('defers an absent task to stdin when stdin is not a terminal', async () => {
+    const { task, observed } = await bootStartup([], { stdinIsTty: false })
+    expect(task).toEqual({ task: undefined, sessionId: undefined, json: false })
+    expect(observed.runnerConfig).toMatchObject({ json: false })
+  })
+
+  it.each([{ args: [] as string[] }, { args: ['   '] }])('rejects an interactive invocation with no task ($args)', async ({ args }) => {
+    const { task, observed } = await bootStartup(args, { stdinIsTty: true })
     expect(observed.out).toContain('a task is required')
     expect(task).toBeUndefined()
     expect(observed.runnerConfig).toBeUndefined()
     expect(observed.exits).toEqual([1])
   })
 
+  it('rejects an explicitly empty Session identity', async () => {
+    const { task, observed } = await bootStartup(['--session-id', '', 'do', 'it'])
+    expect(observed.out).toContain('--session-id requires a non-empty session id')
+    expect(task).toBeUndefined()
+    expect(observed.exits).toEqual([1])
+  })
+
   it('prints its own help and leaves the runner pending', async () => {
     const { task, observed } = await bootStartup(['--help'])
     expect(observed.out).toContain('dsh --profile headless')
     expect(observed.out).toContain('stream reasoning to stderr')
+    expect(observed.out).toContain('--session-id')
     expect(task).toBeUndefined()
     expect(observed.runnerConfig).toBeUndefined()
     expect(observed.exits).toEqual([0])

packages/bundle/headless/tsconfig.json

@@ -31,6 +31,9 @@
     },
     {
       "path": "../../boot/cmdline"
+    },
+    {
+      "path": "../../session-query/session-query"
     }
   ]
 }

pnpm-lock.yaml

@@ -1631,6 +1631,9 @@ importers:
       '@deepseek-ai/dsh-session':
         specifier: workspace:^
         version: link:../../core/session
+      '@deepseek-ai/dsh-session-query':
+        specifier: workspace:^
+        version: link:../../session-query/session-query
 
   packages/bundle/sdk-app:
     dependencies: