Reassemble chunked connector file results
Buffer JSON/base64 connector_file_op_result chunks per pending file operation and resolve only after the complete result is assembled. Validate malformed chunk metadata with a pending-operation error, while preserving the existing single-payload behavior for small file operations.
Alessandro committed
Jun 16, 2026 at 13:59 UTC
2615fb8e3f10d5405ba2557807f87c52e4212b80
2 files changed
+101
-1
plugins/_a0_connector/AGENTS.md
+3
@@ -21,6 +21,9 @@
21
execution metadata enables `code_execution_remote`, and supported enabled
22
Computer Use that does not need re-arming enables `computer_use_remote`.
23
- Do not bypass WebSocket authentication or leak connector session data.
24
+- File operation results may arrive as chunked JSON/base64
25
+ `connector_file_op_result` frames; resolve the pending file operation only
26
+ after all chunks for the `op_id` are assembled.
27
28
## Work Guidance
29
plugins/_a0_connector/helpers/ws_runtime.py
+98
-1
@@ -1,10 +1,13 @@
1
from __future__ import annotations
2
3
import asyncio
4
+import base64
5
+import binascii
6
import copy
7
+import json
8
import threading
9
import time
7
-from dataclasses import dataclass
10
+from dataclasses import dataclass, field
11
from typing import Any
12
13
@@ -14,6 +17,8 @@ class PendingFileOperation:
17
loop: asyncio.AbstractEventLoop
18
future: asyncio.Future[dict[str, Any]]
19
context_id: str | None = None
20
+ chunk_count: int | None = None
21
+ chunks: dict[int, bytes] = field(default_factory=dict)
22
23
24
@dataclass
@@ -570,9 +575,101 @@ def resolve_pending_file_op(
575
sid: str,
576
payload: dict[str, Any],
577
) -> bool:
578
+ if payload.get("chunked") is True:
579
+ return _resolve_pending_file_chunk(op_id, sid=sid, payload=payload)
580
return _resolve_pending(_pending_file_ops, op_id, sid=sid, payload=payload)
581
582
583
+def _resolve_pending_file_chunk(
584
+ op_id: str,
585
+ *,
586
+ sid: str,
587
+ payload: dict[str, Any],
588
+) -> bool:
589
+ error = _validate_file_chunk_payload(payload)
590
+ if error:
591
+ return _fail_pending(
592
+ _pending_file_ops,
593
+ op_id,
594
+ sid=sid,
595
+ error=f"Invalid chunked file operation result: {error}",
596
+ )
597
+
598
+ chunk_index = int(payload["chunk_index"])
599
+ chunk_count = int(payload["chunk_count"])
600
+ encoded = str(payload.get("data") or "")
601
+ try:
602
+ chunk = base64.b64decode(encoded.encode("ascii"), validate=True)
603
+ except (UnicodeEncodeError, binascii.Error) as exc:
604
+ return _fail_pending(
605
+ _pending_file_ops,
606
+ op_id,
607
+ sid=sid,
608
+ error=f"Invalid chunked file operation result: {exc}",
609
+ )
610
+
611
+ with _state_lock:
612
+ pending = _pending_file_ops.get(op_id)
613
+ if pending is None or pending.sid != sid:
614
+ return False
615
+
616
+ if pending.chunk_count is None:
617
+ pending.chunk_count = chunk_count
618
+ elif pending.chunk_count != chunk_count:
619
+ _pending_file_ops.pop(op_id, None)
620
+ pending.loop.call_soon_threadsafe(
621
+ _set_future_result,
622
+ pending.future,
623
+ {
624
+ "op_id": op_id,
625
+ "ok": False,
626
+ "error": "Invalid chunked file operation result: chunk_count changed",
627
+ },
628
+ )
629
+ return True
630
+
631
+ pending.chunks[chunk_index] = chunk
632
+ if len(pending.chunks) < chunk_count:
633
+ return True
634
+
635
+ ordered = [pending.chunks[index] for index in range(chunk_count)]
636
+ _pending_file_ops.pop(op_id, None)
637
+
638
+ try:
639
+ assembled = b"".join(ordered).decode("utf-8")
640
+ result = json.loads(assembled)
641
+ if not isinstance(result, dict):
642
+ raise ValueError("decoded result is not an object")
643
+ except Exception as exc:
644
+ result = {
645
+ "op_id": op_id,
646
+ "ok": False,
647
+ "error": f"Invalid chunked file operation result: {exc}",
648
+ }
649
+
650
+ pending.loop.call_soon_threadsafe(_set_future_result, pending.future, result)
651
+ return True
652
+
653
+
654
+def _validate_file_chunk_payload(payload: dict[str, Any]) -> str:
655
+ if payload.get("encoding") != "json+base64":
656
+ return "encoding must be json+base64"
657
+
658
+ try:
659
+ chunk_index = int(payload.get("chunk_index"))
660
+ chunk_count = int(payload.get("chunk_count"))
661
+ except (TypeError, ValueError):
662
+ return "chunk_index and chunk_count must be integers"
663
+
664
+ if chunk_count <= 0:
665
+ return "chunk_count must be positive"
666
+ if chunk_index < 0 or chunk_index >= chunk_count:
667
+ return "chunk_index out of range"
668
+ if not isinstance(payload.get("data"), str):
669
+ return "data must be a string"
670
+ return ""
671
+
672
+
673
def fail_pending_file_op(
674
op_id: str,
675
*,