From 0ad9b25d693fa5ee7e18543136d3b0de17d04e12 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Wed, 23 Sep 2026 13:12:22 -0700 Subject: [PATCH 1/3] fix(observer): project each run into its channel and scope the link to it The observer link a run printed opened an arbitrary workspace channel with nothing about the flow: no run published anything to Relaycast (only agent chat was projected), and the link was minted workspace-wide after the run had already finished. - kernel: run.start {watch: true} streams the new run's entries on the starting connection from run.spawned on, registered before the first append so each entry arrives exactly once. The only way to observe a run whose id the caller does not know yet. - sdk: a run projection publishes run start, every step transition and the terminal outcome into wf-, each message carrying the run snapshot under metadata.relayflow. YAML runs fold the journal stream; authored runs use the executor's progress events. Resume replays history silently. - cli: the observer token is scoped to the run's channel and printed on stderr as soon as the run exists, then again after RUN as before. - Fail open throughout: a projection or mint failure is one [observer] line; a daemon that predates watch is started again without it. - tests: isolate the agent-relay workspace store so no test publishes into a developer's real workspace. Co-Authored-By: Claude Opus 5.5 (1M context) --- kernel/DESIGN.md | 2 +- kernel/relayflowd/src/engine.rs | 24 +++ kernel/relayflowd/src/server.rs | 11 +- kernel/relayflowd/src/server/tests.rs | 60 +++++++ kernel/relayflowd/src/server/wire.rs | 3 + packages/sdk/src/authored-root.ts | 2 + packages/sdk/src/cli.ts | 46 ++--- packages/sdk/src/cli/direct-run.ts | 1 + packages/sdk/src/cli/observer-session.ts | 104 +++++++++++ packages/sdk/src/cli/run.ts | 40 ++++- packages/sdk/src/journal-client.ts | 9 +- packages/sdk/src/journal-projection.ts | 82 +++++++++ packages/sdk/src/observer-link.ts | 11 +- packages/sdk/src/protocol.ts | 2 + packages/sdk/src/run-projection.ts | 210 ++++++++++++++++++++++ packages/sdk/tests/isolate-workspace.ts | 12 ++ packages/sdk/tests/observer-link.test.ts | 73 ++++++++ packages/sdk/tests/run-projection.test.ts | 154 ++++++++++++++++ packages/sdk/vitest.config.ts | 1 + 19 files changed, 808 insertions(+), 39 deletions(-) create mode 100644 packages/sdk/src/cli/observer-session.ts create mode 100644 packages/sdk/src/journal-projection.ts create mode 100644 packages/sdk/src/run-projection.ts create mode 100644 packages/sdk/tests/isolate-workspace.ts create mode 100644 packages/sdk/tests/run-projection.test.ts diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index 30b9b8830..123f17c58 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -380,7 +380,7 @@ Minimal verb set for gate 1: | verb | params → result | purpose | |---|---|---| | `hello` | `{protocol: 0, client}` → `{protocol: 0, server}` | handshake; version mismatch is a hard error | -| `run.start` | `{spec}` → `{run_id}` | validate spec (zero-agent flows are legal; invalid declarations return `invalid_spec` before storage), create run file, append `run.spawned`, begin scheduling | +| `run.start` | `{spec, watch?}` → `{run_id}` | validate spec (zero-agent flows are legal; invalid declarations return `invalid_spec` before storage), create run file, append `run.spawned`, begin scheduling. `watch: true` pushes the new run's entries to this connection from `run.spawned` on, as `run.watch` does — the only way to observe a run whose id the caller does not know yet | | `run.resume` | `{run_id}` → `{run_id, state}` | §3 memoized resume | | `run.cancel` | `{run_id}` → `{run_id, status, completion_reason}` | append durable intent, close active leases, and append the terminal canceled fact; repeated calls return the existing outcome | | `run.get` | `{run_id}` → `{status, steps, budget}` | snapshot for legibility | diff --git a/kernel/relayflowd/src/engine.rs b/kernel/relayflowd/src/engine.rs index 708f5d824..fb7ae5302 100644 --- a/kernel/relayflowd/src/engine.rs +++ b/kernel/relayflowd/src/engine.rs @@ -256,6 +256,29 @@ impl Engine { options: DriveOptions, reuse_from_run_id: Option<&str>, admission_key: Option<&str>, + ) -> Result { + self.start_observed( + spec, + created_by, + options, + reuse_from_run_id, + admission_key, + &|_| {}, + ) + } + + /// `start_with_admission`, calling `before_first_append` with the new + /// run's id before its journal exists. A watcher registered there sees + /// every entry the run appends, from `run.spawned` on. It is not called + /// when admission returns an existing run. + pub fn start_observed( + &self, + spec: RunSpec, + created_by: &str, + options: DriveOptions, + reuse_from_run_id: Option<&str>, + admission_key: Option<&str>, + before_first_append: &dyn Fn(&str), ) -> Result { spec.validate().context("invalid run spec")?; let reuse = reuse_from_run_id @@ -285,6 +308,7 @@ impl Engine { } } + before_first_append(&run_id); let path = self.run_path(&run_id); let now_ms = self.clock.now_ms(); let started = (|| -> Result { diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index 360f9c092..fe0a63673 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -170,14 +170,23 @@ fn handle_request( .map_err(|error| ("invalid_spec", error.to_string()))?; spec.validate() .map_err(|error| ("invalid_spec", error.to_string()))?; + // Registered before the first append, so there is nothing to + // replay: the watcher goes live and receives every entry once. + let watch = |run_id: &str| { + if params.watch { + hub.watch(connection_id, run_id.to_owned(), writer.clone()); + hub.watch_ready(connection_id, run_id, 0); + } + }; to_value( engine - .start_with_admission( + .start_observed( spec, "protocol-v0", crate::DriveOptions::default(), params.reuse_from_run_id.as_deref(), params.admission_key.as_deref(), + &watch, ) .map_err(run_start_error)?, ) diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index c1f6e237b..cb24cff05 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -885,6 +885,66 @@ fn an_entry_appended_during_watch_registration_is_delivered_exactly_once() { ); } +/// `run.start` with `watch` streams the new run's entries on the starting +/// connection from `run.spawned` on, each exactly once, before the result — +/// the only moment a client that does not yet know the run id can observe it. +#[test] +fn run_start_with_watch_streams_every_entry_once_before_the_result() { + let directory = tempdir().unwrap(); + let data_dir = directory.path(); + let hub = Arc::new(ProtocolHub::default()); + let (writer, peer) = shared_writer(); + let spec = json!({"steps": [ + {"id": "a", "type": "deterministic", "command": ["/bin/sh", "-c", "printf a"]}, + {"id": "b", "type": "deterministic", "command": ["/bin/sh", "-c", "printf b"], "depends_on": ["a"]} + ]}); + let line = json!({"id": "start", "verb": "run.start", "params": {"spec": spec, "watch": true}}) + .to_string(); + let started = request(data_dir, &hub, 1, &writer, &line); + assert!(started.ok, "run.start failed: {:?}", started.error); + let run_id = started.result.unwrap()["run_id"].as_str().unwrap().to_owned(); + + let expected = Engine::new(data_dir) + .journal_entries(&run_id, 1, usize::MAX) + .unwrap(); + assert_eq!(expected.first().unwrap().entry_type, EntryType::RunSpawned); + assert_eq!(expected.last().unwrap().entry_type, EntryType::RunCompleted); + peer.set_read_timeout(Some(Duration::from_millis(500))).unwrap(); + let mut reader = BufReader::new(peer); + let seen = (0..expected.len()) + .map(|_| { + let frame = read_frame(&mut reader); + assert_eq!(frame["event"], "entry"); + assert_eq!(frame["data"]["run_id"], run_id.as_str()); + frame["data"]["seq"].as_i64().unwrap() + }) + .collect::>(); + let mut leftover = String::new(); + assert!( + reader.read_line(&mut leftover).is_err(), + "watcher received a duplicate frame: {leftover}" + ); + assert_eq!(seen, expected.iter().map(|entry| entry.seq).collect::>()); +} + +/// Without `watch`, `run.start` pushes nothing: the flag is opt-in. +#[test] +fn run_start_without_watch_pushes_no_entries() { + let directory = tempdir().unwrap(); + let data_dir = directory.path(); + let hub = Arc::new(ProtocolHub::default()); + let (writer, peer) = shared_writer(); + let spec = json!({"steps": [{"id": "a", "type": "deterministic", "command": ["/bin/sh", "-c", "printf a"]}]}); + let line = json!({"id": "start", "verb": "run.start", "params": {"spec": spec}}).to_string(); + assert!(request(data_dir, &hub, 1, &writer, &line).ok); + peer.set_read_timeout(Some(Duration::from_millis(200))).unwrap(); + let mut leftover = String::new(); + assert!( + BufReader::new(peer).read_line(&mut leftover).is_err(), + "an unwatched start pushed a frame: {leftover}" + ); +} + /// Finding 5: when the journal append for a disconnect's crashed completion /// fails, the abandonment is surfaced and retained for the reconciler — never /// silently dropped — and the reconciler journals it once the journal heals. diff --git a/kernel/relayflowd/src/server/wire.rs b/kernel/relayflowd/src/server/wire.rs index b06d8867c..ca8adf6ed 100644 --- a/kernel/relayflowd/src/server/wire.rs +++ b/kernel/relayflowd/src/server/wire.rs @@ -41,6 +41,9 @@ pub(super) struct RunStartParams { pub spec: Value, pub reuse_from_run_id: Option, pub admission_key: Option, + /// Stream the new run's entries to this connection, as `run.watch` does. + #[serde(default)] + pub watch: bool, } #[derive(Deserialize)] diff --git a/packages/sdk/src/authored-root.ts b/packages/sdk/src/authored-root.ts index 72b916d1c..bf2aefa49 100644 --- a/packages/sdk/src/authored-root.ts +++ b/packages/sdk/src/authored-root.ts @@ -111,6 +111,7 @@ export async function executeDurableAuthoredFlow( return await completedRootResult(journal, outcome.run_id); } assertRootCanDispatch(outcome); + options.lifecycle?.onRunStarted?.({ runId: outcome.run_id, flow: definition.name }); // `run.start` is an idempotent receipt. If the first caller died after // the daemon dispatched this root, a same-daemon retry sees the existing // active run but receives no second dispatch from start itself. Resume is @@ -157,6 +158,7 @@ export async function resumeDurableAuthoredFlow( } assertRootCanDispatch(outcome); await assertNoOpenHumanWait(journal, outcome); + options.lifecycle?.onRunStarted?.({ runId: rootRunId, flow: metadata.flowName, resumed: true }); const dispatch = await dispatchWait.promise; return await driveRoot(loaded, metadata, journal, peer, dispatch, options); } finally { diff --git a/packages/sdk/src/cli.ts b/packages/sdk/src/cli.ts index 8dc8e64e9..2952c6872 100644 --- a/packages/sdk/src/cli.ts +++ b/packages/sdk/src/cli.ts @@ -7,6 +7,8 @@ import { describeFlowRequirements } from './flow-requirements.js'; import type { CliModelSource } from './cli-adapter.js'; import { renderProgress, type ProgressEvent } from './progress.js'; +import type { JournalEvent } from './journal-reader.js'; +import { createObserverSession } from './cli/observer-session.js'; import { realpathSync } from 'node:fs'; import { pathToFileURL } from 'node:url'; import { @@ -287,9 +289,11 @@ export async function runCli( // not here. Hoisting it above the dispatch would start a daemon as a side // effect of an invocation that is about to be refused for bad input. const startedSteps = new Map(); + const observer = parsed.noObserverLink ? undefined : createObserverSession(parsed.command, io); const showProgress = (event: ProgressEvent): void => { if (event.type === 'step.started') startedSteps.set(event.stepId, performance.now()); if (!parsed.json) for (const line of renderProgress([event])) io.stderr(line); + observer?.onProgress(event); }; const lifecycle = { ...(parsed.command === 'run' ? { bucket: parsed.bucket } : {}), @@ -299,6 +303,10 @@ export async function runCli( localAgent: parsed.localAgent, ...(parsed.agentCapacity === undefined ? {} : { agentCapacity: parsed.agentCapacity }), onProgress: showProgress, + ...(observer === undefined ? {} : { + onJournalEntry: (entry: JournalEvent) => observer.onJournalEntry(entry), + onRunStarted: (run: { runId: string; flow: string; resumed?: boolean }) => observer.onRunStarted(run), + }), onWait: (progress: RunProgress) => { emitWait(progress, io); const now = performance.now(); @@ -308,16 +316,15 @@ export async function runCli( }, daemon: { spawn: parsed.spawn && spawnAllowedByEnv() }, }; - // Mint the observer token in parallel with the run so the mint round-trip - // never adds to the RUN summary latency. The outcome is only consulted at - // emit time; a rejected promise here can never fail the run (see - // `observerUrlFrom`, which swallows every failure into `warning`). - const observerMint = startObserverMint(parsed); const execution = parsed.command === 'run' ? isAuthoredFlowPath(parsed.value) ? await runDirectFlow(parsed.value, parsed.input, parsed.dataDir, lifecycle) : await runFlow(parsed.value, parsed.dataDir, lifecycle) : await resumeFlow(parsed.value, parsed.dataDir, lifecycle); + // The link is scoped to the run's channel, so it exists only once the run + // does. `finish` drains the projection and settles the mint, both bounded; + // it never rejects (see `createObserverSession`). + const observerMint = observer?.finish(execution.report); // In `--json` mode the report is a single machine-readable object that // MUST carry `observerUrl` when one is available, so a consumer sees one // authoritative signal. That justifies blocking up to `MINT_TIMEOUT_MS` @@ -369,35 +376,6 @@ async function checkAuthoredFlowComposed(path: string): Promise<{ report: CheckR }; } -/** - * Start the observer-token mint if the environment says one should happen. - * Returns `undefined` when no attempt should be made — no workspace key - * configured, or `FLOWS_NO_OBSERVER=1` / `--no-observer-link` — which is the - * silent-skip branch. The returned promise always resolves; a rejection here - * would slip past `observerUrlFrom` and could fail the run, which the feature - * expressly forbids. - */ -function startObserverMint( - parsed: { command: 'run' | 'resume'; noObserverLink: boolean }, - env: NodeJS.ProcessEnv = process.env, - mint: (options: MintObserverOptions) => Promise<{ observerUrl?: string; warning?: string }> = mintObserverUrl, -): Promise<{ observerUrl?: string; warning?: string }> | undefined { - if (parsed.noObserverLink) return undefined; - // `resolveObserverLinkEnv` (not `readObserverLinkEnv`) falls back to the - // `agent-relay cloud login` workspace store (~/.agentworkforce/relay/ - // workspaces.json) when RELAYCAST_WORKSPACE_KEY is unset. Env wins if set; - // FLOWS_NO_OBSERVER=1 still suppresses regardless of source. - const link = resolveObserverLinkEnv(env); - if (link.suppressed || link.workspaceKey === undefined) return undefined; - return mint({ - workspaceKey: link.workspaceKey, - ...(link.baseUrl !== undefined ? { baseUrl: link.baseUrl } : {}), - ...(link.dashboardUrl !== undefined ? { dashboardUrl: link.dashboardUrl } : {}), - }).catch((error) => ({ - warning: error instanceof Error ? error.message : 'unknown mint error', - })); -} - /** * Resolve the pending observer-mint into a URL (or nothing), and route any * mint warning to stderr under a `[observer]` label. Silent-skip (`undefined` diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index 97bb4732d..f33b6ea02 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -95,6 +95,7 @@ export async function runDirectFlow( ...(localAgent === undefined ? {} : { workerCapacity }), lifecycle: { onProgress: options.onProgress, + ...(options.onRunStarted !== undefined ? { onRunStarted: options.onRunStarted } : {}), ...(options.signal !== undefined ? { signal: options.signal } : {}), ...(options.onWait !== undefined ? { onWait: options.onWait } : {}), }, diff --git a/packages/sdk/src/cli/observer-session.ts b/packages/sdk/src/cli/observer-session.ts new file mode 100644 index 000000000..fe80a8ed2 --- /dev/null +++ b/packages/sdk/src/cli/observer-session.ts @@ -0,0 +1,104 @@ +/** + * The observer half of `flows run` / `flows resume`: once the run id is known, + * project the run into its `wf-` channel and mint a read-only link + * scoped to that channel, so the link opens on this run's step graph instead + * of an arbitrary workspace channel. The link is printed on stderr the moment + * it exists, so a human can follow the run live, and again after `RUN`. + * + * Best-effort throughout: no workspace key means no session, and a projection + * or mint failure is a labeled stderr line. The run never depends on it. + */ + +import type { CliIo } from '../cli.js'; +import { createJournalProjector } from '../journal-projection.js'; +import type { JournalEvent } from '../journal-reader.js'; +import { mintObserverUrl, resolveObserverLinkEnv, type MintObserverOptions } from '../observer-link.js'; +import type { ProgressEvent } from '../progress.js'; +import { + createRunProjection, runChannelName, type DeclaredStep, type ProjectionFetch, type RunProjection, +} from '../run-projection.js'; +import type { RunReport } from './run.js'; + +type MintOutcome = { observerUrl?: string; warning?: string }; + +/** Bounded so an unreachable Relaycast cannot hold the CLI open. */ +const DRAIN_GRACE_MS = 5_000; + +export interface ObserverSession { + /** YAML runs: every entry `run.start {watch}` / `run.watch` pushes. */ + onJournalEntry(entry: JournalEvent): void; + /** Authored runs: the root's id and name, known once it is admitted. */ + onRunStarted(run: { runId: string; flow: string; resumed?: boolean }): void; + /** Authored runs: step transitions from the body's executor. */ + onProgress(event: ProgressEvent): void; + /** Close the projection from the final report and settle the link. */ + finish(report: RunReport): Promise | undefined; +} + +export interface ObserverSessionDeps { + fetch?: ProjectionFetch; + mint?: (options: MintObserverOptions) => Promise; +} + +export function createObserverSession( + command: 'run' | 'resume', + io: CliIo, + env: NodeJS.ProcessEnv = process.env, + deps: ObserverSessionDeps = {}, +): ObserverSession | undefined { + const link = resolveObserverLinkEnv(env); + if (link.suppressed || link.workspaceKey === undefined) return undefined; + const workspaceKey = link.workspaceKey; + const mint = deps.mint ?? mintObserverUrl; + const liveSinceMs = command === 'resume' ? Date.now() : 0; + let projection: RunProjection | undefined; + let minted: Promise | undefined; + + const mintFor = (runId: string): Promise => minted ??= mint({ + workspaceKey, + channel: runChannelName(runId), + ...(link.baseUrl === undefined ? {} : { baseUrl: link.baseUrl }), + ...(link.dashboardUrl === undefined ? {} : { dashboardUrl: link.dashboardUrl }), + }).catch((error: unknown) => ({ warning: error instanceof Error ? error.message : 'unknown mint error' })); + + const open = (run: { runId: string; flow: string; steps?: DeclaredStep[]; resumed?: boolean }): RunProjection => { + if (projection !== undefined) return projection; + projection = createRunProjection({ + workspaceKey, + ...(link.baseUrl === undefined ? {} : { baseUrl: link.baseUrl }), + ...(deps.fetch === undefined ? {} : { fetch: deps.fetch }), + diagnostic: message => io.stderr(`[observer] ${message}`), + }, run); + void mintFor(run.runId).then(outcome => { + if (outcome.observerUrl !== undefined) io.stderr(`Observer: ${outcome.observerUrl}`); + }); + return projection; + }; + const project = createJournalProjector( + run => open({ ...run, ...(liveSinceMs > 0 ? { resumed: true } : {}) }), liveSinceMs, + ); + + return { + onJournalEntry(entry) { + try { project(entry); } catch (error) { + io.stderr(`[observer] could not project journal entry ${entry.seq}: ${error instanceof Error ? error.message : String(error)}`); + } + }, + onRunStarted: run => { open(run); }, + onProgress(event) { projection?.step(event); }, + finish(report) { + if (report.runId === undefined) return undefined; + if (projection === undefined) { + io.stderr('[observer] this run was not projected (the daemon did not stream its journal); ' + + 'the observer link opens an empty channel'); + } + projection?.finish({ + status: report.status === 'parked' ? 'parked' : report.completionReason === 'canceled' ? 'canceled' + : report.ok ? 'completed' : 'failed', + ...(report.completionReason === undefined ? {} : { completionReason: report.completionReason }), + }); + const drained = projection?.drain(DRAIN_GRACE_MS) ?? Promise.resolve(); + return drained.then(() => mintFor(report.runId!)); + }, + }; +} diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index f71cb8183..e323191aa 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -10,6 +10,7 @@ import { AuthoredFlowExecutionError, AuthoredHumanParked, type AuthoredHumanWait import { answerCommand, resumeCommand } from '../authored-human.js'; import { join, resolve } from 'node:path'; import type { ProgressEvent } from '../progress.js'; +import type { JournalEvent } from '../journal-reader.js'; import { toKernelSpec } from '../compile.js'; import { socketPathFor } from '../daemon-connection.js'; import { ensureDaemon, type EnsureDaemonOptions } from '../daemon-lifecycle.js'; @@ -102,6 +103,10 @@ export interface RunLifecycleOptions { onPtyReady?: (path: string) => void; reuseFromRunId?: string; onProgress?: (event: ProgressEvent) => void; + /** Stream the run's journal entries while it runs (YAML runs, via `run.start {watch}` / `run.watch`). */ + onJournalEntry?: (entry: JournalEvent) => void; + /** An authored root was admitted: its id is known before its body runs. */ + onRunStarted?: (run: { runId: string; flow: string; resumed?: boolean }) => void; localAgent?: boolean; /** `--agent-capacity`: the local workers' concurrency; the default is `DEFAULT_LOCAL_AGENT_CAPACITY`. */ agentCapacity?: number; @@ -169,7 +174,7 @@ async function executeCheckedFlow( const { attachCommunicationWorkers } = await import('../communication/local.js'); communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir); } - const outcome = await client.runStart(spec, options.reuseFromRunId); + const outcome = await startWatched(client, spec, options); const execution = await classifyOutcome(client, 'run', outcome, base, socketPath, { ...options, dataDir }); if (options.reuseFromRunId !== undefined) { execution.report.reuse = await reuseSummary(client, outcome.run_id, options.reuseFromRunId); @@ -193,6 +198,31 @@ async function executeCheckedFlow( } } +/** + * `run.start`, streaming the run's entries to `onJournalEntry` when one is + * given. A daemon that predates `watch` refuses the field while decoding, + * before any run exists, so starting again without it is the same request + * minus the observation — never a second run. + */ +async function startWatched( + client: JournalClient, + spec: ReturnType, + options: RunLifecycleOptions, +): Promise { + const onEntry = options.onJournalEntry; + if (onEntry === undefined) return client.runStart(spec, options.reuseFromRunId); + client.on('entry', onEntry); + try { + return await client.runStart(spec, options.reuseFromRunId, undefined, true); + } catch (error) { + if (!(error instanceof JournalProtocolError) || error.code !== 'bad_request' + || !/unknown field `watch`/.test(error.message)) throw error; + return await client.runStart(spec, options.reuseFromRunId); + } finally { + client.off('entry', onEntry); + } +} + export async function resumeFlow( runId: string, dataDir: string, @@ -255,10 +285,18 @@ export async function resumeFlow( communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir); } } + const onEntry = options.onJournalEntry; + if (onEntry !== undefined) { + client.on('entry', onEntry); + // Observation never decides a resume: a watch the daemon refuses leaves + // the run unprojected, and the resume below reports the run's own fate. + await client.runWatch(runId).catch(() => client.off('entry', onEntry)); + } let outcome = await client.runResume(runId, options.allowHumanInfluenced); if (await resumeHelperEffect(client, runId, dataDir)) { outcome = await client.runResume(runId, options.allowHumanInfluenced); } + if (onEntry !== undefined) client.off('entry', onEntry); return await classifyOutcome(client, 'resume', outcome, base, socketPath, { ...options, dataDir }); } catch (error) { if (error instanceof CommunicationEnvironmentError) return { exitCode: 2, report: { ...base, runId, socketPath, diff --git a/packages/sdk/src/journal-client.ts b/packages/sdk/src/journal-client.ts index 378f722de..df294301b 100644 --- a/packages/sdk/src/journal-client.ts +++ b/packages/sdk/src/journal-client.ts @@ -224,12 +224,19 @@ export class JournalClient extends EventEmitter { * Validate a compiled kernel-dialect spec (zero-agent flows legal), create * the run file, and append `run.spawned`. Authoring specs must be compiled * with `toKernelSpec` before crossing this journal-protocol boundary. + * `watch` streams the run's entries as `'entry'` events while it runs. */ - runStart(spec: KernelRunSpec, reuseFromRunId?: string, admissionKey?: string): Promise { + runStart( + spec: KernelRunSpec, + reuseFromRunId?: string, + admissionKey?: string, + watch = false, + ): Promise { return this.request('run.start', { spec, ...(reuseFromRunId === undefined ? {} : { reuse_from_run_id: reuseFromRunId }), ...(admissionKey === undefined ? {} : { admission_key: admissionKey }), + ...(watch ? { watch: true } : {}), }, null); } diff --git a/packages/sdk/src/journal-projection.ts b/packages/sdk/src/journal-projection.ts new file mode 100644 index 000000000..093206f1f --- /dev/null +++ b/packages/sdk/src/journal-projection.ts @@ -0,0 +1,82 @@ +/** + * Fold a run's journal entries, as `run.start {watch}` or `run.watch` pushes + * them, into a `RunProjection`: `run.spawned` opens it with the declared step + * graph, attempt starts and completions become step transitions, and + * `run.completed` closes it. Entries journaled before `liveSinceMs` (a resume + * replaying history) update the snapshot without publishing a message each. + */ + +import type { JournalEvent } from './journal-reader.js'; +import type { ProgressEvent } from './progress.js'; +import type { DeclaredStep, RunProjection, RunSnapshot } from './run-projection.js'; +import type { StepType } from './spec.js'; + +export type OpenProjection = (run: { runId: string; flow: string; steps: DeclaredStep[] }) => RunProjection; + +export function createJournalProjector(open: OpenProjection, liveSinceMs = 0): (entry: JournalEvent) => void { + let projection: RunProjection | undefined; + const types = new Map(); + const started = new Map(); + + return entry => { + const payload = (entry.payload ?? {}) as Record; + const live = entry.at_ms >= liveSinceMs; + if (entry.entry_type === 'run.spawned') { + if (projection !== undefined) return; + const spec = (payload['spec'] ?? {}) as { name?: unknown; steps?: unknown }; + const steps = (Array.isArray(spec.steps) ? spec.steps : []).flatMap(declaredStep); + for (const step of steps) types.set(step.id, step.type); + projection = open({ runId: entry.run_id, flow: typeof spec.name === 'string' ? spec.name : 'flow', steps }); + return; + } + if (projection === undefined || entry.step_id === null && entry.entry_type !== 'run.completed') return; + const stepId = entry.step_id ?? ''; + const stepType = types.get(stepId) ?? 'deterministic'; + const attempt = entry.attempt ?? undefined; + const key = `${stepId}#${attempt ?? 0}`; + const event = (type: ProgressEvent['type'], completionReason?: string): void => { + const elapsedMs = type === 'step.started' ? 0 : entry.at_ms - (started.get(key) ?? entry.at_ms); + projection!.step({ + type, stepId, stepType, elapsedMs, + ...(attempt === undefined ? {} : { attempt }), + ...(completionReason === undefined ? {} : { completionReason: completionReason as never }), + }, live); + }; + switch (entry.entry_type) { + case 'step.attempt.started': + started.set(key, entry.at_ms); + event('step.started'); + return; + case 'wait.human': + event('step.parked'); + return; + case 'step.completed': { + const reason = typeof payload['completionReason'] === 'string' ? payload['completionReason'] : undefined; + const parked = payload['disposition'] === 'park'; + event(parked ? 'step.parked' : reason === 'success' ? 'step.completed' : 'step.failed', reason); + return; + } + case 'run.completed': { + const reason = typeof payload['completionReason'] === 'string' ? payload['completionReason'] : undefined; + projection.finish({ status: runStatus(reason), ...(reason === undefined ? {} : { completionReason: reason }) }); + return; + } + } + }; +} + +function declaredStep(value: unknown): DeclaredStep[] { + if (typeof value !== 'object' || value === null) return []; + const step = value as { id?: unknown; type?: unknown; depends_on?: unknown }; + if (typeof step.id !== 'string') return []; + const type = step.type === 'llm' || step.type === 'agent' ? step.type : 'deterministic'; + const dependsOn = Array.isArray(step.depends_on) + ? step.depends_on.filter((id): id is string => typeof id === 'string') : []; + return [{ id: step.id, type, dependsOn }]; +} + +function runStatus(reason: string | undefined): RunSnapshot['status'] { + if (reason === 'success') return 'completed'; + if (reason === 'canceled') return 'canceled'; + return 'failed'; +} diff --git a/packages/sdk/src/observer-link.ts b/packages/sdk/src/observer-link.ts index ca7a1ce20..ff00f7d99 100644 --- a/packages/sdk/src/observer-link.ts +++ b/packages/sdk/src/observer-link.ts @@ -238,6 +238,13 @@ export interface MintObserverOptions { * (default `https://agentrelay.com`). Separate axis from `baseUrl` — the * mint API lives on a different subdomain from the dashboard. */ dashboardUrl?: string; + /** + * Scope the token to this one channel, so the dashboard opens on it — the + * run's `wf-` channel — rather than on whichever channel the + * workspace lists first. DMs are excluded: a run projects its + * conversation into that channel. Omitted: the whole workspace. + */ + channel?: string; fetch?: ObserverFetch; now?: () => number; /** Called for the token's uniquely-suffixed name; injectable for tests. */ @@ -299,7 +306,9 @@ export async function mintObserverUrl( name: `flows-run-${uuid}`, description: 'Auto-minted by `flows run` for the observer dashboard link.', scopes: OBSERVER_SCOPES, - filters: { include_dms: true }, + filters: options.channel === undefined + ? { include_dms: true } + : { channel_names: [options.channel], include_dms: false }, expires_at: new Date(now() + OBSERVER_TOKEN_TTL_MS).toISOString(), }; diff --git a/packages/sdk/src/protocol.ts b/packages/sdk/src/protocol.ts index 21f8b3423..a5176c0c1 100644 --- a/packages/sdk/src/protocol.ts +++ b/packages/sdk/src/protocol.ts @@ -96,6 +96,8 @@ export type StepUsage = export interface RunStartParams { /** Caller-owned retry identity. Reuse with a different spec is refused. */ admission_key?: string; + /** Push the new run's entries to this connection from `run.spawned` on, as `run.watch` does. */ + watch?: boolean; reuse_from_run_id?: string; /** * The kernel spec dialect — the ONE boundary shape `RunSpec::parse` diff --git a/packages/sdk/src/run-projection.ts b/packages/sdk/src/run-projection.ts new file mode 100644 index 000000000..9052901ca --- /dev/null +++ b/packages/sdk/src/run-projection.ts @@ -0,0 +1,210 @@ +/** + * Project a run's lifecycle into its Relaycast channel, `wf-`, so the + * observer dashboard shows the flow: one message per step transition, each + * carrying the whole run snapshot under `metadata.relayflow` for a renderer + * that draws the step graph. + * + * A projection, never a source of truth (RFC-0001 settled decision 7): the + * journal is the record. Publication is serialized and fire-and-forget; a + * failure is reported once through `diagnostic` and never reaches the run. + */ + +import { randomUUID } from 'node:crypto'; +import { renderProgress, type ProgressEvent } from './progress.js'; +import type { StepType } from './spec.js'; + +export const RELAYFLOW_METADATA_VERSION = 1; + +export type ProjectedStepState = 'pending' | 'running' | 'completed' | 'failed' | 'parked'; + +export interface ProjectedStep { + id: string; + type: StepType; + dependsOn: string[]; + state: ProjectedStepState; + attempt?: number; + elapsedMs?: number; + completionReason?: string; +} + +export interface RunSnapshot { + runId: string; + flow: string; + status: 'running' | 'completed' | 'failed' | 'parked' | 'canceled'; + completionReason?: string; + steps: ProjectedStep[]; +} + +export interface DeclaredStep { + id: string; + type: StepType; + dependsOn?: string[]; +} + +export interface RunProjection { + readonly channel: string; + /** Apply a transition; `publish: false` folds replayed history into the snapshot silently. */ + step(event: ProgressEvent & { attempt?: number }, publish?: boolean): void; + /** Close the run; only the first call publishes. */ + finish(outcome: { status: RunSnapshot['status']; completionReason?: string }): void; + /** Resolves when every queued publication settled, or after `timeoutMs`. */ + drain(timeoutMs: number): Promise; +} + +export type ProjectionFetch = (url: string, init: { + method: string; + headers: Record; + body?: string; + signal?: AbortSignal; +}) => Promise<{ ok: boolean; status: number; json: () => Promise }>; + +export interface RunProjectionOptions { + workspaceKey: string; + /** Relaycast API base, default `https://cast.agentrelay.com`. */ + baseUrl?: string; + fetch?: ProjectionFetch; + diagnostic: (message: string) => void; +} + +const DEFAULT_BASE_URL = 'https://cast.agentrelay.com'; +const REQUEST_TIMEOUT_MS = 5_000; + +export function runChannelName(runId: string): string { + return `wf-${runId.toLowerCase()}`; +} + +export function createRunProjection( + options: RunProjectionOptions, + run: { runId: string; flow: string; steps?: DeclaredStep[]; resumed?: boolean }, +): RunProjection { + const doFetch = options.fetch ?? (globalThis.fetch as unknown as ProjectionFetch); + const base = options.baseUrl ?? DEFAULT_BASE_URL; + const channel = runChannelName(run.runId); + const snapshot: RunSnapshot = { + runId: run.runId, flow: run.flow, status: 'running', + steps: (run.steps ?? []).map(step => ({ + id: step.id, type: step.type, dependsOn: step.dependsOn ?? [], state: 'pending', + })), + }; + + const call = async (path: string, token: string, body?: unknown, idempotencyKey?: string): Promise => { + const response = await doFetch(new URL(`/v1${path}`, base).toString(), { + method: body === undefined ? 'GET' : 'POST', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + ...(idempotencyKey === undefined ? {} : { 'Idempotency-Key': idempotencyKey }), + }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS), + }); + const parsed = await response.json().catch(() => undefined) as + { data?: unknown; error?: { code?: string } } | undefined; + if (!response.ok) { + throw new RelaycastError(response.status, parsed?.error?.code ?? `http_${response.status}`); + } + return parsed?.data; + }; + + // A per-session publisher: agent names are unique per workspace, and a + // resumed run publishes from a fresh session into the same channel. + const setup = (async (): Promise => { + const agent = await call('/agents', options.workspaceKey, { + name: `${slug(run.flow)}-${randomUUID().slice(0, 6)}`, type: 'agent', auto_join_general: false, + }) as { token?: unknown } | undefined; + if (typeof agent?.token !== 'string') throw new RelaycastError(0, 'agent_token_missing'); + const token = agent.token; + try { + await call('/channels', token, { name: channel, topic: `relayflow ${run.flow} · run ${run.runId}` }); + } catch (error) { + // Agent communication in the same run may have created it first. + if (!(error instanceof RelaycastError) || error.code !== 'channel_already_exists') throw error; + await call(`/channels/${encodeURIComponent(channel)}/join`, token, {}); + } + return token; + })(); + + let queue: Promise = setup; + let failed = false; + let sequence = 0; + const sessionId = randomUUID().slice(0, 8); + const publish = (event: string, text: string): void => { + const data = { relayflow: { version: RELAYFLOW_METADATA_VERSION, event, run: structuredClone(snapshot) } }; + const key = `${sessionId}-${++sequence}`; + queue = queue.then(async () => { + if (failed) return; + await call(`/channels/${encodeURIComponent(channel)}/messages`, await setup, { text, data }, key); + }).catch((error: unknown) => { + if (failed) return; + failed = true; + options.diagnostic(`run projection to #${channel} failed (${errorMessage(error)}); ` + + 'the observer will not show this run, which is unaffected'); + }); + }; + + const declared = snapshot.steps.length; + publish('run.started', `▶ ${run.flow} ${run.resumed ? 'resumed' : 'started'}` + + `${declared > 0 ? ` · ${declared} step${declared === 1 ? '' : 's'}` : ''} · run ${run.runId}`); + + return { + channel, + step(event, shouldPublish = true) { + let step = snapshot.steps.find(candidate => candidate.id === event.stepId); + if (step === undefined) { + step = { id: event.stepId, type: event.stepType, dependsOn: [], state: 'pending' }; + snapshot.steps.push(step); + } + step.state = STATE[event.type]; + if (event.attempt !== undefined) step.attempt = event.attempt; + if (event.type === 'step.started') { + delete step.elapsedMs; + delete step.completionReason; + } else { + step.elapsedMs = Math.max(0, Math.round(event.elapsedMs)); + } + if (event.completionReason !== undefined) step.completionReason = event.completionReason; + if (!shouldPublish || event.type === 'step.running') return; + publish(event.type, event.type === 'step.started' + ? `○ ${step.id} (${step.type}) started${step.attempt !== undefined && step.attempt > 1 ? ` · attempt ${step.attempt}` : ''}` + : renderProgress([event])[0]!); + }, + finish(outcome) { + if (snapshot.status !== 'running') return; + snapshot.status = outcome.status; + if (outcome.completionReason !== undefined) snapshot.completionReason = outcome.completionReason; + const icon = outcome.status === 'completed' ? '■' : outcome.status === 'parked' ? '⏸' : '✗'; + publish('run.completed', `${icon} ${run.flow} ${outcome.status}` + + `${outcome.completionReason === undefined ? '' : ` · completionReason: ${outcome.completionReason}`}`); + }, + async drain(timeoutMs) { + let timer: NodeJS.Timeout | undefined; + await Promise.race([ + queue, + new Promise(resolve => { timer = setTimeout(resolve, timeoutMs); timer.unref?.(); }), + ]); + clearTimeout(timer); + }, + }; +} + +const STATE: Record = { + 'step.started': 'running', + 'step.running': 'running', + 'step.completed': 'completed', + 'step.failed': 'failed', + 'step.parked': 'parked', +}; + +class RelaycastError extends Error { + constructor(readonly status: number, readonly code: string) { + super(status === 0 ? code : `HTTP ${status} ${code}`); + } +} + +function slug(name: string): string { + return name.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 40) || 'relayflow'; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/packages/sdk/tests/isolate-workspace.ts b/packages/sdk/tests/isolate-workspace.ts new file mode 100644 index 000000000..d1a03fe06 --- /dev/null +++ b/packages/sdk/tests/isolate-workspace.ts @@ -0,0 +1,12 @@ +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +// A developer logged in with `agent-relay` has a real workspace key in +// ~/.agentworkforce/relay/workspaces.json, which `flows run` reads to project +// runs into Relaycast. No test may publish into that workspace: point the +// store at an empty directory and clear the env key, for this process and the +// CLI children that inherit its environment. Tests that exercise the observer +// stub their own key. +process.env['AGENT_RELAY_HOME'] = mkdtempSync(join(tmpdir(), 'flows-test-relay-home-')); +delete process.env['RELAYCAST_WORKSPACE_KEY']; diff --git a/packages/sdk/tests/observer-link.test.ts b/packages/sdk/tests/observer-link.test.ts index 129fd8151..4f6543282 100644 --- a/packages/sdk/tests/observer-link.test.ts +++ b/packages/sdk/tests/observer-link.test.ts @@ -540,6 +540,79 @@ describe('flows run: observer link integration', () => { expect(fetch).toHaveBeenCalledOnce(); }); + it('streams the run into wf- and scopes the link to that channel', async () => { + const dataDir = temporaryProject(); + const runId = '01RUNOBSERVED'; + const entry = (seq: number, entry_type: string, step_id: string | null, payload: unknown = {}) => ({ + event: 'entry', + data: { seq, segment_id: 1, entry_type, run_id: runId, step_id, attempt: step_id === null ? null : 1, at_ms: 1_000 + seq, payload }, + }); + let watched: unknown; + await startCliLoopback(dataDir, { + hello: sendOk, + 'run.start': (ctx, params) => { + watched = params['watch']; + ctx.send(entry(1, 'run.spawned', null, { spec: { name: 'hello-deterministic', steps: [ + { id: 'greet', type: 'deterministic', depends_on: [] }, + ] } })); + ctx.send(entry(2, 'step.attempt.started', 'greet')); + ctx.send(entry(3, 'step.completed', 'greet', { completionReason: 'success', disposition: 'step_done' })); + ctx.send(entry(4, 'run.completed', null, { completionReason: 'success' })); + sendResult(ctx, { run_id: runId, status: 'completed', completion_reason: 'success', completed_steps: 1 }); + }, + }); + const requests: Array<{ path: string; body: Record }> = []; + vi.stubGlobal('fetch', vi.fn(async (url: string, init: { body?: string }) => { + const path = new URL(url).pathname; + requests.push({ path, body: init.body === undefined ? {} : JSON.parse(init.body) as Record }); + if (path === '/v1/observer-tokens') return jsonResponse(200, { data: { token: 'ot_live_scoped' } }); + if (path === '/v1/agents') return jsonResponse(201, { data: { token: 'at_live_pub' } }); + return jsonResponse(200, { data: {} }); + })); + vi.stubEnv('RELAYCAST_WORKSPACE_KEY', 'rk_live_operator'); + vi.stubEnv('FLOWS_NO_OBSERVER', ''); + + const output = capture(); + expect(await runCli(RUN_ARGS(dataDir), output.io)).toBe(0); + + expect(watched).toBe(true); + const link = 'Observer: https://agentrelay.com/observer?key=ot_live_scoped'; + expect(output.stderr).toContain(link); + expect(output.stdout).toContain(link); + const mint = requests.find(request => request.path === '/v1/observer-tokens'); + expect(mint?.body['filters']).toEqual({ channel_names: [`wf-${runId.toLowerCase()}`], include_dms: false }); + const posts = requests.filter(request => request.path === `/v1/channels/wf-${runId.toLowerCase()}/messages`); + expect(posts.map(post => (post.body['data'] as { relayflow: { event: string } }).relayflow.event)) + .toEqual(['run.started', 'step.started', 'step.completed', 'run.completed']); + }); + + it('starts again without watch when the daemon predates it, and says the run was not projected', async () => { + const dataDir = temporaryProject(); + const starts: unknown[] = []; + await startCliLoopback(dataDir, { + hello: sendOk, + 'run.start': (ctx, params) => { + starts.push(params['watch']); + if (params['watch'] === true) { + ctx.send({ id: ctx.id, ok: false, error: { code: 'bad_request', + message: 'unknown field `watch`, expected one of `spec`, `reuse_from_run_id`, `admission_key`' } }); + return; + } + sendResult(ctx, { run_id: 'run-old-daemon', status: 'completed', completion_reason: 'success', completed_steps: 2 }); + }, + }); + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(jsonResponse(200, { data: { token: 'ot_live_old' } }))); + vi.stubEnv('RELAYCAST_WORKSPACE_KEY', 'rk_live_operator'); + vi.stubEnv('FLOWS_NO_OBSERVER', ''); + + const output = capture(); + expect(await runCli(RUN_ARGS(dataDir), output.io)).toBe(0); + + expect(starts).toEqual([true, undefined]); + expect(output.stdout.some(line => line.startsWith('RUN run-old-daemon completed'))).toBe(true); + expect(output.stderr.some(line => line.includes('this run was not projected'))).toBe(true); + }); + it('emits no observer line and no fetch when no workspace key is set', async () => { const dataDir = temporaryProject(); await startCliLoopback(dataDir, { diff --git a/packages/sdk/tests/run-projection.test.ts b/packages/sdk/tests/run-projection.test.ts new file mode 100644 index 000000000..2fb3d90f2 --- /dev/null +++ b/packages/sdk/tests/run-projection.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it, vi } from 'vitest'; +import { createJournalProjector } from '../src/journal-projection.js'; +import type { JournalEvent } from '../src/journal-reader.js'; +import { createRunProjection, type ProjectionFetch } from '../src/run-projection.js'; + +type Call = { url: string; method: string; body: Record | undefined; headers: Record }; + +/** A Relaycast stand-in: records every request, answers by path. */ +function relaycast(overrides: Record = {}) { + const calls: Call[] = []; + const fetch: ProjectionFetch = vi.fn(async (url, init) => { + const path = new URL(url).pathname; + calls.push({ url: path, method: init.method, headers: init.headers, + body: init.body === undefined ? undefined : JSON.parse(init.body) as Record }); + const override = overrides[path]; + const status = override?.status ?? (path === '/v1/agents' ? 201 : 200); + const body = override?.body ?? (path === '/v1/agents' ? { ok: true, data: { token: 'at_live_pub' } } : { ok: true, data: {} }); + return { ok: status < 300, status, json: async () => body }; + }); + return { fetch, calls }; +} + +const run = { runId: '01RUN', flow: 'hello-deterministic', steps: [ + { id: 'greet', type: 'deterministic' as const, dependsOn: [] }, + { id: 'shout', type: 'deterministic' as const, dependsOn: ['greet'] }, +] }; + +describe('createRunProjection', () => { + it('registers a publisher, opens wf-, and posts each transition with the run snapshot', async () => { + const { fetch, calls } = relaycast(); + const projection = createRunProjection({ workspaceKey: 'rk_live_k', fetch, diagnostic: vi.fn() }, run); + projection.step({ type: 'step.started', stepId: 'greet', stepType: 'deterministic', elapsedMs: 0, attempt: 1 }); + projection.step({ type: 'step.completed', stepId: 'greet', stepType: 'deterministic', elapsedMs: 12.4, completionReason: 'success' }); + projection.finish({ status: 'completed', completionReason: 'success' }); + projection.finish({ status: 'failed' }); + await projection.drain(1_000); + + expect(calls.map(call => `${call.method} ${call.url}`)).toEqual([ + 'POST /v1/agents', + 'POST /v1/channels', + 'POST /v1/channels/wf-01run/messages', + 'POST /v1/channels/wf-01run/messages', + 'POST /v1/channels/wf-01run/messages', + 'POST /v1/channels/wf-01run/messages', + ]); + expect(calls[0]!.headers['Authorization']).toBe('Bearer rk_live_k'); + expect(calls[1]!.headers['Authorization']).toBe('Bearer at_live_pub'); + const posts = calls.slice(2).map(call => call.body as { text: string; data: { relayflow: { event: string; run: { + status: string; steps: Array<{ id: string; state: string; elapsedMs?: number }> } } } }); + expect(posts.map(post => post.text)).toEqual([ + '▶ hello-deterministic started · 2 steps · run 01RUN', + '○ greet (deterministic) started', + '✓ greet (deterministic) 0.01s completionReason: success', + '■ hello-deterministic completed · completionReason: success', + ]); + expect(posts.map(post => post.data.relayflow.event)).toEqual(['run.started', 'step.started', 'step.completed', 'run.completed']); + expect(posts[0]!.data.relayflow.run.steps.map(step => step.state)).toEqual(['pending', 'pending']); + expect(posts[2]!.data.relayflow.run.steps[0]).toMatchObject({ id: 'greet', state: 'completed', elapsedMs: 12 }); + expect(posts[3]!.data.relayflow.run.status).toBe('completed'); + // Every post carries its own idempotency key. + expect(new Set(calls.slice(2).map(call => call.headers['Idempotency-Key'])).size).toBe(4); + }); + + it('joins the channel when agent communication created it first', async () => { + const { fetch, calls } = relaycast({ '/v1/channels': { status: 409, body: { ok: false, error: { code: 'channel_already_exists' } } } }); + const projection = createRunProjection({ workspaceKey: 'rk_live_k', fetch, diagnostic: vi.fn() }, run); + await projection.drain(1_000); + expect(calls.map(call => call.url)).toEqual(['/v1/agents', '/v1/channels', '/v1/channels/wf-01run/join', '/v1/channels/wf-01run/messages']); + }); + + it('reports a failure once, stops publishing, and never throws into the run', async () => { + const { fetch, calls } = relaycast({ '/v1/agents': { status: 401, body: { ok: false, error: { code: 'unauthorized' } } } }); + const diagnostic = vi.fn(); + const projection = createRunProjection({ workspaceKey: 'rk_live_bad', fetch, diagnostic }, run); + projection.step({ type: 'step.started', stepId: 'greet', stepType: 'deterministic', elapsedMs: 0 }); + projection.finish({ status: 'completed' }); + await projection.drain(1_000); + expect(calls).toHaveLength(1); + expect(diagnostic).toHaveBeenCalledOnce(); + expect(diagnostic.mock.calls[0]![0]).toContain('HTTP 401 unauthorized'); + }); +}); + +describe('createJournalProjector', () => { + const entry = (seq: number, entry_type: string, at_ms: number, step_id: string | null = null, + payload: unknown = {}, attempt: number | null = step_id === null ? null : 1): JournalEvent => + ({ seq, segment_id: 1, entry_type, run_id: '01RUN', step_id, attempt, at_ms, payload }); + const spawned = entry(1, 'run.spawned', 1_000, null, { spec: { name: 'hello', steps: [ + { id: 'greet', type: 'deterministic', depends_on: [] }, + { id: 'plan', type: 'llm', depends_on: ['greet'] }, + ] } }); + + function recorder() { + const opened: unknown[] = []; + const steps: Array<{ type: string; stepId: string; stepType: string; elapsedMs: number; completionReason?: string; publish: boolean }> = []; + const finished: unknown[] = []; + const open = vi.fn((declared: unknown) => { + opened.push(declared); + return { + channel: 'wf-01run', + step: (event: { type: string; stepId: string; stepType: string; elapsedMs: number; completionReason?: string }, publish = true) => + { steps.push({ ...event, publish }); }, + finish: (outcome: unknown) => { finished.push(outcome); }, + drain: async () => {}, + }; + }); + return { open, opened, steps, finished }; + } + + it('opens on run.spawned with the declared graph and maps attempts and completions', () => { + const r = recorder(); + const project = createJournalProjector(r.open); + for (const e of [ + spawned, + entry(2, 'step.routed', 1_000, 'greet'), + entry(3, 'step.attempt.started', 1_010, 'greet'), + entry(4, 'step.completed', 1_260, 'greet', { completionReason: 'success', disposition: 'step_done' }), + entry(5, 'step.attempt.started', 1_300, 'plan'), + entry(6, 'step.completed', 1_500, 'plan', { completionReason: 'verification_failed', disposition: 'retry' }), + entry(7, 'run.completed', 1_600, null, { completionReason: 'step_failed' }), + ]) project(e); + + expect(r.opened).toEqual([{ runId: '01RUN', flow: 'hello', steps: [ + { id: 'greet', type: 'deterministic', dependsOn: [] }, + { id: 'plan', type: 'llm', dependsOn: ['greet'] }, + ] }]); + expect(r.steps.map(s => [s.type, s.stepId, s.stepType, s.elapsedMs, s.completionReason])).toEqual([ + ['step.started', 'greet', 'deterministic', 0, undefined], + ['step.completed', 'greet', 'deterministic', 250, 'success'], + ['step.started', 'plan', 'llm', 0, undefined], + ['step.failed', 'plan', 'llm', 200, 'verification_failed'], + ]); + expect(r.finished).toEqual([{ status: 'failed', completionReason: 'step_failed' }]); + }); + + it('folds entries journaled before liveSinceMs without publishing them', () => { + const r = recorder(); + const project = createJournalProjector(r.open, 1_200); + project(spawned); + project(entry(3, 'step.attempt.started', 1_010, 'greet')); + project(entry(4, 'step.completed', 1_260, 'greet', { completionReason: 'success', disposition: 'step_done' })); + expect(r.steps.map(s => s.publish)).toEqual([false, true]); + }); + + it('ignores step entries before run.spawned and a second run.spawned', () => { + const r = recorder(); + const project = createJournalProjector(r.open); + project(entry(3, 'step.attempt.started', 1_010, 'greet')); + project(spawned); + project(spawned); + expect(r.open).toHaveBeenCalledOnce(); + expect(r.steps).toEqual([]); + }); +}); diff --git a/packages/sdk/vitest.config.ts b/packages/sdk/vitest.config.ts index efa710b61..04c7db2fc 100644 --- a/packages/sdk/vitest.config.ts +++ b/packages/sdk/vitest.config.ts @@ -5,5 +5,6 @@ export default defineConfig({ environment: 'node', include: ['tests/**/*.test.ts'], globals: false, + setupFiles: ['tests/isolate-workspace.ts'], }, }); From e1d9052b7aabd7afadcf41af998228d0994d8046 Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 24 Sep 2026 08:20:17 +0200 Subject: [PATCH 2/3] fix(observer): retain lifecycle watches and retire run publishers --- docs/OBSERVER-RUN-PROJECTION.md | 31 +++++ kernel/DESIGN.md | 2 +- kernel/relayflowd/src/engine.rs | 12 +- kernel/relayflowd/src/server.rs | 45 ++++--- kernel/relayflowd/src/server/session.rs | 11 ++ kernel/relayflowd/src/server/tests.rs | 143 +++++++++++++++++++++- packages/sdk/src/cli/observer-session.ts | 9 +- packages/sdk/src/cli/run.ts | 7 +- packages/sdk/src/journal-projection.ts | 30 ++++- packages/sdk/src/run-projection.ts | 53 ++++++-- packages/sdk/tests/isolate-workspace.ts | 7 +- packages/sdk/tests/observer-link.test.ts | 53 ++++++++ packages/sdk/tests/run-projection.test.ts | 60 +++++++++ packages/sdk/tsconfig.tests.json | 3 + 14 files changed, 421 insertions(+), 45 deletions(-) create mode 100644 docs/OBSERVER-RUN-PROJECTION.md diff --git a/docs/OBSERVER-RUN-PROJECTION.md b/docs/OBSERVER-RUN-PROJECTION.md new file mode 100644 index 000000000..cb8f6e295 --- /dev/null +++ b/docs/OBSERVER-RUN-PROJECTION.md @@ -0,0 +1,31 @@ +# Local run observer projection + +`flows run` and `flows resume` project journal facts into `wf-` +when a Relaycast workspace is configured. The printed link uses a scoped +`ot_live_` observer token, never the workspace key. `--no-observer-link` and +`FLOWS_NO_OBSERVER=1` suppress both token creation and publication. + +YAML observation spans initial admission, out-of-band worker completion, +retries and final classification. An idempotent or recovered start watches +the existing run instead of creating another. Resume folds old facts silently; +an old terminal fact cannot close the resumed projection. Epoch summaries reset +the projected step state to the journal's retained done/open steps. Authored +child failures keep the root's channel and observer link. + +The producer sends `{text, data: {relayflow: {version: 1, event, run}}}` to +Relaycast's message endpoint. Relaycast exposes that payload as +`message.metadata.relayflow` to the dashboard. Changing the request field to +`metadata` is not compatible with the message API. + +After queued publication settles, the CLI retires its session publisher using +Relaycast's history-preserving agent deletion endpoint. Current Relaycast +tombstones the agent and revokes its credentials without deleting its messages. +Cleanup and publication are best-effort and bounded; an unavailable service or +abrupt process death cannot affect execution and can leave cleanup unfinished. +The journal remains the record (RFC-0001 decision 7). + +Full visual acceptance requires the Relaycast observer renderer (#450) in +addition to this producer. The joint proof must show live step progression, +the final status, single-channel token scope, retained history after publisher +retirement, and unchanged execution with observation disabled/unavailable. +Mocked transport tests do not establish that deployed end-to-end behavior. diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index 123f17c58..0dcb1928a 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -380,7 +380,7 @@ Minimal verb set for gate 1: | verb | params → result | purpose | |---|---|---| | `hello` | `{protocol: 0, client}` → `{protocol: 0, server}` | handshake; version mismatch is a hard error | -| `run.start` | `{spec, watch?}` → `{run_id}` | validate spec (zero-agent flows are legal; invalid declarations return `invalid_spec` before storage), create run file, append `run.spawned`, begin scheduling. `watch: true` pushes the new run's entries to this connection from `run.spawned` on, as `run.watch` does — the only way to observe a run whose id the caller does not know yet | +| `run.start` | `{spec, watch?}` → `{run_id}` | validate spec (zero-agent flows are legal; invalid declarations return `invalid_spec` before storage), create run file, append `run.spawned`, begin scheduling. `watch: true` pushes the new run's entries to this connection from `run.spawned` on; an existing/recovered admission replays then watches the same run. Failed starts roll back their watcher. Observation failure never changes admission or execution. | | `run.resume` | `{run_id}` → `{run_id, state}` | §3 memoized resume | | `run.cancel` | `{run_id}` → `{run_id, status, completion_reason}` | append durable intent, close active leases, and append the terminal canceled fact; repeated calls return the existing outcome | | `run.get` | `{run_id}` → `{status, steps, budget}` | snapshot for legibility | diff --git a/kernel/relayflowd/src/engine.rs b/kernel/relayflowd/src/engine.rs index fb7ae5302..cccc1d28f 100644 --- a/kernel/relayflowd/src/engine.rs +++ b/kernel/relayflowd/src/engine.rs @@ -263,14 +263,14 @@ impl Engine { options, reuse_from_run_id, admission_key, - &|_| {}, + &|_, _| {}, ) } /// `start_with_admission`, calling `before_first_append` with the new /// run's id before its journal exists. A watcher registered there sees - /// every entry the run appends, from `run.spawned` on. It is not called - /// when admission returns an existing run. + /// every entry the run appends, from `run.spawned` on. For an existing + /// admission, the second argument is true: replay and watch that run. pub fn start_observed( &self, spec: RunSpec, @@ -278,7 +278,7 @@ impl Engine { options: DriveOptions, reuse_from_run_id: Option<&str>, admission_key: Option<&str>, - before_first_append: &dyn Fn(&str), + before_first_append: &dyn Fn(&str, bool), ) -> Result { spec.validate().context("invalid run spec")?; let reuse = reuse_from_run_id @@ -293,9 +293,11 @@ impl Engine { validate_admission_key(key)?; match registry.claim_run_admission(key, &spec_hash, &run_id, self.boot_id())? { RunAdmissionClaim::Existing(existing_run_id) => { + before_first_append(&existing_run_id, true); return self.current_outcome(&existing_run_id); } RunAdmissionClaim::Recover(existing_run_id) => { + before_first_append(&existing_run_id, true); return self.resume_with_options(&existing_run_id, options); } RunAdmissionClaim::Conflict => { @@ -308,7 +310,7 @@ impl Engine { } } - before_first_append(&run_id); + before_first_append(&run_id, false); let path = self.run_path(&run_id); let now_ms = self.clock.now_ms(); let started = (|| -> Result { diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index fe0a63673..45abb96cc 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -172,24 +172,39 @@ fn handle_request( .map_err(|error| ("invalid_spec", error.to_string()))?; // Registered before the first append, so there is nothing to // replay: the watcher goes live and receives every entry once. - let watch = |run_id: &str| { + let watched = std::cell::RefCell::new(None::); + let watch = |run_id: &str, existing: bool| { if params.watch { - hub.watch(connection_id, run_id.to_owned(), writer.clone()); - hub.watch_ready(connection_id, run_id, 0); + if existing { + // Projection failure must never gate admission/recovery. + if watch_with_replay(&engine, hub, connection_id, run_id, writer, || ()) + .is_err() + { + return; + } + } else { + hub.watch(connection_id, run_id.to_owned(), writer.clone()); + hub.watch_ready(connection_id, run_id, 0); + } + *watched.borrow_mut() = Some(run_id.to_owned()); } }; - to_value( - engine - .start_observed( - spec, - "protocol-v0", - crate::DriveOptions::default(), - params.reuse_from_run_id.as_deref(), - params.admission_key.as_deref(), - &watch, - ) - .map_err(run_start_error)?, - ) + let outcome = engine + .start_observed( + spec, + "protocol-v0", + crate::DriveOptions::default(), + params.reuse_from_run_id.as_deref(), + params.admission_key.as_deref(), + &watch, + ) + .map_err(run_start_error); + if outcome.is_err() { + if let Some(run_id) = watched.borrow().as_deref() { + hub.unwatch(connection_id, run_id); + } + } + to_value(outcome?) } "run.resume" => { let params: RunResumeParams = decode_params(request.params)?; diff --git a/kernel/relayflowd/src/server/session.rs b/kernel/relayflowd/src/server/session.rs index 3d9a647ac..05af34aaf 100644 --- a/kernel/relayflowd/src/server/session.rs +++ b/kernel/relayflowd/src/server/session.rs @@ -101,6 +101,17 @@ pub struct ProtocolHub { } impl ProtocolHub { + #[cfg(test)] + pub fn watcher_count(&self, connection_id: u64) -> usize { + self.sessions + .lock() + .expect("protocol sessions lock") + .watchers + .values() + .flat_map(|watchers| watchers.iter()) + .filter(|watcher| watcher.connection_id == connection_id) + .count() + } pub fn run_lock(&self, run_id: &str) -> Arc> { self.run_locks .lock() diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index cb24cff05..c01ba3bfb 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -902,14 +902,18 @@ fn run_start_with_watch_streams_every_entry_once_before_the_result() { .to_string(); let started = request(data_dir, &hub, 1, &writer, &line); assert!(started.ok, "run.start failed: {:?}", started.error); - let run_id = started.result.unwrap()["run_id"].as_str().unwrap().to_owned(); + let run_id = started.result.unwrap()["run_id"] + .as_str() + .unwrap() + .to_owned(); let expected = Engine::new(data_dir) .journal_entries(&run_id, 1, usize::MAX) .unwrap(); assert_eq!(expected.first().unwrap().entry_type, EntryType::RunSpawned); assert_eq!(expected.last().unwrap().entry_type, EntryType::RunCompleted); - peer.set_read_timeout(Some(Duration::from_millis(500))).unwrap(); + peer.set_read_timeout(Some(Duration::from_millis(500))) + .unwrap(); let mut reader = BufReader::new(peer); let seen = (0..expected.len()) .map(|_| { @@ -924,7 +928,10 @@ fn run_start_with_watch_streams_every_entry_once_before_the_result() { reader.read_line(&mut leftover).is_err(), "watcher received a duplicate frame: {leftover}" ); - assert_eq!(seen, expected.iter().map(|entry| entry.seq).collect::>()); + assert_eq!( + seen, + expected.iter().map(|entry| entry.seq).collect::>() + ); } /// Without `watch`, `run.start` pushes nothing: the flag is opt-in. @@ -937,7 +944,8 @@ fn run_start_without_watch_pushes_no_entries() { let spec = json!({"steps": [{"id": "a", "type": "deterministic", "command": ["/bin/sh", "-c", "printf a"]}]}); let line = json!({"id": "start", "verb": "run.start", "params": {"spec": spec}}).to_string(); assert!(request(data_dir, &hub, 1, &writer, &line).ok); - peer.set_read_timeout(Some(Duration::from_millis(200))).unwrap(); + assert_eq!(hub.watcher_count(1), 0); + peer.set_nonblocking(true).unwrap(); let mut leftover = String::new(); assert!( BufReader::new(peer).read_line(&mut leftover).is_err(), @@ -945,6 +953,133 @@ fn run_start_without_watch_pushes_no_entries() { ); } +#[test] +fn run_start_watch_replays_an_existing_admission() { + let directory = tempdir().unwrap(); + let hub = Arc::new(ProtocolHub::default()); + let (writer, peer) = shared_writer(); + let params = json!({"admission_key":"watch-retry", "spec":{"steps":[]}}); + let first = request( + directory.path(), + &hub, + 1, + &writer, + &json!({"id":"first","verb":"run.start","params":params}).to_string(), + ); + assert!(first.ok); + let run_id = first.result.unwrap()["run_id"].as_str().unwrap().to_owned(); + let mut params = params; + params["watch"] = json!(true); + let retry = request( + directory.path(), + &hub, + 1, + &writer, + &json!({"id":"retry","verb":"run.start","params":params}).to_string(), + ); + assert!(retry.ok); + assert_eq!(retry.result.unwrap()["run_id"], run_id); + let expected = Engine::new(directory.path()) + .journal_entries(&run_id, 1, usize::MAX) + .unwrap(); + peer.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + let mut reader = BufReader::new(peer); + for entry in expected { + assert_eq!(read_frame(&mut reader)["data"]["seq"], entry.seq); + } + assert_eq!(hub.watcher_count(1), 1); + // A boot change takes the Recover branch rather than Existing. + rusqlite::Connection::open(directory.path().join("relayflowd.sqlite3")) + .unwrap() + .execute("UPDATE run_admissions SET boot_id = 'dead-boot'", []) + .unwrap(); + let (writer2, peer2) = shared_writer(); + let recovered = request( + directory.path(), + &hub, + 2, + &writer2, + &json!({"id":"recover","verb":"run.start","params":params}).to_string(), + ); + assert!(recovered.ok); + assert_eq!(recovered.result.unwrap()["run_id"], run_id); + peer2 + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + assert_eq!( + read_frame(&mut BufReader::new(peer2))["data"]["entry_type"], + "run.spawned" + ); + assert_eq!(hub.watcher_count(2), 1); +} + +#[test] +fn run_start_watch_rolls_back_after_journal_creation_failure() { + let directory = tempdir().unwrap(); + // Registry creation succeeds, but creating the per-run journal cannot. + std::fs::write(directory.path().join("runs"), "not a directory").unwrap(); + let hub = Arc::new(ProtocolHub::default()); + let (writer, _peer) = shared_writer(); + let response = request( + directory.path(), + &hub, + 1, + &writer, + &json!({"id":"bad","verb":"run.start","params":{"watch":true,"spec":{"steps":[]}}}) + .to_string(), + ); + assert!(!response.ok); + assert_eq!(hub.watcher_count(1), 0); +} + +#[test] +fn run_start_watch_streams_while_the_step_is_still_blocked() { + let directory = tempdir().unwrap(); + let release = directory.path().join("release-step"); + let command = format!( + "while [ ! -f '{}' ]; do sleep 0.01; done", + release.display() + ); + let hub = Arc::new(ProtocolHub::default()); + let (writer, peer) = shared_writer(); + let path = directory.path().to_owned(); + let (tx, rx) = mpsc::channel(); + let worker = thread::spawn(move || { + let response = request( + &path, + &hub, + 1, + &writer, + &json!({"id":"live","verb":"run.start","params":{"watch":true,"spec":{"steps":[ + {"id":"blocked","type":"deterministic","command":["/bin/sh","-c",command]} + ]}}}) + .to_string(), + ); + tx.send(response).unwrap(); + }); + peer.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut reader = BufReader::new(peer); + let first = read_frame(&mut reader); + assert_eq!(first["data"]["entry_type"], "run.spawned"); + assert!( + rx.try_recv().is_err(), + "run completed before its step was released" + ); + std::fs::write(&release, "go").unwrap(); + let mut last = 1; + loop { + let frame = read_frame(&mut reader); + let seq = frame["data"]["seq"].as_i64().unwrap(); + assert!(seq > last); + last = seq; + if frame["data"]["entry_type"] == "run.completed" { + break; + } + } + assert!(rx.recv_timeout(Duration::from_secs(5)).unwrap().ok); + worker.join().unwrap(); +} + /// Finding 5: when the journal append for a disconnect's crashed completion /// fails, the abandonment is surfaced and retained for the reconciler — never /// silently dropped — and the reconciler journals it once the journal heals. diff --git a/packages/sdk/src/cli/observer-session.ts b/packages/sdk/src/cli/observer-session.ts index fe80a8ed2..802758556 100644 --- a/packages/sdk/src/cli/observer-session.ts +++ b/packages/sdk/src/cli/observer-session.ts @@ -53,6 +53,7 @@ export function createObserverSession( const liveSinceMs = command === 'resume' ? Date.now() : 0; let projection: RunProjection | undefined; let minted: Promise | undefined; + let rootRunId: string | undefined; const mintFor = (runId: string): Promise => minted ??= mint({ workspaceKey, @@ -63,6 +64,7 @@ export function createObserverSession( const open = (run: { runId: string; flow: string; steps?: DeclaredStep[]; resumed?: boolean }): RunProjection => { if (projection !== undefined) return projection; + rootRunId = run.runId; projection = createRunProjection({ workspaceKey, ...(link.baseUrl === undefined ? {} : { baseUrl: link.baseUrl }), @@ -87,7 +89,8 @@ export function createObserverSession( onRunStarted: run => { open(run); }, onProgress(event) { projection?.step(event); }, finish(report) { - if (report.runId === undefined) return undefined; + const runId = rootRunId ?? report.runId; + if (runId === undefined) return undefined; if (projection === undefined) { io.stderr('[observer] this run was not projected (the daemon did not stream its journal); ' + 'the observer link opens an empty channel'); @@ -97,8 +100,8 @@ export function createObserverSession( : report.ok ? 'completed' : 'failed', ...(report.completionReason === undefined ? {} : { completionReason: report.completionReason }), }); - const drained = projection?.drain(DRAIN_GRACE_MS) ?? Promise.resolve(); - return drained.then(() => mintFor(report.runId!)); + const drained = projection?.close(DRAIN_GRACE_MS) ?? Promise.resolve(); + return drained.then(() => mintFor(runId)); }, }; } diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index e323191aa..15c86ff32 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -174,6 +174,7 @@ async function executeCheckedFlow( const { attachCommunicationWorkers } = await import('../communication/local.js'); communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir); } + if (options.onJournalEntry !== undefined) client.on('entry', options.onJournalEntry); const outcome = await startWatched(client, spec, options); const execution = await classifyOutcome(client, 'run', outcome, base, socketPath, { ...options, dataDir }); if (options.reuseFromRunId !== undefined) { @@ -194,6 +195,7 @@ async function executeCheckedFlow( } return protocolFailure('run', base, socketPath, communicationWorkers?.failure ?? localAgent?.failure ?? error); } finally { + if (options.onJournalEntry !== undefined) client.off('entry', options.onJournalEntry); try { await communicationWorkers?.close(); } finally { try { await localAgent?.close(); } finally { client.close(); } } } } @@ -211,15 +213,12 @@ async function startWatched( ): Promise { const onEntry = options.onJournalEntry; if (onEntry === undefined) return client.runStart(spec, options.reuseFromRunId); - client.on('entry', onEntry); try { return await client.runStart(spec, options.reuseFromRunId, undefined, true); } catch (error) { if (!(error instanceof JournalProtocolError) || error.code !== 'bad_request' || !/unknown field `watch`/.test(error.message)) throw error; return await client.runStart(spec, options.reuseFromRunId); - } finally { - client.off('entry', onEntry); } } @@ -296,7 +295,6 @@ export async function resumeFlow( if (await resumeHelperEffect(client, runId, dataDir)) { outcome = await client.runResume(runId, options.allowHumanInfluenced); } - if (onEntry !== undefined) client.off('entry', onEntry); return await classifyOutcome(client, 'resume', outcome, base, socketPath, { ...options, dataDir }); } catch (error) { if (error instanceof CommunicationEnvironmentError) return { exitCode: 2, report: { ...base, runId, socketPath, @@ -346,6 +344,7 @@ export async function resumeFlow( }; } finally { try { + if (options.onJournalEntry !== undefined) client.off('entry', options.onJournalEntry); await communicationWorkers?.close(); await authoredLlm?.close(); authoredLlmClient?.close(); diff --git a/packages/sdk/src/journal-projection.ts b/packages/sdk/src/journal-projection.ts index 093206f1f..48f5dc8e1 100644 --- a/packages/sdk/src/journal-projection.ts +++ b/packages/sdk/src/journal-projection.ts @@ -17,6 +17,7 @@ export function createJournalProjector(open: OpenProjection, liveSinceMs = 0): ( let projection: RunProjection | undefined; const types = new Map(); const started = new Map(); + const parked = new Set(); return entry => { const payload = (entry.payload ?? {}) as Record; @@ -29,7 +30,22 @@ export function createJournalProjector(open: OpenProjection, liveSinceMs = 0): ( projection = open({ runId: entry.run_id, flow: typeof spec.name === 'string' ? spec.name : 'flow', steps }); return; } - if (projection === undefined || entry.step_id === null && entry.entry_type !== 'run.completed') return; + if (projection === undefined) return; + if (entry.entry_type === 'epoch.summary') { + started.clear(); + parked.clear(); + const done = (payload['steps_done'] ?? {}) as Record; + const pending = (payload['steps_open'] ?? {}) as Record; + projection.epoch([ + ...Object.entries(done).map(([id, value]) => ({ id, + state: value.completionReason === 'success' ? 'completed' as const : 'failed' as const, + completionReason: value.completionReason })), + ...Object.entries(pending).map(([id, value]) => ({ id, attempt: value.attempt, + state: value.state === 'running' ? 'running' as const : value.state === 'needs_human' ? 'parked' as const : 'pending' as const })), + ], live); + return; + } + if (entry.step_id === null && entry.entry_type !== 'run.completed') return; const stepId = entry.step_id ?? ''; const stepType = types.get(stepId) ?? 'deterministic'; const attempt = entry.attempt ?? undefined; @@ -44,19 +60,27 @@ export function createJournalProjector(open: OpenProjection, liveSinceMs = 0): ( }; switch (entry.entry_type) { case 'step.attempt.started': + parked.delete(key); started.set(key, entry.at_ms); event('step.started'); return; case 'wait.human': + if (parked.has(key)) return; + parked.add(key); event('step.parked'); return; case 'step.completed': { const reason = typeof payload['completionReason'] === 'string' ? payload['completionReason'] : undefined; - const parked = payload['disposition'] === 'park'; - event(parked ? 'step.parked' : reason === 'success' ? 'step.completed' : 'step.failed', reason); + const isParked = payload['disposition'] === 'park'; + if (isParked && parked.has(key)) return; + if (isParked) parked.add(key); + event(isParked ? 'step.parked' : reason === 'success' ? 'step.completed' : 'step.failed', reason); return; } case 'run.completed': { + // Historical outcomes belong to earlier epochs. The current resume's + // report closes the projection if no new terminal entry is appended. + if (!live) return; const reason = typeof payload['completionReason'] === 'string' ? payload['completionReason'] : undefined; projection.finish({ status: runStatus(reason), ...(reason === undefined ? {} : { completionReason: reason }) }); return; diff --git a/packages/sdk/src/run-projection.ts b/packages/sdk/src/run-projection.ts index 9052901ca..31bb2d0dc 100644 --- a/packages/sdk/src/run-projection.ts +++ b/packages/sdk/src/run-projection.ts @@ -45,10 +45,14 @@ export interface RunProjection { readonly channel: string; /** Apply a transition; `publish: false` folds replayed history into the snapshot silently. */ step(event: ProgressEvent & { attempt?: number }, publish?: boolean): void; + /** Rebuild an epoch boundary from journal facts, never from old terminal state. */ + epoch(states: Array<{ id: string; state: ProjectedStepState; attempt?: number; completionReason?: string }>, publish?: boolean): void; /** Close the run; only the first call publishes. */ finish(outcome: { status: RunSnapshot['status']; completionReason?: string }): void; /** Resolves when every queued publication settled, or after `timeoutMs`. */ drain(timeoutMs: number): Promise; + /** Stop publication and remove this session's publisher after its queue settles. */ + close(timeoutMs: number): Promise; } export type ProjectionFetch = (url: string, init: { @@ -87,9 +91,9 @@ export function createRunProjection( })), }; - const call = async (path: string, token: string, body?: unknown, idempotencyKey?: string): Promise => { + const call = async (path: string, token: string, body?: unknown, idempotencyKey?: string, method?: string): Promise => { const response = await doFetch(new URL(`/v1${path}`, base).toString(), { - method: body === undefined ? 'GET' : 'POST', + method: method ?? (body === undefined ? 'GET' : 'POST'), headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', @@ -108,10 +112,22 @@ export function createRunProjection( // A per-session publisher: agent names are unique per workspace, and a // resumed run publishes from a fresh session into the same channel. + const publisherName = `flow-${randomUUID()}`; + let registered = false; + let cleanup: Promise | undefined; + const removePublisher = (): Promise => cleanup ??= (async () => { + if (!registered) return; + try { + await call(`/agents/${encodeURIComponent(publisherName)}`, options.workspaceKey, undefined, undefined, 'DELETE'); + } catch { + options.diagnostic(`publisher cleanup for #${channel} failed; the run is unaffected`); + } + })(); const setup = (async (): Promise => { const agent = await call('/agents', options.workspaceKey, { - name: `${slug(run.flow)}-${randomUUID().slice(0, 6)}`, type: 'agent', auto_join_general: false, + name: publisherName, type: 'agent', auto_join_general: false, }) as { token?: unknown } | undefined; + registered = true; if (typeof agent?.token !== 'string') throw new RelaycastError(0, 'agent_token_missing'); const token = agent.token; try { @@ -122,13 +138,15 @@ export function createRunProjection( await call(`/channels/${encodeURIComponent(channel)}/join`, token, {}); } return token; - })(); + })().catch(async (error: unknown) => { await removePublisher(); throw error; }); let queue: Promise = setup; let failed = false; let sequence = 0; + let closed = false; const sessionId = randomUUID().slice(0, 8); const publish = (event: string, text: string): void => { + if (closed) return; const data = { relayflow: { version: RELAYFLOW_METADATA_VERSION, event, run: structuredClone(snapshot) } }; const key = `${sessionId}-${++sequence}`; queue = queue.then(async () => { @@ -148,6 +166,19 @@ export function createRunProjection( return { channel, + epoch(states, shouldPublish = true) { + snapshot.status = 'running'; + delete snapshot.completionReason; + for (const step of snapshot.steps) { + step.state = 'pending'; + delete step.attempt; + delete step.elapsedMs; + delete step.completionReason; + const state = states.find(candidate => candidate.id === step.id); + if (state !== undefined) Object.assign(step, state); + } + if (shouldPublish) publish('run.started', `▶ ${run.flow} resumed · run ${run.runId}`); + }, step(event, shouldPublish = true) { let step = snapshot.steps.find(candidate => candidate.id === event.stepId); if (step === undefined) { @@ -184,6 +215,16 @@ export function createRunProjection( ]); clearTimeout(timer); }, + async close(timeoutMs) { + if (!closed) { + closed = true; + queue = queue.then(removePublisher, removePublisher); + } + let timer: NodeJS.Timeout | undefined; + try { + await Promise.race([queue, new Promise(resolve => { timer = setTimeout(resolve, timeoutMs); timer.unref?.(); })]); + } finally { clearTimeout(timer); } + }, }; } @@ -201,10 +242,6 @@ class RelaycastError extends Error { } } -function slug(name: string): string { - return name.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 40) || 'relayflow'; -} - function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } diff --git a/packages/sdk/tests/isolate-workspace.ts b/packages/sdk/tests/isolate-workspace.ts index d1a03fe06..358504e0b 100644 --- a/packages/sdk/tests/isolate-workspace.ts +++ b/packages/sdk/tests/isolate-workspace.ts @@ -1,4 +1,5 @@ -import { mkdtempSync } from 'node:fs'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { afterAll } from 'vitest'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -8,5 +9,7 @@ import { join } from 'node:path'; // store at an empty directory and clear the env key, for this process and the // CLI children that inherit its environment. Tests that exercise the observer // stub their own key. -process.env['AGENT_RELAY_HOME'] = mkdtempSync(join(tmpdir(), 'flows-test-relay-home-')); +const isolatedHome = mkdtempSync(join(tmpdir(), 'flows-test-relay-home-')); +process.env['AGENT_RELAY_HOME'] = isolatedHome; +afterAll(() => rmSync(isolatedHome, { recursive: true, force: true })); delete process.env['RELAYCAST_WORKSPACE_KEY']; diff --git a/packages/sdk/tests/observer-link.test.ts b/packages/sdk/tests/observer-link.test.ts index 4f6543282..d870317fb 100644 --- a/packages/sdk/tests/observer-link.test.ts +++ b/packages/sdk/tests/observer-link.test.ts @@ -15,6 +15,7 @@ import { type ObserverFetch, } from '../src/observer-link.js'; import { socketPathFor } from '../src/daemon-connection.js'; +import { createObserverSession } from '../src/cli/observer-session.js'; import { sendOk, sendResult, startLoopback, type LoopbackHandlers } from './journal-client-loopback.js'; const TESTDATA = join(dirname(fileURLToPath(import.meta.url)), '..', '..', '..', 'testdata'); @@ -91,6 +92,15 @@ function jsonResponse(status: number, body: unknown): Awaited { + it('keeps an authored root channel when the failure report names its child', async () => { + const mint = vi.fn(async (_options: import('../src/observer-link.js').MintObserverOptions) => ({ observerUrl: 'https://observer.test/root' })); + const fetch = vi.fn(async () => jsonResponse(200, { data: { token: 'at_live_test' } })); + const session = createObserverSession('run', capture().io, { RELAYCAST_WORKSPACE_KEY: 'rk_live_test' }, { mint, fetch })!; + session.onRunStarted({ runId: 'ROOT', flow: 'authored' }); + await session.finish({ command: 'run', ok: false, runId: 'CHILD', status: 'failed', diagnostics: [], resolutions: [] }); + expect(mint).toHaveBeenCalledOnce(); + expect(mint.mock.calls[0]?.[0]).toMatchObject({ channel: 'wf-root' }); + }); it('mints an observer token and returns an /observer?key= URL', async () => { const fetch = vi.fn().mockResolvedValue( jsonResponse(200, { data: { token: 'ot_live_abc123', id: 'ot_id_1' } }), @@ -586,6 +596,49 @@ describe('flows run: observer link integration', () => { .toEqual(['run.started', 'step.started', 'step.completed', 'run.completed']); }); + it.each(['run', 'resume'] as const)('keeps %s projection listening after the initial parked response', async command => { + const dataDir = temporaryProject(); + const runId = '01LATE'; + const at = Date.now() + 1_000; + const entry = (seq: number, entry_type: string, step_id: string | null, payload: unknown = {}) => ({ + event: 'entry', data: { seq, segment_id: 1, entry_type, run_id: runId, step_id, + attempt: step_id === null ? null : 1, at_ms: at + seq, payload }, + }); + const beginning = (ctx: Parameters[0]) => { + ctx.send(entry(1, 'run.spawned', null, { spec: { name: 'late', steps: [{ id: 'greet', type: 'llm' }] } })); + ctx.send(entry(2, 'step.attempt.started', 'greet')); + }; + let resumes = 0; + await startCliLoopback(dataDir, { + hello: sendOk, + 'journal.read': ctx => sendResult(ctx, { entries: [] }), + 'run.watch': ctx => { beginning(ctx); sendResult(ctx, { watching: runId }); }, + 'run.start': ctx => { beginning(ctx); sendResult(ctx, { run_id: runId, status: 'parked', completion_reason: null, completed_steps: 0 }); }, + 'run.get': ctx => sendResult(ctx, { run_id: runId, status: 'completed', steps: {} }), + 'run.resume': ctx => { + if (command === 'resume' && resumes++ === 0) { + sendResult(ctx, { run_id: runId, status: 'parked', completion_reason: null, completed_steps: 0 }); + return; + } + ctx.send(entry(3, 'step.completed', 'greet', { completionReason: 'success', disposition: 'step_done' })); + ctx.send(entry(4, 'run.completed', null, { completionReason: 'success' })); + sendResult(ctx, { run_id: runId, status: 'completed', completion_reason: 'success', completed_steps: 1 }); + }, + }); + const posted: Array<{ relayflow: { event: string; run: { steps: Array<{ state: string }> } } }> = []; + vi.stubGlobal('fetch', vi.fn(async (url: string, init: { body?: string }) => { + const path = new URL(url).pathname; + if (path.endsWith('/messages')) posted.push(JSON.parse(init.body!)['data']); + return jsonResponse(200, { data: { token: path === '/v1/observer-tokens' ? 'ot_live_test' : 'at_live_test' } }); + })); + vi.stubEnv('RELAYCAST_WORKSPACE_KEY', 'rk_live_test'); + vi.stubEnv('FLOWS_NO_OBSERVER', ''); + const args = command === 'run' ? RUN_ARGS(dataDir) : ['resume', runId, '--data-dir', dataDir]; + expect(await runCli(args, capture().io)).toBe(0); + expect(posted.map(post => post.relayflow.event)).toContain('step.completed'); + expect(posted.at(-1)?.relayflow.run.steps[0]?.state).toBe('completed'); + }); + it('starts again without watch when the daemon predates it, and says the run was not projected', async () => { const dataDir = temporaryProject(); const starts: unknown[] = []; diff --git a/packages/sdk/tests/run-projection.test.ts b/packages/sdk/tests/run-projection.test.ts index 2fb3d90f2..756d16fa0 100644 --- a/packages/sdk/tests/run-projection.test.ts +++ b/packages/sdk/tests/run-projection.test.ts @@ -26,6 +26,26 @@ const run = { runId: '01RUN', flow: 'hello-deterministic', steps: [ ] }; describe('createRunProjection', () => { + it('removes its publisher only after the final message, once across repeated close', async () => { + const { fetch, calls } = relaycast(); + const projection = createRunProjection({ workspaceKey: 'rk_live_k', fetch, diagnostic: vi.fn() }, run); + projection.finish({ status: 'completed' }); + await projection.close(1_000); + await projection.close(1_000); + const name = calls[0]!.body!['name']; + expect(calls.at(-1)).toMatchObject({ method: 'DELETE', url: `/v1/agents/${name}`, headers: { Authorization: 'Bearer rk_live_k' } }); + expect(calls.filter(call => call.method === 'DELETE')).toHaveLength(1); + expect(calls.filter(call => call.url.endsWith('/messages'))).toHaveLength(2); + }); + + it('cleans up after channel setup fails without failing the run', async () => { + const { fetch, calls } = relaycast({ '/v1/channels': { status: 500, body: {} } }); + const diagnostic = vi.fn(); + const projection = createRunProjection({ workspaceKey: 'rk_live_k', fetch, diagnostic }, run); + await projection.close(1_000); + expect(calls.filter(call => call.method === 'DELETE')).toHaveLength(1); + expect(diagnostic).toHaveBeenCalledOnce(); + }); it('registers a publisher, opens wf-, and posts each transition with the run snapshot', async () => { const { fetch, calls } = relaycast(); const projection = createRunProjection({ workspaceKey: 'rk_live_k', fetch, diagnostic: vi.fn() }, run); @@ -98,10 +118,12 @@ describe('createJournalProjector', () => { opened.push(declared); return { channel: 'wf-01run', + epoch: vi.fn(), step: (event: { type: string; stepId: string; stepType: string; elapsedMs: number; completionReason?: string }, publish = true) => { steps.push({ ...event, publish }); }, finish: (outcome: unknown) => { finished.push(outcome); }, drain: async () => {}, + close: async () => {}, }; }); return { open, opened, steps, finished }; @@ -142,6 +164,44 @@ describe('createJournalProjector', () => { expect(r.steps.map(s => s.publish)).toEqual([false, true]); }); + it('does not close a resumed projection on an earlier epoch completion', () => { + const r = recorder(); + const project = createJournalProjector(r.open, 2_000); + project(spawned); + project(entry(2, 'run.completed', 1_500, null, { completionReason: 'success' })); + project(entry(3, 'step.attempt.started', 2_100, 'greet')); + project(entry(4, 'run.completed', 2_200, null, { completionReason: 'step_failed' })); + expect(r.finished).toEqual([{ status: 'failed', completionReason: 'step_failed' }]); + expect(r.steps).toHaveLength(1); + }); + + it('resets terminal and stale steps at the next epoch and publishes its new outcome', async () => { + const { fetch, calls } = relaycast(); + const projection = createRunProjection({ workspaceKey: 'rk_live_test', fetch, diagnostic: vi.fn() }, run); + const project = createJournalProjector(() => projection); + project(spawned); + project(entry(2, 'step.completed', 1_100, 'greet', { completionReason: 'success' })); + project(entry(3, 'run.completed', 1_200, null, { completionReason: 'success' })); + project(entry(4, 'epoch.summary', 2_000, null, { steps_done: {}, steps_open: {} })); + project(entry(5, 'run.completed', 2_100, null, { completionReason: 'step_failed' })); + await projection.close(1_000); + const posts = calls.filter(call => call.url.endsWith('/messages')).map(call => call.body!['data'] as { relayflow: { run: { status: string; steps: Array<{ state: string }> } } }); + expect(posts.at(-2)?.relayflow.run).toMatchObject({ status: 'running', steps: [{ state: 'pending' }, { state: 'pending' }] }); + expect(posts.at(-1)?.relayflow.run.status).toBe('failed'); + }); + + it('publishes a manual park once per attempt, including retry attempts', () => { + const r = recorder(); + const project = createJournalProjector(r.open); + project(spawned); + for (const attempt of [1, 2]) { + project(entry(attempt * 3, 'step.attempt.started', 1_100, 'greet', {}, attempt)); + project(entry(attempt * 3 + 1, 'step.completed', 1_200, 'greet', { disposition: 'park', completionReason: 'crashed' }, attempt)); + project(entry(attempt * 3 + 2, 'wait.human', 1_200, 'greet', {}, attempt)); + } + expect(r.steps.filter(step => step.type === 'step.parked')).toHaveLength(2); + }); + it('ignores step entries before run.spawned and a second run.spawned', () => { const r = recorder(); const project = createJournalProjector(r.open); diff --git a/packages/sdk/tsconfig.tests.json b/packages/sdk/tsconfig.tests.json index 8305cc33a..2a2a36546 100644 --- a/packages/sdk/tsconfig.tests.json +++ b/packages/sdk/tsconfig.tests.json @@ -11,6 +11,9 @@ }, "include": [ "src/**/*.ts", + "tests/isolate-workspace.ts", + "tests/run-projection.test.ts", + "tests/observer-link.test.ts", "tests/bundle.test.ts", "tests/deploy.test.ts", "tests/run-from-digest.test.ts", From 63c71cc481d4dd2edab560a92645550394cb71e4 Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 24 Sep 2026 08:28:28 +0200 Subject: [PATCH 3/3] fix(observer): retire publishers after ambiguous registration --- kernel/DESIGN.md | 2 +- packages/sdk/src/run-projection.ts | 16 ++++++++-------- packages/sdk/tests/observer-link.test.ts | 9 +++++---- packages/sdk/tests/run-projection.test.ts | 20 +++++++++++++++++++- 4 files changed, 33 insertions(+), 14 deletions(-) diff --git a/kernel/DESIGN.md b/kernel/DESIGN.md index 0dcb1928a..d3cd99d64 100644 --- a/kernel/DESIGN.md +++ b/kernel/DESIGN.md @@ -380,7 +380,7 @@ Minimal verb set for gate 1: | verb | params → result | purpose | |---|---|---| | `hello` | `{protocol: 0, client}` → `{protocol: 0, server}` | handshake; version mismatch is a hard error | -| `run.start` | `{spec, watch?}` → `{run_id}` | validate spec (zero-agent flows are legal; invalid declarations return `invalid_spec` before storage), create run file, append `run.spawned`, begin scheduling. `watch: true` pushes the new run's entries to this connection from `run.spawned` on; an existing/recovered admission replays then watches the same run. Failed starts roll back their watcher. Observation failure never changes admission or execution. | +| `run.start` | `{spec, watch?, admission_key?, reuse_from_run_id?}` → `{run_id}` | validate spec (zero-agent flows are legal; invalid declarations return `invalid_spec` before storage), create run file, append `run.spawned`, begin scheduling. An admission key deduplicates starts of the same spec; reuse names a prior run whose compatible results may be reused. `watch: true` pushes the new run's entries from `run.spawned` on; an existing/recovered admission replays then watches the same run. Failed starts roll back their watcher. Observation failure never changes admission or execution. | | `run.resume` | `{run_id}` → `{run_id, state}` | §3 memoized resume | | `run.cancel` | `{run_id}` → `{run_id, status, completion_reason}` | append durable intent, close active leases, and append the terminal canceled fact; repeated calls return the existing outcome | | `run.get` | `{run_id}` → `{status, steps, budget}` | snapshot for legibility | diff --git a/packages/sdk/src/run-projection.ts b/packages/sdk/src/run-projection.ts index 31bb2d0dc..538326b76 100644 --- a/packages/sdk/src/run-projection.ts +++ b/packages/sdk/src/run-projection.ts @@ -113,21 +113,21 @@ export function createRunProjection( // A per-session publisher: agent names are unique per workspace, and a // resumed run publishes from a fresh session into the same channel. const publisherName = `flow-${randomUUID()}`; - let registered = false; let cleanup: Promise | undefined; - const removePublisher = (): Promise => cleanup ??= (async () => { - if (!registered) return; + const removePublisher = (reportFailure = true): Promise => cleanup ??= (async () => { + // The unique name is known before POST. A lost create response is not + // proof that creation failed: retire it even after an ambiguous outcome. try { await call(`/agents/${encodeURIComponent(publisherName)}`, options.workspaceKey, undefined, undefined, 'DELETE'); - } catch { - options.diagnostic(`publisher cleanup for #${channel} failed; the run is unaffected`); + } catch (error) { + if (error instanceof RelaycastError && error.status === 404) return; + if (reportFailure) options.diagnostic(`publisher cleanup for #${channel} failed; the run is unaffected`); } })(); const setup = (async (): Promise => { const agent = await call('/agents', options.workspaceKey, { name: publisherName, type: 'agent', auto_join_general: false, }) as { token?: unknown } | undefined; - registered = true; if (typeof agent?.token !== 'string') throw new RelaycastError(0, 'agent_token_missing'); const token = agent.token; try { @@ -138,7 +138,7 @@ export function createRunProjection( await call(`/channels/${encodeURIComponent(channel)}/join`, token, {}); } return token; - })().catch(async (error: unknown) => { await removePublisher(); throw error; }); + })().catch(async (error: unknown) => { await removePublisher(false); throw error; }); let queue: Promise = setup; let failed = false; @@ -218,7 +218,7 @@ export function createRunProjection( async close(timeoutMs) { if (!closed) { closed = true; - queue = queue.then(removePublisher, removePublisher); + queue = queue.then(() => removePublisher(), () => removePublisher()); } let timer: NodeJS.Timeout | undefined; try { diff --git a/packages/sdk/tests/observer-link.test.ts b/packages/sdk/tests/observer-link.test.ts index d870317fb..20d16fc90 100644 --- a/packages/sdk/tests/observer-link.test.ts +++ b/packages/sdk/tests/observer-link.test.ts @@ -600,13 +600,13 @@ describe('flows run: observer link integration', () => { const dataDir = temporaryProject(); const runId = '01LATE'; const at = Date.now() + 1_000; - const entry = (seq: number, entry_type: string, step_id: string | null, payload: unknown = {}) => ({ + const entry = (seq: number, entry_type: string, step_id: string | null, payload: unknown = {}, replay = false) => ({ event: 'entry', data: { seq, segment_id: 1, entry_type, run_id: runId, step_id, - attempt: step_id === null ? null : 1, at_ms: at + seq, payload }, + attempt: step_id === null ? null : 1, at_ms: (replay ? at - 60_000 : at) + seq, payload }, }); const beginning = (ctx: Parameters[0]) => { - ctx.send(entry(1, 'run.spawned', null, { spec: { name: 'late', steps: [{ id: 'greet', type: 'llm' }] } })); - ctx.send(entry(2, 'step.attempt.started', 'greet')); + ctx.send(entry(1, 'run.spawned', null, { spec: { name: 'late', steps: [{ id: 'greet', type: 'llm' }] } }, command === 'resume')); + ctx.send(entry(2, 'step.attempt.started', 'greet', {}, command === 'resume')); }; let resumes = 0; await startCliLoopback(dataDir, { @@ -636,6 +636,7 @@ describe('flows run: observer link integration', () => { const args = command === 'run' ? RUN_ARGS(dataDir) : ['resume', runId, '--data-dir', dataDir]; expect(await runCli(args, capture().io)).toBe(0); expect(posted.map(post => post.relayflow.event)).toContain('step.completed'); + if (command === 'resume') expect(posted.map(post => post.relayflow.event)).not.toContain('step.started'); expect(posted.at(-1)?.relayflow.run.steps[0]?.state).toBe('completed'); }); diff --git a/packages/sdk/tests/run-projection.test.ts b/packages/sdk/tests/run-projection.test.ts index 756d16fa0..83e5daaad 100644 --- a/packages/sdk/tests/run-projection.test.ts +++ b/packages/sdk/tests/run-projection.test.ts @@ -95,10 +95,28 @@ describe('createRunProjection', () => { projection.step({ type: 'step.started', stepId: 'greet', stepType: 'deterministic', elapsedMs: 0 }); projection.finish({ status: 'completed' }); await projection.drain(1_000); - expect(calls).toHaveLength(1); + expect(calls).toHaveLength(2); + expect(calls[1]!.method).toBe('DELETE'); expect(diagnostic).toHaveBeenCalledOnce(); expect(diagnostic.mock.calls[0]![0]).toContain('HTTP 401 unauthorized'); }); + + it.each([204, 404])('retires an uncertain publisher create and treats DELETE %s as settled', async status => { + const calls: Array<{ url: string; method: string; body?: string }> = []; + const fetch: ProjectionFetch = async (url, init) => { + calls.push({ url, method: init.method, body: init.body }); + if (init.method === 'POST') throw new Error('create response lost'); + return { ok: status === 204, status, json: async () => ({ error: { code: 'agent_not_found' } }) }; + }; + const diagnostic = vi.fn(); + const projection = createRunProjection({ workspaceKey: 'rk_live_test', fetch, diagnostic }, run); + await projection.close(1_000); + const name = (JSON.parse(calls[0]!.body!) as { name: string }).name; + expect(calls[1]).toMatchObject({ method: 'DELETE', url: `https://cast.agentrelay.com/v1/agents/${name}` }); + expect(calls).toHaveLength(2); + expect(diagnostic).toHaveBeenCalledOnce(); + expect(diagnostic.mock.calls[0]![0]).toContain('create response lost'); + }); }); describe('createJournalProjector', () => {