diff --git a/SESSIONS.md b/SESSIONS.md
index 695907f..088522f 100644
--- a/SESSIONS.md
+++ b/SESSIONS.md
@@ -419,3 +419,69 @@
 1. Real GitHub/Jira webhook URLs (ngrok or deploy)
 2. Phase 3: `cortex.query()` hardening + MCP live inject wiring
 3. Graph schema validation against multi-source decisions
+
+---
+
+## Session — 2026-06-10 — Phase 4 kickoff: scoring pipeline enforcement
+**Duration:** ~45m
+**Phase:** Phase 4 — Importance scorer + Trust scorer + RBAC
+
+### Gap analysis (start of session)
+- **Already implemented:** `scoring/importance.py`, `scoring/trust_scorer.py`, `graph/rbac.py`, pipeline worker calling both scorers, `GraphWriter` discard/quarantine gates, RBAC filtering on all `GraphQueryService` read paths + API `resolve_roles`.
+- **Gaps found:** No single enforcement module for write-path scoring; `GraphWriter` did not reject explicitly unscored events (0.0/0.0); `scripts/seed_demo.py` bypassed scorers with hardcoded scores; pipeline discarded low-importance only via writer `ValueError` (late); CMVK majority voting, GDPR cascade, `V004__rbac_enforcement.cypher` migration still outstanding.
+
+### Built
+- **`scoring/write_pipeline.py`** — `DecisionScoringPipeline`, `assert_scored_for_write()`, `write_reject_reason()` — shared write-path gate
+- **`graph/writer.py`** — rejects unscored decisions before threshold checks
+- **`pipeline/extraction_worker.py`** — uses scoring pipeline; early discard/quarantine before Neo4j write
+- **`scripts/seed_demo.py`** — runs scorers before write; skips sub-threshold decisions
+- Tests: `tests/scoring/test_write_pipeline.py`; extended pipeline, writer, RBAC, integration mocks
+
+### State at end
+- Branch **`feature/phase-4-scoring-rbac`** from `origin/main` (Phase 3 not merged — no dependency)
+- **306 passed**, 1 failed (`tests/api/test_telemetry.py` — pre-existing OTLP optional-deps mock; unrelated to Phase 4)
+- Changes **uncommitted** per session instructions
+
+### Next session starts with
+1. GDPR delete cascade + `graph/migrations/V009__rbac_enforcement.cypher`
+2. Commit + PR for Phase 4 branch
+3. LLM-backed CMVK verifiers (replace heuristic stand-ins for production)
+
+---
+
+## Session — 2026-06-10 — Phase 4: CMVK + quarantine audit store
+**Duration:** ~30m
+**Phase:** Phase 4 — Importance scorer + Trust scorer + RBAC
+
+### Built
+- **`scoring/cmvk.py`** — `CrossModelVerificationKernel` with 3-verifier majority vote (2/3) for importance > 0.8; heuristic verifiers for dev/test; `CORTEX_CMVK_ENABLED` toggle
+- **`memory/quarantine.py`** — TimescaleDB audit store for rejected writes (`cortex_quarantine_events`)
+- **`scoring/write_pipeline.py`** — CMVK step between importance and trust; new `cmvk_disagreement` reject reason
+- **`pipeline/extraction_worker.py`** + **`scripts/seed_demo.py`** — persist quarantine records on reject paths
+- Tests: `tests/scoring/test_cmvk.py`, `tests/memory/test_quarantine.py`; extended write_pipeline and extraction_worker tests
+- Fixed pre-existing **`tests/api/test_telemetry.py`** mock path (`api.telemetry.FastAPIInstrumentor`)
+
+### State at end
+- Branch **`feature/phase-4-scoring-rbac`** — uncommitted
+- **317 passed** (full suite, `--no-cov`)
+
+---
+
+## Session — 2026-06-10 — Phase 4: RBAC migration + GDPR cascade
+**Duration:** ~25m
+**Phase:** Phase 4 — Importance scorer + Trust scorer + RBAC
+
+### Built
+- **`graph/migrations/V009__rbac_enforcement.cypher`** — access_policy indices on all memory node types; `GdprAuditLog` constraint + workspace index
+- **`graph/gdpr.py`** — `GdprErasureService.erase_subject()` cascade: Rationale → Contradiction → Decision → Person + audit log
+- **`graph/rbac.py`** — `can_erase()`, `is_gdpr_subject()` helpers
+- Tests: `tests/graph/test_gdpr.py`; extended `test_rbac.py`, `test_migrate_strip.py`
+
+### State at end
+- Branch **`feature/phase-4-scoring-rbac`** — uncommitted
+- **324 passed** (full suite, `--no-cov`)
+
+### Next session starts with
+1. Commit + open PR for Phase 4 branch
+2. LLM-backed CMVK verifiers (production)
+3. GDPR erasure API route (admin-only)
diff --git a/graph/gdpr.py b/graph/gdpr.py
new file mode 100644
index 0000000..c39fa91
--- /dev/null
+++ b/graph/gdpr.py
@@ -0,0 +1,221 @@
+"""GDPR Right to Erasure — cascade delete with audit log.
+
+Architecture: Phase 4 — graph-level RBAC (D-008) + thermodynamic lifecycle (D-007).
+Deletes all Decision/Rationale/Contradiction nodes linked to a data subject,
+then removes the Person node. Every erasure is recorded as a GdprAuditLog node.
+
+Requires V009__rbac_enforcement.cypher (GdprAuditLog schema).
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from dataclasses import dataclass
+from typing import Any
+
+import structlog
+from neo4j import Driver, GraphDatabase
+
+from graph.rbac import can_erase
+
+log = structlog.get_logger(__name__)
+
+_COLLECT_SUBJECT = """
+MATCH (p:Person {id: $person_id, workspace_id: $workspace_id})
+OPTIONAL MATCH (p)-[:MADE]->(d:Decision)
+RETURN collect(DISTINCT d.id) AS decision_ids
+"""
+
+_DELETE_RATIONALES = """
+UNWIND $decision_ids AS decision_id
+MATCH (d:Decision {id: decision_id, workspace_id: $workspace_id})-[:HAS_RATIONALE]->(r:Rationale)
+DETACH DELETE r
+"""
+
+_DELETE_CONTRADICTIONS = """
+UNWIND $decision_ids AS decision_id
+MATCH (c:Contradiction {workspace_id: $workspace_id})-[:INVOLVES_NEW|INVOLVES_PRIOR]->(d:Decision {id: decision_id})
+DETACH DELETE c
+"""
+
+_DELETE_DECISIONS = """
+UNWIND $decision_ids AS decision_id
+MATCH (d:Decision {id: decision_id, workspace_id: $workspace_id})
+DETACH DELETE d
+"""
+
+_DELETE_PERSON = """
+MATCH (p:Person {id: $person_id, workspace_id: $workspace_id})
+DETACH DELETE p
+RETURN count(p) AS deleted
+"""
+
+_CREATE_AUDIT = """
+CREATE (a:GdprAuditLog {
+    id: $audit_id,
+    workspace_id: $workspace_id,
+    subject_id: $person_id,
+    requested_by: $requested_by,
+    reason: $reason,
+    decisions_deleted: $decisions_deleted,
+    deleted_at: datetime()
+})
+RETURN a.id AS id
+"""
+
+
+@dataclass(frozen=True)
+class GdprErasureResult:
+    """Outcome of a GDPR cascade delete."""
+
+    audit_id: str
+    workspace_id: str
+    person_id: str
+    decisions_deleted: int
+    requested_by: str
+
+
+class GdprErasureService:
+    """Cascade-delete a data subject and linked memory nodes."""
+
+    def __init__(
+        self,
+        uri: str | None = None,
+        user: str | None = None,
+        password: str | None = None,
+    ) -> None:
+        self._uri = uri or os.environ.get("NEO4J_URI", "bolt://localhost:7687")
+        self._user = user or os.environ.get("NEO4J_USER", "neo4j")
+        self._password = password or os.environ.get("NEO4J_PASSWORD", "cortex_local")
+        self._driver: Driver = GraphDatabase.driver(
+            self._uri,
+            auth=(self._user, self._password),
+        )
+        log.info("graph.gdpr.initialized", uri=self._uri)
+
+    def erase_subject(
+        self,
+        *,
+        workspace_id: str,
+        person_id: str,
+        requested_by: str,
+        caller_roles: list[str],
+        reason: str = "gdpr_right_to_erasure",
+    ) -> GdprErasureResult:
+        """Delete all memory for a data subject and write an audit log entry.
+
+        Args:
+            workspace_id: Org scope for the erasure.
+            person_id: Canonical person id (email or username).
+            requested_by: DID or user id initiating the request.
+            caller_roles: Roles of the caller — must include admin/gdpr_officer/legal.
+            reason: Human-readable reason stored on the audit node.
+
+        Returns:
+            GdprErasureResult with counts and audit id.
+
+        Raises:
+            PermissionError: When caller lacks erasure privileges.
+            ValueError: When the person node does not exist in the workspace.
+        """
+        if not can_erase(caller_roles):
+            raise PermissionError(
+                "GDPR erasure requires admin, gdpr_officer, or legal role."
+            )
+
+        audit_id = str(uuid.uuid4())
+
+        with self._driver.session() as session:
+            result = session.execute_write(
+                self._erase_transaction,
+                workspace_id=workspace_id,
+                person_id=person_id,
+                requested_by=requested_by,
+                reason=reason,
+                audit_id=audit_id,
+            )
+
+        log.info(
+            "graph.gdpr.erased",
+            workspace_id=workspace_id,
+            person_id=person_id,
+            decisions_deleted=result.decisions_deleted,
+            audit_id=result.audit_id,
+            requested_by=requested_by,
+        )
+        return result
+
+    def close(self) -> None:
+        """Close the Neo4j driver."""
+        self._driver.close()
+
+    @staticmethod
+    def _erase_transaction(
+        tx: Any,
+        *,
+        workspace_id: str,
+        person_id: str,
+        requested_by: str,
+        reason: str,
+        audit_id: str,
+    ) -> GdprErasureResult:
+        """Run cascade delete atomically inside one write transaction."""
+        collect = tx.run(
+            _COLLECT_SUBJECT,
+            person_id=person_id,
+            workspace_id=workspace_id,
+        ).single()
+        if collect is None:
+            raise ValueError(
+                f"Person {person_id!r} not found in workspace {workspace_id!r}."
+            )
+
+        decision_ids: list[str] = [
+            str(item) for item in (collect["decision_ids"] or []) if item
+        ]
+
+        if decision_ids:
+            tx.run(
+                _DELETE_RATIONALES,
+                decision_ids=decision_ids,
+                workspace_id=workspace_id,
+            )
+            tx.run(
+                _DELETE_CONTRADICTIONS,
+                decision_ids=decision_ids,
+                workspace_id=workspace_id,
+            )
+            tx.run(
+                _DELETE_DECISIONS,
+                decision_ids=decision_ids,
+                workspace_id=workspace_id,
+            )
+
+        deleted = tx.run(
+            _DELETE_PERSON,
+            person_id=person_id,
+            workspace_id=workspace_id,
+        ).single()
+        if not deleted or deleted["deleted"] == 0:
+            raise ValueError(
+                f"Person {person_id!r} not found in workspace {workspace_id!r}."
+            )
+
+        tx.run(
+            _CREATE_AUDIT,
+            audit_id=audit_id,
+            workspace_id=workspace_id,
+            person_id=person_id,
+            requested_by=requested_by,
+            reason=reason,
+            decisions_deleted=len(decision_ids),
+        )
+
+        return GdprErasureResult(
+            audit_id=audit_id,
+            workspace_id=workspace_id,
+            person_id=person_id,
+            decisions_deleted=len(decision_ids),
+            requested_by=requested_by,
+        )
diff --git a/graph/migrations/V009__rbac_enforcement.cypher b/graph/migrations/V009__rbac_enforcement.cypher
new file mode 100644
index 0000000..43db1f1
--- /dev/null
+++ b/graph/migrations/V009__rbac_enforcement.cypher
@@ -0,0 +1,34 @@
+// V009 — RBAC enforcement indices + GDPR audit log schema
+// Decision: D-008 — Graph-level RBAC with DID agent identity.
+// Extends V002 access_policy coverage to all memory node types and adds
+// GdprAuditLog for Right-to-Erasure cascade deletes (graph/gdpr.py).
+
+CREATE INDEX person_access_policy IF NOT EXISTS
+  FOR (p:Person) ON (p.access_policy);
+
+CREATE INDEX system_access_policy IF NOT EXISTS
+  FOR (s:System) ON (s.access_policy);
+
+CREATE INDEX rationale_access_policy IF NOT EXISTS
+  FOR (r:Rationale) ON (r.access_policy);
+
+CREATE INDEX exception_access_policy IF NOT EXISTS
+  FOR (e:Exception) ON (e.access_policy);
+
+CREATE INDEX team_access_policy IF NOT EXISTS
+  FOR (t:Team) ON (t.access_policy);
+
+CREATE INDEX outcome_access_policy IF NOT EXISTS
+  FOR (o:Outcome) ON (o.access_policy);
+
+CREATE INDEX contradiction_access_policy IF NOT EXISTS
+  FOR (c:Contradiction) ON (c.access_policy);
+
+CREATE CONSTRAINT gdpr_audit_id_unique IF NOT EXISTS
+  FOR (a:GdprAuditLog) REQUIRE a.id IS UNIQUE;
+
+CREATE INDEX gdpr_audit_workspace IF NOT EXISTS
+  FOR (a:GdprAuditLog) ON (a.workspace_id, a.deleted_at);
+
+MERGE (v:SchemaVersion {version: 9})
+SET v.applied_at = datetime(), v.description = "RBAC enforcement + GDPR audit";
diff --git a/graph/rbac.py b/graph/rbac.py
index 40f3def..22226aa 100644
--- a/graph/rbac.py
+++ b/graph/rbac.py
@@ -12,6 +12,8 @@
     "gdpr_subject": False,
 }
 
+_GDPR_ERASURE_ROLES = frozenset({"admin", "gdpr_officer", "legal"})
+
 
 def normalize_access_policy(policy: Any) -> dict[str, Any]:
     """Coerce stored access_policy values into a dict."""
@@ -42,6 +44,17 @@ def can_access(policy: Any, caller_roles: list[str]) -> bool:
     return bool(allowed_roles.intersection(caller_roles))
 
 
+def can_erase(caller_roles: list[str]) -> bool:
+    """Return True when caller may invoke GDPR Right to Erasure."""
+    return bool(_GDPR_ERASURE_ROLES.intersection(caller_roles))
+
+
+def is_gdpr_subject(policy: Any) -> bool:
+    """Return True when a node access policy marks GDPR-subject data."""
+    normalized = normalize_access_policy(policy)
+    return bool(normalized.get("gdpr_subject", False))
+
+
 def serialize_access_policy(policy: dict[str, Any] | None) -> str:
     """Serialize access policy for Neo4j storage."""
     payload = policy or _DEFAULT_POLICY
diff --git a/graph/writer.py b/graph/writer.py
index eb1f107..bc55a24 100644
--- a/graph/writer.py
+++ b/graph/writer.py
@@ -27,6 +27,7 @@
 
 from graph.rbac import serialize_access_policy
 from scoring.trust_scorer import is_writable
+from scoring.write_pipeline import assert_scored_for_write
 from shared.models import IMPORTANCE_DISCARD, DecisionEvent
 
 log = structlog.get_logger(__name__)
@@ -187,9 +188,11 @@ def write(
             decision.event_id on success.
 
         Raises:
-            ValueError: If importance_score is below discard threshold.
+            ValueError: If scores are unset or below write thresholds.
             Neo4jError: On graph write failure.
         """
