diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 1d34ab513..fb9eb4dce 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed the bundled workflows `/workflow resume` experience to mix successful completed workflows into the existing globally newest-first, deduplicated picker with green completed styling, resolve full IDs and prefixes across live, durable, and completed targets, and reopen retained stage chats for follow-up without re-running workflow code or replaying side effects. Completed durable state is retained for authoritative inspection; rows need checkpoints and at least one strictly valid retained conversation, invalid per-stage transcript paths cannot open chat, repeated inspection refreshes changed authoritative chat handles, and selector mount failures close safely. ([#1532](https://github.com/bastani-inc/atomic/issues/1532)) + ## [0.9.8] - 2026-07-12 ### Changed diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index 0a83dfdcc..955268ce0 100644 --- a/packages/coding-agent/docs/workflows.md +++ b/packages/coding-agent/docs/workflows.md @@ -1048,13 +1048,13 @@ Atomic workflows support **cross-session resumability** via a durable workflow b - **Durable `ctx.tool`**: `ctx.tool(name, args, fn)` runs TypeScript and caches the result by stable call order plus content hash of `name` + `args`. Repeated same-name/same-args calls are distinct within one workflow and replay in order after resume. DBOS-backed checkpoint writes are serialized and awaited before the tool returns, so a completed side effect is not exposed to workflow code before its checkpoint is durable; retry backoff observes workflow cancellation before later attempts run. - **Durable `ctx.ui`**: Completed `ctx.ui.input` / `confirm` / `select` / `editor` / `custom` responses are cached by prompt identity, including prompt kind, label/message, options, and call order so repeated prompts do not collide. On resume an already-answered prompt returns its cached response instead of re-asking the user, including `ctx.ui.custom` prompts that intentionally complete with `undefined`/void. - **Durable `ctx.stage` / `ctx.task` / `ctx.chain` / `ctx.parallel` / `ctx.workflow`**: Completed stage-like operations are recorded with stable replay keys. `ctx.chain` and `ctx.parallel` reuse the durable task primitive for each item, and child `ctx.workflow` calls checkpoint the completed child result at the parent workflow boundary with a dedicated replay-key counter so repeated same-child calls do not desync ordinal sequencing across resume. Schema-backed `ctx.stage(..., { schema }).prompt(...)` checkpoints preserve the parsed structured value, not just the rendered text. The live stage finalizer awaits durable writes before returning control where practical; if finalization fails, Atomic releases stage handles and concurrency limiter slots before reporting the finalization error instead of marking the workflow completed without a durable checkpoint. Workflow terminal status — including `ctx.exit` terminal states (`cancelled`, `blocked`, `skipped`) — is flushed to durable metadata before the run reports completion. On resume, cached operations are skipped and represented in the workflow graph as completed durable-replay nodes so status/overlay views still show progress. -- **Session-file cache**: Because the default backend is persistent, top-level workflow metadata is mirrored as Atomic custom entries (`customType: "workflow.durable.checkpoint"`) in the session JSONL so a new session can discover resumable root workflows without scanning the durable store directly. Child workflows are hidden from this catalog. The durable backend remains the checkpoint source of truth; the session cache is a discovery index. `/workflow resume` refuses a cache-only entry as `stale` when a root workflow has session metadata but no checkpoint state in the durable backend, instead of silently re-running from scratch — re-run the workflow explicitly to start fresh. +- **Session-file cache**: Because the default backend is persistent, top-level workflow metadata is mirrored as Atomic custom entries (`customType: "workflow.durable.checkpoint"`) in the session JSONL so a new session can discover resumable root workflows without scanning the durable store directly. Child workflows are hidden from this cache. The durable backend remains the checkpoint source of truth; the session cache is only a discovery index. `/workflow resume` refuses a cache-only resumable entry as `stale` instead of silently re-running from scratch. Completed inspection is cataloged separately from resumability and requires an authoritative completed backend handle, durable checkpoints, and at least one retained stage conversation. Each retained transcript must be a regular, parseable Atomic session file with prior message context; empty, malformed, header-only, missing, and directory paths are stale. - **Child side-effect scoping**: A child workflow launched via `ctx.workflow(child)` checkpoints its internal `ctx.tool`/`ctx.ui`/`ctx.stage` side effects under the parent (root) durable workflow, keyed by a stable child-boundary scope. If the parent is interrupted while a child is mid-flight, resuming the parent re-runs the child but replays the child's already-completed side effects from the root store instead of re-executing them (no split-brain per-run UUID checkpoints). -- **Crash-safe file backend**: Default file-backed durability stores each root workflow in its own lock-protected JSON file under `~/.atomic/workflow-durable` instead of one shared state file. Completed/cancelled/non-resumable terminal workflow files are pruned, writes use `0700` directories and `0600` state/owner files where the platform supports chmod, and stale locks are reclaimed only when their owner marker belongs to a dead process. If Atomic cannot resolve `HOME`/`USERPROFILE`, it fails closed to in-memory durability instead of writing state under `/tmp`. Durable replay identities use a SHA-256 digest over canonical JSON so distinct tool/stage identities never collide. +- **Crash-safe file backend**: Default file-backed durability stores each root workflow in its own lock-protected JSON file under `~/.atomic/workflow-durable` instead of one shared state file. Successful completed workflow files are retained for read-only inspection through `/workflow resume`; cancelled and explicitly non-resumable failed/blocked files are pruned. Writes use `0700` directories and `0600` state/owner files where the platform supports chmod, and stale locks are reclaimed only when their owner marker belongs to a dead process. If Atomic cannot resolve `HOME`/`USERPROFILE`, it fails closed to in-memory durability instead of writing state under `/tmp`. Durable replay identities use a SHA-256 digest over canonical JSON so distinct tool/stage identities never collide. **Privacy and retention.** File and DBOS durability persist workflow inputs, completed `ctx.tool` outputs, answered `ctx.ui` responses, stage outputs, and stage-session paths as plaintext durable state so replay can skip completed work. Treat `~/.atomic/workflow-durable` and the configured DBOS database as sensitive. To opt out of cross-session checkpoint persistence for a session, start Atomic with `ATOMIC_WORKFLOW_DURABLE=0` (also accepts `false`, `off`, `memory`, or `in-memory`); workflows then use process-local in-memory durability and cannot be resumed after the process exits. -The default file backend prunes completed, cancelled, and explicitly non-resumable failed/blocked workflow files when terminal status is recorded. Running, paused (quit), and recoverably failed workflows with checkpoint progress are retained indefinitely by design so crash recovery and `/workflow resume` remain available across sessions. Use `/workflow kill ` to mark a durable workflow cancelled/non-resumable and remove its file-backed state, or delete the corresponding `workflow-.json` file under `~/.atomic/workflow-durable` if you intentionally want to discard abandoned resumable state outside Atomic. +The default file backend retains running, paused (quit), recoverably failed, and successful completed workflows with checkpoint progress. Completed state remains available so `/workflow resume` can reconstruct an immutable final graph and reopen retained stage transcripts without dispatching the workflow. Cancelled and explicitly non-resumable failed/blocked files are removed. Retained files contain the plaintext durable data described above; use `/workflow kill ` before completion when you want to cancel and remove a durable run, or delete the corresponding `workflow-.json` file under `~/.atomic/workflow-durable` when you intentionally want to discard retained completed or abandoned state outside Atomic. **Resume after editing a workflow.** Replay identity is based on workflow id plus each `ctx.*` call's stable content hash and ordinal. Editing a workflow between quit/crash and resume can intentionally invalidate or shift checkpoints: changed prompts/arguments/stage names re-run, and inserted/reordered `ctx.*` calls may not match old ordinal checkpoints. Prefer finishing or killing old durable runs before deploying reordered workflow definitions; otherwise treat resume as best-effort from compatible checkpoint identities. @@ -1087,14 +1087,18 @@ export default workflow({ ### `/workflow resume` — cross-session resume selector -The `/workflow resume` command mirrors `/resume` ergonomics and now uses the same Atomic session-selector tree chrome as `/resume` for its interactive picker. With no id, interactive sessions show one `/resume`-style searchable/threaded selector that lists both live/paused runs and compatible cross-session durable workflows together, so the user can choose from all resumable workflows in one view. The selector uses async DBOS hydration (`prepareDurableResumable`) so cross-session durable entries discovered from Postgres are included even when live runs exist. Dismissing the selector returns to chat without opening a second picker, and selecting a failed-but-resumable live run uses the normal continuation path rather than opening a read-only snapshot. On a successful durable resume, the overlay connects to the resumed run automatically (matching live resume behavior). Headless sessions print the filtered durable catalog and durable resume dispatch preserves the same non-interactive tool/UI restrictions as normal workflow dispatch. With a target id, Atomic first checks live runs in the current session; when the target id is not a live run, it falls back to the cross-session durable resume catalog and resumes by top-level workflow id, then opens the overlay. Resume re-dispatches the workflow with its cached inputs and the **original workflow id**, so completed `ctx.tool`, `ctx.ui`, stage/task/chain/parallel items, and child workflow boundaries return cached results instead of re-executing — DBOS-style replay without Postgres. Replayed stages preserve graph parent/frontier lineage so subsequent stages connect correctly. Empty string stage outputs are preserved distinctly from undefined/no-result during checkpointing. Stale session-cache entries with no matching durable backend handle are hidden from the selector because they came from older non-checkpointed workflow engines and cannot resume usefully. Durable `running`/`failed`/`blocked` rows whose workflow definition is no longer registered are also hidden from no-arg discovery, which keeps old test or pre-engine rows out of the picker without changing targeted `/workflow resume ` diagnostics. Stale cache entries for workflows the durable backend knows are terminal (completed/cancelled/non-resumable) are also suppressed, so terminal workflows cannot be resurrected from old JSONL cache metadata. +The `/workflow resume` command mirrors `/resume` ergonomics and uses the same searchable/threaded Atomic session-selector chrome. With no id, it builds one deduplicated picker containing paused or recoverably failed live runs, compatible cross-session resumable runs, and authoritative successful completed runs, globally ordered newest-first by run or stage activity with deterministic workflow-id ties. Live rows take precedence over matching durable rows, and resumable durable rows take precedence over matching completed rows. Completed rows use the existing green `✓ completed` visual language. Durable catalogs are hydrated from DBOS when configured, stale cache-only data is filtered, and actively executing runs remain excluded to prevent double dispatch. A selector mount failure closes cleanly instead of leaving the command pending. + +Selecting a paused, failed, blocked, or crash-recovery target follows the existing resume path unchanged: Atomic re-dispatches the workflow with its cached inputs and the **original workflow id**, so previously completed `ctx.tool`, `ctx.ui`, stage/task/chain/parallel items, and child workflow boundaries replay from durable checkpoints rather than executing again. Selecting a completed target follows a separate open path. Atomic reconstructs a completed run/stage snapshot from authoritative checkpoints and opens the detail/chat overlay without calling the durable resume dispatcher or re-running workflow stages, tools, tasks, prompts, or workflow code. + +Completed detail state is read-only. A retained stage transcript can still be opened and continued conversationally: the stage's Atomic session is reopened lazily and a new follow-up is appended to that chat only. Follow-up chat does not change the completed run or durable handle back to `running`, does not replay workflow side effects, and does not make the completed workflow resumable. ```text -/workflow resume # Show resumable workflows (live + durable) -/workflow resume # Resume by top-level id; completed checkpoints replay +/workflow resume # Mixed picker: resumable + completed +/workflow resume # Resume unfinished work or open completed detail/chat ``` -The selector displays workflow rows through the same `/resume` tree UI and lists only **inactive, resume-applicable** workflows: paused (quit) or recoverably-failed durable runs with checkpoint progress, plus any live paused/failed runs. Actively-running workflows are hidden from the selector — resuming one that is already executing (in this or another session) would double-dispatch, so `/workflow resume ` is refused with an intuitive error pointing at `/workflow connect` (to attach) or `/workflow kill` (to clear a stuck run). Quitting the CLI/panel flips the durable handle from `running` to `paused`, which is what makes a workflow re-enter the resumable set; a fresh dispatch or durable resume flips it back to `running`. Running/paused durable rows appear only after the workflow has at least one durable checkpoint or pending checkpointable prompt, so a just-started workflow with no resumable state does not clutter the picker. Eligible LM stages also write their Atomic/Pi session metadata as soon as the stage session opens; on durable resume, Atomic reopens that exact session file and sends `Continue` instead of replaying the original workflow prompt, so repeated quit/resume cycles append to the active LM chat rather than duplicating prompts or emptying prior chats. Completed tool/UI/stage checkpoints still replay all-or-nothing, and successful completion marks the durable handle `completed`, prunes the file-backed state for that workflow, and removes the row once the whole workflow reaches the end. Quitting the CLI is treated as a resumable process boundary rather than explicit workflow cancellation, so durable-progress workflows remain available to `/workflow resume`; use `/workflow kill` when you want a workflow removed as cancelled/non-resumable. If no resumable workflows remain after filtering, interactive `/workflow resume` still opens the same empty selector rather than printing an error, matching `/resume` behavior. In fresh sessions, Atomic scans workflow-specific `workflow.durable.checkpoint` JSONL history for discovery, but the durable backend remains authoritative: cache rows without backend checkpoint state are treated as old incompatible metadata and are hidden instead of prompting a stale resume failure. File-backed durability is enabled by default under `~/.atomic/workflow-durable`, so `/workflow resume` can discover cross-session workflows without any setup. +Explicit full IDs take precedence, while prefixes resolve across live, resumable durable, and completed targets as one namespace. Ambiguous prefixes use the existing-style ambiguity diagnostic. A completed backend row with no checkpoints or no usable retained stage conversation is hidden from the picker; an explicit target reports that it is stale or missing required durable checkpoint/session data. A completed run remains inspectable when at least one stage has a usable transcript; missing, empty, directory, context-empty, or partially malformed transcript paths are omitted from stage chat attachment. Validation uses the final retained transcript for a repeated stage replay key, so an obsolete superseded checkpoint path does not hide an otherwise valid completed run. Reopening inspection refreshes a changed authoritative retained-chat handle. Session-cache-only rows are likewise hidden because the backend is authoritative. Cancelled, killed, non-resumable failed, and other terminal non-success states are never added. Normal `/resume`, `atomic -r`, and `--continue` behavior for internal workflow stage sessions is unchanged. ### Cancellation, failure, and retry semantics diff --git a/packages/coding-agent/src/core/session-manager-types.ts b/packages/coding-agent/src/core/session-manager-types.ts index b9caa0369..90a083baf 100644 --- a/packages/coding-agent/src/core/session-manager-types.ts +++ b/packages/coding-agent/src/core/session-manager-types.ts @@ -210,6 +210,8 @@ export interface SessionInfo { messageCount: number; firstMessage: string; allMessagesText: string; + /** Optional semantic color for synthetic selector rows. */ + messageColor?: "success" | "warning" | "accent"; } export interface ContextDeletionFilters { diff --git a/packages/coding-agent/src/modes/interactive/components/session-selector-list.ts b/packages/coding-agent/src/modes/interactive/components/session-selector-list.ts index 9e29b3d13..6bbf151e3 100644 --- a/packages/coding-agent/src/modes/interactive/components/session-selector-list.ts +++ b/packages/coding-agent/src/modes/interactive/components/session-selector-list.ts @@ -220,11 +220,13 @@ export class SessionList implements Component, Focusable { const truncatedMsg = truncateToWidth(normalizedMessage, Math.max(10, availableForMsg), "…"); // Style message - let messageColor: "error" | "warning" | "accent" | null = null; + let messageColor: "error" | "warning" | "accent" | "success" | null = null; if (isConfirmingDelete) { messageColor = "error"; } else if (isCurrent) { messageColor = "accent"; + } else if (session.messageColor !== undefined) { + messageColor = session.messageColor; } else if (hasName) { messageColor = "warning"; } diff --git a/packages/workflows/CHANGELOG.md b/packages/workflows/CHANGELOG.md index 5fcb8f589..c28c664dc 100644 --- a/packages/workflows/CHANGELOG.md +++ b/packages/workflows/CHANGELOG.md @@ -18,6 +18,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - Fixed the builtin `open-claude-design` artifact directory fallback to use a per-user OS tmpdir namespace (`open-claude-design-`), preventing `EACCES` failures on shared hosts where another user already owns the plain `/open-claude-design` directory. The legacy shared path remains a secondary fallback, and the last-resort synthesized path now also uses the per-user namespace so best-effort feedback persistence can actually create it. - Prevented the synced Impeccable fixture's nested `git init` from inheriting hook-local Git repository variables and corrupting a linked repository's shared `core.worktree`. Runner-managed reusable worktrees now remap propagated invoking-repository `cwd` values and relative direct outputs into the selected worktree; reject blank, self, nested, foreign, lexical-escape, symlink-escape, and changed cached targets before opening a later session; key cached identity by canonical repository/target across equivalent path and ref spellings; revalidate newly created targets; and expose clearer workflow tool guidance/schema descriptions so natural-language worktree requests cannot be mistaken for runtime configuration. Temporary direct task and parallel worktrees also derive omitted/relative task cwd values from the runner invocation cwd and clean up when startup fails before their workflow callback, while relative outputs are persisted to stable runner-owned artifacts before cleanup and reject parent traversal, child symlink escapes, or a linked trusted artifact root across direct, parallel, and chain modes; blank `chainDir` values no longer bypass reusable-worktree output routing. - Fixed the active Workflow Orchestrator graph pane to pan left and right for horizontal trackpad and terminal mouse-wheel events while preserving vertical wheel panning and pane-local input capture ([#1756](https://github.com/bastani-inc/atomic/issues/1756)). +- Fixed `/workflow resume` so its single globally newest-first, deduplicated picker includes current paused or recoverably failed and durable resumable entries alongside authoritative successful completed workflows. Completed rows use green `✓ completed` styling and open an immutable detail/chat snapshot with transcript follow-up conversation, without durable re-dispatch or workflow side-effect replay. Full IDs take precedence and prefixes resolve across the mixed namespace; completed rows require checkpoints and at least one strictly valid retained conversation, invalid per-stage paths cannot attach chat, repeated inspection refreshes changed authoritative chat handles, and selector mount failures close safely. Existing resumable workflow and ordinary internal-session history behavior remains on the original paths. ([#1532](https://github.com/bastani-inc/atomic/issues/1532)) ## [0.9.8] - 2026-07-12 diff --git a/packages/workflows/src/durable/backend.ts b/packages/workflows/src/durable/backend.ts index a01f9fb22..807f17bd5 100644 --- a/packages/workflows/src/durable/backend.ts +++ b/packages/workflows/src/durable/backend.ts @@ -111,6 +111,8 @@ export interface DurableWorkflowBackend { * resumable after failure). Used by the `/workflow resume` selector. */ listResumableWorkflows(): readonly ResumableWorkflowEntry[]; + /** List successful completed root workflows with durable checkpoint progress. */ + listCompletedWorkflows(): readonly ResumableWorkflowEntry[]; /** Export a session-cache entry for the given workflow (for JSONL persistence). */ toCacheEntry(workflowId: string): DurableCheckpointEntry | undefined; @@ -274,6 +276,12 @@ export class InMemoryDurableBackend implements DurableWorkflowBackend { .map((rec) => toResumableEntry(rec.handle)); } + listCompletedWorkflows(): readonly ResumableWorkflowEntry[] { + return [...this.workflows.values()] + .filter((rec) => isRootWorkflow(rec.handle) && isCompletedHandle(rec.handle)) + .map((rec) => toResumableEntry(rec.handle)); + } + toCacheEntry(workflowId: string): DurableCheckpointEntry | undefined { const rec = this.workflows.get(workflowId); if (!rec) return undefined; @@ -333,6 +341,10 @@ function isResumableHandle(handle: DurableWorkflowHandle): boolean { return (handle.status === "running" || handle.status === "paused") && hasResumeProgress(handle); } +function isCompletedHandle(handle: DurableWorkflowHandle): boolean { + return handle.status === "completed" && hasResumeProgress(handle); +} + function toResumableEntry(handle: DurableWorkflowHandle): ResumableWorkflowEntry { return { workflowId: handle.workflowId, diff --git a/packages/workflows/src/durable/completed-catalog.ts b/packages/workflows/src/durable/completed-catalog.ts new file mode 100644 index 000000000..fc8aa51c8 --- /dev/null +++ b/packages/workflows/src/durable/completed-catalog.ts @@ -0,0 +1,245 @@ +import { readFileSync, statSync } from "node:fs"; +import type { RunSnapshot, StageSnapshot } from "../shared/store-types.js"; +import type { WorkflowInputValues } from "../shared/types.js"; +import type { DurableWorkflowBackend } from "./backend.js"; +import type { + DurableCheckpoint, + DurableStageCheckpoint, + ResumableWorkflowEntry, +} from "./types.js"; +import { resolveDurableEntry } from "./resume-runtime.js"; + +export type CompletedWorkflowResolution = + | { readonly kind: "found"; readonly entry: ResumableWorkflowEntry; readonly snapshot: RunSnapshot } + | { readonly kind: "ambiguous"; readonly matches: readonly ResumableWorkflowEntry[] } + | { 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; + readonly name: string; + readonly firstCompletedAt: number; + readonly output?: DurableStageCheckpoint["output"]; + readonly result?: string; + readonly sessionId?: string; + readonly sessionFile?: string; + readonly startedAt?: number; + readonly endedAt?: number; + readonly durationMs?: number; + readonly model?: string; + readonly fastMode?: boolean; + readonly attemptedModels?: readonly string[]; + readonly modelAttempts?: DurableStageCheckpoint["modelAttempts"]; +} + +/** Authoritative completed rows. This path is deliberately separate from resumability. */ +export function listCompletedFromBackend( + backend: DurableWorkflowBackend, +): readonly ResumableWorkflowEntry[] { + return backend.listCompletedWorkflows(); +} + +/** Completed rows whose authoritative checkpoints and referenced transcripts still exist. */ +export function listOpenableCompletedWorkflows( + backend: DurableWorkflowBackend, +): readonly ResumableWorkflowEntry[] { + return listCompletedFromBackend(backend) + .filter((entry) => completedWorkflowSnapshot(backend, entry) !== undefined) + .sort((a, b) => b.updatedAt - a.updatedAt); +} + +export function resolveCompletedWorkflow( + workflowIdOrPrefix: string, + backend: DurableWorkflowBackend, + openableCatalog: readonly ResumableWorkflowEntry[] = listOpenableCompletedWorkflows(backend), +): CompletedWorkflowResolution { + const resolved = resolveDurableEntry(workflowIdOrPrefix, openableCatalog); + if (resolved !== undefined) { + if ("kind" in resolved) return { kind: "ambiguous", matches: resolved.matches }; + const snapshot = completedWorkflowSnapshot(backend, resolved); + return snapshot === undefined + ? { kind: "stale", entry: resolved } + : { kind: "found", entry: resolved, snapshot }; + } + + const authoritative = resolveDurableEntry(workflowIdOrPrefix, listCompletedFromBackend(backend)); + if (authoritative === undefined) return { kind: "not_found" }; + if ("kind" in authoritative) return { kind: "ambiguous", matches: authoritative.matches }; + return { kind: "stale", entry: authoritative }; +} + +export function completedWorkflowSnapshot( + backend: DurableWorkflowBackend, + entry: ResumableWorkflowEntry, +): RunSnapshot | undefined { + const handle = backend.getWorkflow(entry.workflowId); + if (handle === undefined || handle.status !== "completed") return undefined; + const checkpoints = backend.listCheckpoints(entry.workflowId); + if (checkpoints.length === 0) return undefined; + const stages = stageSnapshotsFromCheckpoints(checkpoints, handle.updatedAt).map(validatedStageTranscript); + if (!stages.some((stage) => stage.sessionFile !== undefined)) return undefined; + + return { + id: handle.workflowId, + name: handle.name, + inputs: { ...handle.inputs } as WorkflowInputValues, + status: "completed", + stages, + startedAt: handle.createdAt, + endedAt: handle.updatedAt, + durationMs: Math.max(0, handle.updatedAt - handle.createdAt), + resumable: false, + }; +} + +function validatedStageTranscript(stage: StageSnapshot): StageSnapshot { + if (stage.sessionFile === undefined || isReopenableSessionTranscript(stage.sessionFile)) return stage; + const { sessionFile, ...withoutSessionFile } = stage; + void sessionFile; + 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[], + fallbackCompletedAt: number, +): StageSnapshot[] { + const drafts = new Map(); + for (const checkpoint of checkpoints) { + if (checkpoint.kind !== "stage") continue; + const existing = drafts.get(checkpoint.replayKey); + drafts.set(checkpoint.replayKey, mergeStageDraft(existing, checkpoint)); + } + const ordered = [...drafts.values()].sort((a, b) => a.firstCompletedAt - b.firstCompletedAt); + if (ordered.length === 0) return [syntheticCheckpointStage(checkpoints.length, fallbackCompletedAt)]; + return ordered.map(stageSnapshotFromDraft); +} + +function mergeStageDraft( + existing: StageDraft | undefined, + checkpoint: DurableStageCheckpoint, +): StageDraft { + return { + replayKey: checkpoint.replayKey, + name: existing?.name ?? checkpoint.name, + firstCompletedAt: Math.min(existing?.firstCompletedAt ?? checkpoint.completedAt, checkpoint.completedAt), + ...valueOrExisting("output", checkpoint, existing), + ...valueOrExisting("result", checkpoint, existing), + ...valueOrExisting("sessionId", checkpoint, existing), + ...valueOrExisting("sessionFile", checkpoint, existing), + ...valueOrExisting("startedAt", checkpoint, existing), + ...valueOrExisting("endedAt", checkpoint, existing), + ...valueOrExisting("durationMs", checkpoint, existing), + ...valueOrExisting("model", checkpoint, existing), + ...valueOrExisting("fastMode", checkpoint, existing), + ...valueOrExisting("attemptedModels", checkpoint, existing), + ...valueOrExisting("modelAttempts", checkpoint, existing), + }; +} + +function valueOrExisting< + K extends keyof Omit +>(key: K, checkpoint: DurableStageCheckpoint, existing: StageDraft | undefined): Pick | object { + const checkpointValue = checkpoint[key]; + if (checkpointValue !== undefined) return { [key]: checkpointValue } as Pick; + const existingValue = existing?.[key]; + return existingValue === undefined ? {} : { [key]: existingValue } as Pick; +} + +function stageSnapshotFromDraft(draft: StageDraft, index: number): StageSnapshot { + const startedAt = draft.startedAt ?? draft.firstCompletedAt; + const endedAt = draft.endedAt ?? draft.firstCompletedAt; + return { + id: `completed-stage-${index + 1}`, + name: draft.name, + status: "completed", + parentIds: [], + startedAt, + endedAt, + durationMs: draft.durationMs ?? Math.max(0, endedAt - startedAt), + ...(stageResult(draft) !== undefined ? { result: stageResult(draft) } : {}), + replayKey: draft.replayKey, + toolEvents: [], + attachable: false, + ...(draft.sessionId !== undefined ? { sessionId: draft.sessionId } : {}), + ...(draft.sessionFile !== undefined ? { sessionFile: draft.sessionFile } : {}), + ...(draft.model !== undefined ? { model: draft.model } : {}), + ...(draft.fastMode !== undefined ? { fastMode: draft.fastMode } : {}), + ...(draft.attemptedModels !== undefined ? { attemptedModels: draft.attemptedModels } : {}), + ...(draft.modelAttempts !== undefined ? { modelAttempts: draft.modelAttempts } : {}), + }; +} + +function stageResult(draft: StageDraft): string | undefined { + if (draft.result !== undefined) return draft.result; + if (draft.output === undefined) return undefined; + return typeof draft.output === "string" ? draft.output : JSON.stringify(draft.output); +} + +function syntheticCheckpointStage(checkpointCount: number, completedAt: number): StageSnapshot { + return { + id: "completed-checkpoints", + name: "durable checkpoints", + status: "completed", + parentIds: [], + startedAt: completedAt, + endedAt: completedAt, + durationMs: 0, + result: `${checkpointCount} durable checkpoint${checkpointCount === 1 ? "" : "s"}`, + toolEvents: [], + attachable: false, + }; +} diff --git a/packages/workflows/src/durable/completed-inspection.ts b/packages/workflows/src/durable/completed-inspection.ts new file mode 100644 index 000000000..3417492a7 --- /dev/null +++ b/packages/workflows/src/durable/completed-inspection.ts @@ -0,0 +1,213 @@ +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 { + stageControlRegistry as defaultStageControlRegistry, + type AgentSessionEventListener, + type StageControlHandle, + type StageControlRegistry, +} from "../runs/foreground/stage-control-registry.js"; +import type { DurableWorkflowBackend } from "./backend.js"; +import { + listOpenableCompletedWorkflows, + resolveCompletedWorkflow, +} from "./completed-catalog.js"; +import type { ResumableWorkflowEntry } from "./types.js"; + +export type OpenCompletedDurableResult = + | { readonly ok: true; readonly runId: string; readonly workflowId: string; readonly name: string; readonly message: string } + | { readonly ok: false; readonly reason: "not_found" | "ambiguous" | "stale" | "active"; readonly message: string }; + +export interface OpenCompletedDurableDeps { + readonly durableBackend: DurableWorkflowBackend; + readonly store: Store; + readonly adapters?: StageAdapters; + readonly stageControlRegistry?: StageControlRegistry; + readonly cwd?: string; + readonly defaultSessionDir?: string; +} + +interface CompletedChatRegistration { + readonly handle: StageControlHandle; + readonly unregister: () => void; +} + +const completedChatRegistrations = new WeakMap< + StageControlRegistry, + Map +>(); + +/** + * Open a completed durable workflow as an immutable run snapshot. The only + * mutable surface is a lazily reopened stage chat, which appends follow-up + * conversation to its retained Atomic session without dispatching the workflow. + */ +export function openCompletedDurableWorkflow( + workflowIdOrPrefix: string, + deps: OpenCompletedDurableDeps, + catalog: readonly ResumableWorkflowEntry[] = listOpenableCompletedWorkflows(deps.durableBackend), +): OpenCompletedDurableResult { + const resolved = resolveCompletedWorkflow(workflowIdOrPrefix, deps.durableBackend, catalog); + if (resolved.kind === "not_found") { + return failure("not_found", `No completed durable workflow found for id/prefix: ${workflowIdOrPrefix}`); + } + if (resolved.kind === "ambiguous") { + const matches = resolved.matches.map((entry) => `${entry.name} (${entry.workflowId.slice(0, 8)})`).join(", "); + return failure("ambiguous", `Ambiguous completed workflow prefix "${workflowIdOrPrefix}" matches: ${matches}`); + } + if (resolved.kind === "stale") { + return failure( + "stale", + `Completed workflow ${resolved.entry.workflowId.slice(0, 8)} is stale or missing durable checkpoint/session data and cannot be opened.`, + ); + } + + const existing = deps.store.runs().find((run) => run.id === resolved.snapshot.id); + if (existing !== undefined && existing.status !== "completed") { + return failure( + "active", + `Workflow ${resolved.snapshot.id.slice(0, 8)} is already active in this session; attach with /workflow connect ${resolved.snapshot.id.slice(0, 8)} instead.`, + ); + } + const snapshot = resolved.snapshot; + if (existing !== undefined) deps.store.removeRun(existing.id); + deps.store.recordRunStart(snapshot); + registerCompletedChatHandles(snapshot, deps); + return { + ok: true, + runId: snapshot.id, + workflowId: snapshot.id, + name: snapshot.name, + message: `Opened completed durable workflow "${snapshot.name}" (${snapshot.id.slice(0, 8)}) for read-only inspection and follow-up chat.`, + }; +} + +function failure( + reason: "not_found" | "ambiguous" | "stale" | "active", + message: string, +): OpenCompletedDurableResult { + return { ok: false, reason, message }; +} + +function registerCompletedChatHandles( + snapshot: RunSnapshot, + deps: OpenCompletedDurableDeps, +): void { + if (deps.adapters?.agentSession === undefined) return; + const registry = deps.stageControlRegistry ?? defaultStageControlRegistry; + const registrations = completedChatRegistrations.get(registry) ?? new Map(); + completedChatRegistrations.set(registry, registrations); + const desiredKeys = new Set( + snapshot.stages + .filter((stage) => stage.sessionFile !== undefined) + .map((stage) => completedChatKey(snapshot.id, stage.id)), + ); + for (const [key, registration] of registrations) { + if (!key.startsWith(`${snapshot.id}:`) || desiredKeys.has(key)) continue; + removeCompletedChatRegistration(registrations, key, registration); + } + for (const stage of snapshot.stages) { + if (stage.sessionFile === undefined) continue; + const key = completedChatKey(snapshot.id, stage.id); + const registration = registrations.get(key); + const existing = registry.get(snapshot.id, stage.id); + if (existing?.sessionFile === stage.sessionFile && !existing.isDisposed) continue; + if (registration !== undefined) { + removeCompletedChatRegistration(registrations, key, registration); + } else if (existing !== undefined) { + disposeCompletedChatHandle(existing); + } + const handle = createCompletedChatHandle(snapshot, 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); + } +} + +function completedChatKey(runId: string, stageId: string): string { + return `${runId}:${stageId}`; +} + +function removeCompletedChatRegistration( + registrations: Map, + key: string, + registration: CompletedChatRegistration, +): void { + registration.unregister(); + registrations.delete(key); + disposeCompletedChatHandle(registration.handle); +} + +function disposeCompletedChatHandle(handle: StageControlHandle): void { + void Promise.resolve(handle.dispose?.()).catch((error: Error) => { + 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/durable/dbos-backend.ts b/packages/workflows/src/durable/dbos-backend.ts index 738d075fe..445770a71 100644 --- a/packages/workflows/src/durable/dbos-backend.ts +++ b/packages/workflows/src/durable/dbos-backend.ts @@ -256,6 +256,7 @@ export class DbosDurableBackend implements DurableWorkflowBackend { } listResumableWorkflows(): readonly ResumableWorkflowEntry[] { return this.mem.listResumableWorkflows(); } + listCompletedWorkflows(): readonly ResumableWorkflowEntry[] { return this.mem.listCompletedWorkflows(); } toCacheEntry(workflowId: string) { return this.mem.toCacheEntry(workflowId); } reset(): void { this.mem.reset(); this.hydrated.clear(); this.writeQueue = Promise.resolve(); this.writeErrors = []; } async flush(): Promise { diff --git a/packages/workflows/src/durable/file-backend.ts b/packages/workflows/src/durable/file-backend.ts index 149872a78..f1003b056 100644 --- a/packages/workflows/src/durable/file-backend.ts +++ b/packages/workflows/src/durable/file-backend.ts @@ -126,6 +126,11 @@ export class FileDurableBackend implements DurableWorkflowBackend { return this.mem.listResumableWorkflows(); } + listCompletedWorkflows() { + this.ensureLoaded(); + return this.mem.listCompletedWorkflows(); + } + toCacheEntry(workflowId: string) { this.ensureLoaded(); return this.mem.toCacheEntry(workflowId); @@ -194,6 +199,12 @@ export class WorkflowFileDurableBackend implements DurableWorkflowBackend { return mem.listResumableWorkflows(); } + listCompletedWorkflows() { + const mem = new InMemoryDurableBackend(); + mem.importAll(mergeRecords([], this.readAllRecords())); + return mem.listCompletedWorkflows(); + } + toCacheEntry(workflowId: string) { return this.backendFor(workflowId).toCacheEntry(workflowId); } @@ -407,7 +418,7 @@ function chmodBestEffort(path: string, mode: number): void { } function isPrunableTerminalStatus(status: DurableWorkflowStatus, resumable?: boolean): boolean { - if (status === "completed" || status === "cancelled") return true; + if (status === "cancelled") return true; return (status === "failed" || status === "blocked") && resumable === false; } diff --git a/packages/workflows/src/durable/index.ts b/packages/workflows/src/durable/index.ts index 3f2915fb8..8c6fa9d07 100644 --- a/packages/workflows/src/durable/index.ts +++ b/packages/workflows/src/durable/index.ts @@ -49,6 +49,18 @@ export { persistDurableCacheEntry, formatResumableWorkflowList, } from "./resume-catalog.js"; +export { + completedWorkflowSnapshot, + listCompletedFromBackend, + listOpenableCompletedWorkflows, + resolveCompletedWorkflow, + type CompletedWorkflowResolution, +} from "./completed-catalog.js"; +export { + openCompletedDurableWorkflow, + type OpenCompletedDurableDeps, + type OpenCompletedDurableResult, +} from "./completed-inspection.js"; export { createToolPrimitive, createCheckpointIdGenerator, diff --git a/packages/workflows/src/durable/resume-catalog.ts b/packages/workflows/src/durable/resume-catalog.ts index 6c955adb1..4983f0679 100644 --- a/packages/workflows/src/durable/resume-catalog.ts +++ b/packages/workflows/src/durable/resume-catalog.ts @@ -217,13 +217,14 @@ export function persistDurableCacheEntry( * Format the resumable workflow list for display in the selector. */ export function formatResumableWorkflowList(entries: readonly ResumableWorkflowEntry[]): string { - if (entries.length === 0) return "No resumable workflows found."; - const lines = entries.map((e, i) => { - const id = e.workflowId.slice(0, 8); - const status = e.status.padEnd(8); - const checkpoints = `${e.completedCheckpoints} checkpoint${e.completedCheckpoints === 1 ? "" : "s"}`; - const label = e.label ? ` "${e.label}"` : ""; - return ` ${i + 1}. ${id} ${status} ${e.name}${label} (${checkpoints})`; + if (entries.length === 0) return "No resumable or completed workflows found."; + const hasCompleted = entries.some((entry) => entry.status === "completed"); + const lines = entries.map((entry, index) => { + const id = entry.workflowId.slice(0, 8); + const status = entry.status === "completed" ? "✓ completed" : entry.status.padEnd(8); + const checkpoints = `${entry.completedCheckpoints} checkpoint${entry.completedCheckpoints === 1 ? "" : "s"}`; + const label = entry.label ? ` "${entry.label}"` : ""; + return ` ${index + 1}. ${id} ${status} ${entry.name}${label} (${checkpoints})`; }); - return `Resumable workflows:\n${lines.join("\n")}`; + return `${hasCompleted ? "Workflow resume targets" : "Resumable workflows"}:\n${lines.join("\n")}`; } diff --git a/packages/workflows/src/durable/scoped-backend.ts b/packages/workflows/src/durable/scoped-backend.ts index 3266c08a0..5df7c2e6a 100644 --- a/packages/workflows/src/durable/scoped-backend.ts +++ b/packages/workflows/src/durable/scoped-backend.ts @@ -13,9 +13,9 @@ * child boundary key, so the same side effects are recovered on resume. * * Only checkpoint read/write methods are scoped. Lifecycle methods - * (`registerWorkflow`, `setWorkflowStatus`, `listResumableWorkflows`, + * (`registerWorkflow`, `setWorkflowStatus`, completed/resumable listing, * `toCacheEntry`, `getWorkflow`) are no-ops for scoped children because child - * runs are never independently resumable — only the root workflow is resumed. + * runs are never independently addressable — only the root workflow is. * * cross-ref: issue #1498 — child side effects under the root durable workflow. */ @@ -114,6 +114,10 @@ export class ScopedDurableBackend implements DurableWorkflowBackend { return []; } + listCompletedWorkflows(): readonly ResumableWorkflowEntry[] { + return []; + } + toCacheEntry(_workflowId: string): undefined { return undefined; } diff --git a/packages/workflows/src/extension/extension-runtime-state.ts b/packages/workflows/src/extension/extension-runtime-state.ts index 4bd52e6b9..61517f1de 100644 --- a/packages/workflows/src/extension/extension-runtime-state.ts +++ b/packages/workflows/src/extension/extension-runtime-state.ts @@ -148,6 +148,16 @@ export function createWorkflowExtensionRuntimeState( resumeDurableWorkflow(workflowIdOrPrefix, options) { return runtimeRef.current.resumeDurableWorkflow(workflowIdOrPrefix, options); }, listDurableResumable(sessionDir) { return runtimeRef.current.listDurableResumable(sessionDir); }, prepareDurableResumable(workflowIdOrPrefix, sessionDir) { return runtimeRef.current.prepareDurableResumable(workflowIdOrPrefix, sessionDir); }, + prepareCompletedDurable() { + return runtimeRef.current.prepareCompletedDurable?.() ?? Promise.resolve([]); + }, + openCompletedDurableWorkflow(workflowIdOrPrefix, catalog) { + const open = runtimeRef.current.openCompletedDurableWorkflow; + if (open === undefined) { + return { ok: false, reason: "not_found", message: `No completed durable workflow found for id/prefix: ${workflowIdOrPrefix}` }; + } + return open(workflowIdOrPrefix, catalog); + }, }; function workflowModelCatalogFromContext(ctx?: PiModelContext): WorkflowModelCatalogPort | undefined { diff --git a/packages/workflows/src/extension/runtime-durable-resume.ts b/packages/workflows/src/extension/runtime-durable-resume.ts new file mode 100644 index 000000000..da7a08c36 --- /dev/null +++ b/packages/workflows/src/extension/runtime-durable-resume.ts @@ -0,0 +1,115 @@ +import type { WorkflowExecutionPolicy } from "../shared/types.js"; +import type { StageAdapters } from "../runs/foreground/stage-runner.js"; +import type { RunOpts } from "../runs/foreground/executor.js"; +import type { Store } from "../shared/store.js"; +import type { WorkflowRegistry } from "../workflows/registry.js"; +import { + isBackendTerminal, + prepareRuntimeDurableResumable, + resumeDurableWorkflow as resumeDurableWorkflowAdapter, + type ResumeDurableDeps, + type ResumeDurableResult, +} from "../durable/resume-runtime.js"; +import { getDurableBackend } from "../durable/factory.js"; +import { scanResumableWorkflows } from "../durable/resume-catalog.js"; +import { listOpenableCompletedWorkflows } from "../durable/completed-catalog.js"; +import { + openCompletedDurableWorkflow as openCompletedSnapshot, + type OpenCompletedDurableResult, +} from "../durable/completed-inspection.js"; +import type { ResumableWorkflowEntry } from "../durable/types.js"; + +export interface DurableResumeRuntime { + resumeDurableWorkflow( + workflowIdOrPrefix: string, + options?: { readonly policy?: WorkflowExecutionPolicy }, + ): ResumeDurableResult; + listDurableResumable(sessionDir?: string): readonly ResumableWorkflowEntry[]; + prepareDurableResumable( + workflowIdOrPrefix?: string, + sessionDir?: string, + ): Promise; + prepareCompletedDurable?(): Promise; + openCompletedDurableWorkflow?( + workflowIdOrPrefix: string, + catalog?: readonly ResumableWorkflowEntry[], + ): OpenCompletedDurableResult; +} + +export interface DurableResumeRuntimeDeps { + readonly registry: WorkflowRegistry; + readonly store: Store; + readonly adapters?: StageAdapters; + readonly runtimeCwd: string; + readonly ensureReady: () => Promise; + readonly resolveDefaultStageSessionDir?: () => string | undefined; + readonly baseRunOpts: (policy?: WorkflowExecutionPolicy) => RunOpts; +} + +export function createDurableResumeRuntime( + deps: DurableResumeRuntimeDeps, +): DurableResumeRuntime { + let preparedCatalog: readonly ResumableWorkflowEntry[] = []; + return { + resumeDurableWorkflow(workflowIdOrPrefix, options): ResumeDurableResult { + const adapterDeps: ResumeDurableDeps = { + registry: deps.registry, + baseRunOpts: deps.baseRunOpts(options?.policy), + durableBackend: getDurableBackend(), + }; + return resumeDurableWorkflowAdapter(workflowIdOrPrefix, adapterDeps, preparedCatalog); + }, + listDurableResumable(sessionDir): readonly ResumableWorkflowEntry[] { + const backend = getDurableBackend(); + const live = backend.listResumableWorkflows(); + const dir = sessionDir ?? deps.resolveDefaultStageSessionDir?.(); + if (dir === undefined) return live; + const scanned = scanResumableWorkflows(dir); + const liveIds = new Set(live.map((entry) => entry.workflowId)); + const compatible = scanned.filter((entry) => + !liveIds.has(entry.workflowId) && + backend.getWorkflow(entry.workflowId) !== undefined && + !isBackendTerminal(backend, entry.workflowId) + ); + return [...live, ...compatible]; + }, + async prepareDurableResumable(workflowIdOrPrefix, sessionDir) { + await deps.ensureReady(); + preparedCatalog = await prepareRuntimeDurableResumable( + getDurableBackend, + () => deps.resolveDefaultStageSessionDir?.(), + workflowIdOrPrefix, + sessionDir, + ); + return preparedCatalog; + }, + async prepareCompletedDurable() { + await deps.ensureReady(); + const backend = getDurableBackend(); + await backend.hydrateResumableWorkflows?.(); + return listOpenableCompletedWorkflows(backend); + }, + openCompletedDurableWorkflow(workflowIdOrPrefix, catalog) { + const backend = getDurableBackend(); + const entry = resolveCatalogEntry(workflowIdOrPrefix, catalog ?? []); + const handle = backend.getWorkflow(entry?.workflowId ?? workflowIdOrPrefix); + return openCompletedSnapshot(workflowIdOrPrefix, { + durableBackend: backend, + store: deps.store, + adapters: deps.adapters, + cwd: handle?.workflowCwd ?? handle?.invocationCwd ?? deps.runtimeCwd, + defaultSessionDir: deps.resolveDefaultStageSessionDir?.(), + }, catalog); + }, + }; +} + +function resolveCatalogEntry( + workflowIdOrPrefix: string, + catalog: readonly ResumableWorkflowEntry[], +): ResumableWorkflowEntry | undefined { + const exact = catalog.find((entry) => entry.workflowId === workflowIdOrPrefix); + if (exact !== undefined) return exact; + const matches = catalog.filter((entry) => entry.workflowId.startsWith(workflowIdOrPrefix)); + return matches.length === 1 ? matches[0] : undefined; +} diff --git a/packages/workflows/src/extension/runtime.ts b/packages/workflows/src/extension/runtime.ts index f89230f41..a1f0e84d6 100644 --- a/packages/workflows/src/extension/runtime.ts +++ b/packages/workflows/src/extension/runtime.ts @@ -43,11 +43,12 @@ import { runDetached } from "../runs/background/runner.js"; import type { JobTracker } from "../runs/background/job-tracker.js"; import { appendRunEnd } from "../shared/persistence-session-entries.js"; import { classifyWorkflowFailure } from "../shared/workflow-failures.js"; -import { resumeDurableWorkflow as resumeDurableWorkflowAdapter, prepareRuntimeDurableResumable, isBackendTerminal, type ResumeDurableDeps, type ResumeDurableResult } from "../durable/resume-runtime.js"; -import { getDurableBackend, initializeDbosDurableBackendFromEnv } from "../durable/factory.js"; -import { scanResumableWorkflows } from "../durable/resume-catalog.js"; -import type { ResumableWorkflowEntry } from "../durable/types.js"; +import { initializeDbosDurableBackendFromEnv } from "../durable/factory.js"; import { directMode, directModelRequests, directOptions, directProgressTotal } from "./runtime-direct.js"; +import { + createDurableResumeRuntime, + type DurableResumeRuntime, +} from "./runtime-durable-resume.js"; // --------------------------------------------------------------------------- // Options @@ -98,7 +99,7 @@ export type ResumeFailedRunResult = | { ok: true; runId: string; sourceRunId: string; resumeFromStageId: string; message: string } | { ok: false; reason: "run_not_found" | "not_resumable" | "workflow_not_found" | "insufficient_state"; message: string }; -export interface ExtensionRuntime { +export interface ExtensionRuntime extends DurableResumeRuntime { /** * Live registry — read-only reference. * Reflects all definitions registered at startup. @@ -117,16 +118,6 @@ export interface ExtensionRuntime { /** Start a linked continuation for a failed resumable named workflow run. */ resumeFailedRun(sourceRunId: string, stageId?: string, options?: RuntimeDispatchOptions): ResumeFailedRunResult; - /** - * Resume a durable workflow by top-level workflow id when no live run exists. - * Re-dispatches the workflow with the cached inputs and original workflow id - * so durable checkpoints replay (skipping completed side effects). - * - * cross-ref: issue #1498 — cross-session /workflow resume selector. - */ - resumeDurableWorkflow(workflowIdOrPrefix: string, options?: RuntimeDispatchOptions): import("../durable/resume-runtime.js").ResumeDurableResult; - listDurableResumable(sessionDir?: string): readonly import("../durable/types.js").ResumableWorkflowEntry[]; - prepareDurableResumable(workflowIdOrPrefix?: string, sessionDir?: string): Promise; } export interface RuntimeDispatchOptions { readonly policy?: WorkflowExecutionPolicy; @@ -163,7 +154,6 @@ export function createExtensionRuntime(opts: ExtensionRuntimeOpts = {}): Extensi const resolveDefaultStageSessionDir = opts.resolveDefaultStageSessionDir; const dbosReady = initializeDbosDurableBackendFromEnv().catch((err) => process.emitWarning(`Atomic workflow DBOS durability unavailable; using file-backed durability: ${err instanceof Error ? err.message : String(err)}`)); const ensureDbosReady = async (): Promise => { await dbosReady; }; - let preparedDurableCatalog: readonly ResumableWorkflowEntry[] = []; function runOptions(args: WorkflowToolArgs, policy?: WorkflowExecutionPolicy): RunOpts { const argConcurrency = @@ -471,30 +461,15 @@ export function createExtensionRuntime(opts: ExtensionRuntimeOpts = {}): Extensi }, resumeFailedRun, + ...createDurableResumeRuntime({ + registry, + store: activeStore, + adapters, + runtimeCwd, + ensureReady: ensureDbosReady, + resolveDefaultStageSessionDir, + baseRunOpts: (policy) => runOptions({ workflow: "", inputs: {} }, policy), + }), - resumeDurableWorkflow(workflowIdOrPrefix: string, options?: RuntimeDispatchOptions): ResumeDurableResult { - const adapterDeps: ResumeDurableDeps = { - registry, - baseRunOpts: runOptions({ workflow: "", inputs: {} }, options?.policy), - durableBackend: getDurableBackend(), - }; - return resumeDurableWorkflowAdapter(workflowIdOrPrefix, adapterDeps, preparedDurableCatalog); - }, - listDurableResumable(sessionDir?: string): readonly ResumableWorkflowEntry[] { - const backend = getDurableBackend(); - const live = backend.listResumableWorkflows(); - const dir = sessionDir ?? resolveDefaultStageSessionDir?.(); - if (dir === undefined) return live; - const scanned = scanResumableWorkflows(dir); - const liveIds = new Set(live.map((e) => e.workflowId)); - const compatible = scanned.filter((e) => !liveIds.has(e.workflowId) && backend.getWorkflow(e.workflowId) !== undefined && !isBackendTerminal(backend, e.workflowId)); - return [...live, ...compatible]; - }, - - async prepareDurableResumable(workflowIdOrPrefix?: string, sessionDir?: string): Promise { - await ensureDbosReady(); - preparedDurableCatalog = await prepareRuntimeDurableResumable(getDurableBackend, () => resolveDefaultStageSessionDir?.(), workflowIdOrPrefix, sessionDir); - return preparedDurableCatalog; - }, }; } diff --git a/packages/workflows/src/extension/workflow-durable-resume-command.ts b/packages/workflows/src/extension/workflow-durable-resume-command.ts new file mode 100644 index 000000000..fe0704b73 --- /dev/null +++ b/packages/workflows/src/extension/workflow-durable-resume-command.ts @@ -0,0 +1,226 @@ +import type { WorkflowPersistencePort } from "../shared/types.js"; +import { store } from "../shared/store.js"; +import type { RunSnapshot } from "../shared/store-types.js"; +import { openWorkflowResumeSelector } from "../tui/workflow-resume-selector.js"; +import type { GraphOverlayPort } from "../tui/overlay-adapter.js"; +import { formatResumableWorkflowList } from "../durable/resume-catalog.js"; +import { getDurableBackend } from "../durable/factory.js"; +import { listOpenableCompletedWorkflows } from "../durable/completed-catalog.js"; +import { openCompletedDurableWorkflow } from "../durable/completed-inspection.js"; +import type { ResumableWorkflowEntry } from "../durable/types.js"; +import type { ExtensionRuntime } from "./runtime.js"; +import type { ExtensionAPI, PiCommandContext } from "./public-types.js"; +import type { WorkflowCommandReporter } from "./workflow-command-utils.js"; +import { workflowPolicyFromContext } from "./workflow-policy.js"; +import { formatWorkflowResourceLoadWarning } from "./workflow-command-surfaces.js"; +import { overlaySurfaceFromContext } from "./workflow-targets.js"; + +export interface WorkflowRunControlDeps { + pi: ExtensionAPI; + overlay: GraphOverlayPort; + getPersistence: () => WorkflowPersistencePort | undefined; + runtimeForContext: (ctx?: PiCommandContext) => ExtensionRuntime; + ensureWorkflowResourcesLoaded: () => Promise | void; +} + +export interface WorkflowResumeCatalog { + readonly resumable: readonly ResumableWorkflowEntry[]; + readonly completed: readonly ResumableWorkflowEntry[]; +} + +export interface WorkflowResumeTarget { + readonly kind: "live" | "durable" | "completed"; + readonly workflowId: string; + readonly name: string; +} + +export type WorkflowResumeTargetResolution = WorkflowResumeTarget + | { readonly kind: "ambiguous"; readonly matches: readonly WorkflowResumeTarget[] } + | { readonly kind: "not_found" }; + +export async function prepareWorkflowResumeCatalog( + runtime: ExtensionRuntime, + activeLiveIds: ReadonlySet, + target?: string, +): Promise { + const prepared = await runtime.prepareDurableResumable(target); + const resumable = filterSelectorDurableEntries(runtime, prepared) + .filter((entry) => !activeLiveIds.has(entry.workflowId)); + const backend = getDurableBackend(); + const completed = runtime.prepareCompletedDurable !== undefined + ? await runtime.prepareCompletedDurable() + : listOpenableCompletedWorkflows(backend); + return { + resumable, + completed: completed.filter((entry) => !activeLiveIds.has(entry.workflowId)), + }; +} + +export async function handleDurableResume( + target: string | undefined, + ctx: PiCommandContext, + reporter: WorkflowCommandReporter, + deps: WorkflowRunControlDeps, +): Promise { + const print = (message: string): void => reporter.info(message); + const fail = (message: string): void => reporter.error(message); + try { + await deps.ensureWorkflowResourcesLoaded(); + } catch (error) { + ctx.ui?.notify(formatWorkflowResourceLoadWarning(error), "warning"); + } + const runtime = deps.runtimeForContext(ctx); + const policy = workflowPolicyFromContext(ctx); + const catalog = await prepareWorkflowResumeCatalog(runtime, new Set(), target); + const allOpenable = [...catalog.resumable, ...catalog.completed]; + + if (target !== undefined) { + const resolved = resolveWorkflowResumeTarget( + target, + [], + catalog.resumable, + getDurableBackend().listCompletedWorkflows(), + ); + if (resolved.kind === "ambiguous") { + fail(`Ambiguous workflow prefix "${target}" matches: ${formatMatches(resolved.matches)}`); + return true; + } + if (resolved.kind === "completed") { + return openCompletedTarget(resolved.workflowId, catalog.completed, ctx, reporter, deps, runtime); + } + if (resolved.kind === "durable") { + return resumeDurableTarget(resolved.workflowId, ctx, reporter, deps, runtime); + } + + const completedAttempt = openCompleted(runtime, target, catalog.completed); + if (!completedAttempt.ok && completedAttempt.reason !== "not_found") { + fail(completedAttempt.message); + return true; + } + const result = runtime.resumeDurableWorkflow(target, { policy }); + fail(allOpenable.length === 0 + ? result.message + : `${result.message}\n\n${formatResumableWorkflowList(allOpenable)}`); + return true; + } + + if (allOpenable.length === 0) { + fail("No resumable or completed durable workflows found. Usage: /workflow resume (or /resume for Atomic sessions)."); + return true; + } + if (!policy.allowInputPicker) { + const instruction = catalog.completed.length === 0 ? "Resume with" : "Resume/open with"; + print(`${formatResumableWorkflowList(allOpenable)}\n\n${instruction}: /workflow resume `); + return true; + } + const picked = await openWorkflowResumeSelector(ctx.ui, [], catalog.resumable, catalog.completed); + if (picked.kind === "durable") { + return resumeDurableTarget(picked.workflowId, ctx, reporter, deps, runtime); + } + if (picked.kind === "completed") { + return openCompletedTarget(picked.workflowId, catalog.completed, ctx, reporter, deps, runtime); + } + return true; +} + +function filterSelectorDurableEntries( + runtime: ExtensionRuntime, + entries: readonly ResumableWorkflowEntry[], +): readonly ResumableWorkflowEntry[] { + const registry = runtime.registry as { has(name: string): boolean } | undefined; + if (registry === undefined) return entries; + return entries.filter((entry) => { + const requiresDefinition = entry.status === "running" || entry.status === "failed" || entry.status === "blocked"; + return !requiresDefinition || registry.has(entry.name); + }); +} + +export function resolveWorkflowResumeTarget( + target: string, + liveRuns: readonly RunSnapshot[], + resumable: readonly ResumableWorkflowEntry[], + completed: readonly ResumableWorkflowEntry[], +): WorkflowResumeTargetResolution { + const targets = new Map(); + for (const entry of resumable) { + targets.set(entry.workflowId, { kind: "durable", workflowId: entry.workflowId, name: entry.name }); + } + for (const entry of completed) { + targets.set(entry.workflowId, { kind: "completed", workflowId: entry.workflowId, name: entry.name }); + } + for (const run of liveRuns.filter(isExplicitResumeCandidate)) { + targets.set(run.id, { + kind: run.status === "completed" ? "completed" : "live", + workflowId: run.id, + name: run.name, + }); + } + const exact = targets.get(target); + if (exact !== undefined) return exact; + const matches = [...targets.values()].filter((candidate) => candidate.workflowId.startsWith(target)); + if (matches.length === 0) return { kind: "not_found" }; + if (matches.length === 1) return matches[0]!; + return { kind: "ambiguous", matches }; +} + +function isExplicitResumeCandidate(run: RunSnapshot): boolean { + if (run.status === "completed" || run.status === "paused" || run.exitReason === "quit") return true; + if (run.stages.some((stage) => stage.status === "paused")) return true; + if (run.status === "failed") return run.resumable !== false; + if (run.endedAt !== undefined) return false; + if (run.status === "running") return true; + return run.resumable === true && run.failureRecoverability === "recoverable"; +} + +function resumeDurableTarget( + workflowId: string, + ctx: PiCommandContext, + reporter: WorkflowCommandReporter, + deps: WorkflowRunControlDeps, + runtime: ExtensionRuntime, +): boolean { + const result = runtime.resumeDurableWorkflow(workflowId, { policy: workflowPolicyFromContext(ctx) }); + if (!result.ok) reporter.error(result.message); + else { + reporter.info(result.message); + if (workflowPolicyFromContext(ctx).allowInputPicker) { + deps.overlay.open(result.runId, overlaySurfaceFromContext(ctx)); + } + } + return true; +} + +function openCompletedTarget( + workflowId: string, + catalog: readonly ResumableWorkflowEntry[], + ctx: PiCommandContext, + reporter: WorkflowCommandReporter, + deps: WorkflowRunControlDeps, + runtime: ExtensionRuntime, +): boolean { + const result = openCompleted(runtime, workflowId, catalog); + if (!result.ok) reporter.error(result.message); + else { + reporter.info(result.message); + if (workflowPolicyFromContext(ctx).allowInputPicker) { + deps.overlay.open(result.runId, overlaySurfaceFromContext(ctx)); + } + } + return true; +} + +function openCompleted( + runtime: ExtensionRuntime, + workflowIdOrPrefix: string, + catalog: readonly ResumableWorkflowEntry[], +) { + return runtime.openCompletedDurableWorkflow?.(workflowIdOrPrefix, catalog) + ?? openCompletedDurableWorkflow(workflowIdOrPrefix, { + durableBackend: getDurableBackend(), + store, + }, catalog); +} + +function formatMatches(entries: readonly WorkflowResumeTarget[]): string { + return entries.map((entry) => `${entry.name} (${entry.workflowId.slice(0, 8)})`).join(", "); +} diff --git a/packages/workflows/src/extension/workflow-run-control-command.ts b/packages/workflows/src/extension/workflow-run-control-command.ts index b94211373..4cdd49139 100644 --- a/packages/workflows/src/extension/workflow-run-control-command.ts +++ b/packages/workflows/src/extension/workflow-run-control-command.ts @@ -1,20 +1,17 @@ import { cancellationRegistry } from "../runs/background/cancellation-registry.js"; import { interruptAllRuns, interruptRun, killAllRuns, killRun, pauseRun, resumeRun } from "../runs/background/status.js"; -import type { WorkflowPersistencePort } from "../shared/types.js"; +import { getDurableBackend } from "../durable/factory.js"; import { store } from "../shared/store.js"; import { topLevelWorkflowRuns } from "../shared/run-visibility.js"; import { renderSessionList } from "../tui/session-list.js"; import { openKillConfirm, openSessionPicker } from "../tui/session-overlays.js"; -import { openWorkflowResumeSelector } from "../tui/workflow-resume-selector.js"; import { deriveGraphTheme } from "../tui/graph-theme.js"; +import { openWorkflowResumeSelector } from "../tui/workflow-resume-selector.js"; import { emitChatSurface } from "../tui/chat-surface-message.js"; -import type { GraphOverlayPort } from "../tui/overlay-adapter.js"; -import type { ExtensionRuntime } from "./runtime.js"; -import type { ExtensionAPI, PiCommandContext } from "./public-types.js"; +import type { PiCommandContext } from "./public-types.js"; import type { WorkflowCommandReporter } from "./workflow-command-utils.js"; import { stripYesFlag } from "./workflow-command-utils.js"; import { workflowPolicyFromContext } from "./workflow-policy.js"; -import { formatResumableWorkflowList } from "../durable/resume-catalog.js"; import type { ResumableWorkflowEntry } from "../durable/types.js"; import { formatAlreadyEndedRetainedMessage, @@ -23,14 +20,14 @@ import { resolveStageTarget, } from "./workflow-targets.js"; import { formatWorkflowResourceLoadWarning } from "./workflow-command-surfaces.js"; +import { + handleDurableResume, + prepareWorkflowResumeCatalog, + resolveWorkflowResumeTarget, + type WorkflowRunControlDeps, +} from "./workflow-durable-resume-command.js"; -export interface WorkflowRunControlDeps { - pi: ExtensionAPI; - overlay: GraphOverlayPort; - getPersistence: () => WorkflowPersistencePort | undefined; - runtimeForContext: (ctx?: PiCommandContext) => ExtensionRuntime; - ensureWorkflowResourcesLoaded: () => Promise | void; -} +export type { WorkflowRunControlDeps } from "./workflow-durable-resume-command.js"; function resolveAttachStageId(runId: string, stageTarget: string | undefined): string | undefined | false { if (!stageTarget) return undefined; @@ -42,79 +39,6 @@ function resolveAttachStageId(runId: string, stageTarget: string | undefined): s return byName?.id ?? false; } -function filterSelectorDurableEntries( - runtime: ExtensionRuntime, - entries: readonly ResumableWorkflowEntry[], -): readonly ResumableWorkflowEntry[] { - const registry = runtime.registry as { has(name: string): boolean } | undefined; - if (registry === undefined) return entries; - return entries.filter((entry) => { - const requiresCurrentDefinition = entry.status === "running" || entry.status === "failed" || entry.status === "blocked"; - return !requiresCurrentDefinition || registry.has(entry.name); - }); -} - - -async function handleDurableResume( - target: string | undefined, - ctx: PiCommandContext, - reporter: WorkflowCommandReporter, - deps: WorkflowRunControlDeps, -): Promise { - const print = (msg: string): void => reporter.info(msg); - const fail = (msg: string): void => reporter.error(msg); - try { - await deps.ensureWorkflowResourcesLoaded(); - } catch (error) { - ctx.ui?.notify(formatWorkflowResourceLoadWarning(error), "warning"); - } - const runtime = deps.runtimeForContext(ctx); - const policy = workflowPolicyFromContext(ctx); - // Hydrate the durable backend from DBOS (if configured) before listing so a - // fresh process discovers workflows persisted by a prior session. - const prepared = await runtime.prepareDurableResumable(target); - const durable = filterSelectorDurableEntries(runtime, prepared); - if (target !== undefined) { - // Attempt resume by id/prefix against the durable catalog. - const result = runtime.resumeDurableWorkflow(target, { policy }); - if (result.ok) { - print(result.message); - // Open/connect the overlay to the resumed run, analogous to live resume. - if (policy.allowInputPicker) deps.overlay.open(result.runId, overlaySurfaceFromContext(ctx)); - return true; - } - // Not a durable workflow either — surface the catalog for discovery. - if (durable.length > 0) { - fail(`${result.message}\n\n${formatResumableWorkflowList(durable)}`); - } else { - fail(result.message); - } - return true; - } - // No target: show the durable selector when interactive, otherwise print. - if (durable.length === 0) { - fail("No resumable durable workflows found. Usage: /workflow resume (or /resume for Atomic sessions)."); - return true; - } - if (!policy.allowInputPicker) { - print(`${formatResumableWorkflowList(durable)}\n\nResume with: /workflow resume `); - return true; - } - const picked = await openWorkflowResumeSelector(ctx.ui, [], durable); - if (picked.kind === "durable") { - const result = runtime.resumeDurableWorkflow(picked.workflowId, { policy }); - if (result.ok) { - print(result.message); - if (policy.allowInputPicker) deps.overlay.open(result.runId, overlaySurfaceFromContext(ctx)); - } else { - fail(result.message); - } - } - if (picked.kind !== "durable") { - print(`${formatResumableWorkflowList(durable)}\n\nResume with: /workflow resume `); - } - return true; -} export async function handleRunControlCommand( action: "connect" | "interrupt" | "kill" | "attach" | "pause" | "resume", @@ -307,30 +231,30 @@ export async function handleRunControlCommand( const liveRuns = topLevelWorkflowRuns(store.runs()).filter((run) => run.status === "paused" || (run.status === "failed" && run.resumable !== false), ); + const activeLiveIds = new Set( + topLevelWorkflowRuns(store.runs()) + .filter((run) => run.endedAt === undefined && run.status === "running" && run.exitReason !== "quit") + .map((run) => run.id), + ); + if (liveRuns.length === 0) await ensureWorkflowResourcesVisible(); + const runtime = deps.runtimeForContext(ctx); let durableEntries: readonly ResumableWorkflowEntry[] = []; - if (liveRuns.length === 0) { - // Durable entries: a `running` durable handle may be a crashed process - // (cross-session crash recovery), so it stays selectable UNLESS it - // matches an actively-executing live run in this session. - const activeLiveIds = new Set( - topLevelWorkflowRuns(store.runs()) - .filter((run) => run.endedAt === undefined && run.status === "running" && run.exitReason !== "quit") - .map((run) => run.id), - ); - await ensureWorkflowResourcesVisible(); - const runtime = deps.runtimeForContext(ctx); - try { - const prepared = await runtime.prepareDurableResumable(undefined); - durableEntries = filterSelectorDurableEntries(runtime, prepared) - .filter((entry) => !activeLiveIds.has(entry.workflowId)); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - fail(`Failed to list resumable workflows: ${message}`); + let completedEntries: readonly ResumableWorkflowEntry[] = []; + try { + const catalog = await prepareWorkflowResumeCatalog(runtime, activeLiveIds); + durableEntries = catalog.resumable; + completedEntries = catalog.completed; + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + if (liveRuns.length === 0) { + fail(`Failed to list workflow resume targets: ${errorMessage}`); return true; } } - const picked = await openWorkflowResumeSelector(ctx.ui, liveRuns, durableEntries); - if (picked.kind === "durable") return await handleDurableResume(picked.workflowId, ctx, reporter, deps); + const picked = await openWorkflowResumeSelector(ctx.ui, liveRuns, durableEntries, completedEntries); + if (picked.kind === "durable" || picked.kind === "completed") { + return await handleDurableResume(picked.workflowId, ctx, reporter, deps); + } if (picked.kind === "live") { const resolved = resolveRunIdPrefix(picked.runId); if (resolved.kind !== "exact") { @@ -361,20 +285,55 @@ export async function handleRunControlCommand( runId = picked.runId; } else { const resolved = resolveRunIdPrefix(target); - if (resolved.kind === "not_found") { - // Not a live run — fall back to the cross-session durable resume catalog. - // cross-ref: issue #1498 — /workflow resume by top-level workflow id. - if (action === "resume") { - return await handleDurableResume(target, ctx, reporter, deps); + const exactLocal = store.runs().find((run) => run.id === target); + if (action === "resume" && exactLocal?.status === "completed") { + return await handleDurableResume(target, ctx, reporter, deps); + } + if (action === "resume" && exactLocal === undefined) { + try { + await ensureWorkflowResourcesVisible(); + const runtime = deps.runtimeForContext(ctx); + const durable = await runtime.prepareDurableResumable(target); + const combined = resolveWorkflowResumeTarget( + target, + topLevelWorkflowRuns(store.runs()), + durable, + getDurableBackend().listCompletedWorkflows(), + ); + if (combined.kind === "ambiguous") { + fail(`Ambiguous workflow prefix "${target}" matches: ${combined.matches.map((match) => `${match.name} (${match.workflowId.slice(0, 8)})`).join(", ")}`); + return true; + } + if (combined.kind === "completed" || combined.kind === "durable") { + return await handleDurableResume(combined.workflowId, ctx, reporter, deps); + } + if (combined.kind === "live") runId = combined.workflowId; + else if (resolved.kind === "not_found") return await handleDurableResume(target, ctx, reporter, deps); + else if (resolved.kind === "ambiguous") { + fail(`Ambiguous run prefix "${target}" matches: ${resolved.matches.map((id) => id.slice(0, 12)).join(", ")}`); + return true; + } else runId = resolved.runId; + } catch (error) { + if (resolved.kind === "not_found") { + fail(`Failed to resolve workflow resume target: ${error instanceof Error ? error.message : String(error)}`); + return true; + } + if (resolved.kind === "ambiguous") { + fail(`Ambiguous run prefix "${target}" matches: ${resolved.matches.map((id) => id.slice(0, 12)).join(", ")}`); + return true; + } + runId = resolved.runId; } + } else if (resolved.kind === "not_found") { + if (action === "resume") return await handleDurableResume(target, ctx, reporter, deps); fail(`Run not found: ${target}`); return true; - } - if (resolved.kind === "ambiguous") { + } else if (resolved.kind === "ambiguous") { fail(`Ambiguous run prefix "${target}" matches: ${resolved.matches.map((id) => id.slice(0, 12)).join(", ")}`); return true; + } else { + runId = resolved.runId; } - runId = resolved.runId; } if (action === "attach") { const stageId = resolveAttachStageId(runId, stageTarget); diff --git a/packages/workflows/src/tui/workflow-resume-selector.ts b/packages/workflows/src/tui/workflow-resume-selector.ts index a98504cdd..dd4cfb74a 100644 --- a/packages/workflows/src/tui/workflow-resume-selector.ts +++ b/packages/workflows/src/tui/workflow-resume-selector.ts @@ -10,6 +10,7 @@ import type { RunSnapshot, StageSnapshot } from "../shared/store-types.js"; export type WorkflowResumeSelectorResult = | { kind: "live"; runId: string } | { kind: "durable"; workflowId: string } + | { kind: "completed"; workflowId: string } | { kind: "close" }; export interface WorkflowResumeSelectorUiSurface { @@ -54,47 +55,67 @@ function liveRunSession(run: RunSnapshot): WorkflowResumeSelectorItem { }; } -function durableWorkflowSession(entry: ResumableWorkflowEntry): WorkflowResumeSelectorItem { +function durableWorkflowSession( + entry: ResumableWorkflowEntry, + kind: "durable" | "completed", +): WorkflowResumeSelectorItem { const checkpointText = `${entry.completedCheckpoints} checkpoints`; const promptText = `${entry.pendingPrompts} prompts`; - const firstMessage = `${entry.name} ${entry.status} ${checkpointText} ${promptText}`; + const statusText = kind === "completed" ? "✓ completed" : entry.status; return { - result: { kind: "durable", workflowId: entry.workflowId }, + result: { kind, workflowId: entry.workflowId }, session: { - path: `workflow-durable:${entry.workflowId}`, + path: `workflow-${kind}:${entry.workflowId}`, id: entry.workflowId, - cwd: "Durable workflow runs", + cwd: kind === "completed" ? "Completed workflow runs" : "Durable workflow runs", created: new Date(entry.createdAt), modified: new Date(entry.updatedAt), messageCount: entry.completedCheckpoints, - firstMessage, - allMessagesText: `${entry.workflowId} ${entry.name} ${entry.status} ${checkpointText} ${promptText}`, + firstMessage: `${entry.name} ${statusText} ${checkpointText} ${promptText}`, + allMessagesText: `${entry.workflowId} ${entry.name} ${statusText} ${checkpointText} ${promptText}`, + ...(kind === "completed" ? { messageColor: "success" as const } : {}), }, }; } +function compareResumeItemsByRecency( + left: WorkflowResumeSelectorItem, + right: WorkflowResumeSelectorItem, +): number { + const recencyDifference = right.session.modified.getTime() - left.session.modified.getTime(); + if (recencyDifference !== 0) return recencyDifference; + const idDifference = left.session.id.localeCompare(right.session.id); + return idDifference !== 0 ? idDifference : left.session.path.localeCompare(right.session.path); +} + export function workflowResumeSelectorItems( liveRuns: readonly RunSnapshot[], durableEntries: readonly ResumableWorkflowEntry[], + completedEntries: readonly ResumableWorkflowEntry[] = [], ): WorkflowResumeSelectorItem[] { const liveIds = new Set(liveRuns.map((run) => run.id)); + const durableIds = new Set(durableEntries.map((entry) => entry.workflowId)); return [ ...liveRuns.map(liveRunSession), ...durableEntries .filter((entry) => !liveIds.has(entry.workflowId)) - .map(durableWorkflowSession), - ]; + .map((entry) => durableWorkflowSession(entry, "durable")), + ...completedEntries + .filter((entry) => !liveIds.has(entry.workflowId) && !durableIds.has(entry.workflowId)) + .map((entry) => durableWorkflowSession(entry, "completed")), + ].sort(compareResumeItemsByRecency); } export function openWorkflowResumeSelector( ui: WorkflowResumeSelectorUiSurface, liveRuns: readonly RunSnapshot[], durableEntries: readonly ResumableWorkflowEntry[], + completedEntries: readonly ResumableWorkflowEntry[] = [], ): Promise { const custom = ui.custom; if (typeof custom !== "function") return Promise.resolve({ kind: "close" }); - const items = workflowResumeSelectorItems(liveRuns, durableEntries); + const items = workflowResumeSelectorItems(liveRuns, durableEntries, completedEntries); const resultByPath = new Map(items.map((item) => [item.session.path, item.result])); const sessions = items.map((item) => item.session); @@ -108,8 +129,11 @@ export function openWorkflowResumeSelector( const settle = (result: WorkflowResumeSelectorResult, done?: (result: undefined) => void): void => { if (settled) return; settled = true; - resolve(result); - done?.(undefined); + try { + done?.(undefined); + } finally { + resolve(result); + } }; const factory = ( @@ -146,6 +170,12 @@ export function openWorkflowResumeSelector( }; }; - void custom(factory, { overlay: false }); + try { + void Promise.resolve(custom(factory, { overlay: false })).catch(() => { + settle({ kind: "close" }); + }); + } catch { + settle({ kind: "close" }); + } }); } diff --git a/test/integration/overlay-entrypoints-commands.test.ts b/test/integration/overlay-entrypoints-commands.test.ts index 9695902d4..1467733a7 100644 --- a/test/integration/overlay-entrypoints-commands.test.ts +++ b/test/integration/overlay-entrypoints-commands.test.ts @@ -1,5 +1,8 @@ import { beforeEach, afterEach, 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 { InMemoryDurableBackend } from "../../packages/workflows/src/durable/backend.js"; import { setDurableBackend } from "../../packages/workflows/src/durable/factory.js"; import { @@ -31,6 +34,21 @@ import { void [buildGraphOverlayAdapter, buildInteractiveHostCustomUi, buildMockPi, buildMockUi, buildOverlayHandle, buildPrintCtx, buildPrintCtxWithRealCustom, attachHostCustomUiState, createCancellationRegistry, createJobTracker, createStore, workflow, delay, factory, runDetached, setupBranchingRun, setupSequentialRun, setupWideFanoutRun, singletonStore, Type, visibleText, waitForRenderCount, waitForRunEnded, waitForStagePendingPrompt]; +function registerInspectableCompleted(backend: InMemoryDurableBackend, workflowId: string, name: string): () => void { + const dir = mkdtempSync(join(tmpdir(), "atomic-completed-overlay-")); + const sessionFile = join(dir, "stage.jsonl"); + writeFileSync(sessionFile, [ + JSON.stringify({ type: "session", version: 3, id: `${workflowId}-session`, timestamp: new Date().toISOString(), cwd: dir }), + JSON.stringify({ type: "message", id: `${workflowId}-message`, parentId: null, timestamp: new Date().toISOString(), message: { role: "user", content: "prior context", timestamp: Date.now() } }), + ].join("\n") + "\n"); + backend.registerWorkflow({ workflowId, name, inputs: {}, createdAt: 1, status: "completed" }); + backend.recordCheckpoint({ + kind: "stage", workflowId, checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", output: "done", sessionFile, completedAt: 2, + }); + return () => rmSync(dir, { recursive: true, force: true }); +} + describe("/workflow resume — overlay integration", () => { beforeEach(() => { setDurableBackend(new InMemoryDurableBackend()); @@ -146,7 +164,7 @@ describe("/workflow resume — overlay integration", () => { ); }); - test("resume with no runId prioritizes live picker when live runs exist", async () => { + test("resume with no runId mixes live and durable entries", async () => { singletonStore.clear(); const liveRunId = `live-run-${Date.now()}`; singletonStore.recordRunStart({ id: liveRunId, name: "live-wf", inputs: {}, status: "running", stages: [], startedAt: Date.now() }); @@ -163,7 +181,7 @@ describe("/workflow resume — overlay integration", () => { assert.ok(customCalls.length >= 1); const text = visibleText(customCalls[0]!.component.render(80)).replace(/\n/g, " "); assert.match(text, /live-wf/); - assert.doesNotMatch(text, /durable-cross-session/); + assert.match(text, /durable-cross-session/); customCalls[0]!.component.handleInput?.("\u001b"); await handlerPromise; } finally { @@ -171,29 +189,36 @@ describe("/workflow resume — overlay integration", () => { } }); - test("resume with known completed runId calls overlay.open", async () => { + test("resume with known authoritative completed runId calls overlay.open", async () => { + singletonStore.clear(); const runId = `test-resume-run-${Date.now()}`; - + const backend = new InMemoryDurableBackend(); + const cleanup = registerInspectableCompleted(backend, runId, "test-wf"); + setDurableBackend(backend); singletonStore.recordRunStart({ id: runId, - name: "test-wf", + name: "stale-local", inputs: {}, - status: "running", + status: "completed", stages: [], - startedAt: Date.now(), + startedAt: 1, + endedAt: 2, + resumable: false, }); - singletonStore.recordRunEnd(runId, "completed", {}); - - const { pi, commands, customCalls } = buildMockPi(); - factory(pi); - - const wfCmd = commands["workflow"]!; - const { ctx } = buildPrintCtx(); + try { + const { pi, commands, customCalls } = buildMockPi(); + factory(pi); + const { ctx } = buildPrintCtx(); - await wfCmd.options.handler(`resume ${runId}`, ctx); + await commands["workflow"]!.options.handler(`resume ${runId}`, ctx); - assert.ok(customCalls.length >= 1); - assert.equal(customCalls[0]!.options.overlay, true); + assert.ok(customCalls.length >= 1); + assert.equal(customCalls[0]!.options.overlay, true); + assert.equal(singletonStore.runs()[0]?.name, "test-wf"); + assert.equal(singletonStore.runs()[0]?.stages[0]?.name, "final"); + } finally { + cleanup(); + } }); test("resume of an actively-running run is refused (use /workflow connect)", async () => { @@ -289,25 +314,27 @@ describe("/workflow pause — top-level command", () => { }); describe("/workflow resume — paused vs non-paused branching", () => { - test("resume on a non-paused run still reopens the overlay", async () => { + beforeEach(() => setDurableBackend(new InMemoryDurableBackend())); + afterEach(() => setDurableBackend(undefined)); + test("resume refuses a completed local snapshot without authoritative data", async () => { singletonStore.clear(); const runId = `test-non-paused-${Date.now()}`; singletonStore.recordRunStart({ id: runId, name: "snap-only-wf", inputs: {}, - status: "running", + status: "completed", stages: [], - startedAt: Date.now(), + startedAt: 1, + endedAt: 2, + resumable: false, }); - singletonStore.recordRunEnd(runId, "completed", {}); const { pi, commands, customCalls } = buildMockPi(); factory(pi); - const wfCmd = commands["workflow"]!; - const { ctx } = buildPrintCtx(); - await wfCmd.options.handler(`resume ${runId}`, ctx); - assert.ok(customCalls.length >= 1); - assert.equal(customCalls[0]!.options.overlay, true); + const { ctx, messages } = buildPrintCtx(); + await commands["workflow"]!.options.handler(`resume ${runId}`, ctx); + assert.equal(customCalls.length, 0); + assert.match(messages.join("\n"), /No durable workflow|No completed durable workflow|stale/); }); }); @@ -361,7 +388,7 @@ describe("/workflow attach — top-level command", () => { assert.equal(customCalls.length, 0); }); - test("no-arg resume with live + durable opens the live /resume-style selector without durable discovery", async () => { + test("no-arg resume with live + durable opens one mixed /resume-style selector", async () => { singletonStore.clear(); const liveRunId = `live-combined-${Date.now()}`; singletonStore.recordRunStart({ id: liveRunId, name: "live-wf", inputs: {}, status: "running", stages: [], startedAt: Date.now() }); @@ -379,7 +406,7 @@ describe("/workflow attach — top-level command", () => { assert.equal(customCalls[0]!.options.overlay, false); const text = visibleText(customCalls[0]!.component.render(80)).replace(/\n/g, " "); assert.match(text, /live-wf/); - assert.doesNotMatch(text, /durable-wf/); + assert.match(text, /durable-wf/); customCalls[0]!.component.handleInput?.("\u001b"); await handlerPromise; } finally { @@ -437,7 +464,7 @@ describe("/workflow attach — top-level command", () => { }); - test("no-arg resume with live runs skips async durable hydration", async () => { + test("no-arg resume with live runs includes asynchronously hydrated durable entries", async () => { singletonStore.clear(); const liveRunId = `live-hydrate-${Date.now()}`; singletonStore.recordRunStart({ id: liveRunId, name: "live-hydrate-wf", inputs: {}, status: "running", stages: [], startedAt: Date.now() }); @@ -454,7 +481,7 @@ describe("/workflow attach — top-level command", () => { assert.ok(customCalls.length >= 1); const text = visibleText(customCalls[0]!.component.render(80)).replace(/\n/g, " "); assert.match(text, /live-hydrate-wf/); - assert.doesNotMatch(text, /durable-hydrate/); + assert.match(text, /durable-hydrate/); customCalls[0]!.component.handleInput?.("\u001b"); await handlerPromise; } finally { diff --git a/test/unit/durable-backend.test.ts b/test/unit/durable-backend.test.ts index 5f018a176..809f0076e 100644 --- a/test/unit/durable-backend.test.ts +++ b/test/unit/durable-backend.test.ts @@ -86,15 +86,17 @@ describe("InMemoryDurableBackend", () => { assert.equal(cps[1]!.checkpointId, "cp-1"); }); - test("listResumableWorkflows includes running/paused after checkpoint progress", () => { + test("keeps completed workflows out of resumable listing and in completed listing", () => { // A `running` durable handle may belong to a crashed process (cross-session // crash recovery), so it is resumable at the backend level alongside // `paused`. Same-session double-resume is filtered by the command layer. assert.equal(backend.listResumableWorkflows().length, 0); + assert.equal(backend.listCompletedWorkflows().length, 0); backend.recordCheckpoint(makeToolCheckpoint(WORKFLOW_ID, "progress", "h-progress", "done")); assert.equal(backend.listResumableWorkflows().length, 1); backend.setWorkflowStatus(WORKFLOW_ID, "completed"); assert.equal(backend.listResumableWorkflows().length, 0); + assert.deepEqual(backend.listCompletedWorkflows().map((entry) => entry.workflowId), [WORKFLOW_ID]); }); test("listResumableWorkflows filters children and non-recoverable failures", () => { @@ -240,14 +242,15 @@ describe("WorkflowFileDurableBackend", () => { assert.deepEqual(ids, ["wf-a", "wf-b"]); }); - test("prunes completed workflow files", () => { + test("retains completed workflow files for authoritative inspection", () => { const backend = new WorkflowFileDurableBackend(tmpDir); backend.registerWorkflow({ workflowId: "wf-done", name: "done", inputs: {}, createdAt: 1, status: "running" }); backend.recordCheckpoint(makeToolCheckpoint("wf-done", "done", "hash-done", "ok", "cp-done")); backend.setWorkflowStatus("wf-done", "completed"); - assert.equal(existsSync(durableStateFileFor(tmpDir, "wf-done")), false); + assert.equal(existsSync(durableStateFileFor(tmpDir, "wf-done")), true); assert.equal(backend.listResumableWorkflows().length, 0); + assert.deepEqual(backend.listCompletedWorkflows().map((entry) => entry.workflowId), ["wf-done"]); }); test("reset clears workflow files without wiping unrelated durable-root files", () => { diff --git a/test/unit/durable-completed-catalog.test.ts b/test/unit/durable-completed-catalog.test.ts new file mode 100644 index 000000000..7ebd81b75 --- /dev/null +++ b/test/unit/durable-completed-catalog.test.ts @@ -0,0 +1,217 @@ +import { afterEach, beforeEach, describe, test } from "bun:test"; +import assert from "node:assert/strict"; +import { mkdirSync, 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 { + completedWorkflowSnapshot, + listCompletedFromBackend, + listOpenableCompletedWorkflows, + resolveCompletedWorkflow, +} from "../../packages/workflows/src/durable/completed-catalog.js"; +import { listResumableFromBackend } from "../../packages/workflows/src/durable/resume-catalog.js"; + +let tempDir = ""; + +beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), "atomic-completed-catalog-")); }); +afterEach(() => { rmSync(tempDir, { recursive: true, force: true }); }); + +function writeSessionTranscript(path: string, id: string): void { + writeFileSync(path, [ + JSON.stringify({ type: "session", version: 3, id, timestamp: new Date().toISOString(), cwd: tempDir }), + JSON.stringify({ type: "message", id: `${id}-message`, parentId: null, timestamp: new Date().toISOString(), message: { role: "user", content: "prior context", timestamp: Date.now() } }), + ].join("\n") + "\n"); +} + +function registerCompleted(backend: InMemoryDurableBackend, id: string): void { + backend.registerWorkflow({ + workflowId: id, + name: "completed-flow", + inputs: { topic: "done" }, + createdAt: 10, + updatedAt: 30, + status: "completed", + }); +} + +describe("completed durable catalog", () => { + test("keeps completed listing distinct from resumability predicates", () => { + const backend = new InMemoryDurableBackend(); + backend.registerWorkflow({ workflowId: "paused", name: "paused-flow", inputs: {}, createdAt: 1, status: "paused", completedCheckpoints: 1 }); + registerCompleted(backend, "completed"); + backend.recordCheckpoint({ kind: "tool", workflowId: "completed", checkpointId: "tool:1", name: "read", argsHash: "hash", output: "ok", completedAt: 20 }); + + assert.deepEqual(listResumableFromBackend(backend).map((entry) => entry.workflowId), ["paused"]); + assert.deepEqual(listCompletedFromBackend(backend).map((entry) => entry.workflowId), ["completed"]); + }); + + test("filters stale completed rows and reconstructs authoritative stage detail", () => { + const backend = new InMemoryDurableBackend(); + const transcript = join(tempDir, "stage.jsonl"); + writeSessionTranscript(transcript, "valid-session"); + registerCompleted(backend, "valid-completed"); + backend.recordCheckpoint({ + kind: "stage", + workflowId: "valid-completed", + checkpointId: "stage:1", + name: "summarize", + replayKey: "stage:summarize:1", + output: "finished", + sessionFile: transcript, + model: "provider/model", + completedAt: 20, + }); + registerCompleted(backend, "stale-completed"); + backend.recordCheckpoint({ + kind: "stage", + workflowId: "stale-completed", + checkpointId: "stage:1", + name: "missing", + replayKey: "stage:missing:1", + sessionFile: join(tempDir, "missing.jsonl"), + completedAt: 20, + }); + + assert.deepEqual(listOpenableCompletedWorkflows(backend).map((entry) => entry.workflowId), ["valid-completed"]); + const snapshot = completedWorkflowSnapshot(backend, listCompletedFromBackend(backend)[0]!); + assert.equal(snapshot?.status, "completed"); + assert.equal(snapshot?.stages[0]?.result, "finished"); + assert.equal(snapshot?.stages[0]?.model, "provider/model"); + assert.equal(resolveCompletedWorkflow("stale", backend).kind, "stale"); + }); + + test("hides completed rows without a reopenable retained conversation", () => { + const backend = new InMemoryDurableBackend(); + const cases = [ + { id: "no-session", sessionFile: undefined }, + { id: "empty-session", sessionFile: join(tempDir, "empty.jsonl") }, + { id: "malformed-session", sessionFile: join(tempDir, "malformed.jsonl") }, + { id: "directory-session", sessionFile: join(tempDir, "directory.jsonl") }, + { id: "header-only", sessionFile: join(tempDir, "header-only.jsonl") }, + { id: "invalid-message", sessionFile: join(tempDir, "invalid-message.jsonl") }, + ] as const; + writeFileSync(cases[1].sessionFile, ""); + writeFileSync(cases[2].sessionFile, "not-json\n"); + mkdirSync(cases[3].sessionFile); + writeFileSync(cases[4].sessionFile, `${JSON.stringify({ type: "session", id: "header-only" })}\n`); + writeFileSync(cases[5].sessionFile, [ + JSON.stringify({ type: "session", id: "invalid-message" }), + JSON.stringify({ type: "message" }), + ].join("\n")); + for (const item of cases) { + registerCompleted(backend, item.id); + backend.recordCheckpoint({ + kind: "stage", + workflowId: item.id, + checkpointId: "stage:1", + name: "final", + replayKey: "stage:final:1", + ...(item.sessionFile === undefined ? {} : { sessionFile: item.sessionFile }), + completedAt: 20, + }); + } + registerCompleted(backend, "tool-only"); + backend.recordCheckpoint({ kind: "tool", workflowId: "tool-only", checkpointId: "tool:1", name: "read", argsHash: "hash", output: "ok", completedAt: 20 }); + + assert.deepEqual(listOpenableCompletedWorkflows(backend), []); + for (const item of [...cases, { id: "tool-only" }]) { + assert.equal(resolveCompletedWorkflow(item.id, backend).kind, "stale"); + } + }); + + test("validates the retained transcript after merging repeated stage checkpoints", () => { + const backend = new InMemoryDurableBackend(); + const validTranscript = join(tempDir, "retained.jsonl"); + writeSessionTranscript(validTranscript, "retained-session"); + registerCompleted(backend, "merged-stage"); + backend.recordCheckpoint({ + kind: "stage", workflowId: "merged-stage", checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", sessionFile: join(tempDir, "obsolete-missing.jsonl"), completedAt: 20, + }); + backend.recordCheckpoint({ + kind: "stage", workflowId: "merged-stage", checkpointId: "stage:2", name: "final", + replayKey: "stage:final:1", sessionFile: validTranscript, output: "done", completedAt: 30, + }); + + assert.deepEqual(listOpenableCompletedWorkflows(backend).map((entry) => entry.workflowId), ["merged-stage"]); + assert.equal(completedWorkflowSnapshot(backend, listCompletedFromBackend(backend)[0]!)?.stages[0]?.sessionFile, validTranscript); + }); + + test("keeps a completed workflow when at least one stage has a usable transcript", () => { + const backend = new InMemoryDurableBackend(); + const validTranscript = join(tempDir, "usable.jsonl"); + writeSessionTranscript(validTranscript, "usable-session"); + registerCompleted(backend, "partially-retained"); + backend.recordCheckpoint({ + kind: "stage", workflowId: "partially-retained", checkpointId: "stage:1", name: "retained", + replayKey: "stage:retained:1", sessionFile: validTranscript, completedAt: 20, + }); + backend.recordCheckpoint({ + kind: "stage", workflowId: "partially-retained", checkpointId: "stage:2", name: "stale", + replayKey: "stage:stale:1", sessionFile: join(tempDir, "missing.jsonl"), completedAt: 21, + }); + + const snapshot = completedWorkflowSnapshot(backend, listCompletedFromBackend(backend)[0]!); + assert.deepEqual(listOpenableCompletedWorkflows(backend).map((item) => item.workflowId), ["partially-retained"]); + assert.equal(snapshot?.stages[0]?.sessionFile, validTranscript); + assert.equal(snapshot?.stages[1]?.sessionFile, undefined); + }); + + test("rejects partially malformed and context-empty transcripts", () => { + const backend = new InMemoryDurableBackend(); + const malformed = join(tempDir, "partially-malformed.jsonl"); + writeFileSync(malformed, [ + JSON.stringify({ type: "session", id: "partially-malformed" }), + JSON.stringify({ type: "message", id: "valid", timestamp: new Date().toISOString(), message: { role: "user", content: "context" } }), + "not-json", + ].join("\n")); + const emptyContent = [ + { id: "blank-string", content: " " }, + { id: "empty-array", content: [] }, + { id: "empty-object", content: {} }, + { id: "empty-block", content: [{}] }, + { id: "blank-text-block", content: [{ type: "text", text: "" }] }, + ] as const; + registerCompleted(backend, "partially-malformed"); + backend.recordCheckpoint({ + kind: "stage", workflowId: "partially-malformed", checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", sessionFile: malformed, completedAt: 20, + }); + for (const item of emptyContent) { + const path = join(tempDir, `${item.id}.jsonl`); + writeFileSync(path, [ + JSON.stringify({ type: "session", id: item.id }), + JSON.stringify({ type: "message", id: `${item.id}-message`, timestamp: new Date().toISOString(), message: { role: "user", content: item.content } }), + ].join("\n")); + registerCompleted(backend, item.id); + backend.recordCheckpoint({ + kind: "stage", workflowId: item.id, checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", sessionFile: path, completedAt: 20, + }); + } + + assert.deepEqual(listOpenableCompletedWorkflows(backend), []); + }); + + test("accepts a retained transcript with a meaningful structured content block", () => { + const backend = new InMemoryDurableBackend(); + const path = join(tempDir, "structured-context.jsonl"); + writeFileSync(path, [ + JSON.stringify({ type: "session", id: "structured-context" }), + JSON.stringify({ + type: "message", + id: "structured-context-message", + timestamp: new Date().toISOString(), + message: { role: "user", content: [{ type: "text", text: "retained context" }] }, + }), + ].join("\n")); + registerCompleted(backend, "structured-context"); + backend.recordCheckpoint({ + kind: "stage", workflowId: "structured-context", checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", sessionFile: path, completedAt: 20, + }); + + assert.deepEqual(listOpenableCompletedWorkflows(backend).map((item) => item.workflowId), ["structured-context"]); + }); +}); diff --git a/test/unit/durable-resume-runtime.test.ts b/test/unit/durable-resume-runtime.test.ts index 309bb4102..3b0eb0355 100644 --- a/test/unit/durable-resume-runtime.test.ts +++ b/test/unit/durable-resume-runtime.test.ts @@ -295,6 +295,7 @@ describe("resumeDurableWorkflow", () => { getWorkflow = this.mem.getWorkflow.bind(this.mem); setWorkflowStatus = this.mem.setWorkflowStatus.bind(this.mem); listResumableWorkflows = this.mem.listResumableWorkflows.bind(this.mem); + listCompletedWorkflows = this.mem.listCompletedWorkflows.bind(this.mem); toCacheEntry = this.mem.toCacheEntry.bind(this.mem) as (workflowId: string) => DurableCheckpointEntry | undefined; reset = this.mem.reset.bind(this.mem); async hydrateResumableWorkflows(): Promise { diff --git a/test/unit/workflow-completed-inspection.test.ts b/test/unit/workflow-completed-inspection.test.ts new file mode 100644 index 000000000..14bc67e34 --- /dev/null +++ b/test/unit/workflow-completed-inspection.test.ts @@ -0,0 +1,218 @@ +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 { SessionManager } from "../../packages/coding-agent/src/core/session-manager.js"; +import { InMemoryDurableBackend } from "../../packages/workflows/src/durable/backend.js"; +import { openCompletedDurableWorkflow } from "../../packages/workflows/src/durable/completed-inspection.js"; +import { createStageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import { createStore } from "../../packages/workflows/src/shared/store.js"; +import { mockSession, type StageSessionRuntime } from "./executor-shared.js"; + +let tempDir = ""; + +beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), "atomic-completed-inspection-")); }); +afterEach(() => { rmSync(tempDir, { recursive: true, force: true }); }); + +function retainedSession(name: string, internal = false): string { + const path = join(tempDir, `${name}.jsonl`); + writeFileSync(path, [ + JSON.stringify({ + type: "session", + version: 3, + id: `${name}-session`, + timestamp: new Date().toISOString(), + cwd: tempDir, + ...(internal ? { internal: true, workflow: { runId: name, stageId: "final", stageName: "final" } } : {}), + }), + JSON.stringify({ type: "message", id: `${name}-message`, parentId: null, timestamp: new Date().toISOString(), message: { role: "user", content: "Original workflow request", timestamp: Date.now() } }), + ].join("\n") + "\n"); + return path; +} + +describe("completed workflow inspection", () => { + test("opens immutable detail and appends follow-up chat without durable re-dispatch", async () => { + const backend = new InMemoryDurableBackend(); + const store = createStore(); + const registry = createStageControlRegistry(); + const sessionFile = retainedSession("completed-inspection"); + const promptCalls: string[] = []; + const session: StageSessionRuntime = { + ...mockSession(), + sessionFile, + async prompt(text: string) { promptCalls.push(text); }, + }; + backend.registerWorkflow({ + workflowId: "completed-inspection", + name: "completed-flow", + inputs: { topic: "done" }, + createdAt: 1, + updatedAt: 3, + status: "completed", + }); + backend.recordCheckpoint({ + kind: "stage", + workflowId: "completed-inspection", + checkpointId: "stage:1", + name: "final", + replayKey: "stage:final:1", + output: "done", + sessionFile, + completedAt: 2, + }); + + let sessionCreates = 0; + let restoredMessageCount = 0; + const opened = openCompletedDurableWorkflow("completed-ins", { + durableBackend: backend, + store, + stageControlRegistry: registry, + adapters: { + agentSession: { + async create(options) { + restoredMessageCount = options.sessionManager?.getEntries().length ?? 0; + sessionCreates += 1; + return session; + }, + }, + }, + cwd: tempDir, + }); + + assert.equal(opened.ok, true); + assert.equal(store.runs()[0]?.status, "completed"); + assert.equal(backend.getWorkflow("completed-inspection")?.status, "completed"); + const handle = registry.get("completed-inspection", "completed-stage-1"); + assert.ok(handle); + assert.deepEqual(registry.run("completed-inspection").stages(), []); + await handle.prompt("What should I do next?"); + assert.equal(sessionCreates, 1); + assert.equal(restoredMessageCount, 1); + assert.deepEqual(promptCalls, ["What should I do next?"]); + assert.equal(store.runs()[0]?.status, "completed"); + assert.equal(backend.getWorkflow("completed-inspection")?.status, "completed"); + }); + + test("refuses to replace an active run with the same id", () => { + const backend = new InMemoryDurableBackend(); + const store = createStore(); + const sessionFile = retainedSession("same-id"); + backend.registerWorkflow({ workflowId: "same-id", name: "completed-flow", inputs: {}, createdAt: 1, status: "completed" }); + backend.recordCheckpoint({ kind: "stage", workflowId: "same-id", checkpointId: "stage:1", name: "final", replayKey: "stage:final:1", sessionFile, completedAt: 2 }); + store.recordRunStart({ id: "same-id", name: "active", inputs: {}, status: "running", stages: [], startedAt: 1 }); + + const opened = openCompletedDurableWorkflow("same-id", { durableBackend: backend, store }); + assert.equal(opened.ok, false); + if (!opened.ok) assert.equal(opened.reason, "active"); + assert.equal(store.runs()[0]?.status, "running"); + }); + + test("replaces a retained completed snapshot with authoritative durable detail", () => { + const backend = new InMemoryDurableBackend(); + const store = createStore(); + const sessionFile = retainedSession("authoritative"); + backend.registerWorkflow({ workflowId: "authoritative", name: "durable-name", inputs: {}, createdAt: 1, status: "completed" }); + backend.recordCheckpoint({ + kind: "stage", workflowId: "authoritative", checkpointId: "stage:1", name: "durable-stage", + replayKey: "stage:durable:1", output: "durable result", sessionFile, completedAt: 2, + }); + store.recordRunStart({ + id: "authoritative", name: "stale-local-name", inputs: {}, status: "completed", + stages: [], startedAt: 1, endedAt: 2, resumable: false, + }); + + const opened = openCompletedDurableWorkflow("authoritative", { durableBackend: backend, store }); + + assert.equal(opened.ok, true); + assert.equal(store.runs()[0]?.name, "durable-name"); + assert.equal(store.runs()[0]?.stages[0]?.name, "durable-stage"); + assert.equal(store.runs()[0]?.stages[0]?.sessionFile, sessionFile); + }); + + test("refreshes a retained chat handle when authoritative transcript detail changes", () => { + const backend = new InMemoryDurableBackend(); + const store = createStore(); + const registry = createStageControlRegistry(); + const firstSessionFile = retainedSession("first-authoritative"); + const secondSessionFile = retainedSession("second-authoritative"); + backend.registerWorkflow({ + workflowId: "refresh-chat", name: "completed-flow", inputs: {}, createdAt: 1, status: "completed", + }); + backend.recordCheckpoint({ + kind: "stage", workflowId: "refresh-chat", checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", sessionFile: firstSessionFile, completedAt: 2, + }); + const deps = { + durableBackend: backend, + store, + stageControlRegistry: registry, + adapters: { agentSession: { async create() { return mockSession(); } } }, + }; + + assert.equal(openCompletedDurableWorkflow("refresh-chat", deps).ok, true); + const firstHandle = registry.get("refresh-chat", "completed-stage-1"); + assert.equal(firstHandle?.sessionFile, firstSessionFile); + backend.recordCheckpoint({ + kind: "stage", workflowId: "refresh-chat", checkpointId: "stage:2", name: "final", + replayKey: "stage:final:1", sessionFile: secondSessionFile, completedAt: 3, + }); + + assert.equal(openCompletedDurableWorkflow("refresh-chat", deps).ok, true); + assert.equal(firstHandle?.isDisposed, true); + assert.equal(registry.get("refresh-chat", "completed-stage-1")?.sessionFile, secondSessionFile); + }); + + test("removes a retained chat handle when its transcript becomes invalid", () => { + const backend = new InMemoryDurableBackend(); + const store = createStore(); + const registry = createStageControlRegistry(); + const invalidatedSessionFile = retainedSession("invalidated-stage"); + const retainedSessionFile = retainedSession("still-retained-stage"); + backend.registerWorkflow({ + workflowId: "invalidate-chat", name: "completed-flow", inputs: {}, createdAt: 1, status: "completed", + }); + backend.recordCheckpoint({ + kind: "stage", workflowId: "invalidate-chat", checkpointId: "stage:1", name: "first", + replayKey: "stage:first:1", sessionFile: invalidatedSessionFile, completedAt: 2, + }); + backend.recordCheckpoint({ + kind: "stage", workflowId: "invalidate-chat", checkpointId: "stage:2", name: "second", + replayKey: "stage:second:1", sessionFile: retainedSessionFile, completedAt: 3, + }); + const deps = { + durableBackend: backend, + store, + stageControlRegistry: registry, + adapters: { agentSession: { async create() { return mockSession(); } } }, + }; + + assert.equal(openCompletedDurableWorkflow("invalidate-chat", deps).ok, true); + const invalidatedHandle = registry.get("invalidate-chat", "completed-stage-1"); + assert.ok(invalidatedHandle); + rmSync(invalidatedSessionFile); + + assert.equal(openCompletedDurableWorkflow("invalidate-chat", deps).ok, true); + assert.equal(invalidatedHandle.isDisposed, true); + assert.equal(registry.get("invalidate-chat", "completed-stage-1"), undefined); + assert.equal(registry.get("invalidate-chat", "completed-stage-2")?.sessionFile, retainedSessionFile); + }); + + test("opens a retained internal stage transcript without exposing it in ordinary history", async () => { + const backend = new InMemoryDurableBackend(); + const store = createStore(); + const internalSessionFile = retainedSession("internal-completed", true); + retainedSession("regular-history"); + backend.registerWorkflow({ + workflowId: "internal-completed", name: "completed-flow", inputs: {}, createdAt: 1, status: "completed", + }); + backend.recordCheckpoint({ + kind: "stage", workflowId: "internal-completed", checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", sessionFile: internalSessionFile, completedAt: 2, + }); + + assert.equal(openCompletedDurableWorkflow("internal-completed", { durableBackend: backend, store }).ok, true); + assert.equal(store.runs()[0]?.stages[0]?.sessionFile, internalSessionFile); + assert.deepEqual((await SessionManager.list(tempDir, tempDir)).map((session) => session.id), ["regular-history-session"]); + }); +}); diff --git a/test/unit/workflow-resume-selector.test.ts b/test/unit/workflow-resume-selector.test.ts new file mode 100644 index 000000000..a1382805b --- /dev/null +++ b/test/unit/workflow-resume-selector.test.ts @@ -0,0 +1,114 @@ +import { describe, test } from "bun:test"; +import assert from "node:assert/strict"; +import { + openWorkflowResumeSelector, + workflowResumeSelectorItems, +} from "../../packages/workflows/src/tui/workflow-resume-selector.js"; +import type { ResumableWorkflowEntry } from "../../packages/workflows/src/durable/types.js"; +import type { RunSnapshot, StageSnapshot } from "../../packages/workflows/src/shared/store-types.js"; + +function entry( + id: string, + status: ResumableWorkflowEntry["status"], + updatedAt = status === "completed" ? 300 : 200, +): ResumableWorkflowEntry { + return { + workflowId: id, + name: `${status}-workflow`, + status, + completedCheckpoints: 2, + pendingPrompts: 0, + createdAt: 1, + updatedAt, + }; +} + +function stage(id: string, endedAt: number): StageSnapshot { + return { + id, + name: id, + status: "completed", + parentIds: [], + startedAt: endedAt - 1, + endedAt, + toolEvents: [], + }; +} + +function pausedLiveRun(id = "live-paused", activityAt = 100): RunSnapshot { + return { + id, + name: "live-workflow", + inputs: {}, + status: "paused", + stages: [], + startedAt: 1, + pausedAt: activityAt, + resumable: true, + }; +} + +describe("workflow resume selector", () => { + test("globally orders mixed rows and renders completed rows with a green semantic", () => { + const items = workflowResumeSelectorItems( + [pausedLiveRun()], + [entry("durable-paused", "paused")], + [entry("durable-completed", "completed")], + ); + + assert.deepEqual(items.map((item) => item.result.kind), ["completed", "durable", "live"]); + const completed = items[0]!; + assert.match(completed.session.firstMessage, /✓ completed/); + assert.equal(completed.session.messageColor, "success"); + assert.equal(completed.session.path, "workflow-completed:durable-completed"); + }); + + test("uses latest stage activity and deterministic ids for equal-time ties", () => { + const live = pausedLiveRun("zulu-live", 50); + live.stages.push(stage("recent", 500)); + const reversed = workflowResumeSelectorItems( + [live, pausedLiveRun("alpha-live", 400)], + [entry("zulu-durable", "paused", 400), entry("alpha-durable", "paused", 400)], + [entry("middle-completed", "completed", 450)], + ); + + assert.deepEqual(reversed.map((item) => item.session.id), [ + "zulu-live", + "middle-completed", + "alpha-durable", + "alpha-live", + "zulu-durable", + ]); + assert.deepEqual( + workflowResumeSelectorItems( + [pausedLiveRun("alpha-live", 400), live], + [entry("alpha-durable", "paused", 400), entry("zulu-durable", "paused", 400)], + [entry("middle-completed", "completed", 450)], + ).map((item) => item.session.id), + reversed.map((item) => item.session.id), + ); + }); + + test("deduplicates before sorting and keeps live then durable precedence", () => { + const items = workflowResumeSelectorItems( + [pausedLiveRun()], + [entry("same-id", "paused", 500)], + [entry("same-id", "completed", 900), entry("live-paused", "completed", 1_000)], + ); + + assert.deepEqual(items.map((item) => item.session.id), ["same-id", "live-paused"]); + assert.deepEqual(items.map((item) => item.result.kind), ["durable", "live"]); + }); + + test("closes when the custom selector mount throws or rejects", async () => { + const thrown = await openWorkflowResumeSelector({ + custom: () => { throw new Error("mount failed"); }, + }, [pausedLiveRun()], []); + const rejected = await openWorkflowResumeSelector({ + custom: async () => { throw new Error("async mount failed"); }, + }, [pausedLiveRun()], []); + + assert.deepEqual(thrown, { kind: "close" }); + assert.deepEqual(rejected, { kind: "close" }); + }); +}); diff --git a/test/unit/workflow-run-control-completed-resume.test.ts b/test/unit/workflow-run-control-completed-resume.test.ts new file mode 100644 index 000000000..e7f7cc533 --- /dev/null +++ b/test/unit/workflow-run-control-completed-resume.test.ts @@ -0,0 +1,260 @@ +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 { InMemoryDurableBackend } from "../../packages/workflows/src/durable/backend.js"; +import { setDurableBackend } from "../../packages/workflows/src/durable/factory.js"; +import { createExtensionRuntime, type ExtensionRuntime } from "../../packages/workflows/src/extension/runtime.js"; +import { handleRunControlCommand, type WorkflowRunControlDeps } from "../../packages/workflows/src/extension/workflow-run-control-command.js"; +import { store } from "../../packages/workflows/src/shared/store.js"; + +let tempDir = ""; + +beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), "atomic-completed-command-")); }); +afterEach(() => { + setDurableBackend(undefined); + store.clear(); + 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: `Prior context for ${name}`, timestamp: Date.now() } }), + ].join("\n") + "\n"); + return path; +} + +function registerCompleted(backend: InMemoryDurableBackend, id: string, sessionFile = retainedSession(id)): void { + backend.registerWorkflow({ workflowId: id, name: `${id}-flow`, inputs: {}, createdAt: 1, status: "completed" }); + backend.recordCheckpoint({ + kind: "stage", workflowId: id, checkpointId: "stage:1", name: "final", + replayKey: "stage:final:1", output: "ok", sessionFile, completedAt: 2, + }); +} + +function commandDeps(runtime: ExtensionRuntime, opened: string[]): WorkflowRunControlDeps { + return { + pi: {}, + overlay: { open: (runId) => { if (runId) opened.push(runId); }, toggle: () => undefined, close: () => undefined }, + getPersistence: () => undefined, + runtimeForContext: () => runtime, + ensureWorkflowResourcesLoaded: () => undefined, + }; +} + +async function resume(target: string, runtime: ExtensionRuntime, opened: string[] = []): Promise<{ messages: string[]; errors: string[] }> { + const messages: string[] = []; + const errors: string[] = []; + await handleRunControlCommand( + "resume", + [target], + { hasUI: true, ui: { notify: () => undefined } }, + { info: (message) => messages.push(message), error: (message) => errors.push(message) }, + commandDeps(runtime, opened), + ); + return { messages, errors }; +} + +describe("/workflow resume completed target", () => { + test("opens a unique completed id prefix without invoking durable resume dispatch", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + registerCompleted(backend, "completed-command-target"); + const baseRuntime = createExtensionRuntime({ store }); + let resumeCalls = 0; + const runtime: ExtensionRuntime = { + ...baseRuntime, + resumeDurableWorkflow(workflowIdOrPrefix, options) { + resumeCalls += 1; + return baseRuntime.resumeDurableWorkflow(workflowIdOrPrefix, options); + }, + }; + const opened: string[] = []; + + const result = await resume("completed-command", runtime, opened); + + assert.equal(resumeCalls, 0); + assert.deepEqual(opened, ["completed-command-target"]); + assert.match(result.messages.join("\n"), /read-only inspection and follow-up chat/); + assert.equal(store.runs()[0]?.status, "completed"); + assert.equal(backend.getWorkflow("completed-command-target")?.status, "completed"); + }); + + test("opens an exact completed id and reports completed-prefix ambiguity", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + registerCompleted(backend, "completed-exact-alpha"); + registerCompleted(backend, "completed-exact-beta"); + const runtime = createExtensionRuntime({ store }); + const opened: string[] = []; + + const exact = await resume("completed-exact-alpha", runtime, opened); + store.clear(); + const ambiguous = await resume("completed-exact-", runtime); + + assert.deepEqual(opened, ["completed-exact-alpha"]); + assert.match(exact.messages.join("\n"), /Opened completed durable workflow/); + assert.match(ambiguous.errors.join("\n"), /Ambiguous workflow prefix/); + assert.match(ambiguous.errors.join("\n"), /completed-exact-alpha-flow/); + assert.match(ambiguous.errors.join("\n"), /completed-exact-beta-flow/); + }); + + test("reports a clear missing target without dispatching completed inspection", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + registerCompleted(backend, "known-completed"); + + const result = await resume("missing-workflow", createExtensionRuntime({ store })); + + assert.match(result.errors.join("\n"), /No durable workflow found for id\/prefix: missing-workflow/); + }); + + test("reports a stale completed target instead of dispatching it", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + backend.registerWorkflow({ workflowId: "stale-completed-target", name: "completed-flow", inputs: {}, createdAt: 1, status: "completed", completedCheckpoints: 1 }); + const baseRuntime = createExtensionRuntime({ store }); + let resumeCalls = 0; + const runtime: ExtensionRuntime = { + ...baseRuntime, + resumeDurableWorkflow(workflowIdOrPrefix, options) { + resumeCalls += 1; + return baseRuntime.resumeDurableWorkflow(workflowIdOrPrefix, options); + }, + }; + + const result = await resume("stale-completed", runtime); + + assert.equal(resumeCalls, 0); + assert.match(result.errors.join("\n"), /stale or missing durable checkpoint\/session data/); + }); + + test("does not let a retained completed snapshot bypass authoritative stale checks", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + backend.registerWorkflow({ workflowId: "retained-stale", name: "completed-flow", inputs: {}, createdAt: 1, status: "completed", completedCheckpoints: 1 }); + store.recordRunStart({ id: "retained-stale", name: "completed-flow", inputs: {}, status: "completed", stages: [], startedAt: 1, endedAt: 2, resumable: false }); + const baseRuntime = createExtensionRuntime({ store }); + let resumeCalls = 0; + const runtime: ExtensionRuntime = { ...baseRuntime, resumeDurableWorkflow(target, options) { resumeCalls += 1; return baseRuntime.resumeDurableWorkflow(target, options); } }; + const opened: string[] = []; + + const result = await resume("retained-stale", runtime, opened); + + assert.equal(resumeCalls, 0); + assert.deepEqual(opened, []); + assert.match(result.errors.join("\n"), /stale or missing durable checkpoint\/session data/); + }); + + test("reports ambiguity across live and completed workflow prefixes", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + registerCompleted(backend, "shared-completed"); + store.recordRunStart({ id: "shared-live", name: "live-flow", inputs: {}, status: "paused", stages: [], startedAt: 1, resumable: true }); + const result = await resume("shared-", createExtensionRuntime({ store })); + + assert.match(result.errors.join("\n"), /Ambiguous workflow prefix/); + assert.match(result.errors.join("\n"), /live-flow/); + assert.match(result.errors.join("\n"), /shared-completed-flow/); + }); + + test("excludes cancelled, killed, and non-resumable failed locals from prefix resolution", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + registerCompleted(backend, "excluded-completed"); + store.recordRunStart({ id: "excluded-cancelled", name: "cancelled", inputs: {}, status: "cancelled", stages: [], startedAt: 1, endedAt: 2, resumable: false }); + store.recordRunStart({ id: "excluded-killed", name: "killed", inputs: {}, status: "killed", stages: [], startedAt: 1, endedAt: 2, resumable: false }); + store.recordRunStart({ id: "excluded-failed", name: "failed", inputs: {}, status: "failed", stages: [], startedAt: 1, endedAt: 2, resumable: false }); + const opened: string[] = []; + + const result = await resume("excluded-", createExtensionRuntime({ store }), opened); + + assert.equal(result.errors.length, 0); + assert.deepEqual(opened, ["excluded-completed"]); + }); + + test("keeps quit shadows on the durable resume path", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + backend.registerWorkflow({ workflowId: "quit-shadow", name: "quit-flow", inputs: {}, createdAt: 1, status: "paused", completedCheckpoints: 1 }); + store.recordRunStart({ id: "quit-shadow", name: "quit-flow", inputs: {}, status: "running", stages: [], startedAt: 1, endedAt: 2, exitReason: "quit", resumable: true }); + const baseRuntime = createExtensionRuntime({ store }); + let durableResumeCalls = 0; + const runtime: ExtensionRuntime = { + ...baseRuntime, + resumeDurableWorkflow() { + durableResumeCalls += 1; + return { ok: true, runId: "quit-shadow", workflowId: "quit-shadow", name: "quit-flow", message: "resumed quit shadow" }; + }, + }; + + const result = await resume("quit-shadow", runtime); + + assert.equal(durableResumeCalls, 1); + assert.match(result.messages.join("\n"), /resumed quit shadow/); + }); + + for (const status of ["running", "failed", "blocked"] as const) { + test(`keeps durable ${status} targets on the durable resume path`, async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + const id = `durable-${status}`; + const entry = { workflowId: id, name: `${status}-flow`, status, completedCheckpoints: 1, pendingPrompts: 0, createdAt: 1, updatedAt: 2, resumable: true }; + let resumeCalls = 0; + const runtime = { + registry: { has: () => true }, + prepareDurableResumable: async () => [entry], + prepareCompletedDurable: async () => [], + resumeDurableWorkflow: () => { + resumeCalls += 1; + return { ok: true as const, runId: id, workflowId: id, name: entry.name, message: `resumed ${status}` }; + }, + } as unknown as ExtensionRuntime; + + const result = await resume(id, runtime); + + assert.equal(resumeCalls, 1); + assert.match(result.messages.join("\n"), new RegExp(`resumed ${status}`)); + }); + } + + test("keeps exact full live ids on the existing paused resume path", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + registerCompleted(backend, "exact-live-other-completed"); + store.recordRunStart({ id: "exact-live", name: "live-flow", inputs: {}, status: "paused", stages: [], startedAt: 1, resumable: true }); + const opened: string[] = []; + + const result = await resume("exact-live", createExtensionRuntime({ store }), opened); + + 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/); + }); + + test("keeps recoverable failed and active-running explicit behavior unchanged", async () => { + const backend = new InMemoryDurableBackend(); + setDurableBackend(backend); + store.recordRunStart({ id: "failed-live", name: "failed-flow", inputs: {}, status: "failed", stages: [], startedAt: 1, endedAt: 2, resumable: true }); + store.recordRunStart({ id: "running-live", name: "running-flow", inputs: {}, status: "running", stages: [], startedAt: 1 }); + const baseRuntime = createExtensionRuntime({ store }); + let failedResumeCalls = 0; + const runtime: ExtensionRuntime = { + ...baseRuntime, + resumeFailedRun() { + failedResumeCalls += 1; + return { ok: true, runId: "continued-run", sourceRunId: "failed-live", resumeFromStageId: "failed-stage", message: "continued failed workflow" }; + }, + }; + + const failedResult = await resume("failed-live", runtime); + const runningResult = await resume("running-live", runtime); + + assert.equal(failedResumeCalls, 1); + assert.match(failedResult.messages.join("\n"), /continued failed workflow/); + assert.match(runningResult.errors.join("\n"), /already running.*connect/i); + }); +});