fix: revoke stale UI surfaces when durable state goes terminal

The "Tool approval needed" toast (and several sibling surfaces) could
linger after the state they represent was already gone. User-scoped SSE
events (tool.approval.required, schedule.autopaused, attachment.queued,
…) are durable and replayed on reconnect, but no terminal path emitted a
matching clearing event and the reconciler only wrote operator-facing
stack_logs — so a failed/expired message replayed its approval prompt
with nothing to act on, and the toast trusted event presence over the
actual message state.

Backend — emit a user-facing event on every terminal path:
- reconciler deletes pending_tool_state and publishes
  tool.approval.cleared when a stuck message is failed;
  cleanup_pending_tool_state does the same for TTL-reaped rows
- reconciler now publishes source.ingest.failed (stalled ingest),
  schedule.run.failed (timeout/pending) and schedule.completed (once)
- schedules PATCH-resume / DELETE publish schedule.resumed / .cancelled
  so a stale schedule.autopaused can't outvote them on replay
- store_attachment gains the on_poison terminal-event hook the ingest
  tasks already have; mcp_oauth_task gains a soft/hard time limit so a
  hung flow self-reports mcp.oauth.failed
- tighten the reconciler exemption to the (conversation_id, user_id)
  composite key

Frontend — stop trusting event presence over truth:
- notificationsSlice.resolveToolApproval evicts the matching
  tool.approval.required and persists its (stable) id dismissed so the
  backlog replay stays suppressed
- ToolApprovalToast drops approval events older than the resumable TTL
  window as a backstop for a lost clearing event
- schedulesSlice handles schedule.resumed / .cancelled / .completed
- upload dismissals now outlive the SSE backlog retention window

Tests cover the new clearing events at the reconciler, slice and
dispatch layers.

application/api/user/reconciliation.py

@@ -12,6 +12,9 @@ from sqlalchemy import Connection
 from application.api.user.idempotency import MAX_TASK_ATTEMPTS
 from application.core.settings import settings
 from application.storage.db.engine import get_engine
+from application.storage.db.repositories.pending_tool_state import (
+    PendingToolStateRepository,
+)
 from application.storage.db.repositories.reconciliation import (
     ReconciliationRepository,
 )
@@ -52,9 +55,14 @@ def run_reconciliation() -> Dict[str, Any]:
         "idempotency_pending_failed": 0,
         "schedule_runs_failed": 0,
     }
+    # User-facing events to fan out once their DB writes have committed
+    # (publish-after-commit). Each item is
+    # ``(user_id, event_type, payload, scope)``.
+    events: list[tuple] = []
 
     with engine.begin() as conn:
         repo = ReconciliationRepository(conn)
+        pt_repo = PendingToolStateRepository(conn)
         for msg in repo.find_and_lock_stuck_messages():
             new_count = repo.increment_message_reconcile_attempts(msg["id"])
             if new_count >= MAX_MESSAGE_RECONCILE_ATTEMPTS:
@@ -66,6 +74,30 @@ def run_reconciliation() -> Dict[str, Any]:
                     ),
                 )
                 summary["messages_failed"] += 1
+                # Revoke any awaiting-approval prompt: the resumable state
+                # dies with the message, so the durable
+                # ``tool.approval.required`` envelope must be cleared or the
+                # UI toast lingers on reconnect. Only emit when a row was
+                # actually deleted so non-approval failures stay quiet.
+                user_id = msg.get("user_id")
+                conversation_id = msg.get("conversation_id")
+                if (
+                    user_id
+                    and conversation_id
+                    and pt_repo.delete_state(str(conversation_id), str(user_id))
+                ):
+                    events.append(
+                        (
+                            str(user_id),
+                            "tool.approval.cleared",
+                            {
+                                "conversation_id": str(conversation_id),
+                                "message_id": str(msg["id"]),
+                                "reason": "failed",
+                            },
+                            {"kind": "conversation", "id": str(conversation_id)},
+                        )
+                    )
                 _emit_alert(
                     conn,
                     name="reconciler_message_failed",
@@ -131,7 +163,7 @@ def run_reconciliation() -> Dict[str, Any]:
             _emit_alert(
                 conn,
                 name="reconciler_ingest_stalled",
-                user_id=None,
+                user_id=row.get("user_id"),
                 detail={
                     "source_id": str(row.get("source_id")),
                     "embedded_chunks": row.get("embedded_chunks"),
@@ -140,6 +172,25 @@ def run_reconciliation() -> Dict[str, Any]:
                 },
             )
             repo.mark_ingest_stalled(str(row["source_id"]))
+            # Tell the upload toast the ingest is done-for. Without it the
+            # toast spins on "Training…" for the whole live session; only
+            # ``source.ingest.failed`` flips it to a terminal error.
+            user_id = row.get("user_id")
+            source_id = row.get("source_id")
+            if user_id and source_id:
+                events.append(
+                    (
+                        str(user_id),
+                        "source.ingest.failed",
+                        {
+                            "source_id": str(source_id),
+                            "filename": row.get("source_name") or "",
+                            "operation": "upload",
+                            "error": "Indexing stopped after a stall.",
+                        },
+                        {"kind": "source", "id": str(source_id)},
+                    )
+                )
 
     # Q5: idempotency rows whose lease expired with attempts exhausted.
     # The wrapper's poison-loop guard normally finalises these, but if
@@ -201,10 +252,21 @@ def run_reconciliation() -> Dict[str, Any]:
                 },
             )
             schedules_repo.bump_failure_count(str(run["schedule_id"]))