+        assert_scored_for_write(decision)
+
         if decision.importance_score < IMPORTANCE_DISCARD:
             log.info(
                 "graph.write.discarded",
diff --git a/memory/quarantine.py b/memory/quarantine.py
new file mode 100644
index 0000000..2722b1a
--- /dev/null
+++ b/memory/quarantine.py
@@ -0,0 +1,101 @@
+"""Quarantine store — audit trail for rejected graph writes.
+
+Architecture: Phase 4 — low-trust and CMVK-failed decisions are not written to
+Neo4j but must be persisted for audit and human review instead of silent drop.
+Uses TimescaleDB when TIMESCALE_HOST is configured (same pattern as episodic).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+from typing import Any
+from urllib.parse import quote_plus
+
+import structlog
+
+from scoring.write_pipeline import WriteRejectReason
+from shared.models import DecisionEvent
+
+log = structlog.get_logger(__name__)
+
+
+def _timescale_dsn() -> str | None:
+    host = os.environ.get("TIMESCALE_HOST")
+    if not host:
+        return None
+    port = int(os.environ.get("TIMESCALE_PORT", "5433"))
+    user = os.environ.get("TIMESCALE_USER", "cortex")
+    password = os.environ.get("TIMESCALE_PASSWORD", "cortex_local")
+    db = os.environ.get("TIMESCALE_DB", "cortex_events")
+    return (
+        f"postgresql://{quote_plus(user)}:{quote_plus(password)}"
+        f"@{host}:{port}/{quote_plus(db)}"
+    )
+
+
+async def _ensure_table(conn: Any) -> None:
+    await conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS cortex_quarantine_events (
+            event_id TEXT NOT NULL,
+            workspace_id TEXT NOT NULL,
+            reject_reason TEXT NOT NULL,
+            importance_score DOUBLE PRECISION,
+            trust_score DOUBLE PRECISION,
+            quarantined_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            payload JSONB NOT NULL,
+            PRIMARY KEY (event_id, reject_reason)
+        );
+        """
+    )
+
+
+async def _persist_async(decision: DecisionEvent, reason: WriteRejectReason) -> None:
+    import asyncpg
+
+    dsn = _timescale_dsn()
+    if dsn is None:
+        return
+    conn = await asyncpg.connect(dsn)
+    try:
+        await _ensure_table(conn)
+        payload = json.dumps(decision.model_dump(mode="json"))
+        await conn.execute(
+            """
+            INSERT INTO cortex_quarantine_events (
+                event_id, workspace_id, reject_reason,
+                importance_score, trust_score, payload
+            )
+            VALUES ($1, $2, $3, $4, $5, $6::jsonb)
+            ON CONFLICT (event_id, reject_reason) DO NOTHING
+            """,
+            decision.event_id,
+            decision.workspace_id,
+            reason,
+            decision.importance_score,
+            decision.trust_score,
+            payload,
+        )
+        log.info(
+            "quarantine.persisted",
+            event_id=decision.event_id,
+            reject_reason=reason,
+            importance_score=decision.importance_score,
+            trust_score=decision.trust_score,
+        )
+    finally:
+        await conn.close()
+
+
+def persist_quarantine(decision: DecisionEvent, reason: WriteRejectReason) -> None:
+    """Persist a rejected decision when Timescale is configured."""
+    if _timescale_dsn() is None:
+        log.debug(
+            "quarantine.skipped_no_dsn",
+            event_id=decision.event_id,
+            reject_reason=reason,
+        )
+        return
+    asyncio.run(_persist_async(decision, reason))
diff --git a/pipeline/extraction_worker.py b/pipeline/extraction_worker.py
index 66ee8a0..1d74191 100644
--- a/pipeline/extraction_worker.py
+++ b/pipeline/extraction_worker.py
@@ -16,9 +16,9 @@
 from graph.writer import GraphWriter
 from intelligence.contradiction_detector import ContradictionDetector
 from memory.episodic import append_raw_event
