"""进程执行器的端到端接线：``TaskRun`` / ``Controller`` / 存储。

单元测试盯的是句柄本身（暂停、强杀、心跳、崩溃）。这里盯的是**接进主线之后**的
三件承诺：

1. 快照里真的记下了子进程的 pid（排障时 `ps` 对得上号的就是它）；
2. 不响应停止的任务会被**超时机制真的杀掉**，而不是留一个卡死的线程（线程执行器
   只能把它记成 `LOST`，这是进程执行器存在的理由）；
3. 这套东西写进存储之后仍然对得上（pid 是 store 契约里的字段，不是内存里的临时值）。
"""

from __future__ import annotations

import multiprocessing
import os
import time
from pathlib import Path
from typing import Any

from chongming_task import (
    Controller,
    FailureKind,
    FixedBackoff,
    ProcessExecutor,
    RetryPolicy,
    TaskStatus,
    Timeout,
    run_task,
)
from chongming_task.stores.sqlite import SqliteTaskStore

_WINDOW = 20.0


def _add(ctx: Any, left: int, right: int) -> int:
    """两数相加。"""
    del ctx
    return left + right


def _boom(ctx: Any) -> None:
    """总是失败：用来让任务反复重试。"""
    del ctx
    raise RuntimeError("总是失败")


def _deaf_loop(ctx: Any) -> str:
    """**不**调用 checkpoint：停不掉，只能强杀。"""
    del ctx
    end = time.monotonic() + 120
    while time.monotonic() < end:
        pass
    return "never"


def test_a_task_run_records_the_child_pid_in_its_snapshot() -> None:
    outcome = run_task(
        _add,
        2,
        40,
        task_id="proc-1",
        executor=ProcessExecutor(),
        timeout=Timeout(per_attempt=_WINDOW),
    )

    assert outcome.ok, outcome.failure
    assert outcome.value == 42
    snapshot = outcome.snapshot
    assert snapshot.executor == "process"
    assert snapshot.pid is not None
    assert snapshot.pid != os.getpid(), "任务跑在另一个进程里"


def test_an_idle_task_is_actually_killed_by_the_timeout() -> None:
    """线程执行器下这条只能"放弃等待"并落 `LOST`（线程杀不掉）；进程执行器**真的**把它停掉。

    超时仍然如实记成超时（不是"已停止"）——但载体确实没了，资源也确实释放了。
    """
    started = time.monotonic()
    outcome = run_task(
        _deaf_loop,
        task_id="proc-deaf",
        executor=ProcessExecutor(),
        timeout=Timeout(idle=1),
        retry=RetryPolicy(max_attempts=1),
        graceful_stop=0.5,
    )
    elapsed = time.monotonic() - started

    assert outcome.status is TaskStatus.FAILED
    assert outcome.failure is not None
    assert outcome.failure.kind is FailureKind.TIMEOUT_IDLE
    assert outcome.snapshot.pid is not None

    # 它自己那个循环要跑两分钟，这里几秒就回来了——只有"真的被停掉"解释得通。
    assert elapsed < _WINDOW
    # 载体不在了：那个 pid 已经不再是本进程活着的子进程。
    alive = {child.pid for child in multiprocessing.active_children()}
    assert outcome.snapshot.pid not in alive, "子进程必须真的被终止并回收"


def test_a_controller_can_run_tasks_in_child_processes(tmp_path: Path) -> None:
    store = SqliteTaskStore(tmp_path / "proc.db")
    controller = Controller(store=store, executor=ProcessExecutor(), owner="node-a")
    try:
        controller.register("add", _add)
        controller.submit("add", 20, 22, task_id="t1")
        outcome = controller.wait("t1", timeout=_WINDOW)

        assert outcome.ok and outcome.value == 42
        assert outcome.snapshot.executor == "process"

        # 落库之后仍然对得上：pid 是 store 契约里的字段，不是内存里的临时值。
        stored = store.get("t1")
        assert stored is not None
        assert stored.pid == outcome.snapshot.pid
        assert stored.executor == "process"
    finally:
        controller.close(timeout=_WINDOW, close_store=True)


def test_a_target_that_cannot_cross_the_boundary_fails_fast() -> None:
    """不可 pickle 的目标：任务落 ``failed`` 并带上诊断，而不是让调用方干等。

    （直接调 ``ProcessExecutor.launch()`` 会当场抛 ``ConfigurationError``；这条测的是
    走 ``run_task`` 那条路——异常在那里必须变成任务的结论。）
    """

    def local_closure(ctx: Any) -> str:  # 局部函数：子进程里找不到它
        del ctx
        return "nope"

    started = time.monotonic()
    outcome = run_task(local_closure, task_id="bad-1", executor=ProcessExecutor())
    elapsed = time.monotonic() - started

    assert outcome.status is TaskStatus.FAILED
    assert outcome.failure is not None
    assert "可 pickle" in outcome.failure.message
    assert outcome.failure.retryable is False
    assert elapsed < _WINDOW, "它必须立刻有结论，而不是等到天荒地老"


def test_a_retry_does_not_carry_the_previous_attempts_pid() -> None:
    """重试之间采一次快照：上一次的进程号必须已经清掉。

    留着它的危害很具体：运维看到一个 pid 就去 `ps`，而对上的可能是一个早在别处跑着的
    无关进程——比"没有 pid"更容易误导。
    """
    controller = Controller(executor=ProcessExecutor())
    controller.register("boom", _boom)
    try:
        controller.submit(
            "boom",
            task_id="t-retry",
            retry=RetryPolicy(max_attempts=3, backoff=FixedBackoff(2.0)),
        )
        deadline = time.monotonic() + _WINDOW
        while time.monotonic() < deadline:
            snapshot = controller.get("t-retry")
            if snapshot is not None and snapshot.status is TaskStatus.RETRY_WAIT:
                break
            time.sleep(0.02)
        else:  # pragma: no cover - 没等到就说明退避策略没生效
            raise AssertionError("任务没有进入 retry_wait")

        assert snapshot is not None
        assert snapshot.pid is None, "上一次尝试的进程号必须清掉"
        assert snapshot.executor == "process", "执行器是这一轮的配置，不随尝试变化"
    finally:
        controller.close(timeout=_WINDOW)