refactor(message): canonical Outcome sum type on OutboxInnerMessage

CLAUDE.md

@@ -30,7 +30,7 @@ The package wires a FastStream `Broker`/`Registrator`/`Subscriber` trio whose tr
 - **Relay to foreign broker** — native FastStream publisher-chain (`@kafka_pub @broker_outbox.subscriber("q")`); bad chain composition raises `_OutboxConfigError` and retries via **lease expiry**, not the retry strategy. → [relay.md](architecture/relay.md)
 - **Timers** — `activate_in`/`activate_at` are mutually exclusive and gate eligibility; `timer_id` is "at most one *live* row per `(queue, timer_id)`", not a global idempotency key; `cancel_timer`'s `acquired_token IS NULL` guard is load-bearing. → [timers.md](architecture/timers.md)
 - **User-owned schema** — caller owns the table; the three partial indexes + the `<table>_lease_ck` CHECK are load-bearing; **no `state` column**; `validate_schema()` is opt-in. → [schema.md](architecture/schema.md)
-- **Opt-in DLQ** — with `dlq_table=None` every path is **bit-for-bit identical**; the `DLQFailureReason` `Literal` is the **public contract** (changes are API-breaking); branch on `terminal_failure_reason` **before** `last_exception`. → [dlq.md](architecture/dlq.md)
+- **Opt-in DLQ** — with `dlq_table=None` every path is **bit-for-bit identical**; the `DLQFailureReason` `Literal` is the **public contract** (changes are API-breaking); delivery records one disjoint `Outcome` (`Ack`/`Retry`/`Terminal`) that `dispatch_one` matches on (`terminal_failure_reason`/`pending_delay_seconds`/`state_set` are read-only views). → [dlq.md](architecture/dlq.md)
 - **Two-loop subscriber + lease-token invariant** — fetch loop + N worker loops; **every terminal write filters on `acquired_token`** (a stale write finds `rowcount==0` and is dropped) — any new fetch/terminal path must preserve this; `AckPolicy.ACK_FIRST` is rejected at registration; `lease_ttl_seconds` must exceed handler P99. → [subscriber.md](architecture/subscriber.md)
 - **Drain on stop** — custom `stop()` overrides + the `dispatch_one` shutdown-race guard prevent row leaks on rolling deploys; **re-check both overrides when touching shutdown**. → [drain.md](architecture/drain.md)
 - **Test broker** — `TestOutboxBroker` swaps in `FakeOutboxClient`; sync mode dispatches the handler before `publish` returns; `test_client_contract.py` pins real-vs-fake parity. → [test-broker.md](architecture/test-broker.md)

architecture/dlq.md

@@ -12,15 +12,15 @@ User-facing: `docs/usage/dlq.md`. Invariant summary: `CLAUDE.md` § Opt-in DLQ.
 
 The CTE's `RETURNING` / `INSERT` / `SELECT` column lists are not hand-written — they derive from `_DLQ_PROJECTION` (the `(outbox_col, dlq_col)` pairs copied verbatim) plus `_DLQ_INJECTED_COLUMNS` (`failure_reason`, `last_exception`, supplied by the caller) in `schema.py`. The fake (`FakeOutboxClient.delete_with_lease`) builds its audit dict from the same constants, so the two substrates can't drift on which columns the archive carries — a DLQ column change is one edit in `schema.py`, verified across both adapters by `tests/test_client_contract.py`. `failed_at` is not in the projection; it rides the DLQ column's `server_default`.
 
-## `terminal_failure_reason` routing
+## Terminal-outcome routing
 
-`OutboxInnerMessage.terminal_failure_reason` is set on the three failure paths:
+`OutboxInnerMessage.outcome` records a single canonical `Outcome` (`message.py`): `Ack` (success), `Retry(delay_seconds)`, or `Terminal(reason)`. The three failure paths record a `Terminal`:
 
-- `allow_delivery` False → `"max_deliveries"`
-- `_nack` strategy-exhausted → `"retry_terminal"`
-- `_reject` → `"rejected"`
+- `allow_delivery` False → `Terminal("max_deliveries")`
+- `_nack` strategy-exhausted → `Terminal("retry_terminal")`
+- `_reject` → `Terminal("rejected")`
 
