aps-agent/tests/golden/test_automation_gateway.py

# ============================================================
# 自动化网关驱动黄金测试（round-40 方向 S · 矩阵 76）
# 覆盖：网关定时驱动（POST /api/automation/tick 到期触发 + 状态落盘）、
#       调度器持久化往返（规则/到期基线/暂停/运行记录宽容恢复）、
#       真实业务动作接线（G2 只读建议 / G3 出确认卡 / G4 受控自动执行，
#       未登记动作仍被 P3 门禁拦截）。
# ============================================================
from __future__ import annotations

import os
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from server.agent_core.automation import (
    AutomationGate,
    AutomationScheduler,
    Rule,
    RuleExecutor,
    RuleRegistry,
    automation_authorization,
)
from server.aps_domain import workflow as wf
from server.auth.context import IdentityContext, bind_identity, reset_identity
from server.contracts import IntentResult


class _MemStore:
    """极简内存 store：world + next_id + save（黄金测试隔离）。"""

    def __init__(self, world: dict | None = None) -> None:
        self.data: dict = world if world is not None else {"auditEvents": []}
        self._seq = 0

    def next_id(self, _kind: str) -> int:
        self._seq += 1
        return self._seq

    def save(self) -> None:
        pass


class _CheckpointStore:
    """极简检查点仓（避免 G4 自动执行时写真实快照文件）。"""

    def create(self, *_args, **_kwargs) -> dict:
        return {"pairId": "pair-auto"}


def _approve_escalation_twice(gate: AutomationGate, confirm_id: str) -> dict:
    """P3 升权确认卡需两重审批且由不同用户完成（SOD），返回最终决策记录。"""
    first = gate.approve(confirm_id)
    assert first is not None and first["needsSecondConfirm"] is True
    token = bind_identity(IdentityContext(
        2002, "collaborator", "Collaborator", "platform", roles=("planner",),
    ))
    try:
        second = gate.approve(confirm_id)
    finally:
        reset_identity(token)
    assert second is not None and second["needsSecondConfirm"] is False
    return second


def _scheduled_world_store() -> _MemStore:
    """种子世界 + 规则引擎试排一版：scheduleVersions 含 DRAFT 版本（发布动作可用）。"""
    from server.aps_domain.rush import _engine_params, _sandbox_counter
    from server.engines import get_engine
    from server.state.seed import seed_world
    world = seed_world()
    get_engine("RULE").solve(world, _engine_params(world, "DELIVERY_FIRST"), _sandbox_counter())
    assert world.get("scheduleVersions"), "种子世界必须含 DRAFT 版本"
    # 演示种子按约束配置会产出硬约束冲突（发布门禁 fail-closed）；
    # 测试关注自动化接线，因此把演示冲突标记为已解决，使发布路径可达。
    for _conflict in world.get("conflicts") or []:
        _conflict["isResolved"] = True
    return _MemStore(world)


def _business_bridge(store):
    """与网关 AutomationRuntime 相同的业务动作桥（rule action → 真实意图）。"""
    def bridge(binding, params, ctx):
        payload = dict(binding.params or {})
        payload.update(dict((params or {}).get("payload") or {}))
        intent = IntentResult(intent=binding.intent, params=payload,
                              confidence=1.0, source="RULE_FAST")
        return wf.run_automation_intent(
            store, str(ctx.get("session_id") or "s"), intent,
            actor=str(ctx.get("actor") or "automation"),
            auto=bool(ctx.get("auto")),
        )
    return bridge


@pytest.fixture(autouse=True)
def _cleanup(monkeypatch, tmp_path):
    """自动化运行时/门禁队列隔离：状态文件指向临时目录，运行时每次重建。"""
    import server.gateway.app as gw
    from server.agent_core import harness
    harness._approval_store.clear()
    monkeypatch.setenv("APS_AUTOMATION_STATE_PATH", str(tmp_path / "automation_state.json"))
    gw._reset_automation_runtime()
    yield
    gw._reset_automation_runtime()
    harness._approval_store.clear()


# ---------------- P1：调度器持久化往返（规则/到期基线/暂停/运行记录） ----------------

