205 lines
8.1 KiB
Python
205 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import pathlib
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
|
||
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
||
MAP = ROOT / "routing/persona-execution-limb-agent-map.json"
|
||
PERSONAS = ROOT / "routing/persona-system-canonical-map.json"
|
||
PROFILES = pathlib.Path(__file__).resolve().parent / "profiles"
|
||
RUNTIMES = {
|
||
"qwen": "/Users/bingshuolingdianyuanhe/.npm-global/bin/qwen",
|
||
"opencode": "/Users/bingshuolingdianyuanhe/.npm-global/bin/opencode",
|
||
"codex": "/Applications/ChatGPT.app/Contents/Resources/codex",
|
||
}
|
||
QWEN_AUTH_TYPE = "openai"
|
||
QWEN_MODELS = ["qwen3.8-max", "qwen3.7-plus", "deepseek-v4-pro", "deepseek-v4-flash-0731", "glm-5.2"]
|
||
|
||
|
||
def load(path: pathlib.Path) -> dict:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def registered_personas() -> dict[str, dict]:
|
||
return {item["id"]: item for item in load(PERSONAS)["persona_systems"]}
|
||
|
||
|
||
def profile(persona_id: str) -> dict:
|
||
personas = registered_personas()
|
||
if persona_id not in personas:
|
||
raise ValueError("PERSONA_NOT_REGISTERED_NO_SYNTHETIC_WAKE")
|
||
exact = PROFILES / f"{persona_id}.json"
|
||
value = load(exact if exact.is_file() else PROFILES / "default.json")
|
||
if value.get("parent_persona_id") is None:
|
||
value["profile_id"] = f"LIMB::{persona_id}::PRIMARY"
|
||
value["parent_persona_id"] = persona_id
|
||
value["state"] = "DERIVED_REQUIRES_PARENT_TASK_ENVELOPE"
|
||
return value
|
||
|
||
|
||
def probe() -> dict:
|
||
return {
|
||
"schema": "guanghu.persona-execution-limb-tool-probe/v1",
|
||
"runtimes": {
|
||
name: {"path": path, "available": pathlib.Path(path).is_file() and bool(shutil.which(path))}
|
||
for name, path in RUNTIMES.items()
|
||
},
|
||
"secrets_read": False,
|
||
}
|
||
|
||
|
||
def envelope(persona_id: str, task_id: str, task: str, runtime: str) -> dict:
|
||
tools = probe()["runtimes"]
|
||
if runtime not in tools or not tools[runtime]["available"]:
|
||
raise ValueError("HOST_TOOL_UNAVAILABLE_REPORT_NOT_INVENT")
|
||
p = profile(persona_id)
|
||
return {
|
||
"schema": "guanghu.persona-execution-limb-task/v1",
|
||
"task_id": task_id,
|
||
"parent_persona_id": persona_id,
|
||
"limb_profile_id": p["profile_id"],
|
||
"runtime": runtime,
|
||
"task": task,
|
||
"working_defaults": p["working_defaults"],
|
||
"authority": "CURRENT_TASK_ENVELOPE_ONLY",
|
||
"controller": persona_id,
|
||
"receipt_required": True,
|
||
"prohibited": load(MAP)["prohibited"],
|
||
}
|
||
|
||
|
||
def prompt(value: dict) -> str:
|
||
return (
|
||
"你是父人格训练的有界Agent执行手脚，不是独立主控人格。"
|
||
"不得审核父人格目的、扩大权限、读取秘密或虚构完成。"
|
||
"请完成任务中安全且明确的部分；如有精确事实冲突，返回证据和可继续路径。"
|
||
"最终只输出JSON，字段为outcome,actions,evidence,tests,unresolved,training_candidate。\n"
|
||
+ json.dumps(value, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
def qwen_succeeded(completed: subprocess.CompletedProcess[str]) -> bool:
|
||
if completed.returncode != 0 or "[API Error:" in completed.stdout:
|
||
return False
|
||
try:
|
||
stream = json.loads(completed.stdout)
|
||
except json.JSONDecodeError:
|
||
return False
|
||
if not isinstance(stream, list):
|
||
return False
|
||
results = [item for item in stream if item.get("type") == "result"]
|
||
return bool(results) and results[-1].get("is_error") is False and bool(str(results[-1].get("result", "")).strip())
|
||
|
||
|
||
def codex_succeeded(completed: subprocess.CompletedProcess[str]) -> bool:
|
||
if completed.returncode != 0:
|
||
return False
|
||
try:
|
||
events = [json.loads(line) for line in completed.stdout.splitlines() if line.strip()]
|
||
except json.JSONDecodeError:
|
||
return False
|
||
messages = [
|
||
event.get("item", {}).get("text")
|
||
for event in events
|
||
if event.get("type") == "item.completed" and event.get("item", {}).get("type") == "agent_message"
|
||
]
|
||
return any(isinstance(message, str) and message.strip() for message in messages)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
sub = parser.add_subparsers(dest="command", required=True)
|
||
sub.add_parser("probe")
|
||
prepare = sub.add_parser("prepare")
|
||
for p in (prepare,):
|
||
p.add_argument("--persona", required=True)
|
||
p.add_argument("--task-id", required=True)
|
||
p.add_argument("--task", required=True)
|
||
p.add_argument("--runtime", required=True, choices=tuple(RUNTIMES))
|
||
run = sub.add_parser("run")
|
||
run.add_argument("--persona", required=True)
|
||
run.add_argument("--task-id", required=True)
|
||
run.add_argument("--task", required=True)
|
||
run.add_argument("--runtime", required=True, choices=("qwen", "codex"))
|
||
run.add_argument("--cwd", required=True)
|
||
args = parser.parse_args()
|
||
try:
|
||
if args.command == "probe":
|
||
result = probe()
|
||
else:
|
||
value = envelope(args.persona, args.task_id, args.task, args.runtime)
|
||
if args.command == "prepare":
|
||
result = value
|
||
else:
|
||
cwd = pathlib.Path(args.cwd).resolve()
|
||
if not cwd.is_dir():
|
||
raise ValueError("TASK_CWD_REQUIRED")
|
||
attempts = []
|
||
completed = None
|
||
selected_model = None
|
||
if args.runtime == "qwen":
|
||
for model in QWEN_MODELS:
|
||
candidate = subprocess.run(
|
||
[
|
||
RUNTIMES["qwen"],
|
||
"--auth-type", QWEN_AUTH_TYPE,
|
||
"--model", model,
|
||
"--prompt", prompt(value),
|
||
"--output-format", "json",
|
||
],
|
||
cwd=cwd, capture_output=True, text=True, timeout=180, check=False,
|
||
)
|
||
ok = qwen_succeeded(candidate)
|
||
attempts.append({"model": model, "exit_code": candidate.returncode, "semantic_success": ok})
|
||
completed = candidate
|
||
if ok:
|
||
selected_model = model
|
||
break
|
||
else:
|
||
candidate = subprocess.run(
|
||
[
|
||
RUNTIMES["codex"],
|
||
"-a", "never",
|
||
"exec",
|
||
"--ephemeral",
|
||
"--ignore-rules",
|
||
"--sandbox", "read-only",
|
||
"--cd", str(cwd),
|
||
"--json",
|
||
prompt(value),
|
||
],
|
||
cwd=cwd, capture_output=True, text=True, timeout=180, check=False,
|
||
)
|
||
ok = codex_succeeded(candidate)
|
||
attempts.append({"model": "codex-configured-default", "exit_code": candidate.returncode, "semantic_success": ok})
|
||
completed = candidate
|
||
if ok:
|
||
selected_model = "codex-configured-default"
|
||
assert completed is not None
|
||
success = selected_model is not None
|
||
result = {
|
||
"schema": "guanghu.persona-execution-limb-runtime-receipt/v1",
|
||
"task_id": value["task_id"],
|
||
"parent_persona_id": value["parent_persona_id"],
|
||
"runtime": args.runtime,
|
||
"selected_model": selected_model,
|
||
"attempts": attempts,
|
||
"exit_code": completed.returncode,
|
||
"stdout": completed.stdout[-20000:],
|
||
"stderr": completed.stderr[-4000:],
|
||
"outcome": "PASS" if success else "FAIL",
|
||
}
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
return 0
|
||
except Exception as error:
|
||
print(json.dumps({"outcome": "FAIL", "error": str(error)}, ensure_ascii=False))
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|