test(control): adversarial restart replay invariants

src/forge_loop/eventlog/__init__.py

@@ -8,7 +8,11 @@
 
 from forge_loop.eventlog.legacy_mirror import LegacyEventMirror, LegacyRunnerEventKind
 from forge_loop.eventlog.models import EventEnvelope, EventId, EventKind, EventRef
-from forge_loop.eventlog.projections import ProjectionCursor
+from forge_loop.eventlog.projections import (
+    ProjectionCursor,
+    ProjectionReplayError,
+    replay_projection,
+)
 from forge_loop.eventlog.sqlite import SqliteEventLog
 from forge_loop.eventlog.store import EventLog, InMemoryEventLog
 
@@ -22,5 +26,7 @@
     "LegacyEventMirror",
     "LegacyRunnerEventKind",
     "ProjectionCursor",
+    "ProjectionReplayError",
     "SqliteEventLog",
+    "replay_projection",
 ]

src/forge_loop/eventlog/projections.py

@@ -2,12 +2,17 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import dataclass
 from typing import Protocol
 
 from forge_loop.eventlog.models import EventEnvelope
 
 
+class ProjectionReplayError(RuntimeError):
+    """Raised when replay or cursor advancement would make a projection unsafe."""
+
+
 @dataclass(frozen=True)
 class ProjectionCursor:
     """Where a projection last caught up to the durable event log."""
@@ -22,3 +27,39 @@ class Projection(Protocol):
 
     def apply(self, event: EventEnvelope) -> None:
         """Apply one event. Implementations should be idempotent by sequence."""
+
+
+class ProjectionEventLog(Protocol):
+    """Event-log operations needed for deterministic projection replay."""
+
+    def since(self, sequence: int = 0) -> Iterable[EventEnvelope]:
+        """Yield events with sequence greater than ``sequence`` in log order."""
+        ...
+
+    def advance_projection_cursor(
+        self,
+        projection_name: str,
+        cursor: ProjectionCursor,
+    ) -> None:
+        """Persist a cursor only when it is monotonic and within the log tail."""
+        ...
+
+
+def replay_projection(
+    event_log: ProjectionEventLog,
+    projection_name: str,
+    projection: Projection,
+) -> ProjectionCursor:
+    """Replay events from a projection cursor and persist the replayed tail.
+
+    The cursor advances only after every yielded event is applied. A projection
+    that receives events out of order should raise before the stored cursor
+    moves, which keeps restart recovery from blessing a partial replay.
+    """
+
+    cursor = projection.cursor
+    for event in event_log.since(cursor.sequence):
+        projection.apply(event)
+        cursor = ProjectionCursor(sequence=event.sequence)
+    event_log.advance_projection_cursor(projection_name, cursor)
+    return cursor

src/forge_loop/eventlog/sqlite.py

@@ -11,7 +11,7 @@
 from typing import Any
 
 from forge_loop.eventlog.models import EventEnvelope, EventId, EventKind, EventRef
-from forge_loop.eventlog.projections import ProjectionCursor
+from forge_loop.eventlog.projections import ProjectionCursor, ProjectionReplayError
 
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS events (
@@ -178,6 +178,27 @@ def set_projection_cursor(
                 (projection_name, cursor.sequence),
             )
 
+    def advance_projection_cursor(
+        self,
+        projection_name: str,
+        cursor: ProjectionCursor,
+    ) -> None:
+        """Persist a cursor only when it advances monotonically within the log."""
+
+        current = self.get_projection_cursor(projection_name)
+        if cursor.sequence < current.sequence:
+            raise ProjectionReplayError(
+                f"stale projection cursor for {projection_name}: "
+                f"{cursor.sequence} < {current.sequence}"
+            )
+        latest = self.latest_sequence()
+        if cursor.sequence > latest:
+            raise ProjectionReplayError(
+                f"projection cursor for {projection_name} is past latest event sequence: "
+                f"{cursor.sequence} > {latest}"
+            )
+        self.set_projection_cursor(projection_name, cursor)
+
     def list_projection_cursors(self) -> Mapping[str, ProjectionCursor]:
         """Return saved projection cursors by projection name."""
 

src/forge_loop/eventlog/store.py

@@ -13,7 +13,7 @@
 from typing import Any, Protocol
 
 from forge_loop.eventlog.models import EventEnvelope, EventId, EventKind
-from forge_loop.eventlog.projections import ProjectionCursor
+from forge_loop.eventlog.projections import ProjectionCursor, ProjectionReplayError
 
 
 class EventLog(Protocol):
@@ -51,6 +51,14 @@ def set_projection_cursor(
         """Persist a projection cursor."""
         ...
 
+    def advance_projection_cursor(
+        self,
+        projection_name: str,
+        cursor: ProjectionCursor,
+    ) -> None:
+        """Persist a cursor only when it is monotonic and within the log tail."""
+        ...
+
     def list_projection_cursors(self) -> Mapping[str, ProjectionCursor]:
         """Return all saved projection cursors by projection name."""
         ...
@@ -106,5 +114,24 @@ def set_projection_cursor(
     ) -> None:
         self._projection_cursors[projection_name] = cursor
 
+    def advance_projection_cursor(
+        self,
+        projection_name: str,
+        cursor: ProjectionCursor,
+    ) -> None:
+        current = self.get_projection_cursor(projection_name)
+        if cursor.sequence < current.sequence:
+            raise ProjectionReplayError(
+                f"stale projection cursor for {projection_name}: "
+                f"{cursor.sequence} < {current.sequence}"
+            )
+        latest = self.latest_sequence()
+        if cursor.sequence > latest:
+            raise ProjectionReplayError(
+                f"projection cursor for {projection_name} is past latest event sequence: "
+                f"{cursor.sequence} > {latest}"
+            )
+        self.set_projection_cursor(projection_name, cursor)
+
     def list_projection_cursors(self) -> Mapping[str, ProjectionCursor]:
         return dict(self._projection_cursors)

tests/test_boot_context.py

@@ -14,7 +14,7 @@
     MemoryProvenance,
     SqliteMemoryStore,
 )
-from forge_loop.tasks import SqliteTaskSagaStore, TaskSaga, TaskState
+from forge_loop.tasks import Compensation, SqliteTaskSagaStore, TaskSaga, TaskState
 
 
 def _frontier() -> FrontierCursor:
@@ -133,6 +133,13 @@ def test_boot_context_accepts_no_active_tasks_after_reopen(tmp_path: Path) -> No
             saga_id="saga-165-failed",
             state=TaskState.FAILED,
             issue=165,
+            compensations=(
+                Compensation(
+                    kind="cleanup-worktree",
+                    target="/tmp/wt-loop-165",
+                    reason="failed terminal tasks must keep their cleanup record",
+                ),
+            ),
         )
     )