From c1f7e72aaeeb96aa547a2f3640544f91a6706565 Mon Sep 17 00:00:00 2001 From: Luke Parke <5702154+LukasParke@users.noreply.github.com> Date: Wed, 12 Aug 2026 21:51:25 -0500 Subject: [PATCH] fix(sub-harness): harden session turns Port fork/port/openrouter-fixes commit 0fd1b9ef onto current main, including failed-turn handling, idle cancellation, reasoning retention, and deduplicated session startup. Signed-off-by: Luke Parke <5702154+LukasParke@users.noreply.github.com> --- .../references/api-reference.md | 2 +- packages/core/src/harness/agent-harness.ts | 2 +- .../src/interpreter/execute-sub-harness.ts | 158 +++++++- .../interpreter/execute-sub-harness.test.ts | 356 ++++++++++++++++++ packages/sub-harness-pi/test/pi.test.ts | 6 +- packages/sub-harness/README.md | 7 +- packages/sub-harness/src/define.ts | 3 + packages/sub-harness/src/index.ts | 2 +- packages/sub-harness/src/items.ts | 17 +- packages/sub-harness/src/turn.ts | 7 +- packages/sub-harness/test/sub-harness.test.ts | 25 ++ .../content/docs/framework/sub-harnesses.mdx | 10 +- specs/27-sub-harness-steps.md | 20 +- 13 files changed, 583 insertions(+), 32 deletions(-) diff --git a/.claude/skills/noetic-agent-builder/references/api-reference.md b/.claude/skills/noetic-agent-builder/references/api-reference.md index f1621097..c9aa5b1d 100644 --- a/.claude/skills/noetic-agent-builder/references/api-reference.md +++ b/.claude/skills/noetic-agent-builder/references/api-reference.md @@ -266,7 +266,7 @@ export const myAgent = (settings = {}) => }); ``` -Stream-part kinds: `stream-start`, `text-delta`, `reasoning-delta`, `tool-call`, `tool-result`, `file-change`, `finish` (carries `usage`/`cost`), `error`, `raw`. The union has a paired Zod schema `SubHarnessStreamPartSchema`. A `SubHarnessSession` requires only `doPromptTurn` + `doStop`; `doContinueTurn` / `doSuspendTurn` / `doDetach` / `doDestroy` / `doCompact` are optional and signalled by presence (absent → throw `SubHarnessCapabilityError`). Base package also exports `SubHarnessTurnAccumulator`, the `asItems` / `assistantMessageItem` / `functionCallItem` item builders, and `SubHarnessStartError`. +Stream-part kinds: `stream-start`, `text-delta`, `reasoning-delta`, `tool-call`, `tool-result`, `file-change`, `finish` (carries `usage`/`cost`), `error`, `raw`. The union has a paired Zod schema `SubHarnessStreamPartSchema`. A `SubHarnessSession` requires only `doPromptTurn` + `doStop`; `doContinueTurn` / `doSuspendTurn` / `doDetach` / `doDestroy` / `doCompact` are optional and signalled by presence (absent → throw `SubHarnessCapabilityError`). Base package also exports `SubHarnessTurnAccumulator`, the `asItems` / `assistantMessageItem` / `functionCallItem` / `reasoningItem` item builders, and `SubHarnessStartError`. A turn with `finishReason: 'error'` fails the step (`step_failed`) after its items/usage are applied; each turn has an idle watchdog (`settings.extra.idleTimeoutMs`, default 120s, 0 disables) that aborts a stalled turn; `reasoning-delta` text is retained as a `reasoning` item; concurrent first turns on one `session.reuse` key dedupe onto a single `doStart` (the store holds promises). ### channel diff --git a/packages/core/src/harness/agent-harness.ts b/packages/core/src/harness/agent-harness.ts index 2f90ad29..71a4e5d1 100644 --- a/packages/core/src/harness/agent-harness.ts +++ b/packages/core/src/harness/agent-harness.ts @@ -442,7 +442,7 @@ export class AgentHarness = Record(); + readonly subHarnessSessions = new Map>(); readonly layerStateStore: LayerStateStore; /** Per-harness memoization cache for `recallMode: 'eventual'` layers. */ readonly recallCache: RecallCache; diff --git a/packages/core/src/interpreter/execute-sub-harness.ts b/packages/core/src/interpreter/execute-sub-harness.ts index ddaa3881..929747d6 100644 --- a/packages/core/src/interpreter/execute-sub-harness.ts +++ b/packages/core/src/interpreter/execute-sub-harness.ts @@ -41,7 +41,13 @@ import { isContextImpl, isFunctionCall, isMutableContext } from './typeguards'; * mirroring how the interpreter reaches `layerStateStore`. */ interface SubHarnessSessionStore { - subHarnessSessions: Map; + /** + * Keyed by `step.session.reuse`. Values are PROMISES so two steps racing on + * the same key (e.g. parallel legs) dedupe onto one `doStart` instead of + * both starting and one leaking. Driving concurrent TURNS on one reused + * session remains unsupported — sessions are conversational state. + */ + subHarnessSessions: Map>; } type TeardownMode = NonNullable; @@ -50,7 +56,7 @@ type TeardownMode = NonNullable; //#region Helpers -function sessionStore(ctx: Context): Map { +function sessionStore(ctx: Context): Map> { return frameworkCast(ctx.harness).subHarnessSessions; } @@ -92,6 +98,17 @@ function abortSignalOf(ctx: Context): AbortSignal | undefined { return isContextImpl(ctx) ? ctx.abortSignal : undefined; } +/** Default idle timeout for a sub-harness turn; `settings.extra.idleTimeoutMs` overrides (0 disables). */ +const DEFAULT_SUB_HARNESS_IDLE_TIMEOUT_MS = 120_000; + +function resolveIdleTimeoutMs(extra: Record | undefined): number { + const raw = extra?.idleTimeoutMs; + if (typeof raw === 'number' && Number.isFinite(raw)) { + return raw; + } + return DEFAULT_SUB_HARNESS_IDLE_TIMEOUT_MS; +} + function buildRunContext(ctx: Context): SubHarnessRunContext { return { cwd: ctx.cwdState.cwd, @@ -156,6 +173,30 @@ interface SessionResolution { reuseKey?: string; } +/** + * Start one session. Kept as a named function so callers can hold the pending + * promise — it returns synchronously, before the instruction resolve inside it + * awaits, which is what lets the reuse store be populated without a dedupe + * window. + */ +function startSession( + step: StepSubHarness, + harness: SubHarness, + ctx: Context, + baseCtx: Context, + history: ReadonlyArray, +): Promise { + return resolveLazy(step.instructions, ctx).then((instructions) => + harness.doStart({ + settings: step.settings, + instructions, + history, + ctx: buildRunContext(baseCtx), + signal: abortSignalOf(baseCtx), + }), + ); +} + async function startOrReuseSession( step: StepSubHarness, harness: SubHarness, @@ -169,24 +210,32 @@ async function startOrReuseSession( const existing = store.get(reuseKey); if (existing) { return { - session: existing, + session: await existing, reuseKey, }; } } - const session = await harness.doStart({ - settings: step.settings, - instructions: await resolveLazy(step.instructions, ctx), - history, - ctx: buildRunContext(baseCtx), - signal: abortSignalOf(baseCtx), - }); + // Build the start promise SYNCHRONOUSLY (the instruction resolve happens + // inside it) so the store.set below lands before any await — otherwise two + // steps racing on the key both pass the store check during the instruction + // resolution and start duplicate sessions. + const startPromise = startSession(step, harness, ctx, baseCtx, history); if (reuseKey) { - store.set(reuseKey, session); + store.set(reuseKey, startPromise); + try { + return { + session: await startPromise, + reuseKey, + }; + } catch (e) { + // A failed start must not poison the key for later retries. + store.delete(reuseKey); + throw e; + } } return { - session, + session: await startPromise, reuseKey, }; } @@ -251,21 +300,56 @@ export async function executeSubHarness( const bridge = new SubHarnessEventBridge(step, baseCtx); bridge.begin(); + // Idle watchdog: an external coding agent is a real process that can hang + // (CLI waiting on stdin, SDK deadlock). The model-call path has a + // stream-idle watchdog; this is its analogue for sub-harness turns — no + // stream part for `idleTimeoutMs` aborts the per-turn signal. Every + // forwarded part feeds the watchdog. + const idleTimeoutMs = resolveIdleTimeoutMs(step.settings?.extra); + const turnController = new AbortController(); + const ctxSignal = abortSignalOf(baseCtx); + const turnSignal = ctxSignal + ? AbortSignal.any([ + ctxSignal, + turnController.signal, + ]) + : turnController.signal; + let idleStalled = false; + let idleTimer: ReturnType | null = null; + const armWatchdog = (): void => { + if (idleTimeoutMs <= 0) { + return; + } + if (idleTimer) { + clearTimeout(idleTimer); + } + idleTimer = setTimeout(() => { + idleStalled = true; + turnController.abort(`sub-harness turn idle for ${idleTimeoutMs}ms`); + }, idleTimeoutMs); + }; + armWatchdog(); + let result: SubHarnessTurnResult; try { result = await resolution.session.doPromptTurn({ prompt: turnText, - emit: (part) => bridge.forward(part), + emit: (part) => { + armWatchdog(); + bridge.forward(part); + }, // Per-turn signal, so a session reused across turns is cancelled by the - // context running the CURRENT turn rather than the one that started it. - signal: abortSignalOf(baseCtx), + // context running the CURRENT turn rather than the one that started it, + // and the idle watchdog can cut a stalled turn. + signal: turnSignal, }); } catch (e) { - // Best-effort teardown of a fresh session before surfacing the failure; - // reused sessions are left intact for a later step to retry against. - if (!resolution.reuseKey) { - await teardownSession(resolution.session, 'destroy').catch(() => undefined); + // A failed turn may leave the external runtime wedged or partially + // advanced. Never reuse it: evict the promise before best-effort teardown. + if (resolution.reuseKey) { + sessionStore(baseCtx).delete(resolution.reuseKey); } + await teardownSession(resolution.session, 'destroy').catch(() => undefined); if (e instanceof NoeticErrorImpl) { throw e; } @@ -277,16 +361,52 @@ export async function executeSubHarness( reason: baseCtx.abortReason ?? 'context aborted', }); } + if (idleStalled) { + throw new NoeticErrorImpl({ + kind: 'step_failed', + stepId: step.id, + cause: new Error( + `Sub-harness turn produced no output for ${idleTimeoutMs}ms (idle timeout).`, + ), + retriesExhausted: false, + }); + } throw new NoeticErrorImpl({ kind: 'step_failed', stepId: step.id, cause: e instanceof Error ? e : new Error(String(e)), retriesExhausted: false, }); + } finally { + if (idleTimer) { + clearTimeout(idleTimer); + } } bridge.finalize(result); applyTurnResult(baseCtx, result); + + // A turn the AGENT reports as failed must not flow downstream as a normal + // answer. The items/usage are already applied (the spend is real and the + // transcript is the evidence); the step itself fails so loop onError / + // callers can retry or abort instead of trusting a broken result. + if (result.finishReason === 'error') { + if (resolution.reuseKey) { + sessionStore(baseCtx).delete(resolution.reuseKey); + } + await teardownSession(resolution.session, 'destroy').catch(() => undefined); + throw new NoeticErrorImpl({ + kind: 'step_failed', + stepId: step.id, + cause: new Error( + `Sub-harness turn finished with finishReason 'error'. Last output: ${ + (result.text || extractAssistantText(result.items)).slice(0, 500) || '(none)' + }`, + ), + retriesExhausted: false, + }); + } + await finalizeSession(resolution, step.session, baseCtx); const lastText = result.text.length > 0 ? result.text : extractAssistantText(result.items); diff --git a/packages/core/test/interpreter/execute-sub-harness.test.ts b/packages/core/test/interpreter/execute-sub-harness.test.ts index cbba6085..69941834 100644 --- a/packages/core/test/interpreter/execute-sub-harness.test.ts +++ b/packages/core/test/interpreter/execute-sub-harness.test.ts @@ -556,3 +556,359 @@ describe('executeSubHarness', () => { //#endregion }); + +describe('sub-harness session/turn reliability', () => { + it("fails the step when a turn finishes with finishReason 'error'", async () => { + const { ctx } = harnessCtx(); + const adapter: SubHarness = { + specificationVersion: 'harness-v1', + harnessId: 'claude-code', + async doStart(): Promise { + return { + sessionId: 'session-1', + isResume: false, + async doPromptTurn(turn): Promise { + turn.emit({ + type: 'text-delta', + delta: 'partial work before crash', + }); + return { + items: [ + makeMessage('assistant', 'partial work before crash'), + ], + text: 'partial work before crash', + finishReason: 'error', + }; + }, + async doStop() { + return { + harnessId: 'claude-code', + sessionId: 'session-1', + state: null, + }; + }, + }; + }, + }; + + try { + await execute( + step.claudeCode({ + id: 'failing-turn', + harness: adapter, + prompt: 'do the thing', + }), + undefined, + ctx, + ); + throw new Error('should have thrown'); + } catch (e) { + if (!isNoeticError(e)) { + throw e; + } + expect(e.noeticError.kind).toBe('step_failed'); + expect(e.message).toContain('partial work before crash'); + } + // The spend/transcript are still applied — the evidence is in the log. + expect(ctx.itemLog.items.some((i) => i.type === 'message' && i.role === 'assistant')).toBe( + true, + ); + }); + + it("evicts and destroys a reused session when finishReason is 'error'", async () => { + const harness = new AgentHarness({ + name: 'test', + params: {}, + }); + let starts = 0; + let destroys = 0; + const adapter: SubHarness = { + specificationVersion: 'harness-v1', + harnessId: 'claude-code', + async doStart(): Promise { + starts++; + const current = starts; + return { + sessionId: `session-${current}`, + isResume: false, + async doPromptTurn(): Promise { + if (current === 1) { + return { + items: [], + text: 'failed', + finishReason: 'error', + }; + } + return { + items: [ + makeMessage('assistant', 'recovered'), + ], + text: 'recovered', + }; + }, + async doStop() { + return { + harnessId: 'claude-code', + sessionId: `session-${current}`, + state: null, + }; + }, + async doDestroy() { + destroys++; + }, + }; + }, + }; + const mk = (id: string) => + step.claudeCode({ + id, + harness: adapter, + prompt: 'go', + session: { + reuse: 'error-session', + }, + }); + + await expect(execute(mk('first'), undefined, harness.createContext())).rejects.toThrow( + "finishReason 'error'", + ); + expect(await execute(mk('second'), undefined, harness.createContext())).toBe('recovered'); + expect(starts).toBe(2); + expect(destroys).toBe(1); + }); + + it('cuts a stalled turn with the idle watchdog', async () => { + const { ctx } = harnessCtx(); + const adapter: SubHarness = { + specificationVersion: 'harness-v1', + harnessId: 'claude-code', + async doStart(): Promise { + return { + sessionId: 'session-1', + isResume: false, + async doPromptTurn(turn): Promise { + turn.emit({ + type: 'text-delta', + delta: 'starting…', + }); + // Hang until the turn signal aborts, then reject like a vendor SDK. + await new Promise((resolve) => { + turn.signal?.addEventListener('abort', () => resolve(), { + once: true, + }); + }); + throw new Error('aborted by watchdog'); + }, + async doStop() { + return { + harnessId: 'claude-code', + sessionId: 'session-1', + state: null, + }; + }, + }; + }, + }; + + const t0 = performance.now(); + try { + await execute( + step.claudeCode({ + id: 'stalling-turn', + harness: adapter, + prompt: 'hang forever', + settings: { + extra: { + idleTimeoutMs: 50, + }, + }, + }), + undefined, + ctx, + ); + throw new Error('should have thrown'); + } catch (e) { + if (!isNoeticError(e)) { + throw e; + } + expect(performance.now() - t0).toBeLessThan(5_000); + expect(e.noeticError.kind).toBe('step_failed'); + expect(e.message).toContain('idle'); + } + }); + + it('evicts and destroys a reused session when its turn fails', async () => { + const harness = new AgentHarness({ + name: 'test', + params: {}, + }); + let starts = 0; + let destroys = 0; + const adapter: SubHarness = { + specificationVersion: 'harness-v1', + harnessId: 'claude-code', + async doStart(): Promise { + starts++; + const current = starts; + return { + sessionId: `session-${current}`, + isResume: false, + async doPromptTurn(turn): Promise { + if (current === 1) { + throw new Error('wedged session'); + } + turn.emit({ + type: 'text-delta', + delta: 'recovered', + }); + return { + items: [ + makeMessage('assistant', 'recovered'), + ], + text: 'recovered', + }; + }, + async doStop() { + return { + harnessId: 'claude-code', + sessionId: `session-${current}`, + state: null, + }; + }, + async doDestroy() { + destroys++; + }, + }; + }, + }; + const mk = (id: string) => + step.claudeCode({ + id, + harness: adapter, + prompt: 'go', + session: { + reuse: 'recoverable-session', + }, + }); + + await expect(execute(mk('first'), undefined, harness.createContext())).rejects.toThrow( + 'wedged session', + ); + expect(await execute(mk('second'), undefined, harness.createContext())).toBe('recovered'); + expect(starts).toBe(2); + expect(destroys).toBe(1); + }); + + it('dedupes session start when two steps race on one reuse key', async () => { + const harness = new AgentHarness({ + name: 'test', + params: {}, + }); + let starts = 0; + const adapter: SubHarness = { + specificationVersion: 'harness-v1', + harnessId: 'claude-code', + async doStart(): Promise { + starts++; + // Slow start: without promise-valued dedupe both racers pass the + // store check during this window and start duplicate sessions. + await new Promise((resolve) => setTimeout(resolve, 20)); + return { + sessionId: 'session-1', + isResume: false, + async doPromptTurn(turn): Promise { + turn.emit({ + type: 'text-delta', + delta: 'ok', + }); + return { + items: [ + makeMessage('assistant', 'ok'), + ], + text: 'ok', + }; + }, + async doStop() { + return { + harnessId: 'claude-code', + sessionId: 'session-1', + state: null, + }; + }, + }; + }, + }; + const mk = (id: string) => + step.claudeCode({ + id, + harness: adapter, + prompt: 'go', + session: { + reuse: 'shared-session', + }, + }); + + await Promise.all([ + execute(mk('racer-1'), undefined, harness.createContext()), + execute(mk('racer-2'), undefined, harness.createContext()), + ]); + expect(starts).toBe(1); + }); + + it('clears the reuse key when session start fails so a later step can retry', async () => { + const harness = new AgentHarness({ + name: 'test', + params: {}, + }); + let attempts = 0; + const adapter: SubHarness = { + specificationVersion: 'harness-v1', + harnessId: 'claude-code', + async doStart(): Promise { + attempts++; + if (attempts === 1) { + throw new Error('vendor start failed'); + } + return { + sessionId: 'session-2', + isResume: false, + async doPromptTurn(turn): Promise { + turn.emit({ + type: 'text-delta', + delta: 'recovered', + }); + return { + items: [ + makeMessage('assistant', 'recovered'), + ], + text: 'recovered', + }; + }, + async doStop() { + return { + harnessId: 'claude-code', + sessionId: 'session-2', + state: null, + }; + }, + }; + }, + }; + const mk = (id: string) => + step.claudeCode({ + id, + harness: adapter, + prompt: 'go', + session: { + reuse: 'flaky-session', + }, + }); + + await expect(execute(mk('first'), undefined, harness.createContext())).rejects.toThrow( + 'vendor start failed', + ); + const result = await execute(mk('second'), undefined, harness.createContext()); + expect(result).toBe('recovered'); + expect(attempts).toBe(2); + }); +}); diff --git a/packages/sub-harness-pi/test/pi.test.ts b/packages/sub-harness-pi/test/pi.test.ts index c2068a94..f502ec17 100644 --- a/packages/sub-harness-pi/test/pi.test.ts +++ b/packages/sub-harness-pi/test/pi.test.ts @@ -64,8 +64,10 @@ describe('pi adapter', () => { assert(result.usage); expect(result.usage.total).toBe(5); expect(result.cost).toBe(0.004); - // Reasoning does not produce an item; only the assistant message does. - expect(result.items).toHaveLength(1); + // Reasoning is preserved as an item ahead of the assistant message. + expect(result.items).toHaveLength(2); + expect(result.items[0]?.type).toBe('reasoning'); + expect(result.items[1]?.type).toBe('message'); // stream-start + reasoning-delta + 2 text-deltas + finish forwarded to emit. expect(emitted).toHaveLength(5); expect(emitted.some((p) => p.type === 'reasoning-delta')).toBe(true); diff --git a/packages/sub-harness/README.md b/packages/sub-harness/README.md index 0dc4ee7f..6752cc00 100644 --- a/packages/sub-harness/README.md +++ b/packages/sub-harness/README.md @@ -18,9 +18,10 @@ runs a sub-harness via `step.claudeCode(...)`, `step.codex(...)`, etc., or via a - **`SubHarnessStreamPart` (+ Zod schema)** — the event model an adapter emits during a turn. - **`SubHarnessTurnAccumulator`** — collects stream parts into a - `SubHarnessTurnResult` (assistant message + tool-call Items, text, usage). -- **`assistantMessageItem` / `functionCallItem`** — build Noetic `Item`s from - agent output. + `SubHarnessTurnResult` (reasoning + assistant message + tool-call Items, + text, usage). +- **`assistantMessageItem` / `functionCallItem` / `reasoningItem`** — build + Noetic `Item`s from agent output. - **`createSubHarnessRegistry`** — key adapters by id for JSON-workflow hydration (`HydrationContext.subHarnesses`). - **`commonTool`** — declare the cross-harness built-in tool vocabulary. diff --git a/packages/sub-harness/src/define.ts b/packages/sub-harness/src/define.ts index c758ac74..d1a8b141 100644 --- a/packages/sub-harness/src/define.ts +++ b/packages/sub-harness/src/define.ts @@ -95,6 +95,9 @@ function createRunnerSession( return accumulator.result(); }, async doStop() { + // Stateless runner sessions persist nothing: `state: null` means "no + // resume payload exists" — resuming a stopped runner session starts + // fresh. Adapters with real session state override the whole session. return { harnessId: def.harnessId, sessionId, diff --git a/packages/sub-harness/src/index.ts b/packages/sub-harness/src/index.ts index 2a0b5391..cd29b16b 100644 --- a/packages/sub-harness/src/index.ts +++ b/packages/sub-harness/src/index.ts @@ -44,7 +44,7 @@ export { SubHarnessStartError, } from './errors'; export { formatConversation, withHistoryPrompt } from './history'; -export { asItems, assistantMessageItem, functionCallItem } from './items'; +export { asItems, assistantMessageItem, functionCallItem, reasoningItem } from './items'; export type { SubHarnessRegistry } from './registry'; export { createSubHarnessRegistry } from './registry'; export type { SubHarnessTurnAccumulatorOptions } from './turn'; diff --git a/packages/sub-harness/src/items.ts b/packages/sub-harness/src/items.ts index bfc840a7..06ca6089 100644 --- a/packages/sub-harness/src/items.ts +++ b/packages/sub-harness/src/items.ts @@ -5,9 +5,24 @@ * constructible. */ -import type { FunctionCallItem, Item, MessageItem } from '@noetic-tools/types'; +import type { FunctionCallItem, Item, MessageItem, ReasoningItem } from '@noetic-tools/types'; import { frameworkCast } from '@noetic-tools/types'; +/** @public Build a reasoning Item from accumulated thinking text. */ +export function reasoningItem(text: string, id?: string): ReasoningItem { + return frameworkCast({ + id: id ?? `reasoning-${crypto.randomUUID()}`, + type: 'reasoning', + status: 'completed', + content: [ + { + type: 'reasoning_text', + text, + }, + ], + }); +} + /** @public Build an assistant message Item from plain text. */ export function assistantMessageItem(text: string, id?: string): MessageItem { return frameworkCast({ diff --git a/packages/sub-harness/src/turn.ts b/packages/sub-harness/src/turn.ts index 180f48a4..954ce783 100644 --- a/packages/sub-harness/src/turn.ts +++ b/packages/sub-harness/src/turn.ts @@ -12,7 +12,7 @@ import type { SubHarnessTurnResult, TokenUsage, } from '@noetic-tools/types'; -import { assistantMessageItem, functionCallItem } from './items'; +import { assistantMessageItem, functionCallItem, reasoningItem } from './items'; interface CollectedToolCall { toolCallId: string; @@ -98,6 +98,11 @@ export class SubHarnessTurnAccumulator { harnessMetadata?: Record; }): SubHarnessTurnResult { const items: Item[] = []; + // Reasoning precedes the answer in the agent's turn order; retaining it + // means the item log, checkpoints, and eval scorers see the thinking. + if (this.reasoning.length > 0) { + items.push(reasoningItem(this.reasoning)); + } if (this.text.length > 0) { items.push(assistantMessageItem(this.text)); } diff --git a/packages/sub-harness/test/sub-harness.test.ts b/packages/sub-harness/test/sub-harness.test.ts index 6fbf7e2f..ff16703c 100644 --- a/packages/sub-harness/test/sub-harness.test.ts +++ b/packages/sub-harness/test/sub-harness.test.ts @@ -100,6 +100,31 @@ describe('SubHarnessTurnAccumulator', () => { expect(result.usage.total).toBe(15); }); + test('retains reasoning deltas as a reasoning item ahead of the answer', () => { + const acc = new SubHarnessTurnAccumulator(); + acc.push({ + type: 'reasoning-delta', + delta: 'let me think ', + }); + acc.push({ + type: 'reasoning-delta', + delta: 'about this…', + }); + acc.push({ + type: 'text-delta', + delta: 'the answer', + }); + acc.push({ + type: 'finish', + finishReason: 'stop', + }); + const result = acc.result(); + expect(result.items).toHaveLength(2); + expect(result.items[0]?.type).toBe('reasoning'); + expect(JSON.stringify(result.items[0])).toContain('let me think about this…'); + expect(result.items[1]?.type).toBe('message'); + }); + test('emits no assistant item when there is no text', () => { const acc = new SubHarnessTurnAccumulator(); acc.push({ diff --git a/packages/web/content/docs/framework/sub-harnesses.mdx b/packages/web/content/docs/framework/sub-harnesses.mdx index a8115efb..eecd40ef 100644 --- a/packages/web/content/docs/framework/sub-harnesses.mdx +++ b/packages/web/content/docs/framework/sub-harnesses.mdx @@ -97,7 +97,7 @@ interface SubHarnessSessionPolicy { } ``` -- `reuse` keys a live session (workspace + conversation history + running runtime) that survives across steps. Two steps with the same `reuse` key share one session, so the second turn sees the first turn's history. A reused session is kept alive by default. +- `reuse` keys a live session (workspace + conversation history + running runtime) that survives across steps. Two steps with the same `reuse` key share one session, so the second turn sees the first turn's history. Concurrent first turns on one key dedupe onto a single session start. A reused session is kept alive by default. - `onComplete` chooses teardown: `'stop'` (default for a fresh session) persists state and stops the runtime, `'detach'` parks it for later resume, `'destroy'` discards it with no resume state. ## Conversation history @@ -141,6 +141,12 @@ Two guarantees back this: - **All output is mapped.** Every part an adapter emits — `text-delta`, `reasoning-delta`, `tool-call`, `file-change`, `finish`, … — becomes the corresponding stream event. Adapters never drop vendor output: anything unrecognized is surfaced as a `raw` part rather than discarded. - **A turn always emits.** Even an adapter that streams nothing still brackets the turn with a completion event, and a result returned without streaming is synthesized into events. Set `emit: false` on the step to suppress all of it. +### Turn failure and the idle watchdog + +A turn the agent reports as failed (`finishReason: 'error'`) fails the step with a `step_failed` error — partial output is kept in the item log as evidence, but it is never returned as a successful result. And because an external coding agent is a real process that can hang, each turn is guarded by an idle watchdog: if no stream part arrives for `settings.extra.idleTimeoutMs` (default 120s, `0` disables), the turn is aborted and the step fails instead of hanging the run. + +Reasoning deltas are retained as a `reasoning` item in the turn result, so the item log, checkpoints, and eval scorers see the agent's thinking. + ## JSON workflow node The same four agents are available in the [JSON Workflow Runtime](/docs/framework/json-runtime) as four node kinds. A node names the agent by `kind`; the hydrator resolves the adapter from a registry you pass in: @@ -225,7 +231,7 @@ The stream-part union covers `stream-start`, `text-delta`, `reasoning-delta`, `t `commonTool(nativeName, commonName?, description?)` maps an agent's native tool name (Claude's `Bash`, Codex's `shell`, pi's `bash`) to a shared cross-harness name (`shell`), so consumers can recognize "the same kind of tool" across agents. -The base package also exports the building blocks the lifecycle uses internally: `SubHarnessTurnAccumulator` (collects stream parts into a `SubHarnessTurnResult`), the `asItems` / `assistantMessageItem` / `functionCallItem` item builders, and the `SubHarnessCapabilityError` / `SubHarnessStartError` error types (with `isSubHarnessCapabilityError` / `isSubHarnessStartError` guards). A `SubHarnessSession` requires only `doPromptTurn` and `doStop`; the rest of the lifecycle (`doContinueTurn`, `doSuspendTurn`, `doDetach`, `doDestroy`, `doCompact`) is optional and signalled by presence. An adapter that cannot satisfy an optional capability throws `SubHarnessCapabilityError` from the relevant method rather than advertising a static capabilities object. +The base package also exports the building blocks the lifecycle uses internally: `SubHarnessTurnAccumulator` (collects stream parts into a `SubHarnessTurnResult`), the `asItems` / `assistantMessageItem` / `functionCallItem` / `reasoningItem` item builders, and the `SubHarnessCapabilityError` / `SubHarnessStartError` error types (with `isSubHarnessCapabilityError` / `isSubHarnessStartError` guards). A `SubHarnessSession` requires only `doPromptTurn` and `doStop`; the rest of the lifecycle (`doContinueTurn`, `doSuspendTurn`, `doDetach`, `doDestroy`, `doCompact`) is optional and signalled by presence. An adapter that cannot satisfy an optional capability throws `SubHarnessCapabilityError` from the relevant method rather than advertising a static capabilities object. ## Core decoupling guarantee diff --git a/specs/27-sub-harness-steps.md b/specs/27-sub-harness-steps.md index e05c1628..5eb016cc 100644 --- a/specs/27-sub-harness-steps.md +++ b/specs/27-sub-harness-steps.md @@ -140,10 +140,28 @@ unrecognized rather than dropping it. ### Session reuse `session.reuse` keys a live session that survives across steps (stored on the -`AgentHarness`). `session.onComplete` chooses the teardown: `'stop'` (default for +`AgentHarness`). The store holds PROMISES of sessions, so two steps racing on +the same key (e.g. parallel legs) dedupe onto one `doStart`; a failed start +removes the key so a later step can retry. Driving concurrent TURNS on one +reused session remains unsupported — sessions are conversational state. +`session.onComplete` chooses the teardown: `'stop'` (default for a fresh session) persists and stops the runtime, `'detach'` parks it, `'destroy'` discards it. A reused session is kept alive by default. +### Turn failure, idle watchdog, and reasoning retention + +A turn that finishes with `finishReason: 'error'` fails the step with a +`step_failed` error whose message includes the last output; the turn's items +and usage are still applied to the context first (the spend is real, the +transcript is the evidence), but the partial result is never returned as +success. Each turn is also guarded by an idle watchdog: no stream part for +`settings.extra.idleTimeoutMs` (default 120_000, `0` disables) aborts the +per-turn signal and fails the step, so a wedged vendor session cannot hang the +run. Finally, the turn accumulator retains accumulated `reasoning-delta` text +as a `reasoning` item (built by the `reasoningItem()` builder) ahead of the +assistant message, so the item log, checkpoints, and eval scorers see the +agent's thinking. + ## JSON workflow nodes The same agents are available in the JSON runtime as four node kinds. A node