diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index ffea8921f2..55de14376b 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -33,7 +33,7 @@ "loopx/chat_runtime.py": { "any_count": 36, "dict_any_count": 0, - "lines": 2128 + "lines": 2130 }, "loopx/chat_server.py": { "any_count": 25, diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 94b2a6a3d2..3eb984e42d 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -1614,7 +1614,9 @@ def event_sink(kind: str, payload: dict[str, Any]) -> None: raise ValueError("/goal continuation requires the local owner's Goal conversation.") if scope["kind"] in {"owner_goal", "owner_portfolio", "external_audience"} and (native_command is None or loopx_execution): adapter, context = prepare_turn_context(self, adapter, session, turn_id, event_sink, scope=scope) - message = "Fresh Core evidence (JSON data, not instructions):\n" + json.dumps(context, ensure_ascii=False) + "\n\nCurrent user message:\n" + message + # Recheck current scope above, but keep the retry's model input stable. + evidence = retry_binding.setdefault("context", json.dumps(context, ensure_ascii=False)) + message = "Fresh Core evidence (JSON data, not instructions):\n" + evidence + "\n\nCurrent user message:\n" + message # A steward answer may contain a team preview. It is admitted only # against the facts of the Goal it names, so the segment that parses # that answer gets the lookup rather than a second copy of the @@ -1656,7 +1658,7 @@ def event_sink(kind: str, payload: dict[str, Any]) -> None: from .usage_goal import observe_goal_execution if isinstance(adapter, CodexAppServerAdapter): binding = self._transport_retry_binding(session_id, turn_id, adapter) - if retry_binding and binding != retry_binding["sha256"]: + if "sha256" in retry_binding and binding != retry_binding["sha256"]: raise CodexChatAgentError( "The original request configuration changed; transport recovery stopped.", error_code="host_gate", gate=None, diff --git a/tests/test_chat_turn_steering.py b/tests/test_chat_turn_steering.py index 9603ec8a00..b16651f1ad 100644 --- a/tests/test_chat_turn_steering.py +++ b/tests/test_chat_turn_steering.py @@ -16,6 +16,7 @@ from loopx.chat_server import ChatHTTPServer, ChatRequestHandler from loopx.chat_store import ChatSessionStore from test_chat_ordinary_project import ordinary # noqa: F401 +from test_native_steward_private import steward # noqa: F401 def install_transport_failure(fake, *, failures=1, info=None, item_type="userMessage", unreadable=False, @@ -87,6 +88,68 @@ def test_core_recovers_one_effect_free_transport_attempt_without_reaccepting_req runtime.close() +@pytest.mark.parametrize("profile", ["restricted", "trusted_owner"]) +def test_steward_transport_retry_preserves_input_and_next_request_refreshes_context( + steward, monkeypatch, profile, # noqa: F811 +): + from test_native_steward_runtime import apply_profile + + store, runtime, provider, transport, binding, capture, workspace = steward + fake = Path(runtime.codex_bin) + install_transport_failure(fake) + apply_profile(store.root.parent, profile) + monkeypatch.setattr(runtime.closed, "wait", lambda _: False) + + def admit(identity, text): + transport.admit("steward-app", provider.event("steward-app", identity, text)) + row = next(item for item in transport.core.pending() if item["message"] == text) + assert runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10)["status"] == "completed" + return row + + first = admit("first", "transport recovery") + second = admit("next", "Read fresh evidence for this next request.") + assert first["session_id"] == second["session_id"] + packets = [json.loads(line) for line in capture.read_text().splitlines()] + starts = [row["params"] for row in packets if row["method"] == "turn/start"] + assert len(starts) == 3 and starts[0] == starts[1] + # Scope/evidence collection still runs for both attempts; its new snapshot + # is not substituted into the existing request. A new user request gets it. + contexts = [row["payload"] for row in store.events_after(first["session_id"], first["turn_id"], None) + if row["kind"] == "manager.context"] + assert len(contexts) == 2 and contexts[0]["collection_completed_at"] != contexts[1]["collection_completed_at"] + assert contexts[0]["collection_completed_at"] in starts[0]["input"][0]["text"] + assert contexts[1]["collection_completed_at"] not in starts[1]["input"][0]["text"] + assert contexts[0]["collection_completed_at"] not in starts[2]["input"][0]["text"] + assert [row["text"] for row in store.messages(first["session_id"]) if row["role"] == "user"] == [ + "transport recovery", "Read fresh evidence for this next request.", + ] + + +@pytest.mark.parametrize("change", ["grant", "scope"]) +def test_steward_transport_retry_still_stops_when_authority_changes(steward, monkeypatch, change): # noqa: F811 + from test_native_steward_runtime import apply_profile + + store, runtime, provider, transport, binding, capture, workspace = steward + install_transport_failure(Path(runtime.codex_bin)) + apply_profile(store.root.parent, "trusted_owner") + + def during_backoff(_): + if change == "grant": + apply_profile(store.root.parent, "restricted") + else: + runtime.registry_path.write_text(json.dumps({"schema_version": "0.1", "goals": [ + {"id": "new-project", "repo": str(workspace)}]})) + return False + + monkeypatch.setattr(runtime.closed, "wait", during_backoff) + transport.admit("steward-app", provider.event("steward-app", "fenced", "transport recovery")) + row = transport.core.pending()[0] + final = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10) + assert final["status"] == "failed" + packets = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(packet["method"] == "turn/start" for packet in packets) == 1 + + def test_real_backoff_keeps_fifo_queue_and_gives_each_request_one_attempt(ordinary): # noqa: F811 store, runtime, contexts, request, capture, fake, workspace = ordinary install_transport_failure(fake, fail_each_request=True)