Page connector history replay
Bound _a0_connector context snapshots and live stream reads so long chats are replayed in small WebSocket frames instead of one oversized payload. Replay remaining historical entries as connector_context_snapshot pages before switching to live connector_context_event streaming, and preserve LogOutput.end cursor semantics throughout.
Alessandro committed
Jun 3, 2026 at 11:43 UTC
aff70f19bbc37a8c76ccc89672e5252015495c4e
3 files changed
+139
-25
plugins/AGENTS.md
+1
@@ -30,6 +30,7 @@
30
- `hooks.py` runs in the framework runtime. Explicitly target another runtime if a plugin must prepare the agent execution environment.
31
- `execute.py` is manual user-triggered setup, maintenance, repair, migration, or refresh work; automatic framework behavior belongs in hooks or lifecycle extensions.
32
- Plugin routes are `GET /plugins/<name>/<path>`, `POST /api/plugins/<name>/<handler>`, and `POST /api/plugins` for management actions.
33
+- `_a0_connector` WebSocket history replay must stay bounded: emit large chat history as paged `connector_context_snapshot` payloads, keep `last_sequence` as the Agent Zero log-output cursor, and avoid sending an entire long transcript in one frame.
34
- Frontend plugin HTML extensions live under `extensions/webui/<point>/`, include a root Alpine scope, and use `x-move-*` directives when targeting static breakpoints.
35
- Frontend plugin JS extensions live under `extensions/webui/<point>/` and export a default function.
36
- Plugin UI must use the A0 notification system for errors, warnings, success, and info instead of inline success/error boxes.
plugins/_a0_connector/api/ws_connector.py
+123
-24
@@ -62,6 +62,9 @@ WS_FEATURES = [
62
"connector_browser_op",
63
]
64
65
+_SNAPSHOT_REPLAY_PAGE_SIZE = 50
66
+_LIVE_STREAM_PAGE_SIZE = 100
67
+
68
69
class WsConnector(WsHandler):
70
_streaming_tasks: ClassVar[dict[tuple[str, str], asyncio.Task[None]]] = {}
@@ -253,19 +256,25 @@ class WsConnector(WsHandler):
256
)
257
258
subscribe_sid_to_context(sid, context_id)
256
- events, last_sequence = get_context_log_entries(context_id, after=from_sequence)
257
- await self.emit_to(
259
+ events, last_sequence = get_context_log_entries(
260
+ context_id,
261
+ after=from_sequence,
262
+ limit=_SNAPSHOT_REPLAY_PAGE_SIZE,
263
+ )
264
+ await self._emit_context_snapshot(
265
sid,
259
- "connector_context_snapshot",
260
- {
261
- "context_id": context_id,
262
- "events": events,
263
- "last_sequence": last_sequence,
264
- "message_queue": self._queue_items_for_context(context),
265
- },
266
+ context_id=context_id,
267
+ events=events,
268
+ last_sequence=last_sequence,
269
+ context=context,
270
correlation_id=data.get("correlationId"),
271
)
268
- self._start_streaming(sid, context_id, from_sequence=last_sequence)
272
+ self._start_streaming(
273
+ sid,
274
+ context_id,
275
+ from_sequence=last_sequence,
276
+ replay_history=True,
277
+ )
278
279
return {
280
"context_id": context_id,
@@ -349,19 +358,25 @@ class WsConnector(WsHandler):
358
359
if context_id not in subscribed_contexts_for_sid(sid):
360
subscribe_sid_to_context(sid, context_id)
352
- events, last_sequence = get_context_log_entries(context_id, after=0)
353
- await self.emit_to(
361
+ events, last_sequence = get_context_log_entries(
362
+ context_id,
363
+ after=0,
364
+ limit=_SNAPSHOT_REPLAY_PAGE_SIZE,
365
+ )
366
+ await self._emit_context_snapshot(
367
sid,
355
- "connector_context_snapshot",
356
- {
357
- "context_id": context_id,
358
- "events": events,
359
- "last_sequence": last_sequence,
360
- "message_queue": self._queue_items_for_context(context),
361
- },
368
+ context_id=context_id,
369
+ events=events,
370
+ last_sequence=last_sequence,
371
+ context=context,
372
correlation_id=data.get("correlationId"),
373
)
364
- self._start_streaming(sid, context_id, from_sequence=last_sequence)
374
+ self._start_streaming(
375
+ sid,
376
+ context_id,
377
+ from_sequence=last_sequence,
378
+ replay_history=True,
379
+ )
380
381
message_id = client_message_id or data.get("correlationId") or ""
382
context.log.log(
@@ -865,14 +880,48 @@ class WsConnector(WsHandler):
880
f"[a0-connector] failed to emit connector_context_complete to {target_sid}: {exc}"
881
)
882
868
- def _start_streaming(self, sid: str, context_id: str, *, from_sequence: int) -> None:
883
+ async def _emit_context_snapshot(
884
+ self,
885
+ sid: str,
886
+ *,
887
+ context_id: str,
888
+ events: list[dict[str, Any]],
889
+ last_sequence: int,
890
+ context: AgentContext | None = None,
891
+ correlation_id: str | None = None,
892
+ ) -> None:
893
+ await self.emit_to(
894
+ sid,
895
+ "connector_context_snapshot",
896
+ {
897
+ "context_id": context_id,
898
+ "events": events,
899
+ "last_sequence": last_sequence,
900
+ "message_queue": self._queue_items_for_context(context),
901
+ },
902
+ correlation_id=correlation_id,
903
+ )
904
+
905
+ def _start_streaming(
906
+ self,
907
+ sid: str,
908
+ context_id: str,
909
+ *,
910
+ from_sequence: int,
911
+ replay_history: bool = False,
912
+ ) -> None:
913
key = (sid, context_id)
914
task = self._streaming_tasks.get(key)
915
if task is not None and not task.done():
916
return
917
918
task = asyncio.create_task(
875
- self._stream_events(sid, context_id, from_sequence=from_sequence)
919
+ self._stream_events(
920
+ sid,
921
+ context_id,
922
+ from_sequence=from_sequence,
923
+ replay_history=replay_history,
924
+ )
925
)
926
self._streaming_tasks[key] = task
927
@@ -887,14 +936,26 @@ class WsConnector(WsHandler):
936
context_id: str,
937
*,
938
from_sequence: int,
939
+ replay_history: bool = False,
940
) -> None:
941
# `from_sequence` is a log-output cursor (not an event sequence number).
942
cursor = max(int(from_sequence or 0), 0)
943
last_queue_signature, _ = self._queue_state_for_context_id(context_id)
944
was_running = self._context_is_running(context_id)
945
try:
946
+ if replay_history:
947
+ cursor = await self._replay_history_snapshots(
948
+ sid,
949
+ context_id,
950
+ from_sequence=cursor,
951
+ )
952
+
953
while context_id in subscribed_contexts_for_sid(sid):
897
- events, next_cursor = get_context_log_entries(context_id, after=cursor)
954
+ events, next_cursor = get_context_log_entries(
955
+ context_id,
956
+ after=cursor,
957
+ limit=_LIVE_STREAM_PAGE_SIZE,
958
+ )
959
for event in events:
960
await self.emit_to(sid, "connector_context_event", event)
961
cursor = max(cursor, int(next_cursor or cursor))
@@ -920,7 +981,7 @@ class WsConnector(WsHandler):
981
},
982
)
983
was_running = is_running
923
- await asyncio.sleep(0.5)
984
+ await asyncio.sleep(0 if events else 0.5)
985
except asyncio.CancelledError:
986
raise
987
except Exception as exc:
@@ -929,3 +990,41 @@ class WsConnector(WsHandler):
990
)
991
finally:
992
self._streaming_tasks.pop((sid, context_id), None)
993
+
994
+ async def _replay_history_snapshots(
995
+ self,
996
+ sid: str,
997
+ context_id: str,
998
+ *,
999
+ from_sequence: int,
1000
+ ) -> int:
1001
+ cursor = max(int(from_sequence or 0), 0)
1002
+
1003
+ while context_id in subscribed_contexts_for_sid(sid):
1004
+ events, next_cursor = get_context_log_entries(
1005
+ context_id,
1006
+ after=cursor,
1007
+ limit=_SNAPSHOT_REPLAY_PAGE_SIZE,
1008
+ )
1009
+ next_cursor = max(cursor, int(next_cursor or cursor))
1010
+ if not events:
1011
+ return next_cursor
1012
+
1013
+ _, queue_items = self._queue_state_for_context_id(context_id)
1014
+ await self.emit_to(
1015
+ sid,
1016
+ "connector_context_snapshot",
1017
+ {
1018
+ "context_id": context_id,
1019
+ "events": events,
1020
+ "last_sequence": next_cursor,
1021
+ "message_queue": queue_items,
1022
+ },
1023
+ )
1024
+
1025
+ if next_cursor == cursor:
1026
+ return cursor + len(events)
1027
+ cursor = next_cursor
1028
+ await asyncio.sleep(0)
1029
+
1030
+ return cursor
plugins/_a0_connector/helpers/event_bridge.py
+15
-1
@@ -79,6 +79,7 @@ def log_entry_to_connector_event(
79
def get_context_log_entries(
80
context_id: str,
81
after: int = 0,
82
+ limit: int | None = None,
83
) -> tuple[list[dict[str, Any]], int]:
84
"""Return connector events plus the next log cursor for the context."""
85
try:
@@ -88,7 +89,20 @@ def get_context_log_entries(
89
if context is None:
90
return [], 0
91
91
- log_output = context.log.output(start=max(int(after or 0), 0))
92
+ start = max(int(after or 0), 0)
93
+ end: int | None = None
94
+ if limit is not None:
95
+ limit = max(int(limit or 0), 0)
96
+ if limit > 0:
97
+ log_lock = getattr(context.log, "_lock", None)
98
+ log_updates = getattr(context.log, "updates", None)
99
+ if log_lock is not None and isinstance(log_updates, list):
100
+ with log_lock:
101
+ end = min(start + limit, len(log_updates))
102
+ else:
103
+ end = start + limit
104
+
105
+ log_output = context.log.output(start=start, end=end)
106
events = [
107
log_entry_to_connector_event(entry, context_id)
108
for entry in log_output.items