+from memory.quarantine import persist_quarantine
 from memory.semantic import upsert_decision_vector
-from scoring.importance import ImportanceScorer
-from scoring.trust_scorer import TrustScorer, is_writable
+from scoring.write_pipeline import DecisionScoringPipeline, write_reject_reason
 from shared.models import RawEvent
 
 log = structlog.get_logger(__name__)
@@ -111,8 +111,7 @@ def __init__(
             }
         )
         self._extractor = DecisionExtractor()
-        self._importance = ImportanceScorer()
-        self._trust = TrustScorer()
+        self._scoring = DecisionScoringPipeline()
         self._writer = GraphWriter()
         self._contradictions: ContradictionDetector | None = None
         self._processed = _BoundedSeenCache(PROCESSED_CACHE_SIZE)
@@ -151,15 +150,32 @@ def process_raw_event(self, raw_event: RawEvent) -> str | None:
             self._processed.add(raw_event.event_id)
             return None
 
-        self._importance.score(decision)
-        self._trust.score(decision)
-
-        if not is_writable(decision.trust_score):
-            log.info(
-                "pipeline.event.quarantined",
-                event_id=decision.event_id,
-                trust_score=decision.trust_score,
-            )
+        self._scoring.score(decision)
+
+        reject = write_reject_reason(decision)
+        if reject is not None:
+            persist_quarantine(decision, reject)
+            if reject == "importance_discard":
+                log.info(
+                    "pipeline.event.discarded",
+                    event_id=decision.event_id,
+                    importance_score=decision.importance_score,
+                    reason="importance_below_threshold",
+                )
+            elif reject == "cmvk_disagreement":
+                log.info(
+                    "pipeline.event.quarantined",
+                    event_id=decision.event_id,
+                    importance_score=decision.importance_score,
+                    reason="cmvk_majority_vote_failed",
+                )
+            else:
+                log.info(
+                    "pipeline.event.quarantined",
+                    event_id=decision.event_id,
+                    trust_score=decision.trust_score,
+                    reason=reject,
+                )
             self._processed.add(raw_event.event_id)
             return None
 
