merge: black box launch spine (lane 2/5)

This commit is contained in:
Sam Valladares 2026-06-24 17:34:54 -05:00
commit d5002c314b
43 changed files with 6857 additions and 18 deletions

View file

@ -90,6 +90,10 @@ pub mod fts;
pub mod memory;
pub mod storage;
/// Agent Black Box, Memory Receipts & Memory PRs — the cognitive flight
/// recorder, immune system, and reviewable-diff model for agent memory.
pub mod trace;
#[cfg(feature = "embeddings")]
#[cfg_attr(docsrs, doc(cfg(feature = "embeddings")))]
pub mod embeddings;
@ -160,6 +164,13 @@ pub use fsrs::{
// Configuration (vestige.toml output profiles / defaults)
pub use config::{CONFIG_FILE, OutputConfig, OutputDefaults, OutputProfile, VestigeConfig};
// Agent Black Box / Receipts / Memory PRs (the cognitive flight recorder)
pub use trace::{
classify_write, DecayRisk, MemoryPr, MemoryPrAction, MemoryPrKind, MemoryPrStatus,
MemoryTraceEvent, Receipt, ReceiptMutation, ReviewMode, RiskClass, RiskSignal, SuppressReason,
SuppressedReceiptEntry, WriteContext, WriteSource, HIGH_TRUST_FLOOR, LOW_CONFIDENCE_FLOOR,
};
// Storage layer
pub use storage::{
ClassificationResult,
@ -191,6 +202,7 @@ pub use storage::{
Result,
SchedulingState,
SearchQuery,
AgentRunSummary,
SmartIngestResult,
SourceUpsertOutcome,
SourceUpsertResult,

View file

@ -89,6 +89,11 @@ pub const MIGRATIONS: &[Migration] = &[
description: "#57 Source envelope: provenance columns + connector cursor checkpoints for idempotent external-source sync",
up: MIGRATION_V17_UP,
},
Migration {
version: 18,
description: "Agent Black Box + Memory Receipts + Memory PRs: replayable run traces, retrieval receipts, risk-gated brain-change review queue",
up: MIGRATION_V18_UP,
},
];
/// A database migration
@ -1029,6 +1034,105 @@ pub const MIGRATION_V17_ALTER_COLUMNS: &[&str] = &[
"ALTER TABLE knowledge_nodes ADD COLUMN source_author TEXT",
];
/// V18: Agent Black Box + Memory Receipts + Memory PRs.
///
/// Three append-only / review tables that turn Vestige into the *black box,
/// immune system, and cinematic debugger for agent memory*:
///
/// - `agent_traces` — one row per [`crate::trace::MemoryTraceEvent`], ordered by
/// `(run_id, seq)`. Append-only so a run replays exactly as the agent
/// experienced it. `payload` is the full serialized event; `event_type` and
/// `run_id` are denormalized for fast filtering and the `vestige://trace/{id}`
/// resource. `args_hash` (for `mcp.call`) is stored, never the raw args, so
/// traces can't leak prompt contents or secrets.
///
/// - `memory_receipts` — one row per retrieval receipt. `payload` holds the full
/// [`crate::trace::Receipt`]; the scalar columns (`trust_floor`, `decay_risk`)
/// are denormalized for list/sort without parsing JSON.
///
/// - `memory_prs` — the risk-gated review queue. A risky write (contradiction
/// vs high-trust, supersede/forget/merge/protect, sensitive topic, dream
/// consolidation, decay resurrection, low-confidence batch, weak-provenance
/// connector) lands here as `pending` instead of auto-committing. `diff` is the
/// structured before/after, `signals` is the self-explaining risk evidence,
/// `run_id` links the PR back to the black-box trace that produced it.
///
/// `memory_id` / `run_id` references are intentionally *not* foreign keys to
/// `knowledge_nodes`: forgetting or superseding a memory must never erase the
/// audit trail of the trace, receipt, or PR that touched it (same
/// audit-preserving stance as V15's composition tables).
const MIGRATION_V18_UP: &str = r#"
-- Black-box trace events: append-only, ordered by (run_id, seq).
CREATE TABLE IF NOT EXISTS agent_traces (
id TEXT PRIMARY KEY,
run_id TEXT NOT NULL,
seq INTEGER NOT NULL,
event_type TEXT NOT NULL, -- mcp.call | memory.retrieve | ...
tool TEXT, -- denormalized for mcp.call rows
payload TEXT NOT NULL, -- full serialized MemoryTraceEvent (JSON)
at INTEGER NOT NULL, -- wall-clock millis
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_agent_traces_run ON agent_traces(run_id, seq);
CREATE INDEX IF NOT EXISTS idx_agent_traces_type ON agent_traces(event_type);
CREATE INDEX IF NOT EXISTS idx_agent_traces_at ON agent_traces(at);
-- One row per agent run, for the Black Box run list (denormalized roll-up).
CREATE TABLE IF NOT EXISTS agent_runs (
run_id TEXT PRIMARY KEY,
first_tool TEXT,
event_count INTEGER NOT NULL DEFAULT 0,
retrieved_count INTEGER NOT NULL DEFAULT 0,
suppressed_count INTEGER NOT NULL DEFAULT 0,
write_count INTEGER NOT NULL DEFAULT 0,
veto_count INTEGER NOT NULL DEFAULT 0,
started_at INTEGER NOT NULL, -- millis of first event
last_at INTEGER NOT NULL, -- millis of latest event
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_agent_runs_last_at ON agent_runs(last_at DESC);
-- Retrieval receipts (the "nutrition label" for a piece of agent memory).
CREATE TABLE IF NOT EXISTS memory_receipts (
receipt_id TEXT PRIMARY KEY,
run_id TEXT, -- links to the trace, if any
tool TEXT,
query TEXT,
retrieved_count INTEGER NOT NULL DEFAULT 0,
suppressed_count INTEGER NOT NULL DEFAULT 0,
trust_floor REAL NOT NULL DEFAULT 0,
decay_risk TEXT NOT NULL DEFAULT 'low',
payload TEXT NOT NULL, -- full serialized Receipt (JSON)
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_memory_receipts_run ON memory_receipts(run_id);
CREATE INDEX IF NOT EXISTS idx_memory_receipts_created_at ON memory_receipts(created_at DESC);
-- Memory PRs: the risk-gated review queue for brain changes.
CREATE TABLE IF NOT EXISTS memory_prs (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL, -- new_fact | contradiction_detected | ...
status TEXT NOT NULL DEFAULT 'pending',
title TEXT NOT NULL,
subject_id TEXT, -- the memory this PR concerns, if any
run_id TEXT, -- the run that produced it
diff TEXT NOT NULL DEFAULT '{}', -- structured before/after (JSON)
signals TEXT NOT NULL DEFAULT '[]', -- self-explaining RiskSignal[] (JSON)
decision TEXT, -- promote | merge | supersede | ...
created_at TEXT NOT NULL,
decided_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_memory_prs_status ON memory_prs(status);
CREATE INDEX IF NOT EXISTS idx_memory_prs_kind ON memory_prs(kind);
CREATE INDEX IF NOT EXISTS idx_memory_prs_created_at ON memory_prs(created_at DESC);
UPDATE schema_version SET version = 18, applied_at = datetime('now');
"#;
/// Apply pending migrations
pub fn apply_migrations(conn: &rusqlite::Connection) -> rusqlite::Result<u32> {
let current_version = get_current_version(conn)?;
@ -1109,9 +1213,10 @@ mod tests {
// 1. schema_version advanced to the latest migration
let version = get_current_version(&conn).expect("read schema_version");
let latest = MIGRATIONS.last().unwrap().version;
assert_eq!(
version, 17,
"schema_version must be 17 after all migrations"
version, latest,
"schema_version must be the latest migration after all migrations"
);
// 2. knowledge_edges is gone (V11 drops it)
@ -1236,7 +1341,11 @@ mod tests {
// After replaying from V10, the schema advances to the latest version.
let version = get_current_version(&conn).expect("read schema_version");
assert_eq!(version, 17, "schema_version back at latest after replay");
assert_eq!(
version,
MIGRATIONS.last().unwrap().version,
"schema_version back at latest after replay"
);
}
#[test]
@ -1310,7 +1419,11 @@ mod tests {
// V16 uses CREATE TABLE IF NOT EXISTS and idempotent ALTER handling.
apply_migrations(&conn).expect("V16 replay must be idempotent");
let version = get_current_version(&conn).expect("read version");
assert_eq!(version, 17, "schema_version must be latest after replay");
assert_eq!(
version,
MIGRATIONS.last().unwrap().version,
"schema_version must be latest after replay"
);
}
#[test]
@ -1400,7 +1513,11 @@ mod tests {
.expect("rewind to 16");
apply_migrations(&conn).expect("V17 replay must be idempotent");
let version = get_current_version(&conn).expect("read version");
assert_eq!(version, 17, "schema_version must be 17 after replay");
assert_eq!(
version,
MIGRATIONS.last().unwrap().version,
"schema_version must be latest after replay"
);
}
#[test]

View file

@ -6,6 +6,7 @@ mod memory_store;
mod migrations;
mod portable;
mod sqlite;
mod trace_store;
pub use memory_store::{
ClassificationResult, Domain, HealthStatus, LocalMemoryStore, MemoryEdge, MemoryRecord,
@ -25,6 +26,7 @@ pub use sqlite::{
SmartIngestResult, SourceUpsertOutcome, SourceUpsertResult, SqliteMemoryStore,
StateTransitionRecord, StorageError,
};
pub use trace_store::AgentRunSummary;
/// Backwards-compatibility alias. Retained until Phase 4 completes so every
/// existing `Arc<Storage>` call site keeps compiling. Scheduled for removal

View file

@ -302,8 +302,11 @@ const VESTIGE_DISABLE_VECTOR_SEARCH: &str = "VESTIGE_DISABLE_VECTOR_SEARCH";
/// so the MCP layer can use `Arc<Storage>` instead of `Arc<Mutex<Storage>>`.
pub struct SqliteMemoryStore {
db_path: PathBuf,
writer: Mutex<Connection>,
reader: Mutex<Connection>,
// `pub(crate)` so the sibling `trace_store` module (Black Box / Receipts /
// Memory PRs CRUD) can lock the same writer/reader connections and follow
// the established store idiom without duplicating connection management.
pub(crate) writer: Mutex<Connection>,
pub(crate) reader: Mutex<Connection>,
scheduler: Mutex<FSRSScheduler>,
#[cfg(feature = "embeddings")]
embedding_service: EmbeddingService,
@ -1782,6 +1785,61 @@ impl SqliteMemoryStore {
.ok_or_else(|| StorageError::NotFound(id.to_string()))
}
/// Release a memory from quarantine **unconditionally** (no labile-window
/// limit), used when a Memory PR is approved.
///
/// Unlike [`Self::reverse_suppression`] (which models a time-bounded "undo"
/// of an active-forgetting decision and refuses after the labile window),
/// approving a quarantined risky write is an explicit reviewer decision that
/// must always restore the memory's retrieval influence — even days later.
/// Fully clears the suppression (count → 0, `suppressed_at` → NULL) and
/// restores strengths. A no-op (returns the node) if it isn't suppressed.
pub fn release_quarantine(&self, id: &str) -> Result<KnowledgeNode> {
let node = self
.get_node(id)?
.ok_or_else(|| StorageError::NotFound(id.to_string()))?;
if node.suppression_count == 0 && node.suppressed_at.is_none() {
// Nothing to release — idempotent.
return Ok(node);
}
{
let writer = self
.writer
.lock()
.map_err(|_| StorageError::Init("Writer lock poisoned".into()))?;
writer.execute(
"UPDATE knowledge_nodes SET
suppression_count = 0,
suppressed_at = NULL,
retrieval_strength = MIN(1.0, retrieval_strength + 0.15),
retention_strength = MIN(1.0, retention_strength + 0.10),
stability = stability * 1.25
WHERE id = ?1",
params![id],
)?;
}
let _ = self.log_access(id, "release_quarantine");
self.get_node(id)?
.ok_or_else(|| StorageError::NotFound(id.to_string()))
}
/// Test-only: backdate a node's `suppressed_at` to simulate a suppression
/// that happened long ago (e.g. to verify release works past the labile
/// window). `pub(crate)` so sibling test modules can reach it.
#[cfg(test)]
pub(crate) fn set_suppressed_at_for_test(&self, id: &str, when: DateTime<Utc>) {
if let Ok(writer) = self.writer.lock() {
let _ = writer.execute(
"UPDATE knowledge_nodes SET suppressed_at = ?1 WHERE id = ?2",
params![when.to_rfc3339(), id],
);
}
}
/// Count memories currently in a suppressed state (suppression_count > 0).
pub fn count_suppressed(&self) -> Result<usize> {
let reader = self

View file

@ -0,0 +1,732 @@
//! # Black Box / Receipts / Memory PRs — persistence
//!
//! CRUD for the three V18 tables (`agent_traces` + `agent_runs`,
//! `memory_receipts`, `memory_prs`) on [`SqliteMemoryStore`]. The pure data
//! model lives in [`crate::trace`]; this file is the storage half of the
//! Black Box, immune system, and cinematic debugger for agent memory.
//!
//! Every method follows the established store idiom: lock the writer/reader
//! `Mutex<Connection>`, `params![]`-bind, store timestamps as RFC3339 (and
//! event millis as INTEGER), serialize structured fields with `serde_json`, and
//! map rows back through a small closure.
use chrono::Utc;
use rusqlite::{params, OptionalExtension};
use uuid::Uuid;
use super::sqlite::SqliteMemoryStore;
use super::{Result, StorageError};
use crate::trace::{MemoryPr, MemoryPrAction, MemoryPrStatus, MemoryTraceEvent, Receipt};
/// A roll-up summary of one agent run, for the Black Box run list.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
pub struct AgentRunSummary {
/// The run id.
pub run_id: String,
/// The first tool invoked in the run (the run's "entry point").
pub first_tool: Option<String>,
/// Total events recorded.
pub event_count: i64,
/// Memories retrieved across the run.
pub retrieved_count: i64,
/// Memories suppressed across the run.
pub suppressed_count: i64,
/// Memory writes across the run.
pub write_count: i64,
/// Sanhedrin vetoes across the run.
pub veto_count: i64,
/// Millis of the first event.
pub started_at: i64,
/// Millis of the most recent event.
pub last_at: i64,
}
impl SqliteMemoryStore {
// ========================================================================
// BLACK BOX — trace events + run roll-up
// ========================================================================
/// Append one trace event to a run (append-only) and update the run
/// roll-up. Returns the assigned sequence number within the run.
///
/// `seq` is `MAX(seq)+1` for the run, computed under the writer lock so a
/// run's events stay totally ordered even under concurrent tool calls.
pub fn append_trace_event(&self, event: &MemoryTraceEvent) -> Result<i64> {
let now = Utc::now();
let run_id = event.run_id().to_string();
let event_type = event.kind();
let at = event.at();
let payload = serde_json::to_string(event)
.map_err(|e| StorageError::Init(format!("trace event serialize: {e}")))?;
let tool = match event {
MemoryTraceEvent::McpCall { tool, .. } => Some(tool.clone()),
_ => None,
};
// Roll-up deltas this event contributes.
let (d_retrieved, d_suppressed, d_write, d_veto) = match event {
MemoryTraceEvent::MemoryRetrieve { ids, .. } => (ids.len() as i64, 0, 0, 0),
MemoryTraceEvent::MemorySuppress { .. } => (0, 1, 0, 0),
MemoryTraceEvent::MemoryWrite { .. } => (0, 0, 1, 0),
MemoryTraceEvent::SanhedrinVeto { .. } => (0, 0, 0, 1),
_ => (0, 0, 0, 0),
};
let writer = self
.writer
.lock()
.map_err(|_| StorageError::Init("Writer lock poisoned".into()))?;
let seq: i64 = writer
.query_row(
"SELECT COALESCE(MAX(seq), -1) + 1 FROM agent_traces WHERE run_id = ?1",
params![run_id],
|r| r.get(0),
)
.unwrap_or(0);
writer.execute(
"INSERT INTO agent_traces (id, run_id, seq, event_type, tool, payload, at, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![
Uuid::new_v4().to_string(),
run_id,
seq,
event_type,
tool,
payload,
at,
now.to_rfc3339(),
],
)?;
// Upsert the run roll-up. On first event the row is created with the
// event's tool as the entry point; subsequent events accumulate counts
// and advance `last_at`.
writer.execute(
"INSERT INTO agent_runs (run_id, first_tool, event_count, retrieved_count,
suppressed_count, write_count, veto_count, started_at, last_at, created_at)
VALUES (?1, ?2, 1, ?3, ?4, ?5, ?6, ?7, ?7, ?8)
ON CONFLICT(run_id) DO UPDATE SET
first_tool = COALESCE(agent_runs.first_tool, excluded.first_tool),
event_count = agent_runs.event_count + 1,
retrieved_count = agent_runs.retrieved_count + ?3,
suppressed_count = agent_runs.suppressed_count + ?4,
write_count = agent_runs.write_count + ?5,
veto_count = agent_runs.veto_count + ?6,
last_at = MAX(agent_runs.last_at, ?7)",
params![
run_id,
tool,
d_retrieved,
d_suppressed,
d_write,
d_veto,
at,
now.to_rfc3339(),
],
)?;
Ok(seq)
}
/// Fetch every event of a run, in sequence order. The black-box replay.
pub fn get_trace(&self, run_id: &str) -> Result<Vec<MemoryTraceEvent>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
let mut stmt = reader.prepare(
"SELECT payload FROM agent_traces WHERE run_id = ?1 ORDER BY seq ASC",
)?;
let rows = stmt.query_map(params![run_id], |row| {
let payload: String = row.get(0)?;
Ok(payload)
})?;
let mut out = Vec::new();
for r in rows {
let payload = r?;
if let Ok(ev) = serde_json::from_str::<MemoryTraceEvent>(&payload) {
out.push(ev);
}
}
Ok(out)
}
/// List recent runs, newest activity first.
pub fn list_agent_runs(&self, limit: usize) -> Result<Vec<AgentRunSummary>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
let mut stmt = reader.prepare(
"SELECT run_id, first_tool, event_count, retrieved_count, suppressed_count,
write_count, veto_count, started_at, last_at
FROM agent_runs ORDER BY last_at DESC LIMIT ?1",
)?;
let rows = stmt.query_map(params![limit as i64], Self::row_to_run_summary)?;
let mut out = Vec::new();
for r in rows {
out.push(r?);
}
Ok(out)
}
/// Fetch one run summary.
pub fn get_agent_run(&self, run_id: &str) -> Result<Option<AgentRunSummary>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
reader
.query_row(
"SELECT run_id, first_tool, event_count, retrieved_count, suppressed_count,
write_count, veto_count, started_at, last_at
FROM agent_runs WHERE run_id = ?1",
params![run_id],
Self::row_to_run_summary,
)
.optional()
.map_err(StorageError::from)
}
fn row_to_run_summary(row: &rusqlite::Row) -> rusqlite::Result<AgentRunSummary> {
Ok(AgentRunSummary {
run_id: row.get("run_id")?,
first_tool: row.get("first_tool").ok().flatten(),
event_count: row.get("event_count")?,
retrieved_count: row.get("retrieved_count")?,
suppressed_count: row.get("suppressed_count")?,
write_count: row.get("write_count")?,
veto_count: row.get("veto_count")?,
started_at: row.get("started_at")?,
last_at: row.get("last_at")?,
})
}
// ========================================================================
// MEMORY RECEIPTS
// ========================================================================
/// Persist a retrieval receipt. `run_id`/`tool`/`query` are denormalized
/// context for the dashboard; the full [`Receipt`] is stored as JSON.
pub fn save_receipt(
&self,
receipt: &Receipt,
run_id: Option<&str>,
tool: Option<&str>,
query: Option<&str>,
) -> Result<()> {
let payload = serde_json::to_string(receipt)
.map_err(|e| StorageError::Init(format!("receipt serialize: {e}")))?;
let writer = self
.writer
.lock()
.map_err(|_| StorageError::Init("Writer lock poisoned".into()))?;
writer.execute(
"INSERT OR REPLACE INTO memory_receipts
(receipt_id, run_id, tool, query, retrieved_count, suppressed_count,
trust_floor, decay_risk, payload, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
receipt.receipt_id,
run_id,
tool,
query,
receipt.retrieved.len() as i64,
receipt.suppressed.len() as i64,
receipt.trust_floor,
receipt.decay_risk.as_str(),
payload,
Utc::now().to_rfc3339(),
],
)?;
Ok(())
}
/// Fetch one receipt by id.
pub fn get_receipt(&self, receipt_id: &str) -> Result<Option<Receipt>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
let payload: Option<String> = reader
.query_row(
"SELECT payload FROM memory_receipts WHERE receipt_id = ?1",
params![receipt_id],
|row| row.get(0),
)
.optional()?;
Ok(payload.and_then(|p| serde_json::from_str(&p).ok()))
}
/// List recent receipts, newest first.
pub fn list_receipts(&self, limit: usize) -> Result<Vec<Receipt>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
let mut stmt = reader.prepare(
"SELECT payload FROM memory_receipts ORDER BY created_at DESC LIMIT ?1",
)?;
let rows = stmt.query_map(params![limit as i64], |row| {
let p: String = row.get(0)?;
Ok(p)
})?;
let mut out = Vec::new();
for r in rows {
if let Ok(rc) = serde_json::from_str::<Receipt>(&r?) {
out.push(rc);
}
}
Ok(out)
}
/// List the receipts belonging to one run, newest first (B5). The Black Box
/// receipts panel uses this so the receipts it shows actually belong to the
/// selected run, not the global latest.
pub fn list_receipts_for_run(&self, run_id: &str, limit: usize) -> Result<Vec<Receipt>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
let mut stmt = reader.prepare(
"SELECT payload FROM memory_receipts WHERE run_id = ?1
ORDER BY created_at DESC LIMIT ?2",
)?;
let rows = stmt.query_map(params![run_id, limit as i64], |row| {
let p: String = row.get(0)?;
Ok(p)
})?;
let mut out = Vec::new();
for r in rows {
if let Ok(rc) = serde_json::from_str::<Receipt>(&r?) {
out.push(rc);
}
}
Ok(out)
}
// ========================================================================
// MEMORY PRs — the risk-gated review queue
// ========================================================================
/// Open (insert) a Memory PR.
pub fn save_memory_pr(&self, pr: &MemoryPr) -> Result<()> {
let diff = serde_json::to_string(&pr.diff).unwrap_or_else(|_| "{}".to_string());
let signals = serde_json::to_string(&pr.signals).unwrap_or_else(|_| "[]".to_string());
let writer = self
.writer
.lock()
.map_err(|_| StorageError::Init("Writer lock poisoned".into()))?;
writer.execute(
"INSERT OR REPLACE INTO memory_prs
(id, kind, status, title, subject_id, run_id, diff, signals,
decision, created_at, decided_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
params![
pr.id,
pr.kind.as_str(),
pr.status.as_str(),
pr.title,
pr.subject_id,
pr.run_id,
diff,
signals,
pr.decision
.and_then(|d| serde_json::to_value(d).ok())
.and_then(|v| v.as_str().map(|s| s.to_string())),
pr.created_at,
pr.decided_at,
],
)?;
Ok(())
}
/// Fetch one Memory PR by id.
pub fn get_memory_pr(&self, id: &str) -> Result<Option<MemoryPr>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
reader
.query_row(
"SELECT id, kind, status, title, subject_id, run_id, diff, signals,
decision, created_at, decided_at
FROM memory_prs WHERE id = ?1",
params![id],
Self::row_to_memory_pr,
)
.optional()
.map_err(StorageError::from)
}
/// List Memory PRs, optionally filtered by status, newest first.
pub fn list_memory_prs(
&self,
status: Option<MemoryPrStatus>,
limit: usize,
) -> Result<Vec<MemoryPr>> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
let (sql, with_filter) = match status {
Some(_) => (
"SELECT id, kind, status, title, subject_id, run_id, diff, signals,
decision, created_at, decided_at
FROM memory_prs WHERE status = ?1 ORDER BY created_at DESC LIMIT ?2",
true,
),
None => (
"SELECT id, kind, status, title, subject_id, run_id, diff, signals,
decision, created_at, decided_at
FROM memory_prs ORDER BY created_at DESC LIMIT ?1",
false,
),
};
let mut stmt = reader.prepare(sql)?;
let mut out = Vec::new();
if with_filter {
let st = status.unwrap();
let rows =
stmt.query_map(params![st.as_str(), limit as i64], Self::row_to_memory_pr)?;
for r in rows {
out.push(r?);
}
} else {
let rows = stmt.query_map(params![limit as i64], Self::row_to_memory_pr)?;
for r in rows {
out.push(r?);
}
}
Ok(out)
}
/// Count pending Memory PRs (for the nav badge).
pub fn count_pending_memory_prs(&self) -> Result<i64> {
let reader = self
.reader
.lock()
.map_err(|_| StorageError::Init("Reader lock poisoned".into()))?;
let n: i64 = reader
.query_row(
"SELECT COUNT(*) FROM memory_prs WHERE status = 'pending'",
[],
|r| r.get(0),
)
.unwrap_or(0);
Ok(n)
}
/// Record a decision on a Memory PR, moving it out of `pending`. Returns the
/// updated PR. `AskAgentWhy` is read-only and never reaches here.
pub fn decide_memory_pr(&self, id: &str, action: MemoryPrAction) -> Result<MemoryPr> {
let new_status = action.resulting_status().ok_or_else(|| {
StorageError::Init("ask_agent_why is read-only and decides nothing".into())
})?;
let decision = serde_json::to_value(action)
.ok()
.and_then(|v| v.as_str().map(|s| s.to_string()))
.unwrap_or_default();
let now = Utc::now().to_rfc3339();
{
let writer = self
.writer
.lock()
.map_err(|_| StorageError::Init("Writer lock poisoned".into()))?;
let changed = writer.execute(
"UPDATE memory_prs SET status = ?1, decision = ?2, decided_at = ?3 WHERE id = ?4",
params![new_status.as_str(), decision, now, id],
)?;
if changed == 0 {
return Err(StorageError::NotFound(id.to_string()));
}
}
self.get_memory_pr(id)?
.ok_or_else(|| StorageError::NotFound(id.to_string()))
}
fn row_to_memory_pr(row: &rusqlite::Row) -> rusqlite::Result<MemoryPr> {
let kind_s: String = row.get("kind")?;
let status_s: String = row.get("status")?;
let diff_s: String = row.get("diff")?;
let signals_s: String = row.get("signals")?;
let decision_s: Option<String> = row.get("decision").ok().flatten();
let kind = crate::trace::MemoryPrKind::from_label(&kind_s)
.unwrap_or(crate::trace::MemoryPrKind::NewFact);
let status = serde_json::from_value(serde_json::Value::String(status_s))
.unwrap_or(MemoryPrStatus::Pending);
let diff: serde_json::Value = serde_json::from_str(&diff_s).unwrap_or(serde_json::json!({}));
let signals = serde_json::from_str(&signals_s).unwrap_or_default();
let decision = decision_s
.and_then(|s| serde_json::from_value(serde_json::Value::String(s)).ok());
Ok(MemoryPr {
id: row.get("id")?,
kind,
status,
title: row.get("title")?,
diff,
signals,
subject_id: row.get("subject_id").ok().flatten(),
run_id: row.get("run_id").ok().flatten(),
created_at: row.get("created_at")?,
decided_at: row.get("decided_at").ok().flatten(),
decision,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::trace::{
DecayRisk, MemoryPrKind, MemoryTraceEvent, Receipt, RiskSignal, SuppressReason,
SuppressedReceiptEntry,
};
fn store() -> SqliteMemoryStore {
// Temp-file store for isolated, fast tests (mirrors the existing
// sqlite.rs test helpers; there is no in-memory constructor).
let dir = tempfile::tempdir().unwrap();
SqliteMemoryStore::new(Some(dir.path().join("trace_test.db"))).expect("test store")
}
#[test]
fn trace_append_orders_and_rolls_up() {
let s = store();
let run = "run_abc";
s.append_trace_event(&MemoryTraceEvent::McpCall {
run_id: run.into(),
tool: "deep_reference".into(),
args_hash: "h".into(),
at: 100,
})
.unwrap();
let mut activation = std::collections::BTreeMap::new();
activation.insert("m1".to_string(), 0.9);
s.append_trace_event(&MemoryTraceEvent::MemoryRetrieve {
run_id: run.into(),
ids: vec!["m1".into(), "m2".into()],
activation,
at: 110,
})
.unwrap();
s.append_trace_event(&MemoryTraceEvent::MemorySuppress {
run_id: run.into(),
id: "m3".into(),
reason: SuppressReason::Contradicted,
at: 120,
})
.unwrap();
let events = s.get_trace(run).unwrap();
assert_eq!(events.len(), 3);
assert_eq!(events[0].kind(), "mcp.call");
assert_eq!(events[2].kind(), "memory.suppress");
let summary = s.get_agent_run(run).unwrap().unwrap();
assert_eq!(summary.first_tool.as_deref(), Some("deep_reference"));
assert_eq!(summary.event_count, 3);
assert_eq!(summary.retrieved_count, 2);
assert_eq!(summary.suppressed_count, 1);
assert_eq!(summary.started_at, 100);
assert_eq!(summary.last_at, 120);
let runs = s.list_agent_runs(10).unwrap();
assert_eq!(runs.len(), 1);
assert_eq!(runs[0].run_id, run);
}
#[test]
fn receipt_roundtrips() {
let s = store();
let receipt = Receipt {
receipt_id: "r_2026_06_22_abc".into(),
retrieved: vec!["m1".into(), "m2".into()],
suppressed: vec![SuppressedReceiptEntry::new("m3", SuppressReason::LowTrust)],
activation_path: vec!["a -> b".into()],
trust_floor: 0.62,
decay_risk: DecayRisk::Medium,
mutations: vec![],
};
s.save_receipt(&receipt, Some("run_abc"), Some("search"), Some("q"))
.unwrap();
let got = s.get_receipt("r_2026_06_22_abc").unwrap().unwrap();
assert_eq!(got, receipt);
assert_eq!(s.list_receipts(10).unwrap().len(), 1);
}
#[test]
fn receipts_are_listable_per_run_b5() {
let s = store();
let mk = |id: &str| Receipt {
receipt_id: id.into(),
retrieved: vec!["m1".into()],
suppressed: vec![],
activation_path: vec![],
trust_floor: 0.9,
decay_risk: DecayRisk::Low,
mutations: vec![],
};
s.save_receipt(&mk("r_a1"), Some("run_a"), Some("search"), None)
.unwrap();
s.save_receipt(&mk("r_a2"), Some("run_a"), Some("search"), None)
.unwrap();
s.save_receipt(&mk("r_b1"), Some("run_b"), Some("search"), None)
.unwrap();
let run_a = s.list_receipts_for_run("run_a", 10).unwrap();
assert_eq!(run_a.len(), 2, "run_a has exactly its 2 receipts");
assert!(run_a.iter().all(|r| r.receipt_id.starts_with("r_a")));
let run_b = s.list_receipts_for_run("run_b", 10).unwrap();
assert_eq!(run_b.len(), 1, "run_b has only its own receipt");
assert_eq!(run_b[0].receipt_id, "r_b1");
// Global list still sees all three.
assert_eq!(s.list_receipts(10).unwrap().len(), 3);
}
#[test]
fn memory_pr_lifecycle() {
let s = store();
let pr = MemoryPr {
id: "pr_1".into(),
kind: MemoryPrKind::ContradictionDetected,
status: MemoryPrStatus::Pending,
title: "Agent wants to overwrite a high-trust fact".into(),
diff: serde_json::json!({"before": "x", "after": "y"}),
signals: vec![RiskSignal {
code: "contradicts_high_trust".into(),
detail: "Contradicts trust 0.9.".into(),
}],
subject_id: Some("m_old".into()),
run_id: Some("run_abc".into()),
created_at: Utc::now().to_rfc3339(),
decided_at: None,
decision: None,
};
s.save_memory_pr(&pr).unwrap();
assert_eq!(s.count_pending_memory_prs().unwrap(), 1);
let pending = s
.list_memory_prs(Some(MemoryPrStatus::Pending), 10)
.unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].signals[0].code, "contradicts_high_trust");
let decided = s.decide_memory_pr("pr_1", MemoryPrAction::Promote).unwrap();
assert_eq!(decided.status, MemoryPrStatus::Promoted);
assert_eq!(decided.decision, Some(MemoryPrAction::Promote));
assert!(decided.decided_at.is_some());
assert_eq!(s.count_pending_memory_prs().unwrap(), 0);
}
#[test]
fn promote_releases_a_quarantined_memory_end_to_end() {
// B1 regression: the full quarantine→release cycle at the storage layer.
// gate_writes suppresses a risky write; an accept action must reverse it.
let s = store();
let node = s
.ingest(crate::IngestInput {
content: "Risky write that got quarantined.".to_string(),
node_type: "fact".to_string(),
..Default::default()
})
.expect("ingest");
assert_eq!(node.suppression_count, 0, "fresh node not suppressed");
// Quarantine it (what gate_writes does for a risky write).
let suppressed = s.suppress_memory(&node.id).expect("suppress");
assert_eq!(
suppressed.suppression_count, 1,
"quarantined write is suppressed (held out of retrieval)"
);
// Promote = release. (The action releases_memory() == true; the handler
// calls release_quarantine on the subject.)
assert!(crate::MemoryPrAction::Promote.releases_memory());
let released = s
.release_quarantine(&node.id)
.expect("release quarantine");
assert_eq!(
released.suppression_count, 0,
"promoting the PR must release the memory — not leave it suppressed"
);
assert!(
released.suppressed_at.is_none(),
"release must clear suppressed_at"
);
}
#[test]
fn release_quarantine_works_past_the_labile_window_c1() {
// C1: a PR reviewed LATE (past the 24h active-forgetting labile window)
// must still release the memory. reverse_suppression refuses after the
// window; release_quarantine must not.
let s = store();
let node = s
.ingest(crate::IngestInput {
content: "Risky write quarantined and reviewed days later.".to_string(),
node_type: "fact".to_string(),
..Default::default()
})
.expect("ingest");
s.suppress_memory(&node.id).expect("suppress");
// Backdate suppressed_at to 100h ago — well past any labile window.
s.set_suppressed_at_for_test(&node.id, chrono::Utc::now() - chrono::Duration::hours(100));
// reverse_suppression refuses (window expired)...
assert!(
s.reverse_suppression(&node.id, 24).is_err(),
"reverse_suppression must refuse past the labile window"
);
// ...but release_quarantine still releases it.
let released = s.release_quarantine(&node.id).expect("release past window");
assert_eq!(released.suppression_count, 0);
assert!(released.suppressed_at.is_none());
}
#[test]
fn release_quarantine_is_idempotent_on_unsuppressed() {
let s = store();
let node = s
.ingest(crate::IngestInput {
content: "Never suppressed.".to_string(),
node_type: "fact".to_string(),
..Default::default()
})
.expect("ingest");
// No-op when not suppressed — must not error.
let same = s.release_quarantine(&node.id).expect("idempotent release");
assert_eq!(same.suppression_count, 0);
}
#[test]
fn ask_agent_why_is_not_a_decision() {
let s = store();
let pr = MemoryPr {
id: "pr_2".into(),
kind: MemoryPrKind::NewFact,
status: MemoryPrStatus::Pending,
title: "t".into(),
diff: serde_json::json!({}),
signals: vec![],
subject_id: None,
run_id: None,
created_at: Utc::now().to_rfc3339(),
decided_at: None,
decision: None,
};
s.save_memory_pr(&pr).unwrap();
assert!(s
.decide_memory_pr("pr_2", MemoryPrAction::AskAgentWhy)
.is_err());
// Still pending.
assert_eq!(s.count_pending_memory_prs().unwrap(), 1);
}
}

View file

@ -0,0 +1,356 @@
//! # Agent Black Box, Receipts & Memory PRs — the cognitive flight recorder
//!
//! This module holds the **pure** data model and classification logic for three
//! tightly-related capabilities that together make Vestige *the black box,
//! immune system, and cinematic debugger for agent memory*:
//!
//! 1. **Agent Black Box** — a replayable trace of everything an agent run did to
//! memory: prompt → retrieved → suppressed → activated edges → tool calls →
//! writes → contradictions → vetoes → dream consolidation → final answer.
//! The event model is [`MemoryTraceEvent`].
//!
//! 2. **Memory Receipts** — every important retrieval returns a structured
//! [`Receipt`]: what was retrieved, what was suppressed and why, the
//! activation path that surfaced it, the trust floor, the decay risk, and any
//! mutations. A receipt is the "nutrition label" for a piece of agent memory.
//!
//! 3. **Memory PRs** — changes to an agent's *brain* are reviewed like changes
//! to code. Ordinary context auto-commits (and always leaves a receipt), but
//! risky writes — contradictions against high-trust memory, supersede / forget
//! / merge / protect, identity / preference / workflow / positioning facts,
//! permission / auth / security / money / legal facts, dream consolidation
//! proposals, decay-below-threshold resurrection, low-confidence batch
//! imports, and weak-provenance connector writes — open a reviewable
//! [`MemoryPr`]. The gating decision is [`classify_write`].
//!
//! ## Design north star (shared with [`crate::advanced::merge_supersede`])
//!
//! - **append-only** — trace events are never mutated, only appended, so a run
//! replays exactly as the agent experienced it.
//! - **self-explaining** — every gated write carries the [`RiskSignal`]s that
//! explain *why* it needs review, in plain language.
//! - **opt-in friction** — the default [`ReviewMode::RiskGated`] keeps ordinary
//! memory frictionless and only opens a PR when the agent tries to rewrite its
//! own brain. [`ReviewMode::Fast`] never gates; [`ReviewMode::Paranoid`] gates
//! every write.
//! - **DB-free** — this module is pure logic so it is unit-testable without a
//! database. Persistence (the `agent_traces`, `memory_receipts`, and
//! `memory_prs` tables) lives in [`crate::storage`].
//!
//! The killer line, made literal by [`classify_write`]:
//!
//! > Vestige auto-remembers ordinary context, but opens a Memory PR when the
//! > agent tries to rewrite its own brain.
use serde::{Deserialize, Serialize};
mod receipt;
mod review;
pub use receipt::{DecayRisk, Receipt, ReceiptMutation, SuppressedReceiptEntry};
pub use review::{
classify_write, MemoryPr, MemoryPrAction, MemoryPrKind, MemoryPrStatus, ReviewMode, RiskClass,
RiskSignal, WriteContext, HIGH_TRUST_FLOOR, LOW_CONFIDENCE_FLOOR,
};
// ============================================================================
// TRACE EVENTS — the black-box flight recorder
// ============================================================================
/// One append-only event in an agent run's black-box trace.
///
/// Mirrors the TypeScript `MemoryTraceEvent` union exactly (tagged on `type`,
/// camelCase fields) so the dashboard, the `vestige://trace/{runId}` MCP
/// resource, and the exported `.vestige-trace.json` all speak one schema.
///
/// ```ts
/// type MemoryTraceEvent =
/// | { type: "mcp.call"; runId; tool; argsHash; at }
/// | { type: "memory.retrieve"; runId; ids; activation; at }
/// | { type: "memory.suppress"; runId; id; reason }
/// | { type: "memory.write"; runId; id; diff; source }
/// | { type: "sanhedrin.veto"; runId; claim; evidenceIds; confidence }
/// | { type: "dream.patch"; runId; proposalIds; at };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "type")]
pub enum MemoryTraceEvent {
/// An MCP tool was invoked. The args are stored as a hash (not the raw
/// payload) so traces never leak prompt contents or secrets.
#[serde(rename = "mcp.call")]
McpCall {
#[serde(rename = "runId")]
run_id: String,
tool: String,
#[serde(rename = "argsHash")]
args_hash: String,
at: i64,
},
/// Memories were retrieved, with per-id spreading-activation strength so the
/// graph replay can pulse exactly the nodes the agent saw, at their weight.
#[serde(rename = "memory.retrieve")]
MemoryRetrieve {
#[serde(rename = "runId")]
run_id: String,
ids: Vec<String>,
activation: std::collections::BTreeMap<String, f64>,
at: i64,
},
/// A memory that *would* have surfaced was suppressed, with the reason —
/// this is the "what the agent chose NOT to use" channel.
#[serde(rename = "memory.suppress")]
MemorySuppress {
#[serde(rename = "runId")]
run_id: String,
id: String,
reason: SuppressReason,
#[serde(default)]
at: i64,
},
/// A memory was written / strengthened. `diff` is an opaque JSON description
/// of the change; `source` records who caused it.
#[serde(rename = "memory.write")]
MemoryWrite {
#[serde(rename = "runId")]
run_id: String,
id: String,
diff: serde_json::Value,
source: WriteSource,
#[serde(default)]
at: i64,
},
/// A contradiction was detected between memories during a run — its own
/// first-class event (not folded into `memory.suppress`), so the Black Box
/// can show the exact contradiction decision the agent faced.
#[serde(rename = "contradiction.detected")]
ContradictionDetected {
#[serde(rename = "runId")]
run_id: String,
/// The two (or more) memory ids in tension.
ids: Vec<String>,
/// The id the agent trusted (kept), if it resolved the tension.
#[serde(rename = "winnerId", skip_serializing_if = "Option::is_none")]
winner_id: Option<String>,
/// Plain-language description of the contradiction.
detail: String,
#[serde(default)]
at: i64,
},
/// The Sanhedrin verifier vetoed a claim the agent was about to assert,
/// citing the evidence it weighed and its confidence.
#[serde(rename = "sanhedrin.veto")]
SanhedrinVeto {
#[serde(rename = "runId")]
run_id: String,
claim: String,
#[serde(rename = "evidenceIds")]
evidence_ids: Vec<String>,
confidence: f64,
#[serde(default)]
at: i64,
},
/// Dream consolidation proposed a patch to memory (merge / insight / prune).
#[serde(rename = "dream.patch")]
DreamPatch {
#[serde(rename = "runId")]
run_id: String,
#[serde(rename = "proposalIds")]
proposal_ids: Vec<String>,
at: i64,
},
}
impl MemoryTraceEvent {
/// The run this event belongs to.
pub fn run_id(&self) -> &str {
match self {
MemoryTraceEvent::McpCall { run_id, .. }
| MemoryTraceEvent::MemoryRetrieve { run_id, .. }
| MemoryTraceEvent::MemorySuppress { run_id, .. }
| MemoryTraceEvent::MemoryWrite { run_id, .. }
| MemoryTraceEvent::ContradictionDetected { run_id, .. }
| MemoryTraceEvent::SanhedrinVeto { run_id, .. }
| MemoryTraceEvent::DreamPatch { run_id, .. } => run_id,
}
}
/// The wall-clock millisecond timestamp the event was recorded at.
pub fn at(&self) -> i64 {
match self {
MemoryTraceEvent::McpCall { at, .. }
| MemoryTraceEvent::MemoryRetrieve { at, .. }
| MemoryTraceEvent::MemorySuppress { at, .. }
| MemoryTraceEvent::MemoryWrite { at, .. }
| MemoryTraceEvent::ContradictionDetected { at, .. }
| MemoryTraceEvent::SanhedrinVeto { at, .. }
| MemoryTraceEvent::DreamPatch { at, .. } => *at,
}
}
/// Short stable kind label used for filtering / the `event_type` column.
pub fn kind(&self) -> &'static str {
match self {
MemoryTraceEvent::McpCall { .. } => "mcp.call",
MemoryTraceEvent::MemoryRetrieve { .. } => "memory.retrieve",
MemoryTraceEvent::MemorySuppress { .. } => "memory.suppress",
MemoryTraceEvent::MemoryWrite { .. } => "memory.write",
MemoryTraceEvent::ContradictionDetected { .. } => "contradiction.detected",
MemoryTraceEvent::SanhedrinVeto { .. } => "sanhedrin.veto",
MemoryTraceEvent::DreamPatch { .. } => "dream.patch",
}
}
/// Stamp `at` on events that left it defaulted (the recorder fills this so
/// callers don't have to thread a clock through every emit site).
pub fn with_at(mut self, now_ms: i64) -> Self {
match &mut self {
MemoryTraceEvent::McpCall { at, .. }
| MemoryTraceEvent::MemoryRetrieve { at, .. }
| MemoryTraceEvent::MemorySuppress { at, .. }
| MemoryTraceEvent::MemoryWrite { at, .. }
| MemoryTraceEvent::ContradictionDetected { at, .. }
| MemoryTraceEvent::SanhedrinVeto { at, .. }
| MemoryTraceEvent::DreamPatch { at, .. } => {
if *at == 0 {
*at = now_ms;
}
}
}
self
}
}
/// Why a memory was suppressed during a run. Mirrors the TS union member
/// `"low_trust" | "decayed" | "contradicted" | "privacy"`, plus `competition`
/// for the existing spreading-activation competition suppression.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum SuppressReason {
/// Below the trust floor for this retrieval.
LowTrust,
/// FSRS retrievability decayed below the usable threshold.
Decayed,
/// Contradicted by a higher-trust memory.
Contradicted,
/// Withheld for privacy / sensitivity reasons.
Privacy,
/// Lost spreading-activation competition to a stronger memory.
Competition,
}
impl SuppressReason {
/// Stable string label.
pub fn as_str(&self) -> &'static str {
match self {
SuppressReason::LowTrust => "low_trust",
SuppressReason::Decayed => "decayed",
SuppressReason::Contradicted => "contradicted",
SuppressReason::Privacy => "privacy",
SuppressReason::Competition => "competition",
}
}
}
/// Who caused a `memory.write`. Mirrors the TS `"agent" | "user" | "dream"`,
/// plus `connector` for external-source sync writes.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WriteSource {
/// The agent wrote it autonomously.
Agent,
/// The user explicitly asked for it.
User,
/// Produced by dream consolidation.
Dream,
/// Ingested by an external connector (GitHub, Redmine, …).
Connector,
}
impl WriteSource {
/// Stable string label.
pub fn as_str(&self) -> &'static str {
match self {
WriteSource::Agent => "agent",
WriteSource::User => "user",
WriteSource::Dream => "dream",
WriteSource::Connector => "connector",
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn trace_event_roundtrips_with_ts_shape() {
let ev = MemoryTraceEvent::McpCall {
run_id: "run_123".into(),
tool: "deep_reference".into(),
args_hash: "abc".into(),
at: 42,
};
let json = serde_json::to_value(&ev).unwrap();
// Tagged on `type`, camelCase runId/argsHash — exactly the TS contract.
assert_eq!(json["type"], "mcp.call");
assert_eq!(json["runId"], "run_123");
assert_eq!(json["argsHash"], "abc");
assert_eq!(json["at"], 42);
let back: MemoryTraceEvent = serde_json::from_value(json).unwrap();
assert_eq!(back, ev);
}
#[test]
fn retrieve_event_carries_activation_map() {
let mut activation = std::collections::BTreeMap::new();
activation.insert("mem_1".to_string(), 0.91);
activation.insert("mem_7".to_string(), 0.42);
let ev = MemoryTraceEvent::MemoryRetrieve {
run_id: "r".into(),
ids: vec!["mem_1".into(), "mem_7".into()],
activation,
at: 1,
};
let json = serde_json::to_value(&ev).unwrap();
assert_eq!(json["type"], "memory.retrieve");
assert_eq!(json["activation"]["mem_1"], 0.91);
}
#[test]
fn with_at_fills_only_when_unset() {
let ev = MemoryTraceEvent::MemorySuppress {
run_id: "r".into(),
id: "m".into(),
reason: SuppressReason::Contradicted,
at: 0,
}
.with_at(999);
assert_eq!(ev.at(), 999);
let ev2 = MemoryTraceEvent::DreamPatch {
run_id: "r".into(),
proposal_ids: vec!["p".into()],
at: 7,
}
.with_at(999);
assert_eq!(ev2.at(), 7, "explicit timestamp must not be overwritten");
}
#[test]
fn suppress_reason_labels_match_ts() {
assert_eq!(SuppressReason::LowTrust.as_str(), "low_trust");
assert_eq!(SuppressReason::Contradicted.as_str(), "contradicted");
// Serde uses the same snake_case form on the wire.
assert_eq!(
serde_json::to_value(SuppressReason::Privacy).unwrap(),
serde_json::json!("privacy")
);
}
}

View file

@ -0,0 +1,302 @@
//! # Memory Receipts
//!
//! Every important retrieval returns a [`Receipt`] — a structured record of what
//! the agent's memory actually did to answer a query. It is built entirely from
//! data the retrieval pipeline *already computes* (scored memories, suppression
//! decisions, spreading-activation path, FSRS trust), so attaching one is nearly
//! free and never changes the answer.
//!
//! The canonical shape (matching the product spec):
//!
//! ```json
//! {
//! "receipt_id": "r_2026_06_22_abc",
//! "retrieved": ["mem_1", "mem_7", "mem_9"],
//! "suppressed": [{"id": "mem_4", "reason": "contradicted"}],
//! "activation_path": ["project_goal -> design_decision -> current_file"],
//! "trust_floor": 0.62,
//! "decay_risk": "medium",
//! "mutations": []
//! }
//! ```
use serde::{Deserialize, Serialize};
use super::SuppressReason;
/// A structured receipt attached to a retrieval's output.
///
/// Field names are snake_case to match the published product spec and the
/// dashboard receipt card exactly.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Receipt {
/// Stable, human-legible id: `r_<yyyy>_<mm>_<dd>_<short>`.
pub receipt_id: String,
/// Ids of the memories that actually informed the answer, best-first.
pub retrieved: Vec<String>,
/// Memories that were withheld, each with the reason — the "what the agent
/// chose NOT to use" channel that makes retrieval auditable.
pub suppressed: Vec<SuppressedReceiptEntry>,
/// Human-readable spreading-activation path(s) that surfaced the result,
/// e.g. `"project_goal -> design_decision -> current_file"`.
pub activation_path: Vec<String>,
/// The minimum trust score among the retrieved memories — the weakest link
/// the answer rests on.
pub trust_floor: f64,
/// Coarse decay risk for the retrieved set (how stale the evidence is).
pub decay_risk: DecayRisk,
/// Any memory mutations this retrieval triggered (testing-effect
/// strengthening, reconsolidation, supersession). Empty for a pure read.
pub mutations: Vec<ReceiptMutation>,
}
impl Receipt {
/// Build a receipt from already-computed retrieval signals.
///
/// `receipt_id` is `r_<date>_<discriminator8>_<unique6>` — human-legible
/// and dated, with a short random suffix so that **multiple retrievals in
/// the same run never collide** (B3). The discriminator (usually the runId)
/// keeps receipts from one run visually grouped; the suffix guarantees
/// uniqueness so `INSERT OR REPLACE` can't overwrite an earlier receipt.
/// `trust_scores` is the per-id FSRS retrievability/trust the pipeline
/// already produced.
pub fn build(
now: chrono::DateTime<chrono::Utc>,
discriminator: &str,
retrieved: Vec<String>,
suppressed: Vec<SuppressedReceiptEntry>,
activation_path: Vec<String>,
trust_scores: &[f64],
mutations: Vec<ReceiptMutation>,
) -> Self {
Self::build_with_unique(
now,
discriminator,
&uuid::Uuid::new_v4().simple().to_string()[..6],
retrieved,
suppressed,
activation_path,
trust_scores,
mutations,
)
}
/// Like [`Receipt::build`] but with a caller-supplied uniqueness token,
/// so the id is fully deterministic for tests. Production uses
/// [`Receipt::build`] which mints a random token.
#[allow(clippy::too_many_arguments)]
pub fn build_with_unique(
now: chrono::DateTime<chrono::Utc>,
discriminator: &str,
unique: &str,
retrieved: Vec<String>,
suppressed: Vec<SuppressedReceiptEntry>,
activation_path: Vec<String>,
trust_scores: &[f64],
mutations: Vec<ReceiptMutation>,
) -> Self {
let trust_floor = trust_scores
.iter()
.copied()
.fold(f64::INFINITY, f64::min);
let trust_floor = if trust_floor.is_finite() {
(trust_floor * 100.0).round() / 100.0
} else {
0.0
};
let decay_risk = DecayRisk::from_trust_floor(trust_floor);
let short: String = discriminator
.chars()
.filter(|c| c.is_ascii_alphanumeric())
.take(8)
.collect();
let unique_clean: String = unique
.chars()
.filter(|c| c.is_ascii_alphanumeric())
.take(6)
.collect();
let receipt_id = format!(
"r_{}_{}_{}",
now.format("%Y_%m_%d"),
short,
unique_clean
);
Self {
receipt_id,
retrieved,
suppressed,
activation_path,
trust_floor,
decay_risk,
mutations,
}
}
}
/// One suppressed-memory entry in a [`Receipt`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SuppressedReceiptEntry {
/// The id of the suppressed memory.
pub id: String,
/// Why it was withheld.
pub reason: SuppressReason,
}
impl SuppressedReceiptEntry {
/// Convenience constructor.
pub fn new(id: impl Into<String>, reason: SuppressReason) -> Self {
Self {
id: id.into(),
reason,
}
}
}
/// Coarse staleness signal for a retrieved set, derived from the trust floor.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum DecayRisk {
/// Trust floor is healthy; the evidence is fresh.
Low,
/// Some of the evidence is weakening.
Medium,
/// The answer rests on memory that is decaying out.
High,
}
impl DecayRisk {
/// Map the weakest retrieved-trust score to a decay-risk band.
///
/// Thresholds align with the FSRS "due for review" intuition: above 0.7 the
/// memory is comfortably retrievable, 0.40.7 is getting weak, below 0.4 is
/// at risk of being forgotten.
pub fn from_trust_floor(trust_floor: f64) -> Self {
if trust_floor >= 0.7 {
DecayRisk::Low
} else if trust_floor >= 0.4 {
DecayRisk::Medium
} else {
DecayRisk::High
}
}
/// Stable string label.
pub fn as_str(&self) -> &'static str {
match self {
DecayRisk::Low => "low",
DecayRisk::Medium => "medium",
DecayRisk::High => "high",
}
}
}
/// A memory mutation that a retrieval triggered, recorded on the receipt so the
/// side effects of "just reading" are never invisible.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ReceiptMutation {
/// The mutated memory id.
pub id: String,
/// What changed: `"strengthened"`, `"reconsolidated"`, `"superseded"`, …
pub kind: String,
/// Optional human note about the change.
#[serde(skip_serializing_if = "Option::is_none")]
pub note: Option<String>,
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::TimeZone;
fn fixed_now() -> chrono::DateTime<chrono::Utc> {
chrono::Utc.with_ymd_and_hms(2026, 6, 22, 15, 0, 0).unwrap()
}
#[test]
fn receipt_id_is_human_legible_and_dated() {
let r = Receipt::build_with_unique(
fixed_now(),
"abc123!!",
"u1u2u3",
vec!["mem_1".into()],
vec![],
vec![],
&[0.9],
vec![],
);
assert_eq!(r.receipt_id, "r_2026_06_22_abc123_u1u2u3");
}
#[test]
fn receipt_ids_unique_within_a_run_b3() {
// B3: two retrievals in the SAME run (same date + discriminator) must
// get DISTINCT ids so INSERT OR REPLACE can't overwrite the first.
let a = Receipt::build(fixed_now(), "run_x", vec![], vec![], vec![], &[], vec![]);
let b = Receipt::build(fixed_now(), "run_x", vec![], vec![], vec![], &[], vec![]);
assert_ne!(
a.receipt_id, b.receipt_id,
"same-run receipts must not collide"
);
assert!(a.receipt_id.starts_with("r_2026_06_22_runx_"));
assert!(b.receipt_id.starts_with("r_2026_06_22_runx_"));
}
#[test]
fn trust_floor_is_the_weakest_link() {
let r = Receipt::build(
fixed_now(),
"x",
vec!["a".into(), "b".into(), "c".into()],
vec![],
vec![],
&[0.91, 0.62, 0.78],
vec![],
);
assert_eq!(r.trust_floor, 0.62);
assert_eq!(r.decay_risk, DecayRisk::Medium);
}
#[test]
fn empty_trust_scores_floor_to_zero_high_risk() {
let r = Receipt::build(fixed_now(), "x", vec![], vec![], vec![], &[], vec![]);
assert_eq!(r.trust_floor, 0.0);
assert_eq!(r.decay_risk, DecayRisk::High);
}
#[test]
fn decay_bands() {
assert_eq!(DecayRisk::from_trust_floor(0.95), DecayRisk::Low);
assert_eq!(DecayRisk::from_trust_floor(0.55), DecayRisk::Medium);
assert_eq!(DecayRisk::from_trust_floor(0.20), DecayRisk::High);
}
#[test]
fn matches_published_spec_shape() {
let r = Receipt {
receipt_id: "r_2026_06_22_abc".into(),
retrieved: vec!["mem_1".into(), "mem_7".into(), "mem_9".into()],
suppressed: vec![SuppressedReceiptEntry::new(
"mem_4",
SuppressReason::Contradicted,
)],
activation_path: vec!["project_goal -> design_decision -> current_file".into()],
trust_floor: 0.62,
decay_risk: DecayRisk::Medium,
mutations: vec![],
};
let json = serde_json::to_value(&r).unwrap();
assert_eq!(json["receipt_id"], "r_2026_06_22_abc");
assert_eq!(json["suppressed"][0]["reason"], "contradicted");
assert_eq!(json["decay_risk"], "medium");
assert_eq!(json["trust_floor"], 0.62);
assert!(json["mutations"].as_array().unwrap().is_empty());
}
}

