temp/features/steps/async_worker_coverage_boost_steps.py

"""Step definitions for async_worker_coverage_boost.feature.

These steps target specific uncovered lines in async_worker.py:
- Lines 515-516: cancel_job InvalidJobTransitionError on queued job
- Lines 543-544: detect_stuck_jobs InvalidJobTransitionError
- Line 604: _poll_loop shutdown break during dispatch
- Lines 606-607: _poll_loop exception handler
- Lines 614-615: _dispatch_job fallback with no thread pool
- Line 618: _dispatch_job duplicate dispatch guard
- Lines 632-633: _signal_handler direct invocation
- Lines 642, 644: _install_signal_handlers from non-main thread
- Lines 653-654: _restore_signal_handlers from non-main thread
"""

from __future__ import annotations

import signal
import threading
from concurrent.futures import Future, ThreadPoolExecutor
from datetime import UTC, datetime, timedelta
from unittest.mock import MagicMock, patch

from behave import given, then, when
from behave.runner import Context
from ulid import ULID

from cleveragents.application.services.async_worker import (
    AsyncWorker,
    AsyncWorkerConfig,
    InMemoryJobStore,
)
from cleveragents.domain.models.core.async_job import (
    AsyncJob,
    AsyncJobStatus,
    InvalidJobTransitionError,
)

PLAN_ID = "01HXYZ01HXYZ01HXYZ01HXYZ01"


def _make_job_id() -> str:
    return str(ULID())


# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------


@given("a coverage-boost async worker is initialised")
def step_cb_worker_init(context: Context) -> None:
    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=4,
        poll_interval=0.05,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.cb_job_store = InMemoryJobStore()
    context.cb_worker = AsyncWorker(config, context.cb_job_store)
    context.cb_job = None
    context.cb_error = None
    context.cb_jobs = []


@given("a coverage-boost queued job exists")
def step_cb_queued_job(context: Context) -> None:
    job = AsyncJob(
        job_id=_make_job_id(),
        plan_id=PLAN_ID,
        phase="execute",
    )
    context.cb_job_store.add(job)
    context.cb_job = job


@given("the queued job mark_cancelled is patched to raise InvalidJobTransitionError")
def step_cb_patch_mark_cancelled(context: Context) -> None:
    """Prepare a class-level patch for mark_cancelled to simulate a race
    condition where the job transitions between the is_terminal check and
    the mark_cancelled call in cancel_job (lines 515-516)."""
    context.cb_patch_cancel = patch.object(
        AsyncJob,
        "mark_cancelled",
        side_effect=InvalidJobTransitionError(
            AsyncJobStatus.CANCELLED, AsyncJobStatus.CANCELLED
        ),
    )
    context.cb_patch_cancel.start()


@given("a coverage-boost async worker with job_timeout {timeout:d}")
def step_cb_worker_timeout(context: Context, timeout: int) -> None:
    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.05,
        job_timeout=timeout,
        job_ttl=86400,
    )
    context.cb_job_store = InMemoryJobStore()
    context.cb_worker = AsyncWorker(config, context.cb_job_store)
    context.cb_job = None
    context.cb_jobs = []


@given("a coverage-boost running job with expired heartbeat exists")
def step_cb_running_job_expired(context: Context) -> None:
    job = AsyncJob(
        job_id=_make_job_id(),
        plan_id=PLAN_ID,
        phase="execute",
    )
    context.cb_job_store.add(job)
    job.mark_running("worker-stuck")
    # Backdate heartbeat to exceed the timeout
    job.last_heartbeat = datetime.now(UTC) - timedelta(seconds=10)
    job.started_at = datetime.now(UTC) - timedelta(seconds=10)
    context.cb_job_store.update(job)
    context.cb_job = job


@given("the stuck job mark_failed is patched to raise InvalidJobTransitionError")
def step_cb_patch_mark_failed(context: Context) -> None:
    """Prepare a class-level patch for mark_failed to simulate a race
    where the stuck job transitions to terminal between detection and
    the mark_failed call (lines 543-544)."""
    context.cb_patch_failed = patch.object(
        AsyncJob,
        "mark_failed",
        side_effect=InvalidJobTransitionError(
            AsyncJobStatus.CANCELLED, AsyncJobStatus.FAILED
        ),
    )
    context.cb_patch_failed.start()


@given("{count:d} coverage-boost queued jobs exist")
def step_cb_n_queued_jobs(context: Context, count: int) -> None:
    context.cb_jobs = []
    for _ in range(count):
        job = AsyncJob(
            job_id=_make_job_id(),
            plan_id=PLAN_ID,
            phase="execute",
        )
        context.cb_job_store.add(job)
        context.cb_jobs.append(job)


