From 4ceb6433aa4fe5054c1b45b221b17694ae2c397d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 14 Nov 2026 21:18:39 -0500 Subject: [PATCH] Revert "Observer caches decoded claim rows, so a ping decodes only what moved" This reverts commit 64d0e5661c6b023cca033d086c25f3f5bf98fb32. --- .../execution/selected/participant.py | 1 + src/agent_comms/notification_assignment.py | 67 +++++++++---------- tests/test_outbound_transcript_source.py | 5 +- 3 files changed, 33 insertions(+), 40 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/notification_assignment.py b/src/agent_comms/notification_assignment.py index 1106ebe744..a54c999b46 100644 --- a/src/agent_comms/notification_assignment.py +++ b/src/agent_comms/notification_assignment.py @@ -471,15 +471,13 @@ 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, evidence) per raw join row, reused while that row is byte-identical. - # Decoding a row (FieldCodec, ~a third of the observer's per-ping CPU) is a pure function of - # it, so a ping re-decodes only the rows that actually moved, not every open window's rows. - # Consumption, bus responses and stored inputs are never cached and are read fresh every - # pass, so nothing displayed can go stale. The key is the whole raw row, so any change to a - # claim, triage context, queue pointer or published response is a different key and re-decodes. - _decoded_row: ClassVar[dict[ - tuple, tuple[WakeAssignment, AssignmentActivityEvidence]]] = {} - _decoded_limit: ClassVar[int] = 4096 + # 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( @@ -487,6 +485,7 @@ def capture( connection: sqlite3.Connection, selection: NotificationSelection, privacy: RegistryPrivacy, + root: Path | None = None, ) -> NotificationAssignmentRead: """Probe only selected frozen keys before joining any claim/context/proof. @@ -498,7 +497,7 @@ def capture( keys = tuple(dict.fromkeys(selection.keys)) if not keys: return NotificationAssignmentRead(selection, ()) - rows = cls._decoded_claims(connection, keys, privacy) + rows = cls._decoded_claims(connection, selection, keys, privacy, root) consumption: dict[ tuple[ThreadLookup, MessageReference], tuple[ConsumptionEvidence, ...] ] = {} @@ -513,34 +512,28 @@ def capture( @classmethod def _decoded_claims( - cls, connection: sqlite3.Connection, + 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) - out: list[tuple[WakeAssignment, AssignmentActivityEvidence] | None] = [] - pending_rows: list[sqlite3.Row] = [] - pending_keys: list[tuple] = [] - pending_slots: list[int] = [] - for row in raw: - key = tuple(row) - hit = cls._decoded_row.pop(key, None) - if hit is not None: - cls._decoded_row[key] = hit # most-recently-used - out.append(hit) - else: - pending_slots.append(len(out)) - out.append(None) - pending_rows.append(row) - pending_keys.append(key) - if pending_rows: - fresh = WakeAssignment.joined( - _DecodedRows(pending_rows, description), AssignmentActivityEvidence) - for key, decoded, slot in zip(pending_keys, fresh, pending_slots, strict=True): - cls._decoded_row[key] = decoded - out[slot] = decoded - while len(cls._decoded_row) > cls._decoded_limit: - del cls._decoded_row[next(iter(cls._decoded_row))] - return tuple(out) + 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( @@ -602,6 +595,7 @@ def missing_store( store: MissingStore, selection: NotificationSelection, privacy: RegistryPrivacy, + root: Path | None = None, ) -> NotificationAssignmentRead: return privacy.missing_notification_assignments(selection) @@ -611,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}") @@ -667,7 +662,7 @@ 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] diff --git a/tests/test_outbound_transcript_source.py b/tests/test_outbound_transcript_source.py index 86efb6b54b..62d2c3ac2c 100644 --- a/tests/test_outbound_transcript_source.py +++ b/tests/test_outbound_transcript_source.py @@ -102,10 +102,7 @@ def count_decodes(frame, event, argument): own = NotificationAssignment.read(root, NotificationSelection.of(viewer, (initial,))) finally: sys.setprofile(profiling) - # The read decodes at most its own one row; the row cache may reuse a row already - # decoded for another viewer, so it never decodes more, and may decode none. own.rows - # below proves the viewer still gets exactly its own row. - assert decoded <= 1 + assert decoded == 1 assert own.selection.viewer == viewer and own.sources == (initial,) assert own.rows == tuple(row for row in everyone.rows if row[0].recipient == name) assert own.evidence[reference].claims == own.rows