Integration batch 10-25: land #3189 over #3186

src/agent_comms/cursor_owner.py

@@ -1,3 +1,3 @@
 """Captured owner and typed initial/recorded states for native source cursors."""
 from __future__ import annotations
 
@@ -14,7 +14,6 @@
 from .coordination_database import CoordinationStore
 from .coordination_errors import IdentityConflict, StaleFence
 from .coordinator import Coordination
-from .historical_native_inputs import first_foreign_claim_source
 from .message_bus import MessageBus
 from .native_input_owner import ParticipantOwner, RegistryOwner
 from .native_input_record import NativeInputReference
@@ -120,13 +119,8 @@
             sources = SourceCoverage(bus, store, wire_root_id=self.wire_root_id,
                                      recipient_lookup=self.lookup)
             with store.session.boundary(bus), store.session.read():
-                db = store.session._connection
-                state = self.state(db)
-                stop_seq = first_foreign_claim_source(
-                    db, wire_root_id=self.wire_root_id, recipient_lookup=self.lookup,
-                    after_seq=state.covered_seq, claimed_generation=self.claimed_generation,
-                )
-            coverage = sources.since(state, stop_seq=stop_seq, source_reads=source_reads)
+                state = self.state(store.session._connection)
+            coverage = sources.since(state, source_reads=source_reads)
             proof = coverage.proof
             with store.session.boundary(bus) as boundary, store.session.writing() as db:
                 coverage.source_prefix.require_current(bus.bus)
@@ -161,22 +155,27 @@
             raise StaleFence(reason)
 
     def cursor(self, db: sqlite3.Connection) -> CurrentNativeCursor | None:
