cleveragents-core/features/steps/async_execution_steps.py

"""Step definitions for async execution tests."""

from __future__ import annotations

import json
import threading
import time
from datetime import UTC, datetime, timedelta
from unittest.mock import MagicMock

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

from cleveragents.application.services.async_worker import (
    AsyncWorker,
    AsyncWorkerConfig,
    CancellationToken,
    InMemoryJobStore,
    WorkerHealthReport,
)
from cleveragents.application.services.plan_lifecycle_service import (
    PlanLifecycleService,
)
from cleveragents.domain.models.core.async_job import (
    AsyncJob,
    AsyncJobStatus,
    InvalidJobTransitionError,
    can_transition_job,
    deserialize_job_payload,
    serialize_job_payload,
)


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


PLAN_ID = "01HXYZ01HXYZ01HXYZ01HXYZ01"


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


use_step_matcher("parse")


@given("an async job store is initialised")
def step_async_job_store_initialised(context: Context) -> None:
    context.job_store = InMemoryJobStore()
    context.async_job = None
    context.error = None


@given('I create a new async job for plan "{plan_id}" phase "{phase}"')
def step_create_async_job(context: Context, plan_id: str, phase: str) -> None:
    job = AsyncJob(
        job_id=_make_job_id(),
        plan_id=plan_id,
        phase=phase,
    )
    context.job_store.add(job)
    context.async_job = job


@given("I create a running async job")
def step_create_running_async_job(context: Context) -> None:
    job = AsyncJob(
        job_id=_make_job_id(),
        plan_id=PLAN_ID,
        phase="execute",
    )
    context.job_store.add(job)
    job.mark_running("worker-test")
    context.job_store.update(job)
    context.async_job = job


@given("I create {count:d} queued async jobs")
def step_create_n_queued_jobs(context: Context, count: int) -> None:
    context.created_jobs = []
    for _ in range(count):
        job = AsyncJob(
            job_id=_make_job_id(),
            plan_id=PLAN_ID,
            phase="execute",
        )
        context.job_store.add(job)
        context.created_jobs.append(job)


@given("an async worker is initialised with max_workers {max_workers:d}")
def step_async_worker_initialised(context: Context, max_workers: int) -> None:
    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=max_workers,
        poll_interval=0.1,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store)
    context.async_job = None
    context.error = None


@given("an async worker is initialised with a failing executor")
def step_async_worker_failing_executor(context: Context) -> None:
    def failing_executor(job: AsyncJob, token: CancellationToken) -> None:
        raise RuntimeError("Execution failed")

    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.1,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store, failing_executor)
    context.async_job = None
    context.error = None


@given("an async worker is initialised with job_timeout {timeout:d}")
def step_async_worker_timeout(context: Context, timeout: int) -> None:
    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.1,
        job_timeout=timeout,
        job_ttl=86400,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store)
    context.async_job = None


@given("an async worker is initialised with job_ttl {ttl:d}")
def step_async_worker_ttl(context: Context, ttl: int) -> None:
    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.1,
        job_timeout=3600,
        job_ttl=ttl,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store)
    context.async_job = None


@given("a queued async job exists")
def step_queued_job_exists(context: Context) -> None:
    job = AsyncJob(
        job_id=_make_job_id(),
        plan_id=PLAN_ID,
        phase="execute",
    )
    context.job_store.add(job)
    context.async_job = job


@given("a running async job with expired heartbeat exists")
def step_running_job_expired_heartbeat(context: Context) -> None:
    job = AsyncJob(
        job_id=_make_job_id(),
        plan_id=PLAN_ID,
        phase="execute",
    )
    context.job_store.add(job)
    job.mark_running("worker-stuck")
    # Backdate heartbeat
    job.last_heartbeat = datetime.now(UTC) - timedelta(seconds=10)
    job.started_at = datetime.now(UTC) - timedelta(seconds=10)
    context.job_store.update(job)
    context.async_job = job


@given("a completed async job older than TTL exists")
def step_completed_job_older_than_ttl(context: Context) -> None:
    job = AsyncJob(
        job_id=_make_job_id(),
        plan_id=PLAN_ID,
        phase="execute",
    )
    context.job_store.add(job)
    job.mark_running("worker-old")
    job.mark_succeeded()
    # Backdate finished_at
    job.finished_at = datetime.now(UTC) - timedelta(seconds=10)
    context.job_store.update(job)
    context.async_job = job


@given("a cancellation token is created")
def step_cancellation_token_created(context: Context) -> None:
    context.token = CancellationToken()


@given("I cancel the async token")
def step_cancel_token(context: Context) -> None:
    context.token.cancel()


@given('a worker health report for "{worker_id}"')
def step_health_report_created(context: Context, worker_id: str) -> None:
    context.health_report = WorkerHealthReport(worker_id)


@given("I mark the async job as succeeded")
def step_given_mark_succeeded(context: Context) -> None:
    context.async_job.mark_succeeded()
    context.job_store.update(context.async_job)


@given("I mark the async job as failed")
def step_given_mark_failed(context: Context) -> None:
    context.async_job.mark_failed()
    context.job_store.update(context.async_job)


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


@when('I mark the async job as running with worker "{worker_id}"')
def step_mark_running(context: Context, worker_id: str) -> None:
    context.async_job.mark_running(worker_id)
    context.job_store.update(context.async_job)


@when("I mark the async job as succeeded")
def step_mark_succeeded(context: Context) -> None:
    context.async_job.mark_succeeded()
    context.job_store.update(context.async_job)