-`_flush_terminal` reads it to decide whether to build a DLQ payload; `dispatch_one` also reads it to pick the `nacked_terminal(reason=…)` tag value. **Branch on `terminal_failure_reason` BEFORE `last_exception`**, so manual `await msg.reject()` (no exception raised) routes correctly to `nacked_terminal(reason="rejected")` instead of the previously-incorrect `acked`. Success (`_ack`) leaves the field None; success rows never touch the DLQ.
+`terminal_failure_reason` is now a **read-only view** of `outcome` (`Terminal.reason`, else `None`), so existing readers are unchanged: `_flush_terminal` reads it to decide whether to build a DLQ payload; the `nacked_terminal(reason=…)` tag value comes from the same reason. `dispatch_one` **matches on the disjoint `Outcome` variant** — `Terminal → nacked_terminal`, `Retry → reschedule`, `Ack → delete`. Because the variants are mutually exclusive there is no ordering dependence between the arms: a manual `await msg.reject()` (no exception raised) records `Terminal("rejected")` directly, independent of `last_exception`. Success (`_ack` → `Ack`) leaves `terminal_failure_reason` None; success rows never touch the DLQ.
 
 **The `DLQFailureReason` `Literal` type (`message.py`) is the public contract** for this string — operator queries and dashboard labels key off these values, so changing them is API-breaking.
 

faststream_outbox/message.py

@@ -36,6 +36,32 @@
 DLQFailureReason = Literal["max_deliveries", "retry_terminal", "rejected"]
 
 
+# Canonical delivery outcome. ``ack``/``nack``/``reject`` record exactly one of these
+# on ``OutboxInnerMessage.outcome``; the worker loop matches on the variant to pick the
+# DB write. The legacy ``state_set``/``terminal_failure_reason``/``pending_delay_seconds``
+# attributes are read-only views derived from it.
+@dataclass(frozen=True, slots=True)
+class Ack:
+    """Handler succeeded; the row is deleted (no DLQ)."""
+
+
+@dataclass(frozen=True, slots=True)
+class Retry:
+    """Handler failed; the strategy scheduled another attempt after ``delay_seconds``."""
+
+    delay_seconds: float
+
+
+@dataclass(frozen=True, slots=True)
+class Terminal:
+    """Delivery failed terminally; the row is deleted (and DLQ'd when configured)."""
+
+    reason: DLQFailureReason
+
+
+Outcome = Ack | Retry | Terminal
+
+
 # Header keys the outbox envelope owns on a row's ``headers`` dict. ``_encode_payload``
 # (``envelope.py``) writes them, ``OutboxParser.parse_message`` (``parser.py``) reads
 # them back, and the relay header-propagation (``_maybe_propagate_inbound_headers``)
@@ -50,9 +76,11 @@
 class OutboxInnerMessage:
     """In-memory copy of a claimed outbox row, plus ack/nack/reject intent helpers.
 
-    The ack/nack/reject methods set in-memory intent flags (``terminal_failure_reason``,
-    ``pending_delay_seconds``). The worker loop reads those flags and issues the
-    actual DB write, scoped by ``acquired_token``.
+    The ack/nack/reject methods record a single canonical ``Outcome``
+    (``Ack``/``Retry``/``Terminal``) on ``outcome``. The worker loop matches on the
+    variant and issues the actual DB write, scoped by ``acquired_token``.
+    ``state_set``/``terminal_failure_reason``/``pending_delay_seconds`` are read-only
+    views of ``outcome`` kept for existing readers.
     """
 
     id: int
@@ -75,16 +103,32 @@ class OutboxInnerMessage:
     retry_strategy: "RetryStrategyProto | None" = None
     last_exception: BaseException | None = None
 
-    state_set: bool = field(default=False, init=False)
-    # Set by ``_nack`` when the strategy schedules a retry; consumed by the
-    # subscriber's ``_flush_retry`` to drive ``mark_pending_with_lease``.
-    pending_delay_seconds: float | None = field(default=None, init=False)
-    # Set on terminal-failure paths (``allow_delivery`` False, ``_nack`` exhausted,
-    # ``_reject``). ``_flush_terminal`` reads it to decide whether to build a DLQ
-    # payload; ``dispatch_one`` reads it to pick the ``nacked_terminal`` reason
-    # tag. Stays ``None`` on the success (``_ack``) path so handler-success
-    # never touches the DLQ.
-    terminal_failure_reason: "DLQFailureReason | None" = field(default=None, init=False)
+    # Canonical delivery outcome, set exactly once by ``_ack``/``_nack``/``_reject``/
+    # ``allow_delivery``. ``None`` until intent is recorded. The three properties below
+    # derive the legacy views the worker loop and metrics read.
+    outcome: "Outcome | None" = field(default=None, init=False)
+
+    @property
+    def state_set(self) -> bool:
+        return self.outcome is not None
+
+    @property
+    def terminal_failure_reason(self) -> "DLQFailureReason | None":
+        """Terminal reason (``allow_delivery`` False, ``_nack`` exhausted, ``_reject``), else ``None``.
+
+        ``_flush_terminal`` reads it to decide whether to build a DLQ payload;
+        ``dispatch_one`` reads it to pick the ``nacked_terminal`` reason tag. ``None``
+        on the ``Ack``/``Retry`` paths so handler-success never touches the DLQ.
+        """
+        return self.outcome.reason if isinstance(self.outcome, Terminal) else None
+
+    @property
+    def pending_delay_seconds(self) -> float | None:
+        """Retry delay set by ``_nack`` when the strategy schedules a retry, else ``None``.
+
+        Consumed by ``_flush_retry`` to drive ``mark_pending_with_lease``.
+        """
+        return self.outcome.delay_seconds if isinstance(self.outcome, Retry) else None
 
     # ``**_options`` are accepted-and-ignored: FastStream's AcknowledgementMiddleware
     # forwards ``message.nack(**extra_options)`` for native idioms like