-        return CurrentNativeCursor.one(
-            db, wire_root_id=self.wire_root_id, recipient_lookup=self.lookup,
-            owner_generation=self.generation, owner_claimed_generation=self.claimed_generation,
+        """The thread's newest cursor; a restart's new claim continues it."""
+        rows = CurrentNativeCursor.select(
+            db, where="{owner_claimed_generation}<=?", parameters=(self.claimed_generation,),
+            order_by=("owner_claimed_generation",), descending=True, limit=1,
+            wire_root_id=self.wire_root_id, recipient_lookup=self.lookup, owner_generation=self.generation,
         )
+        return next(iter(rows), None)
 
     def state(self, db: sqlite3.Connection) -> CursorState:
         """Classify recipient progress once, at its storage boundary."""
         row = self.cursor(db)
         if row is None:
             return UnstartedCursor()
         self.require_recorded_input(db, row)
+        if row.owner_claimed_generation != self.claimed_generation:
+            return InheritedCursor(row)
         return RecordedCursor(row)
 
     def _matches_input(self, row: NativeRuntimeInput | None, reference: NativeInputReference) -> bool:
         return (row is not None and row.owner_identity == self.participant_identity
-                and row.sent_owner_claimed_generation.matches(self.claimed_generation)
+                and row.sent_owner_claimed_generation.sent_by_lineage(self.claimed_generation)
                 and row.reference == reference)
 
     def matches_prefix(self, db: Connection, evidence: tuple[HistoricalNativeInput, ...]) -> bool:
@@ -197,7 +196,7 @@
 
     def require_recorded_reference(self, db: Connection, reference: NativeInputReference) -> None:
         row = NativeRuntimeInput.one(db, input_id=reference.input_id)
-        if row is None or not row.sent_owner_claimed_generation.matches(self.claimed_generation):
+        if row is None or not row.sent_owner_claimed_generation.sent_by_lineage(self.claimed_generation):
             raise IdentityConflict("current cursor input claim_scope differs")
         if not self._matches_input(row, reference):
             raise IdentityConflict("current cursor proof differs from journal")
@@ -311,7 +310,7 @@
         return self.row.injected_seq
 
     def without_input(self, proof: NoInjectedInput, owner: CursorOwner, coverage: ProvenSourceCoverage) -> CurrentNativeCursor:
-        return replace(self.row, covered_seq=coverage.covered_seq)
+        return replace(self.row, owner_claimed_generation=owner.claimed_generation, covered_seq=coverage.covered_seq)
 
     def observation(self) -> CursorObservation:
         return VerifiedCursorObservation(self.row)
@@ -338,6 +337,9 @@
         cursor = proof.candidate(self, owner, coverage)
         if cursor.covered_seq == self.row.covered_seq and cursor.injected_seq == self.row.injected_seq:
             return self
+        return self.write(db, owner, cursor)
+
+    def write(self, db: Connection, owner: CursorOwner, cursor: CurrentNativeCursor) -> CursorState:
         updated = CurrentNativeCursor.update(
             db, where="wire_root_id=? AND recipient_lookup=? AND owner_generation=? "
             "AND {owner_claimed_generation}=? AND covered_seq=? AND injected_seq=?",
@@ -354,3 +356,16 @@
     def require_coverage(self, owner: CursorOwner, db: Connection, coverage: ProvenSourceCoverage, proof: CoverageInput) -> None:
         owner.require_coverage(db, self.row, coverage)
         proof.require_reference(owner, self.row)
+
+
+@dataclass(frozen=True)
+class InheritedCursor(RecordedCursor):
+    """An earlier claim's row; this claim's first publish writes its own row from it."""
+
+    def observation(self) -> CursorObservation:
+        # ACP scopes a cursor to the live claim (acp_extension.py validate_scope): none until it publishes.
+        return EmptyCursorObservation()
+
+    def write(self, db: Connection, owner: CursorOwner, cursor: CurrentNativeCursor) -> CursorState:
+        cursor.insert(db)
+        return RecordedCursor(cursor)

src/agent_comms/historical_native_inputs.py

@@ -1,3 +1,3 @@
 """Historical native input evidence, not an injected-message cursor.
 
 A SQL row recorded only after a live Pi result can be corroborated against the
@@ -8,9 +8,6 @@
 
 from __future__ import annotations
 
-import sqlite3
-import sys
-
 from agent_comms.native_session_files import NativeSessionFiles
 from abc import abstractmethod
 from dataclasses import dataclass
@@ -100,41 +97,6 @@
     message_id: MessageId
 
 
-def _selected_sources(columns: str, condition: str) -> str:
-    """One recipient's sealed selected sources joined to the native inputs that took them."""
-    return (
-        f"SELECT {columns} FROM native_runtime_input n "
-        f"JOIN ({NativeRuntimeInput.source_membership_sql()}) member ON member.input_id=n.input_id "
-        "JOIN wake_claims c ON c.assignment_id=member.assignment_id "
-        "JOIN claim_batch_members m ON m.claim_id=c.assignment_id "
-        "AND m.recipient_lookup=c.recipient_lookup "
-        "JOIN claim_batch_receipts r ON r.wire_root_id=m.wire_root_id "
-        "AND r.wire_seq=m.wire_seq AND r.message_id=c.message_id AND r.sealed=1 "
-        "JOIN cohort_delivery_receipts d ON d.wire_root_id=r.wire_root_id "
-        "AND d.wire_seq=r.wire_seq AND d.claim_id=c.assignment_id "
-        "AND d.recipient_lookup=n.owner_lookup AND d.kind='selected' "
-        f"WHERE r.wire_root_id=? AND c.recipient_lookup=? AND {condition} "
-        "AND n.owner_lookup=c.recipient_lookup AND n.session_entry_id IS NOT NULL"
-    )
-
-
-def first_foreign_claim_source(
-    db: sqlite3.Connection, *, wire_root_id: RootId, recipient_lookup: ThreadLookup,
-    after_seq: int, claimed_generation: int,
-) -> int:
-    """The first source after ``after_seq`` taken by an input another claim sent.
-
-    A cursor owner proves only inputs its own claim sent (CursorOwner.matches_prefix),
-    so no coverage scan publishes past this source. With none, the scan may reach the bus end.
-    """
-    (seq,) = db.execute(
-        _selected_sources("COALESCE(MIN(c.wire_seq), ?)",
-                          "c.wire_seq>? AND n.sent_owner_claimed_generation IS NOT ?"),
-        (sys.maxsize, wire_root_id, recipient_lookup, after_seq, claimed_generation),
-    ).fetchone()
-    return int(seq)
-
-
 def read_historical_native_inputs(
     store: Coordination,
     *,
@@ -164,7 +126,18 @@
             NativeRuntimeTable.require_installed(db)
             sources = _HistoricalSource.read(
                 db.execute(
-                    _selected_sources("n.input_id,c.assignment_id,c.wire_seq,c.message_id", "c.wire_seq=?"),
+                    "SELECT n.input_id,c.assignment_id,c.wire_seq,c.message_id FROM native_runtime_input n "
+                    f"JOIN ({NativeRuntimeInput.source_membership_sql()}) member ON member.input_id=n.input_id "
+                    "JOIN wake_claims c ON c.assignment_id=member.assignment_id "
+                    "JOIN claim_batch_members m ON m.claim_id=c.assignment_id "
+                    "AND m.recipient_lookup=c.recipient_lookup "
+                    "JOIN claim_batch_receipts r ON r.wire_root_id=m.wire_root_id "
+                    "AND r.wire_seq=m.wire_seq AND r.message_id=c.message_id AND r.sealed=1 "
+                    "JOIN cohort_delivery_receipts d ON d.wire_root_id=r.wire_root_id "
+                    "AND d.wire_seq=r.wire_seq AND d.claim_id=c.assignment_id "
+                    "AND d.recipient_lookup=n.owner_lookup AND d.kind='selected' "
+                    "WHERE r.wire_root_id=? AND c.recipient_lookup=? AND c.wire_seq=? "
+                    "AND n.owner_lookup=c.recipient_lookup AND n.session_entry_id IS NOT NULL",
                     (wire_root_id, recipient_lookup, source_seq),
                 )
             )

src/agent_comms/native_claim_epoch.py

@@ -26,6 +26,10 @@ class NativeClaimEpoch(DeclaredFamily, JsonShapeFamily, affix="NativeClaim"):
     @abstractmethod
     def matches(self, claimed_generation: int) -> bool: ...
 
+    @abstractmethod
+    def sent_by_lineage(self, claimed_generation: int) -> bool:
+        """This thread sent the input under this claim or an earlier, fenced one."""
+
     @abstractmethod
     def reservation_violation(self) -> bool: ...
 
@@ -48,6 +52,9 @@ class UnrecordedNativeClaim(NativeClaimEpoch, JsonShapeMember):
     def matches(self, claimed_generation: int) -> bool:
         return False
 
+    def sent_by_lineage(self, claimed_generation: int) -> bool:
+        return False
+
     def reservation_violation(self) -> bool:
         return False
 
@@ -83,6 +90,10 @@ def __post_init__(self) -> None:
     def matches(self, claimed_generation: int) -> bool:
         return self.value == claimed_generation
 
+    def sent_by_lineage(self, claimed_generation: int) -> bool:
+        # Claims only grow per name; a later one is a newer writer, never ours.
+        return self.value <= claimed_generation
+
     def reservation_violation(self) -> bool:
         return True
 

src/agent_comms/proven_source_coverage.py

@@ -10,7 +10,7 @@
 
 import sqlite3
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field, replace
+from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
 from agent_comms.coordination_errors import IdentityConflict
@@ -57,10 +57,6 @@ def require_start(self, covered_seq: int) -> None:
         if covered_seq > self.through_seq:
             raise IdentityConflict("current source coverage exceeds the bus")
 
-    def before(self, stop_seq: int) -> SourcePrefix:
-        """This prefix ending just before ``stop_seq``, when that source is inside it."""
-        return replace(self, through_seq=min(self.through_seq, stop_seq - 1))
-
     def require_current(self, bus: Bus) -> None:
         if not ObservedPath.at(bus.path).same_file(self.identity):
             raise IdentityConflict("native source database changed")
@@ -145,7 +141,7 @@ def allows(self, owner: CursorOwner, db: sqlite3.Connection, state: CursorState)
         if not owner.permits_input(self.original.input_id):
             return False
         row = NativeRuntimeInput.one(db, input_id=self.original.input_id)
-        if row is not None and not row.sent_owner_claimed_generation.matches(owner.claimed_generation):
+        if row is not None and not row.sent_owner_claimed_generation.sent_by_lineage(owner.claimed_generation):
             return state.claim_mismatch(self.original.source_seq)
         owner.require_recorded_reference(db, self.original.reference)
         return True
@@ -329,16 +325,11 @@ def native_inputs(self, source_seq: int, *, source_reads: NativeEvidenceScope) -
             source_seq=source_seq, source_reads=source_reads,
         )
 
-    def since(self, state: CursorState, *, stop_seq: int,
+    def since(self, state: CursorState, *,
               source_reads: NativeEvidenceScope) -> ProvenSourceCoverage:
-        """Prove sources beyond the cursor, up to the first one the owner cannot prove.
-
-        ``stop_seq`` is the owner's first source taken by another claim's input. Every
-        scan past it rereads the whole history only to be refused at publication, so the
-        captured prefix ends before it and an unchanged source reads nothing.
-        """
+        """Prove sources beyond the cursor, bounded to one captured SQLite prefix."""
         with source_reads.borrow():
-            prefix = SourcePrefix.capture(self.bus.bus, state.covered_seq).before(stop_seq)
+            prefix = SourcePrefix.capture(self.bus.bus, state.covered_seq)
             covered = state.covered_seq
             inputs, no_wake = [], []
             for _ in range(self.page_budget):

tests/test_certified_bootstrap.py

@@ -4,7 +4,6 @@
 
 
 import os
-import sys
 
 import pytest
 
@@ -25,16 +24,16 @@ def test_fresh_root_claims_source_gaps_and_empty_cursor(tmp_path, *, source_read
         assert comms.bus.bus.revision() == 0
         empty = SourceCoverage(
             comms.bus, store, wire_root_id=root_id, recipient_lookup=lookups["bob"]
-        ).since(UnstartedCursor(), stop_seq=sys.maxsize, source_reads=source_reads)
+        ).since(UnstartedCursor(), source_reads=source_reads)
         assert empty.covered_seq == 0 and empty.injected_source_seqs == ()
         sent = comms.messaging.send_user_message(CommandTarget.named('bob'), "fresh selected input", worktree=str(tmp_path))
         reader = SourceCoverage(
             comms.bus, store, wire_root_id=root_id, recipient_lookup=lookups["bob"]
         )
-        before = reader.since(UnstartedCursor(), stop_seq=sys.maxsize, source_reads=source_reads)
+        before = reader.since(UnstartedCursor(), source_reads=source_reads)
         assert before.blocked_seq == sent.seq and before.injected_source_seqs == ()
         accept_initial(comms.bus, root_id, sent.seq, store)
-        selected = reader.since(UnstartedCursor(), stop_seq=sys.maxsize, source_reads=source_reads)
+        selected = reader.since(UnstartedCursor(), source_reads=source_reads)
         assert selected.blocked_seq == sent.seq  # Selection cannot manufacture native proof.
         assert selected.covered_seq == 0 and selected.injected_source_seqs == ()
         assert wire(comms.root).bus.bus.revision() == sent.seq

tests/test_native_context_reference.py

@@ -15,10 +15,11 @@
 from agent_comms.cursor_owner import (
     CommittedCursorOwner,
     CursorOwner,
+    InheritedCursor,
     RecordedCursor,
     UnstartedCursor,
 )
-from agent_comms.field_codec import FieldCodec, InputId, MessageId, ThreadLookup
+from agent_comms.field_codec import FieldCodec, InputId, MessageId, RootId, ThreadLookup
 from agent_comms.historical_native_inputs import (
     FullHistoricalNativeInput,
     TriageHistoricalNativeInput,
@@ -36,7 +37,8 @@
 )
 from agent_comms.native_pi import NativeContextProof
 from agent_comms.native_runtime_input import CurrentNativeCursor, NativeRuntimeInput
-from agent_comms.proven_source_coverage import InjectedInput
+from agent_comms.private_path import FileIdentity
+from agent_comms.proven_source_coverage import InjectedInput, NoInjectedInput, ProvenSourceCoverage, SourcePrefix
 from agent_comms.selected_triage import (
     AbsentTriageDecisionRecord,
     DecidedTriageRecord,
@@ -200,3 +202,46 @@ def test_original_native_sql_acquires_required_triage_or_full_recorded_proof(rec
             )
         with pytest.raises(TypeError):
             TriageHistoricalNativeInput(execution=TriageNativeExecution(), **source)
+
+
+def test_a_restarted_claim_advances_past_the_inputs_its_earlier_claim_sent(
+    tmp_path, recorded, historical_source,
+):
+    """Every install mints a new claim; the thread's cursor and its inputs stay its own."""
+    thread = Thread("owner", frozenset(), str(tmp_path))
+    lookup = thread.incarnation.require_recorded().lookup
+    root = FieldCodec.decode(RootId, "a" * 32)
+    sent = replace(recorded, owner_lookup=lookup, verdict=DecidedTriageRecord(IgnoreSelectedTriage))
+    restarted = CursorOwner(thread=thread, claimed_generation=5, generation=1, wire_root_id=root)
+    with sqlite3.connect(":memory:") as db:
+        NativeRuntimeInput.create(db)
+        CurrentNativeCursor.create(db)
+        sent.insert(db)
+        proof = sent.execution.historical_proof(
+            sent, lifecycle=TriagePendingAssignment(), **{**historical_source, "owner_lookup": lookup},
+        )
+        assert restarted.matches_prefix(db, (proof,))
+        assert InjectedInput(proof).allows(restarted, db, UnstartedCursor())
+        earlier = CursorOwner(thread=thread, claimed_generation=2, generation=1, wire_root_id=root)
+        earlier_row = InjectedInput(proof).build(earlier, 1)
+        earlier_row.insert(db)
+        inherited = restarted.state(db)
+        assert type(inherited) is InheritedCursor and inherited.covered_seq == 1
+        assert inherited.observation() == EmptyCursorObservation()
+        coverage = ProvenSourceCoverage(
+            root, lookup, 3, (), (), None, source_prefix=SourcePrefix(FileIdentity.from_stat(tmp_path.stat()), 3),
+        )
+        published = inherited.publish(db, restarted, coverage, NoInjectedInput())
+        assert restarted.state(db) == published == RecordedCursor(replace(
+            earlier_row, owner_claimed_generation=5, covered_seq=3,
+        ))
+        assert earlier.state(db) == RecordedCursor(earlier_row)
+        newer = replace(sent, input_id=FieldCodec.decode(InputId, "9" * 32), sent_owner_claimed_generation=RecordedNativeClaim(6))
+        newer.insert(db)
+        newer_proof = newer.execution.historical_proof(newer, lifecycle=TriagePendingAssignment(), **{
+            **historical_source, "owner_lookup": lookup, "input_id": newer.input_id,
+            "context": replace(historical_source["context"], input_id=newer.input_id),
+            "native_reference": newer.reference,
+        })
+        assert restarted.matches_prefix(db, (proof,))
+        assert not restarted.matches_prefix(db, (newer_proof,))