from __future__ import annotations

import json
import os
import shutil
import sqlite3
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

from gyxx_flow.adapters import (
    COOKIE_SKIP_EXIT_CODE,
    WORKFLOW_ACCEPTANCE_RECIPIENT_OPEN_ID,
)
from gyxx_flow.modules.supply_chain import run as supply_run
from gyxx_flow.modules.supply_chain.orchestrator import config as supply_config
from gyxx_flow.modules.supply_chain.orchestrator import (
    mcp_workflow,
    pg_writer,
    runner,
)
from gyxx_flow.modules.supply_chain.orchestrator.scripts import (
    send_card_notification,
)

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SUPPLY_WRAPPERS = (
    PROJECT_ROOT
    / "src"
    / "gyxx_flow"
    / "modules"
    / "supply_chain"
    / "orchestrator"
    / "scripts"
)


@pytest.fixture
def acceptance_environment(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> Path:
    evidence = tmp_path / "evidence.jsonl"
    monkeypatch.setenv("GYXX_WORKFLOW_ACCEPTANCE", "1")
    monkeypatch.setenv("GYXX_ACCEPTANCE_EVIDENCE_FILE", str(evidence))
    monkeypatch.setenv("GYXX_DATA_ROOT", str(tmp_path / "data"))
    return evidence


def _forbidden(message: str):
    def fail(*_args, **_kwargs):
        pytest.fail(message)

    return fail


def test_frontend_supply_run_uses_stable_business_day_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026-08-03")

    assert mcp_workflow._resolve_workflow_id("purchase-confirmation", None) == (
        "purchase-confirmation_20260803"
    )
    assert mcp_workflow._resolve_workflow_id(
        "purchase-confirmation", "explicit-run"
    ) == "explicit-run"


def test_invalid_business_date_keeps_unique_run_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026/08/03")

    first = mcp_workflow._resolve_workflow_id("purchase-confirmation", None)
    second = mcp_workflow._resolve_workflow_id("purchase-confirmation", None)

    assert first.startswith("purchase-confirmation_")
    assert first != second


def test_local_hermes_profile_api_key_is_used_without_copying_secret(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    profile = tmp_path / "profiles" / "data-analyzer"
    profile.mkdir(parents=True)
    (profile / ".env").write_text(
        "API_SERVER_PORT=8642\nAPI_SERVER_KEY=local-only-key\n",
        encoding="utf-8",
    )
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    monkeypatch.delenv("GYXX_SUPPLY_HERMES_TOKEN", raising=False)
    monkeypatch.delenv("GYXX_HERMES_API_KEY", raising=False)
    monkeypatch.delenv("HERMES_API_KEY", raising=False)

    assert supply_config._hermes_profile_api_key("data-analyzer") == (
        "local-only-key"
    )


def test_stable_run_manifest_accepts_only_current_attempt_files(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    run_dir = tmp_path / "purchase-confirmation" / "run_id=stable-run"
    run_dir.mkdir(parents=True)
    old_file = run_dir / "old.csv"
    current_file = run_dir / "current.csv"
    old_file.write_text("old", encoding="utf-8")
    current_file.write_text("current", encoding="utf-8")
    (run_dir / "_run_manifest.json").write_text(
        json.dumps(
            {
                "run_id": "stable-run",
                "files": [old_file.name, current_file.name],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(mcp_workflow, "SHARED_DIR", tmp_path)

    assert mcp_workflow._validate_run_manifest(
        "purchase-confirmation", "stable-run", [str(current_file)]
    ) == (True, "")
    assert set(
        mcp_workflow._reusable_collected_files(
            "purchase-confirmation", "stable-run"
        )
    ) == {str(old_file), str(current_file)}


def test_replenishment_batch_uses_stable_business_day_ids(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from gyxx_flow.modules.supply_chain.orchestrator.scripts import batch_process

    monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026-08-03")

    assert batch_process._stable_run_id("replenishment") == (
        "replenishment_20260803"
    )
    assert batch_process._stable_run_id("replenishment-alert") == (
        "replenishment-alert_20260803"
    )

    monkeypatch.setenv("AUTOFLOW_TARGET_WORKFLOW", "replenishment-alert")
    assert batch_process._is_alert_only_target()

    monkeypatch.setenv("AUTOFLOW_TARGET_WORKFLOW", "replenishment")
    assert not batch_process._is_alert_only_target()


def test_alert_batch_stops_before_replenishment_side_effects(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from gyxx_flow.modules.supply_chain.orchestrator.scripts import batch_process

    threshold_file = tmp_path / "threshold_alert_data.json"
    monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026-08-03")
    monkeypatch.setenv("AUTOFLOW_TARGET_WORKFLOW", "replenishment-alert")
    monkeypatch.setattr(batch_process, "DATA_DIR", tmp_path)
    monkeypatch.setattr(batch_process, "THRESHOLD_ALERT_FILE", threshold_file)
    monkeypatch.setattr(batch_process, "_PG_IMPORT_OK", False)
    monkeypatch.setattr(batch_process, "_run_product_replenishment", lambda: None)
    monkeypatch.setattr(
        batch_process,
        "_ensure_threshold_alert_file",
        lambda: {"alert_count": 1, "alerts": [{"sku": "sku-1"}]},
    )
    monkeypatch.setattr(
        batch_process,
        "_pick_latest_xlsx",
        _forbidden("alert workflow must not enter replenishment persistence"),
    )

    assert batch_process.main() == 0
    payload = json.loads(threshold_file.read_text(encoding="utf-8"))
    assert payload["run_id"] == "replenishment-alert_20260803"
    assert payload["pg_ok"] is False


def _forbid_gateway_side_effects(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(
        mcp_workflow,
        "_acquire_workflow_lock",
        _forbidden("cookie preflight must run before workflow locking"),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_send_collector_heartbeat",
        _forbidden("cookie preflight must run before Feishu heartbeats"),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "call_hermes",
        _forbidden("cookie preflight must run before Hermes"),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_run_local_collection_script",
        _forbidden("cookie preflight must not start a browser collector"),
    )
    monkeypatch.setattr(
        mcp_workflow.requests,
        "Session",
        _forbidden("cookie preflight must not create an HTTP session"),
    )
    monkeypatch.setattr(
        mcp_workflow.requests,
        "post",
        _forbidden("cookie preflight must not send an HTTP request"),
    )


@pytest.mark.parametrize(
    ("workflow_name", "expected_script_id"),
    [
        (
            "purchase-confirmation",
            "supply_chain:orchestrator/scripts/PurchaseConfirmation.py",
        ),
        (
            "replenishment-alert",
            "supply_chain:orchestrator/scripts/ProductReplenishment.py",
        ),
        (
            "replenishment",
            "supply_chain:orchestrator/scripts/ProductReplenishment.py",
        ),
    ],
)
def test_gateway_skips_missing_cookie_before_all_external_effects(
    workflow_name: str,
    expected_script_id: str,
    acceptance_environment: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("GYXX_SUPPLY_ERP_PASSWORD", raising=False)
    _forbid_gateway_side_effects(monkeypatch)

    result = mcp_workflow.run_mcp_workflow(workflow_name, "acceptance-run")

    assert result["status"] == "skipped_cookie"
    assert result["workflow_id"] == "acceptance-run"
    assert result["cookie_preflight"]["script_id"] == expected_script_id
    assert result["cookie_preflight"]["status"] == "SKIPPED_COOKIE"
    assert result["reason"] == "browser state is missing"

    events = [
        json.loads(line)
        for line in acceptance_environment.read_text(encoding="utf-8").splitlines()
    ]
    gateway_event = next(
        item for item in events if item["event"] == "supply_gateway_cookie_skipped"
    )
    assert gateway_event["operation"] == f"supply.{workflow_name}.cookie_preflight"
    assert gateway_event["details"]["script_id"] == expected_script_id
    assert gateway_event["details"]["workflow_id"] == "acceptance-run"


def test_profile_without_required_platform_cookie_is_still_skipped(
    acceptance_environment: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("GYXX_SUPPLY_ERP_PASSWORD", raising=False)
    binding = mcp_workflow._acceptance_binding_for_workflow("purchase-confirmation")
    cookie_db = binding.profile_dir / "Default" / "Network" / "Cookies"
    cookie_db.parent.mkdir(parents=True)
    with sqlite3.connect(cookie_db) as connection:
        connection.execute("CREATE TABLE cookies (expires_utc INTEGER)")
        connection.execute(
            "INSERT INTO cookies (expires_utc) VALUES (?)",
            (999_999_999_999_999_999,),
        )
    (binding.profile_dir / "Local State").write_text("{}", encoding="utf-8")
    _forbid_gateway_side_effects(monkeypatch)

    result = mcp_workflow.run_mcp_workflow(
        "purchase-confirmation",
        "nonempty-profile-run",
    )

    assert result["status"] == "skipped_cookie"
    assert result["cookie_preflight"]["script_id"].endswith(
        "/PurchaseConfirmation.py"
    )
    assert result["reason"] == "browser state is missing"
    events = [
        json.loads(line)
        for line in acceptance_environment.read_text(encoding="utf-8").splitlines()
    ]
    assert events[-1]["event"] == "supply_gateway_cookie_skipped"
    assert events[-1]["details"]["reason"] == "browser state is missing"


def test_supply_credentials_allow_preflight_without_cookie_or_live_cdp(
    acceptance_environment: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    del acceptance_environment
    monkeypatch.setenv("GYXX_SUPPLY_ERP_PASSWORD", "configured-for-test")
    lock_calls: list[tuple[str, str]] = []

    def reject_lock(workflow_name: str, workflow_id: str) -> tuple[bool, str]:
        lock_calls.append((workflow_name, workflow_id))
        return False, "stop-after-preflight"

    monkeypatch.setattr(mcp_workflow, "_acquire_workflow_lock", reject_lock)

    result = mcp_workflow.run_mcp_workflow(
        "purchase-confirmation",
        "credential-fallback-run",
    )

    assert result["status"] == "failed"
    assert result["error"] == "stop-after-preflight"
    assert lock_calls == [("purchase-confirmation", "credential-fallback-run")]


def test_analyzer_facts_include_authoritative_absolute_collected_files(
    tmp_path: Path,
) -> None:
    collected_file = tmp_path / "purchase_confirmation.csv"

    facts = mcp_workflow._build_analyze_facts(
        "purchase-confirmation",
        [str(collected_file)],
    )

    assert facts == [
        "AUTHORITATIVE COLLECTED FILES (absolute paths): "
        f"{collected_file.resolve()}"
    ]


def test_close_workflow_browser_terminates_only_bound_cdp_tree(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class FakeProcess:
        def __init__(
            self,
            pid: int,
            cmdline: list[str],
            children: list[FakeProcess] | None = None,
        ) -> None:
            self.pid = pid
            self.info = {"pid": pid, "cmdline": cmdline}
            self._children = children or []
            self.terminated = False
            self.killed = False

        def children(self, *, recursive: bool) -> list[FakeProcess]:
            assert recursive is True
            return self._children

        def terminate(self) -> None:
            self.terminated = True

        def kill(self) -> None:
            self.killed = True

    child = FakeProcess(102, ["chrome.exe", "--type=renderer"])
    bound = FakeProcess(
        101,
        ["chrome.exe", "--remote-debugging-port=22116"],
        [child],
    )
    unrelated = FakeProcess(
        201,
        ["chrome.exe", "--remote-debugging-port=22115"],
    )
    waited: list[tuple[int, ...]] = []

    monkeypatch.setattr(
        mcp_workflow,
        "_acceptance_binding_for_workflow",
        lambda _workflow_name: SimpleNamespace(cdp_port=22116),
    )
    monkeypatch.setattr(
        mcp_workflow.psutil,
        "process_iter",
        lambda _attrs: [bound, unrelated],
    )

    def fake_wait_procs(processes, *, timeout: int):
        assert timeout == 5
        waited.append(tuple(process.pid for process in processes))
        return list(processes), []

    monkeypatch.setattr(mcp_workflow.psutil, "wait_procs", fake_wait_procs)

    closed = mcp_workflow._close_workflow_browser("purchase-confirmation")

    assert closed == [101, 102]
    assert bound.terminated is True
    assert child.terminated is True
    assert unrelated.terminated is False
    assert unrelated.killed is False
    assert waited == [(102, 101)]


def test_close_workflow_browser_ignores_other_workflows(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        mcp_workflow,
        "_acceptance_binding_for_workflow",
        _forbidden("unmanaged workflow must not resolve a browser binding"),
    )

    assert mcp_workflow._close_workflow_browser("purchase-order-update") == []


def test_purchase_order_update_block_precedes_cookie_preflight(
    acceptance_environment: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        mcp_workflow,
        "_acceptance_cookie_preflight",
        _forbidden("ERP mutation block must take precedence over cookie inspection"),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_acquire_workflow_lock",
        _forbidden("blocked ERP workflow must not acquire a lock"),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_send_collector_heartbeat",
        _forbidden("blocked ERP workflow must not send a heartbeat"),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_send_purchase_order_update_notification",
        _forbidden("blocked ERP workflow must not notify a user or group"),
    )

    result = mcp_workflow.run_mcp_workflow(
        "purchase-order-update",
        "blocked-erp-run",
    )

    assert result["status"] == "blocked_business_mutation"
    assert "business_mutation_blocked" in acceptance_environment.read_text(
        encoding="utf-8"
    )


def test_analyzer_prompt_does_not_rely_on_two_hop_acceptance_environment(
    acceptance_environment: Path,
) -> None:
    prompt = mcp_workflow._build_analyze_prompt("replenishment")

    assert "Hermes 网关不会继承编排进程的验收环境变量" in prompt
    assert "不得调用 insert_replenishment_bitable.py" in prompt
    assert prompt.rfind("【验收最终覆盖规则】") > prompt.rfind(
        "多维表写入统一走 Python 脚本"
    )
    assert mcp_workflow.ANALYZER_OWNER_OPEN_ID in prompt
    assert "ou_8ee224968aa26a74c7d30ba27fed5eeb" not in prompt


@pytest.mark.parametrize(
    "response_text",
    [
        "没有向业务用户发送任何过程或进度消息，仅投递最终业务通知。",
        "未向目标用户推送额外进度更新，只发送最终结果。",
        (
            "No additional process/progress messages were sent to users; "
            "final business notification delivered."
        ),
    ],
)
def test_notification_policy_allows_explicitly_negated_process_messages(
    response_text: str,
) -> None:
    assert not mcp_workflow._contains_process_notification_violation(response_text)


@pytest.mark.parametrize(
    "response_text",
    [
        "已向业务用户发送进度通知。",
        "Progress notification sent to the target user.",
    ],
)
def test_notification_policy_rejects_process_messages_sent_to_users(
    response_text: str,
) -> None:
    assert mcp_workflow._contains_process_notification_violation(response_text)


def test_notification_receipt_rejects_empty_message_ids() -> None:
    with pytest.raises(ValueError, match="non-empty message_id"):
        pg_writer.mark_notification_sent("replenishment", "run-1", [])


def test_notification_disabled_callback_accepts_an_explicit_empty_receipt(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    statements: list[tuple[str, object]] = []

    class FakeConnection:
        def cursor(self):
            return self

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return None

        def execute(self, sql, params):
            statements.append((sql, params))

        def commit(self):
            return None

        def close(self):
            return None

    monkeypatch.setattr(pg_writer, "_conn", FakeConnection)

    pg_writer.mark_notification_sent(
        "replenishment",
        "run-disabled",
        [],
        notification_disabled=True,
        expected_recipients=[],
    )

    assert len(statements) == 1
    sql, params = statements[0]
    assert "notification_disabled" in sql
    assert "notification_disabled_at" in sql
    assert "notification_message_ids" not in sql
    assert params == ("replenishment", "run-disabled")


def test_notification_receipt_requires_complete_explicit_recipient_mapping(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    statements: list[tuple[str, object]] = []

    class FakeConnection:
        def cursor(self):
            return self

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return None

        def execute(self, sql, params):
            statements.append((sql, params))

        def commit(self):
            return None

        def close(self):
            return None

    monkeypatch.setattr(pg_writer, "_conn", FakeConnection)
    recipients = ["ou_first", "ou_second", "ou_first"]

    with pytest.raises(ValueError, match="recipient_message_ids"):
        pg_writer.mark_notification_sent(
            "replenishment",
            "run-partial",
            message_ids=["mid-first"],
            expected_recipients=recipients,
            recipient_message_ids={"ou_first": "mid-first"},
        )

    assert statements == []

    pg_writer.mark_notification_sent(
        "replenishment",
        "run-complete",
        expected_recipients=recipients,
        recipient_message_ids={
            "ou_first": "mid-first",
            "ou_second": "mid-second",
        },
    )

    assert len(statements) == 3
    run_statement = next(
        (sql, params)
        for sql, params in statements
        if "UPDATE workflow_runs" in sql
    )
    assert "notification_recipient_message_ids" in run_statement[0]
    assert json.loads(run_statement[1][1]) == {
        "ou_first": "mid-first",
        "ou_second": "mid-second",
    }
    assert "'notification_disabled', false" in run_statement[0]


def test_reenabled_notification_writes_receipts_without_disabled_overwrite(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    statements: list[tuple[str, object]] = []

    class FakeConnection:
        def cursor(self):
            return self

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return None

        def execute(self, sql, params):
            statements.append((sql, params))

        def commit(self):
            return None

        def close(self):
            return None

    monkeypatch.setattr(pg_writer, "_conn", FakeConnection)

    pg_writer.mark_notification_sent(
        "purchase-confirmation",
        "same-run",
        notification_disabled=True,
        expected_recipients=[],
    )
    pg_writer.mark_notification_sent(
        "purchase-confirmation",
        "same-run",
        expected_recipients=["ou_owner"],
        recipient_message_ids={"ou_owner": "mid-owner"},
    )

    disabled_sql = statements[0][0]
    assert "notification_message_ids" not in disabled_sql
    enabled_run_sql = next(
        sql
        for sql, _params in statements[1:]
        if "UPDATE workflow_runs" in sql
    )
    assert "notification_recipient_message_ids" in enabled_run_sql
    assert "'notification_disabled', false" in enabled_run_sql


def test_supply_route_resolution_does_not_recurse_and_keeps_all_recipients(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[tuple[str, tuple[str, ...]]] = []
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)

    def fake_resolve(workflow_id, defaults):
        calls.append((workflow_id, tuple(defaults)))
        return SimpleNamespace(enabled=True, open_ids=("ou_first", "ou_second"))

    monkeypatch.setattr(mcp_workflow, "resolve_notification_route", fake_resolve)

    enabled, recipients = mcp_workflow._resolve_business_notification_route(
        "purchase-confirmation"
    )

    assert enabled is True
    assert recipients == ("ou_first", "ou_second")
    assert calls == [
        (
            "supply.purchase_confirmation.daily",
            mcp_workflow.PURCHASE_CONFIRMATION_TARGETS,
        )
    ]


def test_supply_sender_hard_rejects_disabled_or_drifted_route_snapshot(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(send_card_notification, "STATE_ROOT", tmp_path)

    disabled_run = "disabled-run"
    disabled_path = send_card_notification._notification_snapshot_path(
        "replenishment",
        disabled_run,
    )
    disabled_path.parent.mkdir(parents=True)
    disabled_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": "replenishment",
                "run_id": disabled_run,
                "enabled": False,
                "recipients": [],
                "acceptance": False,
            }
        ),
        encoding="utf-8",
    )

    with pytest.raises(RuntimeError, match="route is disabled"):
        send_card_notification._resolved_receivers(
            "replenishment",
            disabled_run,
            ["ou_launch123"],
        )

    enabled_run = "enabled-run"
    enabled_path = send_card_notification._notification_snapshot_path(
        "replenishment",
        enabled_run,
    )
    enabled_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": "replenishment",
                "run_id": enabled_run,
                "enabled": True,
                "recipients": ["ou_launch123"],
                "acceptance": False,
            }
        ),
        encoding="utf-8",
    )
    with pytest.raises(RuntimeError, match="snapshot mismatch"):
        send_card_notification._resolved_receivers(
            "replenishment",
            enabled_run,
            ["ou_arbitrary123"],
        )


def test_supply_sender_keeps_exact_launch_snapshot_and_partial_delivery_mapping(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(send_card_notification, "STATE_ROOT", tmp_path)
    run_id = "run-1"
    launch_recipients = ["ou_first123", "ou_middle123", "ou_last1234"]
    snapshot_path = send_card_notification._notification_snapshot_path(
        "replenishment",
        run_id,
    )
    snapshot_path.parent.mkdir(parents=True)
    snapshot_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": "replenishment",
                "run_id": run_id,
                "enabled": True,
                "recipients": launch_recipients,
                "acceptance": False,
            }
        ),
        encoding="utf-8",
    )
    def fake_send(open_id, _card, _key):
        if open_id == "ou_middle123":
            raise RuntimeError("middle failed")
        return f"om_{open_id.removeprefix('ou_')}"

    monkeypatch.setattr(send_card_notification, "_send_card", fake_send)
    monkeypatch.setattr(send_card_notification.pg_writer, "is_available", lambda: True)

    result = send_card_notification.run(
        launch_recipients,
        {"elements": []},
        workflow="replenishment",
        run_id=run_id,
        idempotency_key="replenishment:run-1",
    )

    assert result == {
        "message_ids": ["om_first123", "om_last1234"],
        "recipient_message_ids": {
            "ou_first123": "om_first123",
            "ou_last1234": "om_last1234",
        },
        "errors": [{"open_id": "ou_middle123", "error": "middle failed"}],
    }


@pytest.mark.parametrize(
    "workflow",
    ("purchase-confirmation", "replenishment-alert", "replenishment"),
)
def test_supply_sender_persists_and_verifies_complete_receipts(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    workflow: str,
) -> None:
    monkeypatch.setattr(send_card_notification, "STATE_ROOT", tmp_path)
    run_id = "run-complete"
    recipients = ["ou_first123", "ou_second123"]
    snapshot_path = send_card_notification._notification_snapshot_path(
        workflow,
        run_id,
    )
    snapshot_path.parent.mkdir(parents=True)
    snapshot_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": workflow,
                "run_id": run_id,
                "enabled": True,
                "recipients": recipients,
                "acceptance": False,
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(
        send_card_notification,
        "_send_card",
        lambda open_id, _card, _key: f"om_{open_id.removeprefix('ou_')}",
    )
    monkeypatch.setattr(send_card_notification.pg_writer, "is_available", lambda: True)
    persisted: list[tuple[tuple[object, ...], dict[str, object]]] = []
    monkeypatch.setattr(
        send_card_notification.pg_writer,
        "mark_notification_sent",
        lambda *args, **kwargs: persisted.append((args, kwargs)),
    )
    monkeypatch.setattr(
        send_card_notification.pg_writer,
        "notification_already_sent",
        lambda checked_workflow, checked_run_id, *, expected_recipients: (
            checked_workflow == workflow
            and checked_run_id == run_id
            and tuple(expected_recipients) == tuple(recipients)
        ),
    )

    result = send_card_notification.run(
        recipients,
        {"elements": []},
        workflow=workflow,
        run_id=run_id,
        idempotency_key=f"{workflow}:{run_id}",
    )

    assert result["errors"] == []
    assert persisted == [
        (
            (workflow, run_id),
            {
                "expected_recipients": recipients,
                "recipient_message_ids": {
                    "ou_first123": "om_first123",
                    "ou_second123": "om_second123",
                },
            },
        )
    ]


def test_supply_sender_fails_closed_when_complete_receipt_is_not_readable(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(send_card_notification, "STATE_ROOT", tmp_path)
    workflow = "replenishment-alert"
    run_id = "run-unverified"
    recipients = ["ou_owner123"]
    snapshot_path = send_card_notification._notification_snapshot_path(
        workflow,
        run_id,
    )
    snapshot_path.parent.mkdir(parents=True)
    snapshot_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": workflow,
                "run_id": run_id,
                "enabled": True,
                "recipients": recipients,
                "acceptance": False,
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(
        send_card_notification,
        "_send_card",
        lambda _open_id, _card, _key: "om_owner123",
    )
    monkeypatch.setattr(send_card_notification.pg_writer, "is_available", lambda: True)
    monkeypatch.setattr(
        send_card_notification.pg_writer,
        "mark_notification_sent",
        lambda *_args, **_kwargs: None,
    )
    monkeypatch.setattr(
        send_card_notification.pg_writer,
        "notification_already_sent",
        lambda *_args, **_kwargs: False,
    )

    with pytest.raises(RuntimeError, match="could not be verified"):
        send_card_notification.run(
            recipients,
            {"elements": []},
            workflow=workflow,
            run_id=run_id,
            idempotency_key=f"{workflow}:{run_id}",
        )


def test_supply_sender_refuses_send_when_receipt_store_is_unavailable(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    workflow = "purchase-confirmation"
    run_id = "run-no-pg"
    recipients = ["ou_owner123"]
    monkeypatch.setattr(send_card_notification, "STATE_ROOT", tmp_path)
    snapshot_path = send_card_notification._notification_snapshot_path(workflow, run_id)
    snapshot_path.parent.mkdir(parents=True)
    snapshot_path.write_text(
        json.dumps({"schema_version": 1, "workflow": workflow, "run_id": run_id, "enabled": True, "recipients": recipients, "acceptance": False}),
        encoding="utf-8",
    )
    sent: list[str] = []
    monkeypatch.setattr(send_card_notification, "_send_card", lambda *_args: sent.append("sent"))
    monkeypatch.setattr(send_card_notification.pg_writer, "is_available", lambda: False)

    with pytest.raises(RuntimeError, match="receipt store is unavailable"):
        send_card_notification.run(recipients, {"elements": []}, workflow=workflow, run_id=run_id)

    assert sent == []


def test_supply_acceptance_snapshot_survives_missing_wrapper_environment(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    acceptance_recipient = supply_config.FEISHU_CONFIG["apps"]["analyzer"][
        "owner_open_id"
    ]
    monkeypatch.setenv("GYXX_WORKFLOW_ACCEPTANCE", "1")
    monkeypatch.setenv("GYXX_FEISHU_TABLE_WRITE_DISABLED", "1")
    monkeypatch.setenv("GYXX_COOKIE_INVALID_SKIP", "1")
    monkeypatch.setenv(
        "GYXX_NOTIFICATION_RECIPIENT_OPEN_ID",
        WORKFLOW_ACCEPTANCE_RECIPIENT_OPEN_ID,
    )
    monkeypatch.setattr(mcp_workflow, "STATE_ROOT", tmp_path)
    monkeypatch.setattr(send_card_notification, "STATE_ROOT", tmp_path)
    monkeypatch.setattr(
        mcp_workflow,
        "_resolve_business_notification_route",
        lambda _workflow: (True, (acceptance_recipient,)),
    )
    run_id = "acceptance-run"
    snapshot = mcp_workflow._load_or_create_notification_snapshot(
        "replenishment",
        run_id,
    )
    assert snapshot == (True, (acceptance_recipient,), True)
    prompt = mcp_workflow._build_analyze_prompt(
        "replenishment",
        notification_route=(snapshot[0], snapshot[1]),
        notification_acceptance=snapshot[2],
        run_id=run_id,
    )
    assert "Hermes must not send any Feishu message" in prompt
    assert "--acceptance-override" not in prompt
    assert acceptance_recipient in prompt

    # The analyzer wrapper is a separate process and does not inherit the
    # orchestrator's acceptance environment. The external run snapshot remains the
    # sole trust source even if the mutable page route changes after launch.
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
    monkeypatch.setattr(
        mcp_workflow,
        "_resolve_business_notification_route",
        lambda _workflow: (True, ("ou_newroute123",)),
    )

    assert mcp_workflow._load_or_create_notification_snapshot(
        "replenishment",
        run_id,
    ) == snapshot
    assert send_card_notification._resolved_receivers(
        "replenishment",
        run_id,
        [acceptance_recipient],
    ) == [acceptance_recipient]
    with pytest.raises(RuntimeError, match="snapshot mismatch"):
        send_card_notification._resolved_receivers(
            "replenishment",
            run_id,
            ["ou_newroute123"],
        )


def test_supply_sender_uses_analyzer_bot_with_stable_business_key(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    sends: list[dict[str, object]] = []

    def fake_send(**kwargs):
        sends.append(kwargs)
        return {"message_id": f"om_{len(sends)}"}

    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
    monkeypatch.setattr(send_card_notification, "send_lark_bot_message", fake_send)

    send_card_notification._send_card(
        "ou_first123",
        {"elements": []},
        "replenishment:run-1",
    )
    send_card_notification._send_card(
        "ou_first123",
        {"elements": []},
        "replenishment:run-1",
    )
    send_card_notification._send_card(
        "ou_second123",
        {"elements": []},
        "replenishment:run-1",
    )

    assert sends == [
        {
            "user_id": "ou_first123",
            "content": {"elements": []},
            "msg_type": "interactive",
            "idempotency_key": "replenishment:run-1",
        },
        {
            "user_id": "ou_first123",
            "content": {"elements": []},
            "msg_type": "interactive",
            "idempotency_key": "replenishment:run-1",
        },
        {
            "user_id": "ou_second123",
            "content": {"elements": []},
            "msg_type": "interactive",
            "idempotency_key": "replenishment:run-1",
        },
    ]


def test_supply_heartbeat_uses_collector_bot_and_collector_open_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    sends: list[dict[str, object]] = []

    def forbidden_resolve(*_args, **_kwargs):
        pytest.fail("heartbeat must not translate a collector-app open_id")

    def fake_send(**kwargs):
        sends.append(kwargs)
        return {"message_id": "om_heartbeat1"}

    monkeypatch.setattr(
        mcp_workflow,
        "resolve_notification_recipients",
        forbidden_resolve,
    )
    monkeypatch.setattr(mcp_workflow, "send_lark_bot_message", fake_send)
    result = {
        "workflow": "replenishment",
        "workflow_name": "replenishment 补货建议",
        "workflow_id": "run-1",
        "status": "running",
        "current_phase": "collecting",
        "heartbeat_response": "",
    }

    mcp_workflow._send_collector_heartbeat(result, phase="progress")

    assert sends == [
        {
            "user_id": mcp_workflow.COLLECTOR_OWNER_OPEN_ID,
            "content": mcp_workflow._build_heartbeat_card(result, phase="progress"),
            "msg_type": "interactive",
            "profile": mcp_workflow.COLLECTOR_FEISHU["app_id"],
            "timeout": 30,
        }
    ]
    assert "feishu_msg_id=om_heartbeat1" in result["heartbeat_response"]


def test_purchase_order_update_notification_uses_chat_id_and_user_id(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    result_file = tmp_path / "update_result.json"
    result_file.write_text(
        json.dumps(
            {
                "total_updated": 1,
                "target_date": "2026-08-18",
                "sku_list": ["SKU-1"],
                "updated_records": [
                    {"sku": "SKU-1", "name": "旅行箱", "spec": "20寸黑色"}
                ],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    sends: list[dict[str, object]] = []

    def fake_resolve(recipients, *, app_profile=None):
        assert tuple(recipients) == (mcp_workflow.ANALYZER_OWNER_OPEN_ID,)
        assert app_profile == "hermes-analyzer"
        return ("ou_analyzerowner123",)

    def fake_send(**kwargs):
        sends.append(kwargs)
        return {"message_id": f"om_delivery{len(sends)}"}

    monkeypatch.setattr(mcp_workflow, "resolve_notification_recipients", fake_resolve)
    monkeypatch.setattr(mcp_workflow, "send_lark_bot_message", fake_send)

    receipt = mcp_workflow._send_purchase_order_update_notification(
        [str(result_file)],
        "purchase-order-update_20260810",
    )

    assert sends[0]["chat_id"] == mcp_workflow.SHENDAN_GROUP_CHAT_ID
    assert "user_id" not in sends[0]
    assert sends[1]["user_id"] == "ou_analyzerowner123"
    assert "chat_id" not in sends[1]
    assert all(send["profile"] == "hermes-analyzer" for send in sends)
    assert all(send["msg_type"] == "interactive" for send in sends)
    assert {
        send["idempotency_key"] for send in sends
    } == {"purchase-order-update:purchase-order-update_20260810"}
    card_text = json.dumps(sends[0]["content"], ensure_ascii=False)
    assert all(
        value in card_text
        for value in ("SKU-1", "旅行箱", "20寸黑色", "2026-08-18", "1")
    )
    assert receipt == {
        "message_ids": ["om_delivery1", "om_delivery2"],
        "recipient_message_ids": {
            mcp_workflow.SHENDAN_GROUP_CHAT_ID: "om_delivery1",
            "ou_analyzerowner123": "om_delivery2",
        },
        "errors": [],
    }


def test_supply_analyzer_contract_has_no_feishu_message_plugin() -> None:
    configured_steps = [
        step
        for workflow in supply_config.WORKFLOWS.values()
        for step in workflow.get("analyzer_scripts", [])
    ]

    assert "plugin:feishu.im.send_message" not in configured_steps
    assert "adapter:send_lark_bot_message" in configured_steps


@pytest.mark.parametrize(
    "workflow_name",
    ["purchase-confirmation", "replenishment-alert", "replenishment"],
)
def test_supply_disabled_route_forbids_send_and_requires_empty_callback(
    workflow_name: str,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
    monkeypatch.setattr(
        mcp_workflow,
        "_resolve_business_notification_route",
        lambda _workflow_name: (False, ()),
    )

    prompt = mcp_workflow._build_analyze_prompt(workflow_name)

    assert "动态通知配置已禁用" in prompt
    assert "不得向任何用户或群聊发送业务消息" in prompt
    assert "message_ids=[]" in prompt
    assert "notification_disabled=True" in prompt
    assert "Dynamic notification routing is disabled" in prompt
    assert "No final business notification is required or allowed" in prompt
    assert "must be sent as exactly ONE" not in prompt


@pytest.mark.parametrize(
    "workflow_name",
    ["purchase-confirmation", "replenishment-alert", "replenishment"],
)
def test_supply_enabled_prompt_forbids_llm_delivery_and_keeps_route_context(
    workflow_name: str,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
    monkeypatch.setattr(
        mcp_workflow,
        "_resolve_business_notification_route",
        _forbidden("prompt must preserve the route snapshot resolved at launch"),
    )

    prompt = mcp_workflow._build_analyze_prompt(
        workflow_name,
        notification_route=(True, ("ou_first", "ou_second")),
        run_id="run-20260807",
    )

    assert "Hermes must not send any Feishu message" in prompt
    assert "Python orchestrator alone delivers the final business card" in prompt
    assert f"--workflow {workflow_name}" not in prompt
    assert f"p.mark_notification_sent('{workflow_name}','run-20260807'" not in prompt


def test_notification_receipt_requires_persisted_message_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        pg_writer,
        "notification_already_sent",
        lambda *_args, **_kwargs: False,
    )
    assert not mcp_workflow._notification_receipt_exists("replenishment", "run-1")

    monkeypatch.setattr(
        pg_writer,
        "notification_already_sent",
        lambda *_args, **_kwargs: True,
    )
    assert mcp_workflow._notification_receipt_exists("replenishment", "run-1")


def test_notification_receipt_check_forwards_current_route_recipients(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured: dict[str, object] = {}

    def fake_already_sent(workflow, run_id, *, expected_recipients=None):
        captured.update(
            workflow=workflow,
            run_id=run_id,
            expected_recipients=expected_recipients,
        )
        return True

    monkeypatch.setattr(pg_writer, "notification_already_sent", fake_already_sent)

    assert mcp_workflow._notification_receipt_exists(
        "replenishment",
        "run-routed",
        ("ou_first", "ou_second"),
    )
    assert captured == {
        "workflow": "replenishment",
        "run_id": "run-routed",
        "expected_recipients": ("ou_first", "ou_second"),
    }


@pytest.mark.parametrize(
    ("mapping", "recipients", "expected"),
    [
        (
            {"ou_first": "mid-first"},
            ("ou_first", "ou_second"),
            False,
        ),
        (
            {"ou_first": "mid-first", "ou_second": "mid-second"},
            ("ou_first", "ou_second"),
            True,
        ),
        (
            {"ou_first": "mid-first", "ou_second": "mid-second"},
            ("ou_new",),
            False,
        ),
    ],
)
def test_notification_already_sent_requires_current_recipient_coverage(
    mapping: dict[str, str],
    recipients: tuple[str, ...],
    expected: bool,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class FakeCursor:
        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return None

        def execute(self, _sql, _params):
            return None

        def fetchone(self):
            return ({"notification_recipient_message_ids": mapping},)

    class FakeConnection:
        def cursor(self):
            return FakeCursor()

        def close(self):
            return None

    monkeypatch.setattr(pg_writer, "_conn", FakeConnection)

    assert (
        pg_writer.notification_already_sent(
            "replenishment",
            "run-multi",
            expected_recipients=recipients,
        )
        is expected
    )


def test_notification_receipt_falls_back_to_zero_row_workflow_batch(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class FakeCursor:
        def __init__(self) -> None:
            self.fetches = iter((None, None, (1,)))

        def __enter__(self):
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def execute(self, _sql: str, _params: object) -> None:
            return None

        def fetchone(self):
            return next(self.fetches)

    class FakeConnection:
        def __init__(self) -> None:
            self.cursor_instance = FakeCursor()

        def cursor(self) -> FakeCursor:
            return self.cursor_instance

        def close(self) -> None:
            return None

    monkeypatch.setattr(pg_writer, "_conn", FakeConnection)

    assert pg_writer.notification_already_sent(
        "purchase-confirmation", "zero-row-run"
    )


def test_production_mode_preserves_the_existing_lock_first_flow(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
    monkeypatch.setattr(
        mcp_workflow,
        "_acceptance_cookie_preflight",
        _forbidden("production mode must not run the acceptance preflight"),
    )
    lock_calls: list[tuple[str, str]] = []

    def reject_lock(workflow_name: str, workflow_id: str) -> tuple[bool, str]:
        lock_calls.append((workflow_name, workflow_id))
        return False, "already running"

    monkeypatch.setattr(mcp_workflow, "_acquire_workflow_lock", reject_lock)

    result = mcp_workflow.run_mcp_workflow("replenishment", "production-run")

    assert result["status"] == "failed"
    assert result["error"] == "already running"
    assert lock_calls == [("replenishment", "production-run")]


@pytest.mark.parametrize(
    ("status", "expected"),
    [
        ("completed", 0),
        ("skipped_cookie", COOKIE_SKIP_EXIT_CODE),
        ("blocked_business_mutation", runner.BLOCKED_EXIT_CODE),
        ("failed", 1),
    ],
)
def test_both_supply_clis_propagate_structured_result_exit_codes(
    status: str,
    expected: int,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    result = {"workflow": "replenishment", "status": status}
    monkeypatch.setattr(runner, "run_mcp", lambda _workflow: result)

    monkeypatch.setattr(
        sys,
        "argv",
        ["runner.py", "mcp", "replenishment"],
    )
    assert runner.main() == expected

    monkeypatch.setattr(
        sys,
        "argv",
        ["run.py", "mcp-run", "replenishment"],
    )
    assert supply_run.main() == expected
    capsys.readouterr()


def test_batch_exit_code_uses_failure_then_block_then_cookie_precedence() -> None:
    assert runner.result_exit_code(
        {
            "results": [
                {"status": "completed"},
                {"status": "skipped_cookie"},
            ]
        }
    ) == COOKIE_SKIP_EXIT_CODE
    assert runner.result_exit_code(
        {
            "results": [
                {"status": "skipped_cookie"},
                {"status": "blocked_business_mutation"},
            ]
        }
    ) == runner.BLOCKED_EXIT_CODE
    assert runner.result_exit_code(
        {
            "results": [
                {"status": "blocked_business_mutation"},
                {"status": "failed"},
            ]
        }
    ) == 1


def test_idempotent_replay_uses_a_distinct_skip_exit_code() -> None:
    assert runner.result_exit_code(
        {
            "workflow": "purchase-confirmation",
            "status": "completed",
            "idempotent_replay": True,
        }
    ) == runner.IDEMPOTENT_REPLAY_EXIT_CODE


@pytest.mark.parametrize("value", ["1", "true", "TRUE", "yes", "on"])
def test_force_refresh_environment_values(
    value: str,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("GYXX_FORCE_REFRESH", value)
    assert mcp_workflow._force_refresh_requested() is True


def test_force_refresh_is_opt_in(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("GYXX_FORCE_REFRESH", raising=False)
    assert mcp_workflow._force_refresh_requested() is False


def test_force_refresh_recollects_and_notifies_deterministically_instead_of_reusing(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
    monkeypatch.setenv("GYXX_FORCE_REFRESH", "true")
    collected_file = tmp_path / "purchase-confirmation.csv"
    collected_file.write_text("sku\n1001\n", encoding="utf-8")
    collection_calls: list[str] = []
    notification_calls: list[tuple[str, str, list[str]]] = []
    receipt_calls: list[tuple[str, str, tuple[str, ...] | None]] = []

    monkeypatch.setattr(mcp_workflow, "_acquire_workflow_lock", lambda *_args: (True, ""))
    monkeypatch.setattr(mcp_workflow, "_release_workflow_lock", lambda *_args: None)
    monkeypatch.setattr(mcp_workflow, "_acquire_collection_lock", lambda *_args: (True, ""))
    monkeypatch.setattr(mcp_workflow, "_release_collection_lock", lambda *_args: None)
    monkeypatch.setattr(
        mcp_workflow,
        "_reusable_collected_files",
        _forbidden("force refresh must not inspect reusable collection files"),
    )
    monkeypatch.setattr(mcp_workflow, "_clear_shared_outputs", lambda *_args: [])

    def collect(workflow_name: str, workflow_id: str) -> tuple[bool, str]:
        collection_calls.append(f"{workflow_name}:{workflow_id}")
        return True, "fresh collection"

    monkeypatch.setattr(mcp_workflow, "_run_local_collection_script", collect)
    monkeypatch.setattr(
        mcp_workflow,
        "_ensure_shared_outputs",
        lambda *_args: [str(collected_file)],
    )
    monkeypatch.setattr(mcp_workflow, "_validate_run_manifest", lambda *_args: (True, ""))
    monkeypatch.setattr(mcp_workflow, "_extract_pending_run_id", lambda *_args: None)
    monkeypatch.setattr(
        mcp_workflow,
        "_load_or_create_notification_snapshot",
        lambda *_args: (True, ("ou_test",), False),
    )

    def receipt_exists(
        workflow_name: str,
        run_id: str,
        recipients: tuple[str, ...] | None = None,
    ) -> bool:
        receipt_calls.append((workflow_name, run_id, recipients))
        return True

    monkeypatch.setattr(mcp_workflow, "_notification_receipt_exists", receipt_exists)

    def send_purchase_confirmation(workflow_name, workflow_run_id, collected_files):
        notification_calls.append((workflow_name, workflow_run_id, list(collected_files)))
        return {
            "message_ids": ["om_test"],
            "recipient_message_ids": {"ou_test": "om_test"},
            "errors": [],
        }

    monkeypatch.setattr(
        mcp_workflow,
        "_send_purchase_confirmation_notification",
        send_purchase_confirmation,
    )
    monkeypatch.setattr(mcp_workflow, "_send_collector_heartbeat", lambda *_args, **_kwargs: "")
    monkeypatch.setattr(mcp_workflow, "_start_progress_heartbeat_worker", lambda *_args: None)
    monkeypatch.setattr(mcp_workflow, "_close_workflow_browser", lambda *_args: [])
    monkeypatch.setattr(mcp_workflow, "_cleanup_workflow_artifacts", lambda *_args: [])
    monkeypatch.setattr(pg_writer, "upsert_run", lambda **_kwargs: None)

    result = mcp_workflow.run_mcp_workflow(
        "purchase-confirmation",
        "purchase-confirmation_20260814",
    )

    assert result["status"] == "completed"
    assert result["force_refresh"] is True
    assert result["collection_reused"] is False
    assert collection_calls == [
        "purchase-confirmation:purchase-confirmation_20260814"
    ]
    assert notification_calls == [
        (
            "purchase-confirmation",
            "purchase-confirmation_20260814",
            [str(collected_file)],
        )
    ]
    assert receipt_calls == [
        (
            "purchase-confirmation",
            "purchase-confirmation_20260814",
            ("ou_test",),
        )
    ]


@pytest.mark.parametrize(
    "script_name",
    ["collect_confirmation.ps1", "collect_replenishment.ps1"],
)
def test_direct_collection_wrappers_gate_before_browser_start(
    script_name: str,
) -> None:
    source = (SUPPLY_WRAPPERS / script_name).read_text(encoding="utf-8-sig")

    acceptance_position = source.index('if ($env:GYXX_WORKFLOW_ACCEPTANCE -eq "1")')
    gate_position = source.index("[SKIPPED_COOKIE]")
    browser_start_position = source.index("Start-Process -FilePath $chromePath")
    assert gate_position < browser_start_position
    assert "exit 75" in source[gate_position:browser_start_position]
    assert '$forceFreshBrowser = $false' in source[gate_position - 200:browser_start_position]
    assert "$env:GYXX_SUPPLY_ERP_PASSWORD" in source[
        acceptance_position:browser_start_position
    ]
    assert "starting the target-owned profile for credential fallback" in source


def test_direct_purchase_order_wrapper_blocks_before_task_and_browser_access() -> None:
    source = (SUPPLY_WRAPPERS / "collect_purchase_order_update.ps1").read_text(
        encoding="utf-8-sig"
    )

    gate_position = source.index("[BLOCKED_BUSINESS_MUTATION]")
    assert gate_position < source.index("$taskFile =")
    assert gate_position < source.index("Start-Process -FilePath $chromePath")
    assert "exit 77" in source[gate_position:source.index("$scriptDir =")]


@pytest.mark.parametrize(
    ("script_name", "expected_code", "expected_marker"),
    [
        ("collect_confirmation.ps1", COOKIE_SKIP_EXIT_CODE, "SKIPPED_COOKIE"),
        ("collect_replenishment.ps1", COOKIE_SKIP_EXIT_CODE, "SKIPPED_COOKIE"),
        (
            "collect_purchase_order_update.ps1",
            runner.BLOCKED_EXIT_CODE,
            "BLOCKED_BUSINESS_MUTATION",
        ),
    ],
)
def test_direct_wrapper_process_never_calls_start_process_when_gated(
    script_name: str,
    expected_code: int,
    expected_marker: str,
    tmp_path: Path,
) -> None:
    powershell = shutil.which("powershell.exe")
    if powershell is None:
        pytest.skip("Windows PowerShell is unavailable")

    target = SUPPLY_WRAPPERS / script_name
    escaped_target = str(target).replace("'", "''")
    command = "\n".join(
        [
            "function global:Get-NetTCPConnection {",
            "  param([int]$LocalPort, [string]$State)",
            "  return $null",
            "}",
            "function global:Start-Process { throw 'START_PROCESS_CALLED' }",
            f"& '{escaped_target}'",
            "$code = $LASTEXITCODE",
            "exit $code",
        ]
    )
    environment = os.environ.copy()
    environment.update(
        {
            "GYXX_WORKFLOW_ACCEPTANCE": "1",
            "GYXX_PROJECT_ROOT": str(PROJECT_ROOT),
            "GYXX_DATA_ROOT": str(tmp_path / "data"),
        }
    )

    completed = subprocess.run(
        [
            powershell,
            "-NoProfile",
            "-ExecutionPolicy",
            "Bypass",
            "-Command",
            command,
        ],
        cwd=PROJECT_ROOT,
        env=environment,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=30,
        check=False,
    )

    output = completed.stdout + completed.stderr
    assert completed.returncode == expected_code, output
    assert expected_marker in output
    assert "START_PROCESS_CALLED" not in output


def test_replenishment_alert_card_builds_deterministic_content() -> None:
    payload = {
        "alert_count": 1,
        "generated_at": "2026-08-18 07:00:45",
        "alerts": [
            {
                "sku": "10362011",
                "threshold": 300,
                "actual_stock": 184.0,
                "purchase_all": 0.0,
                "product_name": "L型数据线",
                "product_spec": "弯头磁吸线CTOC；活力橙色 100W",
                "sales_7": 28.0,
                "sales_15": 46.0,
                "sales_30": 92.0,
            }
        ],
    }

    card = mcp_workflow._replenishment_alert_card(payload)

    assert card["header"]["template"] == "red"
    assert card["header"]["title"]["content"] == "库存阈值预警"
    text = json.dumps(card, ensure_ascii=False)
    assert "10362011" in text
    assert "L型数据线" in text
    assert "库存缺口" in text
    assert "116" in text
    assert "92" in text


def test_replenishment_alert_notification_disabled_records_disabled_receipt(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    run_id = "run-disabled"
    collected = tmp_path / "threshold_alert_data.json"
    collected.write_text(
        json.dumps(
            {
                "alert_count": 1,
                "alerts": [
                    {
                        "sku": "10362011",
                        "threshold": 300,
                        "actual_stock": 184.0,
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(mcp_workflow, "STATE_ROOT", tmp_path / "state")
    snapshot = mcp_workflow._notification_snapshot_path(
        "replenishment-alert",
        run_id,
    )
    snapshot.parent.mkdir(parents=True)
    snapshot.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": "replenishment-alert",
                "run_id": run_id,
                "enabled": False,
                "recipients": [],
                "acceptance": False,
            }
        ),
        encoding="utf-8",
    )

    calls: list[tuple[tuple[object, ...], dict[str, object]]] = []

    def fake_mark_notification_sent(*args, **kwargs):
        calls.append((args, kwargs))

    monkeypatch.setattr(
        pg_writer,
        "mark_notification_sent",
        fake_mark_notification_sent,
    )

    result = mcp_workflow._send_replenishment_alert_notification(
        "replenishment-alert",
        run_id,
        [str(collected)],
    )

    assert result["notification_disabled"] is True
    assert len(calls) == 1
    _, kwargs = calls[0]
    assert kwargs["notification_disabled"] is True
    assert kwargs["message_ids"] == []
    assert kwargs["expected_recipients"] == []


def test_replenishment_alert_notification_sends_and_returns_receipt(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    run_id = "run-enabled"
    recipients = ["ou_recipient0001", "ou_recipient0002"]
    collected = tmp_path / "threshold_alert_data.json"
    collected.write_text(
        json.dumps(
            {
                "alert_count": 1,
                "alerts": [
                    {
                        "sku": "10362011",
                        "threshold": 300,
                        "actual_stock": 184.0,
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(mcp_workflow, "STATE_ROOT", tmp_path / "state")
    snapshot = mcp_workflow._notification_snapshot_path(
        "replenishment-alert",
        run_id,
    )
    snapshot.parent.mkdir(parents=True)
    snapshot.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "workflow": "replenishment-alert",
                "run_id": run_id,
                "enabled": True,
                "recipients": recipients,
                "acceptance": False,
            }
        ),
        encoding="utf-8",
    )

    captured: dict[str, object] = {}

    def fake_send_card_run(
        receivers,
        card,
        *,
        workflow,
        run_id,
        idempotency_key,
    ):
        captured.update(
            receivers=list(receivers),
            card=card,
            workflow=workflow,
            run_id=run_id,
            idempotency_key=idempotency_key,
        )
        return {
            "message_ids": ["om_1", "om_2"],
            "recipient_message_ids": {
                "ou_recipient0001": "om_1",
                "ou_recipient0002": "om_2",
            },
            "errors": [],
        }

    monkeypatch.setattr(send_card_notification, "run", fake_send_card_run)

    result = mcp_workflow._send_replenishment_alert_notification(
        "replenishment-alert",
        run_id,
        [str(collected)],
    )

    assert result["message_ids"] == ["om_1", "om_2"]
    assert captured["receivers"] == recipients
    assert captured["workflow"] == "replenishment-alert"
    assert captured["run_id"] == run_id
    assert captured["idempotency_key"] == f"replenishment-alert:{run_id}"
    assert captured["card"]["header"]["template"] == "red"


def test_replenishment_alert_workflow_uses_deterministic_notification(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
    monkeypatch.delenv("GYXX_FORCE_REFRESH", raising=False)
    run_id = "replenishment-alert_20260818"
    collected_file = tmp_path / "threshold_alert_data.json"
    collected_file.write_text(
        json.dumps(
            {
                "run_id": run_id,
                "alert_count": 1,
                "alerts": [
                    {
                        "sku": "10362011",
                        "threshold": 300,
                        "actual_stock": 184.0,
                    }
                ],
            }
        ),
        encoding="utf-8",
    )

    monkeypatch.setattr(mcp_workflow, "_acquire_workflow_lock", lambda *_args: (True, ""))
    monkeypatch.setattr(mcp_workflow, "_release_workflow_lock", lambda *_args: None)
    monkeypatch.setattr(mcp_workflow, "_acquire_collection_lock", lambda *_args: (True, ""))
    monkeypatch.setattr(mcp_workflow, "_release_collection_lock", lambda *_args: None)
    monkeypatch.setattr(
        mcp_workflow,
        "_reusable_collected_files",
        lambda *_args: [],
    )
    monkeypatch.setattr(mcp_workflow, "_clear_shared_outputs", lambda *_args: [])
    monkeypatch.setattr(
        mcp_workflow,
        "_run_local_collection_script",
        lambda *_args: (True, "fresh collection"),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_ensure_shared_outputs",
        lambda *_args: [str(collected_file)],
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_validate_run_manifest",
        lambda *_args: (True, ""),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_extract_pending_run_id",
        lambda *_args: run_id,
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_load_or_create_notification_snapshot",
        lambda *_args: (True, ("ou_test_recipient",), False),
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_notification_receipt_exists",
        lambda *_args: False,
    )

    deterministic_calls: list[tuple[str, str, list[str]]] = []

    def fake_send_notification(workflow_name, workflow_run_id, collected_files):
        deterministic_calls.append(
            (workflow_name, workflow_run_id, list(collected_files))
        )
        return {
            "message_ids": ["om_1"],
            "recipient_message_ids": {"ou_test_recipient": "om_1"},
            "errors": [],
        }

    monkeypatch.setattr(
        mcp_workflow,
        "_send_replenishment_alert_notification",
        fake_send_notification,
    )

    analyzer_calls: list[str] = []

    def fake_analyze(role: str, _prompt: str):
        analyzer_calls.append(role)
        return True, 1, "analyzer must not run for replenishment-alert"

    monkeypatch.setattr(mcp_workflow, "_call_once", fake_analyze)
    monkeypatch.setattr(
        mcp_workflow,
        "_send_collector_heartbeat",
        lambda *_args, **_kwargs: "",
    )
    monkeypatch.setattr(
        mcp_workflow,
        "_start_progress_heartbeat_worker",
        lambda *_args: None,
    )
    monkeypatch.setattr(mcp_workflow, "_close_workflow_browser", lambda *_args: [])
    monkeypatch.setattr(mcp_workflow, "_cleanup_workflow_artifacts", lambda *_args: [])
    monkeypatch.setattr(pg_writer, "upsert_run", lambda **_kwargs: None)

    result = mcp_workflow.run_mcp_workflow("replenishment-alert", run_id)

    assert result["status"] == "completed"
    assert result["current_phase"] == "finishing"
    assert result["notification_receipt"]["message_ids"] == ["om_1"]
    assert deterministic_calls == [
        ("replenishment-alert", run_id, [str(collected_file)])
    ]
    assert analyzer_calls == []


def test_purchase_confirmation_notification_uses_collector_csv_and_receipt(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    run_id = "purchase-confirmation_20260822"
    collected = tmp_path / "purchase-confirmation.csv"
    collected.write_text(
        "商品编码,清洗后的名称,清洗后的规格,可用数,7天销量,采购在途数,汇总原始数据,缺口,协议到货时间\n"
        "'10483001,宙斯ZAir,大号;曜夜黑,108,698.0,1756.0,宙斯Z Air,-590.0,2026-08-21 超时2\n",
        encoding="utf-8",
    )
    recipients = ("ou_test_recipient",)
    captured: dict[str, object] = {}

    monkeypatch.setattr(
        mcp_workflow,
        "_load_or_create_notification_snapshot",
        lambda *_args: (True, recipients, False),
    )
    from gyxx_flow.modules.supply_chain.orchestrator.scripts import (
        send_card_notification,
    )

    def fake_send_card_run(receivers, card, **kwargs):
        captured.update(receivers=receivers, card=card, **kwargs)
        return {
            "message_ids": ["om_confirmation"],
            "recipient_message_ids": {recipients[0]: "om_confirmation"},
            "errors": [],
        }

    monkeypatch.setattr(send_card_notification, "run", fake_send_card_run)

    receipt = mcp_workflow._send_purchase_confirmation_notification(
        "purchase-confirmation", run_id, [str(collected)]
    )

    assert receipt["message_ids"] == ["om_confirmation"]
    assert captured["receivers"] == list(recipients)
    assert captured["workflow"] == "purchase-confirmation"
    assert captured["run_id"] == run_id
    assert captured["idempotency_key"] == f"purchase-confirmation:{run_id}"
    card = captured["card"]
    assert card["header"]["title"]["content"] == "采购确认通知"
    body = "\n".join(item["text"]["content"] for item in card["elements"] if item["tag"] == "div")
    assert "10483001" in body
    assert "宙斯ZAir" in body
    assert "大号;曜夜黑" in body
    assert "108" in body
    assert "-590.0" in body
    assert "2026-08-21 超时2" in body


def test_replenishment_notification_uses_collector_payloads_and_receipt(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    run_id = "replenishment_20260822"
    pending = tmp_path / "pending_insert.json"
    notify = tmp_path / "notify_data.json"
    pending.write_text(
        json.dumps(
            {
                "records": [
                    {
                        "商品编号": "10483001",
                        "商品名称": "宙斯ZAir",
                        "规格名称（新）": "大号;曜夜黑",
                        "补货量": 300,
                        "实际可用数": 18,
                        "采购在途": 50,
                    },
                    {"商品编号": "10483002", "补货量": float("nan")},
                ]
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    notify.write_text(
        json.dumps({"pending_count": 1, "skipped_count": 2}, ensure_ascii=False),
        encoding="utf-8",
    )
    recipients = ("ou_test_recipient",)
    captured: dict[str, object] = {}
    monkeypatch.setattr(
        mcp_workflow,
        "_load_or_create_notification_snapshot",
        lambda *_args: (True, recipients, False),
    )
    from gyxx_flow.modules.supply_chain.orchestrator.scripts import (
        send_card_notification,
    )

    def fake_send_card_run(receivers, card, **kwargs):
        captured.update(receivers=receivers, card=card, **kwargs)
        return {
            "message_ids": ["om_replenishment"],
            "recipient_message_ids": {recipients[0]: "om_replenishment"},
            "errors": [],
        }

    monkeypatch.setattr(send_card_notification, "run", fake_send_card_run)

    receipt = mcp_workflow._send_replenishment_notification(
        "replenishment", run_id, [str(pending), str(notify)]
    )

    assert receipt["message_ids"] == ["om_replenishment"]
    assert captured["receivers"] == list(recipients)
    assert captured["workflow"] == "replenishment"
    assert captured["idempotency_key"] == f"replenishment:{run_id}"
    card = captured["card"]
    assert card["header"]["title"]["content"] == "补货建议通知"
    body = "\n".join(item["text"]["content"] for item in card["elements"] if item["tag"] == "div")
    assert "10483001" in body
    assert "宙斯ZAir" in body
    assert "大号;曜夜黑" in body
    assert "建议补货：300" in body