sched: record resumed stopped results live and idempotently (B3)

src/wf_api/runs.py

@@ -320,6 +320,18 @@ class WorkflowRunApi:
                 updated_at=cleared_at,
             )
         )
+        # Live occurrence history for scheduled resumes: record the resumed
+        # stopped result now (completion, failure, or re-interruption each
+        # carry their own checkpoint id, so each records exactly once and
+        # repeats dedup). Manual runs and scheduler-off resumes skip
+        # quietly — restart recovery reconciles those instead.
+        gate = self.resume_slot_gate
+        if gate is not None:
+            gate.note_resumed_result(
+                run_id,
+                status_value=run.status.value,
+                checkpoint_id=next_record.latest_checkpoint_id,
+            )
         return _run_payload(
             deployment=environment.deployment,
             artifact=environment.root_artifact,

src/wf_scheduling/poll.py

@@ -504,6 +504,54 @@ class Scheduler:
             checkpoint_id=stopped.latest_checkpoint_id,
         )
 
+    def record_resumed_stopped_result(
+        self,
+        run_id: str,
+        *,
+        status_value: str,
+        checkpoint_id: str | None,
+        now: datetime,
+    ) -> bool:
+        """Record one live resumed stopped result, idempotently.
+
+        The run API persists the resumed stopped checkpoint itself; this
+        only appends the matching occurrence-history entry through the
+        same ``(run_id, kind, checkpoint_id)`` idempotency as dispatch
+        and recovery, so repeated polls and restart recovery never
+        duplicate it. Schedule flags are deliberately not consulted:
+        pausing or deleting a schedule never suppresses retained run
+        history. Returns whether an entry was appended.
+        """
+        self._require_ownership()
+        kind = {
+            "completed": "completed",
+            "interrupted": "interrupted",
+            "failed": "failed",
+        }.get(status_value)
+        if kind is None:
+            return False
+        try:
+            admission = self.run_store.get_admission(run_id)
+        except KeyError:
+            return False
+        sched_id = admission.schedule_id
+        if sched_id is None:
+            return False
+        if self.history.has_terminal(sched_id, run_id, kind, checkpoint_id):
+            return False
+        self._record(
+            kind=kind,  # type: ignore[arg-type]
+            sched_id=sched_id,
+            intended=admission.scheduled_at,
+            run_id=run_id,
+            revision=admission.schedule_revision,
+            reason="resumed",
+            now=now,
+            started_at=now,
+            checkpoint_id=checkpoint_id,
+        )
+        return True
+
     # -- polling --------------------------------------------------------
     def poll(self, now: datetime) -> dict[str, str]:
         self._require_ownership()

src/wf_scheduling/resume_gate.py

@@ -20,6 +20,8 @@ from __future__ import annotations
 
 from typing import Any
 
+from wf_scheduling.ownership import SecondOwnerError
+
 
 class ScheduledCapacityBusyError(ValueError):
     """A scheduled resume found every server execution slot occupied."""
@@ -42,15 +44,15 @@ class SchedulerResumeGate:
 
         Returns the schedule admission when the slot is held (the caller
         must :meth:`release` it), or ``None`` when this run is genuinely
