codex exec in CI skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "codex exec in CI"
  short_description: "Headless Codex runs you can gate and audit"
  default_prompt: "$codex-exec-ci Add a GitLab CI job that runs codex exec to triage failing tests"

#!/usr/bin/env python3
"""Summarize a `codex exec --json` event stream for CI logs and gates.

Reads JSON Lines (thread.started, turn.*, item.*, error) from a file or stdin
and prints the commands Codex ran with exit codes, files it changed, MCP tool
calls, errors, token usage and the final agent message. Optional gates turn
the summary into a pass/fail signal.

Usage:
  codex exec --json "..." > events.jsonl
  python3 scripts/summarize_events.py events.jsonl [--forbid-changes]
         [--fail-on-command-error] [--markdown summary.md] [--json-out summary.json]
  codex exec --json "..." | python3 scripts/summarize_events.py -

Exit codes: 0 ok, 1 run failed or a gate tripped, 2 unreadable input.
"""
from __future__ import annotations

import argparse
import json
import sys


def load(path: str) -> list[dict]:
    stream = sys.stdin if path == "-" else open(path, encoding="utf-8")
    events = []
    with stream:
        for number, raw in enumerate(stream, 1):
            raw = raw.strip()
            if not raw:
                continue
            try:
                events.append(json.loads(raw))
            except json.JSONDecodeError:
                print(f"warning: line {number} is not JSON, skipped", file=sys.stderr)
    return events


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("events", help="JSONL file from codex exec --json, or - for stdin")
    ap.add_argument("--forbid-changes", action="store_true", help="fail if Codex changed any file")
    ap.add_argument("--fail-on-command-error", action="store_true", help="fail if any command exited non-zero")
    ap.add_argument("--markdown", help="also write a Markdown summary to this path")
    ap.add_argument("--json-out", help="also write the summary as JSON to this path")
    args = ap.parse_args()

    try:
        events = load(args.events)
    except OSError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    thread_id = None
    turns_completed = 0
    failures: list[str] = []
    usage = {"input_tokens": 0, "cached_input_tokens": 0, "output_tokens": 0, "reasoning_output_tokens": 0}
    commands, changes, tools = [], {}, []
    final_message = ""
    stream_errors: list[tuple[int, str]] = []
    last_completed = -1
    warnings: list[str] = []

    for index, ev in enumerate(events):
        kind = ev.get("type")
        if kind == "thread.started":
            thread_id = ev.get("thread_id")
        elif kind == "turn.completed":
            turns_completed += 1
            last_completed = index
            for key in usage:
                usage[key] += int(ev.get("usage", {}).get(key, 0) or 0)
        elif kind == "turn.failed":
            failures.append("turn failed: " + str(ev.get("error", {}).get("message", "")))
        elif kind == "error":
            stream_errors.append((index, str(ev.get("message", ""))))
        elif kind == "item.completed":
            item = ev.get("item", {})
            itype = item.get("type")
            if itype == "command_execution":
                commands.append({"command": item.get("command", ""), "exit_code": item.get("exit_code"), "status": item.get("status")})
            elif itype == "file_change":
                for change in item.get("changes", []):
                    changes[change.get("path", "?")] = change.get("kind", "?")
            elif itype == "mcp_tool_call":
                tools.append({"server": item.get("server"), "tool": item.get("tool"), "status": item.get("status")})
            elif itype == "agent_message":
                final_message = item.get("text", "")
            elif itype == "error":
                failures.append("item error: " + str(item.get("message", "")))

    for index, message in stream_errors:
        # Errors followed by a completed turn were recovered (for example retried requests).
        (warnings if index < last_completed else failures).append("error: " + message)

    if not events:
        failures.append("no events read")
    elif turns_completed == 0 and not failures:
        failures.append("stream ended without turn.completed (killed or timed out?)")

    gates = []
    failed_cmds = [c for c in commands if c["exit_code"] not in (0, None) or c["status"] in ("failed", "declined")]
    if args.forbid_changes and changes:
        gates.append(f"--forbid-changes: {len(changes)} file(s) changed")
    if args.fail_on_command_error and failed_cmds:
        gates.append(f"--fail-on-command-error: {len(failed_cmds)} command(s) failed")

    lines = [f"## Codex run {thread_id or '(unknown thread)'}", ""]
    lines.append(f"- Turns completed: {turns_completed}")
    lines.append(f"- Tokens: input {usage['input_tokens']} (cached {usage['cached_input_tokens']}), output {usage['output_tokens']}, reasoning {usage['reasoning_output_tokens']}")
    lines.append(f"- Commands: {len(commands)} ({len(failed_cmds)} failed or declined)")
    for c in commands:
        lines.append(f"  - `{c['command'][:160]}` -> exit {c['exit_code']} ({c['status']})")
    lines.append(f"- Files changed: {len(changes)}")
    for path, kind in sorted(changes.items()):
        lines.append(f"  - {kind}: {path}")
    if tools:
        lines.append(f"- MCP tool calls: {len(tools)}")
        for t in tools:
            lines.append(f"  - {t['server']}.{t['tool']} ({t['status']})")
    for warning in warnings:
        lines.append(f"- warning (recovered): {warning}")
    for problem in failures + gates:
        lines.append(f"- PROBLEM: {problem}")
    lines += ["", "### Final message", "", final_message or "(none)"]
    text = "\n".join(lines) + "\n"
    print(text)

    if args.markdown:
        with open(args.markdown, "w", encoding="utf-8") as fh:
            fh.write(text)
    if args.json_out:
        with open(args.json_out, "w", encoding="utf-8") as fh:
            json.dump({"thread_id": thread_id, "turns_completed": turns_completed, "usage": usage,
                       "commands": commands, "changes": changes, "mcp_tool_calls": tools,
                       "problems": failures + gates, "warnings": warnings, "final_message": final_message}, fh, indent=2)
    return 1 if failures or gates else 0


if __name__ == "__main__":
    sys.exit(main())