From 59e952366658cdab614561ebb31398e6dc6f040b Mon Sep 17 00:00:00 2001 From: Alex Date: Sun, 7 Jun 2026 13:17:50 +0100 Subject: [PATCH] fix: revoke stale UI surfaces when durable state goes terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 157 +++++++++++++++++- application/api/user/schedules/routes.py | 32 ++++ application/api/user/tasks.py | 60 ++++++- .../db/repositories/pending_tool_state.py | 15 +- .../storage/db/repositories/reconciliation.py | 19 ++- .../agents/schedules/schedulesSlice.test.ts | 44 +++++ .../src/agents/schedules/schedulesSlice.ts | 25 +++ frontend/src/events/dispatchEvent.test.ts | 23 +++ frontend/src/events/dispatchEvent.ts | 13 ++ .../src/notifications/ToolApprovalToast.tsx | 14 ++ .../notifications/notificationsSlice.test.ts | 86 ++++++++++ .../src/notifications/notificationsSlice.ts | 58 +++++++ frontend/src/upload/uploadSlice.ts | 6 +- tests/api/user/test_reconciliation.py | 147 ++++++++++++++++ .../repositories/test_pending_tool_state.py | 10 +- 15 files changed, 682 insertions(+), 27 deletions(-) diff --git a/application/api/user/reconciliation.py b/application/api/user/reconciliation.py index 72c4a5be..d9656b10 100644 --- a/application/api/user/reconciliation.py +++ b/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( diff --git a/application/api/user/schedules/routes.py b/application/api/user/schedules/routes.py index 9f82bac3..30935190 100644 --- a/application/api/user/schedules/routes.py +++ b/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//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}) diff --git a/application/api/user/tasks.py b/application/api/user/tasks.py index 0e185f0a..9965749f 100644 --- a/application/api/user/tasks.py +++ b/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) diff --git a/application/storage/db/repositories/pending_tool_state.py b/application/storage/db/repositories/pending_tool_state.py index 2c8dadf9..06299215 100644 --- a/application/storage/db/repositories/pending_tool_state.py +++ b/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()] diff --git a/application/storage/db/repositories/reconciliation.py b/application/storage/db/repositories/reconciliation.py index b31a3dab..b4be9c14 100644 --- a/application/storage/db/repositories/reconciliation.py +++ b/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}, diff --git a/frontend/src/agents/schedules/schedulesSlice.test.ts b/frontend/src/agents/schedules/schedulesSlice.test.ts index 1960ce00..cb49d066 100644 --- a/frontend/src/agents/schedules/schedulesSlice.test.ts +++ b/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, + }; + 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 = { diff --git a/frontend/src/agents/schedules/schedulesSlice.ts b/frontend/src/agents/schedules/schedulesSlice.ts index e0a5e48c..3828c95c 100644 --- a/frontend/src/agents/schedules/schedulesSlice.ts +++ b/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; diff --git a/frontend/src/events/dispatchEvent.test.ts b/frontend/src/events/dispatchEvent.test.ts index a7a369f0..3b4144fd 100644 --- a/frontend/src/events/dispatchEvent.test.ts +++ b/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)); + }); }); diff --git a/frontend/src/events/dispatchEvent.ts b/frontend/src/events/dispatchEvent.ts index 51e0beb2..c08be023 100644 --- a/frontend/src/events/dispatchEvent.ts +++ b/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 = 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; diff --git a/frontend/src/notifications/ToolApprovalToast.tsx b/frontend/src/notifications/ToolApprovalToast.tsx index 4e369f83..79e3dd65 100644 --- a/frontend/src/notifications/ToolApprovalToast.tsx +++ b/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) { diff --git a/frontend/src/notifications/notificationsSlice.test.ts b/frontend/src/notifications/notificationsSlice.test.ts index 27839d47..f08a51db 100644 --- a/frontend/src/notifications/notificationsSlice.test.ts +++ b/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 => ({ + 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 => ({ + 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', + ); + }); +}); diff --git a/frontend/src/notifications/notificationsSlice.ts b/frontend/src/notifications/notificationsSlice.ts index 7708d312..908ab594 100644 --- a/frontend/src/notifications/notificationsSlice.ts +++ b/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) => { + 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 => diff --git a/frontend/src/upload/uploadSlice.ts b/frontend/src/upload/uploadSlice.ts index 824a7fc5..c0ecf35a 100644 --- a/frontend/src/upload/uploadSlice.ts +++ b/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( diff --git a/tests/api/user/test_reconciliation.py b/tests/api/user/test_reconciliation.py index e0b8b29d..8dbf3fd6 100644 --- a/tests/api/user/test_reconciliation.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/storage/db/repositories/test_pending_tool_state.py b/tests/storage/db/repositories/test_pending_tool_state.py index 747a9840..4ee250aa 100644 --- a/tests/storage/db/repositories/test_pending_tool_state.py +++ b/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"