-        manual or the scheduler is not live — both keep the legacy
+        manual, not resumable, or the scheduler is not live — all keep
-        resume path unchanged. Raises
+        the legacy resume path unchanged. Raises
         :class:`ScheduledCapacityBusyError` when every slot is occupied;
         nothing is marked in that case, in particular no ACTIVE resume
         attempt.
         """
         service = self._service
         with service._lock:
-            if not service.running:
+            if not service._started:
                 return None
             try:
                 admission = service.run_store.get_admission(run_id)
@@ -89,3 +91,38 @@ class SchedulerResumeGate:
         with service._lock:
             service.run_store.clear_executing(run_id)
             service._live_resumes.discard(run_id)
+
+    def note_resumed_result(
+        self,
+        run_id: str,
+        *,
+        status_value: str,
+        checkpoint_id: str | None,
+    ) -> bool:
+        """Record one live resumed stopped result for a scheduled run.
+
+        No-op (``False``) for manual runs or when the scheduler is not
+        live — restart recovery still reconciles those through its own
+        idempotent path. Live scheduled resumes append exactly once
+        through the scheduler's recording, sharing idempotency with
+        dispatch and recovery. A lost-ownership race (shutdown releasing
+        mid-note) also skips quietly: the resume result itself is already
+        durable, and recovery reconciles the entry at the next start —
+        the note must never break a completed resume.
+        """
+        service = self._service
+        with service._lock:
+            if not service._started:
+                return False
+            scheduler = service._scheduler
+            if scheduler is None:
+                return False
+            try:
+                return scheduler.record_resumed_stopped_result(
+                    run_id,
+                    status_value=status_value,
+                    checkpoint_id=checkpoint_id,
+                    now=service.clock(),
+                )
+            except (SecondOwnerError, OSError, ValueError):
+                return False

tests/scheduling/test_poll.py

@@ -384,3 +384,89 @@ def test_manual_and_other_schedule_runs_are_overlap_independent(
     assert result["a"].startswith("admit:run-")
     assert result["b"] == "exhausted"
     sched.ownership.release()
+
+
+def test_record_resumed_stopped_result_is_idempotent_and_attributed(
+    tmp_path: Path,
+) -> None:
+    """Live resumed results record exactly once with schedule attribution (B3)."""
+    from wf_api.run_lifecycle import persist_admission
+
+    from tests.scheduling.controlled import fixture_environment
+
+    sched, store, runs, sources = _harness(tmp_path, script={"*": "interrupt"})
+    t0 = ts(2026, 9, 8, 12, 0)
+    _add(sched, store, sources, "a", OneShotSource(t0), t0 - timedelta(seconds=1))
+    assert sched.poll(t0)["a"].startswith("admit:run-")
+    [run] = runs.list_runs()
+    assert run.status.value == "interrupted"
+    assert len([r for r in _history(store, "a") if r["kind"] == "interrupted"]) == 1
+
+    now = t0 + timedelta(minutes=1)
+    resumed_ckpt = f"{run.id}.000002"
+    assert (
+        sched.record_resumed_stopped_result(
+            run.id,
+            status_value="completed",
+            checkpoint_id=resumed_ckpt,
+            now=now,
+        )
+        is True
+    )
+    # Repeats (poll retries, restart recovery) never duplicate the entry.
+    assert (
+        sched.record_resumed_stopped_result(
+            run.id,
+            status_value="completed",
+            checkpoint_id=resumed_ckpt,
+            now=now,
+        )
+        is False
+    )
+    completed = [r for r in _history(store, "a") if r["kind"] == "completed"]
+    assert len(completed) == 1
+    assert completed[0]["run_id"] == run.id
+    assert completed[0]["checkpoint_id"] == resumed_ckpt
+    assert completed[0]["revision"] == 1
+    assert completed[0]["reason"] == "resumed"
+    # A resumed re-interruption is a new result (new checkpoint), not a dup.
+    assert (
+        sched.record_resumed_stopped_result(
+            run.id,
+            status_value="interrupted",
+            checkpoint_id=f"{run.id}.000003",
+            now=now,
+        )
+        is True
+    )
+    assert len([r for r in _history(store, "a") if r["kind"] == "interrupted"]) == 2
+    # Unknown runs, manual runs, and bad statuses record nothing.
+    assert (
+        sched.record_resumed_stopped_result(
+            "run-999999",
+            status_value="completed",
+            checkpoint_id=None,
+            now=now,
+        )
+        is False
+    )
+    assert (
+        sched.record_resumed_stopped_result(
+            run.id, status_value="bogus", checkpoint_id=None, now=now
+        )
+        is False
+    )
+    manual = persist_admission(
+        store=runs,
+        run_id=runs.allocate_run_id(),
+        environment=fixture_environment(object()),
+        resolved_input={},
+        max_steps=None,
+    )
+    assert (
+        sched.record_resumed_stopped_result(
+            manual.id, status_value="completed", checkpoint_id=None, now=now
+        )
+        is False
+    )
+    sched.ownership.release()

tests/wf_server/test_scheduler_integration.py

@@ -1260,3 +1260,174 @@ async def test_service_stop_drains_inflight_scheduled_resume(tmp_path: Path) ->
     finally:
         _gate_open.set()
         await service.stop()
+
+
+async def _kinds(root: Path, schedule_id: str) -> list[str]:
+    page = FileScheduleStore(root).list_occurrences(schedule_id, limit=100)
+    rows = cast(list[dict[str, Any]], page["occurrences"])
+    return [row["kind"] for row in rows]
+
+
+async def test_scheduled_resume_records_completion_live(tmp_path: Path) -> None:
+    """A resumed completion lands in occurrence history live (B3).
+
+    No restart recovery is involved: after the API resume plus ordinary
+    polls, history reads admitted/interrupted/completed, and repeated
+    polling plus a restart never duplicate the entries.
+    """
+    root = tmp_path / "store"
+    server = build_local_static_workflow_server(root)
+    await _seed(server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"])
+    store = FileScheduleStore(root)
+    store.create_schedule(_one_shot("asker", "ask1.default", datetime.now(UTC)))
+    service = _scheduler(server, capacity=2)
+    try:
+        await service.start()
+        await _wait_for(lambda: len(_run_ids(root)) == 1)
+        ask_id = _run_ids(root)[0]
+        await _wait_for(
+            lambda: FileRunStore(root).get_run(ask_id).status.value == "interrupted"
+        )
+        resumed = await server.api.resume_run(
+            run_id=ask_id, resume_payload={}, resume_outcome="submitted"
+        )
+        assert resumed["status"] == "completed"
+        assert await _kinds(root, "asker") == ["admitted", "interrupted", "completed"]
+        completed = _entries(root, "asker", "completed")
+        assert len(completed) == 1
+        assert completed[0]["run_id"] == ask_id
+        assert (
+            completed[0]["checkpoint_id"]
+            == FileRunStore(root).get_run(ask_id).latest_checkpoint_id
+        )
+        assert completed[0]["revision"] == 1
+        await service.poll_once(datetime.now(UTC))
+        await service.poll_once(datetime.now(UTC))
+        assert await _kinds(root, "asker") == ["admitted", "interrupted", "completed"]
+    finally:
+        await service.stop()
+
+    server_b = build_local_static_workflow_server(root)
+    revived = _scheduler(server_b)
+    try:
+        await revived.start()
+        await revived.poll_once(datetime.now(UTC))
+        assert await _kinds(root, "asker") == ["admitted", "interrupted", "completed"]
+    finally:
+        await revived.stop()
+
+
+async def test_scheduled_resume_records_failure_and_reinterruption_live(
+    tmp_path: Path,
+) -> None:
+    """Resumed failures and re-interruptions each record live (B3)."""
+    root = tmp_path / "store"
+    server = build_local_static_workflow_server(root)
+    await _seed(
+        server,
+        "ask_twice",
+        "ask_twice.default",
+        _double_interrupt_plan("ask_twice"),
+        ["submitted"],
+    )
+    await _seed(
+        server,
+        "ask_fail",
+        "ask_fail.default",
+        _interrupt_then_fail_plan("ask_fail"),
+        ["submitted"],
+    )
+    store = FileScheduleStore(root)
+    store.create_schedule(_one_shot("twicer", "ask_twice.default", datetime.now(UTC)))
+    store.create_schedule(_one_shot("failer", "ask_fail.default", datetime.now(UTC)))
+    service = _scheduler(server, capacity=2)
+    try:
+        await service.start()
+        await _wait_for(lambda: len(_run_ids(root)) == 2)
+        twice_id = fail_id = ""
+        for rid in _run_ids(root):
+            admission = FileRunStore(root).get_admission(rid)
+            if admission.schedule_id == "twicer":
+                twice_id = rid
+            else:
+                fail_id = rid
+        assert twice_id and fail_id
+        await _wait_for(
+            lambda: FileRunStore(root).get_run(twice_id).status.value == "interrupted"
+        )
+        await _wait_for(
+            lambda: FileRunStore(root).get_run(fail_id).status.value == "interrupted"
+        )
+        first = await server.api.resume_run(
+            run_id=twice_id, resume_payload={}, resume_outcome="submitted"
+        )
+        assert first["status"] == "interrupted"
+        second = await server.api.resume_run(
+            run_id=twice_id, resume_payload={}, resume_outcome="submitted"
+        )
+        assert second["status"] == "completed"
+        failed = await server.api.resume_run(
+            run_id=fail_id, resume_payload={}, resume_outcome="submitted"
+        )
+        assert failed["status"] == "failed"
+
+        assert await _kinds(root, "twicer") == [
+            "admitted",
+            "interrupted",
+            "interrupted",
+            "completed",
+        ]
+        twicer_page = FileScheduleStore(root).list_occurrences("twicer", limit=100)
+        twicer_ckpts = [
+            row["checkpoint_id"]
+            for row in cast(list[dict[str, Any]], twicer_page["occurrences"])
+            if row["kind"] in ("interrupted", "completed")
+        ]
+        assert len(set(twicer_ckpts)) == 3
+        assert await _kinds(root, "failer") == ["admitted", "interrupted", "failed"]
+        fail_rows = _entries(root, "failer", "failed")
+        assert len(fail_rows) == 1
+        assert fail_rows[0]["run_id"] == fail_id
+    finally:
+        await service.stop()
+
+
+async def test_paused_and_deleted_schedules_keep_resume_history(
+    tmp_path: Path,
+) -> None:
+    """Pausing/deleting never suppresses retained resume history (B3)."""
+    root = tmp_path / "store"
+    server = build_local_static_workflow_server(root)
+    await _seed(server, "ask1", "ask1.default", _single_interrupt_plan("ask1"), ["submitted"])
+    store = FileScheduleStore(root)
+    store.create_schedule(_one_shot("pausable", "ask1.default", datetime.now(UTC)))
+    store.create_schedule(_one_shot("doomed", "ask1.default", datetime.now(UTC)))
+    service = _scheduler(server, capacity=2)
+    try:
+        await service.start()
+        await _wait_for(lambda: len(_run_ids(root)) == 2)
+        ids = {
+            FileRunStore(root).get_admission(rid).schedule_id or "": rid
+            for rid in _run_ids(root)
+        }
+        await _wait_for(
+            lambda: all(
+                FileRunStore(root).get_run(rid).status.value == "interrupted"
+                for rid in ids.values()
+            )
+        )
+        paused = store.get_schedule("pausable")
+        paused.paused = True
+        store.save_schedule(paused)
+        gone = store.get_schedule("doomed")
+        gone.deleted = True
+        store.save_schedule(gone)
+        for rid in ids.values():
+            resumed = await server.api.resume_run(
+                run_id=rid, resume_payload={}, resume_outcome="submitted"
+            )
+            assert resumed["status"] == "completed"
+        assert await _kinds(root, "pausable") == ["admitted", "interrupted", "completed"]
+        assert await _kinds(root, "doomed") == ["admitted", "interrupted", "completed"]
+    finally:
+        await service.stop()