692 lines
29 KiB
Python
692 lines
29 KiB
Python
# ============================================================
|
||
# P4 离线/降级 golden（Agent-H）：O-01..O-10。
|
||
# 全部确定性：fake runner / FakeStore / tmp 隔离 / monkeypatch；
|
||
# 零真实 LLM、零网络、零真实 node/pi 子进程。
|
||
# 本文件复制既有 golden 的最小 helper，不新增共享测试模块。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import subprocess
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from typing import ClassVar
|
||
|
||
import pytest
|
||
|
||
from server.agent_core import fallback_highrisk, fallback_lane, harness
|
||
from server.agent_core.feature_flags import load_feature_flags
|
||
from server.agent_core.providers import reset_provider
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.contracts import IntentResult
|
||
from server.state.checkpoints import CheckpointStore
|
||
from server.state.seed import seed_world
|
||
|
||
DEMO_PRODUCT = "CTRL-A"
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _isolate(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_FALLBACK_DIR", str(tmp_path / "fb"))
|
||
monkeypatch.setenv("APS_FEATURES_PATH", str(tmp_path / "features.json"))
|
||
monkeypatch.setenv("APS_FALLBACK_HIGHRISK_PATH", str(tmp_path / "fallback-highrisk.json"))
|
||
monkeypatch.delenv("LLM_API_KEY", raising=False)
|
||
monkeypatch.delenv("LLM_BASE_URL", raising=False)
|
||
monkeypatch.delenv("LLM_MODEL", raising=False)
|
||
monkeypatch.delenv("LLM_PROVIDER", raising=False)
|
||
reset_provider()
|
||
old_cache = dict(fallback_lane._MODEL_CACHE)
|
||
fallback_lane._MODEL_CACHE.update({"model": None, "note": "", "ts": 0.0})
|
||
from server.state import checkpoints as _cp_mod
|
||
from server.state import store as _store_mod
|
||
|
||
kept_scoped = dict(_store_mod._stores)
|
||
kept_cps = dict(_cp_mod._checkpoints)
|
||
yield
|
||
_store_mod._stores.clear()
|
||
_store_mod._stores.update(kept_scoped)
|
||
_cp_mod._checkpoints.clear()
|
||
_cp_mod._checkpoints.update(kept_cps)
|
||
fallback_lane._MODEL_CACHE.clear()
|
||
fallback_lane._MODEL_CACHE.update(old_cache)
|
||
reset_provider()
|
||
|
||
|
||
class FakeStore:
|
||
_KIND_TABLE: ClassVar[dict[str, str]] = {
|
||
"salesOrder": "salesOrders", "material": "materials",
|
||
"audit": "auditEvents", "importBatch": "importBatches"}
|
||
|
||
def __init__(self, tmp_path: Path):
|
||
self.data = seed_world()
|
||
self._counters: dict[str, int] = {}
|
||
self.tenant_uuid = "platform"
|
||
self.world_key = "default"
|
||
self.path = str(tmp_path / "world.json")
|
||
self.checkpoints = CheckpointStore(str(tmp_path / "checkpoints.json"))
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
if kind not in self._counters:
|
||
table = self._KIND_TABLE.get(kind)
|
||
self._counters[kind] = max(
|
||
(x.get("id", 0) for x in self.data.get(table, [])
|
||
if isinstance(x.get("id"), int)), default=0) if table else 0
|
||
self._counters[kind] += 1
|
||
return self._counters[kind]
|
||
|
||
def save(self) -> None:
|
||
pass
|
||
|
||
def restore(self, world: dict) -> None:
|
||
self.data = copy.deepcopy(world)
|
||
self._counters.clear()
|
||
|
||
|
||
class _FakeResponse:
|
||
def __init__(self, body: bytes):
|
||
self._body = body
|
||
|
||
def read(self) -> bytes:
|
||
return self._body
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *exc):
|
||
return False
|
||
|
||
|
||
def _cfg(tmp_path: Path, **kw) -> fallback_lane.FallbackConfig:
|
||
return fallback_lane.FallbackConfig(pi_home=str(tmp_path / "pi-home"), **kw)
|
||
|
||
|
||
def _write_features(tmp_path: Path, features: dict | None = None) -> None:
|
||
doc = {"version": 1, "features": {"fallback": True} if features is None else features}
|
||
(tmp_path / "features.json").write_text(json.dumps(doc, ensure_ascii=False), encoding="utf-8")
|
||
|
||
|
||
def _write_whitelist(tmp_path: Path, doc) -> None:
|
||
path = tmp_path / "fallback-highrisk.json"
|
||
if isinstance(doc, str):
|
||
path.write_text(doc, encoding="utf-8")
|
||
else:
|
||
path.write_text(json.dumps(doc, ensure_ascii=False), encoding="utf-8")
|
||
|
||
|
||
def _wl_doc(*, s4=True, s6=True, s7=True) -> dict:
|
||
scenarios = {
|
||
"S4": {"enabled": s4,
|
||
"intents": ["flex.simulate_due", "flex.compare",
|
||
"scenario.compare", "scenario.sensitivity"],
|
||
"roles": ["planner", "admin"]},
|
||
"S6": {"enabled": s6, "intents": ["mes.report"], "roles": ["planner", "admin"]},
|
||
"S7": {"enabled": s7,
|
||
"intents": ["agent.fallback.ops.config.apply",
|
||
"agent.fallback.policy.update"],
|
||
"roles": ["ops", "admin"]}}
|
||
return {"whitelistVersion": 1, "updatedAt": "2026-09-04T10:00:00",
|
||
"updatedBy": "offline-test", "scenarios": scenarios}
|
||
|
||
|
||
def _intent(query: str) -> IntentResult:
|
||
return IntentResult(intent="unknown", params={"query": query},
|
||
confidence=0.1, source="LLM")
|
||
|
||
|
||
def _fp(world: dict) -> str:
|
||
return harness.world_fingerprint(world)
|
||
|
||
|
||
def _fb_audits(store: FakeStore) -> list[dict]:
|
||
return [e for e in store.data.get("auditEvents") or []
|
||
if e.get("action") == "agent.fallback.propose"]
|
||
|
||
|
||
def _world_write_audits(store: FakeStore) -> list[dict]:
|
||
return [e for e in store.data.get("auditEvents") or []
|
||
if e.get("category") == "WORLD_WRITE"]
|
||
|
||
|
||
def _run_dirs(tmp_path: Path) -> list[Path]:
|
||
root = tmp_path / "fb"
|
||
if not root.is_dir():
|
||
return []
|
||
return [p for p in root.iterdir() if p.is_dir() and p.name != "pi-home"]
|
||
|
||
|
||
def _run_dir_of(tmp_path: Path) -> Path:
|
||
runs = _run_dirs(tmp_path)
|
||
assert len(runs) == 1
|
||
return runs[0]
|
||
|
||
|
||
def _stop_runner(task: str, work_dir: Path):
|
||
yield {"type": "message_end", "message": {"role": "assistant",
|
||
"stopReason": "stop",
|
||
"content": [{"type": "text", "text": "status: success\n\n只读分析草稿。"}]}}
|
||
yield {"type": "agent_end", "messages": []}
|
||
|
||
|
||
def make_plan_runner(plan_builder):
|
||
def runner(task: str, work_dir: Path):
|
||
run_dir = work_dir.parent
|
||
plan = plan_builder(run_dir)
|
||
(run_dir / "outbox").mkdir(parents=True, exist_ok=True)
|
||
(run_dir / "outbox" / "plan.json").write_text(
|
||
json.dumps(plan, ensure_ascii=False), encoding="utf-8")
|
||
yield {"type": "message_end", "message": {"role": "assistant",
|
||
"stopReason": "stop",
|
||
"content": [{"type": "text", "text": "status: success\n\n已生成计划。"}]}}
|
||
yield {"type": "agent_end", "messages": []}
|
||
return runner
|
||
|
||
|
||
async def _stage(store: FakeStore, tmp_path: Path, runner, query: str):
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent(query), runner=runner, config=_cfg(tmp_path))
|
||
confirm_ids = []
|
||
if reply is not None and getattr(reply, "blocks", None):
|
||
confirm_ids = [b.props["confirmId"] for b in reply.blocks
|
||
if getattr(b, "type", "") == "confirm-card"]
|
||
return reply, confirm_ids
|
||
|
||
|
||
def _approve(store: FakeStore, confirm_id: str) -> str:
|
||
return execute_confirmed(store, confirm_id, approve=True, actor="offline-test")
|
||
|
||
|
||
def _s4_plan() -> dict:
|
||
return {"planVersion": 1, "scenario": "S4", "goal": "缺策略沙盒试排",
|
||
"steps": [{"seq": 1, "mode": "frozen", "intent": "flex.simulate_due",
|
||
"summary": "交期探测", "params": {"productCode": DEMO_PRODUCT,
|
||
"quantity": 10},
|
||
"constraints": {}, "expected": []}]}
|
||
|
||
|
||
def _p2_plan() -> dict:
|
||
rows = [{"customerName": "锐扬精密", "productCode": DEMO_PRODUCT,
|
||
"quantity": 10, "deliveryDate": "2026-09-20", "orderNo": "RY-9001"}]
|
||
return {"planVersion": 1, "scenario": "S3", "goal": "导入订单",
|
||
"steps": [{"seq": 1, "mode": "frozen", "intent": "import.commit",
|
||
"summary": "导入 1 行", "params": {
|
||
"batches": [{"kind": "orders", "rows": rows}]},
|
||
"constraints": {"kinds": ["orders"], "maxRows": 500},
|
||
"expected": [{"table": "salesOrders", "added": 1}]}]}
|
||
|
||
|
||
def _seed_s6_world(store: FakeStore, wo_id: int) -> None:
|
||
store.data.setdefault("workOrders", [])
|
||
store.data.setdefault("mesLinks", [])
|
||
store.data["workOrders"].append({
|
||
"id": wo_id, "mesExternalId": "MES-WO-0001", "status": "RUNNING",
|
||
"progressPct": 0, "qtyDone": 0, "productCode": DEMO_PRODUCT})
|
||
store.data["mesLinks"].append({
|
||
"kind": "dispatch", "woId": wo_id, "externalWoId": "MES-WO-0001",
|
||
"idemKey": "idem-1"})
|
||
|
||
|
||
def _seed_mes_mirror(tmp_path: Path, monkeypatch, count: int):
|
||
from server.integrations import mes_stub
|
||
from server.integrations.mes_stub import MockMesClient
|
||
|
||
client = MockMesClient(tmp_path / "mes_mirror.json")
|
||
monkeypatch.setattr(mes_stub, "_client", client)
|
||
for i in range(count):
|
||
client.create_work_order({"productCode": DEMO_PRODUCT}, idem_key=f"idem-m-{i}")
|
||
return client
|
||
|
||
|
||
def _s6_plan(items: list[dict]) -> dict:
|
||
return {"planVersion": 1, "scenario": "S6", "goal": "MES 断连补录",
|
||
"steps": [{"seq": i + 1, "mode": "frozen", "intent": "mes.report",
|
||
"summary": f"补录工单 {p['woId']}", "params": p,
|
||
"constraints": {}, "expected": []}
|
||
for i, p in enumerate(items)]}
|
||
|
||
|
||
def _report_links(store: FakeStore) -> list[dict]:
|
||
return [l for l in store.data.get("mesLinks") or [] if l.get("kind") == "report"]
|
||
|
||
|
||
def _mirror_reports(client) -> list[dict]:
|
||
return client._load().get("reports") or []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-01：features 缺失/损坏/结构非法/非 bool → fallback=False，原路径零副作用
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.mark.parametrize("mode", ["missing", "corrupt", "wrong_top", "nonbool"])
|
||
async def test_o01_features_missing_or_bad_default_off(mode, tmp_path):
|
||
path = tmp_path / "features.json"
|
||
if mode == "corrupt":
|
||
path.write_text("{not json", encoding="utf-8")
|
||
elif mode == "wrong_top":
|
||
path.write_text(json.dumps({"version": 99, "features": {"fallback": True}}),
|
||
encoding="utf-8")
|
||
elif mode == "nonbool":
|
||
path.write_text(json.dumps(
|
||
{"version": 1, "features": {"fallback": "yes", "orders": "no"}}),
|
||
encoding="utf-8")
|
||
flags = load_feature_flags(str(path))
|
||
assert flags["features"]["fallback"]["enabled"] is False
|
||
|
||
store = FakeStore(tmp_path)
|
||
fp_before = _fp(store.data)
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"), runner=_stop_runner, config=_cfg(tmp_path))
|
||
assert reply is None # 调用方原路径（零副作用）
|
||
assert _fb_audits(store) == []
|
||
assert _world_write_audits(store) == []
|
||
assert _run_dirs(tmp_path) == []
|
||
assert _fp(store.data) == fp_before
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-02：无 LLM_BASE_URL / LLM_API_KEY → FallbackUnavailable 显式失败 + FAILED 审计
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.mark.parametrize("mode", ["both", "url_only", "key_only"])
|
||
async def test_o02_missing_llm_env_explicit_failure(mode, tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
cli_path = tmp_path / "pi-cli.cjs"
|
||
cli_path.write_text("// fixture", encoding="utf-8")
|
||
monkeypatch.setattr(fallback_lane, "_resolve_node", lambda _bin: "node-fixture")
|
||
if mode != "url_only":
|
||
monkeypatch.delenv("LLM_BASE_URL", raising=False)
|
||
if mode == "url_only":
|
||
monkeypatch.setenv("LLM_BASE_URL", "https://llm.invalid/v1")
|
||
if mode != "key_only":
|
||
monkeypatch.delenv("LLM_API_KEY", raising=False)
|
||
if mode == "key_only":
|
||
monkeypatch.setenv("LLM_API_KEY", "secret")
|
||
popen_calls: list = []
|
||
monkeypatch.setattr(subprocess, "Popen",
|
||
lambda *a, **kw: popen_calls.append((a, kw)))
|
||
|
||
store = FakeStore(tmp_path)
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"),
|
||
config=_cfg(tmp_path, pi_cli=str(cli_path)))
|
||
assert reply is not None
|
||
assert reply.text == "智能助手服务暂不可用，本次未执行任何操作。请稍后重试或联系管理员。"
|
||
audits = _fb_audits(store)
|
||
assert len(audits) == 1 and audits[0]["result"] == "FAILED"
|
||
assert "unavailable:" in audits[0]["rationale"]["stopReason"]
|
||
assert "无模型配置" in audits[0]["rationale"]["stopReason"]
|
||
assert popen_calls == [] # 配置失败不触子进程
|
||
assert _run_dirs(tmp_path) == []
|
||
assert _world_write_audits(store) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-03：GET /models HTTP 错 / 超时 / 空清单 → resolve 失败，unavailable 显式失败
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.mark.parametrize("mode", ["http_error", "timeout", "empty"])
|
||
async def test_o03_models_endpoint_unavailable_explicit_failure(
|
||
mode, tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
cli_path = tmp_path / "pi-cli.cjs"
|
||
cli_path.write_text("// fixture", encoding="utf-8")
|
||
monkeypatch.setattr(fallback_lane, "_resolve_node", lambda _bin: "node-fixture")
|
||
monkeypatch.setenv("LLM_BASE_URL", "https://llm.invalid/v1")
|
||
monkeypatch.setenv("LLM_API_KEY", "secret")
|
||
monkeypatch.setenv("LLM_MODEL", "model-alpha")
|
||
|
||
def fake_urlopen(request, timeout=5):
|
||
if mode == "http_error":
|
||
raise urllib.error.HTTPError(
|
||
"https://llm.invalid/v1/models", 503, "Service Unavailable", {}, None)
|
||
if mode == "timeout":
|
||
raise TimeoutError("models endpoint timed out")
|
||
return _FakeResponse(b'{"data": []}')
|
||
|
||
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
|
||
popen_calls: list = []
|
||
monkeypatch.setattr(subprocess, "Popen",
|
||
lambda *a, **kw: popen_calls.append((a, kw)))
|
||
|
||
store = FakeStore(tmp_path)
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"),
|
||
config=_cfg(tmp_path, pi_cli=str(cli_path)))
|
||
assert reply is not None
|
||
assert reply.text == "智能助手服务暂不可用，本次未执行任何操作。请稍后重试或联系管理员。"
|
||
audits = _fb_audits(store)
|
||
assert len(audits) == 1 and audits[0]["result"] == "FAILED"
|
||
assert "unavailable:模型协商失败" in audits[0]["rationale"]["stopReason"]
|
||
assert fallback_lane._MODEL_CACHE["model"] is None
|
||
assert popen_calls == []
|
||
assert _run_dirs(tmp_path) == []
|
||
assert _world_write_audits(store) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-04：LLM_MODEL 不在清单 → 取第一模型并记协商 note（锁现状）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def test_o04_model_not_in_catalog_picks_first_with_note(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("LLM_BASE_URL", "https://llm.invalid/v1")
|
||
monkeypatch.setenv("LLM_API_KEY", "secret")
|
||
monkeypatch.setenv("LLM_MODEL", "ghost-model")
|
||
|
||
def fake_urlopen(request, timeout=5):
|
||
return _FakeResponse(b'{"data": [{"id": "model-alpha"}, {"id": "model-beta"}]}')
|
||
|
||
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
|
||
result = fallback_lane.resolve_model(_cfg(tmp_path))
|
||
assert result == "aps-fallback/model-alpha"
|
||
assert fallback_lane._MODEL_CACHE["model"] == result
|
||
note = fallback_lane._MODEL_CACHE["note"] or ""
|
||
assert "ghost-model" in note
|
||
assert "model-alpha" in note
|
||
assert "协商改用" in note
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-05：node 缺失 → unavailable 显式失败，不触子进程
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def test_o05_node_missing_explicit_failure(tmp_path, monkeypatch):
|
||
cli_path = tmp_path / "pi-cli.cjs"
|
||
cli_path.write_text("// fixture", encoding="utf-8")
|
||
missing_node = str(tmp_path / "missing-node.exe")
|
||
assert fallback_lane._resolve_node(missing_node) is None
|
||
|
||
_write_features(tmp_path)
|
||
monkeypatch.setattr(fallback_lane, "_resolve_node", lambda _bin: None)
|
||
popen_calls: list = []
|
||
monkeypatch.setattr(subprocess, "Popen",
|
||
lambda *a, **kw: popen_calls.append((a, kw)))
|
||
|
||
store = FakeStore(tmp_path)
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"),
|
||
config=_cfg(tmp_path, pi_cli=str(cli_path)))
|
||
assert reply is not None
|
||
assert reply.text == "智能助手服务暂不可用，本次未执行任何操作。请稍后重试或联系管理员。"
|
||
audits = _fb_audits(store)
|
||
assert len(audits) == 1 and audits[0]["result"] == "FAILED"
|
||
assert "node 不在 PATH" in audits[0]["rationale"]["stopReason"]
|
||
assert popen_calls == []
|
||
assert _run_dirs(tmp_path) == []
|
||
assert _world_write_audits(store) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-06：pi cli.js 缺失显式 unavailable；损坏现状延迟到 harness_error（不预判内容）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def test_o06_pi_cli_missing_explicit_failure(tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
popen_calls: list = []
|
||
monkeypatch.setattr(subprocess, "Popen",
|
||
lambda *a, **kw: popen_calls.append((a, kw)))
|
||
|
||
store = FakeStore(tmp_path)
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"),
|
||
config=_cfg(tmp_path, pi_cli=str(tmp_path / "missing-cli.cjs")))
|
||
assert reply is not None
|
||
assert reply.text == "智能助手服务暂不可用，本次未执行任何操作。请稍后重试或联系管理员。"
|
||
audits = _fb_audits(store)
|
||
assert len(audits) == 1 and audits[0]["result"] == "FAILED"
|
||
assert "pi cli 不存在" in audits[0]["rationale"]["stopReason"]
|
||
assert popen_calls == []
|
||
assert _run_dirs(tmp_path) == []
|
||
assert _world_write_audits(store) == []
|
||
|
||
|
||
async def test_o06_pi_cli_corrupt_deferred_to_runtime_error_current_semantics(
|
||
tmp_path, monkeypatch):
|
||
# 产品现状：build_pi_runner 只做 is_file() 预检，不解析 pi cli 内容。
|
||
# 损坏文件不会在预检期被判 unavailable，而是延迟到 Popen/事件流阶段；
|
||
# 本用例锁该现状（harness_error + FAILED 审计 + 零世界写）。
|
||
_write_features(tmp_path)
|
||
cli_path = tmp_path / "corrupt-cli.cjs"
|
||
cli_path.write_text("this is not valid pi cli", encoding="utf-8")
|
||
monkeypatch.setattr(fallback_lane, "_resolve_node", lambda _bin: "node-fixture")
|
||
monkeypatch.setenv("LLM_BASE_URL", "https://llm.invalid/v1")
|
||
monkeypatch.setenv("LLM_API_KEY", "secret")
|
||
monkeypatch.setenv("LLM_MODEL", "model-alpha")
|
||
|
||
def fake_urlopen(request, timeout=5):
|
||
return _FakeResponse(b'{"data": [{"id": "model-alpha"}]}')
|
||
|
||
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
|
||
popen_calls: list = []
|
||
|
||
def broken_popen(*args, **kwargs):
|
||
popen_calls.append((args, kwargs))
|
||
raise OSError("cannot execute corrupt cli")
|
||
|
||
monkeypatch.setattr(subprocess, "Popen", broken_popen)
|
||
|
||
store = FakeStore(tmp_path)
|
||
fp_before = _fp(store.data)
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"),
|
||
config=_cfg(tmp_path, pi_cli=str(cli_path), timeout_sec=5))
|
||
assert reply is not None
|
||
assert reply.text == "本次处理未完成，未执行任何操作。请稍后重试。"
|
||
audits = _fb_audits(store)
|
||
assert len(audits) == 1 and audits[0]["result"] == "FAILED"
|
||
assert audits[0]["rationale"]["stopReason"] == "harness_error"
|
||
assert len(popen_calls) == 1 # 损坏内容未被预检，走到 spawn 尝试
|
||
assert _fp(store.data) == fp_before # 世界仍零写
|
||
assert _world_write_audits(store) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-07：执行期 runner 异常（网络/API 错）→ harness_error，FAILED，零世界写
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.mark.parametrize("mode", ["network", "api"])
|
||
async def test_o07_runtime_error_harness_error_failed_zero_write(
|
||
mode, tmp_path):
|
||
_write_features(tmp_path)
|
||
store = FakeStore(tmp_path)
|
||
fp_before = _fp(store.data)
|
||
messages = {
|
||
"network": "network disconnected during execution",
|
||
"api": "upstream api returned 500 during execution",
|
||
}
|
||
|
||
def runner(task: str, work_dir: Path):
|
||
yield {"type": "tool_execution_start", "toolName": "read", "toolCallId": "t1"}
|
||
yield {"type": "tool_execution_end", "toolName": "read", "toolCallId": "t1",
|
||
"result": "snapshot"}
|
||
raise RuntimeError(messages[mode])
|
||
yield # pragma: no cover - 保持生成器形态
|
||
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"), runner=runner, config=_cfg(tmp_path))
|
||
assert reply is not None
|
||
assert reply.text == "本次处理未完成，未执行任何操作。请稍后重试。"
|
||
audits = _fb_audits(store)
|
||
assert len(audits) == 1 and audits[0]["result"] == "FAILED"
|
||
assert audits[0]["rationale"]["stopReason"] == "harness_error"
|
||
assert not getattr(reply, "blocks", None)
|
||
assert _world_write_audits(store) == []
|
||
assert _fp(store.data) == fp_before
|
||
|
||
|
||
async def test_o07b_max_output_breaker_failed_zero_write(tmp_path):
|
||
_write_features(tmp_path)
|
||
store = FakeStore(tmp_path)
|
||
fp_before = _fp(store.data)
|
||
|
||
def huge_runner(task: str, work_dir: Path):
|
||
chunk = "x" * 2048
|
||
for _ in range(2):
|
||
yield {"type": "message_update", "delta": {"text": chunk}}
|
||
yield {"type": "message_end", "message": {"role": "assistant",
|
||
"stopReason": "stop", "content": [{"type": "text", "text": chunk}]}}
|
||
yield {"type": "agent_end", "messages": []}
|
||
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"), runner=huge_runner,
|
||
config=_cfg(tmp_path, max_output_bytes=1024))
|
||
assert reply is not None
|
||
assert reply.text == "本次处理未完成，未执行任何操作。请稍后重试。"
|
||
audits = _fb_audits(store)
|
||
assert len(audits) == 1 and audits[0]["result"] == "FAILED"
|
||
assert audits[0]["rationale"]["stopReason"].startswith("breaker:max_output")
|
||
assert _world_write_audits(store) == []
|
||
assert _fp(store.data) == fp_before
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-08：run 目录创建失败 → propose_reply 不抛，显式失败，不残留 run 目录
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def test_o08_run_dir_create_failure_returns_original(tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
blocking = tmp_path / "blocked-fb"
|
||
blocking.write_text("occupied", encoding="utf-8")
|
||
monkeypatch.setenv("APS_FALLBACK_DIR", str(blocking))
|
||
store = FakeStore(tmp_path)
|
||
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("随便说说"), runner=_stop_runner, config=_cfg(tmp_path))
|
||
assert reply is not None
|
||
assert reply.text == "智能助手服务暂不可用，本次未执行任何操作。请稍后重试。"
|
||
assert _fb_audits(store) == [] # 未到完成审计
|
||
assert _world_write_audits(store) == []
|
||
assert _run_dirs(tmp_path) == [] # 无半成品 run 目录
|
||
assert blocking.is_file() # 原占用未被改写
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-09：白名单缺失/损坏 → P3 全拒；P2 存量可用（fail-closed 与场景隔离）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.mark.parametrize("mode", ["missing", "corrupt"])
|
||
async def test_o09_whitelist_missing_or_corrupt_p3_denied_p2_ok(
|
||
mode, tmp_path):
|
||
_write_features(tmp_path)
|
||
if mode == "corrupt":
|
||
_write_whitelist(tmp_path, "{not a json")
|
||
loaded = fallback_highrisk.load_highrisk_whitelist()
|
||
assert loaded["ok"] is False
|
||
expected_fragment = "白名单文件缺失" if mode == "missing" else "JSON 解析失败"
|
||
assert expected_fragment in loaded["error"]
|
||
|
||
store = FakeStore(tmp_path)
|
||
fp_before = _fp(store.data)
|
||
reply, confirm_ids = await _stage(
|
||
store, tmp_path,
|
||
make_plan_runner(lambda _rd: _s4_plan()),
|
||
query="没有现成策略，帮我沙盒试排")
|
||
assert confirm_ids == []
|
||
assert "未通过校验" in reply.text
|
||
assert expected_fragment in reply.text
|
||
audits = _fb_audits(store)
|
||
assert audits and audits[-1]["result"] == "FAILED"
|
||
assert _world_write_audits(store) == []
|
||
assert _fp(store.data) == fp_before
|
||
|
||
# P2 不受白名单文件缺失影响（P1/P2 路径不依赖 fail-closed 白名单）
|
||
store_p2 = FakeStore(tmp_path)
|
||
reply_p2, confirm_p2 = await _stage(
|
||
store_p2, tmp_path,
|
||
make_plan_runner(lambda _rd: _p2_plan()),
|
||
query="把这份客户表格导进来")
|
||
assert len(confirm_p2) == 1
|
||
assert reply_p2.blocks[0].props["action"] == "agent.fallback.execute"
|
||
assert _world_write_audits(store_p2) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-10：MES 断连。offlineBooking + 断连事实双成立才本地落账；否则拒绝/回滚。
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def test_o10a_offline_booking_double_condition_local_ledger(tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
_write_whitelist(tmp_path, _wl_doc())
|
||
store = FakeStore(tmp_path)
|
||
client = _seed_mes_mirror(tmp_path, monkeypatch, 1)
|
||
_seed_s6_world(store, 9001)
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity", lambda: "failed")
|
||
|
||
_reply, confirm_ids = await _stage(
|
||
store, tmp_path,
|
||
make_plan_runner(lambda _rd: _s6_plan([{
|
||
"woId": 9001, "progressPct": 100, "finish": True,
|
||
"track": "fixed", "offlineBooking": True}])),
|
||
query="MES 连不上了，帮我补录报工")
|
||
assert len(confirm_ids) == 1
|
||
msg = _approve(store, confirm_ids[0])
|
||
assert "执行完成" in msg
|
||
links = _report_links(store)
|
||
assert len(links) == 1
|
||
assert links[0]["syncStatus"] == "PENDING_SYNC"
|
||
assert _mirror_reports(client) == [] # 未同步外部 MES
|
||
audits = [e for e in store.data.get("auditEvents") or []
|
||
if e.get("action") == "mes.report" and e.get("category") == "INTEGRATION"]
|
||
assert len(audits) == 1
|
||
assert audits[0]["rationale"]["offlineBooking"] is True
|
||
assert audits[0]["rationale"]["syncStatus"] == "PENDING_SYNC"
|
||
|
||
|
||
async def test_o10b_mes_down_without_offline_declaration_fails_and_rolls_back(
|
||
tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
_write_whitelist(tmp_path, _wl_doc())
|
||
store = FakeStore(tmp_path)
|
||
client = _seed_mes_mirror(tmp_path, monkeypatch, 1)
|
||
_seed_s6_world(store, 9001)
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity", lambda: "failed")
|
||
monkeypatch.setattr(client, "post_report",
|
||
lambda *a, **kw: (_ for _ in ()).throw(ConnectionError("MES down")))
|
||
|
||
_reply, confirm_ids = await _stage(
|
||
store, tmp_path,
|
||
make_plan_runner(lambda _rd: _s6_plan([{
|
||
"woId": 9001, "progressPct": 80, "track": "fixed"}])),
|
||
query="MES 连不上了，今天的报工帮我补一下")
|
||
assert len(confirm_ids) == 1 # 出卡允许，但没有离线声明
|
||
fp_before = _fp(store.data)
|
||
msg = _approve(store, confirm_ids[0])
|
||
assert "兜底执行失败" in msg
|
||
assert "自动回滚" in msg
|
||
assert _fp(store.data) == fp_before # 回滚后世界零变更
|
||
assert _report_links(store) == []
|
||
assert _mirror_reports(client) == []
|
||
writes = _world_write_audits(store)
|
||
assert len(writes) == 1 # 只允许 FAILED 总账（回滚后补写）
|
||
assert writes[0]["result"] == "FAILED"
|
||
assert writes[0]["rationale"]["rolledBack"] is True
|
||
assert writes[0]["rationale"]["rollbackVerified"] is True
|
||
|
||
|
||
@pytest.mark.parametrize(("mode", "probe", "offline_value", "expected_fragment"), [
|
||
("mes_online", "ok", True, "MES 当前连通"),
|
||
("non_bool", "failed", "yes", "offlineBooking 必须是 bool"),
|
||
])
|
||
async def test_o10c_offline_declaration_abuse_or_bad_shape_rejected(
|
||
mode, probe, offline_value, expected_fragment, tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
_write_whitelist(tmp_path, _wl_doc())
|
||
store = FakeStore(tmp_path)
|
||
client = _seed_mes_mirror(tmp_path, monkeypatch, 1)
|
||
_seed_s6_world(store, 9001)
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity", lambda: probe)
|
||
fp_before = _fp(store.data)
|
||
|
||
reply, confirm_ids = await _stage(
|
||
store, tmp_path,
|
||
make_plan_runner(lambda _rd: _s6_plan([{
|
||
"woId": 9001, "progressPct": 80, "track": "fixed",
|
||
"offlineBooking": offline_value}])),
|
||
query="MES 连不上了，帮我补录报工")
|
||
assert confirm_ids == []
|
||
assert expected_fragment in reply.text
|
||
assert _fp(store.data) == fp_before
|
||
assert _report_links(store) == []
|
||
assert _mirror_reports(client) == []
|
||
assert _world_write_audits(store) == []
|