cleveragents-core/features/steps/devcontainer_health_check_steps.py

"""Step definitions for devcontainer health check and concurrency operations.

Covers: single health probe, health check loop iterations, start/stop
health check threads, thread registration and join, probe workspace
fallback, health check argument validation, concurrent activation,
and concurrent stop.

All mocks are in ``features/mocks/mock_devcontainer_cli.py``.
"""

from __future__ import annotations

import threading
import time
from typing import Any, cast
from unittest.mock import MagicMock
from unittest.mock import patch as _patch

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

from cleveragents.domain.models.core.container_lifecycle import (
    ContainerLifecycleState,
    ContainerLifecycleTracker,
)
from cleveragents.resource.handlers import devcontainer as _dc_mod
from cleveragents.resource.handlers.devcontainer import (
    DevcontainerHandler,
    _health_check_loop,
    _health_check_stop_events,
    _health_check_threads,
    _single_probe,
    _stop_health_check,
    activate_container,
    clear_lifecycle_registry,
    get_lifecycle_tracker,
    set_lifecycle_tracker,
    start_health_check,
    stop_container,
)

# Import _OneShotEvent from the core lifecycle steps module to avoid
# duplicating the helper class.
from features.steps.devcontainer_lifecycle_steps import _OneShotEvent

# ── Given steps ──────────────────────────────────────────────


@given("the runner configured for failed health check")
def step_configure_hc_failure(context: Context) -> None:
    context.mock_runner.set_exec_failure()


@given("the runner configured for successful health check")
def step_configure_hc_success(context: Context) -> None:
    context.mock_runner.set_exec_result(returncode=0, stdout="ping")


@given("the runner configured to raise exception on exec")
def step_configure_exec_exception(context: Context) -> None:
    """Replace mock_runner with a wrapper that raises on exec."""
    inner = context.mock_runner

    def raising_runner(args: list[str], **kwargs: object) -> object:
        inner.calls.append((list(args), dict(kwargs)))
        if len(args) >= 2 and args[0] == "devcontainer" and args[1] == "exec":
            raise OSError("connection refused")
        return inner(args, **kwargs)

    context.mock_runner = cast(Any, raising_runner)


@given('a registered health check for "{resource_id}"')
def step_register_health_check(context: Context, resource_id: str) -> None:
    """Register a real (but idle) health check thread so _stop_health_check
    can join it and we can verify the full thread-join path (F9 fix).
    """
    stop_event = threading.Event()
    _health_check_stop_events[resource_id] = stop_event

    # F9 fix: use a real short-lived thread instead of current_thread()
    # so _stop_health_check actually calls thread.join() rather than
    # skipping the join due to the ``is not current_thread()`` guard.
    def _idle_loop() -> None:
        stop_event.wait(timeout=5.0)

    thread = threading.Thread(target=_idle_loop, daemon=True)
    _health_check_threads[resource_id] = thread
    thread.start()


@given('a running health check for "{resource_id}"')
def step_start_health_check(context: Context, resource_id: str) -> None:
    """Start a health check thread that we can later verify was joined."""
    context.mock_runner.set_exec_result(returncode=0, stdout="ping")
    start_health_check(resource_id, interval=0.1, run_command=context.mock_runner)
    # Give it a moment to start
    time.sleep(0.05)


# ── When steps ───────────────────────────────────────────────


@when('I run a single health check probe for "{resource_id}"')
def step_single_health_probe(context: Context, resource_id: str) -> None:
    """Run a single health check probe via the production _health_check_loop.

    Uses the shared _OneShotEvent to run exactly one probe iteration
    through the real production code path, avoiding reimplementation
    of transition logic in the test step.
    """
    _health_check_loop(
        resource_id,
        interval=0.0,
        stop_event=cast(threading.Event, _OneShotEvent()),
        run_command=context.mock_runner,
    )


@when('I run one iteration of the health check loop for "{resource_id}"')
def step_run_health_check_loop_once(
    context: Context,
    resource_id: str,
) -> None:
    """Run _health_check_loop with the shared _OneShotEvent."""
    context.hc_resource_id = resource_id

    _health_check_loop(
        resource_id,
        interval=0.0,
        stop_event=cast(threading.Event, _OneShotEvent()),
        run_command=context.mock_runner,
    )


@when('I call start_health_check for "{resource_id}"')
def step_call_start_health_check(context: Context, resource_id: str) -> None:
    """Call start_health_check and immediately stop the thread."""
    start_health_check(
        resource_id,
        interval=60.0,  # long interval — we stop it immediately
        run_command=context.mock_runner,
    )
    context.hc_resource_id = resource_id
    # Give thread just enough time to register
    time.sleep(0.05)


