jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-05 17:01:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(accumulator): harden engine, CLI, and desktop seam after adversarial review

Browse Source 
Engine (buzz-accumulator):
- Citations are always judged: a run shown no signals may cite nothing,
  so fabricated citations can no longer ride config-only reruns.
- Hex runs split into event ids only at exact multiples of 64 chars;
  a valid-id prefix with hex garbage appended now fails loudly.
- Oversized transcript lines keep a >=256-char tail or stay pending;
  a sliver can no longer seal an event as covered.
- Runs cap shown signals (250) and stall when the standing artifact
  exceeds a 40k-byte publish ceiling, keeping the NIP-44 65,535-byte
  plaintext cap a priced-before-spend stall instead of a post-spend
  publish failure. Empty selections with no prior stall too.
- Artifacts carry their selection and the chain-wide channel union;
  the cached-run shortcut now misses on selection edits.
- The subprocess runner leads a process group and group-kills on
  timeout, and pipe drains are joined with a deadline so a grandchild
  cannot orphan a billed model call or wedge the run.
- Selections canonicalize lowercase and validate UUID/pubkey/kind
  shapes; the unused freeform@v1 schema is gone.

CLI (buzz folds):
- share checks the artifact's channel provenance, not the live spec:
  history folded from another channel can never be laundered into a
  single-channel share by editing the selection.
- Artifact events publish without plaintext tags (fold names private).
- Signal fetches are exhaustive within the window and error on the
  --limit bound instead of silently keeping only the newest page.
- run salvages the paid-for model output to stdout on every
  post-spend failure (refusal, size ceiling, version fence, publish),
  and aborts before forking the chain if a concurrent run published.
- estimate reports the priced window so run can replay it exactly;
  the redundant --schema flag is gone.

Desktop:
- run_folds_cli scrubs ambient BUZZ_AUTH_TAG/BUZZ_API_TOKEN/
  NOSTR_PRIVATE_KEY, runs from home, leads a process group, and uses
  900s/120s timeouts with headroom over the CLI's own.
- The settings card pins Run to the preflight's priced window and
  re-locks Run when a run fails; dead kind consts removed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Riley Crane <rileycrane@squareup.com>
This commit is contained in:
Riley Crane
2026-08-28 17:29:24 -07:00
committed by Sol
parent 84e86f085f
commit 6c8b1c865a
17 changed files with 663 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -830,6 +830,7 @@ name = "buzz-accumulator"
version = "0.1.0"
dependencies = [
"chrono",
"libc",
"serde",
"serde_json",
"sha2 0.11.0",

3
crates/buzz-accumulator/Cargo.toml
View File

@@ -13,3 +13,6 @@ serde_json = { workspace = true }
thiserror = { workspace = true }
sha2 = { workspace = true }
chrono = { workspace = true }
[target.'cfg(unix)'.dependencies]
libc = "0.2"

16
crates/buzz-accumulator/src/artifact.rs
View File

@@ -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<i64>,
/// 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<i64>,
/// 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<String>,
/// Model that produced this version.
pub model: String,
/// Schema the output conforms to.

4
crates/buzz-accumulator/src/lib.rs
View File

@@ -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;

113
crates/buzz-accumulator/src/run.rs
View File

@@ -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<String> = 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<String> = prior
.map(|p| p.channels.iter().cloned().collect::<BTreeSet<_>>())
.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();

90
crates/buzz-accumulator/src/runner.rs
View File

@@ -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<R: Read + Send + 'static>(
reader: Option<R>,
label: &'static str,
) -> std::thread::JoinHandle<Result<String, Error>> {
) -> std::sync::mpsc::Receiver<Result<String, Error>> {
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<Result<String, Error>>,
label: &str|
-> Result<String, Error> {
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"
);
}
}

13
crates/buzz-accumulator/src/schema.rs
View File

@@ -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());
}
}

78
crates/buzz-accumulator/src/selection.rs
View File

@@ -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 {

2
crates/buzz-accumulator/src/spec.rs
View File

@@ -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![],
},

70
crates/buzz-accumulator/src/transcript.rs
View File

@@ -60,15 +60,24 @@ fn render_line(signal: &Signal, names: &BTreeMap<String, String>) -> 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<String, String>,
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::<String>())
};
let tail_start = line.chars().count() - (max_chars - 1);
line = format!("…{}", line.chars().skip(tail_start).collect::<String>());
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 <message>: 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::<Vec<_>>(),
@@ -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<Signal> = (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<_>>(),
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 <kind-40002>:"));
}
#[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());

156
crates/buzz-accumulator/src/validate.rs
View File