def test_scheduler_persistence_roundtrip_rules_and_due_state(tmp_path):
    """规则 + lastRun 到期基线 + 暂停 + 运行记录落盘，重启宽容恢复且到期连续。"""
    base = datetime(2026, 8, 2, 0, 0, 0, tzinfo=UTC)
    clock = {"now": base}
    store = _MemStore()
    registry = RuleRegistry()
    executor = RuleExecutor(registry=registry)
    registry.register(Rule("r-every", "room-1", {"on": "schedule", "every": 60},
                           gear="G2", action="notify"))
    registry.register(Rule("r-cron", "room-1", {"on": "schedule", "cron": "0 8 * * *"},
                           gear="G2", action="notify"))
    registry.register(Rule("r-paused", "room-1", {"on": "schedule", "every": 120},
                           gear="G2", action="notify"))
    scheduler = AutomationScheduler(executor, now=lambda: clock["now"])
    scheduler.pause_rule("r-paused")
    scheduler.pause()                                        # 整体暂停也要持久化
    # 首个 tick：间隔规则到期触发；cron 仅建基线；暂停规则跳过
    first = scheduler.tick(world=store.data, next_id=store.next_id, save=store.save)
    assert [r.rule_id for r in first] == []
    scheduler.resume()
    first = scheduler.tick(world=store.data, next_id=store.next_id, save=store.save)
    assert [r.rule_id for r in first] == ["r-every"]

    state_path = tmp_path / "automation_state.json"
    scheduler.save_state(state_path)
    assert state_path.exists()

    # 重启：新调度器宽容加载（规则/暂停/到期基线/运行记录）
    clock["now"] = base + timedelta(seconds=61)
    loaded = AutomationScheduler.load_state(state_path)
    assert [r.rule_id for r in loaded.executor.registry.list()] == ["r-every", "r-cron", "r-paused"]
    assert loaded.paused_rules() == ["r-paused"]
    assert loaded.paused is False                          # 整体暂停状态恢复为 resume 后
    # 恢复的 lastRun 基线：+61s 后 r-every 再到期触发；r-cron 未到期；r-paused 仍暂停
    runs = loaded.tick(world=store.data, next_id=store.next_id, save=store.save,
                       now=clock["now"])
    assert [r.rule_id for r in runs] == ["r-every"]
    # 运行记录恢复（可回放）
    assert loaded.executor.runs and any(r.rule_id == "r-every" for r in loaded.executor.runs.values())

    # 宽容：损坏 JSON / 缺失文件 → 空调度不抛错
    bad = tmp_path / "bad.json"
    bad.write_text("{not json", encoding="utf-8")
    assert AutomationScheduler.load_state(bad).executor.registry.list() == []
    assert AutomationScheduler.load_state(tmp_path / "missing.json").executor.registry.list() == []


def test_rule_params_roundtrip_through_persistence():
    """业务动作入参（如重排 level）随规则序列化往返。"""
    rule = Rule("r-x", "room-1", {"on": "schedule", "every": 30}, gear="G2",
                action="reschedule", params={"level": "L3"})
    restored = Rule.from_dict(rule.as_dict())
    assert restored == rule
    assert restored.params == {"level": "L3"}


# ---------------- P2：网关定时驱动（tick 到期触发 + 审计 + 落盘） ----------------

def test_gateway_tick_endpoint_triggers_due_rule_and_persists(monkeypatch):
    """POST /api/automation/tick：到期规则触发（真实业务动作出卡）+ AUTOMATION 审计 + 状态落盘。"""
    import server.gateway.app as gw
    from tests.auth_provider import install_test_auth

    install_test_auth(monkeypatch, "tenant-batch-test")
    store = _scheduled_world_store()
    store.world_key = "personal-1001"
    store.tenant_uuid = "tenant-batch-test"
    monkeypatch.setattr(gw, "get_store", lambda: store)

    client = TestClient(gw.create_app())
    login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
    assert login.status_code == 200, login.text

    runtime = gw._get_automation_runtime()
    runtime.executor.registry.register(
        Rule("r-pub", "room-1", {"on": "schedule", "every": 3600},
             gear="G2", action="commit")
    )
    resp = client.post("/api/automation/tick", json={})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["ok"] is True and body["triggered"] == 1
    run = body["runs"][0]
    assert run["ruleId"] == "r-pub"
    assert run["status"] == "STAGED"                      # G2：P2 真实业务动作只出确认卡
    assert run["payload"]["confirmId"]
    # 既有 AUTOMATION 审计已写（含 gear/rule/trigger）
    run_events = [e for e in store.data["auditEvents"] if e["action"] == "automation.run"]
    assert run_events and run_events[-1]["rationale"]["runId"] == run["runId"]
    assert run_events[-1]["rationale"]["gear"] == "G2"
    # 门禁出卡审计（schedule.publish.stage，GATE 类）
    assert any(e["action"] == "schedule.publish.stage" for e in store.data["auditEvents"])
    # 状态已落盘（规则可跨重启恢复）
    state_path = Path(os.environ["APS_AUTOMATION_STATE_PATH"])
    assert state_path.exists()
    import json as _json
    state = _json.loads(state_path.read_text(encoding="utf-8"))
    assert any(rule.get("ruleId") == "r-pub" for rule in state["rules"])
    # 状态端点只读投影
    status = client.get("/api/automation/status")
    assert status.status_code == 200
    assert status.json()["ruleCount"] >= 1


