Let TurnState answer which idle publication settles a held turn

src/agent_comms/turn_lease.py

@@ -100,6 +100,10 @@ def phase_effects(self, phase: TurnPhase) -> Iterator[TurnDeclaration]: ...
     @abstractmethod
     def matches(self, turn_id: TurnId) -> bool: ...
 
+    @abstractmethod
+    def settled_by(self, state: TurnState) -> bool:
+        """Whether an idle publication settles this held turn."""
+
 
 @dataclass(frozen=True)
 class NoTurn(NullMember, TurnDeclaration):
@@ -146,6 +150,9 @@ def phase_effects(self, phase: TurnPhase) -> Iterator[TurnDeclaration]:
     def matches(self, turn_id: TurnId) -> bool:
         return False
 
+    def settled_by(self, state: TurnState) -> bool:
+        return True
+
 
 class TurnGeneration(SubclassUnion, affix="Generation"):
     """A registry generation the turn file recorded, or none."""
@@ -267,6 +274,9 @@ def require_turn(self, turn_id: TurnId) -> None:
     def matches(self, turn_id: TurnId) -> bool:
         return self.id == turn_id
 
+    def settled_by(self, state: TurnState) -> bool:
+        return state.active.matches(self.id) or state.finished.names(self.id)
+
     def without_claim(self) -> ActiveTurn:
         return replace(self, claimed_generation=UnrecordedGeneration())
 
@@ -487,6 +497,9 @@ def released(self) -> TurnRecord:
     def require_generation(self, generation: int) -> None:
         pass
 
+    def names(self, turn_id: TurnId) -> bool:
+        return False
+
 
     @abstractmethod
     def observed_turn(self, identity: TurnIdentity, claim_scope: int) -> ObservedTurn: ...
@@ -526,6 +539,9 @@ def read(self) -> NoTurn:
     def state(self) -> TurnState:
         return TurnState(finished=self)
 
+    def names(self, turn_id: TurnId) -> bool:
+        return self.turn_id == turn_id
+
     def without_claim(self) -> None:
         pass
 
@@ -654,6 +670,10 @@ def phase_effects(self, phase: TurnPhase) -> Iterator[TurnState]:
     def native_phase_effects(self, phase: TurnPhase) -> Iterator[TurnState]:
         yield from self.phase_effects(self.active.phase.observed(phase))
 
+    def settles(self, held: TurnDeclaration) -> bool:
+        """A busy publication always replaces the held turn; an idle one only settles its own."""
+        return self.active.phase.busy or held.settled_by(self)
+
 
 class TurnCompletion(ABC):
     """Whether terminal publication attested an exact finished turn."""

tests/test_turn_lease_release.py

@@ -7,7 +7,7 @@
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.field_codec import FieldCodec, TurnId
 from agent_comms.threads import Thread
-from agent_comms.turn_lease import FinishedTurnFence, NoFinishedTurn, NoTurnLease
+from agent_comms.turn_lease import FinishedTurnFence, NoFinishedTurn, NoTurn, NoTurnLease
 
 
 def test_old_lease_cannot_clear_same_id_after_delete_rebind(tmp_path):
@@ -53,3 +53,20 @@ def test_unattested_saved_turn_remains_data_without_a_lease(tmp_path, *, thread_
     )
     assert restored.turn_record.read() == expected
     assert restored.turn_lease == NoTurnLease()
+
+
+def test_idle_publication_settles_only_the_turn_it_finished(tmp_path):
+    comms = wire(tmp_path)
+    comms.registry.declare(Thread(
+        'owner', frozenset(), str(tmp_path), process_identity=ProcessIdentity.capture(os.getpid()),
+    ))
+    first = comms.agents.begin_turn("owner", TurnId('3f1c2b0a9d8e7f6a5b4c3d2e1f0a9b8c')).turn_lease
+    held = comms.registry.require("owner").turn_state.active
+    comms.agents.finish_turn(first)
+    finished = comms.registry.require("owner").turn_state
+    comms.agents.begin_turn("owner", TurnId('c8b9a0f1e2d3c4b5a6f7e8d9c0b1a2f3'))
+    replacement = comms.registry.require("owner").turn_state
+    assert finished.settles(held)
+    assert finished.settles(NoTurn())
+    assert not finished.settles(replacement.active)
+    assert replacement.settles(held)