@when("I mark the async job as failed")
def step_mark_failed(context: Context) -> None:
    context.async_job.mark_failed()
    context.job_store.update(context.async_job)


@when("I mark the async job as cancelled")
def step_mark_cancelled(context: Context) -> None:
    context.async_job.mark_cancelled()
    context.job_store.update(context.async_job)


@when("I try to mark the async job as succeeded")
def step_try_mark_succeeded(context: Context) -> None:
    try:
        context.async_job.mark_succeeded()
        context.error = None
    except InvalidJobTransitionError as exc:
        context.error = exc


@when('I try to mark the async job as running with worker "{worker_id}"')
def step_try_mark_running(context: Context, worker_id: str) -> None:
    try:
        context.async_job.mark_running(worker_id)
        context.error = None
    except (InvalidJobTransitionError, ValueError) as exc:
        context.error = exc


@when("I try to mark the async job as running with an empty worker_id")
def step_try_mark_running_empty(context: Context) -> None:
    try:
        context.async_job.mark_running("")
        context.error = None
    except (InvalidJobTransitionError, ValueError) as exc:
        context.error = exc


@when("the worker picks up and executes the job")
def step_worker_pickup_execute(context: Context) -> None:
    context.worker.pickup_and_execute(context.async_job)


@when("I cancel the async job")
def step_cancel_job(context: Context) -> None:
    context.worker.cancel_job(context.async_job.job_id)


@when("the worker detects stuck jobs")
def step_detect_stuck_jobs(context: Context) -> None:
    context.stuck_jobs = context.worker.detect_stuck_jobs()


@when("the worker runs job cleanup")
def step_run_cleanup(context: Context) -> None:
    context.cleaned_count = context.worker.cleanup_completed_jobs()


@when("the worker is started")
def step_worker_started(context: Context) -> None:
    context.worker.start()


@when("a shutdown is requested")
def step_shutdown_requested(context: Context) -> None:
    context.worker.request_shutdown()
    # Wait for the poll thread to observe the shutdown event rather
    # than relying on an arbitrary sleep that can be flaky on slow CI.
    poll_thread = context.worker._poll_thread
    if poll_thread is not None:
        poll_thread.join(timeout=2.0)
    context.worker.stop()


@when('I serialize a job payload for plan "{plan_id}" phase "{phase}"')
def step_serialize_payload(context: Context, plan_id: str, phase: str) -> None:
    context.payload_json = serialize_job_payload(plan_id, phase)
    context.original_plan_id = plan_id
    context.original_phase = phase


@when("I serialize a job payload with extra data")
def step_serialize_payload_extra(context: Context) -> None:
    context.payload_json = serialize_job_payload(
        PLAN_ID, "execute", extra={"key": "value"}
    )


@when("I try to deserialize invalid JSON")
def step_try_deserialize_invalid(context: Context) -> None:
    try:
        deserialize_job_payload("not valid json {{{")
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I request the worker health report")
def step_request_health_report(context: Context) -> None:
    context.health_data = context.worker.get_health_report()


@when("I try to create AsyncWorkerConfig with max_workers {max_workers:d}")
def step_try_create_config_max_workers(context: Context, max_workers: int) -> None:
    try:
        AsyncWorkerConfig(max_workers=max_workers)
        context.config_error = None
    except ValueError as exc:
        context.config_error = exc


@when("I try to create AsyncWorkerConfig with poll_interval {interval:d}")
def step_try_create_config_poll_interval(context: Context, interval: int) -> None:
    try:
        AsyncWorkerConfig(poll_interval=float(interval))
        context.config_error = None
    except ValueError as exc:
        context.config_error = exc


@when("I try to add the same job again")
def step_try_add_duplicate(context: Context) -> None:
    try:
        context.job_store.add(context.async_job)
        context.error = None
    except ValueError as exc:
        context.error = exc


@when('I list jobs by status "{status}"')
def step_list_by_status(context: Context, status: str) -> None:
    context.job_list = context.job_store.list_by_status(AsyncJobStatus(status))


@when("I record a heartbeat on the async job")
def step_record_heartbeat(context: Context) -> None:
    before = context.async_job.last_heartbeat
    # Busy-wait until the clock advances past the previous heartbeat
    # to avoid flaky failures on fast systems where 10ms may not be
    # enough to produce a distinct datetime.now(UTC) value.
    while datetime.now(UTC) <= before:
        time.sleep(0.001)
    context.async_job.record_heartbeat()
    context.heartbeat_before = before


@when("I try to record a heartbeat on the async job")
def step_try_record_heartbeat(context: Context) -> None:
    try:
        context.async_job.record_heartbeat()
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I get the async job CLI dict")
def step_get_cli_dict(context: Context) -> None:
    context.cli_dict = context.async_job.as_cli_dict()


@when('I try to create an async job with phase "{phase}"')
def step_try_create_invalid_phase(context: Context, phase: str) -> None:
    try:
        AsyncJob(
            job_id=_make_job_id(),
            plan_id=PLAN_ID,
            phase=phase,
        )
        context.error = None
    except ValidationError as exc:
        context.error = exc


@when("I try to create an async job with invalid payload JSON")
def step_try_create_invalid_payload(context: Context) -> None:
    try:
        AsyncJob(
            job_id=_make_job_id(),
            plan_id=PLAN_ID,
            phase="execute",
            payload_json="not json!!!",
        )
        context.error = None
    except ValidationError as exc:
        context.error = exc


