mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-23 16:36:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(cron): finalize compressed session tips (86e2darn2)

Browse source
This commit is contained in:
Colin Greig 2026-07-18 18:55:47 -04:00 • committed by Teknium
parent 50026fbaa1
commit 7fa795a674
3 changed files with 197 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

38
cron/scheduler.py
View file

@ -3627,6 +3627,30 @@ def run_job(
for _var_name in _cron_delivery_vars:
_VAR_MAP[_var_name].set("")
if _session_db:
# Compression can rotate the live agent onto a continuation while
# this run is in flight. Finalize that continuation, not the stale
# cron id captured before AIAgent started. SessionDB is the source
# of truth for the lineage; agent.session_id is only a fail-safe
# when the lookup itself is unavailable.
_final_cron_session_id = _cron_session_id
try:
_compression_tip = _session_db.get_compression_tip(
_cron_session_id
)
if _compression_tip:
_final_cron_session_id = _compression_tip
except (Exception, KeyboardInterrupt) as e:
try:
_agent_session_id = getattr(agent, "session_id", None)
if _agent_session_id:
_final_cron_session_id = _agent_session_id
except (Exception, KeyboardInterrupt):
pass
logger.debug(
"Job '%s': failed to resolve cron compression tip: %s",
job_id,
e,
)
# Title the cron session from the job (name -> id) and PERSIST it
# BEFORE end_session()/close() tear the connection down, so the
# close can never run over an in-flight title write (#50536). The
@ -3636,10 +3660,12 @@ def run_job(
try:
_title_base = " ".join(job_name.split())[:60].strip() or f"cron {job_id}"
_cron_title = f"{_title_base} · {_hermes_now().strftime('%b %d %H:%M')}"
if not _set_cron_session_title(_session_db, _cron_session_id, _cron_title):
if not _set_cron_session_title(
_session_db, _final_cron_session_id, _cron_title
):
# Helper returned None (blank base) -> use the id fallback.
_set_cron_session_title(
_session_db, _cron_session_id, f"cron {job_id}"
_session_db, _final_cron_session_id, f"cron {job_id}"
)
except (Exception, KeyboardInterrupt) as e:
logger.debug(
@ -3651,17 +3677,19 @@ def run_job(
getattr(_session_db, "get_next_title_in_lineage", lambda b: b)(
f"cron {job_id}"
),
f"cron {job_id} {_cron_session_id[-6:]}",
f"cron {job_id} {_final_cron_session_id[-6:]}",
):
try:
if _set_cron_session_title(
_session_db, _cron_session_id, _fallback
_session_db, _final_cron_session_id, _fallback
):
break
except (Exception, KeyboardInterrupt):
continue
try:
_session_db.end_session(_cron_session_id, "cron_complete")
_session_db.end_session(
_final_cron_session_id, "cron_complete"
)
except (Exception, KeyboardInterrupt) as e:
logger.debug("Job '%s': failed to end session: %s", job_id, e)
try:

113
tests/cron/test_scheduler.py
View file

@ -965,6 +965,7 @@ class TestRunJobSessionPersistence:
"prompt": "hello",
}
fake_db = MagicMock()
fake_db.get_compression_tip.side_effect = lambda session_id: session_id
with patch("cron.scheduler._hermes_home", tmp_path), \
patch("cron.scheduler._resolve_origin", return_value=None), \
@ -996,9 +997,11 @@ class TestRunJobSessionPersistence:
assert kwargs["session_db"] is fake_db
assert kwargs["platform"] == "cron"
assert kwargs["session_id"].startswith("cron_test-job_")
original_session_id = kwargs["session_id"]
fake_db.get_compression_tip.assert_called_once_with(original_session_id)
fake_db.end_session.assert_called_once()
call_args = fake_db.end_session.call_args
assert call_args[0][0].startswith("cron_test-job_")
assert call_args[0][0] == original_session_id
assert call_args[0][1] == "cron_complete"
fake_db.close.assert_called_once()
mock_agent.close.assert_called_once()
@ -1097,6 +1100,7 @@ class TestRunJobSessionPersistence:
"prompt": "summarize my inbox",
}
fake_db = MagicMock()
fake_db.get_compression_tip.side_effect = lambda session_id: session_id
with patch("cron.scheduler._hermes_home", tmp_path), \
patch("cron.scheduler._resolve_origin", return_value=None), \
@ -1135,6 +1139,7 @@ class TestRunJobSessionPersistence:
"prompt": "hello",
}
fake_db = MagicMock()
fake_db.get_compression_tip.return_value = "failure-compression-tip"
with patch("cron.scheduler._hermes_home", tmp_path), \
patch("cron.scheduler._resolve_origin", return_value=None), \
@ -1160,8 +1165,114 @@ class TestRunJobSessionPersistence:
assert success is False
assert final_response == ""
assert "RuntimeError: boom" in error
original_session_id = mock_agent_cls.call_args.kwargs["session_id"]
fake_db.get_compression_tip.assert_called_once_with(original_session_id)
assert fake_db.set_session_title.call_args.args[0] == "failure-compression-tip"
fake_db.end_session.assert_called_once_with(
"failure-compression-tip", "cron_complete"
)
mock_agent.close.assert_called_once()
def test_run_job_finalizes_compression_tip_and_dedupes_its_title(
self, tmp_path
):
job = {
"id": "compressing-job",
"name": "Compressed digest",
"prompt": "hello",
}
tip_session_id = "cron-compression-tip"
with self._run_job_patches(tmp_path) as (fake_db, mock_agent_cls):
fake_db.get_compression_tip.return_value = tip_session_id
fake_db.set_session_title.side_effect = [ValueError("in use"), True]
fake_db.get_next_title_in_lineage.return_value = (
"Compressed digest #2"
)
success, _output, _final_response, error = run_job(job)
assert success is True
assert error is None
original_session_id = mock_agent_cls.call_args.kwargs["session_id"]
fake_db.get_compression_tip.assert_called_once_with(original_session_id)
assert [call.args[0] for call in fake_db.set_session_title.call_args_list] == [
tip_session_id,
tip_session_id,
]
fake_db.get_next_title_in_lineage.assert_called_once()
fake_db.end_session.assert_called_once_with(
tip_session_id, "cron_complete"
)
@pytest.mark.parametrize(
("agent_session_id", "expected_suffix"),
[("agent-live-tip", "agent-live-tip"), ("", "original")],
)
def test_run_job_compression_tip_lookup_failure_falls_back_safely(
self, tmp_path, agent_session_id, expected_suffix
):
job = {
"id": "lookup-failure-job",
"name": "Lookup failure",
"prompt": "hello",
}
with self._run_job_patches(tmp_path) as (fake_db, mock_agent_cls):
mock_agent = mock_agent_cls.return_value
mock_agent.session_id = agent_session_id
fake_db.get_compression_tip.side_effect = RuntimeError("db busy")
success, _output, _final_response, error = run_job(job)
assert success is True
assert error is None
original_session_id = mock_agent_cls.call_args.kwargs["session_id"]
expected_session_id = (
agent_session_id
if expected_suffix == "agent-live-tip"
else original_session_id
)
assert fake_db.set_session_title.call_args.args[0] == expected_session_id
fake_db.end_session.assert_called_once_with(
expected_session_id, "cron_complete"
)
def test_run_job_timeout_finalizes_original_session(self, tmp_path, monkeypatch):
job = {
"id": "timeout-job",
"name": "Timeout",
"prompt": "hello",
}
monkeypatch.setenv("HERMES_CRON_TIMEOUT", "1")
with self._run_job_patches(tmp_path) as (fake_db, mock_agent_cls), \
patch(
"cron.scheduler.concurrent.futures.wait",
return_value=(set(), set()),
):
mock_agent = mock_agent_cls.return_value
mock_agent.get_activity_summary.return_value = {
"seconds_since_activity": 2.0,
"last_activity_desc": "api_call_streaming",
}
fake_db.get_compression_tip.return_value = "timeout-compression-tip"
success, _output, _final_response, error = run_job(job)
assert success is False
assert "TimeoutError" in error
original_session_id = mock_agent_cls.call_args.kwargs["session_id"]
mock_agent.interrupt.assert_called_once()
fake_db.get_compression_tip.assert_called_once_with(original_session_id)
assert (
fake_db.set_session_title.call_args.args[0]
== "timeout-compression-tip"
)
fake_db.end_session.assert_called_once_with(
"timeout-compression-tip", "cron_complete"
)
def test_run_job_reaps_stale_auxiliary_clients_per_tick(self, tmp_path):
# Regression: auxiliary clients bound to the cron worker's dead
# event loop must be reaped each tick. Without this, ``_client_cache``

