Wire vault_adapter to capability-based PermissionGate: VAULT_APPEND for inserts, VAULT_SCHEMA for migrations (T6 only)

vault_adapter.py

@@ -15,12 +15,16 @@ Why this exists in one place:
   - Migration application is centralised so bootstrap order is
     deterministic.
 
-PermissionGate interface:
-  This module defines a minimal default (`PermissionGate`) that
-  accepts everything but records the request. Replace with the real
-  policy implementation when it exists. The adapter calls
-  `gate.check_write(table, agent_id, agent_tier)` before every INSERT;
-  the gate raises `PermissionDenied` to abort, or returns to allow.
+PermissionGate routing:
+  Every privileged operation calls `gate.check(capability, agent_id,
+  agent_tier)` against the seven-tier model in permission_gate.py.
+  Default is a strict tier-enforcing gate; pass `PermissionGate.permissive()`
+  for unit tests that aren't testing authorization.
+
+  Capability mapping (per the spec):
+    - inserts on profile/record tables → VAULT_APPEND (T2+)
+    - reads of profile/record tables   → VAULT_READ (T1+, not gated here)
+    - apply_migration                   → VAULT_SCHEMA (T6 only)
 
 Idempotency:
   All inserts are `INSERT OR IGNORE` on the migrations' primary keys.
@@ -37,35 +41,9 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Iterator, Optional, Sequence
 
-logger = logging.getLogger("HSAQ.Vault")
-
-
-# ---------------------------------------------------------------------------
-# Permission gate (default + interface)
-# ---------------------------------------------------------------------------
-
-
-class PermissionDenied(Exception):
-    """Raised by a PermissionGate when a write is rejected."""
-
+from permission_gate import Capability, PermissionDenied, PermissionGate
 
-class PermissionGate:
-    """Default gate: permissive, but records every check.
-
-    Override `check_write` for real policy:
-        class RealGate(PermissionGate):
-            def check_write(self, table, agent_id, agent_tier):
-                if agent_tier < self.min_tier_for(table):
-                    raise PermissionDenied(...)
-                super().check_write(table, agent_id, agent_tier)
-    """
-    def __init__(self) -> None:
-        self.audit_log: list[tuple[str, str, int]] = []
-
-    def check_write(self, table: str, agent_id: str, agent_tier: int) -> None:
-        self.audit_log.append((table, agent_id, agent_tier))
-        logger.debug("permission check: write %s by agent %s (tier %d)",
-                     table, agent_id, agent_tier)
+logger = logging.getLogger("HSAQ.Vault")
 
 
 # ---------------------------------------------------------------------------
@@ -134,10 +112,24 @@ class VaultAdapter:
         cur = self.conn.execute("SELECT version FROM schema_migrations")
         return {row[0] for row in cur.fetchall()}
 
-    def apply_migration(self, sql_path: str | Path) -> bool:
-        """Apply a migration file. Idempotent — checks schema_migrations
-        first and skips if already recorded. Returns True if applied,
-        False if skipped."""
+    def apply_migration(
+        self,
+        sql_path: str | Path,
+        *,
+        agent_id: str = "human-operator",
+        agent_tier: int = 6,
+    ) -> bool:
+        """Apply a migration file. Requires VAULT_SCHEMA (T6 only).
+        Idempotent — checks schema_migrations first and skips if already
+        recorded. Returns True if applied, False if skipped.
+
+        Per the spec, migrations are human-applied only; the agent_id /
+        agent_tier args default to the human-operator T6 identity for
+        the typical `sqlite3 vault.db < migration.sql`-shaped invocation,
+        but a future maintenance agent could pass its own identity for
+        the audit row."""
+        self.gate.check(Capability.VAULT_SCHEMA, agent_id, agent_tier)
+
         sql_path = Path(sql_path)
         sql = sql_path.read_text(encoding="utf-8")
 
@@ -202,7 +194,7 @@ class VaultAdapter:
         for r in rows:
             seen_agents.add((r.profiled_by_agent_id, r.profiled_by_agent_tier))
         for agent_id, agent_tier in seen_agents:
-            self.gate.check_write("kv_sensitivity_profile", agent_id, agent_tier)
+            self.gate.check(Capability.VAULT_APPEND, agent_id, agent_tier)
 
         # Convert dataclasses to dict payloads. ProfileRow.to_vault_payload()
         # is the contract — if it changes, this assert will fire loudly.
@@ -319,8 +311,8 @@ class VaultAdapter:
             if k in payload and isinstance(payload[k], bool):
                 payload[k] = 1 if payload[k] else 0
 
-        self.gate.check_write(
-            "candidate_record",
+        self.gate.check(
+            Capability.VAULT_APPEND,
             payload["discovered_by_agent_id"],
             payload["discovered_by_agent_tier"],
         )