@when("I cancel the async token")
def step_when_cancel_token(context: Context) -> None:
    context.token.cancel()


@when("I reset the token")
def step_reset_token(context: Context) -> None:
    context.token.reset()


@when("I record {completed:d} completed jobs and {failed:d} failed job")
def step_record_jobs(context: Context, completed: int, failed: int) -> None:
    for _ in range(completed):
        context.health_report.record_job_completed()
    for _ in range(failed):
        context.health_report.record_job_failed()


@when("I try to serialize a payload with empty plan_id")
def step_try_serialize_empty_plan_id(context: Context) -> None:
    try:
        serialize_job_payload("", "execute")
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I try to deserialize an empty payload")
def step_try_deserialize_empty(context: Context) -> None:
    try:
        deserialize_job_payload("")
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I call can_transition_job with invalid types")
def step_can_transition_invalid(context: Context) -> None:
    try:
        can_transition_job("queued", "running")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


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


@then('the async job status should be "{status}"')
def step_check_status(context: Context, status: str) -> None:
    assert context.async_job.status.value == status, (
        f"Expected {status}, got {context.async_job.status.value}"
    )


@then('the async job phase should be "{phase}"')
def step_check_phase(context: Context, phase: str) -> None:
    assert context.async_job.phase == phase


@then("the async job schema_version should be {version:d}")
def step_check_schema_version(context: Context, version: int) -> None:
    assert context.async_job.schema_version == version


@then('the async job worker_id should be "{worker_id}"')
def step_check_worker_id(context: Context, worker_id: str) -> None:
    assert context.async_job.worker_id == worker_id


@then("the async job started_at should be set")
def step_check_started_at(context: Context) -> None:
    assert context.async_job.started_at is not None


@then("the async job finished_at should be set")
def step_check_finished_at(context: Context) -> None:
    assert context.async_job.finished_at is not None


@then("an InvalidJobTransitionError should be raised")
def step_check_transition_error(context: Context) -> None:
    assert isinstance(context.error, InvalidJobTransitionError), (
        f"Expected InvalidJobTransitionError, got {type(context.error)}: {context.error}"
    )


@then("the worker health should show {count:d} job processed")
def step_check_health_processed(context: Context, count: int) -> None:
    assert context.worker.health.jobs_processed == count


@then("the worker health should show {count:d} job failed")
def step_check_health_failed(context: Context, count: int) -> None:
    assert context.worker.health.jobs_failed == count


@then('the stuck job should be marked as "{status}"')
def step_check_stuck_status(context: Context, status: str) -> None:
    job = context.job_store.get(context.async_job.job_id)
    assert job is not None
    assert job.status.value == status


@then("the completed job should be removed from the store")
def step_check_job_removed(context: Context) -> None:
    job = context.job_store.get(context.async_job.job_id)
    assert job is None


@then("the worker should stop running")
def step_check_worker_stopped(context: Context) -> None:
    assert not context.worker.is_running


@then("the serialized payload should contain schema_version {version:d}")
def step_check_payload_schema(context: Context, version: int) -> None:
    data = json.loads(context.payload_json)
    assert data["schema_version"] == version


@then("the deserialized payload should match the original")
def step_check_deserialized(context: Context) -> None:
    data = deserialize_job_payload(context.payload_json)
    assert data["plan_id"] == context.original_plan_id
    assert data["phase"] == context.original_phase


@then("the serialized payload should contain the extra data")
def step_check_extra_data(context: Context) -> None:
    data = json.loads(context.payload_json)
    assert data["extra"]["key"] == "value"


@then("a ValueError should be raised for async deserialize")
def step_check_value_error(context: Context) -> None:
    assert isinstance(context.error, ValueError), (
        f"Expected ValueError, got {type(context.error)}"
    )


@then("the health report should contain worker_id")
def step_check_health_worker_id(context: Context) -> None:
    assert "worker_id" in context.health_data


@then("the health report should contain config details")
def step_check_health_config(context: Context) -> None:
    assert "config" in context.health_data
    assert context.health_data["config"]["max_workers"] == 4


@then("a ValueError should be raised for config")
def step_check_config_error(context: Context) -> None:
    assert isinstance(context.config_error, ValueError)


@then("a ValueError should be raised for duplicate job")
def step_check_duplicate_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then("the job list should have {count:d} entries")
def step_check_job_list_count(context: Context, count: int) -> None:
    assert len(context.job_list) == count


@then("the async job last_heartbeat should be updated")
def step_check_heartbeat_updated(context: Context) -> None:
    assert context.async_job.last_heartbeat is not None
    if context.heartbeat_before is not None:
        assert context.async_job.last_heartbeat >= context.heartbeat_before


@then("a ValueError should be raised for heartbeat")
def step_check_heartbeat_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then("the async job should be terminal")
def step_check_is_terminal(context: Context) -> None:
    assert context.async_job.is_terminal


@then("the async job should be running")
def step_check_is_running(context: Context) -> None:
    assert context.async_job.is_running


@then('the async CLI dict should contain key "{key}"')
def step_check_cli_dict_key(context: Context, key: str) -> None:
    assert key in context.cli_dict


@then("a validation error should be raised for phase")
def step_check_phase_error(context: Context) -> None:
    assert context.error is not None


@then("a validation error should be raised for payload")
def step_check_payload_error(context: Context) -> None:
    assert context.error is not None


@then("a ValueError should be raised for worker_id")
def step_check_worker_id_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then("the token should be cancelled")
def step_check_token_cancelled(context: Context) -> None:
    assert context.token.is_cancelled


