MemeMeow/tests/test_tasks.py

"""进程内长任务状态机测试。"""

from __future__ import annotations

import time
from threading import Event, Lock

import pytest

from backend.tasks import PersistentTaskService, TaskManager, TaskRecord


def wait_for_terminal(manager: TaskManager, task_id: str, timeout: float = 2.0):
    """轮询测试任务直到终态，超时即失败。"""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        record = manager.get(task_id)
        if record and record.status in {"succeeded", "failed"}:
            return record
        time.sleep(0.01)
    raise AssertionError("任务未在测试时限内完成")


def test_task_succeeds_and_reports_progress():
    """成功任务按 queued/running/succeeded 状态机结束。"""
    manager = TaskManager(max_workers=1)
    record = manager.submit("cache_generation", lambda progress: progress(0.5, "half"))
    completed = wait_for_terminal(manager, record.task_id)
    assert completed.status == "succeeded"
    assert completed.progress == 1.0
    assert completed.completed_at is not None
    manager.shutdown()


def test_duplicate_active_type_returns_same_task():
    """同类型未完成任务不会并发执行。"""
    release = Event()
    manager = TaskManager(max_workers=1)
    first = manager.submit("cache_generation", lambda progress: release.wait(1))
    second = manager.submit("cache_generation", lambda progress: None)
    assert second.task_id == first.task_id
    release.set()
    wait_for_terminal(manager, first.task_id)
    manager.shutdown()


def test_persistent_task_list_orders_by_created_at_and_keeps_cursor_stable(tmp_path):
    """任务更新时间变化不会改变创建顺序，cursor 仍按创建键连续分页。"""
    service = PersistentTaskService(tmp_path / "tasks")
    older = TaskRecord(
        task_id="task-old",
        task_type="cache_generation",
        created_at="2026-08-26T00:00:01Z",
        updated_at="2026-08-26T00:10:00Z",
    )
    newer = TaskRecord(
        task_id="task-new",
        task_type="cache_generation",
        created_at="2026-08-26T00:00:02Z",
        updated_at="2026-08-26T00:00:03Z",
    )
    tie_low = TaskRecord(
        task_id="task-a",
        task_type="cache_generation",
        created_at="2026-08-26T00:00:03Z",
        updated_at="2026-08-26T00:00:04Z",
    )
    tie_high = TaskRecord(
        task_id="task-z",
        task_type="cache_generation",
        created_at="2026-08-26T00:00:03Z",
        updated_at="2026-08-26T00:00:01Z",
    )
    with service._lock:
        service._records = {item.task_id: item for item in (older, newer, tie_low, tie_high)}

    first_page, cursor = service.list(limit=1)
    assert [item.task_id for item in first_page] == ["task-z"]
    second_page, next_cursor = service.list(cursor=cursor, limit=1)
    assert [item.task_id for item in second_page] == ["task-a"]
    assert next_cursor == "task-a"

    older.updated_at = "2026-08-26T23:00:00Z"
    full_page, _ = service.list(limit=10)
    assert [item.task_id for item in full_page] == ["task-z", "task-a", "task-new", "task-old"]


def test_failure_is_diagnostic_and_shutdown_marks_pending_failed():
    """异常和服务关闭都生成稳定失败信息。"""
    manager = TaskManager(max_workers=1)

    def fail(progress):
        raise RuntimeError("boom")

    failed = manager.submit("failure", fail)
    completed = wait_for_terminal(manager, failed.task_id)
    assert completed.status == "failed"
    assert completed.error["error"] == "task_failed"

    release = Event()
    pending = manager.submit("pending", lambda progress: release.wait(1))
    manager.shutdown()
    stopped = manager.get(pending.task_id)
    assert stopped.status == "failed"
    assert stopped.error["error"] == "task_not_recoverable"
    release.set()


@pytest.mark.parametrize("code", ["thumbnail_generation_failed", "thumbnail_output_invalid", "thumbnail_source_changed"])
def test_thumbnail_task_failure_keeps_stable_error_code(tmp_path, code):
    """缩略图处理器抛出的稳定错误码不会被本地任务服务归一为 task_failed。"""
    manager = PersistentTaskService(tmp_path / "tasks")

    def fail(_payload, _progress):
        """模拟缩略图处理器返回带诊断信息的领域错误。"""
        raise RuntimeError(f"{code}: diagnostic")

    manager.register("derived_thumbnail_generation", fail)
    manager.start()
    try:
        record = manager.submit("derived_thumbnail_generation", {"meme_id": code})
        completed = wait_for_terminal(manager, record.task_id)
        assert completed.status == "failed"
        assert completed.error["error"] == code
    finally:
        manager.shutdown()


def test_agent_lane_runs_different_images_in_parallel_and_keeps_cache_lane_available(tmp_path):
    """Agent lane 受并发上限控制，cache 任务仍可在 Agent 满载时启动。"""
    manager = PersistentTaskService(tmp_path / "tasks", max_workers=1, agent_concurrency=2)
    active = 0
    maximum = 0
    lock = Lock()
    release = Event()

    def context(payload, progress):
        nonlocal active, maximum
        with lock:
            active += 1
            maximum = max(maximum, active)
        release.wait(1)
        with lock:
            active -= 1

    cache_started = Event()

    def cache(payload, progress):
        cache_started.set()

    manager.register("meme_context_generation", context)
    manager.register("cache_generation", cache)
    manager.start()
    first = manager.submit("meme_context_generation", {"image_relative_path": "a.png"})
    second = manager.submit("meme_context_generation", {"image_relative_path": "b.png"})
    queued = manager.submit("meme_context_generation", {"image_relative_path": "c.png"})
    manager.submit("cache_generation", {})
    assert queued.message == "等待 Agent 运行槽位"
    assert cache_started.wait(1)
    release.set()
    assert wait_for_terminal(manager, first.task_id).status == "succeeded"
    assert wait_for_terminal(manager, second.task_id).status == "succeeded"
    assert wait_for_terminal(manager, queued.task_id).status == "succeeded"
    assert maximum == 2
    manager.shutdown()