-            _terminal_flip_once_schedule(
+            flipped = _terminal_flip_once_schedule(
                 schedules_repo, str(run["schedule_id"]),
             )
             summary["schedule_runs_failed"] += 1
+            events.extend(
+                _schedule_terminal_events(
+                    run,
+                    error_type="timeout",
+                    error=(
+                        "reconciler: schedule_run stuck in 'running' past "
+                        f"{stuck_age} min"
+                    ),
+                    once_completed=flipped,
+                )
+            )
             _emit_alert(
                 conn,
                 name="reconciler_schedule_run_timeout",
@@ -234,10 +296,21 @@ def run_reconciliation() -> Dict[str, Any]:
                 },
             )
             schedules_repo.bump_failure_count(str(run["schedule_id"]))
-            _terminal_flip_once_schedule(
+            flipped = _terminal_flip_once_schedule(
                 schedules_repo, str(run["schedule_id"]),
             )
             summary["schedule_runs_failed"] += 1
+            events.extend(
+                _schedule_terminal_events(
+                    run,
+                    error_type="internal",
+                    error=(
+                        "reconciler: schedule_run stuck in 'pending' past "
+                        f"{stuck_age} min (worker_never_started)"
+                    ),
+                    once_completed=flipped,
+                )
+            )
             _emit_alert(
                 conn,
                 name="reconciler_schedule_run_pending",
@@ -248,25 +321,95 @@ def run_reconciliation() -> Dict[str, Any]:
                 },
             )
 
+    _publish_events(events)
     return summary
 
 
 def _terminal_flip_once_schedule(
     schedules_repo: "SchedulesRepository", schedule_id: str,
-) -> None:
+) -> bool:
     """Flip a once-schedule to 'completed' after its run terminates.
 
     Recurring schedules keep firing; once-schedules would otherwise read
-    'active forever' since next_run_at is already NULL.
+    'active forever' since next_run_at is already NULL. Returns ``True``
+    when the flip happened so the caller can publish a status event.
     """
     schedule = schedules_repo.get_internal(schedule_id)
     if schedule is None or schedule.get("trigger_type") != "once":
-        return
+        return False
     if schedule.get("status") in {"completed", "cancelled"}:
-        return
+        return False
     schedules_repo.update_internal(
         schedule_id, {"status": "completed", "next_run_at": None},
     )
+    return True
+
+
+def _schedule_terminal_events(
+    run: Dict[str, Any],
+    *,
+    error_type: str,
+    error: str,
+    once_completed: bool,
+) -> list:
+    """Build the user-facing events for a reconciler-failed schedule run.
+
+    Always a ``schedule.run.failed`` (so a watching run-log updates live),
+    plus ``schedule.completed`` when a once-schedule flipped terminal — the
+    UI otherwise shows it 'active' with stale Edit/Cancel actions until a
+    manual refresh.
+    """
+    user_id = run.get("user_id")
+    schedule_id = run.get("schedule_id")
+    if not user_id or not schedule_id:
+        return []
+    scope = {"kind": "schedule", "id": str(schedule_id)}
+    out: list = [
+        (
+            str(user_id),
+            "schedule.run.failed",
+            {
+                "run_id": str(run["id"]),
+                "schedule_id": str(schedule_id),
+                "status": "failed",
+                "error_type": error_type,
+                "error": error,
+            },
+            scope,
+        )
+    ]
+    if once_completed:
+        out.append(
+            (
+                str(user_id),
+                "schedule.completed",
+                {"schedule_id": str(schedule_id), "status": "completed"},
+                scope,
+            )
+        )
+    return out
+
+
+def _publish_events(events: list) -> None:
+    """Fan out user-facing events after their DB writes have committed.
+
+    Each item is ``(user_id, event_type, payload, scope)``. Best-effort:
+    a publish miss is swallowed per event so one failure can't strand the
+    rest, and notifications never surface as a reconciler-task failure.
+    """
+    if not events:
+        return
+    from application.events.publisher import publish_user_event
+
+    for user_id, event_type, payload, scope in events:
+        try:
+            publish_user_event(user_id, event_type, payload, scope=scope)
+        except Exception:
+            logger.exception(
+                "reconciler: failed to publish %s for user=%s",
+                event_type,
+                user_id,
+            )
 
 
 def _emit_alert(

application/api/user/schedules/routes.py

@@ -117,6 +117,31 @@ def _user_id() -> Optional[str]:
     return decoded.get("sub")
 
 
+def _publish_schedule_event(
+    user_id: str, event_type: str, schedule_id: str, *, status: str,
+) -> None:
+    """Best-effort SSE so other clients revoke a stale schedule state.
+
+    A resume/cancel must override an earlier ``schedule.autopaused``; that
+    envelope is durable and replays from the backlog on reconnect, so
+    without a newer event the schedule reads 'paused' even though it will
+    fire on the next tick.
+    """
+    try:
+        from application.events.publisher import publish_user_event
+
+        publish_user_event(
+            user_id,
+            event_type,
+            {"schedule_id": str(schedule_id), "status": status},
+            scope={"kind": "schedule", "id": str(schedule_id)},
+        )
+    except Exception:
+        logger.exception(
+            "schedules: publish %s failed for %s", event_type, schedule_id,
+        )
+
+
 @schedules_ns.route("/agents/<string:agent_id>/schedules")
 class AgentSchedules(Resource):
     @api.doc(description="List schedules for an agent (recurring + one-time).")
@@ -468,6 +493,10 @@ class ScheduleResource(Resource):
             )
             if action == "resume":
                 SchedulesRepository(conn).reset_failure_count(schedule_id)