@then("the token wait should return True")
def step_check_token_wait(context: Context) -> None:
    assert context.token.wait(timeout=0.01)


@then("the token should not be cancelled")
def step_check_token_not_cancelled(context: Context) -> None:
    assert not context.token.is_cancelled


@then("the health report should show {processed:d} processed and {failed:d} failed")
def step_check_health_counts(context: Context, processed: int, failed: int) -> None:
    assert context.health_report.jobs_processed == processed
    assert context.health_report.jobs_failed == failed


@then("a ValueError should be raised for serialize")
def step_check_serialize_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then("a ValueError should be raised for deserialize")
def step_check_deserialize_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then("a TypeError should be raised for async transition")
def step_check_type_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


# ---- Additional Coverage Steps ----


@when("I try to create a WorkerHealthReport with empty worker_id")
def step_try_empty_health_report(context: Context) -> None:
    try:
        WorkerHealthReport("")
        context.error = None
    except ValueError as exc:
        context.error = exc


@then("a ValueError should be raised for empty worker_id")
def step_check_empty_worker_id_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@when("I try to create AsyncWorkerConfig with non-bool enabled")
def step_try_config_nonbool(context: Context) -> None:
    try:
        AsyncWorkerConfig(enabled="yes")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@then("a TypeError should be raised for async config enabled")
def step_check_config_enabled_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


@when("I try to create AsyncWorkerConfig with job_timeout {timeout:d}")
def step_try_config_job_timeout(context: Context, timeout: int) -> None:
    try:
        AsyncWorkerConfig(job_timeout=timeout)
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I try to create AsyncWorkerConfig with job_ttl {ttl:d}")
def step_try_config_job_ttl(context: Context, ttl: int) -> None:
    try:
        AsyncWorkerConfig(job_ttl=ttl)
        context.error = None
    except ValueError as exc:
        context.error = exc


@then("a ValueError should be raised for async config validation")
def step_check_config_validation_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@when("I try to add a non-AsyncJob to the store")
def step_try_add_non_job(context: Context) -> None:
    try:
        context.job_store.add("not a job")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@then("a TypeError should be raised for async store type")
def step_check_store_type_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


@when("I try to get a job with empty job_id")
def step_try_get_empty_id(context: Context) -> None:
    try:
        context.job_store.get("")
        context.error = None
    except ValueError as exc:
        context.error = exc


@then("a ValueError should be raised for async store get")
def step_check_store_get_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@when("I try to list jobs by invalid status type")
def step_try_list_invalid_status(context: Context) -> None:
    try:
        context.job_store.list_by_status("invalid")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@when("I try to update with a non-AsyncJob")
def step_try_update_non_job(context: Context) -> None:
    try:
        context.job_store.update("not a job")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@when("I try to update a non-existent job")
def step_try_update_nonexistent(context: Context) -> None:
    try:
        job = AsyncJob(
            job_id=str(ULID()),
            plan_id="01HXYZ01HXYZ01HXYZ01HXYZ01",
            phase="execute",
        )
        context.job_store.update(job)
        context.error = None
    except ValueError as exc:
        context.error = exc


@then("a ValueError should be raised for async store update")
def step_check_store_update_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@when("I try to remove a job with empty job_id")
def step_try_remove_empty_id(context: Context) -> None:
    try:
        context.job_store.remove("")
        context.error = None
    except ValueError as exc:
        context.error = exc


@then("a ValueError should be raised for async store remove")
def step_check_store_remove_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@when("I try to create AsyncWorker with invalid config")
def step_try_worker_invalid_config(context: Context) -> None:
    try:
        AsyncWorker(config="bad", job_store=InMemoryJobStore())  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@then("a TypeError should be raised for async worker config")
def step_check_worker_config_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


@when("I try to create AsyncWorker with invalid job_store")
def step_try_worker_invalid_store(context: Context) -> None:
    try:
        AsyncWorker(config=AsyncWorkerConfig(), job_store="bad")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@then("a TypeError should be raised for async worker store")
def step_check_worker_store_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


@when("I try to pickup a non-AsyncJob")
def step_try_pickup_non_job(context: Context) -> None:
    try:
        context.worker.pickup_and_execute("bad")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@then("a TypeError should be raised for async worker pickup")
def step_check_worker_pickup_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


@when("I try to cancel a job with empty job_id")
def step_try_cancel_empty_id(context: Context) -> None:
    try:
        context.worker.cancel_job("")
        context.error = None
    except ValueError as exc:
        context.error = exc


@then("a ValueError should be raised for async worker cancel")
def step_check_worker_cancel_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then('the worker should have a worker_id starting with "worker-"')
def step_check_worker_id_prefix(context: Context) -> None:
    assert context.worker.worker_id.startswith("worker-")


@then("the worker config max_workers should be {max_workers:d}")
def step_check_worker_config_value(context: Context, max_workers: int) -> None:
    assert context.worker.config.max_workers == max_workers


@when("I start the worker")
def step_start_worker(context: Context) -> None:
    context.worker.start()


@when("I start the worker again")
def step_start_worker_again(context: Context) -> None:
    context.worker.start()


@then("the worker should still be running")
def step_worker_still_running(context: Context) -> None:
    assert context.worker.is_running


@when("I stop the worker")
def step_stop_worker(context: Context) -> None:
    context.worker.stop()


@then("the worker should not be running")
def step_worker_not_running(context: Context) -> None:
    assert not context.worker.is_running


