From f8a67d50b61520a4471b5c23c785f06c51a243dd Mon Sep 17 00:00:00 2001 From: olegbrok Date: Fri, 31 Jul 2026 07:32:38 -0700 Subject: [PATCH] fix(scheduler): recover undelivered wakes --- src/pinky_daemon/agent_registry.py | 135 ++++++++ src/pinky_daemon/api.py | 67 ++++ src/pinky_daemon/scheduler.py | 355 ++++++++++++++++++++- src/pinky_daemon/tmux_session.py | 12 +- tests/test_api.py | 42 +++ tests/test_scheduler.py | 479 +++++++++++++++++++++++++++++ tests/test_tmux_session.py | 20 ++ 7 files changed, 1093 insertions(+), 17 deletions(-) diff --git a/src/pinky_daemon/agent_registry.py b/src/pinky_daemon/agent_registry.py index a31d7fa5..23c61009 100644 --- a/src/pinky_daemon/agent_registry.py +++ b/src/pinky_daemon/agent_registry.py @@ -652,6 +652,24 @@ def to_dict(self) -> dict: } +@dataclass +class PendingScheduleWake: + """A fired scheduler prompt awaiting confirmed transport acceptance.""" + + id: int = 0 + schedule_id: int = 0 + agent_name: str = "" + schedule_name: str = "" + prompt: str = "" + fired_at: float = 0.0 + created_at: float = 0.0 + + @property + def name(self) -> str: + """Match the ``AgentSchedule`` interface used by receipt waiting.""" + return self.schedule_name + + class ScheduleNameConflictError(ValueError): """An enabled schedule already uses the requested agent/name pair.""" @@ -1249,6 +1267,18 @@ def _init_tables(self) -> None: FOREIGN KEY (agent_name) REFERENCES agents(name) ON DELETE CASCADE ); + CREATE TABLE IF NOT EXISTS pending_schedule_wakes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + schedule_id INTEGER NOT NULL, + agent_name TEXT NOT NULL, + schedule_name TEXT NOT NULL DEFAULT '', + prompt TEXT NOT NULL DEFAULT '', + fired_at REAL NOT NULL, + created_at REAL NOT NULL, + FOREIGN KEY (agent_name) REFERENCES agents(name) ON DELETE CASCADE, + UNIQUE(schedule_id, fired_at) + ); + CREATE TABLE IF NOT EXISTS agent_heartbeats ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_name TEXT NOT NULL, @@ -1392,6 +1422,8 @@ def _init_tables(self) -> None: ON agent_heartbeats(agent_name, timestamp DESC); CREATE INDEX IF NOT EXISTS idx_schedules_agent ON agent_schedules(agent_name); + CREATE INDEX IF NOT EXISTS idx_pending_schedule_wakes_agent + ON pending_schedule_wakes(agent_name, fired_at, id); CREATE INDEX IF NOT EXISTS idx_pending_messages_agent_chat ON pending_messages(agent_name, chat_id, delivered); CREATE INDEX IF NOT EXISTS idx_approval_requests_retry @@ -3066,6 +3098,17 @@ def get_all_schedules(self, *, enabled_only: bool = True) -> list[AgentSchedule] return [self._row_to_schedule(r) for r in rows ] + def get_schedule(self, schedule_id: int) -> AgentSchedule | None: + """Return one schedule regardless of enabled state.""" + row = self._db.execute( + """SELECT id, agent_name, name, cron, prompt, timezone, enabled, + last_run, last_delivered, created_at, direct_send, + target_channel, one_shot + FROM agent_schedules WHERE id=?""", + (schedule_id,), + ).fetchone() + return self._row_to_schedule(row) if row else None + def update_schedule( self, schedule_id: int, @@ -3181,6 +3224,98 @@ def update_schedule_last_delivered( ) self._db.commit() + def persist_schedule_wake( + self, + schedule_id: int, + *, + agent_name: str, + schedule_name: str, + prompt: str, + fired_at: float, + ) -> tuple[PendingScheduleWake, bool]: + """Durably retain one fired wake until exact delivery is confirmed. + + ``(schedule_id, fired_at)`` identifies one cron fire and makes repeated + accounting of the same failed receipt idempotent. Callers must carry + that immutable timestamp with the fired cohort; rereading mutable + ``last_run`` here can collapse overlapping fires into one row. + """ + if fired_at <= 0: + raise ValueError("fired_at must be a positive exact-fire timestamp") + created_at = time.time() + with self._rmw_lock: + cursor = self._db.execute( + """INSERT OR IGNORE INTO pending_schedule_wakes ( + schedule_id, agent_name, schedule_name, prompt, + fired_at, created_at + ) VALUES (?, ?, ?, ?, ?, ?)""", + ( + schedule_id, + agent_name, + schedule_name, + prompt, + fired_at, + created_at, + ), + ) + created = cursor.rowcount > 0 + row = self._db.execute( + """SELECT id, schedule_id, agent_name, schedule_name, prompt, + fired_at, created_at + FROM pending_schedule_wakes + WHERE schedule_id=? AND fired_at=?""", + (schedule_id, fired_at), + ).fetchone() + self._db.commit() + return PendingScheduleWake(*row), created + + def list_pending_schedule_wakes( + self, agent_name: str | None = None + ) -> list[PendingScheduleWake]: + """Return pending scheduler wakes oldest-first.""" + sql = """SELECT id, schedule_id, agent_name, schedule_name, prompt, + fired_at, created_at + FROM pending_schedule_wakes""" + params: tuple = () + if agent_name is not None: + sql += " WHERE agent_name=?" + params = (agent_name,) + sql += " ORDER BY fired_at ASC, id ASC" + rows = self._db.execute(sql, params).fetchall() + return [PendingScheduleWake(*row) for row in rows] + + def confirm_pending_schedule_wake( + self, pending_id: int, *, delivered_at: float = 0.0 + ) -> bool: + """Atomically stamp the schedule delivered and retire its outbox row.""" + timestamp = delivered_at or time.time() + with self._rmw_lock: + row = self._db.execute( + "SELECT schedule_id FROM pending_schedule_wakes WHERE id=?", + (pending_id,), + ).fetchone() + if row is None: + return False + self._db.execute( + "UPDATE agent_schedules SET last_delivered=? WHERE id=?", + (timestamp, row[0]), + ) + cursor = self._db.execute( + "DELETE FROM pending_schedule_wakes WHERE id=?", + (pending_id,), + ) + self._db.commit() + return cursor.rowcount > 0 + + def discard_pending_schedule_wake(self, pending_id: int) -> bool: + """Retire a pending wake without marking its schedule delivered.""" + cursor = self._db.execute( + "DELETE FROM pending_schedule_wakes WHERE id=?", + (pending_id,), + ) + self._db.commit() + return cursor.rowcount > 0 + # ── Heartbeats ───────────────────────────────────────── def record_heartbeat( diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index d14351ab..0de2b6a0 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -3051,6 +3051,8 @@ def _build_streaming_wake_context( # Not part of the public API; harness should not depend on it. app.state._build_streaming_wake_context = _build_streaming_wake_context + _scheduler_holder: dict[str, AgentScheduler] = {} + def _log_agent_wake_event(agent_name: str, reason: WakeReason) -> None: """Fires from SDK / tmux / codex sessions after a wake prompt is successfully delivered. @@ -3077,6 +3079,21 @@ def _log_agent_wake_event(agent_name: str, reason: WakeReason) -> None: f"api: agent_wake log failed for {agent_name} " f"(reason={reason.value}): {e}" ) + # #949 startup catch-up: a durable wake must target a newly-booted + # session, not the in-memory deque that produced its negative receipt. + # This callback fires only after the orientation wake itself lands, so + # persisted scheduler prompts replay behind a proven-live boot. The + # holder is populated before application startup; keeping the callback + # synchronous lets each transport retain its existing delivery hook. + try: + scheduler_instance = _scheduler_holder.get("scheduler") + if scheduler_instance is not None: + scheduler_instance.replay_pending_for_agent(agent_name) + except Exception as e: + _log( + f"api: PERSISTED_WAKE_REPLAY_TRIGGER_FAILURE for " + f"{agent_name} (reason={reason.value}): {e}" + ) # Exposed for unit-test reach-in (verifying centralized wake logging # advances the cycle-gate boundary). Not part of the public API. @@ -10426,6 +10443,52 @@ def _is_resurrectable(agent_name: str) -> bool: return False # deliberately disconnected — leave it alone return True + def _scheduler_delivery_busy(agent_name: str) -> bool: + """Fresh positive-liveness signal used to extend receipt waits.""" + ss = broker._get_streaming_session(agent_name) + if ss is None: + return False + stats = getattr(ss, "stats", None) or {} + return ( + stats.get("inflight_busy_not_wedged") is True + or stats.get("inflight_active") is True + ) + + async def _notify_owner_undelivered( + agent_name: str, message: str + ) -> bool: + """Send scheduler failures through canonical owner destinations.""" + send_callback = broker.send_callback + destinations = agents.get_owner_notification_destinations() + if not send_callback or not destinations: + return False + last_error = "no configured destination accepted the alert" + for destination in destinations: + try: + result = await send_callback( + agent_name, + destination["platform"], + destination["conversation_id"], + message, + account_id=destination["account_id"], + ) + if not ( + isinstance(result, dict) + and result.get("sent") is True + ): + raise RuntimeError( + "send callback did not confirm delivery" + ) + except Exception as exc: + last_error = f"{type(exc).__name__}: {exc}" + continue + return True + _log( + f"api: OWNER_NOTIFY_DELIVERY_FAILURE for scheduler alert " + f"agent '{agent_name}': {last_error}" + ) + return False + scheduler = AgentScheduler( agents, wake_callback=_wake_callback, @@ -10436,9 +10499,13 @@ def _is_resurrectable(agent_name: str) -> bool: librarian_callback=_librarian_callback, streaming_sessions_fn=lambda: broker._streaming, comms_cleanup_fn=comms.cleanup_expired, + delivery_busy_fn=_scheduler_delivery_busy, + owner_notify_callback=_notify_owner_undelivered, trigger_store=trigger_store, activity=activity, ) + _scheduler_holder["scheduler"] = scheduler + app.state.scheduler = scheduler # Autonomy engine — self-directed work loops autonomy = AutonomyEngine( diff --git a/src/pinky_daemon/scheduler.py b/src/pinky_daemon/scheduler.py index 15df0bfa..c079bdd4 100644 --- a/src/pinky_daemon/scheduler.py +++ b/src/pinky_daemon/scheduler.py @@ -168,6 +168,8 @@ def __init__( streaming_sessions_fn=None, is_resurrectable_fn=None, comms_cleanup_fn=None, + delivery_busy_fn=None, + owner_notify_callback=None, trigger_store=None, activity=None, tick_interval: int = 30, @@ -191,6 +193,13 @@ def __init__( # would refuse anyway, but at the cost of a budget slot and a log line). self._is_resurrectable_fn = is_resurrectable_fn # fn(agent_name) -> bool self._comms_cleanup_fn = comms_cleanup_fn # fn() -> int (expired comms bookkeeping cleanup) + # fn(agent_name) -> bool. True means the inflight watchdog has + # positive busy-not-wedged evidence, so a pending scheduler receipt + # must keep waiting instead of expiring behind a healthy long turn. + self._delivery_busy_fn = delivery_busy_fn + # async fn(agent_name, text) -> bool. FIRED BUT UNDELIVERED must leave + # journald and reach the owner through an out-of-band transport. + self._owner_notify_callback = owner_notify_callback self._trigger_store = trigger_store # TriggerStore | None self._activity = activity # ActivityStore | None self._tick_interval = tick_interval @@ -219,6 +228,8 @@ def __init__( # cohorts while allowing unrelated agents to progress independently. self._schedule_delivery_tasks: set[asyncio.Task] = set() self._schedule_delivery_locks: dict[str, asyncio.Lock] = {} + self._pending_replay_tasks: dict[str, asyncio.Task] = {} + self._owner_alert_tasks: set[asyncio.Task] = set() # Resurrection rate-limit: agent_name -> list[timestamp] of recent attempts. # Used by _check_heartbeats to cap how often we ping the heartbeat_callback # for a stuck agent (avoid thrashing on a persistently-broken session). @@ -229,6 +240,10 @@ async def start(self) -> None: if self._running: return self._running = True + for pending in self._registry.list_pending_schedule_wakes(): + self.replay_pending_for_agent(pending.agent_name) + # Queue startup catch-up before the first live tick can enqueue newer + # cron fires. Per-agent delivery locks preserve that ordering. self._task = asyncio.create_task(self._loop()) _log(f"scheduler: started (tick every {self._tick_interval}s)") @@ -261,6 +276,17 @@ async def stop(self) -> None: if delivery_tasks: await asyncio.gather(*delivery_tasks, return_exceptions=True) self._schedule_delivery_tasks.clear() + replay_tasks = list(self._pending_replay_tasks.values()) + for task in replay_tasks: + if not task.done(): + task.cancel() + if replay_tasks: + await asyncio.gather(*replay_tasks, return_exceptions=True) + self._pending_replay_tasks.clear() + alert_tasks = list(self._owner_alert_tasks) + if alert_tasks: + await asyncio.gather(*alert_tasks, return_exceptions=True) + self._owner_alert_tasks.clear() _log("scheduler: stopped") async def _loop(self) -> None: @@ -335,6 +361,11 @@ async def _check_schedules(self, now: float) -> None: except Exception: pass self._registry.update_schedule_last_run(schedule.id, now) + # Carry the exact fire identity on this queued snapshot. A + # later minute can advance the DB row while this cohort still + # waits behind a long turn, so failure paths must never reread + # mutable last_run from storage. + schedule.last_run = now # Auto-disable one-shot schedules after firing if schedule.one_shot: @@ -343,29 +374,77 @@ async def _check_schedules(self, now: float) -> None: due_by_agent.setdefault(schedule.agent_name, []).append(schedule) + cohort_started: list[asyncio.Event] = [] for agent_name, due_schedules in due_by_agent.items(): + attempt_started = asyncio.Event() task = asyncio.create_task( - self._deliver_schedule_group(agent_name, due_schedules) + self._deliver_schedule_group( + agent_name, + due_schedules, + attempt_started=attempt_started, + ) ) self._schedule_delivery_tasks.add(task) task.add_done_callback(self._schedule_delivery_done) - - # Give callbacks that confirm immediately one loop turn without making - # a busy agent's bounded receipt wait part of the scheduler tick. - if due_by_agent: - await asyncio.sleep(0) + cohort_started.append(attempt_started) + + # Synchronize on the actual start condition rather than counting event + # loop turns. The confirmation monitor adds task indirection on Python + # 3.12/3.13, but a due callback must still begin before this check + # returns (#702). This bound never waits for a receipt (or a dream), and + # prevents an older per-agent delivery lock from blocking the tick. + if cohort_started: + try: + await asyncio.wait_for( + asyncio.gather(*(event.wait() for event in cohort_started)), + timeout=1.0, + ) + except asyncio.TimeoutError: + _log( + "scheduler: cohort-start synchronization timed out; " + "blocked delivery tasks remain queued" + ) async def _deliver_schedule_group( - self, agent_name: str, schedules: list + self, + agent_name: str, + schedules: list, + *, + attempt_started: asyncio.Event | None = None, ) -> None: """Deliver one agent's due prompts serially, including receipt waits.""" lock = self._schedule_delivery_locks.setdefault(agent_name, asyncio.Lock()) next_index = 0 try: + # Persisted wakes belong to the NEXT session, never another cron + # tick on the same doomed transport. Only wait when boot/ + # orientation explicitly scheduled catch-up; otherwise leave the + # outbox untouched until that lifecycle boundary occurs. + replay_task = self._pending_replay_tasks.get(agent_name) + if ( + replay_task is not None + and replay_task is not asyncio.current_task() + and not replay_task.done() + ): + try: + await asyncio.shield(replay_task) + except asyncio.CancelledError: + raise + except Exception as exc: + _log( + f"scheduler: persisted wake catch-up failed before " + f"live cohort for '{agent_name}': " + f"{type(exc).__name__}: {exc}" + ) async with lock: for next_index, schedule in enumerate(schedules): try: - await self._deliver_schedule(schedule) + await self._deliver_schedule( + schedule, + attempt_started=( + attempt_started if next_index == 0 else None + ), + ) except asyncio.CancelledError: self._record_schedule_undelivered( schedule, "delivery canceled before confirmation" @@ -384,12 +463,19 @@ async def _deliver_schedule_group( ) raise - async def _deliver_schedule(self, schedule) -> None: + async def _deliver_schedule( + self, + schedule, + *, + attempt_started: asyncio.Event | None = None, + ) -> None: """Attempt one fired schedule and record confirmed delivery separately.""" confirmed = False failure_reason = "no delivery callback configured" if schedule.direct_send: + if attempt_started is not None: + attempt_started.set() if not schedule.target_channel: failure_reason = "direct send has no target channel" elif self._direct_send_callback is None: @@ -411,9 +497,8 @@ async def _deliver_schedule(self, schedule) -> None: ) elif self._wake_callback: try: - confirmed = await asyncio.wait_for( - self._wake_and_confirm(schedule), - timeout=self._schedule_delivery_timeout, + confirmed = await self._wait_for_wake_confirmation( + schedule, attempt_started=attempt_started ) if not confirmed: failure_reason = "wake callback returned no positive receipt" @@ -426,6 +511,8 @@ async def _deliver_schedule(self, schedule) -> None: raise except Exception as e: failure_reason = f"wake callback raised {type(e).__name__}: {e}" + elif attempt_started is not None: + attempt_started.set() if confirmed: delivered_at = time.time() @@ -439,8 +526,12 @@ async def _deliver_schedule(self, schedule) -> None: "schedule_delivered", f"Schedule '{schedule.name}' delivery confirmed", ) - except Exception: - pass + except Exception as exc: + _log( + f"scheduler: failed to record schedule delivery " + f"activity for '{schedule.agent_name}': " + f"{type(exc).__name__}: {exc}" + ) _log( f"scheduler: delivery confirmed for schedule " f"'{schedule.name}' (#{schedule.id}) for agent " @@ -453,7 +544,29 @@ async def _deliver_schedule(self, schedule) -> None: def _record_schedule_undelivered( self, schedule, failure_reason: str ) -> None: - """Loudly account one fired schedule without advancing delivery.""" + """Persist, alert, and loudly account one unconfirmed fired schedule.""" + persisted = False + alert_this_failure = True + if not schedule.direct_send: + try: + _, created = self._registry.persist_schedule_wake( + schedule.id, + agent_name=schedule.agent_name, + schedule_name=schedule.name, + prompt=( + schedule.prompt + or f"Scheduled wake: {schedule.name}" + ), + fired_at=schedule.last_run, + ) + persisted = True + alert_this_failure = created + except Exception as exc: + _log( + f"scheduler: SCHEDULER_WAKE_PERSIST_FAILURE schedule " + f"'{schedule.name}' (#{schedule.id}) for agent " + f"'{schedule.agent_name}': {type(exc).__name__}: {exc}" + ) if self._activity: try: self._activity.log( @@ -468,9 +581,219 @@ def _record_schedule_undelivered( f"'{schedule.name}' (#{schedule.id}) for agent " f"'{schedule.agent_name}': {failure_reason}" ) + if alert_this_failure: + recovery = ( + " The wake was persisted for the agent's next session." + if persisted + else " WARNING: durable wake persistence did not succeed." + ) + self._queue_owner_alert( + schedule.agent_name, + ( + "🚨 FIRED BUT UNDELIVERED: schedule " + f"'{schedule.name}' (#{schedule.id}) for agent " + f"'{schedule.agent_name}' was not confirmed: " + f"{failure_reason}.{recovery}" + ), + ) + + async def _wait_for_wake_confirmation( + self, + schedule, + *, + attempt_started: asyncio.Event | None = None, + ) -> bool: + """Wait for one exact receipt, extending while positive liveness holds.""" + delivery = asyncio.create_task( + self._wake_and_confirm( + schedule, attempt_started=attempt_started + ) + ) + try: + while True: + try: + return await asyncio.wait_for( + asyncio.shield(delivery), + timeout=self._schedule_delivery_timeout, + ) + except asyncio.TimeoutError: + if self._agent_busy_not_wedged(schedule.agent_name): + _log( + f"scheduler: receipt still pending for schedule " + f"'{schedule.name}' (#{schedule.id}) for agent " + f"'{schedule.agent_name}', but inflight watchdog " + "reports busy-not-wedged; extending delivery timeout" + ) + continue + delivery.cancel() + await asyncio.gather(delivery, return_exceptions=True) + raise + except asyncio.CancelledError: + delivery.cancel() + await asyncio.gather(delivery, return_exceptions=True) + raise + + def _agent_busy_not_wedged(self, agent_name: str) -> bool: + """Read the transport's live positive-liveness signal, failing closed.""" + if self._delivery_busy_fn is None: + return False + try: + return self._delivery_busy_fn(agent_name) is True + except Exception as exc: + _log( + f"scheduler: busy-not-wedged check failed for " + f"'{agent_name}': {type(exc).__name__}: {exc}" + ) + return False - async def _wake_and_confirm(self, schedule) -> bool: + def _queue_owner_alert(self, agent_name: str, message: str) -> None: + """Start one owner alert with a strong task reference and loud failure.""" + if self._owner_notify_callback is None: + _log( + f"scheduler: OWNER_NOTIFY_UNAVAILABLE for schedule-delivery " + f"alert agent '{agent_name}'" + ) + return + + async def _notify() -> None: + try: + result = self._owner_notify_callback(agent_name, message) + if inspect.isawaitable(result): + result = await result + if result is not True: + raise RuntimeError("owner notify returned no positive receipt") + except Exception as exc: + _log( + f"scheduler: OWNER_NOTIFY_FAILURE for schedule-delivery " + f"alert agent '{agent_name}': " + f"{type(exc).__name__}: {exc}" + ) + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + _log( + f"scheduler: OWNER_NOTIFY_FAILURE for schedule-delivery " + f"alert agent '{agent_name}': no running event loop" + ) + return + task = loop.create_task(_notify()) + self._owner_alert_tasks.add(task) + task.add_done_callback(self._owner_alert_tasks.discard) + + def replay_pending_for_agent(self, agent_name: str) -> None: + """Replay the durable wake outbox after the agent's next session boot.""" + existing = self._pending_replay_tasks.get(agent_name) + if existing is not None and not existing.done(): + return + try: + loop = asyncio.get_running_loop() + except RuntimeError: + _log( + f"scheduler: persisted wake replay trigger for '{agent_name}' " + "has no running event loop; startup catch-up remains pending" + ) + return + task = loop.create_task(self._replay_pending_for_agent(agent_name)) + self._pending_replay_tasks[agent_name] = task + + def _done(done: asyncio.Task) -> None: + if self._pending_replay_tasks.get(agent_name) is done: + self._pending_replay_tasks.pop(agent_name, None) + if not done.cancelled() and done.exception() is not None: + error = done.exception() + _log( + f"scheduler: PERSISTED_WAKE_REPLAY_FAILURE for " + f"'{agent_name}': {type(error).__name__}: {error}" + ) + + task.add_done_callback(_done) + + async def _replay_pending_for_agent(self, agent_name: str) -> None: + """Deliver one agent's persisted wakes FIFO and retire exact successes.""" + lock = self._schedule_delivery_locks.setdefault(agent_name, asyncio.Lock()) + async with lock: + await self._replay_pending_locked(agent_name) + + async def _replay_pending_locked(self, agent_name: str) -> None: + """Replay FIFO pending wakes while the caller holds the agent lock.""" + pending_wakes = self._registry.list_pending_schedule_wakes(agent_name) + for pending in pending_wakes: + current_schedule = self._registry.get_schedule(pending.schedule_id) + zombie_reason = "" + if current_schedule is None: + zombie_reason = "schedule deleted" + elif current_schedule.agent_name != pending.agent_name: + zombie_reason = "schedule reassigned to another agent" + elif not current_schedule.enabled and not current_schedule.one_shot: + zombie_reason = "schedule disabled" + if zombie_reason: + retired = self._registry.discard_pending_schedule_wake( + pending.id + ) + _log( + f"scheduler: PERSISTED_WAKE_ZOMBIE_DROPPED pending " + f"#{pending.id}, schedule #{pending.schedule_id} for " + f"agent '{pending.agent_name}': {zombie_reason}; " + f"outbox_retired={retired}" + ) + continue + try: + confirmed = await self._wait_for_wake_confirmation(pending) + except asyncio.CancelledError: + raise + except Exception as exc: + _log( + f"scheduler: persisted wake #{pending.id} for " + f"'{agent_name}' remains pending after replay: " + f"{type(exc).__name__}: {exc}" + ) + break + if not confirmed: + _log( + f"scheduler: persisted wake #{pending.id} for " + f"'{agent_name}' remains pending: no positive receipt" + ) + break + delivered_at = time.time() + if not self._registry.confirm_pending_schedule_wake( + pending.id, delivered_at=delivered_at + ): + _log( + f"scheduler: persisted wake #{pending.id} for " + f"'{agent_name}' was confirmed but outbox retirement " + "did not match a row" + ) + break + if self._activity: + try: + self._activity.log( + agent_name, + "schedule_delivered", + f"Schedule '{pending.schedule_name}' persisted " + "wake delivery confirmed", + ) + except Exception as exc: + _log( + f"scheduler: failed to record persisted wake " + f"delivery activity for '{agent_name}': " + f"{type(exc).__name__}: {exc}" + ) + _log( + f"scheduler: persisted wake delivery confirmed for " + f"schedule '{pending.schedule_name}' " + f"(#{pending.schedule_id}) for agent '{agent_name}'" + ) + + async def _wake_and_confirm( + self, + schedule, + *, + attempt_started: asyncio.Event | None = None, + ) -> bool: """Invoke the wake callback and await its exact per-prompt receipt.""" + if attempt_started is not None: + attempt_started.set() main_session_id = f"{schedule.agent_name}-main" result = await self._wake_callback( schedule.agent_name, diff --git a/src/pinky_daemon/tmux_session.py b/src/pinky_daemon/tmux_session.py index 347df5ed..0dc0b150 100644 --- a/src/pinky_daemon/tmux_session.py +++ b/src/pinky_daemon/tmux_session.py @@ -1889,7 +1889,9 @@ def stats(self) -> dict: # read to avoid tearing down a session mid-Workflow. Computed live here # so those paths don't recompute slightly-different truth; cheap when no # turn is in flight (returns early before any filesystem stat). - live = self._watchdog_liveness(time.time()) + now = time.time() + live = self._watchdog_liveness(now) + stall_verdict = self._inflight_stall_verdict(now) return { **self._stats, "state": self.state.value, @@ -1900,6 +1902,14 @@ def stats(self) -> dict: "pending_responses": self._message_queue.qsize(), "inflight_turns": len(self._inflight_metas), "inflight_active": live["active"], + # #949 scheduler receipt waits need the SAME positive verdict the + # inflight watchdog uses. ``inflight_active`` intentionally has a + # narrower recent-write window for outer teardown carve-outs; a + # transcript that grew within the watchdog's full timeout window + # is still proven busy-not-wedged for confirmed wake delivery. + "inflight_busy_not_wedged": ( + live["active"] or stall_verdict == "growing" + ), "inflight_liveness_reason": live["reason"], "inflight_liveness_age_s": live["age_s"], "current_activity": self._current_activity, diff --git a/tests/test_api.py b/tests/test_api.py index 70f3ecc3..56f67c7d 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -7723,6 +7723,48 @@ def test_central_wake_log_advances_cycle_gate_on_warm_wakes(self): ) assert "Grep daemon log for verdict_wedged_inputs" not in out_t3 + def test_central_wake_triggers_persisted_scheduler_catch_up(self): + """#949: every proven session boot wakes the durable scheduler outbox.""" + from pinky_daemon.wake_prompt import WakeReason + + with tempfile.TemporaryDirectory() as tmpdir: + app = self._make_app(os.path.join(tmpdir, "test.db")) + replayed: list[str] = [] + app.state.scheduler.replay_pending_for_agent = replayed.append + + app.state._log_agent_wake_event("dymok", WakeReason.CONTEXT_RESTART) + + assert replayed == ["dymok"] + + @pytest.mark.asyncio + async def test_scheduler_undelivered_alert_uses_owner_destination(self): + """#949 alerting uses the durable #863 owner-notify configuration.""" + with tempfile.TemporaryDirectory() as tmpdir: + app = self._make_app(os.path.join(tmpdir, "test.db")) + app.state.agents.set_owner_notification_destinations([ + { + "platform": "telegram", + "account_id": "acct-1", + "conversation_id": "owner-dm", + "principal_id": "owner-user", + } + ]) + send = AsyncMock(return_value={"sent": True}) + app.state.broker._send_callback = send + + delivered = await app.state.scheduler._owner_notify_callback( + "dymok", "FIRED BUT UNDELIVERED test" + ) + + assert delivered is True + send.assert_awaited_once_with( + "dymok", + "telegram", + "owner-dm", + "FIRED BUT UNDELIVERED test", + account_id="acct-1", + ) + def test_central_wake_log_failure_does_not_advance_gate(self): """#591 P1#2 (Barsik refinement): the callback fires ONLY on successful delivery. If the wake prompt's paste/query fails, diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 4b9fd50b..e2660d93 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -376,6 +376,42 @@ def test_update_last_delivered_is_distinct_from_last_run(self, registry): delivered_at, abs=0.1 ) + def test_pending_schedule_wake_is_idempotent_until_confirmed(self, registry): + registry.register("oleg") + schedule = registry.add_schedule( + "oleg", "0 8 * * *", name="morning", prompt="check mail" + ) + fired_at = time.time() + registry.update_schedule_last_run(schedule.id, fired_at) + + first, first_created = registry.persist_schedule_wake( + schedule.id, + agent_name="oleg", + schedule_name="morning", + prompt="check mail", + fired_at=fired_at, + ) + duplicate, duplicate_created = registry.persist_schedule_wake( + schedule.id, + agent_name="oleg", + schedule_name="morning", + prompt="check mail", + fired_at=fired_at, + ) + + assert first_created is True + assert duplicate_created is False + assert duplicate.id == first.id + assert registry.list_pending_schedule_wakes("oleg") == [first] + + delivered_at = time.time() + assert registry.confirm_pending_schedule_wake( + first.id, delivered_at=delivered_at + ) is True + assert registry.list_pending_schedule_wakes("oleg") == [] + stored = registry.get_schedules("oleg")[0] + assert stored.last_delivered == pytest.approx(delivered_at) + def test_cascade_delete(self, registry): registry.register("oleg") registry.add_schedule("oleg", "0 8 * * *", name="morning") @@ -811,9 +847,441 @@ async def wake_cb(agent_name, session_id, prompt): stored = registry.get_schedules("oleg")[0] assert stored.last_run == pytest.approx(fired_at) assert stored.last_delivered == 0.0 + pending = registry.list_pending_schedule_wakes("oleg") + assert [(wake.schedule_id, wake.prompt) for wake in pending] == [ + (stored.id, "prompt") + ] assert events == ["schedule_fired", "schedule_undelivered"] assert "FIRED BUT UNDELIVERED" in capsys.readouterr().err + @pytest.mark.asyncio + async def test_overlapping_fires_keep_distinct_exact_outbox_rows( + self, registry + ): + """A later tick must not replace an older cohort's fire identity.""" + registry.register("oleg") + registry.add_schedule( + "oleg", "* * * * *", name="overlap", prompt="same schedule" + ) + first_started = asyncio.Event() + attempts: list[str] = [] + + async def no_receipt(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + first_started.set() + return asyncio.get_running_loop().create_future() + + scheduler = AgentScheduler( + registry, + wake_callback=no_receipt, + schedule_delivery_timeout=0.03, + ) + first_fire = 1_800_000_000.0 + second_fire = first_fire + 60 + + await scheduler._check_schedules(first_fire) + await asyncio.wait_for(first_started.wait(), timeout=1) + await scheduler._check_schedules(second_fire) + delivery_tasks = list(scheduler._schedule_delivery_tasks) + await asyncio.wait_for( + asyncio.gather(*delivery_tasks, return_exceptions=True), timeout=1 + ) + + assert attempts == ["same schedule", "same schedule"] + assert [ + pending.fired_at + for pending in registry.list_pending_schedule_wakes("oleg") + ] == [first_fire, second_fire] + + @pytest.mark.asyncio + async def test_busy_not_wedged_extends_receipt_timeout(self, registry, capsys): + registry.register("oleg") + schedule = registry.add_schedule( + "oleg", "* * * * *", name="long-turn", prompt="after busy turn" + ) + receipt = asyncio.get_running_loop().create_future() + busy_checks: list[str] = [] + + async def wake_cb(agent_name, session_id, prompt): + del agent_name, session_id, prompt + asyncio.get_running_loop().call_later( + 0.025, receipt.set_result, True + ) + return receipt + + def busy_fn(agent_name): + busy_checks.append(agent_name) + return True + + scheduler = AgentScheduler( + registry, + wake_callback=wake_cb, + delivery_busy_fn=busy_fn, + schedule_delivery_timeout=0.01, + ) + await scheduler._deliver_schedule(schedule) + + assert len(busy_checks) >= 2 + assert set(busy_checks) == {"oleg"} + assert registry.get_schedules("oleg")[0].last_delivered > 0 + assert registry.list_pending_schedule_wakes("oleg") == [] + assert "busy-not-wedged; extending" in capsys.readouterr().err + + @pytest.mark.asyncio + async def test_undelivered_alerts_owner_and_replays_on_next_boot( + self, registry + ): + registry.register("oleg") + schedule = registry.add_schedule( + "oleg", "* * * * *", name="durable", prompt="durable prompt" + ) + fired_at = time.time() + registry.update_schedule_last_run(schedule.id, fired_at) + schedule.last_run = fired_at + alerts: list[tuple[str, str]] = [] + + async def no_receipt(agent_name, session_id, prompt): + del agent_name, session_id, prompt + return asyncio.get_running_loop().create_future() + + async def owner_notify(agent_name, message): + alerts.append((agent_name, message)) + return True + + first_session = AgentScheduler( + registry, + wake_callback=no_receipt, + owner_notify_callback=owner_notify, + schedule_delivery_timeout=0.01, + ) + await first_session._deliver_schedule(schedule) + await asyncio.sleep(0) + + pending = registry.list_pending_schedule_wakes("oleg") + assert len(pending) == 1 + assert alerts[0][0] == "oleg" + assert "FIRED BUT UNDELIVERED" in alerts[0][1] + assert "persisted for the agent's next session" in alerts[0][1] + + attempts: list[str] = [] + + async def confirmed(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + return True + + next_session = AgentScheduler(registry, wake_callback=confirmed) + next_session.replay_pending_for_agent("oleg") + replay_task = next_session._pending_replay_tasks["oleg"] + await replay_task + + assert attempts == ["durable prompt"] + assert registry.list_pending_schedule_wakes("oleg") == [] + assert registry.get_schedules("oleg")[0].last_delivered > 0 + + @pytest.mark.asyncio + async def test_persisted_fifo_replays_before_new_schedule_cohort( + self, registry + ): + registry.register("oleg") + older = registry.add_schedule( + "oleg", "0 8 * * *", name="older", prompt="older pending" + ) + newer = registry.add_schedule( + "oleg", "0 9 * * *", name="newer", prompt="new live fire" + ) + older_fired_at = time.time() - 60 + registry.update_schedule_last_run(older.id, older_fired_at) + registry.persist_schedule_wake( + older.id, + agent_name="oleg", + schedule_name="older", + prompt="older pending", + fired_at=older_fired_at, + ) + attempts: list[str] = [] + + async def confirmed(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + return True + + scheduler = AgentScheduler(registry, wake_callback=confirmed) + scheduler.replay_pending_for_agent("oleg") + await scheduler._deliver_schedule_group("oleg", [newer]) + + assert attempts == ["older pending", "new live fire"] + assert registry.list_pending_schedule_wakes("oleg") == [] + + @pytest.mark.asyncio + async def test_transient_oldest_replay_failure_halts_fifo(self, registry): + registry.register("oleg") + schedule = registry.add_schedule( + "oleg", "* * * * *", name="fifo", prompt="unused" + ) + registry.persist_schedule_wake( + schedule.id, + agent_name="oleg", + schedule_name="fifo", + prompt="oldest", + fired_at=100.0, + ) + registry.persist_schedule_wake( + schedule.id, + agent_name="oleg", + schedule_name="fifo", + prompt="newer", + fired_at=200.0, + ) + attempts: list[str] = [] + + async def first_fails(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + return False + + scheduler = AgentScheduler(registry, wake_callback=first_fails) + scheduler.replay_pending_for_agent("oleg") + await scheduler._pending_replay_tasks["oleg"] + + assert attempts == ["oldest"] + assert [ + pending.prompt + for pending in registry.list_pending_schedule_wakes("oleg") + ] == ["oldest", "newer"] + + @pytest.mark.asyncio + async def test_terminal_zombie_head_retires_and_fifo_advances( + self, registry, capsys + ): + registry.register("oleg") + zombie = registry.add_schedule( + "oleg", "* * * * *", name="zombie", prompt="never deliver" + ) + live = registry.add_schedule( + "oleg", "* * * * *", name="live", prompt="unused" + ) + registry.persist_schedule_wake( + zombie.id, + agent_name="oleg", + schedule_name="zombie", + prompt="never deliver", + fired_at=100.0, + ) + registry.persist_schedule_wake( + live.id, + agent_name="oleg", + schedule_name="live", + prompt="live one", + fired_at=200.0, + ) + registry.persist_schedule_wake( + live.id, + agent_name="oleg", + schedule_name="live", + prompt="live two", + fired_at=300.0, + ) + registry.remove_schedule(zombie.id) + attempts: list[str] = [] + + async def confirmed(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + return True + + scheduler = AgentScheduler(registry, wake_callback=confirmed) + scheduler.replay_pending_for_agent("oleg") + await scheduler._pending_replay_tasks["oleg"] + + assert attempts == ["live one", "live two"] + assert registry.list_pending_schedule_wakes("oleg") == [] + assert "PERSISTED_WAKE_ZOMBIE_DROPPED" in capsys.readouterr().err + + @pytest.mark.asyncio + async def test_new_cron_fire_does_not_replay_backlog_into_old_session( + self, registry + ): + registry.register("oleg") + older = registry.add_schedule( + "oleg", "0 8 * * *", name="older", prompt="next session only" + ) + newer = registry.add_schedule( + "oleg", "0 9 * * *", name="newer", prompt="new live fire" + ) + older_fired_at = time.time() - 60 + registry.update_schedule_last_run(older.id, older_fired_at) + registry.persist_schedule_wake( + older.id, + agent_name="oleg", + schedule_name="older", + prompt="next session only", + fired_at=older_fired_at, + ) + attempts: list[str] = [] + + async def confirmed(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + return True + + scheduler = AgentScheduler(registry, wake_callback=confirmed) + + await scheduler._deliver_schedule_group("oleg", [newer]) + + assert attempts == ["new live fire"] + assert [ + pending.prompt + for pending in registry.list_pending_schedule_wakes("oleg") + ] == ["next session only"] + + @pytest.mark.asyncio + async def test_canceled_cohort_waiting_for_boot_replay_is_persisted( + self, registry + ): + registry.register("oleg") + older = registry.add_schedule( + "oleg", "0 8 * * *", name="older", prompt="older pending" + ) + newer = registry.add_schedule( + "oleg", "0 9 * * *", name="newer", prompt="new live fire" + ) + older_fired_at = time.time() - 60 + registry.update_schedule_last_run(older.id, older_fired_at) + registry.persist_schedule_wake( + older.id, + agent_name="oleg", + schedule_name="older", + prompt="older pending", + fired_at=older_fired_at, + ) + newer_fired_at = time.time() + registry.update_schedule_last_run(newer.id, newer_fired_at) + newer.last_run = newer_fired_at + replay_started = asyncio.Event() + + async def blocked(agent_name, session_id, prompt): + del agent_name, session_id, prompt + replay_started.set() + return asyncio.get_running_loop().create_future() + + scheduler = AgentScheduler(registry, wake_callback=blocked) + scheduler.replay_pending_for_agent("oleg") + await asyncio.wait_for(replay_started.wait(), timeout=1) + cohort = asyncio.create_task( + scheduler._deliver_schedule_group("oleg", [newer]) + ) + await asyncio.sleep(0) + + cohort.cancel() + with pytest.raises(asyncio.CancelledError): + await cohort + replay = scheduler._pending_replay_tasks["oleg"] + replay.cancel() + await asyncio.gather(replay, return_exceptions=True) + + assert [ + pending.prompt + for pending in registry.list_pending_schedule_wakes("oleg") + ] == ["older pending", "new live fire"] + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("mutation", "reason"), + [("delete", "schedule deleted"), ("disable", "schedule disabled")], + ) + async def test_replay_drops_deleted_or_disabled_zombie_wake( + self, registry, capsys, mutation, reason + ): + registry.register("oleg") + schedule = registry.add_schedule( + "oleg", "0 8 * * *", name="obsolete", prompt="do not deliver" + ) + fired_at = time.time() + registry.update_schedule_last_run(schedule.id, fired_at) + registry.persist_schedule_wake( + schedule.id, + agent_name="oleg", + schedule_name="obsolete", + prompt="do not deliver", + fired_at=fired_at, + ) + if mutation == "delete": + registry.remove_schedule(schedule.id) + else: + registry.toggle_schedule(schedule.id, False) + attempts: list[str] = [] + + async def confirmed(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + return True + + scheduler = AgentScheduler(registry, wake_callback=confirmed) + scheduler.replay_pending_for_agent("oleg") + replay_task = scheduler._pending_replay_tasks["oleg"] + await replay_task + + assert attempts == [] + assert registry.list_pending_schedule_wakes("oleg") == [] + error_log = capsys.readouterr().err + assert "PERSISTED_WAKE_ZOMBIE_DROPPED" in error_log + assert reason in error_log + + @pytest.mark.asyncio + async def test_disabled_one_shot_replays_once_across_scheduler_restart( + self, registry + ): + registry.register("oleg") + schedule = registry.add_schedule( + "oleg", + "0 8 * * *", + name="one-shot", + prompt="deliver exactly once", + one_shot=True, + ) + fired_at = time.time() + registry.update_schedule_last_run(schedule.id, fired_at) + registry.toggle_schedule(schedule.id, False) + registry.persist_schedule_wake( + schedule.id, + agent_name="oleg", + schedule_name="one-shot", + prompt="deliver exactly once", + fired_at=fired_at, + ) + db_path = registry._db_path + registry.close() + reopened = AgentRegistry(db_path=db_path) + attempts: list[str] = [] + + async def confirmed(agent_name, session_id, prompt): + del agent_name, session_id + attempts.append(prompt) + return True + + try: + first_boot = AgentScheduler( + reopened, wake_callback=confirmed, tick_interval=3600 + ) + await first_boot.start() + first_replay = first_boot._pending_replay_tasks["oleg"] + await asyncio.wait_for(first_replay, timeout=1) + await first_boot.stop() + + second_boot = AgentScheduler( + reopened, wake_callback=confirmed, tick_interval=3600 + ) + await second_boot.start() + await asyncio.sleep(0.01) + await second_boot.stop() + + assert attempts == ["deliver exactly once"] + assert reopened.list_pending_schedule_wakes("oleg") == [] + finally: + reopened.close() + @pytest.mark.asyncio async def test_canceled_cohort_accounts_current_and_remaining_as_undelivered( self, registry, capsys @@ -851,6 +1319,10 @@ async def wake_cb(agent_name, session_id, prompt): assert attempts == ["first"] assert all(schedule.last_run > 0 for schedule in schedules) assert all(schedule.last_delivered == 0 for schedule in schedules) + assert [ + pending.prompt + for pending in registry.list_pending_schedule_wakes("oleg") + ] == ["first", "second"] assert events == [ "schedule_fired", "schedule_fired", @@ -1346,12 +1818,14 @@ async def test_schedules_fire_while_dream_runs(self, registry): registry.add_schedule("ivan", "* * * * *", name="mid-dream", prompt="hi") release = asyncio.Event() fired = [] + receipt = asyncio.get_running_loop().create_future() async def dream_cb(agent_name, agent): await release.wait() async def wake_cb(agent_name, session_id, prompt): fired.append(agent_name) + return receipt scheduler = AgentScheduler( registry, dream_callback=dream_cb, wake_callback=wake_cb @@ -1360,6 +1834,11 @@ async def wake_cb(agent_name, session_id, prompt): await asyncio.wait_for(scheduler._check_dreams(now), timeout=2) await asyncio.wait_for(scheduler._check_schedules(now), timeout=2) assert fired == ["ivan"] + delivery_tasks = list(scheduler._schedule_delivery_tasks) + assert delivery_tasks + assert all(not task.done() for task in delivery_tasks) + receipt.set_result(True) + await asyncio.wait_for(asyncio.gather(*delivery_tasks), timeout=2) release.set() await asyncio.wait_for(scheduler._dream_tasks["ivan"], timeout=2) diff --git a/tests/test_tmux_session.py b/tests/test_tmux_session.py index 63dd635a..d220a4fb 100644 --- a/tests/test_tmux_session.py +++ b/tests/test_tmux_session.py @@ -9706,12 +9706,32 @@ def test_watchdog_liveness_surfaced_in_stats(tmp_path) -> None: ss._inflight_tool_calls = {"tool-1": _time.time()} stats = ss.stats assert stats["inflight_active"] is True + assert stats["inflight_busy_not_wedged"] is True assert stats["inflight_liveness_reason"] == "foreground_tool_in_flight" assert stats["inflight_turns"] == 1 +def test_stats_scheduler_busy_uses_full_watchdog_growth_window(tmp_path) -> None: + """#949: scheduler waits must not use only the narrower active window.""" + ss, _ = _make_session() + _seed_inflight(ss) + ss._head_started_at = ( + _time.time() - tmux_session._TURN_DONE_TIMEOUT_SEC - 1 + ) + main = tmp_path / "session.jsonl" + main.write_text("{}") + _age_file(main, 300) + _point_transcript(ss, main) + + stats = ss.stats + + assert stats["inflight_active"] is False + assert stats["inflight_busy_not_wedged"] is True + + def test_stats_inflight_inactive_when_idle(tmp_path) -> None: ss, _ = _make_session() stats = ss.stats assert stats["inflight_active"] is False + assert stats["inflight_busy_not_wedged"] is False assert stats["inflight_liveness_reason"] == "no_inflight_turn"