AgentLab/backend/tests/test_app_backfill.py

"""按新口径补算（D-f；设计文档 §3.4 趋势那一行、X3 的 app_backfill）。

- 只有视图的钉住项最近一次升版选了 recompute 才补算；取旧版本在同一来源上最近 N 期（N 不超过趋势的 last）已封存正式
  运行的输入，用新版本逐个发起正式运行（来源 app_backfill，带 app_id / app_version，按钉住的 graph_hash 核对）；
- 新版本已经算过同一个输入的跳过（补算过的、来源本身按新版本跑过的），再点一次不重复发起；
- 趋势把补算的运行算作新版本的点，按它补的那一期排；「最近一次」不取补算的运行；
- 选了别的处置、视图不存在的拒绝；详情里给管理者列出能补算的视图和会发起几次；
- 有外部副作用的工作流（人工节点、写外部系统、钉作流程）不补算，详情里也不列；
- 同一个视图并发点两次，同一个输入只发起一次。
"""
from __future__ import annotations

import asyncio
import uuid

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select

from app.core.artifact_store import graph_hash
from app.db.base import SessionLocal, utcnow
from app.db.models import Run, Workflow, WorkflowVersion
from app.engine.runner import run_manager
from app.main import app


@pytest.fixture(autouse=True)
async def engine_up():
    await run_manager.setup()
    yield
    await run_manager.shutdown()


@pytest.fixture
async def client():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c


def node(nid, ntype, label=None, **config):
    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}}


def kpi(gmv: int) -> dict:
    nodes = [
        node("start", "input", "开始", fields=[{"name": "week", "required": True}]),
        node("fetch", "transform", "取数", mode="expression", expression=f"{{'gmv': {gmv}}}", assign_to="kpi"),
        node("card", "metrics", "口径卡", caliber="周报口径", caliber_version="v1",
             metrics=[{"id": "gmv", "name": "销售额", "unit": "元", "expression": "vars.kpi.gmv"}]),
        node("out", "output", "成果", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}]),
    ]
    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}


async def workflow(v2: dict | None = None) -> tuple[str, str, str]:
    """v1、v2 两个发布版，口径不同（gmv 100 → 200）。返回 (工作流, v1 哈希, v2 哈希)。"""
    v1, v2 = kpi(100), v2 or kpi(200)
    async with SessionLocal() as session:
        wf = Workflow(name=f"周报-{uuid.uuid4().hex[:6]}", graph=v2, status="published", published_version=2, version=2)
        session.add(wf)
        await session.flush()
        for number, graph in ((1, v1), (2, v2)):
            session.add(WorkflowVersion(workflow_id=wf.id, version=number, graph=graph, graph_hash=graph_hash(graph),
                                        level="published", published_at=utcnow()))
        await session.commit()
        return wf.id, graph_hash(v1), graph_hash(v2)


async def wait(run_id: str) -> Run:
    for _ in range(600):
        async with SessionLocal() as session:
            row = await session.get(Run, run_id)
        if row.status in ("succeeded", "failed", "cancelled") and row.manifest_seq is not None:
            return row
        await asyncio.sleep(0.03)
    raise AssertionError(f"运行 {run_id} 没有结束：{row.status}")


async def formal(wf: str, version: int, week: str) -> Run:
    from app.engine.launch import launch_run

    async with SessionLocal() as session:
        run = await launch_run(session, workflow_id=wf, run_class="formal", version=version,
                               input_payload={"week": week}, actor="周报编辑", origin_kind="launcher")
    row = await wait(run.id)
    assert row.status == "succeeded", row.error
    return row


