From 62f0d68761c3a6a7290a157b7368758dc8ffd2d8 Mon Sep 17 00:00:00 2001 From: Riley Crane Date: Fri, 28 Aug 2026 16:06:50 -0700 Subject: [PATCH] feat(accumulator): fold engine crate + fold event kinds (30640, 4640) A1 of the accumulator experiment: incrementally maintained, provenance- carrying artifacts over the event stream (signal / selection / fold / artifact), shipped as a packaged experiment. - New pure-logic workspace crate buzz-accumulator: selection-to-filter compilation with deterministic materialization, budgeted transcript rendering, zero-spend estimates (unknown model => honest None), plan_run/complete_run around a FoldRunner seam (subprocess claude -p v1 impl with sentinel/exit-code/timeout hardening), output-contract validation (exact H1 sections, cited within shown ids, >=1 citation, refuse => nothing persists), and the engine splice that retains append-section history by construction. 46 unit tests port the honesty invariants: coverage never seals unread, stall-don't-seal, prior citations never satisfy a new run. - New kinds in buzz-core: 30640 fold spec (parameterized replaceable, d = fold name) and 4640 artifact version (regular/immutable), both AUTHOR_ONLY_KINDS with NIP-44 content encrypted to self. - Relay: UsersWrite scope arms + global-only (never channel-scoped). - FTS privacy: both kinds excluded from search_tsv in schema.sql and brownfield migration 0041 (same shape as 0014/0033); embedded migrator tripwire extended. Co-Authored-By: Claude Fable 5 Signed-off-by: Riley Crane --- Cargo.lock | 11 + Cargo.toml | 1 + crates/buzz-accumulator/Cargo.toml | 15 + crates/buzz-accumulator/src/artifact.rs | 41 +++ crates/buzz-accumulator/src/error.rs | 23 ++ crates/buzz-accumulator/src/estimate.rs | 129 +++++++ crates/buzz-accumulator/src/lib.rs | 58 +++ crates/buzz-accumulator/src/run.rs | 421 ++++++++++++++++++++++ crates/buzz-accumulator/src/runner.rs | 261 ++++++++++++++ crates/buzz-accumulator/src/schema.rs | 70 ++++ crates/buzz-accumulator/src/selection.rs | 203 +++++++++++ crates/buzz-accumulator/src/signal.rs | 25 ++ crates/buzz-accumulator/src/spec.rs | 134 +++++++ crates/buzz-accumulator/src/transcript.rs | 217 +++++++++++ crates/buzz-accumulator/src/validate.rs | 371 +++++++++++++++++++ crates/buzz-core/src/kind.rs | 29 ++ crates/buzz-db/src/runtime/migration.rs | 18 +- crates/buzz-relay/src/handlers/ingest.rs | 51 +-- migrations/0045_fold_kinds_fts.sql | 39 ++ schema/schema.sql | 4 +- 20 files changed, 2095 insertions(+), 26 deletions(-) create mode 100644 crates/buzz-accumulator/Cargo.toml create mode 100644 crates/buzz-accumulator/src/artifact.rs create mode 100644 crates/buzz-accumulator/src/error.rs create mode 100644 crates/buzz-accumulator/src/estimate.rs create mode 100644 crates/buzz-accumulator/src/lib.rs create mode 100644 crates/buzz-accumulator/src/run.rs create mode 100644 crates/buzz-accumulator/src/runner.rs create mode 100644 crates/buzz-accumulator/src/schema.rs create mode 100644 crates/buzz-accumulator/src/selection.rs create mode 100644 crates/buzz-accumulator/src/signal.rs create mode 100644 crates/buzz-accumulator/src/spec.rs create mode 100644 crates/buzz-accumulator/src/transcript.rs create mode 100644 crates/buzz-accumulator/src/validate.rs create mode 100644 migrations/0045_fold_kinds_fts.sql diff --git a/Cargo.lock b/Cargo.lock index 9ad2a913e6..eee1331392 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -825,6 +825,17 @@ version = "3.20.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" +[[package]] +name = "buzz-accumulator" +version = "0.1.0" +dependencies = [ + "chrono", + "serde", + "serde_json", + "sha2 0.11.0", + "thiserror 2.0.18", +] + [[package]] name = "buzz-acp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 0af365f52f..eb40077802 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ members = [ "crates/buzz-search", "crates/buzz-audit", "crates/buzz-acp", + "crates/buzz-accumulator", "crates/buzz-agent", "crates/sprig", "crates/buzz-test-client", diff --git a/crates/buzz-accumulator/Cargo.toml b/crates/buzz-accumulator/Cargo.toml new file mode 100644 index 0000000000..5dc3d21b35 --- /dev/null +++ b/crates/buzz-accumulator/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "buzz-accumulator" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Fold engine: incrementally maintained, provenance-carrying artifacts over the event stream" + +[dependencies] +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } +sha2 = { workspace = true } +chrono = { workspace = true } diff --git a/crates/buzz-accumulator/src/artifact.rs b/crates/buzz-accumulator/src/artifact.rs new file mode 100644 index 0000000000..6e66f6389a --- /dev/null +++ b/crates/buzz-accumulator/src/artifact.rs @@ -0,0 +1,41 @@ +//! Artifact versions: the immutable, provenance-carrying output of one run. + +use serde::{Deserialize, Serialize}; + +/// One artifact version — the JSON a caller NIP-44-encrypts into the content +/// of an immutable relay event. +/// +/// Versions form a chain per fold (v1, v2, …), each produced by +/// [`crate::run::complete_run`]. The chain is the coverage ledger: the set of +/// signals a fold has ever folded is exactly the union of `shown_ids` over +/// its versions — computed from what the model was actually shown, so unread +/// signals are never sealed as covered. Append-only is physics here: relay +/// events cannot be rewritten. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ArtifactPayload { + /// Fold name this version belongs to. + pub fold: String, + /// 1-based version number; each run of the fold increments it. + pub version: u32, + /// Full artifact document (after the engine splice). + pub output: String, + /// Exactly the signal ids the model was shown this run. + pub shown_ids: Vec, + /// Earliest shown signal timestamp (unix seconds); `None` when the run + /// showed no signals (config-only rerun). + pub coverage_since: Option, + /// Half-open end of the covered window: latest shown timestamp + 1; + /// `None` when the run showed no signals. + pub coverage_until: Option, + /// Model that produced this version. + pub model: String, + /// Schema the output conforms to. + pub schema: String, + /// SHA-256 of the instructions used (provenance + cache key). + pub prompt_sha256: String, + /// True when the context budget dropped or trimmed pending signals — the + /// remainder stays pending for the next run. + pub truncated: bool, + /// Unix seconds this version was produced (caller-supplied). + pub created_at: i64, +} diff --git a/crates/buzz-accumulator/src/error.rs b/crates/buzz-accumulator/src/error.rs new file mode 100644 index 0000000000..bba0aa1f2c --- /dev/null +++ b/crates/buzz-accumulator/src/error.rs @@ -0,0 +1,23 @@ +//! Error type shared across the fold engine. + +/// Errors surfaced by the fold engine. +/// +/// A [`Error::Nonconforming`] means the model's output broke the artifact +/// contract — the run is refused and nothing may be persisted. This is a +/// deliberate design rule: never build a validator that a model can only +/// satisfy by lying; every rule enforced here is one the model can genuinely +/// meet. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// The fold spec is invalid (bad name, empty selection, unknown schema, …). + #[error("invalid fold spec: {0}")] + InvalidSpec(String), + + /// Model output does not conform to the artifact contract; nothing persists. + #[error("nonconforming model output: {0}")] + Nonconforming(String), + + /// The model runner failed (missing binary, timeout, crash, auth sentinel). + #[error("fold runner failed: {0}")] + Runner(String), +} diff --git a/crates/buzz-accumulator/src/estimate.rs b/crates/buzz-accumulator/src/estimate.rs new file mode 100644 index 0000000000..f4275b1f09 --- /dev/null +++ b/crates/buzz-accumulator/src/estimate.rs @@ -0,0 +1,129 @@ +//! Zero-spend price estimates and context-window fit. +//! +//! Estimates never call a model — by construction, not by discipline: this +//! module has no runner access. Prices and context windows are curated rather +//! than discovered; an unknown model's cost deliberately remains `None` +//! instead of becoming false precision. + +use serde::Serialize; + +/// USD per million *input* tokens (curated; update deliberately with a source). +const PRICING: &[(&str, f64)] = &[("haiku", 0.80), ("sonnet", 3.00), ("opus", 15.00)]; + +/// Documented context window for the Claude aliases above. +const CLAUDE_WINDOW: u64 = 200_000; + +/// Curated input-context capacities by model alias. +const MODEL_WINDOWS: &[(&str, u64)] = &[ + ("haiku", CLAUDE_WINDOW), + ("sonnet", CLAUDE_WINDOW), + ("opus", CLAUDE_WINDOW), +]; + +/// Whether an input estimate fits a model's curated window. +/// +/// `headroom_tokens` is intentionally negative for an overfull known window so +/// the amount of overflow is observable; both fields are `None` for a model +/// whose window is not curated. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct WindowFit { + /// Curated window size, if known. + pub model_window: Option, + /// The estimate being fitted. + pub est_input_tokens: u64, + /// Whether it fits; `None` when the window is unknown. + pub fits: Option, + /// `window - estimate`; negative when overfull, `None` when unknown. + pub headroom_tokens: Option, +} + +/// An input-only estimate for one fold run. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct Estimate { + /// chars/4 heuristic, clearly labeled as such. + pub est_input_tokens: u64, + /// `None` for a model with no curated rate — an honest unknown. + pub est_cost_usd: Option, + /// Window fit for the model. + pub window_fit: WindowFit, +} + +/// The clearly labeled chars/4 input-token heuristic. +pub fn estimate_tokens(chars: usize) -> u64 { + chars.div_ceil(4) as u64 +} + +/// Report whether an input-token estimate fits `model`'s curated window. +pub fn window_fit(model: &str, est_input_tokens: u64) -> WindowFit { + match MODEL_WINDOWS.iter().find(|(m, _)| *m == model) { + None => WindowFit { + model_window: None, + est_input_tokens, + fits: None, + headroom_tokens: None, + }, + Some((_, window)) => { + let headroom = *window as i64 - est_input_tokens as i64; + WindowFit { + model_window: Some(*window), + est_input_tokens, + fits: Some(headroom >= 0), + headroom_tokens: Some(headroom), + } + } + } +} + +/// Estimate one run's input cost; never fabricates a dollar cost for an +/// unknown model. +pub fn estimate(model: &str, chars: usize) -> Estimate { + let tokens = estimate_tokens(chars); + let fit = window_fit(model, tokens); + let cost = PRICING + .iter() + .find(|(m, _)| *m == model) + .map(|(_, rate)| (tokens as f64 * rate / 1_000_000.0 * 1e8).round() / 1e8); + Estimate { + est_input_tokens: tokens, + est_cost_usd: cost, + window_fit: fit, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tokens_round_up() { + assert_eq!(estimate_tokens(0), 0); + assert_eq!(estimate_tokens(1), 1); + assert_eq!(estimate_tokens(4), 1); + assert_eq!(estimate_tokens(5), 2); + } + + #[test] + fn known_model_prices_exactly() { + let e = estimate("haiku", 4_000_000); + assert_eq!(e.est_input_tokens, 1_000_000); + assert_eq!(e.est_cost_usd, Some(0.80)); + assert_eq!(e.window_fit.fits, Some(false)); + assert_eq!(e.window_fit.headroom_tokens, Some(200_000 - 1_000_000)); + } + + #[test] + fn unknown_model_is_honest_none() { + let e = estimate("mystery-model", 400); + assert_eq!(e.est_input_tokens, 100); + assert_eq!(e.est_cost_usd, None); + assert_eq!(e.window_fit.fits, None); + assert_eq!(e.window_fit.model_window, None); + } + + #[test] + fn fitting_input_has_positive_headroom() { + let f = window_fit("sonnet", 10); + assert_eq!(f.fits, Some(true)); + assert_eq!(f.headroom_tokens, Some(199_990)); + } +} diff --git a/crates/buzz-accumulator/src/lib.rs b/crates/buzz-accumulator/src/lib.rs new file mode 100644 index 0000000000..bb298468b6 --- /dev/null +++ b/crates/buzz-accumulator/src/lib.rs @@ -0,0 +1,58 @@ +//! Fold engine core: incrementally maintained, provenance-carrying artifacts +//! over the Buzz event stream. +//! +//! The accumulator maintains small, always-current summaries ("artifacts") of +//! selections of the event stream. Each update is a *fold*: the next artifact +//! is a function of the previous artifact plus only the signals that arrived +//! since the last run — never a re-read of history. +//! +//! Five nouns, one recurrence: +//! - **signal** — a raw relay event, never rewritten; the source of truth. +//! - **selection** — a saveable query over signals (channels, authors, kinds). +//! - **fold** — `artifact' = fold(artifact, new_signals, spec)`. +//! - **artifact** — compact sectioned output where every claim cites the exact +//! signals it came from; published back to the relay as an immutable event. +//! - (a **lens** renders artifacts into UI; rendering is out of scope here.) +//! +//! This crate is deliberately **pure**: it plans runs, prices them, renders +//! transcripts, validates model output, and splices history — but it performs +//! no relay I/O and holds no storage. Callers (the `buzz` CLI, the desktop +//! app) fetch signals and prior artifacts from the relay, hand them to +//! [`run::plan_run`], invoke a [`runner::FoldRunner`] if the plan is ready, +//! and finish with [`run::complete_run`]. All state lives in signed relay +//! events, so the artifact store is append-only by construction. +//! +//! Honesty invariants the engine enforces (ported from the X-Ray POC and +//! covered by unit tests in each module): +//! - Estimates are computed without any model call, and an unknown model's +//! cost is an honest `None`, never a guess ([`estimate`]). +//! - Coverage records exactly the signals the model was shown — an oversized +//! window stalls or truncates honestly; unread signals are never sealed +//! ([`run`]). +//! - Output that breaks the artifact contract — wrong sections, a citation of +//! a signal the model was not shown, or no citation at all — is refused and +//! nothing persists ([`validate`]). +//! - Prior append-section history is retained by the engine, not the model: +//! history cannot be rewritten regardless of model output ([`validate::splice_append_sections`]). + +pub mod artifact; +pub mod error; +pub mod estimate; +pub mod run; +pub mod runner; +pub mod schema; +pub mod selection; +pub mod signal; +pub mod spec; +pub mod transcript; +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 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 new file mode 100644 index 0000000000..2db4bb1228 --- /dev/null +++ b/crates/buzz-accumulator/src/run.rs @@ -0,0 +1,421 @@ +//! Planning and completing one fold run. +//! +//! A run is two pure halves around one model call: +//! +//! 1. [`plan_run`] — decide cached/stalled/ready, render the transcript under +//! budget, and price the *exact* string the model would receive. No model +//! call, no I/O. +//! 2. The caller shows the estimate (priced before spend), invokes a +//! [`crate::runner::FoldRunner`] with `RunPlan::model_input`, then +//! 3. [`complete_run`] — validate the output against the artifact contract, +//! splice history, and produce the next [`ArtifactPayload`]. Nonconforming +//! output returns an error and nothing persists. + +use std::collections::{BTreeMap, BTreeSet}; + +use crate::artifact::ArtifactPayload; +use crate::error::Error; +use crate::estimate::{self, Estimate}; +use crate::schema; +use crate::selection::materialize; +use crate::signal::Signal; +use crate::spec::FoldSpec; +use crate::transcript::{render_transcript, ShownSignal}; +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; + +/// The outcome of planning a run. +#[derive(Debug, Clone, PartialEq)] +pub enum Plan { + /// Nothing new and the config is unchanged — the latest artifact already + /// answers; spend nothing. + Cached, + /// The run cannot honestly proceed; nothing is spent and nothing is + /// sealed. `pending` counts the uncovered signals still waiting. + Stalled { reason: String, pending: usize }, + /// Ready to run: the exact model input and its price. + Ready(RunPlan), +} + +/// A priced, ready-to-execute run. +#[derive(Debug, Clone, PartialEq)] +pub struct RunPlan { + /// The exact string to hand the runner — previews price this same string. + pub model_input: String, + /// Exactly the signals rendered into the input, in transcript order. + pub shown: Vec, + /// True when pending signals were dropped/trimmed to fit the budget; the + /// remainder stays pending for the next run. + pub truncated: bool, + /// Zero-spend price of `model_input` for the spec's model. + pub estimate: Estimate, + /// Total uncovered signals this plan drew from (`shown.len() <=` this). + pub pending: usize, +} + +/// Plan the next run of `spec`. +/// +/// `prior` is the latest artifact version (if any), `covered` the union of +/// `shown_ids` across all prior versions, `fetched` the raw signals the +/// caller fetched for the selection's window, and `names` an optional +/// pubkey→display-name map for transcript lines. +pub fn plan_run( + spec: &FoldSpec, + prior: Option<&ArtifactPayload>, + covered: &BTreeSet, + fetched: Vec, + names: &BTreeMap, +) -> Result { + if schema::builtin(&spec.schema).is_none() { + return Err(Error::InvalidSpec(format!( + "unknown schema {:?}", + spec.schema + ))); + } + let signals = materialize(fetched); + let new: Vec = signals + .into_iter() + .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() + }); + if config_matches && new.is_empty() { + return Ok(Plan::Cached); + } + let parent = match prior { + Some(p) => format!( + "--- PRIOR DIGEST ({} v{}) ---\n{}", + spec.name, p.version, p.output + ), + None => String::new(), + }; + 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); + if !new.is_empty() && render.shown.is_empty() { + return Ok(Plan::Stalled { + reason: "no pending event fits the remaining context budget".to_string(), + pending: new.len(), + }); + } + let mut transcript = [parent.as_str(), render.body.as_str()] + .iter() + .filter(|part| !part.is_empty()) + .copied() + .collect::>() + .join("\n\n"); + if !render.shown.is_empty() { + let ids: Vec<&str> = render.shown.iter().map(|s| s.id.as_str()).collect(); + transcript.push_str("\n\n--- SOURCE EVENT IDS ---\n"); + transcript.push_str(&ids.join("\n")); + } + let model_input = format!( + "{}\n\n--- CONTEXT (time-ordered events) ---\n{}\n", + spec.instructions, transcript + ); + let est = estimate::estimate(&spec.model, model_input.chars().count()); + Ok(Plan::Ready(RunPlan { + model_input, + shown: render.shown, + truncated: render.truncated, + estimate: est, + pending: new.len(), + })) +} + +/// Validate model `output` for `plan` and produce the next artifact version. +/// +/// On any contract violation this returns [`Error::Nonconforming`] and the +/// caller must persist nothing. Coverage is computed from exactly the signals +/// the plan showed; append-section history from `prior` is spliced back +/// mechanically, so it survives regardless of what the model emitted. +pub fn complete_run( + spec: &FoldSpec, + prior: Option<&ArtifactPayload>, + plan: &RunPlan, + output: &str, + created_at: i64, +) -> Result { + let sch = schema::builtin(&spec.schema) + .ok_or_else(|| Error::InvalidSpec(format!("unknown schema {:?}", spec.schema)))?; + 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); + 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), + shown_ids, + model: spec.model.clone(), + schema: spec.schema.clone(), + prompt_sha256: spec.prompt_sha256(), + truncated: plan.truncated, + created_at, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::selection::Selection; + + fn hex_id(c: char) -> String { + std::iter::repeat_n(c, 64).collect() + } + + fn spec() -> FoldSpec { + FoldSpec { + name: "team-digest".to_string(), + selection: Selection { + channels: vec!["ch1".to_string()], + authors: vec![], + kinds: vec![], + }, + schema: "channel-digest@v1".to_string(), + model: "haiku".to_string(), + instructions: "Maintain the digest.".to_string(), + } + } + + fn sig(id: &str, ts: i64, content: &str) -> Signal { + Signal { + id: id.to_string(), + pubkey: "pk".to_string(), + kind: 9, + created_at: ts, + content: content.to_string(), + channel: Some("ch1".to_string()), + } + } + + fn digest(log_body: &str) -> String { + format!("# Working Context\n\nSummary.\n\n# Log\n\n{log_body}\n") + } + + fn artifact_v1(spec: &FoldSpec, shown: &[(&str, i64)]) -> ArtifactPayload { + ArtifactPayload { + fold: spec.name.clone(), + version: 1, + output: digest(&format!("- prior entry [event:{}]", hex_id('a'))), + 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), + model: spec.model.clone(), + schema: spec.schema.clone(), + prompt_sha256: spec.prompt_sha256(), + truncated: false, + created_at: 1_700_000_100, + } + } + + #[test] + fn cached_when_config_matches_and_nothing_new() { + let spec = spec(); + let a = hex_id('a'); + let prior = artifact_v1(&spec, &[(&a, 100)]); + let covered = BTreeSet::from([a.clone()]); + let plan = plan_run( + &spec, + Some(&prior), + &covered, + vec![sig(&a, 100, "seen")], + &BTreeMap::new(), + ) + .expect("plan"); + assert_eq!(plan, Plan::Cached); + } + + #[test] + fn config_change_reruns_without_new_signals() { + let mut spec = spec(); + let a = hex_id('a'); + let prior = artifact_v1(&spec, &[(&a, 100)]); + spec.instructions.push_str(" Now stricter."); + let covered = BTreeSet::from([a.clone()]); + let plan = plan_run( + &spec, + Some(&prior), + &covered, + vec![sig(&a, 100, "seen")], + &BTreeMap::new(), + ) + .expect("plan"); + let Plan::Ready(run) = plan else { + panic!("expected Ready, got {plan:?}"); + }; + assert!(run.shown.is_empty()); + assert!(!run.model_input.contains("--- SOURCE EVENT IDS ---")); + assert!(run + .model_input + .contains("--- PRIOR DIGEST (team-digest v1) ---")); + // No shown signals → citation contract not demanded, coverage stays empty. + let v2 = complete_run(&spec, Some(&prior), &run, &digest(""), 1_700_000_200).expect("v2"); + assert_eq!(v2.version, 2); + assert!(v2.shown_ids.is_empty()); + assert_eq!(v2.coverage_since, None); + assert_eq!(v2.coverage_until, None); + // The engine splice still carried the prior Log forward. + assert!(v2.output.contains("- prior entry")); + } + + #[test] + fn stalled_when_standing_summary_exhausts_budget() { + let spec = spec(); + let mut prior = artifact_v1(&spec, &[]); + prior.output = "x".repeat(MAX_CONTEXT_CHARS); + let plan = plan_run( + &spec, + Some(&prior), + &BTreeSet::new(), + vec![sig(&hex_id('b'), 200, "pending")], + &BTreeMap::new(), + ) + .expect("plan"); + let Plan::Stalled { reason, pending } = plan else { + panic!("expected Stalled, got Ready/Cached"); + }; + assert!(reason.contains("standing summary exhausts")); + assert_eq!(pending, 1); + } + + #[test] + fn ready_plan_prices_the_exact_model_input() { + let spec = spec(); + let b = hex_id('b'); + let plan = plan_run( + &spec, + None, + &BTreeSet::new(), + vec![sig(&b, 200, "hello")], + &BTreeMap::new(), + ) + .expect("plan"); + let Plan::Ready(run) = plan else { + panic!("expected Ready"); + }; + assert_eq!(run.pending, 1); + assert_eq!(run.shown.len(), 1); + assert!(run.model_input.contains("--- SOURCE EVENT IDS ---")); + assert!(run.model_input.contains(&b)); + assert_eq!( + run.estimate, + estimate::estimate("haiku", run.model_input.chars().count()) + ); + assert_eq!(run.estimate.window_fit.fits, Some(true)); + } + + #[test] + fn oversized_backlog_chunks_honestly_and_never_seals_unread() { + let spec = spec(); + // Enough large signals that they cannot all fit one run's budget. + let fetched: Vec = (0..40) + .map(|i| sig(&format!("{i:064}"), 1_000 + i as i64, &"m".repeat(5_000))) + .collect(); + let plan = + plan_run(&spec, None, &BTreeSet::new(), fetched, &BTreeMap::new()).expect("plan"); + let Plan::Ready(run) = plan else { + panic!("expected Ready"); + }; + assert!(run.truncated); + assert!(run.shown.len() < 40, "must not claim the whole backlog"); + assert_eq!(run.pending, 40); + // Coverage seals only what was shown; the rest stays pending. + let cited = run.shown[0].id.clone(); + let out = digest(&format!("- chunk one [event:{cited}]")); + let v1 = complete_run(&spec, None, &run, &out, 1_700_000_300).expect("v1"); + assert_eq!(v1.shown_ids.len(), run.shown.len()); + assert!(v1.truncated); + } + + #[test] + fn complete_run_produces_v1_with_coverage_window() { + let spec = spec(); + let b = hex_id('b'); + let plan = plan_run( + &spec, + None, + &BTreeSet::new(), + vec![sig(&b, 200, "hello")], + &BTreeMap::new(), + ) + .expect("plan"); + let Plan::Ready(run) = plan else { + panic!("expected Ready"); + }; + let out = digest(&format!("- said hello [event:{b}]")); + let v1 = complete_run(&spec, None, &run, &out, 1_700_000_300).expect("v1"); + assert_eq!(v1.version, 1); + assert_eq!(v1.shown_ids, vec![b]); + assert_eq!(v1.coverage_since, Some(200)); + assert_eq!(v1.coverage_until, Some(201)); + assert_eq!(v1.prompt_sha256, spec.prompt_sha256()); + } + + #[test] + fn complete_run_splices_prior_history_and_increments_version() { + let spec = spec(); + let a = hex_id('a'); + let b = hex_id('b'); + let prior = artifact_v1(&spec, &[(&a, 100)]); + let covered = BTreeSet::from([a]); + let plan = plan_run( + &spec, + Some(&prior), + &covered, + vec![sig(&b, 200, "next")], + &BTreeMap::new(), + ) + .expect("plan"); + let Plan::Ready(run) = plan else { + panic!("expected Ready"); + }; + // Model returns ONLY the new entry; engine must carry the prior Log. + let out = digest(&format!("- new entry [event:{b}]")); + let v2 = complete_run(&spec, Some(&prior), &run, &out, 1_700_000_400).expect("v2"); + assert_eq!(v2.version, 2); + assert!(v2.output.contains("- prior entry")); + assert!(v2.output.contains("- new entry")); + } + + #[test] + fn nonconforming_output_is_refused_and_nothing_persists() { + let spec = spec(); + let b = hex_id('b'); + let plan = plan_run( + &spec, + None, + &BTreeSet::new(), + vec![sig(&b, 200, "hello")], + &BTreeMap::new(), + ) + .expect("plan"); + let Plan::Ready(run) = plan else { + panic!("expected Ready"); + }; + // Fabricated citation. + let fabricated = digest(&format!("- invented [event:{}]", hex_id('c'))); + assert!(matches!( + complete_run(&spec, None, &run, &fabricated, 0), + Err(Error::Nonconforming(_)) + )); + // Wrong sections. + assert!(matches!( + complete_run(&spec, None, &run, "# Nope\n\nbody\n", 0), + Err(Error::Nonconforming(_)) + )); + } +} diff --git a/crates/buzz-accumulator/src/runner.rs b/crates/buzz-accumulator/src/runner.rs new file mode 100644 index 0000000000..e10d60d865 --- /dev/null +++ b/crates/buzz-accumulator/src/runner.rs @@ -0,0 +1,261 @@ +//! The model-call seam: a trait plus the proven subprocess implementation. + +use std::io::{Read, Write}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use crate::error::Error; + +/// Executes one fold run's model call. Implementations receive the exact +/// priced `model_input` string and return raw model output text. +/// +/// This is the only impurity seam in the crate. A future implementation can +/// route through `buzz-agent`'s provider client; v1 ships the subprocess +/// runner below. +pub trait FoldRunner { + /// Run the model over `model_input` and return its output text. + fn run(&self, model_input: &str, model: &str) -> Result; +} + +/// Default max thinking-token cap passed to the CLI. 1024 keeps citation +/// accuracy — fully disabling thinking made models drop citations; unbounded +/// thinking dominated run latency. +const DEFAULT_MAX_THINKING_TOKENS: &str = "1024"; + +/// Phrases an unauthenticated CLI prints (with exit 0) instead of output. +const SENTINELS: &[&str] = &["not logged in", "please run /login", "invalid api key"]; + +/// Shells out to `claude -p --system-prompt ` with input on stdin. +/// +/// A fold run is a pure text transform, but a plain agent CLI call is a full +/// harness: its own large system prompt, memory/context injection, and +/// unbounded extended thinking. So we pass a minimal transform-engine system +/// prompt and cap thinking (override with `BUZZ_FOLD_MAX_THINKING_TOKENS`; +/// set it empty to keep the CLI default). The binary defaults to `claude`, +/// overridable with `BUZZ_FOLD_RUNNER_BIN`. +/// +/// Hardening (each gate is load-bearing, learned on real runs): +/// - exit != 0 is invalid even with partial stdout (mid-stream crash); +/// - exit 0 is not trusted: empty output, or a short output that is +/// 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. +pub struct SubprocessRunner { + binary: String, + timeout: Duration, +} + +impl Default for SubprocessRunner { + fn default() -> Self { + Self::new() + } +} + +impl SubprocessRunner { + /// Runner with the environment-configured binary and a 600s timeout. + pub fn new() -> Self { + Self { + binary: std::env::var("BUZZ_FOLD_RUNNER_BIN").unwrap_or_else(|_| "claude".to_string()), + timeout: Duration::from_secs(600), + } + } + + /// Runner with an explicit binary and timeout (tests, pinned installs). + pub fn with_binary(binary: impl Into, timeout: Duration) -> Self { + Self { + binary: binary.into(), + timeout, + } + } + + /// The minimal system prompt that turns an agent CLI into a transform engine. + pub const SYSTEM_PROMPT: &'static str = "You are a headless text-transform engine. There is \ + no filesystem, no tools, no memory. Read the task and input, output ONLY the requested \ + document — no preamble, no questions."; + + fn command(&self, model: &str) -> Command { + let mut cmd = Command::new(&self.binary); + cmd.arg("-p") + .arg("--system-prompt") + .arg(Self::SYSTEM_PROMPT); + if !model.is_empty() { + cmd.arg("--model").arg(model); + } + let thinking = std::env::var("BUZZ_FOLD_MAX_THINKING_TOKENS") + .unwrap_or_else(|_| DEFAULT_MAX_THINKING_TOKENS.to_string()); + if !thinking.is_empty() { + cmd.env("MAX_THINKING_TOKENS", thinking); + } + cmd.stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + cmd + } + + fn wait_with_deadline(&self, child: &mut Child) -> Result { + let start = Instant::now(); + loop { + match child.try_wait() { + Ok(Some(status)) => return Ok(status), + Ok(None) => { + if start.elapsed() > self.timeout { + let _ = child.kill(); + let _ = child.wait(); + return Err(Error::Runner(format!( + "fold runner timed out after {}s", + self.timeout.as_secs() + ))); + } + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => return Err(Error::Runner(format!("failed to wait on fold runner: {e}"))), + } + } + } +} + +/// First 200 chars, for error snippets. +fn snip(s: &str) -> String { + s.chars().take(200).collect() +} + +/// Drain a child pipe to a string on a background thread (avoids pipe-buffer +/// deadlock against the concurrent stdin write). +fn drain( + reader: Option, + label: &'static str, +) -> std::thread::JoinHandle> { + std::thread::spawn(move || { + let mut text = String::new(); + 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), + } + }) +} + +impl FoldRunner for SubprocessRunner { + fn run(&self, model_input: &str, model: &str) -> Result { + let mut child = self.command(model).spawn().map_err(|e| { + Error::Runner(format!( + "failed to launch fold runner {:?}: {e}", + self.binary + )) + })?; + let stdin = child.stdin.take(); + let input = model_input.to_string(); + let writer = std::thread::spawn(move || { + if let Some(mut pipe) = stdin { + let _ = pipe.write_all(input.as_bytes()); + } + }); + let stdout = drain(child.stdout.take(), "stdout"); + 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 out = out.trim().to_string(); + if !status.success() { + return Err(Error::Runner(format!( + "fold runner exited {status} — treating output as invalid. stdout={:?} stderr={:?}", + snip(&out), + snip(&err) + ))); + } + let low = out.to_lowercase(); + let is_sentinel = SENTINELS.iter().any(|s| low.contains(s)); + if out.is_empty() || (is_sentinel && out.chars().count() <= 80) { + return Err(Error::Runner(format!( + "fold runner produced no usable output (auth or empty-response failure). \ + exit={status} stdout={:?} stderr={:?}", + snip(&out), + snip(&err) + ))); + } + Ok(out) + } +} + +#[cfg(all(test, unix))] +mod tests { + use super::*; + use std::os::unix::fs::PermissionsExt; + use std::path::PathBuf; + + /// Write an executable stub script and return its path. + fn stub(name: &str, body: &str) -> PathBuf { + let dir = + std::env::temp_dir().join(format!("buzz-accumulator-test-{}", std::process::id())); + std::fs::create_dir_all(&dir).expect("mkdir"); + let path = dir.join(name); + std::fs::write(&path, format!("#!/bin/sh\n{body}\n")).expect("write stub"); + let mut perms = std::fs::metadata(&path).expect("meta").permissions(); + perms.set_mode(0o755); + std::fs::set_permissions(&path, perms).expect("chmod"); + path + } + + fn runner(path: &std::path::Path) -> SubprocessRunner { + SubprocessRunner::with_binary(path.to_string_lossy(), Duration::from_secs(20)) + } + + #[test] + fn missing_binary_fails_cleanly() { + let r = + SubprocessRunner::with_binary("/nonexistent/buzz-fold-runner", Duration::from_secs(1)); + let err = r.run("input", "haiku"); + assert!(matches!(err, Err(Error::Runner(msg)) if msg.contains("failed to launch"))); + } + + #[test] + fn passes_input_on_stdin_and_returns_stdout() { + let path = stub("echo-runner.sh", "cat -"); + let out = runner(&path).run("the exact input", "haiku").expect("run"); + assert_eq!(out, "the exact input"); + } + + #[test] + fn nonzero_exit_invalidates_partial_stdout() { + let path = stub("crash-runner.sh", "echo partial output; exit 3"); + let err = runner(&path).run("input", "haiku"); + assert!( + matches!(err, Err(Error::Runner(msg)) if msg.contains("treating output as invalid")) + ); + } + + #[test] + fn short_sentinel_output_is_rejected_even_on_exit_zero() { + let path = stub( + "auth-runner.sh", + "echo 'Not logged in · Please run /login'; exit 0", + ); + let err = runner(&path).run("input", "haiku"); + assert!(matches!(err, Err(Error::Runner(msg)) if msg.contains("no usable output"))); + } + + #[test] + fn long_output_discussing_login_is_not_rejected() { + let path = stub( + "discuss-runner.sh", + "printf '%s' 'A long digest that happens to mention someone was not logged in during the incident window, plus plenty of other prose to exceed the sentinel length gate.'", + ); + let out = runner(&path).run("input", "haiku").expect("run"); + assert!(out.contains("not logged in")); + } + + #[test] + fn empty_output_is_rejected() { + let path = stub("silent-runner.sh", "exit 0"); + let err = runner(&path).run("input", "haiku"); + assert!(matches!(err, Err(Error::Runner(msg)) if msg.contains("no usable output"))); + } +} diff --git a/crates/buzz-accumulator/src/schema.rs b/crates/buzz-accumulator/src/schema.rs new file mode 100644 index 0000000000..df7a15a232 --- /dev/null +++ b/crates/buzz-accumulator/src/schema.rs @@ -0,0 +1,70 @@ +//! Artifact schemas: the structural contract a fold's output must satisfy. + +/// A schema names the exact H1 sections an artifact must have, and which of +/// them are *append* sections — sections whose history the engine retains +/// mechanically ([`crate::validate::splice_append_sections`]) so the model +/// only ever contributes new entries. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +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). + pub sections: &'static [&'static str], + /// Sections that accumulate: prior content is preserved by construction + /// and new entries must carry signal citations. + pub append_sections: &'static [&'static str], +} + +/// A bounded digest of a selection: a rewritten standing summary plus an +/// append-only, citation-carrying log. +pub const CHANNEL_DIGEST_V1: ArtifactSchema = ArtifactSchema { + name: "channel-digest@v1", + sections: &["Working Context", "Log"], + 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]; + +/// Look up a built-in schema by name. +pub fn builtin(name: &str) -> Option<&'static ArtifactSchema> { + BUILTIN_SCHEMAS.iter().copied().find(|s| s.name == name) +} + +/// Default fold instructions for [`CHANNEL_DIGEST_V1`]. +/// +/// The contract lines are load-bearing: exact H1 headings, only-new log +/// entries (the engine splices history back), and full-id citations copied +/// from the run's SOURCE EVENT IDS list — ids embedded in message text are +/// never citations. +pub const CHANNEL_DIGEST_PROMPT: &str = "Maintain a bounded digest of the anchored selection.\n\ +Return markdown with exactly these H1 sections, in this order: Working Context, Log.\n\ +Working Context is a concise standing summary rewritten in light of all evidence.\n\ +Log: output ONLY new dated entries for this run's evidence — the engine keeps the\n\ +standing Log and appends your new entries to it; never repeat prior entries.\n\ +For new evidence, cite the source events you actually use as [event:], one id\n\ +per bracket, copied in full from the SOURCE EVENT IDS list. Ids that appear inside\n\ +message text are never citations. Do not invent facts. Output only the document markdown."; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn builtin_lookup() { + assert_eq!( + 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 new file mode 100644 index 0000000000..ae1a8c4801 --- /dev/null +++ b/crates/buzz-accumulator/src/selection.rs @@ -0,0 +1,203 @@ +//! Selections: saveable queries over signals, compiled to relay filters. + +use serde::{Deserialize, Serialize}; + +use crate::error::Error; +use crate::signal::Signal; + +/// Kinds a selection reads when none are specified: chat messages. +pub const DEFAULT_SIGNAL_KINDS: &[u32] = &[9]; + +/// A named-able query over signals. +/// +/// v1 supports the shapes the relay can answer directly: channels (`#h`), +/// authors, and kinds. DMs are deliberately excluded — a selection never +/// reads them. Time is not part of the selection; each run supplies its own +/// half-open window so the same selection replays deterministically. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Selection { + /// Channel ids (NIP-29 `h` tags). One relay filter is emitted per channel. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub channels: Vec, + /// Author pubkeys (hex). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub authors: Vec, + /// Event kinds; empty means [`DEFAULT_SIGNAL_KINDS`]. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub kinds: Vec, +} + +impl Selection { + /// Validate and canonicalize (sort + dedupe) in place, so equal selections + /// serialize identically. + pub fn canonicalize(&mut self) -> Result<(), Error> { + self.channels.sort(); + self.channels.dedup(); + self.authors.sort(); + self.authors.dedup(); + self.kinds.sort_unstable(); + self.kinds.dedup(); + if self.channels.is_empty() && self.authors.is_empty() { + return Err(Error::InvalidSpec( + "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 self.authors.iter().any(|a| a.trim().is_empty()) { + return Err(Error::InvalidSpec( + "selection has an empty author pubkey".into(), + )); + } + Ok(()) + } + + /// Effective kinds for relay filters (relay queries must always name kinds). + pub fn effective_kinds(&self) -> Vec { + if self.kinds.is_empty() { + DEFAULT_SIGNAL_KINDS.to_vec() + } else { + self.kinds.clone() + } + } + + /// Compile to NIP-01 relay filters over the half-open window + /// `[since, until_exclusive)`. + /// + /// One filter is emitted **per channel** rather than a single multi-`#h` + /// filter: per-channel requests match the relay's fan-out expectations and + /// avoid union-filter edge cases. Nostr `since`/`until` are inclusive, so + /// the exclusive end is mapped to `until_exclusive - 1`. + pub fn compile_filters( + &self, + since: i64, + until_exclusive: i64, + limit: usize, + ) -> Vec { + let kinds = self.effective_kinds(); + let until = until_exclusive.saturating_sub(1); + let base = |extra: &mut serde_json::Map| { + extra.insert("kinds".into(), serde_json::json!(kinds)); + extra.insert("since".into(), serde_json::json!(since)); + extra.insert("until".into(), serde_json::json!(until)); + extra.insert("limit".into(), serde_json::json!(limit)); + if !self.authors.is_empty() { + extra.insert("authors".into(), serde_json::json!(self.authors)); + } + }; + if self.channels.is_empty() { + let mut m = serde_json::Map::new(); + base(&mut m); + return vec![serde_json::Value::Object(m)]; + } + self.channels + .iter() + .map(|ch| { + let mut m = serde_json::Map::new(); + base(&mut m); + m.insert("#h".into(), serde_json::json!([ch])); + serde_json::Value::Object(m) + }) + .collect() + } + + /// Human-readable one-line description. + pub fn describe(&self) -> String { + let mut parts = Vec::new(); + if !self.channels.is_empty() { + parts.push(format!("{} channel(s)", self.channels.len())); + } + if !self.authors.is_empty() { + parts.push(format!("{} author(s)", self.authors.len())); + } + parts.push(format!("kinds {:?}", self.effective_kinds())); + parts.join(" · ") + } +} + +/// Deterministically materialize fetched signals: `created_at ASC, id ASC`, +/// duplicates (same id) removed. +/// +/// The same fetched set always materializes to the identical ordered list, so +/// (selection, window) pins exactly which signals a run saw. +pub fn materialize(mut signals: Vec) -> Vec { + signals.sort_by(|a, b| { + a.created_at + .cmp(&b.created_at) + .then_with(|| a.id.cmp(&b.id)) + }); + signals.dedup_by(|a, b| a.id == b.id); + signals +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sig(id: &str, ts: i64) -> Signal { + Signal { + id: id.into(), + pubkey: "pk".into(), + kind: 9, + created_at: ts, + content: "x".into(), + channel: None, + } + } + + #[test] + fn materialize_orders_created_at_then_id_and_dedupes() { + let out = materialize(vec![sig("bb", 5), sig("aa", 5), sig("cc", 1), sig("aa", 5)]); + let ids: Vec<&str> = out.iter().map(|s| s.id.as_str()).collect(); + assert_eq!(ids, vec!["cc", "aa", "bb"]); + } + + #[test] + fn empty_selection_is_rejected() { + let mut s = Selection::default(); + assert!(s.canonicalize().is_err()); + } + + #[test] + fn canonicalize_sorts_and_dedupes() { + let mut s = Selection { + channels: vec!["b".into(), "a".into(), "b".into()], + authors: vec![], + kinds: vec![40002, 9, 9], + }; + s.canonicalize().expect("valid"); + assert_eq!(s.channels, vec!["a", "b"]); + assert_eq!(s.kinds, vec![9, 40002]); + } + + #[test] + fn one_filter_per_channel_with_half_open_window() { + let s = Selection { + channels: vec!["ch1".into(), "ch2".into()], + authors: vec!["p1".into()], + kinds: vec![], + }; + let filters = s.compile_filters(100, 200, 500); + assert_eq!(filters.len(), 2); + assert_eq!(filters[0]["#h"], serde_json::json!(["ch1"])); + assert_eq!(filters[0]["kinds"], serde_json::json!([9])); + assert_eq!(filters[0]["since"], serde_json::json!(100)); + assert_eq!(filters[0]["until"], serde_json::json!(199)); + assert_eq!(filters[0]["authors"], serde_json::json!(["p1"])); + } + + #[test] + fn author_only_selection_compiles_one_filter() { + let s = Selection { + channels: vec![], + authors: vec!["p1".into()], + kinds: vec![9], + }; + let filters = s.compile_filters(0, 10, 5); + assert_eq!(filters.len(), 1); + assert!(filters[0].get("#h").is_none()); + } +} diff --git a/crates/buzz-accumulator/src/signal.rs b/crates/buzz-accumulator/src/signal.rs new file mode 100644 index 0000000000..30c99c0bcd --- /dev/null +++ b/crates/buzz-accumulator/src/signal.rs @@ -0,0 +1,25 @@ +//! The engine's view of a raw relay event. + +use serde::{Deserialize, Serialize}; + +/// One signal: a raw relay event the caller fetched for a selection. +/// +/// Signals are never rewritten — they are the source of truth every artifact +/// claim must be able to point back to. The caller maps relay JSON into this +/// shape; the engine only reads it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Signal { + /// Nostr event id (64 lowercase hex chars). + pub id: String, + /// Author pubkey (hex). + pub pubkey: String, + /// Event kind. + pub kind: u32, + /// Unix seconds. + pub created_at: i64, + /// Plain content. + pub content: String, + /// Channel id (`h` tag) when the event is channel-scoped. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub channel: Option, +} diff --git a/crates/buzz-accumulator/src/spec.rs b/crates/buzz-accumulator/src/spec.rs new file mode 100644 index 0000000000..fb20c54d49 --- /dev/null +++ b/crates/buzz-accumulator/src/spec.rs @@ -0,0 +1,134 @@ +//! Fold specs: the saved definition of one accumulator. +//! +//! A spec is the JSON content of an addressable relay event (`d` tag = fold +//! name, last-write-wins), NIP-44-encrypted to its author. There is +//! deliberately no cadence field: *when* to run belongs to the caller — a +//! human pressing Run today, a `buzz-workflow` schedule trigger later. The +//! spec only says what to read and how to fold it. + +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::error::Error; +use crate::schema; +use crate::selection::Selection; + +/// Maximum fold-name length (also the `d` tag value). +pub const MAX_NAME_LEN: usize = 64; + +/// One fold definition. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct FoldSpec { + /// Fold name: 1–64 chars of `[a-z0-9-]`, no leading/trailing dash. + pub name: String, + /// What to read. + pub selection: Selection, + /// Built-in schema name (see [`crate::schema::BUILTIN_SCHEMAS`]). + pub schema: String, + /// Model alias passed to the runner and priced by [`crate::estimate`]. + pub model: String, + /// Fold instructions (the prompt). + pub instructions: String, +} + +impl FoldSpec { + /// Validate the spec and canonicalize its selection in place. + pub fn validate(&mut self) -> Result<(), Error> { + let name_ok = !self.name.is_empty() + && self.name.len() <= MAX_NAME_LEN + && self + .name + .chars() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-') + && !self.name.starts_with('-') + && !self.name.ends_with('-'); + if !name_ok { + return Err(Error::InvalidSpec(format!( + "name {:?} must be 1-{MAX_NAME_LEN} chars of [a-z0-9-] with no edge dashes", + self.name + ))); + } + if schema::builtin(&self.schema).is_none() { + let known: Vec<&str> = schema::BUILTIN_SCHEMAS.iter().map(|s| s.name).collect(); + return Err(Error::InvalidSpec(format!( + "unknown schema {:?}; built-ins: {known:?}", + self.schema + ))); + } + if self.model.trim().is_empty() { + return Err(Error::InvalidSpec( + "model is required — estimates and runs must name the model".into(), + )); + } + if self.instructions.trim().is_empty() { + return Err(Error::InvalidSpec("instructions must be non-empty".into())); + } + self.selection.canonicalize() + } + + /// SHA-256 hex of the instructions. Stored on every artifact version so a + /// prompt change is visible in provenance and invalidates the cached-run + /// shortcut. + pub fn prompt_sha256(&self) -> String { + let digest = Sha256::digest(self.instructions.as_bytes()); + digest.iter().map(|b| format!("{b:02x}")).collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn valid() -> FoldSpec { + FoldSpec { + name: "team-digest".to_string(), + selection: Selection { + channels: vec!["ch1".to_string()], + authors: vec![], + kinds: vec![], + }, + schema: "channel-digest@v1".to_string(), + model: "haiku".to_string(), + instructions: "Maintain the digest.".to_string(), + } + } + + #[test] + fn valid_spec_passes() { + assert!(valid().validate().is_ok()); + } + + #[test] + fn bad_names_are_rejected() { + for name in ["", "-lead", "trail-", "Has Caps", "way!", &"x".repeat(65)] { + let mut s = valid(); + s.name = name.to_string(); + assert!(s.validate().is_err(), "name {name:?} should be rejected"); + } + } + + #[test] + fn unknown_schema_and_empty_fields_are_rejected() { + let mut s = valid(); + s.schema = "nope@v1".to_string(); + assert!(s.validate().is_err()); + let mut s = valid(); + s.model = " ".to_string(); + assert!(s.validate().is_err()); + let mut s = valid(); + s.instructions = "".to_string(); + assert!(s.validate().is_err()); + } + + #[test] + fn prompt_hash_tracks_instructions_only() { + let a = valid(); + let mut b = valid(); + b.model = "opus".to_string(); + assert_eq!(a.prompt_sha256(), b.prompt_sha256()); + let mut c = valid(); + c.instructions.push('!'); + assert_ne!(a.prompt_sha256(), c.prompt_sha256()); + assert_eq!(a.prompt_sha256().len(), 64); + } +} diff --git a/crates/buzz-accumulator/src/transcript.rs b/crates/buzz-accumulator/src/transcript.rs new file mode 100644 index 0000000000..1434dd3ae4 --- /dev/null +++ b/crates/buzz-accumulator/src/transcript.rs @@ -0,0 +1,217 @@ +//! Transcript rendering: whole recent signals under a hard character budget. +//! +//! The budget is filled newest-first so the freshest evidence always makes the +//! cut, then re-emitted in time order. The render reports the *exact* signals +//! shown — that list is what coverage may seal. A signal that did not fit is +//! simply not shown and stays pending; it is never summarized invisibly. + +use std::collections::BTreeMap; + +use chrono::{TimeZone, Utc}; +use serde::{Deserialize, Serialize}; + +use crate::signal::Signal; + +/// Identity and timestamp of one signal actually included in a transcript. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ShownSignal { + /// Event id. + pub id: String, + /// Unix seconds, used for the coverage window. + pub created_at: i64, +} + +/// A rendered transcript plus the honesty bookkeeping around it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TranscriptRender { + /// Time-ordered transcript body (possibly truncated, with a visible marker). + pub body: String, + /// True when any candidate signal was dropped or tail-trimmed to fit. + pub truncated: bool, + /// Exactly the signals whose text appears in `body`, in body order. + pub shown: Vec, +} + +/// Human label for an event kind in transcript lines. +fn kind_label(kind: u32) -> String { + if kind == 9 { + "message".to_string() + } else { + format!("kind-{kind}") + } +} + +/// Render one signal as a single transcript line: +/// `[YYYY-MM-DD HH:MM] {who} <{kind}>: {whitespace-normalized content}`. +fn render_line(signal: &Signal, names: &BTreeMap) -> String { + let iso = match Utc.timestamp_opt(signal.created_at, 0).single() { + Some(ts) => ts.format("%Y-%m-%d %H:%M").to_string(), + None => format!("ts:{}", signal.created_at), + }; + let who = names + .get(&signal.pubkey) + .cloned() + .unwrap_or_else(|| signal.pubkey.chars().take(8).collect()); + let text = signal + .content + .split_whitespace() + .collect::>() + .join(" "); + format!("[{iso}] {who} <{}>: {text}", kind_label(signal.kind)) +} + +/// Render `signals` (already materialized: time-ordered, deduped) into a +/// transcript of at most `max_chars` characters, 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. +pub fn render_transcript( + signals: &[Signal], + names: &BTreeMap, + max_chars: usize, +) -> TranscriptRender { + let rendered: Vec<(ShownSignal, String)> = signals + .iter() + .map(|s| { + ( + ShownSignal { + id: s.id.clone(), + created_at: s.created_at, + }, + render_line(s, names), + ) + }) + .collect(); + let mut kept: Vec<(ShownSignal, String)> = Vec::new(); + let mut used = 0usize; + let mut trimmed = false; + for (shown, line) in rendered.iter().rev() { + 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 { + 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::()) + }; + cost = line.chars().count(); + trimmed = true; + } + kept.push((shown.clone(), line)); + used += cost; + } + kept.reverse(); + let truncated = trimmed || kept.len() != rendered.len(); + let mut body = kept + .iter() + .map(|(_, line)| line.as_str()) + .collect::>() + .join("\n"); + if truncated { + body = if body.is_empty() { + "[…older events truncated to fit context budget…]".to_string() + } else { + format!("[…older events truncated to fit context budget…]\n{body}") + }; + } + TranscriptRender { + body, + truncated, + shown: kept.into_iter().map(|(shown, _)| shown).collect(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sig(id: &str, ts: i64, content: &str) -> Signal { + Signal { + id: id.to_string(), + pubkey: "aabbccddeeff0011".to_string(), + kind: 9, + created_at: ts, + content: content.to_string(), + channel: None, + } + } + + #[test] + fn renders_all_when_budget_allows() { + let names = BTreeMap::from([("aabbccddeeff0011".to_string(), "riley".to_string())]); + let signals = vec![ + sig("e1", 1_700_000_000, "hello there"), + sig("e2", 1_700_000_060, "hi"), + ]; + let r = render_transcript(&signals, &names, 10_000); + assert!(!r.truncated); + assert_eq!(r.shown.len(), 2); + assert!(r.body.contains("riley : hello there")); + assert!(r.body.lines().count() == 2); + } + + #[test] + 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) + .body + .chars() + .count(); + let r = render_transcript(&signals, &names, newest_len); + assert!(r.truncated); + assert_eq!( + r.shown.iter().map(|s| s.id.as_str()).collect::>(), + vec!["e2"] + ); + assert!(r.body.starts_with("[…older events truncated")); + assert!(r.body.contains("new")); + assert!(!r.body.contains("old old old")); + } + + #[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); + 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); + } + + #[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); + assert!(r.truncated); + assert!(r.shown.is_empty()); + assert_eq!(r.body, "[…older events truncated to fit context budget…]"); + } + + #[test] + fn unknown_author_falls_back_to_pubkey_prefix_and_kind_labelled() { + 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); + assert!(r.body.contains("aabbccdd :")); + } + + #[test] + fn empty_input_is_empty_and_honest() { + let r = render_transcript(&[], &BTreeMap::new(), 100); + 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 new file mode 100644 index 0000000000..a386b0a4d9 --- /dev/null +++ b/crates/buzz-accumulator/src/validate.rs @@ -0,0 +1,371 @@ +//! Output-contract validation and the engine splice. +//! +//! Nonconforming output is refused and nothing persists. The rules here are +//! ones a model can genuinely meet — exact H1 sections, citations drawn only +//! from the ids it was shown — never rules it could only satisfy by lying. + +use std::collections::BTreeSet; + +use crate::error::Error; +use crate::schema::ArtifactSchema; + +/// True for a line the heading grammar recognizes as an H1: `# ` followed by +/// at least one non-`#` character. +fn is_h1(line: &str) -> bool { + match line.strip_prefix("# ") { + Some(rest) => rest.chars().next().is_some_and(|c| c != '#'), + None => false, + } +} + +/// Byte offsets of the section body for H1 `name`: (start just after the +/// heading's newline, end at the next H1 line or end of input). `None` when +/// the heading is absent (a heading with no trailing newline has no body). +fn section_span(output: &str, name: &str) -> Option<(usize, usize)> { + let mut pos = 0; + let mut found: Option = None; + for line in output.split_inclusive('\n') { + let bare = line.strip_suffix('\n'); + let text = bare.unwrap_or(line); + match found { + None => { + // Heading line: `# {name}` plus optional trailing whitespace, + // and it must be newline-terminated to have a body. + let matches = text + .strip_prefix("# ") + .is_some_and(|rest| rest.trim_end() == name && rest.trim_end() == rest.trim()); + if matches && bare.is_some() { + found = Some(pos + line.len()); + } + } + Some(start) => { + if is_h1(text) { + return Some((start, pos)); + } + } + } + pos += line.len(); + } + found.map(|start| (start, output.len())) +} + +/// The trimmed body of H1 section `name`, or `""` when absent. +pub fn section(output: &str, name: &str) -> String { + match section_span(output, name) { + Some((start, end)) => output[start..end].trim().to_string(), + None => String::new(), + } +} + +/// Replace the body of H1 section `name` with `content`, leaving the rest of +/// the document byte-identical. A missing heading leaves the document +/// unchanged. +pub fn replace_section(output: &str, name: &str, content: &str) -> String { + match section_span(output, name) { + Some((start, end)) => { + format!("{}\n{}\n\n{}", &output[..start], content, &output[end..]) + } + None => output.to_string(), + } +} + +/// All H1 headings in document order (text after `# `, verbatim to line end). +pub fn headings(output: &str) -> Vec { + output + .lines() + .filter(|line| is_h1(line)) + .map(|line| line[2..].to_string()) + .collect() +} + +/// Extract cited event ids from `[event:…]` brackets, splitting brackets that +/// 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. +pub fn cited_event_ids(citation_text: &str) -> BTreeSet { + let mut cited = BTreeSet::new(); + let mut rest = citation_text; + while let Some(open) = rest.find("[event:") { + let inner_start = open + "[event:".len(); + let Some(close_rel) = rest[inner_start..].find(']') else { + break; + }; + let chunk = &rest[inner_start..inner_start + close_rel]; + rest = &rest[inner_start + close_rel + 1..]; + if chunk.is_empty() { + continue; + } + let before = cited.len(); + // Consecutive non-overlapping 64-char windows over each maximal hex run. + 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; + } + } + } + if cited.len() == before { + cited.insert(chunk.trim().to_string()); + } + } + cited +} + +/// 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. +/// +/// Any failure is [`Error::Nonconforming`]: the caller persists nothing. +pub fn validate_output( + schema: &ArtifactSchema, + output: &str, + 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!( + "output does not conform to {}: expected H1 sections {:?}, got {:?}", + 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 + ))); + } + 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 + .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(), + )); + } + } + Ok(()) +} + +/// Rebuild append sections so prior content is retained by construction. +/// +/// The model contributes only this run's NEW entries; the engine prepends the +/// prior section verbatim. A model that repeats the prior section anyway is +/// deduplicated (exact-substring strip). History cannot be rewritten or +/// dropped no matter what the model outputs — and the model never has to +/// re-emit an ever-growing log, which kept run latency flat along the chain. +pub fn splice_append_sections( + schema: &ArtifactSchema, + previous_output: Option<&str>, + output: &str, +) -> String { + let Some(previous) = previous_output.filter(|p| !p.is_empty()) else { + return output.to_string(); + }; + let mut doc = output.to_string(); + for name in schema.append_sections { + let prior = section(previous, name); + if prior.is_empty() { + continue; + } + let model_part = section(&doc, name); + let new_part = if model_part.contains(&prior) { + model_part.replacen(&prior, "", 1) + } else { + model_part + }; + let new_part = new_part.trim(); + let combined = if new_part.is_empty() { + prior + } else { + format!("{prior}\n{new_part}") + }; + doc = replace_section(&doc, name, &combined); + } + doc +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::schema::{CHANNEL_DIGEST_V1, FREEFORM_V1}; + + fn id(c: char) -> String { + std::iter::repeat_n(c, 64).collect() + } + + fn digest(log_body: &str) -> String { + format!("# Working Context\n\nSummary.\n\n# Log\n\n{log_body}\n") + } + + #[test] + fn section_and_replace_roundtrip() { + let doc = digest("- entry one"); + assert_eq!(section(&doc, "Log"), "- entry one"); + assert_eq!(section(&doc, "Working Context"), "Summary."); + assert_eq!(section(&doc, "Missing"), ""); + let replaced = replace_section(&doc, "Log", "- swapped"); + assert_eq!(section(&replaced, "Log"), "- swapped"); + assert_eq!(section(&replaced, "Working Context"), "Summary."); + } + + #[test] + fn headings_ignore_h2_and_capture_in_order() { + let doc = "# One\n\n## nested\n\n# Two\nbody\n"; + assert_eq!(headings(doc), vec!["One", "Two"]); + } + + #[test] + fn heading_mismatch_is_refused() { + let out = "# Wrong\n\nx\n\n# Log\n\n- e\n"; + let err = validate_output(&CHANNEL_DIGEST_V1, out, None, &[id('a')]); + 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'))); + let err = validate_output(&CHANNEL_DIGEST_V1, &out, None, &[id('a')]); + assert!(matches!(err, Err(Error::Nonconforming(_)))); + } + + #[test] + fn missing_citation_is_refused() { + let out = digest("- no citation here"); + let err = validate_output(&CHANNEL_DIGEST_V1, &out, None, &[id('a')]); + assert!(matches!(err, Err(Error::Nonconforming(_)))); + } + + #[test] + fn valid_citation_passes() { + let out = digest(&format!("- did a thing [event:{}]", id('a'))); + assert!(validate_output(&CHANNEL_DIGEST_V1, &out, None, &[id('a')]).is_ok()); + } + + #[test] + fn multi_id_bracket_splits_and_short_id_fails_loudly() { + let jam = format!("[event:{}, event:{}]", id('a'), id('b')); + let cited = cited_event_ids(&jam); + assert_eq!(cited.len(), 2); + assert!(cited.contains(&id('a')) && cited.contains(&id('b'))); + let short = cited_event_ids("[event:abc123]"); + assert_eq!(short.into_iter().collect::>(), vec!["abc123"]); + } + + #[test] + fn prior_citations_do_not_satisfy_a_new_run() { + let prior = digest(&format!("- old entry [event:{}]", id('a'))); + // Model re-emits the prior Log verbatim and adds nothing new: refused. + let repeat_only = digest(&format!("- old entry [event:{}]", id('a'))); + let err = validate_output(&CHANNEL_DIGEST_V1, &repeat_only, Some(&prior), &[id('b')]); + assert!(matches!(err, Err(Error::Nonconforming(_)))); + // Repeat plus a genuinely new cited entry: the repeat is stripped, the + // new citation is judged against this run's shown ids. + let repeat_plus_new = digest(&format!( + "- old entry [event:{}]\n- new entry [event:{}]", + id('a'), + id('b') + )); + assert!(validate_output( + &CHANNEL_DIGEST_V1, + &repeat_plus_new, + Some(&prior), + &[id('b')] + ) + .is_ok()); + } + + #[test] + fn splice_retains_prior_history_when_model_omits_it() { + let prior = digest(&format!("- old entry [event:{}]", id('a'))); + let fresh = digest(&format!("- new entry [event:{}]", id('b'))); + let doc = splice_append_sections(&CHANNEL_DIGEST_V1, Some(&prior), &fresh); + let log = section(&doc, "Log"); + assert_eq!( + log, + format!( + "- old entry [event:{}]\n- new entry [event:{}]", + id('a'), + id('b') + ) + ); + } + + #[test] + fn splice_dedupes_a_repeated_prior_section() { + let prior = digest(&format!("- old entry [event:{}]", id('a'))); + let repeat_plus_new = digest(&format!( + "- old entry [event:{}]\n- new entry [event:{}]", + id('a'), + id('b') + )); + let doc = splice_append_sections(&CHANNEL_DIGEST_V1, Some(&prior), &repeat_plus_new); + let log = section(&doc, "Log"); + assert_eq!(log.matches("- old entry").count(), 1); + assert!(log.contains("- new entry")); + } + + #[test] + fn splice_with_no_new_entries_keeps_prior_exactly() { + let prior = digest(&format!("- old entry [event:{}]", id('a'))); + let empty_log = digest(""); + let doc = splice_append_sections(&CHANNEL_DIGEST_V1, Some(&prior), &empty_log); + assert_eq!( + section(&doc, "Log"), + format!("- old entry [event:{}]", id('a')) + ); + } +} diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 4e1ab1c7f5..eb5816a103 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -117,6 +117,24 @@ pub const KIND_PUSH_LEASE: u32 = 30350; /// plus exact public projection bindings. See `docs/nips/NIP-PMA.md`. pub const KIND_PRIVATE_MANAGED_AGENT: u32 = 30179; +/// Accumulator fold spec (parameterized replaceable, author-only). +/// +/// The saved definition of one fold: selection, schema, model, and +/// instructions, addressed by `(pubkey, kind, d_tag = fold name)` with +/// last-write-wins semantics. Content is the `buzz-accumulator` `FoldSpec` +/// JSON, NIP-44 encrypted from the author's key to itself; reads are +/// author-only (see [`AUTHOR_ONLY_KINDS`]). +pub const KIND_FOLD_SPEC: u32 = 30640; + +/// Accumulator artifact version (regular/immutable, author-only). +/// +/// One versioned output of a fold run. Content is the `buzz-accumulator` +/// `ArtifactPayload` JSON (output, shown signal ids, coverage window, +/// provenance), NIP-44 encrypted from the author's key to itself; reads are +/// author-only (see [`AUTHOR_ONLY_KINDS`]). Versions are ordinary immutable +/// events, so a fold's artifact history is append-only by construction. +pub const KIND_FOLD_ARTIFACT: u32 = 4640; + /// Kinds whose stored events are readable only by their author. /// /// The relay must never reveal the existence, count, tags, content, schedule, @@ -130,6 +148,8 @@ pub const AUTHOR_ONLY_KINDS: &[u32] = &[ KIND_EVENT_REMINDER, KIND_PUSH_LEASE, KIND_PRIVATE_MANAGED_AGENT, + KIND_FOLD_SPEC, + KIND_FOLD_ARTIFACT, ]; /// Kinds that require a result-level read gate beyond the filter-layer @@ -766,6 +786,8 @@ pub const ALL_KINDS: &[u32] = &[ KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_PROJECT, + KIND_FOLD_SPEC, + KIND_FOLD_ARTIFACT, ]; /// Returns `true` if `kind` is in the ephemeral range (20000–29999). @@ -867,6 +889,13 @@ const _: () = assert!(is_parameterized_replaceable(KIND_DM_VISIBILITY)); // 3062 const _: () = assert!(is_parameterized_replaceable(KIND_PROJECT)); // 30621 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_THREAD_SUMMARY)); // 39005 ∈ 30000–39999 const _: () = assert!(is_parameterized_replaceable(KIND_WINDOW_BOUNDS)); // 39006 ∈ 30000–39999 +const _: () = assert!(is_parameterized_replaceable(KIND_FOLD_SPEC)); // 30640 ∈ 30000–39999 + // Artifact versions are immutable history: regular-range, never replaceable. +const _: () = assert!( + !is_parameterized_replaceable(KIND_FOLD_ARTIFACT) + && !is_replaceable(KIND_FOLD_ARTIFACT) + && !is_ephemeral(KIND_FOLD_ARTIFACT) +); // 4640 regular // Compile-time: NIP-34 parameterized replaceable kinds are in the correct range. const _: () = assert!( diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 5901512504..7ce1c8e0a7 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -702,7 +702,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 44); + assert_eq!(migrations.len(), 45); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -910,8 +910,20 @@ mod postgres_tests { assert!(migrations[32].sql.as_str().contains("kind = 30179")); assert!(migrations[32].sql.as_str().contains("search_tsv")); assert!(!migrations[0].sql.as_str().contains("30179")); - assert!(include_str!("../../../../schema/schema.sql") - .contains("kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200)")); + + // Accumulator kinds 30640 (fold spec) + 4640 (artifact version) FTS + // exclusion (0041): author-only NIP-44 ciphertext, same + // wrap-the-existing-expression shape as 0014/0033. + assert_eq!(migrations[40].version, 41); + assert!(migrations[40] + .sql + .as_str() + .contains("kind IN (4640, 30640)")); + assert!(migrations[40].sql.as_str().contains("search_tsv")); + assert!(!migrations[0].sql.as_str().contains("30640")); + assert!(include_str!("../../../../schema/schema.sql").contains( + "kind IN (1059, 4640, 30179, 30300, 30350, 30622, 30640, 44100, 44101, 44200)" + )); // Public push-gateway authority is intentionally deployment-global and // durable: immediate revocation and hostile-relay admission cannot be diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ee1d0312be..8b395795dd 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -15,26 +15,26 @@ use buzz_core::kind::{ is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, KIND_AGENT_TURN_METRIC, KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, KIND_AUTH, KIND_BOOKMARK_LIST, KIND_BOOKMARK_SET, KIND_CANVAS, KIND_CONTACT_LIST, KIND_DELETION, KIND_DM_ADD_MEMBER, KIND_DM_HIDE, KIND_DM_OPEN, - KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, KIND_FOLLOW_SET, KIND_FORUM_COMMENT, - KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, KIND_GIT_ISSUE, KIND_GIT_PATCH, - KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, - KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, - KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, KIND_HUDDLE_PARTICIPANT_JOINED, - KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, KIND_IA_ARCHIVE_REQUEST, - KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, KIND_MEMBER_ADDED_NOTIFICATION, - KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, KIND_MODERATION_RESOLVE_REPORT, - KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, - KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, KIND_NIP29_DELETE_GROUP, - KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, KIND_NIP29_LEAVE_REQUEST, - KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, KIND_NIP43_LEAVE_REQUEST, - KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, KIND_PRESENCE_UPDATE, - KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_PROJECT, KIND_REACTION, - KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, - KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, - KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, - KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, - RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, - RELAY_ADMIN_SET_WORKSPACE_PROFILE, + KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, KIND_FOLD_ARTIFACT, KIND_FOLD_SPEC, + KIND_FOLLOW_SET, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, + KIND_GIT_ISSUE, KIND_GIT_PATCH, KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, + KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, + KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, + KIND_HUDDLE_PARTICIPANT_JOINED, KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, + KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, + KIND_MODERATION_RESOLVE_REPORT, KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, + KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, + KIND_NIP29_DELETE_GROUP, KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, + KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, + KIND_NIP43_LEAVE_REQUEST, KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, + KIND_PRESENCE_UPDATE, KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, + KIND_PROJECT, KIND_REACTION, KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, + KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, + KIND_STREAM_REMINDER, KIND_TEAM, KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, + KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, + RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; @@ -440,7 +440,10 @@ fn required_scope_for_kind(kind: u32, event: &Event) -> Result Ok(Scope::MessagesWrite), KIND_CONTACT_LIST | KIND_READ_STATE | KIND_USER_STATUS | KIND_AGENT_ENGRAM | KIND_EVENT_REMINDER | KIND_PERSONA | KIND_TEAM | KIND_MANAGED_AGENT - | KIND_PRIVATE_MANAGED_AGENT | KIND_TEAM_CATALOG | super::push_lease::KIND_PUSH_LEASE => { + | KIND_PRIVATE_MANAGED_AGENT | KIND_TEAM_CATALOG | super::push_lease::KIND_PUSH_LEASE + // Accumulator fold specs + artifact versions: user-owned personal + // state, author-only reads, NIP-44 content encrypted to self. + | KIND_FOLD_SPEC | KIND_FOLD_ARTIFACT => { Ok(Scope::UsersWrite) } // NIP-AM: agent turn metrics are agent-authored global events (encrypted to owner). @@ -700,6 +703,12 @@ pub(crate) fn is_global_only_kind(kind: u32) -> bool { | KIND_AGENT_TURN_METRIC // NIP-PL leases are author-owned, addressable global state. | super::push_lease::KIND_PUSH_LEASE + // Accumulator fold specs (30640, keyed by (pubkey, kind, d_tag)) + // and artifact versions (4640, immutable) are author-owned + // personal state. Selections live encrypted inside the payload — + // a stray `h` tag must not channel-scope them. + | KIND_FOLD_SPEC + | KIND_FOLD_ARTIFACT ) } diff --git a/migrations/0045_fold_kinds_fts.sql b/migrations/0045_fold_kinds_fts.sql new file mode 100644 index 0000000000..7ade2860c5 --- /dev/null +++ b/migrations/0045_fold_kinds_fts.sql @@ -0,0 +1,39 @@ +-- Accumulator kinds 30640 (fold spec) and 4640 (artifact version) carry +-- NIP-44 ciphertext encrypted from the author's key to itself and are +-- author-only (AUTHOR_ONLY_KINDS). Exclude them from full-text search so +-- the ciphertext is never tokenized into search_tsv, where a NIP-50 search +-- from any member could otherwise probe it. +-- +-- Same shape as 0014 (kind:30350) and 0033 (kind:30179): PostgreSQL cannot +-- alter a generated expression in place, so capture the current expression, +-- drop the column, and re-add it wrapped with the new exclusion. Every other +-- kind keeps whatever policy the database had before. +-- +-- Operational cost: DROP COLUMN + ADD ... GENERATED ... STORED rewrites the +-- events heap and rebuilds the GIN index under an ACCESS EXCLUSIVE lock — +-- see the note on 0033. Operators with large brownfield databases should +-- schedule a window. +DO $$ +DECLARE + existing_expression TEXT; +BEGIN + SELECT pg_get_expr(d.adbin, d.adrelid) + INTO existing_expression + FROM pg_attrdef d + JOIN pg_attribute a + ON a.attrelid = d.adrelid + AND a.attnum = d.adnum + WHERE d.adrelid = 'events'::regclass + AND a.attname = 'search_tsv'; + + IF existing_expression IS NULL THEN + RAISE EXCEPTION 'events.search_tsv generated expression not found'; + END IF; + + ALTER TABLE events DROP COLUMN search_tsv; + EXECUTE format( + 'ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (CASE WHEN kind IN (4640, 30640) THEN NULL::tsvector ELSE (%s) END) STORED', + existing_expression + ); + CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv); +END $$; diff --git a/schema/schema.sql b/schema/schema.sql index 0950812562..d5fb35e39b 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -219,9 +219,9 @@ CREATE TABLE events ( -- Privacy: encrypted/private routing wrappers and p-gated membership notices -- must never be discoverable through NIP-50 full-text search. NULL tsvector -- never matches `@@`. - -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033). + -- Keep in sync with migrations (final state: 0001 + 0005 + 0014 + 0033 + 0041). search_tsv TSVECTOR GENERATED ALWAYS AS ( - CASE WHEN kind IN (1059, 30179, 30300, 30350, 30622, 44100, 44101, 44200) THEN NULL::tsvector + CASE WHEN kind IN (1059, 4640, 30179, 30300, 30350, 30622, 30640, 44100, 44101, 44200) THEN NULL::tsvector ELSE to_tsvector('simple', content) END ) STORED,