52
tests/test_hermes_state.py
View file

@ -329,6 +329,58 @@ class TestSessionLifecycle:
assert session["end_reason"] == "compression"
assert session["ended_at"] == first_ended_at
def test_end_session_first_reason_wins_across_concurrent_connections(
self, db
):
"""Concurrent finalizers perform one transition, not last-write-wins."""
import threading
db.create_session(session_id="s1", source="cron")
db._conn.execute(
"CREATE TABLE session_end_audit (reason TEXT NOT NULL)"
)
db._conn.execute(
"""
CREATE TRIGGER audit_session_end
AFTER UPDATE OF ended_at ON sessions
WHEN OLD.ended_at IS NULL AND NEW.ended_at IS NOT NULL
BEGIN
INSERT INTO session_end_audit(reason) VALUES (NEW.end_reason);
END
"""
)
peer = SessionDB(db_path=db.db_path)
barrier = threading.Barrier(2)
errors = []
def _end(session_db, reason):
try:
barrier.wait(timeout=5)
session_db.end_session("s1", reason)
except BaseException as exc:
errors.append(exc)
threads = [
threading.Thread(target=_end, args=(db, "compression")),
threading.Thread(target=_end, args=(peer, "cron_complete")),
]
try:
for thread in threads:
thread.start()
for thread in threads:
thread.join(timeout=10)
assert all(not thread.is_alive() for thread in threads)
assert errors == []
audit_rows = db._conn.execute(
"SELECT reason FROM session_end_audit"
).fetchall()
assert len(audit_rows) == 1
assert db.get_session("s1")["end_reason"] == audit_rows[0]["reason"]
finally:
peer.close()
def test_end_session_after_reopen_allows_re_end(self, db):
"""reopen_session() is the explicit escape hatch for re-ending a
closed session. After reopen, end_session() takes effect again.