View file

@ -0,0 +1,802 @@
//! # Memory PRs — review changes to an agent's brain like code
//!
//! Ordinary context auto-commits and always leaves a receipt. But a *risky*
//! write — one where the agent is rewriting its own brain — opens a reviewable
//! [`MemoryPr`] instead. [`classify_write`] is the immune system: given a
//! [`WriteContext`] and a [`ReviewMode`], it returns the [`RiskClass`] and the
//! [`RiskSignal`]s that explain, in plain language, *why* a write needs review.
//!
//! ## The three modes (one-click in the dashboard)
//!
//! | Mode | Behaviour |
//! |------|-----------|
//! | [`ReviewMode::Fast`] | Never gate. Every write auto-commits. (Demos, trusted solo flows.) |
//! | [`ReviewMode::RiskGated`] | **Default.** Auto-commit ordinary writes; open a PR for risky ones. |
//! | [`ReviewMode::Paranoid`] | Gate *every* write. Nothing enters the brain without approval. |
//!
//! ## What counts as "risky" (the taxonomy)
//!
//! A write is risky when any of these hold:
//! - it **contradicts a high-trust memory**,
//! - it **supersedes / forgets / merges / protects** existing memory,
//! - it touches **identity, user preference, workflow, or project positioning**,
//! - it asserts a **permission / auth / security / money / bounty / legal-ish** fact,
//! - it is a **dream consolidation** proposal,
//! - it **resurrects a decayed** memory (below the retention threshold),
//! - it is part of a **low-confidence batch import**,
//! - it is an **external connector write without strong provenance**.
//!
//! Each rule maps to a [`RiskSignal`] so the resulting Memory PR is fully
//! self-explaining.
use serde::{Deserialize, Serialize};
use super::WriteSource;
/// A memory is "high trust" at or above this FSRS retrievability/trust score.
/// Contradicting something this trusted is always worth a review.
pub const HIGH_TRUST_FLOOR: f64 = 0.7;
/// Writes below this confidence are treated as low-confidence (e.g. a bulk
/// import where the model wasn't sure).
pub const LOW_CONFIDENCE_FLOOR: f64 = 0.5;
// ============================================================================
// REVIEW MODE
// ============================================================================
/// How aggressively the agent's brain gates incoming writes.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
pub enum ReviewMode {
/// Never gate — every write auto-commits.
Fast,
/// Default: auto-commit ordinary writes, open a PR for risky ones.
#[default]
RiskGated,
/// Gate every write — nothing enters the brain without approval.
Paranoid,
}
impl ReviewMode {
/// Stable string label, also the wire form.
pub fn as_str(&self) -> &'static str {
match self {
ReviewMode::Fast => "fast",
ReviewMode::RiskGated => "risk_gated",
ReviewMode::Paranoid => "paranoid",
}
}
/// Parse from a label (case-insensitive, tolerant of `-`/`_`). Falls back to
/// the default [`ReviewMode::RiskGated`] on anything unrecognised.
pub fn from_label(s: &str) -> Self {
match s.trim().to_ascii_lowercase().replace('-', "_").as_str() {
"fast" => ReviewMode::Fast,
"paranoid" => ReviewMode::Paranoid,
_ => ReviewMode::RiskGated,
}
}
}
// ============================================================================
// RISK CLASSIFICATION
// ============================================================================
/// The outcome of [`classify_write`]: does this write auto-commit or open a PR?
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum RiskClass {
/// Ordinary context — auto-commit (a receipt is still generated).
AutoCommit,
/// Risky — open a [`MemoryPr`] for review.
Review,
}
impl RiskClass {
/// Stable string label.
pub fn as_str(&self) -> &'static str {
match self {
RiskClass::AutoCommit => "auto_commit",
RiskClass::Review => "review",
}
}
/// Whether this write should be held for review.
pub fn needs_review(&self) -> bool {
matches!(self, RiskClass::Review)
}
}
/// A single, self-explaining reason a write was flagged for review. The
/// `code` is stable for filtering/telemetry; the `detail` is human prose for
/// the PR card.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RiskSignal {
/// Stable machine code, e.g. `"contradicts_high_trust"`.
pub code: String,
/// Plain-language explanation shown on the Memory PR.
pub detail: String,
}
impl RiskSignal {
fn new(code: &str, detail: impl Into<String>) -> Self {
Self {
code: code.into(),
detail: detail.into(),
}
}
}
/// Everything [`classify_write`] needs to decide whether a write is risky.
///
/// All fields default to the "ordinary, safe" interpretation so callers only
/// set the signals that actually apply to their write.
#[derive(Debug, Clone, Default)]
pub struct WriteContext {
/// Who is performing the write.
pub source: Option<WriteSource>,
/// The node type being written, e.g. `"fact"`, `"preference"`, `"identity"`.
pub node_type: String,
/// The content (or a representative slice) — scanned for sensitive topics.
pub content: String,
/// Tags attached to the write — also scanned for sensitive topics.
pub tags: Vec<String>,
/// The write contradicts an existing memory whose trust is this high.
/// `None` if there is no contradiction.
pub contradicts_trust: Option<f64>,
/// This write supersedes / replaces an existing memory.
pub supersedes: bool,
/// This write forgets / suppresses an existing memory.
pub forgets: bool,
/// This write merges existing memories.
pub merges: bool,
/// This write protects / pins a memory.
pub protects: bool,
/// This write resurrects a memory that had decayed below retention.
pub resurrects_decayed: bool,
/// Confidence of the write (0..1). `None` means "not a batch / unknown".
pub confidence: Option<f64>,
/// This write is one of many in a bulk import.
pub batch_import: bool,
/// For connector writes: whether the source envelope carries strong
/// provenance (a verified `source_system` + `source_id` + URL).
pub strong_provenance: bool,
}
/// Sensitive topic substrings. A write whose content/tags/type mention any of
/// these is treated as touching identity / preference / security / money /
/// legal / workflow / positioning and is routed to review.
const SENSITIVE_TOPICS: &[(&str, &str)] = &[
// identity & preference
("identity", "identity fact"),
("preference", "user preference"),
("workflow", "workflow rule"),
("positioning", "project positioning"),
("persona", "agent persona"),
// permission / auth / security
("permission", "tool permission"),
("auth", "authentication / authorization"),
("token", "credential / token"),
("secret", "secret material"),
("password", "credential"),
("api key", "credential / API key"),
("security", "security-relevant fact"),
("vuln", "security vulnerability"),
("vulnerability", "security vulnerability"),
("credential", "credential material"),
("credentials", "credential material"),
("api key", "credential / API key"),
("apikey", "credential / API key"),
// money / bounty / legal
("money", "financial fact"),
("payment", "financial fact"),
("invoice", "financial fact"),
("bounty", "bounty / payout"),
("salary", "financial fact"),
("license", "legal / license fact"),
("legal", "legal-relevant fact"),
("contract", "legal / contract fact"),
];
/// Node types that are intrinsically sensitive regardless of content.
const SENSITIVE_NODE_TYPES: &[&str] = &[
"identity",
"preference",
"user_preference",
"credential",
"permission",
"security",
"constitution",
];
/// Classify a write into auto-commit vs. review, with the signals explaining the
/// decision.
///
/// This is the immune system. It is pure and deterministic, so the dashboard's
/// "explain this PR" view and the agent's `Ask Agent Why` action see exactly the
/// same reasoning the gate used.
pub fn classify_write(ctx: &WriteContext, mode: ReviewMode) -> (RiskClass, Vec<RiskSignal>) {
// Mode shortcuts.
match mode {
// Fast never gates — but we still collect signals so the receipt/PR
// record can note what *would* have been flagged.
ReviewMode::Fast => return (RiskClass::AutoCommit, Vec::new()),
ReviewMode::Paranoid => {
let mut signals = collect_signals(ctx);
if signals.is_empty() {
signals.push(RiskSignal::new(
"paranoid_mode",
"Paranoid mode: every write is reviewed before entering memory.",
));
}
return (RiskClass::Review, signals);
}
ReviewMode::RiskGated => {}
}
let signals = collect_signals(ctx);
if signals.is_empty() {
(RiskClass::AutoCommit, signals)
} else {
(RiskClass::Review, signals)
}
}
/// Gather every risk signal that applies to a write, independent of mode.
fn collect_signals(ctx: &WriteContext) -> Vec<RiskSignal> {
let mut signals = Vec::new();
// 1. Contradiction against a high-trust memory.
if let Some(trust) = ctx.contradicts_trust
&& trust >= HIGH_TRUST_FLOOR
{
signals.push(RiskSignal::new(
"contradicts_high_trust",
format!(
"Contradicts an existing high-trust memory (trust {:.2} ≥ {:.2}).",
trust, HIGH_TRUST_FLOOR
),
));
}
// 2. Structural rewrites of existing memory.
if ctx.supersedes {
signals.push(RiskSignal::new(
"supersedes_memory",
"Supersedes / replaces an existing memory.",
));
}
if ctx.forgets {
signals.push(RiskSignal::new(
"forgets_memory",
"Forgets / suppresses an existing memory.",
));
}
if ctx.merges {
signals.push(RiskSignal::new(
"merges_memory",
"Merges existing memories into one.",
));
}
if ctx.protects {
signals.push(RiskSignal::new(
"protects_memory",
"Protects / pins a memory against decay and forgetting.",
));
}
// 3. Sensitive node types & topics (identity / preference / workflow /
// positioning / permission / auth / security / money / legal).
let node_type_lc = ctx.node_type.to_ascii_lowercase();
if SENSITIVE_NODE_TYPES.contains(&node_type_lc.as_str()) {
signals.push(RiskSignal::new(
"sensitive_node_type",
format!("Writes a sensitive node type: `{}`.", node_type_lc),
));
}
if let Some(topic) = first_sensitive_topic(&ctx.content, &ctx.tags) {
signals.push(RiskSignal::new(
"sensitive_topic",
format!("Touches a sensitive topic: {topic}."),
));
}
// 4. Dream consolidation proposals.
if matches!(ctx.source, Some(WriteSource::Dream)) {
signals.push(RiskSignal::new(
"dream_consolidation",
"Proposed by dream consolidation — a machine-generated change to memory.",
));
}
// 5. Decay-below-threshold resurrection.
if ctx.resurrects_decayed {
signals.push(RiskSignal::new(
"resurrects_decayed",
"Resurrects a memory that had decayed below the retention threshold.",
));
}
// 6. Low-confidence batch imports.
if ctx.batch_import {
if let Some(conf) = ctx.confidence {
if conf < LOW_CONFIDENCE_FLOOR {
signals.push(RiskSignal::new(
"low_confidence_batch",
format!(
"Low-confidence batch import (confidence {:.2} < {:.2}).",
conf, LOW_CONFIDENCE_FLOOR
),
));
}
} else {
signals.push(RiskSignal::new(
"unscored_batch",
"Batch import with no confidence score.",
));
}
}
// 7. External connector writes without strong provenance.
if matches!(ctx.source, Some(WriteSource::Connector)) && !ctx.strong_provenance {
signals.push(RiskSignal::new(
"weak_provenance_connector",
"External connector write without strong provenance (unverified source envelope).",
));
}
signals
}
/// Return the human label of the first sensitive topic found in content/tags.
///
/// B6: matches on WORD BOUNDARIES, not substrings — so "tokenizer" no longer
/// trips "token", "author" no longer trips "auth", "secretary" no longer trips
/// "secret". Multi-word needles (e.g. "api key") match a consecutive run of
/// words. The text is lowercased and split on any non-alphanumeric char.
fn first_sensitive_topic(content: &str, tags: &[String]) -> Option<&'static str> {
// Tokenize content + tags into lowercased alphanumeric words.
let mut words: Vec<String> = Vec::new();
let mut push_words = |s: &str| {
for w in s
.to_ascii_lowercase()
.split(|c: char| !c.is_ascii_alphanumeric())
{
if !w.is_empty() {
words.push(w.to_string());
}
}
};
push_words(content);
for t in tags {
push_words(t);
}
SENSITIVE_TOPICS
.iter()
.find(|(needle, _)| matches_word_sequence(&words, needle))
.map(|(_, label)| *label)
}
/// Whether `needle` (one or more space-separated words) appears as a consecutive
/// whole-word run in `words`.
fn matches_word_sequence(words: &[String], needle: &str) -> bool {
let needle_words: Vec<&str> = needle.split_whitespace().collect();
if needle_words.is_empty() {
return false;
}
if needle_words.len() == 1 {
return words.iter().any(|w| w == needle_words[0]);
}
words
.windows(needle_words.len())
.any(|win| win.iter().zip(&needle_words).all(|(w, n)| w == n))
}
// ============================================================================
// MEMORY PR DATA MODEL
// ============================================================================
/// What kind of change a Memory PR represents.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum MemoryPrKind {
/// A brand-new fact entering the brain.
NewFact,
/// An existing fact being strengthened / reinforced.
StrengthenedFact,
/// A contradiction was detected against existing memory.
ContradictionDetected,
/// A memory being superseded by a newer one.
MemorySuperseded,
/// A new edge added to the knowledge graph.
EdgeAdded,
/// A node decayed below the retention threshold.
NodeDecayed,
/// Dream consolidation proposed a merge / insight.
DreamConsolidation,
}
impl MemoryPrKind {
/// Stable string label.
pub fn as_str(&self) -> &'static str {
match self {
MemoryPrKind::NewFact => "new_fact",
MemoryPrKind::StrengthenedFact => "strengthened_fact",
MemoryPrKind::ContradictionDetected => "contradiction_detected",
MemoryPrKind::MemorySuperseded => "memory_superseded",
MemoryPrKind::EdgeAdded => "edge_added",
MemoryPrKind::NodeDecayed => "node_decayed",
MemoryPrKind::DreamConsolidation => "dream_consolidation",
}
}
/// Parse from a label; `None` if unrecognised.
pub fn from_label(s: &str) -> Option<Self> {
Some(match s {
"new_fact" => MemoryPrKind::NewFact,
"strengthened_fact" => MemoryPrKind::StrengthenedFact,
"contradiction_detected" => MemoryPrKind::ContradictionDetected,
"memory_superseded" => MemoryPrKind::MemorySuperseded,
"edge_added" => MemoryPrKind::EdgeAdded,
"node_decayed" => MemoryPrKind::NodeDecayed,
"dream_consolidation" => MemoryPrKind::DreamConsolidation,
_ => return None,
})
}
}
/// The review status of a Memory PR.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
pub enum MemoryPrStatus {
/// Awaiting a decision.
#[default]
Pending,
/// Promoted into long-term memory as-is.
Promoted,
/// Merged into an existing memory.
Merged,
/// Superseded an existing memory.
Superseded,
/// Quarantined — held in the firewall, not used for retrieval.
Quarantined,
/// Forgotten — rejected and suppressed.
Forgotten,
}
impl MemoryPrStatus {
/// Stable string label.
pub fn as_str(&self) -> &'static str {
match self {
MemoryPrStatus::Pending => "pending",
MemoryPrStatus::Promoted => "promoted",
MemoryPrStatus::Merged => "merged",
MemoryPrStatus::Superseded => "superseded",
MemoryPrStatus::Quarantined => "quarantined",
MemoryPrStatus::Forgotten => "forgotten",
}
}
}
/// The actions a reviewer can take on a Memory PR (the buttons in the diff UI).
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum MemoryPrAction {
/// Accept the change as-is.
Promote,
/// Fold it into an existing memory.
Merge,
/// Use it to supersede an existing memory.
Supersede,
/// Hold it in the firewall.
Quarantine,
/// Reject and suppress it.
Forget,
/// Ask the agent to explain the change (returns the risk signals).
AskAgentWhy,
}
impl MemoryPrAction {
/// Parse from a URL/path label; `None` if unrecognised.
pub fn from_label(s: &str) -> Option<Self> {
Some(match s {
"promote" => MemoryPrAction::Promote,
"merge" => MemoryPrAction::Merge,
"supersede" => MemoryPrAction::Supersede,
"quarantine" => MemoryPrAction::Quarantine,
"forget" => MemoryPrAction::Forget,
"ask_agent_why" | "ask-agent-why" | "why" => MemoryPrAction::AskAgentWhy,
_ => return None,
})
}
/// The status this action moves the PR into (`None` for `AskAgentWhy`, which
/// is read-only).
pub fn resulting_status(&self) -> Option<MemoryPrStatus> {
Some(match self {
MemoryPrAction::Promote => MemoryPrStatus::Promoted,
MemoryPrAction::Merge => MemoryPrStatus::Merged,
MemoryPrAction::Supersede => MemoryPrStatus::Superseded,
MemoryPrAction::Quarantine => MemoryPrStatus::Quarantined,
MemoryPrAction::Forget => MemoryPrStatus::Forgotten,
MemoryPrAction::AskAgentWhy => return None,
})
}
/// Whether deciding the PR with this action should **release** the subject
/// memory from quarantine (reverse the suppression that gate_writes applied).
///
/// A risky write is committed-then-suppressed; approving it must restore its
/// retrieval influence, otherwise the UI says "promoted" while the memory
/// stays held out — the bug this guards against. Accept actions release;
/// `Quarantine` keeps it held; `Forget` rejects it (stays suppressed);
/// `AskAgentWhy` is read-only.
pub fn releases_memory(&self) -> bool {
matches!(
self,
MemoryPrAction::Promote | MemoryPrAction::Merge | MemoryPrAction::Supersede
)
}
}
/// A reviewable change to the agent's brain — the persisted Memory PR record.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct MemoryPr {
/// UUID.
pub id: String,
/// What kind of change this is.
pub kind: MemoryPrKind,
/// Current review status.
pub status: MemoryPrStatus,
/// Short human title for the PR list.
pub title: String,
/// The proposed change as a structured diff (before/after, ids, payload).
pub diff: serde_json::Value,
/// The self-explaining risk signals that opened this PR.
pub signals: Vec<RiskSignal>,
/// The memory id this PR concerns, if any.
#[serde(skip_serializing_if = "Option::is_none")]
pub subject_id: Option<String>,
/// The run that produced this change, linking the PR back to the black box.
#[serde(skip_serializing_if = "Option::is_none")]
pub run_id: Option<String>,
/// RFC3339 creation time.
pub created_at: String,
/// RFC3339 decision time, once decided.
#[serde(skip_serializing_if = "Option::is_none")]
pub decided_at: Option<String>,
/// The action that resolved this PR, once decided.
#[serde(skip_serializing_if = "Option::is_none")]
pub decision: Option<MemoryPrAction>,
}
#[cfg(test)]
mod tests {
use super::*;
fn ordinary() -> WriteContext {
WriteContext {
source: Some(WriteSource::Agent),
node_type: "fact".into(),
content: "The build uses cargo and pnpm.".into(),
tags: vec!["build".into()],
..Default::default()
}
}
#[test]
fn ordinary_write_auto_commits_in_risk_gated() {
let (class, signals) = classify_write(&ordinary(), ReviewMode::RiskGated);
assert_eq!(class, RiskClass::AutoCommit);
assert!(signals.is_empty());
}
#[test]
fn fast_mode_never_gates_even_risky_writes() {
let mut ctx = ordinary();
ctx.supersedes = true;
ctx.contradicts_trust = Some(0.95);
let (class, _) = classify_write(&ctx, ReviewMode::Fast);
assert_eq!(class, RiskClass::AutoCommit);
}
#[test]
fn paranoid_mode_gates_even_ordinary_writes() {
let (class, signals) = classify_write(&ordinary(), ReviewMode::Paranoid);
assert_eq!(class, RiskClass::Review);
assert_eq!(signals[0].code, "paranoid_mode");
}
#[test]
fn contradiction_against_high_trust_is_risky() {
let mut ctx = ordinary();
ctx.contradicts_trust = Some(0.82);
let (class, signals) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::Review);
assert!(signals.iter().any(|s| s.code == "contradicts_high_trust"));
}
#[test]
fn contradiction_against_low_trust_is_fine() {
let mut ctx = ordinary();
ctx.contradicts_trust = Some(0.3);
let (class, _) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::AutoCommit);
}
#[test]
fn supersede_forget_merge_protect_all_gate() {
for set in [
|c: &mut WriteContext| c.supersedes = true,
|c: &mut WriteContext| c.forgets = true,
|c: &mut WriteContext| c.merges = true,
|c: &mut WriteContext| c.protects = true,
] {
let mut ctx = ordinary();
set(&mut ctx);
let (class, _) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::Review);
}
}
#[test]
fn sensitive_topics_gate() {
for topic in [
"remember my auth token is xyz",
"Sam's salary is confidential",
"the bounty payout terms",
"user preference: dark mode",
"this is a security vulnerability",
] {
let mut ctx = ordinary();
ctx.content = topic.into();
let (class, signals) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::Review, "should gate: {topic}");
assert!(signals.iter().any(|s| s.code == "sensitive_topic"));
}
}
#[test]
fn sensitive_topic_word_boundary_no_false_positives_b6() {
// B6: these ordinary technical writes must NOT gate — they only CONTAIN
// a sensitive substring, they don't USE the sensitive word.
// These each only CONTAIN a sensitive substring; the word-boundary fix
// means they no longer gate. (Note: bare "license"/"contract"/"legal"
// ARE kept as gating words — a license/contract fact is legitimately
// legal-relevant — so they're intentionally not in this benign set.)
for benign in [
"The tokenizer converts input strings to embeddings.",
"The author of this module is documented in the header.",
"The secretary pattern coordinates the worker pool.",
"Contraction of the array happens during compaction.",
"The authority record links to the canonical node.",
"The authentication-free endpoint is for health checks.", // "authentication" != "auth"
] {
let mut ctx = ordinary();
ctx.content = benign.into();
ctx.node_type = "fact".into();
ctx.tags = vec![];
let (class, _) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(
class,
RiskClass::AutoCommit,
"must NOT gate ordinary write: {benign}"
);
}
}
#[test]
fn sensitive_topic_word_boundary_still_catches_real_b6() {
// The real sensitive phrasings must still gate.
for risky in [
"store the auth token for the deploy",
"this is a security vulnerability in the parser",
"the api key for the service",
"remember the user preference for dark mode",
"the bounty payout is configured",
] {
let mut ctx = ordinary();
ctx.content = risky.into();
ctx.node_type = "fact".into();
ctx.tags = vec![];
let (class, signals) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::Review, "must gate: {risky}");
assert!(signals.iter().any(|s| s.code == "sensitive_topic"));
}
}
#[test]
fn sensitive_node_type_gates() {
let mut ctx = ordinary();
ctx.node_type = "identity".into();
let (class, signals) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::Review);
assert!(signals.iter().any(|s| s.code == "sensitive_node_type"));
}
#[test]
fn dream_consolidation_gates() {
let mut ctx = ordinary();
ctx.source = Some(WriteSource::Dream);
let (class, signals) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::Review);
assert!(signals.iter().any(|s| s.code == "dream_consolidation"));
}
#[test]
fn decayed_resurrection_gates() {
let mut ctx = ordinary();
ctx.resurrects_decayed = true;
let (class, _) = classify_write(&ctx, ReviewMode::RiskGated);
assert_eq!(class, RiskClass::Review);
}
#[test]
fn low_confidence_batch_gates_but_confident_batch_does_not() {
let mut low = ordinary();
low.batch_import = true;
low.confidence = Some(0.3);
assert_eq!(
classify_write(&low, ReviewMode::RiskGated).0,
RiskClass::Review
);
let mut high = ordinary();
high.batch_import = true;
high.confidence = Some(0.9);
assert_eq!(
classify_write(&high, ReviewMode::RiskGated).0,
RiskClass::AutoCommit
);
}
#[test]
fn weak_provenance_connector_gates_strong_does_not() {
let mut weak = ordinary();
weak.source = Some(WriteSource::Connector);
weak.strong_provenance = false;
assert_eq!(
classify_write(&weak, ReviewMode::RiskGated).0,
RiskClass::Review
);
let mut strong = ordinary();
strong.source = Some(WriteSource::Connector);
strong.strong_provenance = true;
assert_eq!(
classify_write(&strong, ReviewMode::RiskGated).0,
RiskClass::AutoCommit
);
}
#[test]
fn mode_label_roundtrip() {
assert_eq!(ReviewMode::from_label("FAST"), ReviewMode::Fast);
assert_eq!(ReviewMode::from_label("risk-gated"), ReviewMode::RiskGated);
assert_eq!(ReviewMode::from_label("paranoid"), ReviewMode::Paranoid);
assert_eq!(ReviewMode::from_label("garbage"), ReviewMode::RiskGated);
}
#[test]
fn action_resulting_status() {
assert_eq!(
MemoryPrAction::Promote.resulting_status(),
Some(MemoryPrStatus::Promoted)
);
assert_eq!(MemoryPrAction::AskAgentWhy.resulting_status(), None);
}
#[test]
fn only_accept_actions_release_the_memory() {
// B1: accepting a risky write must release it from quarantine.
assert!(MemoryPrAction::Promote.releases_memory());
assert!(MemoryPrAction::Merge.releases_memory());
assert!(MemoryPrAction::Supersede.releases_memory());
// Rejecting / holding keeps it suppressed.
assert!(!MemoryPrAction::Forget.releases_memory());
assert!(!MemoryPrAction::Quarantine.releases_memory());
assert!(!MemoryPrAction::AskAgentWhy.releases_memory());
}
}

