v2/skills/discipline_gate.py

"""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, scan_destructive_anywhere
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._check_insecure_code(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,
                )

        if not any(f["severity"] == "BLOCK" for f in verdict.flags):
            prose_hits = scan_destructive_anywhere(draft)
            if prose_hits:
                labels = sorted(set(label for _, label in prose_hits))
                verdict.add(
                    Severity.BLOCK,
                    f"Draft mentions destructive SQL outside code blocks "
                    f"({', '.join(labels)}). Even in prose or inline code, "
                    f"destructive SQL gets copy-pasted. Rephrase without "
                    f"the destructive statement.",
                    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"])

    _INSECURE_PATTERNS = [
        (r"""\$\{[^}]*(?:req|params|query|body|input|user)[^}]*\}""",
         "string interpolation with user input (SQL/NoSQL injection risk — CWE-89)"),
        (r"""\beval\s*\([^)]*(?:req|params|query|body|input|user)""",
         "eval() with user-controlled input (code injection — CWE-94)"),
        (r"""['"`]\s*\+\s*(?:req|params|query|body|input|user)""",
         "string concatenation with user input in query context (CWE-89)"),
        (r"""\bshell\s*[=:]\s*True""",
         "subprocess with shell=True (OS command injection — CWE-78)"),
        (r"""\bchild_process\.exec\s*\(""",
         "child_process.exec (OS command injection — CWE-78). Use execFile or spawn instead"),
        (r"""\bos\.system\s*\(""",
         "os.system() (OS command injection — CWE-78). Use subprocess with shell=False"),
        (r"""\bopen\s*\([^)]*(?:filename|filepath|file_path|path|fname)\b[^)]*\)(?!.*(?:sanitize|validate|whitelist|allowlist|os\.path\.basename|realpath))""",
         "file open with unsanitized path (path traversal — CWE-22). Validate against a base directory"),
        (r"""\bpickle\.loads?\s*\(""",
         "pickle.load on untrusted data (deserialization — CWE-502). Use JSON or a safe format"),
        (r"""\byaml\.load\s*\([^)]*\)(?!.*Loader)""",
         "yaml.load without SafeLoader (code execution — CWE-502). Use yaml.safe_load"),
        (r"""\bMath\.random\s*\(\s*\).*(?:token|secret|key|password|auth|session|nonce|salt|iv)""",
         "Math.random() for security-sensitive value (weak PRNG — CWE-338). Use crypto.randomBytes"),
        (r"""(?:md5|sha1)\s*\(.*(?:password|secret|credential)""",
         "weak hash for credentials (CWE-328). Use bcrypt, scrypt, or argon2"),
    ]

    def _check_insecure_code(self, verdict: GateVerdict, draft: str) -> None:
        fenced = re.findall(r"```(?:\w*)\n(.*?)```", draft, re.DOTALL)
        if not fenced:
            return
        code = "\n".join(fenced)
        for pattern, label in self._INSECURE_PATTERNS:
            if re.search(pattern, code, re.IGNORECASE):
                verdict.add(
                    Severity.WARN,
                    f"Code contains {label}. Use safe alternatives.",
                )

    # ------------------------------------------------------------------
    # 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
            ),
        )