diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 44368ff77e..9339c914a8 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1518,6 +1518,10 @@ the transport neither truncates history nor creates a Python decision or new authority. Missing, changed, non-private or over-64-MiB files fail closed. After a handler may have committed, an unverifiable response stays ambiguous and requires exact receipt readback, never automatic mutation retry. +Both inline and private-file responses apply this rule to invalid encoding, +JSON value limits and excessive nesting. A decoder failure cannot enter the +pre-send startup retry path or erase possible durable acceptance; it reports +`runtime_response_ambiguous` through the existing transport adapter. Source read and read/check reduction now compose inside one TypeScript request. Both context reading and commit preflight use one checkpoint request instead of diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 3dee7c57d9..652db16f6c 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -1066,6 +1066,9 @@ Goal 与重放；源码验证不代表已安装宿主采用或整 Goal 资格通 使用同 UID 私有文件及字节数、SHA-256 摘要绑定；其他 effect 的默认传输边界不变。 文件缺失、改变、权限不符或超过 64 MiB 时拒绝。Handler 可能已经提交后，无法验证 的响应仍按不明确结果处理，要求精确回执回读，不能自动重试写入。 +内联响应和私有文件响应都将此规则用于编码无效、JSON 值超限和嵌套过深的情况。 +解码失败不能进入发送前的启动重试路径，也不能抹掉可能已经持久化的事实； +既有传输 adapter 将其报告为 `runtime_response_ambiguous`。 来源读取和 read/check 归约现在在一次 TypeScript 请求内完成。上下文读取与提交前 检查各从两次 checkpoint 请求降为一次，完整权威事实不再先返回 Python 再传回 TS。 diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 8f4e59e89b..8201eef0d0 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -786,7 +786,8 @@ def _request_with_info( raise EffectRuntimeResponseAmbiguous(method, timeout=timeout) from exc try: response = json.loads(b"".join(chunks).split(b"\n", 1)[0]) - except (json.JSONDecodeError, IndexError): + except (ValueError, RecursionError): + # Decoder limits can fail after commit; they cannot authorize a retry. raise EffectRuntimeResponseAmbiguous(method, timeout=timeout) from None if isinstance(response, dict) and "result_ref" in response: response = _read_local_snapshot_response( @@ -832,7 +833,7 @@ def _read_local_snapshot_response( if not isinstance(result, dict): raise ValueError("invalid local snapshot response") return result - except (OSError, ValueError, json.JSONDecodeError) as exc: + except (OSError, ValueError, RecursionError) as exc: raise EffectRuntimeResponseAmbiguous(method, timeout=timeout) from exc diff --git a/tests/control_plane/test_effect_runtime_response_decode.py b/tests/control_plane/test_effect_runtime_response_decode.py new file mode 100644 index 0000000000..4fd6ab6a85 --- /dev/null +++ b/tests/control_plane/test_effect_runtime_response_decode.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import hashlib +import json +import os +import socket +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Event + +import pytest + +from loopx.control_plane import effect_runtime + + +@pytest.mark.parametrize("snapshot", [False, True], ids=["inline", "snapshot"]) +@pytest.mark.parametrize( + "result_bytes", + [b"[" * 100_000 + b"0" + b"]" * 100_000, b'"\xff"'], + ids=["excessive-depth", "invalid-utf8"], +) +def test_undecodable_response_is_ambiguous_without_redispatch( + tmp_path: Path, monkeypatch, snapshot: bool, result_bytes: bytes, +) -> None: + monkeypatch.setattr(effect_runtime, "_runtime_dir", lambda: tmp_path) + fingerprint = effect_runtime._runtime_fingerprint_for_request() + finished = Event() + received_requests = [] + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + listener.listen() + listener.settimeout(0.05) + info = { + "schema_version": effect_runtime.EFFECT_RUNTIME_INFO_SCHEMA_VERSION, + "fingerprint": fingerprint, + "pid": os.getpid(), + "host": "127.0.0.1", + "port": listener.getsockname()[1], + "token": "fixture-token", + } + locator = effect_runtime._runtime_info_path(fingerprint) + locator.write_text(json.dumps(info), encoding="utf-8") + + def serve_responses() -> None: + while not finished.is_set(): + try: + connection = listener.accept()[0] + except TimeoutError: + continue + with connection: + connection.settimeout(2) + request_bytes = b"" + while b"\n" not in request_bytes: + chunk = connection.recv(65536) + if not chunk: + raise AssertionError("request ended before its frame") + request_bytes += chunk + request = json.loads(request_bytes) + received_requests.append(request) + envelope = { + "schema_version": effect_runtime.EFFECT_RUNTIME_RESPONSE_SCHEMA_VERSION, + "request_id": request["request_id"], + "ok": True, + } + response = json.dumps(envelope).encode()[:-1] + b',"result":' + result_bytes + b"}\n" + if snapshot: + Path(request["response_sink"]).write_bytes(response) + envelope["result_ref"] = { + "byte_count": len(response), + "sha256": hashlib.sha256(response).hexdigest(), + } + response = json.dumps(envelope).encode() + b"\n" + connection.sendall(response) + + with ThreadPoolExecutor(max_workers=1) as executor: + server = executor.submit(serve_responses) + try: + with pytest.raises(effect_runtime.EffectRuntimeResponseAmbiguous) as error: + effect_runtime.effect_runtime_request( + "goal.checkpoint_read_context.commit" if snapshot else "turn_journal.write", + {"operation_id": "fixture-operation"}, + timeout=2, + large_local_snapshot=snapshot, + ) + finally: + finished.set() + server.result(timeout=3) + + assert error.value.diagnostic_code == "runtime_response_ambiguous" + assert len(received_requests) == 1 + assert json.loads(locator.read_text(encoding="utf-8")) == info