David/HERMES-GIELIE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: handle remaining stale runtime entrypoints

Browse source
This commit is contained in:
pxxD1998 2026-07-16 18:35:11 +08:00
parent 66dceb10e5
commit 0bab6a3d7a
5 changed files with 280 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
api/routes.py
View file

@ -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)

146
tests/test_agent_runtime_revision_guard.py
View file

@ -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

12
tests/test_goal_command_webui.py
View file

@ -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):

54
tests/test_issue1013_handoff_dock.py
View file

@ -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

32
tests/test_sprint46.py
View file

@ -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 = [