diff --git a/scoring/cmvk.py b/scoring/cmvk.py
new file mode 100644
index 0000000..e494afa
--- /dev/null
+++ b/scoring/cmvk.py
@@ -0,0 +1,151 @@
+"""Cross-Model Verification Kernel (CMVK) — majority voting for high-stakes writes.
+
+Architecture: Phase 4 — events with importance > IMPORTANCE_FULL (0.8) require
+verification by three independent verifiers before graph write. Majority vote
+(2/3) required; disagreement quarantines the event for human review.
+
+Decision: D-006 — Bayesian trust scoring with CMVK majority voting.
+
+Production verifiers will call independent LLMs; v0.1 uses deterministic
+heuristic checks so dev/test paths incur zero API cost.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Protocol
+
+import structlog
+
+from shared.models import IMPORTANCE_FULL, DecisionEvent
+
+log = structlog.get_logger(__name__)
+
+CMVK_VERSION = "0.1.0"
+CMVK_VERIFIER_COUNT = 3
+CMVK_MAJORITY = 2
+
+
+class VerifierVote:
+    """Single verifier outcome."""
+
+    __slots__ = ("verifier_id", "approved", "rationale")
+
+    def __init__(self, verifier_id: str, approved: bool, rationale: str) -> None:
+        self.verifier_id = verifier_id
+        self.approved = approved
+        self.rationale = rationale
+
+
+class CMVKResult:
+    """Aggregated CMVK outcome for one decision."""
+
+    __slots__ = ("approved", "votes", "approved_verifier_ids")
+
+    def __init__(
+        self,
+        approved: bool,
+        votes: tuple[VerifierVote, ...],
+        approved_verifier_ids: list[str],
+    ) -> None:
+        self.approved = approved
+        self.votes = votes
+        self.approved_verifier_ids = approved_verifier_ids
+
+
+class DecisionVerifier(Protocol):
+    """Pluggable verifier — LLM-backed in production, heuristic in dev."""
+
+    verifier_id: str
+
+    def verify(self, decision: DecisionEvent) -> VerifierVote: ...
+
+
+class HeuristicDecisionVerifier:
+    """Deterministic verifier stand-in for LLM verifiers (dev/test)."""
+
+    def __init__(self, verifier_id: str, min_confidence: float, require_rationale: bool) -> None:
+        self.verifier_id = verifier_id
+        self._min_confidence = min_confidence
+        self._require_rationale = require_rationale
+
+    def verify(self, decision: DecisionEvent) -> VerifierVote:
+        if len(decision.content.strip()) < 20:
+            return VerifierVote(self.verifier_id, False, "content too short")
+        if decision.extraction_confidence < self._min_confidence:
+            return VerifierVote(
+                self.verifier_id,
+                False,
+                f"confidence below {self._min_confidence}",
+            )
+        if not decision.made_by:
+            return VerifierVote(self.verifier_id, False, "no authors")
+        if self._require_rationale and not decision.rationale:
+            return VerifierVote(self.verifier_id, False, "missing rationale")
+        if not decision.affects and not decision.rationale:
+            return VerifierVote(self.verifier_id, False, "no affects or rationale")
+        return VerifierVote(self.verifier_id, True, "heuristic checks passed")
+
+
+def default_heuristic_verifiers() -> list[HeuristicDecisionVerifier]:
+    """Three independent heuristic verifiers with slightly different strictness."""
+    return [
+        HeuristicDecisionVerifier("cmvk-heuristic-a", min_confidence=0.50, require_rationale=False),
+        HeuristicDecisionVerifier("cmvk-heuristic-b", min_confidence=0.55, require_rationale=False),
+        HeuristicDecisionVerifier("cmvk-heuristic-c", min_confidence=0.50, require_rationale=True),
+    ]
+
+
+class CrossModelVerificationKernel:
+    """Runs CMVK majority voting when importance exceeds the high-stakes threshold."""
+
+    def __init__(
+        self,
+        verifiers: list[DecisionVerifier] | None = None,
+        *,
+        enabled: bool | None = None,
+    ) -> None:
+        self._verifiers = verifiers or default_heuristic_verifiers()
+        if enabled is None:
+            enabled = os.environ.get("CORTEX_CMVK_ENABLED", "true").lower() in {
+                "1",
+                "true",
+                "yes",
+            }
+        self._enabled = enabled
+        log.info(
+            "cmvk.initialized",
+            version=CMVK_VERSION,
+            enabled=self._enabled,
+            verifier_count=len(self._verifiers),
+        )
+
+    @property
+    def enabled(self) -> bool:
+        return self._enabled
+
+    def requires_verification(self, decision: DecisionEvent) -> bool:
+        """Return True when CMVK must run before write."""
+        return self._enabled and decision.importance_score > IMPORTANCE_FULL
+
+    def verify(self, decision: DecisionEvent) -> CMVKResult:
+        """Run all verifiers and apply majority vote."""
+        votes = tuple(v.verify(decision) for v in self._verifiers)
+        approved_ids = [v.verifier_id for v in votes if v.approved]
+        approved = len(approved_ids) >= CMVK_MAJORITY
+
+        log.info(
+            "cmvk.verified",
+            event_id=decision.event_id,
+            importance_score=decision.importance_score,
+            approved=approved,
+            approve_count=len(approved_ids),
+            majority_required=CMVK_MAJORITY,
+            votes=[{"id": v.verifier_id, "approved": v.approved} for v in votes],
+        )
+
+        return CMVKResult(
+            approved=approved,
+            votes=votes,
+            approved_verifier_ids=approved_ids,
+        )
diff --git a/scoring/write_pipeline.py b/scoring/write_pipeline.py
new file mode 100644
index 0000000..42ec141
--- /dev/null
+++ b/scoring/write_pipeline.py
@@ -0,0 +1,92 @@
+"""Central scoring pipeline for all graph writes.
+
+Architecture: Phase 4 — every DecisionEvent must pass ImportanceScorer then
+TrustScorer before GraphWriter.write(). This module is the single enforcement
+point shared by the Kafka worker, demo seed, and any future write paths.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+import structlog
+
+from scoring.cmvk import CrossModelVerificationKernel
+from scoring.importance import ImportanceScorer
+from scoring.trust_scorer import TrustScorer, is_writable
+from shared.models import IMPORTANCE_DISCARD, IMPORTANCE_FULL, DecisionEvent
+
+log = structlog.get_logger(__name__)
+
+WriteRejectReason = Literal[
+    "unscored",
+    "importance_discard",
+    "trust_quarantine",
+    "cmvk_disagreement",
+]
+
+
+def assert_scored_for_write(decision: DecisionEvent) -> None:
+    """Raise when importance/trust scores were never computed."""
+    if decision.importance_score <= 0.0 and decision.trust_score <= 0.0:
+        raise ValueError(
+            "DecisionEvent must pass ImportanceScorer and TrustScorer before graph write. "
+            "Both importance_score and trust_score are unset (0.0)."
+        )
+
+
+def write_reject_reason(decision: DecisionEvent) -> WriteRejectReason | None:
+    """Return why a scored decision must not be persisted, or None if writable."""
+    assert_scored_for_write(decision)
+    if (
+        decision.importance_score > IMPORTANCE_FULL
+        and decision.status == "under_review"
+    ):
+        return "cmvk_disagreement"
+    if decision.importance_score < IMPORTANCE_DISCARD:
+        return "importance_discard"
+    if not is_writable(decision.trust_score):
+        return "trust_quarantine"
+    return None
+
+
+class DecisionScoringPipeline:
+    """Runs importance → CMVK (high-stakes) → trust scoring in the required order."""
+
+    def __init__(
+        self,
+        importance: ImportanceScorer | None = None,
+        trust: TrustScorer | None = None,
+        cmvk: CrossModelVerificationKernel | None = None,
+    ) -> None:
+        self._importance = importance or ImportanceScorer()
+        self._trust = trust or TrustScorer()
+        self._cmvk = cmvk or CrossModelVerificationKernel()
+
+    def score(self, decision: DecisionEvent) -> DecisionEvent:
+        """Apply importance, optional CMVK, then trust scoring, in place."""
+        self._importance.score(decision)
+
+        if self._cmvk.requires_verification(decision):
+            result = self._cmvk.verify(decision)
+            if result.approved:
+                decision.provenance.verified_by = result.approved_verifier_ids
+            else:
+                decision.status = "under_review"
+                log.info(
+                    "write_pipeline.cmvk_rejected",
+                    event_id=decision.event_id,
+                    importance_score=decision.importance_score,
+                    approve_count=len(result.approved_verifier_ids),
+                )
+                return decision
+
+        self._trust.score(decision)
+        log.debug(
+            "write_pipeline.scored",
+            event_id=decision.event_id,
+            importance_score=decision.importance_score,
+            trust_score=decision.trust_score,
+            verified_by=decision.provenance.verified_by,
+        )
+        return decision
diff --git a/scripts/seed_demo.py b/scripts/seed_demo.py
index d8d6dab..04e4e46 100755
--- a/scripts/seed_demo.py
+++ b/scripts/seed_demo.py
@@ -79,17 +79,29 @@ def main() -> int:
         return 0
 
     from graph.writer import GraphWriter
+    from memory.quarantine import persist_quarantine
+    from scoring.write_pipeline import DecisionScoringPipeline, write_reject_reason
 
+    scoring = DecisionScoringPipeline()
     writer = GraphWriter()
     written: list[str] = []
+    skipped = 0
     try:
         for decision in decisions:
+            scoring.score(decision)
+            reject = write_reject_reason(decision)
+            if reject is not None:
+                persist_quarantine(decision, reject)
+                skipped += 1
+                continue
             written.append(writer.write(decision))
     finally:
         writer.close()
 
     print(f"Demo seed complete for workspace={args.workspace!r} scale={scale!r}")
     print(f"  Wrote {len(written)} decisions")
