"""Mapping tests for the DSH bridge plugin (dcs/bridge/index.js).

Drives the real bridge module under `node -e` against FABRICATED hook
fixtures -- no live DSH, no network, no model. Every fabricated stdout
shape is copied from the corresponding real hook's emit()/deny() literal:
dcs_gate.py:274-283 (PreToolUse permissionDecision), dcs_intake.py:88-95
(UserPromptSubmit additionalContext), register_view_regen.py:64-70
(PostToolUse additionalContext). The matrix labels carry adapter x failure
mode explicitly; run_code covers both criterion-4 paths (verbatim
passthrough reaching the gate; opt-in deny firing only while an incident
is active behind the config flag).
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile
import uuid
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
BRIDGE = str(REPO / "dcs" / "bridge" / "index.js")
REAL_GATE = REPO / "dcs" / "hooks" / "dcs_gate.py"
REAL_INTAKE = REPO / "dcs" / "hooks" / "dcs_intake.py"

os.environ["PYTHONIOENCODING"] = "utf-8"

results = []


def check(name, ok, detail=""):
    results.append(ok)
    print(f"{'PASS' if ok else 'FAIL'}  {name}" + (f"\n      {detail}" if detail and not ok else ""))


# Fabricated hook bodies -- emit shapes copied verbatim from the real hooks.
DENY_GATE_BODY = (
    "import json, sys\n"
    "print(json.dumps({'hookSpecificOutput': {\n"
    "    'hookEventName': 'PreToolUse',\n"
    "    'permissionDecision': 'deny',\n"
    "    'permissionDecisionReason': 'FABRICATION: planning-phase edit denied',\n"
    "}}))\n"
)
INTAKE_CTX_BODY = (
    "import json, sys\n"
    "print(json.dumps({'hookSpecificOutput': {\n"
    "    'hookEventName': 'UserPromptSubmit',\n"
    "    'additionalContext': 'FABRICATION: intake advisory text',\n"
    "}}))\n"
)
REGEN_CTX_BODY = (
    "import json, sys\n"
    "print(json.dumps({'hookSpecificOutput': {\n"
    "    'hookEventName': 'PostToolUse',\n"
    "    'additionalContext': 'FABRICATION: register view regenerated',\n"
    "}}))\n"
)
SILENT_BODY = "import sys\nsys.exit(0)\n"
NONZERO_BODY = "import sys\nsys.stderr.write('fabricated boom')\nsys.exit(3)\n"
GARBAGE_BODY = "import sys\nsys.stdout.write('not-json-at-all')\nsys.exit(0)\n"
DUMPER_BODY = (
    "import json, sys\n"
    "open(sys.argv[0].replace('dcs_gate.py', 'dump.json'), 'w')"
    ".write(sys.stdin.read())\n"
)


def build_fixture(root, gate_body=None, intake_body=None, regen_body=None,
                  active=None, config=None):
    """<root>/proj with .dcs/hooks/<script> copies plus optional state."""
    proj = root / "proj"
    hooks = proj / ".dcs" / "hooks"
    hooks.mkdir(parents=True)
    (proj / "src").mkdir()
    (proj / "src" / "a.txt").write_text("contents", encoding="utf-8")
    bodies = {
        "dcs_gate.py": gate_body,
        "dcs_intake.py": intake_body,
        "register_view_regen.py": regen_body,
    }
    for name, body in bodies.items():
        if body is not None:
            (hooks / name).write_text(body, encoding="utf-8")
    if active is not None:
        (proj / ".dcs" / "ACTIVE").write_text(active, encoding="utf-8")
    if config is not None:
        (proj / ".dcs" / "config.json").write_text(
            json.dumps(config), encoding="utf-8")
    return proj


DRIVER_JS = r"""
// NB: under `node -e`, extra CLI args start at process.argv[1] (no script
// path slot), so bridge/config/input land on argv[1..3].
const bridge = require(process.argv[1]);
const cfg = JSON.parse(process.argv[2]);
const input = JSON.parse(process.argv[3]);
// Classify the plugin's return value against the EFFECT forms cordis
// accepts (dsh-tool-cordis fiber.ts:356-400): nullish, function, thenable,
// iterator. Anything else -- notably handing back its own merged config --
// is an INVALID effect return for a transport-only loader entry.
const classifyRet = (ret) => {
  if (ret === null || ret === undefined) return "nullish";
  if (typeof ret === "function") return "function";
  const isObject = typeof ret === "object";
  if (isObject && typeof ret.then === "function") return "thenable";
  if (isObject && (typeof ret[Symbol.iterator] === "function"
    || typeof ret[Symbol.asyncIterator] === "function")) return "iterator";
  return "INVALID";
};
const captured = {};
const ctx = { on(name, fn) { captured[name] = fn; } };
// Direct call -- `bridge.apply(...)` would hit Function.prototype.apply.
// The return value used to be discarded here; capture + classify it.
const RET = classifyRet(bridge(ctx, cfg));
(async () => {
  const handler = captured[input.seam];
  if (!handler) {
    console.log(JSON.stringify({ registered: Object.keys(captured), ret: RET }));
    return;
  }
  const NEXT = input.nextDecision !== undefined ? input.nextDecision : { kind: "__next__" };
  const next = async () => NEXT;
  let decision;
  if (input.seam === "agent/pre-step") {
    decision = await handler(input.payload, next);
  } else if (input.seam === "tools/post-execute") {
    decision = await handler(input.exec, { isError: false }, next);
  } else {
    decision = await handler(input.exec, next);
  }
  console.log(JSON.stringify({ registered: Object.keys(captured), decision, ret: RET }));
})().catch((e) => console.log(JSON.stringify(
  { error: String(e && e.message || e), ret: RET })));
