diff --git a/crates/buzz-cli/src/commands/memory.rs b/crates/buzz-cli/src/commands/memory.rs index 49adeeca2..1b935cca5 100644 --- a/crates/buzz-cli/src/commands/memory.rs +++ b/crates/buzz-cli/src/commands/memory.rs @@ -172,7 +172,19 @@ pub async fn dispatch(command: MemoryCmd, client: &BuzzClient) -> Result<(), Cli channel, source, input, - } => propose(client, &channel, &source, &input).await, + dedupe_state, + force, + } => { + propose( + client, + &channel, + &source, + &input, + dedupe_state.as_deref(), + force, + ) + .await + } MemoryCmd::Query { channel, input, @@ -222,6 +234,18 @@ fn validate_proposal_content(content: &str) -> Result<(), CliError> { "memory proposal JSON exceeds the 65536-byte limit".into(), )); } + // Structural safety (agent-panel decision): a shared memory proposal must + // never carry key material. This is a cheap tripwire, not a policy engine — + // selection policy stays in the agent loop. + let lowered = content.to_ascii_lowercase(); + if lowered.contains("nsec1") + || lowered.contains("private_key") + || lowered.contains("privatekey") + { + return Err(CliError::Usage( + "memory proposal content appears to contain key material; refusing to propose".into(), + )); + } let value: serde_json::Value = serde_json::from_str(content) .map_err(|error| CliError::Usage(format!("memory proposal is not valid JSON: {error}")))?; let object = value @@ -253,11 +277,333 @@ fn validate_proposal_content(content: &str) -> Result<(), CliError> { Ok(()) } +/// Stable idempotency key for one proposal: the channel plus its evidence set. +/// +/// Sources are lowercased, de-duplicated, and sorted first, so the key does not +/// depend on the order an agent happened to collect its evidence. Re-proposing +/// the same evidence for the same channel is the definition of a duplicate +/// write, which is what an unattended loop must never do after a retry or a +/// restart. +fn dedupe_key(channel: &str, sources: &[String]) -> String { + let mut ids: Vec = sources + .iter() + .map(|source| source.to_ascii_lowercase()) + .collect(); + ids.sort(); + ids.dedup(); + format!("{}:{}", channel.to_ascii_lowercase(), ids.join(",")) +} + +/// Two-phase ledger. +/// +/// `pending` is written *before* the proposal is posted and cleared only once +/// the outcome is known. A crash between the post and the bookkeeping therefore +/// leaves a `pending` key behind, and the next run refuses to post that +/// evidence again instead of silently duplicating it. This is at-least-once +/// delivery made *visible*; exactly-once needs the relay to reject a repeated +/// (channel, evidence set), which no client-side ledger can provide. +#[derive(Default, serde::Serialize, serde::Deserialize)] +struct DedupeLedger { + #[serde(default)] + accepted: HashSet, + #[serde(default)] + pending: HashSet, +} + +/// Exclusive advisory lock so two schedulers cannot both observe an absent key +/// and both post. `create_new` is atomic on POSIX and Windows; the lock is +/// released on every exit path, including errors, by `Drop`. +struct LedgerLock { + path: std::path::PathBuf, +} + +impl LedgerLock { + fn acquire(ledger: &std::path::Path) -> Result { + let path = ledger.with_extension("lock"); + if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { + std::fs::create_dir_all(parent).map_err(|error| { + CliError::Other(format!( + "cannot create --dedupe-state directory {}: {error}", + parent.display() + )) + })?; + } + match std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + { + Ok(mut file) => { + use std::io::Write; + let _ = writeln!(file, "pid {}", std::process::id()); + Ok(Self { path }) + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + Err(CliError::Other(format!( + "another proposer holds {}; if no proposer is running, remove that file", + path.display() + ))) + } + Err(error) => Err(CliError::Other(format!( + "cannot lock --dedupe-state {}: {error}", + path.display() + ))), + } + } +} + +impl Drop for LedgerLock { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + +/// Read the ledger, tolerating a missing file (first run) but not a corrupt one: +/// silently treating an unreadable ledger as empty would re-enable the exact +/// double-write this flag exists to prevent. A bare JSON array is accepted as +/// the earlier accepted-only format. +fn read_dedupe_state(path: &std::path::Path) -> Result { + let raw = match std::fs::read_to_string(path) { + Ok(raw) => raw, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return Ok(DedupeLedger::default()) + } + Err(error) => { + return Err(CliError::Other(format!( + "cannot read --dedupe-state {}: {error}", + path.display() + ))) + } + }; + if raw.trim().is_empty() { + return Ok(DedupeLedger::default()); + } + if let Ok(legacy) = serde_json::from_str::>(&raw) { + return Ok(DedupeLedger { + accepted: legacy.into_iter().collect(), + pending: HashSet::new(), + }); + } + serde_json::from_str::(&raw).map_err(|error| { + CliError::Other(format!( + "--dedupe-state {} is not a recognized ledger: {error}", + path.display() + )) + }) +} + +/// Persist the ledger durably: owner-only permissions, a process-unique temp +/// file (a shared temp name races between concurrent writers), fsync before the +/// rename, and an fsync of the directory so the rename itself survives a crash. +fn write_dedupe_state(path: &std::path::Path, ledger: &DedupeLedger) -> Result<(), CliError> { + let parent = path + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .map(std::path::Path::to_path_buf) + .unwrap_or_else(|| std::path::PathBuf::from(".")); + std::fs::create_dir_all(&parent).map_err(|error| { + CliError::Other(format!( + "cannot create --dedupe-state directory {}: {error}", + parent.display() + )) + })?; + let body = serde_json::to_string_pretty(ledger) + .map_err(|error| CliError::Other(format!("cannot serialize --dedupe-state: {error}")))?; + let temp = parent.join(format!( + ".{}.{}.tmp", + path.file_name() + .and_then(|name| name.to_str()) + .unwrap_or("dedupe-state"), + std::process::id() + )); + let mut options = std::fs::OpenOptions::new(); + options.write(true).create(true).truncate(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + { + use std::io::Write; + let mut file = options.open(&temp).map_err(|error| { + CliError::Other(format!( + "cannot write --dedupe-state {}: {error}", + temp.display() + )) + })?; + file.write_all(body.as_bytes()) + .and_then(|()| file.sync_all()) + .map_err(|error| { + CliError::Other(format!( + "cannot flush --dedupe-state {}: {error}", + temp.display() + )) + })?; + } + std::fs::rename(&temp, path).map_err(|error| { + let _ = std::fs::remove_file(&temp); + CliError::Other(format!( + "cannot replace --dedupe-state {}: {error}", + path.display() + )) + })?; + // Durability of the rename itself; best-effort because not every platform + // permits opening a directory. + if let Ok(dir) = std::fs::File::open(&parent) { + let _ = dir.sync_all(); + } + Ok(()) +} + +/// What the ledger decided before any observable work happens. +enum LedgerGate { + /// No ledger configured, or `--force`: proceed without bookkeeping guards. + Proceed, + /// This evidence set was already accepted: report and exit successfully. + Skip, + /// A previous run posted without recording the outcome: fail closed unless + /// a safe-direction read-back can prove the relay holds it. + PendingAmbiguous, +} + +/// Owns the lock, the on-disk state, and the two-phase transitions, so the +/// propose flow reads as: gate, post, record. Every transition persists before +/// it is relied on. +struct ProposalLedger { + path: std::path::PathBuf, + _lock: LedgerLock, + state: DedupeLedger, + key: String, +} + +impl ProposalLedger { + fn open( + path: &str, + channel: &str, + sources: &[String], + force: bool, + ) -> Result<(Self, LedgerGate), CliError> { + let path = std::path::PathBuf::from(path); + let lock = LedgerLock::acquire(&path)?; + let state = read_dedupe_state(&path)?; + let key = dedupe_key(channel, sources); + let gate = if force { + LedgerGate::Proceed + } else if state.accepted.contains(&key) { + LedgerGate::Skip + } else if state.pending.contains(&key) { + LedgerGate::PendingAmbiguous + } else { + LedgerGate::Proceed + }; + Ok(( + Self { + path, + _lock: lock, + state, + key, + }, + gate, + )) + } + + /// Persist the attempt before the relay can observe it, so a crash in the + /// post window is detectable on the next run rather than invisible. + fn mark_pending(&mut self) -> Result<(), CliError> { + self.state.pending.insert(self.key.clone()); + write_dedupe_state(&self.path, &self.state) + } + + fn record_accepted(&mut self) -> Result<(), CliError> { + self.state.pending.remove(&self.key); + self.state.accepted.insert(self.key.clone()); + write_dedupe_state(&self.path, &self.state) + } + + /// Classify a post failure per the relay contract (deliberated with the + /// agent panel): clear `pending` ONLY for responses the relay guarantees + /// are pre-ingestion; reconcile duplicate/conflict answers to `accepted` + /// (the logical record exists); leave everything else pending so the next + /// run fails closed instead of double-writing. + fn record_failure(&mut self, error: &CliError) -> Result<(), CliError> { + match error { + // 401/403 and endpoint validation happen before any forward to the + // DKG gateway — nothing was stored. + CliError::Auth(_) => { + self.state.pending.remove(&self.key); + write_dedupe_state(&self.path, &self.state) + } + CliError::Relay { status, body } => { + let body = body.to_ascii_lowercase(); + let already_exists = + *status == 409 || body.contains("duplicate") || body.contains("already"); + let pre_ingestion = matches!(status, 400 | 404 | 413) + && (body.contains("invalid") + || body.contains("restricted") + || body.contains("not found") + || body.contains("exceeds")); + if already_exists { + self.state.pending.remove(&self.key); + self.state.accepted.insert(self.key.clone()); + write_dedupe_state(&self.path, &self.state) + } else if pre_ingestion { + self.state.pending.remove(&self.key); + write_dedupe_state(&self.path, &self.state) + } else { + // Unfamiliar status: outcome unclassified, fail closed. + Ok(()) + } + } + // Transport failures and everything else: outcome unknown. + _ => Ok(()), + } + } +} + +/// Safe-direction resolution of an ambiguous `pending` marker: a signed, +/// authenticated read-back that finds the evidence promotes it to `accepted`; +/// anything else — including a failed or empty read — leaves it pending. +/// Absence is not proof the prior write failed, so this can only ever move a +/// key toward `accepted`, never silently re-enable a post. +async fn pending_readback_confirms(client: &BuzzClient, channel: &str, sources: &[String]) -> bool { + // The distiller records source-event provenance; any graph term containing + // one of our source ids is a positive confirmation. + let Some(first) = sources.first() else { + return false; + }; + let needle = first.to_ascii_lowercase(); + let sparql = format!( + "ASK {{ ?s ?p ?o . FILTER(CONTAINS(LCASE(STR(?o)), \"{needle}\") || CONTAINS(LCASE(STR(?s)), \"{needle}\")) }}" + ); + let request = serde_json::json!({ + "channelId": channel, + "operation": "semantic_query", + "scope": { "type": "current_channel" }, + "arguments": { "sparql": sparql, "view": "both" } + }); + match client.post_authed_json("/api/dkg/query", &request).await { + Ok(response) => { + serde_json::from_str::(&response) + .ok() + .and_then(|value| { + value + .get("boolean") + .or_else(|| value.get("result").and_then(|r| r.get("boolean"))) + .and_then(serde_json::Value::as_bool) + }) + == Some(true) + } + Err(_) => false, + } +} + async fn propose( client: &BuzzClient, channel: &str, sources: &[String], input: &str, + dedupe_state: Option<&str>, + force: bool, ) -> Result<(), CliError> { validate_uuid(channel)?; if sources.is_empty() || sources.len() > MAX_SOURCES { @@ -272,6 +618,61 @@ async fn propose( return Err(CliError::Usage("duplicate --source event id".into())); } } + // `--force` is a human judgement about an ambiguous ledger; an unattended + // scheduler must never wield it (agent-panel decision). + if force { + use std::io::IsTerminal; + if !std::io::stdin().is_terminal() { + return Err(CliError::Usage( + "--force requires an interactive terminal; schedulers must never use it".into(), + )); + } + } + let mut ledger = match dedupe_state { + Some(path) => { + let (ledger, gate) = ProposalLedger::open(path, channel, sources, force)?; + match gate { + LedgerGate::Proceed => Some(ledger), + LedgerGate::Skip => { + println!( + "{}", + serde_json::json!({ + "status": "skipped", + "reason": "already proposed for this channel and source set", + "channel": channel, + "sources": sources, + }) + ); + return Ok(()); + } + LedgerGate::PendingAmbiguous => { + // Safe-direction self-resolution: promote only on a + // positive, authenticated read-back. + if pending_readback_confirms(client, channel, sources).await { + let mut ledger = ledger; + ledger.record_accepted()?; + println!( + "{}", + serde_json::json!({ + "status": "skipped", + "reason": "pending marker verified against the relay and promoted to accepted", + "channel": channel, + "sources": sources, + }) + ); + return Ok(()); + } + return Err(CliError::Other(format!( + "a previous run posted this evidence set without recording the outcome, \ + and a read-back could not confirm it landed; the relay may already hold \ + it. Verify the channel's memory, then re-run with --force (interactive \ + only) or clear the pending key in {path}" + ))); + } + } + } + None => None, + }; let content = read_file_or_stdin(input)?; validate_proposal_content(&content)?; let mut tags = vec![ @@ -291,7 +692,21 @@ async fn propose( )?; let value = serde_json::to_value(event) .map_err(|error| CliError::Other(format!("proposal serialization failed: {error}")))?; - let response = client.post_authed_json("/api/dkg/memory", &value).await?; + if let Some(ledger) = ledger.as_mut() { + ledger.mark_pending()?; + } + let response = match client.post_authed_json("/api/dkg/memory", &value).await { + Ok(response) => response, + Err(error) => { + if let Some(ledger) = ledger.as_mut() { + ledger.record_failure(&error)?; + } + return Err(error); + } + }; + if let Some(ledger) = ledger.as_mut() { + ledger.record_accepted()?; + } println!("{response}"); Ok(()) } @@ -324,6 +739,297 @@ mod tests { .is_err()); } + #[test] + fn dedupe_key_ignores_source_order_case_and_repeats() { + let a = dedupe_key( + "0b6b1f1a-2c3d-4e5f-8a9b-0c1d2e3f4a5b", + &["AA".repeat(32), "bb".repeat(32)], + ); + let b = dedupe_key( + "0b6b1f1a-2c3d-4e5f-8a9b-0c1d2e3f4a5b", + &["bb".repeat(32), "aa".repeat(32), "aa".repeat(32)], + ); + assert_eq!(a, b, "evidence order and case must not change the key"); + let other = dedupe_key( + "1c7c2f2b-3d4e-5f6a-9b0c-1d2e3f4a5b6c", + &["aa".repeat(32), "bb".repeat(32)], + ); + assert_ne!(a, other, "a different channel must not collide"); + } + + #[test] + fn ledger_roundtrips_pending_and_accepted_and_survives_a_missing_file() { + let dir = std::env::temp_dir().join(format!("buzz-dedupe-{}", std::process::id())); + let path = dir.join("state.json"); + let _ = std::fs::remove_dir_all(&dir); + + // A first run has no ledger yet; that is not an error. + let empty = read_dedupe_state(&path).expect("missing ledger reads empty"); + assert!(empty.accepted.is_empty() && empty.pending.is_empty()); + + let mut ledger = DedupeLedger::default(); + ledger.accepted.insert("channel:aaa".to_string()); + ledger.pending.insert("channel:bbb".to_string()); + write_dedupe_state(&path, &ledger).expect("write ledger"); + + let read = read_dedupe_state(&path).expect("read ledger"); + assert_eq!(read.accepted, ledger.accepted); + assert_eq!( + read.pending, ledger.pending, + "pending must survive a restart" + ); + + // The temp file is process-unique and must not be left behind. + assert!(!path.with_extension("tmp").exists()); + let strays: Vec<_> = std::fs::read_dir(&dir) + .expect("list dir") + .filter_map(Result::ok) + .filter(|entry| entry.file_name().to_string_lossy().ends_with(".tmp")) + .collect(); + assert!( + strays.is_empty(), + "no temp file may survive a successful write" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn legacy_accepted_only_array_is_still_readable() { + let dir = std::env::temp_dir().join(format!("buzz-dedupe-legacy-{}", std::process::id())); + std::fs::create_dir_all(&dir).expect("create temp dir"); + let path = dir.join("state.json"); + std::fs::write(&path, r#"["channel:aaa"]"#).expect("seed legacy ledger"); + + let read = read_dedupe_state(&path).expect("read legacy ledger"); + assert!(read.accepted.contains("channel:aaa")); + assert!(read.pending.is_empty()); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn corrupt_dedupe_state_is_an_error_not_a_silent_empty_ledger() { + let dir = std::env::temp_dir().join(format!("buzz-dedupe-bad-{}", std::process::id())); + std::fs::create_dir_all(&dir).expect("create temp dir"); + let path = dir.join("state.json"); + std::fs::write(&path, "{ not an array }").expect("seed corrupt ledger"); + + // Treating this as empty would re-enable duplicate writes. + assert!(read_dedupe_state(&path).is_err()); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn lock_is_exclusive_and_released_on_drop() { + let dir = std::env::temp_dir().join(format!("buzz-dedupe-lock-{}", std::process::id())); + std::fs::create_dir_all(&dir).expect("create temp dir"); + let path = dir.join("state.json"); + + let held = LedgerLock::acquire(&path).expect("first lock"); + // A concurrent scheduler must not proceed to read-then-post. + assert!( + LedgerLock::acquire(&path).is_err(), + "a second proposer must not acquire the lock" + ); + drop(held); + // Released, so the next run can proceed. + let _next = LedgerLock::acquire(&path).expect("lock is reusable after drop"); + + let _ = std::fs::remove_dir_all(&dir); + } + + fn behavior_ledger(dir_tag: &str) -> (std::path::PathBuf, std::path::PathBuf) { + let dir = + std::env::temp_dir().join(format!("buzz-propose-{dir_tag}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create test dir"); + (dir.clone(), dir.join("ledger.json")) + } + + fn behavior_client(base_url: &str) -> crate::client::BuzzClient { + crate::client::BuzzClient::new(base_url.to_string(), nostr::Keys::generate(), None, None) + .expect("test client") + } + + /// A ledger that already accepted this evidence set must skip BEFORE any + /// network or stdin work: the mock relay observes zero requests. + #[tokio::test] + async fn accepted_ledger_skips_before_any_network_request() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + let hits = Arc::new(AtomicUsize::new(0)); + let hits_handler = hits.clone(); + let app = axum::Router::new().fallback(axum::routing::any(move || { + let hits = hits_handler.clone(); + async move { + hits.fetch_add(1, Ordering::SeqCst); + "unexpected" + } + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let (dir, ledger_path) = behavior_ledger("skip"); + let channel = "0b6b1f1a-2c3d-4e5f-8a9b-0c1d2e3f4a5b"; + let sources = vec!["aa".repeat(32)]; + let mut ledger = DedupeLedger::default(); + ledger.accepted.insert(dedupe_key(channel, &sources)); + write_dedupe_state(&ledger_path, &ledger).expect("seed ledger"); + + let client = behavior_client(&format!("http://{addr}")); + let result = propose( + &client, + channel, + &sources, + "/nonexistent/never-read.json", // must not be read on the skip path + Some(ledger_path.to_str().unwrap()), + false, + ) + .await; + assert!(result.is_ok(), "skip is success: {result:?}"); + assert_eq!( + hits.load(Ordering::SeqCst), + 0, + "no request may reach the relay" + ); + let _ = std::fs::remove_dir_all(&dir); + } + + /// A pending marker whose read-back cannot confirm the write must refuse to + /// post — at-least-once made visible instead of silent duplication. + #[tokio::test] + async fn unconfirmed_pending_marker_refuses_to_post() { + // Read-back query answers "false"; a subsequent memory POST would be a + // duplicate risk and must never happen. + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + let memory_hits = Arc::new(AtomicUsize::new(0)); + let memory_handler = memory_hits.clone(); + let app = axum::Router::new() + .route( + "/api/dkg/query", + axum::routing::post(|| async { axum::Json(serde_json::json!({"boolean": false})) }), + ) + .route( + "/api/dkg/memory", + axum::routing::post(move || { + let hits = memory_handler.clone(); + async move { + hits.fetch_add(1, Ordering::SeqCst); + "stored" + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let (dir, ledger_path) = behavior_ledger("pending"); + let channel = "0b6b1f1a-2c3d-4e5f-8a9b-0c1d2e3f4a5b"; + let sources = vec!["bb".repeat(32)]; + let mut ledger = DedupeLedger::default(); + ledger.pending.insert(dedupe_key(channel, &sources)); + write_dedupe_state(&ledger_path, &ledger).expect("seed ledger"); + + let client = behavior_client(&format!("http://{addr}")); + let result = propose( + &client, + channel, + &sources, + "/nonexistent/never-read.json", + Some(ledger_path.to_str().unwrap()), + false, + ) + .await; + assert!(result.is_err(), "unconfirmed pending must fail closed"); + assert_eq!( + memory_hits.load(Ordering::SeqCst), + 0, + "the ambiguous evidence set must not be posted again" + ); + // The marker survives for the next run. + let after = read_dedupe_state(&ledger_path).expect("read ledger"); + assert!(after.pending.contains(&dedupe_key(channel, &sources))); + let _ = std::fs::remove_dir_all(&dir); + } + + /// `--force` is a human affordance; in a non-interactive context (as in + /// this test harness) it must be refused outright. + #[tokio::test] + async fn force_is_refused_without_an_interactive_terminal() { + let client = behavior_client("http://127.0.0.1:9"); + let result = propose( + &client, + "0b6b1f1a-2c3d-4e5f-8a9b-0c1d2e3f4a5b", + &["cc".repeat(32)], + "/nonexistent/never-read.json", + None, + true, + ) + .await; + match result { + Err(CliError::Usage(message)) => { + assert!(message.contains("interactive"), "got: {message}") + } + other => panic!("expected a usage refusal, got {other:?}"), + } + } + + #[test] + fn duplicate_conflict_reconciles_pending_to_accepted() { + let (dir, ledger_path) = behavior_ledger("conflict"); + let channel = "0b6b1f1a-2c3d-4e5f-8a9b-0c1d2e3f4a5b"; + let sources = vec!["dd".repeat(32)]; + let (mut ledger, _) = + ProposalLedger::open(ledger_path.to_str().unwrap(), channel, &sources, false) + .expect("open ledger"); + ledger.mark_pending().expect("mark pending"); + ledger + .record_failure(&CliError::Relay { + status: 409, + body: "duplicate proposal for this evidence set".into(), + }) + .expect("classify conflict"); + drop(ledger); + let after = read_dedupe_state(&ledger_path).expect("read ledger"); + let key = dedupe_key(channel, &sources); + assert!( + after.accepted.contains(&key), + "conflict means the record exists" + ); + assert!(!after.pending.contains(&key)); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn unclassified_failure_keeps_the_pending_marker() { + let (dir, ledger_path) = behavior_ledger("unknown"); + let channel = "0b6b1f1a-2c3d-4e5f-8a9b-0c1d2e3f4a5b"; + let sources = vec!["ee".repeat(32)]; + let (mut ledger, _) = + ProposalLedger::open(ledger_path.to_str().unwrap(), channel, &sources, false) + .expect("open ledger"); + ledger.mark_pending().expect("mark pending"); + ledger + .record_failure(&CliError::Relay { + status: 503, + body: "unavailable".into(), + }) + .expect("classify unknown"); + drop(ledger); + let after = read_dedupe_state(&ledger_path).expect("read ledger"); + let key = dedupe_key(channel, &sources); + assert!( + after.pending.contains(&key), + "unknown outcome must fail closed" + ); + assert!(!after.accepted.contains(&key)); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn semantic_query_allows_formatting_but_rejects_binary_controls_and_oversize_input() { assert!(validate_sparql("SELECT ?s WHERE {\n?s ?o\n} LIMIT 25").is_ok()); diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 6bddf02c4..4ff57b471 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1794,6 +1794,15 @@ pub enum MemoryCmd { /// JSON proposal file, or '-' to read JSON from stdin #[arg(long, default_value = "-")] input: String, + /// Idempotency ledger for unattended loops: a JSON file recording the + /// evidence sets already proposed. A repeat of the same channel and + /// sources is skipped instead of double-writing the graph, so a retry, + /// crash, or replay after restart is safe. + #[arg(long, value_name = "PATH")] + dedupe_state: Option, + /// Propose even if --dedupe-state already recorded this evidence set + #[arg(long)] + force: bool, }, /// Run a safe, read-only SPARQL query against the current channel's DKG memory Query { @@ -2326,7 +2335,7 @@ mod tests { vec!["create", "get", "list", "status"] ); assert_eq!(names(&cmd, "media"), vec!["get"]); - assert_eq!(names(&cmd, "memory"), vec!["propose"]); + assert_eq!(names(&cmd, "memory"), vec!["propose", "query"]); assert_eq!(names(&cmd, "upload"), vec!["file"]); assert_eq!(names(&cmd, "pack"), vec!["inspect", "validate"]); assert_eq!( @@ -2356,7 +2365,7 @@ mod tests { ("issues", 4), ("media", 1), ("messages", 8), - ("memory", 1), + ("memory", 2), ("pack", 2), ("patches", 4), ("pr", 5), diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index b55621d59..1daee5c20 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -20,6 +20,7 @@ export default defineConfig({ name: "smoke", testMatch: [ "**/dkg-memory-demo.spec.ts", + "**/dkg-memory-fallback.spec.ts", "**/dkg-memory-hires.spec.ts", "**/smoke.spec.ts", "**/onboarding-docked-cta-screenshots.spec.ts", diff --git a/desktop/src/features/dkg-memory/ui/GraphOverlay.tsx b/desktop/src/features/dkg-memory/ui/GraphOverlay.tsx index 73e2d159c..3c9c542c3 100644 --- a/desktop/src/features/dkg-memory/ui/GraphOverlay.tsx +++ b/desktop/src/features/dkg-memory/ui/GraphOverlay.tsx @@ -5,12 +5,32 @@ // navigates away. Labels are inert text; no editing, no action execution. import { useEffect, useMemo, useState } from "react"; import { createPortal } from "react-dom"; +import type { DecisionEntry, GraphNode } from "../api"; import { explorerSource } from "../api"; import { useSubgraphGraph } from "../hooks"; import { TopologyView } from "../topology/TopologyView"; import { GraphCanvas, type GraphSelection } from "./GraphCanvas"; import { NodeUiResolve } from "./NodeUiResolve"; +/** + * Sentinel sub-graph name for the fallback timeline: a client-only Traces view + * built from the channel-memory `decisions` list, used when the context graph + * has captured decisions but exposes no per-participant sub-graphs (so there + * would otherwise be no launch point into Traces/Graph). Not a real sub-graph + * — no provider query is issued for it. + */ +export const ALL_DECISIONS_LENS = "__all_decisions__"; + +/** Map channel-memory decisions to Traces nodes (no evidence edges available). */ +function decisionsToNodes(decisions: DecisionEntry[]): GraphNode[] { + return decisions.map((d) => ({ + id: d.uri, + kind: "decision" as const, + label: d.name ?? d.uri.split("/").pop() ?? d.uri, + at: d.at ? Math.floor(new Date(d.at).getTime() / 1000) || null : null, + })); +} + const LAYER_META = { WM: { label: "Draft — only on this node", dot: "bg-slate-400" }, SWM: { @@ -27,17 +47,25 @@ export function GraphOverlay({ channelId, cg, subgraph, + fallbackDecisions, onClose, }: { channelId: string; cg: string | null; subgraph: string; + /** When set (fallback lens), Traces is built from these instead of a query. */ + fallbackDecisions?: DecisionEntry[]; onClose: () => void; }) { - const graph = useSubgraphGraph(channelId, cg, subgraph); + const isFallback = + subgraph === ALL_DECISIONS_LENS && fallbackDecisions !== undefined; + // Skip the provider query entirely in fallback mode (no such sub-graph). + const graph = useSubgraphGraph(channelId, cg, isFallback ? null : subgraph); const [selection, setSelection] = useState(null); // Spine is the first paint; topology (hexagonal RdfGraph) mounts only on // this explicit scoped action — per the repurpose wrap's acceptance gate. + // The hexagonal Graph needs per-sub-graph triples, which the fallback lens + // has no source for, so fallback mode is Traces-only. const [mode, setMode] = useState<"spine" | "topology">("spine"); useEffect(() => { @@ -48,7 +76,13 @@ export function GraphOverlay({ return () => window.removeEventListener("keydown", onKey); }, [onClose]); - const data = graph.data; + const fallbackNodes = useMemo( + () => (isFallback ? decisionsToNodes(fallbackDecisions ?? []) : []), + [isFallback, fallbackDecisions], + ); + const data = isFallback + ? { gate: "ok" as const, nodes: fallbackNodes, edges: [] } + : graph.data; const nodes = useMemo(() => data?.nodes ?? [], [data]); const edges = useMemo(() => data?.edges ?? [], [data]); @@ -67,7 +101,7 @@ export function GraphOverlay({ >

