diff --git a/packages/workflows/CHANGELOG.md b/packages/workflows/CHANGELOG.md index 3b74f3330b..d70294f65c 100644 --- a/packages/workflows/CHANGELOG.md +++ b/packages/workflows/CHANGELOG.md @@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ## [Unreleased] +### Added + +- Added workflow tool stage introspection and control actions (`stages`, `stage`, `transcript`, `send`, `pause`, and `reload`) for inspecting stage state, reading transcripts, answering prompts, controlling live stages, pausing runs, and reloading workflow resources ([#1023](https://github.com/flora131/atomic/issues/1023)). + ### Changed - Added a final Ralph PR-preparation phase that reviews changes against the configured base branch, tries available GitHub credentials using local git identity as a hint, posts implementation notes as a PR comment, and creates a pull request when possible. @@ -15,6 +19,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed - Removed biasing stage-output and iteration-count context from Ralph reviewer prompts while making the comparison base branch explicit ([#1037](https://github.com/flora131/atomic/issues/1037)). +- Ordered snapshot transcript fallback entries chronologically before applying `tail`/`limit`, preserving terminal result/error entries after tools for missing or tied timestamps ([#1023](https://github.com/flora131/atomic/issues/1023)). +- Reloaded workflow resources directly for `workflow({ action: "reload" })` instead of queuing a literal `/workflow reload` follow-up ([#1023](https://github.com/flora131/atomic/issues/1023)). +- Kept pending prompts unresolved when `workflow({ action: "send" })` omits `text`, `response`, and `message`, while preserving explicit empty-string answers ([#1023](https://github.com/flora131/atomic/issues/1023)). - Included `deep-research-codebase` discovery-stage handoff files in the persisted run manifest. - Persisted `deep-research-codebase` final reports as dated Markdown research docs while retaining file-only handoffs for bounded aggregation. - Prevented `deep-research-codebase` aggregation from inlining large specialist transcripts by using file-only handoff artifacts ([#1016](https://github.com/flora131/atomic/issues/1016)). diff --git a/packages/workflows/README.md b/packages/workflows/README.md index 7d2a510a71..3243d5e105 100644 --- a/packages/workflows/README.md +++ b/packages/workflows/README.md @@ -192,6 +192,7 @@ registry.get("alpha"); // compiled workflow definition | undefined | `/workflow interrupt [run-id\|--all]` | Pause active/named/all active runs so they can resume | | `/workflow kill [run-id\|--all]` | Kill and remove active/named/all active runs from status | | `/workflow resume ` | Resume paused work or re-open a run snapshot | +| `/workflow reload` | Reload discovered workflow resources in-process | | `/workflow inputs ` | Print the input schema for a workflow | Input overrides are bare `key=value` tokens (no leading `--`). Values are JSON-parsed when possible, so numbers, booleans, and quoted strings work as expected (e.g. `count=3`, `flag=true`, `prompt="multi word value"`). A whole-object override can be passed as a single JSON token (e.g. `{"prompt":"...","count":3}`). @@ -207,11 +208,21 @@ Workflows always run as **background tasks** — the chat editor stays free whil "parameters": { "workflow": "string (optional) — workflow ID or normalized name", "inputs": "object (optional) — key/value map of workflow inputs", - "action": "'run' | 'list' | 'get' | 'inputs' | 'status' | 'interrupt' | 'kill' | 'resume'", - "runId": "optional run id or unique prefix; interrupt/kill default to the active run; use '--all' or all:true for interrupt/kill all", - "stageId": "optional stage id, prefix, or name for resume", - "message": "optional resume message", - "all": "optional boolean for interrupt/kill all", + "action": "'run' | 'list' | 'get' | 'inputs' | 'status' | 'stages' | 'stage' | 'transcript' | 'send' | 'pause' | 'interrupt' | 'kill' | 'resume' | 'reload'", + "runId": "optional run id or unique prefix; control actions default to the active run where safe; use '--all' or all:true for pause/interrupt/kill all", + "stageId": "optional stage id, prefix, or name for stage-scoped actions; cannot be combined with all:true", + "statusFilter": "optional stages filter: pending/running/awaiting_input/paused/blocked/completed/failed/skipped/all", + "format": "optional agent-facing output format: text or json", + "limit": "transcript-only maximum number of most recent entries; default 50", + "tail": "transcript-only last-N entry count; overrides limit", + "includeToolOutput": "transcript-only flag for snapshot tool-event output; live transcripts may not expose tool output", + "text": "optional string payload for send/resume; explicit empty text answers pending prompts", + "response": "optional structured payload for answering pending prompts; explicit empty response is valid", + "message": "optional string payload for send/resume when text is not provided", + "delivery": "optional send delivery mode: auto, answer, prompt, steer, followUp, or resume; auto prioritizes answer, then resume, steer, followUp", + "promptId": "optional pending prompt identifier for send/answer", + "reason": "optional human-readable reload reason", + "all": "optional boolean for pause/interrupt/kill all; cannot be combined with stageId", "task/tasks/chain": "optional direct workflow-native orchestration modes" } } @@ -219,6 +230,9 @@ Workflows always run as **background tasks** — the chat editor stays free whil - **`renderCall`** — renders a compact workflow call summary in the chat scroll. - **`renderResult`** — renders the result or dispatch banner; live progress continues through the widget and graph viewer. Named workflow runs are background-oriented. +- **`transcript`** — uses a registered live stage handle when one exists, even before live messages arrive; otherwise it falls back to stored stage snapshots. Snapshot entries are ordered chronologically before `tail`/`limit` is applied, with terminal result/error entries kept after tool entries when timestamps are missing or tied. `includeToolOutput` applies to snapshot tool events; live session transcripts may not expose tool output. +- **`send`** — answers pending stage prompts only when `text`, `response`, or `message` is present; an explicit empty string is a valid answer, while an omitted payload is a no-op. `delivery: "auto"` answers pending prompts first, then resumes paused stages, steers streaming stages, or queues a follow-up. +- **`reload`** — refreshes workflow resources directly in-process instead of queuing a literal `/workflow reload` chat follow-up. ### F2 keyboard shortcut diff --git a/packages/workflows/src/extension/index.ts b/packages/workflows/src/extension/index.ts index a140d8dc68..5f62f466dc 100644 --- a/packages/workflows/src/extension/index.ts +++ b/packages/workflows/src/extension/index.ts @@ -9,6 +9,7 @@ import { renderInputsSchema } from "../shared/render-inputs-schema.js"; import { WorkflowParametersSchema } from "./workflow-schema.js"; import { renderRunBanner, renderRunSummary } from "./renderers.js"; import type { RunEndPayload, RunStartPayload } from "./renderers.js"; +import type { StageSnapshot, StageStatus, ToolEvent } from "../shared/store-types.js"; import { store } from "../shared/store.js"; import { restoreOnSessionStart } from "../shared/persistence-restore.js"; import type { SessionManager } from "../shared/persistence-restore.js"; @@ -19,11 +20,13 @@ import { destroyAllRuns, resumeRun, pauseRun, + pauseAllRuns, interruptRun, interruptAllRuns, inspectRun, } from "../runs/background/status.js"; import { cancellationRegistry } from "../runs/background/cancellation-registry.js"; +import { stageControlRegistry } from "../runs/foreground/stage-control-registry.js"; import { registerIntercomParentSession } from "../intercom/intercom-bridge.js"; import { subscribeIntercomControl } from "../intercom/result-intercom.js"; import { buildIntercomCallbacks } from "../intercom/intercom-routing.js"; @@ -273,9 +276,8 @@ export interface ExtensionAPI { renderer: (payload: unknown) => string, ) => void; /** - * Inject a custom message into the chat history. Used by the inline - * workflow input form to emit a sticky card under `customType: - * "workflows:input-form"`. The card stays in scrollback and is + * Inject a custom message into chat history. Used by inline workflow surfaces + * such as `workflows:input-form`; cards stay in scrollback and are * re-rendered by the registered renderer on every `tui.requestRender()`. */ sendMessage?: ( @@ -398,9 +400,15 @@ export interface WorkflowToolArgs extends StageOptions { | "list" | "get" | "status" + | "stages" + | "stage" + | "transcript" + | "send" + | "pause" | "interrupt" | "kill" | "resume" + | "reload" | "inputs"; /** Canonical run identifier or unique prefix for status/interrupt/kill/resume. */ runId?: string; @@ -410,6 +418,16 @@ export interface WorkflowToolArgs extends StageOptions { stageId?: string; /** Optional message forwarded when resuming paused work. */ message?: string; + statusFilter?: StageStatus | "all"; + format?: "text" | "json"; + limit?: number; + tail?: number; + includeToolOutput?: boolean; + text?: string; + response?: unknown; + delivery?: "auto" | "answer" | "prompt" | "steer" | "followUp" | "resume"; + promptId?: string; + reason?: string; /** Direct single-task mode, or root task string when chain is present. */ task?: WorkflowDirectTaskItem | string; /** Direct top-level parallel mode. */ @@ -511,6 +529,136 @@ function workflowRunResultFromDetails( }; } +function stringifyWorkflowToolResult(result: WorkflowToolResult): string { + return JSON.stringify(result, null, 2); +} + +function compactWorkflowToolMessage( + result: Extract, +): string { + if (result.action === "reload") { + return `${result.action}: ${result.status} — ${result.message}`; + } + const target = [ + result.runId, + result.action === "send" ? result.stageId : undefined, + ].filter((part): part is string => part !== undefined && part.length > 0) + .join("/"); + return `${result.action}:${target ? ` ${target}` : ""} ${result.status} — ${result.message}`; +} + +function renderTranscriptToolContent( + result: Extract, +): string { + const lines = [ + `action: transcript`, + `runId: ${result.runId}`, + `stageId: ${result.stageId}`, + `source: ${result.source}`, + `truncated: ${result.truncated}`, + ]; + if (result.sessionId) lines.push(`sessionId: ${result.sessionId}`); + if (result.sessionFile) lines.push(`sessionFile: ${result.sessionFile}`); + if (result.entries.length === 0) { + lines.push("entries: none"); + return lines.join("\n"); + } + lines.push("entries:"); + result.entries.forEach((entry, index) => { + const metadata = [ + `[${index + 1}]`, + `role=${entry.role}`, + entry.toolName ? `tool=${entry.toolName}` : undefined, + entry.timestamp !== undefined ? `timestamp=${entry.timestamp}` : undefined, + ].filter((part): part is string => part !== undefined); + lines.push(metadata.join(" ")); + if (entry.text !== undefined) lines.push(entry.text); + if (entry.output !== undefined) { + lines.push("tool output:"); + lines.push(entry.output); + } + if (entry.text === undefined && entry.output === undefined) { + lines.push("(no body)"); + } + }); + return lines.join("\n"); +} + +function renderStagesToolContent( + result: Extract, +): string { + const lines = [ + "action: stages", + `runId: ${result.runId}`, + `filter: ${result.filter}`, + ]; + if (result.error) lines.push(`error: ${result.error}`); + if (result.stages.length === 0) { + lines.push("stages: none"); + return lines.join("\n"); + } + lines.push("stages:"); + result.stages.forEach((stage, index) => { + lines.push(`[${index + 1}] ${stage.name} (${stage.id}) ${stage.status}`); + if (stage.sessionId) lines.push(`sessionId: ${stage.sessionId}`); + if (stage.sessionFile) lines.push(`sessionFile: ${stage.sessionFile}`); + if (stage.error) lines.push(`error: ${stage.error}`); + if (stage.awaitingInputSince !== undefined) { + lines.push(`awaitingInputSince: ${stage.awaitingInputSince}`); + } + if (stage.pendingPrompt !== undefined) { + lines.push("pendingPrompt:"); + lines.push(JSON.stringify(stage.pendingPrompt, null, 2)); + } + }); + return lines.join("\n"); +} + +function renderStageToolContent( + result: Extract, +): string { + const lines = ["action: stage", `runId: ${result.runId}`]; + if (result.error || result.stage === undefined) { + lines.push(`error: ${result.error ?? "stage not found"}`); + return lines.join("\n"); + } + lines.push("stage:"); + lines.push(JSON.stringify(result.stage, null, 2)); + return lines.join("\n"); +} + +function renderWorkflowToolContent( + result: WorkflowToolResult, + args: WorkflowToolArgs, +): string { + if (args.format === "json") return stringifyWorkflowToolResult(result); + + switch (result.action) { + case "transcript": + return renderTranscriptToolContent(result); + case "stages": + return renderStagesToolContent(result); + case "stage": + return renderStageToolContent(result); + case "send": + case "pause": + case "reload": + case "interrupt": + case "kill": + case "resume": + return compactWorkflowToolMessage(result); + case "list": + case "status": + case "statusDetail": + case "inputs": + case "get": + case "run": + return stringifyWorkflowToolResult(result); + } +} + function workflowGetResult( runtime: ExtensionRuntime, args: WorkflowToolArgs, @@ -550,6 +698,243 @@ function workflowGetResult( }; } +// --------------------------------------------------------------------------- +// Stage tool helpers +// --------------------------------------------------------------------------- + +type WorkflowStageSummary = { + id: string; + name: string; + status: StageStatus; + sessionId?: string; + sessionFile?: string; + error?: string; + awaitingInputSince?: number; + pendingPrompt?: StageSnapshot["pendingPrompt"]; +}; + +type WorkflowTranscriptEntry = { + role: string; + text?: string; + toolName?: string; + output?: string; + timestamp?: number; +}; + +type MessageContentBlock = { readonly type?: string; readonly text?: string }; +type MessageLike = { + readonly role?: string; + readonly content?: string | readonly MessageContentBlock[]; + readonly name?: string; + readonly toolName?: string; + readonly timestamp?: number; + readonly createdAt?: number; +}; + +function cloneStage(stage: StageSnapshot): StageSnapshot { + return structuredClone(stage); +} + +function summarizeStage(stage: StageSnapshot): WorkflowStageSummary { + return { + id: stage.id, + name: stage.name, + status: stage.status, + sessionId: stage.sessionId, + sessionFile: stage.sessionFile, + error: stage.error, + awaitingInputSince: stage.awaitingInputSince, + pendingPrompt: stage.pendingPrompt === undefined + ? undefined + : structuredClone(stage.pendingPrompt), + }; +} + +const DEFAULT_TRANSCRIPT_LIMIT = 50; + +function boundedCount(args: WorkflowToolArgs): number { + const raw = args.tail ?? args.limit; + if (raw === undefined) return DEFAULT_TRANSCRIPT_LIMIT; + if (!Number.isFinite(raw) || raw <= 0) return 0; + return Math.floor(raw); +} + +function applyEntryLimit( + entries: readonly T[], + args: WorkflowToolArgs, +): { entries: T[]; truncated: boolean } { + const count = boundedCount(args); + if (count === 0) { + return { entries: [], truncated: false }; + } + if (entries.length <= count) { + return { entries: [...entries], truncated: false }; + } + return { entries: entries.slice(entries.length - count), truncated: true }; +} + +function messageText(content: MessageLike["content"]): string | undefined { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return undefined; + let sawTextBlock = false; + const text = content + .map((block) => { + if (block.type === "text" && typeof block.text === "string") { + sawTextBlock = true; + return block.text; + } + return ""; + }) + .join(""); + return sawTextBlock ? text : undefined; +} + +function transcriptEntryFromMessage(message: MessageLike): WorkflowTranscriptEntry { + const entry: WorkflowTranscriptEntry = { role: message.role ?? "unknown" }; + const text = messageText(message.content); + if (text !== undefined) entry.text = text; + const toolName = message.toolName ?? message.name; + if (toolName !== undefined) entry.toolName = toolName; + const timestamp = message.timestamp ?? message.createdAt; + if (timestamp !== undefined) entry.timestamp = timestamp; + return entry; +} + +function transcriptEntriesFromToolEvents( + events: readonly ToolEvent[], + includeOutput: boolean, +): WorkflowTranscriptEntry[] { + return events.map((event) => ({ + role: "tool", + toolName: event.name, + output: includeOutput ? event.output : undefined, + timestamp: event.endedAt ?? event.startedAt, + })); +} + +function hasPayloadProperty(args: WorkflowToolArgs): boolean { + return ( + args.text !== undefined || + args.response !== undefined || + args.message !== undefined + ); +} + +function promptPayloadFromArgs(args: WorkflowToolArgs): unknown { + if (args.response !== undefined) return args.response; + if (args.text !== undefined) return args.text; + return args.message; +} + +function textPayloadFromArgs(args: WorkflowToolArgs): string | undefined { + if (args.text !== undefined) return args.text; + if (typeof args.response === "string") { + return args.response; + } + if (args.message !== undefined) return args.message; + return undefined; +} + +type WorkflowSendToolResult = Extract; + +function workflowSendResult( + runId: string, + stageId: string, + delivery: WorkflowSendToolResult["delivery"], + status: WorkflowSendToolResult["status"], + message: string, +): WorkflowSendToolResult { + return { action: "send", runId, stageId, delivery, status, message }; +} + +function sortTranscriptEntriesChronologically( + entries: readonly WorkflowTranscriptEntry[], +): WorkflowTranscriptEntry[] { + return entries + .map((entry, index) => ({ entry, index })) + .sort((a, b) => { + const aTimestamp = a.entry.timestamp; + const bTimestamp = b.entry.timestamp; + if ( + typeof aTimestamp === "number" && + typeof bTimestamp === "number" && + aTimestamp !== bTimestamp + ) { + return aTimestamp - bTimestamp; + } + return a.index - b.index; + }) + .map(({ entry }) => entry); +} + +function terminalTranscriptEntry( + role: "assistant" | "notice", + text: string, + endedAt: number | undefined, +): WorkflowTranscriptEntry { + const entry: WorkflowTranscriptEntry = { role, text }; + if (endedAt !== undefined) entry.timestamp = endedAt; + return entry; +} + +function snapshotTranscriptEntries( + snapshot: StageSnapshot | undefined, + includeOutput: boolean, +): WorkflowTranscriptEntry[] { + if (snapshot === undefined) return []; + const entries: WorkflowTranscriptEntry[] = [ + ...transcriptEntriesFromToolEvents(snapshot.toolEvents ?? [], includeOutput), + ]; + if (snapshot.result !== undefined) { + entries.push(terminalTranscriptEntry("assistant", snapshot.result, snapshot.endedAt)); + } + if (snapshot.error !== undefined) { + entries.push(terminalTranscriptEntry("notice", snapshot.error, snapshot.endedAt)); + } + return sortTranscriptEntriesChronologically(entries); +} + +function stageFailureMessage( + runId: string, + resultReason: string, + action: "pause" | "interrupt", +): string { + switch (resultReason) { + case "not_found": + return `Run not found: ${runId}`; + case "already_ended": + return `Run already ended: ${runId}`; + case "stage_not_found": + return `Stage not found for run: ${runId}`; + default: + return `No active stages to ${action} for run: ${runId}`; + } +} + +function inFlightRunCount(): number { + return store.runs().filter((run) => run.endedAt === undefined).length; +} + +function reloadBlockedMessage(count = inFlightRunCount()): string { + return `Reload skipped: ${count} workflow run(s) still in flight. Wait for them to finish, or pause/kill them before reloading workflow resources.`; +} + +function allStageConflictMessage(action: "pause" | "interrupt" | "kill"): string { + return `Cannot ${action} --all with a stageId; omit stageId or target a single run.`; +} + +class WorkflowReloadBlockedError extends Error { + constructor(message: string) { + super(message); + this.name = "WorkflowReloadBlockedError"; + } +} + +function reloadFailureMessage(error: unknown): string { + if (error instanceof WorkflowReloadBlockedError) return error.message; + return `Reload failed: ${error instanceof Error ? error.message : String(error)}`; +} + // --------------------------------------------------------------------------- // Tool execute — dispatch with real registry for list/inputs/run (Phase E) // + real status/interrupt/resume (Phase D) @@ -558,6 +943,7 @@ function workflowGetResult( export function makeExecuteWorkflowTool( runtime: ExtensionRuntime | ((ctx: PiExecuteContext) => ExtensionRuntime), getPersistence: () => WorkflowPersistencePort | undefined, + reloadWorkflowResources: () => Promise | void, ) { return async function executeWorkflowTool( args: WorkflowToolArgs, @@ -615,15 +1001,322 @@ export function makeExecuteWorkflowTool( const snapshots = store.runs().filter((r) => r.endedAt === undefined); return { action: "status", - snapshots: snapshots.map( - (s) => JSON.parse(JSON.stringify(s)) as typeof s, - ), + snapshots: snapshots.map((snapshot) => structuredClone(snapshot)), + }; + } + + case "stages": { + const target = resolveToolRunTarget(args, "No active run to inspect."); + const filter = args.statusFilter ?? "all"; + if (target.kind === "all") { + return { + action: "stages", + runId: "--all", + filter, + stages: [], + error: "Stage listing requires a single run.", + }; + } + if (target.kind === "ambiguous") { + return { + action: "stages", + runId: target.target, + filter, + stages: [], + error: ambiguousRunMessage(target.target, target.matches), + }; + } + if (target.kind === "not_found") { + return { + action: "stages", + runId: target.target, + filter, + stages: [], + error: target.message, + }; + } + const run = store.runs().find((r) => r.id === target.runId); + const stages = (run?.stages ?? []) + .filter((stage) => filter === "all" || stage.status === filter) + .map(summarizeStage); + return { action: "stages", runId: target.runId, filter, stages }; + } + + case "stage": { + const target = resolveToolRunTarget(args, "No active run to inspect."); + if (target.kind === "all") { + return { + action: "stage", + runId: "--all", + error: "Stage inspection requires a single run.", + }; + } + if (target.kind === "ambiguous") { + return { + action: "stage", + runId: target.target, + error: ambiguousRunMessage(target.target, target.matches), + }; + } + if (target.kind === "not_found") { + return { + action: "stage", + runId: target.target, + error: target.message, + }; + } + const stage = resolveToolStageTarget(target.runId, args.stageId); + if (!stage.ok || stage.stageId === undefined) { + return { + action: "stage", + runId: target.runId, + error: stage.ok + ? "Stage id, prefix, or name is required." + : stage.message, + }; + } + const run = store.runs().find((r) => r.id === target.runId); + const snapshot = run?.stages.find((s) => s.id === stage.stageId); + return snapshot + ? { action: "stage", runId: target.runId, stage: cloneStage(snapshot) } + : { + action: "stage", + runId: target.runId, + error: `Stage not found in run ${target.runId.slice(0, 8)}: ${stage.stageId}`, + }; + } + + case "transcript": { + const target = resolveToolRunTarget(args, "No active run to inspect."); + if (target.kind === "all") { + return { + action: "transcript", + runId: "--all", + stageId: "", + source: "error", + entries: [], + truncated: false, + }; + } + if (target.kind === "ambiguous") { + return { + action: "transcript", + runId: target.target, + stageId: "", + source: "error", + entries: [ + { role: "notice", text: ambiguousRunMessage(target.target, target.matches) }, + ], + truncated: false, + }; + } + if (target.kind === "not_found") { + return { + action: "transcript", + runId: target.target, + stageId: "", + source: "error", + entries: [{ role: "notice", text: target.message }], + truncated: false, + }; + } + const stage = resolveToolStageTarget(target.runId, args.stageId); + if (!stage.ok || stage.stageId === undefined) { + return { + action: "transcript", + runId: target.runId, + stageId: "", + source: "error", + entries: [ + { + role: "notice", + text: stage.ok + ? "Stage id, prefix, or name is required." + : stage.message, + }, + ], + truncated: false, + }; + } + const run = store.runs().find((r) => r.id === target.runId); + const snapshot = run?.stages.find((s) => s.id === stage.stageId); + const liveHandle = stageControlRegistry.get(target.runId, stage.stageId); + if (liveHandle !== undefined) { + const limited = applyEntryLimit( + liveHandle.messages.map((m) => transcriptEntryFromMessage(m as MessageLike)), + args, + ); + return { + action: "transcript", + runId: target.runId, + stageId: stage.stageId, + source: "live", + entries: limited.entries, + truncated: limited.truncated, + sessionId: liveHandle.sessionId, + sessionFile: liveHandle.sessionFile, + }; + } + const fallback = snapshotTranscriptEntries(snapshot, args.includeToolOutput === true); + const limited = applyEntryLimit(fallback, args); + return { + action: "transcript", + runId: target.runId, + stageId: stage.stageId, + source: "snapshot", + entries: limited.entries, + truncated: limited.truncated, + sessionId: snapshot?.sessionId, + sessionFile: snapshot?.sessionFile, + }; + } + + case "send": { + const target = resolveToolRunTarget(args, "No active run to message."); + const requestedDelivery = args.delivery ?? "auto"; + if (target.kind === "all") { + return workflowSendResult("--all", "", requestedDelivery, "noop", "Send requires a single run."); + } + if (target.kind === "ambiguous") { + return workflowSendResult(target.target, "", requestedDelivery, "noop", ambiguousRunMessage(target.target, target.matches)); + } + if (target.kind === "not_found") { + return workflowSendResult(target.target, "", requestedDelivery, "noop", target.message); + } + const stage = resolveToolStageTarget(target.runId, args.stageId); + if (!stage.ok || stage.stageId === undefined) { + return workflowSendResult( + target.runId, + "", + requestedDelivery, + "noop", + stage.ok ? "Stage id, prefix, or name is required." : stage.message, + ); + } + const run = store.runs().find((r) => r.id === target.runId); + const snapshot = run?.stages.find((s) => s.id === stage.stageId); + const targetsPrompt = + requestedDelivery === "answer" || + args.promptId !== undefined || + (requestedDelivery === "auto" && snapshot?.pendingPrompt !== undefined); + if (targetsPrompt) { + const promptId = args.promptId ?? snapshot?.pendingPrompt?.id; + if (promptId === undefined) { + return workflowSendResult(target.runId, stage.stageId, "answer", "noop", "No pending prompt to answer."); + } + if (!hasPayloadProperty(args)) { + return workflowSendResult(target.runId, stage.stageId, "answer", "noop", "Send requires text, response, or message."); + } + const ok = store.resolveStagePendingPrompt(target.runId, stage.stageId, promptId, promptPayloadFromArgs(args)); + return workflowSendResult( + target.runId, + stage.stageId, + "answer", + ok ? "ok" : "noop", + ok ? `Answered prompt ${promptId}.` : `No matching pending prompt ${promptId}.`, + ); + } + const text = textPayloadFromArgs(args); + if (text === undefined) { + return workflowSendResult(target.runId, stage.stageId, requestedDelivery, "noop", "Send requires text, response, or message."); + } + const handle = stageControlRegistry.get(target.runId, stage.stageId); + if (handle === undefined) { + return workflowSendResult(target.runId, stage.stageId, requestedDelivery, "noop", "No live handle for stage."); + } + if (requestedDelivery === "resume" || (requestedDelivery === "auto" && handle.status === "paused")) { + await handle.resume(text); + return workflowSendResult(target.runId, stage.stageId, "resume", "ok", "Resumed stage with message."); + } + if (requestedDelivery === "steer" || (requestedDelivery === "auto" && handle.isStreaming)) { + await handle.steer(text); + return workflowSendResult(target.runId, stage.stageId, "steer", "ok", "Steered live stage."); + } + if (requestedDelivery === "prompt") { + await handle.prompt(text); + return workflowSendResult(target.runId, stage.stageId, "prompt", "ok", "Prompt sent to stage."); + } + await handle.followUp(text); + return workflowSendResult(target.runId, stage.stageId, "followUp", "ok", "Follow-up queued for stage."); + } + + case "pause": { + const target = resolveToolRunTarget(args, "No in-flight runs to pause."); + if (target.kind === "all") { + if (args.stageId !== undefined && args.stageId.length > 0) { + return { + action, + runId: "--all", + status: "noop", + message: allStageConflictMessage("pause"), + }; + } + const results = pauseAllRuns(); + const paused = results.filter((r) => r.ok).length; + return { + action, + runId: "--all", + status: paused > 0 ? "paused" : "noop", + message: paused > 0 + ? `Paused ${paused} run(s).` + : "No in-flight runs to pause.", + }; + } + if (target.kind === "ambiguous") return { action, runId: target.target, status: "noop", message: ambiguousRunMessage(target.target, target.matches) }; + if (target.kind === "not_found") return { action, runId: target.target, status: "noop", message: target.message }; + const stage = resolveToolStageTarget(target.runId, args.stageId); + if (!stage.ok) return { action, runId: target.runId, status: "noop", message: stage.message }; + const result = pauseRun(target.runId, { stageId: stage.stageId }); + return result.ok + ? { action, runId: result.runId, status: "paused", message: `Paused ${result.paused.length} stage(s) on run ${result.runId.slice(0, 8)}.` } + : { + action, + runId: target.runId, + status: "noop", + message: stageFailureMessage(target.runId, result.reason, "pause"), + }; + } + + case "reload": { + // Fast UX check; reloadWorkflowResourcesNow re-checks inside the + // serialized reload queue and remains the authoritative TOCTOU guard. + const activeRuns = inFlightRunCount(); + if (activeRuns > 0) { + return { + action: "reload", + status: "noop", + message: reloadBlockedMessage(activeRuns), + }; + } + try { + await reloadWorkflowResources(); + } catch (error) { + return { + action: "reload", + status: "noop", + message: reloadFailureMessage(error), + }; + } + return { + action: "reload", + status: "ok", + message: args.reason?.trim() + ? `Reloaded workflow resources (${args.reason.trim()}).` + : "Reloaded workflow resources.", }; } case "kill": { const target = resolveToolRunTarget(args, "No in-flight runs to kill."); if (target.kind === "all") { + if (args.stageId !== undefined && args.stageId.length > 0) { + return { + action, + runId: "--all", + status: "noop", + message: allStageConflictMessage("kill"), + }; + } const results = destroyAllRuns({ cancellation: cancellationRegistry, persistence: getPersistence(), @@ -669,6 +1362,14 @@ export function makeExecuteWorkflowTool( // Interrupt is resumable: it pauses live work and keeps runs in history/status. const target = resolveToolRunTarget(args, "No in-flight runs to interrupt."); if (target.kind === "all") { + if (args.stageId !== undefined && args.stageId.length > 0) { + return { + action, + runId: "--all", + status: "noop", + message: allStageConflictMessage("interrupt"), + }; + } const results = interruptAllRuns(); const interrupted = results.filter((r) => r.ok).length; return { @@ -687,27 +1388,26 @@ export function makeExecuteWorkflowTool( if (target.kind === "not_found") { return { action, runId: target.target, status: "noop", message: target.message }; } - const result = interruptRun(target.runId); + const stage = resolveToolStageTarget(target.runId, args.stageId); + if (!stage.ok) { + return { action, runId: target.runId, status: "noop", message: stage.message }; + } + const result = interruptRun(target.runId, { stageId: stage.stageId }); if (result.ok) { return { action, runId: result.runId, status: "paused", - message: `Run ${result.runId} interrupted and can be resumed.`, + message: stage.stageId + ? `Stage ${stage.stageId} interrupted on run ${result.runId} and can be resumed.` + : `Run ${result.runId} interrupted and can be resumed.`, }; } return { action, runId: target.runId, status: "noop", - message: - result.reason === "not_found" - ? `Run not found: ${target.runId}` - : result.reason === "already_ended" - ? `Run already ended: ${target.runId}` - : result.reason === "stage_not_found" - ? `Stage not found for run: ${target.runId}` - : `No active stages to interrupt for run: ${target.runId}`, + message: stageFailureMessage(target.runId, result.reason, "interrupt"), }; } @@ -1345,10 +2045,84 @@ function factory(pi: ExtensionAPI): void { } let intercomControlUnsubscribe: (() => void) | null = null; + let workflowReloadQueue: Promise = Promise.resolve(); + + async function reloadWorkflowResources(options?: { allowInFlight?: boolean }): Promise { + const reload = workflowReloadQueue.then(() => reloadWorkflowResourcesNow(options)); + workflowReloadQueue = reload.catch(() => {}); + await reload; + } + + async function reloadWorkflowResourcesNow(options?: { allowInFlight?: boolean }): Promise { + const activeRuns = inFlightRunCount(); + if (options?.allowInFlight !== true) { + if (activeRuns > 0) { + throw new WorkflowReloadBlockedError(reloadBlockedMessage(activeRuns)); + } + } else if (activeRuns > 0 && process.env.ATOMIC_WORKFLOW_DEBUG === "1") { + console.warn( + `Workflow reload bypassed in-flight guard with ${activeRuns} active run(s).`, + ); + } + + const configResult = await loadWorkflowConfig(); + configLoadRef.current = configResult; + + // Build scope-aware DiscoveryConfig: global entries → globalWorkflows (resolved + // under /.atomic/agent), project entries → projectWorkflows (resolved under + // projectRoot). Project keys override global keys. Paths pre-resolved to absolute. + const { homedir } = await import("node:os"); + const hasGlobal = configResult.globalConfig != null; + const hasProject = configResult.projectConfig != null; + const discoveryConfig = + hasGlobal || hasProject + ? toScopedDiscoveryConfig( + configResult.globalConfig ?? null, + configResult.projectConfig ?? null, + { projectRoot: process.cwd(), homeDir: homedir() }, + ) + : undefined; + + const packageWorkflowPaths = (pi.getWorkflowResources?.() ?? []) + .filter((resource) => resource.enabled !== false) + .map((resource) => resource.path); + const result = await discoverWorkflows({ config: discoveryConfig, packageWorkflowPaths }); + discoveryRef.current = result; + + // Resolve effective config (fills in all defaults) and build WorkflowRuntimeConfig. + const effectiveConfig = withWorkflowDefaults(configResult.config ?? {}); + runtimeConfigRef.current = { + maxDepth: effectiveConfig.maxDepth, + defaultConcurrency: effectiveConfig.defaultConcurrency, + persistRuns: effectiveConfig.persistRuns, + statusFile: effectiveConfig.statusFile, + resumeInFlight: effectiveConfig.resumeInFlight, + }; + + // Replace status writer with one that reflects the resolved config. + // Unsubscribe the prior (no-op) writer before creating the new one. + statusWriterRef.unsubscribe(); + statusWriterRef = createStatusWriter(store, runtimeConfigRef.current); + + persistenceRef.current = makePersistencePort( + pi, + effectiveConfig.persistRuns, + ); + runtimeRef.current = createExtensionRuntime({ + registry: result.registry, + adapters, + cancellation: cancellationRegistry, + persistence: persistenceRef.current, + mcp: mcpPort, + intercom: intercomPort, + config: runtimeConfigRef.current, + }); + } const executeWorkflowTool = makeExecuteWorkflowTool( (ctx) => runtimeForContext(ctx), () => persistenceRef.current, + reloadWorkflowResources, ); let storeWidgetUnsubscribe: (() => void) | null = null; @@ -1358,59 +2132,7 @@ function factory(pi: ExtensionAPI): void { // Load startup config before discovery so workflow paths and tunables are applied. const discoveryPromise = pi.disableAsyncDiscovery ? Promise.resolve() - : loadWorkflowConfig().then(async (configResult) => { - configLoadRef.current = configResult; - - // Build scope-aware DiscoveryConfig: global entries → globalWorkflows (resolved - // under /.atomic/agent), project entries → projectWorkflows (resolved under - // projectRoot). Project keys override global keys. Paths pre-resolved to absolute. - const { homedir } = await import("node:os"); - const hasGlobal = configResult.globalConfig != null; - const hasProject = configResult.projectConfig != null; - const discoveryConfig = - hasGlobal || hasProject - ? toScopedDiscoveryConfig( - configResult.globalConfig ?? null, - configResult.projectConfig ?? null, - { projectRoot: process.cwd(), homeDir: homedir() }, - ) - : undefined; - - const packageWorkflowPaths = (pi.getWorkflowResources?.() ?? []) - .filter((resource) => resource.enabled !== false) - .map((resource) => resource.path); - const result = await discoverWorkflows({ config: discoveryConfig, packageWorkflowPaths }); - discoveryRef.current = result; - - // Resolve effective config (fills in all defaults) and build WorkflowRuntimeConfig. - const effectiveConfig = withWorkflowDefaults(configResult.config ?? {}); - runtimeConfigRef.current = { - maxDepth: effectiveConfig.maxDepth, - defaultConcurrency: effectiveConfig.defaultConcurrency, - persistRuns: effectiveConfig.persistRuns, - statusFile: effectiveConfig.statusFile, - resumeInFlight: effectiveConfig.resumeInFlight, - }; - - // Replace status writer with one that reflects the resolved config. - // Unsubscribe the prior (no-op) writer before creating the new one. - statusWriterRef.unsubscribe(); - statusWriterRef = createStatusWriter(store, runtimeConfigRef.current); - - persistenceRef.current = makePersistencePort( - pi, - effectiveConfig.persistRuns, - ); - runtimeRef.current = createExtensionRuntime({ - registry: result.registry, - adapters, - cancellation: cancellationRegistry, - persistence: persistenceRef.current, - mcp: mcpPort, - intercom: intercomPort, - config: runtimeConfigRef.current, - }); - }); + : reloadWorkflowResources({ allowInFlight: true }); // ------------------------------------------------------------------------- // 1. Register the `workflow` tool @@ -1432,7 +2154,7 @@ function factory(pi: ExtensionAPI): void { // tool-call dispatch path. const details = await executeWorkflowTool(params, ctx); return { - content: [{ type: "text", text: renderResult(details, {}) }], + content: [{ type: "text", text: renderWorkflowToolContent(details, params) }], details, }; }, @@ -1931,7 +2653,7 @@ function factory(pi: ExtensionAPI): void { "workflow", { description: - "Run or inspect pi workflows. Usage: /workflow [key=value…] | /workflow [list|status|connect|attach|interrupt|kill|pause|resume|inputs] [args]", + "Run or inspect pi workflows. Usage: /workflow [key=value…] | /workflow [list|status|connect|attach|interrupt|kill|pause|resume|inputs|reload] [args]", handler: async (args: string, ctx: PiCommandContext) => { const print = (msg: string): void => ctx.ui.notify(msg, "info"); // Quote-aware split so `prompt="map the codebase"` stays a single @@ -2019,6 +2741,26 @@ function factory(pi: ExtensionAPI): void { return; } + // ----------------------------------------------------------------------- + // reload — refresh workflow resources in-process when no workflows are + // currently running. Reload swaps runtime/persistence wiring, so doing it + // mid-flight would split active runs across old and new resources. + // ----------------------------------------------------------------------- + if (subcommand === "reload") { + const activeRuns = inFlightRunCount(); + if (activeRuns > 0) { + print(reloadBlockedMessage(activeRuns)); + return; + } + try { + await reloadWorkflowResources(); + print("Reloaded workflow resources."); + } catch (error) { + print(reloadFailureMessage(error)); + } + return; + } + // ----------------------------------------------------------------------- // interrupt — top-level chat fast path (no confirmation overlay). // ----------------------------------------------------------------------- @@ -2344,6 +3086,11 @@ function factory(pi: ExtensionAPI): void { label: "inputs", description: "Show a workflow's input schema", }, + { + value: "reload ", + label: "reload", + description: "Reload workflow resources", + }, ]; const parts = partial.trim().split(/\s+/).filter(Boolean); diff --git a/packages/workflows/src/extension/render-call.ts b/packages/workflows/src/extension/render-call.ts index ed2fdbbb47..d0c91e33b9 100644 --- a/packages/workflows/src/extension/render-call.ts +++ b/packages/workflows/src/extension/render-call.ts @@ -5,10 +5,25 @@ import { truncateToWidth } from "../tui/text-helpers.js"; +/** Renderer-only subset of the canonical WorkflowToolArgs from index.ts. */ export interface WorkflowToolArgs { workflow?: string; inputs?: Record; - action?: "run" | "list" | "get" | "status" | "interrupt" | "kill" | "resume" | "inputs"; + action?: + | "run" + | "list" + | "get" + | "status" + | "stages" + | "stage" + | "transcript" + | "send" + | "pause" + | "interrupt" + | "kill" + | "resume" + | "reload" + | "inputs"; runId?: string; task?: { name?: string; prompt?: string; task?: string } | string; tasks?: readonly unknown[]; @@ -62,6 +77,24 @@ export function renderCall(args: WorkflowToolArgs, opts: RenderCallOpts = {}): s case "run": line = name === undefined ? "workflow: run" : `workflow: run ${quoted(name)}`; break; + case "stages": + line = name === undefined ? "workflow: list stages" : `workflow: list stages for ${quoted(name)}`; + break; + case "stage": + line = name === undefined ? "workflow: inspect stage" : `workflow: inspect stage in ${quoted(name)}`; + break; + case "transcript": + line = name === undefined ? "workflow: read stage transcript" : `workflow: read stage transcript in ${quoted(name)}`; + break; + case "send": + line = name === undefined ? "workflow: send to stage" : `workflow: send to stage in ${quoted(name)}`; + break; + case "pause": + line = name === undefined ? "workflow: pause run" : `workflow: pause run ${quoted(name)}`; + break; + case "reload": + line = "workflow: reload runtime"; + break; case "interrupt": line = name === undefined ? "workflow: interrupt run" diff --git a/packages/workflows/src/extension/render-result.ts b/packages/workflows/src/extension/render-result.ts index 6c2ceaf0d6..c100f2016a 100644 --- a/packages/workflows/src/extension/render-result.ts +++ b/packages/workflows/src/extension/render-result.ts @@ -11,7 +11,7 @@ * - pi-subagents src/extension/index.ts renderResult slot */ -import type { RunSnapshot, StageSnapshot } from "../shared/store-types.js"; +import type { PendingPrompt, RunSnapshot, StageSnapshot, StageStatus } from "../shared/store-types.js"; import type { WorkflowDetails } from "../shared/types.js"; import type { RunDetail } from "../runs/background/status.js"; import { renderInputsSchema } from "../shared/render-inputs-schema.js"; @@ -98,6 +98,32 @@ type RunResult = { */ message?: string; }; +type StageListItem = { + id: string; + name: string; + status: StageStatus; + sessionId?: string; + sessionFile?: string; + error?: string; + awaitingInputSince?: number; + pendingPrompt?: PendingPrompt; +}; +type StageListResult = { action: "stages"; runId: string; filter: string; stages: StageListItem[]; error?: string }; +type StageDetailResult = { action: "stage"; runId: string; stage?: StageSnapshot; error?: string }; +type TranscriptEntry = { role: string; text?: string; toolName?: string; output?: string; timestamp?: number }; +type TranscriptResult = { + action: "transcript"; + runId: string; + stageId: string; + source: "live" | "snapshot" | "error"; + entries: TranscriptEntry[]; + truncated: boolean; + sessionId?: string; + sessionFile?: string; +}; +type SendResult = { action: "send"; runId: string; stageId: string; delivery: string; status: "ok" | "noop"; message: string }; +type PauseResult = { action: "pause"; runId: string; status: string; message: string }; +type ReloadResult = { action: "reload"; status: "ok" | "noop"; message: string }; type InterruptResult = { action: "interrupt"; runId: string; status: string; message: string }; type KillResult = { action: "kill"; runId: string; status: string; message: string }; type ResumeResult = { action: "resume"; runId: string; status: string; message: string }; @@ -109,6 +135,12 @@ export type WorkflowToolResult = | InputsResult | GetResult | RunResult + | StageListResult + | StageDetailResult + | TranscriptResult + | SendResult + | PauseResult + | ReloadResult | InterruptResult | KillResult | ResumeResult; @@ -162,6 +194,21 @@ function renderNotice( }); } +const TRANSCRIPT_NOTICE_ENTRY_LIMIT = 5; +const TRANSCRIPT_NOTICE_CHAR_LIMIT = 240; + +function transcriptNoticeText(entries: readonly TranscriptEntry[]): string { + if (entries.length === 0) return "no transcript entries"; + const shown = entries.slice(0, TRANSCRIPT_NOTICE_ENTRY_LIMIT); + const text = shown + .map((entry) => `${entry.role}: ${entry.text ?? entry.output ?? entry.toolName ?? "(no body)"}`) + .join(" | "); + const entrySuffix = entries.length > shown.length + ? ` … (+${entries.length - shown.length} more)` + : ""; + return fitLine(`${text}${entrySuffix}`, TRANSCRIPT_NOTICE_CHAR_LIMIT); +} + export function renderResult(result: WorkflowToolResult, opts?: RenderResultOpts): string { const partial = opts?.isPartial === true; const themed = opts?.plain !== true; @@ -256,6 +303,42 @@ export function renderResult(result: WorkflowToolResult, opts?: RenderResultOpts ); } + case "stages": { + const r = result as StageListResult; + if (r.error) return renderNotice("WORKFLOW STAGES", `${r.runId || "(none)"}: ${r.error}`, opts, themed); + const counts = r.stages.map((s) => `${s.name} (${s.id.slice(0, 12)}): ${s.status}`).join("; "); + return renderNotice("WORKFLOW STAGES", `${r.runId}: ${r.filter} — ${counts || "no stages"}`, opts, themed); + } + + case "stage": { + const r = result as StageDetailResult; + if (r.error || !r.stage) return renderNotice("WORKFLOW STAGE", `${r.runId}: ${r.error ?? "stage not found"}`, opts, themed); + const extra = r.stage.error ? ` — ${r.stage.error}` : r.stage.result ? ` — ${r.stage.result}` : ""; + return renderNotice("WORKFLOW STAGE", `${r.runId}: ${r.stage.name} (${r.stage.id.slice(0, 12)}) ${r.stage.status}${extra}`, opts, themed); + } + + case "transcript": { + const r = result as TranscriptResult; + const text = transcriptNoticeText(r.entries); + const suffix = r.truncated ? " (truncated)" : ""; + return renderNotice("WORKFLOW TRANSCRIPT", `${r.runId}/${r.stageId.slice(0, 12)} ${r.source}: ${text}${suffix}`, opts, themed); + } + + case "send": { + const r = result as SendResult; + return renderNotice("WORKFLOW SEND", `${r.runId}/${r.stageId.slice(0, 12)} ${r.delivery}: ${r.message}`, opts, themed); + } + + case "pause": { + const r = result as PauseResult; + return renderNotice("WORKFLOW PAUSE", `${r.runId}: ${r.message}`, opts, themed); + } + + case "reload": { + const r = result as ReloadResult; + return renderNotice("WORKFLOW RELOAD", r.message, opts, themed); + } + case "interrupt": { const r = result as InterruptResult; return renderNotice("WORKFLOW INTERRUPT", `${r.runId}: ${r.message}`, opts, themed); diff --git a/packages/workflows/src/extension/workflow-schema.ts b/packages/workflows/src/extension/workflow-schema.ts index 7fbb427a5c..d0648fb207 100644 --- a/packages/workflows/src/extension/workflow-schema.ts +++ b/packages/workflows/src/extension/workflow-schema.ts @@ -99,21 +99,78 @@ export const WorkflowParametersSchema = Type.Object({ Type.Literal("get"), Type.Literal("inputs"), Type.Literal("status"), + Type.Literal("stages"), + Type.Literal("stage"), + Type.Literal("transcript"), + Type.Literal("send"), + Type.Literal("pause"), Type.Literal("interrupt"), Type.Literal("kill"), Type.Literal("resume"), - ])), + Type.Literal("reload"), + ], { + description: "Workflow action: run/list/get/inputs/status, inspect stages/transcripts, send messages or prompt answers, pause/resume/interrupt/kill runs, or reload workflow resources.", + })), runId: Type.Optional(Type.String({ - description: "Run identifier or unique prefix for status/interrupt/kill/resume. Use '--all' for interrupt/kill all.", + description: "Run identifier or unique prefix for status/stages/stage/transcript/send/pause/resume/interrupt/kill. Use '--all' or all:true for supported bulk run-control actions.", })), all: Type.Optional(Type.Boolean({ - description: "Apply supported run-control actions (interrupt/kill) to all in-flight runs.", + description: "Apply supported run-control actions (pause/interrupt/kill) to all in-flight runs instead of one run; cannot be combined with stageId.", })), stageId: Type.Optional(Type.String({ - description: "Stage id, unique prefix, or stage name for stage-scoped resume.", + description: "Stage id, unique prefix, or stage name for stage-scoped inspection, transcript, send, pause, or resume.", })), message: Type.Optional(Type.String({ - description: "Optional message forwarded when resuming paused work.", + description: "Message payload for send/follow-up/prompt/steer/resume, or optional text forwarded when resuming paused work.", + })), + statusFilter: Type.Optional(Type.Union([ + Type.Literal("pending"), + Type.Literal("running"), + Type.Literal("awaiting_input"), + Type.Literal("paused"), + Type.Literal("blocked"), + Type.Literal("completed"), + Type.Literal("failed"), + Type.Literal("skipped"), + Type.Literal("all"), + ], { + description: "Filter stages by status for the stages action; use 'all' to include every stage.", + })), + format: Type.Optional(Type.Union([Type.Literal("text"), Type.Literal("json")], { + description: "Agent-visible output format for data-bearing inspection actions.", + })), + limit: Type.Optional(Type.Integer({ + minimum: 0, + description: "Transcript-only: maximum number of most recent transcript entries to return; applied before tool output is serialized.", + })), + tail: Type.Optional(Type.Integer({ + minimum: 0, + description: "Transcript-only: return only the last N transcript entries; overrides limit when both are provided.", + })), + includeToolOutput: Type.Optional(Type.Boolean({ + description: "Transcript-only: include captured tool output entries when building results from stage snapshots; live session transcripts may not expose tool output.", + })), + text: Type.Optional(Type.String({ + description: "Text to send to a stage for prompt answers, steering, follow-ups, or resume messages.", + })), + response: Type.Optional(Type.Unknown({ + description: "Structured response payload for answering a pending stage prompt.", + })), + delivery: Type.Optional(Type.Union([ + Type.Literal("auto"), + Type.Literal("answer"), + Type.Literal("prompt"), + Type.Literal("steer"), + Type.Literal("followUp"), + Type.Literal("resume"), + ], { + description: "Delivery mode for the send action: auto answers pending prompts first, then resumes paused stages, steers streaming stages, or queues a follow-up.", + })), + promptId: Type.Optional(Type.String({ + description: "Pending prompt identifier to answer when using the send action.", + })), + reason: Type.Optional(Type.String({ + description: "Human-readable reason for the reload action, echoed in the reload result.", })), task: Type.Optional(Type.Union([ DirectTaskSchema, diff --git a/packages/workflows/src/runs/background/status.ts b/packages/workflows/src/runs/background/status.ts index d30cd8a8c0..371db8b554 100644 --- a/packages/workflows/src/runs/background/status.ts +++ b/packages/workflows/src/runs/background/status.ts @@ -294,8 +294,8 @@ export function resumeRun( } // Return a deep copy of the snapshot for safe consumption - const snapshot: RunSnapshot = JSON.parse(JSON.stringify(run)) as RunSnapshot; - const resumedCopy: StageSnapshot[] = JSON.parse(JSON.stringify(resumed)) as StageSnapshot[]; + const snapshot = structuredClone(run); + const resumedCopy = structuredClone(resumed); if (run.status === "failed" && run.endedAt !== undefined && run.resumable === false) { return { ok: true, @@ -366,7 +366,7 @@ export function pauseRun( } void handle.pause(); const stageSnap = run.stages.find((s) => s.id === opts.stageId); - const paused: StageSnapshot[] = stageSnap ? [JSON.parse(JSON.stringify(stageSnap)) as StageSnapshot] : []; + const paused: StageSnapshot[] = stageSnap ? [structuredClone(stageSnap)] : []; // Only mark the whole run paused when every active stage is paused. const stillActive = run.stages.some( (s) => s.status === "running" && s.id !== opts.stageId, @@ -385,12 +385,23 @@ export function pauseRun( for (const handle of handles) { void handle.pause(); const stageSnap = run.stages.find((s) => s.id === handle.stageId); - if (stageSnap) pausedSnaps.push(JSON.parse(JSON.stringify(stageSnap)) as StageSnapshot); + if (stageSnap) pausedSnaps.push(structuredClone(stageSnap)); } activeStore.recordRunPaused(runId); return { ok: true, runId, paused: pausedSnaps }; } +export function pauseAllRuns(opts?: { + store?: Store; + stageControlRegistry?: StageControlRegistry; +}): PauseResult[] { + const activeStore = opts?.store ?? defaultStore; + const inFlight = activeStore.runs().filter((r) => r.endedAt === undefined); + return inFlight.map((r) => + pauseRun(r.id, { store: activeStore, stageControlRegistry: opts?.stageControlRegistry }), + ); +} + // --------------------------------------------------------------------------- // interruptRun // --------------------------------------------------------------------------- @@ -449,7 +460,7 @@ export function inspectRun( } // Deep copy so callers cannot mutate the store via the snapshot. - const copy = JSON.parse(JSON.stringify(candidate)) as RunSnapshot; + const copy = structuredClone(candidate); const detail: RunDetail = { runId: copy.id, diff --git a/test/integration/mcp-entrypoint.test.ts b/test/integration/mcp-entrypoint.test.ts index 2b0d0554f8..e61ecc99de 100644 --- a/test/integration/mcp-entrypoint.test.ts +++ b/test/integration/mcp-entrypoint.test.ts @@ -160,7 +160,7 @@ describe("MCP entrypoints — workflow tool execute", () => { const { pi, emits: e } = makeMockPiWithEvents(); emits = e; const runtime = buildTestRuntime(pi); - toolExecute = makeExecuteWorkflowTool(runtime, () => undefined); + toolExecute = makeExecuteWorkflowTool(runtime, () => undefined, () => undefined); }); test("tool execute emits mcp.scope.set (set then clear) when running mcp-restricted workflow", async () => { @@ -199,7 +199,7 @@ describe("MCP entrypoints — workflow tool execute", () => { adapters: noopAdapters, mcp: makeMcpPort(piNoEvents), }); - const execute = makeExecuteWorkflowTool(runtime, () => undefined); + const execute = makeExecuteWorkflowTool(runtime, () => undefined, () => undefined); // Should not throw, should complete const result = await execute({ action: "run", workflow: "mcp-restricted", inputs: {} }, {}); assert.equal((result as { action: string }).action, "run"); diff --git a/test/integration/mock-extension-api.test.ts b/test/integration/mock-extension-api.test.ts index a46cf40d1b..e65b2b7862 100644 --- a/test/integration/mock-extension-api.test.ts +++ b/test/integration/mock-extension-api.test.ts @@ -371,7 +371,7 @@ describe("MockExtensionAPI — tool registration", () => { }, }, }); - const executeWorkflowTool = makeExecuteWorkflowTool(runtime, () => undefined); + const executeWorkflowTool = makeExecuteWorkflowTool(runtime, () => undefined, () => undefined); const started = await executeWorkflowTool({ task: { name: "blocking-scout", task: "wait for interrupt" }, @@ -870,6 +870,17 @@ describe("renderResult — all action branches", () => { }, }, { action: "run", runId: "run-abcdef", status: "running", message: "A very long background dispatch message." }, + { + action: "transcript", + runId: "run-abcdef", + stageId: "stage-abcdef", + source: "snapshot", + entries: Array.from({ length: 10 }, (_, index) => ({ + role: "assistant", + text: `A very long transcript entry ${index} ${"x".repeat(80)}`, + })), + truncated: false, + }, { action: "interrupt", runId: "run-abcdef", status: "paused", message: "A very long interrupt response message." }, { action: "kill", runId: "run-abcdef", status: "killed", message: "A very long kill response message." }, { action: "resume", runId: "run-abcdef", status: "ok", message: "A very long resume response message." }, diff --git a/test/unit/slash-dispatch.test.ts b/test/unit/slash-dispatch.test.ts index 24d77a5af9..712b3d08dd 100644 --- a/test/unit/slash-dispatch.test.ts +++ b/test/unit/slash-dispatch.test.ts @@ -20,11 +20,15 @@ import { tokenizeWorkflowArgs, makeExecuteWorkflowTool, } from "../../packages/workflows/src/extension/index.js"; +import { renderResult } from "../../packages/workflows/src/extension/render-result.js"; +import type { WorkflowToolResult } from "../../packages/workflows/src/extension/render-result.js"; import type { ExtensionAPI, PiArgumentCompletion, PiCommandContext, PiCommandOptions, + PiToolOpts, + WorkflowToolArgs, } from "../../packages/workflows/src/extension/index.js"; import { createRegistry } from "../../packages/workflows/src/workflows/registry.js"; import { defineWorkflow } from "../../packages/workflows/src/workflows/define-workflow.js"; @@ -43,8 +47,13 @@ import { killAllRuns } from "../../packages/workflows/src/runs/background/status import { cancellationRegistry } from "../../packages/workflows/src/runs/background/cancellation-registry.js"; import { jobTracker } from "../../packages/workflows/src/runs/background/job-tracker.js"; import type { StageSessionRuntime } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; +import { + stageControlRegistry, + type StageControlHandle, +} from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; afterEach(async () => { + stageControlRegistry.clear(); killAllRuns({ store, cancellation: cancellationRegistry }); await Promise.all(jobTracker.runIds().map((runId) => jobTracker.get(runId)?.promise)); store.clear(); @@ -413,6 +422,7 @@ describe("getArgumentCompletions includes workflow names", () => { assert.ok(labels.includes("kill")); assert.ok(labels.includes("resume")); assert.ok(labels.includes("inputs")); + assert.ok(labels.includes("reload")); assert.equal(labels.includes("session"), false); assert.ok(labels.includes("deep-research-codebase")); @@ -788,6 +798,43 @@ describe("/workflow interrupt chat command", () => { assert.equal(run?.status, "running"); assert.equal(msgs.some((m) => m.includes("No active stages to interrupt")), true); }); + + test("top-level /workflow reload is skipped while workflows are in flight", async () => { + const runId = `reload-slash-blocked-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + + const { pi, commands } = buildMockPi(); + addFactoryStubs(pi); + + const factoryModule = await import("../../packages/workflows/src/extension/index.js"); + factoryModule.default(pi); + + const workflowCmd = commands.find((c) => c.name === "workflow")!; + const { ctx, messages } = buildCtx(); + + await workflowCmd.options.handler("reload", ctx); + + assert.equal(messages.some((message) => message.includes("still in flight")), true); + assert.equal(messages.some((message) => message.includes("Reloaded workflow resources")), false); + }); + + test("top-level /workflow reload reports reload failures", async () => { + const { pi, commands } = buildMockPi(); + addFactoryStubs(pi); + pi.getWorkflowResources = () => { + throw new Error("package loader unavailable"); + }; + + const factoryModule = await import("../../packages/workflows/src/extension/index.js"); + factoryModule.default(pi); + + const workflowCmd = commands.find((c) => c.name === "workflow")!; + const { ctx, messages } = buildCtx(); + + await workflowCmd.options.handler("reload", ctx); + + assert.equal(messages.some((message) => message.includes("Reload failed: package loader unavailable")), true); + }); }); // --------------------------------------------------------------------------- @@ -887,9 +934,121 @@ describe("tool run-control actions", () => { function makeToolHandler() { const registry = createRegistry([]); const runtime = createExtensionRuntime({ registry }); - return makeExecuteWorkflowTool(runtime, () => undefined); + return makeExecuteWorkflowTool(runtime, () => undefined, () => undefined); + } + + async function makeRegisteredWorkflowTool(): Promise> { + const { pi } = buildMockPi(); + addFactoryStubs(pi); + let registered: PiToolOpts | undefined; + pi.registerTool = (opts) => { + registered = opts as unknown as PiToolOpts; + }; + const factoryModule = await import("../../packages/workflows/src/extension/index.js"); + factoryModule.default(pi); + assert.ok(registered, "expected workflow tool registration"); + return registered; + } + + function registerLiveStageHandle( + runId: string, + stageId: string, + options?: { + status?: StageControlHandle["status"]; + isStreaming?: boolean; + messages?: StageControlHandle["messages"]; + }, + ): { followUps: string[]; prompts: string[]; steers: string[]; dispose: () => void } { + const followUps: string[] = []; + const prompts: string[] = []; + const steers: string[] = []; + const handle: StageControlHandle = { + runId, + stageId, + stageName: "ask", + status: options?.status ?? "running", + sessionId: undefined, + sessionFile: undefined, + isStreaming: options?.isStreaming ?? false, + messages: options?.messages ?? [], + async ensureAttached(): Promise {}, + async prompt(text: string): Promise { + prompts.push(text); + }, + async steer(text: string): Promise { + steers.push(text); + }, + async followUp(text: string): Promise { + followUps.push(text); + }, + async pause(): Promise {}, + async resume(): Promise {}, + subscribe: () => () => {}, + }; + return { followUps, prompts, steers, dispose: stageControlRegistry.register(handle) }; } + test("registered workflow tool content preserves full transcript text and supports JSON format", async () => { + const runId = `tool-content-transcript-${Date.now()}`; + const longText = `start-${"x".repeat(180)}-sentinel-end`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-tool-content-1", + name: "summarize", + status: "completed", + parentIds: [], + toolEvents: [], + result: longText, + sessionId: "session-tool-content", + sessionFile: "/tmp/tool-content.jsonl", + }); + const tool = await makeRegisteredWorkflowTool(); + + const textResult = await tool.execute( + "tool-content-text", + { action: "transcript", runId, stageId: "summarize" }, + undefined, + undefined, + {} as never, + ); + const textBlock = textResult.content[0]; + assert.equal(textBlock?.type, "text"); + const textContent = textBlock.type === "text" ? textBlock.text : ""; + assert.ok(textContent.includes(longText), "plain tool content should include the full transcript entry"); + assert.equal(textContent.includes("╭"), false, "tool content should not use clipped UI chrome"); + + const jsonResult = await tool.execute( + "tool-content-json", + { action: "transcript", runId, stageId: "summarize", format: "json" }, + undefined, + undefined, + {} as never, + ); + const jsonBlock = jsonResult.content[0]; + assert.equal(jsonBlock?.type, "text"); + const parsed = JSON.parse(jsonBlock.type === "text" ? jsonBlock.text : "{}"); + assert.equal(parsed.entries[0].text, longText); + }); + + test("registered workflow tool content elides empty send targets", async () => { + const tool = await makeRegisteredWorkflowTool(); + + const result = await tool.execute( + "tool-content-send-empty-target", + { action: "send", text: "hello" }, + undefined, + undefined, + {} as never, + ); + + assert.equal(result.details.action, "send"); + const textBlock = result.content[0]; + assert.equal(textBlock?.type, "text"); + const textContent = textBlock.type === "text" ? textBlock.text : ""; + assert.match(textContent, /^send: noop — /); + assert.doesNotMatch(textContent, /^send:\s{2,}noop/); + }); + test("makeExecuteWorkflowTool kill without runId defaults to the active run", async () => { const runId = `kill-tool-active-${Date.now()}`; store.recordRunStart(makeInflightRun(runId)); @@ -938,6 +1097,33 @@ describe("tool run-control actions", () => { assert.equal(store.runs().some((run) => run.id === ended), true); }); + test("makeExecuteWorkflowTool pause all reports noop when no runs are in flight", async () => { + const handler = makeToolHandler(); + + const result = await handler({ action: "pause", all: true }, {} as never); + + assert.equal(result.action, "pause"); + const r = result as { action: string; status: string; runId: string; message: string }; + assert.equal(r.runId, "--all"); + assert.equal(r.status, "noop"); + assert.match(r.message, /No in-flight runs to pause/); + }); + + test("makeExecuteWorkflowTool rejects all run-control with stageId", async () => { + const runId = `pause-tool-all-stage-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + const handler = makeToolHandler(); + + const result = await handler({ action: "pause", all: true, stageId: "stage-a" }, {} as never); + + assert.equal(result.action, "pause"); + const r = result as { action: string; status: string; runId: string; message: string }; + assert.equal(r.runId, "--all"); + assert.equal(r.status, "noop"); + assert.match(r.message, /Cannot pause --all with a stageId/); + assert.equal(store.runs().find((run) => run.id === runId)?.status, "running"); + }); + test("makeExecuteWorkflowTool interrupt without runId defaults to the active run", async () => { const runId = `interrupt-tool-active-${Date.now()}`; store.recordRunStart(makeInflightRun(runId)); @@ -953,6 +1139,660 @@ describe("tool run-control actions", () => { assert.equal(store.runs().find((run) => run.id === runId)?.status, "running"); }); + test("makeExecuteWorkflowTool pause reports pause wording for inactive stages", async () => { + const runId = `pause-tool-inactive-stage-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-paused-1", + name: "paused-stage", + status: "paused", + parentIds: [], + toolEvents: [], + }); + const { dispose } = registerLiveStageHandle(runId, "stage-paused-1", { status: "paused" }); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "pause", runId, stageId: "paused-stage" }, {} as never); + + assert.equal(result.action, "pause"); + const r = result as { action: string; status: string; message: string }; + assert.equal(r.status, "noop"); + assert.match(r.message, /No active stages to pause/); + assert.doesNotMatch(r.message, /interrupt/); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool lists and inspects workflow stages", async () => { + const runId = `stage-tool-list-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-running-1", name: "scan", status: "running", parentIds: [], toolEvents: [] }); + store.recordStageStart(runId, { id: "stage-failed-1", name: "review", status: "failed", parentIds: [], toolEvents: [], error: "boom" }); + const handler = makeToolHandler(); + + const listResult = await handler({ action: "stages", runId, statusFilter: "failed" }, {} as never); + assert.equal(listResult.action, "stages"); + const list = listResult as { action: string; stages: Array<{ name: string; status: string; error?: string }> }; + assert.deepEqual(list.stages.map((stage) => stage.name), ["review"]); + assert.equal(list.stages[0]!.status, "failed"); + + const detailResult = await handler({ action: "stage", runId, stageId: "scan" }, {} as never); + assert.equal(detailResult.action, "stage"); + const detail = detailResult as { action: string; stage?: { id: string; name: string; status: string } }; + assert.equal(detail.stage?.id, "stage-running-1"); + assert.equal(detail.stage?.status, "running"); + }); + + test("makeExecuteWorkflowTool stages clones pending prompts", async () => { + const runId = `stage-tool-prompt-clone-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-clone", name: "ask", status: "awaiting_input", parentIds: [], toolEvents: [] }); + store.recordStagePendingPrompt(runId, "stage-prompt-clone", { id: "prompt-clone", kind: "select", message: "Original?", choices: ["yes"], createdAt: Date.now() }); + const handler = makeToolHandler(); + + const result = await handler({ action: "stages", runId }, {} as never); + + assert.equal(result.action, "stages"); + const stages = result as { action: string; stages: Array<{ pendingPrompt?: { message: string; choices?: string[] } }> }; + assert.equal(stages.stages[0]?.pendingPrompt?.message, "Original?"); + stages.stages[0]!.pendingPrompt!.message = "Mutated"; + stages.stages[0]!.pendingPrompt!.choices!.push("no"); + const storedPrompt = store.runs().find((run) => run.id === runId)?.stages[0]?.pendingPrompt; + assert.equal(storedPrompt?.message, "Original?"); + assert.deepEqual(storedPrompt?.choices, ["yes"]); + }); + + test("makeExecuteWorkflowTool stage rejects all-run inspection", async () => { + const handler = makeToolHandler(); + + const result = await handler({ action: "stage", all: true }, {} as never); + + assert.equal(result.action, "stage"); + const stage = result as { action: string; runId: string; error?: string }; + assert.equal(stage.runId, "--all"); + assert.match(stage.error ?? "", /requires a single run/); + }); + + test("makeExecuteWorkflowTool stages supports all stage status filters", async () => { + const runId = `stage-tool-status-filters-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + for (const status of ["pending", "running", "awaiting_input", "paused", "blocked", "completed", "failed", "skipped"] as const) { + store.recordStageStart(runId, { + id: `stage-${status}`, + name: status, + status, + parentIds: [], + toolEvents: [], + }); + } + const handler = makeToolHandler(); + + const completedResult = await handler({ action: "stages", runId, statusFilter: "completed" }, {} as never); + + assert.equal(completedResult.action, "stages"); + const completed = completedResult as { action: string; stages: Array<{ name: string; status: string }> }; + assert.deepEqual(completed.stages.map(({ name, status }) => ({ name, status })), [ + { name: "completed", status: "completed" }, + ]); + }); + + test("makeExecuteWorkflowTool stages reports missing and ambiguous run targets", async () => { + const handler = makeToolHandler(); + + const missing = await handler({ action: "stages" }, {} as never); + assert.equal(missing.action, "stages"); + const missingStages = missing as { action: string; runId: string; error?: string; stages: unknown[] }; + assert.equal(missingStages.runId, ""); + assert.deepEqual(missingStages.stages, []); + assert.match(missingStages.error ?? "", /No active run to inspect/); + assert.match(renderResult(missing, { plain: true }), /No active run to inspect/); + + store.recordRunStart(makeInflightRun("stages-ambiguous-run-a")); + store.recordRunStart(makeInflightRun("stages-ambiguous-run-b")); + const ambiguous = await handler({ action: "stages", runId: "stages-ambiguous-run" }, {} as never); + assert.equal(ambiguous.action, "stages"); + const ambiguousStages = ambiguous as { action: string; runId: string; error?: string; stages: unknown[] }; + assert.equal(ambiguousStages.runId, "stages-ambiguous-run"); + assert.deepEqual(ambiguousStages.stages, []); + assert.match(ambiguousStages.error ?? "", /Ambiguous run prefix/); + assert.match(renderResult(ambiguous, { plain: true }), /Ambiguous run prefix/); + }); + + test("makeExecuteWorkflowTool returns chronologically final snapshot result after tools", async () => { + const runId = `stage-tool-transcript-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-transcript-1", + name: "summarize", + status: "completed", + parentIds: [], + toolEvents: [{ name: "read", output: "file contents", startedAt: 1, endedAt: 2 }], + result: "done", + sessionId: "session-1", + sessionFile: "/tmp/session.jsonl", + }); + const handler = makeToolHandler(); + + const result = await handler({ action: "transcript", runId, stageId: "summarize", tail: 1, includeToolOutput: true }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; source: string; entries: Array<{ role: string; text?: string; output?: string }>; truncated: boolean; sessionFile?: string }; + assert.equal(transcript.source, "snapshot"); + assert.equal(transcript.sessionFile, "/tmp/session.jsonl"); + assert.equal(transcript.truncated, true); + assert.deepEqual(transcript.entries, [{ role: "assistant", text: "done" }]); + }); + + test("makeExecuteWorkflowTool applies limit and lets tail override limit", async () => { + const runId = `stage-tool-transcript-limit-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-transcript-limit-1", + name: "limited", + status: "completed", + parentIds: [], + toolEvents: [ + { name: "one", output: "1", startedAt: 1, endedAt: 1 }, + { name: "two", output: "2", startedAt: 2, endedAt: 2 }, + { name: "three", output: "3", startedAt: 3, endedAt: 3 }, + ], + result: "done", + endedAt: 4, + }); + const handler = makeToolHandler(); + + const limited = await handler({ action: "transcript", runId, stageId: "limited", limit: 2, includeToolOutput: true }, {} as never); + assert.equal(limited.action, "transcript"); + const limitedTranscript = limited as { action: string; truncated: boolean; entries: Array<{ role: string; toolName?: string; text?: string }> }; + assert.equal(limitedTranscript.truncated, true); + assert.deepEqual(limitedTranscript.entries.map((entry) => entry.toolName ?? entry.text), ["three", "done"]); + + const tailOverride = await handler({ action: "transcript", runId, stageId: "limited", limit: 3, tail: 1, includeToolOutput: true }, {} as never); + assert.equal(tailOverride.action, "transcript"); + const tailTranscript = tailOverride as { action: string; truncated: boolean; entries: Array<{ text?: string }> }; + assert.equal(tailTranscript.truncated, true); + assert.deepEqual(tailTranscript.entries, [{ role: "assistant", text: "done", timestamp: 4 }]); + }); + + test("makeExecuteWorkflowTool labels empty live handles as live transcript source", async () => { + const runId = `stage-tool-live-empty-handle-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-live-empty-handle-1", + name: "live-empty-handle", + status: "running", + parentIds: [], + toolEvents: [], + result: "snapshot-result", + }); + const { dispose } = registerLiveStageHandle(runId, "stage-live-empty-handle-1"); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "transcript", runId, stageId: "live-empty-handle" }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; source: string; entries: unknown[]; truncated: boolean }; + assert.equal(transcript.source, "live"); + assert.equal(transcript.truncated, false); + assert.deepEqual(transcript.entries, []); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool uses error transcript source for target errors", async () => { + const handler = makeToolHandler(); + + const result = await handler({ action: "transcript", runId: "missing-run", stageId: "stage" }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; source: string; entries: Array<{ role: string; text?: string }> }; + assert.equal(transcript.source, "error"); + assert.equal(transcript.entries[0]?.role, "notice"); + }); + + test("makeExecuteWorkflowTool preserves empty live transcript text blocks", async () => { + const runId = `stage-tool-live-empty-block-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-live-empty-block-1", + name: "live-empty", + status: "running", + parentIds: [], + toolEvents: [], + }); + const { dispose } = registerLiveStageHandle(runId, "stage-live-empty-block-1", { + messages: [ + { role: "user", content: [{ type: "text", text: "" }], timestamp: 1 }, + ], + }); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "transcript", runId, stageId: "live-empty" }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; source: string; entries: Array<{ role: string; text?: string }> }; + assert.equal(transcript.source, "live"); + assert.equal(transcript.entries.length, 1); + assert.equal(transcript.entries[0]?.role, "user"); + assert.equal(transcript.entries[0]?.text, ""); + assert.equal(Object.hasOwn(transcript.entries[0]!, "text"), true); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool omits text for live non-text content blocks", async () => { + const runId = `stage-tool-live-non-text-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-live-non-text-1", + name: "live-non-text", + status: "running", + parentIds: [], + toolEvents: [], + }); + const { dispose } = registerLiveStageHandle(runId, "stage-live-non-text-1", { + messages: [ + { role: "user", content: [{ type: "image", data: "", mimeType: "image/png" }], timestamp: 1 }, + ], + }); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "transcript", runId, stageId: "live-non-text" }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; source: string; entries: Array<{ role: string; text?: string }> }; + assert.equal(transcript.source, "live"); + assert.equal(transcript.entries.length, 1); + assert.equal(Object.hasOwn(transcript.entries[0]!, "text"), false); + assert.equal(transcript.entries[0]?.text, undefined); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool returns no truncation marker for tail zero", async () => { + const runId = `stage-tool-transcript-tail-zero-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-transcript-tail-zero-1", + name: "tail-zero", + status: "completed", + parentIds: [], + toolEvents: [{ name: "read", output: "file contents", startedAt: 1, endedAt: 2 }], + result: "done", + }); + const handler = makeToolHandler(); + + const result = await handler({ action: "transcript", runId, stageId: "tail-zero", tail: 0, includeToolOutput: true }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; entries: unknown[]; truncated: boolean }; + assert.equal(transcript.truncated, false); + assert.deepEqual(transcript.entries, []); + }); + + test("makeExecuteWorkflowTool returns final snapshot error after timestamped tools", async () => { + const runId = `stage-tool-transcript-error-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-transcript-error-1", + name: "review", + status: "failed", + parentIds: [], + toolEvents: [{ name: "grep", output: "matches", startedAt: 10, endedAt: 11 }], + error: "boom", + endedAt: 12, + }); + const handler = makeToolHandler(); + + const result = await handler({ action: "transcript", runId, stageId: "review", tail: 1, includeToolOutput: true }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; entries: Array<{ role: string; text?: string; timestamp?: number }>; truncated: boolean }; + assert.equal(transcript.truncated, true); + assert.deepEqual(transcript.entries, [{ role: "notice", text: "boom", timestamp: 12 }]); + }); + + test("makeExecuteWorkflowTool keeps terminal snapshot entries after tools for tied timestamps", async () => { + const runId = `stage-tool-transcript-tie-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-transcript-tie-1", + name: "tie", + status: "completed", + parentIds: [], + toolEvents: [{ name: "read", output: "file contents", startedAt: 4, endedAt: 5 }], + result: "finished", + endedAt: 5, + }); + const handler = makeToolHandler(); + + const result = await handler({ action: "transcript", runId, stageId: "tie", tail: 1, includeToolOutput: true }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; entries: Array<{ role: string; text?: string; timestamp?: number }>; truncated: boolean }; + assert.equal(transcript.truncated, true); + assert.deepEqual(transcript.entries, [{ role: "assistant", text: "finished", timestamp: 5 }]); + }); + + test("makeExecuteWorkflowTool preserves empty final snapshot result after tools", async () => { + const runId = `stage-tool-transcript-empty-result-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-transcript-empty-result-1", + name: "empty-result", + status: "completed", + parentIds: [], + toolEvents: [{ name: "read", output: "file contents", startedAt: 1, endedAt: 2 }], + result: "", + }); + const handler = makeToolHandler(); + + const result = await handler({ action: "transcript", runId, stageId: "empty-result", tail: 1, includeToolOutput: true }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; entries: Array<{ role: string; text?: string }>; truncated: boolean }; + assert.equal(transcript.truncated, true); + assert.deepEqual(transcript.entries, [{ role: "assistant", text: "" }]); + }); + + test("makeExecuteWorkflowTool preserves empty final snapshot error after tools", async () => { + const runId = `stage-tool-transcript-empty-error-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { + id: "stage-transcript-empty-error-1", + name: "empty-error", + status: "failed", + parentIds: [], + toolEvents: [{ name: "grep", output: "matches", startedAt: 10, endedAt: 11 }], + error: "", + }); + const handler = makeToolHandler(); + + const result = await handler({ action: "transcript", runId, stageId: "empty-error", tail: 1, includeToolOutput: true }, {} as never); + + assert.equal(result.action, "transcript"); + const transcript = result as { action: string; entries: Array<{ role: string; text?: string }>; truncated: boolean }; + assert.equal(transcript.truncated, true); + assert.deepEqual(transcript.entries, [{ role: "notice", text: "" }]); + }); + + test("makeExecuteWorkflowTool answers stage pending prompts", async () => { + const runId = `stage-tool-send-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-1", name: "ask", status: "awaiting_input", parentIds: [], toolEvents: [] }); + store.recordStagePendingPrompt(runId, "stage-prompt-1", { id: "prompt-1", kind: "input", message: "Value?", createdAt: Date.now() }); + const handler = makeToolHandler(); + + const result = await handler({ action: "send", runId, stageId: "ask", text: "42" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "answer"); + assert.equal(send.status, "ok"); + assert.match(send.message, /Answered prompt/); + const stage = store.runs().find((run) => run.id === runId)?.stages.find((s) => s.id === "stage-prompt-1"); + assert.equal(stage?.pendingPrompt, undefined); + }); + + test("makeExecuteWorkflowTool leaves pending prompts untouched when payload is omitted", async () => { + const runId = `stage-tool-send-omitted-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-omitted", name: "ask-omitted", status: "awaiting_input", parentIds: [], toolEvents: [] }); + store.recordStagePendingPrompt(runId, "stage-prompt-omitted", { id: "prompt-omitted", kind: "input", message: "Value?", createdAt: Date.now() }); + const handler = makeToolHandler(); + + const result = await handler({ action: "send", runId, stageId: "ask-omitted" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "answer"); + assert.equal(send.status, "noop"); + assert.match(send.message, /requires text, response, or message/); + const stage = store.runs().find((run) => run.id === runId)?.stages.find((s) => s.id === "stage-prompt-omitted"); + assert.equal(stage?.pendingPrompt?.id, "prompt-omitted"); + }); + + test("makeExecuteWorkflowTool delivery answer without a pending prompt does not fall through to live followUp", async () => { + const runId = `stage-tool-send-answer-no-prompt-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-no-prompt", name: "ask", status: "running", parentIds: [], toolEvents: [] }); + const { followUps, dispose } = registerLiveStageHandle(runId, "stage-no-prompt"); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "send", runId, stageId: "ask", delivery: "answer", text: "42" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "answer"); + assert.equal(send.status, "noop"); + assert.match(send.message, /No pending prompt/); + assert.deepEqual(followUps, []); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool auto delivery without a targeted prompt still queues a live followUp", async () => { + const runId = `stage-tool-send-auto-live-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-auto-live", name: "ask", status: "running", parentIds: [], toolEvents: [] }); + const { followUps, dispose } = registerLiveStageHandle(runId, "stage-auto-live"); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "send", runId, stageId: "ask", text: "next" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "followUp"); + assert.equal(send.status, "ok"); + assert.deepEqual(followUps, ["next"]); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool sends explicit prompt delivery to live handles", async () => { + const runId = `stage-tool-send-prompt-live-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-live", name: "ask", status: "running", parentIds: [], toolEvents: [] }); + const { followUps, prompts, steers, dispose } = registerLiveStageHandle(runId, "stage-prompt-live"); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "send", runId, stageId: "ask", delivery: "prompt", text: "start next" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string }; + assert.equal(send.delivery, "prompt"); + assert.equal(send.status, "ok"); + assert.deepEqual(prompts, ["start next"]); + assert.deepEqual(steers, []); + assert.deepEqual(followUps, []); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool sends explicit steer delivery to live handles", async () => { + const runId = `stage-tool-send-steer-live-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-steer-live", name: "ask", status: "running", parentIds: [], toolEvents: [] }); + const { followUps, prompts, steers, dispose } = registerLiveStageHandle(runId, "stage-steer-live", { isStreaming: true }); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "send", runId, stageId: "ask", delivery: "steer", text: "adjust course" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string }; + assert.equal(send.delivery, "steer"); + assert.equal(send.status, "ok"); + assert.deepEqual(steers, ["adjust course"]); + assert.deepEqual(prompts, []); + assert.deepEqual(followUps, []); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool promptId mismatch does not fall through to live followUp", async () => { + const runId = `stage-tool-send-prompt-mismatch-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-mismatch", name: "ask", status: "awaiting_input", parentIds: [], toolEvents: [] }); + store.recordStagePendingPrompt(runId, "stage-prompt-mismatch", { id: "prompt-real", kind: "input", message: "Value?", createdAt: Date.now() }); + const { followUps, dispose } = registerLiveStageHandle(runId, "stage-prompt-mismatch"); + const handler = makeToolHandler(); + + try { + const result = await handler({ action: "send", runId, stageId: "ask", promptId: "prompt-missing", text: "42" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "answer"); + assert.equal(send.status, "noop"); + assert.match(send.message, /No matching pending prompt prompt-missing/); + assert.deepEqual(followUps, []); + const stage = store.runs().find((run) => run.id === runId)?.stages.find((s) => s.id === "stage-prompt-mismatch"); + assert.equal(stage?.pendingPrompt?.id, "prompt-real"); + } finally { + dispose(); + } + }); + + test("makeExecuteWorkflowTool treats explicit empty text prompt payload as an answer", async () => { + const runId = `stage-tool-send-empty-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-empty", name: "ask-empty", status: "awaiting_input", parentIds: [], toolEvents: [] }); + store.recordStagePendingPrompt(runId, "stage-prompt-empty", { id: "prompt-empty", kind: "input", message: "Value?", createdAt: Date.now() }); + const handler = makeToolHandler(); + + const result = await handler({ action: "send", runId, stageId: "ask-empty", text: "" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "answer"); + assert.equal(send.status, "ok"); + assert.match(send.message, /Answered prompt/); + const stage = store.runs().find((run) => run.id === runId)?.stages.find((s) => s.id === "stage-prompt-empty"); + assert.equal(stage?.pendingPrompt, undefined); + }); + + test("makeExecuteWorkflowTool treats explicit empty response prompt payload as an answer", async () => { + const runId = `stage-tool-send-empty-response-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-empty-response", name: "ask-empty-response", status: "awaiting_input", parentIds: [], toolEvents: [] }); + store.recordStagePendingPrompt(runId, "stage-prompt-empty-response", { id: "prompt-empty-response", kind: "input", message: "Value?", createdAt: Date.now() }); + const handler = makeToolHandler(); + + const result = await handler({ action: "send", runId, stageId: "ask-empty-response", response: "" }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "answer"); + assert.equal(send.status, "ok"); + assert.match(send.message, /Answered prompt/); + const stage = store.runs().find((run) => run.id === runId)?.stages.find((s) => s.id === "stage-prompt-empty-response"); + assert.equal(stage?.pendingPrompt, undefined); + }); + + test("makeExecuteWorkflowTool ignores explicit undefined prompt payloads", async () => { + const runId = `stage-tool-send-undefined-${Date.now()}`; + store.recordRunStart(makeInflightRun(runId)); + store.recordStageStart(runId, { id: "stage-prompt-undefined", name: "ask-undefined", status: "awaiting_input", parentIds: [], toolEvents: [] }); + store.recordStagePendingPrompt(runId, "stage-prompt-undefined", { id: "prompt-undefined", kind: "input", message: "Value?", createdAt: Date.now() }); + const handler = makeToolHandler(); + + const result = await handler({ action: "send", runId, stageId: "ask-undefined", text: undefined }, {} as never); + + assert.equal(result.action, "send"); + const send = result as { action: string; delivery: string; status: string; message: string }; + assert.equal(send.delivery, "answer"); + assert.equal(send.status, "noop"); + assert.match(send.message, /requires text, response, or message/); + const stage = store.runs().find((run) => run.id === runId)?.stages.find((s) => s.id === "stage-prompt-undefined"); + assert.equal(stage?.pendingPrompt?.id, "prompt-undefined"); + }); + + test("makeExecuteWorkflowTool reloads directly without sending a literal slash command", async () => { + const registry = createRegistry([]); + const runtime = createExtensionRuntime({ registry }); + let reloads = 0; + const handler = makeExecuteWorkflowTool(runtime, () => undefined, async () => { + reloads += 1; + }); + const sent: string[] = []; + + const result = await handler({ action: "reload", reason: "test" }, { + // Sentinel-only property: production ExtensionAPI does not expose this. + sendUserMessage: (content: string) => { + sent.push(content); + }, + } as never); + + assert.equal(result.action, "reload"); + const reload = result as { action: string; status: string; message: string }; + assert.equal(reload.status, "ok"); + assert.match(reload.message, /Reloaded workflow resources/); + assert.equal(reloads, 1); + assert.deepEqual(sent, []); + }); + + test("makeExecuteWorkflowTool treats explicit empty reload reason as omitted", async () => { + const registry = createRegistry([]); + const runtime = createExtensionRuntime({ registry }); + const handler = makeExecuteWorkflowTool(runtime, () => undefined, () => undefined); + + const result = await handler({ action: "reload", reason: "" }, {} as never); + + assert.equal(result.action, "reload"); + const reload = result as { action: string; status: string; message: string }; + assert.equal(reload.status, "ok"); + assert.equal(reload.message, "Reloaded workflow resources."); + }); + + test("makeExecuteWorkflowTool reload is skipped while workflows are in flight", async () => { + const registry = createRegistry([]); + const runtime = createExtensionRuntime({ registry }); + let reloads = 0; + const handler = makeExecuteWorkflowTool(runtime, () => undefined, () => { + reloads += 1; + }); + store.recordRunStart(makeInflightRun(`reload-blocked-${Date.now()}`)); + + const result = await handler({ action: "reload", reason: "test" }, {} as never); + + assert.equal(result.action, "reload"); + const reload = result as { action: string; status: string; message: string }; + assert.equal(reload.status, "noop"); + assert.match(reload.message, /still in flight/); + assert.equal(reloads, 0); + }); + + test("makeExecuteWorkflowTool reload surfaces callback failures as noop", async () => { + const registry = createRegistry([]); + const runtime = createExtensionRuntime({ registry }); + const handler = makeExecuteWorkflowTool(runtime, () => undefined, async () => { + throw new Error("bad workflow config"); + }); + + const result = await handler({ action: "reload", reason: "test" }, {} as never); + + assert.equal(result.action, "reload"); + const reload = result as { action: string; status: string; message: string }; + assert.equal(reload.status, "noop"); + assert.match(reload.message, /Reload failed: bad workflow config/); + }); + test("makeExecuteWorkflowTool returns ambiguous run-prefix messages", async () => { store.recordRunStart(makeInflightRun("ambiguous-run-a")); store.recordRunStart(makeInflightRun("ambiguous-run-b")); @@ -1089,7 +1929,7 @@ describe("tool run-control actions", () => { store, adapters: { prompt: { prompt: async (text) => { calls.push(text); return "second-new"; } } }, }); - const handler = makeExecuteWorkflowTool(runtime, () => undefined); + const handler = makeExecuteWorkflowTool(runtime, () => undefined, () => undefined); const result = await handler({ action: "resume", runId: sourceRunId }, {} as never); diff --git a/test/unit/workflow-schema.test.ts b/test/unit/workflow-schema.test.ts index 79dd2751c1..d8a3fb5a11 100644 --- a/test/unit/workflow-schema.test.ts +++ b/test/unit/workflow-schema.test.ts @@ -50,6 +50,73 @@ describe("WorkflowParametersSchema stage options", () => { assert.equal(Value.Check(WorkflowParametersSchema, payload), true); }); + test("accepts stage introspection and control actions", () => { + for (const statusFilter of ["pending", "running", "awaiting_input", "paused", "blocked", "completed", "failed", "skipped", "all"] as const) { + assert.equal(Value.Check(WorkflowParametersSchema, { + action: "stages", + runId: "abc123", + statusFilter, + }), true); + } + assert.equal(Value.Check(WorkflowParametersSchema, { + action: "transcript", + runId: "abc123", + stageId: "review", + format: "text", + tail: 20, + includeToolOutput: true, + }), true); + assert.equal(Value.Check(WorkflowParametersSchema, { + action: "send", + runId: "abc123", + stageId: "review", + text: "continue", + delivery: "followUp", + promptId: "prompt-1", + }), true); + assert.equal(Value.Check(WorkflowParametersSchema, { + action: "pause", + runId: "abc123", + stageId: "review", + }), true); + assert.equal(Value.Check(WorkflowParametersSchema, { + action: "reload", + reason: "created a workflow file", + }), true); + }); + + test("exposes descriptions for agent-facing action fields", () => { + const properties = (WorkflowParametersSchema as unknown as { + properties: Record; + }).properties; + + for (const field of [ + "statusFilter", + "format", + "limit", + "tail", + "includeToolOutput", + "text", + "response", + "delivery", + "promptId", + "reason", + ]) { + assert.equal(typeof properties[field]?.description, "string", `${field} description`); + assert.ok((properties[field]?.description ?? "").length > 0, `${field} description`); + } + }); + + test("rejects invalid stage-control enum values and transcript counts", () => { + assert.equal(Value.Check(WorkflowParametersSchema, { action: "stages", statusFilter: "cancelled" }), false); + assert.equal(Value.Check(WorkflowParametersSchema, { action: "transcript", format: "markdown" }), false); + assert.equal(Value.Check(WorkflowParametersSchema, { action: "transcript", limit: -1 }), false); + assert.equal(Value.Check(WorkflowParametersSchema, { action: "transcript", limit: 1.5 }), false); + assert.equal(Value.Check(WorkflowParametersSchema, { action: "transcript", tail: -1 }), false); + assert.equal(Value.Check(WorkflowParametersSchema, { action: "transcript", tail: 1.5 }), false); + assert.equal(Value.Check(WorkflowParametersSchema, { action: "send", delivery: "chat" }), false); + }); + test("rejects non-array and non-string fallbackModels", () => { assert.equal(Value.Check(WorkflowParametersSchema, { task: { name: "planner", prompt: "plan", fallbackModels: "openai/fallback" },