Merge 8f4f127b20 into b19163e4b4

docs/internal/events-strategy.md

@@ -32,6 +32,14 @@ Tracing logs are separate. Tracing is developer diagnostics; the stream is
 the product-facing record other systems consume. If something must be
 visible after a reattach, it has to be a record, not a log line.
 
+The server also mirrors selected lifecycle records into an independent
+global **main event stream** (`fabro-store::MainEventStore`). Its compact
+`fabro.run.*` events have their own sequence and survive run deletion.
+Source-keyed appends deduplicate replay of a run's canonical stream; the
+main stream is a cross-run mirror, not another source for run projections.
+See [the main event design](../superpowers/specs/2026-06-21-main-event-stream-design.md)
+for its event vocabulary, failure policy, and recovery boundary.
+
 ## Recording a fact
 
 Petri's own facts need nothing from Fabro: the engine records them and the

docs/superpowers/specs/2026-06-21-main-event-stream-design.md

@@ -0,0 +1,126 @@
+# Durable main event stream
+
+Updated 2026-10-16 for upstream's Petri run stream and SQLite storage.
+
+## Purpose
+
+A run's canonical history is Petri's execution records and Fabro's platform
+records, delivered together as `RunStreamItem`. This design adds an independent,
+compact server-wide lifecycle stream for future cross-run consumers. It does
+not replace the run stream, its projection, APIs, attach broadcast, or CLI.
+
+The implementation follows the lifecycle mirror maintained in the ipt fork,
+using upstream's current SQLite store in place of the fork's SlateDB store.
+
+## Event contract
+
+`fabro-types::MainEventEnvelope` contains a global `seq` and a `MainEvent`.
+The main event has its own UUID v7 `id`, append timestamp `ts`, event name,
+and compact `properties`. The eight lifecycle names are:
+
+- `fabro.run.created`
+- `fabro.run.started`
+- `fabro.run.running`
+- `fabro.run.completed`
+- `fabro.run.failed`
+- `fabro.run.cancelled`
+- `fabro.run.paused`
+- `fabro.run.unpaused`
+
+Each lifecycle payload preserves the run ID, source item ID, source timestamp,
+and source event name. Source IDs include the stream kind, for example
+`platform:1` or `petri:coordinator/3/0`, so the two logs cannot collide.
+Native records do not carry a request actor; the optional actor stays absent.
+Creation preserves the initiating principal from the immutable run spec.
+
+Creation includes the original title, workflow slug, automation reference,
+Git context, and parent ID. Start includes workflow identity, goal, and Git
+baseline/branch metadata when available. Terminal events include the outcome,
+final Git commit, automation reference, and recorded aggregate Lithos usage
+when available. Completion also retains the recorded cost in USD micros;
+missing cost remains absent. Mirroring does not price usage again.
+
+Unknown event names and their arbitrary JSON properties round-trip unchanged.
+Malformed properties for a known event are rejected rather than silently
+reinterpreted as an unknown event.
+
+## Persistence
+
+`fabro-store::MainEventStore` uses the existing run-summary SQLite pool. The
+`main_events` table has:
+
+- `seq`: a positive global sequence, bounded to `u32`;
+- `source_key`: an optional unique key for idempotent source delivery;
+- `event_json`: the typed main event serialized as JSON.
+
+There is no foreign key to a run. Run deletion removes neither these events
+nor their deduplication identities. A `BEGIN IMMEDIATE` transaction serializes
+writers, checks for an existing source, and commits the next sequence and
+payload together. Retrying a source returns the original envelope, consumes
+no sequence, and emits no new live message. Failed writes publish nothing.
+The next sequence comes from the committed table, including after restart.
+Sequence exhaustion returns an error without overwriting existing events.
+
+The store supports ordinary append, source-keyed append, inclusive replay
+from a global sequence, and in-process subscription. Replay queries return
+at most `limit + 1` rows in sequence order. Subscribers receive committed
+appends; a slow subscriber must replay after a broadcast lag. Broadcasts are
+local to the shared store handle, not a cross-process delivery mechanism.
+
+Storage owns persistence and deduplication. It does not decide which run
+records belong in the lifecycle stream.
+
+## Native lifecycle mapping
+
+`fabro-petri::main_events` maps native stream items into compact payloads:
+
+- The platform `run.created` record supplies creation from its immutable
+  record, including the original title rather than a later renamed title.
+- Petri's `run.started` supplies start.
+- Platform `run.lifecycle` transitions supply running, paused, unpaused,
+  succeeded, failed, and dead outcomes.
+- A terminal failure whose reason is `Cancelled` becomes
+  `fabro.run.cancelled`. A cancellation request alone is not terminal.
+
+Petri's `run.finished` is not mirrored independently: Fabro's terminal
+lifecycle record reflects the coordinator result after required finalization.
+Stage, agent, tool, transcript, sandbox, graph, settings, patch, and output
+records remain exclusively in the canonical run history.
+
+## Server delivery and recovery
+
+The server mirrors creation synchronously after persistence and projection,
+before returning success from create, retry, fork, or rewind. A failed main
+append fails that request; the persisted run is retained and background
+reconciliation can repair the missing delivery. The synchronous path and
+background path use the same source identity.
+
+A separate server task follows projector commit signals and scans all retained
+runs at startup and every 30 seconds. For each run it reads stream pages and
+loads the committed projection after the page, so terminal usage is visible.
+It advances its in-memory cursor only after an item has been processed
+successfully. Mapping or storage errors are logged with run identity and
+retried; they do not interrupt execution or the existing live run broadcast.
+
+After restart, cursors begin at zero. The canonical run stream acts as the
+outbox, and source-keyed appends make replay idempotent. A lost or lagged signal
+costs delivery latency; the periodic scan repairs it. Cursors are an
+optimization, never evidence that an uncommitted append succeeded.
+
+The main sequence describes append order. Recovery may interleave historical
+runs differently from their original wall-clock order; source timestamps
+preserve the time of the underlying records. Events already mirrored survive
+run deletion, but a run deleted before reconciliation cannot supply missing
+records afterward. This is not an exactly-once trigger-delivery contract.
+
+## Scope and verification
+
+This change adds no public main-stream API, CLI command, SSE endpoint, webhook
+ingestion, automation listener, persistent consumer cursor, or delivery receipt
+for external consumers.
+
+Tests cover all lifecycle names and JSON round trips, unknown events and
+malformed known events, native record filtering, cancellation, terminal usage,
+global replay and subscription order, concurrent deduplication, restart
+recovery, failed writes, sequence exhaustion, deletion survival, multi-page
+reconciliation, and startup reconciliation without a new signal.

lib/apps/fabro-server/src/server.rs

