From 24f0d0a393466f3169174d108ed933cb96878160 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Wed, 17 Jun 2026 10:20:03 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Seal durable activity metrics","authority":"manual"} --- apps/decodex/src/orchestrator/status.rs | 27 +------ .../tests/operator/status/history.rs | 70 ++++++++++++++++++ apps/decodex/src/state/internal.rs | 48 +++++++++++- apps/decodex/src/state/models.rs | 59 +++++++++++++++ apps/decodex/src/state/store.rs | 2 +- apps/decodex/src/state/tests.rs | 73 ++++++++++++++++++- 6 files changed, 251 insertions(+), 28 deletions(-) diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index 6430614db..d54eef625 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -7332,32 +7332,13 @@ fn operator_run_child_agent_activity( stored_summary: Option<&ChildAgentActivitySummary>, now_unix_epoch: i64, ) -> Option { - let mut summary = marker - .and_then(RunActivityMarker::child_agent_activity) - .or(stored_summary) - .cloned()?; - - summary.current_elapsed_seconds = - summary.current_started_unix_epoch.and_then(|started_at| { - now_unix_epoch.checked_sub(started_at).filter(|elapsed| *elapsed >= 0) - }); - - if let (Some(current_bucket), Some(current_elapsed_seconds)) = - (summary.current_bucket.as_deref(), summary.current_elapsed_seconds) - && current_elapsed_seconds > 0 + if let Some(marker) = marker + && let Some(summary) = marker.child_agent_activity() { - if let Some(bucket) = summary.buckets.iter_mut().find(|bucket| bucket.name == current_bucket) { - bucket.wall_seconds = bucket.wall_seconds.saturating_add(current_elapsed_seconds); - } else { - summary.buckets.push(ChildAgentActivityBucket { - name: current_bucket.to_owned(), - wall_seconds: current_elapsed_seconds, - ..ChildAgentActivityBucket::default() - }); - } + return Some(summary.clone().live_projection(now_unix_epoch)); } - Some(summary) + stored_summary.cloned().map(ChildAgentActivitySummary::sealed_durable) } fn operator_run_protocol_activity( diff --git a/apps/decodex/src/orchestrator/tests/operator/status/history.rs b/apps/decodex/src/orchestrator/tests/operator/status/history.rs index 0e4ac3cfa..1b0e742e5 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/history.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/history.rs @@ -95,6 +95,17 @@ fn history_lane_child_activity( } } +fn unsealed_history_lane_child_activity() -> ChildAgentActivitySummary { + let mut activity = history_lane_child_activity(12, 3, 1, 100, 30); + + activity.current_bucket = Some(String::from("Model")); + activity.current_detail = Some(String::from("model output")); + activity.current_started_unix_epoch = Some(1); + activity.current_elapsed_seconds = Some(11); + + activity +} + fn seed_grouped_history_lane_lifecycle_metrics(state_store: &StateStore, issue_id: &str) { let first_activity = history_lane_child_activity(10, 2, 1, 100, 30); let second_activity = history_lane_child_activity(20, 3, 4, 200, 40); @@ -255,6 +266,65 @@ fn operator_status_history_lanes_group_attempts_by_issue() { )); } +#[test] +fn operator_history_lifecycle_metrics_use_sealed_durable_activity() { + let (_temp_dir, config, _workflow) = temp_project_layout(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let issue = sample_issue_with_sort_fields( + "issue-324", + "XY-324", + "Done", + &[], + Some(3), + "2026-03-13T04:18:17.133Z", + ); + + state_store + .record_run_attempt("xy-324-attempt-1-1777361523", &issue.id, 1, "failed") + .expect("failed run should record"); + state_store + .upsert_worktree( + "pubfi", + &issue.id, + "x/decodex-xy-324", + &config.worktree_root().join(&issue.identifier).display().to_string(), + ) + .expect("worktree should record"); + state_store + .record_run_activity_summary( + "xy-324-attempt-1-1777361523", + 1, + Some(&unsealed_history_lane_child_activity()), + None, + ) + .expect("activity summary should record"); + + let snapshot = orchestrator::build_operator_status_snapshot(&config, &state_store, 10) + .expect("snapshot should build"); + let grouped_lane = snapshot + .history_lanes + .iter() + .find(|lane| lane.issue_key == "XY-324") + .expect("history lane should exist"); + let activity = grouped_lane + .latest_run + .child_agent_activity + .as_ref() + .expect("sealed activity should remain available"); + let model_bucket = activity + .buckets + .iter() + .find(|bucket| bucket.name == "Model") + .expect("model bucket should remain available"); + + assert_eq!(activity.current_bucket, None); + assert_eq!(activity.current_started_unix_epoch, None); + assert_eq!(activity.current_elapsed_seconds, None); + assert_eq!(model_bucket.wall_seconds, 12); + assert_eq!(grouped_lane.lifecycle_metrics.wall_seconds, 12); + assert_eq!(grouped_lane.lifecycle_metrics.buckets[0].wall_seconds, 12); +} + #[test] fn operator_status_project_waiting_count_ignores_superseded_waiting_attempts() { let (_temp_dir, config, _workflow) = temp_project_layout(); diff --git a/apps/decodex/src/state/internal.rs b/apps/decodex/src/state/internal.rs index 69b8cd7b1..5028e4769 100644 --- a/apps/decodex/src/state/internal.rs +++ b/apps/decodex/src/state/internal.rs @@ -398,6 +398,7 @@ ON linear_execution_events (service_id, issue_id, event_unix, recorded_at_unix); self.bootstrap_program_intake_state_schema()?; self.bootstrap_loop_guardrail_schema()?; self.record_schema_version()?; + self.seal_run_activity_summary_records()?; Ok(()) } @@ -775,6 +776,42 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + fn seal_run_activity_summary_records(&self) -> Result<()> { + let updates = { + let mut statement = self.connection.prepare( + "SELECT run_id, child_agent_activity_json FROM run_activity_summaries \ + WHERE child_agent_activity_json IS NOT NULL", + )?; + let rows = statement.query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })?; + let mut updates = Vec::new(); + + for row in rows { + let (run_id, child_agent_activity_json) = row?; + let sealed_json = serde_json::to_string( + &serde_json::from_str::(&child_agent_activity_json)? + .sealed_durable(), + )?; + + if sealed_json != child_agent_activity_json { + updates.push((run_id, sealed_json)); + } + } + + updates + }; + + for (run_id, child_agent_activity_json) in updates { + self.connection.execute( + "UPDATE run_activity_summaries SET child_agent_activity_json = ?2 WHERE run_id = ?1", + params![run_id, child_agent_activity_json], + )?; + } + + Ok(()) + } + fn load_state(&self) -> Result { let mut state = StateData::default(); @@ -992,7 +1029,9 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; let child_agent_activity_json = summary .child_agent_activity .as_ref() - .map(serde_json::to_string) + .cloned() + .map(ChildAgentActivitySummary::sealed_durable) + .map(|summary| serde_json::to_string(&summary)) .transpose()?; let protocol_activity_json = summary .protocol_activity @@ -3957,7 +3996,9 @@ fn persist_run_activity_summaries( let child_agent_activity_json = summary .child_agent_activity .as_ref() - .map(serde_json::to_string) + .cloned() + .map(ChildAgentActivitySummary::sealed_durable) + .map(|summary| serde_json::to_string(&summary)) .transpose()?; let protocol_activity_json = summary .protocol_activity @@ -5045,7 +5086,8 @@ fn run_activity_summary_record_from_row( Ok(RunActivitySummaryRecord { run_id: row.get(0)?, attempt_number: row.get(1)?, - child_agent_activity: optional_json_from_row(row, 2)?, + child_agent_activity: optional_json_from_row::(row, 2)? + .map(ChildAgentActivitySummary::sealed_durable), protocol_activity: optional_json_from_row(row, 3)?, updated_at: row.get(4)?, updated_at_unix: row.get(5)?, diff --git a/apps/decodex/src/state/models.rs b/apps/decodex/src/state/models.rs index a7e7f3fe8..5c7432257 100644 --- a/apps/decodex/src/state/models.rs +++ b/apps/decodex/src/state/models.rs @@ -1102,6 +1102,65 @@ pub(crate) struct ChildAgentActivitySummary { pub(crate) largest_tool_output_bytes: Option, pub(crate) largest_tool_output_tool: Option, pub(crate) large_output_warnings: Vec, + } + impl ChildAgentActivitySummary { + pub(crate) fn sealed_durable(mut self) -> Self { + self.seal_open_interval(); + + self + } + + pub(crate) fn live_projection(mut self, now_unix_epoch: i64) -> Self { + let observed_elapsed_seconds = self + .current_elapsed_seconds + .filter(|elapsed| *elapsed >= 0) + .unwrap_or(0); + let current_elapsed_seconds = + self.current_started_unix_epoch.and_then(|started_at| { + now_unix_epoch.checked_sub(started_at).filter(|elapsed| *elapsed >= 0) + }); + let open_delta_seconds = current_elapsed_seconds.and_then(|elapsed| { + elapsed + .checked_sub(observed_elapsed_seconds) + .filter(|delta| *delta > 0) + }); + + self.current_elapsed_seconds = current_elapsed_seconds; + + let current_bucket = self.current_bucket.clone(); + + if let (Some(current_bucket), Some(open_delta_seconds)) = + (current_bucket, open_delta_seconds) + { + let bucket = self.bucket_mut(¤t_bucket); + + bucket.wall_seconds = bucket.wall_seconds.saturating_add(open_delta_seconds); + } + + self + } + + fn seal_open_interval(&mut self) { + self.current_bucket = None; + self.current_detail = None; + self.current_started_unix_epoch = None; + self.current_elapsed_seconds = None; + } + + fn bucket_mut(&mut self, name: &str) -> &mut ChildAgentActivityBucket { + if let Some(index) = self.buckets.iter().position(|bucket| bucket.name == name) { + return &mut self.buckets[index]; + } + + self.buckets.push(ChildAgentActivityBucket { + name: name.to_owned(), + ..ChildAgentActivityBucket::default() + }); + + let last_index = self.buckets.len().saturating_sub(1); + + &mut self.buckets[last_index] + } } #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] diff --git a/apps/decodex/src/state/store.rs b/apps/decodex/src/state/store.rs index a66a76254..68bf04559 100644 --- a/apps/decodex/src/state/store.rs +++ b/apps/decodex/src/state/store.rs @@ -1597,7 +1597,7 @@ impl StateStore { let summary = RunActivitySummaryRecord { run_id: run_id.to_owned(), attempt_number, - child_agent_activity: child_agent_activity.cloned(), + child_agent_activity: child_agent_activity.cloned().map(ChildAgentActivitySummary::sealed_durable), protocol_activity: protocol_activity.cloned(), updated_at: now.text, updated_at_unix: now.unix, diff --git a/apps/decodex/src/state/tests.rs b/apps/decodex/src/state/tests.rs index 2fbe7153a..a28688b53 100644 --- a/apps/decodex/src/state/tests.rs +++ b/apps/decodex/src/state/tests.rs @@ -1677,6 +1677,7 @@ fn records_run_activity_summary_for_recent_project_runs() { turn_status: Some(String::from("completed")), ..ProtocolActivitySummary::default() }; + let persisted_child_activity = child_activity.clone().sealed_durable(); { let store = StateStore::open(&state_path).expect("persistent state store should open"); @@ -1702,10 +1703,80 @@ fn records_run_activity_summary_for_recent_project_runs() { assert_eq!(runs.len(), 1); assert_eq!(runs[0].run_id(), "run-1"); - assert_eq!(runs[0].child_agent_activity(), Some(&child_activity)); + assert_eq!(runs[0].child_agent_activity(), Some(&persisted_child_activity)); assert_eq!(runs[0].protocol_activity(), Some(&protocol_activity)); } +#[test] +fn opening_state_store_seals_durable_run_activity_summary_rows() { + let temp_dir = TempDir::new().expect("tempdir should create"); + let state_path = temp_dir.path().join("runtime.sqlite3"); + let child_activity = ChildAgentActivitySummary { + buckets: vec![ChildAgentActivityBucket { + name: String::from("Model"), + wall_seconds: 12, + event_count: 3, + tool_call_count: 0, + input_tokens: 1_200, + output_tokens: 240, + output_bytes: 0, + }], + current_bucket: Some(String::from("Model")), + current_detail: Some(String::from("gpt-5")), + current_started_unix_epoch: Some(10), + current_elapsed_seconds: Some(8), + wall_seconds: 12, + event_count: 3, + tool_call_count: 2, + input_tokens_current: Some(1_200), + input_tokens_max: Some(1_200), + input_tokens_cumulative: 1_200, + output_tokens_cumulative: 240, + largest_tool_output_bytes: Some(4_096), + largest_tool_output_tool: Some(String::from("shell")), + large_output_warnings: vec![String::from("shell output was truncated")], + }; + let unsealed_json = + serde_json::to_string(&child_activity).expect("unsealed activity should serialize"); + + StateStore::open(&state_path).expect("persistent state store should bootstrap"); + + { + let connection = Connection::open(&state_path).expect("sqlite connection should reopen"); + + connection + .execute( + "INSERT INTO run_activity_summaries ( + run_id, attempt_number, child_agent_activity_json, protocol_activity_json, + updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, NULL, ?4, ?5)", + rusqlite::params!["run-old", 1_i64, unsealed_json, "2026-06-17T00:00:00Z", 1_i64], + ) + .expect("unsealed activity row should insert"); + } + + StateStore::open(&state_path).expect("persistent state store should seal stored row"); + + let sealed_json: String = Connection::open(&state_path) + .expect("sqlite connection should reopen") + .query_row( + "SELECT child_agent_activity_json FROM run_activity_summaries WHERE run_id = ?1", + ["run-old"], + |row| row.get(0), + ) + .expect("sealed row should load"); + let sealed_value: Value = + serde_json::from_str(&sealed_json).expect("sealed activity should remain json"); + let sealed_activity: ChildAgentActivitySummary = + serde_json::from_str(&sealed_json).expect("sealed activity should deserialize"); + + assert!(sealed_value["current_bucket"].is_null()); + assert!(sealed_value["current_detail"].is_null()); + assert!(sealed_value["current_started_unix_epoch"].is_null()); + assert!(sealed_value["current_elapsed_seconds"].is_null()); + assert_eq!(sealed_activity, child_activity.sealed_durable()); +} + #[test] fn lists_issue_attempts_and_protocol_event_presence() { let store = StateStore::open_in_memory().expect("in-memory state store should open");