Skip to content

Annotate parameters in triage, cursor, Pi child and transcript modules #2757

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
trissim merged 2 commits into from
Oct 23, 2026
Merged

Annotate parameters in triage, cursor, Pi child and transcript modules #2757

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
abade62
Annotate parameters in triage, cursor, Pi child and transcript modules
trissim Oct 23, 2026
e518a2f
Merge remote-tracking branch 'origin/main' into fix/annotate-params-2
trissim Oct 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 33 additions & 28 deletions src/agent_comms/cursor_owner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from typing import TYPE_CHECKING

from agent_comms.native_runtime_input import NativeRuntimeTable

from .acp_extension import CursorObservation, EmptyCursorObservation, VerifiedCursorObservation
from .agent_backend import InputId
from .coordination_database import CoordinationStore
Expand All @@ -23,8 +24,12 @@
from .thread_identity import GenerationCounter

if TYPE_CHECKING:
from sqlite3 import Connection

from .coordination_tables.participants import OwnerGenerations
from .historical_native_inputs import HistoricalNativeInput
from .native_entries import NativeEvidenceScope
from .proven_source_coverage import CoverageInput, NoInjectedInput, ProvenSourceCoverage


@dataclass(frozen=True)
Expand All @@ -48,7 +53,7 @@ async def read_async(cls, bus: MessageBus, *, wire_root_id: str, owner_name: str
)

@classmethod
def _read_owned(cls, bus, root_id, owner_name, source_reads, store) -> CursorState:
def _read_owned(cls, bus: MessageBus, root_id: str, owner_name: str, source_reads: NativeEvidenceScope, store: Coordination) -> CursorState:
with store.session.boundary(bus) as boundary:
reason = "Current native cursor has no matching live owner"
bus._registry.store.private_guard_unlocked().require_root(root_id, StaleFence(reason))
Expand Down Expand Up @@ -96,14 +101,14 @@ async def refresh_async(cls, bus: MessageBus, *, wire_root_id: str, owner_name:
)

@classmethod
def _refresh_owned(cls, bus, root_id, owner_name, source_reads, store) -> CursorState:
def _refresh_owned(cls, bus: MessageBus, root_id: str, owner_name: str, source_reads: NativeEvidenceScope, store: Coordination) -> CursorState:
thread, generation = bus._registry.live_owner_with_claim(owner_name)
person = store.participants.get(thread.incarnation.require_recorded().lookup)
owner = cls(thread=thread, claimed_generation=generation, wire_root_id=root_id,
generation=person.owner_identity.generation)
return owner._advance_owned(bus, source_reads, store)

def _advance_owned(self, bus, source_reads, store) -> CursorState:
def _advance_owned(self, bus: MessageBus, source_reads: NativeEvidenceScope, store: Coordination) -> CursorState:
with source_reads.borrow():
sources = SourceCoverage(bus, store, wire_root_id=self.wire_root_id,
recipient_lookup=self.lookup)
Expand All @@ -126,7 +131,7 @@ def _advance_owned(self, bus, source_reads, store) -> CursorState:
return state

@property
def participant_identity(self):
def participant_identity(self) -> OwnerGenerations:
return ParticipantOwner(self.thread, self.generation).coordinator_identity(self.lookup)

@property
Expand Down Expand Up @@ -157,19 +162,19 @@ def state(self, db: sqlite3.Connection) -> CursorState:
self.require_recorded_input(db, row)
return RecordedCursor(row)

def _matches_input(self, row, reference) -> bool:
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.reference == reference)

def matches_prefix(self, db, evidence) -> bool:
def matches_prefix(self, db: Connection, evidence: tuple[HistoricalNativeInput, ...]) -> bool:
return all(self._matches_input(NativeRuntimeInput.one(db, input_id=item.input_id), item.reference)
for item in evidence)

def require_recorded_input(self, db, cursor) -> None:
def require_recorded_input(self, db: Connection, cursor: CurrentNativeCursor) -> None:
cursor.reference.require_recorded_input(self, db)

def require_coverage(self, db, cursor, coverage) -> None:
def require_coverage(self, db: Connection, cursor: CurrentNativeCursor, coverage: ProvenSourceCoverage) -> None:
if cursor.owner_identity != self.participant_identity:
raise IdentityConflict("current native cursor owner identity differs")
if cursor.covered_seq > coverage.covered_seq:
Expand All @@ -178,7 +183,7 @@ def require_coverage(self, db, cursor, coverage) -> None:
raise IdentityConflict("current cursor borrows historical owner source proof")
self.require_recorded_input(db, cursor)

def require_recorded_reference(self, db, reference: NativeInputReference) -> None:
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):
raise IdentityConflict("current cursor input claim_scope differs")
Expand Down Expand Up @@ -212,9 +217,9 @@ def cursor(self) -> CurrentNativeCursor | None: ...
def injected_seq(self) -> int: ...

