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

accumulator: three-noun algebra — selection absorbs when, publish verb, schema deleted

Browse Source 
Design of record (Riley's refinement): Selection = frozen-or-live
description of a signal set (who x what x WHEN — since/until_exclusive
live on the selection itself; pinned end = frozen, open = live). Fold =
name + selection + model + instructions, a factory for artifacts. A
frozen fold runs until covered then is cached forever; a live fold is
never done. Rollup machinery stays deleted: publish an artifact back
into a channel and it becomes a signal a later fold can select.

- selection: since/until_exclusive fields; canonicalize rejects empty
  windows; resolve_window(clamp, now) — request bounds can only narrow
  (runs stay pinned to priced preflight windows; frozen selections never
  read outside their freeze). Dead code deleted (compile_filters,
  describe).
- spec/artifact/run: schema label deleted end to end — the output is
  verbatim, a label describing its shape was dead weight. Legacy JSON
  with the field still loads (unknown-field tolerance, test-covered);
  config_matches = model + prompt hash + selection.
- daemon/folds: plan/preflight/run take a WindowClamp instead of a
  resolved window; default_window deleted. New test: frozen fold
  completes then stays cached forever even under a wide clamp.
- daemon/publish (new): Publisher trait + RelayPublisher — short-lived
  authenticated connection per publish, same identity as the mirror.
- http: POST /folds/{name}/artifacts/{version}/publish with a taint
  guard (chain-union channels must equal the target unless
  allow_cross_channel is passed deliberately); preview/select/preflight/
  run resolve through the selection's own window; PutFoldBody loses
  schema. FakePublisher tests for footer, guard, and 404.
- ui/api.ts: type-sync only (schema fields dropped, Selection window
  fields added) — client work is Sol's.
- crate README: the three nouns, honesty invariants, run recipe, API
  table.

Gates: cargo test -p buzz-accumulator 73/73, clippy -D warnings clean,
fmt clean, ui tsc + build clean. Live-smoked on a scratch mirror:
live window ends at now+1, wide clamp cannot widen a frozen fold
(window == freeze exactly), empty window 422s, cross-channel publish
refuses with the deliberate-override hint.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Fizz <fizz@agents.buzz>
This commit is contained in:
Fizz
2026-09-01 15:22:36 -07:00
committed by Sol
parent 0722066d73
commit 3741229235
14 changed files with 604 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -849,6 +849,7 @@ dependencies = [
"tower",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]

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

@@ -27,6 +27,7 @@ daemon = [
"dep:tokio",
"dep:tracing",
"dep:tracing-subscriber",
"dep:uuid",
]
[dependencies]
@@ -52,6 +53,7 @@ sqlx = { workspace = true, features = ["sqlite"], optional = true }
tokio = { workspace = true, optional = true }
tracing = { workspace = true, optional = true }
tracing-subscriber = { workspace = true, optional = true }
uuid = { workspace = true, optional = true }
[target.'cfg(unix)'.dependencies]
libc = "0.2"

100
crates/buzz-accumulator/README.md Normal file
View File

@@ -0,0 +1,100 @@
# buzz-accumulator
Fold engine + standalone daemon: mirrors everything your key can see from the
relay into local SQLite, and folds selections of that stream into small,
always-current, provenance-carrying artifacts.
## The three nouns
Signals — raw relay events, never rewritten — are the substrate. On top of
them:
- **Selection** — a frozen-or-live description of a signal set: who × what ×
when (`channels`, `authors`, `kinds`, plus the selection's own `since` /
`until_exclusive`). A pinned `until_exclusive` **freezes** it; an open end
means **live** — "and whatever comes next".
- **Fold** — name + selection + model + instructions: a **factory for
artifacts**. Each run computes `artifact' = fold(artifact, new_signals)`.
A frozen selection makes the fold run until its set is covered, then it is
done forever (every further preflight says *cached*). A live fold is never
done — the prior version rides along and new signals keep folding in.
- **Artifact** — the model's response, **verbatim**, persisted as an
immutable append-only version chain. Provenance (`shown_ids`, coverage
window, model, prompt hash) is engine-computed from what the model was
actually shown — never parsed out of model output.
There is no rollup machinery: **publish** an artifact back into a channel and
it becomes a signal again, so a later fold can simply select it. Folds all the
way down.
Coverage is **per fold**: the same events can feed any number of folds (one
summarizes yesterday, another counts how often "Tua" was said). Within one
fold's chain, an event folds in exactly once. Note the corollary: editing a
fold's instructions does not re-fold already-covered events — a new lens is a
new fold.
## Honesty invariants
- Preflight is $0: plans and estimates are computed without any model call;
an unknown model's window fit is an honest `null`, never a guess.
- Coverage records exactly the signals the model was shown. An oversized
window truncates honestly and the remainder stays pending; unread signals
are never sealed as covered.
- Runs are pinned to their priced window: the optional request clamp can only
**narrow** the selection's own window, so what you priced is what you run.
- A frozen selection never reads outside its freeze, no matter what clamp the
caller sends.
- The output is free-form; `[event:<id>]` citations in it are
reader-verifiable links (resolve via `GET /events/{id}`), not a validated
contract.
## Running it
```sh
# daemon (loopback HTTP on 127.0.0.1:4640; mirror in ~/.buzz-accumulator/)
BUZZ_PRIVATE_KEY=<nsec-or-hex> cargo run -p buzz-accumulator
# lab UI (its own pnpm workspace; proxies /api to the daemon)
cd crates/buzz-accumulator/ui && pnpm install && pnpm dev # → http://localhost:5173
```
The relay is pinned to the team relay by default; the generic `BUZZ_RELAY_URL`
env var is deliberately ignored (override with `--relay` /
`BUZZ_ACCUMULATOR_RELAY_URL`). Useful overrides: `BUZZ_ACCUMULATOR_DB`,
`BUZZ_ACCUMULATOR_HTTP_ADDR`.
The HTTP API is **loopback-only and unauthenticated by design** —
`POST /folds/{name}/run` spends real money and `publish` posts real messages,
so any non-loopback bind must add authentication first.
## HTTP API
| Route | What |
|---|---|
| `GET /status` | Connection, backfill, mirror counts. |
| `GET /channels` | Discovered channels. |
| `GET /events/{id}` | One mirrored event (the citation-chip endpoint). |
| `POST /select/preview` | `{selection, since?, until_exclusive?}` → count, size, daily rhythm. $0. |
| `POST /select/events` | Same body + `limit`, `after` keyset cursor → the actual events, paged. |
| `GET /folds` | All fold specs. |
| `PUT /folds/{name}` | `{selection, model, instructions?, meta?}` — create/update a fold. |
| `GET /folds/{name}` | Spec + chain length + latest artifact summary. |
| `DELETE /folds/{name}` | Delete the spec (artifacts are history and survive). |
| `POST /folds/{name}/preflight` | `{since?, until_exclusive?, include_input?}` → cached / stalled / ready with exact cost estimate and (opt-in) the exact model input. $0. |
| `POST /folds/{name}/run` | Same window clamp → runs the model, appends one artifact version. **Spends money.** |
| `GET /folds/{name}/artifacts` | The version chain (summaries). |
| `GET /folds/{name}/artifacts/{version}` | One full artifact with provenance. |
| `POST /folds/{name}/artifacts/{version}/publish` | `{channel, allow_cross_channel?}` → posts the artifact into the channel as a message. Guarded: refuses unless the artifact's chain has only ever read that channel, unless crossed deliberately. |
`since` / `until_exclusive` on preview/preflight/run are a **clamp**: the
selection's own window is authoritative and the clamp can only narrow it.
## Layout
- `src/` — pure engine (selection, spec, plan/complete run, estimate,
transcript). No relay I/O, no storage.
- `src/daemon/` — the standalone app (`daemon` feature, on by default):
relay sync → SQLite mirror → HTTP API → relay publish. Engine consumers
depend with `default-features = false`.
- `ui/` — the vite + React lab client. Deliberately its own pnpm workspace
root, invisible to the repo's `just ci`.

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

@@ -19,7 +19,7 @@ pub struct ArtifactPayload {
pub fold: String,
/// 1-based version number; each run of the fold increments it.
pub version: u32,
/// Full artifact document (after the engine splice).
/// The model's response, verbatim.
pub output: String,
/// Exactly the signal ids the model was shown this run.
pub shown_ids: Vec<String>,
@@ -45,8 +45,6 @@ pub struct ArtifactPayload {
pub channels: Vec<String>,
/// 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

104
crates/buzz-accumulator/src/daemon/folds.rs
View File

@@ -135,17 +135,25 @@ impl Drop for RunToken {
}
}
/// Loads spec + chain and plans a run over `[since, until_exclusive)`.
/// An optional narrowing of the selection's own window, `[since,
/// until_exclusive)`. This is how a run stays pinned to the exact window its
/// preflight priced; it can only narrow, never widen (the selection's window
/// wins by intersection).
pub type WindowClamp = (Option<i64>, Option<i64>);
/// Loads spec + chain and plans a run over the selection's window (narrowed
/// by `clamp`). Returns the resolved window alongside the plan.
async fn plan(
store: &Store,
name: &str,
since: i64,
until_exclusive: i64,
) -> Result<(FoldSpec, Vec<ArtifactPayload>, Plan), FoldError> {
clamp: WindowClamp,
) -> Result<(FoldSpec, Vec<ArtifactPayload>, Plan, (i64, i64)), FoldError> {
let spec = store
.get_fold(name)
.await?
.ok_or_else(|| FoldError::NotFound(name.to_string()))?;
let now = chrono::Utc::now().timestamp();
let (since, until_exclusive) = spec.selection.resolve_window(clamp.0, clamp.1, now);
let chain = store.artifacts(name).await?;
let covered: BTreeSet<String> = chain
.iter()
@@ -157,20 +165,19 @@ async fn plan(
let authors: BTreeSet<String> = fetched.iter().map(|s| s.pubkey.clone()).collect();
let names = store.names(&authors).await?;
let plan = plan_run(&spec, chain.last(), &covered, fetched, &names)?;
Ok((spec, chain, plan))
Ok((spec, chain, plan, (since, until_exclusive)))
}
/// Zero-spend preflight: what would a run over this window do, and at what
/// estimated cost? With `include_input`, the Ready outcome carries the exact
/// string the model would receive.
/// Zero-spend preflight: what would a run do, and at what estimated cost?
/// With `include_input`, the Ready outcome carries the exact string the model
/// would receive.
pub async fn preflight(
store: &Store,
name: &str,
since: i64,
until_exclusive: i64,
clamp: WindowClamp,
include_input: bool,
) -> Result<PreflightOutcome, FoldError> {
let (_, _, plan) = plan(store, name, since, until_exclusive).await?;
let (_, _, plan, window) = plan(store, name, clamp).await?;
Ok(match plan {
Plan::Cached => PreflightOutcome::Cached,
Plan::Stalled { reason, pending } => PreflightOutcome::Stalled { reason, pending },
@@ -179,23 +186,22 @@ pub async fn preflight(
pending: rp.pending,
truncated: rp.truncated,
estimate: rp.estimate,
window: (since, until_exclusive),
window,
model_input: include_input.then_some(rp.model_input),
},
})
}
/// Runs one fold turn over `[since, until_exclusive)`.
/// Runs one fold turn over the selection's window (narrowed by `clamp`).
pub async fn run_fold(
store: &Store,
runner: Arc<dyn FoldRunner + Send + Sync>,
guard: &RunGuard,
name: &str,
since: i64,
until_exclusive: i64,
clamp: WindowClamp,
) -> Result<RunOutcome, FoldError> {
let _token = guard.acquire(name)?;
let (spec, chain, plan) = plan(store, name, since, until_exclusive).await?;
let (spec, chain, plan, _) = plan(store, name, clamp).await?;
let run_plan = match plan {
Plan::Cached => return Ok(RunOutcome::Cached),
Plan::Stalled { reason, pending } => return Ok(RunOutcome::Stalled { reason, pending }),
@@ -229,12 +235,6 @@ pub async fn run_fold(
})
}
/// Default window for preflight/run when the caller does not narrow it:
/// everything up to now.
pub fn default_window() -> (i64, i64) {
(0, chrono::Utc::now().timestamp() + 1)
}
#[cfg(test)]
mod tests {
use super::*;
@@ -255,10 +255,8 @@ mod tests {
name: "weekly".into(),
selection: Selection {
channels: vec![CHANNEL.into()],
authors: vec![],
kinds: vec![],
..Selection::default()
},
schema: "channel-digest@v1".into(),
model: "test-model".into(),
instructions: "digest the channel".into(),
meta: None,
@@ -283,7 +281,7 @@ mod tests {
#[tokio::test]
async fn preflight_prices_before_any_spend() {
let store = store_with_fold_and_signal().await;
let out = preflight(&store, "weekly", 0, 1_000, false)
let out = preflight(&store, "weekly", (None, None), false)
.await
.expect("preflight");
match out {
@@ -306,7 +304,7 @@ mod tests {
#[tokio::test]
async fn preflight_can_expose_the_exact_model_input() {
let store = store_with_fold_and_signal().await;
let out = preflight(&store, "weekly", 0, 1_000, true)
let out = preflight(&store, "weekly", (None, None), true)
.await
.expect("preflight");
match out {
@@ -327,7 +325,7 @@ mod tests {
let output = format!("Just a paragraph citing [event:{id}], no headings.");
let runner = Arc::new(FakeRunner(output.clone()));
let guard = RunGuard::default();
let out = run_fold(&store, runner.clone(), &guard, "weekly", 0, 1_000)
let out = run_fold(&store, runner.clone(), &guard, "weekly", (None, None))
.await
.expect("run");
match out {
@@ -342,17 +340,63 @@ mod tests {
other => panic!("expected folded, got {other:?}"),
}
// Same window again: engine reports cached, no new version.
let again = run_fold(&store, runner, &guard, "weekly", 0, 1_000)
let again = run_fold(&store, runner, &guard, "weekly", (None, None))
.await
.expect("rerun");
assert!(matches!(again, RunOutcome::Cached), "got {again:?}");
assert_eq!(store.artifacts("weekly").await.expect("chain").len(), 1);
}
#[tokio::test]
async fn frozen_fold_completes_then_stays_cached_forever() {
let store = store_with_fold_and_signal().await;
// Freeze the selection around the seeded signal (created_at = 100).
let mut spec = store
.get_fold("weekly")
.await
.expect("get")
.expect("present");
spec.selection.since = Some(0);
spec.selection.until_exclusive = Some(1_000);
spec.validate().expect("valid");
store.put_fold(&spec, 2).await.expect("update");
let runner = Arc::new(FakeRunner("the day, summarized".into()));
let guard = RunGuard::default();
let out = run_fold(&store, runner.clone(), &guard, "weekly", (None, None))
.await
.expect("run");
assert!(matches!(out, RunOutcome::Folded { .. }), "got {out:?}");
// Fully covered: the frozen fold is done forever — no clamp needed,
// and later events outside the freeze must never wake it.
store
.upsert_events(&[super::super::store::StoredEvent {
id: "f".repeat(64),
channel: Some(CHANNEL.into()),
pubkey: "a".repeat(64),
kind: 9,
created_at: 5_000,
content: "after the freeze".into(),
raw: "{}".into(),
}])
.await
.expect("late event");
let pf = preflight(&store, "weekly", (None, None), false)
.await
.expect("preflight");
assert!(matches!(pf, PreflightOutcome::Cached), "got {pf:?}");
// A clamp can only narrow, never widen past the freeze.
let pf = preflight(&store, "weekly", (Some(0), Some(10_000)), false)
.await
.expect("preflight");
assert!(matches!(pf, PreflightOutcome::Cached), "got {pf:?}");
}
#[tokio::test]
async fn unknown_fold_is_not_found() {
let store = Store::open(":memory:").await.expect("open");
let err = preflight(&store, "nope", 0, 10, false)
let err = preflight(&store, "nope", (None, None), false)
.await
.expect_err("missing fold");
assert!(matches!(err, FoldError::NotFound(_)));

224
crates/buzz-accumulator/src/daemon/http.rs
View File

@@ -17,7 +17,8 @@ use serde_json::{json, Value};
use crate::runner::FoldRunner;
use crate::{FoldSpec, Selection};
use super::folds::{self, FoldError, RunGuard};
use super::folds::{self, FoldError, RunGuard, WindowClamp};
use super::publish::Publisher;
use super::status::StatusRegistry;
use super::store::Store;
@@ -32,6 +33,8 @@ pub struct AppState {
pub runner: Arc<dyn FoldRunner + Send + Sync>,
/// Per-fold run serialization (concurrent runs 409 instead of racing).
pub runs: RunGuard,
/// Publishes artifacts back into channels as messages.
pub publisher: Arc<dyn Publisher>,
}
/// Builds the router. Separated from serving for tests.
@@ -51,6 +54,10 @@ pub fn router(state: AppState) -> Router {
.route("/folds/{name}/run", post(run_fold))
.route("/folds/{name}/artifacts", get(list_artifacts))
.route("/folds/{name}/artifacts/{version}", get(get_artifact))
.route(
"/folds/{name}/artifacts/{version}/publish",
post(publish_artifact),
)
.with_state(state)
}
@@ -116,8 +123,9 @@ async fn get_channels(State(s): State<AppState>) -> Result<Json<Value>, ApiError
Ok(Json(json!({ "channels": channels })))
}
/// Window bounds shared by preview/preflight/run bodies. Omitted bounds mean
/// "everything up to now".
/// Optional window clamp shared by preview/preflight/run bodies. The
/// selection's own window is authoritative; these bounds can only narrow it
/// (this is how a run stays pinned to its priced preflight window).
#[derive(Debug, Default, serde::Deserialize)]
struct WindowBody {
since: Option<i64>,
@@ -125,12 +133,8 @@ struct WindowBody {
}
impl WindowBody {
fn resolve(&self) -> (i64, i64) {
let (default_since, default_until) = folds::default_window();
(
self.since.unwrap_or(default_since),
self.until_exclusive.unwrap_or(default_until),
)
fn clamp(&self) -> WindowClamp {
(self.since, self.until_exclusive)
}
}
@@ -149,7 +153,9 @@ async fn select_preview(
) -> Result<Json<Value>, ApiError> {
let mut selection = body.selection;
selection.canonicalize()?;
let (since, until) = body.window.resolve();
let now = chrono::Utc::now().timestamp();
let (since, until) =
selection.resolve_window(body.window.since, body.window.until_exclusive, now);
let signals = s.store.query_signals(&selection, since, until).await?;
let total_chars: usize = signals.iter().map(|sig| sig.content.len()).sum();
// Daily histogram (UTC day starts). Sparse: days with no matches are absent.
@@ -199,7 +205,9 @@ async fn select_events(
) -> Result<Json<Value>, ApiError> {
let mut selection = body.selection;
selection.canonicalize()?;
let (since, until) = body.window.resolve();
let now = chrono::Utc::now().timestamp();
let (since, until) =
selection.resolve_window(body.window.since, body.window.until_exclusive, now);
let limit = body.limit.unwrap_or(100).clamp(1, 500);
let after = body.after.as_ref().map(|c| (c.created_at, c.id.as_str()));
// Fetch one extra row purely to learn whether a next page exists.
@@ -281,8 +289,6 @@ struct PutFoldBody {
model: String,
/// Defaults to the built-in running-digest task.
instructions: Option<String>,
/// Provenance label recorded on artifacts; defaults to `freeform@v1`.
schema: Option<String>,
/// Free-form client-owned JSON, stored verbatim and never read by the
/// engine (see [`FoldSpec::meta`]).
meta: Option<serde_json::Value>,
@@ -296,9 +302,6 @@ async fn put_fold(
let mut spec = FoldSpec {
name,
selection: body.selection,
schema: body
.schema
.unwrap_or_else(|| crate::spec::FREEFORM_SCHEMA.to_string()),
model: body.model,
instructions: body
.instructions
@@ -343,8 +346,7 @@ async fn preflight_fold(
body: Option<Json<PreflightBody>>,
) -> Result<Json<Value>, ApiError> {
let Json(body) = body.unwrap_or_default();
let (since, until) = body.window.resolve();
let out = folds::preflight(&s.store, &name, since, until, body.include_input).await?;
let out = folds::preflight(&s.store, &name, body.window.clamp(), body.include_input).await?;
Ok(Json(
serde_json::to_value(out).unwrap_or_else(|_| json!({})),
))
@@ -355,10 +357,8 @@ async fn run_fold(
Path(name): Path<String>,
body: Option<Json<WindowBody>>,
) -> Result<Json<Value>, ApiError> {
let (since, until) = body
.map(|Json(w)| w.resolve())
.unwrap_or_else(|| WindowBody::default().resolve());
let out = folds::run_fold(&s.store, s.runner.clone(), &s.runs, &name, since, until).await?;
let clamp = body.map(|Json(w)| w.clamp()).unwrap_or((None, None));
let out = folds::run_fold(&s.store, s.runner.clone(), &s.runs, &name, clamp).await?;
Ok(Json(
serde_json::to_value(out).unwrap_or_else(|_| json!({})),
))
@@ -373,7 +373,6 @@ fn artifact_summary(a: &crate::ArtifactPayload) -> Value {
"coverage_until": a.coverage_until,
"channels": a.channels,
"model": a.model,
"schema": a.schema,
"truncated": a.truncated,
})
}
@@ -404,6 +403,65 @@ async fn get_artifact(
))
}
#[derive(Debug, serde::Deserialize)]
struct PublishBody {
/// Channel UUID to post into.
channel: String,
/// The taint guard requires the artifact's chain to have read only the
/// target channel; set this to cross that line deliberately.
#[serde(default)]
allow_cross_channel: bool,
}
/// Publishes an artifact version into a channel as an ordinary message —
/// the composition verb. The mirror's live tail picks the message back up,
/// so a later fold can select it: folds all the way down.
async fn publish_artifact(
State(s): State<AppState>,
Path((name, version)): Path<(String, u32)>,
Json(body): Json<PublishBody>,
) -> Result<Json<Value>, ApiError> {
let artifact = s.store.artifact(&name, version).await?.ok_or_else(|| {
ApiError(
StatusCode::NOT_FOUND,
format!("artifact not found: {name} v{version}"),
)
})?;
let channel = body.channel.trim().to_lowercase();
// Taint guard: the chain-union `channels` is what this artifact has ever
// read. Publishing into a channel it didn't come from can leak another
// channel's content, so that path is opt-in only.
if !body.allow_cross_channel && artifact.channels != vec![channel.clone()] {
return Err(ApiError(
StatusCode::UNPROCESSABLE_ENTITY,
format!(
"artifact provenance covers channels {:?}, not just {channel} — \
pass allow_cross_channel to publish it there deliberately",
artifact.channels
),
));
}
let content = format!(
"{}\n\n— fold {} v{} · folded {} event(s) · {}",
artifact.output.trim_end(),
artifact.fold,
artifact.version,
artifact.shown_ids.len(),
artifact.model,
);
let event_id = s
.publisher
.publish(&channel, &content)
.await
.map_err(|e| ApiError(StatusCode::BAD_GATEWAY, e))?;
Ok(Json(json!({
"published": event_id,
"channel": channel,
"fold": name,
"version": version,
})))
}
#[cfg(test)]
mod tests {
use super::*;
@@ -420,12 +478,38 @@ mod tests {
}
}
/// Records publishes instead of touching a relay.
#[derive(Default)]
struct FakePublisher {
calls: std::sync::Mutex<Vec<(String, String)>>,
}
impl Publisher for FakePublisher {
fn publish<'a>(
&'a self,
channel: &'a str,
content: &'a str,
) -> futures_util::future::BoxFuture<'a, Result<String, String>> {
Box::pin(async move {
self.calls
.lock()
.expect("lock")
.push((channel.to_string(), content.to_string()));
Ok("e".repeat(64))
})
}
}
async fn test_state() -> AppState {
test_state_with(Arc::new(FakePublisher::default())).await
}
async fn test_state_with(publisher: Arc<FakePublisher>) -> AppState {
AppState {
store: Store::open(":memory:").await.expect("open"),
registry: StatusRegistry::new("wss://test", "ab", ":memory:", 0),
runner: Arc::new(NoRunner),
runs: RunGuard::default(),
publisher,
}
}
@@ -463,7 +547,6 @@ mod tests {
let resp = app.clone().oneshot(put).await.expect("resp");
assert_eq!(resp.status(), StatusCode::OK);
let saved = body_json(resp).await;
assert_eq!(saved["saved"]["schema"], "freeform@v1");
assert!(!saved["saved"]["instructions"]
.as_str()
.unwrap_or("")
@@ -669,4 +752,97 @@ mod tests {
.expect("resp");
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
}
fn artifact(channels: &[&str]) -> crate::ArtifactPayload {
crate::ArtifactPayload {
fold: "weekly".into(),
version: 1,
output: "the digest".into(),
shown_ids: vec!["s".repeat(64), "t".repeat(64)],
coverage_since: Some(100),
coverage_until: Some(201),
selection: Selection {
channels: channels.iter().map(|s| s.to_string()).collect(),
..Selection::default()
},
channels: channels.iter().map(|s| s.to_string()).collect(),
model: "haiku".into(),
prompt_sha256: "0".repeat(64),
truncated: false,
created_at: 1,
}
}
fn publish_req(body: Value) -> Request<Body> {
Request::post("/folds/weekly/artifacts/1/publish")
.header("content-type", "application/json")
.body(Body::from(body.to_string()))
.expect("req")
}
#[tokio::test]
async fn publish_posts_artifact_with_provenance_footer() {
let fake = Arc::new(FakePublisher::default());
let state = test_state_with(fake.clone()).await;
state
.store
.insert_artifact(&artifact(&[CHANNEL]))
.await
.expect("seed artifact");
let app = router(state);
let resp = app
.oneshot(publish_req(json!({ "channel": CHANNEL })))
.await
.expect("resp");
assert_eq!(resp.status(), StatusCode::OK);
let v = body_json(resp).await;
assert_eq!(v["published"], "e".repeat(64));
assert_eq!(v["channel"], CHANNEL);
let calls = fake.calls.lock().expect("lock");
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].0, CHANNEL);
assert!(calls[0].1.starts_with("the digest\n\n— fold weekly v1"));
assert!(calls[0].1.contains("2 event(s)"));
assert!(calls[0].1.contains("haiku"));
}
#[tokio::test]
async fn publish_refuses_cross_channel_unless_deliberate() {
let state = test_state().await;
// Chain has read a second channel; sharing into CHANNEL alone leaks it.
let other = "0e415c05-98cc-4c39-9a10-e07ba0479560";
let mut a = artifact(&[CHANNEL]);
a.channels = vec![other.to_string(), CHANNEL.to_string()];
state.store.insert_artifact(&a).await.expect("seed");
let app = router(state);
let resp = app
.clone()
.oneshot(publish_req(json!({ "channel": CHANNEL })))
.await
.expect("resp");
assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY);
let v = body_json(resp).await;
assert!(v["error"]
.as_str()
.expect("error string")
.contains("allow_cross_channel"));
// The same publish goes through when crossed deliberately.
let resp = app
.oneshot(publish_req(
json!({ "channel": CHANNEL, "allow_cross_channel": true }),
))
.await
.expect("resp");
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn publish_of_missing_artifact_is_404() {
let app = router(test_state().await);
let resp = app
.oneshot(publish_req(json!({ "channel": CHANNEL })))
.await
.expect("resp");
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
}
}

7
crates/buzz-accumulator/src/daemon/mod.rs
View File

@@ -15,6 +15,7 @@
pub mod folds;
pub mod http;
pub mod publish;
pub mod status;
pub mod store;
pub mod sync;
@@ -152,6 +153,11 @@ pub async fn run(cfg: Config) -> anyhow::Result<()> {
chrono::Utc::now().timestamp(),
);
let publisher = Arc::new(publish::RelayPublisher {
relay_url: cfg.relay.clone(),
keys: keys.clone(),
auth_tag: auth_tag.clone(),
});
let sync_cfg = sync::SyncConfig {
relay_url: cfg.relay.clone(),
keys,
@@ -164,6 +170,7 @@ pub async fn run(cfg: Config) -> anyhow::Result<()> {
registry,
runner: Arc::new(crate::SubprocessRunner::new()),
runs: folds::RunGuard::default(),
publisher,
};
let addr: std::net::SocketAddr = cfg
.http_addr

64
crates/buzz-accumulator/src/daemon/publish.rs Normal file
View File

@@ -0,0 +1,64 @@
//! Publishing artifacts back into channels as ordinary messages.
//!
//! This is the composition seam: a published artifact is just a kind-9
//! message in a channel the mirror watches, so another fold can select it —
//! folds all the way down, no rollup machinery needed.
use futures_util::future::BoxFuture;
use nostr::{Keys, Tag};
/// One publish, bounded end to end (connect + auth + EVENT + OK).
const PUBLISH_TIMEOUT_SECS: u64 = 30;
/// Publishes one message to a channel as the daemon identity.
pub trait Publisher: Send + Sync {
/// Publish `content` as a kind-9 message in `channel` (a channel UUID);
/// resolves to the accepted event id.
fn publish<'a>(
&'a self,
channel: &'a str,
content: &'a str,
) -> BoxFuture<'a, Result<String, String>>;
}
/// Real publisher: a short-lived authenticated connection per publish. Same
/// key as the mirror's read connection, publish-only, no REQs — the pattern
/// the relay expects (events bind to the connection's NIP-42 identity).
pub struct RelayPublisher {
/// Relay websocket URL (same pinned relay the mirror reads from).
pub relay_url: String,
/// Signing identity (the person's key).
pub keys: Keys,
/// Optional NIP-OA ownership tag (agent identities only).
pub auth_tag: Option<Tag>,
}
impl Publisher for RelayPublisher {
fn publish<'a>(
&'a self,
channel: &'a str,
content: &'a str,
) -> BoxFuture<'a, Result<String, String>> {
Box::pin(async move {
let channel = uuid::Uuid::parse_str(channel)
.map_err(|e| format!("channel is not a UUID: {e}"))?;
let event = buzz_sdk::build_message(channel, content, None, &[], false, &[])
.map_err(|e| format!("building message: {e}"))?
.sign_with_keys(&self.keys)
.map_err(|e| format!("signing message: {e}"))?;
let ok = buzz_ws_client::publish_event(
&self.relay_url,
event,
&self.keys,
self.auth_tag.as_ref(),
PUBLISH_TIMEOUT_SECS,
)
.await
.map_err(|e| format!("publishing to relay: {e}"))?;
if !ok.accepted {
return Err(format!("relay refused the message: {}", ok.message));
}
Ok(ok.event_id)
})
}
}

11
crates/buzz-accumulator/src/daemon/store.rs
View File

@@ -585,8 +585,7 @@ mod tests {
fn selection(channels: &[&str]) -> Selection {
Selection {
channels: channels.iter().map(|s| s.to_string()).collect(),
authors: vec![],
kinds: vec![],
..Selection::default()
}
}
@@ -635,7 +634,6 @@ mod tests {
selection: selection(&[]),
channels: vec![],
model: "m".into(),
schema: "channel-digest@v1".into(),
prompt_sha256: "0".repeat(64),
truncated: false,
created_at: 1,
@@ -670,12 +668,7 @@ mod tests {
let store = Store::open(":memory:").await.expect("open");
let mut spec = FoldSpec {
name: "weekly".into(),
selection: Selection {
channels: vec!["6ba7b810-9dad-11d1-80b4-00c04fd430c8".into()],
authors: vec![],
kinds: vec![],
},
schema: "channel-digest@v1".into(),
selection: selection(&["6ba7b810-9dad-11d1-80b4-00c04fd430c8"]),
model: "haiku".into(),
instructions: "digest".into(),
meta: None,

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

@@ -6,14 +6,21 @@
//! 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)`.
//! Three nouns, one recurrence (signals — raw relay events, never rewritten —
//! are the substrate underneath all three):
//! - **selection** — a frozen-or-live description of a signal set: who ×
//! what × when (channels, authors, kinds, and the selection's own window).
//! A pinned `until_exclusive` freezes it; an open end means "and whatever
//! comes next".
//! - **fold** — name + selection + model + instructions: a factory for
//! artifacts. `artifact' = fold(artifact, new_signals, spec)`. A frozen
//! selection makes the fold run until covered, then done forever; a live
//! one is never done.
//! - **artifact** — the model's response, verbatim and free-form, persisted
//! locally as an immutable, append-only version chain whose provenance
//! (exactly which signals went in, and when) is engine-computed.
//! - (a **lens** renders artifacts into UI; rendering is out of scope here.)
//! (exactly which signals went in, and when) is engine-computed. Publishing
//! an artifact back into a channel makes it a signal again — later folds
//! can select it, so composition needs no extra machinery.
//!
//! The engine modules are deliberately **pure**: they plan runs, price them,
//! and render transcripts — but perform no relay I/O and hold no storage.

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

@@ -96,7 +96,6 @@ pub fn plan_run(
.collect();
let config_matches = prior.is_some_and(|p| {
p.model == spec.model
&& p.schema == spec.schema
&& p.prompt_sha256 == spec.prompt_sha256()
&& p.selection == spec.selection
});
@@ -193,7 +192,6 @@ pub fn complete_run(
channels: channels.into_iter().collect(),
shown_ids,
model: spec.model.clone(),
schema: spec.schema.clone(),
prompt_sha256: spec.prompt_sha256(),
truncated: plan.truncated,
created_at,
@@ -214,10 +212,8 @@ mod tests {
name: "team-digest".to_string(),
selection: Selection {
channels: vec!["ch1".to_string()],
authors: vec![],
kinds: vec![],
..Selection::default()
},
schema: "channel-digest@v1".to_string(),
model: "haiku".to_string(),
instructions: "Maintain the digest.".to_string(),
meta: None,
@@ -250,7 +246,6 @@ mod tests {
selection: spec.selection.clone(),
channels: spec.selection.channels.clone(),
model: spec.model.clone(),
schema: spec.schema.clone(),
prompt_sha256: spec.prompt_sha256(),
truncated: false,
created_at: 1_700_000_100,

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

@@ -1,4 +1,4 @@
//! Selections: saveable queries over signals, compiled to relay filters.
//! Selections: frozen-or-live descriptions of a set of signals.
use serde::{Deserialize, Serialize};
@@ -8,15 +8,18 @@ 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.
/// A named-able description of a set of signals: who × what × when.
///
/// 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.
/// The *when* is part of the selection. A pinned `until_exclusive` makes the
/// selection **frozen** — it describes a fixed set that a fold works through
/// and is then done with forever. An open `until_exclusive` makes it **live**
/// — it keeps extending to "now", so a fold over it is never done.
///
/// v1 supports the shapes the mirror can answer directly: channels, authors,
/// and kinds. DMs are deliberately excluded — a selection never reads them.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Selection {
/// Channel ids (NIP-29 `h` tags). One relay filter is emitted per channel.
/// Channel ids (NIP-29 `h` tags).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub channels: Vec<String>,
/// Author pubkeys (hex).
@@ -25,6 +28,13 @@ pub struct Selection {
/// Event kinds; empty means [`DEFAULT_SIGNAL_KINDS`].
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub kinds: Vec<u32>,
/// Window start (unix seconds, inclusive). `None` = beginning of time.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub since: Option<i64>,
/// Window end (unix seconds, exclusive). `None` = live: the selection
/// keeps extending to "now". Pinned = frozen.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub until_exclusive: Option<i64>,
}
impl Selection {
@@ -64,10 +74,23 @@ impl Selection {
"kind {k} is outside the event-kind range"
)));
}
if let (Some(s), Some(u)) = (self.since, self.until_exclusive) {
if s >= u {
return Err(Error::InvalidSpec(format!(
"window is empty: since {s} must be before until_exclusive {u}"
)));
}
}
Ok(())
}
/// Effective kinds for relay filters (relay queries must always name kinds).
/// Whether the selection describes a fixed set (pinned end) rather than a
/// live one that keeps extending to "now".
pub fn is_frozen(&self) -> bool {
self.until_exclusive.is_some()
}
/// Effective kinds for signal queries (queries must always name kinds).
pub fn effective_kinds(&self) -> Vec<u32> {
if self.kinds.is_empty() {
DEFAULT_SIGNAL_KINDS.to_vec()
@@ -76,57 +99,26 @@ impl Selection {
}
}
/// Compile to NIP-01 relay filters over the half-open window
/// `[since, until_exclusive)`.
/// The concrete half-open window `[since, until_exclusive)` a run over
/// this selection reads, optionally narrowed by a caller clamp.
///
/// 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(
/// The clamp is how a run stays pinned to its priced preflight window: a
/// live selection resolves `until` to "now" at preflight time, and the run
/// passes that resolved window back as the clamp. Clamps can only narrow —
/// the intersection with the selection's own window is taken, so a frozen
/// selection never reads outside its freeze.
pub fn resolve_window(
&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(" · ")
clamp_since: Option<i64>,
clamp_until_exclusive: Option<i64>,
now: i64,
) -> (i64, i64) {
let since = self.since.unwrap_or(0).max(clamp_since.unwrap_or(0));
let until = self
.until_exclusive
.unwrap_or(now + 1)
.min(clamp_until_exclusive.unwrap_or(i64::MAX));
(since, until)
}
}
@@ -185,6 +177,13 @@ mod tests {
const CH_A: &str = "59ca5528-71ea-4a53-a7f5-90c9fb2b1729";
const CH_B: &str = "a2228687-0000-4000-8000-000000000000";
fn channels(ids: &[&str]) -> Selection {
Selection {
channels: ids.iter().map(|s| s.to_string()).collect(),
..Selection::default()
}
}
#[test]
fn empty_selection_is_rejected() {
let mut s = Selection::default();
@@ -194,9 +193,8 @@ mod tests {
#[test]
fn canonicalize_lowercases_sorts_and_dedupes() {
let mut s = Selection {
channels: vec![CH_B.into(), CH_A.to_uppercase(), CH_B.into()],
authors: vec![],
kinds: vec![40002, 9, 9],
..channels(&[CH_B, &CH_A.to_uppercase(), CH_B])
};
s.canonicalize().expect("valid");
assert_eq!(s.channels, vec![CH_A.to_string(), CH_B.to_string()]);
@@ -205,51 +203,74 @@ mod tests {
#[test]
fn malformed_ids_and_out_of_range_kinds_are_rejected() {
assert!(channels(&["not-a-uuid"]).canonicalize().is_err());
let mut s = Selection {
channels: vec!["not-a-uuid".into()],
authors: vec![],
kinds: vec![],
};
assert!(s.canonicalize().is_err());
let mut s = Selection {
channels: vec![],
authors: vec!["abc123".into()],
kinds: vec![],
..Selection::default()
};
assert!(s.canonicalize().is_err());
let mut s = Selection {
channels: vec![CH_A.into()],
authors: vec![],
kinds: vec![70_000],
..channels(&[CH_A])
};
assert!(s.canonicalize().is_err());
}
#[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![],
fn empty_window_is_rejected_but_open_ends_pass() {
let mut s = Selection {
since: Some(200),
until_exclusive: Some(200),
..channels(&[CH_A])
};
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"]));
assert!(s.canonicalize().is_err());
let mut s = Selection {
since: Some(100),
until_exclusive: Some(200),
..channels(&[CH_A])
};
assert!(s.canonicalize().is_ok());
assert!(s.is_frozen());
let mut s = Selection {
since: Some(100),
..channels(&[CH_A])
};
assert!(s.canonicalize().is_ok());
assert!(!s.is_frozen());
}
#[test]
fn author_only_selection_compiles_one_filter() {
fn live_selection_resolves_to_now_and_clamps_narrow() {
let s = channels(&[CH_A]);
assert_eq!(s.resolve_window(None, None, 1000), (0, 1001));
// The run-pins-to-priced-window clamp.
assert_eq!(s.resolve_window(Some(10), Some(900), 1000), (10, 900));
}
#[test]
fn frozen_selection_never_reads_outside_its_freeze() {
let s = Selection {
channels: vec![],
authors: vec!["p1".into()],
kinds: vec![9],
since: Some(100),
until_exclusive: Some(200),
..channels(&[CH_A])
};
let filters = s.compile_filters(0, 10, 5);
assert_eq!(filters.len(), 1);
assert!(filters[0].get("#h").is_none());
// No clamp: the freeze is the window, even as "now" moves on.
assert_eq!(s.resolve_window(None, None, 10_000), (100, 200));
// A wider clamp cannot widen; a narrower one narrows.
assert_eq!(s.resolve_window(Some(0), Some(10_000), 10_000), (100, 200));
assert_eq!(s.resolve_window(Some(150), Some(180), 10_000), (150, 180));
}
#[test]
fn selection_json_without_window_still_loads_as_live() {
// Specs saved before the when moved into the selection must keep
// loading — and must compare equal to a live selection, so existing
// chains stay cached.
let legacy = format!(r#"{{"channels":["{CH_A}"]}}"#);
let loaded: Selection = serde_json::from_str(&legacy).expect("deserialize");
assert_eq!(loaded, channels(&[CH_A]));
assert!(!loaded.is_frozen());
let out = serde_json::to_string(&loaded).expect("serialize");
assert!(!out.contains("until"), "open ends must not serialize");
}
}

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

@@ -14,26 +14,22 @@ use crate::selection::Selection;
/// Maximum fold-name length (also the `d` tag value).
pub const MAX_NAME_LEN: usize = 64;
/// Default schema label for new folds. The output is free-form — the label is
/// provenance on artifacts (which prompt regime produced them), not a
/// validated contract; chains created under `channel-digest@v1` keep their
/// historical label.
pub const FREEFORM_SCHEMA: &str = "freeform@v1";
/// Default fold *task* instructions when the caller supplies none.
pub const DEFAULT_INSTRUCTIONS: &str = "Maintain a running digest of the selection: rewrite the \
prior version in light of the new events. Keep it concise, concrete, and factual.";
/// One fold definition.
/// One fold definition: a factory for artifacts.
///
/// A fold is exactly name + selection + model + instructions — nothing else.
/// The selection's own window decides its lifecycle: frozen (pinned end) folds
/// run until their set is covered and are then done forever; live folds are
/// never done.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FoldSpec {
/// Fold name: 1–64 chars of `[a-z0-9-]`, no leading/trailing dash.
pub name: String,
/// What to read.
/// What to read (who × what × when; frozen or live).
pub selection: Selection,
/// Schema label recorded on artifacts as provenance ([`FREEFORM_SCHEMA`]
/// by default). Output is free-form; the label is not a validated shape.
pub schema: String,
/// Model alias passed to the runner and priced by [`crate::estimate`].
pub model: String,
/// Fold instructions (the prompt).
@@ -41,7 +37,7 @@ pub struct FoldSpec {
/// Free-form client-owned JSON, persisted verbatim with the spec.
///
/// The engine never reads it and it is deliberately absent from the
/// cached-run comparison ([`crate::plan_run`] compares model, schema,
/// cached-run comparison ([`crate::plan_run`] compares model,
/// prompt hash, and selection field-by-field) — so a client can stash its
/// strategy metadata here without invalidating chains.
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -65,11 +61,6 @@ impl FoldSpec {
self.name
)));
}
if self.schema.trim().is_empty() {
return Err(Error::InvalidSpec(
"schema label must be non-empty (it is recorded on artifacts as provenance)".into(),
));
}
if self.model.trim().is_empty() {
return Err(Error::InvalidSpec(
"model is required — estimates and runs must name the model".into(),
@@ -99,10 +90,8 @@ mod tests {
name: "team-digest".to_string(),
selection: Selection {
channels: vec!["59ca5528-71ea-4a53-a7f5-90c9fb2b1729".to_string()],
authors: vec![],
kinds: vec![],
..Selection::default()
},
schema: "channel-digest@v1".to_string(),
model: "haiku".to_string(),
instructions: "Maintain the digest.".to_string(),
meta: None,
@@ -124,14 +113,7 @@ mod tests {
}
#[test]
fn empty_fields_are_rejected_but_any_schema_label_passes() {
// Schema is a provenance label, not a validated shape.
let mut s = valid();
s.schema = "my-custom-regime@v3".to_string();
assert!(s.validate().is_ok());
let mut s = valid();
s.schema = " ".to_string();
assert!(s.validate().is_err());
fn empty_fields_are_rejected() {
let mut s = valid();
s.model = " ".to_string();
assert!(s.validate().is_err());
@@ -141,12 +123,22 @@ mod tests {
}
#[test]
fn spec_json_without_meta_still_loads() {
// Specs persisted before the meta field existed must keep loading.
fn legacy_spec_json_still_loads() {
// Specs persisted before `meta` existed — and before `schema` was
// removed — must keep loading (unknown fields are ignored).
let legacy = serde_json::to_string(&valid()).expect("serialize");
assert!(!legacy.contains("meta"), "None meta must not serialize");
let loaded: FoldSpec = serde_json::from_str(&legacy).expect("deserialize");
assert_eq!(loaded.meta, None);
let with_schema = r#"{
"name": "old",
"selection": { "channels": ["59ca5528-71ea-4a53-a7f5-90c9fb2b1729"] },
"schema": "channel-digest@v1",
"model": "haiku",
"instructions": "digest"
}"#;
let loaded: FoldSpec = serde_json::from_str(with_schema).expect("legacy schema field");
assert_eq!(loaded.name, "old");
}
#[test]

12
crates/buzz-accumulator/ui/src/api.ts
View File

@@ -4,9 +4,14 @@ export type Selection = {
channels: string[];
authors: string[];
kinds: number[];
/** Selection's own window (who × what × when). A pinned `until_exclusive`
* freezes the selection; open ends mean live ("and whatever comes next"). */
since?: number;
until_exclusive?: number;
};
/** Half-open run window; omitted bounds mean "everything up to now". */
/** Optional clamp sent with preview/preflight/run; it can only narrow the
* selection's own window (this pins a run to its priced preflight window). */
export type TimeWindow = { since?: number; until_exclusive?: number };
export type Status = {
@@ -71,7 +76,6 @@ export type EventsPage = {
export type FoldSpec = {
name: string;
selection: Selection;
schema: string;
model: string;
instructions: string;
meta?: unknown;
@@ -120,7 +124,6 @@ export type Artifact = {
selection: Selection;
channels: string[];
model: string;
schema: string;
prompt_sha256: string;
truncated: boolean;
created_at: number;
@@ -134,7 +137,6 @@ export type ArtifactSummary = {
coverage_until: number | null;
channels: string[];
model: string;
schema: string;
truncated: boolean;
};
@@ -157,6 +159,8 @@ function normalizeSelection(s: Partial<Selection> | undefined): Selection {
channels: s?.channels ?? [],
authors: s?.authors ?? [],
kinds: s?.kinds ?? [],
since: s?.since,
until_exclusive: s?.until_exclusive,
};
}