From c0f5aa467bcfe2bcd7e40e00823fa747ad2c6241 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 14 Nov 2026 20:55:59 -0500 Subject: [PATCH] Observer reuses a window's decoded claims while its rows are unchanged On every #comms message the observer re-decoded all open threads' notification windows, even the N-1 whose evidence did not move. Decoding the four-table claim join (FieldCodec) is ~a third of the observer child's per-ping CPU (observer child 6.6-7.5% per ping, over the 5% target). The decode is a pure function of the raw join rows. capture() now hashes those raw rows and reuses the decoded claims when the hash is unchanged for the same (root, selection), decoding only the window(s) whose rows actually moved. The hash covers the whole join -- claim, triage context, queue pointer and published response -- so any change re-decodes; consumption, bus responses and stored inputs are never cached and are read fresh every pass, so nothing displayed can go stale. The cache is bounded to the most recent selections (a sliding window makes a new selection each move). root is optional: callers outside the observer loop decode fresh. This supersedes PR #4776's batch (it batched reads but not the decode, so it did not move the number; removed prime/read_batch/sliced). Notification + observation suite: 112 passed, same 3 pre-existing main reds as the baseline, zero added. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XbFMSnLEc51gw21bLFFRK1 --- .../execution/selected/participant.py | 1 + .../messaging/message_notification.py | 9 - src/agent_comms/notification_assignment.py | 172 ++++++++++-------- src/agent_comms/ui_model/observation.py | 1 - 4 files changed, 96 insertions(+), 87 deletions(-) diff --git a/src/agent_comms/execution/selected/participant.py b/src/agent_comms/execution/selected/participant.py index 8a10486bab..83059e705a 100644 --- a/src/agent_comms/execution/selected/participant.py +++ b/src/agent_comms/execution/selected/participant.py @@ -473,6 +473,7 @@ def prepare(self, store: Coordination, measured: PublicationMeasurements) -> Pre sent_read = NotificationAssignment.capture( store.session._connection, NotificationSelection.of(viewer, next(pages, ())), privacy, + self.comms.root, ) participant.pointer.require_idle() assignments = AssignmentSource(self.comms.root) diff --git a/src/agent_comms/messaging/message_notification.py b/src/agent_comms/messaging/message_notification.py index 3a305e31ea..5409de7d1b 100644 --- a/src/agent_comms/messaging/message_notification.py +++ b/src/agent_comms/messaging/message_notification.py @@ -132,15 +132,6 @@ def evidence(self) -> CapturedNotificationEvidence: def changed(self, window: NotificationWindow, earlier: RevisionChanges) -> bool: return window.changed(self, earlier) - def prime(self, windows: Iterable[NotificationWindow], earlier: RevisionChanges) -> None: - """Read every window's assignment evidence in one batch before any window's change - check. A check only reads assignments when the global assignment evidence moved, and - then every window reads; batching that read is one coordinator pass for the whole - sidebar instead of one per open thread.""" - if self.revision.changed_since(AssignmentEvidenceView(), earlier): - self.assignments.read_batch( - read.selection for window in windows for read in window.reads) - @dataclass(frozen=True) class CapturedNotificationEvidence: diff --git a/src/agent_comms/notification_assignment.py b/src/agent_comms/notification_assignment.py index 5faf168519..a54c999b46 100644 --- a/src/agent_comms/notification_assignment.py +++ b/src/agent_comms/notification_assignment.py @@ -207,18 +207,6 @@ def differs_from(self, fresh: Mapping[MessageReference, SourceEvidence]) -> bool return any(previous != fresh.get(source, SourceEvidence()) for source, previous in self.evidence.items()) - def sliced(self, selection: NotificationSelection) -> NotificationAssignmentRead: - """This read's rows for one window's keys, as that window's own read. A combined - read covers many windows' keys in one capture; each window owns its slice, so its - evidence and change check answer exactly as a lone read of its selection would.""" - keys = frozenset(selection.keys) - return NotificationAssignmentRead( - selection, - tuple(row for row in self.rows if (row[0].recipient_lookup, row[0].source) in keys), - MappingProxyType({key: value for key, value in self.consumption.items() if key in keys}), - MappingProxyType({key: value for key, value in self.responses.items() if key in keys}), - ) - @dataclass(frozen=True) class NotificationReadMark: @@ -360,6 +348,22 @@ def project( self.published, sender_lookup, observation=observation) +@dataclass(frozen=True) +class _DecodedRows: + """Already-read rows and their column description, presented to joined() as the cursor it + expects. The cache holds the raw rows for its fingerprint; when it must decode, joined reads + the columns from `description` and the rows by iteration, exactly as from a live cursor.""" + + rows: list + description: object + + def __iter__(self): + return iter(self.rows) + + def close(self) -> None: + pass + + @dataclass(frozen=True) class NotificationAssignment(NotificationSource): assignment: WakeAssignment @@ -467,12 +471,21 @@ def capture_detached( """Every bus-side evidence read that follows the coordinator snapshot.""" return cls.capture_inputs(source, cls.capture_responses(source.root, read)) + # One decoded claim set per (root, selection), reused across the cut changes a ping brings + # while the raw join rows are byte-identical. Decoding the rows (FieldCodec, ~a third of the + # observer's per-ping CPU) is a pure function of them; consumption, bus responses and stored + # inputs are never cached and are read fresh every pass, so nothing displayed can go stale. + _decoded: ClassVar[dict[tuple[Path, NotificationSelection], tuple[int, tuple[ + tuple[WakeAssignment, AssignmentActivityEvidence], ...]]]] = {} + _decoded_limit: ClassVar[int] = 512 + @classmethod def capture( cls, connection: sqlite3.Connection, selection: NotificationSelection, privacy: RegistryPrivacy, + root: Path | None = None, ) -> NotificationAssignmentRead: """Probe only selected frozen keys before joining any claim/context/proof. @@ -484,10 +497,57 @@ def capture( keys = tuple(dict.fromkeys(selection.keys)) if not keys: return NotificationAssignmentRead(selection, ()) + rows = cls._decoded_claims(connection, selection, keys, privacy, root) + consumption: dict[ + tuple[ThreadLookup, MessageReference], tuple[ConsumptionEvidence, ...] + ] = {} + for attempt in HarnessDelivery.for_recipients(connection, keys): + key = (attempt.recipient_lookup, attempt.reference) + consumption[key] = (*consumption.get(key, ()), attempt.monitor) + return NotificationAssignmentRead( + selection, + tuple(sorted(rows, key=lambda row: (-row[0].wire_seq, row[0].recipient))), + MappingProxyType(consumption), + ) + + @classmethod + def _decoded_claims( + cls, connection: sqlite3.Connection, selection: NotificationSelection, + keys: tuple[tuple[ThreadLookup, MessageReference], ...], privacy: RegistryPrivacy, + root: Path | None, + ) -> tuple[tuple[WakeAssignment, AssignmentActivityEvidence], ...]: + raw, description = cls._claim_scan(connection, keys, privacy) + if root is None: # a caller outside the observer loop never reuses a decode + return tuple(WakeAssignment.joined( + _DecodedRows(raw, description), AssignmentActivityEvidence)) + fingerprint = hash(frozenset(tuple(row) for row in raw)) + cache_key = (root, selection) + cached = cls._decoded.pop(cache_key, None) + if cached is not None and cached[0] == fingerprint: + cls._decoded[cache_key] = cached # keep as most-recently-used + return cached[1] + decoded = tuple(WakeAssignment.joined( + _DecodedRows(raw, description), AssignmentActivityEvidence)) + cls._decoded[cache_key] = (fingerprint, decoded) + # A sliding window makes a new selection value each time it moves, so bound the cache + # to the most recent (root, selection)s instead of the observer's whole lifetime. + while len(cls._decoded) > cls._decoded_limit: + del cls._decoded[next(iter(cls._decoded))] + return decoded + + @classmethod + def _claim_scan( + cls, connection: sqlite3.Connection, + keys: tuple[tuple[ThreadLookup, MessageReference], ...], privacy: RegistryPrivacy, + ) -> tuple[list[sqlite3.Row], object]: + """The raw claim-and-evidence rows for the keys, undecoded, with the cursor's column + description: the four-table join whose decode the cache reuses while these rows do not + move. The rows feed the fingerprint; rows+description feed joined() when it decodes.""" privacy.coordinator_schema().require_schema(connection) columns = ",".join(f"w.{name}" for name in WakeAssignment.columns()) context = privacy.notification_context_sql(connection) - rows = [] + raw: list[sqlite3.Row] = [] + description: object = () # Each batch's selected frozen keys select their assignments through the # recipient/seq index; message_id corroborates the exact original, and native-input # membership is selected once per batch by its assignment index. @@ -502,38 +562,26 @@ def capture( for lookup, reference in keys ), ): - rows.extend( - WakeAssignment.joined( - connection.execute( - f"WITH {requested.values}, " - f"w AS MATERIALIZED (SELECT * {requested.join}), " - f"context AS MATERIALIZED ({context}) " - f"SELECT {columns},context.triage_inflight," - "pointer.execution_id IS NOT NULL AS queued,context.context_committed_at_ms," - "p.seq AS response_seq, p.message_id AS response_message_id, " - "p.received_at_ms AS response_received_at_ms " - "FROM w " - "JOIN context ON context.assignment_id=w.assignment_id " - f"LEFT JOIN {CurrentExecutions.declared_name} pointer " - "ON pointer.owner_lookup=w.recipient_lookup " - f"LEFT JOIN {PublicationEvidence.declared_name} p " - "ON p.execution_id=w.execution_id AND p.exact_target=w.exact_target", - requested.bindings, - ), - AssignmentActivityEvidence, - ) + cursor = connection.execute( + f"WITH {requested.values}, " + f"w AS MATERIALIZED (SELECT * {requested.join}), " + f"context AS MATERIALIZED ({context}) " + f"SELECT {columns},context.triage_inflight," + "pointer.execution_id IS NOT NULL AS queued,context.context_committed_at_ms," + "p.seq AS response_seq, p.message_id AS response_message_id, " + "p.received_at_ms AS response_received_at_ms " + "FROM w " + "JOIN context ON context.assignment_id=w.assignment_id " + f"LEFT JOIN {CurrentExecutions.declared_name} pointer " + "ON pointer.owner_lookup=w.recipient_lookup " + f"LEFT JOIN {PublicationEvidence.declared_name} p " + "ON p.execution_id=w.execution_id AND p.exact_target=w.exact_target", + requested.bindings, ) - consumption: dict[ - tuple[ThreadLookup, MessageReference], tuple[ConsumptionEvidence, ...] - ] = {} - for attempt in HarnessDelivery.for_recipients(connection, keys): - key = (attempt.recipient_lookup, attempt.reference) - consumption[key] = (*consumption.get(key, ()), attempt.monitor) - return NotificationAssignmentRead( - selection, - tuple(sorted(rows, key=lambda row: (-row[0].wire_seq, row[0].recipient))), - MappingProxyType(consumption), - ) + description = cursor.description # one query shape, so every batch shares it + raw.extend(cursor.fetchall()) + cursor.close() + return raw, description @classmethod def read(cls, root: Path, selection: NotificationSelection) -> NotificationAssignmentRead: @@ -547,6 +595,7 @@ def missing_store( store: MissingStore, selection: NotificationSelection, privacy: RegistryPrivacy, + root: Path | None = None, ) -> NotificationAssignmentRead: return privacy.missing_notification_assignments(selection) @@ -556,6 +605,7 @@ def invalid_store( store: InvalidStore, selection: NotificationSelection, privacy: RegistryPrivacy, + root: Path | None = None, ) -> NoReturn: raise ValueError(f"Channel notification status unavailable: {store.failure_reason}") @@ -612,31 +662,11 @@ def read_assignments(self, selection: NotificationSelection) -> NotificationAssi held = self._current_held() if selection not in held.reads: with self.coordinator.observing( - NotificationAssignment, selection, self.privacy) as result: + NotificationAssignment, selection, self.privacy, self.root) as result: captured = result held.reads[selection] = NotificationAssignment.capture_detached(self, captured) return held.reads[selection] - def read_batch(self, selections: Iterable[NotificationSelection]) -> None: - """Warm the held cache for many windows' selections in as few coordinator reads as - possible. A notification pass checks every open thread's window on one revision; - reading them one window at a time re-stat'd every source, re-checked the schema and - took a coordinator round-trip each. One read per bounded batch covers them, and each - selection owns its slice (read_assignments then returns it without touching SQLite).""" - held = self._current_held() - pending = [selection for selection in dict.fromkeys(selections) - if selection.pages and selection not in held.reads] - batch: list[NotificationSelection] = [] - pages = 0 - for selection in pending: - if batch and pages + len(selection.pages) > NotificationSelection.window_limit: - self._read_combined(held, batch) - batch, pages = [], 0 - batch.append(selection) - pages += len(selection.pages) - if batch: - self._read_combined(held, batch) - def _current_held(self) -> HeldAssignmentReads: cut = tuple(FileRevision.capture(path) for path in NotificationAssignment.source_paths(self.root)) @@ -645,18 +675,6 @@ def _current_held(self) -> HeldAssignmentReads: held = AssignmentSource._held[self.root] = HeldAssignmentReads(cut, self.registry) return held - def _read_combined( - self, held: HeldAssignmentReads, selections: list[NotificationSelection], - ) -> None: - combined = NotificationSelection( - selections[0].viewer, tuple(page for s in selections for page in s.pages)) - with self.coordinator.observing( - NotificationAssignment, combined, self.privacy) as result: - captured = result - full = NotificationAssignment.capture_detached(self, captured) - for selection in selections: - held.reads[selection] = full.sliced(selection) - @dataclass class HeldAssignmentReads: diff --git a/src/agent_comms/ui_model/observation.py b/src/agent_comms/ui_model/observation.py index a3e58abc39..2dab32764a 100644 --- a/src/agent_comms/ui_model/observation.py +++ b/src/agent_comms/ui_model/observation.py @@ -1726,7 +1726,6 @@ def changed_threads(self, revision: WireRevision, comms: Comms, if not self.results[name].same_status(status_batch.view(name))) remaining = {name: view for name, view in readable.items() if name not in changed} notifications = NotificationPass(comms, revision) - notifications.prime((view.notification_window for view in remaining.values()), earlier) changed.update(name for name, view in remaining.items() if view.notification_window.activity_names() & activity_changed or notifications.changed(view.notification_window, earlier))