+        if action == "resume" and updated:
+            _publish_schedule_event(
+                user_id, "schedule.resumed", schedule_id, status="active",
+            )
         return _ok({"schedule": _format_schedule(updated or {})})
 
     @api.doc(description="Cancel / delete a schedule.")
@@ -482,6 +511,9 @@ class ScheduleResource(Resource):
             ok = SchedulesRepository(conn).delete(schedule_id, user_id)
         if not ok:
             return _err("schedule not found", 404)
+        _publish_schedule_event(
+            user_id, "schedule.cancelled", schedule_id, status="cancelled",
+        )
         return _ok({"success": True})
 
 

application/api/user/tasks.py

@@ -149,8 +149,36 @@ def sync_source(
     return resp
 
 
+def _emit_attachment_poison_event(task_name, bound):
+    """Publish a terminal ``attachment.failed`` when the poison-guard trips.
+
+    Mirrors ``_emit_ingest_poison_event``: the guard returns before the
+    worker runs, so ``attachment_worker``'s own events never fire and the
+    upload toast would otherwise spin on "processing" forever.
+    """
+    user = bound.get("user")
+    file_info = bound.get("file_info") or {}
+    attachment_id = file_info.get("attachment_id")
+    if not user or not attachment_id:
+        return
+    from application.events.publisher import publish_user_event
+
+    publish_user_event(
+        user,
+        "attachment.failed",
+        {
+            "attachment_id": str(attachment_id),
+            "filename": file_info.get("filename") or "",
+            "error": "Attachment processing stopped after repeated failures.",
+        },
+        scope={"kind": "attachment", "id": str(attachment_id)},
+    )
+
+
 @celery.task(**DURABLE_TASK)
-@with_idempotency(task_name="store_attachment")
+@with_idempotency(
+    task_name="store_attachment", on_poison=_emit_attachment_poison_event,
+)
 def store_attachment(self, file_info, user, idempotency_key=None):
     resp = attachment_worker(self, file_info, user)
     return resp
@@ -325,7 +353,13 @@ def setup_periodic_tasks(sender, **kwargs):
     )
 
 
-@celery.task(bind=True)
+# Bound time limits so a hung OAuth discovery (user never finishes the
+# consent flow, upstream never redirects) self-terminates instead of
+# stranding the ``mcp.oauth.awaiting_redirect`` envelope forever. The
+# soft limit raises inside ``mcp_oauth``'s ``try`` so it publishes a
+# terminal ``mcp.oauth.failed``; the hard limit is the prefork backstop.
+# Generous so a human actively clicking through OAuth isn't cut off.
+@celery.task(bind=True, soft_time_limit=600, time_limit=660)
 def mcp_oauth_task(self, config, user):
     resp = mcp_oauth(self, config, user)
     return resp
@@ -347,8 +381,26 @@ def cleanup_pending_tool_state(self):
     with engine.begin() as conn:
         repo = PendingToolStateRepository(conn)
         reverted = repo.revert_stale_resuming(grace_seconds=600)
-        deleted = repo.cleanup_expired()
-    return {"deleted": deleted, "reverted": reverted}
+        cleared = repo.cleanup_expired()
+
+    # Reaping the resumable state retires any awaiting-approval prompt
+    # tied to it. Without a clearing event the durable
+    # ``tool.approval.required`` envelope replays on reconnect and the UI
+    # toast lingers for a conversation that can no longer be resumed.
+    from application.events.publisher import publish_user_event
+
+    for row in cleared:
+        user_id = row.get("user_id")
+        conversation_id = row.get("conversation_id")
+        if not user_id or not conversation_id:
+            continue
+        publish_user_event(
+            str(user_id),
+            "tool.approval.cleared",
+            {"conversation_id": str(conversation_id), "reason": "expired"},
+            scope={"kind": "conversation", "id": str(conversation_id)},
+        )
+    return {"deleted": len(cleared), "reverted": reverted}
 
 
 @celery.task(bind=True, acks_late=False)