def test_local_agent_lane_preserves_configured_large_capacity(tmp_path):
    """本地兼容任务服务保留大运行容量，不依赖旧背压默认值。"""
    manager = PersistentTaskService(tmp_path / "tasks", agent_concurrency=500, agent_backpressure=80)
    try:
        assert manager.agent_concurrency == 500
    finally:
        manager.shutdown()


def test_task_record_public_summary_preserves_large_agent_concurrency() -> None:
    """任务公开摘要和磁盘恢复不再把并发值截断到旧的 128。"""
    record = TaskRecord(task_id="task", task_type="meme_context_generation", agent_concurrency=256)
    public = record.as_dict()
    assert public["agent_concurrency"] == 256
    restored = TaskRecord.from_dict(public)
    assert restored.agent_concurrency == 256


def test_local_json_service_explicitly_reports_single_scope_compatibility(tmp_path):
    """本地 JSON 兼容服务不伪装提供跨进程公平调度。"""
    manager = PersistentTaskService(tmp_path / "tasks")
    assert manager.supports_cross_scope_fairness is False
    assert manager.fairness_mode == "single_scope_process_compat"
    manager.shutdown()


def test_batch_finalizer_submits_one_deduplicated_cache_task(tmp_path):
    """批次最后一项终态后只触发一次缓存回调。"""
    manager = PersistentTaskService(tmp_path / "tasks", agent_concurrency=2)
    finalized = []
    manager.set_batch_finalizer(finalized.append)
    manager.register("meme_context_generation", lambda payload, progress: None)
    manager.start()
    first = manager.submit("meme_context_generation", {"image_relative_path": "a.png", "batch_id": "batch"})
    second = manager.submit("meme_context_generation", {"image_relative_path": "b.png", "batch_id": "batch"})
    assert wait_for_terminal(manager, first.task_id).status == "succeeded"
    assert wait_for_terminal(manager, second.task_id).status == "succeeded"
    for _ in range(50):
        if finalized:
            break
        time.sleep(0.01)
    assert finalized == ["batch"]
    manager.shutdown()


def test_context_dedupe_uses_image_identity_not_transient_batch_or_auto_name(tmp_path):
    """同一图片即使来自不同批次或自动命名选项也只保留一个活动任务。"""
    manager = PersistentTaskService(tmp_path / "tasks")
    release = Event()
    manager.register("meme_context_generation", lambda payload, progress: release.wait(1))
    manager.start()
    first = manager.submit("meme_context_generation", {"image_relative_path": "a.png", "image_sha256": "a" * 64, "auto_name": False, "batch_id": "one"})
    second = manager.submit("meme_context_generation", {"image_relative_path": "a.png", "image_sha256": "a" * 64, "auto_name": True, "batch_id": "two"})
    assert second.task_id == first.task_id
    release.set()
    wait_for_terminal(manager, first.task_id)
    manager.shutdown()


def test_reused_context_task_finalizes_every_associated_batch(tmp_path):
    """活动任务跨批次复用时，每个批次仍只触发一次缓存合并。"""
    manager = PersistentTaskService(tmp_path / "tasks")
    finalized = []
    release = Event()
    manager.set_batch_finalizer(finalized.append)
    manager.register("meme_context_generation", lambda payload, progress: release.wait(1))
    manager.start()
    first = manager.submit("meme_context_generation", {"image_relative_path": "a.png", "image_sha256": "a" * 64, "batch_id": "one"})
    second = manager.submit("meme_context_generation", {"image_relative_path": "a.png", "image_sha256": "a" * 64, "batch_id": "two"})
    assert second.task_id == first.task_id
    release.set()
    assert wait_for_terminal(manager, first.task_id).status == "succeeded"
    for _ in range(50):
        if sorted(finalized) == ["one", "two"]:
            break
        time.sleep(0.01)
    assert sorted(finalized) == ["one", "two"]
    manager.shutdown()


def test_standalone_image_stage_dedupe_ignores_submission_nonce(tmp_path):
    """独立图片阶段的并发 nonce 不应绕过去重并重复执行同一输入。"""
    manager = PersistentTaskService(tmp_path / "tasks")
    release = Event()
    manager.register("image_auto_rename", lambda payload, progress: release.wait(1))
    manager.start()
    payload = {
        "submission_mode": "standalone",
        "stage": "auto_rename",
        "meme_id": "meme-1",
        "image_sha256": "a" * 64,
        "expected_storage_key": "source.png",
        "title_fingerprint": "b" * 64,
    }
    first = manager.submit("image_auto_rename", {**payload, "standalone_submission_nonce": "one"})
    second = manager.submit("image_auto_rename", {**payload, "standalone_submission_nonce": "two"})
    assert second.task_id == first.task_id
    release.set()
    assert wait_for_terminal(manager, first.task_id).status == "succeeded"
    manager.shutdown()