diff --git a/AGENTS.md b/AGENTS.md index 2d3939bbb36..136ce255bc4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -51,6 +51,7 @@ crates/ buzz-dev-mcp # Developer MCP server — shell + file-edit tools buzz-persona # Agent persona packs buzz-workflow # YAML-as-code workflow engine (evalexpr conditions) + buzz-budget # Sliding-window cost accounting for agent-to-agent exchanges # Clients + interop buzz-pair-relay # Ephemeral sidecar relay for NIP-AB device pairing buzz-pairing-cli # CLI for NIP-AB device pairing interop testing diff --git a/Cargo.lock b/Cargo.lock index 73ecb249d48..54991d5dbc1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -935,6 +935,19 @@ dependencies = [ "tower", ] +[[package]] +name = "buzz-budget" +version = "0.1.0" +dependencies = [ + "buzz-core", + "chrono", + "nostr", + "thiserror 2.0.18", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "buzz-cli" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index cc1dd0f9dff..bb7f74317e5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,7 @@ members = [ "crates/buzz-dev-mcp", "crates/buzz-voice", "crates/buzz-backend-kubernetes", + "crates/buzz-budget", "examples/countdown-bot", ] exclude = ["desktop/src-tauri"] diff --git a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py index 149a5295a7a..ed883a820aa 100644 --- a/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py +++ b/benchmarks/harbor-buzz-orchestra/src/harbor_buzz_orchestra/container_runtime.py @@ -157,9 +157,17 @@ async def run( ) # The task arrives exactly as it would in production Buzz: a # user prompt @mentioning the orchestrator. The harness never - # speaks as any agent. + # speaks as any agent. The orchestrator is mentioned by pubkey, + # not by name resolution: task text is untrusted payload, and any + # @-token inside it (e.g. Vim's `:%normal! @a`) would otherwise + # fail member resolution and kill the trial before the agent + # ever saw the task. An explicit --mention demotes unresolved + # @-tokens in the text to presentation-only. await self._send( - trial.user, trial, f"@{orchestrator.agent_id} {instruction}" + trial.user, + trial, + f"@{orchestrator.agent_id} {instruction}", + mention=orchestrator.nostr_pubkey, ) final_message = await asyncio.wait_for( self._wait_for_done(environment, orchestrator, trial, agents + infra), @@ -519,18 +527,24 @@ async def _verify_m1_output( ) async def _send( - self, credential: AgentCredential, trial: TrialHandle, content: str + self, + credential: AgentCredential, + trial: TrialHandle, + content: str, + *, + mention: str | None = None, ) -> None: - await self._buzz_json( - credential, - trial, + args = [ "messages", "send", "--channel", trial.channel_id, "--content", content, - ) + ] + if mention is not None: + args += ["--mention", mention] + await self._buzz_json(credential, trial, *args) async def _buzz_json( self, credential: AgentCredential, trial: TrialHandle, *args: str diff --git a/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py b/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py index ebf0eb4b5d2..5fc0e63e549 100644 --- a/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py +++ b/benchmarks/harbor-buzz-orchestra/tests/test_container_runtime.py @@ -372,6 +372,37 @@ async def test_m1_output_probe_matches_grader_and_is_condition_scoped( assert bool(probed) == (condition == "M1-hello-world") +async def test_send_mentions_by_pubkey_so_task_text_stays_inert( + tmp_path, monkeypatch +): + """Task text is untrusted payload: `:%normal! @a` in a task statement must + not be fed to member-name resolution (it would fail and kill the trial). + An explicit --mention pins delivery to the orchestrator's pubkey.""" + rt = runtime(tmp_path) + orch = credential("orch-1", "orchestrator", "orch-model") + trial = trial_handle((orch,)) + calls = [] + + async def buzz_json(credential, trial, *args): + calls.append(args) + return {} + + monkeypatch.setattr(rt, "_buzz_json", buzz_json) + + await rt._send( + trial.user, + trial, + "@orch-1 run `:%normal! @a` on the file", + mention=orch.nostr_pubkey, + ) + assert calls[-1][-2:] == ("--mention", "pubkey-orch-1") + + # Without an explicit mention the send is unchanged (name resolution). + await rt._send(trial.user, trial, "plain content") + assert "--mention" not in calls[-1] + assert calls[-1][-2:] == ("--content", "plain content") + + async def test_wait_for_done_requires_orchestrator_authorship(tmp_path, monkeypatch): rt = runtime(tmp_path, poll_seconds=0) orch = credential("orch-1", "orchestrator", "orch-model") diff --git a/crates/buzz-budget/Cargo.toml b/crates/buzz-budget/Cargo.toml new file mode 100644 index 00000000000..12ba6ef6e93 --- /dev/null +++ b/crates/buzz-budget/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "buzz-budget" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Sliding-window cost accounting for agent-to-agent exchanges" + +[dependencies] +buzz-core = { workspace = true } +chrono = { workspace = true } +nostr = { workspace = true } +thiserror = { workspace = true } +tracing = { workspace = true } +uuid = { workspace = true } + +[dev-dependencies] +tracing-subscriber = { workspace = true } diff --git a/crates/buzz-budget/examples/runaway.rs b/crates/buzz-budget/examples/runaway.rs new file mode 100644 index 00000000000..14b015a17be --- /dev/null +++ b/crates/buzz-budget/examples/runaway.rs @@ -0,0 +1,85 @@ +//! Drive the budget at its real runtime surface and show the sawtooth. +//! +//! Decision D4 of wayfinder ticket #7 justifies a self-healing window by the +//! signature it leaves in the logs — "a runaway stops within minutes and +//! restarts only to stop again, producing a **sawtooth in the logs**, a +//! diagnosable signature rather than a silent drain." +//! +//! That claim is only true if something is actually emitted. This example +//! exists so it can be *observed* rather than asserted: it runs a simulated +//! two-agent runaway alongside an owner working in another channel, with a real +//! `tracing` subscriber attached, and prints what an operator would see. +//! +//! Run with: +//! cargo run -p buzz-budget --example runaway + +use chrono::{DateTime, Duration, Utc}; +use uuid::Uuid; + +use buzz_budget::{Supervisor, DEFAULT_BUDGET_USD, DEFAULT_WINDOW_SECS}; + +fn main() { + tracing_subscriber::fmt() + .with_max_level(tracing::Level::DEBUG) + .with_target(false) + .without_time() + .init(); + + let runaway_channel = Uuid::from_u128(1); + let human_channel = Uuid::from_u128(2); + let t0: DateTime = + DateTime::from_timestamp(1_700_000_000, 0).expect("valid fixed timestamp"); + + let mut sup = Supervisor::new(DEFAULT_BUDGET_USD, DEFAULT_WINDOW_SECS); + + println!("--- two agents talking to each other, nobody watching ---"); + let mut tripped = 0usize; + for i in 0..30i64 { + let at = t0 + Duration::seconds(i * 2); + let (speaker, listener) = if i % 2 == 0 { + ("otto", "eva") + } else { + ("eva", "otto") + }; + sup.observe_message(runaway_channel, speaker, false, at); + let v = sup.on_turn_completed( + runaway_channel, + listener, + Some(0.60), + at + Duration::seconds(1), + ); + if v.is_exhausted() { + tripped += 1; + } + } + println!("=> exhausted verdicts in the first window: {tripped}"); + + println!(); + println!("--- the same agents, one hour later: the window has slid ---"); + let later = t0 + Duration::seconds(DEFAULT_WINDOW_SECS + 60); + sup.observe_message(runaway_channel, "otto", false, later); + let v = sup.on_turn_completed( + runaway_channel, + "eva", + Some(0.60), + later + Duration::seconds(1), + ); + println!("=> after the window slid: {v:?}"); + + println!(); + println!("--- meanwhile, the owner working hard in another channel ---"); + let mut blocked = 0usize; + for i in 0..40i64 { + let at = t0 + Duration::seconds(i * 2); + sup.observe_message(human_channel, "owner", true, at); + let v = sup.on_turn_completed(human_channel, "eva", Some(2.50), at + Duration::seconds(1)); + if v.is_exhausted() { + blocked += 1; + } + } + println!("=> owner turns blocked (must be 0): {blocked}"); + println!( + "=> owner-channel spend recorded (must be 0.00): {:.2}", + sup.spent(human_channel, "eva", "owner", t0 + Duration::seconds(200)) + ); +} diff --git a/crates/buzz-budget/src/ingest.rs b/crates/buzz-budget/src/ingest.rs new file mode 100644 index 00000000000..fdea0bced04 --- /dev/null +++ b/crates/buzz-budget/src/ingest.rs @@ -0,0 +1,350 @@ +//! Turning a `kind:44200` event on the wire into something the ledger can charge. +//! +//! This is the only part of the crate that knows about Nostr. Everything else +//! works on plain values, which is what keeps [`crate::Ledger`] and +//! [`crate::TriggerLog`] testable without keys or events. +//! +//! The metric's content is NIP-44 v2 ciphertext addressed **agent key → owner +//! pubkey** (`crates/buzz-core/src/agent_turn_metric.rs:1-5`), so only the owner +//! can read it. That is why the budget is enforced owner-side: the relay +//! structurally cannot see `costUsd`. + +use chrono::{DateTime, Utc}; +use nostr::{Event, Keys}; +use uuid::Uuid; + +use buzz_core::agent_turn_metric::decrypt_agent_turn_metric; +use buzz_core::kind::KIND_AGENT_TURN_METRIC; + +/// One completed turn, ready to be attributed and charged. +#[derive(Debug, Clone, PartialEq)] +pub struct TurnCharge { + /// Channel the turn served. Carried *inside* the encrypted payload, not as + /// an `h` tag, so it is only readable by the owner. + pub channel_id: Uuid, + /// The agent that took the turn — the event's author. + pub agent_pubkey: String, + /// Spend for this turn. `None` when the harness did not report it; see + /// ticket #7 OQ7.2, such a turn is charged zero and is effectively + /// invisible to the budget. + pub cost_usd: Option, + /// End-of-turn timestamp from the payload. + pub turn_end: DateTime, + /// `false` when the publisher could not observe the previous cumulative + /// baseline, making `cost_usd` unreliable. Surfaced rather than swallowed — + /// ticket #7 OQ7.1 asks how common this is in practice. + pub delta_reliable: bool, +} + +/// Why a `kind:44200` event could not be turned into a [`TurnCharge`]. +#[derive(Debug, thiserror::Error)] +pub enum IngestError { + /// The event is not `kind:44200` — expected when filtering a mixed stream. + #[error("event kind {0} is not the agent turn metric kind ({KIND_AGENT_TURN_METRIC})")] + WrongKind(u16), + /// Decryption or NIP-AM validation failed. Also the case when the metric + /// was addressed to a different owner. + #[error("could not decrypt or validate the turn metric: {0}")] + Payload(String), + /// The payload omitted `channelId`, so the turn cannot be scoped. + #[error("payload has no channelId")] + MissingChannelId, + /// `channelId` was present but is not a UUID. + #[error("payload channelId {0:?} is not a UUID")] + BadChannelId(String), + /// `timestamp` was present but is not RFC 3339. + #[error("payload timestamp {0:?} is not RFC 3339")] + BadTimestamp(String), + /// The self-reported turn time is too far from the signed `created_at` to + /// be trusted. See [`MAX_TIMESTAMP_SKEW_SECS`]. + #[error( + "payload timestamp {payload} is {skew_secs}s from the event's created_at {created_at} \ + (max {MAX_TIMESTAMP_SKEW_SECS}s)" + )] + TimestampSkew { + /// The self-reported end-of-turn time from inside the payload. + payload: String, + /// The event's `created_at`, which is covered by the signature and + /// checkable by the relay. + created_at: String, + /// Absolute difference, in seconds. + skew_secs: i64, + }, +} + +/// How far the payload's self-reported `timestamp` may sit from the event's +/// `created_at` before the metric is rejected. +/// +/// This exists because the payload timestamp is **attacker-controlled by the +/// very agent being budgeted**: it is chosen by the agent, sealed inside its own +/// NIP-44 ciphertext, and `AgentTurnMetricPayload::validate` checks only the +/// numerics. Without this check, one metric dated far in the future moves the +/// ledger's eviction cutoff forward and wipes the pair's whole window — letting +/// a runaway agent zero its own budget at will. +/// +/// `created_at` is not a perfect oracle (it is also chosen by the signer), but +/// it is covered by the signature and is the field a relay can and does police, +/// so tying the two together removes the free-form forgery. +pub const MAX_TIMESTAMP_SKEW_SECS: i64 = 300; + +/// Decrypt a `kind:44200` event with the owner's keys and extract what the +/// budget needs. +/// +/// Returns [`IngestError::WrongKind`] rather than panicking on unrelated +/// events, so a caller can pass a whole subscription stream through this. +pub fn charge_from_metric(owner_keys: &Keys, event: &Event) -> Result { + let kind = event.kind.as_u16(); + if u32::from(kind) != KIND_AGENT_TURN_METRIC { + return Err(IngestError::WrongKind(kind)); + } + + let payload = decrypt_agent_turn_metric(owner_keys, event) + .map_err(|e| IngestError::Payload(e.to_string()))?; + + let channel_raw = payload.channel_id.ok_or(IngestError::MissingChannelId)?; + let channel_id = Uuid::parse_str(&channel_raw) + .map_err(|_| IngestError::BadChannelId(channel_raw.clone()))?; + + let turn_end = DateTime::parse_from_rfc3339(&payload.timestamp) + .map_err(|_| IngestError::BadTimestamp(payload.timestamp.clone()))? + .with_timezone(&Utc); + + // The payload timestamp is chosen by the agent being budgeted. Tie it to + // the signed `created_at` so it cannot be used to shift the ledger window. + let created_at = DateTime::from_timestamp(event.created_at.as_secs() as i64, 0) + .ok_or_else(|| IngestError::BadTimestamp(event.created_at.to_string()))?; + let skew_secs = (turn_end - created_at).num_seconds().abs(); + if skew_secs > MAX_TIMESTAMP_SKEW_SECS { + return Err(IngestError::TimestampSkew { + payload: payload.timestamp.clone(), + created_at: created_at.to_rfc3339(), + skew_secs, + }); + } + + Ok(TurnCharge { + channel_id, + agent_pubkey: event.pubkey.to_hex(), + cost_usd: payload.turn.as_ref().and_then(|t| t.cost_usd), + turn_end, + delta_reliable: payload.delta_reliable, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::agent_turn_metric::{ + encrypt_agent_turn_metric, AgentTurnMetricPayload, TokenCounts, + }; + use nostr::{EventBuilder, Kind}; + + const CHANNEL: &str = "12345678-1234-1234-1234-123456789abc"; + + fn payload(cost: Option, channel: Option<&str>, ts: &str) -> AgentTurnMetricPayload { + AgentTurnMetricPayload { + harness: "goose".into(), + model: None, + channel_id: channel.map(|c| c.to_string()), + session_id: None, + turn_id: None, + turn_seq: None, + timestamp: ts.into(), + turn: Some(TokenCounts { + input_tokens: None, + output_tokens: None, + total_tokens: None, + cost_usd: cost, + cache_read_tokens: None, + cache_write_tokens: None, + }), + cumulative: None, + delta_reliable: true, + stop_reason: None, + } + } + + /// Build a real signed kind:44200 event the way an agent would, with + /// `created_at` consistent with the payload's own timestamp. + fn metric_event(agent: &Keys, owner: &Keys, p: &AgentTurnMetricPayload) -> Event { + let created = DateTime::parse_from_rfc3339(&p.timestamp) + .map(|t| t.timestamp() as u64) + .unwrap_or(1_700_000_000); + metric_event_at(agent, owner, p, created) + } + + /// Same, but with `created_at` chosen independently — used to exercise the + /// skew check. + fn metric_event_at( + agent: &Keys, + owner: &Keys, + p: &AgentTurnMetricPayload, + created_at: u64, + ) -> Event { + let content = encrypt_agent_turn_metric(agent, &owner.public_key(), p) + .expect("payload should encrypt"); + EventBuilder::new(Kind::Custom(KIND_AGENT_TURN_METRIC as u16), content) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(agent) + .expect("event should sign") + } + + #[test] + fn a_real_metric_event_round_trips_into_a_charge() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let p = payload(Some(0.42), Some(CHANNEL), "2026-08-07T12:00:00Z"); + let ev = metric_event(&agent, &owner, &p); + + let charge = charge_from_metric(&owner, &ev).expect("should ingest"); + assert_eq!(charge.channel_id, Uuid::parse_str(CHANNEL).unwrap()); + assert_eq!(charge.agent_pubkey, agent.public_key().to_hex()); + assert_eq!(charge.cost_usd, Some(0.42)); + assert_eq!(charge.turn_end.to_rfc3339(), "2026-08-07T12:00:00+00:00"); + assert!(charge.delta_reliable); + } + + /// A subscription stream carries other kinds; those must be rejected + /// cleanly rather than panicking. + #[test] + fn an_unrelated_kind_is_rejected_without_decrypting() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let ev = EventBuilder::new(Kind::Custom(9), "hello") + .sign_with_keys(&agent) + .expect("event should sign"); + assert!(matches!( + charge_from_metric(&owner, &ev), + Err(IngestError::WrongKind(9)) + )); + } + + /// The metric is addressed to one owner; anyone else must fail closed. + #[test] + fn a_different_owner_key_cannot_read_the_metric() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let stranger = Keys::generate(); + let ev = metric_event( + &agent, + &owner, + &payload(Some(1.0), Some(CHANNEL), "2026-08-07T12:00:00Z"), + ); + assert!(matches!( + charge_from_metric(&stranger, &ev), + Err(IngestError::Payload(_)) + )); + } + + #[test] + fn a_missing_channel_id_is_an_error_not_a_silent_default() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let ev = metric_event( + &agent, + &owner, + &payload(Some(1.0), None, "2026-08-07T12:00:00Z"), + ); + assert!(matches!( + charge_from_metric(&owner, &ev), + Err(IngestError::MissingChannelId) + )); + } + + #[test] + fn a_non_uuid_channel_id_is_rejected() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let ev = metric_event( + &agent, + &owner, + &payload(Some(1.0), Some("not-a-uuid"), "2026-08-07T12:00:00Z"), + ); + assert!(matches!( + charge_from_metric(&owner, &ev), + Err(IngestError::BadChannelId(_)) + )); + } + + #[test] + fn a_bad_timestamp_is_rejected() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let ev = metric_event( + &agent, + &owner, + &payload(Some(1.0), Some(CHANNEL), "yesterday"), + ); + assert!(matches!( + charge_from_metric(&owner, &ev), + Err(IngestError::BadTimestamp(_)) + )); + } + + /// Security regression: the payload timestamp is chosen by the agent being + /// budgeted. A far-future one used to move the ledger's eviction cutoff + /// forward and wipe the pair's whole window, letting a runaway zero its own + /// budget. It must be rejected against the signed `created_at`. + #[test] + fn a_far_future_payload_timestamp_is_rejected() { + let agent = Keys::generate(); + let owner = Keys::generate(); + // created_at says 2023; the payload claims 2099. + let ev = metric_event_at( + &agent, + &owner, + &payload(Some(1.0), Some(CHANNEL), "2099-01-01T00:00:00Z"), + 1_700_000_000, + ); + assert!(matches!( + charge_from_metric(&owner, &ev), + Err(IngestError::TimestampSkew { .. }) + )); + } + + /// A far-past timestamp is equally forged, and equally rejected. + #[test] + fn a_far_past_payload_timestamp_is_rejected() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let ev = metric_event_at( + &agent, + &owner, + &payload(Some(1.0), Some(CHANNEL), "1999-01-01T00:00:00Z"), + 1_700_000_000, + ); + assert!(matches!( + charge_from_metric(&owner, &ev), + Err(IngestError::TimestampSkew { .. }) + )); + } + + /// Ordinary clock jitter between the harness and the signer must not + /// reject a legitimate metric. + #[test] + fn small_clock_skew_is_tolerated() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let p = payload(Some(0.5), Some(CHANNEL), "2023-11-14T22:13:20Z"); + let created = DateTime::parse_from_rfc3339(&p.timestamp) + .unwrap() + .timestamp() as u64; + let ev = metric_event_at(&agent, &owner, &p, created + 60); + assert!(charge_from_metric(&owner, &ev).is_ok()); + } + + /// OQ7.2 — a harness that reports no cost is ingestable, and invisible to + /// the budget. Asserted so the hole cannot close silently unnoticed. + #[test] + fn a_metric_with_no_cost_ingests_as_none() { + let agent = Keys::generate(); + let owner = Keys::generate(); + let ev = metric_event( + &agent, + &owner, + &payload(None, Some(CHANNEL), "2026-08-07T12:00:00Z"), + ); + let charge = charge_from_metric(&owner, &ev).expect("should ingest"); + assert_eq!(charge.cost_usd, None); + } +} diff --git a/crates/buzz-budget/src/lib.rs b/crates/buzz-budget/src/lib.rs new file mode 100644 index 00000000000..c1490a3e744 --- /dev/null +++ b/crates/buzz-budget/src/lib.rs @@ -0,0 +1,389 @@ +//! Sliding-window cost accounting for agent-to-agent exchanges. +//! +//! Implements the accounting half of the loop-protection policy decided in +//! wayfinder ticket #7 (`mfethe1/agent-mesh`): +//! +//! - **D2** — budget measured `cost_usd`, not message count. Buzz already emits +//! `cost_usd` per completed turn on kind 44200 (NIP-AM). +//! - **D3** — $5 per (channel, agent-pair) per rolling hour. +//! - **D4** — self-healing: the window slides, no human reset. +//! - **D5** — only agent-triggered turns count. Human-triggered turns are +//! unbudgeted, because the human is present and can stop it themselves. +//! +//! **This crate deliberately does not enforce.** [`Ledger::record`] returns a +//! [`Verdict`]; acting on `Verdict::Exhausted` is the caller's job. That seam +//! exists because the enforcement mechanism is still an open question — Buzz's +//! owner commands (`!shutdown`, `!cancel`, `!rotate`) cannot mute a single peer, +//! so enforcement needs either a new owner command or a supervisor with +//! process-level control. Keeping accounting pure means that decision can land +//! later without touching this code. +//! +//! Why cost rather than message count: a productive exchange of 200 short +//! messages costs pennies and passes; 30 huge-context turns burn the budget and +//! stop. Message count would penalise the first and permit the second. + +#![deny(unsafe_code)] +#![warn(missing_docs)] + +use std::collections::HashMap; + +use chrono::{DateTime, Duration, Utc}; +use uuid::Uuid; + +pub mod ingest; +pub mod origin; +pub mod supervisor; +pub use ingest::{charge_from_metric, IngestError, TurnCharge}; +pub use origin::TriggerLog; +pub use supervisor::Supervisor; + +/// Default budget from decision D3. +pub const DEFAULT_BUDGET_USD: f64 = 5.0; + +/// Default rolling window from decision D3. +pub const DEFAULT_WINDOW_SECS: i64 = 3600; + +/// What caused an agent to take a turn. +/// +/// Per **D5** only [`TurnOrigin::Agent`] turns consume budget. A turn the human +/// asked for is never charged: the human is watching, and cutting them off is +/// the failure this policy exists to avoid. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TurnOrigin { + /// Triggered by the owner (a human). Never charged. + Human, + /// Triggered by another agent, identified by its pubkey hex. + Agent(String), +} + +/// The unordered pair of agents an exchange runs between, scoped to a channel. +/// +/// Unordered because a runaway is a property of the *pair*, not of a direction: +/// A→B and B→A are the same exchange and must share one budget. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct PairKey { + /// Channel the exchange runs in. Budgets never cross channels. + pub channel_id: Uuid, + lo: String, + hi: String, +} + +impl PairKey { + /// Build a key from two agent pubkeys in either order. + pub fn new(channel_id: Uuid, a: &str, b: &str) -> Self { + let (lo, hi) = if a <= b { + (a.to_string(), b.to_string()) + } else { + (b.to_string(), a.to_string()) + }; + Self { channel_id, lo, hi } + } +} + +/// The result of recording a turn. +#[derive(Debug, Clone, PartialEq)] +pub enum Verdict { + /// Within budget. `spent_usd` is the pair's total across the current + /// window, inclusive of the turn just recorded. + Allow { + /// The pair's spend across the current window. + spent_usd: f64, + }, + /// The turn was not charged, because it had no agent-pair to charge: a + /// human-triggered turn (**D5**), or one with no observable trigger. + /// + /// Distinct from `Allow` so a caller logging spend cannot mistake "not + /// budgeted" for "budgeted, and at zero" — the pair may be holding $4.90. + Unbudgeted, + /// Budget exceeded for this pair in this channel. The caller decides what + /// to do about it; this crate does not act. + Exhausted { + /// The pair's spend across the current window. + spent_usd: f64, + /// The budget it exceeded. + budget_usd: f64, + }, +} + +impl Verdict { + /// True when the exchange should be stopped. + pub fn is_exhausted(&self) -> bool { + matches!(self, Verdict::Exhausted { .. }) + } +} + +#[derive(Debug, Clone, Copy)] +struct Charge { + at: DateTime, + cost_usd: f64, +} + +/// Sliding-window cost ledger. +/// +/// Holds one window of charges per [`PairKey`]. Charges older than the window +/// are dropped whenever that pair is touched, which is what makes the budget +/// self-healing (**D4**) with no reset step. +#[derive(Debug)] +pub struct Ledger { + budget_usd: f64, + window: Duration, + charges: HashMap>, +} + +impl Default for Ledger { + fn default() -> Self { + Self::new(DEFAULT_BUDGET_USD, DEFAULT_WINDOW_SECS) + } +} + +impl Ledger { + /// Create a ledger. `window_secs` must be positive; non-positive values are + /// clamped to 1 second so a misconfiguration cannot disable the window and + /// silently accumulate forever. + pub fn new(budget_usd: f64, window_secs: i64) -> Self { + Self { + budget_usd, + window: Duration::seconds(window_secs.max(1)), + charges: HashMap::new(), + } + } + + /// Record one completed turn and return whether the exchange may continue. + /// + /// `cost_usd` comes from the kind 44200 turn metric. A `None` cost — a + /// harness that does not report spend — is charged as zero and logged by the + /// caller's concern, not silently treated as over budget; see the crate docs + /// and ticket #7 OQ7.2. + pub fn record( + &mut self, + channel_id: Uuid, + agent_pubkey: &str, + origin: &TurnOrigin, + cost_usd: Option, + at: DateTime, + ) -> Verdict { + // D5: human-triggered turns are never charged and never blocked. + let peer = match origin { + TurnOrigin::Human => return Verdict::Unbudgeted, + TurnOrigin::Agent(peer) => peer, + }; + + let key = PairKey::new(channel_id, agent_pubkey, peer); + let cutoff = at - self.window; + let entry = self.charges.entry(key).or_default(); + entry.retain(|c| c.at > cutoff); + entry.push(Charge { + at, + cost_usd: cost_usd.unwrap_or(0.0).max(0.0), + }); + + let spent: f64 = entry.iter().map(|c| c.cost_usd).sum(); + if spent > self.budget_usd { + Verdict::Exhausted { + spent_usd: spent, + budget_usd: self.budget_usd, + } + } else { + Verdict::Allow { spent_usd: spent } + } + } + + /// Spend for a pair across the current window, without recording anything. + pub fn spent(&self, channel_id: Uuid, a: &str, b: &str, now: DateTime) -> f64 { + let key = PairKey::new(channel_id, a, b); + let cutoff = now - self.window; + self.charges + .get(&key) + .map(|cs| { + cs.iter() + .filter(|c| c.at > cutoff) + .map(|c| c.cost_usd) + .sum() + }) + .unwrap_or(0.0) + } + + /// Number of pairs currently holding charges. Charges are only evicted when + /// their pair is touched, so this counts pairs seen, not pairs active. + pub fn tracked_pairs(&self) -> usize { + self.charges.len() + } + + /// Drop every pair whose charges have all aged out. Callers running a + /// long-lived supervisor should call this periodically; without it, a pair + /// that stops talking retains an empty-but-allocated entry. + pub fn evict_expired(&mut self, now: DateTime) { + let cutoff = now - self.window; + self.charges.retain(|_, cs| { + cs.retain(|c| c.at > cutoff); + !cs.is_empty() + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ch() -> Uuid { + Uuid::nil() + } + + fn t0() -> DateTime { + DateTime::from_timestamp(1_700_000_000, 0).expect("valid timestamp") + } + + const A: &str = "aaaa"; + const B: &str = "bbbb"; + + /// The runaway this whole policy exists to stop: two agents exchanging + /// turns with nobody watching. Written first, per the map's Notes. + #[test] + fn runaway_agent_exchange_is_stopped_once_the_budget_is_spent() { + let mut led = Ledger::new(5.0, 3600); + let origin = TurnOrigin::Agent(B.into()); + + // Ten turns at $0.40 = $4.00. Still under. + for i in 0..10 { + let v = led.record(ch(), A, &origin, Some(0.40), t0() + Duration::seconds(i)); + assert!(!v.is_exhausted(), "turn {i} should be allowed"); + } + + // Three more crosses $5.00. + let mut stopped = false; + for i in 10..14 { + if led + .record(ch(), A, &origin, Some(0.40), t0() + Duration::seconds(i)) + .is_exhausted() + { + stopped = true; + break; + } + } + assert!( + stopped, + "runaway must be stopped once the budget is exceeded" + ); + } + + /// D5 — the case that would make this policy worse than nothing. + #[test] + fn human_triggered_turns_are_never_charged_or_blocked() { + let mut led = Ledger::new(5.0, 3600); + for i in 0..100 { + let v = led.record( + ch(), + A, + &TurnOrigin::Human, + Some(1.0), + t0() + Duration::seconds(i), + ); + assert!(!v.is_exhausted(), "human turn {i} must never be blocked"); + } + assert_eq!(led.spent(ch(), A, B, t0() + Duration::seconds(100)), 0.0); + } + + /// D4 — self-healing, no human reset. + #[test] + fn budget_recovers_once_charges_slide_out_of_the_window() { + let mut led = Ledger::new(5.0, 3600); + let origin = TurnOrigin::Agent(B.into()); + + let v = led.record(ch(), A, &origin, Some(6.0), t0()); + assert!(v.is_exhausted(), "one $6 turn exceeds a $5 budget"); + + // Two hours later the charge has aged out. + let later = t0() + Duration::seconds(7200); + assert_eq!(led.spent(ch(), A, B, later), 0.0); + let v = led.record(ch(), A, &origin, Some(0.10), later); + assert!(!v.is_exhausted(), "budget must self-heal without a reset"); + } + + /// A runaway is a property of the pair, not a direction. + #[test] + fn the_pair_budget_is_shared_regardless_of_direction() { + let mut led = Ledger::new(5.0, 3600); + + led.record(ch(), A, &TurnOrigin::Agent(B.into()), Some(3.0), t0()); + // B replying to A must draw on the same budget, not a fresh one. + let v = led.record( + ch(), + B, + &TurnOrigin::Agent(A.into()), + Some(3.0), + t0() + Duration::seconds(1), + ); + assert!( + v.is_exhausted(), + "A→B and B→A must share one budget; got {v:?}" + ); + } + + #[test] + fn budgets_are_scoped_per_channel() { + let mut led = Ledger::new(5.0, 3600); + let other = Uuid::from_u128(1); + let origin = TurnOrigin::Agent(B.into()); + + assert!(led.record(ch(), A, &origin, Some(6.0), t0()).is_exhausted()); + // The same pair in a different channel is unaffected. + assert!(!led + .record(other, A, &origin, Some(0.5), t0()) + .is_exhausted()); + } + + /// OQ7.2 — a harness that reports no cost must not crash or be charged + /// arbitrarily. It is charged zero, which is why #7 flags per-harness + /// coverage as something to establish before relying on this. + #[test] + fn a_turn_with_no_reported_cost_is_charged_zero() { + let mut led = Ledger::new(5.0, 3600); + let origin = TurnOrigin::Agent(B.into()); + for i in 0..50 { + assert!(!led + .record(ch(), A, &origin, None, t0() + Duration::seconds(i)) + .is_exhausted()); + } + assert_eq!(led.spent(ch(), A, B, t0() + Duration::seconds(50)), 0.0); + } + + #[test] + fn negative_costs_cannot_refund_the_budget() { + let mut led = Ledger::new(5.0, 3600); + let origin = TurnOrigin::Agent(B.into()); + led.record(ch(), A, &origin, Some(4.9), t0()); + led.record(ch(), A, &origin, Some(-100.0), t0() + Duration::seconds(1)); + let v = led.record(ch(), A, &origin, Some(0.2), t0() + Duration::seconds(2)); + assert!( + v.is_exhausted(), + "a negative cost must not buy back budget; got {v:?}" + ); + } + + #[test] + fn pair_key_is_order_independent() { + assert_eq!(PairKey::new(ch(), A, B), PairKey::new(ch(), B, A)); + // Same key from either argument order, so both directions of an + // exchange hash to one budget. + use std::collections::HashSet; + let mut set = HashSet::new(); + set.insert(PairKey::new(ch(), A, B)); + set.insert(PairKey::new(ch(), B, A)); + assert_eq!(set.len(), 1); + } + + #[test] + fn evict_expired_drops_pairs_that_have_gone_quiet() { + let mut led = Ledger::new(5.0, 3600); + led.record(ch(), A, &TurnOrigin::Agent(B.into()), Some(1.0), t0()); + assert_eq!(led.tracked_pairs(), 1); + led.evict_expired(t0() + Duration::seconds(7200)); + assert_eq!(led.tracked_pairs(), 0); + } + + #[test] + fn a_non_positive_window_is_clamped_rather_than_disabling_the_budget() { + let led = Ledger::new(5.0, 0); + assert_eq!(led.window, Duration::seconds(1)); + } +} diff --git a/crates/buzz-budget/src/origin.rs b/crates/buzz-budget/src/origin.rs new file mode 100644 index 00000000000..1059d76d38b --- /dev/null +++ b/crates/buzz-budget/src/origin.rs @@ -0,0 +1,341 @@ +//! Attributing a completed turn to whatever triggered it. +//! +//! Decision **D5** of wayfinder ticket #7 says only agent-triggered turns +//! consume budget. The kind 44200 turn metric does not carry that information: +//! [`AgentTurnMetricPayload`] has `harness`, `model`, `channel_id`, +//! `session_id`, `turn_id`, `turn_seq`, `timestamp`, `turn`, `cumulative`, +//! `delta_reliable` and `stop_reason` — and no field naming the author that +//! caused the turn. +//! +//! This module recovers it without changing the wire format, using a property +//! the harness already guarantees: **turns are serialised per channel.** +//! `buzz-acp`'s queue keeps one in-flight prompt per channel and drains *all* +//! pending events for that channel into a single batch +//! (`crates/buzz-acp/src/queue.rs:1-7`). So the messages observed in a channel +//! since that channel's previous turn are exactly the batch that triggered the +//! next one. +//! +//! # Attribution rule +//! +//! Given the un-consumed messages in a channel up to the turn's end time, +//! ignoring the agent's own: +//! +//! - **any message from the owner → [`TurnOrigin::Human`]** (unbudgeted). +//! A batch mixing human and agent messages counts as human because D5 exists +//! to protect the case where a person is present and watching. Charging that +//! turn risks muting an agent mid-conversation with its owner, which is the +//! failure the policy is meant to avoid. Erring the other way only means a +//! runaway costs one extra turn before it trips. +//! - **otherwise, the most recent agent message → [`TurnOrigin::Agent`]**. +//! - **nothing observed → `None`.** Heartbeat turns and turns whose trigger was +//! missed are *not charged*. See the caveat below. +//! +//! # Caveat — the boundary is end-of-turn, not start-of-turn +//! +//! The metric's `timestamp` is *end-of-turn* +//! (`crates/buzz-core/src/agent_turn_metric.rs:111`) and the payload carries no +//! start time. So a message that arrives **while a turn is running** falls +//! inside that turn's window, even though it actually triggered the *next* one. +//! +//! The consequence is concrete: one owner message landing mid-turn attributes +//! that turn to `Human` (free), and the turn it really triggered then finds +//! nothing new and returns `None` (also free). **One human message can free two +//! turns.** Per-agent cursors do not fix this — the boundary is wrong, not the +//! bookkeeping. +//! +//! It errs toward under-charging, which is the same direction as every other +//! approximation here, so it is safe rather than merely tolerable. The durable +//! fix is a turn-start (or trigger) field on NIP-AM — a second argument for +//! ticket #7 OQ7.5. +//! +//! # Caveat — this is best-effort, and it fails open +//! +//! A supervisor that starts mid-conversation, drops a relay subscription, or +//! cannot read a private channel will observe no trigger and return `None`, +//! which is not charged. **An observation gap therefore disables the budget +//! rather than tripping it.** That is the right default for an accident-shaped +//! threat model, but it means coverage of the message stream is a correctness +//! requirement, not a nice-to-have. +//! +//! The durable fix is a `trigger` field on NIP-AM — the spec already says +//! *"Consumers MUST ignore unknown fields (forward compatibility)"* +//! (`crates/buzz-core/src/agent_turn_metric.rs:86`), so adding one is +//! backward-compatible. That needs a `buzz-acp` change; see ticket #7 OQ7.5. + +use std::collections::HashMap; + +use chrono::{DateTime, Duration, Utc}; +use uuid::Uuid; + +use crate::TurnOrigin; + +/// One observed inbound message in a channel. +#[derive(Debug, Clone, PartialEq, Eq)] +struct Observed { + at: DateTime, + author: String, + is_owner: bool, +} + +/// Records inbound messages per channel so a completed turn can be attributed. +/// +/// Feed it every kind:9 the supervisor sees, then ask [`Self::origin_for_turn`] +/// when a kind 44200 metric arrives. +#[derive(Debug, Default)] +pub struct TriggerLog { + seen: HashMap>, + /// Per-(channel, agent) high-water mark of what that agent has already + /// been attributed. Exclusive lower bound on the next lookup. + cursors: HashMap<(Uuid, String), DateTime>, +} + +impl TriggerLog { + /// Create an empty log. + pub fn new() -> Self { + Self::default() + } + + /// Record an inbound message. `is_owner` distinguishes the human from a + /// sibling agent — `buzz-acp` already computes exactly this distinction in + /// `is_owner_or_sibling` (`crates/buzz-acp/src/lib.rs:192-216`). + pub fn observe(&mut self, channel_id: Uuid, author: &str, is_owner: bool, at: DateTime) { + self.seen.entry(channel_id).or_default().push(Observed { + at, + author: author.to_string(), + is_owner, + }); + } + + /// Attribute a turn that ended at `turn_end`. + /// + /// Consumption is tracked **per (channel, agent)**, not per channel. Each + /// agent carries its own cursor, so two agents sharing a channel each see + /// the batch that triggered them. Draining per channel — as this did + /// originally — meant whichever metric the supervisor happened to process + /// first ate the other agent's trigger, leaving the second turn attributed + /// to `None` and therefore uncharged. Relay reordering alone was enough to + /// silently disable the budget. + /// + /// Returns `None` when nothing was observed — a heartbeat turn, or an + /// observation gap. `None` is not charged. + pub fn origin_for_turn( + &mut self, + channel_id: Uuid, + agent_pubkey: &str, + turn_end: DateTime, + ) -> Option { + let entries = self.seen.get(&channel_id)?; + let cursor = self + .cursors + .get(&(channel_id, agent_pubkey.to_string())) + .copied(); + + // Candidates are the messages this agent has not already accounted for, + // up to the end of the turn. `cursor` is exclusive so a message cannot + // trigger the same agent twice. + let triggers: Vec<&Observed> = entries + .iter() + .filter(|o| o.at <= turn_end) + .filter(|o| cursor.is_none_or(|c| o.at > c)) + .filter(|o| o.author != agent_pubkey) + .collect(); + + // Advance this agent's cursor whether or not anything was found, so a + // turn that saw nothing does not re-examine the same window forever. + self.cursors + .insert((channel_id, agent_pubkey.to_string()), turn_end); + + if triggers.is_empty() { + return None; + } + // A mixed batch counts as human — see the module docs for why. + if triggers.iter().any(|o| o.is_owner) { + return Some(TurnOrigin::Human); + } + triggers + .iter() + .max_by_key(|o| o.at) + .map(|o| TurnOrigin::Agent(o.author.clone())) + } + + /// Drop observations older than `max_age`, so a channel that goes quiet + /// with un-consumed messages cannot grow without bound. + pub fn prune(&mut self, now: DateTime, max_age: Duration) { + let cutoff = now - max_age; + self.seen.retain(|_, v| { + v.retain(|o| o.at > cutoff); + !v.is_empty() + }); + // Cursors for channels with nothing left to attribute are dead weight. + self.cursors + .retain(|(ch, _), at| *at > cutoff && self.seen.contains_key(ch)); + } + + /// Number of channels currently holding un-consumed observations. + pub fn tracked_channels(&self) -> usize { + self.seen.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ch() -> Uuid { + Uuid::nil() + } + fn t0() -> DateTime { + DateTime::from_timestamp(1_700_000_000, 0).expect("valid timestamp") + } + + const AGENT: &str = "agent"; + const PEER: &str = "peer"; + const OWNER: &str = "owner"; + + #[test] + fn a_turn_triggered_by_another_agent_is_attributed_to_that_agent() { + let mut log = TriggerLog::new(); + log.observe(ch(), PEER, false, t0()); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)), + Some(TurnOrigin::Agent(PEER.into())) + ); + } + + #[test] + fn a_turn_triggered_by_the_owner_is_human() { + let mut log = TriggerLog::new(); + log.observe(ch(), OWNER, true, t0()); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)), + Some(TurnOrigin::Human) + ); + } + + /// The safety-critical case: a batch containing the human must not be + /// charged, or an agent can be muted mid-conversation with its owner. + #[test] + fn a_mixed_batch_counts_as_human() { + let mut log = TriggerLog::new(); + log.observe(ch(), PEER, false, t0()); + log.observe(ch(), OWNER, true, t0() + Duration::seconds(1)); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)), + Some(TurnOrigin::Human) + ); + } + + #[test] + fn the_agents_own_messages_do_not_trigger_it() { + let mut log = TriggerLog::new(); + log.observe(ch(), AGENT, false, t0()); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)), + None + ); + } + + #[test] + fn nothing_observed_is_unattributed_and_therefore_uncharged() { + let mut log = TriggerLog::new(); + assert_eq!(log.origin_for_turn(ch(), AGENT, t0()), None); + } + + /// Per-channel serialisation is what makes this correlator sound; a turn + /// must not consume messages that arrived after it ended. + #[test] + fn messages_after_the_turn_end_belong_to_the_next_turn() { + let mut log = TriggerLog::new(); + log.observe(ch(), PEER, false, t0()); + log.observe(ch(), OWNER, true, t0() + Duration::seconds(10)); + + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)), + Some(TurnOrigin::Agent(PEER.into())) + ); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(20)), + Some(TurnOrigin::Human) + ); + } + + #[test] + fn a_batch_is_consumed_so_it_cannot_be_counted_twice() { + let mut log = TriggerLog::new(); + log.observe(ch(), PEER, false, t0()); + assert!(log + .origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)) + .is_some()); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(6)), + None, + "the same message must not trigger two turns" + ); + } + + /// Regression: consumption used to be per-channel, so with two agents in + /// one channel whichever metric arrived first ate the other's trigger and + /// the second turn went uncharged. Relay reordering alone disabled the + /// budget. + #[test] + fn two_agents_in_one_channel_each_see_their_own_trigger() { + let mut log = TriggerLog::new(); + // Iris mentions both agents in one message. + log.observe(ch(), "iris", false, t0()); + + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)), + Some(TurnOrigin::Agent("iris".into())) + ); + assert_eq!( + log.origin_for_turn(ch(), "otto", t0() + Duration::seconds(6)), + Some(TurnOrigin::Agent("iris".into())), + "the second agent must still see the trigger that woke it" + ); + } + + /// Each agent's cursor advances independently. + #[test] + fn one_agents_cursor_does_not_advance_another() { + let mut log = TriggerLog::new(); + log.observe(ch(), PEER, false, t0()); + assert!(log + .origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)) + .is_some()); + // AGENT is now caught up, but a later message must reach it again. + log.observe(ch(), PEER, false, t0() + Duration::seconds(10)); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(15)), + Some(TurnOrigin::Agent(PEER.into())) + ); + } + + #[test] + fn channels_are_independent() { + let mut log = TriggerLog::new(); + let other = Uuid::from_u128(1); + log.observe(ch(), PEER, false, t0()); + assert_eq!(log.origin_for_turn(other, AGENT, t0()), None); + assert!(log.origin_for_turn(ch(), AGENT, t0()).is_some()); + } + + #[test] + fn prune_drops_stale_unconsumed_observations() { + let mut log = TriggerLog::new(); + log.observe(ch(), PEER, false, t0()); + assert_eq!(log.tracked_channels(), 1); + log.prune(t0() + Duration::seconds(7200), Duration::seconds(3600)); + assert_eq!(log.tracked_channels(), 0); + } + + #[test] + fn the_most_recent_agent_wins_when_several_peers_spoke() { + let mut log = TriggerLog::new(); + log.observe(ch(), "early", false, t0()); + log.observe(ch(), "late", false, t0() + Duration::seconds(2)); + assert_eq!( + log.origin_for_turn(ch(), AGENT, t0() + Duration::seconds(5)), + Some(TurnOrigin::Agent("late".into())) + ); + } +} diff --git a/crates/buzz-budget/src/supervisor.rs b/crates/buzz-budget/src/supervisor.rs new file mode 100644 index 00000000000..03a2649ce47 --- /dev/null +++ b/crates/buzz-budget/src/supervisor.rs @@ -0,0 +1,337 @@ +//! Composes attribution and accounting into the API a caller actually wants. +//! +//! [`Ledger`] knows how to charge a pair. [`TriggerLog`] knows who triggered a +//! turn. Neither is useful alone: a caller holding both has to remember to +//! attribute before charging, and to prune both on the same schedule. This type +//! owns that sequencing so a caller cannot get it wrong. +//! +//! ```no_run +//! use buzz_budget::{Supervisor, DEFAULT_BUDGET_USD, DEFAULT_WINDOW_SECS}; +//! # use chrono::Utc; use uuid::Uuid; +//! let mut sup = Supervisor::new(DEFAULT_BUDGET_USD, DEFAULT_WINDOW_SECS); +//! +//! // Feed every kind:9 the supervisor sees. +//! sup.observe_message(Uuid::nil(), "peer-pubkey", false, Utc::now()); +//! +//! // Feed every kind:44200 turn metric. +//! let verdict = sup.on_turn_completed(Uuid::nil(), "agent-pubkey", Some(0.25), Utc::now()); +//! if verdict.is_exhausted() { +//! // Enforcement is the caller's job — see the crate docs. +//! } +//! ``` +//! +//! **Still does not enforce.** The verdict comes back; acting on it is the +//! caller's decision, for the reasons in the crate-level docs. + +use chrono::{DateTime, Duration, Utc}; +use uuid::Uuid; + +use crate::{Ledger, TriggerLog, TurnCharge, Verdict}; + +/// Owns a [`Ledger`] and a [`TriggerLog`] and keeps them in step. +#[derive(Debug)] +pub struct Supervisor { + ledger: Ledger, + triggers: TriggerLog, + window: Duration, +} + +impl Supervisor { + /// Create a supervisor. See [`crate::DEFAULT_BUDGET_USD`] and + /// [`crate::DEFAULT_WINDOW_SECS`] for the values decision D3 settled on. + pub fn new(budget_usd: f64, window_secs: i64) -> Self { + Self { + ledger: Ledger::new(budget_usd, window_secs), + triggers: TriggerLog::new(), + window: Duration::seconds(window_secs.max(1)), + } + } + + /// Record an inbound message so a later turn can be attributed to it. + pub fn observe_message( + &mut self, + channel_id: Uuid, + author: &str, + is_owner: bool, + at: DateTime, + ) { + self.triggers.observe(channel_id, author, is_owner, at); + } + + /// Attribute and charge one completed turn. + /// + /// A turn with no observable trigger — a heartbeat, or an observation gap — + /// is **not charged**, and comes back as [`Verdict::Unbudgeted`]. See + /// [`crate::origin`] for why that fails open rather than closed. + /// + /// Emits the trace events decision **D4** relies on: exhaustion at `warn`, + /// ordinary charges at `debug`. D4 justifies a self-healing window by the + /// **sawtooth it leaves in the logs** — "a diagnosable signature rather than + /// a silent drain" — which only exists if something is emitted. + pub fn on_turn_completed( + &mut self, + channel_id: Uuid, + agent_pubkey: &str, + cost_usd: Option, + turn_end: DateTime, + ) -> Verdict { + let Some(origin) = self + .triggers + .origin_for_turn(channel_id, agent_pubkey, turn_end) + else { + return Verdict::Unbudgeted; + }; + + let verdict = self + .ledger + .record(channel_id, agent_pubkey, &origin, cost_usd, turn_end); + + match &verdict { + Verdict::Exhausted { + spent_usd, + budget_usd, + } => tracing::warn!( + %channel_id, + agent = %agent_pubkey, + spent_usd, + budget_usd, + "agent-pair budget exhausted; exchange should be stopped" + ), + Verdict::Allow { spent_usd } => tracing::debug!( + %channel_id, + agent = %agent_pubkey, + spent_usd, + "agent-triggered turn charged" + ), + Verdict::Unbudgeted => {} + } + verdict + } + + /// Charge a turn decoded straight off the wire by + /// [`crate::charge_from_metric`]. + /// + /// Prefer this over [`Self::on_turn_completed`] when the input came from a + /// `kind:44200` event: it is the only path that inspects + /// [`TurnCharge::delta_reliable`], which the metric sets to `false` when the + /// publisher lost its cumulative baseline and the per-turn cost is + /// therefore untrustworthy. + /// + /// The charge is still applied — ticket #7 **OQ7.1** has not decided a + /// fallback (its suggestion is "prefer `cumulative` where present", which + /// this type does not yet carry). It is logged at `warn` so the question can + /// be answered from real data instead of guessed at. + pub fn on_turn_charge(&mut self, charge: &TurnCharge) -> Verdict { + if !charge.delta_reliable { + tracing::warn!( + channel_id = %charge.channel_id, + agent = %charge.agent_pubkey, + cost_usd = ?charge.cost_usd, + "turn metric reported delta_reliable=false; charging it anyway (see #7 OQ7.1)" + ); + } + self.on_turn_completed( + charge.channel_id, + &charge.agent_pubkey, + charge.cost_usd, + charge.turn_end, + ) + } + + /// Spend for a pair across the current window. + pub fn spent(&self, channel_id: Uuid, a: &str, b: &str, now: DateTime) -> f64 { + self.ledger.spent(channel_id, a, b, now) + } + + /// Drop aged-out state from both halves. A long-running supervisor should + /// call this periodically; nothing else evicts channels that go quiet. + pub fn maintain(&mut self, now: DateTime) { + self.ledger.evict_expired(now); + self.triggers.prune(now, self.window); + } + + /// Pairs currently holding charges — for diagnostics. + pub fn tracked_pairs(&self) -> usize { + self.ledger.tracked_pairs() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ch() -> Uuid { + Uuid::nil() + } + fn t0() -> DateTime { + DateTime::from_timestamp(1_700_000_000, 0).expect("valid timestamp") + } + + const EVA: &str = "eva"; + const OTTO: &str = "otto"; + const OWNER: &str = "owner"; + + /// The whole point, end to end: two agents talking to each other with + /// nobody watching eventually get cut off. + #[test] + fn a_two_agent_runaway_is_eventually_stopped() { + let mut sup = Supervisor::new(5.0, 3600); + let mut stopped_at = None; + + // Eva and Otto alternate, each turn costing $0.30. + for i in 0..40i64 { + let at = t0() + Duration::seconds(i * 2); + let (speaker, listener) = if i % 2 == 0 { (OTTO, EVA) } else { (EVA, OTTO) }; + // The peer's message arrives, then the listener's turn completes. + sup.observe_message(ch(), speaker, false, at); + let v = sup.on_turn_completed(ch(), listener, Some(0.30), at + Duration::seconds(1)); + if v.is_exhausted() { + stopped_at = Some(i); + break; + } + } + + let i = stopped_at.expect("a runaway must eventually be stopped"); + // Every iteration charges exactly one turn, so the loop index is the + // charged-turn count minus one. $5 / $0.30 = 16.67, so cumulative first + // exceeds the budget on the 17th charged turn — index 16. + assert_eq!( + i, 16, + "expected to trip on the 17th charged turn (index 16), tripped at index {i}" + ); + } + + /// The failure this policy must never cause: an agent muted while its + /// owner is actively working with it. + #[test] + fn an_intensive_human_session_is_never_cut_off() { + let mut sup = Supervisor::new(5.0, 3600); + for i in 0..200i64 { + let at = t0() + Duration::seconds(i * 2); + sup.observe_message(ch(), OWNER, true, at); + let v = sup.on_turn_completed(ch(), EVA, Some(1.50), at + Duration::seconds(1)); + assert!( + !v.is_exhausted(), + "human-driven turn {i} must never be blocked" + ); + } + assert_eq!( + sup.spent(ch(), EVA, OWNER, t0() + Duration::seconds(400)), + 0.0 + ); + } + + /// A human joining a runaway rescues it: the mixed batch attributes to the + /// human, so that turn is free. + #[test] + fn a_human_joining_the_channel_makes_that_turn_free() { + let mut sup = Supervisor::new(5.0, 3600); + + sup.observe_message(ch(), OTTO, false, t0()); + sup.on_turn_completed(ch(), EVA, Some(2.0), t0() + Duration::seconds(1)); + assert_eq!(sup.spent(ch(), EVA, OTTO, t0() + Duration::seconds(2)), 2.0); + + // Otto speaks again, but so does the owner, before Eva's next turn. + sup.observe_message(ch(), OTTO, false, t0() + Duration::seconds(2)); + sup.observe_message(ch(), OWNER, true, t0() + Duration::seconds(3)); + sup.on_turn_completed(ch(), EVA, Some(9.0), t0() + Duration::seconds(4)); + + assert_eq!( + sup.spent(ch(), EVA, OTTO, t0() + Duration::seconds(5)), + 2.0, + "a batch containing the owner must not be charged" + ); + } + + #[test] + fn a_runaway_recovers_after_the_window_slides() { + let mut sup = Supervisor::new(5.0, 3600); + sup.observe_message(ch(), OTTO, false, t0()); + let v = sup.on_turn_completed(ch(), EVA, Some(6.0), t0() + Duration::seconds(1)); + assert!(v.is_exhausted()); + + let later = t0() + Duration::seconds(7200); + sup.observe_message(ch(), OTTO, false, later); + let v = sup.on_turn_completed(ch(), EVA, Some(0.10), later + Duration::seconds(1)); + assert!(!v.is_exhausted(), "must self-heal with no reset"); + } + + /// An observation gap fails open — documented, and asserted so a future + /// change cannot flip it silently. + #[test] + fn an_unobserved_turn_is_not_charged() { + let mut sup = Supervisor::new(5.0, 3600); + // No observe_message call at all. + for i in 0..50i64 { + let v = sup.on_turn_completed(ch(), EVA, Some(10.0), t0() + Duration::seconds(i)); + assert!(!v.is_exhausted(), "unattributed turns must fail open"); + } + assert_eq!(sup.tracked_pairs(), 0); + } + + /// `on_turn_charge` must be equivalent to the plain path — the + /// `delta_reliable` inspection is additive, not a different policy. + #[test] + fn charging_from_a_wire_decoded_turn_matches_the_plain_path() { + use crate::TurnCharge; + + let mut sup = Supervisor::new(5.0, 3600); + sup.observe_message(ch(), OTTO, false, t0()); + let v = sup.on_turn_charge(&TurnCharge { + channel_id: ch(), + agent_pubkey: EVA.into(), + cost_usd: Some(6.0), + turn_end: t0() + Duration::seconds(1), + delta_reliable: true, + }); + assert!(v.is_exhausted()); + assert_eq!(sup.spent(ch(), EVA, OTTO, t0() + Duration::seconds(2)), 6.0); + } + + /// An unreliable delta is still charged — OQ7.1 is unresolved, so the + /// behaviour must be explicit and tested rather than accidental. + #[test] + fn an_unreliable_delta_is_still_charged() { + use crate::TurnCharge; + + let mut sup = Supervisor::new(5.0, 3600); + sup.observe_message(ch(), OTTO, false, t0()); + sup.on_turn_charge(&TurnCharge { + channel_id: ch(), + agent_pubkey: EVA.into(), + cost_usd: Some(2.5), + turn_end: t0() + Duration::seconds(1), + delta_reliable: false, + }); + assert_eq!(sup.spent(ch(), EVA, OTTO, t0() + Duration::seconds(2)), 2.5); + } + + #[test] + fn maintain_clears_state_for_channels_that_went_quiet() { + let mut sup = Supervisor::new(5.0, 3600); + sup.observe_message(ch(), OTTO, false, t0()); + sup.on_turn_completed(ch(), EVA, Some(1.0), t0() + Duration::seconds(1)); + assert_eq!(sup.tracked_pairs(), 1); + + sup.maintain(t0() + Duration::seconds(7200)); + assert_eq!(sup.tracked_pairs(), 0); + } + + /// Two separate agent pairs in one channel must not pool their budgets. + #[test] + fn separate_pairs_have_separate_budgets() { + let mut sup = Supervisor::new(5.0, 3600); + let third = "iris"; + + sup.observe_message(ch(), OTTO, false, t0()); + let v = sup.on_turn_completed(ch(), EVA, Some(6.0), t0() + Duration::seconds(1)); + assert!(v.is_exhausted(), "eva/otto is over budget"); + + sup.observe_message(ch(), third, false, t0() + Duration::seconds(2)); + let v = sup.on_turn_completed(ch(), EVA, Some(0.5), t0() + Duration::seconds(3)); + assert!( + !v.is_exhausted(), + "eva/iris must have its own budget; got {v:?}" + ); + } +} diff --git a/crates/buzz-cli/src/commands/agents.rs b/crates/buzz-cli/src/commands/agents.rs index 58564a45c2f..fcdee150609 100644 --- a/crates/buzz-cli/src/commands/agents.rs +++ b/crates/buzz-cli/src/commands/agents.rs @@ -93,7 +93,8 @@ pub async fn dispatch(command: AgentsCmd, client: &BuzzClient) -> Result<(), Cli } => { validate_hex64(&target_pubkey)?; let signer_hex = client.keys().public_key().to_hex(); - let auth = resolve_auth(client, &target_pubkey, &signer_hex).await?; + let auth = + resolve_auth(client, &target_pubkey, &signer_hex, &mut std::io::stderr()).await?; let builder = build_archive_identity_request( &target_pubkey, &content, @@ -124,7 +125,8 @@ pub async fn dispatch(command: AgentsCmd, client: &BuzzClient) -> Result<(), Cli } => { validate_hex64(&target_pubkey)?; let signer_hex = client.keys().public_key().to_hex(); - let auth = resolve_auth(client, &target_pubkey, &signer_hex).await?; + let auth = + resolve_auth(client, &target_pubkey, &signer_hex, &mut std::io::stderr()).await?; let builder = build_unarchive_identity_request( &target_pubkey, &content, @@ -160,19 +162,175 @@ fn require_owner(client: &BuzzClient) -> Result { PublicKey::parse(&hex).map_err(|e| CliError::Auth(format!("invalid owner attestation: {e}"))) } +/// Typed reason why NIP-OA owner-auth could not be extracted from a kind:0. +/// +/// Produced by [`classify_owner_auth_tag`] and formatted into a JSON warning +/// by [`resolve_auth`]. One variant per distinguishable failure cause so the +/// diagnostic is always accurate and never duplicates validation logic. +#[derive(Debug, PartialEq)] +enum AuthFailure { + /// kind:0 has no `tags` array or the array is empty of `auth`-labelled entries. + NoAuthTag, + /// kind:0 has more than one `auth`-labelled tag; count included. + AmbiguousAuthTag(usize), + /// Sole `auth` tag has wrong element count; actual count included. + WrongArity(usize), + /// Sole `auth` tag contains a non-string element. + NonStringElement, + /// Sole `auth` tag owner field is not a valid 64-hex pubkey; value included. + InvalidOwnerHex(String), + /// Sole `auth` tag sig field is not a valid 128-hex signature. + InvalidSigHex, + /// Tag is structurally valid but names a different owner; actual owner included. + OwnerMismatch(String), +} + +impl AuthFailure { + /// Human-readable description suitable for the `"warning"` JSON field. + fn message(&self) -> String { + match self { + AuthFailure::NoAuthTag => "target kind:0 has no \"auth\" tag".to_owned(), + AuthFailure::AmbiguousAuthTag(n) => format!( + "target kind:0 has {n} \"auth\" tags (expected exactly 1) — ambiguous ownership" + ), + AuthFailure::WrongArity(n) => format!( + "sole \"auth\" tag has {n} element(s) (expected 4: label, owner, conditions, sig)" + ), + AuthFailure::NonStringElement => { + "sole \"auth\" tag contains a non-string element".to_owned() + } + AuthFailure::InvalidOwnerHex(v) => { + format!("sole \"auth\" tag owner field is not a valid 64-hex pubkey: {v}") + } + AuthFailure::InvalidSigHex => { + "sole \"auth\" tag sig field is not a valid 128-hex signature".to_owned() + } + AuthFailure::OwnerMismatch(actual) => { + format!("sole \"auth\" tag names owner {actual} which does not match your key") + } + } + } +} + +/// Single classifier: either extract the auth tag or return the typed reason +/// for failure. [`extract_owner_auth_tag`] is a thin `.ok()` wrapper kept for +/// the existing tests that assert on `Option`. +fn classify_owner_auth_tag( + tags: &[serde_json::Value], + signer_hex: &str, +) -> Result<[String; 4], AuthFailure> { + let auth_tags: Vec<&serde_json::Value> = tags + .iter() + .filter(|tag| { + tag.as_array() + .and_then(|elems| elems.first()) + .and_then(|v| v.as_str()) + == Some("auth") + }) + .collect(); + match auth_tags.len() { + 0 => return Err(AuthFailure::NoAuthTag), + n if n > 1 => return Err(AuthFailure::AmbiguousAuthTag(n)), + _ => {} + } + + // Exactly one auth tag. + let elems = auth_tags[0] + .as_array() + .ok_or(AuthFailure::NonStringElement)?; + if elems.len() != 4 { + return Err(AuthFailure::WrongArity(elems.len())); + } + let label = elems[0].as_str().ok_or(AuthFailure::NonStringElement)?; + let owner = elems[1].as_str().ok_or(AuthFailure::NonStringElement)?; + let conditions = elems[2].as_str().ok_or(AuthFailure::NonStringElement)?; + let sig = elems[3].as_str().ok_or(AuthFailure::NonStringElement)?; + if owner.len() != 64 || !owner.chars().all(|c| c.is_ascii_hexdigit()) { + return Err(AuthFailure::InvalidOwnerHex(owner.to_owned())); + } + if sig.len() != 128 || !sig.chars().all(|c| c.is_ascii_hexdigit()) { + return Err(AuthFailure::InvalidSigHex); + } + if !owner.eq_ignore_ascii_case(signer_hex) { + return Err(AuthFailure::OwnerMismatch(owner.to_owned())); + } + Ok([ + label.to_owned(), + owner.to_owned(), + conditions.to_owned(), + sig.to_owned(), + ]) +} + +/// Pure sync core of auth resolution: given a fetched kind:0 profile (or +/// `None` when no event was found), either return the extracted auth tag or +/// emit one `{"warning":"..."}` JSON line to `warn_sink` and return `None`. +/// +/// Separated from [`resolve_auth`] so unit tests can call this directly with +/// a `Vec` sink and assert on exactly what hits the wire — without needing +/// a live `BuzzClient` or async runtime. +/// +/// Three warning branches, one success path: +/// 1. `profile == None` → no kind:0 found for target. +/// 2. `profile.get("tags")` absent or non-array → no tags array. +/// 3. [`classify_owner_auth_tag`] returns `Err` → typed failure reason. +/// 4. `classify_owner_auth_tag` returns `Ok` → `Some(tag)`, no warning. +fn resolve_auth_from_profile( + profile: Option<&serde_json::Value>, + target_hex: &str, + signer_hex: &str, + warn_sink: &mut dyn std::io::Write, +) -> Option<[String; 4]> { + let event = match profile { + Some(e) => e, + None => { + let msg = format!( + "no kind:0 profile found for target {target_hex}; \ + proceeding without owner attestation — this succeeds only if your key is a relay admin" + ); + let _ = writeln!(warn_sink, "{}", serde_json::json!({"warning": msg})); + return None; + } + }; + let tags = match event.get("tags").and_then(|v| v.as_array()) { + Some(t) => t, + None => { + let msg = format!( + "target {target_hex} kind:0 has no tags array; \ + proceeding without owner attestation — this succeeds only if your key is a relay admin" + ); + let _ = writeln!(warn_sink, "{}", serde_json::json!({"warning": msg})); + return None; + } + }; + match classify_owner_auth_tag(tags, signer_hex) { + Ok(tag) => Some(tag), + Err(failure) => { + let msg = format!( + "{}; proceeding without owner attestation — \ + this succeeds only if your key is a relay admin", + failure.message() + ); + let _ = writeln!(warn_sink, "{}", serde_json::json!({"warning": msg})); + None + } + } +} + /// Resolve the optional NIP-OA `auth` tag for archive/unarchive requests. /// /// Mirrors the desktop's `maybe_owner_auth_tag`: -/// - `target == signer`: self path — no auth needed → `Ok(None)`. -/// - Otherwise: fetch target's kind:0, look for an `auth` tag whose owner -/// (index 1) matches the signer. Return it when present; `Ok(None)` when -/// absent or structurally malformed. Query/network failures surface as -/// `Err` — silent degradation to bare would make the relay reject the -/// request with a misleading error. +/// - `target == signer`: self path — no auth needed → `Ok(None)`, silent. +/// - Otherwise: fetch target's kind:0, delegate to [`resolve_auth_from_profile`] +/// which either returns the extracted tag or emits one `{"warning":"..."}` JSON +/// line to `warn_sink` and returns `None` — the bare request is still sent so +/// relay admins can succeed without owner attestation. Query/network failures +/// surface as `Err`. async fn resolve_auth( client: &BuzzClient, target_hex: &str, signer_hex: &str, + warn_sink: &mut dyn std::io::Write, ) -> Result, CliError> { if target_hex.eq_ignore_ascii_case(signer_hex) { return Ok(None); @@ -184,15 +342,13 @@ async fn resolve_auth( .map_err(|e| CliError::Other(format!("failed to fetch target kind:0: {e}")))?; let events: Vec = serde_json::from_str(&raw) .map_err(|e| CliError::Other(format!("invalid kind:0 query response: {e}")))?; - let event = match events.into_iter().next() { - Some(e) => e, - None => return Ok(None), - }; - let tags = match event.get("tags").and_then(|v| v.as_array()) { - Some(t) => t, - None => return Ok(None), - }; - Ok(extract_owner_auth_tag(tags, signer_hex)) + let profile = events.into_iter().next(); + Ok(resolve_auth_from_profile( + profile.as_ref(), + target_hex, + signer_hex, + warn_sink, + )) } /// Pure extraction helper: require exactly one kind:0 tag whose first @@ -201,46 +357,11 @@ async fn resolve_auth( /// then structurally validate that sole tag as /// `["auth", owner, conditions, sig]` matching `signer_hex`. /// -/// Malformed tags (wrong arity, non-string elements, non-hex fields) are -/// silently skipped — the contract is "bare" (None), not error. +/// Thin wrapper around [`classify_owner_auth_tag`] that collapses the typed +/// failure reason to `None`. Malformed tags → `None`; valid tag → `Some`. +#[cfg(test)] fn extract_owner_auth_tag(tags: &[serde_json::Value], signer_hex: &str) -> Option<[String; 4]> { - let auth_tags: Vec<&serde_json::Value> = tags - .iter() - .filter(|tag| { - tag.as_array() - .and_then(|elems| elems.first()) - .and_then(|v| v.as_str()) - == Some("auth") - }) - .collect(); - if auth_tags.len() != 1 { - return None; - } - - let elems = auth_tags[0].as_array()?; - if elems.len() != 4 { - return None; - } - let label = elems[0].as_str()?; - let owner = elems[1].as_str()?; - if !owner.eq_ignore_ascii_case(signer_hex) { - return None; - } - let conditions = elems[2].as_str()?; - let sig = elems[3].as_str()?; - if owner.len() != 64 - || !owner.chars().all(|c| c.is_ascii_hexdigit()) - || sig.len() != 128 - || !sig.chars().all(|c| c.is_ascii_hexdigit()) - { - return None; - } - Some([ - label.to_owned(), - owner.to_owned(), - conditions.to_owned(), - sig.to_owned(), - ]) + classify_owner_auth_tag(tags, signer_hex).ok() } /// Validate the NIP-11 relay-info `self` field is a 64-hex pubkey and @@ -521,6 +642,242 @@ mod tests { assert!(extract_owner_auth_tag(&tags, &signer).is_none()); } + // --- (c) auth-failure classifier: classify_owner_auth_tag --- + // + // Tests the typed failure taxonomy. Each case asserts the exact + // AuthFailure variant so a wrong classification causes a compile-time or + // assertion failure — not just a message-substring miss. + + #[test] + fn classify_no_auth_tag_returns_no_auth_tag() { + // Case 3 (zero auth tags): tags array has entries but none labelled "auth". + let signer = hex64('a'); + let tags = vec![json!(["p", hex64('b')]), json!(["e", hex64('c')])]; + assert_eq!( + classify_owner_auth_tag(&tags, &signer), + Err(AuthFailure::NoAuthTag) + ); + } + + #[test] + fn classify_empty_tags_returns_no_auth_tag() { + assert_eq!( + classify_owner_auth_tag(&[], &hex64('a')), + Err(AuthFailure::NoAuthTag) + ); + } + + #[test] + fn classify_duplicate_auth_tags_returns_ambiguous() { + let signer = hex64('a'); + let sig = hex128('b'); + let tags = vec![ + json!(["auth", signer, "conditions", sig]), + json!(["auth", signer, "conditions", sig]), + ]; + assert_eq!( + classify_owner_auth_tag(&tags, &signer), + Err(AuthFailure::AmbiguousAuthTag(2)) + ); + } + + #[test] + fn classify_wrong_arity_returns_wrong_arity() { + let signer = hex64('a'); + let tags = vec![json!(["auth", signer, "conditions"])]; + assert_eq!( + classify_owner_auth_tag(&tags, &signer), + Err(AuthFailure::WrongArity(3)) + ); + } + + #[test] + fn classify_non_string_element_returns_non_string() { + let signer = hex64('a'); + let tags = vec![json!(["auth", signer, 42, hex128('b')])]; + assert_eq!( + classify_owner_auth_tag(&tags, &signer), + Err(AuthFailure::NonStringElement) + ); + } + + #[test] + fn classify_invalid_owner_hex_returns_invalid_owner_hex() { + let bad_owner = "z".repeat(64); + let tags = vec![json!(["auth", bad_owner, "", hex128('a')])]; + assert_eq!( + classify_owner_auth_tag(&tags, &bad_owner), + Err(AuthFailure::InvalidOwnerHex(bad_owner)) + ); + } + + #[test] + fn classify_invalid_sig_hex_returns_invalid_sig_hex() { + let signer = hex64('a'); + let bad_sig = "z".repeat(128); + let tags = vec![json!(["auth", signer, "", bad_sig])]; + assert_eq!( + classify_owner_auth_tag(&tags, &signer), + Err(AuthFailure::InvalidSigHex) + ); + } + + #[test] + fn classify_owner_mismatch_returns_owner_mismatch_with_actual_owner() { + // Case 4: structurally valid tag but owner ≠ signer. The failure must + // carry the actual owner so resolve_auth can print it in the warning. + let actual_owner = hex64('a'); + let signer = hex64('b'); + let sig = hex128('c'); + let tags = vec![json!(["auth", actual_owner, "conditions", sig])]; + assert_eq!( + classify_owner_auth_tag(&tags, &signer), + Err(AuthFailure::OwnerMismatch(actual_owner.clone())) + ); + // Message must include the actual owner for actionability. + let msg = AuthFailure::OwnerMismatch(actual_owner.clone()).message(); + assert!( + msg.contains(&actual_owner), + "OwnerMismatch message must include actual owner, got: {msg}" + ); + } + + // --- (c2) resolve_auth_from_profile emission boundary --- + // + // Observable-boundary tests: each test calls the production function + // `resolve_auth_from_profile` directly with a `Vec` sink and asserts + // on exactly what the production code writes. Deleting any `writeln!` + // call in that function makes at least one of these tests fail. + // + // `resolve_auth` is async and requires a live `BuzzClient`; the sync + // decomposition lets us test the warning logic without a relay connection. + + fn assert_one_json_warning(sink: &[u8], expected_fragment: &str) { + let text = std::str::from_utf8(sink).expect("sink is valid UTF-8"); + let lines: Vec<&str> = text.lines().collect(); + assert_eq!( + lines.len(), + 1, + "expected exactly one warning line, got: {text:?}" + ); + let parsed: serde_json::Value = + serde_json::from_str(lines[0]).expect("warning line must be parseable JSON"); + let warning = parsed["warning"] + .as_str() + .expect("warning line must have a string 'warning' field"); + assert!( + warning.contains(expected_fragment), + "warning must contain {expected_fragment:?}, got: {warning}" + ); + } + + fn assert_no_warning(sink: &[u8]) { + let text = std::str::from_utf8(sink).expect("sink is valid UTF-8"); + assert!(text.is_empty(), "expected no warning output, got: {text:?}"); + } + + // Branch 1: profile == None → no kind:0 found. + #[test] + fn resolve_auth_from_profile_no_kind0_emits_json_warning() { + let target = hex64('t'); + let signer = hex64('s'); + let mut sink: Vec = Vec::new(); + let result = resolve_auth_from_profile(None, &target, &signer, &mut sink); + assert!(result.is_none(), "bare path: must return None"); + assert_one_json_warning(&sink, "no kind:0 profile found"); + } + + // Branch 2: profile present but no tags array. + #[test] + fn resolve_auth_from_profile_no_tags_array_emits_json_warning() { + let target = hex64('t'); + let signer = hex64('s'); + let profile = json!({"kind": 0, "content": "{}"}); + let mut sink: Vec = Vec::new(); + let result = resolve_auth_from_profile(Some(&profile), &target, &signer, &mut sink); + assert!(result.is_none(), "bare path: must return None"); + assert_one_json_warning(&sink, "no tags array"); + } + + // Branch 3a: tags present but no auth tag (NoAuthTag). + #[test] + fn resolve_auth_from_profile_no_auth_tag_emits_json_warning() { + let target = hex64('t'); + let signer = hex64('s'); + let profile = json!({"tags": [["p", hex64('b')]]}); + let mut sink: Vec = Vec::new(); + let result = resolve_auth_from_profile(Some(&profile), &target, &signer, &mut sink); + assert!(result.is_none(), "bare path: must return None"); + assert_one_json_warning(&sink, "no \"auth\" tag"); + } + + // Branch 3b: duplicate auth tags (AmbiguousAuthTag). + #[test] + fn resolve_auth_from_profile_ambiguous_auth_tag_emits_json_warning() { + let signer = hex64('s'); + let sig = hex128('b'); + let profile = json!({"tags": [ + ["auth", signer, "conditions", sig], + ["auth", signer, "conditions", sig], + ]}); + let mut sink: Vec = Vec::new(); + let result = resolve_auth_from_profile(Some(&profile), &hex64('t'), &signer, &mut sink); + assert!(result.is_none(), "bare path: must return None"); + assert_one_json_warning(&sink, "ambiguous"); + } + + // Branch 3c: sole auth tag malformed (WrongArity). + #[test] + fn resolve_auth_from_profile_malformed_tag_emits_json_warning() { + let signer = hex64('s'); + // arity 3 — missing sig field + let profile = json!({"tags": [["auth", signer, "conditions"]]}); + let mut sink: Vec = Vec::new(); + let result = resolve_auth_from_profile(Some(&profile), &hex64('t'), &signer, &mut sink); + assert!(result.is_none(), "bare path: must return None"); + assert_one_json_warning(&sink, "element"); + } + + // Branch 3d: owner mismatch — warning must include the actual owner pubkey. + #[test] + fn resolve_auth_from_profile_owner_mismatch_emits_json_warning_with_actual_owner() { + let actual_owner = hex64('a'); + let signer = hex64('b'); + let sig = hex128('c'); + let profile = json!({"tags": [["auth", actual_owner, "conditions", sig]]}); + let mut sink: Vec = Vec::new(); + let result = resolve_auth_from_profile(Some(&profile), &hex64('t'), &signer, &mut sink); + assert!(result.is_none(), "bare path: must return None"); + assert_one_json_warning(&sink, &actual_owner); + } + + // Success path: valid auth tag → Some returned, sink stays empty. + #[test] + fn resolve_auth_from_profile_valid_auth_tag_returns_some_emits_nothing() { + let signer = hex64('a'); + let sig = hex128('b'); + let profile = json!({"tags": [["auth", signer, "conditions", sig]]}); + let mut sink: Vec = Vec::new(); + let result = resolve_auth_from_profile(Some(&profile), &hex64('t'), &signer, &mut sink); + assert!(result.is_some(), "must return the extracted tag"); + assert_no_warning(&sink); + } + + // Warning output must be valid JSON (serde_json serializes safely). + #[test] + fn resolve_auth_from_profile_warning_is_valid_json() { + let actual_owner = hex64('a'); + let signer = hex64('b'); + let sig = hex128('c'); + let profile = json!({"tags": [["auth", actual_owner, "conditions", sig]]}); + let mut sink: Vec = Vec::new(); + let _ = resolve_auth_from_profile(Some(&profile), &hex64('t'), &signer, &mut sink); + let text = std::str::from_utf8(&sink).unwrap(); + let parsed: serde_json::Value = + serde_json::from_str(text.trim()).expect("warning output must be valid JSON"); + assert!(parsed["warning"].is_string()); + } + // --- (d) NIP-11 self normalization: normalize_relay_self_hex --- #[test] diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index cd9f20b5f4a..55a468144c1 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -28,12 +28,13 @@ use buzz_core::kind::{ KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, KIND_NIP43_LEAVE_REQUEST, KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, KIND_PRESENCE_UPDATE, - KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_PROJECT, KIND_REACTION, KIND_READ_STATE, KIND_REPORT, - KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, - KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, - KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, KIND_TEAM_CATALOG, KIND_TEXT_NOTE, - KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, - RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, + KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_PROJECT, KIND_REACTION, + KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, + KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, + KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, + KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, + RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, + RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; @@ -263,7 +264,7 @@ fn required_scope_for_kind(kind: u32, event: &Event) -> Result Ok(Scope::MessagesWrite), KIND_CONTACT_LIST | KIND_READ_STATE | KIND_USER_STATUS | KIND_AGENT_ENGRAM | KIND_EVENT_REMINDER | KIND_PERSONA | KIND_TEAM | KIND_MANAGED_AGENT - | KIND_TEAM_CATALOG | super::push_lease::KIND_PUSH_LEASE => { + | KIND_PRIVATE_MANAGED_AGENT | KIND_TEAM_CATALOG | super::push_lease::KIND_PUSH_LEASE => { Ok(Scope::UsersWrite) } // NIP-AM: agent turn metrics are agent-authored global events (encrypted to owner). @@ -476,6 +477,7 @@ pub(crate) fn is_global_only_kind(kind: u32) -> bool { // (pubkey, kind, d_tag). A stray `h` tag must not channel-scope them. | KIND_TEAM | KIND_MANAGED_AGENT + | KIND_PRIVATE_MANAGED_AGENT | KIND_TEAM_CATALOG // NIP-34: git events use `a` tags (repo reference), not `h` tags (channel scope). // Parameterized replaceable kinds are keyed by (pubkey, kind, d_tag). @@ -3338,15 +3340,14 @@ mod tests { } #[test] - fn private_managed_agent_kind_remains_rejected_until_atomic_ingest_exists() { - assert!( - required_scope_for_kind( - buzz_core::kind::KIND_PRIVATE_MANAGED_AGENT, - &make_dummy_event(), - ) - .is_err(), - "kind 30179 must not enter generic EVENT ingest before privacy and aggregate CAS deploy" + fn private_managed_agent_kind_is_owner_scoped_global_user_data() { + let event = make_dummy_event(); + assert_eq!( + required_scope_for_kind(KIND_PRIVATE_MANAGED_AGENT, &event), + Ok(Scope::UsersWrite) ); + assert!(is_global_only_kind(KIND_PRIVATE_MANAGED_AGENT)); + assert!(!requires_h_channel_scope(KIND_PRIVATE_MANAGED_AGENT)); } #[test] diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index b929dbb6131..b578326eba3 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -113,13 +113,16 @@ with a TypeScript lookup table or an id comparison in a component. Once the Advanced toggle is visible, its expanded state is exclusively user-controlled: provider, harness, and required-env changes must never open it automatically in defaults, create, or edit flows. In Create mode, - the defaults summary follows preferred-harness changes saved while the - dialog is open, and its configured state includes required credentials as - well as provider/model values. If no available harness can resolve, Create - starts in Customize and lets unavailable catalog entries be selected only - to expose their setup guidance; submission remains blocked. - Advanced-only required credentials mark the collapsed Advanced toggle - without opening it in Global Defaults and Edit, and block incomplete saves. + `Run on` belongs in Advanced directly after **Who can send instructions**; + keep it out of the basic create fields. The defaults summary follows + preferred-harness changes saved while the dialog is open, and its configured + state includes required credentials as well as provider/model values. If no + available harness can resolve, Create starts in Customize and lets unavailable + catalog entries be selected only to expose their setup guidance; submission + remains blocked. + Advanced-only required credentials and incomplete remote **Run on** setup + mark the collapsed Advanced toggle without opening it, and block incomplete + saves. Runtime-file credentials satisfy Global Defaults just as they do Create and Edit. In Edit, selecting Custom command keeps its required command field beside the harness diff --git a/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx b/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx index 12702f45ac4..409ae6a8214 100644 --- a/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx +++ b/desktop/src/features/agents/ui/AgentDefinitionDialog.tsx @@ -109,7 +109,6 @@ type AgentDefinitionDialogProps = { ) => Promise; /** Publishes saved changes when the edited agent is shared in the catalog. */ publishCatalogUpdatesOnSave?: boolean; - /** Rendered below the form fields in create mode only ("Where to run"). */ createRunSection?: React.ReactNode; /** Extra create-mode submit gate (e.g. incomplete provider config). */ createSubmitBlocked?: boolean; @@ -962,9 +961,6 @@ export function AgentDefinitionDialog({ onSaved={selectSavedHarness} open={isAddHarnessOpen} /> - - {isCreateMode ? createRunSection : null} -
) : null} @@ -1109,33 +1111,22 @@ export function VideoPlayer({ ) : null} - {/* Slide (not fade) the pill out: animating opacity on an ancestor - of a backdrop-filter flattens the glass into a plain fill - mid-transition, which reads as a flicker. The video container's - overflow-hidden clips the slid-out pill. */} - {showControls ? ( + {!hasError ? (
-
- - + +
(null); + const [hasVisibleFrame, setHasVisibleFrame] = React.useState(false); const [videoAreaSize, setVideoAreaSize] = React.useState<{ height: number; width: number; @@ -1364,6 +1356,7 @@ function VideoReviewDialog({ React.useEffect(() => { if (!open) { setIsComposerMounted(false); + setHasVisibleFrame(false); return; } // Two frames: one for the dialog to paint, one for the browser to @@ -1715,7 +1708,7 @@ function VideoReviewDialog({ className="h-full w-full min-h-0 object-contain" playsInline poster={poster} - preload="metadata" + preload="auto" src={src} onClick={togglePlay} onDurationChange={(event) => @@ -1740,6 +1733,7 @@ function VideoReviewDialog({ syncCurrentTime(pendingSeekSeconds); } }} + onLoadedData={() => setHasVisibleFrame(true)} onPause={(event) => { syncCurrentTime(event.currentTarget.currentTime); setIsPlaying(false); @@ -1748,7 +1742,15 @@ function VideoReviewDialog({ syncCurrentTime(event.currentTarget.currentTime); setIsPlaying(true); }} - onSeeked={reviewSeek.handleSeeked} + onSeeked={(event) => { + reviewSeek.handleSeeked(); + if ( + event.currentTarget.readyState >= + HTMLMediaElement.HAVE_CURRENT_DATA + ) { + setHasVisibleFrame(true); + } + }} onTimeUpdate={(event) => { syncCurrentTime(event.currentTarget.currentTime); }} @@ -1757,6 +1759,10 @@ function VideoReviewDialog({ setMuted(event.currentTarget.muted); }} /> +
@@ -1903,12 +1909,12 @@ function VideoReviewDialog({ {showCommentsPanel ? (