View file

@ -443,7 +443,10 @@ async fn handle_event(
| VestigeEvent::ConsolidationCompleted { .. }
| VestigeEvent::RetentionDecayed { .. }
| VestigeEvent::ConnectionDiscovered { .. }
| VestigeEvent::ActivationSpread { .. } => {}
| VestigeEvent::ActivationSpread { .. }
| VestigeEvent::TraceEvent { .. }
| VestigeEvent::MemoryPrOpened { .. }
| VestigeEvent::MemoryPrDecided { .. } => {}
}
}

View file

@ -167,6 +167,39 @@ pub enum VestigeEvent {
timestamp: DateTime<Utc>,
},
// -- Agent Black Box (v2.2) --
// One replayable trace event from an agent run. The dashboard Black Box tab
// appends these to the live timeline and pulses the graph exactly as the
// agent experienced it. The inner event is the canonical
// `vestige_core::MemoryTraceEvent`, serialized with its own `type` tag, so
// the wire shape is `{ "type": "TraceEvent", "data": { "runId": ..., "event": { "type": "mcp.call", ... } } }`.
TraceEvent {
run_id: String,
seq: i64,
event: vestige_core::MemoryTraceEvent,
timestamp: DateTime<Utc>,
},
// -- Memory PRs (v2.2) — the cognitive immune system --
// A risky write opened a Memory PR. The dashboard raises the PR-queue badge
// and can surface a toast: "Vestige opened a Memory PR — the agent tried to
// rewrite its own brain."
MemoryPrOpened {
id: String,
kind: String,
title: String,
signal_count: usize,
run_id: Option<String>,
timestamp: DateTime<Utc>,
},
// A Memory PR was decided (promote / merge / supersede / quarantine / forget).
MemoryPrDecided {
id: String,
decision: String,
status: String,
timestamp: DateTime<Utc>,
},
// -- System --
Heartbeat {
uptime_secs: u64,

View file

@ -1983,6 +1983,358 @@ pub async fn deep_reference_query(
Ok(Json(response))
}
// ============================================================================
// AGENT BLACK BOX (v2.2) — replayable agent-run traces
// ============================================================================
#[derive(Debug, Deserialize)]
pub struct TraceListParams {
pub limit: Option<usize>,
/// Optional run filter — receipts/traces scoped to one run (B5).
pub run: Option<String>,
}
/// List recent agent runs (newest activity first) for the Black Box run picker.
pub async fn list_traces(
State(state): State<AppState>,
Query(params): Query<TraceListParams>,
) -> Result<Json<Value>, StatusCode> {
let limit = params.limit.unwrap_or(50).clamp(1, 500);
let runs = state
.storage
.list_agent_runs(limit)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let runs_json: Vec<Value> = runs
.into_iter()
.map(|r| {
serde_json::json!({
"runId": r.run_id,
"firstTool": r.first_tool,
"eventCount": r.event_count,
"retrievedCount": r.retrieved_count,
"suppressedCount": r.suppressed_count,
"writeCount": r.write_count,
"vetoCount": r.veto_count,
"startedAt": r.started_at,
"lastAt": r.last_at,
})
})
.collect();
Ok(Json(serde_json::json!({
"total": runs_json.len(),
"runs": runs_json,
})))
}
/// Fetch the full event timeline for one run — the black-box replay payload.
pub async fn get_trace(
State(state): State<AppState>,
Path(run_id): Path<String>,
) -> Result<Json<Value>, StatusCode> {
let events = state
.storage
.get_trace(&run_id)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
if events.is_empty() {
return Err(StatusCode::NOT_FOUND);
}
let summary = state.storage.get_agent_run(&run_id).ok().flatten();
Ok(Json(serde_json::json!({
"runId": run_id,
"summary": summary.map(|s| serde_json::json!({
"firstTool": s.first_tool,
"eventCount": s.event_count,
"retrievedCount": s.retrieved_count,
"suppressedCount": s.suppressed_count,
"writeCount": s.write_count,
"vetoCount": s.veto_count,
"startedAt": s.started_at,
"lastAt": s.last_at,
})),
"events": events,
})))
}
/// Export a run as a downloadable `.vestige-trace.json` artifact.
pub async fn export_trace(
State(state): State<AppState>,
Path(run_id): Path<String>,
) -> Result<([(axum::http::HeaderName, String); 2], Json<Value>), StatusCode> {
let events = state
.storage
.get_trace(&run_id)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
if events.is_empty() {
return Err(StatusCode::NOT_FOUND);
}
let summary = state.storage.get_agent_run(&run_id).ok().flatten();
let body = serde_json::json!({
"format": "vestige-trace",
"version": 1,
"runId": run_id,
"exportedAt": Utc::now().to_rfc3339(),
"summary": summary.map(|s| serde_json::json!({
"firstTool": s.first_tool,
"eventCount": s.event_count,
"retrievedCount": s.retrieved_count,
"suppressedCount": s.suppressed_count,
"writeCount": s.write_count,
"vetoCount": s.veto_count,
"startedAt": s.started_at,
"lastAt": s.last_at,
})),
"events": events,
});
// B7: sanitize the run_id before putting it in the download filename so a
// crafted run_id (quotes, path separators, control chars) can't break the
// Content-Disposition header or the filename. Falls back to "trace".
let safe: String = run_id
.chars()
.map(|c| if c.is_ascii_alphanumeric() || c == '_' || c == '-' { c } else { '_' })
.collect();
let safe = if safe.trim_matches('_').is_empty() {
"trace".to_string()
} else {
safe
};
let headers = [
(
axum::http::header::CONTENT_TYPE,
"application/json".to_string(),
),
(
axum::http::header::CONTENT_DISPOSITION,
format!("attachment; filename=\"{safe}.vestige-trace.json\""),
),
];
Ok((headers, Json(body)))
}
// ============================================================================
// MEMORY RECEIPTS (v2.2)
// ============================================================================
/// List recent retrieval receipts.
pub async fn list_receipts(
State(state): State<AppState>,
Query(params): Query<TraceListParams>,
) -> Result<Json<Value>, StatusCode> {
let limit = params.limit.unwrap_or(50).clamp(1, 500);
// B5: when a run is given, scope to that run's receipts so the Black Box
// panel shows only receipts that actually belong to the selected run.
let receipts = match params.run.as_deref().filter(|r| !r.is_empty()) {
Some(run_id) => state.storage.list_receipts_for_run(run_id, limit),
None => state.storage.list_receipts(limit),
}
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(serde_json::json!({
"total": receipts.len(),
"receipts": receipts,
})))
}
/// Fetch one receipt by id — the payload behind "Open receipt in Cinema".
pub async fn get_receipt(
State(state): State<AppState>,
Path(receipt_id): Path<String>,
) -> Result<Json<Value>, StatusCode> {
let receipt = state
.storage
.get_receipt(&receipt_id)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
Ok(Json(serde_json::to_value(receipt).unwrap_or_default()))
}
// ============================================================================
// MEMORY PRs (v2.2) — risk-gated brain-change review queue
// ============================================================================
#[derive(Debug, Deserialize)]
pub struct MemoryPrListParams {
pub status: Option<String>,
pub limit: Option<usize>,
}
/// List Memory PRs, optionally filtered by status.
pub async fn list_memory_prs(
State(state): State<AppState>,
Query(params): Query<MemoryPrListParams>,
) -> Result<Json<Value>, StatusCode> {
let limit = params.limit.unwrap_or(100).clamp(1, 500);
let status = params.status.as_deref().and_then(|s| {
serde_json::from_value::<vestige_core::MemoryPrStatus>(serde_json::Value::String(
s.to_string(),
))
.ok()
});
let prs = state
.storage
.list_memory_prs(status, limit)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let pending = state.storage.count_pending_memory_prs().unwrap_or(0);
Ok(Json(serde_json::json!({
"total": prs.len(),
"pendingCount": pending,
"mode": read_review_mode(&state).as_str(),
"prs": prs,
})))
}
/// Fetch one Memory PR by id.
pub async fn get_memory_pr(
State(state): State<AppState>,
Path(id): Path<String>,
) -> Result<Json<Value>, StatusCode> {
let pr = state
.storage
.get_memory_pr(&id)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
Ok(Json(serde_json::to_value(pr).unwrap_or_default()))
}
/// Act on a Memory PR: promote / merge / supersede / quarantine / forget /
/// ask_agent_why. `ask_agent_why` is read-only and returns the risk signals.
pub async fn act_on_memory_pr(
State(state): State<AppState>,
Path((id, action)): Path<(String, String)>,
) -> Result<Json<Value>, StatusCode> {
let action = vestige_core::MemoryPrAction::from_label(&action)
.ok_or(StatusCode::BAD_REQUEST)?;
// Ask Agent Why is read-only — return the self-explaining signals.
if matches!(action, vestige_core::MemoryPrAction::AskAgentWhy) {
let pr = state
.storage
.get_memory_pr(&id)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
return Ok(Json(serde_json::json!({
"id": pr.id,
"kind": pr.kind.as_str(),
"title": pr.title,
"why": pr.signals,
"explanation": "These are the risk signals that opened this Memory PR.",
})));
}
let decided = state
.storage
.decide_memory_pr(&id, action)
.map_err(|_| StatusCode::NOT_FOUND)?;
// B1: an accept action (promote/merge/supersede) must RELEASE the subject
// memory from quarantine — gate_writes suppressed it, so deciding the PR
// without un-suppressing would leave it "promoted" yet still held out of
// retrieval. Forget/Quarantine intentionally keep it suppressed.
let mut released = false;
if action.releases_memory()
&& let Some(subject_id) = decided.subject_id.as_deref()
{
// Use the UNCONDITIONAL quarantine release, not reverse_suppression:
// approving a PR must restore the memory even if reviewed days later,
// past the active-forgetting labile window (the C1 fix).
match state.storage.release_quarantine(subject_id) {
Ok(node) => {
released = true;
state.emit(VestigeEvent::MemoryUnsuppressed {
id: node.id.clone(),
remaining_count: node.suppression_count,
timestamp: Utc::now(),
});
}
Err(e) => {
// Best-effort: the PR is decided regardless, but surface the
// failure so a stuck-suppressed memory isn't silent.
tracing::warn!(
"memory PR {} {}d but failed to release subject {}: {}",
id,
action_label(action),
subject_id,
e
);
}
}
}
state.emit(VestigeEvent::MemoryPrDecided {
id: decided.id.clone(),
decision: decided
.decision
.and_then(|d| serde_json::to_value(d).ok())
.and_then(|v| v.as_str().map(String::from))
.unwrap_or_default(),
status: decided.status.as_str().to_string(),
timestamp: Utc::now(),
});
let mut out = serde_json::to_value(&decided).unwrap_or_default();
if let Some(obj) = out.as_object_mut() {
obj.insert("subjectReleased".to_string(), serde_json::json!(released));
}
Ok(Json(out))
}
/// Short label for a Memory PR action, for log lines.
fn action_label(action: vestige_core::MemoryPrAction) -> &'static str {
use vestige_core::MemoryPrAction::*;
match action {
Promote => "promote",
Merge => "merge",
Supersede => "supersede",
Quarantine => "quarantine",
Forget => "forget",
AskAgentWhy => "ask_agent_why",
}
}
#[derive(Debug, Deserialize)]
pub struct ReviewModeBody {
pub mode: String,
}
/// Get the current review mode (fast / risk_gated / paranoid).
pub async fn get_review_mode(State(state): State<AppState>) -> Json<Value> {
let mode = read_review_mode(&state);
Json(serde_json::json!({
"mode": mode.as_str(),
"pendingCount": state.storage.count_pending_memory_prs().unwrap_or(0),
}))
}
/// Set the review mode. Persisted to a small JSON file in the data dir so it
/// survives restarts (local-first, no extra config service).
pub async fn set_review_mode(
State(state): State<AppState>,
Json(body): Json<ReviewModeBody>,
) -> Result<Json<Value>, StatusCode> {
let mode = vestige_core::ReviewMode::from_label(&body.mode);
let path = review_mode_path(&state);
let payload = serde_json::json!({ "mode": mode.as_str() });
// B7: atomic write (temp + rename) so a concurrent read can never see a
// partially-written / corrupt review_mode.json, reusing the same helper the
// Sanhedrin receipt path uses.
write_atomic(&path, &serde_json::to_vec_pretty(&payload).unwrap_or_default())?;
Ok(Json(serde_json::json!({ "mode": mode.as_str() })))
}
/// Path to the persisted review-mode file.
fn review_mode_path(state: &AppState) -> PathBuf {
state.storage.data_dir().join("review_mode.json")
}
/// Read the persisted review mode, defaulting to RiskGated.
pub fn read_review_mode(state: &AppState) -> vestige_core::ReviewMode {
let path = review_mode_path(state);
fs::read_to_string(&path)
.ok()
.and_then(|s| serde_json::from_str::<Value>(&s).ok())
.and_then(|v| v.get("mode").and_then(|m| m.as_str()).map(String::from))
.map(|s| vestige_core::ReviewMode::from_label(&s))
.unwrap_or_default()
}
#[cfg(test)]
mod tests {
use super::*;

View file

@ -183,6 +183,32 @@ fn build_router_inner(state: AppState, port: u16) -> (Router, AppState) {
get(handlers::get_sanhedrin_telemetry),
)
.route("/api/sanhedrin/appeal", post(handlers::appeal_sanhedrin))
// ============================================================
// AGENT BLACK BOX (v2.2) — replayable agent-run traces
// ============================================================
.route("/api/traces", get(handlers::list_traces))
.route("/api/traces/{run_id}", get(handlers::get_trace))
.route("/api/traces/{run_id}/export", get(handlers::export_trace))
// ============================================================
// MEMORY RECEIPTS (v2.2) — the nutrition label for a retrieval
// ============================================================
.route("/api/receipts", get(handlers::list_receipts))
.route("/api/receipts/{receipt_id}", get(handlers::get_receipt))
// ============================================================
// MEMORY PRs (v2.2) — risk-gated brain-change review queue
// ============================================================
.route("/api/memory-prs", get(handlers::list_memory_prs))
// Static `/mode` routes declared BEFORE the dynamic `/{id}` route (B7
// hygiene). axum 0.8/matchit already prioritizes static segments, but
// declaring them first makes the intent unambiguous and guards against
// a future router that doesn't.
.route("/api/memory-prs/mode", get(handlers::get_review_mode))
.route("/api/memory-prs/mode", post(handlers::set_review_mode))
.route("/api/memory-prs/{id}", get(handlers::get_memory_pr))
.route(
"/api/memory-prs/{id}/{action}",
post(handlers::act_on_memory_pr),
)
.layer(
ServiceBuilder::new()
.concurrency_limit(50)

View file

@ -9,3 +9,4 @@ pub mod protocol;
pub mod resources;
pub mod server;
pub mod tools;
pub mod trace_recorder;

View file

@ -4,3 +4,4 @@
pub mod codebase;
pub mod memory;
pub mod trace;

View file

@ -0,0 +1,103 @@
//! Agent Black Box Resources
//!
//! `trace://` URI scheme — exposes replayable agent-run traces as MCP resources
//! so a coding agent can read its *own* black box back. This closes the trace
//! correlation spine on the MCP side: the same `runId` an agent received in a
//! tool result's `traceUri` resolves here to the full event timeline.
//!
//! - `trace://{runId}` — the full ordered event log for a run.
//! - `trace://{runId}/summary` — just the roll-up counts.
//! - `trace://runs` — recent runs (the run picker).
//! - `trace://latest` — the most recently active run's full trace.
use std::sync::Arc;
use vestige_core::Storage;
/// Read a `trace://` resource.
pub async fn read(storage: &Arc<Storage>, uri: &str) -> Result<String, String> {
let path = uri.strip_prefix("trace://").unwrap_or("");
let (path, _query) = match path.split_once('?') {
Some((p, q)) => (p, Some(q)),
None => (path, None),
};
match path {
"" | "runs" => read_runs(storage).await,
"latest" => read_latest(storage).await,
other => {
if let Some(run_id) = other.strip_suffix("/summary") {
read_summary(storage, run_id).await
} else {
read_run(storage, other).await
}
}
}
}
async fn read_runs(storage: &Arc<Storage>) -> Result<String, String> {
let runs = storage.list_agent_runs(50).map_err(|e| e.to_string())?;
let json: Vec<_> = runs
.into_iter()
.map(|r| {
serde_json::json!({
"runId": r.run_id,
"firstTool": r.first_tool,
"eventCount": r.event_count,
"retrievedCount": r.retrieved_count,
"suppressedCount": r.suppressed_count,
"writeCount": r.write_count,
"vetoCount": r.veto_count,
"startedAt": r.started_at,
"lastAt": r.last_at,
})
})
.collect();
serde_json::to_string_pretty(&serde_json::json!({ "runs": json }))
.map_err(|e| e.to_string())
}
async fn read_latest(storage: &Arc<Storage>) -> Result<String, String> {
let runs = storage.list_agent_runs(1).map_err(|e| e.to_string())?;
let run = runs
.into_iter()
.next()
.ok_or_else(|| "No agent runs recorded yet".to_string())?;
read_run(storage, &run.run_id).await
}
async fn read_run(storage: &Arc<Storage>, run_id: &str) -> Result<String, String> {
let events = storage.get_trace(run_id).map_err(|e| e.to_string())?;
if events.is_empty() {
return Err(format!("No trace found for run: {run_id}"));
}
let summary = storage.get_agent_run(run_id).ok().flatten();
let body = serde_json::json!({
"runId": run_id,
"summary": summary.map(summary_json),
"events": events,
});
serde_json::to_string_pretty(&body).map_err(|e| e.to_string())
}
async fn read_summary(storage: &Arc<Storage>, run_id: &str) -> Result<String, String> {
let summary = storage
.get_agent_run(run_id)
.map_err(|e| e.to_string())?
.ok_or_else(|| format!("No run: {run_id}"))?;
serde_json::to_string_pretty(&summary_json(summary)).map_err(|e| e.to_string())
}
fn summary_json(s: vestige_core::AgentRunSummary) -> serde_json::Value {
serde_json::json!({
"runId": s.run_id,
"firstTool": s.first_tool,
"eventCount": s.event_count,
"retrievedCount": s.retrieved_count,
"suppressedCount": s.suppressed_count,
"writeCount": s.write_count,
"vetoCount": s.veto_count,
"startedAt": s.started_at,
"lastAt": s.last_at,
})
}

View file

@ -129,6 +129,23 @@ impl McpServer {
}
}
/// Read the active Memory PR review mode from `<data_dir>/review_mode.json`,
/// defaulting to `RiskGated`. Shared shape with the dashboard handler so the
/// MCP write path and the UI agree on the mode.
fn review_mode(&self) -> vestige_core::ReviewMode {
let path = self.storage.data_dir().join("review_mode.json");
std::fs::read_to_string(&path)
.ok()
.and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
.and_then(|v| {
v.get("mode")
.and_then(|m| m.as_str())
.map(|s| s.to_string())
})
.map(|s| vestige_core::ReviewMode::from_label(&s))
.unwrap_or_default()
}
/// Handle an incoming JSON-RPC request
pub async fn handle_request(&mut self, request: JsonRpcRequest) -> Option<JsonRpcResponse> {
debug!("Handling request: {}", request.method);
@ -240,8 +257,8 @@ impl McpServer {
/// Handle tools/list request
async fn handle_tools_list(&self) -> Result<serde_json::Value, JsonRpcError> {
// v2.1.21: 25 tools (verified by the `tools.len() == 25` assertion in the
// handle_tools_list test below — the `suppress` tool landed in v2.0.5).
// v2.1.27: 34 tools (verified by the `tools.len() == 34` assertion in the
// handle_tools_list test below).
// Deprecated tools still work via redirects in handle_tools_call.
let mut tools = vec![
// ================================================================
@ -503,7 +520,7 @@ impl McpServer {
// Per-tool caps below are sized at ~2× observed peak with growth
// headroom; max permitted by Anthropic is 500_000. Only the four
// empirically-measured high-payload tools carry the annotation today;
// the remaining 21 tools deliberately do NOT (cargo-cult prevention —
// the remaining 30 tools deliberately do NOT (cargo-cult prevention —
// annotating a small-payload tool dilutes the signal).
//
// Other tools that COULD plausibly grow into the annotated set with
@ -563,6 +580,21 @@ impl McpServer {
None
};
// ================================================================
// AGENT BLACK BOX (v2.2)
// Open/continue a run for this call and record the opening `mcp.call`
// event (args are hashed, never stored raw). Downstream memory events
// are recorded from the result after dispatch.
// ================================================================
let run_id = crate::trace_recorder::run_id_for(&request.arguments);
crate::trace_recorder::record_call(
&self.storage,
self.event_tx.as_ref(),
&run_id,
&request.name,
&request.arguments,
);
let result = match request.name.as_str() {
// ================================================================
// UNIFIED TOOLS (v1.1+) - Preferred API
@ -1083,10 +1115,81 @@ impl McpServer {
// ================================================================
if let Ok(ref content) = result {
self.emit_tool_event(&request.name, &saved_args, content);
// Black Box: record the downstream memory events (retrieve /
// suppress / write / veto / dream) the agent experienced.
crate::trace_recorder::record_result(
&self.storage,
self.event_tx.as_ref(),
&run_id,
&request.name,
content,
);
}
// ================================================================
// RISK-GATED MEMORY PRs (v2.2) — quarantine review, the cognitive
// immune system. Normal writes auto-land. Risky writes (contradiction
// vs high-trust, supersede/forget/merge, sensitive topics, …) are
// *committed then quarantined*: the row is recorded (audit history
// preserved) but suppressed out of retrieval until a Memory PR is
// decided. This is quarantine review, NOT pre-write blocking — the
// write happens inside the tool before the gate sees it; we hold its
// influence, not its existence. Centralized here so tools stay
// untouched.
// ================================================================
let opened_prs = if let Ok(ref content) = result {
crate::trace_recorder::gate_writes(
&self.storage,
self.event_tx.as_ref(),
&run_id,
&request.name,
content,
self.review_mode(),
)
} else {
Vec::new()
};
let response = match result {
Ok(content) => {
Ok(mut content) => {
// ============================================================
// TRACE SPINE (Phase 0)
// Stamp the runId + a pointer to the full trace onto the tool
// output itself. This is the first hop of the correlation
// chain: the same runId now appears in the tool result, the
// SQLite trace rows, the WebSocket events, /api/traces/{runId},
// and vestige://trace/{runId}. One id, end to end.
// ============================================================
// Memory Receipt: for retrieval tools, build + persist a
// receipt from what the tool already computed and attach it.
// Done before the runId stamp so the receipt's own suppressed
// list is part of the same payload the agent reads.
let receipt =
crate::trace_recorder::build_and_save_receipt(&self.storage, &run_id, &request.name, &content);
if let Some(obj) = content.as_object_mut() {
obj.insert("runId".to_string(), serde_json::json!(run_id));
obj.insert(
"traceUri".to_string(),
serde_json::json!(format!("vestige://trace/{run_id}")),
);
if let Some(r) = receipt {
obj.insert("receipt".to_string(), r);
}
// Surface opened Memory PRs so the agent learns its risky
// write is held for review, not silently committed.
if !opened_prs.is_empty() {
obj.insert(
"memoryPrsOpened".to_string(),
serde_json::json!(opened_prs),
);
obj.insert(
"memoryPrNotice".to_string(),
serde_json::json!(
"Vestige opened a Memory PR (quarantine review): this write was recorded but is held out of retrieval until reviewed — its audit history is preserved while its influence is suspended. See the Memory PRs queue."
),
);
}
}
let call_result = CallToolResult {
content: vec![crate::protocol::messages::ToolResultContent {
content_type: "text".to_string(),
@ -1228,6 +1331,27 @@ impl McpServer {
description: Some("Intentions that have been triggered or are overdue".to_string()),
mime_type: Some("application/json".to_string()),
},
// Agent Black Box (v2.2) — replayable agent-run traces. Individual
// runs are read via the templated `vestige://trace/{runId}` (or
// `trace://{runId}`) URI; these concrete entries list the runs and
// the latest trace so a client can discover them.
ResourceDescription {
uri: "trace://runs".to_string(),
name: "Agent Runs (Black Box)".to_string(),
description: Some(
"Recent agent runs. Read vestige://trace/{runId} for a full replayable trace."
.to_string(),
),
mime_type: Some("application/json".to_string()),
},
ResourceDescription {
uri: "trace://latest".to_string(),
name: "Latest Agent Trace".to_string(),
description: Some(
"The most recently active agent run's full black-box trace.".to_string(),
),
mime_type: Some("application/json".to_string()),
},
];
let result = ListResourcesResult { resources };
@ -1250,7 +1374,17 @@ impl McpServer {
// OpenCode and other MCP clients may send "vestige/memory://recent"
// but we register resources as "memory://recent"
let normalized_uri = uri.strip_prefix("vestige/").unwrap_or(uri);
let content = if normalized_uri.starts_with("memory://") {
// The trace resource is specced as `vestige://trace/{runId}`. Accept
// both that form and the bare `trace://{runId}` scheme, normalizing the
// former to the latter so the resource module sees one shape.
let trace_uri = normalized_uri
.strip_prefix("vestige://trace/")
.map(|rest| format!("trace://{rest}"));
let content = if let Some(ref tu) = trace_uri {
resources::trace::read(&self.storage, tu).await
} else if normalized_uri.starts_with("trace://") {
resources::trace::read(&self.storage, normalized_uri).await
} else if normalized_uri.starts_with("memory://") {
resources::memory::read(&self.storage, normalized_uri).await
} else if normalized_uri.starts_with("codebase://") {
resources::codebase::read(&self.storage, normalized_uri).await
@ -1820,9 +1954,9 @@ mod tests {
let result = response.result.unwrap();
let tools = result["tools"].as_array().unwrap();
// 34 tools: 25 from v2.1.21 + 7 Phase 3 merge/supersede tools
// (merge_candidates, plan_merge, plan_supersede, apply_plan, merge_undo,
// protect, merge_policy, composed_graph) + 1 connector tool (source_sync, #57).
// 34 tools in v2.1.27: the unified memory surface, Phase 3
// merge/supersede controls, ComposedGraph, and the #57 source_sync
// connector tool.
assert_eq!(tools.len(), 34, "Expected exactly 34 tools");
let tool_names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect();
@ -2248,4 +2382,245 @@ mod tests {
"search tool has un-renamed `meta` key (regression — serde rename broke)"
);
}
// ========================================================================
// TRACE SPINE (Phase 0) — one runId, end to end
// ========================================================================
/// Every tools/call must stamp a runId + a trace pointer onto its output,
/// persist an `mcp.call` trace row under that same runId, and that runId
/// must resolve through the `vestige://trace/{runId}` resource. This is the
/// load-bearing correlation guarantee.
#[tokio::test]
async fn test_trace_spine_runid_end_to_end() {
let (mut server, _dir) = test_server().await;
server
.handle_request(make_request("initialize", Some(init_params())))
.await;
// A client-supplied runId must be honoured so a whole session
// correlates under one id.
let call = make_request(
"tools/call",
Some(serde_json::json!({
"name": "memory_health",
"arguments": { "runId": "run_spine_test" }
})),
);
let response = server.handle_request(call).await.unwrap();
let result = response.result.expect("tools/call ok");
// 1. The tool output itself carries the runId + trace pointer.
let structured = &result["structuredContent"];
assert_eq!(
structured["runId"].as_str(),
Some("run_spine_test"),
"tool output must echo the runId (spine hop 1)"
);
assert_eq!(
structured["traceUri"].as_str(),
Some("vestige://trace/run_spine_test"),
"tool output must carry the trace resource pointer"
);
// 2. The same runId persisted a trace row (the mcp.call event).
let events = server.storage.get_trace("run_spine_test").unwrap();
assert!(
events.iter().any(|e| e.kind() == "mcp.call"),
"an mcp.call event must be persisted under the runId (spine hop 2)"
);
// 3. The run roll-up exists with the right entry tool.
let run = server
.storage
.get_agent_run("run_spine_test")
.unwrap()
.expect("run summary persisted");
assert_eq!(run.first_tool.as_deref(), Some("memory_health"));
// 4. The MCP resource resolves the same runId (spine hop 3).
let read = make_request(
"resources/read",
Some(serde_json::json!({ "uri": "vestige://trace/run_spine_test" })),
);
let read_resp = server.handle_request(read).await.unwrap();
let read_result = read_resp.result.expect("resource read ok");
let text = read_result["contents"][0]["text"]
.as_str()
.expect("resource text");
assert!(
text.contains("run_spine_test") && text.contains("mcp.call"),
"vestige://trace/{{runId}} must return the run's events"
);
}
/// Trace events must be broadcast to a live WebSocket subscriber, not just
/// persisted. This guards the spine hop from SQLite → WebSocket → pulse.
#[tokio::test]
async fn test_trace_event_is_broadcast_to_subscriber() {
let (storage, _dir) = test_storage().await;
let cognitive = Arc::new(Mutex::new(CognitiveEngine::new()));
let (event_tx, mut event_rx) = broadcast::channel(64);
let mut server = McpServer::new_with_events(storage, cognitive, event_tx);
server
.handle_request(make_request("initialize", Some(init_params())))
.await;
let call = make_request(
"tools/call",
Some(serde_json::json!({
"name": "memory_health",
"arguments": { "runId": "run_ws" }
})),
);
server.handle_request(call).await.unwrap();
// Drain the broadcast: at least one TraceEvent for run_ws must arrive.
let mut saw_trace = false;
while let Ok(ev) = event_rx.try_recv() {
if let VestigeEvent::TraceEvent { run_id, .. } = ev {
if run_id == "run_ws" {
saw_trace = true;
}
}
}
assert!(
saw_trace,
"a TraceEvent for the run must be broadcast to subscribers (spine hop: WebSocket)"
);
}
/// Risk-gated Memory PRs default: an ordinary tool call opens no PR.
#[tokio::test]
async fn test_no_memory_pr_for_non_write_tool() {
let (mut server, _dir) = test_server().await;
server
.handle_request(make_request("initialize", Some(init_params())))
.await;
let call = make_request(
"tools/call",
Some(serde_json::json!({
"name": "memory_health",
"arguments": { "runId": "run_no_pr" }
})),
);
server.handle_request(call).await.unwrap();
assert_eq!(
server.storage.count_pending_memory_prs().unwrap(),
0,
"a read-only tool must never open a Memory PR"
);
}
/// PROOF LOCK: the complete spine in one test. A single runId must cross
/// every hop, and the value must be byte-identical at each:
/// MCP output → SQLite trace → WebSocket event → API response shape →
/// MCP resource.
/// If any hop drops or rewrites the runId, this fails. This is the
/// "impossible to doubt" guarantee for the receipt chain.
#[tokio::test]
async fn test_full_spine_one_runid_crosses_every_hop() {
const RUN: &str = "run_full_spine";
let (storage, _dir) = test_storage().await;
let cognitive = Arc::new(Mutex::new(CognitiveEngine::new()));
let (event_tx, mut event_rx) = broadcast::channel(256);
let mut server = McpServer::new_with_events(storage, cognitive, event_tx);
server
.handle_request(make_request("initialize", Some(init_params())))
.await;
// ---- HOP 1: MCP tool output carries the runId + trace pointer ----
let call = make_request(
"tools/call",
Some(serde_json::json!({
"name": "memory_health",
"arguments": { "runId": RUN }
})),
);
let response = server.handle_request(call).await.unwrap();
let structured = response.result.expect("tools/call ok")["structuredContent"].clone();
assert_eq!(structured["runId"].as_str(), Some(RUN), "HOP 1: tool output runId");
assert_eq!(
structured["traceUri"].as_str(),
Some(&format!("vestige://trace/{RUN}")[..]),
"HOP 1: tool output traceUri"
);
// ---- HOP 2: SQLite trace rows persisted under the same runId ----
let events = server.storage.get_trace(RUN).unwrap();
assert!(!events.is_empty(), "HOP 2: trace rows exist");
assert!(
events.iter().all(|e| e.run_id() == RUN),
"HOP 2: every persisted trace row carries the SAME runId"
);
// ---- HOP 3: WebSocket broadcast carries the same runId ----
let mut ws_run: Option<String> = None;
while let Ok(ev) = event_rx.try_recv() {
if let VestigeEvent::TraceEvent { run_id, .. } = ev {
ws_run = Some(run_id);
break;
}
}
assert_eq!(
ws_run.as_deref(),
Some(RUN),
"HOP 3: the broadcast TraceEvent carries the same runId"
);
// ---- HOP 4: API response shape (what the dashboard renders) ----
// Exercise the exact handler the dashboard /api/traces/:runId calls by
// going through storage the same way, and assert the render-critical
// shape: a summary roll-up + an ordered event list, all under runId.
let summary = server
.storage
.get_agent_run(RUN)
.unwrap()
.expect("HOP 4: run summary the list view renders");
assert_eq!(summary.run_id, RUN, "HOP 4: API run summary runId");
assert!(summary.event_count >= 1, "HOP 4: event_count rendered in the list");
// The detail view renders these events in sequence order.
let detail_events = server.storage.get_trace(RUN).unwrap();
assert_eq!(
detail_events.len() as i64,
summary.event_count,
"HOP 4: detail event count matches the roll-up the list shows"
);
// ---- HOP 5: MCP resource resolves the same runId ----
let read = make_request(
"resources/read",
Some(serde_json::json!({ "uri": format!("vestige://trace/{RUN}") })),
);
let read_resp = server.handle_request(read).await.unwrap();
let text = read_resp.result.expect("resource read ok")["contents"][0]["text"]
.as_str()
.expect("resource text")
.to_string();
let parsed: serde_json::Value = serde_json::from_str(&text).unwrap();
assert_eq!(
parsed["runId"].as_str(),
Some(RUN),
"HOP 5: vestige://trace/{{runId}} resolves the same runId"
);
assert!(
parsed["events"].as_array().map(|a| !a.is_empty()).unwrap_or(false),
"HOP 5: the resource returns the run's events"
);
// ---- INVARIANT: one id, every hop, byte-identical ----
// Collect the runId as seen at each hop and assert they are all equal.
let seen = [
structured["runId"].as_str().unwrap().to_string(), // hop 1
events[0].run_id().to_string(), // hop 2
ws_run.unwrap(), // hop 3
summary.run_id, // hop 4
parsed["runId"].as_str().unwrap().to_string(), // hop 5
];
assert!(
seen.iter().all(|r| r == RUN),
"the SAME runId must appear, unchanged, at every hop: {seen:?}"
);
}
}

File diff suppressed because it is too large Load diff