Implement P05 checked services and safe selected delivery recovery

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c

src/user_engine/service.py

@@ -2867,71 +2867,65 @@ class UserEngineService:
         return self.store.pending_outbox()
 
     def deliver_outbox(
-        self,
-        actor: Actor,
-        deliver: Callable[[OutboxEvent], None],
-        *,
-        worker_id: str,
-        max_attempts: int = 3,
-        correlation_id: str | None = None,
+        self, actor: Actor, deliver: Callable[[OutboxEvent], None], *,
+        worker_id: str, max_attempts: int = 3, correlation_id: str | None = None,
+        event_id: str | None = None, expected_attempts: int | None = None,
     ) -> tuple[OutboxEvent, ...]:
-        """Claim and deliver pending events, retaining bounded failure details."""
-        if not worker_id.strip():
-            raise ValidationError("outbox worker_id is required")
-        if not 1 <= max_attempts <= 20:
-            raise ValidationError("outbox max_attempts must be between 1 and 20")
+        """Serialize each external attempt and retain only classified failures."""
+        if not worker_id.strip() or not 1 <= max_attempts <= 20:
+            raise ValidationError("invalid outbox worker or attempt limit")
         correlation_id = correlation_id or new_id("corr")
-        self._authorize(
-            actor, action="outbox.deliver", resource_type="user-engine:outbox",
-            resource_id="pending", tenant=actor.tenant,
-            correlation_id=correlation_id,
-        )
-        results: list[OutboxEvent] = []
-        for event in self.store.pending_outbox():
-            claimed = replace(
-                event, claimed_by=worker_id, claimed_at=utc_now(),
-                delivery_attempts=event.delivery_attempts + 1,
-                failed_at=None, failure_reason=None,
-            )
-            with self.store.transaction():
-                self.store.save_outbox(claimed)
-            try:
-                deliver(claimed)
-            except Exception as exc:
-                reason = str(exc).strip()[:200] or type(exc).__name__
-                failed = replace(
-                    claimed, failed_at=utc_now(), failure_reason=reason,
-                    dead_lettered_at=(utc_now() if claimed.delivery_attempts >= max_attempts else None),
-                )
-                with self.store.transaction():
-                    self.store.save_outbox(failed)
-                results.append(failed)
-            else:
-                delivered = replace(claimed, delivered_at=utc_now())
-                with self.store.transaction():
-                    self.store.save_outbox(delivered)
-                results.append(delivered)
+        self._authorize(actor, action="outbox.deliver", resource_type="user-engine:outbox",
+            resource_id="pending", tenant=actor.tenant, correlation_id=correlation_id)
+        selected = self.store.outbox_event(event_id) if event_id else None
+        if event_id and selected is None:
+            raise NotFoundError("delivery record not found")
+        events = (selected,) if selected else self.store.pending_outbox()
+        results = []
+        for candidate in events:
+            with self.store.outbox_delivery_guard(candidate.event_id):
+                event = self.store.outbox_event(candidate.event_id)
+                if event is None or event.delivered_at is not None:
+                    continue
+                if expected_attempts is not None and event.delivery_attempts != expected_attempts:
+                    raise ConflictError("Delivery changed. Refresh and review the attempt again.")
+                # Owning this cross-process guard proves any previous claim is orphaned.
+                # Receiver idempotency reconciles a lost receipt without blind resending.
+                if event.dead_lettered_at and not event_id:
+                    continue
+                claimed = replace(event, claimed_by=worker_id, claimed_at=utc_now(),
+                    delivery_attempts=event.delivery_attempts+1, failed_at=None, failure_reason=None)
+                with self.store.transaction(): self.store.save_outbox(claimed)
+                try:
+                    deliver(claimed)
+                except Exception as exc:
+                    code = getattr(exc, "code", "delivery_unavailable")
+                    if code not in {"mail_unconfigured", "delivery_outcome_unknown", "temporary_deferral", "permanent_rejection", "recipient_suppressed", "provider_timeout", "provider_unavailable"}:
+                        code = "delivery_unavailable"
+                    outcome = replace(claimed, claimed_by=None, claimed_at=None,
+                        failed_at=utc_now(), failure_reason=code,
+                        dead_lettered_at=utc_now() if claimed.delivery_attempts >= max_attempts else None)
+                else:
+                    outcome = replace(claimed, claimed_by=None, claimed_at=None,
+                        delivered_at=utc_now(), dead_lettered_at=None)
+                with self.store.transaction(): self.store.save_outbox(outcome)
+                results.append(outcome)
         return tuple(results)
 
-    def replay_outbox(
-        self, actor: Actor, event_id: str, *, correlation_id: str | None = None
-    ) -> OutboxEvent:
+    def replay_outbox(self, actor: Actor, event_id: str, *, correlation_id: str | None = None) -> OutboxEvent:
         correlation_id = correlation_id or new_id("corr")
-        event = self.store.outbox_event(event_id)
-        if event is None:
-            raise NotFoundError("outbox event not found")
-        self.resolve_tenant_context(actor, event.tenant)
-        self._authorize(
-            actor, action="outbox.replay", resource_type="user-engine:outbox-event",
-            resource_id=event_id, tenant=event.tenant, correlation_id=correlation_id,
-        )
-        replayed = replace(
-            event, claimed_by=None, claimed_at=None, failed_at=None,
-            failure_reason=None, dead_lettered_at=None,
-        )
-        with self.store.transaction():
-            self.store.save_outbox(replayed)
-        return replayed
+        with self.store.outbox_delivery_guard(event_id):
+            event = self.store.outbox_event(event_id)
+            if event is None: raise NotFoundError("outbox event not found")
+            self.resolve_tenant_context(actor, event.tenant)
+            self._authorize(actor, action="outbox.replay", resource_type="user-engine:outbox-event",
+                resource_id=event_id, tenant=event.tenant, correlation_id=correlation_id)
+            if event.delivered_at or (event.claimed_by and not event.failed_at):
+                raise ConflictError("Delivery is completed or its previous attempt is unresolved.")
+            replayed = replace(event, claimed_by=None, claimed_at=None, failed_at=None,
+                failure_reason=None, dead_lettered_at=None)
+            with self.store.transaction(): self.store.save_outbox(replayed)
+            return replayed
 
     def outbox_diagnostics(self) -> OutboxDiagnostics:
         event_types: dict[str, int] = {}