diff --git a/crates/tracedecay-application/src/feedback/concrete.rs b/crates/tracedecay-application/src/feedback/concrete.rs index ad155e7419..1bbf8cd031 100644 --- a/crates/tracedecay-application/src/feedback/concrete.rs +++ b/crates/tracedecay-application/src/feedback/concrete.rs @@ -268,29 +268,34 @@ impl ProjectFeedbackObservationSinkV1 { }) } - #[tracing::instrument(name = "usecases.feedback.close_drain", level = "trace", skip_all)] - async fn close_and_drain(&self) -> Result<(), FeedbackRuntimeError> { - let terminal = { - let _admission = self - .admission - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if self.closed.load(Ordering::Acquire) { - None - } else { - let terminal = self.prepare_terminal_envelope()?; - self.closed.store(true, Ordering::Release); - Some(terminal) + fn close_and_drain( + self: &Arc, + ) -> impl Future> + Send + 'static + use<> { + let terminal = self.close_admission(); + let sink = Arc::clone(self); + async move { + terminal?; + sink.join_worker().await + } + } + + fn close_admission(&self) -> Result<(), FeedbackRuntimeError> { + let _admission = self + .admission + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !self.closed.load(Ordering::Acquire) { + let terminal = self.prepare_terminal_envelope()?; + self.closed.store(true, Ordering::Release); + if self.control_sender.try_send(terminal).is_err() { + self.record_shutdown_failure(&FeedbackRuntimeError::ObservationWorker); } - }; - let terminal_outcome = if let Some(terminal) = terminal { - self.control_sender - .try_send(terminal) - .map_err(|_| FeedbackRuntimeError::ObservationWorker) - } else { - Ok(()) - }; + } + Ok(()) + } + #[tracing::instrument(name = "usecases.feedback.close_drain", level = "trace", skip_all)] + async fn join_worker(&self) -> Result<(), FeedbackRuntimeError> { let mut worker = self.worker.lock().await; let outcome = match worker.as_ref() { Some(task) => match task.wait().await { @@ -303,7 +308,6 @@ impl ProjectFeedbackObservationSinkV1 { None => Ok(()), }; worker.take(); - let outcome = terminal_outcome.and(outcome); if let Err(error) = &outcome { self.record_shutdown_failure(error); } @@ -525,10 +529,12 @@ impl FeedbackRuntime { }) } - /// Stop observation admission, persist the producer's terminal record, and - /// join the worker that owns the project database before store retirement. - pub async fn close_and_drain_observations(&self) -> Result<(), FeedbackRuntimeError> { - self.observation_sink.close_and_drain().await + /// Close admission immediately and return a join for the terminal write + /// and worker that owns the project database. + pub fn close_and_drain_observations( + &self, + ) -> impl Future> + Send + 'static + use<> { + self.observation_sink.close_and_drain() } pub fn owner(&self) -> Arc { @@ -2471,11 +2477,12 @@ mod tests { let boot_id = sink.boot_id.clone(); let worker_guard = sink.worker.lock().await; - let closing_sink = Arc::clone(&sink); - let close = tokio::spawn(async move { closing_sink.close_and_drain().await }); - while !sink.closed.load(Ordering::Acquire) { - tokio::task::yield_now().await; - } + let drain = sink.close_and_drain(); + assert!( + sink.closed.load(Ordering::Acquire), + "admission closes before joining the worker" + ); + let close = tokio::spawn(drain); close.abort(); assert!(close.await.unwrap_err().is_cancelled()); drop(worker_guard); @@ -2515,7 +2522,7 @@ mod tests { Err(()) }) .unwrap(); - let sink = ProjectFeedbackObservationSinkV1 { + let sink = Arc::new(ProjectFeedbackObservationSinkV1 { sender, control_sender, boot_id: canonical_sha256(&"feedback-terminal-failure").unwrap(), @@ -2525,7 +2532,7 @@ mod tests { closed: AtomicBool::new(false), shutdown_failure: AtomicU8::new(OBSERVATION_SHUTDOWN_CLEAN), worker: AsyncMutex::new(Some(worker)), - }; + }); assert!(matches!( sink.close_and_drain().await, diff --git a/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs b/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs index e44d951220..bc79b87b69 100644 --- a/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs +++ b/crates/tracedecay-daemon-service/src/project_runtime/shutdown.rs @@ -185,8 +185,9 @@ impl ProjectRuntimeRegistryV1 { /// Shut every project runtime down and leave the registry empty. /// /// Routers become unavailable before feedback owners drop and Work - /// providers are joined. The drain is single-flight, so the first - /// caller's `deadline` bounds the producer drains for every caller. + /// providers are joined. The first caller supplies the observability flush + /// deadline for the shared task. The daemon bounds its own wait separately, + /// and the registry retains the task until its database workers finish. #[tracing::instrument( name = "daemon.service.project_runtime.shutdown", level = "trace", @@ -368,12 +369,22 @@ async fn shut_down_advisory(runtimes: &BTreeMap) -> Vec } async fn shut_down_feedback(runtimes: &BTreeMap) -> Vec { + // Start every project's terminal write before joining its independent + // database worker. The retained drain keeps ownership of each store + // through its terminal write, even if the daemon stops waiting. + let drains: Vec<_> = runtimes + .iter() + .filter_map(|(project_root, runtime)| { + let feedback = runtime.feedback.as_ref()?; + Some(( + project_root, + feedback.runtime().close_and_drain_observations(), + )) + }) + .collect(); let mut lost_work = Vec::new(); - for (project_root, runtime) in runtimes { - let Some(feedback) = runtime.feedback.as_ref() else { - continue; - }; - if let Err(error) = feedback.runtime().close_and_drain_observations().await { + for (project_root, drain) in drains { + if let Err(error) = drain.await { lost_work.push(format!( "{}: feedback observations: {error}", project_root.display() diff --git a/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs b/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs index 1739c89b54..d83f14556e 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/code_graph_manifest.rs @@ -2674,9 +2674,159 @@ mod tests { &std::fs::read(fixture.scope_root.join("active-code-generation-v1.json")).unwrap(), ) .unwrap(); + let successor_digest = + SealedGraphStateDigest::try_from(pointer.state_digest.clone()).unwrap(); let successor_path = generations_root.join(pointer.generation_file); let mut successor: serde_json::Value = serde_json::from_slice(&std::fs::read(&successor_path).unwrap()).unwrap(); + + // A real one-file successor inherits unchanged source segments. + // Count production proof read callbacks rather than elapsed time: + // re-verifying the whole generation must fail even with warm OS caches. + let parent_file_digests = first_manifest["generation"]["file_segments"] + .as_array() + .unwrap() + .iter() + .map(|segment| segment["segment_digest"].as_str().unwrap()) + .collect::>(); + let inherited_files = successor["generation"]["file_segments"] + .as_array() + .unwrap() + .iter() + .filter(|segment| { + parent_file_digests.contains(segment["segment_digest"].as_str().unwrap()) + }) + .count(); + assert!( + inherited_files > 0, + "a one-file edit must retain sealed files" + ); + let count_proof_reads = |record: &VerifiedSealedSegmentsV1| { + let reads = AtomicUsize::new(0); + assert_eq!( + verify_sealed_generation_source_from_roots( + &generations_root, + replay_root, + &successor_digest, + record, + &|| { + reads.fetch_add(1, Ordering::SeqCst); + Ok(()) + }, + ), + Ok(()) + ); + reads.load(Ordering::SeqCst) + }; + let cold_reads = count_proof_reads(&VerifiedSealedSegmentsV1::default()); + let warm_reads = count_proof_reads(&verified); + if cfg!(unix) { + assert!( + cold_reads >= warm_reads + inherited_files, + "a one-file successor still re-verified inherited files: \ + cold={cold_reads}, warm={warm_reads}, inherited={inherited_files}" + ); + } else { + assert_eq!( + cold_reads, warm_reads, + "without reliable file fingerprints both passes must read all segments" + ); + } + + // A new segment must still be authenticated after the successor is + // warm. A proof that skips *all* segments could satisfy the read-count + // delta but would silently accept a corrupted edited file. + let changed_segment = successor["generation"]["file_segments"] + .as_array() + .unwrap() + .iter() + .find(|segment| { + !parent_file_digests.contains(segment["segment_digest"].as_str().unwrap()) + }) + .expect("a one-file successor must seal a changed file"); + let changed_digest = tracedecay_domain::ManifestDigest::new( + changed_segment["segment_digest"] + .as_str() + .unwrap() + .to_owned(), + ) + .unwrap(); + let changed_path = super::segment_path(&fixture.segments_root, &changed_digest).unwrap(); + let original_bytes = std::fs::read(&changed_path).unwrap(); + assert!(!original_bytes.is_empty()); + let successor_record = VerifiedSealedSegmentsV1::default(); + assert_eq!( + verify_sealed_generation_source_from_roots( + &generations_root, + replay_root, + &successor_digest, + &successor_record, + &|| Ok(()), + ), + Ok(()) + ); + // The check callback also runs for manifest and lifecycle checks, + // so a positive count alone does not prove a segment was rehashed. + let warmed_reads = count_proof_reads(&successor_record); + let original_permissions = std::fs::metadata(&changed_path).unwrap().permissions(); + let mut writable_permissions = original_permissions.clone(); + #[allow(clippy::permissions_set_readonly_false)] + writable_permissions.set_readonly(false); + std::fs::set_permissions(&changed_path, writable_permissions).unwrap(); + let mut tampered_bytes = original_bytes.clone(); + tampered_bytes[0] ^= 1; + std::fs::write(&changed_path, tampered_bytes).unwrap(); + let corrupt_changed = verify_sealed_generation_source_from_roots( + &generations_root, + replay_root, + &successor_digest, + &successor_record, + &|| Ok(()), + ); + assert!( + matches!(corrupt_changed, Err(GraphDbError::Corrupt { .. })), + "a warmed successor accepted a corrupt changed segment: {corrupt_changed:?}" + ); + std::fs::write(&changed_path, original_bytes).unwrap(); + std::fs::set_permissions(&changed_path, original_permissions).unwrap(); + + // A failed proof must not evict the last complete source proof. + // Rechecking the repaired changed segment is necessary, but rehashing + // every untouched inherited segment would make recovery cold again. + let recovered_reads = count_proof_reads(&successor_record); + if cfg!(unix) { + assert!( + recovered_reads > warmed_reads, + "recovery did not re-authenticate repaired source: warmed={warmed_reads}, recovered={recovered_reads}" + ); + } else { + // Without fingerprint reuse, a warm proof still rehashes every + // segment. The corruption refusal above checks edited source. + assert_eq!( + recovered_reads, warmed_reads, + "a platform without fingerprints must rehash every segment" + ); + } + let rewarmed_reads = count_proof_reads(&successor_record); + assert_eq!( + rewarmed_reads, warmed_reads, + "successful recovery must restore the stable warm proof cost" + ); + let recovery_cold_reads = count_proof_reads(&VerifiedSealedSegmentsV1::default()); + if cfg!(unix) { + assert!( + recovery_cold_reads >= recovered_reads + inherited_files, + "a corrupt one-file successor evicted authenticated parent segments: \ + cold={recovery_cold_reads}, recovered={recovered_reads}, \ + inherited={inherited_files}" + ); + } else { + assert_eq!( + recovery_cold_reads, recovered_reads, + "platforms without reliable fingerprints must rehash every segment" + ); + } + let inherited = successor["generation"]["file_segments"] .as_array_mut() .unwrap() diff --git a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs index a33657895d..7a44db51a0 100644 --- a/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs +++ b/crates/tracedecay-store-runtime/src/session_registry/verified_sealed_segments.rs @@ -12,7 +12,7 @@ use std::collections::{BTreeMap, HashMap}; use std::path::{Path, PathBuf}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use tracedecay_domain::ManifestDigest; @@ -53,30 +53,34 @@ impl SealedSegmentFingerprintV1 { } } -/// Per segments root, the segments of the generation last proven there. +type VerifiedSegmentMap = BTreeMap; + +/// Per segments root, the last successfully committed proof. In-flight proofs +/// share that immutable snapshot without evicting or copying its entries. #[derive(Debug, Default)] pub(crate) struct VerifiedSealedSegmentsV1 { - roots: Mutex>>, + roots: Mutex>>, } /// One proof's view: what it may skip, and what it hashed. pub(super) struct SealedSegmentProofV1<'v> { verified: &'v VerifiedSealedSegmentsV1, segments_root: PathBuf, - prior: BTreeMap, - proven: BTreeMap, + prior: Option>, + proven: VerifiedSegmentMap, } impl VerifiedSealedSegmentsV1 { - /// Start a proof under `segments_root`. The record leaves with it, so a - /// proof that fails leaves the next one to read every segment. + /// Start from the last successful proof. Cancellation, corruption or an + /// overlapping proof must not evict already authenticated identities. + /// New observations remain private until the whole proof succeeds. pub(super) fn begin(&self, segments_root: &Path) -> SealedSegmentProofV1<'_> { let prior = self .roots .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .remove(segments_root) - .unwrap_or_default(); + .get(segments_root) + .cloned(); SealedSegmentProofV1 { verified: self, segments_root: segments_root.to_path_buf(), @@ -95,7 +99,12 @@ impl SealedSegmentProofV1<'_> { expected_size: u64, path: &Path, ) -> bool { - let Some(prior) = self.prior.get(digest).copied() else { + let Some(prior) = self + .prior + .as_ref() + .and_then(|prior| prior.get(digest)) + .copied() + else { return false; }; let current = path @@ -121,13 +130,179 @@ impl SealedSegmentProofV1<'_> { } } - /// The proof authenticated every segment: remember exactly this - /// generation's segments, so the record stays one generation large. + /// Remember exactly this generation, but do not let a late proof replace + /// a snapshot another successful proof has committed since we began. pub(super) fn commit(self) { - self.verified + let mut roots = self + .verified .roots .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert(self.segments_root, self.proven); + .unwrap_or_else(std::sync::PoisonError::into_inner); + let unchanged = match (roots.get(&self.segments_root), self.prior.as_ref()) { + (None, None) => true, + (Some(current), Some(prior)) => Arc::ptr_eq(current, prior), + _ => false, + }; + if unchanged { + roots.insert(self.segments_root, Arc::new(self.proven)); + } + } +} + +#[cfg(all(test, unix))] +mod tests { + use std::fs::{File, FileTimes}; + use std::io::Read; + use std::path::{Path, PathBuf}; + + use sha2::{Digest, Sha256}; + use tempfile::TempDir; + use tracedecay_domain::{ManifestDigest, sha256_hex_suffix}; + + use super::{SealedSegmentFingerprintV1, SealedSegmentProofV1, VerifiedSealedSegmentsV1}; + + fn write_segment(root: &Path, bytes: &[u8]) -> (ManifestDigest, PathBuf) { + let digest = ManifestDigest::from_sha256_bytes(&Sha256::digest(bytes)).unwrap(); + let path = root.join(format!( + "segment-{}.json", + sha256_hex_suffix(digest.as_str()).unwrap() + )); + std::fs::write(&path, bytes).unwrap(); + (digest, path) + } + + fn authenticate(proof: &mut SealedSegmentProofV1<'_>, digest: &ManifestDigest, path: &Path) { + let mut file = File::open(path).unwrap(); + let fingerprint = SealedSegmentFingerprintV1::of(&file.metadata().unwrap()); + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes).unwrap(); + assert!(!bytes.is_empty()); + assert_eq!( + ManifestDigest::from_sha256_bytes(&Sha256::digest(&bytes)).unwrap(), + *digest + ); + proof.hashed(digest, fingerprint); + } + + #[test] + fn aborted_proof_keeps_parent_but_never_publishes_tentative_hashes() { + let root = TempDir::new().unwrap(); + let (parent, parent_path) = write_segment(root.path(), b"[1]"); + let (candidate, candidate_path) = write_segment(root.path(), b"[2]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &parent, &parent_path); + first.commit(); + + let mut interrupted = verified.begin(root.path()); + assert!(interrupted.already_verified(&parent, 3, &parent_path)); + // The reader records an identity before the caller checks its digest. + // A failed or cancelled proof must never publish that tentative fact. + std::fs::write(&candidate_path, b"[3]").unwrap(); + interrupted.hashed( + &candidate, + SealedSegmentFingerprintV1::of(&std::fs::metadata(&candidate_path).unwrap()), + ); + drop(interrupted); + + let mut retry = verified.begin(root.path()); + assert!( + retry.already_verified(&parent, 3, &parent_path), + "an interrupted successor must not make its authenticated parent cold" + ); + assert!( + !retry.already_verified(&candidate, 3, &candidate_path), + "a tentative digest from a failed proof must not become reusable" + ); + } + + #[test] + fn overlapping_proofs_share_parent_and_stale_commit_cannot_replace_successor() { + let root = TempDir::new().unwrap(); + let (parent, parent_path) = write_segment(root.path(), b"[1]"); + let (successor, successor_path) = write_segment(root.path(), b"[2]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &parent, &parent_path); + first.commit(); + + let mut stale = verified.begin(root.path()); + assert!(stale.already_verified(&parent, 3, &parent_path)); + let mut sibling = verified.begin(root.path()); + assert!( + sibling.already_verified(&parent, 3, &parent_path), + "an in-flight proof must not evict the parent's authenticated identities" + ); + drop(sibling); + let mut winner = verified.begin(root.path()); + authenticate(&mut winner, &successor, &successor_path); + winner.commit(); + stale.commit(); + + let mut latest = verified.begin(root.path()); + assert!(latest.already_verified(&successor, 3, &successor_path)); + assert!( + !latest.already_verified(&parent, 3, &parent_path), + "a stale proof must not resurrect segments pruned by a committed successor" + ); + } + + #[test] + fn retained_parent_rejects_same_size_replacement_with_restored_mtime() { + let root = TempDir::new().unwrap(); + let (digest, path) = write_segment(root.path(), b"[1]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &digest, &path); + first.commit(); + drop(verified.begin(root.path())); + + // Hold the original inode open so replacement cannot reuse it, even + // when the filesystem happens to assign identical timestamps. + let original = File::open(&path).unwrap(); + let metadata = original.metadata().unwrap(); + let before = SealedSegmentFingerprintV1::of(&metadata).unwrap(); + let (_, replacement_path) = write_segment(root.path(), b"[2]"); + File::options() + .write(true) + .open(&replacement_path) + .unwrap() + .set_times(FileTimes::new().set_modified(metadata.modified().unwrap())) + .unwrap(); + std::fs::rename(&replacement_path, &path).unwrap(); + let after = SealedSegmentFingerprintV1::of(&std::fs::metadata(&path).unwrap()).unwrap(); + assert_eq!(before.len, after.len); + assert_eq!(before.modified, after.modified); + assert_ne!(before.inode, after.inode); + assert_ne!( + ManifestDigest::from_sha256_bytes(&Sha256::digest(std::fs::read(&path).unwrap())) + .unwrap(), + digest + ); + let mut retry = verified.begin(root.path()); + assert!( + !retry.already_verified(&digest, 3, &path), + "retaining an old proof must not trust a replacement's same length and mtime" + ); + drop(original); + } + + #[test] + fn retained_proof_still_checks_root_extent_and_missing_source() { + let root = TempDir::new().unwrap(); + let other_root = TempDir::new().unwrap(); + let (digest, path) = write_segment(root.path(), b"[1]"); + let verified = VerifiedSealedSegmentsV1::default(); + let mut first = verified.begin(root.path()); + authenticate(&mut first, &digest, &path); + first.commit(); + + let mut foreign = verified.begin(other_root.path()); + assert!(!foreign.already_verified(&digest, 3, &path)); + let mut retry = verified.begin(root.path()); + assert!(!retry.already_verified(&digest, 4, &path)); + assert!(retry.already_verified(&digest, 3, &path)); + std::fs::remove_file(&path).unwrap(); + assert!(!retry.already_verified(&digest, 3, &path)); } } diff --git a/crates/tracedecay/src/daemon/engine/shutdown.rs b/crates/tracedecay/src/daemon/engine/shutdown.rs index 33ddcf77ea..f8e0c4d1ce 100644 --- a/crates/tracedecay/src/daemon/engine/shutdown.rs +++ b/crates/tracedecay/src/daemon/engine/shutdown.rs @@ -128,12 +128,19 @@ impl DaemonEngine { automation_join.shutdown_automation_schedulers().await; }, ), - ShutdownOwner::new("session_temporal_refresh", || {}, async move { - session_refresh.shutdown().await; - refresh_services - .release_profile_session_refresh_services() - .await; - }), + ShutdownOwner::new( + "session_temporal_refresh", + { + let refresh_cancel = Arc::clone(&session_refresh); + move || refresh_cancel.cancel_supervisors() + }, + async move { + session_refresh.shutdown().await; + refresh_services + .release_profile_session_refresh_services() + .await; + }, + ), ShutdownOwner::new( "host_admission_replay", { @@ -315,3 +322,68 @@ impl DaemonEngine { .await } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use tracedecay_daemon_service::shutdown::prepare_shutdown_owner_phases; + use tracedecay_domain::ProjectId; + use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; + use tracedecay_session_runtime::StoreOwnerKey; + + use super::DaemonEngine; + + #[tokio::test] + async fn session_refresh_is_cancelled_even_if_shutdown_joins_are_dropped() { + let directory = tempfile::tempdir().unwrap(); + let profile_root = directory.path().join("profile"); + let project_root = directory.path().join("project"); + let project_id = ProjectId::new("project.refresh-shutdown").unwrap(); + let fixture = RegisteredGlobalDbTestRuntime::project( + &profile_root, + &project_root, + project_id.clone(), + ) + .await + .unwrap(); + let database = fixture.project_database_arc().unwrap(); + let store_root = database.db_path().parent().unwrap().to_path_buf(); + let owner = StoreOwnerKey { + global_db_path: profile_root.join("global.db"), + profile_root, + project_id: Some(project_id.as_str().to_owned()), + graph_db_path: store_root.join("graph.db"), + store_root, + }; + let engine = DaemonEngine::default(); + let registry = Arc::clone( + engine + .store_administration + .session_temporal_refresh_schedulers(), + ); + let wake = registry.ensure_project(owner.clone(), database).await; + let state = registry.project_state(&owner).await.unwrap(); + assert_eq!(registry.project_worker_count().await, 1); + assert!(wake.status().is_available()); + + let owner_phases = engine.shutdown_owner_phases().await; + assert!(!state.is_cancelled()); + let prepared = prepare_shutdown_owner_phases(owner_phases); + let cancelled_before_join = state.is_cancelled(); + + // An earlier phase can exhaust the shared deadline before this + // owner's join is polled. Cancellation must already have reached + // the real supervisor even if every prepared join is discarded. + drop(prepared); + tokio::time::timeout(Duration::from_secs(5), registry.shutdown()) + .await + .expect("the cancelled refresh supervisor must join"); + assert_eq!(registry.project_worker_count().await, 0); + assert!( + cancelled_before_join, + "refresh must be cancelled during preparation, not first inside its join" + ); + } +} diff --git a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs index db5f5392b9..562b8d2bf7 100644 --- a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs +++ b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs @@ -5,10 +5,9 @@ use std::path::Path; use std::sync::{Arc, Weak}; -use tracedecay_contracts::ApplicationProblem; -use tracedecay_contracts::ResolvedScope; use tracedecay_contracts::graph_tool::GraphToolResultV1; use tracedecay_contracts::retrieval::HookRuntimeResultV1; +use tracedecay_contracts::{ApplicationProblem, CancellationStage, ResolvedScope, RetryDirective}; use tracedecay_daemon_service::{ DaemonInvocationService, GraphToolFuture, GraphToolInvocationV1, ProjectGraphToolPortV1, RegisteredGraphToolOwnerV1, @@ -51,14 +50,48 @@ impl ProjectGraphToolPortV1 for McpGraphToolPort { { return Err(ApplicationProblem::runtime_mounting()); } + let operation = invocation.operation; server .compute_graph_tool(invocation) .await - .map_err(|error| graph_tool_error_problem(&error)) + .map_err(|error| graph_tool_execution_problem(operation, &error)) }) } } +/// Dispatch and worker timers can both settle an admitted read. A raw +/// interruption returned by either must retain that read's terminal kind, +/// not become an authority-unavailable refusal from before admission. +/// Owner-supplied records/details and side-effect settlement stay authoritative. +fn graph_tool_execution_problem( + operation: ApplicationSurfaceOperation, + error: &TraceDecayError, +) -> ApplicationProblem { + if operation.owner_side_effect().is_none() + && error.project_route_problem_record().is_none() + && error.project_route_typed_detail().is_none() + { + match error.project_route_context().map(|(reason, _, _)| reason) { + Some("tool_dispatch_deadline_exceeded") => { + return ApplicationProblem::TimedOut { + stage: CancellationStage::DuringRead, + retry: RetryDirective::Never, + legal_actions: Vec::new(), + }; + } + Some("tool_dispatch_cancelled") => { + return ApplicationProblem::Cancelled { + stage: CancellationStage::DuringRead, + retry: RetryDirective::Never, + legal_actions: Vec::new(), + }; + } + _ => {} + } + } + graph_tool_error_problem(error) +} + impl McpServer { /// Registry options carrying every owner this server composes; each /// dispatch overrides only its per-call fields. @@ -249,3 +282,105 @@ impl McpServer { }) } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use serde_json::json; + use tracedecay_contracts::{ApplicationProblem, CancellationStage, RequestId}; + use tracedecay_domain::errors::TraceDecayError; + use tracedecay_mcp::server::{DispatchSettlement, dispatch_cancelled_error}; + use tracedecay_mcp::tools::dispatch_ceiling::tool_dispatch_deadline_error; + use tracedecay_tool_catalog::ApplicationSurfaceOperation; + + use super::graph_tool_execution_problem; + use crate::mcp::tools::graph_tool_error_problem; + + #[test] + fn admitted_graph_read_interruptions_preserve_terminal_kind_and_stage() { + let operation = ApplicationSurfaceOperation::RenamePreview; + assert!(operation.owner_side_effect().is_none()); + for (error, expected, kind) in [ + ( + tool_dispatch_deadline_error(operation.mcp_tool_name(), Duration::ZERO), + ApplicationProblem::timed_out(CancellationStage::DuringRead).unwrap(), + "timed_out", + ), + ( + dispatch_cancelled_error( + operation.mcp_tool_name(), + DispatchSettlement::Settling, + false, + ), + ApplicationProblem::cancelled(CancellationStage::DuringRead).unwrap(), + "cancelled", + ), + ] { + let problem = graph_tool_execution_problem(operation, &error); + assert_eq!(problem, expected); + let envelope = tracedecay_api::adapter_problem( + RequestId::new("request.graph-read.interrupted").unwrap(), + problem, + ) + .unwrap(); + let wire = serde_json::to_value(envelope).unwrap(); + assert_eq!(wire["problem"]["kind"], kind); + assert_eq!(wire["problem"]["code"], kind); + assert_eq!(wire["problem"]["cancellation_stage"], "during_read"); + assert_eq!(wire["problem"]["terminality"], "admitted_terminal"); + assert_eq!(wire["problem"]["retry"], "never"); + assert_eq!(wire["problem"]["legal_actions"], json!([])); + } + } + + #[test] + fn graph_read_execution_preserves_owner_problem_records_and_details() { + let operation = ApplicationSurfaceOperation::RenamePreview; + let expected = ApplicationProblem::timed_out_before_admission(); + let envelope = tracedecay_api::adapter_problem( + RequestId::new("request.graph-read.owner-refusal").unwrap(), + expected.clone(), + ) + .unwrap(); + let error = tool_dispatch_deadline_error(operation.mcp_tool_name(), Duration::ZERO) + .with_problem_record(serde_json::to_value(&envelope.problem).unwrap()); + assert_eq!(graph_tool_execution_problem(operation, &error), expected); + + let error = tracedecay_graph_query::map_code_graph_read_runtime_error( + tracedecay_graph_query::CodeGraphReadError::Rewarming { + retry_after_millis: 7_940, + }, + ); + let detail = error.project_route_typed_detail().unwrap().clone(); + let expected = graph_tool_error_problem(&error); + let error = TraceDecayError::project_route_with_detail( + "tool_dispatch_deadline_exceeded", + true, + detail, + ); + assert_eq!(graph_tool_execution_problem(operation, &error), expected); + } + + #[test] + fn graph_read_interruption_mapping_does_not_reclassify_effects_or_other_failures() { + let effect = ApplicationSurfaceOperation::RunAffectedTests; + assert!(effect.owner_side_effect().is_some()); + let timeout = tool_dispatch_deadline_error(effect.mcp_tool_name(), Duration::ZERO); + assert_eq!( + graph_tool_execution_problem(effect, &timeout), + graph_tool_error_problem(&timeout) + ); + for reason in [ + "tool_dispatch_effect_unknown", + "tool_dispatch_shutdown", + "verified-code-graph-read-unavailable", + ] { + let error = TraceDecayError::project_route(reason, true, "owner refusal"); + assert_eq!( + graph_tool_execution_problem(ApplicationSurfaceOperation::RenamePreview, &error), + graph_tool_error_problem(&error) + ); + } + } +}