Seal durable activity metrics

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<ChildAgentActivitySummary> {
-	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(

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();

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::<ChildAgentActivitySummary>(&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<StateData> {
 		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::<ChildAgentActivitySummary>(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)?,

apps/decodex/src/state/models.rs

@@ -1102,6 +1102,65 @@ pub(crate) struct ChildAgentActivitySummary {
 	pub(crate) largest_tool_output_bytes: Option<i64>,
 	pub(crate) largest_tool_output_tool: Option<String>,
 	pub(crate) large_output_warnings: Vec<String>,
+	}
+	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(&current_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)]

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,

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");