application/storage/db/repositories/pending_tool_state.py

@@ -163,16 +163,21 @@ class PendingToolStateRepository:
         )
         return result.rowcount
 
-    def cleanup_expired(self) -> int:
-        """Delete rows where ``expires_at < now()``.
+    def cleanup_expired(self) -> list[dict]:
+        """Delete TTL-expired rows; return their ``(conversation_id, user_id)``.
 
         Replaces Mongo's ``expireAfterSeconds=0`` TTL index. Intended to
-        be called from a Celery beat task every 60 seconds.
+        be called from a Celery beat task every 60 seconds. The deleted
+        rows are returned so the caller can revoke any approval prompt
+        tied to the now-gone resumable state.
         """
         # clock_timestamp() — not now() — since the latter is frozen to the
         # start of the transaction, which would let state that has just
         # expired survive one more cleanup tick.
         result = self._conn.execute(
-            text("DELETE FROM pending_tool_state WHERE expires_at < clock_timestamp()")
+            text(
+                "DELETE FROM pending_tool_state WHERE expires_at < clock_timestamp() "
+                "RETURNING conversation_id, user_id"
+            )
         )
-        return result.rowcount
+        return [row_to_dict(r) for r in result.fetchall()]

application/storage/db/repositories/reconciliation.py

@@ -44,6 +44,7 @@ class ReconciliationRepository:
                       SELECT 1
                       FROM pending_tool_state pts
                       WHERE pts.conversation_id = cm.conversation_id
+                        AND pts.user_id = cm.user_id
                         AND (
                             (pts.status = 'pending'
                              AND pts.expires_at > now())
@@ -115,15 +116,17 @@ class ReconciliationRepository:
         result = self._conn.execute(
             text(
                 """
-                SELECT source_id, total_chunks, embedded_chunks,
-                       last_index, last_updated
-                FROM ingest_chunk_progress
-                WHERE last_updated < now() - make_interval(mins => :age)
-                  AND embedded_chunks < total_chunks
-                  AND status = 'active'
-                ORDER BY last_updated ASC
+                SELECT icp.source_id, icp.total_chunks, icp.embedded_chunks,
+                       icp.last_index, icp.last_updated,
+                       s.user_id, s.name AS source_name
+                FROM ingest_chunk_progress icp
+                LEFT JOIN sources s ON s.id = icp.source_id
+                WHERE icp.last_updated < now() - make_interval(mins => :age)
+                  AND icp.embedded_chunks < icp.total_chunks
+                  AND icp.status = 'active'
+                ORDER BY icp.last_updated ASC
                 LIMIT :limit