@given("the shutdown event is pre-set before dispatch")
def step_cb_pre_set_shutdown(context: Context) -> None:
    """Flag that shutdown should be set during the poll loop iteration."""
    context.cb_pre_set_shutdown = True


@given("the job store list_by_status is patched to raise RuntimeError")
def step_cb_patch_list_by_status(context: Context) -> None:
    """Flag that list_by_status should be patched to raise."""
    context.cb_patch_list_by_status = True


@given("the worker thread pool is None")
def step_cb_thread_pool_none(context: Context) -> None:
    """Ensure the thread pool is None (worker not started)."""
    context.cb_worker._thread_pool = None


@given("a coverage-boost async worker is initialised with a started pool")
def step_cb_worker_with_pool(context: Context) -> None:
    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=4,
        poll_interval=0.05,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.cb_job_store = InMemoryJobStore()
    context.cb_worker = AsyncWorker(config, context.cb_job_store)
    context.cb_job = None
    context.cb_jobs = []
    # Create a thread pool without starting the full worker
    context.cb_worker._thread_pool = ThreadPoolExecutor(
        max_workers=4, thread_name_prefix="test-pool"
    )


@given("the job is already registered in the futures dict")
def step_cb_register_future(context: Context) -> None:
    """Pre-register a mock future for the job in _futures."""
    mock_future = MagicMock(spec=Future)
    mock_future.done.return_value = False
    with context.cb_worker._futures_lock:
        context.cb_worker._futures[context.cb_job.job_id] = mock_future


@given("the worker has original signal handler references set")
def step_cb_set_original_handlers(context: Context) -> None:
    """Set fake original signal handler references so _restore_signal_handlers
    tries to call signal.signal()."""
    context.cb_worker._original_sigint = signal.SIG_DFL
    context.cb_worker._original_sigterm = signal.SIG_DFL


# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------


@when("I call cancel_job on the coverage-boost worker")
def step_cb_call_cancel_job(context: Context) -> None:
    try:
        context.cb_cancel_result = context.cb_worker.cancel_job(context.cb_job.job_id)
    finally:
        # Stop the class-level patch if active
        if hasattr(context, "cb_patch_cancel"):
            context.cb_patch_cancel.stop()


@when("the coverage-boost worker detects stuck jobs")
def step_cb_detect_stuck(context: Context) -> None:
    try:
        context.cb_stuck_jobs = context.cb_worker.detect_stuck_jobs()
    finally:
        # Stop the class-level patch if active
        if hasattr(context, "cb_patch_failed"):
            context.cb_patch_failed.stop()


@when("I run one poll loop iteration")
def step_cb_run_poll_iteration(context: Context) -> None:
    """Run the actual _poll_loop method in a background thread.

    A counting wrapper around _dispatch_job signals shutdown after
    the first dispatch, causing the inner for-loop to break at
    line 604 and the outer while-loop to exit.
    """
    worker = context.cb_worker
    dispatched_count_holder = [0]
    original_dispatch = worker._dispatch_job

    def counting_dispatch(job: AsyncJob) -> None:
        dispatched_count_holder[0] += 1
        # After first dispatch, signal shutdown to trigger line 604 break
        if getattr(context, "cb_pre_set_shutdown", False):
            worker._shutdown_event.set()
        original_dispatch(job)

    worker._dispatch_job = counting_dispatch  # type: ignore[assignment]

    # Run the actual _poll_loop in a thread; it will exit when
    # _shutdown_event is set.
    t = threading.Thread(target=worker._poll_loop, daemon=True)
    t.start()
    t.join(timeout=5.0)

    context.cb_dispatched_count = dispatched_count_holder[0]
    worker._dispatch_job = original_dispatch  # type: ignore[assignment]


@when("I run one poll loop iteration catching the exception")
def step_cb_run_poll_with_exception(context: Context) -> None:
    """Run the actual _poll_loop with a patched list_by_status that raises
    on the first call, exercising the exception handler at lines 606-607.

    On the second call the patch signals shutdown so the loop exits.
    """
    worker = context.cb_worker
    context.cb_poll_exception_handled = False

    original_list = context.cb_job_store.list_by_status
    call_count = [0]

    if getattr(context, "cb_patch_list_by_status", False):

        def raising_then_shutdown(status: AsyncJobStatus) -> list:
            call_count[0] += 1
            if call_count[0] <= 1:
                raise RuntimeError("Simulated poll loop error")
            # Second call: stop the loop
            worker._shutdown_event.set()
            return original_list(status)

        context.cb_job_store.list_by_status = raising_then_shutdown  # type: ignore[assignment]

    # Run the actual _poll_loop in a thread
    t = threading.Thread(target=worker._poll_loop, daemon=True)
    t.start()
    t.join(timeout=5.0)

    context.cb_poll_exception_handled = call_count[0] >= 1

    # Restore
    if getattr(context, "cb_patch_list_by_status", False):
        context.cb_job_store.list_by_status = original_list  # type: ignore[assignment]


