From ad67fa1168a0e9489d0ba8440550cd653e2b2f94 Mon Sep 17 00:00:00 2001 From: Punisheroot <44579963+Punisheroot@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:40:04 +0200 Subject: [PATCH] feat(runtime): add lifecycle execution kernel --- PROJECT_STATUS.md | 14 +- crates/needle-runtime/src/lib.rs | 22 +- .../needle-runtime/src/lifecycle_executor.rs | 721 +++++++++++++++++ .../src/lifecycle_executor/tests.rs | 725 ++++++++++++++++++ crates/needle-runtime/src/store.rs | 4 +- crates/needle-runtime/src/store/changes.rs | 103 +++ docs/ARCHITECTURE.md | 11 +- docs/ROADMAP.md | 13 +- docs/VERIFIED_CHANGES.md | 6 +- 9 files changed, 1594 insertions(+), 25 deletions(-) create mode 100644 crates/needle-runtime/src/lifecycle_executor.rs create mode 100644 crates/needle-runtime/src/lifecycle_executor/tests.rs diff --git a/PROJECT_STATUS.md b/PROJECT_STATUS.md index 9aaf734..cc04e5d 100644 --- a/PROJECT_STATUS.md +++ b/PROJECT_STATUS.md @@ -29,8 +29,8 @@ one. | Multi-need and structured MCP | Sequential and steer delivery, bounded ledger, structured JSON tools, cancellation, shared resolver | App Server simulator and one structured MCP cache-hit observation | **Implemented; offline validated; live calibration** | | 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 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** | +| 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; lifecycle execution uses only injected adapters, not concrete Codex processes | **Implemented; offline validated** | +| 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, journaled apply integration, and a parent-owned injected-adapter execution kernel; concrete Codex adapters and lifecycle UI are not wired | Deterministic transition matrix, adversarial domain/store tests, fake-adapter execution, repair, cancellation, stale-CAS, and restart tests 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 lifecycle domain and change-journal integration | **Implemented; offline validated; executor pending** | +| Parent-owned lifecycle domain, change-journal integration, and injected-adapter kernel | **Implemented; offline validated; concrete adapters 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** | @@ -127,10 +127,10 @@ validation. - Verified changes have no provider-backed patcher or verifier observation. - Canonical role-profile definitions, revision persistence, bounded HTTP/editor 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. + provenance, and the parent-owned lifecycle domain/journal and injected-adapter + kernel are implemented and offline validated. No concrete adapter launches or + supervises Codex workers, and there is 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-runtime/src/lib.rs b/crates/needle-runtime/src/lib.rs index 9398a48..3d5e562 100644 --- a/crates/needle-runtime/src/lib.rs +++ b/crates/needle-runtime/src/lib.rs @@ -4,6 +4,7 @@ mod approval; mod artifact_cache; mod changes; mod claim_proof; +mod lifecycle_executor; mod model_ladder; mod orchestrator; mod proof; @@ -25,12 +26,12 @@ pub use snapshot::{ validate_need_result, }; pub use store::{ - 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, + CacheRecord, ChangeAttemptRecord, ConfigExport, LifecycleChangeContext, LifecycleProjection, + MainTurnObservationRecord, NeedShadowRecord, NeedShadowWrite, NeedStepEventRecord, + NeedStepRequestRecord, NegativeAttemptRecord, OperatorCostKey, OperatorCostObservation, + PatchFileBlob, PreparedChangeRecord, ProofAccountingRecord, RoleProfileAuditOperation, + RoleProfileAuditRecord, RoleProfileStateRecord, RouteCostObservation, RoutePromotionRecord, + RuntimeSettings, RuntimeStore, SessionRecord, StoreError, WorkerRunRecord, }; use needle_core::{NeedKey, Preset, Route, RouteMatcher}; @@ -89,6 +90,15 @@ pub use claim_proof::{ build_claim_set_certificate, claim_proof_engine_definition, claim_validation_certificate_is_fresh, replay_claim_set_certificate, }; +pub use lifecycle_executor::{ + LIFECYCLE_ADAPTER_OUTCOME_SCHEMA, LIFECYCLE_ADAPTER_REQUEST_SCHEMA, LifecycleAdapterCleanup, + LifecycleAdapterFailure, LifecycleAdapterResult, LifecycleCancellation, + LifecycleExecutionDisposition, LifecycleExecutionError, LifecycleExecutionKernel, + LifecycleExecutionOutcome, LifecyclePhaseAdapter, LifecyclePhaseAdapterOutcome, + LifecyclePhaseAdapterRequest, LifecyclePhaseAdapters, LifecycleRemainingBudget, + MAX_LIFECYCLE_ADAPTER_DETAIL_BYTES, MAX_LIFECYCLE_ADAPTER_OUTCOME_BYTES, + MAX_LIFECYCLE_ADAPTER_REQUEST_BYTES, NeverCancel, +}; pub use model_ladder::{ LadderAttempt, LadderAttemptRecord, ModelLadder, ModelLadderError, ModelLadderOutcome, }; diff --git a/crates/needle-runtime/src/lifecycle_executor.rs b/crates/needle-runtime/src/lifecycle_executor.rs new file mode 100644 index 0000000..54ece8a --- /dev/null +++ b/crates/needle-runtime/src/lifecycle_executor.rs @@ -0,0 +1,721 @@ +use crate::{LifecycleChangeContext, LifecycleProjection, RuntimeStore, StoreError}; +use needle_core::{ + CanonicalHasher, ChangeId, DevelopmentLifecycle, Digest, LifecyclePhase, LifecycleReason, + LifecycleStatus, LifecycleTransition, LifecycleUsage, PatchId, RoleProfileProvenance, + VerificationArtifact, VerificationStatus, +}; +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +pub const LIFECYCLE_ADAPTER_REQUEST_SCHEMA: &str = "needle.lifecycle-adapter-request/1"; +pub const LIFECYCLE_ADAPTER_OUTCOME_SCHEMA: &str = "needle.lifecycle-adapter-outcome/1"; +pub const MAX_LIFECYCLE_ADAPTER_REQUEST_BYTES: usize = 128 * 1024; +pub const MAX_LIFECYCLE_ADAPTER_OUTCOME_BYTES: usize = 128 * 1024; +pub const MAX_LIFECYCLE_ADAPTER_DETAIL_BYTES: usize = 1024; + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleRemainingBudget { + pub worker_turns: u32, + pub output_tokens: u64, + pub cost_microusd: u64, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecyclePhaseAdapterRequest { + pub schema: String, + pub invocation_id: Digest, + pub expected_state_digest: Digest, + pub phase: LifecyclePhase, + pub profile: RoleProfileProvenance, + pub remaining_budget: LifecycleRemainingBudget, + pub lifecycle: DevelopmentLifecycle, + pub change: LifecycleChangeContext, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub repair_verification: Option, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleAdapterFailure { + pub code: String, + pub detail: String, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleAdapterCleanup { + pub succeeded: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure: Option, +} + +impl LifecycleAdapterCleanup { + pub fn succeeded() -> Self { + Self { succeeded: true, failure: None } + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum LifecycleAdapterResult { + Completed { transition: Box }, + Failed { failure: LifecycleAdapterFailure }, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecyclePhaseAdapterOutcome { + pub schema: String, + pub invocation_id: Digest, + pub expected_state_digest: Digest, + pub phase: LifecyclePhase, + pub usage: LifecycleUsage, + pub result: LifecycleAdapterResult, + pub cleanup: LifecycleAdapterCleanup, +} + +/// A phase adapter is invoked with a deterministic `invocation_id`. Concrete +/// adapters must make that identity idempotent: repeating the same request may +/// return the recorded outcome, but must not repeat its external side effect. +/// The adapter receives no lifecycle-store transition capability. +pub trait LifecyclePhaseAdapter { + fn invoke(&self, request: &LifecyclePhaseAdapterRequest) -> LifecyclePhaseAdapterOutcome; +} + +#[derive(Clone, Copy)] +pub struct LifecyclePhaseAdapters<'a> { + pub explore: &'a dyn LifecyclePhaseAdapter, + pub implement: &'a dyn LifecyclePhaseAdapter, + pub test: &'a dyn LifecyclePhaseAdapter, + pub review: &'a dyn LifecyclePhaseAdapter, + pub verify: &'a dyn LifecyclePhaseAdapter, +} + +impl LifecyclePhaseAdapters<'_> { + fn for_phase(&self, phase: LifecyclePhase) -> Option<&dyn LifecyclePhaseAdapter> { + 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 trait LifecycleCancellation { + fn is_cancelled(&self, change_id: &ChangeId, expected_state_digest: Digest) -> bool; +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct NeverCancel; + +impl LifecycleCancellation for NeverCancel { + fn is_cancelled(&self, _change_id: &ChangeId, _expected_state_digest: Digest) -> bool { + false + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LifecycleExecutionDisposition { + Advanced, + RepairResumed, + AwaitingApproval, + Terminal, + Cancelled, + BudgetExhausted, + AdapterFailed, + CleanupFailed, + InvalidAdapterOutput, + StaleState, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleExecutionOutcome { + pub disposition: LifecycleExecutionDisposition, + pub projection: LifecycleProjection, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub invocation_id: Option, + pub adapter_usage: LifecycleUsage, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cleanup: Option, +} + +#[derive(Debug, Error)] +pub enum LifecycleExecutionError { + #[error(transparent)] + Store(#[from] StoreError), + #[error("lifecycle adapter contract could not be serialized")] + ContractSerialization, +} + +pub struct LifecycleExecutionKernel<'a> { + store: &'a RuntimeStore, + adapters: LifecyclePhaseAdapters<'a>, + cancellation: &'a dyn LifecycleCancellation, +} + +impl<'a> LifecycleExecutionKernel<'a> { + pub fn new( + store: &'a RuntimeStore, + adapters: LifecyclePhaseAdapters<'a>, + cancellation: &'a dyn LifecycleCancellation, + ) -> Self { + Self { store, adapters, cancellation } + } + + pub fn step( + &self, + change_id: &ChangeId, + ) -> Result { + execute_step(self.store, &self.adapters, self.cancellation, change_id) + } +} + +trait LifecycleKernelStore { + fn replay(&self, change_id: &ChangeId) -> Result; + fn change_context( + &self, + change_id: &ChangeId, + ) -> Result, StoreError>; + fn latest_verification( + &self, + change_id: &ChangeId, + ) -> Result, StoreError>; + fn parent_transition( + &self, + change_id: &ChangeId, + expected_state_digest: Digest, + transition: LifecycleTransition, + ) -> Result; + fn begin_repair(&self, change_id: &ChangeId, patch_id: PatchId) -> Result<(), StoreError>; +} + +impl LifecycleKernelStore for RuntimeStore { + fn replay(&self, change_id: &ChangeId) -> Result { + self.replay_lifecycle(change_id) + } + + fn change_context( + &self, + change_id: &ChangeId, + ) -> Result, StoreError> { + self.lifecycle_change_context(change_id) + } + + fn latest_verification( + &self, + change_id: &ChangeId, + ) -> Result, StoreError> { + self.latest_verification_artifact(change_id) + } + + fn parent_transition( + &self, + change_id: &ChangeId, + expected_state_digest: Digest, + transition: LifecycleTransition, + ) -> Result { + self.parent_transition_lifecycle(change_id, expected_state_digest, transition) + } + + fn begin_repair(&self, change_id: &ChangeId, patch_id: PatchId) -> Result<(), StoreError> { + self.begin_change_repair(change_id, patch_id) + } +} + +fn execute_step( + store: &S, + adapters: &LifecyclePhaseAdapters<'_>, + cancellation: &dyn LifecycleCancellation, + change_id: &ChangeId, +) -> Result { + let projection = store.replay(change_id)?; + if projection.lifecycle.status.terminal() { + return Ok(outcome(LifecycleExecutionDisposition::Terminal, projection)); + } + if projection.lifecycle.phase == LifecyclePhase::Apply { + return Ok(outcome(LifecycleExecutionDisposition::AwaitingApproval, projection)); + } + let expected_state_digest = projection.state_digest; + if cancellation.is_cancelled(change_id, expected_state_digest) { + return commit_terminal( + store, + change_id, + projection, + LifecycleExecutionDisposition::Cancelled, + LifecycleTransition::Cancel { + reason: fixed_reason("cancelled", b"lifecycle cancelled before adapter invocation"), + }, + None, + LifecycleUsage::default(), + None, + ); + } + if projection.lifecycle.status == LifecycleStatus::RepairReserved { + return resume_repair(store, change_id, projection); + } + if projection.lifecycle.status != LifecycleStatus::Active { + return Ok(outcome(LifecycleExecutionDisposition::Terminal, projection)); + } + let remaining_budget = remaining_budget(&projection.lifecycle); + if remaining_budget.worker_turns == 0 + || remaining_budget.output_tokens == 0 + || remaining_budget.cost_microusd == 0 + { + return commit_terminal( + store, + change_id, + projection, + LifecycleExecutionDisposition::BudgetExhausted, + LifecycleTransition::Fail { + reason: fixed_reason("budget_exhausted", b"lifecycle aggregate budget exhausted"), + }, + None, + LifecycleUsage::default(), + None, + ); + } + let Some(profile) = + projection.lifecycle.spec.profiles.for_phase(projection.lifecycle.phase).cloned() + else { + return invalid_output( + store, + change_id, + projection, + None, + LifecycleUsage::default(), + None, + b"active lifecycle phase has no frozen profile", + ); + }; + let Some(change) = store.change_context(change_id)? else { + return Err(StoreError::LifecycleNotFound(format!( + "{change_id}: immutable change request" + )) + .into()); + }; + if change.source_snapshot != projection.lifecycle.source_snapshot { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: lifecycle and request source differ" + )) + .into()); + } + let repair_verification = repair_verification(store, &projection)?; + let invocation_id = invocation_id(&projection); + let request = LifecyclePhaseAdapterRequest { + schema: LIFECYCLE_ADAPTER_REQUEST_SCHEMA.to_owned(), + invocation_id, + expected_state_digest, + phase: projection.lifecycle.phase, + profile, + remaining_budget, + lifecycle: projection.lifecycle.clone(), + change, + repair_verification, + }; + if serde_json::to_vec(&request) + .map_err(|_| LifecycleExecutionError::ContractSerialization)? + .len() + > MAX_LIFECYCLE_ADAPTER_REQUEST_BYTES + { + return invalid_output( + store, + change_id, + projection, + Some(invocation_id), + LifecycleUsage::default(), + None, + b"lifecycle adapter request exceeds the byte bound", + ); + } + let adapter = + adapters.for_phase(request.phase).expect("active worker phases have an injected adapter"); + let adapter_outcome = adapter.invoke(&request); + finish_adapter_step(store, cancellation, change_id, projection, request, adapter_outcome) +} + +fn resume_repair( + store: &S, + change_id: &ChangeId, + projection: LifecycleProjection, +) -> Result { + if projection.lifecycle.phase != LifecyclePhase::Verify { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: repair is outside verify" + )) + .into()); + } + let patch_id = projection + .lifecycle + .patch + .as_ref() + .map(|patch| patch.patch_id) + .ok_or_else(|| StoreError::LifecycleCorruption(format!("{change_id}: repair patch")))?; + match store.begin_repair(change_id, patch_id) { + Ok(()) => { + let resumed = store.replay(change_id)?; + Ok(outcome(LifecycleExecutionDisposition::RepairResumed, resumed)) + } + Err(error @ (StoreError::ChangeConflict(_) | StoreError::LifecycleConflict(_))) => { + let current = store.replay(change_id)?; + if current.state_digest != projection.state_digest { + Ok(outcome(LifecycleExecutionDisposition::StaleState, current)) + } else { + Err(error.into()) + } + } + Err(error) => Err(error.into()), + } +} + +fn finish_adapter_step( + store: &S, + cancellation: &dyn LifecycleCancellation, + change_id: &ChangeId, + projection: LifecycleProjection, + request: LifecyclePhaseAdapterRequest, + adapter_outcome: LifecyclePhaseAdapterOutcome, +) -> Result { + let serialized = serde_json::to_vec(&adapter_outcome) + .map_err(|_| LifecycleExecutionError::ContractSerialization)?; + if serialized.len() > MAX_LIFECYCLE_ADAPTER_OUTCOME_BYTES + || !adapter_outcome_matches(&request, &adapter_outcome) + || !usage_within(&adapter_outcome.usage, &request.remaining_budget) + { + return invalid_output( + store, + change_id, + projection, + Some(request.invocation_id), + adapter_outcome.usage, + Some(adapter_outcome.cleanup), + b"adapter outcome identity, shape, or bounds are invalid", + ); + } + if !adapter_outcome.cleanup.succeeded { + let detail = adapter_outcome + .cleanup + .failure + .as_ref() + .map(problem_detail) + .unwrap_or_else(|| b"adapter cleanup failed without detail".to_vec()); + return commit_terminal( + store, + change_id, + projection, + LifecycleExecutionDisposition::CleanupFailed, + LifecycleTransition::Fail { reason: fixed_reason("adapter_cleanup_failed", detail) }, + Some(request.invocation_id), + adapter_outcome.usage, + Some(adapter_outcome.cleanup), + ); + } + if cancellation.is_cancelled(change_id, request.expected_state_digest) { + return commit_terminal( + store, + change_id, + projection, + LifecycleExecutionDisposition::Cancelled, + LifecycleTransition::Cancel { + reason: fixed_reason("cancelled", b"lifecycle cancelled after adapter invocation"), + }, + Some(request.invocation_id), + adapter_outcome.usage, + Some(adapter_outcome.cleanup), + ); + } + match adapter_outcome.result { + LifecycleAdapterResult::Failed { failure } => commit_terminal( + store, + change_id, + projection, + LifecycleExecutionDisposition::AdapterFailed, + LifecycleTransition::Fail { + reason: LifecycleReason::new(failure.code, failure.detail.as_bytes()) + .expect("validated adapter failure has a bounded reason"), + }, + Some(request.invocation_id), + adapter_outcome.usage, + Some(adapter_outcome.cleanup), + ), + LifecycleAdapterResult::Completed { transition } => { + if !completion_matches(&request, &transition, &adapter_outcome.usage) { + return invalid_output( + store, + change_id, + projection, + Some(request.invocation_id), + adapter_outcome.usage, + Some(adapter_outcome.cleanup), + b"adapter completion differs from the active phase, profile, or budget", + ); + } + commit_completion( + store, + change_id, + projection, + request.invocation_id, + *transition, + adapter_outcome.usage, + adapter_outcome.cleanup, + ) + } + } +} + +fn commit_completion( + store: &S, + change_id: &ChangeId, + projection: LifecycleProjection, + invocation_id: Digest, + transition: LifecycleTransition, + usage: LifecycleUsage, + cleanup: LifecycleAdapterCleanup, +) -> Result { + match store.parent_transition(change_id, projection.state_digest, transition) { + Ok(next) => { + let disposition = if next.lifecycle.status.terminal() { + LifecycleExecutionDisposition::Terminal + } else if next.lifecycle.phase == LifecyclePhase::Apply + && next.lifecycle.status == LifecycleStatus::AwaitingApproval + { + LifecycleExecutionDisposition::AwaitingApproval + } else { + LifecycleExecutionDisposition::Advanced + }; + Ok(LifecycleExecutionOutcome { + disposition, + projection: next, + invocation_id: Some(invocation_id), + adapter_usage: usage, + cleanup: Some(cleanup), + }) + } + Err(error @ (StoreError::Lifecycle(_) | StoreError::LifecycleConflict(_))) => { + let current = store.replay(change_id)?; + if current.state_digest != projection.state_digest { + return Ok(LifecycleExecutionOutcome { + disposition: LifecycleExecutionDisposition::StaleState, + projection: current, + invocation_id: Some(invocation_id), + adapter_usage: usage, + cleanup: Some(cleanup), + }); + } + invalid_output( + store, + change_id, + projection, + Some(invocation_id), + usage, + Some(cleanup), + error.to_string().as_bytes(), + ) + } + Err(error) => Err(error.into()), + } +} + +fn invalid_output( + store: &S, + change_id: &ChangeId, + projection: LifecycleProjection, + invocation_id: Option, + usage: LifecycleUsage, + cleanup: Option, + detail: impl AsRef<[u8]>, +) -> Result { + commit_terminal( + store, + change_id, + projection, + LifecycleExecutionDisposition::InvalidAdapterOutput, + LifecycleTransition::Fail { reason: fixed_reason("adapter_invalid_output", detail) }, + invocation_id, + usage, + cleanup, + ) +} + +#[allow(clippy::too_many_arguments)] +fn commit_terminal( + store: &S, + change_id: &ChangeId, + projection: LifecycleProjection, + disposition: LifecycleExecutionDisposition, + transition: LifecycleTransition, + invocation_id: Option, + usage: LifecycleUsage, + cleanup: Option, +) -> Result { + match store.parent_transition(change_id, projection.state_digest, transition) { + Ok(next) => Ok(LifecycleExecutionOutcome { + disposition, + projection: next, + invocation_id, + adapter_usage: usage, + cleanup, + }), + Err(error @ StoreError::LifecycleConflict(_)) => { + let current = store.replay(change_id)?; + if current.state_digest != projection.state_digest { + Ok(LifecycleExecutionOutcome { + disposition: LifecycleExecutionDisposition::StaleState, + projection: current, + invocation_id, + adapter_usage: usage, + cleanup, + }) + } else { + Err(error.into()) + } + } + Err(error) => Err(error.into()), + } +} + +fn adapter_outcome_matches( + request: &LifecyclePhaseAdapterRequest, + outcome: &LifecyclePhaseAdapterOutcome, +) -> bool { + outcome.schema == LIFECYCLE_ADAPTER_OUTCOME_SCHEMA + && outcome.invocation_id == request.invocation_id + && outcome.expected_state_digest == request.expected_state_digest + && outcome.phase == request.phase + && cleanup_valid(&outcome.cleanup) + && match &outcome.result { + LifecycleAdapterResult::Completed { .. } => true, + LifecycleAdapterResult::Failed { failure } => problem_valid(failure), + } +} + +fn cleanup_valid(cleanup: &LifecycleAdapterCleanup) -> bool { + cleanup.succeeded == cleanup.failure.is_none() + && cleanup.failure.as_ref().is_none_or(problem_valid) +} + +fn problem_valid(problem: &LifecycleAdapterFailure) -> bool { + !problem.code.is_empty() + && problem.code.len() <= needle_core::MAX_LIFECYCLE_REASON_CODE_BYTES + && problem + .code + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':')) + && problem.detail.len() <= MAX_LIFECYCLE_ADAPTER_DETAIL_BYTES +} + +fn completion_matches( + request: &LifecyclePhaseAdapterRequest, + transition: &LifecycleTransition, + usage: &LifecycleUsage, +) -> bool { + let phase_matches = matches!( + (request.phase, transition), + (LifecyclePhase::Explore, LifecycleTransition::CompleteExplore { .. }) + | (LifecyclePhase::Implement, LifecycleTransition::CompleteImplement { .. }) + | (LifecyclePhase::Test, LifecycleTransition::CompleteTest { .. }) + | (LifecyclePhase::Review, LifecycleTransition::CompleteReview { .. }) + | (LifecyclePhase::Verify, LifecycleTransition::CompleteVerify { .. }) + ); + let Some(worker) = transition.worker() else { + return false; + }; + phase_matches + && worker.profile == request.profile + && worker.worker_depth == 1 + && worker.logical_worker_spawns == 1 + && worker.usage == *usage + && usage.worker_turns > 0 +} + +fn usage_within(usage: &LifecycleUsage, remaining: &LifecycleRemainingBudget) -> bool { + usage.worker_turns <= remaining.worker_turns + && usage.output_tokens <= remaining.output_tokens + && usage.cost_microusd <= remaining.cost_microusd +} + +fn remaining_budget(lifecycle: &DevelopmentLifecycle) -> LifecycleRemainingBudget { + LifecycleRemainingBudget { + worker_turns: lifecycle + .spec + .budget + .max_worker_turns + .saturating_sub(lifecycle.usage.worker_turns), + output_tokens: lifecycle + .spec + .budget + .max_output_tokens + .saturating_sub(lifecycle.usage.output_tokens), + cost_microusd: lifecycle + .spec + .budget + .max_cost_microusd + .saturating_sub(lifecycle.usage.cost_microusd), + } +} + +fn repair_verification( + store: &S, + projection: &LifecycleProjection, +) -> Result, LifecycleExecutionError> { + if !projection.lifecycle.repair_consumed { + return Ok(None); + } + let verification = + store.latest_verification(&projection.lifecycle.change_id)?.ok_or_else(|| { + StoreError::LifecycleCorruption(format!( + "{}: consumed repair has no verification artifact", + projection.lifecycle.change_id + )) + })?; + if verification.change_id != projection.lifecycle.change_id + || verification.verdict != VerificationStatus::Repairable + || !verification.is_canonical() + { + return Err(StoreError::LifecycleCorruption(format!( + "{}: consumed repair verification is invalid", + projection.lifecycle.change_id + )) + .into()); + } + Ok(Some(verification)) +} + +fn invocation_id(projection: &LifecycleProjection) -> Digest { + let mut hasher = CanonicalHasher::new(b"needle-lifecycle-adapter-invocation"); + hasher.field_digest(projection.lifecycle.id.0); + hasher.field_digest(projection.state_digest); + hasher.field_str(projection.lifecycle.phase.as_str()); + hasher.finish() +} + +fn problem_detail(problem: &LifecycleAdapterFailure) -> Vec { + serde_json::to_vec(problem).unwrap_or_else(|_| b"invalid adapter failure detail".to_vec()) +} + +fn fixed_reason(code: &'static str, detail: impl AsRef<[u8]>) -> LifecycleReason { + LifecycleReason::new(code, detail).expect("fixed lifecycle reason is bounded") +} + +fn outcome( + disposition: LifecycleExecutionDisposition, + projection: LifecycleProjection, +) -> LifecycleExecutionOutcome { + LifecycleExecutionOutcome { + disposition, + projection, + invocation_id: None, + adapter_usage: LifecycleUsage::default(), + cleanup: None, + } +} + +#[cfg(test)] +#[path = "lifecycle_executor/tests.rs"] +mod tests; diff --git a/crates/needle-runtime/src/lifecycle_executor/tests.rs b/crates/needle-runtime/src/lifecycle_executor/tests.rs new file mode 100644 index 0000000..f78049c --- /dev/null +++ b/crates/needle-runtime/src/lifecycle_executor/tests.rs @@ -0,0 +1,725 @@ +use super::*; +use needle_core::{ + AcceptanceCoverage, AcceptanceStatus, ChangeRequest, LifecycleAcceptanceReview, + LifecycleArtifactKind, LifecycleArtifactRef, LifecycleBudget, LifecyclePatchRef, + LifecycleReviewVerdict, LifecycleSpec, LifecycleTestPlanBinding, LifecycleTestResult, + LifecycleVerificationRef, LifecycleWorkerCompletion, LifecycleWorkerProfiles, ReviewArtifact, + RoleProfileId, TestPlan, +}; +use std::{ + cell::{Cell, RefCell}, + collections::{BTreeMap, BTreeSet}, +}; + +fn profile(name: &str) -> RoleProfileProvenance { + RoleProfileProvenance::new( + RoleProfileId::new(name).unwrap(), + 1, + Digest::blake3(name.as_bytes()), + ) + .unwrap() +} + +fn lifecycle_spec(max_worker_turns: u32) -> LifecycleSpec { + LifecycleSpec { + worker_depth_limit: 1, + profiles: LifecycleWorkerProfiles { + explore: profile("kernel.explore"), + implement: profile("kernel.implement"), + test: profile("kernel.test"), + review: profile("kernel.review"), + verify: profile("kernel.verify"), + }, + budget: LifecycleBudget { + max_worker_turns, + max_output_tokens: 100, + max_cost_microusd: 100, + max_concurrent_workers: 1, + }, + test_plans: vec![LifecycleTestPlanBinding { + plan: TestPlan { + runner: "cargo".to_owned(), + argv: vec!["cargo".to_owned(), "test".to_owned(), "kernel".to_owned()], + cwd_relative: ".".to_owned(), + test_identifier: "kernel".to_owned(), + requires_approval: true, + execution_evidence_id: None, + }, + certificate_digest: Digest::blake3(b"kernel-test-certificate"), + }], + } +} + +fn context(change_id: &ChangeId, source_snapshot: Digest) -> LifecycleChangeContext { + let request = ChangeRequest { + task: format!("Execute {change_id} through the lifecycle kernel."), + acceptance_criteria: vec!["the bounded lifecycle completes".to_owned()], + allowed_paths: Vec::new(), + artifact_ids: Vec::new(), + claim_ids: Vec::new(), + constraints: Vec::new(), + }; + LifecycleChangeContext { + request_digest: request.digest(source_snapshot), + request, + repository_id: Digest::blake3(b"kernel-repository"), + source_snapshot, + } +} + +fn projection(lifecycle: DevelopmentLifecycle) -> LifecycleProjection { + LifecycleProjection { state_digest: lifecycle.state_digest(), lifecycle } +} + +struct FakeStore { + projection: RefCell, + context: LifecycleChangeContext, + latest_verification: RefCell>, + repair_calls: Cell, + fail_next_parent_commit: Cell, + concurrent_transition: RefCell>, + corrupt_replay: Cell, +} + +impl FakeStore { + fn new(max_worker_turns: u32) -> (Self, ChangeId) { + let change_id = ChangeId::from_digest(Digest::blake3(b"kernel-change")); + let source_snapshot = Digest::blake3(b"kernel-source"); + let lifecycle = DevelopmentLifecycle::new( + change_id.clone(), + source_snapshot, + lifecycle_spec(max_worker_turns), + 1, + ) + .unwrap(); + ( + Self { + projection: RefCell::new(projection(lifecycle)), + context: context(&change_id, source_snapshot), + latest_verification: RefCell::new(None), + repair_calls: Cell::new(0), + fail_next_parent_commit: Cell::new(false), + concurrent_transition: RefCell::new(None), + corrupt_replay: Cell::new(false), + }, + change_id, + ) + } + + fn apply(&self, transition: LifecycleTransition) -> Result { + let current = self.projection.borrow().clone(); + let (next, _) = + current.lifecycle.transition(transition, current.lifecycle.updated_unix_ms + 1)?; + let next = projection(next); + *self.projection.borrow_mut() = next.clone(); + Ok(next) + } +} + +impl LifecycleKernelStore for FakeStore { + fn replay(&self, change_id: &ChangeId) -> Result { + if self.corrupt_replay.get() { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: injected replay corruption" + ))); + } + Ok(self.projection.borrow().clone()) + } + + fn change_context( + &self, + _change_id: &ChangeId, + ) -> Result, StoreError> { + Ok(Some(self.context.clone())) + } + + fn latest_verification( + &self, + _change_id: &ChangeId, + ) -> Result, StoreError> { + Ok(self.latest_verification.borrow().clone()) + } + + fn parent_transition( + &self, + change_id: &ChangeId, + expected_state_digest: Digest, + transition: LifecycleTransition, + ) -> Result { + let current = self.projection.borrow().clone(); + if current.state_digest != expected_state_digest { + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: stale injected state" + ))); + } + if self.fail_next_parent_commit.replace(false) { + return Err(StoreError::ConnectionLock); + } + if let Some(concurrent) = self.concurrent_transition.borrow_mut().take() { + self.apply(concurrent)?; + return Err(StoreError::LifecycleConflict(format!( + "{change_id}: injected concurrent transition" + ))); + } + + let verification_artifact = match &transition { + LifecycleTransition::CompleteVerify { verification, .. } => { + let artifact = verification_artifact( + ¤t.lifecycle, + &self.context.request, + verification.verdict, + ); + if artifact.id != verification.verification_id { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: fake verification identity" + ))); + } + Some(artifact) + } + _ => None, + }; + let next = self.apply(transition)?; + if let Some(artifact) = verification_artifact { + *self.latest_verification.borrow_mut() = Some(artifact); + } + Ok(next) + } + + fn begin_repair(&self, change_id: &ChangeId, patch_id: PatchId) -> Result<(), StoreError> { + self.repair_calls.set(self.repair_calls.get() + 1); + let current = self.projection.borrow().clone(); + if current.lifecycle.patch.as_ref().map(|patch| patch.patch_id) != Some(patch_id) { + return Err(StoreError::ChangeConflict(format!( + "{change_id}: injected repair patch mismatch" + ))); + } + self.apply(LifecycleTransition::ConsumeRepair)?; + Ok(()) + } +} + +#[derive(Clone, Copy)] +enum FakeAdapterMode { + Success, + RepairOnce, + RejectVerification, + WrongPhase, + WrongProfile, + BadWorkerDepth, + Failure, + CleanupFailure, + OversizedFailure, +} + +struct FakeAdapter { + mode: FakeAdapterMode, + calls: Cell, + side_effects: RefCell>, + cached: RefCell>, + phases: RefCell>, + repair_contexts: Cell, +} + +impl FakeAdapter { + fn new(mode: FakeAdapterMode) -> Self { + Self { + mode, + calls: Cell::new(0), + side_effects: RefCell::new(BTreeSet::new()), + cached: RefCell::new(BTreeMap::new()), + phases: RefCell::new(Vec::new()), + repair_contexts: Cell::new(0), + } + } +} + +impl LifecyclePhaseAdapter for FakeAdapter { + fn invoke(&self, request: &LifecyclePhaseAdapterRequest) -> LifecyclePhaseAdapterOutcome { + self.calls.set(self.calls.get() + 1); + self.phases.borrow_mut().push(request.phase); + if request.repair_verification.is_some() { + self.repair_contexts.set(self.repair_contexts.get() + 1); + } + if let Some(outcome) = self.cached.borrow().get(&request.invocation_id) { + return outcome.clone(); + } + self.side_effects.borrow_mut().insert(request.invocation_id); + + let usage = LifecycleUsage { worker_turns: 1, output_tokens: 1, cost_microusd: 1 }; + let cleanup = match self.mode { + FakeAdapterMode::CleanupFailure => LifecycleAdapterCleanup { + succeeded: false, + failure: Some(LifecycleAdapterFailure { + code: "cleanup_failed".to_owned(), + detail: "the fake adapter could not clean up".to_owned(), + }), + }, + _ => LifecycleAdapterCleanup::succeeded(), + }; + let result = match self.mode { + FakeAdapterMode::Failure => LifecycleAdapterResult::Failed { + failure: LifecycleAdapterFailure { + code: "adapter_failed".to_owned(), + detail: "the fake adapter failed".to_owned(), + }, + }, + FakeAdapterMode::OversizedFailure => LifecycleAdapterResult::Failed { + failure: LifecycleAdapterFailure { + code: "adapter_failed".to_owned(), + detail: "x".repeat(MAX_LIFECYCLE_ADAPTER_DETAIL_BYTES + 1), + }, + }, + FakeAdapterMode::WrongPhase => LifecycleAdapterResult::Completed { + transition: Box::new(LifecycleTransition::CompleteImplement { + worker: worker(request, usage.clone(), 1, false), + patch: LifecyclePatchRef { + patch_id: PatchId(Digest::blake3(b"wrong-phase-patch")), + revision: 1, + }, + }), + }, + mode => { + let verify_status = match mode { + FakeAdapterMode::RepairOnce + if request.phase == LifecyclePhase::Verify + && !request.lifecycle.repair_consumed => + { + VerificationStatus::Repairable + } + FakeAdapterMode::RejectVerification + if request.phase == LifecyclePhase::Verify => + { + VerificationStatus::Rejected + } + _ => VerificationStatus::Verified, + }; + LifecycleAdapterResult::Completed { + transition: Box::new(completion_transition( + request, + usage.clone(), + if matches!(mode, FakeAdapterMode::BadWorkerDepth) { 2 } else { 1 }, + matches!(mode, FakeAdapterMode::WrongProfile), + verify_status, + )), + } + } + }; + let outcome = LifecyclePhaseAdapterOutcome { + schema: LIFECYCLE_ADAPTER_OUTCOME_SCHEMA.to_owned(), + invocation_id: request.invocation_id, + expected_state_digest: request.expected_state_digest, + phase: request.phase, + usage, + result, + cleanup, + }; + self.cached.borrow_mut().insert(request.invocation_id, outcome.clone()); + outcome + } +} + +fn worker( + request: &LifecyclePhaseAdapterRequest, + usage: LifecycleUsage, + worker_depth: u8, + wrong_profile: bool, +) -> LifecycleWorkerCompletion { + LifecycleWorkerCompletion { + profile: if wrong_profile { + profile("kernel.wrong-profile") + } else { + request.profile.clone() + }, + worker_depth, + logical_worker_spawns: 1, + usage, + } +} + +fn completion_transition( + request: &LifecyclePhaseAdapterRequest, + usage: LifecycleUsage, + worker_depth: u8, + wrong_profile: bool, + verify_status: VerificationStatus, +) -> LifecycleTransition { + let worker = worker(request, usage, worker_depth, wrong_profile); + match request.phase { + LifecyclePhase::Explore => LifecycleTransition::CompleteExplore { + worker, + artifacts: vec![LifecycleArtifactRef { + kind: LifecycleArtifactKind::Exploration, + id: Digest::blake3(b"kernel-exploration"), + source_snapshot: request.lifecycle.source_snapshot, + }], + }, + LifecyclePhase::Implement => LifecycleTransition::CompleteImplement { + worker, + patch: LifecyclePatchRef { + patch_id: PatchId(Digest::blake3(if request.lifecycle.repair_consumed { + b"kernel-patch-2" + } else { + b"kernel-patch-1" + })), + revision: if request.lifecycle.repair_consumed { 2 } else { 1 }, + }, + }, + LifecyclePhase::Test => { + let binding = &request.lifecycle.spec.test_plans[0]; + LifecycleTransition::CompleteTest { + worker, + results: vec![LifecycleTestResult { + plan_digest: binding.plan_digest(), + certificate_digest: binding.certificate_digest, + available: true, + executed: true, + passed: true, + evidence_id: Some("fake:test-evidence".to_owned()), + failure_code: None, + }], + } + } + LifecyclePhase::Review => { + let patch_id = request.lifecycle.patch.as_ref().unwrap().patch_id; + let review = ReviewArtifact::new( + request.lifecycle.change_id.clone(), + patch_id, + LifecycleReviewVerdict::Approved, + vec![LifecycleAcceptanceReview::new( + request.change.request.acceptance_criteria[0].as_bytes(), + AcceptanceStatus::Addressed, + b"fake review evidence", + )], + Vec::new(), + request.profile.definition_digest, + request.lifecycle.updated_unix_ms, + ) + .unwrap(); + LifecycleTransition::CompleteReview { worker, review } + } + LifecyclePhase::Verify => { + let verification = + verification_artifact(&request.lifecycle, &request.change.request, verify_status); + LifecycleTransition::CompleteVerify { + worker, + verification: LifecycleVerificationRef { + verification_id: verification.id, + patch_id: verification.patch_id, + verdict: verification.verdict, + }, + } + } + LifecyclePhase::Apply => unreachable!("the kernel never invokes an apply adapter"), + } +} + +fn verification_artifact( + lifecycle: &DevelopmentLifecycle, + request: &ChangeRequest, + verdict: VerificationStatus, +) -> VerificationArtifact { + let patch_id = lifecycle.patch.as_ref().unwrap().patch_id; + let acceptance_coverage = vec![AcceptanceCoverage { + criterion: request.acceptance_criteria[0].clone(), + status: AcceptanceStatus::Addressed, + evidence: "fake verification evidence".to_owned(), + }]; + let findings = if verdict == VerificationStatus::Repairable { + vec!["one bounded repair is required".to_owned()] + } else { + Vec::new() + }; + let test_evidence_ids = Vec::new(); + let verifier_definition = lifecycle.spec.profiles.verify.definition_digest; + let id = VerificationArtifact::compute_id( + &lifecycle.change_id, + patch_id, + verdict, + &acceptance_coverage, + &findings, + &test_evidence_ids, + verifier_definition, + ); + let artifact = VerificationArtifact { + id, + change_id: lifecycle.change_id.clone(), + patch_id, + verdict, + acceptance_coverage, + findings, + test_evidence_ids, + test_plan_results: Vec::new(), + test_plans_over_cap: false, + verifier_definition, + created_unix_ms: lifecycle.updated_unix_ms, + }; + assert!(artifact.is_canonical()); + artifact +} + +fn adapters(adapter: &FakeAdapter) -> LifecyclePhaseAdapters<'_> { + LifecyclePhaseAdapters { + explore: adapter, + implement: adapter, + test: adapter, + review: adapter, + verify: adapter, + } +} + +fn step( + store: &FakeStore, + adapter: &FakeAdapter, + cancellation: &dyn LifecycleCancellation, + change_id: &ChangeId, +) -> Result { + execute_step(store, &adapters(adapter), cancellation, change_id) +} + +struct CancelOnCheck { + cancel_on: u32, + checks: Cell, +} + +impl LifecycleCancellation for CancelOnCheck { + fn is_cancelled(&self, _change_id: &ChangeId, _expected_state_digest: Digest) -> bool { + let check = self.checks.get() + 1; + self.checks.set(check); + check == self.cancel_on + } +} + +#[test] +fn success_invokes_one_adapter_per_phase_and_stops_at_approval() { + let (store, change_id) = FakeStore::new(10); + let adapter = FakeAdapter::new(FakeAdapterMode::Success); + let mut last = None; + + for _ in 0..5 { + last = Some(step(&store, &adapter, &NeverCancel, &change_id).unwrap()); + } + + let last = last.unwrap(); + assert_eq!(last.disposition, LifecycleExecutionDisposition::AwaitingApproval); + assert_eq!(last.projection.lifecycle.phase, LifecyclePhase::Apply); + assert_eq!(last.projection.lifecycle.status, LifecycleStatus::AwaitingApproval); + assert_eq!(last.projection.lifecycle.usage.worker_turns, 5); + assert_eq!(adapter.calls.get(), 5); + assert_eq!(adapter.side_effects.borrow().len(), 5); + assert_eq!( + adapter.phases.borrow().as_slice(), + &[ + LifecyclePhase::Explore, + LifecyclePhase::Implement, + LifecyclePhase::Test, + LifecyclePhase::Review, + LifecyclePhase::Verify, + ] + ); + + let stopped = step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + assert_eq!(stopped.disposition, LifecycleExecutionDisposition::AwaitingApproval); + assert_eq!(adapter.calls.get(), 5); +} + +#[test] +fn repair_is_consumed_transactionally_once_then_resumes_implement() { + let (store, change_id) = FakeStore::new(12); + let adapter = FakeAdapter::new(FakeAdapterMode::RepairOnce); + for _ in 0..5 { + step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + } + assert_eq!(store.projection.borrow().lifecycle.status, LifecycleStatus::RepairReserved); + + let resumed = step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + assert_eq!(resumed.disposition, LifecycleExecutionDisposition::RepairResumed); + assert_eq!(resumed.projection.lifecycle.phase, LifecyclePhase::Implement); + assert!(resumed.projection.lifecycle.repair_consumed); + assert_eq!(store.repair_calls.get(), 1); + assert_eq!(adapter.calls.get(), 5); + + for _ in 0..4 { + step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + } + let finished = store.projection.borrow().clone(); + assert_eq!(finished.lifecycle.status, LifecycleStatus::AwaitingApproval); + assert_eq!(finished.lifecycle.patch.as_ref().unwrap().revision, 2); + assert_eq!(finished.lifecycle.usage.worker_turns, 9); + assert_eq!(store.repair_calls.get(), 1); + assert_eq!(adapter.calls.get(), 9); + assert_eq!(adapter.repair_contexts.get(), 4); +} + +#[test] +fn cancellation_wins_before_a_reserved_repair_is_consumed() { + let (store, change_id) = FakeStore::new(12); + let adapter = FakeAdapter::new(FakeAdapterMode::RepairOnce); + for _ in 0..5 { + step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + } + assert_eq!(store.projection.borrow().lifecycle.status, LifecycleStatus::RepairReserved); + let cancellation = CancelOnCheck { cancel_on: 1, checks: Cell::new(0) }; + + let outcome = step(&store, &adapter, &cancellation, &change_id).unwrap(); + + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::Cancelled); + assert_eq!(outcome.projection.lifecycle.status, LifecycleStatus::Cancelled); + assert_eq!(store.repair_calls.get(), 0); + assert_eq!(adapter.calls.get(), 5); +} + +#[test] +fn rejection_is_a_typed_terminal_completion() { + let (store, change_id) = FakeStore::new(10); + let adapter = FakeAdapter::new(FakeAdapterMode::RejectVerification); + let mut outcome = None; + for _ in 0..5 { + outcome = Some(step(&store, &adapter, &NeverCancel, &change_id).unwrap()); + } + let outcome = outcome.unwrap(); + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::Terminal); + assert_eq!(outcome.projection.lifecycle.status, LifecycleStatus::Failed); + assert_eq!( + outcome.projection.lifecycle.terminal_reason.as_ref().unwrap().code, + "verification_rejected" + ); +} + +#[test] +fn phase_profile_mismatch_and_bad_depth_fail_closed_as_invalid_output() { + for mode in [ + FakeAdapterMode::WrongPhase, + FakeAdapterMode::WrongProfile, + FakeAdapterMode::BadWorkerDepth, + ] { + let (store, change_id) = FakeStore::new(10); + let adapter = FakeAdapter::new(mode); + let outcome = step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::InvalidAdapterOutput); + assert_eq!(outcome.projection.lifecycle.status, LifecycleStatus::Failed); + assert_eq!( + outcome.projection.lifecycle.terminal_reason.as_ref().unwrap().code, + "adapter_invalid_output" + ); + } +} + +#[test] +fn cancellation_before_and_after_adapter_is_fail_closed() { + let (before_store, before_id) = FakeStore::new(10); + let before_adapter = FakeAdapter::new(FakeAdapterMode::Success); + let before = CancelOnCheck { cancel_on: 1, checks: Cell::new(0) }; + let outcome = step(&before_store, &before_adapter, &before, &before_id).unwrap(); + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::Cancelled); + assert_eq!(outcome.projection.lifecycle.status, LifecycleStatus::Cancelled); + assert_eq!(before_adapter.calls.get(), 0); + + let (after_store, after_id) = FakeStore::new(10); + let after_adapter = FakeAdapter::new(FakeAdapterMode::Success); + let after = CancelOnCheck { cancel_on: 2, checks: Cell::new(0) }; + let outcome = step(&after_store, &after_adapter, &after, &after_id).unwrap(); + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::Cancelled); + assert_eq!(outcome.projection.lifecycle.status, LifecycleStatus::Cancelled); + assert_eq!(outcome.adapter_usage.worker_turns, 1); + assert_eq!(after_adapter.calls.get(), 1); +} + +#[test] +fn exhausted_aggregate_budget_stops_before_another_adapter() { + let (store, change_id) = FakeStore::new(1); + let adapter = FakeAdapter::new(FakeAdapterMode::Success); + step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + + let outcome = step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::BudgetExhausted); + assert_eq!(outcome.projection.lifecycle.status, LifecycleStatus::Failed); + assert_eq!( + outcome.projection.lifecycle.terminal_reason.as_ref().unwrap().code, + "budget_exhausted" + ); + assert_eq!(adapter.calls.get(), 1); +} + +#[test] +fn adapter_and_cleanup_failures_are_distinct_terminal_outcomes() { + let (failed_store, failed_id) = FakeStore::new(10); + let failed_adapter = FakeAdapter::new(FakeAdapterMode::Failure); + let failed = step(&failed_store, &failed_adapter, &NeverCancel, &failed_id).unwrap(); + assert_eq!(failed.disposition, LifecycleExecutionDisposition::AdapterFailed); + assert_eq!( + failed.projection.lifecycle.terminal_reason.as_ref().unwrap().code, + "adapter_failed" + ); + + let (cleanup_store, cleanup_id) = FakeStore::new(10); + let cleanup_adapter = FakeAdapter::new(FakeAdapterMode::CleanupFailure); + let cleanup = step(&cleanup_store, &cleanup_adapter, &NeverCancel, &cleanup_id).unwrap(); + assert_eq!(cleanup.disposition, LifecycleExecutionDisposition::CleanupFailed); + assert_eq!( + cleanup.projection.lifecycle.terminal_reason.as_ref().unwrap().code, + "adapter_cleanup_failed" + ); + assert!(!cleanup.cleanup.unwrap().succeeded); +} + +#[test] +fn oversized_adapter_failure_is_bounded_and_rejected() { + let (store, change_id) = FakeStore::new(10); + let adapter = FakeAdapter::new(FakeAdapterMode::OversizedFailure); + let outcome = step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::InvalidAdapterOutput); + assert_eq!( + outcome.projection.lifecycle.terminal_reason.as_ref().unwrap().code, + "adapter_invalid_output" + ); +} + +#[test] +fn stale_cas_does_not_overwrite_a_concurrent_parent_transition() { + let (store, change_id) = FakeStore::new(10); + let adapter = FakeAdapter::new(FakeAdapterMode::Success); + store.concurrent_transition.replace(Some(LifecycleTransition::Cancel { + reason: LifecycleReason::new("concurrent_cancel", b"concurrent parent won").unwrap(), + })); + + let outcome = step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + assert_eq!(outcome.disposition, LifecycleExecutionDisposition::StaleState); + assert_eq!(outcome.projection.lifecycle.status, LifecycleStatus::Cancelled); + assert!(outcome.projection.lifecycle.exploration_artifacts.is_empty()); +} + +#[test] +fn retry_reuses_invocation_identity_without_repeating_adapter_side_effect() { + let (store, change_id) = FakeStore::new(10); + let adapter = FakeAdapter::new(FakeAdapterMode::Success); + let initial_digest = store.projection.borrow().state_digest; + store.fail_next_parent_commit.set(true); + + let first = step(&store, &adapter, &NeverCancel, &change_id); + assert!(matches!(first, Err(LifecycleExecutionError::Store(StoreError::ConnectionLock)))); + assert_eq!(store.projection.borrow().state_digest, initial_digest); + + let second = step(&store, &adapter, &NeverCancel, &change_id).unwrap(); + assert_eq!(second.disposition, LifecycleExecutionDisposition::Advanced); + assert_eq!(second.projection.lifecycle.phase, LifecyclePhase::Implement); + assert_eq!(adapter.calls.get(), 2); + assert_eq!(adapter.side_effects.borrow().len(), 1); + assert_eq!( + adapter.phases.borrow().as_slice(), + &[LifecyclePhase::Explore, LifecyclePhase::Explore] + ); +} + +#[test] +fn replay_corruption_is_rejected_before_adapter_invocation() { + let (store, change_id) = FakeStore::new(10); + let adapter = FakeAdapter::new(FakeAdapterMode::Success); + store.corrupt_replay.set(true); + + let outcome = step(&store, &adapter, &NeverCancel, &change_id); + assert!(matches!( + outcome, + Err(LifecycleExecutionError::Store(StoreError::LifecycleCorruption(_))) + )); + assert_eq!(adapter.calls.get(), 0); +} diff --git a/crates/needle-runtime/src/store.rs b/crates/needle-runtime/src/store.rs index 10fa4e7..8bee054 100644 --- a/crates/needle-runtime/src/store.rs +++ b/crates/needle-runtime/src/store.rs @@ -30,7 +30,9 @@ mod lifecycles; #[path = "store/role_profiles.rs"] mod role_profiles; -pub use changes::{ChangeAttemptRecord, PatchFileBlob, PreparedChangeRecord}; +pub use changes::{ + ChangeAttemptRecord, LifecycleChangeContext, PatchFileBlob, PreparedChangeRecord, +}; pub use lifecycles::LifecycleProjection; pub use role_profiles::{ RoleProfileAuditOperation, RoleProfileAuditRecord, RoleProfileStateRecord, diff --git a/crates/needle-runtime/src/store/changes.rs b/crates/needle-runtime/src/store/changes.rs index 5ca5315..7bbb2ce 100644 --- a/crates/needle-runtime/src/store/changes.rs +++ b/crates/needle-runtime/src/store/changes.rs @@ -96,6 +96,15 @@ pub struct PreparedChangeRecord { pub role_profile_provenance: Option, } +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LifecycleChangeContext { + pub request: ChangeRequest, + pub request_digest: Digest, + pub repository_id: Digest, + pub source_snapshot: Digest, +} + #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ChangeAttemptRecord { @@ -109,6 +118,40 @@ pub struct ChangeAttemptRecord { } impl RuntimeStore { + pub fn lifecycle_change_context( + &self, + change_id: &ChangeId, + ) -> Result, StoreError> { + let connection = self.connection()?; + let row: Option<(String, String, String, String)> = connection + .query_row( + "SELECT request_digest, repository_id, source_snapshot_digest, request_json + 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)?)), + ) + .optional()?; + let Some((request_digest, repository_id, source_snapshot, request_json)) = row else { + return Ok(None); + }; + let request_digest = Digest::parse(&request_digest) + .map_err(|_| StoreError::LifecycleCorruption(format!("{change_id}: request digest")))?; + let repository_id = Digest::parse(&repository_id).map_err(|_| { + StoreError::LifecycleCorruption(format!("{change_id}: repository digest")) + })?; + let source_snapshot = Digest::parse(&source_snapshot) + .map_err(|_| StoreError::LifecycleCorruption(format!("{change_id}: source digest")))?; + let request: ChangeRequest = serde_json::from_str(&request_json).map_err(|_| { + StoreError::LifecycleCorruption(format!("{change_id}: immutable request JSON")) + })?; + if request.digest(source_snapshot) != request_digest { + return Err(StoreError::LifecycleCorruption(format!( + "{change_id}: immutable request digest" + ))); + } + Ok(Some(LifecycleChangeContext { request, request_digest, repository_id, source_snapshot })) + } + pub fn record_change_request( &self, change_id: &ChangeId, @@ -1375,6 +1418,66 @@ mod tests { RoleProfileProvenance::from_revision(&revision).unwrap() } + #[test] + fn lifecycle_change_context_round_trips_and_rejects_a_digest_mismatch() { + let suffix = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos(); + let path = std::env::temp_dir().join(format!( + "needle-lifecycle-change-context-{}-{suffix}.sqlite3", + std::process::id() + )); + let store = RuntimeStore::new(&path); + store.initialize().unwrap(); + let source_snapshot = Digest::blake3(b"lifecycle-context-source"); + let repository_id = Digest::blake3(b"lifecycle-context-repository"); + let request = ChangeRequest { + task: "Run a bounded lifecycle adapter.".to_owned(), + acceptance_criteria: vec!["The immutable request is preserved.".to_owned()], + allowed_paths: Vec::new(), + artifact_ids: Vec::new(), + claim_ids: Vec::new(), + constraints: Vec::new(), + }; + let change_id = ChangeId::from_digest(Digest::blake3(b"lifecycle-context-change")); + let request_digest = request.digest(source_snapshot); + store + .record_change_request( + &change_id, + repository_id, + source_snapshot, + request_digest, + &request, + ) + .unwrap(); + + assert_eq!( + store.lifecycle_change_context(&change_id).unwrap(), + Some(LifecycleChangeContext { + request: request.clone(), + request_digest, + repository_id, + source_snapshot, + }) + ); + + let mismatched_id = ChangeId::from_digest(Digest::blake3(b"lifecycle-context-mismatch")); + store + .record_change_request( + &mismatched_id, + repository_id, + source_snapshot, + Digest::blake3(b"wrong-request-digest"), + &request, + ) + .unwrap(); + assert!(matches!( + store.lifecycle_change_context(&mismatched_id), + Err(StoreError::LifecycleCorruption(_)) + )); + + drop(store); + fs::remove_file(path).unwrap(); + } + #[test] fn failed_request_is_audited_before_any_patch_exists() { let suffix = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos(); diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3c887c7..f1f9785 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -106,9 +106,14 @@ 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. +Runtime adds a parent-owned execution kernel over this durable contract. The +kernel replays the projection, selects one active phase, invokes one injected +adapter with bounded request/result contracts, and alone commits the typed +transition through digest CAS. Deterministic invocation identities let adapters +deduplicate side effects after an uncommitted attempt; replay prevents a +committed phase from running again. The kernel consumes the one transactional +repair reservation and stops at `apply/awaiting_approval`. Concrete Codex +process adapters and the lifecycle read UI remain separate consumers. ## Request flow diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index dbd19bf..a67eab6 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -56,12 +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. +The durable contract, SQLite journal, and parent-owned injected-adapter kernel +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. The kernel runs at most one adapter for +the validated phase and stops at explicit apply approval. Concrete Codex process +adapters and the lifecycle read/timeline UI remain separate pending slices. ## Milestone 3: Other-host subagent configuration diff --git a/docs/VERIFIED_CHANGES.md b/docs/VERIFIED_CHANGES.md index fa51886..1488dc3 100644 --- a/docs/VERIFIED_CHANGES.md +++ b/docs/VERIFIED_CHANGES.md @@ -81,8 +81,10 @@ 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. +stored projection. Runtime's parent-owned kernel selects one validated phase, +invokes one injected depth-one adapter with a deterministic invocation identity, +and commits through digest CAS. It stops at `apply/awaiting_approval`. Concrete +Codex worker processes and a lifecycle HTTP/UI surface remain unwired. ## One repair