Update vault smoke tests: T2 default for inserts, T1 denied test, audit log severity verification

smoke_test_vault.py

@@ -1,14 +1,16 @@
 """
-Smoke test for vault_adapter.
+Smoke test for vault_adapter + permission_gate integration.
 
 Coverage:
-  1. Open Vault on a tempdir, schema_migrations bootstrap works.
+  1. Open Vault on a tempdir; schema_migrations bootstraps.
   2. Apply migration 003 — idempotent (re-apply is a no-op).
-  3. Insert KV profile rows, count matches, gate audit log populated.
-  4. Re-insert the same rows — duplicates silently dropped (idempotent).
-  5. Fetch by invalidation key returns the inserted rows.
-  6. has_kv_profile() works on hit and miss.
-  7. PermissionGate rejection actually blocks writes.
+  3. Migration requires VAULT_SCHEMA: T2 agent denied, T6 succeeds.
+  4. Insert KV profile rows as a T2 agent (default I/O worker tier).
+  5. Re-insert the same rows — duplicates silently dropped (idempotent).
+  6. Fetch by invalidation key returns the inserted rows.
+  7. has_kv_profile() works on hit and miss.
+  8. T1 (read-only) agent denied at insert with capability-correct error.
+  9. Audit log records every gate check with full severity metadata.
 """
 
 import sys
@@ -19,33 +21,31 @@ sys.path.insert(0, str(Path(__file__).resolve().parent))
 
 import vault_adapter as va
 import kv_profiler as kvp
+import permission_gate as pg
 
 
 def hr(title: str) -> None:
     print(f"\n{'=' * 6} {title} {'=' * 6}")
 
 
-# Build a minimal ProfileRow-shaped object for testing without needing a real
-# model. We use kvp.ProfileRow directly since it's the canonical shape.
-def make_row(layer_idx: int, k: int, v: int, quantizer: str, drift: float,
-             model_hash: str = "testhash") -> kvp.ProfileRow:
+def make_row(layer_idx, k, v, quantizer, drift,
+             model_hash="testhash",
+             agent_id="smoke-test",
+             agent_tier=2):
     return kvp.ProfileRow(
         model_hash=model_hash,
         calibration_hash="calhash0",
         pipeline_version="1.0.0",
         layer_idx=layer_idx,
-        k_bits=k,
-        v_bits=v,
-        quantizer=quantizer,
+        k_bits=k, v_bits=v, quantizer=quantizer,
         drift_attn_output=drift,
         drift_metric="mse_normalised",
         bytes_per_kv_token=128.0,
         max_seq_len_observed=1024,
-        num_kv_heads=8,
-        head_dim=128,
+        num_kv_heads=8, head_dim=128,
         profiled_at="2026-05-18T12:00:00+00:00",
-        profiled_by_agent_id="smoke-test",
-        profiled_by_agent_tier=1,
+        profiled_by_agent_id=agent_id,
+        profiled_by_agent_tier=agent_tier,
     )
 
 
@@ -65,112 +65,138 @@ vault = va.VaultAdapter(db_path)
 cur = vault.conn.execute(
     "SELECT name FROM sqlite_master WHERE type='table' AND name='schema_migrations'"
 )
-assert cur.fetchone() is not None, "schema_migrations not created"
+assert cur.fetchone() is not None
 print(f"  schema_migrations exists ✓")
 
 
 # ===========================================================================
-# 2. Apply migration 003 — idempotent
+# 2. Apply migration 003 — idempotent (default agent_tier=6)
 # ===========================================================================
-hr("2. apply migration 003 (idempotent)")
+hr("2. apply migration 003 (idempotent) as T6")
 applied = vault.apply_migration(migration_path)
-print(f"  first apply: {'YES' if applied else 'NO'}")
 assert applied, "first apply should have run"
+print(f"  first apply: YES ✓")
 
 applied2 = vault.apply_migration(migration_path)
-print(f"  second apply: {'YES' if applied2 else 'NO'}")
 assert not applied2, "second apply should have been skipped"
-
-print(f"  applied migrations: {vault.applied_migrations()}")
+print(f"  second apply: NO (skipped) ✓")
 assert "003" in vault.applied_migrations()
 
 
 # ===========================================================================
-# 3. Insert KV profile rows
+# 3. Migration requires VAULT_SCHEMA: T2 denied, T6 succeeds
+# ===========================================================================
+hr("3. apply_migration as T2 must be denied")
+# Reset for a fresh migration attempt (drop the version row)
+vault2_path = tmp / "vault2.db"
+vault2 = va.VaultAdapter(vault2_path)
+try:
+    vault2.apply_migration(migration_path, agent_id="bad-agent", agent_tier=2)
+    print(f"  FAIL — T2 should not have been able to apply migration")
+    sys.exit(1)
+except pg.PermissionDenied as e:
+    print(f"  caught PermissionDenied: {e} ✓")
+
+# Now apply as T6 — works
+applied_t6 = vault2.apply_migration(
+    migration_path, agent_id="operator", agent_tier=6
+)
+assert applied_t6
+print(f"  T6 apply succeeded ✓")
+vault2.close()
+
+
+# ===========================================================================
+# 4. Insert KV profile rows as T2 (default working tier)
 # ===========================================================================
-hr("3. insert KV profile rows")
+hr("4. insert KV profile rows as T2")
 rows = [
-    make_row(layer_idx=i, k=4, v=4, quantizer="hqq_g64", drift=0.01 + i * 0.001)
-    for i in range(5)
+    make_row(i, 4, 4, "hqq_g64", 0.01 + i * 0.001) for i in range(5)
 ] + [
-    make_row(layer_idx=i, k=8, v=8, quantizer="hqq_g64", drift=0.001 + i * 0.0001)
-    for i in range(5)
+    make_row(i, 8, 8, "hqq_g64", 0.001 + i * 0.0001) for i in range(5)
 ]
