dds

#!/usr/bin/env python3
"""witness.py — deterministic ash_phoenix witness extraction (tree-sitter).

The mechanical front-end for the dds witness: parses Elixir source with
tree-sitter (grammar maintained by the elixir-lang org) and emits the same
per-file fact records the LLM transcriber produces under extraction.md —
every fact quoting its declaring line, so check_join's quote-verification
runs unchanged over mechanical output.

Scope (exactly what the derivations consume):
  * Ash resources: store/table, columns, belongs_to, policies block facts,
    actions (defaults / destroy / archive / redaction), domain_registered;
  * the router: pipelines × plugs;
  * the secrets census: System.get_env / fetch_env! reads.

Class signals are JUDGMENT and are never generated here — they live in
the risk files' `## Signals` ledgers (bound to classes through each
risk's `Class:` declarations), which the instrument parses directly.
On mechanical platforms witness.json is EPHEMERAL: regenerated at
check time, not committed.

Protocol 2: the doc stamps `protocol: 2` and declares its mode
explicitly (`mode: "mechanical"`) — check_join keys currency-by-
construction on the mode field, never on generator-string sniffing.

Usage:
  witness.py --repo DIR [--out docs/registers/witness.json] [--diff]
"""

import argparse
import json
import os
import re
import sys

import tree_sitter
import tree_sitter_elixir

LANG = tree_sitter.Language(tree_sitter_elixir.language())
PARSER = tree_sitter.Parser(LANG)

COLUMN_CALLS = {
    "attribute",
    "uuid_primary_key",
    "uuid_v7_primary_key",
    "integer_primary_key",
    "create_timestamp",
    "update_timestamp",
}
REDACTION_RE = re.compile(r"redact|scrub|purge|expunge|anonymi")
ARCHIVE_COL_RE = re.compile(r"archived?_at$")

# witnessing.md §1: the extension IS the declaration — some libraries
# declare columns via extension, not attribute lines. WHICH extensions
# declare WHAT is a platform fact, carried by the platform profile
# (extension_columns), never by this extractor.
def extension_columns():
    prof = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                        "platforms", "ash_phoenix.json")
    try:
        return json.load(open(prof)).get("extension_columns") or {}
    except (OSError, json.JSONDecodeError):
        return {}


def read(path):
    with open(path, "rb") as f:
        return f.read()


def text(src, node):
    return src[node.start_byte : node.end_byte].decode("utf-8", errors="replace")


def first_line(src, node):
    return text(src, node).split("\n")[0].strip()


def calls(node, src, name=None):
    """Every `call` descendant (optionally filtered by target identifier)."""
    out = []
    stack = [node]
    while stack:
        n = stack.pop()
        if n.type == "call":
            target = n.children[0] if n.children else None
            if target is not None and target.type == "identifier":
                if name is None or text(src, target) == name:
                    out.append(n)
        stack.extend(reversed(n.children))
    return out


def call_name(src, call):
    return text(src, call.children[0]) if call.children else ""


def args_node(call):
    for c in call.children:
        if c.type == "arguments":
            return c
    return None


def do_block(call):
    for c in call.children:
        if c.type == "do_block":
            return c
    return None


def first_atom(src, call):
    a = args_node(call)
    if a is None:
        return None
    for c in a.children:
        if c.type == "atom":
            return text(src, c).lstrip(":")
        if c.type == "keywords":
            break
    return None


def defaults_list(src, node):
    """`[:read, :destroy, update: :*]` → ["read", "destroy", "update: :*"]."""
    out = []
    for lst in descendants(node, "list"):
        for c in lst.children:
            if c.type == "atom":
                out.append(text(src, c).lstrip(":"))
            elif c.type == "keywords":
                for pair in descendants(c, "pair"):
                    out.append(text(src, pair).strip())
        return out
    return out


def atoms_in(src, node):
    return [text(src, n).lstrip(":") for n in descendants(node, "atom")]


def descendants(node, typ):
    out = []
    stack = [node]
    while stack:
        n = stack.pop()
        if n.type == typ:
            out.append(n)
        stack.extend(reversed(n.children))
    return out


def module_name(src, root):
    for call in calls(root, src, "defmodule"):
        a = args_node(call)
        if a is not None:
            for c in a.children:
                if c.type == "alias":
                    return text(src, c)
    return None


def use_line(src, root):
    for call in calls(root, src, "use"):
        a = args_node(call)
        if a is not None and text(src, a).startswith("Ash.Resource"):
            return first_line(src, call)
    return None


# ── the resource record ─────────────────────────────────────────────────