@@ -164,6 +164,7 @@ use crate::{
 
 mod automation_scheduler;
 mod handler;
+mod main_event_mirror;
 pub(crate) mod petri_runs;
 mod pull_request_supervisor;
 pub(crate) mod resource_sampler;
@@ -4471,6 +4472,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
 /// Background task that promotes runnable runs when capacity is available.
 pub fn spawn_scheduler(state: Arc<AppState>) {
     stream_follower::spawn_stream_follower(Arc::clone(&state));
+    main_event_mirror::spawn_native_mirror(Arc::clone(&state));
     tokio::spawn(async move {
         loop {
             tokio::select! {

lib/apps/fabro-server/src/server/handler/lineage.rs

@@ -34,7 +34,8 @@ use fabro_workflow::operations::{self, ForkTarget, ResolvedForkTarget, RunTimeli
 use tracing::{error, warn};
 
 use super::super::{
-    ApiError, AppState, RequireRunManagementTarget, RequiredUser, parse_run_id_path, run_records,
+    ApiError, AppState, RequireRunManagementTarget, RequiredUser, main_event_mirror,
+    parse_run_id_path, run_records,
 };
 use super::lifecycle::{ArchiveAction, queue_run, run_archive_operation, run_response};
 use super::runs::run_provenance;
@@ -179,6 +180,7 @@ async fn retry_run(
         ))
         .await
         .map_err(workflow_operation_error)?;
+        mirror_created(state.as_ref(), new_run_id).await?;
         let projection = run_records::require_projection(state.as_ref(), new_run_id).await?;
         queue_run(state.as_ref(), new_run_id, &projection, false, actor).await?;
         Ok::<_, ApiError>(new_run_id)
@@ -308,6 +310,7 @@ async fn fork_at(
     .await
     .map_err(workflow_operation_error)?;
 
+    mirror_created(state, new_run_id).await?;
     let seeded = petri_fork::fork(ForkRequest {
         source: id,
         fork: new_run_id,
@@ -350,6 +353,18 @@ async fn fork_at(
     })
 }
 
+async fn mirror_created(state: &AppState, run_id: RunId) -> Result<(), ApiError> {
+    main_event_mirror::mirror_created(state, run_id)
+        .await
+        .map_err(|error| {
+            error!(%run_id, error = %error, "Failed to mirror persisted run creation");
+            ApiError::new(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to persist run lifecycle event",
+            )
+        })
+}
+
 /// A refused position is the caller's mistake; anything else is the
 /// server's.
 fn fork_error(err: &ForkError) -> ApiError {

lib/apps/fabro-server/src/server/handler/runs.rs

@@ -47,7 +47,7 @@ use tracing::info;
 use super::super::{
     AppState, DeleteRunOutcome, ListResponse, RunExecutionMode, VariableError, answer_from_request,
     api_question_from_pending_interview, clamp_page_limit, clamp_page_offset, default_page_limit,
-    delete_run_internal, load_pending_interview, managed_run, parse_run_id_path,
+    delete_run_internal, load_pending_interview, main_event_mirror, managed_run, parse_run_id_path,
     parse_stage_id_path, petri_runs, reject_if_archived, run_records,
     submit_pending_interview_answer,
 };
@@ -778,7 +778,14 @@ async fn finalize_created_run(
     let created_at = created.run_id.created_at();
     // The run's summary row is the projector's: wait for the pass that
     // folds the run's first records before reading the run back.
-    state.petri_projector.settle(created.run_id).await;
+    if let Err(error) = main_event_mirror::mirror_created(&state, created.run_id).await {
+        tracing::error!(run_id = %created.run_id, error = %error, "Failed to mirror persisted run creation");
+        return intent_error(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "failed to persist run lifecycle event",
+            "run_persistence_failed",
+        );
+    }
     let summary = match state
         .stores
         .run_summaries

lib/apps/fabro-server/src/server/main_event_mirror.rs

@@ -0,0 +1,336 @@
+//! Replay selected lifecycle records into the independent global stream.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Context as _;
+use fabro_petri::main_events;
+use fabro_store::MainEventSourceKey;
+use fabro_types::{RunId, RunProjection, RunStreamItem, RunStreamItemKind};
+use tokio::sync::broadcast::error::RecvError;
+use tokio::time;
+use tracing::warn;
+
+use super::AppState;
+
+const BATCH_LIMIT: usize = 256;
+
+/// Mirror a native stream item with a stable source key. The SQL stream is the
+/// outbox: replay after a crash retries the append without duplication.
+async fn mirror_native_item(
+    state: &AppState,
+    item: &RunStreamItem,
+    projection: &RunProjection,
+) -> anyhow::Result<()> {
+    let Some(body) = main_events::from_stream_item(item, projection)? else {
+        return Ok(());
+    };
+    state
+        .stores
+        .runs
+        .main_events()
+        .append_once(
+            MainEventSourceKey::RunStream {
+                run_id:  item.run_id,
+                item_id: format!("{}:{}", item.kind, item.id),
+            },
+            body,
+        )
+        .await?;
+    Ok(())
+}
+
+/// Reconcile all native streams on startup and periodically. Signals shorten
+/// delivery latency; the scan also repairs gaps when a signal is lost or a
+/// process exits between the run record commit and the main event append.
+pub(crate) fn spawn_native_mirror(state: Arc<AppState>) {
+    tokio::spawn(async move {
+        let mut cursors = HashMap::new();
+        let mut signals = state.petri_projector.subscribe();
+        let mut tick = time::interval(Duration::from_secs(30));
+        let shutdown = state.shutdown_token();
+        loop {
+            tokio::select! {
+                () = shutdown.cancelled() => break,
+                _ = tick.tick() => {
+                    match state.stores.run_summaries.list_identities().await {
+                        Ok(runs) => for run in runs {
+                            let cursor = cursors.entry(run.id).or_insert(0);
+                            if let Err(error) = reconcile_native_run(&state, run.id, cursor).await {
+                                warn!(run_id = %run.id, error = %error, "Native main event mirror will retry");
+                            }
+                        },
+                        Err(error) => warn!(error = %error, "Main event mirror could not list runs; will retry"),
+                    }
+                }
+                signal = signals.recv() => match signal {
+                    Ok(run_id) => {
+                        let cursor = cursors.entry(run_id).or_insert(0);
+                        if let Err(error) = reconcile_native_run(&state, run_id, cursor).await {
+                            warn!(%run_id, error = %error, "Native main event mirror will retry");
+                        }
+                    }
+                    Err(RecvError::Closed) => break,
+                    Err(RecvError::Lagged(_)) => {},
+                },
+            }
+        }
+    });
+}
+
+async fn reconcile_native_run(
+    state: &AppState,
+    run_id: RunId,
+    cursor: &mut u64,
+) -> anyhow::Result<()> {
+    loop {
+        let items = state
+            .petri_projector
+            .stream_after(run_id, *cursor, BATCH_LIMIT)
+            .await?;
+        if items.is_empty() {
+            return Ok(());
+        }
+        // Read after the stream page: the projection transaction that published
+        // these items must be visible before terminal usage is mirrored.
+        let Some(projection) = state
+            .stores
+            .run_summaries
+            .load_petri_projection(&run_id)
+            .await?
+        else {
+            return Ok(());
+        };
+        let drained = items.len() < BATCH_LIMIT;
+        for item in items {
+            mirror_native_item(state, &item, &projection).await?;
+            *cursor = item.stream_seq;
+        }
+        if drained {
+            return Ok(());
+        }
+    }
+}
+
+/// Creation is mirrored before a successful create response. The background
+/// mirror uses the same source key, so either delivery can win safely.
+pub(crate) async fn mirror_created(state: &AppState, run_id: RunId) -> anyhow::Result<()> {
+    state.petri_projector.settle(run_id).await;
+    let items = state.petri_projector.stream_after(run_id, 0, 1).await?;
+    let created = items
+        .first()
+        .filter(|item| {
+            item.kind == RunStreamItemKind::Platform && item.name() == Some("run.created")
+        })
+        .context("created run has no initial run.created stream item")?;
+    let projection = state
+        .stores
+        .run_summaries
+        .load_petri_projection(&run_id)
+        .await?
+        .context("created run has no projection")?;
+    mirror_native_item(state, created, &projection).await
+}
+
+#[cfg(test)]
+mod tests {
+    use fabro_store::platform_records::{
+        PlatformRecord, RunCreatedRecord, RunLifecycleKind, RunLifecycleRecord,
+    };
+    use fabro_types::{FailureReason, MainEventBody, RunStatus, test_support};
+    use tokio::sync::broadcast::error::TryRecvError;
+
+    use super::*;
+    use crate::server::run_records;
+
+    async fn create(state: &AppState) -> RunId {
+        let mut spec = test_support::test_run_spec();
+        spec.run_id = RunId::new();
+        let run_id = spec.run_id;
+        run_records::append(
+            state,
+            run_id,
+            PlatformRecord::RunCreated(RunCreatedRecord {
+                spec,
+                title: Some("Original title".into()),
+                parent_id: None,
+                retried_from: None,
+                web_url: None,
+            }),
+        )
+        .await
+        .unwrap();
+        run_id
+    }
+
+    #[tokio::test]
+    async fn main_event_mirror_replays_lifecycle_once_and_keeps_events_after_deletion() {
+        let state = crate::test_support::test_app_state();
+        let run_id = create(&state).await;
+        mirror_created(&state, run_id).await.unwrap();
+        let events = state.stores.runs.main_events();
+        let created = events.list_from_with_limit(1, 10).await.unwrap();
+        assert_eq!(created.len(), 1);
+        let MainEventBody::RunCreated(created) = &created[0].event.body else {
+            panic!("created");
+        };
+        assert_eq!(created.title.as_deref(), Some("Original title"));
+
+        for transition in [
+            RunLifecycleKind::Running,
+            RunLifecycleKind::PauseRequested,
+            RunLifecycleKind::Paused,
+            RunLifecycleKind::Unpaused,
+            RunLifecycleKind::CancelRequested,
+        ] {
+            run_records::lifecycle(&state, run_id, RunLifecycleRecord::new(transition))
+                .await
+                .unwrap();
+        }
+        run_records::lifecycle(
+            &state,
+            run_id,
+            run_records::failed(FailureReason::Cancelled, "Stopped by user"),
+        )
+        .await
+        .unwrap();
+        let mut cursor = 0;
+        reconcile_native_run(&state, run_id, &mut cursor)
+            .await
+            .unwrap();
+        let mirrored = events.list_from_with_limit(1, 20).await.unwrap();
+        assert_eq!(
+            mirrored
+                .iter()
+                .map(|event| event.event.event_name())
+                .collect::<Vec<_>>(),
+            [
+                "fabro.run.created",
+                "fabro.run.running",
+                "fabro.run.paused",
+                "fabro.run.unpaused",
+                "fabro.run.cancelled"
+            ]
+        );
+        let MainEventBody::RunCancelled(cancelled) = &mirrored[4].event.body else {
+            panic!("cancelled");
+        };
+        assert_eq!(cancelled.message, "Stopped by user");
+        let head = state
+            .petri_projector
+            .stream_head(run_id)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(cursor, head);
+
+        let mut live = events.subscribe();
+        // A new process starts with no in-memory cursor, then replays its outbox.
+        reconcile_native_run(&state, run_id, &mut 0).await.unwrap();
+        assert_eq!(live.try_recv(), Err(TryRecvError::Empty));
+        assert_eq!(events.list_from_with_limit(1, 20).await.unwrap(), mirrored);
+        state.petri_projector.delete_run(run_id).await.unwrap();
+        assert_eq!(events.list_from_with_limit(1, 20).await.unwrap(), mirrored);
+    }
+
+    #[tokio::test]
+    async fn main_event_mirror_retries_failed_delivery_without_advancing_cursor() {
+        let state = crate::test_support::test_app_state();
+        let run_id = create(&state).await;
+        let pool = state.stores.run_summaries.pool();
+        sqlx::raw_sql("CREATE TRIGGER fail_main_event BEFORE INSERT ON main_events BEGIN SELECT RAISE(ABORT, 'injected failure'); END;")
+            .execute(&pool).await.unwrap();
+        assert!(mirror_created(&state, run_id).await.is_err());
+        let mut cursor = 0;
+        assert!(
+            reconcile_native_run(&state, run_id, &mut cursor)
+                .await
+                .is_err()
+        );
+        assert_eq!(cursor, 0);
+        sqlx::query("DROP TRIGGER fail_main_event")
+            .execute(&pool)
+            .await
+            .unwrap();
+        reconcile_native_run(&state, run_id, &mut cursor)
+            .await
+            .unwrap();
+        assert!(cursor > 0);
+        mirror_created(&state, run_id).await.unwrap();
+        let events = state
+            .stores
+            .runs
+            .main_events()
+            .list_from_with_limit(1, 10)
+            .await
+            .unwrap();
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].seq, 1);
+    }
+
+    #[tokio::test]
+    async fn main_event_mirror_drains_multiple_pages_without_changing_the_run_stream() {
+        let state = crate::test_support::test_app_state();
+        let run_id = create(&state).await;
+        let records = state.stores.run_summaries.platform_records();
+        for _ in 0..BATCH_LIMIT {
+            records
+                .append(
+                    &run_id,
+                    &PlatformRecord::RunLifecycle(
+                        RunLifecycleRecord::new(RunLifecycleKind::Running)
+                            .with_status(RunStatus::Running),
+                    ),
+                    None,
+                )
+                .await
+                .unwrap();
+        }
+        state.stores.run_summaries.notify_platform_record(run_id);
+        state.petri_projector.settle(run_id).await;
+        let original = state
+            .petri_projector
+            .stream_after(run_id, 0, BATCH_LIMIT * 2)
+            .await
+            .unwrap();
+        let mut cursor = 0;
+        reconcile_native_run(&state, run_id, &mut cursor)
+            .await
+            .unwrap();
+        assert_eq!(cursor, original.last().unwrap().stream_seq);
+        assert_eq!(
+            state
+                .petri_projector
+                .stream_after(run_id, 0, BATCH_LIMIT * 2)
+                .await
+                .unwrap(),
+            original
+        );
+        let events = state
+            .stores
+            .runs
+            .main_events()
+            .list_from_with_limit(1, BATCH_LIMIT * 2)
+            .await
+            .unwrap();
+        assert_eq!(events.len(), BATCH_LIMIT + 1);
+    }
+
+    #[tokio::test]
+    async fn main_event_mirror_startup_reconciles_without_a_new_signal() {
+        let state = crate::test_support::test_app_state();
+        let run_id = create(&state).await;
+        let events = state.stores.runs.main_events();
+        let mut live = events.subscribe();
+        spawn_native_mirror(Arc::clone(&state));
+        let event = time::timeout(Duration::from_secs(5), live.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(matches!(event.event.body, MainEventBody::RunCreated(_)));
+        mirror_created(&state, run_id).await.unwrap();
+        assert_eq!(events.list_from_with_limit(1, 10).await.unwrap().len(), 1);
+        state.shutdown_token().cancel();
+    }
+}

lib/apps/fabro-server/src/server/tests.rs

@@ -3525,6 +3525,171 @@ file = "goal.md"
     );
 }
 
+#[tokio::test]
+async fn main_event_mirror_follows_an_executed_petri_workflow() {
+    let workspace = tempfile::tempdir().unwrap();
+    let state = TestAppStateBuilder::new()
+        .default_environment_provider(Some(SandboxProviderKind::LOCAL))
+        .vault_entries([(fabro_static::EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+        .in_process_execution()
+        .build();
+    let app = test_app_with_scheduler(Arc::clone(&state));
+    let version = store_workflow_version(
+        &state,
+        r#"digraph Command {
+        graph [goal="Run one command"]
+        start [shape=Mdiamond]
+        exit [shape=Msquare]
+        say [shape=parallelogram, script="echo hello from petri"]
+        start -> say -> exit
+    }"#,
+        None,
+    )
+    .await;
+    let events = state.stores.runs.main_events();
+    let mut live = events.subscribe();
+    let created = post_run_intent(&app, folder_intent(version, workspace.path())).await;
+    let run_id: RunId = created["id"].as_str().unwrap().parse().unwrap();
+    let request = Request::builder()
+        .method("POST")
+        .uri(api(&format!("/runs/{run_id}/start")))
+        .body(Body::empty())
+        .unwrap();
+    let response = app.oneshot(request).await.unwrap();
+    response_json!(response, StatusCode::OK).await;
+    tokio::time::timeout(Duration::from_secs(10), async {
+        loop {
+            let event = live.recv().await.unwrap();
+            if matches!(
+                event.event.body,
+                fabro_types::MainEventBody::RunCompleted(_)
+            ) {
+                break;
+            }
+            assert!(!matches!(
+                event.event.body,
+                fabro_types::MainEventBody::RunFailed(_)
+            ));
+        }
+    })
+    .await
+    .expect("workflow lifecycle should reach the main stream");
+    let mirrored = events.list_from_with_limit(1, 20).await.unwrap();
+    let mut names = mirrored
+        .iter()
+        .map(|event| event.event.event_name())
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+    assert_eq!(names, [
+        "fabro.run.completed",
+        "fabro.run.created",
+        "fabro.run.running",
+        "fabro.run.started"
+    ]);
+    let stream = state
+        .petri_projector
+        .stream_after(run_id, 0, 100)
+        .await
+        .unwrap();
+    for event in mirrored {
+        let properties = event.event.properties().unwrap();
+        assert_eq!(properties["run_id"], run_id.to_string());
+        assert!(
+            stream
+                .iter()
+                .any(|item| properties["source_event_id"] == format!("{}:{}", item.kind, item.id))
+        );
+    }
+    state.shutdown_token().cancel();
+}
+
+#[tokio::test]
+async fn main_event_mirror_create_is_persisted_before_the_api_returns() {
+    let state = test_app_state();
+    let app = crate::test_support::build_test_router(Arc::clone(&state));
+    let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await;
+    let body = post_run_intent(
+        &app,
+        json!({
+            "workflow_version_id": workflow_version_id,
+            "target": { "kind": "none" },
+            "args": {}
+        }),
+    )
+    .await;
+    let run_id: RunId = body["id"].as_str().unwrap().parse().unwrap();
+    let events = state
+        .stores
+        .runs
+        .main_events()
+        .list_from_with_limit(1, 10)
+        .await
+        .unwrap();
+    assert_eq!(events.len(), 1);
+    assert_eq!(events[0].seq, 1);
+    let fabro_types::MainEventBody::RunCreated(created) = &events[0].event.body else {
+        panic!("expected run creation");
+    };
+    assert_eq!(created.source.run_id, run_id);
+    assert_eq!(created.source.source_event_id, "platform:1");
+    let projection = state.load_run_projection(&run_id).await.unwrap();
+    assert_eq!(
+        created.provenance_subject.as_ref(),
+        Some(&projection.spec.provenance.subject)
+    );
+}
+
+#[tokio::test]
+async fn main_event_mirror_create_failure_retains_the_run_for_reconciliation() {
+    let state = test_app_state();
+    let app = crate::test_support::build_test_router(Arc::clone(&state));
+    let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await;
+    let pool = state.stores.run_summaries.pool();
+    sqlx::raw_sql("CREATE TRIGGER fail_main_event BEFORE INSERT ON main_events BEGIN SELECT RAISE(ABORT, 'injected failure'); END;")
+        .execute(&pool).await.unwrap();
+    let response = post_run_intent_response(
+        &app,
+        json!({
+            "workflow_version_id": workflow_version_id,
+            "target": { "kind": "none" },
+            "args": {}
+        }),
+    )
+    .await;
+    let body = response_json!(response, StatusCode::INTERNAL_SERVER_ERROR).await;
+    assert_eq!(body["errors"][0]["code"], "run_persistence_failed");
+    let runs = state.stores.run_summaries.list_identities().await.unwrap();
+    assert_eq!(runs.len(), 1);
+    assert!(
+        state
+            .stores
+            .runs
+            .main_events()
+            .list_from_with_limit(1, 10)
+            .await
+            .unwrap()
+            .is_empty()
+    );
+    sqlx::query("DROP TRIGGER fail_main_event")
+        .execute(&pool)
+        .await
+        .unwrap();
+    main_event_mirror::mirror_created(&state, runs[0].id)
+        .await
+        .unwrap();
+    assert_eq!(
+        state
+            .stores
+            .runs
+            .main_events()
+            .list_from_with_limit(1, 10)
+            .await
+            .unwrap()
+            .len(),
+        1
+    );
+}
+
 #[tokio::test]
 async fn post_runs_run_intent_creates_submitted_none_target_without_git_projection() {
     let state = test_app_state();

lib/components/fabro-petri/src/lib.rs

@@ -74,6 +74,7 @@ pub mod hooks;
 pub mod host_tools;
 pub mod http_store;
 pub mod interview;
+pub mod main_events;
 pub mod petri;
 pub mod platform_records;
 pub mod projection;

lib/components/fabro-petri/src/main_events.rs

@@ -0,0 +1,360 @@
+//! Native run records projected into Fabro's compact main lifecycle vocabulary.
+//! Storage and idempotent delivery remain the server's main event store
+//! boundary.
+use chrono::DateTime;
+use fabro_types::{
+    FailureReason, MainEventBody, MainEventCancelledProps, MainEventCompletedProps,
+    MainEventCreatedProps, MainEventFailedProps, MainEventLifecycleProps, MainEventSource,
+    MainEventStartedProps, RunProjection, RunStreamItem,
+};
+
+pub fn from_stream_item(
+    item: &RunStreamItem,
+    projection: &RunProjection,
+) -> anyhow::Result<Option<MainEventBody>> {
+    use fabro_store::platform_records::{PlatformRecord, RunLifecycleKind, StoredPlatformRecord};
+    use fabro_types::{FailureCategory, RunStatus, RunStreamItemKind};
+    let timestamp = i64::try_from(item.recorded_at)
+        .ok()
+        .and_then(DateTime::from_timestamp_millis)
+        .ok_or_else(|| {
+            anyhow::anyhow!(
+                "invalid timestamp in run {} stream item {}",
+                item.run_id,
+                item.id
+            )
+        })?;
+    let source = |name: &str| MainEventSource {
+        run_id:            item.run_id,
+        source_event_id:   format!("{}:{}", item.kind, item.id),
+        source_event_ts:   timestamp,
+        source_event_name: name.to_owned(),
+        actor:             None,
+    };
+    if item.kind == RunStreamItemKind::Petri && item.name() == Some("run.started") {
+        return Ok(Some(MainEventBody::RunStarted(MainEventStartedProps {
+            source:       source("run.started"),
+            name:         projection.spec.graph.name.clone(),
+            base_branch:  projection.spec.git.as_ref().map(|git| git.branch.clone()),
+            base_sha:     projection
+                .start
+                .as_ref()
+                .and_then(|start| start.base_sha.clone()),
+            run_branch:   projection
+                .start
+                .as_ref()
+                .and_then(|start| start.run_branch.clone()),
+            worktree_dir: None,
+            goal:         Some(projection.spec.graph.goal.clone()).filter(|goal| !goal.is_empty()),
+        })));
+    }
+    if item.kind != RunStreamItemKind::Platform {
+        return Ok(None);
+    }
+    let stored: StoredPlatformRecord = serde_json::from_value(item.item.clone())?;
+    Ok(match stored.record {
+        PlatformRecord::RunCreated(created) => {
+            Some(MainEventBody::RunCreated(MainEventCreatedProps {
+                source:             source("run.created"),
+                title:              created.title,
+                workflow_slug:      created.spec.workflow_slug,
+                automation:         created.spec.automation,
+                git:                created.spec.git,
+                parent_id:          created.parent_id,
+                provenance_subject: Some(created.spec.provenance.subject),
+            }))
+        }
+        PlatformRecord::RunLifecycle(record) => {
+            let conclusion = projection.conclusion.as_ref();
+            let usage = conclusion.and_then(|conclusion| conclusion.usage);
+            let final_git_commit_sha =
+                conclusion.and_then(|conclusion| conclusion.final_git_commit_sha.clone());
+            match record.transition {
+                RunLifecycleKind::Running => {
+                    Some(MainEventBody::RunRunning(MainEventLifecycleProps {
+                        source: source("run.running"),
+                    }))
+                }
+                RunLifecycleKind::Paused => {
+                    Some(MainEventBody::RunPaused(MainEventLifecycleProps {
+                        source: source("run.paused"),
+                    }))
+                }
+                RunLifecycleKind::Unpaused => {
+                    Some(MainEventBody::RunUnpaused(MainEventLifecycleProps {
+                        source: source("run.unpaused"),
+                    }))
+                }
+                RunLifecycleKind::Succeeded => {
+                    let Some(RunStatus::Succeeded { reason }) = record.status else {
+                        anyhow::bail!("successful lifecycle record has no success reason");
+                    };
+                    Some(MainEventBody::RunCompleted(MainEventCompletedProps {
+                        source: source("run.completed"),
+                        status: "succeeded".into(),
+                        reason,
+                        total_usd_micros: usage
+                            .and_then(|usage| usage.cost)
+                            .map(|cost| i64::try_from(cost.usd_micros).unwrap_or(i64::MAX)),
+                        final_git_commit_sha,
+                        automation: projection.spec.automation.clone(),
+                        usage,
+                    }))
+                }
+                RunLifecycleKind::Failed | RunLifecycleKind::Dead => {
+                    let failure = conclusion.and_then(|conclusion| conclusion.failure.as_ref());
+                    let reason = match record.status {
+                        Some(RunStatus::Failed { reason }) => reason,
+                        _ => failure.map_or(FailureReason::WorkflowError, |failure| failure.reason),
+                    };
+                    let category = failure.map_or(
+                        if reason == FailureReason::Cancelled {
+                            FailureCategory::Canceled
+                        } else {
+                            FailureCategory::Deterministic
+                        },
+                        |failure| failure.detail.category,
+                    );
+                    let message = record
+                        .reason
+                        .or_else(|| failure.map(|failure| failure.detail.message.clone()))
+                        .unwrap_or_default();
+                    if reason == FailureReason::Cancelled {
+                        Some(MainEventBody::RunCancelled(MainEventCancelledProps {
+                            source: source("run.cancelled"),
+                            reason,
+                            category,
+                            message,
+                            final_git_commit_sha,
+                            automation: projection.spec.automation.clone(),
+                            usage,
+                        }))
+                    } else {
+                        Some(MainEventBody::RunFailed(MainEventFailedProps {
+                            source: source("run.failed"),
+                            reason,
+                            category,
+                            message,
+                            final_git_commit_sha,
+                            automation: projection.spec.automation.clone(),
+                            usage,
+                        }))
+                    }
+                }
+                _ => None,
+            }
+        }
+        _ => None,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use chrono::Utc;
+    use fabro_store::platform_records::{PlatformRecord, StoredPlatformRecord};
+    use fabro_types::{FailureCategory, FailureReason, MainEventBody, fixtures, test_support};
+    fn native_item(record: PlatformRecord) -> fabro_types::RunStreamItem {
+        fabro_types::RunStreamItem {
+            run_id:      fixtures::RUN_1,
+            stream_seq:  42,
+            kind:        fabro_types::RunStreamItemKind::Platform,
+            id:          "7".into(),
+            recorded_at: 1_800_000_000_000,
+            item:        serde_json::to_value(StoredPlatformRecord {
+                seq: 7,
+                recorded_at: 1_800_000_000_000,
+                record,
+                position: None,
+            })
+            .unwrap(),
+        }
+    }
+
+    #[test]
+    fn native_creation_uses_the_immutable_record_not_current_projection() {
+        use fabro_store::platform_records::{PlatformRecord, RunCreatedRecord};
+        let spec = test_support::test_run_spec();
+        let projection =
+            fabro_types::RunProjection::new("Renamed later".into(), spec.clone(), Utc::now());
+        let item = native_item(PlatformRecord::RunCreated(RunCreatedRecord {
+            spec,
+            title: Some("Original title".into()),
+            parent_id: None,
+            retried_from: None,
+            web_url: None,
+        }));
+        let Some(MainEventBody::RunCreated(created)) =
+            super::from_stream_item(&item, &projection).unwrap()
+        else {
+            panic!("created");
+        };
+        assert_eq!(created.title.as_deref(), Some("Original title"));
+        assert_eq!(created.source.source_event_id, "platform:7");
+        assert_eq!(
+            created.source.source_event_ts.timestamp_millis(),
+            1_800_000_000_000
+        );
+    }
+
+    #[test]
+    fn native_cancel_request_is_not_a_terminal_event() {
+        use fabro_store::platform_records::{PlatformRecord, RunLifecycleKind, RunLifecycleRecord};
+        let projection = fabro_types::RunProjection::new(
+            "Test".into(),
+            test_support::test_run_spec(),
+            Utc::now(),
+        );
+        let request = native_item(PlatformRecord::RunLifecycle(RunLifecycleRecord::new(
+            RunLifecycleKind::CancelRequested,
+        )));
+        assert!(
+            super::from_stream_item(&request, &projection)
+                .unwrap()
+                .is_none()
+        );
+        let mut record = RunLifecycleRecord::new(RunLifecycleKind::Failed).with_status(
+            fabro_types::RunStatus::Failed {
+                reason: FailureReason::Cancelled,
+            },
+        );
+        record.reason = Some("Stopped by user".into());
+        let cancelled = native_item(PlatformRecord::RunLifecycle(record));
+        let Some(MainEventBody::RunCancelled(cancelled)) =
+            super::from_stream_item(&cancelled, &projection).unwrap()
+        else {
+            panic!("cancelled");
+        };
+        assert_eq!(cancelled.reason, FailureReason::Cancelled);
+        assert_eq!(cancelled.category, FailureCategory::Canceled);
+        assert_eq!(cancelled.message, "Stopped by user");
+        assert_eq!(cancelled.usage, None);
+    }
+
+    #[test]
+    fn native_lifecycle_maps_success_usage_and_live_transitions() {
+        use fabro_store::platform_records::{RunLifecycleKind, RunLifecycleRecord};
+        use fabro_types::{Conclusion, RunStatus, StageOutcome, SuccessReason};
+        use lithos_llm::types::{Cost, CostSource, TokenCounts, Usage};
+
+        let mut projection = fabro_types::RunProjection::new(
+            "Test".into(),
+            test_support::test_run_spec(),
+            Utc::now(),
+        );
+        let usage = Usage {
+            tokens: TokenCounts {
+                input: 10,
+                cache_read: 4,
+                ..TokenCounts::default()
+            },
+            cost:   Some(Cost {
+                usd_micros: 42,
+                source:     CostSource::Catalog,
+            }),
+        };
+        let mut conclusion = Conclusion::outcome_only(Utc::now(), StageOutcome::Succeeded, None);
+        conclusion.usage = Some(usage);
+        conclusion.final_git_commit_sha = Some("abc123".into());
+        projection.conclusion = Some(conclusion);
+        let item = native_item(PlatformRecord::RunLifecycle(
+            RunLifecycleRecord::new(RunLifecycleKind::Succeeded).with_status(
+                RunStatus::Succeeded {
+                    reason: SuccessReason::Completed,
+                },
+            ),
+        ));
+        let Some(MainEventBody::RunCompleted(completed)) =
+            super::from_stream_item(&item, &projection).unwrap()
+        else {
+            panic!("completed");
+        };
+        assert_eq!(completed.total_usd_micros, Some(42));
+        assert_eq!(completed.usage, Some(usage));
+        assert_eq!(completed.final_git_commit_sha.as_deref(), Some("abc123"));
+
+        for (transition, name) in [
+            (RunLifecycleKind::Running, "fabro.run.running"),
+            (RunLifecycleKind::Paused, "fabro.run.paused"),
+            (RunLifecycleKind::Unpaused, "fabro.run.unpaused"),
+        ] {
+            let item = native_item(PlatformRecord::RunLifecycle(RunLifecycleRecord::new(
+                transition,
+            )));
+            assert_eq!(
+                super::from_stream_item(&item, &projection)
+                    .unwrap()
+                    .unwrap()
+                    .event_name(),
+                name
+            );
+        }
+    }
+
+    #[test]
+    fn native_started_maps_petri_identity_and_ignores_detailed_events() {
+        use fabro_types::{RunStreamItemKind, StartRecord};
+        use serde_json::json;
+
+        let mut projection = fabro_types::RunProjection::new(
+            "Test".into(),
+            test_support::test_run_spec(),
+            Utc::now(),
+        );
+        projection.start = Some(StartRecord {
+            start_time: Utc::now(),
+            base_sha:   Some("abc123".into()),
+            run_branch: Some("fabro/run/test".into()),
+        });
+        let mut item = fabro_types::RunStreamItem {
+            run_id:      fixtures::RUN_1,
+            stream_seq:  3,
+            kind:        RunStreamItemKind::Petri,
+            id:          "coordinator/3/0".into(),
+            recorded_at: 1_800_000_000_000,
+            item:        json!({"record": {"body": {"event": "run.started"}}}),
+        };
+        let Some(MainEventBody::RunStarted(started)) =
+            super::from_stream_item(&item, &projection).unwrap()
+        else {
+            panic!("started");
+        };
+        assert_eq!(started.source.source_event_id, "petri:coordinator/3/0");
+        assert_eq!(started.base_sha.as_deref(), Some("abc123"));
+        assert_eq!(started.run_branch.as_deref(), Some("fabro/run/test"));
+        for event in ["step.started", "step.progress.recorded", "run.finished"] {
+            item.item["record"]["body"]["event"] = json!(event);
+            assert!(
+                super::from_stream_item(&item, &projection)
+                    .unwrap()
+                    .is_none()
+            );
+        }
+    }
+
+    #[test]
+    fn native_failed_and_dead_remain_distinct_from_cancellation() {
+        use fabro_store::platform_records::{RunLifecycleKind, RunLifecycleRecord};
+        use fabro_types::RunStatus;
+
+        let projection = fabro_types::RunProjection::new(
+            "Test".into(),
+            test_support::test_run_spec(),
+            Utc::now(),
+        );
+        for transition in [RunLifecycleKind::Failed, RunLifecycleKind::Dead] {
+            let mut record = RunLifecycleRecord::new(transition).with_status(RunStatus::Failed {
+                reason: FailureReason::WorkflowError,
+            });
+            record.reason = Some("worker failed".into());
+            let item = native_item(PlatformRecord::RunLifecycle(record));
+            let Some(MainEventBody::RunFailed(failed)) =
+                super::from_stream_item(&item, &projection).unwrap()
+            else {
+                panic!("failed");
+            };
+            assert_eq!(failed.message, "worker failed");
+            assert_eq!(failed.reason, FailureReason::WorkflowError);
+            assert_eq!(failed.usage, None);
+        }
+    }
+}

lib/components/fabro-store/src/database.rs

@@ -3,19 +3,21 @@
 //!
 //! A run's history is Petri's records (`petri_records`) and Fabro's
 //! platform records; its view is the projection the projector commits.
-//! This handle carries the two stores every reader of a run reaches, and
-//! the run-level operations that span them.
+//! This handle carries the stores every reader of a run reaches, the
+//! independent main lifecycle stream, and the run-level operations that
+//! span them.
 
 use std::sync::Arc;
 
 use fabro_types::{RunId, RunProjection};
 
-use crate::{BlobStore, PlatformRecordHook, Result, RunSummaryStore};
+use crate::{BlobStore, MainEventStore, PlatformRecordHook, Result, RunSummaryStore};
 
 #[derive(Clone)]
 pub struct Database {
     blobs:             Arc<BlobStore>,
     run_summary_store: Arc<RunSummaryStore>,
+    main_events:       Arc<MainEventStore>,
 }
 
 impl std::fmt::Debug for Database {
@@ -28,6 +30,7 @@ impl Database {
     #[must_use]
     pub fn new(blobs: Arc<BlobStore>, run_summary_store: Arc<RunSummaryStore>) -> Self {
         Self {
+            main_events: Arc::new(MainEventStore::new(run_summary_store.pool())),
             blobs,
             run_summary_store,
         }
@@ -43,6 +46,11 @@ impl Database {
         Arc::clone(&self.blobs)
     }
 
+    #[must_use]
+    pub fn main_events(&self) -> Arc<MainEventStore> {
+        Arc::clone(&self.main_events)
+    }
+
     /// The run's projection: the one its projector last committed over
     /// Petri's records and the platform records, or `None` before the
     /// first view pass commits (or for no such run).

lib/components/fabro-store/src/lib.rs

@@ -5,6 +5,7 @@ mod blob_store;
 mod database;
 mod error;
 mod keyed_mutex;
+mod main_event_store;
 pub mod platform_records;
 mod run_session_event_store;
 mod run_session_record_store;
@@ -31,6 +32,7 @@ pub use fabro_types::{
     BlobHash, PendingInterviewRecord, Run, RunProjection, StageId, StageProjection,
 };
 pub use keyed_mutex::{KeyedMutex, KeyedMutexGuard};
+pub use main_event_store::{AppendMainEventOutcome, MainEventSourceKey, MainEventStore};
 pub use platform_records::{
     PlatformRecord, PlatformRecordHook, PlatformRecordKind, PlatformRecordStore, StagePosition,
     StoredPlatformRecord,

lib/components/fabro-store/src/main_event_store.rs

@@ -0,0 +1,353 @@
+//! The compact server-wide event stream, independent of any run's lifetime.
+//!
+//! A source key and its event commit together. The run stream is an outbox:
+//! retrying a delivery after a crash returns the original event without
+//! allocating another global sequence or publishing another live message.
+
+use std::sync::Arc;
+
+use chrono::Utc;
+use fabro_types::{MainEvent, MainEventBody, MainEventEnvelope, RunId};
+use sqlx::SqlitePool;
+use tokio::sync::{Mutex, broadcast};
+use uuid::Uuid;
+
+use crate::{Error, Result};
+
+const LIVE_CAPACITY: usize = 1024;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum MainEventSourceKey {
+    RunStream { run_id: RunId, item_id: String },
+}
+
+impl MainEventSourceKey {
+    fn store_key(&self) -> String {
+        let Self::RunStream { run_id, item_id } = self;
+        format!("runs:{run_id}:{item_id}")
+    }
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum AppendMainEventOutcome {
+    Appended(MainEventEnvelope),
+    Duplicate(MainEventEnvelope),
+}
+
+#[derive(Clone)]
+pub struct MainEventStore {
+    pool:        SqlitePool,
+    append_lock: Arc<Mutex<()>>,
+    live:        broadcast::Sender<MainEventEnvelope>,
+}
+
+impl std::fmt::Debug for MainEventStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MainEventStore").finish_non_exhaustive()
+    }
+}
+
+impl MainEventStore {
+    #[must_use]
+    pub fn new(pool: SqlitePool) -> Self {
+        let (live, _) = broadcast::channel(LIVE_CAPACITY);
+        Self {
+            pool,
+            append_lock: Arc::new(Mutex::new(())),
+            live,
+        }
+    }
+
+    pub async fn append(&self, body: MainEventBody) -> Result<MainEventEnvelope> {
+        match self.append_inner(None, body).await? {
+            AppendMainEventOutcome::Appended(event) | AppendMainEventOutcome::Duplicate(event) => {
+                Ok(event)
+            }
+        }
+    }
+
+    pub async fn append_once(
+        &self,
+        source: MainEventSourceKey,
+        body: MainEventBody,
+    ) -> Result<AppendMainEventOutcome> {
+        self.append_inner(Some(source.store_key()), body).await
+    }
+
+    async fn append_inner(
+        &self,
+        source_key: Option<String>,
+        body: MainEventBody,
+    ) -> Result<AppendMainEventOutcome> {
+        // Keep broadcasts in commit order for all clones of this handle.
+        // SQLite's write lock also serializes independent store handles.
+        let _guard = self.append_lock.lock().await;
+        let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?;
+        let existing = match &source_key {
+            Some(source_key) => {
+                sqlx::query_as::<_, (i64, String)>(
+                    "SELECT seq, event_json FROM main_events WHERE source_key = ?",
+                )
+                .bind(source_key)
+                .fetch_optional(&mut *transaction)
+                .await?
+            }
+            None => None,
+        };
+        if let Some((seq, json)) = existing {
+            let event = decode_event(seq, &json)?;
+            transaction.commit().await?;
+            return Ok(AppendMainEventOutcome::Duplicate(event));
+        }
+        let last_seq: Option<i64> = sqlx::query_scalar("SELECT MAX(seq) FROM main_events")
+            .fetch_one(&mut *transaction)
+            .await?;
+        let seq = u32::try_from(last_seq.unwrap_or(0))
+            .ok()
+            .and_then(|last| last.checked_add(1))
+            .ok_or(Error::EventSequenceExhausted { max_seq: u32::MAX })?;
+        let event = MainEvent {
+            id: Uuid::now_v7().to_string(),
+            ts: Utc::now(),
+            body,
+        };
+        sqlx::query("INSERT INTO main_events (seq, source_key, event_json) VALUES (?, ?, ?)")
+            .bind(i64::from(seq))
+            .bind(source_key)
+            .bind(serde_json::to_string(&event)?)
+            .execute(&mut *transaction)
+            .await?;
+        transaction.commit().await?;
+        let envelope = MainEventEnvelope { seq, event };
+        let _ = self.live.send(envelope.clone());
+        Ok(AppendMainEventOutcome::Appended(envelope))
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<MainEventEnvelope> {
+        self.live.subscribe()
+    }
+
+    /// Replay from an inclusive sequence, returning at most `limit + 1`
+    /// events so a consumer can determine whether another page exists.
+    pub async fn list_from_with_limit(
+        &self,
+        start_seq: u32,
+        limit: usize,
+    ) -> Result<Vec<MainEventEnvelope>> {
+        let rows = sqlx::query_as::<_, (i64, String)>(
+            "SELECT seq, event_json FROM main_events WHERE seq >= ? ORDER BY seq LIMIT ?",
+        )
+        .bind(i64::from(start_seq))
+        .bind(i64::try_from(limit.saturating_add(1)).unwrap_or(i64::MAX))
+        .fetch_all(&self.pool)
+        .await?;
+        rows.into_iter()
+            .map(|(seq, json)| decode_event(seq, &json))
+            .collect()
+    }
+}
+
+fn decode_event(seq: i64, json: &str) -> Result<MainEventEnvelope> {
+    Ok(MainEventEnvelope {
+        seq:   u32::try_from(seq)
+            .map_err(|_| Error::EventSequenceExhausted { max_seq: u32::MAX })?,
+        event: serde_json::from_str(json)?,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use fabro_types::{MainEventLifecycleProps, MainEventSource, fixtures, test_support};
+    use tokio::sync::broadcast::error::TryRecvError;
+
+    use super::*;
+    use crate::platform_records::{PlatformRecord, RunCreatedRecord};
+    use crate::test_support as store_test_support;
+
+    fn body(run_id: RunId) -> MainEventBody {
+        MainEventBody::RunRunning(MainEventLifecycleProps {
+            source: MainEventSource {
+                run_id,
+                source_event_id: "platform:1".into(),
+                source_event_ts: Utc::now(),
+                source_event_name: "run.running".into(),
+                actor: None,
+            },
+        })
+    }
+
+    fn source(run_id: RunId) -> MainEventSourceKey {
+        MainEventSourceKey::RunStream {
+            run_id,
+            item_id: "platform:1".into(),
+        }
+    }
+
+    #[tokio::test]
+    async fn main_events_replay_and_broadcast_in_global_order() {
+        let database = store_test_support::test_database();
+        let store = database.main_events();
+        let mut live = store.subscribe();
+        let first = store.append(body(fixtures::RUN_1)).await.unwrap();
+        let second = store.append(body(fixtures::RUN_2)).await.unwrap();
+        let third = store.append(body(fixtures::RUN_1)).await.unwrap();
+        assert_eq!((first.seq, second.seq, third.seq), (1, 2, 3));
+        assert_ne!(first.event.id, second.event.id);
+        assert_eq!(live.recv().await.unwrap(), first);
+        assert_eq!(live.recv().await.unwrap(), second);
+        assert_eq!(live.recv().await.unwrap(), third);
+        assert_eq!(store.list_from_with_limit(2, 1).await.unwrap(), vec![
+            second, third
+        ]);
+        assert!(store.list_from_with_limit(4, 10).await.unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn main_event_source_survives_reopen_and_is_scoped_to_run() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("fabro.sqlite3");
+        let database = fabro_db::Database::connect(&path).await.unwrap();
+        database.migrate().await.unwrap();
+        let store = MainEventStore::new(database.clone_pool());
+        let AppendMainEventOutcome::Appended(first) = store
+            .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1))
+            .await
+            .unwrap()
+        else {
+            panic!("first append");
+        };
+        database.pool().close().await;
+        drop(store);
+        drop(database);
+
+        let database = fabro_db::Database::connect(&path).await.unwrap();
+        database.migrate().await.unwrap();
+        let reopened = MainEventStore::new(database.clone_pool());
+        let mut live = reopened.subscribe();
+        let duplicate = reopened
+            .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1))
+            .await
+            .unwrap();
+        assert_eq!(duplicate, AppendMainEventOutcome::Duplicate(first.clone()));
+        assert_eq!(live.try_recv(), Err(TryRecvError::Empty));
+        let AppendMainEventOutcome::Appended(second) = reopened
+            .append_once(source(fixtures::RUN_2), body(fixtures::RUN_2))
+            .await
+            .unwrap()
+        else {
+            panic!("a different run has a different source key");
+        };
+        assert_eq!(second.seq, 2);
+        assert_eq!(reopened.list_from_with_limit(1, 10).await.unwrap(), vec![
+            first, second
+        ]);
+    }
+
+    #[tokio::test]
+    async fn concurrent_main_event_sources_consume_one_sequence_across_handles() {
+        let directory = tempfile::tempdir().unwrap();
+        let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3"))
+            .await
+            .unwrap();
+        database.migrate().await.unwrap();
+        let left = MainEventStore::new(database.clone_pool());
+        let right = MainEventStore::new(database.clone_pool());
+        let (first, second) = tokio::join!(
+            left.append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)),
+            right.append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)),
+        );
+        assert_ne!(
+            matches!(first.unwrap(), AppendMainEventOutcome::Appended(_)),
+            matches!(second.unwrap(), AppendMainEventOutcome::Appended(_))
+        );
+        assert_eq!(left.list_from_with_limit(1, 10).await.unwrap().len(), 1);
+        assert_eq!(right.append(body(fixtures::RUN_2)).await.unwrap().seq, 2);
+    }
+
+    #[tokio::test]
+    async fn failed_main_event_append_leaves_no_receipt_or_broadcast() {
+        let database = store_test_support::test_database();
+        let store = database.main_events();
+        let pool = database.run_summary_store().pool();
+        // Connect the lazy fixture and then inject a database write failure.
+        store.list_from_with_limit(1, 1).await.unwrap();
+        sqlx::raw_sql("CREATE TRIGGER fail_main_event BEFORE INSERT ON main_events BEGIN SELECT RAISE(ABORT, 'injected failure'); END;")
+            .execute(&pool).await.unwrap();
+        let mut live = store.subscribe();
+        assert!(
+            store
+                .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1))
+                .await
+                .is_err()
+        );
+        assert_eq!(live.try_recv(), Err(TryRecvError::Empty));
+        sqlx::query("DROP TRIGGER fail_main_event")
+            .execute(&pool)
+            .await
+            .unwrap();
+        let AppendMainEventOutcome::Appended(event) = store
+            .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1))
+            .await
+            .unwrap()
+        else {
+            panic!("retry must append");
+        };
+        assert_eq!(event.seq, 1);
+        assert_eq!(live.recv().await.unwrap(), event);
+    }
+
+    #[tokio::test]
+    async fn main_event_sequence_exhaustion_preserves_existing_events() {
+        let database = store_test_support::test_database();
+        let store = database.main_events();
+        let first = store.append(body(fixtures::RUN_1)).await.unwrap();
+        sqlx::query("UPDATE main_events SET seq = ? WHERE seq = 1")
+            .bind(i64::from(u32::MAX))
+            .execute(&database.run_summary_store().pool())
+            .await
+            .unwrap();
+        assert!(matches!(
+            store.append(body(fixtures::RUN_2)).await,
+            Err(Error::EventSequenceExhausted { .. })
+        ));
+        let events = store.list_from_with_limit(u32::MAX, 1).await.unwrap();
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].event, first.event);
+    }
+
+    #[tokio::test]
+    async fn deleting_a_run_keeps_its_main_events_and_source_receipts() {
+        let database = store_test_support::test_database();
+        let spec = test_support::test_run_spec();
+        let record = PlatformRecord::RunCreated(RunCreatedRecord {
+            spec,
+            title: None,
+            parent_id: None,
+            retried_from: None,
+            web_url: None,
+        });
+        database
+            .run_summary_store()
+            .platform_records()
+            .append(&fixtures::RUN_1, &record, None)
+            .await
+            .unwrap();
+        let events = database.main_events();
+        let first = events
+            .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1))
+            .await
+            .unwrap();
+        database.delete_run(&fixtures::RUN_1).await.unwrap();
+        let duplicate = events
+            .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1))
+            .await
+            .unwrap();
+        let AppendMainEventOutcome::Appended(first) = first else {
+            panic!("first append");
+        };
+        assert_eq!(duplicate, AppendMainEventOutcome::Duplicate(first.clone()));
+        assert_eq!(events.list_from_with_limit(1, 10).await.unwrap(), vec![
+            first
+        ]);
+    }
+}

