"""Opt-in server scheduler lifecycle (T12): service behavior over real stores.

The service owns canonical ownership, startup recovery (never executes),
periodic polling that never blocks on long workflows, bounded execution
tasks behind the async-completion seam, and stop-admission-first drain
with truthful abandonment. Execution itself is scripted here through a
stub async runtime; real-workflow and real-server integration arrives in
dedicated integration tests.
"""

from __future__ import annotations

import asyncio
import subprocess
import sys
import threading
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, cast

import pytest
from pydantic import ValidationError

from tests.artifacts.test_run_store import artifact as _artifact
from tests.artifacts.test_run_store import deployment as _deployment
from wf_artifacts.runs.store import FileRunStore
from wf_artifacts.store import FileWorkflowArtifactStore
from wf_core import END, RunState, RunStatus
from wf_scheduling.lifecycle import (
    DrainReport,
    SchedulerService,
    SchedulerServiceConfig,
    SchedulerStartupError,
    StoreDeploymentDirectory,
)
from wf_scheduling.models import Schedule
from wf_scheduling.ownership import SchedulerOwnership, SecondOwnerError
from wf_scheduling.poll import Scheduler
from wf_scheduling.store import FileScheduleStore


def ts(y: int, mo: int, d: int, h: int = 0, mi: int = 0) -> datetime:
    return datetime(y, mo, d, h, mi, tzinfo=UTC)


def _sched_model(sid: str, intended: datetime, **kw: Any) -> Schedule:
    base: dict[str, Any] = {
        "id": sid,
        "deployment_id": "parent.personal",
        "trigger": {"kind": "oneshot", "at": intended.isoformat()},
        "input_bindings": [],
        "created_at": intended.isoformat(),
        "updated_at": intended.isoformat(),
    }
    base.update(kw)
    return Schedule.model_validate(base)


def _stopped(status: RunStatus) -> RunState:
    return RunState(
        workflow_name="sched",
        status=status,
        workflow_input={},
        state={},
    )


class ScriptedRuntime:
    """Stub async runtime with per-test outcomes (never touches stores)."""

    def __init__(self, outcome: Any = "complete") -> None:
        self.outcome = outcome
        self.calls: list[dict[str, Any]] = []
        self.started = asyncio.Event()
        self.release = asyncio.Event()

    async def run_workflow_from_plan(
        self,
        plan: Any,
        workflow_input: dict[str, Any],
        deployment: Any = None,
        artifact: Any = None,
        saved_subgraph_tree: Any = None,
        limits: Any = None,
    ) -> RunState:
        self.calls.append(
            {
                "plan": plan,
                "input": workflow_input,
                "deployment": deployment,
                "artifact": artifact,
                "limits": limits,
            }
        )
        spec = self.outcome
        if callable(spec):
            spec = spec(len(self.calls))
        if spec == "hang":
            self.started.set()
            await self.release.wait()
            return _stopped(RunStatus.COMPLETED)
        if spec == "raise":
            raise RuntimeError("injected execution failure")
        assert isinstance(spec, str), f"unknown outcome spec {spec!r}"
        return _stopped(
            {
                "complete": RunStatus.COMPLETED,
                "fail": RunStatus.FAILED,
                "interrupt": RunStatus.INTERRUPTED,
            }[spec]
        )

    async def resume_workflow_from_plan(self, *args: Any, **kwargs: Any) -> RunState:
        raise AssertionError("scheduler never resumes through the runtime")


def _artifact_store(tmp_path: Path) -> FileWorkflowArtifactStore:
    store = FileWorkflowArtifactStore(tmp_path / "wf")
    store.save_artifact(_artifact().model_copy(update={"plan": _plan_dict()}))
    store.save_deployment(_deployment())
    return store


