feat(connector): page tail history snapshots
Serve a bounded tail snapshot and cursor-based older pages for interactive clients. Preserve complete replay for clients that do not request history hints.
Alessandro committed
Jul 23, 2026 at 03:43 UTC
1831c7b47e6d74fca5e6f748fe47bce202e20e8f
2 files changed
+107
-7
plugins/_a0_connector/api/ws_connector.py
+81
-7
@@ -9,7 +9,10 @@ from helpers.ws import WsHandler
9
from helpers.ws_manager import WsResult
10
11
from plugins._a0_connector.helpers.exec_config import build_exec_config
12
-from plugins._a0_connector.helpers.event_bridge import get_context_log_entries
12
+from plugins._a0_connector.helpers.event_bridge import (
13
+ get_context_log_entries,
14
+ get_context_log_entry_count,
15
+)
16
from plugins._a0_connector.helpers.version import agent_zero_version
17
from plugins._a0_connector.helpers.ws_runtime import (
18
clear_remote_tree_snapshot,
@@ -68,6 +71,7 @@ WS_FEATURES = [
71
]
72
73
_SNAPSHOT_REPLAY_PAGE_SIZE = 50
74
+_TAIL_HISTORY_PAGE_SIZE = 100
75
_LIVE_STREAM_PAGE_SIZE = 100
76
77
@@ -257,6 +261,8 @@ class WsConnector(WsHandler):
261
262
context_id = str(data.get("context_id", "")).strip()
263
from_sequence = int(data.get("from", 0) or 0)
264
+ history_mode = str(data.get("history", "")).strip().lower()
265
+ history_before = data.get("history_before")
266
267
if not context_id:
268
return WsResult.error(
@@ -274,6 +280,67 @@ class WsConnector(WsHandler):
280
)
281
282
subscribe_sid_to_context(sid, context_id)
283
+
284
+ if history_before is not None:
285
+ before = min(
286
+ max(int(history_before or 0), 0),
287
+ get_context_log_entry_count(context_id),
288
+ )
289
+ start = max(before - _TAIL_HISTORY_PAGE_SIZE, 0)
290
+ events, last_sequence = get_context_log_entries(
291
+ context_id,
292
+ after=start,
293
+ limit=before - start,
294
+ )
295
+ await self._emit_context_snapshot(
296
+ sid,
297
+ context_id=context_id,
298
+ events=events,
299
+ last_sequence=last_sequence,
300
+ context=context,
301
+ correlation_id=data.get("correlationId"),
302
+ history_before=start,
303
+ has_more_history=bool(start),
304
+ )
305
+ return {
306
+ "context_id": context_id,
307
+ "subscribed": True,
308
+ "last_sequence": last_sequence,
309
+ "history_before": start,
310
+ "has_more_history": bool(start),
311
+ }
312
+
313
+ if history_mode == "tail":
314
+ total = get_context_log_entry_count(context_id)
315
+ start = max(total - _TAIL_HISTORY_PAGE_SIZE, 0)
316
+ events, last_sequence = get_context_log_entries(
317
+ context_id,
318
+ after=start,
319
+ limit=_TAIL_HISTORY_PAGE_SIZE,
320
+ )
321
+ await self._emit_context_snapshot(
322
+ sid,
323
+ context_id=context_id,
324
+ events=events,
325
+ last_sequence=last_sequence,
326
+ context=context,
327
+ correlation_id=data.get("correlationId"),
328
+ history_before=start,
329
+ has_more_history=bool(start),
330
+ )
331
+ self._start_streaming(
332
+ sid,
333
+ context_id,
334
+ from_sequence=last_sequence,
335
+ )
336
+ return {
337
+ "context_id": context_id,
338
+ "subscribed": True,
339
+ "last_sequence": last_sequence,
340
+ "history_before": start,
341
+ "has_more_history": bool(start),
342
+ }
343
+
344
events, last_sequence = get_context_log_entries(
345
context_id,
346
after=from_sequence,
@@ -927,16 +994,23 @@ class WsConnector(WsHandler):
994
last_sequence: int,
995
context: AgentContext | None = None,
996
correlation_id: str | None = None,
997
+ history_before: int | None = None,
998
+ has_more_history: bool | None = None,
999
) -> None:
1000
+ payload: dict[str, Any] = {
1001
+ "context_id": context_id,
1002
+ "events": events,
1003
+ "last_sequence": last_sequence,
1004
+ "message_queue": self._queue_items_for_context(context),
1005
+ }
1006
+ if history_before is not None:
1007
+ payload["history_before"] = history_before
1008
+ if has_more_history is not None:
1009
+ payload["has_more_history"] = has_more_history
1010
await self.emit_to(
1011
sid,
1012
"connector_context_snapshot",
934
- {
935
- "context_id": context_id,
936
- "events": events,
937
- "last_sequence": last_sequence,
938
- "message_queue": self._queue_items_for_context(context),
939
- },
1013
+ payload,
1014
correlation_id=correlation_id,
1015
)
1016
plugins/_a0_connector/helpers/event_bridge.py
+26
@@ -116,6 +116,32 @@ def get_context_log_entries(
116
return [], max(int(after or 0), 0)
117
118
119
+def get_context_log_entry_count(context_id: str) -> int:
120
+ """Return the current log-output cursor for a context."""
121
+ try:
122
+ from agent import AgentContext
123
+
124
+ context = AgentContext.get(context_id)
125
+ if context is None:
126
+ return 0
127
+
128
+ log = context.log
129
+ log_lock = getattr(log, "_lock", None)
130
+ updates = getattr(log, "updates", None)
131
+ if isinstance(updates, list):
132
+ if log_lock is not None:
133
+ with log_lock:
134
+ return len(updates)
135
+ return len(updates)
136
+
137
+ return int(log.output().end)
138
+ except Exception as exc:
139
+ PrintStyle.error(
140
+ f"[a0-connector] event_bridge cursor error for context {context_id}: {exc}"
141
+ )
142
+ return 0
143
+
144
+
145
async def stream_context_events(
146
context_id: str,
147
from_sequence: int = 0,