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..1455139b4 --- /dev/null +++ b/apps/decodex/src/program_intake.rs @@ -0,0 +1,1075 @@ +//! 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())?; + + register_intake_project_config_for_persist(&state_store, &config_path, request.persist)?; + + 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 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() + .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> { + let mut dependencies = BTreeMap::new(); + + 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> { + 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); + } + + Ok(snapshots.into_values().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)?); + } + } + } + + domains.sort_by(|left, right| { + left.kind().as_str().cmp(right.kind().as_str()).then_with(|| left.key().cmp(right.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(); + + let mut blockers = + issue.blockers.iter().map(|blocker| blocker.identifier.clone()).collect::>(); + + 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()), + 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, + conflict_domains, + } +} + +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, fs, path::Path}; + + use tempfile::TempDir; + + use crate::{ + program_intake::{self, IssueBatchIntakeClassification}, + state::StateStore, + tracker::{ + 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)] + 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 + } + + fn with_label(mut self, name: &str) -> Self { + self.labels.push(TrackerLabel { id: format!("label-{name}"), name: name.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-20", "Todo") + .with_blocker("XY-10", "Todo") + .with_label("repo:zeta") + .with_label("repo:alpha"), + ]); + 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); + + 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"); + 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(workflow_markdown()).expect("workflow should parse") + } + + fn workflow_markdown() -> &'static str { + 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 = [] ++++ +"# + } + + 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 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}"), + 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.