From 1e1d295cdeefacd11e188710e88ba8838a3011fd Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Thu, 11 Jun 2026 13:37:40 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Optimize Decodex operator status reads","authority":"XY-911","related":["XY-914"]} --- apps/decodex/src/orchestrator/entrypoints.rs | 2 +- .../decodex/src/orchestrator/operator_http.rs | 11 +- apps/decodex/src/orchestrator/status.rs | 103 +-- apps/decodex/src/runtime.rs | 5 + apps/decodex/src/state.rs | 2 +- apps/decodex/src/state/internal.rs | 684 ++++++++++++++---- apps/decodex/src/state/store.rs | 327 ++++++++- apps/decodex/src/state/tests.rs | 144 ++++ plugins/decodex/skills/automation/SKILL.md | 251 ++----- plugins/decodex/skills/manual-cli/SKILL.md | 161 +---- 10 files changed, 1142 insertions(+), 548 deletions(-) diff --git a/apps/decodex/src/orchestrator/entrypoints.rs b/apps/decodex/src/orchestrator/entrypoints.rs index 75cbf2af3..7ee15ddc4 100644 --- a/apps/decodex/src/orchestrator/entrypoints.rs +++ b/apps/decodex/src/orchestrator/entrypoints.rs @@ -238,7 +238,7 @@ pub(crate) fn print_status( eyre::bail!("`status --limit` must be greater than zero."); } - let state_store = runtime::open_runtime_store()?; + let state_store = runtime::open_runtime_store_lazy()?; let Some(config_path) = resolve_config_path(config_path, &state_store)? else { eyre::bail!( "No Decodex project config found. Pass this command's --config or register one with `decodex project add `." diff --git a/apps/decodex/src/orchestrator/operator_http.rs b/apps/decodex/src/orchestrator/operator_http.rs index 851fb1f01..968261386 100644 --- a/apps/decodex/src/orchestrator/operator_http.rs +++ b/apps/decodex/src/orchestrator/operator_http.rs @@ -584,16 +584,23 @@ fn build_operator_run_activity_event( }; let (leased_runs, recent_runs) = state_store.list_project_runs(project.service_id(), DEFAULT_OPERATOR_DASHBOARD_RUN_LIMIT)?; + let loop_evidence = state_store.project_loop_evidence_snapshot(project.service_id())?; let project_display_name = operator_project_display_name(&project); let recent_runs = recent_runs .into_iter() .map(|run| { - operator_run_status(&project, state_store, &project_display_name, run, now_unix_epoch) + operator_run_status( + &project, + &loop_evidence, + &project_display_name, + run, + now_unix_epoch, + ) }) .collect::>>()?; let project_active_runs = operator_active_run_statuses( &project, - state_store, + &loop_evidence, &project_display_name, leased_runs, &recent_runs, diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index 0d33354cd..064808ca7 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -16,6 +16,7 @@ use github::GhCommandResolution; use state::WORKTREE_PROVENANCE_FILESYSTEM_SCAN; use state::WORKTREE_PROVENANCE_GIT_HYGIENE_SCAN; use state::WORKTREE_PROVENANCE_LEGACY_UNKNOWN; +use state::ProjectLoopEvidenceSnapshot; use crate::pull_request::{self, PullRequestLandingGateView}; use crate::worktree; @@ -306,16 +307,23 @@ fn build_operator_status_snapshot_with_account_mode( ) -> crate::prelude::Result { let now_unix_epoch = OffsetDateTime::now_utc().unix_timestamp(); let (leased_runs, recent_runs) = state_store.list_project_runs(project.service_id(), limit)?; + let loop_evidence = state_store.project_loop_evidence_snapshot(project.service_id())?; let project_display_name = operator_project_display_name(project); let recent_runs = recent_runs .into_iter() .map(|run| { - operator_run_status(project, state_store, &project_display_name, run, now_unix_epoch) + operator_run_status( + project, + &loop_evidence, + &project_display_name, + run, + now_unix_epoch, + ) }) .collect::>>()?; let active_runs = operator_active_run_statuses( project, - state_store, + &loop_evidence, &project_display_name, leased_runs, &recent_runs, @@ -376,6 +384,7 @@ fn build_lane_inspect_operator_runs( ) -> crate::prelude::Result> { let now_unix_epoch = OffsetDateTime::now_utc().unix_timestamp(); let (active_runs, recent_runs) = state_store.list_project_runs(project.service_id(), limit)?; + let loop_evidence = state_store.project_loop_evidence_snapshot(project.service_id())?; let project_display_name = operator_project_display_name(project); let mut seen_run_ids = HashSet::new(); let mut runs = Vec::new(); @@ -393,7 +402,7 @@ fn build_lane_inspect_operator_runs( runs.push(operator_run_status( project, - state_store, + &loop_evidence, &project_display_name, run, now_unix_epoch, @@ -421,7 +430,7 @@ fn project_run_status_issue_matches(run: &ProjectRunStatus, issue: &str) -> bool fn operator_active_run_statuses( project: &ServiceConfig, - state_store: &StateStore, + loop_evidence: &ProjectLoopEvidenceSnapshot, project_display_name: &str, leased_runs: Vec, recent_runs: &[OperatorRunStatus], @@ -429,7 +438,15 @@ fn operator_active_run_statuses( ) -> crate::prelude::Result> { let mut active_runs = leased_runs .into_iter() - .map(|run| operator_run_status(project, state_store, project_display_name, run, now_unix_epoch)) + .map(|run| { + operator_run_status( + project, + loop_evidence, + project_display_name, + run, + now_unix_epoch, + ) + }) .collect::>>()? .into_iter() .filter(operator_run_counts_as_active) @@ -4960,7 +4977,7 @@ fn append_primary_account_if_missing( fn operator_run_status( project: &ServiceConfig, - state_store: &StateStore, + loop_evidence: &ProjectLoopEvidenceSnapshot, project_display_name: &str, run: ProjectRunStatus, now_unix_epoch: i64, @@ -4970,7 +4987,7 @@ fn operator_run_status( let app_server_state = operator_run_app_server_state(&run, marker.as_ref()); let protocol_summary = operator_run_protocol_summary(&run, marker.as_ref()); let terminal_finalize_projection = - operator_run_terminal_finalize_projection(project, state_store, &run)?; + operator_run_terminal_finalize_projection(loop_evidence, &run); let lifecycle = operator_run_lifecycle_projection( &run, marker.as_ref(), @@ -5005,7 +5022,7 @@ fn operator_run_status( let loop_status = operator_run_loop_status( project, - state_store, + loop_evidence, &run, &lifecycle.status, &lifecycle.phase, @@ -5188,15 +5205,15 @@ fn operator_run_private_evidence( fn operator_run_loop_status( project: &ServiceConfig, - state_store: &StateStore, + loop_evidence: &ProjectLoopEvidenceSnapshot, run: &ProjectRunStatus, status: &str, phase: &str, current_operation: &str, ) -> crate::prelude::Result { - operator_loop_status_for_run( + operator_loop_status_for_run_with_evidence( project, - state_store, + loop_evidence, run.issue_id(), run.run_id(), run.attempt_number(), @@ -5260,22 +5277,38 @@ fn operator_loop_status_for_run( default_review_phase: Option<&str>, lifecycle_summary: Option, ) -> crate::prelude::Result { - let review_level = project.codex().review_level(); - let review = operator_review_loop_status( - review_level, - state_store, - project.service_id(), + let loop_evidence = state_store.project_loop_evidence_snapshot(project.service_id())?; + + operator_loop_status_for_run_with_evidence( + project, + &loop_evidence, issue_id, run_id, attempt_number, default_review_phase, - )?; - let events = state_store.list_private_execution_events( - project.service_id(), + lifecycle_summary, + ) +} + +fn operator_loop_status_for_run_with_evidence( + project: &ServiceConfig, + loop_evidence: &ProjectLoopEvidenceSnapshot, + issue_id: &str, + run_id: &str, + attempt_number: i64, + default_review_phase: Option<&str>, + lifecycle_summary: Option, +) -> crate::prelude::Result { + let review_level = project.codex().review_level(); + let review = operator_review_loop_status( + review_level, + loop_evidence, issue_id, run_id, attempt_number, + default_review_phase, )?; + let events = loop_evidence.private_events(issue_id, run_id, attempt_number); let architecture_recovery = events .iter() .rev() @@ -5320,8 +5353,7 @@ fn operator_loop_status_for_run( fn operator_review_loop_status( review_level: ReviewLevel, - state_store: &StateStore, - project_id: &str, + loop_evidence: &ProjectLoopEvidenceSnapshot, issue_id: &str, run_id: &str, attempt_number: i64, @@ -5330,12 +5362,8 @@ fn operator_review_loop_status( let latest_checkpoint = ["handoff", "repair"] .into_iter() .filter_map(|phase| { - state_store - .review_policy_checkpoint(project_id, issue_id, run_id, attempt_number, phase) - .transpose() + loop_evidence.review_policy_checkpoint(issue_id, run_id, attempt_number, phase) }) - .collect::>>()? - .into_iter() .max_by(|left, right| { left.updated_at_unix() .cmp(&right.updated_at_unix()) @@ -5797,27 +5825,18 @@ fn operator_run_protocol_summary( } fn operator_run_terminal_finalize_projection( - project: &ServiceConfig, - state_store: &StateStore, + loop_evidence: &ProjectLoopEvidenceSnapshot, run: &ProjectRunStatus, -) -> crate::prelude::Result> { - let events = state_store.list_private_execution_events( - project.service_id(), - run.issue_id(), - run.run_id(), - run.attempt_number(), - )?; - let Some(path) = events +) -> Option { + let events = loop_evidence.private_events(run.issue_id(), run.run_id(), run.attempt_number()); + let path = events .iter() .rev() .find(|event| event.event_type() == "terminal_finalize") .and_then(|event| event.payload().get("path")) - .and_then(Value::as_str) - else { - return Ok(None); - }; + .and_then(Value::as_str)?; - Ok(match path { + match path { "review_handoff" => Some(OperatorTerminalFinalizeProjection { status: "review_handoff_pending", phase: "terminal_pending", @@ -5843,7 +5862,7 @@ fn operator_run_terminal_finalize_projection( current_operation: RUN_OPERATION_REVIEW_WRITEBACK, }), _ => None, - }) + } } fn operator_run_visible_status( diff --git a/apps/decodex/src/runtime.rs b/apps/decodex/src/runtime.rs index ce87a608c..912c5ea39 100644 --- a/apps/decodex/src/runtime.rs +++ b/apps/decodex/src/runtime.rs @@ -144,6 +144,11 @@ pub(crate) fn open_runtime_store() -> Result { StateStore::open(runtime_db_path()?) } +/// Open the global runtime database without preloading all durable rows. +pub(crate) fn open_runtime_store_lazy() -> Result { + StateStore::open_lazy(runtime_db_path()?) +} + /// Register or refresh one project config in the global runtime DB. pub(crate) fn register_project_config( state_store: &StateStore, diff --git a/apps/decodex/src/state.rs b/apps/decodex/src/state.rs index 20a997cba..d85478fca 100644 --- a/apps/decodex/src/state.rs +++ b/apps/decodex/src/state.rs @@ -8,7 +8,7 @@ use std::{ cmp, collections::{HashMap, HashSet}, fs::{self, File, OpenOptions, TryLockError}, - io::{Error, ErrorKind, Read, Seek, SeekFrom, Write}, + io::{ErrorKind, Read, Seek, SeekFrom, Write}, path::{Path, PathBuf}, process, sync::{Mutex, MutexGuard, OnceLock}, diff --git a/apps/decodex/src/state/internal.rs b/apps/decodex/src/state/internal.rs index 0030e6b15..95bc52c20 100644 --- a/apps/decodex/src/state/internal.rs +++ b/apps/decodex/src/state/internal.rs @@ -14,6 +14,7 @@ use libc::{ }; #[cfg(target_os = "macos")] use process::Command; +use rusqlite::{self, Row}; static RUN_ACTIVITY_MARKER_WRITE_SEQUENCE: AtomicU64 = AtomicU64::new(0); @@ -158,14 +159,20 @@ impl StateData { self.connector_backoffs = loaded.connector_backoffs; } - fn replace_project_run_state(&mut self, loaded: Self) { + fn replace_project_run_metadata_state(&mut self, loaded: Self) { self.leases = loaded.leases; self.run_attempts = loaded.run_attempts; self.control_channels = loaded.control_channels; - self.event_summaries = loaded.event_summaries; self.worktrees = loaded.worktrees; } + fn replace_project_loop_evidence_state(&mut self, project_id: &str, loaded: Self) { + self.private_execution_events.retain(|record| record.project_id != project_id); + self.private_execution_events.extend(loaded.private_execution_events); + self.review_policy_checkpoints.retain(|key, _record| key.project_id != project_id); + self.review_policy_checkpoints.extend(loaded.review_policy_checkpoints); + } + fn replace_project_registry_state(&mut self, loaded: Self) { self.projects = loaded.projects; } @@ -314,6 +321,8 @@ CREATE TABLE IF NOT EXISTS run_attempts ( updated_at TEXT NOT NULL, updated_at_unix INTEGER NOT NULL ); +CREATE INDEX IF NOT EXISTS run_attempts_issue_attempt_idx +ON run_attempts (issue_id, attempt_number, updated_at_unix, run_id); CREATE TABLE IF NOT EXISTS protocol_events ( run_id TEXT NOT NULL, sequence_number INTEGER NOT NULL, @@ -341,6 +350,8 @@ CREATE TABLE IF NOT EXISTS worktrees ( created_at_unix INTEGER, updated_at_unix INTEGER ); +CREATE INDEX IF NOT EXISTS worktrees_project_issue_idx +ON worktrees (project_id, issue_id); CREATE TABLE IF NOT EXISTS linear_execution_events ( idempotency_key TEXT PRIMARY KEY NOT NULL, service_id TEXT NOT NULL, @@ -641,14 +652,22 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(state) } - fn load_project_run_state_for_project(&self, project_id: &str) -> Result { + fn load_project_run_metadata_for_project(&self, project_id: &str) -> Result { let mut state = StateData::default(); self.load_leases(&mut state)?; - self.load_run_attempts(&mut state)?; + self.load_run_attempts_for_project(&mut state, project_id)?; self.load_worktrees(&mut state)?; self.load_run_control_channels_for_project(&mut state, project_id)?; - self.load_protocol_event_summaries_for_project_runs(&mut state, project_id)?; + + Ok(state) + } + + fn load_project_loop_evidence_for_project(&self, project_id: &str) -> Result { + let mut state = StateData::default(); + + self.load_private_execution_events_for_project(&mut state, project_id)?; + self.load_review_policy_checkpoints_for_project(&mut state, project_id)?; Ok(state) } @@ -714,6 +733,31 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + fn upsert_project(&self, project: &ProjectRegistration) -> Result<()> { + self.connection.execute( + "INSERT OR REPLACE INTO projects ( + service_id, config_path, repo_root, worktree_root, workflow_path, + tracker_api_key_env_var, github_token_env_var, enabled, config_fingerprint, + updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", + params![ + project.service_id(), + project.config_path().to_string_lossy().as_ref(), + project.repo_root().to_string_lossy().as_ref(), + project.worktree_root().to_string_lossy().as_ref(), + project.workflow_path().to_string_lossy().as_ref(), + project.tracker_api_key_env_var(), + project.github_token_env_var(), + if project.enabled() { 1_i64 } else { 0_i64 }, + project.config_fingerprint(), + project.updated_at(), + project.updated_at_unix(), + ], + )?; + + Ok(()) + } + fn delete_connector_backoff(&self, project_id: &str, connector: &str) -> Result<()> { self.connection.execute( "DELETE FROM connector_backoffs WHERE project_id = ?1 AND connector = ?2", @@ -723,6 +767,23 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + fn connector_backoff( + &self, + project_id: &str, + connector: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, connector, sync_phase, quota_class, reset_unix_epoch, + reset_source, warning, updated_at, updated_at_unix + FROM connector_backoffs + WHERE project_id = ?1 AND connector = ?2 + LIMIT 1", + )?; + let mut rows = statement.query(params![project_id, connector])?; + + Ok(rows.next()?.map(connector_backoff_from_row).transpose()?) + } + fn upsert_run_attempt(&self, attempt: &RunAttemptRecord) -> Result<()> { self.connection.execute( "INSERT OR REPLACE INTO run_attempts ( @@ -1258,6 +1319,23 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + fn load_run_attempts_for_project(&self, state: &mut StateData, project_id: &str) -> Result<()> { + let mut statement = self.connection.prepare( + "SELECT run_id, project_id, issue_id, attempt_number, status, thread_id, turn_id, \ + updated_at, updated_at_unix FROM run_attempts \ + WHERE project_id = ?1", + )?; + let rows = statement.query_map(params![project_id], run_attempt_record_from_row)?; + + for row in rows { + let attempt = row?; + + state.run_attempts.insert(attempt.run_id.clone(), attempt); + } + + Ok(()) + } + fn load_run_control_channels(&self, state: &mut StateData) -> Result<()> { let mut statement = self.connection.prepare( "SELECT run_id, project_id, issue_id, attempt_number, transport, channel_path, status, \ @@ -1353,6 +1431,48 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(count > 0) } + fn run_attempt_for_issue_attempt( + &self, + issue_id: &str, + attempt_number: i64, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT run_id, project_id, issue_id, attempt_number, status, thread_id, turn_id, \ + updated_at, updated_at_unix FROM run_attempts \ + WHERE issue_id = ?1 AND attempt_number = ?2 \ + ORDER BY updated_at_unix DESC, run_id DESC \ + LIMIT 1", + )?; + let mut rows = statement.query(params![issue_id, attempt_number])?; + + Ok(rows.next()?.map(run_attempt_record_from_row).transpose()?) + } + + fn latest_run_attempt_for_issue(&self, issue_id: &str) -> Result> { + let mut statement = self.connection.prepare( + "SELECT run_id, project_id, issue_id, attempt_number, status, thread_id, turn_id, \ + updated_at, updated_at_unix FROM run_attempts \ + WHERE issue_id = ?1 \ + ORDER BY attempt_number DESC, updated_at_unix DESC, run_id DESC \ + LIMIT 1", + )?; + let mut rows = statement.query(params![issue_id])?; + + Ok(rows.next()?.map(run_attempt_record_from_row).transpose()?) + } + + fn list_run_attempts_for_issue(&self, issue_id: &str) -> Result> { + let mut statement = self.connection.prepare( + "SELECT run_id, project_id, issue_id, attempt_number, status, thread_id, turn_id, \ + updated_at, updated_at_unix FROM run_attempts \ + WHERE issue_id = ?1 \ + ORDER BY attempt_number ASC, run_id ASC", + )?; + let rows = statement.query_map(params![issue_id], run_attempt_record_from_row)?; + + rows.collect::>>().map_err(Into::into) + } + fn load_protocol_event_summaries(&self, state: &mut StateData) -> Result<()> { self.load_compacted_protocol_event_summaries(state)?; @@ -1390,24 +1510,15 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } - fn load_protocol_event_summaries_for_project_runs( + fn load_protocol_event_summaries_for_runs( &self, state: &mut StateData, - project_id: &str, + run_ids: &[String], ) -> Result<()> { - let mut run_ids = state - .run_attempts - .values() - .filter(|attempt| state.project_run_status(project_id, attempt).is_some()) - .map(|attempt| attempt.run_id.clone()) - .collect::>(); - - run_ids.sort(); - run_ids.dedup(); - for run_id in run_ids { - self.load_compacted_protocol_event_summary_for_run(state, &run_id)?; - self.load_protocol_event_summary_for_run(state, &run_id)?; + state.event_summaries.remove(run_id); + self.load_compacted_protocol_event_summary_for_run(state, run_id)?; + self.load_protocol_event_summary_for_run(state, run_id)?; } Ok(()) @@ -1419,24 +1530,22 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; run_id: &str, ) -> Result<()> { let mut statement = self.connection.prepare( - "SELECT totals.event_count, totals.last_sequence_number, last.event_type, \ - last.created_at, last.created_at_unix \ - FROM ( - SELECT COUNT(*) AS event_count, MAX(sequence_number) AS last_sequence_number \ - FROM protocol_events WHERE run_id = ?1 - ) totals \ - JOIN protocol_events last \ - ON last.run_id = ?1 \ - AND last.sequence_number = totals.last_sequence_number", + "SELECT sequence_number, event_type, created_at, created_at_unix \ + FROM protocol_events \ + WHERE run_id = ?1 \ + ORDER BY sequence_number DESC \ + LIMIT 1", )?; let summary = statement .query_row(params![run_id], |row| { + let last_sequence_number = row.get(0)?; + Ok(ProtocolEventSummaryRecord { - event_count: row.get(0)?, - last_sequence_number: Some(row.get(1)?), - last_event_type: Some(row.get(2)?), - last_event_at: Some(row.get(3)?), - last_event_at_unix: Some(row.get(4)?), + event_count: last_sequence_number, + last_sequence_number: Some(last_sequence_number), + last_event_type: Some(row.get(1)?), + last_event_at: Some(row.get(2)?), + last_event_at_unix: Some(row.get(3)?), }) }) .optional()?; @@ -1512,20 +1621,9 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; FROM worktrees", )?; let rows = statement.query_map([], |row| { - let issue_id: String = row.get(0)?; + let mapping = worktree_mapping_record_from_row(row)?; - Ok(( - issue_id.clone(), - WorktreeMappingRecord { - issue_id, - project_id: row.get(1)?, - branch_name: row.get(2)?, - worktree_path: PathBuf::from(row.get::<_, String>(3)?), - provenance_source: row.get(4)?, - created_at_unix: row.get(5)?, - updated_at_unix: row.get(6)?, - }, - )) + Ok((mapping.issue_id.clone(), mapping)) })?; for row in rows { @@ -1537,6 +1635,19 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + fn worktree_for_issue(&self, issue_id: &str) -> Result> { + let mut statement = self.connection.prepare( + "SELECT issue_id, project_id, branch_name, worktree_path, + provenance_source, created_at_unix, updated_at_unix + FROM worktrees + WHERE issue_id = ?1 + LIMIT 1", + )?; + let mut rows = statement.query(params![issue_id])?; + + Ok(rows.next()?.map(worktree_mapping_record_from_row).transpose()?) + } + fn load_linear_execution_events(&self, state: &mut StateData) -> Result<()> { let mut statement = self.connection.prepare( "SELECT payload_json, event_unix, recorded_at, recorded_at_unix \ @@ -1655,22 +1766,27 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } - fn load_decision_contracts(&self, state: &mut StateData) -> Result<()> { + fn load_private_execution_events_for_project( + &self, + state: &mut StateData, + project_id: &str, + ) -> Result<()> { let mut statement = self.connection.prepare( - "SELECT project_id, contract_id, source_issue_id, status, payload_json, created_at, \ - created_at_unix, updated_at, updated_at_unix \ - FROM decision_contracts \ - ORDER BY project_id ASC, contract_id ASC", + "SELECT record_id, project_id, issue_id, run_id, attempt_number, event_type, \ + payload_json, recorded_at, recorded_at_unix \ + FROM private_execution_events \ + WHERE project_id = ?1 \ + ORDER BY record_id ASC", )?; - let rows = statement.query_map([], |row| { + let rows = statement.query_map(params![project_id], |row| { Ok(( - row.get::<_, String>(0)?, + row.get::<_, i64>(0)?, row.get::<_, String>(1)?, - row.get::<_, Option>(2)?, + row.get::<_, String>(2)?, row.get::<_, String>(3)?, - row.get::<_, String>(4)?, + row.get::<_, i64>(4)?, row.get::<_, String>(5)?, - row.get::<_, i64>(6)?, + row.get::<_, String>(6)?, row.get::<_, String>(7)?, row.get::<_, i64>(8)?, )) @@ -1678,53 +1794,99 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; for row in rows { let ( + record_id, project_id, - contract_id, - source_issue_id, - status, + issue_id, + run_id, + attempt_number, + event_type, payload_json, - created_at, - created_at_unix, - updated_at, - updated_at_unix, + recorded_at, + recorded_at_unix, ) = row?; - let contract = serde_json::from_str::(&payload_json)?; - let contract_status = contract.status(); + let payload = serde_json::from_str::(&payload_json)?; - contract.validate()?; + state.private_execution_events.push(PrivateExecutionEventRuntimeRecord { + record_id, + project_id, + issue_id, + run_id, + attempt_number, + event_type, + payload, + recorded_at, + recorded_at_unix, + }); + } - if contract_id != contract.contract_id() { - eyre::bail!( - "Decision contract row `{contract_id}` contained payload `{}`.", - contract.contract_id() - ); - } - if status != contract_status.as_str() { - tracing::warn!( - project_id = %project_id, - contract_id = %contract_id, - "decision contract status column differed from payload status" - ); - } + Ok(()) + } - state.decision_contracts.insert( - DecisionContractKey::new(&project_id, &contract_id), - DecisionContractRuntimeRecord { - project_id, - source_issue_id, - status: contract_status, - contract, - created_at, - created_at_unix, - updated_at, - updated_at_unix, - }, - ); + fn load_decision_contracts(&self, state: &mut StateData) -> Result<()> { + let mut statement = self.connection.prepare( + "SELECT project_id, contract_id, source_issue_id, status, payload_json, created_at, \ + created_at_unix, updated_at, updated_at_unix \ + FROM decision_contracts \ + ORDER BY project_id ASC, contract_id ASC", + )?; + let rows = statement.query_map([], decision_contract_runtime_row_parts)?; + + for row in rows { + let record = decision_contract_record_from_row_parts(row?)?; + + state.decision_contracts.insert(record.key(), record); } Ok(()) } + fn decision_contract( + &self, + project_id: &str, + contract_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, contract_id, source_issue_id, status, payload_json, created_at, \ + created_at_unix, updated_at, updated_at_unix \ + FROM decision_contracts \ + WHERE project_id = ?1 AND contract_id = ?2 \ + LIMIT 1", + )?; + let mut rows = statement.query(params![project_id, contract_id])?; + + rows + .next()? + .map(decision_contract_runtime_row_parts) + .transpose()? + .map(decision_contract_record_from_row_parts) + .transpose() + } + + fn list_decision_contracts_for_issue( + &self, + project_id: &str, + source_issue_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, contract_id, source_issue_id, status, payload_json, created_at, \ + created_at_unix, updated_at, updated_at_unix \ + FROM decision_contracts \ + WHERE project_id = ?1 AND source_issue_id = ?2 \ + ORDER BY created_at_unix ASC, contract_id ASC", + )?; + let rows = statement.query_map( + params![project_id, source_issue_id], + decision_contract_runtime_row_parts, + )?; + let mut records = Vec::new(); + + for row in rows { + records.push(decision_contract_record_from_row_parts(row?)?); + } + + Ok(records) + } + fn load_execution_programs(&self, state: &mut StateData) -> Result<()> { let mut statement = self.connection.prepare( "SELECT project_id, program_id, source_contract_id, payload_json, created_at, \ @@ -1732,62 +1894,83 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; FROM execution_programs \ ORDER BY project_id ASC, program_id ASC", )?; - let rows = statement.query_map([], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, String>(1)?, - row.get::<_, String>(2)?, - row.get::<_, String>(3)?, - row.get::<_, String>(4)?, - row.get::<_, i64>(5)?, - row.get::<_, String>(6)?, - row.get::<_, i64>(7)?, - )) - })?; + let rows = statement.query_map([], execution_program_runtime_row_parts)?; for row in rows { - let ( - project_id, - program_id, - source_contract_id, - payload_json, - created_at, - created_at_unix, - updated_at, - updated_at_unix, - ) = row?; - let program = serde_json::from_str::(&payload_json)?; + let record = execution_program_record_from_row_parts(row?)?; - program.validate()?; + state.execution_programs.insert(record.key(), record); + } - if program_id != program.program_id() { - eyre::bail!( - "Execution program row `{program_id}` contained payload `{}`.", - program.program_id() - ); - } - if source_contract_id != program.source_contract_id() { - eyre::bail!( - "Execution program row `{program_id}` carried source contract `{source_contract_id}` but payload references `{}`.", - program.source_contract_id() - ); - } + Ok(()) + } - state.execution_programs.insert( - ExecutionProgramKey::new(&project_id, &program_id), - ExecutionProgramRuntimeRecord { - project_id, - source_contract_id, - program, - created_at, - created_at_unix, - updated_at, - updated_at_unix, - }, - ); + fn execution_program( + &self, + project_id: &str, + program_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, source_contract_id, payload_json, created_at, \ + created_at_unix, updated_at, updated_at_unix \ + FROM execution_programs \ + WHERE project_id = ?1 AND program_id = ?2 \ + LIMIT 1", + )?; + let mut rows = statement.query(params![project_id, program_id])?; + + rows + .next()? + .map(execution_program_runtime_row_parts) + .transpose()? + .map(execution_program_record_from_row_parts) + .transpose() + } + + fn list_execution_programs_for_contract( + &self, + project_id: &str, + source_contract_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, source_contract_id, payload_json, created_at, \ + created_at_unix, updated_at, updated_at_unix \ + FROM execution_programs \ + WHERE project_id = ?1 AND source_contract_id = ?2 \ + ORDER BY created_at_unix ASC, program_id ASC", + )?; + let rows = statement.query_map( + params![project_id, source_contract_id], + execution_program_runtime_row_parts, + )?; + let mut records = Vec::new(); + + for row in rows { + records.push(execution_program_record_from_row_parts(row?)?); } - Ok(()) + Ok(records) + } + + fn list_execution_programs( + &self, + project_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, source_contract_id, payload_json, created_at, \ + created_at_unix, updated_at, updated_at_unix \ + FROM execution_programs \ + WHERE project_id = ?1 \ + ORDER BY created_at_unix ASC, program_id ASC", + )?; + let rows = statement.query_map(params![project_id], execution_program_runtime_row_parts)?; + let mut records = Vec::new(); + + for row in rows { + records.push(execution_program_record_from_row_parts(row?)?); + } + + Ok(records) } fn load_review_handoffs(&self, state: &mut StateData) -> Result<()> { @@ -1933,6 +2116,50 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + fn load_review_policy_checkpoints_for_project( + &self, + state: &mut StateData, + project_id: &str, + ) -> Result<()> { + let mut statement = self.connection.prepare( + "SELECT project_id, issue_id, run_id, attempt_number, phase, status, head_sha, \ + nonclean_rounds, details_json, updated_at, updated_at_unix FROM review_policy_checkpoints \ + WHERE project_id = ?1", + )?; + let rows = statement.query_map(params![project_id], |row| { + let project_id: String = row.get(0)?; + let issue_id: String = row.get(1)?; + let run_id: String = row.get(2)?; + let attempt_number: i64 = row.get(3)?; + let phase: String = row.get(4)?; + + Ok(( + ReviewPolicyKey::new(&project_id, &issue_id, &run_id, attempt_number, &phase), + ReviewPolicyRuntimeRecord { + project_id, + issue_id, + run_id, + attempt_number, + phase, + status: row.get(5)?, + head_sha: row.get(6)?, + nonclean_rounds: row.get(7)?, + details_json: row.get(8)?, + updated_at: row.get(9)?, + updated_at_unix: row.get(10)?, + }, + )) + })?; + + for row in rows { + let (key, record) = row?; + + state.review_policy_checkpoints.insert(key, record); + } + + Ok(()) + } + fn load_loop_guardrail_checkpoints(&self, state: &mut StateData) -> Result<()> { let mut statement = self.connection.prepare( "SELECT project_id, issue_id, reason, fingerprint, run_id, attempt_number, \ @@ -2449,6 +2676,29 @@ struct RunActivityMarkerRecord { review_policy_nonclean_rounds: Option, } +struct DecisionContractRuntimeRowParts { + project_id: String, + contract_id: String, + source_issue_id: Option, + status: String, + payload_json: String, + created_at: String, + created_at_unix: i64, + updated_at: String, + updated_at_unix: i64, +} + +struct ExecutionProgramRuntimeRowParts { + project_id: String, + program_id: String, + source_contract_id: String, + payload_json: String, + created_at: String, + created_at_unix: i64, + updated_at: String, + updated_at_unix: i64, +} + #[derive(Clone, Copy, Eq, PartialEq)] enum GuardRetention { Local, @@ -4106,6 +4356,150 @@ fn compare_attempt_records(left: &RunAttemptRecord, right: &RunAttemptRecord) -> .then_with(|| left.run_id.cmp(&right.run_id)) } +fn run_attempt_record_from_row( + row: &Row<'_>, +) -> std::result::Result { + Ok(RunAttemptRecord { + run_id: row.get(0)?, + project_id: row.get(1)?, + issue_id: row.get(2)?, + attempt_number: row.get(3)?, + status: row.get(4)?, + thread_id: row.get(5)?, + turn_id: row.get(6)?, + updated_at: row.get(7)?, + updated_at_unix: row.get(8)?, + }) +} + +fn worktree_mapping_record_from_row( + row: &Row<'_>, +) -> std::result::Result { + Ok(WorktreeMappingRecord { + issue_id: row.get(0)?, + project_id: row.get(1)?, + branch_name: row.get(2)?, + worktree_path: PathBuf::from(row.get::<_, String>(3)?), + provenance_source: row.get(4)?, + created_at_unix: row.get(5)?, + updated_at_unix: row.get(6)?, + }) +} + +fn decision_contract_runtime_row_parts( + row: &Row<'_>, +) -> std::result::Result { + Ok(DecisionContractRuntimeRowParts { + project_id: row.get(0)?, + contract_id: row.get(1)?, + source_issue_id: row.get(2)?, + status: row.get(3)?, + payload_json: row.get(4)?, + created_at: row.get(5)?, + created_at_unix: row.get(6)?, + updated_at: row.get(7)?, + updated_at_unix: row.get(8)?, + }) +} + +fn decision_contract_record_from_row_parts( + parts: DecisionContractRuntimeRowParts, +) -> Result { + let contract = serde_json::from_str::(&parts.payload_json)?; + let contract_status = contract.status(); + + contract.validate()?; + + if parts.contract_id != contract.contract_id() { + eyre::bail!( + "Decision contract row `{}` contained payload `{}`.", + parts.contract_id, + contract.contract_id() + ); + } + if parts.status != contract_status.as_str() { + tracing::warn!( + project_id = %parts.project_id, + contract_id = %parts.contract_id, + "decision contract status column differed from payload status" + ); + } + + Ok(DecisionContractRuntimeRecord { + project_id: parts.project_id, + source_issue_id: parts.source_issue_id, + status: contract_status, + contract, + created_at: parts.created_at, + created_at_unix: parts.created_at_unix, + updated_at: parts.updated_at, + updated_at_unix: parts.updated_at_unix, + }) +} + +fn execution_program_runtime_row_parts( + row: &Row<'_>, +) -> std::result::Result { + Ok(ExecutionProgramRuntimeRowParts { + project_id: row.get(0)?, + program_id: row.get(1)?, + source_contract_id: row.get(2)?, + payload_json: row.get(3)?, + created_at: row.get(4)?, + created_at_unix: row.get(5)?, + updated_at: row.get(6)?, + updated_at_unix: row.get(7)?, + }) +} + +fn execution_program_record_from_row_parts( + parts: ExecutionProgramRuntimeRowParts, +) -> Result { + let program = serde_json::from_str::(&parts.payload_json)?; + + program.validate()?; + + if parts.program_id != program.program_id() { + eyre::bail!( + "Execution program row `{}` contained payload `{}`.", + parts.program_id, + program.program_id() + ); + } + if parts.source_contract_id != program.source_contract_id() { + eyre::bail!( + "Execution program row `{}` carried source contract `{}` but payload references `{}`.", + parts.program_id, + parts.source_contract_id, + program.source_contract_id() + ); + } + + Ok(ExecutionProgramRuntimeRecord { + project_id: parts.project_id, + source_contract_id: parts.source_contract_id, + program, + created_at: parts.created_at, + created_at_unix: parts.created_at_unix, + updated_at: parts.updated_at, + updated_at_unix: parts.updated_at_unix, + }) +} + +fn connector_backoff_from_row(row: &Row<'_>) -> std::result::Result { + Ok(ConnectorBackoff { + project_id: row.get(0)?, + connector: row.get(1)?, + sync_phase: row.get(2)?, + quota_class: row.get(3)?, + reset_unix_epoch: row.get(4)?, + reset_source: row.get(5)?, + warning: row.get(6)?, + updated_at: row.get(7)?, + updated_at_unix: row.get(8)?, + }) +} + fn compare_linear_execution_event_runtime_records( left: &LinearExecutionEventRuntimeRecord, right: &LinearExecutionEventRuntimeRecord, @@ -4158,7 +4552,7 @@ fn clear_close_on_exec(file: &File) -> Result<()> { let existing_flags = unsafe { libc::fcntl(fd, F_GETFD) }; if existing_flags == -1 { - return Err(Error::last_os_error().into()); + return Err(std::io::Error::last_os_error().into()); } let new_flags = existing_flags & !FD_CLOEXEC; @@ -4167,7 +4561,7 @@ fn clear_close_on_exec(file: &File) -> Result<()> { let result = unsafe { libc::fcntl(fd, F_SETFD, new_flags) }; if result == -1 { - return Err(Error::last_os_error().into()); + return Err(std::io::Error::last_os_error().into()); } } @@ -4180,7 +4574,7 @@ fn set_close_on_exec(file: &File) -> Result<()> { let existing_flags = unsafe { libc::fcntl(fd, F_GETFD) }; if existing_flags == -1 { - return Err(Error::last_os_error().into()); + return Err(std::io::Error::last_os_error().into()); } let new_flags = existing_flags | FD_CLOEXEC; @@ -4189,7 +4583,7 @@ fn set_close_on_exec(file: &File) -> Result<()> { let result = unsafe { libc::fcntl(fd, F_SETFD, new_flags) }; if result == -1 { - return Err(Error::last_os_error().into()); + return Err(std::io::Error::last_os_error().into()); } } diff --git a/apps/decodex/src/state/store.rs b/apps/decodex/src/state/store.rs index 00ecb2e0b..02d325224 100644 --- a/apps/decodex/src/state/store.rs +++ b/apps/decodex/src/state/store.rs @@ -24,6 +24,74 @@ pub(crate) struct ReviewPolicyCheckpointInput<'a> { pub(crate) details_json: &'a str, } +/// Project-scoped loop evidence cached for one operator status render. +#[derive(Clone, Debug, Default)] +pub(crate) struct ProjectLoopEvidenceSnapshot { + private_events: HashMap<(String, String, i64), Vec>, + review_checkpoints: HashMap<(String, String, i64, String), ReviewPolicyCheckpoint>, +} +impl ProjectLoopEvidenceSnapshot { + fn insert_private_event(&mut self, event: PrivateExecutionEvent) { + self.private_events + .entry(( + event.issue_id().to_owned(), + event.run_id().to_owned(), + event.attempt_number(), + )) + .or_default() + .push(event); + } + + fn insert_review_checkpoint(&mut self, checkpoint: ReviewPolicyCheckpoint) { + self.review_checkpoints.insert( + ( + checkpoint.issue_id().to_owned(), + checkpoint.run_id().to_owned(), + checkpoint.attempt_number(), + checkpoint.phase().to_owned(), + ), + checkpoint, + ); + } + + fn sort_private_events(&mut self) { + for events in self.private_events.values_mut() { + events.sort_by(|left, right| { + left.recorded_at_unix() + .cmp(&right.recorded_at_unix()) + .then_with(|| left.record_id().cmp(&right.record_id())) + }); + } + } + + pub(crate) fn private_events( + &self, + issue_id: &str, + run_id: &str, + attempt_number: i64, + ) -> &[PrivateExecutionEvent] { + self.private_events + .get(&(issue_id.to_owned(), run_id.to_owned(), attempt_number)) + .map(Vec::as_slice) + .unwrap_or(&[]) + } + + pub(crate) fn review_policy_checkpoint( + &self, + issue_id: &str, + run_id: &str, + attempt_number: i64, + phase: &str, + ) -> Option<&ReviewPolicyCheckpoint> { + self.review_checkpoints.get(&( + issue_id.to_owned(), + run_id.to_owned(), + attempt_number, + phase.to_owned(), + )) + } +} + /// Input fields for recording the latest loop-guardrail checkpoint. pub(crate) struct LoopGuardrailCheckpointInput<'a> { pub(crate) project_id: &'a str, @@ -50,6 +118,13 @@ impl StateStore { Ok(Self { inner: Mutex::new(state), sqlite: Some(Mutex::new(sqlite)) }) } + /// Open the local persistent runtime store without preloading durable rows. + pub fn open_lazy(path: impl AsRef) -> Result { + let sqlite = SqliteStateStore::open(path.as_ref())?; + + Ok(Self { inner: Mutex::new(StateData::default()), sqlite: Some(Mutex::new(sqlite)) }) + } + /// Open an in-memory runtime store for tests. pub fn open_in_memory() -> Result { Ok(Self::default()) @@ -63,7 +138,10 @@ impl StateStore { &self, registration: &ProjectRegistration, ) -> Result { - let mut state = self.lock()?; + let mut state = self.lock_without_refresh()?; + + self.refresh_project_registry_state_locked(&mut state)?; + let mut registration = registration.clone(); if let Some(enabled) = state.projects.get(registration.service_id()).map(ProjectRegistration::enabled) @@ -75,7 +153,7 @@ impl StateStore { state .projects .insert(registration.service_id().to_owned(), registration.clone()); - self.persist_runtime_state_locked(&state)?; + self.upsert_project_locked(®istration)?; Ok(registration) } @@ -154,6 +232,12 @@ impl StateStore { project_id: &str, connector: &str, ) -> Result> { + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite.connector_backoff(project_id, connector); + } + let state = self.lock()?; Ok(state @@ -184,7 +268,7 @@ impl StateStore { ) -> Result<()> { let slot_limit = slot_limit.into(); let worktree_root = worktree_root.as_ref().to_path_buf(); - let mut state = self.lock()?; + let mut state = self.lock_without_refresh()?; slot_limit.validate()?; @@ -436,7 +520,7 @@ impl StateStore { pub fn list_leases(&self, project_id: &str) -> Result> { let mut state = self.lock_without_refresh()?; - self.refresh_project_run_state_locked(&mut state, project_id)?; + self.refresh_project_run_metadata_state_locked(&mut state, project_id)?; let mut leases = state .leases @@ -455,7 +539,7 @@ impl StateStore { let (mut leases_by_issue, dispatch_slot_config) = { let mut state = self.lock_without_refresh()?; - self.refresh_project_run_state_locked(&mut state, project_id)?; + self.refresh_project_run_metadata_state_locked(&mut state, project_id)?; let leases = state .leases @@ -1213,6 +1297,14 @@ impl StateStore { issue_id: &str, attempt_number: i64, ) -> Result> { + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite + .run_attempt_for_issue_attempt(issue_id, attempt_number) + .map(|attempt| attempt.map(|attempt| attempt.as_public())); + } + let state = self.lock()?; let attempt = state .run_attempts @@ -1228,6 +1320,14 @@ impl StateStore { /// Read the latest run attempt for one issue. pub fn latest_run_attempt_for_issue(&self, issue_id: &str) -> Result> { + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite + .latest_run_attempt_for_issue(issue_id) + .map(|attempt| attempt.map(|attempt| attempt.as_public())); + } + let state = self.lock()?; let attempt = state .run_attempts @@ -1241,6 +1341,17 @@ impl StateStore { /// List all locally recorded run attempts for one issue. pub fn list_run_attempts_for_issue(&self, issue_id: &str) -> Result> { + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + let attempts = sqlite + .list_run_attempts_for_issue(issue_id)? + .into_iter() + .map(|attempt| attempt.as_public()) + .collect(); + + return Ok(attempts); + } + let state = self.lock()?; let mut attempts = state .run_attempts @@ -1273,20 +1384,7 @@ impl StateStore { project_id: &str, limit: usize, ) -> Result> { - let mut state = self.lock_without_refresh()?; - - self.refresh_project_run_state_locked(&mut state, project_id)?; - - let mut runs = state - .run_attempts - .values() - .filter_map(|attempt| state.project_run_status(project_id, attempt)) - .collect::>(); - - runs.sort_by(compare_project_run_status); - runs.truncate(limit); - - Ok(runs) + self.list_project_runs(project_id, limit).map(|(_active, recent)| recent) } /// List active and recent run attempts for one project from one durable snapshot. @@ -1297,7 +1395,7 @@ impl StateStore { ) -> Result<(Vec, Vec)> { let mut state = self.lock_without_refresh()?; - self.refresh_project_run_state_locked(&mut state, project_id)?; + self.refresh_project_run_metadata_state_locked(&mut state, project_id)?; let mut runs = state .run_attempts @@ -1313,7 +1411,37 @@ impl StateStore { .cloned() .collect::>(); let recent_limit = base_recent_limit.saturating_add(active_runs.len()); - let mut recent_runs = runs; + let recent_run_ids = runs + .iter() + .take(recent_limit) + .map(|run| run.run_id().to_owned()) + .collect::>(); + let mut summary_run_ids = active_runs + .iter() + .map(|run| run.run_id().to_owned()) + .collect::>(); + + summary_run_ids.extend(recent_run_ids); + summary_run_ids.sort(); + summary_run_ids.dedup(); + self.refresh_protocol_event_summaries_for_runs_locked(&mut state, &summary_run_ids)?; + + let summary_run_id_set = summary_run_ids.iter().cloned().collect::>(); + let mut selected_runs = state + .run_attempts + .values() + .filter(|attempt| summary_run_id_set.contains(&attempt.run_id)) + .filter_map(|attempt| state.project_run_status(project_id, attempt)) + .collect::>(); + + selected_runs.sort_by(compare_project_run_status); + + let active_runs = selected_runs + .iter() + .filter(|status| status.active_lease()) + .cloned() + .collect::>(); + let mut recent_runs = selected_runs; recent_runs.truncate(recent_limit); @@ -1324,7 +1452,7 @@ impl StateStore { pub fn list_active_runs(&self, project_id: &str) -> Result> { let mut state = self.lock_without_refresh()?; - self.refresh_project_run_state_locked(&mut state, project_id)?; + self.refresh_project_run_metadata_state_locked(&mut state, project_id)?; let mut runs = state .run_attempts @@ -1335,6 +1463,22 @@ impl StateStore { status.active_lease.then_some(status) }) .collect::>(); + let mut run_ids = runs.iter().map(|run| run.run_id().to_owned()).collect::>(); + + run_ids.sort(); + run_ids.dedup(); + self.refresh_protocol_event_summaries_for_runs_locked(&mut state, &run_ids)?; + + runs = state + .run_attempts + .values() + .filter(|attempt| run_ids.contains(&attempt.run_id)) + .filter_map(|attempt| { + let status = state.project_run_status(project_id, attempt)?; + + status.active_lease.then_some(status) + }) + .collect::>(); runs.sort_by(compare_project_run_status); @@ -1562,6 +1706,36 @@ impl StateStore { Ok(records.into_iter().map(|record| record.as_public()).collect()) } + /// Build one project-scoped loop evidence snapshot for operator status rendering. + pub(crate) fn project_loop_evidence_snapshot( + &self, + project_id: &str, + ) -> Result { + let mut state = self.lock_without_refresh()?; + let mut snapshot = ProjectLoopEvidenceSnapshot::default(); + + self.refresh_project_loop_evidence_state_locked(&mut state, project_id)?; + + for record in state + .private_execution_events + .iter() + .filter(|record| record.project_id == project_id) + { + snapshot.insert_private_event(record.as_public()); + } + for record in state + .review_policy_checkpoints + .values() + .filter(|record| record.project_id == project_id) + { + snapshot.insert_review_checkpoint(record.as_public()); + } + + snapshot.sort_private_events(); + + Ok(snapshot) + } + /// Create or replace one local Loop/Decision Contract payload. #[allow(dead_code)] pub(crate) fn upsert_decision_contract( @@ -1608,6 +1782,14 @@ impl StateStore { validate_required_decision_contract_field("project_id", project_id)?; validate_required_decision_contract_field("contract_id", contract_id)?; + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite + .decision_contract(project_id, contract_id) + .map(|record| record.map(|record| record.as_public())); + } + let state = self.lock()?; Ok(state @@ -1626,6 +1808,17 @@ impl StateStore { validate_required_decision_contract_field("project_id", project_id)?; validate_required_decision_contract_field("source_issue_id", source_issue_id)?; + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + let records = sqlite + .list_decision_contracts_for_issue(project_id, source_issue_id)? + .into_iter() + .map(|record| record.as_public()) + .collect(); + + return Ok(records); + } + let state = self.lock()?; let mut records = state .decision_contracts @@ -1758,6 +1951,14 @@ impl StateStore { validate_required_execution_program_field("project_id", project_id)?; validate_required_execution_program_field("program_id", program_id)?; + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite + .execution_program(project_id, program_id) + .map(|record| record.map(|record| record.as_public())); + } + let state = self.lock()?; Ok(state @@ -1776,6 +1977,17 @@ impl StateStore { validate_required_execution_program_field("project_id", project_id)?; validate_required_execution_program_field("source_contract_id", source_contract_id)?; + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + let records = sqlite + .list_execution_programs_for_contract(project_id, source_contract_id)? + .into_iter() + .map(|record| record.as_public()) + .collect(); + + return Ok(records); + } + let state = self.lock()?; let mut records = state .execution_programs @@ -1799,6 +2011,17 @@ impl StateStore { ) -> Result> { validate_required_execution_program_field("project_id", project_id)?; + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + let records = sqlite + .list_execution_programs(project_id)? + .into_iter() + .map(|record| record.as_public()) + .collect(); + + return Ok(records); + } + let state = self.lock()?; let mut records = state .execution_programs @@ -2198,6 +2421,14 @@ impl StateStore { /// Read the worktree mapping for one issue. pub fn worktree_for_issue(&self, issue_id: &str) -> Result> { + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite + .worktree_for_issue(issue_id) + .map(|mapping| mapping.map(|mapping| mapping.as_public())); + } + let state = self.lock()?; Ok(state.worktrees.get(issue_id).map(WorktreeMappingRecord::as_public)) @@ -2207,7 +2438,7 @@ impl StateStore { pub fn list_worktrees(&self, project_id: &str) -> Result> { let mut state = self.lock_without_refresh()?; - self.refresh_project_run_state_locked(&mut state, project_id)?; + self.refresh_project_run_metadata_state_locked(&mut state, project_id)?; let mut mappings = state .worktrees @@ -2264,7 +2495,40 @@ impl StateStore { Ok(()) } - fn refresh_project_run_state_locked( + fn refresh_project_run_metadata_state_locked( + &self, + state: &mut StateData, + project_id: &str, + ) -> Result<()> { + let Some(sqlite) = self.sqlite.as_ref() else { + return Ok(()); + }; + let sqlite = sqlite + .lock() + .map_err(|_| eyre::eyre!("StateStore SQLite mutex is poisoned."))?; + let loaded = sqlite.load_project_run_metadata_for_project(project_id)?; + + state.replace_project_run_metadata_state(loaded); + + Ok(()) + } + + fn refresh_protocol_event_summaries_for_runs_locked( + &self, + state: &mut StateData, + run_ids: &[String], + ) -> Result<()> { + let Some(sqlite) = self.sqlite.as_ref() else { + return Ok(()); + }; + let sqlite = sqlite + .lock() + .map_err(|_| eyre::eyre!("StateStore SQLite mutex is poisoned."))?; + + sqlite.load_protocol_event_summaries_for_runs(state, run_ids) + } + + fn refresh_project_loop_evidence_state_locked( &self, state: &mut StateData, project_id: &str, @@ -2275,9 +2539,9 @@ impl StateStore { let sqlite = sqlite .lock() .map_err(|_| eyre::eyre!("StateStore SQLite mutex is poisoned."))?; - let loaded = sqlite.load_project_run_state_for_project(project_id)?; + let loaded = sqlite.load_project_loop_evidence_for_project(project_id)?; - state.replace_project_run_state(loaded); + state.replace_project_loop_evidence_state(project_id, loaded); Ok(()) } @@ -2318,6 +2582,17 @@ impl StateStore { sqlite.delete_project(service_id) } + fn upsert_project_locked(&self, project: &ProjectRegistration) -> Result<()> { + let Some(sqlite) = self.sqlite.as_ref() else { + return Ok(()); + }; + let sqlite = sqlite + .lock() + .map_err(|_| eyre::eyre!("StateStore SQLite mutex is poisoned."))?; + + sqlite.upsert_project(project) + } + fn delete_connector_backoff_locked(&self, project_id: &str, connector: &str) -> Result<()> { let Some(sqlite) = self.sqlite.as_ref() else { return Ok(()); diff --git a/apps/decodex/src/state/tests.rs b/apps/decodex/src/state/tests.rs index eb55cfd81..679b89567 100644 --- a/apps/decodex/src/state/tests.rs +++ b/apps/decodex/src/state/tests.rs @@ -2661,6 +2661,88 @@ fn private_execution_events_persist_reload_and_keep_append_order() { assert!(!events[0].recorded_at().is_empty()); } +#[test] +fn project_loop_evidence_snapshot_filters_project_evidence_once() { + let temp_dir = TempDir::new().expect("tempdir should create"); + let state_path = temp_dir.path().join("runtime.sqlite3"); + let store = StateStore::open(&state_path).expect("state store should open"); + let first = store + .append_private_execution_event( + "decodex", + "XY-520", + "run-1", + 2, + "evidence_snapshot", + serde_json::json!({"match": true}), + ) + .expect("first private event should append"); + let second = store + .append_private_execution_event( + "decodex", + "XY-520", + "run-1", + 2, + "terminal_finalize", + serde_json::json!({"path": "review_handoff"}), + ) + .expect("second private event should append"); + + store + .append_private_execution_event( + "other", + "XY-520", + "run-1", + 2, + "other_project", + serde_json::json!({"match": false}), + ) + .expect("other project private event should append"); + store + .upsert_review_policy_checkpoint(ReviewPolicyCheckpointInput { + project_id: "decodex", + issue_id: "XY-520", + run_id: "run-1", + attempt_number: 2, + phase: "handoff", + status: "clean", + head_sha: "abc123", + nonclean_rounds: 0, + details_json: "{}", + }) + .expect("review policy checkpoint should persist"); + store + .upsert_review_policy_checkpoint(ReviewPolicyCheckpointInput { + project_id: "other", + issue_id: "XY-520", + run_id: "run-1", + attempt_number: 2, + phase: "handoff", + status: "findings", + head_sha: "def456", + nonclean_rounds: 1, + details_json: "{}", + }) + .expect("other project checkpoint should persist"); + + let snapshot = StateStore::open(&state_path) + .expect("state store should reopen") + .project_loop_evidence_snapshot("decodex") + .expect("project loop evidence should load"); + let events = snapshot.private_events("XY-520", "run-1", 2); + let checkpoint = snapshot + .review_policy_checkpoint("XY-520", "run-1", 2, "handoff") + .expect("matching checkpoint should exist"); + + assert_eq!( + events.iter().map(|event| event.record_id()).collect::>(), + vec![first.record_id(), second.record_id()], + "snapshot should preserve append order and exclude other projects" + ); + assert_eq!(events[1].event_type(), "terminal_finalize"); + assert_eq!(checkpoint.status(), "clean"); + assert!(snapshot.private_events("XY-521", "run-1", 2).is_empty()); +} + #[test] fn private_execution_events_filter_issue_run_attempt_and_stay_out_of_linear_cache() { let store = StateStore::open_in_memory().expect("in-memory state store should open"); @@ -3047,6 +3129,68 @@ fn state_store_open_refreshes_pubfi_project_registry_across_instances() { ); } +#[test] +fn lazy_project_registry_refresh_preserves_runtime_rows() { + let temp_dir = TempDir::new().expect("tempdir should create"); + let state_path = temp_dir.path().join("runtime.db"); + let full_store = StateStore::open(&state_path).expect("state store should open"); + let registration = ProjectRegistration { + service_id: String::from("pubfi"), + config_path: temp_dir.path().join("project.toml"), + repo_root: temp_dir.path().join("repo"), + worktree_root: temp_dir.path().join("repo/.worktrees"), + workflow_path: temp_dir.path().join("repo/WORKFLOW.md"), + tracker_api_key_env_var: String::from("LINEAR_API_KEY_HACKINK"), + github_token_env_var: String::from("GITHUB_PAT_Y"), + enabled: true, + config_fingerprint: String::from("abc123"), + updated_at: String::from("2026-04-29T00:00:00Z"), + updated_at_unix: 1_777_392_000, + }; + let refreshed_registration = ProjectRegistration { + config_fingerprint: String::from("def456"), + updated_at: String::from("2026-04-30T00:00:00Z"), + updated_at_unix: 1_777_478_400, + ..registration.clone() + }; + + full_store.upsert_project(®istration).expect("project should persist"); + full_store.record_run_attempt("run-1", "PUB-101", 1, "running").expect("run should record"); + full_store + .append_event("run-1", 1, "item/agentMessage/delta", "{}") + .expect("event should append"); + full_store + .upsert_worktree( + "pubfi", + "PUB-101", + "x/pub-101", + temp_dir.path().join("repo/.worktrees/PUB-101").to_string_lossy().as_ref(), + ) + .expect("worktree should persist"); + + let lazy_store = StateStore::open_lazy(&state_path).expect("lazy state store should open"); + + lazy_store.upsert_project(&refreshed_registration).expect("project should refresh"); + + let reopened = StateStore::open(&state_path).expect("state store should reopen"); + let attempt = reopened + .latest_run_attempt_for_issue("PUB-101") + .expect("attempt lookup should succeed") + .expect("attempt should survive lazy project refresh"); + let mapping = reopened + .worktree_for_issue("PUB-101") + .expect("worktree lookup should succeed") + .expect("worktree should survive lazy project refresh"); + + assert_eq!(attempt.run_id(), "run-1"); + assert_eq!(reopened.event_count("run-1").expect("event count should survive"), 1); + assert_eq!(mapping.project_id(), "pubfi"); + assert_eq!( + reopened.list_projects().expect("project registry should load")[0].config_fingerprint(), + "def456" + ); +} + #[test] fn remove_project_deletes_persistent_registry_row() { let temp_dir = TempDir::new().expect("tempdir should create"); diff --git a/plugins/decodex/skills/automation/SKILL.md b/plugins/decodex/skills/automation/SKILL.md index 9a2f27fcf..eabeab0f1 100644 --- a/plugins/decodex/skills/automation/SKILL.md +++ b/plugins/decodex/skills/automation/SKILL.md @@ -9,35 +9,42 @@ description: "Use when Decodex owns runtime automation: registered projects, ser Operate Decodex as the retained-lane control plane for automatic development. -Automation starts only after execution authority exists. A natural-language -`research X` request can create a latent Decision Contract, but latent research must -not dispatch retained lanes, set Codex goals, mutate tracker state, or apply queue -labels. A later promotion request such as `arrange this`, `push this forward`, `推进`, -or `做` may feed planning and ready-node queueing only after the accepted contract is -clear. - -## Governing Surfaces - -- `project.toml` under `~/.codex/decodex/projects//` owns repo paths, - service identity, and credential env-var names. -- `WORKFLOW.md` next to that `project.toml` owns execution policy, tracker state names, - validation commands, and context files. -- `docs/spec/runtime.md` owns runtime state and reconciliation rules. -- `docs/spec/tracker-tools.md` owns issue-scoped tracker tool semantics. -- `docs/spec/post-review-lifecycle.md` owns post-`In Review` repair, landing, closeout, - and cleanup phases. -- `docs/spec/lane-control.md` owns CLI/API-first lane-control capabilities, including - inspect, pause/resume, scan, interrupt, steer, retained resume/retry, manual - attention, and deferred controls. -- `docs/runbook/lane-control-recovery.md` owns the post-control decision trees for - agents after interrupt, hard fallback, broad steer, task replacement, or ambiguous - recovery evidence. -- `docs/spec/workflow-file.md` owns `WORKFLOW.md` schema and field semantics. -- `docs/reference/operator-control-plane.md` owns the current status/dashboard field map. - -## Start Sequence - -From an installed runtime: +Automation starts only after execution authority exists. Natural-language `research X` +may create a latent Decision Contract, but latent research must not dispatch retained +lanes, set Codex goals, mutate tracker state, or apply queue labels until a later +promotion request clearly accepts the contract. + +## Read First + +- Read `project.toml` and `WORKFLOW.md` under + `~/.codex/decodex/projects//` before interpreting runtime policy. +- Read `README.md`, `docs/index.md`, and `Makefile.toml` when developing Decodex + itself. +- Read `docs/spec/runtime.md`, `docs/spec/tracker-tools.md`, + `docs/spec/post-review-lifecycle.md`, and `docs/reference/operator-control-plane.md` + when the exact runtime/status contract matters. +- Read `docs/spec/lane-control.md` and `docs/runbook/lane-control-recovery.md` before + interrupting, steering, retrying, resuming, relabeling, or escalating a lane. + +## Core Workflow + +1. Confirm authority: accepted/promoted Decision Contract, explicit human execution + request, or normal issue brief that grants implementation authority. +2. Inspect current state with `decodex status`, `decodex status --json`, or + `decodex lane inspect ` before mutating anything. +3. Queue only ready issues with `decodex:queued:`. Use the `labels` skill + for Decodex Linear labels. + blocked, stale, paused, active, terminal, or unmapped internal nodes remain + unqueued. +4. Request `POST /api/linear-scan` after creating or relabeling queued work when the + scheduler should observe it before the next 5-minute poll. +5. Let retained lanes finish through runtime-owned review, repair, handoff, landing, + closeout, and cleanup unless the operator explicitly moves the lane to a manual + path. + +## Immediate Commands + +Installed runtime: ```sh decodex probe stdio:// @@ -48,181 +55,37 @@ decodex run decodex serve ``` -From the Decodex repo while developing the runtime: +Decodex repo development: ```sh cargo run -p decodex --bin decodex -- probe stdio:// -cargo run -p decodex --bin decodex -- project add "$HOME/.codex/decodex/projects/" cargo run -p decodex --bin decodex -- status cargo run -p decodex --bin decodex -- run --dry-run cargo run -p decodex --bin decodex -- run cargo run -p decodex --bin decodex -- serve ``` -Use `decodex serve --config ` or -`cargo run -p decodex --bin decodex -- serve --config ` when the operator -wants to register that project and start the scheduler in one command. -Use `decodex run ` or `cargo run -p decodex --bin decodex -- run ` only -for a deliberate one-issue automation pass; it still uses the same retained-lane -eligibility and lifecycle rules. -Do not use hidden `serve --dev` for automation. That mode is for isolated local -development: it serves local dashboard/account/app snapshot APIs, but it does not -register projects, poll Linear, or dispatch lanes, and it rejects `--config`. -Decodex App's fallback server uses ordinary `serve` when no compatible local listener -is already running. - -`serve` owns hardcoded scheduler cadences: local operator snapshots publish every -15 seconds, while Linear-backed queue/status scans run at most every 5 minutes per -project. After creating or relabeling queued issues, request the next scan instead of -waiting for the 5-minute window: +## Required Signals -```sh -curl -sS -X POST http://127.0.0.1:8192/api/linear-scan \ - -H 'Content-Type: application/json' \ - -d '{"projectId":""}' -``` - -Omit the JSON body to queue a scan for all enabled projects. The request is consumed -on the next 15-second control-plane tick and still respects tracker rate-limit backoff. - -## Intake and Ownership - -- Automatic intake starts from issues carrying `decodex:queued:`. -- Active lane ownership uses `decodex:active:`. -- `decodex:manual-only` opts an issue out of automation. -- `decodex:needs-attention` marks a human-required stop that automation must not - silently retry. -- Use the `labels` skill before adding, clearing, or interpreting these labels. - -For research-to-execution work, automatic intake also requires: - -- an accepted/promoted Decision Contract or an equivalent explicit human instruction - that grants execution authority -- normal Linear issues with natural-language briefs, acceptance, dependencies, and - validation expectations -- readiness under the registered `WORKFLOW.md`, including startable state, dependency - policy, opt-out labels, active ownership, terminal-state rules, and local capacity -- queue labels only on ready issues; blocked, stale, paused, active, terminal, or - unmapped internal nodes remain unqueued - -Do not expose graph ids, DAG edge editing, hidden goal ids, or queue-label mechanics as -the ordinary user workflow. Operators may inspect status, but automation policy owns -the backstage readiness and intake mechanics. - -## Lane Completion - -The coding agent must leave exactly one terminal path for the leased issue: - -- `review_handoff`, finalized by `issue_terminal_finalize(path = "review_handoff")`. -- `manual_attention`, finalized by `issue_terminal_finalize(path = "manual_attention")`. - -An execution-state checkpoint, a summary message, or a passing local test run is not a -terminal automation signal. - -## Operator Inspection - -- Use `status` and the dashboard to distinguish live execution, retry delay, review - wait, retained repair, closeout, recovery worktrees, and cleanup debt. -- Treat runtime DB rows, app-server protocol activity, and Linear execution-ledger - comments as different evidence surfaces. -- When interpreting history, prefer terminal Run Ledger outcomes and the projected - issue-level `latest_run` status over raw historical attempt rows. A failed raw - attempt that remains in an issue's attempt timeline is diagnostic history, not proof - that the lane is currently blocked, when the Run Ledger shows terminal closeout, - cleanup, or landed completion and the active/backlog/recovery/post-review sections - are empty. -- After an owned agent records `issue_terminal_finalize`, status or lane inspect may - briefly project `phase = terminal_pending` with statuses such as - `review_handoff_pending`, `review_repair_pending`, `closeout_pending`, or - `manual_attention_pending`. Treat this as terminal lifecycle writeback, not active - execution or a stall. Do not hard-interrupt, requeue, or clear labels from that row; - wait for the retained review, closeout, cleanup, or manual-attention path to settle. -- When app-server preflight mentions `skills/list`, distinguish non-blocking scan - diagnostics from real blockers. If the run cwd is present and at least one skill is - enabled, preserve `error_count`, `first_error_path`, and `first_error` as evidence - but do not stop the lane solely because unrelated installed skill metadata failed to - scan. Missing cwd coverage or zero enabled skills remain blockers. -- When status reports loop guardrail reasons, inspect the current recovery state before - treating the lane as human-required. Engineering convergence failures may still be - autonomous after an Architecture Recovery Packet and Authority Boundary Check prove - the next strategy is inside the Authority Envelope. Boundary, external, - uncovered-direction, ownership, insufficient evidence, and exhausted-recovery - outcomes remain human-required stops. -- Before assuming a lane is stuck, compare lane phase, wait reason, last run activity, - protocol activity, active lease state, and child-agent activity when present. - -## Lane Controls - -Read `docs/spec/lane-control.md` before using or explaining operator controls. -Read `docs/runbook/lane-control-recovery.md` before retrying, resuming, relabeling, or -escalating after a control action or ambiguous recovery signal. - -Rules for agents: - -- Inspect first with `decodex lane inspect `, `decodex status`, - `decodex status --json`, `decodex diagnose --json`, `decodex evidence `, or - the dashboard snapshot. Confirm project id, issue id, branch, run id, attempt, - thread/turn evidence, process liveness, tracker state, and PR lineage before mutating - anything. -- Use project dispatch pause/resume only for future intake. `decodex project disable - ` pauses new dispatch; `decodex project enable ` resumes it. - Neither command kills active lanes. -- Request Linear refresh with `POST /api/linear-scan` when a newly queued or relabeled - issue should be observed before the next 5-minute poll. -- Prefer `decodex lane interrupt --run-id ` or - `POST /api/lane/interrupt` for soft `turn/interrupt` when the active turn can be - targeted. Use hard process interruption only with `--force` or `"force": true` and - only as `hard_interrupt_fallback` after soft interrupt is unavailable, timed out, or - impossible. -- Use steer only through the CLI/API lane-control surface and only when the operator - supplies the steer text. The CLI form is `decodex lane steer --run-id - --expected-turn-id --message `; API callers use - canonical `POST /api/lane/steer` or legacy alias `POST /api/lane-steer`. - Bottom-layer steer support is broad; policy, audit, privacy, workflow, recovery, - and skills provide the guardrails. -- Treat task replacement as explicit lifecycle work, not steer. If the operator wants a - different objective or acceptance contract, pause or stop if needed, update/requeue - the issue or create a new lane, and preserve audit evidence. -- Use retained resume/retry through runtime lifecycle paths such as `decodex run - ` only after inspection proves the retained worktree, issue, branch, runtime - evidence, and PR lineage still match. -- Do not expose or use raw `thread/inject_items` as an operator feature. -- Do not mutate Linear tracker state directly to simulate lane controls. During an - owned agent run, use issue-scoped tools for progress, review handoff, manual - attention, and terminal finalization. Outside the owned lane, use documented - CLI/API controls and the labels skill. -- Do not directly kill hidden `_attempt` children or edit runtime DB rows to force a - lane state. Use the supported interrupt, retained retry/resume, recovery, and - manual-attention paths. If an operator had to stop a process for immediate host - safety outside Decodex controls, treat the lane as evidence-ambiguous until - `status`, `diagnose`, `evidence`, and the retained worktree have been inspected. - -Post-control decision tree for automation agents: - -1. Inspect the current lane and private evidence before deciding whether the control - succeeded, failed, timed out, or fell back to `hard_interrupt_fallback`. -2. If the lane is still active and identity still matches the issue, branch, run id, - attempt, and current turn, let the runtime continue or wait for the control result; - do not requeue or clear labels. -3. If the lane is interrupted, failed, or retained with useful local work, resume only - when the retained worktree, branch, issue, runtime evidence, and PR lineage still - prove the same lane. Use runtime lifecycle entrypoints such as `decodex run - `; do not restart from a guessed branch. -4. If a queued or relabeled issue should be observed sooner, request a Linear scan with - `POST /api/linear-scan`. Keep or remove queue labels only through the labels skill - or the supported tracker-tool path for the owned issue. -5. If a broad steer materially changes the requested objective, acceptance criteria, or - issue authority, preserve the local control audit and resolve lifecycle explicitly: - update and requeue the issue, create a new lane, or route the owned run to manual - attention. Do not silently hand off a PR whose diff no longer matches the issue. -6. If evidence cannot prove whether to resume, retry, requeue, or discard retained - work, stop automatic recovery and use manual attention with structured public - blockers. +- Intake starts from `decodex:queued:` and active ownership uses + `decodex:active:`. +- `decodex:manual-only` opts out of automation. +- `decodex:needs-attention` is a human-required stop that automation must not silently + retry. +- A lane is terminal only after exactly one terminal path is finalized: + `review_handoff` or `manual_attention`. +- `phase = terminal_pending` means the agent already called terminal finalize and + Decodex is finishing lifecycle writeback. Do not interrupt, requeue, or manually + clear it from the side. ## Boundaries -- Do not substitute manual `decodex land` for runtime-owned retained-lane landing unless - the operator has explicitly moved the lane to a human-driven landing path. -- Do not directly mutate Linear state outside the issue-scoped tool bridge or the - documented operator procedure. -- Do not infer service-scoped labels from repo name; read the registered project config. +- Do not expose graph ids, DAG edge editing, hidden goal ids, or queue-label mechanics + as the ordinary user workflow. +- Do not directly edit runtime DB rows, kill hidden `_attempt` children, or mutate + Linear state to simulate lane controls. +- Do not substitute manual `decodex land` for runtime-owned retained-lane landing + unless the operator explicitly moves the lane to a human-driven landing path. +- Treat `skills/list` preflight diagnostics as evidence to inspect. Missing cwd + coverage or zero enabled skills are blockers; unrelated installed-skill scan + diagnostics alone are not. diff --git a/plugins/decodex/skills/manual-cli/SKILL.md b/plugins/decodex/skills/manual-cli/SKILL.md index 4e66ef178..3b1f2f987 100644 --- a/plugins/decodex/skills/manual-cli/SKILL.md +++ b/plugins/decodex/skills/manual-cli/SKILL.md @@ -7,28 +7,26 @@ description: "Use when a human is driving Decodex CLI workflows: project registr ## Goal -Use Decodex as a CLI assistant for human-driven development without taking over the +Use Decodex as a CLI assistant for human-driven work without accidentally taking over runtime-owned retained-lane lifecycle. ## Read First -- `README.md` for the current CLI shape. -- `Makefile.toml` before running repo-native checks. -- `docs/spec/lane-control.md` before using CLI/API lane controls. -- `docs/runbook/lane-control-recovery.md` before deciding what to do after interrupt, - hard fallback, broad steer, task replacement, or ambiguous recovery evidence. +- `README.md` for current CLI shape. +- `Makefile.toml` before repository-native checks. - `docs/reference/operator-control-plane.md` when interpreting `status` or dashboard fields. +- `docs/spec/lane-control.md` and `docs/runbook/lane-control-recovery.md` before + interrupting, steering, retrying, resuming, relabeling, or escalating lanes. - `docs/runbook/linear-archive-hygiene.md` before archiving old terminal Linear issues. - `docs/runbook/self-dogfood-pilot.md` for the full self-dogfood operator sequence. ## Command Surface -Use the installed `decodex` binary when the operator is working from an installed -runtime. Use `cargo run -p decodex --bin decodex -- ...` when developing this -repository itself. +Use installed `decodex` when operating an installed runtime. Use +`cargo run -p decodex --bin decodex -- ...` when developing this repository itself. -Common manual checks and dry-run probes: +Common installed-runtime probes: ```sh decodex probe stdio:// @@ -37,150 +35,39 @@ decodex project list decodex status decodex status --live decodex run --dry-run -decodex archive-linear --repo-label repo: --older-than-days 30 ``` -Development equivalents from the Decodex repo root: +Repo-development equivalents: ```sh cargo run -p decodex --bin decodex -- probe stdio:// -cargo run -p decodex --bin decodex -- project add "$HOME/.codex/decodex/projects/" cargo run -p decodex --bin decodex -- status cargo run -p decodex --bin decodex -- status --live cargo run -p decodex --bin decodex -- run --dry-run -cargo run -p decodex --bin decodex -- archive-linear --repo-label repo: --older-than-days 30 ``` -Live `run` commands enter the runtime-owned automation path, even when an operator -starts one pass manually: +## Live Run Warning -```sh -decodex run -decodex run -cargo run -p decodex --bin decodex -- run -cargo run -p decodex --bin decodex -- run -``` +`decodex run` and `decodex run ` enter runtime-owned automation, even when a +human starts one pass manually. Before live run, read the `automation` skill and the +registered project's `WORKFLOW.md`. -Before starting a live run, read the `automation` skill and the registered project's -`WORKFLOW.md`. Treat live `run` as orchestration, not as a status check. - -CLI/API lane controls: - -- Inspect first with `decodex lane inspect `, `decodex status`, - `decodex status --json`, `decodex diagnose --json`, or - `decodex evidence `. -- Use `decodex project disable ` to pause future dispatch for a registered - project, and `decodex project enable ` to resume it. -- Use `POST /api/linear-scan` to request an intake/status refresh before the next - scheduled Linear poll. -- Use `decodex run ` only for deliberate one-issue automation or retained - retry/resume when the lane remains eligible under the registered workflow. -- Use `decodex lane interrupt --run-id ` for soft active-turn - interruption. Add `--force` only when explicit operator intent allows hard - process-kill fallback after soft interrupt is unavailable or fails. -- Use `decodex lane steer --run-id --expected-turn-id - --message ` only with explicit operator-supplied text after inspection proves - the active lane identity. The expected turn id is a fail-closed precondition. - Broad steer text is allowed at the bottom layer; lifecycle ambiguity should route to - explicit recovery or manual attention instead of silently changing the issue - contract. -- Do not use active-lane UI controls, direct runtime DB edits, raw - `thread/inject_items`, or tracker-state mutations as substitutes for the lane-control - contract. -- Do not kill hidden `_attempt` children to simulate interrupt. Use - `decodex lane interrupt ... --force` or the API `"force": true` only when explicit - operator intent allows hard fallback. If an emergency host-safety stop happens - outside Decodex controls, inspect local evidence and route recovery explicitly before - retrying or cleaning labels. - -Post-control CLI recovery: - -1. Inspect again with `decodex lane inspect `, `decodex status --json`, and - `decodex evidence ` when a control request returns, times out, or reports - `hard_interrupt_fallback`. -2. If identity still matches an active lane, wait for the runtime-owned attempt or use - the next supported control. Do not remove `decodex:active:` by hand. -3. If the lane is retained and lineage is exact, use the registered workflow path such - as `decodex run ` for retry/resume. If status reports a retained review - handoff mismatch, use `docs/runbook/recover-review-handoff.md`. -4. If status reports loop guardrail or architecture recovery state, inspect private - evidence before clearing attention labels or retrying. Autonomous recovery is valid - only when the Authority Boundary Check is `within_authority` and recovery budget - remains for an engineering convergence reason. Boundary, dependency, - uncovered-direction, ownership/lineage ambiguity, external, - insufficient-evidence, and exhausted-recovery outcomes are not cleared by - Authority Boundary Check alone; follow `docs/runbook/lane-control-recovery.md` - for the reason-specific resume rule before returning the lane to automation. -5. If the operator changed labels or issue state and wants the scheduler to notice - before the next poll, request `POST /api/linear-scan`; this is a refresh request, - not a retry command. -6. If the new operator text replaces the task or changes acceptance materially, do not - hide that as steer. Resolve the old lane explicitly, then update/requeue the same - issue or create a new issue for the replacement work. -7. If the evidence is ambiguous or useful retained work would be overwritten, route to - manual attention instead of direct Linear label mutation. - -Manual commit and landing are separate narrow workflows: +## Manual Lifecycle - Use `commit` before creating a Decodex-formatted local commit. - Use `land` only when the user asks to land a human-driven PR through Decodex. - -## Project Registration - -- `project add` registers or refreshes a project directory in the local runtime DB. -- The project directory must contain `project.toml` and `WORKFLOW.md`. -- `project.toml` owns `[paths].repo_root`, service identity, and credential env-var - names. -- Project-scoped commands without `--config` resolve through the explicit registry; - they do not scan arbitrary checkouts for repo-local config files. - -## Status and Dry Run - -- Use `status` to inspect the local runtime snapshot for active lanes, retained local - state, recovery worktrees, account-pool configuration, and the run ledger without - refreshing live tracker, pull-request, or ChatGPT account usage observers. -- For completed issue history, treat the Run Ledger outcome and projected - `history_lanes[].latest_run.status` as the primary issue-level result. Raw failed - attempts under `history_lanes[].attempts` are retained for diagnosis and should not - be read as current lane failure when the primary sections show no active, queued, - recovery, or post-review work for that issue. -- If `status` or `lane inspect` shows `phase = terminal_pending`, the agent already - called `issue_terminal_finalize` and Decodex is finishing terminal lifecycle - writeback. Statuses such as `review_handoff_pending`, `review_repair_pending`, - `closeout_pending`, and `manual_attention_pending` are not active lanes and should - not be interrupted, requeued, or manually cleaned up from the side. -- Use `status --live` when the operator needs fresh Linear/GitHub observer readback - before acting. Use `/api/accounts?refresh=1` for fresh ChatGPT account usage probes. -- Use `run --dry-run` before live automation to validate project loading, issue - discovery, eligibility, and worktree planning without tracker mutation. +- Use `run --dry-run` only as an intake/worktree-planning check. It does not prove live + tracker writes, PR handoff, closeout, or app-server execution will succeed. - Use `probe stdio://` before relying on the Codex app-server boundary. -- Treat hidden `serve --dev` as isolated local-development infrastructure only. It - serves dashboard, account, and app snapshot APIs, but it does not register projects, - poll Linear, dispatch work, or accept `--config`. Decodex App's fallback server uses - ordinary `serve` when no compatible local listener is already running. -- `serve` has no interval override. It publishes local operator snapshots every - 15 seconds and runs Linear-backed queue/status scans at most every 5 minutes per - project. -- After creating or relabeling queue issues, request the next scan instead of waiting - for the 5-minute Linear poll: - - ```sh - curl -sS -X POST http://127.0.0.1:8192/api/linear-scan \ - -H 'Content-Type: application/json' \ - -d '{"projectId":""}' - ``` - - Omit the JSON body to scan all enabled projects. The request is consumed on the next - 15-second control-plane tick and still respects tracker rate-limit backoff. -- For `skills/list` app-server preflight output, enabled skills plus scan diagnostics - are local evidence, not a lane blocker. Missing cwd coverage or zero enabled skills - are blockers; inspect `first_error_path` and `first_error` before changing plugin or - skill installs. +- Use `POST /api/linear-scan` after label or issue-state changes when the scheduler + should refresh before its next 5-minute Linear poll. ## Boundaries -- Do not treat `run --dry-run` as proof that a live run can complete tracker writes, - PR handoff, or closeout. - Do not hand-edit runtime DB state unless a runbook explicitly says to. -- Do not clean up retained automation worktrees from the side when `status` shows a - live or recovery-owned lane. +- Do not clean up retained automation worktrees from the side when status shows a live + or recovery-owned lane. +- Do not kill hidden `_attempt` children or directly mutate Linear tracker state to + simulate lane controls. +- Do not treat app-server `skills/list` diagnostics as blockers without checking + whether cwd coverage and enabled skills are actually missing.