@abstractmethod
def without_input(self, proof, owner, coverage): ...
def without_input(self, proof: NoInjectedInput, owner: CursorOwner, coverage: ProvenSourceCoverage) -> CurrentNativeCursor: ...

def require_progress(self, owner, coverage, injected) -> None:
def require_progress(self, owner: CursorOwner, coverage: ProvenSourceCoverage, injected: int) -> None:
if coverage.covered_seq < self.covered_seq:
raise IdentityConflict("current cursor would regress")

Expand All @@ -231,24 +236,24 @@ def require_input_progress(self, proof: HistoricalNativeInput) -> None: ...
def claim_mismatch(self, injected_seq: int) -> bool: ...

@abstractmethod
def publish(self, db, owner, coverage, proof) -> CursorState: ...
def publish(self, db: Connection, owner: CursorOwner, coverage: ProvenSourceCoverage, proof: CoverageInput) -> CursorState: ...

@abstractmethod
def require_coverage(self, owner, db, coverage, proof) -> None: ...
def require_coverage(self, owner: CursorOwner, db: Connection, coverage: ProvenSourceCoverage, proof: CoverageInput) -> None: ...


@dataclass(frozen=True)
class UnstartedCursor(CursorState):
@property
def covered_seq(self):
def covered_seq(self) -> int:
return 0

@property
def cursor(self):
def cursor(self) -> CurrentNativeCursor | None:
return None

@property
def injected_seq(self):
def injected_seq(self) -> int:
return 0 # No injected native input; cursor presence is this state's type.

def proves_input(self, input_id: str) -> bool:
Expand All @@ -260,20 +265,20 @@ def require_input_progress(self, proof: HistoricalNativeInput) -> None:
def claim_mismatch(self, injected_seq: int) -> bool:
return False

def without_input(self, proof, owner, coverage):
def without_input(self, proof: NoInjectedInput, owner: CursorOwner, coverage: ProvenSourceCoverage) -> CurrentNativeCursor:
return proof.build(owner, coverage.covered_seq)

def observation(self):
def observation(self) -> CursorObservation:
return EmptyCursorObservation()

def publish(self, db, owner, coverage, proof):
def publish(self, db: Connection, owner: CursorOwner, coverage: ProvenSourceCoverage, proof: CoverageInput) -> CursorState:
if coverage.covered_seq <= self.covered_seq:
return self # A blocked first source has not established a cursor.
cursor = proof.candidate(self, owner, coverage)
cursor.insert(db)
return RecordedCursor(cursor)

def require_coverage(self, owner, db, coverage, proof):
def require_coverage(self, owner: CursorOwner, db: Connection, coverage: ProvenSourceCoverage, proof: CoverageInput) -> None:
pass # No durable cursor exists to assert coverage.


Expand All @@ -282,27 +287,27 @@ class RecordedCursor(CursorState):
row: CurrentNativeCursor

@property
def covered_seq(self):
def covered_seq(self) -> int:
return self.row.covered_seq

@property
def cursor(self):
def cursor(self) -> CurrentNativeCursor | None:
return self.row

@property
def injected_seq(self):
def injected_seq(self) -> int:
return self.row.injected_seq

def without_input(self, proof, owner, coverage):
def without_input(self, proof: NoInjectedInput, owner: CursorOwner, coverage: ProvenSourceCoverage) -> CurrentNativeCursor:
return replace(self.row, covered_seq=coverage.covered_seq)

def observation(self):
def observation(self) -> CursorObservation:
return VerifiedCursorObservation(self.row)

def proves_input(self, input_id: str) -> bool:
return self.row.input_id.holds(input_id)

def require_progress(self, owner, coverage, injected):
def require_progress(self, owner: CursorOwner, coverage: ProvenSourceCoverage, injected: int) -> None:
super().require_progress(owner, coverage, injected)
if (self.row.owner_identity != owner.participant_identity
or injected < self.row.injected_seq):
Expand All @@ -317,7 +322,7 @@ def claim_mismatch(self, injected_seq: int) -> bool:
return False
raise IdentityConflict("current cursor input claim_scope differs")

def publish(self, db, owner, coverage, proof):
def publish(self, db: Connection, owner: CursorOwner, coverage: ProvenSourceCoverage, proof: CoverageInput) -> CursorState:
cursor = proof.candidate(self, owner, coverage)
if cursor.covered_seq == self.row.covered_seq and cursor.injected_seq == self.row.injected_seq:
return self
Expand All @@ -334,6 +339,6 @@ def publish(self, db, owner, coverage, proof):
raise StaleFence("current cursor monotonic update lost its fence")
return RecordedCursor(cursor)

def require_coverage(self, owner, db, coverage, proof):
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)
Loading
Loading