@when('I stop the health check for "{resource_id}"')
def step_stop_health_check(context: Context, resource_id: str) -> None:
    _stop_health_check(resource_id)
    context.health_check_resource_id = resource_id


@when("I attempt to start health check with empty resource_id")
def step_start_hc_empty_id(context: Context) -> None:
    context.hc_error = None
    try:
        start_health_check("")
    except ValueError as exc:
        context.hc_error = exc


@when("I run a single probe with no workspace_path set")
def step_probe_no_workspace(context: Context) -> None:
    """Run _single_probe with a tracker that has workspace_path=None."""
    tracker = ContainerLifecycleTracker(
        resource_id="01TESTLIFECYCLE0000000320",
        current_state=ContainerLifecycleState.RUNNING,
        container_id="ctr-t3",
        workspace_path=None,
    )
    set_lifecycle_tracker(tracker)
    context.mock_runner.set_exec_result(returncode=0, stdout="ping")
    _single_probe(tracker, context.mock_runner)
    context.probe_calls = context.mock_runner.exec_calls


# ── Then steps ───────────────────────────────────────────────


@then('the health check should have probed "{resource_id}"')
def step_check_health_probe(context: Context, resource_id: str) -> None:
    exec_calls = context.mock_runner.exec_calls
    assert len(exec_calls) > 0, "Expected at least one exec call from health check"


@then('a health check thread should be registered for "{resource_id}"')
def step_check_thread_registered(context: Context, resource_id: str) -> None:
    assert resource_id in _health_check_threads, (
        f"Thread not registered for {resource_id}"
    )
    assert resource_id in _health_check_stop_events, (
        f"Stop event not registered for {resource_id}"
    )
    # Clean up — stop the thread
    clear_lifecycle_registry()


@then('the health check for "{resource_id}" should be unregistered')
def step_check_hc_unregistered(context: Context, resource_id: str) -> None:
    assert resource_id not in _health_check_stop_events, (
        f"Stop event still registered for {resource_id}"
    )
    assert resource_id not in _health_check_threads, (
        f"Thread still registered for {resource_id}"
    )


@then("the health check should raise ValueError")
def step_check_hc_value_error(context: Context) -> None:
    assert context.hc_error is not None
    assert isinstance(context.hc_error, ValueError)


@then("the health check thread should have been joined")
def step_check_thread_joined(context: Context) -> None:
    rid = context.health_check_resource_id
    # After _stop_health_check, thread should have been removed from registry
    assert rid not in _health_check_threads, (
        f"Thread for {rid} still in _health_check_threads after stop"
    )
    assert rid not in _health_check_stop_events, (
        f"Stop event for {rid} still in _health_check_stop_events after stop"
    )


@then('the probe command should use "{path}" as workspace folder')
def step_check_probe_workspace(context: Context, path: str) -> None:
    assert len(context.probe_calls) >= 1, "Expected at least one exec call"
    args, _ = context.probe_calls[-1]
    # args: ["devcontainer", "exec", "--workspace-folder", path, "echo", "ping"]
    ws_idx = args.index("--workspace-folder") + 1
    assert args[ws_idx] == path, f"Expected workspace '{path}', got '{args[ws_idx]}'"


# ── Concurrent operation steps ───────────────────────────────


@when("I activate the container concurrently from two threads")
def step_concurrent_activation(context: Context) -> None:
    """Launch two threads both trying to activate the same resource.

    TEST-1 fix: use ``threading.Barrier(2)`` so both threads reach the
    activation call at the same time, preventing the first thread from
    completing before the second starts.
    """
    resource_id = context.resource_id
    results: list[tuple[str, object]] = []
    lock = threading.Lock()
    barrier = threading.Barrier(2, timeout=5.0)

    def _activate(label: str) -> None:
        try:
            barrier.wait()
            activate_container(
                resource_id,
                "/workspace",
                run_command=context.mock_runner,
            )
            with lock:
                results.append((label, "ok"))
        except (ValueError, RuntimeError) as exc:
            with lock:
                results.append((label, exc))

    t1 = threading.Thread(target=_activate, args=("t1",))
    t2 = threading.Thread(target=_activate, args=("t2",))
    t1.start()
    t2.start()
    t1.join(timeout=5.0)
    t2.join(timeout=5.0)

    context.concurrent_results = results


