From 0489a08557b7608c511ef539c4782d161b7aff7b Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Thu, 11 Jun 2026 14:32:42 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Reuse operator snapshots for status","authority":"XY-913"} --- README.md | 16 +- apps/decodex/src/cli.rs | 2 +- apps/decodex/src/orchestrator.rs | 5 + .../src/orchestrator/agent_evidence.rs | 2 +- apps/decodex/src/orchestrator/entrypoints.rs | 311 +++++++++++++++++- apps/decodex/src/orchestrator/status.rs | 24 +- .../tests/operator/status/agent_evidence.rs | 2 + .../tests/operator/status/control_plane.rs | 160 +++++++++ .../tests/operator/status/text.rs | 18 + .../tests/operator/status_support.rs | 31 +- apps/decodex/src/orchestrator/types.rs | 55 ++-- docs/reference/operator-control-plane.md | 11 + 12 files changed, 590 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index ac45531a8..b70016f5f 100644 --- a/README.md +++ b/README.md @@ -129,11 +129,17 @@ cargo run -p decodex --bin decodex -- serve --listen-address 127.0.0.1:8192 Project-scoped commands accept `--config ` after the subcommand when the operator wants to override registry-based project resolution for that command. -`decodex status` prints the local runtime snapshot without refreshing live -tracker, pull-request, or Codex account usage observers. Use `decodex status --live` -when the operator needs fresh Linear/GitHub readback before acting; use the Accounts -API refresh path, such as `GET /api/accounts?refresh=1`, when the operator needs -fresh ChatGPT account usage probes. +`decodex status` first tries to reuse the default local operator listener's published +`GET /api/operator-snapshot` when the snapshot is recent, covers the requested project, +and has at least the requested `--limit`. JSON output marks this as +`"status_source": "operator_snapshot_cache"` and includes `snapshot_age_seconds`. +If that cache is missing, stale, mismatched, or too small, the command falls back to a +direct local runtime snapshot and reports `status_cached_snapshot_unavailable` in +`warning_details`. Use `decodex status --live` when the operator needs fresh +Linear/GitHub readback before acting; `--live` bypasses the cached snapshot path and +marks JSON output as `"status_source": "live_observers"`. Use the Accounts API refresh +path, such as `GET /api/accounts?refresh=1`, when the operator needs fresh ChatGPT +account usage probes. `decodex serve` uses hardcoded scheduler cadences: the local control-plane loop publishes snapshots every 15 seconds, and Linear-backed queue/status scans run at most every 5 minutes per project unless an operator or agent requests an explicit diff --git a/apps/decodex/src/cli.rs b/apps/decodex/src/cli.rs index f46f2fe49..188cbaad9 100644 --- a/apps/decodex/src/cli.rs +++ b/apps/decodex/src/cli.rs @@ -334,7 +334,7 @@ struct ServeCommand { #[command(flatten)] project_config: ProjectConfigArgs, /// Operator UI listen address. - #[arg(long, value_name = "ADDR", default_value = "127.0.0.1:8192")] + #[arg(long, value_name = "ADDR", default_value_t = orchestrator::DEFAULT_OPERATOR_LISTEN_ADDRESS.to_owned())] listen_address: String, /// Start the local dev endpoint without polling or dispatching projects. #[arg(long, hide = true)] diff --git a/apps/decodex/src/orchestrator.rs b/apps/decodex/src/orchestrator.rs index 14c69ab51..9f4f85611 100644 --- a/apps/decodex/src/orchestrator.rs +++ b/apps/decodex/src/orchestrator.rs @@ -80,6 +80,7 @@ include!("orchestrator/agent_evidence.rs"); pub(crate) const DEFAULT_STATUS_RUN_LIMIT: usize = 10; pub(crate) const DEFAULT_OPERATOR_DASHBOARD_RUN_LIMIT: usize = 25; +pub(crate) const DEFAULT_OPERATOR_LISTEN_ADDRESS: &str = "127.0.0.1:8192"; pub(crate) const EXTERNAL_REVIEW_ACTOR_LOGIN: &str = "codex"; pub(crate) const EXTERNAL_REVIEW_REQUEST_BODY: &str = "@codex review"; pub(crate) const EXTERNAL_REVIEW_PASS_PHRASE: &str = "Didn't find any major issues."; @@ -109,6 +110,10 @@ const OPERATOR_LANE_STEER_ALIAS_ENDPOINT_PATH: &str = "/api/lane-steer"; const OPERATOR_STATE_MAX_REQUEST_BYTES: usize = 256 * 1_024; const OPERATOR_DASHBOARD_WS_CLIENT_MESSAGE_MAX_BYTES: usize = 64 * 1_024; const OPERATOR_STATE_HEADER_TERMINATOR: &[u8] = b"\r\n\r\n"; +const STATUS_OPERATOR_SNAPSHOT_MAX_AGE: Duration = Duration::from_secs(60); +const STATUS_OPERATOR_SNAPSHOT_CONNECT_TIMEOUT: Duration = Duration::from_millis(250); +const STATUS_OPERATOR_SNAPSHOT_IO_TIMEOUT: Duration = Duration::from_millis(500); +const STATUS_OPERATOR_SNAPSHOT_WARNING: &str = "status_cached_snapshot_unavailable"; const OPERATOR_DASHBOARD_WS_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(20); const OPERATOR_RUN_ACTIVITY_STREAM_INTERVAL: Duration = Duration::from_secs(1); const OPERATOR_DEV_SNAPSHOT_STREAM_INTERVAL: Duration = Duration::from_secs(1); diff --git a/apps/decodex/src/orchestrator/agent_evidence.rs b/apps/decodex/src/orchestrator/agent_evidence.rs index dfdc61850..b8a77bcf2 100644 --- a/apps/decodex/src/orchestrator/agent_evidence.rs +++ b/apps/decodex/src/orchestrator/agent_evidence.rs @@ -83,7 +83,7 @@ struct AgentConnectorBackoff { next_action: String, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct AgentPrivateEvidenceRef { evidence_ref: String, source: String, diff --git a/apps/decodex/src/orchestrator/entrypoints.rs b/apps/decodex/src/orchestrator/entrypoints.rs index 7ee15ddc4..70dbd33b1 100644 --- a/apps/decodex/src/orchestrator/entrypoints.rs +++ b/apps/decodex/src/orchestrator/entrypoints.rs @@ -19,6 +19,17 @@ struct ConnectorBackoffStatusParts<'a> { warning: &'a str, } +#[derive(Clone, Debug, Eq, PartialEq)] +struct StatusSnapshotCacheMiss { + reason: String, + next_action: String, +} + +struct StatusSnapshotHttpResponse { + body: Vec, + published_at_unix_epoch: Option, +} + impl TrackerConnectorBackoff { fn to_operator_status( &self, @@ -250,14 +261,40 @@ pub(crate) fn print_status( runtime::register_project_config(&state_store, &config_path, true)?; let snapshot = if live { - build_live_status_command_snapshot(&config, &workflow, &state_store, limit)? + let mut snapshot = build_live_status_command_snapshot(&config, &workflow, &state_store, limit)?; + + snapshot.status_source = Some(String::from("live_observers")); + + snapshot + } else if status_should_attempt_operator_snapshot_cache(live) { + match status_snapshot_from_local_operator_cache(&config, limit) { + Ok(snapshot) => snapshot, + Err(cache_miss) => { + let mut snapshot = build_operator_state_snapshot_without_live_observers( + &config, + &workflow, + &state_store, + limit, + )?; + + snapshot.status_source = Some(String::from("local_runtime")); + + add_status_snapshot_cache_miss_warning(&mut snapshot, &config, cache_miss); + + snapshot + }, + } } else { - build_operator_state_snapshot_without_live_observers( + let mut snapshot = build_operator_state_snapshot_without_live_observers( &config, &workflow, &state_store, limit, - )? + )?; + + snapshot.status_source = Some(String::from("local_runtime")); + + snapshot }; print_operator_status_snapshot(&snapshot, json)?; @@ -356,6 +393,272 @@ pub(crate) fn print_private_evidence(request: EvidenceRequest<'_>) -> Result<()> Ok(()) } +fn status_should_attempt_operator_snapshot_cache(live: bool) -> bool { + !live +} + +fn status_snapshot_from_local_operator_cache( + project: &ServiceConfig, + limit: usize, +) -> std::result::Result { + let response = fetch_local_operator_snapshot_response() + .map_err(|reason| status_snapshot_cache_miss(reason, "start or restart the local Decodex operator listener, or use `decodex status --live` for a fresh direct read"))?; + + status_snapshot_from_operator_cache_response( + project, + limit, + response, + OffsetDateTime::now_utc().unix_timestamp(), + ) +} + +fn status_snapshot_from_operator_cache_response( + project: &ServiceConfig, + limit: usize, + response: StatusSnapshotHttpResponse, + now_unix_epoch: i64, +) -> std::result::Result { + let published_at_unix_epoch = response.published_at_unix_epoch.ok_or_else(|| { + status_snapshot_cache_miss( + "operator snapshot response omitted X-Decodex-Snapshot-Unix-Epoch", + "wait for the operator listener to publish a fresh snapshot, or use `decodex status --live`", + ) + })?; + let snapshot_age_seconds = now_unix_epoch.saturating_sub(published_at_unix_epoch); + + if snapshot_age_seconds > STATUS_OPERATOR_SNAPSHOT_MAX_AGE.as_secs() as i64 { + return Err(status_snapshot_cache_miss( + format!( + "operator snapshot is stale: age {snapshot_age_seconds}s exceeds {}s", + STATUS_OPERATOR_SNAPSHOT_MAX_AGE.as_secs() + ), + "wait for the next control-plane tick or use `decodex status --live`", + )); + } + + let snapshot = serde_json::from_slice::(&response.body).map_err(|error| { + status_snapshot_cache_miss( + format!("operator snapshot JSON could not be read by this CLI: {error}"), + "install the matching Decodex CLI/app build or use `decodex status --live`", + ) + })?; + + project_status_snapshot_from_operator_cache(project, snapshot, limit, snapshot_age_seconds) +} + +fn fetch_local_operator_snapshot_response() + -> std::result::Result +{ + let address = DEFAULT_OPERATOR_LISTEN_ADDRESS + .parse::() + .map_err(|error| format!("default operator listener address is invalid: {error}"))?; + let mut stream = TcpStream::connect_timeout(&address, STATUS_OPERATOR_SNAPSHOT_CONNECT_TIMEOUT) + .map_err(|error| format!("local operator listener is unavailable at {address}: {error}"))?; + + stream + .set_read_timeout(Some(STATUS_OPERATOR_SNAPSHOT_IO_TIMEOUT)) + .map_err(|error| format!("failed to set operator snapshot read timeout: {error}"))?; + stream + .set_write_timeout(Some(STATUS_OPERATOR_SNAPSHOT_IO_TIMEOUT)) + .map_err(|error| format!("failed to set operator snapshot write timeout: {error}"))?; + stream + .write_all( + format!( + "GET {OPERATOR_APP_SNAPSHOT_ENDPOINT_PATH} HTTP/1.1\r\nHost: {DEFAULT_OPERATOR_LISTEN_ADDRESS}\r\nConnection: close\r\n\r\n" + ) + .as_bytes(), + ) + .map_err(|error| format!("failed to request local operator snapshot: {error}"))?; + + let mut response = Vec::new(); + + stream + .read_to_end(&mut response) + .map_err(|error| format!("failed to read local operator snapshot: {error}"))?; + + parse_operator_snapshot_http_response(&response) +} + +fn parse_operator_snapshot_http_response( + response: &[u8], +) -> std::result::Result { + let header_end = response + .windows(OPERATOR_STATE_HEADER_TERMINATOR.len()) + .position(|window| window == OPERATOR_STATE_HEADER_TERMINATOR) + .ok_or_else(|| String::from("local operator snapshot response omitted HTTP headers"))?; + let headers = std::str::from_utf8(&response[..header_end]) + .map_err(|error| format!("local operator snapshot headers were not UTF-8: {error}"))?; + let Some(status_line) = headers.lines().next() else { + return Err(String::from("local operator snapshot response omitted HTTP status")); + }; + + if !status_line.contains(" 200 ") { + return Err(format!("local operator snapshot request returned `{status_line}`")); + } + + let published_at_unix_epoch = headers.lines().find_map(|line| { + line.strip_prefix("X-Decodex-Snapshot-Unix-Epoch: ") + .and_then(|value| value.trim().parse::().ok()) + }); + let body = response[header_end + OPERATOR_STATE_HEADER_TERMINATOR.len()..].to_vec(); + + if body.is_empty() || body.as_slice() == b"{}" { + return Err(String::from("local operator listener has not published a status snapshot yet")); + } + + Ok(StatusSnapshotHttpResponse { body, published_at_unix_epoch }) +} + +fn project_status_snapshot_from_operator_cache( + project: &ServiceConfig, + mut snapshot: OperatorStatusSnapshot, + limit: usize, + snapshot_age_seconds: i64, +) -> std::result::Result { + if snapshot.run_limit < limit { + return Err(status_snapshot_cache_miss( + format!( + "operator snapshot run limit {} is lower than requested status limit {limit}", + snapshot.run_limit + ), + "rerun with a lower `--limit`, wait for a larger published snapshot, or use `decodex status --live`", + )); + } + if snapshot.project_id == project.service_id() { + mark_cached_status_snapshot(&mut snapshot, limit, snapshot_age_seconds); + + return Ok(snapshot); + } + + let Some(project_status) = snapshot + .projects + .iter() + .find(|status| status.project_id == project.service_id()) + .cloned() + else { + return Err(status_snapshot_cache_miss( + format!( + "operator snapshot does not include project `{}`", + project.service_id() + ), + "check that the local listener is serving the same registered project set, or use `decodex status --live`", + )); + }; + let mut project_snapshot = empty_control_plane_snapshot(limit); + + project_snapshot.project_id = project.service_id().to_owned(); + project_snapshot.status_source = Some(String::from("operator_snapshot_cache")); + project_snapshot.snapshot_age_seconds = Some(snapshot_age_seconds); + project_snapshot.account_control = snapshot.account_control; + project_snapshot.accounts = snapshot.accounts; + project_snapshot.projects = vec![project_status]; + project_snapshot.connector_backoffs = snapshot + .connector_backoffs + .into_iter() + .filter(|backoff| backoff.project_id == project.service_id()) + .collect(); + project_snapshot.active_runs = snapshot + .active_runs + .into_iter() + .filter(|run| run.project_id == project.service_id()) + .collect(); + project_snapshot.recent_runs = snapshot + .recent_runs + .into_iter() + .filter(|run| run.project_id == project.service_id()) + .collect(); + project_snapshot.history_lanes = snapshot + .history_lanes + .into_iter() + .filter(|lane| lane.project_id == project.service_id()) + .collect(); + project_snapshot.queued_candidates = snapshot + .queued_candidates + .into_iter() + .filter(|candidate| candidate.project_id == project.service_id()) + .collect(); + project_snapshot.worktrees = snapshot + .worktrees + .into_iter() + .filter(|worktree| worktree.project_id == project.service_id()) + .collect(); + project_snapshot.post_review_lanes = snapshot + .post_review_lanes + .into_iter() + .filter(|lane| lane.project_id == project.service_id()) + .collect(); + project_snapshot.warning_details = snapshot + .warning_details + .into_iter() + .filter(|detail| { + detail.project_id.as_deref().is_none() + || detail.project_id.as_deref() == Some(project.service_id()) + }) + .collect(); + project_snapshot.warnings = project_warnings_from_details(&project_snapshot.warning_details); + + truncate_status_snapshot_to_limit(&mut project_snapshot, limit); + refresh_operator_project_summary(&mut project_snapshot); + + Ok(project_snapshot) +} + +fn mark_cached_status_snapshot( + snapshot: &mut OperatorStatusSnapshot, + limit: usize, + snapshot_age_seconds: i64, +) { + snapshot.run_limit = limit; + snapshot.status_source = Some(String::from("operator_snapshot_cache")); + snapshot.snapshot_age_seconds = Some(snapshot_age_seconds); + + truncate_status_snapshot_to_limit(snapshot, limit); + refresh_operator_project_summary(snapshot); +} + +fn truncate_status_snapshot_to_limit(snapshot: &mut OperatorStatusSnapshot, limit: usize) { + snapshot.recent_runs.truncate(limit); + snapshot.history_lanes.truncate(limit); +} + +fn project_warnings_from_details(details: &[OperatorSnapshotWarningDetail]) -> Vec { + let mut warnings = Vec::new(); + + for detail in details { + if !warnings.iter().any(|warning| warning == &detail.warning) { + warnings.push(detail.warning.clone()); + } + } + + warnings +} + +fn status_snapshot_cache_miss( + reason: impl Into, + next_action: impl Into, +) -> StatusSnapshotCacheMiss { + StatusSnapshotCacheMiss { + reason: reason.into(), + next_action: next_action.into(), + } +} + +fn add_status_snapshot_cache_miss_warning( + snapshot: &mut OperatorStatusSnapshot, + project: &ServiceConfig, + cache_miss: StatusSnapshotCacheMiss, +) { + add_operator_snapshot_warning(snapshot, STATUS_OPERATOR_SNAPSHOT_WARNING); + + snapshot.warning_details.push(OperatorSnapshotWarningDetail { + warning: String::from(STATUS_OPERATOR_SNAPSHOT_WARNING), + project_id: Some(project.service_id().to_owned()), + repo_root: Some(project.repo_root().display().to_string()), + reason: cache_miss.reason, + next_action: Some(cache_miss.next_action), + }); +} + fn build_live_status_command_snapshot( config: &ServiceConfig, workflow: &WorkflowDocument, @@ -1605,6 +1908,8 @@ fn empty_control_plane_snapshot(limit: usize) -> OperatorStatusSnapshot { OperatorStatusSnapshot { project_id: String::from("all"), run_limit: limit, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index 064808ca7..d68e98721 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -336,6 +336,8 @@ fn build_operator_status_snapshot_with_account_mode( let mut snapshot = OperatorStatusSnapshot { project_id: project.service_id().to_owned(), run_limit: limit, + status_source: None, + snapshot_age_seconds: None, warnings, warning_details, connector_backoffs: Vec::new(), @@ -1430,6 +1432,7 @@ fn operator_status_worktrees( .list_worktrees(project.service_id())? .into_iter() .map(|mapping| OperatorWorktreeStatus { + project_id: project.service_id().to_owned(), issue_id: mapping.issue_id().to_owned(), issue_identifier: issue_identifier_in_text(mapping.branch_name()) .or_else(|| issue_identifier_in_text(&mapping.worktree_path().display().to_string())), @@ -1464,6 +1467,7 @@ fn operator_status_worktrees( .unwrap_or_else(|| issue_identifier.clone()); worktrees.push(OperatorWorktreeStatus { + project_id: project.service_id().to_owned(), issue_identifier: Some(issue_identifier.clone()), issue_id: issue_identifier, issue_state: None, @@ -1534,7 +1538,11 @@ fn append_merged_worktree_cleanup_debts( for debt in debts { let relative_path = relative_worktree_path_for_path(project, &debt.path); let is_dirty = debt.cleanliness.is_dirty(); - let debt_status = operator_worktree_status_from_cleanup_debt(debt, relative_path.clone()); + let debt_status = operator_worktree_status_from_cleanup_debt( + project.service_id(), + debt, + relative_path.clone(), + ); if !seen_paths.insert(relative_path.clone()) { if let Some(existing) = @@ -1576,6 +1584,7 @@ fn worktree_hygiene_unavailable_warning_detail( } fn operator_worktree_status_from_cleanup_debt( + project_id: &str, debt: MergedWorktreeCleanupDebt, relative_path: String, ) -> OperatorWorktreeStatus { @@ -1596,6 +1605,7 @@ fn operator_worktree_status_from_cleanup_debt( let branch_name = debt.branch_name; OperatorWorktreeStatus { + project_id: project_id.to_owned(), issue_id: branch_name.clone(), issue_identifier: issue_identifier_in_text(&branch_name) .or_else(|| issue_identifier_in_text(&relative_path)), @@ -2333,6 +2343,7 @@ where )?; Ok(OperatorQueuedIssueStatus { + project_id: project.service_id().to_owned(), issue_id: issue.id, issue_identifier: issue.identifier, title: issue.title, @@ -3075,6 +3086,7 @@ fn degraded_post_review_lane_status_from_classification( )?; Ok(OperatorPostReviewLaneStatus { + project_id: project.service_id().to_owned(), issue_id: worktree.issue_id().to_owned(), issue_identifier, issue_state: String::from("tracker_readback_degraded"), @@ -3283,6 +3295,7 @@ fn post_review_lane_status_from_classification( let loop_status = operator_post_review_loop_status(project, state_store, snapshot)?; Ok(OperatorPostReviewLaneStatus { + project_id: project.service_id().to_owned(), issue_id: snapshot.issue.id.clone(), issue_identifier: snapshot.issue.identifier.clone(), issue_state: snapshot.issue.state.name.clone(), @@ -4273,6 +4286,7 @@ fn blocked_post_review_lane_status( reason: &str, ) -> OperatorPostReviewLaneStatus { OperatorPostReviewLaneStatus { + project_id: project.service_id().to_owned(), issue_id: issue.id.clone(), issue_identifier: issue.identifier.clone(), issue_state: issue.state.name.clone(), @@ -6183,6 +6197,14 @@ fn render_operator_status(snapshot: &OperatorStatusSnapshot) -> String { let mut output = String::new(); output.push_str(&format!("Project: {}\n", snapshot.project_id)); + + if let Some(status_source) = snapshot.status_source.as_deref() { + output.push_str(&format!("Status source: {status_source}\n")); + } + if let Some(snapshot_age_seconds) = snapshot.snapshot_age_seconds { + output.push_str(&format!("Snapshot age: {snapshot_age_seconds}s\n")); + } + output.push_str(&format!("Warnings: {}\n", snapshot.warnings.len())); if !snapshot.warnings.is_empty() { diff --git a/apps/decodex/src/orchestrator/tests/operator/status/agent_evidence.rs b/apps/decodex/src/orchestrator/tests/operator/status/agent_evidence.rs index d8f1cecf4..3e36714ad 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/agent_evidence.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/agent_evidence.rs @@ -16,6 +16,8 @@ fn agent_evidence_snapshot_writes_index_blockers_capsules_and_event_stream() { let snapshot = OperatorStatusSnapshot { project_id: String::from(TEST_SERVICE_ID), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), diff --git a/apps/decodex/src/orchestrator/tests/operator/status/control_plane.rs b/apps/decodex/src/orchestrator/tests/operator/status/control_plane.rs index feecb1f8a..2ff00098b 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/control_plane.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/control_plane.rs @@ -1,4 +1,5 @@ use orchestrator::ProjectDaemonRuntime; +use orchestrator::StatusSnapshotHttpResponse; #[test] fn control_plane_snapshot_lists_disabled_registered_projects() { @@ -429,6 +430,165 @@ fn control_plane_snapshot_aggregates_top_level_lanes_for_all_registered_projects assert_eq!(snapshot.active_runs[0].phase, "executing"); } +#[test] +fn status_cache_projects_aggregate_snapshot_to_requested_project() { + let (active_temp_dir, active_config, _active_workflow) = temp_project_layout(); + let (_idle_temp_dir, idle_base_config, _idle_workflow) = temp_project_layout(); + let _home_guard = TestEnvVarGuard::set( + "HOME", + active_temp_dir.path().to_str().expect("home should be utf-8"), + ); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let issue = sample_issue_with_sort_fields( + "issue-active", + "PUB-101", + "Todo", + &[], + Some(3), + "2026-04-30T03:01:00Z", + ); + let worktree_path = active_config.worktree_root().join("PUB-101"); + + write_service_config( + idle_base_config.repo_root(), + &sample_service_config_toml("rsnap", "HOME", "HOME", None, ReviewLevel::Strict), + ); + + let idle_config = load_service_config(idle_base_config.repo_root()); + let active_registration = ProjectRegistration::from_config( + active_config.service_id(), + &service_config_path(active_config.repo_root()), + &active_config, + true, + "active-fingerprint", + ); + let idle_registration = ProjectRegistration::from_config( + idle_config.service_id(), + &service_config_path(idle_config.repo_root()), + &idle_config, + true, + "idle-fingerprint", + ); + + state_store + .record_run_attempt("run-active", &issue.id, 1, "running") + .expect("active run should record"); + state_store + .upsert_lease(active_config.service_id(), &issue.id, "run-active", "In Progress") + .expect("active lease should record"); + state_store + .upsert_worktree( + active_config.service_id(), + &issue.id, + "x/pubfi-pub-101", + &worktree_path.display().to_string(), + ) + .expect("active worktree should record"); + + let active_snapshot = + orchestrator::build_operator_status_snapshot(&active_config, &state_store, 25) + .expect("active project snapshot should build"); + let idle_snapshot = + orchestrator::build_operator_status_snapshot(&idle_config, &state_store, 25) + .expect("idle project snapshot should build"); + let mut active_snapshot = Some(active_snapshot); + let mut idle_snapshot = Some(idle_snapshot); + let aggregate = orchestrator::collect_control_plane_snapshot( + vec![active_registration, idle_registration], + |project, _project_warnings| { + let project_snapshot = match project.service_id() { + "pubfi" => active_snapshot.take().expect("active snapshot should be used once"), + "rsnap" => idle_snapshot.take().expect("idle snapshot should be used once"), + service_id => panic!("unexpected project {service_id}"), + }; + let project_status = project_snapshot + .projects + .first() + .cloned() + .map(|status| orchestrator::complete_project_status(project, status)); + + ControlPlaneProjectTick { + snapshot: Some(project_snapshot), + project_status, + } + }, + ); + let cached = orchestrator::status_snapshot_from_operator_cache_response( + &active_config, + 10, + StatusSnapshotHttpResponse { + body: serde_json::to_vec(&aggregate).expect("snapshot should serialize"), + published_at_unix_epoch: Some(100), + }, + 105, + ) + .expect("matching cached snapshot should project"); + + assert_eq!(cached.project_id, "pubfi"); + assert_eq!(cached.run_limit, 10); + assert_eq!(cached.status_source.as_deref(), Some("operator_snapshot_cache")); + assert_eq!(cached.snapshot_age_seconds, Some(5)); + assert_eq!(cached.projects.len(), 1); + assert_eq!(cached.projects[0].project_id, "pubfi"); + assert_eq!(cached.active_runs.len(), 1); + assert_eq!(cached.active_runs[0].project_id, "pubfi"); + assert!(cached.worktrees.iter().all(|worktree| worktree.project_id == "pubfi")); + assert!(cached.recent_runs.iter().all(|run| run.project_id == "pubfi")); +} + +#[test] +fn status_cache_rejects_missing_stale_or_too_small_snapshot() { + let (_temp_dir, config, _workflow) = temp_project_layout(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let snapshot = orchestrator::build_operator_status_snapshot(&config, &state_store, 25) + .expect("project snapshot should build"); + let body = serde_json::to_vec(&snapshot).expect("snapshot should serialize"); + let missing_timestamp = orchestrator::status_snapshot_from_operator_cache_response( + &config, + 10, + StatusSnapshotHttpResponse { + body: body.clone(), + published_at_unix_epoch: None, + }, + 105, + ) + .expect_err("missing publish timestamp should fall back"); + + assert!(missing_timestamp.reason.contains("omitted X-Decodex-Snapshot-Unix-Epoch")); + + let stale = orchestrator::status_snapshot_from_operator_cache_response( + &config, + 10, + StatusSnapshotHttpResponse { + body: body.clone(), + published_at_unix_epoch: Some(1), + }, + 1 + 61, + ) + .expect_err("stale snapshot should fall back"); + + assert!(stale.reason.contains("stale")); + + let too_small = orchestrator::status_snapshot_from_operator_cache_response( + &config, + 100, + StatusSnapshotHttpResponse { + body, + published_at_unix_epoch: Some(100), + }, + 101, + ) + .expect_err("lower snapshot limit should fall back"); + + assert!(too_small.reason.contains("run limit")); +} + +#[test] +fn status_cache_is_bypassed_for_live_status() { + assert!(orchestrator::status_should_attempt_operator_snapshot_cache(false)); + assert!(!orchestrator::status_should_attempt_operator_snapshot_cache(true)); +} + #[test] fn control_plane_snapshot_keeps_queued_project_summaries_service_scoped() { let (_decodex_temp_dir, decodex_base_config, _decodex_workflow) = temp_project_layout(); diff --git a/apps/decodex/src/orchestrator/tests/operator/status/text.rs b/apps/decodex/src/orchestrator/tests/operator/status/text.rs index 41b42fbfd..bce57b240 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/text.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/text.rs @@ -5,6 +5,8 @@ fn operator_status_text_surfaces_github_cli_authority() { let snapshot = OperatorStatusSnapshot { project_id: String::from("pubfi"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), @@ -62,6 +64,8 @@ fn operator_status_text_renders_human_readable_sections() { let snapshot = OperatorStatusSnapshot { project_id: String::from("pubfi"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), @@ -160,6 +164,8 @@ fn operator_status_text_surfaces_execution_program_summary() { let snapshot = OperatorStatusSnapshot { project_id: String::from("decodex"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), @@ -440,6 +446,8 @@ fn operator_status_text_explains_empty_backlog_checks() { let snapshot = OperatorStatusSnapshot { project_id: String::from("pubfi"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), @@ -476,6 +484,8 @@ fn operator_status_text_surfaces_cleanup_blocker_pr_url() { let snapshot = OperatorStatusSnapshot { project_id: String::from("pubfi"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), @@ -491,6 +501,7 @@ fn operator_status_text_surfaces_cleanup_blocker_pr_url() { history_lanes: Vec::new(), execution_programs: Vec::new(), worktrees: vec![orchestrator::OperatorWorktreeStatus { + project_id: String::from("pubfi"), issue_id: String::from("issue-3"), issue_identifier: Some(String::from("PUB-103")), issue_state: Some(String::from("Done")), @@ -510,6 +521,7 @@ fn operator_status_text_surfaces_cleanup_blocker_pr_url() { hygiene: None, }], post_review_lanes: vec![orchestrator::OperatorPostReviewLaneStatus { + project_id: String::from("pubfi"), issue_id: String::from("issue-3"), issue_identifier: String::from("PUB-103"), issue_state: String::from("Done"), @@ -552,6 +564,8 @@ fn operator_status_text_terminal_run_freshness_uses_terminal_update() { let snapshot = OperatorStatusSnapshot { project_id: String::from("pubfi"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), @@ -591,6 +605,8 @@ fn operator_status_text_active_run_without_live_activity_does_not_promote_update let snapshot = OperatorStatusSnapshot { project_id: String::from("pubfi"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), @@ -625,6 +641,8 @@ fn operator_status_text_explains_unleased_live_running_lane() { let snapshot = OperatorStatusSnapshot { project_id: String::from("pubfi"), run_limit: 10, + status_source: None, + snapshot_age_seconds: None, warnings: Vec::new(), warning_details: Vec::new(), connector_backoffs: Vec::new(), diff --git a/apps/decodex/src/orchestrator/tests/operator/status_support.rs b/apps/decodex/src/orchestrator/tests/operator/status_support.rs index 7d3982b6f..58b5d1e00 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status_support.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status_support.rs @@ -341,8 +341,9 @@ fn operator_status_text_active_run() -> orchestrator::OperatorRunStatus { fn operator_status_text_queued_candidates() -> Vec { vec![ - orchestrator::OperatorQueuedIssueStatus { - issue_id: String::from("issue-1"), + orchestrator::OperatorQueuedIssueStatus { + project_id: String::from(TEST_SERVICE_ID), + issue_id: String::from("issue-1"), issue_identifier: String::from("PUB-101"), title: String::from("Running lane still has a backlog claim"), author: Some(String::from("Yvette")), @@ -354,8 +355,9 @@ fn operator_status_text_queued_candidates() -> Vec { attention: None, blocker_identifiers: vec![], }, - orchestrator::OperatorQueuedIssueStatus { - issue_id: String::from("issue-2"), + orchestrator::OperatorQueuedIssueStatus { + project_id: String::from(TEST_SERVICE_ID), + issue_id: String::from("issue-2"), issue_identifier: String::from("PUB-102"), title: String::from("Implement backlog surface"), author: Some(String::from("Yvette")), @@ -367,8 +369,9 @@ fn operator_status_text_queued_candidates() -> Vec { attention: None, blocker_identifiers: vec![], }, - orchestrator::OperatorQueuedIssueStatus { - issue_id: String::from("issue-5"), + orchestrator::OperatorQueuedIssueStatus { + project_id: String::from(TEST_SERVICE_ID), + issue_id: String::from("issue-5"), issue_identifier: String::from("PUB-105"), title: String::from("Remove stale queue label"), author: Some(String::from("Yvette")), @@ -385,8 +388,9 @@ fn operator_status_text_queued_candidates() -> Vec { fn operator_status_text_worktrees() -> Vec { vec![ - orchestrator::OperatorWorktreeStatus { - issue_id: String::from("issue-4"), + orchestrator::OperatorWorktreeStatus { + project_id: String::from(TEST_SERVICE_ID), + issue_id: String::from("issue-4"), issue_identifier: Some(String::from("PUB-104")), issue_state: None, branch_name: String::from("x/pubfi-pub-104"), @@ -399,8 +403,9 @@ fn operator_status_text_worktrees() -> Vec { recovery_next_action: None, hygiene: None, }, - orchestrator::OperatorWorktreeStatus { - issue_id: String::from("issue-1"), + orchestrator::OperatorWorktreeStatus { + project_id: String::from(TEST_SERVICE_ID), + issue_id: String::from("issue-1"), issue_identifier: Some(String::from("PUB-101")), issue_state: Some(String::from("In Progress")), branch_name: String::from("x/pubfi-pub-101"), @@ -411,8 +416,9 @@ fn operator_status_text_worktrees() -> Vec { recovery_next_action: None, hygiene: None, }, - orchestrator::OperatorWorktreeStatus { - issue_id: String::from("issue-3"), + orchestrator::OperatorWorktreeStatus { + project_id: String::from(TEST_SERVICE_ID), + issue_id: String::from("issue-3"), issue_identifier: Some(String::from("PUB-103")), issue_state: Some(String::from("In Review")), branch_name: String::from("x/pubfi-pub-103"), @@ -439,6 +445,7 @@ fn test_worktree_provenance(source: &str) -> orchestrator::OperatorWorktreeProve fn operator_status_text_post_review_lanes() -> Vec { vec![orchestrator::OperatorPostReviewLaneStatus { + project_id: String::from(TEST_SERVICE_ID), issue_id: String::from("issue-3"), issue_identifier: String::from("PUB-103"), issue_state: String::from("In Review"), diff --git a/apps/decodex/src/orchestrator/types.rs b/apps/decodex/src/orchestrator/types.rs index f24a34b07..9b0d1b77c 100644 --- a/apps/decodex/src/orchestrator/types.rs +++ b/apps/decodex/src/orchestrator/types.rs @@ -1221,10 +1221,14 @@ struct TerminalFailureOutcome { retry_guarded_by_state: bool, } -#[derive(Clone, Debug, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] struct OperatorStatusSnapshot { project_id: String, run_limit: usize, + #[serde(skip_serializing_if = "Option::is_none")] + status_source: Option, + #[serde(skip_serializing_if = "Option::is_none")] + snapshot_age_seconds: Option, warnings: Vec, warning_details: Vec, connector_backoffs: Vec, @@ -1240,7 +1244,7 @@ struct OperatorStatusSnapshot { post_review_lanes: Vec, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorSnapshotWarningDetail { warning: String, project_id: Option, @@ -1249,7 +1253,7 @@ struct OperatorSnapshotWarningDetail { next_action: Option, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorConnectorBackoffStatus { project_id: String, connector: String, @@ -1263,7 +1267,7 @@ struct OperatorConnectorBackoffStatus { warning: String, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorProjectStatus { project_id: String, config_path: String, @@ -1284,7 +1288,7 @@ struct OperatorProjectStatus { warning_count: usize, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorExecutionProgramStatus { program_id: String, source_contract_id: String, @@ -1340,7 +1344,7 @@ impl OperatorExecutionProgramStatus { } } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorGitHubCliAuthority { command_path: String, resolved_path: Option, @@ -1350,13 +1354,13 @@ struct OperatorGitHubCliAuthority { next_action: String, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorCodexAccountControlStatus { mode: String, account_selector: Option, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorHistoryLaneStatus { project_id: String, issue_id: String, @@ -1370,7 +1374,7 @@ struct OperatorHistoryLaneStatus { attempts: Vec, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorHistoryLedgerOutcome { ledger_status: String, final_outcome: String, @@ -1388,7 +1392,7 @@ struct OperatorHistoryLedgerOutcome { record_count: usize, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorRunStatus { project_id: String, project_display_name: String, @@ -1445,7 +1449,7 @@ struct OperatorRunStatus { worktree_path: Option, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorRunControlCapability { project_id: String, issue_id: String, @@ -1460,8 +1464,9 @@ struct OperatorRunControlCapability { updated_at: String, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorQueuedIssueStatus { + project_id: String, issue_id: String, issue_identifier: String, title: String, @@ -1475,7 +1480,7 @@ struct OperatorQueuedIssueStatus { blocker_identifiers: Vec, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorQueuedIssueAttentionStatus { summary: String, #[serde(skip_serializing_if = "Option::is_none")] @@ -1502,7 +1507,7 @@ struct OperatorQueuedIssueAttentionStatus { worktree_has_tracked_changes: bool, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorAuthorityDecisionRequestStatus { phase: String, reason: String, @@ -1513,7 +1518,7 @@ struct OperatorAuthorityDecisionRequestStatus { resume_condition: Option, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorLoopStatus { review_level: String, autonomy: String, @@ -1525,14 +1530,14 @@ struct OperatorLoopStatus { decision_request: Option, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorReviewLoopStatus { phase: String, status: String, checkpoint: Option, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorReviewCheckpointStatus { head_sha: String, round: i64, @@ -1540,7 +1545,7 @@ struct OperatorReviewCheckpointStatus { updated_at: String, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorArchitectureRecoveryStatus { status: String, reason_code: String, @@ -1551,13 +1556,13 @@ struct OperatorArchitectureRecoveryStatus { next_action: String, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorRecoveryBudgetStatus { attempt: u64, max_attempts: u64, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorBoundaryStatus { disposition: String, reason: Option, @@ -1566,8 +1571,9 @@ struct OperatorBoundaryStatus { improvement_signal_count: usize, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorWorktreeStatus { + project_id: String, issue_id: String, issue_identifier: Option, issue_state: Option, @@ -1580,7 +1586,7 @@ struct OperatorWorktreeStatus { hygiene: Option, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorWorktreeProvenanceStatus { source: String, created_at_unix: Option, @@ -1588,7 +1594,7 @@ struct OperatorWorktreeProvenanceStatus { audit_required: bool, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorWorktreeHygieneStatus { classification: String, default_branch: String, @@ -1596,8 +1602,9 @@ struct OperatorWorktreeHygieneStatus { reason: String, } -#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorPostReviewLaneStatus { + project_id: String, issue_id: String, issue_identifier: String, issue_state: String, diff --git a/docs/reference/operator-control-plane.md b/docs/reference/operator-control-plane.md index 76c51475c..94062d1d2 100644 --- a/docs/reference/operator-control-plane.md +++ b/docs/reference/operator-control-plane.md @@ -212,6 +212,17 @@ acknowledgements. `GET /api/operator-snapshot` is the Decodex App read API over same runtime database, not a browser-dashboard polling authority and not a sign that the dev listener owns scheduling. +`decodex status` uses that same local `GET /api/operator-snapshot` as a fast read path +when the default listener is reachable, the published snapshot is recent, includes the +requested project, and its run limit is large enough for the requested status limit. +The CLI projects aggregate snapshots down to the requested project before printing. +JSON output identifies cache hits with `"status_source": "operator_snapshot_cache"` and +`snapshot_age_seconds`. If the snapshot is missing, stale, mismatched, or too small, +the command falls back to a direct local runtime read and emits +`status_cached_snapshot_unavailable` in `warning_details`. `decodex status --live` +always bypasses the cached snapshot and rebuilds status with fresh Linear/GitHub +observers. + For the lane-control rollout, active-lane UI posture is observe-only. The dashboard renders active-lane state, protocol activity, liveness, private-evidence references, local run-control capability metadata, and local acknowledgement/account controls, but