fix: notify origins and close runs for work interrupted by a gateway restart

src/praisonai-agents/praisonaiagents/background/job_manager.py

@@ -92,13 +92,48 @@ def list_unreconciled(self) -> List[JobInfo]:
 
         Specifically: jobs left ``PENDING`` or ``RUNNING`` at the last write
         (their process died before or during the run — both are orphaned and
-        reconciled to ``LOST``) and jobs ``COMPLETED`` with an ``origin`` that
-        were never marked ``delivered``. Returned in a stable order for
+        reconciled to ``LOST``) and jobs ``COMPLETED`` or ``LOST`` with an
+        ``origin`` that were never marked ``delivered``. Returned in a stable order for
         deterministic replay.
         """
         ...
 
 
+def _emit_job_completed(info: "JobInfo") -> None:
+    """Emit the JOB_COMPLETED hook for a job that reached a terminal state.
+
+    Fired for every background job, not only the ones carrying a ``deliver``
+    target: ``on_complete`` is a delivery callback the gateway may or may not
+    install, whereas ``HookEvent.JOB_COMPLETED`` is the observability contract
+    plugins subscribe to. Also fired once when restart reconciliation marks an
+    orphaned job ``LOST``. Best-effort -- a missing hooks package or a raising
+    subscriber never affects the job's own result.
+    """
+    try:
+        from praisonaiagents.hooks import fire_hook
+
+        fire_hook(
+            "job_completed",
+            {
+                "job_id": info.job_id,
+                "status": getattr(info.status, "value", str(info.status)),
+                "result": info.result,
+                "error": info.error,
+                "agent_name": "background",
+                "session_id": str(info.origin.get("session_id", "") or ""),
+                "deliver": str(info.origin.get("deliver", "") or ""),
+                "platform": str(info.origin.get("platform", "") or ""),
+                "chat_id": str(info.origin.get("chat_id", "") or ""),
+                "thread_id": str(info.origin.get("thread_id", "") or ""),
+            },
+        )
+    except Exception as e:  # noqa: BLE001 - observability must not break the job
+        logger.debug(
+            "JOB_COMPLETED hook emit for job %s failed (non-fatal): %s",
+            info.job_id, e,
+        )
+
+
 class BackgroundJobManager:
     """
     Manages background jobs for agent tasks.
