Module adcp.decisioning.task_registry

Task registry for the DecisioningPlatform handoff path.

Defines:

• :class:TaskRegistry Protocol — the seam adopters substitute when they need a durable backing store (PostgreSQL, Redis, etc.). The Protocol shape is pinned with per-method contract docstrings; D7 of the dispatch design names every invariant.
• :class:InMemoryTaskRegistry — the v6.0 reference implementation. Process-local, lossy on restart. Suitable for local dev, CI, and test fixtures; production deployments running sales-broadcast-tv or any HITL flow refuse to start without an explicit opt-in (see :func:adcp.decisioning.serve.serve Stage 3 wiring).
• :class:TaskHandoffContext — what the framework passes into the adopter's handoff callable when ctx.handoff_to_task(fn) fires. Carries the framework-issued task id plus update(progress) and heartbeat() affordances.

The registry's storage shape is intentionally minimal: {task_id → TaskRecord} keyed by the framework-allocated UUID. Cross-tenant access control is enforced via the optional expected_account_id argument on :meth:TaskRegistry.get() — sellers threading ctx.account.id through to tasks/get get a None return on mismatch (no principal-enumeration via task_id probing).

Production-mode gate (Emma #8 / round-4): :func:adcp.decisioning.serve.serve reads ADCP_ENV (case-insensitive {"prod", "production"} — same as :func:adcp.validation.client_hooks._default_response_mode) and refuses to wire :class:InMemoryTaskRegistry in production unless ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1 is set. Sales-broadcast-tv adopters are structurally forced into the HITL path which depends on the registry — silent in-memory fallback is a real prod foot-gun.

Global variables

var TaskState

Terminal task states per AdCP 3.0 spec (enums/task-status.json). submitted = task created but not yet started; working = adopter callback running; completed / failed = terminal.

Classes

class InMemoryTaskRegistry

Expand source code

class InMemoryTaskRegistry:
    """Process-local task registry — v6.0 reference implementation.

    Storage is a plain ``dict[str, TaskRecord]`` guarded by an
    :class:`asyncio.Lock`. Adequate for local dev, CI, and test
    fixtures; production deployments wire a durable counterpart
    (PostgreSQL, Redis, etc.) implementing the same :class:`TaskRegistry`
    Protocol.

    Production-mode gate: :func:`adcp.decisioning.serve.serve` refuses
    to wire this when ``ADCP_ENV`` indicates production unless
    ``ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1`` is set. The gate
    reads ``registry.is_durable``; subclassing this class for
    instrumentation does NOT bypass the gate (the ``False`` is
    inherited). Custom durable impls set ``is_durable = True``
    explicitly. Production sellers running ``sales-broadcast-tv``
    or any HITL flow get the explicit refusal so silent in-memory
    fallback can't bite oncall.
    """

    is_durable: ClassVar[bool] = False

    def __init__(self) -> None:
        self._records: dict[str, TaskRecord] = {}
        self._lock = asyncio.Lock()

    async def issue(
        self,
        *,
        account_id: str,
        task_type: str,
        request_context: dict[str, Any] | None = None,
        **_extra: Any,
    ) -> str:
        # Forward-compat: log unrecognized kwargs at DEBUG so adopters
        # who haven't yet upgraded notice when they're missing new
        # framework fields. Don't raise — that would break adopters
        # the moment a new version ships.
        if _extra:
            logger.debug(
                "InMemoryTaskRegistry.issue ignoring unrecognized kwargs: %s",
                list(_extra.keys()),
            )
        # Reject empty/unset account_id at issue-time. Without this,
        # two tenants whose AccountStore returns Account(id="") or the
        # default Account(id="<unset>") share a cache scope class and
        # can read each other's tasks via cross-tenant probe (the
        # equality check passes when both are empty). See
        # tests/test_decisioning_task_registry_cross_tenant.py for the
        # regression suite.
        if not account_id or not account_id.strip() or account_id == "<unset>":
            raise ValueError(
                f"account_id must be a non-empty, non-default string; "
                f"got {account_id!r}. AccountStore.resolve must always "
                "return Account(id=<non-empty>) so cross-tenant cache "
                "scoping works correctly."
            )
        task_id = f"task_{uuid.uuid4().hex[:16]}"
        async with self._lock:
            self._records[task_id] = TaskRecord(
                task_id=task_id,
                account_id=account_id,
                state="submitted",
                task_type=task_type,
                request_context=(dict(request_context) if request_context is not None else None),
            )
        return task_id

    async def update_progress(
        self,
        task_id: str,
        progress: dict[str, Any],
    ) -> None:
        async with self._lock:
            record = self._records.get(task_id)
            if record is None:
                # Silent no-op — the dispatch wrapper expects this method
                # to never raise on transient lookup failure (see Protocol
                # docstring).
                return
            if record.state in ("completed", "failed"):
                # Terminal-state guard: a late progress update from a
                # straggler coroutine MUST NOT mutate a finalized record
                # — it would resurrect "working" appearance against
                # ``tasks/get`` reads that already saw the terminal
                # state. Log + drop is the safe choice (the dispatch
                # wrapper is expected to swallow update failures
                # anyway).
                logger.warning(
                    "InMemoryTaskRegistry.update_progress(task_id=%s) "
                    "dropped: task is already in terminal state %r",
                    task_id,
                    record.state,
                )
                return
            record.progress = dict(progress)
            if record.state == "submitted":
                record.state = "working"
            record.updated_at = time.time()

    async def complete(
        self,
        task_id: str,
        result: dict[str, Any],
    ) -> None:
        # Defense-in-depth credential strip at the persistence boundary.
        # The dispatcher's TaskHandoff path also strips before calling
        # complete() (see ``_project_handoff`` in dispatch.py); the
        # WorkflowHandoff path does NOT (the adopter's external
        # workflow calls ``registry.complete`` directly, off the
        # framework's call stack). Stripping here closes that gap and
        # protects custom registry consumers that bypass the dispatcher
        # entirely. The strip is method-gated by ``record.task_type``
        # (the wire verb name persisted at ``issue()``) and idempotent
        # on already-stripped payloads, so the dispatcher-side double
        # strip is a no-op.
        from adcp.decisioning.account_projection import strip_credentials_from_wire_result

        async with self._lock:
            record = self._records.get(task_id)
            if record is None:
                raise ValueError(f"Task {task_id!r} not found")
            stripped = strip_credentials_from_wire_result(record.task_type, result)
            if record.state == "completed":
                if record.result == stripped:
                    return  # idempotent
                raise ValueError(f"Task {task_id!r} already completed with a different result")
            record.state = "completed"
            record.result = dict(stripped) if isinstance(stripped, dict) else stripped
            record.updated_at = time.time()

    async def fail(
        self,
        task_id: str,
        error: dict[str, Any],
    ) -> None:
        async with self._lock:
            record = self._records.get(task_id)
            if record is None:
                raise ValueError(f"Task {task_id!r} not found")
            if record.state == "failed":
                if record.error == error:
                    return  # idempotent
                raise ValueError(f"Task {task_id!r} already failed with a different error")
            record.state = "failed"
            record.error = dict(error)
            record.updated_at = time.time()

    async def get(
        self,
        task_id: str,
        *,
        expected_account_id: str | None = None,
    ) -> dict[str, Any] | None:
        async with self._lock:
            record = self._records.get(task_id)
            if record is None:
                return None
            if expected_account_id is not None and record.account_id != expected_account_id:
                # Cross-tenant probe — return None, NOT raw record.
                # Critical security boundary: returning the record
                # here enables principal-enumeration via task_id
                # probing. The dispatch path that calls this method
                # always passes the authenticated principal's
                # account_id; adopter impls implementing this Protocol
                # MUST preserve this behavior.
                return None
            return record.to_dict()

    async def discard(self, task_id: str) -> None:
        async with self._lock:
            # Idempotent: pop with default. The Protocol contract
            # tolerates discarding an unknown id (no raise) so the
            # WorkflowHandoff projection's rollback can be unconditional.
            self._records.pop(task_id, None)

Process-local task registry — v6.0 reference implementation.

Storage is a plain dict[str, TaskRecord] guarded by an :class:asyncio.Lock. Adequate for local dev, CI, and test fixtures; production deployments wire a durable counterpart (PostgreSQL, Redis, etc.) implementing the same :class:TaskRegistry Protocol.

Production-mode gate: :func:adcp.decisioning.serve.serve refuses to wire this when ADCP_ENV indicates production unless ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1 is set. The gate reads registry.is_durable; subclassing this class for instrumentation does NOT bypass the gate (the False is inherited). Custom durable impls set is_durable = True explicitly. Production sellers running sales-broadcast-tv or any HITL flow get the explicit refusal so silent in-memory fallback can't bite oncall.

Class variables

var is_durable : ClassVar[bool]

Methods

async def complete(self, task_id: str, result: dict[str, Any]) ‑> None

Expand source code

async def complete(
    self,
    task_id: str,
    result: dict[str, Any],
) -> None:
    # Defense-in-depth credential strip at the persistence boundary.
    # The dispatcher's TaskHandoff path also strips before calling
    # complete() (see ``_project_handoff`` in dispatch.py); the
    # WorkflowHandoff path does NOT (the adopter's external
    # workflow calls ``registry.complete`` directly, off the
    # framework's call stack). Stripping here closes that gap and
    # protects custom registry consumers that bypass the dispatcher
    # entirely. The strip is method-gated by ``record.task_type``
    # (the wire verb name persisted at ``issue()``) and idempotent
    # on already-stripped payloads, so the dispatcher-side double
    # strip is a no-op.
    from adcp.decisioning.account_projection import strip_credentials_from_wire_result

    async with self._lock:
        record = self._records.get(task_id)
        if record is None:
            raise ValueError(f"Task {task_id!r} not found")
        stripped = strip_credentials_from_wire_result(record.task_type, result)
        if record.state == "completed":
            if record.result == stripped:
                return  # idempotent
            raise ValueError(f"Task {task_id!r} already completed with a different result")
        record.state = "completed"
        record.result = dict(stripped) if isinstance(stripped, dict) else stripped
        record.updated_at = time.time()

async def discard(self, task_id: str) ‑> None

Expand source code

async def discard(self, task_id: str) -> None:
    async with self._lock:
        # Idempotent: pop with default. The Protocol contract
        # tolerates discarding an unknown id (no raise) so the
        # WorkflowHandoff projection's rollback can be unconditional.
        self._records.pop(task_id, None)

async def fail(self, task_id: str, error: dict[str, Any]) ‑> None

Expand source code

async def fail(
    self,
    task_id: str,
    error: dict[str, Any],
) -> None:
    async with self._lock:
        record = self._records.get(task_id)
        if record is None:
            raise ValueError(f"Task {task_id!r} not found")
        if record.state == "failed":
            if record.error == error:
                return  # idempotent
            raise ValueError(f"Task {task_id!r} already failed with a different error")
        record.state = "failed"
        record.error = dict(error)
        record.updated_at = time.time()

async def get(self, task_id: str, *, expected_account_id: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

async def get(
    self,
    task_id: str,
    *,
    expected_account_id: str | None = None,
) -> dict[str, Any] | None:
    async with self._lock:
        record = self._records.get(task_id)
        if record is None:
            return None
        if expected_account_id is not None and record.account_id != expected_account_id:
            # Cross-tenant probe — return None, NOT raw record.
            # Critical security boundary: returning the record
            # here enables principal-enumeration via task_id
            # probing. The dispatch path that calls this method
            # always passes the authenticated principal's
            # account_id; adopter impls implementing this Protocol
            # MUST preserve this behavior.
            return None
        return record.to_dict()

async def issue(self,
*,
account_id: str,
task_type: str,
request_context: dict[str, Any] | None = None,
**_extra: Any) ‑> str

Expand source code

async def issue(
    self,
    *,
    account_id: str,
    task_type: str,
    request_context: dict[str, Any] | None = None,
    **_extra: Any,
) -> str:
    # Forward-compat: log unrecognized kwargs at DEBUG so adopters
    # who haven't yet upgraded notice when they're missing new
    # framework fields. Don't raise — that would break adopters
    # the moment a new version ships.
    if _extra:
        logger.debug(
            "InMemoryTaskRegistry.issue ignoring unrecognized kwargs: %s",
            list(_extra.keys()),
        )
    # Reject empty/unset account_id at issue-time. Without this,
    # two tenants whose AccountStore returns Account(id="") or the
    # default Account(id="<unset>") share a cache scope class and
    # can read each other's tasks via cross-tenant probe (the
    # equality check passes when both are empty). See
    # tests/test_decisioning_task_registry_cross_tenant.py for the
    # regression suite.
    if not account_id or not account_id.strip() or account_id == "<unset>":
        raise ValueError(
            f"account_id must be a non-empty, non-default string; "
            f"got {account_id!r}. AccountStore.resolve must always "
            "return Account(id=<non-empty>) so cross-tenant cache "
            "scoping works correctly."
        )
    task_id = f"task_{uuid.uuid4().hex[:16]}"
    async with self._lock:
        self._records[task_id] = TaskRecord(
            task_id=task_id,
            account_id=account_id,
            state="submitted",
            task_type=task_type,
            request_context=(dict(request_context) if request_context is not None else None),
        )
    return task_id

async def update_progress(self, task_id: str, progress: dict[str, Any]) ‑> None

Expand source code

async def update_progress(
    self,
    task_id: str,
    progress: dict[str, Any],
) -> None:
    async with self._lock:
        record = self._records.get(task_id)
        if record is None:
            # Silent no-op — the dispatch wrapper expects this method
            # to never raise on transient lookup failure (see Protocol
            # docstring).
            return
        if record.state in ("completed", "failed"):
            # Terminal-state guard: a late progress update from a
            # straggler coroutine MUST NOT mutate a finalized record
            # — it would resurrect "working" appearance against
            # ``tasks/get`` reads that already saw the terminal
            # state. Log + drop is the safe choice (the dispatch
            # wrapper is expected to swallow update failures
            # anyway).
            logger.warning(
                "InMemoryTaskRegistry.update_progress(task_id=%s) "
                "dropped: task is already in terminal state %r",
                task_id,
                record.state,
            )
            return
        record.progress = dict(progress)
        if record.state == "submitted":
            record.state = "working"
        record.updated_at = time.time()

class TaskHandoffContext (id: str,
_registry: TaskRegistry)

Expand source code

@dataclass
class TaskHandoffContext:
    """Per-task context passed to the handoff fn registered via
    :meth:`adcp.decisioning.RequestContext.handoff_to_task`.

    Adopter pattern::

        def create_media_buy(self, req, ctx):
            if self._needs_review(req):
                return ctx.handoff_to_task(self._async_review)

            return CreateMediaBuySuccess(media_buy_id="mb_1", ...)

        async def _async_review(self, task_ctx: TaskHandoffContext):
            await task_ctx.update({"message": "Trafficker reviewing"})
            decision = await self._wait_for_trafficker(task_ctx.id)
            return CreateMediaBuySuccess(media_buy_id=decision.id, ...)

    The framework allocates ``task_ctx.id`` BEFORE invoking the
    handoff fn so the adopter can persist the id to its own backend
    (storyboard runner row, Slack thread reference, etc.) before
    kicking off slow work. This fixes a documented v1 ergonomics bug
    where adopters could only learn the task_id AFTER returning.

    Constructed by :func:`adcp.decisioning.dispatch._build_handoff_context`;
    never instantiated by adopter code.
    """

    id: str
    _registry: TaskRegistry
    _heartbeat_impl: Callable[[], Awaitable[None]] = field(default_factory=lambda: _noop_heartbeat)

    async def update(self, progress: dict[str, Any]) -> None:
        """Write a progress payload. Transitions ``submitted`` →
        ``working`` on first call.

        Errors are swallowed (logged at WARNING with traceback):
        a transient registry write failure must not abort the handoff.
        Buyer-facing impact is a missed progress event, not a failed
        task. Adopters who need delivery guarantees plug a durable
        registry; the warning surfaces the transient via existing
        observability hooks so silent loss isn't truly invisible.
        """
        try:
            await self._registry.update_progress(self.id, progress)
        except Exception:
            logger.warning(
                "TaskHandoffContext.update(task_id=%s) suppressed "
                "registry transient — progress event lost; handoff "
                "continues",
                self.id,
                exc_info=True,
            )
            return

    async def heartbeat(self) -> None:
        """Liveness signal for operator infrastructure. v6.1 stub.

        v6.0 ships as a no-op so adopter code calling
        ``await task_ctx.heartbeat()`` future-proofs against the
        eventual implementation. Operator-side TTL-reset wiring lands
        with the durable registry impl.
        """
        await self._heartbeat_impl()

Per-task context passed to the handoff fn registered via :meth:RequestContext.handoff_to_task().

Adopter pattern::

def create_media_buy(self, req, ctx):
    if self._needs_review(req):
        return ctx.handoff_to_task(self._async_review)

    return CreateMediaBuySuccess(media_buy_id="mb_1", ...)

async def _async_review(self, task_ctx: TaskHandoffContext):
    await task_ctx.update({"message": "Trafficker reviewing"})
    decision = await self._wait_for_trafficker(task_ctx.id)
    return CreateMediaBuySuccess(media_buy_id=decision.id, ...)

The framework allocates task_ctx.id BEFORE invoking the handoff fn so the adopter can persist the id to its own backend (storyboard runner row, Slack thread reference, etc.) before kicking off slow work. This fixes a documented v1 ergonomics bug where adopters could only learn the task_id AFTER returning.

Constructed by :func:adcp.decisioning.dispatch._build_handoff_context; never instantiated by adopter code.

Instance variables

var id : str

Methods

async def heartbeat(self) ‑> None

Expand source code

async def heartbeat(self) -> None:
    """Liveness signal for operator infrastructure. v6.1 stub.

    v6.0 ships as a no-op so adopter code calling
    ``await task_ctx.heartbeat()`` future-proofs against the
    eventual implementation. Operator-side TTL-reset wiring lands
    with the durable registry impl.
    """
    await self._heartbeat_impl()

Liveness signal for operator infrastructure. v6.1 stub.

v6.0 ships as a no-op so adopter code calling await task_ctx.heartbeat() future-proofs against the eventual implementation. Operator-side TTL-reset wiring lands with the durable registry impl.

async def update(self, progress: dict[str, Any]) ‑> None

Expand source code

async def update(self, progress: dict[str, Any]) -> None:
    """Write a progress payload. Transitions ``submitted`` →
    ``working`` on first call.

    Errors are swallowed (logged at WARNING with traceback):
    a transient registry write failure must not abort the handoff.
    Buyer-facing impact is a missed progress event, not a failed
    task. Adopters who need delivery guarantees plug a durable
    registry; the warning surfaces the transient via existing
    observability hooks so silent loss isn't truly invisible.
    """
    try:
        await self._registry.update_progress(self.id, progress)
    except Exception:
        logger.warning(
            "TaskHandoffContext.update(task_id=%s) suppressed "
            "registry transient — progress event lost; handoff "
            "continues",
            self.id,
            exc_info=True,
        )
        return

Write a progress payload. Transitions submitted → working on first call.

Errors are swallowed (logged at WARNING with traceback): a transient registry write failure must not abort the handoff. Buyer-facing impact is a missed progress event, not a failed task. Adopters who need delivery guarantees plug a durable registry; the warning surfaces the transient via existing observability hooks so silent loss isn't truly invisible.

class TaskRecord (task_id: str,
account_id: str,
state: TaskState,
task_type: str,
progress: dict[str, Any] | None = None,
result: dict[str, Any] | None = None,
error: dict[str, Any] | None = None,
request_context: dict[str, Any] | None = None,
created_at: float = <factory>,
updated_at: float = <factory>)

Expand source code

@dataclass
class TaskRecord:
    """The framework's per-task storage row.

    Internal to the registry impl — adopters don't construct these.
    The Protocol surface returns dicts on :meth:`TaskRegistry.get`
    rather than the dataclass directly so the storage shape stays
    swappable (a Postgres impl might return a different row class).

    :param task_id: Framework-allocated UUID. Stable across the
        task's lifetime.
    :param account_id: Account that owns the task. Used for the
        cross-tenant access-control check in :meth:`TaskRegistry.get`.
    :param state: Terminal state lifecycle. Transitions are
        framework-driven; adopters drive completion via
        :meth:`TaskHandoffContext.update` and the dispatcher calls
        :meth:`TaskRegistry.complete` / :meth:`TaskRegistry.fail` at
        the end of the handoff fn.
    :param task_type: Wire-spec task type (``'create_media_buy'``,
        ``'sync_creatives'``, etc.). Stored on the registry record so
        ``tasks/get`` can return it on the response payload; NOT part
        of the synchronous Submitted envelope (per
        ``schemas/cache/core/protocol-envelope.json``).
    :param progress: Latest progress payload written by
        :meth:`TaskHandoffContext.update`. Buyers see this on
        ``tasks/get`` while the task is in the ``working`` state.
    :param result: Terminal artifact set by :meth:`TaskRegistry.complete`.
        MUST be the JSON-serialized spec response shape (e.g. a
        ``CreateMediaBuySuccessResponse`` projected through
        ``model_dump()``). v6.1 adds size enforcement; for now the
        registry trusts adopters.
    :param error: Terminal failure payload set by
        :meth:`TaskRegistry.fail`. MUST be the
        :meth:`AdcpError.to_wire` shape so ``tasks/get`` returns the
        spec ``adcp_error`` envelope verbatim.
    :param created_at: Monotonic creation timestamp (Unix epoch
        seconds). Adopters get the exact value the framework stored;
        useful for SLA dashboards.
    :param updated_at: Last-touched timestamp. Updated on every state
        transition AND every :meth:`TaskHandoffContext.update` call.
    """

    task_id: str
    account_id: str
    state: TaskState
    task_type: str
    progress: dict[str, Any] | None = None
    result: dict[str, Any] | None = None
    error: dict[str, Any] | None = None
    request_context: dict[str, Any] | None = None
    """Buyer-supplied ``context`` extension from the request that
    issued this task. Echoed to ``tasks/get`` responses at the
    top-level ``context`` field per
    ``schemas/cache/core/tasks_get_response.json`` (sibling of
    ``result`` / ``error``). Captured at ``issue()`` time and
    immutable afterwards — terminal-state transitions
    (``complete`` / ``fail``) MUST NOT touch this field. Closes #563.
    """
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)

    def to_dict(self) -> dict[str, Any]:
        """Serialize for buyer consumption via ``tasks/get``.

        Adopters or middleware reading the dict shape get the exact
        wire-relevant fields. ``created_at`` / ``updated_at`` are
        included so admin tooling can build SLA reports.

        ``context`` lands at the top level — sibling of ``result``
        and ``error`` — matching the spec's
        ``TasksGetResponse.context`` placement (#563).
        """
        out: dict[str, Any] = {
            "task_id": self.task_id,
            "account_id": self.account_id,
            "state": self.state,
            "task_type": self.task_type,
            "progress": self.progress,
            "result": self.result,
            "error": self.error,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }
        if self.request_context is not None:
            out["context"] = self.request_context
        return out

The framework's per-task storage row.

Internal to the registry impl — adopters don't construct these. The Protocol surface returns dicts on :meth:TaskRegistry.get() rather than the dataclass directly so the storage shape stays swappable (a Postgres impl might return a different row class).

:param task_id: Framework-allocated UUID. Stable across the task's lifetime. :param account_id: Account that owns the task. Used for the cross-tenant access-control check in :meth:TaskRegistry.get(). :param state: Terminal state lifecycle. Transitions are framework-driven; adopters drive completion via :meth:TaskHandoffContext.update() and the dispatcher calls :meth:TaskRegistry.complete() / :meth:TaskRegistry.fail() at the end of the handoff fn. :param task_type: Wire-spec task type ('create_media_buy', 'sync_creatives', etc.). Stored on the registry record so tasks/get can return it on the response payload; NOT part of the synchronous Submitted envelope (per schemas/cache/core/protocol-envelope.json). :param progress: Latest progress payload written by :meth:TaskHandoffContext.update(). Buyers see this on tasks/get while the task is in the working state. :param result: Terminal artifact set by :meth:TaskRegistry.complete(). MUST be the JSON-serialized spec response shape (e.g. a CreateMediaBuySuccessResponse projected through model_dump()). v6.1 adds size enforcement; for now the registry trusts adopters. :param error: Terminal failure payload set by :meth:TaskRegistry.fail(). MUST be the :meth:AdcpError.to_wire shape so tasks/get returns the spec adcp_error envelope verbatim. :param created_at: Monotonic creation timestamp (Unix epoch seconds). Adopters get the exact value the framework stored; useful for SLA dashboards. :param updated_at: Last-touched timestamp. Updated on every state transition AND every :meth:TaskHandoffContext.update() call.

Instance variables

var account_id : str

var created_at : float

var error : dict[str, typing.Any] | None

var progress : dict[str, typing.Any] | None

var request_context : dict[str, typing.Any] | None

Buyer-supplied context extension from the request that issued this task. Echoed to tasks/get responses at the top-level context field per schemas/cache/core/tasks_get_response.json (sibling of result / error). Captured at issue() time and immutable afterwards — terminal-state transitions (complete / fail) MUST NOT touch this field. Closes #563.

var result : dict[str, typing.Any] | None

var state : Literal['submitted', 'working', 'completed', 'failed']

var task_id : str

var task_type : str

var updated_at : float

Methods

def to_dict(self) ‑> dict[str, typing.Any]

Expand source code

def to_dict(self) -> dict[str, Any]:
    """Serialize for buyer consumption via ``tasks/get``.

    Adopters or middleware reading the dict shape get the exact
    wire-relevant fields. ``created_at`` / ``updated_at`` are
    included so admin tooling can build SLA reports.

    ``context`` lands at the top level — sibling of ``result``
    and ``error`` — matching the spec's
    ``TasksGetResponse.context`` placement (#563).
    """
    out: dict[str, Any] = {
        "task_id": self.task_id,
        "account_id": self.account_id,
        "state": self.state,
        "task_type": self.task_type,
        "progress": self.progress,
        "result": self.result,
        "error": self.error,
        "created_at": self.created_at,
        "updated_at": self.updated_at,
    }
    if self.request_context is not None:
        out["context"] = self.request_context
    return out

Serialize for buyer consumption via tasks/get.

Adopters or middleware reading the dict shape get the exact wire-relevant fields. created_at / updated_at are included so admin tooling can build SLA reports.

context lands at the top level — sibling of result and error — matching the spec's TasksGetResponse.context placement (#563).

class TaskRegistry (*args, **kwargs)

Expand source code

@runtime_checkable
class TaskRegistry(Protocol):
    """Per-account task store — the seam adopters substitute for a
    durable backing implementation.

    **Durability marker** (``is_durable: ClassVar[bool]``):

    Production deployments running ``sales-broadcast-tv`` or any HITL
    flow refuse to start with a non-durable registry unless the
    operator explicitly opts in via
    ``ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1``. The framework reads
    ``registry.is_durable`` to make this decision; subclassing
    :class:`InMemoryTaskRegistry` for instrumentation does NOT bypass
    the gate (the subclass inherits ``is_durable = False``). Custom
    durable impls MUST set ``is_durable = True`` explicitly. The
    Protocol declares this as a class-level ``bool``.

    Lifecycle (framework-driven; adopters call only :meth:`TaskHandoffContext`
    methods, not these directly):

    1. Dispatch detects ``ctx.handoff_to_task(fn)`` returned from a
       platform method. Allocates a task_id and calls :meth:`issue` to
       persist the ``submitted`` row.
    2. Dispatch projects the wire ``Submitted`` envelope to the buyer.
    3. Dispatch runs ``fn(task_handoff_ctx)`` in the background. The
       adopter calls ``task_handoff_ctx.update(progress)`` zero or
       more times; the framework routes each to :meth:`update_progress`
       (also transitions ``submitted`` → ``working`` on first update).
    4. When ``fn`` returns, dispatch calls :meth:`complete` with the
       terminal artifact (a JSON-serialized spec response).
    5. When ``fn`` raises :class:`adcp.decisioning.AdcpError` (or any
       exception, wrapped to ``INTERNAL_ERROR``), dispatch calls
       :meth:`fail` with the wire-shaped error payload.

    All write paths set ``updated_at = now``. The registry is
    expected to be safe for concurrent reads; concurrent writes to
    the same task are serialized by the dispatcher (one ``fn`` per
    handoff, no concurrent `update_progress`/`complete` against the
    same task_id).

    Cross-tenant safety: every read MUST be account-scoped. The
    :meth:`get` method takes an optional ``expected_account_id`` —
    when supplied (the wire ``tasks/get`` path always supplies it),
    a mismatch returns ``None``, NOT the raw record. Adopters
    implementing custom registries MUST honor this: returning a
    cross-tenant record on probe enables principal-enumeration via
    task_id guessing. See
    ``tests/test_decisioning_task_registry_cross_tenant.py`` for
    the regression suite.

    **``account_id`` is opaque.** The framework threads
    ``ctx.account.id`` (whatever the adopter's
    :class:`~adcp.decisioning.AccountStore.resolve` returned) into
    every method. The registry MUST NOT parse it or re-derive tenant
    scope from it. Multi-tenant adopters encode their tenant scope
    into ``Account.id`` once at the
    :class:`~adcp.decisioning.AccountStore` layer; the registry then
    gets a globally-unique scope key and the cross-tenant probe
    check above degenerates to a simple equality. See the
    AccountStore docstring's "Multi-tenant deployments" section for
    the canonical encoding pattern.
    """

    #: Whether this registry persists tasks across process restarts.
    #: ``False`` for in-memory / lossy impls; ``True`` for durable
    #: backings (PostgreSQL, Redis, etc.). The framework's
    #: production-mode gate refuses non-durable registries unless
    #: the operator explicitly opts in via
    #: ``ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1``.
    is_durable: ClassVar[bool]

    async def issue(
        self,
        *,
        account_id: str,
        task_type: str,
        request_context: dict[str, Any] | None = None,
        **_extra: Any,
    ) -> str:
        """Allocate a fresh task_id, persist a ``submitted`` row, and
        return the id.

        :param account_id: Account that owns the task. Drives the
            cross-tenant access check on subsequent reads.
        :param task_type: Wire-spec task type (``'create_media_buy'``,
            etc.). Persisted on the row and surfaced on ``tasks/get``
            reads; NOT included in the synchronous Submitted envelope
            (per ``schemas/cache/core/protocol-envelope.json``).
        :param request_context: Buyer-supplied ``context`` extension
            from the request that issued this task. Persisted on the
            row and surfaced at the top level of ``tasks/get``
            responses (sibling of ``result`` / ``error``) so buyers
            can correlate polled task state with the kick-off
            request. ``None`` when the request carried no context
            field; the framework supplies it from the original
            request params. Adopters writing custom registries SHOULD
            store and surface this field; older registry impls that
            ignore it are functionally compatible (no echo on
            ``tasks/get`` reads, identical to pre-#563 behavior).
        :param _extra: Forward-compat slot for kwargs added by future
            framework versions. Custom registry impls MUST include
            ``**_extra: Any`` on their ``issue()`` signature so the
            framework can introduce new optional kwargs without
            breaking adopters who haven't yet adopted the new field.
            Implementations that don't recognize an extra kwarg
            should silently ignore it (the framework only relies on
            kwargs the Protocol explicitly declares). Logging the
            unrecognized keys at DEBUG level is encouraged so
            adopters notice when they've fallen behind.
        :returns: The framework-allocated task_id (string UUID).
        """
        ...

    async def update_progress(
        self,
        task_id: str,
        progress: dict[str, Any],
    ) -> None:
        """Write a progress payload and transition ``submitted`` →
        ``working`` on first call. No-op transition on subsequent
        calls (already in ``working``).

        Errors here are swallowed by the dispatch wrapper — a transient
        registry write failure must NOT abort the adopter's background
        handoff. Buyer-facing impact is a missed progress event, not a
        failed task. Adopter impls of this method that need durability
        guarantees should buffer + retry internally.
        """
        ...

    async def complete(
        self,
        task_id: str,
        result: dict[str, Any],
    ) -> None:
        """Mark the task ``completed`` with ``result`` as the terminal
        artifact.

        ``result`` MUST be the JSON-serialized spec response shape
        (e.g. ``CreateMediaBuySuccessResponse`` via ``model_dump()``).
        Idempotent on repeated calls with equal ``result``;
        non-idempotent re-completion with different result raises
        ``ValueError``.
        """
        ...

    async def fail(
        self,
        task_id: str,
        error: dict[str, Any],
    ) -> None:
        """Mark the task ``failed`` with ``error`` as the terminal
        wire-shaped error payload.

        ``error`` MUST be the :meth:`AdcpError.to_wire` shape so
        ``tasks/get`` round-trips the spec ``adcp_error`` envelope
        verbatim. Idempotent on repeated calls with equal ``error``.
        """
        ...

    async def get(
        self,
        task_id: str,
        *,
        expected_account_id: str | None = None,
    ) -> dict[str, Any] | None:
        """Look up a task record. Cross-tenant probes return ``None``.

        :param task_id: Framework-allocated id from a prior :meth:`issue`.
        :param expected_account_id: When supplied, the registry MUST
            return ``None`` if the stored record's ``account_id`` does
            not match. The wire ``tasks/get`` path always supplies the
            authenticated principal's account_id so adopters can't
            probe across tenants.
        :returns: The record dict (per :meth:`TaskRecord.to_dict`) or
            ``None`` if the id is unknown OR a cross-tenant mismatch.
        """
        ...

    async def discard(self, task_id: str) -> None:
        """Remove a task_id from the registry — rollback path.

        Used by the WorkflowHandoff dispatch projection
        (:func:`adcp.decisioning.dispatch._project_workflow_handoff`)
        when the adopter's enqueue fn raises after the task_id has
        been allocated. Without rollback, the buyer would receive a
        Submitted envelope referencing an orphan task_id their
        external workflow never registered.

        Idempotent: discarding an unknown task_id is a no-op (no
        raise). The discard window is tightly scoped — between
        ``issue()`` and the framework's projection step, with the
        adopter's enqueue fn in between. In practice this is a few
        milliseconds.

        Adopters MUST NOT call ``discard`` on a task that has
        progressed past ``submitted`` — that's the wrong recovery
        path; use ``fail()`` instead.
        """
        ...

Per-account task store — the seam adopters substitute for a durable backing implementation.

Durability marker (is_durable: ClassVar[bool]):

Production deployments running sales-broadcast-tv or any HITL flow refuse to start with a non-durable registry unless the operator explicitly opts in via ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1. The framework reads registry.is_durable to make this decision; subclassing :class:InMemoryTaskRegistry for instrumentation does NOT bypass the gate (the subclass inherits is_durable = False). Custom durable impls MUST set is_durable = True explicitly. The Protocol declares this as a class-level bool.

Lifecycle (framework-driven; adopters call only :meth:TaskHandoffContext methods, not these directly):

1. Dispatch detects ctx.handoff_to_task(fn) returned from a platform method. Allocates a task_id and calls :meth:issue to persist the submitted row.
2. Dispatch projects the wire Submitted envelope to the buyer.
3. Dispatch runs fn(task_handoff_ctx) in the background. The adopter calls task_handoff_ctx.update(progress) zero or more times; the framework routes each to :meth:update_progress (also transitions submitted → working on first update).
4. When fn returns, dispatch calls :meth:complete with the terminal artifact (a JSON-serialized spec response).
5. When fn raises :class:AdcpError (or any exception, wrapped to INTERNAL_ERROR), dispatch calls :meth:fail with the wire-shaped error payload.

All write paths set updated_at = now. The registry is expected to be safe for concurrent reads; concurrent writes to the same task are serialized by the dispatcher (one fn per handoff, no concurrent update_progress/complete against the same task_id).

Cross-tenant safety: every read MUST be account-scoped. The :meth:get method takes an optional expected_account_id — when supplied (the wire tasks/get path always supplies it), a mismatch returns None, NOT the raw record. Adopters implementing custom registries MUST honor this: returning a cross-tenant record on probe enables principal-enumeration via task_id guessing. See tests/test_decisioning_task_registry_cross_tenant.py for the regression suite.

account_id is opaque. The framework threads ctx.account.id (whatever the adopter's :class:~adcp.decisioning.AccountStore.resolve returned) into every method. The registry MUST NOT parse it or re-derive tenant scope from it. Multi-tenant adopters encode their tenant scope into Account.id once at the :class:~adcp.decisioning.AccountStore layer; the registry then gets a globally-unique scope key and the cross-tenant probe check above degenerates to a simple equality. See the AccountStore docstring's "Multi-tenant deployments" section for the canonical encoding pattern.

Ancestors

• typing.Protocol
• typing.Generic

Class variables

var is_durable : ClassVar[bool]

Whether this registry persists tasks across process restarts. False for in-memory / lossy impls; True for durable backings (PostgreSQL, Redis, etc.). The framework's production-mode gate refuses non-durable registries unless the operator explicitly opts in via ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1.

Methods

async def complete(self, task_id: str, result: dict[str, Any]) ‑> None

Expand source code

async def complete(
    self,
    task_id: str,
    result: dict[str, Any],
) -> None:
    """Mark the task ``completed`` with ``result`` as the terminal
    artifact.

    ``result`` MUST be the JSON-serialized spec response shape
    (e.g. ``CreateMediaBuySuccessResponse`` via ``model_dump()``).
    Idempotent on repeated calls with equal ``result``;
    non-idempotent re-completion with different result raises
    ``ValueError``.
    """
    ...

Mark the task completed with result as the terminal artifact.

result MUST be the JSON-serialized spec response shape (e.g. CreateMediaBuySuccessResponse via model_dump()). Idempotent on repeated calls with equal result; non-idempotent re-completion with different result raises ValueError.

async def discard(self, task_id: str) ‑> None

Expand source code

async def discard(self, task_id: str) -> None:
    """Remove a task_id from the registry — rollback path.

    Used by the WorkflowHandoff dispatch projection
    (:func:`adcp.decisioning.dispatch._project_workflow_handoff`)
    when the adopter's enqueue fn raises after the task_id has
    been allocated. Without rollback, the buyer would receive a
    Submitted envelope referencing an orphan task_id their
    external workflow never registered.

    Idempotent: discarding an unknown task_id is a no-op (no
    raise). The discard window is tightly scoped — between
    ``issue()`` and the framework's projection step, with the
    adopter's enqueue fn in between. In practice this is a few
    milliseconds.

    Adopters MUST NOT call ``discard`` on a task that has
    progressed past ``submitted`` — that's the wrong recovery
    path; use ``fail()`` instead.
    """
    ...

Remove a task_id from the registry — rollback path.

Used by the WorkflowHandoff dispatch projection (:func:adcp.decisioning.dispatch._project_workflow_handoff) when the adopter's enqueue fn raises after the task_id has been allocated. Without rollback, the buyer would receive a Submitted envelope referencing an orphan task_id their external workflow never registered.

Idempotent: discarding an unknown task_id is a no-op (no raise). The discard window is tightly scoped — between issue() and the framework's projection step, with the adopter's enqueue fn in between. In practice this is a few milliseconds.

Adopters MUST NOT call discard on a task that has progressed past submitted — that's the wrong recovery path; use fail() instead.

async def fail(self, task_id: str, error: dict[str, Any]) ‑> None

Expand source code

async def fail(
    self,
    task_id: str,
    error: dict[str, Any],
) -> None:
    """Mark the task ``failed`` with ``error`` as the terminal
    wire-shaped error payload.

    ``error`` MUST be the :meth:`AdcpError.to_wire` shape so
    ``tasks/get`` round-trips the spec ``adcp_error`` envelope
    verbatim. Idempotent on repeated calls with equal ``error``.
    """
    ...

Mark the task failed with error as the terminal wire-shaped error payload.

error MUST be the :meth:AdcpError.to_wire shape so tasks/get round-trips the spec adcp_error envelope verbatim. Idempotent on repeated calls with equal error.

async def get(self, task_id: str, *, expected_account_id: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

async def get(
    self,
    task_id: str,
    *,
    expected_account_id: str | None = None,
) -> dict[str, Any] | None:
    """Look up a task record. Cross-tenant probes return ``None``.

    :param task_id: Framework-allocated id from a prior :meth:`issue`.
    :param expected_account_id: When supplied, the registry MUST
        return ``None`` if the stored record's ``account_id`` does
        not match. The wire ``tasks/get`` path always supplies the
        authenticated principal's account_id so adopters can't
        probe across tenants.
    :returns: The record dict (per :meth:`TaskRecord.to_dict`) or
        ``None`` if the id is unknown OR a cross-tenant mismatch.
    """
    ...

Look up a task record. Cross-tenant probes return None.

:param task_id: Framework-allocated id from a prior :meth:issue. :param expected_account_id: When supplied, the registry MUST return None if the stored record's account_id does not match. The wire tasks/get path always supplies the authenticated principal's account_id so adopters can't probe across tenants. :returns: The record dict (per :meth:TaskRecord.to_dict()) or None if the id is unknown OR a cross-tenant mismatch.

async def issue(self,
*,
account_id: str,
task_type: str,
request_context: dict[str, Any] | None = None,
**_extra: Any) ‑> str

Expand source code

async def issue(
    self,
    *,
    account_id: str,
    task_type: str,
    request_context: dict[str, Any] | None = None,
    **_extra: Any,
) -> str:
    """Allocate a fresh task_id, persist a ``submitted`` row, and
    return the id.

    :param account_id: Account that owns the task. Drives the
        cross-tenant access check on subsequent reads.
    :param task_type: Wire-spec task type (``'create_media_buy'``,
        etc.). Persisted on the row and surfaced on ``tasks/get``
        reads; NOT included in the synchronous Submitted envelope
        (per ``schemas/cache/core/protocol-envelope.json``).
    :param request_context: Buyer-supplied ``context`` extension
        from the request that issued this task. Persisted on the
        row and surfaced at the top level of ``tasks/get``
        responses (sibling of ``result`` / ``error``) so buyers
        can correlate polled task state with the kick-off
        request. ``None`` when the request carried no context
        field; the framework supplies it from the original
        request params. Adopters writing custom registries SHOULD
        store and surface this field; older registry impls that
        ignore it are functionally compatible (no echo on
        ``tasks/get`` reads, identical to pre-#563 behavior).
    :param _extra: Forward-compat slot for kwargs added by future
        framework versions. Custom registry impls MUST include
        ``**_extra: Any`` on their ``issue()`` signature so the
        framework can introduce new optional kwargs without
        breaking adopters who haven't yet adopted the new field.
        Implementations that don't recognize an extra kwarg
        should silently ignore it (the framework only relies on
        kwargs the Protocol explicitly declares). Logging the
        unrecognized keys at DEBUG level is encouraged so
        adopters notice when they've fallen behind.
    :returns: The framework-allocated task_id (string UUID).
    """
    ...

Allocate a fresh task_id, persist a submitted row, and return the id.

:param account_id: Account that owns the task. Drives the cross-tenant access check on subsequent reads. :param task_type: Wire-spec task type ('create_media_buy', etc.). Persisted on the row and surfaced on tasks/get reads; NOT included in the synchronous Submitted envelope (per schemas/cache/core/protocol-envelope.json). :param request_context: Buyer-supplied context extension from the request that issued this task. Persisted on the row and surfaced at the top level of tasks/get responses (sibling of result / error) so buyers can correlate polled task state with the kick-off request. None when the request carried no context field; the framework supplies it from the original request params. Adopters writing custom registries SHOULD store and surface this field; older registry impls that ignore it are functionally compatible (no echo on tasks/get reads, identical to pre-#563 behavior). :param _extra: Forward-compat slot for kwargs added by future framework versions. Custom registry impls MUST include **_extra: Any on their issue() signature so the framework can introduce new optional kwargs without breaking adopters who haven't yet adopted the new field. Implementations that don't recognize an extra kwarg should silently ignore it (the framework only relies on kwargs the Protocol explicitly declares). Logging the unrecognized keys at DEBUG level is encouraged so adopters notice when they've fallen behind. :returns: The framework-allocated task_id (string UUID).

async def update_progress(self, task_id: str, progress: dict[str, Any]) ‑> None

Expand source code

async def update_progress(
    self,
    task_id: str,
    progress: dict[str, Any],
) -> None:
    """Write a progress payload and transition ``submitted`` →
    ``working`` on first call. No-op transition on subsequent
    calls (already in ``working``).

    Errors here are swallowed by the dispatch wrapper — a transient
    registry write failure must NOT abort the adopter's background
    handoff. Buyer-facing impact is a missed progress event, not a
    failed task. Adopter impls of this method that need durability
    guarantees should buffer + retry internally.
    """
    ...

Write a progress payload and transition submitted → working on first call. No-op transition on subsequent calls (already in working).

Errors here are swallowed by the dispatch wrapper — a transient registry write failure must NOT abort the adopter's background handoff. Buyer-facing impact is a missed progress event, not a failed task. Adopter impls of this method that need durability guarantees should buffer + retry internally.