def test_gateway_confirm_mints_bound_g4_grant_and_executes_real_bridge(monkeypatch):
    import server.aps_domain.workflow as wf_module
    import server.gateway.app as gw
    from tests.auth_provider import install_test_auth

    class ScopedProjectStore:
        def active_world_key(self):
            return "p1"

        def require_active_write(self):
            return None

        def snapshot(self, include_messages=False):
            data = {
                "projects": [{"id": "p1", "name": "Automation", "workDir": ""}],
                "sessions": [{"id": "s-g4", "projectId": "p1"}],
                "files": [],
                "activeProjectId": "p1",
                "activeSessionId": "s-g4",
            }
            if include_messages:
                data["messages"] = {"s-g4": []}
            return data

    install_test_auth(monkeypatch, "tenant-g4-gateway")
    store = _scheduled_world_store()
    store.world_key = "p1"
    store.tenant_uuid = "tenant-g4-gateway"
    monkeypatch.setattr(gw, "get_store", lambda: store)
    monkeypatch.setattr("server.state.projects.get_project_store", lambda: ScopedProjectStore())
    monkeypatch.setattr(wf_module, "get_checkpoints", lambda: _CheckpointStore())

    client = TestClient(gw.create_app())
    assert client.post(
        "/api/auth/login", json={"username": "planner", "password": "test"},
    ).status_code == 200
    runtime = gw._get_automation_runtime()
    rule = Rule(
        "r-g4-api", "room-1", {"on": "schedule", "every": 60},
        gear="G4", action="commit", params={"track": "fixed"},
    )
    runtime.executor.registry.register(rule)
    token = bind_identity(IdentityContext(
        1001, "planner", "计划员", "tenant-g4-gateway", roles=("planner",),
    ))
    try:
        staged = runtime.executor.execute(
            rule,
            rule.trigger,
            world=store.data,
            next_id=store.next_id,
            session_id="s-g4",
            save=store.save,
        )
    finally:
        reset_identity(token)
    assert staged.status == "GATE_REQUIRED"
    confirm_id = staged.payload["escalateConfirmId"]

    first = client.post(
        "/api/actions/confirm", json={"confirmId": confirm_id, "approve": True},
    )
    assert first.status_code == 200, first.text
    assert first.json()["secondConfirmRequired"] is True, first.json()

    assert client.post(
        "/api/auth/login", json={"username": "collaborator", "password": "test"},
    ).status_code == 200
    second = client.post(
        "/api/actions/confirm", json={"confirmId": confirm_id, "approve": True},
    )
    assert second.status_code == 200, second.text
    assert second.json()["secondConfirmRequired"] is False
    assert "已获 G4" in second.json()["message"]
    grant = runtime.executor.gate.grant_for(rule.rule_id)
    assert grant is not None and grant.authorization_digest

    token = bind_identity(IdentityContext(
        1002, "collaborator", "协作成员", "tenant-g4-gateway", roles=("planner",),
    ))
    try:
        executed = runtime.executor.execute(
            rule,
            rule.trigger,
            world=store.data,
            next_id=store.next_id,
            session_id="s-g4",
            save=store.save,
        )
    finally:
        reset_identity(token)
    assert executed.status == "EXECUTED" and executed.gear == "G4"
    assert store.data["scheduleVersions"][-1]["status"] == "PUBLISHED"
    assert any(
        event.get("action") == "automation.gear.escalate.granted"
        for event in store.data["auditEvents"]
    )


# ---------------- P3：真实业务动作接线（G2/G3/G4 档位语义） ----------------

