diff --git a/crates/tracedecay-application/src/feedback/concrete.rs b/crates/tracedecay-application/src/feedback/concrete.rs index ad155e74195..700a4b54089 100644 --- a/crates/tracedecay-application/src/feedback/concrete.rs +++ b/crates/tracedecay-application/src/feedback/concrete.rs @@ -903,11 +903,9 @@ impl FeedbackCycleDedupePort for ProjectFeedbackStore { if !publication_matches_context(publication, context) { return FeedbackPublicationRecordState::Unavailable; } - match self.record_publication(publication.clone()).await { - Ok(true) => FeedbackPublicationRecordState::Recorded, - Ok(false) => FeedbackPublicationRecordState::Duplicate, - Err(_) => FeedbackPublicationRecordState::Unavailable, - } + self.record_publication(publication.clone()) + .await + .unwrap_or(FeedbackPublicationRecordState::Unavailable) }) } } @@ -1208,7 +1206,7 @@ impl FeedbackReadPort for ProjectFeedbackStore { match latest.get(&finding.finding_id) { Some((prior, _)) if publication_order(&publications[*prior], publication) - != std::cmp::Ordering::Less => {} + == std::cmp::Ordering::Greater => {} _ => { latest.insert(finding.finding_id.clone(), (index, finding.clone())); } @@ -1368,7 +1366,7 @@ impl ProjectFeedbackStore { async fn record_publication( &self, publication: FeedbackPublicationV1, - ) -> Result { + ) -> Result { publication .validate() .map_err(|_| FeedbackRuntimeError::Corrupt)?; @@ -1397,23 +1395,61 @@ impl ProjectFeedbackStore { .map(decode_ledger) .transpose()? .unwrap_or_default(); - if publications.iter().any(|stored| { - stored.result.result_id == publication.result.result_id - || (stored.dedupe_key == publication.dedupe_key - && matches!( - stored.result.termination, - FeedbackCycleTerminationV1::Clean | FeedbackCycleTerminationV1::Blocked - ) - && matches!( - publication.result.termination, - FeedbackCycleTerminationV1::Clean | FeedbackCycleTerminationV1::Blocked - )) - }) { + // A durable result identity may replay but may not be rebound to + // different evidence or a different admission scope. + if let Some(stored) = publications + .iter() + .find(|stored| stored.result.result_id == publication.result.result_id) + { + // The complete committed record owns a result ID. Replaying it + // cannot silently change admission, trigger, budget, or the + // revalidation receipt while keeping that durable identity. + let exact_identity = stored == &publication; + transaction + .rollback() + .await + .map_err(|_| FeedbackRuntimeError::Store)?; + return if !exact_identity { + Err(FeedbackRuntimeError::Corrupt) + } else if stored.result.termination == FeedbackCycleTerminationV1::IncompleteCoverage { + Ok(FeedbackPublicationRecordState::Reused(stored.clone())) + } else { + Ok(FeedbackPublicationRecordState::Duplicate) + }; + } + let completed_duplicate = publications.iter().any(|stored| { + stored.dedupe_key == publication.dedupe_key + && matches!( + stored.result.termination, + FeedbackCycleTerminationV1::Clean | FeedbackCycleTerminationV1::Blocked + ) + && matches!( + publication.result.termination, + FeedbackCycleTerminationV1::Clean | FeedbackCycleTerminationV1::Blocked + ) + }); + // Historic incomplete evidence must not suppress a later recovery. + // Only the latest published verdict for this exact key can coalesce + // an identical incomplete evaluation. + if completed_duplicate { + transaction + .rollback() + .await + .map_err(|_| FeedbackRuntimeError::Store)?; + return Ok(FeedbackPublicationRecordState::Duplicate); + } + if let Some(stored) = publications + .iter() + .filter(|stored| stored.dedupe_key == publication.dedupe_key) + .max_by(|left, right| publication_order(left, right)) + .filter(|stored| stored.same_incomplete_evidence(&publication)) + { + let committed = stored.clone(); transaction .rollback() .await .map_err(|_| FeedbackRuntimeError::Store)?; - return Ok(false); + return Ok(FeedbackPublicationRecordState::Reused(committed)); } if publications.len() >= MAX_STORED_PUBLICATIONS { transaction @@ -1423,12 +1459,9 @@ impl ProjectFeedbackStore { return Err(FeedbackRuntimeError::Store); } publications.push(publication); - publications.sort_by(|left, right| { - left.authority - .revalidated_at - .cmp(&right.authority.revalidated_at) - .then_with(|| left.result.result_id.cmp(&right.result.result_id)) - }); + // Sort stably by revalidation time. Equal timestamps must retain + // their actual commit order, never sort by result identity. + publications.sort_by_key(|entry| entry.authority.revalidated_at); let encoded = serde_json::to_string(&StoredPublicationLedgerV1 { schema_version: PUBLICATION_LEDGER_SCHEMA_VERSION, publications, @@ -1449,7 +1482,7 @@ impl ProjectFeedbackStore { .commit() .await .map_err(|_| FeedbackRuntimeError::Store)?; - Ok(true) + Ok(FeedbackPublicationRecordState::Recorded) } fn finding_view( @@ -2000,7 +2033,6 @@ fn publication_order( left.authority .revalidated_at .cmp(&right.authority.revalidated_at) - .then_with(|| left.result.result_id.cmp(&right.result.result_id)) } fn latest_finding<'a>( @@ -2051,12 +2083,20 @@ fn validate_request(request: &FeedbackReadRequestV1) -> Result<(), ApplicationCo #[cfg(test)] mod tests { use super::*; + use tracedecay_contracts::feedback::FeedbackRuntimeStateV1; use tracedecay_contracts::{CancellationContext, CapabilityGrantSnapshot, Deadline, RequestId}; use tracedecay_domain::configuration::{ AuthorityRef, ConfigurationRevisionId, ScopeSourceBinding, SourceBindingId, SourceKindV1, }; + use tracedecay_domain::feedback::{ + FeedbackActorContextV1, FeedbackAuthoritativeRuntimeStateV1, FeedbackBaselineStateV1, + FeedbackBudgetV1, FeedbackContentIdentityV1, FeedbackCycleId, FeedbackCycleRequestV1, + FeedbackCycleResultV1, FeedbackCycleRuntimeSnapshotV1, FeedbackEvaluationInputV1, + FeedbackScopeV1, FeedbackTargetV1, FeedbackTriggerV1, ProviderEvaluationStateV1, + }; use tracedecay_domain::{ - LocatorDigest, ProjectId, RefId, RepositoryId, WorktreeId, canonical_sha256, + CodeGenerationId, LocatorDigest, ProjectId, RefId, RepositoryId, WorktreeId, + canonical_sha256, }; use tracedecay_runtime_core::db::{DatabaseAuthority, TestDatabaseRuntimeMode}; @@ -2133,6 +2173,405 @@ mod tests { } } + fn incomplete_publication( + cycle_id: &str, + evidence: &str, + remote_state: tracedecay_domain::feedback::ProviderEvaluationStateV1, + ) -> FeedbackPublicationV1 { + let admitted_scope = scope(); + let request = FeedbackCycleRequestV1::new( + id::(cycle_id), + FeedbackScopeV1 { + project_id: admitted_scope.project_id.clone(), + repository_id: admitted_scope.repository_id.clone(), + worktree_id: admitted_scope.worktree_id.clone(), + branch_ref: "refs/heads/main".to_owned(), + head_commit_id: id::("commit.feedback.incomplete.fixture"), + }, + FeedbackContentIdentityV1::SavedContent { + generation_digest: canonical_sha256(&"feedback.incomplete.generation").unwrap(), + file_digest: canonical_sha256(&"feedback.incomplete.file").unwrap(), + }, + FeedbackTriggerV1::PostEditHook, + canonical_sha256(&"feedback.incomplete.policy").unwrap(), + canonical_sha256(&"feedback.incomplete.configuration").unwrap(), + FeedbackBudgetV1::bounded(100, 100, 1_024, 100), + ) + .unwrap(); + let input = FeedbackEvaluationInputV1 { + target: FeedbackTargetV1 { + file: id("file.feedback.incomplete.fixture"), + span: None, + symbol: None, + generation_id: Some(id::("generation.feedback.incomplete")), + }, + request, + actor: FeedbackActorContextV1::default(), + observed_at: UtcMicros(10), + }; + let result = FeedbackCycleResultV1::new( + &input.request, + FeedbackCycleTerminationV1::IncompleteCoverage, + vec![ + ProviderEvaluationStateV1::SupportedCompletedComplete, + remote_state, + ], + vec![FeedbackBaselineStateV1::NoPriorBaseline], + None, + None, + None, + Vec::new(), + 0, + 0, + 0, + ) + .unwrap(); + let runtime = FeedbackRuntimeStateV1::new( + FeedbackAuthoritativeRuntimeStateV1 { + snapshot: FeedbackCycleRuntimeSnapshotV1::from_request(&input.request), + baseline_horizon: None, + runtime_watermark: canonical_sha256(&"feedback.incomplete.runtime").unwrap(), + }, + input.target.generation_id.clone(), + ) + .unwrap(); + let operation = feedback_surface_operation("feedback_list") + .unwrap() + .unwrap(); + let context = context(&operation, "actor.feedback-reader.requester"); + let authority = AuthorityReceipt::from_context( + &context, + PolicyDecisionRef::new( + "policy.feedback.incomplete", + 1, + canonical_sha256(&"feedback.incomplete.grant-policy").unwrap(), + ComponentVersion::new("policy.feedback.incomplete.v1").unwrap(), + ) + .unwrap(), + UtcMicros(10), + ) + .unwrap(); + FeedbackPublicationV1::new( + input.clone(), + input + .dedupe_key(&canonical_sha256(&evidence).unwrap()) + .unwrap(), + result, + runtime, + admitted_scope, + authority, + ) + .unwrap() + } + + #[tokio::test] + async fn repeated_incomplete_local_saves_do_not_fill_the_durable_ledger() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("feedback-incomplete-dedupe.db"); + crate::register_test_schema_installer(); + let authority = + DatabaseAuthority::acquire_test(&path, "feedback incomplete dedupe").unwrap(); + let (database, _) = + Database::publish_test_runtime(&path, &authority, TestDatabaseRuntimeMode::Initialize) + .await + .unwrap(); + let store = ProjectFeedbackStore { + database, + response_handle_root: root.path().join("response-handles"), + source_observations: None, + }; + + let unavailable = tracedecay_domain::feedback::ProviderEvaluationStateV1::Unavailable; + let initial = incomplete_publication( + "cycle.feedback.incomplete.initial", + "feedback.incomplete.stable-evidence", + unavailable, + ); + let state = store.record_publication(initial.clone()).await.unwrap(); + assert_eq!(state, FeedbackPublicationRecordState::Recorded); + // Every re-run has a fresh result identity but the same exact evidence. + for ordinal in 0..64 { + let repeated = incomplete_publication( + &format!("cycle.feedback.incomplete.repeated.{ordinal}"), + "feedback.incomplete.stable-evidence", + unavailable, + ); + assert_ne!(repeated.result.result_id, initial.result.result_id); + let state = store.record_publication(repeated).await.unwrap(); + assert_eq!( + state, + FeedbackPublicationRecordState::Reused(initial.clone()) + ); + } + let retained = store.load_publications().await.unwrap(); + assert_eq!( + retained.len(), + 1, + "identical incomplete saves must not grow storage" + ); + assert_eq!( + retained[0].result.termination, + FeedbackCycleTerminationV1::IncompleteCoverage + ); + let operation = feedback_surface_operation("feedback_list") + .unwrap() + .unwrap(); + let admitted = context(&operation, "actor.feedback-reader.requester"); + assert_eq!( + store.lookup_completed(&admitted, &initial.dedupe_key).await, + FeedbackCycleDedupeState::Unique, + "incomplete evidence must not grant a completed dedupe verdict" + ); + + // A changed remote verdict is new inspectable evidence even if the + // source file and request dedupe key have not changed. + let mut failed = incomplete_publication( + "cycle.feedback.incomplete.failed", + "feedback.incomplete.stable-evidence", + tracedecay_domain::feedback::ProviderEvaluationStateV1::Failed, + ); + failed.authority.revalidated_at = UtcMicros(11); + assert_eq!(failed.dedupe_key, initial.dedupe_key); + assert_ne!(failed.result.result_id, initial.result.result_id); + let state = store.record_publication(failed.clone()).await.unwrap(); + assert_eq!(state, FeedbackPublicationRecordState::Recorded); + + // Returning to an older verdict is a new transition, not a duplicate + // of the historical A. Latest readers must observe A -> B -> A. + let mut recovered = incomplete_publication( + "cycle.feedback.incomplete.recovered", + "feedback.incomplete.stable-evidence", + unavailable, + ); + recovered.authority.revalidated_at = UtcMicros(12); + assert!(recovered.validate().is_ok()); + let state = store.record_publication(recovered.clone()).await.unwrap(); + assert_eq!(state, FeedbackPublicationRecordState::Recorded); + assert_eq!( + store + .doctor_latest_publication(&admitted) + .await + .unwrap() + .unwrap() + .result + .result_id, + recovered.result.result_id + ); + let unchanged = incomplete_publication( + "cycle.feedback.incomplete.recovered-again", + "feedback.incomplete.stable-evidence", + unavailable, + ); + let state = store.record_publication(unchanged).await.unwrap(); + assert_eq!( + state, + FeedbackPublicationRecordState::Reused(recovered.clone()) + ); + + // An unchanged verdict with a new runtime watermark also remains + // inspectable: source identity is not the current runtime receipt. + let mut changed_runtime = incomplete_publication( + "cycle.feedback.incomplete.runtime", + "feedback.incomplete.stable-evidence", + unavailable, + ); + changed_runtime.runtime.authoritative.runtime_watermark = + canonical_sha256(&"feedback.incomplete.new-runtime").unwrap(); + changed_runtime.authority.revalidated_at = UtcMicros(13); + assert!(changed_runtime.validate().is_ok()); + let state = store + .record_publication(changed_runtime.clone()) + .await + .unwrap(); + assert_eq!(state, FeedbackPublicationRecordState::Recorded); + let retained = store.load_publications().await.unwrap(); + assert_eq!(retained.len(), 4); + assert!( + retained + .iter() + .any(|item| item.result.result_id == failed.result.result_id) + ); + assert!( + retained + .iter() + .any(|item| item.runtime == changed_runtime.runtime) + ); + + // Fresh evidence remains publishable and incomplete, not silently + // hidden by the earlier project's incomplete publication. + let updated = incomplete_publication( + "cycle.feedback.incomplete.updated", + "feedback.incomplete.updated-evidence", + unavailable, + ); + assert_ne!(updated.dedupe_key, initial.dedupe_key); + let state = store.record_publication(updated.clone()).await.unwrap(); + assert_eq!(state, FeedbackPublicationRecordState::Recorded); + let retained = store.load_publications().await.unwrap(); + assert_eq!(retained.len(), 5); + assert!(retained.iter().any(|publication| { + publication.dedupe_key == updated.dedupe_key + && publication.result.termination == FeedbackCycleTerminationV1::IncompleteCoverage + })); + } + + #[tokio::test] + async fn equal_timestamp_provider_verdicts_follow_durable_commit_order() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("feedback-equal-timestamp.db"); + crate::register_test_schema_installer(); + let authority = DatabaseAuthority::acquire_test(&path, "feedback equal timestamp").unwrap(); + let (database, _) = + Database::publish_test_runtime(&path, &authority, TestDatabaseRuntimeMode::Initialize) + .await + .unwrap(); + let store = ProjectFeedbackStore { + database, + response_handle_root: root.path().join("response-handles"), + source_observations: None, + }; + let unavailable = incomplete_publication( + "cycle.feedback.equal-timestamp.unavailable", + "feedback.equal-timestamp.same-content", + ProviderEvaluationStateV1::Unavailable, + ); + let failed = incomplete_publication( + "cycle.feedback.equal-timestamp.failed", + "feedback.equal-timestamp.same-content", + ProviderEvaluationStateV1::Failed, + ); + // Deliberately publish the lexicographically larger result identity + // first. Its identity must not win a timestamp collision. + let (older, newer) = if unavailable.result.result_id > failed.result.result_id { + (unavailable, failed) + } else { + (failed, unavailable) + }; + assert_eq!(older.dedupe_key, newer.dedupe_key); + assert_eq!( + older.authority.revalidated_at, + newer.authority.revalidated_at + ); + assert!(older.result.result_id > newer.result.result_id); + assert_ne!(older.result.provider_states, newer.result.provider_states); + assert_eq!( + store.record_publication(older.clone()).await.unwrap(), + FeedbackPublicationRecordState::Recorded + ); + assert_eq!( + store.record_publication(newer.clone()).await.unwrap(), + FeedbackPublicationRecordState::Recorded + ); + let operation = feedback_surface_operation("feedback_list") + .unwrap() + .unwrap(); + let admitted = context(&operation, "actor.feedback-reader.requester"); + let latest = store + .doctor_latest_publication(&admitted) + .await + .unwrap() + .unwrap(); + assert_eq!(latest.result.result_id, newer.result.result_id); + + // A new cycle with the latest verdict must reuse its committed + // publication, not resurrect the earlier provider state. + let replay = incomplete_publication( + "cycle.feedback.equal-timestamp.replay", + "feedback.equal-timestamp.same-content", + newer.result.provider_states[1], + ); + assert_ne!(replay.result.result_id, newer.result.result_id); + assert_eq!( + store.record_publication(replay).await.unwrap(), + FeedbackPublicationRecordState::Reused(newer) + ); + let ordered = store.load_publications().await.unwrap(); + assert_eq!(ordered.len(), 2); + assert_eq!( + ordered.last().unwrap().result.result_id, + newer.result.result_id, + "equal-timestamp publications retain their durable commit order" + ); + + // A later return to the first verdict is a new publication, not + // a lookup of its older, lexicographically larger identity. + let recovered = incomplete_publication( + "cycle.feedback.equal-timestamp.recovered", + "feedback.equal-timestamp.same-content", + older.result.provider_states[1], + ); + assert_eq!( + store.record_publication(recovered.clone()).await.unwrap(), + FeedbackPublicationRecordState::Recorded + ); + assert_eq!( + store + .doctor_latest_publication(&admitted) + .await + .unwrap() + .unwrap() + .result + .result_id, + recovered.result.result_id + ); + let ordered = store.load_publications().await.unwrap(); + assert_eq!(ordered.len(), 3); + assert_eq!( + ordered.last().unwrap().result.result_id, + recovered.result.result_id + ); + } + + #[tokio::test] + async fn conflicting_durable_result_identity_refuses_rebinding() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join("feedback-incomplete-identity.db"); + crate::register_test_schema_installer(); + let authority = + DatabaseAuthority::acquire_test(&path, "feedback incomplete identity").unwrap(); + let (database, _) = + Database::publish_test_runtime(&path, &authority, TestDatabaseRuntimeMode::Initialize) + .await + .unwrap(); + let store = ProjectFeedbackStore { + database, + response_handle_root: root.path().join("response-handles"), + source_observations: None, + }; + let original = incomplete_publication( + "cycle.feedback.incomplete.identity", + "feedback.incomplete.same-result", + tracedecay_domain::feedback::ProviderEvaluationStateV1::Unavailable, + ); + let state = store.record_publication(original.clone()).await.unwrap(); + assert_eq!(state, FeedbackPublicationRecordState::Recorded); + let state = store.record_publication(original.clone()).await.unwrap(); + assert_eq!( + state, + FeedbackPublicationRecordState::Reused(original.clone()) + ); + + let mut conflicting = original.clone(); + conflicting.runtime.authoritative.runtime_watermark = + canonical_sha256(&"feedback.incomplete.conflicting-runtime").unwrap(); + assert!(conflicting.validate().is_ok()); + assert!(matches!( + store.record_publication(conflicting).await, + Err(FeedbackRuntimeError::Corrupt) + )); + // A replayed result cannot silently renew or rewrite its original + // authorization receipt. This was previously accepted as a duplicate + // because the identity check compared only selected record fields. + let mut rebinding_receipt = original.clone(); + rebinding_receipt.authority.revalidated_at = UtcMicros(11); + assert!(rebinding_receipt.validate().is_ok()); + assert!(matches!( + store.record_publication(rebinding_receipt).await, + Err(FeedbackRuntimeError::Corrupt) + )); + assert_eq!(store.load_publications().await.unwrap(), vec![original]); + } + #[test] fn concurrent_request_authorities_keep_handles_in_their_own_store() { let stores = [tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap()]; diff --git a/crates/tracedecay-application/src/lsp_runtime/advisory_source_tests.rs b/crates/tracedecay-application/src/lsp_runtime/advisory_source_tests.rs index 6b250ef4207..1000b0798c8 100644 --- a/crates/tracedecay-application/src/lsp_runtime/advisory_source_tests.rs +++ b/crates/tracedecay-application/src/lsp_runtime/advisory_source_tests.rs @@ -954,7 +954,8 @@ async fn incomplete_publication_remains_readable_without_consuming_completed_ded .expect("repeated incomplete publication"); assert_eq!( store.record_publication(&context, &repeated).await, - FeedbackPublicationRecordState::Recorded + FeedbackPublicationRecordState::Reused(publication.clone()), + "a repeated incomplete cycle must return its actual committed identity" ); assert_eq!( store @@ -963,6 +964,72 @@ async fn incomplete_publication_remains_readable_without_consuming_completed_ded FeedbackCycleDedupeState::Unique, "repeated incomplete publications must not consume completed dedupe" ); + + // Run the real service again, not just the store port. A new local + // evaluation must reuse the committed finding identity and expose its + // read handles without converting incomplete remote coverage into Clean. + let mut repeated_request = cycle_request(digest('a'), observed_at); + repeated_request.input.request.cycle_id = repeated_input.request.cycle_id.clone(); + let repeated_service = feedback_service(runtime.clone(), &repeated_request); + let mut repeated_providers = complete_advisory_providers(); + repeated_providers[1].state = ProviderEvaluationStateV1::Unavailable; + let repeated_execution = repeated_service + .execute_with_advisory( + &context, + repeated_request, + FeedbackCycleAdvisoryV1 { + providers: repeated_providers, + findings: vec![github_finding()], + }, + ) + .await + .expect("replayed incomplete cycle"); + assert_eq!( + repeated_execution.cycle.termination, + tracedecay_domain::feedback::FeedbackCycleTerminationV1::IncompleteCoverage + ); + assert_eq!(repeated_execution.publication, Some(publication.clone())); + assert_eq!( + repeated_execution.cycle.result_id, + publication.result.result_id + ); + assert_ne!(repeated_input.request.cycle_id, publication.result.cycle_id); + let replayed_canonical = compose_canonical_result( + &runtime, + repeated_execution, + tracedecay_domain::feedback::FeedbackDurabilityV1::Durable, + ) + .expect("reused incomplete result handles"); + assert_eq!(replayed_canonical.finding_handles.len(), 1); + let replayed_handles = replayed_canonical + .read_handles + .as_ref() + .expect("reused diagnostics/list handles"); + let replayed_read = runtime + .owner() + .invoke( + FeedbackReadOperationV1::Diagnostics, + &replayed_handles.diagnostics_handle, + read_at, + ) + .await + .expect("reused diagnostics handle"); + let FeedbackReadInvocationResultV1::Diagnostics(replayed_read) = replayed_read else { + panic!("reused diagnostics result variant"); + }; + let ApplicationOutcome::Evidence(replayed_diagnostics) = + replayed_read.expect("reused diagnostics read").outcome + else { + panic!("reused diagnostics evidence"); + }; + assert_eq!( + replayed_diagnostics + .payload + .expect("reused diagnostics payload") + .cycle, + publication.result.clone() + ); + let complete_result = tracedecay_domain::feedback::FeedbackCycleResultV1::new_with_advisory_provider_states( &repeated_input.request, diff --git a/crates/tracedecay-contracts/src/feedback/ports.rs b/crates/tracedecay-contracts/src/feedback/ports.rs index 975c26f5d17..a8ad6489f60 100644 --- a/crates/tracedecay-contracts/src/feedback/ports.rs +++ b/crates/tracedecay-contracts/src/feedback/ports.rs @@ -261,6 +261,44 @@ impl FeedbackPublicationV1 { Ok(publication) } + /// Two incomplete evaluations can share the same committed evidence even + /// when their cycle and result identities differ. No provider, authorization, + /// finding, runtime or policy evidence may be dropped to make that true. + pub fn same_incomplete_evidence(&self, current: &Self) -> bool { + let left = &self.result; + let right = ¤t.result; + self.dedupe_key == current.dedupe_key + && self.runtime == current.runtime + && self.authorized_scope == current.authorized_scope + && self.authority.grant_id == current.authority.grant_id + && self.authority.grant_revision == current.authority.grant_revision + && self.authority.grant_digest == current.authority.grant_digest + && self.authority.disclosure == current.authority.disclosure + && self.authority.policy == current.authority.policy + && self.input.request.trigger == current.input.request.trigger + && self.input.request.budget == current.input.request.budget + && self.input.target == current.input.target + && self.input.actor == current.input.actor + && left.termination == FeedbackCycleTerminationV1::IncompleteCoverage + && right.termination == FeedbackCycleTerminationV1::IncompleteCoverage + && left.scope == right.scope + && left.content_identity == right.content_identity + && left.durability == right.durability + && left.policy_digest == right.policy_digest + && left.configuration_digest == right.configuration_digest + && left.provider_states == right.provider_states + && left.advisory_provider_states == right.advisory_provider_states + && left.baseline_states == right.baseline_states + && left.impact == right.impact + && left.impact_state == right.impact_state + && left.affected_tests_state == right.affected_tests_state + && left.findings == right.findings + && left.total_findings == right.total_findings + && left.returned_findings == right.returned_findings + && left.omitted_findings == right.omitted_findings + && left.advisory_only == right.advisory_only + } + pub fn validate(&self) -> Result<(), ApplicationContractError> { self.input.validate()?; self.input.saved()?; @@ -302,12 +340,14 @@ impl FeedbackPublicationV1 { } /// Result of the daemon-serialized publication compare-and-insert. -/// `Duplicate` means the same result, or another complete result with the same -/// dedupe key, won the race. Incomplete publications never consume that key. -/// `Cancelled`, `TimedOut`, and `Unavailable` leave no row behind. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +/// `Duplicate` is reserved for a completed result or its exact replay. +/// `Reused` carries the actual committed incomplete result, preserving its +/// read handles without inserting a false identity or claiming remote success. +/// Cancellation, timeout and unavailable states do not write a publication. +#[derive(Clone, Debug, PartialEq, Eq)] pub enum FeedbackPublicationRecordState { Recorded, + Reused(FeedbackPublicationV1), Duplicate, Cancelled, TimedOut, diff --git a/crates/tracedecay-contracts/src/feedback/service.rs b/crates/tracedecay-contracts/src/feedback/service.rs index 288397f86ec..6701a3c0969 100644 --- a/crates/tracedecay-contracts/src/feedback/service.rs +++ b/crates/tracedecay-contracts/src/feedback/service.rs @@ -305,9 +305,8 @@ pub struct FeedbackCycleExecutionResult { pub dedupe_key: Option, pub authority: Option, pub usage: FeedbackBudgetUsage, - /// Present only after the shared durable store atomically records this - /// exact inspectable publication. Duplicate, stale, cancelled, timed-out, - /// unavailable, and non-durable outcomes never expose a delivery handoff. + /// Present for the committed result, including an explicitly reused + /// incomplete publication. Never expose handles for an uncommitted identity. pub publication: Option, } @@ -1291,6 +1290,30 @@ where publication: Some(publication), ..result }), + FeedbackPublicationRecordState::Reused(existing) => { + if !existing.same_incomplete_evidence(&publication) { + return Err(ApplicationContractError::Inconsistent { + field: "feedback incomplete publication reuse", + }); + } + existing.validate()?; + Ok(FeedbackCycleExecutionResult { + cycle: existing.result.clone(), + dedupe_key: Some(existing.dedupe_key.clone()), + authority: Some(existing.authority.clone()), + publication: Some(existing), + ..result + }) + } + // A bare Duplicate cannot represent incomplete evidence: that + // loses the committed reference and therefore all read handles. + FeedbackPublicationRecordState::Duplicate + if result.cycle.termination == FeedbackCycleTerminationV1::IncompleteCoverage => + { + Err(ApplicationContractError::Inconsistent { + field: "feedback incomplete publication duplicate", + }) + } FeedbackPublicationRecordState::Duplicate => self.assemble( request, Some(dedupe_key), diff --git a/crates/tracedecay-contracts/tests/contracts_suite/feedback_cycle.rs b/crates/tracedecay-contracts/tests/contracts_suite/feedback_cycle.rs index 33e747d5949..bfcbef8eb7d 100644 --- a/crates/tracedecay-contracts/tests/contracts_suite/feedback_cycle.rs +++ b/crates/tracedecay-contracts/tests/contracts_suite/feedback_cycle.rs @@ -223,6 +223,43 @@ impl FeedbackCycleDedupePort for DedupeFixture { } } +#[derive(Clone, Default)] +struct ReusingIncompleteAtRecordFixture { + committed: Rc>>, +} + +impl FeedbackCycleDedupePort for ReusingIncompleteAtRecordFixture { + fn lookup_completed<'a>( + &'a self, + _context: &'a RequestContext, + _key: &'a tracedecay_domain::feedback::FeedbackDedupeKeyV1, + ) -> tracedecay_contracts::feedback::FeedbackPortFuture<'a, FeedbackCycleDedupeState> { + Box::pin(async { FeedbackCycleDedupeState::Unique }) + } + + fn record_publication<'a>( + &'a self, + _context: &'a RequestContext, + publication: &'a FeedbackPublicationV1, + ) -> tracedecay_contracts::feedback::FeedbackPortFuture<'a, FeedbackPublicationRecordState> + { + let state = { + let mut committed = self.committed.borrow_mut(); + match committed.as_ref() { + Some(previous) if previous.same_incomplete_evidence(publication) => { + FeedbackPublicationRecordState::Reused(previous.clone()) + } + Some(_) => FeedbackPublicationRecordState::Unavailable, + None => { + *committed = Some(publication.clone()); + FeedbackPublicationRecordState::Recorded + } + } + }; + Box::pin(async move { state }) + } +} + #[derive(Clone)] struct RecordingDedupeFixture { state: FeedbackCycleDedupeState, @@ -1405,6 +1442,76 @@ fn dedupe_key_changes_when_authoritative_evidence_changes() { assert_ne!(keys[0], keys[1]); } +#[test] +fn repeated_incomplete_feedback_reuses_inspectable_committed_identity() { + let input = saved_input(); + let provider = provider_identity(&input); + let dedupe = ReusingIncompleteAtRecordFixture::default(); + let service = FeedbackCycleService::new( + runtime_port(&input), + DiagnosticsFixture { + calls: Rc::new(Cell::new(0)), + results: vec![complete_result(provider.clone(), Vec::new())], + }, + ImpactFixture { + calls: Rc::new(Cell::new(0)), + outcome: FeedbackImpactPortOutcome::Unavailable, + }, + dedupe.clone(), + ObservationFixture::default(), + RouteAuthorizationFixture::default(), + common::operation(), + ); + let first = block_on(service.execute( + &common::context(&common::operation()), + execution_request(input.clone(), provider), + )) + .unwrap(); + assert_eq!( + first.cycle.termination, + FeedbackCycleTerminationV1::IncompleteCoverage + ); + assert!(first.publication.is_some()); + + let mut repeated = input; + repeated.request.cycle_id = FeedbackCycleId::new("cycle.feedback.incomplete.repeated").unwrap(); + let repeated_provider = provider_identity(&repeated); + let repeated_service = FeedbackCycleService::new( + runtime_port(&repeated), + DiagnosticsFixture { + calls: Rc::new(Cell::new(0)), + results: vec![complete_result(repeated_provider.clone(), Vec::new())], + }, + ImpactFixture { + calls: Rc::new(Cell::new(0)), + outcome: FeedbackImpactPortOutcome::Unavailable, + }, + dedupe, + ObservationFixture::default(), + RouteAuthorizationFixture::default(), + common::operation(), + ); + let replayed = block_on(repeated_service.execute( + &common::context(&common::operation()), + execution_request(repeated.clone(), repeated_provider), + )) + .unwrap(); + + assert_ne!(repeated.request.cycle_id, first.cycle.cycle_id); + assert_eq!(replayed.cycle, first.cycle); + assert_eq!(replayed.publication, first.publication); + assert_eq!( + replayed.cycle.termination, + FeedbackCycleTerminationV1::IncompleteCoverage, + "a durable incomplete result must not claim a clean pass" + ); + assert_eq!( + replayed.cycle.impact_state, + Some(FeedbackImpactStateV1::Unavailable) + ); + assert!(!replayed.cycle.provider_states.is_empty()); +} + #[test] fn unavailable_authoritative_baseline_cannot_produce_clean() { let input = saved_input(); diff --git a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs index bd756c6db29..5d4f40b456e 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime.rs @@ -1,6 +1,7 @@ //! Project-open composition for the canonical feedback and advisory owners. use std::fmt::Write as _; +use std::future::Future; use std::path::Path; use std::process::Command; use std::sync::Arc; @@ -151,6 +152,41 @@ struct ProjectOpenAdvisoryCycleExecutionV1 { feedback_cycle: Arc, } +/// Local editor/hook feedback must not wait for remote refresh. The explicit +/// advisory operation keeps its existing GitHub and CI refresh semantics. +#[derive(Clone, Copy)] +enum AdvisoryRemoteReadMode { + LocalFeedback, + ExplicitRefresh, +} + +async fn advisory_remote_requests( + mode: AdvisoryRemoteReadMode, + github: Option, + ci_configured: bool, + discover_ci: DiscoverCi, +) -> ( + Option, + ProductionCiFailureDiscoveryOutcomeV1, +) +where + DiscoverCi: Future, +{ + match mode { + AdvisoryRemoteReadMode::LocalFeedback => ( + None, + if ci_configured { + // A configured remote source was deliberately not queried in + // this local cycle; it has not established a clean result. + ProductionCiFailureDiscoveryOutcomeV1::Unavailable + } else { + ProductionCiFailureDiscoveryOutcomeV1::NotConfigured + }, + ), + AdvisoryRemoteReadMode::ExplicitRefresh => (github, discover_ci.await), + } +} + #[derive(Clone)] struct ProjectOpenFeedbackCyclePinV1 { revision_id: ConfigurationRevisionId, @@ -220,6 +256,7 @@ impl ProjectOpenAdvisoryFeedbackCycleV1 { deadline: MonotonicDeadline, agent_stop_gate: bool, expected_generation: Option<&tracedecay_domain::CodeGenerationId>, + remote_mode: AdvisoryRemoteReadMode, ) -> std::result::Result { let pin = self.producer.feedback_cycle.read().await.clone(); let lsp_input = pin.runtime.lsp_input(); @@ -242,7 +279,7 @@ impl ProjectOpenAdvisoryFeedbackCycleV1 { Some(&latest.generation().manifest().generation_id) == invocation.request.input.target.generation_id.as_ref() }); - self.run_cycle_with_invocation(pin, invocation, deadline, agent_stop_gate) + self.run_cycle_with_invocation(pin, invocation, deadline, agent_stop_gate, remote_mode) .await } @@ -252,6 +289,7 @@ impl ProjectOpenAdvisoryFeedbackCycleV1 { mut invocation: FeedbackCycleInvocation, deadline: MonotonicDeadline, agent_stop_gate: bool, + remote_mode: AdvisoryRemoteReadMode, ) -> std::result::Result { if agent_stop_gate { invocation.request.input.request.trigger = FeedbackTriggerV1::AgentStopGate; @@ -265,17 +303,36 @@ impl ProjectOpenAdvisoryFeedbackCycleV1 { .request .configuration_digest .clone(); - let ci = match self.ci_discovery_config.as_ref() { - Some(config) => { - discover_production_ci_failure_request_v1( - &invocation.context, - config, - &self.feedback_scope, - ) - .await - } - None => ProductionCiFailureDiscoveryOutcomeV1::NotConfigured, - }; + // Construct the discovery future without polling it on editor or hook + // paths. A pending/denied remote provider cannot consume local feedback's + // deadline or cancel its independently authorized diagnostics. + let github_request = self + .github_review_read + .clone() + .map(|(pull_request_id, operation)| GitHubReviewReadRequestV1 { + operation, + scope: self.feedback_scope.clone(), + pull_request_id, + }); + let (github, ci) = advisory_remote_requests( + remote_mode, + github_request, + self.ci_discovery_config.is_some(), + async { + match self.ci_discovery_config.as_ref() { + Some(config) => { + discover_production_ci_failure_request_v1( + &invocation.context, + config, + &self.feedback_scope, + ) + .await + } + None => ProductionCiFailureDiscoveryOutcomeV1::NotConfigured, + } + }, + ) + .await; let expires_at = UtcMicros(observed_at.0.saturating_add(5 * 60 * 1_000_000)); let operation = daemon_operation_event_authority() .begin( @@ -294,46 +351,40 @@ impl ProjectOpenAdvisoryFeedbackCycleV1 { ); LspRuntimeFailure::new("feedback-cycle-advisory-operation") })?; - let outcome = - pin.registration - .runtime() - .run_once( - &invocation.context, - AdvisoryCycleControl { - operation, - deadline, - }, - AdvisoryCycleRequest { - feedback: invocation.request, - github: self.github_review_read.clone().map( - |(pull_request_id, operation)| GitHubReviewReadRequestV1 { - operation, - scope: self.feedback_scope.clone(), - pull_request_id, - }, - ), - ci, - proximity: Some(ProximityEvaluationRequestV1 { - scope: self.feedback_scope.clone(), - observed_at, - }), - validity: tracedecay_contracts::AdvisoryFindingValidityWindowV1 { - valid_at: observed_at, - expires_at, - }, + let outcome = pin + .registration + .runtime() + .run_once( + &invocation.context, + AdvisoryCycleControl { + operation, + deadline, + }, + AdvisoryCycleRequest { + feedback: invocation.request, + github, + ci, + proximity: Some(ProximityEvaluationRequestV1 { + scope: self.feedback_scope.clone(), + observed_at, + }), + validity: tracedecay_contracts::AdvisoryFindingValidityWindowV1 { + valid_at: observed_at, + expires_at, }, - ) - .await - .map_err(|error| { - tracing::warn!( - target: "tracedecay::feedback_advisory_cycle", - project_id = self.feedback_scope.project_id.as_str(), - worktree_id = self.feedback_scope.worktree_id.as_str(), - ?error, - "advisory feedback cycle execution failed" - ); - LspRuntimeFailure::new(error.lsp_failure_class()) - })?; + }, + ) + .await + .map_err(|error| { + tracing::warn!( + target: "tracedecay::feedback_advisory_cycle", + project_id = self.feedback_scope.project_id.as_str(), + worktree_id = self.feedback_scope.worktree_id.as_str(), + ?error, + "advisory feedback cycle execution failed" + ); + LspRuntimeFailure::new(error.lsp_failure_class()) + })?; if outcome.publication().is_some() { self.deliver_completed_publication(&pin.registration, &outcome); } @@ -458,6 +509,7 @@ impl FeedbackCycleRuntimePort for ProjectOpenAdvisoryFeedbackCycleV1 { MonotonicDeadline::at(Instant::now() + Duration::from_secs(5)), false, None, + AdvisoryRemoteReadMode::LocalFeedback, ) .await?; Ok(()) @@ -549,6 +601,7 @@ impl DaemonAdvisoryCycleInvocationPort for ProjectOpenAdvisoryFeedbackCycleV1 { monotonic_deadline, false, None, + AdvisoryRemoteReadMode::ExplicitRefresh, ) .await .map_err(|failure| { @@ -1137,6 +1190,7 @@ async fn run_production_hook_cycle( MonotonicDeadline::at(Instant::now() + Duration::from_secs(5)), request.trigger == HookOrchestrationTriggerV1::Stop, Some(&indexed_generation.metadata().manifest().generation_id), + AdvisoryRemoteReadMode::LocalFeedback, ) .await { diff --git a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime/tests.rs b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime/tests.rs index aed4188bc6d..fdf115c27a3 100644 --- a/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime/tests.rs +++ b/crates/tracedecay/src/daemon/project_open_owners/advisory_runtime/tests.rs @@ -13,6 +13,94 @@ fn feedback_scope() -> FeedbackScopeV1 { } } +/// A blocked CI discovery must not even be polled for an editor/hook run. +/// The matching GitHub request is also excluded from local admission. +#[tokio::test] +async fn local_advisory_feedback_does_not_poll_remote_refresh() { + let github_request = GitHubReviewReadRequestV1 { + operation: GitHubReviewReadOperationV1::RestGetPullRequest, + scope: feedback_scope(), + pull_request_id: GitHubPullRequestIdV1::new("42").expect("pull request"), + }; + let forbidden_ci = std::future::poll_fn( + |_| -> std::task::Poll { + panic!("local feedback polled remote CI discovery") + }, + ); + let (github, ci) = advisory_remote_requests( + AdvisoryRemoteReadMode::LocalFeedback, + Some(github_request), + true, + forbidden_ci, + ) + .await; + assert!( + github.is_none(), + "local feedback must not request GitHub refresh" + ); + assert!(matches!( + ci, + ProductionCiFailureDiscoveryOutcomeV1::Unavailable + )); +} + +/// A local cycle with no CI binding must not pretend a remote provider exists. +#[tokio::test] +async fn local_feedback_without_ci_config_is_not_configured() { + let forbidden_ci = std::future::poll_fn( + |_| -> std::task::Poll { + panic!("unconfigured local feedback polled CI discovery") + }, + ); + let (github, ci) = advisory_remote_requests( + AdvisoryRemoteReadMode::LocalFeedback, + None, + false, + forbidden_ci, + ) + .await; + assert!(github.is_none()); + assert!(matches!( + ci, + ProductionCiFailureDiscoveryOutcomeV1::NotConfigured + )); +} + +/// Explicit advisory demand still owns the remote read and propagates its +/// terminal result. The barrier proves the operation awaits actual discovery. +#[tokio::test] +async fn explicit_advisory_refresh_waits_for_its_own_remote_discovery() { + let (started_tx, started_rx) = tokio::sync::oneshot::channel::<()>(); + let (release_tx, release_rx) = + tokio::sync::oneshot::channel::(); + let github_request = GitHubReviewReadRequestV1 { + operation: GitHubReviewReadOperationV1::RestGetPullRequest, + scope: feedback_scope(), + pull_request_id: GitHubPullRequestIdV1::new("42").expect("pull request"), + }; + let mut refresh = Box::pin(advisory_remote_requests( + AdvisoryRemoteReadMode::ExplicitRefresh, + Some(github_request.clone()), + true, + async move { + started_tx + .send(()) + .expect("reader still owns the admission barrier"); + release_rx.await.expect("explicit discovery must settle") + }, + )); + tokio::select! { + result = &mut refresh => panic!("explicit refresh completed before its remote read: {result:?}"), + started = started_rx => started.expect("remote read started"), + } + release_tx + .send(ProductionCiFailureDiscoveryOutcomeV1::Denied) + .expect("remote read still awaited"); + let (github, ci) = refresh.await; + assert_eq!(github, Some(github_request)); + assert!(matches!(ci, ProductionCiFailureDiscoveryOutcomeV1::Denied)); +} + #[test] fn hook_notice_registration_is_released_with_the_published_owner() { let scope = feedback_scope();