From 37412292352ab713ac792b6049d83555a87a17ef Mon Sep 17 00:00:00 2001 From: Fizz Date: Tue, 1 Sep 2026 15:22:36 -0700 Subject: [PATCH] =?UTF-8?q?accumulator:=20three-noun=20algebra=20=E2=80=94?= =?UTF-8?q?=20selection=20absorbs=20when,=20publish=20verb,=20schema=20del?= =?UTF-8?q?eted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Signed-off-by: Fizz --- Cargo.lock | 1 + crates/buzz-accumulator/Cargo.toml | 2 + crates/buzz-accumulator/README.md | 100 ++++++++ crates/buzz-accumulator/src/artifact.rs | 4 +- crates/buzz-accumulator/src/daemon/folds.rs | 104 +++++--- crates/buzz-accumulator/src/daemon/http.rs | 224 ++++++++++++++++-- crates/buzz-accumulator/src/daemon/mod.rs | 7 + crates/buzz-accumulator/src/daemon/publish.rs | 64 +++++ crates/buzz-accumulator/src/daemon/store.rs | 11 +- crates/buzz-accumulator/src/lib.rs | 19 +- crates/buzz-accumulator/src/run.rs | 7 +- crates/buzz-accumulator/src/selection.rs | 197 ++++++++------- crates/buzz-accumulator/src/spec.rs | 52 ++-- crates/buzz-accumulator/ui/src/api.ts | 12 +- 14 files changed, 604 insertions(+), 200 deletions(-) create mode 100644 crates/buzz-accumulator/README.md create mode 100644 crates/buzz-accumulator/src/daemon/publish.rs diff --git a/Cargo.lock b/Cargo.lock index b556406971..9edebe5ed7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -849,6 +849,7 @@ dependencies = [ "tower", "tracing", "tracing-subscriber", + "uuid", ] [[package]] diff --git a/crates/buzz-accumulator/Cargo.toml b/crates/buzz-accumulator/Cargo.toml index 0cf6203803..555b06d582 100644 --- a/crates/buzz-accumulator/Cargo.toml +++ b/crates/buzz-accumulator/Cargo.toml @@ -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" diff --git a/crates/buzz-accumulator/README.md b/crates/buzz-accumulator/README.md new file mode 100644 index 0000000000..f6b03905b1 --- /dev/null +++ b/crates/buzz-accumulator/README.md @@ -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:]` 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= 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`. diff --git a/crates/buzz-accumulator/src/artifact.rs b/crates/buzz-accumulator/src/artifact.rs index 01bb484bf4..8dcd52b4db 100644 --- a/crates/buzz-accumulator/src/artifact.rs +++ b/crates/buzz-accumulator/src/artifact.rs @@ -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, @@ -45,8 +45,6 @@ pub struct ArtifactPayload { pub channels: Vec, /// 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 diff --git a/crates/buzz-accumulator/src/daemon/folds.rs b/crates/buzz-accumulator/src/daemon/folds.rs index 83bc02425e..868e1bd51c 100644 --- a/crates/buzz-accumulator/src/daemon/folds.rs +++ b/crates/buzz-accumulator/src/daemon/folds.rs @@ -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, Option); + +/// 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, Plan), FoldError> { + clamp: WindowClamp, +) -> Result<(FoldSpec, Vec, 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 = chain .iter() @@ -157,20 +165,19 @@ async fn plan( let authors: BTreeSet = 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 { - 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, guard: &RunGuard, name: &str, - since: i64, - until_exclusive: i64, + clamp: WindowClamp, ) -> Result { 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(_))); diff --git a/crates/buzz-accumulator/src/daemon/http.rs b/crates/buzz-accumulator/src/daemon/http.rs index c425e11da1..efb516fabd 100644 --- a/crates/buzz-accumulator/src/daemon/http.rs +++ b/crates/buzz-accumulator/src/daemon/http.rs @@ -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, /// Per-fold run serialization (concurrent runs 409 instead of racing). pub runs: RunGuard, + /// Publishes artifacts back into channels as messages. + pub publisher: Arc, } /// 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) -> Result, 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, @@ -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, 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, 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, - /// Provenance label recorded on artifacts; defaults to `freeform@v1`. - schema: Option, /// Free-form client-owned JSON, stored verbatim and never read by the /// engine (see [`FoldSpec::meta`]). meta: Option, @@ -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>, ) -> Result, 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, body: Option>, ) -> Result, 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, + Path((name, version)): Path<(String, u32)>, + Json(body): Json, +) -> Result, 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>, + } + impl Publisher for FakePublisher { + fn publish<'a>( + &'a self, + channel: &'a str, + content: &'a str, + ) -> futures_util::future::BoxFuture<'a, Result> { + 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) -> 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 { + 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); + } } diff --git a/crates/buzz-accumulator/src/daemon/mod.rs b/crates/buzz-accumulator/src/daemon/mod.rs index 1f8164ffde..a2cea06640 100644 --- a/crates/buzz-accumulator/src/daemon/mod.rs +++ b/crates/buzz-accumulator/src/daemon/mod.rs @@ -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 diff --git a/crates/buzz-accumulator/src/daemon/publish.rs b/crates/buzz-accumulator/src/daemon/publish.rs new file mode 100644 index 0000000000..8905b747d6 --- /dev/null +++ b/crates/buzz-accumulator/src/daemon/publish.rs @@ -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>; +} + +/// 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, +} + +impl Publisher for RelayPublisher { + fn publish<'a>( + &'a self, + channel: &'a str, + content: &'a str, + ) -> BoxFuture<'a, Result> { + 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) + }) + } +} diff --git a/crates/buzz-accumulator/src/daemon/store.rs b/crates/buzz-accumulator/src/daemon/store.rs index d2e4e98692..893603f559 100644 --- a/crates/buzz-accumulator/src/daemon/store.rs +++ b/crates/buzz-accumulator/src/daemon/store.rs @@ -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, diff --git a/crates/buzz-accumulator/src/lib.rs b/crates/buzz-accumulator/src/lib.rs index 27df0abbb5..31b895a434 100644 --- a/crates/buzz-accumulator/src/lib.rs +++ b/crates/buzz-accumulator/src/lib.rs @@ -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. diff --git a/crates/buzz-accumulator/src/run.rs b/crates/buzz-accumulator/src/run.rs index d92c0ae107..957172146a 100644 --- a/crates/buzz-accumulator/src/run.rs +++ b/crates/buzz-accumulator/src/run.rs @@ -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, diff --git a/crates/buzz-accumulator/src/selection.rs b/crates/buzz-accumulator/src/selection.rs index e6b4fbf844..7758001c1d 100644 --- a/crates/buzz-accumulator/src/selection.rs +++ b/crates/buzz-accumulator/src/selection.rs @@ -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, /// 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, + /// Window start (unix seconds, inclusive). `None` = beginning of time. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub since: Option, + /// 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, } 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 { 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 { - let kinds = self.effective_kinds(); - let until = until_exclusive.saturating_sub(1); - let base = |extra: &mut serde_json::Map| { - extra.insert("kinds".into(), serde_json::json!(kinds)); - extra.insert("since".into(), serde_json::json!(since)); - extra.insert("until".into(), serde_json::json!(until)); - extra.insert("limit".into(), serde_json::json!(limit)); - if !self.authors.is_empty() { - extra.insert("authors".into(), serde_json::json!(self.authors)); - } - }; - if self.channels.is_empty() { - let mut m = serde_json::Map::new(); - base(&mut m); - return vec![serde_json::Value::Object(m)]; - } - self.channels - .iter() - .map(|ch| { - let mut m = serde_json::Map::new(); - base(&mut m); - m.insert("#h".into(), serde_json::json!([ch])); - serde_json::Value::Object(m) - }) - .collect() - } - - /// Human-readable one-line description. - pub fn describe(&self) -> String { - let mut parts = Vec::new(); - if !self.channels.is_empty() { - parts.push(format!("{} channel(s)", self.channels.len())); - } - if !self.authors.is_empty() { - parts.push(format!("{} author(s)", self.authors.len())); - } - parts.push(format!("kinds {:?}", self.effective_kinds())); - parts.join(" · ") + clamp_since: Option, + clamp_until_exclusive: Option, + 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"); } } diff --git a/crates/buzz-accumulator/src/spec.rs b/crates/buzz-accumulator/src/spec.rs index d3355f768a..f24c21c29a 100644 --- a/crates/buzz-accumulator/src/spec.rs +++ b/crates/buzz-accumulator/src/spec.rs @@ -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] diff --git a/crates/buzz-accumulator/ui/src/api.ts b/crates/buzz-accumulator/ui/src/api.ts index 228083280f..3bc8be4070 100644 --- a/crates/buzz-accumulator/ui/src/api.ts +++ b/crates/buzz-accumulator/ui/src/api.ts @@ -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 | undefined): Selection { channels: s?.channels ?? [], authors: s?.authors ?? [], kinds: s?.kinds ?? [], + since: s?.since, + until_exclusive: s?.until_exclusive, }; }