Merge remote-tracking branch 'origin/main'

# Conflicts: # apps/fabro-web/app/routes/run-overview.tsx # apps/fabro-web/app/routes/workflow-detail.tsx # apps/fabro-web/app/routes/workflows.tsx # lib/crates/fabro-workflow/src/lifecycle/artifact.rs # lib/crates/fabro-workflow/src/pipeline/finalize.rs
2026-09-15 23:32:46 +00:00 · 2026-04-07 15:24:07 -04:00 · 2026-04-07 15:24:07 -04:00 · 9ca0113f66
commit 9ca0113f66
parent ba7dc77be0 4f02f5c42f
47 changed files with 1107 additions and 475 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -1526,6 +1526,7 @@ dependencies = [
 "async-trait",
 "axum",
 "base64",
+ "bytes",
 "chrono",
 "clap",
 "clap_complete",
@ -2068,6 +2069,7 @@ dependencies = [
 "assert_cmd",
 "async-trait",
 "base64",
+ "bytes",
 "chrono",
 "dirs",
 "fabro-agent",
--- a/apps/fabro-web/app/lib/workflow-api.ts
+++ b/apps/fabro-web/app/lib/workflow-api.ts
@ -0,0 +1,32 @@
+import type { PaginationMeta, RunSettings } from "@qltysh/fabro-api-client";
+
+export interface WorkflowScheduleSummary {
+  expression: string;
+  next_run?: string | null;
+}
+
+export interface WorkflowLastRunSummary {
+  ran_at?: string | null;
+}
+
+export interface WorkflowListItem {
+  name: string;
+  slug: string;
+  filename: string;
+  last_run?: WorkflowLastRunSummary | null;
+  schedule?: WorkflowScheduleSummary | null;
+}
+
+export interface PaginatedWorkflowListResponse {
+  data: WorkflowListItem[];
+  pagination?: PaginationMeta;
+}
+
+export interface WorkflowDetailResponse {
+  name: string;
+  slug: string;
+  description: string;
+  filename: string;
+  settings: RunSettings;
+  graph: string;
+}
--- a/apps/fabro-web/app/routes/run-overview.tsx
+++ b/apps/fabro-web/app/routes/run-overview.tsx
@ -8,10 +8,7 @@ import { getGraphTheme } from "../lib/graph-theme";
 import { apiJson } from "../api";
 import { formatDurationSecs } from "../lib/format";
 import type { PaginatedRunStageList, PaginatedRunList } from "@qltysh/fabro-api-client";
-
-interface WorkflowGraphResponse {
-  graph: string;
-}
+import type { WorkflowDetailResponse } from "../lib/workflow-api";

 export const handle = { wide: true };

@ -39,7 +36,7 @@ export async function loader({ request, params }: any) {
  let graphDot: string | null = null;
  if (run) {
    try {
-      const workflow = await apiJson<WorkflowGraphResponse>(`/workflows/${run.workflow}`, { request });
+      const workflow = await apiJson<WorkflowDetailResponse>(`/workflows/${run.workflow}`, { request });
      graphDot = workflow.graph;
    } catch {
      // workflow not found — leave graphDot null
--- a/apps/fabro-web/app/routes/workflow-detail.tsx
+++ b/apps/fabro-web/app/routes/workflow-detail.tsx
@ -2,15 +2,7 @@ import { ChevronRightIcon } from "@heroicons/react/20/solid";
 import { Link, Outlet, useLocation, useParams } from "react-router";
 import { apiJson } from "../api";
 import type { RunSettings } from "@qltysh/fabro-api-client";
-
-interface ApiWorkflowDetail {
-  name: string;
-  slug: string;
-  description: string;
-  filename: string;
-  settings: RunSettings;
-  graph: string;
-}
+import type { WorkflowDetailResponse as ApiWorkflowDetail } from "../lib/workflow-api";

 export interface WorkflowEntry {
  name: string;
--- a/apps/fabro-web/app/routes/workflows.tsx
+++ b/apps/fabro-web/app/routes/workflows.tsx
@ -15,27 +15,7 @@ import {
 import { Link } from "react-router";
 import { apiJson } from "../api";
 import { timeAgo, timeUntil } from "../lib/time";
-
-interface WorkflowRunSummary {
-  ran_at?: string | null;
-}
-
-interface WorkflowScheduleSummary {
-  expression: string;
-  next_run?: string | null;
-}
-
-interface WorkflowListItem {
-  name: string;
-  slug: string;
-  filename: string;
-  last_run?: WorkflowRunSummary | null;
-  schedule?: WorkflowScheduleSummary | null;
-}
-
-interface PaginatedWorkflowList {
-  data: WorkflowListItem[];
-}
+import type { PaginatedWorkflowListResponse } from "../lib/workflow-api";

 export function meta({}: any) {
  return [{ title: "Workflows — Fabro" }];
@ -125,7 +105,7 @@ interface WorkflowData {
 }

 export async function loader({ request }: any) {
-  const { data: apiWorkflows } = await apiJson<PaginatedWorkflowList>("/workflows", { request });
+  const { data: apiWorkflows } = await apiJson<PaginatedWorkflowListResponse>("/workflows", { request });
  const workflows: WorkflowData[] = apiWorkflows.map((w) => ({
    name: w.name,
    slug: w.slug,
--- a/docs/api-reference/fabro-api.yaml
+++ b/docs/api-reference/fabro-api.yaml
@ -529,7 +529,7 @@ paths:
      operationId: attachRunEvents
      tags: [Run Internals]
      summary: Attach Run Events
-      description: Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
+      description: Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
      parameters:
        - $ref: "#/components/parameters/RunId"
        - $ref: "#/components/parameters/SinceSeq"
@ -546,12 +546,6 @@ paths:
            application/json:
              schema:
                $ref: "#/components/schemas/ErrorResponse"
-        "410":
-          description: Run is not live on this server
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/ErrorResponse"

  /api/v1/runs/{id}/blobs:
    post:
--- a/lib/crates/fabro-cli/Cargo.toml
+++ b/lib/crates/fabro-cli/Cargo.toml
@ -82,6 +82,7 @@ sha2.workspace = true
 shlex = "1"
 walkdir.workspace = true
 object_store.workspace = true
+bytes.workspace = true

 [target.'cfg(target_os = "macos")'.dependencies]
 core-foundation = { version = "0.9", optional = true }
--- a/lib/crates/fabro-cli/src/commands/pr/create.rs
+++ b/lib/crates/fabro-cli/src/commands/pr/create.rs
@ -103,7 +103,7 @@ pub(super) async fn create_command(
        &model,
        true,
        None,
-        &run_store,
+        &run_store.clone().into(),
        None,
    )
    .await
--- a/lib/crates/fabro-cli/src/commands/run/attach.rs
+++ b/lib/crates/fabro-cli/src/commands/run/attach.rs
@ -4,7 +4,7 @@ use std::path::Path;
 #[cfg(test)]
 use std::path::PathBuf;
 use std::process::ExitCode;
-use std::time::{Duration, Instant};
+use std::time::Duration;

 use anyhow::Result;
 use fabro_types::{EventBody, RunEvent, RunId};
@ -17,7 +17,7 @@ use fabro_util::terminal::Styles;
 use fabro_workflow::outcome::StageStatus;
 use fabro_workflow::run_status::RunStatus;
 use tokio::signal::ctrl_c;
-use tokio::time::{sleep, timeout};
+use tokio::time::sleep;

 use super::run_progress;
 use crate::server_client;
@ -25,10 +25,7 @@ use crate::server_client;
 const INTERVIEW_UNANSWERED_MESSAGE: &str =
    "Interview ended without an answer. The run is still waiting for input; reattach to answer it.";
 const JSON_INTERVIEW_MESSAGE: &str = "This run is waiting for human input, but --json is non-interactive. Reattach without --json to answer it.";
-#[cfg(test)]
-const ATTACH_FINAL_STATUS_GRACE: Duration = Duration::from_millis(250);
-#[cfg(not(test))]
-const ATTACH_FINAL_STATUS_GRACE: Duration = Duration::from_secs(2);
+const ATTACH_PREMATURE_EOF_MESSAGE: &str = "Attach stream ended before terminal run event.";

 /// Attach to a running (or finished) workflow run, rendering progress live.
 ///
@ -73,57 +70,51 @@ pub(crate) async fn attach_run_with_client(
    let replay_events = events.clone();
    let next_seq = events.last().map_or(1, |event| event.seq.saturating_add(1));
    let initial_exit_code = events.iter().rev().find_map(event_exit_code);
+    let state_exit_code = state_exit_code(&state);

    if state_is_terminal(&state) || initial_exit_code.is_some() {
-        return replay_run_with_client(client, run_id, verbose, events, json_output).await;
+        return replay_run_with_client(
+            verbose,
+            events,
+            initial_exit_code
+                .or(state_exit_code)
+                .unwrap_or(ExitCode::from(1)),
+            json_output,
+        )
+        .await;
    }

-    match client.attach_run_events(run_id, Some(next_seq)).await {
-        Ok(stream) => {
-            attach_live_run_with_client(
-                client,
-                run_id,
-                verbose,
-                events,
-                stream,
-                kill_on_detach,
-                styles,
-                json_output,
-            )
-            .await
-        }
-        Err(server_client::RunAttachStreamError::Gone) => {
-            replay_run_with_client(client, run_id, verbose, replay_events, json_output).await
-        }
-        Err(server_client::RunAttachStreamError::Other(err)) => Err(err),
-    }
+    let stream = client.attach_run_events(run_id, Some(next_seq)).await?;
+    attach_live_run_with_client(
+        client,
+        run_id,
+        verbose,
+        replay_events,
+        stream,
+        kill_on_detach,
+        styles,
+        json_output,
+    )
+    .await
 }

 async fn replay_run_with_client(
-    client: &server_client::ServerStoreClient,
-    run_id: &RunId,
    verbose: bool,
    events: Vec<EventEnvelope>,
+    exit_code: ExitCode,
    json_output: bool,
 ) -> Result<ExitCode> {
    let is_tty = std::io::stderr().is_terminal();
    let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose);
-    let mut terminal_exit_code = None;

    for event in events {
-        if let Some(exit_code) = event_exit_code(&event) {
-            terminal_exit_code = Some(exit_code);
-        }
        let line = event_payload_line(&event)?;
        emit_progress_line(&mut progress_ui, &line, json_output)?;
    }

    finish_progress(&mut progress_ui, json_output);

-    Ok(match terminal_exit_code {
-        Some(exit_code) => exit_code,
-        None => determine_exit_code_with_server(client, run_id).await,
-    })
+    Ok(exit_code)
 }

 async fn attach_live_run_with_client(
@ -141,16 +132,7 @@ async fn attach_live_run_with_client(
    let ctrl_c_signal = ctrl_c();
    tokio::pin!(ctrl_c_signal);

-    let mut next_seq = 1;
-    let mut terminal_exit_code = None;
-    let mut terminal_event_seen_at: Option<Instant> = None;
-
    for event in existing_events {
-        next_seq = event.seq.saturating_add(1);
-        if let Some(exit_code) = event_exit_code(&event) {
-            terminal_exit_code = Some(exit_code);
-            terminal_event_seen_at = Some(Instant::now());
-        }
        let line = event_payload_line(&event)?;
        emit_progress_line(&mut progress_ui, &line, json_output)?;
    }
@ -163,46 +145,28 @@ async fn attach_live_run_with_client(
    }

    loop {
-        let next_event = if let Some(seen_at) = terminal_event_seen_at {
-            let remaining = ATTACH_FINAL_STATUS_GRACE.saturating_sub(seen_at.elapsed());
-            if remaining.is_zero() {
-                break;
-            }
-            tokio::select! {
-                _ = &mut ctrl_c_signal => {
-                    handle_detach_signal(client, run_id, kill_on_detach).await;
-                    break;
-                }
-                result = timeout(remaining, stream.next_event()) => {
-                    match result {
-                        Ok(result) => result?,
-                        Err(_) => break,
-                    }
-                }
-            }
-        } else {
-            tokio::select! {
-                _ = &mut ctrl_c_signal => {
-                    handle_detach_signal(client, run_id, kill_on_detach).await;
-                    break;
-                }
-                result = stream.next_event() => result?,
+        let next_event = tokio::select! {
+            _ = &mut ctrl_c_signal => {
+                handle_detach_signal(client, run_id, kill_on_detach).await;
+                finish_progress(&mut progress_ui, json_output);
+                return Ok(ExitCode::from(1));
            }
+            result = stream.next_event() => result?,
        };

        let Some(event) = next_event else {
-            break;
+            finish_progress(&mut progress_ui, json_output);
+            return Err(anyhow::anyhow!(ATTACH_PREMATURE_EOF_MESSAGE));
        };

-        next_seq = event.seq.saturating_add(1);
-        if let Some(exit_code) = event_exit_code(&event) {
-            terminal_exit_code = Some(exit_code);
-            terminal_event_seen_at = Some(Instant::now());
-        }
-
        let line = event_payload_line(&event)?;
        emit_progress_line(&mut progress_ui, &line, json_output)?;

+        if let Some(exit_code) = event_exit_code(&event) {
+            finish_progress(&mut progress_ui, json_output);
+            return Ok(exit_code);
+        }
+
        if event_starts_interview(&event) {
            if let Some(exit_code) = handle_pending_server_interview(
                client,
@ -217,20 +181,6 @@ async fn attach_live_run_with_client(
            }
        }
    }
-
-    if terminal_exit_code.is_none() {
-        let (_, trailing_exit_code) =
-            emit_server_events_from(client, run_id, next_seq, &mut progress_ui, json_output)
-                .await?;
-        terminal_exit_code = trailing_exit_code;
-    }
-
-    finish_progress(&mut progress_ui, json_output);
-
-    Ok(match terminal_exit_code {
-        Some(exit_code) => exit_code,
-        None => determine_exit_code_with_server(client, run_id).await,
-    })
 }

 async fn handle_pending_server_interview(
@ -287,33 +237,6 @@ async fn handle_detach_signal(
    }
 }

-async fn emit_server_events_from(
-    client: &server_client::ServerStoreClient,
-    run_id: &RunId,
-    next_seq: u32,
-    progress_ui: &mut run_progress::ProgressUI,
-    json_output: bool,
-) -> Result<(u32, Option<ExitCode>)> {
-    let events = match client.list_run_events(run_id, Some(next_seq), None).await {
-        Ok(events) => events,
-        Err(err) if is_run_not_found_error(&err) => Vec::new(),
-        Err(err) => return Err(err),
-    };
-
-    let mut current_seq = next_seq;
-    let mut terminal_exit_code = None;
-    for event in events {
-        if let Some(exit_code) = event_exit_code(&event) {
-            terminal_exit_code = Some(exit_code);
-        }
-        let line = event_payload_line(&event)?;
-        emit_progress_line(progress_ui, &line, json_output)?;
-        current_seq = event.seq.saturating_add(1);
-    }
-
-    Ok((current_seq, terminal_exit_code))
-}
-
 fn api_question_to_question(question: &types::ApiQuestion) -> Question {
    let question_type = match question.question_type {
        types::QuestionType::YesNo => QuestionType::YesNo,
@ -363,11 +286,6 @@ async fn submit_server_interview_answer(
    Ok(true)
 }

-fn is_run_not_found_error(err: &anyhow::Error) -> bool {
-    err.chain()
-        .any(|cause| cause.to_string() == "Run not found.")
-}
-
 fn state_is_terminal(state: &server_client::RunProjection) -> bool {
    state.conclusion.is_some()
        || state
@ -455,38 +373,23 @@ fn answer_requires_reattach(answer: &fabro_interview::Answer) -> bool {
    matches!(answer.value, AnswerValue::Aborted | AnswerValue::Skipped)
 }

-async fn determine_exit_code_with_server(
-    client: &server_client::ServerStoreClient,
-    run_id: &RunId,
-) -> ExitCode {
-    let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE;
-    loop {
-        if let Ok(state) = client.get_run_state(run_id).await {
-            if let Some(conclusion) = state.conclusion {
-                let success = matches!(
-                    conclusion.status,
-                    StageStatus::Success | StageStatus::PartialSuccess
-                );
-                return if success {
-                    ExitCode::from(0)
-                } else {
-                    ExitCode::from(1)
-                };
-            }
+fn state_exit_code(state: &server_client::RunProjection) -> Option<ExitCode> {
+    if let Some(conclusion) = &state.conclusion {
+        let success = matches!(
+            conclusion.status,
+            StageStatus::Success | StageStatus::PartialSuccess
+        );
+        return Some(if success {
+            ExitCode::from(0)
+        } else {
+            ExitCode::from(1)
+        });
+    }

-            match state.status {
-                Some(record) if matches!(record.status, RunStatus::Succeeded) => {
-                    return ExitCode::from(0);
-                }
-                Some(record) if record.status.is_terminal() => return ExitCode::from(1),
-                Some(_) | None => {}
-            }
-        }
-
-        if Instant::now() >= deadline {
-            return ExitCode::from(1);
-        }
-        sleep(Duration::from_millis(100)).await;
+    match state.status.as_ref() {
+        Some(record) if record.status == RunStatus::Succeeded => Some(ExitCode::from(0)),
+        Some(record) if record.status.is_terminal() => Some(ExitCode::from(1)),
+        Some(_) | None => None,
    }
 }

--- a/lib/crates/fabro-cli/src/commands/run/runner.rs
+++ b/lib/crates/fabro-cli/src/commands/run/runner.rs
@ -4,21 +4,28 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;

 use anyhow::{Context, Result, anyhow};
+use async_trait::async_trait;
 use fabro_config::RunScratch;
 use fabro_interview::FileInterviewer;
-use fabro_store::{Database, EventPayload, RunDatabase};
-use fabro_types::{EventBody, RunEvent, RunId, Settings, StatusReason};
+use fabro_store::{EventEnvelope, EventPayload, RunProjection};
+use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, Settings, StatusReason};
 use fabro_workflow::event::{Emitter, RunEventSink};
 use fabro_workflow::run_control::RunControlState;
-use object_store::memory::InMemory as MemoryObjectStore;
+use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle};
 #[cfg(unix)]
 use tokio::signal::unix::{SignalKind, signal};
+use tokio::sync::Mutex;
+use tokio::time::sleep;

 use crate::args::RunWorkerMode;
 use crate::server_client;
 use crate::shared::github::build_github_app_credentials;

-const STORE_FLUSH_INTERVAL: Duration = Duration::from_millis(100);
+const RUN_STORE_RETRY_DELAYS: [Duration; 3] = [
+    Duration::from_millis(50),
+    Duration::from_millis(100),
+    Duration::from_millis(250),
+];

 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum WorkerTitlePhase {
@ -43,7 +50,7 @@ pub(crate) async fn execute(
    set_worker_title(&run_id, initial_worker_title_phase(mode));

    let client = server_client::connect_server_target_direct(&server).await?;
-    let run_store = load_seed_run_store(&client, &run_id).await?;
+    let run_store = HttpRunStore::connect(run_id, client.clone_for_reuse()).await?;
    let run_state = run_store
        .state()
        .await
@ -62,7 +69,6 @@ pub(crate) async fn execute(
    let cancel_token = Arc::new(AtomicBool::new(false));
    install_signal_handlers(Arc::clone(&run_control), Arc::clone(&cancel_token))?;
    let github_app = maybe_build_github_app_credentials(&run_record.settings)?;
-    let event_client = client.clone_for_reuse();
    let services = fabro_workflow::operations::StartServices {
        run_id,
        cancel_token: Some(Arc::clone(&cancel_token)),
@ -70,11 +76,10 @@ pub(crate) async fn execute(
        interviewer,
        run_store: run_store.clone(),
        event_sink: RunEventSink::fanout(vec![
-            RunEventSink::store(run_store),
+            RunEventSink::backend(run_store),
            RunEventSink::callback(move |event| {
                update_worker_title_from_event(&event);
-                let client = event_client.clone_for_reuse();
-                async move { client.append_run_event(&event.run_id, &event).await }
+                async move { Ok(()) }
            }),
        ]),
        run_control: Some(run_control),
@ -95,42 +100,139 @@ pub(crate) async fn execute(
    Ok(())
 }

-fn open_memory_store() -> Arc<Database> {
-    Arc::new(Database::new(
-        Arc::new(MemoryObjectStore::new()),
-        "",
-        STORE_FLUSH_INTERVAL,
-    ))
+#[derive(Clone)]
+struct HttpRunStore {
+    run_id: RunId,
+    client: server_client::ServerStoreClient,
+    state: Arc<Mutex<RunProjection>>,
+    events: Arc<Mutex<Option<Vec<EventEnvelope>>>>,
 }

-async fn load_seed_run_store(
-    client: &server_client::ServerStoreClient,
-    run_id: &RunId,
-) -> Result<RunDatabase> {
-    let events = client
-        .list_run_events(run_id, None, None)
-        .await
-        .with_context(|| format!("failed to fetch run events for {run_id}"))?;
-    let payloads = events
-        .into_iter()
-        .map(|event| event.payload)
-        .collect::<Vec<_>>();
-    seed_run_store(run_id, &payloads).await
-}
-
-async fn seed_run_store(run_id: &RunId, events: &[EventPayload]) -> Result<RunDatabase> {
-    let store = open_memory_store();
-    let run_store = store
-        .create_run(run_id)
-        .await
-        .with_context(|| format!("failed to create in-memory run store for {run_id}"))?;
-    for payload in events {
-        run_store
-            .append_event(payload)
+impl HttpRunStore {
+    async fn connect(
+        run_id: RunId,
+        client: server_client::ServerStoreClient,
+    ) -> Result<RunStoreHandle> {
+        let state = client
+            .get_run_state(&run_id)
            .await
-            .with_context(|| format!("failed to seed in-memory run store for {run_id}"))?;
+            .with_context(|| format!("failed to fetch run state for {run_id}"))?;
+        Ok(RunStoreHandle::new(Arc::new(Self {
+            run_id,
+            client,
+            state: Arc::new(Mutex::new(state)),
+            events: Arc::new(Mutex::new(None)),
+        })))
+    }
+
+    async fn with_retries<T, F, Fut>(&self, operation: &'static str, mut op: F) -> Result<T>
+    where
+        F: FnMut() -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        let mut last_error = None;
+        for attempt in 0..=RUN_STORE_RETRY_DELAYS.len() {
+            match op().await {
+                Ok(value) => return Ok(value),
+                Err(err) => last_error = Some(err),
+            }
+            if let Some(delay) = RUN_STORE_RETRY_DELAYS.get(attempt) {
+                sleep(*delay).await;
+            }
+        }
+        Err(last_error
+            .unwrap_or_else(|| anyhow!("run store operation failed"))
+            .context(format!(
+                "worker lost canonical run store during {operation}"
+            )))
+    }
+
+    async fn refresh_state_from_server(&self) -> Result<RunProjection> {
+        self.with_retries("refresh state", || {
+            let client = self.client.clone_for_reuse();
+            let run_id = self.run_id;
+            async move { client.get_run_state(&run_id).await }
+        })
+        .await
+    }
+
+    async fn apply_acknowledged_event(&self, seq: u32, event: &RunEvent) -> Result<()> {
+        let payload = EventPayload::new(event.to_value()?, &self.run_id)?;
+        let envelope = EventEnvelope { seq, payload };
+
+        {
+            let mut state = self.state.lock().await;
+            if let Err(err) = state.apply_event(&envelope) {
+                tracing::warn!(run_id = %self.run_id, error = %err, "failed to apply acknowledged event to local run-state mirror; refreshing from server");
+                drop(state);
+                let refreshed = self.refresh_state_from_server().await?;
+                *self.state.lock().await = refreshed;
+            }
+        }
+
+        let mut events = self.events.lock().await;
+        if let Some(cached) = events.as_mut() {
+            cached.push(envelope);
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl RunStoreBackend for HttpRunStore {
+    async fn load_state(&self) -> Result<RunProjection> {
+        Ok(self.state.lock().await.clone())
+    }
+
+    async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
+        let mut cached = self.events.lock().await;
+        if let Some(events) = cached.as_ref() {
+            return Ok(events.clone());
+        }
+
+        let events = self
+            .with_retries("list run events", || {
+                let client = self.client.clone_for_reuse();
+                let run_id = self.run_id;
+                async move { client.list_run_events(&run_id, None, None).await }
+            })
+            .await?;
+        *cached = Some(events.clone());
+        Ok(events)
+    }
+
+    async fn append_run_event(&self, event: &RunEvent) -> Result<()> {
+        let seq = self
+            .with_retries("append run event", || {
+                let client = self.client.clone_for_reuse();
+                let run_id = self.run_id;
+                let event = event.clone();
+                async move { client.append_run_event(&run_id, &event).await }
+            })
+            .await?;
+        self.apply_acknowledged_event(seq, event).await
+    }
+
+    async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
+        self.with_retries("write run blob", || {
+            let client = self.client.clone_for_reuse();
+            let run_id = self.run_id;
+            let data = data.to_vec();
+            async move { client.write_run_blob(&run_id, &data).await }
+        })
+        .await
+    }
+
+    async fn read_blob(&self, id: &RunBlobId) -> Result<Option<bytes::Bytes>> {
+        self.with_retries("read run blob", || {
+            let client = self.client.clone_for_reuse();
+            let run_id = self.run_id;
+            let blob_id = *id;
+            async move { client.read_run_blob(&run_id, &blob_id).await }
+        })
+        .await
    }
-    Ok(run_store)
 }

 fn set_worker_title(run_id: &RunId, phase: WorkerTitlePhase) {
@ -247,8 +349,12 @@ fn install_signal_handlers(

 #[cfg(test)]
 mod tests {
+    use httpmock::MockServer;
+    use serde_json::json;
+
    use super::{
-        WorkerTitlePhase, initial_worker_title_phase, worker_title, worker_title_phase_for_event,
+        WorkerTitlePhase, execute, initial_worker_title_phase, worker_title,
+        worker_title_phase_for_event,
    };
    use crate::args::RunWorkerMode;
    use fabro_types::fixtures;
@ -342,4 +448,61 @@ mod tests {
            Some(WorkerTitlePhase::Failed)
        );
    }
+
+    #[tokio::test]
+    async fn worker_bootstrap_loads_run_state_without_prefetching_run_events() {
+        let server = MockServer::start_async().await;
+        let run_id = fixtures::RUN_1;
+
+        let state_mock = server
+            .mock_async(|when, then| {
+                when.method("GET")
+                    .path(format!("/api/v1/runs/{run_id}/state"));
+                then.status(200)
+                    .header("Content-Type", "application/json")
+                    .body(
+                        json!({
+                            "run": null,
+                            "graph_source": null,
+                            "start": null,
+                            "status": null,
+                            "checkpoint": null,
+                            "checkpoints": [],
+                            "conclusion": null,
+                            "retro": null,
+                            "retro_prompt": null,
+                            "retro_response": null,
+                            "sandbox": null,
+                            "final_patch": null,
+                            "pull_request": null,
+                            "nodes": {}
+                        })
+                        .to_string(),
+                    );
+            })
+            .await;
+        let events_mock = server
+            .mock_async(|when, then| {
+                when.method("GET")
+                    .path(format!("/api/v1/runs/{run_id}/events"));
+                then.status(200)
+                    .header("Content-Type", "application/json")
+                    .body(json!({ "data": [], "meta": { "has_more": false } }).to_string());
+            })
+            .await;
+
+        let run_dir = tempfile::tempdir().unwrap();
+        let error = execute(
+            run_id,
+            format!("{}/api/v1", server.base_url()),
+            run_dir.path().to_path_buf(),
+            RunWorkerMode::Start,
+        )
+        .await
+        .unwrap_err();
+
+        assert!(error.to_string().contains("has no run record"));
+        state_mock.assert_async().await;
+        assert_eq!(events_mock.calls_async().await, 0);
+    }
 }
--- a/lib/crates/fabro-cli/src/server_client.rs
+++ b/lib/crates/fabro-cli/src/server_client.rs
@ -4,10 +4,11 @@ use std::path::{Path, PathBuf};
 use std::time::Duration;

 use anyhow::{Context as _, Result, anyhow, bail};
+use bytes::Bytes;
 use fabro_api::types;
 use fabro_server::bind::Bind;
 use fabro_store::{EventEnvelope, RunSummary, StageId};
-use fabro_types::{RunEvent, RunId, Settings};
+use fabro_types::{RunBlobId, RunEvent, RunId, Settings};
 use futures::StreamExt;
 use serde::de::DeserializeOwned;
 use tokio::time::sleep;
@ -34,11 +35,6 @@ pub(crate) struct RunAttachEventStream {
    buffered_events: VecDeque<EventEnvelope>,
 }

-pub(crate) enum RunAttachStreamError {
-    Gone,
-    Other(anyhow::Error),
-}
-
 impl RunAttachEventStream {
    fn new(stream: progenitor_client::ByteStream) -> Self {
        Self {
@ -355,12 +351,12 @@ impl ServerStoreClient {
        &self,
        run_id: &RunId,
        since_seq: Option<u32>,
-    ) -> std::result::Result<RunAttachEventStream, RunAttachStreamError> {
+    ) -> Result<RunAttachEventStream> {
        let mut request = self.client.attach_run_events().id(run_id.to_string());
        if let Some(seq) = since_seq.and_then(non_zero_u64_from_u32) {
            request = request.since_seq(seq);
        }
-        let response = request.send().await.map_err(map_attach_run_stream_error)?;
+        let response = request.send().await.map_err(map_api_error)?;
        Ok(RunAttachEventStream::new(response.into_inner()))
    }

@ -403,16 +399,65 @@ impl ServerStoreClient {
        Ok(())
    }

-    pub(crate) async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result<()> {
+    pub(crate) async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result<u32> {
        let body: types::RunEvent = convert_type(event)?;
-        self.client
+        let response = self
+            .client
            .append_run_event()
            .id(run_id.to_string())
            .body(body)
            .send()
            .await
            .map_err(map_api_error)?;
-        Ok(())
+        u32::try_from(response.into_inner().seq).context("append_run_event returned invalid seq")
+    }
+
+    pub(crate) async fn write_run_blob(&self, run_id: &RunId, data: &[u8]) -> Result<RunBlobId> {
+        let response = self
+            .client
+            .write_run_blob()
+            .id(run_id.to_string())
+            .body(data.to_vec())
+            .send()
+            .await
+            .map_err(map_api_error)?;
+        response
+            .into_inner()
+            .id
+            .parse()
+            .context("write_run_blob returned invalid blob id")
+    }
+
+    pub(crate) async fn read_run_blob(
+        &self,
+        run_id: &RunId,
+        blob_id: &RunBlobId,
+    ) -> Result<Option<Bytes>> {
+        let response = self
+            .client
+            .read_run_blob()
+            .id(run_id.to_string())
+            .blob_id(blob_id.to_string())
+            .send()
+            .await;
+        match response {
+            Ok(response) => {
+                let mut stream = response.into_inner();
+                let mut bytes = Vec::new();
+                while let Some(chunk) = stream.next().await {
+                    let chunk = chunk.map_err(|err| anyhow!("{err}"))?;
+                    bytes.extend_from_slice(&chunk);
+                }
+                Ok(Some(Bytes::from(bytes)))
+            }
+            Err(err) => {
+                if is_not_found_error(&err) {
+                    Ok(None)
+                } else {
+                    Err(map_api_error(err))
+                }
+            }
+        }
    }

    pub(crate) async fn delete_store_run(&self, run_id: &RunId) -> Result<()> {
@ -584,19 +629,20 @@ where
    }
 }

-fn map_attach_run_stream_error(
-    err: progenitor_client::Error<types::ErrorResponse>,
-) -> RunAttachStreamError {
-    match &err {
-        progenitor_client::Error::ErrorResponse(response)
-            if response.status() == reqwest::StatusCode::GONE =>
-        {
-            RunAttachStreamError::Gone
+fn is_not_found_error<E>(err: &progenitor_client::Error<E>) -> bool
+where
+    E: serde::Serialize + std::fmt::Debug,
+{
+    match err {
+        progenitor_client::Error::ErrorResponse(response) => {
+            response.status() == reqwest::StatusCode::NOT_FOUND
        }
-        _ => RunAttachStreamError::Other(map_api_error(err)),
+        progenitor_client::Error::UnexpectedResponse(response) => {
+            response.status() == reqwest::StatusCode::NOT_FOUND
+        }
+        _ => false,
    }
 }
-
 fn convert_type<TInput, TOutput>(value: TInput) -> Result<TOutput>
 where
    TInput: serde::Serialize,
--- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs
+++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs
@ -196,6 +196,103 @@ fn attach_uses_configured_server_target_without_server_flag() {
    assert!(stdout.contains("\"event\":\"run.completed\""), "{stdout}");
 }

+#[test]
+fn attach_errors_when_live_stream_ends_before_terminal_event() {
+    let context = test_context!();
+    let server = MockServer::start();
+    let run_id = unique_run_id();
+
+    server.mock(|when, then| {
+        when.method("GET").path("/api/v1/runs");
+        then.status(200)
+            .header("Content-Type", "application/json")
+            .body(
+                serde_json::json!([
+                    {
+                        "run_id": run_id,
+                        "workflow_name": "Remote Workflow",
+                        "workflow_slug": "remote-workflow",
+                        "goal": "Remote output",
+                        "labels": {},
+                        "host_repo_path": null,
+                        "start_time": "2026-04-05T12:00:00Z",
+                        "status": "running",
+                        "status_reason": null,
+                        "duration_ms": 12,
+                        "total_cost": null
+                    }
+                ])
+                .to_string(),
+            );
+    });
+    server.mock(|when, then| {
+        when.method("GET")
+            .path(format!("/api/v1/runs/{run_id}/events"));
+        then.status(200)
+            .header("Content-Type", "application/json")
+            .body(
+                serde_json::json!({
+                    "data": [{
+                        "seq": 1,
+                        "payload": {
+                            "event": "run.running",
+                            "id": "evt-run-running",
+                            "run_id": run_id,
+                            "ts": "2026-04-05T12:00:00Z",
+                            "properties": {}
+                        }
+                    }],
+                    "meta": { "has_more": false }
+                })
+                .to_string(),
+            );
+    });
+    server.mock(|when, then| {
+        when.method("GET")
+            .path(format!("/api/v1/runs/{run_id}/state"));
+        then.status(200)
+            .header("Content-Type", "application/json")
+            .body(live_run_state_response().to_string());
+    });
+    server.mock(|when, then| {
+        when.method("GET")
+            .path(format!("/api/v1/runs/{run_id}/questions"))
+            .query_param("page[limit]", "100")
+            .query_param("page[offset]", "0");
+        then.status(200)
+            .header("Content-Type", "application/json")
+            .body(r#"{"data":[],"meta":{"has_more":false}}"#);
+    });
+    server.mock(|when, then| {
+        when.method("GET")
+            .path(format!("/api/v1/runs/{run_id}/attach"))
+            .query_param("since_seq", "2");
+        then.status(200)
+            .header("Content-Type", "text/event-stream")
+            .body("");
+    });
+    context.write_home(
+        ".fabro/settings.toml",
+        format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()),
+    );
+
+    let output = context
+        .command()
+        .args(["attach", &run_id])
+        .output()
+        .expect("attach should execute");
+
+    assert!(
+        !output.status.success(),
+        "attach should fail on premature EOF"
+    );
+    let stderr = String::from_utf8(output.stderr).expect("stderr should be UTF-8");
+    assert!(
+        stderr.contains("terminal run event"),
+        "expected a protocol error, got:\n{stderr}"
+    );
+}
+
 #[test]
 fn attach_replays_completed_detached_run() {
    let context = test_context!();
--- a/lib/crates/fabro-cli/tests/it/cmd/run.rs
+++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs
@ -4,8 +4,8 @@ use httpmock::MockServer;
 use serde_json::Value;

 use super::support::{
-    output_stderr, resolve_run, run_state, wait_for_no_process_match, wait_for_status,
-    write_gated_workflow,
+    output_stderr, resolve_run, run_state, wait_for_event_names, wait_for_no_process_match,
+    wait_for_status, write_gated_workflow,
 };
 use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id};

@ -477,7 +477,8 @@ fn dry_run_persists_event_history_in_store() {
        .assert()
        .success();

-    context.find_run_dir(&run_id);
+    let run_dir = context.find_run_dir(&run_id);
+    wait_for_event_names(&run_dir, &["run.completed", "sandbox.cleanup.completed"]);
    let output = context
        .command()
        .args(["logs", &run_id])
@ -510,6 +511,12 @@ fn dry_run_persists_event_history_in_store() {
            .and_then(Value::as_bool),
        Some(true)
    );
+    assert!(
+        progress
+            .iter()
+            .any(|event| event["event"].as_str() == Some("run.completed")),
+        "store-backed event history should include run.completed"
+    );
    assert_eq!(
        progress.last().and_then(|event| event["event"].as_str()),
        Some("sandbox.cleanup.completed")
@ -1275,25 +1282,6 @@ fn json_run_implies_auto_approve_for_human_gates() {
        },
        "run_id": "[ULID]",
        "ts": "[TIMESTAMP]"
-      },
-      {
-        "event": "sandbox.cleanup.started",
-        "id": "[EVENT_ID]",
-        "properties": {
-          "provider": "local"
-        },
-        "run_id": "[ULID]",
-        "ts": "[TIMESTAMP]"
-      },
-      {
-        "event": "sandbox.cleanup.completed",
-        "id": "[EVENT_ID]",
-        "properties": {
-          "duration_ms": "[DURATION_MS]",
-          "provider": "local"
-        },
-        "run_id": "[ULID]",
-        "ts": "[TIMESTAMP]"
      }
    ]
    "#);
--- a/lib/crates/fabro-cli/tests/it/cmd/support.rs
+++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs
@ -143,10 +143,15 @@ fn run_completed_dry_run(context: &TestContext, workflow: &Path) -> RunSetup {
            stderr(&output)
        );
    }
-    RunSetup {
+    let run_setup = RunSetup {
        run_dir: context.find_run_dir(&run_id),
        run_id,
-    }
+    };
+    wait_for_event_names(
+        &run_setup.run_dir,
+        &["run.completed", "sandbox.cleanup.completed"],
+    );
+    run_setup
 }

 pub(crate) fn setup_created_dry_run(context: &TestContext) -> RunSetup {
@ -691,6 +696,37 @@ pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
    serde_json::from_value(response["data"].clone()).expect("event list should parse")
 }

+pub(crate) fn wait_for_event_names(run_dir: &Path, expected: &[&str]) {
+    let deadline = std::time::Instant::now() + COMMAND_TIMEOUT;
+
+    loop {
+        let event_names = run_events(run_dir)
+            .into_iter()
+            .filter_map(|event| {
+                event
+                    .payload
+                    .as_value()
+                    .get("event")
+                    .and_then(serde_json::Value::as_str)
+                    .map(ToString::to_string)
+            })
+            .collect::<Vec<_>>();
+
+        if expected
+            .iter()
+            .all(|expected_name| event_names.iter().any(|name| name == expected_name))
+        {
+            return;
+        }
+
+        assert!(
+            std::time::Instant::now() < deadline,
+            "timed out waiting for events {expected:?}; saw {event_names:?}"
+        );
+        std::thread::sleep(std::time::Duration::from_millis(50));
+    }
+}
+
 pub(crate) fn git_stdout(repo_dir: &Path, args: &[&str]) -> String {
    stdout(&git_success(repo_dir, args))
 }
--- a/lib/crates/fabro-cli/tests/it/scenario/recovery.rs
+++ b/lib/crates/fabro-cli/tests/it/scenario/recovery.rs
@ -5,6 +5,7 @@ use fabro_checkpoint::branch::BranchStore;
 use fabro_checkpoint::git::Store as GitStore;
 use fabro_test::{fabro_snapshot, test_context};
 use fabro_types::Checkpoint;
+use fabro_workflow::operations::build_timeline;
 use git2::{Repository, Signature};

 use crate::support::unique_run_id;
@ -56,6 +57,17 @@ fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint {
    serde_json::from_slice(&store.read_blob_at(tip, "checkpoint.json").unwrap().unwrap()).unwrap()
 }

+fn timeline_run_shas(repo_dir: &Path, run_id: &str) -> Vec<Option<String>> {
+    let repo = Repository::discover(repo_dir).unwrap();
+    let store = GitStore::new(repo);
+    build_timeline(&store, run_id)
+        .unwrap()
+        .entries
+        .into_iter()
+        .map(|entry| entry.run_commit_sha)
+        .collect()
+}
+
 fn init_repo_with_workflow(repo_dir: &Path) {
    std::fs::write(repo_dir.join("README.md"), "recovery test\n").unwrap();
    std::fs::write(
@ -151,10 +163,9 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
    exit_code: 0
    ----- stdout -----
    ----- stderr -----
-    @   Node   Details         
-     @1  start  (no run commit) 
-     @2  plan                   
-     @3  build
+    @   Node   Details 
+     @1  plan           
+     @2  build
    ");

    let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), &source_run_id);
@ -165,12 +176,9 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
        None
    );
    assert!(rebuilt_checkpoints.len() >= 2);
-    let plan_sha = rebuilt_checkpoints[rebuilt_checkpoints.len() - 2]
-        .git_commit_sha
-        .clone();
-    let build_sha = rebuilt_checkpoints
-        .last()
-        .and_then(|checkpoint| checkpoint.git_commit_sha.clone());
+
+    let timeline_shas = timeline_run_shas(repo_dir.path(), &source_run_id);
+    let build_sha = timeline_shas.last().cloned().flatten();
    assert!(build_sha.is_some());

    let before_child = list_metadata_run_ids(repo_dir.path());
@ -204,14 +212,14 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
    exit_code: 0
    ----- stdout -----
    ----- stderr -----
-    Rewound metadata branch to @2 (plan)
+    Rewound metadata branch to @2 (build)
    Rewound run branch fabro/run/[ULID] to [SHA]

    To resume: fabro resume [RUN_PREFIX]
    ");

-    let rewound_child = latest_metadata_checkpoint(repo_dir.path(), &source_run_id);
-    assert_eq!(rewound_child.git_commit_sha, plan_sha);
+    let rewound_timeline_shas = timeline_run_shas(repo_dir.path(), &source_run_id);
+    assert_eq!(rewound_timeline_shas.last().cloned().flatten(), build_sha);

    let before_grandchild = list_metadata_run_ids(repo_dir.path());
    context
@ -229,5 +237,5 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
    assert_eq!(grandchild_run_ids.len(), 1, "expected one grandchild run");

    let grandchild_checkpoint = latest_metadata_checkpoint(repo_dir.path(), &grandchild_run_ids[0]);
-    assert_eq!(grandchild_checkpoint.git_commit_sha, plan_sha);
+    assert_eq!(grandchild_checkpoint.git_commit_sha, build_sha);
 }
--- a/lib/crates/fabro-retro/src/retro_agent.rs
+++ b/lib/crates/fabro-retro/src/retro_agent.rs
@ -10,7 +10,7 @@ use fabro_agent::{
 use fabro_llm::client::Client;
 use fabro_llm::provider::Provider;
 use fabro_llm::types::ToolDefinition;
-use fabro_store::RunDatabase;
+use fabro_store::{EventEnvelope, RunProjection};
 use tokio::task::JoinHandle;

 use crate::retro::{RetroNarrative, SmoothnessRating};
@ -135,7 +135,8 @@ pub fn build_retro_prompt(retro_data_dir: &str) -> String {
 /// files via tool access, then calls `submit_retro` with its analysis.
 pub async fn run_retro_agent(
    sandbox: &Arc<dyn Sandbox>,
-    run_store: &RunDatabase,
+    state: &RunProjection,
+    events: &[EventEnvelope],
    run_dir: &Path,
    llm_client: &Client,
    provider: Provider,
@ -144,7 +145,7 @@ pub async fn run_retro_agent(
 ) -> anyhow::Result<RetroAgentResult> {
    // Upload data files into sandbox (needed for Daytona; no-op effect for local
    // since the agent can also read from the original paths via tools).
-    upload_data_files(sandbox, run_store, run_dir, RETRO_DATA_DIR).await?;
+    upload_data_files(sandbox, state, events, run_dir, RETRO_DATA_DIR).await?;

    // Build provider profile with the submit_retro tool
    let captured: Arc<Mutex<Option<RetroNarrative>>> = Arc::new(Mutex::new(None));
@ -292,7 +293,8 @@ fn build_profile(provider: Provider, model: &str) -> Box<dyn AgentProfile> {

 async fn upload_data_files(
    sandbox: &Arc<dyn Sandbox>,
-    run_store: &RunDatabase,
+    state: &RunProjection,
+    events: &[EventEnvelope],
    _run_dir: &Path,
    target_dir: &str,
 ) -> anyhow::Result<()> {
@ -302,19 +304,16 @@ async fn upload_data_files(
        .await
        .map_err(|e| anyhow::anyhow!("Failed to create retro data dir: {e}"))?;

-    let progress_content = match run_store.list_events().await {
-        Ok(envelopes) => {
-            let lines: Vec<String> = envelopes
-                .into_iter()
-                .filter_map(|env| serde_json::to_string(env.payload.as_value()).ok())
-                .collect();
-            if lines.is_empty() {
-                None
-            } else {
-                Some(lines.join("\n") + "\n")
-            }
+    let progress_content = {
+        let lines: Vec<String> = events
+            .iter()
+            .filter_map(|env| serde_json::to_string(env.payload.as_value()).ok())
+            .collect();
+        if lines.is_empty() {
+            None
+        } else {
+            Some(lines.join("\n") + "\n")
        }
-        Err(e) => return Err(anyhow::anyhow!("Failed to load events from store: {e}")),
    };
    if let Some(content) = progress_content {
        sandbox
@ -323,24 +322,23 @@ async fn upload_data_files(
            .map_err(|e| anyhow::anyhow!("Failed to upload progress.jsonl: {e}"))?;
    }

-    let state = run_store
-        .state()
-        .await
-        .map_err(|e| anyhow::anyhow!("Failed to load run state from store: {e}"))?;
    let checkpoint_content = state
        .checkpoint
+        .clone()
        .map(|cp| serde_json::to_string_pretty(&cp))
        .transpose()?;
    upload_file(sandbox, target_dir, "checkpoint.json", checkpoint_content).await?;

    let run_content = state
        .run
+        .clone()
        .map(|run| serde_json::to_string_pretty(&run))
        .transpose()?;
    upload_file(sandbox, target_dir, "run.json", run_content).await?;

    let start_content = state
        .start
+        .clone()
        .map(|start| serde_json::to_string_pretty(&start))
        .transpose()?;
    upload_file(sandbox, target_dir, "start.json", start_content).await?;
--- a/lib/crates/fabro-server/src/demo/mod.rs
+++ b/lib/crates/fabro-server/src/demo/mod.rs
@ -215,7 +215,26 @@ pub(crate) async fn run_events_stub(
    State(_state): State<Arc<AppState>>,
    Path(_id): Path<String>,
 ) -> Response {
-    ApiError::new(StatusCode::GONE, "Event stream closed.").into_response()
+    let events = vec![Ok::<_, std::convert::Infallible>(
+        Event::default().data(
+            json!({
+                "seq": 2,
+                "payload": {
+                    "id": "evt_demo_attach_completed",
+                    "ts": "2026-04-06T15:00:02Z",
+                    "run_id": "01JQ0000000000000000000001",
+                    "event": "run.completed",
+                    "properties": {
+                        "duration_ms": 42,
+                        "artifact_count": 0,
+                        "status": "success"
+                    }
+                }
+            })
+            .to_string(),
+        ),
+    )];
+    Sse::new(tokio_stream::iter(events)).into_response()
 }

 pub(crate) async fn checkpoint_stub(
--- a/lib/crates/fabro-server/src/server.rs
+++ b/lib/crates/fabro-server/src/server.rs
@ -31,7 +31,7 @@ use fabro_llm::types::{
 use fabro_model::{BilledModelUsage, BilledTokenCounts};
 use fabro_store::{ArtifactStore, Database, EventEnvelope, EventPayload, StageId};
 use fabro_types::{
-    RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance,
+    EventBody, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance,
    RunServerProvenance, RunSubjectProvenance, Settings,
 };
 use fabro_util::redact::redact_jsonl_line;
@ -49,11 +49,12 @@ use tokio::sync::Notify;
 use tokio::sync::RwLock as AsyncRwLock;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::error::RecvError;
+use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio::task::spawn_blocking;
 use tokio::time::sleep;
 use tokio_stream::StreamExt;
-use tokio_stream::wrappers::BroadcastStream;
+use tokio_stream::wrappers::{BroadcastStream, UnboundedReceiverStream};
 use tower::{ServiceExt, service_fn};
 use ulid::Ulid;

@ -1036,6 +1037,23 @@ fn sse_event_from_store(event: &EventEnvelope) -> Option<Event> {
    Some(Event::default().data(data))
 }

+fn attach_event_is_terminal(event: &EventEnvelope) -> bool {
+    let Ok(run_event) = RunEvent::try_from(&event.payload) else {
+        return false;
+    };
+    matches!(
+        run_event.body,
+        EventBody::RunCompleted(_) | EventBody::RunFailed(_)
+    )
+}
+
+fn run_projection_is_active(state: &fabro_store::RunProjection) -> bool {
+    state
+        .status
+        .as_ref()
+        .is_some_and(|record| record.status.is_active())
+}
+
 fn dir_size(path: &std::path::Path) -> u64 {
    walkdir::WalkDir::new(path)
        .into_iter()
@ -1325,21 +1343,20 @@ async fn get_aggregate_billing(
            stages: totals.stages,
        })
        .collect();
-    let total_billing =
-        by_model
-            .iter()
-            .fold(BilledTokenCounts::default(), |mut acc, model| {
-                acc.input_tokens += model.billing.input_tokens;
-                acc.output_tokens += model.billing.output_tokens;
-                acc.reasoning_tokens += model.billing.reasoning_tokens.unwrap_or(0);
-                acc.cache_read_tokens += model.billing.cache_read_tokens.unwrap_or(0);
-                acc.cache_write_tokens += model.billing.cache_write_tokens.unwrap_or(0);
-                acc.total_tokens += model.billing.total_tokens;
-                if let Some(value) = model.billing.total_usd_micros {
-                    *acc.total_usd_micros.get_or_insert(0) += value;
-                }
-                acc
-            });
+    let total_billing = by_model
+        .iter()
+        .fold(BilledTokenCounts::default(), |mut acc, model| {
+            acc.input_tokens += model.billing.input_tokens;
+            acc.output_tokens += model.billing.output_tokens;
+            acc.reasoning_tokens += model.billing.reasoning_tokens.unwrap_or(0);
+            acc.cache_read_tokens += model.billing.cache_read_tokens.unwrap_or(0);
+            acc.cache_write_tokens += model.billing.cache_write_tokens.unwrap_or(0);
+            acc.total_tokens += model.billing.total_tokens;
+            if let Some(value) = model.billing.total_usd_micros {
+                *acc.total_usd_micros.get_or_insert(0) += value;
+            }
+            acc
+        });
    let response = AggregateBilling {
        totals: AggregateBillingTotals {
            cache_read_tokens: nonzero_i64(total_billing.cache_read_tokens),
@ -2774,7 +2791,7 @@ async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
        run_store.subscribe(),
        state.global_event_tx.clone(),
    ));
-    let persisted = match Persisted::load_from_store(&run_store, &run_dir).await {
+    let persisted = match Persisted::load_from_store(&run_store.clone().into(), &run_dir).await {
        Ok(persisted) => persisted,
        Err(e) => {
            tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run");
@ -2810,7 +2827,7 @@ async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
        cancel_token: Some(Arc::clone(&cancel_token)),
        emitter: Arc::clone(&emitter),
        interviewer: Arc::clone(&interviewer) as Arc<dyn Interviewer>,
-        run_store: run_store.clone(),
+        run_store: run_store.clone().into(),
        event_sink: workflow_event::RunEventSink::store(run_store.clone()),
        run_control: None,
        github_app,
@ -3427,20 +3444,6 @@ async fn attach_run_events(
        Ok(id) => id,
        Err(response) => return response,
    };
-    {
-        let runs = state.runs.lock().expect("runs lock poisoned");
-        let Some(managed_run) = runs.get(&id) else {
-            return ApiError::not_found("Run not found.").into_response();
-        };
-        if !matches!(
-            managed_run.status,
-            RunStatus::Queued | RunStatus::Starting | RunStatus::Running | RunStatus::Paused
-        ) {
-            return ApiError::new(StatusCode::GONE, "Run is not live on this server.")
-                .into_response();
-        }
-    }
-
    let Ok(run_store) = state.store.open_run_reader(&id).await else {
        return ApiError::not_found("Run not found.").into_response();
    };
@ -3455,19 +3458,110 @@ async fn attach_run_events(
            }
        },
    };
-    let stream = match run_store.watch_events_from(start_seq) {
-        Ok(stream) => stream,
-        Err(err) => {
-            return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
-                .into_response();
+    const ATTACH_REPLAY_BATCH_LIMIT: usize = 256;
+
+    let (sender, receiver) = mpsc::unbounded_channel();
+    tokio::spawn(async move {
+        let mut next_seq = start_seq;
+
+        loop {
+            let replay_batch = match run_store
+                .list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT)
+                .await
+            {
+                Ok(events) => events,
+                Err(_) => return,
+            };
+            let replay_has_more = replay_batch.len() > ATTACH_REPLAY_BATCH_LIMIT;
+
+            for event in replay_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) {
+                next_seq = event.seq.saturating_add(1);
+                let terminal = attach_event_is_terminal(&event);
+                if let Some(sse_event) = sse_event_from_store(&event) {
+                    if sender
+                        .send(Ok::<Event, std::convert::Infallible>(sse_event))
+                        .is_err()
+                    {
+                        return;
+                    }
+                }
+                if terminal {
+                    return;
+                }
+            }
+
+            if replay_has_more {
+                continue;
+            }
+
+            let state = match run_store.state().await {
+                Ok(state) => state,
+                Err(_) => return,
+            };
+
+            if run_projection_is_active(&state) {
+                break;
+            }
+
+            let tail_batch = match run_store
+                .list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT)
+                .await
+            {
+                Ok(events) => events,
+                Err(_) => return,
+            };
+            let tail_has_more = tail_batch.len() > ATTACH_REPLAY_BATCH_LIMIT;
+
+            for event in tail_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) {
+                next_seq = event.seq.saturating_add(1);
+                let terminal = attach_event_is_terminal(&event);
+                if let Some(sse_event) = sse_event_from_store(&event) {
+                    if sender
+                        .send(Ok::<Event, std::convert::Infallible>(sse_event))
+                        .is_err()
+                    {
+                        return;
+                    }
+                }
+                if terminal {
+                    return;
+                }
+            }
+
+            if tail_has_more {
+                continue;
+            }
+
+            return;
+        }
+
+        let mut live_stream = match run_store.watch_events_from(next_seq) {
+            Ok(stream) => stream,
+            Err(_) => return,
+        };
+
+        while let Some(result) = live_stream.next().await {
+            let Ok(event) = result else {
+                return;
+            };
+            let terminal = attach_event_is_terminal(&event);
+            if let Some(sse_event) = sse_event_from_store(&event) {
+                if sender
+                    .send(Ok::<Event, std::convert::Infallible>(sse_event))
+                    .is_err()
+                {
+                    return;
+                }
+            }
+            if terminal {
+                return;
+            }
        }
-    };
-    let stream = stream.filter_map(|result| match result {
-        Ok(event) => sse_event_from_store(&event).map(Ok::<Event, std::convert::Infallible>),
-        Err(_) => None,
    });

-    Sse::new(stream).into_response()
+    Sse::new(UnboundedReceiverStream::new(receiver))
+        .keep_alive(KeepAlive::default())
+        .into_response()
 }

 async fn get_checkpoint(
@ -6301,7 +6395,7 @@ mod tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn cancel_before_run_transitions_to_running_closes_event_stream() {
+    async fn cancel_before_run_transitions_to_running_returns_empty_attach_stream() {
        let state = create_app_state_with_registry_factory(|interviewer| {
            std::thread::sleep(std::time::Duration::from_millis(200));
            fabro_workflow::handler::default_registry(interviewer, || None)
@ -6330,7 +6424,9 @@ mod tests {
            .body(Body::empty())
            .unwrap();
        let response = app.oneshot(req).await.unwrap();
-        assert_eq!(response.status(), StatusCode::GONE);
+        assert_eq!(response.status(), StatusCode::OK);
+        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        assert!(body.is_empty(), "expected an empty attach stream");
    }

    #[tokio::test]
--- a/lib/crates/fabro-server/tests/it/scenario/run_completion.rs
+++ b/lib/crates/fabro-server/tests/it/scenario/run_completion.rs
@ -1,4 +1,4 @@
-use axum::body::Body;
+use axum::body::{Body, to_bytes};
 use axum::http::{Request, StatusCode};
 use tokio::time::sleep;
 use tower::ServiceExt;
@ -36,22 +36,52 @@ async fn attach_run_events_returns_sse_stream() {
        .unwrap();

    let response = app.oneshot(req).await.unwrap();
-    let status = response.status();
+    assert_eq!(response.status(), StatusCode::OK);
+    let content_type = response
+        .headers()
+        .get("content-type")
+        .expect("content-type header should be present")
+        .to_str()
+        .unwrap();
    assert!(
-        status == StatusCode::OK || status == StatusCode::GONE,
-        "unexpected status: {status}"
+        content_type.contains("text/event-stream"),
+        "expected text/event-stream, got: {content_type}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn attach_run_events_replays_terminal_event_after_completion() {
+    let state = test_app_state_with_options(dry_run_settings(), 5);
+    let app = test_app_with_scheduler(state);
+
+    let run_id = create_and_start_run(&app, MINIMAL_DOT).await;
+    let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
+    assert_eq!(status, "succeeded");
+
+    let req = Request::builder()
+        .method("GET")
+        .uri(api(&format!("/runs/{run_id}/attach?since_seq=1")))
+        .body(Body::empty())
+        .unwrap();
+
+    let response = app.oneshot(req).await.unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+
+    let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    let body = String::from_utf8(body.to_vec()).unwrap();
+    let event_names = body
+        .lines()
+        .filter_map(|line| line.strip_prefix("data:"))
+        .filter_map(|line| serde_json::from_str::<serde_json::Value>(line.trim()).ok())
+        .filter_map(|event| event["payload"]["event"].as_str().map(ToString::to_string))
+        .collect::<Vec<_>>();
+
+    assert!(
+        event_names.iter().any(|event| event == "run.completed"),
+        "expected a replayed terminal event, got {event_names:?}"
+    );
+    assert_eq!(
+        event_names.last().map(String::as_str),
+        Some("run.completed")
    );
-
-    if status == StatusCode::OK {
-        let content_type = response
-            .headers()
-            .get("content-type")
-            .expect("content-type header should be present")
-            .to_str()
-            .unwrap();
-        assert!(
-            content_type.contains("text/event-stream"),
-            "expected text/event-stream, got: {content_type}"
-        );
-    }
 }
--- a/lib/crates/fabro-server/tests/it/scenario/sse.rs
+++ b/lib/crates/fabro-server/tests/it/scenario/sse.rs
@ -54,15 +54,12 @@ async fn sse_stream_contains_expected_event_types() {
        .body(Body::empty())
        .unwrap();
    let response = app.clone().oneshot(req).await.unwrap();
-    // May be 200 (stream open) or 410 (run completed before connect)
    let sse_status = response.status();
-    assert!(
-        sse_status == StatusCode::OK || sse_status == StatusCode::GONE,
-        "expected 200 or 410, got: {sse_status}"
+    assert_eq!(
+        sse_status,
+        StatusCode::OK,
+        "expected 200, got: {sse_status}"
    );
-    if sse_status == StatusCode::GONE {
-        return;
-    }

    let content_type = response
        .headers()
@ -96,8 +93,8 @@ async fn sse_stream_contains_expected_event_types() {

    // Because we subscribe while the run is only guaranteed to be past
    // "queued", a live stream should include at least one stage event.
-    // A 410 response above still covers the case where the run completed
-    // before we managed to attach.
+    // If the run completes before we attach with no unread events, an empty
+    // stream is still a valid 200 response.
    if !event_types.is_empty() {
        assert!(
            event_types
--- a/lib/crates/fabro-store/src/run_state.rs
+++ b/lib/crates/fabro-store/src/run_state.rs
@ -57,7 +57,7 @@ pub(crate) struct EventProjectionCache {
 }

 impl RunProjection {
-    pub(crate) fn apply_events(events: &[EventEnvelope]) -> Result<Self> {
+    pub fn apply_events(events: &[EventEnvelope]) -> Result<Self> {
        let mut state = Self::default();
        for event in events {
            state.apply_event(event)?;
@ -65,7 +65,7 @@ impl RunProjection {
        Ok(state)
    }

-    pub(crate) fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> {
+    pub fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> {
        let stored = RunEvent::from_ref(event.payload.as_value())
            .map_err(|err| StoreError::InvalidEvent(format!("invalid stored event: {err}")))?;
        let ts = stored.ts;
--- a/lib/crates/fabro-store/src/slate/run_store.rs
+++ b/lib/crates/fabro-store/src/slate/run_store.rs
@ -240,6 +240,7 @@ impl RunDatabase {
        let inner = Arc::clone(&self.inner);
        let (sender, receiver) = mpsc::unbounded_channel();
        tokio::spawn(async move {
+            let mut rx = inner.event_tx.subscribe();
            let cached = {
                let recent_events = inner.recent_events.lock().await;
                recent_events
@ -256,8 +257,29 @@ impl RunDatabase {
                }
            }

-            let mut rx = inner.event_tx.subscribe();
-            while let Ok(event) = rx.recv().await {
+            loop {
+                loop {
+                    match rx.try_recv() {
+                        Ok(event) => {
+                            if event.seq < next_seq {
+                                continue;
+                            }
+                            next_seq = event.seq.saturating_add(1);
+                            if sender.send(Ok(event)).is_err() {
+                                return;
+                            }
+                        }
+                        Err(broadcast::error::TryRecvError::Empty) => break,
+                        Err(broadcast::error::TryRecvError::Lagged(_)) => continue,
+                        Err(broadcast::error::TryRecvError::Closed) => return,
+                    }
+                }
+
+                let event = match rx.recv().await {
+                    Ok(event) => event,
+                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(broadcast::error::RecvError::Closed) => return,
+                };
                if event.seq < next_seq {
                    continue;
                }
--- a/lib/crates/fabro-types/src/run_event/mod.rs
+++ b/lib/crates/fabro-types/src/run_event/mod.rs
@ -701,7 +701,7 @@ mod tests {
                status: crate::StageStatus::Success,
                preferred_label: None,
                suggested_next_ids: vec!["next".to_string()],
-                usage: None,
+                billing: None,
                failure: None,
                notes: Some("done".to_string()),
                files_touched: vec!["src/main.rs".to_string()],
@ -770,7 +770,7 @@ mod tests {
            status: crate::StageStatus::Success,
            preferred_label: None,
            suggested_next_ids: vec!["next".to_string()],
-            usage: None,
+            billing: None,
            failure: None,
            notes: Some("done".to_string()),
            files_touched: vec!["src/main.rs".to_string()],
--- a/lib/crates/fabro-workflow/Cargo.toml
+++ b/lib/crates/fabro-workflow/Cargo.toml
@ -40,6 +40,7 @@ thiserror.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
+bytes.workspace = true
 object_store.workspace = true
 ulid.workspace = true
 uuid.workspace = true
--- a/lib/crates/fabro-workflow/src/artifact.rs
+++ b/lib/crates/fabro-workflow/src/artifact.rs
@ -4,9 +4,9 @@ use std::path::Path;
 use serde_json::Value;

 use fabro_agent::Sandbox;
-use fabro_store::RunDatabase;

 use crate::error::{FabroError, Result};
+use crate::runtime_store::RunStoreHandle;

 /// Threshold above which values are persisted as blobs and materialized to disk (100KB).
 const BLOB_OFFLOAD_THRESHOLD: usize = 100 * 1024;
@ -26,7 +26,7 @@ const ARTIFACT_POINTER_PREFIX: &str = "file://";
 /// Returns an error if blob persistence or cache materialization fails.
 pub async fn offload_large_values(
    updates: &mut HashMap<String, Value>,
-    run_store: &RunDatabase,
+    run_store: &RunStoreHandle,
    cache_dir: &Path,
 ) -> Result<()> {
    std::fs::create_dir_all(cache_dir)?;
@ -161,7 +161,7 @@ mod tests {
        let mut updates = HashMap::new();
        updates.insert("response.plan".to_string(), serde_json::json!(large_string));

-        offload_large_values(&mut updates, &run_store, dir.path())
+        offload_large_values(&mut updates, &run_store.clone().into(), dir.path())
            .await
            .unwrap();

@ -196,7 +196,7 @@ mod tests {
        let mut updates = HashMap::new();
        updates.insert("small_key".to_string(), small_value.clone());

-        offload_large_values(&mut updates, &run_store, dir.path())
+        offload_large_values(&mut updates, &run_store.clone().into(), dir.path())
            .await
            .unwrap();

--- a/lib/crates/fabro-workflow/src/event.rs
+++ b/lib/crates/fabro-workflow/src/event.rs
@ -24,6 +24,8 @@ use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback
 use fabro_llm::types::TokenCounts as LlmTokenCounts;
 use fabro_util::redact::redact_json_value;

+use crate::runtime_store::RunStoreHandle;
+
 pub use fabro_types::{EventBody, RunNoticeLevel};

 /// Events emitted during workflow run execution for observability.
@ -2372,7 +2374,7 @@ pub async fn append_event_to_sink(

 #[derive(Clone)]
 pub enum RunEventSink {
-    Store(RunDatabase),
+    Store(RunStoreHandle),
    JsonLines(Arc<AsyncMutex<Pin<Box<dyn AsyncWrite + Send>>>>),
    Callback(Arc<RunEventSinkCallback>),
    Composite(Vec<Self>),
@ -2384,6 +2386,11 @@ type RunEventSinkCallback = dyn Fn(RunEvent) -> RunEventSinkFuture + Send + Sync
 impl RunEventSink {
    #[must_use]
    pub fn store(run_store: RunDatabase) -> Self {
+        Self::Store(RunStoreHandle::local(run_store))
+    }
+
+    #[must_use]
+    pub fn backend(run_store: RunStoreHandle) -> Self {
        Self::Store(run_store)
    }

@ -2421,12 +2428,7 @@ impl RunEventSink {
        while let Some(sink) = pending.pop() {
            match sink {
                Self::Store(run_store) => {
-                    let payload = build_redacted_event_payload(event, &event.run_id)?;
-                    run_store
-                        .append_event(&payload)
-                        .await
-                        .map(|_| ())
-                        .map_err(anyhow::Error::from)?;
+                    run_store.append_run_event(event).await?;
                }
                Self::JsonLines(writer) => {
                    let line = redacted_event_json(event)?;
@ -2507,9 +2509,9 @@ pub struct StoreProgressLogger {

 impl StoreProgressLogger {
    #[must_use]
-    pub fn new(run_store: RunDatabase) -> Self {
+    pub fn new(run_store: impl Into<RunStoreHandle>) -> Self {
        Self {
-            inner: RunEventLogger::new(RunEventSink::store(run_store)),
+            inner: RunEventLogger::new(RunEventSink::backend(run_store.into())),
        }
    }

--- a/lib/crates/fabro-workflow/src/handler/agent.rs
+++ b/lib/crates/fabro-workflow/src/handler/agent.rs
@ -423,7 +423,7 @@ mod tests {
        let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
        let services = EngineServices {
            emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)),
-            run_store: run_store.clone(),
+            run_store: run_store.clone().into(),
            ..EngineServices::test_default()
        };
        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
--- a/lib/crates/fabro-workflow/src/handler/command.rs
+++ b/lib/crates/fabro-workflow/src/handler/command.rs
@ -202,7 +202,7 @@ mod tests {
        let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
        let services = EngineServices {
            emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)),
-            run_store: run_store.clone(),
+            run_store: run_store.clone().into(),
            ..EngineServices::test_default()
        };
        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
--- a/lib/crates/fabro-workflow/src/handler/manager_loop.rs
+++ b/lib/crates/fabro-workflow/src/handler/manager_loop.rs
@ -251,7 +251,7 @@ impl Handler for SubWorkflowHandler {
                run_options: child_run_options,
                workflow_path: child_workflow_path,
                workflow_bundle,
-                run_store,
+                run_store: run_store.into(),
                checkpoint: None,
                seed_context: Some(child_context),
                emitter,
--- a/lib/crates/fabro-workflow/src/handler/mod.rs
+++ b/lib/crates/fabro-workflow/src/handler/mod.rs
@ -22,7 +22,6 @@ use async_trait::async_trait;
 use fabro_agent::Sandbox;
 #[cfg(test)]
 use fabro_store::Database;
-use fabro_store::RunDatabase;
 #[cfg(test)]
 use object_store::memory::InMemory;

@ -30,6 +29,7 @@ use crate::context::Context;
 use crate::error::FabroError;
 use crate::event::Emitter;
 use crate::outcome::{Outcome, OutcomeExt};
+use crate::runtime_store::RunStoreHandle;
 use crate::sandbox_git::GitState;
 use crate::workflow_bundle::WorkflowBundle;
 use fabro_graphviz::graph::{Graph, Node, shape_to_handler_type};
@ -43,7 +43,7 @@ pub struct EngineServices {
    pub registry: Arc<HandlerRegistry>,
    pub emitter: Arc<Emitter>,
    pub sandbox: Arc<dyn Sandbox>,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    /// Git state for the current run. Set via `set_git_state` at the start of
    /// `run_via_core` and read by parallel/fan-in handlers.
    pub(crate) git_state: std::sync::RwLock<Option<Arc<GitState>>>,
@ -137,7 +137,8 @@ impl EngineServices {
                    })
            })
            .join()
-            .expect("test run store thread should join"),
+            .expect("test run store thread should join")
+            .into(),
            git_state: std::sync::RwLock::new(None),
            hook_runner: None,
            env: HashMap::new(),
--- a/lib/crates/fabro-workflow/src/handler/parallel.rs
+++ b/lib/crates/fabro-workflow/src/handler/parallel.rs
@ -634,7 +634,7 @@ mod tests {
        let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
        let services = EngineServices {
            emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)),
-            run_store: run_store.clone(),
+            run_store: run_store.clone().into(),
            ..EngineServices::test_default()
        };
        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
@ -686,7 +686,7 @@ mod tests {
        let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
        let services = EngineServices {
            emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)),
-            run_store: run_store.clone(),
+            run_store: run_store.clone().into(),
            ..EngineServices::test_default()
        };
        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
--- a/lib/crates/fabro-workflow/src/handler/prompt.rs
+++ b/lib/crates/fabro-workflow/src/handler/prompt.rs
@ -203,7 +203,7 @@ mod tests {
        let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
        let services = EngineServices {
            emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)),
-            run_store: run_store.clone(),
+            run_store: run_store.clone().into(),
            ..EngineServices::test_default()
        };
        let logger = crate::event::StoreProgressLogger::new(run_store.clone());
--- a/lib/crates/fabro-workflow/src/lib.rs
+++ b/lib/crates/fabro-workflow/src/lib.rs
@ -140,6 +140,7 @@ pub mod run_dump;
 pub mod run_lookup;
 pub mod run_options;
 pub mod run_status;
+pub mod runtime_store;
 pub mod sandbox_git;
 #[doc(hidden)]
 pub mod test_support;
--- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs
+++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs
@ -3,7 +3,6 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};

 use async_trait::async_trait;
-use fabro_store::RunDatabase;

 use fabro_core::graph::NodeSpec;
 use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, RunLifecycle};
@ -16,6 +15,7 @@ use crate::event::{Emitter, Event, RunNoticeLevel};
 use crate::graph::WorkflowGraph;
 use crate::graph::WorkflowNode;
 use crate::outcome::BilledModelUsage;
+use crate::runtime_store::RunStoreHandle;
 use fabro_core::error::Result as CoreResult;
 use fabro_core::lifecycle::NodeDecision;

@ -26,7 +26,7 @@ type WfNodeDecision = NodeDecision<Option<BilledModelUsage>>;
 /// Sub-lifecycle responsible for artifact collection, offloading, and syncing.
 pub(crate) struct ArtifactLifecycle {
    pub sandbox: Arc<dyn fabro_sandbox::Sandbox>,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub blob_cache_dir: PathBuf,
    pub emitter: Arc<Emitter>,
    pub artifacts_dir: PathBuf,
@ -40,7 +40,7 @@ impl ArtifactLifecycle {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        sandbox: Arc<dyn fabro_sandbox::Sandbox>,
-        run_store: RunDatabase,
+        run_store: RunStoreHandle,
        blob_cache_dir: PathBuf,
        emitter: Arc<Emitter>,
        artifacts_dir: PathBuf,
--- a/lib/crates/fabro-workflow/src/lifecycle/git.rs
+++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs
@ -4,7 +4,6 @@ use std::sync::{Arc, Mutex};

 use async_trait::async_trait;
 use fabro_config::RunScratch;
-use fabro_store::RunDatabase;
 use fabro_types::RunId;
 use tokio::fs;

@ -21,6 +20,7 @@ use crate::graph::WorkflowNode;
 use crate::outcome::{BilledModelUsage, Outcome, StageStatus};
 use crate::run_dump::RunDump;
 use crate::run_options::RunOptions;
+use crate::runtime_store::RunStoreHandle;
 use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host};

 type WfRunState = ExecutionState<Option<BilledModelUsage>>;
@ -67,7 +67,7 @@ pub(crate) struct GitLifecycle {
    pub emitter: Arc<Emitter>,
    pub run_dir: PathBuf,
    pub run_id: RunId,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub run_options: Arc<RunOptions>,
    pub start_node_id: Option<String>,
    // Cross-lifecycle data (shared with EventLifecycle)
--- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs
+++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs
@ -14,7 +14,6 @@ use std::time::Instant;

 use async_trait::async_trait;
 use fabro_config::RunScratch;
-use fabro_store::RunDatabase;
 use fabro_types::RunId;

 use fabro_core::error::Result as CoreResult;
@ -33,6 +32,7 @@ use crate::graph::WorkflowNode;
 use crate::outcome::{BilledModelUsage, Outcome};
 use crate::run_control::RunControlState;
 use crate::run_options::RunOptions;
+use crate::runtime_store::RunStoreHandle;
 use fabro_graphviz::graph::types::Graph as GvGraph;
 use fabro_hooks::HookRunner;
 use fabro_sandbox::Sandbox;
@ -84,7 +84,7 @@ impl WorkflowLifecycle {
        sandbox: &Arc<dyn Sandbox>,
        graph: Arc<GvGraph>,
        run_dir: &PathBuf,
-        run_store: &RunDatabase,
+        run_store: &RunStoreHandle,
        run_options: &Arc<RunOptions>,
        is_resume: bool,
        on_node: crate::OnNodeCallback,
--- a/lib/crates/fabro-workflow/src/operations/start.rs
+++ b/lib/crates/fabro-workflow/src/operations/start.rs
@ -9,7 +9,6 @@ use fabro_config::{project as project_config, run as run_config, sandbox as sand
 use fabro_interview::{AutoApproveInterviewer, Interviewer};
 use fabro_model::{Catalog, FallbackTarget, Provider};
 use fabro_sandbox::{SandboxProvider, SandboxSpec};
-use fabro_store::RunDatabase;
 use fabro_types::{RunId, Settings};

 use crate::context::Context;
@ -29,6 +28,7 @@ use crate::records::Checkpoint;
 use crate::run_control::RunControlState;
 use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
 use crate::run_status::{RunStatus, StatusReason};
+use crate::runtime_store::RunStoreHandle;
 use crate::workflow_bundle::{StoredWorkflowBundle, WorkflowBundle};
 use fabro_config::run::PullRequestSettings;
 use fabro_retro::retro::Retro;
@ -48,7 +48,7 @@ struct RunSession {
    sandbox_env: SandboxEnvSpec,
    devcontainer: Option<DevcontainerSpec>,
    seed_context: Option<Context>,
-    run_store: RunDatabase,
+    run_store: RunStoreHandle,
    event_sink: RunEventSink,
    git: Option<GitCheckpointOptions>,
    github_app: Option<fabro_github::GitHubAppCredentials>,
@ -70,7 +70,7 @@ pub struct StartServices {
    pub cancel_token: Option<Arc<AtomicBool>>,
    pub emitter: Arc<Emitter>,
    pub interviewer: Arc<dyn Interviewer>,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub event_sink: RunEventSink,
    pub run_control: Option<Arc<RunControlState>>,
    pub github_app: Option<fabro_github::GitHubAppCredentials>,
@ -221,7 +221,7 @@ pub(super) async fn execute_persisted_run(

 async fn persist_terminal_engine_failure(
    run_id: RunId,
-    run_store: &RunDatabase,
+    run_store: &RunStoreHandle,
    event_sink: &RunEventSink,
    _run_dir: &Path,
    error: &FabroError,
@ -867,7 +867,7 @@ mod tests {
            cancel_token: None,
            emitter,
            interviewer: Arc::new(fabro_interview::AutoApproveInterviewer),
-            run_store: store.open_run(&fixtures::RUN_1).await.unwrap(),
+            run_store: store.open_run(&fixtures::RUN_1).await.unwrap().into(),
            event_sink: RunEventSink::store(store.open_run(&fixtures::RUN_1).await.unwrap()),
            run_control: None,
            github_app: None,
@ -1005,7 +1005,7 @@ mod tests {
            node_visits: HashMap::new(),
        };
        crate::event::append_event(
-            &services.run_store,
+            &store.open_run(&fixtures::RUN_1).await.unwrap(),
            &services.run_id,
            &Event::CheckpointCompleted {
                node_id: checkpoint.current_node.clone(),
--- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
@ -185,7 +185,7 @@ async fn execute_test_run_with_options(
        persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value),
        InitOptions {
            run_id: run_id_value,
-            run_store,
+            run_store: run_store.into(),
            dry_run: false,
            emitter,
            sandbox: SandboxSpec::Local {
@ -241,7 +241,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
        persisted_workflow(graph, source, &run_dir, test_run_id("run-test")),
        InitOptions {
            run_id: test_run_id("run-test"),
-            run_store: test_run_store(&test_run_id("run-test")).await,
+            run_store: test_run_store(&test_run_id("run-test")).await.into(),
            dry_run: false,
            emitter: test_emitter_arc("run-test"),
            sandbox: SandboxSpec::Local {
@ -311,7 +311,7 @@ async fn run_with_lifecycle(
        persisted_workflow(graph.clone(), String::new(), &run_dir, run_id),
        InitOptions {
            run_id,
-            run_store: test_run_store(&run_id).await,
+            run_store: test_run_store(&run_id).await.into(),
            dry_run: false,
            emitter,
            sandbox: SandboxSpec::Local {
--- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs
@ -8,9 +8,9 @@ use crate::records::{Checkpoint, Conclusion, StageSummary};
 use crate::run_dump::RunDump;
 use crate::run_options::RunOptions;
 use crate::run_status::{RunStatus, StatusReason};
+use crate::runtime_store::RunStoreHandle;
 use crate::sandbox_git::git_push_host;
 use fabro_hooks::{HookContext, HookEvent, HookRunner};
-use fabro_store::RunDatabase;
 use fabro_types::BilledTokenCounts;

 use super::types::{Concluded, FinalizeOptions, Retroed};
@ -64,7 +64,7 @@ pub fn classify_engine_result(
 }

 pub(crate) async fn build_conclusion_from_store(
-    run_store: &RunDatabase,
+    run_store: &RunStoreHandle,
    status: StageStatus,
    failure_reason: Option<String>,
    run_duration_ms: u64,
@ -153,7 +153,7 @@ fn build_conclusion_from_parts(
 ///
 /// This captures the last diff.patch (written after the final checkpoint) and retro.json.
 /// Best-effort: errors are logged as warnings.
-pub async fn write_finalize_commit(run_options: &RunOptions, run_store: &RunDatabase) {
+pub async fn write_finalize_commit(run_options: &RunOptions, run_store: &RunStoreHandle) {
    let (Some(meta_branch), Some(repo_path)) = (
        run_options
            .git
@ -356,7 +356,7 @@ mod tests {
            graph: Graph::new("test"),
            outcome: Ok(Outcome::success()),
            run_options: test_run_options(&run_dir),
-            run_store: run_store.clone(),
+            run_store: run_store.clone().into(),
            hook_runner: None,
            emitter,
            sandbox: Arc::new(fabro_agent::LocalSandbox::new(
@ -371,7 +371,7 @@ mod tests {
            &FinalizeOptions {
                run_dir: run_dir.clone(),
                run_id: test_run_id(),
-                run_store: run_store.clone(),
+                run_store: run_store.clone().into(),
                workflow_name: "test".to_string(),
                hook_runner: None,
                preserve_sandbox: true,
--- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs
@ -769,7 +769,7 @@ mod tests {
                run_store: {
                    let store = memory_store();
                    let inner = store.create_run(&test_run_id()).await.unwrap();
-                    inner
+                    inner.into()
                },
                dry_run: false,
                emitter,
@ -846,7 +846,7 @@ mod tests {
            persisted,
            InitOptions {
                run_id: test_run_id(),
-                run_store,
+                run_store: run_store.into(),
                dry_run: false,
                emitter,
                sandbox: SandboxSpec::Local {
--- a/lib/crates/fabro-workflow/src/pipeline/persist.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs
@ -1,8 +1,7 @@
 use std::path::Path;

-use fabro_store::RunDatabase;
-
 use crate::error::FabroError;
+use crate::runtime_store::RunStoreHandle;

 use super::types::{PersistOptions, Persisted, Validated};

@ -26,7 +25,7 @@ pub(crate) fn persist(
 }

 pub(crate) async fn load_from_store(
-    run_store: &RunDatabase,
+    run_store: &RunStoreHandle,
    run_dir: &Path,
 ) -> Result<Persisted, FabroError> {
    let state = run_store
@ -231,7 +230,9 @@ mod tests {
        .unwrap();

        let run_store = seeded_store(&run_dir, &expected, Some(&source)).await;
-        let loaded = load_from_store(&run_store, &run_dir).await.unwrap();
+        let loaded = load_from_store(&run_store.clone().into(), &run_dir)
+            .await
+            .unwrap();

        let loaded_record = loaded.run_record();
        assert_eq!(loaded_record.run_id, expected.run_id);
@ -284,7 +285,9 @@ mod tests {
        record.graph = graph;

        let run_store = seeded_store(&run_dir, &record, None).await;
-        let loaded = load_from_store(&run_store, &run_dir).await.unwrap();
+        let loaded = load_from_store(&run_store.clone().into(), &run_dir)
+            .await
+            .unwrap();

        assert!(loaded.source().is_empty());
    }
@ -300,7 +303,9 @@ mod tests {
        record.graph = graph.clone();

        let run_store = seeded_store(&run_dir, &record, Some(&source)).await;
-        let loaded = load_from_store(&run_store, &run_dir).await.unwrap();
+        let loaded = load_from_store(&run_store.clone().into(), &run_dir)
+            .await
+            .unwrap();

        assert_eq!(
            serde_json::to_value(loaded.graph()).unwrap(),
--- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs
@ -1,5 +1,5 @@
 use fabro_config::run::MergeStrategy;
-use fabro_store::{RunDatabase, RunProjection};
+use fabro_store::RunProjection;
 use fabro_types::PullRequestRecord;
 use tracing::{debug, info};

@ -12,6 +12,7 @@ use super::types::{Concluded, Finalized, PullRequestOptions};
 use crate::event::{Emitter, Event, RunNoticeLevel};
 use crate::outcome::{StageStatus, format_cost as outcome_format_cost};
 use crate::records::{Conclusion, RunRecord};
+use crate::runtime_store::RunStoreHandle;
 use fabro_retro::retro::Retro;

 /// Derive a PR title from the workflow goal.
@ -282,7 +283,7 @@ fn emit_run_notice(
    });
 }

-async fn load_pull_request_diff(run_store: &RunDatabase) -> String {
+async fn load_pull_request_diff(run_store: &RunStoreHandle) -> String {
    run_store
        .state()
        .await
@ -300,7 +301,7 @@ pub async fn build_pr_body(
    diff: &str,
    goal: &str,
    model: &str,
-    run_store: &RunDatabase,
+    run_store: &RunStoreHandle,
    conclusion: Option<&Conclusion>,
 ) -> Result<String, String> {
    debug!("Building PR body");
@ -408,7 +409,7 @@ pub async fn maybe_open_pull_request(
    model: &str,
    draft: bool,
    auto_merge: Option<AutoMergeOptions>,
-    run_store: &RunDatabase,
+    run_store: &RunStoreHandle,
    conclusion: Option<&Conclusion>,
 ) -> Result<Option<PullRequestRecord>, String> {
    if diff.is_empty() {
@ -1060,7 +1061,7 @@ mod tests {
            "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n",
            "Implement feature",
            "mock-model",
-            &run_store,
+            &run_store.clone().into(),
            Some(&conclusion),
        )
        .await
@ -1130,7 +1131,7 @@ mod tests {
            "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n",
            "Implement feature",
            "mock-model",
-            &run_store,
+            &run_store.clone().into(),
            Some(&conclusion),
        )
        .await
@ -1216,7 +1217,7 @@ mod tests {
            "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n",
            "Implement feature",
            "mock-model",
-            &run_store,
+            &run_store.clone().into(),
            Some(&make_test_conclusion()),
        )
        .await
@ -1359,7 +1360,7 @@ mod tests {
            "claude-sonnet-4-20250514",
            false,
            None,
-            &run_store,
+            &run_store.clone().into(),
            None,
        )
        .await;
@ -1425,7 +1426,7 @@ mod tests {
        .await
        .unwrap();

-        let diff = load_pull_request_diff(&run_store).await;
+        let diff = load_pull_request_diff(&run_store.clone().into()).await;

        assert!(diff.contains("from_store"));
    }
--- a/lib/crates/fabro-workflow/src/pipeline/retro.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs
@ -23,7 +23,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
            return None;
        }
    };
-    let Some(cp) = state.checkpoint else {
+    let Some(ref cp) = state.checkpoint else {
        tracing::warn!("Could not load checkpoint, skipping retro");
        if let Some(ref emitter) = options.emitter {
            emitter.emit(&Event::RetroFailed {
@ -80,9 +80,23 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
                    }
                })
            });
+        let events = match options.run_store.list_events().await {
+            Ok(events) => events,
+            Err(err) => {
+                tracing::warn!(error = %err, "Could not load events from store, skipping retro");
+                if let Some(ref emitter) = options.emitter {
+                    emitter.emit(&Event::RetroFailed {
+                        error: err.to_string(),
+                        duration_ms: 0,
+                    });
+                }
+                return None;
+            }
+        };
        run_retro_agent(
            &options.sandbox,
-            &options.run_store,
+            &state,
+            &events,
            &options.run_dir,
            client,
            options.provider,
@ -319,7 +333,7 @@ mod tests {
            graph: Graph::new("test"),
            outcome: Ok(crate::outcome::Outcome::success()),
            run_options: test_run_options(&run_dir),
-            run_store: run_store.clone(),
+            run_store: run_store.clone().into(),
            hook_runner: None,
            emitter: Arc::clone(&emitter),
            sandbox: Arc::clone(&sandbox),
@ -334,7 +348,7 @@ mod tests {
            executed,
            &RetroOptions {
                run_id: test_run_id(),
-                run_store,
+                run_store: run_store.into(),
                workflow_name: "test".to_string(),
                goal: "Ship it".to_string(),
                run_dir: run_dir.clone(),
@ -371,7 +385,7 @@ mod tests {
        let retro = run_retro(
            &RetroOptions {
                run_id: test_run_id(),
-                run_store: test_run_store(&run_dir, &checkpoint).await,
+                run_store: test_run_store(&run_dir, &checkpoint).await.into(),
                workflow_name: "test".to_string(),
                goal: "Ship it".to_string(),
                run_dir: run_dir.clone(),
--- a/lib/crates/fabro-workflow/src/pipeline/types.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/types.rs
@ -11,7 +11,6 @@ use fabro_llm::Provider;
 use fabro_mcp::config::McpServerSettings;
 use fabro_model::FallbackTarget;
 use fabro_sandbox::SandboxSpec;
-use fabro_store::RunDatabase;
 use fabro_types::RunId;
 use fabro_validate::Diagnostic;

@ -24,6 +23,7 @@ use crate::outcome::Outcome;
 use crate::records::{Checkpoint, Conclusion, RunRecord};
 use crate::run_control::RunControlState;
 use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
+use crate::runtime_store::RunStoreHandle;
 use crate::transforms::Transform;
 use crate::workflow_bundle::WorkflowBundle;
 use fabro_config::run::PullRequestSettings;
@ -198,7 +198,7 @@ impl Persisted {
    }

    pub async fn load_from_store(
-        run_store: &RunDatabase,
+        run_store: &RunStoreHandle,
        run_dir: &Path,
    ) -> Result<Self, FabroError> {
        super::persist::load_from_store(run_store, run_dir).await
@ -230,7 +230,7 @@ pub struct DevcontainerSpec {

 pub struct InitOptions {
    pub run_id: RunId,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub dry_run: bool,
    pub emitter: Arc<Emitter>,
    pub sandbox: SandboxSpec,
@ -259,7 +259,7 @@ pub struct Initialized {
    pub run_options: RunOptions,
    pub workflow_path: Option<PathBuf>,
    pub workflow_bundle: Option<Arc<WorkflowBundle>>,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub(crate) checkpoint: Option<Checkpoint>,
    pub(crate) seed_context: Option<Context>,
    pub emitter: Arc<Emitter>,
@ -281,7 +281,7 @@ pub struct Executed {
    pub graph: Graph,
    pub outcome: Result<Outcome, FabroError>,
    pub run_options: RunOptions,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub hook_runner: Option<Arc<HookRunner>>,
    pub emitter: Arc<Emitter>,
    pub sandbox: Arc<dyn Sandbox>,
@ -298,7 +298,7 @@ pub struct Retroed {
    pub graph: Graph,
    pub outcome: Result<Outcome, FabroError>,
    pub run_options: RunOptions,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub hook_runner: Option<Arc<HookRunner>>,
    pub emitter: Arc<Emitter>,
    pub sandbox: Arc<dyn Sandbox>,
@ -338,7 +338,7 @@ pub struct TransformOptions {
 /// Options for the RETRO phase.
 pub struct RetroOptions {
    pub run_id: RunId,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub workflow_name: String,
    pub goal: String,
    pub run_dir: PathBuf,
@ -356,7 +356,7 @@ pub struct RetroOptions {
 pub struct FinalizeOptions {
    pub run_dir: PathBuf,
    pub run_id: RunId,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub workflow_name: String,
    pub hook_runner: Option<Arc<HookRunner>>,
    pub preserve_sandbox: bool,
@ -366,7 +366,7 @@ pub struct FinalizeOptions {
 /// Options for the PULL_REQUEST phase.
 pub struct PullRequestOptions {
    pub run_dir: PathBuf,
-    pub run_store: RunDatabase,
+    pub run_store: RunStoreHandle,
    pub pr_config: Option<PullRequestSettings>,
    pub github_app: Option<fabro_github::GitHubAppCredentials>,
    pub origin_url: Option<String>,
--- a/lib/crates/fabro-workflow/src/runtime_store.rs
+++ b/lib/crates/fabro-workflow/src/runtime_store.rs
@ -0,0 +1,206 @@
+use std::sync::Arc;
+
+use anyhow::Result;
+use async_trait::async_trait;
+use bytes::Bytes;
+use fabro_store::{EventEnvelope, RunDatabase, RunProjection};
+use fabro_types::{RunBlobId, RunEvent};
+
+use crate::event::build_redacted_event_payload;
+
+#[async_trait]
+pub trait RunStoreBackend: Send + Sync {
+    async fn load_state(&self) -> Result<RunProjection>;
+    async fn list_events(&self) -> Result<Vec<EventEnvelope>>;
+    async fn append_run_event(&self, event: &RunEvent) -> Result<()>;
+    async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId>;
+    async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>>;
+}
+
+#[derive(Clone)]
+pub struct RunStoreHandle {
+    backend: Arc<dyn RunStoreBackend>,
+}
+
+impl RunStoreHandle {
+    #[must_use]
+    pub fn new(backend: Arc<dyn RunStoreBackend>) -> Self {
+        Self { backend }
+    }
+
+    #[must_use]
+    pub fn local(run_store: RunDatabase) -> Self {
+        Self::new(Arc::new(LocalRunStoreBackend { run_store }))
+    }
+
+    pub async fn state(&self) -> Result<RunProjection> {
+        self.backend.load_state().await
+    }
+
+    pub async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
+        self.backend.list_events().await
+    }
+
+    pub async fn append_run_event(&self, event: &RunEvent) -> Result<()> {
+        self.backend.append_run_event(event).await
+    }
+
+    pub async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
+        self.backend.write_blob(data).await
+    }
+
+    pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
+        self.backend.read_blob(id).await
+    }
+}
+
+impl From<RunDatabase> for RunStoreHandle {
+    fn from(value: RunDatabase) -> Self {
+        Self::local(value)
+    }
+}
+
+struct LocalRunStoreBackend {
+    run_store: RunDatabase,
+}
+
+#[async_trait]
+impl RunStoreBackend for LocalRunStoreBackend {
+    async fn load_state(&self) -> Result<RunProjection> {
+        self.run_store.state().await.map_err(anyhow::Error::from)
+    }
+
+    async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
+        self.run_store
+            .list_events()
+            .await
+            .map_err(anyhow::Error::from)
+    }
+
+    async fn append_run_event(&self, event: &RunEvent) -> Result<()> {
+        let payload = build_redacted_event_payload(event, &event.run_id)?;
+        self.run_store
+            .append_event(&payload)
+            .await
+            .map(|_| ())
+            .map_err(anyhow::Error::from)
+    }
+
+    async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
+        self.run_store
+            .write_blob(data)
+            .await
+            .map_err(anyhow::Error::from)
+    }
+
+    async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
+        self.run_store
+            .read_blob(id)
+            .await
+            .map_err(anyhow::Error::from)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::path::PathBuf;
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use chrono::Utc;
+    use fabro_graphviz::graph::Graph;
+    use fabro_store::Database;
+    use fabro_types::fixtures;
+    use fabro_types::run_event::RunStatusTransitionProps;
+    use fabro_types::{EventBody, RunEvent, Settings};
+    use object_store::memory::InMemory;
+
+    use super::RunStoreHandle;
+    use crate::event::{Event, append_event};
+    use crate::records::RunRecord;
+
+    async fn test_run_store() -> fabro_store::RunDatabase {
+        let store = Arc::new(Database::new(
+            Arc::new(InMemory::new()),
+            "",
+            Duration::from_millis(1),
+        ));
+        store.create_run(&fixtures::RUN_1).await.unwrap()
+    }
+
+    fn test_run_record() -> RunRecord {
+        RunRecord {
+            run_id: fixtures::RUN_1,
+            settings: Settings::default(),
+            graph: Graph::new("test"),
+            workflow_slug: Some("test".to_string()),
+            working_directory: PathBuf::from("/tmp/test"),
+            host_repo_path: None,
+            repo_origin_url: None,
+            base_branch: None,
+            labels: HashMap::new(),
+            provenance: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn local_handle_loads_state_and_events() {
+        let run_store = test_run_store().await;
+        let record = test_run_record();
+        append_event(
+            &run_store,
+            &fixtures::RUN_1,
+            &Event::RunCreated {
+                run_id: fixtures::RUN_1,
+                settings: serde_json::to_value(&record.settings).unwrap(),
+                graph: serde_json::to_value(&record.graph).unwrap(),
+                workflow_source: Some("digraph test {}".to_string()),
+                workflow_config: None,
+                labels: std::collections::BTreeMap::new(),
+                run_dir: "/tmp/test".to_string(),
+                working_directory: "/tmp/test".to_string(),
+                host_repo_path: None,
+                repo_origin_url: None,
+                base_branch: None,
+                workflow_slug: Some("test".to_string()),
+                db_prefix: None,
+                provenance: None,
+            },
+        )
+        .await
+        .unwrap();
+
+        let handle = RunStoreHandle::local(run_store);
+        let state = handle.state().await.unwrap();
+        let events = handle.list_events().await.unwrap();
+
+        assert_eq!(state.run.unwrap().workflow_slug.as_deref(), Some("test"));
+        assert_eq!(events.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn local_handle_appends_events_and_roundtrips_blobs() {
+        let run_store = test_run_store().await;
+        let handle = RunStoreHandle::local(run_store);
+
+        let event = RunEvent {
+            id: "evt-run-submitted".to_string(),
+            ts: Utc::now(),
+            run_id: fixtures::RUN_1,
+            node_id: None,
+            node_label: None,
+            session_id: None,
+            parent_session_id: None,
+            body: EventBody::RunSubmitted(RunStatusTransitionProps { reason: None }),
+        };
+        handle.append_run_event(&event).await.unwrap();
+
+        let blob_id = handle.write_blob(br#"{"ok":true}"#).await.unwrap();
+        let blob = handle.read_blob(&blob_id).await.unwrap().unwrap();
+        let events = handle.list_events().await.unwrap();
+
+        assert_eq!(events.len(), 1);
+        assert_eq!(blob.as_ref(), br#"{"ok":true}"#);
+    }
+}
--- a/lib/crates/fabro-workflow/src/test_support.rs
+++ b/lib/crates/fabro-workflow/src/test_support.rs
@ -106,7 +106,7 @@ async fn initialized(
            run_options: run_options.clone(),
            workflow_path: None,
            workflow_bundle: None,
-            run_store,
+            run_store: run_store.into(),
            checkpoint: options.checkpoint,
            seed_context: None,
            emitter,
--- a/lib/packages/fabro-api-client/src/api/run-internals-api.ts
+++ b/lib/packages/fabro-api-client/src/api/run-internals-api.ts
@ -97,7 +97,7 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
            };
        },
        /**
-         * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
+         * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
         * @summary Attach Run Events
         * @param {string} id Unique run identifier (ULID).
         * @param {number} [sinceSeq] First event sequence number to include.
@ -732,7 +732,7 @@ export const RunInternalsApiFp = function(configuration?: Configuration) {
            return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
        },
        /**
-         * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
+         * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
         * @summary Attach Run Events
         * @param {string} id Unique run identifier (ULID).
         * @param {number} [sinceSeq] First event sequence number to include.
@ -937,7 +937,7 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b
            return localVarFp.appendRunEvent(id, runEvent, options).then((request) => request(axios, basePath));
        },
        /**
-         * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
+         * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
         * @summary Attach Run Events
         * @param {string} id Unique run identifier (ULID).
         * @param {number} [sinceSeq] First event sequence number to include.
@ -1102,7 +1102,7 @@ export class RunInternalsApi extends BaseAPI {
    }

    /**
-     * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
+     * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
     * @summary Attach Run Events
     * @param {string} id Unique run identifier (ULID).
     * @param {number} [sinceSeq] First event sequence number to include.