From 0bab6a3d7a541dd3204cc85d18c123e5de5076ee Mon Sep 17 00:00:00 2001 From: pxxD1998 <214340659+pxxD1998@users.noreply.github.com> Date: Thu, 16 Jul 2026 18:35:11 +0800 Subject: [PATCH] fix: handle remaining stale runtime entrypoints --- api/routes.py | 41 +++++- tests/test_agent_runtime_revision_guard.py | 146 +++++++++++++++++++++ tests/test_goal_command_webui.py | 12 +- tests/test_issue1013_handoff_dock.py | 54 ++++++++ tests/test_sprint46.py | 32 +++++ 5 files changed, 280 insertions(+), 5 deletions(-) diff --git a/api/routes.py b/api/routes.py index 698325b8..fc851642 100644 --- a/api/routes.py +++ b/api/routes.py @@ -20782,8 +20782,14 @@ def _active_run_stream_for_session(session_id: str | None) -> str | None: return None -def _agent_runtime_barrier_response(*, runner_local_owned: bool = False) -> dict | None: +def _agent_runtime_barrier_response( + *, + runner_local_owned: bool = False, + external_runtime_owned: bool | None = None, +) -> dict | None: """Return the typed stale-runtime response for local in-process turns.""" + if external_runtime_owned is True: + return None if runner_local_owned and webui_gateway_chat_enabled(get_config()): return None from api.runtime_adapter import runtime_adapter_runner_enabled @@ -20814,11 +20820,14 @@ def _start_chat_stream_for_session( goal_related: bool = False, source: str = "webui", moa_config=None, - external_runtime_owned: bool = False, + external_runtime_owned: bool | None = None, ): """Persist pending state, register an SSE channel, and start an agent turn.""" + if external_runtime_owned is None: + external_runtime_owned = webui_gateway_chat_enabled(get_config()) + backend_is_gateway = bool(external_runtime_owned) stale_response = _agent_runtime_barrier_response( - runner_local_owned=external_runtime_owned, + external_runtime_owned=backend_is_gateway, ) if stale_response is not None: stale_response["_status"] = 409 @@ -20941,7 +20950,6 @@ def _start_chat_stream_for_session( if goal_related: STREAM_GOAL_RELATED[stream_id] = True diag.stage("worker_thread_start") if diag else None - backend_is_gateway = webui_gateway_chat_enabled(get_config()) worker_target = _run_gateway_chat_streaming if backend_is_gateway else _run_agent_streaming worker_kwargs = {"model_provider": model_provider, "goal_related": goal_related} if moa_config and not backend_is_gateway: @@ -21686,6 +21694,7 @@ def _handle_goal_command(handler, body): model_provider=model_provider, normalized_model=normalized_model, goal_related=True, + external_runtime_owned=webui_gateway_chat_enabled(get_config()), ) status = int(stream_response.pop("_status", 200) or 200) payload.update(stream_response) @@ -22737,6 +22746,12 @@ def _handle_git_commit_message(handler, body): return bad(handler, str(e)) except GitWorkspaceError as e: return _git_bad(handler, e) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.exception("git commit message generation failed") return bad(handler, _sanitize_error(e), 500) @@ -22768,6 +22783,12 @@ def _handle_git_commit_message_selected(handler, body): return bad(handler, str(e)) except GitWorkspaceError as e: return _git_bad(handler, e) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.exception("selected git commit message generation failed") return bad(handler, _sanitize_error(e), 500) @@ -24230,6 +24251,12 @@ def _handle_session_compress(handler, body): "focus_topic": focus_topic, }, ) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.warning("Manual session compression failed: %s", e) return bad(handler, f"Compression failed: {_sanitize_error(e)}") @@ -24879,6 +24906,12 @@ def _handle_handoff_summary(handler, body): "rounds": rounds, "fallback": fallback, }) + except AgentRuntimeChangedError as e: + return j(handler, { + "error": str(e), + "type": "agent_runtime_stale", + "retryable": True, + }, status=409) except Exception as e: logger.warning("Handoff summary generation failed: %s", e) summary_text = _fallback_handoff_summary(msgs) diff --git a/tests/test_agent_runtime_revision_guard.py b/tests/test_agent_runtime_revision_guard.py index 10ad81ae..625f3e59 100644 --- a/tests/test_agent_runtime_revision_guard.py +++ b/tests/test_agent_runtime_revision_guard.py @@ -371,6 +371,152 @@ def test_runner_owned_start_run_does_not_enter_local_stream_barrier(monkeypatch) assert len(calls) == 1 +@pytest.mark.parametrize("gateway_owned", [False, True]) +def test_stream_admission_uses_one_gateway_ownership_snapshot(monkeypatch, gateway_owned): + """The barrier and worker must share one immutable backend decision.""" + from api import routes + from api import turn_journal + from api.config import unregister_stream_owner + + gateway_reads = [] + revision_checks = [] + worker_targets = [] + session = types.SimpleNamespace( + session_id="gateway-snapshot", + profile=None, + title="Gateway snapshot", + active_stream_id=None, + pending_started_at=None, + ) + + def read_gateway(_cfg): + gateway_reads.append(True) + if len(gateway_reads) > 1: + raise AssertionError("gateway ownership was re-read after admission") + return gateway_owned + + def prepare(current, **kwargs): + current.active_stream_id = kwargs["stream_id"] + current.pending_started_at = 123.0 + + class FakeThread: + def __init__(self, *, target, args, kwargs, daemon): + worker_targets.append(target) + + def start(self): + return None + + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", read_gateway) + monkeypatch.setattr(routes, "get_config", lambda: {}) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: revision_checks.append(True), + ) + monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda _sid: None) + monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda _session: False) + monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", prepare) + monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace: None) + monkeypatch.setattr(routes.threading, "Thread", FakeThread) + monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *_args, **_kwargs: {}) + + response = routes._start_chat_stream_for_session( + session, + msg="goal", + attachments=[], + workspace="/tmp/workspace", + model="test-model", + goal_related=True, + ) + + try: + assert response["session_id"] == session.session_id + assert len(gateway_reads) == 1 + assert revision_checks == ([] if gateway_owned else [True]) + expected_worker = ( + routes._run_gateway_chat_streaming + if gateway_owned + else routes._run_agent_streaming + ) + assert worker_targets == [expected_worker] + finally: + stream_id = str(response.get("stream_id") or "") + with routes.STREAMS_LOCK: + routes.STREAMS.pop(stream_id, None) + unregister_stream_owner(stream_id) + routes.STREAM_GOAL_RELATED.pop(stream_id, None) + + +@pytest.mark.parametrize( + ("handler_name", "body"), + [ + ("_handle_git_commit_message", {"session_id": "git-message"}), + ( + "_handle_git_commit_message_selected", + {"session_id": "git-message", "paths": ["selected.py"]}, + ), + ], +) +def test_git_commit_message_stale_runtime_returns_typed_409( + monkeypatch, tmp_path, handler_name, body +): + """Commit-message generation must preserve the stale-runtime contract.""" + from api import routes + from api import workspace_git + + session = types.SimpleNamespace(workspace=str(tmp_path)) + monkeypatch.setattr(routes, "require", lambda _body, *keys: None) + monkeypatch.setattr(routes, "get_session", lambda _sid: session) + monkeypatch.setattr(routes, "_git_paths_from_body", lambda _body: ["selected.py"]) + monkeypatch.setattr( + workspace_git, + "staged_commit_message_prompt", + lambda _workspace: {"system_prompt": "system", "user_prompt": "user"}, + ) + monkeypatch.setattr( + workspace_git, + "selected_commit_message_prompt", + lambda _workspace, _paths: { + "system_prompt": "system", + "user_prompt": "user", + }, + ) + monkeypatch.setattr( + routes, + "_llm_git_commit_message", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + monkeypatch.setattr( + routes, + "j", + lambda _handler, payload, status=200, **_kwargs: { + "status": status, + "payload": payload, + }, + ) + monkeypatch.setattr( + routes, + "bad", + lambda _handler, message, status=400: { + "status": status, + "payload": {"error": message}, + }, + ) + + response = getattr(routes, handler_name)(object(), body) + + assert response == { + "status": 409, + "payload": { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + }, + } + + def test_gateway_owned_start_run_bypasses_local_runtime_barrier(monkeypatch): """Gateway chat must reach its worker without inspecting the local Agent.""" from api import routes diff --git a/tests/test_goal_command_webui.py b/tests/test_goal_command_webui.py index 2389c08b..b16f3892 100644 --- a/tests/test_goal_command_webui.py +++ b/tests/test_goal_command_webui.py @@ -161,7 +161,10 @@ def test_goal_continuation_decision_emits_status_and_normal_user_prompt(monkeypa assert decision["continuation_prompt"].startswith("[Continuing toward your standing goal]") -def test_goal_endpoint_sets_goal_and_starts_kickoff_stream(monkeypatch, tmp_path): +@pytest.mark.parametrize("gateway_owned", [False, True]) +def test_goal_endpoint_sets_goal_and_starts_kickoff_stream( + monkeypatch, tmp_path, gateway_owned +): """POST /api/goal uses GoalManager state and launches the first goal turn.""" from api import goals as webui_goals from api import routes @@ -199,6 +202,12 @@ def test_goal_endpoint_sets_goal_and_starts_kickoff_stream(monkeypatch, tmp_path monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager) monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession()) monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path) + monkeypatch.setattr( + routes, + "webui_gateway_chat_enabled", + lambda _cfg: gateway_owned, + ) + monkeypatch.setattr(routes, "get_config", lambda: {}) monkeypatch.setattr( routes, "_resolve_compatible_session_model_state", @@ -229,6 +238,7 @@ def test_goal_endpoint_sets_goal_and_starts_kickoff_stream(monkeypatch, tmp_path assert result["payload"]["stream_id"] == "goal-stream" assert started and started[0]["msg"] == "ship the feature" assert started[0]["model_provider"] == "openai-codex" + assert started[0]["external_runtime_owned"] is gateway_owned def test_goal_endpoint_preserves_response_shape_under_runtime_adapter_flag(monkeypatch, tmp_path): diff --git a/tests/test_issue1013_handoff_dock.py b/tests/test_issue1013_handoff_dock.py index 39121ebc..2a970bac 100644 --- a/tests/test_issue1013_handoff_dock.py +++ b/tests/test_issue1013_handoff_dock.py @@ -248,6 +248,60 @@ def test_no_api_key_handoff_summary_persists_fallback_summary(monkeypatch): assert persisted[0]["rounds"] == models.CONVERSATION_ROUND_THRESHOLD +def test_stale_runtime_handoff_summary_returns_typed_409_without_persisting(monkeypatch): + """A stale runtime must remain retryable instead of becoming fallback success.""" + import api.routes as routes + import api.models as models + + monkeypatch.setattr(routes, "require", lambda body, *keys: None) + monkeypatch.setattr( + routes, + "j", + lambda _handler, payload, status=200, extra_headers=None: { + "status": status, + "payload": payload, + }, + ) + monkeypatch.setattr( + models, + "count_conversation_rounds", + lambda sid, since=None: models.CONVERSATION_ROUND_THRESHOLD, + ) + monkeypatch.setattr( + models, + "get_cli_session_messages", + lambda sid: [ + {"role": "user", "content": "Need a handoff", "timestamp": 1.0}, + {"role": "assistant", "content": "Context is ready", "timestamp": 2.0}, + ], + ) + monkeypatch.setattr( + routes, + "_persist_handoff_summary", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("stale runtime persisted a fallback summary") + ), + ) + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + + response = routes._handle_handoff_summary(object(), {"session_id": "stale-handoff"}) + + assert response == { + "status": 409, + "payload": { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + }, + } + + def test_exception_handoff_summary_persists_fallback_summary(monkeypatch): """Unhandled summary exception should still persist a fallback handoff marker.""" import api.routes as routes diff --git a/tests/test_sprint46.py b/tests/test_sprint46.py index 272f8668..ff10828e 100644 --- a/tests/test_sprint46.py +++ b/tests/test_sprint46.py @@ -129,6 +129,38 @@ def test_session_compress_requires_session_id(cleanup_test_sessions): assert handler.payload()["error"] == "Missing required field(s): session_id" +def test_session_compress_stale_runtime_returns_typed_409_before_mutation( + monkeypatch, cleanup_test_sessions +): + import api.routes as routes + + sid = _make_session() + cleanup_test_sessions.append(sid) + loaded_before = Session.load(sid) + assert loaded_before is not None + before = loaded_before.compact() + monkeypatch.setattr( + routes, + "ensure_agent_runtime_current", + lambda: (_ for _ in ()).throw( + routes.AgentRuntimeChangedError("restart required") + ), + ) + + handler = _FakeHandler() + _handle_session_compress(handler, {"session_id": sid}) + + assert handler.status == 409 + assert handler.payload() == { + "error": "restart required", + "type": "agent_runtime_stale", + "retryable": True, + } + loaded_after = Session.load(sid) + assert loaded_after is not None + assert loaded_after.compact() == before + + def test_session_compress_roundtrip(monkeypatch, cleanup_test_sessions): created = cleanup_test_sessions original_messages = [