@then("cancelling a non-existent job should return false")
def step_cancel_nonexistent_job(context: Context) -> None:
    result = context.worker.cancel_job("01ZZZZZZZZZZZZZZZZZZZZZZZZ")
    assert result is False


@then("the async job payload property should return a dict")
def step_check_payload_property(context: Context) -> None:
    payload = context.async_job.payload
    assert isinstance(payload, dict)


@when('I try to create an async job with payload_json "not-json{{"')
def step_try_invalid_payload_json(context: Context) -> None:
    try:
        AsyncJob(
            job_id=str(ULID()),
            plan_id="01HXYZ01HXYZ01HXYZ01HXYZ01",
            phase="execute",
            payload_json="not-json{",
        )
        context.error = None
    except Exception as exc:
        context.error = exc


# ---------------------------------------------------------------------------
# Review-fix steps: concurrent execution (T2)
# ---------------------------------------------------------------------------


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


@when("the worker executes all queued jobs concurrently")
def step_worker_executes_concurrently(context: Context) -> None:
    """Use threads to verify concurrent execution."""
    threads: list[threading.Thread] = []
    for job in context.created_jobs:
        t = threading.Thread(target=context.worker.pickup_and_execute, args=(job,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join(timeout=5.0)


@then("all {count:d} jobs should be in a terminal state")
def step_all_jobs_terminal(context: Context, count: int) -> None:
    terminal = [j for j in context.created_jobs if j.is_terminal]
    assert len(terminal) == count, f"Expected {count} terminal, got {len(terminal)}"


@then("the worker health should show {count:d} jobs processed total")
def step_check_total_processed(context: Context, count: int) -> None:
    assert context.worker.health.jobs_processed == count, (
        f"Expected {count} processed, got {context.worker.health.jobs_processed}"
    )


@then("the worker health should show {count:d} jobs processed")
def step_check_health_processed_exact(context: Context, count: int) -> None:
    assert context.worker.health.jobs_processed == count


@then("the worker health should show {count:d} jobs failed")
def step_check_health_failed_exact(context: Context, count: int) -> None:
    assert context.worker.health.jobs_failed == count


# ---------------------------------------------------------------------------
# Review-fix steps: error_message audit trail (SEC1)
# ---------------------------------------------------------------------------


@then("the async job should have an error_message set")
def step_check_error_message_set(context: Context) -> None:
    assert context.async_job.error_message is not None, (
        "Expected error_message to be set, but it was None"
    )
    assert len(context.async_job.error_message) > 0


# ---------------------------------------------------------------------------
# Review-fix steps: AsyncJobModel DB round-trip (T1)
# ---------------------------------------------------------------------------


@when("I create and persist an AsyncJobModel to the database")
def step_create_persist_db_model(context: Context) -> None:
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

    from cleveragents.infrastructure.database.models import AsyncJobModel, Base

    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    context.db_engine = engine
    context.db_session_factory = sessionmaker(bind=engine)

    job_id = str(ULID())
    context.db_job_id = job_id
    with context.db_session_factory() as session:
        model = AsyncJobModel(
            job_id=job_id,
            plan_id=PLAN_ID,
            phase="execute",
            status="queued",
            payload_json='{"schema_version": 1}',
            created_at=datetime.now(UTC).isoformat(),
            schema_version=1,
        )
        session.add(model)
        session.commit()


@then("the AsyncJobModel should be retrievable with correct fields")
def step_check_db_model_retrieve(context: Context) -> None:
    from cleveragents.infrastructure.database.models import AsyncJobModel

    with context.db_session_factory() as session:
        row = session.query(AsyncJobModel).filter_by(job_id=context.db_job_id).first()
        assert row is not None, "AsyncJobModel not found in DB"
        assert row.plan_id == PLAN_ID
        assert row.phase == "execute"
        assert row.status == "queued"
        assert row.schema_version == 1


@then("the database CHECK constraint should reject invalid status")
def step_check_db_constraint(context: Context) -> None:
    from sqlalchemy.exc import IntegrityError

    from cleveragents.infrastructure.database.models import AsyncJobModel

    with context.db_session_factory() as session:
        bad = AsyncJobModel(
            job_id=str(ULID()),
            plan_id=PLAN_ID,
            phase="execute",
            status="invalid_status",
            payload_json="{}",
            created_at=datetime.now(UTC).isoformat(),
            schema_version=1,
        )
        session.add(bad)
        try:
            session.commit()
            # SQLite enforces CHECK constraints; this should fail
            context.db_constraint_enforced = False
        except IntegrityError:
            session.rollback()
            context.db_constraint_enforced = True

    assert context.db_constraint_enforced, (
        "Expected CHECK constraint to reject invalid status"
    )


# ---------------------------------------------------------------------------
# Coverage steps: Safe initialisation and cleanup
# ---------------------------------------------------------------------------


@given("an async worker is initialised with a blocking executor")
def step_worker_with_blocking_executor(context: Context) -> None:
    """Create a worker whose executor blocks until a threading.Event is set."""
    entered_event = threading.Event()
    release_event = threading.Event()
    context.blocking_entered = entered_event
    context.blocking_release = release_event

    def blocking_executor(job: AsyncJob, token: CancellationToken) -> None:
        entered_event.set()
        # Block until released or cancelled
        while not release_event.is_set() and not token.is_cancelled:
            release_event.wait(timeout=0.05)

    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.05,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store, blocking_executor)
    context.async_job = None
    context.error = None


@given("an async worker is initialised with a terminal-forcing executor")
def step_worker_with_terminal_executor(context: Context) -> None:
    """Create a worker whose executor forcefully marks the job succeeded."""

    def terminal_executor(job: AsyncJob, token: CancellationToken) -> None:
        # Force the job to succeeded during execution — pickup_and_execute
        # should detect this and skip its own transition.
        job.mark_succeeded()

    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.1,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store, terminal_executor)
    context.async_job = None
    context.error = None


