From bfc568c1b5bc8f02df6644e34591d5017503d1df Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Tue, 28 Jul 2026 14:53:20 -0700 Subject: [PATCH 1/3] fix(workflows): keep retained follow-up Working visible An attached workflow-stage chat sat idle while a workflow-authored idle follow-up was already running: nothing told the chat a delivery had been admitted until the public agent_start cleared the session's ordered extension event queue. Report accepted idle-delivery start and settlement on a workflow-internal channel and map it onto the chat's ordinary Working lifecycle. The channel replays active deliveries to late subscribers, so attaching or remounting mid-delivery paints immediately. A pre-turn compaction_end reasserts the lifecycle for a still-active delivery while preserving factual status text, and a terminal transition drops its pre-terminal leases so a leftover agent_start or turn_start cannot restart Working on a finished stage. Live-turn followUp/steer queueing, controlled pause, terminal-root rejection, post-mortem chat, and retry/fallback/compaction/error status precedence are unchanged. Assistant-model: Claude Opus 5 --- packages/coding-agent/CHANGELOG.md | 4 + packages/coding-agent/docs/workflows.md | 2 +- .../components/chat-session-host-actions.ts | 19 +- .../components/chat-session-host-runtime.ts | 68 ++++ .../components/chat-session-host.ts | 27 ++ .../runs/foreground/executor-stage-control.ts | 3 + .../runs/foreground/executor-stage-replay.ts | 1 + .../runs/foreground/postmortem-stage-chat.ts | 1 + .../runs/foreground/stage-control-registry.ts | 10 + .../foreground/stage-delivery-activity.ts | 76 ++++ .../runs/foreground/stage-runner-context.ts | 4 + .../foreground/stage-runner-controller.ts | 8 +- .../stage-runner-send-user-message.ts | 8 + .../src/runs/foreground/stage-runner-types.ts | 9 + .../tui/stage-chat-view-delivery-activity.ts | 86 +++++ .../src/tui/stage-chat-view-live-events.ts | 42 +++ .../src/tui/stage-chat-view-state.ts | 16 + .../src/tui/stage-chat-view-types.ts | 4 + ...orkflow-followup-working-lifecycle.test.ts | 337 ++++++++++++++++++ .../stage-chat-view-delivery-activity.test.ts | 334 +++++++++++++++++ test/unit/stage-chat-view-helpers.ts | 32 +- test/unit/stage-delivery-activity.test.ts | 244 +++++++++++++ 22 files changed, 1314 insertions(+), 21 deletions(-) create mode 100644 packages/workflows/src/runs/foreground/stage-delivery-activity.ts create mode 100644 packages/workflows/src/tui/stage-chat-view-delivery-activity.ts create mode 100644 test/integration/workflow-followup-working-lifecycle.test.ts create mode 100644 test/unit/stage-chat-view-delivery-activity.test.ts create mode 100644 test/unit/stage-delivery-activity.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index bdecf9fe9..c6dfa7228 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed attached workflow-stage chats missing Working during workflow-authored idle follow-ups, including chats attached or remounted while the turn is already starting. Working now persists through prompt startup and pre-turn compaction, while live-turn `followUp`/`steer` queueing and factual retry, fallback, compaction, cancellation, and error status precedence remain intact; terminal cleanup keeps a leftover start from restarting it. + ## [0.9.11-alpha.6] - 2026-07-28 ### Added diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index 6e96435bb..8b7399637 100644 --- a/packages/coding-agent/docs/workflows.md +++ b/packages/coding-agent/docs/workflows.md @@ -2819,7 +2819,7 @@ Surface behavior: - **Wheel and trackpad** - While the workflow graph is active, vertical wheel/trackpad gestures pan it up and down, and horizontal gestures pan wide graphs left and right when the terminal exposes horizontal wheel events; these gestures remain scoped to the graph instead of leaking into 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. - **Tool and node detail** - Attached stage chats match main chat's tool-detail expansion behavior while keeping expansion state local to the workflow UI context. Press Ctrl+O (the configurable `app.tools.expand` binding) to expand every visible workflow node and tool card, including single, parallel, and chain subagent progress, current tool activity, and artifact paths; press it again to collapse them. The toggle works for active, completed, and archived stage views, including at the supported 40-column terminal minimum. A mounted prompt, custom question, or other input-owning overlay keeps the key instead of changing expansion. - **Footer context** - An attached live stage chat carries the main chat's current-folder and Git-branch identity into its themed footer and mirrors live extension status lines such as the MCP server indicator. Branch changes trigger a repaint through the host's cached footer provider, and extension status changes are read from that same provider rather than recomputed by the workflow UI. -- **Working animation lifecycle** - Ordinary attached-stage work keeps the same exact one-cell `∀` visible while following the active workflow theme's dark → accent → bright/bold → accent → dark luminance ramp every 88ms. Every agent and SDK turn resets to the dark regular phase with a fresh lifecycle-relative cadence; turn, terminal, error, replacement, and disposal cleanup stop the active timer without stale repaint. In an eligible retained terminal-stage chat, an accepted follow-up shows Working immediately while Atomic attaches the retained conversation and starts the prompt, then hands the same visible status to the agent turn without a gap. A no-turn result, prompt error, or terminal completion removes it; an accepted manual retry clears stale status from the prior prompt before showing new pre-stream activity. `NO_COLOR` retains regular/bold activity without foreground-color escapes. Reduced motion uses a static regular accent `∀` without an animation timer; factual automatic retry, fallback, compaction, cancellation, and error copy retains precedence. +- **Working animation lifecycle** - Ordinary attached-stage work keeps the same exact one-cell `∀` visible while following the active workflow theme's dark → accent → bright/bold → accent → dark luminance ramp every 88ms. Every agent and SDK turn resets to the dark regular phase with a fresh lifecycle-relative cadence; turn, terminal, error, replacement, and disposal cleanup stop the active timer without stale repaint. In an eligible retained-stage chat, every accepted idle follow-up — including a workflow-authored `stage.sendUserMessage(...)` after a prior turn ended — shows Working on admission or on attach, and keeps it through pre-stream startup, pre-turn compaction, and the handoff to the agent turn. Attaching or remounting mid-delivery paints immediately rather than waiting for the turn's first event. A message queued into a live turn with `followUp`/`steer` uses that turn's existing status instead of starting a new one. A no-turn result, prompt error, or terminal completion removes it, and a start left over from work that terminal cleanup already finished cannot bring it back; an accepted manual retry clears stale status from the prior prompt before showing new pre-stream activity. `NO_COLOR` retains regular/bold activity without foreground-color escapes. Reduced motion uses a static regular accent `∀` without an animation timer; factual automatic retry, fallback, compaction, cancellation, and error copy retains precedence. - **Async statusline** - If an async/background subagent is running while the fullscreen workflow graph is open, the graph statusline mirrors the async summary so the background run remains visible; hide the graph with `h`, leave it with `ctrl+x`, or reconnect later to return to the full below-editor async widget. - **Copy mode** - Press `ctrl+t` inside an attached stage chat 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 transcript or prompt scrolling. Archived read-only stage transcripts expose the same footer and copy-mode status, so their text can also be selected and copied; `esc` closes the transcript and `ctrl+x` returns to the graph. 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. - **Run control** - Use `interrupt`, `pause`, and `resume` for resumable live work. Pause/interrupt holds a stage's queued steering and follow-up items in place without dequeuing them or starting continuation; `resume` releases those items once in their existing per-queue order, but queue release alone does not start a model turn. `resume` on a non-paused run reopens the saved snapshot or overlay. Use `quit` to pause a live run gracefully while preserving it for `/workflow resume`. diff --git a/packages/coding-agent/src/modes/interactive/components/chat-session-host-actions.ts b/packages/coding-agent/src/modes/interactive/components/chat-session-host-actions.ts index 1e88ae408..15355da0c 100644 --- a/packages/coding-agent/src/modes/interactive/components/chat-session-host-actions.ts +++ b/packages/coding-agent/src/modes/interactive/components/chat-session-host-actions.ts @@ -12,6 +12,7 @@ import { notifyChatSessionStatus, notifyChatSessionWarning, requiredChatSessionCommand, + settleChatSessionPromptLifecycle, startChatSessionWorkingLifecycle, stopChatSessionWorkingLifecycle, syncChatSessionAnimationTick, @@ -202,23 +203,7 @@ export async function submitChatSession( - state: ChatSessionHostState, - submittedGeneration: number | undefined, -): void { - if (submittedGeneration === undefined) { - state.sdkBusy = false; - return; - } - const lifecycleWasReplaced = state.workingLifecycleGeneration !== submittedGeneration; - if (lifecycleWasReplaced && state.workingLifecycleActive) return; - state.sdkBusy = false; - if (!lifecycleWasReplaced && !isChatSessionStreaming(state)) { - stopChatSessionWorkingLifecycle(state); - } -} +const settleSubmittedPromptLifecycle = settleChatSessionPromptLifecycle; export async function abortChatSessionCompaction< TExtraEntry extends ChatTranscriptEntryLike, diff --git a/packages/coding-agent/src/modes/interactive/components/chat-session-host-runtime.ts b/packages/coding-agent/src/modes/interactive/components/chat-session-host-runtime.ts index 5f95c7fcd..98dae4e21 100644 --- a/packages/coding-agent/src/modes/interactive/components/chat-session-host-runtime.ts +++ b/packages/coding-agent/src/modes/interactive/components/chat-session-host-runtime.ts @@ -66,6 +66,74 @@ export function stopChatSessionWorkingLifecycle< state.immediateEventRenderPending = !state.disposed && immediateEventRender; } +/** + * Begin a Working lifecycle for a prompt this host did not submit itself — + * currently a workflow-authored `sendUserMessage()` accepted on an idle + * retained stage. Mirrors the manual submit path so an accepted delivery paints + * immediately and then hands the same visible period to the agent turn. + * + * Returns the lifecycle generation that identifies this delivery. Pass it back + * to `settleChatSessionPromptLifecycle` so a stale settlement cannot clear a + * newer turn. + */ +export function startChatSessionExternalPromptLifecycle< + TExtraEntry extends ChatTranscriptEntryLike, +>(state: ChatSessionHostState): number | undefined { + // Compaction status outranks ordinary Working and owns its own indicator. + return openChatSessionExternalPromptLifecycle(state, state.compacting); +} + +/** + * Reopen the Working lifecycle for a delivery that is still active after a + * temporary overlay ended — a pre-turn compaction clears busy state and the + * lifecycle even though the accepted delivery has not started its turn yet. + * + * Unlike the initial start, this preserves `statusMessage` verbatim, so a + * compaction error stays visible and keeps outranking ordinary Working. + */ +export function reassertChatSessionExternalPromptLifecycle< + TExtraEntry extends ChatTranscriptEntryLike, +>(state: ChatSessionHostState): number | undefined { + return openChatSessionExternalPromptLifecycle(state, true); +} + +function openChatSessionExternalPromptLifecycle< + TExtraEntry extends ChatTranscriptEntryLike, +>( + state: ChatSessionHostState, + preserveStatusMessage: boolean, +): number | undefined { + if (state.disposed) return undefined; + if (!preserveStatusMessage) state.statusMessage = ""; + state.sdkBusy = true; + startChatSessionWorkingLifecycle(state); + syncChatSessionAnimationTick(state); + state.requestRender?.(); + return state.workingLifecycleGeneration; +} + +/** + * Settle a prompt lifecycle started by this host. A replaced generation means a + * newer turn owns the indicator, so only its own owner may clear busy state. + */ +export function settleChatSessionPromptLifecycle< + TExtraEntry extends ChatTranscriptEntryLike, +>( + state: ChatSessionHostState, + submittedGeneration: number | undefined, +): void { + if (submittedGeneration === undefined) { + state.sdkBusy = false; + return; + } + const lifecycleWasReplaced = state.workingLifecycleGeneration !== submittedGeneration; + if (lifecycleWasReplaced && state.workingLifecycleActive) return; + state.sdkBusy = false; + if (!lifecycleWasReplaced && !isChatSessionStreaming(state)) { + stopChatSessionWorkingLifecycle(state); + } +} + function clearChatSessionAnimation< TExtraEntry extends ChatTranscriptEntryLike, >(state: ChatSessionHostState): void { diff --git a/packages/coding-agent/src/modes/interactive/components/chat-session-host.ts b/packages/coding-agent/src/modes/interactive/components/chat-session-host.ts index 1bc91b2f3..2f2b476ef 100644 --- a/packages/coding-agent/src/modes/interactive/components/chat-session-host.ts +++ b/packages/coding-agent/src/modes/interactive/components/chat-session-host.ts @@ -30,6 +30,9 @@ import { disposeChatSession, isChatSessionBashRunning, isChatSessionStreaming, + reassertChatSessionExternalPromptLifecycle, + settleChatSessionPromptLifecycle, + startChatSessionExternalPromptLifecycle, syncChatSessionAnimationTick, } from "./chat-session-host-runtime.ts"; import { ChatSessionHostState } from "./chat-session-host-state.ts"; @@ -204,6 +207,30 @@ export class ChatSessionHost rejectReplayMutation("steer"), followUp: async () => rejectReplayMutation("follow up"), subscribe: () => () => {}, + __subscribeDeliveryActivity: () => () => {}, get sessionFile() { return replaySource.sessionFile; }, get sessionId() { return replaySource.sessionId ?? ""; }, setModel: async () => rejectReplayMutation("set model"), diff --git a/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts index 4306ed370..3af5f7cc0 100644 --- a/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts +++ b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts @@ -194,6 +194,7 @@ export function createPostMortemStageHandle( throw new Error("Post-mortem stage chat cannot pause or resume workflow execution."); }, subscribe(listener: AgentSessionEventListener) { return context.subscribe(listener); }, + subscribeDeliveryActivity(listener) { return context.__subscribeDeliveryActivity(listener); }, async dispose() { if (disposed) return; disposed = true; diff --git a/packages/workflows/src/runs/foreground/stage-control-registry.ts b/packages/workflows/src/runs/foreground/stage-control-registry.ts index 60e4ddc40..0ad2dc620 100644 --- a/packages/workflows/src/runs/foreground/stage-control-registry.ts +++ b/packages/workflows/src/runs/foreground/stage-control-registry.ts @@ -26,6 +26,7 @@ */ import type { AgentSession, AgentSessionEvent } from "@bastani/atomic"; +import type { StageDeliveryActivityEvent } from "./stage-delivery-activity.js"; import type { StageUserMessageDeliveryAction } from "./stage-runner-types.js"; export type StageControlStatus = @@ -98,6 +99,15 @@ export interface StageControlHandle { * before the session exists are buffered and bound on first attach. */ subscribe(listener: AgentSessionEventListener): () => void; + /** + * Subscribe to workflow-owned delivery lifecycle facts. Present only for + * handles backed by a live stage runner. An accepted idle delivery reports + * `delivery_start` before the public `agent_start` is published, letting an + * attached chat show Working for the whole turn. + */ + subscribeDeliveryActivity?( + listener: (event: StageDeliveryActivityEvent) => void, + ): () => void; /** Release the underlying SDK session and unregister this direct chat handle. */ dispose?(): void | Promise; } diff --git a/packages/workflows/src/runs/foreground/stage-delivery-activity.ts b/packages/workflows/src/runs/foreground/stage-delivery-activity.ts new file mode 100644 index 000000000..02e90748f --- /dev/null +++ b/packages/workflows/src/runs/foreground/stage-delivery-activity.ts @@ -0,0 +1,76 @@ +/** + * Internal stage delivery-activity channel. + * + * An accepted *idle* stage delivery starts a real backend turn long before the + * public `agent_start` clears the session's ordered extension event queue. The + * attached stage chat therefore had no signal to paint `Working` with, and the + * pane looked idle while the model was already running. + * + * This channel carries only that fact — "an idle delivery was admitted and is + * now starting a turn" — as a workflow-internal event, so the chat can start + * its existing Working lifecycle without fabricating SDK lifecycle events or + * reordering the public event stream. + * + * Streaming deliveries (`steer` / `followUp`) never emit here: they join an + * in-flight turn that already owns the indicator. + * + * cross-ref: + * - src/runs/foreground/stage-runner-send-user-message.ts (emitter) + * - src/tui/stage-chat-view-delivery-activity.ts (consumer) + */ + +export type StageDeliveryActivityEvent = + | { readonly type: "delivery_start"; readonly deliveryId: number } + | { readonly type: "delivery_settled"; readonly deliveryId: number }; + +export type StageDeliveryActivityListener = (event: StageDeliveryActivityEvent) => void; + +export class StageDeliveryActivity { + private readonly listeners = new Set(); + private readonly active = new Set(); + private nextDeliveryId = 0; + + /** + * Register `listener`, then synchronously replay every currently-active + * delivery as `delivery_start`. + * + * A stage chat can attach after a delivery was admitted — a late `/workflow + * attach`, or a remount — and would otherwise see nothing until the public + * `agent_start` finally arrives. Registering before replaying matters: + * snapshotting first would drop a start or settlement that lands in between. + */ + subscribe(listener: StageDeliveryActivityListener): () => void { + this.listeners.add(listener); + for (const deliveryId of [...this.active]) { + if (!this.listeners.has(listener)) break; + listener({ type: "delivery_start", deliveryId }); + } + return () => { + this.listeners.delete(listener); + }; + } + + /** Report an admitted idle delivery that is about to start a turn. */ + start(): number { + const deliveryId = ++this.nextDeliveryId; + this.active.add(deliveryId); + this.emit({ type: "delivery_start", deliveryId }); + return deliveryId; + } + + /** Report that a delivery finished, failed, or started no turn at all. */ + settle(deliveryId: number | undefined): void { + if (deliveryId === undefined || !this.active.delete(deliveryId)) return; + this.emit({ type: "delivery_settled", deliveryId }); + } + + /** Settle every outstanding delivery, then drop listeners. */ + dispose(): void { + for (const deliveryId of [...this.active]) this.settle(deliveryId); + this.listeners.clear(); + } + + private emit(event: StageDeliveryActivityEvent): void { + for (const listener of [...this.listeners]) listener(event); + } +} diff --git a/packages/workflows/src/runs/foreground/stage-runner-context.ts b/packages/workflows/src/runs/foreground/stage-runner-context.ts index b79af0b96..3efad5d57 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-context.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-context.ts @@ -126,6 +126,10 @@ export function createStageContext(opts: StageRunnerOpts): InternalStageContext return controller.subscribe(listener); }, + __subscribeDeliveryActivity(listener) { + return controller.subscribeDeliveryActivity(listener); + }, + get sessionFile() { return controller.currentSession?.sessionFile; }, diff --git a/packages/workflows/src/runs/foreground/stage-runner-controller.ts b/packages/workflows/src/runs/foreground/stage-runner-controller.ts index f03c9d6c7..1e8cfbe5d 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-controller.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-controller.ts @@ -9,6 +9,7 @@ import { structuredOutputToolErrorFromEvent } from "./stage-runner-structured-ou import { buildStageSessionOptions } from "./stage-runner-session-options.js"; import { sendStageUserMessage } from "./stage-runner-send-user-message.js"; import { StageMessageAdmission } from "./stage-runner-message-admission.js"; +import { StageDeliveryActivity, type StageDeliveryActivityListener } from "./stage-delivery-activity.js"; import { nextResumedContextOverflowFallbackIndex, terminatingToolCallId, unresolvedContextOverflowFailure, unresolvedContextOverflowMessage } from "./stage-runner-unresolved-overflow.js"; import type { AgentSessionConsumer, StageModelFallbackMeta, StageRunnerOpts, StageSessionCreateOptions, StageSessionCreateResult, StageSessionEvent, StageSessionRuntime, WorkflowFastModeSettingsManager } from "./stage-runner-types.js"; import { StageSessionReplacement } from "./stage-runner-replacement.js"; @@ -44,6 +45,7 @@ export class StageSessionController { private sessionSettingsManager: WorkflowFastModeSettingsManager | undefined; private readonly replacement = new StageSessionReplacement(); private readonly messageAdmission = new StageMessageAdmission(); + private readonly deliveryActivity = new StageDeliveryActivity(); constructor( private readonly opts: StageRunnerOpts, @@ -117,10 +119,12 @@ export class StageSessionController { return pausedDelivery; } return sendStageUserMessage( - await this.ensureSession("prompt"), content, options, beforeDelivery, release, this.messageAdmission, + await this.ensureSession("prompt"), content, options, beforeDelivery, release, this.messageAdmission, this.deliveryActivity, ); }); } + /** Internal pre-stream delivery lifecycle consumed by an attached stage chat. */ + subscribeDeliveryActivity(listener: StageDeliveryActivityListener): () => void { return this.deliveryActivity.subscribe(listener); } sealGeneration(): void { this.generationSealed = true; @@ -201,7 +205,7 @@ export class StageSessionController { this.pendingListeners.clear(); this.unsubscribeTerminateWatcher?.(); this.unsubscribeTerminateWatcher = undefined; - this.terminatingToolCallIds.clear(); this.messageAdmission.dispose(); + this.terminatingToolCallIds.clear(); this.messageAdmission.dispose(); this.deliveryActivity.dispose(); await this.replacement.dispose(); await disposeStageSession(this.session); } diff --git a/packages/workflows/src/runs/foreground/stage-runner-send-user-message.ts b/packages/workflows/src/runs/foreground/stage-runner-send-user-message.ts index 4b6c0a0ad..f4b20f182 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-send-user-message.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-send-user-message.ts @@ -1,4 +1,5 @@ import type { StageSendUserMessageOptions, StageUserMessageContent } from "../../shared/types.js"; +import type { StageDeliveryActivity } from "./stage-delivery-activity.js"; import type { StageMessageAdmission, StageMessageTurn } from "./stage-runner-message-admission.js"; import type { StageSessionRuntime, StageUserMessageDeliveryAction } from "./stage-runner-types.js"; @@ -59,6 +60,7 @@ export async function sendStageUserMessage( beforeDelivery?: () => void, promptStarted?: () => void, admission?: StageMessageAdmission, + activity?: StageDeliveryActivity, ): Promise { const streaming = activeSession.isStreaming || admission?.isOwned(activeSession) === true; const deliverAs = streaming ? options?.deliverAs ?? "followUp" : options?.deliverAs; @@ -71,6 +73,9 @@ export async function sendStageUserMessage( ? createLocalPromptOwnershipObserver(activeSession, promptStarted) : coordinatedPromptOwnershipObserver(admission.startTurn(activeSession, promptStarted ?? (() => {}))); ownership?.arm(); + // An admitted idle delivery owns the visible Working period from here until + // it settles; a streaming delivery joins a turn that already owns it. + const deliveryId = streaming ? undefined : activity?.start(); try { const delivery = activeSession.sendUserMessage(content, { ...(deliverAs === undefined ? {} : { deliverAs }), @@ -89,6 +94,7 @@ export async function sendStageUserMessage( throw error; } finally { ownership?.dispose(); + activity?.settle(deliveryId); } } if (typeof content !== "string") throw unsupportedContentError(); @@ -102,6 +108,7 @@ export async function sendStageUserMessage( ? createLocalPromptOwnershipObserver(activeSession, promptStarted) : coordinatedPromptOwnershipObserver(admission.startTurn(activeSession, promptStarted ?? (() => {}))); ownership.arm(); + const deliveryId = activity?.start(); try { const turn = activeSession.prompt(content); ownership.observeStreaming(); @@ -113,5 +120,6 @@ export async function sendStageUserMessage( throw error; } finally { ownership.dispose(); + activity?.settle(deliveryId); } } diff --git a/packages/workflows/src/runs/foreground/stage-runner-types.ts b/packages/workflows/src/runs/foreground/stage-runner-types.ts index 93cc583da..c039a8ec3 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-types.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-types.ts @@ -153,6 +153,15 @@ export interface InternalStageContext extends StageContext { options?: StageSendUserMessageOptions, beforeDelivery?: () => void, ): Promise; + /** + * Internal: subscribe to workflow-owned delivery lifecycle facts for this + * stage. An accepted idle delivery reports `delivery_start` before the public + * `agent_start` reaches subscribers, so an attached chat can paint Working + * for the whole turn rather than only after the SDK event queue drains. + */ + __subscribeDeliveryActivity( + listener: (event: import("./stage-delivery-activity.js").StageDeliveryActivityEvent) => void, + ): () => void; /** Internal: synchronously reject new detached traffic without waiting for active work. */ __sealGeneration(): void; /** Internal: atomically stop detached traffic admission and drain admitted work. */ diff --git a/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts b/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts new file mode 100644 index 000000000..aee785bf6 --- /dev/null +++ b/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts @@ -0,0 +1,86 @@ +/** + * Maps workflow-owned stage delivery lifecycle facts onto the attached chat's + * existing Working lifecycle. + * + * A workflow definition that auto-sends to its own retained stage starts a real + * backend turn well before the public `agent_start` reaches this view. Without + * this bridge the pane sits with `sdkBusy=false` and no requested render while + * the model is already running, so it looks idle. + * + * Starting the host's ordinary lifecycle here — rather than synthesising an SDK + * event — keeps `agent_start`, `turn_start`, `turn_end`, `agent_end`, retry, + * fallback, compaction, and terminal cleanup in charge of everything after + * startup, and lets a settled delivery that never produced a turn clean up. + * + * cross-ref: src/runs/foreground/stage-delivery-activity.ts + */ +import type { StageDeliveryActivityEvent } from "../runs/foreground/stage-delivery-activity.js"; +import type { StageChatViewContext } from "./stage-chat-view-types.js"; + +export function subscribeStageChatDeliveryActivity( + ctx: StageChatViewContext, +): (() => void) | null { + const handle = ctx.handle; + if (!handle?.subscribeDeliveryActivity) return null; + return handle.subscribeDeliveryActivity((event) => { + applyStageChatDeliveryActivityEvent(ctx, event); + }); +} + +export function applyStageChatDeliveryActivityEvent( + ctx: StageChatViewContext, + event: StageDeliveryActivityEvent, +): void { + if (event.type === "delivery_start") { + // Replaces any lifecycle this same delivery id already owned; the map keeps + // one live token per delivery so settlement stays correlated. A start that + // arrives after terminal cleanup is fresh authorization for a retained + // post-mortem turn, so it also lifts the stale-start fence. + ctx.terminalLifecycleFenced = false; + ctx.deliveryLifecycles.set(event.deliveryId, ctx.chatHost.beginExternalPromptLifecycle()); + return; + } + if (!ctx.deliveryLifecycles.has(event.deliveryId)) return; + const generation = ctx.deliveryLifecycles.get(event.deliveryId); + ctx.deliveryLifecycles.delete(event.deliveryId); + ctx.chatHost.settleExternalPromptLifecycle(generation); +} + +/** True while at least one admitted delivery still owns this chat's Working period. */ +export function hasActiveStageChatDelivery(ctx: StageChatViewContext): boolean { + return ctx.deliveryLifecycles.size > 0; +} + +/** + * Restore delivery-owned Working after a temporary overlay ended. + * + * A pre-turn `compaction_end` clears busy state and the Working lifecycle even + * though the accepted delivery has not started its turn yet, so the pane would + * sit idle until the public `agent_start`. Reassertion preserves factual status + * text, so a compaction error still outranks ordinary Working. + */ +export function reconcileStageChatDeliveryLifecycle(ctx: StageChatViewContext): void { + if (!hasActiveStageChatDelivery(ctx)) return; + if (ctx.chatHost.isStreaming() && ctx.chatHost.hasAnimationTick()) return; + const generation = ctx.chatHost.reassertExternalPromptLifecycle(); + for (const deliveryId of [...ctx.deliveryLifecycles.keys()]) { + ctx.deliveryLifecycles.set(deliveryId, generation); + } +} + +/** + * Drop every delivery lease admitted before a workflow terminal transition and + * arm the stale-start fence. + * + * Those turns can no longer authorize activity on this chat, and keeping them + * would let a queued late `agent_start` restart Working on a finished stage. A + * settlement for a dropped id is then an ordinary unknown settlement. + * + * The fence is armed only by an observed transition. A chat that mounts onto an + * already-terminal stage never cleaned up in-flight work, so a later + * `agent_start` there is genuine retained-session work and stays visible. + */ +export function invalidateStageChatDeliveryLifecycles(ctx: StageChatViewContext): void { + ctx.deliveryLifecycles.clear(); + ctx.terminalLifecycleFenced = true; +} diff --git a/packages/workflows/src/tui/stage-chat-view-live-events.ts b/packages/workflows/src/tui/stage-chat-view-live-events.ts index 0e60bd8c6..e44d4e6b4 100644 --- a/packages/workflows/src/tui/stage-chat-view-live-events.ts +++ b/packages/workflows/src/tui/stage-chat-view-live-events.ts @@ -1,18 +1,60 @@ import type { AgentSessionEvent } from "@bastani/atomic"; import type { StageChatViewContext } from "./stage-chat-view-types.js"; import { isTerminalStageChatState } from "./stage-chat-view-status.js"; +import { + hasActiveStageChatDelivery, + reconcileStageChatDeliveryLifecycle, +} from "./stage-chat-view-delivery-activity.js"; export function applyStageChatLiveHandleEvent( ctx: StageChatViewContext, event: AgentSessionEvent, ): void { + const staleTerminalStart = isStaleTerminalLifecycleStart(ctx, event); ctx.chatHost.applyAgentEvent(event); + if (staleTerminalStart) { + retainTerminalCleanup(ctx); + return; + } + if (isCompactionEndEvent(event)) reconcileStageChatDeliveryLifecycle(ctx); if (!shouldCleanupAfterLiveEvent(ctx, event)) return; + retainTerminalCleanup(ctx); +} + +function retainTerminalCleanup(ctx: StageChatViewContext): void { const hadAnimationTick = ctx.chatHost.hasAnimationTick(); ctx.chatHost.clearBusyForTerminalWorkflowStage(); if (hadAnimationTick !== ctx.chatHost.hasAnimationTick()) ctx.requestRender?.(); } +/** + * A turn start queued before the stage finished must not repaint Working on a + * settled stage. + * + * The fence is armed only when this chat actually observed a terminal + * transition and cleaned up in-flight work. Work admitted after that carries + * its own authorization and is left alone: a live workflow delivery lease + * (including one replayed at attach time), or a prompt this chat submitted + * itself, which is already busy by the time its `agent_start` arrives. + * `turn_start` is guarded too, otherwise it would restart the lifecycle a + * suppressed `agent_start` just stopped. + */ +function isStaleTerminalLifecycleStart( + ctx: StageChatViewContext, + event: AgentSessionEvent, +): boolean { + if (!ctx.terminalLifecycleFenced) return false; + const type = String((event as { type?: unknown }).type ?? ""); + if (type !== "agent_start" && type !== "turn_start") return false; + if (hasActiveStageChatDelivery(ctx)) return false; + if (ctx.chatHost.isStreaming()) return false; + return isCurrentRunOrStageTerminal(ctx); +} + +function isCompactionEndEvent(event: AgentSessionEvent): boolean { + return String((event as { type?: unknown }).type ?? "") === "compaction_end"; +} + function shouldCleanupAfterLiveEvent( ctx: StageChatViewContext, event: AgentSessionEvent, diff --git a/packages/workflows/src/tui/stage-chat-view-state.ts b/packages/workflows/src/tui/stage-chat-view-state.ts index 8b4a595f2..47434c0df 100644 --- a/packages/workflows/src/tui/stage-chat-view-state.ts +++ b/packages/workflows/src/tui/stage-chat-view-state.ts @@ -18,6 +18,10 @@ import { } from "./stage-chat-view-types.js"; import { noticeRow, noticeSummary } from "./stage-chat-view-transcript.js"; import { applyStageChatLiveHandleEvent } from "./stage-chat-view-live-events.js"; +import { + invalidateStageChatDeliveryLifecycles, + subscribeStageChatDeliveryActivity, +} from "./stage-chat-view-delivery-activity.js"; import { replayPendingToolExecutions } from "./stage-chat-view-pending-tools.js"; import { chatHostStyle, stageChatRenderSettings } from "./stage-chat-view-render-settings.js"; import { hexToAnsi, RESET } from "./color-utils.js"; @@ -66,8 +70,11 @@ export function initializeStageChatView( ctx.lastObservedStageStatus = undefined; ctx.lastObservedRunStatus = undefined; ctx.seenNoticeIds = new Set(); + ctx.deliveryLifecycles = new Map(); + ctx.terminalLifecycleFenced = false; ctx._unsubscribeStore = null; ctx._unsubscribeHandle = null; + ctx._unsubscribeDeliveryActivity = null; ctx._unsubscribeFooterData = null; ctx._unregisterStageUiHost = null; installFocusHold(ctx); @@ -98,6 +105,10 @@ export function initializeStageChatView( ctx._unsubscribeFooterData = ctx.footerData?.onBranchChange(() => ctx.requestRender?.()) ?? null; if (ctx.handle) { + // Delivery activity first: its replay must be in place before the SDK + // stream, so a synchronously replayed `agent_start` on a retained terminal + // stage can see that a live delivery authorizes it. + ctx._unsubscribeDeliveryActivity = subscribeStageChatDeliveryActivity(ctx); ctx._unsubscribeHandle = ctx.handle.subscribe((event) => applyStageChatLiveHandleEvent(ctx, event)); } ctx.chatHost.syncAnimationTick(); @@ -209,6 +220,8 @@ function handleStoreUpdate(ctx: StageChatViewContext): void { } if (isTerminalStageChatTransition(ctx.lastObservedStageStatus, currentStageStatus) || isTerminalStageChatTransition(ctx.lastObservedRunStatus, currentRunStatus)) { ctx.chatHost.clearBusyForTerminalWorkflowStage(); + // Leases admitted before this transition can no longer authorize activity. + invalidateStageChatDeliveryLifecycles(ctx); changed = true; } ctx.lastObservedRunStatus = currentRunStatus; @@ -468,6 +481,9 @@ export function disposeStageChatView(ctx: StageChatViewContext): void { ctx._unsubscribeStore = null; ctx._unsubscribeHandle?.(); ctx._unsubscribeHandle = null; + ctx._unsubscribeDeliveryActivity?.(); + ctx._unsubscribeDeliveryActivity = null; + ctx.deliveryLifecycles.clear(); ctx._unsubscribeFooterData?.(); ctx._unsubscribeFooterData = null; releaseMountedCustomUi(ctx); diff --git a/packages/workflows/src/tui/stage-chat-view-types.ts b/packages/workflows/src/tui/stage-chat-view-types.ts index fa28e55cf..05ebe37e2 100644 --- a/packages/workflows/src/tui/stage-chat-view-types.ts +++ b/packages/workflows/src/tui/stage-chat-view-types.ts @@ -162,8 +162,12 @@ export interface StageChatViewContext { lastObservedStageStatus: StageStatus | undefined; lastObservedRunStatus: RunStatus | undefined; seenNoticeIds: Set; + deliveryLifecycles: Map; + /** True once this chat observed a terminal transition and cleaned up in-flight work. */ + terminalLifecycleFenced: boolean; _unsubscribeStore: (() => void) | null; _unsubscribeHandle: (() => void) | null; + _unsubscribeDeliveryActivity: (() => void) | null; _unsubscribeFooterData: (() => void) | null; _unregisterStageUiHost: (() => void) | null; } diff --git a/test/integration/workflow-followup-working-lifecycle.test.ts b/test/integration/workflow-followup-working-lifecycle.test.ts new file mode 100644 index 000000000..fbf34d527 --- /dev/null +++ b/test/integration/workflow-followup-working-lifecycle.test.ts @@ -0,0 +1,337 @@ +/** + * Regression: a workflow definition that auto-sends a follow-on user message to + * its own retained stage after the previous stream terminated must show + * `Working` in an attached stage chat immediately, without further user input. + * + * The reproduced fault was a missing pre-stream UI lifecycle signal: the second + * backend turn started while the attached chat still had `sdkBusy=false`, + * `workingLifecycleActive=false`, and no requested render, so the pane looked + * idle until the public `agent_start` finally arrived. + * + * cross-ref: + * - packages/workflows/src/runs/foreground/stage-runner-send-user-message.ts + * - packages/workflows/src/tui/stage-chat-view-delivery-activity.ts + */ +import { afterEach, beforeAll, test } from "bun:test"; +import assert from "node:assert/strict"; +import { initTheme, type AgentSessionEvent } from "@bastani/atomic"; +import { workflow } from "../../packages/workflows/src/authoring/workflow.js"; +import { run } from "../../packages/workflows/src/runs/foreground/executor.js"; +import { stageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { StageSessionRuntime } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; +import { store } from "../../packages/workflows/src/shared/store.js"; +import { deriveGraphTheme } from "../../packages/workflows/src/tui/graph-theme.js"; +import { StageChatView } from "../../packages/workflows/src/tui/stage-chat-view.js"; + +const activeRuns: Promise[] = []; + +beforeAll(() => { + initTheme("dark", false); +}); + +afterEach(async () => { + stageControlRegistry.clear(); + store.clear(); + await Promise.allSettled(activeRuns.splice(0)); +}); + +function stripAnsi(text: string): string { + return text + .replace(/\x1B\][^\x07]*(?:\x07|\x1B\\)/g, "") + .replace(/\x1B\[[0-?]*[ -/]*[@-~]/g, ""); +} + +function renderText(view: StageChatView): string { + return stripAnsi(view.render(96).join("\n")); +} + +/** + * The Atomic working row: spinner glyph plus a label. `turn_start` swaps the + * default "Working..." text for a whimsical message, so the row itself — not + * one label — is what must stay visible without a gap. + */ +const WORKING_ROW = /∀\s+\S+/; + +interface SecondTurnGates { + readonly backendStarted: PromiseWithResolvers; + readonly publishAgentStart: PromiseWithResolvers; + readonly agentTurnVisible: PromiseWithResolvers; + readonly finishTurn: PromiseWithResolvers; +} + +/** + * Mirrors the real `AgentSession` delivery contract: the backend turn is live + * (and `promptStarted`/`delivered` have fired) before the public `agent_start` + * clears the ordered extension event queue. + */ +function retainedStageSession( + promptCalls: string[], + gates: SecondTurnGates, +): StageSessionRuntime { + type Listener = Parameters[0]; + const listeners = new Set(); + let streaming = false; + const emit = (event: AgentSessionEvent): void => { + for (const listener of [...listeners]) listener(event as never); + }; + const session = { + async prompt(text: string) { + promptCalls.push(text); + streaming = true; + emit({ type: "agent_start" } as AgentSessionEvent); + emit({ type: "turn_start" } as AgentSessionEvent); + emit({ type: "turn_end" } as AgentSessionEvent); + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + streaming = false; + }, + async sendUserMessage( + content: string, + options?: { + readonly __workflowDelivery?: { + readonly promptStarted?: () => void; + readonly delivered?: (action: string) => void; + }; + }, + ) { + promptCalls.push(content); + // The provider turn is live here; the public agent_start is still queued. + streaming = true; + options?.__workflowDelivery?.promptStarted?.(); + options?.__workflowDelivery?.delivered?.("prompt"); + gates.backendStarted.resolve(); + await gates.publishAgentStart.promise; + emit({ type: "agent_start" } as AgentSessionEvent); + emit({ type: "turn_start" } as AgentSessionEvent); + gates.agentTurnVisible.resolve(); + await gates.finishTurn.promise; + emit({ type: "turn_end" } as AgentSessionEvent); + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + streaming = false; + }, + steer: async () => {}, + followUp: async () => {}, + subscribe(listener: Listener) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + sessionFile: undefined, + sessionId: crypto.randomUUID(), + setModel: async () => {}, + setThinkingLevel: () => {}, + cycleModel: async () => ({ model: undefined, thinkingLevel: undefined }), + cycleThinkingLevel: () => undefined, + agent: {} as never, + model: undefined, + thinkingLevel: "off", + messages: [], + navigateTree: async () => {}, + compact: async () => undefined, + abortCompaction: () => {}, + abort: async () => {}, + dispose: () => {}, + } as unknown as StageSessionRuntime; + Object.defineProperty(session, "isStreaming", { get: () => streaming }); + return session; +} + +test("workflow-authored follow-up after a terminated stream shows Working immediately", async () => { + const promptCalls: string[] = []; + const firstTurnDone = Promise.withResolvers(); + const followUpGate = Promise.withResolvers(); + const gates: SecondTurnGates = { + backendStarted: Promise.withResolvers(), + publishAgentStart: Promise.withResolvers(), + agentTurnVisible: Promise.withResolvers(), + finishTurn: Promise.withResolvers(), + }; + const ids = Promise.withResolvers<{ runId: string; stageId: string }>(); + + const def = workflow({ + name: "workflow-followup-working-lifecycle", + description: "", + inputs: {}, + outputs: {}, + run: async (ctx) => { + const stage = ctx.stage("retained"); + await stage.prompt("first turn"); + firstTurnDone.resolve(); + await followUpGate.promise; + await stage.sendUserMessage("automatic second turn"); + return {}; + }, + }); + + const runPromise = run(def, {}, { + store, + stageControlRegistry, + adapters: { + agentSession: { + async create() { + return retainedStageSession(promptCalls, gates); + }, + }, + }, + onStageStart(runId, stage) { + ids.resolve({ runId, stageId: stage.id }); + }, + confirmStageReadiness: async () => true, + }); + activeRuns.push(runPromise); + + const target = await ids.promise; + await firstTurnDone.promise; + + const handle = stageControlRegistry.get(target.runId, target.stageId); + assert.ok(handle, "expected a retained stage-control handle after the first turn"); + + let renderRequests = 0; + const view = new StageChatView({ + store, + graphTheme: deriveGraphTheme({}), + runId: target.runId, + stageId: target.stageId, + workflowName: "workflow-followup-working-lifecycle", + handle, + onDetach: () => {}, + onClose: () => {}, + requestRender: () => { + renderRequests += 1; + }, + }); + + try { + assert.deepEqual(promptCalls, ["first turn"]); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + + // No further user input: the workflow definition itself sends the message. + followUpGate.resolve(); + await gates.backendStarted.promise; + + assert.deepEqual(promptCalls, ["first turn", "automatic second turn"]); + assert.match( + renderText(view), + /Working/, + "expected Working while the workflow-authored second turn is starting", + ); + assert.equal(view._hasAnimationTick, true); + assert.ok(renderRequests > 0, "expected an immediate render request"); + + // Handing over to the real agent turn must not blink the status off. + gates.publishAgentStart.resolve(); + await gates.agentTurnVisible.promise; + assert.match(renderText(view), WORKING_ROW); + assert.equal(view._hasAnimationTick, true); + + gates.finishTurn.resolve(); + const completed = await runPromise; + assert.equal(completed.status, "completed"); + assert.doesNotMatch(renderText(view), WORKING_ROW); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + gates.publishAgentStart.resolve(); + gates.finishTurn.resolve(); + view.dispose(); + } +}); + +test("a chat attached mid-delivery shows Working without waiting for agent_start", async () => { + const promptCalls: string[] = []; + const firstTurnDone = Promise.withResolvers(); + const followUpGate = Promise.withResolvers(); + const gates: SecondTurnGates = { + backendStarted: Promise.withResolvers(), + publishAgentStart: Promise.withResolvers(), + agentTurnVisible: Promise.withResolvers(), + finishTurn: Promise.withResolvers(), + }; + const ids = Promise.withResolvers<{ runId: string; stageId: string }>(); + + const def = workflow({ + name: "workflow-followup-late-attach", + description: "", + inputs: {}, + outputs: {}, + run: async (ctx) => { + const stage = ctx.stage("retained"); + await stage.prompt("first turn"); + firstTurnDone.resolve(); + await followUpGate.promise; + await stage.sendUserMessage("automatic second turn"); + return {}; + }, + }); + + const runPromise = run(def, {}, { + store, + stageControlRegistry, + adapters: { + agentSession: { + async create() { + return retainedStageSession(promptCalls, gates); + }, + }, + }, + onStageStart(runId, stage) { + ids.resolve({ runId, stageId: stage.id }); + }, + confirmStageReadiness: async () => true, + }); + activeRuns.push(runPromise); + + const target = await ids.promise; + await firstTurnDone.promise; + + // Nothing is attached yet: the delivery is admitted and the backend turn + // starts while the pane does not exist. + followUpGate.resolve(); + await gates.backendStarted.promise; + assert.deepEqual(promptCalls, ["first turn", "automatic second turn"]); + + const handle = stageControlRegistry.get(target.runId, target.stageId); + assert.ok(handle, "expected a retained stage-control handle mid-delivery"); + + let renderRequests = 0; + const view = new StageChatView({ + store, + graphTheme: deriveGraphTheme({}), + runId: target.runId, + stageId: target.stageId, + workflowName: "workflow-followup-late-attach", + handle, + onDetach: () => {}, + onClose: () => {}, + requestRender: () => { + renderRequests += 1; + }, + }); + + try { + assert.match( + renderText(view), + /Working/, + "expected Working on attach while the workflow-authored turn was already running", + ); + assert.equal(view._hasAnimationTick, true); + assert.ok(renderRequests > 0, "expected an immediate render request on attach"); + + gates.publishAgentStart.resolve(); + await gates.agentTurnVisible.promise; + assert.match(renderText(view), WORKING_ROW); + assert.equal(view._hasAnimationTick, true); + + gates.finishTurn.resolve(); + const completed = await runPromise; + assert.equal(completed.status, "completed"); + assert.doesNotMatch(renderText(view), WORKING_ROW); + assert.equal(view._hasAnimationTick, false); + assert.deepEqual(promptCalls, ["first turn", "automatic second turn"]); + } finally { + gates.publishAgentStart.resolve(); + gates.finishTurn.resolve(); + view.dispose(); + } +}); diff --git a/test/unit/stage-chat-view-delivery-activity.test.ts b/test/unit/stage-chat-view-delivery-activity.test.ts new file mode 100644 index 000000000..91ebcd973 --- /dev/null +++ b/test/unit/stage-chat-view-delivery-activity.test.ts @@ -0,0 +1,334 @@ +/** + * Attached stage chat: workflow-authored delivery lifecycle. + * + * A workflow definition auto-sending to its own retained stage must paint + * `Working` at admission — before the public `agent_start` clears the session's + * ordered event queue — and must hand that same visible period to the agent + * turn without a gap. Settlement must clean up when no turn ever starts, and a + * stale settlement must never clear a newer turn. + * + * cross-ref: packages/workflows/src/tui/stage-chat-view-delivery-activity.ts + */ +import { test } from "bun:test"; +import { + assert, + createStore, + deriveGraphTheme, + makeHandle, + setupRun, + StageChatView, + stripAnsi, + type AgentSessionEvent, + type StageControlHandle, +} from "./stage-chat-view-helpers.js"; + +function renderText(view: StageChatView): string { + return stripAnsi(view.render(96).join("\n")); +} + +function stageViewFor( + handle: StageControlHandle, + onRender: () => void = () => {}, + status: "running" | "completed" = "completed", + store: ReturnType = createStore(), + seedRun = true, +): { view: StageChatView; store: ReturnType } { + if (seedRun) setupRun(store, "run-1", "stage-a", status); + const view = new StageChatView({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-1", + stageId: "stage-a", + workflowName: "test-wf", + handle, + onDetach: () => {}, + onClose: () => {}, + requestRender: onRender, + }); + return { view, store }; +} + +function completedStageView( + handle: StageControlHandle, + onRender: () => void, +): { view: StageChatView; store: ReturnType } { + return stageViewFor(handle, onRender, "completed"); +} + +function completeStage(store: ReturnType): void { + const stage = store.runs()[0]!.stages[0]!; + store.recordStageEnd("run-1", { + ...stage, + status: "completed", + endedAt: Date.now(), + durationMs: 1, + }); +} + +test("workflow-authored delivery paints Working before the public agent_start", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "completed"); + let renderRequests = 0; + const { view } = completedStageView(handle, () => { + renderRequests += 1; + }); + + try { + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + + renderRequests = 0; + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + assert.match(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, true); + assert.ok(renderRequests > 0, "expected an immediate render request on admission"); + + // The real turn takes over the same visible period. + emit({ type: "agent_start" } as AgentSessionEvent); + assert.match(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, true); + + emit({ type: "turn_end" } as AgentSessionEvent); + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + + // The delivery settles last; it must not resurrect or disturb the idle pane. + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("a delivery that starts no turn clears Working on settlement", () => { + const { handle, emitDeliveryActivity } = makeHandle(undefined, [], "completed"); + const { view } = completedStageView(handle, () => {}); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 7 }); + assert.match(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, true); + + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 7 }); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("a stale delivery settlement cannot clear a newer turn", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "completed"); + const { view } = completedStageView(handle, () => {}); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + emitDeliveryActivity({ type: "delivery_start", deliveryId: 2 }); + emit({ type: "agent_start" } as AgentSessionEvent); + assert.match(renderText(view), /Working/); + + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + assert.match(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, true); + + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 2 }); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("an unknown settlement id is ignored", () => { + const { handle, emitDeliveryActivity } = makeHandle(undefined, [], "completed"); + const { view } = completedStageView(handle, () => {}); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 3 }); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 99 }); + assert.match(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, true); + } finally { + view.dispose(); + } +}); + +test("compaction status keeps precedence over an accepted delivery", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "completed"); + const { view } = completedStageView(handle, () => {}); + + try { + emit({ type: "compaction_start", reason: "manual" } as AgentSessionEvent); + assert.match(renderText(view), /Compacting context/); + + emitDeliveryActivity({ type: "delivery_start", deliveryId: 5 }); + assert.match(renderText(view), /Compacting context/); + assert.doesNotMatch(renderText(view), /Working\.\.\./); + } finally { + view.dispose(); + } +}); + +test("disposal unsubscribes the delivery channel and stops the animation", () => { + const { handle, emitDeliveryActivity, deliveryActivityListeners } = makeHandle( + undefined, + [], + "completed", + ); + const { view } = completedStageView(handle, () => {}); + + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + assert.equal(deliveryActivityListeners(), 1); + assert.equal(view._hasAnimationTick, true); + + view.dispose(); + assert.equal(deliveryActivityListeners(), 0); + assert.equal(view._hasAnimationTick, false); +}); + +test("pre-turn compaction returns Working to the still-active delivery", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "completed"); + const { view } = completedStageView(handle, () => {}); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + emit({ type: "compaction_start", reason: "threshold" } as AgentSessionEvent); + assert.match(renderText(view), /Auto-compacting/); + + emit({ type: "compaction_end", midTurn: false } as AgentSessionEvent); + assert.match( + renderText(view), + /Working/, + "expected Working to return while the workflow delivery was still active", + ); + assert.equal(view._hasAnimationTick, true); + + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("a compaction error stays authoritative over the active delivery's Working", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "completed"); + const { view } = completedStageView(handle, () => {}); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 4 }); + emit({ type: "compaction_start", reason: "manual" } as AgentSessionEvent); + emit({ + type: "compaction_end", + midTurn: false, + errorMessage: "compaction hit the hard input limit", + } as AgentSessionEvent); + + assert.match(renderText(view), /compaction hit the hard input limit/); + assert.doesNotMatch(renderText(view), /Working\.\.\./); + + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 4 }); + assert.match(renderText(view), /compaction hit the hard input limit/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("a late agent_start after terminal cleanup cannot resurrect Working", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "running"); + const { view, store } = stageViewFor(handle, () => {}, "running"); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + assert.match(renderText(view), /Working/); + + completeStage(store); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + + emit({ type: "agent_start" } as AgentSessionEvent); + assert.doesNotMatch( + renderText(view), + /Working/, + "a start queued before terminal cleanup must not restart Working", + ); + assert.equal(view._hasAnimationTick, false); + + emit({ type: "turn_start" } as AgentSessionEvent); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + + // The stale lease was invalidated, so its settlement is an unknown id. + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("a delivery admitted after terminal cleanup still authorizes a retained turn", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "running"); + const { view, store } = stageViewFor(handle, () => {}, "running"); + + try { + completeStage(store); + emitDeliveryActivity({ type: "delivery_start", deliveryId: 9 }); + assert.match(renderText(view), /Working/); + + emit({ type: "agent_start" } as AgentSessionEvent); + assert.match( + renderText(view), + /Working/, + "post-mortem conversation admitted after cleanup must keep Working", + ); + assert.equal(view._hasAnimationTick, true); + + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 9 }); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("remounting while a delivery is active replays it onto the new view", () => { + const { handle, deliveryActivity } = makeHandle(undefined, [], "completed"); + const store = createStore(); + setupRun(store, "run-1", "stage-a", "completed"); + const first = stageViewFor(handle, () => {}, "completed", store, false).view; + const deliveryId = deliveryActivity.start(); + assert.match(renderText(first), /Working/); + first.dispose(); + + let renderRequests = 0; + const second = stageViewFor( + handle, + () => { + renderRequests += 1; + }, + "completed", + store, + false, + ).view; + + try { + assert.match( + renderText(second), + /Working/, + "a view attached mid-delivery must replay the active delivery", + ); + assert.equal(second._hasAnimationTick, true); + assert.ok(renderRequests > 0, "expected an immediate render request on replay"); + + deliveryActivity.settle(deliveryId); + assert.doesNotMatch(renderText(second), /Working/); + assert.equal(second._hasAnimationTick, false); + } finally { + second.dispose(); + } +}); diff --git a/test/unit/stage-chat-view-helpers.ts b/test/unit/stage-chat-view-helpers.ts index ff2ddc032..404a487e8 100644 --- a/test/unit/stage-chat-view-helpers.ts +++ b/test/unit/stage-chat-view-helpers.ts @@ -29,6 +29,8 @@ import { type TUI, } from "@earendil-works/pi-tui"; import type { StageControlHandle } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import { StageDeliveryActivity } from "../../packages/workflows/src/runs/foreground/stage-delivery-activity.js"; +import type { StageDeliveryActivityEvent } from "../../packages/workflows/src/runs/foreground/stage-delivery-activity.js"; import { StageToolExecutionBuffer } from "../../packages/workflows/src/runs/foreground/stage-tool-execution-buffer.js"; import type { PendingPrompt } from "../../packages/workflows/src/shared/store-types.js"; import { @@ -67,7 +69,14 @@ export function makeHandle( handle: StageControlHandle; state: HandleState; emit: (event: AgentSessionEvent) => void; + emitDeliveryActivity: (event: StageDeliveryActivityEvent) => void; + deliveryActivity: StageDeliveryActivity; + deliveryActivityListeners: () => number; } { + const rawDeliveryListeners = new Set<(event: StageDeliveryActivityEvent) => void>(); + // A real channel so replay-on-subscribe is exercised end to end; the raw set + // stays available for tests that need exact synthetic delivery ids. + const deliveryActivity = new StageDeliveryActivity(); let listener: ((e: AgentSessionEvent) => void) | undefined; const toolExecutions = new StageToolExecutionBuffer(); let handleStatus = status; @@ -113,6 +122,14 @@ export function makeHandle( listener = undefined; }; }, + subscribeDeliveryActivity(l) { + rawDeliveryListeners.add(l); + const unsubscribeChannel = deliveryActivity.subscribe(l); + return () => { + rawDeliveryListeners.delete(l); + unsubscribeChannel(); + }; + }, }; return { handle, @@ -121,6 +138,11 @@ export function makeHandle( toolExecutions.record(event); listener?.(event); }, + emitDeliveryActivity: (event: StageDeliveryActivityEvent) => { + for (const deliveryListener of [...rawDeliveryListeners]) deliveryListener(event); + }, + deliveryActivity, + deliveryActivityListeners: () => rawDeliveryListeners.size, }; } @@ -398,4 +420,12 @@ export { StageUiBroker, SessionManager, }; -export type { AgentSession, AgentSessionEvent, Component, EditorComponent, StageControlHandle, TUI }; +export type { + AgentSession, + AgentSessionEvent, + Component, + EditorComponent, + StageControlHandle, + StageDeliveryActivityEvent, + TUI, +}; diff --git a/test/unit/stage-delivery-activity.test.ts b/test/unit/stage-delivery-activity.test.ts new file mode 100644 index 000000000..779bb4708 --- /dev/null +++ b/test/unit/stage-delivery-activity.test.ts @@ -0,0 +1,244 @@ +/** + * Contract for the workflow-internal stage delivery-activity channel. + * + * Only an *accepted idle* delivery reports `delivery_start`, because only that + * delivery starts a new turn whose Working period the attached chat must own + * before the public `agent_start` is published. Streaming steer/follow-up, a + * rejected admission, and a slash-command style "handled" send must not open a + * lifecycle the UI would have to guess its way out of. + * + * cross-ref: packages/workflows/src/runs/foreground/stage-delivery-activity.ts + */ +import { describe, test } from "bun:test"; +import { StageDeliveryActivity, type StageDeliveryActivityEvent } from "../../packages/workflows/src/runs/foreground/stage-delivery-activity.js"; +import type { AgentSessionAdapter, InternalStageContext, StageSessionRuntime } from "./stage-runner-helpers.js"; +import { assert, createStageContext, makeMockSession, makeOpts } from "./stage-runner-helpers.js"; + +function recordActivity(ctx: InternalStageContext): { + readonly events: StageDeliveryActivityEvent[]; + readonly types: () => string[]; +} { + const events: StageDeliveryActivityEvent[] = []; + ctx.__subscribeDeliveryActivity((event) => events.push(event)); + return { events, types: () => events.map((event) => event.type) }; +} + +function contextFor(session: StageSessionRuntime): InternalStageContext { + const agentSession: AgentSessionAdapter = { async create() { return session; } }; + return createStageContext(makeOpts({ adapters: { agentSession } })) as InternalStageContext; +} + +describe("StageDeliveryActivity", () => { + test("settles outstanding deliveries exactly once on dispose", () => { + const activity = new StageDeliveryActivity(); + const events: StageDeliveryActivityEvent[] = []; + activity.subscribe((event) => events.push(event)); + + const first = activity.start(); + const second = activity.start(); + activity.settle(first); + activity.settle(first); + activity.dispose(); + + assert.deepEqual(events, [ + { type: "delivery_start", deliveryId: first }, + { type: "delivery_start", deliveryId: second }, + { type: "delivery_settled", deliveryId: first }, + { type: "delivery_settled", deliveryId: second }, + ]); + assert.notEqual(first, second); + }); + + test("unsubscribed listeners stop receiving events", () => { + const activity = new StageDeliveryActivity(); + const events: StageDeliveryActivityEvent[] = []; + const unsubscribe = activity.subscribe((event) => events.push(event)); + unsubscribe(); + activity.settle(activity.start()); + assert.deepEqual(events, []); + }); + + test("replays active deliveries to a late subscriber in insertion order", () => { + const activity = new StageDeliveryActivity(); + const first = activity.start(); + const second = activity.start(); + const alreadySettled = activity.start(); + activity.settle(alreadySettled); + + const events: StageDeliveryActivityEvent[] = []; + activity.subscribe((event) => events.push(event)); + + assert.deepEqual(events, [ + { type: "delivery_start", deliveryId: first }, + { type: "delivery_start", deliveryId: second }, + ]); + }); + + test("a replayed delivery still settles live for the late subscriber", () => { + const activity = new StageDeliveryActivity(); + const deliveryId = activity.start(); + const events: StageDeliveryActivityEvent[] = []; + activity.subscribe((event) => events.push(event)); + + activity.settle(deliveryId); + + assert.deepEqual(events, [ + { type: "delivery_start", deliveryId }, + { type: "delivery_settled", deliveryId }, + ]); + }); + + test("a delivery started during replay is seen live exactly once", () => { + const activity = new StageDeliveryActivity(); + const existing = activity.start(); + const events: StageDeliveryActivityEvent[] = []; + let nested: number | undefined; + activity.subscribe((event) => { + events.push(event); + // Registering the listener before replay means a reentrant start is seen + // live exactly once rather than being dropped or replayed twice. + if (nested === undefined && event.deliveryId === existing) nested = activity.start(); + }); + + assert.deepEqual(events.map((event) => event.deliveryId), [existing, nested]); + assert.equal(events.every((event) => event.type === "delivery_start"), true); + }); + + test("an unsubscribed late listener stops receiving replayed deliveries", () => { + const activity = new StageDeliveryActivity(); + activity.start(); + const events: StageDeliveryActivityEvent[] = []; + const unsubscribe = activity.subscribe((event) => events.push(event)); + assert.equal(events.length, 1); + unsubscribe(); + + activity.start(); + assert.equal(events.length, 1); + }); +}); + +describe("stage runner delivery activity", () => { + test("an accepted idle delivery starts before the turn resolves and settles after", async () => { + const turn = Promise.withResolvers(); + const { session } = makeMockSession({ + async sendUserMessage() { + await turn.promise; + }, + }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + const delivery = ctx.__sendUserMessage("workflow-authored follow-up"); + await Promise.resolve(); + assert.deepEqual(activity.types(), ["delivery_start"]); + + turn.resolve(); + assert.equal(await delivery, "prompt"); + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + assert.equal(activity.events[0]!.deliveryId, activity.events[1]!.deliveryId); + }); + + test("streaming steer and follow-up deliveries report no activity", async () => { + const { session } = makeMockSession({ + isStreaming: true, + async sendUserMessage() {}, + }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + assert.equal(await ctx.__sendUserMessage("queued while streaming"), "followUp"); + assert.equal(await ctx.__sendUserMessage("steer while streaming", { deliverAs: "steer" }), "steer"); + + assert.deepEqual(activity.types(), []); + }); + + test("a delivery rejected at admission reports no activity", async () => { + const handled: string[] = []; + const { session } = makeMockSession({ + async sendUserMessage(text) { + handled.push(String(text)); + }, + }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + await assert.rejects( + ctx.__sendUserMessage("rejected", undefined, () => { + throw new DOMException("workflow exited", "AbortError"); + }), + /workflow exited/, + ); + + assert.deepEqual(activity.types(), []); + assert.deepEqual(handled, []); + }); + + test("a handled delivery that starts no turn still settles", async () => { + const { session } = makeMockSession({ + async sendUserMessage(_text, options) { + options?.__workflowDelivery?.delivered?.("handled"); + }, + }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + assert.equal(await ctx.__sendUserMessage("/slash-style"), "handled"); + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + }); + + test("a failed delivery settles instead of leaving a stuck lifecycle", async () => { + const { session } = makeMockSession({ + async sendUserMessage() { + throw new Error("prompt preflight failed"); + }, + }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + await assert.rejects(ctx.__sendUserMessage("doomed"), /prompt preflight failed/); + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + }); + + test("runtimes without native sendUserMessage still report their idle prompt", async () => { + const prompts: string[] = []; + const { session } = makeMockSession({ + async prompt(text) { + prompts.push(text); + }, + }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + assert.equal(await ctx.__sendUserMessage("fallback idle prompt"), "prompt"); + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + assert.deepEqual(prompts, ["fallback idle prompt"]); + }); + + test("disposing the stage settles an in-flight delivery", async () => { + const turn = Promise.withResolvers(); + const { session } = makeMockSession({ + async sendUserMessage() { + await turn.promise; + }, + }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + const delivery = ctx.__sendUserMessage("in flight at disposal"); + await Promise.resolve(); + assert.deepEqual(activity.types(), ["delivery_start"]); + + await ctx.__dispose(); + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + + turn.resolve(); + await delivery; + }); +}); From edbee596b6c65e2d377eca2840ac3b9c6d09301c Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Tue, 28 Jul 2026 16:15:14 -0700 Subject: [PATCH 2/3] fix(workflows): cover retained restore activity Start one correlated delivery lease before lazy session preparation while preserving the final SDK admission guard. Rearm stale lifecycle fencing after the last post-terminal delivery settles. Assistant-model: GPT-5.6 Sol --- packages/coding-agent/CHANGELOG.md | 2 +- packages/coding-agent/docs/workflows.md | 2 +- .../runs/foreground/executor-stage-control.ts | 29 ++- .../runs/foreground/postmortem-stage-chat.ts | 24 +- .../foreground/stage-delivery-activity.ts | 10 + .../runs/foreground/stage-runner-context.ts | 4 +- .../foreground/stage-runner-controller.ts | 26 ++- .../src/runs/foreground/stage-runner-types.ts | 7 + .../tui/stage-chat-view-delivery-activity.ts | 16 ++ ...xecutor-stage-control-handled-race.test.ts | 59 +++++ .../postmortem-stage-chat-working.test.ts | 208 ++++++++++++++++++ .../stage-chat-view-delivery-activity.test.ts | 122 ++++++++++ test/unit/stage-delivery-activity.test.ts | 165 ++++++++++++++ ...ge-runner-public-adapter-ownership.test.ts | 13 +- 14 files changed, 653 insertions(+), 34 deletions(-) create mode 100644 test/unit/postmortem-stage-chat-working.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index c6dfa7228..3d5bd088a 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -4,7 +4,7 @@ ### Fixed -- Fixed attached workflow-stage chats missing Working during workflow-authored idle follow-ups, including chats attached or remounted while the turn is already starting. Working now persists through prompt startup and pre-turn compaction, while live-turn `followUp`/`steer` queueing and factual retry, fallback, compaction, cancellation, and error status precedence remain intact; terminal cleanup keeps a leftover start from restarting it. +- Fixed attached workflow-stage chats missing Working during workflow-authored idle follow-ups, including while a saved stage session restores or a chat attaches mid-startup. Working persists through prompt startup and pre-turn compaction; live-turn queueing and factual status precedence remain intact, and late events cannot restart it after the final retained delivery settles. ## [0.9.11-alpha.6] - 2026-07-28 diff --git a/packages/coding-agent/docs/workflows.md b/packages/coding-agent/docs/workflows.md index 8b7399637..f46349194 100644 --- a/packages/coding-agent/docs/workflows.md +++ b/packages/coding-agent/docs/workflows.md @@ -2819,7 +2819,7 @@ Surface behavior: - **Wheel and trackpad** - While the workflow graph is active, vertical wheel/trackpad gestures pan it up and down, and horizontal gestures pan wide graphs left and right when the terminal exposes horizontal wheel events; these gestures remain scoped to the graph instead of leaking into 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. - **Tool and node detail** - Attached stage chats match main chat's tool-detail expansion behavior while keeping expansion state local to the workflow UI context. Press Ctrl+O (the configurable `app.tools.expand` binding) to expand every visible workflow node and tool card, including single, parallel, and chain subagent progress, current tool activity, and artifact paths; press it again to collapse them. The toggle works for active, completed, and archived stage views, including at the supported 40-column terminal minimum. A mounted prompt, custom question, or other input-owning overlay keeps the key instead of changing expansion. - **Footer context** - An attached live stage chat carries the main chat's current-folder and Git-branch identity into its themed footer and mirrors live extension status lines such as the MCP server indicator. Branch changes trigger a repaint through the host's cached footer provider, and extension status changes are read from that same provider rather than recomputed by the workflow UI. -- **Working animation lifecycle** - Ordinary attached-stage work keeps the same exact one-cell `∀` visible while following the active workflow theme's dark → accent → bright/bold → accent → dark luminance ramp every 88ms. Every agent and SDK turn resets to the dark regular phase with a fresh lifecycle-relative cadence; turn, terminal, error, replacement, and disposal cleanup stop the active timer without stale repaint. In an eligible retained-stage chat, every accepted idle follow-up — including a workflow-authored `stage.sendUserMessage(...)` after a prior turn ended — shows Working on admission or on attach, and keeps it through pre-stream startup, pre-turn compaction, and the handoff to the agent turn. Attaching or remounting mid-delivery paints immediately rather than waiting for the turn's first event. A message queued into a live turn with `followUp`/`steer` uses that turn's existing status instead of starting a new one. A no-turn result, prompt error, or terminal completion removes it, and a start left over from work that terminal cleanup already finished cannot bring it back; an accepted manual retry clears stale status from the prior prompt before showing new pre-stream activity. `NO_COLOR` retains regular/bold activity without foreground-color escapes. Reduced motion uses a static regular accent `∀` without an animation timer; factual automatic retry, fallback, compaction, cancellation, and error copy retains precedence. +- **Working animation lifecycle** - Ordinary attached-stage work keeps the same exact one-cell `∀` visible while following the active workflow theme's dark → accent → bright/bold → accent → dark luminance ramp every 88ms. Every agent and SDK turn resets to the dark regular phase with a fresh lifecycle-relative cadence; turn, terminal, error, replacement, and disposal cleanup stop the active timer without stale repaint. In an eligible retained-stage chat, every accepted idle follow-up — including a workflow-authored `stage.sendUserMessage(...)` after a prior turn ended — shows Working on admission or attach, including while Atomic restores a saved retained conversation, and keeps it through prompt startup, pre-turn compaction, and agent handoff. Attaching or remounting mid-delivery paints immediately rather than waiting for the turn's first event. A message queued into a live turn with `followUp`/`steer` uses that turn's existing status instead of starting a new one. A no-turn result, prompt or restore error, or terminal completion removes it; once the last accepted post-terminal delivery settles, a leftover start cannot bring it back. An accepted manual retry clears stale status from the prior prompt before showing new pre-stream activity. `NO_COLOR` retains regular/bold activity without foreground-color escapes. Reduced motion uses a static regular accent `∀` without an animation timer; factual automatic retry, fallback, compaction, cancellation, and error copy retains precedence. - **Async statusline** - If an async/background subagent is running while the fullscreen workflow graph is open, the graph statusline mirrors the async summary so the background run remains visible; hide the graph with `h`, leave it with `ctrl+x`, or reconnect later to return to the full below-editor async widget. - **Copy mode** - Press `ctrl+t` inside an attached stage chat 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 transcript or prompt scrolling. Archived read-only stage transcripts expose the same footer and copy-mode status, so their text can also be selected and copied; `esc` closes the transcript and `ctrl+x` returns to the graph. 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. - **Run control** - Use `interrupt`, `pause`, and `resume` for resumable live work. Pause/interrupt holds a stage's queued steering and follow-up items in place without dequeuing them or starting continuation; `resume` releases those items once in their existing per-queue order, but queue release alone does not start a model turn. `resume` on a non-paused run reopens the saved snapshot or overlay. Use `quit` to pause a live run gracefully while preserving it for `/workflow resume`. diff --git a/packages/workflows/src/runs/foreground/executor-stage-control.ts b/packages/workflows/src/runs/foreground/executor-stage-control.ts index 4252128e4..89b058303 100644 --- a/packages/workflows/src/runs/foreground/executor-stage-control.ts +++ b/packages/workflows/src/runs/foreground/executor-stage-control.ts @@ -1,21 +1,31 @@ import type { StageControlHandle, AgentSessionEventListener } from "./stage-control-registry.js"; import type { LiveStageRuntime } from "./executor-stage-types.js"; -import type { StageUserMessageDeliveryAction } from "./stage-runner-types.js"; +import type { StageUserMessageDeliveryAction, StageUserMessagePreparation } from "./stage-runner-types.js"; import { isTerminalStage } from "./executor-scheduler.js"; import { StageToolExecutionBuffer } from "./stage-tool-execution-buffer.js"; export function createStageControlHandle(runtime: LiveStageRuntime): StageControlHandle { - const ensureMessagingSession = async (): Promise => { + const messagePreparation = (): StageUserMessagePreparation => { const meta = runtime.innerCtx.__sessionMeta(); - if (meta.sessionId !== undefined || meta.sessionFile !== undefined) return; + if (meta.sessionId !== undefined || meta.sessionFile !== undefined) { + return { beforePreparation: runtime.throwIfStageMutationBlocked }; + } if (runtime.stageSnapshot.sessionFile !== undefined) { - await runtime.innerCtx.__ensureSessionFromFile(runtime.stageSnapshot.sessionFile); - runtime.captureStageSessionMeta(); - return; + return { + sessionFile: runtime.stageSnapshot.sessionFile, + beforePreparation: runtime.throwIfStageMutationBlocked, + }; } if (isTerminalStage(runtime.stageSnapshot)) { throw new Error(`atomic-workflows: cannot message stage "${runtime.name}" because no retained session metadata is available.`); } + return { beforePreparation: runtime.throwIfStageMutationBlocked }; + }; + const ensureMessagingSession = async (): Promise => { + const sessionFile = messagePreparation().sessionFile; + if (sessionFile === undefined) return; + await runtime.innerCtx.__ensureSessionFromFile(sessionFile); + runtime.captureStageSessionMeta(); }; const toolExecutions = new StageToolExecutionBuffer(); const unsubscribeToolExecutions = runtime.innerCtx.subscribe((event) => toolExecutions.record(event)); @@ -57,8 +67,7 @@ export function createStageControlHandle(runtime: LiveStageRuntime): StageContro }, async sendUserMessage(text, options, beforeDelivery) { runtime.throwIfStageMutationBlocked(); - await ensureMessagingSession(); - runtime.throwIfStageMutationBlocked(); + const preparation = messagePreparation(); const admitDelivery = (): void => { runtime.throwIfStageMutationBlocked(); beforeDelivery?.(); @@ -68,6 +77,7 @@ export function createStageControlHandle(runtime: LiveStageRuntime): StageContro text, options, admitDelivery, + preparation, ); if (action === "steer" || action === "followUp") { runtime.state.resumeContinuationPending = "queued-user-message"; @@ -79,13 +89,14 @@ export function createStageControlHandle(runtime: LiveStageRuntime): StageContro }, async prompt(text: string) { runtime.throwIfStageMutationBlocked(); - await ensureMessagingSession(); + const preparation = messagePreparation(); let action: StageUserMessageDeliveryAction | undefined; try { action = await runtime.innerCtx.__sendUserMessage( text, undefined, runtime.throwIfStageMutationBlocked, + preparation, ); } finally { runtime.captureStageSessionMeta(); diff --git a/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts index 3af5f7cc0..80ad964bc 100644 --- a/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts +++ b/packages/workflows/src/runs/foreground/postmortem-stage-chat.ts @@ -34,6 +34,7 @@ import { type StageControlHandle, type StageControlRegistry, } from "./stage-control-registry.js"; +import type { StageUserMessagePreparation } from "./stage-runner-types.js"; /** Why a terminal stage could not be revived as an interactive post-mortem chat. */ export type PostMortemUnavailableReason = @@ -153,10 +154,18 @@ export function createPostMortemStageHandle( ...(defaultSessionDir !== undefined ? { defaultSessionDir } : {}), }); let disposed = false; - const ensureAttached = async (): Promise => { + const throwIfClosed = (): void => { if (disposed) throw new Error(`Post-mortem stage chat "${stage.name}" is closed.`); - if (context.__sessionMeta().sessionFile === undefined) { - await context.__ensureSessionFromFile(sessionFile); + }; + const messagePreparation = (): StageUserMessagePreparation => ({ + ...(context.__sessionMeta().sessionFile === undefined ? { sessionFile } : {}), + beforePreparation: throwIfClosed, + }); + const ensureAttached = async (): Promise => { + throwIfClosed(); + const sessionFileToRestore = messagePreparation().sessionFile; + if (sessionFileToRestore !== undefined) { + await context.__ensureSessionFromFile(sessionFileToRestore); } }; return { @@ -172,8 +181,13 @@ export function createPostMortemStageHandle( get agentSession() { return context.__agentSession(); }, async ensureAttached() { await ensureAttached(); }, async sendUserMessage(text, options, beforeDelivery) { - await ensureAttached(); - return context.__sendUserMessage(text, options, beforeDelivery); + throwIfClosed(); + const preparation = messagePreparation(); + const admitDelivery = (): void => { + throwIfClosed(); + beforeDelivery?.(); + }; + return context.__sendUserMessage(text, options, admitDelivery, preparation); }, async prompt(text: string) { await ensureAttached(); diff --git a/packages/workflows/src/runs/foreground/stage-delivery-activity.ts b/packages/workflows/src/runs/foreground/stage-delivery-activity.ts index 02e90748f..0497d9696 100644 --- a/packages/workflows/src/runs/foreground/stage-delivery-activity.ts +++ b/packages/workflows/src/runs/foreground/stage-delivery-activity.ts @@ -58,6 +58,16 @@ export class StageDeliveryActivity { return deliveryId; } + /** Keep one correlated delivery active for the full asynchronous operation. */ + async runWithLease(operation: () => Promise): Promise { + const deliveryId = this.start(); + try { + return await operation(); + } finally { + this.settle(deliveryId); + } + } + /** Report that a delivery finished, failed, or started no turn at all. */ settle(deliveryId: number | undefined): void { if (deliveryId === undefined || !this.active.delete(deliveryId)) return; diff --git a/packages/workflows/src/runs/foreground/stage-runner-context.ts b/packages/workflows/src/runs/foreground/stage-runner-context.ts index 3efad5d57..aa77acec4 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-context.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-context.ts @@ -110,8 +110,8 @@ export function createStageContext(opts: StageRunnerOpts): InternalStageContext await controller.sendUserMessage(text, options); }, - async __sendUserMessage(text, options, beforeDelivery) { - return controller.sendUserMessage(text, options, beforeDelivery); + async __sendUserMessage(text, options, beforeDelivery, preparation) { + return controller.sendUserMessage(text, options, beforeDelivery, preparation); }, async steer(text) { diff --git a/packages/workflows/src/runs/foreground/stage-runner-controller.ts b/packages/workflows/src/runs/foreground/stage-runner-controller.ts index 1e8cfbe5d..d9b371c79 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-controller.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-controller.ts @@ -11,7 +11,7 @@ import { sendStageUserMessage } from "./stage-runner-send-user-message.js"; import { StageMessageAdmission } from "./stage-runner-message-admission.js"; import { StageDeliveryActivity, type StageDeliveryActivityListener } from "./stage-delivery-activity.js"; import { nextResumedContextOverflowFallbackIndex, terminatingToolCallId, unresolvedContextOverflowFailure, unresolvedContextOverflowMessage } from "./stage-runner-unresolved-overflow.js"; -import type { AgentSessionConsumer, StageModelFallbackMeta, StageRunnerOpts, StageSessionCreateOptions, StageSessionCreateResult, StageSessionEvent, StageSessionRuntime, WorkflowFastModeSettingsManager } from "./stage-runner-types.js"; +import type { AgentSessionConsumer, StageModelFallbackMeta, StageRunnerOpts, StageSessionCreateOptions, StageSessionCreateResult, StageSessionEvent, StageSessionRuntime, StageUserMessagePreparation, WorkflowFastModeSettingsManager } from "./stage-runner-types.js"; import { StageSessionReplacement } from "./stage-runner-replacement.js"; import { StageSessionPause, type StageSessionPauseResumeResult } from "./stage-runner-pause.js"; @@ -108,19 +108,21 @@ export class StageSessionController { return session; } - async sendUserMessage(content: StageUserMessageContent, options?: StageSendUserMessageOptions, beforeDelivery?: () => void): - Promise>> { + async sendUserMessage(content: StageUserMessageContent, options?: StageSendUserMessageOptions, + beforeDelivery?: () => void, preparation?: StageUserMessagePreparation): Promise>> { return this.messageAdmission.run(async (release) => { - const pausedDelivery = this.pauseControl.deferRunnerOwnedDelivery( - () => this.sendUserMessage(content, options, beforeDelivery), - ); - if (pausedDelivery !== undefined) { - release(); - return pausedDelivery; - } - return sendStageUserMessage( - await this.ensureSession("prompt"), content, options, beforeDelivery, release, this.messageAdmission, this.deliveryActivity, + const pausedDelivery = this.pauseControl.deferRunnerOwnedDelivery(() => + this.sendUserMessage(content, options, beforeDelivery, preparation)); + if (pausedDelivery !== undefined) { release(); return pausedDelivery; } + preparation?.beforePreparation?.(); + const sessionFile = preparation?.sessionFile; + const deliver = async (activity?: StageDeliveryActivity) => sendStageUserMessage( + sessionFile === undefined ? await this.ensureSession("prompt") + : await this.ensureSessionFromFile(sessionFile, "prompt"), + content, options, beforeDelivery, release, this.messageAdmission, activity, ); + if (this.session === undefined || sessionFile !== undefined) return this.deliveryActivity.runWithLease(() => deliver()); + return deliver(this.deliveryActivity); }); } /** Internal pre-stream delivery lifecycle consumed by an attached stage chat. */ diff --git a/packages/workflows/src/runs/foreground/stage-runner-types.ts b/packages/workflows/src/runs/foreground/stage-runner-types.ts index c039a8ec3..54eea2fcb 100644 --- a/packages/workflows/src/runs/foreground/stage-runner-types.ts +++ b/packages/workflows/src/runs/foreground/stage-runner-types.ts @@ -29,6 +29,12 @@ export type WorkflowFastModeSettingsManager = { export type StageUserMessageDeliveryAction = "prompt" | "steer" | "followUp" | "handled"; +/** Internal session preparation and side-effect-free eligibility boundary. */ +export interface StageUserMessagePreparation { + readonly sessionFile?: string; + readonly beforePreparation?: () => void; +} + export interface StageUserMessageDeliveryHooks { readonly beforeDelivery?: () => void; /** Releases serialized admission once an idle prompt synchronously owns the turn. */ @@ -152,6 +158,7 @@ export interface InternalStageContext extends StageContext { content: StageUserMessageContent, options?: StageSendUserMessageOptions, beforeDelivery?: () => void, + preparation?: StageUserMessagePreparation, ): Promise; /** * Internal: subscribe to workflow-owned delivery lifecycle facts for this diff --git a/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts b/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts index aee785bf6..9df2ca156 100644 --- a/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts +++ b/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts @@ -15,6 +15,7 @@ * cross-ref: src/runs/foreground/stage-delivery-activity.ts */ import type { StageDeliveryActivityEvent } from "../runs/foreground/stage-delivery-activity.js"; +import { isTerminalStageChatState } from "./stage-chat-view-status.js"; import type { StageChatViewContext } from "./stage-chat-view-types.js"; export function subscribeStageChatDeliveryActivity( @@ -44,6 +45,13 @@ export function applyStageChatDeliveryActivityEvent( const generation = ctx.deliveryLifecycles.get(event.deliveryId); ctx.deliveryLifecycles.delete(event.deliveryId); ctx.chatHost.settleExternalPromptLifecycle(generation); + // The map is the reference count: overlapping leases keep authorization open, + // and the last one to settle on a finished stage restores stale-start + // suppression. Otherwise the fence stayed open forever after the first + // post-terminal delivery and a late start could repaint a settled chat. + if (!hasActiveStageChatDelivery(ctx) && isStageChatContextTerminal(ctx)) { + ctx.terminalLifecycleFenced = true; + } } /** True while at least one admitted delivery still owns this chat's Working period. */ @@ -51,6 +59,14 @@ export function hasActiveStageChatDelivery(ctx: StageChatViewContext): boolean { return ctx.deliveryLifecycles.size > 0; } +/** True when the last observed run or stage status is terminal. */ +export function isStageChatContextTerminal(ctx: StageChatViewContext): boolean { + return ( + isTerminalStageChatState(ctx.lastObservedRunStatus) || + isTerminalStageChatState(ctx.lastObservedStageStatus) + ); +} + /** * Restore delivery-owned Working after a temporary overlay ended. * diff --git a/test/unit/executor-stage-control-handled-race.test.ts b/test/unit/executor-stage-control-handled-race.test.ts index d2e284faf..089b40400 100644 --- a/test/unit/executor-stage-control-handled-race.test.ts +++ b/test/unit/executor-stage-control-handled-race.test.ts @@ -93,3 +93,62 @@ test("stage-control resume rechecks admission after awaited __resume setup", asy assert.equal(nativeReleaseMutations, 0); assert.equal(snapshotMutations, 0); }); + +test("stage-control send keeps saved-session restore inside final message admission", async () => { + const sendEntered = Promise.withResolvers(); + const finishRestore = Promise.withResolvers(); + const events: string[] = []; + let terminal = false; + let sdkMutations = 0; + const runtime = { + runId: "run-restore-race", + stageId: "stage-restore-race", + name: "restore race", + stageSnapshot: { + status: "completed", + sessionFile: "/tmp/retained-restore.jsonl", + }, + state: { liveHandleReleased: false }, + innerCtx: { + __sessionMeta: () => ({ sessionId: undefined, sessionFile: undefined }), + subscribe: () => () => {}, + async __ensureSessionFromFile() { + events.push("outer_restore"); + }, + async __sendUserMessage( + _text: string, + _options: undefined, + beforeDelivery: () => void, + preparation: { + sessionFile?: string; + beforePreparation?: () => void; + } | undefined, + ) { + events.push("inner_send"); + preparation?.beforePreparation?.(); + sendEntered.resolve(); + assert.equal(preparation?.sessionFile, "/tmp/retained-restore.jsonl"); + await finishRestore.promise; + beforeDelivery(); + sdkMutations += 1; + return "prompt" as const; + }, + }, + throwIfStageMutationBlocked() { + if (terminal) throw new DOMException("root terminated during restore", "AbortError"); + }, + captureStageSessionMeta() {}, + }; + const handle = createStageControlHandle(runtime as never); + + const delivery = handle.sendUserMessage?.("late retained message"); + assert.ok(delivery); + void delivery.catch(() => {}); + await sendEntered.promise; + assert.deepEqual(events, ["inner_send"], "restore must not happen before controller admission"); + + terminal = true; + finishRestore.resolve(); + await assert.rejects(delivery, /root terminated during restore/); + assert.equal(sdkMutations, 0, "final admission must run before SDK mutation"); +}); diff --git a/test/unit/postmortem-stage-chat-working.test.ts b/test/unit/postmortem-stage-chat-working.test.ts new file mode 100644 index 000000000..8b3c83f7f --- /dev/null +++ b/test/unit/postmortem-stage-chat-working.test.ts @@ -0,0 +1,208 @@ +/** + * A post-mortem stage chat restores its saved session lazily, on the first + * accepted message. That restore is a real wait, so the pane must show Working + * for it rather than sitting idle until the session finally exists. + * + * Also checks that restore stays conversation-only: the workflow stage snapshot + * must remain terminal throughout, and a failed restore must not leave Working + * stuck. + * + * cross-ref: + * - packages/workflows/src/runs/foreground/postmortem-stage-chat.ts + * - packages/workflows/src/runs/foreground/stage-runner-controller.ts + */ +import { afterEach, beforeEach, test } from "bun:test"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { ensurePostMortemStageHandle } from "../../packages/workflows/src/runs/foreground/postmortem-stage-chat.js"; +import { createStageControlRegistry } from "../../packages/workflows/src/runs/foreground/stage-control-registry.js"; +import type { StageAdapters } from "../../packages/workflows/src/runs/foreground/stage-runner.js"; +import type { StageSnapshot } from "../../packages/workflows/src/shared/store-types.js"; +import { + assert, + createStore, + deriveGraphTheme, + setupRun, + StageChatView, + stripAnsi, + type StageControlHandle, +} from "./stage-chat-view-helpers.js"; +import { mockSession, type StageSessionRuntime } from "./executor-shared.js"; + +let tempDir = ""; +beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), "atomic-postmortem-working-")); +}); +afterEach(() => { + rmSync(tempDir, { recursive: true, force: true }); +}); + +function retainedSessionFile(name: string): string { + const path = join(tempDir, `${name}.jsonl`); + writeFileSync(path, [ + JSON.stringify({ type: "session", version: 3, id: `${name}-session`, timestamp: new Date().toISOString(), cwd: tempDir }), + JSON.stringify({ type: "message", id: `${name}-message`, parentId: null, timestamp: new Date().toISOString(), message: { role: "user", content: "Original stage request" } }), + ].join("\n") + "\n"); + return path; +} + +function completedStage(sessionFile: string): StageSnapshot { + return { + id: "stage-a", + name: "review-a", + status: "completed", + parentIds: [], + toolEvents: [], + sessionFile, + }; +} + +interface LazyHandleFixture { + readonly handle: StageControlHandle; + readonly view: StageChatView; + readonly store: ReturnType; + readonly creationEntered: Promise; + readonly releaseCreation: () => void; + readonly failCreation: (error: Error) => void; + readonly promptCalls: string[]; + readonly lifecycleEvents: string[]; + renderRequests: number; +} + +function lazyPostMortemFixture(name: string, sessionFile: string): LazyHandleFixture { + const entered = Promise.withResolvers(); + const gate = Promise.withResolvers(); + const promptCalls: string[] = []; + const lifecycleEvents: string[] = []; + const session: StageSessionRuntime = { + ...mockSession(), + sessionFile, + async prompt(text: string) { + lifecycleEvents.push("prompt"); + promptCalls.push(text); + }, + async sendUserMessage(content: unknown) { + lifecycleEvents.push("send"); + promptCalls.push(String(content)); + }, + async closeWorkflowStageGeneration() { + lifecycleEvents.push("close_generation"); + }, + } as StageSessionRuntime; + const adapters: StageAdapters = { + agentSession: { + async create() { + entered.resolve(); + await gate.promise; + return session; + }, + }, + }; + const registry = createStageControlRegistry(); + const stage = completedStage(sessionFile); + const result = ensurePostMortemStageHandle(`run-${name}`, stage, { + registry, + adapters, + cwd: tempDir, + }); + assert.equal(result.ok, true); + if (!result.ok) throw new Error("expected an eligible post-mortem stage"); + + const store = createStore(); + setupRun(store, "run-1", "stage-a", "completed"); + const fixture: LazyHandleFixture = { + handle: result.handle, + store, + creationEntered: entered.promise, + releaseCreation: () => gate.resolve(), + failCreation: (error) => gate.reject(error), + promptCalls, + lifecycleEvents, + renderRequests: 0, + view: undefined as unknown as StageChatView, + }; + (fixture as { view: StageChatView }).view = new StageChatView({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-1", + stageId: "stage-a", + workflowName: "test-wf", + handle: result.handle, + onDetach: () => {}, + onClose: () => {}, + requestRender: () => { + fixture.renderRequests += 1; + }, + }); + return fixture; +} + +function renderText(view: StageChatView): string { + return stripAnsi(view.render(96).join("\n")); +} + +test("a post-mortem chat shows Working while its saved session restores", async () => { + const sessionFile = retainedSessionFile("restore-working"); + const fixture = lazyPostMortemFixture("restore", sessionFile); + + try { + assert.doesNotMatch(renderText(fixture.view), /Working/); + fixture.renderRequests = 0; + + const delivery = fixture.handle.sendUserMessage?.("follow up after the fact"); + assert.ok(delivery, "post-mortem handles expose sendUserMessage"); + await fixture.creationEntered; + + assert.match( + renderText(fixture.view), + /Working/, + "expected Working while the retained session was still restoring", + ); + assert.equal(fixture.view._hasAnimationTick, true); + assert.ok(fixture.renderRequests > 0, "expected an immediate render request"); + assert.deepEqual(fixture.promptCalls, [], "no SDK mutation before restore completes"); + + fixture.releaseCreation(); + await delivery; + assert.deepEqual(fixture.promptCalls, ["follow up after the fact"]); + assert.deepEqual( + fixture.lifecycleEvents, + ["close_generation", "send"], + "post-mortem generation closure must precede SDK delivery", + ); + + // Conversation-only: the workflow snapshot stays terminal. + assert.equal(fixture.store.runs()[0]!.stages[0]!.status, "completed"); + } finally { + fixture.releaseCreation(); + fixture.view.dispose(); + } +}); + +test("a failed restore settles the post-mortem chat instead of sticking on Working", async () => { + const sessionFile = retainedSessionFile("restore-failure"); + const fixture = lazyPostMortemFixture("failure", sessionFile); + + try { + const delivery = fixture.handle.sendUserMessage?.("follow up that cannot land"); + assert.ok(delivery); + await fixture.creationEntered; + assert.match(renderText(fixture.view), /Working/); + + fixture.failCreation(new Error("retained session is unreadable")); + await assert.rejects(delivery, /retained session is unreadable/); + + assert.doesNotMatch( + renderText(fixture.view), + /Working/, + "a failed restore must not leave Working stuck", + ); + assert.equal(fixture.view._hasAnimationTick, false); + assert.deepEqual(fixture.promptCalls, []); + assert.deepEqual(fixture.lifecycleEvents, [], "failed restore must not close or deliver"); + assert.equal(fixture.store.runs()[0]!.stages[0]!.status, "completed"); + } finally { + fixture.view.dispose(); + } +}); diff --git a/test/unit/stage-chat-view-delivery-activity.test.ts b/test/unit/stage-chat-view-delivery-activity.test.ts index 91ebcd973..30c406674 100644 --- a/test/unit/stage-chat-view-delivery-activity.test.ts +++ b/test/unit/stage-chat-view-delivery-activity.test.ts @@ -332,3 +332,125 @@ test("remounting while a delivery is active replays it onto the new view", () => second.dispose(); } }); + +test("the terminal fence returns after a no-turn delivery settles", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "running"); + const { view, store } = stageViewFor(handle, () => {}, "running"); + + try { + completeStage(store); + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + assert.doesNotMatch(renderText(view), /Working/); + + emit({ type: "agent_start" } as AgentSessionEvent); + assert.doesNotMatch( + renderText(view), + /Working/, + "a start after the last post-terminal delivery settled must stay suppressed", + ); + emit({ type: "turn_start" } as AgentSessionEvent); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("the terminal fence returns after a completed retained turn settles", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "running"); + const { view, store } = stageViewFor(handle, () => {}, "running"); + + try { + completeStage(store); + emitDeliveryActivity({ type: "delivery_start", deliveryId: 2 }); + emit({ type: "agent_start" } as AgentSessionEvent); + emit({ type: "turn_start" } as AgentSessionEvent); + assert.match(renderText(view), /∀\s+\S/, "the accepted retained turn must show Working"); + + emit({ type: "turn_end" } as AgentSessionEvent); + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 2 }); + assert.doesNotMatch(renderText(view), /Working/); + + emit({ type: "agent_start" } as AgentSessionEvent); + emit({ type: "turn_start" } as AgentSessionEvent); + assert.doesNotMatch( + renderText(view), + /Working/, + "a start after the retained turn finished must stay suppressed", + ); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("settling one of several leases keeps the others authorized", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "running"); + const { view, store } = stageViewFor(handle, () => {}, "running"); + + try { + completeStage(store); + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + emitDeliveryActivity({ type: "delivery_start", deliveryId: 2 }); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + + emit({ type: "agent_start" } as AgentSessionEvent); + assert.match(renderText(view), /Working/, "lease 2 must still authorize the turn"); + assert.equal(view._hasAnimationTick, true); + + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 2 }); + emit({ type: "agent_start" } as AgentSessionEvent); + assert.doesNotMatch(renderText(view), /Working/, "the final settlement must rearm the fence"); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("unknown and invalidated settlements leave the fence untouched", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "running"); + const { view, store } = stageViewFor(handle, () => {}, "running"); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + completeStage(store); + // Lease 1 was dropped by the transition; its settlement must not reopen it. + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 42 }); + + emit({ type: "agent_start" } as AgentSessionEvent); + assert.doesNotMatch(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, false); + + // A genuinely new delivery still reopens authorization. + emitDeliveryActivity({ type: "delivery_start", deliveryId: 5 }); + emit({ type: "agent_start" } as AgentSessionEvent); + assert.match(renderText(view), /Working/); + assert.equal(view._hasAnimationTick, true); + } finally { + view.dispose(); + } +}); + +test("a non-terminal stage keeps ordinary agent work visible after a delivery settles", () => { + const { handle, emit, emitDeliveryActivity } = makeHandle(undefined, [], "running"); + const { view } = stageViewFor(handle, () => {}, "running"); + + try { + emitDeliveryActivity({ type: "delivery_start", deliveryId: 1 }); + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + + emit({ type: "agent_start" } as AgentSessionEvent); + assert.match( + renderText(view), + /Working/, + "a running stage must keep showing ordinary agent work", + ); + assert.equal(view._hasAnimationTick, true); + } finally { + view.dispose(); + } +}); diff --git a/test/unit/stage-delivery-activity.test.ts b/test/unit/stage-delivery-activity.test.ts index 779bb4708..9ea35e2a5 100644 --- a/test/unit/stage-delivery-activity.test.ts +++ b/test/unit/stage-delivery-activity.test.ts @@ -28,6 +28,29 @@ function contextFor(session: StageSessionRuntime): InternalStageContext { return createStageContext(makeOpts({ adapters: { agentSession } })) as InternalStageContext; } +function gatedContext(session: StageSessionRuntime): { + readonly ctx: InternalStageContext; + readonly creationEntered: Promise; + readonly releaseCreation: () => void; + readonly failCreation: (error: Error) => void; +} { + const entered = Promise.withResolvers(); + const gate = Promise.withResolvers(); + const agentSession: AgentSessionAdapter = { + async create() { + entered.resolve(); + await gate.promise; + return session; + }, + }; + return { + ctx: createStageContext(makeOpts({ adapters: { agentSession } })) as InternalStageContext, + creationEntered: entered.promise, + releaseCreation: () => gate.resolve(), + failCreation: (error) => gate.reject(error), + }; +} + describe("StageDeliveryActivity", () => { test("settles outstanding deliveries exactly once on dispose", () => { const activity = new StageDeliveryActivity(); @@ -242,3 +265,145 @@ describe("stage runner delivery activity", () => { await delivery; }); }); + +describe("stage runner delivery activity during session restore", () => { + test("an accepted delivery reports activity while the saved session is still restoring", async () => { + const { session } = makeMockSession({ async sendUserMessage() {} }); + const gated = gatedContext(session); + const activity = recordActivity(gated.ctx); + + const delivery = gated.ctx.__sendUserMessage("restore then deliver"); + await gated.creationEntered; + + assert.deepEqual( + activity.types(), + ["delivery_start"], + "expected Working activity to cover the retained-session restore wait", + ); + + gated.releaseCreation(); + assert.equal(await delivery, "prompt"); + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + assert.equal(activity.events[0]!.deliveryId, activity.events[1]!.deliveryId); + }); + + test("a failed session restore settles the correlated delivery", async () => { + const { session } = makeMockSession({ async sendUserMessage() {} }); + const gated = gatedContext(session); + const activity = recordActivity(gated.ctx); + + const delivery = gated.ctx.__sendUserMessage("doomed restore"); + await gated.creationEntered; + assert.deepEqual(activity.types(), ["delivery_start"]); + + gated.failCreation(new Error("saved session is unreadable")); + await assert.rejects(delivery, /saved session is unreadable/); + + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + assert.equal(activity.events[0]!.deliveryId, activity.events[1]!.deliveryId); + }); + + test("a controlled pause starts one restore lease only after resume", async () => { + const { session } = makeMockSession({ async sendUserMessage() {} }); + let creationStarted = false; + const creationEntered = Promise.withResolvers(); + const releaseCreation = Promise.withResolvers(); + const agentSession: AgentSessionAdapter = { + async create() { + creationStarted = true; + creationEntered.resolve(); + await releaseCreation.promise; + return session; + }, + }; + const ctx = createStageContext(makeOpts({ adapters: { agentSession } })) as InternalStageContext; + const activity = recordActivity(ctx); + + await ctx.__requestPause(); + const delivery = ctx.__sendUserMessage("held until resume"); + await Promise.resolve(); + await Promise.resolve(); + + assert.deepEqual(activity.types(), [], "a paused delivery must stay invisible"); + assert.equal(creationStarted, false, "a paused delivery must not start session restore"); + + const resume = ctx.__resume(); + await creationEntered.promise; + assert.deepEqual( + activity.types(), + ["delivery_start"], + "exactly one lease must cover the pending restore after resume", + ); + + releaseCreation.resolve(); + assert.equal(await delivery, "prompt"); + await resume; + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + assert.equal(activity.events[0]!.deliveryId, activity.events[1]!.deliveryId); + }); + + test("initial eligibility rejects before activity or session creation", async () => { + const { session } = makeMockSession({ async sendUserMessage() {} }); + let creationStarted = false; + const agentSession: AgentSessionAdapter = { + async create() { + creationStarted = true; + return session; + }, + }; + const ctx = createStageContext(makeOpts({ adapters: { agentSession } })) as InternalStageContext; + const activity = recordActivity(ctx); + + await assert.rejects( + ctx.__sendUserMessage("already terminal", undefined, undefined, { + beforePreparation() { + throw new DOMException("root already terminal", "AbortError"); + }, + }), + /root already terminal/, + ); + + assert.equal(creationStarted, false); + assert.deepEqual(activity.types(), []); + }); + + test("a root that terminates during restore is rechecked before SDK mutation", async () => { + const sdkMessages: string[] = []; + const { session } = makeMockSession({ + async sendUserMessage(content) { + sdkMessages.push(String(content)); + }, + }); + const gated = gatedContext(session); + const activity = recordActivity(gated.ctx); + let terminal = false; + + const delivery = gated.ctx.__sendUserMessage( + "must not land after terminal restore", + undefined, + () => { + if (terminal) throw new DOMException("root terminated during restore", "AbortError"); + }, + ); + await gated.creationEntered; + assert.deepEqual(activity.types(), ["delivery_start"]); + + terminal = true; + gated.releaseCreation(); + await assert.rejects(delivery, /root terminated during restore/); + + assert.deepEqual(sdkMessages, []); + assert.deepEqual(activity.types(), ["delivery_start", "delivery_settled"]); + assert.equal(activity.events[0]!.deliveryId, activity.events[1]!.deliveryId); + }); + + test("a streaming delivery reports no activity even when a session already exists", async () => { + const { session } = makeMockSession({ isStreaming: true, async sendUserMessage() {} }); + const ctx = contextFor(session); + await ctx.__ensureSession(); + const activity = recordActivity(ctx); + + assert.equal(await ctx.__sendUserMessage("queued while streaming"), "followUp"); + assert.deepEqual(activity.types(), []); + }); +}); diff --git a/test/unit/stage-runner-public-adapter-ownership.test.ts b/test/unit/stage-runner-public-adapter-ownership.test.ts index 400b08518..c73161182 100644 --- a/test/unit/stage-runner-public-adapter-ownership.test.ts +++ b/test/unit/stage-runner-public-adapter-ownership.test.ts @@ -69,6 +69,7 @@ describe("public AgentSessionAdapter prompt ownership", () => { test("retains public agent_start ownership while isStreaming publication lags", async () => { const allowPromptStart = Promise.withResolvers(); const promptStarted = Promise.withResolvers(); + const followUpRouted = Promise.withResolvers(); const firstTurn = Promise.withResolvers(); let streaming = false; let promptStarts = 0; @@ -92,7 +93,10 @@ describe("public AgentSessionAdapter prompt ownership", () => { for (const listener of listeners) listener({ type: "agent_end", messages: [] }); streaming = false; }, - async followUp(text) { consumed.push(text); }, + async followUp(text) { + consumed.push(text); + followUpRouted.resolve(); + }, }); Object.defineProperty(session, "isStreaming", { get: () => streaming }); const adapter = { @@ -107,14 +111,15 @@ describe("public AgentSessionAdapter prompt ownership", () => { void second.then(() => { secondSettled = true; }); allowPromptStart.resolve(); await promptStarted.promise; - await flushMicrotasks(); + await followUpRouted.promise; + const secondOutcome = await second; const settledBeforeTurnEnd = secondSettled; firstTurn.resolve(); - const outcomes = await Promise.all([first, second]); + const firstOutcome = await first; assert.equal(settledBeforeTurnEnd, true); - assert.deepEqual(outcomes, ["prompt", "followUp"]); + assert.deepEqual([firstOutcome, secondOutcome], ["prompt", "followUp"]); assert.equal(promptStarts, 1); assert.deepEqual(consumed, ["first", "second"]); }); From 2cb36e05bff56aa3fa0f097daf1c18a00efe6eae Mon Sep 17 00:00:00 2001 From: Norin Lavaee Date: Tue, 28 Jul 2026 17:03:55 -0700 Subject: [PATCH 3/3] fix(workflows): retain shared delivery lifecycle ownership A pre-turn compaction_end clears the host Working lifecycle, so the stage chat reasserts one lifecycle and writes that same generation into every delivery lease still open. Two overlapping workflow-authored deliveries then aliased one token, and settling the first one called settleExternalPromptLifecycle() with the current generation, stopping Working and its animation while the second lease was still accepted and its public agent_start still pending. Settle a host generation only when no remaining active lease references that exact token. Distinct generations, including stale ones, still reach the host's own generation fence unchanged, and the terminal stale-start fence stays map-wide so it rearms only after the last lease settles. Assistant-model: Claude Opus 5 --- .../tui/stage-chat-view-delivery-activity.ts | 21 +- .../stage-chat-view-delivery-overlap.test.ts | 267 ++++++++++++++++++ 2 files changed, 287 insertions(+), 1 deletion(-) create mode 100644 test/unit/stage-chat-view-delivery-overlap.test.ts diff --git a/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts b/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts index 9df2ca156..fef466d7a 100644 --- a/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts +++ b/packages/workflows/src/tui/stage-chat-view-delivery-activity.ts @@ -44,7 +44,15 @@ export function applyStageChatDeliveryActivityEvent( if (!ctx.deliveryLifecycles.has(event.deliveryId)) return; const generation = ctx.deliveryLifecycles.get(event.deliveryId); ctx.deliveryLifecycles.delete(event.deliveryId); - ctx.chatHost.settleExternalPromptLifecycle(generation); + // A pre-turn compaction reasserts one lifecycle and hands the same token to + // every lease still open, so several deliveries can alias one generation. + // Only the last owner may settle it; otherwise the first lease to finish + // would stop Working while another accepted delivery is still running. + // Distinct generations — including stale ones — keep reaching the host's own + // generation fence exactly as before. + if (!ownsStageChatLifecycleGeneration(ctx, generation)) { + ctx.chatHost.settleExternalPromptLifecycle(generation); + } // The map is the reference count: overlapping leases keep authorization open, // and the last one to settle on a finished stage restores stale-start // suppression. Otherwise the fence stayed open forever after the first @@ -54,6 +62,17 @@ export function applyStageChatDeliveryActivityEvent( } } +/** True while another still-active delivery references this exact lifecycle token. */ +function ownsStageChatLifecycleGeneration( + ctx: StageChatViewContext, + generation: number | undefined, +): boolean { + for (const active of ctx.deliveryLifecycles.values()) { + if (active === generation) return true; + } + return false; +} + /** True while at least one admitted delivery still owns this chat's Working period. */ export function hasActiveStageChatDelivery(ctx: StageChatViewContext): boolean { return ctx.deliveryLifecycles.size > 0; diff --git a/test/unit/stage-chat-view-delivery-overlap.test.ts b/test/unit/stage-chat-view-delivery-overlap.test.ts new file mode 100644 index 000000000..211020ac5 --- /dev/null +++ b/test/unit/stage-chat-view-delivery-overlap.test.ts @@ -0,0 +1,267 @@ +/** + * Attached stage chat: overlapping delivery leases across a pre-turn compaction. + * + * A pre-turn `compaction_end` clears the host's Working lifecycle, so the view + * reasserts one fresh lifecycle generation and writes it into every active + * delivery lease. Two overlapping leases then alias the same generation, and + * settling the first one must not clear a period the second still owns. + * + * These use real `StageDeliveryActivity` leases gated on resolvers, so + * settlement order is explicit and no assertion depends on a sleep. + * + * cross-ref: packages/workflows/src/tui/stage-chat-view-delivery-activity.ts + */ +import { test } from "bun:test"; +import { + assert, + createStore, + deriveGraphTheme, + makeHandle, + setupRun, + StageChatView, + stripAnsi, + type AgentSessionEvent, + type StageControlHandle, +} from "./stage-chat-view-helpers.js"; +import { StageDeliveryActivity } from "../../packages/workflows/src/runs/foreground/stage-delivery-activity.js"; + +function renderText(view: StageChatView): string { + return stripAnsi(view.render(96).join("\n")); +} + +function isWorking(view: StageChatView): boolean { + return /Working/.test(renderText(view)); +} + +function stageViewFor( + handle: StageControlHandle, + store: ReturnType, + onRender: () => void = () => {}, +): StageChatView { + return new StageChatView({ + store, + graphTheme: deriveGraphTheme({}), + runId: "run-1", + stageId: "stage-a", + workflowName: "test-wf", + handle, + onDetach: () => {}, + onClose: () => {}, + requestRender: onRender, + }); +} + +function completeStage(store: ReturnType): void { + const stage = store.runs()[0]!.stages[0]!; + store.recordStageEnd("run-1", { + ...stage, + status: "completed", + endedAt: Date.now(), + durationMs: 1, + }); +} + +/** A real delivery lease held open until its gate resolves. */ +function gatedLease(activity: StageDeliveryActivity): { + release: () => Promise; +} { + const gate = Promise.withResolvers(); + const lease = activity.runWithLease(() => gate.promise); + return { + release: async () => { + gate.resolve(); + await lease; + }, + }; +} + +function compactBeforeTurn(emit: (event: AgentSessionEvent) => void): void { + emit({ type: "compaction_start", reason: "threshold" } as AgentSessionEvent); + emit({ type: "compaction_end", midTurn: false } as AgentSessionEvent); +} + +test("settling one of two leases sharing a reconciled generation keeps Working", async () => { + const { handle, emit, deliveryActivity } = makeHandle(undefined, [], "completed"); + const store = createStore(); + setupRun(store, "run-1", "stage-a", "completed"); + let renderRequests = 0; + const view = stageViewFor(handle, store, () => { + renderRequests += 1; + }); + + try { + const a = gatedLease(deliveryActivity); + const b = gatedLease(deliveryActivity); + compactBeforeTurn(emit); + assert.equal(isWorking(view), true, "the reasserted lifecycle must repaint Working"); + assert.equal(view._hasAnimationTick, true); + + renderRequests = 0; + await a.release(); + assert.equal( + isWorking(view), + true, + "lease B still owns the reconciled generation, so Working must stay", + ); + assert.equal(view._hasAnimationTick, true); + assert.equal( + renderRequests, + 0, + "dropping an internal map entry must not request a render on its own", + ); + + await b.release(); + assert.equal(isWorking(view), false, "the final correlated settlement must clear Working"); + assert.equal(view._hasAnimationTick, false); + assert.ok(renderRequests > 0, "the final settlement must request a render"); + } finally { + view.dispose(); + } +}); + +test("reversed settlement order clears the shared generation only on the last lease", async () => { + const { handle, emit, deliveryActivity } = makeHandle(undefined, [], "completed"); + const store = createStore(); + setupRun(store, "run-1", "stage-a", "completed"); + const view = stageViewFor(handle, store); + + try { + const a = gatedLease(deliveryActivity); + const b = gatedLease(deliveryActivity); + compactBeforeTurn(emit); + + await b.release(); + assert.equal(isWorking(view), true, "lease A still owns the reconciled generation"); + assert.equal(view._hasAnimationTick, true); + + await a.release(); + assert.equal(isWorking(view), false); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("a third overlapping lease holds the shared generation past two settlements", async () => { + const { handle, emit, deliveryActivity } = makeHandle(undefined, [], "completed"); + const store = createStore(); + setupRun(store, "run-1", "stage-a", "completed"); + const view = stageViewFor(handle, store); + + try { + const a = gatedLease(deliveryActivity); + const b = gatedLease(deliveryActivity); + const c = gatedLease(deliveryActivity); + compactBeforeTurn(emit); + + await b.release(); + assert.equal(isWorking(view), true); + await a.release(); + assert.equal(isWorking(view), true, "lease C still owns the reconciled generation"); + assert.equal(view._hasAnimationTick, true); + + await c.release(); + assert.equal(isWorking(view), false); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("unknown and stale settlements cannot clear a shared reconciled generation", async () => { + const { handle, emit, emitDeliveryActivity, deliveryActivity } = makeHandle( + undefined, + [], + "completed", + ); + const store = createStore(); + setupRun(store, "run-1", "stage-a", "completed"); + const view = stageViewFor(handle, store); + + try { + const a = gatedLease(deliveryActivity); + const b = gatedLease(deliveryActivity); + compactBeforeTurn(emit); + + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 4242 }); + assert.equal(isWorking(view), true, "an unknown id must not clear the shared generation"); + assert.equal(view._hasAnimationTick, true); + + await a.release(); + // Lease A already settled: replaying its id is now an unknown settlement. + emitDeliveryActivity({ type: "delivery_settled", deliveryId: 1 }); + assert.equal(isWorking(view), true, "a replayed settlement must not clear lease B's period"); + assert.equal(view._hasAnimationTick, true); + + await b.release(); + assert.equal(isWorking(view), false); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("the terminal fence returns only after the last shared-generation lease settles", async () => { + const { handle, emit, deliveryActivity } = makeHandle(undefined, [], "running"); + const store = createStore(); + setupRun(store, "run-1", "stage-a", "running"); + const view = stageViewFor(handle, store); + + try { + completeStage(store); + const a = gatedLease(deliveryActivity); + const b = gatedLease(deliveryActivity); + compactBeforeTurn(emit); + assert.equal(isWorking(view), true); + + await a.release(); + emit({ type: "agent_start" } as AgentSessionEvent); + assert.equal(isWorking(view), true, "lease B still authorizes retained work"); + assert.equal(view._hasAnimationTick, true); + + emit({ type: "agent_end", messages: [] } as AgentSessionEvent); + await b.release(); + assert.equal(isWorking(view), false); + assert.equal(view._hasAnimationTick, false); + + emit({ type: "agent_start" } as AgentSessionEvent); + emit({ type: "turn_start" } as AgentSessionEvent); + assert.equal( + isWorking(view), + false, + "a start after the last retained lease settled must stay fenced", + ); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +}); + +test("distinct lease generations keep their existing independent settlement", async () => { + const { handle, deliveryActivity } = makeHandle(undefined, [], "completed"); + const store = createStore(); + setupRun(store, "run-1", "stage-a", "completed"); + const view = stageViewFor(handle, store); + + try { + // No compaction, so each lease owns its own generation and B's is current. + const a = gatedLease(deliveryActivity); + const b = gatedLease(deliveryActivity); + assert.equal(isWorking(view), true); + + await b.release(); + assert.equal( + isWorking(view), + false, + "settling the current distinct generation must still clear it", + ); + assert.equal(view._hasAnimationTick, false); + + // A's stale token must reach the host fence and stay a no-op. + await a.release(); + assert.equal(isWorking(view), false); + assert.equal(view._hasAnimationTick, false); + } finally { + view.dispose(); + } +});