feat(archive): add agent turn-metric (kind 44200) local archive (#1555)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
2026-07-06 18:30:22 -04:00
committed by GitHub
parent d8355d3188
commit 429f54b4f4
17 changed files with 1170 additions and 70 deletions

View File

@@ -69,7 +69,25 @@ const overrides = new Map([
// E2E test-depth hardening added the owner_p content round-trip assert and
// two empty-table drop asserts (~24 lines). Queued to split the test module
// into archive/mod_tests.rs in a follow-up.
["src-tauri/src/archive/mod.rs", 1465],
// agent-metric-archive PR added 4 new unit tests (owner_p+44200 routing,
// decrypt-success plaintext storage, decrypt-fail-closed, 24200 still
// ephemeral) + run_batch_sync_with_keys helper (~175 lines). Same test-growth
// category as above. Still queued to split.
// merge_save_subscription_kinds command + owner_p-kinds TOCTOU fix adds ~30
// lines. Atomic merge to close the concurrent-seed race. Still queued to split.
// IMMEDIATE-tx fix: BEGIN IMMEDIATE replaces DEFERRED unchecked_transaction
// in merge_owner_p_kinds comment block (~5 lines). Still queued to split.
// doc-comment: mixed row-shape invariant on read_archived_events (~5 lines).
["src-tauri/src/archive/mod.rs", 1705],
// archive/store.rs: merge_owner_p_kinds fn (read+union+upsert under a single
// SQLite tx) + 4 unit tests (create-when-none, adds-kind, idempotent,
// concurrent-interleave). Load-bearing TOCTOU fix for the owner_p shared row.
// Queued to split test module into store_tests.rs in a follow-up.
// IMMEDIATE-tx fix: replaces mislabeled sequential test with a real
// two-connection WAL regression test (tempfile + std::thread + Barrier,
// ~85 lines). The new test exercises the actual concurrent write path and
// fails fast if IMMEDIATE guard is removed. Still queued to split.
["src-tauri/src/archive/store.rs", 1179],
["src-tauri/src/commands/agents.rs", 1437],
// #1418 read-path fix: get_thread_replies' blocker fix (shared TIMELINE_KINDS
// const + build_thread_replies_filter helper, mirroring the channel sibling so

View File

@@ -14,6 +14,7 @@ fn main() {
println!("cargo:rerun-if-env-changed=BUZZ_BUILD_AGENT_ENV");
println!("cargo:rerun-if-env-changed=BUZZ_BUILD_RELAY_RECONNECT_CMD");
println!("cargo:rerun-if-env-changed=BUZZ_BUILD_OBSERVER_ARCHIVE_DEFAULT");
println!("cargo:rerun-if-env-changed=BUZZ_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT");
println!("cargo:rustc-check-cfg=cfg(buzz_updater_enabled)");
if let Ok(relay_url) = std::env::var("BUZZ_RELAY_URL") {
@@ -81,6 +82,13 @@ fn main() {
println!("cargo:rustc-env=BUZZ_DESKTOP_BUILD_OBSERVER_ARCHIVE_DEFAULT=1");
}
// Presence-only flag: when set (any non-empty value), agent-turn-metric
// archive defaults to ON for the current identity on first run. OSS builds
// leave this unset → default OFF.
if std::env::var("BUZZ_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT").is_ok() {
println!("cargo:rustc-env=BUZZ_DESKTOP_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT=1");
}
let updater_public_key = std::env::var("BUZZ_UPDATER_PUBLIC_KEY")
.ok()
.map(|value| value.trim().to_string())

View File

@@ -5,9 +5,11 @@
//!
//! Two access proof paths, chosen by event kind:
//!
//! **Persistent scopes** (`channel_h`, `referenced_e`): the relay is the
//! source of truth. Candidates are grouped and re-queried via a batched
//! authed `/query`; only events the relay returns are inserted.
//! **Persistent scopes** (`channel_h`, `referenced_e`, and `owner_p`+44200):
//! the relay is the source of truth. Candidates are grouped and re-queried via
//! a batched authed `/query`; only events the relay returns are inserted.
//! For kind-44200 (agent turn metrics), content is decrypted at ingest and
//! stored as plaintext JSON — fail-closed (decrypt error → drop).
//!
//! **Ephemeral scope** (`owner_p`, kind 24200 observer frames): the relay
//! never stores these, so `/query` cannot verify them. The relay's REQ-time
@@ -32,6 +34,7 @@ use crate::relay::{query_relay, relay_ws_url_with_override};
// ── Constants ───────────────────────────────────────────────────────────────
const KIND_AGENT_OBSERVER_FRAME: u16 = 24200;
const KIND_AGENT_TURN_METRIC: u16 = 44200;
const OBSERVER_FRAME_TELEMETRY: &str = "telemetry";
// ── DB helpers ───────────────────────────────────────────────────────────────
@@ -143,12 +146,18 @@ pub async fn archive_events(
// ── Phase 3: persist (sync) ──────────────────────────────────────────────
let conn = open_db()?;
let owner_keys = {
let keys_guard = state.keys.lock().map_err(|e| e.to_string())?;
keys_guard.clone()
// guard drops here
};
commit_archive(
bucket_results,
plan.ephemeral,
plan.pre_dropped,
&identity_pk,
&relay_url,
&owner_keys,
now,
&conn,
)
@@ -375,6 +384,32 @@ async fn probe_event_readable(state: &AppState, event_id: &str) -> Result<(), St
Ok(())
}
// ── merge_save_subscription_kinds ────────────────────────────────────────────
/// Atomically merge `kind` into the `owner_p` save subscription for the
/// current identity + relay.
///
/// Reads the existing `kinds` array, unions in `kind`, and writes back — all
/// inside a single SQLite transaction. This prevents the TOCTOU race where two
/// concurrent seed hooks (observer seed + metric seed) each read an empty row
/// and the last writer clobbers the first.
///
/// Called by both `useObserverArchiveSeed` (kind 24200) and
/// `useAgentMetricArchiveSeed` (kind 44200) instead of the former
/// list → merge-in-TS → create pattern.
#[tauri::command]
pub fn merge_save_subscription_kinds(state: State<'_, AppState>, kind: u32) -> Result<(), String> {
if kind > u32::from(u16::MAX) {
return Err(format!("kind {kind} is out of the valid range 0..=65535"));
}
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
let now = now_secs();
let conn = open_db()?;
store::merge_owner_p_kinds(&conn, &identity_pk, &relay_url, &identity_pk, kind, now)
}
// ── list_save_subscriptions ──────────────────────────────────────────────────
/// List all save subscriptions for the current identity + relay.
@@ -429,6 +464,11 @@ const DEFAULT_READ_LIMIT: i64 = 50;
///
/// `kinds` is an optional filter; an empty array means "no kinds matched"
/// (not "all kinds") — callers should pass `null`/`None` when they want all.
///
/// Note: stored row payloads are not uniform — kind 44200 rows store the raw
/// metric payload JSON, while all other kinds store full Nostr Event JSON. A
/// caller doing `Event::from_json` on an unfiltered read must filter by kind
/// first (today's only reader filters `kinds: [24200]`).
#[tauri::command]
pub fn read_archived_events(
state: State<'_, AppState>,
@@ -519,6 +559,26 @@ mod tests {
relay_url: &str,
conn: &Connection,
fake_relay_events: Vec<Event>,
) -> ArchiveBatchResult {
let owner_keys = Keys::generate();
run_batch_sync_with_keys(
candidates,
identity_pk,
relay_url,
conn,
fake_relay_events,
&owner_keys,
)
}
/// Like `run_batch_sync` but with a specific owner `Keys` for decrypt.
fn run_batch_sync_with_keys(
candidates: Vec<ArchiveCandidate>,
identity_pk: &str,
relay_url: &str,
conn: &Connection,
fake_relay_events: Vec<Event>,
owner_keys: &Keys,
) -> ArchiveBatchResult {
let plan = plan_archive(candidates, identity_pk, relay_url, conn).unwrap();
@@ -544,6 +604,7 @@ mod tests {
plan.pre_dropped,
identity_pk,
relay_url,
owner_keys,
0,
conn,
)
@@ -1064,6 +1125,183 @@ mod tests {
);
}
// ── Kind-44200 agent-turn-metric archive tests ───────────────────────────
fn make_turn_metric_event(owner_keys: &Keys, agent_keys: &Keys) -> Event {
use buzz_core_pkg::agent_turn_metric::{
encrypt_agent_turn_metric, AgentTurnMetricPayload, TokenCounts,
};
let owner_pk = owner_keys.public_key().to_hex();
let payload = AgentTurnMetricPayload {
harness: "test-harness".to_string(),
model: Some("test-model".to_string()),
channel_id: None,
session_id: Some("sess-1".to_string()),
turn_id: Some("turn-1".to_string()),
turn_seq: Some(1),
timestamp: "2026-07-01T00:00:00Z".to_string(),
turn: Some(TokenCounts {
input_tokens: Some(100),
output_tokens: Some(50),
total_tokens: Some(150),
cost_usd: Some(0.001),
cache_read_tokens: None,
cache_write_tokens: None,
}),
cumulative: None,
delta_reliable: true,
stop_reason: None,
};
let ciphertext =
encrypt_agent_turn_metric(agent_keys, &owner_keys.public_key(), &payload).unwrap();
let tags = vec![
Tag::parse(["p", &owner_pk]).unwrap(),
Tag::parse(["agent", &agent_keys.public_key().to_hex()]).unwrap(),
];
EventBuilder::new(Kind::Custom(44200), &ciphertext)
.tags(tags)
.sign_with_keys(agent_keys)
.unwrap()
}
/// A kind-44200 event with `owner_p` scope must route to the persistent
/// (relay-query) path, NOT the ephemeral path.
#[test]
fn test_owner_p_44200_routes_to_persistent_path() {
let conn = in_memory();
let owner_keys = Keys::generate();
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
// Subscription for kind 44200 under owner_p.
add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[44200]");
let ev = make_turn_metric_event(&owner_keys, &agent_keys);
let cand = candidate(&ev, ScopeType::OwnerP, &owner_pk);
let plan = plan_archive(vec![cand], &owner_pk, relay_url, &conn).unwrap();
// Must be in persistent buckets, NOT ephemeral list.
assert_eq!(plan.buckets.len(), 1, "kind-44200 must land in a bucket");
assert_eq!(
plan.ephemeral.len(),
0,
"kind-44200 must NOT be on the ephemeral path"
);
assert_eq!(
plan.buckets[0].scope_type_str, "owner_p",
"bucket scope_type must be owner_p"
);
}
/// A kind-24200 event with `owner_p` scope must still route to ephemeral.
#[test]
fn test_owner_p_24200_still_routes_to_ephemeral() {
let conn = in_memory();
let owner_keys = Keys::generate();
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
let cand = candidate(&ev, ScopeType::OwnerP, &owner_pk);
let plan = plan_archive(vec![cand], &owner_pk, relay_url, &conn).unwrap();
assert_eq!(
plan.buckets.len(),
0,
"kind-24200 must NOT land in a bucket"
);
assert_eq!(
plan.ephemeral.len(),
1,
"kind-24200 must be on the ephemeral path"
);
}
/// Decrypt success: plaintext payload JSON is stored, not raw ciphertext.
#[test]
fn test_turn_metric_decrypt_success_stores_plaintext() {
let conn = in_memory();
let owner_keys = Keys::generate();
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[44200]");
let ev = make_turn_metric_event(&owner_keys, &agent_keys);
let cand = candidate(&ev, ScopeType::OwnerP, &owner_pk);
let result = run_batch_sync_with_keys(
vec![cand],
&owner_pk,
relay_url,
&conn,
vec![ev.clone()],
&owner_keys,
);
assert_eq!(result.persisted, 1, "event must be persisted");
assert_eq!(result.dropped, 0, "no drops on successful decrypt");
// The stored raw_json must be plaintext JSON, not NIP-44 ciphertext.
let raw_json: String = conn
.query_row("SELECT raw_json FROM archived_events", [], |r| r.get(0))
.unwrap();
// Plaintext JSON should be a valid object with "harness" key.
let parsed: serde_json::Value =
serde_json::from_str(&raw_json).expect("stored raw_json must be valid JSON");
assert_eq!(
parsed["harness"], "test-harness",
"stored plaintext must decode to AgentTurnMetricPayload"
);
// Sanity: must NOT be the original NIP-44 ciphertext (which is not JSON).
assert_ne!(
raw_json, ev.content,
"stored content must differ from original ciphertext"
);
}
/// Decrypt fail: event is dropped, nothing written to the store (fail-closed).
#[test]
fn test_turn_metric_decrypt_fail_drops_fail_closed() {
let conn = in_memory();
let owner_keys = Keys::generate();
let wrong_keys = Keys::generate(); // wrong owner key — decrypt will fail
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
// Register subscription under owner_pk so the event passes plan-phase,
// but use `wrong_keys` in commit so decrypt fails.
add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[44200]");
let ev = make_turn_metric_event(&owner_keys, &agent_keys);
let cand = candidate(&ev, ScopeType::OwnerP, &owner_pk);
let result = run_batch_sync_with_keys(
vec![cand],
&owner_pk,
relay_url,
&conn,
vec![ev.clone()],
&wrong_keys, // wrong key → decrypt fails
);
assert_eq!(
result.persisted, 0,
"decrypt failure must not persist the event"
);
assert_eq!(result.dropped, 1, "decrypt failure must count as dropped");
let event_count: i64 = conn
.query_row("SELECT COUNT(*) FROM archived_events", [], |r| r.get(0))
.unwrap();
assert_eq!(
event_count, 0,
"no rows must be written to archived_events on decrypt failure"
);
}
// ── Real-relay integration tests ──────────────────────────────────────────
//
// Gated on `#[cfg(not(target_os = "windows"))]` because `build_app_state()`
@@ -1202,12 +1440,14 @@ mod tests {
// Phase 3: persist (sync). Fresh connection, same file.
let conn = store::open_archive_db(db_path).expect("open archive db for commit");
let owner_keys = state.keys.lock().unwrap().clone();
commit_archive(
bucket_results,
plan.ephemeral,
plan.pre_dropped,
&identity_pk,
&relay_url,
&owner_keys,
0,
&conn,
)

View File

@@ -124,7 +124,15 @@ pub(super) fn plan_archive(
continue;
}
if cand.matched_scope.scope_type.is_ephemeral() {
// owner_p scope splits by kind:
// kind 24200 (observer frames) → ephemeral path (relay never stores them).
// kind 44200 (turn metrics) → persistent path (relay stores, #p-gated).
// Any other kind under owner_p follows the same ephemeral path as 24200
// (conservative default for unknowns).
let is_ephemeral = cand.matched_scope.scope_type.is_ephemeral()
&& raw_kind != super::KIND_AGENT_TURN_METRIC as u64;
if is_ephemeral {
ephemeral.push(Parsed {
event,
raw_json: cand.raw_event_json,
@@ -188,6 +196,11 @@ pub(super) fn plan_archive(
"#e": [&scope_value],
"kinds": allowed_kinds,
}),
"owner_p" => serde_json::json!({
"ids": ids,
"#p": [&scope_value],
"kinds": allowed_kinds,
}),
_ => {
pre_dropped += group.len() as u32;
continue;
@@ -249,6 +262,7 @@ pub(super) fn commit_archive(
pre_dropped: u32,
identity_pk: &str,
relay_url: &str,
owner_keys: &nostr::Keys,
now: i64,
conn: &Connection,
) -> Result<ArchiveBatchResult, String> {
@@ -257,16 +271,19 @@ pub(super) fn commit_archive(
// Collect writes; count drops first, then execute inside a single
// transaction so event and scope rows are always committed atomically.
struct WriteRow<'a> {
//
// raw_json is owned so kind-44200 rows can store decrypted plaintext
// instead of the original NIP-44 ciphertext.
struct WriteRow {
eid: String,
kind: i64,
pubkey: String,
created_at: i64,
raw_json: &'a str,
scope_type: &'a str,
scope_value: &'a str,
raw_json: String,
scope_type: String,
scope_value: String,
}
let mut writes: Vec<WriteRow<'_>> = Vec::new();
let mut writes: Vec<WriteRow> = Vec::new();
// ── Persistent path ──────────────────────────────────────────────────────
for result in &bucket_results {
@@ -293,7 +310,33 @@ pub(super) fn commit_archive(
continue;
}
// The relay returning this event for {ids, #h/#e, kinds} IS the
// For kind-44200 (agent turn metrics): decrypt at ingest and store
// the plaintext payload JSON so token-usage calculators can read
// the archive without needing the owner key. Fail-closed: if
// decrypt fails for any reason, drop the event — never store
// ciphertext or partial output.
let stored_json =
if p.event.kind.as_u16() as u64 == super::KIND_AGENT_TURN_METRIC as u64 {
match buzz_core_pkg::agent_turn_metric::decrypt_agent_turn_metric(
owner_keys, &p.event,
) {
Ok(payload) => match serde_json::to_string(&payload) {
Ok(json) => json,
Err(_) => {
dropped += 1;
continue;
}
},
Err(_) => {
dropped += 1;
continue;
}
}
} else {
p.raw_json.clone()
};
// The relay returning this event for {ids, #h/#e/#p, kinds} IS the
// proof of scope membership. Use scope_value directly; no local
// tag re-derivation (which would incorrectly drop h-less events
// matched via the relay's StoredEvent.channel_id fallback).
@@ -302,9 +345,9 @@ pub(super) fn commit_archive(
kind: p.event.kind.as_u16() as i64,
pubkey: p.event.pubkey.to_hex(),
created_at: p.event.created_at.as_secs() as i64,
raw_json: &p.raw_json,
scope_type: &result.scope_type_str,
scope_value: &result.scope_value,
raw_json: stored_json,
scope_type: result.scope_type_str.clone(),
scope_value: result.scope_value.clone(),
});
}
}
@@ -343,7 +386,7 @@ pub(super) fn commit_archive(
w.kind,
&w.pubkey,
w.created_at,
w.raw_json,
&w.raw_json,
now,
)?;
store::upsert_event_scope(
@@ -351,8 +394,8 @@ pub(super) fn commit_archive(
identity_pk,
relay_url,
&w.eid,
w.scope_type,
w.scope_value,
&w.scope_type,
&w.scope_value,
now,
)?;
persisted += 1;

View File

@@ -195,6 +195,85 @@ pub fn has_save_subscription(
Ok(count > 0)
}
/// Atomically merge `new_kind` into the `owner_p` save subscription for the
/// given identity + relay + scope_value.
///
/// Reads the current `kinds` array, unions in `new_kind`, and writes back —
/// all inside a single `BEGIN IMMEDIATE` SQLite transaction. `IMMEDIATE`
/// acquires the write lock at `BEGIN` (before the `SELECT`), so a second
/// concurrent caller blocks on `busy_timeout` (5000 ms, set by
/// `open_archive_db`) and then reads the first caller's committed row. This
/// guarantees the union is complete; a `DEFERRED` transaction would let two
/// concurrent callers both read the empty snapshot and produce a
/// `BUSY_SNAPSHOT` failure on the losing writer.
///
/// If no row exists yet it is created with `[new_kind]`. `now` is used as
/// `created_at` only on insert (the conflict clause never updates it).
pub fn merge_owner_p_kinds(
conn: &Connection,
identity_pubkey: &str,
relay_url: &str,
scope_value: &str,
new_kind: u32,
now: i64,
) -> Result<(), String> {
// BEGIN IMMEDIATE takes the write lock before the SELECT so concurrent
// callers serialize here rather than racing to a BUSY_SNAPSHOT error.
conn.execute_batch("BEGIN IMMEDIATE")
.map_err(|e| format!("merge_owner_p_kinds begin immediate: {e}"))?;
let result = (|| -> Result<(), String> {
// Read the current kinds, if any.
let existing_json: Option<String> = conn
.query_row(
"SELECT kinds FROM save_subscriptions
WHERE identity_pubkey = ?1
AND relay_url = ?2
AND scope_type = 'owner_p'
AND scope_value = ?3",
params![identity_pubkey, relay_url, scope_value],
|row| row.get::<_, String>(0),
)
.optional()
.map_err(|e| format!("merge_owner_p_kinds read: {e}"))?;
// Parse, union, re-serialize.
let mut kinds: Vec<u32> = match existing_json {
Some(ref json) => serde_json::from_str(json).unwrap_or_default(),
None => vec![],
};
if !kinds.contains(&new_kind) {
kinds.push(new_kind);
}
let kinds_json = serde_json::to_string(&kinds)
.map_err(|e| format!("merge_owner_p_kinds serialize: {e}"))?;
// Upsert — INSERT on first call, UPDATE kinds on subsequent.
conn.execute(
"INSERT INTO save_subscriptions
(identity_pubkey, relay_url, scope_type, scope_value, kinds, created_at)
VALUES (?1, ?2, 'owner_p', ?3, ?4, ?5)
ON CONFLICT (identity_pubkey, relay_url, scope_type, scope_value)
DO UPDATE SET kinds = excluded.kinds",
params![identity_pubkey, relay_url, scope_value, kinds_json, now],
)
.map_err(|e| format!("merge_owner_p_kinds upsert: {e}"))?;
Ok(())
})();
if result.is_ok() {
conn.execute_batch("COMMIT")
.map_err(|e| format!("merge_owner_p_kinds commit: {e}"))?;
} else {
// Best-effort rollback; ignore the rollback error to surface the
// original error to the caller.
let _ = conn.execute_batch("ROLLBACK");
}
result
}
/// Return the `kinds` JSON string for a matching save subscription, or `None`
/// if no subscription exists.
pub fn get_subscription_kinds(
@@ -524,6 +603,130 @@ mod tests {
assert!(!has_save_subscription(&conn, "pk", "wss://r", "owner_p", "other").unwrap());
}
// ── merge_owner_p_kinds ──────────────────────────────────────────────────
#[test]
fn test_merge_owner_p_kinds_creates_row_when_none_exists() {
let conn = in_memory();
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
assert_eq!(subs.len(), 1);
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
assert_eq!(kinds, [24200]);
}
#[test]
fn test_merge_owner_p_kinds_adds_new_kind_to_existing_row() {
let conn = in_memory();
// Seed with 24200 first.
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
// Now merge 44200 in — must produce [24200, 44200].
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap();
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
assert_eq!(subs.len(), 1);
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
assert!(kinds.contains(&24200), "must still contain 24200");
assert!(kinds.contains(&44200), "must now contain 44200");
assert_eq!(kinds.len(), 2, "no duplicates");
}
#[test]
fn test_merge_owner_p_kinds_idempotent_on_existing_kind() {
let conn = in_memory();
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 1).unwrap();
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap();
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
assert_eq!(kinds, [44200], "no duplicates after idempotent call");
}
/// Two-connection WAL regression test for the BEGIN IMMEDIATE fix.
///
/// Opens TWO separate connections to the same WAL file (mirroring the
/// real scenario where the observer-archive seed hook and the
/// metric-archive seed hook each open their own connection via
/// `open_archive_db`). Both threads call `merge_owner_p_kinds`
/// concurrently from an empty row. With `BEGIN IMMEDIATE` the losing
/// thread blocks on `busy_timeout` until the winner commits, then
/// reads the committed row and merges its kind in. Both calls must
/// resolve `Ok` and the final row must contain exactly `[24200, 44200]`.
///
/// A `DEFERRED` transaction would produce `SQLITE_BUSY_SNAPSHOT` on the
/// loser (not retried by busy_timeout), causing one kind to be silently
/// dropped. This test fails in <10 ms if the IMMEDIATE guard is removed.
#[test]
fn test_merge_owner_p_kinds_two_conn_wal_both_kinds_survive() {
use std::sync::{Arc, Barrier};
use std::thread;
use tempfile::NamedTempFile;
// A real file DB is required for WAL mode (in-memory dbs don't support
// shared-cache WAL across multiple connections in the same process).
let db_file = NamedTempFile::new().unwrap();
let db_path = db_file.path().to_path_buf();
// Initialise schema on the file DB via conn-A so both threads see it.
let init_conn = open_archive_db(&db_path).unwrap();
drop(init_conn);
// Barrier ensures both threads are inside `merge_owner_p_kinds` before
// either one issues `BEGIN IMMEDIATE`, maximising the race window.
let barrier = Arc::new(Barrier::new(2));
let path_a = db_path.clone();
let path_b = db_path.clone();
let bar_a = Arc::clone(&barrier);
let bar_b = Arc::clone(&barrier);
let handle_observer = thread::spawn(move || {
let conn = open_archive_db(&path_a).unwrap();
bar_a.wait(); // sync: both threads ready
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1)
});
let handle_metric = thread::spawn(move || {
let conn = open_archive_db(&path_b).unwrap();
bar_b.wait(); // sync: both threads ready
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2)
});
let res_observer = handle_observer.join().expect("observer thread panicked");
let res_metric = handle_metric.join().expect("metric thread panicked");
assert!(
res_observer.is_ok(),
"observer seed must succeed: {:?}",
res_observer
);
assert!(
res_metric.is_ok(),
"metric seed must succeed: {:?}",
res_metric
);
// Verify the final row contains both kinds.
let verify_conn = open_archive_db(&db_path).unwrap();
let subs = list_save_subscriptions(&verify_conn, "pk", "wss://r").unwrap();
assert_eq!(subs.len(), 1, "exactly one owner_p row");
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
assert!(
kinds.contains(&24200),
"observer kind 24200 must survive concurrent metric seed; got {:?}",
kinds
);
assert!(
kinds.contains(&44200),
"metric kind 44200 must be present after concurrent seed; got {:?}",
kinds
);
assert_eq!(
kinds.len(),
2,
"exactly two kinds, no duplicates; got {:?}",
kinds
);
}
// ── Archived events ──────────────────────────────────────────────────────
#[test]

View File

@@ -0,0 +1,35 @@
//! Build-time flag for agent-turn-metric archive default.
//!
//! When `BUZZ_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT` is set at build time
//! (internal builds), `agent_metric_archive_default_enabled()` returns `true`
//! and the frontend auto-seeds an `owner_p` save subscription for kind 44200
//! (agent turn metrics) on first run for the current identity.
//!
//! OSS builds (env var unset) return `false` — no auto-seeding, user opts in
//! manually via the Local Archive settings card.
/// Returns `true` when an internal build has agent-turn-metric archive
/// default-on.
///
/// The frontend calls this once at startup to decide whether to seed the
/// `owner_p` [44200] save subscription. The result is stable for the lifetime
/// of the binary — it is baked at compile time.
#[tauri::command]
pub fn agent_metric_archive_default_enabled() -> bool {
option_env!("BUZZ_DESKTOP_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT").is_some()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_agent_metric_archive_default_enabled_returns_false_in_oss_build() {
// In a standard OSS/test build (no BUZZ_DESKTOP_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT
// baked in), this must return false.
assert!(
!agent_metric_archive_default_enabled(),
"expected false in OSS/test build"
);
}
}

View File

@@ -1,5 +1,6 @@
mod agent_config;
mod agent_discovery;
mod agent_metric_archive;
mod agent_models;
mod agent_providers;
mod agent_settings;
@@ -42,6 +43,7 @@ mod workspace;
pub use agent_config::*;
pub use agent_discovery::*;
pub use agent_metric_archive::*;
pub use agent_models::*;
pub use agent_providers::*;
pub use agent_settings::*;

View File

@@ -616,8 +616,10 @@ pub fn run() {
relay_reconnect_hook,
relay_reconnect_hook_configured,
observer_archive_default_enabled,
agent_metric_archive_default_enabled,
archive::archive_events,
archive::create_save_subscription,
archive::merge_save_subscription_kinds,
archive::list_save_subscriptions,
archive::delete_save_subscription,
archive::read_archived_events,

View File

@@ -51,6 +51,7 @@ import {
import { useWorkspaceEmojiLiveUpdates } from "@/features/custom-emoji/hooks";
import { useArchiveSync } from "@/features/local-archive/archiveSyncManager";
import { useObserverArchiveSeed } from "@/features/local-archive/useObserverArchiveSeed";
import { useAgentMetricArchiveSeed } from "@/features/local-archive/useAgentMetricArchiveSeed";
import { useProfileQuery } from "@/features/profile/hooks";
import {
DEFAULT_SETTINGS_SECTION,
@@ -151,6 +152,7 @@ export function AppShell() {
useAgentsDataRefresh();
useArchiveSync();
useObserverArchiveSeed(identityQuery.data?.pubkey);
useAgentMetricArchiveSeed(identityQuery.data?.pubkey);
const profileQuery = useProfileQuery();
const deferredPubkey = startupReady ? identityQuery.data?.pubkey : undefined;
useRelayAutoHeal();

View File

@@ -0,0 +1,71 @@
/**
* Persists whether the user has made an explicit choice about the
* agent-turn-metric archive default-on feature.
*
* The key is identity-scoped so toggling off on one identity doesn't suppress
* the default-on for another identity. The value is:
* "1" → user explicitly enabled (or accepted the default)
* "0" → user explicitly disabled
* null → no explicit choice yet (default-on seeding may still fire)
*
* Device-level localStorage — intentionally not reset on workspace switch
* (the archive subscription itself is identity-scoped in SQLite; this flag
* is just the UI gate that prevents re-seeding after an explicit opt-out).
*/
const KEY_PREFIX = "buzz:agent-metric-archive-default-seeded";
function storageKey(identityPubkey: string): string {
return `${KEY_PREFIX}:${identityPubkey}`;
}
/**
* Returns `true` if the user has already made an explicit choice for this
* identity (either opted in or opted out). When `false`, the seeding path
* may fire.
*/
export function hasExplicitAgentMetricArchiveChoice(
identityPubkey: string,
): boolean {
if (typeof window === "undefined") return true; // SSR/test: treat as set
try {
return window.localStorage.getItem(storageKey(identityPubkey)) !== null;
} catch {
return true; // storage error → treat as set, never auto-seed
}
}
/**
* Mark that the user has made an explicit choice for this identity.
* `enabled` should reflect whether the `owner_p` subscription exists after
* the action (true = seeded/enabled, false = opted out).
*/
export function setExplicitAgentMetricArchiveChoice(
identityPubkey: string,
enabled: boolean,
): void {
if (typeof window === "undefined") return;
try {
window.localStorage.setItem(
storageKey(identityPubkey),
enabled ? "1" : "0",
);
} catch {
// Best-effort — the seeding guard will re-fire on next startup if storage
// is unavailable, but that is safe (create_save_subscription is idempotent).
}
}
/**
* Clear the explicit choice for this identity (for testing / reset flows).
*/
export function clearExplicitAgentMetricArchiveChoice(
identityPubkey: string,
): void {
if (typeof window === "undefined") return;
try {
window.localStorage.removeItem(storageKey(identityPubkey));
} catch {
// ignore
}
}

View File

@@ -9,7 +9,10 @@ import {
type SaveSubscription,
type ScopeType,
} from "@/shared/api/tauriArchive";
import { KIND_AGENT_OBSERVER_FRAME } from "@/shared/constants/kinds";
import {
KIND_AGENT_OBSERVER_FRAME,
KIND_AGENT_TURN_METRIC,
} from "@/shared/constants/kinds";
import { useChannelsQuery } from "@/features/channels/hooks";
import { useIdentityQuery } from "@/shared/api/hooks";
import { Button } from "@/shared/ui/button";
@@ -21,6 +24,7 @@ import {
} from "@/features/settings/ui/SettingsOptionGroup";
import { SettingsSectionHeader } from "@/features/settings/ui/SettingsSectionHeader";
import { setExplicitObserverArchiveChoice } from "../observerArchivePreference";
import { setExplicitAgentMetricArchiveChoice } from "../agentMetricArchivePreference";
import {
buildSubscriptionRequest,
@@ -42,6 +46,9 @@ function scopeLabel(
return channelNameById.get(sub.scopeValue) ?? sub.scopeValue;
}
if (sub.scopeType === "owner_p") {
if (sub.kinds.includes(KIND_AGENT_TURN_METRIC)) {
return "My agents' turn metrics";
}
return "My agent session frames";
}
return sub.scopeValue;
@@ -97,6 +104,50 @@ function ObserverArchiveSection({
);
}
// ── Agent-turn-metric archive section ────────────────────────────────────────
type AgentMetricSectionProps = {
enabled: boolean;
toggling: boolean;
onToggle: (checked: boolean) => void;
};
function AgentMetricArchiveSection({
enabled,
toggling,
onToggle,
}: AgentMetricSectionProps) {
return (
<div className="space-y-3" data-testid="local-archive-agent-metric-section">
<h3 className="text-sm font-medium">Agent turn metrics</h3>
<SettingsOptionGroup>
<SettingsOptionRow>
<div className="min-w-0 flex-1">
<label
className="text-sm font-medium"
htmlFor="local-archive-agent-metric-toggle"
>
Archive my agents' turn metrics
</label>
<p className="text-sm font-normal text-muted-foreground">
Saves kind {KIND_AGENT_TURN_METRIC} turn-metric events addressed
to your pubkey. Stored as plaintext in your local archive so
token-usage calculators can read them directly.
</p>
</div>
<Switch
checked={enabled}
data-testid="local-archive-agent-metric-toggle"
disabled={toggling}
id="local-archive-agent-metric-toggle"
onCheckedChange={onToggle}
/>
</SettingsOptionRow>
</SettingsOptionGroup>
</div>
);
}
// ── Add-subscription form ─────────────────────────────────────────────────────
type KindChecklistProps = {
@@ -331,6 +382,7 @@ export function LocalArchiveSettingsCard() {
const [deletingKey, setDeletingKey] = React.useState<string | null>(null);
const [isAddingOpen, setIsAddingOpen] = React.useState(false);
const [observerToggling, setObserverToggling] = React.useState(false);
const [metricToggling, setMetricToggling] = React.useState(false);
const pubkey = identityQuery.data?.pubkey ?? "";
@@ -381,24 +433,42 @@ export function LocalArchiveSettingsCard() {
[reload],
);
const observerEnabled = subs.some((s) => s.scopeType === "owner_p");
const observerEnabled = subs.some(
(s) =>
s.scopeType === "owner_p" && s.kinds.includes(KIND_AGENT_OBSERVER_FRAME),
);
const metricEnabled = subs.some(
(s) =>
s.scopeType === "owner_p" && s.kinds.includes(KIND_AGENT_TURN_METRIC),
);
const handleObserverToggle = React.useCallback(
async (checked: boolean) => {
if (!pubkey) return;
setObserverToggling(true);
try {
if (checked) {
await createSaveSubscription("owner_p", pubkey, [
KIND_AGENT_OBSERVER_FRAME,
]);
setExplicitObserverArchiveChoice(pubkey, true);
toast.success("Observer feed archive enabled.");
// The owner_p row is keyed by (scope_type, scope_value) — both observer
// (24200) and metric (44200) share the same row. Merge kinds atomically:
// read current kinds, add or remove 24200, upsert the result.
const currentKinds =
subs
.find((s) => s.scopeType === "owner_p" && s.scopeValue === pubkey)
?.kinds.filter((k) => k !== KIND_AGENT_OBSERVER_FRAME) ?? [];
const nextKinds = checked
? [...currentKinds, KIND_AGENT_OBSERVER_FRAME]
: currentKinds;
if (nextKinds.length > 0) {
await createSaveSubscription("owner_p", pubkey, nextKinds);
} else {
await deleteSaveSubscription("owner_p", pubkey);
setExplicitObserverArchiveChoice(pubkey, false);
toast.success("Observer feed archive disabled.");
}
setExplicitObserverArchiveChoice(pubkey, checked);
toast.success(
checked
? "Observer feed archive enabled."
: "Observer feed archive disabled.",
);
await reload();
} catch (err) {
toast.error(
@@ -410,10 +480,51 @@ export function LocalArchiveSettingsCard() {
setObserverToggling(false);
}
},
[pubkey, reload],
[pubkey, subs, reload],
);
// Non-observer subscriptions shown in the active-subscriptions list.
const handleMetricToggle = React.useCallback(
async (checked: boolean) => {
if (!pubkey) return;
setMetricToggling(true);
try {
// Same row as observer — merge 44200 in or out.
const currentKinds =
subs
.find((s) => s.scopeType === "owner_p" && s.scopeValue === pubkey)
?.kinds.filter((k) => k !== KIND_AGENT_TURN_METRIC) ?? [];
const nextKinds = checked
? [...currentKinds, KIND_AGENT_TURN_METRIC]
: currentKinds;
if (nextKinds.length > 0) {
await createSaveSubscription("owner_p", pubkey, nextKinds);
} else {
await deleteSaveSubscription("owner_p", pubkey);
}
setExplicitAgentMetricArchiveChoice(pubkey, checked);
toast.success(
checked
? "Agent turn metric archive enabled."
: "Agent turn metric archive disabled.",
);
await reload();
} catch (err) {
toast.error(
err instanceof Error
? err.message
: "Failed to update agent metric archive.",
);
} finally {
setMetricToggling(false);
}
},
[pubkey, subs, reload],
);
// Non-owner_p subscriptions shown in the active-subscriptions list.
// observer (24200) and metric (44200) owner_p subs each have their own
// dedicated section above.
const channelSubs = subs.filter((s) => s.scopeType !== "owner_p");
return (
@@ -431,6 +542,13 @@ export function LocalArchiveSettingsCard() {
toggling={observerToggling}
/>
{/* Agent-turn-metric archive — dedicated first-class section */}
<AgentMetricArchiveSection
enabled={metricEnabled}
onToggle={(checked) => void handleMetricToggle(checked)}
toggling={metricToggling}
/>
{/* Channel subscriptions */}
<div className="space-y-3" data-testid="local-archive-subscriptions">
<h3 className="text-sm font-medium">

View File

@@ -0,0 +1,222 @@
/**
* Tests for useAgentMetricArchiveSeed seeding logic.
*
* Mirrors the pattern in useObserverArchiveSeed.test.mjs — drives the async
* seed logic via the deps-injection interface, no React required.
*/
import assert from "node:assert/strict";
import test from "node:test";
// ── Fake deps factory ────────────────────────────────────────────────────────
function makeDeps({
defaultOn = false,
hasExplicitChoice = false,
mergeShouldFail = false,
} = {}) {
const calls = { mergeSaveSubscriptionKinds: [], setExplicitChoice: [] };
return {
calls,
agentMetricArchiveDefaultEnabled: async () => defaultOn,
mergeSaveSubscriptionKinds: async (kind) => {
if (mergeShouldFail) throw new Error("merge failed");
calls.mergeSaveSubscriptionKinds.push({ kind });
},
hasExplicitChoice: (_pubkey) => hasExplicitChoice,
setExplicitChoice: (pubkey, enabled) => {
calls.setExplicitChoice.push({ pubkey, enabled });
},
};
}
// Minimal re-implementation of the seeding logic from useAgentMetricArchiveSeed.ts.
// Kept in sync with the source by structural mirroring.
const KIND_AGENT_TURN_METRIC = 44200;
async function runSeed(pubkey, deps) {
if (!pubkey) return;
if (deps.hasExplicitChoice(pubkey)) return;
let defaultOn;
try {
defaultOn = await deps.agentMetricArchiveDefaultEnabled();
} catch {
return;
}
if (!defaultOn) return;
try {
await deps.mergeSaveSubscriptionKinds(KIND_AGENT_TURN_METRIC);
} catch {
return; // transient failure — do NOT set explicit choice
}
deps.setExplicitChoice(pubkey, true);
}
// ── Tests ────────────────────────────────────────────────────────────────────
test("test_internal_build_unset_seeds_owner_p_subscription", async () => {
const deps = makeDeps({ defaultOn: true, hasExplicitChoice: false });
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.mergeSaveSubscriptionKinds.length,
1,
"should call mergeSaveSubscriptionKinds once",
);
const call = deps.calls.mergeSaveSubscriptionKinds[0];
assert.equal(call.kind, 44200);
});
test("test_internal_build_unset_persists_explicit_choice_after_seed", async () => {
const deps = makeDeps({ defaultOn: true, hasExplicitChoice: false });
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.setExplicitChoice.length,
1,
"should persist explicit choice after successful seed",
);
assert.equal(deps.calls.setExplicitChoice[0].pubkey, "pubkey123");
assert.equal(deps.calls.setExplicitChoice[0].enabled, true);
});
test("test_explicit_choice_set_does_not_reseed", async () => {
const deps = makeDeps({ defaultOn: true, hasExplicitChoice: true });
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.mergeSaveSubscriptionKinds.length,
0,
"should not call mergeSaveSubscriptionKinds when explicit choice is already set",
);
assert.equal(
deps.calls.setExplicitChoice.length,
0,
"should not update explicit choice when already set",
);
});
test("test_oss_build_does_not_seed", async () => {
const deps = makeDeps({ defaultOn: false, hasExplicitChoice: false });
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.mergeSaveSubscriptionKinds.length,
0,
"should not call mergeSaveSubscriptionKinds in OSS build",
);
assert.equal(
deps.calls.setExplicitChoice.length,
0,
"should not persist explicit choice in OSS build",
);
});
test("test_merge_failure_does_not_persist_explicit_choice", async () => {
const deps = makeDeps({
defaultOn: true,
hasExplicitChoice: false,
mergeShouldFail: true,
});
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.setExplicitChoice.length,
0,
"should NOT persist explicit choice after a transient merge failure",
);
});
test("test_empty_pubkey_does_nothing", async () => {
const deps = makeDeps({ defaultOn: true, hasExplicitChoice: false });
await runSeed("", deps);
assert.equal(deps.calls.mergeSaveSubscriptionKinds.length, 0);
assert.equal(deps.calls.setExplicitChoice.length, 0);
});
test("test_undefined_pubkey_does_nothing", async () => {
const deps = makeDeps({ defaultOn: true, hasExplicitChoice: false });
await runSeed(undefined, deps);
assert.equal(deps.calls.mergeSaveSubscriptionKinds.length, 0);
assert.equal(deps.calls.setExplicitChoice.length, 0);
});
// ── Concurrent-interleave test ───────────────────────────────────────────────
//
// Verifies the scenario Paul identified: on an internal-build first run with
// both flags on and no prior owner_p row, the observer and metric seeds race.
// With the old TS-side list+merge+create pattern the interleave could be:
//
// 1. observer seed: await list() → []
// 2. metric seed: await list() → [] (row not yet written)
// 3. observer writes [24200]
// 4. metric writes [44200] → clobbers 24200
//
// The new pattern delegates the merge to Rust under a single SQLite tx.
// Here we model that by tracking a shared "db state" and verifying that
// running both seeds concurrently (Promise.all) leaves both kinds present.
test("test_concurrent_seeds_both_kinds_survive", async () => {
// Shared in-memory "db" — the atomic merge impl would serialize via SQLite
// write lock; here we simulate by letting calls accumulate and checking
// the final state.
const db = new Set(); // kinds present after all merges
function makeConcurrentDeps(defaultOn = true) {
return {
agentMetricArchiveDefaultEnabled: async () => defaultOn,
// Simulates the atomic merge: each call simply adds its kind to the set,
// regardless of what was there before (atomicity guarantee).
mergeSaveSubscriptionKinds: async (kind) => {
db.add(kind);
},
hasExplicitChoice: (_pubkey) => false,
setExplicitChoice: () => {},
};
}
const observerDeps = makeConcurrentDeps();
const metricDeps = makeConcurrentDeps();
// Replace the kind constant in the observer runSeed equivalent with 24200.
async function runObserverSeed(pubkey, deps) {
if (!pubkey) return;
if (deps.hasExplicitChoice(pubkey)) return;
let defaultOn;
try {
defaultOn = await deps.agentMetricArchiveDefaultEnabled();
} catch {
return;
}
if (!defaultOn) return;
try {
await deps.mergeSaveSubscriptionKinds(24200);
} catch {
return;
}
deps.setExplicitChoice(pubkey, true);
}
// Run both seeds concurrently — no await between them.
await Promise.all([
runObserverSeed("pubkey123", observerDeps),
runSeed("pubkey123", metricDeps),
]);
assert.ok(
db.has(24200),
"observer kind 24200 must be present after concurrent seeds",
);
assert.ok(
db.has(44200),
"metric kind 44200 must be present after concurrent seeds",
);
assert.equal(db.size, 2, "exactly two kinds, no extras");
});

View File

@@ -0,0 +1,114 @@
/**
* First-run seeding for agent-turn-metric archive.
*
* When an internal build has `BUZZ_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT` set and
* the current identity has not yet made an explicit choice, this hook
* auto-creates an `owner_p` save subscription including kind 44200 agent turn
* metrics, scoped to the current identity's pubkey.
*
* Uses `mergeSaveSubscriptionKinds` (atomic DB-side merge) so a concurrently
* running observer seed (24200) cannot clobber this kind — the union happens
* under a single SQLite transaction regardless of await ordering.
*
* OSS builds return `false` from `agent_metric_archive_default_enabled` →
* no-op. After any explicit user action (seeding or opt-out), the localStorage
* flag prevents re-seeding on subsequent starts.
*/
import * as React from "react";
import { KIND_AGENT_TURN_METRIC } from "@/shared/constants/kinds";
import {
mergeSaveSubscriptionKinds,
agentMetricArchiveDefaultEnabled,
} from "@/shared/api/tauriArchive";
import {
hasExplicitAgentMetricArchiveChoice,
setExplicitAgentMetricArchiveChoice,
} from "./agentMetricArchivePreference";
/**
* Deps interface for testing. Production callers pass nothing.
*/
export interface AgentMetricArchiveSeedDeps {
agentMetricArchiveDefaultEnabled: () => Promise<boolean>;
mergeSaveSubscriptionKinds: (kind: number) => Promise<void>;
hasExplicitChoice: (pubkey: string) => boolean;
setExplicitChoice: (pubkey: string, enabled: boolean) => void;
}
const defaultDeps: AgentMetricArchiveSeedDeps = {
agentMetricArchiveDefaultEnabled,
mergeSaveSubscriptionKinds,
hasExplicitChoice: hasExplicitAgentMetricArchiveChoice,
setExplicitChoice: setExplicitAgentMetricArchiveChoice,
};
/**
* Seed the agent-turn-metric archive subscription for `pubkey` once per
* identity per device on internal builds.
*
* @param pubkey - current identity pubkey. When undefined (identity not yet
* loaded), the hook waits until it becomes available.
* @param deps - optional dep-injection for tests.
*/
export function useAgentMetricArchiveSeed(
pubkey: string | undefined,
deps: AgentMetricArchiveSeedDeps = defaultDeps,
): void {
React.useEffect(() => {
if (!pubkey) return;
// Already made an explicit choice for this identity — never re-seed.
if (deps.hasExplicitChoice(pubkey)) return;
let cancelled = false;
async function maybeSeed(): Promise<void> {
// pubkey is checked above but TypeScript doesn't narrow across the async
// boundary — re-guard here so the call below is type-safe.
if (!pubkey) return;
let defaultOn: boolean;
try {
defaultOn = await deps.agentMetricArchiveDefaultEnabled();
} catch (err) {
console.warn("[useAgentMetricArchiveSeed] flag check failed:", err);
return;
}
if (cancelled) return;
if (!defaultOn) {
// OSS build (flag off): don't persist a choice — leave null so seeding
// can still fire if this identity later runs an internal build.
return;
}
// Internal build + no prior choice → auto-seed via atomic DB merge.
try {
await deps.mergeSaveSubscriptionKinds(KIND_AGENT_TURN_METRIC);
} catch (err) {
console.warn(
"[useAgentMetricArchiveSeed] mergeSaveSubscriptionKinds failed:",
err,
);
// Do NOT set the localStorage flag — a transient failure (relay
// unreachable, archive DB not yet initialized) should retry on next
// startup rather than permanently suppress seeding.
return;
}
if (cancelled) return;
// Persist the explicit choice so this never re-fires.
deps.setExplicitChoice(pubkey, true);
}
void maybeSeed();
return () => {
cancelled = true;
};
}, [pubkey, deps]);
}

View File

@@ -19,16 +19,16 @@ import test from "node:test";
function makeDeps({
defaultOn = false,
hasExplicitChoice = false,
createShouldFail = false,
mergeShouldFail = false,
} = {}) {
const calls = { createSaveSubscription: [], setExplicitChoice: [] };
const calls = { mergeSaveSubscriptionKinds: [], setExplicitChoice: [] };
return {
calls,
observerArchiveDefaultEnabled: async () => defaultOn,
createSaveSubscription: async (scopeType, scopeValue, kinds) => {
if (createShouldFail) throw new Error("create failed");
calls.createSaveSubscription.push({ scopeType, scopeValue, kinds });
mergeSaveSubscriptionKinds: async (kind) => {
if (mergeShouldFail) throw new Error("merge failed");
calls.mergeSaveSubscriptionKinds.push({ kind });
},
hasExplicitChoice: (_pubkey) => hasExplicitChoice,
setExplicitChoice: (pubkey, enabled) => {
@@ -40,6 +40,8 @@ function makeDeps({
// Minimal re-implementation of the seeding logic from useObserverArchiveSeed.ts.
// Kept in sync with the source by structural mirroring (not bytewise copy) —
// change the source and update this accordingly.
const KIND_AGENT_OBSERVER_FRAME = 24200;
async function runSeed(pubkey, deps) {
if (!pubkey) return;
if (deps.hasExplicitChoice(pubkey)) return;
@@ -54,7 +56,7 @@ async function runSeed(pubkey, deps) {
if (!defaultOn) return;
try {
await deps.createSaveSubscription("owner_p", pubkey, [24200]);
await deps.mergeSaveSubscriptionKinds(KIND_AGENT_OBSERVER_FRAME);
} catch {
return; // transient failure — do NOT set explicit choice
}
@@ -69,14 +71,12 @@ test("test_internal_build_unset_seeds_owner_p_subscription", async () => {
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.createSaveSubscription.length,
deps.calls.mergeSaveSubscriptionKinds.length,
1,
"should call createSaveSubscription once",
"should call mergeSaveSubscriptionKinds once",
);
const call = deps.calls.createSaveSubscription[0];
assert.equal(call.scopeType, "owner_p");
assert.equal(call.scopeValue, "pubkey123");
assert.deepEqual(call.kinds, [24200]);
const call = deps.calls.mergeSaveSubscriptionKinds[0];
assert.equal(call.kind, 24200);
});
test("test_internal_build_unset_persists_explicit_choice_after_seed", async () => {
@@ -97,9 +97,9 @@ test("test_explicit_choice_set_does_not_reseed", async () => {
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.createSaveSubscription.length,
deps.calls.mergeSaveSubscriptionKinds.length,
0,
"should not call createSaveSubscription when explicit choice is already set",
"should not call mergeSaveSubscriptionKinds when explicit choice is already set",
);
assert.equal(
deps.calls.setExplicitChoice.length,
@@ -113,9 +113,9 @@ test("test_oss_build_does_not_seed", async () => {
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.createSaveSubscription.length,
deps.calls.mergeSaveSubscriptionKinds.length,
0,
"should not call createSaveSubscription in OSS build",
"should not call mergeSaveSubscriptionKinds in OSS build",
);
assert.equal(
deps.calls.setExplicitChoice.length,
@@ -124,23 +124,18 @@ test("test_oss_build_does_not_seed", async () => {
);
});
test("test_create_failure_does_not_persist_explicit_choice", async () => {
test("test_merge_failure_does_not_persist_explicit_choice", async () => {
const deps = makeDeps({
defaultOn: true,
hasExplicitChoice: false,
createShouldFail: true,
mergeShouldFail: true,
});
await runSeed("pubkey123", deps);
assert.equal(
deps.calls.createSaveSubscription.length,
0,
"createSaveSubscription should throw (called internally but errors)",
);
assert.equal(
deps.calls.setExplicitChoice.length,
0,
"should NOT persist explicit choice after a transient create failure",
"should NOT persist explicit choice after a transient merge failure",
);
});
@@ -148,7 +143,7 @@ test("test_empty_pubkey_does_nothing", async () => {
const deps = makeDeps({ defaultOn: true, hasExplicitChoice: false });
await runSeed("", deps);
assert.equal(deps.calls.createSaveSubscription.length, 0);
assert.equal(deps.calls.mergeSaveSubscriptionKinds.length, 0);
assert.equal(deps.calls.setExplicitChoice.length, 0);
});
@@ -156,6 +151,6 @@ test("test_undefined_pubkey_does_nothing", async () => {
const deps = makeDeps({ defaultOn: true, hasExplicitChoice: false });
await runSeed(undefined, deps);
assert.equal(deps.calls.createSaveSubscription.length, 0);
assert.equal(deps.calls.mergeSaveSubscriptionKinds.length, 0);
assert.equal(deps.calls.setExplicitChoice.length, 0);
});

View File

@@ -2,9 +2,13 @@
* First-run seeding for observer-feed archive.
*
* When an internal build has `BUZZ_BUILD_OBSERVER_ARCHIVE_DEFAULT` set and the
* current identity has not yet made an explicit choice, this hook auto-creates
* an `owner_p` save subscription (kind 24200 observer frames, scoped to the
* current identity's pubkey).
* current identity has not yet made an explicit choice, this hook
* auto-creates an `owner_p` save subscription including kind 24200 observer
* frames, scoped to the current identity's pubkey.
*
* Uses `mergeSaveSubscriptionKinds` (atomic DB-side merge) so a concurrently
* running metric seed (44200) cannot clobber this kind — the union happens
* under a single SQLite transaction regardless of await ordering.
*
* OSS builds return `false` from `observer_archive_default_enabled` → no-op.
* After any explicit user action (seeding or opt-out), the localStorage flag
@@ -15,7 +19,7 @@ import * as React from "react";
import { KIND_AGENT_OBSERVER_FRAME } from "@/shared/constants/kinds";
import {
createSaveSubscription,
mergeSaveSubscriptionKinds,
observerArchiveDefaultEnabled,
} from "@/shared/api/tauriArchive";
import {
@@ -28,18 +32,14 @@ import {
*/
export interface ObserverArchiveSeedDeps {
observerArchiveDefaultEnabled: () => Promise<boolean>;
createSaveSubscription: (
scopeType: "owner_p",
scopeValue: string,
kinds: number[],
) => Promise<void>;
mergeSaveSubscriptionKinds: (kind: number) => Promise<void>;
hasExplicitChoice: (pubkey: string) => boolean;
setExplicitChoice: (pubkey: string, enabled: boolean) => void;
}
const defaultDeps: ObserverArchiveSeedDeps = {
observerArchiveDefaultEnabled,
createSaveSubscription,
mergeSaveSubscriptionKinds,
hasExplicitChoice: hasExplicitObserverArchiveChoice,
setExplicitChoice: setExplicitObserverArchiveChoice,
};
@@ -85,14 +85,12 @@ export function useObserverArchiveSeed(
return;
}
// Internal build + no prior choice → auto-seed.
// Internal build + no prior choice → auto-seed via atomic DB merge.
try {
await deps.createSaveSubscription("owner_p", pubkey, [
KIND_AGENT_OBSERVER_FRAME,
]);
await deps.mergeSaveSubscriptionKinds(KIND_AGENT_OBSERVER_FRAME);
} catch (err) {
console.warn(
"[useObserverArchiveSeed] createSaveSubscription failed:",
"[useObserverArchiveSeed] mergeSaveSubscriptionKinds failed:",
err,
);
// Do NOT set the localStorage flag — a transient failure (relay

View File

@@ -98,6 +98,34 @@ export async function observerArchiveDefaultEnabled(): Promise<boolean> {
return invokeTauri<boolean>("observer_archive_default_enabled");
}
/**
* Returns `true` when the build has agent-turn-metric archive default-on.
*
* Internal builds set `BUZZ_BUILD_AGENT_METRIC_ARCHIVE_DEFAULT` at build time;
* OSS builds never set it, so this returns `false`. The frontend calls this
* once at startup to decide whether to auto-seed an `owner_p` [44200]
* subscription.
*/
export async function agentMetricArchiveDefaultEnabled(): Promise<boolean> {
return invokeTauri<boolean>("agent_metric_archive_default_enabled");
}
/**
* Atomically merge `kind` into the `owner_p` save subscription for the
* current identity + relay.
*
* Performs a read-modify-write inside a single SQLite transaction on the Rust
* side, so concurrent callers (e.g. observer seed + metric seed racing on
* first run) cannot clobber each other's kind.
*
* Called by both `useObserverArchiveSeed` and `useAgentMetricArchiveSeed`
* instead of the former list → merge-in-TS → create pattern.
*/
export async function mergeSaveSubscriptionKinds(kind: number): Promise<void> {
await invokeTauri("merge_save_subscription_kinds", { kind });
notifySubscriptionChange();
}
/**
* Create a save subscription.
* Runs an access probe on the backend (channel membership, event readability).

View File

@@ -43,6 +43,7 @@ export const KIND_TEAM = 30176;
export const KIND_MANAGED_AGENT = 30177;
export const KIND_USER_STATUS = 30315;
export const KIND_AGENT_OBSERVER_FRAME = 24200;
export const KIND_AGENT_TURN_METRIC = 44200;
export const KIND_MESH_STATUS_REPORT = 24620;
export const KIND_MESH_CONNECT_REQUEST = 24621;
export const KIND_MESH_CALL_ME_NOW = 24622;