fix(cron): mark session as ended after job completes (#2998)

Cron was the only execution path that never called end_session(),
leaving ended_at = NULL permanently. This made cron sessions invisible
to hermes prune --older-than and indistinguishable from active sessions.

Captures session_id in a local variable before agent construction so
it's available in the finally block even if AIAgent() fails, then calls
end_session(session_id, 'cron_complete') before close().

Cherry-picked from PR #2979 by ygd58. Fixed bug: original PR called
end_session() with zero arguments (TypeError — method requires
session_id and end_reason).

Fixes #2972.

Co-authored-by: ygd58 <ygd58@users.noreply.github.com>

cron/scheduler.py

@@ -280,6 +280,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
     job_name = job["name"]
     prompt = _build_job_prompt(job)
     origin = _resolve_origin(job)
+    _cron_session_id = f"cron_{job_id}_{_hermes_now().strftime('%Y%m%d_%H%M%S')}"
 
     logger.info("Running job '%s' (ID: %s)", job_name, job_id)
     logger.info("Prompt: %s", prompt[:100])
@@ -411,7 +412,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             disabled_toolsets=["cronjob", "messaging", "clarify"],
             quiet_mode=True,
             platform="cron",
-            session_id=f"cron_{job_id}_{_hermes_now().strftime('%Y%m%d_%H%M%S')}",
+            session_id=_cron_session_id,
             session_db=_session_db,
         )
         
@@ -476,6 +477,10 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
         ):
             os.environ.pop(key, None)
         if _session_db:
+            try:
+                _session_db.end_session(_cron_session_id, "cron_complete")
+            except Exception as e:
+                logger.debug("Job '%s': failed to end session: %s", job_id, e)
             try:
                 _session_db.close()
             except Exception as e:

tests/cron/test_scheduler.py

@@ -254,6 +254,10 @@ class TestRunJobSessionPersistence:
         assert kwargs["session_db"] is fake_db
         assert kwargs["platform"] == "cron"
         assert kwargs["session_id"].startswith("cron_test-job_")
+        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][1] == "cron_complete"
         fake_db.close.assert_called_once()
 
     def test_run_job_empty_response_returns_empty_not_placeholder(self, tmp_path):