aps-agent/tests/golden/test_workflow_pi_contract.py

from __future__ import annotations

import asyncio
from pathlib import Path

import pytest

from server.agent_core import harness
from server.aps_domain.workflow import (
    execute_confirmed,
    handle_intent,
    stage_schedule_publish,
)
from server.contracts import IntentResult
from server.state.seed import seed_world
from tests.golden.test_flex_publish_mes_gate import _MemStore, _scheduled_world


def _run(store: _MemStore, intent: IntentResult, session_id: str = "workflow-contract"):
    return asyncio.run(handle_intent(store, session_id, intent, actor="test"))


def test_knowledge_catalog_requires_explicit_mode(tmp_path: Path):
    store = _MemStore(seed_world(), tmp_path / "knowledge-mode-checkpoints.json")

    reply = _run(
        store,
        IntentResult(
            intent="knowledge.query",
            params={"query": "知识清单"},
            confidence=1.0,
            source="LLM",
        ),
    )

    assert "mode=catalog" in reply.text
    assert "不再从 query 文本推断" in reply.text
    assert not reply.blocks


@pytest.mark.parametrize("enabled", [None, "false", 0, 1])
def test_skill_enable_requires_boolean_enabled_not_query(tmp_path: Path, enabled):
    store = _MemStore(seed_world(), tmp_path / f"skill-enable-{enabled}-checkpoints.json")

    reply = _run(
        store,
        IntentResult(
            intent="skill.enable",
            params={"skillId": "algo.stub", "enabled": enabled, "query": "停用 skill algo.stub"},
            confidence=1.0,
            source="LLM",
        ),
    )

    assert "enabled" in reply.text
    assert "不再从 query 文本推断" in reply.text
    assert not any(block.type == "confirm-card" for block in reply.blocks)


def test_skill_enable_accepts_explicit_false(tmp_path: Path):
    store = _MemStore(seed_world(), tmp_path / "skill-enable-false-checkpoints.json")

    reply = _run(
        store,
        IntentResult(
            intent="skill.enable",
            params={"skillId": "algo.stub", "enabled": False},
            confidence=1.0,
            source="LLM",
        ),
    )

    card = next(block for block in reply.blocks if block.type == "confirm-card")
    assert "停用算法 Skill algo.stub" in reply.text
    assert card.props["confirmId"]


@pytest.mark.parametrize("std_min", [None, "not-a-number", True, 0, float("inf")])
def test_flex_time_update_requires_structured_positive_minutes(tmp_path: Path, std_min):
    store = _MemStore(seed_world(), tmp_path / "time-update-checkpoints.json")

    reply = _run(
        store,
        IntentResult(
            intent="flex.time.update",
            params={"query": "把 OP-1 工时改成 30 分钟", "operationCode": "OP-1", "stdMin": std_min},
            confidence=1.0,
            source="LLM",
        ),
    )

    assert "stdMin" in reply.text or "operationCode" in reply.text
    assert not any(block.type == "confirm-card" for block in reply.blocks)


def test_flex_time_update_accepts_explicit_structured_values(tmp_path: Path):
    store = _MemStore(seed_world(), tmp_path / "time-update-valid-checkpoints.json")

    reply = _run(
        store,
        IntentResult(
            intent="flex.time.update",
            params={"operationCode": "OP-1", "stdMin": 30},
            confidence=1.0,
            source="LLM",
        ),
    )

    assert any(block.type == "confirm-card" for block in reply.blocks)


def test_sop_compile_requires_explicit_selection(tmp_path: Path):
    store = _MemStore(seed_world(), tmp_path / "sop-compile-checkpoints.json")

    reply = _run(
        store,
        IntentResult(
            intent="sop.compile",
            params={"query": "   "},
            confidence=1.0,
            source="LLM",
        ),
    )

    assert "请先选择要查看的作业标准" in reply.text
    assert not any(block.type == "confirm-card" for block in reply.blocks)


@pytest.mark.parametrize(
    "unsafe_version",
    [
        {"trialOnly": True, "productionReady": True},
        {"trialOnly": False, "productionReady": False},
    ],
)
def test_trial_or_nonproduction_version_cannot_stage_publish(
    tmp_path: Path, unsafe_version: dict,
):
    world, version_id = _scheduled_world()
    version = next(row for row in world["flexScheduleVersions"] if row["id"] == version_id)
    version.update(unsafe_version)
    store = _MemStore(world, tmp_path / "unsafe-publish-checkpoints.json")

    reply = stage_schedule_publish(
        store,
        session_id="unsafe-publish",
        actor="planner",
        track="flex",
        version_id=version_id,
    )

    assert "不可发布" in reply.text
    assert "trialOnly=true" in reply.text or "productionReady=false" in reply.text
    assert not any(block.type == "confirm-card" for block in reply.blocks)
    assert version["status"] == "DRAFT"


def test_execute_confirmed_rechecks_trial_only_before_publish(tmp_path: Path):
    world, version_id = _scheduled_world()
    version = next(row for row in world["flexScheduleVersions"] if row["id"] == version_id)
    version.update({"trialOnly": True, "productionReady": True})
    store = _MemStore(world, tmp_path / "unsafe-publish-execute-checkpoints.json")
    block = harness.stage_confirmation(
        "unsafe-publish-execute",
        "schedule.publish",
        {"track": "flex", "versionId": version_id, "evidenceRefs": [f"schedule-version:{version_id}"]},
        title="unsafe publish",
        summary_lines=["test"],
    )
    confirm_id = str(block.props["confirmId"])

    text = execute_confirmed(store, confirm_id, True, actor="planner")

    assert "发布被拒绝" in text
    assert version["status"] == "DRAFT"
    assert world.get("mesLinks") in (None, [])
    assert not any(row.get("action") == "schedule.publish" for row in world.get("auditEvents") or [])