Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: retain safe Journal compaction failure diagnostics

Browse Source 
Keep stage/reason classification in the Journal Feature without changing frozen input, metadata or fallback behavior. Exercise connection, Host read failures, size budget and secret-bearing fixtures through compaction persistence.

Validation: old freeze function with new tests fails connection and snapshot/size diagnostics (3 failures); new code passes all six selected cases and cargo test -p worker. Serialization is currently unreachable for the derived JSON-safe snapshot fields; no artificial runtime failure is claimed.
This commit is contained in:
Hare
2026-10-21 05:32:07 +09:00
parent 4ef7b03651
commit 62e2b44305
5 changed files with 324 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -6558,6 +6558,7 @@ dependencies = [
"toml",
"tools",
"tracing",
"tracing-subscriber",
"tungstenite 0.28.0",
"uuid",
"wip-client",
crates/worker/Cargo.toml
+1
View File
@@ -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"
crates/worker/src/feature/builtin/journal.rs
+136 -16
View File
@@ -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<String> = (!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<Vec<JournalJobRequest>>,
crates/worker/tests/common/diagnostics.rs
+33
View File
@@ -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<Mutex<Vec<u8>>>);
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<usize> {
self.0.lock().unwrap().extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
crates/worker/tests/compact_events_test.rs
+153 -22
View File
@@ -26,6 +26,9 @@ use tokio::sync::broadcast;
use worker::{Worker, WorkerController, WorkerError};
#[path = "common/diagnostics.rs"]
mod diagnostics;
type TestStore = CombinedStore<FsStore, FsWorkerStore>;
#[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<subjektiv::journal::startup::JournalStartupSnapshot>,
reads: Mutex<Vec<worker::subjektiv::SubjektivHostContext>>,
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::<serde_json::Value>(json).unwrap()
);
assert_eq!(
payload["snapshot_sha256"],
Sha256::digest(json.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
);
} 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")),