diff --git a/packages/subagents/src/runs/foreground/chain-execution.ts b/packages/subagents/src/runs/foreground/chain-execution.ts index 6817de7e0e..cbc4728789 100644 --- a/packages/subagents/src/runs/foreground/chain-execution.ts +++ b/packages/subagents/src/runs/foreground/chain-execution.ts @@ -405,9 +405,9 @@ export async function executeChain(params: ChainExecutionParams): Promise(WORKFLOW_LIFECYCLE_NOTICE_KINDS); // --------------------------------------------------------------------------- // Public types @@ -34,6 +40,13 @@ export interface WorkflowConfigEntry { * The parsed shape of a workflow extension config file. * All fields optional; absence means "use default". */ +export interface WorkflowNotificationsConfig { + /** Emit lifecycle notices into the main chat. Default: true. */ + readonly enabled?: boolean; + /** Lifecycle states that should create chat notices. */ + readonly notifyOn?: readonly WorkflowLifecycleNoticeKind[]; +} + export interface WorkflowExtensionConfig { /** Explicit named workflows to register by module path. */ readonly workflows?: Readonly>; @@ -47,6 +60,8 @@ export interface WorkflowExtensionConfig { readonly statusFile?: boolean; /** Behaviour on session_start for in-flight runs. Default: "ask". */ readonly resumeInFlight?: "ask" | "auto" | "never"; + /** Main-chat workflow lifecycle notices. */ + readonly workflowNotifications?: WorkflowNotificationsConfig; } /** Severity of a config diagnostic. */ @@ -126,6 +141,10 @@ async function tryReadFile(filePath: string): Promise { * Validate a parsed JSON value as a WorkflowExtensionConfig. * Returns null when valid, or a human-readable rejection reason. */ +function isWorkflowLifecycleNoticeKind(value: unknown): value is WorkflowLifecycleNoticeKind { + return typeof value === "string" && WORKFLOW_LIFECYCLE_NOTICE_KIND_SET.has(value); +} + function validateConfig(value: unknown): string | null { if (value === null || typeof value !== "object" || Array.isArray(value)) { return "config must be a JSON object"; @@ -155,6 +174,31 @@ function validateConfig(value: unknown): string | null { } } + if ("workflowNotifications" in c) { + const value = c["workflowNotifications"]; + if (value === null || typeof value !== "object" || Array.isArray(value)) { + return `"workflowNotifications" must be a JSON object, got ${JSON.stringify(typeof value)}`; + } + const notifications = value as Record; + if ("enabled" in notifications && typeof notifications["enabled"] !== "boolean") { + return `"workflowNotifications.enabled" must be a boolean, got ${JSON.stringify(notifications["enabled"])}`; + } + if ("notifyOn" in notifications) { + const notifyOn = notifications["notifyOn"]; + if (!Array.isArray(notifyOn)) { + return `"workflowNotifications.notifyOn" must be an array, got ${JSON.stringify(typeof notifyOn)}`; + } + if (notifyOn.length === 0) { + return `"workflowNotifications.notifyOn" must be a non-empty array`; + } + for (const item of notifyOn) { + if (!isWorkflowLifecycleNoticeKind(item)) { + return `"workflowNotifications.notifyOn" entries must be "completed", "failed", or "awaiting_input", got ${JSON.stringify(item)}`; + } + } + } + } + if ("workflows" in c) { if (c["workflows"] === null || typeof c["workflows"] !== "object" || Array.isArray(c["workflows"])) { return `"workflows" must be a JSON object, got ${JSON.stringify(typeof c["workflows"])}`; @@ -268,6 +312,14 @@ function mergeConfigs( ...(base.resumeInFlight !== undefined || override.resumeInFlight !== undefined ? { resumeInFlight: override.resumeInFlight ?? base.resumeInFlight } : {}), + ...(base.workflowNotifications !== undefined || override.workflowNotifications !== undefined + ? { + workflowNotifications: { + ...(base.workflowNotifications ?? {}), + ...(override.workflowNotifications ?? {}), + }, + } + : {}), ...(workflows !== undefined ? { workflows } : {}), }; } @@ -289,6 +341,10 @@ export const WORKFLOW_CONFIG_DEFAULTS = { persistRuns: true, statusFile: false, resumeInFlight: "ask" as const, + workflowNotifications: { + enabled: true, + notifyOn: ["completed", "failed", "awaiting_input"] as const, + }, } as const; /** @@ -301,6 +357,10 @@ export interface WorkflowEffectiveConfig { readonly persistRuns: boolean; readonly statusFile: boolean; readonly resumeInFlight: "ask" | "auto" | "never"; + readonly workflowNotifications: { + readonly enabled: boolean; + readonly notifyOn: readonly WorkflowLifecycleNoticeKind[]; + }; readonly workflows?: Readonly>; } @@ -321,6 +381,14 @@ export function withWorkflowDefaults( statusFile: config.statusFile ?? WORKFLOW_CONFIG_DEFAULTS.statusFile, resumeInFlight: config.resumeInFlight ?? WORKFLOW_CONFIG_DEFAULTS.resumeInFlight, + workflowNotifications: { + enabled: + config.workflowNotifications?.enabled + ?? WORKFLOW_CONFIG_DEFAULTS.workflowNotifications.enabled, + notifyOn: + config.workflowNotifications?.notifyOn + ?? WORKFLOW_CONFIG_DEFAULTS.workflowNotifications.notifyOn, + }, ...(config.workflows !== undefined ? { workflows: config.workflows } : {}), }; } diff --git a/packages/workflows/src/extension/index.ts b/packages/workflows/src/extension/index.ts index 35efe497e7..1b55952304 100644 --- a/packages/workflows/src/extension/index.ts +++ b/packages/workflows/src/extension/index.ts @@ -64,6 +64,15 @@ import { WORKFLOW_CONFIG_DEFAULTS, withWorkflowDefaults, } from "./config-loader.js"; +import { + createWorkflowLifecycleNotificationState, + installWorkflowLifecycleNotifications, + registerLifecycleNoticeRenderer, + resetWorkflowLifecycleNotificationState, + seedWorkflowLifecycleNotificationState, + withWorkflowLifecycleNotificationsSuppressed, +} from "./lifecycle-notifications.js"; +import type { WorkflowLifecycleNotificationConfig } from "./lifecycle-notifications.js"; import type { ConfigLoadResult } from "./config-loader.js"; import type { WorkflowPersistencePort, @@ -113,6 +122,13 @@ export interface PiRenderComponent { includes(searchString: string): boolean; } +export interface PiMessageRenderComponent { + render(width: number): string[]; + invalidate?: () => void; +} + +export type PiMessageRendererResult = string | PiMessageRenderComponent | undefined; + function textRenderComponent(text: string): PiRenderComponent { return dynamicTextRenderComponent(() => text); } @@ -273,7 +289,7 @@ export interface ExtensionAPI { registerCommand?: (name: string, options: PiCommandOptions) => void; registerMessageRenderer?: ( event: string, - renderer: (payload: unknown) => string, + renderer: (payload: unknown) => PiMessageRendererResult, ) => void; /** * Inject a custom message into chat history. Used by inline workflow surfaces @@ -1977,6 +1993,32 @@ function factory(pi: ExtensionAPI): void { store, runtimeConfigRef.current, ); + let lifecycleNotificationsUnsubscribe: (() => void) | null = null; + let lifecycleNotificationsActive = false; + const lifecycleNotificationState = createWorkflowLifecycleNotificationState(); + const lifecycleNotificationConfigRef: { current: WorkflowLifecycleNotificationConfig } = { + current: WORKFLOW_CONFIG_DEFAULTS.workflowNotifications, + }; + registerLifecycleNoticeRenderer({ + rendererHost: pi, + registerMessageRenderer: pi.registerMessageRenderer + ? (event, renderer) => pi.registerMessageRenderer?.(event, renderer) + : undefined, + }); + const reinstallLifecycleNotifications = (): void => { + lifecycleNotificationsUnsubscribe?.(); + lifecycleNotificationsUnsubscribe = null; + if (!lifecycleNotificationsActive) return; + lifecycleNotificationsUnsubscribe = installWorkflowLifecycleNotifications({ + store, + config: lifecycleNotificationConfigRef.current, + state: lifecycleNotificationState, + seedExisting: true, + sendMessage: pi.sendMessage + ? (message, options) => pi.sendMessage?.(message, options) + : undefined, + }); + }; let intercomParentSession: string | null = null; const intercomPort = { emit: @@ -2127,6 +2169,8 @@ function factory(pi: ExtensionAPI): void { statusFile: effectiveConfig.statusFile, resumeInFlight: effectiveConfig.resumeInFlight, }; + lifecycleNotificationConfigRef.current = effectiveConfig.workflowNotifications; + reinstallLifecycleNotifications(); // Replace status writer with one that reflects the resolved config. // Unsubscribe the prior (no-op) writer before creating the new one. @@ -3333,6 +3377,7 @@ function factory(pi: ExtensionAPI): void { persistence: persistenceRef.current, }); store.clear(); + resetWorkflowLifecycleNotificationState(lifecycleNotificationState); stageControlRegistry.clear(); // pi-intercom session naming lives here so we don't trip the @@ -3343,6 +3388,8 @@ function factory(pi: ExtensionAPI): void { // Ensure config+discovery are ready before restoring in-flight runs — // tunables must be resolved first. await discoveryPromise; + lifecycleNotificationsActive = true; + reinstallLifecycleNotifications(); if (ctx?.ui) { const diagnostics = formatStartupDiagnostics(configLoadRef.current, discoveryRef.current); if (diagnostics !== null) { @@ -3355,13 +3402,25 @@ function factory(pi: ExtensionAPI): void { const sessionManager = ctx?.sessionManager ?? pi.sessionManager; if (sessionManager) { const cfg = configLoadRef.current?.config; - restoreOnSessionStart( - sessionManager, - { - resumeInFlight: cfg?.resumeInFlight ?? "ask", - persistRuns: cfg?.persistRuns ?? true, + withWorkflowLifecycleNotificationsSuppressed( + lifecycleNotificationState, + () => { + restoreOnSessionStart( + sessionManager, + { + resumeInFlight: cfg?.resumeInFlight ?? "ask", + persistRuns: cfg?.persistRuns ?? true, + }, + store, + ); + // The suppressed subscriber observes restore replay and marks matching + // notices delivered. Seed explicitly as a defensive backstop for + // runtimes without a lifecycle-notification subscriber installed. + seedWorkflowLifecycleNotificationState( + lifecycleNotificationState, + store.snapshot(), + ); }, - store, ); } }); @@ -3385,6 +3444,9 @@ function factory(pi: ExtensionAPI): void { } storeWidgetUnsubscribe?.(); storeWidgetUnsubscribe = null; + lifecycleNotificationsActive = false; + lifecycleNotificationsUnsubscribe?.(); + lifecycleNotificationsUnsubscribe = null; }); } diff --git a/packages/workflows/src/extension/lifecycle-notifications.ts b/packages/workflows/src/extension/lifecycle-notifications.ts new file mode 100644 index 0000000000..12854d0f88 --- /dev/null +++ b/packages/workflows/src/extension/lifecycle-notifications.ts @@ -0,0 +1,364 @@ +import type { + ExtensionAPI, + PiMessageRenderComponent, + PiMessageRendererResult, +} from "./index.js"; +import type { Store } from "../shared/store.js"; +import type { + PendingPrompt, + PromptKind, + RunSnapshot, + RunStatus, + StageSnapshot, + StageStatus, + StoreSnapshot, +} from "../shared/store-types.js"; + +export const LIFECYCLE_NOTICE_CUSTOM_TYPE = "workflows:lifecycle-notice"; +export const LIFECYCLE_NOTICE_SNIPPET_LIMIT = 240; + +export type WorkflowLifecycleNoticeKind = "completed" | "failed" | "awaiting_input"; + +export const WORKFLOW_LIFECYCLE_NOTICE_KINDS = [ + "completed", + "failed", + "awaiting_input", +] as const satisfies readonly WorkflowLifecycleNoticeKind[]; + +export interface WorkflowLifecycleNotificationConfig { + readonly enabled: boolean; + readonly notifyOn: readonly WorkflowLifecycleNoticeKind[]; +} + +export interface WorkflowLifecycleNoticeDetails { + readonly kind: WorkflowLifecycleNoticeKind; + readonly scope: "run" | "stage"; + readonly runId: string; + readonly workflowName: string; + readonly status: RunStatus | StageStatus; + readonly stageId?: string; + readonly stageName?: string; + readonly promptId?: string; + readonly promptKind?: PromptKind; + readonly promptMessage?: string; + readonly error?: string; + readonly failedStageId?: string; + readonly durationMs?: number; + readonly createdAt: number; +} + +export interface WorkflowLifecycleNotificationState { + readonly deliveredTerminalRuns: Set; + readonly deliveredInputPrompts: Set; + suppressionDepth: number; +} + +export interface WorkflowLifecycleNotificationOptions { + readonly store: Store; + readonly sendMessage?: ExtensionAPI["sendMessage"]; + readonly registerMessageRenderer?: ExtensionAPI["registerMessageRenderer"]; + readonly rendererHost?: object; + readonly config: WorkflowLifecycleNotificationConfig; + readonly state?: WorkflowLifecycleNotificationState; + readonly seedExisting?: boolean; +} + +type RawRenderer = (payload: unknown) => PiMessageRendererResult; + +// Process-lifetime registration dedupe: extension hosts are object identities +// and may be garbage-collected, but renderer registrations are not unregistered. +const rendererRegisteredHosts = new WeakSet(); + +export function createWorkflowLifecycleNotificationState(): WorkflowLifecycleNotificationState { + return { + deliveredTerminalRuns: new Set(), + deliveredInputPrompts: new Set(), + suppressionDepth: 0, + }; +} + +export function resetWorkflowLifecycleNotificationState( + state: WorkflowLifecycleNotificationState, +): void { + state.deliveredTerminalRuns.clear(); + state.deliveredInputPrompts.clear(); + state.suppressionDepth = 0; +} + +export function seedWorkflowLifecycleNotificationState( + state: WorkflowLifecycleNotificationState, + snapshot: StoreSnapshot, +): void { + for (const run of snapshot.runs) { + if ((run.status === "completed" || run.status === "failed") && run.endedAt !== undefined) { + state.deliveredTerminalRuns.add(terminalRunKey(run.status, run.id)); + } + if (run.pendingPrompt !== undefined) { + state.deliveredInputPrompts.add(runAwaitingInputKey(run.id, run.pendingPrompt)); + } + for (const stage of run.stages) { + if (stage.status === "awaiting_input") { + state.deliveredInputPrompts.add(awaitingInputKey(run.id, stage)); + } + } + } +} + +/** + * Suppress lifecycle notice emission while still observing snapshot changes and + * marking matching lifecycle states as delivered. This is intended for restore + * or replay paths where historical workflow states should seed dedupe state + * without notifying the current chat; it is not a generic temporary mute that + * should emit the same notices later. + */ +export function withWorkflowLifecycleNotificationsSuppressed( + state: WorkflowLifecycleNotificationState, + fn: () => T, +): T { + state.suppressionDepth += 1; + try { + return fn(); + } finally { + state.suppressionDepth -= 1; + } +} + +export function installWorkflowLifecycleNotifications( + options: WorkflowLifecycleNotificationOptions, +): () => void { + registerLifecycleNoticeRenderer(options); + + if (!options.config.enabled) return () => undefined; + const send = options.sendMessage; + if (typeof send !== "function") return () => undefined; + + const notifyOn = new Set(options.config.notifyOn); + const state = options.state ?? createWorkflowLifecycleNotificationState(); + if (options.seedExisting !== false) { + seedWorkflowLifecycleNotificationState(state, options.store.snapshot()); + } + + const emit = (details: WorkflowLifecycleNoticeDetails): void => { + const content = formatWorkflowLifecycleNoticeText(details); + try { + // Store subscribers are notified in a tight loop. A lifecycle notice + // failure must never abort sibling subscribers such as status writers. + void Promise.resolve( + send( + { + customType: LIFECYCLE_NOTICE_CUSTOM_TYPE, + content, + display: true, + details, + }, + { triggerTurn: true, deliverAs: "steer" }, + ), + ).catch((error: unknown) => warnLifecycleSendFailure(error)); + } catch (error) { + warnLifecycleSendFailure(error); + // Best-effort notification only; keep store delivery isolated. + } + }; + + const emitTerminalNoticeOnce = ( + run: RunSnapshot, + kind: "completed" | "failed", + ): void => { + if (run.status !== kind || run.endedAt === undefined || !notifyOn.has(kind)) { + return; + } + + const key = terminalRunKey(kind, run.id); + if (state.deliveredTerminalRuns.has(key)) return; + + state.deliveredTerminalRuns.add(key); + if (state.suppressionDepth > 0) return; + emit(makeTerminalNotice(run, kind)); + }; + + const emitStageAwaitingInputNoticeOnce = ( + run: RunSnapshot, + stage: StageSnapshot, + ): void => { + if (stage.status !== "awaiting_input") return; + + const key = awaitingInputKey(run.id, stage); + if (state.deliveredInputPrompts.has(key)) return; + + state.deliveredInputPrompts.add(key); + if (state.suppressionDepth > 0) return; + emit(makeStageAwaitingInputNotice(run, stage)); + }; + + const emitRunAwaitingInputNoticeOnce = (run: RunSnapshot): void => { + if (run.pendingPrompt === undefined) return; + + const key = runAwaitingInputKey(run.id, run.pendingPrompt); + if (state.deliveredInputPrompts.has(key)) return; + + state.deliveredInputPrompts.add(key); + if (state.suppressionDepth > 0) return; + emit(makeRunAwaitingInputNotice(run, run.pendingPrompt)); + }; + + const inspect = (snapshot: StoreSnapshot): void => { + for (const run of snapshot.runs) { + emitTerminalNoticeOnce(run, "completed"); + emitTerminalNoticeOnce(run, "failed"); + + if (!notifyOn.has("awaiting_input")) continue; + emitRunAwaitingInputNoticeOnce(run); + for (const stage of run.stages) { + emitStageAwaitingInputNoticeOnce(run, stage); + } + } + }; + + return options.store.subscribe(inspect); +} + +export function registerLifecycleNoticeRenderer( + options: Pick, +): void { + const register = options.registerMessageRenderer; + if (typeof register !== "function") return; + + const host = options.rendererHost ?? register; + if (rendererRegisteredHosts.has(host)) return; + + const renderer: RawRenderer = (raw) => { + const message = raw as { details?: WorkflowLifecycleNoticeDetails }; + if (!message.details) return undefined; + return makeNoticeComponent(message.details); + }; + + register(LIFECYCLE_NOTICE_CUSTOM_TYPE, renderer); + rendererRegisteredHosts.add(host); +} + +export function formatWorkflowLifecycleNoticeText(details: WorkflowLifecycleNoticeDetails): string { + const workflowName = escapeQuotedText(details.workflowName); + if (details.kind === "completed") { + return `✅ Workflow "${workflowName}" completed (run ${details.runId}). Inspect: /workflow status ${details.runId}`; + } + if (details.kind === "failed") { + const stage = details.stageName ?? details.failedStageId; + const stageText = stage ? `, stage ${stage}` : ""; + const errorText = details.error ? `: ${details.error}` : ""; + return `❌ Workflow "${workflowName}" failed (run ${details.runId}${stageText})${errorText}. Inspect: /workflow status ${details.runId}`; + } + const prompt = details.promptMessage ? ` Prompt: ${details.promptMessage}` : ""; + if (details.scope === "run") { + return `❓ Workflow "${workflowName}" needs input (run ${details.runId}).${prompt} Respond: /workflow connect ${details.runId} to answer this run-level prompt.`; + } + const stage = details.stageName ?? details.stageId ?? "unknown"; + const responseHint = details.stageId && details.promptId + ? `/workflow connect ${details.runId} or workflow({ action: "send", runId: ${jsonString(details.runId)}, stageId: ${jsonString(details.stageId)}, promptId: ${jsonString(details.promptId)}, response: ... })` + : `/workflow connect ${details.runId}`; + return `❓ Workflow "${workflowName}" needs input (run ${details.runId}, stage ${stage}).${prompt} Respond: ${responseHint}.`; +} + +function makeTerminalNotice( + run: RunSnapshot, + kind: "completed" | "failed", +): WorkflowLifecycleNoticeDetails { + const failedStage = run.failedStageId + ? run.stages.find((stage) => stage.id === run.failedStageId) + : undefined; + return { + kind, + scope: "run", + runId: run.id, + workflowName: run.name, + status: run.status, + ...(run.error ? { error: truncateSnippet(run.error) } : {}), + ...(run.failedStageId ? { failedStageId: run.failedStageId } : {}), + ...(failedStage ? { stageId: failedStage.id, stageName: failedStage.name } : {}), + ...(run.durationMs !== undefined ? { durationMs: run.durationMs } : {}), + // Normal store paths stamp endedAt; Date.now() is defensive for malformed restored snapshots. + createdAt: run.endedAt ?? Date.now(), + }; +} + +function makeStageAwaitingInputNotice(run: RunSnapshot, stage: StageSnapshot): WorkflowLifecycleNoticeDetails { + const prompt = stage.pendingPrompt; + return { + kind: "awaiting_input", + scope: "stage", + runId: run.id, + workflowName: run.name, + status: stage.status, + stageId: stage.id, + stageName: stage.name, + ...(prompt ? promptFields(prompt) : {}), + // Normal store paths stamp awaitingInputSince; Date.now() is defensive for malformed restored snapshots. + createdAt: prompt?.createdAt ?? stage.awaitingInputSince ?? Date.now(), + }; +} + +function makeRunAwaitingInputNotice(run: RunSnapshot, prompt: PendingPrompt): WorkflowLifecycleNoticeDetails { + return { + kind: "awaiting_input", + scope: "run", + runId: run.id, + workflowName: run.name, + status: run.status, + ...promptFields(prompt), + createdAt: prompt.createdAt, + }; +} + +function warnLifecycleSendFailure(error: unknown): void { + if (process.env.ATOMIC_WORKFLOW_DEBUG !== "1") return; + const message = error instanceof Error ? error.message : String(error); + console.warn("[workflows] workflow lifecycle notice send failed", message); +} + +function escapeQuotedText(value: string): string { + return value.replace(/\\/g, "\\\\").replace(/"/g, '\\"'); +} + +function jsonString(value: string): string { + return JSON.stringify(value); +} + +function terminalRunKey(kind: "completed" | "failed", runId: string): string { + return `${kind}:${runId}`; +} + +function promptFields( + prompt: PendingPrompt, +): Pick { + return { + promptId: prompt.id, + promptKind: prompt.kind, + promptMessage: truncateSnippet(prompt.message), + }; +} + +function awaitingInputKey(runId: string, stage: StageSnapshot): string { + const promptId = stage.pendingPrompt?.id; + if (promptId) return `awaiting_input:${runId}:stage:${stage.id}:${promptId}`; + return `awaiting_input:${runId}:stage:${stage.id}:${stage.awaitingInputSince ?? "active"}`; +} + +function runAwaitingInputKey(runId: string, prompt: PendingPrompt): string { + return `awaiting_input:${runId}:run:${prompt.id}`; +} + +function truncateSnippet(value: string): string { + const normalized = value.replace(/\s+/g, " ").trim(); + if (normalized.length <= LIFECYCLE_NOTICE_SNIPPET_LIMIT) return normalized; + return `${normalized.slice(0, LIFECYCLE_NOTICE_SNIPPET_LIMIT - 1)}…`; +} + +function makeNoticeComponent(details: WorkflowLifecycleNoticeDetails): PiMessageRenderComponent { + return { + render(): string[] { + return [formatWorkflowLifecycleNoticeText(details)]; + }, + invalidate() { + /* stored lifecycle notices are immutable */ + }, + }; +} diff --git a/packages/workflows/src/tui/chat-surface-message.ts b/packages/workflows/src/tui/chat-surface-message.ts index 83c47e0359..3bf9c4eee6 100644 --- a/packages/workflows/src/tui/chat-surface-message.ts +++ b/packages/workflows/src/tui/chat-surface-message.ts @@ -135,16 +135,8 @@ export function registerChatSurfaceRenderer( return makeComponent(payload, theme); }; - // The project's local `ExtensionAPI` types `registerMessageRenderer` as - // returning a plain string. pi's runtime also accepts a Component (see - // docs/extensions.md §Custom UI). Cast through `unknown` so the call - // typechecks against both shapes. `.call(pi, …)` preserves `this` for - // pi's class-backed ExtensionAPI. - (register as unknown as (event: string, r: RawRenderer) => void).call( - pi, - CHAT_SURFACE_CUSTOM_TYPE, - renderer, - ); + // `.call(pi, …)` preserves `this` for pi's class-backed ExtensionAPI. + register.call(pi, CHAT_SURFACE_CUSTOM_TYPE, renderer); rendererRegisteredHosts.add(pi); } diff --git a/packages/workflows/src/tui/inline-form-overlay.ts b/packages/workflows/src/tui/inline-form-overlay.ts index 643be469ff..9b1df2b096 100644 --- a/packages/workflows/src/tui/inline-form-overlay.ts +++ b/packages/workflows/src/tui/inline-form-overlay.ts @@ -124,16 +124,8 @@ export function registerInlineFormRenderer(pi: ExtensionAPI, theme: GraphTheme): }; }; - // The project's local `ExtensionAPI` types `registerMessageRenderer` as - // returning a plain string. The real pi runtime also accepts a Component - // (see docs/extensions.md §Custom UI). Cast through `unknown` so the call - // typechecks against both shapes. Call through `pi` so pi's - // class-backed ExtensionAPI keeps its `this` binding. - (register as unknown as (event: string, r: RawRenderer) => void).call( - pi, - CUSTOM_TYPE, - renderer, - ); + // Call through `pi` so pi's class-backed ExtensionAPI keeps its `this` binding. + register.call(pi, CUSTOM_TYPE, renderer); rendererRegisteredHosts.add(pi); } diff --git a/test/integration/mock-extension-api.test.ts b/test/integration/mock-extension-api.test.ts index 8568df648a..685efad769 100644 --- a/test/integration/mock-extension-api.test.ts +++ b/test/integration/mock-extension-api.test.ts @@ -18,6 +18,7 @@ import factory, { type PiToolOpts, type PiCommandOptions, type PiFlagNamedOpts, + type PiMessageRendererResult, type WorkflowToolArgs, } from "../../packages/workflows/src/extension/index.js"; import type { WorkflowToolResult } from "../../packages/workflows/src/extension/render-result.js"; @@ -44,7 +45,7 @@ interface RegisteredCommand { interface RegisteredRenderer { event: string; - renderer: (payload: Record) => string; + renderer: (payload: Record) => PiMessageRendererResult; } interface RegisteredFlag { @@ -93,7 +94,7 @@ function makeMock(): ExtensionAPI & { commands.push({ name, options }); }, - registerMessageRenderer(event: string, renderer: (payload: Record) => string) { + registerMessageRenderer(event: string, renderer: (payload: Record) => PiMessageRendererResult) { renderers.push({ event, renderer }); }, @@ -154,10 +155,17 @@ function getCommand(commands: RegisteredCommand[], name: string): RegisteredComm function getRenderer( renderers: RegisteredRenderer[], event: string, -): ((payload: Record) => string) | undefined { +): ((payload: Record) => PiMessageRendererResult) | undefined { return renderers.find((r) => r.event === event)?.renderer; } +function expectStringRendererOutput(output: PiMessageRendererResult): string { + if (typeof output !== "string") { + throw new Error("Expected renderer to return a string"); + } + return output; +} + function expectRegisteredCommand( commands: RegisteredCommand[], name: string, @@ -747,8 +755,7 @@ describe("MockExtensionAPI — message renderer registration", () => { test("workflow.run.start renderer returns non-empty string", () => { const renderer = getRenderer(mock.renderers, "workflow.run.start")!; - const out = renderer({ runId: "r1", name: "my-wf", inputs: { foo: "bar" } }); - assert.equal(typeof out, "string"); + const out = expectStringRendererOutput(renderer({ runId: "r1", name: "my-wf", inputs: { foo: "bar" } })); assert.ok(out.length > 0); assert.ok(out.includes("my-wf")); assert.ok(out.includes("r1")); @@ -756,20 +763,20 @@ describe("MockExtensionAPI — message renderer registration", () => { test("workflow.run.start renderer shows input count", () => { const renderer = getRenderer(mock.renderers, "workflow.run.start")!; - const out = renderer({ runId: "r1", name: "wf", inputs: { a: 1, b: 2 } }); + const out = expectStringRendererOutput(renderer({ runId: "r1", name: "wf", inputs: { a: 1, b: 2 } })); assert.ok(out.includes("2")); }); test("workflow.run.end renderer ok status shows success emoji", () => { const renderer = getRenderer(mock.renderers, "workflow.run.end")!; - const out = renderer({ runId: "r1", status: "ok" }); + const out = expectStringRendererOutput(renderer({ runId: "r1", status: "ok" })); assert.ok(out.includes("✅")); assert.ok(out.includes("r1")); }); test("workflow.run.end renderer error status shows failure emoji", () => { const renderer = getRenderer(mock.renderers, "workflow.run.end")!; - const out = renderer({ runId: "r1", status: "error" }); + const out = expectStringRendererOutput(renderer({ runId: "r1", status: "error" })); assert.ok(out.includes("❌")); }); @@ -1392,7 +1399,7 @@ describe("MockExtensionAPI — graceful degradation", () => { test("factory with partial API (only registerMessageRenderer) does not throw", () => { const api: ExtensionAPI = { - registerMessageRenderer(event: string, renderer: (payload: Record) => string) { + registerMessageRenderer(event: string, renderer: (payload: Record) => PiMessageRendererResult) { void event; void renderer; }, diff --git a/test/unit/config-loader-helpers.test.ts b/test/unit/config-loader-helpers.test.ts index 70cdbd20b3..061198e58c 100644 --- a/test/unit/config-loader-helpers.test.ts +++ b/test/unit/config-loader-helpers.test.ts @@ -39,6 +39,13 @@ describe("withWorkflowDefaults — empty config applies all defaults", () => { assert.equal(withWorkflowDefaults({}).resumeInFlight, WORKFLOW_CONFIG_DEFAULTS.resumeInFlight,); }); + test("workflowNotifications defaults are applied", () => { + assert.deepEqual( + withWorkflowDefaults({}).workflowNotifications, + WORKFLOW_CONFIG_DEFAULTS.workflowNotifications, + ); + }); + test("workflows is undefined when absent from config", () => { assert.equal(withWorkflowDefaults({}).workflows, undefined); }); @@ -69,6 +76,15 @@ describe("withWorkflowDefaults — explicit values are preserved", () => { assert.equal(withWorkflowDefaults({ resumeInFlight: "never" }).resumeInFlight, "never"); }); + test("workflowNotifications override is merged with defaults", () => { + assert.deepEqual(withWorkflowDefaults({ + workflowNotifications: { enabled: false, notifyOn: ["failed"] }, + }).workflowNotifications, { + enabled: false, + notifyOn: ["failed"], + }); + }); + test("workflows map is passed through unchanged", () => { const wf = { deploy: { path: "/deploy.ts" } }; assert.deepEqual(withWorkflowDefaults({ workflows: wf }).workflows, wf); @@ -99,6 +115,7 @@ describe("withWorkflowDefaults — partial config: only absent fields get defaul persistRuns: false, statusFile: true, resumeInFlight: "never", + workflowNotifications: { enabled: false, notifyOn: ["completed"] }, workflows: { wf: { path: "/x.ts" } }, }; const result = withWorkflowDefaults(config); @@ -107,6 +124,7 @@ describe("withWorkflowDefaults — partial config: only absent fields get defaul assert.equal(result.persistRuns, false); assert.equal(result.statusFile, true); assert.equal(result.resumeInFlight, "never"); + assert.deepEqual(result.workflowNotifications, { enabled: false, notifyOn: ["completed"] }); assert.deepEqual(result.workflows, { wf: { path: "/x.ts" } }); }); }); @@ -140,6 +158,13 @@ describe("withWorkflowDefaults — WORKFLOW_CONFIG_DEFAULTS constants", () => { test("WORKFLOW_CONFIG_DEFAULTS.resumeInFlight is 'ask'", () => { assert.equal(WORKFLOW_CONFIG_DEFAULTS.resumeInFlight, "ask"); }); + + test("WORKFLOW_CONFIG_DEFAULTS.workflowNotifications enables all lifecycle steer notices", () => { + assert.deepEqual(WORKFLOW_CONFIG_DEFAULTS.workflowNotifications, { + enabled: true, + notifyOn: ["completed", "failed", "awaiting_input"], + }); + }); }); // --------------------------------------------------------------------------- diff --git a/test/unit/config-loader.test.ts b/test/unit/config-loader.test.ts index dc6d42f007..c24e0ed055 100644 --- a/test/unit/config-loader.test.ts +++ b/test/unit/config-loader.test.ts @@ -542,3 +542,74 @@ describe("ConfigDiagnostic shape", () => { assert.ok(diag.source!.includes("config.json")); }); }); + +describe("loadWorkflowConfig — workflowNotifications", () => { + let tmpHome: string; + let tmpProject: string; + + beforeAll(async () => { + tmpHome = await mkdtemp(join(tmpdir(), "wf-home-notifications-")); + tmpProject = await mkdtemp(join(tmpdir(), "wf-project-notifications-")); + }); + + afterAll(async () => { + await rm(tmpHome, { recursive: true, force: true }); + await rm(tmpProject, { recursive: true, force: true }); + }); + + test("accepts valid notification config", async () => { + const dir = await makeDir(tmpProject, ".atomic", "extensions", "workflow"); + await writeJson(dir, "config.json", { + workflowNotifications: { + enabled: false, + notifyOn: ["failed", "awaiting_input"], + }, + }); + + const result = await loadWorkflowConfig({ homeDir: tmpHome, projectRoot: tmpProject }); + assert.deepEqual(result.config?.workflowNotifications, { + enabled: false, + notifyOn: ["failed", "awaiting_input"], + }); + assert.equal(result.diagnostics.length, 0); + }); + + test("rejects invalid notifyOn entries", async () => { + const home = await mkdtemp(join(tmpdir(), "wf-home-notifications-invalid-")); + const project = await mkdtemp(join(tmpdir(), "wf-project-notifications-invalid-")); + try { + const dir = await makeDir(project, ".atomic", "extensions", "workflow"); + await writeJson(dir, "config.json", { + workflowNotifications: { notifyOn: ["completed", "killed"] }, + }); + + const result = await loadWorkflowConfig({ homeDir: home, projectRoot: project }); + assert.equal(result.config, null); + assert.equal(result.diagnostics.length, 1); + assert.match(result.diagnostics[0]?.message ?? "", /workflowNotifications\.notifyOn/); + } finally { + await rm(home, { recursive: true, force: true }); + await rm(project, { recursive: true, force: true }); + } + }); + + test("rejects empty notifyOn arrays", async () => { + const home = await mkdtemp(join(tmpdir(), "wf-home-notifications-empty-")); + const project = await mkdtemp(join(tmpdir(), "wf-project-notifications-empty-")); + try { + const dir = await makeDir(project, ".atomic", "extensions", "workflow"); + await writeJson(dir, "config.json", { + workflowNotifications: { notifyOn: [] }, + }); + + const result = await loadWorkflowConfig({ homeDir: home, projectRoot: project }); + assert.equal(result.config, null); + assert.equal(result.diagnostics.length, 1); + assert.match(result.diagnostics[0]?.message ?? "", /workflowNotifications\.notifyOn/); + assert.match(result.diagnostics[0]?.message ?? "", /non-empty|at least one/); + } finally { + await rm(home, { recursive: true, force: true }); + await rm(project, { recursive: true, force: true }); + } + }); +}); diff --git a/test/unit/status-writer.test.ts b/test/unit/status-writer.test.ts index e07d38c541..deb5e23fb3 100644 --- a/test/unit/status-writer.test.ts +++ b/test/unit/status-writer.test.ts @@ -328,7 +328,7 @@ describe("createStatusWriter — statusFile:true", () => { startedAt: Date.now(), }); - await sleep(50); + await writer.flush(); writer.unsubscribe(); const raw = await readFile(expectedPath, "utf8"); diff --git a/test/unit/workflow-lifecycle-notifications.test.ts b/test/unit/workflow-lifecycle-notifications.test.ts new file mode 100644 index 0000000000..39b836be02 --- /dev/null +++ b/test/unit/workflow-lifecycle-notifications.test.ts @@ -0,0 +1,591 @@ +import { describe, test } from "bun:test"; +import assert from "node:assert/strict"; +import { + createWorkflowLifecycleNotificationState, + installWorkflowLifecycleNotifications, + formatWorkflowLifecycleNoticeText, + LIFECYCLE_NOTICE_CUSTOM_TYPE, + LIFECYCLE_NOTICE_SNIPPET_LIMIT, + registerLifecycleNoticeRenderer, + resetWorkflowLifecycleNotificationState, + seedWorkflowLifecycleNotificationState, + withWorkflowLifecycleNotificationsSuppressed, + type WorkflowLifecycleNoticeDetails, +} from "../../packages/workflows/src/extension/lifecycle-notifications.js"; +import { restoreOnSessionStart, type SessionEntry } from "../../packages/workflows/src/shared/persistence-restore.js"; +import { createStore } from "../../packages/workflows/src/shared/store.js"; +import type { PendingPrompt, StageSnapshot } from "../../packages/workflows/src/shared/store-types.js"; + +interface SentMessage { + readonly customType: string; + readonly content?: string; + readonly display?: boolean; + readonly details?: WorkflowLifecycleNoticeDetails; +} + +interface CardComponent { + render(width: number): string[]; + invalidate?(): void; +} + +interface RegisteredRenderer { + readonly event: string; + readonly renderer: (payload: unknown) => unknown; +} + +type SendOptions = { + readonly triggerTurn?: boolean; + readonly deliverAs?: "steer" | "followUp" | "nextTurn"; +}; + +const config = { + enabled: true, + notifyOn: ["completed", "failed", "awaiting_input"] as const, +}; + +function runningStage(overrides: Partial = {}): StageSnapshot { + return { + id: "stage-1", + name: "planner", + status: "running", + parentIds: [], + toolEvents: [], + ...overrides, + }; +} + +function prompt(overrides: Partial = {}): PendingPrompt { + return { + id: "prompt-1", + kind: "confirm", + message: "Proceed with this plan?", + createdAt: 10, + ...overrides, + }; +} + +function install() { + const store = createStore(); + const sent: SentMessage[] = []; + const options: SendOptions[] = []; + const unsubscribe = installWorkflowLifecycleNotifications({ + store, + config, + sendMessage(message, sendOptions) { + sent.push(message as SentMessage); + options.push(sendOptions ?? {}); + }, + }); + return { store, sent, options, unsubscribe }; +} + +function installWithState( + store: ReturnType, + state: ReturnType, + sent: SentMessage[], +): () => void { + return installWorkflowLifecycleNotifications({ + store, + config, + state, + seedExisting: true, + sendMessage(message) { sent.push(message as SentMessage); }, + }); +} + +function startRun(store: ReturnType, id: string, name = id): void { + store.recordRunStart({ id, name, inputs: {}, status: "running", stages: [], startedAt: 1 }); +} + +describe("installWorkflowLifecycleNotifications", () => { + test("emits one completion notice when a run completes", () => { + const { store, sent, options } = install(); + store.recordRunStart({ id: "run-1", name: "release", inputs: {}, status: "running", stages: [], startedAt: 1 }); + + assert.equal(store.recordRunEnd("run-1", "completed", {}, undefined), true); + store.recordNotice({ id: "nudge", level: "info", message: "force notify", createdAt: 3 }); + + assert.equal(sent.length, 1); + assert.deepEqual(options, [{ triggerTurn: true, deliverAs: "steer" }]); + assert.equal(sent[0]?.customType, LIFECYCLE_NOTICE_CUSTOM_TYPE); + assert.equal(sent[0]?.display, true); + assert.equal(sent[0]?.details?.kind, "completed"); + assert.equal(sent[0]?.details?.scope, "run"); + assert.equal(sent[0]?.details?.workflowName, "release"); + assert.match(sent[0]?.content ?? "", /\/workflow status run-1/); + }); + + test("emits failure notice with stage and truncated error context", () => { + const { store, sent, options } = install(); + const longError = `${"No API key. ".repeat(40)}tail`; + store.recordRunStart({ id: "run-2", name: "deploy", inputs: {}, status: "running", stages: [], startedAt: 1 }); + store.recordStageStart("run-2", runningStage({ id: "stage-2", name: "publish" })); + + assert.equal(store.recordRunEnd("run-2", "failed", undefined, longError, { failedStageId: "stage-2" }), true); + + assert.equal(sent.length, 1); + assert.deepEqual(options, [{ triggerTurn: true, deliverAs: "steer" }]); + assert.equal(sent[0]?.details?.kind, "failed"); + assert.equal(sent[0]?.details?.stageName, "publish"); + assert.equal(sent[0]?.details?.error?.length, LIFECYCLE_NOTICE_SNIPPET_LIMIT); + assert.match(sent[0]?.details?.error ?? "", /…$/); + }); + + test("emits awaiting-input notice for a stage pending prompt", () => { + const { store, sent, options } = install(); + store.recordRunStart({ id: "run-3", name: "review", inputs: {}, status: "running", stages: [], startedAt: 1 }); + store.recordStageStart("run-3", runningStage()); + + assert.equal(store.recordStagePendingPrompt("run-3", "stage-1", prompt()), true); + + assert.equal(sent.length, 1); + assert.deepEqual(options, [{ triggerTurn: true, deliverAs: "steer" }]); + assert.equal(sent[0]?.details?.kind, "awaiting_input"); + assert.equal(sent[0]?.details?.scope, "stage"); + assert.equal(sent[0]?.details?.promptId, "prompt-1"); + assert.match(sent[0]?.content ?? "", /workflow\(\{ action: "send"/); + }); + + test("emits awaiting-input notice for ask_user_question-style stages", () => { + const { store, sent } = install(); + store.recordRunStart({ id: "run-4", name: "qa", inputs: {}, status: "running", stages: [], startedAt: 1 }); + store.recordStageStart("run-4", runningStage({ id: "stage-ask", name: "question" })); + + assert.equal(store.recordStageAwaitingInput("run-4", "stage-ask", true, 123), true); + + assert.equal(sent.length, 1); + assert.equal(sent[0]?.details?.kind, "awaiting_input"); + assert.equal(sent[0]?.details?.scope, "stage"); + assert.equal(sent[0]?.details?.stageId, "stage-ask"); + assert.equal(sent[0]?.details?.createdAt, 123); + assert.match(sent[0]?.content ?? "", /Respond: \/workflow connect run-4\./); + assert.doesNotMatch(sent[0]?.content ?? "", /workflow\(\{ action: "send"/); + assert.doesNotMatch(sent[0]?.content ?? "", /promptId: ""/); + }); + + test("emits promptless awaiting-input after resolving a structured stage prompt", () => { + const { store, sent } = install(); + const runId = "run-stale-footprint"; + const stageId = "stage-mixed"; + + startRun(store, runId, "stale footprint"); + store.recordStageStart(runId, runningStage({ id: stageId, name: "mixed" })); + + assert.equal( + store.recordStagePendingPrompt( + runId, + stageId, + prompt({ id: "prompt-1", message: "Old structured prompt", createdAt: 10 }), + ), + true, + ); + assert.equal(store.resolveStagePendingPrompt(runId, stageId, "prompt-1", "accepted"), true); + assert.equal(store.recordStageAwaitingInput(runId, stageId, true, 123), true); + + const structuredPromptNotice = sent[0]?.details; + const promptlessNotice = sent[1]?.details; + + assert.equal(sent.length, 2); + assert.equal(structuredPromptNotice?.promptId, "prompt-1"); + assert.equal(promptlessNotice?.stageId, stageId); + assert.equal(promptlessNotice?.createdAt, 123); + assert.equal(promptlessNotice?.promptId, undefined); + assert.equal(promptlessNotice?.promptKind, undefined); + assert.equal(promptlessNotice?.promptMessage, undefined); + assert.doesNotMatch(sent[1]?.content ?? "", /Old structured prompt/); + }); + + test("dedupes repeated promptless pauses by awaitingInputSince instead of stale prompt footprint", () => { + const { store, sent } = install(); + const runId = "run-promptless-dedupe"; + const stageId = "stage-repeat"; + + startRun(store, runId, "promptless dedupe"); + store.recordStageStart(runId, runningStage({ id: stageId, name: "repeat" })); + + assert.equal( + store.recordStagePendingPrompt(runId, stageId, prompt({ id: "prompt-1", createdAt: 10 })), + true, + ); + assert.equal(store.resolveStagePendingPrompt(runId, stageId, "prompt-1", true), true); + assert.equal(store.recordStageAwaitingInput(runId, stageId, true, 123), true); + store.recordNotice({ id: "same-pause-tick", level: "info", message: "tick", createdAt: 124 }); + assert.equal(store.recordStageAwaitingInput(runId, stageId, false), true); + assert.equal(store.recordStageAwaitingInput(runId, stageId, true, 456), true); + + assert.deepEqual(sent.map((message) => message.details?.createdAt), [10, 123, 456]); + assert.deepEqual(sent.map((message) => message.details?.promptId), ["prompt-1", undefined, undefined]); + }); + + test("uses a new prompt id for a second structured stage prompt", () => { + const { store, sent } = install(); + const runId = "run-second-prompt"; + const stageId = "stage-structured"; + + startRun(store, runId, "second prompt"); + store.recordStageStart(runId, runningStage({ id: stageId, name: "structured" })); + + assert.equal( + store.recordStagePendingPrompt(runId, stageId, prompt({ id: "prompt-1", createdAt: 10 })), + true, + ); + assert.equal(store.resolveStagePendingPrompt(runId, stageId, "prompt-1", false), true); + assert.equal( + store.recordStagePendingPrompt( + runId, + stageId, + prompt({ id: "prompt-2", message: "New prompt", createdAt: 20 }), + ), + true, + ); + + assert.deepEqual(sent.map((message) => message.details?.promptId), ["prompt-1", "prompt-2"]); + assert.equal(sent[1]?.details?.createdAt, 20); + assert.equal(sent[1]?.details?.promptMessage, "New prompt"); + }); + + test("respects disabled and notifyOn filtering", () => { + const store = createStore(); + const sent: SentMessage[] = []; + installWorkflowLifecycleNotifications({ + store, + config: { enabled: true, notifyOn: ["failed"] }, + sendMessage(message) { sent.push(message as SentMessage); }, + }); + store.recordRunStart({ id: "run-5", name: "filtered", inputs: {}, status: "running", stages: [], startedAt: 1 }); + store.recordRunEnd("run-5", "completed", {}); + assert.equal(sent.length, 0); + + installWorkflowLifecycleNotifications({ + store, + config: { enabled: false, notifyOn: ["completed", "failed", "awaiting_input"] }, + sendMessage(message) { sent.push(message as SentMessage); }, + }); + store.recordRunStart({ id: "run-6", name: "disabled", inputs: {}, status: "running", stages: [], startedAt: 1 }); + store.recordRunEnd("run-6", "failed", undefined, "boom"); + assert.equal(sent.length, 1); + }); + + test("emits awaiting-input notice for a run-level pending prompt", () => { + const { store, sent } = install(); + startRun(store, "run-prompt", "legacy"); + + assert.equal(store.recordPendingPrompt("run-prompt", prompt({ id: "run-prompt-1" })), true); + + assert.equal(sent.length, 1); + assert.equal(sent[0]?.details?.kind, "awaiting_input"); + assert.equal(sent[0]?.details?.scope, "run"); + assert.equal(sent[0]?.details?.promptId, "run-prompt-1"); + assert.equal(sent[0]?.details?.stageId, undefined); + assert.match(sent[0]?.content ?? "", /run-level prompt/); + assert.doesNotMatch(sent[0]?.content ?? "", /stageId/); + }); + + test("suppresses run-level pending prompt when notifyOn excludes awaiting_input", () => { + const store = createStore(); + const sent: SentMessage[] = []; + installWorkflowLifecycleNotifications({ + store, + config: { enabled: true, notifyOn: ["completed", "failed"] }, + sendMessage(message) { sent.push(message as SentMessage); }, + }); + startRun(store, "run-filtered-prompt", "legacy filtered"); + + assert.equal(store.recordPendingPrompt("run-filtered-prompt", prompt({ id: "filtered-prompt" })), true); + + assert.equal(sent.length, 0); + }); + + test("shared state dedupes terminal notices across reinstall", () => { + const store = createStore(); + const state = createWorkflowLifecycleNotificationState(); + const sent: SentMessage[] = []; + const unsubscribe = installWithState(store, state, sent); + startRun(store, "run-dedupe", "dedupe"); + store.recordRunEnd("run-dedupe", "completed", {}); + unsubscribe(); + installWithState(store, state, sent); + startRun(store, "run-other", "other"); + + assert.deepEqual(sent.map((message) => message.details?.runId), ["run-dedupe"]); + }); + + test("omitted seedExisting treats current terminal runs and prompts as history", () => { + const store = createStore(); + startRun(store, "run-old", "old"); + store.recordRunEnd("run-old", "completed", {}); + startRun(store, "run-old-prompt", "old prompt"); + store.recordPendingPrompt("run-old-prompt", prompt({ id: "old-prompt" })); + + const sent: SentMessage[] = []; + installWorkflowLifecycleNotifications({ + store, + config, + state: createWorkflowLifecycleNotificationState(), + sendMessage(message) { sent.push(message as SentMessage); }, + }); + store.recordNotice({ id: "tick", level: "info", message: "tick", createdAt: 11 }); + startRun(store, "run-new", "new"); + store.recordRunEnd("run-new", "completed", {}); + + assert.deepEqual(sent.map((message) => message.details?.runId), ["run-new"]); + }); + + test("resetting shared state allows reused run IDs across session boundaries", () => { + const store = createStore(); + const state = createWorkflowLifecycleNotificationState(); + const sent: SentMessage[] = []; + let unsubscribe = installWithState(store, state, sent); + startRun(store, "run-reused", "first session"); + store.recordRunEnd("run-reused", "completed", {}); + unsubscribe(); + + store.clear(); + resetWorkflowLifecycleNotificationState(state); + unsubscribe = installWithState(store, state, sent); + startRun(store, "run-reused", "second session"); + store.recordRunEnd("run-reused", "completed", {}); + unsubscribe(); + + assert.deepEqual(sent.map((message) => message.details?.workflowName), ["first session", "second session"]); + }); + + test("restore suppression after reset seeds restored history without emitting", () => { + const store = createStore(); + const state = createWorkflowLifecycleNotificationState(); + const sent: SentMessage[] = []; + installWorkflowLifecycleNotifications({ + store, + config, + state, + sendMessage(message) { sent.push(message as SentMessage); }, + }); + + startRun(store, "run-before-reset", "before reset"); + store.recordRunEnd("run-before-reset", "completed", {}); + store.clear(); + resetWorkflowLifecycleNotificationState(state); + + const entries: SessionEntry[] = [ + { id: "e1", type: "workflow.run.start", payload: { runId: "run-restored-after-reset", name: "restored after reset", inputs: {}, ts: 1 } }, + { id: "e2", type: "workflow.run.end", payload: { runId: "run-restored-after-reset", status: "completed", result: {}, ts: 2 } }, + ]; + + withWorkflowLifecycleNotificationsSuppressed(state, () => { + restoreOnSessionStart({ getEntries: () => entries }, { resumeInFlight: "never", persistRuns: true }, store); + seedWorkflowLifecycleNotificationState(state, store.snapshot()); + }); + store.recordNotice({ id: "after-reset-restore", level: "info", message: "tick", createdAt: 12 }); + startRun(store, "run-live-after-reset", "live after reset"); + store.recordRunEnd("run-live-after-reset", "completed", {}); + + assert.deepEqual(sent.map((message) => message.details?.runId), ["run-before-reset", "run-live-after-reset"]); + }); + + test("suppression seeds actual restore replay without emitting", () => { + const store = createStore(); + const state = createWorkflowLifecycleNotificationState(); + const sent: SentMessage[] = []; + installWorkflowLifecycleNotifications({ + store, + config, + state, + sendMessage(message) { sent.push(message as SentMessage); }, + }); + const entries: SessionEntry[] = [ + { id: "e1", type: "workflow.run.start", payload: { runId: "run-restored", name: "restored", inputs: {}, ts: 1 } }, + { id: "e2", type: "workflow.run.end", payload: { runId: "run-restored", status: "failed", error: "old failure", ts: 2 } }, + ]; + + withWorkflowLifecycleNotificationsSuppressed(state, () => { + restoreOnSessionStart({ getEntries: () => entries }, { resumeInFlight: "never", persistRuns: true }, store); + }); + store.recordNotice({ id: "after-restore", level: "info", message: "tick", createdAt: 12 }); + startRun(store, "run-live", "live"); + store.recordRunEnd("run-live", "failed", undefined, "live failure"); + + assert.deepEqual(sent.map((message) => message.details?.runId), ["run-live"]); + }); + + test("escapes workflow names and structured response ids in notice text", () => { + const runId = 'run"\\id'; + const stageId = 'stage"\\id'; + const promptId = 'prompt"\\id'; + const text = formatWorkflowLifecycleNoticeText({ + kind: "awaiting_input", + scope: "stage", + runId, + workflowName: 'release "canary"', + status: "awaiting_input", + stageId, + stageName: 'review "gate"', + promptId, + promptKind: "confirm", + promptMessage: "Approve?", + createdAt: 1, + }); + + assert.match(text, /Workflow "release \\"canary\\"" needs input/); + assert.match(text, /workflow\(\{ action: "send"/); + assert.ok(text.includes(`runId: ${JSON.stringify(runId)}`)); + assert.ok(text.includes(`stageId: ${JSON.stringify(stageId)}`)); + assert.ok(text.includes(`promptId: ${JSON.stringify(promptId)}`)); + }); + + test("always triggers a steer turn for emitted lifecycle notices", () => { + const store = createStore(); + const options: SendOptions[] = []; + installWorkflowLifecycleNotifications({ + store, + config: { enabled: true, notifyOn: ["completed"] }, + sendMessage(_message, sendOptions) { options.push(sendOptions ?? {}); }, + }); + store.recordRunStart({ id: "run-7", name: "turn", inputs: {}, status: "running", stages: [], startedAt: 1 }); + store.recordRunEnd("run-7", "completed", {}); + assert.deepEqual(options, [{ triggerTurn: true, deliverAs: "steer" }]); + }); + + test("warns about send failures when workflow debug logging is enabled", () => { + const store = createStore(); + const previousDebug = process.env.ATOMIC_WORKFLOW_DEBUG; + const originalWarn = console.warn; + const warnings: unknown[][] = []; + process.env.ATOMIC_WORKFLOW_DEBUG = "1"; + console.warn = (...args: unknown[]) => { warnings.push(args); }; + try { + installWorkflowLifecycleNotifications({ + store, + config: { enabled: true, notifyOn: ["completed"] }, + sendMessage() { + throw new Error("send failed"); + }, + }); + store.recordRunStart({ id: "run-debug-throw", name: "debug", inputs: {}, status: "running", stages: [], startedAt: 1 }); + assert.equal(store.recordRunEnd("run-debug-throw", "completed", {}), true); + } finally { + console.warn = originalWarn; + if (previousDebug === undefined) { + delete process.env.ATOMIC_WORKFLOW_DEBUG; + } else { + process.env.ATOMIC_WORKFLOW_DEBUG = previousDebug; + } + } + + assert.equal(warnings.length, 1); + assert.match(String(warnings[0]?.[0] ?? ""), /workflow lifecycle notice/i); + assert.match(String(warnings[0]?.[1] ?? ""), /send failed/); + }); + + test("does not warn about send failures unless workflow debug logging is enabled", () => { + const store = createStore(); + const previousDebug = process.env.ATOMIC_WORKFLOW_DEBUG; + const originalWarn = console.warn; + const warnings: unknown[][] = []; + delete process.env.ATOMIC_WORKFLOW_DEBUG; + console.warn = (...args: unknown[]) => { warnings.push(args); }; + try { + installWorkflowLifecycleNotifications({ + store, + config: { enabled: true, notifyOn: ["completed"] }, + sendMessage() { + throw new Error("send failed"); + }, + }); + store.recordRunStart({ id: "run-debug-off", name: "debug off", inputs: {}, status: "running", stages: [], startedAt: 1 }); + assert.equal(store.recordRunEnd("run-debug-off", "completed", {}), true); + } finally { + console.warn = originalWarn; + if (previousDebug === undefined) { + delete process.env.ATOMIC_WORKFLOW_DEBUG; + } else { + process.env.ATOMIC_WORKFLOW_DEBUG = previousDebug; + } + } + + assert.equal(warnings.length, 0); + }); + + test("swallows synchronous send failures so sibling subscribers still receive snapshots", () => { + const store = createStore(); + const seenStatuses: string[] = []; + installWorkflowLifecycleNotifications({ + store, + config: { enabled: true, notifyOn: ["completed"] }, + sendMessage() { + throw new Error("send failed"); + }, + }); + const unsubscribeSibling = store.subscribe((snapshot) => { + const run = snapshot.runs.find((candidate) => candidate.id === "run-send-throw"); + if (run) seenStatuses.push(run.status); + }); + + store.recordRunStart({ id: "run-send-throw", name: "throw", inputs: {}, status: "running", stages: [], startedAt: 1 }); + assert.doesNotThrow(() => { + assert.equal(store.recordRunEnd("run-send-throw", "completed", {}), true); + }); + unsubscribeSibling(); + + assert.deepEqual(seenStatuses, ["running", "completed"]); + }); + + test("swallows rejected send promises without surfacing unhandled rejections", async () => { + const store = createStore(); + let siblingSawCompletion = false; + installWorkflowLifecycleNotifications({ + store, + config: { enabled: true, notifyOn: ["completed"] }, + sendMessage() { + return Promise.reject(new Error("send rejected")); + }, + }); + const unsubscribeSibling = store.subscribe((snapshot) => { + siblingSawCompletion ||= snapshot.runs.some( + (run) => run.id === "run-send-reject" && run.status === "completed", + ); + }); + + store.recordRunStart({ id: "run-send-reject", name: "reject", inputs: {}, status: "running", stages: [], startedAt: 1 }); + assert.equal(store.recordRunEnd("run-send-reject", "completed", {}), true); + await Promise.resolve(); + unsubscribeSibling(); + + assert.equal(siblingSawCompletion, true); + }); + + test("registers lifecycle renderer once per host and returns a notice card", () => { + const host = {}; + const registered: RegisteredRenderer[] = []; + registerLifecycleNoticeRenderer({ + rendererHost: host, + registerMessageRenderer(event, renderer) { + registered.push({ event, renderer: renderer as (payload: unknown) => unknown }); + }, + }); + registerLifecycleNoticeRenderer({ + rendererHost: host, + registerMessageRenderer(event, renderer) { + registered.push({ event, renderer: renderer as (payload: unknown) => unknown }); + }, + }); + + assert.equal(registered.length, 1); + assert.equal(registered[0]?.event, LIFECYCLE_NOTICE_CUSTOM_TYPE); + const rendered = registered[0]?.renderer({ + details: { + kind: "completed", + scope: "run", + runId: "run-card", + workflowName: "cards", + status: "completed", + createdAt: 1, + } satisfies WorkflowLifecycleNoticeDetails, + }); + + assert.equal(typeof rendered, "object"); + assert.notEqual(rendered, null); + assert.deepEqual((rendered as CardComponent).render(80), [ + '✅ Workflow "cards" completed (run run-card). Inspect: /workflow status run-card', + ]); + }); +});