#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Codex adapter for the CASAN agentic bridge (Plan-20, Wave 3). Thin renderer that maps Codex lifecycle hook payloads (on stdin) into the client-agnostic bridge contract and renders the response back. It reuses the core bridge module so the bridge never learns Codex's JSON shape. Codex project hooks load only AFTER a trust review (Spike-20 §4.2) — the devkit `doctor` surfaces the trust/onboarding state so this is never hidden from a member. Codex tool hooks are a guardrail, not a complete boundary: a turn that uses a hosted/specialized tool outside hook coverage is DOWNGRADED, not silently certified. Output contract (kept portable across Codex versions, which is why the exact field mapping is a Wave-3 experiment): a decision JSON on stdout AND an exit code — 0 = allow, 2 = block/deny — so a host that reads either signal fails closed the same way. Field names are accepted defensively (tool_name|tool, cwd|project, session_id|session) to absorb payload drift. """ from __future__ import annotations import argparse import hashlib import json import os import sys _ADAPTER_DIR = os.path.dirname(os.path.abspath(__file__)) _BRIDGE_DIR = os.path.abspath(os.path.join(_ADAPTER_DIR, "..", "..", "scripts", "python")) if _BRIDGE_DIR not in sys.path: sys.path.insert(0, _BRIDGE_DIR) import agentic_bridge as bridge # noqa: E402 ADAPTER_VERSION = "20.1.0-codex" def integration_mode(): # Managed/MDM deployments set this to managed_hook; default project_hook. return os.environ.get("CASAN_AGENTIC_INTEGRATION_MODE", "project_hook") def _first(payload, *keys): for k in keys: if payload.get(k) is not None: return payload.get(k) return None def _ptr_path(session): h = hashlib.sha256(("codex:" + (session or "nosession")).encode("utf-8")).hexdigest()[:24] return os.path.join(bridge.sessions_dir(), "ptr-%s.json" % h) def store_pointer(session, admission_id, trace_id): bridge.atomic_write_json(_ptr_path(session), { "admission_id": admission_id, "trace_id": trace_id}) def load_pointer(session): path = _ptr_path(session) if not os.path.exists(path): return None try: with open(path, "r", encoding="utf-8") as fh: return json.load(fh) except (OSError, ValueError): return None def clear_pointer(session): try: os.unlink(_ptr_path(session)) except OSError: pass def _emit(obj, exit_code): sys.stdout.write(json.dumps(obj, ensure_ascii=False)) sys.stdout.write("\n") return exit_code def handle_user_prompt_submit(payload): session = _first(payload, "session_id", "session", "conversation_id") resp = bridge.op_begin({ "op": "begin", "client": "codex", "client_version": _first(payload, "client_version", "codex_version"), "adapter_version": ADAPTER_VERSION, "project": _first(payload, "cwd", "project", "project_dir", "workspace"), "session": session, "turn": _first(payload, "turn_id", "message_id"), "prompt": _first(payload, "prompt", "input", "message") or "", "integration_mode": integration_mode(), }) if resp.get("admission_id"): store_pointer(session, resp["admission_id"], resp.get("trace_id")) if resp.get("decision") == "block": return _emit({"decision": "block", "reason": resp.get("reason"), "trace_id": resp.get("trace_id")}, 2) return _emit({"decision": "allow", "context": resp.get("context"), "certification_strength": resp.get("certification_strength"), "warnings": resp.get("warnings", [])}, 0) def handle_pre_tool_use(payload): session = _first(payload, "session_id", "session", "conversation_id") ptr = load_pointer(session) resp = bridge.op_pre_tool({ "op": "pre-tool", "admission_id": (ptr or {}).get("admission_id") or "", "tool": _first(payload, "tool_name", "tool", "name") or "", "tool_input": _first(payload, "tool_input", "input", "arguments", "args"), "project": _first(payload, "cwd", "project", "workspace"), }) if resp.get("decision") == "allow": return _emit({"decision": "allow", "reason": resp.get("reason")}, 0) return _emit({"decision": "deny", "reason": resp.get("reason"), "trace_id": resp.get("trace_id")}, 2) def handle_post_tool_use(payload): session = _first(payload, "session_id", "session", "conversation_id") ptr = load_pointer(session) if not ptr: return _emit({"decision": "allow", "reason": "no_admission"}, 0) status = _first(payload, "status") or ("error" if _first(payload, "error", "is_error") else "success") bridge.op_post_tool({ "op": "post-tool", "admission_id": ptr.get("admission_id"), "tool": _first(payload, "tool_name", "tool", "name") or "", "status": status, "duration_ms": _first(payload, "duration_ms", "elapsed_ms"), "result": None, }) # Opportunistic usage/cost capture — only when Codex actually supplies a # source. Absent a stable source, we record nothing (bridge keeps it null + # partial rather than inventing a number). See Spike-20 X6. usage = _first(payload, "usage") if isinstance(usage, dict) and usage.get("source"): bridge.op_telemetry({ "op": "telemetry", "admission_id": ptr.get("admission_id"), "model": _first(payload, "model"), "input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens"), "cost_amount": usage.get("cost_amount"), "cost_currency": usage.get("cost_currency"), "cost_source": usage.get("source"), }) return _emit({"decision": "allow"}, 0) def handle_stop(payload): if payload.get("stop_hook_active"): return _emit({"decision": "allow"}, 0) session = _first(payload, "session_id", "session", "conversation_id") ptr = load_pointer(session) if not ptr: return _emit({"decision": "allow", "reason": "no_admission"}, 0) resp = bridge.op_finalize({ "op": "finalize", "admission_id": ptr.get("admission_id"), "stop_reason": _first(payload, "stop_reason") or "completed", "assistant_summary": _first(payload, "last_assistant_message", "assistant_summary"), }) clear_pointer(session) return _emit({"decision": "allow", "certified": resp.get("decision") == "certified", "certification_strength": resp.get("certification_strength")}, 0) HANDLERS = { "UserPromptSubmit": handle_user_prompt_submit, "PreToolUse": handle_pre_tool_use, "PostToolUse": handle_post_tool_use, "Stop": handle_stop, } def main(argv=None): parser = argparse.ArgumentParser(description="CASAN Codex hook adapter") parser.add_argument("--event", help="Codex hook event name (else read from payload)") args = parser.parse_args(argv) try: payload = json.loads(sys.stdin.read() or "{}") except ValueError: payload = {} event = args.event or _first(payload, "hook_event_name", "event", "type") handler = HANDLERS.get(event) if handler is None: return _emit({"decision": "allow", "reason": "unknown_event"}, 0) try: return handler(payload) except Exception as exc: # noqa: BLE001 if event == "PreToolUse": return _emit({"decision": "deny", "reason": "adapter_error:%s" % exc}, 2) return _emit({"decision": "allow", "reason": "adapter_error:%s" % exc}, 0) if __name__ == "__main__": sys.exit(main())