muse-agent-example/docs/write-chapter/probe_sandbox.py

#!/usr/bin/env python3
"""sandbox 探针：走官方 run_claude（sandbox-exec + 最小环境）真调一次 claude。

目的：核实 claude_runtime 的 sandbox 路径在本机能否跑通（前会话实测空 envelope /
EPERM，但未留档原始 stdout/stderr）。本探针用一个只记录不改行为的 runner 包住
subprocess，把完整 command / env 键 / returncode / stdout / stderr 落 artifacts，
失败即可定位 claude 在 sandbox 里到底缺什么。
"""
import json
import subprocess
import sys
import uuid
from datetime import datetime
from decimal import Decimal
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
SKILLS = SCRIPT_DIR.parents[1] / ".claude" / "skills"
for sub in (
    "write-next-chapter/scripts",
    "execute-claude-task/scripts",
    "assemble-context/scripts",
):
    p = str(SKILLS / sub)
    if p not in sys.path:
        sys.path.insert(0, p)

from claude_runtime import ClaudeRuntimeError, run_claude  # noqa: E402
from run_writer import build_writer_execution_profile  # noqa: E402
from writer_contract import validate_writer_draft  # noqa: E402

CLAUDE_BIN = "/Users/qingse/.nvm/versions/node/v24.15.0/bin/claude"
CLAUDE_SHA256 = "ba790279cab6ef77b713864d4bf5f764fcea87d3a3eb7591a41f741e45212b5c"
GATE_A = json.loads((SKILLS / "evaluate-frozen-replay" / "configs" /
                     "writer-gate-a-deep-space-v1.json").read_text(encoding="utf-8"))
WRITER_PROFILE_CFG = GATE_A["executionProfiles"]["writer"]
ARTIFACTS = SCRIPT_DIR / "artifacts"
ARTIFACTS.mkdir(exist_ok=True)

LOG_PATH = ARTIFACTS / f"probe-sandbox-{uuid.uuid4().hex[:8]}.json"


def logging_runner(command, *, input, text, capture_output, timeout, check, cwd, env,
                   start_new_session):
    """透传 subprocess.run，只把现场原样记到留档文件。"""
    record = {"command": list(command), "envKeys": sorted(env.keys()),
              "cwd": cwd, "inputChars": len(input), "startedAt": str(datetime.now())}
    try:
        completed = subprocess.run(list(command), input=input, text=text,
                                   capture_output=capture_output, timeout=timeout,
                                   check=check, cwd=cwd, env=dict(env),
                                   start_new_session=start_new_session)
    except subprocess.TimeoutExpired as exc:
        record["exception"] = f"TimeoutExpired: {exc}"
        LOG_PATH.write_text(json.dumps(record, ensure_ascii=False, indent=1), encoding="utf-8")
        raise
    record.update({"returncode": completed.returncode,
                   "stdout": completed.stdout or "", "stderr": completed.stderr or ""})
    LOG_PATH.write_text(json.dumps(record, ensure_ascii=False, indent=1), encoding="utf-8")
    return completed


def main():
    profile = build_writer_execution_profile(
        claude_executable_path=CLAUDE_BIN, claude_executable_sha256=CLAUDE_SHA256,
        claude_cli_version="2.1.211", resolved_model_id="claude-opus-4-8[1m]",
        effort="high", max_budget_usd_per_call=Decimal("1.0"), timeout_seconds=300,
        max_context_chars=10000, system_prompt=WRITER_PROFILE_CFG["systemPrompt"])
    probe_input = {"fineOutline": {"hardConstraints": ["写一句至少二十字的测试正文。"],
                                   "adjustableBeats": [], "declaredNewFacts": []},
                   "narrativeState": {"time": "", "location": "", "characterPositions": {},
                                      "immediateSituation": "探针"},
                   "factConstraints": [], "proseExcerpts": [], "patternReferences": [],
                   "lengthContract": {"targetChars": 40, "minChars": 10, "maxChars": 200,
                                      "frontmatterRequired": False},
                   "styleConstraints": []}
    try:
        invocation = run_claude(profile, probe_input, runner=logging_runner,
                                business_validator=validate_writer_draft)
    except ClaudeRuntimeError as exc:
        print(f"[探针失败] primary={exc.primary_code}")
        print(f"  causes: {sorted(exc.causes)}")
        receipt = exc.details.get("executionReceipt") or {}
        print(f"  receipt: exitCode={receipt.get('exitCode')} "
              f"actualModelId={receipt.get('actualModelId')} "
              f"terminalReason={receipt.get('terminalReason')}")
        print(f"  现场留档: {LOG_PATH}")
        raise SystemExit(1)
    print("[探针通过] sandbox 路径可跑通")
    print(f"  structured_output: {json.dumps(invocation.structured_output, ensure_ascii=False)[:200]}")
    print(f"  actual_model_id: {invocation.receipt.actual_model_id}")
    print(f"  total_cost_usd: {invocation.receipt.total_cost_usd}")
    print(f"  现场留档: {LOG_PATH}")


if __name__ == "__main__":
    main()