diff --git a/src/agent_comms/cursor_owner.py b/src/agent_comms/cursor_owner.py index 6d1c8ef81c..f23e162847 100644 --- a/src/agent_comms/cursor_owner.py +++ b/src/agent_comms/cursor_owner.py @@ -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 @@ def _advance_owned(self, bus: MessageBus, source_reads: NativeEvidenceScope, sto 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,10 +155,13 @@ def require_participant(self, store: Coordination, reason: str) -> None: 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.""" @@ -172,11 +169,13 @@ def state(self, db: sqlite3.Connection) -> CursorState: 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_coverage(self, db: Connection, cursor: CurrentNativeCursor, coverage 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 @@ def injected_seq(self) -> int: 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 @@ def publish(self, db: Connection, owner: CursorOwner, coverage: ProvenSourceCove 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 publish(self, db: Connection, owner: CursorOwner, coverage: ProvenSourceCove 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) diff --git a/src/agent_comms/historical_native_inputs.py b/src/agent_comms/historical_native_inputs.py index 1b04860619..67df91c7a6 100644 --- a/src/agent_comms/historical_native_inputs.py +++ b/src/agent_comms/historical_native_inputs.py @@ -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 @@ class _HistoricalSource(TypedRow): 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 @@ def read_historical_native_inputs( 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), ) ) diff --git a/src/agent_comms/native_claim_epoch.py b/src/agent_comms/native_claim_epoch.py index 1ddf2a3c2b..067c8ae87b 100644 --- a/src/agent_comms/native_claim_epoch.py +++ b/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 diff --git a/src/agent_comms/proven_source_coverage.py b/src/agent_comms/proven_source_coverage.py index d7d61b9c97..1f560b3501 100644 --- a/src/agent_comms/proven_source_coverage.py +++ b/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): diff --git a/tests/test_certified_bootstrap.py b/tests/test_certified_bootstrap.py index f2beb768c8..4f31d9bf6e 100644 --- a/tests/test_certified_bootstrap.py +++ b/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 diff --git a/tests/test_native_context_reference.py b/tests/test_native_context_reference.py index a002f7e5bf..2c7463b640 100644 --- a/tests/test_native_context_reference.py +++ b/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,)) diff --git a/tests/test_sqlite_source_coverage.py b/tests/test_sqlite_source_coverage.py index 9109bac244..7445a7d6cb 100644 --- a/tests/test_sqlite_source_coverage.py +++ b/tests/test_sqlite_source_coverage.py @@ -1,6 +1,5 @@ """SQLite delivery prefixes preserve native gaps without a file checkpoint.""" -import sys import pytest @@ -13,7 +12,6 @@ from agent_comms.coordinator import Coordination from agent_comms.cursor_owner import UnstartedCursor from agent_comms.field_codec import ThreadLookup -from agent_comms.historical_native_inputs import first_foreign_claim_source from agent_comms.messages import InfoKind, Message from agent_comms.proven_source_coverage import SourceCoverage, SourcePrefix from agent_comms.publisher import ResponseWrites @@ -44,7 +42,7 @@ def test_native_coverage_uses_delivery_membership_and_keeps_unproven_barrier( def test_unstarted_coverage_needs_neither_sender_claim_nor_activation(coverage, source_reads): - result = coverage.since(UnstartedCursor(), stop_seq=sys.maxsize, source_reads=source_reads) + result = coverage.since(UnstartedCursor(), source_reads=source_reads) assert result.covered_seq == 0 assert result.blocked_seq is None assert result.inputs == () @@ -100,35 +98,3 @@ def slow(self, *args, **kwargs): assert result.blocked_seq == next((row.seq for row in record.deliveries()), None) assert result.covered_seq == (0 if record.deliveries() else record.seq) assert result.source_prefix.through_seq == record.seq - - -def test_coverage_stops_before_the_first_source_another_claim_took(coverage, source_reads, monkeypatch): - """A source another claim's input took is never scanned, so a cursor parked before it reads nothing.""" - bus = coverage.bus.bus - first, foreign, _later = (bus.append(lambda seq, db, text=text: PublicRecord.from_message( - Message("author", "worker", text, InfoKind, seq=seq))) for text in ("first", "foreign", "later")) - scanned = [] - received = type(bus).received - - def spy(self, lookup, *, after, before, **kwargs): - rows = received(self, lookup, after=after, before=before, **kwargs) - scanned.append((after, before, len(rows))) - return rows - - monkeypatch.setattr(type(bus), "received", spy) - result = coverage.since(UnstartedCursor(), stop_seq=foreign.seq, source_reads=source_reads) - assert result.covered_seq == first.seq - assert result.source_prefix.through_seq == first.seq - assert scanned == [(0, foreign.seq, 0)] - scanned.clear() - parked = coverage.since(UnstartedCursor(), stop_seq=first.seq, source_reads=source_reads) - assert parked.covered_seq == 0 - assert scanned == [(0, first.seq, 0)] - - -def test_a_fresh_owner_has_no_foreign_claim_source(coverage): - with coverage.store.session.read(): - assert first_foreign_claim_source( - coverage.store.session._connection, wire_root_id=coverage.wire_root_id, - recipient_lookup=coverage.recipient_lookup, after_seq=0, claimed_generation=1, - ) == sys.maxsize