Codex 无头模式实战

npm install -g @openai/codex
codex --version
codex exec --help
python3 --version
git --version

codex login
codex login status

# 无浏览器环境的可选登录方式
codex login --device-auth

mkdir codex-headless-demo
cd codex-headless-demo
git init
mkdir -p data scripts .runs
printf ".runs/\n__pycache__/\n" > .gitignore

[
  {"sku":"P100","name":"键盘","price_cents":12900},
  {"sku":"P200","name":"数据线","price_cents":3900},
  {"sku":"P200","name":"鼠标","price_cents":-500}
]

codex exec --sandbox read-only \
  '读取 data/products.json。
按数组顺序检查重复 SKU，以及不为正整数的 price_cents。
列出问题所在记录，不修改文件，不访问网络。' \
  > .runs/first-report.txt 2> .runs/first-run.log

cat .runs/first-report.txt

codex exec --sandbox read-only --json \
  --output-last-message .runs/final.txt \
  "只读检查data/products.json并汇总发现" \
  > .runs/events.jsonl 2> .runs/run.log

只读检查 data/products.json，不修改任何文件，不访问网络。
按数组顺序扫描全部记录，行号从1开始。
SKU第一次出现不算重复，后续相同SKU记为duplicate_sku。
price_cents必须是大于0的整数，否则记为invalid_price。
一条记录可同时产生两项问题。checked为记录总数。
有问题时status为needs_review，否则为ok。
最终答复必须符合提供的JSON Schema，不要添加解释或Markdown。

{
  "type": "object",
  "properties": {
    "status": {"type": "string", "enum": ["ok", "needs_review"]},
    "checked": {"type": "integer"},
    "issues": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "row": {"type": "integer"},
          "sku": {"type": "string"},
          "code": {"type": "string", "enum": ["duplicate_sku", "invalid_price"]}
        },
        "required": ["row", "sku", "code"],
        "additionalProperties": false
      }
    }
  },
  "required": ["status", "checked", "issues"],
  "additionalProperties": false
}

{
  "status": "needs_review",
  "checked": 3,
  "issues": [
    {"row": 3, "sku": "P200", "code": "duplicate_sku"},
    {"row": 3, "sku": "P200", "code": "invalid_price"}
  ]
}

#!/usr/bin/env python3
"""Validate this tutorial's exact output contract, then recompute the findings."""
import json
import sys
from pathlib import Path


def require(condition, message):
    if not condition:
        raise ValueError(message)


def unique_object(pairs):
    obj = {}
    for key, value in pairs:
        require(key not in obj, f"duplicate JSON key: {key}")
        obj[key] = value
    return obj


def reject_constant(value):
    raise ValueError(f"non-standard JSON number: {value}")


def read_json(path):
    # Read one byte past the limit, not the entire untrusted file.
    with Path(path).open("rb") as stream:
        raw = stream.read(65537)
    require(0 < len(raw) <= 65536, "empty or oversized JSON")
    return json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object,
                      parse_constant=reject_constant)


def verify(result_path, data_path):
    result, rows = read_json(result_path), read_json(data_path)
    require(type(rows) is list, "input must be an array")
    require(type(result) is dict, "result must be an object")
    require(set(result) == {"status", "checked", "issues"}, "wrong result keys")
    require(result["status"] in ("ok", "needs_review"), "invalid status")
    require(type(result["checked"]) is int, "checked must be an integer")
    require(type(result["issues"]) is list, "issues must be an array")

    expected, seen = [], set()
    for number, row in enumerate(rows, start=1):
        require(type(row) is dict and type(row.get("sku")) is str,
                "input row needs a string sku")
        sku, price = row["sku"], row.get("price_cents")
        if sku in seen:
            expected.append((number, sku, "duplicate_sku"))
        seen.add(sku)
        if type(price) is not int or price <= 0:
            expected.append((number, sku, "invalid_price"))

    actual = []
    for issue in result["issues"]:
        require(type(issue) is dict, "issue must be an object")
        require(set(issue) == {"row", "sku", "code"}, "wrong issue keys")
        require(type(issue["row"]) is int and 1 <= issue["row"] <= len(rows),
                "invalid row number")
        require(type(issue["sku"]) is str, "sku must be a string")
        require(issue["code"] in ("duplicate_sku", "invalid_price"), "invalid code")
        actual.append((issue["row"], issue["sku"], issue["code"]))

    require(result["checked"] == len(rows), "wrong checked count")
    require(sorted(actual) == sorted(expected), "missing or invented findings")
    expected_status = "needs_review" if expected else "ok"
    require(result["status"] == expected_status, "status contradicts findings")
    print(f"VALID: {len(rows)} rows, {len(expected)} verified issues")