-print(f"  inserting {len(rows)} rows")
 result = vault.insert_kv_profile_rows(rows)
-print(f"  requested={result.requested} inserted={result.inserted} table={result.table}")
+print(f"  requested={result.requested} inserted={result.inserted}")
 assert result.requested == 10
 assert result.inserted == 10
-assert len(vault.gate.audit_log) >= 1, "permission gate audit log not populated"
-print(f"  gate audit log: {vault.gate.audit_log}")
 
 
 # ===========================================================================
-# 4. Re-insert same rows — idempotent dedup
+# 5. Re-insert same rows — idempotent dedup
 # ===========================================================================
-hr("4. re-insert same rows (idempotent)")
+hr("5. re-insert same rows (idempotent)")
 result2 = vault.insert_kv_profile_rows(rows)
 print(f"  requested={result2.requested} inserted={result2.inserted}")
 assert result2.requested == 10
-assert result2.inserted == 0, "duplicate insert should have inserted 0 new rows"
+assert result2.inserted == 0
 
 
 # ===========================================================================
-# 5. Fetch by invalidation key
+# 6. Fetch by invalidation key
 # ===========================================================================
-hr("5. fetch_kv_profile_rows")
+hr("6. fetch_kv_profile_rows")
 fetched = vault.fetch_kv_profile_rows("testhash", "calhash0", "1.0.0")
 print(f"  fetched {len(fetched)} rows")
 assert len(fetched) == 10
-assert all("drift_attn_output" in r for r in fetched)
-assert fetched[0]["model_hash"] == "testhash"
-
-# Wrong invalidation key returns empty
 missing = vault.fetch_kv_profile_rows("testhash", "calhash0", "9.9.9")
-print(f"  wrong pipeline_version: {len(missing)} rows")
 assert len(missing) == 0
 
 
 # ===========================================================================
-# 6. has_kv_profile
+# 7. has_kv_profile
 # ===========================================================================
-hr("6. has_kv_profile hit/miss")
+hr("7. has_kv_profile hit/miss")
 assert vault.has_kv_profile("testhash", "calhash0", "1.0.0") is True
-print(f"  hit: True ✓")
 assert vault.has_kv_profile("doesnotexist", "calhash0", "1.0.0") is False
-print(f"  miss: False ✓")
+print(f"  hit/miss correctly distinguished ✓")
 
 
 # ===========================================================================
-# 7. PermissionGate rejection
+# 8. T1 (read-only) agent denied insert with correct capability error
 # ===========================================================================
-hr("7. PermissionGate rejection actually blocks writes")
-
-
-class BlockingGate(va.PermissionGate):
-    def check_write(self, table: str, agent_id: str, agent_tier: int) -> None:
-        if agent_tier < 5:
-            raise va.PermissionDenied(
-                f"agent {agent_id} tier {agent_tier} below min tier 5 for {table}"
-            )
-        super().check_write(table, agent_id, agent_tier)
-
-
-vault2 = va.VaultAdapter(db_path, permission_gate=BlockingGate())
-new_row = make_row(layer_idx=99, k=2, v=2, quantizer="hqq_g64", drift=0.5,
-                   model_hash="willnotinsert")
+hr("8. T1 agent denied insert (vault.append requires T2+)")
+t1_row = make_row(
+    99, 2, 2, "hqq_g64", 0.5,
+    model_hash="willnotinsert",
+    agent_id="readonly-agent",
+    agent_tier=1,
+)
 try:
-    vault2.insert_kv_profile_rows([new_row])
-    print(f"  FAIL: should have raised PermissionDenied")
+    vault.insert_kv_profile_rows([t1_row])
+    print(f"  FAIL — T1 should not be able to insert")
     sys.exit(1)
-except va.PermissionDenied as e:
+except pg.PermissionDenied as e:
     print(f"  caught PermissionDenied: {e} ✓")
+    assert "vault.append" in str(e)
+    print(f"  capability cited correctly ✓")
 
-# Verify nothing actually got in
-missing = vault2.fetch_kv_profile_rows("willnotinsert", "calhash0", "1.0.0")
+# Confirm nothing got in
+missing = vault.fetch_kv_profile_rows("willnotinsert", "calhash0", "1.0.0")
 assert len(missing) == 0
-print(f"  row not in DB after rejection: ✓")
-vault2.close()
+
+
+# ===========================================================================
+# 9. Audit log: every gate check recorded with severity metadata
+# ===========================================================================
+hr("9. audit log captures every check + severity")
+log = vault.gate.audit_log
+print(f"  total audit entries: {len(log)}")
+# The migration apply checked VAULT_SCHEMA at HIGH severity
+schema_checks = [r for r in log if r.capability == "vault.schema"]
+assert len(schema_checks) > 0, "VAULT_SCHEMA never logged"
+assert all(r.severity == "HIGH" for r in schema_checks)
+print(f"  VAULT_SCHEMA entries: {len(schema_checks)} (all severity=HIGH) ✓")
+
+# The inserts checked VAULT_APPEND at NORMAL severity
+append_checks = [r for r in log if r.capability == "vault.append"]
+assert len(append_checks) > 0
+assert all(r.severity == "NORMAL" for r in append_checks)
+print(f"  VAULT_APPEND entries: {len(append_checks)} (all severity=NORMAL) ✓")
+
+# The T1 attempt was a denied VAULT_APPEND
+denials = [r for r in log if not r.granted]
+assert any(r.agent_id == "readonly-agent" for r in denials)
+print(f"  denial entries: {len(denials)} (incl. T1 agent) ✓")
 
 
 vault.close()