@@ -103,13 +147,13 @@ async def reject(self, **_options: Any) -> None:
         await self._update_state_if_not_set(self._reject)
 
     async def _update_state_if_not_set(self, fn: Callable[[], Awaitable[None]]) -> None:
-        if self.state_set:
+        if self.outcome is not None:
             return
         await fn()
-        self.state_set = True
 
     async def _ack(self) -> None:
         self._record_attempt()
+        self.outcome = Ack()
 
     async def _nack(self) -> None:
         self._record_attempt()
@@ -135,16 +179,16 @@ async def _nack(self) -> None:
                     self.retry_strategy,
                     self,
                 )
-                self.terminal_failure_reason = "retry_terminal"
+                self.outcome = Terminal("retry_terminal")
                 return
         if delay is None:
-            self.terminal_failure_reason = "retry_terminal"
+            self.outcome = Terminal("retry_terminal")
         else:
-            self.pending_delay_seconds = delay
+            self.outcome = Retry(delay)
 
     async def _reject(self) -> None:
         self._record_attempt()
-        self.terminal_failure_reason = "rejected"
+        self.outcome = Terminal("rejected")
 
     def _record_attempt(self) -> None:
         self.attempts_count += 1
@@ -156,8 +200,7 @@ def _record_attempt(self) -> None:
     def allow_delivery(self, *, max_deliveries: int | None, logger: "LoggerProto | None") -> bool:
         """If ``max_deliveries`` is set and exceeded, mark for deletion without invoking the handler."""
         if max_deliveries is not None and self.deliveries_count > max_deliveries:
-            self.state_set = True
-            self.terminal_failure_reason = "max_deliveries"
+            self.outcome = Terminal("max_deliveries")
             if logger is not None:
                 logger.log(
                     logging.ERROR,

faststream_outbox/subscriber/usecase.py

@@ -39,7 +39,7 @@
 from faststream.specification.schema import Message, Operation, SubscriberSpec
 from typing_extensions import override
 
-from faststream_outbox.message import ENVELOPE_MANAGED_HEADERS, OutboxInnerMessage
+from faststream_outbox.message import ENVELOPE_MANAGED_HEADERS, OutboxInnerMessage, Retry, Terminal
 from faststream_outbox.parser.parser import OutboxParser
 from faststream_outbox.publisher.fake import OutboxFakePublisher
 from faststream_outbox.response import OutboxResponse
@@ -701,23 +701,23 @@ async def dispatch_one(  # linear pipeline: guard, consume, branch on outcome, f
         await row.assert_state_set(logger)
         duration_seconds = time.perf_counter() - start_perf
         common = {**base, "deliveries_count": row.deliveries_count, "duration_seconds": duration_seconds}
-        # Branch on ``terminal_failure_reason`` (set by ``_nack``/``_reject``) before
-        # ``last_exception`` so manual ``msg.reject()`` (no exception raised) still
-        # surfaces as ``nacked_terminal(reason="rejected")``, and ack-policy-driven
-        # rejects (REJECT_ON_ERROR) carry ``reason="rejected"`` rather than the
-        # incorrect ``"retry_terminal"`` they got under the old ``last_exception``-first
-        # ordering. Successful handlers leave ``terminal_failure_reason=None``.
-        if row.terminal_failure_reason is not None:
-            terminal_tags = self._with_exception_type({**common, "reason": row.terminal_failure_reason}, row)
-            event, tags = "nacked_terminal", terminal_tags
-        elif row.pending_delay_seconds is not None:
-            retry_tags = self._with_exception_type({**common, "next_delay_seconds": row.pending_delay_seconds}, row)
-            # Retry is a per-row UPDATE -- never batched. Flush inline, unchanged.
+        # Match on the disjoint ``Outcome`` variant recorded by ack/nack/reject:
+        # Terminal -> nacked_terminal (delete + DLQ), Retry -> reschedule, Ack -> delete.
+        # The variants are mutually exclusive, so there is no ordering dependence between
+        # the arms (e.g. a manual ``msg.reject()`` records ``Terminal("rejected")``
+        # directly, independent of ``last_exception``).
+        outcome = row.outcome
+        if isinstance(outcome, Terminal):
+            tags = self._with_exception_type({**common, "reason": outcome.reason}, row)
+            event = "nacked_terminal"
+        elif isinstance(outcome, Retry):
+            retry_tags = self._with_exception_type({**common, "next_delay_seconds": outcome.delay_seconds}, row)
+            # Retry is a per-row UPDATE -- never batched. Flush inline.
             # P17: emit only after the flush lands; a lease-lost update emits lease_lost instead.
             if await self._safe_flush(row, terminal=False, writer_conn=writer_conn):
                 self._emit_metric("nacked_retried", retry_tags)
             return False
-        else:
+        else:  # Ack (assert_state_set guaranteed outcome is set, so this is Ack, never None)
             event, tags = "acked", common
         # Terminal delete: batch it if batchable, else flush inline. P17 holds either way —
         # the metric fires only when the delete lands (inline here, or deferred in _flush_buffer).

planning/changes/2026-07-17.02-outcome-sum-type.md

@@ -0,0 +1,55 @@
+---
+summary: Replaced OutboxInnerMessage's terminal_failure_reason / pending_delay_seconds / state_set fields with one canonical Outcome sum type (Ack | Retry | Terminal); the old attributes become read-only views. Behavior-preserving; dissolves the terminal-before-retry ordering invariant.
+---
+
+# Change: Canonical Outcome sum type on OutboxInnerMessage
+
+**Lane:** lightweight. A behavior-preserving refactor of the load-bearing
+terminal/dispatch path that reshapes an internal representation and dissolves a
+documented ordering invariant — recorded here so the change is legible.
+
+## Goal
+
+Deepen `OutboxInnerMessage`: `ack`/`nack`/`reject`/`allow_delivery` recorded delivery
+intent across three loosely-coupled fields (`terminal_failure_reason`,
+`pending_delay_seconds`, `state_set`), where "success" was the implicit "all None"
+state and "reason and delay both set" was representable-but-never-used. Replace them
+with one canonical `Outcome` sum type so success is an explicit variant and the
+impossible combination is unrepresentable.
+
+## Approach (Option A — canonical outcome + derived views)
+
+- `message.py` defines `Ack` / `Retry(delay_seconds)` / `Terminal(reason)`
+  (frozen, slotted) and `Outcome = Ack | Retry | Terminal`. `OutboxInnerMessage.outcome`
+  is the single canonical field; the setters assign it (`_ack` → `Ack()`, `_nack` →
+  `Retry`/`Terminal("retry_terminal")`, `_reject` → `Terminal("rejected")`,
+  `allow_delivery` → `Terminal("max_deliveries")`).
+- `terminal_failure_reason` / `pending_delay_seconds` / `state_set` become **read-only
+  `@property` views** of `outcome` (zero-drift derivations, no setters). So the five
+  other subscriber readers (`_flush_terminal`, `_flush_retry`, `_terminal_has_dlq`, the
+  shutdown-race `state_set` guard, metric tags) and every read-assertion in the tests
+  are unchanged.
+- `dispatch_one` matches on the disjoint variant (`Terminal → nacked_terminal`,
+  `Retry → reschedule`, `Ack → delete`). Because the variants are mutually exclusive,
+  the **terminal-before-retry / before-`last_exception` ordering invariant is dissolved**
+  — recorded in `architecture/dlq.md` and the `CLAUDE.md` DLQ invariant line.
+
+The `DLQFailureReason` `Literal` (the public wire contract queried by operators) is
+unchanged — it is now `Terminal.reason`'s type. `dlq_table=None` stays bit-for-bit
+identical.
+
+## Files
+
+- `faststream_outbox/message.py` — `Outcome` types; `outcome` field + three property
+  views; setters wired to `outcome`.
+- `faststream_outbox/subscriber/usecase.py` — `dispatch_one` matches on `outcome`
+  (imports `Retry`, `Terminal`); no other reader changed.
+- `tests/test_unit.py` — the 12 sites that *wrote* the old fields now set `outcome`
+  (`Retry(...)` / `Terminal(...)`); read-assertions unchanged (properties).
+- `architecture/dlq.md`, `CLAUDE.md` — truth-home promotion of the routing model.
+
+## Verification
+
+- [x] `just test` — full suite green: 603 passed, Postgres 17, coverage 100.00%.
+- [x] `just lint` — clean (ruff + ty).
+- [x] `just check-planning` — passes.