if __name__ == "__main__":
    try:
        require(len(sys.argv) == 3, "usage: verify.py RESULT_JSON INPUT_JSON")
        verify(sys.argv[1], sys.argv[2])
    except (OSError, UnicodeError, ValueError, TypeError) as exc:
        print(f"INVALID: {exc}", file=sys.stderr)
        raise SystemExit(2)

python3 scripts/verify.py sample-result.json data/products.json

#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
mkdir -p .runs
run_dir=$(mktemp -d "$PWD/.runs/audit.XXXXXX")
printf '%s\n' "$run_dir" > .runs/latest
codex --version > "$run_dir/cli-version.txt"
git rev-parse HEAD > "$run_dir/commit.txt"

if codex exec --sandbox read-only \
    --json --output-schema audit.schema.json \
    --output-last-message "$run_dir/result.json" \
    - < audit.prompt.txt \
    > "$run_dir/events.jsonl" 2> "$run_dir/run.log"; then
    :
else
    rc=$?
    printf 'Codex failed: exit=%s; evidence=%s\n' "$rc" "$run_dir" >&2
    exit "$rc"
fi

python3 scripts/verify.py "$run_dir/result.json" data/products.json
printf 'Verified result: %s/result.json\n' "$run_dir"

git add .
git commit -m "Add headless audit example"
bash run-audit.sh

#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
if [[ -n $(git status --porcelain) ]]; then
  printf 'Start from a clean tutorial working tree.\n' >&2
  exit 1
fi
mkdir -p .runs
run_dir=$(mktemp -d "$PWD/.runs/edit.XXXXXX")
codex exec --sandbox workspace-write \
  --output-last-message "$run_dir/summary.txt" \
  '按照audit.prompt.txt的检查规则，
生成reports/catalog-audit.json，结构遵守audit.schema.json。
仅创建或覆盖该报告，不改输入数据、提示词、Schema和脚本。
不执行提交、推送、网络或发布操作。'
python3 scripts/verify.py reports/catalog-audit.json data/products.json
# This narrow example permits only one new report; reject every other change.
python3 - <<'PY'
import subprocess
allowed = b'?? reports/catalog-audit.json'
records = subprocess.check_output(
    ['git', 'status', '--porcelain=v1', '-z', '--untracked-files=all']
).split(b'\0')
if [r for r in records if r] != [allowed]:
    raise SystemExit('Unexpected workspace changes; inspect git status and diff')
print('Only the expected report was added')
PY
git diff --check
git status --short

name: Codex catalog audit
on:
  workflow_dispatch:

permissions:
  contents: read

concurrency:
  group: codex-catalog-audit-${{ github.ref }}
  cancel-in-progress: false

jobs:
  audit:
    if: >-
      github.ref ==
      format('refs/heads/{0}', github.event.repository.default_branch)
    runs-on: ubuntu-latest
    timeout-minutes: 10
    outputs:
      report: ${{ steps.codex.outputs.final-message }}
    steps:
      - uses: actions/checkout@v5
        with:
          ref: ${{ github.sha }}
          persist-credentials: false
      # Keep Codex last in this job. Do not run repository build/test hooks here.
      - uses: openai/codex-action@v1
        id: codex
        with:
          openai-api-key: ${{ secrets.OPENAI_API_KEY }}
          safety-strategy: drop-sudo
          permission-profile: ':read-only'
          prompt-file: audit.prompt.txt
          output-schema-file: audit.schema.json

  verify:
    needs: audit
    runs-on: ubuntu-latest
    timeout-minutes: 3
    steps:
      - uses: actions/checkout@v5
        with:
          ref: ${{ github.sha }}
          persist-credentials: false
      - name: Validate the report as data
        env:
          CODEX_REPORT: ${{ needs.audit.outputs.report }}
        run: |
          python3 - <<'PY'
          import os
          from pathlib import Path
          raw = os.environ['CODEX_REPORT'].encode('utf-8')
          if not 0 < len(raw) <= 65536:
              raise SystemExit('Missing or oversized report')
          Path('result.json').write_bytes(raw)
          PY
          python3 scripts/verify.py result.json data/products.json

#!/usr/bin/env python3
"""Standalone POSIX wrapper with waitid/WNOWAIT; timeout exit is 124."""
import argparse
import math
import os
import signal
import subprocess
import sys
import time