-                FOR UPDATE SKIP LOCKED
+                FOR UPDATE OF icp SKIP LOCKED
                 """
             ),
             {"age": age_minutes, "limit": limit},

frontend/src/agents/schedules/schedulesSlice.test.ts

@@ -111,6 +111,50 @@ describe('schedulesSlice SSE event handling', () => {
     expect(schedules[0].status).toBe('paused');
   });
 
+  it('schedule.resumed flips a paused schedule back to active', () => {
+    let state = seedState();
+    state = {
+      ...state,
+      byAgent: {
+        'agent-1': [sampleSchedule({ status: 'paused' })],
+      } as Record<string, Schedule[]>,
+    };
+    state = reducer(
+      state,
+      sseEventReceived({
+        id: 'evt-r',
+        ts: '2026-05-19T12:07:30Z',
+        type: 'schedule.resumed',
+        payload: { schedule_id: 'sched-1', status: 'active' },
+      }),
+    );
+    const schedules = selectSchedulesForAgent({ schedules: state }, 'agent-1');
+    expect(schedules[0].status).toBe('active');
+  });
+
+  it('schedule.cancelled and schedule.completed set their terminal status', () => {
+    for (const [type, expected] of [
+      ['schedule.cancelled', 'cancelled'],
+      ['schedule.completed', 'completed'],
+    ] as const) {
+      let state = seedWithSchedule();
+      state = reducer(
+        state,
+        sseEventReceived({
+          id: `evt-${type}`,
+          ts: '2026-05-19T12:09:00Z',
+          type,
+          payload: { schedule_id: 'sched-1', status: expected },
+        }),
+      );
+      const schedules = selectSchedulesForAgent(
+        { schedules: state },
+        'agent-1',
+      );
+      expect(schedules[0].status).toBe(expected);
+    }
+  });
+
   it('schedule.message.appended is acknowledged without mutating run state', () => {
     let state = seedWithSchedule();
     const envelope: SSEEvent = {

frontend/src/agents/schedules/schedulesSlice.ts

@@ -9,6 +9,7 @@ import type {
   Schedule,
   ScheduleCreatePayload,
   ScheduleRun,
+  ScheduleStatus,
   ScheduleUpdatePayload,
 } from '../types/schedule';
 
@@ -294,6 +295,30 @@ const schedulesSlice = createSlice({
               }
               break;
             }
+            // Schedule-level status transitions. These revoke a stale
+            // ``schedule.autopaused`` (resumed/cancelled) or surface a
+            // reconciler/once-schedule finish (completed) that the run
+            // deltas above don't carry — otherwise the schedule reads
+            // 'active' (or 'paused') until a manual refresh.
+            case 'schedule.resumed':
+            case 'schedule.cancelled':
+            case 'schedule.completed': {
+              const status: ScheduleStatus =
+                envelope.type === 'schedule.resumed'
+                  ? 'active'
+                  : envelope.type === 'schedule.cancelled'
+                    ? 'cancelled'
+                    : 'completed';
+              const found = findAgentForSchedule(state, scheduleId);
+              if (found) {
+                const next: Schedule = { ...found.schedule, status };
+                state.byAgent[found.agentId] = upsert(
+                  state.byAgent[found.agentId],
+                  next,
+                );
+              }
+              break;
+            }
             case 'schedule.message.appended':
               // Handled by conversationSlice; nothing to mutate here.
               break;

frontend/src/events/dispatchEvent.test.ts

@@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
 import type { AppDispatch } from '../store';
 import {
+  resolveToolApproval,
   sseEventReceived,
   sseLastEventIdReset,
 } from '../notifications/notificationsSlice';
@@ -52,9 +53,31 @@ describe('dispatchSSEEvent', () => {
     'schedule.run.failed',
     'schedule.autopaused',
     'schedule.message.appended',
+    'tool.approval.cleared',
+    'schedule.resumed',
+    'schedule.cancelled',
+    'schedule.completed',
   ])('treats %s as a known envelope (no debug noise)', (type) => {
     const dispatch = vi.fn() as unknown as AppDispatch;
     dispatchSSEEvent({ id: `e-${type}`, type }, dispatch);
     expect(debugSpy).not.toHaveBeenCalled();
   });
+
+  it('dispatches resolveToolApproval AND sseEventReceived for tool.approval.cleared', () => {
+    const dispatch = vi.fn() as unknown as AppDispatch;
+    const envelope = {
+      id: 'clr-1',
+      type: 'tool.approval.cleared' as const,
+      scope: { kind: 'conversation', id: 'conv-1' },
+      payload: { conversation_id: 'conv-1', message_id: 'msg-1' },
+    };
+
+    dispatchSSEEvent(envelope, dispatch);
+
+    const calls = (dispatch as unknown as { mock: { calls: unknown[][] } }).mock
+      .calls;
+    expect(calls).toHaveLength(2);
+    expect(calls[0][0]).toEqual(resolveToolApproval(envelope));
+    expect(calls[1][0]).toEqual(sseEventReceived(envelope));
+  });
 });

frontend/src/events/dispatchEvent.ts

@@ -1,5 +1,6 @@
 import type { AppDispatch } from '../store';
 import {
+  resolveToolApproval,
   sseEventReceived,
   sseLastEventIdReset,
   type SSEEvent,
@@ -24,10 +25,16 @@ const KNOWN_TYPES: ReadonlySet<string> = new Set([
   'mcp.oauth.completed',
   'mcp.oauth.failed',
   'tool.approval.required',
+  // Revokes a stale tool.approval.required (reconciler / TTL cleanup).
+  'tool.approval.cleared',
   // Scheduler envelopes (scheduler_worker.py); consumed by schedulesSlice.
   'schedule.run.completed',
   'schedule.run.failed',
   'schedule.autopaused',
+  // Revoke a stale schedule.autopaused / surface a once-schedule finish.
+  'schedule.resumed',
+  'schedule.cancelled',
+  'schedule.completed',
   'schedule.message.appended',
 ]);
 
@@ -54,6 +61,12 @@ export function dispatchSSEEvent(
       // can subscribe to ``sseEventReceived`` and refetch.
       dispatch(sseLastEventIdReset());
       break;
+    case 'tool.approval.cleared':
+      // Evict the matching tool.approval.required and persist its id as
+      // dismissed BEFORE it lands in the ring, so the toast surface never
+      // sees a revoked approval (live or on backlog replay).
+      dispatch(resolveToolApproval(envelope));
+      break;
     default:
       // No central side effect; rely on slice-level extraReducers.
       break;

frontend/src/notifications/ToolApprovalToast.tsx

@@ -12,6 +12,10 @@ import {
   selectRecentEvents,
 } from './notificationsSlice';
 
+// Backend ``PENDING_STATE_TTL_SECONDS`` is 30 min; allow a margin for
+// clock skew before treating a still-present approval event as moot.
+const APPROVAL_EVENT_MAX_AGE_MS = 35 * 60 * 1000;
+
 /**
  * Surface ``tool.approval.required`` events as toasts that look like
  * ``UploadToast`` (same fixed bottom-right rail) — but only when the
@@ -62,10 +66,20 @@ export default function ToolApprovalToast() {
     string,
     { eventId: string; conversationId: string }
   >();
+  const now = Date.now();
   for (const event of events) {
     if (event.type !== 'tool.approval.required') continue;
     if (!event.id) continue; // can't dismiss without an id
     if (dismissedSet.has(event.id)) continue;
+    // Timestamp backstop: an approval is only resumable inside the
+    // backend pending-tool-state TTL. Past that window the prompt is
+    // moot, so don't surface it even if no clearing event arrived (lost
+    // publish, older backend). Keyed off the envelope's own ``ts`` so it
+    // holds across reload/backlog replay.
+    if (event.ts) {
+      const age = now - Date.parse(event.ts);
+      if (Number.isFinite(age) && age > APPROVAL_EVENT_MAX_AGE_MS) continue;
+    }
     const conversationId = event.scope?.id;
     if (!conversationId) continue;
     if (currentConversationId && conversationId === currentConversationId) {

frontend/src/notifications/notificationsSlice.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it, vi, afterEach } from 'vitest';
 
 import reducer, {
   dismissToolApproval,
+  resolveToolApproval,
   sseEventReceived,
   sseLastEventIdReset,
   type SSEEvent,
@@ -107,3 +108,88 @@ describe('dismissToolApproval', () => {
     expect(state.dismissedToolApprovals[199].id).toBe('id-204');
   });
 });
+
+describe('resolveToolApproval', () => {
+  const required = (overrides: Partial<SSEEvent> = {}): SSEEvent => ({
+    id: 'req-1',
+    type: 'tool.approval.required',
+    scope: { kind: 'conversation', id: 'conv-1' },
+    payload: { conversation_id: 'conv-1', message_id: 'msg-1' },
+    ...overrides,
+  });
+
+  const cleared = (overrides: Partial<SSEEvent> = {}): SSEEvent => ({
+    id: 'clr-1',
+    type: 'tool.approval.cleared',
+    scope: { kind: 'conversation', id: 'conv-1' },
+    payload: {
+      conversation_id: 'conv-1',
+      message_id: 'msg-1',
+      reason: 'failed',
+    },
+    ...overrides,
+  });
+
+  it('evicts the matching required by message_id and persists its id dismissed', () => {
+    let state = seedState();
+    state = reducer(state, sseEventReceived(required()));
+    expect(state.recentEvents).toHaveLength(1);
+
+    state = reducer(state, resolveToolApproval(cleared()));
+
+    expect(
+      state.recentEvents.some((e) => e.type === 'tool.approval.required'),
+    ).toBe(false);
+    expect(state.dismissedToolApprovals.map((d) => d.id)).toContain('req-1');
+  });
+
+  it('matches by conversation scope when the clear carries no message_id', () => {
+    let state = seedState();
+    state = reducer(
+      state,
+      sseEventReceived(
+        required({
+          id: 'req-2',
+          scope: { kind: 'conversation', id: 'conv-2' },
+          payload: { conversation_id: 'conv-2' },
+        }),
+      ),
+    );
+
+    state = reducer(
+      state,
+      resolveToolApproval(
+        cleared({
+          id: 'clr-2',
+          scope: { kind: 'conversation', id: 'conv-2' },
+          payload: { conversation_id: 'conv-2', reason: 'expired' },
+        }),
+      ),
+    );
+
+    expect(
+      state.recentEvents.some((e) => e.type === 'tool.approval.required'),
+    ).toBe(false);
+    expect(state.dismissedToolApprovals.map((d) => d.id)).toContain('req-2');
+  });
+
+  it('leaves a different message untouched so a new pause still surfaces', () => {
+    let state = seedState();
+    state = reducer(
+      state,
+      sseEventReceived(
+        required({
+          id: 'req-3',
+          payload: { conversation_id: 'conv-1', message_id: 'msg-OTHER' },
+        }),
+      ),
+    );
+
+    state = reducer(state, resolveToolApproval(cleared()));
+
+    expect(state.recentEvents.some((e) => e.id === 'req-3')).toBe(true);
+    expect(state.dismissedToolApprovals.map((d) => d.id)).not.toContain(
+      'req-3',
+    );
+  });
+});

frontend/src/notifications/notificationsSlice.ts

@@ -160,6 +160,63 @@ export const notificationsSlice = createSlice({
         state.dismissedToolApprovals,
       );
     },
+    /**
+     * Revoke a pending approval when the backend says it can no longer be
+     * acted on (``tool.approval.cleared`` — the message failed or the
+     * resumable state was reaped). Evicts the matching
+     * ``tool.approval.required`` from the ring AND records its (stable
+     * Redis-stream) id in the dismissed set, so the durable envelope
+     * replayed on reconnect stays suppressed without re-popping the toast.
+     *
+     * Match by ``payload.message_id`` when the clearing event carries one
+     * (precise — a later pause of the same conversation has a new message
+     * id and still surfaces); otherwise fall back to the conversation
+     * ``scope.id``.
+     */
+    resolveToolApproval: (state, action: PayloadAction<SSEEvent>) => {
+      const cleared = action.payload;
+      const conversationId = cleared.scope?.id;
+      const messageId = (cleared.payload as { message_id?: string } | undefined)
+        ?.message_id;
+      if (!conversationId && !messageId) return;
+
+      const now = Date.now();
+      const cutoff = now - DISMISSED_TOOL_APPROVALS_TTL_MS;
+      state.dismissedToolApprovals = state.dismissedToolApprovals.filter(
+        (entry) => entry.at >= cutoff,
+      );
+
+      const dismissedIds = new Set(
+        state.dismissedToolApprovals.map((entry) => entry.id),
+      );
+      let changed = false;
+      state.recentEvents = state.recentEvents.filter((e) => {
+        if (e.type !== 'tool.approval.required' || !e.id) return true;
+        const eMsg = (e.payload as { message_id?: string } | undefined)
+          ?.message_id;
+        const match = messageId
+          ? eMsg === messageId
+          : e.scope?.id === conversationId;
+        if (!match) return true;
+        if (!dismissedIds.has(e.id)) {
+          state.dismissedToolApprovals.push({ id: e.id, at: now });
+          dismissedIds.add(e.id);
+        }
+        changed = true;
+        return false;
+      });
+
+      if (!changed) return;
+      if (state.dismissedToolApprovals.length > DISMISSED_TOOL_APPROVALS_CAP) {
+        state.dismissedToolApprovals = state.dismissedToolApprovals.slice(
+          -DISMISSED_TOOL_APPROVALS_CAP,
+        );
+      }
+      saveDismissed(
+        DISMISSED_TOOL_APPROVALS_STORAGE_KEY,
+        state.dismissedToolApprovals,
+      );
+    },
   },
 });
 
@@ -171,6 +228,7 @@ export const {
   sseLastEventIdAdvanced,
   clearRecentEvents,
   dismissToolApproval,
+  resolveToolApproval,
 } = notificationsSlice.actions;
 
 export const selectSseHealth = (state: RootState): PushHealth =>

frontend/src/upload/uploadSlice.ts

@@ -10,7 +10,11 @@ import { sseEventReceived } from '../notifications/notificationsSlice';
 import { RootState } from '../store';
 
 const DISMISSED_SOURCE_IDS_STORAGE_KEY = 'docsgpt:dismissedUploadSourceIds';
-const DISMISSED_SOURCE_IDS_TTL_MS = 24 * 60 * 60 * 1000;
+// Must outlive the backend SSE backlog (``EVENTS_STREAM_MAXLEN`` ≈ 24h,
+// longer under light traffic). If the dismissal expires first, a replayed
+// ``source.ingest.*`` for a dismissed source auto-creates a fresh task
+// with ``dismissed=false`` and the toast re-pops. 7 days clears the window.
+const DISMISSED_SOURCE_IDS_TTL_MS = 7 * 24 * 60 * 60 * 1000;
 const DISMISSED_SOURCE_IDS_CAP = 200;
 
 function recordDismissedSourceId(

tests/api/user/test_reconciliation.py

@@ -581,6 +581,38 @@ def _ingest_status(conn, source_id: str) -> str | None:
     return row[0] if row is not None else None
 
 
+def _seed_source(
+    conn, *, source_id: str, user_id: str = "u-1", name: str = "My Doc.pdf",
+) -> str:
+    """Insert a minimal ``sources`` row so the ingest sweep can resolve its owner."""
+    conn.execute(
+        text(
+            "INSERT INTO sources (id, user_id, name, type) "
+            "VALUES (CAST(:id AS uuid), :user_id, :name, 'file')"
+        ),
+        {"id": source_id, "user_id": user_id, "name": name},
+    )
+    return source_id
+
+
+def _capture_published(pg_conn):
+    """Patch ``publish_user_event`` and collect ``(user, type, payload, scope)``.
+
+    Returns a ``(context_manager, captured_list)`` pair. The reconciler
+    imports the publisher lazily inside ``_publish_events``, so patching the
+    function on its home module is what intercepts the call.
+    """
+    captured: list = []
+
+    def _fake(user_id, event_type, payload, *, scope=None):
+        captured.append((user_id, event_type, payload, scope))
+        return "1-0"
+
+    return patch(
+        "application.events.publisher.publish_user_event", _fake,
+    ), captured
+
+
 class TestStalledIngests:
     @pytest.mark.unit
     def test_stalled_ingest_escalated_with_alert(self, pg_conn, caplog):
@@ -784,6 +816,121 @@ class TestStuckIdempotencyPending:
         assert r["idempotency_pending_failed"] == 0
 
 
+# ---------------------------------------------------------------------------
+# Clearing / terminal user-facing events (revoke stale UI surfaces)
+# ---------------------------------------------------------------------------
+
+
+class TestApprovalClearedEvents:
+    @pytest.mark.unit
+    def test_message_failed_clears_pending_approval(self, pg_conn):
+        """A reconciled-to-failed message deletes its resumable state and
+        publishes ``tool.approval.cleared`` so the approval toast doesn't
+        linger after reconnect.
+        """
+        from application.api.user import reconciliation as recon
+
+        msg = _seed_pending_message(pg_conn)
+        # Expired PT row: doesn't shield the message (past TTL) but is the
+        # resumable state the failure path must delete + revoke.
+        _seed_pending_state(
+            pg_conn, msg["conversation_id"], msg["user_id"],
+            expires_in_minutes=-1,
+        )
+
+        ctx, published = _capture_published(pg_conn)
+        with _route_engine_to(pg_conn), ctx:
+            recon.run_reconciliation()
+            recon.run_reconciliation()
+            recon.run_reconciliation()
+
+        status = pg_conn.execute(
+            text(
+                "SELECT status FROM conversation_messages "
+                "WHERE id = CAST(:id AS uuid)"
+            ),
+            {"id": msg["id"]},
+        ).scalar()
+        assert status == "failed"
+
+        # Resumable state is gone.
+        pt_count = pg_conn.execute(
+            text(
+                "SELECT count(*) FROM pending_tool_state "
+                "WHERE conversation_id = CAST(:c AS uuid)"
+            ),
+            {"c": msg["conversation_id"]},
+        ).scalar()
+        assert pt_count == 0
+
+        cleared = [p for p in published if p[1] == "tool.approval.cleared"]
+        assert len(cleared) == 1
+        user_id, _, payload, scope = cleared[0]
+        assert user_id == msg["user_id"]
+        assert payload["conversation_id"] == msg["conversation_id"]
+        assert payload["message_id"] == msg["id"]
+        assert payload["reason"] == "failed"
+        assert scope == {"kind": "conversation", "id": msg["conversation_id"]}
+
+    @pytest.mark.unit
+    def test_message_failed_without_approval_emits_no_clear(self, pg_conn):
+        """A plain stuck message (no resumable state) must not emit a
+        spurious clearing event.
+        """
+        from application.api.user import reconciliation as recon
+
+        _seed_pending_message(pg_conn)
+
+        ctx, published = _capture_published(pg_conn)
+        with _route_engine_to(pg_conn), ctx:
+            recon.run_reconciliation()
+            recon.run_reconciliation()
+            recon.run_reconciliation()
+
+        assert not any(p[1] == "tool.approval.cleared" for p in published)
+
+
+class TestStalledIngestEvent:
+    @pytest.mark.unit
+    def test_stalled_ingest_emits_source_failed_event(self, pg_conn):
+        from application.api.user import reconciliation as recon
+
+        sid = "1a000000-0000-0000-0000-0000000000b1"
+        _seed_source(pg_conn, source_id=sid, user_id="u-ingest", name="report.pdf")
+        _seed_ingest_progress(pg_conn, source_id=sid, embedded=2, total=50)
+
+        ctx, published = _capture_published(pg_conn)
+        with _route_engine_to(pg_conn), ctx:
+            r = recon.run_reconciliation()
+
+        assert r["ingests_stalled"] == 1
+        failed = [p for p in published if p[1] == "source.ingest.failed"]
+        assert len(failed) == 1
+        user_id, _, payload, scope = failed[0]
+        assert user_id == "u-ingest"
+        assert payload["source_id"] == sid
+        assert payload["filename"] == "report.pdf"
+        assert scope == {"kind": "source", "id": sid}
+
+    @pytest.mark.unit
+    def test_orphan_source_stalls_without_event(self, pg_conn):
+        """An ingest row with no matching ``sources`` row (deleted source)
+        still escalates to 'stalled' but emits no user event.
+        """
+        from application.api.user import reconciliation as recon
+
+        sid = "1a000000-0000-0000-0000-0000000000b2"
+        _seed_ingest_progress(pg_conn, source_id=sid, embedded=1, total=20)
+
+        ctx, published = _capture_published(pg_conn)
+        with _route_engine_to(pg_conn), ctx:
+            r = recon.run_reconciliation()
+
+        assert r["ingests_stalled"] == 1
+        assert _ingest_status(pg_conn, sid) == "stalled"
+        assert not any(p[1] == "source.ingest.failed" for p in published)
+
+
 # ---------------------------------------------------------------------------
 # Skip path
 # ---------------------------------------------------------------------------

tests/storage/db/repositories/test_pending_tool_state.py

@@ -90,7 +90,13 @@ class TestCleanupExpired:
         # Create a state with TTL of 0 seconds (already expired)
         repo.save_state(conv["id"], "user-1", **_sample_state(), ttl_seconds=0)
         deleted = repo.cleanup_expired()
-        assert deleted >= 1
+        assert len(deleted) >= 1
+        # Deleted rows carry the keys needed to revoke a stale approval toast.
+        assert any(
+            str(row["conversation_id"]) == conv["id"]
+            and row["user_id"] == "user-1"
+            for row in deleted
+        )
         assert repo.load_state(conv["id"], "user-1") is None
 
 
@@ -204,7 +210,7 @@ class TestRevertStaleResuming:
         # and is well past — without the bump, cleanup_expired would
         # delete the just-reverted row.
         deleted = repo.cleanup_expired()
-        assert deleted == 0
+        assert deleted == []
         loaded = repo.load_state(conv["id"], "user-1")
         assert loaded is not None
         assert loaded["status"] == "pending"