diff --git a/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs index 27cdd9dab3d..674f012bc11 100644 --- a/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs +++ b/crates/tracedecay-cli/tests/core_cli_suite/monitor_test.rs @@ -5,7 +5,8 @@ use tempfile::TempDir; fn write(dir: &Path, project: &Path, prefix: &str, tool: &str, delta: u64, before: u64) { tracedecay_session_memory::monitor_ring::write_entry_to( dir, project, prefix, tool, delta, before, - ); + ) + .expect("write monitor fixture entry"); } /// Helper: open reader at a specific mmap dir. diff --git a/crates/tracedecay-session-memory/src/monitor_ring.rs b/crates/tracedecay-session-memory/src/monitor_ring.rs index 71140fbbddd..0ad4e350485 100644 --- a/crates/tracedecay-session-memory/src/monitor_ring.rs +++ b/crates/tracedecay-session-memory/src/monitor_ring.rs @@ -63,8 +63,8 @@ impl MonitorEntry { /// `dir`. /// /// `project_root` is used to derive the folder name. `prefix` identifies -/// the tool suite (e.g. "tracedecay"). Best-effort: silently returns on -/// any failure. +/// the tool suite (e.g. "tracedecay"). Writer failures are returned to +/// the caller so accounting can report a degraded live-monitor projection. pub fn write_entry_to( dir: &Path, project_root: &Path, @@ -72,13 +72,13 @@ pub fn write_entry_to( tool_name: &str, delta: u64, before: u64, -) { - let _ = std::fs::create_dir_all(dir); +) -> std::io::Result<()> { + std::fs::create_dir_all(dir)?; let mmap_path = dir.join(MMAP_FILENAME); let project = project_root .file_name() .map_or_default(|n| n.to_string_lossy().to_string()); - let _ = write_entry_inner(&mmap_path, prefix, &project, tool_name, delta, before); + write_entry_inner(&mmap_path, prefix, &project, tool_name, delta, before) } fn write_str(mmap: &mut memmap2::MmapMut, offset: usize, value: &str) { @@ -326,7 +326,8 @@ mod tests { fn write_values(dir: &Path, values: std::ops::RangeInclusive) { let project = Path::new("/projects/alpha"); for value in values { - write_entry_to(dir, project, "tracedecay", "search", value, 0); + write_entry_to(dir, project, "tracedecay", "search", value, 0) + .expect("write monitor entry"); } } @@ -442,16 +443,41 @@ mod tests { ); } + #[test] + fn writer_reports_unavailable_directory_and_ring_without_clobbering_them() { + let dir = tempfile::tempdir().expect("monitor fixture directory"); + let project = Path::new("/projects/alpha"); + + // A non-directory owner path cannot be turned into a monitor cache. + let occupied_dir = dir.path().join("occupied"); + std::fs::write(&occupied_dir, b"owner data").expect("create occupied path"); + write_entry_to(&occupied_dir, project, "tracedecay", "search", 7, 10) + .expect_err("a file used as the monitor directory must be refused"); + assert_eq!( + std::fs::read(&occupied_dir).expect("read original owner file"), + b"owner data" + ); + + // Opening a directory as the mmap is a separate post-mkdir failure. + let occupied_ring = dir.path().join(MMAP_FILENAME); + std::fs::create_dir(&occupied_ring).expect("create occupied mmap path"); + write_entry_to(dir.path(), project, "tracedecay", "search", 7, 10) + .expect_err("an occupied mmap path must be refused"); + assert!(occupied_ring.is_dir()); + } + /// Writers no longer msync, so an already-open reader must still observe /// every entry through its existing mapping, without a refresh. #[test] fn open_reader_observes_every_unflushed_entry() { let dir = tempfile::tempdir().expect("monitor dir"); let project = tempfile::tempdir().expect("project dir"); - write_entry_to(dir.path(), project.path(), "tracedecay", "first", 1, 2); + write_entry_to(dir.path(), project.path(), "tracedecay", "first", 1, 2) + .expect("write initial monitor entry"); let reader = MmapReader::open_at(dir.path()).expect("open monitor"); for call in 1..40_u64 { - write_entry_to(dir.path(), project.path(), "tracedecay", "poll", call, 2); + write_entry_to(dir.path(), project.path(), "tracedecay", "poll", call, 2) + .expect("write monitor poll entry"); } assert_eq!(reader.write_idx(), 40); let last = reader.entry(39).expect("last slot"); diff --git a/crates/tracedecay/src/mcp/server/freshness_tests.rs b/crates/tracedecay/src/mcp/server/freshness_tests.rs index 49aefd70556..97588de6b24 100644 --- a/crates/tracedecay/src/mcp/server/freshness_tests.rs +++ b/crates/tracedecay/src/mcp/server/freshness_tests.rs @@ -84,7 +84,8 @@ async fn savings_without_a_display_baseline_still_commit_to_the_project_store() let (cg, _dir, _authority) = init_indexed_repo().await; assert_eq!(cg.get_tokens_saved().await.unwrap(), 0); - super::ledger::commit_token_savings(&cg, None, super::ledger::LedgerSink::NotMounted, 13).await; + super::ledger::commit_token_savings(&cg, None, super::ledger::LedgerSink::NotMounted, 13, None) + .await; assert_eq!(cg.get_tokens_saved().await.unwrap(), 13); assert_eq!(cg.get_local_counter().await.unwrap(), 13); @@ -96,6 +97,7 @@ async fn savings_without_a_display_baseline_still_commit_to_the_project_store() Some(&display), super::ledger::LedgerSink::NotMounted, 7, + None, ) .await; assert_eq!(display.load(Ordering::Relaxed), 20); diff --git a/crates/tracedecay/src/mcp/server/ledger.rs b/crates/tracedecay/src/mcp/server/ledger.rs index 764a5abedde..6f8587d411b 100644 --- a/crates/tracedecay/src/mcp/server/ledger.rs +++ b/crates/tracedecay/src/mcp/server/ledger.rs @@ -89,18 +89,28 @@ pub(crate) struct McpToolErrorAnalyticsRequest<'a> { /// Commits a response delta even when startup could not read the display /// baseline. The optional cache and global projection consume committed data; /// neither can suppress the authoritative project-store write. +/// Whether the canonical project-store counter committed. A failed optional +/// global projection cannot undo the local commit; the live monitor credits +/// only this durable outcome, not the response's optimistic estimate. pub(super) async fn commit_token_savings( cg: &TraceDecay, tokens_saved: Option<&AtomicU64>, sink: LedgerSink, delta: u64, -) { + commit_sender: Option>, +) -> bool { match cg.record_token_savings(delta).await { Ok(committed_total) => { // Two completions may resume in reverse order after their commits. if let Some(tokens_saved) = tokens_saved { tokens_saved.fetch_max(committed_total, Ordering::Relaxed); } + // A successful project-store transaction is the response's durable + // credit. Release its receipt before the optional profile summary: + // that projection has a different writer gate and may stall or fail. + if let Some(sender) = commit_sender { + let _ = sender.send(true); + } if let LedgerSink::Mounted(gdb) = sink && let Err(error) = gdb .try_upsert_project_tokens(cg.project_root(), committed_total) @@ -112,13 +122,18 @@ pub(super) async fn commit_token_savings( "background token-accounting projection failed" ); } + true } Err(error) => { + if let Some(sender) = commit_sender { + let _ = sender.send(false); + } tracing::warn!( project_root = %cg.project_root().display(), %error, "background token-accounting commit failed" ); + false } } } @@ -168,7 +183,16 @@ impl McpServer { tool_name: &str, net_saved_tokens: u64, raw_file_tokens: u64, - ) { + ) -> Option> { + // The independently scheduled analytics writer may report positive + // savings only after this exact response's store transaction commits. + // A dropped task closes the channel and means refusal, not success. + let (commit_sender, commit_receiver) = if net_saved_tokens == 0 { + (None, None) + } else { + let (sender, receiver) = tokio::sync::oneshot::channel(); + (Some(sender), Some(receiver)) + }; // An unreadable baseline suppresses only the approximate display cache. // It must not disable future durable deltas if that read failure clears. let persist = (net_saved_tokens != 0).then(|| { @@ -183,9 +207,26 @@ impl McpServer { .map(|profile| profile.data_dir().to_path_buf()); let tool_name = tool_name.to_owned(); self.spawn_observed_ledger_write(async move { - if let Some((tokens_saved, sink)) = persist { - commit_token_savings(&cg, tokens_saved.as_deref(), sink, net_saved_tokens).await; - } + // The monitor is an activity feed, not a second savings authority. + // Preserve the activity's raw input size after failure, but never + // credit a token delta that the project store could not commit. + let committed_saved_tokens = if let Some((tokens_saved, sink)) = persist { + if commit_token_savings( + &cg, + tokens_saved.as_deref(), + sink, + net_saved_tokens, + commit_sender, + ) + .await + { + net_saved_tokens + } else { + 0 + } + } else { + 0 + }; // The monitor entry opens, locks, and mmaps a file; keep that // off the async workers. // A server without an owning profile has no monitor ring. @@ -199,14 +240,21 @@ impl McpServer { &monitor_project_root, "tracedecay", &tool_name, - net_saved_tokens, + committed_saved_tokens, raw_file_tokens, - ); + ) }); - if let Err(error) = monitor_write.await { - tracing::warn!(error = %error, "live monitor entry write failed"); + match monitor_write.await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + tracing::warn!(%error, "live monitor entry I/O failed"); + } + Err(error) => { + tracing::warn!(%error, "live monitor entry task failed"); + } } }); + commit_receiver } /// Resolves once every savings-ledger write spawned so far has @@ -501,6 +549,7 @@ impl McpServer { #[cfg(test)] mod tests { use serde_json::json; + use tracedecay_mcp::tools::response_trailers::ToolTokenAccounting; use tracedecay_mcp::transport::JsonRpcRequest; use super::*; @@ -532,7 +581,7 @@ mod tests { .expect("tool call should produce a response"); server.ledger_writes_settled().await; let accounted = tokens_saved.load(Ordering::Acquire); - server.spawn_token_accounting_persist( + let _ = server.spawn_token_accounting_persist( server.cg_snapshot().await, &project_root, "tracedecay_search", @@ -567,7 +616,7 @@ mod tests { ) .await; - first.spawn_token_accounting_persist( + let _ = first.spawn_token_accounting_persist( first.cg_snapshot().await, &project_root, "tracedecay_search", @@ -577,7 +626,7 @@ mod tests { first.ledger_writes_settled().await; // The second server was created with the same baseline before the // first credit. Its own completion must add to the committed total. - second.spawn_token_accounting_persist( + let _ = second.spawn_token_accounting_persist( second.cg_snapshot().await, &project_root, "tracedecay_search", @@ -614,6 +663,273 @@ mod tests { assert_eq!(reopened.get_local_counter().await.unwrap(), 35); } + #[tokio::test] + async fn profile_projection_writer_cannot_hold_a_committed_response_receipt() { + let (cg, _project, authority) = + super::super::writer_test_support::init_indexed_repo().await; + let project_root = cg.project_root().to_path_buf(); + let server = super::super::McpServer::new_with_context( + super::super::writer_test_support::registered_context(cg, &authority), + ) + .await; + let accounting_db = server + .accounting_db + .as_ref() + .expect("registered profile projection") + .clone(); + + // Hold only the optional profile store's write gate. The canonical + // project store is independent and must still settle this response. + let held_projection = accounting_db + .begin_write_transaction() + .await + .expect("hold profile writer while project commit proceeds"); + let receipt = server + .spawn_token_accounting_persist( + server.cg_snapshot().await, + &project_root, + "tracedecay_search", + 7, + 20, + ) + .expect("nonzero savings need an explicit commit receipt"); + let credited = tokio::time::timeout(Duration::from_secs(5), receipt) + .await + .expect("the project receipt cannot depend on the profile projection") + .expect("accepted accounting writer must settle the receipt"); + assert!(credited, "the canonical project-store credit must commit"); + assert_eq!( + server.cg_snapshot().await.get_tokens_saved().await.unwrap(), + 7 + ); + + held_projection + .rollback() + .await + .expect("release the held profile writer"); + assert!( + server + .ledger_writes_settled_within(Duration::from_secs(10)) + .await, + "the optional projection should finish after its writer is released" + ); + assert_eq!( + accounting_db + .try_get_project_tokens(&project_root) + .await + .expect("committed profile projection"), + 7 + ); + server.shutdown_background_tasks().await; + } + + #[tokio::test] + async fn live_monitor_reports_only_committed_token_savings() { + let (cg, _project, authority) = + super::super::writer_test_support::init_indexed_repo().await; + let project_root = cg.project_root().to_path_buf(); + let server = super::super::McpServer::new_with_context( + super::super::writer_test_support::registered_context(cg, &authority), + ) + .await; + let monitor_dir = server + .owner_profile() + .expect("fixture has an isolated owner profile") + .data_dir() + .to_path_buf(); + + let _ = server.spawn_token_accounting_persist( + server.cg_snapshot().await, + &project_root, + "tracedecay_search", + 7, + 20, + ); + server.ledger_writes_settled().await; + { + let ring = tracedecay_session_memory::monitor_ring::MmapReader::open_at(&monitor_dir) + .expect("committed call created monitor"); + assert_eq!(ring.write_idx(), 1); + let entry = ring.entry(0).expect("committed response entry"); + assert_eq!(entry.delta, 7); + assert_eq!(entry.before, 20); + assert_eq!(entry.tool_name, "tracedecay_search"); + } + assert_eq!( + server.cg_snapshot().await.get_tokens_saved().await.unwrap(), + 7 + ); + + // A real committed maximum forces the next recorder's checked-add + // refusal, without modifying production code or faking an error. + server + .cg_snapshot() + .await + .record_token_savings(u64::MAX - 7) + .await + .expect("seed a durable lifetime overflow boundary"); + let _ = server.spawn_token_accounting_persist( + server.cg_snapshot().await, + &project_root, + "tracedecay_search", + 1, + 30, + ); + server.ledger_writes_settled().await; + let ring = tracedecay_session_memory::monitor_ring::MmapReader::open_at(&monitor_dir) + .expect("failed accounting still records activity"); + assert_eq!(ring.write_idx(), 2); + let entry = ring.entry(1).expect("failed response activity entry"); + assert_eq!(entry.delta, 0, "uncommitted savings must not be credited"); + assert_eq!(entry.before, 30, "raw tool activity remains observable"); + assert_eq!(entry.tool_name, "tracedecay_search"); + assert_eq!( + server.cg_snapshot().await.get_tokens_saved().await.unwrap(), + u64::MAX + ); + assert_eq!( + server + .cg_snapshot() + .await + .get_local_counter() + .await + .unwrap(), + u64::MAX, + "the refused delta must not change the resettable counter" + ); + server.shutdown_background_tasks().await; + } + + #[tokio::test] + async fn refused_counter_commit_never_credits_independent_savings_history() { + let (cg, _project, authority) = + super::super::writer_test_support::init_indexed_repo().await; + let project_root = cg.project_root().to_path_buf(); + let server = super::super::McpServer::new_with_context( + super::super::writer_test_support::registered_context(cg, &authority), + ) + .await; + let accounting_db = server + .accounting_db + .as_ref() + .expect("registered accounting writer") + .clone(); + let normal = ToolTokenAccounting { + raw_file_tokens: 20, + response_tokens: 13, + }; + let result = tracedecay_mcp::ToolResult::new(json!({"content": []}), Vec::new()); + + let first_commit = server.spawn_token_accounting_persist( + server.cg_snapshot().await, + &project_root, + "tracedecay_search", + normal.net_saved_tokens(), + normal.raw_file_tokens, + ); + server.spawn_success_analytics( + &project_root, + "tracedecay_search", + &json!("request.first"), + &json!({}), + &None, + true, + None, + normal, + &result, + None, + None, + first_commit, + ); + server.ledger_writes_settled().await; + let project_key = project_root.to_str().expect("UTF-8 fixture project"); + let history = accounting_db + .sum_savings(Some(project_key), 0) + .await + .expect("registered savings history"); + assert_eq!(history.saved_tokens, 7); + assert_eq!(history.calls, 1); + + // A real checked-add failure prevents the second response's positive + // durable credit. The independent analytics writer used to race ahead + // and persist that failed response as another saved-token event. + server + .cg_snapshot() + .await + .record_token_savings(u64::MAX - 7) + .await + .expect("seed canonical overflow boundary"); + let refused = ToolTokenAccounting { + raw_file_tokens: 30, + response_tokens: 29, + }; + let second_commit = server.spawn_token_accounting_persist( + server.cg_snapshot().await, + &project_root, + "tracedecay_search", + refused.net_saved_tokens(), + refused.raw_file_tokens, + ); + server.spawn_success_analytics( + &project_root, + "tracedecay_search", + &json!("request.refused"), + &json!({}), + &None, + true, + None, + refused, + &result, + None, + None, + second_commit, + ); + server.ledger_writes_settled().await; + + let history = accounting_db + .sum_savings(Some(project_key), 0) + .await + .expect("savings history after refused commit"); + assert_eq!(history.saved_tokens, 7, "failed commit cannot add savings"); + assert_eq!(history.calls, 1, "failed commit cannot add a savings row"); + assert_eq!( + server.cg_snapshot().await.get_tokens_saved().await.unwrap(), + u64::MAX, + "refused accounting cannot alter the canonical project counter" + ); + + // Cost and raw activity remain independently observable; their + // metadata must not claim a saving that failed its canonical commit. + let snapshot = accounting_db + .read_snapshot() + .await + .expect("analytics snapshot"); + let mut rows = snapshot + .query( + "SELECT metadata_json FROM analytics_events + WHERE event_kind = 'mcp_tool_call' AND tool_name = 'tracedecay_search' + ORDER BY id", + (), + ) + .await + .expect("registered analytics query"); + let mut metadata = Vec::new(); + while let Some(row) = rows.next().await.expect("analytics row") { + let encoded = row.get::(0).expect("metadata JSON"); + metadata.push( + serde_json::from_str::(&encoded) + .expect("valid analytics metadata"), + ); + } + assert_eq!(metadata.len(), 2, "both tool activities remain visible"); + assert_eq!(metadata[0]["tokens_saved"], json!(7)); + assert_eq!(metadata[1]["tokens_saved"], json!(0)); + assert_eq!(metadata[1]["savings_commit"], json!("refused")); + assert_eq!(metadata[1]["before_tokens"], json!(30)); + assert_eq!(metadata[1]["after_tokens"], json!(29)); + server.shutdown_background_tasks().await; + } + #[tokio::test] async fn a_failed_accounting_commit_does_not_advance_the_display_cache() { let (cg, _project, authority) = @@ -624,7 +940,7 @@ mod tests { super::super::writer_test_support::registered_context(cg, &authority), ) .await; - server.spawn_token_accounting_persist( + let _ = server.spawn_token_accounting_persist( server.cg_snapshot().await, &project_root, "tracedecay_search", diff --git a/crates/tracedecay/src/mcp/server/requests.rs b/crates/tracedecay/src/mcp/server/requests.rs index 805918b00e0..ccdee99054b 100644 --- a/crates/tracedecay/src/mcp/server/requests.rs +++ b/crates/tracedecay/src/mcp/server/requests.rs @@ -1057,21 +1057,24 @@ impl McpServer { cg: &Arc, tool_name: &str, result: &mut ToolResult, - ) -> ToolTokenAccounting { + ) -> ( + ToolTokenAccounting, + Option>, + ) { // A result the shared renderer already accounted carries its figures // and footer; only persist them. let accounting = match result.token_accounting() { Some(accounting) => accounting, None => self.account_unrendered_result(cg, result).await, }; - self.spawn_token_accounting_persist( + let commitment = self.spawn_token_accounting_persist( Arc::clone(cg), cg.project_root(), tool_name, accounting.net_saved_tokens(), accounting.raw_file_tokens, ); - accounting + (accounting, commitment) } async fn account_unrendered_result( @@ -1096,7 +1099,7 @@ impl McpServer { } #[allow(clippy::too_many_arguments)] - fn spawn_success_analytics( + pub(super) fn spawn_success_analytics( &self, accounting_project_root: &Path, tool_name: &str, @@ -1109,6 +1112,7 @@ impl McpServer { result: &ToolResult, connection_client_name: Option<&str>, connection_instance_id: Option<&str>, + commit_confirmation: Option>, ) { let analytics_outcome = if tool_result_has_semantic_error(result) { "error" @@ -1136,7 +1140,7 @@ impl McpServer { let failure_reason = (analytics_outcome == "error") .then(|| semantic_failure_reason(result)) .flatten(); - let analytics_event = mcp_tool_analytics_event(McpToolAnalyticsEvent { + let mut analytics_event = mcp_tool_analytics_event(McpToolAnalyticsEvent { project_root: accounting_project_root, session_id: analytics_session_id.clone(), tool_name, @@ -1155,7 +1159,34 @@ impl McpServer { cost: result.cost().as_ref(), }); self.spawn_observed_ledger_write(async move { + // Two independently scheduled writes may run in either order. + // Only the receipt for this response (not a project total + // sampled after another response) proves its durable credit. + let committed = match commit_confirmation { + Some(receiver) => receiver.await.unwrap_or(false), + None => net_saved_tokens == 0, + }; + if !committed { + // Keep the observed before/after activity and actual cost, + // but never claim a positive saving in analytics or history. + let Some(metadata_json) = analytics_event.metadata_json.as_mut() else { + tracing::warn!("MCP analytics credit reconciliation lacked metadata"); + return; + }; + let Ok(mut metadata) = serde_json::from_str::(metadata_json) else { + tracing::warn!("MCP analytics credit reconciliation rejected metadata"); + return; + }; + if !metadata.is_object() { + tracing::warn!("MCP analytics credit reconciliation rejected a nonobject"); + return; + } + metadata["tokens_saved"] = json!(0); + metadata["savings_commit"] = json!("refused"); + *metadata_json = metadata.to_string(); + } if let Some(registered) = savings_db + && committed && let Err(e) = registered .try_record_savings( &project_path_str, @@ -1197,7 +1228,8 @@ impl McpServer { connection_client_name: Option<&str>, connection_instance_id: Option<&str>, ) { - let accounting = self.apply_token_accounting(cg, tool_name, result).await; + let (accounting, commit_confirmation) = + self.apply_token_accounting(cg, tool_name, result).await; self.spawn_success_analytics( accounting_project_root, tool_name, @@ -1210,6 +1242,7 @@ impl McpServer { result, connection_client_name, connection_instance_id, + commit_confirmation, ); }