diff --git a/cron/scheduler.py b/cron/scheduler.py index c5957070021e..4c872c855e08 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -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: diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index f5317bf2820f..bf99f4153ad3 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -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`` diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 38222176b973..edeec0332d30 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -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.