lib/components/fabro-store/src/test_support/mod.rs

@@ -27,6 +27,7 @@ const RUN_SUMMARY_MIGRATIONS: &[&str] = &[
     fabro_db::RUNS_MIGRATION_SQL,
     fabro_db::DROP_RUN_EVENTS_MIGRATION_SQL,
     fabro_db::PETRI_PROJECTION_MIGRATION_SQL,
+    fabro_db::MAIN_EVENTS_MIGRATION_SQL,
 ];
 
 /// Returns an isolated SQLite run-summary store backed by its own in-memory

lib/foundation/fabro-db/migrations/2026101601_main_events.sql

@@ -0,0 +1,7 @@
+-- The global lifecycle mirror outlives individual runs. Deliberately no
+-- foreign key to runs: deleting a run must keep its events and source keys.
+CREATE TABLE main_events (
+    seq INTEGER PRIMARY KEY CHECK (seq > 0 AND seq <= 4294967295),
+    source_key TEXT UNIQUE,
+    event_json TEXT NOT NULL
+);

lib/foundation/fabro-db/src/lib.rs

@@ -42,6 +42,10 @@ pub const RUN_SESSION_RECORDS_MIGRATION_SQL: &str =
 pub const RUN_SESSION_EVENTS_MIGRATION_SQL: &str =
     include_str!("../migrations/2026091802_run_session_events.sql");
 