def test_business_action_wiring_gear_semantics(monkeypatch):
    """G2 只读建议 / G2 出卡 / G3 出确认卡 / G4 受控自动执行 / 未登记动作 P3 拦截。"""
    import server.aps_domain.workflow as wf_module
    store = _scheduled_world_store()
    monkeypatch.setattr(wf_module, "get_checkpoints", lambda: _CheckpointStore())

    registry = RuleRegistry()
    gate = AutomationGate()
    executor = RuleExecutor(registry=registry, gate=gate, business_bridge=_business_bridge(store))

    # 动作注册表：commit → schedule.publish；reschedule → flex.reschedule；未登记 → None
    assert executor.binding_for("commit").intent == "schedule.publish"
    assert executor.binding_for("reschedule").intent == "flex.reschedule"
    assert executor.binding_for("reschedule").params == {"level": "L2"}
    assert executor.binding_for("notify") is None

    # G2 只读建议：suggest 动作不产生任何写入
    s = Rule("r-s", "room-1", {"on": "schedule", "every": 60}, gear="G2", action="suggest")
    registry.register(s)
    run = executor.execute(s, s.trigger, world=store.data, next_id=store.next_id,
                           session_id="s", save=store.save)
    assert run.status == "PROPOSED" and run.payload["summary"]

    # G2 沙盒：commit（P2 真实业务动作）只出确认卡，绝不自动执行
    g2 = Rule("r-g2", "room-1", {"on": "schedule", "every": 60}, gear="G2", action="commit")
    registry.register(g2)
    run = executor.execute(g2, g2.trigger, world=store.data, next_id=store.next_id,
                           session_id="s", save=store.save)
    assert run.status == "STAGED"
    assert run.payload["confirmId"]
    assert store.data["scheduleVersions"][-1]["status"] == "DRAFT"   # 未批准 → 不发布

    # G3 监督：升权凭据到手后出确认卡（P2 卡）
    g3 = Rule("r-g3", "room-1", {"on": "schedule", "every": 60}, gear="G3", action="commit")
    registry.register(g3)
    card = gate.request("s", "r-g3", "G3", room_id="room-1")
    gate.grant_from_decision(card["confirmId"], _approve_escalation_twice(gate, card["confirmId"]),
                             rule_id="r-g3", params=card["params"])
    run = executor.execute(g3, g3.trigger, world=store.data, next_id=store.next_id,
                           session_id="s", save=store.save)
    assert run.status == "STAGED" and run.gear == "G3"
    assert run.payload["confirmId"]
    assert store.data["scheduleVersions"][-1]["status"] == "DRAFT"

    # G4 全自治：升权凭据 + 业务桥 → 自动批准执行真实发布（版本 PUBLISHED，全审计）
    g4 = Rule("r-g4", "room-1", {"on": "schedule", "every": 60}, gear="G4", action="commit")
    registry.register(g4)
    card = gate.request(
        "s",
        "r-g4",
        "G4",
        room_id="room-1",
        authorization=automation_authorization(g4, executor.binding_for(g4.action)),
    )
    gate.grant_from_decision(card["confirmId"], _approve_escalation_twice(gate, card["confirmId"]),
                             rule_id="r-g4", params=card["params"])
    run = executor.execute(g4, g4.trigger, world=store.data, next_id=store.next_id,
                           session_id="s", save=store.save)
    assert run.status == "EXECUTED" and run.gear == "G4"
    assert store.data["scheduleVersions"][-1]["status"] == "PUBLISHED"   # 真实业务动作已执行
    g4_events = [e for e in store.data["auditEvents"] if e["action"] == "automation.run"
                 and e["rationale"]["runId"] == run.run_id]
    assert g4_events and g4_events[0]["rationale"]["gear"] == "G4"


def test_unregistered_action_blocked_by_p3_gate_even_at_g4(monkeypatch):
    """未登记真实业务动作（无 binding / 未接线业务桥）→ G4 受控自动仍被 P3 门禁拦截。"""
    import server.aps_domain.workflow as wf_module
    store = _scheduled_world_store()
    monkeypatch.setattr(wf_module, "get_checkpoints", lambda: _CheckpointStore())

    gate = AutomationGate()
    # 未注入业务桥、未注册处理器 → commit 无可执行的真实业务实现
    bare = RuleExecutor(registry=RuleRegistry(), gate=gate)
    bare._business_bindings.pop("commit", None)            # 显式模拟「未登记动作」
    rule = Rule("r-g4x", "room-1", {"on": "schedule", "every": 60}, gear="G4", action="commit")
    bare.registry.register(rule)
    with pytest.raises(ValueError, match="BusinessActionBinding"):
        gate.request(
            "s",
            "r-g4x",
            "G4",
            room_id="room-1",
            authorization=automation_authorization(rule, bare.binding_for(rule.action)),
        )

    run = bare.execute(rule, rule.trigger, world=store.data, next_id=store.next_id,
                       session_id="s", save=store.save, escalate_on_demand=False)
    assert run.status == "DENIED"
    assert run.payload.get("unregisteredAction") is True
    denied = [e for e in store.data["auditEvents"] if e["action"] == "automation.run"][-1]
    assert denied["power"] == "P3"                          # fail-closed：未登记默认 P3
    assert store.data["scheduleVersions"][-1]["status"] == "DRAFT"  # 未执行任何写入


def test_automation_intent_unbound_fails_closed():
    """未接线的真实业务意图 → ValueError（fail closed，不产生任何写入）。"""
    store = _scheduled_world_store()
    with pytest.raises(ValueError):
        wf.run_automation_intent(
            store, "s", IntentResult(intent="order.cancel", params={}), auto=False)