def _plan_dict() -> dict[str, Any]:
    """Minimal executable plan (constant workflow, proven shape)."""
    return {
        "name": "parent",
        "input_schema": {"type": "object", "properties": {}},
        "state_schema": {
            "type": "object",
            "properties": {"result": {"type": "string"}},
        },
        "output_schema": {
            "type": "object",
            "properties": {"result": {"type": "string"}},
        },
        "outcomes": ["ok"],
        "start": "constant",
        "nodes": [
            {
                "id": "constant",
                "type": "node",
                "node": "wf.std.constant",
                "input": [
                    {
                        "value": "hello from scheduler",
                        "target": {"root": "local", "parts": ["value"]},
                    }
                ],
                "output": [
                    {
                        "source": {"root": "local", "parts": ["value"]},
                        "target": {"root": "state", "parts": ["result"]},
                    }
                ],
            }
        ],
        "edges": [{"from": "constant", "outcome": "ok", "to": END}],
        "output": [
            {
                "path": {"root": "state", "parts": ["result"]},
                "target": {"root": "local", "parts": ["result"]},
            }
        ],
    }


def _service(
    tmp_path: Path,
    runtime: Any,
    config: SchedulerServiceConfig | None = None,
) -> SchedulerService:
    if config is None:
        # Deterministic tests drive ticks explicitly through poll_once;
        # the background loop has its own test below.
        config = SchedulerServiceConfig(poll_interval_s=0.01, auto_tick=False)
    return SchedulerService(
        schedule_store=FileScheduleStore(tmp_path),
        run_store=FileRunStore(tmp_path),
        runtime=runtime,
        artifact_store=_artifact_store(tmp_path),
        ownership=SchedulerOwnership(tmp_path, owner="test"),
        config=config,
    )


async def _wait_for(cond: Any, timeout: float = 10.0) -> None:
    async with asyncio.timeout(timeout):
        while not cond():
            await asyncio.sleep(0.01)


def _entries(store: FileScheduleStore, sid: str, kind: str) -> list[dict[str, Any]]:
    page = store.list_occurrences(sid, limit=100)
    rows = cast(list[dict[str, Any]], page["occurrences"])
    return [r for r in rows if r["kind"] == kind]


def _only_run_id(run_store: Any) -> str:
    runs = list(run_store.list_runs())
    assert len(runs) == 1
    return runs[0].id


