Add replay-triggered reconciliation for delivery_uncertain

app/services/idempotency.py

@@ -84,6 +84,117 @@ def _replay_from_record(
 
 
 class IdempotencyService:
+    async def _try_reconcile_and_replay(
+        self,
+        session: AsyncSession,
+        *,
+        existing: IdempotencyRecordModel,
+        request_hash: str,
+        wallet_id: str,
+        endpoint: str,
+        idempotency_key: str,
+        wait_timeout_seconds: float,
+        poll_interval_seconds: float,
+    ) -> IdempotencyReplay:
+        """Try immediate reconciliation for dispatched/terminal attempts before wait."""
+        # Check if there's a dispatch attempt linked to this idempotency record
+        # that is either dispatched (ambiguous) or terminal but not finalized.
+        dispatch_attempt = (
+            await session.execute(
+                select(McpDispatchAttemptModel).where(
+                    cast(
+                        ColumnElement[bool],
+                        McpDispatchAttemptModel.idempotency_record_id
+                        == existing.record_id,
+                    )
+                )
+            )
+        ).scalar_one_or_none()
+
+        if dispatch_attempt is not None:
+            # Import here to avoid circular dependency
+            from app.services.mcp_dispatch_attempts import (
+                DISPATCH_TERMINAL_STATES,
+            )
+            from app.services.mcp_dispatch_reconciliation import (
+                get_mcp_dispatch_reconciliation_service,
+            )
+
+            # If attempt is already terminal and the idempotency record has a response,
+            # return it directly without re-reconciling.
+            if dispatch_attempt.state in DISPATCH_TERMINAL_STATES:
+                if existing.response_json is not None:
+                    replay = _replay_from_record(existing, request_hash)
+                    if replay is not None:
+                        return replay
+                # Terminal but no response yet - fall through to wait
+
+            # Only reconcile PREPARED or DISPATCHED attempts that are truly stale
+            # (updated >300s ago, matching the periodic sweep's idle threshold).
+            # Live attempts with active owners must wait, not be stolen.
+            should_reconcile = False
+
+            # Check staleness using updated_at (last state change), matching
+            # list_stale_contexts logic in mcp_dispatch_attempts.py
+            if dispatch_attempt.updated_at is not None:
+                from datetime import timedelta
+
+                from app.core.time import to_naive_utc, utc_now
+
+                # 300 seconds = standard staleness threshold used by periodic sweep
+                STALE_IDLE_SECONDS = 300
+                cutoff = to_naive_utc(utc_now() - timedelta(seconds=STALE_IDLE_SECONDS))
+                updated_naive = to_naive_utc(dispatch_attempt.updated_at)
+                is_stale = updated_naive < cutoff
+
+                if is_stale:
+                    # Attempt hasn't been updated in 300s: truly stale/crashed
+                    should_reconcile = True
+                # else: fresh update, assume live owner, wait
+
+            if should_reconcile:
+                reconciler = get_mcp_dispatch_reconciliation_service()
+                try:
+                    await reconciler.reconcile_attempt(
+                        dispatch_attempt.attempt_id,
+                        prepared_error_code="reconciled_stale_prepared",
+                    )
+                except Exception:
+                    # Reconciliation failed; fall through to normal wait path
+                    pass
+                else:
+                    # Reconciliation succeeded; re-read the record
+                    await session.rollback()
+                    session.expire_all()
+                    fresh = (
+                        await session.execute(
+                            select(IdempotencyRecordModel).where(
+                                *_idempotency_predicates(
+                                    wallet_id,
+                                    endpoint,
+                                    idempotency_key,
+                                )
+                            )
+                        )
+                    ).scalar_one_or_none()
+                    if fresh is not None and fresh.response_json is not None:
+                        replay = _replay_from_record(fresh, request_hash)
+                        if replay is not None:
+                            return replay
+
+        # No dispatch attempt or reconciliation failed/incomplete; wait normally
+        if wait_timeout_seconds <= 0:
+            raise IdempotencyInProgressError("idempotency_in_progress")
+        return await self._wait_for_replay(
+            session,
+            wallet_id=wallet_id,
+            endpoint=endpoint,
+            idempotency_key=idempotency_key,
+            request_hash=request_hash,
+            timeout_seconds=wait_timeout_seconds,
+            poll_interval_seconds=poll_interval_seconds,
+        )
+
     async def _wait_for_replay(
         self,
         session: AsyncSession,
@@ -148,15 +259,16 @@ async def begin_with_record(
                 try:
                     replay = _replay_from_record(existing, request_hash)
                 except IdempotencyInProgressError:
-                    if wait_timeout_seconds <= 0:
-                        raise
-                    replay = await self._wait_for_replay(
+                    # Before declaring in-progress, check if there's a terminal or
+                    # dispatched attempt that can be reconciled immediately.
+                    replay = await self._try_reconcile_and_replay(
                         session,
+                        existing=existing,
+                        request_hash=request_hash,
                         wallet_id=wallet_id,
                         endpoint=endpoint,
                         idempotency_key=idempotency_key,
-                        request_hash=request_hash,
-                        timeout_seconds=wait_timeout_seconds,
+                        wait_timeout_seconds=wait_timeout_seconds,
                         poll_interval_seconds=poll_interval_seconds,
                     )
                 return IdempotencyBegin(
@@ -193,15 +305,14 @@ async def begin_with_record(
                 try:
                     replay = _replay_from_record(existing, request_hash)
                 except IdempotencyInProgressError:
-                    if wait_timeout_seconds <= 0:
-                        raise
-                    replay = await self._wait_for_replay(
+                    replay = await self._try_reconcile_and_replay(
                         session,
+                        existing=existing,
+                        request_hash=request_hash,
                         wallet_id=wallet_id,
                         endpoint=endpoint,
                         idempotency_key=idempotency_key,
-                        request_hash=request_hash,
-                        timeout_seconds=wait_timeout_seconds,
+                        wait_timeout_seconds=wait_timeout_seconds,
                         poll_interval_seconds=poll_interval_seconds,
                     )
                 return IdempotencyBegin(
@@ -231,15 +342,14 @@ async def begin_with_record(
                 try:
                     replay = _replay_from_record(existing, request_hash)
                 except IdempotencyInProgressError:
-                    if wait_timeout_seconds <= 0:
-                        raise
-                    replay = await self._wait_for_replay(
+                    replay = await self._try_reconcile_and_replay(
                         session,
+                        existing=existing,
+                        request_hash=request_hash,
                         wallet_id=wallet_id,
                         endpoint=endpoint,
                         idempotency_key=idempotency_key,
-                        request_hash=request_hash,
-                        timeout_seconds=wait_timeout_seconds,
+                        wait_timeout_seconds=wait_timeout_seconds,
                         poll_interval_seconds=poll_interval_seconds,
                     )
                 return IdempotencyBegin(