diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index 654ce3989..34313017d 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" })`. 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. 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. @@ -1038,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 36f184fa6..d9487e7e8 100644 --- a/packages/workflows/CHANGELOG.md +++ b/packages/workflows/CHANGELOG.md @@ -6,6 +6,15 @@ 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. 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 + +- 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 ### Changed 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..e6d9e0db9 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 { PostMortemHandleResolution } from "../tui/workflow-attach-pane-types.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) => PostMortemHandleResolution, ): 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-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/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..96d52c4dd --- /dev/null +++ b/packages/workflows/src/extension/postmortem-deps.ts @@ -0,0 +1,73 @@ +/** + * 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 EnsurePostMortemStageHandleResult, + type PostMortemStageChatDeps, +} from "../runs/foreground/postmortem-stage-chat.js"; +import { stageControlRegistry } 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 tree, when still available. */ +function resolveStageCwd(runId: string): string | undefined { + try { + 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; + } +} + +/** 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) => result | undefined` resolver for the attach pane. + * Unknown stages return `undefined`; known but non-revivable stages retain the + * resolver's explicit reason so the pane can explain why chat is unavailable. + */ +export function createPostMortemHandleResolver( + deps: PostMortemResolverDeps, +): (runId: string, stageId: string) => EnsurePostMortemStageHandleResult | 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; + return ensurePostMortemStageHandle(runId, stage, postMortemDepsForRun(runId, deps)); + }; +} diff --git a/packages/workflows/src/extension/workflow-run-control-command.ts b/packages/workflows/src/extension/workflow-run-control-command.ts index 58e8b1348..6f867f4e7 100644 --- a/packages/workflows/src/extension/workflow-run-control-command.ts +++ b/packages/workflows/src/extension/workflow-run-control-command.ts @@ -29,15 +29,6 @@ import { export type { WorkflowRunControlDeps } from "./workflow-durable-resume-command.js"; -function resolveAttachStageId(runId: string, stageTarget: string | undefined): string | undefined | false { - if (!stageTarget) return undefined; - const run = store.runs().find((r) => r.id === runId); - if (!run) return undefined; - const exact = run.stages.find((s) => s.id === stageTarget); - const prefix = exact ?? run.stages.find((s) => s.id.startsWith(stageTarget)); - const byName = prefix ?? run.stages.find((s) => s.name === stageTarget); - return byName?.id ?? false; -} export async function handleRunControlCommand( @@ -353,13 +344,15 @@ export async function handleRunControlCommand( runId = resolved.runId; } if (action === "attach") { - const stageId = resolveAttachStageId(runId, stageTarget); - if (stageId === false) { - fail(`Stage not found in run ${runId.slice(0, 8)}: ${stageTarget}`); + const resolvedStage = resolveStageTarget(runId, stageTarget); + if (!resolvedStage.ok) { + fail(resolvedStage.message); return true; } + const stageId = resolvedStage.stageId; + const stageRunId = resolvedStage.runId ?? runId; if (failHeadlessAttachCommand("attach", runId, stageId)) return true; - if (policy.allowInputPicker) deps.overlay.open(runId, overlaySurfaceFromContext(ctx), stageId); + if (policy.allowInputPicker) deps.overlay.open(runId, overlaySurfaceFromContext(ctx), stageId, stageRunId); print(stageId ? `Attached to ${runId.slice(0, 8)} stage ${stageId.slice(0, 8)}. ctrl+d return to graph · esc close.` : `Attached to ${runId.slice(0, 8)}. ↵ chat · ctrl+d detach.`); return true; } @@ -376,7 +369,7 @@ export async function handleRunControlCommand( fail(result.reason === "not_found" ? `Run not found: ${stageRunId.slice(0, 8)}` : result.reason === "already_ended" ? `Run ${stageRunId.slice(0, 8)} already ended.` : result.reason === "no_active_stages" ? `No pausable stages on run ${stageRunId.slice(0, 8)}.` : `Stage not found: ${stageTarget ?? "(unknown)"}`); return true; } - if (policy.allowInputPicker) deps.overlay.open(runId, overlaySurfaceFromContext(ctx), stageId); + if (policy.allowInputPicker) deps.overlay.open(runId, overlaySurfaceFromContext(ctx), stageId, stageRunId); print(result.paused.length === 0 ? `No stages were paused on run ${stageRunId.slice(0, 8)}.` : `Paused ${result.paused.length} stage(s) on run ${stageRunId.slice(0, 8)}: ${result.paused.map((s) => s.name).join(", ")}`); return true; } @@ -412,7 +405,7 @@ export async function handleRunControlCommand( print(result.message ?? `Snapshot available: run ${result.runId} (${result.snapshot.name}) — status: ${result.snapshot.status}, stages: ${result.snapshot.stages.length}`); return true; } - if (!message && stageId && policy.allowInputPicker) deps.overlay.open(runId, overlaySurfaceFromContext(ctx), stageId); + if (!message && stageId && policy.allowInputPicker) deps.overlay.open(runId, overlaySurfaceFromContext(ctx), stageId, stageRunId); if (result.resumed.length === 0) { const runLevelResumed = hadPausedRunState && !hadPausedStageState && stageId === undefined && result.snapshot.status === "running"; runLevelResumed ? print(`Resumed run ${stageRunId.slice(0, 8)}.`) : fail(`No paused stages on run ${stageRunId.slice(0, 8)}.`); diff --git a/packages/workflows/src/extension/workflow-tool-send.ts b/packages/workflows/src/extension/workflow-tool-send.ts index 42fadd596..a24d86866 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,15 +159,34 @@ 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..8cad651cc --- /dev/null +++ b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts @@ -0,0 +1,169 @@ +/** + * 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() { + throw new Error("Post-mortem stage chat cannot pause or resume workflow execution."); + }, + 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/runs/foreground/stage-runner-controller.ts b/packages/workflows/src/runs/foreground/stage-runner-controller.ts index 5b91dce64..c74eebda9 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-controller.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-controller.ts @@ -3,7 +3,7 @@ import type { StageContext, StageExecutionMeta, StageOptions, StageSendUserMessa import { buildModelCandidatesFromCatalog, errorMessage, isRetryableModelFailure, workflowModelId, type WorkflowResolvedModelCandidate } from "../shared/model-fallback.js"; import { WorkflowPromptModelFailure, lastAssistantTextFromSession, latestTerminalAssistantFailureSince } from "./stage-runner-messages.js"; import { missingAdapter, stripWorkflowOnlyOptions, unavailableSync } from "./stage-runner-options.js"; -import { asAgentSession, disposeStageSession, normalizeSessionCreateResult } from "./stage-runner-session.js"; +import { asAgentSession, attachCreatedStageSession, disposeStageSession, normalizeSessionCreateResult } from "./stage-runner-session.js"; import { structuredOutputToolErrorFromEvent } from "./stage-runner-structured-output.js"; import { sendStageUserMessage } from "./stage-runner-send-user-message.js"; import { nextResumedContextOverflowFallbackIndex, terminatingToolCallId, unresolvedContextOverflowFailure, unresolvedContextOverflowMessage } from "./stage-runner-unresolved-overflow.js"; @@ -310,7 +310,7 @@ export class StageSessionController { { ...this.meta, stageOptions }, ) : missingAdapter(consumer); - return this.attachSession(created); + return attachCreatedStageSession(created, this.disposed, this.opts.stageName, (result) => this.attachSession(result)); } private attachSession(created: StageSessionRuntime | StageSessionCreateResult): StageSessionRuntime { diff --git a/packages/workflows/src/runs/foreground/stage-runner-session.ts b/packages/workflows/src/runs/foreground/stage-runner-session.ts index e9efd4c10..c9bbff6ca 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-session.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-session.ts @@ -53,3 +53,22 @@ export function normalizeSessionCreateResult( if ("session" in created) return created; return { session: created }; } + +export function attachCreatedStageSession( + created: StageSessionRuntime | StageSessionCreateResult, + disposed: boolean, + stageName: string, + attach: (result: StageSessionCreateResult) => T, +): T | Promise { + const result = normalizeSessionCreateResult(created); + if (!disposed) return attach(result); + return rejectDisposedCreatedSession(result, stageName); +} + +async function rejectDisposedCreatedSession( + result: StageSessionCreateResult, + stageName: string, +): Promise { + await disposeStageSession(result.session); + throw new Error(`atomic-workflows: stage "${stageName}" session has been disposed`); +} 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..3c5bdf632 100644 --- a/packages/workflows/src/tui/overlay-adapter.ts +++ b/packages/workflows/src/tui/overlay-adapter.ts @@ -25,6 +25,7 @@ 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 { StageUiBroker } from "../shared/stage-ui-broker.js"; +import type { PostMortemHandleResolution } from "./workflow-attach-pane-types.js"; import type { PiCustomComponent, PiCustomOverlayFactoryTui, @@ -64,12 +65,12 @@ export interface OverlayPiSurface { * `toggle(runId)`— show if hidden, hide if visible, create if absent. * `close()` — permanently dismiss. * - * Optional `stageId` (on `open`) opens directly on the stage-chat - * surface for that node — used by `/workflow attach ` + * Optional `stageId` and owning `stageRunId` (on `open`) open directly on + * that exact stage-chat node — used by `/workflow attach ` * and the picker overlay's connect-to-stage flow. */ export interface GraphOverlayPort { - open(runId: string | null, surface?: OverlayPiSurface, stageId?: string): void; + open(runId: string | null, surface?: OverlayPiSurface, stageId?: string, stageRunId?: string): void; toggle(runId: string | null, surface?: OverlayPiSurface): void; close(): void; } @@ -130,6 +131,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; + * unavailable results provide the pane's actionable read-only reason. + */ + resolvePostMortemHandle?: (runId: string, stageId: string) => PostMortemHandleResolution; /** Broker used to route stage-local custom UI into attached stage chats. */ stageUiBroker?: StageUiBroker; /** @@ -149,6 +158,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, @@ -305,13 +315,14 @@ export function buildGraphOverlayAdapter( runId: string | null, surface?: OverlayPiSurface, stageId?: string, + stageRunId?: string, ): void { const ui = surface?.ui ?? pi.ui; observeHostCustomUi(ui); // Already mounted but hidden — flip visibility without remounting. if (mounted && currentHandle?.isHidden()) { - currentView?.retarget(runId, stageId); + currentView?.retarget(runId, stageId, stageRunId); currentView?.setVisible(true); updateTerminalAutowrap(true); updateMouseScrollTracking(currentView?.wantsMouseScrollTracking() ?? true); @@ -320,7 +331,7 @@ export function buildGraphOverlayAdapter( return; } if (mounted) { - currentView?.retarget(runId, stageId); + currentView?.retarget(runId, stageId, stageRunId); updateTerminalAutowrap(true); updateMouseScrollTracking(currentView?.wantsMouseScrollTracking() ?? true); // Restore keyboard focus to the visible overlay after retargeting. @@ -367,12 +378,14 @@ export function buildGraphOverlayAdapter( graphTheme: deriveGraphThemeFromPiTheme(theme), runId, stageControlRegistry: registry, + resolvePostMortemHandle, stageUiBroker, uiStatus, onClose: finish, onHide: hideMounted, onQuit: quitRun, initialAttachStageId: stageId, + initialAttachRunId: stageRunId, piTui: tui, piTheme: theme, piKeybindings: keybindings, diff --git a/packages/workflows/src/tui/stage-chat-view-archive-history.ts b/packages/workflows/src/tui/stage-chat-view-archive-history.ts index d490664e8..a22094733 100644 --- a/packages/workflows/src/tui/stage-chat-view-archive-history.ts +++ b/packages/workflows/src/tui/stage-chat-view-archive-history.ts @@ -13,6 +13,21 @@ import { import type { StageChatViewContext } from "./stage-chat-view-types.js"; import { hexToAnsi, RESET } from "./color-utils.js"; +function postMortemUnavailableMessage(reason: StageChatViewContext["postMortemUnavailableReason"]): string | undefined { + switch (reason) { + case "no_adapter": + return "Post-mortem chat is unavailable because no agent session adapter is configured."; + case "not_terminal": + return "Post-mortem chat is available only after the stage completes."; + case "no_session": + return "No retained agent session is available for this stage."; + case "invalid_session": + return "The retained session is missing, unreadable, or invalid. Check that the session file still exists and is readable."; + case undefined: + return undefined; + } +} + export function renderReadOnlyArchiveBody( ctx: StageChatViewContext, width: number, @@ -24,28 +39,33 @@ export function renderReadOnlyArchiveBody( } const t = ctx.theme; - const calloutRows = 5; - const transcriptBudget = Math.max(1, budget - calloutRows); - const lines = ctx.chatHost.renderBody(width, transcriptBudget); + const unavailableMessage = postMortemUnavailableMessage(ctx.postMortemUnavailableReason); const callout: string[] = []; callout.push(blankLine(width)); callout.push( ...bannerLines( ctx, width, - "info", - "◌", - "READ-ONLY SESSION", - stage?.sessionFile ? "archived transcript" : "no live chat session", + unavailableMessage === undefined ? "info" : "warning", + unavailableMessage === undefined ? "◌" : "!", + unavailableMessage === undefined ? "READ-ONLY SESSION" : "SESSION UNAVAILABLE", + unavailableMessage === undefined + ? stage?.sessionFile ? "archived transcript" : "no live chat session" + : "post-mortem chat cannot be reopened", ), ); callout.push( ...new Text( - paint("This node is no longer attached to a live chat session.", t.textMuted), + paint( + unavailableMessage ?? "This node is no longer attached to a live chat session.", + t.textMuted, + ), 2, 0, ).render(width), ); + const transcriptBudget = Math.max(0, budget - callout.length); + const lines = transcriptBudget > 0 ? ctx.chatHost.renderBody(width, transcriptBudget) : []; lines.push(...callout); while (lines.length < budget) lines.push(blankLine(width)); if (lines.length > budget) lines.length = budget; diff --git a/packages/workflows/src/tui/stage-chat-view-state.ts b/packages/workflows/src/tui/stage-chat-view-state.ts index ec35bd7bb..813dfdaca 100644 --- a/packages/workflows/src/tui/stage-chat-view-state.ts +++ b/packages/workflows/src/tui/stage-chat-view-state.ts @@ -45,6 +45,7 @@ export function initializeStageChatView( ctx.stageId = opts.stageId; ctx.workflowName = opts.workflowName; ctx.handle = opts.handle; + ctx.postMortemUnavailableReason = opts.postMortemUnavailableReason; ctx.onDetach = opts.onDetach; ctx.onClose = opts.onClose; ctx.requestRender = opts.requestRender; diff --git a/packages/workflows/src/tui/stage-chat-view-types.ts b/packages/workflows/src/tui/stage-chat-view-types.ts index f14def351..8ec062692 100644 --- a/packages/workflows/src/tui/stage-chat-view-types.ts +++ b/packages/workflows/src/tui/stage-chat-view-types.ts @@ -6,6 +6,7 @@ import type { } from "@bastani/atomic"; import type { EditorComponent, EditorTheme, TUI } from "@earendil-works/pi-tui"; import type { StageControlHandle } from "../runs/foreground/stage-control-registry.js"; +import type { PostMortemUnavailableReason } from "../runs/foreground/postmortem-stage-chat.js"; import type { Store } from "../shared/store.js"; import type { MountedStageCustomUi, @@ -37,6 +38,8 @@ export interface StageChatViewOpts { * inspect-only (settled stage with no live handle). */ handle?: StageControlHandle; + /** Why a retained terminal stage could not be reopened for follow-up chat. */ + postMortemUnavailableReason?: PostMortemUnavailableReason; /** Called when the user presses Ctrl+D (back to graph). */ onDetach: (reason?: StageChatDetachReason, metadata?: StageChatDetachMetadata) => void; /** Called when the user presses Escape (close the whole popup). */ @@ -127,6 +130,7 @@ export interface StageChatViewContext { stageId: string; workflowName: string; handle: StageControlHandle | undefined; + postMortemUnavailableReason: PostMortemUnavailableReason | undefined; onDetach: (reason?: StageChatDetachReason, metadata?: StageChatDetachMetadata) => void; onClose: () => void; requestRender: (() => void) | undefined; diff --git a/packages/workflows/src/tui/stage-chat-view.ts b/packages/workflows/src/tui/stage-chat-view.ts index 554bcf751..4d0fa973f 100644 --- a/packages/workflows/src/tui/stage-chat-view.ts +++ b/packages/workflows/src/tui/stage-chat-view.ts @@ -74,6 +74,7 @@ export class StageChatView implements Component, Focusable { private stageId!: StageChatViewContext["stageId"]; private workflowName!: StageChatViewContext["workflowName"]; private handle!: StageChatViewContext["handle"]; + private postMortemUnavailableReason!: StageChatViewContext["postMortemUnavailableReason"]; private onDetach!: StageChatViewContext["onDetach"]; private onClose!: StageChatViewContext["onClose"]; private requestRender!: StageChatViewContext["requestRender"]; @@ -208,6 +209,7 @@ export class StageChatView implements Component, Focusable { void this.stageId; void this.workflowName; void this.handle; + void this.postMortemUnavailableReason; void this.onDetach; void this.onClose; void this.requestRender; diff --git a/packages/workflows/src/tui/workflow-attach-pane-handle.ts b/packages/workflows/src/tui/workflow-attach-pane-handle.ts new file mode 100644 index 000000000..97235ef00 --- /dev/null +++ b/packages/workflows/src/tui/workflow-attach-pane-handle.ts @@ -0,0 +1,26 @@ +import type { PostMortemUnavailableReason } from "../runs/foreground/postmortem-stage-chat.js"; +import type { + StageControlHandle, + StageControlRegistry, +} from "../runs/foreground/stage-control-registry.js"; +import type { PostMortemHandleResolution } from "./workflow-attach-pane-types.js"; + +interface AttachedStageHandleResolution { + readonly handle?: StageControlHandle; + readonly postMortemUnavailableReason?: PostMortemUnavailableReason; +} + +export function resolveAttachedStageHandle( + registry: StageControlRegistry | undefined, + resolvePostMortemHandle: ((runId: string, stageId: string) => PostMortemHandleResolution) | undefined, + runId: string, + stageId: string, +): AttachedStageHandleResolution { + const liveHandle = registry?.get(runId, stageId); + if (liveHandle !== undefined) return { handle: liveHandle }; + const postMortem = resolvePostMortemHandle?.(runId, stageId); + if (postMortem?.ok === true) return { handle: postMortem.handle }; + return postMortem?.ok === false + ? { postMortemUnavailableReason: postMortem.reason } + : {}; +} diff --git a/packages/workflows/src/tui/workflow-attach-pane-types.ts b/packages/workflows/src/tui/workflow-attach-pane-types.ts index aca03cda8..8546306ff 100644 --- a/packages/workflows/src/tui/workflow-attach-pane-types.ts +++ b/packages/workflows/src/tui/workflow-attach-pane-types.ts @@ -4,11 +4,14 @@ 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 { EnsurePostMortemStageHandleResult } from "../runs/foreground/postmortem-stage-chat.js"; export interface AttachUiStatusSurface { setStatus?: (key: string, value: string | undefined) => void; } +export type PostMortemHandleResolution = EnsurePostMortemStageHandleResult | undefined; + export interface WorkflowAttachPaneOpts { store: Store; graphTheme: GraphTheme; @@ -18,6 +21,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). Unavailable results retain their reason so the pane can render + * a truthful actionable read-only fallback. Called only after a live + * `stageControlRegistry.get()` miss. + */ + resolvePostMortemHandle?: (runId: string, stageId: string) => PostMortemHandleResolution; /** Broker used to route stage-local custom UI such as ask_user_question into attached chats. */ stageUiBroker?: StageUiBroker; /** @@ -53,6 +65,8 @@ export interface WorkflowAttachPaneOpts { * directly on the node's chat. */ initialAttachStageId?: string; + /** Owning nested run for `initialAttachStageId`; omitted stages resolve through the root graph. */ + initialAttachRunId?: string; /** * Optional accessor returning the current terminal row count. Threaded * into both `GraphView` and `StageChatView` so the overlay renders a diff --git a/packages/workflows/src/tui/workflow-attach-pane.ts b/packages/workflows/src/tui/workflow-attach-pane.ts index 91470dee0..f901ae9aa 100644 --- a/packages/workflows/src/tui/workflow-attach-pane.ts +++ b/packages/workflows/src/tui/workflow-attach-pane.ts @@ -1,15 +1,13 @@ /** * WorkflowAttachPane — outer in-place attach shell. * - * Wraps a single `pi.ui.custom` overlay popup whose interior swaps - * between the orchestrator `GraphView` and a stage-scoped + * Wraps a single `pi.ui.custom` overlay popup whose interior swaps between the orchestrator `GraphView` and a stage-scoped * `StageChatView`. Pressing Enter on a graph node attaches the popup * to that node's chat; Ctrl+D in chat mode swaps back to graph mode * with the same node still focused (see ui/attach-mockup.html), including * paused stage chats. * - * The shell never remounts the overlay — it only flips a `mode` - * field and re-renders, so the popup stays in pi-tui's overlay layer + * The shell never remounts the overlay — it only flips a `mode` field and re-renders, so the popup stays in pi-tui's overlay layer * across attach/detach cycles. This matches the mockup contract * (`only the popup interior swapped — outer chrome is unchanged`). * @@ -26,11 +24,12 @@ import type { GraphTheme } from "./graph-theme.js"; import { GraphView } from "./graph-view.js"; import { StageChatView, type StageChatDetachMetadata, type StageChatDetachReason } from "./stage-chat-view.js"; import { Key, matchesKey } from "./text-helpers.js"; -import type { StageControlHandle, StageControlRegistry } from "../runs/foreground/stage-control-registry.js"; +import type { StageControlRegistry } from "../runs/foreground/stage-control-registry.js"; import type { StageUiBroker } from "../shared/stage-ui-broker.js"; import type { StageSnapshot, StoreSnapshot } from "../shared/store-types.js"; import { expandWorkflowGraph } from "../shared/expanded-workflow-graph.js"; import { WORKFLOW_STATUS_KEY } from "./workflow-status.js"; +import { resolveAttachedStageHandle } from "./workflow-attach-pane-handle.js"; /** * Surface for the overlay footer/status tag. Passing `undefined` on dispose * prevents `pi-workflows/[/]` from lingering in later chat @@ -45,6 +44,7 @@ export class WorkflowAttachPane implements Component { private theme: GraphTheme; private runId: string | null; private registry: StageControlRegistry | undefined; + private resolvePostMortemHandle?: WorkflowAttachPaneOpts["resolvePostMortemHandle"]; private stageUiBroker: StageUiBroker | undefined; private uiStatus: AttachUiStatusSurface | undefined; private onClose: () => void; @@ -82,7 +82,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; @@ -105,7 +105,7 @@ export class WorkflowAttachPane implements Component { this.unsubscribeStore = this.store.subscribe((snapshot) => this._handleStoreUpdate(snapshot)); this.graphView = this._buildGraphView(); if (opts.initialAttachStageId !== undefined && this.runId) { - const target = this._resolveGraphStageTarget(this.runId, opts.initialAttachStageId); + const target = opts.initialAttachRunId === undefined ? this._resolveGraphStageTarget(this.runId, opts.initialAttachStageId) : { runId: opts.initialAttachRunId, stageId: opts.initialAttachStageId }; this._attachToStage(target.runId, target.stageId); } else { this._syncAwaitingInputKeys(this.store.snapshot()); @@ -175,7 +175,7 @@ export class WorkflowAttachPane implements Component { : 0; this.attachedRunId = runId; this.lastAttachedStageId = stageId; - const handle: StageControlHandle | undefined = this.registry?.get(runId, stageId); + const { handle, postMortemUnavailableReason } = resolveAttachedStageHandle(this.registry, this.resolvePostMortemHandle, runId, stageId); this.chatView?.dispose(); let chatView!: StageChatView; chatView = new StageChatView({ @@ -185,6 +185,7 @@ export class WorkflowAttachPane implements Component { stageId, workflowName: this._workflowName(runId), handle, + postMortemUnavailableReason, onDetach: (reason, metadata) => this._detachFromStage(reason, metadata), onClose: this.onClose, requestRender: this.hostRequestRender, @@ -237,7 +238,7 @@ export class WorkflowAttachPane implements Component { this._setBaseStatus(); this._syncMouseScrollTracking(); } - retarget(runId: string | null, stageId?: string): void { + retarget(runId: string | null, stageId?: string, stageRunId?: string): void { if (this.chatView && this.attachedRunId && this.lastAttachedStageId) { this.store.recordStageAttached(this.attachedRunId, this.lastAttachedStageId, false); } @@ -253,7 +254,7 @@ export class WorkflowAttachPane implements Component { this.graphView = this._buildGraphView(); this._syncAwaitingInputKeys(this.store.snapshot()); if (stageId !== undefined && runId) { - const target = this._resolveGraphStageTarget(runId, stageId); + const target = stageRunId === undefined ? this._resolveGraphStageTarget(runId, stageId) : { runId: stageRunId, stageId }; this._attachToStage(target.runId, target.stageId); return; } 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/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); + } +}); diff --git a/test/unit/postmortem-stage-chat.test.ts b/test/unit/postmortem-stage-chat.test.ts new file mode 100644 index 000000000..6dab1c920 --- /dev/null +++ b/test/unit/postmortem-stage-chat.test.ts @@ -0,0 +1,236 @@ +/** + * 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("disposes a pending lazy session when the registry is cleared", async () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("clear-race"); + const created = Promise.withResolvers(); + let disposeCalls = 0; + const session: StageSessionRuntime = { + ...mockSession(), + sessionFile, + dispose() { disposeCalls += 1; }, + }; + const stage = completedStage({ sessionFile }); + const result = ensurePostMortemStageHandle("run-1", stage, { + registry, + adapters: { + agentSession: { + async create() { return created.promise; }, + }, + }, + cwd: tempDir, + }); + assert.equal(result.ok, true); + if (!result.ok) return; + + const attaching = result.handle.ensureAttached(); + await Promise.resolve(); + registry.clear(); + created.resolve(session); + + await assert.rejects(attaching, /session has been disposed/); + assert.equal(disposeCalls, 1); + assert.equal(result.handle.isDisposed, true); + assert.deepEqual(registry.forRun("run-1"), []); + }); + + test("rejects post-mortem pause and resume without appending a prompt", async () => { + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("no-resume"); + 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, counter), + cwd: tempDir, + }); + assert.equal(result.ok, true); + if (!result.ok) return; + + 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", () => { + 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/slash-dispatch-resume.ts b/test/unit/slash-dispatch-resume.ts index 25904a482..9e3212f76 100644 --- a/test/unit/slash-dispatch-resume.ts +++ b/test/unit/slash-dispatch-resume.ts @@ -268,6 +268,80 @@ 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; + let attachedOwner: string | undefined; + const notifications: string[] = []; + const { pi, commands } = buildMockPi(); + addFactoryStubs(pi); + pi.ui = { + notify: (message: string) => { notifications.push(message); }, + setWidget: () => {}, + custom: (factory) => { + overlayOpens += 1; + const component = factory({ requestRender: () => {}, terminal: { rows: 32, columns: 96 } }, {}, {}, () => {}); + attachedOwner = store.runs().find((run) => run.stages.some((stage) => stage.id === "review" && stage.attached === true))?.id; + component.dispose?.(); + 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 childOneRunId = `attach-child-one-${Date.now()}`; + const childTwoRunId = `attach-child-two-${Date.now()}`; + const nestedStageId = "review"; + const boundary = (id: string, childRunId: string) => ({ + id, + name: id, + status: "completed" as const, + parentIds: [], + startedAt: Date.now(), + endedAt: Date.now(), + toolEvents: [], + workflowChild: { + alias: id, + workflow: "child-workflow", + runId: childRunId, + status: "completed" as const, + outputs: {}, + }, + }); + store.recordRunStart({ + ...makeInflightRun(rootRunId), + stages: [boundary("child-one", childOneRunId), boundary("child-two", childTwoRunId)], + }); + for (const childRunId of [childOneRunId, childTwoRunId]) { + store.recordRunStart({ + ...makeInflightRun(childRunId), + parentRunId: rootRunId, + rootRunId, + stages: [{ + id: nestedStageId, + name: "review", + status: "completed", + parentIds: [], + startedAt: Date.now(), + endedAt: Date.now(), + toolEvents: [], + sessionFile: `/tmp/${childRunId}.jsonl`, + }], + }); + } + + await handler(`attach ${rootRunId} ${childTwoRunId}:${nestedStageId}`, { hasUI: true, ui: pi.ui }); + + assert.equal(overlayOpens, 1); + assert.equal(attachedOwner, childTwoRunId); + const content = notifications.join("\n"); + assert.match(content, /Attached to .* stage review/); + assert.doesNotMatch(content, /Stage not found/); + }); +}); + // --------------------------------------------------------------------------- // resume regression: tool action "resume" against active run returns status:"ok" // --------------------------------------------------------------------------- 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..869eeec7c --- /dev/null +++ b/test/unit/workflow-attach-pane-11.test.ts @@ -0,0 +1,171 @@ +// @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, store as globalStore } 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 { createPostMortemHandleResolver } from "../../packages/workflows/src/extension/postmortem-deps.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 { ok: true, handle: 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(); + }); + + test("renders the post-mortem unavailability reason instead of a generic archive", () => { + const store = createStore(); + setupCompletedRun(store, "run-1"); + const pane = new WorkflowAttachPane({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-1", + stageControlRegistry: createStageControlRegistry(), + resolvePostMortemHandle: () => ({ ok: false, reason: "invalid_session" }), + onClose: () => {}, + initialAttachStageId: "stage-a", + }); + + const rendered = pane.render(40).join("\n"); + const visible = rendered.replace(/\x1b\[[0-9;]*m/g, "").replace(/\s+/g, " "); + assert.match(visible, /SESSION UNAVAILABLE/); + assert.match(visible, /The retained session is missing, unreadable, or invalid\. Check that the session file still exists and is readable\./); + assert.doesNotMatch(visible, /archived transcript/); + pane.dispose(); + }); + + test.serial("the extension resolver preserves invalid-session reasons", () => { + globalStore.clear(); + try { + setupCompletedRun(globalStore, "resolver-run"); + const resolver = createPostMortemHandleResolver({ + adapters: { agentSession: { async create() { throw new Error("must not create"); } } }, + resolveDefaultStageSessionDir: () => undefined, + }); + + assert.deepEqual( + resolver("resolver-run", "stage-a"), + { ok: false, reason: "invalid_session" }, + ); + } finally { + globalStore.clear(); + } + }); +}); 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 () => { 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..2089e1d1a --- /dev/null +++ b/test/unit/workflow-tool-send-postmortem.test.ts @@ -0,0 +1,205 @@ +/** + * 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, + 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"; + +let tempDir = ""; +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); +}); + +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 }); +} +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 () => { + 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("returns a structured noop for explicit resume without appending or mutating terminal status", async () => { + const sessionFile = retainedSession("send-no-resume"); + 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: "resume should be rejected", delivery: "resume" }, + { resolvePostMortemDeps: resolvePostMortemDeps(session, counter) }, + ); + + 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); + 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 }; + 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); + }); +});