#!/usr/bin/env python3 """OpenHands brief -> local Codex execution card helper. OpenHands only returns a small planning brief for continuation: - goal - scope - in_scope - out_of_scope The local wrapper then synthesizes the real Codex execution card from repo truth sources, so OpenHands no longer decides files_to_touch/tests_to_run. """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path from typing import Any, Iterable from urllib import error, parse, request DEFAULT_README = "tests/apifox/README.md" DEFAULT_TODO = "prompt/raw-requirements/feat-dengbao/Codex_TODO_智慧食堂_04_整改执行版_20260415.md" DEFAULT_SANDBOX_ROOT = "/workspace/project" BRIEF_FIELDS = ("goal", "scope", "in_scope", "out_of_scope") LOCAL_FOCUS_PRESETS = { "http_real_flow": { "focus_key": "http_real_flow", "files_to_touch": [ "tests/apifox/http/run_extended_smoke.sh", "tests/apifox/scenarios/core_flows.md", "tests/apifox/README.md", ], "tests_to_run": [ "bash tests/apifox/http/run_extended_smoke.sh", ], "done_definition": [ "至少补齐一条真实 HTTP 业务联调链路，并能被扩展 smoke 直接执行", "README 回写新增链路与执行方式", "相关脚本或场景文档与当前仓库接口事实保持一致", ], }, "mobile_deeper": { "focus_key": "mobile_deeper", "files_to_touch": [ "tests/apifox/mobile/MobileDishesContractValidationTest.php", "tests/apifox/mobile/MobileDateMenuSuccessContractTest.php", "tests/apifox/mobile/MobileStaffSuccessContractTest.php", "tests/apifox/README.md", ], "tests_to_run": [ "$(brew --prefix php@8.2)/bin/php /tmp/phpunit-11.phar --bootstrap tests/bootstrap.php tests/apifox/mobile/MobileDishesGuardMatrixTest.php tests/apifox/mobile/MobileDishesContractValidationTest.php tests/apifox/mobile/MobileDateMenuSuccessContractTest.php", "$(brew --prefix php@8.2)/bin/php /tmp/phpunit-11.phar --bootstrap tests/bootstrap.php tests/apifox/mobile/MobileStaffSuccessContractTest.php", ], "done_definition": [ "至少补齐一条 /m/** 深层成功分支或参数契约缺口", "README 回写新增覆盖点与执行命令", "移动端对应 PHPUnit 用例在本机兼容命令下可执行", ], }, "pc_deeper": { "focus_key": "pc_deeper", "files_to_touch": [ "tests/apifox/pc/PcAdminGuardMatrixTest.php", "tests/apifox/pc/PcAdminContractValidationTest.php", "tests/apifox/README.md", ], "tests_to_run": [ "$(brew --prefix php@8.2)/bin/php /tmp/phpunit-11.phar --bootstrap tests/bootstrap.php tests/apifox/pc/PcAdminGuardMatrixTest.php tests/apifox/pc/PcAdminContractValidationTest.php", ], "done_definition": [ "至少补齐一条 PC 端 Role/User/Menu 更深成功链路或参数契约", "README 回写新增覆盖点与执行命令", "PC 端对应 PHPUnit 用例在本机兼容命令下可执行", ], }, } def http_json( method: str, url: str, headers: dict[str, str], params: dict[str, Any] | None = None, payload: Any | None = None, ) -> Any: if params: url = f"{url}?{parse.urlencode(params)}" data = None if payload is not None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {**headers, "Content-Type": "application/json"} req = request.Request(url=url, data=data, headers=headers, method=method) with request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) def safe_http_json(*args: Any, **kwargs: Any) -> Any: try: return http_json(*args, **kwargs) except error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"HTTP {exc.code} {exc.reason}: {body}") from exc def read_lines(path: Path, max_lines: int) -> list[str]: if not path.exists(): return [f"[missing] {path}"] lines = path.read_text(encoding="utf-8").splitlines() return lines[:max_lines] def extract_readme_todos(lines: Iterable[str]) -> list[str]: keep: list[str] = [] for line in lines: striped = line.strip() if striped.startswith("- [ ]"): keep.append(striped) return keep[:12] def list_focus_files(repo_root: Path) -> list[str]: apifox_root = repo_root / "tests" / "apifox" paths: list[Path] = [] for pattern in ( "Support/*.php", "openapi/*Test.php", "pc/*Test.php", "mobile/*Test.php", "http/*.sh", ): paths.extend(sorted(apifox_root.glob(pattern))) rels = [ p.relative_to(repo_root).as_posix() for p in paths if p.is_file() ] return rels[:30] def build_prompt(repo_root: Path, sandbox_root: str, readme_rel: str, todo_rel: str) -> str: readme_path = repo_root / readme_rel todo_path = repo_root / todo_rel readme_lines = read_lines(readme_path, 220) todo_lines = read_lines(todo_path, 180) todos = extract_readme_todos(readme_lines) readme_excerpt = "\n".join(readme_lines[:120]) todo_excerpt = "\n".join(todo_lines[:80]) todo_section = "\n".join(todos) if todos else "- [ ] 未从 README 抽取到未完成项，请仅基于真实文件决定最小闭环" return f"""你现在在 OpenHands sandbox 中继续同一条 conversation。仓库根目录是 {sandbox_root}。 不要访问或引用 /Users/jack/... 绝对路径。 你只能基于以下真实上下文，输出“继续让 Codex 工作所需的最小 brief”，不要替 Codex 决定文件清单、测试命令或完成定义。 【tests/apifox/README.md 摘要】 {readme_excerpt} 【README 中仍未完成的事项】 {todo_section} 【整改 TODO 摘要】 {todo_excerpt} 硬约束： - 只输出 brief，不要寒暄，不要解释。 - 输出字段固定为：goal、scope、in_scope、out_of_scope。 - 任务粒度要小，只选一个最小闭环，不要把多个方向混在一张卡里。 - 不要输出 files_to_touch、tests_to_run、done_definition。 - 不要虚构文件名、命令、覆盖率数字或环境变量文件。 - 目标必须能直接交给 Codex 在当前仓库执行。 """ def extract_section_items(raw: str, section_name: str) -> list[str]: lines = raw.splitlines() items: list[str] = [] in_section = False for raw in lines: line = raw.strip() if not line: continue if line.startswith(section_name): in_section = True continue if in_section and any( line.startswith(prefix) for prefix in ( *tuple(f"{name}:" for name in BRIEF_FIELDS), "files_to_touch:", "tests_to_run:", "done_definition:", ) ): break if in_section and line.startswith("- "): items.append(line[2:].strip()) return items def parse_brief(message: str) -> dict[str, Any]: result: dict[str, Any] = {} lines = message.splitlines() simple_fields = {"goal", "scope"} for field in BRIEF_FIELDS: if field in simple_fields: value = "" for line in lines: if line.strip().startswith(f"{field}:"): value = line.split(":", 1)[1].strip() break result[field] = value else: result[field] = extract_section_items(message, f"{field}:") return result def validate_brief(message: str) -> dict[str, Any]: parsed = parse_brief(message) missing = [] for field in BRIEF_FIELDS: value = parsed.get(field) if isinstance(value, list): if not value: missing.append(field) elif not value: missing.append(field) forbidden_fields = [ name for name in ("files_to_touch", "tests_to_run", "done_definition") if f"{name}:" in message ] return { "brief": parsed, "missing_fields": missing, "forbidden_fields": forbidden_fields, "has_required_brief": not missing, "is_valid": not missing and not forbidden_fields, } def infer_focus_key(message: str) -> str: lowered = message.lower() if "http" in lowered or "smoke" in lowered or "联调" in lowered: return "http_real_flow" if "/m/" in lowered or "mobile" in lowered or "dishes" in lowered or "datemenu" in lowered or "staff" in lowered: return "mobile_deeper" return "pc_deeper" def synthesize_execution_card( validation: dict[str, Any], repo_root: Path, sandbox_root: str, ) -> dict[str, Any]: brief = validation["brief"] focus_key = infer_focus_key( "\n".join( [ brief.get("goal", ""), brief.get("scope", ""), *brief.get("in_scope", []), *brief.get("out_of_scope", []), ] ) ) preset = LOCAL_FOCUS_PRESETS[focus_key] return { "focus_key": focus_key, "goal": brief["goal"], "scope": brief["scope"], "in_scope": brief["in_scope"], "out_of_scope": brief["out_of_scope"], "files_to_touch": [ { "repo_relative": rel, "sandbox_path": f"{sandbox_root}/{rel}", "local_path": str((repo_root / rel).resolve()), } for rel in preset["files_to_touch"] ], "tests_to_run": preset["tests_to_run"], "done_definition": preset["done_definition"], } def latest_agent_message(items: list[dict[str, Any]], since_ts: str | None = None) -> dict[str, Any] | None: for item in items: if item.get("source") != "agent" or item.get("kind") != "MessageEvent": continue if since_ts and item.get("timestamp", "") <= since_ts: continue return item return None def message_text(item: dict[str, Any]) -> str: chunks = [] for chunk in item.get("llm_message", {}).get("content", []): if chunk.get("type") == "text": chunks.append(chunk.get("text", "")) return "".join(chunks).strip() def fetch_events(agent_url: str, conversation_id: str, headers: dict[str, str], limit: int = 12) -> dict[str, Any]: return safe_http_json( "GET", f"{agent_url}/api/conversations/{conversation_id}/events/search", headers=headers, params={"limit": limit, "sort_order": "TIMESTAMP_DESC"}, ) def fetch_conversation(agent_url: str, conversation_id: str, headers: dict[str, str]) -> dict[str, Any]: return safe_http_json( "GET", f"{agent_url}/api/conversations/{conversation_id}", headers=headers, ) def send_prompt(agent_url: str, conversation_id: str, headers: dict[str, str], prompt: str) -> None: payload = { "role": "user", "content": [{"type": "text", "text": prompt}], "run": True, } safe_http_json( "POST", f"{agent_url}/api/conversations/{conversation_id}/events", headers=headers, payload=payload, ) def print_result(data: dict[str, Any]) -> None: print(json.dumps(data, ensure_ascii=False, indent=2)) def main() -> int: parser = argparse.ArgumentParser(description="Send OpenHands brief prompts and synthesize a local Codex execution card.") parser.add_argument("--agent-url", required=True, help="Example: http://localhost:37511") parser.add_argument("--conversation-id", required=True, help="OpenHands agent-server conversation id") parser.add_argument("--session-api-key", required=True, help="X-Session-API-Key value") parser.add_argument("--repo-root", default=".", help="Local repo root used to build grounded context") parser.add_argument("--sandbox-root", default=DEFAULT_SANDBOX_ROOT, help="Repo root path inside OpenHands sandbox") parser.add_argument("--readme", default=DEFAULT_README, help="Repo-relative README path") parser.add_argument("--todo", default=DEFAULT_TODO, help="Repo-relative TODO path") parser.add_argument("--poll-seconds", type=int, default=5, help="Polling interval after sending a prompt") parser.add_argument("--timeout-seconds", type=int, default=180, help="Timeout waiting for a new agent message") parser.add_argument("--latest-only", action="store_true", help="Only inspect the latest agent message without sending a new prompt") args = parser.parse_args() repo_root = Path(args.repo_root).resolve() headers = {"X-Session-API-Key": args.session_api_key} if args.latest_only: conversation = fetch_conversation(args.agent_url, args.conversation_id, headers) events = fetch_events(args.agent_url, args.conversation_id, headers) latest = latest_agent_message(events.get("items", [])) latest_text = message_text(latest) if latest else None validation = validate_brief(latest_text) if latest_text else None print_result( { "mode": "latest_only", "execution_status": conversation.get("execution_status"), "last_user_message_id": conversation.get("last_user_message_id"), "latest_agent_message_timestamp": latest.get("timestamp") if latest else None, "latest_agent_message": latest_text, "validation": validation, "execution_card": synthesize_execution_card(validation, repo_root, args.sandbox_root) if validation and validation.get("has_required_brief") else None, } ) if not latest: return 1 return 0 if validation and validation.get("has_required_brief") else 3 before = fetch_events(args.agent_url, args.conversation_id, headers) previous_latest = latest_agent_message(before.get("items", [])) since_ts = previous_latest.get("timestamp") if previous_latest else None prompt = build_prompt(repo_root, args.sandbox_root, args.readme, args.todo) send_prompt(args.agent_url, args.conversation_id, headers, prompt) deadline = time.time() + args.timeout_seconds while time.time() < deadline: conversation = fetch_conversation(args.agent_url, args.conversation_id, headers) events = fetch_events(args.agent_url, args.conversation_id, headers) latest = latest_agent_message(events.get("items", []), since_ts=since_ts) if latest: latest_text = message_text(latest) validation = validate_brief(latest_text) print_result( { "mode": "send_and_wait", "status": "ok" if validation.get("is_valid") else ("brief_with_extra_fields" if validation.get("has_required_brief") else "invalid_brief"), "execution_status": conversation.get("execution_status"), "last_user_message_id": conversation.get("last_user_message_id"), "agent_message_timestamp": latest.get("timestamp"), "agent_message": latest_text, "validation": validation, "execution_card": synthesize_execution_card(validation, repo_root, args.sandbox_root) if validation.get("has_required_brief") else None, } ) return 0 if validation.get("has_required_brief") else 3 time.sleep(args.poll_seconds) conversation = fetch_conversation(args.agent_url, args.conversation_id, headers) events = fetch_events(args.agent_url, args.conversation_id, headers) print_result( { "mode": "send_and_wait", "status": "timeout", "execution_status": conversation.get("execution_status"), "last_user_message_id": conversation.get("last_user_message_id"), "latest_seen_agent_message_timestamp": since_ts, "latest_seen_agent_message": message_text(previous_latest) if previous_latest else None, "latest_event_timestamps": [item.get("timestamp") for item in events.get("items", [])[:6]], } ) return 2 if __name__ == "__main__": sys.exit(main())