380 lines
17 KiB
Python
380 lines
17 KiB
Python
# ============================================================
|
||
# FB-03-BUG-1 修复专项（Agent-O，全部确定性）：
|
||
# S6 断连补录的显式离线落账模式（offlineBooking）。
|
||
# 语义：显式声明 + 断连事实双成立才走离线落账（本地落账 +
|
||
# syncStatus=PENDING_SYNC）；在线路径逐字节不变；非断连声明即滥用，拒绝。
|
||
# 手法与 test_fallback_highrisk.py 一致：fake runner / FakeStore / tmp 隔离 /
|
||
# monkeypatch _probe_mes_connectivity seam（零网络零真 LLM）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
from pathlib import Path
|
||
from typing import ClassVar
|
||
|
||
import pytest
|
||
|
||
from server.agent_core import fallback_lane, harness
|
||
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()
|
||
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)
|
||
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()
|
||
|
||
|
||
def _cfg(tmp_path: Path) -> fallback_lane.FallbackConfig:
|
||
return fallback_lane.FallbackConfig(pi_home=str(tmp_path / "pi-home"))
|
||
|
||
|
||
def _write_features(tmp_path: Path) -> None:
|
||
(tmp_path / "features.json").write_text(
|
||
json.dumps({"version": 1, "features": {"fallback": True}}, ensure_ascii=False),
|
||
encoding="utf-8")
|
||
|
||
|
||
def _write_whitelist(tmp_path: Path) -> None:
|
||
doc = {"whitelistVersion": 1, "updatedAt": "2026-09-05T10:00:00",
|
||
"updatedBy": "offline-booking-test",
|
||
"scenarios": {
|
||
"S4": {"enabled": True,
|
||
"intents": ["flex.simulate_due", "flex.compare",
|
||
"scenario.compare", "scenario.sensitivity"],
|
||
"roles": ["planner", "admin"]},
|
||
"S6": {"enabled": True, "intents": ["mes.report"],
|
||
"roles": ["planner", "admin"]},
|
||
"S7": {"enabled": True,
|
||
"intents": ["agent.fallback.ops.config.apply",
|
||
"agent.fallback.policy.update"],
|
||
"roles": ["ops", "admin"]}}}
|
||
(tmp_path / "fallback-highrisk.json").write_text(
|
||
json.dumps(doc, ensure_ascii=False), encoding="utf-8")
|
||
|
||
|
||
def _intent(query: str, **params) -> IntentResult:
|
||
"""Pi 的结构化信封：query 是原话，params 是 Pi 已选定的工具槽位。"""
|
||
return IntentResult(intent="unknown", params={"query": query, **params},
|
||
confidence=0.1, source="LLM")
|
||
|
||
|
||
def _fp(world: dict) -> str:
|
||
return harness.world_fingerprint(world)
|
||
|
||
|
||
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": "ok"}]}}
|
||
yield {"type": "agent_end", "messages": []}
|
||
return runner
|
||
|
||
|
||
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": []}
|
||
|
||
|
||
async def _stage(store: FakeStore, tmp_path: Path, plan: dict, query: str):
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent(query), runner=_make_plan_runner(lambda rd: plan),
|
||
config=_cfg(tmp_path))
|
||
confirm_ids = [b.props["confirmId"] for b in (reply.blocks or [])
|
||
if getattr(b, "type", "") == "confirm-card"] if reply else []
|
||
return reply, confirm_ids
|
||
|
||
|
||
def _approve(store: FakeStore, confirm_id: str, *, approve: bool = True) -> str:
|
||
return execute_confirmed(store, confirm_id, approve=approve, actor="offline-test")
|
||
|
||
|
||
def _seed_s6_world(store: FakeStore, items: list[tuple]) -> None:
|
||
store.data.setdefault("workOrders", [])
|
||
store.data.setdefault("mesLinks", [])
|
||
for wo_id, ext_id, status, pct, qty in items:
|
||
store.data["workOrders"].append({
|
||
"id": wo_id, "mesExternalId": ext_id, "status": status,
|
||
"progressPct": pct, "qtyDone": qty, "productCode": DEMO_PRODUCT})
|
||
store.data["mesLinks"].append({
|
||
"kind": "dispatch", "woId": wo_id, "externalWoId": ext_id,
|
||
"idemKey": f"idem-{wo_id}"})
|
||
|
||
|
||
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 _setup(tmp_path: Path, monkeypatch, wo_ids=(9001, 9002)):
|
||
_write_features(tmp_path)
|
||
_write_whitelist(tmp_path)
|
||
store = FakeStore(tmp_path)
|
||
client = _seed_mes_mirror(tmp_path, monkeypatch, len(wo_ids))
|
||
_seed_s6_world(store, [
|
||
(w, f"MES-WO-{i + 1:04d}", "RUNNING", 0, 0) for i, w in enumerate(wo_ids)])
|
||
return store, client
|
||
|
||
|
||
def _report_links(store: FakeStore) -> list[dict]:
|
||
return [l for l in store.data.get("mesLinks") or [] if l.get("kind") == "report"]
|
||
|
||
|
||
def _mes_report_audits(store: FakeStore) -> list[dict]:
|
||
return [e for e in store.data.get("auditEvents") or []
|
||
if e.get("action") == "mes.report" and e.get("category") == "INTEGRATION"]
|
||
|
||
|
||
def _mirror_reports(client) -> list[dict]:
|
||
return client._load().get("reports") or []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-1：离线落账成功路径——声明 + 断连事实双成立 → 本地落账 + PENDING_SYNC 标记，
|
||
# post_report 全程未被调用；确认卡摘要明示「MES 断连，本地落账待同步」
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_o1_offline_booking_success_path(tmp_path, monkeypatch):
|
||
store, client = _setup(tmp_path, monkeypatch)
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity", lambda: "failed")
|
||
items = [{"woId": 9001, "progressPct": 100, "finish": True,
|
||
"track": "fixed", "offlineBooking": True},
|
||
{"woId": 9002, "progressPct": 60, "track": "fixed",
|
||
"offlineBooking": True}]
|
||
reply, confirm_ids = await _stage(
|
||
store, tmp_path, _s6_plan(items), "MES 连不上了，今天的报工帮我补一下")
|
||
assert len(confirm_ids) == 2
|
||
# 审批人知情：卡摘要明示断连本地落账 + 待同步标记
|
||
summary_blob = json.dumps(
|
||
[b.props.get("summary") for b in reply.blocks], ensure_ascii=False)
|
||
assert "MES 断连" in summary_blob and "PENDING_SYNC" in summary_blob
|
||
|
||
msg1 = _approve(store, confirm_ids[0])
|
||
msg2 = _approve(store, confirm_ids[1])
|
||
assert "执行完成" in msg1 and "执行完成" in msg2
|
||
wos = {w["id"]: w for w in store.data["workOrders"]}
|
||
assert wos[9001]["status"] == "COMPLETED"
|
||
assert wos[9002]["progressPct"] == 60
|
||
# 可识别状态：两笔报工记录均带 syncStatus=PENDING_SYNC
|
||
links = _report_links(store)
|
||
assert len(links) == 2
|
||
assert all(l.get("syncStatus") == "PENDING_SYNC" for l in links)
|
||
# 关键断言：post_report 全程未被调用（MES 镜像报工流水零新增）
|
||
assert _mirror_reports(client) == []
|
||
# 审计 rationale 同步标记（链上可回放）
|
||
audits = _mes_report_audits(store)
|
||
assert len(audits) == 2
|
||
assert all(e["rationale"].get("offlineBooking") is True
|
||
and e["rationale"].get("syncStatus") == "PENDING_SYNC"
|
||
for e in audits)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-2：在线路径不变断言——未声明 offlineBooking：post_report 照常同步推送、
|
||
# mesLinks/审计形态逐字节口径不变；且全程不触发连通性探测（零新增副作用）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_o2_online_path_unchanged(tmp_path, monkeypatch):
|
||
store, client = _setup(tmp_path, monkeypatch, wo_ids=(9001,))
|
||
def _probe_must_not_be_called():
|
||
raise AssertionError("在线路径不应触发连通性探测")
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity",
|
||
_probe_must_not_be_called)
|
||
items = [{"woId": 9001, "progressPct": 80, "track": "fixed"}]
|
||
_reply, confirm_ids = await _stage(
|
||
store, tmp_path, _s6_plan(items), "MES 连不上了，今天的报工帮我补一下")
|
||
assert len(confirm_ids) == 1
|
||
msg = _approve(store, confirm_ids[0])
|
||
assert "执行完成" in msg
|
||
wos = {w["id"]: w for w in store.data["workOrders"]}
|
||
assert wos[9001]["progressPct"] == 80
|
||
# post_report 被照常调用（镜像报工流水 +1，原语义）
|
||
assert len(_mirror_reports(client)) == 1
|
||
# mesLinks 报工记录无 syncStatus 键（在线路径形态逐字节不变）
|
||
links = _report_links(store)
|
||
assert len(links) == 1 and "syncStatus" not in links[0]
|
||
# 审计 rationale 键集与原口径完全一致
|
||
audits = _mes_report_audits(store)
|
||
assert len(audits) == 1
|
||
assert set(audits[0]["rationale"]) == {"pct", "status", "orderDone"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-3：滥用拒绝（出卡闸）——MES 连通时声明 offlineBooking → 拒绝出卡 + 零变更
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_o3_abuse_rejected_at_stage_gate(tmp_path, monkeypatch):
|
||
store, client = _setup(tmp_path, monkeypatch, wo_ids=(9001,))
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity", lambda: "ok")
|
||
fp_before = _fp(store.data)
|
||
items = [{"woId": 9001, "progressPct": 80, "track": "fixed",
|
||
"offlineBooking": True}]
|
||
reply, confirm_ids = await _stage(
|
||
store, tmp_path, _s6_plan(items), "MES 连不上了，今天的报工帮我补一下")
|
||
assert confirm_ids == []
|
||
assert "未通过校验" in reply.text
|
||
assert "MES 当前连通" in reply.text
|
||
assert _fp(store.data) == fp_before # 世界零变更
|
||
assert _report_links(store) == []
|
||
assert _mirror_reports(client) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-4：滥用拒绝（执行端）——出卡时断连、审批窗口内 MES 恢复 → 熔断回滚，
|
||
# 零落账（前提不再成立即拒，绝不借离线模式跳过 MES 同步）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_o4_abuse_rejected_at_execute_gate(tmp_path, monkeypatch):
|
||
store, client = _setup(tmp_path, monkeypatch, wo_ids=(9001,))
|
||
probe_state = {"v": "failed"}
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity",
|
||
lambda: probe_state["v"])
|
||
items = [{"woId": 9001, "progressPct": 80, "track": "fixed",
|
||
"offlineBooking": True}]
|
||
_reply, confirm_ids = await _stage(
|
||
store, tmp_path, _s6_plan(items), "MES 连不上了，今天的报工帮我补一下")
|
||
assert len(confirm_ids) == 1 # 断连事实下允许出卡
|
||
probe_state["v"] = "ok" # 审批窗口内 MES 恢复
|
||
msg = _approve(store, confirm_ids[0])
|
||
assert "已熔断并自动回滚" in msg
|
||
assert "未留下任何变更" in msg
|
||
wos = {w["id"]: w for w in store.data["workOrders"]}
|
||
assert wos[9001]["progressPct"] == 0 # 零落账
|
||
assert _report_links(store) == []
|
||
assert _mirror_reports(client) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-5：PENDING_SYNC 对账呈现——恢复后对账识别离线落账记录并如实呈现
|
||
# （回复文案 / 对账报告 / ALGO_RUN rationale 三处一致）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_o5_pending_sync_presented_in_reconcile(tmp_path, monkeypatch):
|
||
_write_features(tmp_path)
|
||
_write_whitelist(tmp_path)
|
||
store = FakeStore(tmp_path)
|
||
_seed_s6_world(store, [(9101, "MES-WO-0001", "RUNNING", 50, 5)])
|
||
store.data["mesLinks"].append({ # 断连期离线落账的事实记录
|
||
"kind": "report", "woId": 9101, "externalWoId": "MES-WO-0001",
|
||
"track": "fixed", "progressPct": 50, "status": "RUNNING",
|
||
"syncedAt": "2026-09-05", "actor": "pi-fallback",
|
||
"syncStatus": "PENDING_SYNC"})
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity", lambda: "ok")
|
||
monkeypatch.setattr(
|
||
fallback_lane, "_fetch_external_wo",
|
||
lambda ext_id: {"id": ext_id, "status": "RUNNING",
|
||
"progressPct": 50, "qtyDone": 5})
|
||
fp_before = _fp(store.data)
|
||
|
||
reply = await fallback_lane.propose_reply(
|
||
store, "s1", _intent("MES 恢复了，对一下账",
|
||
p3Scenario="S6", p3Action="reconcile"),
|
||
runner=_stop_runner, config=_cfg(tmp_path))
|
||
assert reply is not None
|
||
assert "一致 1" in reply.text
|
||
assert "PENDING_SYNC" in reply.text and "1 笔" in reply.text
|
||
|
||
runs = [p for p in (tmp_path / "fb").iterdir() if p.is_dir() and p.name != "pi-home"]
|
||
assert len(runs) == 1
|
||
report = (runs[0] / "outbox" / "reconcile-report.md").read_text(encoding="utf-8")
|
||
assert "待同步（PENDING_SYNC）1 笔" in report
|
||
assert "本地待同步（PENDING_SYNC）" in report
|
||
recon = [e for e in store.data.get("auditEvents") or []
|
||
if e.get("action") == "agent.fallback.reconcile"]
|
||
assert recon and recon[-1]["rationale"].get("pendingSync") == 1
|
||
assert _fp(store.data) == fp_before # 对账纯读：世界零变更
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# O-6：形态校验——offlineBooking 非 bool → 出卡闸显式拒绝（参数形态防线）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_o6_offline_booking_non_bool_rejected(tmp_path, monkeypatch):
|
||
store, _client = _setup(tmp_path, monkeypatch, wo_ids=(9001,))
|
||
monkeypatch.setattr(fallback_lane, "_probe_mes_connectivity", lambda: "failed")
|
||
items = [{"woId": 9001, "progressPct": 80, "track": "fixed",
|
||
"offlineBooking": "yes"}]
|
||
reply, confirm_ids = await _stage(
|
||
store, tmp_path, _s6_plan(items), "MES 连不上了，今天的报工帮我补一下")
|
||
assert confirm_ids == []
|
||
assert "未通过校验" in reply.text
|
||
assert "offlineBooking 必须是 bool" in reply.text
|