diff --git a/desktop/src-tauri/src/archive/causal_ledger.rs b/desktop/src-tauri/src/archive/causal_ledger.rs new file mode 100644 index 0000000000..9202249bec --- /dev/null +++ b/desktop/src-tauri/src/archive/causal_ledger.rs @@ -0,0 +1,301 @@ +use rusqlite::{params, Connection, OptionalExtension}; +use serde::Deserialize; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use tauri::State; + +use crate::app_state::AppState; + +use super::{identity_pubkey, now_secs, run_archive_db_task}; + +const GENESIS_HASH: &str = "0000000000000000000000000000000000000000000000000000000000000000"; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct LedgerEntryEnvelope { + sequence: u64, + previous_hash: String, + hash: String, + experiment: Value, +} + +fn experiment_id(experiment: &Value) -> Result<&str, String> { + experiment + .get("experimentId") + .and_then(Value::as_str) + .filter(|id| !id.trim().is_empty()) + .ok_or_else(|| "causal ledger experimentId is required".to_string()) +} + +fn canonical_json(value: &Value) -> Result { + match value { + Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => { + serde_json::to_string(value) + .map_err(|error| format!("failed to canonicalize causal ledger value: {error}")) + } + Value::Array(values) => { + let values = values + .iter() + .map(canonical_json) + .collect::, _>>()?; + Ok(format!("[{}]", values.join(","))) + } + Value::Object(entries) => { + let mut entries = entries.iter().collect::>(); + entries.sort_unstable_by_key(|(key, _)| *key); + let entries = entries + .into_iter() + .map(|(key, value)| { + let key = serde_json::to_string(key).map_err(|error| { + format!("failed to canonicalize causal ledger key: {error}") + })?; + Ok(format!("{key}:{}", canonical_json(value)?)) + }) + .collect::, String>>()?; + Ok(format!("{{{}}}", entries.join(","))) + } + } +} + +fn entry_hash(entry: &LedgerEntryEnvelope) -> Result { + let hash_input = serde_json::json!({ + "sequence": entry.sequence, + "previousHash": entry.previous_hash, + "experiment": entry.experiment, + }); + Ok(hex::encode(Sha256::digest( + canonical_json(&hash_input)?.as_bytes(), + ))) +} + +fn validate_entry(entry: &LedgerEntryEnvelope) -> Result<(), String> { + experiment_id(&entry.experiment)?; + if entry.sequence == 0 + || entry.hash.len() != 64 + || entry.previous_hash.len() != 64 + || !entry.hash.bytes().all(|byte| byte.is_ascii_hexdigit()) + || !entry + .previous_hash + .bytes() + .all(|byte| byte.is_ascii_hexdigit()) + { + return Err("invalid causal ledger chain fields".to_string()); + } + let expected = entry_hash(entry)?; + if entry.hash != expected { + return Err(format!( + "causal ledger integrity failure at sequence {}", + entry.sequence + )); + } + Ok(()) +} + +/// Return the current owner's immutable causal-ledger journal in chain order. +#[tauri::command] +pub async fn read_causal_ledger(state: State<'_, AppState>) -> Result, String> { + let identity = identity_pubkey(&state)?; + run_archive_db_task(move |conn| read_entries(conn, &identity)).await +} + +fn read_entries(conn: &Connection, identity: &str) -> Result, String> { + let mut statement = conn + .prepare( + "SELECT entry_json FROM causal_ledger_entries + WHERE identity_pubkey = ?1 ORDER BY sequence ASC", + ) + .map_err(|error| format!("failed to prepare causal ledger read: {error}"))?; + let rows = statement + .query_map(params![identity], |row| row.get::<_, String>(0)) + .map_err(|error| format!("failed to read causal ledger: {error}"))?; + let entries = rows + .collect::, _>>() + .map_err(|error| format!("failed to decode causal ledger row: {error}"))?; + let mut previous_hash = GENESIS_HASH.to_string(); + for (index, entry_json) in entries.iter().enumerate() { + let entry: LedgerEntryEnvelope = serde_json::from_str(entry_json) + .map_err(|error| format!("invalid stored causal ledger entry: {error}"))?; + validate_entry(&entry)?; + let expected_sequence = + u64::try_from(index).map_err(|_| "causal ledger sequence overflow".to_string())? + 1; + if entry.sequence != expected_sequence || entry.previous_hash != previous_hash { + return Err(format!( + "causal ledger integrity failure at sequence {}", + entry.sequence + )); + } + previous_hash = entry.hash; + } + Ok(entries) +} + +/// Transactionally append one owner-scoped hash-linked causal-ledger entry. +#[tauri::command] +pub async fn append_causal_ledger_entry( + state: State<'_, AppState>, + entry_json: String, +) -> Result<(), String> { + let identity = identity_pubkey(&state)?; + let entry: LedgerEntryEnvelope = serde_json::from_str(&entry_json) + .map_err(|error| format!("invalid causal ledger entry: {error}"))?; + validate_entry(&entry)?; + let recorded_at = now_secs(); + run_archive_db_task(move |conn| append_entry(conn, &identity, &entry_json, entry, recorded_at)) + .await +} + +fn append_entry( + conn: &Connection, + identity: &str, + entry_json: &str, + entry: LedgerEntryEnvelope, + recorded_at: i64, +) -> Result<(), String> { + validate_entry(&entry)?; + conn.execute_batch("BEGIN IMMEDIATE") + .map_err(|error| format!("failed to begin causal ledger append: {error}"))?; + let result = (|| { + let tail = conn + .query_row( + "SELECT sequence, hash FROM causal_ledger_entries + WHERE identity_pubkey = ?1 ORDER BY sequence DESC LIMIT 1", + params![identity], + |row| Ok((row.get::<_, u64>(0)?, row.get::<_, String>(1)?)), + ) + .optional() + .map_err(|error| format!("failed to read causal ledger tail: {error}"))?; + let expected_sequence = tail.as_ref().map_or(1, |(sequence, _)| sequence + 1); + let expected_previous = tail + .as_ref() + .map_or(GENESIS_HASH, |(_, hash)| hash.as_str()); + + if entry.sequence != expected_sequence || entry.previous_hash != expected_previous { + return Err(format!( + "causal ledger append conflict: expected sequence {expected_sequence}" + )); + } + conn + .execute( + "INSERT INTO causal_ledger_entries + (identity_pubkey, sequence, experiment_id, previous_hash, hash, entry_json, recorded_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + identity, + entry.sequence, + experiment_id(&entry.experiment)?, + entry.previous_hash, + entry.hash, + entry_json, + recorded_at + ], + ) + .map_err(|error| format!("failed to append causal ledger entry: {error}"))?; + Ok(()) + })(); + match result { + Ok(()) => conn + .execute_batch("COMMIT") + .map_err(|error| format!("failed to commit causal ledger append: {error}")), + Err(error) => { + let _ = conn.execute_batch("ROLLBACK"); + Err(error) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::archive::store::open_archive_db; + + fn envelope(sequence: u64, previous_hash: &str) -> (String, LedgerEntryEnvelope) { + let experiment_id = format!("experiment-{sequence}"); + let mut value = serde_json::json!({ + "sequence": sequence, + "previousHash": previous_hash, + "hash": "", + "experiment": { "experimentId": experiment_id } + }); + let mut parsed: LedgerEntryEnvelope = + serde_json::from_value(value.clone()).expect("test envelope should decode"); + parsed.hash = entry_hash(&parsed).expect("hash test entry"); + value["hash"] = Value::String(parsed.hash.clone()); + let json = value.to_string(); + (json, parsed) + } + + #[test] + fn persists_ten_thousand_owner_scoped_entries_on_disk_and_reopens() { + let directory = tempfile::tempdir().expect("temp directory"); + let path = directory.path().join("archive.db"); + let conn = open_archive_db(&path).expect("open archive database"); + let mut previous = GENESIS_HASH.to_string(); + for sequence in 1..=10_000 { + let (json, entry) = envelope(sequence, &previous); + previous = entry.hash.clone(); + append_entry(&conn, "owner-a", &json, entry, 0).expect("append entry"); + } + drop(conn); + + let reopened = open_archive_db(&path).expect("reopen archive database"); + assert_eq!( + read_entries(&reopened, "owner-a").expect("read").len(), + 10_000 + ); + assert!(read_entries(&reopened, "owner-b").expect("read").is_empty()); + } + + #[test] + fn rejects_a_non_contiguous_chain_without_writing_it() { + let conn = Connection::open_in_memory().expect("open in-memory database"); + conn.execute_batch(crate::archive::store::SCHEMA) + .expect("initialize schema"); + let (json, entry) = envelope(2, GENESIS_HASH); + assert!(append_entry(&conn, "owner", &json, entry, 0).is_err()); + assert!(read_entries(&conn, "owner").expect("read").is_empty()); + } + + #[test] + fn rejects_a_forged_hash_without_writing_it() { + let conn = Connection::open_in_memory().expect("open in-memory database"); + conn.execute_batch(crate::archive::store::SCHEMA) + .expect("initialize schema"); + let (json, mut entry) = envelope(1, GENESIS_HASH); + entry.hash = "f".repeat(64); + assert!(append_entry(&conn, "owner", &json, entry, 0).is_err()); + assert!(read_entries(&conn, "owner").expect("read").is_empty()); + } + + #[test] + fn rejects_a_tampered_stored_experiment_on_read() { + let conn = Connection::open_in_memory().expect("open in-memory database"); + conn.execute_batch(crate::archive::store::SCHEMA) + .expect("initialize schema"); + let (json, entry) = envelope(1, GENESIS_HASH); + append_entry(&conn, "owner", &json, entry, 0).expect("append entry"); + conn.execute( + "UPDATE causal_ledger_entries SET entry_json = replace(entry_json, 'experiment-1', 'experiment-x')", + [], + ) + .expect("tamper stored entry"); + assert!(read_entries(&conn, "owner").is_err()); + } + + #[test] + fn hash_matches_the_browser_canonical_json_contract() { + let entry = LedgerEntryEnvelope { + sequence: 1, + previous_hash: GENESIS_HASH.to_string(), + hash: String::new(), + experiment: serde_json::json!({ + "schema": "causal-experiment/v1", + "experimentId": "golden", + "nested": { "z": true, "a": [1, "two", null] } + }), + }; + assert_eq!( + entry_hash(&entry).expect("hash golden entry"), + "514a88ba21863a1ea56a88e91e08aa382d249f46ffaf6c1eb797c745160490d8" + ); + } +} diff --git a/desktop/src-tauri/src/archive/mod.rs b/desktop/src-tauri/src/archive/mod.rs index 42c6812674..3c0dec414a 100644 --- a/desktop/src-tauri/src/archive/mod.rs +++ b/desktop/src-tauri/src/archive/mod.rs @@ -17,6 +17,7 @@ //! validation (sig/id + kind + p-tag + agent tag + frame=telemetry + author //! == agent) is applied fail-closed. +pub(crate) mod causal_ledger; mod pipeline; pub mod store; diff --git a/desktop/src-tauri/src/archive/store.rs b/desktop/src-tauri/src/archive/store.rs index ae0ef92e4b..935baa6a12 100644 --- a/desktop/src-tauri/src/archive/store.rs +++ b/desktop/src-tauri/src/archive/store.rs @@ -75,6 +75,20 @@ CREATE TABLE IF NOT EXISTS archive_migrations ( name TEXT PRIMARY KEY, applied_at INTEGER NOT NULL ); + +CREATE TABLE IF NOT EXISTS causal_ledger_entries ( + identity_pubkey TEXT NOT NULL, + sequence INTEGER NOT NULL, + experiment_id TEXT NOT NULL, + previous_hash TEXT NOT NULL, + hash TEXT NOT NULL, + entry_json TEXT NOT NULL, + recorded_at INTEGER NOT NULL, + PRIMARY KEY (identity_pubkey, sequence), + UNIQUE (identity_pubkey, experiment_id) +); +CREATE INDEX IF NOT EXISTS idx_causal_ledger_experiment + ON causal_ledger_entries (identity_pubkey, experiment_id); "; // ── Open / init ───────────────────────────────────────────────────────────── diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 2847b87877..0adc3b2f0c 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -85,11 +85,8 @@ use tray_menu::show_main_window; #[cfg_attr(mobile, tauri::mobile_entry_point)] pub fn run() { - // mesh-llm's async chains (model download, node start/join) overflow - // tokio's default 2 MiB worker stacks — a stack-guard SIGABRT, not a - // panic. Upstream mesh-llm and mesh-console both run on 8 MiB worker - // stacks for this reason; give Tauri's command runtime the same headroom - // before anything else touches tauri::async_runtime. + // mesh-llm's async chains overflow tokio's default 2 MiB worker stacks. + // Match upstream's 8 MiB stacks before anything touches tauri::async_runtime. #[cfg(feature = "mesh-llm")] match tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -903,6 +900,8 @@ pub fn run() { archive::read_archived_observer_events_for_channel, archive::index_observer_channel_id, archive::read_unindexed_observer_rows, + archive::causal_ledger::read_causal_ledger, + archive::causal_ledger::append_causal_ledger_entry, is_auto_update_supported, set_window_vibrancy, #[cfg(target_os = "macos")] diff --git a/desktop/src-tauri/src/managed_agents/types.rs b/desktop/src-tauri/src/managed_agents/types.rs index e5be105fed..17abacd43b 100644 --- a/desktop/src-tauri/src/managed_agents/types.rs +++ b/desktop/src-tauri/src/managed_agents/types.rs @@ -809,7 +809,8 @@ pub struct UpdateTeamRequest { pub const DEFAULT_ACP_COMMAND: &str = "buzz-acp"; /// ~5 min (320s) — matches the CLI harness default (BUZZ_ACP_IDLE_TIMEOUT). pub const DEFAULT_AGENT_TURN_TIMEOUT_SECONDS: u64 = 320; -pub const DEFAULT_AGENT_PARALLELISM: u32 = 10; +/// Use one worker by default because ACP adapters may launch helper processes. +pub const DEFAULT_AGENT_PARALLELISM: u32 = 1; fn default_agent_parallelism() -> u32 { DEFAULT_AGENT_PARALLELISM diff --git a/desktop/src-tauri/src/managed_agents/types/tests.rs b/desktop/src-tauri/src/managed_agents/types/tests.rs index 1db7b9b524..bf7d6ef27b 100644 --- a/desktop/src-tauri/src/managed_agents/types/tests.rs +++ b/desktop/src-tauri/src/managed_agents/types/tests.rs @@ -468,6 +468,11 @@ fn sample_agent_record() -> ManagedAgentRecord { .expect("sample record") } +#[test] +fn records_without_parallelism_default_to_one_worker() { + assert_eq!(sample_agent_record().parallelism, 1); +} + // ── AgentDefinition ↔ ManagedAgentRecord fold mapping (Phase 1A) ───────────────────── fn sample_persona() -> AgentDefinition { diff --git a/desktop/src/features/agents/lib/causalLedger.test.mjs b/desktop/src/features/agents/lib/causalLedger.test.mjs new file mode 100644 index 0000000000..384aaf171b --- /dev/null +++ b/desktop/src/features/agents/lib/causalLedger.test.mjs @@ -0,0 +1,169 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { CAUSAL_EXPERIMENT_SCHEMA, CausalLedger } from "./causalLedger.ts"; + +function experiment(id, overrides = {}) { + return { + schema: CAUSAL_EXPERIMENT_SCHEMA, + experimentId: id, + recordedAt: "2026-08-05T18:44:22Z", + task: { + description: "Build the Buzz Causal Ledger", + sourceMessageId: "c5b7084", + }, + execution: { + sessionId: `session-${id}`, + turnId: `turn-${id}`, + replayOf: null, + }, + failureFingerprint: "host-write-without-acp-permission/v1", + context: { + codeVersion: "2330ec7", + policyVersion: "workspace-write/v1", + modelVersion: "gpt-5", + toolVersion: "buzz-acp/v1", + environmentVersion: "macos-arm64", + }, + hypothesis: { + cause: "Host write bypassed ACP", + evidenceIds: [`evidence-${id}`], + }, + intervention: { + remedyId: "route-through-acp/v1", + changedVariable: "execution_path", + }, + result: { outcome: "validated", evidenceIds: [`result-${id}`] }, + coverage: { + acp: "observed", + host_workspace: "observed", + os_sandbox: "observed", + }, + relations: { supports: [], contradicts: [], invalidates: [] }, + ...overrides, + }; +} + +test("appends immutable, hash-linked causal experiments", async () => { + const ledger = new CausalLedger(); + const first = await ledger.append(experiment("one")); + const second = await ledger.append(experiment("two")); + + assert.equal(first.sequence, 1); + assert.equal(second.previousHash, first.hash); + assert.equal(await ledger.verify(), true); + await assert.rejects(() => ledger.append(experiment("one")), /Duplicate/); + assert.throws(() => { + first.experiment.result.outcome = "rejected"; + }, /read only/); +}); + +test("restores an append-only journal and rejects a tampered restart", async () => { + const ledger = new CausalLedger(); + await ledger.append(experiment("before-crash")); + await ledger.append(experiment("after-restart")); + const journal = ledger.toJournal(); + + const restored = await CausalLedger.fromJournal(journal); + assert.equal(restored.size, 2); + assert.equal(await restored.verify(), true); + + const tampered = journal.replace("Host write bypassed ACP", "Invented cause"); + await assert.rejects( + () => CausalLedger.fromJournal(tampered), + /integrity failure/, + ); +}); + +test("does not let unrelated or version-drifted successes strengthen a finding", async () => { + const ledger = new CausalLedger(); + const target = experiment("target"); + await ledger.append(target); + await ledger.append(experiment("same-context")); + await ledger.append( + experiment("unrelated", { failureFingerprint: "different-failure/v1" }), + ); + await ledger.append( + experiment("drifted", { + context: { ...target.context, policyVersion: "workspace-write/v2" }, + }), + ); + + const finding = ledger.findingFor(target); + assert.equal(finding.comparableExperiments, 2); + assert.equal(finding.validated, 2); + assert.equal(finding.confidence, 1); +}); + +test("missing coverage and contradictions lower confidence", async () => { + const ledger = new CausalLedger(); + const target = experiment("target"); + await ledger.append(target); + await ledger.append( + experiment("contradiction", { + result: { outcome: "rejected", evidenceIds: ["contradiction-evidence"] }, + relations: { supports: [], contradicts: ["target"], invalidates: [] }, + }), + ); + await ledger.append( + experiment("live-dogfood-gap", { + result: { outcome: "inconclusive", evidenceIds: [] }, + coverage: { + acp: "observed", + host_workspace: "missing", + os_sandbox: "missing", + }, + }), + ); + + const finding = ledger.findingFor(target); + assert.deepEqual( + { + validated: finding.validated, + rejected: finding.rejected, + inconclusive: finding.inconclusive, + }, + { validated: 1, rejected: 1, inconclusive: 1 }, + ); + assert.equal(finding.confidence, 1 / 3); +}); + +test("keeps 10,000 adversarial experiments inside their causal boundaries", async () => { + const ledger = new CausalLedger(); + const target = experiment("target"); + await ledger.append(target); + for (let index = 1; index < 10_000; index += 1) { + const poisoned = index % 5 === 0; + const versionDrift = index % 7 === 0; + await ledger.append( + experiment(`scale-${index}`, { + failureFingerprint: poisoned + ? "poisoned-unrelated/v1" + : target.failureFingerprint, + context: versionDrift + ? { ...target.context, codeVersion: `drift-${index}` } + : target.context, + result: { + outcome: index % 11 === 0 ? "rejected" : "validated", + evidenceIds: [`result-${index}`], + }, + }), + ); + } + + const finding = ledger.findingFor(target); + assert.equal(ledger.size, 10_000); + assert.equal(await ledger.verify(), true); + assert.equal( + finding.comparableExperiments, + ledger + .entries() + .filter( + ({ experiment: item }) => + item.failureFingerprint === target.failureFingerprint && + item.context.codeVersion === target.context.codeVersion, + ).length, + ); + assert.equal(finding.experimentIds.includes("scale-5"), false); + assert.equal(finding.experimentIds.includes("scale-7"), false); +}); diff --git a/desktop/src/features/agents/lib/causalLedger.ts b/desktop/src/features/agents/lib/causalLedger.ts new file mode 100644 index 0000000000..d7aed368b4 --- /dev/null +++ b/desktop/src/features/agents/lib/causalLedger.ts @@ -0,0 +1,221 @@ +export const CAUSAL_EXPERIMENT_SCHEMA = "causal-experiment/v1" as const; + +export type ExperimentOutcome = + | "validated" + | "rejected" + | "inconclusive" + | "untested"; + +export type EvidenceCoverage = "observed" | "missing"; + +export type CausalExperiment = { + schema: typeof CAUSAL_EXPERIMENT_SCHEMA; + experimentId: string; + recordedAt: string; + task: { description: string; sourceMessageId: string | null }; + execution: { sessionId: string; turnId: string; replayOf: string | null }; + failureFingerprint: string; + context: { + codeVersion: string; + policyVersion: string; + modelVersion: string; + toolVersion: string; + environmentVersion: string; + }; + hypothesis: { cause: string; evidenceIds: string[] }; + intervention: { + remedyId: string; + changedVariable: string; + successCriteria?: string; + approvedAt?: string; + }; + result: { outcome: ExperimentOutcome; evidenceIds: string[] }; + evaluation?: { + evaluator: "owner-independent"; + rationale: string; + evaluatedAt: string; + }; + coverage: Record; + relations: { + supports: string[]; + contradicts: string[]; + invalidates: string[]; + }; +}; + +export type LedgerEntry = { + sequence: number; + previousHash: string; + hash: string; + experiment: CausalExperiment; +}; + +export type CausalFinding = { + comparableExperiments: number; + validated: number; + rejected: number; + inconclusive: number; + confidence: number; + experimentIds: string[]; +}; + +const GENESIS_HASH = "0".repeat(64); + +function canonical(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`; + if (value && typeof value === "object") { + return `{${Object.entries(value as Record) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, entry]) => `${JSON.stringify(key)}:${canonical(entry)}`) + .join(",")}}`; + } + return JSON.stringify(value); +} + +async function sha256(value: string): Promise { + const bytes = new TextEncoder().encode(value); + const digest = await crypto.subtle.digest("SHA-256", bytes); + return Array.from(new Uint8Array(digest), (byte) => + byte.toString(16).padStart(2, "0"), + ).join(""); +} + +function sameContext( + left: CausalExperiment["context"], + right: CausalExperiment["context"], +): boolean { + return canonical(left) === canonical(right); +} + +function hasCompleteCoverage(experiment: CausalExperiment): boolean { + return Object.values(experiment.coverage).every( + (value) => value === "observed", + ); +} + +function immutableClone(value: T): T { + if (Array.isArray(value)) { + return Object.freeze(value.map(immutableClone)) as T; + } + if (value && typeof value === "object") { + const clone = Object.fromEntries( + Object.entries(value as Record).map(([key, entry]) => [ + key, + immutableClone(entry), + ]), + ); + return Object.freeze(clone) as T; + } + return value; +} + +export class CausalLedger { + readonly #entries: LedgerEntry[] = []; + readonly #ids = new Set(); + + get size(): number { + return this.#entries.length; + } + + entries(): readonly LedgerEntry[] { + return this.#entries; + } + + toJournal(): string { + return this.#entries.map((entry) => canonical(entry)).join("\n"); + } + + static async fromJournal(journal: string): Promise { + const ledger = new CausalLedger(); + const lines = journal.split("\n").filter((line) => line.trim()); + for (const line of lines) { + const persisted = JSON.parse(line) as LedgerEntry; + const restored = await ledger.append(persisted.experiment); + if ( + restored.sequence !== persisted.sequence || + restored.previousHash !== persisted.previousHash || + restored.hash !== persisted.hash + ) { + throw new Error( + `Causal ledger integrity failure at sequence ${persisted.sequence}`, + ); + } + } + return ledger; + } + + async append(experiment: CausalExperiment): Promise { + if (experiment.schema !== CAUSAL_EXPERIMENT_SCHEMA) { + throw new Error( + `Unsupported causal experiment schema: ${experiment.schema}`, + ); + } + if (this.#ids.has(experiment.experimentId)) { + throw new Error( + `Duplicate causal experiment: ${experiment.experimentId}`, + ); + } + const sequence = this.#entries.length + 1; + const previousHash = this.#entries.at(-1)?.hash ?? GENESIS_HASH; + const frozenExperiment = immutableClone(experiment); + const hash = await sha256( + canonical({ sequence, previousHash, experiment: frozenExperiment }), + ); + const entry = Object.freeze({ + sequence, + previousHash, + hash, + experiment: frozenExperiment, + }); + this.#entries.push(entry); + this.#ids.add(experiment.experimentId); + return entry; + } + + async verify(): Promise { + let previousHash = GENESIS_HASH; + for (const entry of this.#entries) { + if (entry.previousHash !== previousHash) return false; + const expected = await sha256( + canonical({ + sequence: entry.sequence, + previousHash: entry.previousHash, + experiment: entry.experiment, + }), + ); + if (entry.hash !== expected) return false; + previousHash = entry.hash; + } + return true; + } + + findingFor(target: CausalExperiment): CausalFinding { + const comparable = this.#entries + .map((entry) => entry.experiment) + .filter( + (experiment) => + experiment.failureFingerprint === target.failureFingerprint && + experiment.intervention.remedyId === target.intervention.remedyId && + sameContext(experiment.context, target.context), + ); + const validated = comparable.filter( + (experiment) => experiment.result.outcome === "validated", + ).length; + const rejected = comparable.filter( + (experiment) => experiment.result.outcome === "rejected", + ).length; + const inconclusive = comparable.length - validated - rejected; + const complete = comparable.filter(hasCompleteCoverage).length; + const decisive = validated + rejected; + const evidenceFactor = comparable.length ? complete / comparable.length : 0; + const confidence = decisive ? (validated / decisive) * evidenceFactor : 0; + return { + comparableExperiments: comparable.length, + validated, + rejected, + inconclusive, + confidence, + experimentIds: comparable.map((experiment) => experiment.experimentId), + }; + } +} diff --git a/desktop/src/features/agents/lib/causalReplayProposal.test.mjs b/desktop/src/features/agents/lib/causalReplayProposal.test.mjs new file mode 100644 index 0000000000..eabfedd116 --- /dev/null +++ b/desktop/src/features/agents/lib/causalReplayProposal.test.mjs @@ -0,0 +1,162 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + buildApprovedReplayProposal, + buildIndependentEvaluation, + buildReplayDispatchMessage, + buildReplayDispatchReceipt, + proposalIdFromReplayTask, +} from "./causalReplayProposal.ts"; + +const candidate = { + schema: "causal-experiment/v1", + experimentId: "live:agent:session-1", + recordedAt: "2026-08-05T00:00:00Z", + task: { description: "Write the file", sourceMessageId: "message-1" }, + execution: { sessionId: "session-1", turnId: "turn-1", replayOf: null }, + failureFingerprint: "unclassified-live-candidate/v1", + context: { + codeVersion: "code-1", + policyVersion: "policy-1", + modelVersion: "model-1", + toolVersion: "tool-1", + environmentVersion: "env-1", + }, + hypothesis: { cause: "unclassified", evidenceIds: [] }, + intervention: { remedyId: "unclassified", changedVariable: "unclassified" }, + result: { outcome: "untested", evidenceIds: ["receipt-1"] }, + coverage: { acp_observer: "observed", os_sandbox: "missing" }, + relations: { supports: [], contradicts: [], invalidates: [] }, +}; + +test("builds one owner-approved controlled replay without inventing a result", () => { + const proposal = buildApprovedReplayProposal( + candidate, + { + failureFingerprint: "host-write-bypass/v1", + cause: "The host tool bypassed the governed path", + changedVariable: "execution path: host tool → ACP workspace tool", + successCriteria: + "Permission is requested before the write and the task succeeds", + }, + { experimentId: "proposal-1", recordedAt: "2026-08-05T01:00:00Z" }, + ); + + assert.equal(proposal.execution.replayOf, candidate.experimentId); + assert.equal(proposal.intervention.approvedAt, "2026-08-05T01:00:00Z"); + assert.equal(proposal.result.outcome, "untested"); + assert.deepEqual(proposal.result.evidenceIds, []); + assert.deepEqual(proposal.hypothesis.evidenceIds, ["receipt-1"]); +}); + +test("requires every causal and evaluation field before approval", () => { + assert.throws( + () => + buildApprovedReplayProposal( + candidate, + { + failureFingerprint: "host-write-bypass/v1", + cause: "A cause", + changedVariable: "one variable", + successCriteria: " ", + }, + { experimentId: "proposal-1", recordedAt: "2026-08-05T01:00:00Z" }, + ), + /Success criteria is required/, + ); +}); + +test("refuses to replay an already evaluated candidate", () => { + assert.throws( + () => + buildApprovedReplayProposal( + { ...candidate, result: { outcome: "validated", evidenceIds: [] } }, + { + failureFingerprint: "failure/v1", + cause: "A cause", + changedVariable: "one variable", + successCriteria: "A measured result", + }, + { experimentId: "proposal-1", recordedAt: "2026-08-05T01:00:00Z" }, + ), + /Only an untested candidate/, + ); +}); + +test("dispatches an approved replay with a durable correlation marker", () => { + const proposal = buildApprovedReplayProposal( + candidate, + { + failureFingerprint: "failure/v1", + cause: "A cause", + changedVariable: "one variable", + successCriteria: "A measured result", + }, + { experimentId: "proposal-1", recordedAt: "2026-08-05T01:00:00Z" }, + ); + const message = buildReplayDispatchMessage(proposal); + assert.equal(proposalIdFromReplayTask(message), "proposal-1"); + assert.match(message, /disposable workspace/); + assert.match(message, /Change exactly one variable: one variable/); + + const receipt = buildReplayDispatchReceipt( + proposal, + "event-1", + "2026-08-05T01:01:00Z", + ); + assert.equal(receipt.task.sourceMessageId, "event-1"); + assert.equal(receipt.execution.replayOf, proposal.experimentId); + assert.deepEqual(receipt.result.evidenceIds, ["message:event-1"]); +}); + +test("requires cited evidence before sealing an independent verdict", () => { + const replay = { + ...candidate, + experimentId: "replay-1", + execution: { ...candidate.execution, replayOf: "proposal-1" }, + }; + assert.throws( + () => + buildIndependentEvaluation( + replay, + { outcome: "validated", evidenceIds: " ", rationale: "It worked" }, + { experimentId: "evaluation-1", recordedAt: "2026-08-05T02:00:00Z" }, + ), + /At least one evidence ID/, + ); + const evaluation = buildIndependentEvaluation( + replay, + { + outcome: "rejected", + evidenceIds: "observer:1\nreceipt:2", + rationale: "The success criterion was not met.", + }, + { experimentId: "evaluation-1", recordedAt: "2026-08-05T02:00:00Z" }, + ); + assert.equal(evaluation.execution.replayOf, replay.experimentId); + assert.equal(evaluation.result.outcome, "rejected"); + assert.deepEqual(evaluation.relations.contradicts, [replay.experimentId]); + assert.equal(evaluation.evaluation.evaluator, "owner-independent"); +}); + +test("refuses to validate a replay while an execution layer is missing", () => { + const replay = { + ...candidate, + experimentId: "replay-1", + execution: { ...candidate.execution, replayOf: "proposal-1" }, + }; + assert.throws( + () => + buildIndependentEvaluation( + replay, + { + outcome: "validated", + evidenceIds: "observer:1", + rationale: "The observed portion worked.", + }, + { experimentId: "evaluation-1", recordedAt: "2026-08-05T02:00:00Z" }, + ), + /missing coverage cannot be validated/, + ); +}); diff --git a/desktop/src/features/agents/lib/causalReplayProposal.ts b/desktop/src/features/agents/lib/causalReplayProposal.ts new file mode 100644 index 0000000000..e3bfddeebd --- /dev/null +++ b/desktop/src/features/agents/lib/causalReplayProposal.ts @@ -0,0 +1,169 @@ +import { + CAUSAL_EXPERIMENT_SCHEMA, + type CausalExperiment, +} from "./causalLedger"; + +export type ReplayProposalInput = { + failureFingerprint: string; + cause: string; + changedVariable: string; + successCriteria: string; +}; + +const REPLAY_MARKER_PREFIX = "buzz-controlled-replay:"; + +function required(label: string, value: string): string { + const normalized = value.trim(); + if (!normalized) throw new Error(`${label} is required.`); + return normalized; +} + +export function buildApprovedReplayProposal( + candidate: CausalExperiment, + input: ReplayProposalInput, + identity: { experimentId: string; recordedAt: string }, +): CausalExperiment { + if (candidate.result.outcome !== "untested") { + throw new Error( + "Only an untested candidate can start a controlled replay.", + ); + } + const failureFingerprint = required( + "Failure fingerprint", + input.failureFingerprint, + ); + const cause = required("Cause hypothesis", input.cause); + const changedVariable = required("Changed variable", input.changedVariable); + const successCriteria = required("Success criteria", input.successCriteria); + + return { + schema: CAUSAL_EXPERIMENT_SCHEMA, + experimentId: identity.experimentId, + recordedAt: identity.recordedAt, + task: candidate.task, + execution: { + sessionId: `replay-proposal:${candidate.execution.sessionId}`, + turnId: "pending-owner-approved-replay", + replayOf: candidate.experimentId, + }, + failureFingerprint, + context: candidate.context, + hypothesis: { + cause, + evidenceIds: [...candidate.result.evidenceIds], + }, + intervention: { + remedyId: `remedy:${identity.experimentId}`, + changedVariable, + successCriteria, + approvedAt: identity.recordedAt, + }, + result: { outcome: "untested", evidenceIds: [] }, + coverage: candidate.coverage, + relations: { + supports: [], + contradicts: [], + invalidates: [], + }, + }; +} + +export function replayMarker(proposalId: string): string { + return `[${REPLAY_MARKER_PREFIX}${proposalId}]`; +} + +export function proposalIdFromReplayTask(description: string): string | null { + const start = description.indexOf(`[${REPLAY_MARKER_PREFIX}`); + if (start < 0) return null; + const valueStart = start + REPLAY_MARKER_PREFIX.length + 1; + const end = description.indexOf("]", valueStart); + const value = end < 0 ? "" : description.slice(valueStart, end).trim(); + return value || null; +} + +export function buildReplayDispatchMessage(proposal: CausalExperiment): string { + if ( + !proposal.intervention.approvedAt || + !proposal.intervention.successCriteria + ) { + throw new Error("The replay must be approved before it can run."); + } + return [ + replayMarker(proposal.experimentId), + "Run this as a controlled verification replay in a disposable workspace.", + `Original task: ${proposal.task.description}`, + `Change exactly one variable: ${proposal.intervention.changedVariable}`, + `Keep fixed: code, model, tools, policy, and environment versions unless that named variable explicitly changes one of them.`, + `Success criteria: ${proposal.intervention.successCriteria}`, + "Report observable evidence and do not claim the remedy is validated; an independent owner evaluation happens after this turn.", + ].join("\n\n"); +} + +export function buildReplayDispatchReceipt( + proposal: CausalExperiment, + eventId: string, + recordedAt: string, +): CausalExperiment { + if (!proposal.intervention.approvedAt) { + throw new Error("The replay must be approved before it can be dispatched."); + } + return { + ...proposal, + experimentId: `dispatch:${proposal.experimentId}:${eventId}`, + recordedAt, + task: { ...proposal.task, sourceMessageId: eventId }, + execution: { + sessionId: `replay-dispatch:${eventId}`, + turnId: "awaiting-agent-session", + replayOf: proposal.experimentId, + }, + result: { outcome: "untested", evidenceIds: [`message:${eventId}`] }, + relations: { supports: [], contradicts: [], invalidates: [] }, + }; +} + +export function buildIndependentEvaluation( + replay: CausalExperiment, + input: { + outcome: Exclude; + evidenceIds: string; + rationale: string; + }, + identity: { experimentId: string; recordedAt: string }, +): CausalExperiment { + if (replay.result.outcome !== "untested" || !replay.execution.replayOf) { + throw new Error("Only a completed, unevaluated replay can be evaluated."); + } + const evidenceIds = input.evidenceIds + .split(/[\n,]/) + .map((value) => value.trim()) + .filter(Boolean); + if (!evidenceIds.length) + throw new Error("At least one evidence ID is required."); + if ( + input.outcome === "validated" && + Object.values(replay.coverage).some((coverage) => coverage !== "observed") + ) { + throw new Error( + "A replay with missing coverage cannot be validated; choose inconclusive or rejected.", + ); + } + const rationale = required("Evaluation rationale", input.rationale); + return { + ...replay, + experimentId: identity.experimentId, + recordedAt: identity.recordedAt, + execution: { ...replay.execution, replayOf: replay.experimentId }, + result: { outcome: input.outcome, evidenceIds }, + evaluation: { + evaluator: "owner-independent", + rationale, + evaluatedAt: identity.recordedAt, + }, + relations: { + supports: input.outcome === "validated" ? [replay.experimentId] : [], + contradicts: input.outcome === "rejected" ? [replay.experimentId] : [], + invalidates: [], + }, + }; +} diff --git a/desktop/src/features/agents/lib/liveCausalLedger.test.mjs b/desktop/src/features/agents/lib/liveCausalLedger.test.mjs new file mode 100644 index 0000000000..06a3cc5067 --- /dev/null +++ b/desktop/src/features/agents/lib/liveCausalLedger.test.mjs @@ -0,0 +1,217 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + browserLedgerPersistence, + LiveCausalLedger, +} from "./liveCausalLedger.ts"; + +function memoryStorage() { + const values = new Map(); + return { + getItem: (key) => values.get(key) ?? null, + setItem: (key, value) => values.set(key, value), + }; +} + +function observer(seq, kind, payload = {}) { + return { + seq, + timestamp: `2026-08-05T18:5${seq}:00Z`, + kind, + agentIndex: 0, + channelId: "2db0f46d", + sessionId: "live-session-1", + turnId: "turn-1", + payload, + }; +} + +test("automatically closes a live session into an owner-local candidate", async () => { + const storage = memoryStorage(); + const ledger = new LiveCausalLedger( + "OWNER", + browserLedgerPersistence("OWNER", storage), + ); + await ledger.ingest( + "agent", + observer(1, "task_captured", { + description: "Wire this session into the causal ledger", + sourceMessageId: "414ccec", + }), + ); + await ledger.ingest( + "agent", + observer(2, "permission_decision", { + decision: "allow_once", + }), + ); + await ledger.ingest("agent", observer(3, "turn_completed")); + + const [entry] = await ledger.entries(); + assert.equal(entry.experiment.result.outcome, "untested"); + assert.equal(entry.experiment.task.sourceMessageId, "414ccec"); + assert.equal(entry.experiment.coverage.acp_permission_gate, "observed"); + assert.equal(entry.experiment.coverage.host_workspace, "missing"); + assert.equal(entry.experiment.coverage.os_sandbox, "missing"); + assert.ok(storage.getItem(ledger.storageKey)); +}); + +test("captures the task and OS failure from raw ACP evidence", async () => { + const ledger = new LiveCausalLedger( + "owner", + browserLedgerPersistence("owner", memoryStorage()), + ); + await ledger.ingest( + "agent", + observer(1, "acp_write", { + method: "session/prompt", + params: { + prompt: [ + { + type: "text", + text: `[Buzz event: @mention]\nEvent ID: ${"a".repeat(64)}\nContent: Touch the protected file once.`, + }, + ], + }, + }), + ); + await ledger.ingest( + "agent", + observer(2, "acp_read", { + method: "session/update", + params: { + update: { + sessionUpdate: "tool_call_update", + toolCallId: "touch-library", + status: "in_progress", + content: [ + { + type: "text", + text: "touch: /Library/file: Permission denied", + }, + ], + }, + }, + }), + ); + await ledger.ingest( + "agent", + observer(3, "acp_read", { + method: "session/update", + params: { + update: { + sessionUpdate: "tool_call_update", + toolCallId: "touch-library", + status: "failed", + content: [], + }, + }, + }), + ); + await ledger.ingest("agent", observer(4, "turn_completed")); + + const [entry] = await ledger.entries(); + assert.equal( + entry.experiment.task.description, + "Touch the protected file once.", + ); + assert.equal(entry.experiment.task.sourceMessageId, "a".repeat(64)); + assert.equal(entry.experiment.coverage.os_sandbox, "observed"); +}); + +test("records each turn in a reused session as its own candidate", async () => { + const ledger = new LiveCausalLedger( + "owner", + browserLedgerPersistence("owner", memoryStorage()), + ); + await ledger.ingest("agent", observer(1, "turn_completed")); + await ledger.ingest("agent", { + ...observer(2, "turn_completed"), + turnId: "turn-2", + }); + + const entries = await ledger.entries(); + assert.equal(entries.length, 2); + assert.equal(entries[0].experiment.execution.turnId, "turn-1"); + assert.equal(entries[1].experiment.execution.turnId, "turn-2"); +}); + +test("restores the candidate after restart and does not duplicate terminal replay", async () => { + const storage = memoryStorage(); + const first = new LiveCausalLedger( + "owner", + browserLedgerPersistence("owner", storage), + ); + await first.ingest("agent", observer(1, "turn_completed")); + + const restarted = new LiveCausalLedger( + "owner", + browserLedgerPersistence("owner", storage), + ); + await restarted.ingest("agent", observer(1, "turn_completed")); + assert.equal((await restarted.entries()).length, 1); +}); + +test("links a marked terminal replay to its approved proposal", async () => { + const storage = memoryStorage(); + const persistence = browserLedgerPersistence("owner", storage); + const ledger = new LiveCausalLedger("owner", persistence); + const approved = { + schema: "causal-experiment/v1", + experimentId: "proposal-1", + recordedAt: "2026-08-05T00:00:00Z", + task: { description: "Write the file", sourceMessageId: "message-1" }, + execution: { + sessionId: "proposal", + turnId: "pending", + replayOf: "candidate-1", + }, + failureFingerprint: "host-write/v1", + context: { + codeVersion: "code-1", + policyVersion: "policy-1", + modelVersion: "model-1", + toolVersion: "tool-1", + environmentVersion: "env-1", + }, + hypothesis: { cause: "Host bypass", evidenceIds: ["evidence-1"] }, + intervention: { + remedyId: "remedy-1", + changedVariable: "tool path", + successCriteria: "Permission before write", + approvedAt: "2026-08-05T00:00:00Z", + }, + result: { outcome: "untested", evidenceIds: [] }, + coverage: { acp_observer: "observed", os_sandbox: "missing" }, + relations: { supports: [], contradicts: [], invalidates: [] }, + }; + const seed = new (await import("./causalLedger.ts")).CausalLedger(); + await persistence.appendEntry(await seed.append(approved)); + await persistence.appendEntry( + await seed.append({ + ...approved, + experimentId: "dispatch-1", + task: { ...approved.task, sourceMessageId: "dispatch-1" }, + execution: { + sessionId: "replay-dispatch:dispatch-1", + turnId: "awaiting-agent-session", + replayOf: approved.experimentId, + }, + result: { outcome: "untested", evidenceIds: ["message:dispatch-1"] }, + }), + ); + await ledger.ingest( + "agent", + observer(1, "task_captured", { + description: "[buzz-controlled-replay:proposal-1]\n\nRun it", + sourceMessageId: "dispatch-1", + }), + ); + await ledger.ingest("agent", observer(2, "turn_completed")); + const replay = (await ledger.entries()).at(-1).experiment; + assert.equal(replay.execution.replayOf, "proposal-1"); + assert.equal(replay.failureFingerprint, approved.failureFingerprint); + assert.equal(replay.intervention.changedVariable, "tool path"); + assert.equal(replay.result.outcome, "untested"); +}); diff --git a/desktop/src/features/agents/lib/liveCausalLedger.ts b/desktop/src/features/agents/lib/liveCausalLedger.ts new file mode 100644 index 0000000000..75e99b9894 --- /dev/null +++ b/desktop/src/features/agents/lib/liveCausalLedger.ts @@ -0,0 +1,259 @@ +import type { ObserverEvent } from "../ui/agentSessionTypes"; +import { + extractPromptText, + extractToolResult, + parsePromptText, +} from "../ui/agentSessionTranscriptHelpers"; +import { asRecord, asString } from "../ui/agentSessionUtils"; +import { + CAUSAL_EXPERIMENT_SCHEMA, + CausalLedger, + type CausalExperiment, +} from "./causalLedger"; +import { proposalIdFromReplayTask } from "./causalReplayProposal"; + +export type CausalLedgerPersistence = { + loadJournal(): Promise; + appendEntry(entry: import("./causalLedger").LedgerEntry): Promise; +}; + +export function browserLedgerPersistence( + owner: string, + storage: Pick, +): CausalLedgerPersistence { + const storageKey = `buzz-causal-ledger.v1:${owner.toLowerCase()}`; + return { + async loadJournal() { + return storage.getItem(storageKey) ?? ""; + }, + async appendEntry(entry) { + const existing = storage.getItem(storageKey); + storage.setItem( + storageKey, + existing + ? `${existing}\n${JSON.stringify(entry)}` + : JSON.stringify(entry), + ); + }, + }; +} + +function payload(event: ObserverEvent): Record { + return event.payload && typeof event.payload === "object" + ? (event.payload as Record) + : {}; +} + +function text(value: unknown): string | null { + return typeof value === "string" && value.trim() ? value.trim() : null; +} + +function eventId(event: ObserverEvent): string { + return `observer:${event.sessionId ?? "unknown"}:${event.seq}:${event.timestamp}`; +} + +function taskFromEvents(events: readonly ObserverEvent[]): { + description: string; + sourceMessageId: string | null; +} { + const captured = events.find((event) => event.kind === "task_captured"); + if (captured) { + const capturedPayload = payload(captured); + return { + description: + text(capturedPayload.description) ?? + "Task unavailable from observer evidence", + sourceMessageId: text(capturedPayload.sourceMessageId), + }; + } + + const promptEvent = events.find((event) => { + const eventPayload = payload(event); + return ( + event.kind === "acp_write" && + asString(eventPayload.method) === "session/prompt" + ); + }); + if (!promptEvent) { + return { + description: "Task unavailable from observer evidence", + sourceMessageId: null, + }; + } + const prompt = extractPromptText(payload(promptEvent)); + const parsed = parsePromptText(prompt); + return { + description: + parsed.userText || + prompt.trim() || + "Task unavailable from observer evidence", + sourceMessageId: parsed.userEventId, + }; +} + +function hasOsSandboxEvidence(events: readonly ObserverEvent[]): boolean { + const toolOutputById = new Map(); + for (const event of events) { + if (event.kind !== "acp_read") continue; + const eventPayload = payload(event); + if (asString(eventPayload.method) !== "session/update") continue; + const update = asRecord(asRecord(eventPayload.params).update); + if (asString(update.sessionUpdate) !== "tool_call_update") continue; + const toolCallId = asString(update.toolCallId); + const output = extractToolResult(update); + if (toolCallId && output) toolOutputById.set(toolCallId, output); + if (asString(update.status) !== "failed") continue; + const detail = output || (toolCallId ? toolOutputById.get(toolCallId) : ""); + if (/permission denied|operation not permitted/i.test(detail ?? "")) { + return true; + } + } + return false; +} + +export class LiveCausalLedger { + readonly #persistence: CausalLedgerPersistence; + readonly #owner: string; + readonly #sessions = new Map(); + #ledger = new CausalLedger(); + #ready: Promise; + #writes: Promise = Promise.resolve(); + + constructor(owner: string, persistence: CausalLedgerPersistence) { + this.#owner = owner.toLowerCase(); + this.#persistence = persistence; + this.#ready = this.#restore(); + } + + get storageKey(): string { + return `buzz-causal-ledger.v1:${this.#owner}`; + } + + async #restore() { + const journal = await this.#persistence.loadJournal(); + if (journal) this.#ledger = await CausalLedger.fromJournal(journal); + } + + ingest(agentPubkey: string, event: ObserverEvent): Promise { + this.#writes = this.#writes.then(async () => { + await this.#ready; + if (!event.sessionId) return; + const runId = event.turnId ?? `terminal-${event.seq}`; + const key = `${agentPubkey.toLowerCase()}:${event.sessionId}:${runId}`; + const session = this.#sessions.get(key) ?? []; + session.push(event); + this.#sessions.set(key, session); + if (event.kind !== "turn_completed" && event.kind !== "turn_error") + return; + + // Owner actions can append approvals/evaluations while this live ingestor + // remains mounted. Refresh before closing the turn so correlation and the + // next hash are based on the same durable ledger the UI just changed. + const latestJournal = await this.#persistence.loadJournal(); + this.#ledger = latestJournal + ? await CausalLedger.fromJournal(latestJournal) + : new CausalLedger(); + + const experimentId = `live:${agentPubkey.toLowerCase()}:${event.sessionId}:${runId}`; + if ( + this.#ledger + .entries() + .some((entry) => entry.experiment.experimentId === experimentId) + ) { + this.#sessions.delete(key); + return; + } + const entry = await this.#ledger.append( + this.#candidate(experimentId, event.sessionId, event.turnId, session), + ); + await this.#persistence.appendEntry(entry); + this.#sessions.delete(key); + }); + return this.#writes; + } + + async entries() { + await this.#ready; + await this.#writes; + return this.#ledger.entries(); + } + + #candidate( + experimentId: string, + sessionId: string, + turnId: string | null, + events: ObserverEvent[], + ): CausalExperiment { + const task = taskFromEvents(events); + const taskDescription = task.description; + const sourceMessageId = task.sourceMessageId; + const proposalId = proposalIdFromReplayTask(taskDescription); + const markedProposal = proposalId + ? this.#ledger + .entries() + .find((entry) => entry.experiment.experimentId === proposalId) + ?.experiment + : undefined; + const hasDispatchReceipt = this.#ledger + .entries() + .some( + (entry) => + entry.experiment.execution.replayOf === + markedProposal?.experimentId && + entry.experiment.execution.sessionId.startsWith("replay-dispatch:") && + entry.experiment.task.sourceMessageId === sourceMessageId, + ); + const proposal = hasDispatchReceipt ? markedProposal : undefined; + const layers = new Set( + events.map((event) => { + if (event.kind === "host_operation") return "host_workspace"; + if (event.kind === "permission_decision") return "acp_permission_gate"; + return "acp_observer"; + }), + ); + return { + schema: CAUSAL_EXPERIMENT_SCHEMA, + experimentId, + recordedAt: events.at(-1)?.timestamp ?? new Date().toISOString(), + task: { + description: proposal?.task.description ?? taskDescription, + sourceMessageId, + }, + execution: { + sessionId, + turnId: turnId ?? "unknown", + replayOf: proposal?.experimentId ?? null, + }, + failureFingerprint: + proposal?.failureFingerprint ?? "unclassified-live-candidate/v1", + context: proposal?.context ?? { + codeVersion: "unknown", + policyVersion: "unknown", + modelVersion: "unknown", + toolVersion: "buzz-acp-observer/v1", + environmentVersion: "unknown", + }, + hypothesis: proposal?.hypothesis ?? { + cause: "unclassified", + evidenceIds: [], + }, + intervention: proposal?.intervention ?? { + remedyId: "unclassified", + changedVariable: "unclassified", + }, + result: { + outcome: "untested", + evidenceIds: events.map(eventId), + }, + coverage: { + acp_observer: layers.has("acp_observer") ? "observed" : "missing", + acp_permission_gate: layers.has("acp_permission_gate") + ? "observed" + : "missing", + host_workspace: layers.has("host_workspace") ? "observed" : "missing", + os_sandbox: hasOsSandboxEvidence(events) ? "observed" : "missing", + }, + relations: { supports: [], contradicts: [], invalidates: [] }, + }; + } +} diff --git a/desktop/src/features/agents/observerRelayStore.ts b/desktop/src/features/agents/observerRelayStore.ts index 611fdd489d..2ef9b891f7 100644 --- a/desktop/src/features/agents/observerRelayStore.ts +++ b/desktop/src/features/agents/observerRelayStore.ts @@ -45,6 +45,9 @@ const EMPTY_EVENTS: ObserverEvent[] = []; const EMPTY_TRANSCRIPT: TranscriptItem[] = []; const listeners = new Set<() => void>(); +const eventListeners = new Set< + (agentPubkey: string, event: ObserverEvent) => void +>(); const eventsByAgent = new Map(); const transcriptByAgent = new Map(); const snapshotByAgent = new Map(); @@ -211,6 +214,9 @@ function appendAgentEvent(agentPubkey: string, event: ObserverEvent) { ? sorted.slice(sorted.length - MAX_OBSERVER_EVENTS) : sorted; eventsByAgent.set(key, final); + for (const listener of eventListeners) { + listener(key, event); + } // Determine whether the new event landed at the end of the sorted array. // If it did (common case), we can incrementally process just this event. @@ -233,6 +239,16 @@ function appendAgentEvent(agentPubkey: string, event: ObserverEvent) { notifyListeners(); } +/** Subscribe to each newly accepted live observer event after deduplication. */ +export function subscribeObserverEvents( + listener: (agentPubkey: string, event: ObserverEvent) => void, +) { + eventListeners.add(listener); + return () => { + eventListeners.delete(listener); + }; +} + /** * Compose the map key for the channel-scoped archive transcript. * Separates agent identity from channel with `:` — the same delimiter used by diff --git a/desktop/src/features/agents/ui/CausalCandidateCard.tsx b/desktop/src/features/agents/ui/CausalCandidateCard.tsx new file mode 100644 index 0000000000..13af339501 --- /dev/null +++ b/desktop/src/features/agents/ui/CausalCandidateCard.tsx @@ -0,0 +1,495 @@ +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { AlertTriangle, Eye, GitBranch, ShieldCheck } from "lucide-react"; +import * as React from "react"; +import { toast } from "sonner"; + +import { + appendCausalExperiment, + readCausalLedger, +} from "@/shared/api/tauriCausalLedger"; +import { sendChannelMessage } from "@/shared/api/tauri"; +import { Badge } from "@/shared/ui/badge"; +import { Button } from "@/shared/ui/button"; +import { Input } from "@/shared/ui/input"; +import { Textarea } from "@/shared/ui/textarea"; + +import { + buildApprovedReplayProposal, + buildIndependentEvaluation, + buildReplayDispatchMessage, + buildReplayDispatchReceipt, +} from "../lib/causalReplayProposal"; +import { inspectCausalCandidate } from "./causalCandidateInspection"; + +export function CausalCandidateCard({ + agentPubkey, + channelId, + sessionId, +}: { + agentPubkey: string; + channelId: string | null; + sessionId: string | null; +}) { + const queryClient = useQueryClient(); + const query = useQuery({ + queryKey: ["causal-ledger", sessionId], + queryFn: readCausalLedger, + enabled: Boolean(sessionId), + refetchInterval: 5_000, + }); + const inspection = inspectCausalCandidate(query.data ?? [], sessionId); + const experiments = query.data?.map((entry) => entry.experiment) ?? []; + const candidate = experiments.find( + (experiment) => experiment.experimentId === inspection?.experimentId, + ); + const activeEvaluation = candidate?.evaluation ? candidate : undefined; + const replayFromEvaluation = activeEvaluation + ? experiments.find( + (experiment) => + experiment.experimentId === activeEvaluation.execution.replayOf, + ) + : undefined; + const parentOfActive = candidate?.execution.replayOf + ? experiments.find( + (experiment) => + experiment.experimentId === candidate.execution.replayOf, + ) + : undefined; + const activeReplay = + !candidate?.evaluation && + parentOfActive?.intervention.approvedAt && + !candidate?.execution.sessionId.startsWith("replay-dispatch:") + ? candidate + : undefined; + const approvedProposal = replayFromEvaluation + ? experiments.find( + (experiment) => + experiment.experimentId === replayFromEvaluation.execution.replayOf, + ) + : activeReplay + ? parentOfActive + : experiments.find( + (experiment) => + experiment.execution.replayOf === inspection?.experimentId && + experiment.intervention.approvedAt, + ); + const dispatchReceipt = experiments.find( + (experiment) => + experiment.execution.replayOf === approvedProposal?.experimentId && + experiment.execution.sessionId.startsWith("replay-dispatch:"), + ); + const completedReplay = + replayFromEvaluation ?? + activeReplay ?? + experiments.find( + (experiment) => + experiment.execution.replayOf === approvedProposal?.experimentId && + !experiment.execution.sessionId.startsWith("replay-dispatch:"), + ); + const evaluation = + activeEvaluation ?? + experiments.find( + (experiment) => + experiment.execution.replayOf === completedReplay?.experimentId && + experiment.evaluation, + ); + const [draft, setDraft] = React.useState({ + failureFingerprint: "", + cause: "", + changedVariable: "", + successCriteria: "", + }); + const approveMutation = useMutation({ + mutationFn: async () => { + if (!candidate) throw new Error("The candidate is no longer available."); + const recordedAt = new Date().toISOString(); + const proposal = buildApprovedReplayProposal(candidate, draft, { + experimentId: `proposal:${candidate.experimentId}:${crypto.randomUUID()}`, + recordedAt, + }); + return appendCausalExperiment(proposal); + }, + onSuccess: async () => { + await queryClient.invalidateQueries({ queryKey: ["causal-ledger"] }); + toast.success("Controlled replay approved and sealed in the ledger."); + }, + onError: (error) => { + toast.error( + error instanceof Error + ? error.message + : "Could not approve the replay.", + ); + }, + }); + const dispatchMutation = useMutation({ + mutationFn: async () => { + if (!approvedProposal || !channelId) { + throw new Error( + "Open this candidate inside its channel to run the replay.", + ); + } + const result = await sendChannelMessage( + channelId, + buildReplayDispatchMessage(approvedProposal), + undefined, + undefined, + [agentPubkey], + ); + return appendCausalExperiment( + buildReplayDispatchReceipt( + approvedProposal, + result.eventId, + new Date().toISOString(), + ), + ); + }, + onSuccess: async () => { + await queryClient.invalidateQueries({ queryKey: ["causal-ledger"] }); + toast.success("Controlled replay dispatched to a fresh agent turn."); + }, + onError: (error) => { + toast.error( + error instanceof Error + ? error.message + : "Could not dispatch the replay.", + ); + }, + }); + if (!inspection) return null; + + return ( +
+
+
+