@when("I stop the container concurrently from two threads")
def step_concurrent_stop(context: Context) -> None:
    """Launch two threads both trying to stop the same resource.

    TEST-1 fix: use ``threading.Barrier(2)`` so both threads reach the
    stop call at the same time, preventing sequential execution.
    """
    resource_id = "01TESTLIFECYCLE0000000390"
    results: list[tuple[str, object]] = []
    lock = threading.Lock()
    barrier = threading.Barrier(2, timeout=5.0)

    def _stop(label: str) -> None:
        try:
            barrier.wait()
            stop_container(
                resource_id,
                run_command=context.mock_runner,
            )
            with lock:
                results.append((label, "ok"))
        except (ValueError, RuntimeError) as exc:
            with lock:
                results.append((label, exc))

    t1 = threading.Thread(target=_stop, args=("t1",))
    t2 = threading.Thread(target=_stop, args=("t2",))
    t1.start()
    t2.start()
    t1.join(timeout=5.0)
    t2.join(timeout=5.0)

    context.concurrent_stop_results = results


@then("exactly one activation should succeed")
def step_one_success(context: Context) -> None:
    successes = [r for r in context.concurrent_results if r[1] == "ok"]
    assert len(successes) == 1, (
        f"Expected exactly 1 success, got {len(successes)}: {context.concurrent_results}"
    )


@then("exactly one activation should fail with ValueError")
def step_one_failure(context: Context) -> None:
    failures = [r for r in context.concurrent_results if isinstance(r[1], ValueError)]
    assert len(failures) == 1, (
        f"Expected exactly 1 ValueError, got {len(failures)}: {context.concurrent_results}"
    )


@then("both stops should succeed")
def step_both_stops_succeed(context: Context) -> None:
    # R8-F2 fix: stop_container is now idempotent — the second caller
    # sees a terminal state and returns early instead of raising.
    successes = [r for r in context.concurrent_stop_results if r[1] == "ok"]
    assert len(successes) == 2, (
        f"Expected 2 stop successes (idempotent), got {len(successes)}: "
        f"{context.concurrent_stop_results}"
    )


@then('the container state for "{resource_id}" should not be "{state}"')
def step_check_state_not(context: Context, resource_id: str, state: str) -> None:
    tracker = get_lifecycle_tracker(resource_id)
    assert tracker.current_state != ContainerLifecycleState(state), (
        f"Expected state NOT to be {state}, but got {tracker.current_state.value}"
    )


@then("exactly {count:d} docker stop call should have been made")
def step_check_docker_stop_count(context: Context, count: int) -> None:
    actual = len(context.mock_runner.stop_calls)
    assert actual == count, (
        f"Expected exactly {count} docker stop call(s), got {actual}"
    )


# ── F16: resolve() lazy activation steps ─────────────────────


@when('I resolve a devcontainer-instance "{resource_id}" at "{location}"')
def step_resolve_devcontainer(
    context: Context, resource_id: str, location: str
) -> None:
    """Call DevcontainerHandler.resolve() with mocked Resource and SandboxManager.

    The handler's resolve() triggers activate_container() for
    non-running devcontainer-instance resources (F16 fix), then
    delegates to super().resolve() which needs a SandboxManager.
    We patch activate_container with a side_effect that injects
    the mock runner, so the real activation logic executes.
    """
    handler = DevcontainerHandler()
    mock_resource = MagicMock()
    mock_resource.resource_id = resource_id
    mock_resource.resource_type_name = "devcontainer-instance"
    mock_resource.location = location
    mock_resource.sandbox_strategy = None

    mock_sandbox_mgr = MagicMock()
    mock_sandbox_ctx = MagicMock()
    mock_sandbox_ctx.sandbox_path = "/tmp/sandbox"
    mock_sandbox = MagicMock()
    mock_sandbox.context = mock_sandbox_ctx
    mock_sandbox_mgr.get_or_create_sandbox.return_value = mock_sandbox

    real_activate = _dc_mod.activate_container

    def _activate_with_mock(
        rid: str,
        ws: str,
        session_id: str | None = None,
        **kwargs: object,
    ) -> object:
        return real_activate(
            rid, ws, run_command=context.mock_runner, session_id=session_id
        )

    context.resolve_error = None
    with _patch.object(
        _dc_mod, "activate_container", side_effect=_activate_with_mock
    ) as mock_activate:
        try:
            context.bound_resource = handler.resolve(
                resource=mock_resource,
                plan_id="plan-001",
                slot_name="workspace",
                sandbox_manager=mock_sandbox_mgr,
            )
        except (RuntimeError, ValueError) as exc:
            context.resolve_error = exc
        context.resolve_activate_mock = mock_activate