def spec(wf: str, version: int, hash_: str) -> dict:
    return {
        "format": "agentlab.app/1", "name": "周报看板",
        "pins": [{"alias": "main", "workflow_id": wf, "version": version, "graph_hash": hash_, "use": "view"}],
        "views": {"latest": {"source": {"kind": "latest_sealed", "pin": "main", "from": "formal"}}},
        "pages": [{"id": "home", "title": "首页", "body": [
            {"id": "trend", "type": "series", "props": {"label": "销售额趋势"},
             "bind": {"view": "latest", "series": "card.gmv", "last": 6}},
            {"id": "gmv", "type": "metric", "props": {"label": "销售额"}, "bind": {"view": "latest", "metric": "card.gmv"}},
        ]}],
    }


async def upgraded_app(client, wf: str, h1: str, h2: str, disposition: str) -> str:
    res = await client.post("/api/apps", json={"name": "周报看板", "spec": spec(wf, 1, h1)})
    assert res.status_code == 201, res.text
    app_id = res.json()["id"]
    assert (await client.post(f"/api/apps/{app_id}/publish", json={})).json()["ok"]
    draft = (await client.get(f"/api/apps/{app_id}")).json()["draft"]
    draft["pins"][0].update(version=2, graph_hash=h2)
    assert (await client.patch(f"/api/apps/{app_id}", json={"draft": draft})).status_code == 200
    out = (await client.post(f"/api/apps/{app_id}/publish", json={"dispositions": {"main": disposition}})).json()
    assert out["ok"] and out["version"] == 2, out
    return app_id


def find(vm, cid):
    return next(n for n in vm["tree"] if n["id"] == cid)


async def test_backfill_recomputes_the_old_periods_under_the_new_version(client):
    wf, h1, h2 = await workflow()
    old = [await formal(wf, 1, w) for w in ("W1", "W2", "W3")]
    app_id = await upgraded_app(client, wf, h1, h2, "recompute")
    current = await formal(wf, 2, "W3")         # 升版之后新版本已经跑过 W3：这一期不补

    plan = (await client.get(f"/api/apps/{app_id}/views/latest/backfill")).json()
    assert plan == {"view": "latest", "alias": "main", "source": "formal", "from_version": 1, "to_version": 2,
                    "periods": 6, "will_launch": 2, "will_skip": 1}
    detail = (await client.get(f"/api/apps/{app_id}")).json()
    assert detail["backfill"] == [plan], "管理者在详情里看得到能补算的视图和会发起几次"

    out = (await client.post(f"/api/apps/{app_id}/views/latest/backfill")).json()
    assert (out["launched"], out["skipped"]) == (2, 1) and len(out["run_ids"]) == 2
    runs = [await wait(r) for r in out["run_ids"]]
    assert all(r.status == "succeeded" for r in runs), [r.error for r in runs]
    assert [(r.origin_kind, r.app_id, r.app_version, r.version, r.version_hash, r.run_class) for r in runs] == \
        [("app_backfill", app_id, 2, 2, h2, "formal")] * 2
    assert [r.input for r in runs] == [{"week": "W1"}, {"week": "W2"}]
    assert [r.origin["backfill_of"] for r in runs] == [old[0].id, old[1].id]
    assert runs[0].origin["source"] == "formal" and runs[0].origin["period_at"]

    again = (await client.post(f"/api/apps/{app_id}/views/latest/backfill")).json()
    assert (again["launched"], again["skipped"]) == (0, 3), "同一个输入不再补第二次"

    vm = (await client.get(f"/api/apps/{app_id}/pages/home")).json()
    assert vm["sources"]["latest"]["run_id"] == current.id, "补算的运行不是「最近一次」"
    trend = find(vm, "trend")["data"]
    assert trend["mode"] == "recompute"
    [segment] = trend["series"]
    assert segment["version"] == 2
    assert [p["run_id"] for p in segment["points"]] == [runs[0].id, runs[1].id, current.id], "按补的那一期排"
    assert [p.get("backfill", False) for p in segment["points"]] == [True, True, False]
    assert [p["value"] for p in segment["points"]] == [200, 200, 200]
    assert segment["points"][0]["as_of"].startswith(old[0].finished_at.isoformat()[:19])