@when("the worker processes all queued jobs within {seconds:d} seconds")
def step_worker_processes_queued(context: Context, seconds: int) -> None:
    """Wait until context.async_job reaches a terminal state."""
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        if context.async_job.is_terminal:
            return
        time.sleep(0.05)
    assert context.async_job.is_terminal, (
        f"Job did not reach terminal state within {seconds}s "
        f"(status={context.async_job.status.value})"
    )


@when("all created jobs reach a terminal state within {seconds:d} seconds")
def step_all_created_jobs_terminal(context: Context, seconds: int) -> None:
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        if all(j.is_terminal for j in context.created_jobs):
            return
        time.sleep(0.05)
    pending = [j for j in context.created_jobs if not j.is_terminal]
    assert not pending, f"{len(pending)} job(s) still not terminal after {seconds}s"


@when("the blocking executor has been entered")
def step_blocking_executor_entered(context: Context) -> None:
    entered = context.blocking_entered.wait(timeout=5.0)
    assert entered, "Blocking executor was not entered within 5s"


@when("I dispatch the job in a background thread")
def step_dispatch_job_background(context: Context) -> None:
    t = threading.Thread(
        target=context.worker.pickup_and_execute,
        args=(context.async_job,),
    )
    context.background_job_thread = t
    t.start()


@when("I cancel the running async job via the worker")
def step_cancel_running_job(context: Context) -> None:
    result = context.worker.cancel_job(context.async_job.job_id)
    context.cancel_result = result
    # Release the blocking executor so the thread can finish
    if hasattr(context, "blocking_release"):
        context.blocking_release.set()


@when("the background job thread completes")
def step_background_thread_completes(context: Context) -> None:
    context.background_job_thread.join(timeout=5.0)
    assert not context.background_job_thread.is_alive(), (
        "Background job thread did not complete within 5s"
    )


@given("the async job is already cancelled")
def step_job_already_cancelled(context: Context) -> None:
    context.async_job.mark_cancelled()
    context.job_store.update(context.async_job)


@then("the cancellation token for the job should be signalled")
def step_check_cancel_token_signalled(context: Context) -> None:
    # After stop(), all tokens should have been cancelled
    # We verify the job reached a terminal state or the token was signalled
    # The _cancel_all_running_jobs path fires during stop()
    # Since the job was in-flight, it should be terminal
    job = context.job_store.get(context.async_job.job_id)
    assert job is not None
    # The job should be terminal (cancelled or succeeded) after stop
    # releases executor and cancels tokens
    assert job.is_terminal or context.async_job.is_terminal, (
        f"Expected terminal job after stop, got {job.status.value}"
    )


@then("the worker health should show {count:d} jobs cancelled")
def step_check_health_cancelled(context: Context, count: int) -> None:
    assert context.worker.health.jobs_cancelled == count, (
        f"Expected {count} cancelled, got {context.worker.health.jobs_cancelled}"
    )


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


# ---------------------------------------------------------------------------
# Coverage steps: Job store operations
# ---------------------------------------------------------------------------


@given("I add {count:d} jobs in mixed statuses to the store")
def step_add_mixed_jobs(context: Context, count: int) -> None:
    context.store_jobs = []
    statuses_to_set = ["queued", "running", "succeeded"]
    for i in range(count):
        job = AsyncJob(job_id=_make_job_id(), plan_id=PLAN_ID, phase="execute")
        target = statuses_to_set[i % len(statuses_to_set)]
        if target == "running":
            job.mark_running(f"worker-{i}")
        elif target == "succeeded":
            job.mark_running(f"worker-{i}")
            job.mark_succeeded()
        context.job_store.add(job)
        context.store_jobs.append(job)


@when("I call list_all on the job store")
def step_call_list_all(context: Context) -> None:
    context.all_jobs = context.job_store.list_all()


@then("the store should return {count:d} jobs")
def step_store_returns_count(context: Context, count: int) -> None:
    assert len(context.all_jobs) == count, (
        f"Expected {count}, got {len(context.all_jobs)}"
    )


@then("the store count should be {count:d}")
def step_store_count(context: Context, count: int) -> None:
    assert context.job_store.count() == count, (
        f"Expected count {count}, got {context.job_store.count()}"
    )


@given("a queued async job exists in the store")
def step_queued_job_in_store(context: Context) -> None:
    job = AsyncJob(job_id=_make_job_id(), plan_id=PLAN_ID, phase="execute")
    context.job_store.add(job)
    context.async_job = job


@when("I remove the job from the store")
def step_remove_job_from_store(context: Context) -> None:
    context.job_store.remove(context.async_job.job_id)


# ---------------------------------------------------------------------------
# Coverage steps: Domain model validators
# ---------------------------------------------------------------------------


@given("a queued async job exists standalone")
def step_queued_job_standalone(context: Context) -> None:
    context.job_store = InMemoryJobStore()
    context.async_job = AsyncJob(
        job_id=_make_job_id(), plan_id=PLAN_ID, phase="execute"
    )
    context.job_store.add(context.async_job)
    context.error = None


