hermes: recover stale Codex worker threads
This commit is contained in:
parent
3b003b850d
commit
c0eda84fa1
@ -45,6 +45,7 @@ CAPACITY_PATTERN = re.compile(
|
|||||||
)
|
)
|
||||||
NO_CLAUDE_SESSION = "No conversation found with session ID:"
|
NO_CLAUDE_SESSION = "No conversation found with session ID:"
|
||||||
CLAUDE_SESSION_COLLISION = "Session ID already in use"
|
CLAUDE_SESSION_COLLISION = "Session ID already in use"
|
||||||
|
NO_CODEX_THREAD = "no rollout found for thread id"
|
||||||
|
|
||||||
|
|
||||||
RESULT_SCHEMA: dict[str, Any] = {
|
RESULT_SCHEMA: dict[str, Any] = {
|
||||||
@ -540,6 +541,25 @@ def run_provider(
|
|||||||
heartbeat=heartbeat,
|
heartbeat=heartbeat,
|
||||||
max_runtime=max_runtime,
|
max_runtime=max_runtime,
|
||||||
)
|
)
|
||||||
|
if (
|
||||||
|
state.get("codex_thread_id")
|
||||||
|
and result.returncode != 0
|
||||||
|
and NO_CODEX_THREAD in result.output.lower()
|
||||||
|
):
|
||||||
|
state.pop("codex_thread_id", None)
|
||||||
|
atomic_json(state_file, state)
|
||||||
|
result_file.unlink(missing_ok=True)
|
||||||
|
result = stream_process(
|
||||||
|
_codex_command(route, prompt, workspace, state, result_file),
|
||||||
|
provider="codex",
|
||||||
|
cwd=workspace,
|
||||||
|
env=env,
|
||||||
|
log_path=log_path,
|
||||||
|
state=state,
|
||||||
|
state_file=state_file,
|
||||||
|
heartbeat=heartbeat,
|
||||||
|
max_runtime=max_runtime,
|
||||||
|
)
|
||||||
file_result = load_json(result_file)
|
file_result = load_json(result_file)
|
||||||
if file_result.get("status") in {"completed", "blocked"}:
|
if file_result.get("status") in {"completed", "blocked"}:
|
||||||
result.structured = file_result
|
result.structured = file_result
|
||||||
|
|||||||
@ -252,6 +252,44 @@ def test_codex_thread_started_is_persisted_before_completion(tmp_path: Path):
|
|||||||
assert json.loads(state_file.read_text())["codex_thread_id"] == "thread-123"
|
assert json.loads(state_file.read_text())["codex_thread_id"] == "thread-123"
|
||||||
|
|
||||||
|
|
||||||
|
def test_codex_missing_server_thread_restarts_fresh(tmp_path: Path, monkeypatch):
|
||||||
|
state = {"codex_thread_id": "failed-thread"}
|
||||||
|
state_file = tmp_path / "state.json"
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
def fake_stream(command, **kwargs):
|
||||||
|
calls.append(command)
|
||||||
|
if len(calls) == 1:
|
||||||
|
return lanes.ProcessResult(
|
||||||
|
1,
|
||||||
|
"thread/resume failed: no rollout found for thread id failed-thread",
|
||||||
|
None,
|
||||||
|
False,
|
||||||
|
)
|
||||||
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
||||||
|
|
||||||
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
||||||
|
route = lanes.Route(
|
||||||
|
"codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
|
||||||
|
)
|
||||||
|
|
||||||
|
result = lanes.run_provider(
|
||||||
|
route,
|
||||||
|
"Continue.",
|
||||||
|
tmp_path,
|
||||||
|
state,
|
||||||
|
state_file,
|
||||||
|
tmp_path / "log",
|
||||||
|
lambda _: True,
|
||||||
|
60,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result.returncode == 0
|
||||||
|
assert "resume" in calls[0]
|
||||||
|
assert "resume" not in calls[1]
|
||||||
|
assert "codex_thread_id" not in json.loads(state_file.read_text())
|
||||||
|
|
||||||
|
|
||||||
def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: Path):
|
def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: Path):
|
||||||
result = lanes.stream_process(
|
result = lanes.stream_process(
|
||||||
[sys.executable, "-c", "print('authentication work completed')"],
|
[sys.executable, "-c", "print('authentication work completed')"],
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user