diff --git a/Cargo.lock b/Cargo.lock index fd2babc6f..fa56fd6d7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6559,6 +6559,7 @@ dependencies = [ "toml", "tools", "tracing", + "tracing-subscriber", "tungstenite 0.28.0", "uuid", "wip-client", diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 9cdbc33c9..a96239ba3 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -58,6 +58,7 @@ wip-protocol = { workspace = true } wip-text-view = { workspace = true } [dev-dependencies] +tracing-subscriber.workspace = true axum = { workspace = true } workdir = { workspace = true, features = ["http-client"] } dotenv = "0.15.0" diff --git a/crates/worker/src/feature/builtin/journal.rs b/crates/worker/src/feature/builtin/journal.rs index e5de44038..7db8b5e35 100644 --- a/crates/worker/src/feature/builtin/journal.rs +++ b/crates/worker/src/feature/builtin/journal.rs @@ -59,6 +59,53 @@ impl JournalReadStatus { } } +// Only bounded classifications cross into diagnostics. Host messages/codes and +// io::Error payloads can contain Journal text, credentials, proofs or host paths. +// These diagnostics belong to the Feature, not persisted/model-visible input. +enum JournalCompactionFailure { + Connection(std::io::ErrorKind), + SnapshotConflict, + SnapshotInvalidArgument, + SnapshotUnavailable, + Serialization, + ByteBudget, +} + +impl JournalCompactionFailure { + fn snapshot(error: SubjektivHostError) -> Self { + match error { + SubjektivHostError::Conflict { .. } => Self::SnapshotConflict, + SubjektivHostError::InvalidArgument(_) => Self::SnapshotInvalidArgument, + SubjektivHostError::Unavailable(_) => Self::SnapshotUnavailable, + } + } + + fn stage(&self) -> &'static str { + match self { + Self::Connection(_) => "connection", + Self::SnapshotConflict | Self::SnapshotInvalidArgument | Self::SnapshotUnavailable => { + "snapshot" + } + Self::Serialization => "serialization", + Self::ByteBudget => "byte_budget", + } + } + + fn reason(&self) -> &'static str { + match self { + Self::Connection(std::io::ErrorKind::PermissionDenied) => "permission_denied", + Self::Connection(std::io::ErrorKind::InvalidInput) => "invalid_input", + Self::Connection(std::io::ErrorKind::NotConnected) => "not_connected", + Self::Connection(_) => "io_error", + Self::SnapshotConflict => "host_conflict", + Self::SnapshotInvalidArgument => "host_invalid_argument", + Self::SnapshotUnavailable => "host_unavailable", + Self::Serialization => "json_serialization_failed", + Self::ByteBudget => "startup_max_bytes_exceeded", + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct JournalCompactionInput { @@ -203,33 +250,46 @@ pub(crate) fn freeze_compaction_journal( .and_then(|connection| connection.host.workspace_id()) }); let read = match connection { - Ok(Some(connection)) if connection.host.journal_compaction_granted() => { - Some(connection.journal_compaction_context().map_err(|_| ())) - } + Ok(Some(connection)) if connection.host.journal_compaction_granted() => Some( + connection + .journal_compaction_context() + .map_err(JournalCompactionFailure::snapshot), + ), Ok(_) => None, - Err(_) => Some(Err(())), + Err(error) => Some(Err(JournalCompactionFailure::Connection(error.kind()))), }; let snapshot = read.as_ref().and_then(|read| read.as_ref().ok()).cloned(); - let (status, json) = match read { - None => (Disabled, String::new()), + let prepared = match read { + None => Ok((Disabled, String::new())), Some(Ok(snapshot)) => { let status = if snapshot.journals.is_empty() { Absent } else { Available }; - match serde_json::to_string(&snapshot) { - Ok(json) if json.len() <= subjektiv::journal::startup::STARTUP_MAX_BYTES => { - (status, json) - } - _ => (Unavailable, String::new()), - } + // The current snapshot has only derived, JSON-safe fields (strings, + // integers, vectors, enums and a UUID). There is no known serialization + // failure fixture; retain this guard if the domain type changes. + serde_json::to_string(&snapshot) + .map_err(|_| JournalCompactionFailure::Serialization) + .and_then(|json| { + if json.len() <= subjektiv::journal::startup::STARTUP_MAX_BYTES { + Ok((status, json)) + } else { + Err(JournalCompactionFailure::ByteBudget) + } + }) } - Some(Err(())) => (Unavailable, String::new()), + Some(Err(failure)) => Err(failure), }; - if status == Unavailable { - tracing::warn!("Journal compaction snapshot unavailable; continuing without Journal"); - } + let (status, json) = prepared.unwrap_or_else(|failure| { + tracing::warn!( + stage = failure.stage(), + reason = failure.reason(), + "Journal compaction snapshot unavailable; continuing without Journal" + ); + (Unavailable, String::new()) + }); use sha2::{Digest, Sha256}; let snapshot_sha256: Option = (!json.is_empty()).then(|| { Sha256::digest(json.as_bytes()) @@ -726,12 +786,72 @@ fn tool_schema(name: &str) -> Value { } } +#[cfg(test)] +#[path = "../../../tests/common/diagnostics.rs"] +mod diagnostic_test_support; + #[cfg(test)] mod tests { use super::*; use std::sync::Mutex; use subjektiv::journal::JournalReceipt; + #[test] + fn compaction_connection_failure_keeps_safe_reason_without_exposing_error_or_changing_input() { + let prompts = crate::prompt::catalog::PromptCatalog::builtins_only().unwrap(); + for (kind, reason) in [ + (std::io::ErrorKind::PermissionDenied, "permission_denied"), + (std::io::ErrorKind::InvalidInput, "invalid_input"), + (std::io::ErrorKind::NotConnected, "not_connected"), + (std::io::ErrorKind::Other, "io_error"), + ] { + let logs = diagnostic_test_support::CapturedWarnings::default(); + let input = tracing::subscriber::with_default(logs.subscriber(), || { + freeze_compaction_journal( + Err(std::io::Error::new( + kind, + "PRIVATE_JOURNAL_BODY /private/host/path proof=PRIVATE_PROOF credential=PRIVATE_CREDENTIAL raw_error=PRIVATE_ERROR", + )), + &prompts, + ) + .unwrap() + }); + let log = logs.text(); + assert!(log.contains("stage=\"connection\""), "{log}"); + assert!(log.contains(&format!("reason=\"{reason}\"")), "{log}"); + assert_eq!(log.matches("continuing without Journal").count(), 1); + assert!(!log.contains("PRIVATE_")); + assert!(!log.contains("/private/host/path")); + assert!(input.body().contains("Journal read status: unavailable.")); + let persisted = serde_json::to_string(&input).unwrap(); + assert!(!persisted.contains("PRIVATE_")); + assert!(!persisted.contains("/private/host/path")); + assert!(!persisted.contains(reason)); + let payload: JournalCompactionInput = input + .metadata() + .feature_payload() + .unwrap() + .decode("journal:compaction", 1) + .unwrap(); + assert!(payload.snapshot.is_none()); + assert!(payload.snapshot_sha256.is_none()); + assert_eq!( + payload.prompt_provenance.source_digest, + prompts.projection().source_digest + ); + assert_eq!( + payload.prompt_provenance.projection_digest, + prompts.projection().catalog_digest + ); + } + let logs = diagnostic_test_support::CapturedWarnings::default(); + let input = tracing::subscriber::with_default(logs.subscriber(), || { + freeze_compaction_journal(Ok(None), &prompts).unwrap() + }); + assert!(input.body().contains("Journal read status: disabled.")); + assert!(logs.text().is_empty()); + } + #[derive(Default)] struct Host { requests: Mutex>, diff --git a/crates/worker/tests/common/diagnostics.rs b/crates/worker/tests/common/diagnostics.rs new file mode 100644 index 000000000..d5768f599 --- /dev/null +++ b/crates/worker/tests/common/diagnostics.rs @@ -0,0 +1,33 @@ +use std::io::{self, Write}; +use std::sync::{Arc, Mutex}; + +/// Per-test warning capture; never installs a global subscriber or writes fixture secrets. +#[derive(Clone, Default)] +pub struct CapturedWarnings(Arc>>); + +impl CapturedWarnings { + pub fn subscriber(&self) -> impl tracing::Subscriber + Send + Sync + 'static { + let output = self.clone(); + tracing_subscriber::fmt() + .with_writer(move || output.clone()) + .without_time() + .with_ansi(false) + .with_max_level(tracing::Level::WARN) + .finish() + } + + pub fn text(&self) -> String { + String::from_utf8(self.0.lock().unwrap().clone()).unwrap() + } +} + +impl Write for CapturedWarnings { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 0869c0b5c..c4de54e11 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -26,6 +26,9 @@ use tokio::sync::broadcast; use worker::{Worker, WorkerController, WorkerError}; +#[path = "common/diagnostics.rs"] +mod diagnostics; + type TestStore = CombinedStore; #[derive(Clone)] @@ -2876,7 +2879,7 @@ const COMPACTION_HOST_WORKSPACE_ID: &str = "journal-transport-host-workspace"; // Host authorization/selection or live-model semantic continuation quality. struct CompactionJournalHost { granted: bool, - unavailable: bool, + failure: Option<&'static str>, snapshot: Mutex, reads: Mutex>, startup_reads: AtomicUsize, @@ -2886,7 +2889,7 @@ impl CompactionJournalHost { fn new(snapshot: subjektiv::journal::startup::JournalStartupSnapshot) -> Self { Self { granted: true, - unavailable: false, + failure: None, snapshot: Mutex::new(snapshot), reads: Mutex::new(vec![]), startup_reads: AtomicUsize::new(0), @@ -2917,10 +2920,18 @@ impl worker::subjektiv::SubjektivHost for CompactionJournalHost { worker::subjektiv::SubjektivHostError, > { self.reads.lock().unwrap().push(context.clone()); - if self.unavailable { - return Err(worker::subjektiv::SubjektivHostError::Unavailable( - "injected Journal read failure".into(), - )); + if let Some(failure) = self.failure { + use worker::subjektiv::SubjektivHostError; + let secret = "PRIVATE_JOURNAL_BODY /private/host/path proof=PRIVATE_PROOF credential=PRIVATE_CREDENTIAL raw_error=PRIVATE_ERROR"; + return Err(match failure { + "unavailable" => SubjektivHostError::Unavailable(secret.into()), + "invalid-argument" => SubjektivHostError::InvalidArgument(secret.into()), + "conflict" => SubjektivHostError::Conflict { + code: secret.into(), + message: secret.into(), + }, + _ => panic!("unknown failure fixture"), + }); } Ok(self.snapshot.lock().unwrap().clone()) } @@ -3466,26 +3477,37 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai } #[tokio::test] -async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversation_compaction() { - for case in [ - "absent", - "disabled", - "unavailable", - "over-byte-budget", - "no-host", - ] { +async fn journal_empty_disabled_and_available_compaction_preserve_inputs_without_failure_warnings() +{ + assert_journal_compaction_cases(&["absent", "available", "disabled", "no-host"]).await; +} + +#[tokio::test] +async fn journal_snapshot_failures_keep_safe_diagnostics_and_continue_compaction() { + assert_journal_compaction_cases(&["unavailable", "invalid-argument", "conflict"]).await; +} + +#[tokio::test] +async fn journal_over_byte_budget_keeps_safe_diagnostic_and_continues_compaction() { + assert_journal_compaction_cases(&["over-byte-budget"]).await; +} + +async fn assert_journal_compaction_cases(cases: &[&'static str]) { + for &case in cases { let mut snapshot = compaction_snapshot(0); if case == "absent" { snapshot.journals.clear(); } if case == "over-byte-budget" { snapshot.journals[0].current_position = - "x".repeat(subjektiv::journal::startup::STARTUP_MAX_BYTES); + "PRIVATE_JOURNAL_BODY".repeat(subjektiv::journal::startup::STARTUP_MAX_BYTES); } - let expected_json = (case == "absent").then(|| serde_json::to_string(&snapshot).unwrap()); + let expected_json = matches!(case, "absent" | "available") + .then(|| serde_json::to_string(&snapshot).unwrap()); let mut host = CompactionJournalHost::new(snapshot); host.granted = case != "disabled"; - host.unavailable = case == "unavailable"; + host.failure = + matches!(case, "unavailable" | "invalid-argument" | "conflict").then_some(case); let host = Arc::new(host); let client = MockClient::new(vec![ single_text_events("seed"), @@ -3517,7 +3539,51 @@ async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversat true, ) .await; - fixture.worker.compact(0).await.unwrap(); + // Scope the subscriber to every poll, not a thread-local guard across await. + // This observes the real Feature warning while compacting into FsStore. + let logs = diagnostics::CapturedWarnings::default(); + use tracing::instrument::WithSubscriber; + fixture + .worker + .compact(0) + .with_subscriber(logs.subscriber()) + .await + .unwrap(); + let log = logs.text(); + let failure = match case { + "unavailable" => Some(("snapshot", "host_unavailable")), + "invalid-argument" => Some(("snapshot", "host_invalid_argument")), + "conflict" => Some(("snapshot", "host_conflict")), + "over-byte-budget" => Some(("byte_budget", "startup_max_bytes_exceeded")), + _ => None, + }; + if let Some((stage, reason)) = failure { + assert!(log.contains(&format!("stage=\"{stage}\"")), "{case}: {log}"); + assert!( + log.contains(&format!("reason=\"{reason}\"")), + "{case}: {log}" + ); + assert_eq!( + log.matches("continuing without Journal").count(), + 1, + "{case}" + ); + } else { + assert!( + !log.contains("Journal compaction snapshot unavailable"), + "{case}: {log}" + ); + } + for secret in [ + "PRIVATE_", + "/private/host/path", + "JOURNAL_POSITION_0", + "JOURNAL_BODY_0", + "state-token-0", + "fixed-event-reference-0", + ] { + assert!(!log.contains(secret), "{case}: diagnostic leaked {secret}"); + } let input = first_compactor_input(&observed.requests()[1]); let body = if matches!(case, "disabled" | "no-host") { assert_eq!( @@ -3532,8 +3598,8 @@ async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversat String::new() } else { let body = journal_body(&fixture.worker); - let expected = if case == "absent" { - "absent" + let expected = if matches!(case, "absent" | "available") { + case } else { "unavailable" }; @@ -3548,6 +3614,71 @@ async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversat &fixture.worker.prompts().load_full(), Some(COMPACTION_HOST_WORKSPACE_ID), ); + let seed = replacement_seed(&fixture.worker); + let metadata = seed[1].metadata.input_metadata.as_ref().unwrap(); + assert_eq!( + metadata.usage().recording_material, + session_store::InputEvidenceUse::Exclude + ); + assert_eq!( + metadata.usage().memory_evidence, + session_store::InputEvidenceUse::Exclude + ); + assert_eq!( + metadata.usage().observation, + session_store::InputObservationUse::Hidden + ); + let payload: serde_json::Value = metadata + .feature_payload() + .unwrap() + .decode("journal:compaction", 1) + .unwrap(); + assert_eq!(payload["status"], expected); + assert_eq!( + payload["prompt_provenance"]["workspace_id"], + COMPACTION_HOST_WORKSPACE_ID + ); + let projection = fixture.worker.prompts().load_full().projection().clone(); + assert_eq!( + payload["prompt_provenance"]["source_digest"], + projection.source_digest + ); + assert_eq!( + payload["prompt_provenance"]["projection_digest"], + projection.catalog_digest + ); + if let Some(json) = expected_json.as_deref() { + use sha2::{Digest, Sha256}; + assert_eq!( + payload["snapshot"], + serde_json::from_str::(json).unwrap() + ); + assert_eq!( + payload["snapshot_sha256"], + Sha256::digest(json.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::() + ); + } else { + assert!(payload["snapshot"].is_null()); + assert!(payload["snapshot_sha256"].is_null()); + } + // Feature diagnostics do not become model input or a new metadata contract. + let saved = serde_json::to_string(&seed[1]).unwrap(); + for forbidden in [ + "PRIVATE_", + "/private/host/path", + "host_unavailable", + "host_conflict", + "host_invalid_argument", + "startup_max_bytes_exceeded", + ] { + assert!( + !saved.contains(forbidden), + "{case}: saved input leaked {forbidden}" + ); + } body }; assert_full_initial_summary(&input, &inherited); @@ -3559,8 +3690,8 @@ async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversat .is_some_and(|text| text.contains("PREVIOUS_JOURNAL_DELEGATED_CONTEXT"))), "fallback retains only the newly frozen reference, not an old snapshot as latest" ); - assert!(!body.contains("JOURNAL_POSITION_0")); - assert!(!body.contains("journal-reference-0")); + assert_eq!(body.contains("JOURNAL_POSITION_0"), case == "available"); + assert_eq!(body.contains("journal-reference-0"), case == "available"); assert_eq!( host.reads.lock().unwrap().len(), usize::from(!matches!(case, "disabled" | "no-host")),