From 434d2ffe426236e7de30efed9ce89b7d4b92373e Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Fri, 12 Jun 2026 00:09:10 +0800 Subject: [PATCH 1/2] {"schema":"decodex/commit/1","summary":"Implement Program Intake state and issue-batch dry-run CLI","authority":"XY-938"} --- README.md | 1 + apps/decodex/src/cli.rs | 127 ++- apps/decodex/src/execution_program.rs | 194 +++- apps/decodex/src/lib.rs | 1 + .../src/orchestrator/dispatch_policy.rs | 8 +- .../src/orchestrator/harness_improvement.rs | 4 +- apps/decodex/src/orchestrator/status.rs | 18 +- .../tests/operator/status/text.rs | 2 +- apps/decodex/src/orchestrator/types.rs | 6 +- apps/decodex/src/program_intake.rs | 968 ++++++++++++++++++ apps/decodex/src/state/internal.rs | 820 ++++++++++++++- apps/decodex/src/state/models.rs | 222 +++- apps/decodex/src/state/store.rs | 117 ++- apps/decodex/src/state/tests.rs | 32 +- docs/spec/loop-runtime.md | 17 +- docs/spec/runtime.md | 10 + docs/spec/tracker-tools.md | 5 + 17 files changed, 2478 insertions(+), 74 deletions(-) create mode 100644 apps/decodex/src/program_intake.rs diff --git a/README.md b/README.md index b70016f5f..f51140524 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,7 @@ cargo run -p decodex --bin decodex -- lane steer --run-id --exp cargo run -p decodex --bin decodex -- research compile --intent "research X" cargo run -p decodex --bin decodex -- research compile --input research-design-run.json cargo run -p decodex --bin decodex -- research promote +cargo run -p decodex --bin decodex -- intake issues --project decodex XY-1 XY-2 --dry-run cargo run -p decodex --bin decodex -- radar refresh-upstream-queue cargo run -p decodex --bin decodex -- radar refresh-release-delta cargo run -p decodex --bin decodex -- radar validate diff --git a/apps/decodex/src/cli.rs b/apps/decodex/src/cli.rs index 188cbaad9..f5d126709 100644 --- a/apps/decodex/src/cli.rs +++ b/apps/decodex/src/cli.rs @@ -26,6 +26,7 @@ use crate::{ LaneSteerRequest, RunOnceRequest, ServeRequest, }, prelude::{Result, eyre}, + program_intake::{self, IssueBatchIntakeCommandRequest}, radar::{ self, RadarBackfillReleaseRangeRequest, RadarBundleBuildRequest, RadarBundleValidateRequest, RadarLedgerArtifactLinkRequest, RadarLedgerBootstrapRequest, @@ -77,6 +78,7 @@ impl Cli { Command::Diagnose(args) => args.run(), Command::Evidence(args) => args.run(), Command::Research(args) => args.run(), + Command::Intake(args) => args.run(), Command::Recover(args) => args.run(), Command::ArchiveLinear(args) => args.run(), Command::Maintenance(args) => args.run(), @@ -618,6 +620,61 @@ impl EvidenceCommand { } } +#[derive(Debug, Args)] +struct IntakeCommand { + #[command(subcommand)] + command: IntakeSubcommand, +} +impl IntakeCommand { + fn run(&self) -> Result<()> { + match &self.command { + IntakeSubcommand::Issues(args) => args.run(), + } + } +} + +#[derive(Debug, Args)] +struct IntakeIssuesCommand { + #[command(flatten)] + project_config: ProjectConfigArgs, + /// Registered Decodex service id to intake against. + #[arg(long, value_name = "SERVICE_ID", conflicts_with = "config")] + project: Option, + /// Read tracker state and print the deterministic intake report without local persistence. + #[arg(long, conflicts_with = "persist", required_unless_present = "persist")] + dry_run: bool, + /// Persist only local runtime Program Intake records; never mutates Linear queue labels. + #[arg(long, conflicts_with = "dry_run")] + persist: bool, + /// Emit structured JSON instead of human-readable text. + #[arg(long)] + json: bool, + /// Existing Linear issue identifiers to intake into an internal Execution Program. + #[arg(value_name = "ISSUE")] + #[arg(required = true)] + issues: Vec, +} +impl IntakeIssuesCommand { + fn run(&self) -> Result<()> { + let report = + program_intake::run_issue_batch_intake_command(IssueBatchIntakeCommandRequest { + config_path: self.project_config.as_path(), + project_id: self.project.as_deref(), + issue_identifiers: self.issues.clone(), + dry_run: self.dry_run, + persist: self.persist, + })?; + + if self.json { + println!("{}", serde_json::to_string_pretty(&report)?); + } else { + print!("{}", program_intake::render_issue_batch_intake_report(&report)); + } + + Ok(()) + } +} + #[derive(Debug, Args)] struct ResearchCommand { #[command(flatten)] @@ -1477,6 +1534,8 @@ enum Command { Evidence(EvidenceCommand), /// Compile or promote Decodex-native research/design contracts. Research(ResearchCommand), + /// Operator issue-batch intake into internal Execution Programs, not a graph editor. + Intake(IntakeCommand), /// Diagnose or explicitly repair supported retained-lane recovery cases. Recover(RecoverCommand), /// Dry-run or archive old terminal Linear issues by repo label. @@ -1544,6 +1603,12 @@ enum ResearchSubcommand { Promote(ResearchPromoteCommand), } +#[derive(Debug, Subcommand)] +enum IntakeSubcommand { + /// Dry-run or persist existing Linear issues as an internal program intake batch. + Issues(IntakeIssuesCommand), +} + #[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] #[value(rename_all = "kebab-case")] enum ResearchOutcomeArg { @@ -1708,18 +1773,19 @@ mod tests { use crate::cli::{ AccountCommand, AccountSubcommand, AccountUseCommand, AttemptCommand, Cli, Command, - CommitCommand, DiagnoseCommand, EvidenceCommand, LandCommand, LaneCommand, - LaneInspectCommand, LaneInterruptCommand, LaneSteerCommand, LaneSubcommand, - LegacyCloseoutRecoveryCommand, ProbeCommand, ProjectCommand, ProjectConfigArgs, - ProjectSubcommand, RadarBackfillReleaseRangeCommand, RadarBundleBuildCommand, - RadarBundleCommand, RadarBundleSubcommand, RadarBundleValidateCommand, RadarCommand, - RadarLedgerCommand, RadarLedgerIngestExistingCommand, RadarLedgerSubcommand, - RadarLedgerSummaryCommand, RadarRefreshReleaseDeltaCommand, - RadarRefreshUpstreamQueueCommand, RadarRenderSignalCommand, RadarSubcommand, - RadarValidateCommand, RecoverCommand, RecoverSubcommand, ResearchCommand, - ResearchCompileCommand, ResearchOutcomeArg, ResearchPromoteCommand, ResearchSubcommand, - ReviewHandoffDiagnoseCommand, ReviewHandoffRebindCommand, ReviewHandoffRecoveryCommand, - ReviewHandoffRecoverySubcommand, RunCommand, ServeCommand, StatusCommand, + CommitCommand, DiagnoseCommand, EvidenceCommand, IntakeCommand, IntakeIssuesCommand, + IntakeSubcommand, LandCommand, LaneCommand, LaneInspectCommand, LaneInterruptCommand, + LaneSteerCommand, LaneSubcommand, LegacyCloseoutRecoveryCommand, ProbeCommand, + ProjectCommand, ProjectConfigArgs, ProjectSubcommand, RadarBackfillReleaseRangeCommand, + RadarBundleBuildCommand, RadarBundleCommand, RadarBundleSubcommand, + RadarBundleValidateCommand, RadarCommand, RadarLedgerCommand, + RadarLedgerIngestExistingCommand, RadarLedgerSubcommand, RadarLedgerSummaryCommand, + RadarRefreshReleaseDeltaCommand, RadarRefreshUpstreamQueueCommand, + RadarRenderSignalCommand, RadarSubcommand, RadarValidateCommand, RecoverCommand, + RecoverSubcommand, ResearchCommand, ResearchCompileCommand, ResearchOutcomeArg, + ResearchPromoteCommand, ResearchSubcommand, ReviewHandoffDiagnoseCommand, + ReviewHandoffRebindCommand, ReviewHandoffRecoveryCommand, ReviewHandoffRecoverySubcommand, + RunCommand, ServeCommand, StatusCommand, }; #[test] @@ -2471,6 +2537,43 @@ mod tests { )); } + #[test] + fn parses_intake_issues_dry_run_with_project() { + let cli = Cli::parse_from([ + "decodex", + "intake", + "issues", + "--project", + "decodex", + "XY-1", + "XY-2", + "--dry-run", + "--json", + ]); + + assert!(matches!( + cli.command, + Command::Intake(IntakeCommand { + command: IntakeSubcommand::Issues(IntakeIssuesCommand { + project: Some(_), + dry_run: true, + persist: false, + json: true, + issues, + .. + }) + }) if issues == vec![String::from("XY-1"), String::from("XY-2")] + )); + } + + #[test] + fn rejects_intake_issues_without_explicit_mode() { + let error = Cli::try_parse_from(["decodex", "intake", "issues", "XY-1"]) + .expect_err("intake issues requires dry-run or persist"); + + assert!(error.to_string().contains("--dry-run") || error.to_string().contains("--persist")); + } + #[test] fn parses_research_promote_with_acceptance_metadata() { let cli = Cli::parse_from([ diff --git a/apps/decodex/src/execution_program.rs b/apps/decodex/src/execution_program.rs index 4a6987c55..d298c05e3 100644 --- a/apps/decodex/src/execution_program.rs +++ b/apps/decodex/src/execution_program.rs @@ -112,6 +112,11 @@ impl ProgramIntakePlan { &self.plan_id } + /// Service id that owns this intake plan. + pub(crate) fn service_id(&self) -> &str { + &self.service_id + } + /// Intake source kind. pub(crate) fn intake_kind(&self) -> ProgramIntakeKind { self.intake_kind @@ -122,6 +127,16 @@ impl ProgramIntakePlan { self.source_contract_id.as_deref() } + /// Stable authority fingerprint for this intake boundary. + pub(crate) fn accepted_contract_fingerprint(&self) -> &str { + &self.accepted_contract_fingerprint + } + + /// Public-safe summary suitable for operator readback. + pub(crate) fn public_summary(&self) -> &str { + &self.public_summary + } + fn validate(&self) -> Result<()> { validate_required("program intake plan schema", &self.schema)?; validate_required("program intake plan plan_id", &self.plan_id)?; @@ -151,6 +166,14 @@ impl ProgramIntakePlan { { eyre::bail!("Goal intake plan `{}` must reference a source contract.", self.plan_id); } + if self.intake_kind == ProgramIntakeKind::IssueBatchIntake + && self.source_contract_id.as_deref().is_some_and(|id| !id.is_empty()) + { + eyre::bail!( + "Issue-batch intake plan `{}` must not reference a source contract.", + self.plan_id + ); + } Ok(()) } @@ -530,6 +553,26 @@ impl ExecutionLinearIssueMapping { self.queue_label_owned_by_program_reconciler } + /// Whether the service active label is currently present. + pub(crate) fn has_active_label(&self) -> bool { + self.has_active_label + } + + /// Whether the configured opt-out label is currently present. + pub(crate) fn has_opt_out_label(&self) -> bool { + self.has_opt_out_label + } + + /// Whether the configured human-attention label is currently present. + pub(crate) fn has_needs_attention_label(&self) -> bool { + self.has_needs_attention_label + } + + /// Whether the issue description is usable as a generic dispatch briefing. + pub(crate) fn has_generic_dispatch_briefing(&self) -> bool { + self.has_generic_dispatch_briefing + } + fn validate(&self) -> Result<()> { validate_required("execution program issue_mapping.issue_id", &self.issue_id)?; validate_required( @@ -749,7 +792,8 @@ pub(crate) struct ExecutionProgram { record_version: u16, program_id: String, service_id: String, - source_contract_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + source_contract_id: Option, accepted_contract_fingerprint: String, #[serde(skip_serializing_if = "Option::is_none")] program_intake_plan: Option, @@ -779,7 +823,7 @@ impl ExecutionProgram { record_version: EXECUTION_PROGRAM_RECORD_VERSION, program_id: program_id.clone(), service_id: service_id.clone(), - source_contract_id: contract.contract_id().to_owned(), + source_contract_id: Some(contract.contract_id().to_owned()), accepted_contract_fingerprint: fingerprint.clone(), program_intake_plan: Some(ProgramIntakePlan::goal_intake( program_id, @@ -795,6 +839,43 @@ impl ExecutionProgram { Ok(program) } + /// Build an internal Execution Program from an accepted issue-batch intake boundary. + pub(crate) fn from_issue_batch_intake( + program_id: impl Into, + service_id: impl Into, + accepted_batch_fingerprint: impl Into, + public_summary: impl Into, + mut nodes: Vec, + ) -> Result { + let program_id = program_id.into(); + let service_id = service_id.into(); + let fingerprint = accepted_batch_fingerprint.into(); + + 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.clone(), + service_id: service_id.clone(), + source_contract_id: None, + accepted_contract_fingerprint: fingerprint.clone(), + program_intake_plan: Some(ProgramIntakePlan::issue_batch_intake( + program_id, + service_id, + fingerprint, + public_summary, + )?), + nodes, + }; + + program.validate()?; + + Ok(program) + } + /// Stable internal program id. pub(crate) fn program_id(&self) -> &str { &self.program_id @@ -805,9 +886,14 @@ impl ExecutionProgram { &self.service_id } - /// Accepted Decision Contract id that authorized this program. - pub(crate) fn source_contract_id(&self) -> &str { - &self.source_contract_id + /// Accepted Decision Contract id that authorized this program, for goal intake. + pub(crate) fn source_contract_id(&self) -> Option<&str> { + self.source_contract_id.as_deref() + } + + /// Stable authority fingerprint for this program. + pub(crate) fn accepted_contract_fingerprint(&self) -> &str { + &self.accepted_contract_fingerprint } /// Durable program-intake plan metadata, when the payload is not a legacy row. @@ -826,6 +912,41 @@ impl ExecutionProgram { current_contract: &DecisionContract, policy: &ExecutionWorkflowPolicy, context: &ExecutionProgramReadinessContext, + ) -> Result { + if self.source_contract_id.as_deref().is_none() { + eyre::bail!( + "Execution program `{}` came from issue-batch intake and must be evaluated without a Decision Contract.", + self.program_id + ); + } + + let current_fingerprint = decision_contract_fingerprint(current_contract)?; + + self.evaluate_with_authority(Some(current_contract), ¤t_fingerprint, policy, context) + } + + /// Evaluate every node for an issue-batch intake program. + pub(crate) fn evaluate_issue_batch( + &self, + policy: &ExecutionWorkflowPolicy, + context: &ExecutionProgramReadinessContext, + ) -> Result { + if self.source_contract_id.as_deref().is_some() { + eyre::bail!( + "Execution program `{}` came from goal intake and must be evaluated with a Decision Contract.", + self.program_id + ); + } + + self.evaluate_with_authority(None, &self.accepted_contract_fingerprint, policy, context) + } + + fn evaluate_with_authority( + &self, + current_contract: Option<&DecisionContract>, + current_fingerprint: &str, + policy: &ExecutionWorkflowPolicy, + context: &ExecutionProgramReadinessContext, ) -> Result { self.validate()?; @@ -838,7 +959,6 @@ impl ExecutionProgram { ); } - 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(); @@ -850,7 +970,7 @@ impl ExecutionProgram { program: self, node, current_contract, - current_fingerprint: ¤t_fingerprint, + current_fingerprint, policy, node_lookup: &node_lookup, dependency_lookup: &dependency_lookup, @@ -866,7 +986,10 @@ impl ExecutionProgram { 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_optional( + "execution program source_contract_id", + self.source_contract_id.as_deref(), + )?; validate_required( "execution program accepted_contract_fingerprint", &self.accepted_contract_fingerprint, @@ -900,16 +1023,32 @@ impl ExecutionProgram { } if let Some(source_contract_id) = plan.source_contract_id() - && source_contract_id != self.source_contract_id + && Some(source_contract_id) != self.source_contract_id.as_deref() { eyre::bail!( "Execution program `{}` belongs to source contract `{}` but intake plan belongs to `{}`.", self.program_id, - self.source_contract_id, + self.source_contract_id.as_deref().unwrap_or("none"), source_contract_id ); } + if plan.intake_kind == ProgramIntakeKind::GoalIntake + && self.source_contract_id.as_deref().is_none_or(str::is_empty) + { + eyre::bail!( + "Goal intake execution program `{}` must reference a source contract.", + self.program_id + ); + } + if plan.intake_kind == ProgramIntakeKind::IssueBatchIntake + && self.source_contract_id.as_deref().is_some_and(|id| !id.is_empty()) + { + eyre::bail!( + "Issue-batch execution program `{}` must not reference a source contract.", + self.program_id + ); + } if plan.accepted_contract_fingerprint != self.accepted_contract_fingerprint { eyre::bail!( "Execution program `{}` has an intake plan fingerprint mismatch.", @@ -931,16 +1070,12 @@ impl ExecutionProgram { ); } } - 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 - ); - } - } + + if self.program_intake_plan.is_none() && self.source_contract_id.as_deref().is_none() { + eyre::bail!( + "Execution program `{}` without a source contract must carry an issue-batch intake plan.", + self.program_id + ); } Ok(()) @@ -1298,7 +1433,7 @@ pub(crate) struct ExecutionProgramOperatorSummary { struct EvaluateNodeInput<'a> { program: &'a ExecutionProgram, node: &'a ExecutionProgramNode, - current_contract: &'a DecisionContract, + current_contract: Option<&'a DecisionContract>, current_fingerprint: &'a str, policy: &'a ExecutionWorkflowPolicy, node_lookup: &'a BTreeMap<&'a str, &'a ExecutionProgramNode>, @@ -1317,22 +1452,29 @@ fn evaluate_node(input: EvaluateNodeInput<'_>) -> Result bool { + !description_is_machine_only_fenced_block(&issue.description) +} + fn issue_passes_dispatch_policy( tracker: &T, issue: &TrackerIssue, @@ -48,10 +52,6 @@ where Ok(true) } -fn issue_has_generic_dispatch_briefing(issue: &TrackerIssue) -> bool { - !description_is_machine_only_fenced_block(&issue.description) -} - fn description_is_machine_only_fenced_block(description: &str) -> bool { let trimmed = description.trim(); diff --git a/apps/decodex/src/orchestrator/harness_improvement.rs b/apps/decodex/src/orchestrator/harness_improvement.rs index 641f33678..b19a9585b 100644 --- a/apps/decodex/src/orchestrator/harness_improvement.rs +++ b/apps/decodex/src/orchestrator/harness_improvement.rs @@ -117,7 +117,7 @@ struct HarnessOutcomeContract { #[derive(Clone, Debug, Eq, PartialEq, Serialize)] struct HarnessOutcomeProgram { program_id: String, - source_contract_id: String, + source_contract_id: Option, node_count: usize, nodes: Vec, } @@ -497,7 +497,7 @@ fn harness_outcome_program(record: &ExecutionProgramRecord) -> HarnessOutcomePro HarnessOutcomeProgram { program_id: record.program_id().to_owned(), - source_contract_id: record.source_contract_id().to_owned(), + source_contract_id: record.source_contract_id().map(str::to_owned), node_count: nodes.len(), nodes, } diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index 6c9bc04bb..b6087fe1b 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -673,14 +673,18 @@ fn operator_execution_program_statuses( 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)); + let evaluation = if let Some(source_contract_id) = record.source_contract_id() { + let Some(contract) = state_store.decision_contract(project.service_id(), source_contract_id)? + else { + statuses.push(OperatorExecutionProgramStatus::missing_contract(&record)); - continue; + continue; + }; + + record.program().evaluate(contract.contract(), &policy, &context)? + } else { + record.program().evaluate_issue_batch(&policy, &context)? }; - let evaluation = record.program().evaluate(contract.contract(), &policy, &context)?; statuses.push(OperatorExecutionProgramStatus::from_summary( &record, @@ -6513,7 +6517,7 @@ fn append_rendered_execution_programs(output: &mut String, snapshot: &OperatorSt output.push_str(&format!( "- 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.source_contract_id.as_deref().unwrap_or("none"), program.node_count, program.planned_count, program.mapped_count, diff --git a/apps/decodex/src/orchestrator/tests/operator/status/text.rs b/apps/decodex/src/orchestrator/tests/operator/status/text.rs index e43f1ca88..4c4b594bf 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/text.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/text.rs @@ -181,7 +181,7 @@ fn operator_status_text_surfaces_execution_program_summary() { history_lanes: Vec::new(), execution_programs: vec![OperatorExecutionProgramStatus { program_id: String::from("program-853"), - source_contract_id: String::from("contract-852"), + source_contract_id: Some(String::from("contract-852")), node_count: 3, planned_count: 0, mapped_count: 0, diff --git a/apps/decodex/src/orchestrator/types.rs b/apps/decodex/src/orchestrator/types.rs index c3a806c94..21c6d90b1 100644 --- a/apps/decodex/src/orchestrator/types.rs +++ b/apps/decodex/src/orchestrator/types.rs @@ -1291,7 +1291,7 @@ struct OperatorProjectStatus { #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct OperatorExecutionProgramStatus { program_id: String, - source_contract_id: String, + source_contract_id: Option, node_count: usize, planned_count: usize, mapped_count: usize, @@ -1315,7 +1315,7 @@ impl OperatorExecutionProgramStatus { ) -> Self { Self { program_id: summary.program_id, - source_contract_id: record.source_contract_id().to_owned(), + source_contract_id: record.source_contract_id().map(str::to_owned), node_count: record.program().nodes().len(), planned_count: summary.planned_count, mapped_count: summary.mapped_count, @@ -1339,7 +1339,7 @@ impl OperatorExecutionProgramStatus { Self { program_id: record.program_id().to_owned(), - source_contract_id: record.source_contract_id().to_owned(), + source_contract_id: record.source_contract_id().map(str::to_owned), node_count, planned_count: 0, mapped_count: 0, diff --git a/apps/decodex/src/program_intake.rs b/apps/decodex/src/program_intake.rs new file mode 100644 index 000000000..25da101c0 --- /dev/null +++ b/apps/decodex/src/program_intake.rs @@ -0,0 +1,968 @@ +//! Operator issue-batch intake for internal Execution Programs. + +use std::{ + collections::{BTreeMap, BTreeSet}, + env, + path::{Path, PathBuf}, +}; + +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::{ + config::ServiceConfig, + execution_program::{ + ExecutionConflictDomain, ExecutionConflictDomainKind, ExecutionDependencySnapshot, + ExecutionLinearIssueMapping, ExecutionNodeEvaluation, ExecutionProgram, + ExecutionProgramDependency, ExecutionProgramNode, ExecutionProgramNodeLifecycleState, + ExecutionProgramNodeStage, ExecutionProgramReadinessContext, ExecutionQueueIntent, + ExecutionQueueLabelAction, ExecutionWorkflowPolicy, + }, + orchestrator, + prelude::{Result, eyre}, + runtime, + state::StateStore, + tracker::{self, IssueTracker, TrackerIssue, linear::LinearClient}, + workflow::WorkflowDocument, +}; + +/// CLI/runtime request for issue-batch Program Intake. +pub(crate) struct IssueBatchIntakeCommandRequest<'a> { + pub(crate) config_path: Option<&'a Path>, + pub(crate) project_id: Option<&'a str>, + pub(crate) issue_identifiers: Vec, + pub(crate) dry_run: bool, + pub(crate) persist: bool, +} + +/// Deterministic report for one issue-batch Program Intake run. +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +pub(crate) struct IssueBatchIntakeReport { + /// Registered service id that owns this intake. + pub(crate) service_id: String, + /// Internal program id derived from the accepted issue batch. + pub(crate) program_id: String, + /// Whether this run was explicitly dry-run only. + pub(crate) dry_run: bool, + /// Whether local runtime state was persisted. + pub(crate) persisted: bool, + /// Service-scoped queue label that would be used by later reconciliation. + pub(crate) queue_label: String, + /// Deterministic classification counts. + pub(crate) counts: IssueBatchIntakeCounts, + /// Per-issue classification rows. + pub(crate) issues: Vec, +} + +/// Count summary for an issue-batch intake report. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)] +pub(crate) struct IssueBatchIntakeCounts { + /// Issues ready for queue intent. + pub(crate) ready: usize, + /// Issues intentionally held from queueing. + pub(crate) held: usize, + /// Issues blocked by dependencies, attention, or briefing. + pub(crate) blocked: usize, + /// Issues that are stale or terminal for the accepted batch. + pub(crate) stale: usize, + /// Supplied identifiers that did not map to Linear issues. + pub(crate) unmapped: usize, +} + +/// Per-issue report row for issue-batch intake. +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +pub(crate) struct IssueBatchIntakeIssueReport { + /// Linear issue identifier supplied by the operator. + pub(crate) issue_identifier: String, + /// Linear issue id, when the identifier resolved. + pub(crate) issue_id: Option, + /// Current Linear workflow state, when the identifier resolved. + pub(crate) issue_state: Option, + /// Normalized intake classification. + pub(crate) classification: IssueBatchIntakeClassification, + /// Queue intent stored on the internal program node, when available. + pub(crate) queue_intent: Option, + /// Readiness-derived queue-label action for later reconciliation. + pub(crate) queue_label_action: Option, + /// Deterministic local readback reasons. + pub(crate) reasons: Vec, + /// Known blocker issue identifiers. + pub(crate) blockers: Vec, + /// Coarse conflict-domain hints. + pub(crate) conflict_domains: Vec, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct IssueFacts { + has_queue_label: bool, + queue_label_program_owned: bool, + has_active_label: bool, + has_opt_out_label: bool, + has_needs_attention_label: bool, + has_generic_dispatch_briefing: bool, + has_open_blockers: bool, + has_human_owned_queue_label: bool, +} + +/// Normalized issue-batch intake classification. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum IssueBatchIntakeClassification { + /// Ready for later queue-label reconciliation. + Ready, + /// Intentionally held from queueing. + Held, + /// Blocked by issue state, dependency, attention, or briefing evidence. + Blocked, + /// Terminal or stale relative to the accepted intake boundary. + Stale, + /// Supplied identifier did not map to a tracker issue. + Unmapped, +} +impl IssueBatchIntakeClassification { + fn as_str(self) -> &'static str { + match self { + Self::Ready => "ready", + Self::Held => "held", + Self::Blocked => "blocked", + Self::Stale => "stale", + Self::Unmapped => "unmapped", + } + } +} + +/// Run issue-batch intake through the configured Linear tracker. +pub(crate) fn run_issue_batch_intake_command( + request: IssueBatchIntakeCommandRequest<'_>, +) -> Result { + if request.dry_run == request.persist { + eyre::bail!("Issue-batch intake requires exactly one of --dry-run or --persist."); + } + + let state_store = runtime::open_runtime_store()?; + let config_path = + resolve_intake_project_config_path(request.config_path, request.project_id, &state_store)?; + let config = ServiceConfig::from_path(&config_path)?; + let workflow = WorkflowDocument::from_path(config.workflow_path())?; + + runtime::register_project_config(&state_store, &config_path, true)?; + + let tracker = LinearClient::new(config.tracker().resolve_api_key()?)?; + + run_issue_batch_intake( + &state_store, + &tracker, + &config, + &workflow, + request.issue_identifiers, + request.dry_run, + request.persist, + ) +} + +/// Build and optionally persist a non-mutating issue-batch intake report. +pub(crate) fn run_issue_batch_intake( + state_store: &StateStore, + tracker: &T, + config: &ServiceConfig, + workflow: &WorkflowDocument, + issue_identifiers: Vec, + dry_run: bool, + persist: bool, +) -> Result +where + T: IssueTracker + ?Sized, +{ + if dry_run == persist { + eyre::bail!("Issue-batch intake requires exactly one of dry_run or persist."); + } + + let issue_identifiers = normalize_issue_identifiers(issue_identifiers)?; + let queue_label = tracker::automation_queue_label(config.service_id()); + let active_label = tracker::automation_active_label(config.service_id()); + let policy = ExecutionWorkflowPolicy::from_workflow(config.service_id(), workflow)?; + let mut resolved = BTreeMap::new(); + let mut missing = Vec::new(); + + for identifier in &issue_identifiers { + match tracker.get_issue_by_identifier(identifier)? { + Some(issue) => { + resolved.insert(identifier.clone(), issue); + }, + None => missing.push(identifier.clone()), + } + } + + let batch_fingerprint = + issue_batch_fingerprint(config.service_id(), &issue_identifiers, &resolved); + let program_id = issue_batch_program_id(config.service_id(), &batch_fingerprint); + let supplied_node_ids = issue_identifiers + .iter() + .map(|identifier| (identifier.clone(), node_id_for_issue(identifier))) + .collect::>(); + let mut nodes = Vec::new(); + let mut dependency_snapshots = Vec::new(); + let mut facts_by_identifier = BTreeMap::new(); + + for identifier in &issue_identifiers { + if let Some(issue) = resolved.get(identifier) { + let facts = issue_facts( + tracker, + state_store, + config.service_id(), + workflow, + issue, + &queue_label, + &active_label, + )?; + + dependency_snapshots.extend(dependency_snapshots_for(issue, &supplied_node_ids)?); + nodes.push(issue_node(issue, &facts, workflow, &supplied_node_ids)?); + facts_by_identifier.insert(identifier.clone(), facts); + } else { + nodes.push(unmapped_node(identifier)?); + } + } + + let program = ExecutionProgram::from_issue_batch_intake( + &program_id, + config.service_id(), + &batch_fingerprint, + format!("Issue-batch intake for {} issue(s).", issue_identifiers.len()), + nodes, + )?; + let context = + ExecutionProgramReadinessContext::new().with_dependency_snapshots(dependency_snapshots); + let evaluation = program.evaluate_issue_batch(&policy, &context)?; + let evaluation_by_issue = evaluation + .nodes() + .iter() + .filter_map(|node| { + let issue = node.linear_issue()?; + + Some((issue.issue_identifier().to_owned(), node)) + }) + .collect::>(); + let mut rows = Vec::new(); + + for identifier in &issue_identifiers { + if missing.iter().any(|missing| missing == identifier) { + rows.push(unmapped_report_row(identifier)); + + continue; + } + + let issue = resolved + .get(identifier) + .ok_or_else(|| eyre::eyre!("Resolved issue `{identifier}` disappeared from intake."))?; + let facts = facts_by_identifier + .get(identifier) + .ok_or_else(|| eyre::eyre!("Issue facts for `{identifier}` disappeared."))?; + let evaluation = evaluation_by_issue + .get(identifier) + .ok_or_else(|| eyre::eyre!("Issue evaluation for `{identifier}` disappeared."))?; + + rows.push(issue_report_row(issue, facts, evaluation, workflow)); + } + + let counts = classify_counts(&rows); + + if persist { + state_store.upsert_execution_program(config.service_id(), program)?; + } + + Ok(IssueBatchIntakeReport { + service_id: config.service_id().to_owned(), + program_id, + dry_run, + persisted: persist, + queue_label, + counts, + issues: rows, + }) +} + +/// Render a compact human-readable intake report. +pub(crate) fn render_issue_batch_intake_report(report: &IssueBatchIntakeReport) -> String { + let mode = if report.persisted { "persist" } else { "dry-run" }; + let mut output = format!( + "program intake {mode}: service={} program={} queue_label={} ready={} held={} blocked={} stale={} unmapped={}\n", + report.service_id, + report.program_id, + report.queue_label, + report.counts.ready, + report.counts.held, + report.counts.blocked, + report.counts.stale, + report.counts.unmapped, + ); + + for row in &report.issues { + let state = row.issue_state.as_deref().unwrap_or("unmapped"); + let action = row.queue_label_action.as_deref().unwrap_or("none"); + let reasons = + if row.reasons.is_empty() { String::from("none") } else { row.reasons.join("; ") }; + + output.push_str(&format!( + "- {} classification={} state={} queue_action={} reasons={}\n", + row.issue_identifier, + row.classification.as_str(), + state, + action, + reasons + )); + } + + output +} + +fn resolve_intake_project_config_path( + config_path: Option<&Path>, + project_id: Option<&str>, + state_store: &StateStore, +) -> Result { + if let Some(config_path) = config_path { + return ServiceConfig::resolve_project_config_path(config_path); + } + if let Some(project_id) = project_id { + let Some(project) = state_store + .list_projects()? + .into_iter() + .find(|project| project.service_id() == project_id) + else { + eyre::bail!( + "Decodex project `{project_id}` is not registered; pass --config ." + ); + }; + + return Ok(project.config_path().to_path_buf()); + } + + runtime::registered_config_path_for_cwd(state_store, &env::current_dir()?)?.ok_or_else(|| { + eyre::eyre!( + "Current directory is not registered to a Decodex project; pass --config or --project ." + ) + }) +} + +fn normalize_issue_identifiers(issue_identifiers: Vec) -> Result> { + let mut normalized = issue_identifiers + .into_iter() + .map(|identifier| identifier.trim().to_owned()) + .filter(|identifier| !identifier.is_empty()) + .collect::>(); + + normalized.sort(); + normalized.dedup(); + + if normalized.is_empty() { + eyre::bail!("Issue-batch intake requires at least one issue identifier."); + } + + Ok(normalized) +} + +fn issue_batch_fingerprint( + service_id: &str, + issue_identifiers: &[String], + resolved: &BTreeMap, +) -> String { + let mut digest = Sha256::new(); + + digest.update(service_id.as_bytes()); + + for identifier in issue_identifiers { + digest.update(b"\0identifier:"); + digest.update(identifier.as_bytes()); + + if let Some(issue) = resolved.get(identifier) { + digest.update(b"\0issue:"); + digest.update(issue.id.as_bytes()); + digest.update(b"\0state:"); + digest.update(issue.state.name.as_bytes()); + digest.update(b"\0updated:"); + digest.update(issue.updated_at.as_bytes()); + } + } + + digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn issue_batch_program_id(service_id: &str, fingerprint: &str) -> String { + format!("issue-batch-{service_id}-{}", &fingerprint[..16]) +} + +fn node_id_for_issue(identifier: &str) -> String { + format!("issue:{identifier}") +} + +fn unmapped_node(identifier: &str) -> Result { + ExecutionProgramNode::new( + format!("unmapped:{identifier}"), + ExecutionProgramNodeStage::Runtime, + format!("Resolve supplied Linear issue identifier `{identifier}` before queueing."), + ExecutionQueueIntent::NotReady, + )? + .with_acceptance_expectations([format!( + "`{identifier}` maps to a normal Linear issue before execution." + )])? + .with_validation_expectations([String::from("Tracker lookup succeeds before queue intent.")]) +} + +fn issue_facts( + tracker: &T, + state_store: &StateStore, + service_id: &str, + workflow: &WorkflowDocument, + issue: &TrackerIssue, + queue_label: &str, + active_label: &str, +) -> Result +where + T: IssueTracker + ?Sized, +{ + let tracker_policy = workflow.frontmatter().tracker(); + let has_queue_label = + tracker::issue_has_label_with_server_confirmation(tracker, issue, queue_label)?; + let queue_label_program_owned = has_queue_label + && !state_store + .program_queue_label_ownership_for_issue(service_id, &issue.id, queue_label)? + .is_empty(); + let has_active_label = + tracker::issue_has_label_with_server_confirmation(tracker, issue, active_label)?; + let has_opt_out_label = tracker::issue_has_label_with_server_confirmation( + tracker, + issue, + tracker_policy.opt_out_label(), + )?; + let has_needs_attention_label = tracker::issue_has_label_with_server_confirmation( + tracker, + issue, + tracker_policy.needs_attention_label(), + )?; + let has_open_blockers = + issue.blockers.iter().any(|blocker| !state_name_is_terminal(&blocker.state.name, workflow)); + let has_human_owned_queue_label = has_queue_label && !queue_label_program_owned; + + Ok(IssueFacts { + has_queue_label, + queue_label_program_owned, + has_active_label, + has_opt_out_label, + has_needs_attention_label, + has_generic_dispatch_briefing: issue_has_generic_dispatch_briefing(issue), + has_open_blockers, + has_human_owned_queue_label, + }) +} + +fn issue_node( + issue: &TrackerIssue, + facts: &IssueFacts, + workflow: &WorkflowDocument, + supplied_node_ids: &BTreeMap, +) -> Result { + let queue_intent = issue_queue_intent(issue, facts, workflow); + let mut mapping = + ExecutionLinearIssueMapping::new(&issue.id, &issue.identifier, &issue.state.name)?; + + mapping = if facts.queue_label_program_owned { + mapping.with_program_owned_queue_label(true) + } else { + mapping.with_queue_label(facts.has_queue_label) + }; + mapping = mapping + .with_active_label(facts.has_active_label) + .with_opt_out_label(facts.has_opt_out_label) + .with_needs_attention_label(facts.has_needs_attention_label) + .with_generic_dispatch_briefing(facts.has_generic_dispatch_briefing); + + ExecutionProgramNode::new( + node_id_for_issue(&issue.identifier), + ExecutionProgramNodeStage::Runtime, + issue.title.clone(), + queue_intent, + )? + .with_objective_lineage([format!("Issue-batch intake supplied `{}`.", issue.identifier)])? + .with_dependencies(issue_dependencies(issue, supplied_node_ids)?)? + .with_conflict_domains(issue_conflict_domains(issue)?)? + .with_acceptance_expectations([format!( + "`{}` remains a normal Linear issue with an executable brief.", + issue.identifier + )])? + .with_validation_expectations([String::from( + "Run the issue-specific repository validation before review handoff.", + )])? + .with_linear_issue(mapping) +} + +fn issue_queue_intent( + issue: &TrackerIssue, + facts: &IssueFacts, + workflow: &WorkflowDocument, +) -> ExecutionQueueIntent { + if state_name_is_terminal(&issue.state.name, workflow) { + return ExecutionQueueIntent::Done; + } + if facts.has_active_label { + return ExecutionQueueIntent::Active; + } + if facts.has_opt_out_label || facts.has_human_owned_queue_label { + return ExecutionQueueIntent::NotReady; + } + if !workflow + .frontmatter() + .tracker() + .startable_states() + .iter() + .any(|state| state == &issue.state.name) + { + return ExecutionQueueIntent::NotReady; + } + + ExecutionQueueIntent::ReadyToQueue +} + +fn issue_dependencies( + issue: &TrackerIssue, + supplied_node_ids: &BTreeMap, +) -> Result> { + issue + .blockers + .iter() + .map(|blocker| { + let dependency_id = supplied_node_ids + .get(&blocker.identifier) + .cloned() + .unwrap_or_else(|| blocker.identifier.clone()); + + ExecutionProgramDependency::new(dependency_id) + }) + .collect() +} + +fn dependency_snapshots_for( + issue: &TrackerIssue, + supplied_node_ids: &BTreeMap, +) -> Result> { + issue + .blockers + .iter() + .map(|blocker| { + let dependency_id = supplied_node_ids + .get(&blocker.identifier) + .cloned() + .unwrap_or_else(|| blocker.identifier.clone()); + + ExecutionDependencySnapshot::tracker_state(dependency_id, blocker.state.name.clone()) + }) + .collect() +} + +fn issue_conflict_domains(issue: &TrackerIssue) -> Result> { + let mut domains = vec![ExecutionConflictDomain::new( + ExecutionConflictDomainKind::TrackerOwnership, + issue.identifier.clone(), + )?]; + let mut seen = BTreeSet::from([format!( + "{}:{}", + ExecutionConflictDomainKind::TrackerOwnership.as_str(), + issue.identifier + )]); + + for label in &issue.labels { + if let Some(module) = label.name.strip_prefix("repo:") + && !module.trim().is_empty() + { + let key = module.trim().to_owned(); + let seen_key = format!("{}:{key}", ExecutionConflictDomainKind::Module.as_str()); + + if seen.insert(seen_key) { + domains + .push(ExecutionConflictDomain::new(ExecutionConflictDomainKind::Module, key)?); + } + } + } + + Ok(domains) +} + +fn issue_report_row( + issue: &TrackerIssue, + facts: &IssueFacts, + evaluation: &ExecutionNodeEvaluation, + workflow: &WorkflowDocument, +) -> IssueBatchIntakeIssueReport { + let classification = classify_issue(issue, facts, evaluation, workflow); + let mut reasons = evaluation.reasons().to_vec(); + + if facts.has_human_owned_queue_label { + reasons.push(String::from("service queue label is present without program-owned evidence")); + } + + reasons.sort(); + reasons.dedup(); + IssueBatchIntakeIssueReport { + issue_identifier: issue.identifier.clone(), + issue_id: Some(issue.id.clone()), + issue_state: Some(issue.state.name.clone()), + classification, + queue_intent: Some( + evaluation + .linear_issue() + .map_or(ExecutionQueueIntent::NotReady, |_| { + issue_queue_intent(issue, facts, workflow) + }) + .as_str() + .to_owned(), + ), + queue_label_action: evaluation.queue_label_action().map(queue_label_action_name), + reasons, + blockers: issue.blockers.iter().map(|blocker| blocker.identifier.clone()).collect(), + conflict_domains: issue_conflict_domains(issue) + .unwrap_or_default() + .into_iter() + .map(|domain| format!("{}:{}", domain.kind().as_str(), domain.key())) + .collect(), + } +} + +fn classify_issue( + issue: &TrackerIssue, + facts: &IssueFacts, + evaluation: &ExecutionNodeEvaluation, + workflow: &WorkflowDocument, +) -> IssueBatchIntakeClassification { + if state_name_is_terminal(&issue.state.name, workflow) { + return IssueBatchIntakeClassification::Stale; + } + if facts.has_active_label || facts.has_opt_out_label || facts.has_human_owned_queue_label { + return IssueBatchIntakeClassification::Held; + } + if facts.has_needs_attention_label + || facts.has_open_blockers + || !facts.has_generic_dispatch_briefing + { + return IssueBatchIntakeClassification::Blocked; + } + + match evaluation.lifecycle_state() { + ExecutionProgramNodeLifecycleState::Ready | ExecutionProgramNodeLifecycleState::Queued => + IssueBatchIntakeClassification::Ready, + ExecutionProgramNodeLifecycleState::Planned + | ExecutionProgramNodeLifecycleState::Mapped + | ExecutionProgramNodeLifecycleState::Active => IssueBatchIntakeClassification::Held, + ExecutionProgramNodeLifecycleState::Blocked + | ExecutionProgramNodeLifecycleState::NeedsAttention => IssueBatchIntakeClassification::Blocked, + ExecutionProgramNodeLifecycleState::Completed + | ExecutionProgramNodeLifecycleState::Stale + | ExecutionProgramNodeLifecycleState::Superseded => IssueBatchIntakeClassification::Stale, + } +} + +fn queue_label_action_name(action: ExecutionQueueLabelAction) -> String { + match action { + ExecutionQueueLabelAction::Apply => "apply", + ExecutionQueueLabelAction::Retain => "retain", + ExecutionQueueLabelAction::Remove => "remove", + } + .to_owned() +} + +fn unmapped_report_row(identifier: &str) -> IssueBatchIntakeIssueReport { + IssueBatchIntakeIssueReport { + issue_identifier: identifier.to_owned(), + issue_id: None, + issue_state: None, + classification: IssueBatchIntakeClassification::Unmapped, + queue_intent: None, + queue_label_action: None, + reasons: vec![String::from("tracker issue identifier did not resolve")], + blockers: Vec::new(), + conflict_domains: Vec::new(), + } +} + +fn classify_counts(rows: &[IssueBatchIntakeIssueReport]) -> IssueBatchIntakeCounts { + let mut counts = IssueBatchIntakeCounts::default(); + + for row in rows { + match row.classification { + IssueBatchIntakeClassification::Ready => counts.ready += 1, + IssueBatchIntakeClassification::Held => counts.held += 1, + IssueBatchIntakeClassification::Blocked => counts.blocked += 1, + IssueBatchIntakeClassification::Stale => counts.stale += 1, + IssueBatchIntakeClassification::Unmapped => counts.unmapped += 1, + } + } + + counts +} + +fn state_name_is_terminal(state_name: &str, workflow: &WorkflowDocument) -> bool { + workflow.frontmatter().tracker().terminal_states().iter().any(|state| state == state_name) +} + +fn issue_has_generic_dispatch_briefing(issue: &TrackerIssue) -> bool { + orchestrator::issue_has_generic_dispatch_briefing(issue) +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use crate::{ + program_intake::{self, IssueBatchIntakeClassification}, + state::StateStore, + tracker::{ + IssueTracker, TrackerComment, TrackerIssue, TrackerIssueBlocker, TrackerState, + TrackerTeam, + }, + workflow::WorkflowDocument, + }; + + trait TestIssueExt { + fn with_blocker(self, identifier: &str, state: &str) -> Self; + } + + #[derive(Default)] + struct FakeTracker { + issues: HashMap, + } + + impl FakeTracker { + fn with_issues(mut self, issues: impl IntoIterator) -> Self { + for issue in issues { + self.issues.insert(issue.identifier.clone(), issue); + } + + self + } + } + + impl IssueTracker for FakeTracker { + fn list_issues_with_label( + &self, + label_name: &str, + ) -> crate::prelude::Result> { + Ok(self.issues.values().filter(|issue| issue.has_label(label_name)).cloned().collect()) + } + + fn find_team_label_id( + &self, + _team_id: &str, + label_name: &str, + ) -> crate::prelude::Result> { + Ok(Some(format!("label-{label_name}"))) + } + + fn get_issue_by_identifier( + &self, + issue_identifier: &str, + ) -> crate::prelude::Result> { + Ok(self.issues.get(issue_identifier).cloned()) + } + + fn refresh_issues( + &self, + _issue_ids: &[String], + ) -> crate::prelude::Result> { + Ok(Vec::new()) + } + + fn list_comments(&self, _issue_id: &str) -> crate::prelude::Result> { + Ok(Vec::new()) + } + + fn update_issue_state( + &self, + _issue_id: &str, + _state_id: &str, + ) -> crate::prelude::Result<()> { + Ok(()) + } + + fn add_issue_labels( + &self, + _issue_id: &str, + _label_ids: &[String], + ) -> crate::prelude::Result<()> { + Ok(()) + } + + fn remove_issue_labels( + &self, + _issue_id: &str, + _label_ids: &[String], + ) -> crate::prelude::Result<()> { + Ok(()) + } + + fn create_comment(&self, _issue_id: &str, _body: &str) -> crate::prelude::Result<()> { + Ok(()) + } + } + + impl TestIssueExt for TrackerIssue { + fn with_blocker(mut self, identifier: &str, state: &str) -> Self { + self.blockers.push(TrackerIssueBlocker { + id: format!("id-{identifier}"), + identifier: identifier.to_owned(), + state: TrackerState { id: format!("state-{state}"), name: state.to_owned() }, + }); + + self + } + } + + #[test] + fn issue_batch_dry_run_classifies_without_persisting() { + let store = StateStore::open_in_memory().expect("store should open"); + let workflow = workflow(); + let config = test_config(); + let tracker = FakeTracker::default().with_issues([ + issue("XY-1", "Todo"), + issue("XY-2", "In Progress"), + issue("XY-3", "Done"), + issue("XY-4", "Todo").with_blocker("XY-10", "Todo"), + ]); + let report = program_intake::run_issue_batch_intake( + &store, + &tracker, + &config, + &workflow, + vec![ + String::from("XY-4"), + String::from("XY-2"), + String::from("XY-404"), + String::from("XY-1"), + String::from("XY-3"), + ], + true, + false, + ) + .expect("dry-run should classify"); + + assert_eq!(report.counts.ready, 1); + assert_eq!(report.counts.held, 1); + assert_eq!(report.counts.blocked, 1); + assert_eq!(report.counts.stale, 1); + assert_eq!(report.counts.unmapped, 1); + assert_eq!(report.issues[0].issue_identifier, "XY-1"); + assert_eq!(report.issues[0].classification, IssueBatchIntakeClassification::Ready); + assert!( + store.list_execution_programs("decodex").expect("program list should read").is_empty() + ); + } + + #[test] + fn issue_batch_persist_writes_program_and_adjacent_intake_state() { + let store = StateStore::open_in_memory().expect("store should open"); + let workflow = workflow(); + let config = test_config(); + let tracker = FakeTracker::default().with_issues([issue("XY-1", "Todo")]); + let report = program_intake::run_issue_batch_intake( + &store, + &tracker, + &config, + &workflow, + vec![String::from("XY-1")], + false, + true, + ) + .expect("persist should write local state"); + + assert!(report.persisted); + assert_eq!(store.list_execution_programs("decodex").expect("programs").len(), 1); + assert_eq!(store.list_program_intake_plans("decodex").expect("plans").len(), 1); + assert_eq!( + store + .list_program_issue_mappings("decodex", &report.program_id) + .expect("mappings") + .len(), + 1 + ); + assert_eq!( + store.list_program_intake_plans("decodex").expect("plans")[0].intake_kind(), + "issue_batch_intake" + ); + } + + fn workflow() -> WorkflowDocument { + WorkflowDocument::parse_markdown( + r#"+++ +version = 1 +[tracker] +provider = "linear" +startable_states = ["Todo"] +terminal_states = ["Done", "Canceled", "Duplicate"] +in_progress_state = "In Progress" +success_state = "In Review" +completed_state = "Done" +failure_state = "Todo" +opt_out_label = "decodex:manual-only" +needs_attention_label = "decodex:needs-attention" +[agent] +transport = "stdio://" +[execution] +max_attempts = 3 +max_turns = 3 +max_retry_backoff_ms = 300000 +max_concurrent_agents = 0 +gate_profiles = {} +canonicalize_commands = ["cargo make fmt"] +verify_commands = ["cargo make test"] +[execution.workspace_hooks] +after_create_commands = [] +before_remove_commands = [] +timeout_seconds = 60 +[context] +read_first = [] ++++ +"#, + ) + .expect("workflow should parse") + } + + fn test_config() -> crate::config::ServiceConfig { + crate::config::ServiceConfig::parse_toml( + r#" +service_id = "decodex" +[tracker] +api_key_env_var = "HOME" +[github] +token_env_var = "HOME" +[codex] +review = "standard" +[paths] +repo_root = "." +worktree_root = ".worktrees" +"#, + ) + .expect("config should parse") + } + + fn issue(identifier: &str, state: &str) -> TrackerIssue { + TrackerIssue { + id: format!("id-{identifier}"), + identifier: identifier.to_owned(), + project_slug: None, + title: format!("Issue {identifier}"), + author: None, + description: format!("Implement {identifier}."), + priority: None, + created_at: String::from("2026-06-01T00:00:00Z"), + updated_at: String::from("2026-06-01T00:00:00Z"), + state: TrackerState { id: format!("state-{state}"), name: state.to_owned() }, + team: TrackerTeam { + id: String::from("team"), + name: String::from("Team"), + states: Vec::new(), + labels: Vec::new(), + }, + labels_complete: true, + labels: Vec::new(), + blockers: Vec::new(), + } + } +} diff --git a/apps/decodex/src/state/internal.rs b/apps/decodex/src/state/internal.rs index 95bc52c20..256c4e639 100644 --- a/apps/decodex/src/state/internal.rs +++ b/apps/decodex/src/state/internal.rs @@ -16,6 +16,8 @@ use libc::{ use process::Command; use rusqlite::{self, Row}; +use crate::tracker; + static RUN_ACTIVITY_MARKER_WRITE_SEQUENCE: AtomicU64 = AtomicU64::new(0); pub(crate) struct EffectiveRuntimeMarker<'a> { @@ -130,6 +132,10 @@ struct StateData { private_execution_events: Vec, decision_contracts: HashMap, execution_programs: HashMap, + program_intake_plans: HashMap, + program_issue_mappings: HashMap, + program_queue_label_ownership: + HashMap, review_handoffs: HashMap, review_orchestrations: HashMap, review_policy_checkpoints: HashMap, @@ -152,6 +158,9 @@ impl StateData { self.private_execution_events = loaded.private_execution_events; self.decision_contracts = loaded.decision_contracts; self.execution_programs = loaded.execution_programs; + self.program_intake_plans = loaded.program_intake_plans; + self.program_issue_mappings = loaded.program_issue_mappings; + self.program_queue_label_ownership = loaded.program_queue_label_ownership; self.review_handoffs = loaded.review_handoffs; self.review_orchestrations = loaded.review_orchestrations; self.review_policy_checkpoints = loaded.review_policy_checkpoints; @@ -374,6 +383,7 @@ ON linear_execution_events (service_id, issue_id, event_unix, recorded_at_unix); self.bootstrap_private_execution_events_schema()?; self.bootstrap_decision_contracts_schema()?; self.bootstrap_execution_programs_schema()?; + self.bootstrap_program_intake_state_schema()?; self.bootstrap_loop_guardrail_schema()?; self.record_schema_version()?; @@ -598,7 +608,7 @@ ON decision_contracts (project_id, status, updated_at_unix); CREATE TABLE IF NOT EXISTS execution_programs ( project_id TEXT NOT NULL, program_id TEXT NOT NULL, - source_contract_id TEXT NOT NULL, + source_contract_id TEXT, payload_json TEXT NOT NULL, created_at TEXT NOT NULL, created_at_unix INTEGER NOT NULL, @@ -610,6 +620,145 @@ CREATE INDEX IF NOT EXISTS execution_programs_source_contract_idx ON execution_programs (project_id, source_contract_id, updated_at_unix); "#, )?; + self.ensure_execution_program_source_contract_nullable()?; + + Ok(()) + } + + fn ensure_execution_program_source_contract_nullable(&self) -> Result<()> { + let mut statement = self.connection.prepare("PRAGMA table_info(execution_programs)")?; + let columns = statement.query_map([], |row| { + Ok((row.get::<_, String>(1)?, row.get::<_, i64>(3)?)) + })?; + let mut source_contract_not_null = false; + + for column in columns { + let (name, not_null) = column?; + + if name == "source_contract_id" { + source_contract_not_null = not_null != 0; + + break; + } + } + + if !source_contract_not_null { + return Ok(()); + } + + self.connection.execute_batch( + r#" +ALTER TABLE execution_programs RENAME TO execution_programs_legacy_contract_required; +CREATE TABLE execution_programs ( + project_id TEXT NOT NULL, + program_id TEXT NOT NULL, + source_contract_id TEXT, + 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) +); +INSERT INTO execution_programs ( + project_id, program_id, source_contract_id, payload_json, created_at, created_at_unix, + updated_at, updated_at_unix +) +SELECT project_id, program_id, source_contract_id, payload_json, created_at, created_at_unix, + updated_at, updated_at_unix +FROM execution_programs_legacy_contract_required; +DROP TABLE execution_programs_legacy_contract_required; +CREATE INDEX IF NOT EXISTS execution_programs_source_contract_idx +ON execution_programs (project_id, source_contract_id, updated_at_unix); +"#, + )?; + + Ok(()) + } + + fn bootstrap_program_intake_state_schema(&self) -> Result<()> { + self.connection.execute_batch( + r#" +CREATE TABLE IF NOT EXISTS program_intake_plans ( + project_id TEXT NOT NULL, + program_id TEXT NOT NULL, + plan_id TEXT NOT NULL, + intake_kind TEXT NOT NULL, + source_contract_id TEXT, + accepted_contract_fingerprint TEXT NOT NULL, + public_summary 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, plan_id) +); +CREATE INDEX IF NOT EXISTS program_intake_plans_project_idx +ON program_intake_plans (project_id, intake_kind, updated_at_unix); +CREATE TABLE IF NOT EXISTS program_issue_mappings ( + project_id TEXT NOT NULL, + program_id TEXT NOT NULL, + node_id TEXT NOT NULL, + issue_id TEXT NOT NULL, + issue_identifier TEXT NOT NULL, + issue_state TEXT NOT NULL, + queue_intent TEXT NOT NULL, + has_queue_label INTEGER NOT NULL, + queue_label_owned_by_program_reconciler INTEGER NOT NULL, + has_active_label INTEGER NOT NULL, + has_opt_out_label INTEGER NOT NULL, + has_needs_attention_label INTEGER NOT NULL, + has_generic_dispatch_briefing INTEGER 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, node_id) +); +CREATE INDEX IF NOT EXISTS program_issue_mappings_issue_idx +ON program_issue_mappings (project_id, issue_id, updated_at_unix); +CREATE TABLE IF NOT EXISTS program_queue_label_ownership ( + project_id TEXT NOT NULL, + program_id TEXT NOT NULL, + node_id TEXT NOT NULL, + issue_id TEXT NOT NULL, + issue_identifier TEXT NOT NULL, + label_name TEXT NOT NULL, + service_id 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, node_id, label_name) +); +CREATE INDEX IF NOT EXISTS program_queue_label_ownership_issue_idx +ON program_queue_label_ownership (project_id, issue_id, label_name, updated_at_unix); +"#, + )?; + self.backfill_program_intake_state_from_execution_programs()?; + + Ok(()) + } + + fn backfill_program_intake_state_from_execution_programs(&self) -> 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([], execution_program_runtime_row_parts)?; + let mut records = Vec::new(); + + for row in rows { + records.push(execution_program_record_from_row_parts(row?)?); + } + + drop(statement); + + for record in records { + self.replace_program_intake_state(&record)?; + } Ok(()) } @@ -622,7 +771,7 @@ CREATE TABLE IF NOT EXISTS schema_meta ( value TEXT NOT NULL ); INSERT INTO schema_meta (key, value) -VALUES ('schema_version', '10') +VALUES ('schema_version', '11') ON CONFLICT(key) DO UPDATE SET value = excluded.value; "#, )?; @@ -643,6 +792,7 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; self.load_private_execution_events(&mut state)?; self.load_decision_contracts(&mut state)?; self.load_execution_programs(&mut state)?; + self.load_program_intake_state(&mut state)?; self.load_review_handoffs(&mut state)?; self.load_review_orchestrations(&mut state)?; self.load_review_policy_checkpoints(&mut state)?; @@ -693,6 +843,7 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; persist_private_execution_events(&transaction, state)?; persist_decision_contracts(&transaction, state)?; persist_execution_programs(&transaction, state)?; + persist_program_intake_state(&transaction, state)?; persist_review_handoffs(&transaction, state)?; persist_review_orchestrations(&transaction, state)?; persist_review_policy_checkpoints(&transaction, state)?; @@ -724,6 +875,18 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; "DELETE FROM execution_programs WHERE project_id = ?1", params![service_id], )?; + transaction.execute( + "DELETE FROM program_intake_plans WHERE project_id = ?1", + params![service_id], + )?; + transaction.execute( + "DELETE FROM program_issue_mappings WHERE project_id = ?1", + params![service_id], + )?; + transaction.execute( + "DELETE FROM program_queue_label_ownership WHERE project_id = ?1", + params![service_id], + )?; transaction.execute( "DELETE FROM loop_guardrail_checkpoints WHERE project_id = ?1", params![service_id], @@ -1002,7 +1165,7 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; params![ &record.project_id, record.program.program_id(), - &record.source_contract_id, + record.source_contract_id.as_deref(), payload_json, &record.created_at, record.created_at_unix, @@ -1010,10 +1173,28 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; record.updated_at_unix, ], )?; + self.replace_program_intake_state(record)?; Ok(()) } + fn replace_program_intake_state(&self, record: &ExecutionProgramRuntimeRecord) -> Result<()> { + self.connection.execute( + "DELETE FROM program_intake_plans WHERE project_id = ?1 AND program_id = ?2", + params![&record.project_id, record.program.program_id()], + )?; + self.connection.execute( + "DELETE FROM program_issue_mappings WHERE project_id = ?1 AND program_id = ?2", + params![&record.project_id, record.program.program_id()], + )?; + self.connection.execute( + "DELETE FROM program_queue_label_ownership WHERE project_id = ?1 AND program_id = ?2", + params![&record.project_id, record.program.program_id()], + )?; + + insert_program_intake_state(&self.connection, record) + } + fn delete_lease(&mut self, issue_id: &str) -> Result<()> { self.connection .execute("DELETE FROM leases WHERE issue_id = ?1", params![issue_id])?; @@ -1061,6 +1242,14 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; "UPDATE decision_contracts SET source_issue_id = ?2 WHERE source_issue_id = ?1", params![previous_issue_id, canonical_issue_id], )?; + transaction.execute( + "UPDATE program_issue_mappings SET issue_id = ?2 WHERE issue_id = ?1", + params![previous_issue_id, canonical_issue_id], + )?; + transaction.execute( + "UPDATE program_queue_label_ownership SET issue_id = ?2 WHERE issue_id = ?1", + params![previous_issue_id, canonical_issue_id], + )?; transaction.execute( "INSERT OR IGNORE INTO loop_guardrail_checkpoints ( project_id, issue_id, reason, fingerprint, run_id, attempt_number, @@ -1973,6 +2162,169 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; Ok(records) } + fn load_program_intake_state(&self, state: &mut StateData) -> Result<()> { + for record in self.list_all_program_intake_plans()? { + state.program_intake_plans.insert( + ProgramIntakePlanKey::new(&record.project_id, &record.program_id, &record.plan_id), + record, + ); + } + for record in self.list_all_program_issue_mappings()? { + state.program_issue_mappings.insert( + ProgramIssueMappingKey::new( + &record.project_id, + &record.program_id, + &record.node_id, + ), + record, + ); + } + for record in self.list_all_program_queue_label_ownership()? { + state.program_queue_label_ownership.insert( + ProgramQueueLabelOwnershipKey::new( + &record.project_id, + &record.program_id, + &record.node_id, + &record.label_name, + ), + record, + ); + } + + Ok(()) + } + + fn list_all_program_intake_plans(&self) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, plan_id, intake_kind, source_contract_id, \ + accepted_contract_fingerprint, public_summary, created_at, created_at_unix, \ + updated_at, updated_at_unix \ + FROM program_intake_plans \ + ORDER BY project_id ASC, program_id ASC, plan_id ASC", + )?; + let rows = statement.query_map([], program_intake_plan_row)?; + let mut records = Vec::new(); + + for row in rows { + records.push(row?); + } + + Ok(records) + } + + fn list_program_intake_plans( + &self, + project_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, plan_id, intake_kind, source_contract_id, \ + accepted_contract_fingerprint, public_summary, created_at, created_at_unix, \ + updated_at, updated_at_unix \ + FROM program_intake_plans \ + WHERE project_id = ?1 \ + ORDER BY updated_at_unix ASC, program_id ASC, plan_id ASC", + )?; + let rows = statement.query_map(params![project_id], program_intake_plan_row)?; + let mut records = Vec::new(); + + for row in rows { + records.push(row?); + } + + Ok(records) + } + + fn list_all_program_issue_mappings(&self) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, node_id, issue_id, issue_identifier, issue_state, \ + queue_intent, has_queue_label, queue_label_owned_by_program_reconciler, \ + has_active_label, has_opt_out_label, has_needs_attention_label, \ + has_generic_dispatch_briefing, created_at, created_at_unix, updated_at, \ + updated_at_unix \ + FROM program_issue_mappings \ + ORDER BY project_id ASC, program_id ASC, node_id ASC", + )?; + let rows = statement.query_map([], program_issue_mapping_row)?; + let mut records = Vec::new(); + + for row in rows { + records.push(row?); + } + + Ok(records) + } + + fn list_program_issue_mappings( + &self, + project_id: &str, + program_id: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, node_id, issue_id, issue_identifier, issue_state, \ + queue_intent, has_queue_label, queue_label_owned_by_program_reconciler, \ + has_active_label, has_opt_out_label, has_needs_attention_label, \ + has_generic_dispatch_briefing, created_at, created_at_unix, updated_at, \ + updated_at_unix \ + FROM program_issue_mappings \ + WHERE project_id = ?1 AND program_id = ?2 \ + ORDER BY updated_at_unix ASC, node_id ASC", + )?; + let rows = + statement.query_map(params![project_id, program_id], program_issue_mapping_row)?; + let mut records = Vec::new(); + + for row in rows { + records.push(row?); + } + + Ok(records) + } + + fn list_all_program_queue_label_ownership( + &self, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, node_id, issue_id, issue_identifier, label_name, \ + service_id, created_at, created_at_unix, updated_at, updated_at_unix \ + FROM program_queue_label_ownership \ + ORDER BY project_id ASC, program_id ASC, node_id ASC, label_name ASC", + )?; + let rows = statement.query_map([], program_queue_label_ownership_row)?; + let mut records = Vec::new(); + + for row in rows { + records.push(row?); + } + + Ok(records) + } + + fn program_queue_label_ownership_for_issue( + &self, + project_id: &str, + issue_id: &str, + label_name: &str, + ) -> Result> { + let mut statement = self.connection.prepare( + "SELECT project_id, program_id, node_id, issue_id, issue_identifier, label_name, \ + service_id, created_at, created_at_unix, updated_at, updated_at_unix \ + FROM program_queue_label_ownership \ + WHERE project_id = ?1 AND issue_id = ?2 AND label_name = ?3 \ + ORDER BY updated_at_unix ASC, program_id ASC, node_id ASC", + )?; + let rows = statement.query_map( + params![project_id, issue_id, label_name], + program_queue_label_ownership_row, + )?; + let mut records = Vec::new(); + + for row in rows { + records.push(row?); + } + + Ok(records) + } + 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, \ @@ -2419,7 +2771,7 @@ impl ExecutionProgramKey { #[derive(Clone, Debug)] struct ExecutionProgramRuntimeRecord { project_id: String, - source_contract_id: String, + source_contract_id: Option, program: ExecutionProgram, created_at: String, created_at_unix: i64, @@ -2446,6 +2798,56 @@ impl ExecutionProgramRuntimeRecord { } } +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct ProgramIntakePlanKey { + project_id: String, + program_id: String, + plan_id: String, +} +impl ProgramIntakePlanKey { + fn new(project_id: &str, program_id: &str, plan_id: &str) -> Self { + Self { + project_id: project_id.to_owned(), + program_id: program_id.to_owned(), + plan_id: plan_id.to_owned(), + } + } +} + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct ProgramIssueMappingKey { + project_id: String, + program_id: String, + node_id: String, +} +impl ProgramIssueMappingKey { + fn new(project_id: &str, program_id: &str, node_id: &str) -> Self { + Self { + project_id: project_id.to_owned(), + program_id: program_id.to_owned(), + node_id: node_id.to_owned(), + } + } +} + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct ProgramQueueLabelOwnershipKey { + project_id: String, + program_id: String, + node_id: String, + label_name: String, +} +impl ProgramQueueLabelOwnershipKey { + fn new(project_id: &str, program_id: &str, node_id: &str, label_name: &str) -> Self { + Self { + project_id: project_id.to_owned(), + program_id: program_id.to_owned(), + node_id: node_id.to_owned(), + label_name: label_name.to_owned(), + } + } +} + #[derive(Clone, Debug)] struct WorktreeMappingRecord { project_id: String, @@ -2691,7 +3093,7 @@ struct DecisionContractRuntimeRowParts { struct ExecutionProgramRuntimeRowParts { project_id: String, program_id: String, - source_contract_id: String, + source_contract_id: Option, payload_json: String, created_at: String, created_at_unix: i64, @@ -3552,7 +3954,7 @@ fn persist_execution_programs( params![ &record.project_id, record.program.program_id(), - &record.source_contract_id, + record.source_contract_id.as_deref(), payload_json, &record.created_at, record.created_at_unix, @@ -3565,6 +3967,165 @@ fn persist_execution_programs( Ok(()) } +fn persist_program_intake_state(transaction: &Transaction<'_>, state: &StateData) -> Result<()> { + for record in state.program_intake_plans.values() { + transaction.execute( + "INSERT OR REPLACE INTO program_intake_plans ( + project_id, program_id, plan_id, intake_kind, source_contract_id, + accepted_contract_fingerprint, public_summary, created_at, created_at_unix, + updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", + params![ + &record.project_id, + &record.program_id, + &record.plan_id, + &record.intake_kind, + record.source_contract_id.as_deref(), + &record.accepted_contract_fingerprint, + &record.public_summary, + &record.created_at, + record.created_at_unix, + &record.updated_at, + record.updated_at_unix, + ], + )?; + } + for record in state.program_issue_mappings.values() { + transaction.execute( + "INSERT OR REPLACE INTO program_issue_mappings ( + project_id, program_id, node_id, issue_id, issue_identifier, issue_state, + queue_intent, has_queue_label, queue_label_owned_by_program_reconciler, + has_active_label, has_opt_out_label, has_needs_attention_label, + has_generic_dispatch_briefing, created_at, created_at_unix, updated_at, + updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)", + params![ + &record.project_id, + &record.program_id, + &record.node_id, + &record.issue_id, + &record.issue_identifier, + &record.issue_state, + &record.queue_intent, + sqlite_bool_value(record.has_queue_label), + sqlite_bool_value(record.queue_label_owned_by_program_reconciler), + sqlite_bool_value(record.has_active_label), + sqlite_bool_value(record.has_opt_out_label), + sqlite_bool_value(record.has_needs_attention_label), + sqlite_bool_value(record.has_generic_dispatch_briefing), + &record.created_at, + record.created_at_unix, + &record.updated_at, + record.updated_at_unix, + ], + )?; + } + for record in state.program_queue_label_ownership.values() { + transaction.execute( + "INSERT OR REPLACE INTO program_queue_label_ownership ( + project_id, program_id, node_id, issue_id, issue_identifier, label_name, + service_id, created_at, created_at_unix, updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", + params![ + &record.project_id, + &record.program_id, + &record.node_id, + &record.issue_id, + &record.issue_identifier, + &record.label_name, + &record.service_id, + &record.created_at, + record.created_at_unix, + &record.updated_at, + record.updated_at_unix, + ], + )?; + } + + Ok(()) +} + +fn insert_program_intake_state( + connection: &Connection, + record: &ExecutionProgramRuntimeRecord, +) -> Result<()> { + for plan in derived_program_intake_plan_records(record) { + connection.execute( + "INSERT OR REPLACE INTO program_intake_plans ( + project_id, program_id, plan_id, intake_kind, source_contract_id, + accepted_contract_fingerprint, public_summary, created_at, created_at_unix, + updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", + params![ + &plan.project_id, + &plan.program_id, + &plan.plan_id, + &plan.intake_kind, + plan.source_contract_id.as_deref(), + &plan.accepted_contract_fingerprint, + &plan.public_summary, + &plan.created_at, + plan.created_at_unix, + &plan.updated_at, + plan.updated_at_unix, + ], + )?; + } + for mapping in derived_program_issue_mapping_records(record) { + connection.execute( + "INSERT OR REPLACE INTO program_issue_mappings ( + project_id, program_id, node_id, issue_id, issue_identifier, issue_state, + queue_intent, has_queue_label, queue_label_owned_by_program_reconciler, + has_active_label, has_opt_out_label, has_needs_attention_label, + has_generic_dispatch_briefing, created_at, created_at_unix, updated_at, + updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)", + params![ + &mapping.project_id, + &mapping.program_id, + &mapping.node_id, + &mapping.issue_id, + &mapping.issue_identifier, + &mapping.issue_state, + &mapping.queue_intent, + sqlite_bool_value(mapping.has_queue_label), + sqlite_bool_value(mapping.queue_label_owned_by_program_reconciler), + sqlite_bool_value(mapping.has_active_label), + sqlite_bool_value(mapping.has_opt_out_label), + sqlite_bool_value(mapping.has_needs_attention_label), + sqlite_bool_value(mapping.has_generic_dispatch_briefing), + &mapping.created_at, + mapping.created_at_unix, + &mapping.updated_at, + mapping.updated_at_unix, + ], + )?; + } + for ownership in derived_program_queue_label_ownership_records(record) { + connection.execute( + "INSERT OR REPLACE INTO program_queue_label_ownership ( + project_id, program_id, node_id, issue_id, issue_identifier, label_name, + service_id, created_at, created_at_unix, updated_at, updated_at_unix + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", + params![ + &ownership.project_id, + &ownership.program_id, + &ownership.node_id, + &ownership.issue_id, + &ownership.issue_identifier, + &ownership.label_name, + &ownership.service_id, + &ownership.created_at, + ownership.created_at_unix, + &ownership.updated_at, + ownership.updated_at_unix, + ], + )?; + } + + Ok(()) +} + fn persist_review_handoffs(transaction: &Transaction<'_>, state: &StateData) -> Result<()> { for record in state.review_handoffs.values() { transaction.execute( @@ -4466,12 +5027,12 @@ fn execution_program_record_from_row_parts( program.program_id() ); } - if parts.source_contract_id != program.source_contract_id() { + if parts.source_contract_id.as_deref() != program.source_contract_id() { eyre::bail!( "Execution program row `{}` carried source contract `{}` but payload references `{}`.", parts.program_id, - parts.source_contract_id, - program.source_contract_id() + parts.source_contract_id.as_deref().unwrap_or("none"), + program.source_contract_id().unwrap_or("none") ); } @@ -4486,6 +5047,74 @@ fn execution_program_record_from_row_parts( }) } +fn program_intake_plan_row( + row: &Row<'_>, +) -> std::result::Result { + Ok(ProgramIntakePlanRecord { + project_id: row.get(0)?, + program_id: row.get(1)?, + plan_id: row.get(2)?, + intake_kind: row.get(3)?, + source_contract_id: row.get(4)?, + accepted_contract_fingerprint: row.get(5)?, + public_summary: row.get(6)?, + created_at: row.get(7)?, + created_at_unix: row.get(8)?, + updated_at: row.get(9)?, + updated_at_unix: row.get(10)?, + }) +} + +fn program_issue_mapping_row( + row: &Row<'_>, +) -> std::result::Result { + Ok(ProgramIssueMappingRecord { + project_id: row.get(0)?, + program_id: row.get(1)?, + node_id: row.get(2)?, + issue_id: row.get(3)?, + issue_identifier: row.get(4)?, + issue_state: row.get(5)?, + queue_intent: row.get(6)?, + has_queue_label: sqlite_bool(row, 7)?, + queue_label_owned_by_program_reconciler: sqlite_bool(row, 8)?, + has_active_label: sqlite_bool(row, 9)?, + has_opt_out_label: sqlite_bool(row, 10)?, + has_needs_attention_label: sqlite_bool(row, 11)?, + has_generic_dispatch_briefing: sqlite_bool(row, 12)?, + created_at: row.get(13)?, + created_at_unix: row.get(14)?, + updated_at: row.get(15)?, + updated_at_unix: row.get(16)?, + }) +} + +fn program_queue_label_ownership_row( + row: &Row<'_>, +) -> std::result::Result { + Ok(ProgramQueueLabelOwnershipRecord { + project_id: row.get(0)?, + program_id: row.get(1)?, + node_id: row.get(2)?, + issue_id: row.get(3)?, + issue_identifier: row.get(4)?, + label_name: row.get(5)?, + service_id: row.get(6)?, + created_at: row.get(7)?, + created_at_unix: row.get(8)?, + updated_at: row.get(9)?, + updated_at_unix: row.get(10)?, + }) +} + +fn sqlite_bool(row: &Row<'_>, index: usize) -> std::result::Result { + Ok(row.get::<_, i64>(index)? != 0) +} + +fn sqlite_bool_value(value: bool) -> i64 { + if value { 1 } else { 0 } +} + fn connector_backoff_from_row(row: &Row<'_>) -> std::result::Result { Ok(ConnectorBackoff { project_id: row.get(0)?, @@ -4537,6 +5166,179 @@ fn compare_execution_program_runtime_records( .then_with(|| left.program.program_id().cmp(right.program.program_id())) } +fn compare_program_intake_plan_records( + left: &ProgramIntakePlanRecord, + right: &ProgramIntakePlanRecord, +) -> cmp::Ordering { + left.updated_at_unix + .cmp(&right.updated_at_unix) + .then_with(|| left.program_id.cmp(&right.program_id)) + .then_with(|| left.plan_id.cmp(&right.plan_id)) +} + +fn compare_program_issue_mapping_records( + left: &ProgramIssueMappingRecord, + right: &ProgramIssueMappingRecord, +) -> cmp::Ordering { + left.updated_at_unix + .cmp(&right.updated_at_unix) + .then_with(|| left.program_id.cmp(&right.program_id)) + .then_with(|| left.node_id.cmp(&right.node_id)) +} + +fn compare_program_queue_label_ownership_records( + left: &ProgramQueueLabelOwnershipRecord, + right: &ProgramQueueLabelOwnershipRecord, +) -> cmp::Ordering { + left.updated_at_unix + .cmp(&right.updated_at_unix) + .then_with(|| left.program_id.cmp(&right.program_id)) + .then_with(|| left.node_id.cmp(&right.node_id)) + .then_with(|| left.label_name.cmp(&right.label_name)) +} + +fn remove_derived_program_intake_state( + state: &mut StateData, + project_id: &str, + program_id: &str, +) { + state + .program_intake_plans + .retain(|key, _record| key.project_id != project_id || key.program_id != program_id); + state + .program_issue_mappings + .retain(|key, _record| key.project_id != project_id || key.program_id != program_id); + state.program_queue_label_ownership.retain(|key, _record| { + key.project_id != project_id || key.program_id != program_id + }); +} + +fn apply_derived_program_intake_state( + state: &mut StateData, + record: &ExecutionProgramRuntimeRecord, +) { + remove_derived_program_intake_state(state, &record.project_id, record.program.program_id()); + + for plan in derived_program_intake_plan_records(record) { + state.program_intake_plans.insert( + ProgramIntakePlanKey::new(&plan.project_id, &plan.program_id, &plan.plan_id), + plan, + ); + } + for mapping in derived_program_issue_mapping_records(record) { + state.program_issue_mappings.insert( + ProgramIssueMappingKey::new( + &mapping.project_id, + &mapping.program_id, + &mapping.node_id, + ), + mapping, + ); + } + for ownership in derived_program_queue_label_ownership_records(record) { + state.program_queue_label_ownership.insert( + ProgramQueueLabelOwnershipKey::new( + &ownership.project_id, + &ownership.program_id, + &ownership.node_id, + &ownership.label_name, + ), + ownership, + ); + } +} + +fn derived_program_intake_plan_records( + record: &ExecutionProgramRuntimeRecord, +) -> Vec { + record + .program + .program_intake_plan() + .map(|plan| { + vec![ProgramIntakePlanRecord { + project_id: record.project_id.clone(), + program_id: record.program.program_id().to_owned(), + plan_id: plan.plan_id().to_owned(), + intake_kind: plan.intake_kind().as_str().to_owned(), + source_contract_id: plan.source_contract_id().map(str::to_owned), + accepted_contract_fingerprint: plan.accepted_contract_fingerprint().to_owned(), + public_summary: plan.public_summary().to_owned(), + created_at: record.created_at.clone(), + created_at_unix: record.created_at_unix, + updated_at: record.updated_at.clone(), + updated_at_unix: record.updated_at_unix, + }] + }) + .unwrap_or_default() +} + +fn derived_program_issue_mapping_records( + record: &ExecutionProgramRuntimeRecord, +) -> Vec { + record + .program + .nodes() + .iter() + .filter_map(|node| { + let issue = node.linear_issue()?; + + Some(ProgramIssueMappingRecord { + project_id: record.project_id.clone(), + program_id: record.program.program_id().to_owned(), + node_id: node.node_id().to_owned(), + issue_id: issue.issue_id().to_owned(), + issue_identifier: issue.issue_identifier().to_owned(), + issue_state: issue.issue_state().to_owned(), + queue_intent: node.queue_intent().as_str().to_owned(), + has_queue_label: issue.has_queue_label(), + queue_label_owned_by_program_reconciler: issue + .queue_label_owned_by_program_reconciler(), + has_active_label: issue.has_active_label(), + has_opt_out_label: issue.has_opt_out_label(), + has_needs_attention_label: issue.has_needs_attention_label(), + has_generic_dispatch_briefing: issue.has_generic_dispatch_briefing(), + created_at: record.created_at.clone(), + created_at_unix: record.created_at_unix, + updated_at: record.updated_at.clone(), + updated_at_unix: record.updated_at_unix, + }) + }) + .collect() +} + +fn derived_program_queue_label_ownership_records( + record: &ExecutionProgramRuntimeRecord, +) -> Vec { + let label_name = tracker::automation_queue_label(record.program.service_id()); + + record + .program + .nodes() + .iter() + .filter_map(|node| { + let issue = node.linear_issue()?; + + if !issue.queue_label_owned_by_program_reconciler() { + return None; + } + + Some(ProgramQueueLabelOwnershipRecord { + project_id: record.project_id.clone(), + program_id: record.program.program_id().to_owned(), + node_id: node.node_id().to_owned(), + issue_id: issue.issue_id().to_owned(), + issue_identifier: issue.issue_identifier().to_owned(), + label_name: label_name.clone(), + service_id: record.program.service_id().to_owned(), + created_at: record.created_at.clone(), + created_at_unix: record.created_at_unix, + updated_at: record.updated_at.clone(), + updated_at_unix: record.updated_at_unix, + }) + }) + .collect() +} + 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 9c0476c0e..832b68bbb 100644 --- a/apps/decodex/src/state/models.rs +++ b/apps/decodex/src/state/models.rs @@ -595,7 +595,7 @@ impl DecisionContractRecord { pub(crate) struct ExecutionProgramRecord { project_id: String, program: ExecutionProgram, - source_contract_id: String, + source_contract_id: Option, created_at: String, created_at_unix: i64, updated_at: String, @@ -615,8 +615,224 @@ impl ExecutionProgramRecord { self.program.program_id() } - pub(crate) fn source_contract_id(&self) -> &str { - &self.source_contract_id + pub(crate) fn source_contract_id(&self) -> Option<&str> { + self.source_contract_id.as_deref() + } + + 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 + } +} + +/// SQLite-backed Program Intake Plan projection retained by the local runtime. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ProgramIntakePlanRecord { + project_id: String, + program_id: String, + plan_id: String, + intake_kind: String, + source_contract_id: Option, + accepted_contract_fingerprint: String, + public_summary: String, + created_at: String, + created_at_unix: i64, + updated_at: String, + updated_at_unix: i64, +} +#[allow(dead_code)] +impl ProgramIntakePlanRecord { + pub(crate) fn project_id(&self) -> &str { + &self.project_id + } + + pub(crate) fn program_id(&self) -> &str { + &self.program_id + } + + pub(crate) fn plan_id(&self) -> &str { + &self.plan_id + } + + pub(crate) fn intake_kind(&self) -> &str { + &self.intake_kind + } + + pub(crate) fn source_contract_id(&self) -> Option<&str> { + self.source_contract_id.as_deref() + } + + pub(crate) fn accepted_contract_fingerprint(&self) -> &str { + &self.accepted_contract_fingerprint + } + + pub(crate) fn public_summary(&self) -> &str { + &self.public_summary + } + + 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 + } +} + +/// SQLite-backed normal Linear issue mapping for one internal program node. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ProgramIssueMappingRecord { + project_id: String, + program_id: String, + node_id: String, + issue_id: String, + issue_identifier: String, + issue_state: String, + queue_intent: String, + has_queue_label: bool, + queue_label_owned_by_program_reconciler: bool, + has_active_label: bool, + has_opt_out_label: bool, + has_needs_attention_label: bool, + has_generic_dispatch_briefing: bool, + created_at: String, + created_at_unix: i64, + updated_at: String, + updated_at_unix: i64, +} +#[allow(dead_code)] +impl ProgramIssueMappingRecord { + pub(crate) fn project_id(&self) -> &str { + &self.project_id + } + + pub(crate) fn program_id(&self) -> &str { + &self.program_id + } + + pub(crate) fn node_id(&self) -> &str { + &self.node_id + } + + pub(crate) fn issue_id(&self) -> &str { + &self.issue_id + } + + pub(crate) fn issue_identifier(&self) -> &str { + &self.issue_identifier + } + + pub(crate) fn issue_state(&self) -> &str { + &self.issue_state + } + + pub(crate) fn queue_intent(&self) -> &str { + &self.queue_intent + } + + pub(crate) fn has_queue_label(&self) -> bool { + self.has_queue_label + } + + pub(crate) fn queue_label_owned_by_program_reconciler(&self) -> bool { + self.queue_label_owned_by_program_reconciler + } + + pub(crate) fn has_active_label(&self) -> bool { + self.has_active_label + } + + pub(crate) fn has_opt_out_label(&self) -> bool { + self.has_opt_out_label + } + + pub(crate) fn has_needs_attention_label(&self) -> bool { + self.has_needs_attention_label + } + + pub(crate) fn has_generic_dispatch_briefing(&self) -> bool { + self.has_generic_dispatch_briefing + } + + 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 + } +} + +/// SQLite-backed queue-label ownership proof for one program-owned label. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ProgramQueueLabelOwnershipRecord { + project_id: String, + program_id: String, + node_id: String, + issue_id: String, + issue_identifier: String, + label_name: String, + service_id: String, + created_at: String, + created_at_unix: i64, + updated_at: String, + updated_at_unix: i64, +} +#[allow(dead_code)] +impl ProgramQueueLabelOwnershipRecord { + pub(crate) fn project_id(&self) -> &str { + &self.project_id + } + + pub(crate) fn program_id(&self) -> &str { + &self.program_id + } + + pub(crate) fn node_id(&self) -> &str { + &self.node_id + } + + pub(crate) fn issue_id(&self) -> &str { + &self.issue_id + } + + pub(crate) fn issue_identifier(&self) -> &str { + &self.issue_identifier + } + + pub(crate) fn label_name(&self) -> &str { + &self.label_name + } + + pub(crate) fn service_id(&self) -> &str { + &self.service_id } pub(crate) fn created_at(&self) -> &str { diff --git a/apps/decodex/src/state/store.rs b/apps/decodex/src/state/store.rs index 02d325224..f9d3e5823 100644 --- a/apps/decodex/src/state/store.rs +++ b/apps/decodex/src/state/store.rs @@ -363,6 +363,20 @@ impl StateStore { { record.source_issue_id = Some(canonical_issue_id.to_owned()); } + for record in state + .program_issue_mappings + .values_mut() + .filter(|record| record.issue_id == previous_issue_id) + { + record.issue_id = canonical_issue_id.to_owned(); + } + for record in state + .program_queue_label_ownership + .values_mut() + .filter(|record| record.issue_id == previous_issue_id) + { + record.issue_id = canonical_issue_id.to_owned(); + } self.retarget_issue_identity_locked(previous_issue_id, canonical_issue_id) } @@ -1927,7 +1941,7 @@ impl StateStore { }); let record = ExecutionProgramRuntimeRecord { project_id: project_id.to_owned(), - source_contract_id: program.source_contract_id().to_owned(), + source_contract_id: program.source_contract_id().map(str::to_owned), program, created_at, created_at_unix, @@ -1936,6 +1950,9 @@ impl StateStore { }; state.execution_programs.insert(record.key(), record.clone()); + + apply_derived_program_intake_state(&mut state, &record); + self.upsert_execution_program_locked(&record)?; Ok(record.as_public()) @@ -1993,7 +2010,8 @@ impl StateStore { .execution_programs .values() .filter(|record| { - record.project_id == project_id && record.source_contract_id == source_contract_id + record.project_id == project_id + && record.source_contract_id.as_deref() == Some(source_contract_id) }) .cloned() .collect::>(); @@ -2035,6 +2053,101 @@ impl StateStore { Ok(records.into_iter().map(|record| record.as_public()).collect()) } + /// List local Program Intake Plan records retained for one project. + #[allow(dead_code)] + pub(crate) fn list_program_intake_plans( + &self, + project_id: &str, + ) -> Result> { + validate_required_execution_program_field("project_id", project_id)?; + + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite.list_program_intake_plans(project_id); + } + + let state = self.lock()?; + let mut records = state + .program_intake_plans + .values() + .filter(|record| record.project_id == project_id) + .cloned() + .collect::>(); + + records.sort_by(compare_program_intake_plan_records); + + Ok(records) + } + + /// List local issue mappings for one internal Execution Program. + #[allow(dead_code)] + pub(crate) fn list_program_issue_mappings( + &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)?; + + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite.list_program_issue_mappings(project_id, program_id); + } + + let state = self.lock()?; + let mut records = state + .program_issue_mappings + .values() + .filter(|record| record.project_id == project_id && record.program_id == program_id) + .cloned() + .collect::>(); + + records.sort_by(compare_program_issue_mapping_records); + + Ok(records) + } + + /// Read program-owned queue-label evidence for one mapped issue and label. + #[allow(dead_code)] + pub(crate) fn program_queue_label_ownership_for_issue( + &self, + project_id: &str, + issue_id: &str, + label_name: &str, + ) -> Result> { + validate_required_execution_program_field("project_id", project_id)?; + validate_required_execution_program_field("issue_id", issue_id)?; + validate_required_execution_program_field("label_name", label_name)?; + + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite.program_queue_label_ownership_for_issue( + project_id, + issue_id, + label_name, + ); + } + + let state = self.lock()?; + let mut records = state + .program_queue_label_ownership + .values() + .filter(|record| { + record.project_id == project_id + && record.issue_id == issue_id + && record.label_name == label_name + }) + .cloned() + .collect::>(); + + records.sort_by(compare_program_queue_label_ownership_records); + + Ok(records) + } + /// Count protocol journal records for one run. pub fn event_count(&self, run_id: &str) -> Result { let state = self.lock()?; diff --git a/apps/decodex/src/state/tests.rs b/apps/decodex/src/state/tests.rs index 679b89567..b1e6d694a 100644 --- a/apps/decodex/src/state/tests.rs +++ b/apps/decodex/src/state/tests.rs @@ -106,7 +106,8 @@ fn sample_execution_program(contract: &DecisionContract) -> ExecutionProgram { .expect("validation expectations should attach") .with_linear_issue( ExecutionLinearIssueMapping::new("issue-853", "XY-853", "Todo") - .expect("issue mapping should validate"), + .expect("issue mapping should validate") + .with_program_owned_queue_label(true), ) .expect("issue mapping should attach"); @@ -2939,7 +2940,7 @@ fn execution_programs_persist_reload_and_list_by_contract() { 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.source_contract_id(), Some("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()); @@ -2951,7 +2952,7 @@ fn execution_programs_persist_reload_and_list_by_contract() { .expect("execution program should exist"); assert_eq!(reloaded.created_at(), record.created_at()); - assert_eq!(reloaded.program().source_contract_id(), "research-x-loop-contract"); + assert_eq!(reloaded.program().source_contract_id(), Some("research-x-loop-contract")); let contract_programs = reopened .list_execution_programs_for_contract("decodex", "research-x-loop-contract") @@ -2965,6 +2966,31 @@ fn execution_programs_persist_reload_and_list_by_contract() { assert_eq!(project_programs.len(), 1); assert_eq!(project_programs[0].program_id(), "program-853"); + + let intake_plans = + reopened.list_program_intake_plans("decodex").expect("program intake plans should list"); + + assert_eq!(intake_plans.len(), 1); + assert_eq!(intake_plans[0].program_id(), "program-853"); + assert_eq!(intake_plans[0].intake_kind(), "goal_intake"); + assert_eq!(intake_plans[0].source_contract_id(), Some("research-x-loop-contract")); + + let issue_mappings = reopened + .list_program_issue_mappings("decodex", "program-853") + .expect("program issue mappings should list"); + + assert_eq!(issue_mappings.len(), 1); + assert_eq!(issue_mappings[0].node_id(), "runtime-readiness"); + assert_eq!(issue_mappings[0].issue_identifier(), "XY-853"); + assert!(issue_mappings[0].queue_label_owned_by_program_reconciler()); + + let queue_ownership = reopened + .program_queue_label_ownership_for_issue("decodex", "issue-853", "decodex:queued:decodex") + .expect("program queue label ownership should read"); + + assert_eq!(queue_ownership.len(), 1); + assert_eq!(queue_ownership[0].program_id(), "program-853"); + assert_eq!(queue_ownership[0].issue_identifier(), "XY-853"); } #[test] diff --git a/docs/spec/loop-runtime.md b/docs/spec/loop-runtime.md index 057c6d262..931362f30 100644 --- a/docs/spec/loop-runtime.md +++ b/docs/spec/loop-runtime.md @@ -306,8 +306,12 @@ 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. +`decodex.execution_program/1` record. The adjacent runtime readback rows are +`program_intake_plans`, `program_issue_mappings`, and +`program_queue_label_ownership`; they are derived from the versioned program payload +and exist so operator status, reconciliation, and tests can query intake state without +copying private graph payloads into Linear. Linear issue descriptions, Linear comments, +and operator summaries may expose only sparse public projections. The payload carries: @@ -329,6 +333,15 @@ 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. +The first operator CLI surface for existing issues is +`decodex intake issues --project ... --dry-run`, or the same +command with `--config `. Dry-run reads tracker state and prints a +deterministic ready/held/blocked/stale/unmapped report without mutating Linear and +without persisting local runtime rows. `--persist` is an explicit local-runtime write: +it stores the Program Intake Plan, Execution Program payload, issue mappings, and any +already-proven program-owned queue-label evidence, but it still must not apply or +remove `decodex:queued:`. + ## Internal Execution Program An Execution Program is internal loop-runtime state derived from an accepted Program diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index e52de79e7..4e190af92 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -63,6 +63,13 @@ state or this state machine. - `decodex research compile` and `decodex research promote` are runtime-local Decision Contract writes. They update the SQLite `decision_contracts` surface and do not by themselves create Linear issues, queue intent, goals, or executable lanes. +- `decodex intake issues ... --dry-run` is a tracker-read-only operator + surface for existing Linear issues. It classifies the supplied batch as ready, + held, blocked, stale, or unmapped and builds the same internal program model used + by later persistence, but it must not mutate Linear or write local runtime rows. + `--persist` writes only local runtime Program Intake and Execution Program state; + queue labels remain untouched until a later reconciliation surface is explicitly + authorized. The evidence boundary is ordered from private runtime authority to public collaboration mirror: @@ -72,6 +79,9 @@ 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 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 `program_intake_plans` | Queryable local projection of `decodex.program_intake_plan/1` metadata, including intake kind, source contract when present, authority fingerprint, and public-safe summary. | +| Runtime SQLite `program_issue_mappings` | Queryable local projection of each internal program node's mapped Linear issue, tracker state, queue intent, service label facts, dispatch-briefing fact, and program-owned queue-label flag. | +| Runtime SQLite `program_queue_label_ownership` | Fail-closed evidence that program reconciliation owns a specific service queue label for one service, program, node, and issue. Queue-label removal authority must come from this table or the equivalent canonical program payload evidence. | | 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. | diff --git a/docs/spec/tracker-tools.md b/docs/spec/tracker-tools.md index 89ec6548d..5f58ff896 100644 --- a/docs/spec/tracker-tools.md +++ b/docs/spec/tracker-tools.md @@ -125,6 +125,11 @@ In either invalid case, `decodex` must fail the attempt rather than infer which 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. +- `decodex intake issues ... --dry-run` may read tracker state for supplied issues + and render a local ready/held/blocked/stale/unmapped report, but it must not call + tracker label mutation tools or write Linear comments. `--persist` may write local + runtime Program Intake records and issue mappings, but it still must not apply or + remove `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. From 8d9315b8d80df9720eadb6d6612e73a7c3979899 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Fri, 12 Jun 2026 00:22:41 +0800 Subject: [PATCH 2/2] {"schema":"decodex/commit/1","summary":"Repair Program Intake dry-run and report determinism","authority":"XY-938"} --- apps/decodex/src/program_intake.rs | 183 +++++++++++++++++++++++------ 1 file changed, 145 insertions(+), 38 deletions(-) diff --git a/apps/decodex/src/program_intake.rs b/apps/decodex/src/program_intake.rs index 25da101c0..1455139b4 100644 --- a/apps/decodex/src/program_intake.rs +++ b/apps/decodex/src/program_intake.rs @@ -145,7 +145,7 @@ pub(crate) fn run_issue_batch_intake_command( let config = ServiceConfig::from_path(&config_path)?; let workflow = WorkflowDocument::from_path(config.workflow_path())?; - runtime::register_project_config(&state_store, &config_path, true)?; + register_intake_project_config_for_persist(&state_store, &config_path, request.persist)?; let tracker = LinearClient::new(config.tracker().resolve_api_key()?)?; @@ -345,6 +345,18 @@ fn resolve_intake_project_config_path( }) } +fn register_intake_project_config_for_persist( + state_store: &StateStore, + config_path: &Path, + persist: bool, +) -> Result<()> { + if persist { + runtime::register_project_config(state_store, config_path, true)?; + } + + Ok(()) +} + fn normalize_issue_identifiers(issue_identifiers: Vec) -> Result> { let mut normalized = issue_identifiers .into_iter() @@ -527,36 +539,42 @@ fn issue_dependencies( issue: &TrackerIssue, supplied_node_ids: &BTreeMap, ) -> Result> { - issue - .blockers - .iter() - .map(|blocker| { - let dependency_id = supplied_node_ids - .get(&blocker.identifier) - .cloned() - .unwrap_or_else(|| blocker.identifier.clone()); + let mut dependencies = BTreeMap::new(); - ExecutionProgramDependency::new(dependency_id) - }) - .collect() + for blocker in &issue.blockers { + let dependency_id = supplied_node_ids + .get(&blocker.identifier) + .cloned() + .unwrap_or_else(|| blocker.identifier.clone()); + + dependencies + .entry(dependency_id.clone()) + .or_insert(ExecutionProgramDependency::new(dependency_id)?); + } + + Ok(dependencies.into_values().collect()) } fn dependency_snapshots_for( issue: &TrackerIssue, supplied_node_ids: &BTreeMap, ) -> Result> { - issue - .blockers - .iter() - .map(|blocker| { - let dependency_id = supplied_node_ids - .get(&blocker.identifier) - .cloned() - .unwrap_or_else(|| blocker.identifier.clone()); + let mut snapshots = BTreeMap::new(); + + for blocker in &issue.blockers { + let dependency_id = supplied_node_ids + .get(&blocker.identifier) + .cloned() + .unwrap_or_else(|| blocker.identifier.clone()); + let snapshot = ExecutionDependencySnapshot::tracker_state( + dependency_id.clone(), + blocker.state.name.clone(), + )?; + + snapshots.entry(dependency_id).or_insert(snapshot); + } - ExecutionDependencySnapshot::tracker_state(dependency_id, blocker.state.name.clone()) - }) - .collect() + Ok(snapshots.into_values().collect()) } fn issue_conflict_domains(issue: &TrackerIssue) -> Result> { @@ -584,6 +602,10 @@ fn issue_conflict_domains(issue: &TrackerIssue) -> Result>(); + + blockers.sort(); + blockers.dedup(); + + let mut conflict_domains = issue_conflict_domains(issue) + .unwrap_or_default() + .into_iter() + .map(|domain| format!("{}:{}", domain.kind().as_str(), domain.key())) + .collect::>(); + + conflict_domains.sort(); + conflict_domains.dedup(); IssueBatchIntakeIssueReport { issue_identifier: issue.identifier.clone(), issue_id: Some(issue.id.clone()), @@ -618,12 +655,8 @@ fn issue_report_row( ), queue_label_action: evaluation.queue_label_action().map(queue_label_action_name), reasons, - blockers: issue.blockers.iter().map(|blocker| blocker.identifier.clone()).collect(), - conflict_domains: issue_conflict_domains(issue) - .unwrap_or_default() - .into_iter() - .map(|domain| format!("{}:{}", domain.kind().as_str(), domain.key())) - .collect(), + blockers, + conflict_domains, } } @@ -709,20 +742,23 @@ fn issue_has_generic_dispatch_briefing(issue: &TrackerIssue) -> bool { #[cfg(test)] mod tests { - use std::collections::HashMap; + use std::{collections::HashMap, fs, path::Path}; + + use tempfile::TempDir; use crate::{ program_intake::{self, IssueBatchIntakeClassification}, state::StateStore, tracker::{ - IssueTracker, TrackerComment, TrackerIssue, TrackerIssueBlocker, TrackerState, - TrackerTeam, + IssueTracker, TrackerComment, TrackerIssue, TrackerIssueBlocker, TrackerLabel, + TrackerState, TrackerTeam, }, workflow::WorkflowDocument, }; trait TestIssueExt { fn with_blocker(self, identifier: &str, state: &str) -> Self; + fn with_label(self, name: &str) -> Self; } #[derive(Default)] @@ -813,6 +849,12 @@ mod tests { self } + + fn with_label(mut self, name: &str) -> Self { + self.labels.push(TrackerLabel { id: format!("label-{name}"), name: name.to_owned() }); + + self + } } #[test] @@ -824,7 +866,11 @@ mod tests { issue("XY-1", "Todo"), issue("XY-2", "In Progress"), issue("XY-3", "Done"), - issue("XY-4", "Todo").with_blocker("XY-10", "Todo"), + issue("XY-4", "Todo") + .with_blocker("XY-20", "Todo") + .with_blocker("XY-10", "Todo") + .with_label("repo:zeta") + .with_label("repo:alpha"), ]); let report = program_intake::run_issue_batch_intake( &store, @@ -850,11 +896,48 @@ mod tests { assert_eq!(report.counts.unmapped, 1); assert_eq!(report.issues[0].issue_identifier, "XY-1"); assert_eq!(report.issues[0].classification, IssueBatchIntakeClassification::Ready); + + let blocked = report + .issues + .iter() + .find(|issue| issue.issue_identifier == "XY-4") + .expect("blocked issue should be reported"); + + assert_eq!(blocked.blockers, vec![String::from("XY-10"), String::from("XY-20")]); + assert_eq!( + blocked.conflict_domains, + vec![ + String::from("module:alpha"), + String::from("module:zeta"), + String::from("tracker_ownership:XY-4"), + ] + ); assert!( store.list_execution_programs("decodex").expect("program list should read").is_empty() ); } + #[test] + fn project_registration_is_persist_only_for_command_path() { + let store = StateStore::open_in_memory().expect("store should open"); + let temp_dir = TempDir::new().expect("temp dir should create"); + let config_path = write_project_files(temp_dir.path()); + + program_intake::register_intake_project_config_for_persist(&store, &config_path, false) + .expect("dry-run registration should no-op"); + + assert!(store.list_projects().expect("projects should list").is_empty()); + + program_intake::register_intake_project_config_for_persist(&store, &config_path, true) + .expect("persist registration should write"); + + let projects = store.list_projects().expect("projects should list"); + + assert_eq!(projects.len(), 1); + assert_eq!(projects[0].service_id(), "decodex"); + assert!(projects[0].enabled()); + } + #[test] fn issue_batch_persist_writes_program_and_adjacent_intake_state() { let store = StateStore::open_in_memory().expect("store should open"); @@ -889,8 +972,11 @@ mod tests { } fn workflow() -> WorkflowDocument { - WorkflowDocument::parse_markdown( - r#"+++ + WorkflowDocument::parse_markdown(workflow_markdown()).expect("workflow should parse") + } + + fn workflow_markdown() -> &'static str { + r#"+++ version = 1 [tracker] provider = "linear" @@ -919,9 +1005,7 @@ timeout_seconds = 60 [context] read_first = [] +++ -"#, - ) - .expect("workflow should parse") +"# } fn test_config() -> crate::config::ServiceConfig { @@ -942,6 +1026,29 @@ worktree_root = ".worktrees" .expect("config should parse") } + fn write_project_files(project_dir: &Path) -> std::path::PathBuf { + fs::write(project_dir.join("WORKFLOW.md"), workflow_markdown()) + .expect("workflow should write"); + fs::write( + project_dir.join("project.toml"), + r#" +service_id = "decodex" +[tracker] +api_key_env_var = "HOME" +[github] +token_env_var = "HOME" +[codex] +review = "standard" +[paths] +repo_root = "." +worktree_root = ".worktrees" +"#, + ) + .expect("project config should write"); + + project_dir.join("project.toml") + } + fn issue(identifier: &str, state: &str) -> TrackerIssue { TrackerIssue { id: format!("id-{identifier}"),