From a1621f3702129eed91f2f18a2308644f57593ba4 Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Tue, 14 Jul 2026 21:33:48 -0700 Subject: [PATCH 1/7] feat(workflows): unify post-mortem stage chat across attach, restore/replay, and send MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Generalize the completed-workflow-inspection detached-handle pattern (#1758) into a shared post-mortem stage-chat resolver so any eligible terminal agent stage with a valid retained Atomic session reopens as an interactive follow-up conversation — not only completed-inspection, but also generic /workflow attach and /workflow connect, restored/replayed durable snapshots after a process restart, and workflow send. Adds ensurePostMortemStageHandle: reuses an existing non-disposed handle, validates the retained sessionFile is an existing readable context-bearing transcript, and lazily reopens it through a detached, single-flight StageControlRegistry.getOrCreateDetached handle keyed by the real runId/stageId. Follow-ups append in place; run/stage status, results, timings, checkpoints, replay metadata, and graph topology stay immutable, and post-mortem chat can never resume/retry/rewind/pause/re-dispatch execution. Stages without a valid retained agent session keep the read-only transcript; recoverably failed stages keep execution-resume semantics. Refs: #1811 Assistant-model: Claude Opus 4.8 --- packages/coding-agent/docs/workflows.md | 2 + packages/workflows/CHANGELOG.md | 4 + packages/workflows/README.md | 2 +- .../src/durable/completed-catalog.ts | 54 +---- .../src/durable/completed-inspection.ts | 76 +------ .../src/extension/extension-factory.ts | 11 +- .../src/extension/extension-runtime-state.ts | 3 + .../src/extension/postmortem-deps.ts | 68 +++++++ .../src/extension/workflow-tool-send.ts | 24 ++- .../workflows/src/extension/workflow-tool.ts | 5 +- .../runs/foreground/postmortem-stage-chat.ts | 172 ++++++++++++++++ .../runs/foreground/stage-control-registry.ts | 30 +++ .../src/shared/session-transcript.ts | 74 +++++++ packages/workflows/src/tui/overlay-adapter.ts | 12 +- .../src/tui/workflow-attach-pane-types.ts | 11 +- .../workflows/src/tui/workflow-attach-pane.ts | 5 +- test/unit/postmortem-stage-chat.test.ts | 188 ++++++++++++++++++ test/unit/stage-control-registry.test.ts | 42 ++++ test/unit/workflow-attach-pane-11.test.ts | 131 ++++++++++++ .../workflow-tool-send-postmortem.test.ts | 108 ++++++++++ 20 files changed, 887 insertions(+), 135 deletions(-) create mode 100644 packages/workflows/src/extension/postmortem-deps.ts create mode 100644 packages/workflows/src/runs/foreground/postmortem-stage-chat.ts create mode 100644 packages/workflows/src/shared/session-transcript.ts create mode 100644 test/unit/postmortem-stage-chat.test.ts create mode 100644 test/unit/workflow-attach-pane-11.test.ts create mode 100644 test/unit/workflow-tool-send-postmortem.test.ts diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index e27556744..e6216d555 100644 --- a/packages/coding-agent/docs/workflows.md +++ b/packages/coding-agent/docs/workflows.md @@ -416,6 +416,8 @@ When a paused stage is resumed with a message, Atomic lets the stage answer that Durable `/workflow resume` preserves completed stage metadata, active-stage elapsed time, and graph topology. While an LM stage or task is active, repeated durable checkpoints refresh its accumulated pause-adjusted duration even when its session file does not change. Each new Atomic process that reopens the unfinished session mid-chat starts from the latest saved baseline and uses the same continuation prompt shown above, so repeated process-boundary resumes keep status, graph, stored, and lifecycle duration cumulative without double-counting pauses from earlier process segments. Replayed `ctx.stage`, `ctx.task`, `ctx.chain`, `ctx.parallel`, and child-workflow checkpoints keep their original summaries, timing, session/model metadata, and parallel fanout parentage instead of appearing as freshly flattened replay nodes. +**Post-mortem chat vs. execution resume.** These are distinct operations. *Resuming workflow execution* (`/workflow resume`) is for paused, interrupted, recoverably failed, or unfinished durable work; it may replay checkpoints, continue an incomplete stage, and dispatch remaining DAG work. *Opening a post-mortem chat* reopens one terminal agent stage's retained conversation for follow-up only — it never resumes, retries, rewinds, or otherwise changes workflow execution. Any eligible terminal agent stage with a valid retained session opens as an interactive post-mortem chat regardless of how you reach it: same-process `task`/`tasks`/`chain` stages, completed-workflow inspection, generic `/workflow attach` / `/workflow connect`, restored/replayed durable snapshots after a restart, and `workflow({ action: "send" })`. Follow-up turns are appended in place to the stage's retained session (no separate fork), so the agent may still invoke its ordinary tools and cause side effects; only the workflow DAG, run/stage status, results, timings, checkpoints, and topology are immutable. A stage stays a **read-only transcript** when it has no valid retained agent session — prompt/HIL and boundary/summary nodes, skipped nodes without a completed conversation, non-terminal handle-less stages (another process may still own the session), and missing/malformed/deleted session files. Recoverably failed stages keep their execution-resume semantics and are not silently reopened as post-mortem chat. + Workflow stage sessions and first-party subagent transcripts created inside them are classified as **internal** at creation and excluded from the standard `/resume`, `atomic -r`, `--continue`, and global history surfaces. Fork-context stages and subagents inherit the owning run/stage marker in their initial JSONL header, avoiding a briefly visible ordinary session. They remain resumable and inspectable through the workflow-specific commands and tool actions shown here (`/workflow resume`, `/workflow attach`, `workflow({ action: "status" | "stages" | "stage" | "resume" })`), which read the run/stage store and its `sessionFile` links directly. Passing a stage session's file path to `--session` still opens it explicitly. Classification requires exact `internal: true` plus complete run/stage metadata; malformed legacy markers and ordinary user forks remain in standard history. Legacy workflow sessions created before this marker behavior lack provable ownership and continue to appear until they age out. Human-in-the-loop prompts from `ctx.ui.input`, `ctx.ui.confirm`, `ctx.ui.select`, `ctx.ui.editor`, and `ctx.ui.custom` appear as awaiting-input nodes in the workflow graph viewer, not as chat modals — use `/workflow connect ` (or F2), then press Enter on the focused node or click a visible graph node directly to focus and open/attach it for local answers. diff --git a/packages/workflows/CHANGELOG.md b/packages/workflows/CHANGELOG.md index 20e553f2b..e9852e041 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 + +- Unified workflow **post-mortem stage chat** across every inspection surface. Any eligible terminal agent stage with a valid retained session now reopens as an interactive follow-up conversation — not only completed-workflow inspection (#1758), but also generic `/workflow attach` / `/workflow connect`, restored/replayed durable snapshots after a process restart, and `workflow({ action: "send" })`. A shared runtime resolver (`ensurePostMortemStageHandle`) validates the retained session is an existing, readable, context-bearing Atomic transcript, then lazily reopens it through a detached, single-flight stage-control handle keyed by the real `{ runId, stageId }` (including nested/expanded child stages). Follow-up turns are appended in place to the retained session and the agent keeps its ordinary tools, while run/stage status, results, timings, checkpoints, replay metadata, and graph topology remain immutable — post-mortem chat can never resume, retry, rewind, pause, or re-dispatch workflow execution. Stages without a valid retained agent session (prompt/HIL and boundary/summary nodes, skipped nodes, non-terminal handle-less stages, and missing/malformed/deleted session files) keep the existing read-only transcript, and recoverably failed stages retain their execution-resume semantics. `workflow send` now revives such a stage on a registry miss and delivers the message as a conversational follow-up instead of reporting `No live handle for stage.` ([#1811](https://github.com/bastani-inc/atomic/issues/1811)) + ### Fixed - Fixed workflow resource reload to build and atomically publish a fresh registry for additions, edits, renames, deletions, config paths, conventional/legacy directories, and package resources without restarting Atomic. Reloads now serialize and coalesce, reject stale session generations, retain the active registry on fatal failures, remain safe during in-flight runs, and return visible config/discovery diagnostics through both slash-command and tool surfaces while preserving valid siblings. diff --git a/packages/workflows/README.md b/packages/workflows/README.md index d0ff6b78d..b2998ff61 100644 --- a/packages/workflows/README.md +++ b/packages/workflows/README.md @@ -600,7 +600,7 @@ Prompt answer replay is live-memory only. `StageSnapshot.promptAnswerState` repo - **`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`** — path-only by default when a transcript file exists: use `status`, `stages`, or `stage` to identify the stage and its `sessionFile`/`transcriptPath`, quote the exact path without changing platform separators (for example, preserve Windows backslashes), then search that file with `rg`/`grep` for targeted terms and read only small surrounding ranges. Default text results include JSON-escaped `sessionFileJson`/`transcriptPathJson` lines for copy-safe path literals plus a `lazyReadPrompt`, with `entries: not inlined` so transcript bodies and tool outputs stay out of model context. Passing explicit `tail` or `limit` opts into a bounded inline preview for quick context checks. If no transcript path is available, the action falls back to a bounded preview of up to 5 recent entries with a `fallbackNote`. A registered live stage handle is used when one exists, even before live messages arrive; otherwise the action 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 only to inlined snapshot previews or no-path fallback previews; live session transcripts may not expose tool output. -- **`send`** — answers pending primitive/structured 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. Follow-ups to completed or failed stages reuse retained `sessionFile` metadata when available so the conversation resumes from the archived stage transcript instead of starting empty; if no session metadata was retained, the follow-up is refused instead of silently resetting. Arbitrary `ctx.ui.custom` widget prompts require the interactive workflow graph and return a clear unsupported message when targeted through `send`. `delivery: "auto"` answers pending prompts first, then resumes paused stages, steers streaming stages, or queues a follow-up. +- **`send`** — answers pending primitive/structured 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. Follow-ups to eligible terminal agent stages revive an interactive **post-mortem chat** through the shared resolver: on a live-handle miss the stage's retained `sessionFile` is validated (existing, readable, context-bearing) and reopened as a detached, single-flight handle so the message is delivered as a conversational follow-up appended in place — the same path used by `/workflow attach`, restored/replayed durable snapshots, and completed-workflow inspection — without resuming, retrying, or re-dispatching workflow execution. If no valid retained session exists, the follow-up is refused (`No live handle for stage.`) instead of silently resetting or exposing a handle-less non-terminal session. Arbitrary `ctx.ui.custom` widget prompts require the interactive workflow graph and return a clear unsupported message when targeted through `send`. `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/durable/completed-catalog.ts b/packages/workflows/src/durable/completed-catalog.ts index fc8aa51c8..c81d19249 100644 --- a/packages/workflows/src/durable/completed-catalog.ts +++ b/packages/workflows/src/durable/completed-catalog.ts @@ -1,6 +1,6 @@ -import { readFileSync, statSync } from "node:fs"; import type { RunSnapshot, StageSnapshot } from "../shared/store-types.js"; import type { WorkflowInputValues } from "../shared/types.js"; +import { isReopenableSessionTranscript } from "../shared/session-transcript.js"; import type { DurableWorkflowBackend } from "./backend.js"; import type { DurableCheckpoint, @@ -15,15 +15,6 @@ export type CompletedWorkflowResolution = | { readonly kind: "not_found" } | { readonly kind: "stale"; readonly entry: ResumableWorkflowEntry }; -interface SessionTranscriptEntry { - readonly type?: string; - readonly id?: string; - readonly timestamp?: string; - readonly message?: { - readonly role?: string; - readonly content?: string | object; - }; -} interface StageDraft { readonly replayKey: string; @@ -109,49 +100,6 @@ function validatedStageTranscript(stage: StageSnapshot): StageSnapshot { return withoutSessionFile; } -function isReopenableSessionTranscript(path: string): boolean { - try { - const stats = statSync(path); - if (!stats.isFile() || stats.size === 0) return false; - const lines = readFileSync(path, "utf8").split("\n").filter((line) => line.trim().length > 0); - if (lines.length < 2) return false; - const entries: SessionTranscriptEntry[] = []; - for (const line of lines) { - const parsed = JSON.parse(line) as object; - if (typeof parsed !== "object" || parsed === null) return false; - entries.push(parsed as SessionTranscriptEntry); - } - const header = entries[0]; - return header?.type === "session" && typeof header.id === "string" && entries.some(isUsableContextMessage); - } catch { - return false; - } -} - -function isUsableContextMessage(entry: SessionTranscriptEntry): boolean { - return entry.type === "message" - && typeof entry.id === "string" - && typeof entry.timestamp === "string" - && typeof entry.message?.role === "string" - && hasUsableMessageContent(entry.message.content); -} - -function hasUsableMessageContent(content: string | object | undefined): boolean { - if (typeof content === "string") return content.trim().length > 0; - return Array.isArray(content) && content.some(hasUsableContentBlock); -} - -function hasUsableContentBlock(block: object): boolean { - if (typeof block !== "object" || block === null) return false; - const contentBlock = block as { - readonly text?: string; - readonly thinking?: string; - readonly data?: string; - readonly name?: string; - }; - return [contentBlock.text, contentBlock.thinking, contentBlock.data, contentBlock.name] - .some((value) => typeof value === "string" && value.trim().length > 0); -} function stageSnapshotsFromCheckpoints( checkpoints: readonly DurableCheckpoint[], diff --git a/packages/workflows/src/durable/completed-inspection.ts b/packages/workflows/src/durable/completed-inspection.ts index 3417492a7..dae21f971 100644 --- a/packages/workflows/src/durable/completed-inspection.ts +++ b/packages/workflows/src/durable/completed-inspection.ts @@ -1,9 +1,9 @@ import type { Store } from "../shared/store.js"; -import type { RunSnapshot, StageSnapshot } from "../shared/store-types.js"; -import { createStageContext, type StageAdapters } from "../runs/foreground/stage-runner.js"; +import type { RunSnapshot } from "../shared/store-types.js"; +import type { StageAdapters } from "../runs/foreground/stage-runner.js"; +import { createPostMortemStageHandle } from "../runs/foreground/postmortem-stage-chat.js"; import { stageControlRegistry as defaultStageControlRegistry, - type AgentSessionEventListener, type StageControlHandle, type StageControlRegistry, } from "../runs/foreground/stage-control-registry.js"; @@ -117,7 +117,7 @@ function registerCompletedChatHandles( } else if (existing !== undefined) { disposeCompletedChatHandle(existing); } - const handle = createCompletedChatHandle(snapshot, stage, stage.sessionFile, deps.adapters, deps.cwd, deps.defaultSessionDir); + const handle = createPostMortemStageHandle(snapshot.id, stage, stage.sessionFile, deps.adapters, deps.cwd, deps.defaultSessionDir); const unregister = registry.register(handle); registrations.set(key, { handle, unregister }); registry.detachControl(snapshot.id, stage.id, handle); @@ -143,71 +143,3 @@ function disposeCompletedChatHandle(handle: StageControlHandle): void { console.warn("atomic-workflows: completed chat handle dispose failed", error); }); } - -function createCompletedChatHandle( - run: RunSnapshot, - stage: StageSnapshot, - sessionFile: string, - adapters: StageAdapters, - cwd: string | undefined, - defaultSessionDir: string | undefined, -): StageControlHandle { - const context = createStageContext({ - runId: run.id, - stageId: stage.id, - stageName: stage.name, - adapters, - stageOptions: { - resumeFromSessionFile: sessionFile, - ...(cwd !== undefined ? { cwd } : {}), - }, - ...(defaultSessionDir !== undefined ? { defaultSessionDir } : {}), - }); - let disposed = false; - const ensureAttached = async (): Promise => { - if (disposed) throw new Error(`Completed stage chat "${stage.name}" is closed.`); - if (context.__sessionMeta().sessionFile === undefined) { - await context.__ensureSessionFromFile(sessionFile); - } - }; - return { - runId: run.id, - stageId: stage.id, - stageName: stage.name, - status: "completed", - get sessionId() { return context.__sessionMeta().sessionId ?? stage.sessionId; }, - get sessionFile() { return context.__sessionMeta().sessionFile ?? sessionFile; }, - get isStreaming() { return context.isStreaming; }, - get isDisposed() { return disposed; }, - get messages() { return context.messages; }, - get agentSession() { return context.__agentSession(); }, - async ensureAttached() { await ensureAttached(); }, - async prompt(text: string) { - await ensureAttached(); - await context.prompt(text); - }, - async steer(text: string) { - await ensureAttached(); - await context.steer(text); - }, - async followUp(text: string) { - await ensureAttached(); - await context.followUp(text); - }, - async pause() { - throw new Error("Completed workflow snapshots cannot be paused or resumed."); - }, - async resume(message?: string) { - if (message !== undefined && message.trim().length > 0) { - await ensureAttached(); - await context.prompt(message); - } - }, - subscribe(listener: AgentSessionEventListener) { return context.subscribe(listener); }, - async dispose() { - if (disposed) return; - disposed = true; - await context.__dispose(); - }, - }; -} diff --git a/packages/workflows/src/extension/extension-factory.ts b/packages/workflows/src/extension/extension-factory.ts index a5680a997..d83f0665d 100644 --- a/packages/workflows/src/extension/extension-factory.ts +++ b/packages/workflows/src/extension/extension-factory.ts @@ -15,6 +15,8 @@ import { createWorkflowExtensionRuntimeState } from "./extension-runtime-state.j import { registerWorkflowLifecycleHandlers } from "./extension-lifecycle.js"; import { dynamicTextRenderComponent } from "./render-component.js"; import { makeExecuteWorkflowTool } from "./workflow-tool.js"; +import { createPostMortemHandleResolver, postMortemDepsForRun } from "./postmortem-deps.js"; +import type { StageControlHandle } from "../runs/foreground/stage-control-registry.js"; import { registerWorkflowTool } from "./workflow-tool-registration.js"; import { registerWorkflowSlashCommand } from "./workflow-command-registration.js"; import { installInputInterceptor, type WorkflowCommandHandler } from "./workflow-command-utils.js"; @@ -34,8 +36,10 @@ function registerWorkflowMessageRenderers(pi: ExtensionAPI): void { function buildWorkflowOverlay( pi: ExtensionAPI, + resolvePostMortemHandle: (runId: string, stageId: string) => StageControlHandle | undefined, ): GraphOverlayPort { return buildGraphOverlayAdapter(pi, store, { + resolvePostMortemHandle, onQuitRun: (runId) => { quitRun(runId, { store }); pi.ui?.notify?.(`Workflow quit; resume with /workflow resume.`, "info"); @@ -77,7 +81,11 @@ function registerIntercomControl( function factory(pi: ExtensionAPI): void { const adapters = buildRuntimeAdapters(pi); const runtimeState = createWorkflowExtensionRuntimeState(pi, adapters); - const overlay = buildWorkflowOverlay(pi); + const postMortemResolverDeps = { + adapters, + resolveDefaultStageSessionDir: runtimeState.resolveDefaultStageSessionDir, + }; + const overlay = buildWorkflowOverlay(pi, createPostMortemHandleResolver(postMortemResolverDeps)); const workflowCommands = new Map(); const storeWidgetRef: { current: (() => void) | null } = { current: null }; const intercomControlRef: { current: (() => void) | null } = { current: null }; @@ -86,6 +94,7 @@ function factory(pi: ExtensionAPI): void { () => runtimeState.persistenceRef.current, runtimeState.reloadWorkflowResources, runtimeState.ensureWorkflowResourcesLoaded, + { resolvePostMortemDeps: (runId) => postMortemDepsForRun(runId, postMortemResolverDeps) }, ); registerWorkflowTool(pi, executeWorkflowTool, runtimeState.runWithLifecycleSuppressedForPolicy); diff --git a/packages/workflows/src/extension/extension-runtime-state.ts b/packages/workflows/src/extension/extension-runtime-state.ts index 922a5ec4c..99d56ee78 100644 --- a/packages/workflows/src/extension/extension-runtime-state.ts +++ b/packages/workflows/src/extension/extension-runtime-state.ts @@ -60,6 +60,8 @@ export interface WorkflowExtensionRuntimeState { setNotificationsActive(active: boolean): void; setIntercomParentSession(session: string | null): void; updateHostStageSessionDir(sessionManager: SessionManager | undefined): void; + /** Current default stage session directory, when the host set a non-default one. */ + resolveDefaultStageSessionDir(): string | undefined; } export function createWorkflowExtensionRuntimeState( @@ -442,5 +444,6 @@ export function createWorkflowExtensionRuntimeState( hostStageSessionDir.current = undefined; } }, + resolveDefaultStageSessionDir, }; } diff --git a/packages/workflows/src/extension/postmortem-deps.ts b/packages/workflows/src/extension/postmortem-deps.ts new file mode 100644 index 000000000..9269884a5 --- /dev/null +++ b/packages/workflows/src/extension/postmortem-deps.ts @@ -0,0 +1,68 @@ +/** + * Extension-side wiring for the shared post-mortem stage-chat resolver. + * + * Builds `PostMortemStageChatDeps` from live extension runtime surfaces + * (stage adapters, durable per-run cwd, and the default stage session dir after + * a host restart) so both the TUI attach pane and `workflow send` revive an + * eligible terminal agent stage through the same detached, single-flight + * resolver instead of process-local handle presence alone. + * + * cross-ref: + * - src/runs/foreground/postmortem-stage-chat.ts (resolver) + * - src/tui/overlay-adapter.ts (attach pane wiring) + * - src/extension/workflow-tool-send.ts (send parity) + */ +import { store } from "../shared/store.js"; +import { getDurableBackend } from "../durable/factory.js"; +import { + ensurePostMortemStageHandle, + type PostMortemStageChatDeps, +} from "../runs/foreground/postmortem-stage-chat.js"; +import { stageControlRegistry } from "../runs/foreground/stage-control-registry.js"; +import type { StageControlHandle } from "../runs/foreground/stage-control-registry.js"; +import type { StageAdapters } from "../runs/foreground/stage-runner.js"; + +export interface PostMortemResolverDeps { + readonly adapters: StageAdapters; + readonly resolveDefaultStageSessionDir: () => string | undefined; +} + +/** Persisted original/resolved cwd for a durable run, when still available. */ +function resolveStageCwd(runId: string): string | undefined { + try { + const handle = getDurableBackend().getWorkflow(runId); + return handle?.workflowCwd ?? handle?.invocationCwd ?? undefined; + } catch { + return undefined; + } +} + +/** Resolver deps for a specific run, keyed so revived handles use the real run cwd. */ +export function postMortemDepsForRun( + runId: string, + deps: PostMortemResolverDeps, +): PostMortemStageChatDeps { + return { + registry: stageControlRegistry, + adapters: deps.adapters, + cwd: resolveStageCwd(runId), + defaultSessionDir: deps.resolveDefaultStageSessionDir(), + }; +} + +/** + * Build a `(runId, stageId) => handle | undefined` resolver for the attach pane. + * Returns `undefined` when the stage is unknown or not revivable so the pane + * keeps its read-only transcript fallback. + */ +export function createPostMortemHandleResolver( + deps: PostMortemResolverDeps, +): (runId: string, stageId: string) => StageControlHandle | undefined { + return (runId, stageId) => { + const run = store.snapshot().runs.find((candidate) => candidate.id === runId); + const stage = run?.stages.find((candidate) => candidate.id === stageId); + if (stage === undefined) return undefined; + const result = ensurePostMortemStageHandle(runId, stage, postMortemDepsForRun(runId, deps)); + return result.ok ? result.handle : undefined; + }; +} diff --git a/packages/workflows/src/extension/workflow-tool-send.ts b/packages/workflows/src/extension/workflow-tool-send.ts index 42fadd596..5880fbbb8 100644 --- a/packages/workflows/src/extension/workflow-tool-send.ts +++ b/packages/workflows/src/extension/workflow-tool-send.ts @@ -1,4 +1,9 @@ import { stageControlRegistry } from "../runs/foreground/stage-control-registry.js"; +import { + ensurePostMortemStageHandle, + type PostMortemStageChatDeps, +} from "../runs/foreground/postmortem-stage-chat.js"; +import type { StageControlHandle } from "../runs/foreground/stage-control-registry.js"; import { store } from "../shared/store.js"; import { stageUiBroker } from "../shared/stage-ui-broker.js"; import { @@ -14,6 +19,14 @@ import { resolveToolStageTarget, } from "./workflow-targets.js"; +/** + * Optional dependencies enabling `workflow send` to revive an eligible terminal + * agent stage as a post-mortem chat when no process-local handle exists. + */ +export interface WorkflowSendDeps { + readonly resolvePostMortemDeps?: (runId: string) => PostMortemStageChatDeps; +} + function hasPayloadProperty(args: WorkflowToolArgs): boolean { return args.text !== undefined || args.response !== undefined || args.message !== undefined; } @@ -52,7 +65,10 @@ function workflowSendResult( return { action: "send", runId, stageId, delivery, status, message }; } -export async function workflowSendAction(args: WorkflowToolArgs): Promise { +export async function workflowSendAction( + args: WorkflowToolArgs, + deps: WorkflowSendDeps = {}, +): Promise { const target = resolveToolRunTarget(args, "No active run to message."); const requestedDelivery = args.delivery ?? "auto"; if (target.kind === "all") { @@ -143,7 +159,11 @@ export async function workflowSendAction(args: WorkflowToolArgs): Promise WorkflowPersistencePort | undefined, reloadWorkflowResources: () => Promise | void, ensureWorkflowResourcesLoaded: () => Promise | void = () => {}, + sendDeps: WorkflowSendDeps = {}, ): (args: WorkflowToolArgs, ctx: PiExecuteContext) => Promise { return async function executeWorkflowTool( args: WorkflowToolArgs, @@ -115,7 +116,7 @@ export function makeExecuteWorkflowTool( case "transcript": return workflowTranscriptResult(args); case "send": - return workflowSendAction(args); + return workflowSendAction(args, sendDeps); case "pause": return workflowPauseAction(args); case "reload": diff --git a/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts new file mode 100644 index 000000000..9f623ba90 --- /dev/null +++ b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts @@ -0,0 +1,172 @@ +/** + * Post-mortem stage-chat resolver. + * + * Generalizes the safe detached-handle pattern introduced by completed + * durable-workflow inspection (#1758) so any eligible terminal agent stage + * with a valid retained Atomic session can be reopened as an interactive + * post-mortem chat — from generic `/workflow attach` / `/workflow connect`, + * restored/replayed durable snapshots, and `workflow send` — not only the + * dedicated completed-inspection path. + * + * A post-mortem chat appends follow-up conversation to the stage's retained + * session (append-in-place, matching same-process handles and #1758). It never + * resumes, retries, rewinds, or otherwise mutates workflow execution: the + * handle is detached from run-level pause/resume control from birth and cannot + * pause/resume the workflow DAG. Tool side effects from follow-up turns are + * possible (the agent keeps its ordinary tools); only the workflow execution + * state is immutable. + * + * The resolver deliberately does NOT fabricate a `LiveStageRuntime`: that + * carries scheduler, store-mutation, exit, failure, and finalization + * dependencies a post-mortem conversation must never receive. + * + * cross-ref: + * - src/durable/completed-inspection.ts (authoritative completed catalog/open) + * - src/runs/foreground/stage-control-registry.ts (get-or-create ownership) + * - src/shared/session-transcript.ts (retained-session validation) + */ +import type { StageSnapshot } from "../../shared/store-types.js"; +import { isReopenableSessionTranscript } from "../../shared/session-transcript.js"; +import { createStageContext, type StageAdapters } from "./stage-runner.js"; +import { + type AgentSessionEventListener, + type StageControlHandle, + type StageControlRegistry, +} from "./stage-control-registry.js"; + +/** Why a terminal stage could not be revived as an interactive post-mortem chat. */ +export type PostMortemUnavailableReason = + | "no_adapter" + | "not_terminal" + | "no_session" + | "invalid_session"; + +export type EnsurePostMortemStageHandleResult = + | { readonly ok: true; readonly handle: StageControlHandle } + | { readonly ok: false; readonly reason: PostMortemUnavailableReason }; + +export interface PostMortemStageChatDeps { + readonly registry: StageControlRegistry; + readonly adapters?: StageAdapters; + /** Working directory used when reopening the retained session. */ + readonly cwd?: string; + /** Default stage session directory to restore after a host restart. */ + readonly defaultSessionDir?: string; +} + +/** Terminal statuses whose retained agent session may be reopened for follow-up. */ +const TERMINAL_POSTMORTEM_STATUSES = new Set(["completed"]); + +/** + * True when the snapshot is an eligible terminal agent stage: a completed stage + * that retains a `sessionFile`. Prompt/boundary/summary/skipped nodes without + * their own agent session are excluded because they have no `sessionFile`. + */ +export function isPostMortemEligibleStage(stage: StageSnapshot): boolean { + return TERMINAL_POSTMORTEM_STATUSES.has(stage.status) + && typeof stage.sessionFile === "string" + && stage.sessionFile.length > 0; +} + +/** + * Resolve an interactive post-mortem chat handle for a terminal agent stage. + * + * Reuses an existing non-disposed registry handle when present; otherwise + * validates the retained session and lazily creates a detached, single-flight + * handle keyed by the real `{ runId, stageId }`. Returns an explicit + * unavailable reason so callers can preserve the read-only transcript. + */ +export function ensurePostMortemStageHandle( + runId: string, + stage: StageSnapshot, + deps: PostMortemStageChatDeps, +): EnsurePostMortemStageHandleResult { + const existing = deps.registry.get(runId, stage.id); + if (existing !== undefined && existing.isDisposed !== true) { + return { ok: true, handle: existing }; + } + if (deps.adapters?.agentSession === undefined) return { ok: false, reason: "no_adapter" }; + if (!TERMINAL_POSTMORTEM_STATUSES.has(stage.status)) return { ok: false, reason: "not_terminal" }; + const sessionFile = stage.sessionFile; + if (typeof sessionFile !== "string" || sessionFile.length === 0) return { ok: false, reason: "no_session" }; + if (!isReopenableSessionTranscript(sessionFile)) return { ok: false, reason: "invalid_session" }; + + const adapters = deps.adapters; + const handle = deps.registry.getOrCreateDetached(runId, stage.id, () => + createPostMortemStageHandle(runId, stage, sessionFile, adapters, deps.cwd, deps.defaultSessionDir), + ); + return { ok: true, handle }; +} + +/** + * Build a lazy session-only stage-control handle that reopens `sessionFile` on + * first use and appends follow-up conversation without dispatching the + * workflow. Pause/resume of workflow execution is rejected. + */ +export function createPostMortemStageHandle( + runId: string, + stage: Pick, + sessionFile: string, + adapters: StageAdapters, + cwd: string | undefined, + defaultSessionDir: string | undefined, +): StageControlHandle { + const context = createStageContext({ + runId, + stageId: stage.id, + stageName: stage.name, + adapters, + stageOptions: { + resumeFromSessionFile: sessionFile, + ...(cwd !== undefined ? { cwd } : {}), + }, + ...(defaultSessionDir !== undefined ? { defaultSessionDir } : {}), + }); + let disposed = false; + const ensureAttached = async (): Promise => { + if (disposed) throw new Error(`Post-mortem stage chat "${stage.name}" is closed.`); + if (context.__sessionMeta().sessionFile === undefined) { + await context.__ensureSessionFromFile(sessionFile); + } + }; + return { + runId, + stageId: stage.id, + stageName: stage.name, + status: "completed", + get sessionId() { return context.__sessionMeta().sessionId ?? stage.sessionId; }, + get sessionFile() { return context.__sessionMeta().sessionFile ?? sessionFile; }, + get isStreaming() { return context.isStreaming; }, + get isDisposed() { return disposed; }, + get messages() { return context.messages; }, + get agentSession() { return context.__agentSession(); }, + async ensureAttached() { await ensureAttached(); }, + async prompt(text: string) { + await ensureAttached(); + await context.prompt(text); + }, + async steer(text: string) { + await ensureAttached(); + await context.steer(text); + }, + async followUp(text: string) { + await ensureAttached(); + await context.followUp(text); + }, + async pause() { + throw new Error("Post-mortem stage chat cannot pause or resume workflow execution."); + }, + async resume(message?: string) { + if (message !== undefined && message.trim().length > 0) { + await ensureAttached(); + await context.prompt(message); + } + }, + subscribe(listener: AgentSessionEventListener) { return context.subscribe(listener); }, + async dispose() { + if (disposed) return; + disposed = true; + await context.__dispose(); + }, + }; +} diff --git a/packages/workflows/src/runs/foreground/stage-control-registry.ts b/packages/workflows/src/runs/foreground/stage-control-registry.ts index 5125bd73c..d1cfe63f2 100644 --- a/packages/workflows/src/runs/foreground/stage-control-registry.ts +++ b/packages/workflows/src/runs/foreground/stage-control-registry.ts @@ -123,6 +123,18 @@ export interface StageControlRegistry { * keep using its direct handle reference. */ register(handle: StageControlHandle): () => void; + /** + * Atomically resolve an existing non-disposed handle for `runId + stageId` + * or create one via `create`, register it, and immediately detach it from + * run-level pause/resume control. Used by the post-mortem stage-chat + * resolver so repeated attach/send calls single-flight onto one detached + * writer per real stage instead of racing competing sessions. + */ + getOrCreateDetached( + runId: string, + stageId: string, + create: () => StageControlHandle, + ): StageControlHandle; /** * Remove this stage from run-level pause/resume aggregates while keeping * `get()` chat attachment live until the registration disposer runs. @@ -252,6 +264,24 @@ export function createStageControlRegistry(): StageControlRegistry { if (existing.size === 0) _byRun.delete(handle.runId); }; }, + getOrCreateDetached( + runId: string, + stageId: string, + create: () => StageControlHandle, + ): StageControlHandle { + const runMap = ensureRun(runId); + const existing = runMap.get(stageId); + if (existing !== undefined && existing.handle.isDisposed !== true) return existing.handle; + if (existing !== undefined) { + runMap.delete(stageId); + void Promise.resolve(existing.handle.dispose?.()).catch((err: unknown) => { + console.warn("atomic-workflows: stale stage handle dispose failed", err); + }); + } + const handle = create(); + runMap.set(handle.stageId, { handle, controlsDependencies: false }); + return handle; + }, detachControl(runId: string, stageId: string, handle?: StageControlHandle): boolean { const entry = _byRun.get(runId)?.get(stageId); if (!entry) return false; diff --git a/packages/workflows/src/shared/session-transcript.ts b/packages/workflows/src/shared/session-transcript.ts new file mode 100644 index 000000000..b1bb12f78 --- /dev/null +++ b/packages/workflows/src/shared/session-transcript.ts @@ -0,0 +1,74 @@ +/** + * Shared retained-session transcript validation. + * + * A stage snapshot's `sessionFile` is only reopenable as a post-mortem chat + * when the referenced path is an existing, readable, regular file whose JSONL + * contents parse into a genuine Atomic session with at least one usable + * context-bearing message. This guards the completed-workflow catalog and the + * post-mortem stage-chat resolver against blank / missing / truncated + * transcripts, and prevents `SessionManager.open()` from turning a missing path + * into an empty session. + * + * cross-ref: + * - src/durable/completed-catalog.ts (catalog eligibility) + * - src/runs/foreground/postmortem-stage-chat.ts (revival eligibility) + */ +import { readFileSync, statSync } from "node:fs"; + +interface SessionTranscriptEntry { + readonly type?: string; + readonly id?: string; + readonly timestamp?: string; + readonly message?: { + readonly role?: string; + readonly content?: string | object; + }; +} + +/** + * True when `path` points at a readable regular file containing a parseable + * Atomic session header plus at least one usable context message. + */ +export function isReopenableSessionTranscript(path: string): boolean { + try { + const stats = statSync(path); + if (!stats.isFile() || stats.size === 0) return false; + const lines = readFileSync(path, "utf8").split("\n").filter((line) => line.trim().length > 0); + if (lines.length < 2) return false; + const entries: SessionTranscriptEntry[] = []; + for (const line of lines) { + const parsed = JSON.parse(line) as object; + if (typeof parsed !== "object" || parsed === null) return false; + entries.push(parsed as SessionTranscriptEntry); + } + const header = entries[0]; + return header?.type === "session" && typeof header.id === "string" && entries.some(isUsableContextMessage); + } catch { + return false; + } +} + +function isUsableContextMessage(entry: SessionTranscriptEntry): boolean { + return entry.type === "message" + && typeof entry.id === "string" + && typeof entry.timestamp === "string" + && typeof entry.message?.role === "string" + && hasUsableMessageContent(entry.message.content); +} + +function hasUsableMessageContent(content: string | object | undefined): boolean { + if (typeof content === "string") return content.trim().length > 0; + return Array.isArray(content) && content.some(hasUsableContentBlock); +} + +function hasUsableContentBlock(block: object): boolean { + if (typeof block !== "object" || block === null) return false; + const contentBlock = block as { + readonly text?: string; + readonly thinking?: string; + readonly data?: string; + readonly name?: string; + }; + return [contentBlock.text, contentBlock.thinking, contentBlock.data, contentBlock.name] + .some((value) => typeof value === "string" && value.trim().length > 0); +} diff --git a/packages/workflows/src/tui/overlay-adapter.ts b/packages/workflows/src/tui/overlay-adapter.ts index 9d85f4577..dec950c11 100644 --- a/packages/workflows/src/tui/overlay-adapter.ts +++ b/packages/workflows/src/tui/overlay-adapter.ts @@ -23,7 +23,7 @@ import { WORKFLOW_STATUS_KEY } from "./workflow-status.js"; import { deriveGraphThemeFromPiTheme } from "./graph-theme.js"; import { quitRun as defaultQuitRun } from "../runs/background/quit.js"; import { stageControlRegistry as defaultStageControlRegistry } from "../runs/foreground/stage-control-registry.js"; -import type { StageControlRegistry } from "../runs/foreground/stage-control-registry.js"; +import type { StageControlHandle, StageControlRegistry } from "../runs/foreground/stage-control-registry.js"; import type { StageUiBroker } from "../shared/stage-ui-broker.js"; import type { PiCustomComponent, @@ -130,6 +130,14 @@ export interface BuildGraphOverlayAdapterOpts { * Defaults to the singleton registry registered alongside the store. */ stageControlRegistry?: StageControlRegistry; + /** + * Resolver that revives a post-mortem chat handle for an eligible terminal + * agent stage with a valid retained session but no process-local handle. + * Threaded into every `WorkflowAttachPane` so generic attach/connect and + * restored/replayed durable snapshots open as interactive follow-up chats + * instead of read-only archives. `undefined` results keep the archive. + */ + resolvePostMortemHandle?: (runId: string, stageId: string) => StageControlHandle | undefined; /** Broker used to route stage-local custom UI into attached stage chats. */ stageUiBroker?: StageUiBroker; /** @@ -149,6 +157,7 @@ export function buildGraphOverlayAdapter( buildOpts: BuildGraphOverlayAdapterOpts = {}, ): GraphOverlayPort { const registry = buildOpts.stageControlRegistry ?? defaultStageControlRegistry; + const resolvePostMortemHandle = buildOpts.resolvePostMortemHandle; const stageUiBroker = buildOpts.stageUiBroker; const terminalOutput = buildOpts.terminalOutput ?? { platform: process.platform, @@ -367,6 +376,7 @@ export function buildGraphOverlayAdapter( graphTheme: deriveGraphThemeFromPiTheme(theme), runId, stageControlRegistry: registry, + resolvePostMortemHandle, stageUiBroker, uiStatus, onClose: finish, diff --git a/packages/workflows/src/tui/workflow-attach-pane-types.ts b/packages/workflows/src/tui/workflow-attach-pane-types.ts index aca03cda8..0b772242d 100644 --- a/packages/workflows/src/tui/workflow-attach-pane-types.ts +++ b/packages/workflows/src/tui/workflow-attach-pane-types.ts @@ -3,7 +3,7 @@ import type { EditorComponent, EditorTheme, TUI } from "@earendil-works/pi-tui"; import type { Store } from "../shared/store.js"; import type { StageUiBroker } from "../shared/stage-ui-broker.js"; import type { GraphTheme } from "./graph-theme.js"; -import type { StageControlRegistry } from "../runs/foreground/stage-control-registry.js"; +import type { StageControlHandle, StageControlRegistry } from "../runs/foreground/stage-control-registry.js"; export interface AttachUiStatusSurface { setStatus?: (key: string, value: string | undefined) => void; @@ -18,6 +18,15 @@ export interface WorkflowAttachPaneOpts { * the user attaches to a node. Defaults to the singleton registry. */ stageControlRegistry?: StageControlRegistry; + /** + * Resolver that revives an interactive post-mortem chat handle for an + * eligible terminal agent stage that has a valid retained session but no + * process-local handle (generic attach/connect, restored/replayed durable + * snapshots). Returns `undefined` when the stage is not revivable, so the + * pane preserves the read-only transcript. Called only after a live + * `stageControlRegistry.get()` miss. + */ + resolvePostMortemHandle?: (runId: string, stageId: string) => StageControlHandle | undefined; /** Broker used to route stage-local custom UI such as ask_user_question into attached chats. */ stageUiBroker?: StageUiBroker; /** diff --git a/packages/workflows/src/tui/workflow-attach-pane.ts b/packages/workflows/src/tui/workflow-attach-pane.ts index 91470dee0..d4e21b039 100644 --- a/packages/workflows/src/tui/workflow-attach-pane.ts +++ b/packages/workflows/src/tui/workflow-attach-pane.ts @@ -45,6 +45,7 @@ export class WorkflowAttachPane implements Component { private theme: GraphTheme; private runId: string | null; private registry: StageControlRegistry | undefined; + private resolvePostMortemHandle?: (runId: string, stageId: string) => StageControlHandle | undefined; private stageUiBroker: StageUiBroker | undefined; private uiStatus: AttachUiStatusSurface | undefined; private onClose: () => void; @@ -82,7 +83,7 @@ export class WorkflowAttachPane implements Component { this.store = opts.store; this.theme = opts.graphTheme; this.runId = opts.runId; - this.registry = opts.stageControlRegistry; + this.registry = opts.stageControlRegistry; this.resolvePostMortemHandle = opts.resolvePostMortemHandle; this.stageUiBroker = opts.stageUiBroker; this.uiStatus = opts.uiStatus; this.onClose = opts.onClose; @@ -175,7 +176,7 @@ export class WorkflowAttachPane implements Component { : 0; this.attachedRunId = runId; this.lastAttachedStageId = stageId; - const handle: StageControlHandle | undefined = this.registry?.get(runId, stageId); + const handle = this.registry?.get(runId, stageId) ?? this.resolvePostMortemHandle?.(runId, stageId); this.chatView?.dispose(); let chatView!: StageChatView; chatView = new StageChatView({ diff --git a/test/unit/postmortem-stage-chat.test.ts b/test/unit/postmortem-stage-chat.test.ts new file mode 100644 index 000000000..d7c7ffb08 --- /dev/null +++ b/test/unit/postmortem-stage-chat.test.ts @@ -0,0 +1,188 @@ +/** + * Unit tests for the shared post-mortem stage-chat resolver. + * + * Verifies: + * - eligible completed agent stages with a valid retained session revive into + * a detached, interactive handle that appends follow-up without mutating + * run/stage status; + * - single-flight: repeated calls reuse one handle / one session create; + * - explicit unavailable reasons for non-terminal / session-less / invalid / + * adapter-less stages so callers preserve the read-only transcript. + * + * cross-ref: src/runs/foreground/postmortem-stage-chat.ts + */ +import { afterEach, beforeEach, describe, test } from "bun:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + ensurePostMortemStageHandle, + isPostMortemEligibleStage, +} from "../../packages/workflows/src/runs/foreground/postmortem-stage-chat.js"; +import { createStageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { StageAdapters } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; +import type { StageSnapshot } from "../../packages/workflows/src/shared/store-types.js"; +import { mockSession, type StageSessionRuntime } from "./executor-shared.js"; + +let tempDir = ""; +beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), "atomic-postmortem-")); }); +afterEach(() => { rmSync(tempDir, { recursive: true, force: true }); }); + +function retainedSession(name: string): string { + const path = join(tempDir, `${name}.jsonl`); + writeFileSync(path, [ + JSON.stringify({ type: "session", version: 3, id: `${name}-session`, timestamp: new Date().toISOString(), cwd: tempDir }), + JSON.stringify({ type: "message", id: `${name}-message`, parentId: null, timestamp: new Date().toISOString(), message: { role: "user", content: "Original stage request" } }), + ].join("\n") + "\n"); + return path; +} + +function completedStage(overrides: Partial = {}): StageSnapshot { + return { + id: "stage-1", + name: "final", + status: "completed", + parentIds: [], + toolEvents: [], + ...overrides, + }; +} + +function adaptersRecording(session: StageSessionRuntime, counter: { creates: number }): StageAdapters { + return { + agentSession: { + async create() { + counter.creates += 1; + return session; + }, + }, + }; +} + +describe("ensurePostMortemStageHandle", () => { + test("revives a detached interactive handle and appends follow-up without mutating status", async () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("revive"); + const promptCalls: string[] = []; + const session: StageSessionRuntime = { ...mockSession(), sessionFile, async prompt(text: string) { promptCalls.push(text); } }; + const counter = { creates: 0 }; + const stage = completedStage({ sessionFile }); + + const result = ensurePostMortemStageHandle("run-1", stage, { + registry, + adapters: adaptersRecording(session, counter), + cwd: tempDir, + }); + + assert.equal(result.ok, true); + if (!result.ok) return; + // Detached from run-level control, still resolvable as a chat handle. + assert.equal(registry.get("run-1", "stage-1"), result.handle); + assert.deepEqual(registry.run("run-1").stages(), []); + + await result.handle.prompt("What next?"); + assert.equal(counter.creates, 1); + assert.deepEqual(promptCalls, ["What next?"]); + assert.equal(stage.status, "completed"); + }); + + test("single-flights repeated calls onto one handle and one session create", async () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("single-flight"); + const counter = { creates: 0 }; + const session: StageSessionRuntime = { ...mockSession(), sessionFile }; + const stage = completedStage({ sessionFile }); + const deps = { registry, adapters: adaptersRecording(session, counter), cwd: tempDir }; + + const first = ensurePostMortemStageHandle("run-1", stage, deps); + const second = ensurePostMortemStageHandle("run-1", stage, deps); + assert.equal(first.ok && second.ok, true); + if (!first.ok || !second.ok) return; + assert.equal(first.handle, second.handle); + await first.handle.ensureAttached(); + await second.handle.ensureAttached(); + assert.equal(counter.creates, 1); + }); + + test("rejects a post-mortem pause/resume of workflow execution", async () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("no-resume"); + const session: StageSessionRuntime = { ...mockSession(), sessionFile }; + const result = ensurePostMortemStageHandle("run-1", completedStage({ sessionFile }), { + registry, + adapters: adaptersRecording(session, { creates: 0 }), + cwd: tempDir, + }); + assert.equal(result.ok, true); + if (!result.ok) return; + await assert.rejects(() => result.handle.pause()); + }); + + test("returns not_terminal for a running stage", () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("running"); + const result = ensurePostMortemStageHandle("run-1", completedStage({ status: "running", sessionFile }), { + registry, + adapters: adaptersRecording({ ...mockSession(), sessionFile }, { creates: 0 }), + }); + assert.equal(result.ok, false); + if (result.ok) return; + assert.equal(result.reason, "not_terminal"); + }); + + test("returns no_session when a completed stage has no retained session", () => { + const registry = createStageControlRegistry(); + const result = ensurePostMortemStageHandle("run-1", completedStage(), { + registry, + adapters: adaptersRecording(mockSession(), { creates: 0 }), + }); + assert.equal(result.ok, false); + if (result.ok) return; + assert.equal(result.reason, "no_session"); + }); + + test("returns invalid_session for a missing / malformed transcript", () => { + const registry = createStageControlRegistry(); + const missing = join(tempDir, "does-not-exist.jsonl"); + const result = ensurePostMortemStageHandle("run-1", completedStage({ sessionFile: missing }), { + registry, + adapters: adaptersRecording(mockSession(), { creates: 0 }), + }); + assert.equal(result.ok, false); + if (result.ok) return; + assert.equal(result.reason, "invalid_session"); + }); + + test("returns no_adapter when no agent-session adapter is available", () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("no-adapter"); + const result = ensurePostMortemStageHandle("run-1", completedStage({ sessionFile }), { registry }); + assert.equal(result.ok, false); + if (result.ok) return; + assert.equal(result.reason, "no_adapter"); + }); + + test("reuses an existing non-disposed handle instead of validating again", () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("existing"); + const counter = { creates: 0 }; + const stage = completedStage({ sessionFile }); + const deps = { registry, adapters: adaptersRecording({ ...mockSession(), sessionFile }, counter), cwd: tempDir }; + const first = ensurePostMortemStageHandle("run-1", stage, deps); + assert.equal(first.ok, true); + // Even if the session file is deleted, the existing handle is reused. + rmSync(sessionFile); + const second = ensurePostMortemStageHandle("run-1", stage, deps); + assert.equal(first.ok && second.ok && first.handle === second.handle, true); + }); +}); + +describe("isPostMortemEligibleStage", () => { + test("true only for completed stages with a retained session file", () => { + assert.equal(isPostMortemEligibleStage(completedStage({ sessionFile: "/tmp/x.jsonl" })), true); + assert.equal(isPostMortemEligibleStage(completedStage()), false); + assert.equal(isPostMortemEligibleStage(completedStage({ status: "running", sessionFile: "/tmp/x.jsonl" })), false); + assert.equal(isPostMortemEligibleStage(completedStage({ status: "failed", sessionFile: "/tmp/x.jsonl" })), false); + }); +}); diff --git a/test/unit/stage-control-registry.test.ts b/test/unit/stage-control-registry.test.ts index b2fbcfb33..61313b79c 100644 --- a/test/unit/stage-control-registry.test.ts +++ b/test/unit/stage-control-registry.test.ts @@ -247,5 +247,47 @@ describe("stageControlRegistry — resume fan-out", () => { const stages = r.run("run-1").pausedStages(); assert.equal(stages.length, 1); assert.equal(stages[0]!.stageId, "b"); + }); +}); + +describe("stageControlRegistry — getOrCreateDetached", () => { + test("creates a detached handle excluded from run-level control", () => { + const r = createStageControlRegistry(); + let creates = 0; + const handle = r.getOrCreateDetached("run-1", "a", () => { + creates += 1; + return makeHandle("run-1", "a", { status: "completed" }); + }); + assert.equal(creates, 1); + assert.equal(r.get("run-1", "a"), handle); + assert.deepEqual(r.run("run-1").stages(), []); + }); + + test("reuses an existing non-disposed handle without re-invoking the factory", () => { + const r = createStageControlRegistry(); + let creates = 0; + const factory = (): StageControlHandle => { + creates += 1; + return makeHandle("run-1", "a", { status: "completed" }); + }; + const first = r.getOrCreateDetached("run-1", "a", factory); + const second = r.getOrCreateDetached("run-1", "a", factory); + assert.equal(creates, 1); + assert.equal(first, second); + }); + + test("replaces and disposes a disposed handle", () => { + const r = createStageControlRegistry(); + let disposed = false; + const stale: StageControlHandle = { + ...makeHandle("run-1", "a", { status: "completed" }), + get isDisposed() { return true; }, + dispose() { disposed = true; }, + }; + r.register(stale); + const fresh = r.getOrCreateDetached("run-1", "a", () => makeHandle("run-1", "a", { status: "completed" })); + assert.notEqual(fresh, stale); + assert.equal(disposed, true); + assert.equal(r.get("run-1", "a"), fresh); }); }); diff --git a/test/unit/workflow-attach-pane-11.test.ts b/test/unit/workflow-attach-pane-11.test.ts new file mode 100644 index 000000000..b4fd137f7 --- /dev/null +++ b/test/unit/workflow-attach-pane-11.test.ts @@ -0,0 +1,131 @@ +// @ts-nocheck +/** + * Unit tests for `WorkflowAttachPane` post-mortem handle revival. + * + * Verifies: + * - a completed stage with no process-local handle revives an interactive + * post-mortem handle through `resolvePostMortemHandle`, and prompts reach it; + * - a live registry handle short-circuits revival (resolver not consulted); + * - an unavailable stage (resolver returns undefined) stays read-only. + * + * cross-ref: src/tui/workflow-attach-pane.ts, src/runs/foreground/postmortem-stage-chat.ts + */ + +import { describe, test } from "bun:test"; +import assert from "node:assert/strict"; +import { createStore } from "../../packages/workflows/src/shared/store.js"; +import { WorkflowAttachPane } from "../../packages/workflows/src/tui/workflow-attach-pane.js"; +import { deriveGraphTheme } from "../../packages/workflows/src/tui/graph-theme.js"; +import { createStageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { StageControlHandle } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { AgentSession } from "@bastani/atomic"; +function setupCompletedRun(store: ReturnType, runId: string) { + store.recordRunStart({ id: runId, name: "test-wf", inputs: {}, status: "completed", stages: [], startedAt: 1 }); + store.recordStageStart(runId, { + id: "stage-a", + name: "A", + status: "completed", + parentIds: [], + toolEvents: [], + result: "done", + sessionFile: "/tmp/a.jsonl", + attachable: false, + }); +} + +function makeHandle(runId: string, stageId: string, promptCalls: string[]): StageControlHandle { + return { + runId, + stageId, + stageName: `stage-${stageId}`, + status: "completed", + sessionId: undefined, + sessionFile: "/tmp/a.jsonl", + isStreaming: false, + messages: [] as AgentSession["messages"], + async ensureAttached() {}, + async prompt(text: string) { promptCalls.push(text); }, + async steer() {}, + async followUp() {}, + async pause() {}, + async resume() {}, + subscribe() { return () => {}; }, + }; +} + +async function flush(): Promise { + for (let i = 0; i < 6; i += 1) await Promise.resolve(); +} + +function submit(chatView: { handleInput(data: string): boolean }, text: string): void { + for (const ch of text) chatView.handleInput(ch); + chatView.handleInput("\r"); +} + +describe("WorkflowAttachPane post-mortem revival", () => { + test("revives a post-mortem handle when the registry misses", async () => { + const store = createStore(); + setupCompletedRun(store, "run-1"); + const registry = createStageControlRegistry(); + const promptCalls: string[] = []; + const resolverCalls: Array<[string, string]> = []; + const pane = new WorkflowAttachPane({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-1", + stageControlRegistry: registry, + resolvePostMortemHandle: (runId, stageId) => { + resolverCalls.push([runId, stageId]); + return makeHandle(runId, stageId, promptCalls); + }, + onClose: () => {}, + initialAttachStageId: "stage-a", + }); + assert.deepEqual(resolverCalls, [["run-1", "stage-a"]]); + const chatView = (pane as unknown as { chatView: { handleInput(data: string): boolean } | null }).chatView; + assert.ok(chatView, "expected an interactive stage chat"); + submit(chatView, "follow up question"); + await flush(); + assert.deepEqual(promptCalls, ["follow up question"]); + pane.dispose(); + }); + + test("uses the live registry handle and never consults the resolver", () => { + const store = createStore(); + setupCompletedRun(store, "run-1"); + const registry = createStageControlRegistry(); + const promptCalls: string[] = []; + registry.register(makeHandle("run-1", "stage-a", promptCalls)); + let resolverCalls = 0; + const pane = new WorkflowAttachPane({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-1", + stageControlRegistry: registry, + resolvePostMortemHandle: () => { resolverCalls += 1; return undefined; }, + onClose: () => {}, + initialAttachStageId: "stage-a", + }); + assert.equal(resolverCalls, 0); + assert.equal(pane._mode, "stage-chat"); + pane.dispose(); + }); + + test("keeps a read-only archive when the stage is not revivable", () => { + const store = createStore(); + setupCompletedRun(store, "run-1"); + const registry = createStageControlRegistry(); + const pane = new WorkflowAttachPane({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-1", + stageControlRegistry: registry, + resolvePostMortemHandle: () => undefined, + onClose: () => {}, + initialAttachStageId: "stage-a", + }); + assert.equal(pane._mode, "stage-chat"); + assert.equal(pane._hasChatView, true); + pane.dispose(); + }); +}); diff --git a/test/unit/workflow-tool-send-postmortem.test.ts b/test/unit/workflow-tool-send-postmortem.test.ts new file mode 100644 index 000000000..81132af1f --- /dev/null +++ b/test/unit/workflow-tool-send-postmortem.test.ts @@ -0,0 +1,108 @@ +/** + * Unit tests for `workflow send` post-mortem parity. + * + * Verifies: + * - a registry miss + valid retained session revives the stage and delivers + * the text as a conversational follow-up (not execution resume); + * - an invalid/missing retained session stays a no-op with an explicit reason; + * - run/stage status is unchanged by a post-mortem follow-up. + * + * cross-ref: src/extension/workflow-tool-send.ts, src/runs/foreground/postmortem-stage-chat.ts + */ +import { afterEach, beforeEach, describe, test } from "bun:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { workflowSendAction } from "../../packages/workflows/src/extension/workflow-tool-send.js"; +import { store } from "../../packages/workflows/src/shared/store.js"; +import { createStageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { PostMortemStageChatDeps } from "../../packages/workflows/src/runs/foreground/postmortem-stage-chat.js"; +import type { StageAdapters } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; +import { mockSession, type StageSessionRuntime } from "./executor-shared.js"; + +let tempDir = ""; +const RUN_ID = "postmortem-send-run"; + +beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), "atomic-send-postmortem-")); }); +afterEach(() => { + rmSync(tempDir, { recursive: true, force: true }); + store.removeRun(RUN_ID); +}); + +function retainedSession(name: string): string { + const path = join(tempDir, `${name}.jsonl`); + writeFileSync(path, [ + JSON.stringify({ type: "session", version: 3, id: `${name}-session`, timestamp: new Date().toISOString(), cwd: tempDir }), + JSON.stringify({ type: "message", id: `${name}-msg`, parentId: null, timestamp: new Date().toISOString(), message: { role: "user", content: "Original request" } }), + ].join("\n") + "\n"); + return path; +} + +function seedCompletedRun(sessionFile: string | undefined): void { + store.recordRunStart({ id: RUN_ID, name: "send-flow", inputs: {}, status: "completed", stages: [], startedAt: 1 }); + store.recordStageStart(RUN_ID, { + id: "stage-a", + name: "final", + status: "completed", + parentIds: [], + toolEvents: [], + result: "done", + attachable: false, + ...(sessionFile !== undefined ? { sessionFile } : {}), + }); +} + +function resolvePostMortemDeps(session: StageSessionRuntime, counter: { creates: number }): (runId: string) => PostMortemStageChatDeps { + const adapters: StageAdapters = { + agentSession: { + async create() { counter.creates += 1; return session; }, + }, + }; + return () => ({ registry: createStageControlRegistry(), adapters, cwd: tempDir }); +} + +describe("workflow send — post-mortem parity", () => { + test("revives a retained session and delivers a follow-up", async () => { + const sessionFile = retainedSession("send-ok"); + seedCompletedRun(sessionFile); + const followUps: string[] = []; + const counter = { creates: 0 }; + const session: StageSessionRuntime = { ...mockSession(), sessionFile, async followUp(text: string) { followUps.push(text); } }; + + const result = await workflowSendAction( + { runId: RUN_ID, stageId: "stage-a", text: "any regressions?" }, + { resolvePostMortemDeps: resolvePostMortemDeps(session, counter) }, + ); + + assert.equal(result.status, "ok"); + assert.equal(result.delivery, "followUp"); + assert.deepEqual(followUps, ["any regressions?"]); + assert.equal(counter.creates, 1); + assert.equal(store.runs().find((r) => r.id === RUN_ID)?.status, "completed"); + assert.equal(store.runs().find((r) => r.id === RUN_ID)?.stages[0]?.status, "completed"); + }); + + test("stays a no-op with an explicit reason when the session is invalid", async () => { + seedCompletedRun(join(tempDir, "missing.jsonl")); + const counter = { creates: 0 }; + const result = await workflowSendAction( + { runId: RUN_ID, stageId: "stage-a", text: "hello" }, + { resolvePostMortemDeps: resolvePostMortemDeps(mockSession(), counter) }, + ); + assert.equal(result.status, "noop"); + assert.equal(result.message, "No live handle for stage."); + assert.equal(counter.creates, 0); + }); + + test("stays a no-op when the stage has no retained session", async () => { + seedCompletedRun(undefined); + const counter = { creates: 0 }; + const result = await workflowSendAction( + { runId: RUN_ID, stageId: "stage-a", text: "hello" }, + { resolvePostMortemDeps: resolvePostMortemDeps(mockSession(), counter) }, + ); + assert.equal(result.status, "noop"); + assert.equal(counter.creates, 0); + }); +}); From 7973a77469e1f20abf28a52c6a2ce614f6542c0b Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Tue, 14 Jul 2026 21:47:55 -0700 Subject: [PATCH 2/7] fix(workflows): reject post-mortem execution resume Assistant-model: GPT-5.6 Sol --- .../runs/foreground/postmortem-stage-chat.ts | 7 ++--- test/ci/ci-workflow-contracts.test.ts | 4 ++- test/unit/postmortem-stage-chat.test.ts | 24 +++++++++++++---- .../workflow-tool-send-postmortem.test.ts | 26 +++++++++++++++++++ 4 files changed, 50 insertions(+), 11 deletions(-) diff --git a/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts index 9f623ba90..8cad651cc 100644 --- a/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts +++ b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts @@ -156,11 +156,8 @@ export function createPostMortemStageHandle( async pause() { throw new Error("Post-mortem stage chat cannot pause or resume workflow execution."); }, - async resume(message?: string) { - if (message !== undefined && message.trim().length > 0) { - await ensureAttached(); - await context.prompt(message); - } + async resume() { + throw new Error("Post-mortem stage chat cannot pause or resume workflow execution."); }, subscribe(listener: AgentSessionEventListener) { return context.subscribe(listener); }, async dispose() { diff --git a/test/ci/ci-workflow-contracts.test.ts b/test/ci/ci-workflow-contracts.test.ts index 7654ae771..28303268b 100644 --- a/test/ci/ci-workflow-contracts.test.ts +++ b/test/ci/ci-workflow-contracts.test.ts @@ -49,6 +49,8 @@ test("publish replaces full-suite reruns with release integrity and preserves re assert.match(workflow, /name: Reconfirm release tag is immutable[\s\S]*current_sha[\s\S]*VERIFIED_SHA/); }); +// This process-heavy contract invokes the verifier twice through temporary Git +// worktrees; native Windows Git can exceed Bun's 5s default test timeout. test("release verifier accepts generated release and rejects an extra forged file", async () => { const tag = "0.9.7-alpha.1"; const integrityWorktrees = async () => (await $`git worktree list --porcelain`.cwd(root).text()) @@ -85,4 +87,4 @@ test("release verifier accepts generated release and rejects an extra forged fil } finally { rmSync(temp, { recursive: true, force: true }); } -}); +}, 30_000); diff --git a/test/unit/postmortem-stage-chat.test.ts b/test/unit/postmortem-stage-chat.test.ts index d7c7ffb08..cd5f4394f 100644 --- a/test/unit/postmortem-stage-chat.test.ts +++ b/test/unit/postmortem-stage-chat.test.ts @@ -105,18 +105,32 @@ describe("ensurePostMortemStageHandle", () => { assert.equal(counter.creates, 1); }); - test("rejects a post-mortem pause/resume of workflow execution", async () => { + test("rejects post-mortem pause and resume without appending a prompt", async () => { const registry = createStageControlRegistry(); const sessionFile = retainedSession("no-resume"); - const session: StageSessionRuntime = { ...mockSession(), sessionFile }; - const result = ensurePostMortemStageHandle("run-1", completedStage({ sessionFile }), { + const promptCalls: string[] = []; + const counter = { creates: 0 }; + const session: StageSessionRuntime = { + ...mockSession(), + sessionFile, + async prompt(text: string) { promptCalls.push(text); }, + }; + const stage = completedStage({ sessionFile }); + const result = ensurePostMortemStageHandle("run-1", stage, { registry, - adapters: adaptersRecording(session, { creates: 0 }), + adapters: adaptersRecording(session, counter), cwd: tempDir, }); assert.equal(result.ok, true); if (!result.ok) return; - await assert.rejects(() => result.handle.pause()); + + const expected = /Post-mortem stage chat cannot pause or resume workflow execution\./; + await assert.rejects(() => result.handle.pause(), expected); + await assert.rejects(() => result.handle.resume("resume should be rejected"), expected); + assert.deepEqual(promptCalls, []); + assert.equal(counter.creates, 0); + assert.equal(result.handle.status, "completed"); + assert.equal(stage.status, "completed"); }); test("returns not_terminal for a running stage", () => { diff --git a/test/unit/workflow-tool-send-postmortem.test.ts b/test/unit/workflow-tool-send-postmortem.test.ts index 81132af1f..e893ffd4d 100644 --- a/test/unit/workflow-tool-send-postmortem.test.ts +++ b/test/unit/workflow-tool-send-postmortem.test.ts @@ -83,6 +83,32 @@ describe("workflow send — post-mortem parity", () => { assert.equal(store.runs().find((r) => r.id === RUN_ID)?.stages[0]?.status, "completed"); }); + test("rejects explicit resume without appending or mutating terminal status", async () => { + const sessionFile = retainedSession("send-no-resume"); + seedCompletedRun(sessionFile); + const promptCalls: string[] = []; + const counter = { creates: 0 }; + const session: StageSessionRuntime = { + ...mockSession(), + sessionFile, + async prompt(text: string) { promptCalls.push(text); }, + }; + + await assert.rejects( + () => workflowSendAction( + { runId: RUN_ID, stageId: "stage-a", text: "resume should be rejected", delivery: "resume" }, + { resolvePostMortemDeps: resolvePostMortemDeps(session, counter) }, + ), + /Post-mortem stage chat cannot pause or resume workflow execution\./, + ); + + assert.deepEqual(promptCalls, []); + assert.equal(counter.creates, 0); + const run = store.runs().find((candidate) => candidate.id === RUN_ID); + assert.equal(run?.status, "completed"); + assert.equal(run?.stages[0]?.status, "completed"); + }); + test("stays a no-op with an explicit reason when the session is invalid", async () => { seedCompletedRun(join(tempDir, "missing.jsonl")); const counter = { creates: 0 }; From 15fc88fbd6a422a3b077ae91c47d654671c98b37 Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Tue, 14 Jul 2026 22:41:32 -0700 Subject: [PATCH 3/7] fix(workflows): clarify terminal send delivery Assistant-model: GPT-5.6 Sol --- packages/coding-agent/docs/workflows.md | 2 +- packages/workflows/CHANGELOG.md | 2 +- .../src/extension/workflow-tool-send.ts | 11 ++++ .../workflow-tool-send-postmortem.test.ts | 57 ++++++++++++++----- 4 files changed, 57 insertions(+), 15 deletions(-) diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index e6216d555..bb046b458 100644 --- a/packages/coding-agent/docs/workflows.md +++ b/packages/coding-agent/docs/workflows.md @@ -1040,7 +1040,7 @@ Control behavior: - `stages` lists stage summaries, including flattened stages from nested `ctx.workflow(...)` imports and `sessionFile`/`transcriptPath` when a stage has a persisted session. Use `statusFilter: "all"` to include completed, failed, skipped, and pending stages. - `stage` returns details for one stage by stage id, unique prefix, or stage name, including nested child stages shown in the expanded graph and the persisted `sessionFile` when available. Abbreviated stage IDs printed in graph/control messages use this same unique-prefix resolver; collisions return an ambiguity diagnostic rather than selecting a stage. - `transcript` is reference-first with a small preview by default: it returns metadata, transcript paths, and up to 5 recent entries. For targeted lookup, quote the exact `sessionFile`/`transcriptPath` value without changing platform separators (preserve Windows backslashes), search it with `rg` or `grep`, then read only small surrounding ranges. Text results include JSON-escaped `sessionFileJson`/`transcriptPathJson` lines for copy-safe path literals. Pass explicit `tail` or `limit` to override the 5-entry preview; `tail` overrides `limit`; `includeToolOutput` includes captured snapshot tool output in snapshot transcript results. -- `send` delivery modes are `auto`, `answer`, `prompt`, `steer`, `followUp`, and `resume`. Prompt answers can include `promptId` and can carry answer content in `response`, `text`, or `message`; structured UI prompts usually prefer `response`. Follow-up messaging to completed or failed stages reuses the retained `sessionFile` when available so the conversation resumes from the archived stage transcript instead of starting empty; if no session metadata was retained, Atomic refuses the follow-up rather than silently resetting. Arbitrary `ctx.ui.custom` widget prompts require the interactive workflow graph and return a clear unsupported message when targeted through `send`. +- `send` delivery modes are `auto`, `answer`, `prompt`, `steer`, `followUp`, and `resume`. Prompt answers can include `promptId` and can carry answer content in `response`, `text`, or `message`; structured UI prompts usually prefer `response`. Follow-up messaging to completed or failed stages reuses the retained `sessionFile` when available so the conversation resumes from the archived stage transcript instead of starting empty; if no session metadata was retained, Atomic refuses the follow-up rather than silently resetting. Explicit `delivery: "resume"` or `delivery: "steer"` against a completed post-mortem stage returns a structured `noop` with guidance to use `followUp` or `prompt`; it never appends the supplied text or mutates workflow execution. Arbitrary `ctx.ui.custom` widget prompts require the interactive workflow graph and return a clear unsupported message when targeted through `send`. - `delivery: "auto"` first answers a pending prompt, then resumes paused work, then steers a streaming stage, then queues a follow-up. - `pause`, `interrupt`, and `kill` can target one top-level run or `all: true`; `stageId` cannot be combined with `all: true`. Stage-scoped controls can target a visible nested child stage from the expanded graph; Atomic routes the operation to the owning nested run internally. - `interrupt` is resumable: it pauses live work when pausable stages exist and keeps the run in live history/status. diff --git a/packages/workflows/CHANGELOG.md b/packages/workflows/CHANGELOG.md index e9852e041..5253fa8af 100644 --- a/packages/workflows/CHANGELOG.md +++ b/packages/workflows/CHANGELOG.md @@ -8,7 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added -- Unified workflow **post-mortem stage chat** across every inspection surface. Any eligible terminal agent stage with a valid retained session now reopens as an interactive follow-up conversation — not only completed-workflow inspection (#1758), but also generic `/workflow attach` / `/workflow connect`, restored/replayed durable snapshots after a process restart, and `workflow({ action: "send" })`. A shared runtime resolver (`ensurePostMortemStageHandle`) validates the retained session is an existing, readable, context-bearing Atomic transcript, then lazily reopens it through a detached, single-flight stage-control handle keyed by the real `{ runId, stageId }` (including nested/expanded child stages). Follow-up turns are appended in place to the retained session and the agent keeps its ordinary tools, while run/stage status, results, timings, checkpoints, replay metadata, and graph topology remain immutable — post-mortem chat can never resume, retry, rewind, pause, or re-dispatch workflow execution. Stages without a valid retained agent session (prompt/HIL and boundary/summary nodes, skipped nodes, non-terminal handle-less stages, and missing/malformed/deleted session files) keep the existing read-only transcript, and recoverably failed stages retain their execution-resume semantics. `workflow send` now revives such a stage on a registry miss and delivers the message as a conversational follow-up instead of reporting `No live handle for stage.` ([#1811](https://github.com/bastani-inc/atomic/issues/1811)) +- Unified workflow **post-mortem stage chat** across every inspection surface. Any eligible terminal agent stage with a valid retained session now reopens as an interactive follow-up conversation — not only completed-workflow inspection (#1758), but also generic `/workflow attach` / `/workflow connect`, restored/replayed durable snapshots after a process restart, and `workflow({ action: "send" })`. A shared runtime resolver (`ensurePostMortemStageHandle`) validates the retained session is an existing, readable, context-bearing Atomic transcript, then lazily reopens it through a detached, single-flight stage-control handle keyed by the real `{ runId, stageId }` (including nested/expanded child stages). Follow-up turns are appended in place to the retained session and the agent keeps its ordinary tools, while run/stage status, results, timings, checkpoints, replay metadata, and graph topology remain immutable — post-mortem chat can never resume, retry, rewind, pause, or re-dispatch workflow execution. Explicit `workflow send` delivery modes `resume` and `steer` on a completed post-mortem stage return a structured `noop` with follow-up guidance and do not append the supplied text. Stages without a valid retained agent session (prompt/HIL and boundary/summary nodes, skipped nodes, non-terminal handle-less stages, and missing/malformed/deleted session files) keep the existing read-only transcript, and recoverably failed stages retain their execution-resume semantics. `workflow send` now revives such a stage on a registry miss and delivers the message as a conversational follow-up instead of reporting `No live handle for stage.` ([#1811](https://github.com/bastani-inc/atomic/issues/1811)) ### Fixed diff --git a/packages/workflows/src/extension/workflow-tool-send.ts b/packages/workflows/src/extension/workflow-tool-send.ts index 5880fbbb8..45908cb8e 100644 --- a/packages/workflows/src/extension/workflow-tool-send.ts +++ b/packages/workflows/src/extension/workflow-tool-send.ts @@ -167,11 +167,22 @@ export async function workflowSendAction( if (handle === undefined) { return workflowSendResult(stageRunId, stage.stageId, requestedDelivery, "noop", "No live handle for stage."); } + // A completed post-mortem handle is not live execution: its handle-level + // resume()/pause() reject by contract. Return a structured noop with guidance + // instead of letting that rejection cross the tool boundary. Failed handles + // remain eligible for their existing recoverable execution-resume semantics. + const isTerminalPostMortemStage = handle.status === "completed"; if (requestedDelivery === "resume" || (requestedDelivery === "auto" && handle.status === "paused")) { + if (isTerminalPostMortemStage) { + return workflowSendResult(stageRunId, stage.stageId, "resume", "noop", "Cannot resume a terminal post-mortem stage; use delivery \"followUp\" or \"prompt\" to continue its retained conversation."); + } await handle.resume(text); return workflowSendResult(stageRunId, stage.stageId, "resume", "ok", "Resumed stage with message."); } if (requestedDelivery === "steer" || (requestedDelivery === "auto" && handle.isStreaming)) { + if (isTerminalPostMortemStage) { + return workflowSendResult(stageRunId, stage.stageId, "steer", "noop", "Cannot steer a terminal post-mortem stage; use delivery \"followUp\" or \"prompt\" to continue its retained conversation."); + } await handle.steer(text); return workflowSendResult(stageRunId, stage.stageId, "steer", "ok", "Steered live stage."); } diff --git a/test/unit/workflow-tool-send-postmortem.test.ts b/test/unit/workflow-tool-send-postmortem.test.ts index e893ffd4d..f5a6d3b7d 100644 --- a/test/unit/workflow-tool-send-postmortem.test.ts +++ b/test/unit/workflow-tool-send-postmortem.test.ts @@ -61,6 +61,11 @@ function resolvePostMortemDeps(session: StageSessionRuntime, counter: { creates: }; return () => ({ registry: createStageControlRegistry(), adapters, cwd: tempDir }); } +function runExecutionSnapshot(): object { + const run = store.runs().find((candidate) => candidate.id === RUN_ID); + assert.ok(run); + return structuredClone(run); +} describe("workflow send — post-mortem parity", () => { test("revives a retained session and delivers a follow-up", async () => { @@ -83,30 +88,56 @@ describe("workflow send — post-mortem parity", () => { assert.equal(store.runs().find((r) => r.id === RUN_ID)?.stages[0]?.status, "completed"); }); - test("rejects explicit resume without appending or mutating terminal status", async () => { + test("returns a structured noop for explicit resume without appending or mutating terminal status", async () => { const sessionFile = retainedSession("send-no-resume"); seedCompletedRun(sessionFile); - const promptCalls: string[] = []; + const deliveryCalls: string[] = []; const counter = { creates: 0 }; const session: StageSessionRuntime = { ...mockSession(), sessionFile, - async prompt(text: string) { promptCalls.push(text); }, + async prompt(text: string) { deliveryCalls.push(`prompt:${text}`); }, + async followUp(text: string) { deliveryCalls.push(`followUp:${text}`); }, + async steer(text: string) { deliveryCalls.push(`steer:${text}`); }, }; + const before = runExecutionSnapshot(); - await assert.rejects( - () => workflowSendAction( - { runId: RUN_ID, stageId: "stage-a", text: "resume should be rejected", delivery: "resume" }, - { resolvePostMortemDeps: resolvePostMortemDeps(session, counter) }, - ), - /Post-mortem stage chat cannot pause or resume workflow execution\./, + const result = await workflowSendAction( + { runId: RUN_ID, stageId: "stage-a", text: "resume should be rejected", delivery: "resume" }, + { resolvePostMortemDeps: resolvePostMortemDeps(session, counter) }, ); - assert.deepEqual(promptCalls, []); + assert.equal(result.status, "noop"); + assert.equal(result.delivery, "resume"); + assert.match(result.message, /Cannot resume a terminal post-mortem stage/); + assert.deepEqual(deliveryCalls, []); + assert.equal(counter.creates, 0); + assert.deepEqual(runExecutionSnapshot(), before); + }); + + test("returns a structured noop for explicit steer of a terminal stage", async () => { + const sessionFile = retainedSession("send-no-steer"); + seedCompletedRun(sessionFile); + const deliveryCalls: string[] = []; + const counter = { creates: 0 }; + const session: StageSessionRuntime = { + ...mockSession(), + sessionFile, + async prompt(text: string) { deliveryCalls.push(`prompt:${text}`); }, + async followUp(text: string) { deliveryCalls.push(`followUp:${text}`); }, + async steer(text: string) { deliveryCalls.push(`steer:${text}`); }, + }; + const before = runExecutionSnapshot(); + const result = await workflowSendAction( + { runId: RUN_ID, stageId: "stage-a", text: "steer attempt", delivery: "steer" }, + { resolvePostMortemDeps: resolvePostMortemDeps(session, counter) }, + ); + assert.equal(result.status, "noop"); + assert.equal(result.delivery, "steer"); + assert.match(result.message, /Cannot steer a terminal post-mortem stage/); + assert.deepEqual(deliveryCalls, []); assert.equal(counter.creates, 0); - const run = store.runs().find((candidate) => candidate.id === RUN_ID); - assert.equal(run?.status, "completed"); - assert.equal(run?.stages[0]?.status, "completed"); + assert.deepEqual(runExecutionSnapshot(), before); }); test("stays a no-op with an explicit reason when the session is invalid", async () => { From adaa0ca3174cabe9048948f17fedcc6940304616 Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Wed, 15 Jul 2026 08:36:24 -0700 Subject: [PATCH 4/7] fix(workflows): queue terminal auto sends Route automatic delivery on a streaming post-mortem stage to follow-up instead of rejecting it as an explicit execution steer. Assistant-model: GPT-5.6 Sol --- .../src/extension/workflow-tool-send.ts | 6 ++- .../workflow-tool-send-postmortem.test.ts | 42 ++++++++++++++++++- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/packages/workflows/src/extension/workflow-tool-send.ts b/packages/workflows/src/extension/workflow-tool-send.ts index 45908cb8e..a24d86866 100644 --- a/packages/workflows/src/extension/workflow-tool-send.ts +++ b/packages/workflows/src/extension/workflow-tool-send.ts @@ -179,13 +179,17 @@ export async function workflowSendAction( await handle.resume(text); return workflowSendResult(stageRunId, stage.stageId, "resume", "ok", "Resumed stage with message."); } - if (requestedDelivery === "steer" || (requestedDelivery === "auto" && handle.isStreaming)) { + if (requestedDelivery === "steer") { if (isTerminalPostMortemStage) { return workflowSendResult(stageRunId, stage.stageId, "steer", "noop", "Cannot steer a terminal post-mortem stage; use delivery \"followUp\" or \"prompt\" to continue its retained conversation."); } await handle.steer(text); return workflowSendResult(stageRunId, stage.stageId, "steer", "ok", "Steered live stage."); } + if (requestedDelivery === "auto" && handle.isStreaming && !isTerminalPostMortemStage) { + await handle.steer(text); + return workflowSendResult(stageRunId, stage.stageId, "steer", "ok", "Steered live stage."); + } if (requestedDelivery === "prompt") { await handle.prompt(text); return workflowSendResult(stageRunId, stage.stageId, "prompt", "ok", "Prompt sent to stage."); diff --git a/test/unit/workflow-tool-send-postmortem.test.ts b/test/unit/workflow-tool-send-postmortem.test.ts index f5a6d3b7d..2089e1d1a 100644 --- a/test/unit/workflow-tool-send-postmortem.test.ts +++ b/test/unit/workflow-tool-send-postmortem.test.ts @@ -16,7 +16,11 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { workflowSendAction } from "../../packages/workflows/src/extension/workflow-tool-send.js"; import { store } from "../../packages/workflows/src/shared/store.js"; -import { createStageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import { + createStageControlRegistry, + stageControlRegistry, + type StageControlHandle, +} from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; import type { PostMortemStageChatDeps } from "../../packages/workflows/src/runs/foreground/postmortem-stage-chat.js"; import type { StageAdapters } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; import { mockSession, type StageSessionRuntime } from "./executor-shared.js"; @@ -26,6 +30,7 @@ const RUN_ID = "postmortem-send-run"; beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), "atomic-send-postmortem-")); }); afterEach(() => { + stageControlRegistry.clear(); rmSync(tempDir, { recursive: true, force: true }); store.removeRun(RUN_ID); }); @@ -140,6 +145,41 @@ describe("workflow send — post-mortem parity", () => { assert.deepEqual(runExecutionSnapshot(), before); }); + test("queues an auto delivery to a streaming terminal post-mortem chat", async () => { + seedCompletedRun(undefined); + const deliveryCalls: string[] = []; + const handle: StageControlHandle = { + runId: RUN_ID, + stageId: "stage-a", + stageName: "final", + status: "completed", + sessionId: "retained-session", + sessionFile: undefined, + isStreaming: true, + messages: [], + async ensureAttached() {}, + async prompt(text: string) { deliveryCalls.push(`prompt:${text}`); }, + async followUp(text: string) { deliveryCalls.push(`followUp:${text}`); }, + async steer(text: string) { deliveryCalls.push(`steer:${text}`); }, + async pause() {}, + async resume() {}, + subscribe() { return () => {}; }, + }; + stageControlRegistry.register(handle); + const before = runExecutionSnapshot(); + + const result = await workflowSendAction({ + runId: RUN_ID, + stageId: "stage-a", + text: "queue after the active turn", + }); + + assert.equal(result.status, "ok"); + assert.equal(result.delivery, "followUp"); + assert.deepEqual(deliveryCalls, ["followUp:queue after the active turn"]); + assert.deepEqual(runExecutionSnapshot(), before); + }); + test("stays a no-op with an explicit reason when the session is invalid", async () => { seedCompletedRun(join(tempDir, "missing.jsonl")); const counter = { creates: 0 }; From 005c2e3967475df0a52458be591f15a22dfc8fb8 Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Wed, 15 Jul 2026 08:48:10 -0700 Subject: [PATCH 5/7] fix(workflows): bypass exact resume catalog scan Resolve exact compatible live workflow IDs before enumerating the completed durable catalog, avoiding unbounded resume latency from unrelated retained runs. Assistant-model: GPT-5.6 Sol --- .../src/extension/workflow-run-control-command.ts | 12 +++++++++--- .../workflow-run-control-completed-resume.test.ts | 9 ++++++++- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/packages/workflows/src/extension/workflow-run-control-command.ts b/packages/workflows/src/extension/workflow-run-control-command.ts index d9c6a84fa..9d5f92fb9 100644 --- a/packages/workflows/src/extension/workflow-run-control-command.ts +++ b/packages/workflows/src/extension/workflow-run-control-command.ts @@ -299,11 +299,15 @@ export async function handleRunControlCommand( fail(`Workflow ${exactBeforePreparation.id.slice(0, 8)} is already running in this session. Attach with \`/workflow connect ${exactBeforePreparation.id.slice(0, 8)}\` instead of resuming.`); return true; } + const exactLocalResolution = exactBeforePreparation !== undefined + && backend.isWorkflowLoadable(exactBeforePreparation.id) + ? resolveWorkflowResumeTarget(target, [exactBeforePreparation], [], []) + : undefined; let durable: readonly ResumableWorkflowEntry[] = []; let preparationError: string | undefined; - const needsDurablePreparation = exactBeforePreparation === undefined + const needsDurablePreparation = exactLocalResolution?.kind !== "live" && (exactBeforePreparation === undefined || (!backend.isWorkflowLoadable(exactBeforePreparation.id) - && (!exactHasPausedState || backend.hydrateResumableWorkflows !== undefined)); + && (!exactHasPausedState || backend.hydrateResumableWorkflows !== undefined))); if (needsDurablePreparation) { await ensureWorkflowResourcesVisible(); const runtime = deps.runtimeForContext(ctx); @@ -314,7 +318,9 @@ export async function handleRunControlCommand( } } const loadableRuns = topLevelWorkflowRuns(store.runs()).filter((run) => backend.isWorkflowLoadable(run.id)); - const combined = resolveWorkflowResumeTarget(target, loadableRuns, durable, backend.listCompletedWorkflows()); + const combined = exactLocalResolution?.kind === "live" + ? exactLocalResolution + : resolveWorkflowResumeTarget(target, loadableRuns, durable, backend.listCompletedWorkflows()); if (combined.kind === "ambiguous") { fail(`Ambiguous workflow prefix "${target}" matches: ${combined.matches.map((match) => `${match.name} (${match.workflowId.slice(0, 8)})`).join(", ")}`); return true; diff --git a/test/unit/workflow-run-control-completed-resume.test.ts b/test/unit/workflow-run-control-completed-resume.test.ts index e44ae513a..1065228d0 100644 --- a/test/unit/workflow-run-control-completed-resume.test.ts +++ b/test/unit/workflow-run-control-completed-resume.test.ts @@ -226,8 +226,14 @@ describe("/workflow resume completed target", () => { }); } - test("keeps exact full live ids on the existing paused resume path", async () => { + test("keeps exact full live ids on the existing paused resume path without listing completed durable runs", async () => { const backend = new InMemoryDurableBackend(); + let completedCatalogReads = 0; + const listCompletedWorkflows = backend.listCompletedWorkflows.bind(backend); + backend.listCompletedWorkflows = () => { + completedCatalogReads += 1; + return listCompletedWorkflows(); + }; setDurableBackend(backend); registerCompleted(backend, "exact-live-other-completed"); store.recordRunStart({ id: "exact-live", name: "live-flow", inputs: {}, status: "paused", stages: [], startedAt: 1, resumable: true }); @@ -238,6 +244,7 @@ describe("/workflow resume completed target", () => { assert.equal(result.errors.length, 0); assert.equal(store.runs().find((run) => run.id === "exact-live")?.status, "running"); assert.match(result.messages.join("\n"), /Resumed run exact-li/); + assert.equal(completedCatalogReads, 0, "an exact live run must bypass durable completed-catalog enumeration"); }); test("keeps recoverable failed and active-running explicit behavior unchanged", async () => { From 2f905403b83580e91e7052ae61ea93ff35e48a7f Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Wed, 15 Jul 2026 11:01:01 -0700 Subject: [PATCH 6/7] test(workflows): isolate nested attach setup Initialize slash-command resources before seeding the shared workflow store so concurrent test cleanup cannot erase the nested attach fixture during module loading. Assistant-model: GPT-5.6 Sol --- test/unit/slash-dispatch-resume.ts | 32 +++++++++++++++--------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/test/unit/slash-dispatch-resume.ts b/test/unit/slash-dispatch-resume.ts index 064894059..531da37ad 100644 --- a/test/unit/slash-dispatch-resume.ts +++ b/test/unit/slash-dispatch-resume.ts @@ -270,6 +270,22 @@ describe("/workflow resume — exact live fast path", () => { describe("/workflow attach ", () => { test.serial("routes an explicit nested stage through the root graph overlay", async () => { + let overlayOpens = 0; + const notifications: string[] = []; + const { pi, commands } = buildMockPi(); + addFactoryStubs(pi); + pi.ui = { + notify: (message: string) => { notifications.push(message); }, + setWidget: () => {}, + custom: () => { + overlayOpens += 1; + return undefined; + }, + }; + const factoryModule = await import("../../packages/workflows/src/extension/index.js"); + factoryModule.default(pi); + const handler = commands.find((command) => command.name === "workflow")!.options.handler; + const rootRunId = `attach-root-${Date.now()}`; const childRunId = `attach-child-${Date.now()}`; const nestedStageId = "nested-review"; @@ -308,22 +324,6 @@ describe("/workflow attach ", () => { }], }); - let overlayOpens = 0; - const notifications: string[] = []; - const { pi, commands } = buildMockPi(); - addFactoryStubs(pi); - pi.ui = { - notify: (message: string) => { notifications.push(message); }, - setWidget: () => {}, - custom: () => { - overlayOpens += 1; - return undefined; - }, - }; - const factoryModule = await import("../../packages/workflows/src/extension/index.js"); - factoryModule.default(pi); - const handler = commands.find((command) => command.name === "workflow")!.options.handler; - await handler(`attach ${rootRunId} ${nestedStageId}`, { hasUI: true, ui: pi.ui }); assert.equal(overlayOpens, 1); From dc4a4c004df7f2079b6940b435e80aa7b300ffe3 Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Wed, 15 Jul 2026 12:15:33 -0700 Subject: [PATCH 7/7] fix(workflows): fence post-mortem session boundaries Assistant-model: GPT-5.6 Sol --- packages/coding-agent/docs/workflows.md | 2 +- packages/workflows/CHANGELOG.md | 4 +- .../src/extension/extension-lifecycle.ts | 5 +- .../src/extension/postmortem-deps.ts | 12 +- .../extension-shutdown-postmortem.test.ts | 101 ++++++++++++++++ test/unit/postmortem-deps.test.ts | 111 ++++++++++++++++++ 6 files changed, 228 insertions(+), 7 deletions(-) create mode 100644 test/unit/extension-shutdown-postmortem.test.ts create mode 100644 test/unit/postmortem-deps.test.ts diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index fe9bc5a4f..34313017d 100644 --- a/packages/coding-agent/docs/workflows.md +++ b/packages/coding-agent/docs/workflows.md @@ -416,7 +416,7 @@ When a paused stage is resumed with a message, Atomic lets the stage answer that Durable `/workflow resume` preserves completed stage metadata, active-stage elapsed time, and graph topology. While an LM stage or task is active, repeated durable checkpoints refresh its accumulated pause-adjusted duration even when its session file does not change. Each new Atomic process that reopens the unfinished session mid-chat starts from the latest saved baseline and uses the same continuation prompt shown above, so repeated process-boundary resumes keep status, graph, stored, and lifecycle duration cumulative without double-counting pauses from earlier process segments. Replayed `ctx.stage`, `ctx.task`, `ctx.chain`, `ctx.parallel`, and child-workflow checkpoints keep their original summaries, timing, session/model metadata, and parallel fanout parentage instead of appearing as freshly flattened replay nodes. -**Post-mortem chat vs. execution resume.** These are distinct operations. *Resuming workflow execution* (`/workflow resume`) is for paused, interrupted, recoverably failed, or unfinished durable work; it may replay checkpoints, continue an incomplete stage, and dispatch remaining DAG work. *Opening a post-mortem chat* reopens one terminal agent stage's retained conversation for follow-up only — it never resumes, retries, rewinds, or otherwise changes workflow execution. Any eligible terminal agent stage with a valid retained session opens as an interactive post-mortem chat regardless of how you reach it: same-process `task`/`tasks`/`chain` stages, completed-workflow inspection, generic `/workflow attach` / `/workflow connect`, restored/replayed durable snapshots after a restart, and `workflow({ action: "send" })`. Explicit `/workflow attach ` targets are resolved through the expanded graph and routed to the child run that owns the stage while the overlay remains rooted on the requested graph; the resolved owner is preserved when sibling child workflows reuse the same local stage ID. Follow-up turns are appended in place to the stage's retained session (no separate fork), so the agent may still invoke its ordinary tools and cause side effects; only the workflow DAG, run/stage status, results, timings, checkpoints, and topology are immutable. Session teardown also owns a post-mortem session whose lazy reopen is still pending: if the host session is cleared while creation is in flight, Atomic disposes the newly created session before rejecting the stale attachment. A stage stays a **read-only transcript** when it has no valid retained agent session — prompt/HIL and boundary/summary nodes, skipped nodes without a completed conversation, non-terminal handle-less stages (another process may still own the session), and missing/malformed/deleted session files. When a known stage cannot be reopened, the attached chat shows the complete `SESSION UNAVAILABLE` explanation down to the supported 40-column minimum instead of incorrectly labeling an invalid file as an archived transcript. Recoverably failed stages keep their execution-resume semantics and are not silently reopened as post-mortem chat. +**Post-mortem chat vs. execution resume.** These are distinct operations. *Resuming workflow execution* (`/workflow resume`) is for paused, interrupted, recoverably failed, or unfinished durable work; it may replay checkpoints, continue an incomplete stage, and dispatch remaining DAG work. *Opening a post-mortem chat* reopens one terminal agent stage's retained conversation for follow-up only — it never resumes, retries, rewinds, or otherwise changes workflow execution. Any eligible terminal agent stage with a valid retained session opens as an interactive post-mortem chat regardless of how you reach it: same-process `task`/`tasks`/`chain` stages, completed-workflow inspection, generic `/workflow attach` / `/workflow connect`, restored/replayed durable snapshots after a restart, and `workflow({ action: "send" })`. Explicit `/workflow attach ` targets are resolved through the expanded graph and routed to the child run that owns the stage while the overlay remains rooted on the requested graph; the resolved owner is preserved when sibling child workflows reuse the same local stage ID. When a nested stage is reopened after a restart or from another checkout, its session cwd comes from the durable root workflow (resolved workflow cwd first, then original invocation cwd) while stage-control ownership remains with the actual child run. Follow-up turns are appended in place to the stage's retained session (no separate fork), so the agent may still invoke its ordinary tools and cause side effects; only the workflow DAG, run/stage status, results, timings, checkpoints, and topology are immutable. Every host session replacement or shutdown invalidates post-mortem handles, including a session whose lazy reopen is still pending: if creation finishes after the boundary, Atomic disposes the newly created session and rejects the already-submitted prompt before it can execute. A stage stays a **read-only transcript** when it has no valid retained agent session — prompt/HIL and boundary/summary nodes, skipped nodes without a completed conversation, non-terminal handle-less stages (another process may still own the session), and missing/malformed/deleted session files. When a known stage cannot be reopened, the attached chat shows the complete `SESSION UNAVAILABLE` explanation down to the supported 40-column minimum instead of incorrectly labeling an invalid file as an archived transcript. Recoverably failed stages keep their execution-resume semantics and are not silently reopened as post-mortem chat. Workflow stage sessions and first-party subagent transcripts created inside them are classified as **internal** at creation and excluded from the standard `/resume`, `atomic -r`, `--continue`, and global history surfaces. Fork-context stages and subagents inherit the owning run/stage marker in their initial JSONL header, avoiding a briefly visible ordinary session. They remain resumable and inspectable through the workflow-specific commands and tool actions shown here (`/workflow resume`, `/workflow attach`, `workflow({ action: "status" | "stages" | "stage" | "resume" })`), which read the run/stage store and its `sessionFile` links directly. Passing a stage session's file path to `--session` still opens it explicitly. Classification requires exact `internal: true` plus complete run/stage metadata; malformed legacy markers and ordinary user forks remain in standard history. Legacy workflow sessions created before this marker behavior lack provable ownership and continue to appear until they age out. diff --git a/packages/workflows/CHANGELOG.md b/packages/workflows/CHANGELOG.md index 907524d3f..d9487e7e8 100644 --- a/packages/workflows/CHANGELOG.md +++ b/packages/workflows/CHANGELOG.md @@ -12,8 +12,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed -- Fixed a session-boundary race in lazy post-mortem stage-chat attachment so clearing the stage-control registry while retained-session creation is pending disposes the newly created session and rejects the stale attachment instead of leaking an unowned SDK session. -- Fixed explicit `/workflow attach ` commands to resolve expanded child stages through their owning nested run, preserve that owner through overlay retargeting when sibling child runs reuse the same local stage ID, and retain post-mortem resolver failure reasons so invalid or unavailable sessions render a complete actionable `SESSION UNAVAILABLE` explanation—even at the supported 40-column minimum—instead of a misleading archived-transcript label. +- Fixed session-boundary races in lazy post-mortem stage-chat attachment so every host session replacement or shutdown (`new`, `resume`, `fork`, `reload`, and `quit`) synchronously invalidates detached handles; a retained-session creation that finishes after the boundary is disposed and its already-submitted prompt is rejected instead of executing in the replacement session or leaking an unowned SDK session. +- Fixed explicit `/workflow attach ` commands to resolve expanded child stages through their owning nested run, preserve that owner through overlay retargeting when sibling child runs reuse the same local stage ID, and retain post-mortem resolver failure reasons so invalid or unavailable sessions render a complete actionable `SESSION UNAVAILABLE` explanation—even at the supported 40-column minimum—instead of a misleading archived-transcript label. Reopened nested terminal stages now restore cwd from their durable root workflow metadata (preferring the resolved workflow cwd, then the original invocation cwd) rather than silently falling back to the current review checkout. ## [0.9.9] - 2026-07-15 diff --git a/packages/workflows/src/extension/extension-lifecycle.ts b/packages/workflows/src/extension/extension-lifecycle.ts index 3294fa960..e919b6f3f 100644 --- a/packages/workflows/src/extension/extension-lifecycle.ts +++ b/packages/workflows/src/extension/extension-lifecycle.ts @@ -140,8 +140,11 @@ export function registerWorkflowLifecycleHandlers( // `/workflow kill`. Durable-progress workflows stay available through // `/workflow resume`; stage handles are disposed after being paused. quitAllRuns({ store, stageControlRegistry }); - stageControlRegistry.clear(); } + // Every host-session boundary invalidates detached lazy handles. Clearing + // synchronously marks pending session creation as disposed before it can + // attach to the replacement session and submit an already-queued prompt. + stageControlRegistry.clear(); deps.storeWidgetRef.current?.(); deps.storeWidgetRef.current = null; runtimeState.resetWorkflowDiscoveryForSession(); diff --git a/packages/workflows/src/extension/postmortem-deps.ts b/packages/workflows/src/extension/postmortem-deps.ts index 094d9bb0f..96d52c4dd 100644 --- a/packages/workflows/src/extension/postmortem-deps.ts +++ b/packages/workflows/src/extension/postmortem-deps.ts @@ -27,11 +27,17 @@ export interface PostMortemResolverDeps { readonly resolveDefaultStageSessionDir: () => string | undefined; } -/** Persisted original/resolved cwd for a durable run, when still available. */ +/** Persisted original/resolved cwd for a durable run tree, when still available. */ function resolveStageCwd(runId: string): string | undefined { try { - const handle = getDurableBackend().getWorkflow(runId); - return handle?.workflowCwd ?? handle?.invocationCwd ?? undefined; + const backend = getDurableBackend(); + const owningHandle = backend.getWorkflow(runId); + const run = store.snapshot().runs.find((candidate) => candidate.id === runId); + const rootRunId = run?.rootRunId ?? owningHandle?.rootWorkflowId; + const cwdHandle = rootRunId === undefined + ? owningHandle + : backend.getWorkflow(rootRunId) ?? owningHandle; + return cwdHandle?.workflowCwd ?? cwdHandle?.invocationCwd ?? undefined; } catch { return undefined; } diff --git a/test/unit/extension-shutdown-postmortem.test.ts b/test/unit/extension-shutdown-postmortem.test.ts new file mode 100644 index 000000000..bb28e1bfb --- /dev/null +++ b/test/unit/extension-shutdown-postmortem.test.ts @@ -0,0 +1,101 @@ +import { afterEach, test } from "bun:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import factory, { type ExtensionAPI } from "../../packages/workflows/src/extension/index.js"; +import { ensurePostMortemStageHandle } from "../../packages/workflows/src/runs/foreground/postmortem-stage-chat.js"; +import { stageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { StageSessionRuntime } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; +import { mockSession } from "./executor-shared.js"; + +type SessionShutdownHandler = (event: { readonly reason: string }) => unknown; + +function captureSessionShutdown(): SessionShutdownHandler { + let shutdown: SessionShutdownHandler | undefined; + const pi: ExtensionAPI = { + registerTool: () => undefined, + registerCommand: () => undefined, + registerMessageRenderer: () => undefined, + registerFlag: () => undefined, + registerShortcut: () => undefined, + on: (event, handler) => { + if (event === "session_shutdown") shutdown = handler as SessionShutdownHandler; + }, + disableAsyncDiscovery: true, + }; + factory(pi); + assert.notEqual(shutdown, undefined); + return shutdown!; +} + +afterEach(() => stageControlRegistry.clear()); + +test("every non-quit session shutdown invalidates a post-mortem prompt whose session creation is pending", async () => { + for (const reason of ["new", "resume", "fork", "reload"] as const) { + stageControlRegistry.clear(); + const root = mkdtempSync(join(tmpdir(), "atomic-shutdown-postmortem-")); + try { + const sessionFile = join(root, `${reason}.jsonl`); + writeFileSync(sessionFile, [ + JSON.stringify({ + type: "session", + version: 3, + id: `${reason}-session`, + timestamp: new Date().toISOString(), + cwd: root, + }), + JSON.stringify({ + type: "message", + id: `${reason}-message`, + parentId: null, + timestamp: new Date().toISOString(), + message: { role: "user", content: "Original stage request" }, + }), + ].join("\n") + "\n"); + const creationStarted = Promise.withResolvers(); + const created = Promise.withResolvers(); + let disposeCalls = 0; + let promptCalls = 0; + const result = ensurePostMortemStageHandle("run-1", { + id: "stage-1", + name: "completed-stage", + status: "completed", + parentIds: [], + toolEvents: [], + sessionFile, + }, { + registry: stageControlRegistry, + cwd: root, + adapters: { + agentSession: { + async create() { + creationStarted.resolve(); + return created.promise; + }, + }, + }, + }); + assert.equal(result.ok, true); + if (!result.ok) continue; + + const submittedPrompt = result.handle.prompt("must not cross the host-session boundary"); + await creationStarted.promise; + await Promise.resolve(captureSessionShutdown()({ reason })); + created.resolve({ + ...mockSession(), + sessionFile, + async prompt() { promptCalls += 1; }, + dispose() { disposeCalls += 1; }, + }); + + await assert.rejects(submittedPrompt, /session has been disposed/); + assert.equal(promptCalls, 0); + assert.equal(disposeCalls, 1); + assert.equal(result.handle.isDisposed, true); + assert.equal(stageControlRegistry.get("run-1", "stage-1"), undefined); + } finally { + rmSync(root, { recursive: true, force: true }); + } + } +}); diff --git a/test/unit/postmortem-deps.test.ts b/test/unit/postmortem-deps.test.ts new file mode 100644 index 000000000..d1d57757d --- /dev/null +++ b/test/unit/postmortem-deps.test.ts @@ -0,0 +1,111 @@ +import { afterEach, beforeEach, test } from "bun:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { InMemoryDurableBackend } from "../../packages/workflows/src/durable/backend.js"; +import { setDurableBackend } from "../../packages/workflows/src/durable/factory.js"; +import { postMortemDepsForRun } from "../../packages/workflows/src/extension/postmortem-deps.js"; +import { ensurePostMortemStageHandle } from "../../packages/workflows/src/runs/foreground/postmortem-stage-chat.js"; +import { stageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { StageSessionCreateOptions } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; +import { store } from "../../packages/workflows/src/shared/store.js"; +import { mockSession } from "./executor-shared.js"; + +let tempRoot = ""; + +beforeEach(() => { + tempRoot = mkdtempSync(join(tmpdir(), "atomic-nested-postmortem-")); + stageControlRegistry.clear(); + store.clear(); +}); + +afterEach(() => { + stageControlRegistry.clear(); + store.clear(); + setDurableBackend(undefined); + rmSync(tempRoot, { recursive: true, force: true }); +}); + +test("nested post-mortem chat reopens in the durable root cwd while retaining child ownership", async () => { + for (const durableCwdField of ["workflowCwd", "invocationCwd"] as const) { + stageControlRegistry.clear(); + store.clear(); + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + const durableCwd = join(tempRoot, durableCwdField); + backend.registerWorkflow({ + workflowId: "root-run", + rootWorkflowId: "root-run", + name: "root-workflow", + inputs: {}, + createdAt: 1, + status: "completed", + invocationCwd: durableCwdField === "invocationCwd" ? durableCwd : join(tempRoot, "invocation"), + ...(durableCwdField === "workflowCwd" ? { workflowCwd: durableCwd } : {}), + }); + assert.equal(backend.getWorkflow("child-run"), undefined); + + const sessionFile = join(tempRoot, `${durableCwdField}.jsonl`); + writeFileSync(sessionFile, [ + JSON.stringify({ + type: "session", + version: 3, + id: `${durableCwdField}-session`, + timestamp: new Date().toISOString(), + cwd: durableCwd, + }), + JSON.stringify({ + type: "message", + id: `${durableCwdField}-message`, + parentId: null, + timestamp: new Date().toISOString(), + message: { role: "user", content: "Original stage request" }, + }), + ].join("\n") + "\n"); + const stage = { + id: "duplicate-stage-id", + name: "nested-completed-stage", + status: "completed" as const, + parentIds: [], + toolEvents: [], + sessionFile, + }; + store.recordRunStart({ + id: "child-run", + name: "nested-workflow", + inputs: {}, + status: "completed", + stages: [stage], + startedAt: 2, + endedAt: 3, + parentRunId: "root-run", + parentStageId: "nested-call", + rootRunId: "root-run", + }); + + let createOptions: StageSessionCreateOptions | undefined; + const deps = postMortemDepsForRun("child-run", { + adapters: { + agentSession: { + async create(options) { + createOptions = options; + return { ...mockSession(), sessionFile }; + }, + }, + }, + resolveDefaultStageSessionDir: () => undefined, + }); + assert.equal(deps.cwd, durableCwd); + assert.notEqual(deps.cwd, process.cwd()); + + const result = ensurePostMortemStageHandle("child-run", stage, deps); + assert.equal(result.ok, true); + if (!result.ok) continue; + await result.handle.ensureAttached(); + + assert.equal(createOptions?.cwd, durableCwd); + assert.equal(stageControlRegistry.get("child-run", "duplicate-stage-id"), result.handle); + assert.equal(stageControlRegistry.get("root-run", "duplicate-stage-id"), undefined); + } +});