Fix confirmed-delivery wake loss across session restart

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(

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(