diff --git a/Cargo.lock b/Cargo.lock index 803dcf1ac8..bdba801c23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -830,6 +830,7 @@ name = "buzz-accumulator" version = "0.1.0" dependencies = [ "chrono", + "libc", "serde", "serde_json", "sha2 0.11.0", diff --git a/crates/buzz-accumulator/Cargo.toml b/crates/buzz-accumulator/Cargo.toml index 5dc3d21b35..37270fc92a 100644 --- a/crates/buzz-accumulator/Cargo.toml +++ b/crates/buzz-accumulator/Cargo.toml @@ -13,3 +13,6 @@ serde_json = { workspace = true } thiserror = { workspace = true } sha2 = { workspace = true } chrono = { workspace = true } + +[target.'cfg(unix)'.dependencies] +libc = "0.2" diff --git a/crates/buzz-accumulator/src/artifact.rs b/crates/buzz-accumulator/src/artifact.rs index 6e66f6389a..9963f14972 100644 --- a/crates/buzz-accumulator/src/artifact.rs +++ b/crates/buzz-accumulator/src/artifact.rs @@ -2,6 +2,8 @@ use serde::{Deserialize, Serialize}; +use crate::selection::Selection; + /// One artifact version — the JSON a caller NIP-44-encrypts into the content /// of an immutable relay event. /// @@ -26,7 +28,21 @@ pub struct ArtifactPayload { pub coverage_since: Option, /// Half-open end of the covered window: latest shown timestamp + 1; /// `None` when the run showed no signals. + /// + /// The window is advisory provenance: coverage *truth* is `shown_ids`. + /// Signals timestamped inside the window that were dropped to fit the + /// budget stay pending — the next plan filters by id, not by window. pub coverage_until: Option, + /// The exact selection this version was planned from. A later selection + /// change makes the cached-run shortcut miss, so edits always re-fold. + #[serde(default)] + pub selection: Selection, + /// Every channel any version in this chain has ever read (sorted union + /// over the chain). Sharing is taint-checked against this, not the live + /// spec: an artifact that ever folded another channel's events never + /// leaks them into a single-channel share. + #[serde(default)] + pub channels: Vec, /// Model that produced this version. pub model: String, /// Schema the output conforms to. diff --git a/crates/buzz-accumulator/src/lib.rs b/crates/buzz-accumulator/src/lib.rs index bb298468b6..97920fe3ab 100644 --- a/crates/buzz-accumulator/src/lib.rs +++ b/crates/buzz-accumulator/src/lib.rs @@ -49,10 +49,8 @@ pub mod validate; pub use artifact::ArtifactPayload; pub use error::Error; -pub use estimate::Estimate; -pub use run::{complete_run, plan_run, Plan, RunPlan, MAX_CONTEXT_CHARS}; +pub use run::{complete_run, plan_run, Plan, RunPlan}; pub use runner::{FoldRunner, SubprocessRunner}; -pub use schema::ArtifactSchema; pub use selection::Selection; pub use signal::Signal; pub use spec::FoldSpec; diff --git a/crates/buzz-accumulator/src/run.rs b/crates/buzz-accumulator/src/run.rs index 2db4bb1228..79c45b123c 100644 --- a/crates/buzz-accumulator/src/run.rs +++ b/crates/buzz-accumulator/src/run.rs @@ -26,6 +26,17 @@ use crate::validate::{splice_append_sections, validate_output}; /// Hard character budget for one run's model input (prior digest + transcript). pub const MAX_CONTEXT_CHARS: usize = 120_000; +/// Most signals one run may show (and therefore seal). Bounds `shown_ids` so +/// an artifact payload cannot outgrow its encrypted relay event; the rest of +/// an oversized backlog stays pending for the next run. +pub const MAX_SHOWN_PER_RUN: usize = 250; + +/// Largest prior artifact document a run will build on. The artifact JSON +/// must fit one NIP-44-encrypted relay event (65,535-byte plaintext cap); +/// stalling here keeps that failure priced-before-spend instead of a model +/// call whose output can never be published. +pub const MAX_PRIOR_OUTPUT_BYTES: usize = 40_000; + /// The outcome of planning a run. #[derive(Debug, Clone, PartialEq)] pub enum Plan { @@ -80,11 +91,29 @@ pub fn plan_run( .filter(|s| !covered.contains(&s.id)) .collect(); let config_matches = prior.is_some_and(|p| { - p.model == spec.model && p.schema == spec.schema && p.prompt_sha256 == spec.prompt_sha256() + p.model == spec.model + && p.schema == spec.schema + && p.prompt_sha256 == spec.prompt_sha256() + && p.selection == spec.selection }); if config_matches && new.is_empty() { return Ok(Plan::Cached); } + if prior.is_none() && new.is_empty() { + return Ok(Plan::Stalled { + reason: "selection matched no signals in the window; nothing to fold yet".to_string(), + pending: 0, + }); + } + if prior.is_some_and(|p| p.output.len() > MAX_PRIOR_OUTPUT_BYTES) { + return Ok(Plan::Stalled { + reason: format!( + "standing artifact is over the {MAX_PRIOR_OUTPUT_BYTES}-byte publish ceiling; \ + compact it (tighter instructions, or start a fresh fold) before retrying" + ), + pending: new.len(), + }); + } let parent = match prior { Some(p) => format!( "--- PRIOR DIGEST ({} v{}) ---\n{}", @@ -92,18 +121,10 @@ pub fn plan_run( ), None => String::new(), }; + // A prior under the publish ceiling always leaves ample transcript budget + // (40k bytes of parent against a 120k-char context). let raw_budget = MAX_CONTEXT_CHARS.saturating_sub(parent.chars().count() + 2); - if !new.is_empty() && raw_budget == 0 { - // The standing summary alone fills the context. Calling the model - // could not cover any pending signal and would persist an - // empty-coverage version forever. - return Ok(Plan::Stalled { - reason: "standing summary exhausts the context budget; compact it before retrying" - .to_string(), - pending: new.len(), - }); - } - let render = render_transcript(&new, names, raw_budget); + let render = render_transcript(&new, names, raw_budget, MAX_SHOWN_PER_RUN); if !new.is_empty() && render.shown.is_empty() { return Ok(Plan::Stalled { reason: "no pending event fits the remaining context budget".to_string(), @@ -153,12 +174,22 @@ pub fn complete_run( let shown_ids: Vec = plan.shown.iter().map(|s| s.id.clone()).collect(); validate_output(sch, output, prior.map(|p| p.output.as_str()), &shown_ids)?; let spliced = splice_append_sections(sch, prior.map(|p| p.output.as_str()), output); + // Taint travels with the chain: once a channel's events fold in, every + // later version keeps carrying that channel, even after a selection edit. + let channels: BTreeSet = prior + .map(|p| p.channels.iter().cloned().collect::>()) + .unwrap_or_default() + .union(&spec.selection.channels.iter().cloned().collect()) + .cloned() + .collect(); Ok(ArtifactPayload { fold: spec.name.clone(), version: prior.map_or(1, |p| p.version + 1), output: spliced, coverage_since: plan.shown.iter().map(|s| s.created_at).min(), coverage_until: plan.shown.iter().map(|s| s.created_at).max().map(|t| t + 1), + selection: spec.selection.clone(), + channels: channels.into_iter().collect(), shown_ids, model: spec.model.clone(), schema: spec.schema.clone(), @@ -214,6 +245,8 @@ mod tests { shown_ids: shown.iter().map(|(id, _)| id.to_string()).collect(), coverage_since: shown.iter().map(|(_, ts)| *ts).min(), coverage_until: shown.iter().map(|(_, ts)| *ts).max().map(|t| t + 1), + selection: spec.selection.clone(), + channels: spec.selection.channels.clone(), model: spec.model.clone(), schema: spec.schema.clone(), prompt_sha256: spec.prompt_sha256(), @@ -273,10 +306,10 @@ mod tests { } #[test] - fn stalled_when_standing_summary_exhausts_budget() { + fn stalled_when_standing_artifact_exceeds_publish_ceiling() { let spec = spec(); let mut prior = artifact_v1(&spec, &[]); - prior.output = "x".repeat(MAX_CONTEXT_CHARS); + prior.output = "x".repeat(MAX_PRIOR_OUTPUT_BYTES + 1); let plan = plan_run( &spec, Some(&prior), @@ -288,10 +321,60 @@ mod tests { let Plan::Stalled { reason, pending } = plan else { panic!("expected Stalled, got Ready/Cached"); }; - assert!(reason.contains("standing summary exhausts")); + assert!(reason.contains("publish ceiling")); assert_eq!(pending, 1); } + #[test] + fn selection_change_is_not_cached() { + let mut spec = spec(); + let a = hex_id('a'); + let prior = artifact_v1(&spec, &[(&a, 100)]); + spec.selection.channels.push("ch2".to_string()); + let covered = BTreeSet::from([a]); + let plan = plan_run(&spec, Some(&prior), &covered, vec![], &BTreeMap::new()).expect("plan"); + assert!(matches!(plan, Plan::Ready(_)), "got {plan:?}"); + } + + #[test] + fn no_prior_and_no_signals_stalls_instead_of_folding_nothing() { + let spec = spec(); + let plan = plan_run(&spec, None, &BTreeSet::new(), vec![], &BTreeMap::new()).expect("plan"); + let Plan::Stalled { reason, pending } = plan else { + panic!("expected Stalled, got {plan:?}"); + }; + assert!(reason.contains("nothing to fold")); + assert_eq!(pending, 0); + } + + #[test] + fn channels_union_survives_a_selection_change() { + let mut spec = spec(); + let a = hex_id('a'); + let b = hex_id('b'); + let prior = artifact_v1(&spec, &[(&a, 100)]); + // The fold once read ch1; the selection now reads only ch2. + spec.selection.channels = vec!["ch2".to_string()]; + let covered = BTreeSet::from([a]); + let mut new_sig = sig(&b, 200, "from ch2"); + new_sig.channel = Some("ch2".to_string()); + let plan = plan_run( + &spec, + Some(&prior), + &covered, + vec![new_sig], + &BTreeMap::new(), + ) + .expect("plan"); + let Plan::Ready(run) = plan else { + panic!("expected Ready"); + }; + let out = digest(&format!("- new entry [event:{b}]")); + let v2 = complete_run(&spec, Some(&prior), &run, &out, 1_700_000_500).expect("v2"); + assert_eq!(v2.channels, vec!["ch1".to_string(), "ch2".to_string()]); + assert_eq!(v2.selection, spec.selection); + } + #[test] fn ready_plan_prices_the_exact_model_input() { let spec = spec(); diff --git a/crates/buzz-accumulator/src/runner.rs b/crates/buzz-accumulator/src/runner.rs index e10d60d865..76d79c94f4 100644 --- a/crates/buzz-accumulator/src/runner.rs +++ b/crates/buzz-accumulator/src/runner.rs @@ -40,12 +40,32 @@ const SENTINELS: &[&str] = &["not logged in", "please run /login", "invalid api /// essentially just an auth sentinel ("Not logged in"), is refused — /// a real digest may legitimately *discuss* login issues, so the sentinel /// check applies only to outputs ≤ 80 chars; -/// - a hard timeout kills the child rather than waiting forever. +/// - a hard timeout kills the child rather than waiting forever — on unix the +/// whole process group, so a wrapper script cannot leave a billed model +/// call running as an orphan; +/// - pipe drains are joined with a deadline, so a grandchild inheriting the +/// pipes cannot wedge the run after the child exits (the group is killed +/// in that case too). pub struct SubprocessRunner { binary: String, timeout: Duration, } +/// How long after child exit the pipe drains may take to finish. +const DRAIN_TIMEOUT: Duration = Duration::from_secs(5); + +/// Kill the child — on unix the entire process group it leads (see +/// `process_group(0)` below), elsewhere the direct child only. +fn kill_group(child: &mut Child) { + #[cfg(unix)] + // SAFETY: plain syscall; a negative pid signals the process group. + unsafe { + libc::kill(-(child.id() as i32), libc::SIGKILL); + } + let _ = child.kill(); + let _ = child.wait(); +} + impl Default for SubprocessRunner { fn default() -> Self { Self::new() @@ -90,6 +110,13 @@ impl SubprocessRunner { cmd.stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); + // Lead a fresh process group so a timeout kill reaches every + // descendant (the model CLI may itself be a wrapper script). + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + cmd.process_group(0); + } cmd } @@ -100,8 +127,7 @@ impl SubprocessRunner { Ok(Some(status)) => return Ok(status), Ok(None) => { if start.elapsed() > self.timeout { - let _ = child.kill(); - let _ = child.wait(); + kill_group(child); return Err(Error::Runner(format!( "fold runner timed out after {}s", self.timeout.as_secs() @@ -121,21 +147,26 @@ fn snip(s: &str) -> String { } /// Drain a child pipe to a string on a background thread (avoids pipe-buffer -/// deadlock against the concurrent stdin write). +/// deadlock against the concurrent stdin write). The result arrives on a +/// channel so the caller can bound how long it waits: a grandchild holding +/// the pipe open must not wedge the run after the child exits. fn drain( reader: Option, label: &'static str, -) -> std::thread::JoinHandle> { +) -> std::sync::mpsc::Receiver> { + let (tx, rx) = std::sync::mpsc::channel(); std::thread::spawn(move || { let mut text = String::new(); - match reader { + let result = match reader { Some(mut r) => r .read_to_string(&mut text) .map(|_| text) .map_err(|e| Error::Runner(format!("failed reading runner {label}: {e}"))), None => Ok(text), - } - }) + }; + let _ = tx.send(result); + }); + rx } impl FoldRunner for SubprocessRunner { @@ -157,12 +188,22 @@ impl FoldRunner for SubprocessRunner { let stderr = drain(child.stderr.take(), "stderr"); let status = self.wait_with_deadline(&mut child)?; let _ = writer.join(); - let out = stdout - .join() - .map_err(|_| Error::Runner("runner stdout reader panicked".to_string()))??; - let err = stderr - .join() - .map_err(|_| Error::Runner("runner stderr reader panicked".to_string()))??; + let mut recv = |rx: std::sync::mpsc::Receiver>, + label: &str| + -> Result { + rx.recv_timeout(DRAIN_TIMEOUT).map_err(|_| { + // Something (a grandchild) is still holding the pipe open + // after exit — kill the group so nothing billed lives on. + kill_group(&mut child); + Error::Runner(format!( + "runner {label} still open {}s after exit (orphaned descendant?); killed \ + the process group and treated the run as failed", + DRAIN_TIMEOUT.as_secs() + )) + })? + }; + let out = recv(stdout, "stdout")?; + let err = recv(stderr, "stderr")?; let out = out.trim().to_string(); if !status.success() { return Err(Error::Runner(format!( @@ -258,4 +299,25 @@ mod tests { let err = runner(&path).run("input", "haiku"); assert!(matches!(err, Err(Error::Runner(msg)) if msg.contains("no usable output"))); } + + #[test] + fn grandchild_holding_the_pipe_cannot_wedge_the_run() { + // The child exits immediately but leaves a background grandchild + // holding stdout open. The drain must give up on a deadline and kill + // the group instead of blocking until the grandchild finishes. + let path = stub( + "orphan-runner.sh", + "(sleep 30; echo late) & echo now; exit 0", + ); + let started = Instant::now(); + let err = runner(&path).run("input", "haiku"); + assert!( + matches!(err, Err(Error::Runner(msg)) if msg.contains("still open")), + "expected drain-deadline error" + ); + assert!( + started.elapsed() < Duration::from_secs(20), + "run should fail on the drain deadline, not the grandchild's sleep" + ); + } } diff --git a/crates/buzz-accumulator/src/schema.rs b/crates/buzz-accumulator/src/schema.rs index df7a15a232..69cef5fb6b 100644 --- a/crates/buzz-accumulator/src/schema.rs +++ b/crates/buzz-accumulator/src/schema.rs @@ -8,8 +8,7 @@ pub struct ArtifactSchema { /// Versioned schema name, e.g. `channel-digest@v1`. pub name: &'static str, - /// Required H1 sections, in order. Empty means freeform (no structural or - /// citation contract; the only dishonest freeform output is an empty one). + /// Required H1 sections, in order. pub sections: &'static [&'static str], /// Sections that accumulate: prior content is preserved by construction /// and new entries must carry signal citations. @@ -24,15 +23,8 @@ pub const CHANNEL_DIGEST_V1: ArtifactSchema = ArtifactSchema { append_sections: &["Log"], }; -/// No structural contract; output must merely be non-empty. -pub const FREEFORM_V1: ArtifactSchema = ArtifactSchema { - name: "freeform@v1", - sections: &[], - append_sections: &[], -}; - /// All built-in schemas. -pub const BUILTIN_SCHEMAS: &[&ArtifactSchema] = &[&CHANNEL_DIGEST_V1, &FREEFORM_V1]; +pub const BUILTIN_SCHEMAS: &[&ArtifactSchema] = &[&CHANNEL_DIGEST_V1]; /// Look up a built-in schema by name. pub fn builtin(name: &str) -> Option<&'static ArtifactSchema> { @@ -64,7 +56,6 @@ mod tests { builtin("channel-digest@v1").map(|s| s.name), Some("channel-digest@v1") ); - assert_eq!(builtin("freeform@v1").map(|s| s.sections.len()), Some(0)); assert!(builtin("nope@v9").is_none()); } } diff --git a/crates/buzz-accumulator/src/selection.rs b/crates/buzz-accumulator/src/selection.rs index ae1a8c4801..e6b4fbf844 100644 --- a/crates/buzz-accumulator/src/selection.rs +++ b/crates/buzz-accumulator/src/selection.rs @@ -28,9 +28,16 @@ pub struct Selection { } impl Selection { - /// Validate and canonicalize (sort + dedupe) in place, so equal selections - /// serialize identically. + /// Validate and canonicalize (lowercase + sort + dedupe) in place, so + /// equal selections serialize identically and case variants of the same + /// id cannot silently select nothing. pub fn canonicalize(&mut self) -> Result<(), Error> { + for c in &mut self.channels { + *c = c.trim().to_ascii_lowercase(); + } + for a in &mut self.authors { + *a = a.trim().to_ascii_lowercase(); + } self.channels.sort(); self.channels.dedup(); self.authors.sort(); @@ -42,15 +49,20 @@ impl Selection { "selection must name at least one channel or author".into(), )); } - if self.channels.iter().any(|c| c.trim().is_empty()) { - return Err(Error::InvalidSpec( - "selection has an empty channel id".into(), - )); + if let Some(c) = self.channels.iter().find(|c| !is_uuid(c)) { + return Err(Error::InvalidSpec(format!( + "channel {c:?} is not a channel UUID" + ))); } - if self.authors.iter().any(|a| a.trim().is_empty()) { - return Err(Error::InvalidSpec( - "selection has an empty author pubkey".into(), - )); + if let Some(a) = self.authors.iter().find(|a| !is_hex_pubkey(a)) { + return Err(Error::InvalidSpec(format!( + "author {a:?} is not a 64-hex pubkey" + ))); + } + if let Some(k) = self.kinds.iter().find(|k| **k > u16::MAX as u32) { + return Err(Error::InvalidSpec(format!( + "kind {k} is outside the event-kind range" + ))); } Ok(()) } @@ -118,6 +130,21 @@ impl Selection { } } +/// Lowercase hyphenated UUID: 8-4-4-4-12 hex digits. +fn is_uuid(s: &str) -> bool { + let b = s.as_bytes(); + b.len() == 36 + && b.iter().enumerate().all(|(i, c)| match i { + 8 | 13 | 18 | 23 => *c == b'-', + _ => c.is_ascii_hexdigit() && !c.is_ascii_uppercase(), + }) +} + +/// Lowercase 64-hex pubkey. +fn is_hex_pubkey(s: &str) -> bool { + s.len() == 64 && s.bytes().all(|c| matches!(c, b'0'..=b'9' | b'a'..=b'f')) +} + /// Deterministically materialize fetched signals: `created_at ASC, id ASC`, /// duplicates (same id) removed. /// @@ -155,6 +182,9 @@ mod tests { assert_eq!(ids, vec!["cc", "aa", "bb"]); } + const CH_A: &str = "59ca5528-71ea-4a53-a7f5-90c9fb2b1729"; + const CH_B: &str = "a2228687-0000-4000-8000-000000000000"; + #[test] fn empty_selection_is_rejected() { let mut s = Selection::default(); @@ -162,17 +192,39 @@ mod tests { } #[test] - fn canonicalize_sorts_and_dedupes() { + fn canonicalize_lowercases_sorts_and_dedupes() { let mut s = Selection { - channels: vec!["b".into(), "a".into(), "b".into()], + channels: vec![CH_B.into(), CH_A.to_uppercase(), CH_B.into()], authors: vec![], kinds: vec![40002, 9, 9], }; s.canonicalize().expect("valid"); - assert_eq!(s.channels, vec!["a", "b"]); + assert_eq!(s.channels, vec![CH_A.to_string(), CH_B.to_string()]); assert_eq!(s.kinds, vec![9, 40002]); } + #[test] + fn malformed_ids_and_out_of_range_kinds_are_rejected() { + let mut s = Selection { + channels: vec!["not-a-uuid".into()], + authors: vec![], + kinds: vec![], + }; + assert!(s.canonicalize().is_err()); + let mut s = Selection { + channels: vec![], + authors: vec!["abc123".into()], + kinds: vec![], + }; + assert!(s.canonicalize().is_err()); + let mut s = Selection { + channels: vec![CH_A.into()], + authors: vec![], + kinds: vec![70_000], + }; + assert!(s.canonicalize().is_err()); + } + #[test] fn one_filter_per_channel_with_half_open_window() { let s = Selection { diff --git a/crates/buzz-accumulator/src/spec.rs b/crates/buzz-accumulator/src/spec.rs index fb20c54d49..e06339d196 100644 --- a/crates/buzz-accumulator/src/spec.rs +++ b/crates/buzz-accumulator/src/spec.rs @@ -83,7 +83,7 @@ mod tests { FoldSpec { name: "team-digest".to_string(), selection: Selection { - channels: vec!["ch1".to_string()], + channels: vec!["59ca5528-71ea-4a53-a7f5-90c9fb2b1729".to_string()], authors: vec![], kinds: vec![], }, diff --git a/crates/buzz-accumulator/src/transcript.rs b/crates/buzz-accumulator/src/transcript.rs index 1434dd3ae4..a4b8552ead 100644 --- a/crates/buzz-accumulator/src/transcript.rs +++ b/crates/buzz-accumulator/src/transcript.rs @@ -60,15 +60,24 @@ fn render_line(signal: &Signal, names: &BTreeMap) -> String { format!("[{iso}] {who} <{}>: {text}", kind_label(signal.kind)) } +/// Smallest tail of an oversized line worth showing the model. A shorter +/// sliver (a few characters of a long message) is not honest evidence — the +/// line is dropped and stays pending rather than being sealed as covered. +pub const MIN_TAIL_CHARS: usize = 256; + /// Render `signals` (already materialized: time-ordered, deduped) into a -/// transcript of at most `max_chars` characters, newest-first fill. +/// transcript of at most `max_chars` characters and at most `max_items` +/// lines, newest-first fill. /// /// If even the single newest line exceeds the budget, its identity is kept -/// with a bounded `…`-prefixed tail rather than claiming an unseen event. +/// with a `…`-prefixed tail of at least [`MIN_TAIL_CHARS`] — when the budget +/// cannot afford even that, the line is dropped (stays pending) rather than +/// sealed on a sliver. pub fn render_transcript( signals: &[Signal], names: &BTreeMap, max_chars: usize, + max_items: usize, ) -> TranscriptRender { let rendered: Vec<(ShownSignal, String)> = signals .iter() @@ -86,21 +95,20 @@ pub fn render_transcript( let mut used = 0usize; let mut trimmed = false; for (shown, line) in rendered.iter().rev() { + if kept.len() >= max_items { + break; + } let mut line = line.clone(); let mut cost = line.chars().count() + usize::from(!kept.is_empty()); if !kept.is_empty() && used + cost > max_chars { break; } if kept.is_empty() && cost > max_chars { - if max_chars == 0 { + if max_chars < MIN_TAIL_CHARS { break; } - line = if max_chars == 1 { - "…".to_string() - } else { - let tail_start = line.chars().count() - (max_chars - 1); - format!("…{}", line.chars().skip(tail_start).collect::()) - }; + let tail_start = line.chars().count() - (max_chars - 1); + line = format!("…{}", line.chars().skip(tail_start).collect::()); cost = line.chars().count(); trimmed = true; } @@ -150,7 +158,7 @@ mod tests { sig("e1", 1_700_000_000, "hello there"), sig("e2", 1_700_000_060, "hi"), ]; - let r = render_transcript(&signals, &names, 10_000); + let r = render_transcript(&signals, &names, 10_000, usize::MAX); assert!(!r.truncated); assert_eq!(r.shown.len(), 2); assert!(r.body.contains("riley : hello there")); @@ -161,11 +169,11 @@ mod tests { fn drops_oldest_first_and_marks_truncation() { let names = BTreeMap::new(); let signals = vec![sig("e1", 100, "old old old"), sig("e2", 200, "new")]; - let newest_len = render_transcript(&signals[1..], &names, 10_000) + let newest_len = render_transcript(&signals[1..], &names, 10_000, usize::MAX) .body .chars() .count(); - let r = render_transcript(&signals, &names, newest_len); + let r = render_transcript(&signals, &names, newest_len, usize::MAX); assert!(r.truncated); assert_eq!( r.shown.iter().map(|s| s.id.as_str()).collect::>(), @@ -179,20 +187,46 @@ mod tests { #[test] fn oversized_single_line_keeps_identity_with_bounded_tail() { let names = BTreeMap::new(); - let signals = vec![sig("e1", 100, &"x".repeat(500))]; - let r = render_transcript(&signals, &names, 40); + let signals = vec![sig("e1", 100, &"x".repeat(5_000))]; + let r = render_transcript(&signals, &names, 400, usize::MAX); assert!(r.truncated); assert_eq!(r.shown.len(), 1); let last_line = r.body.lines().last().unwrap_or(""); assert!(last_line.starts_with('…')); - assert_eq!(last_line.chars().count(), 40); + assert_eq!(last_line.chars().count(), 400); + } + + #[test] + fn sliver_budget_drops_the_line_instead_of_sealing_it() { + // A tail below MIN_TAIL_CHARS is not honest evidence: show nothing, + // seal nothing — the event stays pending. + let names = BTreeMap::new(); + let signals = vec![sig("e1", 100, &"x".repeat(5_000))]; + let r = render_transcript(&signals, &names, MIN_TAIL_CHARS - 1, usize::MAX); + assert!(r.truncated); + assert!(r.shown.is_empty()); + assert_eq!(r.body, "[…older events truncated to fit context budget…]"); + } + + #[test] + fn item_cap_keeps_newest_and_marks_truncation() { + let names = BTreeMap::new(); + let signals: Vec = (0..10) + .map(|i| sig(&format!("e{i}"), 100 + i, "m")) + .collect(); + let r = render_transcript(&signals, &names, 10_000, 3); + assert!(r.truncated); + assert_eq!( + r.shown.iter().map(|s| s.id.as_str()).collect::>(), + vec!["e7", "e8", "e9"] + ); } #[test] fn zero_budget_shows_nothing_and_says_so() { let names = BTreeMap::new(); let signals = vec![sig("e1", 100, "hello")]; - let r = render_transcript(&signals, &names, 0); + let r = render_transcript(&signals, &names, 0, usize::MAX); assert!(r.truncated); assert!(r.shown.is_empty()); assert_eq!(r.body, "[…older events truncated to fit context budget…]"); @@ -203,13 +237,13 @@ mod tests { let names = BTreeMap::new(); let mut s = sig("e1", 1_700_000_000, "deployed"); s.kind = 40002; - let r = render_transcript(&[s], &names, 10_000); + let r = render_transcript(&[s], &names, 10_000, usize::MAX); assert!(r.body.contains("aabbccdd :")); } #[test] fn empty_input_is_empty_and_honest() { - let r = render_transcript(&[], &BTreeMap::new(), 100); + let r = render_transcript(&[], &BTreeMap::new(), 100, usize::MAX); assert_eq!(r.body, ""); assert!(!r.truncated); assert!(r.shown.is_empty()); diff --git a/crates/buzz-accumulator/src/validate.rs b/crates/buzz-accumulator/src/validate.rs index a386b0a4d9..f8769a85bf 100644 --- a/crates/buzz-accumulator/src/validate.rs +++ b/crates/buzz-accumulator/src/validate.rs @@ -82,8 +82,11 @@ pub fn headings(output: &str) -> Vec { /// jam several full ids together. /// /// `[event:<64 hex>]` yields that id; `[event:, event:]` yields both. -/// A bracket without a full 64-hex id inside (shortened id, prose) is kept -/// verbatim so it still fails validation loudly. +/// A hex run splits into ids only when its length is an exact multiple of 64 — +/// a 64-hex prefix with hex garbage appended must not pass as a valid id, so +/// any other run of ≥ 64 is kept whole. A bracket without a full 64-hex id +/// inside (shortened id, prose) is kept verbatim. Either way the impostor +/// cannot match a real shown id, so it still fails validation loudly. pub fn cited_event_ids(citation_text: &str) -> BTreeSet { let mut cited = BTreeSet::new(); let mut rest = citation_text; @@ -98,16 +101,22 @@ pub fn cited_event_ids(citation_text: &str) -> BTreeSet { continue; } let before = cited.len(); - // Consecutive non-overlapping 64-char windows over each maximal hex run. + // Non-overlapping 64-char windows over each maximal hex run whose + // length is an exact multiple of 64. let mut run_start: Option = None; for (i, c) in chunk.char_indices().chain([(chunk.len(), ' ')]) { if matches!(c, '0'..='9' | 'a'..='f') { run_start.get_or_insert(i); } else if let Some(s) = run_start.take() { - let mut at = s; - while i - at >= 64 { - cited.insert(chunk[at..at + 64].to_string()); - at += 64; + let len = i - s; + if len >= 64 && len % 64 == 0 { + let mut at = s; + while at < i { + cited.insert(chunk[at..at + 64].to_string()); + at += 64; + } + } else if len > 64 { + cited.insert(chunk[s..i].to_string()); } } } @@ -120,13 +129,13 @@ pub fn cited_event_ids(citation_text: &str) -> BTreeSet { /// Validate model output against the artifact contract. /// -/// - Freeform schemas (no sections) require only non-empty output. -/// - Sectioned schemas require exactly the schema's H1 sections, in order. -/// - When signals were shown (`source_event_ids` non-empty), the append -/// sections must cite ≥ 1 of them and may cite nothing else. Citations the -/// model merely re-emitted from the prior artifact's append sections are -/// stripped before checking, so only this run's citations are judged -/// against this run's shown ids. +/// - The output must have exactly the schema's H1 sections, in order. +/// - Append-section citations are always judged: every cited id must be one of +/// this run's `source_event_ids` — a run that was shown nothing may cite +/// nothing. Citations the model merely re-emitted from the prior artifact's +/// append sections are stripped first, so only this run's citations are +/// judged against this run's shown ids. +/// - When signals were shown, the append sections must cite ≥ 1 of them. /// /// Any failure is [`Error::Nonconforming`]: the caller persists nothing. pub fn validate_output( @@ -135,15 +144,6 @@ pub fn validate_output( previous_output: Option<&str>, source_event_ids: &[String], ) -> Result<(), Error> { - if schema.sections.is_empty() { - if output.trim().is_empty() { - return Err(Error::Nonconforming(format!( - "schema {:?} requires non-empty output", - schema.name - ))); - } - return Ok(()); - } let found = headings(output); if found != schema.sections { return Err(Error::Nonconforming(format!( @@ -151,45 +151,46 @@ pub fn validate_output( schema.name, schema.sections, found ))); } - if !source_event_ids.is_empty() { - if schema.append_sections.is_empty() { - return Err(Error::Nonconforming(format!( - "schema {:?} has no append section for event citations", - schema.name - ))); + if schema.append_sections.is_empty() { + if source_event_ids.is_empty() { + return Ok(()); } - let joined = |doc: &str| { - schema - .append_sections - .iter() - .map(|name| section(doc, name)) - .collect::>() - .join("\n") - }; - let citation_text = joined(output); - let prior_citation_text = joined(previous_output.unwrap_or("")); - let new_citation_text = if !prior_citation_text.is_empty() { - citation_text.replacen(&prior_citation_text, "", 1) - } else { - citation_text - }; - let cited = cited_event_ids(&new_citation_text); - let allowed: BTreeSet<&str> = source_event_ids.iter().map(String::as_str).collect(); - let fabricated: Vec<&str> = cited + return Err(Error::Nonconforming(format!( + "schema {:?} has no append section for event citations", + schema.name + ))); + } + let joined = |doc: &str| { + schema + .append_sections .iter() - .map(String::as_str) - .filter(|id| !allowed.contains(id)) - .collect(); - if !fabricated.is_empty() { - return Err(Error::Nonconforming(format!( - "append sections cite unknown source event ids: {fabricated:?}" - ))); - } - if cited.is_empty() { - return Err(Error::Nonconforming( - "append sections are missing a source event citation".into(), - )); - } + .map(|name| section(doc, name)) + .collect::>() + .join("\n") + }; + let citation_text = joined(output); + let prior_citation_text = joined(previous_output.unwrap_or("")); + let new_citation_text = if !prior_citation_text.is_empty() { + citation_text.replacen(&prior_citation_text, "", 1) + } else { + citation_text + }; + let cited = cited_event_ids(&new_citation_text); + let allowed: BTreeSet<&str> = source_event_ids.iter().map(String::as_str).collect(); + let fabricated: Vec<&str> = cited + .iter() + .map(String::as_str) + .filter(|id| !allowed.contains(id)) + .collect(); + if !fabricated.is_empty() { + return Err(Error::Nonconforming(format!( + "append sections cite event ids that were not shown this run: {fabricated:?}" + ))); + } + if !source_event_ids.is_empty() && cited.is_empty() { + return Err(Error::Nonconforming( + "append sections are missing a source event citation".into(), + )); } Ok(()) } @@ -235,7 +236,7 @@ pub fn splice_append_sections( #[cfg(test)] mod tests { use super::*; - use crate::schema::{CHANNEL_DIGEST_V1, FREEFORM_V1}; + use crate::schema::CHANNEL_DIGEST_V1; fn id(c: char) -> String { std::iter::repeat_n(c, 64).collect() @@ -269,12 +270,6 @@ mod tests { assert!(matches!(err, Err(Error::Nonconforming(_)))); } - #[test] - fn freeform_requires_only_non_empty() { - assert!(validate_output(&FREEFORM_V1, " \n", None, &[]).is_err()); - assert!(validate_output(&FREEFORM_V1, "anything", None, &[]).is_ok()); - } - #[test] fn fabricated_citation_is_refused() { let out = digest(&format!("- made up [event:{}]", id('c'))); @@ -282,6 +277,16 @@ mod tests { assert!(matches!(err, Err(Error::Nonconforming(_)))); } + #[test] + fn run_shown_nothing_may_cite_nothing() { + // A config-only rerun shows no signals: an empty Log passes… + assert!(validate_output(&CHANNEL_DIGEST_V1, &digest(""), None, &[]).is_ok()); + // …but any citation is a fabrication and is refused. + let out = digest(&format!("- invented [event:{}]", id('c'))); + let err = validate_output(&CHANNEL_DIGEST_V1, &out, None, &[]); + assert!(matches!(err, Err(Error::Nonconforming(_)))); + } + #[test] fn missing_citation_is_refused() { let out = digest("- no citation here"); @@ -305,6 +310,23 @@ mod tests { assert_eq!(short.into_iter().collect::>(), vec!["abc123"]); } + #[test] + fn hex_run_with_trailing_garbage_is_not_a_valid_prefix_id() { + // 64 valid hex chars + 6 more hex chars: must NOT yield the 64-char + // prefix (which could match a real shown id); the run is kept whole. + let run = format!("{}abcdef", id('a')); + let cited = cited_event_ids(&format!("[event:{run}]")); + assert!(!cited.contains(&id('a'))); + assert!(cited.contains(&run)); + let err = validate_output( + &CHANNEL_DIGEST_V1, + &digest(&format!("- padded [event:{run}]")), + None, + &[id('a')], + ); + assert!(matches!(err, Err(Error::Nonconforming(_)))); + } + #[test] fn prior_citations_do_not_satisfy_a_new_run() { let prior = digest(&format!("- old entry [event:{}]", id('a'))); diff --git a/crates/buzz-cli/src/commands/folds.rs b/crates/buzz-cli/src/commands/folds.rs index 96a0817faf..3080018907 100644 --- a/crates/buzz-cli/src/commands/folds.rs +++ b/crates/buzz-cli/src/commands/folds.rs @@ -41,6 +41,11 @@ use crate::validate::read_or_stdin; /// would re-show already-folded history; `query_all_bounded` errors instead. const MAX_ARTIFACT_CHAIN: u32 = 2_000; +/// Largest artifact JSON the CLI will publish. NIP-44 v2 caps plaintext at +/// 65,535 bytes; staying under it with headroom keeps "encrypt then publish" +/// a step that cannot fail after the model spend. +const MAX_ARTIFACT_PLAINTEXT: usize = 60_000; + fn now_secs() -> i64 { SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) @@ -157,29 +162,37 @@ async fn load_all_specs(client: &BuzzClient) -> Result, CliError> }); let events = client.query_all_bounded(filter, 500).await?; let mut specs = Vec::new(); + let mut skipped = 0usize; for ev in &events { let Some(content) = event_field(ev, "content") else { + skipped += 1; continue; }; // Skip undecryptable or malformed entries rather than failing the // whole listing; `get` on the specific name surfaces the error. let Ok(plain) = decrypt_from_self(client, content) else { + skipped += 1; continue; }; - if let Ok(Some(spec)) = parse_spec_envelope(&plain) { - specs.push(spec); + match parse_spec_envelope(&plain) { + Ok(Some(spec)) => specs.push(spec), + Ok(None) => {} + Err(_) => skipped += 1, } } + if skipped > 0 { + eprintln!("warning: skipped {skipped} unreadable fold-spec event(s)"); + } specs.sort_by(|a, b| a.name.cmp(&b.name)); Ok(specs) } /// Load the full artifact chain for `fold`, sorted by version ascending. /// -/// Artifact events carry a plaintext `f` tag, but generic-tag filters are not -/// pushed to storage for regular kinds, so this reads the caller's whole -/// kind-4640 set (bounded, erroring on overflow rather than silently -/// truncating the covered-ids union) and filters by payload. +/// Artifact events carry no plaintext tags (the fold name stays private), so +/// this reads the caller's whole kind-4640 set (bounded, erroring on overflow +/// rather than silently truncating the covered-ids union) and filters by the +/// decrypted payload's `fold` field. async fn load_artifacts(client: &BuzzClient, fold: &str) -> Result, CliError> { let me = client.keys().public_key().to_hex(); let filter = serde_json::json!({ @@ -189,24 +202,27 @@ async fn load_artifacts(client: &BuzzClient, fold: &str) -> Result(&plain) else { + skipped += 1; continue; }; if payload.fold == fold { artifacts.push(payload); } } + if skipped > 0 { + eprintln!("warning: skipped {skipped} unreadable artifact event(s)"); + } artifacts.sort_by_key(|a| a.version); Ok(artifacts) } @@ -223,7 +239,11 @@ fn chain_state(artifacts: &[ArtifactPayload]) -> (Option<&ArtifactPayload>, BTre /// Fetch the signals a selection matches over `[since, until_exclusive)`. /// /// One relay query per compiled filter (multi-channel union filters are not -/// reliable); the engine dedupes and orders during materialization. +/// reliable), each read exhaustively up to `limit` — a window with more +/// matches errors loudly (narrow `--since`/`--until` or raise `--limit`) +/// instead of silently keeping only the newest page, which would leave older +/// backlog permanently unfolded while reporting the fold as covered. The +/// engine dedupes and orders during materialization. async fn fetch_signals( client: &BuzzClient, selection: &Selection, @@ -233,7 +253,7 @@ async fn fetch_signals( ) -> Result, CliError> { let mut signals = Vec::new(); for filter in selection.compile_filters(since, until_exclusive, limit as usize) { - let events = client.query_paginated(filter, limit).await?; + let events = client.query_all_bounded(filter, limit).await?; for ev in &events { let (Some(id), Some(pubkey)) = (event_field(ev, "id"), event_field(ev, "pubkey")) else { @@ -337,6 +357,10 @@ struct Preflight { spec: FoldSpec, prior: Option, plan: Plan, + /// The exact half-open window this plan drew from. `estimate` reports it + /// so a later `run --since … --until …` can replay the priced set. + since: i64, + until_exclusive: i64, } /// Load spec + artifact chain, fetch pending signals, and plan the run. @@ -361,7 +385,13 @@ async fn preflight( let names = fetch_names(client, &authors).await; let plan = plan_run(&spec, prior.as_ref(), &covered, fetched, &names) .map_err(|e| CliError::Other(e.to_string()))?; - Ok(Preflight { spec, prior, plan }) + Ok(Preflight { + spec, + prior, + plan, + since, + until_exclusive, + }) } fn plan_json(name: &str, pre: &Preflight) -> serde_json::Value { @@ -389,34 +419,28 @@ fn plan_json(name: &str, pre: &Preflight) -> serde_json::Value { "est_cost_usd": run.estimate.est_cost_usd, "window_fit": run.estimate.window_fit, "next_version": pre.prior.as_ref().map_or(1, |p| p.version + 1), + // Replay window: `run --since --until ` + // executes the same fetch this estimate priced. + "since": pre.since, + "until_exclusive": pre.until_exclusive, }), } } // --- subcommands ---------------------------------------------------------------- -#[allow(clippy::too_many_arguments)] async fn cmd_set( client: &BuzzClient, name: String, channels: Vec, authors: Vec, kinds: Vec, - schema: String, model: String, instructions: Option, ) -> Result<(), CliError> { let instructions = match instructions { Some(value) => read_or_stdin(&value)?, - None if schema == buzz_accumulator::schema::CHANNEL_DIGEST_V1.name => { - buzz_accumulator::schema::CHANNEL_DIGEST_PROMPT.to_string() - } - None => { - return Err(CliError::Usage(format!( - "--instructions is required for schema {schema:?} (only {:?} has a default prompt)", - buzz_accumulator::schema::CHANNEL_DIGEST_V1.name - ))); - } + None => buzz_accumulator::schema::CHANNEL_DIGEST_PROMPT.to_string(), }; let mut spec = FoldSpec { name: name.clone(), @@ -425,7 +449,7 @@ async fn cmd_set( authors, kinds, }, - schema, + schema: buzz_accumulator::schema::CHANNEL_DIGEST_V1.name.to_string(), model, instructions, }; @@ -555,21 +579,80 @@ async fn cmd_run( let output = runner .run(&run.model_input, &pre.spec.model) .map_err(|e| CliError::Other(e.to_string()))?; - // Refusal path: nonconforming output errors here and nothing persists. - let artifact = complete_run(&pre.spec, pre.prior.as_ref(), run, &output, now_secs()) - .map_err(|e| CliError::Other(e.to_string()))?; - let plaintext = serde_json::to_string(&artifact) - .map_err(|e| CliError::Other(format!("artifact serialization failed: {e}")))?; - let ciphertext = encrypt_to_self(client, &plaintext)?; - let tags = [ - Tag::parse(["f", &artifact.fold]) - .map_err(|e| CliError::Other(format!("failed to build f tag: {e}")))?, - Tag::parse(["v", &artifact.version.to_string()]) - .map_err(|e| CliError::Other(format!("failed to build v tag: {e}")))?, - ]; - let builder = EventBuilder::new(Kind::Custom(KIND_FOLD_ARTIFACT as u16), ciphertext).tags(tags); - let event = client.sign_event(builder)?; - let event_id = submit_checked(client, event, "artifact write was reported duplicate").await?; + // The model spend is behind us: from here on, any failure salvages the + // paid-for output to stdout instead of discarding it with the error. + let salvage = |action: &str, reason: &str| { + println!( + "{}", + serde_json::json!({ + "fold": name, + "action": action, + "reason": reason, + "accepted": false, + "model_output": output, + }) + ); + }; + // Refusal path: nonconforming output persists nothing. + let artifact = match complete_run(&pre.spec, pre.prior.as_ref(), run, &output, now_secs()) { + Ok(artifact) => artifact, + Err(e) => { + salvage("refused", &e.to_string()); + return Err(CliError::Other(e.to_string())); + } + }; + let plaintext = match serde_json::to_string(&artifact) { + Ok(p) => p, + Err(e) => { + let reason = format!("artifact serialization failed: {e}"); + salvage("unpublished", &reason); + return Err(CliError::Other(reason)); + } + }; + if plaintext.len() > MAX_ARTIFACT_PLAINTEXT { + let reason = format!( + "artifact JSON is {} bytes, over the {MAX_ARTIFACT_PLAINTEXT}-byte encrypted-event \ + ceiling; compact the fold before running again", + plaintext.len() + ); + salvage("unpublished", &reason); + return Err(CliError::Other(reason)); + } + // Version fence: if a concurrent run published while the model was + // thinking, abort instead of forking the chain at the same version. + let head = match load_artifacts(client, &name).await { + Ok(chain) => chain.last().map(|a| a.version), + Err(e) => { + salvage( + "unpublished", + &format!("pre-publish chain re-read failed: {e}"), + ); + return Err(e); + } + }; + if head.is_some_and(|v| v >= artifact.version) { + let reason = format!( + "a concurrent run already published v{} for this fold; nothing persisted — \ + re-run to fold what is still pending", + head.unwrap_or_default() + ); + salvage("unpublished", &reason); + return Err(CliError::Other(reason)); + } + // No plaintext tags: the fold name lives only inside the encrypted payload. + let publish = async { + let ciphertext = encrypt_to_self(client, &plaintext)?; + let builder = EventBuilder::new(Kind::Custom(KIND_FOLD_ARTIFACT as u16), ciphertext); + let event = client.sign_event(builder)?; + submit_checked(client, event, "artifact write was reported duplicate").await + }; + let event_id = match publish.await { + Ok(id) => id, + Err(e) => { + salvage("unpublished", &format!("artifact publish failed: {e}")); + return Err(e); + } + }; println!( "{}", serde_json::json!({ @@ -638,25 +721,22 @@ async fn cmd_artifact( } /// The share taint rule: an artifact may be republished into a channel only -/// when its selection reads from exactly that one channel — then every cited -/// signal is already visible to the audience by construction. Author/kind -/// narrowing within the channel is fine; a second channel or a channel-less -/// selection is not. -fn share_allowed(selection: &Selection, channel: &str) -> bool { - selection.channels.len() == 1 && selection.channels[0] == channel +/// when every channel that ever fed its chain (`artifact.channels`, the +/// union over all versions) is exactly that one channel — then every signal +/// that ever folded in is already visible to the audience by construction. +/// The check reads the artifact's provenance, not the live spec: a selection +/// edit cannot launder history folded from elsewhere. Author/kind narrowing +/// within the channel is fine; a second channel, a channel-less selection, +/// or a pre-provenance artifact is not. +fn share_allowed(artifact: &ArtifactPayload, channel: &str) -> bool { + artifact.channels.len() == 1 && artifact.channels[0] == channel } async fn cmd_share(client: &BuzzClient, name: String, channel: String) -> Result<(), CliError> { - let spec = load_spec(client, &name) - .await? - .ok_or_else(|| CliError::NotFound(format!("no fold named {name:?}")))?; - if !share_allowed(&spec.selection, &channel) { - return Err(CliError::Usage(format!( - "refusing to share: fold {name:?} reads {}, not exactly channel {channel} — \ - sharing is only allowed into the single channel the fold reads from, so every \ - cited signal is already visible to the audience", - spec.selection.describe(), - ))); + // Provenance channels are canonicalized lowercase; match that. + let channel = channel.trim().to_ascii_lowercase(); + if load_spec(client, &name).await?.is_none() { + return Err(CliError::NotFound(format!("no fold named {name:?}"))); } let artifacts = load_artifacts(client, &name).await?; let artifact = artifacts.last().ok_or_else(|| { @@ -664,6 +744,15 @@ async fn cmd_share(client: &BuzzClient, name: String, channel: String) -> Result "fold {name:?} has no artifact yet (run `buzz folds run {name}`)" )) })?; + if !share_allowed(artifact, &channel) { + return Err(CliError::Usage(format!( + "refusing to share: this artifact's history folded events from channel(s) {:?}, \ + not exactly channel {channel} — sharing is only allowed into the single channel \ + an artifact has ever read from, so everything in it is already visible to the \ + audience", + artifact.channels, + ))); + } let channel_uuid = channel .parse::() .map_err(|e| CliError::Usage(format!("--channel must be a channel UUID: {e}")))?; @@ -699,22 +788,9 @@ pub async fn dispatch(cmd: crate::FoldsCmd, client: &BuzzClient) -> Result<(), C channel, author, kind, - schema, model, instructions, - } => { - cmd_set( - client, - name, - channel, - author, - kind, - schema, - model, - instructions, - ) - .await - } + } => cmd_set(client, name, channel, author, kind, model, instructions).await, FoldsCmd::List => cmd_list(client).await, FoldsCmd::Get { name } => cmd_get(client, name).await, FoldsCmd::Delete { name } => cmd_delete(client, name).await, @@ -752,14 +828,37 @@ mod tests { } } + fn artifact_with_channels(channels: &[&str]) -> ArtifactPayload { + ArtifactPayload { + fold: "x".into(), + version: 1, + output: "# Working Context\n\nS.\n\n# Log\n".into(), + shown_ids: vec![], + coverage_since: None, + coverage_until: None, + selection: selection(channels, &[]), + channels: channels.iter().map(|s| s.to_string()).collect(), + model: "haiku".into(), + schema: "channel-digest@v1".into(), + prompt_sha256: "0".repeat(64), + truncated: false, + created_at: 0, + } + } + #[test] - fn share_requires_exactly_the_target_channel() { - assert!(share_allowed(&selection(&["ch1"], &[]), "ch1")); - // Author-narrowing within the channel keeps taint ⊆ {channel}. - assert!(share_allowed(&selection(&["ch1"], &["pk1"]), "ch1")); - assert!(!share_allowed(&selection(&["ch1"], &[]), "ch2")); - assert!(!share_allowed(&selection(&["ch1", "ch2"], &[]), "ch1")); - assert!(!share_allowed(&selection(&[], &["pk1"]), "ch1")); + fn share_requires_exactly_the_target_channel_in_provenance() { + assert!(share_allowed(&artifact_with_channels(&["ch1"]), "ch1")); + assert!(!share_allowed(&artifact_with_channels(&["ch1"]), "ch2")); + // A chain that EVER read a second channel stays untouchable, no + // matter what the live spec reads now. + assert!(!share_allowed( + &artifact_with_channels(&["ch1", "ch2"]), + "ch1" + )); + // Pre-provenance or author-only artifacts have no single-channel + // proof: blocked. + assert!(!share_allowed(&artifact_with_channels(&[]), "ch1")); } #[test] diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index ac5f6015b6..a195d97d56 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1843,14 +1843,11 @@ pub enum FoldsCmd { /// Event kind to include (repeatable; default 9 = channel messages) #[arg(long = "kind")] kind: Vec, - /// Artifact schema: channel-digest@v1 or freeform@v1 - #[arg(long, default_value = "channel-digest@v1")] - schema: String, /// Model handed to the fold runner (e.g. haiku, sonnet, opus) #[arg(long, default_value = "haiku")] model: String, - /// Fold instructions; pass `-` to read from stdin. Only - /// channel-digest@v1 has a built-in default prompt. + /// Fold instructions; pass `-` to read from stdin (defaults to the + /// built-in channel-digest prompt) #[arg(long)] instructions: Option, }, @@ -1869,7 +1866,8 @@ pub enum FoldsCmd { /// Only consider signals created strictly before this unix timestamp #[arg(long)] until: Option, - /// Max signals fetched per compiled selection filter + /// Exhaustive-read bound per compiled selection filter; a window with + /// more matches errors (narrow --since/--until or raise this) #[arg(long, default_value_t = 500)] limit: u32, }, @@ -1882,7 +1880,8 @@ pub enum FoldsCmd { /// Only consider signals created strictly before this unix timestamp #[arg(long)] until: Option, - /// Max signals fetched per compiled selection filter + /// Exhaustive-read bound per compiled selection filter; a window with + /// more matches errors (narrow --since/--until or raise this) #[arg(long, default_value_t = 500)] limit: u32, }, @@ -1902,8 +1901,9 @@ pub enum FoldsCmd { /// Republish the latest artifact as a message in the fold's channel Share { name: String, - /// Target channel UUID — must be exactly the one channel the fold - /// reads from, so every cited signal is already visible there + /// Target channel UUID — must be exactly the one channel the + /// artifact's whole history has read from, so everything in it is + /// already visible there #[arg(long)] channel: String, }, diff --git a/desktop/src-tauri/src/commands/folds_exec.rs b/desktop/src-tauri/src/commands/folds_exec.rs index cbe2bf36da..e1d60a6ab1 100644 --- a/desktop/src-tauri/src/commands/folds_exec.rs +++ b/desktop/src-tauri/src/commands/folds_exec.rs @@ -28,9 +28,11 @@ const MAX_ARGS: usize = 64; const MAX_ARG_LEN: usize = 16 * 1024; /// `folds run` invokes a model subprocess (the fold runner's own timeout is -/// 600s); every other verb is relay I/O only. -const RUN_TIMEOUT: Duration = Duration::from_secs(660); -const QUERY_TIMEOUT: Duration = Duration::from_secs(60); +/// 600s, plus relay reads and the publish); every other verb is relay I/O +/// only. Both leave generous headroom over the inner timeouts so the CLI's +/// own error reporting wins the race against this outer kill. +const RUN_TIMEOUT: Duration = Duration::from_secs(900); +const QUERY_TIMEOUT: Duration = Duration::from_secs(120); #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -75,6 +77,18 @@ fn resolve_buzz_cli() -> Result { .ok_or_else(|| "buzz CLI not found".to_string()) } +/// Kill the child — on unix the entire process group it leads, so the model +/// subprocess a `folds run` spawned dies with it. +fn kill_group(child: &mut std::process::Child) { + #[cfg(unix)] + // SAFETY: plain syscall; a negative pid signals the process group. + unsafe { + libc::kill(-(child.id() as i32), libc::SIGKILL); + } + let _ = child.kill(); + let _ = child.wait(); +} + fn read_pipe_lossy(pipe: Option) -> String { let Some(mut pipe) = pipe else { return String::new(); @@ -101,14 +115,30 @@ fn run_folds_cli_blocking( command.arg("folds").args(&args); command.env("BUZZ_PRIVATE_KEY", &nsec); command.env("BUZZ_RELAY_URL", &relay_url); + // Scrub ambient credentials the desktop process may carry: an inherited + // BUZZ_AUTH_TAG would be silently injected into every event this CLI + // signs (folds are personal, never membership-delegated). + command.env_remove("BUZZ_AUTH_TAG"); + command.env_remove("BUZZ_API_TOKEN"); + command.env_remove("NOSTR_PRIVATE_KEY"); // The fold runner shells out to the user's model CLI (`claude` by // default); the augmented PATH makes that resolvable from a DMG launch. if let Some(path) = crate::managed_agents::readiness::cli_probe::augmented_path() { command.env("PATH", path); } + if let Some(home) = dirs::home_dir() { + command.current_dir(home); + } command.stdin(Stdio::null()); command.stdout(Stdio::piped()); command.stderr(Stdio::piped()); + // Lead a fresh process group so a timeout kill reaches the model CLI the + // fold runner spawned, not just the `buzz` wrapper. + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + command.process_group(0); + } crate::util::configure_no_window(&mut command); let mut child = command @@ -128,8 +158,7 @@ fn run_folds_cli_blocking( Ok(Some(status)) => break status, Ok(None) => { if started.elapsed() > timeout { - let _ = child.kill(); - let _ = child.wait(); + kill_group(&mut child); let _ = stdout_thread.join(); let _ = stderr_thread.join(); return Err(format!("buzz folds timed out after {}s", timeout.as_secs())); @@ -137,8 +166,7 @@ fn run_folds_cli_blocking( std::thread::sleep(Duration::from_millis(50)); } Err(error) => { - let _ = child.kill(); - let _ = child.wait(); + kill_group(&mut child); return Err(format!("failed to wait for buzz folds: {error}")); } } diff --git a/desktop/src/features/accumulator/ui/AccumulatorSettingsCard.tsx b/desktop/src/features/accumulator/ui/AccumulatorSettingsCard.tsx index 59545c9e31..9cf1134e45 100644 --- a/desktop/src/features/accumulator/ui/AccumulatorSettingsCard.tsx +++ b/desktop/src/features/accumulator/ui/AccumulatorSettingsCard.tsx @@ -36,6 +36,10 @@ type FoldPlan = { next_version?: number; shown_signals?: number; coverage_until?: number | null; + // Ready plans carry the exact half-open window the estimate priced, so Run + // can replay the same fetch instead of racing new arrivals. + since?: number; + until_exclusive?: number; }; const foldsListQueryKey = ["accumulator", "folds"] as const; @@ -85,7 +89,20 @@ function FoldRow({ fold }: { fold: FoldSummary }) { }); const runMutation = useMutation({ - mutationFn: () => runFoldsCliJson(["run", fold.fold]), + // Replay exactly the window the preflight priced — arrivals since then + // stay pending for the next preflight instead of silently joining a run + // priced without them. + mutationFn: (priced: FoldPlan) => + runFoldsCliJson([ + "run", + fold.fold, + ...(priced.since !== undefined + ? ["--since", String(priced.since)] + : []), + ...(priced.until_exclusive !== undefined + ? ["--until", String(priced.until_exclusive)] + : []), + ]), onSuccess: async (result) => { setPlan(result); setError(null); @@ -96,7 +113,12 @@ function FoldRow({ fold }: { fold: FoldSummary }) { }), ]); }, - onError: (mutationError: Error) => setError(mutationError.message), + onError: (mutationError: Error) => { + // A failed run invalidates the priced plan: Run re-locks until the + // next preflight rather than offering a stale "ready". + setPlan(null); + setError(mutationError.message); + }, }); const busy = estimateMutation.isPending || runMutation.isPending; @@ -160,7 +182,7 @@ function FoldRow({ fold }: { fold: FoldSummary }) {