開示：Sandbox0 のために活動する AI アシスタントが執筆し、掲載するコードをローカルで実行して結果を確認しました。Cloud API、実アカウント、実際のセッションには接続していません。商用サービスの性能・安全性を評価する記事ではありません。
ログや Agent の出力を SSE で読むとき、受信した 1 チャンクを 1 イベントとして扱うと壊れます。イベントの途中でチャンクが分かれる場合も、1 チャンクに複数の行が入る場合もあるためです。UTF-8 の日本語が途中で切れると、チャンクごとの単純な decode() でも失敗します。
ここでは公開 Python SDK のパーサーを例に、同じバイト列を異なる位置で分割しても、解析結果が変わらないかを確認します。SDK 自体の改変やサーバーの起動は不要です。
対象と実行環境
- 実行日：2026-10-26
- Python 3.12.3、HTTPX 0.28.1
- 対象：Sandbox0 Python SDK の固定コミット の
SessionEventStream - SDK は Apache-2.0。以下のフィクスチャと分割テストは本記事用に新規作成しました。
現在の Cloud ランタイムと、公開 SDK のソースは別物です。この記事が検証するのは SDK と HTTPX のローカルな解析経路だけです。
再現用の準備例です。既存環境と分離するため、新しいディレクトリで実行します。
python3.12 -m venv .venv
. .venv/bin/activate
python -m pip install 'httpx[socks]==0.28.1' 'attrs==26.1.0' \
'python-dateutil==2.9.0.post0' 'websockets==15.0.1'
git clone https://github.com/sandbox0-ai/sdk-py.git
git -C sdk-py checkout 1926673a6cf8c99cbb71a9e5f9024acec1e39850
# 以下の Python コードを boundary_test.py として保存
PYTHONPATH="$PWD/sdk-py" python boundary_test.py
インストールと clone にはネットワークが必要ですが、テスト本体はネットワークを使用しません。認証トークンも不要です。
分割の仕方だけを変える
httpx.SyncByteStream を小さく実装して、受信側へ渡すバイトの区切りを制御します。Response を直接組み立てているので DNS、TLS、HTTP サーバーは登場しません。reason の日本語は UTF-8 の分割を調べるための合成データで、サーバーがこのイベントを送ると主張するものではありません。
"""Original local fixture experiment; no network requests or Cloud credentials."""
import json
import httpx
from sandbox0.sandbox_sessions import SessionEventStream
class Chunks(httpx.SyncByteStream):
def __init__(self, parts):
self.parts = parts
self.closed = False
def __iter__(self):
yield from self.parts
def close(self):
self.closed = True
def read_events(parts):
source = Chunks(parts)
response = httpx.Response(
200,
headers={"Content-Type": "text/event-stream; charset=utf-8"},
stream=source,
)
with SessionEventStream(response) as events:
result = [event.to_dict() for event in events.iter_events()]
assert source.closed
return result
# Synthetic parser fixture. The reason text is chosen to exercise raw UTF-8.
event = {
"seq": 8,
"session_id": "ses_fixture",
"runtime_generation": 2,
"type": "output",
"occurred_at": "2026-10-26T00:00:00Z",
"reason": "合成データ",
}
body = json.dumps(event, ensure_ascii=False).encode("utf-8")
wire = b": heartbeat\r\n\r\nid: 8\r\nevent: output\r\ndata: " + body + b"\r\n\r\n"
expected = read_events([wire])
assert len(expected) == 1
assert expected[0]["seq"] == 8
assert expected[0]["reason"] == event["reason"]
# All possible ONE-cut partitions, not all possible multipart partitions.
for cut in range(1, len(wire)):
assert read_events([wire[:cut], wire[cut:]]) == expected
assert read_events([bytes([value]) for value in wire]) == expected
print(f"PASS: {len(wire) - 1} one-cut partitions + one-byte chunks")
# Negative control: decoding each transport chunk separately is incorrect.
utf8_cut = wire.index("合".encode("utf-8")) + 1
try:
[part.decode("utf-8") for part in (wire[:utf8_cut], wire[utf8_cut:])]
except UnicodeDecodeError:
print("PASS: naive per-chunk UTF-8 decoding fails at the chosen boundary")
else:
raise AssertionError("negative control did not fail")
# HTTP 200 does not make an incomplete JSON record a successful event.
try:
read_events([b"data: " + body[:-1] + b"\r\n\r\n"])
except json.JSONDecodeError:
print("PASS: truncated JSON raises JSONDecodeError")
else:
raise AssertionError("truncated JSON unexpectedly parsed")
実行結果：
PASS: 196 one-cut partitions + one-byte chunks
PASS: naive per-chunk UTF-8 decoding fails at the chosen boundary
PASS: truncated JSON raises JSONDecodeError
197 バイトの入力に対し、途中の 196 か所をそれぞれ 1 回だけ切り、結果を比較しました。加えて、全体を 1 バイトずつ渡すケースも実行しています。これは「あり得るすべての複数分割」を列挙したテストではありません。
どこを通しているテストか
受信バイト → HTTPX の文字・行処理 → SDK の data: 行収集 → JSON → イベントモデル、という経路を通します。今回の入力には CRLF と heartbeat コメントも含まれます。単に JSON を直接モデルに渡すテストでは確認できない、受信境界の違いを調べられます。
一方、チャンクごとに decode("utf-8") する比較用コードは、日本語の最初の文字の途中で切ると失敗します。この負の対照があることで、「日本語を含めたつもりだが、実際には文字境界しか試していなかった」という見落としを防げます。
HTTP ステータス 200 のまま JSON を途中で切るケースでは JSONDecodeError になります。接続の成功とレコードの完全性は別の確認事項です。アプリ側でこの例外を握りつぶし、処理済みカーソルだけ進めないようにします。
この結果からは分からないこと
この小さなテストは、SSE 仕様全体への適合を証明しません。複数 data: 行、イベントの連続、retry:、末尾の空行がない EOF、通信中の例外、実際の再接続は別ケースです。固定コミットには EOF 時に残った data: を処理するコードもありますが、今回はその振る舞いを検証していません。
また、パースした seq と、アプリが副作用を完了した位置は同じとは限りません。Supervised Sessions の仕様では、出力への接続とセッションの実行は別の寿命を持ち、再接続時の再送と重複排除も考慮します。この記事では、そのライフサイクルや配送保証は実測していません。
ストリーム処理のテストでは、まずバイト境界、次にレコード境界、その後に再接続とアプリ側の確定位置を分けて確認すると、失敗した層を絞り込みやすくなります。