def resource_record(rel, src, root, domains):
    use_q = use_line(src, root)
    if use_q is None:
        return None

    store = None
    store_quote = None
    for pg in calls(root, src, "postgres"):
        for t in calls(pg, src, "table"):
            a = args_node(t)
            if a is not None:
                strings = descendants(a, "string")
                if strings:
                    store = text(src, strings[0]).strip('"')
                    store_quote = first_line(src, t)
    if store is None:
        return None  # not a store-bearing resource (embedded/no table)

    mod = module_name(src, root)

    columns = []
    for attrs in calls(root, src, "attributes"):
        blk = do_block(attrs)
        if blk is None:
            continue
        for call in calls(blk, src):
            fn = call_name(src, call)
            if fn in COLUMN_CALLS:
                name = first_atom(src, call)
                if name:
                    columns.append({"name": name, "quote": first_line(src, call)})
            elif fn == "timestamps":
                q = first_line(src, call)
                columns.append({"name": "inserted_at", "quote": q})
                columns.append({"name": "updated_at", "quote": q})

    text_src = src.decode(errors="replace") if isinstance(src, bytes) else src
    for ext, names in extension_columns().items():
        m = re.search(rf"^.*\b{re.escape(ext)}\b.*$", text_src, re.M)
        if m:
            have = {c["name"] for c in columns}
            q = m.group(0).strip()
            columns.extend(
                {"name": name, "quote": q} for name in names if name not in have
            )

    belongs = []
    for call in calls(root, src, "belongs_to"):
        name = first_atom(src, call)
        if name:
            belongs.append({"name": name, "quote": first_line(src, call)})

    policies = {
        "present": False,
        "quote": use_q,
        "default_deny": None,
        "admit_kinds": [],
        "bypass": False,
    }
    pol_calls = [c for c in calls(root, src, "policies") if do_block(c) is not None]
    if pol_calls:
        blk = do_block(pol_calls[0])
        policies["present"] = True
        policies["quote"] = first_line(src, pol_calls[0])
        policies["bypass"] = any(do_block(b) for b in calls(blk, src, "bypass"))
        policies["default_deny"] = any(
            "always()" in text(src, f) for f in calls(blk, src, "forbid_if")
        )
        policies["admit_kinds"] = [first_line(src, a) for a in calls(blk, src, "authorize_if")]

    actions = {
        "defaults": [],
        "defaults_quote": None,
        "destroy": False,
        "archive": False,
        "redaction": [],
    }
    for acts in calls(root, src, "actions"):
        blk = do_block(acts)
        if blk is None:
            continue
        for call in calls(blk, src, "defaults"):
            actions["defaults"] = defaults_list(src, args_node(call) or call)
            actions["defaults_quote"] = first_line(src, call)
        for call in calls(blk, src):
            fn = call_name(src, call)
            name = first_atom(src, call)
            if fn == "destroy":
                actions["destroy"] = True
            if fn in ("create", "update", "action", "destroy") and name and REDACTION_RE.search(name):
                sets = []
                for sa in calls(call, src, "set_attribute"):
                    col = first_atom(src, sa)
                    if col:
                        sets.append(col)
                actions["redaction"].append(
                    {"name": name, "sets": sets, "quote": first_line(src, call)}
                )
    if "destroy" in actions["defaults"]:
        actions["destroy"] = True
    if any(ARCHIVE_COL_RE.search(c["name"]) for c in columns) or calls(root, src, "archive"):
        actions["archive"] = True

    rec = {
        "file": rel,
        "module": mod,
        "store": store,
        "store_quote": store_quote,
        "columns": columns,
        "belongs_to": belongs,
        "policies": policies,
        "actions": actions,
    }

    registered = domains.get(mod)
    rec["domain_registered"] = bool(registered)
    if registered:
        rec["domain_registered_quote"] = registered
    return rec


# ── domains: module → the `resource Mod` line in its domain file ────────


def domain_registrations(repo, files):
    out = {}
    for rel in files:
        src = read(os.path.join(repo, rel))
        if b"use Ash.Domain" not in src and b"resources do" not in src:
            continue
        root = PARSER.parse(src).root_node
        for call in calls(root, src, "resources"):
            blk = do_block(call)
            if blk is None:
                continue
            for r in calls(blk, src, "resource"):
                a = args_node(r)
                if a is None:
                    continue
                for c in a.children:
                    if c.type == "alias":
                        out[text(src, c)] = first_line(src, r)
    return out


# ── router + secrets ────────────────────────────────────────────────────


def router_record(rel, src, root):
    pipelines = []
    for call in calls(root, src, "pipeline"):
        blk = do_block(call)
        name = first_atom(src, call)
        if blk is None or not name:
            continue
        plugs = []
        for pl in calls(blk, src, "plug"):
            a = args_node(pl)
            if a is None:
                continue
            for c in a.children:
                if c.type == "atom":
                    plugs.append(text(src, c).lstrip(":"))
                    break
                if c.type == "alias":
                    plugs.append(text(src, c))
                    break
        pipelines.append({"name": name, "plugs": plugs, "quote": first_line(src, call)})
    if not pipelines:
        return None
    return {"file": rel, "router": {"pipelines": pipelines}}


