diff --git a/README.md b/README.md index 0e99274..1cf9483 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ The installed extension generates this compact index from its executable capabil | Name | Classification | Signature | Options and defaults | | --- | --- | --- | --- | -| agent | runtime-global | `agent(prompt, options?) => Promise` | `label`: string (optional; default: derived from phase and call count)
`phase`: string (optional; default: current phase)
`schema`: plain JSON Schema (optional)
`model`: string (optional)
`tier`: string (optional)
`isolation`: "worktree" (optional)
`agentType`: string (optional)
`timeoutMs`: number \| null (optional; default: run timeout; null disables)
`retries`: number (optional; default: run retry count) | +| agent | runtime-global | `agent(prompt, options?) => Promise` | `label`: string (optional; default: derived from phase and call count)
`phase`: string (optional; default: current phase)
`schema`: plain JSON Schema (optional)
`model`: string (optional)
`tier`: string (optional)
`isolation`: "worktree" (optional)
`thread`: string (optional)
`agentType`: string (optional)
`timeoutMs`: number \| null (optional; default: run timeout; null disables)
`retries`: number (optional; default: run retry count) | | parallel | runtime-global | `parallel(thunks) => Promise>` | — | | pipeline | runtime-global | `pipeline(items, ...stages) => Promise>` | — | | workflow | runtime-global | `workflow(savedName, childArgs?) => Promise` | — | @@ -250,7 +250,7 @@ Extension state lives outside the repository under `~/.pi/workflows`: - project runs, journals, locks, and saved overrides: `~/.pi/workflows/projects//` - older project-local `.pi/workflows/runs` and `.pi/workflows/saved` remain readable as fallbacks -Subagents are in-memory by default. Set `persistAgentSessions: true` to retain full transcripts in Pi's standard session directory. This creates one file per agent and may store sensitive material that an agent read, so enable it deliberately. +Subagents are in-memory by default. Set `persistAgentSessions: true` to retain full transcripts in Pi's standard session directory. This creates one file per unthreaded call or named thread and may store sensitive material that an agent read, so enable it deliberately. Completed background runs persist their full result in the project run JSON. The conversation delivery includes a pointer to that file when the visible summary is shortened. diff --git a/docs/workflow-authoring.md b/docs/workflow-authoring.md index 1a87218..b95ffab 100644 --- a/docs/workflow-authoring.md +++ b/docs/workflow-authoring.md @@ -11,7 +11,7 @@ See [Workflow prompt guidance rationale](workflow-prompt-guidance-rationale.md) | Name | Classification | Signature | Options and defaults | | --- | --- | --- | --- | -| agent | runtime-global | `agent(prompt, options?) => Promise` | `label`: string (optional; default: derived from phase and call count)
`phase`: string (optional; default: current phase)
`schema`: plain JSON Schema (optional)
`model`: string (optional)
`tier`: string (optional)
`isolation`: "worktree" (optional)
`agentType`: string (optional)
`timeoutMs`: number \| null (optional; default: run timeout; null disables)
`retries`: number (optional; default: run retry count) | +| agent | runtime-global | `agent(prompt, options?) => Promise` | `label`: string (optional; default: derived from phase and call count)
`phase`: string (optional; default: current phase)
`schema`: plain JSON Schema (optional)
`model`: string (optional)
`tier`: string (optional)
`isolation`: "worktree" (optional)
`thread`: string (optional)
`agentType`: string (optional)
`timeoutMs`: number \| null (optional; default: run timeout; null disables)
`retries`: number (optional; default: run retry count) | | parallel | runtime-global | `parallel(thunks) => Promise>` | — | | pipeline | runtime-global | `pipeline(items, ...stages) => Promise>` | — | | workflow | runtime-global | `workflow(savedName, childArgs?) => Promise` | — | diff --git a/docs/workflow-context-surfaces.json b/docs/workflow-context-surfaces.json index ab58d4b..5a7de50 100644 --- a/docs/workflow-context-surfaces.json +++ b/docs/workflow-context-surfaces.json @@ -38,11 +38,11 @@ "workflowAuthoringSkillCorpus": { "serialization": "sum of UTF-8 bytes for every file under skills/workflow-authoring", "files": 28, - "bytes": 73626 + "bytes": 74599 }, "representativeAuthoringProfiles": { "serialization": "sum of UTF-8 bytes for each profile's declared package-relative files", - "medianBytes": 12366, + "medianBytes": 12917, "profiles": [ { "name": "write", @@ -54,7 +54,7 @@ "skills/workflow-authoring/examples/fan-out-and-synthesize.js", "skills/workflow-authoring/examples/structured-output.js" ], - "bytes": 12260 + "bytes": 12811 }, { "name": "edit", @@ -66,7 +66,7 @@ "skills/workflow-authoring/examples/phased-budgets.js", "skills/workflow-authoring/examples/saved-nested-workflows.js" ], - "bytes": 16972 + "bytes": 17523 }, { "name": "review", @@ -77,7 +77,7 @@ "skills/workflow-authoring/references/quality-helpers.md", "skills/workflow-authoring/examples/adversarial-verification.js" ], - "bytes": 11106 + "bytes": 11657 }, { "name": "debug", @@ -88,7 +88,7 @@ "skills/workflow-authoring/references/specialized-helpers.md", "skills/workflow-authoring/examples/validated-gate.js" ], - "bytes": 12472 + "bytes": 13023 }, { "name": "loop", @@ -101,7 +101,7 @@ "skills/workflow-authoring/examples/loop-until-done.js", "skills/workflow-authoring/examples/structured-output.js" ], - "bytes": 17508 + "bytes": 18059 }, { "name": "retry", @@ -113,7 +113,7 @@ "skills/workflow-authoring/examples/bounded-semantic-retry.js", "skills/workflow-authoring/examples/structured-output.js" ], - "bytes": 11238 + "bytes": 11789 } ] } diff --git a/docs/workflow-guidance-baseline.json b/docs/workflow-guidance-baseline.json index a057405..2667beb 100644 --- a/docs/workflow-guidance-baseline.json +++ b/docs/workflow-guidance-baseline.json @@ -3,6 +3,6 @@ "algorithm": "sha256", "surfaces": { "compactGuidance": "7c699635fe8677afdcb3b57fddc457ea68f9250ef4b4413dc6f385dc1b910d48", - "detailedProse": "12b7410148045c90edecdc22e113b40d863b5ad4d23844ea204632e454960400" + "detailedProse": "155acccb7e3003543f5263d7f7ea3f52b873248e82f37d9d7860f7a1952c0c81" } } diff --git a/skills/workflow-authoring/references/capabilities.md b/skills/workflow-authoring/references/capabilities.md index e94895b..09d29fc 100644 --- a/skills/workflow-authoring/references/capabilities.md +++ b/skills/workflow-authoring/references/capabilities.md @@ -11,7 +11,7 @@ This compact generated index covers supported runtime globals and workflow-tool | Name | Classification | Signature | Options and defaults | | --- | --- | --- | --- | -| agent | runtime-global | `agent(prompt, options?) => Promise` | `label`: string (optional; default: derived from phase and call count)
`phase`: string (optional; default: current phase)
`schema`: plain JSON Schema (optional)
`model`: string (optional)
`tier`: string (optional)
`isolation`: "worktree" (optional)
`agentType`: string (optional)
`timeoutMs`: number \| null (optional; default: run timeout; null disables)
`retries`: number (optional; default: run retry count) | +| agent | runtime-global | `agent(prompt, options?) => Promise` | `label`: string (optional; default: derived from phase and call count)
`phase`: string (optional; default: current phase)
`schema`: plain JSON Schema (optional)
`model`: string (optional)
`tier`: string (optional)
`isolation`: "worktree" (optional)
`thread`: string (optional)
`agentType`: string (optional)
`timeoutMs`: number \| null (optional; default: run timeout; null disables)
`retries`: number (optional; default: run retry count) | | parallel | runtime-global | `parallel(thunks) => Promise>` | — | | pipeline | runtime-global | `pipeline(items, ...stages) => Promise>` | — | | workflow | runtime-global | `workflow(savedName, childArgs?) => Promise` | — | diff --git a/skills/workflow-authoring/references/capability-details.md b/skills/workflow-authoring/references/capability-details.md index 41922c9..4c6d335 100644 --- a/skills/workflow-authoring/references/capability-details.md +++ b/skills/workflow-authoring/references/capability-details.md @@ -19,6 +19,7 @@ Every exact fact below is projected from the installed extension's capability co - `model`: string (optional; highest-priority exact model selector) - `tier`: string (optional; configured route name; dynamic reference: model-routes) - `isolation`: "worktree" (optional) +- `thread`: string (optional; non-empty name; same-name calls must be sequential) - `agentType`: string (optional; must come from provided context; dynamic reference: agent-types) - `timeoutMs`: number | null (optional; default: run timeout; null disables) - `retries`: number (optional; default: run retry count; finite values are floored and clamped to 0..3) @@ -26,6 +27,9 @@ Every exact fact below is projected from the installed extension's capability co - Constraint: schema noncompliance after bounded structured-output repair is nonrecoverable and bypasses agent retries - Constraint: per-agent retries override invocation retries; retries are floored and clamped to 0..3 - Constraint: resume replays only the longest unchanged prefix; the first miss and every later call execute live +- Constraint: a named thread retains its full Pi transcript and session identity only within one uninterrupted workflow invocation +- Constraint: threaded calls are live-execution resume barriers and are never journaled +- Constraint: same-thread calls must be sequential; threads cannot use worktree isolation - Constraint: selector priority is explicit model > agentType model > tier > phase model > metadata model > implicit medium > session default - Constraint: an explicit model, agentType model, tier, or phase model that resolves to an unavailable model throws MODEL_NOT_FOUND naming the source (e.g. the tier and what it resolved to) instead of falling back - Constraint: only the implicit default medium tier (no explicit model, tier, agentType, or phase model requested) degrades to the session default when unavailable, logging a one-time run-visible warning instead of throwing diff --git a/skills/workflow-authoring/references/runtime.md b/skills/workflow-authoring/references/runtime.md index a98d809..460a1bc 100644 --- a/skills/workflow-authoring/references/runtime.md +++ b/skills/workflow-authoring/references/runtime.md @@ -18,6 +18,8 @@ The runtime supplies `agent`, `parallel`, `pipeline`, `workflow`, quality/contro Call `agent(prompt, { label, schema? })`; it returns text, a schema-validated value, or recoverable `null`. Nonrecoverable limit, validation, and budget failures throw. Record each intended work ID before filtering. A `null` means missing coverage, never a negative finding. +Use `agent(prompt, { thread: "implementer" })` when the same subagent must receive a later follow-up with its complete conversation intact, such as implementer → separate reviewer → implementer revision. Reuse a thread name sequentially; never put same-thread calls in one `parallel()` batch. Nested workflows share thread names with their parent. Threads exist only during the current uninterrupted workflow invocation, cannot use worktree isolation, and restart from the beginning after pause/resume because threaded results are not journaled. + When JavaScript reads fields, pass a small plain JSON Schema. Schema noncompliance after repair throws and bypasses agent retries. Catch it only to return an explicit incomplete outcome without reading missing fields. Return objects, arrays, strings, numbers, booleans, and `null`—not functions, promises, cycles, `BigInt`, or runtime handles. ## Routing and support diff --git a/src/agent.ts b/src/agent.ts index b13cdcf..1b7c0de 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -492,6 +492,8 @@ export interface AgentRunOptions = TSchemaDef extends TSchema @@ -553,6 +555,13 @@ export class WorkflowAgent { * run. See onModelFallback below for the (still-loud) degrade path. */ private warnedDefaultTierUnavailable = false; + /** + * Named conversations live for this WorkflowAgent instance. Production creates + * one instance per workflow invocation; embedders that inject and reuse an + * agent are responsible for choosing the longer thread lifetime deliberately. + */ + private readonly threadSessions = new Map(); + private readonly activeThreads = new Set(); constructor(options: WorkflowAgentOptions = {}) { this.cwd = options.cwd ?? process.cwd(); @@ -676,21 +685,32 @@ export class WorkflowAgent { * agent to an in-memory session instead — the run continues, just without a * persisted transcript. */ - private createSessionManager(): SessionManager { - if (!this.persistAgentSessions) return SessionManager.inMemory(); - try { - const manager = SessionManager.create(this.cwd); - this.assertSessionDirWritable(manager.getSessionDir()); - warnPersistSecretsOnce(manager.getSessionDir()); - return manager; - } catch (error) { - console.warn( - `[workflow] persistAgentSessions: could not persist this agent's session (${ - error instanceof Error ? error.message : String(error) - }); continuing with an in-memory session`, - ); - return SessionManager.inMemory(); + private createSessionManager(thread?: string): SessionManager { + if (thread) { + const existing = this.threadSessions.get(thread); + if (existing) return existing; } + + let manager: SessionManager; + if (!this.persistAgentSessions) { + manager = SessionManager.inMemory(); + } else { + try { + manager = SessionManager.create(this.cwd); + this.assertSessionDirWritable(manager.getSessionDir()); + warnPersistSecretsOnce(manager.getSessionDir()); + } catch (error) { + console.warn( + `[workflow] persistAgentSessions: could not persist this agent's session (${ + error instanceof Error ? error.message : String(error) + }); continuing with an in-memory session`, + ); + manager = SessionManager.inMemory(); + } + } + + if (thread) this.threadSessions.set(thread, manager); + return manager; } /** Best-effort write probe: throws if the session directory isn't actually writable. */ @@ -703,6 +723,29 @@ export class WorkflowAgent { async run( prompt: string, options: AgentRunOptions = {}, + ): Promise> { + const thread = options.thread; + if (thread && this.activeThreads.has(thread)) { + throw new WorkflowError( + `agent thread "${thread}" is already running; same-thread calls must be sequential`, + WorkflowErrorCode.SCRIPT_VALIDATION_ERROR, + { + recoverable: false, + agentLabel: options.label, + }, + ); + } + if (thread) this.activeThreads.add(thread); + try { + return await this.runTurn(prompt, options); + } finally { + if (thread) this.activeThreads.delete(thread); + } + } + + private async runTurn( + prompt: string, + options: AgentRunOptions = {}, ): Promise> { const capture: StructuredOutputCapture = { called: false, value: undefined }; // Per-call cwd (e.g. a worktree) needs coding tools bound to that directory, @@ -808,39 +851,57 @@ export class WorkflowAgent { // per-call runCwd: agents working in short-lived git worktrees should still // group under the project's session dir instead of scattering across // temporary worktree paths. - const sessionManager = this.createSessionManager(); - const { session } = await createAgentSession({ - cwd: runCwd, - agentDir, - sessionManager, - // Use real SettingsManager to inherit user's default provider/model settings. - // SettingsManager.inMemory() doesn't load ~/.pi/settings.json, so subagents - // would fall back to the first available model (e.g. openai-codex) which may - // not have valid auth, causing silent empty responses. - settingsManager: SettingsManager.create(this.cwd, agentDir), - customTools, - // Shared per-run loader with no host extensions (#109) — see - // getSharedResourceLoader. An injected resourceLoader (tests / embedders) - // wins and skips the shared build entirely; the ...this.sessionOptions - // spread below re-applies the same injected value harmlessly. - resourceLoader: this.sessionOptions.resourceLoader ?? (await this.getSharedResourceLoader(agentDir)), - // Share the resolved registry's ModelRuntime (catalog + auth, including - // extension-registered providers) with the subagent session. pi >= 0.80.8 - // takes modelRuntime here; the old modelRegistry option is gone. - ...(modelRuntime ? { modelRuntime } : {}), - ...this.sessionOptions, - // Per-call model/thinking wins over any sessionOptions defaults. - ...(resolvedModel ? { model: resolvedModel } : {}), - ...(resolvedThinkingLevel ? { thinkingLevel: resolvedThinkingLevel } : {}), - // Deny recursive-orchestration tools in the subagent (#107). Placed after - // the sessionOptions spread so it always applies; folds in any denylist - // the caller set on sessionOptions rather than dropping it. - excludeTools: subagentExcludedTools(this.excludeTools, this.sessionOptions.excludeTools), - }); + const sessionManager = this.createSessionManager(options.thread); + const threadLeaf = options.thread ? sessionManager.getLeafId() : null; + let session: Awaited>["session"]; + try { + ({ session } = await createAgentSession({ + cwd: runCwd, + agentDir, + sessionManager, + // Use real SettingsManager to inherit user's default provider/model settings. + // SettingsManager.inMemory() doesn't load ~/.pi/settings.json, so subagents + // would fall back to the first available model (e.g. openai-codex) which may + // not have valid auth, causing silent empty responses. + settingsManager: SettingsManager.create(this.cwd, agentDir), + customTools, + // Shared per-run loader with no host extensions (#109) — see + // getSharedResourceLoader. An injected resourceLoader (tests / embedders) + // wins and skips the shared build entirely; the ...this.sessionOptions + // spread below re-applies the same injected value harmlessly. + resourceLoader: this.sessionOptions.resourceLoader ?? (await this.getSharedResourceLoader(agentDir)), + // Share the resolved registry's ModelRuntime (catalog + auth, including + // extension-registered providers) with the subagent session. pi >= 0.80.8 + // takes modelRuntime here; the old modelRegistry option is gone. + ...(modelRuntime ? { modelRuntime } : {}), + ...this.sessionOptions, + // Named threads must retain their own manager even when an embedder supplied + // a default manager for ordinary one-shot calls. + ...(options.thread ? { sessionManager } : {}), + // Per-call model/thinking wins over any sessionOptions defaults. + ...(resolvedModel ? { model: resolvedModel } : {}), + ...(resolvedThinkingLevel ? { thinkingLevel: resolvedThinkingLevel } : {}), + // Deny recursive-orchestration tools in the subagent (#107). Placed after + // the sessionOptions spread so it always applies; folds in any denylist + // the caller set on sessionOptions rather than dropping it. + excludeTools: subagentExcludedTools(this.excludeTools, this.sessionOptions.excludeTools), + })); + } catch (error) { + if (options.thread) this.restoreThreadLeaf(sessionManager, threadLeaf); + throw error; + } + const usageBeforeTurn = options.thread ? session.getSessionStats() : undefined; + const messagesBeforeTurn = session.messages.length; // Name the persisted session so it's identifiable in session pickers. - // Skip when an injected session.sessionManager override won (tests/embedders). - if (this.persistAgentSessions && !this.sessionOptions.sessionManager && options.sessionName) { + // Skip when an injected session.sessionManager override won (tests/embedders), + // and name a threaded session only on its first turn. + if ( + this.persistAgentSessions && + (!this.sessionOptions.sessionManager || options.thread) && + (!options.thread || !threadLeaf) && + options.sessionName + ) { try { sessionManager.appendSessionInfo(options.sessionName); } catch { @@ -851,6 +912,7 @@ export class WorkflowAgent { let removeAbortListener: (() => void) | undefined; let removeHistoryListener: (() => void) | undefined; let lastHistoryEmit = 0; + let threadTurnSucceeded = false; const emitHistory = () => options.onHistory?.(compactAgentHistory(session.messages)); const maybeEmitHistory = () => { if (!options.onHistory) return; @@ -881,22 +943,26 @@ export class WorkflowAgent { throwIfProviderLimit(session.messages, options.label); if (options.schema) { - return (await resolveStructuredOutput(session, capture, options.schema, options, (m) => - this.lastAssistantText(m), + const result = (await resolveStructuredOutput(session, capture, options.schema, options, (messages) => + this.lastAssistantText(messages.slice(messagesBeforeTurn)), )) as AgentRunResult; + threadTurnSucceeded = true; + return result; } // Unstructured result: require assistant text AFTER the last tool result. // Text emitted before it is stale progress (the agent's last real action was // a tool call) — accepting it would report an incomplete run as successful - // and suppress the AGENT_EMPTY_OUTPUT retry (#111). - const text = this.finalAssistantText(session.messages); + // and suppress the AGENT_EMPTY_OUTPUT retry (#111). A threaded session's + // restored transcript is excluded so an empty turn cannot reuse an old answer. + const text = this.finalAssistantText(session.messages.slice(messagesBeforeTurn)); if (!text.trim()) { throw new WorkflowError("Subagent produced no assistant output", WorkflowErrorCode.AGENT_EMPTY_OUTPUT, { recoverable: true, agentLabel: options.label, }); } + threadTurnSucceeded = true; return text as AgentRunResult; } finally { removeAbortListener?.(); @@ -906,10 +972,27 @@ export class WorkflowAgent { } catch { // History is diagnostic only; never let it mask the real result/error. } + if (options.thread && !threadTurnSucceeded) { + this.restoreThreadLeaf(sessionManager, threadLeaf); + } // Read real usage before disposing — dispose tears down the session state. if (options.onUsage) { try { - const usage = usageFromStats(session.getSessionStats()); + const stats = session.getSessionStats(); + const usage = usageFromStats( + usageBeforeTurn + ? { + tokens: { + input: Math.max(0, stats.tokens.input - usageBeforeTurn.tokens.input), + output: Math.max(0, stats.tokens.output - usageBeforeTurn.tokens.output), + cacheRead: Math.max(0, stats.tokens.cacheRead - usageBeforeTurn.tokens.cacheRead), + cacheWrite: Math.max(0, stats.tokens.cacheWrite - usageBeforeTurn.tokens.cacheWrite), + total: Math.max(0, stats.tokens.total - usageBeforeTurn.tokens.total), + }, + cost: Math.max(0, stats.cost - usageBeforeTurn.cost), + } + : stats, + ); if (usage) options.onUsage(usage); } catch { // Usage is best-effort; never let stats failure mask the real result/error. @@ -919,6 +1002,11 @@ export class WorkflowAgent { } } + private restoreThreadLeaf(sessionManager: SessionManager, leafId: string | null): void { + if (leafId) sessionManager.branch(leafId); + else sessionManager.resetLeaf(); + } + private buildPrompt(prompt: string, options: AgentRunOptions, structured: boolean): string { const parts = [ this.instructions, diff --git a/src/workflow-authoring-coverage.ts b/src/workflow-authoring-coverage.ts index 0b2d439..739018b 100644 --- a/src/workflow-authoring-coverage.ts +++ b/src/workflow-authoring-coverage.ts @@ -37,7 +37,7 @@ export const WORKFLOW_AUTHORING_FROZEN_FILES = [ }, { path: "skills/workflow-authoring/references/runtime.md", - sha256: "13b710424bc262a3951e9ad8f7387cdf4d630b89f27c07bbcbc6b47c37e24242", + sha256: "9c4a693f6916330462450a8b031866409f234ffe29b4e9f9a4d221d46997a209", }, { path: "skills/workflow-authoring/references/helpers.md", @@ -126,6 +126,13 @@ const CAPABILITY_SCENARIOS: Readonly> = { }; const FROZEN_GUIDANCE_BY_CAPABILITY: Readonly> = { + "workflow.runtime.agent": [ + { + path: RUNTIME_PATH, + requiredText: + 'Use `agent(prompt, { thread: "implementer" })` when the same subagent must receive a later follow-up with its complete conversation intact, such as implementer → separate reviewer → implementer revision. Reuse a thread name sequentially; never put same-thread calls in one `parallel()` batch. Threads exist only during the current uninterrupted workflow invocation, cannot use worktree isolation, and restart from the beginning after pause/resume because threaded results are not journaled.', + }, + ], "workflow.runtime.pipeline": [ { path: RUNTIME_PATH, diff --git a/src/workflow-capability-contract.ts b/src/workflow-capability-contract.ts index b3d09d4..3670c36 100644 --- a/src/workflow-capability-contract.ts +++ b/src/workflow-capability-contract.ts @@ -180,6 +180,7 @@ const AGENT_OPTIONS: OptionShape = { option("model", "string", true, null, ["highest-priority exact model selector"]), option("tier", "string", true, null, ["configured route name"], "model-routes"), option("isolation", '"worktree"', true), + option("thread", "string", true, null, ["non-empty name; same-name calls must be sequential"]), option("agentType", "string", true, null, ["must come from provided context"], "agent-types"), option("timeoutMs", "number | null", true, "run timeout; null disables"), option("retries", "number", true, "run retry count", ["finite values are floored and clamped to 0..3"]), @@ -308,6 +309,9 @@ const capabilities: readonly CapabilityDescriptor[] = [ "schema noncompliance after bounded structured-output repair is nonrecoverable and bypasses agent retries", "per-agent retries override invocation retries; retries are floored and clamped to 0..3", "resume replays only the longest unchanged prefix; the first miss and every later call execute live", + "a named thread retains its full Pi transcript and session identity only within one uninterrupted workflow invocation", + "threaded calls are live-execution resume barriers and are never journaled", + "same-thread calls must be sequential; threads cannot use worktree isolation", "selector priority is explicit model > agentType model > tier > phase model > metadata model > implicit medium > session default", "an explicit model, agentType model, tier, or phase model that resolves to an unavailable model throws MODEL_NOT_FOUND naming the source (e.g. the tier and what it resolved to) instead of falling back", "only the implicit default medium tier (no explicit model, tier, agentType, or phase model requested) degrades to the session default when unavailable, logging a one-time run-visible warning instead of throwing", diff --git a/src/workflow.ts b/src/workflow.ts index f32f6dc..c193dc0 100644 --- a/src/workflow.ts +++ b/src/workflow.ts @@ -140,6 +140,10 @@ export interface SharedRuntime { * after the run has been marked complete and torn down. See the drain below. */ inFlight: Set>; + /** Named conversations currently executing anywhere in this run tree. */ + activeThreads: Set; + /** Whether a threaded call has invalidated journal replay for the remaining run tree. */ + resumeBarrierReached: boolean; } /** Runtime instrumentation for workflow boundaries, quality helpers, and control attempts. */ @@ -314,6 +318,12 @@ export interface AgentOptions.md`, project > * user). Binds that definition's tool allow/denylist, model, and body prompt @@ -470,6 +480,8 @@ export async function runWorkflow( nestedCallSeq: 0, runFatalController: new AbortController(), inFlight: new Set>(), + activeThreads: new Set(), + resumeBarrierReached: false, }; const limiter = shared.limiter; // This frame created `shared` fresh (rather than inheriting a parent @@ -534,11 +546,32 @@ export async function runWorkflow( }; const agent = (prompt: string, agentOptions: AgentOptions = {}): Promise => { + const rawThread = agentOptions.thread; + const thread = rawThread === undefined ? undefined : typeof rawThread === "string" ? rawThread.trim() : ""; + let call: Promise; + if (rawThread !== undefined && !thread) { + call = Promise.reject( + new WorkflowError("agent() thread must be a non-empty string", WorkflowErrorCode.SCRIPT_VALIDATION_ERROR, { + recoverable: false, + }), + ); + } else if (thread && shared.activeThreads.has(thread)) { + call = Promise.reject( + new WorkflowError( + `agent thread "${thread}" is already running; same-thread calls must be sequential`, + WorkflowErrorCode.SCRIPT_VALIDATION_ERROR, + { recoverable: false }, + ), + ); + } else { + if (thread) shared.activeThreads.add(thread); + call = agentImpl(prompt, thread === rawThread ? agentOptions : { ...agentOptions, thread }); + if (thread) call = call.finally(() => shared.activeThreads.delete(thread)); + } // Track every call (awaited or not) so the top-level run can drain // outstanding calls before completing (see SharedRuntime.inFlight and the // drain in the finally below) — this is what stops a forgotten `await` // from letting an agent mutate state after the run is torn down. - const call = agentImpl(prompt, agentOptions); shared.inFlight.add(call); // Attaching a handler here (independent of whatever the script itself does // with the returned promise) also means an un-awaited call's eventual @@ -599,6 +632,13 @@ export async function runWorkflow( // Resolve a named agentType to its bound definition (tools/model/prompt). const agentDef = resolveAgentType(agentOptions.agentType, agentRegistry); + if (agentOptions.thread && (agentOptions.isolation === "worktree" || agentDef?.isolation === "worktree")) { + throw new WorkflowError( + `agent thread "${agentOptions.thread}" cannot use worktree isolation because worktrees are removed after each call`, + WorkflowErrorCode.SCRIPT_VALIDATION_ERROR, + { recoverable: false }, + ); + } if (agentOptions.agentType && !agentDef) { log(`unknown agentType "${agentOptions.agentType}"; using default tools/model`); } @@ -650,10 +690,11 @@ export async function runWorkflow( // exact `${runId}:${callIndex}` string) so a nested workflow()'s // callIndex-0 can never accidentally replay the parent's callIndex-0 // entry, or vice versa (see JournalEntry.runId). - const cached = options.resumeJournal?.get(deltaKey); + if (agentOptions.thread) shared.resumeBarrierReached = true; + const cached = agentOptions.thread ? undefined : options.resumeJournal?.get(deltaKey); const hashMatches = cached != null && cached.hash === callHash; const cachedEmptyOutput = hashMatches && isEmptyTextAgentResult(cached.result, agentOptions.schema); - if (hashMatches && !cachedEmptyOutput && callIndex < state.firstMiss) { + if (!shared.resumeBarrierReached && hashMatches && !cachedEmptyOutput && callIndex < state.firstMiss) { options.onAgentStart?.({ id: deltaKey, label, phase: assignedPhase, prompt, model: displayModel }); options.onAgentEnd?.({ id: deltaKey, @@ -717,6 +758,7 @@ export async function runWorkflow( const externalSignal = options.signal; let onExternalAbort: (() => void) | undefined; let onRunFatal: (() => void) | undefined; + let attemptRunPromise: Promise | undefined; try { throwIfAborted(); // This agent's own fan-out already breached maxAgents while this @@ -748,7 +790,9 @@ export async function runWorkflow( const runPromise = agentRunner.run(prompt, { label, // Identifiable name for persisted sessions (persistAgentSessions). - sessionName: `workflow:${runId} ${label}`, + sessionName: agentOptions.thread + ? `workflow:${runId} thread:${agentOptions.thread}` + : `workflow:${runId} ${label}`, schema: agentOptions.schema, signal: agentController.signal, instructions: buildAgentInstructions(assignedPhase, agentOptions, agentDef, resolvedIsolation), @@ -779,7 +823,9 @@ export async function runWorkflow( onHistory: (history: AgentHistoryEntry[]) => { options.onAgentHistory?.({ id: deltaKey, label, phase: assignedPhase, history }); }, + thread: agentOptions.thread, }); + attemptRunPromise = runPromise; // After a timeout the run() promise still settles later, rejecting with // "aborted" once agentController fires; the race has already resolved, // so swallow that to avoid an unhandled rejection. @@ -795,13 +841,17 @@ export async function runWorkflow( } const tokens = recordTokens(result); - options.onAgentJournal?.({ - index: callIndex, - runId, - hash: callHash, - result, - storeDelta: store.commitDelta(deltaKey), - }); + if (!agentOptions.thread) { + options.onAgentJournal?.({ + index: callIndex, + runId, + hash: callHash, + result, + storeDelta: store.commitDelta(deltaKey), + }); + } else { + store.commitDelta(deltaKey); + } options.onAgentEnd?.({ id: deltaKey, label, @@ -814,6 +864,10 @@ export async function runWorkflow( }); return result; } catch (error) { + // A named thread cannot start its next turn while an aborted wrapper + // is still unwinding against the shared SessionManager. Wait for the + // wrapper to restore its prior leaf before retrying or returning. + if (agentOptions.thread && attemptRunPromise) await attemptRunPromise.catch(() => {}); if (isAborted()) throw error; const workflowError = wrapError(error, { agentLabel: label }); @@ -997,6 +1051,9 @@ export async function runWorkflow( sharedStore: store, resumeJournal: prefixIntact ? options.resumeJournal : undefined, resumeFromRunId: undefined, + // Reuse the same runner so named threads span parent/child frames but + // still die with this one top-level runWorkflow invocation. + agent: agentRunner, // shared.nestedCallSeq, not shared.depth — see its doc comment: depth // returns to 0 between sequential sibling calls, which would otherwise // mint the same child runId (and hence colliding deltaKeys/event ids) @@ -1199,7 +1256,7 @@ export async function runWorkflow( // Namespaced by runId like agent()'s deltaKey — see JournalEntry.runId. const journalKey = `${runId}:${callIndex}`; const cached = options.resumeJournal?.get(journalKey); - if (cached != null && cached.hash === callHash && callIndex < state.firstMiss) { + if (!shared.resumeBarrierReached && cached != null && cached.hash === callHash && callIndex < state.firstMiss) { shared.agentCount++; return cached.result; // replay the journaled human reply } @@ -1529,6 +1586,7 @@ function hashAgentCall( tier: options.tier ?? null, phase: phase ?? null, agentType: options.agentType ?? null, + ...(options.thread ? { thread: options.thread } : {}), // Resolved definition (tools/model/prompt) so editing an agent .md invalidates // this call's cached result on a later resume. agentDef: agentDefKey, diff --git a/tests/agent.test.ts b/tests/agent.test.ts index d799120..63cd354 100644 --- a/tests/agent.test.ts +++ b/tests/agent.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import test from "node:test"; import { createFauxCore, fauxAssistantMessage, fauxToolCall } from "@earendil-works/pi-ai"; -import { ModelRegistry, ModelRuntime } from "@earendil-works/pi-coding-agent"; +import { ModelRegistry, ModelRuntime, SessionManager } from "@earendil-works/pi-coding-agent"; import { Type } from "typebox"; import type { AgentRunOptions, AgentUsage } from "../src/agent.js"; import { @@ -26,7 +26,16 @@ type WorkflowAgentPrivates = { buildPrompt(prompt: string, options: AgentRunOptions, structured: boolean): string; lastAssistantText(messages: unknown[]): string; finalAssistantText(messages: unknown[]): string; - createSessionManager(): { isPersisted(): boolean; getCwd(): string }; + createSessionManager(thread?: string): { + isPersisted(): boolean; + getCwd(): string; + getSessionId(): string; + getSessionDir(): string; + getSessionFile(): string | undefined; + getLeafId(): string | null; + appendSessionInfo(name: string): string; + }; + restoreThreadLeaf(manager: ReturnType, leafId: string | null): void; }; // ═══════════════════════════════════════════════════════════════════════ @@ -65,6 +74,57 @@ test("WorkflowAgent with persistAgentSessions=true creates a file-backed manager } }); +test("WorkflowAgent retains one session manager per named thread", () => { + const agent = new WorkflowAgent({ cwd: "/tmp" }) as unknown as WorkflowAgentPrivates; + const first = agent.createSessionManager("implementer"); + const again = agent.createSessionManager("implementer"); + const reviewer = agent.createSessionManager("reviewer"); + + assert.equal(again, first); + assert.equal(again.getSessionId(), first.getSessionId()); + assert.notEqual(reviewer.getSessionId(), first.getSessionId()); + const nextInvocation = new WorkflowAgent({ cwd: "/tmp" }) as unknown as WorkflowAgentPrivates; + assert.notEqual(nextInvocation.createSessionManager("implementer").getSessionId(), first.getSessionId()); + assert.notEqual(agent.createSessionManager(), agent.createSessionManager(), "unthreaded calls remain one-shot"); +}); + +test("WorkflowAgent restores a failed named turn to its previous leaf", () => { + const agent = new WorkflowAgent({ cwd: "/tmp" }) as unknown as WorkflowAgentPrivates; + const manager = agent.createSessionManager("implementer"); + const acceptedLeaf = manager.appendSessionInfo("accepted"); + manager.appendSessionInfo("failed-attempt"); + + agent.restoreThreadLeaf(manager, acceptedLeaf); + assert.equal(manager.getLeafId(), acceptedLeaf); +}); + +test("persistAgentSessions uses one file per named thread", () => { + const dir = mkdtempSync(join(tmpdir(), "pi-dynamic-workflows-thread-files-")); + const projectCwd = join(dir, "project"); + const fakeHome = join(dir, "home"); + try { + withFakeHome(fakeHome, () => { + const agent = new WorkflowAgent({ + cwd: projectCwd, + persistAgentSessions: true, + }) as unknown as WorkflowAgentPrivates; + const implementer = agent.createSessionManager("implementer"); + implementer.appendSessionInfo("first turn"); + agent.createSessionManager("implementer").appendSessionInfo("second turn"); + agent.createSessionManager("reviewer").appendSessionInfo("review"); + + const sessionFiles = new Set([ + implementer.getSessionFile(), + agent.createSessionManager("implementer").getSessionFile(), + agent.createSessionManager("reviewer").getSessionFile(), + ]); + assert.equal(sessionFiles.size, 2); + }); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + test("WorkflowAgent degrades to in-memory when the session directory can't be created", () => { const dir = mkdtempSync(join(tmpdir(), "pi-dynamic-workflows-persist-agent-fail-")); const projectCwd = join(dir, "project"); @@ -310,6 +370,68 @@ test("WorkflowAgent.run(): tier routing resolves correctly through the real (non // parameters schema and fail with an opaque transport-level 400. // ═══════════════════════════════════════════════════════════════════════════ +test("failed or empty named turns restore the active transcript before the next call", async () => { + const home = mkdtempSync(join(tmpdir(), "pi-dw-thread-rollback-home-")); + const cwd = mkdtempSync(join(tmpdir(), "pi-dw-thread-rollback-cwd-")); + const core = createFauxCore({ + provider: "fauxtest-thread", + models: [{ id: "faux-model", name: "Faux Model", contextWindow: 128000, maxTokens: 4096 }], + }); + try { + await withFakeHomeAsync(home, async () => { + const runtime = await ModelRuntime.create({ authPath: join(home, "auth.json"), modelsPath: null }); + runtime.registerProvider("fauxtest-thread", { + name: "Faux Thread", + baseUrl: "http://127.0.0.1:9/faux", + apiKey: "faux-dummy-key-not-used", + api: core.api, + streamSimple: core.streamSimple as never, + models: core.models.map((model) => ({ + ...model, + input: ["text"] as ("text" | "image")[], + })), + }); + const contexts: unknown[] = []; + core.setResponses([ + () => { + throw new Error("attempt failed"); + }, + fauxAssistantMessage("accepted", { stopReason: "stop" }), + fauxAssistantMessage("", { stopReason: "stop" }), + (context) => { + contexts.push(context); + return fauxAssistantMessage("followed up", { stopReason: "stop" }); + }, + ]); + const injectedManager = SessionManager.inMemory(); + const agent = new WorkflowAgent({ + cwd, + modelRegistry: new ModelRegistry(runtime), + session: { sessionManager: injectedManager }, + }); + const runOptions = { thread: "implementer", model: "fauxtest-thread/faux-model" } as const; + + await assert.rejects(agent.run("FAILED_TURN_MARKER", runOptions)); + await agent.run("ACCEPTED_TURN_MARKER", runOptions); + await assert.rejects(agent.run("EMPTY_TURN_MARKER", runOptions), (error: unknown) => { + assert.ok(error instanceof WorkflowError); + assert.equal(error.code, WorkflowErrorCode.AGENT_EMPTY_OUTPUT); + return true; + }); + await agent.run("FOLLOWUP_TURN_MARKER", runOptions); + + const contextText = JSON.stringify(contexts[0]); + assert.doesNotMatch(contextText, /FAILED_TURN_MARKER|EMPTY_TURN_MARKER/); + assert.match(contextText, /ACCEPTED_TURN_MARKER/); + assert.match(contextText, /FOLLOWUP_TURN_MARKER/); + assert.equal(injectedManager.getLeafId(), null, "a named thread must not use the injected one-shot manager"); + }); + } finally { + rmSync(home, { recursive: true, force: true }); + rmSync(cwd, { recursive: true, force: true }); + } +}); + test("WorkflowAgent.run() rejects a non-object top-level schema before touching the model registry", async () => { const agent = new WorkflowAgent({ cwd: "/tmp" }); await assert.rejects( @@ -529,15 +651,23 @@ test("WorkflowAgent.run() still completes with a normal object schema (no regres const registry = new ModelRegistry(runtime); core.setResponses([ fauxAssistantMessage(fauxToolCall("structured_output", { verdict: "ok" }), { stopReason: "toolUse" }), + fauxAssistantMessage("", { stopReason: "stop" }), ]); const agent = new WorkflowAgent({ cwd, modelRegistry: registry }); - const result = await agent.run("task", { + const runOptions = { model: "fauxtest-schema/faux-model", schema: Type.Object({ verdict: Type.String() }), - }); + thread: "structured", + } as const; + const result = await agent.run("task", runOptions); assert.deepEqual(result, { verdict: "ok" }); + await assert.rejects(agent.run("empty follow-up", { ...runOptions, maxSchemaRetries: 0 }), (error: unknown) => { + assert.ok(error instanceof WorkflowError); + assert.equal(error.code, WorkflowErrorCode.SCHEMA_NONCOMPLIANCE); + return true; + }); }); } finally { rmSync(home, { recursive: true, force: true }); diff --git a/tests/workflow-runtime.test.ts b/tests/workflow-runtime.test.ts index bc4a81f..ad2dba6 100644 --- a/tests/workflow-runtime.test.ts +++ b/tests/workflow-runtime.test.ts @@ -1,4 +1,5 @@ import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; import test from "node:test"; import type { AgentUsage } from "../src/agent.js"; import { WorkflowError, WorkflowErrorCode } from "../src/errors.js"; @@ -79,6 +80,143 @@ return xs`; assert.equal(result.agentCount, 4); }); +test("named agent threads can be re-entered around a separate reviewer", async () => { + const calls: Array<{ prompt: string; thread?: string }> = []; + const result = await runWorkflow( + `export const meta = { name: 'thread_reentry', description: 're-enter implementer' } +const first = await agent('implement', { thread: 'implementer' }) +const review = await agent('review', { thread: 'reviewer' }) +const second = await agent('address review', { thread: 'implementer' }) +return { first, review, second }`, + { + agent: { + async run(prompt, options) { + calls.push({ prompt, thread: options?.thread }); + return `${options?.thread}:${prompt}`; + }, + }, + persistLogs: false, + }, + ); + + assert.deepEqual(calls, [ + { prompt: "implement", thread: "implementer" }, + { prompt: "review", thread: "reviewer" }, + { prompt: "address review", thread: "implementer" }, + ]); + assert.deepEqual(JSON.parse(JSON.stringify(result.result)), { + first: "implementer:implement", + review: "reviewer:review", + second: "implementer:address review", + }); +}); + +test("concurrent calls on one named thread are rejected before a second agent starts", async () => { + let starts = 0; + await assert.rejects( + runWorkflow( + `export const meta = { name: 'thread_concurrency', description: 'reject overlap' } +return await parallel([ + () => agent('first', { thread: 'implementer' }), + () => agent('second', { thread: 'implementer' }) +])`, + { + agent: { + async run() { + starts++; + await new Promise((resolve) => setTimeout(resolve, 20)); + return "ok"; + }, + }, + persistLogs: false, + }, + ), + /same-thread calls must be sequential/, + ); + assert.equal(starts, 1); +}); + +test("named threads reject worktree isolation", async () => { + await assert.rejects( + runWorkflow( + `export const meta = { name: 'thread_worktree', description: 'reject worktree' } +return await agent('work', { thread: 'implementer', isolation: 'worktree' })`, + { agent: countingAgent().runner, persistLogs: false }, + ), + /cannot use worktree isolation/, + ); +}); + +test("threaded calls are live resume barriers and are not journaled", async () => { + const firstJournal: JournalEntry[] = []; + const first = countingAgent(); + await runWorkflow( + `export const meta = { name: 'thread_resume', description: 'thread barrier' } +const before = await agent('before') +const threaded = await agent('threaded', { thread: 'implementer' }) +const after = await agent('after') +return { before, threaded, after }`, + { + agent: first.runner, + runId: "thread-run", + persistLogs: false, + onAgentJournal: (entry) => firstJournal.push(entry), + }, + ); + assert.deepEqual( + firstJournal.map((entry) => entry.index), + [0, 2], + ); + + const resumed = countingAgent(); + await runWorkflow( + `export const meta = { name: 'thread_resume', description: 'thread barrier' } +const before = await agent('before') +const threaded = await agent('threaded', { thread: 'implementer' }) +const after = await agent('after') +return { before, threaded, after }`, + { + agent: resumed.runner, + runId: "thread-run", + persistLogs: false, + resumeJournal: new Map(firstJournal.map((entry) => [`thread-run:${entry.index}`, entry])), + resumeFromRunId: "thread-run", + }, + ); + assert.equal(resumed.state.calls, 2, "the prefix replays, then the threaded call and all later calls run live"); +}); + +test("a timed-out named turn fully settles before retrying the thread", async () => { + let calls = 0; + let active = 0; + let maxActive = 0; + const result = await runWorkflow( + `export const meta = { name: 'thread_timeout_retry', description: 'safe retry' } +return await agent('work', { thread: 'implementer', timeoutMs: 5, retries: 1 })`, + { + agent: { + async run(_prompt, options) { + calls++; + active++; + maxActive = Math.max(maxActive, active); + if (calls === 1) { + await new Promise((_resolve, reject) => { + options?.signal?.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); + }).finally(() => active--); + } else { + active--; + return "ok"; + } + }, + }, + persistLogs: false, + }, + ); + + assert.equal(result.result, "ok"); + assert.equal(maxActive, 1); +}); + test("runWorkflow retries recoverable empty output then succeeds", async () => { let calls = 0; const journal: JournalEntry[] = []; @@ -366,6 +504,30 @@ test("resume replays cached results without re-running agents", async () => { assert.equal(JSON.stringify(r2.result), JSON.stringify(r1.result)); }); +test("unthreaded agent journal hashes remain compatible with pre-thread runs", async () => { + const journal: JournalEntry[] = []; + await runWorkflow( + `export const meta = { name: 'hash_compat', description: 'stable unthreaded hash' } +return await agent('work')`, + { + agent: countingAgent().runner, + persistLogs: false, + onAgentJournal: (entry) => journal.push(entry), + }, + ); + + const oldIdentity = JSON.stringify({ + prompt: "work", + model: null, + tier: null, + phase: null, + agentType: null, + agentDef: null, + schema: null, + }); + assert.equal(journal[0]?.hash, createHash("sha256").update(oldIdentity).digest("hex")); +}); + test("resume re-runs only the changed call (hash mismatch)", async () => { const first = countingAgent(); const journal: JournalEntry[] = []; @@ -483,6 +645,78 @@ return { a, nested }`; assert.equal(result.result.nested.child, "ran:child task"); }); +test("nested workflows share named agent threads with their parent", async () => { + const turns = new Map(); + const runner = { + async run(prompt: string, options?: { thread?: string }) { + const thread = options?.thread ?? "one-shot"; + const prior = turns.get(thread) ?? []; + prior.push(prompt); + turns.set(thread, prior); + return prior.join(" -> "); + }, + }; + const child = `export const meta = { name: 'child_thread', description: 'continue parent thread' } +return await agent('child', { thread: 'implementer' })`; + const parent = `export const meta = { name: 'parent_thread', description: 'share thread with child' } +const first = await agent('parent-before', { thread: 'implementer' }) +const nested = await workflow('child') +const last = await agent('parent-after', { thread: 'implementer' }) +return { first, nested, last }`; + + const result = await runWorkflow<{ first: string; nested: string; last: string }>(parent, { + agent: runner, + persistLogs: false, + loadSavedWorkflow: (name) => (name === "child" ? child : undefined), + }); + + assert.deepEqual(turns.get("implementer"), ["parent-before", "child", "parent-after"]); + assert.deepEqual(JSON.parse(JSON.stringify(result.result)), { + first: "parent-before", + nested: "parent-before -> child", + last: "parent-before -> child -> parent-after", + }); +}); + +test("a nested threaded call invalidates later parent journal entries", async () => { + const script = `export const meta = { name: 'parent_resume_barrier', description: 'propagate child barrier' } +const before = await agent('before') +await workflow('child') +const after = await agent('after') +const confirmed = await checkpoint('confirm', { default: false }) +return { before, after, confirmed }`; + const child = `export const meta = { name: 'child_resume_barrier', description: 'thread barrier' } +return await agent('threaded child', { thread: 'implementer' })`; + const journal: JournalEntry[] = []; + await runWorkflow(script, { + agent: countingAgent().runner, + runId: "nested-thread-barrier", + persistLogs: false, + loadSavedWorkflow: (name) => (name === "child" ? child : undefined), + confirm: async () => true, + onAgentJournal: (entry) => journal.push(entry), + }); + + const resumed = countingAgent(); + let confirmations = 0; + const result = await runWorkflow<{ before: string; after: string; confirmed: boolean }>(script, { + agent: resumed.runner, + runId: "nested-thread-barrier", + persistLogs: false, + loadSavedWorkflow: (name) => (name === "child" ? child : undefined), + resumeJournal: new Map(journal.map((entry) => [`${entry.runId}:${entry.index}`, entry])), + resumeFromRunId: "nested-thread-barrier", + confirm: async () => { + confirmations++; + return false; + }, + }); + + assert.equal(resumed.state.calls, 2, "the child thread and later parent agent both run live"); + assert.equal(confirmations, 1, "the later parent checkpoint also runs live"); + assert.equal(result.result.confirmed, false); +}); + test("workflow() nesting is one level deep (second level throws)", async () => { const map: Record = { gc: `export const meta = { name: 'gc', description: 'g' }