"""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
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, cast

import pytest

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_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_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(Exception):
            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"
        )
        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"
        )
        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