Support historical run events during migration

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

@@ -50,6 +50,31 @@ impl RunProjectionReducer for RunProjection {
         };
         let mut state = projection_from_created(first)?;
         for event in rest {
+            // Runs written before the runnable state was introduced can move
+            // directly from submitted to starting. Replay that historical
+            // shape through the equivalent current transition while keeping
+            // live single-event transitions strict.
+            if matches!(event.event.body, EventBody::RunStarting(_))
+                && matches!(state.status, RunStatus::Submitted)
+            {
+                state.try_apply_status(RunStatus::Runnable, event.event.ts)?;
+            }
+            if let EventBody::RunFailed(props) = &event.event.body {
+                let failed = RunStatus::Failed {
+                    reason: props.failure.reason,
+                };
+                if !state.status.can_transition_to(failed) {
+                    if matches!(
+                        state.status,
+                        RunStatus::Submitted | RunStatus::Pending { .. }
+                    ) {
+                        state.try_apply_status(RunStatus::Runnable, event.event.ts)?;
+                    }
+                    if matches!(state.status, RunStatus::Runnable) {
+                        state.try_apply_status(RunStatus::Starting, event.event.ts)?;
+                    }
+                }
+            }
             state.apply_event(event)?;
         }
         Ok(state)
@@ -2664,6 +2689,71 @@ mod tests {
         }
     }
 
