apunkt/vestige
1
0
Fork 0
mirror of https://github.com/samvallad33/vestige.git synced 2026-07-24 23:41:01 +02:00
Code Issues Projects Releases Packages Wiki Activity

merge: launch polish — dashboard auth + review UX (lane 3/5)

Browse source
This commit is contained in:
Sam Valladares 2026-06-24 17:35:00 -05:00
commit e95b772cfe
14 changed files with 2302 additions and 662 deletions
Download patch file Download diff file Expand all files Collapse all files

67
crates/vestige-core/src/storage/trace_store.rs
View file

@ -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);
}

586
crates/vestige-mcp/src/dashboard/handlers.rs
View file

@ -161,6 +161,19 @@ pub async fn delete_memory(
State(state): State<AppState>,
Path(id): Path<String>,
) -> Result<Json<Value>, 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<Option<Value>, 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<TraceListParams>,
) -> Result<Json<Value>, StatusCode> {
let limit = params.limit.unwrap_or(50).clamp(1, 500);
let runs = state
.storage
.list_agent_runs(limit)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let runs = 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<Value> = 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<Value> = 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<AppState>,
Path((id, action)): Path<(String, String)>,
) -> Result<Json<Value>, 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();

134
crates/vestige-mcp/src/dashboard/mod.rs
View file

@ -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<Storage>,
@ -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::<axum::http::HeaderValue>()
.expect("valid origin"),
format!("http://localhost:{}", port)
.parse::<axum::http::HeaderValue>()
.expect("valid origin"),
];
// SvelteKit dev server — only in debug builds
#[cfg(debug_assertions)]
{
origins.push(
"http://localhost:5173"
.parse::<axum::http::HeaderValue>()
.expect("valid origin"),
);
origins.push(
"http://127.0.0.1:5173"
.parse::<axum::http::HeaderValue>()
.expect("valid origin"),
);
}
let origin_strings = dashboard_allowed_origins(port);
let origins: Vec<HeaderValue> = origin_strings
.iter()
.map(|origin| origin.parse::<HeaderValue>().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<String> {
#[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<AppState>,
request: Request<Body>,
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<Storage>,
@ -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);

62
crates/vestige-mcp/src/dashboard/state.rs
View file

@ -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<Arc<Mutex<CognitiveEngine>>>,
pub event_tx: broadcast::Sender<VestigeEvent>,
pub start_time: Instant,
dashboard_token: Arc<str>,
dashboard_allowed_origins: Arc<Vec<String>>,
}
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<String>) -> 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
}

29
crates/vestige-mcp/src/dashboard/websocket.rs
View file

@ -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<String>,
}
/// 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<WsAuthQuery>,
ws: WebSocketUpgrade,
State(state): State<AppState>,
) -> 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();
}

1162
crates/vestige-mcp/src/server.rs
View file

File diff suppressed because it is too large Load diff

429
crates/vestige-mcp/src/trace_recorder.rs
View file

@ -97,7 +97,7 @@ pub fn gate_writes(
mode: vestige_core::ReviewMode,
) -> Vec<serde_json::Value> {
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<String>,
}
/// 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<Storage>,
event_tx: Option<&broadcast::Sender<VestigeEvent>>,
run_id: &str,
tool: &str,
args: &Option<serde_json::Value>,
mode: vestige_core::ReviewMode,
) -> Result<Option<serde_json::Value>, 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<Storage>,
subject_id: &str,
pending_action: &str,
) -> Option<vestige_core::MemoryPr> {
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<serde_json::Value>,
) -> Option<PendingMemoryMutation> {
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<String>, 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"));