From 004f9a4562a54827ac6bf086098f98c2158ce026 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Thu, 11 Jun 2026 23:21:59 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Define program intake contract and queue ownership","authority":"XY-937"} --- apps/decodex/src/execution_program.rs | 459 ++++++++++++++++-- apps/decodex/src/orchestrator/status.rs | 9 +- .../tests/operator/status/text.rs | 9 +- .../tests/runtime/loop_scenarios.rs | 9 +- apps/decodex/src/orchestrator/types.rs | 21 +- docs/reference/operator-control-plane.md | 10 +- docs/spec/loop-runtime.md | 120 +++-- docs/spec/runtime.md | 12 +- docs/spec/tracker-tools.md | 5 + 9 files changed, 579 insertions(+), 75 deletions(-) diff --git a/apps/decodex/src/execution_program.rs b/apps/decodex/src/execution_program.rs index d5b8b3caf..4a6987c55 100644 --- a/apps/decodex/src/execution_program.rs +++ b/apps/decodex/src/execution_program.rs @@ -16,6 +16,145 @@ use crate::{ pub(crate) const EXECUTION_PROGRAM_SCHEMA: &str = "decodex.execution_program/1"; pub(crate) const EXECUTION_PROGRAM_RECORD_VERSION: u16 = 1; +pub(crate) const PROGRAM_INTAKE_PLAN_SCHEMA: &str = "decodex.program_intake_plan/1"; +pub(crate) const PROGRAM_INTAKE_PLAN_RECORD_VERSION: u16 = 1; + +/// Source shape for a Program Intake Plan. +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ProgramIntakeKind { + /// Natural-language goal promoted through an accepted Decision Contract. + GoalIntake, + /// Operator-supplied batch of normal issue briefs. + IssueBatchIntake, +} +impl ProgramIntakeKind { + /// Stable machine-readable intake kind. + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::GoalIntake => "goal_intake", + Self::IssueBatchIntake => "issue_batch_intake", + } + } +} + +/// Durable planning metadata for first-class program intake. +#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)] +pub(crate) struct ProgramIntakePlan { + #[serde(default = "program_intake_plan_schema")] + schema: String, + #[serde(default = "program_intake_plan_record_version")] + record_version: u16, + plan_id: String, + service_id: String, + intake_kind: ProgramIntakeKind, + #[serde(skip_serializing_if = "Option::is_none")] + source_contract_id: Option, + accepted_contract_fingerprint: String, + public_summary: String, +} +impl ProgramIntakePlan { + /// Build program-intake metadata for a promoted natural-language goal. + pub(crate) fn goal_intake( + plan_id: impl Into, + service_id: impl Into, + contract: &DecisionContract, + accepted_contract_fingerprint: impl Into, + ) -> Result { + ensure_accepted_contract(contract)?; + + let public_summary = + contract.accepted_authority().accepted_objectives().first().cloned().unwrap_or_else( + || format!("Accepted Decision Contract `{}`.", contract.contract_id()), + ); + let plan = Self { + schema: program_intake_plan_schema(), + record_version: PROGRAM_INTAKE_PLAN_RECORD_VERSION, + plan_id: plan_id.into(), + service_id: service_id.into(), + intake_kind: ProgramIntakeKind::GoalIntake, + source_contract_id: Some(contract.contract_id().to_owned()), + accepted_contract_fingerprint: accepted_contract_fingerprint.into(), + public_summary, + }; + + plan.validate()?; + + Ok(plan) + } + + /// Build program-intake metadata for an accepted issue batch. + #[allow(dead_code)] + pub(crate) fn issue_batch_intake( + plan_id: impl Into, + service_id: impl Into, + accepted_contract_fingerprint: impl Into, + public_summary: impl Into, + ) -> Result { + let plan = Self { + schema: program_intake_plan_schema(), + record_version: PROGRAM_INTAKE_PLAN_RECORD_VERSION, + plan_id: plan_id.into(), + service_id: service_id.into(), + intake_kind: ProgramIntakeKind::IssueBatchIntake, + source_contract_id: None, + accepted_contract_fingerprint: accepted_contract_fingerprint.into(), + public_summary: public_summary.into(), + }; + + plan.validate()?; + + Ok(plan) + } + + /// Program intake plan id. + pub(crate) fn plan_id(&self) -> &str { + &self.plan_id + } + + /// Intake source kind. + pub(crate) fn intake_kind(&self) -> ProgramIntakeKind { + self.intake_kind + } + + /// Accepted Decision Contract id for goal intake. + pub(crate) fn source_contract_id(&self) -> Option<&str> { + self.source_contract_id.as_deref() + } + + fn validate(&self) -> Result<()> { + validate_required("program intake plan schema", &self.schema)?; + validate_required("program intake plan plan_id", &self.plan_id)?; + validate_required("program intake plan service_id", &self.service_id)?; + validate_required( + "program intake plan accepted_contract_fingerprint", + &self.accepted_contract_fingerprint, + )?; + validate_required("program intake plan public_summary", &self.public_summary)?; + + if self.schema != PROGRAM_INTAKE_PLAN_SCHEMA { + eyre::bail!( + "Program intake plan `{}` has unsupported schema `{}`.", + self.plan_id, + self.schema + ); + } + if self.record_version != PROGRAM_INTAKE_PLAN_RECORD_VERSION { + eyre::bail!( + "Program intake plan `{}` has unsupported record_version `{}`.", + self.plan_id, + self.record_version + ); + } + if self.intake_kind == ProgramIntakeKind::GoalIntake + && self.source_contract_id.as_deref().is_none_or(str::is_empty) + { + eyre::bail!("Goal intake plan `{}` must reference a source contract.", self.plan_id); + } + + Ok(()) + } +} /// Stage for one internal Execution Program node. #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Deserialize, Serialize)] @@ -156,6 +295,48 @@ impl ExecutionReadinessState { } } +/// Durable lifecycle state for one program node. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ExecutionProgramNodeLifecycleState { + /// Node exists only as an internal plan and has no normal Linear issue yet. + Planned, + /// Node is mapped to a normal Linear issue but is intentionally held. + Mapped, + /// Node is ready to receive the service queue label. + Ready, + /// Node is ready and already carries the service queue label. + Queued, + /// Node already has an active lane. + Active, + /// Node is blocked by dependency, conflict, issue, or briefing evidence. + Blocked, + /// Node is stopped on human-required issue attention. + NeedsAttention, + /// Node is terminal. + Completed, + /// Node no longer matches the accepted contract. + Stale, + /// Node belongs to a superseded contract. + Superseded, +} +impl ExecutionProgramNodeLifecycleState { + /// Stable machine-readable state name. + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Planned => "planned", + Self::Mapped => "mapped", + Self::Ready => "ready", + Self::Queued => "queued", + Self::Active => "active", + Self::Blocked => "blocked", + Self::NeedsAttention => "needs_attention", + Self::Completed => "completed", + Self::Stale => "stale", + Self::Superseded => "superseded", + } + } +} + /// Queue-label action allowed for a mapped node. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum ExecutionQueueLabelAction { @@ -250,6 +431,8 @@ pub(crate) struct ExecutionLinearIssueMapping { issue_identifier: String, issue_state: String, has_queue_label: bool, + #[serde(default, skip_serializing_if = "is_false")] + queue_label_owned_by_program_reconciler: bool, has_active_label: bool, has_opt_out_label: bool, has_needs_attention_label: bool, @@ -267,6 +450,7 @@ impl ExecutionLinearIssueMapping { issue_identifier: issue_identifier.into(), issue_state: issue_state.into(), has_queue_label: false, + queue_label_owned_by_program_reconciler: false, has_active_label: false, has_opt_out_label: false, has_needs_attention_label: false, @@ -285,6 +469,14 @@ impl ExecutionLinearIssueMapping { self } + /// Mark that the service queue label was applied through the program reconciler. + pub(crate) fn with_program_owned_queue_label(mut self, present: bool) -> Self { + self.has_queue_label = present; + self.queue_label_owned_by_program_reconciler = 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; @@ -333,14 +525,26 @@ impl ExecutionLinearIssueMapping { self.has_queue_label } + /// Whether Decodex may remove the queue label as program-reconciler-owned. + pub(crate) fn queue_label_owned_by_program_reconciler(&self) -> bool { + self.queue_label_owned_by_program_reconciler + } + 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)?; - validate_required("execution program issue_mapping.issue_state", &self.issue_state) + if self.queue_label_owned_by_program_reconciler && !self.has_queue_label { + eyre::bail!( + "Execution program issue_mapping.queue_label_owned_by_program_reconciler requires has_queue_label.", + ); + } + + Ok(()) } } @@ -547,6 +751,8 @@ pub(crate) struct ExecutionProgram { service_id: String, source_contract_id: String, accepted_contract_fingerprint: String, + #[serde(skip_serializing_if = "Option::is_none")] + program_intake_plan: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] nodes: Vec, } @@ -560,6 +766,8 @@ impl ExecutionProgram { ) -> Result { ensure_accepted_contract(contract)?; + let program_id = program_id.into(); + let service_id = service_id.into(); let fingerprint = decision_contract_fingerprint(contract)?; for node in &mut nodes { @@ -569,10 +777,16 @@ impl ExecutionProgram { let program = Self { schema: execution_program_schema(), record_version: EXECUTION_PROGRAM_RECORD_VERSION, - program_id: program_id.into(), - service_id: service_id.into(), + program_id: program_id.clone(), + service_id: service_id.clone(), source_contract_id: contract.contract_id().to_owned(), - accepted_contract_fingerprint: fingerprint, + accepted_contract_fingerprint: fingerprint.clone(), + program_intake_plan: Some(ProgramIntakePlan::goal_intake( + program_id, + service_id, + contract, + fingerprint.clone(), + )?), nodes, }; @@ -596,6 +810,11 @@ impl ExecutionProgram { &self.source_contract_id } + /// Durable program-intake plan metadata, when the payload is not a legacy row. + pub(crate) fn program_intake_plan(&self) -> Option<&ProgramIntakePlan> { + self.program_intake_plan.as_ref() + } + /// Program nodes. pub(crate) fn nodes(&self) -> &[ExecutionProgramNode] { &self.nodes @@ -668,6 +887,37 @@ impl ExecutionProgram { ); } + if let Some(plan) = &self.program_intake_plan { + plan.validate()?; + + if plan.service_id != self.service_id { + eyre::bail!( + "Execution program `{}` belongs to service `{}` but intake plan belongs to `{}`.", + self.program_id, + self.service_id, + plan.service_id + ); + } + + if let Some(source_contract_id) = plan.source_contract_id() + && source_contract_id != self.source_contract_id + { + eyre::bail!( + "Execution program `{}` belongs to source contract `{}` but intake plan belongs to `{}`.", + self.program_id, + self.source_contract_id, + source_contract_id + ); + } + + if plan.accepted_contract_fingerprint != self.accepted_contract_fingerprint { + eyre::bail!( + "Execution program `{}` has an intake plan fingerprint mismatch.", + self.program_id + ); + } + } + let mut node_ids = HashSet::new(); for node in &self.nodes { @@ -877,6 +1127,7 @@ impl ExecutionProgramReadinessContext { pub(crate) struct ExecutionNodeEvaluation { node_id: String, state: ExecutionReadinessState, + lifecycle_state: ExecutionProgramNodeLifecycleState, reasons: Vec, queue_label_action: Option, linear_issue: Option, @@ -892,6 +1143,11 @@ impl ExecutionNodeEvaluation { self.state } + /// Durable lifecycle state used for operator program-intake readback. + pub(crate) fn lifecycle_state(&self) -> ExecutionProgramNodeLifecycleState { + self.lifecycle_state + } + /// Human-readable readiness reasons. pub(crate) fn reasons(&self) -> &[String] { &self.reasons @@ -950,25 +1206,44 @@ impl ExecutionProgramEvaluation { pub(crate) fn operator_summary(&self) -> ExecutionProgramOperatorSummary { let mut summary = ExecutionProgramOperatorSummary { program_id: self.program_id.clone(), + planned_count: 0, + mapped_count: 0, ready_count: 0, + queued_count: 0, blocked_count: 0, - paused_count: 0, + held_count: 0, active_count: 0, + needs_attention_count: 0, completed_count: 0, stale_count: 0, + superseded_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, + match node.lifecycle_state { + ExecutionProgramNodeLifecycleState::Planned => { + summary.planned_count += 1; + summary.held_count += 1; + }, + ExecutionProgramNodeLifecycleState::Mapped => { + summary.mapped_count += 1; + summary.held_count += 1; + }, + ExecutionProgramNodeLifecycleState::Ready => summary.ready_count += 1, + ExecutionProgramNodeLifecycleState::Queued => summary.queued_count += 1, + ExecutionProgramNodeLifecycleState::Blocked => summary.blocked_count += 1, + ExecutionProgramNodeLifecycleState::Active => { + summary.active_count += 1; + summary.held_count += 1; + }, + ExecutionProgramNodeLifecycleState::NeedsAttention => { + summary.needs_attention_count += 1; + }, + ExecutionProgramNodeLifecycleState::Completed => summary.completed_count += 1, + ExecutionProgramNodeLifecycleState::Stale => summary.stale_count += 1, + ExecutionProgramNodeLifecycleState::Superseded => summary.superseded_count += 1, } if node.queue_label_eligible() { @@ -992,18 +1267,28 @@ impl ExecutionProgramEvaluation { pub(crate) struct ExecutionProgramOperatorSummary { /// Program id. pub(crate) program_id: String, + /// Count of planned nodes without a normal Linear issue mapping. + pub(crate) planned_count: usize, + /// Count of mapped nodes that are intentionally held from queueing. + pub(crate) mapped_count: usize, /// Count of ready nodes. pub(crate) ready_count: usize, + /// Count of queued nodes. + pub(crate) queued_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 held nodes that are planned, mapped, or active. + pub(crate) held_count: usize, /// Count of active nodes. pub(crate) active_count: usize, + /// Count of human-attention nodes. + pub(crate) needs_attention_count: usize, /// Count of done or canceled nodes. pub(crate) completed_count: usize, /// Count of stale nodes. pub(crate) stale_count: usize, + /// Count of superseded nodes. + pub(crate) superseded_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. @@ -1034,6 +1319,7 @@ fn evaluate_node(input: EvaluateNodeInput<'_>) -> Result) -> Result) -> Result Option { + (issue.has_queue_label() && issue.queue_label_owned_by_program_reconciler()) + .then_some(ExecutionQueueLabelAction::Remove) +} + +fn lifecycle_state_for( + node: &ExecutionProgramNode, + state: ExecutionReadinessState, + queue_label_action: Option, +) -> ExecutionProgramNodeLifecycleState { + if let Some(issue) = node.linear_issue() + && issue.has_needs_attention_label + { + return ExecutionProgramNodeLifecycleState::NeedsAttention; + } + + match state { + ExecutionReadinessState::NotReady | ExecutionReadinessState::Paused => + if node.linear_issue().is_some() { + ExecutionProgramNodeLifecycleState::Mapped + } else { + ExecutionProgramNodeLifecycleState::Planned + }, + ExecutionReadinessState::Ready => { + if matches!(queue_label_action, Some(ExecutionQueueLabelAction::Retain)) { + ExecutionProgramNodeLifecycleState::Queued + } else { + ExecutionProgramNodeLifecycleState::Ready + } + }, + ExecutionReadinessState::Blocked => ExecutionProgramNodeLifecycleState::Blocked, + ExecutionReadinessState::Active => ExecutionProgramNodeLifecycleState::Active, + ExecutionReadinessState::Completed => ExecutionProgramNodeLifecycleState::Completed, + ExecutionReadinessState::Stale => ExecutionProgramNodeLifecycleState::Stale, + } } fn ensure_accepted_contract(contract: &DecisionContract) -> Result<()> { @@ -1272,6 +1607,14 @@ fn execution_program_record_version() -> u16 { EXECUTION_PROGRAM_RECORD_VERSION } +fn program_intake_plan_schema() -> String { + PROGRAM_INTAKE_PLAN_SCHEMA.to_owned() +} + +fn program_intake_plan_record_version() -> u16 { + PROGRAM_INTAKE_PLAN_RECORD_VERSION +} + fn validate_required(name: &str, value: &str) -> Result<()> { if value.trim().is_empty() { eyre::bail!("{name} must not be empty."); @@ -1292,6 +1635,10 @@ fn non_empty_optional(value: &str) -> Option<&str> { if value.is_empty() { None } else { Some(value) } } +fn is_false(value: &bool) -> bool { + !*value +} + fn validate_string_list(name: &str, values: &[String]) -> Result<()> { for value in values { validate_required(name, value)?; @@ -1306,9 +1653,9 @@ mod tests { execution_program::{ ExecutionConflictDomain, ExecutionConflictDomainKind, ExecutionDependencySnapshot, ExecutionLinearIssueMapping, ExecutionProgram, ExecutionProgramDependency, - ExecutionProgramNode, ExecutionProgramNodeStage, ExecutionProgramReadinessContext, - ExecutionQueueIntent, ExecutionQueueLabelAction, ExecutionReadinessState, - ExecutionWorkflowPolicy, + ExecutionProgramNode, ExecutionProgramNodeLifecycleState, ExecutionProgramNodeStage, + ExecutionProgramReadinessContext, ExecutionQueueIntent, ExecutionQueueLabelAction, + ExecutionReadinessState, ExecutionWorkflowPolicy, ProgramIntakeKind, }, loop_contract::{DecisionContract, DecisionPromotion, DecisionPromotionActorKind}, }; @@ -1409,6 +1756,35 @@ mod tests { assert_eq!(evaluation.operator_summary().blocked_count, 1); } + #[test] + fn accepted_contract_program_carries_goal_intake_metadata() { + let (contract, program) = program_with(vec![ready_node("node-ready", "XY-900")]); + let plan = program.program_intake_plan().expect("new programs should carry intake plan"); + + assert_eq!(plan.plan_id(), "program-1"); + assert_eq!(plan.intake_kind(), ProgramIntakeKind::GoalIntake); + assert_eq!(plan.source_contract_id(), Some(contract.contract_id())); + } + + #[test] + fn legacy_execution_program_payload_without_intake_plan_still_validates() { + let (_contract, program) = program_with(vec![ready_node("node-ready", "XY-900")]); + let mut payload = + serde_json::to_value(&program).expect("program payload should serialize to json"); + + payload + .as_object_mut() + .expect("program payload should be an object") + .remove("program_intake_plan"); + + let legacy_program: ExecutionProgram = + serde_json::from_value(payload).expect("legacy program should deserialize"); + + legacy_program.validate().expect("legacy program should validate"); + + assert!(legacy_program.program_intake_plan().is_none()); + } + #[test] fn dependency_blocking_respects_workflow_terminal_states() { let dependent = ready_node("node-dependent", "XY-902") @@ -1491,12 +1867,19 @@ mod tests { 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)) + .with_linear_issue(issue("XY-906", "Todo").with_program_owned_queue_label(true)) + .expect("issue should attach"); + let ready_human_owned = ready_node("node-ready-human-owned", "XY-910") + .with_linear_issue(issue("XY-910", "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)) + .with_linear_issue(issue("XY-907", "In Progress").with_program_owned_queue_label(true)) + .expect("issue should attach"); + let human_owned = ready_node("node-human-owned", "XY-909") + .with_linear_issue(issue("XY-909", "In Progress").with_queue_label(true)) .expect("issue should attach"); - let (contract, program) = program_with(vec![apply, retain, remove]); + let (contract, program) = + program_with(vec![apply, retain, ready_human_owned, remove, human_owned]); let evaluation = program .evaluate(&contract, &workflow_policy(), &ExecutionProgramReadinessContext::new()) .expect("program should evaluate"); @@ -1511,7 +1894,9 @@ mod tests { assert_eq!(action_for("node-apply"), Some(ExecutionQueueLabelAction::Apply)); assert_eq!(action_for("node-retain"), Some(ExecutionQueueLabelAction::Retain)); + assert_eq!(action_for("node-ready-human-owned"), None); assert_eq!(action_for("node-remove"), Some(ExecutionQueueLabelAction::Remove)); + assert_eq!(action_for("node-human-owned"), None); assert_eq!( evaluation .nodes() @@ -1521,6 +1906,24 @@ mod tests { .state(), ExecutionReadinessState::Blocked ); + assert_eq!( + evaluation + .nodes() + .iter() + .find(|node| node.node_id() == "node-retain") + .expect("node should exist") + .lifecycle_state(), + ExecutionProgramNodeLifecycleState::Queued + ); + assert_eq!( + evaluation + .nodes() + .iter() + .find(|node| node.node_id() == "node-ready-human-owned") + .expect("node should exist") + .lifecycle_state(), + ExecutionProgramNodeLifecycleState::Ready + ); } #[test] diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index cf1f2f4f0..6c9bc04bb 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -6511,16 +6511,21 @@ fn append_rendered_execution_programs(output: &mut String, snapshot: &OperatorSt .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_id: {} source_contract_id: {} nodes={} planned={} mapped={} ready={} queued={} blocked={} held={} active={} attention={} completed={} stale={} superseded={} queue_label_eligible={} mapped_issues={}{}\n", program.program_id, program.source_contract_id, program.node_count, + program.planned_count, + program.mapped_count, program.ready_count, + program.queued_count, program.blocked_count, - program.paused_count, + program.held_count, program.active_count, + program.needs_attention_count, program.completed_count, program.stale_count, + program.superseded_count, program.queue_label_eligible_count, mapped_issues, readback_warning, diff --git a/apps/decodex/src/orchestrator/tests/operator/status/text.rs b/apps/decodex/src/orchestrator/tests/operator/status/text.rs index bce57b240..e43f1ca88 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/text.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/text.rs @@ -183,12 +183,17 @@ fn operator_status_text_surfaces_execution_program_summary() { program_id: String::from("program-853"), source_contract_id: String::from("contract-852"), node_count: 3, + planned_count: 0, + mapped_count: 0, ready_count: 1, + queued_count: 0, blocked_count: 1, - paused_count: 0, + held_count: 0, active_count: 0, + needs_attention_count: 0, completed_count: 1, stale_count: 0, + superseded_count: 0, queue_label_eligible_count: 1, mapped_issue_identifiers: vec![String::from("XY-853")], readback_warning: None, @@ -201,7 +206,7 @@ fn operator_status_text_surfaces_execution_program_summary() { 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" + "program_id: program-853 source_contract_id: contract-852 nodes=3 planned=0 mapped=0 ready=1 queued=0 blocked=1 held=0 active=0 attention=0 completed=1 stale=0 superseded=0 queue_label_eligible=1 mapped_issues=XY-853" )); } diff --git a/apps/decodex/src/orchestrator/tests/runtime/loop_scenarios.rs b/apps/decodex/src/orchestrator/tests/runtime/loop_scenarios.rs index a6ce2ccce..079305cb6 100644 --- a/apps/decodex/src/orchestrator/tests/runtime/loop_scenarios.rs +++ b/apps/decodex/src/orchestrator/tests/runtime/loop_scenarios.rs @@ -142,9 +142,12 @@ impl LoopScenarioHarness { assert_eq!(evaluation.ready_node_ids(), vec!["node-ready"]); assert_eq!(evaluation.startable_node_ids(), vec!["node-ready"]); assert_eq!(summary.ready_count, 1); + assert_eq!(summary.queued_count, 0); assert_eq!(summary.blocked_count, 1); - assert_eq!(summary.paused_count, 1); + assert_eq!(summary.mapped_count, 1); + assert_eq!(summary.held_count, 2); assert_eq!(summary.active_count, 1); + assert_eq!(summary.needs_attention_count, 0); assert_eq!(summary.queue_label_eligible_count, 1); assert_eq!( loop_scenario_queue_action(evaluation, "node-ready"), @@ -609,7 +612,7 @@ fn loop_scenario_active_node() -> crate::execution_program::ExecutionProgramNode .with_linear_issue( ExecutionLinearIssueMapping::new("linear-node-active", "XY-864", "Todo") .expect("issue mapping should build") - .with_queue_label(true) + .with_program_owned_queue_label(true) .with_active_label(true), ) .expect("active issue should attach") @@ -759,7 +762,7 @@ fn loop_scenario_node( "Todo", ) .expect("issue mapping should build") - .with_queue_label(has_queue_label), + .with_program_owned_queue_label(has_queue_label), ) .expect("issue mapping should attach") } diff --git a/apps/decodex/src/orchestrator/types.rs b/apps/decodex/src/orchestrator/types.rs index 9b0d1b77c..c3a806c94 100644 --- a/apps/decodex/src/orchestrator/types.rs +++ b/apps/decodex/src/orchestrator/types.rs @@ -1293,12 +1293,17 @@ struct OperatorExecutionProgramStatus { program_id: String, source_contract_id: String, node_count: usize, + planned_count: usize, + mapped_count: usize, ready_count: usize, + queued_count: usize, blocked_count: usize, - paused_count: usize, + held_count: usize, active_count: usize, + needs_attention_count: usize, completed_count: usize, stale_count: usize, + superseded_count: usize, queue_label_eligible_count: usize, mapped_issue_identifiers: Vec, readback_warning: Option, @@ -1312,12 +1317,17 @@ impl OperatorExecutionProgramStatus { program_id: summary.program_id, source_contract_id: record.source_contract_id().to_owned(), node_count: record.program().nodes().len(), + planned_count: summary.planned_count, + mapped_count: summary.mapped_count, ready_count: summary.ready_count, + queued_count: summary.queued_count, blocked_count: summary.blocked_count, - paused_count: summary.paused_count, + held_count: summary.held_count, active_count: summary.active_count, + needs_attention_count: summary.needs_attention_count, completed_count: summary.completed_count, stale_count: summary.stale_count, + superseded_count: summary.superseded_count, queue_label_eligible_count: summary.queue_label_eligible_count, mapped_issue_identifiers: summary.mapped_issue_identifiers, readback_warning: None, @@ -1331,12 +1341,17 @@ impl OperatorExecutionProgramStatus { program_id: record.program_id().to_owned(), source_contract_id: record.source_contract_id().to_owned(), node_count, + planned_count: 0, + mapped_count: 0, ready_count: 0, + queued_count: 0, blocked_count: 0, - paused_count: 0, + held_count: 0, active_count: 0, + needs_attention_count: 0, completed_count: 0, stale_count: node_count, + superseded_count: 0, queue_label_eligible_count: 0, mapped_issue_identifiers: Vec::new(), readback_warning: Some(String::from("source_decision_contract_missing")), diff --git a/docs/reference/operator-control-plane.md b/docs/reference/operator-control-plane.md index 640b795b4..50c27d0db 100644 --- a/docs/reference/operator-control-plane.md +++ b/docs/reference/operator-control-plane.md @@ -36,9 +36,11 @@ 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 +- Program intake readback is intentionally summarized: operators may see counts of + ready, queued, blocked, held, stale, attention, completed, and + queue-label-eligible nodes plus mapped issue identifiers. Optional planned, mapped, + active, and superseded counts may appear when the runtime has that detail. Low-level + node edges, graph operations, and queue-label ownership records remain internal runtime state. Decodex App is a native shell over the same local runtime and account-pool state. On @@ -164,7 +166,7 @@ hidden unless `--include-payload` is explicitly requested for local repair. | Surface | Owns | Does Not Own | | --- | --- | --- | -| 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, loop-guardrail checkpoints, 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, Program Intake Plans, internal Execution Programs, program-owned queue-label evidence, worktree mappings, retry state, retained PR state, review-policy checkpoints with structured independent-review detail, loop-guardrail checkpoints, 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/loop-runtime.md b/docs/spec/loop-runtime.md index a7e641a73..057c6d262 100644 --- a/docs/spec/loop-runtime.md +++ b/docs/spec/loop-runtime.md @@ -295,11 +295,45 @@ Promotion may create or update normal Linear issues, dependencies, labels, or qu intent. It must not expose an ordinary user workflow that depends on graph ids, manual DAG edge editing, or hidden Codex goal state. +## Program Intake Plan + +Program intake is first-class loop-runtime behavior after a Decision Contract is +accepted or after the user supplies an explicit executable issue-batch intake. The +ordinary user workflow stays natural language: the user may ask Decodex to push an +accepted goal forward or provide a batch of issue briefs, but the user does not edit a +DAG, set queue labels by hand, or operate hidden graph commands. + +The durable intake-planning payload is versioned as +`decodex.program_intake_plan/1` with `record_version = 1`. The runtime stores it in +runtime SQLite as part of, or directly adjacent to, the internal +`decodex.execution_program/1` record. Linear issue descriptions, Linear comments, and +operator summaries may expose only sparse public projections. + +The payload carries: + +| Field | Meaning | +| --- | --- | +| `plan_id` | Stable runtime id for this intake plan. | +| `service_id` | Registered service that owns program reconciliation for this plan. | +| `intake_kind` | `goal_intake` for promoted natural-language goals, or `issue_batch_intake` for a supplied batch of executable issue briefs. | +| `source_contract_id` | Accepted Decision Contract id for goal intake, when present. | +| `accepted_contract_fingerprint` | Fingerprint used to detect drift from the accepted contract or batch boundary. | +| `public_summary` | Public-safe objective/readiness summary suitable for status readback. | +| `node_projection` | Optional public-safe node summary or count metadata. The full internal nodes, dependencies, conflict domains, issue mapping, queue-label ownership metadata, and lifecycle evaluation inputs live in the paired `decodex.execution_program/1` payload. | + +Goal intake and issue-batch intake both materialize normal Linear issues. A goal +intake starts from an accepted Decision Contract and shapes one or more issue briefs. +An issue-batch intake starts from a supplied batch of issue briefs and records the +batch boundary as the accepted authority. In both cases, dependencies and ordering may +be represented internally as a DAG, but executable work still enters Decodex as +ordinary Linear issue lanes with generic natural-language descriptions, tracker +states, validation expectations, and Decodex lifecycle writeback. + ## Internal Execution Program -An Execution Program is internal loop-runtime state derived from accepted Decision -Contracts. It may use DAG semantics, but the graph is backstage state rather than the -user-facing workflow. +An Execution Program is internal loop-runtime state derived from an accepted Program +Intake Plan. It may use DAG semantics, but the graph is backstage state rather than +the user-facing workflow. The runtime-facing Execution Program payload is versioned as `decodex.execution_program/1` with `record_version = 1`. It is stored in runtime @@ -308,14 +342,15 @@ 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. | +| `service_id` | Registered Decodex service that owns program reconciliation. | +| `source_contract_id` | Accepted Decision Contract that authorized the program, when the program came from goal intake. | +| `accepted_contract_fingerprint` | Fingerprint of the accepted contract or batch authority used for drift detection. | +| `program_intake_plan` | The embedded or linked `decodex.program_intake_plan/1` payload. | | `nodes` | Internal executable nodes. | Each program node carries: -- objective lineage back to the accepted Decision Contract +- objective lineage back to the accepted Decision Contract or issue-batch authority - executable stage: `research`, `design`, `spec`, `schema`, `runtime`, `plugin`, `eval`, or `handoff` - explicit dependencies with optional terminal-state requirements; when omitted, the @@ -323,11 +358,15 @@ Each program node carries: - conflict domains for `file`, `module`, `state`, `credentials`, `tracker_ownership`, and `review_surface` - acceptance expectations and validation expectations +- runtime-derived lifecycle state: `planned`, `mapped`, `ready`, `queued`, `active`, + `blocked`, `needs_attention`, `completed`, `stale`, or `superseded` - queue intent: `not_ready`, `ready_to_queue`, `queued`, `active`, `paused`, `done`, - or `canceled` + or `canceled`; `paused` is legacy queue intent and renders as held lifecycle + readback, not as a user-facing graph state - linked normal Linear issue identity and startability facts when the node becomes executable -- accepted-contract fingerprint used to detect node-level drift +- queue-label ownership metadata for labels applied by program reconciliation +- accepted authority 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 @@ -335,29 +374,52 @@ 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. -Readiness evaluation is runtime-owned. It classifies nodes as: +Lifecycle 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. +| `planned` | The node exists only inside the Program Intake Plan and has no normal Linear issue mapping yet. | +| `mapped` | The node has a normal Linear issue mapping but is intentionally held from queueing. | +| `ready` | Dependencies, conflicts, acceptance expectations, validation expectations, and issue mapping allow normal execution, but the queue label has not been applied or retained yet. | +| `queued` | The node is ready and its mapped issue currently carries a program-owned service queue label. | +| `active` | The mapped issue already has an active lane and must not retain program-owned queue labels. | +| `blocked` | A dependency, conflict domain, missing expectation, non-startable issue state, missing issue mapping, or missing briefing blocks execution. | +| `needs_attention` | The mapped issue carries the configured human-attention label or equivalent human-required stop. | +| `completed` | The node is done or canceled under the accepted program. | +| `stale` | The node or program no longer matches the accepted authority fingerprint. | +| `superseded` | A later accepted contract or batch authority replaced this node. | + +Queue-label action is derived from lifecycle 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:`. + +Queue-label ownership is fail-closed: + +- Apply the service queue label only when the node is `ready`, the mapped issue lacks + the label, and program reconciliation records ownership for that node, issue, label, + service id, and program id. +- Retain the label only while the node remains `ready` or `queued`, the mapped issue + remains startable, and the program-owned label record still matches the same node + and service label. A human-owned queued label may be left in place but must not be + converted into program-owned removal authority. +- Remove the label only when a matching program-owned label record proves Decodex + applied it through program reconciliation and the node is no longer queueable, for + example `blocked`, `active`, `needs_attention`, `completed`, `stale`, + `superseded`, unmapped, terminal, or missing a generic dispatch briefing. +- If the queue label is present but no matching program-owned ownership record exists, + treat it as human-owned or ownership-unknown. Do not remove it during program + reconciliation; surface a public-safe ownership warning or local operator reason + instead. + +Operator readback must summarize program progress without exposing graph operations as +workflow. Public-safe readback fields are: intake kind, public summary, ready count, +queued count, blocked count, held count, stale count, attention count, completed +count, optional planned/mapped/active/superseded counts, queue-label-eligible count, +mapped issue identifiers, and coarse next action. It must not expose internal graph +ids, edge lists, raw dependency payloads, host-local paths, transcripts, credentials, +or private evidence references as public/team-visible status. ## Drift Handling diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index 524c132a0..e52de79e7 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -71,7 +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, recovery detail, and `decodex.harness_outcome/1` feedback records 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 `execution_programs` | Versioned `decodex.execution_program/1` payloads with embedded or linked `decodex.program_intake_plan/1` planning data. They hold internal node lifecycle/readiness, dependency, conflict-domain, queue-intent, drift, normal-issue mapping, and program-owned queue-label evidence; 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. | | Runtime SQLite `loop_guardrail_checkpoints` | Latest convergence checkpoint for one project, issue, and guardrail reason. It stores the fingerprint, consecutive count, run id, attempt number, and structured detail used to stop non-converging loops without replaying Linear comments. | @@ -138,9 +138,10 @@ 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. The runtime-facing serialized payload is - `decodex.execution_program/1`. It may use DAG semantics, but normal Linear issues - remain the executable lanes. + Contracts or accepted issue-batch intake. The durable planning payload is + `decodex.program_intake_plan/1`, stored with or adjacent to the + `decodex.execution_program/1` payload. It may use DAG semantics, but normal Linear + issues remain the executable lanes. - Authority Envelope: The accepted boundary from the Decision Contract, project policy, issue briefing, and explicit user direction. Lane recovery may change implementation details inside this envelope, but product goals, accepted behavior, @@ -571,6 +572,9 @@ The runtime database stores at least: - private execution events scoped by project, issue, run, and attempt - Decision Contracts scoped by project and contract id, with optional source issue linkage for later issue shaping +- Program Intake Plans and internal Execution Programs scoped by project, with + lifecycle/readiness state, normal Linear issue mappings, and program-owned + queue-label ownership evidence - worktree mappings - retained PR and post-review state - review-policy checkpoints diff --git a/docs/spec/tracker-tools.md b/docs/spec/tracker-tools.md index 3329a6663..89ec6548d 100644 --- a/docs/spec/tracker-tools.md +++ b/docs/spec/tracker-tools.md @@ -120,6 +120,11 @@ In either invalid case, `decodex` must fail the attempt rather than infer which 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:`. + Program reconciliation must record when it applied that label for a specific + service, program, node, and issue. It may remove only labels with matching + program-owned evidence. If the same queue label is present without matching + ownership evidence, the label is human-owned or ownership-unknown and must be left + in place. - `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.