+    if skipped:
+        print(f"  Skipped {skipped} decisions (importance/trust thresholds)")
     sample = json.dumps(
         {
             "query": "Why CockroachDB payments?",
diff --git a/tests/api/test_telemetry.py b/tests/api/test_telemetry.py
index 6c15eb7..a60a785 100644
--- a/tests/api/test_telemetry.py
+++ b/tests/api/test_telemetry.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import sys
 from unittest.mock import MagicMock, patch
 
 from fastapi import FastAPI
@@ -10,6 +11,30 @@
 from api.telemetry import setup_telemetry
 
 
+def _install_fake_otel_modules(instrumentor_cls: MagicMock) -> None:
+    """Inject stub opentelemetry modules (optional dep may be absent in CI)."""
+    trace_mod = MagicMock()
+    exporter_mod = MagicMock(OTLPSpanExporter=MagicMock())
+    resource_mod = MagicMock(Resource=MagicMock(create=MagicMock(return_value="resource")))
+    provider_mod = MagicMock(TracerProvider=MagicMock(return_value=MagicMock()))
+    export_mod = MagicMock(BatchSpanProcessor=MagicMock())
+
+    sys.modules["opentelemetry"] = MagicMock(trace=trace_mod)
+    sys.modules["opentelemetry.instrumentation"] = MagicMock()
+    sys.modules["opentelemetry.instrumentation.fastapi"] = MagicMock(
+        FastAPIInstrumentor=instrumentor_cls,
+    )
+    sys.modules["opentelemetry.exporter"] = MagicMock()
+    sys.modules["opentelemetry.exporter.otlp"] = MagicMock()
+    sys.modules["opentelemetry.exporter.otlp.proto"] = MagicMock()
+    sys.modules["opentelemetry.exporter.otlp.proto.http"] = MagicMock()
+    sys.modules["opentelemetry.exporter.otlp.proto.http.trace_exporter"] = exporter_mod
+    sys.modules["opentelemetry.sdk"] = MagicMock()
+    sys.modules["opentelemetry.sdk.resources"] = resource_mod
+    sys.modules["opentelemetry.sdk.trace"] = provider_mod
+    sys.modules["opentelemetry.sdk.trace.export"] = export_mod
+
+
 def test_setup_telemetry_skipped_without_endpoint(monkeypatch) -> None:
     telemetry_module._CONFIGURED = False
     monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
@@ -23,7 +48,8 @@ def test_setup_telemetry_enables_with_endpoint(monkeypatch) -> None:
     monkeypatch.setenv("OTEL_SERVICE_NAME", "cortex-api-test")
     app = FastAPI()
 
-    with patch("opentelemetry.instrumentation.fastapi.FastAPIInstrumentor") as instrumentor:
-        instrumentor.instrument_app = MagicMock()
+    instrumentor_cls = MagicMock()
+    with patch.dict(sys.modules, {}, clear=False):
+        _install_fake_otel_modules(instrumentor_cls)
         assert setup_telemetry(app) is True
-        instrumentor.instrument_app.assert_called_once()
+        instrumentor_cls.instrument_app.assert_called_once()
diff --git a/tests/graph/test_gdpr.py b/tests/graph/test_gdpr.py
new file mode 100644
index 0000000..4061d4c
--- /dev/null
+++ b/tests/graph/test_gdpr.py
@@ -0,0 +1,103 @@
+"""Tests for graph/gdpr.py — GDPR cascade delete."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from graph.gdpr import GdprErasureService
+
+
+@patch("graph.gdpr.GraphDatabase")
+def test_erase_subject_requires_privileged_role(mock_gdb: MagicMock) -> None:
+    mock_gdb.driver.return_value = MagicMock()
+    service = GdprErasureService(uri="bolt://x", user="u", password="p")
+
+    with pytest.raises(PermissionError, match="GDPR erasure requires"):
+        service.erase_subject(
+            workspace_id="ws-1",
+            person_id="alice@company.com",
+            requested_by="agent-1",
+            caller_roles=["viewer"],
+        )
+
+
+@patch("graph.gdpr.GraphDatabase")
+def test_erase_subject_cascade_and_audit(mock_gdb: MagicMock) -> None:
+    driver = MagicMock()
+    mock_gdb.driver.return_value = driver
+    session = MagicMock()
+    driver.session.return_value.__enter__.return_value = session
+
+    def _execute_write(fn: object, **kwargs: object) -> object:
+        tx = MagicMock()
+
+        def _single() -> dict[str, object]:
+            return {"decision_ids": ["dec-1", "dec-2"]}
+
+        def _single_person() -> dict[str, object]:
+            return {"deleted": 1}
+
+        tx.run.side_effect = [
+            MagicMock(single=_single),
+            MagicMock(),
+            MagicMock(),
+            MagicMock(),
+            MagicMock(single=_single_person),
+            MagicMock(),
+        ]
+        return fn(
+            tx,
+            workspace_id="ws-1",
+            person_id="alice@company.com",
+            requested_by="admin-1",
+            reason="gdpr_right_to_erasure",
+            audit_id="audit-123",
+        )
+
+    session.execute_write.side_effect = _execute_write
+
+    service = GdprErasureService(uri="bolt://x", user="u", password="p")
+    result = service.erase_subject(
+        workspace_id="ws-1",
+        person_id="alice@company.com",
+        requested_by="admin-1",
+        caller_roles=["admin"],
+    )
+
+    assert result.decisions_deleted == 2
+    assert result.person_id == "alice@company.com"
+    assert result.requested_by == "admin-1"
+    assert result.audit_id
+
+
+@patch("graph.gdpr.GraphDatabase")
+def test_erase_subject_raises_when_person_missing(mock_gdb: MagicMock) -> None:
+    driver = MagicMock()
+    mock_gdb.driver.return_value = driver
+    session = MagicMock()
+    driver.session.return_value.__enter__.return_value = session
+
+    def _execute_write(fn: object, **kwargs: object) -> object:
+        tx = MagicMock()
+        tx.run.return_value = MagicMock(single=lambda: None)
+        return fn(
+            tx,
+            workspace_id="ws-1",
+            person_id="missing@company.com",
+            requested_by="admin-1",
+            reason="gdpr_right_to_erasure",
+            audit_id="audit-123",
+        )
+
+    session.execute_write.side_effect = _execute_write
+
+    service = GdprErasureService(uri="bolt://x", user="u", password="p")
+    with pytest.raises(ValueError, match="not found"):
+        service.erase_subject(
+            workspace_id="ws-1",
+            person_id="missing@company.com",
+            requested_by="admin-1",
+            caller_roles=["legal"],
+        )
diff --git a/tests/graph/test_migrate_strip.py b/tests/graph/test_migrate_strip.py
index b3217b8..2aa9de0 100644
--- a/tests/graph/test_migrate_strip.py
+++ b/tests/graph/test_migrate_strip.py
@@ -2,7 +2,12 @@
 
 from __future__ import annotations
 
-from graph.migrate import _strip_leading_comments
+from graph.migrate import _strip_leading_comments, get_migration_files
+
+
+def test_v009_rbac_enforcement_migration_registered() -> None:
+    versions = {version for version, _, _ in get_migration_files()}
+    assert 9 in versions
 
 
 def test_strip_leading_comments_preserves_create_after_header() -> None:
diff --git a/tests/graph/test_rbac.py b/tests/graph/test_rbac.py
index 776d3b6..eeecaba 100644
--- a/tests/graph/test_rbac.py
+++ b/tests/graph/test_rbac.py
@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from graph.rbac import can_access, normalize_access_policy, serialize_access_policy
+from graph.rbac import can_access, can_erase, is_gdpr_subject, normalize_access_policy, serialize_access_policy
 
 
 def test_normalize_json_policy() -> None:
@@ -17,3 +17,23 @@ def test_can_access_allows_authenticated_role() -> None:
 def test_can_access_respects_deny_list() -> None:
     policy = {"roles": ["authenticated"], "deny": ["contractor"]}
     assert not can_access(policy, ["contractor"])
+
+
+def test_can_access_requires_explicit_role_when_not_authenticated() -> None:
+    policy = {"roles": ["admin", "legal"], "deny": []}
+    assert can_access(policy, ["admin"])
+    assert not can_access(policy, ["viewer"])
+
+
+def test_can_erase_allows_privileged_roles() -> None:
+    assert can_erase(["admin"])
+    assert can_erase(["gdpr_officer"])
+    assert can_erase(["legal"])
+    assert not can_erase(["viewer"])
+    assert not can_erase(["authenticated"])
+
+
+def test_is_gdpr_subject_reads_policy_flag() -> None:
+    assert is_gdpr_subject({"roles": ["authenticated"], "gdpr_subject": True})
+    assert not is_gdpr_subject({"roles": ["authenticated"], "gdpr_subject": False})
+    assert not is_gdpr_subject(None)
diff --git a/tests/graph/test_writer.py b/tests/graph/test_writer.py
index 8c13dac..e14c693 100644
--- a/tests/graph/test_writer.py
+++ b/tests/graph/test_writer.py
@@ -52,6 +52,7 @@ def _make_provenance() -> Provenance:
 
 def _make_decision(
     importance_score: float = 0.75,
+    trust_score: float = 0.80,
     made_by: list[str] | None = None,
     affects: list[str] | None = None,
     rationale: list[str] | None = None,
@@ -68,7 +69,7 @@ def _make_decision(
         replaces=replaces,
         extraction_confidence=0.90,
         importance_score=importance_score,
-        trust_score=0.80,
+        trust_score=trust_score,
         provenance=_make_provenance(),
         extracted_at=NOW,
     )
@@ -109,6 +110,15 @@ def test_initialises_with_explicit_params(self, mock_gdb: MagicMock) -> None:
 
 
 class TestWriteImportanceThreshold:
+    @patch("graph.writer.GraphDatabase")
+    def test_raises_when_unscored(self, mock_gdb: MagicMock) -> None:
+        mock_gdb.driver.return_value = MagicMock()
+        writer = GraphWriter(uri="bolt://x", user="u", password="p")
+        decision = _make_decision(importance_score=0.0, trust_score=0.0)
+
+        with pytest.raises(ValueError, match="ImportanceScorer and TrustScorer"):
+            writer.write(decision)
+
     @patch("graph.writer.GraphDatabase")
     def test_raises_below_importance_discard(self, mock_gdb: MagicMock) -> None:
         mock_gdb.driver.return_value = MagicMock()
diff --git a/tests/integration/test_github_pipeline_e2e.py b/tests/integration/test_github_pipeline_e2e.py
index 4f00871..3f2c2da 100644
--- a/tests/integration/test_github_pipeline_e2e.py
+++ b/tests/integration/test_github_pipeline_e2e.py
@@ -50,8 +50,7 @@ def _github_decision(raw: RawEvent) -> DecisionEvent:
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -59,25 +58,20 @@ def test_github_pr_through_worker_scores_before_write(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     """GitHub RawEvent runs extract → importance → trust → Neo4j write in order."""
     raw_event = _github_raw()
     decision = _github_decision(raw_event)
 
-    def _apply_importance(d: DecisionEvent) -> DecisionEvent:
+    def _apply_scores(d: DecisionEvent) -> DecisionEvent:
         d.importance_score = 0.82
-        return d
-
-    def _apply_trust(d: DecisionEvent) -> DecisionEvent:
         d.trust_score = 0.76
         return d
 
     extractor_cls.return_value.extract.return_value = decision
-    importance_cls.return_value.score.side_effect = _apply_importance
-    trust_cls.return_value.score.side_effect = _apply_trust
+    scoring_cls.return_value.score.side_effect = _apply_scores
     writer_cls.return_value.write.return_value = decision.event_id
 
     worker = ExtractionWorker(bootstrap_servers="localhost:9092")
@@ -85,6 +79,5 @@ def _apply_trust(d: DecisionEvent) -> DecisionEvent:
 
     assert event_id == decision.event_id
     extractor_cls.return_value.extract.assert_called_once_with(raw_event)
-    importance_cls.return_value.score.assert_called_once()
-    trust_cls.return_value.score.assert_called_once()
+    scoring_cls.return_value.score.assert_called_once()
     writer_cls.return_value.write.assert_called_once_with(decision)
diff --git a/tests/integration/test_jira_pipeline_e2e.py b/tests/integration/test_jira_pipeline_e2e.py
index 3887963..a538185 100644
--- a/tests/integration/test_jira_pipeline_e2e.py
+++ b/tests/integration/test_jira_pipeline_e2e.py
@@ -50,8 +50,7 @@ def _jira_decision(raw: RawEvent) -> DecisionEvent:
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -59,25 +58,20 @@ def test_jira_comment_through_worker_scores_before_write(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     """Jira RawEvent runs extract → importance → trust → Neo4j write in order."""
     raw_event = _jira_raw()
     decision = _jira_decision(raw_event)
 
-    def _apply_importance(d: DecisionEvent) -> DecisionEvent:
+    def _apply_scores(d: DecisionEvent) -> DecisionEvent:
         d.importance_score = 0.8
-        return d
-
-    def _apply_trust(d: DecisionEvent) -> DecisionEvent:
         d.trust_score = 0.74
         return d
 
     extractor_cls.return_value.extract.return_value = decision
-    importance_cls.return_value.score.side_effect = _apply_importance
-    trust_cls.return_value.score.side_effect = _apply_trust
+    scoring_cls.return_value.score.side_effect = _apply_scores
     writer_cls.return_value.write.return_value = decision.event_id
 
     worker = ExtractionWorker(bootstrap_servers="localhost:9092")
@@ -85,6 +79,5 @@ def _apply_trust(d: DecisionEvent) -> DecisionEvent:
 
     assert event_id == decision.event_id
     extractor_cls.return_value.extract.assert_called_once_with(raw_event)
-    importance_cls.return_value.score.assert_called_once()
-    trust_cls.return_value.score.assert_called_once()
+    scoring_cls.return_value.score.assert_called_once()
     writer_cls.return_value.write.assert_called_once_with(decision)
diff --git a/tests/integration/test_pipeline_e2e.py b/tests/integration/test_pipeline_e2e.py
index a065bc2..403a6fa 100644
--- a/tests/integration/test_pipeline_e2e.py
+++ b/tests/integration/test_pipeline_e2e.py
@@ -12,8 +12,7 @@
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -21,8 +20,7 @@ def test_webhook_to_graph_pipeline(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     raw_event = RawEvent(
@@ -59,8 +57,7 @@ def test_webhook_to_graph_pipeline(
     )
 
     extractor_cls.return_value.extract.return_value = decision
-    importance_cls.return_value.score.side_effect = lambda item: item
-    trust_cls.return_value.score.side_effect = lambda item: item
+    scoring_cls.return_value.score.side_effect = lambda item: item
     writer_cls.return_value.write.return_value = decision.event_id
 
     worker = ExtractionWorker(bootstrap_servers="localhost:9092")
diff --git a/tests/integration/test_pipeline_failures.py b/tests/integration/test_pipeline_failures.py
index 45096f4..0dc99d1 100644
--- a/tests/integration/test_pipeline_failures.py
+++ b/tests/integration/test_pipeline_failures.py
@@ -25,8 +25,7 @@ def _raw() -> RawEvent:
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -34,8 +33,7 @@ def test_extractor_returns_none_skips_write(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     extractor_cls.return_value.extract.return_value = None
@@ -45,8 +43,7 @@ def test_extractor_returns_none_skips_write(
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -54,8 +51,7 @@ def test_low_importance_discarded_before_write(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     raw = _raw()
@@ -81,18 +77,15 @@ def test_low_importance_discarded_before_write(
         extracted_at=NOW,
     )
     extractor_cls.return_value.extract.return_value = decision
-    importance_cls.return_value.score.side_effect = lambda item: item
-    trust_cls.return_value.score.side_effect = lambda item: item
-    writer_cls.return_value.write.side_effect = ValueError("below discard threshold")
+    scoring_cls.return_value.score.side_effect = lambda item: item
 
     worker = ExtractionWorker(bootstrap_servers="localhost:9092")
     assert worker.process_raw_event(raw) is None
-    writer_cls.return_value.write.assert_called_once()
+    writer_cls.return_value.write.assert_not_called()
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -100,8 +93,7 @@ def test_invalid_message_routed_to_dlq(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     worker = ExtractionWorker(bootstrap_servers="localhost:9092")
diff --git a/tests/integration/test_slack_pipeline_e2e.py b/tests/integration/test_slack_pipeline_e2e.py
index d84defa..a47b2be 100644
--- a/tests/integration/test_slack_pipeline_e2e.py
+++ b/tests/integration/test_slack_pipeline_e2e.py
@@ -50,8 +50,7 @@ def _slack_decision(raw: RawEvent) -> DecisionEvent:
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -59,33 +58,27 @@ def test_slack_message_through_worker_scores_before_write(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     """Slack RawEvent runs extract → importance → trust → Neo4j write in order."""
     raw_event = _slack_raw()
     decision = _slack_decision(raw_event)
 
-    def _apply_importance(d: DecisionEvent) -> DecisionEvent:
+    def _apply_scores(d: DecisionEvent) -> DecisionEvent:
         d.importance_score = 0.85
-        return d
-
-    def _apply_trust(d: DecisionEvent) -> DecisionEvent:
         d.trust_score = 0.78
         return d
 
     extractor_cls.return_value.extract.return_value = decision
-    importance_cls.return_value.score.side_effect = _apply_importance
-    trust_cls.return_value.score.side_effect = _apply_trust
+    scoring_cls.return_value.score.side_effect = _apply_scores
     writer_cls.return_value.write.return_value = decision.event_id
 
     worker = ExtractionWorker(bootstrap_servers="localhost:9092")
     assert worker.process_raw_event(raw_event) == decision.event_id
 
     extractor_cls.return_value.extract.assert_called_once_with(raw_event)
-    importance_cls.return_value.score.assert_called_once()
-    trust_cls.return_value.score.assert_called_once()
+    scoring_cls.return_value.score.assert_called_once()
     writer_cls.return_value.write.assert_called_once()
 
     written = writer_cls.return_value.write.call_args[0][0]
diff --git a/tests/memory/test_quarantine.py b/tests/memory/test_quarantine.py
new file mode 100644
index 0000000..c01fe46
--- /dev/null
+++ b/tests/memory/test_quarantine.py
@@ -0,0 +1,51 @@
+"""Tests for memory/quarantine.py."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from unittest.mock import patch
+
+from memory.quarantine import persist_quarantine
+from shared.models import DecisionEvent, Provenance
+
+NOW = datetime(2026, 5, 11, 12, 0, 0, tzinfo=UTC)
+
+
+def _decision() -> DecisionEvent:
+    return DecisionEvent(
+        source_raw_event_id="raw-1",
+        workspace_id="ws-1",
+        event_type="decision",
+        content="We decided to migrate payments to CockroachDB.",
+        made_by=["alice@company.com"],
+        affects=["payments-service"],
+        rationale=["Scale ceiling"],
+        extraction_confidence=0.9,
+        importance_score=0.75,
+        trust_score=0.1,
+        provenance=Provenance(
+            source="slack",
+            channel="C-engineering",
+            original_timestamp=NOW,
+            extractor_version="0.1.0",
+            extractor_model="gpt-4o",
+            verified_by=[],
+            raw_event_id="raw-1",
+        ),
+        extracted_at=NOW,
+    )
+
+
+@patch("memory.quarantine._timescale_dsn", return_value=None)
+def test_persist_quarantine_noop_without_dsn(_dsn: object) -> None:
+    persist_quarantine(_decision(), "trust_quarantine")
+
+
+@patch("memory.quarantine.asyncio.run")
+@patch("memory.quarantine._timescale_dsn", return_value="postgresql://x")
+def test_persist_quarantine_calls_async_when_dsn_set(
+    _dsn: object,
+    run_mock: object,
+) -> None:
+    persist_quarantine(_decision(), "cmvk_disagreement")
+    run_mock.assert_called_once()
diff --git a/tests/pipeline/test_extraction_worker.py b/tests/pipeline/test_extraction_worker.py
index 7ae1c69..f4c7c1c 100644
--- a/tests/pipeline/test_extraction_worker.py
+++ b/tests/pipeline/test_extraction_worker.py
@@ -50,8 +50,7 @@ def _decision() -> DecisionEvent:
 
 
 @patch("pipeline.extraction_worker.GraphWriter")
-@patch("pipeline.extraction_worker.TrustScorer")
-@patch("pipeline.extraction_worker.ImportanceScorer")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
 @patch("pipeline.extraction_worker.DecisionExtractor")
 @patch("pipeline.extraction_worker.Producer")
 @patch("pipeline.extraction_worker.Consumer")
@@ -59,16 +58,13 @@ def test_process_raw_event_writes_and_publishes(
     consumer_cls: MagicMock,
     producer_cls: MagicMock,
     extractor_cls: MagicMock,
-    importance_cls: MagicMock,
-    trust_cls: MagicMock,
+    scoring_cls: MagicMock,
     writer_cls: MagicMock,
 ) -> None:
     extractor = extractor_cls.return_value
     extractor.extract.return_value = _decision()
-    importance = importance_cls.return_value
-    importance.score.side_effect = lambda decision: decision
-    trust = trust_cls.return_value
-    trust.score.side_effect = lambda decision: decision
+    scoring = scoring_cls.return_value
+    scoring.score.side_effect = lambda decision: decision
     writer = writer_cls.return_value
     writer.write.return_value = "decision-1"
 
@@ -76,10 +72,61 @@ def test_process_raw_event_writes_and_publishes(
     event_id = worker.process_raw_event(_raw_event())
 
     assert event_id == "decision-1"
+    scoring.score.assert_called_once()
     writer.write.assert_called_once()
     producer_cls.return_value.produce.assert_called_once()
 
 
+@patch("pipeline.extraction_worker.persist_quarantine")
+@patch("pipeline.extraction_worker.GraphWriter")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
+@patch("pipeline.extraction_worker.DecisionExtractor")
+@patch("pipeline.extraction_worker.Producer")
+@patch("pipeline.extraction_worker.Consumer")
+def test_process_raw_event_quarantines_low_trust(
+    consumer_cls: MagicMock,
+    producer_cls: MagicMock,
+    extractor_cls: MagicMock,
+    scoring_cls: MagicMock,
+    writer_cls: MagicMock,
+    quarantine_mock: MagicMock,
+) -> None:
+    decision = _decision()
+    decision.trust_score = 0.1
+    decision.importance_score = 0.75
+    extractor_cls.return_value.extract.return_value = decision
+    scoring_cls.return_value.score.side_effect = lambda d: d
+
+    worker = ExtractionWorker(bootstrap_servers="localhost:9092")
+    assert worker.process_raw_event(_raw_event()) is None
+    writer_cls.return_value.write.assert_not_called()
+    producer_cls.return_value.produce.assert_not_called()
+    quarantine_mock.assert_called_once()
+
+
+@patch("pipeline.extraction_worker.GraphWriter")
+@patch("pipeline.extraction_worker.DecisionScoringPipeline")
+@patch("pipeline.extraction_worker.DecisionExtractor")
+@patch("pipeline.extraction_worker.Producer")
+@patch("pipeline.extraction_worker.Consumer")
+def test_process_raw_event_discards_low_importance(
+    consumer_cls: MagicMock,
+    producer_cls: MagicMock,
+    extractor_cls: MagicMock,
+    scoring_cls: MagicMock,
+    writer_cls: MagicMock,
+) -> None:
+    decision = _decision()
+    decision.importance_score = 0.1
+    decision.trust_score = 0.8
+    extractor_cls.return_value.extract.return_value = decision
+    scoring_cls.return_value.score.side_effect = lambda d: d
+
+    worker = ExtractionWorker(bootstrap_servers="localhost:9092")
+    assert worker.process_raw_event(_raw_event()) is None
+    writer_cls.return_value.write.assert_not_called()
+
+
 def test_bounded_seen_cache_evicts_lru() -> None:
     """The dedup cache must drop the oldest entry once capacity is reached."""
     cache = _BoundedSeenCache(capacity=3)
diff --git a/tests/scoring/test_cmvk.py b/tests/scoring/test_cmvk.py
new file mode 100644
index 0000000..c936816
--- /dev/null
+++ b/tests/scoring/test_cmvk.py
@@ -0,0 +1,112 @@
+"""Tests for scoring/cmvk.py."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from scoring.cmvk import (
+    CMVK_MAJORITY,
+    CrossModelVerificationKernel,
+    HeuristicDecisionVerifier,
+    VerifierVote,
+)
+from shared.models import IMPORTANCE_FULL, DecisionEvent, Provenance
+
+NOW = datetime(2026, 5, 11, 12, 0, 0, tzinfo=UTC)
+
+
+def _decision(**overrides: object) -> DecisionEvent:
+    base = DecisionEvent(
+        source_raw_event_id="raw-1",
+        workspace_id="ws-1",
+        event_type="decision",
+        content="We decided to migrate payments to CockroachDB for scale.",
+        made_by=["alice@company.com"],
+        affects=["payments-service"],
+        rationale=["Scale ceiling at 10M txn/day"],
+        extraction_confidence=0.9,
+        importance_score=IMPORTANCE_FULL + 0.05,
+        provenance=Provenance(
+            source="github",
+            channel="payments",
+            original_timestamp=NOW,
+            extractor_version="0.1.0",
+            extractor_model="gpt-4o",
+            verified_by=[],
+            raw_event_id="raw-1",
+        ),
+        extracted_at=NOW,
+    )
+    for key, value in overrides.items():
+        setattr(base, key, value)
+    return base
+
+
+class _FixedVerifier:
+    def __init__(self, verifier_id: str, approved: bool) -> None:
+        self.verifier_id = verifier_id
+        self._approved = approved
+
+    def verify(self, decision: DecisionEvent) -> VerifierVote:
+        return VerifierVote(self.verifier_id, self._approved, "fixed")
+
+
+def test_requires_verification_above_threshold() -> None:
+    kernel = CrossModelVerificationKernel(enabled=True)
+    assert kernel.requires_verification(_decision(importance_score=IMPORTANCE_FULL + 0.01))
+    assert not kernel.requires_verification(_decision(importance_score=IMPORTANCE_FULL))
+    assert not kernel.requires_verification(_decision(importance_score=0.75))
+
+
+def test_majority_vote_approves_with_two_of_three() -> None:
+    kernel = CrossModelVerificationKernel(
+        verifiers=[
+            _FixedVerifier("v1", True),
+            _FixedVerifier("v2", True),
+            _FixedVerifier("v3", False),
+        ],
+        enabled=True,
+    )
+    result = kernel.verify(_decision())
+    assert result.approved
+    assert result.approved_verifier_ids == ["v1", "v2"]
+    assert len(result.votes) == 3
+
+
+def test_majority_vote_rejects_with_one_of_three() -> None:
+    kernel = CrossModelVerificationKernel(
+        verifiers=[
+            _FixedVerifier("v1", True),
+            _FixedVerifier("v2", False),
+            _FixedVerifier("v3", False),
+        ],
+        enabled=True,
+    )
+    result = kernel.verify(_decision())
+    assert not result.approved
+    assert result.approved_verifier_ids == ["v1"]
+
+
+def test_heuristic_verifiers_approve_well_formed_decision() -> None:
+    kernel = CrossModelVerificationKernel(enabled=True)
+    result = kernel.verify(_decision())
+    assert result.approved
+    assert len(result.approved_verifier_ids) >= CMVK_MAJORITY
+
+
+def test_heuristic_verifiers_reject_sparse_decision() -> None:
+    kernel = CrossModelVerificationKernel(enabled=True)
+    sparse = _decision(
+        content="ok",
+        made_by=[],
+        affects=[],
+        rationale=[],
+        extraction_confidence=0.2,
+    )
+    result = kernel.verify(sparse)
+    assert not result.approved
+
+
+def test_disabled_kernel_skips_verification_requirement() -> None:
+    kernel = CrossModelVerificationKernel(enabled=False)
+    assert not kernel.requires_verification(_decision(importance_score=0.95))
diff --git a/tests/scoring/test_write_pipeline.py b/tests/scoring/test_write_pipeline.py
new file mode 100644
index 0000000..1516c61
--- /dev/null
+++ b/tests/scoring/test_write_pipeline.py
@@ -0,0 +1,154 @@
+"""Tests for scoring/write_pipeline.py."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from unittest.mock import MagicMock
+
+import pytest
+
+from scoring.write_pipeline import (
+    DecisionScoringPipeline,
+    assert_scored_for_write,
+    write_reject_reason,
+)
+from shared.models import IMPORTANCE_DISCARD, IMPORTANCE_FULL, TRUST_QUARANTINED, DecisionEvent, Provenance
+
+NOW = datetime(2026, 5, 11, 12, 0, 0, tzinfo=UTC)
+
+
+def _decision(**overrides: object) -> DecisionEvent:
+    base = DecisionEvent(
+        source_raw_event_id="raw-1",
+        workspace_id="ws-1",
+        event_type="decision",
+        content="We decided to migrate payments to CockroachDB for scale.",
+        made_by=["alice@company.com"],
+        affects=["payments-service"],
+        rationale=["Scale ceiling at 10M txn/day"],
+        extraction_confidence=0.9,
+        provenance=Provenance(
+            source="github",
+            channel="payments",
+            original_timestamp=NOW,
+            extractor_version="0.1.0",
+            extractor_model="gpt-4o",
+            verified_by=[],
+            raw_event_id="raw-1",
+        ),
+        extracted_at=NOW,
+    )
+    for key, value in overrides.items():
+        setattr(base, key, value)
+    return base
+
+
+def test_assert_scored_for_write_rejects_unscored() -> None:
+    decision = _decision(importance_score=0.0, trust_score=0.0)
+    with pytest.raises(ValueError, match="ImportanceScorer and TrustScorer"):
+        assert_scored_for_write(decision)
+
+
+def test_scoring_pipeline_sets_both_scores() -> None:
+    importance = MagicMock()
+    trust = MagicMock()
+
+    def _set_importance(decision: DecisionEvent) -> DecisionEvent:
+        decision.importance_score = 0.75
+        return decision
+
+    def _set_trust(decision: DecisionEvent) -> DecisionEvent:
+        decision.trust_score = 0.8
+        return decision
+
+    importance.score.side_effect = _set_importance
+    trust.score.side_effect = _set_trust
+
+    pipeline = DecisionScoringPipeline(importance=importance, trust=trust)
+    decision = _decision()
+    pipeline.score(decision)
+
+    importance.score.assert_called_once()
+    trust.score.assert_called_once()
+    assert decision.importance_score == 0.75
+    assert decision.trust_score == 0.8
+
+
+def test_write_reject_reason_importance_discard() -> None:
+    decision = _decision(
+        importance_score=IMPORTANCE_DISCARD - 0.05,
+        trust_score=0.8,
+    )
+    assert write_reject_reason(decision) == "importance_discard"
+
+
+def test_write_reject_reason_trust_quarantine() -> None:
+    decision = _decision(
+        importance_score=0.75,
+        trust_score=TRUST_QUARANTINED - 0.05,
+    )
+    assert write_reject_reason(decision) == "trust_quarantine"
+
+
+def test_write_reject_reason_none_when_writable() -> None:
+    decision = _decision(importance_score=0.75, trust_score=0.8)
+    assert write_reject_reason(decision) is None
+
+
+def test_write_reject_reason_cmvk_disagreement() -> None:
+    decision = _decision(
+        importance_score=IMPORTANCE_FULL + 0.05,
+        trust_score=0.0,
+        status="under_review",
+    )
+    assert write_reject_reason(decision) == "cmvk_disagreement"
+
+
+def test_scoring_pipeline_runs_cmvk_for_high_importance() -> None:
+    importance = MagicMock()
+    trust = MagicMock()
+    cmvk = MagicMock()
+    cmvk.requires_verification.return_value = True
+    cmvk.verify.return_value = MagicMock(approved=True, approved_verifier_ids=["v1", "v2"])
+
+    def _set_importance(decision: DecisionEvent) -> DecisionEvent:
+        decision.importance_score = IMPORTANCE_FULL + 0.05
+        return decision
+
+    importance.score.side_effect = _set_importance
+
+    def _set_trust(decision: DecisionEvent) -> DecisionEvent:
+        decision.trust_score = 0.85
+        return decision
+
+    trust.score.side_effect = _set_trust
+
+    pipeline = DecisionScoringPipeline(importance=importance, trust=trust, cmvk=cmvk)
+    decision = _decision()
+    pipeline.score(decision)
+
+    cmvk.verify.assert_called_once()
+    trust.score.assert_called_once()
+    assert decision.provenance.verified_by == ["v1", "v2"]
+
+
+def test_scoring_pipeline_skips_trust_on_cmvk_rejection() -> None:
+    importance = MagicMock()
+    trust = MagicMock()
+    cmvk = MagicMock()
+    cmvk.requires_verification.return_value = True
+    cmvk.verify.return_value = MagicMock(approved=False, approved_verifier_ids=[])
+
+    def _set_importance(decision: DecisionEvent) -> DecisionEvent:
+        decision.importance_score = 0.9
+        return decision
+
+    importance.score.side_effect = _set_importance
+
+    pipeline = DecisionScoringPipeline(importance=importance, trust=trust, cmvk=cmvk)
+    decision = _decision()
+    pipeline.score(decision)
+
+    trust.score.assert_not_called()
+    assert decision.status == "under_review"
+    assert write_reject_reason(decision) == "cmvk_disagreement"