From 18f934f2f9ba8d3115e5a8adf081eaebb142f069 Mon Sep 17 00:00:00 2001 From: dp-web4 Date: Wed, 12 Aug 2026 19:36:50 -0700 Subject: [PATCH 1/2] hub F0.1 (R7a): degraded-event recording + conduct-vs-infra delta classification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements issue 700 / PRD_HUB_V2_FEDERATED R7a — infrastructure failure can never score as member conduct, and the reputation seam opens staged. web4-core: additive DeltaClass {conduct, infra, unclassified} on ReputationDelta, serde-default Unclassified (the sovereign_strength fail-closed precedent): a pre-classification delta can never fabricate Conduct. hub-lib: DegradedLog — local append-only JSONL diagnostic log, the fallback witness for windows in which the signer (the normal witness) is the unreachable thing; snapshot/commit two-phase reconciliation (a failed witness duplicates auditablely, never loses). SwappableSigner records Transport and locked-shell refusals at its choke point (policy denials from a live vault are NOT recorded). EmitMode {classify_only (default), apply} on ReputationEmitPolicy — the staged seam-opening in law. ReputationRecorded gains applied (decided at record time from mode x class, honored on replay; fold re-checks class as belt-and-braces). DegradedReconciled event binds drained log bytes by digest. HubState: ReputationIngest + DegradedCounters for the observation window. hub-daemon: both witness sites (record_reputation verb, obligation sweep) decide applied at record time; ignition reconciles the degraded log into the witnessed ledger (snapshot -> witness -> commit); admin renders class/applied and the reconciliation event. Intentional behavior changes: (1) sovereign record_reputation ingest is now also staged (previously folded unconditionally; production impact nil — the seam has never been open); (2) the obligation sweep holds its Conduct deltas until law ratifies mode: apply — the e2e test now asserts the staged semantics. Tests: 417 hub workspace + 200 web4-core + 52 trust-core, green. New: signer choke-point discrimination (transport/locked recorded, policy denial not), class-gate fold (only applied Conduct reaches tensors; adversarial applied-Infra held), snapshot/commit reconciliation, EmitMode serde defaults. GitNexus impact analysis: mandated by CLAUDE.md but GitNexus is not registered on this machine (classified per fleet memory); blast radius enumerated by compiler instead (workspace + tests both directions). surface: record_reputation act: mutate governed (subject,role) reputation S: med/reversible [construct: ReputationDelta fold] R: pass [construct: reputation_emit_decision, Pin #1 authenticated caller] W: pass [construct: law rules match authenticated emitter] O: pass [construct: law decision + applied computed before witness_event] A: pass [construct: ReputationRecorded{delta,applied} single witnessed entry] V: present [construct: EmitMode::ClassifyOnly default + escalate decision arm] verdict: PASS Co-Authored-By: Claude Fable 5 --- hub/docs/HUB-LAW.md | 1 + hub/hub-daemon/src/admin.rs | 16 +- hub/hub-daemon/src/rest.rs | 111 +++++++++++-- hub/hub-lib/src/degraded.rs | 300 ++++++++++++++++++++++++++++++++++++ hub/hub-lib/src/events.rs | 31 +++- hub/hub-lib/src/law.rs | 67 ++++++++ hub/hub-lib/src/lib.rs | 1 + hub/hub-lib/src/signer.rs | 166 +++++++++++++++++++- hub/hub-lib/src/state.rs | 190 ++++++++++++++++++++--- web4-core/src/r6.rs | 38 +++++ 10 files changed, 881 insertions(+), 40 deletions(-) create mode 100644 hub/hub-lib/src/degraded.rs diff --git a/hub/docs/HUB-LAW.md b/hub/docs/HUB-LAW.md index 964d0d9e..f6f62e7d 100644 --- a/hub/docs/HUB-LAW.md +++ b/hub/docs/HUB-LAW.md @@ -79,6 +79,7 @@ lockstep with `HubEvent::kind()` by a test that reads the function's own source. | pairing + channels | `pairing_requested`, `pairing_confirmed`, `pairing_revoked`, `pair_message_posted`, `intro_requested`, `intro_responded` | | devices | `device_enrolled`, `device_revoked` | | obligations + trust | `obligation_opened`, `obligation_resolved`, `reputation_recorded` | +| degraded-window audit | `degraded_reconciled` (F0.1: the witnessed summary of infrastructure-failure windows — recorded by the hub at ignition, never member conduct) | | vault unlock (audit) | `vault_unlock_requested`, `vault_unlock_attested`, `vault_unlock_resolved` | | records | `event_recorded`, `lct_published`, `referenced_act` | diff --git a/hub/hub-daemon/src/admin.rs b/hub/hub-daemon/src/admin.rs index 88703a21..735e3c44 100644 --- a/hub/hub-daemon/src/admin.rs +++ b/hub/hub-daemon/src/admin.rs @@ -910,13 +910,25 @@ fn event_summary(event: &HubEvent) -> String { html_escape(&act.substance.uri), ) } - HubEvent::ReputationRecorded { delta } => format!( + HubEvent::DegradedReconciled { count, first_ts, last_ts, by_source, .. } => format!( + "🛠 degraded window reconciled — {} infrastructure event(s), \ + {} → {}, by source: {}", + count, + first_ts.format("%Y-%m-%d %H:%M"), + last_ts.format("%Y-%m-%d %H:%M"), + html_escape( + &by_source.iter().map(|(k, v)| format!("{k}×{v}")).collect::>().join(", ") + ), + ), + HubEvent::ReputationRecorded { delta, applied } => format!( "📊 reputation Δ for {} in role {} \ - (ΔT3 {:+.3}, ΔV3 {:+.3}) — {}", + (ΔT3 {:+.3}, ΔV3 {:+.3}, class {:?}, {}) — {}", html_escape(&delta.subject_lct), html_escape(&delta.role_lct), delta.net_trust_change(), delta.net_value_change(), + delta.class, + if *applied { "applied" } else { "recorded-only" }, html_escape(&delta.reason), ), HubEvent::ObligationOpened { request_id, subject_lct, role_lct, due_at, criticality, .. } => format!( diff --git a/hub/hub-daemon/src/rest.rs b/hub/hub-daemon/src/rest.rs index e56b9eca..ef0a087d 100644 --- a/hub/hub-daemon/src/rest.rs +++ b/hub/hub-daemon/src/rest.rs @@ -104,6 +104,11 @@ pub struct RestState { /// process); HestiaCallbackSigner for Hestia-mode chapters (hub holds NO /// keys; signs via Hestia HTTP callback); LockedSigner while sealed. pub signer: Arc, + /// F0.1 (R7a): the degraded-event diagnostic log — the fallback witness + /// for windows in which the signer (the normal witness) is the + /// unreachable thing. The `SwappableSigner` appends to it at its choke + /// point; ignition reconciles it into the witnessed ledger. + pub degraded_log: Arc, pub ledger: Arc>, /// Incremental projection cache. The ledger is append-only, so a cached /// `HubState` can be advanced by folding only entries appended since it was @@ -376,12 +381,20 @@ impl RestState { } let _ = locked; // lock state now derives from the installed signer kind + // F0.1 (R7a): one degraded log per hub root; the swappable signer + // records infrastructure failures to it at its choke point. + let degraded_log = Arc::new(hub_lib::degraded::DegradedLog::new( + paths.root.join("degraded.jsonl"), + )); + let signer_sw = Arc::new(SwappableSigner::new(signer)); + signer_sw.set_degraded_log(degraded_log.clone()); Ok(Self { paths, hub_id: society.lct_id, hub_name: society.name.clone(), sovereign_lct_id, - signer: Arc::new(SwappableSigner::new(signer)), + signer: signer_sw, + degraded_log, ledger, state_cache: Arc::new(std::sync::Mutex::new(ProjectionCache::default())), nonces: Arc::new(NonceStore::new()), @@ -543,12 +556,21 @@ impl RestState { ) })?; let signer: Arc = Arc::new(LockedSigner::new(pid.founding_sovereign_lct_id)); + // F0.1 (R7a): the locked shell is exactly the boot most likely to + // produce degraded events — attach the log so locked refusals are + // recorded and reconciled at ignition. + let degraded_log = Arc::new(hub_lib::degraded::DegradedLog::new( + paths.root.join("degraded.jsonl"), + )); + let signer_sw = Arc::new(SwappableSigner::new(signer)); + signer_sw.set_degraded_log(degraded_log.clone()); Ok(Self { paths, hub_id: pid.hub_id, hub_name: pid.hub_name, sovereign_lct_id: pid.founding_sovereign_lct_id, - signer: Arc::new(SwappableSigner::new(signer)), + signer: signer_sw, + degraded_log, ledger: placeholder_ledger, state_cache: Arc::new(std::sync::Mutex::new(ProjectionCache::default())), nonces: Arc::new(NonceStore::new()), @@ -841,6 +863,37 @@ impl RestState { // the durable mailbox opens with the same now-live key. self.hydrate_mailbox().await; self.unlock_gate.record_success(); + // F0.1 (R7a): the signer is live again — reconcile the degraded + // diagnostic log into the witnessed ledger. The local log was + // the fallback witness for the window in which the signer (the + // normal witness) was the unreachable thing; this entry binds + // its exact drained bytes by digest. Non-fatal on failure: the + // drain fails toward duplication, never loss, and ignition + // must not be blocked by its own bookkeeping. + if let Some(snap) = self.degraded_log.snapshot() { + let count = snap.summary.count; + let ev = HubEvent::DegradedReconciled { + count, + first_ts: snap.summary.first_ts, + last_ts: snap.summary.last_ts, + entries_digest: snap.summary.entries_digest.clone(), + by_source: snap.summary.by_source.clone(), + }; + match witness_event(self, ev).await { + Ok(idx) => { + // Witnessed — only now remove the reconciled bytes. + self.degraded_log.commit_reconciled(&snap); + tracing::info!( + "ignition: reconciled {count} degraded event(s) into ledger entry {idx}" + ); + } + Err(e) => tracing::warn!( + "ignition: degraded-log reconciliation failed (non-fatal; \ + entries re-report on next ignition): {}", + e.message + ), + } + } // Refresh the clear public-identity tier-0 file so a future locked-shell // boot knows who this hub is (to serve well-known + accept `hub unlock`). let pid = PublicIdentity { @@ -3757,6 +3810,10 @@ fn temporal_delta( // Hub-internal obligation-outcome delta: the hub does not hardware-attest // the subject's sovereign here, so it stays at the fail-closed default. sovereign_strength: web4_core::r6::SovereignStrength::default(), + // Adjudicated by the hub from witnessed timestamps — conduct evidence. + // (F0.1/R7a: an infrastructure-caused miss is the emitter's to classify + // as `Infra`; the deadline sweep judges only what the ledger witnessed.) + class: web4_core::r6::DeltaClass::Conduct, action_type: action_kind.to_string(), action_target: "hub".to_string(), action_id: action_id.to_string(), @@ -3947,8 +4004,27 @@ async fn dispatch_channel( } } } - let index = witness_event(s, HubEvent::ReputationRecorded { delta }).await?; - Ok(serde_json::json!({ "recorded": true, "entry_index": index })) + // F0.1 (R7a): decide application AT RECORD TIME from the law's + // staged mode × the delta's class, and capture it into the event — + // replay honors the law in force when the delta landed. Fail-closed + // on every arm: no law / no section / classify_only mode / non- + // Conduct class ⇒ recorded, applied to nothing. + let applied = { + use hub_lib::law::{EmitMode, HubLawExt}; + let law_guard = s.law.read().await; + let mode = law_guard.as_ref() + .map(|law| law.reputation_emit_mode()) + .unwrap_or_default(); + mode == EmitMode::Apply && delta.class == web4_core::r6::DeltaClass::Conduct + }; + let class = delta.class; + let index = witness_event(s, HubEvent::ReputationRecorded { delta, applied }).await?; + Ok(serde_json::json!({ + "recorded": true, + "entry_index": index, + "applied": applied, + "class": class, + })) } "list_members" => { let all: Vec = state.members.values() @@ -4273,7 +4349,19 @@ async fn dispatch_channel( _ => "met", }; let delta = temporal_delta(&ob, &request_id, outcome_str, &impact, index, ¬ice_kind, now); - witness_event(s, HubEvent::ReputationRecorded { delta }).await?; + // F0.1 (R7a): the obligation sweep obeys the same staged + // mode as every other emitter — hub-adjudicated Conduct + // applies only when the law's mode says apply. + let applied = { + use hub_lib::law::{EmitMode, HubLawExt}; + let law_guard = s.law.read().await; + let mode = law_guard.as_ref() + .map(|law| law.reputation_emit_mode()) + .unwrap_or_default(); + mode == EmitMode::Apply + && delta.class == web4_core::r6::DeltaClass::Conduct + }; + witness_event(s, HubEvent::ReputationRecorded { delta, applied }).await?; witness_event(s, HubEvent::ObligationResolved { request_id: request_id.clone(), outcome: outcome_str.to_string(), @@ -9139,12 +9227,17 @@ pub(crate) mod channel_e2e_tests { assert_eq!(out2["outcome"], serde_json::json!("met")); assert!(out2["t3_temperament"].as_f64().unwrap() > 0.0, "on-time = positive temperament"); - // Projected: obligation cleared, reputation folded at (subject, role). + // Projected: obligation cleared. F0.1 (R7a): with no `reputation_emit` + // section the mode is CLASSIFY-ONLY — the hub-adjudicated Conduct delta + // is WITNESSED to the ledger but folds into no tensor until the law + // ratifies `mode: apply` (the staged seam opening). The observation + // window reads the held counter instead. let proj2 = { let l = state.ledger.lock().await; HubState::project(&l) }; assert!(!proj2.obligations.contains_key("req-1"), "resolved obligation removed"); - let rep = proj2.reputation.get(&(lct.to_string(), "citizen".to_string())) - .expect("reputation folded for (subject, citizen)"); - assert_eq!(rep.observations, 1); + assert!(proj2.reputation.is_empty(), + "classify-only mode holds the delta out of the tensors"); + assert_eq!(proj2.reputation_ingest.held_classify_only, 1, + "the delta was witnessed and counted, not lost"); } #[tokio::test] diff --git a/hub/hub-lib/src/degraded.rs b/hub/hub-lib/src/degraded.rs new file mode 100644 index 00000000..d4286231 --- /dev/null +++ b/hub/hub-lib/src/degraded.rs @@ -0,0 +1,300 @@ +//! Degraded-event recording — Sprint F0.1 / PRD_HUB_V2_FEDERATED R7a. +//! +//! When the hub's signing/referee dependencies fail, that fact must be +//! **recorded**, append-only, and **distinguishable from member conduct** +//! (the measured fleet failure this prevents: infrastructure fail-closed +//! denies scored as conduct — hestia PR #357 class). The recorder cannot be +//! the witnessed ledger itself, because the signer is frequently the +//! unreachable thing: a witnessed `DegradedRecorded` event would need the +//! very signature that just failed. So the mechanism is the fleet's ratified +//! pattern (hestia PRD_GATE_CONSOLIDATION criterion 9c): a **local +//! append-only JSONL diagnostic log** — the fallback witness precisely when +//! the normal witness is the unreachable thing — reconciled into one +//! witnessed [`crate::events::HubEvent::DegradedReconciled`] entry when +//! signing capability returns (at unlock / signer recovery). +//! +//! Design properties: +//! - **Best-effort, never load-bearing for serving**: an `append` that cannot +//! write logs an error and returns; it never panics and never blocks the +//! request path on anything slower than a local file append. +//! - **Append-only**: the file is opened `O_APPEND`; `drain` (reconciliation) +//! is the only truncation, and it happens under the same lock as appends, +//! after the drained bytes have been captured and digested. +//! - **No secrets**: `context` is a bounded operator-facing string (operation +//! + error text), never key material or payload bytes. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use std::io::Write; +use std::path::{Path, PathBuf}; +use web4_core::crypto::sha256_hex; + +/// Why the hub was degraded for one call. Every variant is an infrastructure +/// fact by construction — there is deliberately no conduct-shaped variant. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum DegradedSource { + /// The vault is sealed (`LockedSigner`) — a signing act was refused + /// because the hub has no signing capability, not because policy said no. + LockedRefusal, + /// The remote signer (hestia callback / HSM) did not answer (transport). + SignerUnreachable, + /// A federation peer did not answer. Forward slot for R1; unused until + /// federation edges exist. + PeerUnreachable, + /// A law-gate / referee round-trip timed out. + GateTimeout, +} + +impl DegradedSource { + /// Stable snake_case token used in reconciliation summaries. + pub fn token(&self) -> &'static str { + match self { + DegradedSource::LockedRefusal => "locked_refusal", + DegradedSource::SignerUnreachable => "signer_unreachable", + DegradedSource::PeerUnreachable => "peer_unreachable", + DegradedSource::GateTimeout => "gate_timeout", + } + } +} + +/// One degraded event, as persisted to the diagnostic log (one JSON object +/// per line). +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct DegradedEntry { + pub source: DegradedSource, + pub ts: DateTime, + /// Bounded, operator-facing context: what was attempted and what failed. + pub context: String, +} + +/// Maximum stored context length — a transport error carrying a huge body +/// must not bloat the diagnostic log. +const MAX_CONTEXT: usize = 400; + +/// The result of draining the log for reconciliation: everything a +/// `DegradedReconciled` ledger event needs. +#[derive(Clone, Debug)] +pub struct DrainSummary { + pub count: u32, + pub first_ts: DateTime, + pub last_ts: DateTime, + /// SHA-256 hex over the exact drained JSONL bytes, binding the local log + /// content to the witnessed reconciliation event. + pub entries_digest: String, + /// Per-source counts, keyed by [`DegradedSource::token`]. + pub by_source: BTreeMap, +} + +/// Append-only JSONL diagnostic log for degraded events. +pub struct DegradedLog { + path: PathBuf, + lock: std::sync::Mutex<()>, +} + +impl DegradedLog { + /// A log at `path`. The file is created lazily on first append; a missing + /// file reads as empty. + pub fn new(path: impl Into) -> Self { + Self { path: path.into(), lock: std::sync::Mutex::new(()) } + } + + pub fn path(&self) -> &Path { + &self.path + } + + /// Record one degraded event. Best-effort: failures to write are logged + /// via `tracing::error!` and swallowed — the serving path never fails + /// because its diagnostic log did. + pub fn append(&self, source: DegradedSource, context: &str) { + let mut ctx = context.to_string(); + if ctx.len() > MAX_CONTEXT { + ctx.truncate(MAX_CONTEXT); + ctx.push_str("…"); + } + let entry = DegradedEntry { source, ts: Utc::now(), context: ctx }; + let line = match serde_json::to_string(&entry) { + Ok(l) => l, + Err(e) => { + tracing::error!("degraded-log serialize failed: {e}"); + return; + } + }; + let _guard = self.lock.lock().expect("degraded log lock poisoned"); + let res = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&self.path) + .and_then(|mut f| writeln!(f, "{line}")); + if let Err(e) = res { + tracing::error!("degraded-log append failed at {}: {e}", self.path.display()); + } + } + + /// Read all entries without consuming them. Malformed lines are skipped + /// (counted via `tracing::warn!`), never fatal: a torn final line from a + /// crash must not brick reconciliation. + pub fn read_all(&self) -> Vec { + let _guard = self.lock.lock().expect("degraded log lock poisoned"); + self.read_unlocked().0 + } + + fn read_unlocked(&self) -> (Vec, Vec) { + let bytes = std::fs::read(&self.path).unwrap_or_default(); + let mut out = Vec::new(); + for line in bytes.split(|b| *b == b'\n') { + if line.is_empty() { + continue; + } + match serde_json::from_slice::(line) { + Ok(e) => out.push(e), + Err(e) => tracing::warn!("degraded-log skipping malformed line: {e}"), + } + } + (out, bytes) + } + + /// Number of readable entries (operator surface). + pub fn len(&self) -> usize { + self.read_all().len() + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Phase 1 of reconciliation: capture + digest the current contents + /// WITHOUT consuming them. Returns `None` when there is nothing to + /// reconcile. The caller witnesses the summary to the ledger and, only on + /// success, calls [`Self::commit_reconciled`] — so a failed witness loses + /// nothing (the entries re-report on the next snapshot, with the same + /// digest, which is auditable duplication rather than silent loss). + pub fn snapshot(&self) -> Option { + let _guard = self.lock.lock().expect("degraded log lock poisoned"); + let (entries, bytes) = self.read_unlocked(); + if entries.is_empty() { + return None; + } + let digest = sha256_hex(&bytes); + let mut by_source: BTreeMap = BTreeMap::new(); + for e in &entries { + *by_source.entry(e.source.token().to_string()).or_insert(0) += 1; + } + Some(Snapshot { + summary: DrainSummary { + count: entries.len() as u32, + first_ts: entries.first().expect("non-empty").ts, + last_ts: entries.last().expect("non-empty").ts, + entries_digest: digest, + by_source, + }, + byte_len: bytes.len(), + }) + } + + /// Phase 2: remove exactly the bytes the snapshot summarized (now + /// witnessed), preserving any entries appended since the snapshot was + /// taken. Runs under the append lock, so nothing lands between the read + /// and the rewrite. + pub fn commit_reconciled(&self, snap: &Snapshot) { + let _guard = self.lock.lock().expect("degraded log lock poisoned"); + let bytes = std::fs::read(&self.path).unwrap_or_default(); + let tail: &[u8] = if bytes.len() >= snap.byte_len { + &bytes[snap.byte_len..] + } else { + // The file shrank underneath us (external interference). The + // witnessed summary stands; start clean rather than guess. + tracing::warn!( + "degraded-log at {} shorter than reconciled snapshot — clearing", + self.path.display() + ); + b"" + }; + if let Err(e) = std::fs::write(&self.path, tail) { + tracing::error!("degraded-log commit failed at {}: {e}", self.path.display()); + } + } +} + +/// A captured, digested view of the log pending witness — see +/// [`DegradedLog::snapshot`] / [`DegradedLog::commit_reconciled`]. +#[derive(Clone, Debug)] +pub struct Snapshot { + pub summary: DrainSummary, + byte_len: usize, +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmp_log(name: &str) -> DegradedLog { + let dir = std::env::temp_dir().join("hub-degraded-tests"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join(format!("{name}-{}.jsonl", uuid::Uuid::new_v4())); + DegradedLog::new(path) + } + + #[test] + fn append_read_roundtrip() { + let log = tmp_log("roundtrip"); + log.append(DegradedSource::SignerUnreachable, "sign member_admit: connect refused"); + log.append(DegradedSource::LockedRefusal, "sign law_amend: vault locked"); + let entries = log.read_all(); + assert_eq!(entries.len(), 2); + assert_eq!(entries[0].source, DegradedSource::SignerUnreachable); + assert_eq!(entries[1].source, DegradedSource::LockedRefusal); + } + + #[test] + fn snapshot_then_commit_reconciles() { + let log = tmp_log("snapshot"); + log.append(DegradedSource::SignerUnreachable, "a"); + log.append(DegradedSource::SignerUnreachable, "b"); + log.append(DegradedSource::GateTimeout, "c"); + let snap = log.snapshot().expect("snapshot"); + assert_eq!(snap.summary.count, 3); + assert_eq!(snap.summary.by_source.get("signer_unreachable"), Some(&2)); + assert_eq!(snap.summary.by_source.get("gate_timeout"), Some(&1)); + assert_eq!(snap.summary.entries_digest.len(), 64); + // Snapshot does NOT consume — a failed witness loses nothing. + assert_eq!(log.read_all().len(), 3); + log.commit_reconciled(&snap); + assert!(log.is_empty(), "commit must remove the reconciled bytes"); + assert!(log.snapshot().is_none(), "empty log has nothing to reconcile"); + } + + #[test] + fn commit_preserves_entries_appended_after_snapshot() { + let log = tmp_log("tail"); + log.append(DegradedSource::LockedRefusal, "before"); + let snap = log.snapshot().expect("snapshot"); + log.append(DegradedSource::GateTimeout, "after"); + log.commit_reconciled(&snap); + let remaining = log.read_all(); + assert_eq!(remaining.len(), 1, "post-snapshot entry must survive commit"); + assert_eq!(remaining[0].source, DegradedSource::GateTimeout); + } + + #[test] + fn malformed_line_is_skipped_not_fatal() { + let log = tmp_log("malformed"); + log.append(DegradedSource::LockedRefusal, "good"); + { + let mut f = std::fs::OpenOptions::new().append(true).open(log.path()).unwrap(); + writeln!(f, "{{torn line").unwrap(); + } + log.append(DegradedSource::GateTimeout, "also good"); + let entries = log.read_all(); + assert_eq!(entries.len(), 2, "torn line skipped, valid lines kept"); + } + + #[test] + fn context_is_bounded() { + let log = tmp_log("bounded"); + log.append(DegradedSource::SignerUnreachable, &"x".repeat(10_000)); + let entries = log.read_all(); + assert!(entries[0].context.len() <= MAX_CONTEXT + 4); + } +} diff --git a/hub/hub-lib/src/events.rs b/hub/hub-lib/src/events.rs index 0415ea03..cd19660c 100644 --- a/hub/hub-lib/src/events.rs +++ b/hub/hub-lib/src/events.rs @@ -510,7 +510,34 @@ pub enum HubEvent { /// society-law weights); the hub records + applies it, never invents the math. /// Reputation is NEVER global — the delta's `(subject_lct, role_lct)` scopes /// it to an MRH role-pairing link (RFC #403). - ReputationRecorded { delta: web4_core::r6::ReputationDelta }, + ReputationRecorded { + delta: web4_core::r6::ReputationDelta, + /// F0.1 (R7a): whether this delta was APPLIED to the `(subject, role)` + /// tensors, decided **at record time** from the law's `EmitMode` and + /// the delta's `DeltaClass` (`applied = mode == Apply && class == + /// Conduct`). Captured into the event so replay honors the law in + /// force when the delta was recorded — never re-evaluated against a + /// later law. Fail-closed default `false`: an event from before this + /// field existed replays as recorded-not-applied. + #[serde(default)] + applied: bool, + }, + + /// F0.1 (R7a): reconciliation of the local degraded-event diagnostic log + /// into the witnessed ledger. Appended when signing capability returns + /// (at unlock / signer recovery) — the diagnostic log is the fallback + /// witness for the window in which the normal witness (the signer) was + /// the unreachable thing, and this entry binds its exact drained bytes + /// by digest so the two records are auditable against each other. + DegradedReconciled { + count: u32, + first_ts: DateTime, + last_ts: DateTime, + /// SHA-256 hex over the drained JSONL bytes. + entries_digest: String, + /// Per-source counts, keyed by `DegradedSource::token()`. + by_source: std::collections::BTreeMap, + }, /// §5.1 R7 carrier: an accountability *obligation* opened by a coordination /// act that carried an `r7` block with a deadline. The subject committed to @@ -620,6 +647,7 @@ impl HubEvent { "council_member_added", "council_member_removed", "council_threshold_changed", + "degraded_reconciled", "device_enrolled", "device_revoked", "event_recorded", @@ -678,6 +706,7 @@ impl HubEvent { Self::IntroResponded { .. } => "intro_responded", Self::MemberProfileUpdated { .. } => "member_profile_updated", Self::LawAmended { .. } => "law_amended", + Self::DegradedReconciled { .. } => "degraded_reconciled", Self::CouncilMemberAdded { .. } => "council_member_added", Self::CouncilMemberRemoved { .. } => "council_member_removed", Self::CouncilThresholdChanged { .. } => "council_threshold_changed", diff --git a/hub/hub-lib/src/law.rs b/hub/hub-lib/src/law.rs index c1330ddc..bf89eaaf 100644 --- a/hub/hub-lib/src/law.rs +++ b/hub/hub-lib/src/law.rs @@ -202,6 +202,23 @@ pub struct ReputationEmitRule { pub description: Option, } +/// How authorized reputation deltas are APPLIED — the staged seam-opening +/// (Sprint F0.1 / PRD R7a, decision recorded in `SPRINTS.md` 2026-08-13). +/// +/// - `classify_only` (fail-closed default): authorized deltas are witnessed to +/// the ledger with their conduct-vs-infra class, visible on the operator +/// surface, and applied to **no** tensor. The observation-window mode. +/// - `apply`: authorized `Conduct`-class deltas fold into `(subject, role)` +/// tensors. `Infra` and `Unclassified` deltas are recorded, never applied, +/// in **every** mode. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum EmitMode { + #[default] + ClassifyOnly, + Apply, +} + /// The hub's `reputation_emit` law section — who, other than the Sovereign, may /// record reputation deltas, and for which subject-roles. Absent section ⇒ the /// emit path is fully dark (Sovereign-only, the pre-wiring behavior). @@ -209,6 +226,10 @@ pub struct ReputationEmitRule { pub struct ReputationEmitPolicy { #[serde(default, skip_serializing_if = "Vec::is_empty")] pub rules: Vec, + /// Staged application mode (F0.1). Absent ⇒ `classify_only` — a law that + /// opens the seam without stating a mode observes before it actuates. + #[serde(default)] + pub mode: EmitMode, } /// A `subject` token that references `constellation:` — the v2 attestation @@ -414,6 +435,11 @@ pub trait HubLawExt { /// governs only non-Sovereign emitters. fn reputation_emit_decision(&self, emitter: &str, delta_role_lct: &str) -> ReputationEmitOutcome; + /// F0.1 (R7a): the staged application mode of the `reputation_emit` + /// section. Fail-closed: no section ⇒ `ClassifyOnly` — with the seam dark + /// or unstated, nothing applies to tensors. + fn reputation_emit_mode(&self) -> EmitMode; + /// Norms that test `r6.request.action` for equality against a value the gate /// can never produce. /// @@ -475,6 +501,10 @@ impl HubLawExt for Law { out } + fn reputation_emit_mode(&self) -> EmitMode { + self.ext.reputation_emit.as_ref().map(|re| re.mode).unwrap_or_default() + } + fn reputation_emit_decision(&self, emitter: &str, delta_role_lct: &str) -> ReputationEmitOutcome { let deny = || ReputationEmitOutcome { decision: Decision::Deny, matched_rule: None }; let Some(re) = self.ext.reputation_emit.as_ref() else { @@ -1679,3 +1709,40 @@ norms: assert!(checked >= 3, "only {checked} block(s) were actually normalised and checked"); } } + +#[cfg(test)] +mod emit_mode_tests { + use super::*; + + #[test] + fn mode_defaults_to_classify_only() { + // F0.1: a `reputation_emit` section that says nothing about mode + // observes before it actuates — and no section at all is the same. + let yaml = r#" +rules: + - emitter: "hestia-gate" + subject: "citizen" + decision: allow + priority: 10 +"#; + let policy: ReputationEmitPolicy = serde_yaml::from_str(yaml).expect("parses"); + assert_eq!(policy.mode, EmitMode::ClassifyOnly); + } + + #[test] + fn mode_apply_parses_and_round_trips() { + let yaml = r#" +mode: apply +rules: + - emitter: "hestia-gate" + subject: "citizen" + decision: allow + priority: 10 +"#; + let policy: ReputationEmitPolicy = serde_yaml::from_str(yaml).expect("parses"); + assert_eq!(policy.mode, EmitMode::Apply); + let back = serde_yaml::to_string(&policy).expect("serializes"); + let again: ReputationEmitPolicy = serde_yaml::from_str(&back).expect("round-trips"); + assert_eq!(again.mode, EmitMode::Apply); + } +} diff --git a/hub/hub-lib/src/lib.rs b/hub/hub-lib/src/lib.rs index 34793854..7733c78c 100644 --- a/hub/hub-lib/src/lib.rs +++ b/hub/hub-lib/src/lib.rs @@ -27,6 +27,7 @@ pub mod atomic_file; pub mod build_info; pub mod charter; pub mod constellation; +pub mod degraded; pub mod envelope; pub mod events; pub mod identity; diff --git a/hub/hub-lib/src/signer.rs b/hub/hub-lib/src/signer.rs index b34d521d..fa189291 100644 --- a/hub/hub-lib/src/signer.rs +++ b/hub/hub-lib/src/signer.rs @@ -432,11 +432,23 @@ impl RemoteSigner for LockedSigner { /// of the currently-installed inner signer (`signer_kind() == Locked`). pub struct SwappableSigner { inner: std::sync::RwLock>, + /// F0.1 (R7a): the degraded-event diagnostic log. Every signer failure + /// that is an *infrastructure* fact — the vault is locked, or the remote + /// signer's transport failed — is recorded here, at this single choke + /// point all signing paths flow through, so no caller has to remember to. + /// Policy denials (`SignError::Denied` from an unlocked vault) are NOT + /// degraded events and are not recorded here. + degraded: std::sync::OnceLock>, } impl SwappableSigner { pub fn new(initial: Arc) -> Self { - Self { inner: std::sync::RwLock::new(initial) } + Self { inner: std::sync::RwLock::new(initial), degraded: std::sync::OnceLock::new() } + } + /// Attach the degraded-event diagnostic log (once, at boot). Absent a log + /// the signer behaves exactly as before — recording is additive. + pub fn set_degraded_log(&self, log: Arc) { + let _ = self.degraded.set(log); } /// Promote (or demote) the backing signer. Used by the unlock slot to swap /// a `LockedSigner` for the real `LocalKeypairSigner` once ignited. @@ -446,6 +458,23 @@ impl SwappableSigner { fn current(&self) -> Arc { self.inner.read().expect("swappable signer lock poisoned").clone() } + /// Classify a signer failure as a degraded (infrastructure) event and + /// record it; conduct-shaped outcomes (policy denials) pass through + /// unrecorded. `locked` is the *backing signer's* kind at call time — + /// a `Denied` from a locked shell is a capability refusal, not policy. + fn record_if_degraded(&self, op: &str, locked: bool, err: &SignError) { + let Some(log) = self.degraded.get() else { return }; + use crate::degraded::DegradedSource; + match err { + SignError::Transport(msg) => { + log.append(DegradedSource::SignerUnreachable, &format!("{op}: {msg}")); + } + SignError::Denied(msg) if locked => { + log.append(DegradedSource::LockedRefusal, &format!("{op}: {msg}")); + } + _ => {} + } + } } #[async_trait] @@ -458,7 +487,12 @@ impl RemoteSigner for SwappableSigner { ) -> std::result::Result { // Snapshot the Arc, then drop the lock before the await point. let cur = self.current(); - cur.sign(actor_lct_id, signing_bytes, intent).await + let locked = cur.signer_kind() == SignerKind::Locked; + let res = cur.sign(actor_lct_id, signing_bytes, intent).await; + if let Err(e) = &res { + self.record_if_degraded(&format!("sign:{}", intent.event_kind), locked, e); + } + res } fn signer_kind(&self) -> SignerKind { self.current().signer_kind() @@ -468,11 +502,23 @@ impl RemoteSigner for SwappableSigner { } fn channel_seal(&self, peer: &PublicKey, pair_id: Uuid, plaintext: &[u8]) -> std::result::Result { - self.current().channel_seal(peer, pair_id, plaintext) + let cur = self.current(); + let locked = cur.signer_kind() == SignerKind::Locked; + let res = cur.channel_seal(peer, pair_id, plaintext); + if let Err(e) = &res { + self.record_if_degraded("channel_seal", locked, e); + } + res } fn channel_open(&self, peer: &PublicKey, pair_id: Uuid, sealed_b64: &str) -> std::result::Result, SignError> { - self.current().channel_open(peer, pair_id, sealed_b64) + let cur = self.current(); + let locked = cur.signer_kind() == SignerKind::Locked; + let res = cur.channel_open(peer, pair_id, sealed_b64); + if let Err(e) = &res { + self.record_if_degraded("channel_open", locked, e); + } + res } } @@ -773,3 +819,115 @@ mod tests { } } } + +#[cfg(test)] +mod degraded_recording_tests { + use super::*; + use crate::degraded::{DegradedLog, DegradedSource}; + + /// A remote signer whose transport always fails — the unit-level stand-in + /// for "the hestia callback was killed mid-session" (criterion 7). + struct DeadTransportSigner; + #[async_trait] + impl RemoteSigner for DeadTransportSigner { + async fn sign(&self, _a: Uuid, _b: &[u8], _i: &SignIntent) + -> std::result::Result { + Err(SignError::Transport("connect refused (test)".into())) + } + fn signer_kind(&self) -> SignerKind { SignerKind::HestiaCallback } + fn public_key(&self) -> Option { None } + fn channel_seal(&self, _p: &PublicKey, _id: Uuid, _pt: &[u8]) + -> std::result::Result { + Err(SignError::Transport("connect refused (test)".into())) + } + fn channel_open(&self, _p: &PublicKey, _id: Uuid, _s: &str) + -> std::result::Result, SignError> { + Err(SignError::Transport("connect refused (test)".into())) + } + } + + /// An UNLOCKED signer that denies on policy — a conduct-shaped refusal + /// that must NOT be recorded as degraded. + struct PolicyDenyingSigner; + #[async_trait] + impl RemoteSigner for PolicyDenyingSigner { + async fn sign(&self, _a: Uuid, _b: &[u8], _i: &SignIntent) + -> std::result::Result { + Err(SignError::Denied("vault policy said no (test)".into())) + } + fn signer_kind(&self) -> SignerKind { SignerKind::HestiaCallback } + fn public_key(&self) -> Option { None } + fn channel_seal(&self, _p: &PublicKey, _id: Uuid, _pt: &[u8]) + -> std::result::Result { + Err(SignError::Denied("no (test)".into())) + } + fn channel_open(&self, _p: &PublicKey, _id: Uuid, _s: &str) + -> std::result::Result, SignError> { + Err(SignError::Denied("no (test)".into())) + } + } + + fn intent(kind: &str) -> SignIntent { + SignIntent { + request_id: Uuid::new_v4(), + hub_id: Uuid::new_v4(), + hub_name: "Test".into(), + actor_lct_id: Uuid::new_v4(), + ledger_index: 1, + event_kind: kind.into(), + event: serde_json::json!({}), + } + } + + fn tmp_log(name: &str) -> Arc { + let dir = std::env::temp_dir().join("hub-signer-degraded-tests"); + std::fs::create_dir_all(&dir).unwrap(); + Arc::new(DegradedLog::new(dir.join(format!("{name}-{}.jsonl", Uuid::new_v4())))) + } + + #[tokio::test] + async fn transport_failure_records_signer_unreachable() { + // F0.1 criterion 7, unit form: the callback dies mid-session → every + // failed signing act lands in the degraded log as infrastructure. + let log = tmp_log("transport"); + let sw = SwappableSigner::new(Arc::new(DeadTransportSigner)); + sw.set_degraded_log(log.clone()); + let err = sw.sign(Uuid::new_v4(), b"bytes", &intent("member_added")).await.unwrap_err(); + assert!(matches!(err, SignError::Transport(_)), "error passes through unchanged"); + let entries = log.read_all(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].source, DegradedSource::SignerUnreachable); + assert!(entries[0].context.contains("sign:member_added")); + } + + #[tokio::test] + async fn locked_refusal_records_locked_not_policy() { + let log = tmp_log("locked"); + let sw = SwappableSigner::new(Arc::new(LockedSigner::new(Uuid::new_v4()))); + sw.set_degraded_log(log.clone()); + let _ = sw.sign(Uuid::new_v4(), b"bytes", &intent("law_amended")).await.unwrap_err(); + let entries = log.read_all(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].source, DegradedSource::LockedRefusal); + } + + #[tokio::test] + async fn policy_denial_from_unlocked_signer_is_not_degraded() { + // The discrimination criterion: a conduct-shaped policy denial from a + // live vault is NOT an infrastructure event and must not be recorded. + let log = tmp_log("policy"); + let sw = SwappableSigner::new(Arc::new(PolicyDenyingSigner)); + sw.set_degraded_log(log.clone()); + let _ = sw.sign(Uuid::new_v4(), b"bytes", &intent("member_added")).await.unwrap_err(); + assert!(log.is_empty(), "policy denial must leave no degraded entry"); + } + + #[tokio::test] + async fn recording_is_additive_absent_a_log() { + // No log attached → identical behavior to pre-F0.1 (recording is + // strictly additive). + let sw = SwappableSigner::new(Arc::new(DeadTransportSigner)); + let err = sw.sign(Uuid::new_v4(), b"bytes", &intent("x")).await.unwrap_err(); + assert!(matches!(err, SignError::Transport(_))); + } +} diff --git a/hub/hub-lib/src/state.rs b/hub/hub-lib/src/state.rs index b6568e7c..6c6c0a05 100644 --- a/hub/hub-lib/src/state.rs +++ b/hub/hub-lib/src/state.rs @@ -154,6 +154,15 @@ pub struct HubState { #[serde(serialize_with = "serialize_reputation")] pub reputation: BTreeMap<(String, String), RoleReputation>, + /// F0.1 (R7a): ingest accounting for the reputation seam — how many deltas + /// were applied vs recorded-only, by why. The observation-window review + /// (staged seam opening) reads these before `EmitMode::Apply` is ratified. + pub reputation_ingest: ReputationIngest, + + /// F0.1 (R7a): degraded-window accounting folded from `DegradedReconciled` + /// events — the witnessed summary of the local diagnostic log. + pub degraded: DegradedCounters, + /// §5.1 R7 carrier: open accountability obligations, keyed by `request_id`. /// An `ObligationOpened` inserts; `ObligationResolved` removes. String-keyed, /// so serialization-safe. The timeout sweep (P2) reads this to debit misses. @@ -199,6 +208,33 @@ pub struct Obligation { pub opened_at: DateTime, } +/// F0.1 (R7a): reputation-seam ingest counters. `applied` folded into tensors; +/// the three `held_*` buckets were witnessed to the ledger but applied to +/// nothing — split by why, so the observation window can distinguish "law says +/// classify-only" from "the delta itself was infra/unclassified". +#[derive(Clone, Debug, Default, Serialize)] +pub struct ReputationIngest { + pub applied: u64, + /// Conduct-class deltas recorded while the law's mode was `classify_only`. + pub held_classify_only: u64, + /// Infra-class deltas — never applied in any mode. + pub held_infra: u64, + /// Unclassified deltas — held for review, never applied in any mode. + pub held_unclassified: u64, +} + +/// F0.1 (R7a): counters folded from witnessed `DegradedReconciled` events. +#[derive(Clone, Debug, Default, Serialize)] +pub struct DegradedCounters { + /// Reconciliation events witnessed. + pub reconciliations: u64, + /// Total degraded entries those reconciliations summarized. + pub entries: u64, + /// Per-source totals, keyed by `DegradedSource::token()`. + pub by_source: BTreeMap, + pub last_reconciled: Option>, +} + /// Accumulated role-contextualized reputation for one `(subject, role)` pairing. /// Folded from `ReputationRecorded` deltas via `T3/V3::apply_delta` — the hub /// applies the math but never invents it (weights are a society-law hook). @@ -832,28 +868,51 @@ impl HubState { // R7 reputation: fold the delta into the (subject, role) tensors. The // hub applies the delta (via T3/V3::apply_delta); the *weights* that // produced it are a society-law hook, computed by the recorder. - HubEvent::ReputationRecorded { delta } => { - let rep = self - .reputation - .entry((delta.subject_lct.clone(), delta.role_lct.clone())) - .or_insert_with(|| RoleReputation::new(ts)); - for (dim, td) in &delta.t3_delta { - if let Some(d) = trust_dim(dim) { - rep.t3.apply_delta(d, td.change); + HubEvent::ReputationRecorded { delta, applied } => { + use web4_core::r6::DeltaClass; + // F0.1 (R7a): the class gate. `applied` was decided at record + // time (law mode × class) and is honored on replay — but the + // fold re-checks the class as a belt-and-braces invariant: an + // event claiming `applied` for a non-Conduct delta (corrupt or + // adversarial) folds NOTHING. Infra never scores as conduct. + if *applied && delta.class == DeltaClass::Conduct { + let rep = self + .reputation + .entry((delta.subject_lct.clone(), delta.role_lct.clone())) + .or_insert_with(|| RoleReputation::new(ts)); + for (dim, td) in &delta.t3_delta { + if let Some(d) = trust_dim(dim) { + rep.t3.apply_delta(d, td.change); + } } - } - for (dim, td) in &delta.v3_delta { - if let Some(d) = value_dim(dim) { - rep.v3.apply_delta(d, td.change); + for (dim, td) in &delta.v3_delta { + if let Some(d) = value_dim(dim) { + rep.v3.apply_delta(d, td.change); + } + } + // Fail-closed provenance: the bucket is only as strong as the + // weakest delta that fed it. A placeholder-attested delta pins the + // bucket to `placeholder` and no later hardware delta can upgrade + // the placeholder-era observations back to `hardware`. + rep.sovereign_strength = rep.sovereign_strength.min(delta.sovereign_strength); + rep.observations += 1; + rep.last_updated = ts; + self.reputation_ingest.applied += 1; + } else { + match delta.class { + DeltaClass::Conduct => self.reputation_ingest.held_classify_only += 1, + DeltaClass::Infra => self.reputation_ingest.held_infra += 1, + DeltaClass::Unclassified => self.reputation_ingest.held_unclassified += 1, } } - // Fail-closed provenance: the bucket is only as strong as the - // weakest delta that fed it. A placeholder-attested delta pins the - // bucket to `placeholder` and no later hardware delta can upgrade - // the placeholder-era observations back to `hardware`. - rep.sovereign_strength = rep.sovereign_strength.min(delta.sovereign_strength); - rep.observations += 1; - rep.last_updated = ts; + } + HubEvent::DegradedReconciled { count, last_ts, by_source, .. } => { + self.degraded.reconciliations += 1; + self.degraded.entries += u64::from(*count); + for (src, n) in by_source { + *self.degraded.by_source.entry(src.clone()).or_insert(0) += u64::from(*n); + } + self.degraded.last_reconciled = Some(*last_ts); } HubEvent::ObligationOpened { request_id, @@ -1240,6 +1299,7 @@ mod tests { subject_lct: subject.clone(), role_lct: "citizen".to_string(), sovereign_strength: web4_core::r6::SovereignStrength::Hardware, + class: web4_core::r6::DeltaClass::Conduct, action_type: "handoff".into(), action_target: "hub".into(), action_id: "act-1".into(), @@ -1256,7 +1316,7 @@ mod tests { hub_name: "Test".into(), charter_hash: "sha256:0".into(), founding_sovereign_lct_id: sov.lct.id, created_at: Utc::now(), }), - (sov.lct.id, &kp, HubEvent::ReputationRecorded { delta }), + (sov.lct.id, &kp, HubEvent::ReputationRecorded { delta, applied: true }), ]).await; let state = HubState::project(&ledger); let rep = state.reputation.get(&(subject.clone(), "citizen".to_string())) @@ -1289,6 +1349,7 @@ mod tests { subject_lct: subject.clone(), role_lct: "role:constellation:mesh-worker".to_string(), sovereign_strength: strength, + class: web4_core::r6::DeltaClass::Conduct, action_type: "act".into(), action_target: "hub".into(), action_id: id.into(), rule_triggered: "r".into(), reason: "x".into(), t3_delta: t3, v3_delta: HashMap::new(), @@ -1301,9 +1362,9 @@ mod tests { founding_sovereign_lct_id: sov.lct.id, created_at: Utc::now(), }), // hardware first, then a placeholder delta, then hardware again. - (sov.lct.id, &kp, HubEvent::ReputationRecorded { delta: mk(SovereignStrength::Hardware, "a1") }), - (sov.lct.id, &kp, HubEvent::ReputationRecorded { delta: mk(SovereignStrength::Placeholder, "a2") }), - (sov.lct.id, &kp, HubEvent::ReputationRecorded { delta: mk(SovereignStrength::Hardware, "a3") }), + (sov.lct.id, &kp, HubEvent::ReputationRecorded { applied: true, delta: mk(SovereignStrength::Hardware, "a1") }), + (sov.lct.id, &kp, HubEvent::ReputationRecorded { applied: true, delta: mk(SovereignStrength::Placeholder, "a2") }), + (sov.lct.id, &kp, HubEvent::ReputationRecorded { applied: true, delta: mk(SovereignStrength::Hardware, "a3") }), ]).await; let state = HubState::project(&ledger); let rep = state.reputation @@ -1330,6 +1391,7 @@ mod tests { subject_lct: subject.clone(), role_lct: "citizen".to_string(), sovereign_strength: Default::default(), + class: web4_core::r6::DeltaClass::Conduct, action_type: "handoff".into(), action_target: "hub".into(), action_id: "a".into(), rule_triggered: "r".into(), reason: "x".into(), t3_delta, v3_delta: HashMap::new(), @@ -1340,7 +1402,7 @@ mod tests { hub_name: "Test".into(), charter_hash: "sha256:0".into(), founding_sovereign_lct_id: sov.lct.id, created_at: Utc::now(), }), - (sov.lct.id, &kp, HubEvent::ReputationRecorded { delta }), + (sov.lct.id, &kp, HubEvent::ReputationRecorded { delta, applied: true }), ]).await; let state = HubState::project(&ledger); let json = serde_json::to_value(&state).expect("HubState must serialize"); @@ -1351,6 +1413,86 @@ mod tests { assert_eq!(arr[0]["observations"], serde_json::json!(1)); } + #[tokio::test] + async fn class_gate_holds_everything_but_applied_conduct() { + // F0.1 (R7a) criterion 7, fold form: of four witnessed deltas, ONLY + // the applied Conduct delta reaches the tensors. The classify-only + // Conduct, the Infra (even adversarially marked applied), and the + // Unclassified are each recorded, counted, and folded into NOTHING. + use std::collections::HashMap; + use web4_core::r6::DeltaClass; + let sov = IdentityFile::generate(EntityType::Human); + let kp = sov.keypair().unwrap(); + let subject = Uuid::new_v4().to_string(); + let mk = |class: DeltaClass, id: &str| { + let mut t3_delta = HashMap::new(); + t3_delta.insert("talent".to_string(), + web4_core::r6::TensorDelta { change: 0.1, from_value: 0.0, to_value: 0.1 }); + web4_core::r6::ReputationDelta { + subject_lct: subject.clone(), + role_lct: "citizen".to_string(), + sovereign_strength: Default::default(), + class, + action_type: "act".into(), action_target: "hub".into(), action_id: id.into(), + rule_triggered: "r".into(), reason: "t".into(), + t3_delta, v3_delta: HashMap::new(), + contributing_factors: vec![], witnesses: vec![], timestamp: Utc::now(), + } + }; + let (_tmp, ledger) = make_ledger_with(vec![ + (sov.lct.id, &kp, HubEvent::Genesis { + hub_name: "Test".into(), charter_hash: "sha256:0".into(), + founding_sovereign_lct_id: sov.lct.id, created_at: Utc::now(), + }), + // applied Conduct — the only one that may fold + (sov.lct.id, &kp, HubEvent::ReputationRecorded { + delta: mk(DeltaClass::Conduct, "a1"), applied: true }), + // Conduct held by classify-only mode + (sov.lct.id, &kp, HubEvent::ReputationRecorded { + delta: mk(DeltaClass::Conduct, "a2"), applied: false }), + // Infra ADVERSARIALLY marked applied — the belt-and-braces gate + (sov.lct.id, &kp, HubEvent::ReputationRecorded { + delta: mk(DeltaClass::Infra, "a3"), applied: true }), + // Unclassified (pre-F0.1 emitter) — held + (sov.lct.id, &kp, HubEvent::ReputationRecorded { + delta: mk(DeltaClass::Unclassified, "a4"), applied: false }), + ]).await; + let state = HubState::project(&ledger); + let rep = state.reputation.get(&(subject.clone(), "citizen".to_string())) + .expect("the applied Conduct delta folded"); + assert_eq!(rep.observations, 1, "exactly ONE delta reached the tensors"); + assert_eq!(state.reputation_ingest.applied, 1); + assert_eq!(state.reputation_ingest.held_classify_only, 1); + assert_eq!(state.reputation_ingest.held_infra, 1, + "infra never scores as conduct, even marked applied"); + assert_eq!(state.reputation_ingest.held_unclassified, 1); + } + + #[tokio::test] + async fn degraded_reconciled_folds_counters_only() { + let sov = IdentityFile::generate(EntityType::Human); + let kp = sov.keypair().unwrap(); + let mut by_source = std::collections::BTreeMap::new(); + by_source.insert("signer_unreachable".to_string(), 2u32); + by_source.insert("locked_refusal".to_string(), 1u32); + let now = Utc::now(); + let (_tmp, ledger) = make_ledger_with(vec![ + (sov.lct.id, &kp, HubEvent::Genesis { + hub_name: "Test".into(), charter_hash: "sha256:0".into(), + founding_sovereign_lct_id: sov.lct.id, created_at: Utc::now(), + }), + (sov.lct.id, &kp, HubEvent::DegradedReconciled { + count: 3, first_ts: now, last_ts: now, + entries_digest: "d".repeat(64), by_source, + }), + ]).await; + let state = HubState::project(&ledger); + assert_eq!(state.degraded.reconciliations, 1); + assert_eq!(state.degraded.entries, 3); + assert_eq!(state.degraded.by_source.get("signer_unreachable"), Some(&2)); + assert!(state.reputation.is_empty(), "degraded windows touch no tensor"); + } + #[tokio::test] async fn member_added_and_removed_round_trip() { let sov = IdentityFile::generate(EntityType::Human); diff --git a/web4-core/src/r6.rs b/web4-core/src/r6.rs index 9952cb3b..f26b9b32 100644 --- a/web4-core/src/r6.rs +++ b/web4-core/src/r6.rs @@ -249,6 +249,33 @@ impl Default for SovereignStrength { } } +/// Classification of what a [`ReputationDelta`] evidences — member **conduct**, +/// or an **infrastructure** condition that is not the member's doing (timeout, +/// locked vault, unreachable referee or peer). The measured fleet failure this +/// prevents (hestia PR #357 class): infrastructure fail-closed denies scored as +/// member conduct. The class rides the delta so the scoring side can never +/// conflate the two. Fail-closed: absent ⇒ `Unclassified`, which is **held** — +/// recorded, surfaced for review, never applied to tensors — so a +/// pre-classification delta can never fabricate `Conduct`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum DeltaClass { + /// Evidences member conduct — eligible to apply to `(subject, role)` tensors. + Conduct, + /// Evidences an infrastructure condition. Recorded for audit; never applied. + Infra, + /// Not yet classified (e.g., emitted before this field existed). Held: + /// recorded, not applied, surfaced for review. + Unclassified, +} + +impl Default for DeltaClass { + fn default() -> Self { + // Fail-closed: an unstated class is held, never scored as conduct. + DeltaClass::Unclassified + } +} + /// R7's first-class reputation output — explicit trust/value delta per action. /// /// Key: reputation is ROLE-CONTEXTUALIZED. The `role_lct` field determines @@ -264,6 +291,12 @@ pub struct ReputationDelta { /// as `Placeholder` (the fail-closed default), never fabricating `Hardware`. #[serde(default)] pub sovereign_strength: SovereignStrength, + /// What this delta evidences: member conduct vs an infrastructure condition + /// (see [`DeltaClass`]). Backward-compatible: deltas emitted before this + /// field existed deserialize as `Unclassified` (held, never applied to + /// tensors), never fabricating `Conduct`. + #[serde(default)] + pub class: DeltaClass, /// What action caused the change pub action_type: String, /// Target of the action @@ -488,6 +521,11 @@ impl R7Action { subject_lct: self.role.actor_lct.clone(), role_lct: self.role.role_lct.clone(), sovereign_strength: SovereignStrength::default(), + // The R6 engine scores an ADJUDICATED action outcome — conduct + // evidence by construction. An emitter that knows the outcome was + // infrastructure-caused (transport failure, referee timeout) + // overrides this to `Infra` before recording. + class: DeltaClass::Conduct, action_type: self.request.action.clone(), action_target: self.request.target.clone(), action_id: self.action_id.clone(), From 63e3b47ca7e6f26a9e54597327cac026d5bd76f3 Mon Sep 17 00:00:00 2001 From: dp-web4 Date: Thu, 13 Aug 2026 10:30:00 -0700 Subject: [PATCH 2/2] hub F0.1 fixes: UTF-8 truncation panic + opportunistic degraded reconciliation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses both review findings on PR 703. BLOCKER (reproduced, exit=101): `String::truncate` panics when the split lands mid-codepoint, and `append`'s input is `format!("{op}: {msg}")` over a `SignError::Transport` payload — a peer error body or non-ASCII hostname. One leading ASCII byte shifts every following multibyte char across byte 400. The panic fired inside the signer's error path, i.e. exactly during the degraded window this module exists to record: the record lost AND the caller unwinding 500 instead of receiving its SignError, inverting the module's own invariant. Fix: walk down to a char boundary before truncating. The regression test uses a straddling multibyte payload (`format!("a{}", "é".repeat(300))` — the exact shape from the review) plus 3-byte and 4-byte cases; the old all-ASCII bound test could not see this because every ASCII byte is a boundary. Verified the new test FAILS against the unfixed code (panic at the truncate line) before keeping the fix — a guard not run against what it guards proves nothing. REVIEW POINT 2: reconciliation was ignition-only, so a dependency recovering on a live hub left its window unwitnessed until the next restart — and hubs run for weeks between ignitions. Reconciliation is now opportunistic: it fires at ignition AND after any successful signed append while entries are pending. Guards: a `dirty` atomic so the hot path pays one load when nothing is degraded; a `reconciling` slot (released via Drop, so an early return or panic cannot strand it) collapsing concurrent attempts; and the recursion is broken STRUCTURALLY via `witness_event_inner` rather than relying on the runtime guard. Both signing paths are hooked — `witness_event` and `commit_pair_event` — since a trigger on one signing path and not its sibling would leave whole classes of recovery unreconciled (the admission path signs directly, which is what the new test caught). Tests: 419 hub workspace green. New: straddling-multibyte truncation, and live-recovery reconciliation asserting the window folds in with NO ignition and does not re-witness an empty window on later acts. Co-Authored-By: Claude Fable 5 --- hub/hub-daemon/src/rest.rs | 168 ++++++++++++++++++++++++++++-------- hub/hub-lib/src/degraded.rs | 111 ++++++++++++++++++++++-- 2 files changed, 239 insertions(+), 40 deletions(-) diff --git a/hub/hub-daemon/src/rest.rs b/hub/hub-daemon/src/rest.rs index ef0a087d..25e0ca1b 100644 --- a/hub/hub-daemon/src/rest.rs +++ b/hub/hub-daemon/src/rest.rs @@ -864,36 +864,8 @@ impl RestState { self.hydrate_mailbox().await; self.unlock_gate.record_success(); // F0.1 (R7a): the signer is live again — reconcile the degraded - // diagnostic log into the witnessed ledger. The local log was - // the fallback witness for the window in which the signer (the - // normal witness) was the unreachable thing; this entry binds - // its exact drained bytes by digest. Non-fatal on failure: the - // drain fails toward duplication, never loss, and ignition - // must not be blocked by its own bookkeeping. - if let Some(snap) = self.degraded_log.snapshot() { - let count = snap.summary.count; - let ev = HubEvent::DegradedReconciled { - count, - first_ts: snap.summary.first_ts, - last_ts: snap.summary.last_ts, - entries_digest: snap.summary.entries_digest.clone(), - by_source: snap.summary.by_source.clone(), - }; - match witness_event(self, ev).await { - Ok(idx) => { - // Witnessed — only now remove the reconciled bytes. - self.degraded_log.commit_reconciled(&snap); - tracing::info!( - "ignition: reconciled {count} degraded event(s) into ledger entry {idx}" - ); - } - Err(e) => tracing::warn!( - "ignition: degraded-log reconciliation failed (non-fatal; \ - entries re-report on next ignition): {}", - e.message - ), - } - } + // diagnostic log into the witnessed ledger. + reconcile_degraded_log(self, "ignition").await; // Refresh the clear public-identity tier-0 file so a future locked-shell // boot knows who this hub is (to serve well-known + accept `hub unlock`). let pid = PublicIdentity { @@ -1106,7 +1078,91 @@ pub(crate) async fn law_integrity_write_gate( /// append). Returns the committed entry index. Requires an ignited signer. /// HUB-001: refuses on law-integrity mismatch — except `LawAmended` itself, /// which is the recovery path (re-witnessing the law is how a mismatch clears). +/// Releases the degraded-log reconciliation slot on drop, so an early return +/// (or a panic) cannot strand the guard and disable reconciliation for the +/// life of the process. +struct ReconcileSlot(Arc); +impl Drop for ReconcileSlot { + fn drop(&mut self) { + self.0.end_reconcile(); + } +} +fn scopeguard_release(s: &RestState) -> ReconcileSlot { + ReconcileSlot(s.degraded_log.clone()) +} + +/// F0.1 (R7a): fold the local degraded diagnostic log into the witnessed +/// ledger. The local log is the fallback witness for a window in which the +/// signer — the normal witness — was the unreachable thing; this binds its +/// exact drained bytes by digest so the two records are auditable against each +/// other. +/// +/// **Reconciliation is opportunistic, not ignition-only.** It runs at ignition +/// AND after any signing act that succeeds while unreconciled entries are +/// pending, so a dependency that recovers on a live hub (a hestia callback +/// coming back, a peer reachable again) does not leave its window unwitnessed +/// until the next restart. The hub may run for weeks between ignitions; a +/// record that waits that long is not a record anyone can act on. +/// +/// Best-effort by construction: snapshot → witness → commit, so a failed +/// witness re-reports on the next attempt (auditable duplication with a +/// matching digest) rather than losing entries. Never blocks or fails its +/// caller. +async fn reconcile_degraded_log(s: &RestState, occasion: &str) { + // Cheap guard: an atomic load, so the witness path pays nothing when + // nothing is degraded (the overwhelmingly common case). + if !s.degraded_log.has_pending() { + return; + } + // Claim the slot: this both prevents recursion (the witness below re-enters + // here) and collapses concurrent attempts into one. + if !s.degraded_log.try_begin_reconcile() { + return; + } + let _release = scopeguard_release(s); + let Some(snap) = s.degraded_log.snapshot() else { return }; + let count = snap.summary.count; + let ev = HubEvent::DegradedReconciled { + count, + first_ts: snap.summary.first_ts, + last_ts: snap.summary.last_ts, + entries_digest: snap.summary.entries_digest.clone(), + by_source: snap.summary.by_source.clone(), + }; + match witness_event_inner(s, ev).await { + Ok(idx) => { + // Witnessed — only now remove the reconciled bytes. + s.degraded_log.commit_reconciled(&snap); + tracing::info!( + "{occasion}: reconciled {count} degraded event(s) into ledger entry {idx}" + ); + } + Err(e) => tracing::warn!( + "{occasion}: degraded-log reconciliation failed (non-fatal; entries \ + re-report on the next attempt): {}", + e.message + ), + } +} + async fn witness_event(s: &RestState, event: HubEvent) -> Result { + let index = witness_event_inner(s, event).await?; + // F0.1 (R7a): this act just signed and witnessed successfully, so signing + // capability is live NOW. If a degraded window is still unwitnessed, fold + // it in — a dependency that recovers on a running hub must not leave its + // record waiting for the next ignition (hubs run for weeks between + // restarts). Costs one atomic load when nothing is pending, and never + // fails this caller: the act above is already committed, and + // reconciliation is bookkeeping. + reconcile_degraded_log(s, "post-witness").await; + Ok(index) +} + +/// The witness path WITHOUT the reconciliation hook. Reconciliation witnesses +/// its own event, so it calls this — breaking the recursion structurally +/// rather than relying on a runtime guard to unwind it (the guard still exists +/// for concurrency, but the call graph is acyclic by construction). +async fn witness_event_inner(s: &RestState, event: HubEvent) -> Result { let (unsigned, signature) = authorize_event(s, event).await?; let mut ledger = s.ledger.lock().await; let entry = ledger @@ -7583,10 +7639,20 @@ async fn commit_pair_event(s: &RestState, event: HubEvent) -> Result<(u64, Strin .sign(s.sovereign_lct_id, &signing_bytes, &intent) .await .map_err(|e| ApiError::internal(anyhow::anyhow!("Sovereign signer denied/failed: {}", e)))?; - let mut ledger = s.ledger.lock().await; - let entry = ledger.append_signed(unsigned, signature).await - .map_err(ApiError::internal)?; - Ok((entry.index, entry.entry_hash.clone())) + let (index, hash) = { + let mut ledger = s.ledger.lock().await; + let entry = ledger.append_signed(unsigned, signature).await + .map_err(ApiError::internal)?; + (entry.index, entry.entry_hash.clone()) + }; + // F0.1 (R7a): a successful signed append proves signing capability is live + // — fold any still-unwitnessed degraded window in now, without waiting for + // an ignition. (This path signs Sovereign acts directly rather than through + // `witness_event`, so it needs its own hook: a reconciliation trigger on one + // signing path and not its sibling would leave whole classes of recovery + // unreconciled.) + reconcile_degraded_log(s, "post-commit").await; + Ok((index, hash)) } #[cfg(test)] @@ -9201,6 +9267,40 @@ pub(crate) mod channel_e2e_tests { Ok(open_resp(m, hub_pub, pid, &r.0.sealed)) } + /// F0.1 (R7a), review follow-up: a dependency that recovers while the hub + /// stays LIVE must not leave its degraded window unwitnessed until the next + /// ignition. Hubs run for weeks between restarts; a record that waits that + /// long is not a record anyone can act on. + #[tokio::test] + async fn live_recovery_reconciles_without_an_ignition() { + let (_tmp, state) = fresh_rest_state(None).await; + let hub_pub = state.signer.public_key().unwrap(); + + // A degraded window happens (the signer was unreachable for a while). + state.degraded_log.append( + hub_lib::degraded::DegradedSource::SignerUnreachable, "sign:member_added: refused"); + state.degraded_log.append( + hub_lib::degraded::DegradedSource::GateTimeout, "law gate: timed out"); + assert!(state.degraded_log.has_pending(), "entries are pending"); + + // The dependency recovers — no restart, no unlock. The very next + // successful witnessed act proves capability is back. + let (_kp, _lct) = pin_member(&state, &hub_pub).await; + + assert!(!state.degraded_log.has_pending(), + "the window reconciled on live recovery, not at some future ignition"); + let proj = { let l = state.ledger.lock().await; HubState::project(&l) }; + assert_eq!(proj.degraded.reconciliations, 1); + assert_eq!(proj.degraded.entries, 2); + assert_eq!(proj.degraded.by_source.get("signer_unreachable"), Some(&1)); + assert_eq!(proj.degraded.by_source.get("gate_timeout"), Some(&1)); + + // And it does not re-witness an empty window on every later act. + let (_kp2, _lct2) = pin_member(&state, &hub_pub).await; + let proj2 = { let l = state.ledger.lock().await; HubState::project(&l) }; + assert_eq!(proj2.degraded.reconciliations, 1, "no empty re-reconciliation"); + } + #[tokio::test] async fn r7_obligation_opens_then_resolves_on_time() { let (_tmp, state) = fresh_rest_state(None).await; diff --git a/hub/hub-lib/src/degraded.rs b/hub/hub-lib/src/degraded.rs index d4286231..8f34e3d5 100644 --- a/hub/hub-lib/src/degraded.rs +++ b/hub/hub-lib/src/degraded.rs @@ -11,7 +11,13 @@ //! append-only JSONL diagnostic log** — the fallback witness precisely when //! the normal witness is the unreachable thing — reconciled into one //! witnessed [`crate::events::HubEvent::DegradedReconciled`] entry when -//! signing capability returns (at unlock / signer recovery). +//! signing capability returns. +//! +//! **Reconciliation is opportunistic, not ignition-only.** It runs at ignition +//! AND after any successful signed append while entries are pending, so a +//! dependency that recovers on a LIVE hub (a callback coming back, a peer +//! reachable again) is folded in immediately. A hub can run for weeks between +//! ignitions; a record that waits that long is not one anyone can act on. //! //! Design properties: //! - **Best-effort, never load-bearing for serving**: an `append` that cannot @@ -28,6 +34,7 @@ use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::io::Write; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; use web4_core::crypto::sha256_hex; /// Why the hub was degraded for one call. Every variant is an infrastructure @@ -91,13 +98,51 @@ pub struct DrainSummary { pub struct DegradedLog { path: PathBuf, lock: std::sync::Mutex<()>, + /// Cheap "something is pending" flag so the reconciliation hook on the + /// witness path costs an atomic load, not a file read, in the common case + /// (nothing degraded). Set on append; cleared only when a commit leaves + /// the file empty. + dirty: AtomicBool, + /// Re-entrancy + concurrency guard. Reconciliation witnesses a ledger + /// event, and the witness path itself triggers reconciliation — without + /// this, that recurses. It also collapses concurrent attempts, so two + /// signing acts finishing together cannot both witness the same window. + reconciling: AtomicBool, } impl DegradedLog { /// A log at `path`. The file is created lazily on first append; a missing /// file reads as empty. pub fn new(path: impl Into) -> Self { - Self { path: path.into(), lock: std::sync::Mutex::new(()) } + let path = path.into(); + // A log left behind by a previous process still needs reconciling. + let dirty = std::fs::metadata(&path).map(|m| m.len() > 0).unwrap_or(false); + Self { + path, + lock: std::sync::Mutex::new(()), + dirty: AtomicBool::new(dirty), + reconciling: AtomicBool::new(false), + } + } + + /// Is there anything to reconcile? An atomic load — safe to call on the + /// witness path for every event. + pub fn has_pending(&self) -> bool { + self.dirty.load(Ordering::Acquire) + } + + /// Claim the reconciliation slot. `false` = another reconciliation is in + /// flight (or this call is re-entering from inside one); the caller must + /// do nothing. Pair every `true` with [`Self::end_reconcile`]. + pub fn try_begin_reconcile(&self) -> bool { + self.reconciling + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + } + + /// Release the reconciliation slot. + pub fn end_reconcile(&self) { + self.reconciling.store(false, Ordering::Release); } pub fn path(&self) -> &Path { @@ -110,8 +155,22 @@ impl DegradedLog { pub fn append(&self, source: DegradedSource, context: &str) { let mut ctx = context.to_string(); if ctx.len() > MAX_CONTEXT { - ctx.truncate(MAX_CONTEXT); - ctx.push_str("…"); + // Walk DOWN to a char boundary before truncating. `String::truncate` + // panics when the split lands mid-codepoint, and this input is + // attacker-shaped: the context is `format!("{op}: {msg}")` over a + // `SignError::Transport` payload — a peer error body or a non-ASCII + // hostname. A single leading ASCII byte shifts every following + // multibyte character across byte 400. A panic here would fire + // inside the signer's error path, i.e. exactly during the degraded + // window this module exists to record: the record would be lost and + // the caller would unwind 500 instead of receiving its `SignError`, + // inverting this module's stated best-effort invariant. + let mut cut = MAX_CONTEXT; + while cut > 0 && !ctx.is_char_boundary(cut) { + cut -= 1; + } + ctx.truncate(cut); + ctx.push('…'); } let entry = DegradedEntry { source, ts: Utc::now(), context: ctx }; let line = match serde_json::to_string(&entry) { @@ -127,8 +186,10 @@ impl DegradedLog { .append(true) .open(&self.path) .and_then(|mut f| writeln!(f, "{line}")); - if let Err(e) = res { - tracing::error!("degraded-log append failed at {}: {e}", self.path.display()); + match res { + Ok(()) => self.dirty.store(true, Ordering::Release), + Err(e) => tracing::error!( + "degraded-log append failed at {}: {e}", self.path.display()), } } @@ -211,9 +272,15 @@ impl DegradedLog { ); b"" }; + let remaining = tail.len(); if let Err(e) = std::fs::write(&self.path, tail) { tracing::error!("degraded-log commit failed at {}: {e}", self.path.display()); + // Write failed: entries stand, so the log stays dirty. + return; } + // Entries appended after the snapshot survive the commit and still need + // reconciling — clear the flag only when nothing is left. + self.dirty.store(remaining > 0, Ordering::Release); } } @@ -297,4 +364,36 @@ mod tests { let entries = log.read_all(); assert!(entries[0].context.len() <= MAX_CONTEXT + 4); } + + /// Regression: a multibyte character STRADDLING the truncation boundary. + /// The all-ASCII bound test above cannot see this — every ASCII byte is a + /// char boundary, so it never exercises the walk. One leading ASCII byte is + /// all it takes to shift every following 2-byte character across byte 400, + /// and `String::truncate` panics on a non-boundary split. + #[test] + fn context_truncation_survives_a_straddling_multibyte_char() { + let log = tmp_log("multibyte"); + // 1 + 600 bytes; byte 400 lands mid-'é' (the exact shape reproduced in + // review: `assertion failed: self.is_char_boundary(new_len)`, exit 101). + let payload = format!("a{}", "é".repeat(300)); + assert!(!payload.is_char_boundary(MAX_CONTEXT), "the test input must straddle"); + log.append(DegradedSource::SignerUnreachable, &payload); + let entries = log.read_all(); + assert_eq!(entries.len(), 1, "the degraded record survives, not panics"); + assert!(entries[0].context.ends_with('…'), "truncation marker retained"); + + // Non-Latin scripts (3-byte chars) and emoji (4-byte, plus ZWJ + // sequences) exercise different walk depths. + for payload in [ + format!("ab{}", "日".repeat(200)), + format!("abc{}", "🔒".repeat(150)), + ] { + log.append(DegradedSource::GateTimeout, &payload); + } + let all = log.read_all(); + assert_eq!(all.len(), 3, "every multibyte shape recorded"); + for e in &all { + assert!(e.context.len() <= MAX_CONTEXT + 4, "still bounded"); + } + } }