class IdentityLost(RuntimeError):
    pass


def positive_seconds(value):
    number = float(value)
    if not math.isfinite(number) or not 0 < number <= 86400:
        raise ValueError('seconds must be finite and within (0, 86400]')
    return number


def require_support():
    needed = ('waitid', 'P_PID', 'WEXITED', 'WNOHANG', 'WNOWAIT', 'CLD_EXITED')
    if os.name != 'posix' or not all(hasattr(os, key) for key in needed):
        raise ValueError('POSIX waitid/WNOWAIT support is required')
    if signal.getsignal(signal.SIGCHLD) != signal.SIG_DFL:
        raise ValueError('standalone wrapper requires default SIGCHLD')


def peek_child(pid):
    try:
        return os.waitid(os.P_PID, pid, os.WEXITED | os.WNOHANG | os.WNOWAIT)
    except OSError as exc:
        raise IdentityLost('child identity uncertain; no more signals') from exc


def group_signal(pgid, sig):
    try:
        os.killpg(pgid, sig)
        return True
    except ProcessLookupError:
        return False


def stop_group(process, grace):
    # Do not reap the leader: its reserved PID protects the group identity.
    peek_child(process.pid)
    if not group_signal(process.pid, signal.SIGTERM):
        return
    time.sleep(grace)
    peek_child(process.pid)
    group_signal(process.pid, signal.SIGKILL)


def run(seconds, command, grace=10):
    seconds, grace = positive_seconds(seconds), positive_seconds(grace)
    require_support()
    process = subprocess.Popen(command, start_new_session=True)
    identity_valid = True
    try:
        until = time.monotonic() + seconds
        while True:
            status = peek_child(process.pid)
            if status is not None:
                if status.si_code == os.CLD_EXITED:
                    return status.si_status
                return 128 + status.si_status
            remaining = until - time.monotonic()
            if remaining <= 0:
                print('Deadline exceeded; inspect outputs before retrying',
                      file=sys.stderr)
                return 124
            time.sleep(min(0.02, remaining))
    except IdentityLost:
        identity_valid = False
        raise
    finally:
        if identity_valid:
            # If identity becomes uncertain here, skip wait as well as signals.
            stop_group(process, grace)
            process.wait()  # The only reap; no later group signal is allowed.


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--grace', type=positive_seconds, default=10)
    parser.add_argument('seconds', type=positive_seconds)
    parser.add_argument('command', nargs=argparse.REMAINDER)
    args = parser.parse_args()
    if not args.command:
        parser.error('COMMAND [ARG ...] is required')
    try:
        return run(args.seconds, args.command, args.grace)
    except (OSError, ValueError, IdentityLost) as exc:
        print(f'Wrapper failed: {exc}', file=sys.stderr)
        return 2


if __name__ == '__main__':
    raise SystemExit(main())

python3 scripts/with_deadline.py 600 bash run-audit.sh

python3 scripts/with_deadline.py --grace 0.2 2 \
  python3 -c 'print("offline success")'
printf 'exit=%s\n' "$?"

if python3 scripts/with_deadline.py --grace 0.2 0.2 \
    python3 -c 'import time; time.sleep(60)'; then
  rc=0
else
  rc=$?
fi
printf 'exit=%s\n' "$rc"

python3 -m unittest discover -s scripts -p test_deadline.py -v

codex exec --sandbox read-only resume SESSION_ID \
  "继续只读检查，说明尚未完成的部分"

#!/usr/bin/env python3
"""Local synthetic Bash/tee experiment. Python standard library only."""
import json
import os
from pathlib import Path
import platform
import shutil
import subprocess
import sys
import tempfile

PRODUCER = '''import sys
print("fixture-event: one", flush=True)
sys.exit(int(sys.argv[1]))
'''

PROBE = r'''#!/usr/bin/env bash
set -eu
set +o pipefail
if [[ $1 == on ]]; then set -o pipefail; fi
py=$2; producer=$3; code=$4; log=$5; ledger=$6
if "$py" "$producer" "$code" | tee "$log"; then
  snapshot=( "$?" "${PIPESTATUS[@]}" )
else
  snapshot=( "$?" "${PIPESTATUS[@]}" )
fi
# The assignment above also changes PIPESTATUS; use only snapshot from here.
if (( ${#snapshot[@]} != 3 )); then exit 90; fi
aggregate=${snapshot[0]}; producer_rc=${snapshot[1]}; logger_rc=${snapshot[2]}
if (( producer_rc != 0 && logger_rc != 0 )); then
  gate=stop_both
elif (( producer_rc != 0 )); then
  gate=stop_producer
elif (( logger_rc != 0 )); then
  gate=stop_logger
else
  gate=verify_next
fi
# A failed ledger write is itself a failure; do not turn it into success.
printf '%s\t%s\t%s\t%s\n' \
  "$aggregate" "$producer_rc" "$logger_rc" "$gate" > "$ledger" || exit 91
if (( aggregate != 0 )); then exit "$aggregate"; fi
if (( producer_rc != 0 )); then exit "$producer_rc"; fi
if (( logger_rc != 0 )); then exit "$logger_rc"; fi
exit 0
'''

