From 0ff9527d2b0f413fc4621ea2367403f5dcced709 Mon Sep 17 00:00:00 2001 From: Sam Valladares Date: Wed, 24 Jun 2026 17:28:38 -0500 Subject: [PATCH] feat(launch-polish): finish dashboard auth trust boundary + pending-review UX MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the launch-polish lane Codex started (rate-limited mid-wiring): Trust boundary (was: dashboard API/WS unauthenticated): - Token middleware on all /api/* and /ws, reusing the persisted Vestige auth token with constant-time compare (subtle); Sec-Fetch-Site + exact-Origin checks - Token delivered via URL fragment (never sent to server / logged), installed into localStorage on app load — WIRES installDashboardTokenFromLocation() into the root +layout onMount BEFORE websocket.connect()/any fetch (the missing piece that otherwise 401s the dashboard against its own API) Fail-closed memory governance: - Dashboard delete/suppress pre-gate through Memory PRs in Risk-Gated/Paranoid (Fast still mutates directly); decision-before-mutation; replay -> 409 - Duplicate pending-PR reuse; PR content redacted to preview+hash at the API layer Black Box / receipts: - Live refresh of run list + receipts on real TraceEvents (trace-before-receipt race handled via retry); synthetic events excluded from proof counters Fixes on top of Codex's work: - websocket.ts: source 'synthetic' -> `as const` (TS error, blocked pnpm check) - memories: scoped a11y-ignore on the non-interactive review-notice - collapsed sanitize_memory_pr_json (clippy collapsible_if); rustfmt the 7 touched Rust files Gates: pnpm check 0/0 (905 files); vestige-mcp 453 tests; core trace/review/ receipt 67 tests; clippy -D warnings clean. All green. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/dashboard/src/lib/stores/api.ts | 102 +- apps/dashboard/src/lib/stores/websocket.ts | 19 +- apps/dashboard/src/lib/types/index.ts | 22 + .../src/routes/(app)/blackbox/+page.svelte | 93 +- .../src/routes/(app)/memories/+page.svelte | 220 +++- .../src/routes/(app)/memory-prs/+page.svelte | 33 +- apps/dashboard/src/routes/+layout.svelte | 6 + .../vestige-core/src/storage/trace_store.rs | 67 +- crates/vestige-mcp/src/dashboard/handlers.rs | 586 ++++++++- crates/vestige-mcp/src/dashboard/mod.rs | 134 +- crates/vestige-mcp/src/dashboard/state.rs | 62 + crates/vestige-mcp/src/dashboard/websocket.rs | 29 +- crates/vestige-mcp/src/server.rs | 1162 ++++++++++------- crates/vestige-mcp/src/trace_recorder.rs | 429 +++++- 14 files changed, 2302 insertions(+), 662 deletions(-) diff --git a/apps/dashboard/src/lib/stores/api.ts b/apps/dashboard/src/lib/stores/api.ts index 950ec99..456d14d 100644 --- a/apps/dashboard/src/lib/stores/api.ts +++ b/apps/dashboard/src/lib/stores/api.ts @@ -11,7 +11,8 @@ import type { RetentionDistribution, ConsolidationResult, IntentionItem, - SuppressResult, + DeleteMemoryResult, + SuppressMemoryResult, UnsuppressResult, SanhedrinAppealReason, SanhedrinAppealResponse, @@ -20,14 +21,81 @@ import type { } from '$types'; const BASE = '/api'; +const DASHBOARD_TOKEN_STORAGE_KEY = 'vestige.dashboard.token'; +const DASHBOARD_TOKEN_HEADER = 'X-Vestige-Dashboard-Token'; async function fetcher(path: string, options?: RequestInit): Promise { + const headers = withDashboardAuthHeaders(options?.headers); const res = await fetch(`${BASE}${path}`, { - headers: { 'Content-Type': 'application/json' }, - ...options + ...options, + headers + }); + if (!res.ok) { + if (res.status === 401 || res.status === 403) { + throw new Error( + `API ${res.status}: dashboard auth failed. Open Vestige from the CLI or set VESTIGE_AUTH_TOKEN.` + ); + } + throw new Error(`API ${res.status}: ${res.statusText}`); + } + return res.json(); +} + +function withDashboardAuthHeaders(headers?: HeadersInit): Headers { + const next = new Headers(headers); + if (!next.has('Content-Type')) { + next.set('Content-Type', 'application/json'); + } + const token = getDashboardAuthToken(); + if (token) { + next.set(DASHBOARD_TOKEN_HEADER, token); + next.set('Authorization', `Bearer ${token}`); + } + return next; +} + +export function getDashboardAuthToken(): string | null { + if (typeof window === 'undefined') return null; + return window.localStorage.getItem(DASHBOARD_TOKEN_STORAGE_KEY); +} + +export function installDashboardTokenFromLocation(): boolean { + if (typeof window === 'undefined') return false; + const hash = window.location.hash.startsWith('#') + ? window.location.hash.slice(1) + : window.location.hash; + if (!hash) return false; + + const params = new URLSearchParams(hash); + const token = params.get('vestige_token') ?? params.get('token'); + if (!token) return false; + + window.localStorage.setItem(DASHBOARD_TOKEN_STORAGE_KEY, token); + params.delete('vestige_token'); + params.delete('token'); + + const remainingHash = params.toString(); + const nextUrl = `${window.location.pathname}${window.location.search}${ + remainingHash ? `#${remainingHash}` : '' + }`; + window.history.replaceState(null, document.title, nextUrl); + return true; +} + +async function downloadJson(path: string, filename: string): Promise { + const res = await fetch(`${BASE}${path}`, { + headers: withDashboardAuthHeaders() }); if (!res.ok) throw new Error(`API ${res.status}: ${res.statusText}`); - return res.json(); + const blob = await res.blob(); + const url = URL.createObjectURL(blob); + const anchor = document.createElement('a'); + anchor.href = url; + anchor.download = filename; + document.body.appendChild(anchor); + anchor.click(); + anchor.remove(); + URL.revokeObjectURL(url); } export const api = { @@ -37,16 +105,16 @@ export const api = { const qs = params ? '?' + new URLSearchParams(params).toString() : ''; return fetcher(`/memories${qs}`); }, - get: (id: string) => fetcher(`/memories/${id}`), - delete: (id: string) => fetcher<{ deleted: boolean }>(`/memories/${id}`, { method: 'DELETE' }), - promote: (id: string) => fetcher(`/memories/${id}/promote`, { method: 'POST' }), - demote: (id: string) => fetcher(`/memories/${id}/demote`, { method: 'POST' }), + get: (id: string) => fetcher(`/memories/${id}`), + delete: (id: string) => fetcher(`/memories/${id}`, { method: 'DELETE' }), + promote: (id: string) => fetcher(`/memories/${id}/promote`, { method: 'POST' }), + demote: (id: string) => fetcher(`/memories/${id}/demote`, { method: 'POST' }), // v2.0.7: suppress + unsuppress. Anderson 2025 top-down inhibitory // control. Each suppress call compounds; reversible within 24h. The // backend emits MemorySuppressed / MemoryUnsuppressed so the 3D graph // plays the violet implosion / rainbow reversal. - suppress: (id: string, reason?: string) => - fetcher(`/memories/${id}/suppress`, { + suppress: (id: string, reason?: string) => + fetcher(`/memories/${id}/suppress`, { method: 'POST', body: reason ? JSON.stringify({ reason }) : undefined }), @@ -138,9 +206,19 @@ export const api = { // Agent Black Box (v2.2): replayable agent-run traces. The runId in a tool // result threads through here unchanged — one id, end to end. traces: { - list: (limit = 50) => fetcher(`/traces?limit=${limit}`), + list: (limit = 50, runId?: string) => { + const qs = new URLSearchParams(); + qs.set('limit', String(limit)); + if (runId) qs.set('run', runId); + return fetcher(`/traces?${qs.toString()}`); + }, get: (runId: string) => fetcher(`/traces/${encodeURIComponent(runId)}`), - exportUrl: (runId: string) => `${BASE}/traces/${encodeURIComponent(runId)}/export` + exportUrl: (runId: string) => `${BASE}/traces/${encodeURIComponent(runId)}/export`, + download: (runId: string) => + downloadJson( + `/traces/${encodeURIComponent(runId)}/export`, + `${runId}.vestige-trace.json` + ) }, // Memory Receipts (v2.2): the nutrition label for a retrieval. diff --git a/apps/dashboard/src/lib/stores/websocket.ts b/apps/dashboard/src/lib/stores/websocket.ts index 2e591ac..08c39df 100644 --- a/apps/dashboard/src/lib/stores/websocket.ts +++ b/apps/dashboard/src/lib/stores/websocket.ts @@ -1,5 +1,6 @@ import { writable, derived } from 'svelte/store'; import type { VestigeEvent } from '$types'; +import { getDashboardAuthToken } from './api'; const MAX_EVENTS = 200; @@ -23,9 +24,11 @@ function createWebSocketStore() { let reconnectAttempts = 0; function connect(url?: string) { - const wsUrl = url || (window.location.port === '5173' + const baseWsUrl = url || (window.location.port === '5173' ? `ws://${window.location.hostname}:3927/ws` : `ws://${window.location.host}/ws`); + const token = getDashboardAuthToken(); + const wsUrl = token ? withToken(baseWsUrl, token) : baseWsUrl; if (ws?.readyState === WebSocket.OPEN) return; @@ -93,7 +96,7 @@ function createWebSocketStore() { */ function injectEvent(event: VestigeEvent) { update(s => { - const events = [event, ...s.events].slice(0, MAX_EVENTS); + const events = [{ ...event, source: 'synthetic' as const }, ...s.events].slice(0, MAX_EVENTS); return { ...s, events }; }); } @@ -107,6 +110,12 @@ function createWebSocketStore() { }; } +function withToken(url: string, token: string): string { + const next = new URL(url, window.location.href); + next.searchParams.set('token', token); + return next.toString(); +} + export const websocket = createWebSocketStore(); // Derived stores for specific event types @@ -136,19 +145,19 @@ export const uptimeSeconds = derived(websocket, $ws => // is a real `VestigeEvent::TraceEvent` backed by a persisted `agent_traces` // row — the dashboard pulse is only ever driven by these, never by fakes. export const traceEvents = derived(websocket, $ws => - $ws.events.filter((e) => e.type === 'TraceEvent') + $ws.events.filter((e) => e.type === 'TraceEvent' && e.source !== 'synthetic') ); // The most recent runId seen on the live feed — the "current run" indicator in // Proof Mode / the Black Box live header. export const liveRunId = derived(websocket, $ws => { - const latest = $ws.events.find((e) => e.type === 'TraceEvent'); + const latest = $ws.events.find((e) => e.type === 'TraceEvent' && e.source !== 'synthetic'); return (latest?.data?.run_id as string) ?? null; }); // The single most recent trace event (for the "last event" readout). export const lastTraceEvent = derived(websocket, $ws => - $ws.events.find((e) => e.type === 'TraceEvent') ?? null + $ws.events.find((e) => e.type === 'TraceEvent' && e.source !== 'synthetic') ?? null ); // Live Memory PR notifications (opened / decided) for the queue badge + toasts. diff --git a/apps/dashboard/src/lib/types/index.ts b/apps/dashboard/src/lib/types/index.ts index 2989391..e8d8a41 100644 --- a/apps/dashboard/src/lib/types/index.ts +++ b/apps/dashboard/src/lib/types/index.ts @@ -176,6 +176,7 @@ export type VestigeEventType = export interface VestigeEvent { type: VestigeEventType; data: Record; + source?: 'socket' | 'synthetic'; } // v2.0.7: active-forgetting response shapes. Each suppress call COMPOUNDS; @@ -196,6 +197,27 @@ export interface SuppressResult { reason: string | null; } +export interface MemoryPrOpenedSummary { + id: string; + kind: string; + title: string; + signals?: Array>; + subjectId?: string; +} + +export interface PendingReviewResult { + action: string; + success: false; + pendingReview: true; + nodeId: string; + message: string; + memoryPrsOpened: MemoryPrOpenedSummary[]; + memoryPrNotice?: string; +} + +export type DeleteMemoryResult = { deleted: true; id: string } | PendingReviewResult; +export type SuppressMemoryResult = SuppressResult | PendingReviewResult; + export interface UnsuppressResult { unsuppressed: true; id: string; diff --git a/apps/dashboard/src/routes/(app)/blackbox/+page.svelte b/apps/dashboard/src/routes/(app)/blackbox/+page.svelte index 3187bf8..d4cfe7d 100644 --- a/apps/dashboard/src/routes/(app)/blackbox/+page.svelte +++ b/apps/dashboard/src/routes/(app)/blackbox/+page.svelte @@ -13,7 +13,7 @@ // Live events are real — they arrive over the WebSocket backed by trace // rows. No fake demo events. // ═══════════════════════════════════════════════════════════════════════ - import { onMount } from 'svelte'; + import { onDestroy, onMount } from 'svelte'; import PageHeader from '$components/PageHeader.svelte'; import Icon from '$components/Icon.svelte'; import AnimatedNumber from '$components/AnimatedNumber.svelte'; @@ -46,6 +46,8 @@ let scrubIndex = $state(0); // index into detail.events let proofMode = $state(false); let receipts = $state([]); + let liveRefreshTimer: ReturnType | null = null; + let receiptRetryTimer: ReturnType | null = null; // The events up to and including the scrubber position — what the agent had // "experienced" at that moment in the run. @@ -69,11 +71,15 @@ detail?.events.some((e) => e.type === 'contradiction.detected') ?? false ); - async function loadRuns() { + async function loadRuns(preferredRunId?: string | null) { try { const res = await api.traces.list(100); runs = res.runs; - if (!selectedRunId && runs.length) selectRun(runs[0].runId); + if (preferredRunId) { + await selectRun(preferredRunId); + } else if (!selectedRunId && runs.length) { + await selectRun(runs[0].runId); + } } catch (e) { error = String(e); } @@ -84,11 +90,11 @@ loading = true; error = null; try { - detail = await api.traces.get(runId); - scrubIndex = Math.max(0, (detail.events.length || 1) - 1); - // Receipts are the proof behind THIS run's retrievals — scoped to - // the selected run (B5), not the global latest. - receipts = (await api.receipts.listForRun(runId, 8)).receipts; + const next = await api.traces.get(runId); + if (selectedRunId !== runId) return; + detail = next; + scrubIndex = Math.max(0, (next.events.length || 1) - 1); + await loadReceiptsForRun(runId); } catch (e) { error = String(e); detail = null; @@ -97,30 +103,77 @@ } } + async function refreshSelectedRun(runId: string, pinToLatest = false) { + const wasPinnedToLatest = !detail || scrubIndex >= detail.events.length - 1; + const next = await api.traces.get(runId); + if (selectedRunId !== runId) return; + detail = next; + if (pinToLatest || wasPinnedToLatest) { + scrubIndex = Math.max(0, next.events.length - 1); + } + } + + async function loadReceiptsForRun(runId: string) { + const next = (await api.receipts.listForRun(runId, 8)).receipts; + if (selectedRunId === runId) { + receipts = next; + } + } + + function scheduleReceiptRetry(runId: string) { + if (receiptRetryTimer) clearTimeout(receiptRetryTimer); + receiptRetryTimer = setTimeout(() => { + if (selectedRunId === runId) { + void loadReceiptsForRun(runId); + } + }, 300); + } + + function scheduleLiveRefresh(runId: string, eventType?: TraceEvent['type']) { + if (liveRefreshTimer) clearTimeout(liveRefreshTimer); + liveRefreshTimer = setTimeout(async () => { + await loadRuns(); + if (!selectedRunId) { + await selectRun(runId); + return; + } + if (selectedRunId !== runId) return; + await refreshSelectedRun(runId, true); + await loadReceiptsForRun(runId); + if (eventType === 'memory.retrieve') { + scheduleReceiptRetry(runId); + } + }, 75); + } + function exportTrace() { if (!selectedRunId) return; // Direct browser download of the .vestige-trace.json artifact. window.location.href = api.traces.exportUrl(selectedRunId); } - // Live: when a trace event for the *currently open* run arrives, refresh it - // so the timeline grows in real time. Also refresh the run list so new runs - // appear at the top. + // Live: when a real trace event arrives, refresh the run picker and the open + // trace. Receipts can be saved just after the TraceEvent broadcast, so + // retrieval events get one delayed retry to avoid a stale proof panel. $effect(() => { const last = $lastTraceEvent; if (!last) return; - const evRunId = last.data?.run_id as string | undefined; - if (evRunId && evRunId === selectedRunId) { - // Re-fetch the open run (cheap; trace rows are local SQLite). - api.traces.get(selectedRunId).then((d) => { - detail = d; - // Keep the scrubber pinned to the newest event in live mode. - scrubIndex = Math.max(0, d.events.length - 1); - }); + const event = last.data?.event as TraceEvent | undefined; + const evRunId = (last.data?.run_id as string | undefined) ?? event?.runId; + if (evRunId) { + scheduleLiveRefresh(evRunId, event?.type); } }); - onMount(loadRuns); + onMount(() => { + const preferredRunId = new URLSearchParams(window.location.search).get('run'); + loadRuns(preferredRunId); + }); + + onDestroy(() => { + if (liveRefreshTimer) clearTimeout(liveRefreshTimer); + if (receiptRetryTimer) clearTimeout(receiptRetryTimer); + });
diff --git a/apps/dashboard/src/routes/(app)/memories/+page.svelte b/apps/dashboard/src/routes/(app)/memories/+page.svelte index 393f84b..73cdbf8 100644 --- a/apps/dashboard/src/routes/(app)/memories/+page.svelte +++ b/apps/dashboard/src/routes/(app)/memories/+page.svelte @@ -1,7 +1,7 @@
@@ -270,11 +274,15 @@
{kindLabel[selected.kind] ?? selected.kind} {selected.status} - {#if selected.run_id} - - {selected.run_id.replace('run_', '').slice(0, 8)} - - {/if} + {#if selected.run_id} + + {selected.run_id.replace('run_', '').slice(0, 8)} + + {/if}

{selected.title}

@@ -284,10 +292,13 @@ {#if diffNodeType(selected)} type: {diffNodeType(selected)} {/if} - {#each diffTags(selected) as t (t)} - #{t} - {/each} -
+ {#each diffTags(selected) as t (t)} + #{t} + {/each} + {#if diffHash(selected)} + hash: {diffHash(selected)} + {/if} +
{#if diffContent(selected)} 
+{diffContent(selected)}
{/if} diff --git a/apps/dashboard/src/routes/+layout.svelte b/apps/dashboard/src/routes/+layout.svelte index 53f5a52..805e889 100644 --- a/apps/dashboard/src/routes/+layout.svelte +++ b/apps/dashboard/src/routes/+layout.svelte @@ -20,6 +20,7 @@ import ThemeToggle from '$lib/components/ThemeToggle.svelte'; import Icon, { type IconName } from '$lib/components/Icon.svelte'; import { initTheme } from '$stores/theme'; + import { installDashboardTokenFromLocation } from '$stores/api'; let { children } = $props(); let showCommandPalette = $state(false); @@ -31,6 +32,11 @@ let isMarketingRoute = $derived(dashboardPath === '/waitlist' || dashboardPath.startsWith('/waitlist/')); onMount(() => { + // Vestige opens the dashboard with the auth token in the URL fragment. + // Install it into localStorage (and scrub it from the address bar) BEFORE + // any API call or WebSocket connect, or the now-authenticated /api and /ws + // endpoints reject every request. + installDashboardTokenFromLocation(); if (!isMarketingRoute) { websocket.connect(); } diff --git a/crates/vestige-core/src/storage/trace_store.rs b/crates/vestige-core/src/storage/trace_store.rs index ee39751..bff0b6d 100644 --- a/crates/vestige-core/src/storage/trace_store.rs +++ b/crates/vestige-core/src/storage/trace_store.rs @@ -11,7 +11,7 @@ //! map rows back through a small closure. use chrono::Utc; -use rusqlite::{params, OptionalExtension}; +use rusqlite::{OptionalExtension, params}; use uuid::Uuid; use super::sqlite::SqliteMemoryStore; @@ -136,9 +136,8 @@ impl SqliteMemoryStore { .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 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) @@ -266,9 +265,8 @@ impl SqliteMemoryStore { .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 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) @@ -436,7 +434,9 @@ impl SqliteMemoryStore { .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", + "UPDATE memory_prs + SET status = ?1, decision = ?2, decided_at = ?3 + WHERE id = ?4 AND status = 'pending'", params![new_status.as_str(), decision, now, id], )?; if changed == 0 { @@ -458,10 +458,11 @@ impl SqliteMemoryStore { .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 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()); + let decision = + decision_s.and_then(|s| serde_json::from_value(serde_json::Value::String(s)).ok()); Ok(MemoryPr { id: row.get("id")?, @@ -625,6 +626,39 @@ mod tests { assert_eq!(s.count_pending_memory_prs().unwrap(), 0); } + #[test] + fn decided_memory_pr_cannot_be_decided_again() { + let s = store(); + let pr = MemoryPr { + id: "pr_once".into(), + kind: MemoryPrKind::ContradictionDetected, + status: MemoryPrStatus::Pending, + title: "Decide once".into(), + diff: serde_json::json!({"before": "x", "after": "y"}), + signals: vec![], + 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(); + + let decided = s + .decide_memory_pr("pr_once", MemoryPrAction::Promote) + .unwrap(); + assert_eq!(decided.status, MemoryPrStatus::Promoted); + + assert!( + s.decide_memory_pr("pr_once", MemoryPrAction::Forget) + .is_err(), + "a decided Memory PR must not accept a second decision" + ); + let still_promoted = s.get_memory_pr("pr_once").unwrap().unwrap(); + assert_eq!(still_promoted.status, MemoryPrStatus::Promoted); + assert_eq!(still_promoted.decision, Some(MemoryPrAction::Promote)); + } + #[test] fn promote_releases_a_quarantined_memory_end_to_end() { // B1 regression: the full quarantine→release cycle at the storage layer. @@ -649,9 +683,7 @@ mod tests { // 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"); + 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" @@ -723,9 +755,10 @@ mod tests { decision: None, }; s.save_memory_pr(&pr).unwrap(); - assert!(s - .decide_memory_pr("pr_2", MemoryPrAction::AskAgentWhy) - .is_err()); + assert!( + s.decide_memory_pr("pr_2", MemoryPrAction::AskAgentWhy) + .is_err() + ); // Still pending. assert_eq!(s.count_pending_memory_prs().unwrap(), 1); } diff --git a/crates/vestige-mcp/src/dashboard/handlers.rs b/crates/vestige-mcp/src/dashboard/handlers.rs index 6bff8f4..2ac3c33 100644 --- a/crates/vestige-mcp/src/dashboard/handlers.rs +++ b/crates/vestige-mcp/src/dashboard/handlers.rs @@ -161,6 +161,19 @@ pub async fn delete_memory( State(state): State, Path(id): Path, ) -> Result, StatusCode> { + if let Some(gated) = gate_dashboard_pending_memory_mutation( + &state, + "memory", + serde_json::json!({ + "action": "delete", + "id": id.clone(), + "confirm": true, + "reason": "Dashboard delete requested", + }), + )? { + return Ok(Json(gated)); + } + let deleted = state .storage .delete_node(&id) @@ -250,6 +263,17 @@ pub async fn suppress_memory( .and_then(|r| r.as_str()) .map(String::from); + if let Some(gated) = gate_dashboard_pending_memory_mutation( + &state, + "suppress", + serde_json::json!({ + "id": id.clone(), + "reason": reason.clone(), + }), + )? { + return Ok(Json(gated)); + } + let sys = ActiveForgettingSystem::new(); // Pre-count to surface in the response + for the event payload. @@ -310,6 +334,26 @@ pub async fn suppress_memory( }))) } +fn gate_dashboard_pending_memory_mutation( + state: &AppState, + tool: &str, + args: Value, +) -> Result, StatusCode> { + let run_id = format!("dashboard_{}", uuid::Uuid::new_v4().simple()); + crate::trace_recorder::gate_pending_memory_mutation( + &state.storage, + Some(&state.event_tx), + &run_id, + tool, + &Some(args), + read_review_mode(state), + ) + .map_err(|e| { + tracing::warn!("dashboard memory mutation review gate failed closed: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + }) +} + /// Reverse a prior suppression, if still inside the 24-hour labile /// window. Emits `MemoryUnsuppressed` so the graph plays the rainbow /// reversal burst. Returns the current suppression state so the UI @@ -2000,10 +2044,18 @@ pub async fn list_traces( Query(params): Query, ) -> Result, 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 = match params.run.as_deref().filter(|r| !r.is_empty()) { + Some(run_id) => state + .storage + .get_agent_run(run_id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .into_iter() + .collect(), + None => state + .storage + .list_agent_runs(limit) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?, + }; let runs_json: Vec = runs .into_iter() .map(|r| { @@ -2090,7 +2142,13 @@ pub async fn export_trace( // 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 { '_' }) + .map(|c| { + if c.is_ascii_alphanumeric() || c == '_' || c == '-' { + c + } else { + '_' + } + }) .collect(); let safe = if safe.trim_matches('_').is_empty() { "trace".to_string() @@ -2172,6 +2230,10 @@ pub async fn list_memory_prs( .storage .list_memory_prs(status, limit) .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let prs: Vec = prs + .into_iter() + .map(|pr| sanitize_memory_pr_json(serde_json::to_value(pr).unwrap_or_default())) + .collect(); let pending = state.storage.count_pending_memory_prs().unwrap_or(0); Ok(Json(serde_json::json!({ "total": prs.len(), @@ -2191,7 +2253,25 @@ pub async fn get_memory_pr( .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())) + Ok(Json(sanitize_memory_pr_json( + serde_json::to_value(pr).unwrap_or_default(), + ))) +} + +fn sanitize_memory_pr_json(mut value: Value) -> Value { + if let Some(node) = value + .get_mut("diff") + .and_then(|diff| diff.get_mut("node")) + .and_then(|node| node.as_object_mut()) + && node.remove("content").is_some() + && !node.contains_key("contentPreview") + { + node.insert( + "contentPreview".to_string(), + serde_json::json!("[redacted legacy content]"), + ); + } + value } /// Act on a Memory PR: promote / merge / supersede / quarantine / forget / @@ -2200,25 +2280,141 @@ pub async fn act_on_memory_pr( State(state): State, Path((id, action)): Path<(String, String)>, ) -> Result, StatusCode> { - let action = vestige_core::MemoryPrAction::from_label(&action) - .ok_or(StatusCode::BAD_REQUEST)?; + let action = + vestige_core::MemoryPrAction::from_label(&action).ok_or(StatusCode::BAD_REQUEST)?; + + let current_pr = state + .storage + .get_memory_pr(&id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .ok_or(StatusCode::NOT_FOUND)?; // 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, + "id": current_pr.id, + "kind": current_pr.kind.as_str(), + "title": current_pr.title, + "why": current_pr.signals, "explanation": "These are the risk signals that opened this Memory PR.", }))); } + if current_pr.status != vestige_core::MemoryPrStatus::Pending { + return Err(StatusCode::CONFLICT); + } + + let pending_action = current_pr + .diff + .get("pendingAction") + .and_then(|v| v.as_str()) + .map(str::to_string); + + let mut mutation_applied = false; + let mut mutation_report = serde_json::Value::Null; + + // Pending destructive/suppressive PRs are opened before the mutation runs. + // Deciding them is therefore the place where the requested operation is + // either applied (`forget`/`quarantine`) or kept (`promote`/accept actions). + if let Some(pending) = pending_action.as_deref() + && let Some(subject_id) = current_pr.subject_id.as_deref() + { + match (pending, action) { + ("purge" | "delete", vestige_core::MemoryPrAction::Forget) => { + let reason = current_pr + .diff + .get("reason") + .and_then(|v| v.as_str()) + .unwrap_or("approved destructive Memory PR"); + let report = state + .storage + .purge_node(subject_id, Some(reason)) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + if report.deleted { + state.emit(VestigeEvent::MemoryDeleted { + id: subject_id.to_string(), + timestamp: Utc::now(), + }); + } + mutation_applied = report.deleted; + mutation_report = serde_json::json!({ + "action": pending, + "deleted": report.deleted, + "deletedAt": report.deleted_at.to_rfc3339(), + "edgesPruned": report.edges_pruned, + "insightsRewritten": report.insights_rewritten, + "insightsDeleted": report.insights_deleted, + "childrenOrphaned": report.children_orphaned, + }); + if let Some(run_id) = current_pr.run_id.as_deref() { + crate::trace_recorder::record_result( + &state.storage, + Some(&state.event_tx), + run_id, + "memory", + &serde_json::json!({ + "action": pending, + "nodeId": subject_id, + "success": report.deleted + }), + ); + } + } + ( + "purge" | "delete" | "suppress", + vestige_core::MemoryPrAction::Quarantine | vestige_core::MemoryPrAction::Forget, + ) => { + let node = state + .storage + .suppress_memory(subject_id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let estimated_cascade = state + .storage + .get_connections_for_memory(subject_id) + .map(|edges| edges.len().min(100)) + .unwrap_or(0); + let sys = + vestige_core::neuroscience::active_forgetting::ActiveForgettingSystem::new(); + let reversible_until = node + .suppressed_at + .map(|t| sys.reversible_until(t)) + .unwrap_or_else(Utc::now); + state.emit(VestigeEvent::MemorySuppressed { + id: node.id.clone(), + suppression_count: node.suppression_count, + estimated_cascade, + reversible_until, + timestamp: Utc::now(), + }); + mutation_applied = true; + mutation_report = serde_json::json!({ + "action": "suppress", + "suppressionCount": node.suppression_count, + "reversibleUntil": reversible_until.to_rfc3339(), + }); + if let Some(run_id) = current_pr.run_id.as_deref() { + crate::trace_recorder::record_result( + &state.storage, + Some(&state.event_tx), + run_id, + "suppress", + &serde_json::json!({ + "receipt": { + "suppressed": [ + { + "id": subject_id, + "reason": "low_trust" + } + ] + } + }), + ); + } + } + _ => {} + } + } + let decided = state .storage .decide_memory_pr(&id, action) @@ -2230,6 +2426,7 @@ pub async fn act_on_memory_pr( // retrieval. Forget/Quarantine intentionally keep it suppressed. let mut released = false; if action.releases_memory() + && pending_action.is_none() && let Some(subject_id) = decided.subject_id.as_deref() { // Use the UNCONDITIONAL quarantine release, not reverse_suppression: @@ -2272,6 +2469,13 @@ pub async fn act_on_memory_pr( 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)); + obj.insert( + "pendingMutationApplied".to_string(), + serde_json::json!(mutation_applied), + ); + if !mutation_report.is_null() { + obj.insert("mutationReport".to_string(), mutation_report); + } } Ok(Json(out)) } @@ -2315,7 +2519,10 @@ pub async fn set_review_mode( // 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())?; + write_atomic( + &path, + &serde_json::to_vec_pretty(&payload).unwrap_or_default(), + )?; Ok(Json(serde_json::json!({ "mode": mode.as_str() }))) } @@ -2424,6 +2631,349 @@ mod tests { node.id } + fn pending_mutation_pr(id: &str, pr_id: &str, pending_action: &str) -> vestige_core::MemoryPr { + vestige_core::MemoryPr { + id: pr_id.to_string(), + kind: vestige_core::MemoryPrKind::NodeDecayed, + status: vestige_core::MemoryPrStatus::Pending, + title: format!("Pending {pending_action}"), + diff: serde_json::json!({ + "pendingAction": pending_action, + "reason": format!("test approved {pending_action}"), + "node": { + "id": id, + "deleted": false + } + }), + signals: vec![vestige_core::RiskSignal { + code: "forgets_memory".to_string(), + detail: "Forgets / suppresses an existing memory.".to_string(), + }], + subject_id: Some(id.to_string()), + run_id: Some(format!("run_{pr_id}")), + created_at: Utc::now().to_rfc3339(), + decided_at: None, + decision: None, + } + } + + #[tokio::test] + async fn dashboard_delete_is_pre_gated_before_mutation() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "dashboard delete target"); + let state = AppState::new(storage.clone(), None); + + let Json(out) = delete_memory(State(state), Path(id.clone())) + .await + .expect("dashboard delete should open review"); + + assert_eq!(out["pendingReview"], serde_json::json!(true)); + assert_eq!(out["action"], serde_json::json!("delete_pending_review")); + assert!( + storage.get_node(&id).unwrap().is_some(), + "dashboard delete must not mutate before review in risk-gated mode" + ); + let prs = storage + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); + assert_eq!(prs.len(), 1); + assert_eq!(prs[0].diff["pendingAction"], serde_json::json!("delete")); + } + + #[tokio::test] + async fn dashboard_delete_fast_mode_preserves_direct_mutation() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "dashboard fast delete target"); + let state = AppState::new(storage.clone(), None); + write_atomic(&review_mode_path(&state), br#"{"mode":"fast"}"#).unwrap(); + + let Json(out) = delete_memory(State(state), Path(id.clone())) + .await + .expect("fast mode delete should execute directly"); + + assert_eq!(out["deleted"], serde_json::json!(true)); + assert!(storage.get_node(&id).unwrap().is_none()); + assert_eq!(storage.count_pending_memory_prs().unwrap(), 0); + } + + #[tokio::test] + async fn dashboard_suppress_is_pre_gated_before_mutation() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "dashboard suppress target"); + let state = AppState::new(storage.clone(), None); + + let Json(out) = suppress_memory(State(state), Path(id.clone()), None) + .await + .expect("dashboard suppress should open review"); + + assert_eq!(out["pendingReview"], serde_json::json!(true)); + assert_eq!(out["action"], serde_json::json!("suppress_pending_review")); + let node = storage.get_node(&id).unwrap().unwrap(); + assert_eq!( + node.suppression_count, 0, + "dashboard suppress must not change retrieval influence before review" + ); + let prs = storage + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); + assert_eq!(prs.len(), 1); + assert_eq!(prs[0].diff["pendingAction"], serde_json::json!("suppress")); + } + + #[tokio::test] + async fn dashboard_suppress_fast_mode_preserves_direct_mutation() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "dashboard fast suppress target"); + let state = AppState::new(storage.clone(), None); + write_atomic(&review_mode_path(&state), br#"{"mode":"fast"}"#).unwrap(); + + let Json(out) = suppress_memory(State(state), Path(id.clone()), None) + .await + .expect("fast mode suppress should execute directly"); + + assert_eq!(out["suppressed"], serde_json::json!(true)); + assert_eq!(storage.get_node(&id).unwrap().unwrap().suppression_count, 1); + assert_eq!(storage.count_pending_memory_prs().unwrap(), 0); + } + + #[tokio::test] + async fn list_traces_run_filter_returns_only_requested_run() { + let (_dir, storage) = seed_storage(); + storage + .append_trace_event(&vestige_core::MemoryTraceEvent::McpCall { + run_id: "run_a".to_string(), + tool: "search".to_string(), + args_hash: "ha".to_string(), + at: 100, + }) + .unwrap(); + storage + .append_trace_event(&vestige_core::MemoryTraceEvent::McpCall { + run_id: "run_b".to_string(), + tool: "deep_reference".to_string(), + args_hash: "hb".to_string(), + at: 200, + }) + .unwrap(); + let state = AppState::new(storage, None); + + let Json(out) = list_traces( + State(state), + Query(TraceListParams { + limit: Some(100), + run: Some("run_b".to_string()), + }), + ) + .await + .expect("trace list should load"); + + assert_eq!(out["total"], serde_json::json!(1)); + assert_eq!(out["runs"][0]["runId"], serde_json::json!("run_b")); + } + + #[tokio::test] + async fn pending_purge_pr_forget_applies_delete() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "pending purge target"); + let pr = vestige_core::MemoryPr { + id: "pr_pending_purge".to_string(), + kind: vestige_core::MemoryPrKind::NodeDecayed, + status: vestige_core::MemoryPrStatus::Pending, + title: "Pending purge".to_string(), + diff: serde_json::json!({ + "pendingAction": "purge", + "reason": "test approved purge", + "node": { + "id": id, + "deleted": false + } + }), + signals: vec![vestige_core::RiskSignal { + code: "forgets_memory".to_string(), + detail: "Forgets / suppresses an existing memory.".to_string(), + }], + subject_id: Some(id.clone()), + run_id: Some("run_pending_purge".to_string()), + created_at: Utc::now().to_rfc3339(), + decided_at: None, + decision: None, + }; + storage.save_memory_pr(&pr).unwrap(); + let state = AppState::new(storage.clone(), None); + + let Json(out) = act_on_memory_pr( + State(state), + Path(("pr_pending_purge".to_string(), "forget".to_string())), + ) + .await + .expect("decision should succeed"); + + assert_eq!(out["pendingMutationApplied"], serde_json::json!(true)); + assert_eq!(out["mutationReport"]["deleted"], serde_json::json!(true)); + assert!( + storage.get_node(&id).unwrap().is_none(), + "forget on pending purge PR must perform the irreversible purge" + ); + } + + #[tokio::test] + async fn pending_purge_pr_promote_keeps_memory() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "pending purge keep target"); + let pr = vestige_core::MemoryPr { + id: "pr_keep_purge".to_string(), + kind: vestige_core::MemoryPrKind::NodeDecayed, + status: vestige_core::MemoryPrStatus::Pending, + title: "Pending purge".to_string(), + diff: serde_json::json!({ + "pendingAction": "purge", + "node": { + "id": id, + "deleted": false + } + }), + signals: vec![vestige_core::RiskSignal { + code: "forgets_memory".to_string(), + detail: "Forgets / suppresses an existing memory.".to_string(), + }], + subject_id: Some(id.clone()), + run_id: Some("run_keep_purge".to_string()), + created_at: Utc::now().to_rfc3339(), + decided_at: None, + decision: None, + }; + storage.save_memory_pr(&pr).unwrap(); + let state = AppState::new(storage.clone(), None); + + let Json(out) = act_on_memory_pr( + State(state), + Path(("pr_keep_purge".to_string(), "promote".to_string())), + ) + .await + .expect("decision should succeed"); + + assert_eq!(out["pendingMutationApplied"], serde_json::json!(false)); + assert!( + storage.get_node(&id).unwrap().is_some(), + "promote on pending purge PR must keep the memory" + ); + } + + #[tokio::test] + async fn decided_pending_purge_pr_cannot_be_replayed() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "pending purge replay target"); + storage + .save_memory_pr(&pending_mutation_pr(&id, "pr_replay_purge", "purge")) + .unwrap(); + let state = AppState::new(storage.clone(), None); + + let Json(first) = act_on_memory_pr( + State(state.clone()), + Path(("pr_replay_purge".to_string(), "promote".to_string())), + ) + .await + .expect("first decision should succeed"); + assert_eq!(first["status"], serde_json::json!("promoted")); + + let err = match act_on_memory_pr( + State(state), + Path(("pr_replay_purge".to_string(), "forget".to_string())), + ) + .await + { + Ok(_) => panic!("decided PR must not accept a second destructive decision"), + Err(status) => status, + }; + + assert_eq!(err, StatusCode::CONFLICT); + assert!( + storage.get_node(&id).unwrap().is_some(), + "replaying forget after promote must not delete the memory" + ); + } + + #[tokio::test] + async fn decided_pending_suppress_pr_cannot_be_replayed_or_released() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "pending suppress replay target"); + storage + .save_memory_pr(&pending_mutation_pr(&id, "pr_replay_suppress", "suppress")) + .unwrap(); + let state = AppState::new(storage.clone(), None); + + let Json(first) = act_on_memory_pr( + State(state.clone()), + Path(("pr_replay_suppress".to_string(), "forget".to_string())), + ) + .await + .expect("first decision should apply suppression"); + assert_eq!(first["status"], serde_json::json!("forgotten")); + assert_eq!(storage.get_node(&id).unwrap().unwrap().suppression_count, 1); + + let err = match act_on_memory_pr( + State(state), + Path(("pr_replay_suppress".to_string(), "promote".to_string())), + ) + .await + { + Ok(_) => panic!("decided PR must not accept a second release decision"), + Err(status) => status, + }; + + assert_eq!(err, StatusCode::CONFLICT); + assert_eq!( + storage.get_node(&id).unwrap().unwrap().suppression_count, + 1, + "replaying promote after forget must not release the approved suppression" + ); + } + + #[tokio::test] + async fn pending_suppress_pr_forget_applies_suppression() { + let (_dir, storage) = seed_storage(); + let id = ingest(&storage, "pending suppress target"); + let pr = vestige_core::MemoryPr { + id: "pr_pending_suppress".to_string(), + kind: vestige_core::MemoryPrKind::NodeDecayed, + status: vestige_core::MemoryPrStatus::Pending, + title: "Pending suppress".to_string(), + diff: serde_json::json!({ + "pendingAction": "suppress", + "reason": "test approved suppress", + "node": { + "id": id, + "deleted": false + } + }), + signals: vec![vestige_core::RiskSignal { + code: "forgets_memory".to_string(), + detail: "Forgets / suppresses an existing memory.".to_string(), + }], + subject_id: Some(id.clone()), + run_id: Some("run_pending_suppress".to_string()), + created_at: Utc::now().to_rfc3339(), + decided_at: None, + decision: None, + }; + storage.save_memory_pr(&pr).unwrap(); + let state = AppState::new(storage.clone(), None); + + let Json(out) = act_on_memory_pr( + State(state), + Path(("pr_pending_suppress".to_string(), "forget".to_string())), + ) + .await + .expect("decision should succeed"); + + assert_eq!(out["pendingMutationApplied"], serde_json::json!(true)); + let node = storage.get_node(&id).unwrap().unwrap(); + assert_eq!( + node.suppression_count, 1, + "forget on pending suppress PR must apply suppression" + ); + } + #[test] fn default_center_id_recent_returns_newest_node() { let (_dir, storage) = seed_storage(); diff --git a/crates/vestige-mcp/src/dashboard/mod.rs b/crates/vestige-mcp/src/dashboard/mod.rs index 8f0c77f..681fb7b 100644 --- a/crates/vestige-mcp/src/dashboard/mod.rs +++ b/crates/vestige-mcp/src/dashboard/mod.rs @@ -12,6 +12,11 @@ pub mod static_files; pub mod websocket; use axum::Router; +use axum::body::Body; +use axum::extract::State; +use axum::http::{HeaderMap, HeaderValue, Method, Request, StatusCode, header}; +use axum::middleware::{self, Next}; +use axum::response::{IntoResponse, Response}; use axum::routing::{delete, get, post}; use std::net::SocketAddr; use std::sync::Arc; @@ -25,6 +30,8 @@ use crate::cognitive::CognitiveEngine; use state::AppState; use vestige_core::Storage; +const DASHBOARD_TOKEN_HEADER: &str = "x-vestige-dashboard-token"; + /// Build the axum router with all dashboard routes pub fn build_router( storage: Arc, @@ -47,30 +54,12 @@ pub fn build_router_with_event_tx( } fn build_router_inner(state: AppState, port: u16) -> (Router, AppState) { - #[allow(unused_mut)] - let mut origins = vec![ - format!("http://127.0.0.1:{}", port) - .parse::() - .expect("valid origin"), - format!("http://localhost:{}", port) - .parse::() - .expect("valid origin"), - ]; - - // SvelteKit dev server — only in debug builds - #[cfg(debug_assertions)] - { - origins.push( - "http://localhost:5173" - .parse::() - .expect("valid origin"), - ); - origins.push( - "http://127.0.0.1:5173" - .parse::() - .expect("valid origin"), - ); - } + let origin_strings = dashboard_allowed_origins(port); + let origins: Vec = origin_strings + .iter() + .map(|origin| origin.parse::().expect("valid origin")) + .collect(); + let state = state.with_dashboard_allowed_origins(origin_strings); let cors = CorsLayer::new() .allow_origin(AllowOrigin::list(origins)) @@ -83,6 +72,7 @@ fn build_router_inner(state: AppState, port: u16) -> (Router, AppState) { .allow_headers([ axum::http::header::CONTENT_TYPE, axum::http::header::AUTHORIZATION, + axum::http::HeaderName::from_static(DASHBOARD_TOKEN_HEADER), ]); // Security: restrict WebSocket connections to localhost only (prevents cross-site WS hijacking) @@ -212,6 +202,10 @@ fn build_router_inner(state: AppState, port: u16) -> (Router, AppState) { .layer( ServiceBuilder::new() .concurrency_limit(50) + .layer(middleware::from_fn_with_state( + state.clone(), + require_dashboard_auth, + )) .layer(cors) .layer(csp) .layer(x_frame_options) @@ -224,6 +218,92 @@ fn build_router_inner(state: AppState, port: u16) -> (Router, AppState) { (router, state) } +fn dashboard_allowed_origins(port: u16) -> Vec { + #[allow(unused_mut)] + let mut origins = vec![ + format!("http://127.0.0.1:{}", port), + format!("http://localhost:{}", port), + ]; + + // SvelteKit dev server — only in debug builds. + #[cfg(debug_assertions)] + { + origins.push("http://localhost:5173".to_string()); + origins.push("http://127.0.0.1:5173".to_string()); + } + + origins +} + +async fn require_dashboard_auth( + State(state): State, + request: Request, + next: Next, +) -> Response { + let path = request.uri().path(); + if !path.starts_with("/api/") || request.method() == Method::OPTIONS { + return next.run(request).await; + } + + let headers = request.headers(); + if let Err((status, message)) = validate_dashboard_origin(headers, &state) { + return (status, message).into_response(); + } + if let Err((status, message)) = validate_dashboard_token(headers, &state) { + return (status, message).into_response(); + } + + next.run(request).await +} + +fn validate_dashboard_origin( + headers: &HeaderMap, + state: &AppState, +) -> Result<(), (StatusCode, &'static str)> { + if let Some(fetch_site) = headers + .get("sec-fetch-site") + .and_then(|value| value.to_str().ok()) + && fetch_site == "cross-site" + { + return Err(( + StatusCode::FORBIDDEN, + "Cross-site dashboard request rejected", + )); + } + + let Some(origin) = headers.get(header::ORIGIN).and_then(|v| v.to_str().ok()) else { + return Ok(()); + }; + + if state.is_allowed_dashboard_origin(origin) { + Ok(()) + } else { + Err((StatusCode::FORBIDDEN, "Dashboard origin not allowed")) + } +} + +fn validate_dashboard_token( + headers: &HeaderMap, + state: &AppState, +) -> Result<(), (StatusCode, &'static str)> { + let token = headers + .get(header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .or_else(|| { + headers + .get(DASHBOARD_TOKEN_HEADER) + .and_then(|value| value.to_str().ok()) + }) + .ok_or((StatusCode::UNAUTHORIZED, "Missing dashboard auth token"))?; + + if state.is_valid_dashboard_token(token) { + Ok(()) + } else { + Err((StatusCode::FORBIDDEN, "Invalid dashboard auth token")) + } +} + /// Start the dashboard HTTP server (blocking — use in CLI mode) pub async fn start_dashboard( storage: Arc, @@ -237,7 +317,11 @@ pub async fn start_dashboard( info!("Dashboard starting at http://127.0.0.1:{}", port); if open_browser { - let url = format!("http://127.0.0.1:{}", port); + let url = format!( + "http://127.0.0.1:{}/dashboard#vestige_token={}", + port, + _state.dashboard_token_fragment_value() + ); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_millis(500)).await; let _ = open::that(&url); diff --git a/crates/vestige-mcp/src/dashboard/state.rs b/crates/vestige-mcp/src/dashboard/state.rs index 5a41266..d2f35f3 100644 --- a/crates/vestige-mcp/src/dashboard/state.rs +++ b/crates/vestige-mcp/src/dashboard/state.rs @@ -3,10 +3,12 @@ use std::sync::Arc; use std::time::Instant; use tokio::sync::{Mutex, broadcast}; +use tracing::warn; use vestige_core::Storage; use super::events::VestigeEvent; use crate::cognitive::CognitiveEngine; +use subtle::ConstantTimeEq; /// Broadcast channel capacity — how many events can buffer before old ones drop. const EVENT_CHANNEL_CAPACITY: usize = 1024; @@ -18,6 +20,8 @@ pub struct AppState { pub cognitive: Option>>, pub event_tx: broadcast::Sender, pub start_time: Instant, + dashboard_token: Arc, + dashboard_allowed_origins: Arc>, } impl AppState { @@ -29,6 +33,8 @@ impl AppState { cognitive, event_tx, start_time: Instant::now(), + dashboard_token: load_dashboard_token().into(), + dashboard_allowed_origins: Arc::new(Vec::new()), } } @@ -48,12 +54,68 @@ impl AppState { cognitive, event_tx, start_time: Instant::now(), + dashboard_token: load_dashboard_token().into(), + dashboard_allowed_origins: Arc::new(Vec::new()), } } + /// Attach the exact origins allowed to drive the dashboard API and WS. + pub fn with_dashboard_allowed_origins(mut self, origins: Vec) -> Self { + self.dashboard_allowed_origins = Arc::new(origins); + self + } + + /// Return true when a dashboard token matches the shared Vestige auth token. + pub fn is_valid_dashboard_token(&self, token: &str) -> bool { + let expected = self.dashboard_token.as_bytes(); + let candidate = token.as_bytes(); + candidate.len() == expected.len() && candidate.ct_eq(expected).unwrap_u8() == 1 + } + + /// Return true when an Origin header exactly matches the configured dashboard origins. + pub fn is_allowed_dashboard_origin(&self, origin: &str) -> bool { + self.dashboard_allowed_origins + .iter() + .any(|allowed| allowed == origin) + } + + /// Percent-encode the dashboard token for use inside a URL fragment. + pub fn dashboard_token_fragment_value(&self) -> String { + percent_encode_fragment_value(self.dashboard_token.as_ref()) + } + /// Emit an event to all connected clients. pub fn emit(&self, event: VestigeEvent) { // Ignore send errors (no receivers connected) let _ = self.event_tx.send(event); } } + +fn load_dashboard_token() -> String { + match crate::protocol::auth::get_or_create_auth_token() { + Ok(token) => token, + Err(err) => { + warn!( + "Could not load persisted auth token for dashboard; using process-local token: {}", + err + ); + uuid::Uuid::new_v4().to_string() + } + } +} + +fn percent_encode_fragment_value(value: &str) -> String { + let mut encoded = String::with_capacity(value.len()); + for byte in value.bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => { + encoded.push(byte as char); + } + _ => { + encoded.push('%'); + encoded.push_str(&format!("{:02X}", byte)); + } + } + } + encoded +} diff --git a/crates/vestige-mcp/src/dashboard/websocket.rs b/crates/vestige-mcp/src/dashboard/websocket.rs index 3ee6fac..a023fb7 100644 --- a/crates/vestige-mcp/src/dashboard/websocket.rs +++ b/crates/vestige-mcp/src/dashboard/websocket.rs @@ -3,35 +3,46 @@ //! Clients connect to `/ws` and receive all VestigeEvents as JSON. //! Also sends heartbeats every 5 seconds with system stats. -use axum::extract::State; use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::extract::{Query, State}; use axum::http::{HeaderMap, StatusCode}; use axum::response::IntoResponse; use chrono::Utc; use futures_util::{SinkExt, StreamExt}; +use serde::Deserialize; use tokio::sync::broadcast; use tracing::{debug, warn}; use super::events::VestigeEvent; use super::state::AppState; +#[derive(Debug, Deserialize)] +pub struct WsAuthQuery { + token: Option, +} + /// WebSocket upgrade handler — GET /ws -/// Validates Origin header to prevent cross-site WebSocket hijacking. +/// Validates token + Origin header to prevent cross-site WebSocket hijacking. pub async fn ws_handler( headers: HeaderMap, + Query(query): Query, ws: WebSocketUpgrade, State(state): State, ) -> impl IntoResponse { + let Some(token) = query.token.as_deref() else { + warn!("Rejected WebSocket connection without dashboard token"); + return StatusCode::UNAUTHORIZED.into_response(); + }; + if !state.is_valid_dashboard_token(token) { + warn!("Rejected WebSocket connection with invalid dashboard token"); + return StatusCode::FORBIDDEN.into_response(); + } + // Validate Origin header (browsers always send it for WebSocket upgrades). - // Non-browser clients (curl, wscat) won't have Origin — allowed since localhost-only. + // Non-browser clients (curl, wscat) won't have Origin; they still need the token. match headers.get("origin").and_then(|v| v.to_str().ok()) { Some(origin) => { - let allowed = - origin.starts_with("http://127.0.0.1:") || origin.starts_with("http://localhost:"); - #[cfg(debug_assertions)] - let allowed = - allowed || origin == "http://localhost:5173" || origin == "http://127.0.0.1:5173"; - if !allowed { + if !state.is_allowed_dashboard_origin(origin) { warn!("Rejected WebSocket connection from origin: {}", origin); return StatusCode::FORBIDDEN.into_response(); } diff --git a/crates/vestige-mcp/src/server.rs b/crates/vestige-mcp/src/server.rs index 0511d31..8e7853e 100644 --- a/crates/vestige-mcp/src/server.rs +++ b/crates/vestige-mcp/src/server.rs @@ -595,517 +595,567 @@ impl McpServer { &request.arguments, ); - let result = match request.name.as_str() { - // ================================================================ - // UNIFIED TOOLS (v1.1+) - Preferred API - // ================================================================ - "search" => { - tools::search_unified::execute( - &self.storage, - &self.cognitive, - &self.output_config, - request.arguments, - ) - .await - } - "memory" => { - tools::memory_unified::execute(&self.storage, &self.cognitive, request.arguments) + let pre_gated = crate::trace_recorder::gate_pending_memory_mutation( + &self.storage, + self.event_tx.as_ref(), + &run_id, + &request.name, + &request.arguments, + self.review_mode(), + ); + + let result = if let Some(content) = + pre_gated.map_err(|e| JsonRpcError::internal_error(&e))? + { + Ok(content) + } else { + match request.name.as_str() { + // ================================================================ + // UNIFIED TOOLS (v1.1+) - Preferred API + // ================================================================ + "search" => { + tools::search_unified::execute( + &self.storage, + &self.cognitive, + &self.output_config, + request.arguments, + ) .await - } - "codebase" => { - tools::codebase_unified::execute( - &self.storage, - &self.cognitive, - &self.output_config, - request.arguments, - ) - .await - } - "intention" => { - tools::intention_unified::execute(&self.storage, &self.cognitive, request.arguments) + } + "memory" => { + tools::memory_unified::execute( + &self.storage, + &self.cognitive, + request.arguments, + ) .await - } - - // ================================================================ - // Core memory (v1.7: smart_ingest absorbs ingest + checkpoint) - // ================================================================ - "smart_ingest" => { - tools::smart_ingest::execute(&self.storage, &self.cognitive, request.arguments) + } + "codebase" => { + tools::codebase_unified::execute( + &self.storage, + &self.cognitive, + &self.output_config, + request.arguments, + ) .await - } - - // ================================================================ - // External-source connectors (#57) - // ================================================================ - "source_sync" => tools::source_sync::execute(&self.storage, request.arguments).await, - - // ================================================================ - // DEPRECATED (v1.7): ingest → smart_ingest - // ================================================================ - "ingest" => { - warn!("Tool 'ingest' is deprecated in v1.7. Use 'smart_ingest' instead."); - tools::smart_ingest::execute(&self.storage, &self.cognitive, request.arguments) + } + "intention" => { + tools::intention_unified::execute( + &self.storage, + &self.cognitive, + request.arguments, + ) .await - } + } - // ================================================================ - // DEPRECATED (v1.7): session_checkpoint → smart_ingest (batch mode) - // ================================================================ - "session_checkpoint" => { - warn!( - "Tool 'session_checkpoint' is deprecated in v1.7. Use 'smart_ingest' with 'items' parameter instead." - ); - tools::smart_ingest::execute(&self.storage, &self.cognitive, request.arguments) - .await - } + // ================================================================ + // Core memory (v1.7: smart_ingest absorbs ingest + checkpoint) + // ================================================================ + "smart_ingest" => { + tools::smart_ingest::execute(&self.storage, &self.cognitive, request.arguments) + .await + } - // ================================================================ - // DEPRECATED (v1.7): promote_memory → memory(action='promote') - // ================================================================ - "promote_memory" => { - warn!( - "Tool 'promote_memory' is deprecated in v1.7. Use 'memory' with action='promote' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert("action".to_string(), serde_json::json!("promote")); - } - Some(new_args) - } - None => Some(serde_json::json!({"action": "promote"})), - }; - tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args).await - } - "demote_memory" => { - warn!( - "Tool 'demote_memory' is deprecated in v1.7. Use 'memory' with action='demote' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert("action".to_string(), serde_json::json!("demote")); - } - Some(new_args) - } - None => Some(serde_json::json!({"action": "demote"})), - }; - tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args).await - } + // ================================================================ + // External-source connectors (#57) + // ================================================================ + "source_sync" => { + tools::source_sync::execute(&self.storage, request.arguments).await + } - // ================================================================ - // DEPRECATED (v1.7): health_check, stats → system_status - // ================================================================ - "health_check" => { - warn!("Tool 'health_check' is deprecated in v1.7. Use 'system_status' instead."); - tools::maintenance::execute_system_status( - &self.storage, - &self.cognitive, - request.arguments, - ) - .await - } - "stats" => { - warn!("Tool 'stats' is deprecated in v1.7. Use 'system_status' instead."); - tools::maintenance::execute_system_status( - &self.storage, - &self.cognitive, - request.arguments, - ) - .await - } + // ================================================================ + // DEPRECATED (v1.7): ingest → smart_ingest + // ================================================================ + "ingest" => { + warn!("Tool 'ingest' is deprecated in v1.7. Use 'smart_ingest' instead."); + tools::smart_ingest::execute(&self.storage, &self.cognitive, request.arguments) + .await + } - // ================================================================ - // SYSTEM STATUS (v1.7: replaces health_check + stats) - // ================================================================ - "system_status" => { - tools::maintenance::execute_system_status( - &self.storage, - &self.cognitive, - request.arguments, - ) - .await - } + // ================================================================ + // DEPRECATED (v1.7): session_checkpoint → smart_ingest (batch mode) + // ================================================================ + "session_checkpoint" => { + warn!( + "Tool 'session_checkpoint' is deprecated in v1.7. Use 'smart_ingest' with 'items' parameter instead." + ); + tools::smart_ingest::execute(&self.storage, &self.cognitive, request.arguments) + .await + } - "mark_reviewed" => tools::review::execute(&self.storage, request.arguments).await, - - // ================================================================ - // DEPRECATED: Search tools - redirect to unified 'search' - // ================================================================ - "recall" | "semantic_search" | "hybrid_search" => { - warn!( - "Tool '{}' is deprecated. Use 'search' instead.", - request.name - ); - tools::search_unified::execute( - &self.storage, - &self.cognitive, - &self.output_config, - request.arguments, - ) - .await - } - - // ================================================================ - // DEPRECATED: Memory tools - redirect to unified 'memory' - // ================================================================ - "get_knowledge" => { - warn!( - "Tool 'get_knowledge' is deprecated. Use 'memory' with action='get' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let id = args.get("id").cloned().unwrap_or(serde_json::Value::Null); - Some(serde_json::json!({ - "action": "get", - "id": id - })) - } - None => None, - }; - tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args).await - } - "delete_knowledge" => { - warn!( - "Tool 'delete_knowledge' is deprecated. Use 'memory' with action='purge', confirm=true instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let id = args.get("id").cloned().unwrap_or(serde_json::Value::Null); - let confirm = args - .get("confirm") - .cloned() - .unwrap_or(serde_json::Value::Bool(false)); - Some(serde_json::json!({ - "action": "delete", - "id": id, - "confirm": confirm - })) - } - None => None, - }; - tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args).await - } - "get_memory_state" => { - warn!( - "Tool 'get_memory_state' is deprecated. Use 'memory' with action='state' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let id = args - .get("memory_id") - .cloned() - .unwrap_or(serde_json::Value::Null); - Some(serde_json::json!({ - "action": "state", - "id": id - })) - } - None => None, - }; - tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args).await - } - - // ================================================================ - // DEPRECATED: Codebase tools - redirect to unified 'codebase' - // ================================================================ - "remember_pattern" => { - warn!( - "Tool 'remember_pattern' is deprecated. Use 'codebase' with action='remember_pattern' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert("action".to_string(), serde_json::json!("remember_pattern")); - } - Some(new_args) - } - None => Some(serde_json::json!({"action": "remember_pattern"})), - }; - tools::codebase_unified::execute( - &self.storage, - &self.cognitive, - &self.output_config, - unified_args, - ) - .await - } - "remember_decision" => { - warn!( - "Tool 'remember_decision' is deprecated. Use 'codebase' with action='remember_decision' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert( - "action".to_string(), - serde_json::json!("remember_decision"), - ); - } - Some(new_args) - } - None => Some(serde_json::json!({"action": "remember_decision"})), - }; - tools::codebase_unified::execute( - &self.storage, - &self.cognitive, - &self.output_config, - unified_args, - ) - .await - } - "get_codebase_context" => { - warn!( - "Tool 'get_codebase_context' is deprecated. Use 'codebase' with action='get_context' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert("action".to_string(), serde_json::json!("get_context")); - } - Some(new_args) - } - None => Some(serde_json::json!({"action": "get_context"})), - }; - tools::codebase_unified::execute( - &self.storage, - &self.cognitive, - &self.output_config, - unified_args, - ) - .await - } - - // ================================================================ - // DEPRECATED: Intention tools - redirect to unified 'intention' - // ================================================================ - "set_intention" => { - warn!( - "Tool 'set_intention' is deprecated. Use 'intention' with action='set' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert("action".to_string(), serde_json::json!("set")); - } - Some(new_args) - } - None => Some(serde_json::json!({"action": "set"})), - }; - tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) - .await - } - "check_intentions" => { - warn!( - "Tool 'check_intentions' is deprecated. Use 'intention' with action='check' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert("action".to_string(), serde_json::json!("check")); - } - Some(new_args) - } - None => Some(serde_json::json!({"action": "check"})), - }; - tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) - .await - } - "complete_intention" => { - warn!( - "Tool 'complete_intention' is deprecated. Use 'intention' with action='update', status='complete' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let id = args - .get("intentionId") - .cloned() - .unwrap_or(serde_json::Value::Null); - Some(serde_json::json!({ - "action": "update", - "id": id, - "status": "complete" - })) - } - None => None, - }; - tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) - .await - } - "snooze_intention" => { - warn!( - "Tool 'snooze_intention' is deprecated. Use 'intention' with action='update', status='snooze' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let id = args - .get("intentionId") - .cloned() - .unwrap_or(serde_json::Value::Null); - let minutes = args - .get("minutes") - .cloned() - .unwrap_or(serde_json::json!(30)); - Some(serde_json::json!({ - "action": "update", - "id": id, - "status": "snooze", - "snooze_minutes": minutes - })) - } - None => None, - }; - tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) - .await - } - "list_intentions" => { - warn!( - "Tool 'list_intentions' is deprecated. Use 'intention' with action='list' instead." - ); - let unified_args = match request.arguments { - Some(ref args) => { - let mut new_args = args.clone(); - if let Some(obj) = new_args.as_object_mut() { - obj.insert("action".to_string(), serde_json::json!("list")); - if let Some(status) = obj.remove("status") { - obj.insert("filter_status".to_string(), status); + // ================================================================ + // DEPRECATED (v1.7): promote_memory → memory(action='promote') + // ================================================================ + "promote_memory" => { + warn!( + "Tool 'promote_memory' is deprecated in v1.7. Use 'memory' with action='promote' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert("action".to_string(), serde_json::json!("promote")); } + Some(new_args) } - Some(new_args) - } - None => Some(serde_json::json!({"action": "list"})), - }; - tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) + None => Some(serde_json::json!({"action": "promote"})), + }; + tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + "demote_memory" => { + warn!( + "Tool 'demote_memory' is deprecated in v1.7. Use 'memory' with action='demote' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert("action".to_string(), serde_json::json!("demote")); + } + Some(new_args) + } + None => Some(serde_json::json!({"action": "demote"})), + }; + tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + + // ================================================================ + // DEPRECATED (v1.7): health_check, stats → system_status + // ================================================================ + "health_check" => { + warn!( + "Tool 'health_check' is deprecated in v1.7. Use 'system_status' instead." + ); + tools::maintenance::execute_system_status( + &self.storage, + &self.cognitive, + request.arguments, + ) .await - } - - // ================================================================ - // Neuroscience tools (internal, not in tools/list) - // ================================================================ - "list_by_state" => { - tools::memory_states::execute_list(&self.storage, request.arguments).await - } - "state_stats" => tools::memory_states::execute_stats(&self.storage).await, - "trigger_importance" => { - tools::tagging::execute_trigger(&self.storage, request.arguments).await - } - "find_tagged" => tools::tagging::execute_find(&self.storage, request.arguments).await, - "tagging_stats" => tools::tagging::execute_stats(&self.storage).await, - "match_context" => tools::context::execute(&self.storage, request.arguments).await, - - // ================================================================ - // Feedback (internal, still used by request_feedback) - // ================================================================ - "request_feedback" => { - tools::feedback::execute_request_feedback(&self.storage, request.arguments).await - } - - // ================================================================ - // TEMPORAL TOOLS (v1.2+) - // ================================================================ - "memory_timeline" => { - tools::timeline::execute(&self.storage, &self.output_config, request.arguments) + } + "stats" => { + warn!("Tool 'stats' is deprecated in v1.7. Use 'system_status' instead."); + tools::maintenance::execute_system_status( + &self.storage, + &self.cognitive, + request.arguments, + ) .await - } - "memory_changelog" => tools::changelog::execute(&self.storage, request.arguments).await, + } - // ================================================================ - // MAINTENANCE TOOLS (v1.2+, non-deprecated) - // ================================================================ - "consolidate" => { - self.emit(VestigeEvent::ConsolidationStarted { - timestamp: chrono::Utc::now(), - }); - tools::maintenance::execute_consolidate(&self.storage, request.arguments).await - } - "backup" => tools::maintenance::execute_backup(&self.storage, request.arguments).await, - "export" => tools::maintenance::execute_export(&self.storage, request.arguments).await, - "gc" => tools::maintenance::execute_gc(&self.storage, request.arguments).await, - - // ================================================================ - // AUTO-SAVE & DEDUP TOOLS (v1.3+) - // ================================================================ - "importance_score" => { - tools::importance::execute(&self.storage, &self.cognitive, request.arguments).await - } - "find_duplicates" => tools::dedup::execute(&self.storage, request.arguments).await, - - // ================================================================ - // MERGE / SUPERSEDE CONTROLS (v2.1.25 — Phase 3) - // ================================================================ - "merge_candidates" | "plan_merge" | "plan_supersede" | "apply_plan" | "merge_undo" - | "protect" | "merge_policy" => { - tools::merge::execute(&self.storage, request.name.as_str(), request.arguments).await - } - - // ================================================================ - // COGNITIVE TOOLS (v1.5+) - // ================================================================ - "dream" => { - self.emit(VestigeEvent::DreamStarted { - memory_count: self - .storage - .get_stats() - .map(|s| s.total_nodes as usize) - .unwrap_or(0), - timestamp: chrono::Utc::now(), - }); - tools::dream::execute(&self.storage, &self.cognitive, request.arguments).await - } - "explore_connections" => { - tools::explore::execute(&self.storage, &self.cognitive, request.arguments).await - } - "predict" => { - tools::predict::execute(&self.storage, &self.cognitive, request.arguments).await - } - "restore" => tools::restore::execute(&self.storage, request.arguments).await, - - // ================================================================ - // CONTEXT PACKETS (v1.8+) - // ================================================================ - "session_context" => { - tools::session_context::execute( - &self.storage, - &self.cognitive, - &self.output_config, - request.arguments, - ) - .await - } - - // ================================================================ - // AUTONOMIC TOOLS (v1.9+) - // ================================================================ - "memory_health" => tools::health::execute(&self.storage, request.arguments).await, - "memory_graph" => tools::graph::execute(&self.storage, request.arguments).await, - "composed_graph" => { - tools::composed_graph::execute(&self.storage, request.arguments).await - } - "deep_reference" | "cross_reference" => { - tools::cross_reference::execute(&self.storage, &self.cognitive, request.arguments) + // ================================================================ + // SYSTEM STATUS (v1.7: replaces health_check + stats) + // ================================================================ + "system_status" => { + tools::maintenance::execute_system_status( + &self.storage, + &self.cognitive, + request.arguments, + ) .await - } - "contradictions" => { - tools::contradictions::execute(&self.storage, request.arguments).await - } + } - // ================================================================ - // ACTIVE FORGETTING (v2.0.5) — top-down suppression - // ================================================================ - "suppress" => tools::suppress::execute(&self.storage, request.arguments).await, + "mark_reviewed" => tools::review::execute(&self.storage, request.arguments).await, - name => { - return Err(JsonRpcError::invalid_params(&format!( - "Unknown tool: {}", - name - ))); + // ================================================================ + // DEPRECATED: Search tools - redirect to unified 'search' + // ================================================================ + "recall" | "semantic_search" | "hybrid_search" => { + warn!( + "Tool '{}' is deprecated. Use 'search' instead.", + request.name + ); + tools::search_unified::execute( + &self.storage, + &self.cognitive, + &self.output_config, + request.arguments, + ) + .await + } + + // ================================================================ + // DEPRECATED: Memory tools - redirect to unified 'memory' + // ================================================================ + "get_knowledge" => { + warn!( + "Tool 'get_knowledge' is deprecated. Use 'memory' with action='get' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let id = args.get("id").cloned().unwrap_or(serde_json::Value::Null); + Some(serde_json::json!({ + "action": "get", + "id": id + })) + } + None => None, + }; + tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + "delete_knowledge" => { + warn!( + "Tool 'delete_knowledge' is deprecated. Use 'memory' with action='purge', confirm=true instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let id = args.get("id").cloned().unwrap_or(serde_json::Value::Null); + let confirm = args + .get("confirm") + .cloned() + .unwrap_or(serde_json::Value::Bool(false)); + Some(serde_json::json!({ + "action": "delete", + "id": id, + "confirm": confirm + })) + } + None => None, + }; + tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + "get_memory_state" => { + warn!( + "Tool 'get_memory_state' is deprecated. Use 'memory' with action='state' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let id = args + .get("memory_id") + .cloned() + .unwrap_or(serde_json::Value::Null); + Some(serde_json::json!({ + "action": "state", + "id": id + })) + } + None => None, + }; + tools::memory_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + + // ================================================================ + // DEPRECATED: Codebase tools - redirect to unified 'codebase' + // ================================================================ + "remember_pattern" => { + warn!( + "Tool 'remember_pattern' is deprecated. Use 'codebase' with action='remember_pattern' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert( + "action".to_string(), + serde_json::json!("remember_pattern"), + ); + } + Some(new_args) + } + None => Some(serde_json::json!({"action": "remember_pattern"})), + }; + tools::codebase_unified::execute( + &self.storage, + &self.cognitive, + &self.output_config, + unified_args, + ) + .await + } + "remember_decision" => { + warn!( + "Tool 'remember_decision' is deprecated. Use 'codebase' with action='remember_decision' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert( + "action".to_string(), + serde_json::json!("remember_decision"), + ); + } + Some(new_args) + } + None => Some(serde_json::json!({"action": "remember_decision"})), + }; + tools::codebase_unified::execute( + &self.storage, + &self.cognitive, + &self.output_config, + unified_args, + ) + .await + } + "get_codebase_context" => { + warn!( + "Tool 'get_codebase_context' is deprecated. Use 'codebase' with action='get_context' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert("action".to_string(), serde_json::json!("get_context")); + } + Some(new_args) + } + None => Some(serde_json::json!({"action": "get_context"})), + }; + tools::codebase_unified::execute( + &self.storage, + &self.cognitive, + &self.output_config, + unified_args, + ) + .await + } + + // ================================================================ + // DEPRECATED: Intention tools - redirect to unified 'intention' + // ================================================================ + "set_intention" => { + warn!( + "Tool 'set_intention' is deprecated. Use 'intention' with action='set' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert("action".to_string(), serde_json::json!("set")); + } + Some(new_args) + } + None => Some(serde_json::json!({"action": "set"})), + }; + tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + "check_intentions" => { + warn!( + "Tool 'check_intentions' is deprecated. Use 'intention' with action='check' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert("action".to_string(), serde_json::json!("check")); + } + Some(new_args) + } + None => Some(serde_json::json!({"action": "check"})), + }; + tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + "complete_intention" => { + warn!( + "Tool 'complete_intention' is deprecated. Use 'intention' with action='update', status='complete' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let id = args + .get("intentionId") + .cloned() + .unwrap_or(serde_json::Value::Null); + Some(serde_json::json!({ + "action": "update", + "id": id, + "status": "complete" + })) + } + None => None, + }; + tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + "snooze_intention" => { + warn!( + "Tool 'snooze_intention' is deprecated. Use 'intention' with action='update', status='snooze' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let id = args + .get("intentionId") + .cloned() + .unwrap_or(serde_json::Value::Null); + let minutes = args + .get("minutes") + .cloned() + .unwrap_or(serde_json::json!(30)); + Some(serde_json::json!({ + "action": "update", + "id": id, + "status": "snooze", + "snooze_minutes": minutes + })) + } + None => None, + }; + tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + "list_intentions" => { + warn!( + "Tool 'list_intentions' is deprecated. Use 'intention' with action='list' instead." + ); + let unified_args = match request.arguments { + Some(ref args) => { + let mut new_args = args.clone(); + if let Some(obj) = new_args.as_object_mut() { + obj.insert("action".to_string(), serde_json::json!("list")); + if let Some(status) = obj.remove("status") { + obj.insert("filter_status".to_string(), status); + } + } + Some(new_args) + } + None => Some(serde_json::json!({"action": "list"})), + }; + tools::intention_unified::execute(&self.storage, &self.cognitive, unified_args) + .await + } + + // ================================================================ + // Neuroscience tools (internal, not in tools/list) + // ================================================================ + "list_by_state" => { + tools::memory_states::execute_list(&self.storage, request.arguments).await + } + "state_stats" => tools::memory_states::execute_stats(&self.storage).await, + "trigger_importance" => { + tools::tagging::execute_trigger(&self.storage, request.arguments).await + } + "find_tagged" => { + tools::tagging::execute_find(&self.storage, request.arguments).await + } + "tagging_stats" => tools::tagging::execute_stats(&self.storage).await, + "match_context" => tools::context::execute(&self.storage, request.arguments).await, + + // ================================================================ + // Feedback (internal, still used by request_feedback) + // ================================================================ + "request_feedback" => { + tools::feedback::execute_request_feedback(&self.storage, request.arguments) + .await + } + + // ================================================================ + // TEMPORAL TOOLS (v1.2+) + // ================================================================ + "memory_timeline" => { + tools::timeline::execute(&self.storage, &self.output_config, request.arguments) + .await + } + "memory_changelog" => { + tools::changelog::execute(&self.storage, request.arguments).await + } + + // ================================================================ + // MAINTENANCE TOOLS (v1.2+, non-deprecated) + // ================================================================ + "consolidate" => { + self.emit(VestigeEvent::ConsolidationStarted { + timestamp: chrono::Utc::now(), + }); + tools::maintenance::execute_consolidate(&self.storage, request.arguments).await + } + "backup" => { + tools::maintenance::execute_backup(&self.storage, request.arguments).await + } + "export" => { + tools::maintenance::execute_export(&self.storage, request.arguments).await + } + "gc" => tools::maintenance::execute_gc(&self.storage, request.arguments).await, + + // ================================================================ + // AUTO-SAVE & DEDUP TOOLS (v1.3+) + // ================================================================ + "importance_score" => { + tools::importance::execute(&self.storage, &self.cognitive, request.arguments) + .await + } + "find_duplicates" => tools::dedup::execute(&self.storage, request.arguments).await, + + // ================================================================ + // MERGE / SUPERSEDE CONTROLS (v2.1.25 — Phase 3) + // ================================================================ + "merge_candidates" | "plan_merge" | "plan_supersede" | "apply_plan" + | "merge_undo" | "protect" | "merge_policy" => { + tools::merge::execute(&self.storage, request.name.as_str(), request.arguments) + .await + } + + // ================================================================ + // COGNITIVE TOOLS (v1.5+) + // ================================================================ + "dream" => { + self.emit(VestigeEvent::DreamStarted { + memory_count: self + .storage + .get_stats() + .map(|s| s.total_nodes as usize) + .unwrap_or(0), + timestamp: chrono::Utc::now(), + }); + tools::dream::execute(&self.storage, &self.cognitive, request.arguments).await + } + "explore_connections" => { + tools::explore::execute(&self.storage, &self.cognitive, request.arguments).await + } + "predict" => { + tools::predict::execute(&self.storage, &self.cognitive, request.arguments).await + } + "restore" => tools::restore::execute(&self.storage, request.arguments).await, + + // ================================================================ + // CONTEXT PACKETS (v1.8+) + // ================================================================ + "session_context" => { + tools::session_context::execute( + &self.storage, + &self.cognitive, + &self.output_config, + request.arguments, + ) + .await + } + + // ================================================================ + // AUTONOMIC TOOLS (v1.9+) + // ================================================================ + "memory_health" => tools::health::execute(&self.storage, request.arguments).await, + "memory_graph" => tools::graph::execute(&self.storage, request.arguments).await, + "composed_graph" => { + tools::composed_graph::execute(&self.storage, request.arguments).await + } + "deep_reference" | "cross_reference" => { + tools::cross_reference::execute( + &self.storage, + &self.cognitive, + request.arguments, + ) + .await + } + "contradictions" => { + tools::contradictions::execute(&self.storage, request.arguments).await + } + + // ================================================================ + // ACTIVE FORGETTING (v2.0.5) — top-down suppression + // ================================================================ + "suppress" => tools::suppress::execute(&self.storage, request.arguments).await, + + name => { + return Err(JsonRpcError::invalid_params(&format!( + "Unknown tool: {}", + name + ))); + } } }; @@ -1164,8 +1214,12 @@ impl McpServer { // 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); + 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( @@ -1178,10 +1232,7 @@ impl McpServer { // 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("memoryPrsOpened".to_string(), serde_json::json!(opened_prs)); obj.insert( "memoryPrNotice".to_string(), serde_json::json!( @@ -2512,6 +2563,101 @@ mod tests { ); } + /// Destructive memory operations must be blocked before execution in the + /// default Risk-Gated mode. This is the real C2 regression test: a purge + /// request opens a Memory PR, but the row is still present until review. + #[tokio::test] + async fn test_memory_purge_is_pre_gated_before_delete() { + let (mut server, _dir) = test_server().await; + server + .handle_request(make_request("initialize", Some(init_params()))) + .await; + let node = server + .storage + .ingest(vestige_core::IngestInput { + content: "A purge target containing auth token sk-live-DO-NOT-LEAK-123".to_string(), + node_type: "fact".to_string(), + ..Default::default() + }) + .unwrap(); + + let call = make_request( + "tools/call", + Some(serde_json::json!({ + "name": "memory", + "arguments": { + "action": "purge", + "id": node.id, + "confirm": true, + "runId": "run_pre_gate_purge" + } + })), + ); + let response = server.handle_request(call).await.unwrap(); + let structured = response.result.unwrap()["structuredContent"].clone(); + + assert_eq!(structured["pendingReview"], serde_json::json!(true)); + assert_eq!(structured["success"], serde_json::json!(false)); + assert!( + server.storage.get_node(&node.id).unwrap().is_some(), + "purge must not delete before Memory PR review" + ); + let prs = server + .storage + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); + assert_eq!(prs.len(), 1); + assert_eq!(prs[0].subject_id.as_deref(), Some(node.id.as_str())); + assert_eq!(prs[0].diff["pendingAction"], serde_json::json!("purge")); + let serialized = serde_json::to_string(&prs[0]).unwrap(); + assert!( + !serialized.contains("DO-NOT-LEAK") && !serialized.contains("sk-live"), + "pending Memory PR must not expose raw sensitive content" + ); + } + + #[tokio::test] + async fn test_direct_suppress_is_pre_gated_before_mutation() { + let (mut server, _dir) = test_server().await; + server + .handle_request(make_request("initialize", Some(init_params()))) + .await; + let node = server + .storage + .ingest(vestige_core::IngestInput { + content: "A suppress target awaiting review.".to_string(), + node_type: "fact".to_string(), + ..Default::default() + }) + .unwrap(); + + let call = make_request( + "tools/call", + Some(serde_json::json!({ + "name": "suppress", + "arguments": { + "id": node.id, + "reason": "test suppress", + "runId": "run_pre_gate_suppress" + } + })), + ); + let response = server.handle_request(call).await.unwrap(); + let structured = response.result.unwrap()["structuredContent"].clone(); + + assert_eq!(structured["pendingReview"], serde_json::json!(true)); + let current = server.storage.get_node(&node.id).unwrap().unwrap(); + assert_eq!( + current.suppression_count, 0, + "suppress must not mutate retrieval influence before review" + ); + let prs = server + .storage + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); + assert_eq!(prs[0].diff["pendingAction"], serde_json::json!("suppress")); + } + /// 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 → @@ -2540,7 +2686,11 @@ mod tests { ); 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["runId"].as_str(), + Some(RUN), + "HOP 1: tool output runId" + ); assert_eq!( structured["traceUri"].as_str(), Some(&format!("vestige://trace/{RUN}")[..]), @@ -2579,7 +2729,10 @@ mod tests { .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"); + 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!( @@ -2605,7 +2758,10 @@ mod tests { "HOP 5: vestige://trace/{{runId}} resolves the same runId" ); assert!( - parsed["events"].as_array().map(|a| !a.is_empty()).unwrap_or(false), + parsed["events"] + .as_array() + .map(|a| !a.is_empty()) + .unwrap_or(false), "HOP 5: the resource returns the run's events" ); diff --git a/crates/vestige-mcp/src/trace_recorder.rs b/crates/vestige-mcp/src/trace_recorder.rs index 2aa28d5..cc24ff8 100644 --- a/crates/vestige-mcp/src/trace_recorder.rs +++ b/crates/vestige-mcp/src/trace_recorder.rs @@ -97,7 +97,7 @@ pub fn gate_writes( mode: vestige_core::ReviewMode, ) -> Vec { use vestige_core::{ - classify_write, MemoryPr, MemoryPrKind, MemoryPrStatus, RiskClass, WriteContext, + MemoryPr, MemoryPrKind, MemoryPrStatus, RiskClass, WriteContext, classify_write, }; if !is_write_tool(tool) { @@ -139,7 +139,10 @@ pub fn gate_writes( let ctx = WriteContext { source: Some(WriteSource::Agent), - node_type: node.as_ref().map(|n| n.node_type.clone()).unwrap_or_default(), + node_type: node + .as_ref() + .map(|n| n.node_type.clone()) + .unwrap_or_default(), content: node.as_ref().map(|n| n.content.clone()).unwrap_or_default(), tags: node.as_ref().map(|n| n.tags.clone()).unwrap_or_default(), contradicts_trust, @@ -176,9 +179,9 @@ pub fn gate_writes( // secret, and the PR row is read by the dashboard and may be exported). // Store a short, redacted preview + a content hash instead. The preview // is dropped entirely when the write was gated for a sensitive topic. - let sensitive = signals.iter().any(|s| { - s.code == "sensitive_topic" || s.code == "sensitive_node_type" - }); + let sensitive = signals + .iter() + .any(|s| s.code == "sensitive_topic" || s.code == "sensitive_node_type"); let raw_content = node.as_ref().map(|n| n.content.as_str()).unwrap_or(""); let preview = content_preview(raw_content, sensitive); let content_hash = hash_content(raw_content); @@ -238,6 +241,232 @@ pub fn gate_writes( opened } +struct PendingMemoryMutation { + action: String, + id: String, + reason: Option, +} + +/// Pre-gate memory mutations that would otherwise be irreversible or directly +/// inhibitory before the reviewer sees them. +/// +/// Normal risky writes are still handled post-commit by [`gate_writes`]. Purge, +/// delete, and suppress are different: executing the tool first either removes +/// the row or mutates retrieval influence before review. Under Risk-Gated and +/// Paranoid modes this function opens a pending Memory PR and returns a tool +/// response without performing the mutation. Fast mode keeps the historical +/// direct-execution behavior. +pub fn gate_pending_memory_mutation( + storage: &Arc, + event_tx: Option<&broadcast::Sender>, + run_id: &str, + tool: &str, + args: &Option, + mode: vestige_core::ReviewMode, +) -> Result, String> { + use vestige_core::{ + MemoryPr, MemoryPrKind, MemoryPrStatus, RiskClass, WriteContext, classify_write, + }; + + if matches!(mode, vestige_core::ReviewMode::Fast) { + return Ok(None); + } + + let Some(pending) = pending_memory_mutation(tool, args) else { + return Ok(None); + }; + let node = match storage.get_node(&pending.id) { + Ok(Some(node)) => node, + Ok(None) => return Ok(None), + Err(e) => return Err(format!("failed to inspect memory before review gate: {e}")), + }; + + let ctx = WriteContext { + source: Some(WriteSource::Agent), + node_type: node.node_type.clone(), + content: node.content.clone(), + tags: node.tags.clone(), + forgets: true, + ..Default::default() + }; + let (class, signals) = classify_write(&ctx, mode); + if class != RiskClass::Review { + return Ok(None); + } + + if let Some(existing) = find_pending_mutation_pr(storage, &pending.id, &pending.action) { + if let Some(tx) = event_tx { + let _ = tx.send(VestigeEvent::MemoryPrOpened { + id: existing.id.clone(), + kind: existing.kind.as_str().to_string(), + title: existing.title.clone(), + signal_count: existing.signals.len(), + run_id: existing.run_id.clone().or_else(|| Some(run_id.to_string())), + timestamp: Utc::now(), + }); + } + return Ok(Some(pending_review_response(&pending, &existing))); + } + + let sensitive = signals + .iter() + .any(|s| s.code == "sensitive_topic" || s.code == "sensitive_node_type"); + let preview = content_preview(&node.content, sensitive); + let content_hash = hash_content(&node.content); + let kind = MemoryPrKind::NodeDecayed; + let title = format!("{}: {}", pr_kind_phrase(kind), preview); + let pr = MemoryPr { + id: format!("pr_{}", uuid::Uuid::new_v4().simple()), + kind, + status: MemoryPrStatus::Pending, + title: title.clone(), + diff: serde_json::json!({ + "decision": pending.action, + "pendingAction": pending.action, + "requiresApproval": true, + "reason": pending.reason, + "node": { + "id": pending.id, + "nodeType": node.node_type, + "contentPreview": preview, + "contentHash": content_hash, + "tags": node.tags, + "deleted": false, + }, + }), + signals: signals.clone(), + subject_id: Some(pending.id.clone()), + run_id: Some(run_id.to_string()), + created_at: Utc::now().to_rfc3339(), + decided_at: None, + decision: None, + }; + + if let Err(e) = storage.save_memory_pr(&pr) { + tracing::warn!("pending destructive Memory PR save failed: {e}"); + return Err(format!( + "review gate failed closed: could not open Memory PR for pending mutation: {e}" + )); + } + + if let Some(tx) = event_tx { + let _ = tx.send(VestigeEvent::MemoryPrOpened { + id: pr.id.clone(), + kind: kind.as_str().to_string(), + title: title.clone(), + signal_count: signals.len(), + run_id: Some(run_id.to_string()), + timestamp: Utc::now(), + }); + } + + Ok(Some(pending_review_response(&pending, &pr))) +} + +fn find_pending_mutation_pr( + storage: &Arc, + subject_id: &str, + pending_action: &str, +) -> Option { + storage + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 500) + .ok()? + .into_iter() + .find(|pr| { + pr.subject_id.as_deref() == Some(subject_id) + && pr.diff.get("pendingAction").and_then(|v| v.as_str()) == Some(pending_action) + }) +} + +fn pending_review_response( + pending: &PendingMemoryMutation, + pr: &vestige_core::MemoryPr, +) -> serde_json::Value { + let opened = serde_json::json!({ + "id": pr.id.clone(), + "kind": pr.kind.as_str(), + "title": pr.title.clone(), + "signals": pr.signals.clone(), + "subjectId": pending.id.clone(), + }); + + serde_json::json!({ + "action": format!("{}_pending_review", pending.action), + "success": false, + "pendingReview": true, + "nodeId": pending.id, + "message": "Mutation was not executed. Vestige opened a Memory PR and is waiting for review.", + "memoryPrsOpened": [opened], + "memoryPrNotice": "Vestige opened a Memory PR before applying this destructive or suppressive memory mutation. Approve with `forget`; keep the memory with `promote`; hold it suppressed with `quarantine`.", + }) +} + +fn pending_memory_mutation( + tool: &str, + args: &Option, +) -> Option { + let args = args.as_ref()?; + match tool { + "memory" => { + let action = args.get("action")?.as_str()?.to_ascii_lowercase(); + if !matches!(action.as_str(), "purge" | "delete") { + return None; + } + if !args + .get("confirm") + .and_then(|v| v.as_bool()) + .unwrap_or(false) + { + return None; + } + Some(PendingMemoryMutation { + action, + id: args.get("id")?.as_str()?.to_string(), + reason: args + .get("reason") + .and_then(|v| v.as_str()) + .map(str::to_string), + }) + } + "delete_knowledge" => { + if !args + .get("confirm") + .and_then(|v| v.as_bool()) + .unwrap_or(false) + { + return None; + } + Some(PendingMemoryMutation { + action: "delete".to_string(), + id: args.get("id")?.as_str()?.to_string(), + reason: args + .get("reason") + .and_then(|v| v.as_str()) + .map(str::to_string), + }) + } + "suppress" => { + if args + .get("reverse") + .and_then(|v| v.as_bool()) + .unwrap_or(false) + { + return None; + } + Some(PendingMemoryMutation { + action: "suppress".to_string(), + id: args.get("id")?.as_str()?.to_string(), + reason: args + .get("reason") + .or_else(|| args.get("note")) + .and_then(|v| v.as_str()) + .map(str::to_string), + }) + } + _ => None, + } +} + /// Whether a write decision permanently removes / forgets memory (so the live /// row may already be gone when the gate runs). fn is_destructive_decision(label: &str) -> bool { @@ -719,7 +948,10 @@ fn extract_veto(result: &Value) -> Option<(String, Vec, f64)> { .collect() }) .unwrap_or_default(); - let confidence = veto.get("confidence").and_then(|v| v.as_f64()).unwrap_or(0.0); + let confidence = veto + .get("confidence") + .and_then(|v| v.as_f64()) + .unwrap_or(0.0); Some((claim, evidence_ids, confidence)) } @@ -875,7 +1107,10 @@ mod tests { #[test] fn extract_writes_single_and_batch() { let single = serde_json::json!({ "decision": "create", "nodeId": "n1" }); - assert_eq!(extract_writes(&single), vec![("n1".into(), "create".into())]); + assert_eq!( + extract_writes(&single), + vec![("n1".into(), "create".into())] + ); let batch = serde_json::json!({ "results": [ { "decision": "update", "id": "n2" } ] }); @@ -886,9 +1121,15 @@ mod tests { fn extract_writes_recognizes_action_shape_b2() { // B2: memory promote/demote return `action` + `nodeId`, not `decision`. let promoted = serde_json::json!({ "action": "promoted", "nodeId": "m1" }); - assert_eq!(extract_writes(&promoted), vec![("m1".into(), "promoted".into())]); + assert_eq!( + extract_writes(&promoted), + vec![("m1".into(), "promoted".into())] + ); let demoted = serde_json::json!({ "action": "demoted", "nodeId": "m2" }); - assert_eq!(extract_writes(&demoted), vec![("m2".into(), "demoted".into())]); + assert_eq!( + extract_writes(&demoted), + vec![("m2".into(), "demoted".into())] + ); // codebase remember_decision returns action + nodeId. let decision = serde_json::json!({ "action": "remember_decision", "nodeId": "c1" }); assert_eq!( @@ -908,7 +1149,14 @@ mod tests { #[test] fn destructive_decision_classification_c2() { - for d in ["purge", "delete", "forget", "purged", "deleted", "forgotten"] { + for d in [ + "purge", + "delete", + "forget", + "purged", + "deleted", + "forgotten", + ] { assert!(is_destructive_decision(d), "{d} is destructive"); } for d in ["create", "update", "promote", "reinforce"] { @@ -987,8 +1235,14 @@ mod tests { vestige_core::ReviewMode::RiskGated, ); - assert_eq!(opened.len(), 1, "destructive write must open a PR even with the node gone"); - let pr = s.list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10).unwrap(); + assert_eq!( + opened.len(), + 1, + "destructive write must open a PR even with the node gone" + ); + let pr = s + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); assert_eq!(pr.len(), 1); assert_eq!(pr[0].subject_id.as_deref(), Some(node.id.as_str())); // The diff marks the node as deleted and carries no resurrected content. @@ -1035,6 +1289,157 @@ mod tests { assert!(pr.diff["node"]["contentHash"].as_str().is_some()); } + #[test] + fn pre_gate_blocks_purge_before_deleting_c2() { + let s = store(); + let node = s + .ingest(vestige_core::IngestInput { + content: "A memory awaiting destructive review.".to_string(), + node_type: "fact".to_string(), + ..Default::default() + }) + .unwrap(); + let args = Some(serde_json::json!({ + "action": "purge", + "id": node.id, + "confirm": true, + "reason": "test purge" + })); + + let response = gate_pending_memory_mutation( + &s, + None, + "run_pre_gate", + "memory", + &args, + vestige_core::ReviewMode::RiskGated, + ) + .unwrap() + .expect("purge should be pre-gated"); + + assert_eq!(response["pendingReview"], serde_json::json!(true)); + assert!( + s.get_node(&node.id).unwrap().is_some(), + "pre-gating must not delete before review" + ); + let pr = s + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); + assert_eq!(pr.len(), 1); + assert_eq!(pr[0].diff["pendingAction"], serde_json::json!("purge")); + assert_eq!(pr[0].diff["node"]["deleted"], serde_json::json!(false)); + } + + #[test] + fn pre_gate_leaves_fast_mode_direct() { + let s = store(); + let node = s + .ingest(vestige_core::IngestInput { + content: "Fast mode purge target.".to_string(), + node_type: "fact".to_string(), + ..Default::default() + }) + .unwrap(); + let args = Some(serde_json::json!({ + "action": "purge", + "id": node.id, + "confirm": true + })); + + assert!( + gate_pending_memory_mutation( + &s, + None, + "run_fast", + "memory", + &args, + vestige_core::ReviewMode::Fast, + ) + .unwrap() + .is_none(), + "Fast mode should preserve direct execution" + ); + } + + #[test] + fn pre_gate_blocks_direct_suppress_before_mutating() { + let s = store(); + let node = s + .ingest(vestige_core::IngestInput { + content: "A memory awaiting suppress review.".to_string(), + node_type: "fact".to_string(), + ..Default::default() + }) + .unwrap(); + let args = Some(serde_json::json!({ "id": node.id, "reason": "test suppress" })); + + let response = gate_pending_memory_mutation( + &s, + None, + "run_suppress", + "suppress", + &args, + vestige_core::ReviewMode::RiskGated, + ) + .unwrap() + .expect("suppress should be pre-gated"); + + assert_eq!(response["pendingReview"], serde_json::json!(true)); + let kept = s.get_node(&node.id).unwrap().unwrap(); + assert_eq!(kept.suppression_count, 0, "pre-gate must not suppress yet"); + let pr = s + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); + assert_eq!(pr[0].diff["pendingAction"], serde_json::json!("suppress")); + } + + #[test] + fn pre_gate_reuses_existing_pending_mutation_pr() { + let s = store(); + let node = s + .ingest(vestige_core::IngestInput { + content: "Do not open duplicate destructive PRs.".to_string(), + node_type: "fact".to_string(), + ..Default::default() + }) + .unwrap(); + let args = Some(serde_json::json!({ + "action": "delete", + "id": node.id, + "confirm": true + })); + + let first = gate_pending_memory_mutation( + &s, + None, + "run_repeat_1", + "memory", + &args, + vestige_core::ReviewMode::RiskGated, + ) + .unwrap() + .expect("first delete should open PR"); + let second = gate_pending_memory_mutation( + &s, + None, + "run_repeat_2", + "memory", + &args, + vestige_core::ReviewMode::RiskGated, + ) + .unwrap() + .expect("second delete should reuse PR"); + + assert_eq!( + first["memoryPrsOpened"][0]["id"], second["memoryPrsOpened"][0]["id"], + "repeated pending mutation returns the existing PR" + ); + let prs = s + .list_memory_prs(Some(vestige_core::MemoryPrStatus::Pending), 10) + .unwrap(); + assert_eq!(prs.len(), 1, "only one pending PR is stored"); + } + #[test] fn write_tool_set_includes_codebase_b2() { assert!(is_write_tool("codebase"));