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 <noreply@anthropic.com>
Signed-off-by: Riley Crane <rileycrane@squareup.com>
This commit is contained in:
Riley Crane
2026-08-28 16:06:50 -07:00
committed by Sol
parent d595806fc3
commit 62f0d68761
20 changed files with 2095 additions and 26 deletions

11
Cargo.lock generated
View File

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

View File

@@ -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",

View File

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

View File

@@ -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<String>,
/// Earliest shown signal timestamp (unix seconds); `None` when the run
/// showed no signals (config-only rerun).
pub coverage_since: Option<i64>,
/// Half-open end of the covered window: latest shown timestamp + 1;
/// `None` when the run showed no signals.
pub coverage_until: Option<i64>,
/// 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,
}

View File

@@ -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),
}

View File

@@ -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<u64>,
/// The estimate being fitted.
pub est_input_tokens: u64,
/// Whether it fits; `None` when the window is unknown.
pub fits: Option<bool>,
/// `window - estimate`; negative when overfull, `None` when unknown.
pub headroom_tokens: Option<i64>,
}
/// 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<f64>,
/// 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));
}
}

View File

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

View File

@@ -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<ShownSignal>,
/// 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<String>,
fetched: Vec<Signal>,
names: &BTreeMap<String, String>,
) -> Result<Plan, Error> {
if schema::builtin(&spec.schema).is_none() {
return Err(Error::InvalidSpec(format!(
"unknown schema {:?}",
spec.schema
)));
}
let signals = materialize(fetched);
let new: Vec<Signal> = 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::<Vec<_>>()
.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<ArtifactPayload, Error> {
let sch = schema::builtin(&spec.schema)
.ok_or_else(|| Error::InvalidSpec(format!("unknown schema {:?}", spec.schema)))?;
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);
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<Signal> = (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(_))
));
}
}

View File

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

View File

@@ -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:<id>], 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());
}
}

View File

@@ -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<String>,
/// Author pubkeys (hex).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub authors: Vec<String>,
/// Event kinds; empty means [`DEFAULT_SIGNAL_KINDS`].
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub kinds: Vec<u32>,
}
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<u32> {
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<serde_json::Value> {
let kinds = self.effective_kinds();
let until = until_exclusive.saturating_sub(1);
let base = |extra: &mut serde_json::Map<String, serde_json::Value>| {
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<Signal>) -> Vec<Signal> {
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());
}
}

View File

@@ -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<String>,
}

View File

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

View File

@@ -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<ShownSignal>,
}
/// 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, String>) -> 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::<Vec<_>>()
.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<String, String>,
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::<String>())
};
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::<Vec<_>>()
.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 <message>: 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<_>>(),
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 <kind-40002>:"));
}
#[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());
}
}

View File

@@ -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<usize> = 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<String> {
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:<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.
pub fn cited_event_ids(citation_text: &str) -> BTreeSet<String> {
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<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;
}
}
}
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::<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 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<_>>(), 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'))
);
}
}

View File

@@ -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!(

View File

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

View File

@@ -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<Scope, &'static s
KIND_TEXT_NOTE | KIND_LONG_FORM => 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
)
}

View File

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

View File

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