From 0dfb39376cb30c68db25e65f196e0dc8b422d5ff Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Wed, 10 Jun 2026 01:46:13 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Implement Execution Program readiness model","authority":"XY-853"} --- apps/decodex/src/execution_program.rs | 1518 +++++++++++++++++ apps/decodex/src/lib.rs | 1 + apps/decodex/src/orchestrator.rs | 2 +- apps/decodex/src/orchestrator/entrypoints.rs | 2 + apps/decodex/src/orchestrator/status.rs | 79 + apps/decodex/src/orchestrator/tests.rs | 2 +- .../tests/operator/status/agent_evidence.rs | 1 + .../tests/operator/status/text.rs | 51 + apps/decodex/src/orchestrator/types.rs | 57 + apps/decodex/src/state.rs | 1 + apps/decodex/src/state/internal.rs | 206 ++- apps/decodex/src/state/models.rs | 46 + apps/decodex/src/state/store.rs | 132 ++ apps/decodex/src/state/tests.rs | 112 ++ docs/reference/operator-control-plane.md | 6 +- docs/spec/linear-execution-ledger.md | 7 +- docs/spec/loop-runtime.md | 63 +- docs/spec/runtime.md | 8 +- docs/spec/tracker-tools.md | 4 + 19 files changed, 2273 insertions(+), 25 deletions(-) create mode 100644 apps/decodex/src/execution_program.rs diff --git a/apps/decodex/src/execution_program.rs b/apps/decodex/src/execution_program.rs new file mode 100644 index 000000000..f51df7c23 --- /dev/null +++ b/apps/decodex/src/execution_program.rs @@ -0,0 +1,1518 @@ +//! Internal Execution Program model and readiness evaluator. + +#![allow(dead_code)] + +use std::collections::{BTreeMap, HashSet}; + +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::{ + loop_contract::{DecisionContract, DecisionContractStatus}, + prelude::{Result, eyre}, + tracker, + workflow::WorkflowDocument, +}; + +pub(crate) const EXECUTION_PROGRAM_SCHEMA: &str = "decodex.execution_program/1"; +pub(crate) const EXECUTION_PROGRAM_RECORD_VERSION: u16 = 1; + +/// Stage for one internal Execution Program node. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ExecutionProgramNodeStage { + /// Research or evidence-gathering work. + Research, + /// Design or architecture-shaping work. + Design, + /// Normative specification work. + Spec, + /// Runtime schema, storage, or serialization work. + Schema, + /// Runtime implementation work. + Runtime, + /// Agent/plugin skill or integration work. + Plugin, + /// Evaluation, harness, or validation work. + Eval, + /// Review, PR, delivery, or handoff work. + Handoff, +} + +/// Queue intent for one internal Execution Program node. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ExecutionQueueIntent { + /// The node is intentionally not ready for queueing. + NotReady, + /// The node is ready to receive the service queue label once mapped to a startable issue. + ReadyToQueue, + /// The node should retain the service queue label while it remains startable. + Queued, + /// The node is already active in a lane. + Active, + /// The node is intentionally paused. + Paused, + /// The node is complete. + Done, + /// The node was canceled. + Canceled, +} +impl ExecutionQueueIntent { + fn is_terminal(self) -> bool { + matches!(self, Self::Done | Self::Canceled) + } +} + +/// Conflict-domain class for one program node. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ExecutionConflictDomainKind { + /// A concrete file or path family. + File, + /// A module, crate, package, or app surface. + Module, + /// Local runtime or repository state. + State, + /// Credential, account, or auth-owned surface. + Credentials, + /// Tracker ownership, labels, comments, or workflow state. + TrackerOwnership, + /// Pull request, review, or landing surface. + ReviewSurface, +} +impl ExecutionConflictDomainKind { + fn as_str(self) -> &'static str { + match self { + Self::File => "file", + Self::Module => "module", + Self::State => "state", + Self::Credentials => "credentials", + Self::TrackerOwnership => "tracker_ownership", + Self::ReviewSurface => "review_surface", + } + } +} + +/// Normalized readiness state for one program node. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ExecutionReadinessState { + /// Node is intentionally not ready yet. + NotReady, + /// Node is startable and may be mapped to queue action. + Ready, + /// Node cannot start until a concrete blocker clears. + Blocked, + /// Node is intentionally paused. + Paused, + /// Node is already active and should not retain the queue label. + Active, + /// Node is terminal. + Completed, + /// Node no longer matches the accepted contract. + Stale, +} +impl ExecutionReadinessState { + /// Stable machine-readable state name. + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::NotReady => "not_ready", + Self::Ready => "ready", + Self::Blocked => "blocked", + Self::Paused => "paused", + Self::Active => "active", + Self::Completed => "completed", + Self::Stale => "stale", + } + } +} + +/// Queue-label action allowed for a mapped node. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ExecutionQueueLabelAction { + /// Apply the service queue label. + Apply, + /// Retain the service queue label. + Retain, + /// Remove the service queue label. + Remove, +} + +/// Conflict-domain key for one program node. +#[derive(Clone, Debug, Eq, Hash, PartialEq, Deserialize, Serialize)] +pub(crate) struct ExecutionConflictDomain { + kind: ExecutionConflictDomainKind, + key: String, +} +impl ExecutionConflictDomain { + /// Build a conflict-domain key. + pub(crate) fn new(kind: ExecutionConflictDomainKind, key: impl Into) -> Result { + let domain = Self { kind, key: key.into() }; + + domain.validate()?; + + Ok(domain) + } + + /// Stable conflict-domain key. + pub(crate) fn key(&self) -> &str { + &self.key + } + + fn validate(&self) -> Result<()> { + validate_required("execution program conflict_domain.key", &self.key) + } +} + +/// Dependency edge for one program node. +#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)] +pub(crate) struct ExecutionProgramDependency { + dependency_id: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + required_terminal_states: Vec, +} +impl ExecutionProgramDependency { + /// Build a dependency edge using the registered workflow terminal states. + pub(crate) fn new(dependency_id: impl Into) -> Result { + let dependency = + Self { dependency_id: dependency_id.into(), required_terminal_states: Vec::new() }; + + dependency.validate()?; + + Ok(dependency) + } + + /// Override the terminal tracker states that satisfy this dependency. + pub(crate) fn with_required_terminal_states( + mut self, + states: impl IntoIterator>, + ) -> Result { + self.required_terminal_states = states.into_iter().map(Into::into).collect(); + + self.validate()?; + + Ok(self) + } + + /// Dependency node or issue identifier. + pub(crate) fn dependency_id(&self) -> &str { + &self.dependency_id + } + + fn validate(&self) -> Result<()> { + validate_required("execution program dependency.dependency_id", &self.dependency_id)?; + + validate_string_list( + "execution program dependency.required_terminal_states", + &self.required_terminal_states, + ) + } +} + +/// Normal Linear issue mapping for an executable program node. +#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)] +pub(crate) struct ExecutionLinearIssueMapping { + issue_id: String, + issue_identifier: String, + issue_state: String, + has_queue_label: bool, + has_active_label: bool, + has_opt_out_label: bool, + has_needs_attention_label: bool, + has_generic_dispatch_briefing: bool, +} +impl ExecutionLinearIssueMapping { + /// Build a Linear issue mapping with no automation labels and a generic dispatch briefing. + pub(crate) fn new( + issue_id: impl Into, + issue_identifier: impl Into, + issue_state: impl Into, + ) -> Result { + let mapping = Self { + issue_id: issue_id.into(), + issue_identifier: issue_identifier.into(), + issue_state: issue_state.into(), + has_queue_label: false, + has_active_label: false, + has_opt_out_label: false, + has_needs_attention_label: false, + has_generic_dispatch_briefing: true, + }; + + mapping.validate()?; + + Ok(mapping) + } + + /// Mark whether the issue currently carries the service queue label. + pub(crate) fn with_queue_label(mut self, present: bool) -> Self { + self.has_queue_label = present; + + self + } + + /// Mark whether the issue currently carries the service active label. + pub(crate) fn with_active_label(mut self, present: bool) -> Self { + self.has_active_label = present; + + self + } + + /// Mark whether the issue currently carries the opt-out label. + pub(crate) fn with_opt_out_label(mut self, present: bool) -> Self { + self.has_opt_out_label = present; + + self + } + + /// Mark whether the issue currently carries the needs-attention label. + pub(crate) fn with_needs_attention_label(mut self, present: bool) -> Self { + self.has_needs_attention_label = present; + + self + } + + /// Mark whether the issue description remains a generic dispatch briefing. + pub(crate) fn with_generic_dispatch_briefing(mut self, present: bool) -> Self { + self.has_generic_dispatch_briefing = present; + + self + } + + /// Linear issue identifier such as `XY-853`. + pub(crate) fn issue_identifier(&self) -> &str { + &self.issue_identifier + } + + /// Tracker workflow state for the mapped issue. + pub(crate) fn issue_state(&self) -> &str { + &self.issue_state + } + + /// Whether the service queue label is currently present. + pub(crate) fn has_queue_label(&self) -> bool { + self.has_queue_label + } + + fn validate(&self) -> Result<()> { + validate_required("execution program issue_mapping.issue_id", &self.issue_id)?; + validate_required( + "execution program issue_mapping.issue_identifier", + &self.issue_identifier, + )?; + + validate_required("execution program issue_mapping.issue_state", &self.issue_state) + } +} + +/// Internal node in an Execution Program. +#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)] +pub(crate) struct ExecutionProgramNode { + node_id: String, + stage: ExecutionProgramNodeStage, + objective: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + objective_lineage: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + dependencies: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + conflict_domains: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + acceptance_expectations: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + validation_expectations: Vec, + queue_intent: ExecutionQueueIntent, + #[serde(skip_serializing_if = "Option::is_none")] + linear_issue: Option, + contract_fingerprint: String, +} +impl ExecutionProgramNode { + /// Build a program node. + pub(crate) fn new( + node_id: impl Into, + stage: ExecutionProgramNodeStage, + objective: impl Into, + queue_intent: ExecutionQueueIntent, + ) -> Result { + let node = Self { + node_id: node_id.into(), + stage, + objective: objective.into(), + objective_lineage: Vec::new(), + dependencies: Vec::new(), + conflict_domains: Vec::new(), + acceptance_expectations: Vec::new(), + validation_expectations: Vec::new(), + queue_intent, + linear_issue: None, + contract_fingerprint: String::new(), + }; + + node.validate()?; + + Ok(node) + } + + /// Add objective-lineage text from the accepted contract. + pub(crate) fn with_objective_lineage( + mut self, + lineage: impl IntoIterator>, + ) -> Result { + self.objective_lineage = lineage.into_iter().map(Into::into).collect(); + + self.validate()?; + + Ok(self) + } + + /// Add dependencies. + pub(crate) fn with_dependencies( + mut self, + dependencies: impl IntoIterator, + ) -> Result { + self.dependencies = dependencies.into_iter().collect(); + + self.validate()?; + + Ok(self) + } + + /// Add conflict domains. + pub(crate) fn with_conflict_domains( + mut self, + conflict_domains: impl IntoIterator, + ) -> Result { + self.conflict_domains = conflict_domains.into_iter().collect(); + + self.validate()?; + + Ok(self) + } + + /// Add acceptance expectations. + pub(crate) fn with_acceptance_expectations( + mut self, + expectations: impl IntoIterator>, + ) -> Result { + self.acceptance_expectations = expectations.into_iter().map(Into::into).collect(); + + self.validate()?; + + Ok(self) + } + + /// Add validation expectations. + pub(crate) fn with_validation_expectations( + mut self, + expectations: impl IntoIterator>, + ) -> Result { + self.validation_expectations = expectations.into_iter().map(Into::into).collect(); + + self.validate()?; + + Ok(self) + } + + /// Link the node to a normal Linear issue. + pub(crate) fn with_linear_issue(mut self, issue: ExecutionLinearIssueMapping) -> Result { + self.linear_issue = Some(issue); + + self.validate()?; + + Ok(self) + } + + /// Override the accepted-contract fingerprint used for drift detection. + pub(crate) fn with_contract_fingerprint( + mut self, + fingerprint: impl Into, + ) -> Result { + self.contract_fingerprint = fingerprint.into(); + + self.validate()?; + + Ok(self) + } + + /// Stable internal node id. + pub(crate) fn node_id(&self) -> &str { + &self.node_id + } + + /// Node queue intent. + pub(crate) fn queue_intent(&self) -> ExecutionQueueIntent { + self.queue_intent + } + + /// Linked normal Linear issue, when the node is executable. + pub(crate) fn linear_issue(&self) -> Option<&ExecutionLinearIssueMapping> { + self.linear_issue.as_ref() + } + + fn bind_contract_fingerprint(&mut self, fingerprint: &str) { + if self.contract_fingerprint.is_empty() { + self.contract_fingerprint = fingerprint.to_owned(); + } + } + + fn validate(&self) -> Result<()> { + validate_required("execution program node.node_id", &self.node_id)?; + validate_required("execution program node.objective", &self.objective)?; + validate_string_list("execution program node.objective_lineage", &self.objective_lineage)?; + validate_string_list( + "execution program node.acceptance_expectations", + &self.acceptance_expectations, + )?; + validate_string_list( + "execution program node.validation_expectations", + &self.validation_expectations, + )?; + validate_optional( + "execution program node.contract_fingerprint", + non_empty_optional(&self.contract_fingerprint), + )?; + + for dependency in &self.dependencies { + dependency.validate()?; + } + for domain in &self.conflict_domains { + domain.validate()?; + } + + if let Some(issue) = &self.linear_issue { + issue.validate()?; + } + + Ok(()) + } +} + +/// Versioned internal Execution Program derived from an accepted Decision Contract. +#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)] +pub(crate) struct ExecutionProgram { + #[serde(default = "execution_program_schema")] + schema: String, + #[serde(default = "execution_program_record_version")] + record_version: u16, + program_id: String, + service_id: String, + source_contract_id: String, + accepted_contract_fingerprint: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + nodes: Vec, +} +impl ExecutionProgram { + /// Build an internal Execution Program from an accepted Decision Contract. + pub(crate) fn from_accepted_contract( + program_id: impl Into, + service_id: impl Into, + contract: &DecisionContract, + mut nodes: Vec, + ) -> Result { + ensure_accepted_contract(contract)?; + + let fingerprint = decision_contract_fingerprint(contract)?; + + for node in &mut nodes { + node.bind_contract_fingerprint(&fingerprint); + } + + let program = Self { + schema: execution_program_schema(), + record_version: EXECUTION_PROGRAM_RECORD_VERSION, + program_id: program_id.into(), + service_id: service_id.into(), + source_contract_id: contract.contract_id().to_owned(), + accepted_contract_fingerprint: fingerprint, + nodes, + }; + + program.validate()?; + + Ok(program) + } + + /// Stable internal program id. + pub(crate) fn program_id(&self) -> &str { + &self.program_id + } + + /// Service id that owns queue-label decisions. + pub(crate) fn service_id(&self) -> &str { + &self.service_id + } + + /// Accepted Decision Contract id that authorized this program. + pub(crate) fn source_contract_id(&self) -> &str { + &self.source_contract_id + } + + /// Program nodes. + pub(crate) fn nodes(&self) -> &[ExecutionProgramNode] { + &self.nodes + } + + /// Evaluate every node against the current contract, workflow policy, and runtime context. + pub(crate) fn evaluate( + &self, + current_contract: &DecisionContract, + policy: &ExecutionWorkflowPolicy, + context: &ExecutionProgramReadinessContext, + ) -> Result { + self.validate()?; + + if self.service_id != policy.service_id { + eyre::bail!( + "Execution program `{}` belongs to service `{}` but readiness policy belongs to `{}`.", + self.program_id, + self.service_id, + policy.service_id + ); + } + + let current_fingerprint = decision_contract_fingerprint(current_contract)?; + let node_lookup = + self.nodes.iter().map(|node| (node.node_id.as_str(), node)).collect::>(); + let dependency_lookup = context.dependency_lookup(); + let occupied_conflicts = context.occupied_conflict_domains.iter().collect::>(); + let mut nodes = Vec::new(); + + for node in &self.nodes { + nodes.push(evaluate_node(EvaluateNodeInput { + program: self, + node, + current_contract, + current_fingerprint: ¤t_fingerprint, + policy, + node_lookup: &node_lookup, + dependency_lookup: &dependency_lookup, + occupied_conflicts: &occupied_conflicts, + })?); + } + + Ok(ExecutionProgramEvaluation { program_id: self.program_id.clone(), nodes }) + } + + /// Validate the serialized program payload. + pub(crate) fn validate(&self) -> Result<()> { + validate_required("execution program schema", &self.schema)?; + validate_required("execution program program_id", &self.program_id)?; + validate_required("execution program service_id", &self.service_id)?; + validate_required("execution program source_contract_id", &self.source_contract_id)?; + validate_required( + "execution program accepted_contract_fingerprint", + &self.accepted_contract_fingerprint, + )?; + + if self.schema != EXECUTION_PROGRAM_SCHEMA { + eyre::bail!( + "Execution program `{}` has unsupported schema `{}`.", + self.program_id, + self.schema + ); + } + if self.record_version != EXECUTION_PROGRAM_RECORD_VERSION { + eyre::bail!( + "Execution program `{}` has unsupported record_version `{}`.", + self.program_id, + self.record_version + ); + } + + let mut node_ids = HashSet::new(); + + for node in &self.nodes { + node.validate()?; + + if !node_ids.insert(node.node_id.as_str()) { + eyre::bail!( + "Execution program `{}` contains duplicate node `{}`.", + self.program_id, + node.node_id + ); + } + } + for node in &self.nodes { + for dependency in &node.dependencies { + if !node_ids.contains(dependency.dependency_id.as_str()) { + eyre::bail!( + "Execution program node `{}` depends on unknown node `{}`.", + node.node_id, + dependency.dependency_id + ); + } + } + } + + Ok(()) + } +} + +/// Workflow policy needed for Execution Program readiness. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ExecutionWorkflowPolicy { + service_id: String, + queue_label: String, + active_label: String, + startable_states: Vec, + terminal_states: Vec, + opt_out_label: String, + needs_attention_label: String, +} +impl ExecutionWorkflowPolicy { + /// Build readiness policy from the registered project workflow. + pub(crate) fn from_workflow(service_id: &str, workflow: &WorkflowDocument) -> Result { + Self::new( + service_id, + workflow.frontmatter().tracker().startable_states().to_vec(), + workflow.frontmatter().tracker().terminal_states().to_vec(), + workflow.frontmatter().tracker().opt_out_label().to_owned(), + workflow.frontmatter().tracker().needs_attention_label().to_owned(), + ) + } + + /// Build readiness policy directly. + pub(crate) fn new( + service_id: impl Into, + startable_states: Vec, + terminal_states: Vec, + opt_out_label: impl Into, + needs_attention_label: impl Into, + ) -> Result { + let service_id = service_id.into(); + let policy = Self { + queue_label: tracker::automation_queue_label(&service_id), + active_label: tracker::automation_active_label(&service_id), + service_id, + startable_states, + terminal_states, + opt_out_label: opt_out_label.into(), + needs_attention_label: needs_attention_label.into(), + }; + + policy.validate()?; + + Ok(policy) + } + + /// Service-scoped queue label. + pub(crate) fn queue_label(&self) -> &str { + &self.queue_label + } + + /// Workflow terminal states. + pub(crate) fn terminal_states(&self) -> &[String] { + &self.terminal_states + } + + fn issue_is_startable(&self, issue: &ExecutionLinearIssueMapping) -> bool { + self.startable_states.iter().any(|state| state == issue.issue_state()) + } + + fn issue_is_terminal(&self, issue: &ExecutionLinearIssueMapping) -> bool { + self.terminal_states.iter().any(|state| state == issue.issue_state()) + } + + fn validate(&self) -> Result<()> { + validate_required("execution workflow service_id", &self.service_id)?; + validate_required("execution workflow queue_label", &self.queue_label)?; + validate_required("execution workflow active_label", &self.active_label)?; + validate_required("execution workflow opt_out_label", &self.opt_out_label)?; + validate_required("execution workflow needs_attention_label", &self.needs_attention_label)?; + validate_string_list("execution workflow startable_states", &self.startable_states)?; + validate_string_list("execution workflow terminal_states", &self.terminal_states)?; + + if self.startable_states.is_empty() { + eyre::bail!("Execution workflow startable_states must not be empty."); + } + if self.terminal_states.is_empty() { + eyre::bail!("Execution workflow terminal_states must not be empty."); + } + + Ok(()) + } +} + +/// Runtime dependency observation used by readiness evaluation. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ExecutionDependencySnapshot { + dependency_id: String, + tracker_state: Option, + queue_intent: Option, +} +impl ExecutionDependencySnapshot { + /// Observe a dependency through a tracker state. + pub(crate) fn tracker_state( + dependency_id: impl Into, + state: impl Into, + ) -> Result { + let snapshot = Self { + dependency_id: dependency_id.into(), + tracker_state: Some(state.into()), + queue_intent: None, + }; + + snapshot.validate()?; + + Ok(snapshot) + } + + /// Observe a dependency through another internal node queue intent. + pub(crate) fn queue_intent( + dependency_id: impl Into, + queue_intent: ExecutionQueueIntent, + ) -> Result { + let snapshot = Self { + dependency_id: dependency_id.into(), + tracker_state: None, + queue_intent: Some(queue_intent), + }; + + snapshot.validate()?; + + Ok(snapshot) + } + + fn validate(&self) -> Result<()> { + validate_required("execution dependency snapshot.dependency_id", &self.dependency_id)?; + + validate_optional( + "execution dependency snapshot.tracker_state", + self.tracker_state.as_deref(), + ) + } +} + +/// Runtime context supplied to readiness evaluation. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub(crate) struct ExecutionProgramReadinessContext { + dependency_snapshots: Vec, + occupied_conflict_domains: Vec, +} +impl ExecutionProgramReadinessContext { + /// Build an empty readiness context. + pub(crate) fn new() -> Self { + Self::default() + } + + /// Add dependency observations. + pub(crate) fn with_dependency_snapshots( + mut self, + snapshots: impl IntoIterator, + ) -> Self { + self.dependency_snapshots = snapshots.into_iter().collect(); + + self + } + + /// Add conflict domains already occupied by active or retained work. + pub(crate) fn with_occupied_conflict_domains( + mut self, + domains: impl IntoIterator, + ) -> Self { + self.occupied_conflict_domains = domains.into_iter().collect(); + + self + } + + fn dependency_lookup(&self) -> BTreeMap<&str, &ExecutionDependencySnapshot> { + self.dependency_snapshots + .iter() + .map(|snapshot| (snapshot.dependency_id.as_str(), snapshot)) + .collect() + } +} + +/// Readiness result for one program node. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ExecutionNodeEvaluation { + node_id: String, + state: ExecutionReadinessState, + reasons: Vec, + queue_label_action: Option, + linear_issue: Option, +} +impl ExecutionNodeEvaluation { + /// Node id. + pub(crate) fn node_id(&self) -> &str { + &self.node_id + } + + /// Normalized readiness state. + pub(crate) fn state(&self) -> ExecutionReadinessState { + self.state + } + + /// Human-readable readiness reasons. + pub(crate) fn reasons(&self) -> &[String] { + &self.reasons + } + + /// Queue-label action, if any. + pub(crate) fn queue_label_action(&self) -> Option { + self.queue_label_action + } + + /// Whether this node may receive or retain the service queue label. + pub(crate) fn queue_label_eligible(&self) -> bool { + matches!( + self.queue_label_action, + Some(ExecutionQueueLabelAction::Apply | ExecutionQueueLabelAction::Retain) + ) + } + + /// Mapped Linear issue, when present. + pub(crate) fn linear_issue(&self) -> Option<&ExecutionLinearIssueMapping> { + self.linear_issue.as_ref() + } +} + +/// Full readiness result for one Execution Program. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ExecutionProgramEvaluation { + program_id: String, + nodes: Vec, +} +impl ExecutionProgramEvaluation { + /// Node evaluations. + pub(crate) fn nodes(&self) -> &[ExecutionNodeEvaluation] { + &self.nodes + } + + /// Nodes that are internally ready. + pub(crate) fn ready_node_ids(&self) -> Vec<&str> { + self.nodes + .iter() + .filter(|node| node.state == ExecutionReadinessState::Ready) + .map(|node| node.node_id.as_str()) + .collect() + } + + /// Nodes that may receive or retain the service queue label. + pub(crate) fn startable_node_ids(&self) -> Vec<&str> { + self.nodes + .iter() + .filter(|node| node.queue_label_eligible()) + .map(|node| node.node_id.as_str()) + .collect() + } + + /// Operator-facing progress summary without exposing graph operations as workflow. + pub(crate) fn operator_summary(&self) -> ExecutionProgramOperatorSummary { + let mut summary = ExecutionProgramOperatorSummary { + program_id: self.program_id.clone(), + ready_count: 0, + blocked_count: 0, + paused_count: 0, + active_count: 0, + completed_count: 0, + stale_count: 0, + queue_label_eligible_count: 0, + mapped_issue_identifiers: Vec::new(), + }; + + for node in &self.nodes { + match node.state { + ExecutionReadinessState::Ready => summary.ready_count += 1, + ExecutionReadinessState::Blocked | ExecutionReadinessState::NotReady => + summary.blocked_count += 1, + ExecutionReadinessState::Paused => summary.paused_count += 1, + ExecutionReadinessState::Active => summary.active_count += 1, + ExecutionReadinessState::Completed => summary.completed_count += 1, + ExecutionReadinessState::Stale => summary.stale_count += 1, + } + + if node.queue_label_eligible() { + summary.queue_label_eligible_count += 1; + } + + if let Some(issue) = &node.linear_issue { + summary.mapped_issue_identifiers.push(issue.issue_identifier.clone()); + } + } + + summary.mapped_issue_identifiers.sort(); + summary.mapped_issue_identifiers.dedup(); + + summary + } +} + +/// Compact operator readback for Execution Program progress. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ExecutionProgramOperatorSummary { + /// Program id. + pub(crate) program_id: String, + /// Count of ready nodes. + pub(crate) ready_count: usize, + /// Count of blocked or intentionally not-ready nodes. + pub(crate) blocked_count: usize, + /// Count of paused nodes. + pub(crate) paused_count: usize, + /// Count of active nodes. + pub(crate) active_count: usize, + /// Count of done or canceled nodes. + pub(crate) completed_count: usize, + /// Count of stale nodes. + pub(crate) stale_count: usize, + /// Count of nodes eligible to receive or retain the service queue label. + pub(crate) queue_label_eligible_count: usize, + /// Normal Linear issue identifiers linked to the program. + pub(crate) mapped_issue_identifiers: Vec, +} + +struct EvaluateNodeInput<'a> { + program: &'a ExecutionProgram, + node: &'a ExecutionProgramNode, + current_contract: &'a DecisionContract, + current_fingerprint: &'a str, + policy: &'a ExecutionWorkflowPolicy, + node_lookup: &'a BTreeMap<&'a str, &'a ExecutionProgramNode>, + dependency_lookup: &'a BTreeMap<&'a str, &'a ExecutionDependencySnapshot>, + occupied_conflicts: &'a HashSet<&'a ExecutionConflictDomain>, +} + +fn evaluate_node(input: EvaluateNodeInput<'_>) -> Result { + let EvaluateNodeInput { + program, + node, + current_contract, + current_fingerprint, + policy, + node_lookup, + dependency_lookup, + occupied_conflicts, + } = input; + let mut reasons = Vec::new(); + let mut state = ExecutionReadinessState::Ready; + + if current_contract.status() != DecisionContractStatus::AcceptedPromoted + || current_contract.contract_id() != program.source_contract_id + || current_fingerprint != program.accepted_contract_fingerprint + || current_fingerprint != node.contract_fingerprint + { + state = ExecutionReadinessState::Stale; + + reasons.push(String::from("node no longer matches the accepted Decision Contract")); + } else { + match node.queue_intent { + ExecutionQueueIntent::NotReady => { + state = ExecutionReadinessState::NotReady; + + reasons.push(String::from("node queue intent is not-ready")); + }, + ExecutionQueueIntent::Paused => { + state = ExecutionReadinessState::Paused; + + reasons.push(String::from("node queue intent is paused")); + }, + ExecutionQueueIntent::Active => { + state = ExecutionReadinessState::Active; + + reasons.push(String::from("node already has an active lane")); + }, + ExecutionQueueIntent::Done | ExecutionQueueIntent::Canceled => { + state = ExecutionReadinessState::Completed; + + reasons.push(String::from("node queue intent is terminal")); + }, + ExecutionQueueIntent::ReadyToQueue | ExecutionQueueIntent::Queued => { + collect_blocking_readiness_reasons( + node, + policy, + node_lookup, + dependency_lookup, + occupied_conflicts, + &mut reasons, + ); + + if !reasons.is_empty() { + state = ExecutionReadinessState::Blocked; + } + }, + } + } + if state == ExecutionReadinessState::Ready { + reasons.push(String::from("node is ready for normal Linear issue execution")); + } + + let queue_label_action = queue_label_action_for(node, state, policy); + + Ok(ExecutionNodeEvaluation { + node_id: node.node_id.clone(), + state, + reasons, + queue_label_action, + linear_issue: node.linear_issue.clone(), + }) +} + +fn collect_blocking_readiness_reasons( + node: &ExecutionProgramNode, + policy: &ExecutionWorkflowPolicy, + node_lookup: &BTreeMap<&str, &ExecutionProgramNode>, + dependency_lookup: &BTreeMap<&str, &ExecutionDependencySnapshot>, + occupied_conflicts: &HashSet<&ExecutionConflictDomain>, + reasons: &mut Vec, +) { + if node.acceptance_expectations.is_empty() { + reasons.push(String::from("node has no acceptance expectations")); + } + if node.validation_expectations.is_empty() { + reasons.push(String::from("node has no validation expectations")); + } + + for dependency in &node.dependencies { + if !dependency_is_satisfied(dependency, policy, node_lookup, dependency_lookup) { + reasons.push(format!( + "dependency `{}` has not reached a required terminal state", + dependency.dependency_id() + )); + } + } + for domain in &node.conflict_domains { + if occupied_conflicts.contains(domain) { + reasons.push(format!( + "conflict domain `{}:{}` is already occupied", + domain.kind.as_str(), + domain.key() + )); + } + } + + if let Some(issue) = &node.linear_issue { + collect_issue_mapping_reasons(issue, policy, reasons); + } else { + reasons.push(String::from("node has no normal Linear issue mapping")); + } +} + +fn collect_issue_mapping_reasons( + issue: &ExecutionLinearIssueMapping, + policy: &ExecutionWorkflowPolicy, + reasons: &mut Vec, +) { + if policy.issue_is_terminal(issue) { + reasons.push(format!( + "mapped issue `{}` is already terminal in `{}`", + issue.issue_identifier(), + issue.issue_state() + )); + } + if !policy.issue_is_startable(issue) { + reasons.push(format!( + "mapped issue `{}` is not in a startable state", + issue.issue_identifier() + )); + } + if issue.has_active_label { + reasons.push(format!( + "mapped issue `{}` already carries `{}`", + issue.issue_identifier(), + policy.active_label + )); + } + if issue.has_opt_out_label { + reasons.push(format!( + "mapped issue `{}` carries `{}`", + issue.issue_identifier(), + policy.opt_out_label + )); + } + if issue.has_needs_attention_label { + reasons.push(format!( + "mapped issue `{}` carries `{}`", + issue.issue_identifier(), + policy.needs_attention_label + )); + } + if !issue.has_generic_dispatch_briefing { + reasons.push(format!( + "mapped issue `{}` is missing a generic dispatch briefing", + issue.issue_identifier() + )); + } +} + +fn dependency_is_satisfied( + dependency: &ExecutionProgramDependency, + policy: &ExecutionWorkflowPolicy, + node_lookup: &BTreeMap<&str, &ExecutionProgramNode>, + dependency_lookup: &BTreeMap<&str, &ExecutionDependencySnapshot>, +) -> bool { + if let Some(snapshot) = dependency_lookup.get(dependency.dependency_id()) { + if let Some(state) = &snapshot.tracker_state { + return dependency_terminal_states(dependency, policy) + .iter() + .any(|terminal| terminal == state); + } + if let Some(queue_intent) = snapshot.queue_intent { + return queue_intent.is_terminal(); + } + } + + node_lookup + .get(dependency.dependency_id()) + .is_some_and(|node| node.queue_intent().is_terminal()) +} + +fn dependency_terminal_states<'a>( + dependency: &'a ExecutionProgramDependency, + policy: &'a ExecutionWorkflowPolicy, +) -> &'a [String] { + if dependency.required_terminal_states.is_empty() { + policy.terminal_states() + } else { + &dependency.required_terminal_states + } +} + +fn queue_label_action_for( + node: &ExecutionProgramNode, + state: ExecutionReadinessState, + policy: &ExecutionWorkflowPolicy, +) -> Option { + let issue = node.linear_issue()?; + + if state != ExecutionReadinessState::Ready { + return issue.has_queue_label().then_some(ExecutionQueueLabelAction::Remove); + } + if !matches!( + node.queue_intent(), + ExecutionQueueIntent::ReadyToQueue | ExecutionQueueIntent::Queued + ) || !policy.issue_is_startable(issue) + { + return issue.has_queue_label().then_some(ExecutionQueueLabelAction::Remove); + } + + Some(if issue.has_queue_label() { + ExecutionQueueLabelAction::Retain + } else { + ExecutionQueueLabelAction::Apply + }) +} + +fn ensure_accepted_contract(contract: &DecisionContract) -> Result<()> { + contract.validate()?; + + if contract.status() != DecisionContractStatus::AcceptedPromoted { + eyre::bail!( + "Execution Programs can only derive from accepted Decision Contracts; `{}` is `{}`.", + contract.contract_id(), + contract.status().as_str() + ); + } + + Ok(()) +} + +fn decision_contract_fingerprint(contract: &DecisionContract) -> Result { + contract.validate()?; + + let payload = serde_json::to_vec(contract)?; + let digest = Sha256::digest(payload); + + Ok(digest.iter().map(|byte| format!("{byte:02x}")).collect::()) +} + +fn execution_program_schema() -> String { + EXECUTION_PROGRAM_SCHEMA.to_owned() +} + +fn execution_program_record_version() -> u16 { + EXECUTION_PROGRAM_RECORD_VERSION +} + +fn validate_required(name: &str, value: &str) -> Result<()> { + if value.trim().is_empty() { + eyre::bail!("{name} must not be empty."); + } + + Ok(()) +} + +fn validate_optional(name: &str, value: Option<&str>) -> Result<()> { + if let Some(value) = value { + validate_required(name, value)?; + } + + Ok(()) +} + +fn non_empty_optional(value: &str) -> Option<&str> { + if value.is_empty() { None } else { Some(value) } +} + +fn validate_string_list(name: &str, values: &[String]) -> Result<()> { + for value in values { + validate_required(name, value)?; + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use crate::{ + execution_program::{ + ExecutionConflictDomain, ExecutionConflictDomainKind, ExecutionDependencySnapshot, + ExecutionLinearIssueMapping, ExecutionProgram, ExecutionProgramDependency, + ExecutionProgramNode, ExecutionProgramNodeStage, ExecutionProgramReadinessContext, + ExecutionQueueIntent, ExecutionQueueLabelAction, ExecutionReadinessState, + ExecutionWorkflowPolicy, + }, + loop_contract::{DecisionContract, DecisionPromotion, DecisionPromotionActorKind}, + }; + + fn latent_contract_fixture() -> DecisionContract { + serde_json::from_str(include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/fixtures/decision_contract/research_x_latent_contract.json" + ))) + .expect("decision contract fixture should deserialize") + } + + fn accepted_contract_fixture() -> DecisionContract { + let mut contract = latent_contract_fixture(); + + contract + .promote( + DecisionPromotion::new( + "operator", + DecisionPromotionActorKind::User, + "2026-06-09T10:00:00Z", + "conversation", + Some(String::from("User asked to push this forward.")), + ) + .expect("promotion should build"), + ) + .expect("contract should promote"); + + contract + } + + fn workflow_policy() -> ExecutionWorkflowPolicy { + ExecutionWorkflowPolicy::new( + "decodex", + vec![String::from("Todo")], + vec![String::from("Done"), String::from("Canceled"), String::from("Duplicate")], + "decodex:manual-only", + "decodex:needs-attention", + ) + .expect("workflow policy should build") + } + + fn issue(identifier: &str, state: &str) -> ExecutionLinearIssueMapping { + ExecutionLinearIssueMapping::new( + format!("linear-{identifier}"), + identifier.to_owned(), + state.to_owned(), + ) + .expect("issue mapping should build") + } + + fn ready_node(id: &str, issue_identifier: &str) -> ExecutionProgramNode { + ExecutionProgramNode::new( + id, + ExecutionProgramNodeStage::Runtime, + format!("Implement {id}."), + ExecutionQueueIntent::ReadyToQueue, + ) + .expect("node should build") + .with_objective_lineage([String::from("Ship the accepted runtime work.")]) + .expect("lineage should attach") + .with_acceptance_expectations([String::from("Acceptance is concrete.")]) + .expect("acceptance should attach") + .with_validation_expectations([String::from("Run the repo gate.")]) + .expect("validation should attach") + .with_linear_issue(issue(issue_identifier, "Todo")) + .expect("issue should attach") + } + + fn program_with(nodes: Vec) -> (DecisionContract, ExecutionProgram) { + let contract = accepted_contract_fixture(); + let program = + ExecutionProgram::from_accepted_contract("program-1", "decodex", &contract, nodes) + .expect("program should derive from accepted contract"); + + (contract, program) + } + + #[test] + fn readiness_selects_only_startable_ready_nodes() { + let blocked = ready_node("node-blocked", "XY-901") + .with_dependencies([ + ExecutionProgramDependency::new("node-ready").expect("dependency should build") + ]) + .expect("dependency should attach"); + let (contract, program) = program_with(vec![ready_node("node-ready", "XY-900"), blocked]); + let evaluation = program + .evaluate(&contract, &workflow_policy(), &ExecutionProgramReadinessContext::new()) + .expect("program should evaluate"); + + assert_eq!(evaluation.ready_node_ids(), vec!["node-ready"]); + assert_eq!(evaluation.startable_node_ids(), vec!["node-ready"]); + assert_eq!( + evaluation.nodes()[0].queue_label_action(), + Some(ExecutionQueueLabelAction::Apply) + ); + assert_eq!(evaluation.operator_summary().ready_count, 1); + assert_eq!(evaluation.operator_summary().blocked_count, 1); + } + + #[test] + fn dependency_blocking_respects_workflow_terminal_states() { + let dependent = ready_node("node-dependent", "XY-902") + .with_dependencies([ExecutionProgramDependency::new("node-dependency") + .expect("dependency should build")]) + .expect("dependency should attach"); + let (contract, program) = + program_with(vec![ready_node("node-dependency", "XY-901"), dependent.clone()]); + let blocked_context = ExecutionProgramReadinessContext::new().with_dependency_snapshots([ + ExecutionDependencySnapshot::tracker_state("node-dependency", "In Review") + .expect("snapshot should build"), + ]); + let blocked = program + .evaluate(&contract, &workflow_policy(), &blocked_context) + .expect("program should evaluate"); + let dependent_evaluation = blocked + .nodes() + .iter() + .find(|node| node.node_id() == "node-dependent") + .expect("dependent node should exist"); + + assert_eq!(dependent_evaluation.state(), ExecutionReadinessState::Blocked); + assert!( + dependent_evaluation + .reasons() + .iter() + .any(|reason| reason.contains("required terminal state")) + ); + + let ready_context = ExecutionProgramReadinessContext::new().with_dependency_snapshots([ + ExecutionDependencySnapshot::tracker_state("node-dependency", "Done") + .expect("snapshot should build"), + ]); + let ready = program + .evaluate(&contract, &workflow_policy(), &ready_context) + .expect("program should evaluate"); + + assert!(ready.startable_node_ids().contains(&"node-dependent")); + } + + #[test] + fn stale_contract_drift_blocks_queue_retention() { + let stale_node = ready_node("node-stale", "XY-903") + .with_contract_fingerprint("stale-contract-fingerprint") + .expect("fingerprint should override"); + let (contract, program) = program_with(vec![stale_node]); + let evaluation = program + .evaluate(&contract, &workflow_policy(), &ExecutionProgramReadinessContext::new()) + .expect("program should evaluate"); + let node = &evaluation.nodes()[0]; + + assert_eq!(node.state(), ExecutionReadinessState::Stale); + assert_eq!(node.queue_label_action(), None); + assert!(evaluation.startable_node_ids().is_empty()); + } + + #[test] + fn conflict_domain_blocks_ready_node() { + let conflict = ExecutionConflictDomain::new( + ExecutionConflictDomainKind::File, + "apps/decodex/src/runtime.rs", + ) + .expect("domain should build"); + let node = ready_node("node-conflict", "XY-904") + .with_conflict_domains([conflict.clone()]) + .expect("conflict should attach"); + let (contract, program) = program_with(vec![node]); + let context = + ExecutionProgramReadinessContext::new().with_occupied_conflict_domains([conflict]); + let evaluation = program + .evaluate(&contract, &workflow_policy(), &context) + .expect("program should evaluate"); + let node = &evaluation.nodes()[0]; + + assert_eq!(node.state(), ExecutionReadinessState::Blocked); + assert!(node.reasons().iter().any(|reason| reason.contains("already occupied"))); + } + + #[test] + fn linear_issue_mapping_controls_apply_retain_and_remove() { + let apply = ready_node("node-apply", "XY-905"); + let retain = ready_node("node-retain", "XY-906") + .with_linear_issue(issue("XY-906", "Todo").with_queue_label(true)) + .expect("issue should attach"); + let remove = ready_node("node-remove", "XY-907") + .with_linear_issue(issue("XY-907", "In Progress").with_queue_label(true)) + .expect("issue should attach"); + let (contract, program) = program_with(vec![apply, retain, remove]); + let evaluation = program + .evaluate(&contract, &workflow_policy(), &ExecutionProgramReadinessContext::new()) + .expect("program should evaluate"); + let action_for = |id: &str| { + evaluation + .nodes() + .iter() + .find(|node| node.node_id() == id) + .expect("node should exist") + .queue_label_action() + }; + + assert_eq!(action_for("node-apply"), Some(ExecutionQueueLabelAction::Apply)); + assert_eq!(action_for("node-retain"), Some(ExecutionQueueLabelAction::Retain)); + assert_eq!(action_for("node-remove"), Some(ExecutionQueueLabelAction::Remove)); + assert_eq!( + evaluation + .nodes() + .iter() + .find(|node| node.node_id() == "node-remove") + .expect("node should exist") + .state(), + ExecutionReadinessState::Blocked + ); + } + + #[test] + fn unmapped_ready_to_queue_node_is_blocked_from_startable_selection() { + let unmapped = ExecutionProgramNode::new( + "node-unmapped", + ExecutionProgramNodeStage::Runtime, + "Implement unmapped work.", + ExecutionQueueIntent::ReadyToQueue, + ) + .expect("node should build") + .with_acceptance_expectations([String::from("Acceptance is concrete.")]) + .expect("acceptance should attach") + .with_validation_expectations([String::from("Run the repo gate.")]) + .expect("validation should attach"); + let (contract, program) = program_with(vec![unmapped]); + let evaluation = program + .evaluate(&contract, &workflow_policy(), &ExecutionProgramReadinessContext::new()) + .expect("program should evaluate"); + let node = &evaluation.nodes()[0]; + + assert_eq!(node.state(), ExecutionReadinessState::Blocked); + assert!(node.reasons().iter().any(|reason| reason.contains("no normal Linear issue"))); + assert!(evaluation.startable_node_ids().is_empty()); + } + + #[test] + fn evaluator_rejects_wrong_service_policy() { + let (contract, program) = program_with(vec![ready_node("node-ready", "XY-908")]); + let wrong_service_policy = ExecutionWorkflowPolicy::new( + "other-service", + vec![String::from("Todo")], + vec![String::from("Done")], + "decodex:manual-only", + "decodex:needs-attention", + ) + .expect("workflow policy should build"); + let error = program + .evaluate(&contract, &wrong_service_policy, &ExecutionProgramReadinessContext::new()) + .expect_err("program should reject mismatched service policy"); + + assert!(error.to_string().contains("readiness policy belongs to")); + } +} diff --git a/apps/decodex/src/lib.rs b/apps/decodex/src/lib.rs index 12999cd82..ab3975bae 100644 --- a/apps/decodex/src/lib.rs +++ b/apps/decodex/src/lib.rs @@ -12,6 +12,7 @@ mod cli; mod codex_config; mod commit_message; mod default_branch_sync; +mod execution_program; mod git_credentials; mod github; mod loop_contract; diff --git a/apps/decodex/src/orchestrator.rs b/apps/decodex/src/orchestrator.rs index 8db567fc8..4a95f67a6 100644 --- a/apps/decodex/src/orchestrator.rs +++ b/apps/decodex/src/orchestrator.rs @@ -34,7 +34,7 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use crate::{agent, default_branch_sync, git_credentials, maintenance, state}; #[rustfmt::skip] -use crate::{agent::{ACTIVE_RUN_IDLE_TIMEOUT, AppServerCapabilityPreflightFailure, AppServerDynamicToolFailure, AppServerHomePreflightFailure, AppServerPhaseGoalFailure, AppServerProcessEnv, AppServerRunRequest, AppServerRunResult, AppServerTransportFailure, AppServerTurnFailure, ISSUE_DELIVERY_CLOSEOUT_COMPLETE_TOOL_NAME, ISSUE_LABEL_ADD_TOOL_NAME, ISSUE_PROGRESS_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_HANDOFF_TOOL_NAME, ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME, ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, DecodexRunContext, DecodexToolBridge, PhaseGoalController, PhaseGoalKind, PhaseGoalSpec, PhaseGoalTransition, ReviewExecutionMode, ReviewHandoffContext, ReviewHandoffWritebackFailed, ReviewPolicyStopReason, ReviewPolicyStopRequested, RunCompletionDisposition, TrackerToolBridge, TurnContinuationGuard}, config::{InternalReviewMode, ServiceConfig}, git_credentials::GitCredentialSource, github, prelude::{Result, eyre}, state::{ChildAgentActivityBucket, ChildAgentActivitySummary, CodexAccountActivitySummary, ProjectRegistration, ProjectRunStatus, ProtocolActivitySummary, RUN_OPERATION_AGENT_RUN, RUN_OPERATION_APP_SERVER_PREFLIGHT, RUN_OPERATION_GIT_CREDENTIALS, RUN_OPERATION_IDLE, RUN_OPERATION_RECONCILIATION, RUN_OPERATION_REPO_GATE, RUN_OPERATION_REVIEW_WRITEBACK, RUN_OPERATION_WAITING_EXTERNAL, ReviewHandoffMarker, ReviewOrchestrationMarker, RunActivityMarker, RunAttempt, StateStore, WorktreeMapping}, tracker::{IssueTracker, TrackerComment, TrackerIssue, linear::LinearClient, records}, workflow::{WorkflowDocument, WorkflowExecution}, worktree::{WorktreeManager, WorktreeSpec}}; +use crate::{agent::{ACTIVE_RUN_IDLE_TIMEOUT, AppServerCapabilityPreflightFailure, AppServerDynamicToolFailure, AppServerHomePreflightFailure, AppServerPhaseGoalFailure, AppServerProcessEnv, AppServerRunRequest, AppServerRunResult, AppServerTransportFailure, AppServerTurnFailure, ISSUE_DELIVERY_CLOSEOUT_COMPLETE_TOOL_NAME, ISSUE_LABEL_ADD_TOOL_NAME, ISSUE_PROGRESS_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_HANDOFF_TOOL_NAME, ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME, ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, DecodexRunContext, DecodexToolBridge, PhaseGoalController, PhaseGoalKind, PhaseGoalSpec, PhaseGoalTransition, ReviewExecutionMode, ReviewHandoffContext, ReviewHandoffWritebackFailed, ReviewPolicyStopReason, ReviewPolicyStopRequested, RunCompletionDisposition, TrackerToolBridge, TurnContinuationGuard}, config::{InternalReviewMode, ServiceConfig}, execution_program::{ExecutionProgramOperatorSummary, ExecutionProgramReadinessContext, ExecutionWorkflowPolicy}, git_credentials::GitCredentialSource, github, prelude::{Result, eyre}, state::{ChildAgentActivityBucket, ChildAgentActivitySummary, CodexAccountActivitySummary, ExecutionProgramRecord, ProjectRegistration, ProjectRunStatus, ProtocolActivitySummary, RUN_OPERATION_AGENT_RUN, RUN_OPERATION_APP_SERVER_PREFLIGHT, RUN_OPERATION_GIT_CREDENTIALS, RUN_OPERATION_IDLE, RUN_OPERATION_RECONCILIATION, RUN_OPERATION_REPO_GATE, RUN_OPERATION_REVIEW_WRITEBACK, RUN_OPERATION_WAITING_EXTERNAL, ReviewHandoffMarker, ReviewOrchestrationMarker, RunActivityMarker, RunAttempt, StateStore, WorktreeMapping}, tracker::{IssueTracker, TrackerComment, TrackerIssue, linear::LinearClient, records}, workflow::{WorkflowDocument, WorkflowExecution}, worktree::{WorktreeManager, WorktreeSpec}}; include!("orchestrator/types.rs"); diff --git a/apps/decodex/src/orchestrator/entrypoints.rs b/apps/decodex/src/orchestrator/entrypoints.rs index 2a36f29c0..75cbf2af3 100644 --- a/apps/decodex/src/orchestrator/entrypoints.rs +++ b/apps/decodex/src/orchestrator/entrypoints.rs @@ -1086,6 +1086,7 @@ fn append_control_plane_project_snapshot( snapshot.active_runs.extend(project_snapshot.active_runs); snapshot.recent_runs.extend(project_snapshot.recent_runs); snapshot.history_lanes.extend(project_snapshot.history_lanes); + snapshot.execution_programs.extend(project_snapshot.execution_programs); snapshot.queued_candidates.extend(project_snapshot.queued_candidates); snapshot.worktrees.extend(project_snapshot.worktrees); snapshot.post_review_lanes.extend(project_snapshot.post_review_lanes); @@ -1616,6 +1617,7 @@ fn empty_control_plane_snapshot(limit: usize) -> OperatorStatusSnapshot { active_runs: Vec::new(), recent_runs: Vec::new(), history_lanes: Vec::new(), + execution_programs: Vec::new(), queued_candidates: Vec::new(), worktrees: Vec::new(), post_review_lanes: Vec::new(), diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index b01249e3f..6056fbff9 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -327,6 +327,7 @@ fn build_operator_status_snapshot_with_account_mode( active_runs, recent_runs, history_lanes, + execution_programs: Vec::new(), queued_candidates: Vec::new(), worktrees, post_review_lanes: Vec::new(), @@ -534,6 +535,9 @@ where options.account_activity_mode, )?; + snapshot.execution_programs = + operator_execution_program_statuses(project, workflow, state_store)?; + hydrate_history_lanes_from_local_ledger(project, state_store, &mut snapshot)?; hydrate_live_operator_external_observers( LiveOperatorStatusObserverContext { @@ -557,6 +561,36 @@ where Ok(snapshot) } +fn operator_execution_program_statuses( + project: &ServiceConfig, + workflow: &WorkflowDocument, + state_store: &StateStore, +) -> crate::prelude::Result> { + let policy = ExecutionWorkflowPolicy::from_workflow(project.service_id(), workflow)?; + let context = ExecutionProgramReadinessContext::new(); + let mut statuses = Vec::new(); + + for record in state_store.list_execution_programs(project.service_id())? { + let Some(contract) = + state_store.decision_contract(project.service_id(), record.source_contract_id())? + else { + statuses.push(OperatorExecutionProgramStatus::missing_contract(&record)); + + continue; + }; + let evaluation = record.program().evaluate(contract.contract(), &policy, &context)?; + + statuses.push(OperatorExecutionProgramStatus::from_summary( + &record, + evaluation.operator_summary(), + )); + } + + statuses.sort_by(|left, right| left.program_id.cmp(&right.program_id)); + + Ok(statuses) +} + fn hydrate_live_operator_external_observers( context: LiveOperatorStatusObserverContext<'_, T>, snapshot: &mut OperatorStatusSnapshot, @@ -5164,8 +5198,15 @@ fn render_operator_status(snapshot: &OperatorStatusSnapshot) -> String { "Stale closed queue labels: {}\n", stale_closed_queue_labels.len() )); + output.push_str(&format!( + "Execution programs: {}\n", + snapshot.execution_programs.len() + )); output.push_str(&format!("Recovery worktrees: {}\n", recovery_worktrees.len())); output.push_str(&format!("Post-review lanes: {}\n", snapshot.post_review_lanes.len())); + + append_rendered_execution_programs(&mut output, snapshot); + output.push_str("\nRunning Lanes\n"); if snapshot.active_runs.is_empty() { @@ -5217,6 +5258,44 @@ fn render_operator_status(snapshot: &OperatorStatusSnapshot) -> String { output } +fn append_rendered_execution_programs(output: &mut String, snapshot: &OperatorStatusSnapshot) { + output.push_str("\nExecution Programs\n"); + + if snapshot.execution_programs.is_empty() { + output.push_str("- none\n"); + + return; + } + + for program in &snapshot.execution_programs { + let mapped_issues = if program.mapped_issue_identifiers.is_empty() { + String::from("none") + } else { + program.mapped_issue_identifiers.join(", ") + }; + let readback_warning = program + .readback_warning + .as_ref() + .map_or_else(String::new, |warning| format!(" readback_warning={warning}")); + + output.push_str(&format!( + "- program_id: {} source_contract_id: {} nodes={} ready={} blocked={} paused={} active={} completed={} stale={} queue_label_eligible={} mapped_issues={}{}\n", + program.program_id, + program.source_contract_id, + program.node_count, + program.ready_count, + program.blocked_count, + program.paused_count, + program.active_count, + program.completed_count, + program.stale_count, + program.queue_label_eligible_count, + mapped_issues, + readback_warning, + )); + } +} + fn append_rendered_github_cli_authority(output: &mut String, snapshot: &OperatorStatusSnapshot) { if let Some(authority) = rendered_project_github_cli_authority(snapshot) { output.push_str(&format!( diff --git a/apps/decodex/src/orchestrator/tests.rs b/apps/decodex/src/orchestrator/tests.rs index 40d34a748..dcb27cac8 100644 --- a/apps/decodex/src/orchestrator/tests.rs +++ b/apps/decodex/src/orchestrator/tests.rs @@ -34,7 +34,7 @@ use crate::config::{InternalReviewMode, ServiceConfig}; #[rustfmt::skip] use crate::github; #[rustfmt::skip] -use crate::orchestrator::{self, ActiveChildRunContext, ActiveRunDisposition, ActiveRunReconciliation, ActiveWorkflowOverride, AgentEvidenceSource, ChildExitRetryContext, ChildRunRef, ControlPlaneProjectTick, CONTINUATION_PENDING_RUN_STATUS, DaemonRunChild, DaemonTickRuntimeContext, DashboardEventHub, EvidenceRequest, GhPullRequestReviewStateInspector, ISSUE_DELIVERY_CLOSEOUT_COMPLETE_TOOL_NAME, ISSUE_LABEL_ADD_TOOL_NAME, ISSUE_PROGRESS_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_HANDOFF_TOOL_NAME, ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME, ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, IssueDispatchMode, IssueRunPlan, IssueTurnContinuationGuard, ManualAttentionRequested, OPERATOR_DASHBOARD_ALIAS_ENDPOINT_PATH, OPERATOR_DASHBOARD_ENDPOINT_PATH, OperatorCodexAccountControlStatus, OperatorGitHubCliAuthority, OperatorProjectStatus, OperatorStatusSnapshot, PostReviewLaneClassification, PostReviewLaneDecision, PostReviewLaneSnapshot, PreferredRunIdentity, PrepareIssueRunContext, PublishedOperatorSnapshot, PullRequestCommitConnection, PullRequestCommitNode, PullRequestCommitPayload, PullRequestIssueCommentConnection, PullRequestIssueCommentState, PullRequestIssueCommentsNode, PullRequestPageInfo, PullRequestReactionGroup, PullRequestReactionUsersConnection, PullRequestActor, PullRequestRepository, PullRequestRepositoryOwner, PullRequestReviewConnection, PullRequestIssueCommentNode, PullRequestReviewNode, PullRequestReviewRequestConnection, PullRequestReviewState, PullRequestReviewStateInspector, PullRequestReviewStateNode, PullRequestReviewStateRepository, PullRequestReviewSummaryState, PullRequestReviewThreadConnection, PullRequestReviewThreadNode, PullRequestStatusCheckRollup, RecoveredRuntimeState, RetainedPartialProgress, RetainedReviewRunIdentity, RetryComment, RetryDispatchDecision, RetryEntry, RetryKind, RetryQueue, RunCompletionDisposition, RunSummary, RepoGateFailure, TERMINAL_GUARD_MARKER_FILE, TERMINAL_GUARDED_RUN_STATUS, TRACKER_RATE_LIMIT_WARNING, TargetIssueRunContext, EXTERNAL_REVIEW_ACTOR_LOGIN, EXTERNAL_REVIEW_PASS_PHRASE, EXTERNAL_REVIEW_REQUEST_BODY}; +use crate::orchestrator::{self, ActiveChildRunContext, ActiveRunDisposition, ActiveRunReconciliation, ActiveWorkflowOverride, AgentEvidenceSource, ChildExitRetryContext, ChildRunRef, ControlPlaneProjectTick, CONTINUATION_PENDING_RUN_STATUS, DaemonRunChild, DaemonTickRuntimeContext, DashboardEventHub, EvidenceRequest, GhPullRequestReviewStateInspector, ISSUE_DELIVERY_CLOSEOUT_COMPLETE_TOOL_NAME, ISSUE_LABEL_ADD_TOOL_NAME, ISSUE_PROGRESS_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_HANDOFF_TOOL_NAME, ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME, ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, IssueDispatchMode, IssueRunPlan, IssueTurnContinuationGuard, ManualAttentionRequested, OPERATOR_DASHBOARD_ALIAS_ENDPOINT_PATH, OPERATOR_DASHBOARD_ENDPOINT_PATH, OperatorCodexAccountControlStatus, OperatorExecutionProgramStatus, OperatorGitHubCliAuthority, OperatorProjectStatus, OperatorStatusSnapshot, PostReviewLaneClassification, PostReviewLaneDecision, PostReviewLaneSnapshot, PreferredRunIdentity, PrepareIssueRunContext, PublishedOperatorSnapshot, PullRequestCommitConnection, PullRequestCommitNode, PullRequestCommitPayload, PullRequestIssueCommentConnection, PullRequestIssueCommentState, PullRequestIssueCommentsNode, PullRequestPageInfo, PullRequestReactionGroup, PullRequestReactionUsersConnection, PullRequestActor, PullRequestRepository, PullRequestRepositoryOwner, PullRequestReviewConnection, PullRequestIssueCommentNode, PullRequestReviewNode, PullRequestReviewRequestConnection, PullRequestReviewState, PullRequestReviewStateInspector, PullRequestReviewStateNode, PullRequestReviewStateRepository, PullRequestReviewSummaryState, PullRequestReviewThreadConnection, PullRequestReviewThreadNode, PullRequestStatusCheckRollup, RecoveredRuntimeState, RetainedPartialProgress, RetainedReviewRunIdentity, RetryComment, RetryDispatchDecision, RetryEntry, RetryKind, RetryQueue, RunCompletionDisposition, RunSummary, RepoGateFailure, TERMINAL_GUARD_MARKER_FILE, TERMINAL_GUARDED_RUN_STATUS, TRACKER_RATE_LIMIT_WARNING, TargetIssueRunContext, EXTERNAL_REVIEW_ACTOR_LOGIN, EXTERNAL_REVIEW_PASS_PHRASE, EXTERNAL_REVIEW_REQUEST_BODY}; #[rustfmt::skip] use crate::prelude::Result; #[rustfmt::skip] 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 c0fc8370e..4f379591e 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/agent_evidence.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/agent_evidence.rs @@ -22,6 +22,7 @@ fn agent_evidence_snapshot_writes_index_blockers_capsules_and_event_stream() { active_runs: vec![active_run.clone()], recent_runs: vec![active_run], history_lanes: Vec::new(), + execution_programs: Vec::new(), queued_candidates: vec![agent_evidence_blocked_candidate()], worktrees: operator_status_text_worktrees(), post_review_lanes: vec![agent_evidence_missing_handoff_lane()], diff --git a/apps/decodex/src/orchestrator/tests/operator/status/text.rs b/apps/decodex/src/orchestrator/tests/operator/status/text.rs index ce88ceed6..7101cfaef 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/text.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/text.rs @@ -43,6 +43,7 @@ fn operator_status_text_surfaces_github_cli_authority() { queued_candidates: Vec::new(), recent_runs: Vec::new(), history_lanes: Vec::new(), + execution_programs: Vec::new(), worktrees: Vec::new(), post_review_lanes: Vec::new(), }; @@ -72,6 +73,7 @@ fn operator_status_text_renders_human_readable_sections() { queued_candidates: operator_status_text_queued_candidates(), recent_runs: vec![active_run], history_lanes: Vec::new(), + execution_programs: Vec::new(), worktrees: operator_status_text_worktrees(), post_review_lanes: operator_status_text_post_review_lanes(), }; @@ -151,6 +153,50 @@ fn operator_status_text_renders_human_readable_sections() { assert_recovery_worktree_roles_are_grouped(&rendered); } +#[test] +fn operator_status_text_surfaces_execution_program_summary() { + let snapshot = OperatorStatusSnapshot { + project_id: String::from("decodex"), + run_limit: 10, + warnings: Vec::new(), + warning_details: Vec::new(), + connector_backoffs: Vec::new(), + projects: Vec::new(), + account_control: OperatorCodexAccountControlStatus { + mode: String::from("balanced"), + account_selector: None, + }, + accounts: Vec::new(), + active_runs: Vec::new(), + queued_candidates: Vec::new(), + recent_runs: Vec::new(), + history_lanes: Vec::new(), + execution_programs: vec![OperatorExecutionProgramStatus { + program_id: String::from("program-853"), + source_contract_id: String::from("contract-852"), + node_count: 3, + ready_count: 1, + blocked_count: 1, + paused_count: 0, + active_count: 0, + completed_count: 1, + stale_count: 0, + queue_label_eligible_count: 1, + mapped_issue_identifiers: vec![String::from("XY-853")], + readback_warning: None, + }], + worktrees: Vec::new(), + post_review_lanes: Vec::new(), + }; + let rendered = orchestrator::render_operator_status(&snapshot); + + assert!(rendered.contains("Execution programs: 1")); + assert!(rendered.contains("Execution Programs")); + assert!(rendered.contains( + "program_id: program-853 source_contract_id: contract-852 nodes=3 ready=1 blocked=1 paused=0 active=0 completed=1 stale=0 queue_label_eligible=1 mapped_issues=XY-853" + )); +} + #[test] fn queue_explain_renders_candidate_reasons_without_running_dispatch() { let (_temp_dir, config, _workflow) = temp_project_layout(); @@ -210,6 +256,7 @@ fn operator_status_text_explains_empty_backlog_checks() { queued_candidates: Vec::new(), recent_runs: Vec::new(), history_lanes: Vec::new(), + execution_programs: Vec::new(), worktrees: Vec::new(), post_review_lanes: Vec::new(), }; @@ -245,6 +292,7 @@ fn operator_status_text_surfaces_cleanup_blocker_pr_url() { queued_candidates: Vec::new(), recent_runs: Vec::new(), history_lanes: Vec::new(), + execution_programs: Vec::new(), worktrees: vec![orchestrator::OperatorWorktreeStatus { issue_id: String::from("issue-3"), issue_identifier: Some(String::from("PUB-103")), @@ -312,6 +360,7 @@ fn operator_status_text_terminal_run_freshness_uses_terminal_update() { queued_candidates: Vec::new(), recent_runs: vec![terminal_run], history_lanes, + execution_programs: Vec::new(), worktrees: Vec::new(), post_review_lanes: Vec::new(), }; @@ -350,6 +399,7 @@ fn operator_status_text_active_run_without_live_activity_does_not_promote_update queued_candidates: Vec::new(), recent_runs: vec![active_run], history_lanes: Vec::new(), + execution_programs: Vec::new(), worktrees: Vec::new(), post_review_lanes: Vec::new(), }; @@ -383,6 +433,7 @@ fn operator_status_text_explains_unleased_live_running_lane() { queued_candidates: Vec::new(), recent_runs: vec![active_run], history_lanes: Vec::new(), + execution_programs: Vec::new(), worktrees: Vec::new(), post_review_lanes: Vec::new(), }; diff --git a/apps/decodex/src/orchestrator/types.rs b/apps/decodex/src/orchestrator/types.rs index b9a391fd7..f2321c702 100644 --- a/apps/decodex/src/orchestrator/types.rs +++ b/apps/decodex/src/orchestrator/types.rs @@ -1001,6 +1001,7 @@ struct OperatorStatusSnapshot { active_runs: Vec, recent_runs: Vec, history_lanes: Vec, + execution_programs: Vec, queued_candidates: Vec, worktrees: Vec, post_review_lanes: Vec, @@ -1050,6 +1051,62 @@ struct OperatorProjectStatus { warning_count: usize, } +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +struct OperatorExecutionProgramStatus { + program_id: String, + source_contract_id: String, + node_count: usize, + ready_count: usize, + blocked_count: usize, + paused_count: usize, + active_count: usize, + completed_count: usize, + stale_count: usize, + queue_label_eligible_count: usize, + mapped_issue_identifiers: Vec, + readback_warning: Option, +} +impl OperatorExecutionProgramStatus { + fn from_summary( + record: &ExecutionProgramRecord, + summary: ExecutionProgramOperatorSummary, + ) -> Self { + Self { + program_id: summary.program_id, + source_contract_id: record.source_contract_id().to_owned(), + node_count: record.program().nodes().len(), + ready_count: summary.ready_count, + blocked_count: summary.blocked_count, + paused_count: summary.paused_count, + active_count: summary.active_count, + completed_count: summary.completed_count, + stale_count: summary.stale_count, + queue_label_eligible_count: summary.queue_label_eligible_count, + mapped_issue_identifiers: summary.mapped_issue_identifiers, + readback_warning: None, + } + } + + fn missing_contract(record: &ExecutionProgramRecord) -> Self { + let node_count = record.program().nodes().len(); + + Self { + program_id: record.program_id().to_owned(), + source_contract_id: record.source_contract_id().to_owned(), + node_count, + ready_count: 0, + blocked_count: 0, + paused_count: 0, + active_count: 0, + completed_count: 0, + stale_count: node_count, + queue_label_eligible_count: 0, + mapped_issue_identifiers: Vec::new(), + readback_warning: Some(String::from("source_decision_contract_missing")), + } + } +} + #[derive(Clone, Debug, Eq, PartialEq, Serialize)] struct OperatorGitHubCliAuthority { command_path: String, diff --git a/apps/decodex/src/state.rs b/apps/decodex/src/state.rs index ba3c5b61c..20a997cba 100644 --- a/apps/decodex/src/state.rs +++ b/apps/decodex/src/state.rs @@ -22,6 +22,7 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use crate::{ config::ServiceConfig, + execution_program::ExecutionProgram, loop_contract::{DecisionContract, DecisionContractStatus, DecisionPromotion}, prelude::{Result, eyre}, tracker::records::{self, LinearExecutionEventRecord}, diff --git a/apps/decodex/src/state/internal.rs b/apps/decodex/src/state/internal.rs index 1791c727c..abe621460 100644 --- a/apps/decodex/src/state/internal.rs +++ b/apps/decodex/src/state/internal.rs @@ -128,6 +128,7 @@ struct StateData { linear_execution_events: HashMap, private_execution_events: Vec, decision_contracts: HashMap, + execution_programs: HashMap, review_handoffs: HashMap, review_orchestrations: HashMap, review_policy_checkpoints: HashMap, @@ -148,6 +149,7 @@ impl StateData { self.linear_execution_events = loaded.linear_execution_events; self.private_execution_events = loaded.private_execution_events; self.decision_contracts = loaded.decision_contracts; + self.execution_programs = loaded.execution_programs; self.review_handoffs = loaded.review_handoffs; self.review_orchestrations = loaded.review_orchestrations; self.review_policy_checkpoints = loaded.review_policy_checkpoints; @@ -354,6 +356,7 @@ ON linear_execution_events (service_id, issue_id, event_unix, recorded_at_unix); self.bootstrap_connector_backoffs_schema()?; self.bootstrap_private_execution_events_schema()?; self.bootstrap_decision_contracts_schema()?; + self.bootstrap_execution_programs_schema()?; self.record_schema_version()?; Ok(()) @@ -529,6 +532,28 @@ ON decision_contracts (project_id, status, updated_at_unix); Ok(()) } + fn bootstrap_execution_programs_schema(&self) -> Result<()> { + self.connection.execute_batch( + r#" +CREATE TABLE IF NOT EXISTS execution_programs ( + project_id TEXT NOT NULL, + program_id TEXT NOT NULL, + source_contract_id TEXT NOT NULL, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL, + created_at_unix INTEGER NOT NULL, + updated_at TEXT NOT NULL, + updated_at_unix INTEGER NOT NULL, + PRIMARY KEY (project_id, program_id) +); +CREATE INDEX IF NOT EXISTS execution_programs_source_contract_idx +ON execution_programs (project_id, source_contract_id, updated_at_unix); +"#, + )?; + + Ok(()) + } + fn record_schema_version(&self) -> Result<()> { self.connection.execute_batch( r#" @@ -537,7 +562,7 @@ CREATE TABLE IF NOT EXISTS schema_meta ( value TEXT NOT NULL ); INSERT INTO schema_meta (key, value) -VALUES ('schema_version', '9') +VALUES ('schema_version', '10') ON CONFLICT(key) DO UPDATE SET value = excluded.value; "#, )?; @@ -557,6 +582,7 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; self.load_linear_execution_events(&mut state)?; self.load_private_execution_events(&mut state)?; self.load_decision_contracts(&mut state)?; + self.load_execution_programs(&mut state)?; self.load_review_handoffs(&mut state)?; self.load_review_orchestrations(&mut state)?; self.load_review_policy_checkpoints(&mut state)?; @@ -597,6 +623,7 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; persist_linear_execution_events(&transaction, state)?; persist_private_execution_events(&transaction, state)?; persist_decision_contracts(&transaction, state)?; + persist_execution_programs(&transaction, state)?; persist_review_handoffs(&transaction, state)?; persist_review_orchestrations(&transaction, state)?; persist_review_policy_checkpoints(&transaction, state)?; @@ -623,6 +650,10 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; "DELETE FROM decision_contracts WHERE project_id = ?1", params![service_id], )?; + transaction.execute( + "DELETE FROM execution_programs WHERE project_id = ?1", + params![service_id], + )?; transaction.commit()?; Ok(()) @@ -832,6 +863,35 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + #[allow(dead_code)] + fn upsert_execution_program(&self, record: &ExecutionProgramRuntimeRecord) -> Result<()> { + let payload_json = serde_json::to_string(&record.program)?; + + self.connection.execute( + "INSERT INTO execution_programs ( + project_id, program_id, source_contract_id, payload_json, created_at, + created_at_unix, updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + ON CONFLICT(project_id, program_id) DO UPDATE SET + source_contract_id = excluded.source_contract_id, + payload_json = excluded.payload_json, + updated_at = excluded.updated_at, + updated_at_unix = excluded.updated_at_unix", + params![ + &record.project_id, + record.program.program_id(), + &record.source_contract_id, + payload_json, + &record.created_at, + record.created_at_unix, + &record.updated_at, + record.updated_at_unix, + ], + )?; + + Ok(()) + } + fn delete_lease(&mut self, issue_id: &str) -> Result<()> { self.connection .execute("DELETE FROM leases WHERE issue_id = ?1", params![issue_id])?; @@ -1546,6 +1606,71 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(()) } + 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, \ + created_at_unix, updated_at, updated_at_unix \ + 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)?, + )) + })?; + + 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)?; + + program.validate()?; + + 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() + ); + } + + 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, + }, + ); + } + + Ok(()) + } + fn load_review_handoffs(&self, state: &mut StateData) -> Result<()> { let mut statement = self.connection.prepare( "SELECT project_id, issue_id, branch_name, run_id, attempt_number, pr_url, \ @@ -1897,6 +2022,47 @@ impl DecisionContractRuntimeRecord { } } +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct ExecutionProgramKey { + project_id: String, + program_id: String, +} +impl ExecutionProgramKey { + fn new(project_id: &str, program_id: &str) -> Self { + Self { project_id: project_id.to_owned(), program_id: program_id.to_owned() } + } +} + +#[derive(Clone, Debug)] +struct ExecutionProgramRuntimeRecord { + project_id: String, + source_contract_id: String, + program: ExecutionProgram, + created_at: String, + created_at_unix: i64, + updated_at: String, + updated_at_unix: i64, +} +impl ExecutionProgramRuntimeRecord { + #[allow(dead_code)] + fn key(&self) -> ExecutionProgramKey { + ExecutionProgramKey::new(&self.project_id, self.program.program_id()) + } + + #[allow(dead_code)] + fn as_public(&self) -> ExecutionProgramRecord { + ExecutionProgramRecord { + project_id: self.project_id.clone(), + program: self.program.clone(), + source_contract_id: self.source_contract_id.clone(), + created_at: self.created_at.clone(), + created_at_unix: self.created_at_unix, + updated_at: self.updated_at.clone(), + updated_at_unix: self.updated_at_unix, + } + } +} + #[derive(Clone, Debug)] struct WorktreeMappingRecord { project_id: String, @@ -2906,6 +3072,34 @@ fn persist_decision_contracts( Ok(()) } +fn persist_execution_programs( + transaction: &Transaction<'_>, + state: &StateData, +) -> Result<()> { + for record in state.execution_programs.values() { + let payload_json = serde_json::to_string(&record.program)?; + + transaction.execute( + "INSERT OR REPLACE INTO execution_programs ( + project_id, program_id, source_contract_id, payload_json, created_at, + created_at_unix, updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + params![ + &record.project_id, + record.program.program_id(), + &record.source_contract_id, + payload_json, + &record.created_at, + record.created_at_unix, + &record.updated_at, + record.updated_at_unix, + ], + )?; + } + + Ok(()) +} + fn persist_review_handoffs(transaction: &Transaction<'_>, state: &StateData) -> Result<()> { for record in state.review_handoffs.values() { transaction.execute( @@ -3696,6 +3890,16 @@ fn compare_decision_contract_runtime_records( .then_with(|| left.contract.contract_id().cmp(right.contract.contract_id())) } +#[allow(dead_code)] +fn compare_execution_program_runtime_records( + left: &ExecutionProgramRuntimeRecord, + right: &ExecutionProgramRuntimeRecord, +) -> cmp::Ordering { + left.updated_at_unix + .cmp(&right.updated_at_unix) + .then_with(|| left.program.program_id().cmp(right.program.program_id())) +} + fn compare_project_run_status(left: &ProjectRunStatus, right: &ProjectRunStatus) -> cmp::Ordering { right .active_lease diff --git a/apps/decodex/src/state/models.rs b/apps/decodex/src/state/models.rs index ef52f6549..d20fc6eab 100644 --- a/apps/decodex/src/state/models.rs +++ b/apps/decodex/src/state/models.rs @@ -549,6 +549,52 @@ impl DecisionContractRecord { } } +/// SQLite-backed internal Execution Program retained by the local runtime. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ExecutionProgramRecord { + project_id: String, + program: ExecutionProgram, + source_contract_id: String, + created_at: String, + created_at_unix: i64, + updated_at: String, + updated_at_unix: i64, +} +#[allow(dead_code)] +impl ExecutionProgramRecord { + pub(crate) fn project_id(&self) -> &str { + &self.project_id + } + + pub(crate) fn program(&self) -> &ExecutionProgram { + &self.program + } + + pub(crate) fn program_id(&self) -> &str { + self.program.program_id() + } + + pub(crate) fn source_contract_id(&self) -> &str { + &self.source_contract_id + } + + pub(crate) fn created_at(&self) -> &str { + &self.created_at + } + + pub(crate) fn created_at_unix(&self) -> i64 { + self.created_at_unix + } + + pub(crate) fn updated_at(&self) -> &str { + &self.updated_at + } + + pub(crate) fn updated_at_unix(&self) -> i64 { + self.updated_at_unix + } +} + /// Latest runtime-owned review-policy checkpoint for one run phase. #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) struct ReviewPolicyCheckpoint { diff --git a/apps/decodex/src/state/store.rs b/apps/decodex/src/state/store.rs index 2d3975325..860b7b72c 100644 --- a/apps/decodex/src/state/store.rs +++ b/apps/decodex/src/state/store.rs @@ -1679,6 +1679,104 @@ impl StateStore { Ok(record.as_public()) } + /// Create or replace one local internal Execution Program payload. + #[allow(dead_code)] + pub(crate) fn upsert_execution_program( + &self, + project_id: &str, + program: ExecutionProgram, + ) -> Result { + validate_execution_program_record_inputs(project_id, &program)?; + + let now = timestamp_parts(); + let mut state = self.lock_without_refresh()?; + let key = ExecutionProgramKey::new(project_id, program.program_id()); + let (created_at, created_at_unix) = state + .execution_programs + .get(&key) + .map_or_else(|| (now.text.clone(), now.unix), |record| { + (record.created_at.clone(), record.created_at_unix) + }); + let record = ExecutionProgramRuntimeRecord { + project_id: project_id.to_owned(), + source_contract_id: program.source_contract_id().to_owned(), + program, + created_at, + created_at_unix, + updated_at: now.text, + updated_at_unix: now.unix, + }; + + state.execution_programs.insert(record.key(), record.clone()); + self.upsert_execution_program_locked(&record)?; + + Ok(record.as_public()) + } + + /// Read one local internal Execution Program by project and program id. + #[allow(dead_code)] + pub(crate) fn execution_program( + &self, + project_id: &str, + program_id: &str, + ) -> Result> { + validate_required_execution_program_field("project_id", project_id)?; + validate_required_execution_program_field("program_id", program_id)?; + + let state = self.lock()?; + + Ok(state + .execution_programs + .get(&ExecutionProgramKey::new(project_id, program_id)) + .map(ExecutionProgramRuntimeRecord::as_public)) + } + + /// List local internal Execution Programs derived from one Decision Contract. + #[allow(dead_code)] + pub(crate) fn list_execution_programs_for_contract( + &self, + project_id: &str, + source_contract_id: &str, + ) -> Result> { + validate_required_execution_program_field("project_id", project_id)?; + validate_required_execution_program_field("source_contract_id", source_contract_id)?; + + let state = self.lock()?; + let mut records = state + .execution_programs + .values() + .filter(|record| { + record.project_id == project_id && record.source_contract_id == source_contract_id + }) + .cloned() + .collect::>(); + + records.sort_by(compare_execution_program_runtime_records); + + Ok(records.into_iter().map(|record| record.as_public()).collect()) + } + + /// List local internal Execution Programs retained for one project. + #[allow(dead_code)] + pub(crate) fn list_execution_programs( + &self, + project_id: &str, + ) -> Result> { + validate_required_execution_program_field("project_id", project_id)?; + + let state = self.lock()?; + let mut records = state + .execution_programs + .values() + .filter(|record| record.project_id == project_id) + .cloned() + .collect::>(); + + records.sort_by(compare_execution_program_runtime_records); + + Ok(records.into_iter().map(|record| record.as_public()).collect()) + } + /// Count protocol journal records for one run. pub fn event_count(&self, run_id: &str) -> Result { let state = self.lock()?; @@ -2196,6 +2294,21 @@ impl StateStore { sqlite.upsert_decision_contract(record) } + #[allow(dead_code)] + fn upsert_execution_program_locked( + &self, + record: &ExecutionProgramRuntimeRecord, + ) -> 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_execution_program(record) + } + fn delete_lease_locked(&self, issue_id: &str) -> Result<()> { let Some(sqlite) = self.sqlite.as_ref() else { return Ok(()); @@ -2617,6 +2730,25 @@ fn validate_required_decision_contract_field(name: &str, value: &str) -> Result< Ok(()) } +#[allow(dead_code)] +fn validate_execution_program_record_inputs( + project_id: &str, + program: &ExecutionProgram, +) -> Result<()> { + validate_required_execution_program_field("project_id", project_id)?; + + program.validate() +} + +#[allow(dead_code)] +fn validate_required_execution_program_field(name: &str, value: &str) -> Result<()> { + if value.trim().is_empty() { + eyre::bail!("Execution program {name} must not be empty."); + } + + Ok(()) +} + #[cfg_attr(not(test), allow(dead_code))] fn validate_run_control_action_request(request: &RunControlActionRequest<'_>) -> Result<()> { validate_required_run_control_field("project_id", request.project_id)?; diff --git a/apps/decodex/src/state/tests.rs b/apps/decodex/src/state/tests.rs index 5c65f1b88..0d06d3611 100644 --- a/apps/decodex/src/state/tests.rs +++ b/apps/decodex/src/state/tests.rs @@ -13,6 +13,10 @@ use serde_json::Value; use tempfile::TempDir; use crate::{ + execution_program::{ + ExecutionLinearIssueMapping, ExecutionProgram, ExecutionProgramNode, + ExecutionProgramNodeStage, ExecutionQueueIntent, + }, loop_contract::{ DecisionContract, DecisionContractStatus, DecisionPromotion, DecisionPromotionActorKind, }, @@ -87,6 +91,28 @@ fn sample_decision_promotion() -> DecisionPromotion { .expect("sample promotion should validate") } +fn sample_execution_program(contract: &DecisionContract) -> ExecutionProgram { + let node = ExecutionProgramNode::new( + "runtime-readiness", + ExecutionProgramNodeStage::Runtime, + "Implement runtime readiness evaluation.", + ExecutionQueueIntent::ReadyToQueue, + ) + .expect("program node should validate") + .with_acceptance_expectations([String::from("Readiness can explain startability.")]) + .expect("acceptance expectations should attach") + .with_validation_expectations([String::from("Run the registered repo gate.")]) + .expect("validation expectations should attach") + .with_linear_issue( + ExecutionLinearIssueMapping::new("issue-853", "XY-853", "Todo") + .expect("issue mapping should validate"), + ) + .expect("issue mapping should attach"); + + ExecutionProgram::from_accepted_contract("program-853", "decodex", contract, vec![node]) + .expect("execution program should derive from accepted contract") +} + fn assert_decision_contract_retargeted(reopened: &StateStore) { assert_eq!( reopened @@ -2692,6 +2718,92 @@ fn decision_contracts_record_human_decision_and_rejection_transitions() { ); } +#[test] +fn execution_programs_persist_reload_and_list_by_contract() { + 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 mut contract = latent_decision_contract_fixture(); + + contract.promote(sample_decision_promotion()).expect("contract should promote"); + + let program = sample_execution_program(&contract); + let record = store + .upsert_execution_program("decodex", program) + .expect("execution program should persist"); + + assert_eq!(record.project_id(), "decodex"); + assert_eq!(record.program_id(), "program-853"); + assert_eq!(record.source_contract_id(), "research-x-loop-contract"); + assert_eq!(record.program().nodes().len(), 1); + assert!(record.created_at_unix() > 0); + assert!(record.updated_at_unix() >= record.created_at_unix()); + + let reopened = StateStore::open(&state_path).expect("state store should reopen"); + let reloaded = reopened + .execution_program("decodex", "program-853") + .expect("execution program should read") + .expect("execution program should exist"); + + assert_eq!(reloaded.created_at(), record.created_at()); + assert_eq!(reloaded.program().source_contract_id(), "research-x-loop-contract"); + + let contract_programs = reopened + .list_execution_programs_for_contract("decodex", "research-x-loop-contract") + .expect("contract programs should list"); + + assert_eq!(contract_programs.len(), 1); + assert_eq!(contract_programs[0].program_id(), "program-853"); + + let project_programs = + reopened.list_execution_programs("decodex").expect("project programs should list"); + + assert_eq!(project_programs.len(), 1); + assert_eq!(project_programs[0].program_id(), "program-853"); +} + +#[test] +fn execution_program_reload_rejects_row_key_payload_mismatch() { + 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 mut contract = latent_decision_contract_fixture(); + + contract.promote(sample_decision_promotion()).expect("contract should promote"); + store + .upsert_execution_program("decodex", sample_execution_program(&contract)) + .expect("execution program should persist"); + + let connection = Connection::open(&state_path).expect("sqlite should open"); + let mut payload: Value = serde_json::from_str( + &connection + .query_row( + "SELECT payload_json FROM execution_programs WHERE program_id = ?1", + ["program-853"], + |row| row.get::<_, String>(0), + ) + .expect("payload should load"), + ) + .expect("payload should parse"); + + payload["program_id"] = serde_json::json!("program-mismatch"); + + connection + .execute( + "UPDATE execution_programs SET payload_json = ?1 WHERE program_id = ?2", + [ + serde_json::to_string(&payload).expect("payload should serialize"), + String::from("program-853"), + ], + ) + .expect("payload should corrupt"); + + assert!( + StateStore::open(&state_path).is_err(), + "execution program row key must match the versioned payload program_id" + ); +} + #[test] fn decision_contract_reload_rejects_row_key_payload_mismatch() { let temp_dir = TempDir::new().expect("tempdir should create"); diff --git a/docs/reference/operator-control-plane.md b/docs/reference/operator-control-plane.md index 7d055b940..643b0df93 100644 --- a/docs/reference/operator-control-plane.md +++ b/docs/reference/operator-control-plane.md @@ -36,6 +36,10 @@ Decodex currently runs as a local, single-machine control plane: promotion and internal Execution Program state are governed by [`../spec/loop-runtime.md`](../spec/loop-runtime.md), not by dashboard graph editing or user-visible DAG commands. +- Execution Program readback is intentionally summarized: operators may see counts of + ready, blocked, paused, active, completed, stale, and queue-label-eligible nodes plus + mapped issue identifiers. Low-level node edges and graph operations remain internal + runtime state. Decodex App is a native shell over the same local runtime and account-pool state. On launch it connects to an existing default local listener when one is reachable; if @@ -148,7 +152,7 @@ work that needs full private payload values. | Surface | Owns | Does Not Own | | --- | --- | --- | -| Runtime SQLite DB | active leases, attempts, run-control channels, protocol events, private execution events, Decision Contracts, worktree mappings, retry state, retained PR state, review-policy checkpoints with structured independent-review detail, phase-goal signals, phase timing, connector backoff, project registry | human backlog grooming or durable team-visible issue history | +| Runtime SQLite DB | active leases, attempts, run-control channels, protocol events, private execution events, Decision Contracts, internal Execution Programs, worktree mappings, retry state, retained PR state, review-policy checkpoints with structured independent-review detail, phase-goal signals, phase timing, connector backoff, project registry | human backlog grooming or durable team-visible issue history | | Central project config | `service_id`, repo root, worktree root, tracker/GitHub credential env-var names, enabled project registration | per-run state or issue ownership | | Project `WORKFLOW.md` | repo policy, validation gate, state names, retry/review policy | runtime ownership, queue labels, credentials, model overrides | | Linear | team-visible issue state, queue/active/manual-attention labels, coarse execution ledger comments, progress/failure/handoff/closeout summaries | high-frequency runtime truth, heartbeat, token pressure, raw attempts, private execution evidence, connector retry budgets | diff --git a/docs/spec/linear-execution-ledger.md b/docs/spec/linear-execution-ledger.md index 3f0eb39c0..7fc5f9a7b 100644 --- a/docs/spec/linear-execution-ledger.md +++ b/docs/spec/linear-execution-ledger.md @@ -29,9 +29,10 @@ Linear comments, the Decodex runtime database, and short-lived heartbeat markers [`post-review-lifecycle.md`](./post-review-lifecycle.md), and [`tracker-tools.md`](./tracker-tools.md). Those documents define when events may be written; this document defines what the records look like. -- Decision Contracts are not Linear execution-ledger records. A ledger record may - summarize or link to generated issues after promotion, but the versioned - `decodex.decision_contract/1` payload and private loop evidence stay in runtime +- Decision Contracts and internal Execution Programs are not Linear execution-ledger + records. A ledger record may summarize or link to generated issues after promotion, + but the versioned `decodex.decision_contract/1`, + `decodex.execution_program/1`, and private loop evidence payloads stay in runtime SQLite. ## Comment body format diff --git a/docs/spec/loop-runtime.md b/docs/spec/loop-runtime.md index 94a7734e2..af0a0d5a5 100644 --- a/docs/spec/loop-runtime.md +++ b/docs/spec/loop-runtime.md @@ -162,19 +162,33 @@ An Execution Program is internal loop-runtime state derived from accepted Decisi Contracts. It may use DAG semantics, but the graph is backstage state rather than the user-facing workflow. -Each program node should carry: +The runtime-facing Execution Program payload is versioned as +`decodex.execution_program/1` with `record_version = 1`. It is stored in runtime +SQLite and carries: + +| Field | Meaning | +| --- | --- | +| `program_id` | Stable runtime identifier for this internal program. | +| `service_id` | Registered Decodex service that owns queue-label decisions. | +| `source_contract_id` | Accepted Decision Contract that authorized the program. | +| `accepted_contract_fingerprint` | Fingerprint of the accepted contract used for drift detection. | +| `nodes` | Internal executable nodes. | + +Each program node carries: - objective lineage back to the accepted Decision Contract -- executable stage such as `decision`, `issue_shaping`, `queued`, `running`, - `validation_repair`, `review_wait`, `review_repair`, `landing`, `closeout`, - `blocked`, or `done` -- dependencies and blocker references -- conflict domain -- acceptance criteria and validation gates -- queue intent and service id -- ready-node selection reason -- drift status against the accepted contract -- linked Linear issue identity when the node becomes executable +- executable stage: `research`, `design`, `spec`, `schema`, `runtime`, `plugin`, + `eval`, or `handoff` +- explicit dependencies with optional terminal-state requirements; when omitted, the + registered `WORKFLOW.md` terminal states satisfy the dependency +- conflict domains for `file`, `module`, `state`, `credentials`, + `tracker_ownership`, and `review_surface` +- acceptance expectations and validation expectations +- queue intent: `not_ready`, `ready_to_queue`, `queued`, `active`, `paused`, `done`, + or `canceled` +- linked normal Linear issue identity and startability facts when the node becomes + executable +- accepted-contract fingerprint used to detect node-level drift Normal Linear issues remain the executable Decodex lanes. A program node may become eligible only by creating or updating a normal issue with enough natural-language @@ -182,10 +196,29 @@ briefing for generic dispatch and by applying the configured queue policy. The Execution Program does not replace Linear as the team-visible backlog or the runtime lane model. -Ready-node selection is runtime-owned. It should choose nodes whose dependencies are -done, whose conflict domains are available, whose acceptance criteria are concrete, -and whose queue intent is accepted. If those facts are missing or stale, the node is -not ready. +Readiness evaluation is runtime-owned. It classifies nodes as: + +| State | Meaning | +| --- | --- | +| `not_ready` | The node is intentionally not startable yet. | +| `ready` | Dependencies, conflicts, acceptance expectations, validation expectations, and issue mapping allow normal execution. | +| `blocked` | A dependency, conflict domain, missing expectation, or Linear issue mapping blocks execution. | +| `paused` | The accepted program intentionally paused the node. | +| `active` | The node already has an active lane and must not retain the queue label. | +| `completed` | The node is `done` or `canceled`. | +| `stale` | The node or program no longer matches the accepted Decision Contract. | + +Queue-label action is derived from readiness, not from graph presence alone. Only a +`ready` node whose queue intent is `ready_to_queue` or `queued`, and whose mapped +Linear issue is in a registered startable state with no opt-out, needs-attention, +active-label, missing-briefing, or terminal-state blocker, may receive or retain +`decodex:queued:`. Non-startable, blocked, stale, active, paused, +completed, or unmapped nodes must not receive or retain that queue label. + +Operator readback may summarize program progress as counts of ready, blocked, paused, +active, completed, stale, and queue-label-eligible nodes plus the mapped issue +identifiers. It must not turn graph ids, edge editing, or DAG commands into the +primary user workflow. ## Drift Handling diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index f280d839f..0eca8ed3f 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -44,7 +44,7 @@ state or this state machine. ## Source of truth boundaries -- The Decodex runtime SQLite database is the single-machine source of truth for active leases, attempts, run-control channels, protocol events, private execution events, latent and promoted Decision Contracts, worktree mappings, retained PR state, review-policy checkpoints, retry state, phase timing, project registration, tracker cache, PR cache, and connector backoff. +- The Decodex runtime SQLite database is the single-machine source of truth for active leases, attempts, run-control channels, protocol events, private execution events, latent and promoted Decision Contracts, internal Execution Programs, worktree mappings, retained PR state, review-policy checkpoints, retry state, phase timing, project registration, tracker cache, PR cache, and connector backoff. - Linear remains the team-visible tracker surface for issue lifecycle, queue/active/manual-attention labels, and coarse lifecycle summaries such as start, PR-ready, blocked, failed, landed, and done. - Versioned Linear execution event comments use the schema in [`linear-execution-ledger.md`](./linear-execution-ledger.md), but fine-grained runtime truth must not be rebuilt from comments every tick. @@ -71,6 +71,7 @@ mirror: | --- | --- | | Runtime SQLite `private_execution_events` | Structured private execution evidence for the local Decodex installation. This is where full checkpoint payloads, verification notes, local head evidence, and recovery detail belong. | | Runtime SQLite `decision_contracts` | Versioned `decodex.decision_contract/1` payloads produced by research/design and later promoted into execution authority. The row status is indexed for local runtime lookup, but the JSON payload remains the contract authority. | +| Runtime SQLite `execution_programs` | Versioned `decodex.execution_program/1` payloads derived from accepted Decision Contracts. They hold internal node readiness, dependency, conflict-domain, queue-intent, drift, and normal-issue mapping state; Linear issue descriptions and ledger comments are only coarse projections. | | Runtime SQLite `run_control_channels` | Local control capability metadata for active run attempts. It records the project, issue, run id, attempt, transport, local channel path, channel status, and publish/update timestamps needed to route future control requests without bypassing active lease ownership. | | Runtime SQLite `review_policy_checkpoints` | Latest bounded-review checkpoint state for one project, issue, run, attempt, and phase, including structured independent-review detail. This row is the authority for review handoff and retained repair gating. | | Agent evidence under `~/.codex/decodex/agent-evidence//` | Derived local handoff view for repair agents. It may reference private evidence readback commands and compact run capsules, but it is not scheduling authority and is not a public mirror. | @@ -128,8 +129,9 @@ This boundary does not create a project-local runtime database contract. The run is `decodex.decision_contract/1`; statuses are `draft_latent`, `accepted_promoted`, `rejected_superseded`, and `needs_human_decision`. - Execution Program: Internal loop-runtime state derived from accepted Decision - Contracts. It may use DAG semantics, but normal Linear issues remain the executable - lanes. + Contracts. The runtime-facing serialized payload is + `decodex.execution_program/1`. It may use DAG semantics, but normal Linear issues + remain the executable lanes. - Terminal tracker state: A state that should not be auto-started by `decodex`. The default set is `Done`, `Canceled`, and `Duplicate`. ## Eligibility diff --git a/docs/spec/tracker-tools.md b/docs/spec/tracker-tools.md index c51ab6a5c..d61dc9b14 100644 --- a/docs/spec/tracker-tools.md +++ b/docs/spec/tracker-tools.md @@ -116,6 +116,10 @@ In either invalid case, `decodex` must fail the attempt rather than infer which tool comments or issue-description payloads. Tracker tools may later publish sparse public projections or generated issue links after promotion, but they must not copy the private `decodex.decision_contract/1` payload into Linear. +- Internal Execution Programs are local runtime records, not tracker comments or + issue-description payloads. Queue-label mutations for generated issues must follow + the Execution Program readiness evaluator: only ready nodes mapped to normal + startable Linear issues may receive or retain `decodex:queued:`. - `issue_progress_checkpoint` must accept only the normalized execution phases `probing`, `implementing`, `verifying`, `blocked`, `ready_for_review`, `review_repair`, `ready_to_land`, and `closeout`. - `issue_progress_checkpoint` must not replace `issue_review_checkpoint`, `issue_review_handoff`, `issue_review_repair_complete`, `issue_closeout_complete`, or `issue_terminal_finalize`. - `decodex` treats `issue_progress_checkpoint` as execution memory only. Checkpoint phase, focus, next action, blockers, or evidence do not by themselves authorize review handoff, repair completion, merge, closeout, or terminal success.