From 64d0e5661c6b023cca033d086c25f3f5bf98fb32 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 14 Nov 2026 21:08:32 -0500 Subject: [PATCH] Observer caches decoded claim rows, so a ping decodes only what moved #4777 cached a whole window's decode, keyed by (root, selection); it removed the per-thread fan-out but still re-decoded the one window a ping changes in full, leaving the observer child at ~5.1% median per ping (at the 5% target, not under). Cache the decode at the row instead: a decoded (claim, evidence) keyed by its raw join row. Decoding a row is a pure function of it, so a ping re-decodes only the row or two that actually moved; every unchanged row across every window reuses its decode. This subsumes the window cache (and drops the root threading it needed): the key is the whole raw row, so a change to any of claim, triage context, queue pointer or published response is a different key and re-decodes. Consumption, bus responses and stored inputs are still read fresh each pass, so nothing displayed goes stale. Bounded to the most recent rows (LRU). One test loosened: test_a_read_selects_only_its_viewers_claims_and_attempts_before_decoding asserted exactly one _decode_row call per viewer read; with a shared row cache the viewer's one row is already decoded (the test's prior AllOutcomes read warmed it), so it decodes none. Loosened `== 1` to `<= 1`, which still catches a read that decodes more than its own rows; the `own.rows ==` assertion below already proves the viewer gets exactly its own row. Notification + observation suite: 112 passed, zero added over main's baseline of 3. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01XbFMSnLEc51gw21bLFFRK1 --- .../execution/selected/participant.py | 1 - src/agent_comms/notification_assignment.py | 67 ++++++++++--------- tests/test_outbound_transcript_source.py | 5 +- 3 files changed, 40 insertions(+), 33 deletions(-) diff --git a/src/agent_comms/execution/selected/participant.py b/src/agent_comms/execution/selected/participant.py index 83059e705a..8a10486bab 100644 --- a/src/agent_comms/execution/selected/participant.py +++ b/src/agent_comms/execution/selected/participant.py @@ -473,7 +473,6 @@ 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 a54c999b46..1106ebe744 100644 --- a/src/agent_comms/notification_assignment.py +++ b/src/agent_comms/notification_assignment.py @@ -471,13 +471,15 @@ 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 + # 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 @classmethod def capture( @@ -485,7 +487,6 @@ 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. @@ -497,7 +498,7 @@ def capture( keys = tuple(dict.fromkeys(selection.keys)) if not keys: return NotificationAssignmentRead(selection, ()) - rows = cls._decoded_claims(connection, selection, keys, privacy, root) + rows = cls._decoded_claims(connection, keys, privacy) consumption: dict[ tuple[ThreadLookup, MessageReference], tuple[ConsumptionEvidence, ...] ] = {} @@ -512,28 +513,34 @@ def capture( @classmethod def _decoded_claims( - cls, connection: sqlite3.Connection, selection: NotificationSelection, + cls, connection: sqlite3.Connection, 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 + 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) @classmethod def _claim_scan( @@ -595,7 +602,6 @@ def missing_store( store: MissingStore, selection: NotificationSelection, privacy: RegistryPrivacy, - root: Path | None = None, ) -> NotificationAssignmentRead: return privacy.missing_notification_assignments(selection) @@ -605,7 +611,6 @@ def invalid_store( store: InvalidStore, selection: NotificationSelection, privacy: RegistryPrivacy, - root: Path | None = None, ) -> NoReturn: raise ValueError(f"Channel notification status unavailable: {store.failure_reason}") @@ -662,7 +667,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, self.root) as result: + NotificationAssignment, selection, self.privacy) 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 62d2c3ac2c..86efb6b54b 100644 --- a/tests/test_outbound_transcript_source.py +++ b/tests/test_outbound_transcript_source.py @@ -102,7 +102,10 @@ def count_decodes(frame, event, argument): own = NotificationAssignment.read(root, NotificationSelection.of(viewer, (initial,))) finally: sys.setprofile(profiling) - assert decoded == 1 + # 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 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