fix(observer): project each run into its channel and scope the link to it

docs/OBSERVER-RUN-PROJECTION.md

@@ -0,0 +1,31 @@
+# Local run observer projection
+
+`flows run` and `flows resume` project journal facts into `wf-<root-run-id>`
+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.

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?, 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 |

kernel/relayflowd/src/engine.rs

@@ -256,6 +256,29 @@ impl<C: Clock> Engine<C> {
         options: DriveOptions,
         reuse_from_run_id: Option<&str>,
         admission_key: Option<&str>,
+    ) -> Result<RunOutcome> {
+        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. For an existing
+    /// admission, the second argument is true: replay and watch that 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, bool),
     ) -> Result<RunOutcome> {
         spec.validate().context("invalid run spec")?;
         let reuse = reuse_from_run_id
@@ -270,9 +293,11 @@ impl<C: Clock> Engine<C> {
             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 => {
@@ -285,6 +310,7 @@ impl<C: Clock> Engine<C> {
             }
         }
 
+        before_first_append(&run_id, false);
         let path = self.run_path(&run_id);
         let now_ms = self.clock.now_ms();
         let started = (|| -> Result<RunOutcome> {

kernel/relayflowd/src/server.rs

@@ -170,17 +170,41 @@ fn handle_request(
                 .map_err(|error| ("invalid_spec", error.to_string()))?;
             spec.validate()
                 .map_err(|error| ("invalid_spec", error.to_string()))?;
-            to_value(
-                engine
-                    .start_with_admission(
-                        spec,
-                        "protocol-v0",
-                        crate::DriveOptions::default(),
-                        params.reuse_from_run_id.as_deref(),
-                        params.admission_key.as_deref(),
-                    )
-                    .map_err(run_start_error)?,
-            )
+            // Registered before the first append, so there is nothing to
+            // replay: the watcher goes live and receives every entry once.
+            let watched = std::cell::RefCell::new(None::<String>);
+            let watch = |run_id: &str, existing: bool| {
+                if params.watch {
+                    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());
+                }
+            };
+            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)?;

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<Mutex<()>> {
         self.run_locks
             .lock()

kernel/relayflowd/src/server/tests.rs

@@ -885,6 +885,201 @@ 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::<Vec<_>>();
+    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::<Vec<_>>()
+    );
+}
+
+/// 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);
+    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(),
+        "an unwatched start pushed a frame: {leftover}"
+    );
+}
+
+#[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.

kernel/relayflowd/src/server/wire.rs

@@ -41,6 +41,9 @@ pub(super) struct RunStartParams {
     pub spec: Value,
     pub reuse_from_run_id: Option<String>,
     pub admission_key: Option<String>,
+    /// Stream the new run's entries to this connection, as `run.watch` does.
+    #[serde(default)]
+    pub watch: bool,
 }
 
 #[derive(Deserialize)]

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 {