@@ -82,8 +82,11 @@ pub fn headings(output: &str) -> Vec<String> {
/// jam several full ids together.
///
/// `[event:<64 hex>]` yields that id; `[event:<id>, event:<id>]` 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<String> {
let mut cited = BTreeSet::new();
let mut rest = citation_text;
@@ -98,16 +101,22 @@ pub fn cited_event_ids(citation_text: &str) -> BTreeSet<String> {
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<usize> = 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<String> {
/// 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::<Vec<_>>()
.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::<Vec<_>>()
.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<_>>(), 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')));

253
crates/buzz-cli/src/commands/folds.rs
View File

@@ -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<Vec<FoldSpec>, 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<Vec<ArtifactPayload>, 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<Vec<ArtifactP
});
let events = client.query_all_bounded(filter, MAX_ARTIFACT_CHAIN).await?;
let mut artifacts = Vec::new();
let mut skipped = 0usize;
for ev in &events {
// Cheap pre-filter on the plaintext tag; payload is authoritative.
if event_tag(ev, "f").is_some_and(|f| f != fold) {
continue;
}
let Some(content) = event_field(ev, "content") else {
skipped += 1;
continue;
};
let Ok(plain) = decrypt_from_self(client, content) else {
skipped += 1;
continue;
};
let Ok(payload) = serde_json::from_str::<ArtifactPayload>(&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<Vec<Signal>, 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<ArtifactPayload>,
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 <since> --until <until_exclusive>`
// 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<String>,
authors: Vec<String>,
kinds: Vec<u32>,
schema: String,
model: String,
instructions: Option<String>,
) -> 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::<uuid::Uuid>()
.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]

18
crates/buzz-cli/src/lib.rs
View File

@@ -1843,14 +1843,11 @@ pub enum FoldsCmd {
/// Event kind to include (repeatable; default 9 = channel messages)
#[arg(long = "kind")]
kind: Vec<u32>,
/// 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<String>,
},
@@ -1869,7 +1866,8 @@ pub enum FoldsCmd {
/// Only consider signals created strictly before this unix timestamp
#[arg(long)]
until: Option<i64>,
/// 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<i64>,
/// 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,
},

42
desktop/src-tauri/src/commands/folds_exec.rs
View File

@@ -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<std::path::PathBuf, String> {
.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<impl Read>) -> 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}"));
}
}

41
desktop/src/features/accumulator/ui/AccumulatorSettingsCard.tsx
View File

@@ -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<FoldPlan>(["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<FoldPlan>([
"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 }) {
<Button
data-testid={`accumulator-run-${fold.fold}`}
disabled={!runReady}
onClick={() => runMutation.mutate()}
onClick={() => plan && runMutation.mutate(plan)}
size="sm"
>
{runMutation.isPending
@@ -233,19 +255,6 @@ export function AccumulatorSettingsCard() {
))
)}
</SettingsOptionGroup>
<SettingsOptionGroup title="Automation">
<SettingsOptionRow>
<p
className="text-sm font-normal text-muted-foreground/70"
data-settings-subcopy
>
Everything here is also available to agents and scripts as the
`buzz folds` CLI (set, list, get, delete, estimate, run, artifact,
share). Scheduled runs are a workflow concern — folds deliberately
carry no cadence of their own.
</p>
</SettingsOptionRow>
</SettingsOptionGroup>
</SettingsOptionGroupList>
</section>
);

18
desktop/src/shared/api/tauriFolds.ts
View File

@@ -18,14 +18,24 @@ export async function runFoldsCli(args: string[]): Promise<FoldsCliOutput> {
/**
* Run a folds verb that prints JSON on success and parse it. Throws with the
* CLI's stderr (which carries the actionable message) on nonzero exit.
* CLI's stderr (which carries the actionable message) on nonzero exit. A
* failed `run` may put a structured salvage report on stdout (the paid-for
* model output that was refused or could not be published); its reason is
* folded into the thrown message so the UI surfaces why.
*/
export async function runFoldsCliJson<T>(args: string[]): Promise<T> {
const result = await runFoldsCli(args);
if (result.exitCode !== 0) {
throw new Error(
result.stderr.trim() || `buzz folds exited with code ${result.exitCode}`,
);
let detail = result.stderr.trim();
try {
const salvage = JSON.parse(result.stdout) as { reason?: string };
if (salvage.reason && !detail.includes(salvage.reason)) {
detail = detail ? `${detail} (${salvage.reason})` : salvage.reason;
}
} catch {
// stdout was not a salvage report; stderr already tells the story.
}
throw new Error(detail || `buzz folds exited with code ${result.exitCode}`);
}
return JSON.parse(result.stdout) as T;
}

6
desktop/src/shared/constants/kinds.ts
View File

@@ -82,12 +82,6 @@ export const KIND_GIT_STATUS_DRAFT = 1633;
// NIP-DV: relay-signed per-viewer DM visibility snapshot (d=viewer pubkey,
// h-tags = currently-hidden DM channel ids).
export const KIND_DM_VISIBILITY = 30622;
// Accumulator folds. Mirror of buzz-core's KIND_FOLD_SPEC/KIND_FOLD_ARTIFACT.
// Both author-only and NIP-44 encrypted to self; the desktop reads them via
// the bundled `buzz folds` CLI, never directly.
export const KIND_FOLD_SPEC = 30640;
export const KIND_FOLD_ARTIFACT = 4640;
// Human-visible "new content" message kinds. Used as the unread trigger set
// (sidebar badges, catch-up queries) and as the Home-feed mention query.
// Reactions, edits, diffs, deletions, and system messages are deliberately