fix(trace): observe MCP audit tail and bounded capture loss

crates/webcodex-store/src/connection_observation.rs

@@ -1,9 +1,70 @@
 use rusqlite::Connection;
+use std::cell::RefCell;
 use std::ops::{Deref, DerefMut};
 use std::panic::{catch_unwind, AssertUnwindSafe};
 use std::sync::{Mutex, MutexGuard};
 use std::time::{Duration, Instant};
 
+const SLOW_CONNECTION: Duration = Duration::from_millis(100);
+type SlowConnectionCallback = fn(&str, &'static str, Duration, Duration);
+#[derive(Clone)]
+struct ConnectionTrace {
+    trace_id: String,
+    on_slow: SlowConnectionCallback,
+}
+thread_local! { static CONNECTION_TRACE: RefCell<Option<ConnectionTrace>> = const { RefCell::new(None) }; }
+
+/// Correlate synchronous Store work without depending on a Tokio task-local.
+/// Never hold this scope across an await. Only bounded opaque IDs are admitted;
+/// callbacks run after unlocking, only for slow operations, and fail open.
+pub fn with_connection_trace<T>(
+    trace_id: Option<&str>,
+    on_slow: SlowConnectionCallback,
+    work: impl FnOnce() -> T,
+) -> T {
+    struct Restore(Option<ConnectionTrace>);
+    impl Drop for Restore {
+        fn drop(&mut self) {
+            CONNECTION_TRACE.with(|slot| {
+                *slot.borrow_mut() = self.0.take();
+            });
+        }
+    }
+    let context = trace_id
+        .filter(|id| {
+            !id.is_empty()
+                && id.len() <= 64
+                && id
+                    .bytes()
+                    .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
+        })
+        .map(|id| ConnectionTrace {
+            trace_id: id.to_owned(),
+            on_slow,
+        });
+    let _restore = Restore(CONNECTION_TRACE.with(|slot| slot.replace(context)));
+    work()
+}
+
+fn observe_slow_connection(domain: StoreDomain, wait: Duration, hold: Duration) {
+    if wait < SLOW_CONNECTION && hold < SLOW_CONNECTION {
+        return;
+    }
+    // Clone only on the slow path; never invoke a callback while borrowing the
+    // thread-local slot. A subscriber may itself perform nested Store work.
+    let context = CONNECTION_TRACE.with(|slot| slot.borrow().clone());
+    let trace_id = context
+        .as_ref()
+        .map(|ctx| ctx.trace_id.as_str())
+        .unwrap_or("-");
+    tracing::info!(target: "webcodex_store::connection", event="store_connection_slow", server_trace_id=trace_id,
+            domain=domain.as_str(), wait_ms=wait.as_secs_f64()*1000.0, hold_ms=hold.as_secs_f64()*1000.0,
+            "slow store connection");
+    if let Some(context) = context.as_ref() {
+        (context.on_slow)(&context.trace_id, domain.as_str(), wait, hold);
+    }
+}
+
 pub(crate) const STORE_CONNECTION_ACQUISITIONS_TOTAL: &str = "store_connection_acquisitions_total";
 pub(crate) const STORE_CONNECTION_LOCK_WAIT_SECONDS: &str = "store_connection_lock_wait_seconds";
 pub(crate) const STORE_CONNECTION_HOLD_SECONDS: &str = "store_connection_hold_seconds";
@@ -201,6 +262,7 @@ impl Drop for StoreConnectionGuard<'_> {
         // therefore cannot inflate the measured critical section or serialize
         // another store caller behind a slow subscriber.
         drop(self.guard.take());
+        observe_fail_open(|| observe_slow_connection(self.domain, self.wait, hold));
         observe_fail_open(|| self.observer.record_acquisition(self.domain, self.wait));
         observe_fail_open(|| self.observer.record_hold(self.domain, hold));
         tracing::debug!(target: "webcodex_store::connection", domain=self.domain.as_str(),
@@ -212,3 +274,7 @@ impl Drop for StoreConnectionGuard<'_> {
 fn observe_fail_open(observe: impl FnOnce()) {
     let _ = catch_unwind(AssertUnwindSafe(observe));
 }
+
+#[cfg(test)]
+#[path = "connection_observation_tests.rs"]
+mod tests;

crates/webcodex-store/src/connection_observation_tests.rs

@@ -0,0 +1,47 @@
+use super::*;
+
+thread_local! { static OBSERVED: RefCell<Vec<String>> = const { RefCell::new(Vec::new()) }; }
+fn record(id: &str, domain: &'static str, _: Duration, _: Duration) {
+    assert_eq!(domain, "audit");
+    OBSERVED.with(|events| events.borrow_mut().push(id.into()));
+    // Prove callback re-entry does not retain a borrow of the context slot.
+    with_connection_trace(None, record, || {});
+}
+fn slow_wait() {
+    observe_slow_connection(StoreDomain::Audit, SLOW_CONNECTION, Duration::ZERO);
+}
+
+#[test]
+fn slow_connection_identity_is_bounded_nested_and_unwind_safe() {
+    OBSERVED.with(|events| events.borrow_mut().clear());
+    with_connection_trace(Some("outer"), record, || {
+        observe_slow_connection(StoreDomain::Audit, Duration::ZERO, Duration::ZERO);
+        slow_wait();
+        with_connection_trace(Some("inner"), record, || {
+            observe_slow_connection(StoreDomain::Audit, Duration::ZERO, SLOW_CONNECTION);
+        });
+        for invalid in ["", "secret with spaces", "sql;", &"a".repeat(65)] {
+            with_connection_trace(Some(invalid), record, slow_wait);
+        }
+        let result =
+            catch_unwind(|| with_connection_trace(Some("panics"), record, || panic!("fixture")));
+        assert!(result.is_err());
+        slow_wait();
+    });
+    slow_wait(); // No stale context after leaving the synchronous scope.
+    OBSERVED.with(|events| assert_eq!(*events.borrow(), ["outer", "inner", "outer"]));
+}
+
+#[test]
+fn slow_callback_failure_cannot_poison_or_hold_the_store_mutex() {
+    fn fail(_: &str, _: &'static str, _: Duration, _: Duration) {
+        panic!("fixture callback");
+    }
+    let conn = Mutex::new(Connection::open_in_memory().unwrap());
+    with_connection_trace(Some("trace"), fail, || {
+        let mut guard = lock_connection(&conn, &TracingStoreConnectionObserver, StoreDomain::Audit);
+        guard.set_wait_duration(SLOW_CONNECTION);
+        drop(guard);
+        assert!(conn.try_lock().is_ok());
+    });
+}

crates/webcodex-store/src/lib.rs

@@ -22,6 +22,7 @@ mod trace_query;
 pub use trace_query::{ToolTraceCallRecord, ToolTraceQueryFilter};
 mod communication;
 mod connection_observation;
+pub use connection_observation::with_connection_trace;
 mod history_budget;
 pub use history_budget::{HistoryReadBudget, HistoryReadDiagnostics};
 

crates/webcodex-tool-contracts/src/registry/output_schemas/discovery.rs

@@ -164,6 +164,8 @@ pub(super) fn output_schema_for_tool(name: &str) -> Option<Value> {
             ("trace_mode", schema_type("string", "Observed captured mode: metadata, full, or unknown.")),
             ("capture_mode", schema_type("string", "Current capture switch; separate from retained data.")),
             ("capture_health", open_object_schema("Process-since-start queue/budget/write failure counters; not per-trace completeness or prior-process history.")),
+            ("handler_finished_observed", schema_type("boolean", "The HTTP adapter scope finished after synchronous finalizers. Does not prove client receipt or streamed-body EOF.")),
+            ("known_dropped_events", schema_type("integer", "Lower bound from retained per-trace loss markers. Zero is not proof of complete capture.")),
             ("response_handoff_observed", schema_type("boolean", "A handler-return event exists in the retained index; absence is inconclusive and presence does not prove remote delivery.")),
             ("status", schema_type("string", "available or unavailable; missing capture does not prove absent execution.")),
             ("coverage", schema_type("string", "Coverage boundary; observed events are not a completeness guarantee.")),

docs/DEPLOYMENT.md

@@ -223,6 +223,8 @@ WEBCODEX_TOOL_REQUEST_TRACE_MAX_TOTAL_BYTES=2147483648
 
 `full` tracing can contain source text, patches, command/script input and output, user messages, and secrets that were themselves present inside captured tool payloads. Protect the trace directory like other sensitive diagnostic data and keep retention/budget bounded. Trace capture is diagnostic only; trace-write failure does not change tool execution.
 
+Slow Store connection waits/holds (100 ms or more) are reported at INFO with bounded diagnostic labels. Metadata traces distinguish response handoff from Handler completion after audit persistence. Trace capture is best effort: the diagnostic view marks known gaps, and a completion event alone does not prove a complete trace.
+
 The exact trace file layout, correlation ids, model-facing forensic reader, queueing, compression, and payload-validation rules are implementation/maintainer details and are intentionally omitted here.
 
 ### Public HTTPS

docs/DEPLOYMENT.zh-CN.md

@@ -198,6 +198,8 @@ WEBCODEX_TOOL_REQUEST_TRACE_MAX_TOTAL_BYTES=2147483648
 
 `full` 可能包含源码、patch、command/script 输入输出、user message，以及本身就出现在 tool payload 中的 secret。应像保护其它敏感诊断数据一样保护 trace 目录，并设置有界 retention/budget。Trace capture 只用于诊断；trace 写入失败不会改变 tool execution。
 
+Store 连接等待或持有达到 100 ms 时，会在 INFO 级别记录有界诊断标签。Metadata Trace 区分响应交接与审计持久化后的 Handler 收尾。Trace 尽力采集：诊断界面会标记已知遗漏，单独存在完成事件不代表整条 Trace 完整。
+
 精确的 trace 文件布局、correlation id、model-facing forensic reader、queue/compression 与 payload validation 规则属于 implementation/maintainer detail，有意不放在这份部署指南中。
 
 ### 公网 HTTPS

docs/implementation/bounded-meta-trace-diagnostics.md

@@ -154,3 +154,66 @@ OpenTelemetry pipeline, new Direct tool, alternate trace database or snapshot
 report subsystem is added. Per-Window/time/tool-scoped full arming is a separate
 operator-control feature. This patch solves persistent metadata evidence and
 on-demand discovery/inspection first; it cannot reconstruct unrecorded bodies.
+
+
+## 2026-10-27: request finalization and capture loss
+
+The preceding sections record the original delivery. The current implementation
+adds explicit request-tail evidence without changing ActionAudit's response
+handoff clock or Window continuity semantics.
+
+The MCP adapter owns the lifecycle. `request_received` starts when that adapter
+is entered (after upstream middleware), `request_parsed` follows ingress JSON and
+protocol validation, and `dispatch_finished` ends tool dispatch, not a detached
+Job's lifetime. `response_serialized` describes the constructed response and
+`handler_returned` retains its existing framework-response handoff boundary.
+The synchronous finalizer then persists ActionAudit and completes Window evidence;
+`audit_finalize` reports completed/failed/unavailable and `request_finalize`
+covers the whole finalizer. The lifecycle guard's final `handler_finished` records
+leaving the adapter after these synchronous operations. Cancelled or unwinding
+handlers instead retain `handler_incomplete_drop`. These are observations inside
+the adapter, not client receipt, body-stream EOF, network flush, or model reading.
+No await is permitted between response handoff and synchronous tail completion
+without revisiting this guard contract.
+
+Audit owns its explicit original `server_trace_id`; it does not consult the exited
+Tokio dispatch task-local. A bounded synchronous Store observation context carries
+that identity through audit persistence and restores nested/thread-local state on
+exit or unwind. Store still owns its connection Mutex and wait/hold measurement.
+After releasing the connection, a wait **or** hold of at least 100 ms emits an INFO
+`store_connection_slow` event, plus metadata when capture is enabled. Slow phase
+summaries also work at INFO. Fast calls retain the existing lightweight
+observations. The new fields contain only fixed domain/phase/outcome labels,
+durations and a bounded opaque trace ID; no SQL, parameters, tokens or bodies.
+Other Store callers without explicit context report the domain and timings with
+no invented request identity.
+
+Long historical Window/ActionAudit reads could occupy a shared SQLite connection
+Mutex, so even a cheap tool's required audit write waited behind unrelated history
+work. Separating and bounding those readers in #861/#862 reduced that contention;
+it did not make the remaining audit persistence instantaneous. The new tail span
+makes such residual waits visible while preserving the response boundary used by
+existing continuity consumers.
+
+The writer queue remains **64**, with nonblocking `try_send`. A single background
+writer serializes filesystem writes, compression, retention/budget maintenance
+and occasional accounting reconciliation. Sustained producer traffic or slow disk
+can exceed its service rate; increasing queue capacity alone only postpones loss.
+This change diagnoses loss, not a throughput guarantee: queue/budget/write failure
+counters remain process-wide, and a separate try-lock map coalesces known losses
+for at most **128** bounded trace IDs. The writer drains those markers outside the
+business request and before acknowledging a diagnostic flush. Its map and drained
+batch are bounded, with no recursive retries if marker persistence fails.
+
+Retained `tool_trace_capture_gap` events set reader/UI `coverage` to `incomplete`
+and expose a lower bound `known_dropped_events`. `capture_health.loss_marker_drops`
+also counts losses that could not be indexed or persisted. A missing marker,
+including after crash, disk failure or map overflow, never proves completeness:
+the alternative stays `observed_events_only`, even with `handler_finished` present.
+Missing whole traces remain `trace_not_retained`. Capture is not a durable journal.
+Queue saturation, bounded marker overflow and a blocked writer are tested without
+blocking an actual successful MCP request. Deterministic channels and the actual
+Store Mutex prove two concurrent audit waits keep separate trace identities,
+response handoff precedes audit, and Handler finish includes the wait. Additional
+regressions cover failed audits, cancellation, timeout, Runner disconnect and
+normal fast responses.