OVERWRITTEN = r'''#!/usr/bin/env bash
set +e
set -o pipefail
"$1" "$2" 7 | tee "$3"
aggregate=$?
overwritten=( "${PIPESTATUS[@]}" )
printf '%s\t%s\t%s\n' "$aggregate" "${#overwritten[@]}" \
  "${overwritten[*]}" > "$4" || exit 91
# Reject a one-element vector instead of assuming a missing stage succeeded.
if (( ${#overwritten[@]} != 2 )); then exit 90; fi
exit 0
'''

SHELL_MODE = r'''#!/usr/bin/env bash
# Explicit negative-control setup, not a proposed workflow setting.
if [[ $1 == off ]]; then set +o pipefail; fi
"$2" "$3" 7 | tee "$4"
printf '%s\n' reached > "$5"
'''


def require(ok, message):
    if not ok:
        raise AssertionError(message)


def main():
    if len(sys.argv) != 2:
        raise SystemExit("usage: python3 reproduce.py NEW_OUTPUT_DIRECTORY")
    out = Path(sys.argv[1]).resolve()
    out.mkdir(parents=True, exist_ok=False)
    # No inherited BASH_ENV, SHELLOPTS, exported functions, or credentials.
    env = {"PATH": os.defpath, "LC_ALL": "C", "PYTHONIOENCODING": "utf-8"}
    bash = shutil.which("bash", path=env["PATH"])
    tee = shutil.which("tee", path=env["PATH"])
    require(bash is not None and tee is not None, "Bash and tee are required")
    py = str(Path(sys.executable).resolve())

    def run(args, cwd):
        return subprocess.run(args, cwd=cwd, env=env, capture_output=True,
                              text=True, encoding="utf-8", timeout=10)

    def save(name, result):
        (out / (name + ".stdout.txt")).write_text(result.stdout, encoding="utf-8")
        (out / (name + ".stderr.txt")).write_text(result.stderr, encoding="utf-8")

    tee_version = run([tee, "--version"], out)
    require(tee_version.returncode == 0 and "GNU coreutils" in tee_version.stdout,
            "This expected-result fixture targets GNU tee; other variants untested")
    environment = {
        "os": platform.platform(), "python": platform.python_version(),
        "bash": run([bash, "--version"], out).stdout.splitlines()[0],
        "tee": tee_version.stdout.splitlines()[0],
        "bash_path": bash, "tee_path": tee, "python_path": py,
        "scope": "local synthetic processes; no Codex or hosted CI execution",
    }
    (out / "environment.json").write_text(
        json.dumps(environment, indent=2) + "\n", encoding="utf-8")
    cases = [
        # name, pipefail, producer exit, logger destination, snapshot, gate, exit
        ("healthy", "on", 0, "file", [0, 0, 0], "verify_next", 0),
        ("masked", "off", 7, "file", [0, 7, 0], "stop_producer", 7),
        ("producer_fail", "on", 7, "file", [7, 7, 0], "stop_producer", 7),
        ("logger_fail", "on", 0, "directory", [1, 0, 1], "stop_logger", 1),
        ("both_fail", "on", 7, "directory", [1, 7, 1], "stop_both", 1),
    ]
    lines = ["case pipefail aggregate producer logger aggregate_only stage_gate wrapper_exit"]
    results = []
    with tempfile.TemporaryDirectory(prefix="pipeline-fixture-") as tmp:
        root = Path(tmp)
        scripts = {"producer.py": PRODUCER, "probe.sh": PROBE,
                   "overwritten.sh": OVERWRITTEN, "shell-mode.sh": SHELL_MODE}
        for name, body in scripts.items():
            (root / name).write_text(body, encoding="utf-8")
            (out / name).write_text(body, encoding="utf-8")
            if name.endswith(".sh"):
                require(run([bash, "-n", str(root / name)], root).returncode == 0,
                        name + ": Bash syntax")
        for name, mode, code, destination, expected, gate, expected_exit in cases:
            case = root / name
            case.mkdir()
            log = case / "log.txt"
            if destination == "directory":
                log.mkdir()  # Opening a directory as a file fails; no disk filling.
            ledger = case / "ledger.tsv"
            result = run([bash, "--noprofile", "--norc", str(root / "probe.sh"),
                          mode, py, str(root / "producer.py"), str(code),
                          str(log), str(ledger)], root)
            save(name, result)
            fields = ledger.read_text(encoding="utf-8").strip().split("\t")
            require(len(fields) == 4, name + ": ledger shape")
            vector = [int(value) for value in fields[:3]]
            require(vector == expected, name + ": aggregate/stage snapshot")
            require(fields[3] == gate and result.returncode == expected_exit,
                    name + ": explicit gate/exit")
            require(result.stdout == "fixture-event: one\n", name + ": stdout")
            if destination == "file":
                require(log.read_text() == result.stdout, name + ": complete log")
                require(result.stderr == "", name + ": unexpected stderr")
                (out / (name + ".log.txt")).write_text(result.stdout)
            else:
                require(log.is_dir() and "Is a directory" in result.stderr,
                        name + ": expected logger diagnostic")
            (out / (name + ".ledger.tsv")).write_text(ledger.read_text())
            aggregate_gate = "allow" if vector[0] == 0 else "stop"
            lines.append(f"{name} {mode} {' '.join(map(str, vector))} "
                         f"{aggregate_gate} {gate} {result.returncode}")
            results.append({"case": name, "snapshot": vector, "stage_gate": gate,
                            "wrapper_exit": result.returncode})
        result = run([bash, "--noprofile", "--norc", str(root / "overwritten.sh"),
                      py, str(root / "producer.py"), str(root / "negative.log"),
                      str(root / "negative.tsv")], root)
        save("overwritten", result)
        observed = (root / "negative.tsv").read_text()
        require(observed == "7\t1\t0\n" and result.returncode == 90,
                "overwritten PIPESTATUS must be detected")
        require(result.stdout == "fixture-event: one\n" and result.stderr == "",
                "overwritten control streams")
        lines.append("overwritten saved_aggregate=7 observed=[0] stage_count=1 rejected=90")
        for name, flags, mode, expected_exit, reached in [
            ("bash_e", ["-e"], "off", 0, True),
            ("explicit_bash", ["--noprofile", "--norc", "-eo", "pipefail"],
             "on", 7, False),
        ]:
            sentinel = root / (name + ".sentinel")
            log = root / (name + ".log")
            result = run([bash, *flags, str(root / "shell-mode.sh"), mode,
                          py, str(root / "producer.py"), str(log), str(sentinel)], root)
            save(name, result)
            require(result.returncode == expected_exit and sentinel.exists() == reached,
                    name + ": status and sentinel")
            require(result.stdout == "fixture-event: one\n" and result.stderr == ""
                    and log.read_text() == result.stdout, name + ": streams and log")
            if reached:
                require(sentinel.read_text() == "reached\n", name + ": sentinel content")
            lines.append(f"{name} exit={result.returncode} sentinel={'yes' if reached else 'no'}")
    lines.append("PASS: 5 pipeline cases, 1 overwritten-status control, 2 shell-mode cases")
    text = "\n".join(lines) + "\n"
    (out / "summary.txt").write_text(text, encoding="utf-8")
    (out / "results.json").write_text(json.dumps(results, indent=2) + "\n")
    print(text, end="")


if __name__ == "__main__":
    main()

case pipefail aggregate producer logger aggregate_only stage_gate wrapper_exit
healthy on 0 0 0 allow verify_next 0
masked off 0 7 0 allow stop_producer 7
producer_fail on 7 7 0 stop stop_producer 7
logger_fail on 1 0 1 stop stop_logger 1
both_fail on 1 7 1 stop stop_both 1
overwritten saved_aggregate=7 observed=[0] stage_count=1 rejected=90
bash_e exit=0 sentinel=yes
explicit_bash exit=7 sentinel=no
PASS: 5 pipeline cases, 1 overwritten-status control, 2 shell-mode cases

snapshot=( "$?" "${PIPESTATUS[@]}" )
# snapshot[0] 是管道；snapshot[1] 是生产者；snapshot[2] 是 tee
# 之后只读 snapshot，不再依赖当前的 PIPESTATUS

if [[ $1 == off ]]; then set +o pipefail; fi
"$2" "$3" 7 | tee "$4"
printf '%s\n' reached > "$5"