+/// The global lifecycle stream schema, also used by isolated store fixtures.
+pub const MAIN_EVENTS_MIGRATION_SQL: &str =
+    include_str!("../migrations/2026101601_main_events.sql");
+
 /// The Petri run record migration (`petri_runs`, `petri_records`), exposed
 /// so fixtures in other crates can install the production schema without a
 /// filesystem path into this crate.

lib/foundation/fabro-types/src/lib.rs

@@ -19,6 +19,7 @@ mod id;
 mod input_scalar;
 pub mod interview;
 pub mod llm_backend;
+pub mod main_event;
 pub mod manifest_path;
 pub mod mcp_store;
 pub mod model_test;
@@ -95,6 +96,11 @@ pub use interview::{
     ReviewTargetKind,
 };
 pub use llm_backend::AgentBackend;
+pub use main_event::{
+    MainEvent, MainEventBody, MainEventCancelledProps, MainEventCompletedProps,
+    MainEventCreatedProps, MainEventEnvelope, MainEventFailedProps, MainEventLifecycleProps,
+    MainEventSource, MainEventStartedProps,
+};
 pub use manifest_path::{ManifestPath, ManifestPathParseError};
 pub use mcp_store::{
     McpServerDefinition, McpServerDraft, McpServerId, McpServerReplace, McpServerRevision,

lib/foundation/fabro-types/src/main_event.rs

@@ -0,0 +1,311 @@
+use chrono::{DateTime, Utc};
+use lithos_llm::types::Usage;
+use serde::de::Error as DeError;
+use serde::ser::Error as SerError;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use serde_json::{Map, Value, json};
+
+use crate::{
+    AutomationRef, FailureCategory, FailureReason, GitContext, Principal, RunId, SuccessReason,
+};
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventEnvelope {
+    pub seq:   u32,
+    pub event: MainEvent,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct MainEvent {
+    pub id:   String,
+    pub ts:   DateTime<Utc>,
+    pub body: MainEventBody,
+}
+
+#[allow(
+    clippy::large_enum_variant,
+    reason = "Main event bodies stay inline to match the tagged wire format."
+)]
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+#[serde(tag = "event", content = "properties")]
+pub enum MainEventBody {
+    #[serde(rename = "fabro.run.created")]
+    RunCreated(MainEventCreatedProps),
+    #[serde(rename = "fabro.run.started")]
+    RunStarted(MainEventStartedProps),
+    #[serde(rename = "fabro.run.running")]
+    RunRunning(MainEventLifecycleProps),
+    #[serde(rename = "fabro.run.completed")]
+    RunCompleted(MainEventCompletedProps),
+    #[serde(rename = "fabro.run.failed")]
+    RunFailed(MainEventFailedProps),
+    #[serde(rename = "fabro.run.cancelled")]
+    RunCancelled(MainEventCancelledProps),
+    #[serde(rename = "fabro.run.paused")]
+    RunPaused(MainEventLifecycleProps),
+    #[serde(rename = "fabro.run.unpaused")]
+    RunUnpaused(MainEventLifecycleProps),
+    Unknown {
+        name:       String,
+        properties: Value,
+    },
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventSource {
+    pub run_id:            RunId,
+    pub source_event_id:   String,
+    pub source_event_ts:   DateTime<Utc>,
+    pub source_event_name: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub actor:             Option<Principal>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventCreatedProps {
+    #[serde(flatten)]
+    pub source:             MainEventSource,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub title:              Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub workflow_slug:      Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub automation:         Option<AutomationRef>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub git:                Option<GitContext>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub parent_id:          Option<RunId>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub provenance_subject: Option<Principal>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventLifecycleProps {
+    #[serde(flatten)]
+    pub source: MainEventSource,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventStartedProps {
+    #[serde(flatten)]
+    pub source:       MainEventSource,
+    pub name:         String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub base_branch:  Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub base_sha:     Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub run_branch:   Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub worktree_dir: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub goal:         Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventCompletedProps {
+    #[serde(flatten)]
+    pub source:               MainEventSource,
+    pub status:               String,
+    pub reason:               SuccessReason,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub total_usd_micros:     Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub final_git_commit_sha: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub automation:           Option<AutomationRef>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub usage:                Option<Usage>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventFailedProps {
+    #[serde(flatten)]
+    pub source:               MainEventSource,
+    pub reason:               FailureReason,
+    pub category:             FailureCategory,
+    pub message:              String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub final_git_commit_sha: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub automation:           Option<AutomationRef>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub usage:                Option<Usage>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct MainEventCancelledProps {
+    #[serde(flatten)]
+    pub source:               MainEventSource,
+    pub reason:               FailureReason,
+    pub category:             FailureCategory,
+    pub message:              String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub final_git_commit_sha: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub automation:           Option<AutomationRef>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub usage:                Option<Usage>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct MainEventRaw {
+    id:         String,
+    ts:         DateTime<Utc>,
+    event:      String,
+    #[serde(default = "default_properties")]
+    properties: Value,
+}
+
+struct MainEventParts<'a> {
+    id:         String,
+    ts:         DateTime<Utc>,
+    event:      &'a str,
+    properties: &'a Value,
+}
+
+impl MainEventBody {
+    pub fn event_name(&self) -> &str {
+        match self {
+            Self::RunCreated(_) => "fabro.run.created",
+            Self::RunStarted(_) => "fabro.run.started",
+            Self::RunRunning(_) => "fabro.run.running",
+            Self::RunCompleted(_) => "fabro.run.completed",
+            Self::RunFailed(_) => "fabro.run.failed",
+            Self::RunCancelled(_) => "fabro.run.cancelled",
+            Self::RunPaused(_) => "fabro.run.paused",
+            Self::RunUnpaused(_) => "fabro.run.unpaused",
+            Self::Unknown { name, .. } => name.as_str(),
+        }
+    }
+
+    fn properties_value(&self) -> serde_json::Result<Value> {
+        if let Self::Unknown { properties, .. } = self {
+            return Ok(properties.clone());
+        }
+
+        match serde_json::to_value(self)? {
+            Value::Object(mut map) => {
+                Ok(map.remove("properties").unwrap_or_else(default_properties))
+            }
+            _ => Ok(default_properties()),
+        }
+    }
+}
+
+fn is_known_main_event_name(event: &str) -> bool {
+    matches!(
+        event,
+        "fabro.run.created"
+            | "fabro.run.started"
+            | "fabro.run.running"
+            | "fabro.run.completed"
+            | "fabro.run.failed"
+            | "fabro.run.cancelled"
+            | "fabro.run.paused"
+            | "fabro.run.unpaused"
+    )
+}
+
+impl MainEvent {
+    pub fn from_value(value: Value) -> serde_json::Result<Self> {
+        let raw: MainEventRaw = serde_json::from_value(value)?;
+        Self::from_parts(MainEventParts {
+            id:         raw.id,
+            ts:         raw.ts,
+            event:      &raw.event,
+            properties: &raw.properties,
+        })
+    }
+
+    pub fn from_ref(value: &Value) -> serde_json::Result<Self> {
+        let obj = value.as_object().ok_or_else(|| {
+            <serde_json::Error as DeError>::custom("main event must be a JSON object")
+        })?;
+        let id = obj.get("id").and_then(Value::as_str).ok_or_else(|| {
+            <serde_json::Error as DeError>::custom("missing or non-string field: id")
+        })?;
+        let ts = obj
+            .get("ts")
+            .ok_or_else(|| <serde_json::Error as DeError>::custom("missing field: ts"))
+            .and_then(DateTime::<Utc>::deserialize)?;
+        let event = obj.get("event").and_then(Value::as_str).ok_or_else(|| {
+            <serde_json::Error as DeError>::custom("missing or non-string field: event")
+        })?;
+        let properties = obj
+            .get("properties")
+            .cloned()
+            .unwrap_or_else(default_properties);
+        Self::from_parts(MainEventParts {
+            id: id.to_string(),
+            ts,
+            event,
+            properties: &properties,
+        })
+    }
+
+    fn from_parts(parts: MainEventParts<'_>) -> serde_json::Result<Self> {
+        let body_payload = json!({
+            "event": parts.event,
+            "properties": parts.properties,
+        });
+        let body: MainEventBody = match serde_json::from_value(body_payload) {
+            Ok(body) => body,
+            Err(err) if is_known_main_event_name(parts.event) => return Err(err),
+            Err(_) => MainEventBody::Unknown {
+                name:       parts.event.to_string(),
+                properties: parts.properties.clone(),
+            },
+        };
+        Ok(Self {
+            id: parts.id,
+            ts: parts.ts,
+            body,
+        })
+    }
+
+    pub fn to_value(&self) -> serde_json::Result<Value> {
+        let mut map = Map::new();
+        map.insert("id".to_string(), Value::String(self.id.clone()));
+        map.insert("ts".to_string(), serde_json::to_value(self.ts)?);
+        map.insert(
+            "event".to_string(),
+            Value::String(self.body.event_name().to_string()),
+        );
+        map.insert("properties".to_string(), self.body.properties_value()?);
+        Ok(Value::Object(map))
+    }
+
+    pub fn event_name(&self) -> &str {
+        self.body.event_name()
+    }
+
+    pub fn properties(&self) -> serde_json::Result<Value> {
+        self.body.properties_value()
+    }
+}
+
+impl Serialize for MainEvent {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.to_value()
+            .map_err(S::Error::custom)?
+            .serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for MainEvent {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let value = Value::deserialize(deserializer)?;
+        Self::from_value(value).map_err(D::Error::custom)
+    }
+}
+
+fn default_properties() -> Value {
+    Value::Object(Map::new())
+}

lib/foundation/fabro-types/tests/main_event_serde.rs

@@ -0,0 +1,202 @@
+use chrono::{TimeZone, Utc};
+use fabro_types::{
+    AuthMethod, FailureCategory, FailureReason, IdpIdentity, MainEvent, MainEventBody,
+    MainEventCancelledProps, MainEventCompletedProps, MainEventCreatedProps, MainEventFailedProps,
+    MainEventLifecycleProps, MainEventSource, MainEventStartedProps, Principal, RunId,
+    SuccessReason,
+};
+use serde_json::json;
+
+fn run_id() -> RunId {
+    "01JT56VE4Z5NZ814GZN2JZD65A"
+        .parse()
+        .expect("test run id should parse")
+}
+
+fn actor() -> Principal {
+    Principal::user(
+        IdpIdentity::new("https://github.com", "123").expect("test identity should construct"),
+        "fabian".to_string(),
+        AuthMethod::Github,
+    )
+}
+
+fn source(source_event_name: &str) -> MainEventSource {
+    MainEventSource {
+        run_id:            run_id(),
+        source_event_id:   format!("evt-{source_event_name}"),
+        source_event_ts:   Utc
+            .with_ymd_and_hms(2026, 6, 21, 10, 0, 0)
+            .single()
+            .expect("test source timestamp should be valid"),
+        source_event_name: source_event_name.to_string(),
+        actor:             Some(actor()),
+    }
+}
+
+fn main_event(body: MainEventBody) -> MainEvent {
+    MainEvent {
+        id: "019796d3-d2d0-7c81-a65a-6f3ddc67d4a4".to_string(),
+        ts: Utc
+            .with_ymd_and_hms(2026, 6, 21, 10, 1, 0)
+            .single()
+            .expect("test main event timestamp should be valid"),
+        body,
+    }
+}
+
+#[test]
+fn v1_event_names_are_stable() {
+    let cases = [
+        (
+            MainEventBody::RunCreated(MainEventCreatedProps {
+                source:             source("run.created"),
+                title:              Some("Triage bug".to_string()),
+                workflow_slug:      Some("triage".to_string()),
+                automation:         None,
+                git:                None,
+                parent_id:          None,
+                provenance_subject: Some(actor()),
+            }),
+            "fabro.run.created",
+        ),
+        (
+            MainEventBody::RunStarted(MainEventStartedProps {
+                source:       source("run.started"),
+                name:         "triage".to_string(),
+                base_branch:  Some("main".to_string()),
+                base_sha:     Some("abc123".to_string()),
+                run_branch:   Some("fabro/run-1".to_string()),
+                worktree_dir: None,
+                goal:         Some("Triage bug".to_string()),
+            }),
+            "fabro.run.started",
+        ),
+        (
+            MainEventBody::RunRunning(MainEventLifecycleProps {
+                source: source("run.running"),
+            }),
+            "fabro.run.running",
+        ),
+        (
+            MainEventBody::RunCompleted(MainEventCompletedProps {
+                source:               source("run.completed"),
+                status:               "succeeded".to_string(),
+                reason:               SuccessReason::Completed,
+                total_usd_micros:     Some(12),
+                final_git_commit_sha: Some("abc123".to_string()),
+                automation:           None,
+                usage:                None,
+            }),
+            "fabro.run.completed",
+        ),
+        (
+            MainEventBody::RunFailed(MainEventFailedProps {
+                source:               source("run.failed"),
+                reason:               FailureReason::WorkflowError,
+                category:             FailureCategory::Deterministic,
+                message:              "boom".to_string(),
+                final_git_commit_sha: None,
+                automation:           None,
+                usage:                None,
+            }),
+            "fabro.run.failed",
+        ),
+        (
+            MainEventBody::RunCancelled(MainEventCancelledProps {
+                source:               source("run.failed"),
+                reason:               FailureReason::Cancelled,
+                category:             FailureCategory::Canceled,
+                message:              "cancelled".to_string(),
+                final_git_commit_sha: None,
+                automation:           None,
+                usage:                None,
+            }),
+            "fabro.run.cancelled",
+        ),
+        (
+            MainEventBody::RunPaused(MainEventLifecycleProps {
+                source: source("run.paused"),
+            }),
+            "fabro.run.paused",
+        ),
+        (
+            MainEventBody::RunUnpaused(MainEventLifecycleProps {
+                source: source("run.unpaused"),
+            }),
+            "fabro.run.unpaused",
+        ),
+    ];
+
+    for (body, expected) in cases {
+        assert_eq!(body.event_name(), expected);
+        let event = main_event(body);
+        assert_eq!(event.event_name(), expected);
+        let value = serde_json::to_value(&event).unwrap();
+        assert_eq!(serde_json::from_value::<MainEvent>(value).unwrap(), event);
+    }
+}
+
+#[test]
+fn malformed_known_main_event_is_rejected() {
+    let value = json!({
+        "id": "event-1",
+        "ts": "2026-06-21T10:01:00Z",
+        "event": "fabro.run.created",
+        "properties": {"run_id": 42}
+    });
+    assert!(serde_json::from_value::<MainEvent>(value).is_err());
+}
+
+#[test]
+fn main_event_round_trips_flat_event_properties() {
+    let event = main_event(MainEventBody::RunCompleted(MainEventCompletedProps {
+        source:               source("run.completed"),
+        status:               "succeeded".to_string(),
+        reason:               SuccessReason::Completed,
+        total_usd_micros:     Some(42),
+        final_git_commit_sha: Some("deadbeef".to_string()),
+        automation:           None,
+        usage:                None,
+    }));
+
+    let value = serde_json::to_value(&event).expect("main event should serialize");
+    assert_eq!(value["event"], "fabro.run.completed");
+    assert_eq!(value["properties"]["run_id"], run_id().to_string());
+    assert_eq!(value["properties"]["source_event_name"], "run.completed");
+    assert_eq!(value["properties"]["status"], "succeeded");
+
+    let parsed: MainEvent =
+        serde_json::from_value(value.clone()).expect("main event should deserialize");
+    assert_eq!(
+        serde_json::to_value(parsed).expect("main event should serialize after round trip"),
+        value
+    );
+}
+
+#[test]
+fn unknown_main_event_preserves_name_and_properties() {
+    let value = json!({
+        "id": "019796d3-d2d0-7c81-a65a-6f3ddc67d4a4",
+        "ts": "2026-06-21T10:01:00Z",
+        "event": "github.issue_comment.created",
+        "properties": {
+            "delivery_id": "delivery-1",
+            "action": "created"
+        }
+    });
+
+    let parsed: MainEvent =
+        serde_json::from_value(value.clone()).expect("unknown main event should deserialize");
+    match &parsed.body {
+        MainEventBody::Unknown { name, properties } => {
+            assert_eq!(name, "github.issue_comment.created");
+            assert_eq!(properties["delivery_id"], "delivery-1");
+        }
+        other => panic!("expected unknown event, got {other:?}"),
+    }
+    assert_eq!(
+        serde_json::to_value(parsed).expect("unknown main event should serialize after round trip"),
+        value
+    );
+}