From 4fcb40b1ee19f0d4344784202479811b4972ad46 Mon Sep 17 00:00:00 2001 From: likiosliu Date: Wed, 22 Jul 2026 23:09:01 +0800 Subject: [PATCH 1/5] Add test-session resilience: turn-stall watchdog, idle-TTL reaper, lifecycle logs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three production-confirmed defects in the read-only test-session path (kbc/platform/pod/compile_box.py), all box-side: 1. Test turns hung on "thinking" forever. The test-session receive loop (test_session_driver) was a bare `async for msg in receive_messages()` with no model-stall watchdog — unlike the compile path, which runs _consume_turn_stream under _model_stall_watchdog. A black-holed model request (gateway accepts but never responds) left the turn active and the UI spinning indefinitely, with no recovery. Fix: _test_stall_watchdog + _consume_test_turn_stream reuse the compile idle-latency semantics (any SDK event == alive; a pending read tool relaxes to the tool bound). On a stall the turn is reaped once (no retry): interrupt the SDK stream, emit a turn-level error + turn_done so the UI goes idle, and keep the session live for the next question. The receive loop discards the torn-down ResultMessage. Bound defaults to the compile idle bound, overridable via KBC_TEST_MODEL_IDLE_TIMEOUT_S. 2. Orphaned sessions exhausted the concurrency cap. TEST_SESSIONS had no idle TTL and no way to enumerate sessions, so a server timeout or a persistence failure left box-side sessions that nobody closed, filling KBC_MAX_TEST_SESSIONS (3) until pod recreation. Fix: TestRun now tracks created_at / last_activity_at / a monotonic idle marker (touched on open, each turn, and event consumption). A periodic sweep (_test_session_reaper, KBC_TEST_SESSION_IDLE_TTL_S default 1800s) tears down idle sessions and emits a close event. New GET /test-sessions returns {"sessions":[{tid,parent_run_id,created_at,last_activity_at,done}]} (wire contract agreed with the consumer). 3. Zero lifecycle observability. The box only printed its startup line, so pod logs showed nothing for test-session open/close/turn activity. Fix: _print_test_lifecycle emits one stdout line per lifecycle event (test.open / test.close / turn.start / turn.done / turn.stalled / ttl.reap), carrying tid + parent_run_id only — never user content. Constraints held: CompileRun behavior untouched; destream/unrelated areas not touched; test sessions keep faithful streaming (no de-stream shim), so the new watchdog is their only stall guard. Tests (test_compile_box.py, +3): stall reaper frees a wedged turn and keeps the session live for a follow-up question; idle-TTL sweep drops orphans and spares active sessions; GET /test-sessions returns the agreed wire shape. Full suite green (69 passing). --- kbc/platform/pod/compile_box.py | 241 ++++++++++++++++++++++++++- kbc/platform/pod/test_compile_box.py | 168 +++++++++++++++++++ 2 files changed, 404 insertions(+), 5 deletions(-) diff --git a/kbc/platform/pod/compile_box.py b/kbc/platform/pod/compile_box.py index 58163e3a..f6638aaa 100644 --- a/kbc/platform/pod/compile_box.py +++ b/kbc/platform/pod/compile_box.py @@ -226,6 +226,52 @@ def _test_snapshot_root() -> str: _HIERARCHICAL_MODEL_IDLE_TIMEOUT_S = float(os.environ.get( "KBC_HIERARCHICAL_MODEL_IDLE_TIMEOUT_S", "240")) +# ── Test-session resilience (read-only consumer turns) ─────────────────────── +# A test session streams faithfully (no de-stream shim), so the compile-path +# watchdog does not cover it. It gets its OWN turn-stall guard reusing the same +# idle-latency semantics (SDK event == alive), defaulting to the compile idle +# bound. Unlike compile, a wedged test turn is reaped WITHOUT retry: the UI is +# freed with a turn-level error and the session stays live for the next question. +_TEST_MODEL_IDLE_TIMEOUT_S = float(os.environ.get( + "KBC_TEST_MODEL_IDLE_TIMEOUT_S", str(_MODEL_IDLE_TIMEOUT_S))) +# Orphan reaper: a server timeout / persistence failure can leave a box-side test +# session that nobody closes, holding one of KBC_MAX_TEST_SESSIONS until the pod +# is recreated. A periodic sweep tears down sessions idle past this TTL. +_TEST_SESSION_IDLE_TTL_S = float(os.environ.get("KBC_TEST_SESSION_IDLE_TTL_S", "1800")) +_TEST_SESSION_REAP_POLL_S = float(os.environ.get("KBC_TEST_SESSION_REAP_POLL_S", "60")) + + +def _now_iso() -> str: + """UTC timestamp for the test-session wire contract (created_at/last_activity_at).""" + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _print_test_lifecycle(label: str, run: "TestRun", extra: str = "") -> None: + """One stdout line per test-session lifecycle event (test.open / test.close / + turn.start / turn.done / turn.stalled / ttl.reap). tid + parent_run_id ONLY — + never user content — so pod logs make orphan/stall triage possible without + leaking the KB or the consumer's questions. Same [compile_box] prefix style.""" + tail = f" {extra}" if extra else "" + print(f"[compile_box] {label} tid={run.tid} parent={run.parent_run_id}{tail}", flush=True) + + +def _touch_test_activity(run: "TestRun") -> None: + """Mark liveness for the idle-TTL reaper (open, each turn, event consumption).""" + run._last_activity_monotonic = time.monotonic() + run.last_activity_at = _now_iso() + + +def _arm_test_turn(run: "TestRun") -> None: + """Arm the turn-stall watchdog for a new test-session turn (the /test-message + analogue of CompileRun._begin_turn, minus compile/selfcheck bookkeeping).""" + run._turn_active = True + run._tool_pending = False + run._stall_reaped = False + run._last_sdk_message_type = "query" + run._last_model_activity = time.monotonic() + _touch_test_activity(run) + + # ── Rate-limit resilience (C2) ─────────────────────────────────────────────── # massapi under concurrency (5-10 boxes × the red-blue PK) can return 429/503/529. # The bundled CLI retries internally a few times; past that the turn ends with @@ -445,6 +491,24 @@ def __init__(self, tid: str, cwd: str, parent_run_id: str, snapshot_hash: str, l self.consumer_model: str | None = None self.consumer_max_turns: int | None = None self.consumer_fingerprint: str = "" + # Lifecycle timestamps for the idle-TTL reaper and GET /test-sessions. + # created_at/last_activity_at are wire (iso8601, agreed with sicore); + # _last_activity_monotonic drives the TTL comparison (clock-skew safe). + now = time.monotonic() + self.created_at: str = _now_iso() + self.last_activity_at: str = self.created_at + self._last_activity_monotonic: float = now + # Turn-stall watchdog state (mirrors CompileRun's, minus retry/compile + # bookkeeping). A turn is active from a /test-message query() until its + # ResultMessage; the watchdog only judges an active turn, against model + # latency, relaxing to the tool bound while a read tool runs. + self._turn_active = False + self._tool_pending = False + self._last_model_activity: float = now + self._last_sdk_message_type = "query" # controlled class name only; never content + # Set by the watchdog when it reaps a wedged turn; the receive loop then + # discards the interrupted ResultMessage instead of announcing it. + self._stall_reaped = False async def emit(self, ev: dict): await self.events.put(ev) @@ -4096,6 +4160,118 @@ async def guard(input_data, tool_use_id, context): return guard +async def _consume_test_turn_stream(run: "TestRun", client) -> None: + """Relay a read-only test session's message stream through _emit_message, + owning the test-path stall reaper. Every inbound SDK message is liveness + (resets the idle clock, updates tool_pending) AND refreshes the TTL. Unlike + the compile path there is NO retry: when the watchdog has reaped the turn, + its ResultMessage is the torn-down attempt — discard it (the UI already got a + turn_done from the watchdog) and stay live for the next /test-message. A real + ResultMessage ends the turn normally.""" + async for msg in client.receive_messages(): + _note_model_activity(run, msg) # SDK event == alive; flips tool_pending + _touch_test_activity(run) # any message is also TTL liveness + if type(msg).__name__ == "ResultMessage": + if run._stall_reaped: + # The watchdog interrupted this turn and already emitted the + # turn-level error + turn_done. This is the reaped attempt's + # empty result; drop its partial text and keep the session live. + run._turn_text = [] + run._stall_reaped = False + run._turn_active = False + continue + was_active = run._turn_active + run._turn_active = False + if was_active: + _print_test_lifecycle("turn.done", run) + await _emit_message(run, msg) + continue + await _emit_message(run, msg) + + +async def _test_stall_watchdog(run: "TestRun") -> None: + """Reap a test-session turn wedged on a black-holed model request. The test + path streams faithfully (no de-stream shim), so this is its ONLY stall guard. + It does not retry: it reaps the turn once (interrupt → the receive loop + discards the torn-down result), emits a turn-level error + turn_done so the + UI stops "thinking", and leaves the session live for the next question. Only + judges an ACTIVE turn; a pending read tool relaxes to the tool bound so a slow + Read/Grep over the snapshot is never false-killed (I4).""" + while not run.done: + await asyncio.sleep(_MODEL_WATCHDOG_POLL_S) + if not run._turn_active or run._stall_reaped: + continue + client = run.client + if client is None: + continue + # destream_floor=0: test sessions do not de-stream, so the model bound + # needs no lift; a pending tool still keeps its own (longer) bound. + bound = _watchdog_idle_bound(run._tool_pending, _TEST_MODEL_IDLE_TIMEOUT_S, 0.0) + idle = time.monotonic() - run._last_model_activity + if idle <= bound: + continue + run._stall_reaped = True + run._turn_active = False + run._turn_text = [] # the wedged attempt produced nothing usable + _print_test_lifecycle( + "turn.stalled", run, extra=f"idle={round(idle, 1)}s bound={round(bound, 1)}s") + await run.emit({ + "type": "turn_stalled", + "idle_s": round(idle, 1), + "bound_s": round(bound, 1), + "tool_pending": run._tool_pending, + "last_sdk_message": run._last_sdk_message_type, + }) + await run.emit({"type": "error", "error": _loc(run, + "The answer timed out — please try again.", + "回答超时,请重试。")}) + # turn_done frees the UI (idle) while the session stays open; the next + # /test-message re-arms a fresh turn on the same live client. + await run.emit({"type": "turn_done", "text": _loc(run, + "The answer timed out — please try again.", + "回答超时,请重试。")}) + try: + await client.interrupt() + except Exception as e: + # A swallowed interrupt cannot un-wedge the receive loop, but the UI + # already recovered above; do not disconnect (spec: keep the session + # live). Surface the interrupt failure for pod-log triage. + await run.emit({"type": "summary", "text": _loc(run, + f"Test session: interrupt after stall failed {e!r}", + f"测试会话:停滞后中断失败 {e!r}")}) + + +async def _reap_idle_test_sessions() -> int: + """One idle-TTL sweep: tear down every test session idle past the TTL and emit + a close event. Split from the loop so a test can drive a single deterministic + pass. Returns the number reaped.""" + now = time.monotonic() + stale = [r for r in list(TEST_SESSIONS.values()) + if not r.done and (now - r._last_activity_monotonic) > _TEST_SESSION_IDLE_TTL_S] + for run in stale: + _print_test_lifecycle( + "ttl.reap", run, extra=f"idle={round(now - run._last_activity_monotonic)}s") + try: + await run.emit({"type": "summary", "text": _loc(run, + "Test session closed after being idle.", + "测试会话空闲超时,已关闭。")}) + except Exception: + pass + await _teardown_test_session(run) # cancels the task → wrapper emits `end` + return len(stale) + + +async def _test_session_reaper() -> None: + """Periodic idle-TTL sweep for orphaned test sessions. Fail-open: a sweep + error must never kill the reaper loop (it protects a shared concurrency slot).""" + while True: + await asyncio.sleep(_TEST_SESSION_REAP_POLL_S) + try: + await _reap_idle_test_sessions() + except Exception as e: + print(f"[compile_box] test-session reaper sweep failed: {e!r}", flush=True) + + async def test_session_driver(run: "TestRun"): """Read-only consumer driver: host a ClaudeSDKClient over the pinned snapshot dir, tools limited to the kb-test profile's whitelist (default Read/Glob/Grep), @@ -4146,8 +4322,10 @@ async def test_session_driver(run: "TestRun"): sid = str(getattr(client, "session_id", sid) or sid) run.session_id = sid await run.emit({"type": "session", "session_id": sid}) - async for msg in client.receive_messages(): - await _emit_message(run, msg) + # Persistent conversational stream with the turn-stall reaper (started by + # _test_session_wrapper). Yields this turn's output then blocks for the + # next /test-message, keeping the session alive until close/idle-TTL. + await _consume_test_turn_stream(run, client) finally: run.connected.set() # unblock any /test-message waiters even if connect failed run.client = None @@ -4621,15 +4799,22 @@ async def assist_reference_answer(parent: "CompileRun", payload: dict) -> dict: async def _test_session_wrapper(run: "TestRun"): - """Lifecycle for a read-only test session: run the driver, turn a crash into an - `error` event, always close with `end`. No syncer / bundle fallback (read-only, - nothing to persist). Cancellation (teardown) skips `error`, still emits `end`.""" + """Lifecycle for a read-only test session: run the driver under a turn-stall + watchdog, turn a crash into an `error` event, always close with `end`. No + syncer / bundle fallback (read-only, nothing to persist). Cancellation + (teardown) skips `error`, still emits `end`.""" + watchdog = asyncio.create_task(_test_stall_watchdog(run)) try: await _TEST_SESSION_IMPL(run) except Exception as e: # top-level boundary; CancelledError (teardown) passes through await run.emit({"type": "error", "error": repr(e)}) finally: run.done = True + watchdog.cancel() + try: + await watchdog + except asyncio.CancelledError: + pass await run.emit({"type": "end"}) @@ -4637,6 +4822,7 @@ async def _teardown_test_session(run: "TestRun"): """Cancel the session task (→ driver finally disconnects the client), drop the snapshot dir, and forget the run. The original RUNS machinery has no GC; test sessions are frequent + ephemeral, so they get explicit teardown.""" + _print_test_lifecycle("test.close", run) if run.task and not run.task.done(): run.task.cancel() try: @@ -5218,6 +5404,7 @@ async def handle_open_test(request: web.Request): ) TEST_SESSIONS[tid] = run run.task = asyncio.create_task(_test_session_wrapper(run)) + _print_test_lifecycle("test.open", run, extra=f"pages={pages}") return web.json_response({ "ok": True, "test_session_id": tid, @@ -5310,6 +5497,8 @@ async def handle_test_message(request: web.Request): text = (body.get("message") or "").strip() if not text: return web.json_response({"error": "message is required"}, status=400) + _arm_test_turn(run) # arm the turn-stall watchdog for this turn + _print_test_lifecycle("turn.start", run) await run.client.query(text) return web.json_response({"ok": True}) @@ -5346,6 +5535,45 @@ async def handle_close_test(request: web.Request): return web.json_response({"ok": True}) +async def handle_list_test_sessions(request: web.Request): + """List the box's live test sessions so the consumer can reconcile orphans + against its own records and drive the idle-reap view. Wire contract agreed + with sicore — the field names (tid/parent_run_id/created_at/last_activity_at/ + done) are load-bearing, do not rename.""" + return web.json_response({"sessions": [ + { + "tid": r.tid, + "parent_run_id": r.parent_run_id, + "created_at": r.created_at, + "last_activity_at": r.last_activity_at, + "done": r.done, + } + for r in TEST_SESSIONS.values() + ]}) + + +_TEST_SESSION_REAPER_TASK: "asyncio.Task | None" = None + + +async def _start_test_session_reaper(_app) -> None: + """aiohttp on_startup: launch the idle-TTL orphan reaper (needs a running loop).""" + global _TEST_SESSION_REAPER_TASK + _TEST_SESSION_REAPER_TASK = asyncio.create_task(_test_session_reaper()) + + +async def _stop_test_session_reaper(_app) -> None: + """aiohttp on_cleanup: stop the reaper so it does not outlive the app.""" + global _TEST_SESSION_REAPER_TASK + task = _TEST_SESSION_REAPER_TASK + _TEST_SESSION_REAPER_TASK = None + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + async def _flush_on_shutdown(_app) -> None: """aiohttp on_shutdown (F3): SIGTERM is stopping the box. Emit a final workspace sync for every active run so the last unsynced work reaches the @@ -5391,7 +5619,9 @@ def build_app() -> web.Application: middlewares=[_client_certificate_middleware], ) app.on_startup.append(destream.start) + app.on_startup.append(_start_test_session_reaper) app.on_shutdown.append(_flush_on_shutdown) + app.on_cleanup.append(_stop_test_session_reaper) app.add_routes([ web.post("/sources", handle_sources), web.post("/sources/begin", handle_sources_begin), @@ -5409,6 +5639,7 @@ def build_app() -> web.Application: web.post("/test-reference-assist/{run_id}", handle_test_reference_assist), web.post("/test-message/{tid}", handle_test_message), web.get("/test-events/{tid}", handle_test_events), + web.get("/test-sessions", handle_list_test_sessions), web.post("/test-session/{tid}/close", handle_close_test), web.get("/health", handle_health), ]) diff --git a/kbc/platform/pod/test_compile_box.py b/kbc/platform/pod/test_compile_box.py index e6af936e..df7ef3c5 100644 --- a/kbc/platform/pod/test_compile_box.py +++ b/kbc/platform/pod/test_compile_box.py @@ -6,6 +6,7 @@ """ import asyncio import base64 +import contextlib import hashlib import io import json @@ -13,6 +14,7 @@ import shutil import tarfile import tempfile +import time from pathlib import Path from aiohttp.test_utils import TestClient, TestServer @@ -953,6 +955,169 @@ async def drain(run): print("✓ test-session step frames (retrieval trace); compile runs unaffected") +class _TestStallFake: + """First query black-holes (receive blocks on a gate) so the watchdog must + reap it; interrupt() yields the torn-down ResultMessage. A SECOND query + produces a real answer then closes → proves the session stayed live after the + reap (the whole point of defect 1: recover the UI without killing the session).""" + + def __init__(self): + self.queries = [] + self.interrupts = 0 + self._gate = asyncio.Event() + self._mode = "idle" + self._closed = False + + async def connect(self, prompt=None): + pass + + async def query(self, text, session_id="default"): + self.queries.append(text) + if len(self.queries) == 1: + self._mode = "blackhole" # no gate set → receive blocks (the wedge) + else: + self._mode = "produce" + self._gate.set() + + async def interrupt(self): + self.interrupts += 1 + self._mode = "interrupted" + self._gate.set() + + async def receive_messages(self): + while not self._closed: + await self._gate.wait() + self._gate.clear() + if self._mode == "interrupted": + yield ResultMessage() # the reaped attempt ends + elif self._mode == "produce": + yield AssistantMessage("RoCE is a transport") + yield ResultMessage() + self._closed = True + return + + async def disconnect(self): + self._closed = True + + +async def test_test_session_stall_reaps_turn_keeps_session_live(): + """Defect 1: a black-holed test turn used to hang the UI on "思考中" forever — + the test path had no stall watchdog. Now a wedged turn is reaped (interrupt + + turn-level error + turn_done so the UI goes idle), and the session stays live: + a fresh question after the stall still gets a real answer. Also asserts the + turn.stalled / turn.done stdout lines carry tid+parent and NO user content.""" + saved = (compile_box._TEST_MODEL_IDLE_TIMEOUT_S, compile_box._MODEL_WATCHDOG_POLL_S) + compile_box._TEST_MODEL_IDLE_TIMEOUT_S = 0.15 + compile_box._MODEL_WATCHDOG_POLL_S = 0.03 + buf = io.StringIO() + fake = _TestStallFake() + try: + with tempfile.TemporaryDirectory() as snap, contextlib.redirect_stdout(buf): + run = compile_box.TestRun("t-stall", snap, parent_run_id="p-stall", snapshot_hash="h") + run.client = fake + run.connected.set() + wdog = asyncio.create_task(compile_box._test_stall_watchdog(run)) + consume = asyncio.create_task(compile_box._consume_test_turn_stream(run, fake)) + try: + # turn 1 → black-holed; the watchdog must interrupt/reap it + compile_box._arm_test_turn(run) + await fake.query("what is roce?") + deadline = time.monotonic() + 3 + while time.monotonic() < deadline and fake.interrupts == 0: + await asyncio.sleep(0.02) + assert fake.interrupts == 1, fake.interrupts + # turn 2 → the SAME live session answers a fresh question + compile_box._arm_test_turn(run) + await fake.query("again") + await asyncio.wait_for(consume, timeout=3) + finally: + run.done = True + wdog.cancel() + try: + await wdog + except asyncio.CancelledError: + pass + evs = _drain(run) + types = [e["type"] for e in evs] + assert "turn_stalled" in types and "error" in types and "turn_done" in types, types + td = next(e for e in evs if e["type"] == "turn_done") + assert "timed out" in td["text"], td # UI-facing timeout wording (en default) + # the post-stall question produced a real answer → session stayed usable + assert any(e["type"] == "log" and "RoCE" in e.get("text", "") for e in evs), evs + assert fake.queries == ["what is roce?", "again"], fake.queries + out = buf.getvalue() + assert "turn.stalled tid=t-stall parent=p-stall" in out, out + assert "turn.done tid=t-stall parent=p-stall" in out, out + assert "roce" not in out.lower(), out # never leak the question text to stdout + finally: + (compile_box._TEST_MODEL_IDLE_TIMEOUT_S, compile_box._MODEL_WATCHDOG_POLL_S) = saved + print("✓ test-session stall watchdog reaps a wedged turn and keeps the session live") + + +async def test_test_session_idle_ttl_reaper(): + """Defect 2 (TTL): a test session idle past KBC_TEST_SESSION_IDLE_TTL_S is torn + down by the periodic sweep — the snapshot dir + registry slot are freed and a + ttl.reap stdout line fires. A recently-active session is left alone.""" + compile_box.TEST_SESSIONS.clear() + saved = compile_box._TEST_SESSION_IDLE_TTL_S + compile_box._TEST_SESSION_IDLE_TTL_S = 100.0 + buf = io.StringIO() + stale_dir = tempfile.mkdtemp() + fresh_dir = tempfile.mkdtemp() + try: + stale = compile_box.TestRun("t-stale", stale_dir, parent_run_id="p-ttl", snapshot_hash="h") + stale._last_activity_monotonic = time.monotonic() - 500.0 # idle > TTL + fresh = compile_box.TestRun("t-fresh", fresh_dir, parent_run_id="p-ttl", snapshot_hash="h") + compile_box._touch_test_activity(fresh) # just active + compile_box.TEST_SESSIONS["t-stale"] = stale + compile_box.TEST_SESSIONS["t-fresh"] = fresh + with contextlib.redirect_stdout(buf): + reaped = await compile_box._reap_idle_test_sessions() + assert reaped == 1, reaped + assert "t-stale" not in compile_box.TEST_SESSIONS, compile_box.TEST_SESSIONS + assert "t-fresh" in compile_box.TEST_SESSIONS + assert not Path(stale_dir).exists(), "stale snapshot dir not dropped" + # the reaped session got a close (summary) event before teardown emitted end + se = [stale.events.get_nowait() for _ in range(stale.events.qsize())] + assert any(e["type"] == "summary" for e in se), se + out = buf.getvalue() + assert "ttl.reap tid=t-stale parent=p-ttl" in out, out + assert "test.close tid=t-stale parent=p-ttl" in out, out + finally: + compile_box._TEST_SESSION_IDLE_TTL_S = saved + compile_box.TEST_SESSIONS.clear() + shutil.rmtree(stale_dir, ignore_errors=True) + shutil.rmtree(fresh_dir, ignore_errors=True) + print("✓ test-session idle-TTL reaper drops orphans, spares active sessions") + + +async def test_list_test_sessions_endpoint(): + """Defect 2 (list): GET /test-sessions returns the agreed sicore wire contract + exactly — tid / parent_run_id / created_at / last_activity_at / done — one row + per live session.""" + compile_box.TEST_SESSIONS.clear() + client = TestClient(TestServer(compile_box.build_app())) + await client.start_server() + try: + empty = await (await client.get("/test-sessions")).json() + assert empty == {"sessions": []}, empty + + run = compile_box.TestRun("tl1", "/tmp", parent_run_id="pl1", snapshot_hash="h") + run.done = False + compile_box.TEST_SESSIONS["tl1"] = run + body = await (await client.get("/test-sessions")).json() + assert list(body.keys()) == ["sessions"], body + assert len(body["sessions"]) == 1, body + row = body["sessions"][0] + assert set(row.keys()) == {"tid", "parent_run_id", "created_at", "last_activity_at", "done"}, row + assert row["tid"] == "tl1" and row["parent_run_id"] == "pl1" and row["done"] is False, row + assert row["created_at"].endswith("Z") and row["last_activity_at"].endswith("Z"), row + finally: + await client.close() + compile_box.TEST_SESSIONS.clear() + print("✓ GET /test-sessions lists live sessions in the agreed wire contract") + + async def test_explicit_test_recommendation_http(): """The explicit red-team endpoint is stable-draft-only, single-flight, and returns one validated raw-grounded recommendation without mutating files.""" @@ -3907,6 +4072,9 @@ async def main(): await test_open_close_test_session_http() await test_test_message_path() await test_test_session_step_frames() + await test_test_session_stall_reaps_turn_keeps_session_live() + await test_test_session_idle_ttl_reaper() + await test_list_test_sessions_endpoint() await test_explicit_test_recommendation_http() await test_reference_assist_http_and_validation() await test_reference_assist_driver_is_fast_isolated_and_structured() From 651dc321a627f1b17450e4d1c085ee55f9f27679 Mon Sep 17 00:00:00 2001 From: likiosliu Date: Wed, 22 Jul 2026 23:24:53 +0800 Subject: [PATCH 2/5] Add capability.testSessions runtime method + structured 429 for test-session limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the box-side test-session resilience through the TS gateway so the consumer can reach it, aligned with the sicore side. 1. capability.testSessions (src/gateway/server.ts): new RPC method proxying the box's GET /test-sessions. Read-only reconciliation — mirrors testClose's box discovery and NEVER spawns/rehydrates a box just to list (absent/dead box → empty list). Box rows pass through verbatim; the `tid` wire field is load-bearing for the consumer and is not renamed. Response: {run_id, sessions:[{tid,parent_run_id,created_at,last_activity_at,done}]}. 2. Stable 429 error code (kbc/platform/pod/compile_box.py): the box's "too many concurrent test sessions" refusal now returns the structured {error:{code,message,retriable:false}} shape (same convention as handle_test_recommendation) with code "test_session_limit". The agentbox error mapping (agentBoxResponseError → wrapRpcError) passes the code + retriable=false through unchanged. Without this a bare 429 mapped to the generic TOO_MANY_REQUESTS with retriable=true — indistinguishable from a model rate-limit 429 and wrongly retriable. Not retriable by design: the caller must close an idle session, not auto-retry. 3. Wire contract (src/gateway/capability/contract.ts): CAPABILITY_TEST_SESSIONS, the request/response/summary interfaces, and TEST_SESSION_LIMIT_ERROR_CODE documented so the box↔runtime↔consumer contract is single-sourced. Tests: new src/gateway/server-capability-test-sessions.test.ts (passthrough with tid preserved / empty on absent box / run_id required); box test asserts the structured 429 body. tsc --noEmit clean; gateway vitest green; box suite 69 passing. --- kbc/platform/pod/compile_box.py | 18 ++- kbc/platform/pod/test_compile_box.py | 8 +- src/gateway/capability/contract.ts | 42 ++++++ .../server-capability-test-sessions.test.ts | 126 ++++++++++++++++++ src/gateway/server.ts | 29 ++++ 5 files changed, 221 insertions(+), 2 deletions(-) create mode 100644 src/gateway/server-capability-test-sessions.test.ts diff --git a/kbc/platform/pod/compile_box.py b/kbc/platform/pod/compile_box.py index f6638aaa..515c4c8e 100644 --- a/kbc/platform/pod/compile_box.py +++ b/kbc/platform/pod/compile_box.py @@ -239,6 +239,13 @@ def _test_snapshot_root() -> str: # is recreated. A periodic sweep tears down sessions idle past this TTL. _TEST_SESSION_IDLE_TTL_S = float(os.environ.get("KBC_TEST_SESSION_IDLE_TTL_S", "1800")) _TEST_SESSION_REAP_POLL_S = float(os.environ.get("KBC_TEST_SESSION_REAP_POLL_S", "60")) +# Structured error code for the 429 refusal when the pod's concurrent +# test-session cap is full. Emitted in the box's own {error:{code,message, +# retriable}} shape (same convention as handle_test_recommendation) so the +# runtime's agentbox error mapping passes a STABLE code + retriable=false +# through to the consumer — a bare 429 would otherwise map to the generic +# TOO_MANY_REQUESTS with retriable=true. Mirrored in gateway contract.ts. +TEST_SESSION_LIMIT_ERROR_CODE = "test_session_limit" def _now_iso() -> str: @@ -5376,7 +5383,16 @@ async def handle_open_test(request: web.Request): return web.json_response({"error": "unknown run"}, status=404) active = sum(1 for t in TEST_SESSIONS.values() if not t.done) if active >= _max_test_sessions(): - return web.json_response({"error": "too many concurrent test sessions"}, status=429) + # Structured error (same shape as handle_test_recommendation) so the + # runtime maps a STABLE code + retriable=false, not the generic 429. + # Not retriable: the caller must close an idle session, not auto-retry. + return web.json_response( + {"error": { + "code": TEST_SESSION_LIMIT_ERROR_CODE, + "message": "too many concurrent test sessions", + "retriable": False, + }}, + status=429) body = await request.json() if request.body_exists else {} consumer_tools = _effective_test_allowed_tools(body.get("allowed_tools")) consumer_model = _test_model() diff --git a/kbc/platform/pod/test_compile_box.py b/kbc/platform/pod/test_compile_box.py index df7ef3c5..ea661a93 100644 --- a/kbc/platform/pod/test_compile_box.py +++ b/kbc/platform/pod/test_compile_box.py @@ -828,7 +828,13 @@ async def test_open_close_test_session_http(): # concurrency cap → 429 (deterministic: a non-done stub occupies the only slot) os.environ["KBC_MAX_TEST_SESSIONS"] = "1" compile_box.TEST_SESSIONS["busy"] = compile_box.TestRun("busy", "/tmp/x", "p1", "h") - assert (await client.post("/test-session/p1")).status == 429 + capped = await client.post("/test-session/p1") + assert capped.status == 429 + # structured error so the runtime maps a stable code + retriable=false + err = (await capped.json())["error"] + assert err["code"] == compile_box.TEST_SESSION_LIMIT_ERROR_CODE, err + assert err["retriable"] is False, err + assert "too many concurrent test session" in err["message"], err compile_box.TEST_SESSIONS.clear() os.environ.pop("KBC_MAX_TEST_SESSIONS", None) diff --git a/src/gateway/capability/contract.ts b/src/gateway/capability/contract.ts index e35c4056..172dec39 100644 --- a/src/gateway/capability/contract.ts +++ b/src/gateway/capability/contract.ts @@ -44,6 +44,20 @@ export const CAPABILITY_TEST_MESSAGE = "capability.testMessage" as const; export const CAPABILITY_TEST_CLOSE = "capability.testClose" as const; export const CAPABILITY_TEST_RECOMMEND = "capability.testRecommend" as const; export const CAPABILITY_TEST_REFERENCE_ASSIST = "capability.testReferenceAssist" as const; +/** Enumerate the box's live test sessions for a run (orphan reconciliation). */ +export const CAPABILITY_TEST_SESSIONS = "capability.testSessions" as const; + +/** + * ErrorDetail.code the box emits when testStart is refused because the pod's + * concurrent test-session cap (KBC_MAX_TEST_SESSIONS) is full — HTTP 429 with a + * structured `{error:{code,message,retriable:false}}` body. Sourced in the box + * (compile_box handle_open_test) and passed through unchanged by the agentbox + * error mapping (agentBoxResponseError → wrapRpcError). NOT retriable: the caller + * must close an idle session, not auto-retry. Distinct from the generic + * TOO_MANY_REQUESTS a bare 429 would otherwise map to, so a consumer can tell + * "test-session limit" apart from a model rate-limit 429. + */ +export const TEST_SESSION_LIMIT_ERROR_CODE = "test_session_limit" as const; /** siclaw → consumer: live stream + content sink + input fetch. */ export const CAPABILITY_EVENT = "capability.event" as const; @@ -214,6 +228,34 @@ export interface CapabilityTestCloseRequest { test_session_id: string; } +/** + * Enumerate the box's live test sessions for a run so the consumer can + * reconcile orphans (a lost server ack / persistence failure can leave a + * box-side session nobody closed) against its own records. Read-only: the + * runtime NEVER spawns/rehydrates a box just to list — a dead/absent box has + * no sessions and returns an empty list. + */ +export interface CapabilityTestSessionsRequest { + run_id: string; +} + +export interface CapabilityTestSessionSummary { + /** Box-side test session id. Field name is `tid` (NOT test_session_id) — the + * box's GET /test-sessions rows pass through unchanged; do not rename. */ + tid: string; + parent_run_id: string; + /** iso8601 UTC (e.g. "2026-07-22T09:30:00Z"). */ + created_at: string; + /** iso8601 UTC; refreshed on open, each turn, and event consumption. */ + last_activity_at: string; + done: boolean; +} + +export interface CapabilityTestSessionsResponse { + run_id: string; + sessions: CapabilityTestSessionSummary[]; +} + /** * Explicit, owner-triggered red-team authoring of one regression test. * This is deliberately separate from compilation: the one-shot session can diff --git a/src/gateway/server-capability-test-sessions.test.ts b/src/gateway/server-capability-test-sessions.test.ts new file mode 100644 index 00000000..d747a2c2 --- /dev/null +++ b/src/gateway/server-capability-test-sessions.test.ts @@ -0,0 +1,126 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const getJsonMock = vi.hoisted(() => vi.fn()); + +vi.mock("./chat-repo.js", () => ({ + ensureChatSession: vi.fn(async () => {}), + appendMessage: vi.fn(async () => "msg-id"), + bindMessageTraceId: vi.fn(async () => {}), + updateMessage: vi.fn(async () => {}), + incrementMessageCount: vi.fn(async () => {}), +})); +vi.mock("./output-redactor.js", () => ({ buildRedactionConfigForModelConfig: vi.fn(() => ({})) })); +vi.mock("./sse-consumer.js", () => ({ + consumeAgentSse: vi.fn(async () => ({ resultText: "", taskReportText: "", errorMessage: "", eventCount: 0, durationMs: 0 })), +})); +vi.mock("./agentbox/client.js", () => ({ + AgentBoxClient: class { + endpoint: string; + constructor(endpoint: string) { this.endpoint = endpoint; } + async postJson() { return {}; } + getJson = getJsonMock; + async *streamPath() {} + }, +})); +vi.mock("./capability/materialize.js", () => ({ + materializeCapabilityInputs: vi.fn(async () => ({ locale: undefined, llm: undefined, settings: undefined })), +})); + +const { startRuntime } = await import("./server.js"); + +function fakeFrontendClient() { + return { + request: vi.fn(async () => ({})), + onCommand: vi.fn(), + emitEvent: vi.fn(), + close: vi.fn(), + } as any; +} + +function fakeAgentBoxManager(box: { boxId: string; endpoint: string; agentId: string } | null) { + return { + setCertManager: vi.fn(), + setSpawnEnvResolver: vi.fn(), + setPersistenceResolver: vi.fn(), + getAsync: vi.fn(async () => box), + getOrCreate: vi.fn(async () => box), + stop: vi.fn(async () => {}), + list: vi.fn(() => []), + cleanup: vi.fn(async () => {}), + } as any; +} + +let server: Awaited> | undefined; + +beforeEach(() => { + getJsonMock.mockReset(); +}); + +afterEach(async () => { + if (server) await server.close(); + server = undefined; + vi.clearAllMocks(); +}); + +describe("capability.testSessions", () => { + it("lists the box's sessions verbatim (tid field preserved) without spawning a box", async () => { + const rows = [ + { + tid: "test-1", + parent_run_id: "run-live", + created_at: "2026-07-22T09:30:00Z", + last_activity_at: "2026-07-22T09:31:00Z", + done: false, + }, + ]; + getJsonMock.mockResolvedValue({ sessions: rows }); + const manager = fakeAgentBoxManager({ + boxId: "agentbox-run-live", + endpoint: "https://10.0.0.10:3000", + agentId: "run-live", + }); + server = await startRuntime({ + config: { port: 0, internalPort: 0, host: "127.0.0.1", serverUrl: "", portalSecret: "" } as any, + agentBoxManager: manager, + frontendClient: fakeFrontendClient(), + credentialService: {} as any, + }); + + const list = server.rpcMethods.get("capability.testSessions")!; + await expect(list({ run_id: "run-live" })).resolves.toEqual({ + run_id: "run-live", + sessions: rows, // passed through byte-for-byte: `tid`, not renamed + }); + expect(manager.getAsync).toHaveBeenCalledWith("run-live"); + expect(getJsonMock).toHaveBeenCalledWith("/test-sessions"); + expect(manager.getOrCreate).not.toHaveBeenCalled(); // never spawn to list + }); + + it("returns an empty list when the box is absent — never spawns/rehydrates", async () => { + const manager = fakeAgentBoxManager(null); + server = await startRuntime({ + config: { port: 0, internalPort: 0, host: "127.0.0.1", serverUrl: "", portalSecret: "" } as any, + agentBoxManager: manager, + frontendClient: fakeFrontendClient(), + credentialService: {} as any, + }); + + const list = server.rpcMethods.get("capability.testSessions")!; + await expect(list({ run_id: "run-gone" })).resolves.toEqual({ run_id: "run-gone", sessions: [] }); + expect(getJsonMock).not.toHaveBeenCalled(); + expect(manager.getOrCreate).not.toHaveBeenCalled(); + }); + + it("requires run_id", async () => { + const manager = fakeAgentBoxManager(null); + server = await startRuntime({ + config: { port: 0, internalPort: 0, host: "127.0.0.1", serverUrl: "", portalSecret: "" } as any, + agentBoxManager: manager, + frontendClient: fakeFrontendClient(), + credentialService: {} as any, + }); + + const list = server.rpcMethods.get("capability.testSessions")!; + await expect(list({} as any)).rejects.toThrow("run_id is required"); + }); +}); diff --git a/src/gateway/server.ts b/src/gateway/server.ts index 036d77d6..753d3039 100644 --- a/src/gateway/server.ts +++ b/src/gateway/server.ts @@ -44,6 +44,9 @@ import type { CapabilityTestReferenceAssistRequest, CapabilityTestReferenceAssistResponse, CapabilityTestMessageRequest, + CapabilityTestSessionsRequest, + CapabilityTestSessionsResponse, + CapabilityTestSessionSummary, CapabilityTestStartRequest, CapabilityTestStartResponse, } from "./capability/contract.js"; @@ -989,6 +992,32 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise { + const req = params as unknown as CapabilityTestSessionsRequest; + if (!req.run_id) throw new Error("run_id is required"); + // Read-only reconciliation: NEVER spawn/rehydrate a box just to list (mirrors + // testClose's box discovery). An absent/dead box has no live sessions → []. + // The box's GET /test-sessions rows are passed through verbatim (the `tid` + // wire field is load-bearing for the consumer — do not rename). + let client: AgentBoxClient; + if (localCapabilityBoxEndpoint) { + client = new AgentBoxClient(localCapabilityBoxEndpoint, 30000, agentBoxTlsOptions); + } else { + const alive = await agentBoxManager.getAsync(req.run_id); + if (!alive) { + const empty: CapabilityTestSessionsResponse = { run_id: req.run_id, sessions: [] }; + return empty; + } + client = new AgentBoxClient(alive.endpoint, 30000, agentBoxTlsOptions); + } + const listed = await client.getJson<{ sessions: CapabilityTestSessionSummary[] }>("/test-sessions"); + const response: CapabilityTestSessionsResponse = { + run_id: req.run_id, + sessions: listed.sessions ?? [], + }; + return response; + }); + rpcMethods.set("chat.abort", async (params) => { const agentId = params.agentId as string; const sessionId = params.sessionId as string; From 2a0a995e8794bc9623128d82e4129f0ac2157d38 Mon Sep 17 00:00:00 2001 From: likiosliu Date: Thu, 23 Jul 2026 12:32:07 +0800 Subject: [PATCH 3/5] Add testStart idempotency key + run-scoped test-session listing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two P2 follow-ups from the sicore MR review, layered on the test-session work. 1. testStart idempotency (kbc/platform/pod/compile_box.py + gateway): handle_open_test accepts an optional `idempotency_key`. A retried open with the same key returns the SAME live session (same tid/snapshot_hash/pages) with `idempotent_replay: true`, instead of opening a second session or consuming a second concurrency slot — so a lost ack / client retry cannot leave a ghost session. Checked before the cap (a replay is never 429'd) and before packing (no snapshot work). The key→tid map is process-local and cleared with the session on teardown, so a torn-down key opens fresh and can never replay a dead tid. Absent key → behavior unchanged (old-consumer compat). The gateway's capability.testStart passes the field through and, crucially, SKIPS starting a second event relay on an idempotent_replay — the box's /test-events is single-consumer, so a duplicate relay would split the stream. 2. Run-scoped listing (gateway): capability.testSessions filters the box's rows to the requested run_id. A shared box (SICLAW_COMPILE_BOX_ENDPOINT) hosts sessions for multiple parent runs; one run must never see or reap another's. Double protection with the consumer's own ParentRunID check. Field name `idempotency_key` (aligned with sicore). Tests: box test asserts same-key replay → same tid + replay flag + unchanged active count, different key → new session, teardown drops the key (fresh afterwards), no-key never deduped; gateway test asserts a shared box's two runs are mutually invisible in the list. tsc --noEmit clean; gateway vitest green; box suite 70 passing. --- kbc/platform/pod/compile_box.py | 45 ++++++++- kbc/platform/pod/test_compile_box.py | 96 +++++++++++++++++++ src/gateway/capability/contract.ts | 8 ++ .../server-capability-test-sessions.test.ts | 30 ++++++ src/gateway/server.ts | 40 +++++--- 5 files changed, 204 insertions(+), 15 deletions(-) diff --git a/kbc/platform/pod/compile_box.py b/kbc/platform/pod/compile_box.py index 515c4c8e..90066175 100644 --- a/kbc/platform/pod/compile_box.py +++ b/kbc/platform/pod/compile_box.py @@ -101,6 +101,11 @@ # Read-only "test session" runs — ephemeral consumer sessions over a pinned draft # snapshot (test sessions). Parallel to RUNS, torn down on close/idle. See TestRun. TEST_SESSIONS: dict[str, "TestRun"] = {} +# Consumer-minted idempotency key → tid, for open-test-session retries. A retried +# testStart (same key) must return the SAME live session, never a second session +# or concurrency slot. Process-local; entries are cleared on teardown with their +# session (a torn-down key opens fresh, so a stale mapping can never be replayed). +TEST_SESSION_IDEMPOTENCY: dict[str, str] = {} # One explicit red-team read-only review may inspect a run at a time. Test # recommendation and reference-answer assistance share this guard so two owner # actions cannot spend overlapping model calls against the same workspace. @@ -516,6 +521,11 @@ def __init__(self, tid: str, cwd: str, parent_run_id: str, snapshot_hash: str, l # Set by the watchdog when it reaps a wedged turn; the receive loop then # discards the interrupted ResultMessage instead of announcing it. self._stall_reaped = False + # Snapshot page count (returned by open, replayed on an idempotent open). + self.pages: int = 0 + # Consumer-minted idempotency key that opened this session (if any), so + # teardown can drop its TEST_SESSION_IDEMPOTENCY entry. + self.idempotency_key: str | None = None async def emit(self, ev: dict): await self.events.put(ev) @@ -4830,6 +4840,12 @@ async def _teardown_test_session(run: "TestRun"): snapshot dir, and forget the run. The original RUNS machinery has no GC; test sessions are frequent + ephemeral, so they get explicit teardown.""" _print_test_lifecycle("test.close", run) + # Drop the idempotency mapping WITH the session: a torn-down key must open a + # fresh session, never replay a dead tid (guard on tid so a re-minted key for + # a newer session is not clobbered). + key = run.idempotency_key + if key and TEST_SESSION_IDEMPOTENCY.get(key) == run.tid: + TEST_SESSION_IDEMPOTENCY.pop(key, None) if run.task and not run.task.done(): run.task.cancel() try: @@ -5381,6 +5397,29 @@ async def handle_open_test(request: web.Request): parent = RUNS.get(request.match_info["run_id"]) if not parent: return web.json_response({"error": "unknown run"}, status=404) + body = await request.json() if request.body_exists else {} + # Idempotent open: a retried testStart (same consumer-minted key) returns the + # SAME live session. Checked BEFORE the cap (a replay must not be spuriously + # 429'd) and BEFORE packing (a replay does no snapshot work). `idempotent_replay` + # tells the runtime NOT to start a second event relay on the already-relayed + # session (the box's /test-events is single-consumer). Absent key → unchanged. + idem_key = (body.get("idempotency_key") or "").strip() + if len(idem_key) > 128: + return web.json_response({"error": "idempotency_key must be at most 128 characters"}, status=400) + if idem_key: + existing = TEST_SESSIONS.get(TEST_SESSION_IDEMPOTENCY.get(idem_key, "")) + if existing is not None and not existing.done: + _print_test_lifecycle("test.open.idempotent", existing) + return web.json_response({ + "ok": True, + "test_session_id": existing.tid, + "snapshot_hash": existing.snapshot_hash, + "consumer_fingerprint": existing.consumer_fingerprint, + "pages": existing.pages, + "idempotent_replay": True, + }) + # Stale mapping (session closed/reaped): drop it and open fresh below. + TEST_SESSION_IDEMPOTENCY.pop(idem_key, None) active = sum(1 for t in TEST_SESSIONS.values() if not t.done) if active >= _max_test_sessions(): # Structured error (same shape as handle_test_recommendation) so the @@ -5393,7 +5432,6 @@ async def handle_open_test(request: web.Request): "retriable": False, }}, status=429) - body = await request.json() if request.body_exists else {} consumer_tools = _effective_test_allowed_tools(body.get("allowed_tools")) consumer_model = _test_model() consumer_max_turns = _test_max_turns() @@ -5418,6 +5456,10 @@ async def handle_open_test(request: web.Request): run.locale, run.allowed_tools, run.consumer_model, max_turns=run.consumer_max_turns, ) + run.pages = pages + if idem_key: + run.idempotency_key = idem_key + TEST_SESSION_IDEMPOTENCY[idem_key] = tid TEST_SESSIONS[tid] = run run.task = asyncio.create_task(_test_session_wrapper(run)) _print_test_lifecycle("test.open", run, extra=f"pages={pages}") @@ -5427,6 +5469,7 @@ async def handle_open_test(request: web.Request): "snapshot_hash": snapshot_hash, "consumer_fingerprint": run.consumer_fingerprint, "pages": pages, + "idempotent_replay": False, }) diff --git a/kbc/platform/pod/test_compile_box.py b/kbc/platform/pod/test_compile_box.py index ea661a93..54da59dc 100644 --- a/kbc/platform/pod/test_compile_box.py +++ b/kbc/platform/pod/test_compile_box.py @@ -886,6 +886,101 @@ async def test_open_close_test_session_http(): print("✓ open/close test session over HTTP (snapshot pinned, cap, teardown)") +class _AliveFakeClient: + """A test-session SDK stand-in that connects and then BLOCKS in receive + (emits nothing, session stays live / not-done) until disconnect. The default + _FakeSDKClient self-completes almost immediately, which would race the + idempotency replay against the session going done — this keeps it live so the + replay deterministically hits an active session.""" + + def __init__(self, options=None): + self.options = options + self._closed = asyncio.Event() + + async def connect(self, prompt=None): + pass + + async def query(self, text, session_id="default"): + pass + + async def receive_messages(self): + await self._closed.wait() + return + yield # unreachable; makes this an async generator + + async def disconnect(self): + self._closed.set() + + +async def test_open_test_session_idempotency(): + """A retried testStart (same idempotency_key) returns the SAME live session — + same tid/snapshot_hash/pages, flagged idempotent_replay, no new session, no new + concurrency slot. A different key opens a new session. Teardown drops the key + so a later same-key open starts fresh (never replays a dead tid).""" + orig = compile_box.ClaudeSDKClient + compile_box.ClaudeSDKClient = _AliveFakeClient + compile_box.RUNS.clear() + compile_box.TEST_SESSIONS.clear() + compile_box.TEST_SESSION_IDEMPOTENCY.clear() + snap_root = tempfile.mkdtemp() + os.environ["KBC_TEST_SNAPSHOT_ROOT"] = snap_root + os.environ["KBC_MAX_TEST_SESSIONS"] = "10" # exercise idempotency, not the cap + client = TestClient(TestServer(compile_box.build_app())) + await client.start_server() + + async def active(): + return sum(1 for t in compile_box.TEST_SESSIONS.values() if not t.done) + + try: + wd = tempfile.mkdtemp() + cand = Path(wd) / "candidate" + cand.mkdir() + (cand / "index.md").write_text("# index\n") + compile_box.RUNS["p1"] = compile_box.CompileRun("p1", wd, 1) + + # first open with a key → a fresh session (idempotent_replay False) + b1 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-1"})).json() + tid1 = b1["test_session_id"] + assert b1["idempotent_replay"] is False, b1 + n1 = await active() + + # retry SAME key → same tid + snapshot_hash + pages, replay flagged, count unchanged + b2 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-1"})).json() + assert b2["test_session_id"] == tid1, (b1, b2) + assert b2["snapshot_hash"] == b1["snapshot_hash"] and b2["pages"] == b1["pages"], (b1, b2) + assert b2["idempotent_replay"] is True, b2 + assert await active() == n1, "replay must not consume a new slot" + assert len(compile_box.TEST_SESSIONS) == 1, compile_box.TEST_SESSIONS + + # a DIFFERENT key opens a new session + b3 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-2"})).json() + assert b3["test_session_id"] != tid1 and len(compile_box.TEST_SESSIONS) == 2 + + # teardown drops the key → the same key opens FRESH afterwards (no dead replay) + assert (await client.post(f"/test-session/{tid1}/close")).status == 200 + assert "k-1" not in compile_box.TEST_SESSION_IDEMPOTENCY, compile_box.TEST_SESSION_IDEMPOTENCY + b4 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-1"})).json() + assert b4["test_session_id"] != tid1 and b4["idempotent_replay"] is False, b4 + + # no-key opens are never deduped (old-consumer behavior unchanged) + b5 = await (await client.post("/test-session/p1")).json() + b6 = await (await client.post("/test-session/p1")).json() + assert b5["test_session_id"] != b6["test_session_id"], (b5, b6) + + for t in list(compile_box.TEST_SESSIONS.keys()): + await client.post(f"/test-session/{t}/close") + finally: + await client.close() + compile_box.ClaudeSDKClient = orig + compile_box.RUNS.clear() + compile_box.TEST_SESSIONS.clear() + compile_box.TEST_SESSION_IDEMPOTENCY.clear() + os.environ.pop("KBC_TEST_SNAPSHOT_ROOT", None) + os.environ.pop("KBC_MAX_TEST_SESSIONS", None) + shutil.rmtree(snap_root, ignore_errors=True) + print("✓ test-session open is idempotent per key (same tid, no new slot, replay flagged)") + + async def test_test_message_path(): """/test-message injects a user turn into a LIVE test session (200 + query forwarded); an unknown test session → 404.""" @@ -4076,6 +4171,7 @@ async def main(): await test_test_session_driver_readonly() await test_test_session_driver_uses_captured_contract() await test_open_close_test_session_http() + await test_open_test_session_idempotency() await test_test_message_path() await test_test_session_step_frames() await test_test_session_stall_reaps_turn_keeps_session_live() diff --git a/src/gateway/capability/contract.ts b/src/gateway/capability/contract.ts index 172dec39..6b4f60cc 100644 --- a/src/gateway/capability/contract.ts +++ b/src/gateway/capability/contract.ts @@ -204,6 +204,14 @@ export interface CapabilityTestStartRequest { bundle_base64?: string; /** sha256 of the bundle bytes; the box verifies it at install time. */ bundle_sha256?: string; + /** + * Optional consumer-minted idempotency key. A retried testStart with the same + * key returns the SAME live test session (same tid + snapshot_hash) instead of + * opening a second one / consuming a second concurrency slot — so a lost ack + * or a client retry cannot leave a ghost session. Absent → always open fresh + * (old-consumer behavior unchanged). + */ + idempotency_key?: string; } export interface CapabilityTestStartResponse { diff --git a/src/gateway/server-capability-test-sessions.test.ts b/src/gateway/server-capability-test-sessions.test.ts index d747a2c2..1cd97b65 100644 --- a/src/gateway/server-capability-test-sessions.test.ts +++ b/src/gateway/server-capability-test-sessions.test.ts @@ -96,6 +96,36 @@ describe("capability.testSessions", () => { expect(manager.getOrCreate).not.toHaveBeenCalled(); // never spawn to list }); + it("scopes the list to the requested run — a shared box's other runs stay invisible", async () => { + // A shared box (SICLAW_COMPILE_BOX_ENDPOINT) hosts sessions for many runs; + // the box returns them all, and the runtime must filter to the caller's run + // so one run can never see (or reap) another's session. + process.env.SICLAW_COMPILE_BOX_ENDPOINT = "https://10.0.0.20:3000"; + const iso = "2026-07-22T09:30:00Z"; + getJsonMock.mockResolvedValue({ + sessions: [ + { tid: "a-1", parent_run_id: "run-A", created_at: iso, last_activity_at: iso, done: false }, + { tid: "b-1", parent_run_id: "run-B", created_at: iso, last_activity_at: iso, done: false }, + { tid: "a-2", parent_run_id: "run-A", created_at: iso, last_activity_at: iso, done: true }, + ], + }); + server = await startRuntime({ + config: { port: 0, internalPort: 0, host: "127.0.0.1", serverUrl: "", portalSecret: "" } as any, + agentBoxManager: fakeAgentBoxManager(null), + frontendClient: fakeFrontendClient(), + credentialService: {} as any, + }); + try { + const list = server.rpcMethods.get("capability.testSessions")!; + const a = (await list({ run_id: "run-A" })) as any; + expect(a.sessions.map((s: any) => s.tid)).toEqual(["a-1", "a-2"]); // only run-A's + const b = (await list({ run_id: "run-B" })) as any; + expect(b.sessions.map((s: any) => s.tid)).toEqual(["b-1"]); // run-A's stay invisible + } finally { + delete process.env.SICLAW_COMPILE_BOX_ENDPOINT; + } + }); + it("returns an empty list when the box is absent — never spawns/rehydrates", async () => { const manager = fakeAgentBoxManager(null); server = await startRuntime({ diff --git a/src/gateway/server.ts b/src/gateway/server.ts index 753d3039..e9ccffdb 100644 --- a/src/gateway/server.ts +++ b/src/gateway/server.ts @@ -836,25 +836,34 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise capabilityRunManager.touch(runId), - }).catch((err) => { - // A dead test relay is disposable — log, never fail the authoring run. - console.warn( - `[capability] test relay ended run=${runId} tid=${opened.test_session_id}:`, - err instanceof Error ? err.message : String(err), - ); - }); + // On an idempotent replay the box returned an ALREADY-relayed session; its + // /test-events stream is single-consumer, so a second relay would split the + // frames. Only drive a freshly-opened session. + if (!opened.idempotent_replay) { + driveTestSession({ + client, + runId, + testSessionId: opened.test_session_id, + frontendClient, + touch: () => capabilityRunManager.touch(runId), + }).catch((err) => { + // A dead test relay is disposable — log, never fail the authoring run. + console.warn( + `[capability] test relay ended run=${runId} tid=${opened.test_session_id}:`, + err instanceof Error ? err.message : String(err), + ); + }); + } const res: CapabilityTestStartResponse = { run_id: runId, test_session_id: opened.test_session_id, @@ -1011,9 +1020,12 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise("/test-sessions"); + // Scope to THIS run: a SHARED box (SICLAW_COMPILE_BOX_ENDPOINT) hosts test + // sessions for multiple parent runs, and one run must never see (or reap) + // another's. Double protection with the consumer's own ParentRunID check. const response: CapabilityTestSessionsResponse = { run_id: req.run_id, - sessions: listed.sessions ?? [], + sessions: (listed.sessions ?? []).filter((s) => s.parent_run_id === req.run_id), }; return response; }); From a14548704368ecad39fab9d1b85950c86ac59337 Mon Sep 17 00:00:00 2001 From: likiosliu Date: Thu, 23 Jul 2026 12:35:43 +0800 Subject: [PATCH 4/5] Extract shouldRelayTestSession predicate + unit-test both branches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the testStart idempotency change: the relay-skip guard only fires on the rare idempotent-replay path, and a future refactor of the testStart handler could silently drop it — splitting the box's single-consumer /test-events stream. Extract the decision into a pure predicate (shouldRelayTestSession) in the relay module and unit-test both branches (replay → no relay; fresh open or flag absent → relay); the handler now calls the predicate instead of an inline check. Behavior unchanged. tsc clean; test-relay vitest green. --- src/gateway/capability/test-relay.test.ts | 16 +++++++++++++++- src/gateway/capability/test-relay.ts | 13 +++++++++++++ src/gateway/server.ts | 9 ++++----- 3 files changed, 32 insertions(+), 6 deletions(-) diff --git a/src/gateway/capability/test-relay.test.ts b/src/gateway/capability/test-relay.test.ts index 7b576fdc..b4dae2d5 100644 --- a/src/gateway/capability/test-relay.test.ts +++ b/src/gateway/capability/test-relay.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { driveTestSession } from "./test-relay.js"; +import { driveTestSession, shouldRelayTestSession } from "./test-relay.js"; import type { AgentBoxClient } from "../agentbox/client.js"; import type { FrontendWsClient } from "../frontend-ws-client.js"; import type { CapabilityEventFrame } from "./contract.js"; @@ -87,3 +87,17 @@ describe("driveTestSession", () => { expect(emitted).toHaveLength(2); }); }); + +describe("shouldRelayTestSession", () => { + it("does NOT relay an idempotent replay (already-relayed, single-consumer stream)", () => { + expect(shouldRelayTestSession({ idempotent_replay: true })).toBe(false); + }); + + it("relays a freshly-opened session", () => { + expect(shouldRelayTestSession({ idempotent_replay: false })).toBe(true); + }); + + it("relays when the flag is absent (older box that omits it)", () => { + expect(shouldRelayTestSession({})).toBe(true); + }); +}); diff --git a/src/gateway/capability/test-relay.ts b/src/gateway/capability/test-relay.ts index bcd4b618..6f00250b 100644 --- a/src/gateway/capability/test-relay.ts +++ b/src/gateway/capability/test-relay.ts @@ -39,6 +39,19 @@ export interface DriveTestSessionOptions { touch?: () => void; } +/** + * Whether the runtime should start a test-event relay for a just-returned + * testStart. An idempotent replay reuses an ALREADY-relayed session — the box's + * /test-events is single-consumer, so a second relay would split the frame + * stream — so a replay must NOT be relayed again. A fresh open (or an older box + * that omits the flag) IS relayed. Extracted as a pure predicate so this + * rare-and-subtle replay branch stays regression-proof if the testStart handler + * is later refactored (a dropped inline guard would silently split streams). + */ +export function shouldRelayTestSession(opened: { idempotent_replay?: boolean }): boolean { + return !opened.idempotent_replay; +} + /** * Relay the test-session event stream until the box closes it (`end`, emitted * on teardown/close). Errors propagate to the caller — which only logs: a dead diff --git a/src/gateway/server.ts b/src/gateway/server.ts index e9ccffdb..78eefdaf 100644 --- a/src/gateway/server.ts +++ b/src/gateway/server.ts @@ -28,7 +28,7 @@ import { getBoxProfile } from "./agentbox/box-profile.js"; import { buildSpawnEnv } from "./agentbox/spawn-env.js"; import { CapabilityRunManager } from "./capability/run-manager.js"; import { driveCapabilitySession } from "./capability/session-driver.js"; -import { driveTestSession } from "./capability/test-relay.js"; +import { driveTestSession, shouldRelayTestSession } from "./capability/test-relay.js"; import { CAPABILITY_GET_RUN, isTerminalCapabilityStatus } from "./capability/contract.js"; import type { CapabilityCancelRequest, @@ -846,10 +846,9 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise Date: Thu, 23 Jul 2026 12:47:45 +0800 Subject: [PATCH 5/5] Rename test-session idempotency field to client_request_id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Align the testStart idempotency key with sicore (MR!696) and the codebase's existing convention: every consumer-minted idempotency key here is a `_id` (message_id for a turn, command_id for a typed command, operation_id for a mutation receipt) — there is no `idempotency_key`. So the field is `client_request_id`, not the `idempotency_key` the first pass used. Behavior is unchanged: same client_request_id → same live session (checked before the cap and before packing), empty/absent → open fresh. Box read + TestRun attribute + teardown cleanup, the gateway request type + passthrough, and the box test's wire bodies all renamed. box suite 70 green; tsc clean; gateway vitest green. --- kbc/platform/pod/compile_box.py | 31 ++++++++++++++-------------- kbc/platform/pod/test_compile_box.py | 10 ++++----- src/gateway/capability/contract.ts | 13 ++++++------ src/gateway/server.ts | 6 +++--- 4 files changed, 31 insertions(+), 29 deletions(-) diff --git a/kbc/platform/pod/compile_box.py b/kbc/platform/pod/compile_box.py index 90066175..eee86646 100644 --- a/kbc/platform/pod/compile_box.py +++ b/kbc/platform/pod/compile_box.py @@ -101,10 +101,10 @@ # Read-only "test session" runs — ephemeral consumer sessions over a pinned draft # snapshot (test sessions). Parallel to RUNS, torn down on close/idle. See TestRun. TEST_SESSIONS: dict[str, "TestRun"] = {} -# Consumer-minted idempotency key → tid, for open-test-session retries. A retried -# testStart (same key) must return the SAME live session, never a second session -# or concurrency slot. Process-local; entries are cleared on teardown with their -# session (a torn-down key opens fresh, so a stale mapping can never be replayed). +# Consumer-minted client_request_id → tid, for open-test-session retries. A +# retried testStart (same key) must return the SAME live session, never a second +# session or concurrency slot. Process-local; entries are cleared on teardown with +# their session (a torn-down key opens fresh, so a stale mapping is never replayed). TEST_SESSION_IDEMPOTENCY: dict[str, str] = {} # One explicit red-team read-only review may inspect a run at a time. Test # recommendation and reference-answer assistance share this guard so two owner @@ -523,9 +523,9 @@ def __init__(self, tid: str, cwd: str, parent_run_id: str, snapshot_hash: str, l self._stall_reaped = False # Snapshot page count (returned by open, replayed on an idempotent open). self.pages: int = 0 - # Consumer-minted idempotency key that opened this session (if any), so + # Consumer-minted client_request_id that opened this session (if any), so # teardown can drop its TEST_SESSION_IDEMPOTENCY entry. - self.idempotency_key: str | None = None + self.client_request_id: str | None = None async def emit(self, ev: dict): await self.events.put(ev) @@ -4843,7 +4843,7 @@ async def _teardown_test_session(run: "TestRun"): # Drop the idempotency mapping WITH the session: a torn-down key must open a # fresh session, never replay a dead tid (guard on tid so a re-minted key for # a newer session is not clobbered). - key = run.idempotency_key + key = run.client_request_id if key and TEST_SESSION_IDEMPOTENCY.get(key) == run.tid: TEST_SESSION_IDEMPOTENCY.pop(key, None) if run.task and not run.task.done(): @@ -5398,14 +5398,15 @@ async def handle_open_test(request: web.Request): if not parent: return web.json_response({"error": "unknown run"}, status=404) body = await request.json() if request.body_exists else {} - # Idempotent open: a retried testStart (same consumer-minted key) returns the - # SAME live session. Checked BEFORE the cap (a replay must not be spuriously - # 429'd) and BEFORE packing (a replay does no snapshot work). `idempotent_replay` - # tells the runtime NOT to start a second event relay on the already-relayed - # session (the box's /test-events is single-consumer). Absent key → unchanged. - idem_key = (body.get("idempotency_key") or "").strip() + # Idempotent open: a retried testStart (same consumer-minted client_request_id) + # returns the SAME live session. Checked BEFORE the cap (a replay must not be + # spuriously 429'd) and BEFORE packing (a replay does no snapshot work). + # `idempotent_replay` tells the runtime NOT to start a second event relay on the + # already-relayed session (the box's /test-events is single-consumer). Absent + # key (empty/omitted) → unchanged old-consumer behavior. + idem_key = (body.get("client_request_id") or "").strip() if len(idem_key) > 128: - return web.json_response({"error": "idempotency_key must be at most 128 characters"}, status=400) + return web.json_response({"error": "client_request_id must be at most 128 characters"}, status=400) if idem_key: existing = TEST_SESSIONS.get(TEST_SESSION_IDEMPOTENCY.get(idem_key, "")) if existing is not None and not existing.done: @@ -5458,7 +5459,7 @@ async def handle_open_test(request: web.Request): ) run.pages = pages if idem_key: - run.idempotency_key = idem_key + run.client_request_id = idem_key TEST_SESSION_IDEMPOTENCY[idem_key] = tid TEST_SESSIONS[tid] = run run.task = asyncio.create_task(_test_session_wrapper(run)) diff --git a/kbc/platform/pod/test_compile_box.py b/kbc/platform/pod/test_compile_box.py index 54da59dc..bab53202 100644 --- a/kbc/platform/pod/test_compile_box.py +++ b/kbc/platform/pod/test_compile_box.py @@ -913,7 +913,7 @@ async def disconnect(self): async def test_open_test_session_idempotency(): - """A retried testStart (same idempotency_key) returns the SAME live session — + """A retried testStart (same client_request_id) returns the SAME live session — same tid/snapshot_hash/pages, flagged idempotent_replay, no new session, no new concurrency slot. A different key opens a new session. Teardown drops the key so a later same-key open starts fresh (never replays a dead tid).""" @@ -939,13 +939,13 @@ async def active(): compile_box.RUNS["p1"] = compile_box.CompileRun("p1", wd, 1) # first open with a key → a fresh session (idempotent_replay False) - b1 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-1"})).json() + b1 = await (await client.post("/test-session/p1", json={"client_request_id": "k-1"})).json() tid1 = b1["test_session_id"] assert b1["idempotent_replay"] is False, b1 n1 = await active() # retry SAME key → same tid + snapshot_hash + pages, replay flagged, count unchanged - b2 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-1"})).json() + b2 = await (await client.post("/test-session/p1", json={"client_request_id": "k-1"})).json() assert b2["test_session_id"] == tid1, (b1, b2) assert b2["snapshot_hash"] == b1["snapshot_hash"] and b2["pages"] == b1["pages"], (b1, b2) assert b2["idempotent_replay"] is True, b2 @@ -953,13 +953,13 @@ async def active(): assert len(compile_box.TEST_SESSIONS) == 1, compile_box.TEST_SESSIONS # a DIFFERENT key opens a new session - b3 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-2"})).json() + b3 = await (await client.post("/test-session/p1", json={"client_request_id": "k-2"})).json() assert b3["test_session_id"] != tid1 and len(compile_box.TEST_SESSIONS) == 2 # teardown drops the key → the same key opens FRESH afterwards (no dead replay) assert (await client.post(f"/test-session/{tid1}/close")).status == 200 assert "k-1" not in compile_box.TEST_SESSION_IDEMPOTENCY, compile_box.TEST_SESSION_IDEMPOTENCY - b4 = await (await client.post("/test-session/p1", json={"idempotency_key": "k-1"})).json() + b4 = await (await client.post("/test-session/p1", json={"client_request_id": "k-1"})).json() assert b4["test_session_id"] != tid1 and b4["idempotent_replay"] is False, b4 # no-key opens are never deduped (old-consumer behavior unchanged) diff --git a/src/gateway/capability/contract.ts b/src/gateway/capability/contract.ts index 6b4f60cc..6230699e 100644 --- a/src/gateway/capability/contract.ts +++ b/src/gateway/capability/contract.ts @@ -205,13 +205,14 @@ export interface CapabilityTestStartRequest { /** sha256 of the bundle bytes; the box verifies it at install time. */ bundle_sha256?: string; /** - * Optional consumer-minted idempotency key. A retried testStart with the same - * key returns the SAME live test session (same tid + snapshot_hash) instead of - * opening a second one / consuming a second concurrency slot — so a lost ack - * or a client retry cannot leave a ghost session. Absent → always open fresh - * (old-consumer behavior unchanged). + * Optional consumer-minted request id used as an idempotency key (naming + * mirrors message_id / command_id). A retried testStart with the same + * client_request_id returns the SAME live test session (same tid + + * snapshot_hash) instead of opening a second one / consuming a second + * concurrency slot — so a lost ack or a client retry cannot leave a ghost + * session. Absent/empty → always open fresh (old-consumer behavior unchanged). */ - idempotency_key?: string; + client_request_id?: string; } export interface CapabilityTestStartResponse { diff --git a/src/gateway/server.ts b/src/gateway/server.ts index 78eefdaf..b58e540b 100644 --- a/src/gateway/server.ts +++ b/src/gateway/server.ts @@ -836,9 +836,9 @@ export async function startRuntime(opts: StartRuntimeOptions): Promise