feat(decisioning): wire TaskHandoff finalize path — single-ledger via on_complete hook (#538-impl)

The HITL finalize path that the v1.5 design promised but the previous
commits left as a follow-up. Adopters returning ctx.handoff_to_task(...)
from finalize_proposal now get the expected behaviour: Submitted
envelope to the buyer immediately, handoff fn runs in the background,
proposal commits to the store on completion. Single ledger per § D3.

Mechanism: _project_handoff gains an optional on_complete callback that
runs with the typed result before registry.complete.
maybe_intercept_finalize builds a closure capturing proposal_store +
proposal_id and threads it through. If the handoff fn raises,
on_complete raises, or store.commit raises, the framework calls
registry.fail and the proposal stays DRAFT — no half-committed state.

5 new E2E tests cover: happy-path commit-on-completion; handoff path
log emission; AdcpError from handoff fn; wrong return type; commit
failure under transient store error.

3957 passed, 32 skipped (was 3952). Zero regressions.

Author: bokelley

examples/sales_proposal_mode_seller/src/proposal_manager.py

@@ -239,8 +239,10 @@ async def finalize_proposal(
         from the draft; this method lock-prices and returns. Framework
         commits via ``proposal_store.commit`` after projection.
 
-        For HITL flows, return ``ctx.handoff_to_task(...)`` instead;
-        that's a follow-up for v1.5+.
+        For HITL flows, return ``ctx.handoff_to_task(...)`` instead.
+        The framework projects ``Submitted`` immediately, runs the
+        handoff fn in the background, and commits the proposal to the
+        store on completion (single-ledger guarantee per § D3).
         """
         del ctx
         committed_payload = dict(req.proposal_payload)

src/adcp/decisioning/dispatch.py

@@ -66,6 +66,8 @@
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
     from pydantic import BaseModel
 
     from adcp.decisioning.accounts import AccountStore
@@ -1216,6 +1218,7 @@ async def _project_handoff(
     method_name: str,
     registry: TaskRegistry,
     executor: ThreadPoolExecutor,
+    on_complete: Callable[[Any], Awaitable[None]] | None = None,
 ) -> dict[str, Any]:
     """Promote a TaskHandoff to a background task.
 
@@ -1230,9 +1233,11 @@ async def _project_handoff(
        ``contextvars.copy_context()`` snapshot. ``create_task``
        inherits the snapshot for free; ``run_in_executor`` doesn't,
        hence the explicit copy.
-    3. The background task awaits the handoff fn's return; on success
-       calls ``registry.complete(task_id, result.model_dump() if
-       Pydantic else result)``; on :class:`AdcpError` calls
+    3. The background task awaits the handoff fn's return. On success,
+       if ``on_complete`` is provided, the framework awaits it with the
+       typed result before persisting. Then ``registry.complete(task_id,
+       result.model_dump() if Pydantic else result)``. On
+       :class:`AdcpError` from the handoff fn OR ``on_complete``, calls
        ``registry.fail(task_id, error.to_wire())``; on any other
        exception, wraps to ``INTERNAL_ERROR`` and calls
        ``registry.fail``.
@@ -1244,6 +1249,17 @@ async def _project_handoff(
         etc.) — used as ``task_type`` on the registry row so
         ``tasks/get`` round-trips correctly.
 
+    :param on_complete: Optional framework hook invoked with the typed
+        result of the handoff fn before ``registry.complete``. Used by
+        the proposal-finalize lifecycle to commit the proposal to
+        :class:`ProposalStore` exactly once when the HITL approval
+        lands. If the hook raises, the framework treats it like a
+        handoff fn failure: ``registry.fail`` is called with the wrapped
+        error and ``registry.complete`` is NOT called. This is what
+        gives the v1.5 single-ledger guarantee its teeth — a commit
+        failure cannot leave the task in 'submitted' forever or land
+        the proposal in a half-committed state.
+
     The handoff fn is extracted via the type-identity dispatch in
     :func:`adcp.decisioning.types.is_task_handoff`. Subclassed
     TaskHandoff instances (deliberate non-feature) silently take the
@@ -1292,6 +1308,38 @@ async def _run() -> None:
             await registry.fail(task_id, wrapped.to_wire())
             return
 
+        # Framework completion hook (e.g., proposal_store.commit for
+        # finalize). Runs with the TYPED result before model_dump so
+        # the closure can pull typed fields (.expires_at, .proposal)
+        # off it directly. Failures here are treated identically to a
+        # handoff fn failure: registry.fail, no registry.complete. This
+        # is the load-bearing seam for the v1.5 single-ledger D3
+        # guarantee — if the proposal_store.commit raises, the proposal
+        # stays DRAFT and the task lands in 'failed', so the buyer's
+        # tasks/get returns the failure rather than a phantom success.
+        if on_complete is not None:
+            try:
+                await on_complete(result)
+            except AdcpError as exc:
+                await registry.fail(task_id, exc.to_wire())
+                return
+            except Exception as exc:
+                logger.exception(
+                    "Unhandled exception in on_complete hook for task %s — wrapping",
+                    task_id,
+                )
+                wrapped = AdcpError(
+                    "INTERNAL_ERROR",
+                    message=(
+                        f"Post-completion hook for {method_name!r} raised "
+                        f"{type(exc).__name__}; see details for cause"
+                    ),
+                    recovery="terminal",
+                    details=_internal_error_details(exc),
+                )
+                await registry.fail(task_id, wrapped.to_wire())
+                return
+
         # Persist terminal artifact. Pydantic responses get
         # ``model_dump()``; dict responses pass through.
         #

src/adcp/decisioning/handler.py

@@ -1189,6 +1189,7 @@ async def get_products(  # type: ignore[override]
                 params,
                 ctx,
                 executor=self._executor,
+                registry=self._registry,
             )
             if finalize_response is not None:
                 return cast("GetProductsResponse", finalize_response)

src/adcp/decisioning/proposal_dispatch.py

@@ -80,6 +80,7 @@
     from adcp.decisioning.context import RequestContext
     from adcp.decisioning.proposal_manager import ProposalManager
     from adcp.decisioning.proposal_store import ProposalStore
+    from adcp.decisioning.task_registry import TaskRegistry
 
 logger = logging.getLogger("adcp.decisioning.proposal_dispatch")
 
@@ -146,6 +147,7 @@ async def maybe_intercept_finalize(
     ctx: RequestContext[Any],
     *,
     executor: ThreadPoolExecutor,
+    registry: TaskRegistry,
 ) -> dict[str, Any] | None:
     """Intercept ``buying_mode='refine' + action='finalize'`` requests.
 
@@ -243,29 +245,52 @@ async def maybe_intercept_finalize(
         )
 
     if is_task_handoff(result):
-        # HITL slow path. Per § D2: framework projects Submitted; the
-        # handoff fn completes via the standard TaskRegistry flow. The
-        # framework wraps the handoff so the proposal commit happens at
-        # task-completion time — single ledger, no race.
-        #
-        # NOTE: TaskHandoff projection lives in dispatch._project_handoff.
-        # Returning the handoff back through the standard path requires
-        # threading more state than the seam-helper signature carries
-        # today. v1.5 lands the inline path; the handoff path lands as a
-        # follow-up once the dispatch surface exposes a hook. Adopters
-        # who declare finalize=True + return TaskHandoff today will see
-        # their handoff projected as Submitted but the framework won't
-        # auto-commit on completion — the handoff body must call
-        # store.commit explicitly, or fail closed.
-        raise AdcpError(
-            "INTERNAL_ERROR",
-            message=(
-                "TaskHandoff[FinalizeProposalSuccess] return path is not "
-                "wired in v1.5 — the inline FinalizeProposalSuccess path "
-                "is the supported route. File a follow-up issue if you "
-                "need HITL finalize."
-            ),
-            recovery="terminal",
+        # HITL slow path. Per § D2 + § D3: framework projects Submitted
+        # immediately; the handoff fn completes via the standard
+        # TaskRegistry flow. The framework wires the proposal_store
+        # commit as the on_complete hook so the SAME asyncio task that
+        # calls registry.complete also calls store.commit — single
+        # ledger, no race. If either fails, registry.fail is called and
+        # the proposal stays DRAFT (handler at dispatch._project_handoff
+        # treats on_complete failures identically to handoff fn
+        # failures).
+        from adcp.decisioning.dispatch import _project_handoff
+
+        async def _commit_on_handoff_completion(success: Any) -> None:
+            # Type-check the handoff fn's return shape. Adopter mistakes
+            # here surface as INTERNAL_ERROR on tasks/get rather than
+            # silently corrupting state.
+            if not isinstance(success, FinalizeProposalSuccess):
+                raise AdcpError(
+                    "INTERNAL_ERROR",
+                    message=(
+                        f"finalize_proposal handoff fn returned "
+                        f"{type(success).__name__}; expected "
+                        "FinalizeProposalSuccess."
+                    ),
+                    recovery="terminal",
+                )
+            await _await_maybe(
+                store.commit(
+                    proposal_id,
+                    expires_at=success.expires_at,
+                    proposal_payload=dict(success.proposal),
+                )
+            )
+            finalize_succeeded_log(
+                proposal_id=proposal_id,
+                account_id=account_id,
+                expires_at=success.expires_at,
+                path="handoff",
+            )
+
+        return await _project_handoff(
+            result,
+            ctx,
+            method_name="finalize_proposal",
+            registry=registry,
+            executor=executor,
+            on_complete=_commit_on_handoff_completion,
         )
 
     if not isinstance(result, FinalizeProposalSuccess):