GETENV_RE = re.compile(rb'System\.(?:get_env|fetch_env!?)\(\s*"([^"]+)"')
HUSH_RE = re.compile(rb'\{:hush,\s*[A-Za-z_.]+,\s*"([^"]+)"')


def secrets_census(repo, files):
    out = []
    seen = set()
    for rel in files:
        src = read(os.path.join(repo, rel))
        for regex in (GETENV_RE, HUSH_RE):
            for m in regex.finditer(src):
                name = m.group(1).decode()
                line = src[: m.start()].count(b"\n")
                quote = src.split(b"\n")[line].decode(errors="replace").strip()
                if (rel, name) not in seen:
                    seen.add((rel, name))
                    out.append({"name": name, "quote": quote, "file": rel})
    return out


# ── main ────────────────────────────────────────────────────────────────


def code_files(repo):
    out = []
    for base, dirs, names in os.walk(repo):
        rel_base = os.path.relpath(base, repo)
        if rel_base.split(os.sep)[0] in {".git", "deps", "_build", "assets", "priv", "test", "docs"}:
            dirs[:] = []
            continue
        for n in names:
            rel = os.path.normpath(os.path.join(rel_base, n))
            if re.match(r"^(lib/.*\.ex|config/runtime\.exs)$", rel.replace(os.sep, "/")):
                out.append(rel.replace(os.sep, "/"))
    return sorted(out)


def generate(repo):
    files = code_files(repo)
    domains = domain_registrations(repo, files)

    records = {}
    router = None
    for rel in files:
        src = read(os.path.join(repo, rel))
        root = PARSER.parse(src).root_node
        if rel.endswith("router.ex"):
            r = router_record(rel, src, root)
            if r:
                router = r["router"]
            continue
        rec = resource_record(rel, src, root, domains)
        if rec:
            records[rel] = rec

    secret_files = [f for f in files if f.startswith("config/")] + [
        f for f in files if f.startswith("lib/")
    ]
    return {
        "protocol": 2,
        "mode": "mechanical",
        "generator": "witness.py (tree-sitter)",
        "files": records,
        "router": router,
        "secrets": secrets_census(repo, secret_files),
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--repo", default=".")
    ap.add_argument("--out")
    ap.add_argument("--diff", action="store_true", help="compare against the committed witness")
    args = ap.parse_args()

    doc = generate(args.repo)

    committed_path = os.path.join(args.repo, "docs/registers/witness.json")

    if args.diff:
        committed = json.load(open(committed_path)) if os.path.exists(committed_path) else {}
        rc = diff(doc, committed)
        sys.exit(rc)

    out = args.out or committed_path
    json.dump(doc, open(out, "w"), indent=1, sort_keys=True)
    print(f"wrote {out}: {len(doc['files'])} files, "
          f"{sum(len(r['columns']) for r in doc['files'].values())} columns, "
          f"{len(doc.get('secrets') or [])} secrets")


def diff(doc, committed):
    rc = 0
    old_files = committed.get("files", {})
    new_files = doc["files"]
    for rel in sorted(set(old_files) | set(new_files)):
        o, n = old_files.get(rel), new_files.get(rel)
        if o is None:
            print(f"+ {rel} (new)")
            rc = 1
            continue
        if n is None:
            print(f"- {rel} (dropped)")
            rc = 1
            continue
        for key, get in [
            ("store", lambda r: r.get("store")),
            ("columns", lambda r: sorted(c["name"] for c in r.get("columns", []))),
            ("belongs_to", lambda r: sorted(b["name"] for b in r.get("belongs_to", []))),
            ("policies.present", lambda r: (r.get("policies") or {}).get("present")),
            ("actions.destroy", lambda r: (r.get("actions") or {}).get("destroy")),
            ("actions.defaults", lambda r: sorted((r.get("actions") or {}).get("defaults") or [])),
            ("domain_registered", lambda r: r.get("domain_registered")),
        ]:
            if get(o) != get(n):
                print(f"~ {rel} {key}: {get(o)} → {get(n)}")
                rc = 1
    ns = {s["name"] for s in doc.get("secrets") or []}
    os_ = {s["name"] for s in committed.get("secrets") or []}
    if ns != os_:
        print(f"~ secrets: -{sorted(os_ - ns)} +{sorted(ns - os_)}")
        rc = 1
    return rc


if __name__ == "__main__":
    main()