diff --git a/packages/workflows/CHANGELOG.md b/packages/workflows/CHANGELOG.md index 5e95d6dd7..f864d95a9 100644 --- a/packages/workflows/CHANGELOG.md +++ b/packages/workflows/CHANGELOG.md @@ -20,7 +20,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed -- Fixed workflow graph rendering for large, wide fan-outs by clipping cards, edges, and composed rows to the visible viewport, retaining cached topology and layout across status-only updates, and making idle animation eligibility constant-time ([#2100](https://github.com/bastani-inc/atomic/issues/2100)). +- Fixed workflow graph rendering for large, wide fan-outs and payload-heavy runs by clipping cards, edges, and composed rows to the visible viewport, retaining cached topology and layout across status-only updates, making idle animation eligibility constant-time, and routing interactive store updates through a cached payload-free graph projection instead of cloning full workflow inputs, results, child outputs, and tool bodies ([#2100](https://github.com/bastani-inc/atomic/issues/2100)). - Fixed `open-claude-design` converting a failed `user-feedback-*` stage into design approval (the [#1499](https://github.com/bastani-inc/atomic/issues/1499) catch-then-approve path). A rejected feedback stage — provider errors, model-fallback exhaustion, broken session forks — now propagates and fails the run instead of silently setting `approved_for_export` and exporting an unreviewed design. Resolved results are unchanged: a stage that runs and reports no meaningful notes, including degraded manual/headless display paths, still legitimately reads as approval, and browser/tooling trouble continues to resolve gracefully rather than reject (playwright-cli auto-install, the early browser exit, and the stage's degradation contract are unaffected) ([#2123](https://github.com/bastani-inc/atomic/issues/2123)). - Fixed `open-claude-design` embedding unbounded research context inline in single stage prompts, the same single-message overflow class as the [#1499](https://github.com/bastani-inc/atomic/issues/1499) `413 request_too_large`. The composed project design context now persists to `/design-context.md` and the curated references brief always persists to `/references.md` (with the no-references fallback when discovery is skipped); `reference-discovery` reads the design context, and the generate and exporter stages read both files via `reads` with explicit read-the-file instructions instead of inline `design_system`/`reference_inspiration` embeds. The joined research results are no longer injected as `previous` payloads into `generate-1` or `reference-discovery`. Because the files are required stage inputs, a failed write no longer passes silently: the run exits `blocked` through `ctx.exit` with the artifact paths surfaced (or propagates the error when `ctx.exit` is unavailable) instead of dispatching stages against nonexistent context. Deliberately small inline payloads are unchanged: verbatim-threaded user annotations (the #1464 guardrail) and the word-capped prior design summary ([#2121](https://github.com/bastani-inc/atomic/issues/2121)). - Fixed `open-claude-design` blocking on browser live review while reporting `running`, leaving the user with no input affordance. Each refinement round now raises a deterministic run-level `ctx.ui` prompt before its `user-feedback-*` stage starts, so the needs-attention badge fires and the prompt names the preview path and `file://` URL; choosing the skip option accepts the current design and proceeds to export. Only the executor's unavailable-UI rejection (headless mode or a missing UI adapter) degrades to running the review directly; any other gate failure — interruption, a failed durable checkpoint — propagates and stops the run. The feedback-stage prompt additionally instructs the agent to print the live `http://` review URL before starting any long-poll wait, so the URL is visible at the top of the stage output when attaching mid-run ([#2060](https://github.com/bastani-inc/atomic/issues/2060)). diff --git a/packages/workflows/src/extension/extension-runtime-state.ts b/packages/workflows/src/extension/extension-runtime-state.ts index 3237ecc4e..ef9a0e004 100644 --- a/packages/workflows/src/extension/extension-runtime-state.ts +++ b/packages/workflows/src/extension/extension-runtime-state.ts @@ -3,6 +3,7 @@ import type { StageAdapters } from "../runs/foreground/stage-runner.js"; import type { SessionManager } from "../shared/persistence-restore.js"; import { stageUiBroker } from "../shared/stage-ui-broker.js"; import { store } from "../shared/store.js"; +import { readGraphStoreSnapshot } from "../shared/store-observation.js"; import type { RunSnapshot } from "../shared/store-types.js"; import type { WorkflowExecutionPolicy, @@ -87,7 +88,10 @@ export function createWorkflowExtensionRuntimeState( const lifecycleNotificationState = createWorkflowLifecycleNotificationState(); const hilAnswerNotificationState = createWorkflowHilAnswerNotificationState(); const beforeRestoreCompleted = (snapshots: readonly RunSnapshot[]): void => { - seedWorkflowLifecycleNotificationState(lifecycleNotificationState, { ...store.snapshot(), runs: snapshots }); + seedWorkflowLifecycleNotificationState(lifecycleNotificationState, { + ...readGraphStoreSnapshot(store), + runs: snapshots, + }); }; const lifecycleNotificationConfigRef: { current: WorkflowLifecycleNotificationConfig } = { current: WORKFLOW_CONFIG_DEFAULTS.workflowNotifications, diff --git a/packages/workflows/src/extension/hil-answer-notifications.ts b/packages/workflows/src/extension/hil-answer-notifications.ts index 96151f19d..26f237e18 100644 --- a/packages/workflows/src/extension/hil-answer-notifications.ts +++ b/packages/workflows/src/extension/hil-answer-notifications.ts @@ -1,5 +1,6 @@ import type { StageUiBroker } from "../shared/stage-ui-broker.js"; import type { Store } from "../shared/store.js"; +import { readGraphStoreSnapshot, subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { PendingPrompt, PromptKind, @@ -99,10 +100,12 @@ export function installWorkflowHilAnswerNotifications(options: WorkflowHilAnswer } }; - const unsubscribeStore = options.store.subscribe(inspectWorkflowPromptAnswers); + const unsubscribeStore = subscribeStoreInvalidation(options.store, () => + inspectWorkflowPromptAnswers(readGraphStoreSnapshot(options.store)), + ); const unsubscribeBroker = options.stageUiBroker?.onStagePromptResolved((event) => { if (event.answerSource === "workflow_tool") return; - const answeredStage = findStageSnapshot(options.store.snapshot(), event.runId, event.stageId); + const answeredStage = findStageSnapshot(readGraphStoreSnapshot(options.store), event.runId, event.stageId); if (answeredStage === undefined) return; emitOnce( diff --git a/packages/workflows/src/extension/lifecycle-notifications.ts b/packages/workflows/src/extension/lifecycle-notifications.ts index b4383b81c..045c0a51b 100644 --- a/packages/workflows/src/extension/lifecycle-notifications.ts +++ b/packages/workflows/src/extension/lifecycle-notifications.ts @@ -7,6 +7,7 @@ import { } from "../shared/returned-run-status.js"; import { isTopLevelWorkflowRun } from "../shared/run-visibility.js"; import type { Store } from "../shared/store.js"; +import { readGraphStoreSnapshot, subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { PendingPrompt, PromptKind, @@ -175,7 +176,9 @@ export function installWorkflowLifecycleNotifications(options: WorkflowLifecycle const notifyOn = new Set(options.config.notifyOn); const state = options.state ?? createWorkflowLifecycleNotificationState(); let delivery!: ReturnType; - if (options.seedExisting !== false) seedWorkflowLifecycleNotificationState(state, options.store.snapshot()); + if (options.seedExisting !== false) { + seedWorkflowLifecycleNotificationState(state, readGraphStoreSnapshot(options.store)); + } const emit = (details: WorkflowLifecycleNoticeDetails): boolean | Promise => { try { @@ -266,8 +269,8 @@ export function installWorkflowLifecycleNotifications(options: WorkflowLifecycle if (notifyOn.has(details.kind)) delivery.deliver(key, details); } - const unsubscribe = options.store.subscribe(inspect); - inspect(options.store.snapshot()); + const unsubscribe = subscribeStoreInvalidation(options.store, () => inspect(readGraphStoreSnapshot(options.store))); + inspect(readGraphStoreSnapshot(options.store)); return () => { unsubscribe(); delivery.dispose(); diff --git a/packages/workflows/src/extension/postmortem-deps.ts b/packages/workflows/src/extension/postmortem-deps.ts index 464bcdbe0..cb1d25e55 100644 --- a/packages/workflows/src/extension/postmortem-deps.ts +++ b/packages/workflows/src/extension/postmortem-deps.ts @@ -22,6 +22,7 @@ import { import { stageControlRegistry } from "../runs/foreground/stage-control-registry.js"; import type { StageAdapters } from "../runs/foreground/stage-runner.js"; import { store } from "../shared/store.js"; +import { readGraphStoreSnapshot } from "../shared/store-observation.js"; export interface PostMortemResolverDeps { readonly adapters: StageAdapters; @@ -33,7 +34,7 @@ function resolveStageCwd(runId: string): string | undefined { try { const backend = getDurableBackend(); const owningHandle = backend.getWorkflow(runId); - const run = store.snapshot().runs.find((candidate) => candidate.id === runId); + const run = readGraphStoreSnapshot(store).runs.find((candidate) => candidate.id === runId); const rootRunId = run?.rootRunId ?? owningHandle?.rootWorkflowId; const cwdHandle = rootRunId === undefined ? owningHandle : (backend.getWorkflow(rootRunId) ?? owningHandle); return cwdHandle?.workflowCwd ?? cwdHandle?.invocationCwd ?? undefined; @@ -61,7 +62,7 @@ export function createPostMortemHandleResolver( deps: PostMortemResolverDeps, ): (runId: string, stageId: string) => EnsurePostMortemStageHandleResult | undefined { return (runId, stageId) => { - const run = store.snapshot().runs.find((candidate) => candidate.id === runId); + const run = readGraphStoreSnapshot(store).runs.find((candidate) => candidate.id === runId); const stage = run?.stages.find((candidate) => candidate.id === stageId); if (stage === undefined) return undefined; return ensurePostMortemStageHandle(runId, stage, postMortemDepsForRun(runId, deps)); diff --git a/packages/workflows/src/extension/workflow-resume-picker-rows.ts b/packages/workflows/src/extension/workflow-resume-picker-rows.ts index d315e4972..a58fdaddc 100644 --- a/packages/workflows/src/extension/workflow-resume-picker-rows.ts +++ b/packages/workflows/src/extension/workflow-resume-picker-rows.ts @@ -12,6 +12,7 @@ import { getDurableBackend } from "../durable/factory.js"; import type { ResumableWorkflowEntry } from "../durable/types.js"; import { topLevelWorkflowRuns } from "../shared/run-visibility.js"; import type { Store } from "../shared/store.js"; +import { subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { RunSnapshot } from "../shared/store-types.js"; import type { WorkflowResumeRefresh } from "../tui/workflow-resume-selector.js"; import type { ExtensionRuntime } from "./runtime.js"; @@ -66,7 +67,7 @@ export function resumePickerLiveUpdateOptions( runtime: ExtensionRuntime, ): ResumePickerLiveUpdateOptions { return { - watch: (onChange) => runStore.subscribe(() => onChange()), + watch: (onChange) => subscribeStoreInvalidation(runStore, onChange), refresh: async () => { const current = collectResumePickerLiveRuns(runStore); const catalog = await prepareWorkflowResumeCatalog(runtime, current.activeLiveIds); diff --git a/packages/workflows/src/extension/workflow-resume-shadow.ts b/packages/workflows/src/extension/workflow-resume-shadow.ts index a5f89f57b..02dafb095 100644 --- a/packages/workflows/src/extension/workflow-resume-shadow.ts +++ b/packages/workflows/src/extension/workflow-resume-shadow.ts @@ -5,6 +5,7 @@ import { getLoadableDurableWorkflow } from "../durable/workflow-status-transitio import { type JobTracker, jobTracker } from "../runs/background/job-tracker.js"; import { type StageControlRegistry, stageControlRegistry } from "../runs/foreground/stage-control-registry.js"; import { expandWorkflowGraph } from "../shared/expanded-workflow-graph.js"; +import { readGraphStoreSnapshot } from "../shared/store-observation.js"; import type { Store } from "../shared/store-public-types.js"; import type { RunSnapshot } from "../shared/store-types.js"; @@ -28,7 +29,7 @@ export function classifyDurableResumeShadow( const jobs = deps.jobs ?? jobTracker; if (jobs.has(run.id)) return "not_shadow"; const controls = deps.stageControls ?? stageControlRegistry; - const graph = expandWorkflowGraph(store.snapshot(), run.id); + const graph = expandWorkflowGraph(readGraphStoreSnapshot(store), run.id); const controlRunIds = new Set([run.id]); for (const stage of graph.stages) controlRunIds.add(stage.workflowGraphTarget.runId); if ([...controlRunIds].some((runId) => controls.run(runId).stages().length > 0)) return "not_shadow"; diff --git a/packages/workflows/src/extension/workflow-targets.ts b/packages/workflows/src/extension/workflow-targets.ts index 9615a4dac..348e6686e 100644 --- a/packages/workflows/src/extension/workflow-targets.ts +++ b/packages/workflows/src/extension/workflow-targets.ts @@ -7,6 +7,7 @@ import { } from "../shared/expanded-workflow-graph.js"; import { topLevelWorkflowRuns } from "../shared/run-visibility.js"; import { store } from "../shared/store.js"; +import { readGraphStoreSnapshot } from "../shared/store-observation.js"; import type { RunStatus } from "../shared/store-types.js"; import type { OverlayPiSurface } from "../tui/overlay-adapter.js"; import type { PiExecuteContext, WorkflowToolArgs } from "./public-types.js"; @@ -115,7 +116,7 @@ export type ToolStageTarget = { ok: true; runId?: string; stageId?: string } | { export function resolveStageTarget(runId: string, stageTarget?: string): ToolStageTarget { const target = stageTarget?.trim(); if (!target) return { ok: true, runId }; - const graph = expandWorkflowGraph(store.snapshot(), runId); + const graph = expandWorkflowGraph(readGraphStoreSnapshot(store), runId); const exactVirtualIds = graph.stages.filter((stage) => stage.id === target); if (exactVirtualIds.length === 1) return resolvedStageTarget(exactVirtualIds[0]!); if (exactVirtualIds.length > 1) return ambiguousStageTarget(target, exactVirtualIds); @@ -166,7 +167,7 @@ export type ControlNodeTarget = export function resolveControlNodeTarget(runId: string, stageTarget?: string): ControlNodeTarget { const target = stageTarget?.trim(); if (!target) return { ok: true, kind: "run" }; - const graph = expandWorkflowGraph(store.snapshot(), runId); + const graph = expandWorkflowGraph(readGraphStoreSnapshot(store), runId); const nodes = graph.renderStages; const candidates: Array = [ nodes.filter((node) => node.id === target), diff --git a/packages/workflows/src/extension/workflow-tool-send.ts b/packages/workflows/src/extension/workflow-tool-send.ts index 4256ab869..051645540 100644 --- a/packages/workflows/src/extension/workflow-tool-send.ts +++ b/packages/workflows/src/extension/workflow-tool-send.ts @@ -9,6 +9,7 @@ import { coerceStageInputAnswer, hasStageInputAnswerContent, type StageInputAnsw import { stageUiBroker } from "../shared/stage-ui-broker.js"; import { store } from "../shared/store.js"; import { isTerminalRunStatus } from "../shared/store-internal.js"; +import { subscribeStoreInvalidation } from "../shared/store-observation.js"; import { reciprocalWorkflowRootRunId } from "../shared/workflow-run-ownership.js"; import type { WorkflowToolArgs } from "./public-types.js"; import type { WorkflowToolResult } from "./render-result.js"; @@ -280,7 +281,7 @@ export async function workflowSendAction( admitted = true; }; const terminalPending = Promise.withResolvers(); - const unsubscribeTerminal = store.subscribe(() => { + const unsubscribeTerminal = subscribeStoreInvalidation(store, () => { if (admitted) return; const terminal = terminalWorkflowSendResultForRoot(rootRunId, args.stageId?.trim() ?? resolvedStageId); if (terminal !== undefined) terminalPending.reject(new WorkflowSendAdmissionError(terminal)); diff --git a/packages/workflows/src/runs/background/quit.ts b/packages/workflows/src/runs/background/quit.ts index 29a0da757..9b619bf57 100644 --- a/packages/workflows/src/runs/background/quit.ts +++ b/packages/workflows/src/runs/background/quit.ts @@ -18,6 +18,7 @@ import { WorkflowGracefulQuitError } from "../../engine/workflow-tool-abort.js"; import { expandWorkflowGraph } from "../../shared/expanded-workflow-graph.js"; import { topLevelWorkflowRuns } from "../../shared/run-visibility.js"; import { store as defaultStore } from "../../shared/store.js"; +import { readGraphStoreSnapshot } from "../../shared/store-observation.js"; import type { Store } from "../../shared/store-public-types.js"; import type { RunSnapshot, StageSnapshot } from "../../shared/store-types.js"; import type { WorkflowCancelledToolNode, WorkflowToolNodeIdentity } from "../../shared/types.js"; @@ -100,7 +101,7 @@ export async function quitRun( if (!run) return { ok: false, runId, reason: "not_found" }; if (run.endedAt !== undefined) return { ok: false, runId, reason: "already_ended" }; - const graph = expandWorkflowGraph(activeStore.snapshot(), runId); + const graph = expandWorkflowGraph(readGraphStoreSnapshot(activeStore), runId); const handles = controllableHandles(activeStore, registry, runId); const admissionBoundaries = controllableAdmissionBoundaries(activeStore, toolControls, runId); const promptStages = graph.stages.filter( @@ -303,7 +304,7 @@ function controllableHandles( registry: StageControlRegistry, runId: string, ): Array<{ controlRunId: string; handle: StageControlHandle }> { - const graph = expandWorkflowGraph(activeStore.snapshot(), runId); + const graph = expandWorkflowGraph(readGraphStoreSnapshot(activeStore), runId); const controlRunIds = new Set([runId]); for (const stage of graph.stages) controlRunIds.add(stage.workflowGraphTarget.runId); return [...controlRunIds].flatMap((controlRunId) => diff --git a/packages/workflows/src/runs/background/status.ts b/packages/workflows/src/runs/background/status.ts index 5f0ec4138..3da460e7c 100644 --- a/packages/workflows/src/runs/background/status.ts +++ b/packages/workflows/src/runs/background/status.ts @@ -15,6 +15,7 @@ import { effectiveRunStatus } from "../../shared/returned-run-status.js"; import { topLevelWorkflowRuns } from "../../shared/run-visibility.js"; import type { Store } from "../../shared/store.js"; import { store as defaultStore } from "../../shared/store.js"; +import { readGraphStoreSnapshot } from "../../shared/store-observation.js"; import type { RunSnapshot, RunStatus, StageSnapshot } from "../../shared/store-types.js"; import type { WorkflowPersistencePort } from "../../shared/types.js"; import type { StageControlRegistry } from "../foreground/stage-control-registry.js"; @@ -89,7 +90,7 @@ export { type InspectRunResult, inspectRun, type RunDetail } from "./run-inspect export function statusRuns(opts?: { all?: boolean; store?: Store }): RunStatusEntry[] { const activeStore = opts?.store ?? defaultStore; - const snapshot = activeStore.snapshot(); + const snapshot = readGraphStoreSnapshot(activeStore); return topLevelWorkflowRuns(snapshot.runs).map((run) => { const graph = expandWorkflowGraph(snapshot, run.id); return { diff --git a/packages/workflows/src/runs/background/workflow-lifecycle-aggregate.ts b/packages/workflows/src/runs/background/workflow-lifecycle-aggregate.ts index a93a4a31f..153171124 100644 --- a/packages/workflows/src/runs/background/workflow-lifecycle-aggregate.ts +++ b/packages/workflows/src/runs/background/workflow-lifecycle-aggregate.ts @@ -1,10 +1,11 @@ import { expandWorkflowGraph } from "../../shared/expanded-workflow-graph.js"; +import { readGraphStoreSnapshot } from "../../shared/store-observation.js"; import type { Store } from "../../shared/store-public-types.js"; import { reciprocalWorkflowRootRunId } from "../../shared/workflow-run-ownership.js"; /** Control-run ids visible below one workflow boundary, in graph order. */ export function expandedControlRunIds(store: Store, runId: string): string[] { - const graph = expandWorkflowGraph(store.snapshot(), runId); + const graph = expandWorkflowGraph(readGraphStoreSnapshot(store), runId); const ids = new Set([runId]); for (const stage of graph.stages) ids.add(stage.workflowGraphTarget.runId); // Tool-only nested runs own no stage, yet their in-flight ctx.tool nodes are diff --git a/packages/workflows/src/shared/graph-store-snapshot.ts b/packages/workflows/src/shared/graph-store-snapshot.ts new file mode 100644 index 000000000..e78b5ea49 --- /dev/null +++ b/packages/workflows/src/shared/graph-store-snapshot.ts @@ -0,0 +1,139 @@ +import type { + PendingPrompt, + RunSnapshot, + StageInputRequest, + StageSnapshot, + StoreSnapshot, + ToolEvent, + ToolNodeSnapshot, + WorkflowChildReplaySnapshot, + WorkflowNotice, +} from "./store-types.js"; +import type { WorkflowOutputValues } from "./types.js"; + +const COMPACT_RESULT_FIELD_LIMIT = 1024; + +function compactResultField(value: unknown): string | undefined { + if (typeof value !== "string") return undefined; + return value.length <= COMPACT_RESULT_FIELD_LIMIT ? value : value.slice(0, COMPACT_RESULT_FIELD_LIMIT); +} + +function compactRunResult(result: WorkflowOutputValues | undefined): WorkflowOutputValues | undefined { + if (result === undefined) return undefined; + const status = compactResultField(result.status); + const summary = compactResultField(result.summary); + const remainingWork = compactResultField(result.remaining_work); + const resultText = compactResultField(result.result); + const compact: WorkflowOutputValues = { + ...(status !== undefined ? { status } : {}), + ...(summary !== undefined ? { summary } : {}), + ...(remainingWork !== undefined ? { remaining_work: remainingWork } : {}), + ...(resultText !== undefined ? { result: resultText } : {}), + }; + return Object.keys(compact).length === 0 ? undefined : compact; +} + +function clonePrompt(prompt: PendingPrompt | undefined): PendingPrompt | undefined { + if (prompt === undefined) return undefined; + return { + ...prompt, + ...(prompt.choices !== undefined ? { choices: [...prompt.choices] } : {}), + }; +} + +function cloneInputRequest(request: StageInputRequest | undefined): StageInputRequest | undefined { + if (request === undefined) return undefined; + return { + ...request, + questions: request.questions.map((question) => ({ + ...question, + options: question.options.map((option) => ({ ...option })), + })), + }; +} + +function compactToolEvents(events: readonly ToolEvent[]): ToolEvent[] { + const event = events.at(-1); + return event === undefined ? [] : [{ name: event.name }]; +} + +function compactWorkflowChild(child: WorkflowChildReplaySnapshot): WorkflowChildReplaySnapshot { + return { + ...child, + outputs: {}, + outputCount: child.outputCount ?? Object.keys(child.outputs).length, + }; +} + +function compactStage(stage: StageSnapshot): StageSnapshot { + const { + toolEvents, + workflowChild, + workflowChildRun, + pendingPrompt, + promptFootprint, + inputRequest, + notices, + mcpScope: _mcpScope, + attemptedModels: _attemptedModels, + modelAttempts: _modelAttempts, + // Agent-stage results can be unbounded and graph cards do not render + // them. Durable tool cards use ToolNodeSnapshot.resultSummary instead. + result: _result, + ...metadata + } = stage; + return { + ...metadata, + parentIds: [...stage.parentIds], + toolEvents: compactToolEvents(toolEvents), + ...(workflowChild !== undefined ? { workflowChild: compactWorkflowChild(workflowChild) } : {}), + ...(workflowChildRun !== undefined ? { workflowChildRun: { ...workflowChildRun } } : {}), + ...(pendingPrompt !== undefined ? { pendingPrompt: clonePrompt(pendingPrompt) } : {}), + ...(promptFootprint !== undefined ? { promptFootprint: clonePrompt(promptFootprint) } : {}), + ...(inputRequest !== undefined ? { inputRequest: cloneInputRequest(inputRequest) } : {}), + ...(notices !== undefined ? { notices: notices.map((notice) => ({ ...notice })) } : {}), + }; +} + +function compactToolNode(node: ToolNodeSnapshot): ToolNodeSnapshot { + return { ...node, parentIds: [...node.parentIds] }; +} + +function compactRun(run: RunSnapshot): RunSnapshot { + const { inputs: _inputs, result: sourceResult, stages, toolNodes, pendingPrompt, ...metadata } = run; + const result = compactRunResult(sourceResult); + return { + ...metadata, + inputs: {}, + stages: stages.map(compactStage), + ...(toolNodes !== undefined ? { toolNodes: toolNodes.map(compactToolNode) } : {}), + ...(pendingPrompt !== undefined ? { pendingPrompt: clonePrompt(pendingPrompt) } : {}), + ...(result !== undefined ? { result } : {}), + }; +} + +export function createGraphStoreSnapshot( + runs: readonly RunSnapshot[], + notices: readonly WorkflowNotice[], + version: number, +): StoreSnapshot { + const snapshot: StoreSnapshot = { + runs: runs.map(compactRun), + notices: notices.map((notice) => ({ ...notice })), + version, + }; + deepFreezeGraphValue(snapshot); + return snapshot; +} + +function deepFreezeGraphValue(value: unknown): void { + if (value === null || typeof value !== "object" || Object.isFrozen(value)) return; + Object.freeze(value); + if (Array.isArray(value)) { + for (const item of value) deepFreezeGraphValue(item); + return; + } + for (const key of Object.keys(value as Record)) { + deepFreezeGraphValue((value as Record)[key]); + } +} diff --git a/packages/workflows/src/shared/store-internal.ts b/packages/workflows/src/shared/store-internal.ts index b5f85446e..fc48768e9 100644 --- a/packages/workflows/src/shared/store-internal.ts +++ b/packages/workflows/src/shared/store-internal.ts @@ -1,3 +1,4 @@ +import { createGraphStoreSnapshot } from "./graph-store-snapshot.js"; import type { PromptAnswerRecord, RunEndMetadata } from "./store-public-types.js"; import type { PendingPrompt, @@ -42,6 +43,7 @@ export interface StoreState { readonly runs: RunSnapshot[]; readonly notices: WorkflowNotice[]; readonly listeners: Set<(snap: StoreSnapshot) => void>; + readonly invalidationListeners: Set<() => void>; readonly resolvers: Map; readonly stagePromptAnswers: Map; readonly stagePromptDrafts: Map; @@ -51,6 +53,7 @@ export interface StoreState { export interface StoreContext { readonly state: StoreState; snapshot(): StoreSnapshot; + graphSnapshot(): StoreSnapshot; notify(): void; bumpAndNotify(): void; findRun(runId: string): RunSnapshot | undefined; @@ -128,6 +131,7 @@ export function createStoreState(): StoreState { runs: [], notices: [], listeners: new Set(), + invalidationListeners: new Set(), resolvers: new Map(), stagePromptAnswers: new Map(), stagePromptDrafts: new Map(), @@ -136,16 +140,28 @@ export function createStoreState(): StoreState { } export function createStoreContext(state: StoreState = createStoreState()): StoreContext { + let cachedGraphVersion = -1; + let cachedGraphSnapshot: StoreSnapshot | undefined; + function snapshot(): StoreSnapshot { return JSON.parse( JSON.stringify({ runs: state.runs, notices: state.notices, version: state.version }), ) as StoreSnapshot; } + function graphSnapshot(): StoreSnapshot { + if (cachedGraphSnapshot === undefined || cachedGraphVersion !== state.version) { + cachedGraphSnapshot = createGraphStoreSnapshot(state.runs, state.notices, state.version); + cachedGraphVersion = state.version; + } + return cachedGraphSnapshot; + } + function notify(): void { - const snap = snapshot(); - for (const fn of state.listeners) { - fn(snap); + for (const fn of state.invalidationListeners) fn(); + if (state.listeners.size > 0) { + const snap = snapshot(); + for (const fn of state.listeners) fn(snap); } } @@ -209,6 +225,7 @@ export function createStoreContext(state: StoreState = createStoreState()): Stor return { state, snapshot, + graphSnapshot, notify, bumpAndNotify, findRun, diff --git a/packages/workflows/src/shared/store-observation.ts b/packages/workflows/src/shared/store-observation.ts new file mode 100644 index 000000000..bd384d46e --- /dev/null +++ b/packages/workflows/src/shared/store-observation.ts @@ -0,0 +1,12 @@ +import type { Store } from "./store-public-types.js"; +import type { StoreSnapshot } from "./store-types.js"; + +export function readGraphStoreSnapshot(store: Store): StoreSnapshot { + return typeof store.graphSnapshot === "function" ? store.graphSnapshot() : store.snapshot(); +} + +export function subscribeStoreInvalidation(store: Store, listener: () => void): () => void { + return typeof store.subscribeInvalidation === "function" + ? store.subscribeInvalidation(listener) + : store.subscribe(listener); +} diff --git a/packages/workflows/src/shared/store-public-types.ts b/packages/workflows/src/shared/store-public-types.ts index 510374a99..ee00d0ee0 100644 --- a/packages/workflows/src/shared/store-public-types.ts +++ b/packages/workflows/src/shared/store-public-types.ts @@ -228,5 +228,9 @@ export interface Store { /** Drop every run and notice. */ clear(): void; snapshot(): StoreSnapshot; + /** Return one immutable payload-free graph projection per store version. */ + graphSnapshot?(): StoreSnapshot; subscribe(fn: (snap: StoreSnapshot) => void): () => void; + /** Subscribe synchronously to store invalidation without constructing a full snapshot. */ + subscribeInvalidation?(fn: () => void): () => void; } diff --git a/packages/workflows/src/shared/store-run-methods.ts b/packages/workflows/src/shared/store-run-methods.ts index 94ca797ca..123bebbb0 100644 --- a/packages/workflows/src/shared/store-run-methods.ts +++ b/packages/workflows/src/shared/store-run-methods.ts @@ -27,7 +27,9 @@ type RunStoreMethods = Pick< | "recordRunResumed" | "clear" | "snapshot" + | "graphSnapshot" | "subscribe" + | "subscribeInvalidation" >; export function createRunStoreMethods(context: StoreContext): RunStoreMethods { @@ -233,12 +235,21 @@ export function createRunStoreMethods(context: StoreContext): RunStoreMethods { return context.snapshot(); }, + graphSnapshot(): StoreSnapshot { + return context.graphSnapshot(); + }, + subscribe(fn: (snap: StoreSnapshot) => void): () => void { state.listeners.add(fn); return () => { state.listeners.delete(fn); }; }, + + subscribeInvalidation(fn: () => void): () => void { + state.invalidationListeners.add(fn); + return () => state.invalidationListeners.delete(fn); + }, }; } diff --git a/packages/workflows/src/shared/store-types.ts b/packages/workflows/src/shared/store-types.ts index 92a3228b2..4a41500fb 100644 --- a/packages/workflows/src/shared/store-types.ts +++ b/packages/workflows/src/shared/store-types.ts @@ -153,6 +153,8 @@ export interface WorkflowChildReplaySnapshot { /** True when the child reached this terminal status through ctx.exit(). */ readonly exited?: boolean; readonly outputs: WorkflowOutputValues; + /** Payload-free output count used by compact graph projections. */ + readonly outputCount?: number; readonly exitReason?: string; } diff --git a/packages/workflows/src/tui/graph-view-state.ts b/packages/workflows/src/tui/graph-view-state.ts index 0c7a0a9d8..a65dfab6f 100644 --- a/packages/workflows/src/tui/graph-view-state.ts +++ b/packages/workflows/src/tui/graph-view-state.ts @@ -9,6 +9,7 @@ import { sameExpandedWorkflowTopology, } from "../shared/expanded-workflow-graph.js"; import type { Store } from "../shared/store.js"; +import { readGraphStoreSnapshot, subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { PendingPrompt, RunSnapshot, StageSnapshot, StoreSnapshot } from "../shared/store-types.js"; import type { GraphTheme } from "./graph-theme.js"; import { ANIMATION_TICK_MS } from "./graph-view-constants.js"; @@ -139,11 +140,11 @@ export abstract class GraphViewState { this.footerData = opts.footerData; this.getStageQueuedMessageCount = opts.getStageQueuedMessageCount; - this._unsubscribe = this.store.subscribe((snap) => { - this.currentSnapshot = snap; + this._unsubscribe = subscribeStoreInvalidation(this.store, () => { + this.currentSnapshot = readGraphStoreSnapshot(this.store); this._rebuildLayout(); }); - this.currentSnapshot = this.store.snapshot(); + this.currentSnapshot = readGraphStoreSnapshot(this.store); this._rebuildLayout(); // Animation tick: while the overlay is mounted, fire a render diff --git a/packages/workflows/src/tui/node-card.ts b/packages/workflows/src/tui/node-card.ts index 9abf9c39b..d372a8a0e 100644 --- a/packages/workflows/src/tui/node-card.ts +++ b/packages/workflows/src/tui/node-card.ts @@ -144,7 +144,7 @@ function workflowChildSummaryText(stage: StageSnapshot): string { function workflowChildMetaText(stage: StageSnapshot): string { const completed = stage.workflowChild; if (completed !== undefined) { - const outputCount = Object.keys(completed.outputs).length; + const outputCount = completed.outputCount ?? Object.keys(completed.outputs).length; const outputs = outputCount === 1 ? "1 out" : `${outputCount} outs`; return `run ${shortRunId(completed.runId)} · ${outputs}`; } diff --git a/packages/workflows/src/tui/overlay-adapter.ts b/packages/workflows/src/tui/overlay-adapter.ts index 83bf43c7c..ba31886fb 100644 --- a/packages/workflows/src/tui/overlay-adapter.ts +++ b/packages/workflows/src/tui/overlay-adapter.ts @@ -34,6 +34,7 @@ import type { StageControlRegistry } from "../runs/foreground/stage-control-regi import { stageControlRegistry as defaultStageControlRegistry } from "../runs/foreground/stage-control-registry.js"; import type { StageUiBroker } from "../shared/stage-ui-broker.js"; import type { Store } from "../shared/store.js"; +import { readGraphStoreSnapshot, subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { StoreSnapshot } from "../shared/store-types.js"; import { deriveGraphThemeFromPiTheme } from "./graph-theme.js"; import type { OverlayTerminalOutput } from "./overlay-terminal-modes.js"; @@ -274,7 +275,8 @@ export function buildGraphOverlayAdapter( function makeComponent(view: WorkflowAttachPane, tui: PiCustomOverlayFactoryTui): PiCustomComponent { requestMountedRender = () => tui.requestRender?.(); - const onStoreUpdate = (snapshot: StoreSnapshot): void => { + const onStoreUpdate = (): void => { + const snapshot = readGraphStoreSnapshot(store); // Always invalidate retained view state so a later reopen renders the // current snapshot — but while the overlay is hidden, never ask the // host to render (#1856): each hidden-overlay render request became @@ -285,7 +287,7 @@ export function buildGraphOverlayAdapter( refocusVisibleOverlayForAwaitingInput(snapshot); tui.requestRender?.(); }; - const unsubscribe = store.subscribe(onStoreUpdate); + const unsubscribe = subscribeStoreInvalidation(store, onStoreUpdate); return { render: (width: number) => view.render(width), handleInput: (data: string) => { diff --git a/packages/workflows/src/tui/session-overlays.ts b/packages/workflows/src/tui/session-overlays.ts index 5a5742fb0..12f3a9d79 100644 --- a/packages/workflows/src/tui/session-overlays.ts +++ b/packages/workflows/src/tui/session-overlays.ts @@ -25,6 +25,7 @@ import type { PiCustomComponent, PiCustomOverlayFactoryTui, PiCustomOverlayFunction } from "../extension/wiring.js"; import type { Store } from "../shared/store.js"; +import { subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { GraphTheme } from "./graph-theme.js"; import { createSessionPickerState, @@ -96,7 +97,7 @@ export function openSessionPicker( }; // Re-render on store changes so newly-started runs appear and // status icons refresh without the user having to press a key. - unsubscribe = store.subscribe(() => tui.requestRender?.()); + unsubscribe = subscribeStoreInvalidation(store, () => tui.requestRender?.()); return { render: (width: number) => { const rows = selectRunsForPicker(store.runs(), state.query, state.includeAll); diff --git a/packages/workflows/src/tui/stage-chat-view-state.ts b/packages/workflows/src/tui/stage-chat-view-state.ts index 8a5c22ab0..739f34158 100644 --- a/packages/workflows/src/tui/stage-chat-view-state.ts +++ b/packages/workflows/src/tui/stage-chat-view-state.ts @@ -1,6 +1,7 @@ import { type AgentSessionEvent, ChatSessionHost } from "@bastani/atomic"; import { Editor, type EditorComponent } from "@earendil-works/pi-tui"; import { stageUiBroker } from "../shared/stage-ui-broker.js"; +import { readGraphStoreSnapshot, subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { PendingPrompt, RunSnapshot, StageSnapshot } from "../shared/store-types.js"; import { hexToAnsi, RESET } from "./color-utils.js"; import { createPromptCardState } from "./prompt-card.js"; @@ -104,7 +105,7 @@ export function initializeStageChatView(ctx: StageChatViewContext, opts: StageCh const initialChatIsTerminal = isTerminalStageChatState(initialRun?.status) || isTerminalStageChatState(initialStage?.status); if (initialChatIsTerminal) ctx.chatHost.clearBusyForTerminalWorkflowStage(); - ctx._unsubscribeStore = ctx.store.subscribe(() => handleStoreUpdate(ctx)); + ctx._unsubscribeStore = subscribeStoreInvalidation(ctx.store, () => handleStoreUpdate(ctx)); ctx._unsubscribeFooterData = ctx.footerData?.onBranchChange(() => ctx.requestRender?.()) ?? null; if (ctx.handle) { @@ -345,7 +346,7 @@ function absorbStageNotices(ctx: StageChatViewContext, stage: StageSnapshot | un } export function currentRun(ctx: StageChatViewContext): RunSnapshot | undefined { - return ctx.store.snapshot().runs.find((r) => r.id === ctx.runId); + return readGraphStoreSnapshot(ctx.store).runs.find((r) => r.id === ctx.runId); } export function currentStage(ctx: StageChatViewContext): StageSnapshot | undefined { diff --git a/packages/workflows/src/tui/store-widget-installer.ts b/packages/workflows/src/tui/store-widget-installer.ts index 55eca4e2c..80e74ac22 100644 --- a/packages/workflows/src/tui/store-widget-installer.ts +++ b/packages/workflows/src/tui/store-widget-installer.ts @@ -45,6 +45,7 @@ import { type ReactiveWidgetTimerHandle, } from "@bastani/atomic"; import type { Store } from "../shared/store.js"; +import { readGraphStoreSnapshot, subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { StoreSnapshot } from "../shared/store-types.js"; import { buildThemedWidgetLines, nextWidgetRefreshDelayMs } from "./widget.js"; @@ -106,8 +107,8 @@ export function installStoreWidget( key: WIDGET_KEY, placement: "belowEditor", timers, - getSnapshot: () => storeInstance.snapshot(), - subscribe: (listener) => storeInstance.subscribe(() => listener()), + getSnapshot: () => readGraphStoreSnapshot(storeInstance), + subscribe: (listener) => subscribeStoreInvalidation(storeInstance, listener), getPreviewLines: (snap, now) => buildThemedWidgetLines(snap, undefined, 120, now), render: (snap, { theme, width, now }) => buildThemedWidgetLines(snap, theme as PiTheme | undefined, width, now), getNextRefreshDelayMs: (snap, now) => nextWidgetRefreshDelayMs(snap, now), @@ -206,12 +207,12 @@ export function installToolExecutionHooks(pi: LiveWidgetAPI, storeInstance: Stor return call ? { key, call } : null; } - storeInstance.subscribe(pruneActiveToolCalls); + subscribeStoreInvalidation(storeInstance, () => pruneActiveToolCalls(readGraphStoreSnapshot(storeInstance))); function recordToolStart(payload: unknown): void { if (!isToolExecutionPayload(payload)) return; - const snap = storeInstance.snapshot(); + const snap = readGraphStoreSnapshot(storeInstance); pruneActiveToolCalls(snap); const scope = resolveExplicitStageScope(payload); @@ -234,7 +235,7 @@ export function installToolExecutionHooks(pi: LiveWidgetAPI, storeInstance: Stor function recordToolUpdate(payload: unknown): void { if (!isToolExecutionPayload(payload)) return; - pruneActiveToolCalls(storeInstance.snapshot()); + pruneActiveToolCalls(readGraphStoreSnapshot(storeInstance)); if (!activeToolCallForPayload(payload)) return; // Updates are attach-only until the store has an explicit update API. @@ -243,7 +244,7 @@ export function installToolExecutionHooks(pi: LiveWidgetAPI, storeInstance: Stor function recordToolEnd(payload: unknown): void { if (!isToolExecutionPayload(payload)) return; - pruneActiveToolCalls(storeInstance.snapshot()); + pruneActiveToolCalls(readGraphStoreSnapshot(storeInstance)); const active = activeToolCallForPayload(payload); if (!active) return; diff --git a/packages/workflows/src/tui/workflow-attach-pane.ts b/packages/workflows/src/tui/workflow-attach-pane.ts index 6e34b6477..fe02ec26f 100644 --- a/packages/workflows/src/tui/workflow-attach-pane.ts +++ b/packages/workflows/src/tui/workflow-attach-pane.ts @@ -15,6 +15,7 @@ import { stageQueuedUserMessageCount } from "../runs/foreground/stage-queued-use import { expandWorkflowGraph } from "../shared/expanded-workflow-graph.js"; import type { StageUiBroker } from "../shared/stage-ui-broker.js"; import type { Store } from "../shared/store.js"; +import { readGraphStoreSnapshot, subscribeStoreInvalidation } from "../shared/store-observation.js"; import type { StageSnapshot, StoreSnapshot } from "../shared/store-types.js"; import type { GraphTheme } from "./graph-theme.js"; import { GraphView } from "./graph-view.js"; @@ -86,7 +87,9 @@ export class WorkflowAttachPane implements Component { this.setToolsExpanded = opts.setToolsExpanded; this.footerData = opts.footerData; this.now = opts.now ?? Date.now; - this.unsubscribeStore = this.store.subscribe((snapshot) => this._handleStoreUpdate(snapshot)); + this.unsubscribeStore = subscribeStoreInvalidation(this.store, () => + this._handleStoreUpdate(readGraphStoreSnapshot(this.store)), + ); this.graphView = this._buildGraphView(); const target = opts.initialAttachStageId !== undefined && this.runId @@ -97,7 +100,7 @@ export class WorkflowAttachPane implements Component { if (target) { this._attachToStage(target.runId, target.stageId); } else { - this._syncAwaitingInputKeys(this.store.snapshot()); + this._syncAwaitingInputKeys(readGraphStoreSnapshot(this.store)); this._armGraphEnterQuarantineIfRunNeedsInput(); this._syncMouseScrollTracking(); } @@ -150,7 +153,7 @@ export class WorkflowAttachPane implements Component { return null; } private _workflowName(runId: string): string { - const snap = this.store.snapshot(); + const snap = readGraphStoreSnapshot(this.store); const run = snap.runs.find((r) => r.id === runId); return run?.name ?? "workflow"; } @@ -160,7 +163,7 @@ export class WorkflowAttachPane implements Component { options: { suppressInitialPromptSubmit?: boolean } = { suppressInitialPromptSubmit: true }, ): void { this.graphEnterQuarantineUntil = 0; - const snapshot = this.store.snapshot(); + const snapshot = readGraphStoreSnapshot(this.store); const graphRunId = this._resolveRunId(); this.lastGraphAwaitingInputKey = graphRunId ? this._runAwaitingInputKey(snapshot, graphRunId) : null; this.lastStageAwaitingInputKey = this._stageAwaitingInputKey(snapshot, runId, stageId); @@ -227,7 +230,9 @@ export class WorkflowAttachPane implements Component { this.mode = "graph"; this.stagePromptEnterQuarantineUntil = 0; this.lastStageAwaitingInputKey = null; - this.lastGraphAwaitingInputKey = this.runId ? this._runAwaitingInputKey(this.store.snapshot(), this.runId) : null; + this.lastGraphAwaitingInputKey = this.runId + ? this._runAwaitingInputKey(readGraphStoreSnapshot(this.store), this.runId) + : null; this.graphEnterQuarantineUntil = reason === "prompt-resolved" && metadata.suppressNextGraphSubmit === true ? this.now() + ENTER_TRANSITION_QUARANTINE_MS @@ -248,7 +253,7 @@ export class WorkflowAttachPane implements Component { this.graphEnterQuarantineUntil = 0; this.stagePromptEnterQuarantineUntil = 0; this.graphView = this._buildGraphView(); - this._syncAwaitingInputKeys(this.store.snapshot()); + this._syncAwaitingInputKeys(readGraphStoreSnapshot(this.store)); if (stageId !== undefined && runId) { const target = stageRunId === undefined ? this._resolveGraphStageTarget(runId, stageId) : { runId: stageRunId, stageId }; @@ -264,7 +269,7 @@ export class WorkflowAttachPane implements Component { rootRunId: string, stageId: string, ): { runId: string; stageId: string } | undefined { - const graph = expandWorkflowGraph(this.store.snapshot(), rootRunId); + const graph = expandWorkflowGraph(readGraphStoreSnapshot(this.store), rootRunId); const exact = graph.stages.find((stage) => stage.id === stageId); const localMatches = graph.stages.filter((stage) => stage.workflowGraphTarget.stageId === stageId); if (exact === undefined && localMatches.length > 1) return undefined; @@ -403,7 +408,7 @@ export class WorkflowAttachPane implements Component { runId && this._runNeedsInput(runId) ? this.now() + ENTER_TRANSITION_QUARANTINE_MS : 0; } private _runNeedsInput(runId: string): boolean { - return this._runAwaitingInputKey(this.store.snapshot(), runId) !== null; + return this._runAwaitingInputKey(readGraphStoreSnapshot(this.store), runId) !== null; } private _runAwaitingInputKey(snapshot: StoreSnapshot, runId: string): string | null { const run = snapshot.runs.find((candidate) => candidate.id === runId); @@ -422,7 +427,7 @@ export class WorkflowAttachPane implements Component { return keys[0]!.key; } private _stageNeedsInput(runId: string, stageId: string): boolean { - return this._stageAwaitingInputKey(this.store.snapshot(), runId, stageId) !== null; + return this._stageAwaitingInputKey(readGraphStoreSnapshot(this.store), runId, stageId) !== null; } private _stageAwaitingInputKey(snapshot: StoreSnapshot, runId: string, stageId: string): string | null { const run = snapshot.runs.find((candidate) => candidate.id === runId); @@ -448,7 +453,7 @@ export class WorkflowAttachPane implements Component { return this._stageSnapshot(runId, stageId)?.attached === true; } private _stageSnapshot(runId: string, stageId: string): StageSnapshot | undefined { - const run = this.store.snapshot().runs.find((candidate) => candidate.id === runId); + const run = readGraphStoreSnapshot(this.store).runs.find((candidate) => candidate.id === runId); return run?.stages.find((candidate) => candidate.id === stageId); } private _stageSnapshotNeedsInput(stage: Pick): boolean { diff --git a/test/integration/overlay-entrypoints-animation.test.ts b/test/integration/overlay-entrypoints-animation.test.ts index 54713646b..a4024ac26 100644 --- a/test/integration/overlay-entrypoints-animation.test.ts +++ b/test/integration/overlay-entrypoints-animation.test.ts @@ -60,18 +60,30 @@ void [ waitForStagePendingPrompt, ]; +function setupAnimatingRun(store: ReturnType, runId: string): void { + store.recordRunStart({ + id: runId, + name: "wf", + inputs: {}, + status: "running", + stages: [], + startedAt: Date.now(), + }); + store.recordStageStart(runId, { + id: "running-stage", + name: "running-stage", + status: "running", + parentIds: [], + toolEvents: [], + startedAt: Date.now(), + }); +} + describe("buildGraphOverlayAdapter — animation tick visibility gating", () => { test("requestRender from the view fires tui.requestRender while visible", async () => { const runId = `tick-visible-${Date.now()}`; const store = createStore(); - store.recordRunStart({ - id: runId, - name: "wf", - inputs: {}, - status: "running", - stages: [], - startedAt: Date.now(), - }); + setupAnimatingRun(store, runId); let renderCalls = 0; let component: PiCustomComponent | undefined; @@ -106,14 +118,7 @@ describe("buildGraphOverlayAdapter — animation tick visibility gating", () => test("requestRender suppresses tui.requestRender while overlay is hidden", async () => { const runId = `tick-hidden-${Date.now()}`; const store = createStore(); - store.recordRunStart({ - id: runId, - name: "wf", - inputs: {}, - status: "running", - stages: [], - startedAt: Date.now(), - }); + setupAnimatingRun(store, runId); let renderCalls = 0; let component: PiCustomComponent | undefined; @@ -159,14 +164,7 @@ describe("buildGraphOverlayAdapter — animation tick visibility gating", () => test("tick stops after the component is disposed", async () => { const runId = `tick-dispose-${Date.now()}`; const store = createStore(); - store.recordRunStart({ - id: runId, - name: "wf", - inputs: {}, - status: "running", - stages: [], - startedAt: Date.now(), - }); + setupAnimatingRun(store, runId); let renderCalls = 0; let component: PiCustomComponent | undefined; diff --git a/test/unit/store-payload-observation.test.ts b/test/unit/store-payload-observation.test.ts new file mode 100644 index 000000000..8b71eb5d7 --- /dev/null +++ b/test/unit/store-payload-observation.test.ts @@ -0,0 +1,113 @@ +import assert from "node:assert/strict"; +import { describe, test } from "vitest"; +import { expandWorkflowGraph } from "../../packages/workflows/src/shared/expanded-workflow-graph.js"; +import { createStore } from "../../packages/workflows/src/shared/store.js"; + +describe("payload-free store observation", () => { + test("notifies synchronously without traversing unrelated workflow inputs", () => { + const store = createStore(); + let payloadReads = 0; + const evidence: Record = {}; + Object.defineProperty(evidence, "occurrences", { + enumerable: true, + get() { + payloadReads++; + return "large-unrelated-payload"; + }, + }); + store.recordRunStart({ + id: "run-large", + name: "large-workflow", + inputs: { evidence }, + status: "running", + stages: [ + { + id: "question", + name: "question", + status: "running", + parentIds: [], + toolEvents: [], + }, + ], + startedAt: Date.now(), + }); + payloadReads = 0; + let calls = 0; + const unsubscribe = store.subscribeInvalidation?.(() => { + calls++; + store.graphSnapshot?.(); + }); + assert.ok(unsubscribe); + + assert.equal( + store.recordStageInputRequest("run-large", "question", { + id: "question-1", + kind: "ask_user_question", + questions: [{ question: "Continue?", options: [] }], + createdAt: Date.now(), + }), + true, + ); + + assert.equal(calls, 1); + assert.equal(payloadReads, 0); + assert.deepEqual(store.graphSnapshot?.().runs[0]?.inputs, {}); + unsubscribe(); + }); + + test("preserves legacy full-snapshot subscribers", () => { + const store = createStore(); + const versions: number[] = []; + store.subscribe((snapshot) => versions.push(snapshot.version)); + store.recordRunStart({ + id: "run-legacy", + name: "legacy", + inputs: {}, + status: "running", + stages: [], + startedAt: Date.now(), + }); + assert.deepEqual(versions, [1]); + }); + + test("excludes authored stage results but preserves durable tool summaries", () => { + const store = createStore(); + store.recordRunStart({ + id: "run-results", + name: "results", + inputs: {}, + status: "running", + stages: [ + { + id: "agent-stage", + name: "agent-stage", + status: "completed", + parentIds: [], + result: "unbounded agent output", + toolEvents: [], + }, + ], + toolNodes: [ + { + kind: "tool", + id: "tool-node", + name: "verify", + argsHash: "hash", + ordinal: 1, + parentIds: ["agent-stage"], + status: "completed", + resultSummary: "bounded tool summary", + attachable: false, + }, + ], + startedAt: Date.now(), + }); + + const snapshot = store.graphSnapshot?.(); + assert.ok(snapshot); + assert.equal(snapshot.runs[0]?.stages[0]?.result, undefined); + assert.equal(snapshot.runs[0]?.toolNodes?.[0]?.resultSummary, "bounded tool summary"); + const graph = expandWorkflowGraph(snapshot, "run-results"); + assert.equal(graph.renderStages.find((stage) => stage.nodeKind === "tool")?.result, "bounded tool summary"); + }); +}); diff --git a/test/unit/workflow-large-payload-interaction.test.ts b/test/unit/workflow-large-payload-interaction.test.ts new file mode 100644 index 000000000..f964d520d --- /dev/null +++ b/test/unit/workflow-large-payload-interaction.test.ts @@ -0,0 +1,161 @@ +import assert from "node:assert/strict"; +import type { AgentSession } from "@bastani/atomic"; +import { describe, test } from "vitest"; +import type { StageControlHandle } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import { createStageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import { createStore } from "../../packages/workflows/src/shared/store.js"; +import type { WorkflowInputValues } from "../../packages/workflows/src/shared/types.js"; +import { deriveGraphTheme } from "../../packages/workflows/src/tui/graph-theme.js"; +import { GraphView } from "../../packages/workflows/src/tui/graph-view.js"; +import { WorkflowAttachPane } from "../../packages/workflows/src/tui/workflow-attach-pane.js"; +import { ANSI_RE } from "./overlay-graph-helpers.js"; + +function heavyInput(reads: { count: number }): WorkflowInputValues { + const evidence: Record = {}; + Object.defineProperty(evidence, "occurrences", { + enumerable: true, + get() { + reads.count++; + return "x".repeat(1024); + }, + }); + return { evidence }; +} + +function countFullSnapshots(store: ReturnType): { count: number } { + const counter = { count: 0 }; + const snapshot = store.snapshot.bind(store); + store.snapshot = () => { + counter.count++; + return snapshot(); + }; + return counter; +} + +function mousePressForText(lines: readonly string[], text: string): string { + const plain = lines.map((line) => line.replace(ANSI_RE, "")); + const row = plain.findIndex((line) => line.includes(text)); + assert.notEqual(row, -1, `expected rendered text ${text}`); + const column = plain[row]!.indexOf(text); + return `\x1b[<0;${column + 1};${row + 1}M`; +} + +function handle(runId: string, stageId: string): StageControlHandle { + return { + runId, + stageId, + stageName: stageId, + status: "running", + sessionId: undefined, + sessionFile: undefined, + isStreaming: false, + messages: [] as AgentSession["messages"], + async ensureAttached() {}, + async prompt() {}, + async steer() {}, + async followUp() {}, + async pause() {}, + async resume() {}, + subscribe() { + return () => {}; + }, + }; +} + +describe("large workflow interaction isolation", () => { + test("graph render and mouse attachment avoid full snapshots and payload traversal", () => { + const payloadReads = { count: 0 }; + const store = createStore(); + store.recordRunStart({ + id: "run-large", + name: "large-workflow", + inputs: heavyInput(payloadReads), + status: "running", + stages: [ + { + id: "target-stage", + name: "target-stage", + status: "running", + parentIds: [], + toolEvents: [], + startedAt: Date.now(), + }, + ], + startedAt: Date.now(), + }); + payloadReads.count = 0; + const fullSnapshots = countFullSnapshots(store); + const attached: string[] = []; + const view = new GraphView({ + mode: "overlay", + runId: "run-large", + store, + graphTheme: deriveGraphTheme({}), + onStageAttach: (_runId, stageId) => attached.push(stageId), + }); + + const lines = view.render(96); + assert.equal(view.handleInput(mousePressForText(lines, "target-stage")), true); + assert.deepEqual(attached, ["target-stage"]); + assert.equal(fullSnapshots.count, 0); + assert.equal(payloadReads.count, 0); + view.dispose(); + }); + + test("question preview, input, and submission avoid full snapshots and payload traversal", async () => { + const payloadReads = { count: 0 }; + const store = createStore(); + store.recordRunStart({ + id: "run-large", + name: "large-workflow", + inputs: heavyInput(payloadReads), + status: "running", + stages: [ + { + id: "question-stage", + name: "question-stage", + status: "running", + parentIds: [], + toolEvents: [], + startedAt: Date.now(), + }, + ], + startedAt: Date.now(), + }); + assert.equal( + store.recordStagePendingPrompt("run-large", "question-stage", { + id: "prompt-1", + kind: "input", + message: "What should the workflow use?", + createdAt: Date.now(), + }), + true, + ); + const pending = store.awaitStagePendingPrompt("run-large", "question-stage", "prompt-1"); + payloadReads.count = 0; + const fullSnapshots = countFullSnapshots(store); + const registry = createStageControlRegistry(); + registry.register(handle("run-large", "question-stage")); + let now = 1_000; + const pane = new WorkflowAttachPane({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-large", + initialAttachStageId: "question-stage", + stageControlRegistry: registry, + onClose: () => {}, + now: () => now, + }); + + const preview = pane.render(96).join("\n"); + assert.match(preview, /What should the workflow use\?/); + for (const character of "answer") pane.handleInput(character); + now += 201; + pane.handleInput("\r"); + + assert.equal(await pending, "answer"); + assert.equal(fullSnapshots.count, 0); + assert.equal(payloadReads.count, 0); + pane.dispose(); + }); +});