From b1a08df99ee095b31bd74cd8262210c4b08539e6 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Sun, 4 Oct 2026 14:02:17 +0800 Subject: [PATCH 1/3] fix(runtime): await locator publication before admitting effects Signed-off-by: huangruiteng --- loopx/control_plane/effect_runtime_server.ts | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts index ceaf9a4a69..9d435f20f8 100644 --- a/loopx/control_plane/effect_runtime_server.ts +++ b/loopx/control_plane/effect_runtime_server.ts @@ -109,6 +109,9 @@ const idleMs = parseIdleMs(process.env.LOOPX_EFFECT_RUNTIME_IDLE_MS); let idleTimer: NodeJS.Timeout; let pendingRequests = 0; let resolveDrain: (() => void) | undefined; +let publicationComplete = false; +let resolvePublication: () => void; +const publicationReady = new Promise((resolve) => { resolvePublication = resolve; }); const handlers = createEffectRuntimeHandlers({ fingerprint, requestShutdown: () => { @@ -119,7 +122,7 @@ const handlers = createEffectRuntimeHandlers({ function resetIdleTimer(server: ReturnType): void { clearTimeout(idleTimer); // Idleness starts after effects finish, not when their sockets connect. - if (pendingRequests > 0 || shutdownRequested || !server.listening) return; + if (!publicationComplete || pendingRequests > 0 || shutdownRequested || !server.listening) return; idleTimer = setTimeout(() => server.close(), idleMs); idleTimer.unref(); } @@ -178,6 +181,10 @@ const server = createServer((socket) => { let sink: FileHandle | null = null; let dispatched = false; try { + // The locator becomes visible inside the publication lock. A first + // response must wait for its release: otherwise an immediate process + // death can strand an incomplete cleanup claim before the next write. + await publicationReady; let parsed: unknown; try { parsed = JSON.parse( @@ -299,5 +306,7 @@ server.listen(0, "127.0.0.1", async () => { }); await chmod(infoPath, 0o600); }); + publicationComplete = true; + resolvePublication(); resetIdleTimer(server); }); From c51d3886dfc9d99d2af57ff0c6ea356595375fb4 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Sun, 4 Oct 2026 14:02:17 +0800 Subject: [PATCH 2/3] docs(runtime): record managed publication readiness boundary Signed-off-by: huangruiteng --- .../rfcs/typescript-control-plane-migration-v0.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index b53a0c467e..69dfd3c98b 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -1354,6 +1354,15 @@ If an authority daemon owns a registry/workspace, a CLI process must connect to it instead of opening a second direct writer. Runtime discovery and startup are automatic; users do not configure ports or supervise processes. +For the managed loopback runtime, a visible locator is discovery evidence; +request dispatch and successful replies wait until locator publication and its +awaited lock cleanup finish. Otherwise an immediate exit after the first reply +can leave an incomplete cleanup claim and prevent the next retry-safe write +from restarting within the existing lock budget. The real-Node publication +regression covers first ping and typed write, then abrupt exit and receipt +replay. This repair preserves lock reclaim ages, startup deadlines and retry +classification; broader process/storage recovery remains separately qualified. + ### 2.3 TypeScript owns migrated effects The target is not “TypeScript decides, Python always executes”. TypeScript may From df90e0f090f60406bd3674ea444c6fabe0d14935 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Sun, 4 Oct 2026 14:02:17 +0800 Subject: [PATCH 3/3] test(runtime): fence first request during native publication cleanup Signed-off-by: huangruiteng --- .../test_effect_runtime_publication_ready.py | 120 ++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 tests/control_plane/test_effect_runtime_publication_ready.py diff --git a/tests/control_plane/test_effect_runtime_publication_ready.py b/tests/control_plane/test_effect_runtime_publication_ready.py new file mode 100644 index 0000000000..9e71eb2c66 --- /dev/null +++ b/tests/control_plane/test_effect_runtime_publication_ready.py @@ -0,0 +1,120 @@ +"""A visible locator must not admit effects before its publication settles.""" +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +import json +import os +from pathlib import Path +import signal +import time + +import pytest + +from loopx.control_plane import effect_runtime + + +@pytest.mark.parametrize("method", ["runtime.ping", "turn_journal.write"]) +def test_first_request_waits_for_real_publication_cleanup_and_recovers( + tmp_path: Path, monkeypatch, request: pytest.FixtureRequest, method: str, +) -> None: + runtime_dir = tmp_path / "runtime" + opened = tmp_path / "claim-opened" + release = tmp_path / "release" + preload = tmp_path / "pause-publication.mjs" + # Pause the real Node file handle after exclusive claim creation, before + # owner bytes or lock cleanup. No mocked transport, dispatch or lock owner. + preload.write_text("""import fs from 'node:fs'; +import {syncBuiltinESMExports} from 'node:module'; +const original = fs.promises.open; +fs.promises.open = async function(path, ...args) { + const handle = await original.call(this, path, ...args); + if (String(path).includes('.ts-effect.lock.claim.') && + String(path).includes('runtime-') && + !fs.existsSync(process.env.LOOPX_TEST_PUBLICATION_OPENED)) { + fs.writeFileSync(process.env.LOOPX_TEST_PUBLICATION_OPENED, 'opened'); + while (!fs.existsSync(process.env.LOOPX_TEST_PUBLICATION_RELEASE)) { + await new Promise(resolve => setTimeout(resolve, 5)); + } + } + return handle; +}; +syncBuiltinESMExports(); +""", encoding="utf-8") + monkeypatch.setattr(effect_runtime, "_runtime_dir", lambda: runtime_dir) + request.addfinalizer(effect_runtime.restart_effect_runtime) + monkeypatch.setenv("NODE_OPTIONS", f"--import={preload.as_uri()}") + monkeypatch.setenv("LOOPX_TEST_PUBLICATION_OPENED", str(opened)) + monkeypatch.setenv("LOOPX_TEST_PUBLICATION_RELEASE", str(release)) + monkeypatch.setenv("LOOPX_EFFECT_RUNTIME_IDLE_MS", "60000") + effect_id = "fixture-goal:fixture-agent:todo_fixture0001:publication" + journal_path = tmp_path / "turn.json" + journal = { + "schema_version": "loopx_turn_journal_v0", + "goal_id": "fixture-goal", + "turn_key": "sha256:" + "a" * 64, + "status": "in_progress", + "completed_phases": [], + "plan": { + "turn_envelope": { + "goal_id": "fixture-goal", "agent_id": "fixture-agent", + "action": {"selected_todo": {"todo_id": "todo_fixture0001"}}, + }, + "transaction": { + "turn_key": "sha256:" + "a" * 64, + "turn_instance_id": "publication", + "settlement_plan": { + "schema_version": "quota_settlement_plan_v1", + "identity": { + "schema_version": "quota_settlement_identity_v0", + "effect_id": effect_id, "goal_id": "fixture-goal", + "agent_id": "fixture-agent", "todo_id": "todo_fixture0001", + "turn_instance_id": "publication", + }, + }, + }, + }, + } + params = {"path": str(journal_path), "journal": journal, + "expected_effect_id": effect_id} + info_path = effect_runtime._runtime_info_path(effect_runtime._runtime_fingerprint()) + with ThreadPoolExecutor(max_workers=1) as executor: + response = executor.submit( + effect_runtime.effect_runtime_result, method, + params if method == "turn_journal.write" else {}, retry_safe=False, + ) + try: + deadline = time.monotonic() + 5 + while not opened.exists() and time.monotonic() < deadline: + time.sleep(0.01) + assert opened.exists(), "real publication did not reach its cleanup claim" + assert info_path.exists() + lock_path = Path(str(info_path) + ".ts-effect.lock") + claims = list(runtime_dir.glob("*.ts-effect.lock.claim.*")) + assert lock_path.exists() and len(claims) == 1 + assert claims[0].stat().st_size == 0 + # The old implementation replies/commits while this claim is blank. + time.sleep(0.2) + assert not response.done(), "first request acknowledged unfinished publication" + assert not journal_path.exists(), "typed write ran before publication settled" + finally: + release.touch() + result = response.result(timeout=5) + assert not lock_path.exists() + assert not list(runtime_dir.glob("*.ts-effect.lock.claim.*")) + if method == "turn_journal.write": + assert result["appended"] is True and result["replayed"] is False + assert json.loads(journal_path.read_text(encoding="utf-8")) == journal + original = effect_runtime.effect_runtime_result("runtime.ping", {}) + try: + os.kill(int(original["pid"]), signal.SIGTERM) + time.sleep(0.1) + recovered = effect_runtime.effect_runtime_result( + "turn_journal.write", params, retry_safe=True, + ) + assert recovered["replayed"] is (method == "turn_journal.write") + assert recovered["appended"] is (method == "runtime.ping") + assert json.loads(journal_path.read_text(encoding="utf-8")) == journal + replacement = effect_runtime.effect_runtime_result("runtime.ping", {}) + assert replacement["pid"] != original["pid"] + finally: + effect_runtime.effect_runtime_result("runtime.shutdown", {}, retry_safe=False)