@when("I try to transition_to with a string instead of enum")
def step_transition_with_string(context: Context) -> None:
    try:
        context.async_job.transition_to("running")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@when("I call can_transition_job with a string to_status")
def step_can_transition_with_string(context: Context) -> None:
    try:
        can_transition_job(AsyncJobStatus.QUEUED, "running")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@then("an async TypeError should be raised")
def step_check_async_type_error(context: Context) -> None:
    assert isinstance(context.error, TypeError), (
        f"Expected TypeError, got {type(context.error)}: {context.error}"
    )


@when("I try to serialize a payload with an empty phase")
def step_serialize_empty_phase(context: Context) -> None:
    try:
        serialize_job_payload(PLAN_ID, "")
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I try to serialize a payload with schema_version {version:d}")
def step_serialize_bad_version(context: Context, version: int) -> None:
    try:
        serialize_job_payload(PLAN_ID, "execute", schema_version=version)
        context.error = None
    except ValueError as exc:
        context.error = exc


@then('an async ValueError should be raised matching "{fragment}"')
def step_check_async_value_error_message(context: Context, fragment: str) -> None:
    assert isinstance(context.error, ValueError), (
        f"Expected ValueError, got {type(context.error)}: {context.error}"
    )
    assert fragment in str(context.error), f"Expected '{fragment}' in '{context.error}'"
    assert fragment in str(context.error), f"Expected '{fragment}' in '{context.error}'"


@when('I try to deserialize the JSON string "{json_str}"')
def step_deserialize_non_object(context: Context, json_str: str) -> None:
    try:
        deserialize_job_payload(json_str)
        context.error = None
    except ValueError as exc:
        context.error = exc


@when('I mark the async job as failed with error "{error_msg}"')
def step_mark_failed_with_error(context: Context, error_msg: str) -> None:
    context.async_job.mark_failed(error=error_msg)


@then('the async CLI dict error_message should be "{expected}"')
def step_check_cli_dict_error_message(context: Context, expected: str) -> None:
    assert context.cli_dict["error_message"] == expected, (
        f"Expected '{expected}', got '{context.cli_dict.get('error_message')}'"
    )


# ---------------------------------------------------------------------------
# Coverage steps: Transition-conflict edge cases
# ---------------------------------------------------------------------------


@given('the async job is already running with worker "{worker_id}"')
def step_job_already_running(context: Context, worker_id: str) -> None:
    """Manually transition the job to running so pickup_and_execute
    encounters an InvalidJobTransitionError on mark_running."""
    context.async_job.mark_running(worker_id)
    context.job_store.update(context.async_job)


@given("an async worker is initialised with a crash-after-terminal executor")
def step_worker_crash_after_terminal(context: Context) -> None:
    """Executor that marks the job succeeded then raises, testing the
    inner InvalidJobTransitionError path in the exception handler."""

    def crash_after_terminal(job: AsyncJob, token: CancellationToken) -> None:
        job.mark_succeeded()
        raise RuntimeError("Post-terminal crash")

    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.1,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store, crash_after_terminal)
    context.async_job = None
    context.error = None


@given("the running async job has already been cancelled externally")
def step_running_job_cancelled_externally(context: Context) -> None:
    """Simulate a race where the job becomes cancelled between the
    stuck-detection check and the mark_failed call."""
    context.async_job.mark_cancelled()
    context.job_store.update(context.async_job)


@then("no additional stuck jobs should be reported")
def step_no_stuck_jobs(context: Context) -> None:
    assert len(context.stuck_jobs) == 0, (
        f"Expected 0 stuck jobs, got {len(context.stuck_jobs)}"
    )


# ---------------------------------------------------------------------------
# Review-fix steps: Plan lifecycle async wiring (P1)
# ---------------------------------------------------------------------------


def _make_mock_settings(async_enabled: bool = False) -> MagicMock:
    """Create a mock Settings object with async config."""
    settings = MagicMock()
    settings.async_enabled = async_enabled
    return settings


@given("a plan lifecycle service with async enabled and a job store")
def step_lifecycle_async_enabled(context: Context) -> None:
    context.lifecycle_job_store = InMemoryJobStore()
    context.lifecycle_settings = _make_mock_settings(async_enabled=True)
    context.lifecycle_service = PlanLifecycleService(
        settings=context.lifecycle_settings,
        job_store=context.lifecycle_job_store,
    )
    context.lifecycle_plan = None


@given("a plan lifecycle service with async disabled")
def step_lifecycle_async_disabled(context: Context) -> None:
    context.lifecycle_job_store = InMemoryJobStore()
    context.lifecycle_settings = _make_mock_settings(async_enabled=False)
    context.lifecycle_service = PlanLifecycleService(
        settings=context.lifecycle_settings,
        job_store=context.lifecycle_job_store,
    )
    context.lifecycle_plan = None


@given("a plan lifecycle service with async enabled but no job store")
def step_lifecycle_async_no_store(context: Context) -> None:
    context.lifecycle_job_store = None
    context.lifecycle_settings = _make_mock_settings(async_enabled=True)
    context.lifecycle_service = PlanLifecycleService(
        settings=context.lifecycle_settings,
        job_store=None,
    )
    context.lifecycle_plan = None


@given("a plan in Strategize/COMPLETE state exists")
def step_plan_strategize_complete(context: Context) -> None:
    context.lifecycle_service.create_action(
        name="local/test-action",
        description="Test action",
        definition_of_done="Done",
        strategy_actor="local/strategy",
        execution_actor="local/executor",
    )
    plan = context.lifecycle_service.use_action("local/test-action")
    context.lifecycle_service.start_strategize(plan.identity.plan_id)
    context.lifecycle_service.complete_strategize(plan.identity.plan_id)
    context.lifecycle_plan = context.lifecycle_service.get_plan(plan.identity.plan_id)