@@ -239,43 +274,8 @@ def start_job(
         # and first run still leaves a recoverable trace (reconciled as LOST).
         self._persist(job_info)
 
-        def _emit_job_completed() -> None:
-            """Emit the JOB_COMPLETED hook for a job that reached a terminal state.
-
-            Fired for every background job, not only the ones carrying a
-            ``deliver`` target: ``on_complete`` is a delivery callback the
-            gateway may or may not install, whereas ``HookEvent.JOB_COMPLETED``
-            is the observability contract plugins subscribe to. Best-effort --
-            a missing hooks package or a raising subscriber never affects the
-            job's own result.
-            """
-            info = job_info
-            try:
-                from praisonaiagents.hooks import fire_hook
-
-                fire_hook(
-                    "job_completed",
-                    {
-                        "job_id": info.job_id,
-                        "status": getattr(info.status, "value", str(info.status)),
-                        "result": info.result,
-                        "error": info.error,
-                        "agent_name": "background",
-                        "session_id": str(info.origin.get("session_id", "") or ""),
-                        "deliver": str(info.origin.get("deliver", "") or ""),
-                        "platform": str(info.origin.get("platform", "") or ""),
-                        "chat_id": str(info.origin.get("chat_id", "") or ""),
-                        "thread_id": str(info.origin.get("thread_id", "") or ""),
-                    },
-                )
-            except Exception as e:  # noqa: BLE001 - observability must not break the job
-                logger.debug(
-                    "JOB_COMPLETED hook emit for job %s failed (non-fatal): %s",
-                    job_id, e,
-                )
-
         def _fire_complete() -> None:
-            _emit_job_completed()
+            _emit_job_completed(job_info)
             if on_complete is None:
                 return
             info = job_info
@@ -499,21 +499,24 @@ def reconcile_on_start(
           :attr:`JobStatus.LOST` (a terminal, queryable state) and persisted.
           The abandoned work is not automatically re-run — that decision is
           left to the caller, who can inspect ``LOST`` jobs and re-submit.
-        - A job that ``COMPLETED`` with an ``origin`` but was never
-          ``delivered`` has ``redeliver`` invoked with its :class:`JobInfo`,
-          re-firing the deliver-back-to-origin the crash interrupted. On a
-          successful redeliver the job is marked ``delivered`` and persisted so
-          it is not delivered twice on the next restart.
+          ``JOB_COMPLETED`` fires once with ``status="lost"``.
+        - A job that is ``COMPLETED`` or ``LOST`` with an ``origin`` but was
+          never ``delivered`` has ``redeliver`` invoked with its
+          :class:`JobInfo`, so the origin gets the result or an "interrupted"
+          notice. On a successful redeliver the job is marked ``delivered`` and
+          persisted so it is not delivered twice on the next restart.
 
         Every persisted job is also re-hydrated into this manager's in-memory
         map so it remains queryable by id after the restart.
 
         No-op (returns zero counts) when no store is configured. Safe to call
-        once at startup, mirroring ``OutboundQueue.drain`` wiring.
+        again (jobs live in this process are skipped), mirroring
+        ``OutboundQueue.drain`` wiring.
 
         Args:
-            redeliver: Callable invoked with a COMPLETED-but-undelivered
-                :class:`JobInfo` to replay its deliver-back. Best-effort — an
+            redeliver: Callable invoked with a COMPLETED- or LOST-but-undelivered
+                :class:`JobInfo` (check ``status``) to replay its deliver-back
+                or announce the interruption. Best-effort — an
                 exception it raises is logged and the job is left undelivered
                 for a future retry, never crashing startup. When ``None``,
                 undelivered jobs are re-hydrated but not delivered.
@@ -532,6 +535,14 @@ def reconcile_on_start(
             return counts
 
         for job in unreconciled:
+            # A repeat call (e.g. the gateway's boot pass after the lazy one in
+            # get_job_manager) must not touch jobs started by this process,
+            # whatever their status: the snapshot may predate a completion.
+            # Only start_job registers a future, so it marks ownership.
+            with self._lock:
+                owned = job.job_id in self._futures
+            if owned:
+                continue
             # A job left RUNNING (or still PENDING) when its process died was
             # orphaned by the crash. Transition it to the terminal LOST state
             # *before* it becomes visible in the in-memory map, so a concurrent
@@ -542,6 +553,8 @@ def reconcile_on_start(
             if orphaned:
                 prior_status = job.status.value
                 job.status = JobStatus.LOST
+                if job.error is None:
+                    job.error = "Interrupted by a process restart"
                 if job.completed_at is None:
                     job.completed_at = time.time()
 
@@ -556,8 +569,13 @@ def reconcile_on_start(
                     "Reconciled orphaned %s job %s as LOST",
                     prior_status, job.job_id,
                 )
-            elif (
-                job.status == JobStatus.COMPLETED
+                # LOST is persisted before the emit, so a crash loop observes
+                # each orphan exactly once.
+                _emit_job_completed(job)
+
+            # A COMPLETED result or a LOST notice still owed to the origin.
+            if (
+                job.status in (JobStatus.COMPLETED, JobStatus.LOST)
                 and job.origin
                 and not job.delivered
             ):
@@ -569,8 +587,8 @@ def reconcile_on_start(
                     self._persist(job)
                     counts["redelivered"] += 1
                     logger.info(
-                        "Re-delivered undelivered completed job %s to origin",
-                        job.job_id,
+                        "Re-delivered undelivered %s job %s to origin",
+                        job.status.value, job.job_id,
                     )
                 except Exception as e:  # noqa: BLE001 — redeliver is best-effort
                     logger.warning(

src/praisonai-agents/praisonaiagents/background/sqlite_store.py

@@ -141,21 +141,22 @@ def list_unreconciled(self) -> List[JobInfo]:
         """Return jobs needing reconciliation after a restart.
 
         Jobs left ``PENDING``/``RUNNING`` (orphaned by a crash) and jobs
-        ``COMPLETED`` with an ``origin`` that were never ``delivered``. Returned
-        oldest-first for deterministic replay.
+        ``COMPLETED`` or ``LOST`` with an ``origin`` that were never
+        ``delivered``. Returned oldest-first for deterministic replay.
         """
         with self._lock:
             rows = self._conn.execute(
                 """
                 SELECT * FROM background_jobs
                 WHERE status IN (?, ?)
-                   OR (status = ? AND delivered = 0 AND origin IS NOT NULL)
+                   OR (status IN (?, ?) AND delivered = 0 AND origin IS NOT NULL)
                 ORDER BY created_at ASC
                 """,
                 (
                     JobStatus.PENDING.value,
                     JobStatus.RUNNING.value,
                     JobStatus.COMPLETED.value,
+                    JobStatus.LOST.value,
                 ),
             ).fetchall()
         return [self._row_to_job(r) for r in rows]

src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py

@@ -0,0 +1,113 @@
+"""Restart reconciliation announces LOST jobs (Issue #6038)."""
+
+from praisonaiagents.background.job_manager import (
+    BackgroundJobManager, JobInfo, JobStatus,
+)
+from praisonaiagents.background.sqlite_store import SqliteBackgroundJobStore
+from praisonaiagents.hooks import add_hook
+from praisonaiagents.hooks.registry import remove_hook
+
+
+def _store(tmp_path):
+    return SqliteBackgroundJobStore(str(tmp_path / "jobs.db"))
+
+
+def _orphan(store, origin):
+    store.upsert(JobInfo(job_id="job-42", status=JobStatus.RUNNING,
+                         started_at=1.0, origin=origin))
+
+
+ORIGIN = {"deliver": "telegram:12345", "task": "audit all 40 repos"}
+
+
+def test_lost_job_fires_hook_once_and_is_redelivered(tmp_path):
+    store = _store(tmp_path)
+    _orphan(store, ORIGIN)
+    heard = []
+    hook_id = add_hook("job_completed", lambda inp: heard.append(inp) or None)
+    try:
+        _check_lost_flow(store, heard)
+    finally:
+        remove_hook(hook_id)
+
+
+def _check_lost_flow(store, heard):
+
+    # First pass (get_job_manager's lazy one): no redeliver.
+    BackgroundJobManager(store=store).reconcile_on_start()
+    info = store.get("job-42")
+    assert info.status == JobStatus.LOST
+    assert info.error and "restart" in info.error
+    assert not info.delivered
+    assert len(heard) == 1
+    assert heard[0].job_id == "job-42"
+    assert heard[0].status == "lost"
+
+    # Boot pass with redeliver: exactly one notice, then delivered.
+    sent = []
+    counts = BackgroundJobManager(store=store).reconcile_on_start(redeliver=sent.append)
+    assert [j.status for j in sent] == [JobStatus.LOST]
+    assert counts["redelivered"] == 1
+    assert store.get("job-42").delivered
+    assert len(heard) == 1  # hook not re-fired
+
+    sent.clear()
+    BackgroundJobManager(store=store).reconcile_on_start(redeliver=sent.append)
+    assert sent == []
+
+
+def test_failed_redeliver_retries_next_boot(tmp_path):
+    store = _store(tmp_path)
+    _orphan(store, ORIGIN)
+
+    def boom(job):
+        raise RuntimeError("transport down")
+
+    BackgroundJobManager(store=store).reconcile_on_start(redeliver=boom)
+    assert not store.get("job-42").delivered
+    sent = []
+    BackgroundJobManager(store=store).reconcile_on_start(redeliver=sent.append)
+    assert len(sent) == 1
+
+
+def test_lost_job_without_origin_is_not_redelivered(tmp_path):
+    store = _store(tmp_path)
+    _orphan(store, {})
+    sent = []
+    BackgroundJobManager(store=store).reconcile_on_start(redeliver=sent.append)
+    assert sent == []
+    assert store.list_unreconciled() == []
+
+
+def test_repeat_call_skips_jobs_live_in_this_process(tmp_path):
+    import threading
+
+    store = _store(tmp_path)
+    mgr = BackgroundJobManager(store=store)
+    gate = threading.Event()
+    job_id = mgr.start_job(lambda: gate.wait(5))
+    try:
+        mgr.reconcile_on_start(redeliver=lambda j: None)
+        assert mgr.get_job_info(job_id).status in (JobStatus.PENDING, JobStatus.RUNNING)
+    finally:
+        gate.set()
+        mgr.shutdown()
+
+
+def test_job_completing_during_scan_is_not_marked_lost(tmp_path):
+    store = _store(tmp_path)
+    mgr = BackgroundJobManager(store=store)
+    job_id = mgr.start_job(lambda: "done", origin=ORIGIN)
+    mgr.get_result(job_id, timeout=5)
+    # Stale RUNNING snapshot, as if listed before the worker finished.
+    stale = JobInfo(job_id=job_id, status=JobStatus.RUNNING, origin=ORIGIN)
+    mgr._store = type("S", (), {
+        "list_unreconciled": lambda self: [stale],
+        "upsert": store.upsert, "get": store.get,
+    })()
+    sent = []
+    counts = mgr.reconcile_on_start(redeliver=sent.append)
+    assert counts["lost"] == 0 and sent == []
+    assert mgr.get_job_info(job_id).status == JobStatus.COMPLETED
+    assert store.get(job_id).status == JobStatus.COMPLETED
+    mgr.shutdown()

src/praisonai-bot/praisonai_bot/gateway/api_endpoints.py

@@ -108,8 +108,9 @@ def __init__(self, gateway: Any) -> None:
         # Lightweight and dependency-free: a background run's status/output
         # survives the originating HTTP request so a client can retrieve it by
         # id later. Each entry also holds the run's InterruptController (for
-        # cancel) and the driving asyncio.Task. Not persisted across restarts;
-        # the durable-run ledger (Issue #4028) remains the cross-restart store.
+        # cancel) and the driving asyncio.Task. Not persisted across restarts:
+        # a restart drops in-flight stored responses (the gateway does not
+        # write them to the run ledger).
         self._responses: Dict[str, dict] = {}
         # Cap the in-process store so a sustained stream of background/stored
         # submissions cannot grow memory without bound in a long-lived gateway