diff --git a/crates/tracedecay-usecases/src/advisory/github_runtime.rs b/crates/tracedecay-usecases/src/advisory/github_runtime.rs index 7132e98bc..835316a74 100644 --- a/crates/tracedecay-usecases/src/advisory/github_runtime.rs +++ b/crates/tracedecay-usecases/src/advisory/github_runtime.rs @@ -81,7 +81,7 @@ pub use network::{ }; pub use owner::{ GitHubReviewRuntimeOwnerBuildErrorV1, GitHubReviewRuntimeOwnerConfigV1, - GitHubReviewRuntimeOwnerV1, build_github_review_runtime_owner_v1, + GitHubReviewRuntimeOwnerV1, GitHubStackObservabilityV1, build_github_review_runtime_owner_v1, }; pub use read_requests::{GitHubGraphQlReadRequestV1, GitHubReadResumeV1, GitHubRestReadRequestV1}; pub use releases::{ diff --git a/crates/tracedecay-usecases/src/advisory/github_runtime/owner.rs b/crates/tracedecay-usecases/src/advisory/github_runtime/owner.rs index 6f3560366..4fb918886 100644 --- a/crates/tracedecay-usecases/src/advisory/github_runtime/owner.rs +++ b/crates/tracedecay-usecases/src/advisory/github_runtime/owner.rs @@ -23,10 +23,70 @@ use crate::advisory::{ GitHubCurrentBranchRemapper, GitHubReadOnlyAdmissionError, GitHubReadOnlyConnector, GitHubReadOnlyDescriptorSetV1, GitHubRestDescriptorV1, }; -use crate::stack_coordinator::DaemonGitHubStackCoordinatorV1; +use crate::observability::{ + BoundedObservabilityProducerV1, GitHubStackCapabilityObservationResultV1, + GitHubStackDriftObservationResultV1, GitHubStackProbeOwnerV1, record_github_stack_capability, + record_github_stack_drifts, +}; +use crate::stack_coordinator::{ + DaemonGitHubStackCoordinatorV1, GitHubStackObservationV1, GitHubStackProviderSourceBindingV1, +}; use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use tracedecay_runtime_core::db::Database; +/// Canonical Observatory mount for the coordinator observations this owner +/// produces. Absent when the composition root could not mount the probe +/// owner, producer, and observation database; refresh then keeps producing +/// product anchors without canonical capability/drift receipts. +#[derive(Clone)] +pub struct GitHubStackObservabilityV1 { + pub probe_owner: GitHubStackProbeOwnerV1, + pub producer: Arc, + pub observation_db: RegisteredGlobalDbLeaseV1, +} + +impl GitHubStackObservabilityV1 { + /// Offers one validated coordinator observation as a capability receipt + /// plus one receipt per exact drift interval. Telemetry refusal is + /// logged, never propagated to the refresh product path. + pub fn record( + &self, + source_binding: &GitHubStackProviderSourceBindingV1, + observation: &GitHubStackObservationV1, + ) { + let capability = record_github_stack_capability( + self.observation_db.as_ref(), + Some(self.producer.as_ref()), + &self.probe_owner, + source_binding, + observation, + ); + if capability != GitHubStackCapabilityObservationResultV1::Enqueued { + tracing::warn!( + event = "github_stack_capability_observation_refused", + outcome = ?capability, + "GitHub stack capability receipt did not enter the canonical producer" + ); + } + match record_github_stack_drifts( + self.observation_db.as_ref(), + Some(self.producer.as_ref()), + &self.probe_owner, + source_binding, + observation, + ) { + GitHubStackDriftObservationResultV1::Emitted { dropped: 0, .. } => {} + refused => { + tracing::warn!( + event = "github_stack_drift_observation_refused", + outcome = ?refused, + "GitHub stack drift receipts did not fully enter the canonical producer" + ); + } + } + } +} + pub struct GitHubReviewRuntimeOwnerConfigV1 { pub database: Database, pub resolved_scope: ResolvedScope, @@ -37,6 +97,7 @@ pub struct GitHubReviewRuntimeOwnerConfigV1 { pub identity: GitHubReviewProviderIdentityV1, pub stack_coordinator: Arc, pub stack_anchor_db: RegisteredGlobalDbLeaseV1, + pub stack_observability: Option, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -69,6 +130,7 @@ pub struct GitHubReviewRuntimeOwnerV1 { stack_provider: tracedecay_domain::ProviderId, stack_coordinator: Arc, stack_anchors: super::ProjectGitHubStackAnchorAuthorityV1, + stack_observability: Option, } impl GitHubReviewRuntimeOwnerV1 @@ -146,10 +208,16 @@ where self.stack_scope.clone(), self.stack_provider.clone(), provider_outcome, - source_binding, + source_binding.clone(), stack_observed_at, ) { Ok(observation) => { + // Offered before anchor publication so a publication + // refusal below cannot conceal the observation the + // coordinator already made. + if let Some(stack_observability) = &self.stack_observability { + stack_observability.record(&source_binding, &observation); + } let anchor_publication = self .stack_anchors .publish(context, request, &observation, self.source_access.as_ref()) @@ -224,6 +292,7 @@ where let stack_scope = config.resolved_scope.clone(); let stack_provider = config.identity.provider.clone(); let stack_coordinator = Arc::clone(&config.stack_coordinator); + let stack_observability = config.stack_observability.clone(); let stack_anchors = super::ProjectGitHubStackAnchorAuthorityV1::new( config.stack_anchor_db.clone(), config.feedback_scope.clone(), @@ -252,6 +321,7 @@ where stack_provider, stack_coordinator, stack_anchors, + stack_observability, }) } diff --git a/crates/tracedecay-usecases/src/advisory/mod.rs b/crates/tracedecay-usecases/src/advisory/mod.rs index 027052dc4..67a2b75fa 100644 --- a/crates/tracedecay-usecases/src/advisory/mod.rs +++ b/crates/tracedecay-usecases/src/advisory/mod.rs @@ -76,13 +76,14 @@ pub use github_runtime::{ GitHubReviewRefreshStoreReadOutcomeV1, GitHubReviewRuntimeOwnerBuildErrorV1, GitHubReviewRuntimeOwnerConfigV1, GitHubReviewRuntimeOwnerV1, GitHubReviewStoreManifestEntryV1, GitHubReviewStoreManifestLoadOutcomeV1, GitHubReviewStoreManifestV1, - MAX_GITHUB_READ_RESPONSE_BYTES_V1, MAX_GITHUB_REVIEW_STORE_MANIFEST_ENTRIES_V1, - ProjectGitHubAnchorAuthorityV1, ProjectGitHubRegistrarAuthoritiesV1, - ProjectGitHubReleaseAuthorityOpenOutcomeV1, ProjectGitHubReleasePageV1, - ProjectGitHubReleaseReadAuthorityV1, ProjectGitHubReleaseReadOutcomeV1, - ProjectGitHubReleaseReadRequestV1, ProjectGitHubReviewStoreV1, - build_github_review_runtime_owner_v1, github_anchor_authorities_arc_v1, - github_anchor_authorities_v1, open_project_github_release_read_authority_v1, + GitHubStackObservabilityV1, MAX_GITHUB_READ_RESPONSE_BYTES_V1, + MAX_GITHUB_REVIEW_STORE_MANIFEST_ENTRIES_V1, ProjectGitHubAnchorAuthorityV1, + ProjectGitHubRegistrarAuthoritiesV1, ProjectGitHubReleaseAuthorityOpenOutcomeV1, + ProjectGitHubReleasePageV1, ProjectGitHubReleaseReadAuthorityV1, + ProjectGitHubReleaseReadOutcomeV1, ProjectGitHubReleaseReadRequestV1, + ProjectGitHubReviewStoreV1, build_github_review_runtime_owner_v1, + github_anchor_authorities_arc_v1, github_anchor_authorities_v1, + open_project_github_release_read_authority_v1, register_github_read_only_credential_authority_v1, register_profile_github_read_only_credential_authority_v1, unregister_github_read_only_credential_authority_v1, diff --git a/crates/tracedecay-usecases/src/observability.rs b/crates/tracedecay-usecases/src/observability.rs index e03e94e37..a8b3161b3 100644 --- a/crates/tracedecay-usecases/src/observability.rs +++ b/crates/tracedecay-usecases/src/observability.rs @@ -9,6 +9,7 @@ mod emit; mod execution_emit; mod export; mod github_stack_emit; +mod no_progress_emit; mod producer; mod product_view_emit; mod read; @@ -18,6 +19,7 @@ mod read_model_tests; mod retrieval_emit; mod store; mod work_blocked_interval_emit; +mod work_conflict_emit; mod work_duplicate_emit; mod work_operation_resource_emit; mod work_owner_observation_recovery; @@ -51,6 +53,7 @@ pub use github_stack_emit::{ GitHubStackDriftRecoveryErrorV1, GitHubStackProbeOwnerMountErrorV1, GitHubStackProbeOwnerV1, record_github_stack_capability, record_github_stack_drifts, recover_open_github_stack_drifts, }; +pub use no_progress_emit::{WorkNoProgressObservationV1, record_no_progress_observation}; pub use producer::{ BoundedObservabilityProducerV1, ObservabilityEmissionOutcomeV1, ObservabilityOwnerEmissionOutcomeV1, ObservabilityProducerDeadlinesV1, @@ -76,6 +79,10 @@ pub use tracedecay_global_db::{ pub use work_blocked_interval_emit::{ record_work_blocked_interval_observation, work_blocked_interval_observation_envelope, }; +pub use work_conflict_emit::{ + WorkConflictObservationResultV1, WorkConflictObservationUnavailableV1, + record_work_conflict_observation, +}; pub use work_duplicate_emit::record_work_duplicate_observation; pub use work_operation_resource_emit::record_work_operation_resource; pub use work_owner_observation_recovery::{ diff --git a/crates/tracedecay-usecases/src/observability/no_progress_emit.rs b/crates/tracedecay-usecases/src/observability/no_progress_emit.rs new file mode 100644 index 000000000..def665a62 --- /dev/null +++ b/crates/tracedecay-usecases/src/observability/no_progress_emit.rs @@ -0,0 +1,300 @@ +//! Terminal no-progress receipts for wall-exhausted Work provider attempts. +//! +//! The provider-attempt authority commits no intermediate progress frontier +//! (`Leased -> Running -> terminal`, heartbeats never reset the deadline), so +//! at the wall-exhaustion kill the frontier is provably zero, the stall is +//! the measured wall since the attempt began, no run budget remains, and the +//! unreconciled worktree effect outcome is truthfully unknown. + +use tracedecay_domain::{ + CoverageStateV1, EffectReconciliationOutcomeV1, NoProgressEscalationV1, NoProgressObservedV1, + ObservabilityEnvelopeV1, ObservabilityPayloadV1, ObservabilityTerminalResultV1, UtcMicros, + WorkAttemptIdentityV1, WorkflowStageClassV1, canonical_sha256, +}; + +use super::{ + BoundedObservabilityProducerV1, ExecutionOwnerFactInputV1, ObservabilityEmissionOutcomeV1, + ObservabilityProducerIdentityV1, WorkOwnerObservationResultV1, execution_owner_fact_envelope, +}; + +/// One wall-exhaustion kill measured by the live attempt-execution owner. +/// Every field is a value the owner actually holds at the kill site. +pub struct WorkNoProgressObservationV1<'a> { + pub attempt: &'a WorkAttemptIdentityV1, + pub run_deadline: UtcMicros, + pub concurrency_policy_revision: &'a str, + pub configured_timeout_micros: u64, + pub elapsed_stall_micros: u64, + pub observed_at: UtcMicros, +} + +/// Offers one no-progress terminal fact without awaiting telemetry. The +/// payload contract refuses a zero wall budget and a stall shorter than the +/// armed budget; emission never changes the timed-out product handling. +pub fn record_no_progress_observation( + producer: Option<&BoundedObservabilityProducerV1>, + observation: &WorkNoProgressObservationV1<'_>, +) -> WorkOwnerObservationResultV1 { + let Some(producer) = producer else { + return WorkOwnerObservationResultV1::Unavailable; + }; + let scope = producer.identity().authorized_scope_ref.as_str(); + let envelope = match no_progress_observation_envelope(producer.identity(), scope, observation) { + Ok(envelope) => envelope, + Err(_) => return WorkOwnerObservationResultV1::Unavailable, + }; + match producer.try_emit_owner_fact(envelope) { + Ok(ObservabilityEmissionOutcomeV1::Enqueued) => WorkOwnerObservationResultV1::Enqueued, + Ok(ObservabilityEmissionOutcomeV1::DroppedAtCapacity) => { + WorkOwnerObservationResultV1::DroppedAtCapacity + } + Err(_) => WorkOwnerObservationResultV1::Unavailable, + } +} + +/// Builds the canonical owner envelope for one wall-exhaustion kill. The +/// run-deadline reference hashes the exact attempt identity and deadline, so +/// replays are idempotent and raw identifiers never enter the payload. +fn no_progress_observation_envelope( + identity: &ObservabilityProducerIdentityV1, + canonical_project_scope: &str, + observation: &WorkNoProgressObservationV1<'_>, +) -> Result { + let run_deadline_digest = canonical_sha256(&( + "tracedecay.work.run-deadline.v1", + observation.attempt.task_id().as_str(), + observation.attempt.run_id().as_str(), + observation.attempt.attempt_id().as_str(), + observation.run_deadline, + )) + .map_err(|_| "no_progress_run_deadline_identity")?; + let payload = NoProgressObservedV1 { + run_deadline_ref: format!("work-run-deadline:{}", run_deadline_digest.as_str()), + concurrency_policy_revision: observation.concurrency_policy_revision.to_owned(), + workflow_stage: WorkflowStageClassV1::Execute, + configured_timeout_micros: observation.configured_timeout_micros, + // The provider-attempt authority has no frontier commits between + // Running and the terminal transition; zero is the proven frontier, + // not a default. + last_committed_frontier: 0, + elapsed_stall_micros: observation.elapsed_stall_micros, + // The armed wall budget is the envelope deadline itself; at + // exhaustion no run budget remains above it. + remaining_run_budget_micros: 0, + // Both live timeout sites deliver SIGKILL to the provider's whole + // process group/tree with no graceful rung. + escalation: NoProgressEscalationV1::Kill, + effect_outcome: EffectReconciliationOutcomeV1::Unknown, + }; + payload.validate()?; + let owner_transition_ref = format!( + "work-no-progress:{}/{}/{}", + observation.attempt.task_id().as_str(), + observation.attempt.run_id().as_str(), + observation.attempt.attempt_id().as_str() + ); + execution_owner_fact_envelope( + identity, + canonical_project_scope, + ExecutionOwnerFactInputV1 { + owner_transition_ref: &owner_transition_ref, + operation: "execute_work_attempt", + event_time: observation.observed_at, + valid_from: Some(observation.observed_at), + valid_until: Some(observation.observed_at), + terminal_result: Some(ObservabilityTerminalResultV1::TimedOut), + coverage: CoverageStateV1::Known, + payload: ObservabilityPayloadV1::NoProgress(payload), + }, + ) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + use tracedecay_application::{ObservabilityQueryPort, ObservabilityQueryV1}; + use tracedecay_domain::{AttemptId, RunId, TaskId}; + + use crate::observability::RegisteredObservabilityPortV1; + + const SCOPE: &str = "project.no-progress-emit"; + + fn producer_identity() -> ObservabilityProducerIdentityV1 { + ObservabilityProducerIdentityV1 { + authorized_scope_ref: SCOPE.to_owned(), + process_boot_id: "boot:no-progress-emit".to_owned(), + producer_revision: "producer.v1".to_owned(), + configuration_revision: "configuration.v1".to_owned(), + policy_revision: "policy.v1".to_owned(), + } + } + + fn attempt_identity() -> WorkAttemptIdentityV1 { + WorkAttemptIdentityV1::new( + TaskId::new("task.no-progress".to_owned()).expect("task id"), + RunId::new("run.no-progress".to_owned()).expect("run id"), + AttemptId::new("attempt.no-progress".to_owned()).expect("attempt id"), + ) + .expect("attempt identity") + } + + fn observation( + attempt: &WorkAttemptIdentityV1, + configured_timeout_micros: u64, + elapsed_stall_micros: u64, + ) -> WorkNoProgressObservationV1<'_> { + WorkNoProgressObservationV1 { + attempt, + run_deadline: UtcMicros(2_000_000), + concurrency_policy_revision: "topology-policy.v1", + configured_timeout_micros, + elapsed_stall_micros, + observed_at: UtcMicros(1_550_000), + } + } + + #[tokio::test] + async fn wall_exhausted_attempt_persists_the_no_progress_terminal_fact() { + let harness = tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness::open( + "no-progress-emit", + ) + .await; + let producer = BoundedObservabilityProducerV1::start( + harness.registered.clone(), + producer_identity(), + 8, + ) + .expect("bounded producer"); + let attempt = attempt_identity(); + + assert_eq!( + record_no_progress_observation( + Some(&producer), + &observation(&attempt, 30_000_000, 31_000_000), + ), + WorkOwnerObservationResultV1::Enqueued + ); + // A stall shorter than the armed budget is refused, not persisted. + assert_eq!( + record_no_progress_observation( + Some(&producer), + &observation(&attempt, 30_000_000, 29_000_000), + ), + WorkOwnerObservationResultV1::Unavailable + ); + producer.shutdown().await.expect("producer shutdown"); + + let page = RegisteredObservabilityPortV1::new(&harness.registered) + .query(ObservabilityQueryV1 { + authorized_scope_ref: SCOPE.to_owned(), + event_kinds: vec!["operation.no_progress.terminal.v1".to_owned()], + horizon: tracedecay_application::ObservabilityHorizonV1 { + since_micros: 1_500_000, + until_micros: 1_600_000, + }, + after_watermark: None, + limit: 8, + }) + .await + .expect("no-progress page"); + assert_eq!(page.events.len(), 1); + let envelope = &page.events[0]; + assert_eq!(envelope.event_kind, "operation.no_progress.terminal.v1"); + assert_eq!( + envelope.terminal_result, + Some(ObservabilityTerminalResultV1::TimedOut) + ); + assert_eq!(envelope.coverage, CoverageStateV1::Known); + assert_eq!(envelope.event_time_micros, 1_550_000); + let ObservabilityPayloadV1::NoProgress(observed) = &envelope.payload else { + panic!("expected a no-progress payload, got {:?}", envelope.payload); + }; + let expected_deadline_ref = format!( + "work-run-deadline:{}", + canonical_sha256(&( + "tracedecay.work.run-deadline.v1", + attempt.task_id().as_str(), + attempt.run_id().as_str(), + attempt.attempt_id().as_str(), + UtcMicros(2_000_000), + )) + .expect("run deadline digest") + .as_str() + ); + assert_eq!(observed.run_deadline_ref, expected_deadline_ref); + assert_eq!(observed.concurrency_policy_revision, "topology-policy.v1"); + assert_eq!(observed.workflow_stage, WorkflowStageClassV1::Execute); + assert_eq!(observed.configured_timeout_micros, 30_000_000); + assert_eq!(observed.last_committed_frontier, 0); + assert_eq!(observed.elapsed_stall_micros, 31_000_000); + assert_eq!(observed.remaining_run_budget_micros, 0); + assert_eq!(observed.escalation, NoProgressEscalationV1::Kill); + assert_eq!( + observed.effect_outcome, + EffectReconciliationOutcomeV1::Unknown + ); + // Raw attempt identifiers never enter the exportable envelope. + let wire = serde_json::to_string(envelope).expect("serialize envelope"); + for prohibited in ["task.no-progress", "run.no-progress", "attempt.no-progress"] { + assert!(!wire.contains(prohibited), "leaked {prohibited}"); + } + } + + #[test] + fn absent_producer_is_a_typed_unavailable_state() { + let attempt = attempt_identity(); + assert_eq!( + record_no_progress_observation(None, &observation(&attempt, 30_000_000, 31_000_000)), + WorkOwnerObservationResultV1::Unavailable + ); + } + + #[test] + fn unmeasured_or_invalid_inputs_are_refused_without_panicking() { + let identity = producer_identity(); + let attempt = attempt_identity(); + // A zero wall budget is the deadline-already-elapsed admission state, + // not a measured stall. + assert_eq!( + no_progress_observation_envelope(&identity, SCOPE, &observation(&attempt, 0, 0)), + Err("no_progress_timeout") + ); + // A stall shorter than the armed budget was not a timeout. + assert_eq!( + no_progress_observation_envelope( + &identity, + SCOPE, + &observation(&attempt, 30_000_000, 29_999_999), + ), + Err("no_progress_timeout") + ); + // A non-canonical concurrency-policy revision is refused. + let oversized_revision = "r".repeat(97); + let mut invalid = observation(&attempt, 30_000_000, 31_000_000); + invalid.concurrency_policy_revision = &oversized_revision; + assert_eq!( + no_progress_observation_envelope(&identity, SCOPE, &invalid), + Err("revision") + ); + } + + #[test] + fn replayed_kill_builds_byte_identical_idempotent_envelopes() { + let identity = producer_identity(); + let attempt = attempt_identity(); + let first = no_progress_observation_envelope( + &identity, + SCOPE, + &observation(&attempt, 30_000_000, 31_000_000), + ) + .expect("first envelope"); + let replay = no_progress_observation_envelope( + &identity, + SCOPE, + &observation(&attempt, 30_000_000, 31_000_000), + ) + .expect("replay envelope"); + assert_eq!(first, replay, "owner-measured kill replays byte-identical"); + } +} diff --git a/crates/tracedecay-usecases/src/observability/work_conflict_emit.rs b/crates/tracedecay-usecases/src/observability/work_conflict_emit.rs new file mode 100644 index 000000000..f78d8ca49 --- /dev/null +++ b/crates/tracedecay-usecases/src/observability/work_conflict_emit.rs @@ -0,0 +1,882 @@ +//! Work-conflict predictions and linked outcomes from the mounted +//! native-integration owner. +//! +//! The preflight tree merge is the mechanical conflict oracle; its +//! disposition is recorded as an uncalibrated rule prediction bound to the +//! exact preview identity, and the terminal apply receipt is the independent +//! native-git adjudication linking back through the same deterministic +//! `prediction_ref`. Telemetry failure never changes the owner's result. + +use tracedecay_application::{ + NativeIntegrationPreviewProjectionV1, NativeIntegrationReceiptProjectionV1, + NativeIntegrationSurfaceResultV1, +}; +use tracedecay_domain::{ + ConflictAdjudicatorV1, ConflictKindV1, ConflictOutcomeV1, ConflictPredictionV1, + ConflictScoreKindV1, CoverageStateV1, ManifestDigest, NativeIntegrationPreviewDispositionV1, + NativeIntegrationPreviewId, NativeIntegrationPreviewV1, NativeIntegrationTerminalOutcomeV1, + ObservabilityEnvelopeV1, ObservabilityPayloadV1, ObservabilityTerminalResultV1, + WorkConflictOutcomeLinkedV1, WorkConflictPredictionObservedV1, canonical_sha256, +}; + +use super::{ + BoundedObservabilityProducerV1, ExecutionOwnerFactInputV1, ObservabilityEmissionOutcomeV1, + ObservabilityProducerIdentityV1, execution_owner_fact_envelope, +}; + +const PREDICTION_EVENT_KIND: &str = "work.conflict_prediction.observed.v1"; +const OUTCOME_EVENT_KIND: &str = "work.conflict_outcome.linked.v1"; +const PREDICTION_OPERATION: &str = "preflight_native_integration"; +const OUTCOME_OPERATION: &str = "apply_native_integration"; +/// The preflight tree merge is a deterministic rule oracle, not a calibrated +/// probability model; both revision strings say so rather than claiming a +/// calibration that does not exist. +const CONFLICT_DESCRIPTOR_REVISION: &str = "native-integration-preflight-tree-merge.rule.v1"; +const CONFLICT_CALIBRATION_REVISION: &str = "uncalibrated.rule-oracle.v1"; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum WorkConflictObservationUnavailableV1 { + OwnerUnmounted, + ProducerUnmounted, + ProducerAdmissionUnavailable, + /// The owner result does not adjudicate one mechanical merge relation: + /// reads, refusals, approvals, worktree operations, already-integrated + /// previews, and preflights whose tree merge never completed. + NotAdjudicated, + OwnerEvidenceInvalid, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum WorkConflictObservationResultV1 { + Enqueued { + event_kind: &'static str, + }, + DroppedAtCapacity { + event_kind: &'static str, + }, + Unavailable { + reason: WorkConflictObservationUnavailableV1, + }, +} + +/// Offers the work-conflict fact proved by one native-integration owner +/// result to the long-lived producer. +/// +/// A preflight preview that completed its tree merge proves one mechanical +/// conflict prediction; a terminal apply receipt proves one independently +/// observed outcome linked to that prediction. Queue pressure is the +/// producer's typed drop state and never changes the owner's decided result. +pub fn record_work_conflict_observation( + scope_ref: &str, + producer: Option<&BoundedObservabilityProducerV1>, + surface_operation: &str, + owner_mounted: bool, + result: &NativeIntegrationSurfaceResultV1, + owner_preview: Option<&NativeIntegrationPreviewV1>, +) -> WorkConflictObservationResultV1 { + if !owner_mounted { + return unavailable(WorkConflictObservationUnavailableV1::OwnerUnmounted); + } + let Some(producer) = producer else { + return unavailable(WorkConflictObservationUnavailableV1::ProducerUnmounted); + }; + let identity = producer.identity(); + if identity.authorized_scope_ref != scope_ref { + return unavailable(WorkConflictObservationUnavailableV1::OwnerEvidenceInvalid); + } + let (envelope, event_kind) = match work_conflict_envelope( + identity, + scope_ref, + surface_operation, + result, + owner_preview, + ) { + Ok(Some(built)) => built, + Ok(None) => { + return unavailable(WorkConflictObservationUnavailableV1::NotAdjudicated); + } + Err(_) => { + return unavailable(WorkConflictObservationUnavailableV1::OwnerEvidenceInvalid); + } + }; + match producer.try_emit_owner_fact(envelope) { + Ok(ObservabilityEmissionOutcomeV1::Enqueued) => { + WorkConflictObservationResultV1::Enqueued { event_kind } + } + Ok(ObservabilityEmissionOutcomeV1::DroppedAtCapacity) => { + WorkConflictObservationResultV1::DroppedAtCapacity { event_kind } + } + Err(_) => unavailable(WorkConflictObservationUnavailableV1::ProducerAdmissionUnavailable), + } +} + +const fn unavailable( + reason: WorkConflictObservationUnavailableV1, +) -> WorkConflictObservationResultV1 { + WorkConflictObservationResultV1::Unavailable { reason } +} + +fn work_conflict_envelope( + identity: &ObservabilityProducerIdentityV1, + scope_ref: &str, + surface_operation: &str, + result: &NativeIntegrationSurfaceResultV1, + owner_preview: Option<&NativeIntegrationPreviewV1>, +) -> Result, &'static str> { + match result { + NativeIntegrationSurfaceResultV1::Preview(preview) + if surface_operation == PREDICTION_OPERATION => + { + Ok(prediction_envelope(identity, scope_ref, preview)? + .map(|envelope| (envelope, PREDICTION_EVENT_KIND))) + } + NativeIntegrationSurfaceResultV1::Receipt(receipt) + if surface_operation == OUTCOME_OPERATION => + { + // A receipt exists only when native git ran the transaction to a + // terminal state, and the daemon threads the exact durable + // preview it applied. A missing or mismatched preview is invalid + // owner evidence, not a reason to guess the prediction identity. + let preview = owner_preview + .filter(|preview| { + preview.validate().is_ok() + && preview.preview_id == receipt.status.preview_id + && preview.preview_digest == receipt.status.preview_digest + }) + .ok_or("work_conflict_preview_binding")?; + outcome_envelope(identity, scope_ref, receipt, preview) + .map(|envelope| Some((envelope, OUTCOME_EVENT_KIND))) + } + _ => Ok(None), + } +} + +fn prediction_envelope( + identity: &ObservabilityProducerIdentityV1, + scope_ref: &str, + preview: &NativeIntegrationPreviewProjectionV1, +) -> Result, &'static str> { + let prediction = match &preview.disposition { + NativeIntegrationPreviewDispositionV1::MechanicalIntegrationEligible(_) => { + ConflictPredictionV1::NoConflict + } + NativeIntegrationPreviewDispositionV1::NativeConflict { .. } => { + ConflictPredictionV1::Conflict + } + // The mechanical oracle completed but truthfully defers the + // integration verdict to semantic review. + NativeIntegrationPreviewDispositionV1::SemanticReviewRequired { .. } => { + ConflictPredictionV1::Abstained + } + // Already-integrated work has no pending merge relation to predict, + // and a partial or unavailable preflight never completed its tree + // merge; neither observed a prediction. + NativeIntegrationPreviewDispositionV1::AlreadyIntegrated + | NativeIntegrationPreviewDispositionV1::Partial { .. } + | NativeIntegrationPreviewDispositionV1::Unavailable { .. } => return Ok(None), + }; + let prediction_ref = prediction_ref(&preview.preview_id, &preview.preview_digest)?; + let observation = WorkConflictPredictionObservedV1 { + prediction_ref: prediction_ref.clone(), + kind: ConflictKindV1::Mechanical, + prediction, + score_kind: ConflictScoreKindV1::Rule, + descriptor_revision: CONFLICT_DESCRIPTOR_REVISION.to_owned(), + calibration_revision: CONFLICT_CALIBRATION_REVISION.to_owned(), + // The preflight evaluates exactly one frozen source-to-destination + // merge relation, and it observed that whole relation. + eligible_relation_count: 1, + // The prediction stands exactly as long as the preview it is bound + // to: an expired preview can never be applied, so its prediction can + // never be adjudicated. + expires_at_micros: preview.expires_at.0, + coverage: CoverageStateV1::Known, + local_anchor_refs: Vec::new(), + }; + execution_owner_fact_envelope( + identity, + scope_ref, + ExecutionOwnerFactInputV1 { + owner_transition_ref: &prediction_ref, + operation: PREDICTION_OPERATION, + event_time: preview.created_at, + valid_from: Some(preview.created_at), + valid_until: Some(preview.expires_at), + terminal_result: Some(ObservabilityTerminalResultV1::Succeeded), + coverage: CoverageStateV1::Known, + payload: ObservabilityPayloadV1::WorkConflictPrediction(observation), + }, + ) + .map(Some) +} + +fn outcome_envelope( + identity: &ObservabilityProducerIdentityV1, + scope_ref: &str, + receipt: &NativeIntegrationReceiptProjectionV1, + preview: &NativeIntegrationPreviewV1, +) -> Result { + // The prediction was observed at preview creation; a receipt completing + // before its own preview existed is inconsistent owner evidence. + let horizon_micros = receipt + .completed_at + .0 + .checked_sub(preview.created_at.0) + .and_then(|elapsed| u64::try_from(elapsed).ok()) + .ok_or("work_conflict_outcome_horizon")?; + let (outcome, adjudicator, coverage, terminal_result) = match receipt.terminal_outcome { + // Native git integrated exactly the predicted relation and committed: + // an independent no-conflict adjudication. + NativeIntegrationTerminalOutcomeV1::Committed => ( + ConflictOutcomeV1::NoConflict, + ConflictAdjudicatorV1::NativeGit, + CoverageStateV1::Known, + ObservabilityTerminalResultV1::Succeeded, + ), + // Native git aborted before mutating (drift or cancellation): the + // predicted relation left observation without adjudication. + NativeIntegrationTerminalOutcomeV1::AbortedNoChange => ( + ConflictOutcomeV1::Censored, + ConflictAdjudicatorV1::None, + CoverageStateV1::Known, + ObservabilityTerminalResultV1::Unknown, + ), + // The mutation was undone; the precomputed candidate tree failed for + // an environmental reason, which adjudicates neither conflict nor + // no-conflict. + NativeIntegrationTerminalOutcomeV1::RolledBack => ( + ConflictOutcomeV1::Unknown, + ConflictAdjudicatorV1::None, + CoverageStateV1::Known, + ObservabilityTerminalResultV1::Failed, + ), + NativeIntegrationTerminalOutcomeV1::NeedsInspection => ( + ConflictOutcomeV1::Unknown, + ConflictAdjudicatorV1::None, + CoverageStateV1::Unknown, + ObservabilityTerminalResultV1::Partial, + ), + }; + let prediction_ref = prediction_ref(&preview.preview_id, &preview.preview_digest)?; + let observation = WorkConflictOutcomeLinkedV1 { + prediction_ref: prediction_ref.clone(), + kind: ConflictKindV1::Mechanical, + outcome, + adjudicator, + horizon_micros, + coverage, + correction_revision: 0, + }; + execution_owner_fact_envelope( + identity, + scope_ref, + ExecutionOwnerFactInputV1 { + owner_transition_ref: &prediction_ref, + operation: OUTCOME_OPERATION, + event_time: receipt.completed_at, + valid_from: Some(receipt.completed_at), + valid_until: Some(receipt.completed_at), + terminal_result: Some(terminal_result), + coverage, + payload: ObservabilityPayloadV1::WorkConflictOutcome(observation), + }, + ) +} + +/// Deterministic prediction identity shared by the preflight prediction and +/// the apply-linked outcome: both sides hold the exact preview identity and +/// content digest, so both derive the same reference without exporting either +/// raw identifier. +fn prediction_ref( + preview_id: &NativeIntegrationPreviewId, + preview_digest: &ManifestDigest, +) -> Result { + let digest = canonical_sha256(&( + "tracedecay.work-conflict.prediction-ref.v1", + preview_id.as_str(), + preview_digest.as_str(), + )) + .map_err(|_| "work_conflict_prediction_ref")?; + Ok(format!("work-conflict:{}", digest.as_str())) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + use tracedecay_application::{ + NativeIntegrationSnapshotProjectionV1, NativeIntegrationStatusProjectionV1, + ObservabilityHorizonV1, ObservabilityQueryPort, ObservabilityQueryV1, + }; + use tracedecay_domain::{ + BranchStackEdgeV1, BranchStackId, BranchStackNodeV1, BranchStackRevisionId, + BranchStackRevisionV1, BranchStackSourceV1, CommitId, FrozenBranchStackSnapshotV1, + GitHeadStateV1, GitObjectFormatV1, GitOidV1, GitOperationStateV1, + MechanicalIntegrationModeV1, NativeIntegrationDirectionV1, NativeIntegrationPhaseV1, + NativeIntegrationRepositorySnapshotV1, NativeIntegrationSelectionV1, + NativeIntegrationTransactionId, ProjectId, RefId, RepositoryId, StackNodeId, UtcMicros, + WorktreeId, WorktreeInventoryEpoch, WorktreeInventorySnapshotId, + }; + use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; + + use crate::observability::RegisteredObservabilityPortV1; + + fn digest(byte: char) -> ManifestDigest { + ManifestDigest::new(format!("sha256:{}", byte.to_string().repeat(64))).unwrap() + } + + fn oid(byte: char) -> GitOidV1 { + GitOidV1::new(byte.to_string().repeat(40)).unwrap() + } + + fn identity(scope_ref: &str) -> ObservabilityProducerIdentityV1 { + ObservabilityProducerIdentityV1 { + authorized_scope_ref: scope_ref.to_owned(), + process_boot_id: "boot:work-conflict-test".to_owned(), + producer_revision: "work-conflict-test.v1".to_owned(), + configuration_revision: "work-conflict-test-config.v1".to_owned(), + policy_revision: "work-conflict-test-policy.v1".to_owned(), + } + } + + fn sealed_preview( + disposition: NativeIntegrationPreviewDispositionV1, + ) -> NativeIntegrationPreviewV1 { + let project_id = ProjectId::new("private-project").unwrap(); + let repository_id = RepositoryId::new("private-repository").unwrap(); + let source_node = StackNodeId::new("node.work-conflict.source").unwrap(); + let destination_node = StackNodeId::new("node.work-conflict.destination").unwrap(); + let source_ref = RefId::new("refs/heads/private-source-ref").unwrap(); + let destination_ref = RefId::new("refs/heads/private-target-ref").unwrap(); + let destination_worktree = WorktreeId::new("worktree.work-conflict.destination").unwrap(); + let revision = BranchStackRevisionV1::new( + BranchStackId::new("stack.work-conflict").unwrap(), + BranchStackRevisionId::new("revision.work-conflict").unwrap(), + WorktreeInventorySnapshotId::new("inventory.work-conflict").unwrap(), + WorktreeInventoryEpoch::new(1).unwrap(), + BranchStackSourceV1::ExplicitDeclaration, + vec![ + BranchStackNodeV1 { + node_id: source_node.clone(), + project_id: project_id.clone(), + repository_id: repository_id.clone(), + reference: source_ref.clone(), + tip: CommitId::new("1".repeat(40)).unwrap(), + worktree_id: Some(WorktreeId::new("worktree.work-conflict.source").unwrap()), + }, + BranchStackNodeV1 { + node_id: destination_node.clone(), + project_id: project_id.clone(), + repository_id: repository_id.clone(), + reference: destination_ref.clone(), + tip: CommitId::new("2".repeat(40)).unwrap(), + worktree_id: Some(destination_worktree.clone()), + }, + ], + vec![BranchStackEdgeV1 { + dependency: source_node.clone(), + dependent: destination_node.clone(), + }], + ) + .unwrap(); + let selection = NativeIntegrationSelectionV1::DeclaredStackEdge( + FrozenBranchStackSnapshotV1::new( + revision, + source_node, + destination_node, + NativeIntegrationDirectionV1::PropagateDependencyToDependent, + UtcMicros(10), + ) + .unwrap(), + ); + let repository_snapshot = NativeIntegrationRepositorySnapshotV1 { + project_id, + repository_id, + source_worktree_id: selection.source_worktree_id().unwrap().cloned(), + destination_worktree_id: Some(destination_worktree), + source_ref, + destination_ref: destination_ref.clone(), + source_tip: oid('1'), + destination_tip: oid('2'), + source_tree: oid('3'), + destination_tree: oid('4'), + merge_base: oid('5'), + dependency_commits: vec![oid('1')], + destination_head: GitHeadStateV1::Attached { + branch: destination_ref.as_str().to_owned(), + commit: oid('2'), + }, + refs_digest: digest('b'), + index_digest: digest('c'), + worktree_digest: digest('d'), + attributes_digest: digest('e'), + operation_state: GitOperationStateV1::None, + clean: true, + object_format: GitObjectFormatV1::Sha1, + adapter_revision: "gix.work-conflict.v1".to_owned(), + captured_at: UtcMicros(11), + digest: digest('0'), + } + .seal() + .unwrap(); + let eligible = matches!( + disposition, + NativeIntegrationPreviewDispositionV1::MechanicalIntegrationEligible(_) + ); + NativeIntegrationPreviewV1 { + preview_id: NativeIntegrationPreviewId::new("preview.work-conflict.fixture").unwrap(), + selection, + repository_snapshot, + grant_digest: digest('f'), + policy_digest: digest('1'), + graph_revision_digest: digest('2'), + test_revision_digest: digest('3'), + schema_revision_digest: digest('4'), + migration_revision_digest: digest('5'), + disposition, + candidate_tree: eligible.then(|| oid('6')), + ordered_commits: vec![oid('1')], + created_at: UtcMicros(12), + expires_at: UtcMicros(1_000), + preview_digest: digest('0'), + } + .seal() + .unwrap() + } + + fn eligible_preview() -> NativeIntegrationPreviewV1 { + sealed_preview( + NativeIntegrationPreviewDispositionV1::MechanicalIntegrationEligible( + MechanicalIntegrationModeV1::FastForward, + ), + ) + } + + fn preview_result(preview: &NativeIntegrationPreviewV1) -> NativeIntegrationSurfaceResultV1 { + NativeIntegrationSurfaceResultV1::Preview( + NativeIntegrationPreviewProjectionV1::project(preview).unwrap(), + ) + } + + fn projection_result( + disposition: NativeIntegrationPreviewDispositionV1, + ) -> NativeIntegrationSurfaceResultV1 { + NativeIntegrationSurfaceResultV1::Preview(NativeIntegrationPreviewProjectionV1 { + preview_id: NativeIntegrationPreviewId::new("preview.work-conflict.projection") + .unwrap(), + preview_digest: digest('a'), + selection: NativeIntegrationSnapshotProjectionV1 { + selection_digest: digest('b'), + project_id: ProjectId::new("private-project").unwrap(), + repository_id: RepositoryId::new("private-repository").unwrap(), + source_ref: RefId::new("private-source-ref").unwrap(), + destination_ref: RefId::new("private-target-ref").unwrap(), + inventory_epoch: WorktreeInventoryEpoch::new(1).unwrap(), + frozen_at: UtcMicros(10), + }, + disposition, + ordered_commit_count: 1, + created_at: UtcMicros(12), + expires_at: UtcMicros(1_000), + }) + } + + fn receipt_result( + preview: &NativeIntegrationPreviewV1, + terminal_outcome: NativeIntegrationTerminalOutcomeV1, + completed_at: UtcMicros, + ) -> NativeIntegrationSurfaceResultV1 { + NativeIntegrationSurfaceResultV1::Receipt(NativeIntegrationReceiptProjectionV1 { + status: NativeIntegrationStatusProjectionV1 { + transaction_id: NativeIntegrationTransactionId::new( + "transaction.work-conflict.fixture", + ) + .unwrap(), + preview_id: preview.preview_id.clone(), + preview_digest: preview.preview_digest.clone(), + repository_id: preview.repository_snapshot.repository_id.clone(), + destination_ref: preview.repository_snapshot.destination_ref.clone(), + phase: NativeIntegrationPhaseV1::Terminal, + phase_revision: 5, + cancellation_requested: false, + terminal_outcome: Some(terminal_outcome), + updated_at: completed_at, + }, + terminal_outcome, + final_ref_tip: "private-final-object".to_owned(), + final_tree: "private-final-tree".to_owned(), + completed_at, + receipt_digest: digest('9'), + }) + } + + #[tokio::test] + async fn preflight_prediction_and_apply_outcome_persist_one_linked_pair() { + let _pin = tracedecay_runtime_core::config::PinnedUserDataDir::new(); + let project = tempfile::tempdir().expect("project dir"); + let project_id = ProjectId::new("project.work-conflict.durable").unwrap(); + let scope_ref = project_id.as_str().to_owned(); + let runtime = RegisteredGlobalDbTestRuntime::project( + tracedecay_runtime_core::storage::default_profile_root().expect("profile root"), + project.path(), + project_id.clone(), + ) + .await + .expect("registered runtime"); + let database = runtime.project_database_arc().expect("project database"); + let producer = + BoundedObservabilityProducerV1::start(database.clone(), identity(&scope_ref), 64) + .expect("producer"); + + let preview = eligible_preview(); + assert_eq!( + record_work_conflict_observation( + &scope_ref, + Some(&producer), + PREDICTION_OPERATION, + true, + &preview_result(&preview), + Some(&preview), + ), + WorkConflictObservationResultV1::Enqueued { + event_kind: PREDICTION_EVENT_KIND, + } + ); + assert_eq!( + record_work_conflict_observation( + &scope_ref, + Some(&producer), + OUTCOME_OPERATION, + true, + &receipt_result( + &preview, + NativeIntegrationTerminalOutcomeV1::Committed, + UtcMicros(20), + ), + Some(&preview), + ), + WorkConflictObservationResultV1::Enqueued { + event_kind: OUTCOME_EVENT_KIND, + } + ); + // A denied scope writes nothing. + assert_eq!( + record_work_conflict_observation( + "project.work-conflict.foreign", + Some(&producer), + PREDICTION_OPERATION, + true, + &preview_result(&preview), + Some(&preview), + ), + WorkConflictObservationResultV1::Unavailable { + reason: WorkConflictObservationUnavailableV1::OwnerEvidenceInvalid, + } + ); + producer.shutdown().await.expect("flush producer"); + drop(producer); + + let port = RegisteredObservabilityPortV1::new(database.as_ref()); + let page = port + .query(ObservabilityQueryV1 { + authorized_scope_ref: scope_ref.clone(), + event_kinds: vec![ + PREDICTION_EVENT_KIND.to_owned(), + OUTCOME_EVENT_KIND.to_owned(), + ], + horizon: ObservabilityHorizonV1 { + since_micros: 0, + until_micros: 10_000, + }, + after_watermark: None, + limit: 64, + }) + .await + .expect("durable conflict query"); + assert_eq!(page.events.len(), 2, "exactly one linked pair persists"); + + let prediction = page + .events + .iter() + .find(|event| event.event_kind == PREDICTION_EVENT_KIND) + .expect("persisted prediction"); + let outcome = page + .events + .iter() + .find(|event| event.event_kind == OUTCOME_EVENT_KIND) + .expect("persisted outcome"); + let ObservabilityPayloadV1::WorkConflictPrediction(prediction_payload) = + &prediction.payload + else { + panic!("wrong prediction payload family"); + }; + let ObservabilityPayloadV1::WorkConflictOutcome(outcome_payload) = &outcome.payload else { + panic!("wrong outcome payload family"); + }; + assert_eq!( + prediction_payload.prediction_ref, outcome_payload.prediction_ref, + "outcome links the exact prediction identity" + ); + assert_eq!( + prediction.trace_id, outcome.trace_id, + "prediction and outcome share one owner trace" + ); + assert_eq!(prediction_payload.kind, ConflictKindV1::Mechanical); + assert_eq!( + prediction_payload.prediction, + ConflictPredictionV1::NoConflict + ); + assert_eq!(prediction_payload.score_kind, ConflictScoreKindV1::Rule); + assert_eq!(prediction_payload.eligible_relation_count, 1); + assert_eq!( + prediction_payload.expires_at_micros, preview.expires_at.0, + "prediction expiry is the preview's own expiry" + ); + assert_eq!(outcome_payload.kind, ConflictKindV1::Mechanical); + assert_eq!(outcome_payload.outcome, ConflictOutcomeV1::NoConflict); + assert_eq!( + outcome_payload.adjudicator, + ConflictAdjudicatorV1::NativeGit + ); + assert_eq!( + outcome_payload.horizon_micros, 8, + "horizon is apply completion minus preview creation" + ); + assert_eq!(outcome_payload.correction_revision, 0); + + let wire = serde_json::to_string(&page.events).expect("serialize persisted pair"); + for prohibited in [ + "private-project", + "private-repository", + "private-source-ref", + "private-target-ref", + "private-final-object", + "private-final-tree", + "preview.work-conflict.fixture", + ] { + assert!(!wire.contains(prohibited), "leaked {prohibited}"); + } + } + + #[test] + fn conflict_preview_predicts_conflict_and_semantic_review_abstains() { + let identity = identity("project.scope"); + let (envelope, event_kind) = work_conflict_envelope( + &identity, + "project.scope", + PREDICTION_OPERATION, + &projection_result(NativeIntegrationPreviewDispositionV1::NativeConflict { + conflict_digest: digest('8'), + }), + None, + ) + .expect("conflict prediction") + .expect("adjudicated disposition"); + assert_eq!(event_kind, PREDICTION_EVENT_KIND); + let ObservabilityPayloadV1::WorkConflictPrediction(payload) = &envelope.payload else { + panic!("wrong payload family"); + }; + assert_eq!(payload.prediction, ConflictPredictionV1::Conflict); + assert_eq!(payload.descriptor_revision, CONFLICT_DESCRIPTOR_REVISION); + assert_eq!(payload.calibration_revision, CONFLICT_CALIBRATION_REVISION); + + let (envelope, _) = work_conflict_envelope( + &identity, + "project.scope", + PREDICTION_OPERATION, + &projection_result( + NativeIntegrationPreviewDispositionV1::SemanticReviewRequired { + evidence_digest: digest('7'), + }, + ), + None, + ) + .expect("semantic-review prediction") + .expect("abstaining disposition"); + let ObservabilityPayloadV1::WorkConflictPrediction(payload) = &envelope.payload else { + panic!("wrong payload family"); + }; + assert_eq!(payload.prediction, ConflictPredictionV1::Abstained); + } + + #[test] + fn unadjudicated_dispositions_reads_and_wrong_operations_emit_nothing() { + let identity = identity("project.scope"); + for disposition in [ + NativeIntegrationPreviewDispositionV1::AlreadyIntegrated, + NativeIntegrationPreviewDispositionV1::Partial { + reason: tracedecay_domain::NativeIntegrationUnavailabilityV1::PartialEvidence, + }, + NativeIntegrationPreviewDispositionV1::Unavailable { + reason: tracedecay_domain::NativeIntegrationUnavailabilityV1::Denied, + }, + ] { + assert!( + work_conflict_envelope( + &identity, + "project.scope", + PREDICTION_OPERATION, + &projection_result(disposition), + None, + ) + .expect("typed non-adjudication") + .is_none() + ); + } + // A preview reached under any operation but preflight proves nothing. + assert!( + work_conflict_envelope( + &identity, + "project.scope", + "native_integration_status", + &projection_result( + NativeIntegrationPreviewDispositionV1::MechanicalIntegrationEligible( + MechanicalIntegrationModeV1::FastForward, + ), + ), + None, + ) + .expect("wrong operation") + .is_none() + ); + // Refused applies never reach a receipt; the unavailable result + // truthfully adjudicates nothing. + assert!( + work_conflict_envelope( + &identity, + "project.scope", + OUTCOME_OPERATION, + &NativeIntegrationSurfaceResultV1::unavailable( + tracedecay_application::NativeIntegrationSurfaceUnavailableV1::Denied, + ), + None, + ) + .expect("refused apply") + .is_none() + ); + } + + #[test] + fn aborted_rolled_back_and_uninspectable_applies_never_claim_adjudication() { + let identity = identity("project.scope"); + let preview = eligible_preview(); + let cases = [ + ( + NativeIntegrationTerminalOutcomeV1::AbortedNoChange, + ConflictOutcomeV1::Censored, + CoverageStateV1::Known, + ), + ( + NativeIntegrationTerminalOutcomeV1::RolledBack, + ConflictOutcomeV1::Unknown, + CoverageStateV1::Known, + ), + ( + NativeIntegrationTerminalOutcomeV1::NeedsInspection, + ConflictOutcomeV1::Unknown, + CoverageStateV1::Unknown, + ), + ]; + for (terminal_outcome, expected_outcome, expected_coverage) in cases { + let (envelope, _) = work_conflict_envelope( + &identity, + "project.scope", + OUTCOME_OPERATION, + &receipt_result(&preview, terminal_outcome, UtcMicros(20)), + Some(&preview), + ) + .expect("terminal receipt") + .expect("linked outcome"); + let ObservabilityPayloadV1::WorkConflictOutcome(payload) = &envelope.payload else { + panic!("wrong payload family"); + }; + assert_eq!(payload.outcome, expected_outcome); + assert_eq!(payload.adjudicator, ConflictAdjudicatorV1::None); + assert_eq!(payload.coverage, expected_coverage); + } + } + + #[test] + fn missing_or_mismatched_owner_evidence_is_a_typed_refusal_never_a_panic() { + let identity = identity("project.scope"); + let preview = eligible_preview(); + let receipt = receipt_result( + &preview, + NativeIntegrationTerminalOutcomeV1::Committed, + UtcMicros(20), + ); + // A receipt without its durable preview cannot name a prediction. + assert!( + work_conflict_envelope( + &identity, + "project.scope", + OUTCOME_OPERATION, + &receipt, + None + ) + .is_err() + ); + // A preview that does not match the receipt identity is foreign + // evidence, not a linkable prediction. + let mut foreign = eligible_preview(); + foreign.preview_id = + NativeIntegrationPreviewId::new("preview.work-conflict.foreign").unwrap(); + let foreign = foreign.seal().unwrap(); + assert!( + work_conflict_envelope( + &identity, + "project.scope", + OUTCOME_OPERATION, + &receipt, + Some(&foreign), + ) + .is_err() + ); + // A receipt completing before its preview existed is inconsistent. + assert!( + work_conflict_envelope( + &identity, + "project.scope", + OUTCOME_OPERATION, + &receipt_result( + &preview, + NativeIntegrationTerminalOutcomeV1::Committed, + UtcMicros(11), + ), + Some(&preview), + ) + .is_err() + ); + } + + #[test] + fn absent_producer_or_unmounted_owner_is_typed_unavailable() { + let preview = eligible_preview(); + assert_eq!( + record_work_conflict_observation( + "project.scope", + None, + PREDICTION_OPERATION, + true, + &preview_result(&preview), + Some(&preview), + ), + WorkConflictObservationResultV1::Unavailable { + reason: WorkConflictObservationUnavailableV1::ProducerUnmounted, + } + ); + assert_eq!( + record_work_conflict_observation( + "project.scope", + None, + PREDICTION_OPERATION, + false, + &preview_result(&preview), + Some(&preview), + ), + WorkConflictObservationResultV1::Unavailable { + reason: WorkConflictObservationUnavailableV1::OwnerUnmounted, + } + ); + } +} diff --git a/crates/tracedecay-usecases/tests/github_stack_drift_observability.rs b/crates/tracedecay-usecases/tests/github_stack_drift_observability.rs index 11b43a728..fb80b32cc 100644 --- a/crates/tracedecay-usecases/tests/github_stack_drift_observability.rs +++ b/crates/tracedecay-usecases/tests/github_stack_drift_observability.rs @@ -1,4 +1,5 @@ use std::collections::BTreeSet; +use std::sync::Arc; use tracedecay_application::{ CancellationContext, CapabilityGrantId, CapabilityGrantSnapshot, Deadline, DisclosureClass, @@ -16,6 +17,7 @@ use tracedecay_domain::{ use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; use tracedecay_tool_catalog::{CapabilityId, UseCaseId}; use tracedecay_usecases::{ + advisory::GitHubStackObservabilityV1, observability::{ BoundedObservabilityProducerV1, GitHubStackDriftObservationResultV1, GitHubStackDriftObservationUnavailableV1, GitHubStackProbeOwnerV1, @@ -110,6 +112,96 @@ fn drift( .expect("canonical drift observation") } +/// The review refresh owner's Observatory lane (`GitHubStackObservabilityV1:: +/// record`, invoked from `GitHubReviewRuntimeOwnerV1::refresh` after each +/// coordinator observation) must persist one capability receipt plus one +/// receipt per exact drift interval through the bounded producer. +#[tokio::test] +async fn review_owner_observability_lane_records_capability_and_drift_receipts() { + let _pin = tracedecay_runtime_core::config::PinnedUserDataDir::new(); + let project = tempfile::tempdir().expect("project"); + let scope = resolved_scope("owner-lane"); + let runtime = RegisteredGlobalDbTestRuntime::project( + tracedecay_runtime_core::storage::default_profile_root().expect("profile root"), + project.path(), + scope.project_id.clone(), + ) + .await + .expect("registered runtime"); + let database = runtime.project_database_arc().expect("project database"); + let producer = Arc::new( + BoundedObservabilityProducerV1::start( + database.clone(), + producer_identity(&scope.project_id), + 64, + ) + .expect("producer"), + ); + let lane = GitHubStackObservabilityV1 { + probe_owner: GitHubStackProbeOwnerV1::mount( + scope.clone(), + safe_work_topology_policy_v1(), + "octo-org", + "stack-repository", + true, + ) + .expect("stack probe owner"), + producer: Arc::clone(&producer), + observation_db: database.clone(), + }; + let coordinator = DaemonGitHubStackCoordinatorV1::default(); + coordinator + .register_scope(&scope, GitHubStackedPullRequestPolicyV1::Disabled) + .expect("register scope"); + let source_binding = source_binding(&scope); + let mut observation = coordinator + .observe_policy( + scope.clone(), + ProviderId::new("provider.github").expect("provider"), + source_binding.clone(), + UtcMicros(1_000), + ) + .expect("policy observation"); + let open = drift(&scope, 2_000, 2_000, IntervalStateV1::Open); + observation.observed_at = open.observed_at; + observation.drift_observations = vec![open]; + + lane.record(&source_binding, &observation); + + producer.shutdown().await.expect("flush producer"); + drop(lane); + drop(producer); + let port = RegisteredObservabilityPortV1::new(database.as_ref()); + let query = |event_kind: &str| ObservabilityQueryV1 { + authorized_scope_ref: scope.project_id.as_str().to_owned(), + event_kinds: vec![event_kind.to_owned()], + horizon: ObservabilityHorizonV1 { + since_micros: 0, + until_micros: 10_000, + }, + after_watermark: None, + limit: 16, + }; + let capability_page = port + .query(query("work.github_stack_capability.observed.v1")) + .await + .expect("capability receipts"); + assert_eq!( + capability_page.events.len(), + 1, + "one coordinator observation produces exactly one capability receipt" + ); + let drift_page = port + .query(query("work.stack_drift.observed.v1")) + .await + .expect("drift receipts"); + assert_eq!( + drift_page.events.len(), + 1, + "one open drift interval produces exactly one drift receipt" + ); +} + #[tokio::test] async fn coordinator_drift_is_durable_closed_monotone_and_scope_denial_writes_nothing() { let _pin = tracedecay_runtime_core::config::PinnedUserDataDir::new(); diff --git a/src/daemon.rs b/src/daemon.rs index 7993b0479..00f40c42c 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -166,6 +166,7 @@ use engine::{ ensure_context_scout_owner_before_advertising, ensure_git_index_transactions_for_mutation_owners, }; +mod adoption_observation; mod automation_observation; pub(crate) use automation_observation::{ project_run_observation_producer as project_automation_observation_producer, diff --git a/src/daemon/adoption_observation.rs b/src/daemon/adoption_observation.rs new file mode 100644 index 000000000..c9645b9a8 --- /dev/null +++ b/src/daemon/adoption_observation.rs @@ -0,0 +1,272 @@ +//! Once-per-project-open adoption-eligibility census over the composed +//! application capability catalog. +//! +//! `application_catalog_contributions` is the one closed composition +//! authority the daemon serves, so enumerating it is a complete census +//! (`Known` coverage). Per family: `eligible` = every composed capability, +//! `enabled` = default-profile eligible, `available` = enabled and callable — +//! the exact filter stages of `catalog_composition::application_profile` in +//! funnel order. Families with no composed capability are not emitted: a +//! `Known`-zero census would falsely claim their population is empty. + +use std::collections::BTreeMap; +use std::path::Path; + +use tracedecay_application::{ + APPLICATION_DEFAULT_PROFILE_ID, ApplicationContractError, application_catalog_contributions, +}; +use tracedecay_domain::{AdoptionEligibilityObservedV1, CoverageStateV1}; +use tracedecay_tool_catalog::{CatalogContributionV1, ProfileId}; +use tracedecay_usecases::observability::record_adoption_eligibility; + +use super::log_daemon_event; +use crate::global_db::RegisteredGlobalDb; + +/// Composed capability namespaces mapped onto the closed adoption capability +/// families (`AdoptionEligibilityObservedV1::validate`). Prefix, not equality: +/// each namespace is owned by exactly one catalog contribution. +const FAMILY_NAMESPACES: &[(&str, &str)] = &[ + ("capability.application.symbol-search", "retrieval"), + ("capability.application.primitive.", "retrieval"), + ("capability.application.code-query.", "retrieval"), + ("capability.application.context-scout-", "context_scout"), + ("capability.application.feedback.", "feedback"), + ("capability.application.git.", "git"), + ("capability.application.github-stack.", "git"), + ("capability.application.native-integration.", "git"), + ("capability.git.", "git"), + ("capability.application.lsp.", "lsp"), + // The Observatory read surface exposes only the canonical observability + // and cost read models — the analytics family's one composed capability. + ("capability.application.observatory-read", "analytics"), +]; + +fn adoption_family(capability_id: &str) -> Option<&'static str> { + FAMILY_NAMESPACES + .iter() + .find_map(|(namespace, family)| capability_id.starts_with(namespace).then_some(*family)) +} + +/// Enumerates the complete composed catalog into per-family eligibility +/// observations. Only families with a non-zero eligible population appear. +pub(in crate::daemon) fn adoption_eligibility_census() +-> Result, ApplicationContractError> { + let contributions = application_catalog_contributions()?; + let default_profile = ProfileId::new(APPLICATION_DEFAULT_PROFILE_ID)?; + let mut families: BTreeMap<&'static str, AdoptionEligibilityObservedV1> = BTreeMap::new(); + for capability in contributions + .iter() + .flat_map(CatalogContributionV1::capabilities) + { + let Some(family) = adoption_family(capability.capability_id().as_str()) else { + continue; + }; + let observation = families + .entry(family) + .or_insert_with(|| AdoptionEligibilityObservedV1 { + capability: family.to_owned(), + eligible: 0, + enabled: 0, + available: 0, + }); + observation.eligible = observation.eligible.saturating_add(1); + if capability.profile_eligibility().contains(&default_profile) { + observation.enabled = observation.enabled.saturating_add(1); + if capability.availability().is_callable() { + observation.available = observation.available.saturating_add(1); + } + } + } + Ok(families.into_values().collect()) +} + +/// Records the project-open adoption-eligibility census through the +/// project-bound observation authority. Telemetry only: every failure is +/// logged and discarded so project open never blocks or fails on it. +pub(in crate::daemon) async fn record_project_open_adoption_census( + db: &RegisteredGlobalDb, + project_root: &Path, +) { + let observations = match adoption_eligibility_census() { + Ok(observations) => observations, + Err(error) => { + log_daemon_event( + "adoption_observation", + &[ + ("project", project_root.display().to_string()), + ("outcome", "unavailable".to_owned()), + ("reason", error.to_string()), + ], + ); + return; + } + }; + for observation in observations { + let family = observation.capability.clone(); + // The census enumerated the whole composed catalog, so each family + // observation is a complete count of its eligible population. + if let Err(error) = + record_adoption_eligibility(db, CoverageStateV1::Known, observation).await + { + log_daemon_event( + "adoption_observation", + &[ + ("project", project_root.display().to_string()), + ("family", family), + ("outcome", "failed".to_owned()), + ("reason", format!("{error:?}")), + ], + ); + } + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use std::collections::BTreeMap; + + use tracedecay_application::{ + ObservabilityHorizonV1, ObservabilityQueryPort, ObservabilityQueryV1, + }; + use tracedecay_domain::{CoverageStateV1, ObservabilityPayloadV1, ProjectId}; + use tracedecay_usecases::observability::RegisteredObservabilityPortV1; + + use super::*; + + /// Composed namespaces deliberately outside the closed adoption family + /// vocabulary. Configuration, retained memory/LCM, and source editing + /// have no adoption family, so their capabilities are excluded from every + /// census rather than force-fitted into an unrelated family. + const OUT_OF_SCOPE_NAMESPACES: &[&str] = &[ + "capability.application.configuration.", + "capability.application.retained.", + "capability.application.source-edit.", + ]; + + /// Families the composed application catalog can truthfully census today. + const COMPOSED_FAMILIES: &[&str] = &[ + "retrieval", + "context_scout", + "feedback", + "git", + "lsp", + "analytics", + ]; + + #[test] + fn every_composed_capability_is_classified_or_deliberately_out_of_scope() { + let contributions = application_catalog_contributions().expect("composed catalog"); + for capability in contributions + .iter() + .flat_map(CatalogContributionV1::capabilities) + { + let id = capability.capability_id().as_str(); + let classified = adoption_family(id).is_some() + || OUT_OF_SCOPE_NAMESPACES + .iter() + .any(|namespace| id.starts_with(namespace)); + assert!( + classified, + "{id} joined the composed catalog without an adoption-census decision; \ + map its namespace to a closed family or record it as out of scope" + ); + } + } + + #[test] + fn census_counts_hold_the_funnel_order_for_every_composed_family() { + let census = adoption_eligibility_census().expect("catalog census"); + assert!(!census.is_empty(), "the composed catalog census is empty"); + let by_family: BTreeMap<&str, &AdoptionEligibilityObservedV1> = census + .iter() + .map(|observation| (observation.capability.as_str(), observation)) + .collect(); + for family in COMPOSED_FAMILIES { + let observation = by_family + .get(family) + .unwrap_or_else(|| panic!("{family} family missing from the catalog census")); + assert!( + observation.eligible > 0, + "{family} must census a non-zero eligible population" + ); + assert!(observation.enabled <= observation.eligible); + assert!(observation.available <= observation.enabled); + } + // The default profile serves callable retrieval capabilities, so the + // census must observe them as enabled and available, not merely + // composed. + assert!(by_family["retrieval"].available > 0); + // Families this catalog authority does not compose must be absent + // instead of claiming a Known-zero eligible population. + for family in [ + "automation", + "work", + "workflow", + "hooks", + "mcp", + "dashboard", + ] { + assert!( + !by_family.contains_key(family), + "{family} has no composed catalog capability and must not be emitted" + ); + } + } + + #[tokio::test] + async fn project_open_census_persists_known_coverage_family_observations() { + let _pin = crate::config::PinnedUserDataDir::new(); + let project = tempfile::tempdir().expect("project"); + let project_id = ProjectId::new("project.adoption.census").expect("project id"); + let runtime = crate::global_db::tests::harness::RegisteredGlobalDbTestRuntime::project( + tracedecay_runtime_core::storage::default_profile_root().expect("profile root"), + project.path(), + project_id.clone(), + ) + .await + .expect("registered runtime"); + let db = runtime.project_database().expect("project database"); + + record_project_open_adoption_census(db, project.path()).await; + + let page = RegisteredObservabilityPortV1::new(db) + .query(ObservabilityQueryV1 { + authorized_scope_ref: project_id.as_str().to_owned(), + event_kinds: vec!["adoption.eligibility_observed.v1".to_owned()], + horizon: ObservabilityHorizonV1 { + since_micros: 0, + until_micros: i64::MAX, + }, + after_watermark: None, + limit: 32, + }) + .await + .expect("read persisted eligibility census"); + let expected = adoption_eligibility_census().expect("catalog census"); + assert_eq!( + page.events.len(), + expected.len(), + "one observation must persist per composed family" + ); + let persisted: BTreeMap = page + .events + .iter() + .map(|event| { + assert_eq!(event.coverage, CoverageStateV1::Known); + let ObservabilityPayloadV1::AdoptionEligibility(observation) = &event.payload + else { + panic!("unexpected payload for {}", event.event_kind); + }; + (observation.capability.clone(), observation.clone()) + }) + .collect(); + for observation in expected { + assert_eq!( + persisted.get(&observation.capability), + Some(&observation), + "persisted census must match the composed catalog" + ); + } + } +} diff --git a/src/daemon/code_index_scheduler.rs b/src/daemon/code_index_scheduler.rs index 7be2698a2..46c689d09 100644 --- a/src/daemon/code_index_scheduler.rs +++ b/src/daemon/code_index_scheduler.rs @@ -2867,6 +2867,7 @@ mod git_tree_capture; mod graph_activation; pub(crate) mod identity; mod ignored_dependencies; +pub(in crate::daemon) mod observability; mod privacy; pub(in crate::daemon) mod queries; pub(in crate::daemon) mod query_runtime; diff --git a/src/daemon/code_index_scheduler/observability.rs b/src/daemon/code_index_scheduler/observability.rs new file mode 100644 index 000000000..e7e848224 --- /dev/null +++ b/src/daemon/code_index_scheduler/observability.rs @@ -0,0 +1,176 @@ +//! Canonical Plan 26 observability lane for one mounted code-index worktree. +//! Telemetry never changes the product path: refusals are logged and dropped, +//! and an uninstalled lane records nothing. + +use std::sync::Arc; + +use tracedecay_domain::{ + CoverageStateV1, IndexObservationKindV1, IndexObservedV1, IndexOutcomeV1, QueueDepthBucketV1, + RetrievalBudget, +}; +use tracedecay_query::retrieval::AuthorizedQueryFallbackV1; +use tracedecay_query::retrieval::observation::observe_composition; +use tracedecay_usecases::observability::{ + BoundedObservabilityProducerV1, emit_retrieval_pipeline, record_index, +}; + +use super::CodeIndexReconcileOutcomeV1; + +/// Project-bound observation authority installed once per mounted worktree +/// (`CodeIndexSchedulerRegistryV1::install_index_observability`). The session +/// database carries index lifecycle receipts directly; the bounded producer +/// carries the retrieval-pipeline families off the query hot path. +#[derive(Clone)] +pub(in crate::daemon) struct CodeIndexObservabilityV1 { + session_db: crate::global_db::RegisteredGlobalDbLeaseV1, + producer: Arc, +} + +impl CodeIndexObservabilityV1 { + pub(in crate::daemon) fn new( + session_db: crate::global_db::RegisteredGlobalDbLeaseV1, + producer: Arc, + ) -> Self { + Self { + session_db, + producer, + } + } + + /// Records one terminal reconcile pass as a canonical index lifecycle + /// observation beside the worker's in-memory cadence receipt. + pub(in crate::daemon) async fn record_reconcile_outcome( + &self, + outcome: &CodeIndexReconcileOutcomeV1, + service_micros: u64, + queue_depth_bucket: QueueDepthBucketV1, + ) { + let observation = reconcile_index_observation(outcome, service_micros, queue_depth_bucket); + if let Err(error) = record_index(self.session_db.as_ref(), observation).await { + tracing::debug!( + event = "code_index_observability", + family = "index", + outcome = "unavailable", + error = ?error, + "code-index lifecycle observation could not be recorded" + ); + } + } + + /// Offers the Plan 26 retrieval-pipeline families projected from one + /// completed query composition to the bounded producer, non-blocking on + /// the query hot path. + pub(in crate::daemon) fn record_retrieval_composition( + &self, + authorized: &AuthorizedQueryFallbackV1, + budget: &RetrievalBudget, + ) { + // Tokens are countable only after hydration; the projection reports + // partial synthesis coverage rather than a fabricated zero. + let observation = observe_composition( + &authorized.fallback_lanes, + &authorized.composition, + budget, + None, + ); + let summary = emit_retrieval_pipeline( + self.producer.as_ref(), + self.producer.identity(), + observation, + ); + if summary.dropped > 0 || summary.invalid > 0 { + tracing::debug!( + event = "code_index_observability", + family = "retrieval_pipeline", + enqueued = summary.enqueued, + dropped = summary.dropped, + invalid = summary.invalid, + "retrieval-pipeline observations were partially refused by the bounded producer" + ); + } + } +} + +/// Project one terminal reconcile outcome into the closed index-lifecycle +/// vocabulary. A publication carries its changed-chunk volume; a no-op rescan +/// produced no items and abstains rather than counting as a publication. +fn reconcile_index_observation( + outcome: &CodeIndexReconcileOutcomeV1, + service_micros: u64, + queue_depth_bucket: QueueDepthBucketV1, +) -> IndexObservedV1 { + match outcome { + CodeIndexReconcileOutcomeV1::Published(evidence) => IndexObservedV1 { + kind: IndexObservationKindV1::Publication, + duration_micros: Some(service_micros), + item_count: Some(evidence.changed_chunks as u64), + queue_depth_bucket, + outcome: IndexOutcomeV1::Published, + // The worker fully observed this pass from wake to seal. + coverage: CoverageStateV1::Known, + }, + CodeIndexReconcileOutcomeV1::Noop(_) => IndexObservedV1 { + kind: IndexObservationKindV1::Rescan, + duration_micros: Some(service_micros), + item_count: None, + queue_depth_bucket, + outcome: IndexOutcomeV1::NoOp, + coverage: CoverageStateV1::Known, + }, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tracedecay_domain::{ + CodeGenerationId, ContentDigest, ManifestDigest, ObservabilityPayloadV1, RepositoryId, + }; + + fn published() -> CodeIndexReconcileOutcomeV1 { + CodeIndexReconcileOutcomeV1::Published(super::super::CodeIndexPublishEvidenceV1 { + generation_id: CodeGenerationId::new("generation.observability.fixture") + .expect("generation id"), + repository_id: RepositoryId::new("repository.observability.fixture") + .expect("repository id"), + snapshot_content_identity: ContentDigest::new(format!("sha256:{}", "a".repeat(64))) + .expect("content digest"), + _lane_digest: ManifestDigest::new(format!("sha256:{}", "b".repeat(64))) + .expect("lane digest"), + _file_occurrence_ids: Vec::new(), + reextracted_files: 3, + changed_chunks: 7, + reused_chunks: 11, + overflow_reconciled: false, + }) + } + + #[test] + fn a_publication_projects_as_a_published_lifecycle_observation() { + let observation = reconcile_index_observation(&published(), 900, QueueDepthBucketV1::Zero); + assert_eq!(observation.kind, IndexObservationKindV1::Publication); + assert_eq!(observation.outcome, IndexOutcomeV1::Published); + assert_eq!(observation.duration_micros, Some(900)); + assert_eq!(observation.item_count, Some(7)); + ObservabilityPayloadV1::Index(observation) + .validate() + .expect("publication observation validates"); + } + + #[test] + fn a_noop_rescan_abstains_instead_of_counting_as_a_publication() { + let outcome = CodeIndexReconcileOutcomeV1::Noop(super::super::CodeIndexNoopEvidenceV1 { + snapshot_content_identity: ContentDigest::new(format!("sha256:{}", "c".repeat(64))) + .expect("content digest"), + overflow_reconciled: false, + }); + let observation = + reconcile_index_observation(&outcome, 250, QueueDepthBucketV1::OneToEight); + assert_eq!(observation.kind, IndexObservationKindV1::Rescan); + assert_eq!(observation.outcome, IndexOutcomeV1::NoOp); + assert_eq!(observation.item_count, None); + ObservabilityPayloadV1::Index(observation) + .validate() + .expect("no-op observation validates"); + } +} diff --git a/src/daemon/code_index_scheduler/query_runtime.rs b/src/daemon/code_index_scheduler/query_runtime.rs index 2efe3cda3..794a5c594 100644 --- a/src/daemon/code_index_scheduler/query_runtime.rs +++ b/src/daemon/code_index_scheduler/query_runtime.rs @@ -555,6 +555,11 @@ where input.cursor.as_ref(), ) .await?; + // Canonical Plan 26 retrieval-pipeline observation from the composition + // this query actually ran; an uninstalled lane records nothing. + if let Some(observability) = schedulers.index_observability_for_scope(scope).await { + observability.record_retrieval_composition(&authorized, &request.budget); + } Ok(ExecutedQuerySearchV1 { generation, authorized, diff --git a/src/daemon/code_index_scheduler/registry.rs b/src/daemon/code_index_scheduler/registry.rs index 1af869a4f..1619d0035 100644 --- a/src/daemon/code_index_scheduler/registry.rs +++ b/src/daemon/code_index_scheduler/registry.rs @@ -11,7 +11,7 @@ use std::{ collections::{BTreeMap, BTreeSet}, path::{Component, Path, PathBuf}, sync::{ - Arc, Mutex, RwLock, Weak, + Arc, Mutex, OnceLock, RwLock, Weak, atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, }, time::{Duration, Instant}, @@ -380,6 +380,10 @@ pub(super) struct MountedCodeIndexWorktreeV1 { /// timestamp, and trigger so a cancelling query cannot erase a coalesced /// foreign wake between independent atomic updates. pending_wake: Arc, + /// Canonical Plan 26 observability lane, installed once after project open + /// mounts the project-bound producer. Empty means this worktree records no + /// canonical index or retrieval observations (never a fabricated zero). + index_observability: Arc>, shutting_down: Arc, /// Count of in-flight owner passes; nonzero means activation or reconcile /// work is running for this worktree. @@ -1632,6 +1636,8 @@ impl CodeIndexSchedulerRegistryV1 { state.trigger = Self::pack_trigger(trigger); } + /// Returns the pass's service time so the caller can attach the same + /// measurement to the canonical index-lifecycle observation. fn record_reconcile_receipt( telemetry: &Mutex, project_root: PathBuf, @@ -1639,7 +1645,7 @@ impl CodeIndexSchedulerRegistryV1 { trigger: CodeIndexCadenceTriggerV1, started_micros: i64, outcome: &CodeIndexReconcileOutcomeV1, - ) { + ) -> u64 { let ready_micros = now_micros().0; let (cadence_outcome, overflow_reconciled) = match outcome { CodeIndexReconcileOutcomeV1::Published(evidence) => ( @@ -1698,6 +1704,9 @@ impl CodeIndexSchedulerRegistryV1 { overflow_reconciled = receipt.overflow_reconciled, "code-index reconcile reached a terminal outcome" ); + // `service_micros` is clamped non-negative by construction, so the + // widening cast is exact. + let service_micros = receipt.service_micros().max(0) as u64; let mut telemetry = telemetry .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); @@ -1724,6 +1733,7 @@ impl CodeIndexSchedulerRegistryV1 { "code-index cadence percentile became eligible" ); } + service_micros } /// Latest completed event-to-ready receipt for this registry, if any. @@ -2035,6 +2045,9 @@ impl CodeIndexSchedulerRegistryV1 { let semantic_evaluation_publication_gate = Arc::new(tokio::sync::Mutex::new(())); let ignored_dependency_admissions = Arc::new(Mutex::new(BTreeMap::new())); let pending_wake = Arc::new(PendingWakeV1::default()); + let index_observability = + Arc::new(OnceLock::::new()); + let worker_index_observability = Arc::clone(&index_observability); let worker_scheduler = Arc::clone(&scheduler); let worker_reconcile_in_progress = Arc::clone(&reconcile_in_progress); let worker_serving_generation = Arc::clone(&serving_generation); @@ -2365,7 +2378,7 @@ impl CodeIndexSchedulerRegistryV1 { evidence, ); } - Self::record_reconcile_receipt( + let service_micros = Self::record_reconcile_receipt( &worker_cadence_telemetry, worker_project_root.clone(), arrival, @@ -2373,6 +2386,24 @@ impl CodeIndexSchedulerRegistryV1 { started_micros, outcome, ); + if let Some(observability) = worker_index_observability.get() { + // The pending slot coalesces at most one waiting wake, + // so the queue behind this pass is empty or singular. + let queue_depth_bucket = { + let state = worker_pending_wake + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.micros == 0 { + tracedecay_domain::QueueDepthBucketV1::Zero + } else { + tracedecay_domain::QueueDepthBucketV1::OneToEight + } + }; + observability + .record_reconcile_outcome(outcome, service_micros, queue_depth_bucket) + .await; + } } else { // Surface bounded non-terminal failure without new project-path data. match &result { @@ -2421,6 +2452,7 @@ impl CodeIndexSchedulerRegistryV1 { wake: Arc::clone(&wake), epoch, pending_wake: Arc::clone(&pending_wake), + index_observability, shutting_down, reconcile_in_progress, _active_generation_encoded_bytes: active_generation_encoded_bytes, @@ -2468,6 +2500,53 @@ impl CodeIndexSchedulerRegistryV1 { Ok(()) } + /// Install the canonical Plan 26 observability lane for one mounted + /// worktree. Installation is once per mount: a repeated install against + /// the same mounted worktree keeps the incumbent lane, and a worktree that + /// is not mounted is a typed error so the caller can log the absence. + pub(in crate::daemon) async fn install_index_observability( + &self, + project_root: &Path, + observability: super::observability::CodeIndexObservabilityV1, + ) -> Result<(), CodeIndexSchedulerErrorV1> { + let project_root = project_root.canonicalize()?; + let mounted = self.mounted.lock().await; + let worktree = mounted.get(&project_root).ok_or_else(|| { + CodeIndexSchedulerErrorV1::Identity( + "cannot install index observability before its worktree".to_owned(), + ) + })?; + // A remount creates a fresh empty slot, so an ignored second set here + // can only be a same-mount duplicate carrying the same project lane. + let _ = worktree.index_observability.set(observability); + Ok(()) + } + + /// The installed observability lane for one exact admitted scope, if the + /// worktree is mounted and the lane was installed. + pub(in crate::daemon) async fn index_observability_for_scope( + &self, + scope: &tracedecay_application::ResolvedScope, + ) -> Option { + let mounted = self.mounted.try_lock().ok()?; + let mut matched = None; + for worktree in mounted.values() { + if worktree.repository_id != scope.repository_id + || worktree.worktree_id != scope.worktree_id + { + continue; + } + let Some(observability) = worktree.index_observability.get() else { + continue; + }; + if matched.is_some() { + return None; + } + matched = Some(observability.clone()); + } + matched + } + /// Install the core and optional semantic query routes as one committed /// configuration observation. The provider CAS is repeated while the /// mounted-worktree lock is held, so a delayed observer cannot publish a diff --git a/src/daemon/code_index_scheduler/tests.rs b/src/daemon/code_index_scheduler/tests.rs index 808b11437..71d3ec2d7 100644 --- a/src/daemon/code_index_scheduler/tests.rs +++ b/src/daemon/code_index_scheduler/tests.rs @@ -7696,3 +7696,135 @@ async fn wait_for_event_to_ready( tokio::time::sleep(Duration::from_millis(20)).await; } } + +/// The installed Plan 26 observability lane must persist one canonical index +/// lifecycle observation when a reconcile publishes a generation, and the +/// retrieval-pipeline families when a query composition completes, all in the +/// one project observation store. +#[tokio::test] +async fn installed_observability_lane_records_index_and_retrieval_observations() { + let _pin = crate::config::PinnedUserDataDir::new(); + let fixture = GitFixture::new(&[("src/lib.rs", "pub fn alpha() -> u32 { 1 }\n")]); + let store = TempDir::new().expect("store root"); + let (registry, scope) = mounted_core_query_worktree(&fixture, &store).await; + + let runtime = tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime::project( + tracedecay_runtime_core::storage::default_profile_root().expect("profile root"), + fixture.path(), + scope.project_id.clone(), + ) + .await + .expect("registered runtime"); + let database = runtime.project_database_arc().expect("project database"); + let producer = Arc::new( + tracedecay_usecases::observability::BoundedObservabilityProducerV1::start( + database.clone(), + tracedecay_usecases::observability::ObservabilityProducerIdentityV1 { + authorized_scope_ref: scope.project_id.as_str().to_owned(), + process_boot_id: "boot:code-index-observability".to_owned(), + producer_revision: "code-index-observability-test.v1".to_owned(), + configuration_revision: "code-index-observability-config.v1".to_owned(), + policy_revision: "code-index-observability-policy.v1".to_owned(), + }, + 64, + ) + .expect("bounded producer"), + ); + registry + .install_index_observability( + fixture.path(), + super::observability::CodeIndexObservabilityV1::new( + database.clone(), + Arc::clone(&producer), + ), + ) + .await + .expect("install observability lane"); + + // A reconcile after installation publishes a new generation and must leave + // one canonical index lifecycle observation beside the cadence receipt. + let initial = registry + .latest_generation_id(fixture.path()) + .await + .expect("initial generation"); + fixture.edit("src/lib.rs", "pub fn alpha() -> u32 { 2 }\n"); + assert!( + registry + .notify_path(fixture.path(), fixture.path().join("src/lib.rs")) + .await + ); + let _ = wait_for_generation_change(®istry, fixture.path(), &initial).await; + + // One real query composition through the mounted authority carries the + // retrieval-pipeline families through the bounded producer. + let _executed = registry + .execute_query_search(&scope, core_search_request("alpha")) + .await + .expect("query search"); + + registry.shutdown().await; + producer.shutdown().await.expect("flush producer"); + + let port = + tracedecay_usecases::observability::RegisteredObservabilityPortV1::new(database.as_ref()); + let observability_query = + |event_kinds: Vec| tracedecay_application::ObservabilityQueryV1 { + authorized_scope_ref: scope.project_id.as_str().to_owned(), + event_kinds, + horizon: tracedecay_application::ObservabilityHorizonV1 { + since_micros: 0, + until_micros: i64::MAX, + }, + after_watermark: None, + limit: 256, + }; + let index_page = tracedecay_application::ObservabilityQueryPort::query( + &port, + observability_query(vec!["index.measurement.observed.v1".to_owned()]), + ) + .await + .expect("index lifecycle events"); + assert!( + index_page.events.iter().any(|event| matches!( + &event.payload, + tracedecay_domain::ObservabilityPayloadV1::Index(observation) + if observation.outcome == tracedecay_domain::IndexOutcomeV1::Published + && observation.kind + == tracedecay_domain::IndexObservationKindV1::Publication + )), + "a published reconcile must leave a canonical publication observation" + ); + + let retrieval_page = tracedecay_application::ObservabilityQueryPort::query( + &port, + observability_query(vec![ + "retrieval.planner.decided.v1".to_owned(), + "retrieval.synthesis.completed.v1".to_owned(), + "retrieval.source.observed.v1".to_owned(), + ]), + ) + .await + .expect("retrieval pipeline events"); + let planner = retrieval_page + .events + .iter() + .find_map(|event| match &event.payload { + tracedecay_domain::ObservabilityPayloadV1::RetrievalPlanner(planner) => { + Some(planner.clone()) + } + _ => None, + }) + .expect("one planner observation per composition"); + assert_eq!( + planner.requested_lanes, + vec!["exact_literal", "lexical", "graph"], + "the observation reflects the lanes the composition actually ran" + ); + assert!( + retrieval_page.events.iter().any(|event| matches!( + &event.payload, + tracedecay_domain::ObservabilityPayloadV1::RetrievalSynthesis(_) + )), + "the composition's synthesis observation must be persisted" + ); +} diff --git a/src/daemon/invocation_state.rs b/src/daemon/invocation_state.rs index 196490d04..52f4cde47 100644 --- a/src/daemon/invocation_state.rs +++ b/src/daemon/invocation_state.rs @@ -352,6 +352,42 @@ impl DaemonInvocationState { message: "semantic vector graph provider could not be installed in the mounted code-index authority".to_owned(), }); } + // Canonical Plan 26 observability lane. The deferred code-index mount + // runs after the project-open delivery mount that owns the producer; + // an absent producer leaves the lane uninstalled and nothing records. + match self + .service + .observability_producer_with_database(Some(&canonical_project_root)) + .await + { + Some((session_db, producer)) => { + if let Err(error) = self + .code_index_schedulers + .install_index_observability( + &canonical_project_root, + code_index_scheduler::observability::CodeIndexObservabilityV1::new( + session_db, producer, + ), + ) + .await + { + tracing::warn!( + event = "code_index_observability_mount", + outcome = "unavailable", + error = %error, + "code-index observability lane could not be installed" + ); + } + } + None => { + tracing::debug!( + event = "code_index_observability_mount", + outcome = "unavailable", + reason = "producer_unmounted", + "code-index observability lane has no mounted project producer" + ); + } + } Ok(()) } diff --git a/src/daemon/project_open_owners.rs b/src/daemon/project_open_owners.rs index f9d98434f..14e3fc86c 100644 --- a/src/daemon/project_open_owners.rs +++ b/src/daemon/project_open_owners.rs @@ -1086,6 +1086,20 @@ pub(super) async fn register_project_open_production_owners( delivery_settlements, ); + // Once-per-project-open adoption-eligibility census over the composed + // capability catalog, recorded through the project-bound session + // authority. Fire-and-forget telemetry: project open never blocks or + // fails on observation storage. + let census_db = session_db.clone(); + let census_project_root = project_root.to_path_buf(); + tokio::spawn(async move { + super::adoption_observation::record_project_open_adoption_census( + census_db.as_ref(), + &census_project_root, + ) + .await; + }); + // Semantic restore can decode a large durable generation. Keep that // capability-specific warm-up behind every independent production owner // so diagnostics, tests, feedback, and LSP reads remain available while diff --git a/src/daemon/project_open_owners/advisory_runtime.rs b/src/daemon/project_open_owners/advisory_runtime.rs index 3959f9bfe..3ed2c19be 100644 --- a/src/daemon/project_open_owners/advisory_runtime.rs +++ b/src/daemon/project_open_owners/advisory_runtime.rs @@ -914,6 +914,9 @@ async fn resolve_production_github_provider_config( } else { None }; + let stack_observability = + resolve_github_stack_observability(invocation, project_root, state, &owner, &repository) + .await; let authorization_context = github_discovery_authorization_context(&state.access, feedback_scope); let discovery_request = github_discovery_source_access_request(feedback_scope); @@ -960,6 +963,7 @@ async fn resolve_production_github_provider_config( identity, stack_coordinator: invocation.github_stack_coordinator(), stack_anchor_db: state.session_db.clone(), + stack_observability, }, ) } @@ -972,6 +976,67 @@ async fn resolve_production_github_provider_config( }) } +/// Mounts the canonical Observatory lane for GitHub stack observations. +/// Telemetry mounting failure is logged and yields `None` — the review +/// refresh owner keeps its product path either way. +async fn resolve_github_stack_observability( + invocation: &DaemonInvocationState, + project_root: &Path, + state: &ProjectOpenDependentOwnerState, + github_owner: &str, + github_repository: &str, +) -> Option { + let unavailable = |reason: &str, detail: String| { + tracing::warn!( + event = "github_stack_observability_mount", + outcome = "unavailable", + reason, + detail, + project = %project_root.display(), + "GitHub stack observability lane is not mounted" + ); + }; + let topology_policy = match crate::config::topology::resolved_work_topology_policy( + &state.scout_configuration.snapshot, + ) { + Ok(policy) => policy.clone(), + Err(error) => { + unavailable("work_topology_policy", format!("{error:?}")); + return None; + } + }; + // Mirrors the native-integration mount condition at project open: the + // standard pull-request fallback exists exactly when this project is an + // admitted Git worktree (project open fails earlier otherwise). + let native_git_fallback_mounted = crate::worktree::git_worktree_root(project_root).is_some(); + let probe_owner = match tracedecay_usecases::observability::GitHubStackProbeOwnerV1::mount( + state.scope.clone(), + topology_policy, + github_owner, + github_repository, + native_git_fallback_mounted, + ) { + Ok(probe_owner) => probe_owner, + Err(error) => { + unavailable("probe_owner", format!("{error:?}")); + return None; + } + }; + let Some(producer) = invocation + .service + .observability_producer(Some(project_root)) + .await + else { + unavailable("producer_unmounted", String::new()); + return None; + }; + Some(tracedecay_usecases::advisory::GitHubStackObservabilityV1 { + probe_owner, + producer, + observation_db: state.session_db.clone(), + }) +} + /// Assembles the CI provider config for a credential that already proved /// Actions and Checks read permissions. `None` covers only the statically /// impossible identity-constant failures, never a permission decision. diff --git a/src/daemon/service/invocation/native_integration.rs b/src/daemon/service/invocation/native_integration.rs index 9c337d856..493cebb85 100644 --- a/src/daemon/service/invocation/native_integration.rs +++ b/src/daemon/service/invocation/native_integration.rs @@ -45,7 +45,9 @@ use tracedecay_domain::{ }; use tracedecay_store::NativeIntegrationStore; use tracedecay_usecases::observability::{ - BoundedObservabilityProducerV1, record_native_integration_transition, + BoundedObservabilityProducerV1, WorkConflictObservationResultV1, + WorkConflictObservationUnavailableV1, record_native_integration_transition, + record_work_conflict_observation, }; use tracedecay_usecases::stack_coordinator::StackCoordinatorErrorV1; @@ -168,6 +170,28 @@ pub(super) async fn execute_native_integration( &execution.result, execution.owner_preview.as_ref(), ); + // Telemetry only: the preflight disposition and terminal apply receipt + // additionally prove one mechanical conflict prediction/outcome pair. + match record_work_conflict_observation( + registered.scope.project_id.as_str(), + observability_producer.as_deref(), + surface_operation.as_str(), + owner_mounted, + &execution.result, + execution.owner_preview.as_ref(), + ) { + WorkConflictObservationResultV1::Enqueued { .. } + | WorkConflictObservationResultV1::Unavailable { + reason: WorkConflictObservationUnavailableV1::NotAdjudicated, + } => {} + refused => { + tracing::debug!( + outcome = ?refused, + operation = surface_operation.as_str(), + "work-conflict observation was not recorded" + ); + } + } let Ok(payload) = serde_json::to_value(&execution.result) else { return DaemonInvocationResponse::problem( diff --git a/src/daemon/service/invocation/observability_producer.rs b/src/daemon/service/invocation/observability_producer.rs index 722721cea..6b39488ba 100644 --- a/src/daemon/service/invocation/observability_producer.rs +++ b/src/daemon/service/invocation/observability_producer.rs @@ -129,6 +129,23 @@ impl DaemonInvocationService { .await } + /// The mounted producer together with the exact project session database + /// it writes through, for owners that also record directly through the + /// registered observation authority. + pub(crate) async fn observability_producer_with_database( + &self, + project_root: Option<&Path>, + ) -> Option<( + crate::global_db::RegisteredGlobalDbLeaseV1, + Arc, + )> { + self.project_runtimes + .read::(project_root?, |registered| { + (registered.database(), registered.producer()) + }) + .await + } + pub(crate) fn observability_producer_for_project_root( &self, project_root: &Path, diff --git a/src/daemon/service/invocation/work_attempt_exec.rs b/src/daemon/service/invocation/work_attempt_exec.rs index f1f3f4f6f..517313a22 100644 --- a/src/daemon/service/invocation/work_attempt_exec.rs +++ b/src/daemon/service/invocation/work_attempt_exec.rs @@ -59,17 +59,19 @@ use tracedecay_application::{ WorkAttemptEvidenceRecordV1, WorkAttemptProviderOutcomeV1, WorkProviderAvailabilityV1, WorkProviderFallbackRecordV1, }; +use tracedecay_domain::configuration::TopologyPolicyDigestV1; use tracedecay_domain::{ - ObservationSourceIdentityV1, WorkArtifactId, WorkArtifactRefV1, WorkAttemptIdentityV1, - WorkAttemptV1, WorkExecutableReference, WorkFallbackTopology, WorkProviderBackendV1, - WorkProviderProtocol, WorkProviderRouteV1, WorktreeId, + ObservationSourceIdentityV1, UtcMicros, WorkArtifactId, WorkArtifactRefV1, + WorkAttemptIdentityV1, WorkAttemptV1, WorkExecutableReference, WorkFallbackTopology, + WorkProviderBackendV1, WorkProviderProtocol, WorkProviderRouteV1, WorktreeId, }; use tracedecay_sessions::runtime::codex_app_server::{ CodexAppServerCancellation, CodexAppServerLaunchReceipt, CodexAppServerSummaryConfig, CodexAppServerWorkExecution, run_work_with_codex_app_server, }; use tracedecay_usecases::observability::{ - BoundedObservabilityProducerV1, record_terminal_attempt_product_views, + BoundedObservabilityProducerV1, WorkNoProgressObservationV1, WorkOwnerObservationResultV1, + record_no_progress_observation, record_terminal_attempt_product_views, record_work_operation_resource, }; @@ -299,6 +301,20 @@ async fn run_attempt( }; let attempts = services.attempts(); let identity = attempt.identity().clone(); + // The registration-pinned work topology policy carries the concurrency + // policy this attempt was admitted under; its canonical digest is the + // revision a Plan 26 no-progress terminal must name. + let topology_policy_digest = match registered.work_topology_policy.compute_digest() { + Ok(digest) => Some(digest), + Err(error) => { + tracing::warn!( + task = identity.task_id().as_str(), + ?error, + "work topology policy digest is unavailable; no-progress observations are skipped" + ); + None + } + }; match select_provider(&project_root, &attempt) { Ok(selection) => match selection.provider.protocol { @@ -311,6 +327,7 @@ async fn run_attempt( &admitted_environment, cancel, observability_producer.as_deref(), + topology_policy_digest.as_ref(), timing, ) .await; @@ -324,6 +341,7 @@ async fn run_attempt( &admitted_environment, cancel, observability_producer.as_deref(), + topology_policy_digest.as_ref(), timing, ) .await; @@ -582,6 +600,7 @@ async fn execute_provider_with_environment( admitted_environment: &BTreeMap, cancel: Arc, observability_producer: Option<&BoundedObservabilityProducerV1>, + topology_policy_digest: Option<&TopologyPolicyDigestV1>, timing: AttemptAdmissionTimingV1, ) where S: tracedecay_application::WorkAttemptStoragePort, @@ -672,8 +691,17 @@ async fn execute_provider_with_environment( Err(_) => WorkAttemptProviderOutcomeV1::LaunchFailed, }, () = tokio::time::sleep(wall) => { + let stalled_for = started.elapsed(); terminate(&mut child, TerminationSignal::Kill); let _ = child.wait().await; + offer_no_progress_observation( + observability_producer, + &identity, + envelope.deadline(), + topology_policy_digest, + deadline_micros, + stalled_for, + ); WorkAttemptProviderOutcomeV1::TimedOut } () = cancel.notified() => { @@ -787,6 +815,7 @@ async fn execute_app_server( admitted_environment: &BTreeMap, cancel: Arc, observability_producer: Option<&BoundedObservabilityProducerV1>, + topology_policy_digest: Option<&TopologyPolicyDigestV1>, timing: AttemptAdmissionTimingV1, ) where S: tracedecay_application::WorkAttemptStoragePort, @@ -809,6 +838,7 @@ async fn execute_app_server( return; } }; + let attempt_started = std::time::Instant::now(); let deadline_micros = u64::try_from(envelope.deadline().0.saturating_sub(current_micros().0)).unwrap_or(0); let wall = std::time::Duration::from_micros(deadline_micros); @@ -870,7 +900,19 @@ async fn execute_app_server( let ending = tokio::select! { joined = &mut session => AppServerEnding::Session(joined), () = tokio::time::sleep(wall) => { + let stalled_for = attempt_started.elapsed(); + // Cancelling the app-server session SIGKILLs its whole process + // tree; the escalation observed here is the same kill rung as the + // stdio path. cancellation.cancel(); + offer_no_progress_observation( + observability_producer, + &identity, + envelope.deadline(), + topology_policy_digest, + deadline_micros, + stalled_for, + ); AppServerEnding::TimedOut } () = cancel.notified() => { @@ -968,6 +1010,46 @@ async fn execute_app_server( } } +/// Offers Plan 26's no-progress terminal for one wall-exhausted attempt. The +/// stall is the monotonic elapsed time from attempt start to the deadline arm +/// firing; a zero armed budget is refused by the payload contract, and +/// emission never alters the timed-out product handling. +fn offer_no_progress_observation( + observability_producer: Option<&BoundedObservabilityProducerV1>, + identity: &WorkAttemptIdentityV1, + run_deadline: UtcMicros, + topology_policy_digest: Option<&TopologyPolicyDigestV1>, + configured_timeout_micros: u64, + stalled_for: std::time::Duration, +) { + let Some(topology_policy_digest) = topology_policy_digest else { + tracing::debug!( + task = identity.task_id().as_str(), + "work attempt no-progress observation skipped: topology policy digest unavailable" + ); + return; + }; + let elapsed_stall_micros = u64::try_from(stalled_for.as_micros()).unwrap_or(u64::MAX); + let result = record_no_progress_observation( + observability_producer, + &WorkNoProgressObservationV1 { + attempt: identity, + run_deadline, + concurrency_policy_revision: topology_policy_digest.0.as_str(), + configured_timeout_micros, + elapsed_stall_micros, + observed_at: current_micros(), + }, + ); + if result != WorkOwnerObservationResultV1::Enqueued { + tracing::debug!( + task = identity.task_id().as_str(), + ?result, + "work attempt no-progress observation was not enqueued" + ); + } +} + /// Runs the graceful-interrupt / forced-kill cancellation ladder after the /// durable cancellation request has been observed. async fn cancel_ladder( diff --git a/src/daemon/service/invocation/work_attempt_exec/tests.rs b/src/daemon/service/invocation/work_attempt_exec/tests.rs index 9c2db8c2c..eebf06e9a 100644 --- a/src/daemon/service/invocation/work_attempt_exec/tests.rs +++ b/src/daemon/service/invocation/work_attempt_exec/tests.rs @@ -35,24 +35,29 @@ use crate::config::{PinnedRuntimeConfiguration, RuntimeConfigurationTarget}; use tracedecay_application::{ CancelWorkAttemptCommand, CancellationContext, CapabilityGrantSnapshot, Deadline, - DisclosureClass, RequestId, ResolvedScope, WorkAttemptAdmissionKind, WorkAttemptCapacityV1, + DisclosureClass, ObservabilityHorizonV1, ObservabilityQueryPort, ObservabilityQueryV1, + RequestId, ResolvedScope, WorkAttemptAdmissionKind, WorkAttemptCapacityV1, WorkAttemptCapacityVerdictV1, WorkAttemptInsertOutcome, WorkAttemptListPageV1, WorkAttemptService, WorkAttemptStatusRequestV1, WorkAttemptStorageError, WorkAttemptStoragePort, WorkAttemptStreamChannelV1, WorkAttemptStreamSummaryV1, }; use tracedecay_domain::{ - ActorId, AttemptId, CommitId, ConfigurationRevisionId, ConfigurationSnapshotId, ManifestDigest, - OperationActivationOutcomeV1, OperationStageV1, ProjectId, ProposalId, ProviderId, RefId, - RepositoryId, RunId, SessionId, TaskId, UtcMicros, WorkApprovalPolicy, WorkAttemptIdentityV1, - WorkAttemptProjectionBindingV1, WorkAttemptStateV1, WorkAttemptV1, WorkAuthority, - WorkCancellationStateV1, WorkEffectStateV1, WorkEgressPolicy, WorkExecutableReference, - WorkExecutionEnvelopeV1, WorkExecutionLimits, WorkExecutionSnapshot, + ActorId, AttemptId, CommitId, ConfigurationRevisionId, ConfigurationSnapshotId, + EffectReconciliationOutcomeV1, ManifestDigest, NoProgressEscalationV1, ObservabilityPayloadV1, + ObservabilityTerminalResultV1, OperationActivationOutcomeV1, OperationStageV1, ProjectId, + ProposalId, ProviderId, RefId, RepositoryId, RunId, SessionId, TaskId, UtcMicros, + WorkApprovalPolicy, WorkAttemptIdentityV1, WorkAttemptProjectionBindingV1, WorkAttemptStateV1, + WorkAttemptV1, WorkAuthority, WorkCancellationStateV1, WorkEffectStateV1, WorkEgressPolicy, + WorkExecutableReference, WorkExecutionEnvelopeV1, WorkExecutionLimits, WorkExecutionSnapshot, WorkExecutionSnapshotInput, WorkFallbackTopology, WorkFenceEpochV1, WorkFilesystemPolicy, WorkGraphVersionV1, WorkLeaseFenceV1, WorkLeaseId, WorkProductEventSequenceV1, WorkProductSourceWatermarkV1, WorkProviderRouteId, WorkProviderRouteV1, WorkRecoveryStateV1, - WorkSandboxPolicy, WorkflowOperationRef, WorktreeId, + WorkSandboxPolicy, WorkflowOperationRef, WorkflowStageClassV1, WorktreeId, canonical_sha256, }; use tracedecay_tool_catalog::{CapabilityId, UseCaseId}; +use tracedecay_usecases::observability::{ + ObservabilityProducerIdentityV1, RegisteredObservabilityPortV1, +}; /// argv the module maps onto each admitted `(backend, protocol)` pair. These /// literals live in `provider_arguments`; the spawn tests assert the child @@ -758,6 +763,7 @@ async fn a_clean_provider_run_seals_succeeded_evidence_over_the_captured_stream( &admitted_environment, Arc::new(Notify::new()), None, + None, AttemptAdmissionTimingV1::for_test(), ) .await; @@ -912,6 +918,7 @@ async fn initial_provider_child_uses_values_captured_for_that_spawn() { &admitted_environment, Arc::new(Notify::new()), None, + None, AttemptAdmissionTimingV1::for_test(), ) .await; @@ -984,6 +991,7 @@ async fn stdout_past_the_admitted_cap_is_a_typed_overflow_not_a_silent_success() &admitted_environment, Arc::new(Notify::new()), None, + None, AttemptAdmissionTimingV1::for_test(), ) .await; @@ -1152,6 +1160,7 @@ async fn a_provider_that_ignores_interrupt_is_escalated_to_a_kill_on_the_record( &admitted_environment, Arc::clone(&cancel), None, + None, AttemptAdmissionTimingV1::for_test(), ) => {} _ = driver => unreachable!("the driver loops until execution settles"), @@ -1179,6 +1188,152 @@ async fn a_provider_that_ignores_interrupt_is_escalated_to_a_kill_on_the_record( ); } +// --------------------------------------------------------------------------- +// 3b. Wall exhaustion (Plan 26 no-progress terminal) +// --------------------------------------------------------------------------- + +/// A provider that outlives its envelope deadline is killed and sealed as +/// `TimedOut`, and the kill emits exactly one `operation.no_progress.terminal.v1` +/// owner fact through the mounted producer: the pinned topology-policy digest, +/// a positive armed budget, a measured stall at least that budget, a provably +/// zero frontier, no remaining run budget, the kill escalation, and an unknown +/// effect outcome. This test spends the real two-second wall on purpose — the +/// stall must be a monotonic measurement, not a virtual-clock artifact. +#[cfg(unix)] +#[tokio::test] +async fn a_wall_exhausted_provider_seals_timed_out_and_emits_the_no_progress_terminal() { + let _pin = tracedecay_runtime_core::config::PinnedUserDataDir::new(); + let directory = tempfile::TempDir::new().unwrap(); + let root = directory.path(); + let runtime = crate::global_db::tests::harness::RegisteredGlobalDbTestRuntime::project( + tracedecay_runtime_core::storage::default_profile_root().expect("profile root"), + root, + id::(PROJECT), + ) + .await + .expect("registered runtime"); + let database = runtime.project_database_arc().expect("project database"); + let producer = tracedecay_usecases::observability::BoundedObservabilityProducerV1::start( + database.clone(), + ObservabilityProducerIdentityV1 { + authorized_scope_ref: PROJECT.to_owned(), + process_boot_id: "boot:work-attempt-exec-no-progress".to_owned(), + producer_revision: "producer.work-attempt-exec.v1".to_owned(), + configuration_revision: "configuration.work-attempt-exec.v1".to_owned(), + policy_revision: "policy.work-attempt-exec.v1".to_owned(), + }, + 8, + ) + .expect("bounded producer"); + let topology_policy_digest = tracedecay_domain::safe_work_topology_policy_v1() + .compute_digest() + .expect("topology policy digest"); + + let executable = fake_executable( + root, + "stalled-provider", + "#!/bin/sh\ncat > /dev/null\ni=0\nwhile [ $i -lt 300 ]; do sleep 1; i=$((i+1)); done\n", + ); + // The fixture is built after the database mount so the two-second wall + // budget covers only the execution itself. + let deadline = deadline_in(2); + let fixture = leased_attempt( + root, + "Stall past the wall deadline.", + &SnapshotShape { + deadline, + ..SnapshotShape::default() + }, + ); + let admitted_environment = + admitted_provider_environment(fixture.attempt.execution().execution_snapshot()); + execute_provider_with_environment( + &fixture.attempts, + &fixture.context, + &fixture.attempt, + &preferred( + executable, + WorkProviderProtocol::ClaudeStreamJson, + &CLAUDE_STREAM_JSON_ARGV, + requested_route(WorkProviderBackendV1::ClaudeCodeCli), + ), + &admitted_environment, + Arc::new(Notify::new()), + Some(&producer), + Some(&topology_policy_digest), + AttemptAdmissionTimingV1::for_test(), + ) + .await; + + // The product path is unchanged by the emission: the kill still seals the + // typed timeout terminal. + assert_eq!(fixture.state(), WorkAttemptStateV1::TimedOut); + assert_eq!( + fixture.sealed_evidence().outcome, + WorkAttemptProviderOutcomeV1::TimedOut + ); + + producer.shutdown().await.expect("producer shutdown"); + let page = RegisteredObservabilityPortV1::new(&database) + .query(ObservabilityQueryV1 { + authorized_scope_ref: PROJECT.to_owned(), + event_kinds: vec!["operation.no_progress.terminal.v1".to_owned()], + horizon: ObservabilityHorizonV1 { + since_micros: 0, + until_micros: current_micros().0.saturating_add(1_000_000), + }, + after_watermark: None, + limit: 8, + }) + .await + .expect("no-progress page"); + assert_eq!(page.events.len(), 1, "exactly one no-progress terminal"); + let envelope = &page.events[0]; + assert_eq!( + envelope.terminal_result, + Some(ObservabilityTerminalResultV1::TimedOut) + ); + let ObservabilityPayloadV1::NoProgress(observed) = &envelope.payload else { + panic!("expected a no-progress payload, got {:?}", envelope.payload); + }; + let expected_deadline_ref = format!( + "work-run-deadline:{}", + canonical_sha256(&( + "tracedecay.work.run-deadline.v1", + TASK, + RUN, + "attempt.1", + deadline, + )) + .expect("run deadline digest") + .as_str() + ); + assert_eq!(observed.run_deadline_ref, expected_deadline_ref); + assert_eq!( + observed.concurrency_policy_revision, + topology_policy_digest.0.as_str() + ); + assert_eq!(observed.workflow_stage, WorkflowStageClassV1::Execute); + assert!( + observed.configured_timeout_micros > 0 && observed.configured_timeout_micros <= 2_000_000, + "the armed budget is the truthful remaining envelope budget, got {}", + observed.configured_timeout_micros + ); + assert!( + observed.elapsed_stall_micros >= observed.configured_timeout_micros, + "the stall is measured, not asserted: {} < {}", + observed.elapsed_stall_micros, + observed.configured_timeout_micros + ); + assert_eq!(observed.last_committed_frontier, 0); + assert_eq!(observed.remaining_run_budget_micros, 0); + assert_eq!(observed.escalation, NoProgressEscalationV1::Kill); + assert_eq!( + observed.effect_outcome, + EffectReconciliationOutcomeV1::Unknown + ); +} + // --------------------------------------------------------------------------- // 4. The Codex app-server preference gate (Plan 32) // --------------------------------------------------------------------------- @@ -1337,6 +1492,7 @@ async fn a_disqualified_app_server_falls_back_to_codex_cli_and_says_so_in_the_ev &admitted_environment, Arc::new(Notify::new()), None, + None, AttemptAdmissionTimingV1::for_test(), ) .await; @@ -1621,6 +1777,7 @@ async fn a_missing_provider_executable_seals_a_typed_denial_instead_of_panicking &admitted_environment, Arc::new(Notify::new()), None, + None, AttemptAdmissionTimingV1::for_test(), ) .await; diff --git a/src/hint_outcomes.rs b/src/hint_outcomes.rs index 5e81697d4..f50be3ac7 100644 --- a/src/hint_outcomes.rs +++ b/src/hint_outcomes.rs @@ -17,6 +17,8 @@ use tracedecay_application::{ HintEmission, HintOutcomeCorrelationPort, HintOutcomeObservation, HintOutcomePortError, HintOutcomePortFuture, HintOutcomePortOperation, HintOutcomeResolution, HintToolActivity, }; +use tracedecay_domain::{AdoptionOutcomeLinkedV1, CoverageStateV1}; +use tracedecay_usecases::observability::record_adoption_outcome; use crate::analytics_bridge::HookImportSource; use crate::global_db::{ @@ -276,6 +278,7 @@ pub(crate) async fn settle_project_hint_outcomes( "hint-outcome settlement pass completed" ); } + record_settled_adoption_outcomes(sessions, stats).await; HintOutcomeSettlement::Settled { imported_events: import.imported(), import_errors, @@ -294,6 +297,50 @@ pub(crate) async fn settle_project_hint_outcomes( } } +/// Records this pass's settled hints as one linked adoption-outcome funnel +/// (`adoption.outcome.linked.v1`). Strictly downstream telemetry: the record +/// result is discarded after the settlement outcome is determined. +/// +/// Every stage carries only what settlement proved: the idempotent +/// `hint_outcome` write is the exactly-once terminal ledger, so +/// `invoked`/`terminal` count only hints settled this pass and cross-pass +/// sums never double-count. `independently_useful` = `acted` only — the +/// correlator behaviorally verified a category-matching tool fired in the +/// independently ingested session activity (never display/self-report). +/// `repeat_useful` stays 0 (settlement never verifies repeat use), unresolved +/// hints are re-scanned later rather than carried as per-pass censored mass, +/// and their presence weakens `census_coverage` to `Partial`. +async fn record_settled_adoption_outcomes(sessions: &RegisteredGlobalDb, stats: HintOutcomeStats) { + let settled = stats.written() as u64; + if settled == 0 { + return; + } + let census_coverage = if stats.unresolved == 0 { + CoverageStateV1::Known + } else { + CoverageStateV1::Partial + }; + if let Err(error) = record_adoption_outcome( + sessions, + census_coverage, + AdoptionOutcomeLinkedV1 { + invoked: settled, + terminal: settled, + independently_useful: stats.acted as u64, + repeat_useful: 0, + censored: 0, + unknown: 0, + }, + ) + .await + { + tracing::debug!( + error = ?error, + "settled hint adoption outcome was not recorded; settlement output is unaffected" + ); + } +} + fn required_event_field( value: Option, field: &'static str, @@ -377,3 +424,304 @@ fn outcome_event(outcome: &HintOutcomeObservation) -> AnalyticsEventInsert { ), } } + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use tempfile::TempDir; + use tracedecay_application::{ + ObservabilityHorizonV1, ObservabilityQueryPort, ObservabilityQueryV1, + }; + use tracedecay_domain::{ObservabilityEnvelopeV1, ObservabilityPayloadV1, ProjectId}; + use tracedecay_sessions::runtime::{SessionMessageRecord, SessionRecord}; + use tracedecay_usecases::host_admission::HostAdmissionScope; + use tracedecay_usecases::observability::RegisteredObservabilityPortV1; + + use super::*; + use crate::host_admission::HostAdmissionTestRuntimeV1; + + const HINT_TS: i64 = 1_000_000; + /// Past `hooks::hint_outcomes::HORIZON_SECS` (30 minutes) after `HINT_TS`, + /// so a non-matching window settles as ignored instead of staying open. + const AFTER_HORIZON: i64 = HINT_TS + 2_000; + + struct ProjectSettlementFixture { + _dir: TempDir, + runtime: HostAdmissionTestRuntimeV1, + project_root: std::path::PathBuf, + /// Scope the observability envelopes are attributed to: the registered + /// project session binding, not the analytics project key. + binding_project_id: ProjectId, + } + + impl ProjectSettlementFixture { + async fn open(scope: &str) -> Self { + let dir = TempDir::new().unwrap(); + let project_root = dir.path().join("project"); + std::fs::create_dir_all(&project_root).unwrap(); + let binding_project_id = ProjectId::new(scope).expect("project id"); + let runtime = HostAdmissionTestRuntimeV1::project( + dir.path().join("profile"), + &project_root, + binding_project_id.clone(), + ) + .await + .expect("open registered project runtime"); + Self { + _dir: dir, + runtime, + project_root, + binding_project_id, + } + } + + fn analytics(&self) -> &RegisteredGlobalDb { + self.runtime.profile_database_for_test() + } + + fn sessions(&self) -> &RegisteredGlobalDb { + self.runtime + .registered_database(HostAdmissionScope::Project) + .expect("registered project session database") + } + + fn analytics_project_key(&self) -> String { + RegisteredGlobalDb::canonical_project_key(&self.project_root) + } + + async fn seed_hint(&self, session_id: &str, hint_id: &str) { + self.runtime + .append_profile_analytics_event_for_test(&AnalyticsEventInsert { + provider: "hook_claude".to_owned(), + project_id: self.analytics_project_key(), + session_id: Some(session_id.to_owned()), + timestamp: HINT_TS, + event_kind: "hint_emitted".to_owned(), + hook_name: None, + tool_name: None, + tool_category: None, + skill_name: None, + hint_category: Some("search".to_owned()), + hint_id: Some(hint_id.to_owned()), + outcome: Some("observed".to_owned()), + metadata_json: None, + }) + .await + .expect("seed hint_emitted"); + } + + async fn seed_session_activity(&self, session_id: &str, tools: Option<&str>) { + let inserted = self + .runtime + .upsert_session_for_test( + HostAdmissionScope::Project, + &SessionRecord { + provider: "claude".to_owned(), + session_id: session_id.to_owned(), + project_key: self.analytics_project_key(), + project_path: self.project_root.display().to_string(), + title: None, + started_at: Some(HINT_TS), + ended_at: None, + transcript_path: None, + metadata_json: None, + parent_session_id: None, + is_subagent: false, + agent_id: None, + parent_tool_use_id: None, + }, + ) + .await + .expect("upsert project session"); + assert!(inserted, "session should upsert"); + let Some(tools) = tools else { + return; + }; + let inserted = self + .runtime + .upsert_session_message_for_test( + HostAdmissionScope::Project, + &SessionMessageRecord { + provider: "claude".to_owned(), + message_id: format!("{session_id}:1"), + session_id: session_id.to_owned(), + role: "assistant".to_owned(), + timestamp: Some(HINT_TS + 60), + ordinal: 1, + text: "activity".to_owned(), + kind: None, + model: None, + tool_names: Some(tools.to_owned()), + source_path: None, + source_offset: Some(1), + metadata_json: None, + }, + ) + .await + .expect("upsert project session message"); + assert!(inserted, "session message should upsert"); + } + + async fn settle(&self, now_secs: i64) -> HintOutcomeStats { + let settlement = settle_project_hint_outcomes( + Some(self.analytics()), + Some(self.sessions()), + Vec::new(), + &self.project_root, + now_secs, + ) + .await; + let HintOutcomeSettlement::Settled { stats, .. } = settlement else { + panic!("expected a settled pass, got {settlement:?}"); + }; + stats + } + + async fn adoption_outcome_events(&self) -> Vec { + RegisteredObservabilityPortV1::new(self.sessions()) + .query(ObservabilityQueryV1 { + authorized_scope_ref: self.binding_project_id.as_str().to_owned(), + event_kinds: vec!["adoption.outcome.linked.v1".to_owned()], + horizon: ObservabilityHorizonV1 { + since_micros: 0, + until_micros: i64::MAX, + }, + after_watermark: None, + limit: 8, + }) + .await + .expect("read persisted adoption outcomes") + .events + } + } + + fn outcome_payload(event: &ObservabilityEnvelopeV1) -> &AdoptionOutcomeLinkedV1 { + let ObservabilityPayloadV1::AdoptionOutcome(outcome) = &event.payload else { + panic!("unexpected payload for {}", event.event_kind); + }; + outcome + } + + #[tokio::test] + async fn settlement_records_settled_hints_as_a_linked_adoption_outcome() { + let fixture = ProjectSettlementFixture::open("project.hint.adoption.mixed").await; + // One hint acted on (matching tracedecay tool observed after the + // hint), one ignored (only non-matching activity, horizon elapsed), + // and one still open (no post-hint activity ingested yet). + fixture.seed_hint("s-acted", "h-acted").await; + fixture + .seed_session_activity("s-acted", Some("tracedecay_context")) + .await; + fixture.seed_hint("s-ignored", "h-ignored").await; + fixture + .seed_session_activity("s-ignored", Some("Read")) + .await; + fixture.seed_hint("s-open", "h-open").await; + fixture.seed_session_activity("s-open", None).await; + + let stats = fixture.settle(AFTER_HORIZON).await; + assert_eq!( + stats, + HintOutcomeStats { + scanned: 3, + acted: 1, + ignored: 1, + unresolved: 1, + } + ); + + let events = fixture.adoption_outcome_events().await; + assert_eq!(events.len(), 1, "one funnel record per settlement pass"); + assert_eq!( + outcome_payload(&events[0]), + &AdoptionOutcomeLinkedV1 { + invoked: 2, + terminal: 2, + independently_useful: 1, + repeat_useful: 0, + censored: 0, + unknown: 0, + }, + "only exactly-once settled hints may carry funnel mass" + ); + assert_eq!( + events[0].coverage, + CoverageStateV1::Partial, + "an unresolved remainder must weaken the census, never render Known" + ); + + // A later pass re-scans only the still-open hint, settles nothing, + // and must not re-count it as new funnel mass. + let stats = fixture.settle(AFTER_HORIZON + 240).await; + assert_eq!( + stats, + HintOutcomeStats { + scanned: 1, + acted: 0, + ignored: 0, + unresolved: 1, + } + ); + assert_eq!(fixture.adoption_outcome_events().await.len(), 1); + } + + #[tokio::test] + async fn settlement_with_only_open_hints_emits_no_adoption_outcome() { + let fixture = ProjectSettlementFixture::open("project.hint.adoption.open").await; + fixture.seed_hint("s-open", "h-open").await; + fixture.seed_session_activity("s-open", None).await; + + let stats = fixture.settle(AFTER_HORIZON).await; + assert_eq!( + stats, + HintOutcomeStats { + scanned: 1, + acted: 0, + ignored: 0, + unresolved: 1, + } + ); + assert!( + fixture.adoption_outcome_events().await.is_empty(), + "an all-open pass settled nothing and must not fabricate funnel mass" + ); + } + + #[tokio::test] + async fn fully_settled_pass_records_a_known_coverage_census() { + let fixture = ProjectSettlementFixture::open("project.hint.adoption.known").await; + fixture.seed_hint("s-acted", "h-acted").await; + fixture + .seed_session_activity("s-acted", Some("tracedecay_context")) + .await; + + let stats = fixture.settle(AFTER_HORIZON).await; + assert_eq!( + stats, + HintOutcomeStats { + scanned: 1, + acted: 1, + ignored: 0, + unresolved: 0, + } + ); + let events = fixture.adoption_outcome_events().await; + assert_eq!(events.len(), 1); + assert_eq!( + outcome_payload(&events[0]), + &AdoptionOutcomeLinkedV1 { + invoked: 1, + terminal: 1, + independently_useful: 1, + repeat_useful: 0, + censored: 0, + unknown: 0, + } + ); + assert_eq!( + events[0].coverage, + CoverageStateV1::Known, + "a pass that settled every scanned hint is a complete census" + ); + } +}