824 lines
35 KiB
Python
824 lines
35 KiB
Python
# ============================================================
|
||
# 兜底车道编排器 v1（moduleId: core-fallback-lane, 可重生 ✅）
|
||
# 《Pi-Agent兜底能力详细方案》§4.1/§4.6 + P1-DESIGN §2.1/§3/§5：
|
||
# 意图未识别（assistant.reply/unknown）时，拉起 Pi headless 做一次
|
||
# **只读分析兜底**（草稿语义，不写世界），失败显式回退原话术。
|
||
# - 三重熔断：超时 / 步数上限 / 输出体量上限，触发即杀进程树并显式判败；
|
||
# - 成败只看事件流 stopReason，绝不相信进程退出码（P0 实测坑：pi 恒退 0）；
|
||
# - 环境白名单清洗 + PI_CODING_AGENT_DIR 配置圈禁 + taskkill 进程树回收；
|
||
# - 模型端点协商（GET /models，避开 P0 踩过的 404 坑），结果缓存 300s。
|
||
# 吸收 poc/pi-fallback/orchestrator.py 设计但**产品级重写，不 import poc**。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import queue
|
||
import shutil
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
import urllib.request
|
||
import uuid
|
||
from collections.abc import Callable, Iterator
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# AgentRunner：给定 (task, work_dir) 产出 JSONL 事件 dict 迭代器。
|
||
# 真实实现 = pi headless 子进程（build_pi_runner）；fake runner 只允许测试注入。
|
||
AgentRunner = Callable[[str, Path], Iterator[dict]]
|
||
|
||
|
||
class FallbackUnavailable(Exception):
|
||
"""运行时不可用（无 node / 无 pi / 模型协商失败 / 无模型 key）。一律显式失败。"""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 集中配置（改行为只改这里 + 环境变量）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@dataclass
|
||
class FallbackConfig:
|
||
"""集中配置。全部为类默认值，由 from_env() 覆盖。"""
|
||
timeout_sec: float = 90.0 # 闸 1：单次运行超时（chat 同步预算）
|
||
max_steps: int = 30 # 闸 2：工具调用步数上限
|
||
max_output_bytes: int = 2 * 1024 * 1024 # 闸 3：assistant 输出累计体量（2MiB）
|
||
poll_interval_sec: float = 1.0 # 读事件流轮询间隔（进程挂起也能被闸 1 抓到）
|
||
model: str = "" # 显式模型（APS_FALLBACK_MODEL）；空 = 协商
|
||
pi_cli: str = "" # pi cli.js 路径；空 = 默认解析
|
||
pi_home: str = "" # PI_CODING_AGENT_DIR；空 = <run根>/pi-home
|
||
node_bin: str = "node" # APS_FALLBACK_NODE 可覆盖
|
||
tools: str = "read,grep,find,ls" # pi 启动工具白名单（L1 第一道墙，只读四件套）
|
||
|
||
@classmethod
|
||
def from_env(cls) -> FallbackConfig:
|
||
def _float(name: str, default: float) -> float:
|
||
try:
|
||
return float(os.environ.get(name, "") or default)
|
||
except ValueError:
|
||
return default
|
||
|
||
def _int(name: str, default: int) -> int:
|
||
try:
|
||
return int(os.environ.get(name, "") or default)
|
||
except ValueError:
|
||
return default
|
||
|
||
# P1 复用 P0 安装；打包（桌面 sidecar）留后续阶段，路径必须可配置。
|
||
repo_root = Path(__file__).resolve().parents[2]
|
||
default_cli = (repo_root / "poc" / "pi-fallback" / "runtime" / "node_modules"
|
||
/ "@mariozechner" / "pi-coding-agent" / "dist" / "cli.js")
|
||
return cls(
|
||
timeout_sec=_float("APS_FALLBACK_TIMEOUT_SEC", 90.0),
|
||
max_steps=_int("APS_FALLBACK_MAX_STEPS", 30),
|
||
max_output_bytes=_int("APS_FALLBACK_MAX_OUTPUT_BYTES", 2 * 1024 * 1024),
|
||
model=(os.environ.get("APS_FALLBACK_MODEL") or "").strip(),
|
||
pi_cli=(os.environ.get("APS_FALLBACK_PI_CLI") or "").strip() or str(default_cli),
|
||
pi_home=(os.environ.get("APS_FALLBACK_PI_HOME") or "").strip(),
|
||
node_bin=(os.environ.get("APS_FALLBACK_NODE") or "").strip() or "node",
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class FallbackOutcome:
|
||
"""一次兜底运行的最终判定。ok 只由 stopReason=="stop" 且凭证校验通过决定。"""
|
||
run_id: str
|
||
ok: bool
|
||
stop_reason: str = "" # "stop" / "error" / "breaker:timeout(...)" /
|
||
# "breaker:max_steps(...)" / "breaker:max_output(...)" /
|
||
# "harness_error" / "unavailable:<原因>" / "forged_citation"
|
||
error_message: str = ""
|
||
steps: int = 0
|
||
output_bytes: int = 0
|
||
elapsed_sec: float = 0.0
|
||
report_text: str = "" # Pi 最终 assistant 文本（= outbox/report.md 内容）
|
||
run_dir: str = ""
|
||
citation_check: dict = field(default_factory=dict)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# FF-01 开关查询（默认关语义，单一事实源 = feature_flags.load_feature_flags）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def fallback_feature_enabled() -> bool:
|
||
"""fallback 键显式 true 才为 True（默认关）。任何异常 → False（宁可误关不可误开）。"""
|
||
try:
|
||
from server.agent_core.feature_flags import load_feature_flags
|
||
|
||
flags = load_feature_flags()
|
||
return bool(flags["features"]["fallback"]["enabled"])
|
||
except Exception: # noqa: BLE001 - 设计要求：任何异常 → False（宁可误关不可误开）
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# run 目录与 L4 环境清洗
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def fallback_root() -> Path:
|
||
"""run 根目录：APS_FALLBACK_DIR 或 path_under_data("fallback")（与 aps_home 口径一致）。"""
|
||
configured = (os.environ.get("APS_FALLBACK_DIR") or "").strip()
|
||
if configured:
|
||
return Path(configured).expanduser().resolve()
|
||
from server.aps_home import path_under_data
|
||
|
||
return path_under_data("fallback")
|
||
|
||
|
||
def _pi_home(config: FallbackConfig) -> Path:
|
||
return Path(config.pi_home).expanduser().resolve() if config.pi_home \
|
||
else fallback_root() / "pi-home"
|
||
|
||
|
||
def new_run_id() -> str:
|
||
""""fb-" + 时间戳 + uuid4 短串。贯穿审计/calls.jsonl/run 目录。"""
|
||
return time.strftime("fb-%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:6]
|
||
|
||
|
||
def create_run_dirs(run_id: str, config: FallbackConfig) -> dict[str, Path]:
|
||
"""建 L2 三区：{"root","inbox","work","outbox"}，并确保 pi-home 配置圈禁目录存在。"""
|
||
root = fallback_root() / run_id
|
||
dirs = {
|
||
"root": root,
|
||
"inbox": root / "inbox",
|
||
"work": root / "work",
|
||
"outbox": root / "outbox",
|
||
}
|
||
for p in dirs.values():
|
||
p.mkdir(parents=True, exist_ok=True)
|
||
_pi_home(config).mkdir(parents=True, exist_ok=True)
|
||
return dirs
|
||
|
||
|
||
# 子进程环境变量白名单（L4 进程层围墙）
|
||
_ENV_WHITELIST = (
|
||
"PATH", "PATHEXT", "SYSTEMROOT", "SYSTEMDRIVE", "WINDIR", "COMSPEC",
|
||
"TEMP", "TMP", "USERPROFILE", "APPDATA", "LOCALAPPDATA", "HOME",
|
||
"NODE_OPTIONS",
|
||
)
|
||
# 显式剥离名单（即使同名出现在白名单也再剥一层，双保险）
|
||
_ENV_STRIP_PREFIXES = ("CONDA_", "PYTHON", "PIP_", "VIRTUAL_ENV")
|
||
|
||
|
||
def build_child_env(config: FallbackConfig, extra: dict | None = None) -> dict:
|
||
"""L4 环境清洗：白名单制，剥离 CONDA_*/PYTHON*/PIP_*/VIRTUAL_ENV*；
|
||
强制 PI_CODING_AGENT_DIR=config.pi_home。extra 用于注入 LLM_API_KEY
|
||
（值只进子进程内存，绝不打印/落盘）。"""
|
||
env: dict = {}
|
||
for key in _ENV_WHITELIST:
|
||
if key in os.environ and not key.upper().startswith(_ENV_STRIP_PREFIXES):
|
||
env[key] = os.environ[key]
|
||
env["PI_CODING_AGENT_DIR"] = str(_pi_home(config))
|
||
if extra:
|
||
for k, v in extra.items():
|
||
if not k.upper().startswith(_ENV_STRIP_PREFIXES):
|
||
env[k] = v
|
||
return env
|
||
|
||
|
||
def kill_process_tree(pid: int) -> None:
|
||
"""Windows taskkill /PID /T /F；非 Windows 降级 os.killpg。异常吞掉（尽力回收）。"""
|
||
try:
|
||
if os.name == "nt":
|
||
subprocess.run(
|
||
["taskkill", "/PID", str(pid), "/T", "/F"],
|
||
capture_output=True, timeout=30, check=False,
|
||
)
|
||
else:
|
||
import signal
|
||
|
||
os.killpg(pid, signal.SIGKILL)
|
||
except Exception: # noqa: BLE001, S110 - 尽力回收：杀进程树失败不阻断失败判定
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 模型端点协商（P0 的 404 坑对策）+ pi-home/models.json
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_MODEL_CACHE_TTL_SEC = 300.0
|
||
_MODEL_CACHE: dict[str, Any] = {"model": None, "note": "", "ts": 0.0}
|
||
_PROVIDER_NAME = "aps-fallback"
|
||
|
||
|
||
def resolve_model(config: FallbackConfig) -> str | None:
|
||
"""模型协商。返回 "<provider>/<model>" 或 None（不可用）。
|
||
|
||
顺序：显式 config.model → 直接用（操作员显式负责，不探测）；
|
||
缺 LLM_BASE_URL/LLM_API_KEY → None；GET /models 协商（LLM_MODEL 不在清单
|
||
则取第一个并显式记录协商说明）；进程内缓存 300s。
|
||
"""
|
||
if config.model:
|
||
return config.model
|
||
base_url = (os.environ.get("LLM_BASE_URL") or "").strip().rstrip("/")
|
||
api_key = (os.environ.get("LLM_API_KEY") or "").strip()
|
||
if not base_url or not api_key:
|
||
return None
|
||
now = time.monotonic()
|
||
if _MODEL_CACHE["model"] and now - _MODEL_CACHE["ts"] < _MODEL_CACHE_TTL_SEC:
|
||
return str(_MODEL_CACHE["model"])
|
||
try:
|
||
req = urllib.request.Request(
|
||
f"{base_url}/models",
|
||
headers={"Authorization": f"Bearer {api_key}"},
|
||
)
|
||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||
payload = json.loads(resp.read().decode("utf-8"))
|
||
ids = [m.get("id") for m in (payload.get("data") or []) if m.get("id")]
|
||
except Exception: # noqa: BLE001 - 网络/解析错误异构，统一归并为「不可用」
|
||
return None
|
||
if not ids:
|
||
return None
|
||
wanted = (os.environ.get("LLM_MODEL") or "").strip()
|
||
note = ""
|
||
if wanted and wanted in ids:
|
||
chosen = wanted
|
||
else:
|
||
chosen = ids[0]
|
||
note = f"配置的 LLM_MODEL={wanted or '(空)'} 不可用，协商改用 {chosen}"
|
||
model = f"{_PROVIDER_NAME}/{chosen}"
|
||
_MODEL_CACHE.update({"model": model, "note": note, "ts": now})
|
||
return model
|
||
|
||
|
||
def _write_models_json(config: FallbackConfig, model: str) -> None:
|
||
"""把协商结果写 pi-home/models.json。apiKey 只写环境变量名引用 "LLM_API_KEY"，
|
||
绝不落 key 明文（P0 已实测此机制有效）。"""
|
||
model_id = model.split("/", 1)[1] if "/" in model else model
|
||
base_url = (os.environ.get("LLM_BASE_URL") or "").strip().rstrip("/")
|
||
doc = {
|
||
"providers": {
|
||
_PROVIDER_NAME: {
|
||
"baseUrl": base_url,
|
||
"api": "openai-completions",
|
||
"apiKey": "LLM_API_KEY", # 环境变量名引用，非明文
|
||
"authHeader": True,
|
||
"models": [{
|
||
"id": model_id,
|
||
"name": model_id,
|
||
"reasoning": False,
|
||
"input": ["text"],
|
||
"contextWindow": 262144,
|
||
"maxTokens": 8192,
|
||
"cost": {"input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0},
|
||
}],
|
||
},
|
||
},
|
||
}
|
||
pi_home = _pi_home(config)
|
||
pi_home.mkdir(parents=True, exist_ok=True)
|
||
(pi_home / "models.json").write_text(
|
||
json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# L1 第二道墙：守卫扩展（bash/edit 全 block、文件工具限 run 目录）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_GUARD_TS_TEMPLATE = """// AUTO-GENERATED by fallback_lane.py — 守卫扩展（L1/L2 围墙的 pi 侧执行点）。
|
||
// pi.on("tool_call") 返回 {{ block: true, reason }} 即可在工具执行前拦截（P0 已实测）。
|
||
import fs from "node:fs";
|
||
import path from "node:path";
|
||
|
||
const RUN_ROOT = path.normalize("{RUN_ROOT_POSIX}");
|
||
const BLOCKLOG = path.join(RUN_ROOT, "guard-blocked-calls.jsonl");
|
||
|
||
function inRunRoot(p: string): boolean {{
|
||
const abs = path.resolve(process.cwd(), p);
|
||
const norm = path.normalize(abs);
|
||
return norm === RUN_ROOT || norm.startsWith(RUN_ROOT + path.sep);
|
||
}}
|
||
|
||
function deny(toolName: string, toolCallId: string, reason: string, input: any) {{
|
||
fs.appendFileSync(
|
||
BLOCKLOG,
|
||
JSON.stringify({{ ts: new Date().toISOString(), toolName, toolCallId, reason, input }}) + "\\n",
|
||
);
|
||
return {{ block: true, reason }};
|
||
}}
|
||
|
||
export default function (pi: any) {{
|
||
pi.on("tool_call", async (event: any, _ctx: any) => {{
|
||
const name: string = event.toolName;
|
||
const input: any = event.input || {{}};
|
||
|
||
// 1) bash / edit / write：P1 全禁（只读兜底，无写面、无任意命令执行面）。
|
||
if (name === "bash" || name === "edit" || name === "write") {{
|
||
return deny(name, event.toolCallId, "disabled by fallback guard (read-only lane)", input);
|
||
}}
|
||
|
||
// 2) 文件类工具：路径必须落在 run 目录内（L2 圈禁的 pi 侧执行点）。
|
||
const fileTools = ["read", "grep", "find", "ls"];
|
||
if (fileTools.includes(name)) {{
|
||
const p: string = String(input.path || input.pattern || ".");
|
||
if (!inRunRoot(p)) return deny(name, event.toolCallId, "path escapes run root", input);
|
||
}}
|
||
// 放行
|
||
}});
|
||
}}
|
||
"""
|
||
|
||
|
||
def write_guard_extension(run_dir: Path) -> Path:
|
||
"""生成 guard-<runId>.ts（L1 第二道墙）。返回路径供 pi `-e` 加载，随运行归档。"""
|
||
run_dir = Path(run_dir).resolve()
|
||
content = _GUARD_TS_TEMPLATE.format(RUN_ROOT_POSIX=run_dir.as_posix())
|
||
out = run_dir / f"guard-{run_dir.name}.ts"
|
||
out.write_text(content, encoding="utf-8")
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 真实 pi headless runner
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _resolve_node(node_bin: str) -> str | None:
|
||
"""node 解析顺序（P1 真实冒烟坑 B 对策）：
|
||
1) 显式配置（APS_FALLBACK_NODE / config.node_bin 非默认值）最高优先级，原样命中；
|
||
2) 默认 "node" 且 Windows 时优先 shutil.which("node.exe")——避开 PATH 中先于
|
||
node.exe 命中的 node.CMD 垫片（Popen 起 .cmd 引号语义会炸，pi 秒败）；
|
||
3) 其余情况回退 shutil.which("node")。
|
||
"""
|
||
if node_bin != "node":
|
||
return shutil.which(node_bin)
|
||
if os.name == "nt":
|
||
return shutil.which("node.exe") or shutil.which("node")
|
||
return shutil.which(node_bin)
|
||
|
||
|
||
def build_pi_runner(config: FallbackConfig) -> AgentRunner:
|
||
"""构造真实 pi headless runner（读线程+queue 轮询、心跳事件、finally 杀进程树）。
|
||
|
||
Raises FallbackUnavailable：node/pi_cli 缺失或模型协商失败——调用方把它当
|
||
「不可用」显式失败处理。
|
||
"""
|
||
pi_cli = Path(config.pi_cli) if config.pi_cli else None
|
||
if not pi_cli or not pi_cli.is_file():
|
||
raise FallbackUnavailable(f"pi cli 不存在: {config.pi_cli or '(未配置)'}")
|
||
node = _resolve_node(config.node_bin)
|
||
if not node:
|
||
raise FallbackUnavailable(f"node 不在 PATH（APS_FALLBACK_NODE={config.node_bin}）")
|
||
api_key = (os.environ.get("LLM_API_KEY") or "").strip()
|
||
if not (os.environ.get("LLM_BASE_URL") or "").strip() or not api_key:
|
||
raise FallbackUnavailable("无模型配置（LLM_BASE_URL/LLM_API_KEY 缺失）")
|
||
model = resolve_model(config)
|
||
if not model:
|
||
raise FallbackUnavailable("模型协商失败（GET /models 不可达或清单为空）")
|
||
_write_models_json(config, model)
|
||
|
||
def runner(task: str, work_dir: Path) -> Iterator[dict]:
|
||
guard = write_guard_extension(work_dir.parent)
|
||
cmd = [
|
||
node, str(pi_cli),
|
||
"-p", "--mode", "json",
|
||
"--model", model,
|
||
"--tools", config.tools,
|
||
"-e", str(guard),
|
||
task,
|
||
]
|
||
env = build_child_env(config, extra={"LLM_API_KEY": api_key})
|
||
proc = subprocess.Popen(
|
||
cmd, cwd=str(work_dir), env=env,
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0,
|
||
)
|
||
q: queue.Queue[str | None] = queue.Queue()
|
||
|
||
def reader() -> None:
|
||
try:
|
||
for raw in proc.stdout:
|
||
q.put(raw.decode("utf-8", errors="replace"))
|
||
finally:
|
||
q.put(None) # EOF 哨兵
|
||
|
||
t = threading.Thread(target=reader, daemon=True)
|
||
t.start()
|
||
|
||
killed = False
|
||
try:
|
||
while True:
|
||
try:
|
||
line = q.get(timeout=config.poll_interval_sec)
|
||
except queue.Empty:
|
||
if proc.poll() is not None and not t.is_alive():
|
||
break
|
||
# 让上层熔断循环有机会检查超时：产出一个心跳事件
|
||
yield {"type": "harness_heartbeat"}
|
||
continue
|
||
if line is None:
|
||
break
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
yield json.loads(line)
|
||
except json.JSONDecodeError:
|
||
yield {"type": "harness_nonjson", "raw": line[:500]}
|
||
finally:
|
||
# 进程回收：编排器熔断 break 时会 close 本生成器，
|
||
# 这里只杀进程树、不许再 yield（closing 中 yield 会 RuntimeError）。
|
||
if proc.poll() is None:
|
||
killed = True
|
||
kill_process_tree(proc.pid)
|
||
try:
|
||
proc.wait(timeout=15)
|
||
except Exception: # noqa: BLE001, S110 - 进程已杀，wait 失败无需处理
|
||
pass
|
||
# 以下仅正常 EOF 退出时可达（被 close 时 GeneratorExit 直接传播出去）
|
||
if killed:
|
||
yield {"type": "harness_killed", "pid": proc.pid}
|
||
err = proc.stderr.read().decode("utf-8", errors="replace") if proc.stderr else ""
|
||
if err.strip():
|
||
yield {"type": "harness_stderr", "text": err[-2000:]}
|
||
|
||
return runner
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 事件解析辅助
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _extract_text_delta(event: dict) -> str:
|
||
"""从 message_update/message_end 事件取 assistant 文本增量（用于输出体量闸）。"""
|
||
if event.get("type") == "message_update":
|
||
delta = event.get("delta") or {}
|
||
if isinstance(delta, dict):
|
||
return str(delta.get("text") or "")
|
||
if event.get("type") == "message_end":
|
||
msg = event.get("message") or {}
|
||
for part in (msg.get("content") or []):
|
||
if isinstance(part, dict) and part.get("type") == "text":
|
||
return str(part.get("text") or "")
|
||
return ""
|
||
|
||
|
||
def _extract_stop(event: dict) -> tuple[str | None, str | None, str | None]:
|
||
"""从事件里提取 (stopReason, errorMessage, 文本)。
|
||
兼容两个位置：message_end.message.stopReason 与 agent_end.messages 最后一条。"""
|
||
etype = event.get("type")
|
||
msg = None
|
||
if etype == "message_end":
|
||
msg = event.get("message") or {}
|
||
elif etype == "agent_end":
|
||
msgs = event.get("messages") or []
|
||
assistants = [m for m in msgs if m.get("role") == "assistant"]
|
||
msg = assistants[-1] if assistants else None
|
||
if not msg:
|
||
return None, None, None
|
||
txt = ""
|
||
for part in (msg.get("content") or []):
|
||
if isinstance(part, dict) and part.get("type") == "text":
|
||
txt += str(part.get("text") or "")
|
||
return msg.get("stopReason"), msg.get("errorMessage"), txt
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主循环：三重熔断 + stopReason 判定 + 事件落盘
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _run_events(
|
||
runner: AgentRunner,
|
||
task: str,
|
||
dirs: dict[str, Path],
|
||
config: FallbackConfig,
|
||
run_id: str,
|
||
on_tool_event: Callable[[dict], None] | None = None,
|
||
) -> FallbackOutcome:
|
||
"""消费事件流，执行熔断与判定，落 events.jsonl / orchestrator.log。"""
|
||
run_dir = dirs["root"]
|
||
log_path = run_dir / "orchestrator.log"
|
||
events_path = run_dir / "events.jsonl"
|
||
|
||
def log(msg: str) -> None:
|
||
with open(log_path, "a", encoding="utf-8") as f:
|
||
f.write(f"[{time.strftime('%H:%M:%S')}] {msg}\n")
|
||
|
||
outcome = FallbackOutcome(run_id=run_id, ok=False, run_dir=str(run_dir))
|
||
t0 = time.monotonic()
|
||
log(f"run_id={run_id} task={task[:120]!r}")
|
||
log(f"config: {config}")
|
||
if _MODEL_CACHE.get("note"):
|
||
log(f"模型协商：{_MODEL_CACHE['note']}")
|
||
|
||
breaker_tripped: str | None = None
|
||
last_assistant_text = ""
|
||
stop_reason = ""
|
||
error_message = ""
|
||
|
||
def _write_event(evf, event: dict) -> None:
|
||
try:
|
||
evf.write(json.dumps(event, ensure_ascii=False) + "\n")
|
||
except Exception: # noqa: BLE001 - 事件落盘绝不能中断主循环，降级占位记录
|
||
evf.write(json.dumps({"type": "unserializable_event"}) + "\n")
|
||
|
||
try:
|
||
with open(events_path, "w", encoding="utf-8") as evf:
|
||
for event in runner(task, dirs["work"]):
|
||
elapsed = time.monotonic() - t0
|
||
|
||
# —— 闸 1：超时 ——
|
||
if elapsed > config.timeout_sec:
|
||
breaker_tripped = f"breaker:timeout({elapsed:.1f}s>{config.timeout_sec}s)"
|
||
_write_event(evf, {"type": "breaker", "reason": breaker_tripped})
|
||
break
|
||
|
||
_write_event(evf, event)
|
||
etype = event.get("type", "")
|
||
|
||
# —— 步数统计 + 闸 2 ——
|
||
if etype == "tool_execution_start":
|
||
outcome.steps += 1
|
||
if on_tool_event:
|
||
on_tool_event(event)
|
||
if outcome.steps > config.max_steps:
|
||
breaker_tripped = f"breaker:max_steps({outcome.steps}>{config.max_steps})"
|
||
_write_event(evf, {"type": "breaker", "reason": breaker_tripped})
|
||
break
|
||
elif etype == "tool_execution_end" and on_tool_event:
|
||
on_tool_event(event)
|
||
|
||
# —— 输出体量统计 + 闸 3 ——
|
||
text_delta = _extract_text_delta(event)
|
||
if text_delta:
|
||
outcome.output_bytes += len(text_delta.encode("utf-8"))
|
||
if outcome.output_bytes > config.max_output_bytes:
|
||
breaker_tripped = (
|
||
f"breaker:max_output({outcome.output_bytes}>{config.max_output_bytes})"
|
||
)
|
||
_write_event(evf, {"type": "breaker", "reason": breaker_tripped})
|
||
break
|
||
|
||
# —— stopReason 判定（成败唯一权威）——
|
||
sr, em, txt = _extract_stop(event)
|
||
if sr:
|
||
stop_reason, error_message = sr, em or ""
|
||
if txt:
|
||
last_assistant_text = txt
|
||
|
||
if etype == "auto_retry_start":
|
||
log(f"auto_retry_start attempt={event.get('attempt')}")
|
||
|
||
if etype == "agent_end":
|
||
break
|
||
except Exception as exc: # noqa: BLE001 - runner 抛错/桥违规统一归并显式失败
|
||
stop_reason = "harness_error"
|
||
error_message = f"{type(exc).__name__}: {exc}"
|
||
log(f"HARNESS ERROR: {error_message}")
|
||
|
||
outcome.elapsed_sec = time.monotonic() - t0
|
||
outcome.report_text = last_assistant_text
|
||
|
||
if breaker_tripped:
|
||
outcome.stop_reason = breaker_tripped
|
||
outcome.error_message = "熔断触发，运行显式标记失败"
|
||
log(f"BREAKER TRIPPED: {breaker_tripped} -> failed")
|
||
elif stop_reason == "stop":
|
||
outcome.ok = True
|
||
outcome.stop_reason = "stop"
|
||
log(f"OK stopReason=stop steps={outcome.steps} out={outcome.output_bytes}B "
|
||
f"elapsed={outcome.elapsed_sec:.1f}s")
|
||
else:
|
||
outcome.stop_reason = stop_reason or "error:no_stop_reason"
|
||
outcome.error_message = error_message or "事件流未给出 stopReason=stop，按失败处理"
|
||
log(f"FAILED stopReason={outcome.stop_reason} err={outcome.error_message}")
|
||
return outcome
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 回复组装（精确文案契约，测试可断言）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _stop_reason_cn(stop_reason: str) -> str:
|
||
if stop_reason.startswith("breaker:timeout"):
|
||
return "运行超时（已触发超时熔断）"
|
||
if stop_reason.startswith("breaker:max_steps"):
|
||
return "工具调用步数超限（已触发步数熔断）"
|
||
if stop_reason.startswith("breaker:max_output"):
|
||
return "输出体量超限（已触发输出熔断）"
|
||
if stop_reason.startswith("unavailable:"):
|
||
return f"兜底运行时不可用（{stop_reason.split(':', 1)[1]}）"
|
||
if stop_reason == "forged_citation":
|
||
return "报告引用了不存在的凭证（按伪造成果判失败）"
|
||
if stop_reason == "harness_error":
|
||
return "编排器内部错误"
|
||
return stop_reason or "未知原因"
|
||
|
||
|
||
def _compose_success_reply(outcome: FallbackOutcome) -> str:
|
||
text = f"[智能兜底 · 草稿] run {outcome.run_id}\n\n"
|
||
if outcome.steps == 0:
|
||
text += "（Pi 本次未读取项目数据，以下为纯推理草稿）\n\n"
|
||
return (
|
||
text + outcome.report_text
|
||
+ "\n\n---\n以上为 Pi 只读分析草稿（未改动任何数据），凭证与过程见审计。"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接线入口（workflow.py assistant.reply/unknown 分支调用）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def propose_reply(
|
||
store,
|
||
session_id: str,
|
||
intent,
|
||
*,
|
||
actor: str = "planner",
|
||
runner: AgentRunner | None = None, # 测试注入点；None=真实 pi
|
||
config: FallbackConfig | None = None, # 测试注入点；None=from_env()
|
||
):
|
||
"""unknown/assistant.reply 分支的唯一接线入口。
|
||
|
||
返回语义：
|
||
- None → 未触发（开关关 / query 为空）：调用方走原话术，零副作用零审计；
|
||
- AgentReply → 已触发。成功=Pi 报告正文；失败=原话术+一行显式失败说明。
|
||
|
||
保证：本函数绝不抛出——内部所有异常归并为显式失败，聊天链路永远有回复。
|
||
"""
|
||
# 第一关：开关（False → 零副作用直接放行原路径）
|
||
if not fallback_feature_enabled():
|
||
return None
|
||
|
||
try:
|
||
query = str(intent.params.get("query") or intent.params.get("text") or "").strip()
|
||
if not query:
|
||
return None
|
||
hist = intent.params.get("_history") or []
|
||
config = config or FallbackConfig.from_env()
|
||
run_id = new_run_id()
|
||
|
||
# 运行时可用性（仅真实 runner 检查；注入 runner 为测试路径，跳过）
|
||
if runner is None:
|
||
try:
|
||
runner = build_pi_runner(config)
|
||
except Exception as exc: # noqa: BLE001 - 不可用统一归并显式失败（不装死）
|
||
outcome = FallbackOutcome(
|
||
run_id=run_id, ok=False,
|
||
stop_reason=f"unavailable:{exc}", error_message=str(exc))
|
||
_write_completion_audit(store, actor, outcome, query)
|
||
return await _compose_failure_reply(store, query, hist, session_id, outcome)
|
||
|
||
from server.integrations.pi_bridge import PiBridge, render_task_brief
|
||
|
||
dirs = create_run_dirs(run_id, config)
|
||
bridge = PiBridge(run_id, dirs["root"])
|
||
try:
|
||
snapshot_files = bridge.export_snapshot(store.data, dirs)
|
||
except Exception as exc: # noqa: BLE001 - 快照失败不阻断兜底，降级纯推理并显式记日志
|
||
snapshot_files = []
|
||
_append_run_log(dirs["root"], f"快照导出失败（继续纯推理）: {exc}")
|
||
task = render_task_brief(run_id=run_id, query=query, snapshot_files=snapshot_files)
|
||
|
||
on_tool_event = _make_tool_event_handler(store, bridge, run_id)
|
||
outcome = _run_events(runner, task, dirs, config, run_id,
|
||
on_tool_event=on_tool_event)
|
||
|
||
# 凭证校验：引用不存在的 callId = 伪造成果，物理判失败
|
||
if outcome.ok:
|
||
outcome.citation_check = bridge.validate_report_citations(outcome.report_text)
|
||
if not outcome.citation_check["valid"]:
|
||
outcome.ok = False
|
||
outcome.stop_reason = "forged_citation"
|
||
outcome.error_message = (
|
||
f"报告引用了不存在的 callId: {outcome.citation_check['missing']}")
|
||
elif not outcome.report_text.strip():
|
||
outcome.ok = False
|
||
outcome.stop_reason = "error:empty_report"
|
||
outcome.error_message = "stopReason=stop 但最终报告为空，按失败处理"
|
||
|
||
# 产物唯一出口 + 结果落盘
|
||
report_path = ""
|
||
if outcome.report_text:
|
||
report_path = str(dirs["outbox"] / "report.md")
|
||
(dirs["outbox"] / "report.md").write_text(outcome.report_text, encoding="utf-8")
|
||
outcome.citation_check = outcome.citation_check or {}
|
||
_write_result_json(dirs["root"], outcome)
|
||
|
||
_write_completion_audit(store, actor, outcome, query, report_path=report_path)
|
||
|
||
if outcome.ok:
|
||
from server.contracts import AgentReply
|
||
|
||
return AgentReply(text=_compose_success_reply(outcome))
|
||
return await _compose_failure_reply(store, query, hist, session_id, outcome)
|
||
except Exception: # noqa: BLE001 - 绝不抛出：意外异常回退原话术（用户无感知）
|
||
try:
|
||
from server.agent_core.assistant import reply as assistant_reply
|
||
|
||
q = str(intent.params.get("query") or intent.params.get("text") or "")
|
||
return await assistant_reply(store.data, q,
|
||
history=intent.params.get("_history") or [],
|
||
session_id=session_id)
|
||
except Exception: # noqa: BLE001 - 原话术也失败时返回 None，由调用方兜底
|
||
return None
|
||
|
||
|
||
def _append_run_log(run_dir: Path, msg: str) -> None:
|
||
"""追加一行 orchestrator.log（同步函数，避免在 async 接线入口里做阻塞 IO）。"""
|
||
with open(run_dir / "orchestrator.log", "a", encoding="utf-8") as f:
|
||
f.write(f"[{time.strftime('%H:%M:%S')}] {msg}\n")
|
||
|
||
|
||
def _write_result_json(run_dir: Path, outcome: FallbackOutcome) -> None:
|
||
"""落 result.json（含凭证校验结果；同步函数，理由同上）。"""
|
||
with open(run_dir / "result.json", "w", encoding="utf-8") as f:
|
||
json.dump(outcome.__dict__, f, ensure_ascii=False, indent=2, default=str)
|
||
|
||
|
||
# pi 内置工具 → 桥登记工具的映射（桥侧凭证签发；映射外工具出现即 ToolBridgeViolation，
|
||
# 经主循环 except 归并为 harness_error 显式失败——双保险，正常不会触达）
|
||
_PI_TOOL_MAP = {"read": "fs_read", "grep": "fs_read", "find": "fs_read", "ls": "fs_read"}
|
||
|
||
|
||
def _make_tool_event_handler(store, bridge, run_id: str) -> Callable[[dict], None]:
|
||
"""每个工具事件：签/补 callId 凭证 + 写 tool.run 审计（actor=pi-fallback:<runId>）。"""
|
||
from server.agent_core.audit import write_audit
|
||
|
||
pi_call_ids: dict[str, str] = {}
|
||
|
||
def on_tool_event(event: dict) -> None:
|
||
etype = event.get("type")
|
||
tool = str(event.get("toolName") or "")
|
||
pi_id = str(event.get("toolCallId") or "")
|
||
if etype == "tool_execution_start":
|
||
mapped = _PI_TOOL_MAP.get(tool)
|
||
if mapped is None:
|
||
from server.integrations.pi_bridge import ToolBridgeViolation
|
||
|
||
raise ToolBridgeViolation(f"pi 工具未在桥映射表登记: {tool}")
|
||
call_id = bridge.issue_call(
|
||
mapped,
|
||
params=event.get("args") or event.get("input"),
|
||
pi_tool_call_id=pi_id or None,
|
||
)
|
||
pi_call_ids[pi_id] = call_id
|
||
write_audit(
|
||
store.data, store.next_id,
|
||
actor=f"pi-fallback:{run_id}", category="TOOL", action="tool.run",
|
||
target={"type": "PI_TOOL", "id": f"{tool}/{call_id}"},
|
||
power="P0",
|
||
rationale={"runId": run_id, "piToolCallId": pi_id, "bridgeTool": mapped},
|
||
)
|
||
elif etype == "tool_execution_end":
|
||
call_id = pi_call_ids.get(pi_id)
|
||
if call_id:
|
||
bridge.complete_call(
|
||
call_id,
|
||
result=event.get("result") or event.get("output") or "",
|
||
ok=not event.get("isError"),
|
||
)
|
||
|
||
return on_tool_event
|
||
|
||
|
||
def _write_completion_audit(store, actor: str, outcome: FallbackOutcome, query: str,
|
||
report_path: str = "") -> None:
|
||
"""完成时 1 条审计（成败都写），随后 store.save()。"""
|
||
from server.agent_core.audit import write_audit
|
||
|
||
write_audit(
|
||
store.data, store.next_id,
|
||
actor=actor, category="TOOL", action="agent.fallback.propose",
|
||
target={"type": "FALLBACK_RUN", "id": outcome.run_id},
|
||
power="P1",
|
||
rationale={
|
||
"runId": outcome.run_id,
|
||
"queryDigest": hashlib.sha256(query.encode("utf-8")).hexdigest()[:16],
|
||
"stopReason": outcome.stop_reason,
|
||
"steps": outcome.steps,
|
||
"elapsedSec": round(outcome.elapsed_sec, 2),
|
||
"citationCheck": {
|
||
"cited": len(outcome.citation_check.get("cited") or []),
|
||
"missing": len(outcome.citation_check.get("missing") or []),
|
||
},
|
||
"runDir": outcome.run_dir,
|
||
"reportPath": report_path,
|
||
},
|
||
result="SUCCESS" if outcome.ok else "FAILED",
|
||
evidence_refs=[f"fallback-run:{outcome.run_id}"],
|
||
)
|
||
store.save()
|
||
|
||
|
||
async def _compose_failure_reply(store, query: str, hist, session_id: str,
|
||
outcome: FallbackOutcome):
|
||
"""失败 = 原 assistant_reply 话术全文 + 一行显式失败说明。"""
|
||
from server.agent_core.assistant import reply as assistant_reply
|
||
|
||
original = await assistant_reply(store.data, query, history=hist, session_id=session_id)
|
||
original.text += (
|
||
f"\n\n---\n（智能兜底本次未完成：{_stop_reason_cn(outcome.stop_reason)}，"
|
||
f"已记录审计 run {outcome.run_id}；你的数据未被改动）"
|
||
)
|
||
return original
|