+ Durable candidate +

+

{inspection.task}

+
+ + {inspection.status} + +
+ + + + + +
+
+ +
+

+ Promotion gate +

+

{inspection.nextGate}

+
+
+
+ + {evaluation ? ( + + ) : completedReplay ? ( + { + await queryClient.invalidateQueries({ + queryKey: ["causal-ledger"], + }); + }} + /> + ) : approvedProposal ? ( +
+
+

+ Owner-approved replay +

+ + {dispatchReceipt ? "Replay running" : "Ready to run"} + +
+

+ Change only: {approvedProposal.intervention.changedVariable} +

+

+ Success: {approvedProposal.intervention.successCriteria} +

+

+ Approval does not prove the remedy. Buzz will keep this untested + until a linked replay and independent evidence produce a verdict. +

+ {!dispatchReceipt ? ( + + ) : ( +

+ Buzz linked the dispatch receipt. The independent verdict unlocks + when the agent turn reaches a terminal event. +

+ )} +
+ ) : candidate?.result.outcome === "untested" ? ( + + setDraft((current) => ({ ...current, [field]: value })) + } + onSubmit={() => approveMutation.mutate()} + /> + ) : null} +
+ ); +} + +function IndependentEvaluationForm({ + replay, + onSaved, +}: { + replay: import("../lib/causalLedger").CausalExperiment; + onSaved: () => Promise; +}) { + const [outcome, setOutcome] = React.useState< + "validated" | "rejected" | "inconclusive" + >("inconclusive"); + const [evidenceIds, setEvidenceIds] = React.useState( + replay.result.evidenceIds.join("\n"), + ); + const [rationale, setRationale] = React.useState(""); + const mutation = useMutation({ + mutationFn: () => { + const recordedAt = new Date().toISOString(); + return appendCausalExperiment( + buildIndependentEvaluation( + replay, + { outcome, evidenceIds, rationale }, + { + experimentId: `evaluation:${replay.experimentId}:${crypto.randomUUID()}`, + recordedAt, + }, + ), + ); + }, + onSuccess: async () => { + await onSaved(); + toast.success("Independent verdict sealed in the causal ledger."); + }, + onError: (error) => { + toast.error( + error instanceof Error ? error.message : "Could not save the verdict.", + ); + }, + }); + return ( +
{ + event.preventDefault(); + mutation.mutate(); + }} + > +
+

+ Independent evaluation +

+ Replay complete +
+ + +