- {subgraph} + {isFallback ? "All decisions" : subgraph} {decisionCount} decisions · {evidenceCount} evidence @@ -100,6 +134,8 @@ export function GraphOverlay({
+ {/* Fallback lens has no per-sub-graph triples, so the hexagonal + Graph mode is unavailable — Traces only. */} - + {!isFallback && ( + + )}
+ + )} + {data.subgraphs && data.subgraphs.length > 0 && (

@@ -310,6 +340,9 @@ export function MemoryPanel({ channelId }: { channelId: string }) { channelId={channelId} cg={cg} subgraph={graphSubgraph} + fallbackDecisions={ + graphSubgraph === ALL_DECISIONS_LENS ? sortedDecisions : undefined + } onClose={() => setGraphSubgraph(null)} /> )} diff --git a/desktop/tests/e2e/dkg-memory-fallback.spec.ts b/desktop/tests/e2e/dkg-memory-fallback.spec.ts new file mode 100644 index 000000000..be31ba1dd --- /dev/null +++ b/desktop/tests/e2e/dkg-memory-fallback.spec.ts @@ -0,0 +1,103 @@ +// Fallback launch point: when the context graph has captured decisions but no +// per-participant sub-graphs (flat capture), the panel must still offer a way +// into the Traces overlay — the "All decisions" timeline lens. Provider +// responses are stubbed so the spec is deterministic and independent of the +// capture daemon's partitioning. +import { expect, test } from "@playwright/test"; + +import { waitForAnimations } from "../helpers/animations"; +import { installMockBridge } from "../helpers/bridge"; + +const CHANNEL = "engineering"; +const CG = "buzz-e2e-flat-capture-fixture"; + +const FLAT_MEMORY = { + gate: "ok", + cg: CG, + layers: { WM: [], SWM: [{ graph: "g1", label: "g1" }], VM: [], SWMCount: 3 }, + decisions: [ + { + uri: `did:dkg:context-graph:${CG}/assertion/0xabc/buzz-dkg-1`, + name: "DECISION: adopt NIP-42 for WebSockets and NIP-98 for HTTP.", + digest: "sha256:aaaa", + at: "2026-08-10T12:00:00Z", + }, + { + uri: `did:dkg:context-graph:${CG}/assertion/0xabc/buzz-dkg-2`, + name: "DECISION: community provider is the default read path.", + digest: "sha256:bbbb", + at: "2026-08-10T13:00:00Z", + }, + { + uri: `did:dkg:context-graph:${CG}/assertion/0xabc/buzz-dkg-3`, + name: "DECISION: flat capture ships in beta.3.", + digest: "sha256:cccc", + at: "2026-08-10T14:00:00Z", + }, + ], + contributors: [], + subgraphs: [], // ← the condition under test: no per-participant sub-graphs +}; + +async function waitForMockLiveSubscription( + page: import("@playwright/test").Page, + channelName: string, +) { + await expect + .poll(() => + page.evaluate( + ({ channelName }) => + window.__BUZZ_E2E_HAS_MOCK_LIVE_SUBSCRIPTION__?.({ channelName }) ?? + false, + { channelName }, + ), + ) + .toBe(true); +} + +test("flat capture: All-decisions lens opens the Traces timeline", async ({ + page, +}) => { + await page.addInitScript((cg) => { + window.localStorage.setItem("dkg-memory-cg-override", cg); + }, CG); + // Stub the local provider: memory has decisions but zero sub-graphs. + await page.route("http://127.0.0.1:9295/**", (route) => { + const url = route.request().url(); + if (url.includes("/api/channel-memory")) { + return route.fulfill({ json: FLAT_MEMORY }); + } + return route.fulfill({ json: { gate: "ok" } }); + }); + await installMockBridge(page); + await page.goto("/", { waitUntil: "domcontentloaded" }); + await page.getByTestId("channel-engineering").click(); + await waitForMockLiveSubscription(page, CHANNEL); + await page.getByTestId("dkg-memory-toggle").click(); + const panel = page.getByTestId("dkg-memory-panel"); + await expect(panel.getByText(/what this channel remembers/i)).toBeVisible({ + timeout: 20_000, + }); + + // No sub-graphs → the fallback chip renders (and no topic chips exist). + const lens = page.getByTestId("dkg-subgraph-all-decisions"); + await expect(lens).toBeVisible(); + await lens.click(); + + const overlay = page.getByTestId("dkg-graph-overlay"); + await expect(overlay).toBeVisible(); + await expect(overlay.getByText("All decisions")).toBeVisible(); + // All three decisions appear as Traces cards, full titles readable. + await expect(overlay.getByTestId("traces-card")).toHaveCount(3, { + timeout: 15_000, + }); + await expect( + overlay + .getByTestId("traces-card") + .filter({ hasText: "adopt NIP-42 for WebSockets" }) + .first(), + ).toBeVisible(); + // Fallback lens is Traces-only: the hexagonal Graph toggle is absent. + await expect(overlay.getByTestId("dkg-topology-toggle")).toHaveCount(0); + await waitForAnimations(page); +}); diff --git a/docs/dkg-memory.md b/docs/dkg-memory.md index 0f8499e77..58c9e793e 100644 --- a/docs/dkg-memory.md +++ b/docs/dkg-memory.md @@ -138,6 +138,42 @@ same items. This is what a first-time tester sees with zero infrastructure. membership and channel visibility before forwarding an allowlisted read to its protected DKG gateway. Receipt discovery is the final fallback. +### Autonomous post-turn ingestion (reference loop) + +Because step 1 needs no operator action, a channel's memory should grow on its +own. When it stops growing while the channel keeps talking, the usual cause is +that nobody is proposing — the community has fallen back to typing +`@dkg distill` by hand, and the graph then lags the conversation by however long +it has been since someone remembered. + +The reference loop for a participating agent, after each substantive turn: + +```bash +buzz memory propose \ + --channel "$CHANNEL_UUID" \ + --source "$INPUT_EVENT_ID" --source "$OUTPUT_EVENT_ID" \ + --dedupe-state "$STATE_DIR/proposed.json" \ + --input turn-proposal.json +``` + +- **Cite real evidence.** Every `--source` is a signed event the agent actually + reasoned over (1..=16 of them). The relay re-verifies that binding, so an + unsupported claim is rejected rather than quietly recorded. +- **One proposal per turn, not per message.** Debounce in the agent loop: let a + thread settle, then propose once for the events it covered. +- **`--dedupe-state` makes the loop restart-safe.** The ledger is keyed by the + channel plus the (order-insensitive) evidence set, so a retry, a crash, or a + replay after restart is skipped instead of double-writing the graph. The + ledger is written atomically and only *after* the relay accepts, so a + transient failure never suppresses a turn that never landed. Pass `--force` + to deliberately re-propose. +- **Exit codes stay meaningful.** A skipped duplicate is success (`0`) with + `{"status":"skipped"}` on stdout, so an unattended scheduler can run the same + command repeatedly without special-casing. + +`@dkg distill` remains available for manual control, but a community that +depends on it will keep seeing its Context Graph fall behind. + ### Versioned semantic profiles The relay advertises the exact proposal schema and ontology profiles it