async def test_backfill_needs_recompute_and_a_trend(client):
    wf, h1, h2 = await workflow()
    await formal(wf, 1, "W1")
    app_id = await upgraded_app(client, wf, h1, h2, "incomparable")
    refused = await client.post(f"/api/apps/{app_id}/views/latest/backfill")
    assert refused.status_code == 409 and refused.json()["code"] == "app_backfill_refused"
    assert "按新口径重算" in refused.json()["detail"]
    assert (await client.get(f"/api/apps/{app_id}")).json()["backfill"] == []
    gone = await client.get(f"/api/apps/{app_id}/views/nope/backfill")
    assert gone.status_code == 404 and gone.json()["code"] == "app_view_not_found"
    async with SessionLocal() as session:
        launched = (await session.execute(select(Run.id).where(Run.app_id == app_id))).scalars().all()
    assert launched == []


async def test_backfill_refuses_workflows_with_side_effects(client):
    """新版本里有人工节点：补算会对历史输入逐个给人开待办，收不回来。拒绝、详情里不列、一次也不发起。"""
    v2 = kpi(200)
    v2["nodes"].insert(3, node("ok", "human", "经理确认", mode="approve", title="确认周报"))
    v2["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(v2["nodes"], v2["nodes"][1:])]
    wf, h1, h2 = await workflow(v2)
    await formal(wf, 1, "W1")
    app_id = await upgraded_app(client, wf, h1, h2, "recompute")
    for res in (await client.get(f"/api/apps/{app_id}/views/latest/backfill"),
                await client.post(f"/api/apps/{app_id}/views/latest/backfill")):
        assert res.status_code == 409 and res.json()["code"] == "app_backfill_refused"
        assert "人工节点「经理确认」" in res.json()["detail"] and "不能按新口径补算" in res.json()["detail"]
    assert (await client.get(f"/api/apps/{app_id}")).json()["backfill"] == []
    async with SessionLocal() as session:
        launched = (await session.execute(select(Run.id).where(Run.app_id == app_id))).scalars().all()
    assert launched == []


async def test_backfill_writing_to_an_external_system_is_refused(client, monkeypatch):
    """调用工具节点绑的能力会写外部系统（capabilities.registry.writes_externally）：同样拒绝。"""
    from types import SimpleNamespace

    from app.capabilities import registry

    v2 = kpi(200)
    v2["nodes"].insert(3, node("pay", "tool", "付款", tool="erp_pay", args={}))
    v2["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(v2["nodes"], v2["nodes"][1:])]
    wf, h1, h2 = await workflow(v2)
    await formal(wf, 1, "W1")
    app_id = await upgraded_app(client, wf, h1, h2, "recompute")

    async def get_capability(session, name):
        return SimpleNamespace(name=name, effect="write", effect_system="write", endpoint_id=None) \
            if name == "erp_pay" else None

    monkeypatch.setattr(registry, "get_capability", get_capability)
    res = await client.post(f"/api/apps/{app_id}/views/latest/backfill")
    assert res.status_code == 409 and "会写外部系统（erp_pay）" in res.json()["detail"]


async def test_concurrent_backfills_launch_each_input_once(client):
    wf, h1, h2 = await workflow()
    for week in ("W1", "W2"):
        await formal(wf, 1, week)
    app_id = await upgraded_app(client, wf, h1, h2, "recompute")
    first, second = await asyncio.gather(client.post(f"/api/apps/{app_id}/views/latest/backfill"),
                                         client.post(f"/api/apps/{app_id}/views/latest/backfill"))
    outs = sorted((first.json(), second.json()), key=lambda o: -o["launched"])
    assert [(o["launched"], o["skipped"]) for o in outs] == [(2, 0), (0, 2)]
    for run_id in outs[0]["run_ids"]:
        await wait(run_id)
    async with SessionLocal() as session:
        launched = (await session.execute(select(Run.input).where(Run.app_id == app_id,
                                                                  Run.origin_kind == "app_backfill"))).scalars().all()
    assert sorted(i["week"] for i in launched) == ["W1", "W2"]