+    fn historical_created_event() -> EventEnvelope {
+        test_raw_event(
+            1,
+            "run.created",
+            &json!({
+                "settings": WorkflowSettings::default(),
+                "graph": Graph::new("historical"),
+                "labels": {},
+                "provenance": test_support::test_run_provenance()
+            }),
+            None,
+        )
+    }
+
+    #[test]
+    fn historical_submitted_to_starting_transition_replays() {
+        let state = RunProjection::apply_events(&[
+            historical_created_event(),
+            test_raw_event(2, "run.submitted", &json!({}), None),
+            test_raw_event(3, "run.starting", &json!({}), None),
+        ])
+        .unwrap();
+
+        assert_eq!(state.status, RunStatus::Starting);
+    }
+
+    #[test]
+    fn historical_runnable_to_terminated_transition_replays() {
+        let state = RunProjection::apply_events(&[
+            historical_created_event(),
+            test_raw_event(2, "run.submitted", &json!({}), None),
+            test_raw_event(
+                3,
+                "run.runnable",
+                &json!({ "source": "start_requested" }),
+                None,
+            ),
+            test_raw_event(
+                4,
+                "run.failed",
+                &json!({
+                    "failure": {
+                        "reason": "terminated",
+                        "detail": {
+                            "message": "worker stopped before startup",
+                            "category": "deterministic"
+                        }
+                    },
+                    "timing": {
+                        "wall_time_ms": 1,
+                        "inference_time_ms": 0,
+                        "tool_time_ms": 0,
+                        "active_time_ms": 0
+                    }
+                }),
+                None,
+            ),
+        ])
+        .unwrap();
+
+        assert_eq!(state.status, RunStatus::Failed {
+            reason: FailureReason::Terminated,
+        });
+    }
+
     #[test]
     fn live_run_timing_returns_none_before_run_starts() {
         let state = initialized_projection();

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

@@ -1621,13 +1621,7 @@ mod tests {
             .add(&bad_run_id)
             .await
             .unwrap();
-        let mut run_spec = serde_json::to_value(sample_run_spec("run-2")).unwrap();
-        let run_settings = run_spec
-            .get_mut("settings")
-            .and_then(|settings| settings.get_mut("run"))
-            .and_then(serde_json::Value::as_object_mut)
-            .unwrap();
-        run_settings.remove("integrations");
+        let run_spec = serde_json::to_value(sample_run_spec("run-2")).unwrap();
         store
             .put_unvalidated_run_event(
                 &bad_run_id,
@@ -1658,8 +1652,8 @@ mod tests {
         assert_eq!(unreadable[0].run_id, bad_run_id);
         assert_eq!(unreadable[0].created_at, bad_run_id.created_at());
         assert!(
-            unreadable[0].error.contains("missing field `integrations`"),
-            "expected missing integrations error, got: {}",
+            unreadable[0].error.contains("missing field `provenance`"),
+            "expected missing provenance error, got: {}",
             unreadable[0].error
         );
     }

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

@@ -20,7 +20,7 @@ pub use session::*;
 pub use stage::*;
 pub use todo::*;
 
-use crate::{ParallelBranchId, Principal, RunId, StageId};
+use crate::{ParallelBranchId, Principal, RunId, StageId, UsdMicros};
 
 /// Maximum accepted body size for `POST /runs/{id}/events`.
 ///
@@ -759,7 +759,8 @@ fn is_known_event_name(event: &str) -> bool {
 }
 
 impl RunEvent {
-    pub fn from_value(value: Value) -> serde_json::Result<Self> {
+    pub fn from_value(mut value: Value) -> serde_json::Result<Self> {
+        normalize_legacy_event(&mut value);
         let raw: RunEventRaw = serde_json::from_value(value)?;
         Self::from_parts(RunEventParts {
             id:                 raw.id,
@@ -808,10 +809,11 @@ impl RunEvent {
         let event = obj.get("event").and_then(Value::as_str).ok_or_else(|| {
             <serde_json::Error as DeError>::custom("missing or non-string field: event")
         })?;
-        let properties = obj
+        let mut properties = obj
             .get("properties")
             .cloned()
             .unwrap_or_else(default_properties);
+        normalize_legacy_event_properties(event, &mut properties);
         Self::from_parts(RunEventParts {
             id: id.to_string(),
             ts,
@@ -918,6 +920,291 @@ impl RunEvent {
     }
 }
 
+/// Upgrades historical wire shapes only in the value being decoded. Legacy
+/// importers still retain and compare the original stored JSON.
+fn normalize_legacy_event(value: &mut Value) {
+    let Some(event) = value
+        .get("event")
+        .and_then(Value::as_str)
+        .map(str::to_owned)
+    else {
+        return;
+    };
+    let Some(properties) = value.get_mut("properties") else {
+        return;
+    };
+    normalize_legacy_event_properties(&event, properties);
+}
+
+fn normalize_legacy_event_properties(event: &str, properties: &mut Value) {
+    let Some(object) = properties.as_object_mut() else {
+        return;
+    };
+    match event {
+        "agent.message" => normalize_legacy_agent_message(object),
+        "run.completed" => normalize_legacy_timing(object, false),
+        "run.failed" => {
+            normalize_legacy_run_failure(object);
+            normalize_legacy_timing(object, false);
+        }
+        "stage.completed" => {
+            normalize_legacy_usage_field(object);
+            normalize_legacy_billing_field(object, "billing");
+            normalize_legacy_timing(object, true);
+        }
+        "stage.failed" => normalize_legacy_billing_field(object, "billing"),
+        "prompt.completed" => {
+            normalize_legacy_usage_field(object);
+            normalize_legacy_billing_field(object, "billing");
+        }
+        "checkpoint.completed" => normalize_legacy_checkpoint_billing(object),
+        "sandbox.initialized" => normalize_legacy_sandbox_id(object),
+        _ => {}
+    }
+}
+
+fn normalize_legacy_agent_message(properties: &mut Map<String, Value>) {
+    let speed = properties
+        .get("usage")
+        .and_then(Value::as_object)
+        .and_then(|usage| usage.get("speed"))
+        .and_then(Value::as_str)
+        .map(str::to_owned);
+    if let Some(model_id) = properties.get("model").and_then(Value::as_str) {
+        let mut model = Map::from_iter([
+            (
+                "provider".to_owned(),
+                Value::String(legacy_provider_for_model(model_id).to_owned()),
+            ),
+            ("model_id".to_owned(), Value::String(model_id.to_owned())),
+        ]);
+        if let Some(speed @ ("standard" | "fast")) = speed.as_deref() {
+            model.insert("speed".to_owned(), Value::String(speed.to_owned()));
+        }
+        properties.insert("model".to_owned(), Value::Object(model));
+    }
+    if !properties.contains_key("billing") {
+        if let Some(usage) = properties.remove("usage") {
+            properties.insert("billing".to_owned(), usage);
+        }
+    }
+}
+
+fn normalize_legacy_usage_field(properties: &mut Map<String, Value>) {
+    if !properties.contains_key("billing") {
+        if let Some(usage) = properties.remove("usage") {
+            properties.insert("billing".to_owned(), usage);
+        }
+    }
+}
+
+fn normalize_legacy_billing_field(properties: &mut Map<String, Value>, field: &str) {
+    if let Some(billing) = properties.get_mut(field) {
+        normalize_legacy_billing_values(billing);
+    }
+}
+
+fn normalize_legacy_checkpoint_billing(properties: &mut Map<String, Value>) {
+    let Some(outcomes) = properties
+        .get_mut("node_outcomes")
+        .and_then(Value::as_object_mut)
+    else {
+        return;
+    };
+    for outcome in outcomes.values_mut().filter_map(Value::as_object_mut) {
+        normalize_legacy_billing_field(outcome, "usage");
+    }
+}
+
+fn normalize_legacy_timing(properties: &mut Map<String, Value>, stage: bool) {
+    if properties.contains_key("timing") {
+        return;
+    }
+    let Some(wall_time_ms) = properties.get("duration_ms").and_then(Value::as_u64) else {
+        return;
+    };
+    let timing = json!({
+        "wall_time_ms": wall_time_ms,
+        "inference_time_ms": 0,
+        "tool_time_ms": 0,
+        "active_time_ms": 0,
+    });
+    properties.insert("timing".to_owned(), timing);
+    if stage {
+        properties.remove("duration_ms");
+    }
+}
+
+fn normalize_legacy_run_failure(properties: &mut Map<String, Value>) {
+    if !properties.contains_key("failure") {
+        if let (Some(message), Some(reason)) = (
+            properties.get("error").cloned(),
+            properties.get("reason").cloned(),
+        ) {
+            let causes = properties
+                .get("causes")
+                .cloned()
+                .unwrap_or_else(|| Value::Array(Vec::new()));
+            properties.insert(
+                "failure".to_owned(),
+                json!({
+                    "reason": reason,
+                    "detail": {
+                        "message": message,
+                        "causes": causes,
+                        "category": "deterministic",
+                    }
+                }),
+            );
+        }
+    }
+    if !properties.contains_key("final_git_commit_sha") {
+        if let Some(commit) = properties.get("git_commit_sha").cloned() {
+            properties.insert("final_git_commit_sha".to_owned(), commit);
+        }
+    }
+}
+
+fn normalize_legacy_sandbox_id(properties: &mut Map<String, Value>) {
+    if properties.contains_key("id") {
+        return;
+    }
+    let id = properties
+        .get("identifier")
+        .and_then(Value::as_str)
+        .unwrap_or_default();
+    properties.insert("id".to_owned(), Value::String(id.to_owned()));
+}
+
+fn normalize_legacy_billing_values(value: &mut Value) {
+    if legacy_stage_usage(value) {
+        let legacy = std::mem::take(value);
+        *value = normalized_legacy_stage_usage(&legacy);
+        return;
+    }
+    match value {
+        Value::Array(values) => {
+            for value in values {
+                normalize_legacy_billing_values(value);
+            }
+        }
+        Value::Object(object) => {
+            if let Some(facts) = object.get_mut("facts").and_then(Value::as_object_mut) {
+                if !facts.contains_key("algorithm") {
+                    let provider = facts
+                        .get("provider")
+                        .and_then(Value::as_str)
+                        .map(str::to_owned);
+                    if let Some(provider) = provider {
+                        facts.remove("provider");
+                        facts.insert(
+                            "algorithm".to_owned(),
+                            Value::String(legacy_billing_algorithm(&provider).to_owned()),
+                        );
+                    }
+                }
+            }
+            for value in object.values_mut() {
+                normalize_legacy_billing_values(value);
+            }
+        }
+        _ => {}
+    }
+}
+
+fn legacy_stage_usage(value: &Value) -> bool {
+    let Some(object) = value.as_object() else {
+        return false;
+    };
+    object.get("model").is_some_and(Value::is_string)
+        && object.get("input_tokens").is_some_and(Value::is_number)
+        && object.get("output_tokens").is_some_and(Value::is_number)
+}
+
+fn normalized_legacy_stage_usage(legacy: &Value) -> Value {
+    let object = legacy
+        .as_object()
+        .expect("legacy stage usage was validated as an object");
+    let model_id = object
+        .get("model")
+        .and_then(Value::as_str)
+        .expect("legacy stage usage was validated with a string model");
+    let provider = legacy_provider_for_model(model_id);
+    let mut model = json!({
+        "provider": provider,
+        "model_id": model_id,
+    });
+    if let Some(speed @ ("standard" | "fast")) = object.get("speed").and_then(Value::as_str) {
+        model["speed"] = Value::String(speed.to_owned());
+    }
+    let input_tokens = object
+        .get("input_tokens")
+        .and_then(Value::as_i64)
+        .unwrap_or(0);
+    let output_tokens = object
+        .get("output_tokens")
+        .and_then(Value::as_i64)
+        .unwrap_or(0);
+    let reasoning_tokens = object
+        .get("reasoning_tokens")
+        .and_then(Value::as_i64)
+        .unwrap_or(0);
+    let cache_read_tokens = object
+        .get("cache_read_tokens")
+        .and_then(Value::as_i64)
+        .unwrap_or(0);
+    let cache_write_tokens = object
+        .get("cache_write_tokens")
+        .and_then(Value::as_i64)
+        .unwrap_or(0);
+    let mut normalized = json!({
+        "input": {
+            "usage": {
+                "model": model,
+                "tokens": {
+                    "input_tokens": input_tokens,
+                    "output_tokens": output_tokens,
+                    "reasoning_tokens": reasoning_tokens,
+                    "cache_read_tokens": cache_read_tokens,
+                    "cache_write_tokens": cache_write_tokens,
+                }
+            },
+            "facts": {
+                "algorithm": legacy_billing_algorithm(provider),
+            }
+        }
+    });
+    if let Some(cost) = object.get("cost").and_then(Value::as_f64) {
+        normalized["total_usd_micros"] = Value::from(UsdMicros::from_usd(cost).0);
+    }
+    normalized
+}
+
+fn legacy_provider_for_model(model_id: &str) -> &'static str {
+    if model_id.starts_with("claude-") {
+        "anthropic"
+    } else if model_id.starts_with("gemini-") {
+        "gemini"
+    } else if model_id.starts_with("gpt-")
+        || model_id.starts_with("chatgpt-")
+        || model_id.starts_with("o1")
+        || model_id.starts_with("o3")
+        || model_id.starts_with("o4")
+    {
+        "openai"
+    } else {
+        "legacy"
+    }
+}
+
+fn legacy_billing_algorithm(provider: &str) -> &'static str {
+    match provider {
+        "anthropic" => "anthropic",
+        "gemini" => "gemini",
+        _ => "openai",
+    }
+}
+
 impl Serialize for RunEvent {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
@@ -957,6 +1244,16 @@ mod tests {
         )
     }
 
+    fn stored_event(event: &str, properties: &Value) -> Value {
+        json!({
+            "id": format!("evt_{event}"),
+            "ts": "2026-04-04T12:00:00.000Z",
+            "run_id": fixtures::RUN_1,
+            "event": event,
+            "properties": properties,
+        })
+    }
+
     #[test]
     fn run_event_round_trips_json() {
         let event = RunEvent {
@@ -1054,6 +1351,212 @@ mod tests {
         assert_eq!(props.effective_reasoning_effort, None);
     }
 
+    #[test]
+    fn historical_run_created_defaults_new_run_settings() {
+        let mut settings = serde_json::to_value(WorkflowSettings::default()).unwrap();
+        let run = settings["run"].as_object_mut().unwrap();
+        for field in ["clone", "run_branch", "integrations"] {
+            run.remove(field);
+        }
+        let line = stored_event(
+            "run.created",
+            &json!({
+                "settings": settings,
+                "graph": Graph::new("test"),
+                "labels": {},
+                "source_directory": "/tmp/run",
+                "provenance": test_support::test_run_provenance()
+            }),
+        );
+
+        let parsed = RunEvent::from_value(line).unwrap();
+        let EventBody::RunCreated(props) = parsed.body else {
+            panic!("expected run.created");
+        };
+
+        assert_eq!(props.settings.run, WorkflowSettings::default().run);
+    }
+
+    #[test]
+    fn historical_agent_message_accepts_string_model() {
+        let line = stored_event(
+            "agent.message",
+            &json!({
+                "text": "done",
+                "model": "gemini-3.1-pro-preview",
+                "billing": {
+                    "input_tokens": 10,
+                    "output_tokens": 5,
+                    "total_tokens": 15
+                },
+                "tool_call_count": 0,
+                "visit": 1
+            }),
+        );
+
+        let parsed = RunEvent::from_value(line).unwrap();
+        let normalized = parsed.to_value().unwrap();
+
+        assert_eq!(normalized["properties"]["model"]["provider"], "gemini");
+        assert_eq!(
+            normalized["properties"]["model"]["model_id"],
+            "gemini-3.1-pro-preview"
+        );
+    }
+
+    #[test]
+    fn historical_stage_usage_and_duration_are_upgraded() {
+        let line = stored_event(
+            "stage.completed",
+            &json!({
+                "index": 0,
+                "duration_ms": 42,
+                "status": "succeeded",
+                "usage": {
+                    "model": "claude-sonnet-4-6",
+                    "input_tokens": 100,
+                    "output_tokens": 20,
+                    "cache_read_tokens": 7,
+                    "cache_write_tokens": 3,
+                    "reasoning_tokens": 2,
+                    "speed": "fast",
+                    "cost": 0.012_345
+                },
+                "attempt": 1,
+                "max_attempts": 1
+            }),
+        );
+
+        let parsed = RunEvent::from_value(line).unwrap();
+        let normalized = parsed.to_value().unwrap();
+        let properties = &normalized["properties"];
+
+        assert_eq!(properties["timing"]["wall_time_ms"], 42);
+        assert_eq!(
+            properties["billing"]["input"]["facts"]["algorithm"],
+            "anthropic"
+        );
+        assert_eq!(
+            properties["billing"]["input"]["usage"]["model"]["speed"],
+            "fast"
+        );
+        assert_eq!(properties["billing"]["total_usd_micros"], 12_345);
+        assert!(properties.get("duration_ms").is_none());
+        assert!(properties.get("usage").is_none());
+    }
+
+    #[test]
+    fn historical_billing_provider_tags_are_upgraded() {
+        let legacy_billing = json!({
+            "input": {
+                "usage": {
+                    "model": {
+                        "provider": "anthropic",
+                        "model_id": "claude-sonnet-4-6"
+                    },
+                    "tokens": {
+                        "input_tokens": 100,
+                        "output_tokens": 20,
+                        "reasoning_tokens": 0,
+                        "cache_read_tokens": 7,
+                        "cache_write_tokens": 3
+                    }
+                },
+                "facts": {
+                    "provider": "anthropic",
+                    "cache_write_5m_tokens": 3,
+                    "cache_write_1h_tokens": 0
+                }
+            },
+            "total_usd_micros": 123
+        });
+        let prompt = stored_event(
+            "prompt.completed",
+            &json!({
+                "response": "done",
+                "model": "claude-sonnet-4-6",
+                "provider": "anthropic",
+                "billing": legacy_billing.clone()
+            }),
+        );
+        let checkpoint = stored_event(
+            "checkpoint.completed",
+            &json!({
+                "status": "succeeded",
+                "current_node": "build",
+                "node_outcomes": {
+                    "build": {
+                        "status": "succeeded",
+                        "usage": legacy_billing
+                    }
+                }
+            }),
+        );
+
+        let prompt = RunEvent::from_value(prompt).unwrap().to_value().unwrap();
+        let checkpoint = RunEvent::from_value(checkpoint)
+            .unwrap()
+            .to_value()
+            .unwrap();
+
+        assert_eq!(
+            prompt["properties"]["billing"]["input"]["facts"]["algorithm"],
+            "anthropic"
+        );
+        assert_eq!(
+            checkpoint["properties"]["node_outcomes"]["build"]["usage"]["input"]["facts"]["algorithm"],
+            "anthropic"
+        );
+    }
+
+    #[test]
+    fn historical_terminal_and_sandbox_events_are_upgraded() {
+        let completed = stored_event(
+            "run.completed",
+            &json!({
+                "duration_ms": 123,
+                "artifact_count": 0,
+                "status": "succeeded",
+                "reason": "completed"
+            }),
+        );
+        let failed = stored_event(
+            "run.failed",
+            &json!({
+                "error": "cancelled by user",
+                "causes": ["interrupt requested"],
+                "duration_ms": 456,
+                "reason": "cancelled",
+                "git_commit_sha": "abc123"
+            }),
+        );
+        let sandbox = stored_event(
+            "sandbox.initialized",
+            &json!({
+                "provider": "local",
+                "working_directory": "/tmp/run"
+            }),
+        );
+
+        let completed = RunEvent::from_value(completed).unwrap().to_value().unwrap();
+        let failed = RunEvent::from_value(failed).unwrap().to_value().unwrap();
+        let sandbox = RunEvent::from_value(sandbox).unwrap().to_value().unwrap();
+
+        assert_eq!(completed["properties"]["timing"]["wall_time_ms"], 123);
+        assert_eq!(failed["properties"]["failure"]["reason"], "cancelled");
+        assert_eq!(
+            failed["properties"]["failure"]["detail"]["message"],
+            "cancelled by user"
+        );
+        assert_eq!(
+            failed["properties"]["failure"]["detail"]["causes"],
+            json!(["interrupt requested"])
+        );
+        assert_eq!(failed["properties"]["timing"]["wall_time_ms"], 456);
+        assert_eq!(failed["properties"]["final_git_commit_sha"], "abc123");
+        assert_eq!(sandbox["properties"]["id"], "");
+    }
+
     #[test]
     fn run_created_round_trip_preserves_manifest_blob() {
         let line = json!({

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

@@ -22,6 +22,7 @@ use super::size::Size;
 
 /// A structurally resolved `[run]` view for consumers.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(default)]
 pub struct RunNamespace {
     pub goal:          Option<RunGoal>,
     pub working_dir:   Option<String>,