Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1066,6 +1066,9 @@ Goal 与重放;源码验证不代表已安装宿主采用或整 Goal 资格通
使用同 UID 私有文件及字节数、SHA-256 摘要绑定;其他 effect 的默认传输边界不变。
文件缺失、改变、权限不符或超过 64 MiB 时拒绝。Handler 可能已经提交后,无法验证
的响应仍按不明确结果处理,要求精确回执回读,不能自动重试写入。
内联响应和私有文件响应都将此规则用于编码无效、JSON 值超限和嵌套过深的情况。
解码失败不能进入发送前的启动重试路径,也不能抹掉可能已经持久化的事实;
既有传输 adapter 将其报告为 `runtime_response_ambiguous`。

来源读取和 read/check 归约现在在一次 TypeScript 请求内完成。上下文读取与提交前
检查各从两次 checkpoint 请求降为一次,完整权威事实不再先返回 Python 再传回 TS。
Expand Down
5 changes: 3 additions & 2 deletions loopx/control_plane/effect_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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


Expand Down
91 changes: 91 additions & 0 deletions tests/control_plane/test_effect_runtime_response_decode.py
Original file line number Diff line number Diff line change
@@ -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
Loading