From 082d32d3ae0fd8d86cfddf93987ebf900bc4c2d4 Mon Sep 17 00:00:00 2001 From: Punisheroot <44579963+Punisheroot@users.noreply.github.com> Date: Tue, 4 Aug 2026 14:19:42 +0200 Subject: [PATCH] feat(runtime): add parent-owned lifecycle --- PROJECT_STATUS.md | 13 +- crates/needle-core/src/lib.rs | 2 + crates/needle-core/src/lifecycle.rs | 804 +++++++++++++++ crates/needle-core/src/lifecycle/model.rs | 606 +++++++++++ crates/needle-core/src/lifecycle/tests.rs | 714 +++++++++++++ crates/needle-runtime/src/changes.rs | 42 +- crates/needle-runtime/src/lib.rs | 10 +- .../needle-runtime/src/semantic_validation.rs | 4 +- crates/needle-runtime/src/store.rs | 84 +- crates/needle-runtime/src/store/changes.rs | 64 +- crates/needle-runtime/src/store/lifecycles.rs | 878 ++++++++++++++++ .../src/store/lifecycles/tests.rs | 958 ++++++++++++++++++ .../src/store/role_profiles/tests.rs | 6 +- docs/ARCHITECTURE.md | 39 +- docs/ROADMAP.md | 7 + docs/VERIFIED_CHANGES.md | 27 + 16 files changed, 4236 insertions(+), 22 deletions(-) create mode 100644 crates/needle-core/src/lifecycle.rs create mode 100644 crates/needle-core/src/lifecycle/model.rs create mode 100644 crates/needle-core/src/lifecycle/tests.rs create mode 100644 crates/needle-runtime/src/store/lifecycles.rs create mode 100644 crates/needle-runtime/src/store/lifecycles/tests.rs diff --git a/PROJECT_STATUS.md b/PROJECT_STATUS.md index 6ff4a5d..767f9d7 100644 --- a/PROJECT_STATUS.md +++ b/PROJECT_STATUS.md @@ -30,7 +30,7 @@ one. | Claim-level reuse | Validator-extracted claims, claim proofs, mixed planning, and bounded authoritative location, runtime-flow, and focused-test claims | Deterministic freshness, mutation, negative, projection, economics, and performance cases | **Implemented; offline validated** | | Verified changes | Isolated patch preparation, independent verifier, one repair, explicit journaled apply | Simulator and focused persistence, isolation, drift, and recovery tests | **Implemented; offline validated** | | Codex role-profile control plane | Canonical Codex role definitions, bounded policies, immutable revisions, state-digest CAS, SQLite persistence, audit records, explicit WorkerProfile projection, bounded digest-bound HTTP API, local editor, and frozen session/worker/cache provenance | Focused deterministic Rust and frontend tests; no parent-owned lifecycle execution | **Implemented; offline validated** | -| Codex development lifecycle orchestration | Evidence, patch, test, verification, approval, and apply primitives exist; the configurable parent-owned role lifecycle is not integrated | Component-level offline evidence only | **Pending** | +| Codex development lifecycle contract | Durable depth-one `explore -> implement -> test -> review -> verify -> apply` state machine, frozen role-profile/test-plan anchors, budgets, cancellation, one repair, explicit approval, transactional SQLite projection/events, restart replay, and journaled apply integration; Codex executor and lifecycle UI are not wired | Deterministic transition matrix, adversarial domain tests, persistence/restart tests, and concurrent apply serialization without provider calls | **Implemented; offline validated** | | Other-host subagent configuration | Configuration-only interoperability is planned for Claude Code and Cursor, followed by OpenCode and Antigravity | Not available | **Pending** | | Multi-host orchestration | Execution remains Codex-only; non-Codex execution follows configuration interoperability, a host contract, and conformance evidence | Not available | **Pending** | | Release readiness | Stable packaging, supported installation, compatibility policy, second live platform, powered corpus | Not available | **Pending** | @@ -63,7 +63,7 @@ provider-backed claim-authority observation exists. | Canonical named Codex role-profile domain and revision store | **Implemented; offline validated** | | Named role-profile HTTP/editor | **Implemented; offline validated; configuration mutations only** | | Role-profile session, worker, cache, attempt, and audit provenance | **Implemented; offline validated; Codex-first** | -| Parent-owned role-profile lifecycle integration | **Pending; Codex-first** | +| Parent-owned lifecycle domain and change-journal integration | **Implemented; offline validated; executor pending** | | Non-Codex subagent configuration | **Pending; configuration only before execution** | | Non-Codex execution and orchestration | **Pending; later milestone** | | Stable public API or configuration compatibility | **Pending** | @@ -126,10 +126,11 @@ validation. provider-backed evidence. - Verified changes have no provider-backed patcher or verifier observation. - Canonical role-profile definitions, revision persistence, bounded HTTP/editor - flows, request-time preflight, and frozen session/worker/cache/attempt/audit - provenance are implemented and offline validated. They do not provide a - parent-owned lifecycle executor or automatic profile activation; activation - is an explicit configuration change. + flows, request-time preflight, frozen session/worker/cache/attempt/audit + provenance, and the parent-owned lifecycle domain/journal are implemented and + offline validated. The lifecycle contract does not launch or supervise Codex + workers and has no read/timeline UI; automatic profile activation remains + unsupported and activation is an explicit configuration change. - The verifier handles a deterministic serial set of up to four distinct associated certified test plans; exact duplicates collapse to one execution, while over-cap and unavailable plans fail closed. This behavior is offline diff --git a/crates/needle-core/src/lib.rs b/crates/needle-core/src/lib.rs index 3dd78d8..33e0b62 100644 --- a/crates/needle-core/src/lib.rs +++ b/crates/needle-core/src/lib.rs @@ -10,6 +10,7 @@ mod artifact; mod change; pub mod claim; mod domain; +mod lifecycle; mod multi_need; mod role_profile; mod semantic; @@ -24,6 +25,7 @@ pub use claim::{ MAX_SELECTED_CLAIMS, ProofComponent, runtime_flow_anchor, }; pub use domain::*; +pub use lifecycle::*; pub use multi_need::*; pub use role_profile::*; pub use semantic::*; diff --git a/crates/needle-core/src/lifecycle.rs b/crates/needle-core/src/lifecycle.rs new file mode 100644 index 0000000..146b0ac --- /dev/null +++ b/crates/needle-core/src/lifecycle.rs @@ -0,0 +1,804 @@ +use crate::{ + ChangeApplyId, ChangeApplyStatus, ChangeId, Digest, PatchId, VerificationArtifactId, + VerificationStatus, +}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeSet; +use thiserror::Error; + +mod model; +pub use model::*; + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DevelopmentLifecycle { + pub id: LifecycleId, + pub change_id: ChangeId, + pub source_snapshot: Digest, + pub spec: LifecycleSpec, + pub phase: LifecyclePhase, + pub status: LifecycleStatus, + pub terminal_outcome: Option, + pub terminal_reason: Option, + pub generation: u64, + pub usage: LifecycleUsage, + pub exploration_artifacts: Vec, + pub patch: Option, + pub test_results: Vec, + pub review: Option, + pub verification: Option, + pub repair_reserved: bool, + pub repair_consumed: bool, + pub approval: Option, + pub apply_id: Option, + pub created_unix_ms: u64, + pub updated_unix_ms: u64, +} + +impl DevelopmentLifecycle { + pub fn new( + change_id: ChangeId, + source_snapshot: Digest, + spec: LifecycleSpec, + created_unix_ms: u64, + ) -> Result { + spec.validate()?; + let state = Self { + id: LifecycleId::compute(&change_id, source_snapshot, &spec.profiles), + change_id, + source_snapshot, + spec, + phase: LifecyclePhase::Explore, + status: LifecycleStatus::Active, + terminal_outcome: None, + terminal_reason: None, + generation: 0, + usage: LifecycleUsage::default(), + exploration_artifacts: Vec::new(), + patch: None, + test_results: Vec::new(), + review: None, + verification: None, + repair_reserved: false, + repair_consumed: false, + approval: None, + apply_id: None, + created_unix_ms, + updated_unix_ms: created_unix_ms, + }; + state.validate()?; + Ok(state) + } + + pub fn state_digest(&self) -> Digest { + Digest::blake3(serde_json::to_vec(self).unwrap_or_default()) + } + + pub fn validate(&self) -> Result<(), LifecycleError> { + self.spec.validate()?; + if self.id + != LifecycleId::compute(&self.change_id, self.source_snapshot, &self.spec.profiles) + || self.updated_unix_ms < self.created_unix_ms + || !self.usage.within(&self.spec.budget) + || self.generation >= MAX_LIFECYCLE_EVENTS as u64 + { + return Err(LifecycleError::InvalidState); + } + if self.status.terminal() != self.terminal_outcome.is_some() + || self.status.terminal() + != (self.terminal_reason.is_some() + || self.terminal_outcome == Some(LifecycleTerminalOutcome::Applied) + || self.terminal_outcome == Some(LifecycleTerminalOutcome::RolledBack)) + { + return Err(LifecycleError::InvalidState); + } + if self.repair_reserved && self.repair_consumed { + return Err(LifecycleError::InvalidState); + } + self.validate_artifact_chain()?; + self.validate_status_shape()?; + if serde_json::to_vec(self).map_err(|_| LifecycleError::InvalidState)?.len() + > MAX_LIFECYCLE_STATE_BYTES + { + return Err(LifecycleError::StateTooLarge); + } + Ok(()) + } + + pub fn transition( + &self, + transition: LifecycleTransition, + created_unix_ms: u64, + ) -> Result<(Self, LifecycleEvent), LifecycleError> { + self.validate()?; + if self.status.terminal() { + return Err(LifecycleError::Terminal); + } + if created_unix_ms < self.updated_unix_ms { + return Err(LifecycleError::TimeRegression); + } + match &transition { + LifecycleTransition::CompleteReview { review, .. } + if review.created_unix_ms < self.created_unix_ms + || review.created_unix_ms > created_unix_ms => + { + return Err(LifecycleError::TimeRegression); + } + LifecycleTransition::ApproveApply { approval } + if approval.decided_unix_ms < self.created_unix_ms + || approval.decided_unix_ms > created_unix_ms => + { + return Err(LifecycleError::TimeRegression); + } + _ => {} + } + let prior_digest = self.state_digest(); + let mut next = self.clone(); + next.apply_transition(&transition)?; + next.generation = + next.generation.checked_add(1).ok_or(LifecycleError::GenerationOverflow)?; + next.updated_unix_ms = created_unix_ms; + next.validate()?; + let event = LifecycleEvent::transitioned(&next, prior_digest, transition, created_unix_ms)?; + Ok((next, event)) + } + + fn apply_transition(&mut self, transition: &LifecycleTransition) -> Result<(), LifecycleError> { + match transition { + LifecycleTransition::Cancel { reason } => { + if self.status == LifecycleStatus::Applying { + return Err(LifecycleError::InvalidTransition); + } + reason.validate()?; + self.finish_terminal( + LifecycleStatus::Cancelled, + LifecycleTerminalOutcome::Cancelled, + Some(reason.clone()), + ); + return Ok(()); + } + LifecycleTransition::Fail { reason } => { + if self.status == LifecycleStatus::Applying { + return Err(LifecycleError::InvalidTransition); + } + reason.validate()?; + self.finish_terminal( + LifecycleStatus::Failed, + LifecycleTerminalOutcome::Failed, + Some(reason.clone()), + ); + return Ok(()); + } + _ => {} + } + + match transition { + LifecycleTransition::CompleteExplore { worker, artifacts } => { + self.require_active_phase(LifecyclePhase::Explore)?; + self.accept_worker(LifecyclePhase::Explore, worker)?; + if artifacts.is_empty() || artifacts.len() > MAX_LIFECYCLE_ARTIFACT_REFS { + return Err(LifecycleError::MissingArtifact); + } + let mut ids = BTreeSet::new(); + for artifact in artifacts { + if artifact.source_snapshot != self.source_snapshot || !ids.insert(artifact.id) + { + return Err(LifecycleError::InvalidArtifact); + } + } + self.exploration_artifacts = artifacts.clone(); + self.phase = LifecyclePhase::Implement; + } + LifecycleTransition::CompleteImplement { worker, patch } => { + self.require_active_phase(LifecyclePhase::Implement)?; + self.accept_worker(LifecyclePhase::Implement, worker)?; + let expected_revision = if self.repair_consumed { 2 } else { 1 }; + if patch.revision != expected_revision { + return Err(LifecycleError::InvalidArtifact); + } + self.patch = Some(patch.clone()); + self.test_results.clear(); + self.review = None; + self.verification = None; + self.approval = None; + self.apply_id = None; + self.phase = LifecyclePhase::Test; + } + LifecycleTransition::CompleteTest { worker, results } => { + self.require_active_phase(LifecyclePhase::Test)?; + self.accept_worker(LifecyclePhase::Test, worker)?; + if results.len() != self.spec.test_plans.len() { + return Err(LifecycleError::MissingTestEvidence); + } + for (result, binding) in results.iter().zip(&self.spec.test_plans) { + result.validate()?; + if result.plan_digest != binding.plan_digest() + || result.certificate_digest != binding.certificate_digest + { + return Err(LifecycleError::InvalidTestEvidence); + } + } + self.test_results = results.clone(); + if results.iter().any(|result| !result.available || !result.executed) { + self.finish_terminal( + LifecycleStatus::Inconclusive, + LifecycleTerminalOutcome::Inconclusive, + Some(LifecycleReason::new( + "test_evidence_unavailable", + b"trusted test evidence unavailable", + )?), + ); + } else if results.iter().any(|result| !result.passed) { + self.finish_terminal( + LifecycleStatus::Failed, + LifecycleTerminalOutcome::Failed, + Some(LifecycleReason::new("test_failed", b"trusted test failed")?), + ); + } else { + self.phase = LifecyclePhase::Review; + } + } + LifecycleTransition::CompleteReview { worker, review } => { + self.require_active_phase(LifecyclePhase::Review)?; + self.accept_worker(LifecyclePhase::Review, worker)?; + let patch = self.patch.as_ref().ok_or(LifecycleError::MissingArtifact)?; + if !review.is_canonical() + || review.change_id != self.change_id + || review.patch_id != patch.patch_id + || review.reviewer_definition != worker.profile.definition_digest + { + return Err(LifecycleError::InvalidReview); + } + self.review = Some(review.clone()); + match review.verdict { + LifecycleReviewVerdict::Approved => self.phase = LifecyclePhase::Verify, + LifecycleReviewVerdict::Rejected => self.finish_terminal( + LifecycleStatus::Failed, + LifecycleTerminalOutcome::Failed, + Some(LifecycleReason::new("review_rejected", b"review rejected patch")?), + ), + LifecycleReviewVerdict::Inconclusive => self.finish_terminal( + LifecycleStatus::Inconclusive, + LifecycleTerminalOutcome::Inconclusive, + Some(LifecycleReason::new( + "review_inconclusive", + b"review was inconclusive", + )?), + ), + } + } + LifecycleTransition::CompleteVerify { worker, verification } => { + self.require_active_phase(LifecyclePhase::Verify)?; + self.accept_worker(LifecyclePhase::Verify, worker)?; + let patch = self.patch.as_ref().ok_or(LifecycleError::MissingArtifact)?; + if verification.patch_id != patch.patch_id + || verification.verdict == VerificationStatus::NotRequested + { + return Err(LifecycleError::InvalidVerification); + } + self.verification = Some(verification.clone()); + match verification.verdict { + VerificationStatus::Verified => { + self.phase = LifecyclePhase::Apply; + self.status = LifecycleStatus::AwaitingApproval; + } + VerificationStatus::Repairable => { + if self.repair_reserved || self.repair_consumed { + self.finish_terminal( + LifecycleStatus::Failed, + LifecycleTerminalOutcome::Failed, + Some(LifecycleReason::new( + "repair_limit_exhausted", + b"verification requested a second repair", + )?), + ); + } else { + self.status = LifecycleStatus::RepairReserved; + self.repair_reserved = true; + } + } + VerificationStatus::Rejected => self.finish_terminal( + LifecycleStatus::Failed, + LifecycleTerminalOutcome::Failed, + Some(LifecycleReason::new( + "verification_rejected", + b"verification rejected patch", + )?), + ), + VerificationStatus::Inconclusive => self.finish_terminal( + LifecycleStatus::Inconclusive, + LifecycleTerminalOutcome::Inconclusive, + Some(LifecycleReason::new( + "verification_inconclusive", + b"verification was inconclusive", + )?), + ), + VerificationStatus::NotRequested => unreachable!(), + } + } + LifecycleTransition::ConsumeRepair => { + if self.phase != LifecyclePhase::Verify + || self.status != LifecycleStatus::RepairReserved + || !self.repair_reserved + || self.repair_consumed + { + return Err(LifecycleError::InvalidTransition); + } + self.repair_reserved = false; + self.repair_consumed = true; + self.phase = LifecyclePhase::Implement; + self.status = LifecycleStatus::Active; + self.patch = None; + self.test_results.clear(); + self.review = None; + self.verification = None; + self.approval = None; + self.apply_id = None; + } + LifecycleTransition::ApproveApply { approval } => { + if self.phase != LifecyclePhase::Apply + || self.status != LifecycleStatus::AwaitingApproval + { + return Err(LifecycleError::InvalidTransition); + } + let patch = self.patch.as_ref().ok_or(LifecycleError::MissingArtifact)?; + let verification = + self.verification.as_ref().ok_or(LifecycleError::MissingArtifact)?; + if approval.approved_state_digest != self.state_digest() + || approval.patch_id != patch.patch_id + || approval.verification_id != verification.verification_id + || approval.decision_source != crate::ApprovalDecisionSource::WebUser + { + return Err(LifecycleError::StaleApproval); + } + self.approval = Some(approval.clone()); + self.status = LifecycleStatus::Approved; + } + LifecycleTransition::StartApply { apply_id } => { + if self.phase != LifecyclePhase::Apply || self.status != LifecycleStatus::Approved { + return Err(LifecycleError::InvalidTransition); + } + self.apply_id = Some(*apply_id); + self.status = LifecycleStatus::Applying; + } + LifecycleTransition::FinishApply { apply_id, status } => { + if self.phase != LifecyclePhase::Apply + || self.status != LifecycleStatus::Applying + || self.apply_id != Some(*apply_id) + || *status == ChangeApplyStatus::Applying + { + return Err(LifecycleError::InvalidTransition); + } + match status { + ChangeApplyStatus::Applied => self.finish_terminal( + LifecycleStatus::Completed, + LifecycleTerminalOutcome::Applied, + None, + ), + ChangeApplyStatus::RolledBack => self.finish_terminal( + LifecycleStatus::RolledBack, + LifecycleTerminalOutcome::RolledBack, + None, + ), + ChangeApplyStatus::RollbackFailed => self.finish_terminal( + LifecycleStatus::Failed, + LifecycleTerminalOutcome::RollbackFailed, + Some(LifecycleReason::new( + "rollback_failed", + b"active apply rollback failed", + )?), + ), + ChangeApplyStatus::RecoveryConflict => self.finish_terminal( + LifecycleStatus::Failed, + LifecycleTerminalOutcome::RecoveryConflict, + Some(LifecycleReason::new( + "recovery_conflict", + b"pending apply recovery conflicted", + )?), + ), + ChangeApplyStatus::Applying => unreachable!(), + } + } + LifecycleTransition::Cancel { .. } | LifecycleTransition::Fail { .. } => unreachable!(), + } + Ok(()) + } + + fn require_active_phase(&self, expected: LifecyclePhase) -> Result<(), LifecycleError> { + if self.phase != expected || self.status != LifecycleStatus::Active { + return Err(LifecycleError::InvalidTransition); + } + Ok(()) + } + + fn accept_worker( + &mut self, + phase: LifecyclePhase, + worker: &LifecycleWorkerCompletion, + ) -> Result<(), LifecycleError> { + if worker.worker_depth != 1 || worker.logical_worker_spawns != 1 { + return Err(LifecycleError::NestedWorker); + } + if worker.usage.worker_turns == 0 { + return Err(LifecycleError::InvalidWorkerCompletion); + } + if self.spec.profiles.for_phase(phase) != Some(&worker.profile) { + return Err(LifecycleError::ProfileMismatch); + } + let usage = self.usage.checked_add(&worker.usage)?; + if !usage.within(&self.spec.budget) { + return Err(LifecycleError::BudgetExceeded); + } + self.usage = usage; + Ok(()) + } + + fn finish_terminal( + &mut self, + status: LifecycleStatus, + outcome: LifecycleTerminalOutcome, + reason: Option, + ) { + self.repair_reserved = false; + self.status = status; + self.terminal_outcome = Some(outcome); + self.terminal_reason = reason; + } + + fn validate_artifact_chain(&self) -> Result<(), LifecycleError> { + if self.exploration_artifacts.len() > MAX_LIFECYCLE_ARTIFACT_REFS + || self.exploration_artifacts.iter().any(|artifact| { + artifact.source_snapshot != self.source_snapshot + || artifact.kind != LifecycleArtifactKind::Exploration + }) + || self.exploration_artifacts.windows(2).any(|pair| pair[0].id >= pair[1].id) + { + return Err(LifecycleError::InvalidArtifact); + } + if let Some(patch) = &self.patch { + let expected_revision = if self.repair_consumed { 2 } else { 1 }; + if self.exploration_artifacts.is_empty() || patch.revision != expected_revision { + return Err(LifecycleError::InvalidArtifact); + } + } else if !self.test_results.is_empty() + || self.review.is_some() + || self.verification.is_some() + || self.approval.is_some() + || self.apply_id.is_some() + { + return Err(LifecycleError::InvalidState); + } + if !self.test_results.is_empty() { + if self.test_results.len() != self.spec.test_plans.len() { + return Err(LifecycleError::MissingTestEvidence); + } + for (result, binding) in self.test_results.iter().zip(&self.spec.test_plans) { + result.validate()?; + if result.plan_digest != binding.plan_digest() + || result.certificate_digest != binding.certificate_digest + { + return Err(LifecycleError::InvalidTestEvidence); + } + } + } + if let Some(review) = &self.review { + let patch = self.patch.as_ref().ok_or(LifecycleError::MissingArtifact)?; + if self.test_results.iter().any(|result| !result.passed) + || self.test_results.len() != self.spec.test_plans.len() + || !review.is_canonical() + || review.change_id != self.change_id + || review.patch_id != patch.patch_id + || review.reviewer_definition != self.spec.profiles.review.definition_digest + { + return Err(LifecycleError::InvalidReview); + } + } + if let Some(verification) = &self.verification { + let patch = self.patch.as_ref().ok_or(LifecycleError::MissingArtifact)?; + if self.review.as_ref().map(|review| review.verdict) + != Some(LifecycleReviewVerdict::Approved) + || verification.patch_id != patch.patch_id + || verification.verdict == VerificationStatus::NotRequested + { + return Err(LifecycleError::InvalidVerification); + } + } + if let Some(approval) = &self.approval { + let patch = self.patch.as_ref().ok_or(LifecycleError::MissingArtifact)?; + let verification = self.verification.as_ref().ok_or(LifecycleError::MissingArtifact)?; + if !approval.is_canonical() + || verification.verdict != VerificationStatus::Verified + || approval.patch_id != patch.patch_id + || approval.verification_id != verification.verification_id + { + return Err(LifecycleError::StaleApproval); + } + } + if self.apply_id.is_some() && self.approval.is_none() { + return Err(LifecycleError::InvalidState); + } + Ok(()) + } + + fn validate_status_shape(&self) -> Result<(), LifecycleError> { + if self.status.terminal() { + let valid_terminal = matches!( + (self.status, self.terminal_outcome), + (LifecycleStatus::Completed, Some(LifecycleTerminalOutcome::Applied)) + | (LifecycleStatus::Failed, Some(LifecycleTerminalOutcome::Failed)) + | (LifecycleStatus::Failed, Some(LifecycleTerminalOutcome::RollbackFailed)) + | (LifecycleStatus::Failed, Some(LifecycleTerminalOutcome::RecoveryConflict)) + | (LifecycleStatus::Cancelled, Some(LifecycleTerminalOutcome::Cancelled)) + | (LifecycleStatus::Inconclusive, Some(LifecycleTerminalOutcome::Inconclusive)) + | (LifecycleStatus::RolledBack, Some(LifecycleTerminalOutcome::RolledBack)) + ); + if !valid_terminal + || self.repair_reserved + || matches!( + self.terminal_outcome, + Some( + LifecycleTerminalOutcome::Applied + | LifecycleTerminalOutcome::RolledBack + | LifecycleTerminalOutcome::RollbackFailed + | LifecycleTerminalOutcome::RecoveryConflict + ) + ) && (self.phase != LifecyclePhase::Apply || self.apply_id.is_none()) + { + return Err(LifecycleError::InvalidState); + } + return Ok(()); + } + let valid = match (self.phase, self.status) { + (LifecyclePhase::Explore, LifecycleStatus::Active) => { + self.exploration_artifacts.is_empty() + && self.patch.is_none() + && !self.repair_consumed + } + (LifecyclePhase::Implement, LifecycleStatus::Active) => { + !self.exploration_artifacts.is_empty() + && self.patch.is_none() + && self.test_results.is_empty() + && self.review.is_none() + && self.verification.is_none() + } + (LifecyclePhase::Test, LifecycleStatus::Active) => { + self.patch.is_some() + && self.test_results.is_empty() + && self.review.is_none() + && self.verification.is_none() + } + (LifecyclePhase::Review, LifecycleStatus::Active) => { + self.test_results.len() == self.spec.test_plans.len() + && self.test_results.iter().all(|result| result.passed) + && self.review.is_none() + && self.verification.is_none() + } + (LifecyclePhase::Verify, LifecycleStatus::Active) => { + self.review.as_ref().map(|review| review.verdict) + == Some(LifecycleReviewVerdict::Approved) + && self.verification.is_none() + && !self.repair_reserved + } + (LifecyclePhase::Verify, LifecycleStatus::RepairReserved) => { + self.verification.as_ref().map(|verification| verification.verdict) + == Some(VerificationStatus::Repairable) + && self.repair_reserved + && !self.repair_consumed + } + (LifecyclePhase::Apply, LifecycleStatus::AwaitingApproval) => { + self.verification.as_ref().map(|verification| verification.verdict) + == Some(VerificationStatus::Verified) + && self.approval.is_none() + && self.apply_id.is_none() + } + (LifecyclePhase::Apply, LifecycleStatus::Approved) => { + self.approval.is_some() && self.apply_id.is_none() + } + (LifecyclePhase::Apply, LifecycleStatus::Applying) => { + self.approval.is_some() && self.apply_id.is_some() + } + _ => false, + }; + if !valid || self.terminal_outcome.is_some() || self.terminal_reason.is_some() { + return Err(LifecycleError::InvalidState); + } + Ok(()) + } + + pub fn replay(events: &[LifecycleEvent]) -> Result { + let Some(first) = events.first() else { + return Err(LifecycleError::EventReplay); + }; + let LifecycleEventKind::Created { state } = &first.kind else { + return Err(LifecycleError::EventReplay); + }; + state.validate()?; + if *first != LifecycleEvent::created(state)? { + return Err(LifecycleError::EventReplay); + } + let mut current = state.as_ref().clone(); + for event in &events[1..] { + let LifecycleEventKind::Transitioned { transition } = &event.kind else { + return Err(LifecycleError::EventReplay); + }; + if event.sequence != current.generation + 1 + || event.prior_state_digest != Some(current.state_digest()) + { + return Err(LifecycleError::EventReplay); + } + let (next, reproduced) = + current.transition(transition.as_ref().clone(), event.created_unix_ms)?; + if reproduced != *event || next.state_digest() != event.resulting_state_digest { + return Err(LifecycleError::EventReplay); + } + current = next; + } + Ok(current) + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum LifecycleEventKind { + Created { state: Box }, + Transitioned { transition: Box }, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleEvent { + pub lifecycle_id: LifecycleId, + pub change_id: ChangeId, + pub sequence: u64, + pub phase: LifecyclePhase, + pub status: LifecycleStatus, + pub source_snapshot: Digest, + pub profile_revision_digest: Option, + pub patch_id: Option, + pub verification_id: Option, + pub prior_state_digest: Option, + pub resulting_state_digest: Digest, + pub kind: LifecycleEventKind, + pub created_unix_ms: u64, +} + +impl LifecycleEvent { + pub fn created(state: &DevelopmentLifecycle) -> Result { + let event = Self { + lifecycle_id: state.id, + change_id: state.change_id.clone(), + sequence: 0, + phase: state.phase, + status: state.status, + source_snapshot: state.source_snapshot, + profile_revision_digest: None, + patch_id: None, + verification_id: None, + prior_state_digest: None, + resulting_state_digest: state.state_digest(), + kind: LifecycleEventKind::Created { state: Box::new(state.clone()) }, + created_unix_ms: state.created_unix_ms, + }; + event.validate_size()?; + Ok(event) + } + + fn transitioned( + state: &DevelopmentLifecycle, + prior_state_digest: Digest, + transition: LifecycleTransition, + created_unix_ms: u64, + ) -> Result { + let profile_revision_digest = + transition.worker().map(|worker| worker.profile.definition_digest); + let event = Self { + lifecycle_id: state.id, + change_id: state.change_id.clone(), + sequence: state.generation, + phase: state.phase, + status: state.status, + source_snapshot: state.source_snapshot, + profile_revision_digest, + patch_id: state.patch.as_ref().map(|patch| patch.patch_id), + verification_id: state + .verification + .as_ref() + .map(|verification| verification.verification_id), + prior_state_digest: Some(prior_state_digest), + resulting_state_digest: state.state_digest(), + kind: LifecycleEventKind::Transitioned { transition: Box::new(transition) }, + created_unix_ms, + }; + event.validate_size()?; + Ok(event) + } + + fn validate_size(&self) -> Result<(), LifecycleError> { + if serde_json::to_vec(self).map_err(|_| LifecycleError::EventReplay)?.len() + > MAX_LIFECYCLE_EVENT_BYTES + { + return Err(LifecycleError::EventTooLarge); + } + Ok(()) + } +} + +#[derive(Clone, Debug, Error, Eq, PartialEq)] +pub enum LifecycleError { + #[error("lifecycle budget is invalid")] + InvalidBudget, + #[error("lifecycle budget arithmetic overflowed")] + BudgetOverflow, + #[error("lifecycle budget was exceeded")] + BudgetExceeded, + #[error("worker depth or spawn count exceeds the depth-one contract")] + NestedWorker, + #[error("worker profile does not match the frozen phase binding")] + ProfileMismatch, + #[error("worker completion accounting is invalid")] + InvalidWorkerCompletion, + #[error("lifecycle transition is illegal from the current phase/state")] + InvalidTransition, + #[error("lifecycle is terminal")] + Terminal, + #[error("lifecycle artifact is missing")] + MissingArtifact, + #[error("lifecycle artifact is invalid")] + InvalidArtifact, + #[error("lifecycle test plan is invalid")] + InvalidTestPlan, + #[error("lifecycle test evidence is missing")] + MissingTestEvidence, + #[error("lifecycle test evidence is invalid")] + InvalidTestEvidence, + #[error("review artifact is invalid")] + InvalidReview, + #[error("verification artifact reference is invalid")] + InvalidVerification, + #[error("apply approval is stale or references different artifacts")] + StaleApproval, + #[error("lifecycle reason code is invalid")] + InvalidReason, + #[error("lifecycle timestamp regressed")] + TimeRegression, + #[error("lifecycle generation overflowed")] + GenerationOverflow, + #[error("lifecycle state is invalid")] + InvalidState, + #[error("lifecycle state exceeds the byte bound")] + StateTooLarge, + #[error("lifecycle event exceeds the byte bound")] + EventTooLarge, + #[error("lifecycle event replay failed")] + EventReplay, +} + +fn bounded_identifier(value: &str, max_bytes: usize) -> bool { + !value.is_empty() + && value.len() <= max_bytes + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':')) +} + +fn safe_relative_directory(value: &str) -> bool { + let path = std::path::Path::new(value); + path.is_relative() + && !value.is_empty() + && value.len() <= 512 + && path.components().all(|component| { + matches!(component, std::path::Component::Normal(_) | std::path::Component::CurDir) + }) +} + +fn test_plan_contains_absolute_path(plan: &crate::TestPlan) -> bool { + plan.argv.iter().any(|argument| { + std::path::Path::new(argument).components().any(|component| { + matches!(component, std::path::Component::RootDir | std::path::Component::Prefix(_)) + }) + }) +} + +#[cfg(test)] +#[path = "lifecycle/tests.rs"] +mod tests; diff --git a/crates/needle-core/src/lifecycle/model.rs b/crates/needle-core/src/lifecycle/model.rs new file mode 100644 index 0000000..f2d82bb --- /dev/null +++ b/crates/needle-core/src/lifecycle/model.rs @@ -0,0 +1,606 @@ +use super::{ + LifecycleError, bounded_identifier, safe_relative_directory, test_plan_contains_absolute_path, +}; +use crate::{ + AcceptanceStatus, ApprovalDecisionSource, CanonicalHasher, ChangeApplyId, ChangeApplyStatus, + ChangeId, Digest, PatchId, RoleProfileProvenance, TestPlan, VerificationArtifactId, + VerificationStatus, +}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeSet; +pub const MAX_LIFECYCLE_ARTIFACT_REFS: usize = 32; +pub const MAX_LIFECYCLE_TEST_PLANS: usize = crate::MAX_VERIFIER_TEST_PLANS; +pub const MAX_LIFECYCLE_REVIEW_FINDINGS: usize = 32; +pub const MAX_LIFECYCLE_REASON_CODE_BYTES: usize = 64; +pub const MAX_LIFECYCLE_EVIDENCE_ID_BYTES: usize = 128; +pub const MAX_LIFECYCLE_STATE_BYTES: usize = 64 * 1024; +pub const MAX_LIFECYCLE_EVENT_BYTES: usize = 64 * 1024; +pub const MAX_LIFECYCLE_EVENTS: usize = 16; + +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] +#[serde(transparent)] +pub struct LifecycleId(pub Digest); + +impl LifecycleId { + pub fn compute( + change_id: &ChangeId, + source_snapshot: Digest, + profiles: &LifecycleWorkerProfiles, + ) -> Self { + let mut hasher = CanonicalHasher::new(b"needle-development-lifecycle"); + hasher.field_str(change_id.as_str()); + hasher.field_digest(source_snapshot); + profiles.hash_into(&mut hasher); + Self(hasher.finish()) + } +} + +impl std::fmt::Display for LifecycleId { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(formatter) + } +} + +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LifecyclePhase { + Explore, + Implement, + Test, + Review, + Verify, + Apply, +} + +impl LifecyclePhase { + pub const ALL: [Self; 6] = + [Self::Explore, Self::Implement, Self::Test, Self::Review, Self::Verify, Self::Apply]; + + pub const fn as_str(self) -> &'static str { + match self { + Self::Explore => "explore", + Self::Implement => "implement", + Self::Test => "test", + Self::Review => "review", + Self::Verify => "verify", + Self::Apply => "apply", + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LifecycleStatus { + Active, + RepairReserved, + AwaitingApproval, + Approved, + Applying, + Completed, + Failed, + Cancelled, + Inconclusive, + RolledBack, +} + +impl LifecycleStatus { + pub const fn terminal(self) -> bool { + matches!( + self, + Self::Completed + | Self::Failed + | Self::Cancelled + | Self::Inconclusive + | Self::RolledBack + ) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LifecycleTerminalOutcome { + Applied, + Failed, + Cancelled, + Inconclusive, + RolledBack, + RollbackFailed, + RecoveryConflict, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleReason { + pub code: String, + pub detail_digest: Digest, +} + +impl LifecycleReason { + pub fn new(code: impl Into, detail: impl AsRef<[u8]>) -> Result { + let reason = Self { code: code.into(), detail_digest: Digest::blake3(detail) }; + reason.validate()?; + Ok(reason) + } + + pub(super) fn validate(&self) -> Result<(), LifecycleError> { + if !bounded_identifier(&self.code, MAX_LIFECYCLE_REASON_CODE_BYTES) { + return Err(LifecycleError::InvalidReason); + } + Ok(()) + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleBudget { + pub max_worker_turns: u32, + pub max_output_tokens: u64, + pub max_cost_microusd: u64, + pub max_concurrent_workers: u8, +} + +impl LifecycleBudget { + fn validate(&self) -> Result<(), LifecycleError> { + if self.max_worker_turns == 0 + || self.max_output_tokens == 0 + || self.max_cost_microusd == 0 + || self.max_concurrent_workers != 1 + { + return Err(LifecycleError::InvalidBudget); + } + Ok(()) + } +} + +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleUsage { + pub worker_turns: u32, + pub output_tokens: u64, + pub cost_microusd: u64, +} + +impl LifecycleUsage { + pub(super) fn checked_add(&self, delta: &Self) -> Result { + Ok(Self { + worker_turns: self + .worker_turns + .checked_add(delta.worker_turns) + .ok_or(LifecycleError::BudgetOverflow)?, + output_tokens: self + .output_tokens + .checked_add(delta.output_tokens) + .ok_or(LifecycleError::BudgetOverflow)?, + cost_microusd: self + .cost_microusd + .checked_add(delta.cost_microusd) + .ok_or(LifecycleError::BudgetOverflow)?, + }) + } + + pub(super) fn within(&self, budget: &LifecycleBudget) -> bool { + self.worker_turns <= budget.max_worker_turns + && self.output_tokens <= budget.max_output_tokens + && self.cost_microusd <= budget.max_cost_microusd + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleWorkerProfiles { + pub explore: RoleProfileProvenance, + pub implement: RoleProfileProvenance, + pub test: RoleProfileProvenance, + pub review: RoleProfileProvenance, + pub verify: RoleProfileProvenance, +} + +impl LifecycleWorkerProfiles { + pub fn for_phase(&self, phase: LifecyclePhase) -> Option<&RoleProfileProvenance> { + match phase { + LifecyclePhase::Explore => Some(&self.explore), + LifecyclePhase::Implement => Some(&self.implement), + LifecyclePhase::Test => Some(&self.test), + LifecyclePhase::Review => Some(&self.review), + LifecyclePhase::Verify => Some(&self.verify), + LifecyclePhase::Apply => None, + } + } + + pub(super) fn validate(&self) -> Result<(), LifecycleError> { + for phase in LifecyclePhase::ALL.into_iter().take(5) { + self.for_phase(phase) + .ok_or(LifecycleError::ProfileMismatch)? + .validate() + .map_err(|_| LifecycleError::ProfileMismatch)?; + } + Ok(()) + } + + fn hash_into(&self, hasher: &mut CanonicalHasher) { + for phase in LifecyclePhase::ALL.into_iter().take(5) { + let profile = self.for_phase(phase).expect("worker phase has a profile"); + hasher.field_str(phase.as_str()); + hasher.field_str(profile.profile_id.as_str()); + hasher.field_bytes(&profile.revision.to_le_bytes()); + hasher.field_digest(profile.definition_digest); + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleTestPlanBinding { + pub plan: TestPlan, + pub certificate_digest: Digest, +} + +impl LifecycleTestPlanBinding { + pub fn plan_digest(&self) -> Digest { + self.plan.identity_digest() + } + + fn validate(&self) -> Result<(), LifecycleError> { + if !self.plan.requires_approval + || self.plan.execution_evidence_id.is_some() + || self.plan.test_command().is_err() + || !safe_relative_directory(&self.plan.cwd_relative) + || test_plan_contains_absolute_path(&self.plan) + { + return Err(LifecycleError::InvalidTestPlan); + } + Ok(()) + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleSpec { + pub worker_depth_limit: u8, + pub profiles: LifecycleWorkerProfiles, + pub budget: LifecycleBudget, + pub test_plans: Vec, +} + +impl LifecycleSpec { + pub fn validate(&self) -> Result<(), LifecycleError> { + if self.worker_depth_limit != 1 { + return Err(LifecycleError::NestedWorker); + } + self.profiles.validate()?; + self.budget.validate()?; + if self.test_plans.is_empty() || self.test_plans.len() > MAX_LIFECYCLE_TEST_PLANS { + return Err(LifecycleError::InvalidTestPlan); + } + let mut prior = None; + for binding in &self.test_plans { + binding.validate()?; + let digest = binding.plan_digest(); + if prior.is_some_and(|value| value >= digest) { + return Err(LifecycleError::InvalidTestPlan); + } + prior = Some(digest); + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LifecycleArtifactKind { + Exploration, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleArtifactRef { + pub kind: LifecycleArtifactKind, + pub id: Digest, + pub source_snapshot: Digest, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecyclePatchRef { + pub patch_id: PatchId, + pub revision: u32, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleTestResult { + pub plan_digest: Digest, + pub certificate_digest: Digest, + pub available: bool, + pub executed: bool, + pub passed: bool, + pub evidence_id: Option, + pub failure_code: Option, +} + +impl LifecycleTestResult { + pub(super) fn validate(&self) -> Result<(), LifecycleError> { + if self + .evidence_id + .as_deref() + .is_some_and(|value| !bounded_identifier(value, MAX_LIFECYCLE_EVIDENCE_ID_BYTES)) + || self + .failure_code + .as_deref() + .is_some_and(|value| !bounded_identifier(value, MAX_LIFECYCLE_REASON_CODE_BYTES)) + { + return Err(LifecycleError::InvalidTestEvidence); + } + if self.passed { + if !self.available + || !self.executed + || self.evidence_id.is_none() + || self.failure_code.is_some() + { + return Err(LifecycleError::InvalidTestEvidence); + } + } else if self.failure_code.is_none() { + return Err(LifecycleError::InvalidTestEvidence); + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LifecycleReviewVerdict { + Approved, + Rejected, + Inconclusive, +} + +/// Redacted review coverage. Persisted lifecycle events contain only stable +/// digests, never criterion prose, paths, or a reviewer transcript. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleAcceptanceReview { + pub criterion_digest: Digest, + pub status: AcceptanceStatus, + pub evidence_digest: Digest, +} + +impl LifecycleAcceptanceReview { + pub fn new( + criterion: impl AsRef<[u8]>, + status: AcceptanceStatus, + evidence: impl AsRef<[u8]>, + ) -> Self { + Self { + criterion_digest: Digest::blake3(criterion), + status, + evidence_digest: Digest::blake3(evidence), + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ReviewArtifact { + pub id: Digest, + pub change_id: ChangeId, + pub patch_id: PatchId, + pub verdict: LifecycleReviewVerdict, + pub acceptance_coverage: Vec, + pub findings: Vec, + pub reviewer_definition: Digest, + pub created_unix_ms: u64, +} + +impl ReviewArtifact { + pub fn new( + change_id: ChangeId, + patch_id: PatchId, + verdict: LifecycleReviewVerdict, + mut acceptance_coverage: Vec, + mut findings: Vec, + reviewer_definition: Digest, + created_unix_ms: u64, + ) -> Result { + acceptance_coverage.sort_by_key(|coverage| coverage.criterion_digest); + findings.sort_by(|left, right| { + left.code.cmp(&right.code).then_with(|| left.detail_digest.cmp(&right.detail_digest)) + }); + let mut artifact = Self { + id: Digest::blake3(b"pending-review-artifact"), + change_id, + patch_id, + verdict, + acceptance_coverage, + findings, + reviewer_definition, + created_unix_ms, + }; + artifact.validate_material()?; + artifact.id = artifact.compute_id(); + Ok(artifact) + } + + pub fn is_canonical(&self) -> bool { + self.validate_material().is_ok() && self.id == self.compute_id() + } + + fn validate_material(&self) -> Result<(), LifecycleError> { + if self.acceptance_coverage.is_empty() + || self.acceptance_coverage.len() > MAX_LIFECYCLE_ARTIFACT_REFS + || self.findings.len() > MAX_LIFECYCLE_REVIEW_FINDINGS + { + return Err(LifecycleError::InvalidReview); + } + let mut criteria = BTreeSet::new(); + if self + .acceptance_coverage + .iter() + .any(|coverage| !criteria.insert(coverage.criterion_digest)) + || self + .acceptance_coverage + .windows(2) + .any(|pair| pair[0].criterion_digest >= pair[1].criterion_digest) + || (self.verdict == LifecycleReviewVerdict::Approved + && self + .acceptance_coverage + .iter() + .any(|coverage| coverage.status != AcceptanceStatus::Addressed)) + { + return Err(LifecycleError::InvalidReview); + } + let mut finding_ids = BTreeSet::new(); + for finding in &self.findings { + finding.validate()?; + if !finding_ids.insert((&finding.code, finding.detail_digest)) { + return Err(LifecycleError::InvalidReview); + } + } + if self.findings.windows(2).any(|pair| { + (&pair[0].code, pair[0].detail_digest) >= (&pair[1].code, pair[1].detail_digest) + }) { + return Err(LifecycleError::InvalidReview); + } + Ok(()) + } + + fn compute_id(&self) -> Digest { + #[derive(Serialize)] + struct Material<'a> { + change_id: &'a ChangeId, + patch_id: PatchId, + verdict: LifecycleReviewVerdict, + acceptance_coverage: &'a [LifecycleAcceptanceReview], + findings: &'a [LifecycleReason], + reviewer_definition: Digest, + created_unix_ms: u64, + } + let material = Material { + change_id: &self.change_id, + patch_id: self.patch_id, + verdict: self.verdict, + acceptance_coverage: &self.acceptance_coverage, + findings: &self.findings, + reviewer_definition: self.reviewer_definition, + created_unix_ms: self.created_unix_ms, + }; + Digest::blake3(serde_json::to_vec(&material).unwrap_or_default()) + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleVerificationRef { + pub verification_id: VerificationArtifactId, + pub patch_id: PatchId, + pub verdict: VerificationStatus, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleApplyApproval { + pub id: Digest, + pub approved_state_digest: Digest, + pub patch_id: PatchId, + pub verification_id: VerificationArtifactId, + pub decision_source: ApprovalDecisionSource, + pub decided_unix_ms: u64, +} + +impl LifecycleApplyApproval { + pub fn new( + approved_state_digest: Digest, + patch_id: PatchId, + verification_id: VerificationArtifactId, + decision_source: ApprovalDecisionSource, + decided_unix_ms: u64, + ) -> Self { + let id = Self::compute_id( + approved_state_digest, + patch_id, + verification_id, + decision_source, + decided_unix_ms, + ); + Self { + id, + approved_state_digest, + patch_id, + verification_id, + decision_source, + decided_unix_ms, + } + } + + pub fn is_canonical(&self) -> bool { + self.id + == Self::compute_id( + self.approved_state_digest, + self.patch_id, + self.verification_id, + self.decision_source, + self.decided_unix_ms, + ) + } + + fn compute_id( + approved_state_digest: Digest, + patch_id: PatchId, + verification_id: VerificationArtifactId, + decision_source: ApprovalDecisionSource, + decided_unix_ms: u64, + ) -> Digest { + let mut hasher = CanonicalHasher::new(b"needle-lifecycle-user-apply-approval"); + hasher.field_digest(approved_state_digest); + hasher.field_digest(patch_id.0); + hasher.field_digest(verification_id.0); + hasher.field_u8(match decision_source { + ApprovalDecisionSource::AutoPolicy => 0, + ApprovalDecisionSource::WebUser => 1, + ApprovalDecisionSource::Timeout => 2, + ApprovalDecisionSource::Runtime => 3, + }); + hasher.field_bytes(&decided_unix_ms.to_le_bytes()); + hasher.finish() + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleWorkerCompletion { + pub profile: RoleProfileProvenance, + pub worker_depth: u8, + /// Total logical workers represented by this completion. The single + /// parent-launched worker counts as one; a value above one proves nested + /// or duplicate worker creation and is rejected. + pub logical_worker_spawns: u8, + pub usage: LifecycleUsage, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum LifecycleTransition { + CompleteExplore { worker: LifecycleWorkerCompletion, artifacts: Vec }, + CompleteImplement { worker: LifecycleWorkerCompletion, patch: LifecyclePatchRef }, + CompleteTest { worker: LifecycleWorkerCompletion, results: Vec }, + CompleteReview { worker: LifecycleWorkerCompletion, review: ReviewArtifact }, + CompleteVerify { worker: LifecycleWorkerCompletion, verification: LifecycleVerificationRef }, + ConsumeRepair, + ApproveApply { approval: LifecycleApplyApproval }, + StartApply { apply_id: ChangeApplyId }, + FinishApply { apply_id: ChangeApplyId, status: ChangeApplyStatus }, + Cancel { reason: LifecycleReason }, + Fail { reason: LifecycleReason }, +} + +impl LifecycleTransition { + pub fn worker(&self) -> Option<&LifecycleWorkerCompletion> { + match self { + Self::CompleteExplore { worker, .. } + | Self::CompleteImplement { worker, .. } + | Self::CompleteTest { worker, .. } + | Self::CompleteReview { worker, .. } + | Self::CompleteVerify { worker, .. } => Some(worker), + _ => None, + } + } +} diff --git a/crates/needle-core/src/lifecycle/tests.rs b/crates/needle-core/src/lifecycle/tests.rs new file mode 100644 index 0000000..facde93 --- /dev/null +++ b/crates/needle-core/src/lifecycle/tests.rs @@ -0,0 +1,714 @@ +use super::*; +use crate::{ + AcceptanceCoverage, AcceptanceStatus, RoleProfileId, RoleProfileProvenance, TestPlan, + VerificationArtifact, VerificationPlanResult, +}; + +fn profile(name: &str) -> RoleProfileProvenance { + RoleProfileProvenance::new( + RoleProfileId::new(name).unwrap(), + 1, + Digest::blake3(name.as_bytes()), + ) + .unwrap() +} + +fn spec() -> LifecycleSpec { + LifecycleSpec { + worker_depth_limit: 1, + profiles: LifecycleWorkerProfiles { + explore: profile("lifecycle.explorer"), + implement: profile("lifecycle.implementer"), + test: profile("lifecycle.test-runner"), + review: profile("lifecycle.reviewer"), + verify: profile("lifecycle.verifier"), + }, + budget: LifecycleBudget { + max_worker_turns: 10, + max_output_tokens: 10_000, + max_cost_microusd: 100_000, + max_concurrent_workers: 1, + }, + test_plans: vec![LifecycleTestPlanBinding { + plan: TestPlan { + runner: "cargo".to_owned(), + argv: vec![ + "cargo".to_owned(), + "test".to_owned(), + "--offline".to_owned(), + "focused".to_owned(), + "--".to_owned(), + "--exact".to_owned(), + ], + cwd_relative: ".".to_owned(), + test_identifier: "focused".to_owned(), + requires_approval: true, + execution_evidence_id: None, + }, + certificate_digest: Digest::blake3(b"test-certificate"), + }], + } +} + +fn worker(profile: &RoleProfileProvenance) -> LifecycleWorkerCompletion { + LifecycleWorkerCompletion { + profile: profile.clone(), + worker_depth: 1, + logical_worker_spawns: 1, + usage: LifecycleUsage { worker_turns: 1, output_tokens: 10, cost_microusd: 10 }, + } +} + +fn lifecycle() -> DevelopmentLifecycle { + DevelopmentLifecycle::new( + ChangeId::from_digest(Digest::blake3(b"change")), + Digest::blake3(b"source"), + spec(), + 1, + ) + .unwrap() +} + +fn coverage() -> Vec { + vec![LifecycleAcceptanceReview::new( + b"focused behavior works", + AcceptanceStatus::Addressed, + b"bounded evidence", + )] +} + +fn verification_coverage() -> Vec { + vec![AcceptanceCoverage { + criterion: "focused behavior works".to_owned(), + status: AcceptanceStatus::Addressed, + evidence: "bounded evidence".to_owned(), + }] +} + +fn advance_to_verify(mut state: DevelopmentLifecycle) -> DevelopmentLifecycle { + let source_snapshot = state.source_snapshot; + (state, _) = state + .transition( + LifecycleTransition::CompleteExplore { + worker: worker(&state.spec.profiles.explore), + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"exploration"), + source_snapshot, + }], + }, + 2, + ) + .unwrap(); + let patch_material: &[u8] = if state.repair_consumed { b"repaired-patch" } else { b"patch" }; + let patch_id = PatchId(Digest::blake3(patch_material)); + (state, _) = state + .transition( + LifecycleTransition::CompleteImplement { + worker: worker(&state.spec.profiles.implement), + patch: LifecyclePatchRef { + patch_id, + revision: if state.repair_consumed { 2 } else { 1 }, + }, + }, + 3, + ) + .unwrap(); + let binding = state.spec.test_plans[0].clone(); + (state, _) = state + .transition( + LifecycleTransition::CompleteTest { + worker: worker(&state.spec.profiles.test), + results: vec![LifecycleTestResult { + plan_digest: binding.plan_digest(), + certificate_digest: binding.certificate_digest, + available: true, + executed: true, + passed: true, + evidence_id: Some("evidence:focused".to_owned()), + failure_code: None, + }], + }, + 4, + ) + .unwrap(); + let review = ReviewArtifact::new( + state.change_id.clone(), + patch_id, + LifecycleReviewVerdict::Approved, + coverage(), + Vec::new(), + state.spec.profiles.review.definition_digest, + 5, + ) + .unwrap(); + state + .transition( + LifecycleTransition::CompleteReview { + worker: worker(&state.spec.profiles.review), + review, + }, + 5, + ) + .unwrap() + .0 +} + +fn verify_transition( + state: &DevelopmentLifecycle, + verdict: VerificationStatus, + identity: &[u8], +) -> LifecycleTransition { + LifecycleTransition::CompleteVerify { + worker: worker(&state.spec.profiles.verify), + verification: LifecycleVerificationRef { + verification_id: VerificationArtifactId(Digest::blake3(identity)), + patch_id: state.patch.as_ref().unwrap().patch_id, + verdict, + }, + } +} + +fn candidate_worker_transition( + state: &DevelopmentLifecycle, + phase: LifecyclePhase, +) -> LifecycleTransition { + let patch_id = state + .patch + .as_ref() + .map(|patch| patch.patch_id) + .unwrap_or(PatchId(Digest::blake3(b"candidate-patch"))); + match phase { + LifecyclePhase::Explore => LifecycleTransition::CompleteExplore { + worker: worker(&state.spec.profiles.explore), + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"candidate-exploration"), + source_snapshot: state.source_snapshot, + }], + }, + LifecyclePhase::Implement => LifecycleTransition::CompleteImplement { + worker: worker(&state.spec.profiles.implement), + patch: LifecyclePatchRef { + patch_id, + revision: if state.repair_consumed { 2 } else { 1 }, + }, + }, + LifecyclePhase::Test => { + let binding = &state.spec.test_plans[0]; + LifecycleTransition::CompleteTest { + worker: worker(&state.spec.profiles.test), + results: vec![LifecycleTestResult { + plan_digest: binding.plan_digest(), + certificate_digest: binding.certificate_digest, + available: true, + executed: true, + passed: true, + evidence_id: Some("evidence:candidate".to_owned()), + failure_code: None, + }], + } + } + LifecyclePhase::Review => LifecycleTransition::CompleteReview { + worker: worker(&state.spec.profiles.review), + review: ReviewArtifact::new( + state.change_id.clone(), + patch_id, + LifecycleReviewVerdict::Approved, + coverage(), + Vec::new(), + state.spec.profiles.review.definition_digest, + state.updated_unix_ms, + ) + .unwrap(), + }, + LifecyclePhase::Verify => LifecycleTransition::CompleteVerify { + worker: worker(&state.spec.profiles.verify), + verification: LifecycleVerificationRef { + verification_id: VerificationArtifactId(Digest::blake3(b"candidate-verify")), + patch_id, + verdict: VerificationStatus::Verified, + }, + }, + LifecyclePhase::Apply => unreachable!("apply has no worker completion"), + } +} + +#[test] +fn exact_phase_order_and_distinct_review_verifier_reach_approval() { + let mut state = lifecycle(); + let mut events = vec![LifecycleEvent::created(&state).unwrap()]; + let source = state.source_snapshot; + let explore = LifecycleTransition::CompleteExplore { + worker: worker(&state.spec.profiles.explore), + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"exploration"), + source_snapshot: source, + }], + }; + (state, _) = state.transition(explore, 2).unwrap(); + let patch_id = PatchId(Digest::blake3(b"patch")); + (state, _) = state + .transition( + LifecycleTransition::CompleteImplement { + worker: worker(&state.spec.profiles.implement), + patch: LifecyclePatchRef { patch_id, revision: 1 }, + }, + 3, + ) + .unwrap(); + let binding = &state.spec.test_plans[0]; + (state, _) = state + .transition( + LifecycleTransition::CompleteTest { + worker: worker(&state.spec.profiles.test), + results: vec![LifecycleTestResult { + plan_digest: binding.plan_digest(), + certificate_digest: binding.certificate_digest, + available: true, + executed: true, + passed: true, + evidence_id: Some("evidence:focused".to_owned()), + failure_code: None, + }], + }, + 4, + ) + .unwrap(); + let review = ReviewArtifact::new( + state.change_id.clone(), + patch_id, + LifecycleReviewVerdict::Approved, + coverage(), + Vec::new(), + state.spec.profiles.review.definition_digest, + 5, + ) + .unwrap(); + (state, _) = state + .transition( + LifecycleTransition::CompleteReview { + worker: worker(&state.spec.profiles.review), + review, + }, + 5, + ) + .unwrap(); + let verification_id = VerificationArtifactId(Digest::blake3(b"verification")); + let (next, event) = state + .transition( + LifecycleTransition::CompleteVerify { + worker: worker(&state.spec.profiles.verify), + verification: LifecycleVerificationRef { + verification_id, + patch_id, + verdict: VerificationStatus::Verified, + }, + }, + 6, + ) + .unwrap(); + state = next; + events.push(event); + assert_eq!(state.phase, LifecyclePhase::Apply); + assert_eq!(state.status, LifecycleStatus::AwaitingApproval); + assert_ne!(state.review.as_ref().unwrap().id, verification_id.0); + let approval = LifecycleApplyApproval::new( + state.state_digest(), + patch_id, + verification_id, + crate::ApprovalDecisionSource::WebUser, + 7, + ); + (state, _) = state.transition(LifecycleTransition::ApproveApply { approval }, 7).unwrap(); + assert_eq!(state.status, LifecycleStatus::Approved); + assert_eq!(events[0].sequence, 0); +} + +#[test] +fn skips_duplicates_cycles_and_nested_workers_leave_state_unchanged() { + let state = lifecycle(); + let original = state.clone(); + let patch = LifecycleTransition::CompleteImplement { + worker: worker(&state.spec.profiles.implement), + patch: LifecyclePatchRef { patch_id: PatchId(Digest::blake3(b"patch")), revision: 1 }, + }; + assert_eq!(state.transition(patch, 2), Err(LifecycleError::InvalidTransition)); + let mut nested = worker(&state.spec.profiles.explore); + nested.logical_worker_spawns = 2; + assert_eq!( + state.transition( + LifecycleTransition::CompleteExplore { + worker: nested, + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"artifact"), + source_snapshot: state.source_snapshot, + }], + }, + 2, + ), + Err(LifecycleError::NestedWorker) + ); + assert_eq!(state, original); + + let advanced = state + .transition( + LifecycleTransition::CompleteExplore { + worker: worker(&state.spec.profiles.explore), + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"artifact"), + source_snapshot: state.source_snapshot, + }], + }, + 2, + ) + .unwrap() + .0; + assert_eq!( + advanced.transition( + LifecycleTransition::CompleteImplement { + worker: worker(&advanced.spec.profiles.implement), + patch: LifecyclePatchRef { + patch_id: PatchId(Digest::blake3(b"stale-revision")), + revision: 2, + }, + }, + 3, + ), + Err(LifecycleError::InvalidArtifact) + ); +} + +#[test] +fn transition_matrix_rejects_every_out_of_phase_worker_completion() { + let mut states = vec![lifecycle()]; + let mut state = states[0].clone(); + for (phase, timestamp) in [ + (LifecyclePhase::Explore, 2), + (LifecyclePhase::Implement, 3), + (LifecyclePhase::Test, 4), + (LifecyclePhase::Review, 5), + ] { + state = state.transition(candidate_worker_transition(&state, phase), timestamp).unwrap().0; + states.push(state.clone()); + } + assert_eq!( + states.iter().map(|state| state.phase).collect::>(), + [ + LifecyclePhase::Explore, + LifecyclePhase::Implement, + LifecyclePhase::Test, + LifecyclePhase::Review, + LifecyclePhase::Verify, + ] + ); + for state in states { + for phase in LifecyclePhase::ALL.into_iter().take(5) { + if phase == state.phase { + continue; + } + let original = state.clone(); + assert_eq!( + state.transition( + candidate_worker_transition(&state, phase), + state.updated_unix_ms + 1, + ), + Err(LifecycleError::InvalidTransition), + "{} must reject {} completion", + state.phase.as_str(), + phase.as_str(), + ); + assert_eq!(state, original); + } + } +} + +#[test] +fn unavailable_test_evidence_is_terminal_inconclusive() { + let mut state = lifecycle(); + (state, _) = state + .transition( + LifecycleTransition::CompleteExplore { + worker: worker(&state.spec.profiles.explore), + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"artifact"), + source_snapshot: state.source_snapshot, + }], + }, + 2, + ) + .unwrap(); + (state, _) = state + .transition( + LifecycleTransition::CompleteImplement { + worker: worker(&state.spec.profiles.implement), + patch: LifecyclePatchRef { + patch_id: PatchId(Digest::blake3(b"patch")), + revision: 1, + }, + }, + 3, + ) + .unwrap(); + let binding = &state.spec.test_plans[0]; + (state, _) = state + .transition( + LifecycleTransition::CompleteTest { + worker: worker(&state.spec.profiles.test), + results: vec![LifecycleTestResult { + plan_digest: binding.plan_digest(), + certificate_digest: binding.certificate_digest, + available: false, + executed: false, + passed: false, + evidence_id: None, + failure_code: Some("unavailable".to_owned()), + }], + }, + 4, + ) + .unwrap(); + assert_eq!(state.status, LifecycleStatus::Inconclusive); + assert_eq!(state.terminal_outcome, Some(LifecycleTerminalOutcome::Inconclusive)); +} + +#[test] +fn replay_rejects_tampering() { + let state = lifecycle(); + let mut events = vec![LifecycleEvent::created(&state).unwrap()]; + let (next, event) = state + .transition( + LifecycleTransition::Cancel { + reason: LifecycleReason::new("cancelled", b"user cancelled").unwrap(), + }, + 2, + ) + .unwrap(); + events.push(event); + assert_eq!(DevelopmentLifecycle::replay(&events).unwrap(), next); + let mut envelope_tamper = events.clone(); + envelope_tamper[0].phase = LifecyclePhase::Implement; + assert_eq!(DevelopmentLifecycle::replay(&envelope_tamper), Err(LifecycleError::EventReplay)); + events[1].resulting_state_digest = Digest::blake3(b"tampered"); + assert_eq!(DevelopmentLifecycle::replay(&events), Err(LifecycleError::EventReplay)); +} + +#[test] +fn budget_profile_and_zero_usage_fail_without_mutation() { + let state = lifecycle(); + let original = state.clone(); + let artifacts = vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"exploration"), + source_snapshot: state.source_snapshot, + }]; + let mut over_budget = worker(&state.spec.profiles.explore); + over_budget.usage.worker_turns = state.spec.budget.max_worker_turns + 1; + assert_eq!( + state.transition( + LifecycleTransition::CompleteExplore { + worker: over_budget, + artifacts: artifacts.clone(), + }, + 2, + ), + Err(LifecycleError::BudgetExceeded) + ); + let mut wrong_profile = worker(&state.spec.profiles.explore); + wrong_profile.profile = state.spec.profiles.review.clone(); + assert_eq!( + state.transition( + LifecycleTransition::CompleteExplore { + worker: wrong_profile, + artifacts: artifacts.clone(), + }, + 2, + ), + Err(LifecycleError::ProfileMismatch) + ); + let mut zero_usage = worker(&state.spec.profiles.explore); + zero_usage.usage.worker_turns = 0; + assert_eq!( + state + .transition(LifecycleTransition::CompleteExplore { worker: zero_usage, artifacts }, 2,), + Err(LifecycleError::InvalidWorkerCompletion) + ); + assert_eq!(state, original); +} + +#[test] +fn one_repair_is_consumed_and_second_request_fails_terminally() { + let mut state = advance_to_verify(lifecycle()); + (state, _) = state + .transition(verify_transition(&state, VerificationStatus::Repairable, b"repair-1"), 6) + .unwrap(); + assert_eq!(state.status, LifecycleStatus::RepairReserved); + (state, _) = state.transition(LifecycleTransition::ConsumeRepair, 7).unwrap(); + assert_eq!(state.phase, LifecyclePhase::Implement); + assert!(state.repair_consumed); + + let patch_id = PatchId(Digest::blake3(b"repaired-patch")); + (state, _) = state + .transition( + LifecycleTransition::CompleteImplement { + worker: worker(&state.spec.profiles.implement), + patch: LifecyclePatchRef { patch_id, revision: 2 }, + }, + 8, + ) + .unwrap(); + let binding = state.spec.test_plans[0].clone(); + (state, _) = state + .transition( + LifecycleTransition::CompleteTest { + worker: worker(&state.spec.profiles.test), + results: vec![LifecycleTestResult { + plan_digest: binding.plan_digest(), + certificate_digest: binding.certificate_digest, + available: true, + executed: true, + passed: true, + evidence_id: Some("evidence:repair".to_owned()), + failure_code: None, + }], + }, + 9, + ) + .unwrap(); + let review = ReviewArtifact::new( + state.change_id.clone(), + patch_id, + LifecycleReviewVerdict::Approved, + coverage(), + Vec::new(), + state.spec.profiles.review.definition_digest, + 10, + ) + .unwrap(); + (state, _) = state + .transition( + LifecycleTransition::CompleteReview { + worker: worker(&state.spec.profiles.review), + review, + }, + 10, + ) + .unwrap(); + (state, _) = state + .transition(verify_transition(&state, VerificationStatus::Repairable, b"repair-2"), 11) + .unwrap(); + assert_eq!(state.status, LifecycleStatus::Failed); + assert_eq!(state.terminal_outcome, Some(LifecycleTerminalOutcome::Failed)); + assert_eq!(state.terminal_reason.as_ref().unwrap().code, "repair_limit_exhausted"); + assert_eq!( + state.transition(LifecycleTransition::ConsumeRepair, 12), + Err(LifecycleError::Terminal) + ); +} + +#[test] +fn stale_or_noncanonical_approval_cannot_unlock_apply() { + let mut state = advance_to_verify(lifecycle()); + let verification_id = VerificationArtifactId(Digest::blake3(b"verified")); + (state, _) = state + .transition(verify_transition(&state, VerificationStatus::Verified, b"verified"), 6) + .unwrap(); + let original = state.clone(); + let patch_id = state.patch.as_ref().unwrap().patch_id; + let stale = LifecycleApplyApproval::new( + Digest::blake3(b"stale-state"), + patch_id, + verification_id, + crate::ApprovalDecisionSource::WebUser, + 7, + ); + assert_eq!( + state.transition(LifecycleTransition::ApproveApply { approval: stale }, 7), + Err(LifecycleError::StaleApproval) + ); + let mut noncanonical = LifecycleApplyApproval::new( + state.state_digest(), + patch_id, + verification_id, + crate::ApprovalDecisionSource::WebUser, + 7, + ); + noncanonical.id = Digest::blake3(b"forged-approval"); + assert_eq!( + state.transition(LifecycleTransition::ApproveApply { approval: noncanonical }, 7,), + Err(LifecycleError::StaleApproval) + ); + assert_eq!(state, original); +} + +#[test] +fn impossible_projection_shape_is_rejected() { + let mut state = lifecycle(); + state.phase = LifecyclePhase::Apply; + state.status = LifecycleStatus::AwaitingApproval; + assert_eq!(state.validate(), Err(LifecycleError::InvalidState)); + + let mut approved_partial = coverage(); + approved_partial[0].status = AcceptanceStatus::Partial; + assert_eq!( + ReviewArtifact::new( + state.change_id, + PatchId(Digest::blake3(b"patch")), + LifecycleReviewVerdict::Approved, + approved_partial, + Vec::new(), + Digest::blake3(b"reviewer"), + 2, + ), + Err(LifecycleError::InvalidReview) + ); + + let mut absolute_path_plan = spec(); + absolute_path_plan.test_plans[0].plan.argv.push("/private/secret".to_owned()); + assert_eq!(absolute_path_plan.validate(), Err(LifecycleError::InvalidTestPlan)); +} + +#[test] +fn review_artifact_cannot_be_substituted_for_verification() { + let patch_id = PatchId(Digest::blake3(b"patch")); + let review = ReviewArtifact::new( + ChangeId::from_digest(Digest::blake3(b"change")), + patch_id, + LifecycleReviewVerdict::Approved, + coverage(), + Vec::new(), + Digest::blake3(b"reviewer"), + 1, + ) + .unwrap(); + let verification = VerificationArtifact { + id: VerificationArtifactId(review.id), + change_id: review.change_id.clone(), + patch_id, + verdict: VerificationStatus::Verified, + acceptance_coverage: verification_coverage(), + findings: Vec::new(), + test_evidence_ids: vec!["evidence:focused".to_owned()], + test_plan_results: vec![VerificationPlanResult { + plan_digest: Digest::blake3(b"plan"), + runner: "cargo".to_owned(), + argv: vec!["cargo".to_owned(), "test".to_owned()], + cwd_relative: ".".to_owned(), + test_identifier: "focused".to_owned(), + expected: true, + available: true, + executed: true, + passed: true, + evidence_id: Some("evidence:focused".to_owned()), + failure_reason: None, + }], + test_plans_over_cap: false, + verifier_definition: Digest::blake3(b"verifier"), + created_unix_ms: 1, + }; + assert!(!verification.is_canonical()); +} diff --git a/crates/needle-runtime/src/changes.rs b/crates/needle-runtime/src/changes.rs index 8fe3d36..8cc327c 100644 --- a/crates/needle-runtime/src/changes.rs +++ b/crates/needle-runtime/src/changes.rs @@ -130,6 +130,41 @@ pub fn apply_verified_change( repository_root: &Path, change_id: &ChangeId, expected_change_digest: Digest, +) -> Result { + apply_verified_change_with_lifecycle( + store, + repository_root, + change_id, + expected_change_digest, + None, + ) +} + +/// Apply a verified lifecycle change only after the exact approved lifecycle +/// projection has been supplied. Legacy changes without a lifecycle continue +/// through `apply_verified_change`; the two boundaries cannot be substituted. +pub fn apply_lifecycle_change( + store: &RuntimeStore, + repository_root: &Path, + change_id: &ChangeId, + expected_change_digest: Digest, + expected_lifecycle_digest: Digest, +) -> Result { + apply_verified_change_with_lifecycle( + store, + repository_root, + change_id, + expected_change_digest, + Some(expected_lifecycle_digest), + ) +} + +fn apply_verified_change_with_lifecycle( + store: &RuntimeStore, + repository_root: &Path, + change_id: &ChangeId, + expected_change_digest: Digest, + expected_lifecycle_digest: Option, ) -> Result { let repository_root = fs::canonicalize(repository_root)?; let repository_text = repository_root.to_string_lossy().into_owned(); @@ -177,7 +212,12 @@ pub fn apply_verified_change( "patch_id": prepared.patch.id, "paths": prepared.patch.files.iter().map(|file| &file.path).collect::>() }); - store.begin_change_apply(&record, &journal, expected_change_digest)?; + store.begin_change_apply_with_lifecycle( + &record, + &journal, + expected_change_digest, + expected_lifecycle_digest, + )?; let applied = materialize_patch_artifact(&repository_root, &prepared.patch, &blobs) .map_err(ChangeApplyError::from) .and_then(|_| capture_git_snapshot(&repository_root).map_err(ChangeApplyError::from)); diff --git a/crates/needle-runtime/src/lib.rs b/crates/needle-runtime/src/lib.rs index ea140a7..9398a48 100644 --- a/crates/needle-runtime/src/lib.rs +++ b/crates/needle-runtime/src/lib.rs @@ -25,10 +25,10 @@ pub use snapshot::{ validate_need_result, }; pub use store::{ - CacheRecord, ChangeAttemptRecord, ConfigExport, MainTurnObservationRecord, NeedShadowRecord, - NeedShadowWrite, NeedStepEventRecord, NeedStepRequestRecord, NegativeAttemptRecord, - OperatorCostKey, OperatorCostObservation, PatchFileBlob, PreparedChangeRecord, - ProofAccountingRecord, RoleProfileAuditOperation, RoleProfileAuditRecord, + CacheRecord, ChangeAttemptRecord, ConfigExport, LifecycleProjection, MainTurnObservationRecord, + NeedShadowRecord, NeedShadowWrite, NeedStepEventRecord, NeedStepRequestRecord, + NegativeAttemptRecord, OperatorCostKey, OperatorCostObservation, PatchFileBlob, + PreparedChangeRecord, ProofAccountingRecord, RoleProfileAuditOperation, RoleProfileAuditRecord, RoleProfileStateRecord, RouteCostObservation, RoutePromotionRecord, RuntimeSettings, RuntimeStore, SessionRecord, StoreError, WorkerRunRecord, }; @@ -81,7 +81,7 @@ pub use approval::{ }; pub use artifact_cache::{ArtifactCache, ArtifactCacheError, PlanCacheResult}; pub use changes::{ - ChangeApplyError, ChangeMaterializationError, apply_verified_change, + ChangeApplyError, ChangeMaterializationError, apply_lifecycle_change, apply_verified_change, materialize_patch_artifact, recover_pending_change_applies, validate_patch_artifact_base, }; pub use claim_proof::{ diff --git a/crates/needle-runtime/src/semantic_validation.rs b/crates/needle-runtime/src/semantic_validation.rs index 1dbb3d4..084694d 100644 --- a/crates/needle-runtime/src/semantic_validation.rs +++ b/crates/needle-runtime/src/semantic_validation.rs @@ -165,7 +165,7 @@ pub fn validate_semantic_test_plan( ) } -fn validate_parent_owned_test_plan_binding( +pub(crate) fn validate_parent_owned_test_plan_binding( worker_artifact: &SemanticWorkerArtifact, declared_plan: &TestPlan, ) -> Result<(), SemanticValidationError> { @@ -794,7 +794,7 @@ pub fn manifest_digest(manifest: &DependencyManifest) -> Digest { hasher.finish() } -fn validation_certificate_id( +pub(crate) fn validation_certificate_id( artifact: ArtifactId, input_artifacts: &[ArtifactId], evidence_ids: &[String], diff --git a/crates/needle-runtime/src/store.rs b/crates/needle-runtime/src/store.rs index 1c275be..10fa4e7 100644 --- a/crates/needle-runtime/src/store.rs +++ b/crates/needle-runtime/src/store.rs @@ -25,10 +25,13 @@ use thiserror::Error; mod changes; #[path = "store/claims.rs"] mod claims; +#[path = "store/lifecycles.rs"] +mod lifecycles; #[path = "store/role_profiles.rs"] mod role_profiles; pub use changes::{ChangeAttemptRecord, PatchFileBlob, PreparedChangeRecord}; +pub use lifecycles::LifecycleProjection; pub use role_profiles::{ RoleProfileAuditOperation, RoleProfileAuditRecord, RoleProfileStateRecord, }; @@ -884,6 +887,59 @@ BEGIN END; "#; +const MIGRATION_V16: &str = r#" +CREATE TABLE change_lifecycles ( + lifecycle_id TEXT NOT NULL UNIQUE CHECK(length(lifecycle_id) = 67), + change_id TEXT NOT NULL PRIMARY KEY REFERENCES change_requests(change_id), + source_snapshot_digest TEXT NOT NULL CHECK(length(source_snapshot_digest) = 67), + state_digest TEXT NOT NULL CHECK(length(state_digest) = 67), + generation INTEGER NOT NULL CHECK(generation >= 0), + state_json TEXT NOT NULL CHECK(length(state_json) <= 65536), + created_unix_ms INTEGER NOT NULL CHECK(created_unix_ms >= 0), + updated_unix_ms INTEGER NOT NULL CHECK(updated_unix_ms >= created_unix_ms) +); +ALTER TABLE change_events ADD COLUMN lifecycle_sequence INTEGER; +CREATE UNIQUE INDEX change_events_lifecycle_sequence + ON change_events(change_id, lifecycle_sequence) + WHERE lifecycle_sequence IS NOT NULL; +CREATE TRIGGER change_lifecycles_transition_shape +BEFORE UPDATE ON change_lifecycles +WHEN NEW.lifecycle_id <> OLD.lifecycle_id + OR NEW.change_id <> OLD.change_id + OR NEW.source_snapshot_digest <> OLD.source_snapshot_digest + OR NEW.created_unix_ms <> OLD.created_unix_ms + OR NEW.generation <> OLD.generation + 1 + OR NEW.updated_unix_ms < OLD.updated_unix_ms +BEGIN + SELECT RAISE(ABORT, 'invalid lifecycle projection transition'); +END; +CREATE TRIGGER change_lifecycles_no_delete +BEFORE DELETE ON change_lifecycles +BEGIN + SELECT RAISE(ABORT, 'lifecycle projections are durable'); +END; +CREATE TRIGGER change_events_no_update +BEFORE UPDATE ON change_events +BEGIN + SELECT RAISE(ABORT, 'change events are append-only'); +END; +CREATE TRIGGER change_events_no_delete +BEFORE DELETE ON change_events +BEGIN + SELECT RAISE(ABORT, 'change events are append-only'); +END; +CREATE TRIGGER lifecycle_event_payload_bound +BEFORE INSERT ON change_events +WHEN NEW.lifecycle_sequence IS NOT NULL AND ( + length(NEW.event_type) NOT BETWEEN 1 AND 64 + OR length(NEW.payload_digest) != 67 + OR length(NEW.payload_json) > 65536 +) +BEGIN + SELECT RAISE(ABORT, 'lifecycle event exceeds persisted bounds'); +END; +"#; + #[derive(Debug, Error)] pub enum StoreError { #[error("database operation failed: {0}")] @@ -928,6 +984,14 @@ pub enum StoreError { RoleProfileConflict(String), #[error("role profile was not found: {0}")] RoleProfileNotFound(String), + #[error("lifecycle validation failed: {0}")] + Lifecycle(#[from] needle_core::LifecycleError), + #[error("lifecycle operation conflicts: {0}")] + LifecycleConflict(String), + #[error("lifecycle was not found: {0}")] + LifecycleNotFound(String), + #[error("stored lifecycle is corrupt: {0}")] + LifecycleCorruption(String), #[error("database connection lock was poisoned")] ConnectionLock, } @@ -1251,6 +1315,7 @@ impl RuntimeStore { apply_migration(&mut connection, 13, MIGRATION_V13)?; apply_migration(&mut connection, 14, MIGRATION_V14)?; apply_migration(&mut connection, 15, MIGRATION_V15)?; + apply_migration(&mut connection, 16, MIGRATION_V16)?; connection.execute( "INSERT OR IGNORE INTO settings(key, value) VALUES('utility_gate_passed', '0')", [], @@ -5177,7 +5242,7 @@ mod tests { .unwrap() .collect::, _>>() .unwrap(); - assert_eq!(versions, vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]); + assert_eq!(versions, vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]); let columns = connection .prepare("PRAGMA table_info(worker_runs)") .unwrap() @@ -5284,6 +5349,23 @@ mod tests { ) .unwrap(); assert_eq!(v13_tables, 7); + let v16_tables: u32 = connection + .query_row( + "SELECT COUNT(*) FROM sqlite_master + WHERE type='table' AND name='change_lifecycles'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(v16_tables, 1); + let event_columns = connection + .prepare("PRAGMA table_info(change_events)") + .unwrap() + .query_map([], |row| row.get::<_, String>(1)) + .unwrap() + .collect::, _>>() + .unwrap(); + assert!(event_columns.iter().any(|column| column == "lifecycle_sequence")); drop(connection); let _ = fs::remove_file(path); } diff --git a/crates/needle-runtime/src/store/changes.rs b/crates/needle-runtime/src/store/changes.rs index 5f21d91..5ca5315 100644 --- a/crates/needle-runtime/src/store/changes.rs +++ b/crates/needle-runtime/src/store/changes.rs @@ -1,7 +1,8 @@ use super::*; use needle_core::{ CanonicalHasher, ChangeApplyId, ChangeApplyRecord, ChangeApplyStatus, ChangeId, ChangeRequest, - PatchArtifact, PatchId, VerificationArtifact, VerificationStatus, + LifecyclePhase, LifecycleReason, PatchArtifact, PatchId, VerificationArtifact, + VerificationStatus, }; type PreparedChangeRow = ( @@ -331,6 +332,12 @@ impl RuntimeStore { &patch.change_id, role_profile_provenance, )?; + super::lifecycles::require_lifecycle_worker_phase( + &transaction, + &patch.change_id, + LifecyclePhase::Implement, + role_profile_provenance, + )?; } match &existing { None if patch.revision != 1 => { @@ -551,6 +558,12 @@ impl RuntimeStore { if changed != 1 { return Err(StoreError::ChangeConflict(change_id.to_string())); } + super::lifecycles::fail_lifecycle( + &transaction, + change_id, + LifecycleReason::new("change_preparation_failed", reason.as_bytes())?, + now, + )?; transaction.execute( "INSERT INTO change_events( change_id, event_type, payload_digest, payload_json, created_unix_ms @@ -607,6 +620,13 @@ impl RuntimeStore { if changed != 1 { return Err(StoreError::ChangeConflict(change_id.to_string())); } + super::lifecycles::consume_lifecycle_repair( + &transaction, + change_id, + patch_id, + verification.id, + now, + )?; let payload_json = serde_json::to_string(&with_role_profile_provenance( &serde_json::json!({ "patch_id": patch_id, @@ -695,18 +715,28 @@ impl RuntimeStore { "verification references an unknown patch".to_owned(), )); } - let stored_provenance = require_change_role_profile_provenance( + let lifecycle_bound = super::lifecycles::require_lifecycle_worker_phase( &transaction, &artifact.change_id, + LifecyclePhase::Verify, role_profile_provenance, )?; + let event_provenance = if lifecycle_bound { + role_profile_provenance.cloned() + } else { + require_change_role_profile_provenance( + &transaction, + &artifact.change_id, + role_profile_provenance, + )? + }; let event_payload = with_role_profile_provenance( &serde_json::json!({ "verification_id": artifact.id, "patch_id": artifact.patch_id, "verdict": verdict }), - stored_provenance.as_ref(), + event_provenance.as_ref(), ); let event_json = serde_json::to_string(&event_payload)?; let event_digest = Digest::blake3(event_json.as_bytes()); @@ -915,6 +945,16 @@ impl RuntimeStore { record: &ChangeApplyRecord, journal: &serde_json::Value, expected_change_digest: Digest, + ) -> Result<(), StoreError> { + self.begin_change_apply_with_lifecycle(record, journal, expected_change_digest, None) + } + + pub fn begin_change_apply_with_lifecycle( + &self, + record: &ChangeApplyRecord, + journal: &serde_json::Value, + expected_change_digest: Digest, + expected_lifecycle_digest: Option, ) -> Result<(), StoreError> { if record.status != ChangeApplyStatus::Applying || record.post_snapshot.is_some() @@ -1025,6 +1065,17 @@ impl RuntimeStore { if existing != 0 { return Err(StoreError::ChangeConflict(record.change_id.to_string())); } + let verification_id = + verification.as_ref().expect("verified change validation requires an artifact").id; + super::lifecycles::begin_lifecycle_apply( + &transaction, + &record.change_id, + expected_lifecycle_digest, + record.id, + record.patch_id, + verification_id, + record.created_unix_ms, + )?; transaction.execute( "INSERT INTO change_applies( apply_id, change_id, patch_id, repository_root, pre_snapshot_digest, @@ -1111,6 +1162,13 @@ impl RuntimeStore { "UPDATE change_requests SET state=?2, updated_unix_ms=?3 WHERE change_id=?1", params![change_id, change_state, completed_unix_ms], )?; + super::lifecycles::finish_lifecycle_apply( + &transaction, + &parsed_change_id, + apply_id, + status, + completed_unix_ms, + )?; let event_json = serde_json::to_string(&with_role_profile_provenance( &serde_json::json!({ "apply_id": apply_id, diff --git a/crates/needle-runtime/src/store/lifecycles.rs b/crates/needle-runtime/src/store/lifecycles.rs new file mode 100644 index 0000000..fbb7755 --- /dev/null +++ b/crates/needle-runtime/src/store/lifecycles.rs @@ -0,0 +1,878 @@ +use super::*; +use needle_core::{ + ApprovalDecisionSource, ChangeApplyId, ChangeApplyStatus, ChangeId, ChangeRequest, CodexRole, + CommandExecutionEvidence, DevelopmentLifecycle, Digest, LifecycleApplyApproval, LifecycleError, + LifecycleEvent, LifecyclePhase, LifecycleReason, LifecycleSpec, LifecycleStatus, + LifecycleTransition, LifecycleWorkerProfiles, PatchId, ReviewArtifact, RoleProfileProvenance, + VerificationArtifact, VerificationArtifactId, +}; +use rusqlite::{Transaction, params}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeSet; + +type LifecycleChangeAnchors = + (String, String, u64, bool, Option, Option, Option); +type LifecycleProjectionRow = (String, String, String, String, u64); + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleProjection { + pub lifecycle: DevelopmentLifecycle, + pub state_digest: Digest, +} + +impl LifecycleProjection { + fn new(lifecycle: DevelopmentLifecycle) -> Result { + lifecycle.validate()?; + let state_digest = lifecycle.state_digest(); + Ok(Self { lifecycle, state_digest }) + } +} + +impl RuntimeStore { + pub fn create_lifecycle( + &self, + change_id: &ChangeId, + spec: LifecycleSpec, + ) -> Result { + self.initialize()?; + spec.validate()?; + let now = now_ms(); + let mut connection = self.connection()?; + let transaction = + connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + validate_profile_set(&transaction, &spec.profiles)?; + if lifecycle_projection_in_transaction(&transaction, change_id)?.is_some() { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: lifecycle already exists" + ))); + } + let anchors: Option = transaction + .query_row( + "SELECT source_snapshot_digest, state, latest_patch_revision, + repair_attempted, role_profile_id, role_profile_revision, + role_profile_definition_digest + FROM change_requests WHERE change_id=?1", + [change_id.to_string()], + |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + row.get(5)?, + row.get(6)?, + )) + }, + ) + .optional()?; + let Some(( + source, + state, + latest_patch_revision, + repair_attempted, + profile_id, + profile_revision, + profile_digest, + )) = anchors + else { + return Err(StoreError::LifecycleNotFound(format!( + "{change_id}: immutable change request" + ))); + }; + if state != "requested" || latest_patch_revision != 0 || repair_attempted { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: lifecycle must be created before implementation" + ))); + } + let source_snapshot = Digest::parse(&source) + .map_err(|_| StoreError::LifecycleCorruption(format!("{change_id}: source digest")))?; + validate_test_plan_certificates(&transaction, &spec, source_snapshot)?; + let request_profile = parse_profile_anchor(profile_id, profile_revision, profile_digest)?; + if request_profile.as_ref() != Some(&spec.profiles.implement) { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: request implementer profile differs from lifecycle" + ))); + } + let lifecycle = DevelopmentLifecycle::new(change_id.clone(), source_snapshot, spec, now)?; + let projection = LifecycleProjection::new(lifecycle)?; + let state_json = serde_json::to_string(&projection.lifecycle)?; + transaction.execute( + "INSERT INTO change_lifecycles( + lifecycle_id, change_id, source_snapshot_digest, state_digest, + generation, state_json, created_unix_ms, updated_unix_ms + ) VALUES(?1, ?2, ?3, ?4, 0, ?5, ?6, ?6)", + params![ + projection.lifecycle.id.to_string(), + change_id.to_string(), + source_snapshot.to_string(), + projection.state_digest.to_string(), + state_json, + now, + ], + )?; + let event = LifecycleEvent::created(&projection.lifecycle)?; + insert_lifecycle_event(&transaction, &event, "lifecycle_created")?; + transaction.commit()?; + Ok(projection) + } + + pub fn lifecycle( + &self, + change_id: &ChangeId, + ) -> Result, StoreError> { + let connection = self.connection()?; + lifecycle_projection_in_transaction(&*connection, change_id) + } + + pub fn lifecycle_events( + &self, + change_id: &ChangeId, + ) -> Result, StoreError> { + let connection = self.connection()?; + let mut statement = connection.prepare( + "SELECT payload_json, payload_digest FROM change_events + WHERE change_id=?1 AND lifecycle_sequence IS NOT NULL + ORDER BY lifecycle_sequence LIMIT ?2", + )?; + let events = statement + .query_map( + params![change_id.to_string(), needle_core::MAX_LIFECYCLE_EVENTS + 1], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)), + )? + .map(|row| { + let (json, stored_digest) = row?; + if Digest::blake3(json.as_bytes()).to_string() != stored_digest { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: lifecycle event payload digest" + ))); + } + let event: LifecycleEvent = serde_json::from_str(&json)?; + if event.change_id != *change_id { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: lifecycle event change identity" + ))); + } + Ok(event) + }) + .collect::, StoreError>>()?; + if events.len() > needle_core::MAX_LIFECYCLE_EVENTS { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: lifecycle event count" + ))); + } + Ok(events) + } + + pub fn replay_lifecycle( + &self, + change_id: &ChangeId, + ) -> Result { + let persisted = self + .lifecycle(change_id)? + .ok_or_else(|| StoreError::LifecycleNotFound(change_id.to_string()))?; + let replayed = DevelopmentLifecycle::replay(&self.lifecycle_events(change_id)?)?; + let replayed = LifecycleProjection::new(replayed)?; + if replayed != persisted { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: event replay differs from current projection" + ))); + } + Ok(replayed) + } + + /// Parent-owned transition boundary. Worker outputs are data carried by a + /// transition; workers do not receive this store capability. + pub fn parent_transition_lifecycle( + &self, + change_id: &ChangeId, + expected_state_digest: Digest, + transition: LifecycleTransition, + ) -> Result { + if matches!( + transition, + LifecycleTransition::ConsumeRepair + | LifecycleTransition::ApproveApply { .. } + | LifecycleTransition::StartApply { .. } + | LifecycleTransition::FinishApply { .. } + ) { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: transition is reserved for a typed parent operation" + ))); + } + let now = now_ms(); + let mut connection = self.connection()?; + let transaction = + connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + let projection = lifecycle_projection_in_transaction(&transaction, change_id)? + .ok_or_else(|| StoreError::LifecycleNotFound(change_id.to_string()))?; + if projection.state_digest != expected_state_digest { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: state digest changed" + ))); + } + validate_transition_artifacts(&transaction, &projection.lifecycle, &transition)?; + let result = + persist_transition(&transaction, &projection, expected_state_digest, transition, now)?; + transaction.commit()?; + Ok(result) + } + + /// Record an explicit user approval against the exact verified lifecycle + /// projection. The state digest prevents an approval from floating to a + /// later patch or verification revision. + pub fn approve_lifecycle_apply( + &self, + change_id: &ChangeId, + expected_state_digest: Digest, + source: ApprovalDecisionSource, + ) -> Result { + if source != ApprovalDecisionSource::WebUser { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: apply approval must come from an explicit web user decision" + ))); + } + let projection = self + .lifecycle(change_id)? + .ok_or_else(|| StoreError::LifecycleNotFound(change_id.to_string()))?; + if projection.state_digest != expected_state_digest + || projection.lifecycle.phase != LifecyclePhase::Apply + || projection.lifecycle.status != LifecycleStatus::AwaitingApproval + { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: lifecycle is not awaiting approval at this digest" + ))); + } + let patch = projection.lifecycle.patch.as_ref().ok_or(LifecycleError::MissingArtifact)?; + let verification = + projection.lifecycle.verification.as_ref().ok_or(LifecycleError::MissingArtifact)?; + let approval = LifecycleApplyApproval::new( + expected_state_digest, + patch.patch_id, + verification.verification_id, + source, + now_ms(), + ); + let mut connection = self.connection()?; + let transaction = + connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + let current = lifecycle_projection_in_transaction(&transaction, change_id)? + .ok_or_else(|| StoreError::LifecycleNotFound(change_id.to_string()))?; + let result = persist_transition( + &transaction, + ¤t, + expected_state_digest, + LifecycleTransition::ApproveApply { approval }, + now_ms(), + )?; + transaction.commit()?; + Ok(result) + } +} + +fn validate_profile_set( + transaction: &Transaction<'_>, + profiles: &LifecycleWorkerProfiles, +) -> Result<(), StoreError> { + for (phase, role) in [ + (LifecyclePhase::Explore, CodexRole::Explorer), + (LifecyclePhase::Implement, CodexRole::Implementer), + (LifecyclePhase::Test, CodexRole::TestRunner), + (LifecyclePhase::Review, CodexRole::Reviewer), + (LifecyclePhase::Verify, CodexRole::Verifier), + ] { + let provenance = profiles.for_phase(phase).ok_or(LifecycleError::ProfileMismatch)?; + let matches: u64 = transaction.query_row( + "SELECT COUNT(*) + FROM role_profiles p + JOIN role_profile_state s ON s.profile_id=p.profile_id + JOIN role_profile_revisions r + ON r.profile_id=p.profile_id AND r.revision=s.active_revision + WHERE p.profile_id=?1 AND p.role=?2 AND s.active_revision=?3 + AND r.definition_digest=?4 AND r.activated_unix_ms IS NOT NULL", + params![ + provenance.profile_id.as_str(), + role.as_str(), + provenance.revision, + provenance.definition_digest.to_string(), + ], + |row| row.get(0), + )?; + if matches != 1 { + return Err(StoreError::LifecycleConflict(format!( + "{}: profile is not bound to {}", + provenance.profile_id, + phase.as_str() + ))); + } + } + Ok(()) +} + +fn validate_test_plan_certificates( + transaction: &Transaction<'_>, + spec: &LifecycleSpec, + source_snapshot: Digest, +) -> Result<(), StoreError> { + for binding in &spec.test_plans { + let row: Option<(String, String, String, String)> = transaction + .query_row( + "SELECT a.artifact_id, a.request_id, a.artifact_json, c.certificate_json + FROM artifact_validation_certificates c + JOIN artifacts a ON a.artifact_id=c.artifact_id + JOIN artifact_requests r ON r.request_id=a.request_id + WHERE c.certificate_id=?1 AND a.format_revision=2 + AND r.source_digest=?2", + params![binding.certificate_digest.to_string(), source_snapshot.to_string(),], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + let Some((artifact_id, request_id, artifact_json, certificate_json)) = row else { + return Err(StoreError::LifecycleConflict( + "lifecycle test plan has no persisted validation certificate".to_owned(), + )); + }; + let (_, worker_artifact) = + parse_semantic_artifact(&artifact_id, &request_id, &artifact_json)?; + crate::semantic_validation::validate_parent_owned_test_plan_binding( + &worker_artifact, + &binding.plan, + ) + .map_err(|_| { + StoreError::LifecycleConflict( + "lifecycle test plan differs from its certified artifact".to_owned(), + ) + })?; + let certificate: needle_core::ArtifactValidationCertificate = + serde_json::from_str(&certificate_json)?; + if certificate.id.digest() != binding.certificate_digest + || certificate.artifact.to_string() != artifact_id + || certificate.test_plan_evidence.is_none() + || !validation_certificate_is_structurally_canonical(&certificate) + { + return Err(StoreError::LifecycleConflict( + "lifecycle test-plan certificate identity or evidence status is invalid".to_owned(), + )); + } + } + Ok(()) +} + +fn validation_certificate_is_structurally_canonical( + certificate: &needle_core::ArtifactValidationCertificate, +) -> bool { + certificate.dependency_checks_digest == certificate.coverage.dependency_manifest_digest + && certificate.id + == crate::semantic_validation::validation_certificate_id( + certificate.artifact, + &certificate.input_artifacts, + &certificate.evidence_ids, + &certificate.coverage, + certificate.validator_definition, + certificate.test_plan_evidence, + ) +} + +fn parse_semantic_artifact( + stored_artifact_id: &str, + stored_request_id: &str, + artifact_json: &str, +) -> Result<(needle_core::Artifact, needle_core::SemanticWorkerArtifact), StoreError> { + let artifact: needle_core::Artifact = serde_json::from_str(artifact_json)?; + let worker_artifact: needle_core::SemanticWorkerArtifact = + serde_json::from_value(artifact.payload.clone())?; + let canonical_id = + worker_artifact + .canonical_artifact_id(artifact.contract.definition_digest) + .ok_or_else(|| StoreError::LifecycleCorruption("semantic artifact bound".to_owned()))?; + if artifact.id.to_string() != stored_artifact_id + || artifact.request_id.to_string() != stored_request_id + || artifact.contract.schema_id != needle_core::SEMANTIC_ARTIFACT_RESULT_SCHEMA_ID + || artifact.contract.kind != worker_artifact.kind() + || artifact.contract.cache_scope != artifact.dependency_manifest.scope + || canonical_id.digest() != artifact.id + { + return Err(StoreError::LifecycleCorruption("semantic artifact identity".to_owned())); + } + Ok((artifact, worker_artifact)) +} + +fn parse_profile_anchor( + profile_id: Option, + revision: Option, + digest: Option, +) -> Result, StoreError> { + match (profile_id, revision, digest) { + (None, None, None) => Ok(None), + (Some(profile_id), Some(revision), Some(digest)) => Ok(Some( + RoleProfileProvenance::new( + needle_core::RoleProfileId::new(profile_id) + .map_err(|_| StoreError::LifecycleCorruption("profile id".to_owned()))?, + revision, + Digest::parse(&digest) + .map_err(|_| StoreError::LifecycleCorruption("profile digest".to_owned()))?, + ) + .map_err(|_| StoreError::LifecycleCorruption("profile anchor".to_owned()))?, + )), + _ => Err(StoreError::LifecycleCorruption("partial role-profile anchor".to_owned())), + } +} + +trait LifecycleConnection { + fn query_projection_row( + &self, + change_id: &ChangeId, + ) -> rusqlite::Result>; +} + +impl LifecycleConnection for rusqlite::Connection { + fn query_projection_row( + &self, + change_id: &ChangeId, + ) -> rusqlite::Result> { + self.query_row( + "SELECT lifecycle_id, source_snapshot_digest, state_digest, state_json, generation + FROM change_lifecycles WHERE change_id=?1", + [change_id.to_string()], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)), + ) + .optional() + } +} + +impl LifecycleConnection for Transaction<'_> { + fn query_projection_row( + &self, + change_id: &ChangeId, + ) -> rusqlite::Result> { + self.query_row( + "SELECT lifecycle_id, source_snapshot_digest, state_digest, state_json, generation + FROM change_lifecycles WHERE change_id=?1", + [change_id.to_string()], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)), + ) + .optional() + } +} + +fn lifecycle_projection_in_transaction( + connection: &C, + change_id: &ChangeId, +) -> Result, StoreError> { + let Some((lifecycle_id, source_snapshot, stored_state_digest, state_json, generation)) = + connection.query_projection_row(change_id)? + else { + return Ok(None); + }; + let lifecycle: DevelopmentLifecycle = serde_json::from_str(&state_json)?; + lifecycle.validate().map_err(StoreError::from)?; + let state_digest = lifecycle.state_digest(); + if lifecycle.change_id != *change_id + || lifecycle.id.to_string() != lifecycle_id + || lifecycle.source_snapshot.to_string() != source_snapshot + || state_digest.to_string() != stored_state_digest + || lifecycle.generation != generation + { + return Err(StoreError::LifecycleCorruption(change_id.to_string())); + } + Ok(Some(LifecycleProjection { lifecycle, state_digest })) +} + +fn persist_transition( + transaction: &Transaction<'_>, + projection: &LifecycleProjection, + expected_state_digest: Digest, + transition: LifecycleTransition, + created_unix_ms: u64, +) -> Result { + if projection.state_digest != expected_state_digest { + return Err(StoreError::LifecycleConflict(format!( + "{}: state digest changed", + projection.lifecycle.change_id + ))); + } + let (next, event) = projection.lifecycle.transition(transition, created_unix_ms)?; + let next = LifecycleProjection::new(next)?; + let changed = transaction.execute( + "UPDATE change_lifecycles + SET state_digest=?2, generation=?3, state_json=?4, updated_unix_ms=?5 + WHERE change_id=?1 AND state_digest=?6 AND generation=?7", + params![ + projection.lifecycle.change_id.to_string(), + next.state_digest.to_string(), + next.lifecycle.generation, + serde_json::to_string(&next.lifecycle)?, + created_unix_ms, + expected_state_digest.to_string(), + projection.lifecycle.generation, + ], + )?; + if changed != 1 { + return Err(StoreError::LifecycleConflict(format!( + "{}: concurrent transition", + projection.lifecycle.change_id + ))); + } + insert_lifecycle_event(transaction, &event, "lifecycle_transitioned")?; + Ok(next) +} + +fn insert_lifecycle_event( + transaction: &Transaction<'_>, + event: &LifecycleEvent, + event_type: &str, +) -> Result<(), StoreError> { + let payload_json = serde_json::to_string(event)?; + let payload_digest = Digest::blake3(payload_json.as_bytes()); + transaction.execute( + "INSERT INTO change_events( + change_id, event_type, payload_digest, payload_json, + created_unix_ms, lifecycle_sequence + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6)", + params![ + event.change_id.to_string(), + event_type, + payload_digest.to_string(), + payload_json, + event.created_unix_ms, + event.sequence, + ], + )?; + Ok(()) +} + +fn validate_transition_artifacts( + transaction: &Transaction<'_>, + lifecycle: &DevelopmentLifecycle, + transition: &LifecycleTransition, +) -> Result<(), StoreError> { + match transition { + LifecycleTransition::CompleteExplore { artifacts, .. } => { + for artifact in artifacts { + let row: Option<(String, String, String, String)> = transaction + .query_row( + "SELECT a.artifact_id, a.request_id, a.artifact_json, c.certificate_json + FROM artifacts a + JOIN artifact_requests r ON r.request_id=a.request_id + JOIN artifact_validation_certificates c ON c.artifact_id=a.artifact_id + WHERE a.artifact_id=?1 AND a.format_revision=2 + AND r.source_digest=?2 + ORDER BY c.issued_unix_ms DESC, c.certificate_id LIMIT 1", + params![artifact.id.to_string(), lifecycle.source_snapshot.to_string()], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + let Some((artifact_id, request_id, json, certificate_json)) = row else { + return Err(StoreError::LifecycleConflict(format!( + "{}: exploration artifact is unavailable", + lifecycle.change_id + ))); + }; + let (_, worker_artifact) = + parse_semantic_artifact(&artifact_id, &request_id, &json)?; + if matches!(worker_artifact, needle_core::SemanticWorkerArtifact::TestPlan { .. }) { + return Err(StoreError::LifecycleConflict(format!( + "{}: test-plan artifact cannot satisfy exploration", + lifecycle.change_id + ))); + } + let certificate: needle_core::ArtifactValidationCertificate = + serde_json::from_str(&certificate_json)?; + if certificate.artifact.to_string() != artifact.id.to_string() + || !validation_certificate_is_structurally_canonical(&certificate) + { + return Err(StoreError::LifecycleConflict(format!( + "{}: exploration certificate identity is invalid", + lifecycle.change_id + ))); + } + } + } + LifecycleTransition::CompleteImplement { patch, .. } => { + let row: Option<(u32, String)> = transaction + .query_row( + "SELECT p.revision, p.source_snapshot_digest + FROM patch_artifacts p + JOIN change_requests c + ON c.change_id=p.change_id AND c.latest_patch_revision=p.revision + WHERE p.change_id=?1 AND p.patch_id=?2", + params![lifecycle.change_id.to_string(), patch.patch_id.to_string()], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional()?; + if row != Some((patch.revision, lifecycle.source_snapshot.to_string())) { + return Err(StoreError::LifecycleConflict(format!( + "{}: patch is not the latest source-bound revision", + lifecycle.change_id + ))); + } + } + LifecycleTransition::CompleteTest { results, .. } => { + for (binding, result) in lifecycle.spec.test_plans.iter().zip(results) { + let Some(evidence_id) = result.evidence_id.as_deref() else { + continue; + }; + let evidence_json: Option = transaction + .query_row( + "SELECT evidence_json FROM command_evidence WHERE evidence_id=?1", + [evidence_id], + |row| row.get(0), + ) + .optional()?; + let Some(evidence_json) = evidence_json else { + return Err(StoreError::LifecycleConflict(format!( + "{}: test evidence is unavailable", + lifecycle.change_id + ))); + }; + let evidence: CommandExecutionEvidence = serde_json::from_str(&evidence_json)?; + if evidence.source_snapshot_digest != lifecycle.source_snapshot + || evidence.runner != binding.plan.runner + || evidence.argv != binding.plan.argv + || evidence.test_identifier.as_deref() + != Some(binding.plan.test_identifier.as_str()) + { + return Err(StoreError::LifecycleConflict(format!( + "{}: test evidence differs from the frozen plan", + lifecycle.change_id + ))); + } + if result.passed { + crate::validate_test_evidence(&binding.plan, &evidence).map_err(|_| { + StoreError::LifecycleConflict(format!( + "{}: passing test evidence is invalid", + lifecycle.change_id + )) + })?; + } else if result.executed + && evidence.exit_status == Some(0) + && evidence.infrastructure_failure.is_none() + { + return Err(StoreError::LifecycleConflict(format!( + "{}: failed test result contradicts evidence", + lifecycle.change_id + ))); + } + } + } + LifecycleTransition::CompleteReview { review, .. } => { + validate_review_artifact(transaction, lifecycle, review)?; + } + LifecycleTransition::CompleteVerify { worker, verification } => { + let json: Option = transaction + .query_row( + "SELECT artifact_json FROM verification_artifacts + WHERE verification_id=?1 AND change_id=?2 AND patch_id=?3", + params![ + verification.verification_id.to_string(), + lifecycle.change_id.to_string(), + verification.patch_id.to_string(), + ], + |row| row.get(0), + ) + .optional()?; + let Some(json) = json else { + return Err(StoreError::LifecycleConflict(format!( + "{}: verification artifact is unavailable", + lifecycle.change_id + ))); + }; + let artifact: VerificationArtifact = serde_json::from_str(&json)?; + if !artifact.is_canonical() + || artifact.id != verification.verification_id + || artifact.verdict != verification.verdict + || artifact.verifier_definition != worker.profile.definition_digest + || lifecycle.patch.as_ref().map(|patch| patch.patch_id) != Some(artifact.patch_id) + { + return Err(StoreError::LifecycleConflict(format!( + "{}: verification artifact is not current and canonical", + lifecycle.change_id + ))); + } + } + LifecycleTransition::Cancel { .. } | LifecycleTransition::Fail { .. } => {} + LifecycleTransition::ConsumeRepair + | LifecycleTransition::ApproveApply { .. } + | LifecycleTransition::StartApply { .. } + | LifecycleTransition::FinishApply { .. } => { + return Err(StoreError::LifecycleConflict(format!( + "{}: reserved lifecycle transition", + lifecycle.change_id + ))); + } + } + Ok(()) +} + +fn validate_review_artifact( + transaction: &Transaction<'_>, + lifecycle: &DevelopmentLifecycle, + review: &ReviewArtifact, +) -> Result<(), StoreError> { + let request_json: String = transaction.query_row( + "SELECT request_json FROM change_requests WHERE change_id=?1", + [lifecycle.change_id.to_string()], + |row| row.get(0), + )?; + let request: ChangeRequest = serde_json::from_str(&request_json)?; + let expected = request.acceptance_criteria.iter().map(Digest::blake3).collect::>(); + let observed = review + .acceptance_coverage + .iter() + .map(|coverage| coverage.criterion_digest) + .collect::>(); + if expected != observed { + return Err(StoreError::LifecycleConflict(format!( + "{}: review coverage differs from acceptance criteria", + lifecycle.change_id + ))); + } + Ok(()) +} + +pub(super) fn require_lifecycle_worker_phase( + transaction: &Transaction<'_>, + change_id: &ChangeId, + phase: LifecyclePhase, + provenance: Option<&RoleProfileProvenance>, +) -> Result { + let Some(projection) = lifecycle_projection_in_transaction(transaction, change_id)? else { + return Ok(false); + }; + if projection.lifecycle.phase != phase + || projection.lifecycle.status != LifecycleStatus::Active + || projection.lifecycle.spec.profiles.for_phase(phase) != provenance + { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: worker write is outside the active lifecycle phase" + ))); + } + Ok(true) +} + +pub(super) fn consume_lifecycle_repair( + transaction: &Transaction<'_>, + change_id: &ChangeId, + patch_id: PatchId, + verification_id: VerificationArtifactId, + created_unix_ms: u64, +) -> Result<(), StoreError> { + let Some(projection) = lifecycle_projection_in_transaction(transaction, change_id)? else { + return Ok(()); + }; + if projection.lifecycle.patch.as_ref().map(|patch| patch.patch_id) != Some(patch_id) + || projection + .lifecycle + .verification + .as_ref() + .map(|verification| verification.verification_id) + != Some(verification_id) + { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: repair artifacts differ from lifecycle" + ))); + } + persist_transition( + transaction, + &projection, + projection.state_digest, + LifecycleTransition::ConsumeRepair, + created_unix_ms, + )?; + Ok(()) +} + +pub(super) fn fail_lifecycle( + transaction: &Transaction<'_>, + change_id: &ChangeId, + reason: LifecycleReason, + created_unix_ms: u64, +) -> Result<(), StoreError> { + let Some(projection) = lifecycle_projection_in_transaction(transaction, change_id)? else { + return Ok(()); + }; + if projection.lifecycle.status != LifecycleStatus::Active + || !matches!( + projection.lifecycle.phase, + LifecyclePhase::Explore | LifecyclePhase::Implement + ) + { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: preparation failure is outside an active preparation phase" + ))); + } + persist_transition( + transaction, + &projection, + projection.state_digest, + LifecycleTransition::Fail { reason }, + created_unix_ms, + )?; + Ok(()) +} + +pub(super) fn begin_lifecycle_apply( + transaction: &Transaction<'_>, + change_id: &ChangeId, + expected_state_digest: Option, + apply_id: ChangeApplyId, + patch_id: PatchId, + verification_id: VerificationArtifactId, + created_unix_ms: u64, +) -> Result<(), StoreError> { + let lifecycle = lifecycle_projection_in_transaction(transaction, change_id)?; + match (lifecycle, expected_state_digest) { + (None, None) => Ok(()), + (None, Some(_)) => Err(StoreError::LifecycleNotFound(change_id.to_string())), + (Some(_), None) => Err(StoreError::LifecycleConflict(format!( + "{change_id}: lifecycle apply requires its current state digest" + ))), + (Some(projection), Some(expected)) => { + if projection.lifecycle.patch.as_ref().map(|patch| patch.patch_id) != Some(patch_id) + || projection + .lifecycle + .verification + .as_ref() + .map(|verification| verification.verification_id) + != Some(verification_id) + { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: apply artifacts differ from approved lifecycle" + ))); + } + persist_transition( + transaction, + &projection, + expected, + LifecycleTransition::StartApply { apply_id }, + created_unix_ms, + )?; + Ok(()) + } + } +} + +pub(super) fn finish_lifecycle_apply( + transaction: &Transaction<'_>, + change_id: &ChangeId, + apply_id: ChangeApplyId, + status: ChangeApplyStatus, + completed_unix_ms: u64, +) -> Result<(), StoreError> { + let Some(projection) = lifecycle_projection_in_transaction(transaction, change_id)? else { + return Ok(()); + }; + persist_transition( + transaction, + &projection, + projection.state_digest, + LifecycleTransition::FinishApply { apply_id, status }, + completed_unix_ms, + )?; + Ok(()) +} + +#[cfg(test)] +#[path = "lifecycles/tests.rs"] +mod tests; diff --git a/crates/needle-runtime/src/store/lifecycles/tests.rs b/crates/needle-runtime/src/store/lifecycles/tests.rs new file mode 100644 index 0000000..0878e95 --- /dev/null +++ b/crates/needle-runtime/src/store/lifecycles/tests.rs @@ -0,0 +1,958 @@ +use super::*; +use needle_core::{ + AcceptanceCoverage, AcceptanceStatus, AllowedPath, AllowedPathScope, ApprovalDecisionSource, + Artifact, ArtifactContract, ArtifactValidationCertificate, CacheScope, ChangeApplyId, + ChangeApplyRecord, ChangeApplyStatus, CodexHost, CommandExecutionEvidence, CommandPolicy, + CoverageManifest, DependencyManifest, FallbackPolicy, FilesystemPolicy, + LifecycleAcceptanceReview, LifecycleArtifactKind, LifecycleArtifactRef, LifecycleBudget, + LifecyclePatchRef, LifecycleReviewVerdict, LifecycleTestPlanBinding, LifecycleTestResult, + LifecycleUsage, LifecycleVerificationRef, LifecycleWorkerCompletion, LocationRole, + NetworkPolicy, PatchArtifact, PatchFile, PatchOperation, ReasoningLevel, RepairPolicy, + ReviewArtifact, RoleProfileBudget, RoleProfileDefinition, RoleProfileDefinitionInput, + RoleProfileState, SemanticLocation, SemanticWorkerArtifact, SemanticWorld, ServiceTier, + TestPlan, TestPlanEvidenceStatus, TestPolicy, ToolPolicy, VerificationArtifact, + VerificationPlanResult, VerificationStatus, VerificationTestProjection, +}; +use std::sync::{Arc, Barrier}; +use std::thread; +use std::time::{SystemTime, UNIX_EPOCH}; + +fn temporary_path(name: &str) -> PathBuf { + let suffix = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos(); + std::env::temp_dir() + .join(format!("needle-lifecycle-{name}-{}-{suffix}.sqlite3", std::process::id())) +} + +fn active_profile(store: &RuntimeStore, id: &str, role: CodexRole) -> RoleProfileProvenance { + let test_role = matches!(role, CodexRole::TestRunner | CodexRole::Verifier); + let implementer = role == CodexRole::Implementer; + let definition = RoleProfileDefinition::new(RoleProfileDefinitionInput { + profile_id: needle_core::RoleProfileId::new(id).unwrap(), + role, + host: CodexHost::Codex, + model: "offline-lifecycle-worker".to_owned(), + reasoning: ReasoningLevel::Medium, + service_tier: ServiceTier::Default, + timeout_seconds: 30, + budget: RoleProfileBudget { + max_turns: 2, + max_output_tokens: 1_200, + max_cost_microusd: 10_000, + }, + prompt_profile_digest: Digest::blake3(format!("{id}:prompt")), + output_contract_digest: Digest::blake3(format!("{id}:output")), + tool_policy: if implementer { ToolPolicy::IsolatedWrite } else { ToolPolicy::ReadOnly }, + command_policy: if test_role { + CommandPolicy::CertifiedTests + } else { + CommandPolicy::Denied + }, + filesystem_policy: if implementer { + FilesystemPolicy::DisposableCheckout + } else { + FilesystemPolicy::ReadOnlyCheckout + }, + network_policy: NetworkPolicy::Denied, + test_policy: if test_role { TestPolicy::Certified } else { TestPolicy::Disabled }, + repair_policy: if implementer { RepairPolicy::Once } else { RepairPolicy::None }, + fallback_policy: FallbackPolicy::Disabled, + concurrency: 1, + route_assignments: Vec::new(), + }) + .unwrap(); + let revision = store.create_role_profile(definition).unwrap(); + let state = store.role_profile_state(&revision.profile_id).unwrap(); + let revision = store + .activate_role_profile(&revision.profile_id, revision.revision, state.state_digest) + .unwrap(); + assert_eq!(revision.state, RoleProfileState::Active); + RoleProfileProvenance::from_revision(&revision).unwrap() +} + +fn worker(profile: &RoleProfileProvenance) -> LifecycleWorkerCompletion { + LifecycleWorkerCompletion { + profile: profile.clone(), + worker_depth: 1, + logical_worker_spawns: 1, + usage: LifecycleUsage { worker_turns: 1, output_tokens: 10, cost_microusd: 10 }, + } +} + +fn seed_test_plan_certificate( + store: &RuntimeStore, + name: &str, + source: Digest, + plan: &TestPlan, +) -> Digest { + let request_id = Digest::blake3(format!("{name}:test-plan-request")); + let dependency_digest = Digest::blake3(format!("{name}:dependency-manifest")); + let world = SemanticWorld { + repository_lineage: Digest::blake3(format!("{name}:repository")), + source_selector: source.to_string(), + platform: "test".to_owned(), + features: "offline".to_owned(), + configuration: None, + toolchain: None, + }; + let worker_artifact = SemanticWorkerArtifact::TestPlan { + runner: plan.runner.clone(), + argv: plan.argv.clone(), + cwd_relative: plan.cwd_relative.clone(), + identifiers: vec![plan.test_identifier.clone()], + selection: "focused".to_owned(), + evidence_paths: Vec::new(), + }; + let contract = ArtifactContract::semantic( + format!("{name}.test-plan"), + 1, + worker_artifact.kind(), + CacheScope::SnapshotExact, + ); + let artifact_id = worker_artifact.canonical_artifact_id(contract.definition_digest).unwrap(); + let artifact = Artifact { + id: artifact_id.digest(), + request_id, + contract, + payload: serde_json::to_value(&worker_artifact).unwrap(), + dependency_manifest: DependencyManifest { + scope: CacheScope::SnapshotExact, + observed_files_complete: true, + dependencies: Vec::new(), + gaps: Vec::new(), + }, + validations: Vec::new(), + created_unix_ms: 1, + }; + let coverage = CoverageManifest { + entries: Vec::new(), + world: world.clone(), + dependency_manifest_digest: dependency_digest, + }; + let validator_definition = Digest::blake3(format!("{name}:validator")); + let certificate_id = crate::semantic_validation::validation_certificate_id( + artifact_id, + &[], + &[], + &coverage, + validator_definition, + Some(TestPlanEvidenceStatus::Located), + ); + let certificate_digest = certificate_id.digest(); + let certificate = ArtifactValidationCertificate { + id: certificate_id, + artifact: artifact_id, + input_artifacts: Vec::new(), + evidence_ids: Vec::new(), + test_plan_evidence: Some(TestPlanEvidenceStatus::Located), + coverage, + validator_definition, + dependency_checks_digest: dependency_digest, + issued_unix_ms: 1, + }; + let connection = store.connection().unwrap(); + connection + .execute( + "INSERT INTO artifact_requests( + request_id, logical_id, source_digest, contract_id, route_key, + request_json, created_unix_ms, format_revision + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, 1, 2)", + params![ + request_id.to_string(), + format!("{name}:test-plan"), + source.to_string(), + artifact.contract.id.as_str(), + "test-plan", + "{}", + ], + ) + .unwrap(); + connection + .execute( + "INSERT INTO artifacts( + artifact_id, request_id, contract_id, artifact_json, + created_unix_ms, format_revision + ) VALUES(?1, ?2, ?3, ?4, 1, 2)", + params![ + artifact_id.to_string(), + request_id.to_string(), + artifact.contract.id.as_str(), + serde_json::to_string(&artifact).unwrap(), + ], + ) + .unwrap(); + connection + .execute( + "INSERT INTO semantic_worlds(world_digest, world_json, created_unix_ms) + VALUES(?1, ?2, 1)", + params![world.id().to_string(), serde_json::to_string(&world).unwrap()], + ) + .unwrap(); + connection + .execute( + "INSERT INTO artifact_validation_certificates( + certificate_id, artifact_id, validator_definition_digest, + dependency_manifest_digest, world_digest, certificate_json, issued_unix_ms + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, 1)", + params![ + certificate_digest.to_string(), + artifact_id.to_string(), + certificate.validator_definition.to_string(), + dependency_digest.to_string(), + world.id().to_string(), + serde_json::to_string(&certificate).unwrap(), + ], + ) + .unwrap(); + certificate_digest +} + +fn seed_exploration_certificate(store: &RuntimeStore, name: &str, source: Digest) -> Digest { + let request_id = Digest::blake3(format!("{name}:exploration-request")); + let dependency_digest = Digest::blake3(format!("{name}:exploration-dependencies")); + let world = SemanticWorld { + repository_lineage: Digest::blake3(format!("{name}:repository")), + source_selector: source.to_string(), + platform: "test".to_owned(), + features: "exploration".to_owned(), + configuration: None, + toolchain: None, + }; + let worker_artifact = SemanticWorkerArtifact::CodeLocation { + locations: vec![SemanticLocation { + role: LocationRole::Primary, + path: "fixture.txt".to_owned(), + symbol: None, + byte_start: Some(0), + byte_end: Some(1), + }], + gaps: Vec::new(), + }; + let contract = ArtifactContract::semantic( + format!("{name}.exploration"), + 1, + worker_artifact.kind(), + CacheScope::SnapshotExact, + ); + let artifact_id = worker_artifact.canonical_artifact_id(contract.definition_digest).unwrap(); + let artifact_digest = artifact_id.digest(); + let artifact = Artifact { + id: artifact_digest, + request_id, + contract, + payload: serde_json::to_value(&worker_artifact).unwrap(), + dependency_manifest: DependencyManifest { + scope: CacheScope::SnapshotExact, + observed_files_complete: true, + dependencies: Vec::new(), + gaps: Vec::new(), + }, + validations: Vec::new(), + created_unix_ms: 1, + }; + let coverage = CoverageManifest { + entries: Vec::new(), + world: world.clone(), + dependency_manifest_digest: dependency_digest, + }; + let validator_definition = Digest::blake3(format!("{name}:exploration-validator")); + let certificate_id = crate::semantic_validation::validation_certificate_id( + artifact_id, + &[], + &[], + &coverage, + validator_definition, + None, + ); + let certificate_digest = certificate_id.digest(); + let certificate = ArtifactValidationCertificate { + id: certificate_id, + artifact: artifact_id, + input_artifacts: Vec::new(), + evidence_ids: Vec::new(), + test_plan_evidence: None, + coverage, + validator_definition, + dependency_checks_digest: dependency_digest, + issued_unix_ms: 1, + }; + let connection = store.connection().unwrap(); + connection + .execute( + "INSERT INTO artifact_requests( + request_id, logical_id, source_digest, contract_id, route_key, + request_json, created_unix_ms, format_revision + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, 1, 2)", + params![ + request_id.to_string(), + format!("{name}:exploration"), + source.to_string(), + artifact.contract.id.as_str(), + "code-location", + "{}", + ], + ) + .unwrap(); + connection + .execute( + "INSERT INTO artifacts( + artifact_id, request_id, contract_id, artifact_json, + created_unix_ms, format_revision + ) VALUES(?1, ?2, ?3, ?4, 1, 2)", + params![ + artifact_id.to_string(), + request_id.to_string(), + artifact.contract.id.as_str(), + serde_json::to_string(&artifact).unwrap(), + ], + ) + .unwrap(); + connection + .execute( + "INSERT INTO semantic_worlds(world_digest, world_json, created_unix_ms) + VALUES(?1, ?2, 1)", + params![world.id().to_string(), serde_json::to_string(&world).unwrap()], + ) + .unwrap(); + connection + .execute( + "INSERT INTO artifact_validation_certificates( + certificate_id, artifact_id, validator_definition_digest, + dependency_manifest_digest, world_digest, certificate_json, issued_unix_ms + ) VALUES(?1, ?2, ?3, ?4, ?5, ?6, 1)", + params![ + certificate_digest.to_string(), + artifact_id.to_string(), + certificate.validator_definition.to_string(), + dependency_digest.to_string(), + world.id().to_string(), + serde_json::to_string(&certificate).unwrap(), + ], + ) + .unwrap(); + artifact_digest +} + +fn lifecycle_fixture(name: &str) -> (PathBuf, RuntimeStore, ChangeId, LifecycleProjection) { + let path = temporary_path(name); + let store = RuntimeStore::new(&path); + store.initialize().unwrap(); + let profiles = LifecycleWorkerProfiles { + explore: active_profile(&store, &format!("{name}.explorer"), CodexRole::Explorer), + implement: active_profile(&store, &format!("{name}.implementer"), CodexRole::Implementer), + test: active_profile(&store, &format!("{name}.test"), CodexRole::TestRunner), + review: active_profile(&store, &format!("{name}.review"), CodexRole::Reviewer), + verify: active_profile(&store, &format!("{name}.verify"), CodexRole::Verifier), + }; + let source = Digest::blake3(format!("{name}:source")); + let plan = TestPlan { + runner: "cargo".to_owned(), + argv: vec![ + "cargo".to_owned(), + "test".to_owned(), + "--offline".to_owned(), + "focused".to_owned(), + "--".to_owned(), + "--exact".to_owned(), + ], + cwd_relative: ".".to_owned(), + test_identifier: "focused".to_owned(), + requires_approval: true, + execution_evidence_id: None, + }; + let certificate_digest = seed_test_plan_certificate(&store, name, source, &plan); + let request = ChangeRequest { + task: "Implement the bounded lifecycle fixture.".to_owned(), + acceptance_criteria: vec!["The lifecycle remains fail closed.".to_owned()], + allowed_paths: vec![AllowedPath { + path: "fixture.txt".to_owned(), + scope: AllowedPathScope::Exact, + }], + artifact_ids: Vec::new(), + claim_ids: Vec::new(), + constraints: Vec::new(), + }; + let change_id = ChangeId::from_digest(Digest::blake3(format!("{name}:change"))); + store + .record_change_request_with_provenance( + &change_id, + Digest::blake3(format!("{name}:repository")), + source, + request.digest(source), + &request, + Some(&profiles.implement), + ) + .unwrap(); + let projection = store + .create_lifecycle( + &change_id, + LifecycleSpec { + worker_depth_limit: 1, + profiles, + budget: LifecycleBudget { + max_worker_turns: 10, + max_output_tokens: 10_000, + max_cost_microusd: 100_000, + max_concurrent_workers: 1, + }, + test_plans: vec![LifecycleTestPlanBinding { plan, certificate_digest }], + }, + ) + .unwrap(); + (path, store, change_id, projection) +} + +#[test] +fn full_parent_owned_journal_reaches_apply_only_after_current_user_approval() { + let name = "full-journal"; + let (path, store, change_id, mut projection) = lifecycle_fixture(name); + let source = projection.lifecycle.source_snapshot; + let exploration_id = seed_exploration_certificate(&store, name, source); + projection = store + .parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteExplore { + worker: worker(&projection.lifecycle.spec.profiles.explore), + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: exploration_id, + source_snapshot: source, + }], + }, + ) + .unwrap(); + + let request = ChangeRequest { + task: "Implement the bounded lifecycle fixture.".to_owned(), + acceptance_criteria: vec!["The lifecycle remains fail closed.".to_owned()], + allowed_paths: vec![AllowedPath { + path: "fixture.txt".to_owned(), + scope: AllowedPathScope::Exact, + }], + artifact_ids: Vec::new(), + claim_ids: Vec::new(), + constraints: Vec::new(), + }; + let before = b"before\n".to_vec(); + let after = b"after\n".to_vec(); + let files = vec![PatchFile { + path: "fixture.txt".to_owned(), + operation: PatchOperation::Update, + before_digest: Some(Digest::blake3(&before)), + after_digest: Some(Digest::blake3(&after)), + before_bytes: before.len() as u64, + after_bytes: after.len() as u64, + }]; + let patch_id = PatchArtifact::compute_id(source, &files); + let patch = PatchArtifact { + id: patch_id, + change_id: change_id.clone(), + revision: 1, + source_snapshot: source, + files, + summary: "Bounded lifecycle patch".to_owned(), + acceptance_coverage: vec![AcceptanceCoverage { + criterion: request.acceptance_criteria[0].clone(), + status: AcceptanceStatus::Addressed, + evidence: "fixture patch".to_owned(), + }], + residual_risks: Vec::new(), + declared_output_digest: Digest::blake3(b"declared-output"), + discrepancies: Vec::new(), + }; + let repository_id = Digest::blake3(format!("{name}:repository")); + store + .record_prepared_change_with_provenance( + repository_id, + request.digest(source), + &request, + &patch, + &serde_json::json!({"summary": "Bounded lifecycle patch"}), + &[PatchFileBlob { + path: "fixture.txt".to_owned(), + before: Some(before), + after: Some(after), + }], + Some(&projection.lifecycle.spec.profiles.implement), + ) + .unwrap(); + projection = store + .parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteImplement { + worker: worker(&projection.lifecycle.spec.profiles.implement), + patch: LifecyclePatchRef { patch_id, revision: 1 }, + }, + ) + .unwrap(); + + let plan = projection.lifecycle.spec.test_plans[0].clone(); + let evidence = CommandExecutionEvidence { + id: "lifecycle-focused-evidence".to_owned(), + approval_id: "parent-selected-test".to_owned(), + argv: plan.plan.argv.clone(), + cwd: plan.plan.cwd_relative.clone(), + source_snapshot_digest: source, + runner: plan.plan.runner.clone(), + runner_version: None, + exit_status: Some(0), + duration_ms: 1, + output_digest: Digest::blake3(b"focused-test-output"), + output_preview: "test focused ... ok\ntest result: ok. 1 passed".to_owned(), + test_identifier: Some(plan.plan.test_identifier.clone()), + tests_executed: Some(1), + infrastructure_failure: None, + }; + store.record_command_evidence(None, &evidence).unwrap(); + projection = store + .parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteTest { + worker: worker(&projection.lifecycle.spec.profiles.test), + results: vec![LifecycleTestResult { + plan_digest: plan.plan_digest(), + certificate_digest: plan.certificate_digest, + available: true, + executed: true, + passed: true, + evidence_id: Some(evidence.id.clone()), + failure_code: None, + }], + }, + ) + .unwrap(); + + let criterion = &request.acceptance_criteria[0]; + let review = ReviewArtifact::new( + change_id.clone(), + patch_id, + LifecycleReviewVerdict::Approved, + vec![LifecycleAcceptanceReview::new( + criterion.as_bytes(), + AcceptanceStatus::Addressed, + b"review evidence", + )], + Vec::new(), + projection.lifecycle.spec.profiles.review.definition_digest, + projection.lifecycle.updated_unix_ms, + ) + .unwrap(); + projection = store + .parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteReview { + worker: worker(&projection.lifecycle.spec.profiles.review), + review, + }, + ) + .unwrap(); + + let raw_coverage = vec![AcceptanceCoverage { + criterion: criterion.clone(), + status: AcceptanceStatus::Addressed, + evidence: "verified in a fresh checkout".to_owned(), + }]; + let plan_result = VerificationPlanResult { + plan_digest: plan.plan_digest(), + runner: plan.plan.runner.clone(), + argv: plan.plan.argv.clone(), + cwd_relative: plan.plan.cwd_relative.clone(), + test_identifier: plan.plan.test_identifier.clone(), + expected: true, + available: true, + executed: true, + passed: true, + evidence_id: Some(evidence.id.clone()), + failure_reason: None, + }; + let evidence_ids = vec![evidence.id.clone()]; + let plan_results = vec![plan_result]; + let verifier_definition = projection.lifecycle.spec.profiles.verify.definition_digest; + let verification_id = VerificationArtifact::compute_id_with_plan_results( + &change_id, + patch_id, + VerificationStatus::Verified, + &raw_coverage, + &[], + VerificationTestProjection { + evidence_ids: &evidence_ids, + plan_results: &plan_results, + plans_over_cap: false, + }, + verifier_definition, + ); + let verification = VerificationArtifact { + id: verification_id, + change_id: change_id.clone(), + patch_id, + verdict: VerificationStatus::Verified, + acceptance_coverage: raw_coverage, + findings: Vec::new(), + test_evidence_ids: evidence_ids, + test_plan_results: plan_results, + test_plans_over_cap: false, + verifier_definition, + created_unix_ms: projection.lifecycle.updated_unix_ms, + }; + assert!(verification.is_canonical()); + store + .record_verification_artifact_with_provenance( + &verification, + &serde_json::json!({"fresh_checkout": true}), + &serde_json::json!({}), + None, + Some(&projection.lifecycle.spec.profiles.verify), + ) + .unwrap(); + projection = store + .parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteVerify { + worker: worker(&projection.lifecycle.spec.profiles.verify), + verification: LifecycleVerificationRef { + verification_id, + patch_id, + verdict: VerificationStatus::Verified, + }, + }, + ) + .unwrap(); + assert_eq!(projection.lifecycle.status, LifecycleStatus::AwaitingApproval); + + let mut apply_time = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64 + 1; + let mut apply = ChangeApplyRecord { + id: ChangeApplyId(Digest::blake3(b"full-journal-apply")), + change_id: change_id.clone(), + patch_id, + repository_root: "C:/bounded/repository".to_owned(), + pre_snapshot: source, + post_snapshot: None, + status: ChangeApplyStatus::Applying, + created_unix_ms: apply_time, + completed_unix_ms: None, + }; + let change_digest = store.change_digest(&change_id).unwrap().unwrap(); + assert!(matches!( + store.begin_change_apply(&apply, &serde_json::json!({}), change_digest), + Err(StoreError::LifecycleConflict(_)) + )); + projection = store + .approve_lifecycle_apply( + &change_id, + projection.state_digest, + ApprovalDecisionSource::WebUser, + ) + .unwrap(); + assert_eq!(projection.lifecycle.status, LifecycleStatus::Approved); + apply_time = projection.lifecycle.updated_unix_ms + 1; + apply.created_unix_ms = apply_time; + assert!(matches!( + store.begin_change_apply_with_lifecycle( + &apply, + &serde_json::json!({}), + change_digest, + Some(Digest::blake3(b"stale-lifecycle")), + ), + Err(StoreError::LifecycleConflict(_)) + )); + assert_eq!( + store.lifecycle(&change_id).unwrap().unwrap().lifecycle.status, + LifecycleStatus::Approved + ); + let barrier = Arc::new(Barrier::new(2)); + let handles = (0..2) + .map(|_| { + let path = path.clone(); + let apply = apply.clone(); + let barrier = Arc::clone(&barrier); + let lifecycle_digest = projection.state_digest; + thread::spawn(move || { + let store = RuntimeStore::new(path); + store.initialize().unwrap(); + barrier.wait(); + store.begin_change_apply_with_lifecycle( + &apply, + &serde_json::json!({}), + change_digest, + Some(lifecycle_digest), + ) + }) + }) + .collect::>(); + let outcomes = handles.into_iter().map(|handle| handle.join().unwrap()).collect::>(); + assert_eq!(outcomes.iter().filter(|outcome| outcome.is_ok()).count(), 1); + assert_eq!(outcomes.iter().filter(|outcome| outcome.is_err()).count(), 1); + assert_eq!( + store.lifecycle(&change_id).unwrap().unwrap().lifecycle.status, + LifecycleStatus::Applying + ); + store + .finish_change_apply( + apply.id, + ChangeApplyStatus::Applied, + Some(Digest::blake3(b"post-apply")), + apply_time + 1, + ) + .unwrap(); + let replayed = store.replay_lifecycle(&change_id).unwrap(); + assert_eq!(replayed.lifecycle.status, LifecycleStatus::Completed); + assert_eq!( + replayed.lifecycle.terminal_outcome, + Some(needle_core::LifecycleTerminalOutcome::Applied) + ); + assert!(matches!( + store.record_change_failure(&change_id, "late preparation failure"), + Err(StoreError::LifecycleConflict(_)) + )); + assert_eq!( + store.replay_lifecycle(&change_id).unwrap().lifecycle.status, + LifecycleStatus::Completed + ); + + drop(store); + fs::remove_file(path).unwrap(); +} + +#[test] +fn creation_rejects_stale_test_plan_source_and_inactive_profiles() { + let name = "creation-anchors"; + let (path, store, _change_id, projection) = lifecycle_fixture(name); + let request = ChangeRequest { + task: "Implement the bounded lifecycle fixture.".to_owned(), + acceptance_criteria: vec!["The lifecycle remains fail closed.".to_owned()], + allowed_paths: vec![AllowedPath { + path: "fixture.txt".to_owned(), + scope: AllowedPathScope::Exact, + }], + artifact_ids: Vec::new(), + claim_ids: Vec::new(), + constraints: Vec::new(), + }; + let stale_source = Digest::blake3(b"different-source"); + let stale_change = ChangeId::from_digest(Digest::blake3(b"stale-plan-change")); + store + .record_change_request_with_provenance( + &stale_change, + Digest::blake3(b"creation-anchors-repository"), + stale_source, + request.digest(stale_source), + &request, + Some(&projection.lifecycle.spec.profiles.implement), + ) + .unwrap(); + assert!(matches!( + store.create_lifecycle(&stale_change, projection.lifecycle.spec.clone()), + Err(StoreError::LifecycleConflict(_)) + )); + assert!(store.lifecycle(&stale_change).unwrap().is_none()); + + let explore_profile = &projection.lifecycle.spec.profiles.explore; + let profile_state = store.role_profile_state(&explore_profile.profile_id).unwrap(); + store.deactivate_role_profile(&explore_profile.profile_id, profile_state.state_digest).unwrap(); + let inactive_change = ChangeId::from_digest(Digest::blake3(b"inactive-profile-change")); + let source = projection.lifecycle.source_snapshot; + store + .record_change_request_with_provenance( + &inactive_change, + Digest::blake3(b"creation-anchors-repository"), + source, + request.digest(source), + &request, + Some(&projection.lifecycle.spec.profiles.implement), + ) + .unwrap(); + assert!(matches!( + store.create_lifecycle(&inactive_change, projection.lifecycle.spec), + Err(StoreError::LifecycleConflict(_)) + )); + assert!(store.lifecycle(&inactive_change).unwrap().is_none()); + + drop(store); + fs::remove_file(path).unwrap(); +} + +#[test] +fn exploration_rejects_test_plan_and_stale_source_artifacts_without_mutation() { + let name = "exploration-typing"; + let (path, store, change_id, projection) = lifecycle_fixture(name); + let completion = worker(&projection.lifecycle.spec.profiles.explore); + let connection = store.connection().unwrap(); + let test_plan_artifact: String = connection + .query_row( + "SELECT a.artifact_id FROM artifacts a + JOIN artifact_requests r ON r.request_id=a.request_id + WHERE r.logical_id=?1", + [format!("{name}:test-plan")], + |row| row.get(0), + ) + .unwrap(); + let test_plan_artifact = Digest::parse(&test_plan_artifact).unwrap(); + drop(connection); + assert!(matches!( + store.parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteExplore { + worker: completion.clone(), + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: test_plan_artifact, + source_snapshot: projection.lifecycle.source_snapshot, + }], + }, + ), + Err(StoreError::LifecycleConflict(_)) + )); + let stale_artifact = seed_exploration_certificate( + &store, + "exploration-typing-stale", + Digest::blake3(b"stale-source"), + ); + assert!(matches!( + store.parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteExplore { + worker: completion, + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: stale_artifact, + source_snapshot: projection.lifecycle.source_snapshot, + }], + }, + ), + Err(StoreError::LifecycleConflict(_)) + )); + assert_eq!(store.lifecycle(&change_id).unwrap().unwrap(), projection); + assert_eq!(store.lifecycle_events(&change_id).unwrap().len(), 1); + + drop(store); + fs::remove_file(path).unwrap(); +} + +#[test] +fn projection_and_event_payload_digests_are_verified_on_read() { + let (path, store, change_id, projection) = lifecycle_fixture("stored-digests"); + let connection = store.connection().unwrap(); + connection.execute("DROP TRIGGER change_lifecycles_transition_shape", []).unwrap(); + connection + .execute( + "UPDATE change_lifecycles SET state_digest=?2 WHERE change_id=?1", + params![change_id.to_string(), Digest::blake3(b"tampered-state").to_string()], + ) + .unwrap(); + drop(connection); + assert!(matches!(store.lifecycle(&change_id), Err(StoreError::LifecycleCorruption(_)))); + + let connection = store.connection().unwrap(); + connection + .execute( + "UPDATE change_lifecycles SET state_digest=?2 WHERE change_id=?1", + params![change_id.to_string(), projection.state_digest.to_string()], + ) + .unwrap(); + connection.execute("DROP TRIGGER change_events_no_update", []).unwrap(); + connection + .execute( + "UPDATE change_events SET payload_digest=?2 + WHERE change_id=?1 AND lifecycle_sequence=0", + params![change_id.to_string(), Digest::blake3(b"tampered-event").to_string()], + ) + .unwrap(); + drop(connection); + assert!(matches!(store.lifecycle_events(&change_id), Err(StoreError::LifecycleCorruption(_)))); + + drop(store); + fs::remove_file(path).unwrap(); +} + +#[test] +fn concurrent_cancellation_has_one_winner_and_replays_after_restart() { + let (path, store, change_id, projection) = lifecycle_fixture("cancel-race"); + let barrier = Arc::new(Barrier::new(2)); + let expected_state_digest = projection.state_digest; + let transition = LifecycleTransition::Cancel { + reason: LifecycleReason::new("user_cancelled", b"bounded user cancellation").unwrap(), + }; + let handles = (0..2) + .map(|_| { + let path = path.clone(); + let change_id = change_id.clone(); + let barrier = Arc::clone(&barrier); + let transition = transition.clone(); + thread::spawn(move || { + let store = RuntimeStore::new(path); + store.initialize().unwrap(); + barrier.wait(); + store.parent_transition_lifecycle(&change_id, expected_state_digest, transition) + }) + }) + .collect::>(); + let outcomes = handles.into_iter().map(|handle| handle.join().unwrap()).collect::>(); + assert_eq!(outcomes.iter().filter(|outcome| outcome.is_ok()).count(), 1); + assert_eq!(outcomes.iter().filter(|outcome| outcome.is_err()).count(), 1); + drop(store); + + let restarted = RuntimeStore::new(&path); + restarted.initialize().unwrap(); + let replayed = restarted.replay_lifecycle(&change_id).unwrap(); + assert_eq!(replayed.lifecycle.status, LifecycleStatus::Cancelled); + assert_eq!(replayed.lifecycle.generation, 1); + assert_eq!(restarted.lifecycle_events(&change_id).unwrap().len(), 2); + + let connection = rusqlite::Connection::open(&path).unwrap(); + assert!( + connection + .execute( + "UPDATE change_events SET event_type='tampered' WHERE change_id=?1", + [change_id.to_string()], + ) + .is_err() + ); + assert!( + connection + .execute("DELETE FROM change_events WHERE change_id=?1", [change_id.to_string()]) + .is_err() + ); + assert!( + connection + .execute("DELETE FROM change_lifecycles WHERE change_id=?1", [change_id.to_string()],) + .is_err() + ); + drop(connection); + drop(restarted); + fs::remove_file(path).unwrap(); +} + +#[test] +fn unavailable_exploration_artifact_leaves_projection_unchanged() { + let (path, store, change_id, projection) = lifecycle_fixture("missing-artifact"); + let worker = LifecycleWorkerCompletion { + profile: projection.lifecycle.spec.profiles.explore.clone(), + worker_depth: 1, + logical_worker_spawns: 1, + usage: LifecycleUsage { worker_turns: 1, output_tokens: 10, cost_microusd: 10 }, + }; + let result = store.parent_transition_lifecycle( + &change_id, + projection.state_digest, + LifecycleTransition::CompleteExplore { + worker, + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"missing-artifact"), + source_snapshot: projection.lifecycle.source_snapshot, + }], + }, + ); + assert!(matches!(result, Err(StoreError::LifecycleConflict(_)))); + assert_eq!(store.lifecycle(&change_id).unwrap().unwrap(), projection); + assert_eq!(store.lifecycle_events(&change_id).unwrap().len(), 1); + drop(store); + fs::remove_file(path).unwrap(); +} diff --git a/crates/needle-runtime/src/store/role_profiles/tests.rs b/crates/needle-runtime/src/store/role_profiles/tests.rs index 06c59e3..0327860 100644 --- a/crates/needle-runtime/src/store/role_profiles/tests.rs +++ b/crates/needle-runtime/src/store/role_profiles/tests.rs @@ -90,7 +90,7 @@ fn migration_and_revision_lifecycle_are_atomic_and_immutable() { .unwrap() .collect::>() .unwrap(); - assert_eq!(versions, (1..=15).collect::>()); + assert_eq!(versions, (1..=16).collect::>()); for name in ["role_profiles", "role_profile_revisions", "role_profile_state", "role_profile_audit"] { @@ -550,7 +550,7 @@ fn inconsistent_state_pointers_fail_closed_without_historical_fallback() { } #[test] -fn v15_upgrades_a_valid_v14_database_without_attributing_legacy_rows() { +fn current_schema_upgrades_a_valid_v14_database_without_attributing_legacy_rows() { let (path, store) = temporary_store(); let connection = Connection::open(&path).unwrap(); let migrations = [ @@ -595,7 +595,7 @@ fn v15_upgrades_a_valid_v14_database_without_attributing_legacy_rows() { let version: u32 = connection .query_row("SELECT MAX(version) FROM schema_migrations", [], |row| row.get(0)) .unwrap(); - assert_eq!(version, 15); + assert_eq!(version, 16); let legacy: (Option, Option, Option) = connection .query_row( "SELECT role_profile_id, role_profile_revision, diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 82a6696..3c887c7 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -74,6 +74,42 @@ the worker's writable boundary. The verifier is a separate read-only worker. It receives the patched checkout, acceptance criteria, and certified test context, but no patcher transcript. +## Parent-owned development lifecycle + +Core defines an opt-in, depth-one state machine with exactly this worker order: + +```text +explore -> implement -> test -> review -> verify -> apply +``` + +The parent is the only transition authority. A lifecycle freezes the change +ID, source snapshot, active explorer/implementer/test-runner/reviewer/verifier +profile revisions, sorted certified test plans, cumulative budget, and +concurrency of one. Worker completions carry typed bounded data and cannot +create another worker or write a transition themselves. + +Review and verification are separate contracts. Review consumes the current +patch and redacted acceptance-criterion digests. Verification references a +canonical `VerificationArtifact` created by the distinct verifier profile; it +cannot be replaced by the review artifact or supplied with a patcher +transcript. Missing or unavailable test evidence fails closed, and only one +repair reservation may be consumed. + +Runtime stores the current projection in `change_lifecycles` and appends every +transition to the existing `change_events` journal in one SQLite transaction. +Worker artifacts are persisted and validated before the separate parent +transition; a crash between those steps leaves the lifecycle in its prior phase +rather than advancing without a reference. Repair and apply transitions share +the transaction that mutates their existing change-journal records. Projection +and event payload digests are checked on read, replay must reproduce the same +state, and compare-and-swap state digests serialize concurrent transitions. +Lifecycle apply additionally requires an explicit user approval bound to the +current patch, verification, and lifecycle digest. + +This layer is the durable orchestration contract, not the Codex lifecycle +executor or read UI. Those consumers remain separate and must use the typed +parent operations rather than receiving direct store capability. + ## Request flow ```text @@ -122,7 +158,8 @@ not an unbounded worker tree. SQLite stores immutable definitions, settings, sessions, needs, steps, artifacts, claims, dependencies, certificates, plans, attempts, approvals, -usage, economic observations, changes, verification, and apply journals. +usage, economic observations, changes, lifecycle projections and append-only +events, verification, and apply journals. Migrations are additive and checksummed. Existing migration text is immutable. Sessions retain their initial route set, prompt profile, grammar or transport diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 963119f..dbd19bf 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -56,6 +56,13 @@ and verification are exchanged as bounded typed artifacts rather than raw transcripts. Write-capable roles remain confined to disposable checkouts, and active-worktree mutation remains an explicit parent-owned action. +The durable contract and SQLite journal for this sequence are implemented and +offline validated. They freeze active role-profile revisions, certified test +plans, the source snapshot, cumulative budget, one repair allowance, review and +verifier provenance, and approval against the exact verified state digest. +Codex process supervision and the lifecycle read/timeline UI remain separate +pending slices; the contract alone does not launch a lifecycle worker. + ## Milestone 3: Other-host subagent configuration Add configuration-only interoperability before any non-Codex orchestration. diff --git a/docs/VERIFIED_CHANGES.md b/docs/VERIFIED_CHANGES.md index 9af4b60..fa51886 100644 --- a/docs/VERIFIED_CHANGES.md +++ b/docs/VERIFIED_CHANGES.md @@ -60,6 +60,30 @@ The verifier: Infrastructure failure or missing required evidence is `inconclusive`, never `verified`. +## Parent-owned lifecycle contract + +A change may opt into the durable depth-one sequence +`explore -> implement -> test -> review -> verify -> apply`. Creation is allowed +only before the first patch and binds the immutable change/source identity to +active role-profile revisions and parent-selected certified test plans from the +same source snapshot. + +The parent advances the lifecycle with the current state digest. Each worker +phase checks its frozen profile, depth-one completion, cumulative budget, and +typed persisted artifact. Review records redacted acceptance coverage and is +not verifier evidence. Verify accepts only the current canonical verification +artifact from the frozen verifier definition. Failed or unavailable tests, +missing artifacts, stale digests, skipped phases, duplicate completion, and a +second repair all fail closed without a partial transition. + +Lifecycle projection and append-only events share the existing change journal. +Every parent transition persists its projection and event atomically. Patch and +verification artifacts must already exist, so a crash before their parent +transition leaves the phase unchanged; repair and apply lifecycle transitions +are atomic with their change-journal mutation. Restart replay must equal the +stored projection. This is an offline runtime contract. It does not wire Codex +worker processes or add a lifecycle HTTP/UI surface. + ## One repair The first `repairable` verdict may reserve one repair transactionally. A new @@ -80,6 +104,9 @@ MCP deliberately has no apply tool. The local control plane requires: 5. exact `If-Match` change digest; 6. active source snapshot equal to the preparation base. +For a lifecycle-bound change, apply also requires the exact approved lifecycle +state digest. The legacy apply entry point cannot bypass that requirement. + Apply operations are serialized and journaled before the first write. Failure restores persisted before blobs and verifies the pre-apply snapshot. Drift, replay, stale verification, symlink, path conflict, or recovery ambiguity