Clean up nested duplicate: v2/v2/skills/discipline_gate.py

v2/v2/skills/discipline_gate.py

@@ -1,280 +0,0 @@
-"""Discipline gate — a Kintsugi skill, not a regex filter.
-
-Last node of every plan. Nothing reaches the user without passing here.
-
-Three layers, in order of authority:
-
-  1. BELIEF CHECKS — the gate holds beliefs (from the BDI store) about
-     the shared-db constraint, the auth architecture, and the audit
-     findings. The draft is checked against each. A belief with
-     confidence 1.0 (operator constraint) produces a BLOCK on violation;
-     lower-confidence beliefs produce WARNs citing their source.
-  2. PATTERN CHECKS — the code-level regex layer (destructive SQL,
-     signature bypass shapes, transaction gaps). Code, so it cannot be
-     argued with.
-  3. CONFIDENCE GRADING — the gate grades how much of the answer is
-     actually evidenced: source files read this session → HIGH; packs/
-     architecture map only → MEDIUM; neither → LOW. Failed verification
-     caps at LOW; no verification of generated code caps at MEDIUM.
-
-The gate's output is the final artifact: the (possibly blocked) answer
-plus machine-readable flags for the API response.
-"""
-
-import re
-from dataclasses import dataclass, field
-from enum import Enum
-
-from kintsugi_core import (
-    BaseSkillChip,
-    BDIStore,
-    EFEWeights,
-    SkillContext,
-    SkillDomain,
-    SkillRequest,
-    SkillResponse,
-)
-from skills.migration_safety import classify_sql, extract_sql
-from skills.security_review import (
-    findings_relevant_to,
-    recurring_pattern_hits,
-    touches_auth,
-)
-
-
-class Confidence(str, Enum):
-    HIGH = "HIGH"      # claims backed by source read this session, verified
-    MEDIUM = "MEDIUM"  # backed by architecture map / audit / packs
-    LOW = "LOW"        # reasoning from architecture alone, or verification failed
-
-
-class Severity(str, Enum):
-    BLOCK = "BLOCK"
-    WARN = "WARN"
-    INFO = "INFO"
-
-
-@dataclass
-class GateVerdict:
-    passed: bool = True
-    confidence: Confidence = Confidence.LOW
-    flags: list = field(default_factory=list)
-    requires_review: list = field(default_factory=list)
-    beliefs_consulted: list = field(default_factory=list)
-
-    def add(self, severity: Severity, message: str, belief_id: str = ""):
-        self.flags.append({
-            "severity": severity.value, "message": message,
-            "belief": belief_id,
-        })
-        if severity == Severity.BLOCK:
-            self.passed = False
-
-    def format_annotations(self) -> str:
-        lines = []
-        if self.flags:
-            lines.append("**Discipline Gate:**")
-            icons = {"BLOCK": "[BLOCK]", "WARN": "[WARN]", "INFO": "[info]"}
-            for f in self.flags:
-                src = f" (belief: {f['belief']})" if f["belief"] else ""
-                lines.append(f"- {icons[f['severity']]} {f['message']}{src}")
-        if self.requires_review:
-            lines.append(f"- Review required from: "
-                         f"{', '.join(sorted(set(self.requires_review)))}")
-        lines.append(f"- Confidence: **{self.confidence.value}**")
-        return "\n".join(lines)
-
-
-class DisciplineGateChip(BaseSkillChip):
-    name = "discipline_gate"
-    description = "Belief-checked final gate with earned confidence"
-    version = "2.0.0"
-    domain = SkillDomain.SECURITY
-    efe_weights = EFEWeights(
-        mission_alignment=0.20, stakeholder_benefit=0.30,
-        resource_efficiency=0.05, transparency=0.30, equity=0.15,
-    )
-    consensus_actions = ["release_blocked_answer"]
-
-    def __init__(self, bdi: BDIStore):
-        super().__init__()
-        self.bdi = bdi
-
-    async def handle(self, request: SkillRequest,
-                     context: SkillContext) -> SkillResponse:
-        question = context.metadata.get("question", request.raw_input)
-        session = context.metadata.get("session")
-
-        draft_artifact = request.parameters.get("draft") or {}
-        draft = draft_artifact.get("text", "") if isinstance(draft_artifact, dict) else str(draft_artifact)
-        analysis = request.parameters.get("analysis") or {}
-        migration_report = request.parameters.get("migration_report") or {}
-        security_report = request.parameters.get("security_report") or {}
-        verification = request.parameters.get("verification") or {}
-
-        verdict = GateVerdict()
-
-        if not draft:
-            verdict.add(Severity.BLOCK, "synthesis produced no draft")
-            return self._respond(verdict, draft, question)
-
-        self._check_shared_db(verdict, draft)
-        self._check_auth(verdict, draft, question, security_report)
-        self._check_audit_patterns(verdict, draft)
-        self._grade_confidence(
-            verdict, draft, session, analysis, verification, draft_artifact,
-        )
-
-        return self._respond(verdict, draft, question)
-
-    # ------------------------------------------------------------------
-    # Layer 1+2: belief-backed checks (patterns cited to beliefs)
-    # ------------------------------------------------------------------
-
-    def _check_shared_db(self, verdict: GateVerdict, draft: str) -> None:
-        belief = self.bdi.get_belief("belief_constraint_shared_db")
-        belief_id = belief.id if belief else ""
-        if belief:
-            verdict.beliefs_consulted.append(belief.id)
-
-        for sql in extract_sql(draft):
-            klass = classify_sql(sql)
-            if klass["destructive"]:
-                severity = (Severity.BLOCK
-                            if belief and belief.confidence >= 1.0
-                            else Severity.WARN)
-                verdict.add(
-                    severity,
-                    f"Draft contains destructive SQL "
-                    f"({', '.join(klass['violations'])}). "
-                    f"{belief.content if belief else 'Shared-db rule.'}",
-                    belief_id,
-                )
-
-    def _check_auth(self, verdict: GateVerdict, draft: str, question: str,
-                    security_report: dict) -> None:
-        auth_hit = (touches_auth(draft) or touches_auth(question)
-                    or security_report.get("auth_touched", False))
-        if not auth_hit:
-            return
-        belief = self.bdi.get_belief("belief_arch_auth_flow")
-        if belief:
-            verdict.beliefs_consulted.append(belief.id)
-        verdict.add(
-            Severity.WARN,
-            "This touches authentication. Review with security before "
-            "merging.",
-            belief.id if belief else "",
-        )
-        verdict.requires_review.append("security")
-
-    def _check_audit_patterns(self, verdict: GateVerdict, draft: str) -> None:
-        for hit in findings_relevant_to(draft):
-            if not hit["pattern_matched"]:
-                continue
-            belief_id = f"belief_audit_{hit['id'].lower()}"
-            belief = self.bdi.get_belief(belief_id)
-            if belief:
-                verdict.beliefs_consulted.append(belief_id)
-            severity = (Severity.WARN
-                        if hit["severity"] in ("CRITICAL", "HIGH")
-                        else Severity.INFO)
-            verdict.add(
-                severity,
-                f"Draft walks into audit finding {hit['id']} "
-                f"({hit['title']}). {hit['advice']}",
-                belief_id,
-            )
-        for hit in recurring_pattern_hits(draft):
-            verdict.add(Severity.INFO, hit["message"])
-
-    # ------------------------------------------------------------------
-    # Layer 3: earned confidence
-    # ------------------------------------------------------------------
-
-    def _grade_confidence(self, verdict: GateVerdict, draft: str, session,
-                          analysis: dict, verification: dict,
-                          draft_artifact: dict) -> None:
-        files_read = set(session.files_read) if session else set()
-        files_cited = set(re.findall(
-            r"\b((?:server|client|shared|design-system)/[\w./-]+\.\w+)\b",
-            draft,
-        ))
-
-        has_source = bool(files_read)
-        cites_unread = files_cited - files_read
-        has_knowledge = bool(
-            draft_artifact.get("packs_used")
-            or analysis.get("repo_available") is False
-            or self.bdi.list_beliefs()
-        )
-
-        checks = verification.get("checks", []) if verification else []
-        ran = [c for c in checks if c.get("available")]
-        failed = [c for c in ran if not c.get("passed")]
-        has_code = bool(re.search(r"```\w*\n.{10,}?```", draft, re.DOTALL))
-
-        if failed:
-            confidence = Confidence.LOW
-            verdict.add(
-                Severity.WARN,
-                f"Generated code FAILED verification "
-                f"({len(failed)}/{len(ran)} checks): "
-                f"{failed[0]['output'][:200]}",
-            )
-        elif has_source and (not has_code or ran):
-            confidence = Confidence.HIGH
-        elif has_source or has_knowledge:
-            confidence = Confidence.MEDIUM
-            if has_code and not ran:
-                verdict.add(
-                    Severity.INFO,
-                    "Code in this answer was not verified (tsc/tests "
-                    "unavailable here) — treat as unchecked.",
-                )
-        else:
-            confidence = Confidence.LOW
-
-        if cites_unread:
-            if confidence == Confidence.HIGH:
-                confidence = Confidence.MEDIUM
-            verdict.add(
-                Severity.INFO,
-                f"Answer references files not read this session: "
-                f"{', '.join(sorted(cites_unread)[:5])} — reasoning from "
-                f"architecture for those.",
-            )
-
-        verdict.confidence = confidence
-
-    # ------------------------------------------------------------------
-
-    def _respond(self, verdict: GateVerdict, draft: str,
-                 question: str) -> SkillResponse:
-        annotations = verdict.format_annotations()
-        if verdict.passed:
-            final_text = f"{draft}\n\n---\n{annotations}"
-        else:
-            final_text = (
-                "**BLOCKED by Discipline Gate.**\n\n"
-                "The generated answer violated a hard constraint and was "
-                "withheld.\n\n" + annotations +
-                "\n\nAsk for the safe alternative (e.g. an additive "
-                "multi-step migration) and Rivet will produce it."
-            )
-        return SkillResponse(
-            content=final_text,
-            success=True,
-            data={
-                "text": final_text,
-                "passed": verdict.passed,
-                "confidence": verdict.confidence.value,
-                "flags": verdict.flags,
-                "requires_review": sorted(set(verdict.requires_review)),
-                "beliefs_consulted": sorted(set(verdict.beliefs_consulted)),
-            },
-            requires_consensus=not verdict.passed,
-            consensus_action=(
-                "release_blocked_answer" if not verdict.passed else None
-            ),
-        )