"""


def drive(cfg, seam, exec_obj=None, payload=None, cwd=None, next_decision=None):
    """Run one scenario through the real bridge module under node.

    next_decision is what the fake waterfall continuation returns; it
    defaults to a sentinel that no adapter can mistake for a real DSH
    decision. Pre-step/post-execute happy paths pass realistic decisions
    ({kind:'enter', messages:[...]}/{kind:'accept'}) because the bridge
    folds ONLY into those kinds -- folding into anything else would be
    policy.
    """
    inp = {"seam": seam}
    if exec_obj is not None:
        inp["exec"] = exec_obj
    if payload is not None:
        inp["payload"] = payload
    if next_decision is not None:
        inp["nextDecision"] = next_decision
    p = subprocess.run(
        ["node", "-e", DRIVER_JS, BRIDGE, json.dumps(cfg), json.dumps(inp)],
        capture_output=True, text=True, cwd=str(cwd), timeout=180,
    )
    try:
        return json.loads(p.stdout.strip().splitlines()[-1]), p.stderr
    except Exception:
        return {"error": "driver produced no JSON"}, p.stdout + p.stderr


NEXT_SENTINEL = {"kind": "__next__"}
PY = sys.executable


def base_cfg(extra=None):
    cfg = {"python": PY}
    if extra:
        for k, v in extra.items():
            if isinstance(v, dict) and isinstance(cfg.get(k), dict):
                cfg[k].update(v)
            else:
                cfg[k] = v
    return cfg


root = Path(tempfile.mkdtemp(prefix="dcs_bridge_mapping_"))

try:
    # ---------------- happy-path translations ----------------
    proj = build_fixture(root / "f_deny", gate_body=DENY_GATE_BODY)
    out, _ = drive(base_cfg(), "tools/pre-execute",
                   {"name": "edit", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                   cwd=proj)
    d = out.get("decision", {})
    check("gate: hook deny -> {kind:'deny', reason} end-to-end",
          d.get("kind") == "deny"
          and d.get("reason") == "FABRICATION: planning-phase edit denied",
          json.dumps(out))

    proj = build_fixture(root / "f_allow", gate_body=SILENT_BODY)
    out, _ = drive(base_cfg(), "tools/pre-execute",
                   {"name": "write", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                   cwd=proj)
    check("gate: silent exit 0 -> allow via next()",
          out.get("decision") == NEXT_SENTINEL, json.dumps(out))

    proj = build_fixture(root / "f_intake", intake_body=INTAKE_CTX_BODY)
    orig = [{"id": str(uuid.uuid4()), "role": "user",
             "content": [{"type": "text", "text": "hello"}]}]
    # What a real pre-step waterfall continuation yields when the step may
    # proceed: an enter decision carrying the session's own messages.
    enter_next = {"kind": "enter", "messages": orig}
    out, _ = drive(base_cfg(), "agent/pre-step",
                   payload={"messages": orig, "agent": {"id": "sess-1"}, "step": 1},
                   cwd=proj, next_decision=enter_next)
    d = out.get("decision", {})
    msgs = d.get("messages", []) if d.get("kind") == "enter" else []
    folded = msgs[0] if len(msgs) == 2 else {}
    check("intake: additionalContext folded as prepended user message",
          d.get("kind") == "enter" and len(msgs) == 2
          and folded.get("role") == "user"
          and folded.get("source", {}).get("kind") == "plugin"
          and folded.get("content", [{}])[0].get("text") == "FABRICATION: intake advisory text"
          and msgs[1] == orig[0],
          json.dumps(out))

    proj = build_fixture(root / "f_intake_silent", intake_body=SILENT_BODY)
    out, _ = drive(base_cfg(), "agent/pre-step",
                   payload={"messages": orig, "agent": {"id": "sess-2"}, "step": 1},
                   cwd=proj, next_decision=enter_next)
    check("intake: silent hook -> plain next() passthrough (enter decision intact)",
          out.get("decision") == enter_next, json.dumps(out))

    # ------------- out-of-tree node cwd: root comes from the payload --------
    # :347 roots the intake adapter from the pre-step payload's session
    # header, keeping process.cwd() as last-resort only:
    #   const hc = payload?.agent?.session?.header?.cwd;
    #   resolveProjectRoot(typeof hc === "string" && path.isAbsolute(hc)
    #                      ? path.join(hc, "package.json") : null,
    #                      process.cwd())
    # resolveProjectRoot starts at an absolute target's DIRNAME and walks
    # up (index.js:127-143, the gate's target-path semantics; IC-resolved
    # in 204-TASKING/S1.md:18), so the header cwd below points INSIDE the
    # onboarded project -- the shape a DSH session carries when launched
    # from a project subdirectory. Both fixtures keep node's cwd at the
    # mkdtemp root: outside any .dcs/ ancestor, so process.cwd()
    # starvation is total and only the payload header can root the
    # adapter. State A goes red against the pre-fix bridge by
    # construction (root=null -> no-hook -> no canary, no fold); state B
    # pins the silent non-spawn contract that must survive the fix.
    CANARY_INTAKE_BODY = (
        "import json, os, sys\n"
        "_here = os.path.dirname(os.path.abspath(sys.argv[0]))\n"
        "open(os.path.join(_here, 'intake_spawned.canary'), 'w').write('spawned')\n"
        "print(json.dumps({'hookSpecificOutput': {\n"
        "    'hookEventName': 'UserPromptSubmit',\n"
        "    'additionalContext': 'FABRICATION: out-of-tree intake advisory',\n"
        "}}))\n"
    )
    proj = build_fixture(root / "f_oot_intake", intake_body=CANARY_INTAKE_BODY)
    canary = proj / ".dcs" / "hooks" / "intake_spawned.canary"
    out, _ = drive(base_cfg(), "agent/pre-step",
                   payload={"messages": orig,
                            "agent": {"id": "sess-oot",
                                      "session": {"header": {"cwd": str(proj / "src")}}},
                            "step": 1},
                   cwd=root, next_decision=enter_next)
    d = out.get("decision", {})
    msgs = d.get("messages", []) if d.get("kind") == "enter" else []
    folded = msgs[0] if len(msgs) == 2 else {}
    check("intake: out-of-tree node cwd + absolute session-header cwd -> "
          "hook SPAWNS (canary witness) + additionalContext folds prepended",
          canary.exists() and d.get("kind") == "enter" and len(msgs) == 2
          and folded.get("role") == "user"
          and folded.get("source", {}).get("kind") == "plugin"
          and folded.get("content", [{}])[0].get("text") == "FABRICATION: out-of-tree intake advisory"
          and msgs[1] == orig[0],
          f"canary_exists={canary.exists()} out={json.dumps(out)}")

    proj = build_fixture(root / "f_oot_no_header", intake_body=CANARY_INTAKE_BODY)
    canary = proj / ".dcs" / "hooks" / "intake_spawned.canary"
    out, err = drive(base_cfg(), "agent/pre-step",
                     payload={"messages": orig, "agent": {"id": "sess-nohdr"}, "step": 1},
                     cwd=root)
    check("intake negative twin: out-of-tree node cwd + NO session header -> "
          "untouched next() passthrough, canary ABSENT, stderr silent",
          out.get("decision") == NEXT_SENTINEL and not canary.exists()
          and "[dcs-bridge]" not in err,
          f"canary_exists={canary.exists()} out={json.dumps(out)} stderr={err[:200]}")

    # State C -- ROOT-LAUNCH: the header cwd EQUALS the onboarded root
    # itself, the canonical field-evidence shape (a DSH session launched
    # exactly AT an onboarded project root, e.g. C:\dsh-scratch). This is
    # what the amended :347-348 literal buys: wrapping the header in
    # path.join(hc, "package.json") puts dirname() ON the pointed-at
    # directory instead of skipping past it. The first-pass one-line form
    # `resolveProjectRoot(hc ?? null, process.cwd())` reds here by
    # construction -- at root launch dirname() starts at the PARENT (no
    # .dcs/ above the mkdtemp root), so no hook, no canary, no fold --
    # while state A still greens under both forms; C is the discriminator.
    proj = build_fixture(root / "f_root_launch_intake", intake_body=CANARY_INTAKE_BODY)
    canary = proj / ".dcs" / "hooks" / "intake_spawned.canary"
    out, _ = drive(base_cfg(), "agent/pre-step",
                   payload={"messages": orig,
                            "agent": {"id": "sess-root-launch",
                                      "session": {"header": {"cwd": str(proj)}}},
                            "step": 1},
                   cwd=root, next_decision=enter_next)
    d = out.get("decision", {})
    msgs = d.get("messages", []) if d.get("kind") == "enter" else []
    folded = msgs[0] if len(msgs) == 2 else {}
    check("intake ROOT-LAUNCH: session-header cwd EQUAL to fixture project "
          "root -> hook SPAWNS rooted at that dir (canary witness) + "
          "additionalContext folds prepended",
          canary.exists() and d.get("kind") == "enter" and len(msgs) == 2
          and folded.get("role") == "user"
          and folded.get("source", {}).get("kind") == "plugin"
          and folded.get("content", [{}])[0].get("text") == "FABRICATION: out-of-tree intake advisory"
          and msgs[1] == orig[0],
          f"canary_exists={canary.exists()} out={json.dumps(out)}")

    # Once-per-session dedup stays the HOOK'S job: real intake, same session.
    proj = build_fixture(root / "f_intake_real")
    shutil.copy2(REAL_INTAKE, proj / ".dcs" / "hooks" / "dcs_intake.py")
    sid = uuid.uuid4().hex
    first, _ = drive(base_cfg(), "agent/pre-step",
                     payload={"messages": orig, "agent": {"id": sid}, "step": 1}, cwd=proj,
                     next_decision=enter_next)
    second, _ = drive(base_cfg(), "agent/pre-step",
                      payload={"messages": orig, "agent": {"id": sid}, "step": 1}, cwd=proj,
                      next_decision=enter_next)
    f1 = first.get("decision", {})
    ok_first = (f1.get("kind") == "enter" and len(f1.get("messages", [])) == 2
                and "[DCS]" in f1["messages"][0]["content"][0]["text"])
    check("intake: REAL hook folds on FIRST prompt of a session", ok_first,
          json.dumps(first))
    check("intake: REAL hook dedups second same-session prompt in the HOOK "
          "(no JS dedup, plain passthrough)",
          second.get("decision") == enter_next, json.dumps(second))

    proj = build_fixture(root / "f_regen", regen_body=REGEN_CTX_BODY)
    out, _ = drive(base_cfg(), "tools/post-execute",
                   {"name": "write", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                   cwd=proj, next_decision={"kind": "accept"})
    d = out.get("decision", {})
    contexts = d.get("additionalContexts", []) if d.get("kind") == "accept" else []
    last = contexts[-1] if contexts else {}
    check("regen: additionalContexts surfaced via accept (never block)",
          d.get("kind") == "accept" and len(contexts) == 1
          and last.get("role") == "user"
          and last.get("content", [{}])[0].get("text") == "FABRICATION: register view regenerated",
          json.dumps(out))

    proj = build_fixture(root / "f_regen_silent", regen_body=SILENT_BODY)
    out, _ = drive(base_cfg(), "tools/post-execute",
                   {"name": "write", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                   cwd=proj, next_decision={"kind": "accept"})
    check("regen: silent hook -> plain accept passthrough (no contexts added)",
          out.get("decision") == {"kind": "accept"}, json.dumps(out))

    # ---------------- fail-open matrix: adapter x failure mode ----------------
    modes = [
        ("spawn failure", {"python": "dcs-bridge-no-such-python-xyz"},
         SILENT_BODY, "spawn failed"),
        ("nonzero exit", None, NONZERO_BODY, "nonzero"),
        ("unparseable stdout", None, GARBAGE_BODY, "unparseable"),
    ]
    adapters = [
        ("gate", "tools/pre-execute"),
        ("intake", "agent/pre-step"),
        ("regen", "tools/post-execute"),
    ]
    for mode_name, cfg_override, body, needle in modes:
        for adapter_id, seam in adapters:
            kw = {}
            kw["gate_body" if adapter_id == "gate" else
               "intake_body" if adapter_id == "intake" else "regen_body"] = body
            proj = build_fixture(root / f"f_{adapter_id}_{mode_name.replace(' ', '_')}", **kw)
            cfg = base_cfg(cfg_override)
            if seam == "agent/pre-step":
                out, err = drive(cfg, seam,
                                 payload={"messages": orig, "agent": {"id": uuid.uuid4().hex}, "step": 1},
                                 cwd=proj)
            else:
                out, err = drive(cfg, seam,
                                 {"name": "write", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                                 cwd=proj)
            expect = NEXT_SENTINEL
            ok = out.get("decision") == expect and needle in err and "[dcs-bridge]" in err \
                and f"[dcs-bridge] {adapter_id}" in err
            check(f"fail-open [{adapter_id}] x [{mode_name}] -> pass-through + loud log",
                  ok, f"out={json.dumps(out)}\n      stderr={err[:400]}")

    # ---------------- config toggles disable each adapter independently ------
    for adapter_id, seam in adapters:
        cfg = base_cfg({adapter_id: {"enabled": False}})
        proj = build_fixture(root / f"f_off_{adapter_id}", gate_body=DENY_GATE_BODY,
                             intake_body=INTAKE_CTX_BODY, regen_body=REGEN_CTX_BODY)
        if seam == "agent/pre-step":
            out, err = drive(cfg, seam,
                             payload={"messages": orig, "agent": {"id": uuid.uuid4().hex}, "step": 1},
                             cwd=proj)
        else:
            out, err = drive(cfg, seam,
                             {"name": "edit", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                             cwd=proj)
        quiet_skip = out.get("decision") == NEXT_SENTINEL and not out.get("registered") \
            or adapter_id not in (out.get("registered") or [])
        handler_absent = seam not in (out.get("registered") or [])
        no_loud_skip = "NOT a verified" not in err
        check(f"toggle: {adapter_id}.enabled=false disables ONLY its adapter (quiet)",
              handler_absent and no_loud_skip, json.dumps(out) + "\n      " + err[:200])

    out, _ = drive(base_cfg({"intake": {"enabled": False}}), "tools/pre-execute",
                   {"name": "read", "arguments": {"file_path": "x.txt"}}, cwd=root)
    check("toggle: disabling intake leaves gate registered",
          "tools/pre-execute" in out.get("registered", []), json.dumps(out))
    out, _ = drive(base_cfg({"gate": {"enabled": False}}), "agent/pre-step",
                   payload={"messages": orig, "agent": {"id": "s"}, "step": 1}, cwd=root)
    check("toggle: disabling gate leaves intake registered",
          "agent/pre-step" in out.get("registered", []), json.dumps(out))

    # ---------------- D5 feature detection: unknown seam skipped loudly -----
    proj = build_fixture(root / "f_unknown_seam", gate_body=DENY_GATE_BODY)
    out, err = drive(base_cfg({"gate": {"seam": "tools/pre-executed"}}),
                     "tools/pre-execute",
                     {"name": "edit", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                     cwd=proj)
    check("feature-detect: unknown seam override -> adapter skipped LOUDLY, not bound",
          "tools/pre-execute" not in out.get("registered", [])
          and "NOT a verified" in err and "[dcs-bridge] gate" in err,
          json.dumps(out) + "\n      " + err[:300])

    # ---------------- tool-name mapping units -------------------------------
    out, _ = drive({}, "tools/pre-execute", {"name": "probe", "arguments": {}}, cwd=root)
    js_probe = subprocess.run(
        ["node", "-e",
         "const b=require(process.argv[1]);"
         "const m=b.mapToolName;"
         "console.log(JSON.stringify([m('write'),m('edit'),m('read'),"
         "m('send_message'),m('notebook_edit'),m('run_code'),m('bash')]));",
         BRIDGE],
        capture_output=True, text=True, timeout=60)
    got = json.loads(js_probe.stdout.strip())
    check("tool-name mapping: write/edit/read/send_message/notebook_edit mapped, "
          "run_code+bash pass through",
          got == ["Write", "Edit", "Read", "SendMessage", "NotebookEdit",
                  "run_code", "bash"], js_probe.stdout + js_probe.stderr)

    # ---------------- run_code: criterion 4 both paths ----------------------
    proj = build_fixture(root / "f_rc_dump", gate_body=DUMPER_BODY)
    out, _ = drive(base_cfg(), "tools/pre-execute",
                   {"name": "run_code",
                    "arguments": {"code": "print(1)", "description": "sum thing"}},
                   cwd=proj)
    dump_file = proj / ".dcs" / "hooks" / "dump.json"
    dumped = json.loads(dump_file.read_text(encoding="utf-8")) if dump_file.exists() else {}
    check("run_code passthrough reaches the gate VERBATIM: name unmapped, "
          "{code, description} carried as tool_input",
          out.get("decision") == NEXT_SENTINEL
          and dumped.get("tool_name") == "run_code"
          and dumped.get("tool_input") == {"code": "print(1)", "description": "sum thing"},
          f"out={json.dumps(out)} dump={json.dumps(dumped)}")

    rc_args = {"name": "run_code",
               "arguments": {"code": "require('fs')", "description": "arbitrary code"}}
    proj = build_fixture(root / "f_rc_deny", active="2026-08-23-dsh-port|1|execution",
                         config={"bridge": {"run_code_deny_while_active": True}})
    shutil.copy2(REAL_GATE, proj / ".dcs" / "hooks" / "dcs_gate.py")
    out, _ = drive(base_cfg(), "tools/pre-execute", rc_args, cwd=proj)
    d = out.get("decision", {})
    check("run_code opt-in deny FIRES while incident active + flag set (real gate)",
          d.get("kind") == "deny" and "run_code denied" in d.get("reason", "")
          and "bridge.run_code_deny_while_active" in d.get("reason", ""),
          json.dumps(out))

    proj = build_fixture(root / "f_rc_noflag", active="2026-08-23-dsh-port|1|execution")
    shutil.copy2(REAL_GATE, proj / ".dcs" / "hooks" / "dcs_gate.py")
    out, _ = drive(base_cfg(), "tools/pre-execute", rc_args, cwd=proj)
    check("run_code default OFF: absent config never denies (inspect-only)",
          out.get("decision") == NEXT_SENTINEL, json.dumps(out))

    proj = build_fixture(root / "f_rc_noactive",
                         config={"bridge": {"run_code_deny_while_active": True}})
    shutil.copy2(REAL_GATE, proj / ".dcs" / "hooks" / "dcs_gate.py")
    out, _ = drive(base_cfg(), "tools/pre-execute", rc_args, cwd=proj)
    check("run_code opt-in set but NO active incident -> allowed",
          out.get("decision") == NEXT_SENTINEL, json.dumps(out))

    # ---------------- real-gate end-to-end deny through DSH shapes ----------
    proj = build_fixture(root / "f_real_planning",
                         active="2026-08-23-dsh-port|1|planning")
    shutil.copy2(REAL_GATE, proj / ".dcs" / "hooks" / "dcs_gate.py")
    out, _ = drive(base_cfg(), "tools/pre-execute",
                   {"name": "edit", "arguments": {"file_path": str(proj / "src" / "a.txt")}},
                   cwd=proj)
    d = out.get("decision", {})
    check("real gate end-to-end: planning-phase guarded edit translated to "
          "{kind:'deny'} with the gate's reason body",
          d.get("kind") == "deny" and d.get("reason", "").startswith("DCS gate")
          and "no approved" in d.get("reason", ""),
          json.dumps(out))

    # ---------------- C3 return-shape regression ---------------------------
    # Cordis accepts only four EFFECT return forms from a plugin's main
    # callback (dsh-tool-cordis fiber.ts:356-400): nullish, function,
    # thenable, iterator. Any other object -- such as the bridge handing its
    # own merged config back -- is INVALID. The driver now reports a
    # classified `ret` key; the pre-fix bridge fails both checks below.
    LEGAL_EFFECT_RETURNS = ("nullish", "function", "thenable", "iterator")
    out, _ = drive(base_cfg(), "tools/pre-execute",
                   {"name": "probe", "arguments": {}}, cwd=root)
    check("return shape: bridge(ctx, cfg) returns a legal cordis EFFECT form "
          "(nullish/function/thenable/iterator), never a bare object",
          out.get("ret") in LEGAL_EFFECT_RETURNS,
          f"observed ret={json.dumps(out.get('ret'))} full={json.dumps(out)}")

    src = Path(BRIDGE).read_text(encoding="utf-8")
    apply_span = src[src.index("const apply"):src.index("module.exports")]
    check("source guard: 'const apply' -> 'module.exports' span carries no "
          "'return cfg;' (loader entry must not hand config back)",
          "return cfg;" not in apply_span,
          "found 'return cfg;' inside the apply..module.exports span")

finally:
    shutil.rmtree(root, ignore_errors=True)

failed = sum(1 for r in results if not r)
print(f"\n{len(results) - failed}/{len(results)} passed")
sys.exit(1 if failed else 0)