diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index a0eeabdce..8053f47c3 100644 --- a/packages/coding-agent/docs/workflows.md +++ b/packages/coding-agent/docs/workflows.md @@ -410,7 +410,7 @@ Named runs go to the background. Common controls: When a paused stage is resumed with a message, Atomic lets the stage answer that resume message, then (if the stage has not already finalized) injects `Continue where you left off. If you believe you are finished with your original task (or a redefined task if the user told you), stop.` into the same stage session before normal stage completion/readiness handling. This keeps interrupted work moving without asking you to manually type a second continuation prompt while also preventing stages that already finished their scoped work from overstepping. -Durable `/workflow resume` preserves completed stage metadata and graph topology. When it reopens an unfinished stage session mid-chat, it uses the same continuation prompt shown above. Replayed `ctx.stage`, `ctx.task`, `ctx.chain`, `ctx.parallel`, and child-workflow checkpoints keep their original summaries, timing, session/model metadata, and parallel fanout parentage in status and graph views instead of appearing as freshly flattened replay nodes. +Durable `/workflow resume` preserves completed stage metadata, active-stage elapsed time, and graph topology. While an LM stage or task is active, repeated durable checkpoints refresh its accumulated pause-adjusted duration even when its session file does not change. Each new Atomic process that reopens the unfinished session mid-chat starts from the latest saved baseline and uses the same continuation prompt shown above, so repeated process-boundary resumes keep status, graph, stored, and lifecycle duration cumulative without double-counting pauses from earlier process segments. Replayed `ctx.stage`, `ctx.task`, `ctx.chain`, `ctx.parallel`, and child-workflow checkpoints keep their original summaries, timing, session/model metadata, and parallel fanout parentage instead of appearing as freshly flattened replay nodes. Workflow stage sessions are marked as **internal** and excluded from the standard `/resume`, `atomic -r`, and `--continue` history so they do not clutter your interactive session picker. They remain resumable and inspectable through the workflow-specific commands and tool actions shown here (`/workflow resume`, `/workflow attach`, `workflow({ action: "status" | "stages" | "stage" | "resume" })`), which read the run/stage store and its `sessionFile` links directly. Passing a stage session's file path to `--session` still opens it explicitly. Legacy workflow sessions created before this marker behavior lack the signal and will continue to appear in the standard history until they age out. diff --git a/packages/workflows/CHANGELOG.md b/packages/workflows/CHANGELOG.md index 2ba54004e..ffa97eff9 100644 --- a/packages/workflows/CHANGELOG.md +++ b/packages/workflows/CHANGELOG.md @@ -21,6 +21,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - 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)) - Fixed `/workflow resume` to deduplicate live and durable choices with live-row precedence, then globally sort the combined selector by latest run/stage or durable-update activity, newest first, with deterministic workflow-id tie-breaking independent of source enumeration order ([#1714](https://github.com/bastani-inc/atomic/issues/1714)). - Fixed startup and restored-session workflow notices to use the authoritative durable resume catalog's status/progress rules: failed and blocked root workflows are advertised unless explicitly marked `resumable: false`, running and paused workflows require durable progress, nested child workflows and stale cache-only entries are excluded, and legacy direct or payload-wrapped checkpoint entries are parsed correctly. +- Fixed durable mid-stage workflow resume to refresh pause-adjusted active stage/task duration at repeated session checkpoints, preserve it through file and DBOS hydration, and continue accumulating it across repeated process-boundary resumes without resetting the baseline when concurrent tracked stage calls begin, double-counting earlier pauses, changing replay identity, or re-running completed side effects ([#1713](https://github.com/bastani-inc/atomic/issues/1713)). ## [0.9.8] - 2026-07-12 diff --git a/packages/workflows/README.md b/packages/workflows/README.md index a1585b7ea..dd14d7b24 100644 --- a/packages/workflows/README.md +++ b/packages/workflows/README.md @@ -547,7 +547,7 @@ Input overrides are bare `key=value` tokens (no leading `--`). Values are JSON-p Named workflow launches always run as **background tasks** in interactive sessions. Model-launched direct `task`, `tasks`, and `chain` calls must set top-level `async: true` so the chat editor stays free; inspection and control calls are unaffected. Foreground launches are reserved for explicit user requests or technical requirements, with notice before launch. Press **F2** (or `/workflow connect `) to attach to the live graph viewer; HIL prompts (`ctx.ui.input/confirm/select/editor/custom`) appear as awaiting-input graph nodes. Press Enter on a focused node, or click a visible graph node directly, to open that stage and answer locally, never as a modal dialog over the chat. While the graph pane is active, vertical wheel/trackpad gestures pan vertically and horizontal gestures pan wide graphs left and right when the terminal reports them, without falling through to the main chat or terminal scrollback. Attached stage chats capture mouse/trackpad wheel events by default so scrolling stays inside the active stage transcript or prompt instead of falling through to terminal/main-chat scrollback. Press `ctrl+t` to toggle **copy mode**: copy mode disables workflow-chat mouse reporting so normal terminal/tmux text selection can work; press `ctrl+t` again to leave copy mode and restore workflow-chat scrolling. Archived read-only stage transcripts show the same copy-mode footer/status, allowing their transcript text to be selected and copied while preserving `esc` close and `ctrl+d` graph navigation. While copy mode is on, wheel/trackpad gestures are handled by the terminal/tmux and may scroll terminal scrollback, so leave copy mode before using the wheel again. Human input is detected when those runtime `ctx.ui.*` calls execute; workflows no longer have a declaration-time HIL flag. -Durable resume preserves the original completed-stage shape: replayed stage/task/parallel/child checkpoints retain summaries, durations, session/model metadata, and parallel branch parentage in the graph instead of flattening fanout branches into a sequential replay chain. +Durable resume preserves both completed and active stage timing. Replayed stage/task/parallel/child checkpoints retain summaries, durations, session/model metadata, and parallel branch parentage in the graph instead of flattening fanout branches into a sequential replay chain. While an LM stage or task is active, repeated durable session checkpoints refresh its pause-adjusted accumulated duration even when the session file is unchanged. Every later `/workflow resume` starts from the latest saved baseline, so repeated process-boundary resumes keep status, graph, stored, and lifecycle durations cumulative while subtracting only pauses from the current process segment. Nested `ctx.workflow(...)` calls are displayed as an expanded graph within the top-level run. `/workflow status` and run pickers list only top-level user-launched workflows, not implementation-owned child runs. The `workflow` tool's `stages`, `stage`, `transcript`, `send`, `pause`, `interrupt`, and `resume` actions can still target visible child stage ids, prefixes, or names from the expanded graph; Atomic routes the control action to the owning nested run internally. (`stages`, `stage`, `transcript`, and `send` are `workflow` tool actions, not `/workflow` slash subcommands; the slash command exposes `connect`, `attach`, `pause`, `list`, `status`, `interrupt`, `kill`, `resume`, `reload`, and `inputs`.) diff --git a/packages/workflows/src/durable/backend.ts b/packages/workflows/src/durable/backend.ts index 1fa2f7447..18ee7cb3a 100644 --- a/packages/workflows/src/durable/backend.ts +++ b/packages/workflows/src/durable/backend.ts @@ -95,8 +95,13 @@ export interface DurableWorkflowBackend { /** Look up a cached stage output by replay key. */ getStageOutput(workflowId: string, replayKey: string): WorkflowSerializableValue | undefined; - /** Look up resumable stage session metadata by replay key, when available. */ - getStageSession(workflowId: string, replayKey: string): { sessionId?: string; sessionFile?: string } | undefined; + /** Look up the latest resumable stage session and accumulated active timing. */ + getStageSession(workflowId: string, replayKey: string): { + sessionId?: string; + sessionFile?: string; + startedAt?: number; + durationMs?: number; + } | undefined; /** List all checkpoints for a workflow (in completion order). */ listCheckpoints(workflowId: string): readonly DurableCheckpoint[]; @@ -249,10 +254,20 @@ export class InMemoryDurableBackend implements DurableWorkflowBackend { return checkpoint !== undefined && "output" in checkpoint ? checkpoint.output : undefined; } - getStageSession(workflowId: string, replayKey: string): { sessionId?: string; sessionFile?: string } | undefined { + getStageSession(workflowId: string, replayKey: string): { + sessionId?: string; + sessionFile?: string; + startedAt?: number; + durationMs?: number; + } | undefined { const checkpoint = this.workflows.get(workflowId)?.stageSessionByReplayKey.get(replayKey); if (checkpoint?.sessionId === undefined && checkpoint?.sessionFile === undefined) return undefined; - return { ...(checkpoint.sessionId !== undefined ? { sessionId: checkpoint.sessionId } : {}), ...(checkpoint.sessionFile !== undefined ? { sessionFile: checkpoint.sessionFile } : {}) }; + return { + ...(checkpoint.sessionId !== undefined ? { sessionId: checkpoint.sessionId } : {}), + ...(checkpoint.sessionFile !== undefined ? { sessionFile: checkpoint.sessionFile } : {}), + ...(checkpoint.startedAt !== undefined ? { startedAt: checkpoint.startedAt } : {}), + ...(checkpoint.durationMs !== undefined ? { durationMs: checkpoint.durationMs } : {}), + }; } listCheckpoints(workflowId: string): readonly DurableCheckpoint[] { diff --git a/packages/workflows/src/durable/scoped-backend.ts b/packages/workflows/src/durable/scoped-backend.ts index 5df7c2e6a..44e8d89b6 100644 --- a/packages/workflows/src/durable/scoped-backend.ts +++ b/packages/workflows/src/durable/scoped-backend.ts @@ -86,7 +86,12 @@ export class ScopedDurableBackend implements DurableWorkflowBackend { return this.inner.getStageOutput(this.scope.rootWorkflowId, this.scopeKey(replayKey)); } - getStageSession(_workflowId: string, replayKey: string): { sessionId?: string; sessionFile?: string } | undefined { + getStageSession(_workflowId: string, replayKey: string): { + sessionId?: string; + sessionFile?: string; + startedAt?: number; + durationMs?: number; + } | undefined { return this.inner.getStageSession(this.scope.rootWorkflowId, this.scopeKey(replayKey)); } diff --git a/packages/workflows/src/durable/stage-primitive.ts b/packages/workflows/src/durable/stage-primitive.ts index 1d20b1d59..d22fb2abe 100644 --- a/packages/workflows/src/durable/stage-primitive.ts +++ b/packages/workflows/src/durable/stage-primitive.ts @@ -7,6 +7,7 @@ import type { DurableWorkflowBackend } from "./backend.js"; import type { ParallelFailFastScope } from "../runs/foreground/executor-types.js"; import { durableHash } from "./backend.js"; import { recordCheckpointDurably } from "./tool-primitive.js"; +import { elapsedStageMs } from "../shared/timing.js"; import { RESUME_CONTINUATION_PROMPT } from "../shared/resume-continuation.js"; import type { DurableStageCheckpoint } from "./types.js"; export type DurableCompletedStageCheckpoint = DurableStageCheckpoint & { readonly output: WorkflowSerializableValue }; @@ -17,6 +18,7 @@ export interface DurableStageDeps { readonly nextCheckpointId: () => string; readonly nextReplayKey: (stageName: string) => string; readonly replayKeyForCompletedStage?: (stage: StageSnapshot) => string | undefined; + readonly now?: () => number; } export async function recordStageCheckpoint(deps: DurableStageDeps, stage: StageSnapshot): Promise { @@ -51,17 +53,25 @@ export async function recordStageCheckpoint(deps: DurableStageDeps, stage: Stage export async function recordStageSessionCheckpoint(deps: DurableStageDeps, stage: StageSnapshot): Promise { const replayKey = deps.replayKeyForCompletedStage?.(stage) ?? stage.replayKey ?? deps.nextReplayKey(stage.name); if (stage.sessionFile === undefined) return false; + const checkpointNow = deps.now?.() ?? Date.now(); + const durationMs = elapsedStageMs(stage, checkpointNow) ?? 0; const current = deps.backend.getStageSession(deps.workflowId, replayKey); - if (current?.sessionFile === stage.sessionFile) return false; + if (current !== undefined + && current.sessionId === stage.sessionId + && current.sessionFile === stage.sessionFile + && current.startedAt === stage.startedAt + && current.durationMs === durationMs) return false; const checkpoint: DurableStageCheckpoint = { kind: "stage", workflowId: deps.workflowId, - checkpointId: stageSessionCheckpointId(replayKey, stage), + checkpointId: stageSessionCheckpointId(replayKey, stage, durationMs), name: stage.name, replayKey, ...(stage.sessionId !== undefined ? { sessionId: stage.sessionId } : {}), sessionFile: stage.sessionFile, - completedAt: Date.now(), + ...(stage.startedAt !== undefined ? { startedAt: stage.startedAt } : {}), + durationMs, + completedAt: checkpointNow, }; await recordCheckpointDurably(deps.backend, checkpoint); return true; @@ -107,7 +117,10 @@ export function createDurableStagePrimitive(input: { const liveOptions: StageOptions | undefined = { ...(options ?? {}), durableReplayKey: replayKey, - ...(isMidSessionResume ? { resumeFromSessionFile: session.sessionFile } : {}), + ...(isMidSessionResume ? { + resumeFromSessionFile: session.sessionFile, + durableAccumulatedDurationMs: session.durationMs ?? 0, + } : {}), }; const live = withMidSessionResumePrompt(input.stage(name, liveOptions, replayKey), isMidSessionResume); if (options?.schema === undefined) return live; @@ -139,7 +152,10 @@ export function createDurableTaskPrimitive(input: { const taskOptions: WorkflowTaskOptions = { ...options, durableReplayKey: replayKey, - ...(session?.sessionFile !== undefined ? { resumeFromSessionFile: session.sessionFile } : {}), + ...(session?.sessionFile !== undefined ? { + resumeFromSessionFile: session.sessionFile, + durableAccumulatedDurationMs: session.durationMs ?? 0, + } : {}), }; const result = await input.task(name, taskOptions, stageFailFastScope); await recordCheckpointDurably(input.backend, { @@ -299,8 +315,11 @@ function metadataValue( checkpoints: readonly DurableStageCheckpoint[], key: K, ): Pick | Record { - const value = checkpoints.find((checkpoint) => checkpoint[key] !== undefined)?.[key]; - return value === undefined ? {} : { [key]: value } as Pick; + for (let index = checkpoints.length - 1; index >= 0; index -= 1) { + const value = checkpoints[index]?.[key]; + if (value !== undefined) return { [key]: value } as Pick; + } + return {}; } @@ -317,10 +336,12 @@ export function stableCheckpointId(kind: string, replayKey: string): string { return `${kind}:${replayKey}`; } -function stageSessionCheckpointId(replayKey: string, stage: StageSnapshot): string { +function stageSessionCheckpointId(replayKey: string, stage: StageSnapshot, durationMs: number): string { return `${stableCheckpointId("stage-session", replayKey)}:${durableHash({ sessionId: stage.sessionId ?? "", sessionFile: stage.sessionFile ?? "", + startedAt: stage.startedAt ?? 0, + durationMs, })}`; } diff --git a/packages/workflows/src/runs/foreground/executor-stage-call.ts b/packages/workflows/src/runs/foreground/executor-stage-call.ts index 1885e577f..5ee32f4a4 100644 --- a/packages/workflows/src/runs/foreground/executor-stage-call.ts +++ b/packages/workflows/src/runs/foreground/executor-stage-call.ts @@ -7,6 +7,7 @@ import { raceAbort } from "./executor-abort.js"; import { hasExplicitFastModeCandidate } from "./executor-direct-helpers.js"; import { applyFailureToStage } from "./executor-lifecycle.js"; import { isTerminalStage } from "./executor-scheduler.js"; +import { rebasedStageStartedAt } from "../../shared/timing.js"; export interface TrackedStageCallOptions { readonly eagerSession?: boolean; @@ -110,8 +111,9 @@ export function createTrackedStageCaller(input: { } } if (trackStageLifecycle) { + const now = Date.now(); runtime.stageSnapshot.status = "running"; - runtime.stageSnapshot.startedAt = Date.now(); + runtime.stageSnapshot.startedAt ??= rebasedStageStartedAt(input.options?.durableAccumulatedDurationMs, now); const hasNoExplicitModelConfig = input.options?.model === undefined && input.options?.fallbackModels === undefined; const promptAdapterHandlesInitialPrompt = input.adapters.prompt !== undefined; if (callOptions.eagerSession && !promptAdapterHandlesInitialPrompt && (hasNoExplicitModelConfig || await hasExplicitFastModeCandidate({ diff --git a/packages/workflows/src/runs/foreground/stage-runner-options.ts b/packages/workflows/src/runs/foreground/stage-runner-options.ts index 1a8adad05..db677e8bf 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-options.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-options.ts @@ -20,6 +20,7 @@ export function stripWorkflowOnlyOptions( forkFromSessionFile, resumeFromSessionFile, durableReplayKey: _durableReplayKey, + durableAccumulatedDurationMs: _durableAccumulatedDurationMs, sessionDir, gitWorktreeDir: _gitWorktreeDir, baseBranch: _baseBranch, diff --git a/packages/workflows/src/shared/timing.ts b/packages/workflows/src/shared/timing.ts index 73f8a3279..c51fdbc3a 100644 --- a/packages/workflows/src/shared/timing.ts +++ b/packages/workflows/src/shared/timing.ts @@ -18,6 +18,10 @@ function nonNegative(ms: number): number { return Math.max(0, ms); } +export function rebasedStageStartedAt(accumulatedDurationMs: number | undefined, resumedAt: number): number { + return resumedAt - nonNegative(accumulatedDurationMs ?? 0); +} + function elapsedFromStart( startedAt: number, now: number, diff --git a/packages/workflows/src/shared/types.ts b/packages/workflows/src/shared/types.ts index 3f34c06e6..f66e4dca4 100644 --- a/packages/workflows/src/shared/types.ts +++ b/packages/workflows/src/shared/types.ts @@ -191,6 +191,8 @@ export interface StageOptions { assert.equal(backend2.getWorkflow(WORKFLOW_ID)!.name, "file-workflow"); }); + test("preserves latest active-stage timing across a file process boundary", () => { + const replayKey = "stage:analyze:1"; + backend.recordCheckpoint({ + kind: "stage", workflowId: WORKFLOW_ID, checkpointId: "stage-session:1", name: "analyze", replayKey, + sessionFile: "/tmp/analyze.jsonl", startedAt: 1000, durationMs: 400, completedAt: 1400, + }); + backend.recordCheckpoint({ + kind: "stage", workflowId: WORKFLOW_ID, checkpointId: "stage-session:2", name: "analyze", replayKey, + sessionFile: "/tmp/analyze.jsonl", startedAt: 1000, durationMs: 750, completedAt: 1750, + }); + + const fresh = new FileDurableBackend(join(tmpDir, "state.json")); + assert.deepEqual(fresh.getStageSession(WORKFLOW_ID, replayKey), { + sessionFile: "/tmp/analyze.jsonl", + startedAt: 1000, + durationMs: 750, + }); + }); + test("lists resumable workflows from a new backend instance", () => { backend.recordCheckpoint(makeToolCheckpoint(WORKFLOW_ID, "progress", "h-progress", "done")); backend.setWorkflowStatus(WORKFLOW_ID, "paused"); diff --git a/test/unit/durable-dbos-backend.test.ts b/test/unit/durable-dbos-backend.test.ts index 75c92774f..4794ff66c 100644 --- a/test/unit/durable-dbos-backend.test.ts +++ b/test/unit/durable-dbos-backend.test.ts @@ -300,15 +300,22 @@ describe("DbosDurableBackend hydration (fresh process)", () => { test("hydrateWorkflow reconstructs stage checkpoints from DBOS envelopes", async () => { const cp: DurableStageCheckpoint = { - kind: "stage", workflowId: "wf-h3", checkpointId: "stage:r1", name: "build", replayKey: "stage:build:1", output: "done", completedAt: 3000, + kind: "stage", workflowId: "wf-h3", checkpointId: "stage:r1", name: "build", replayKey: "stage:build:1", output: "done", completedAt: 1400, + sessionFile: "/tmp/build.jsonl", startedAt: 1000, durationMs: 400, }; seedMockWorkflow(sdk, { workflowId: "wf-h3", name: "test", status: "PENDING" }); seedMockCheckpoint(sdk, "wf-h3", cp); + seedMockCheckpoint(sdk, "wf-h3", { + ...cp, checkpointId: "stage-session:r2", output: undefined, durationMs: 900, completedAt: 1900, + }); const fresh = new DbosDurableBackend(sdk); assert.equal(fresh.getStageOutput("wf-h3", "stage:build:1"), undefined); await fresh.hydrateWorkflow("wf-h3"); + await fresh.hydrateWorkflow("wf-h3"); assert.equal(fresh.getStageOutput("wf-h3", "stage:build:1"), "done"); + assert.equal(fresh.getStageSession("wf-h3", "stage:build:1")?.durationMs, 900); + assert.equal(fresh.listCheckpoints("wf-h3").length, 2); }); test("hydrateWorkflow handles legacy/simple payloads gracefully", async () => { diff --git a/test/unit/durable-stage-session-resume.test.ts b/test/unit/durable-stage-session-resume.test.ts index 2267c5167..354475589 100644 --- a/test/unit/durable-stage-session-resume.test.ts +++ b/test/unit/durable-stage-session-resume.test.ts @@ -1,11 +1,18 @@ -import { beforeEach, describe, test } from "bun:test"; +import { afterEach, beforeEach, describe, mock, spyOn, test } from "bun:test"; import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { InMemoryDurableBackend } from "../../packages/workflows/src/durable/backend.js"; +import { FileDurableBackend } from "../../packages/workflows/src/durable/file-backend.js"; import { createCheckpointIdGenerator } from "../../packages/workflows/src/durable/tool-primitive.js"; -import { createDurableStagePrimitive, createStageReplayKeyGenerator, recordStageCheckpoint, recordStageSessionCheckpoint } from "../../packages/workflows/src/durable/stage-primitive.js"; +import { createDurableStagePrimitive, createDurableTaskPrimitive, createStageReplayKeyGenerator, recordStageCheckpoint, recordStageSessionCheckpoint, stageCheckpointWithOutput } from "../../packages/workflows/src/durable/stage-primitive.js"; import { RESUME_CONTINUATION_PROMPT } from "../../packages/workflows/src/runs/foreground/executor.js"; import type { StageSnapshot } from "../../packages/workflows/src/shared/store-types.js"; +import { elapsedStageMs, rebasedStageStartedAt } from "../../packages/workflows/src/shared/timing.js"; +import { createStore, run, Type, workflow } from "./executor-shared.js"; +afterEach(() => mock.restore()); const WORKFLOW_ID = "wf-stage-session-resume"; function makeStage(overrides: Partial = {}): StageSnapshot { @@ -59,12 +66,13 @@ describe("durable stage session resume", () => { }); }); - function deps() { + function deps(now = 2000) { return { workflowId: WORKFLOW_ID, backend, nextCheckpointId: createCheckpointIdGenerator(), nextReplayKey: createStageReplayKeyGenerator(WORKFLOW_ID), + now: () => now, }; } @@ -75,6 +83,8 @@ describe("durable stage session resume", () => { assert.deepEqual(backend.getStageSession(WORKFLOW_ID, "stage:analyze:1"), { sessionId: "sid-1", sessionFile: "/tmp/stage.jsonl", + startedAt: 1000, + durationMs: 1000, }); // Running (active) workflows are hidden from resume; quitting flips the // durable handle to paused, which is when an in-progress stage session @@ -83,6 +93,47 @@ describe("durable stage session resume", () => { assert.equal(backend.listResumableWorkflows().length, 1); }); + test("refreshes accumulated active duration for repeated checkpoints of one session", async () => { + const replayKey = "stage:analyze:1"; + const stage = makeStage({ replayKey, sessionId: "sid-1", sessionFile: "/tmp/stage.jsonl" }); + + assert.equal(await recordStageSessionCheckpoint(deps(1400), stage), true); + assert.equal(await recordStageSessionCheckpoint(deps(1750), stage), true); + assert.equal(await recordStageSessionCheckpoint(deps(1750), stage), false); + + assert.deepEqual(backend.getStageSession(WORKFLOW_ID, replayKey), { + sessionId: "sid-1", + sessionFile: "/tmp/stage.jsonl", + startedAt: 1000, + durationMs: 750, + }); + assert.equal(backend.listCheckpoints(WORKFLOW_ID).length, 2); + }); + + test("checkpoints pause-adjusted duration without double-counting", async () => { + const replayKey = "stage:analyze:1"; + const stage = makeStage({ + replayKey, + sessionFile: "/tmp/stage.jsonl", + pausedDurationMs: 200, + pausedAt: 1800, + }); + + await recordStageSessionCheckpoint(deps(2200), stage); + + assert.equal(backend.getStageSession(WORKFLOW_ID, replayKey)?.durationMs, 600); + }); + + test("counts post-resume elapsed time while excluding a new pause exactly once", () => { + const resumedAt = 5000; + const startedAt = rebasedStageStartedAt(700, resumedAt); + const completedAt = 5500; + + assert.equal(startedAt, 4300); + assert.equal(elapsedStageMs({ startedAt, pausedDurationMs: 200 }, completedAt), 1000); + assert.equal(rebasedStageStartedAt(-50, resumedAt), resumedAt); + }); + test("reopens prior session file when output is not completed", async () => { const replayKey = "stage:analyze:1"; await recordStageSessionCheckpoint(deps(), makeStage({ replayKey, sessionFile: "/tmp/prior.jsonl" })); @@ -108,6 +159,43 @@ describe("durable stage session resume", () => { assert.equal(observedPrompt, RESUME_CONTINUATION_PROMPT); }); + test("hydrates accumulated duration into a new-process live stage", async () => { + const replayKey = "stage:analyze:1"; + await recordStageSessionCheckpoint(deps(1700), makeStage({ replayKey, sessionFile: "/tmp/prior.jsonl" })); + let accumulatedDurationMs: number | undefined; + const stage = createDurableStagePrimitive({ + workflowId: WORKFLOW_ID, + backend, + nextReplayKey: () => replayKey, + stage: (_name, options) => { + accumulatedDurationMs = options?.durableAccumulatedDurationMs; + return fakeStageContext("resumed"); + }, + }); + + await stage("analyze").prompt("continue"); + assert.equal(accumulatedDurationMs, 700); + }); + + test("hydrates accumulated duration into a resumed task", async () => { + const replayKey = "stage:task:analyze:1"; + await recordStageSessionCheckpoint(deps(1700), makeStage({ replayKey, sessionFile: "/tmp/prior-task.jsonl" })); + let observedOptions: { resumeFromSessionFile?: string; durableAccumulatedDurationMs?: number } | undefined; + const task = createDurableTaskPrimitive({ + workflowId: WORKFLOW_ID, + backend, + nextReplayKey: () => replayKey, + task: async (_name, options) => { + observedOptions = options; + return { name: "analyze", stageName: "analyze", text: "resumed task" }; + }, + }); + + assert.equal((await task("analyze", { prompt: "continue" })).text, "resumed task"); + assert.equal(observedOptions?.resumeFromSessionFile, "/tmp/prior-task.jsonl"); + assert.equal(observedOptions?.durableAccumulatedDurationMs, 700); + }); + test("mid-session resume does not eagerly read throwing StageContext getters", async () => { const replayKey = "stage:analyze:1"; await recordStageSessionCheckpoint(deps(), makeStage({ replayKey, sessionFile: "/tmp/prior.jsonl" })); @@ -141,9 +229,86 @@ describe("durable stage session resume", () => { test("updates session metadata across repeated resumes", async () => { const replayKey = "stage:analyze:1"; - assert.equal(await recordStageSessionCheckpoint(deps(), makeStage({ replayKey, sessionFile: "/tmp/first.jsonl" })), true); - assert.equal(await recordStageSessionCheckpoint(deps(), makeStage({ replayKey, sessionFile: "/tmp/second.jsonl" })), true); - assert.deepEqual(backend.getStageSession(WORKFLOW_ID, replayKey), { sessionFile: "/tmp/second.jsonl" }); + assert.equal(await recordStageSessionCheckpoint(deps(1500), makeStage({ replayKey, sessionFile: "/tmp/first.jsonl" })), true); + assert.equal(await recordStageSessionCheckpoint(deps(1800), makeStage({ replayKey, sessionFile: "/tmp/second.jsonl" })), true); + assert.deepEqual(backend.getStageSession(WORKFLOW_ID, replayKey), { + sessionFile: "/tmp/second.jsonl", + startedAt: 1000, + durationMs: 800, + }); + }); + + test("preserves pause-adjusted duration across two process-boundary resumes", async () => { + const dir = mkdtempSync(join(tmpdir(), "atomic-stage-repeated-resume-")); + try { + const runId = "wf-stage-repeated-resume"; + const replayKey = "stage:analyze:1"; + const stateFile = join(dir, "durable.json"); + const firstProcess = new FileDurableBackend(stateFile); + firstProcess.registerWorkflow({ workflowId: runId, name: "repeated-resume", inputs: {}, createdAt: 1000, status: "paused" }); + await recordStageSessionCheckpoint({ + workflowId: runId, + backend: firstProcess, + nextCheckpointId: createCheckpointIdGenerator(), + nextReplayKey: () => replayKey, + now: () => 1800, + }, makeStage({ replayKey, sessionFile: "/tmp/process-a.jsonl", pausedDurationMs: 100 })); + + const secondProcess = new FileDurableBackend(stateFile); + const secondBaseline = secondProcess.getStageSession(runId, replayKey)?.durationMs; + assert.equal(secondBaseline, 700); + await recordStageSessionCheckpoint({ + workflowId: runId, + backend: secondProcess, + nextCheckpointId: createCheckpointIdGenerator(), + nextReplayKey: () => replayKey, + now: () => 5600, + }, makeStage({ + replayKey, + sessionFile: "/tmp/process-b.jsonl", + startedAt: rebasedStageStartedAt(secondBaseline, 5000), + pausedDurationMs: 200, + })); + + const thirdProcess = new FileDurableBackend(stateFile); + const thirdBaseline = thirdProcess.getStageSession(runId, replayKey)?.durationMs; + assert.equal(thirdBaseline, 1100); + const completed = makeStage({ + replayKey, + sessionFile: "/tmp/process-c.jsonl", + status: "completed", + result: "done", + startedAt: rebasedStageStartedAt(thirdBaseline, 9000), + endedAt: 9500, + pausedDurationMs: 300, + durationMs: 1300, + }); + await recordStageSessionCheckpoint({ + workflowId: runId, + backend: thirdProcess, + nextCheckpointId: createCheckpointIdGenerator(), + nextReplayKey: () => replayKey, + now: () => 9500, + }, completed); + await recordStageCheckpoint({ + workflowId: runId, + backend: thirdProcess, + nextCheckpointId: createCheckpointIdGenerator(), + nextReplayKey: () => replayKey, + }, completed); + + const replayBackend = new FileDurableBackend(stateFile); + assert.equal(replayBackend.getStageSession(runId, replayKey)?.durationMs, 1300); + const replayed = createDurableStagePrimitive({ + workflowId: runId, + backend: replayBackend, + nextReplayKey: () => replayKey, + stage: () => { throw new Error("completed stage must not run after repeated resume"); }, + }); + assert.equal(await replayed("analyze").prompt("ignored"), "done"); + } finally { + rmSync(dir, { recursive: true, force: true }); + } }); test("completed output wins over later session metadata", async () => { @@ -151,7 +316,7 @@ describe("durable stage session resume", () => { await recordStageCheckpoint(deps(), makeStage({ status: "completed", replayKey, result: "done", endedAt: 2000 })); await recordStageSessionCheckpoint(deps(), makeStage({ replayKey, sessionFile: "/tmp/later.jsonl" })); assert.equal(backend.getStageOutput(WORKFLOW_ID, replayKey), "done"); - assert.deepEqual(backend.getStageSession(WORKFLOW_ID, replayKey), { sessionFile: "/tmp/later.jsonl" }); + assert.deepEqual(backend.getStageSession(WORKFLOW_ID, replayKey), { sessionFile: "/tmp/later.jsonl", startedAt: 1000, durationMs: 1000 }); const stage = createDurableStagePrimitive({ workflowId: WORKFLOW_ID, backend, @@ -166,6 +331,137 @@ describe("durable stage session resume", () => { await recordStageSessionCheckpoint(deps(), makeStage({ replayKey, sessionFile: "/tmp/first.jsonl" })); await recordStageCheckpoint(deps(), makeStage({ status: "completed", replayKey, result: "done", endedAt: 2000 })); assert.equal(backend.getStageOutput(WORKFLOW_ID, replayKey), "done"); - assert.deepEqual(backend.getStageSession(WORKFLOW_ID, replayKey), { sessionFile: "/tmp/first.jsonl" }); + assert.deepEqual(backend.getStageSession(WORKFLOW_ID, replayKey), { sessionFile: "/tmp/first.jsonl", startedAt: 1000, durationMs: 1000 }); + }); + + test("hydrates schema-backed replay from the latest timing metadata", async () => { + const replayKey = "stage:analyze:1"; + let clock = 1300; + spyOn(Date, "now").mockImplementation(() => clock); + const active = makeStage({ replayKey, sessionFile: "/tmp/schema-stage.jsonl" }); + await recordStageSessionCheckpoint(deps(1111), active); + await recordStageSessionCheckpoint(deps(1222), active); + + let liveStageCalls = 0; + const stage = createDurableStagePrimitive({ + workflowId: WORKFLOW_ID, + backend, + nextReplayKey: () => replayKey, + stage: () => { + liveStageCalls += 1; + return Object.assign(fakeStageContext("") as object, { + prompt: async () => ({ answer: "done" }), + }) as never; + }, + }); + const schema = Type.Object({ answer: Type.String() }); + assert.deepEqual(await stage("analyze", { schema }).prompt("analyze"), { answer: "done" }); + + const activeHydration = stageCheckpointWithOutput(backend, WORKFLOW_ID, replayKey); + assert.deepEqual(activeHydration?.output, { answer: "done" }); + assert.equal(activeHydration?.durationMs, 222); + + clock = 1400; + await recordStageCheckpoint(deps(), makeStage({ + replayKey, + status: "completed", + result: "done", + endedAt: clock, + durationMs: 333, + })); + const completedHydration = stageCheckpointWithOutput(backend, WORKFLOW_ID, replayKey); + assert.deepEqual(completedHydration?.output, { answer: "done" }); + assert.equal(completedHydration?.durationMs, 333); + + const replayed = createDurableStagePrimitive({ + workflowId: WORKFLOW_ID, + backend, + nextReplayKey: () => replayKey, + stage: () => { throw new Error("schema-backed replay must not execute the live stage"); }, + }); + assert.deepEqual(await replayed("analyze", { schema }).prompt("ignored"), { answer: "done" }); + assert.equal(liveStageCalls, 1); + }); + + test("file process-boundary completion preserves duration across concurrent tracked calls and replay", async () => { + const dir = mkdtempSync(join(tmpdir(), "atomic-stage-duration-")); + try { + const runId = "wf-stage-duration-resume"; + const replayKey = "stage:analyze:1"; + const stateFile = join(dir, "durable.json"); + const writer = new FileDurableBackend(stateFile); + writer.registerWorkflow({ workflowId: runId, name: "duration-resume", inputs: {}, createdAt: 1000, status: "paused" }); + await recordStageSessionCheckpoint({ + workflowId: runId, + backend: writer, + nextCheckpointId: createCheckpointIdGenerator(), + nextReplayKey: () => replayKey, + now: () => 1700, + }, makeStage({ replayKey, sessionFile: "/tmp/durable-stage-duration.jsonl" })); + + let clock = 5000; + let liveStageCalls = 0; + let releaseCalls: () => void = () => {}; + const bothCallsStarted = new Promise((resolve) => { releaseCalls = resolve; }); + spyOn(Date, "now").mockImplementation(() => clock); + let lifecycleDurationMs: number | undefined; + const store = createStore(); + const def = workflow({ + name: "duration-resume", + description: "", + inputs: {}, + outputs: { result: Type.String() }, + run: async (ctx) => { + const stage = ctx.stage("analyze"); + await Promise.allSettled([ + stage.complete("first"), + stage.complete("second"), + ]); + return { result: "done" }; + }, + }); + const resumedBackend = new FileDurableBackend(stateFile); + const first = await run(def, {}, { + runId, + store, + durableBackend: resumedBackend, + adapters: { complete: { complete: async (text) => { + liveStageCalls += 1; + if (liveStageCalls === 1) { + clock = 5100; + await bothCallsStarted; + } else { + clock = 5300; + releaseCalls(); + } + return text; + } } }, + onStageEnd: (_stageRunId, snapshot) => { lifecycleDurationMs = snapshot.durationMs; }, + }); + + const storedStage = store.runs()[0]?.stages.find((stage) => stage.name === "analyze"); + const durableStage = resumedBackend.listCheckpoints(runId).find((checkpoint) => + checkpoint.kind === "stage" && checkpoint.replayKey === replayKey && checkpoint.output !== undefined, + ); + assert.equal(first.status, "completed"); + assert.equal(liveStageCalls, 2); + assert.equal(storedStage?.durationMs, 1000); + assert.equal(lifecycleDurationMs, 1000); + assert.equal(durableStage?.kind === "stage" ? durableStage.durationMs : undefined, 1000); + + const replay = await run(def, {}, { + runId, + store: createStore(), + durableBackend: new FileDurableBackend(stateFile), + adapters: { complete: { complete: async () => { + liveStageCalls += 1; + throw new Error("completed stage replay must not execute again"); + } } }, + }); + assert.equal(replay.status, "completed"); + assert.equal(liveStageCalls, 2); + } finally { + rmSync(dir, { recursive: true, force: true }); + } }); });