async def test_scheduled_completion_through_lifecycle(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(tmp_path, ScriptedRuntime("complete"))
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        result = await service.poll_once(intended + timedelta(seconds=1))
        run_id = _only_run_id(service.run_store)
        assert result == {"a": f"admit:{run_id}"}
        await _wait_for(lambda: service.live_executions == 0)
        record = service.run_store.get_run(run_id)
        assert record.status.value == "completed"
        assert len(_entries(service.schedule_store, "a", "completed")) == 1
        # The dispatcher ran the pinned deployment input, not a fixture.
        assert len(service.runtime.calls) == 1
        assert service.runtime.calls[0]["input"] == {}
        assert service.runtime.calls[0]["deployment"].id == "parent.personal"
        assert service.runtime.calls[0]["limits"].max_steps == 10_000
    finally:
        report = await service.stop()
    assert report.settled == 1
    assert report.cancelled == 0


async def test_scheduled_failure_records_failed_history(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(tmp_path, ScriptedRuntime("fail"))
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        result = await service.poll_once(intended + timedelta(seconds=1))
        run_id = _only_run_id(service.run_store)
        assert result == {"a": f"admit:{run_id}"}
        await _wait_for(lambda: service.live_executions == 0)
        assert service.run_store.get_run(run_id).status.value == "failed"
        assert len(_entries(service.schedule_store, "a", "failed")) == 1
    finally:
        await service.stop()


async def test_settlement_does_not_block_the_event_loop(
    tmp_path: Path, monkeypatch: Any
) -> None:
    """Synchronous settlement persistence must not stall other coroutines."""
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(tmp_path, ScriptedRuntime("complete"))
    settlement_started = threading.Event()
    settlement_finished = threading.Event()
    release = threading.Event()
    original = Scheduler.record_stopped_execution

    def _blocking_settle(self: Any, run_id: str, state: Any, now: datetime) -> None:
        settlement_started.set()
        release.wait(timeout=2)
        original(self, run_id, state, now)
        settlement_finished.set()

    monkeypatch.setattr(Scheduler, "record_stopped_execution", _blocking_settle)
    release_timer = threading.Timer(1.0, release.set)
    release_timer.start()
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        await service.poll_once(intended + timedelta(seconds=1))
        await asyncio.sleep(0.05)
        assert settlement_started.is_set()
        assert not settlement_finished.is_set()
        assert service.live_executions == 1
        release.set()
        await _wait_for(lambda: service.live_executions == 0)
    finally:
        release.set()
        release_timer.cancel()
        await service.stop()


async def test_shutdown_joins_cancelled_settlement_before_releasing_ownership(
    tmp_path: Path, monkeypatch: Any
) -> None:
    """Shutdown waits for an offloaded settlement after cancelling its task."""
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(
        tmp_path,
        ScriptedRuntime("complete"),
        SchedulerServiceConfig(
            poll_interval_s=0.01,
            drain_grace_s=0,
            auto_tick=False,
        ),
    )
    settlement_started = threading.Event()
    release = threading.Event()
    original = Scheduler.record_stopped_execution

    def _blocking_settle(self: Any, run_id: str, state: Any, now: datetime) -> None:
        settlement_started.set()
        release.wait(timeout=2)
        original(self, run_id, state, now)

    monkeypatch.setattr(Scheduler, "record_stopped_execution", _blocking_settle)
    release_timer = threading.Timer(1.0, release.set)
    release_timer.start()
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        await service.poll_once(intended + timedelta(seconds=1))
        for _ in range(50):
            if settlement_started.is_set():
                break
            await asyncio.sleep(0.01)
        else:
            pytest.fail("settlement did not start")

        stop_task = asyncio.create_task(service.stop())
        await asyncio.sleep(0.05)
        assert not stop_task.done()
        release.set()
        report = await stop_task
        assert report.settled == 1
        assert report.cancelled == 0
        assert service.run_store.get_run(
            _only_run_id(service.run_store)
        ).status.value == ("completed")
    finally:
        release.set()
        release_timer.cancel()
        await service.stop()


def test_submit_and_stop_do_not_deadlock_on_registration(tmp_path: Path) -> None:
    """The poll worker and event loop must not wait on each other."""
    child = """
import asyncio
import sys
import threading
import time
from pathlib import Path

from tests.scheduling.test_lifecycle import (
    SchedulerServiceConfig,
    ScriptedRuntime,
    _sched_model,
    _service,
    ts,
)
import wf_scheduling.lifecycle as lifecycle


async def main() -> None:
    root = Path(sys.argv[1])
    service = _service(
        root,
        ScriptedRuntime("complete"),
        config=SchedulerServiceConfig(poll_interval_s=0.01, auto_tick=False),
    )
    await service.start()
    service.schedule_store.create_schedule(
        _sched_model("a", ts(2026, 9, 8, 12, 0))
    )
    registration_entered = threading.Event()
    release_registration = threading.Event()
    original = asyncio.run_coroutine_threadsafe

    def blocked_registration(coro, loop):
        registration_entered.set()
        if not release_registration.wait(5):
            raise AssertionError("registration release was not signalled")
        return original(coro, loop)

    lifecycle.asyncio.run_coroutine_threadsafe = blocked_registration
    poll_task = asyncio.create_task(
        service.poll_once(ts(2026, 9, 8, 12, 1))
    )
    if not await asyncio.to_thread(registration_entered.wait, 5):
        raise AssertionError("poll did not reach the registration gate")
    stop_task = asyncio.create_task(service.stop())
    threading.Thread(
        target=lambda: (time.sleep(0.1), release_registration.set()),
        daemon=True,
    ).start()
    await asyncio.wait_for(asyncio.gather(poll_task, stop_task), 4)
    print("registration-stop-complete", flush=True)


asyncio.run(main())
"""
    process = subprocess.Popen(
        [sys.executable, "-c", child, str(tmp_path)],
        cwd=Path(__file__).resolve().parents[2],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
    )
    try:
        output, _ = process.communicate(timeout=7)
    except subprocess.TimeoutExpired:
        process.kill()
        output, _ = process.communicate(timeout=5)
        pytest.fail(f"registration/stop deadlocked; child output: {output}")
    assert process.returncode == 0, output
    assert "registration-stop-complete" in output


async def test_scheduled_interrupt_stays_resumable(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(tmp_path, ScriptedRuntime("interrupt"))
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        await service.poll_once(intended + timedelta(seconds=1))
        run_id = _only_run_id(service.run_store)
        await _wait_for(lambda: service.live_executions == 0)
        record = service.run_store.get_run(run_id)
        assert record.status.value == "interrupted"
        assert record.resume_readiness.value == "ready"
        assert len(_entries(service.schedule_store, "a", "interrupted")) == 1
    finally:
        await service.stop()


async def test_sources_refresh_after_start(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(tmp_path, ScriptedRuntime("complete"))
    try:
        await service.start()
        assert await service.poll_once(intended) == {}
        service.schedule_store.create_schedule(_sched_model("a", intended))
        result = await service.poll_once(intended + timedelta(seconds=1))
        assert result["a"].startswith("admit:run-")
        await _wait_for(lambda: service.live_executions == 0)
    finally:
        await service.stop()


async def test_hang_then_late_settlement_frees_capacity(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    runtime = ScriptedRuntime("hang")
    service = _service(
        tmp_path,
        runtime,
        SchedulerServiceConfig(poll_interval_s=0.01, capacity=1, auto_tick=False),
    )
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        service.schedule_store.create_schedule(_sched_model("b", intended))
        first = await service.poll_once(intended + timedelta(seconds=1))
        assert first["a"].startswith("admit:run-")
        await _wait_for(lambda: runtime.started.is_set())
        run_a = _only_run_id(service.run_store)
        assert service.run_store.get_run(run_a).status.value == "admitted"
        assert service.live_executions == 1
        # Capacity is saturated: no second admission while the first hangs.
        await service.poll_once(intended + timedelta(seconds=2))
        assert [r.id for r in list(service.run_store.list_runs())] == [run_a]
        runtime.release.set()
        await _wait_for(lambda: service.live_executions == 0)
        assert service.run_store.get_run(run_a).status.value == "completed"
        # The freed slot admits the waiting schedule on the next tick.
        later = await service.poll_once(intended + timedelta(seconds=3))
        assert later["b"].startswith("admit:run-")
    finally:
        runtime.release.set()
        await service.stop()


async def test_long_run_does_not_block_calendar_polling(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    calls = {"n": 0}

    def outcome(n: int) -> str:
        calls["n"] = n
        return "hang" if n == 1 else "complete"

    runtime = ScriptedRuntime(outcome)
    service = _service(
        tmp_path,
        runtime,
        SchedulerServiceConfig(poll_interval_s=0.01, capacity=2, auto_tick=False),
    )
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        await service.poll_once(intended + timedelta(seconds=1))
        await _wait_for(lambda: runtime.started.is_set())
        run_a = _only_run_id(service.run_store)
        # A second schedule admitted while the first still executes.
        service.schedule_store.create_schedule(_sched_model("b", intended))
        result = await service.poll_once(intended + timedelta(seconds=2))
        assert result["b"].startswith("admit:run-")
        await _wait_for(lambda: service.live_executions == 1)
        run_b = result["b"].split(":", 1)[1]
        assert service.run_store.get_run(run_b).status.value == "completed"
        assert service.run_store.get_run(run_a).status.value == "admitted"
        assert service.live_executions <= 2
    finally:
        runtime.release.set()
        await service.stop()


async def test_graceful_shutdown_cancels_hanging_execution(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    runtime = ScriptedRuntime("hang")
    service = _service(
        tmp_path,
        runtime,
        SchedulerServiceConfig(
            poll_interval_s=0.01, drain_grace_s=0.05, auto_tick=False
        ),
    )
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        await service.poll_once(intended + timedelta(seconds=1))
        await _wait_for(lambda: runtime.started.is_set())
        run_id = _only_run_id(service.run_store)
        report = await service.stop()
        assert report.cancelled == 1
        assert report.settled == 0
        # Truthful handling: the run keeps its executing mark, settled by
        # restart recovery instead of a fabricated outcome.
        record = FileRunStore(tmp_path).get_run(run_id)
        assert record.status.value == "admitted"
        assert FileRunStore(tmp_path).is_executing(run_id) is True
        assert service.live_executions == 0
    finally:
        runtime.release.set()
        await service.stop()


async def test_restart_after_shutdown_abandons_without_replay(
    tmp_path: Path,
) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    runtime = ScriptedRuntime("hang")
    first = _service(
        tmp_path,
        runtime,
        SchedulerServiceConfig(
            poll_interval_s=0.01, drain_grace_s=0.05, auto_tick=False
        ),
    )
    try:
        await first.start()
        first.schedule_store.create_schedule(_sched_model("a", intended))
        await first.poll_once(intended + timedelta(seconds=1))
        await _wait_for(lambda: runtime.started.is_set())
        run_id = _only_run_id(first.run_store)
        await first.stop()
    finally:
        runtime.release.set()
        await first.stop()
    assert len(runtime.calls) == 1
    # Fresh instances across the restart, like a real process boundary.
    second = _service(tmp_path, ScriptedRuntime("complete"))
    try:
        await second.start()
        record = second.run_store.get_run(run_id)
        assert record.status.value == "failed"
        assert record.resume_readiness.value == "not_applicable"
        assert len(_entries(second.schedule_store, "a", "failed")) == 1
        assert len(second.runtime.calls) == 0
    finally:
        await second.stop()


async def test_dispatcher_error_abandons_without_replay(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(tmp_path, ScriptedRuntime("raise"))
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        await service.poll_once(intended + timedelta(seconds=1))
        run_id = _only_run_id(service.run_store)
        await _wait_for(lambda: service.live_executions == 0)
        record = service.run_store.get_run(run_id)
        assert record.status.value == "failed"
        assert record.resume_readiness.value == "not_applicable"
        assert len(_entries(service.schedule_store, "a", "failed")) == 1
        assert any(run_id in message for message in service.errors)
        # The slot is freed: a later schedule still admits.
        service.schedule_store.create_schedule(_sched_model("b", intended))
        later = await service.poll_once(intended + timedelta(seconds=2))
        assert later["b"].startswith("admit:run-")
    finally:
        await service.stop()


async def test_failed_startup_releases_lock(tmp_path: Path) -> None:
    squatter = SchedulerOwnership(tmp_path, owner="squatter").acquire()
    try:
        service = _service(tmp_path, ScriptedRuntime("complete"))
        with pytest.raises(SchedulerStartupError):
            await service.start()
        assert service.running is False
    finally:
        squatter.release()
    service = _service(tmp_path, ScriptedRuntime("complete"))
    try:
        await service.start()
        assert service.running is True
    finally:
        await service.stop()


async def test_cancelled_startup_joins_gated_recovery_before_releasing_ownership(
    tmp_path: Path, monkeypatch: Any
) -> None:
    """Cancelled startup waits for recovery before releasing its store lock."""
    service = _service(tmp_path, ScriptedRuntime("complete"))
    recovery_started = threading.Event()
    recovery_release = threading.Event()
    recovery_ready = threading.Event()
    recovery_continue = threading.Event()
    recovery_finished = threading.Event()
    original = SchedulerService._recover

    def gated_recover(self: SchedulerService) -> list[str]:
        recovery_started.set()
        if not recovery_release.wait(5):
            raise AssertionError("recovery release was not signalled")
        recovery_ready.set()
        if not recovery_continue.wait(5):
            raise AssertionError("recovery continuation was not signalled")
        result = original(self)
        recovery_finished.set()
        return result

    monkeypatch.setattr(SchedulerService, "_recover", gated_recover)
    start_task = asyncio.create_task(service.start())
    try:
        assert await asyncio.to_thread(recovery_started.wait, 5)
        start_task.cancel()
        recovery_release.set()
        assert await asyncio.to_thread(recovery_ready.wait, 5)
        # The cancelled start is still joining the active recovery worker.
        assert not start_task.done()

        retry = _service(tmp_path, ScriptedRuntime("complete"))
        with pytest.raises(SecondOwnerError):
            retry.ownership.acquire()

        recovery_continue.set()
        with pytest.raises(asyncio.CancelledError):
            await asyncio.wait_for(start_task, 5)
        assert recovery_finished.is_set()
        assert service.running is False
        assert service.live_executions == 0
        assert service._tick_task is None

        await retry.start()
        await retry.stop()
    finally:
        recovery_release.set()
        recovery_continue.set()
        if not start_task.done():
            start_task.cancel()
            await asyncio.gather(start_task, return_exceptions=True)
        await service.stop()


async def test_corrupt_store_startup_releases_lock(tmp_path: Path) -> None:
    class BrokenRuns(FileRunStore):
        def list_admissions(self) -> list[Any]:
            raise OSError("injected store failure")

    service = SchedulerService(
        schedule_store=FileScheduleStore(tmp_path),
        run_store=BrokenRuns(tmp_path),
        runtime=ScriptedRuntime("complete"),
        artifact_store=_artifact_store(tmp_path),
        ownership=SchedulerOwnership(tmp_path, owner="test"),
        config=SchedulerServiceConfig(poll_interval_s=0.01, auto_tick=False),
    )
    with pytest.raises(SchedulerStartupError):
        await service.start()
    # The lock is free for a retry after the operator fixes the store.
    retry = SchedulerOwnership(tmp_path, owner="retry").acquire()
    retry.release()


async def test_double_start_rejected_and_stop_idempotent(
    tmp_path: Path,
) -> None:
    assert await _service(tmp_path, ScriptedRuntime()).stop() == DrainReport()
    service = _service(tmp_path, ScriptedRuntime("complete"))
    try:
        await service.start()
        with pytest.raises(SchedulerStartupError):
            await service.start()
    finally:
        await service.stop()
    # Ownership is released only after the drain finishes.
    freed = SchedulerOwnership(tmp_path, owner="freed").acquire()
    freed.release()


async def test_second_owner_rejected_while_running(tmp_path: Path) -> None:
    service = _service(tmp_path, ScriptedRuntime("complete"))
    try:
        await service.start()
        with pytest.raises(SecondOwnerError):
            SchedulerOwnership(tmp_path, owner="intruder").acquire()
    finally:
        await service.stop()


async def test_tick_error_recorded_and_loop_continues(tmp_path: Path) -> None:
    intended = ts(2026, 9, 8, 12, 0)
    service = _service(tmp_path, ScriptedRuntime("complete"))
    try:
        await service.start()
        service.schedule_store.create_schedule(_sched_model("a", intended))
        service.schedule_store.create_schedule(_sched_model("broken", intended))
        # Corrupt one schedule file behind the model's back: the listing
        # fails loudly instead of ticking a half-blind schedule set.
        broken_path = tmp_path / "schedules" / "broken" / "schedule.json"
        good_text = broken_path.read_text(encoding="utf-8")
        broken_path.write_text('{"id": "broken", "trigger": {"kind": "bogus"}}')
        with pytest.raises(ValidationError):
            await service.poll_once(intended + timedelta(seconds=1))
        assert service.last_tick_error is not None
        assert service.run_store.list_runs() == []
        # Repairing the file resumes normal ticking on the next call.
        broken_path.write_text(good_text, encoding="utf-8")
        result = await service.poll_once(intended + timedelta(seconds=2))
        assert result["a"].startswith("admit:run-")
        await _wait_for(lambda: service.live_executions == 0)
    finally:
        await service.stop()


async def test_poll_once_before_start_rejected(tmp_path: Path) -> None:
    service = _service(tmp_path, ScriptedRuntime("complete"))
    with pytest.raises(SchedulerStartupError):
        await service.poll_once(ts(2026, 9, 8, 12, 0))
    await service.stop()


async def test_background_loop_ticks_and_stops(tmp_path: Path) -> None:
    from datetime import timezone

    service = _service(
        tmp_path,
        ScriptedRuntime("complete"),
        SchedulerServiceConfig(poll_interval_s=0.01, auto_tick=True),
    )
    try:
        await service.start()
        # A one-shot due in real time: the background loop admits it
        # without any explicit poll_once call.
        due = datetime.now(timezone.utc) + timedelta(seconds=0.2)
        service.schedule_store.create_schedule(_sched_model("a", due))

        def _status() -> str | None:
            runs = list(service.run_store.list_runs())
            return runs[0].status.value if runs else None

        # Status moves monotonically admitted -> completed: waiting on it
        # cannot miss a fast execution the way a live-task edge can.
        await _wait_for(lambda: _status() == "completed")
        run_id = _only_run_id(service.run_store)
        assert service.run_store.get_run(run_id).status.value == "completed"
    finally:
        report = await service.stop()
    assert report.settled == 1
    assert service.tick_count >= 2


async def test_store_deployment_directory_contract(tmp_path: Path) -> None:
    store = _artifact_store(tmp_path)
    directory = StoreDeploymentDirectory(store)
    assert directory.deployment_revision("parent.personal") >= 1
    assert directory.required_inputs("parent.personal") == []
    with pytest.raises(KeyError):
        directory.deployment_revision("missing.deployment")
    with pytest.raises(KeyError):
        directory.required_inputs("missing.deployment")


async def _poll_two(service: SchedulerService, intended: datetime) -> tuple[str, str]:
    service.schedule_store.create_schedule(_sched_model("a", intended))
    service.schedule_store.create_schedule(_sched_model("b", intended))
    result = await service.poll_once(intended + timedelta(seconds=1))
    hang_id = result["a"].removeprefix("admit:")
    sib_id = result["b"].removeprefix("admit:")
    assert hang_id.startswith("run-") and sib_id.startswith("run-")
    return hang_id, sib_id


def _assert_sibling_live(service: SchedulerService, hang_id: str) -> None:
    """The healthy sibling keeps its markers, status, and clean history."""
    sibling = service.run_store.get_run(hang_id)
    assert sibling.status.value == "admitted"
    assert service.run_store.is_executing(hang_id)
    assert not service.run_store.is_pending_dispatch(hang_id)
    assert len(_entries(service.schedule_store, "a", "failed")) == 0


async def test_live_runtime_failure_spares_healthy_sibling(
    tmp_path: Path,
) -> None:
    """A live execution failure repairs only the broken run (B1).

    Whole-store recovery during live execution used to fail the still
    hanging sibling too, whose late success was then rejected as
    ``settle non-admitted run``.
    """
    intended = ts(2026, 9, 8, 12, 0)
    runtime = ScriptedRuntime(lambda n: "hang" if n == 1 else "raise")
    service = _service(
        tmp_path,
        runtime,
        SchedulerServiceConfig(poll_interval_s=0.01, capacity=2, auto_tick=False),
    )
    try:
        await service.start()
        hang_id, sib_id = await _poll_two(service, intended)
        await _wait_for(lambda: runtime.started.is_set())
        await _wait_for(
            lambda: (
                service.run_store.get_run(sib_id).status.value == "failed"
                and not service.run_store.is_executing(sib_id)
            )
        )
        failed = service.run_store.get_run(sib_id)
        assert failed.status.value == "failed"
        assert not service.run_store.is_executing(sib_id)
        assert not service.run_store.is_pending_dispatch(sib_id)
        assert len(_entries(service.schedule_store, "b", "failed")) == 1
        _assert_sibling_live(service, hang_id)
        runtime.release.set()
        await _wait_for(lambda: service.live_executions == 0)
        assert service.run_store.get_run(hang_id).status.value == "completed"
        assert len(_entries(service.schedule_store, "a", "completed")) == 1
        assert not any("settle non-admitted run" in m for m in service.errors)
    finally:
        runtime.release.set()
        report = await service.stop()
    assert report.settled == 1
    assert report.abandoned == 1


async def test_live_settlement_failure_spares_healthy_sibling(
    tmp_path: Path, monkeypatch: Any
) -> None:
    """A torn settle write repairs only the broken run (B1).

    The settlement seam raises after genuine execution for one run while
    its sibling hangs: only that run fails closed, and the sibling still
    settles exactly once with its capacity slot freed.
    """
    intended = ts(2026, 9, 8, 12, 0)
    runtime = ScriptedRuntime(lambda n: "hang" if n == 1 else "complete")
    service = _service(
        tmp_path,
        runtime,
        SchedulerServiceConfig(poll_interval_s=0.01, capacity=2, auto_tick=False),
    )
    try:
        await service.start()
        original = Scheduler.record_stopped_execution
        fired = {"done": False}

        def _flaky_settle(self: Any, run_id: str, state: Any, now: datetime) -> None:
            try:
                owned_by_b = self.run_store.get_admission(run_id).schedule_id == "b"
            except KeyError:
                owned_by_b = False
            if owned_by_b and not fired["done"]:
                fired["done"] = True
                raise RuntimeError("injected settlement failure")
            original(self, run_id, state, now)

        monkeypatch.setattr(Scheduler, "record_stopped_execution", _flaky_settle)
        hang_id, sib_id = await _poll_two(service, intended)
        await _wait_for(lambda: runtime.started.is_set())
        await _wait_for(
            lambda: (
                service.run_store.get_run(sib_id).status.value == "failed"
                and not service.run_store.is_executing(sib_id)
            )
        )
        assert fired["done"]
        assert not service.run_store.is_executing(sib_id)
        assert len(_entries(service.schedule_store, "b", "failed")) == 1
        _assert_sibling_live(service, hang_id)
        runtime.release.set()
        await _wait_for(lambda: service.live_executions == 0)
        assert service.run_store.get_run(hang_id).status.value == "completed"
        assert len(_entries(service.schedule_store, "a", "completed")) == 1
        assert not any("settle non-admitted run" in m for m in service.errors)
    finally:
        runtime.release.set()
        report = await service.stop()
    assert report.settled == 1
    assert report.abandoned == 1


async def test_live_torn_settlement_reconciles_genuine_result(
    tmp_path: Path, monkeypatch: Any
) -> None:
    """A torn history write keeps the genuine stopped result (B1).

    The stopped result is durably persisted but the history append is
    lost: scoped repair must reconcile the genuine completion (with
    checkpoint authority), never fail the run, and never touch the
    hanging sibling.
    """
    intended = ts(2026, 9, 8, 12, 0)
    runtime = ScriptedRuntime(lambda n: "hang" if n == 1 else "complete")
    service = _service(
        tmp_path,
        runtime,
        SchedulerServiceConfig(poll_interval_s=0.01, capacity=2, auto_tick=False),
    )
    try:
        await service.start()
        original = Scheduler._record
        fired = {"done": False}

        def _flaky_record(self: Any, **kwargs: Any) -> None:
            run_id = kwargs.get("run_id")
            try:
                owned_by_b = (
                    run_id is not None
                    and self.run_store.get_admission(run_id).schedule_id == "b"
                )
            except KeyError:
                owned_by_b = False
            if kwargs.get("kind") == "completed" and owned_by_b and not fired["done"]:
                fired["done"] = True
                raise RuntimeError("injected history failure")
            original(self, **kwargs)

        monkeypatch.setattr(Scheduler, "_record", _flaky_record)
        hang_id, sib_id = await _poll_two(service, intended)
        await _wait_for(lambda: runtime.started.is_set())
        await _wait_for(
            lambda: len(_entries(service.schedule_store, "b", "completed")) == 1
        )
        assert fired["done"]
        repaired = service.run_store.get_run(sib_id)
        assert repaired.status.value == "completed"
        assert not service.run_store.is_executing(sib_id)
        assert len(_entries(service.schedule_store, "b", "failed")) == 0
        _assert_sibling_live(service, hang_id)
        runtime.release.set()
        await _wait_for(lambda: service.live_executions == 0)
        assert service.run_store.get_run(hang_id).status.value == "completed"
        assert len(_entries(service.schedule_store, "a", "completed")) == 1
    finally:
        runtime.release.set()
        report = await service.stop()
    assert report.settled == 2
    assert report.abandoned == 0