@given("a plan in Execute/COMPLETE state exists")
def step_plan_execute_complete(context: Context) -> None:
    context.lifecycle_service.create_action(
        name="local/test-action-apply",
        description="Test action for apply",
        definition_of_done="Done",
        strategy_actor="local/strategy",
        execution_actor="local/executor",
    )
    plan = context.lifecycle_service.use_action("local/test-action-apply")
    context.lifecycle_service.start_strategize(plan.identity.plan_id)
    plan = context.lifecycle_service.complete_strategize(plan.identity.plan_id)
    # complete_strategize may auto-progress; if not, manually advance
    plan = context.lifecycle_service.get_plan(plan.identity.plan_id)
    if plan.phase.value == "strategize":
        plan = context.lifecycle_service.execute_plan(plan.identity.plan_id)
    # Clear any jobs enqueued by execute_plan before testing apply
    if context.lifecycle_job_store is not None:
        for j in context.lifecycle_job_store.list_all():
            context.lifecycle_job_store.remove(j.job_id)
    context.lifecycle_service.start_execute(plan.identity.plan_id)
    context.lifecycle_service.complete_execute(plan.identity.plan_id)
    context.lifecycle_plan = context.lifecycle_service.get_plan(plan.identity.plan_id)


@when("I execute the plan via the lifecycle service")
def step_execute_plan_lifecycle(context: Context) -> None:
    plan = context.lifecycle_plan
    context.lifecycle_plan = context.lifecycle_service.execute_plan(
        plan.identity.plan_id
    )


@when("I apply the plan via the lifecycle service")
def step_apply_plan_lifecycle(context: Context) -> None:
    plan = context.lifecycle_plan
    context.lifecycle_plan = context.lifecycle_service.apply_plan(plan.identity.plan_id)


@then("the plan should be in Execute/QUEUED state")
def step_check_plan_execute_queued(context: Context) -> None:
    assert context.lifecycle_plan.phase.value == "execute", (
        f"Expected execute, got {context.lifecycle_plan.phase.value}"
    )
    assert context.lifecycle_plan.processing_state.value == "queued", (
        f"Expected queued, got {context.lifecycle_plan.processing_state.value}"
    )


@then("the plan should be in Apply/QUEUED state")
def step_check_plan_apply_queued(context: Context) -> None:
    assert context.lifecycle_plan.phase.value == "apply", (
        f"Expected apply, got {context.lifecycle_plan.phase.value}"
    )
    assert context.lifecycle_plan.processing_state.value == "queued", (
        f"Expected queued, got {context.lifecycle_plan.processing_state.value}"
    )


@then('the job store should contain {count:d} async job for phase "{phase}"')
def step_check_job_store_phase(context: Context, count: int, phase: str) -> None:
    jobs = context.lifecycle_job_store.list_by_status(AsyncJobStatus.QUEUED)
    matching = [j for j in jobs if j.phase == phase]
    assert len(matching) == count, (
        f"Expected {count} job(s) for phase {phase}, got {len(matching)}"
    )


@then("the enqueued job plan_id should match the plan")
def step_check_enqueued_plan_id(context: Context) -> None:
    jobs = context.lifecycle_job_store.list_all()
    assert len(jobs) > 0, "No jobs in store"
    plan_id = context.lifecycle_plan.identity.plan_id
    assert any(j.plan_id == plan_id for j in jobs), (
        f"No job with plan_id {plan_id} found"
    )


@then("no async jobs should be in the job store")
def step_check_no_async_jobs(context: Context) -> None:
    assert context.lifecycle_job_store.count() == 0, (
        f"Expected 0 jobs, got {context.lifecycle_job_store.count()}"
    )


# ---------------------------------------------------------------------------
# Review-fix steps: Error message redaction (P2)
# ---------------------------------------------------------------------------


@given("an async worker is initialised with a secret-leaking executor")
def step_worker_secret_leaking_executor(context: Context) -> None:
    """Create a worker whose executor raises an error containing a secret."""

    def leaking_executor(job: AsyncJob, token: CancellationToken) -> None:
        raise RuntimeError(
            "Connection failed: api_key=sk-proj-ABCDEFGHIJ1234567890abcdef"
        )

    config = AsyncWorkerConfig(
        enabled=True,
        max_workers=2,
        poll_interval=0.1,
        job_timeout=3600,
        job_ttl=86400,
    )
    context.job_store = InMemoryJobStore()
    context.worker = AsyncWorker(config, context.job_store, leaking_executor)
    context.async_job = None
    context.error = None


@then("the async job error_message should not contain the raw secret")
def step_check_no_raw_secret(context: Context) -> None:
    error_msg = context.async_job.error_message
    assert error_msg is not None, "error_message is None"
    assert "sk-proj-ABCDEFGHIJ1234567890abcdef" not in error_msg, (
        f"Raw secret found in error_message: {error_msg}"
    )


@then('the async job error_message should contain "***REDACTED***"')
def step_check_redacted_marker(context: Context) -> None:
    error_msg = context.async_job.error_message
    assert error_msg is not None, "error_message is None"
    assert "***REDACTED***" in error_msg, (
        f"Expected ***REDACTED*** in error_message, got: {error_msg}"
    )