From 1a967c5cc0c3e97a6ae93cc855c31143ac3e62a6 Mon Sep 17 00:00:00 2001 From: MervinPraison <454862+MervinPraison@users.noreply.github.com> Date: Mon, 26 Oct 2026 19:27:29 +0000 Subject: [PATCH 1/4] fix: notify origins and close runs for work interrupted by a gateway restart (fixes #6038) - Coalesced hooks: record idempotency keys only after the burst runs; release them on failure or shutdown so sender retries are accepted. - Background jobs: reconcile marks orphans LOST with an error, fires job_completed (status=lost) once, and offers LOST jobs with an origin to redeliver. The gateway runs this at boot and sends one durable 'interrupted' notice via the outbox. - Kanban: reclaiming a stale claim closes the run as crashed, records a failure (so crash loops trip the breaker) and adds an interrupted comment. - Correct comments claiming the gateway writes the run ledger. Co-Authored-By: Claude Opus 5.5 --- .../praisonaiagents/background/job_manager.py | 118 ++++++++++-------- .../background/sqlite_store.py | 7 +- .../test_reconcile_notifies_lost.py | 94 ++++++++++++++ .../praisonai_bot/gateway/api_endpoints.py | 5 +- .../praisonai_bot/gateway/hook_coalescer.py | 77 +++++++++--- .../praisonai_bot/gateway/server.py | 90 ++++++++++++- .../praisonai_bot/kanban/sqlite_store.py | 16 +++ .../gateway/test_hook_coalescer_restart.py | 66 ++++++++++ .../gateway/test_kanban_reclaim_closes_run.py | 47 +++++++ .../unit/gateway/test_ledger_recovery.py | 38 ++++++ 10 files changed, 478 insertions(+), 80 deletions(-) create mode 100644 src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py create mode 100644 src/praisonai-bot/tests/unit/gateway/test_hook_coalescer_restart.py create mode 100644 src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py diff --git a/src/praisonai-agents/praisonaiagents/background/job_manager.py b/src/praisonai-agents/praisonaiagents/background/job_manager.py index 9de8a4a004..1154c25323 100644 --- a/src/praisonai-agents/praisonaiagents/background/job_manager.py +++ b/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. + with self._lock: + live = self._jobs.get(job.job_id) + if live is not None and live.status in ( + JobStatus.PENDING, JobStatus.RUNNING, + ): + 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( diff --git a/src/praisonai-agents/praisonaiagents/background/sqlite_store.py b/src/praisonai-agents/praisonaiagents/background/sqlite_store.py index 2f626c10d9..94bf897751 100644 --- a/src/praisonai-agents/praisonaiagents/background/sqlite_store.py +++ b/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] diff --git a/src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py b/src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py new file mode 100644 index 0000000000..fa6979a53a --- /dev/null +++ b/src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py @@ -0,0 +1,94 @@ +"""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() diff --git a/src/praisonai-bot/praisonai_bot/gateway/api_endpoints.py b/src/praisonai-bot/praisonai_bot/gateway/api_endpoints.py index 67ae12eeb9..8b198bb276 100644 --- a/src/praisonai-bot/praisonai_bot/gateway/api_endpoints.py +++ b/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 diff --git a/src/praisonai-bot/praisonai_bot/gateway/hook_coalescer.py b/src/praisonai-bot/praisonai_bot/gateway/hook_coalescer.py index 5f03fdafc9..178efcbe2f 100644 --- a/src/praisonai-bot/praisonai_bot/gateway/hook_coalescer.py +++ b/src/praisonai-bot/praisonai_bot/gateway/hook_coalescer.py @@ -10,6 +10,11 @@ The number of pending groups is capped at ``max_groups`` so a burst across many distinct entities cannot grow the buffer without bound; ``submit`` then returns ``None`` and the gateway answers with a retryable 503. + +Each delivery's idempotency key stays an in-flight reservation until its group +settles (Issue #6038): ``settle(keys, ok)`` records them after a successful run +and releases them on failure or shutdown, so a sender's retry of a burst that +never ran is accepted instead of deduplicated. """ from __future__ import annotations @@ -17,16 +22,19 @@ import asyncio import logging import time -from typing import Any, Awaitable, Callable, Dict, Optional, Set +from typing import Any, Awaitable, Callable, Dict, List, Optional, Set logger = logging.getLogger(__name__) -# ``dispatch(hook, latest_payload, count)`` runs the coalesced turn. +# ``dispatch(hook, latest_payload, count)`` runs the coalesced turn and returns +# whether it succeeded. Dispatch = Callable[[Any, Dict[str, Any], int], Awaitable[Any]] +# ``settle(idempotency_keys, ok)`` records (ok) or releases the group's keys. +Settle = Callable[[List[str], bool], None] class _Group: - __slots__ = ("hook", "payload", "count", "first_at", "last_at", "task") + __slots__ = ("hook", "payload", "count", "first_at", "last_at", "task", "keys") def __init__(self, hook: Any, payload: Dict[str, Any], now: float) -> None: self.hook = hook @@ -35,6 +43,7 @@ def __init__(self, hook: Any, payload: Dict[str, Any], now: float) -> None: self.first_at = now self.last_at = now self.task: Optional[asyncio.Task] = None + self.keys: List[str] = [] class HookCoalescer: @@ -45,8 +54,10 @@ def __init__( dispatch: Dispatch, clock: Callable[[], float] = time.monotonic, max_groups: int = 10_000, + settle: Optional[Settle] = None, ) -> None: self._dispatch = dispatch + self._settle = settle self._clock = clock self._max_groups = max_groups self._groups: Dict[str, _Group] = {} @@ -57,7 +68,10 @@ def __init__( def __len__(self) -> int: return len(self._groups) - def submit(self, key: str, hook: Any, payload: Dict[str, Any]) -> Optional[int]: + def submit( + self, key: str, hook: Any, payload: Dict[str, Any], + idempotency_key: Optional[str] = None, + ) -> Optional[int]: """Buffer a delivery; returns how many deliveries the group now holds. Returns ``None`` (nothing buffered) when a new group would exceed @@ -78,28 +92,53 @@ def submit(self, key: str, hook: Any, payload: Dict[str, Any]) -> Optional[int]: group.payload = payload group.count += 1 group.last_at = now + if idempotency_key is not None: + group.keys.append(idempotency_key) return group.count async def _fire_when_due(self, key: str, group: _Group) -> None: - cfg = group.hook.coalesce - while True: - remaining = cfg.deadline(group.first_at, group.last_at) - self._clock() - if remaining <= 0: - break - await asyncio.sleep(remaining) - if self._groups.get(key) is group: - del self._groups[key] + ok = False try: - await self._dispatch(group.hook, group.payload, group.count) - except Exception as e: # noqa: BLE001 - never kill the loop - logger.error( - "Coalesced hook '%s' run failed: %s", - getattr(group.hook, "path", "?"), e, - ) + cfg = group.hook.coalesce + while True: + remaining = cfg.deadline(group.first_at, group.last_at) - self._clock() + if remaining <= 0: + break + await asyncio.sleep(remaining) + if self._groups.get(key) is group: + del self._groups[key] + try: + ok = bool(await self._dispatch(group.hook, group.payload, group.count)) + except Exception as e: # noqa: BLE001 - never kill the loop + logger.error( + "Coalesced hook '%s' run failed: %s", + getattr(group.hook, "path", "?"), e, + ) + finally: + # Also runs on cancellation (shutdown), releasing unrun deliveries. + if self._settle is not None and group.keys: + try: + self._settle(group.keys, ok) + except Exception as e: # noqa: BLE001 - best-effort + logger.error("Settling coalesced hook keys failed: %s", e) async def aclose(self) -> None: - """Cancel pending groups and in-progress runs (gateway shutdown).""" + """Cancel pending groups and in-progress runs (gateway shutdown). + + Cancelled groups release their idempotency keys so senders can retry. + """ pending = len(self._groups) + # A task cancelled before it first runs never reaches its ``finally``, + # so release pending groups' keys here (and empty them so a started + # task does not settle twice). + if self._settle is not None: + for group in self._groups.values(): + keys, group.keys = group.keys, [] + if keys: + try: + self._settle(keys, False) + except Exception as e: # noqa: BLE001 - best-effort + logger.error("Releasing coalesced hook keys failed: %s", e) self._groups.clear() tasks = list(self._tasks) if tasks: diff --git a/src/praisonai-bot/praisonai_bot/gateway/server.py b/src/praisonai-bot/praisonai_bot/gateway/server.py index d89f341d30..5dbc8d845b 100644 --- a/src/praisonai-bot/praisonai_bot/gateway/server.py +++ b/src/praisonai-bot/praisonai_bot/gateway/server.py @@ -2456,17 +2456,21 @@ async def hook_handler(request) -> JSONResponse: # Burst coalescing (#5867): buffer the delivery under its group key # and acknowledge it now; one run fires for the whole burst. The - # delivery is recorded as accepted so a sender retry stays deduped. + # key stays reserved (a retry while pending is deduped) and is only + # recorded once the burst runs; a failed or cancelled run releases + # it, and a crash leaves it to expire, so the sender can retry + # (#6038). group = hook.resolve_coalesce_key(payload) if group is not None: - pending = self._get_hook_coalescer().submit(group, hook, payload) + pending = self._get_hook_coalescer().submit( + group, hook, payload, idempotency_key=idem, + ) if pending is None: self._hook_release(idem) return JSONResponse( {"ok": False, "error": "too many pending coalesced groups"}, status_code=503, headers={"Retry-After": "30"}, ) - self._hook_record(idem) return JSONResponse( {"ok": True, "coalesced": True, "pending": pending}, status_code=202, @@ -2670,6 +2674,13 @@ async def _wrapped(request): except Exception: logger.exception("Failed to recover orphaned runs on boot") + # Issue #6038: tell the origin chat about background jobs a restart + # interrupted (or finished but never delivered). + try: + await self._reconcile_background_jobs() + except Exception: + logger.exception("Failed to reconcile background jobs on boot") + logger.info(f"Gateway started on ws://{self._host}:{self._port}") # Issue #3410: arm the opt-in event-loop liveness watchdog around the @@ -6083,15 +6094,24 @@ def _get_hook_coalescer(self) -> Any: if self._hook_coalescer is None: from .hook_coalescer import HookCoalescer - async def _dispatch(hook: Any, payload: Dict[str, Any], count: int) -> None: + async def _dispatch(hook: Any, payload: Dict[str, Any], count: int) -> bool: result = await self._run_hook(hook, payload, coalesced=count) if not result.get("ok", False): logger.error( "Coalesced hook '%s' run failed: %s", hook.path, result.get("error"), ) + return False + return True - self._hook_coalescer = HookCoalescer(_dispatch) + def _settle(keys: List[str], ok: bool) -> None: + for key in keys: + if ok: + self._hook_record(key) + else: + self._hook_release(key) + + self._hook_coalescer = HookCoalescer(_dispatch, settle=_settle) return self._hook_coalescer async def _run_hook( @@ -8037,7 +8057,10 @@ async def _recover_orphaned_runs(self) -> int: Best-effort and default-safe: a no-op when core lacks the ledger or the ledger DB does not exist yet, so gateways that never used the ledger are - unaffected. Idempotency is keyed on the ``run_id`` so a crash loop + unaffected. The gateway does not write the ledger itself; it is + populated by code using the public ``praisonaiagents.runs`` ledger at + its default path. Background jobs are recovered separately by + :meth:`_reconcile_background_jobs`. Idempotency is keyed on the ``run_id`` so a crash loop notifies at most once per interrupted run. """ ledger = self._open_run_ledger() @@ -8106,6 +8129,61 @@ async def _recover_orphaned_runs(self) -> int: except Exception: # pragma: no cover - defensive pass + async def _reconcile_background_jobs(self) -> int: + """Notify origins of background jobs a previous process left behind. + + Issue #6038: ``get_job_manager()`` reconciles orphaned jobs to ``LOST`` + lazily and with no ``redeliver``, so the chat that spawned the job was + never told. Run the reconcile pass at boot with a ``redeliver`` that + sends an "interrupted" notice (or the undelivered result) through the + durable outbox. A job is only marked delivered once that succeeds. + No-op when no job store exists yet. + """ + try: + from praisonaiagents.background.job_manager import ( + JobStatus, get_job_manager, + ) + from praisonaiagents.paths import get_runs_dir + except Exception as exc: # pragma: no cover - old/absent core + logger.debug("background jobs unavailable for boot recovery: %s", exc) + return 0 + if not (get_runs_dir() / "background_jobs.db").exists(): + return 0 + + loop = asyncio.get_running_loop() + + def _redeliver(job: Any) -> None: + origin = job.origin or {} + route = str(origin.get("deliver") or "") + if ":" not in route: + platform, chat_id = origin.get("platform"), origin.get("chat_id") + if not (platform and chat_id): + return # no route: mark delivered so it does not resurface + route = f"{platform}:{chat_id}" + if job.status == JobStatus.LOST: + task = str(origin.get("task") or "")[:80] + text = ( + f"Your background task '{task}' was interrupted by a " + "restart and did not finish. Please resend it to retry." + ) + else: + from praisonai_bot.bots.botos import BotOS + + text = BotOS._summarize_job_result(job.result) + if not text: + return + ok = asyncio.run_coroutine_threadsafe( + self._deliver_lost_run_notice(route, text, f"job:{job.job_id}"), + loop, + ).result(timeout=60) + if not ok: + raise RuntimeError(f"delivery to {route} failed") + + counts = await asyncio.to_thread( + lambda: get_job_manager().reconcile_on_start(redeliver=_redeliver) + ) + return counts.get("redelivered", 0) + async def _deliver_lost_run_notice( self, channel_target: str, text: str, run_id: str, ) -> bool: diff --git a/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py b/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py index 0c3210155d..4ff980b594 100644 --- a/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py +++ b/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py @@ -1069,6 +1069,10 @@ def reclaim_stale_claims(self, *, stale_timeout_seconds: int = 1800) -> List[str recently heartbeating are left untouched (the worker may extend its own lease via heartbeat()). + Each reclaim closes the interrupted run as ``crashed``, counts it via + :meth:`record_failure` (so a crash loop trips the circuit breaker) and + adds an ``interrupted`` comment. + Args: stale_timeout_seconds: Heartbeat staleness threshold. @@ -1157,6 +1161,7 @@ def _parse(value: Any) -> Optional[datetime]: 'reason': reason, } self._log_event(conn, row['id'], 'reclaimed', reclaim_details[row['id']]) + reclaim_details[row['id']]['run_id'] = row.get('current_run_id') # Reclamation is a real running -> ready transition, so it announces # KANBAN_TASK_MOVED like any other move (tagged with why it happened). @@ -1169,6 +1174,17 @@ def _parse(value: Any) -> Optional[datetime]: reclaimed=True, reason=detail.get('reason'), ) + # Issue #6038: the interrupted attempt is a crash, not a silent + # restart. Close its run, count it toward the circuit breaker and + # leave a comment the next attempt sees in get_retry_context. + error = ( + f"Worker {detail.get('worker_id')} was interrupted " + f"({detail.get('reason')}); task reclaimed and will restart from scratch" + ) + if detail.get('run_id') is not None: + self.close_run(detail['run_id'], RunOutcome.CRASHED, error=error) + self.record_failure(task_id, error=error) + self.add_comment(task_id, 'system', f"interrupted: {error}") return reclaimed # ------------------------------------------------------------------ diff --git a/src/praisonai-bot/tests/unit/gateway/test_hook_coalescer_restart.py b/src/praisonai-bot/tests/unit/gateway/test_hook_coalescer_restart.py new file mode 100644 index 0000000000..668a795e14 --- /dev/null +++ b/src/praisonai-bot/tests/unit/gateway/test_hook_coalescer_restart.py @@ -0,0 +1,66 @@ +"""Coalesced hook keys are only recorded once the burst runs (Issue #6038).""" + +import asyncio + +from praisonaiagents.gateway import CoalesceConfig, HookConfig + +from praisonai_bot.gateway.hook_coalescer import HookCoalescer + + +def _hook(): + return HookConfig( + path="github", + message="push", + coalesce=CoalesceConfig(group_key="{pr}", window_seconds=0.05), + ) + + +async def _run(dispatch_ok, close_early=False): + settled = [] + + async def dispatch(hook, payload, count): + return dispatch_ok + + c = HookCoalescer(dispatch, settle=lambda keys, ok: settled.append((keys, ok))) + hook = _hook() + c.submit("k", hook, {"pr": 1}, idempotency_key="a") + c.submit("k", hook, {"pr": 1}, idempotency_key="b") + if close_early: + await c.aclose() + else: + await asyncio.sleep(0.3) + return settled + + +async def test_success_records_every_key_in_the_burst(): + assert await _run(True) == [(["a", "b"], True)] + + +async def test_failed_run_releases_keys(): + assert await _run(False) == [(["a", "b"], False)] + + +async def test_shutdown_before_run_releases_keys(): + assert await _run(True, close_early=True) == [(["a", "b"], False)] + + +async def test_retry_accepted_after_unrun_burst(): + """End to end with the durable store: a cancelled burst does not dedupe.""" + from praisonai_bot.bots._idempotency import SqliteIdempotencyStore + import tempfile, os + + store = SqliteIdempotencyStore(os.path.join(tempfile.mkdtemp(), "i.sqlite")) + + def settle(keys, ok): + for k in keys: + store.record(k) if ok else store.release(k) + + async def dispatch(hook, payload, count): + return True + + c = HookCoalescer(dispatch, settle=settle) + assert store.reserve("delivery-1") + c.submit("k", _hook(), {"pr": 1}, idempotency_key="delivery-1") + assert not store.reserve("delivery-1") # still pending: deduped + await c.aclose() + assert store.reserve("delivery-1") # sender retry accepted diff --git a/src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py b/src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py new file mode 100644 index 0000000000..5c66a4ac91 --- /dev/null +++ b/src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py @@ -0,0 +1,47 @@ +"""Reclaiming a stale kanban claim closes the crashed run (Issue #6038).""" + +import pytest + +from praisonai_bot.kanban.sqlite_store import SQLiteKanbanStore + +DEAD_PID = 2 ** 22 + 12345 # above pid_max on Linux, so never alive + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setenv("PRAISONAI_KANBAN_DB", str(tmp_path / "kanban.db")) + return SQLiteKanbanStore() + + +def _crash(store, task_id="k1"): + """Claim + start a run whose worker died and whose lease has expired.""" + assert store.claim_task(task_id, "gateway_before", ttl_seconds=-1, + worker_pid=DEAD_PID) + return store.start_run(task_id, profile="gateway_before") + + +def test_reclaim_closes_run_counts_failure_and_comments(store): + store.create_task({"id": "k1", "title": "migrate db", "status": "ready", + "workspace_kind": "default"}) + run_id = _crash(store) + + assert store.reclaim_stale_claims() == ["k1"] + + run = next(r for r in store.get_runs("k1") if r.id == run_id) + assert run.outcome == "crashed" + assert run.ended_at + assert "interrupted" in run.error + task = store.get_task("k1") + assert task.status == "ready" + assert task.consecutive_failures == 1 + assert any(c.text.startswith("interrupted:") for c in store.get_comments("k1")) + assert store.get_retry_context("k1")[0]["outcome"] == "crashed" + + +def test_crash_loop_trips_circuit_breaker(store): + store.create_task({"id": "k2", "title": "loop", "status": "ready", + "workspace_kind": "default", "max_retries": 2}) + for _ in range(2): + _crash(store, "k2") + store.reclaim_stale_claims() + assert store.get_task("k2").status == "blocked" diff --git a/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py b/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py index e2ef0d05c9..a4c14bbbde 100644 --- a/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py +++ b/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py @@ -279,3 +279,41 @@ def test_open_run_ledger_returns_none_when_db_missing(tmp_path, monkeypatch): gateway = _make_gateway() assert gateway._open_run_ledger() is None assert not (tmp_path / "ledger.db").exists() + + +def test_reconcile_background_jobs_notifies_lost_job_once(tmp_path, monkeypatch): + """Issue #6038: a job RUNNING at restart gets one 'interrupted' notice.""" + import praisonaiagents.background.job_manager as jm + from praisonaiagents.background.job_manager import JobInfo, JobStatus + from praisonaiagents.background.sqlite_store import SqliteBackgroundJobStore + + monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path)) + pre = SqliteBackgroundJobStore() + pre.upsert(JobInfo( + job_id="job-42", status=JobStatus.RUNNING, started_at=1.0, + origin={"deliver": "origin", "platform": "telegram", + "chat_id": "12345", "task": "audit all 40 repos"}, + )) + pre.close() + monkeypatch.setattr(jm, "_global_job_manager", None) + + gateway = _make_gateway() + sent = [] + + async def _notice(target, text, run_id): + sent.append((target, text, run_id)) + return True + + monkeypatch.setattr(gateway, "_deliver_lost_run_notice", _notice) + try: + assert asyncio.run(gateway._reconcile_background_jobs()) == 1 + assert asyncio.run(gateway._reconcile_background_jobs()) == 0 + finally: + mgr, jm._global_job_manager = jm._global_job_manager, None + if mgr is not None: + mgr.shutdown(wait=False) + assert len(sent) == 1 + target, text, run_id = sent[0] + assert target == "telegram:12345" + assert "audit all 40 repos" in text and "interrupted" in text + assert run_id == "job:job-42" From 8a65b5a599bbf53ebd19249247b05a2495358194 Mon Sep 17 00:00:00 2001 From: MervinPraison <454862+MervinPraison@users.noreply.github.com> Date: Mon, 26 Oct 2026 20:35:16 +0000 Subject: [PATCH 2/4] fix: route job recovery notices by delivery rules and reclaim kanban attempts atomically - _reconcile_background_jobs resolves the saved deliver target the same way live delivery does: origin keeps thread_id, all fans out to home channels, platform names and aliases resolve through the router. A target that cannot be resolved yet raises, so the job is retried on the next boot instead of being marked delivered. - reclaim_stale_claims closes the crashed run, counts the failure and adds the interrupted comment in the reclaim transaction. Hooks fire after commit, so another dispatcher cannot claim the task before the old attempt is counted. Co-Authored-By: Claude Opus 5.5 --- .../praisonai_bot/gateway/server.py | 62 ++++-- .../praisonai_bot/kanban/sqlite_store.py | 177 +++++++++++------- .../gateway/test_kanban_reclaim_closes_run.py | 21 +++ .../unit/gateway/test_ledger_recovery.py | 28 ++- 4 files changed, 211 insertions(+), 77 deletions(-) diff --git a/src/praisonai-bot/praisonai_bot/gateway/server.py b/src/praisonai-bot/praisonai_bot/gateway/server.py index 5dbc8d845b..d944c86cfb 100644 --- a/src/praisonai-bot/praisonai_bot/gateway/server.py +++ b/src/praisonai-bot/praisonai_bot/gateway/server.py @@ -8154,12 +8154,9 @@ async def _reconcile_background_jobs(self) -> int: def _redeliver(job: Any) -> None: origin = job.origin or {} - route = str(origin.get("deliver") or "") - if ":" not in route: - platform, chat_id = origin.get("platform"), origin.get("chat_id") - if not (platform and chat_id): - return # no route: mark delivered so it does not resurface - route = f"{platform}:{chat_id}" + routes = self._resolve_job_routes(origin) + if not routes: + return # nothing deliverable: mark delivered so it does not resurface if job.status == JobStatus.LOST: task = str(origin.get("task") or "")[:80] text = ( @@ -8172,18 +8169,59 @@ def _redeliver(job: Any) -> None: text = BotOS._summarize_job_result(job.result) if not text: return - ok = asyncio.run_coroutine_threadsafe( - self._deliver_lost_run_notice(route, text, f"job:{job.job_id}"), - loop, - ).result(timeout=60) - if not ok: - raise RuntimeError(f"delivery to {route} failed") + delivered = False + for route in routes: + ok = asyncio.run_coroutine_threadsafe( + self._deliver_lost_run_notice( + route, text, f"job:{job.job_id}:{route}", + ), + loop, + ).result(timeout=60) + delivered = delivered or bool(ok) + if not delivered: + raise RuntimeError(f"delivery to {routes} failed") counts = await asyncio.to_thread( lambda: get_job_manager().reconcile_on_start(redeliver=_redeliver) ) return counts.get("redelivered", 0) + def _resolve_job_routes(self, origin: Dict[str, Any]) -> List[str]: + """Resolve a background job's saved ``deliver`` target to concrete routes. + + Mirrors the live delivery rules (``BotOS.on_background_job_complete``): + ``origin`` keeps the thread, ``all`` fans out to every home channel, and + platform names / aliases resolve through the router. Returns ``[]`` only + when nothing can ever be delivered (no target requested, or ``origin`` + without chat fields); a target that does not resolve *now* raises so the + job is retried on the next boot instead of being marked delivered. + """ + deliver = str(origin.get("deliver") or "") + if not deliver: + return [] + if deliver == "origin": + platform, chat_id = origin.get("platform"), origin.get("chat_id") + if not (platform and chat_id): + logger.warning("Background job origin has no chat; dropping notice") + return [] + thread_id = origin.get("thread_id") + return [f"{platform}:{chat_id}:{thread_id}" if thread_id else f"{platform}:{chat_id}"] + router = self.delivery_router + if deliver == "all": + directory = getattr(router, "directory", None) + targets = directory.describe_targets() if directory is not None else [] + routes = [ + f"{t['platform']}:{t['channel_id']}" + for t in targets if t.get("kind") == "home" + ] + if not routes: + raise RuntimeError("deliver='all' has no home channels") + return routes + if router is None: + return [deliver] + platform, chat_id, thread_id = router.resolve(deliver) + return [f"{platform}:{chat_id}:{thread_id}" if thread_id else f"{platform}:{chat_id}"] + async def _deliver_lost_run_notice( self, channel_target: str, text: str, run_id: str, ) -> bool: diff --git a/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py b/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py index 4ff980b594..540c2706e7 100644 --- a/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py +++ b/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py @@ -5,7 +5,7 @@ import uuid from contextlib import contextmanager from datetime import datetime, timezone -from typing import Optional, Dict, Any, List +from typing import Optional, Dict, Any, List, Tuple from pathlib import Path from .models import ( @@ -1069,9 +1069,9 @@ def reclaim_stale_claims(self, *, stale_timeout_seconds: int = 1800) -> List[str recently heartbeating are left untouched (the worker may extend its own lease via heartbeat()). - Each reclaim closes the interrupted run as ``crashed``, counts it via - :meth:`record_failure` (so a crash loop trips the circuit breaker) and - adds an ``interrupted`` comment. + Each reclaim closes the interrupted run as ``crashed``, counts it as a + failure (so a crash loop trips the circuit breaker) and adds an + ``interrupted`` comment, all in the reclaim's own transaction. Args: stale_timeout_seconds: Heartbeat staleness threshold. @@ -1161,7 +1161,25 @@ def _parse(value: Any) -> Optional[datetime]: 'reason': reason, } self._log_event(conn, row['id'], 'reclaimed', reclaim_details[row['id']]) - reclaim_details[row['id']]['run_id'] = row.get('current_run_id') + # Issue #6038: the interrupted attempt is a crash, not a + # silent restart. Close its run, count it toward the + # circuit breaker and leave a comment the next attempt sees + # -- in this same transaction, so no other dispatcher can + # claim the task before the old attempt is accounted for. + error = ( + f"Worker {row['claim_lock']} was interrupted ({reason}); " + "task reclaimed and will restart from scratch" + ) + self._close_crashed_run(conn, row['id'], row.get('current_run_id'), error, now) + reclaim_details[row['id']]['failure'] = ( + error, self._count_failure(conn, row['id'], error, now.isoformat()) + ) + conn.execute( + "INSERT INTO task_comments (id, task_id, author, text, created_at) " + "VALUES (?, ?, ?, ?, ?)", + (f"comment_{uuid.uuid4().hex[:12]}", row['id'], 'system', + f"interrupted: {error}", now.replace(tzinfo=None).isoformat()), + ) # Reclamation is a real running -> ready transition, so it announces # KANBAN_TASK_MOVED like any other move (tagged with why it happened). @@ -1174,19 +1192,32 @@ def _parse(value: Any) -> Optional[datetime]: reclaimed=True, reason=detail.get('reason'), ) - # Issue #6038: the interrupted attempt is a crash, not a silent - # restart. Close its run, count it toward the circuit breaker and - # leave a comment the next attempt sees in get_retry_context. - error = ( - f"Worker {detail.get('worker_id')} was interrupted " - f"({detail.get('reason')}); task reclaimed and will restart from scratch" - ) - if detail.get('run_id') is not None: - self.close_run(detail['run_id'], RunOutcome.CRASHED, error=error) - self.record_failure(task_id, error=error) - self.add_comment(task_id, 'system', f"interrupted: {error}") + error, counted = detail['failure'] + if counted is not None: + self._emit_failure(task_id, error, *counted) return reclaimed + def _close_crashed_run( + self, conn: sqlite3.Connection, task_id: str, run_id: Optional[int], + error: str, now: datetime, + ) -> None: + """Close an interrupted attempt as ``crashed`` inside ``conn``'s transaction.""" + if run_id is None: + return + result = conn.execute( + "UPDATE task_runs SET outcome = ?, error = ?, ended_at = ? " + "WHERE id = ? AND (ended_at IS NULL OR ended_at = '')", + (RunOutcome.CRASHED.value, error, now.isoformat(), run_id), + ) + conn.execute( + "UPDATE tasks SET current_run_id = NULL WHERE id = ? AND current_run_id = ?", + (task_id, run_id), + ) + if result.rowcount > 0: + self._log_event(conn, task_id, 'run_closed', { + 'run_id': run_id, 'outcome': RunOutcome.CRASHED.value, + }) + # ------------------------------------------------------------------ # Task runs (attempt history) + per-task retry / circuit-breaker # ------------------------------------------------------------------ @@ -1323,55 +1354,74 @@ def record_failure(self, task_id: str, *, error: Optional[str] = None) -> bool: from datetime import timezone now = datetime.now(timezone.utc).isoformat() with self._get_connection() as conn: - cursor = conn.execute( - "SELECT consecutive_failures, max_retries, status FROM tasks WHERE id = ?", - (task_id,) - ) - row = cursor.fetchone() - if not row: - raise ValueError(f"Task {task_id} not found") - # A terminal task (already completed/archived by another caller, - # e.g. kanban_complete) must not be re-failed: a stale process-exit - # handler could otherwise increment the counter and re-block a task - # that already finished. Treat as a no-op (not circuit-broken). - if (row['status'] or '') in ('done', 'archived'): + counted = self._count_failure(conn, task_id, error, now) + if counted is None: return False - failures = (row['consecutive_failures'] or 0) + 1 - # Normalize the persisted limit too: a task created before create-path - # validation (or via migration) may carry max_retries=0/negative, - # which would otherwise auto-block on the first failure. Fall back to - # the board default for any non-positive / non-integer stored value. - stored_limit = row['max_retries'] - try: - stored_limit = int(stored_limit) if stored_limit is not None else None - except (TypeError, ValueError): - stored_limit = None - limit = stored_limit if (stored_limit is not None and stored_limit >= 1) else DEFAULT_MAX_RETRIES - blocked = failures >= limit - if blocked: - conn.execute(""" - UPDATE tasks - SET consecutive_failures = ?, status = 'blocked', - claim_lock = NULL, updated_at = ?, version = version + 1 - WHERE id = ? - """, (failures, now, task_id)) - self._log_event(conn, task_id, 'circuit_broken', { - 'consecutive_failures': failures, - 'max_retries': limit, - 'error': (error or '')[:500], - }) - else: - conn.execute( - "UPDATE tasks SET consecutive_failures = ?, updated_at = ? WHERE id = ?", - (failures, now, task_id) - ) - self._log_event(conn, task_id, 'failure_recorded', { - 'consecutive_failures': failures, - 'max_retries': limit, - }) - old_status = row['status'] or '' + self._emit_failure(task_id, error, *counted) + return counted[3] - # Committed. Every recorded attempt failure announces KANBAN_TASK_FAILED; + def _count_failure( + self, conn: sqlite3.Connection, task_id: str, error: Optional[str], now: str, + ) -> Optional[Tuple[str, int, int, bool]]: + """Count one failed attempt inside ``conn``'s transaction. + + Returns ``(old_status, failures, limit, blocked)`` for + :meth:`_emit_failure`, or ``None`` when the task is already terminal. + """ + cursor = conn.execute( + "SELECT consecutive_failures, max_retries, status FROM tasks WHERE id = ?", + (task_id,) + ) + row = cursor.fetchone() + if not row: + raise ValueError(f"Task {task_id} not found") + # A terminal task (already completed/archived by another caller, + # e.g. kanban_complete) must not be re-failed: a stale process-exit + # handler could otherwise increment the counter and re-block a task + # that already finished. Treat as a no-op (not circuit-broken). + if (row['status'] or '') in ('done', 'archived'): + return None + failures = (row['consecutive_failures'] or 0) + 1 + # Normalize the persisted limit too: a task created before create-path + # validation (or via migration) may carry max_retries=0/negative, + # which would otherwise auto-block on the first failure. Fall back to + # the board default for any non-positive / non-integer stored value. + stored_limit = row['max_retries'] + try: + stored_limit = int(stored_limit) if stored_limit is not None else None + except (TypeError, ValueError): + stored_limit = None + limit = stored_limit if (stored_limit is not None and stored_limit >= 1) else DEFAULT_MAX_RETRIES + blocked = failures >= limit + if blocked: + conn.execute(""" + UPDATE tasks + SET consecutive_failures = ?, status = 'blocked', + claim_lock = NULL, updated_at = ?, version = version + 1 + WHERE id = ? + """, (failures, now, task_id)) + self._log_event(conn, task_id, 'circuit_broken', { + 'consecutive_failures': failures, + 'max_retries': limit, + 'error': (error or '')[:500], + }) + else: + conn.execute( + "UPDATE tasks SET consecutive_failures = ?, updated_at = ? WHERE id = ?", + (failures, now, task_id) + ) + self._log_event(conn, task_id, 'failure_recorded', { + 'consecutive_failures': failures, + 'max_retries': limit, + }) + return (row['status'] or '', failures, limit, blocked) + + def _emit_failure( + self, task_id: str, error: Optional[str], + old_status: str, failures: int, limit: int, blocked: bool, + ) -> None: + """Announce a committed failure counted by :meth:`_count_failure`.""" + # Every recorded attempt failure announces KANBAN_TASK_FAILED; # tripping the circuit breaker additionally moves the task to 'blocked' # in-place (it does not route through move_task), so announce that too. self._emit_hook( @@ -1393,7 +1443,6 @@ def record_failure(self, task_id: str, *, error: Optional[str] = None) -> bool: circuit_broken=True, error=(error or '')[:500], ) - return blocked def get_retry_context(self, task_id: str) -> List[Dict[str, Any]]: """Prior attempts' outcomes/summaries/errors for a retrying worker. diff --git a/src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py b/src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py index 5c66a4ac91..eb3a527c37 100644 --- a/src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py +++ b/src/praisonai-bot/tests/unit/gateway/test_kanban_reclaim_closes_run.py @@ -45,3 +45,24 @@ def test_crash_loop_trips_circuit_breaker(store): _crash(store, "k2") store.reclaim_stale_claims() assert store.get_task("k2").status == "blocked" + + +def test_reclaim_bookkeeping_commits_with_the_reclaim(store, monkeypatch): + """No window where the task is 'ready' but the old attempt is unaccounted.""" + store.create_task({"id": "k3", "title": "atomic", "status": "ready", + "workspace_kind": "default", "max_retries": 1}) + run_id = _crash(store, "k3") + seen = [] + emit = store._emit_hook + + def _spy(event, task_id, **payload): + if event == "kanban_task_moved": + run = next(r for r in store.get_runs(task_id) if r.id == run_id) + seen.append((run.outcome, store.get_task(task_id).status)) + return emit(event, task_id, **payload) + + monkeypatch.setattr(store, "_emit_hook", _spy) + assert store.reclaim_stale_claims() == ["k3"] + # At the first post-commit announcement the run is already closed and the + # circuit breaker already tripped, so no dispatcher could claim it as ready. + assert seen == [("crashed", "blocked")] diff --git a/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py b/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py index a4c14bbbde..32eaa65933 100644 --- a/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py +++ b/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py @@ -316,4 +316,30 @@ async def _notice(target, text, run_id): target, text, run_id = sent[0] assert target == "telegram:12345" assert "audit all 40 repos" in text and "interrupted" in text - assert run_id == "job:job-42" + assert run_id == "job:job-42:telegram:12345" + + +def test_resolve_job_routes_keeps_thread_and_resolves_targets(): + """Recovery uses the live delivery rules: thread kept, aliases resolved.""" + gateway = _make_gateway() + origin = {"platform": "slack", "chat_id": "C1", "thread_id": "T9"} + assert gateway._resolve_job_routes({**origin, "deliver": "origin"}) == ["slack:C1:T9"] + assert gateway._resolve_job_routes({"deliver": ""}) == [] + assert gateway._resolve_job_routes({"deliver": "origin"}) == [] + + class _Router: + def resolve(self, target): + if target == "ops": + return ("slack", "C2", None) + raise ValueError(target) + + gateway._delivery_router = _Router() + try: + assert gateway._resolve_job_routes({**origin, "deliver": "ops"}) == ["slack:C2"] + # Not resolvable now: raise so the job is retried, not marked delivered. + with pytest.raises(ValueError): + gateway._resolve_job_routes({**origin, "deliver": "nowhere"}) + with pytest.raises(RuntimeError): + gateway._resolve_job_routes({**origin, "deliver": "all"}) + finally: + gateway._delivery_router = None From de8c65067d755a4f434297c7090cb9fbec502440 Mon Sep 17 00:00:00 2001 From: MervinPraison <454862+MervinPraison@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:40:19 +0000 Subject: [PATCH 3/4] fix: drop duplicate reclaim run-close and failure count after rebase onto main Main's #6048 closed the reclaimed run inline and had the dispatcher count the failure; this PR does both atomically in the store, so keep only the store's _close_crashed_run/_count_failure path. Co-Authored-By: Claude Opus 5.5 --- .../praisonai_bot/gateway/kanban_dispatcher.py | 7 ++----- .../praisonai_bot/kanban/sqlite_store.py | 14 -------------- 2 files changed, 2 insertions(+), 19 deletions(-) diff --git a/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py b/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py index 81bb6e0d15..e50cb76ed5 100644 --- a/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py +++ b/src/praisonai-bot/praisonai_bot/gateway/kanban_dispatcher.py @@ -514,11 +514,8 @@ def _reclaim_stale_claims(self, store: Any): for task_id in reclaimed or []: logger.warning(f"Reclaimed stale kanban task {task_id} back to 'ready'") - # The store closed the abandoned run as 'crashed' atomically with - # the reclaim; count the attempt so the circuit breaker sees it. - self._record_failure_safe( - store, task_id, error='Worker died or went stale; claim reclaimed' - ) + # The store closed the abandoned run as 'crashed' and counted the + # failure atomically with the reclaim, so nothing more to do here. # The running -> ready transition is announced as # KANBAN_TASK_MOVED (reclaimed=True) by the store itself; # 'KANBAN_TASK_RECLAIMED' was never a HookEvent member and so diff --git a/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py b/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py index 540c2706e7..f2f5c3e6cc 100644 --- a/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py +++ b/src/praisonai-bot/praisonai_bot/kanban/sqlite_store.py @@ -1141,20 +1141,6 @@ def _parse(value: Any) -> Optional[datetime]: if result.rowcount > 0: reclaimed.append(row['id']) reason = 'dead' if not worker_alive else 'stale_heartbeat' - # Close the abandoned attempt in the same transaction, so a - # dispatcher that claims the task next can never have its - # own fresh run mistaken for this one. - if row.get('current_run_id') is not None and conn.execute( - "UPDATE task_runs SET outcome = ?, error = ?, ended_at = ? " - "WHERE id = ? AND ended_at IS NULL", - (RunOutcome.CRASHED.value, - f'Worker {reason}; claim reclaimed', - now.isoformat(), row['current_run_id']), - ).rowcount: - self._log_event(conn, row['id'], 'run_closed', { - 'run_id': row['current_run_id'], - 'outcome': RunOutcome.CRASHED.value, - }) reclaim_details[row['id']] = { 'worker_id': row['claim_lock'], 'worker_pid': pid, From be05884c76a6fb69b9d34972349ba16eb6e78806 Mon Sep 17 00:00:00 2001 From: MervinPraison <454862+MervinPraison@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:50:40 +0000 Subject: [PATCH 4/4] fix: never reconcile jobs owned by this process and require every recovery route to succeed Co-Authored-By: Claude Opus 5.5 --- .../praisonaiagents/background/job_manager.py | 10 ++--- .../test_reconcile_notifies_lost.py | 19 +++++++++ .../praisonai_bot/gateway/server.py | 11 +++-- .../unit/gateway/test_ledger_recovery.py | 42 +++++++++++++++++++ 4 files changed, 73 insertions(+), 9 deletions(-) diff --git a/src/praisonai-agents/praisonaiagents/background/job_manager.py b/src/praisonai-agents/praisonaiagents/background/job_manager.py index 1154c25323..01e9055ee3 100644 --- a/src/praisonai-agents/praisonaiagents/background/job_manager.py +++ b/src/praisonai-agents/praisonaiagents/background/job_manager.py @@ -536,12 +536,12 @@ def reconcile_on_start( 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. + # 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: - live = self._jobs.get(job.job_id) - if live is not None and live.status in ( - JobStatus.PENDING, JobStatus.RUNNING, - ): + 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 diff --git a/src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py b/src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py index fa6979a53a..6f3f0785a2 100644 --- a/src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py +++ b/src/praisonai-agents/tests/unit/background/test_reconcile_notifies_lost.py @@ -92,3 +92,22 @@ def test_repeat_call_skips_jobs_live_in_this_process(tmp_path): 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() diff --git a/src/praisonai-bot/praisonai_bot/gateway/server.py b/src/praisonai-bot/praisonai_bot/gateway/server.py index d944c86cfb..3662714446 100644 --- a/src/praisonai-bot/praisonai_bot/gateway/server.py +++ b/src/praisonai-bot/praisonai_bot/gateway/server.py @@ -8169,7 +8169,9 @@ def _redeliver(job: Any) -> None: text = BotOS._summarize_job_result(job.result) if not text: return - delivered = False + # Every route must succeed before the job is marked delivered; the + # per-route outbox key skips routes that already went out on retry. + failed = [] for route in routes: ok = asyncio.run_coroutine_threadsafe( self._deliver_lost_run_notice( @@ -8177,9 +8179,10 @@ def _redeliver(job: Any) -> None: ), loop, ).result(timeout=60) - delivered = delivered or bool(ok) - if not delivered: - raise RuntimeError(f"delivery to {routes} failed") + if not ok: + failed.append(route) + if failed: + raise RuntimeError(f"delivery to {failed} failed") counts = await asyncio.to_thread( lambda: get_job_manager().reconcile_on_start(redeliver=_redeliver) diff --git a/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py b/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py index 32eaa65933..f888ea4df3 100644 --- a/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py +++ b/src/praisonai-bot/tests/unit/gateway/test_ledger_recovery.py @@ -343,3 +343,45 @@ def resolve(self, target): gateway._resolve_job_routes({**origin, "deliver": "all"}) finally: gateway._delivery_router = None + + +def test_reconcile_background_jobs_retries_until_every_route_succeeds(tmp_path, monkeypatch): + """A partial fan-out failure leaves the job undelivered for the next boot.""" + import praisonaiagents.background.job_manager as jm + from praisonaiagents.background.job_manager import JobInfo, JobStatus + from praisonaiagents.background.sqlite_store import SqliteBackgroundJobStore + + monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path)) + pre = SqliteBackgroundJobStore() + pre.upsert(JobInfo( + job_id="job-7", status=JobStatus.RUNNING, started_at=1.0, + origin={"deliver": "all", "task": "nightly report"}, + )) + pre.close() + monkeypatch.setattr(jm, "_global_job_manager", None) + + gateway = _make_gateway() + monkeypatch.setattr( + gateway, "_resolve_job_routes", + lambda origin: ["slack:C1", "telegram:42"], + ) + down = {"telegram:42"} + sent = [] + + async def _notice(target, text, run_id): + if target in down: + return False + sent.append(target) + return True + + monkeypatch.setattr(gateway, "_deliver_lost_run_notice", _notice) + try: + assert asyncio.run(gateway._reconcile_background_jobs()) == 0 + assert not jm._global_job_manager.get_job_info("job-7").delivered + down.clear() + assert asyncio.run(gateway._reconcile_background_jobs()) == 1 + finally: + mgr, jm._global_job_manager = jm._global_job_manager, None + if mgr is not None: + mgr.shutdown(wait=False) + assert "telegram:42" in sent