@when("I dispatch the job directly via _dispatch_job")
def step_cb_dispatch_directly(context: Context) -> None:
    context.cb_worker._dispatch_job(context.cb_job)


@when("I invoke the signal handler directly with SIGINT")
def step_cb_invoke_signal_handler(context: Context) -> None:
    context.cb_worker._signal_handler(signal.SIGINT, None)


@when("I install signal handlers from a background thread")
def step_cb_install_from_thread(context: Context) -> None:
    """Call _install_signal_handlers from a non-main thread, which
    triggers ValueError caught at lines 642/644."""
    context.cb_thread_error = None

    def install_in_thread() -> None:
        try:
            context.cb_worker._install_signal_handlers()
        except Exception as exc:
            context.cb_thread_error = exc

    t = threading.Thread(target=install_in_thread)
    t.start()
    t.join(timeout=5.0)


@when("I restore signal handlers from a background thread")
def step_cb_restore_from_thread(context: Context) -> None:
    """Call _restore_signal_handlers from a non-main thread, which
    triggers ValueError caught at lines 653/654."""
    context.cb_thread_error = None

    def restore_in_thread() -> None:
        try:
            context.cb_worker._restore_signal_handlers()
        except Exception as exc:
            context.cb_thread_error = exc

    t = threading.Thread(target=restore_in_thread)
    t.start()
    t.join(timeout=5.0)


# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------


@then("the cancel_job result should be False")
def step_cb_check_cancel_false(context: Context) -> None:
    assert context.cb_cancel_result is False, (
        f"Expected cancel_job to return False, got {context.cb_cancel_result}"
    )


@then("the stuck jobs list should be empty")
def step_cb_stuck_empty(context: Context) -> None:
    assert len(context.cb_stuck_jobs) == 0, (
        f"Expected 0 stuck jobs, got {len(context.cb_stuck_jobs)}"
    )


@then("not all 5 queued jobs should have been dispatched")
def step_cb_not_all_dispatched(context: Context) -> None:
    # The shutdown break should prevent dispatching all 5 jobs.
    # At least 1 should be dispatched (triggers shutdown), but not all 5.
    assert context.cb_dispatched_count < 5, (
        f"Expected fewer than 5 dispatched, got {context.cb_dispatched_count}"
    )
    assert context.cb_dispatched_count >= 1, (
        f"Expected at least 1 dispatched, got {context.cb_dispatched_count}"
    )


@then("the poll loop should have handled the exception gracefully")
def step_cb_poll_exception_handled(context: Context) -> None:
    assert context.cb_poll_exception_handled is True, (
        "Expected the poll loop exception handler to fire"
    )


@then("the coverage-boost job should be in a terminal state")
def step_cb_job_terminal(context: Context) -> None:
    assert context.cb_job.is_terminal, (
        f"Expected terminal state, got {context.cb_job.status.value}"
    )


@then("the futures dict should still contain exactly one entry for the job")
def step_cb_futures_one_entry(context: Context) -> None:
    with context.cb_worker._futures_lock:
        assert context.cb_job.job_id in context.cb_worker._futures, (
            "Expected job_id in futures dict"
        )
        # The future should be the original mock, not a new one
        assert isinstance(
            context.cb_worker._futures[context.cb_job.job_id], MagicMock
        ), "Expected the original mock future, not a replacement"
    # Clean up the thread pool
    if context.cb_worker._thread_pool is not None:
        context.cb_worker._thread_pool.shutdown(wait=False)
        context.cb_worker._thread_pool = None


@then("the worker shutdown event should be set")
def step_cb_shutdown_set(context: Context) -> None:
    assert context.cb_worker._shutdown_event.is_set(), (
        "Expected shutdown event to be set after signal handler invocation"
    )


@then("no exception should be raised during signal installation")
def step_cb_no_install_exception(context: Context) -> None:
    assert context.cb_thread_error is None, (
        f"Expected no exception, got {context.cb_thread_error}"
    )


@then("no exception should be raised during signal restoration")
def step_cb_no_restore_exception(context: Context) -> None:
    assert context.cb_thread_error is None, (
        f"Expected no exception, got {context.cb_thread_error}"
    )