Genarrative/rust/scripts/probe-codex-app-server.sh

#!/usr/bin/env bash
set -euo pipefail

# 可选的 P5 真实 wire 探测：只验证本机 Codex app-server 的 initialize、
# initialized 和 thread/start。它不发送 turn/start，不需要 API key，也不
# 进入默认 CI；真实 Provider/工具副作用仍由调用方显式承担。
codex_bin="${CODEX_BIN:-codex}"
expected_version="${CODEX_EXPECTED_VERSION:-0.153.4}"
tmp_parent="${AGENT_CODEX_PROBE_TMPDIR:-${TMPDIR:-${HOME:?HOME 未设置}/data/tmp}}"
emit_schema=0

while (($# > 0)); do
    case "$1" in
        --schema)
            emit_schema=1
            ;;
        -h|--help)
            echo "用法：$0 [--schema]"
            echo "  --schema 额外生成并摘要真实 app-server v2 schema（仍不发送 turn/start）"
            exit 0
            ;;
        *)
            echo "未知参数：$1" >&2
            exit 2
            ;;
    esac
    shift
done

if ! command -v python3 >/dev/null 2>&1; then
    echo "需要 python3 才能运行 Codex wire probe" >&2
    exit 1
fi
if ! command -v "$codex_bin" >/dev/null 2>&1 && [[ "$codex_bin" != */* ]]; then
    echo "找不到 Codex 可执行文件：$codex_bin" >&2
    exit 1
fi

version_output="$($codex_bin --version 2>/dev/null || true)"
if [[ "$version_output" != *"$expected_version"* ]]; then
    echo "Codex 版本不匹配：期望包含 $expected_version" >&2
    exit 1
fi

mkdir -p -- "$tmp_parent"
tmp_parent="$(cd -- "$tmp_parent" && pwd -P)"
probe_dir="$(mktemp -d "$tmp_parent/codex-app-server-probe.XXXXXX")"
cleanup() {
    # probe_dir 是本次脚本刚创建的精确目录，不触碰 tmp_parent 中其它文件。
    rm -rf -- "$probe_dir"
}
trap cleanup EXIT
mkdir -p -- "$probe_dir/home" "$probe_dir/workspace" "$probe_dir/tmp"

if ((emit_schema == 1)); then
    mkdir -p -- "$probe_dir/schema"
    CODEX_HOME="$probe_dir/home" TMPDIR="$probe_dir/tmp" "$codex_bin" \
        app-server generate-json-schema --out "$probe_dir/schema" --experimental \
        >/dev/null
    schema_file="$probe_dir/schema/codex_app_server_protocol.v2.schemas.json"
    [[ -s "$schema_file" ]] || {
        echo "Codex schema probe 未生成 v2 schema" >&2
        exit 1
    }
    schema_bytes="$(wc -c < "$schema_file")"
    schema_sha="$(sha256sum "$schema_file" | awk '{print $1}')"
    echo "Codex app-server schema: version=$expected_version bytes=$schema_bytes sha256=$schema_sha"
fi

CODEX_HOME="$probe_dir/home" TMPDIR="$probe_dir/tmp" python3 - "$codex_bin" \
    "$probe_dir/home" "$probe_dir/workspace" <<'PY'
import json
import os
import selectors
import signal
import subprocess
import sys
import time

codex_bin, codex_home, workspace = sys.argv[1:]
env = dict(os.environ)
env["CODEX_HOME"] = codex_home
env["TMPDIR"] = os.environ.get("TMPDIR", os.path.join(codex_home, "tmp"))

process = subprocess.Popen(
    [codex_bin, "app-server", "--stdio"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.DEVNULL,
    text=True,
    bufsize=1,
    env=env,
    cwd=workspace,
    start_new_session=True,
)
selector = selectors.DefaultSelector()
selector.register(process.stdout, selectors.EVENT_READ)


def send(frame):
    process.stdin.write(json.dumps(frame, separators=(",", ":")) + "\n")
    process.stdin.flush()


def read_until(frame_id, timeout=8.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        ready = selector.select(max(0.0, deadline - time.monotonic()))
        if not ready:
            continue
        line = process.stdout.readline()
        if not line:
            break
        try:
            frame = json.loads(line)
        except json.JSONDecodeError as error:
            raise RuntimeError(f"返回了非法 JSON（{error.msg}）") from error
        if frame.get("id") != frame_id:
            continue
        if "error" in frame:
            error = frame["error"]
            raise RuntimeError(f"request {frame_id} 失败（code={error.get('code', 'unknown')}）")
        return frame.get("result")
    raise RuntimeError(f"等待 request {frame_id} 响应超时")


try:
    send(
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "clientInfo": {"name": "agent-runtime-probe", "version": "0.1.0"},
                "capabilities": {},
            },
        }
    )
    initialize = read_until(1)
    if not isinstance(initialize, dict) or not isinstance(initialize.get("userAgent"), str):
        raise RuntimeError("initialize result 缺少 userAgent")

    send({"jsonrpc": "2.0", "method": "initialized", "params": {}})
    send(
        {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "thread/start",
            "params": {"cwd": workspace, "approvalPolicy": "never"},
        }
    )
    thread_start = read_until(2)
    thread = thread_start.get("thread") if isinstance(thread_start, dict) else None
    if not isinstance(thread, dict) or not isinstance(thread.get("id"), str):
        raise RuntimeError("thread/start result 缺少 string thread.id")
    print("Codex app-server probe passed: initialize + thread/start")
finally:
    try:
        process.stdin.close()
    except Exception:
        pass
    try:
        process.wait(timeout=2)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(process.pid, signal.SIGTERM)
        except ProcessLookupError:
            pass
        try:
            process.wait(timeout=2)
        except subprocess.TimeoutExpired:
            try:
                os.killpg(process.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            process.wait(timeout=2)
PY

echo "Codex app-server probe finished (version=$expected_version)"