Module adcp.decisioning.state

Sync workflow-state reader for :class:RequestContext.

Defines:

• :class:StateReader — Protocol for sync reads of framework-owned in-flight workflow state. Platform methods read this without re-querying their own DB; the framework owns the cache.
• :class:WorkflowStep, :class:WorkflowObjectType, :data:GovernanceContextJWS — framework-internal types referenced by :class:StateReader methods. Defined here (not in adcp.types.generated_poc/) because they're framework-only — not on the wire.
• :class:_NotYetWiredStateReader — v6.0 stub. Returns type-correct empty values; emits a one-time :class:UserWarning per method on first call so adopters notice they're reading uninitialized state. Backing store lands in v6.1.

The asymmetry between this stub (returns empty) and :class:adcp.decisioning.resolve._NotYetWiredResolver (raises) is deliberate. state.* reads are read-only inspections of framework-owned state — an empty workflow-steps list IS the correct answer for a fresh tenant. resolve.* fetches are validated lookups — an empty :class:PropertyList in v6.0 vs. a real one in v6.1 is divergence the framework cannot silently paper over. See docs/proposals/decisioning-platform-dispatch-design.md#d15 for the full rationale.

Global variables

var GovernanceContextJWS

JWS-signed governance context. The framework verifies signature, plan-binding, seller-binding, and phase-binding before exposing the token to platform code; adopters can trust the value. Don't unwrap or modify — re-pass to downstream framework calls instead.

var WorkflowObjectType

Object types a workflow step can touch. Framework-internal — not on the wire (the wire-side status-change-resource-type.json enum covers a different surface).

Classes

class Proposal (**data: Any)

Expand source code

class Proposal(AdCPBaseModel):
    model_config = ConfigDict(
        extra='allow',
    )
    proposal_id: Annotated[
        str,
        Field(
            description='Unique identifier for this proposal. Used to finalize a draft proposal and to execute a committed proposal via create_media_buy.',
            max_length=255,
        ),
    ]
    name: Annotated[
        str, Field(description='Human-readable name for this media plan proposal', max_length=500)
    ]
    description: Annotated[
        str | None,
        Field(
            description='Explanation of the proposal strategy and what it achieves', max_length=2000
        ),
    ] = None
    allocations: Annotated[
        list[product_allocation.ProductAllocation],
        Field(
            description='Budget allocations across products. Allocation percentages MUST sum to 100. Publishers are responsible for ensuring the sum equals 100; buyers SHOULD validate this before execution.',
            min_length=1,
        ),
    ]
    proposal_status: Annotated[
        proposal_status_1.ProposalStatus | None,
        Field(
            description="Lifecycle status of this proposal and the per-proposal source of truth for whether finalization is required before create_media_buy. When absent, the proposal is ready to buy (backward compatible). 'draft' means indicative pricing — finalize via refine before purchasing. 'committed' means firm pricing with inventory reserved until expires_at and executable via create_media_buy."
        ),
    ] = None
    expires_at: Annotated[
        AwareDatetime | None,
        Field(
            description='When this proposal expires and can no longer be executed. For draft proposals, indicates when indicative pricing becomes stale. For committed proposals, indicates when the inventory hold lapses — the buyer must call create_media_buy before this time.'
        ),
    ] = None
    insertion_order: Annotated[
        insertion_order_1.InsertionOrder | None,
        Field(
            description='Formal insertion order attached to a committed proposal. Present when the seller requires a signed agreement before the media buy can proceed. The buyer references the io_id in io_acceptance on create_media_buy.'
        ),
    ] = None
    total_budget_guidance: Annotated[
        TotalBudgetGuidance | None, Field(description='Optional budget guidance for this proposal')
    ] = None
    brief_alignment: Annotated[
        str | None,
        Field(
            description='Explanation of how this proposal aligns with the campaign brief',
            max_length=2000,
        ),
    ] = None
    forecast: Annotated[
        delivery_forecast.DeliveryForecast | None,
        Field(
            description='Aggregate forecasted delivery metrics for the entire proposal. When both proposal-level and allocation-level forecasts are present, the proposal-level forecast is authoritative for total delivery estimation.'
        ),
    ] = None
    ext: ext_1.ExtensionObject | None = None

Base model for AdCP types with spec-compliant serialization.

Defaults to extra='ignore' so unknown fields from newer spec versions are silently dropped rather than causing validation errors. Generated types whose schemas set additionalProperties: true override this with extra='allow' in their own model_config.

Set ADCP_STRICT_VALIDATION=1 in the environment ("1", "true", "yes", "on" are accepted) to flip the default to extra='forbid'. Use this during spec upgrades to catch silently-dropped renamed fields in tests. See :func:_resolve_extra_policy.

Important

The env var is resolved once at module import time. Set it in your shell or CI environment before import adcp runs — mutating os.environ["ADCP_STRICT_VALIDATION"] after the first adcp import has no effect on already-imported model classes (they captured the policy at class-body evaluation).

Consumers who want per-model strict validation can override model_config on their subclass.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• AdCPBaseModel
• pydantic.main.BaseModel

Class variables

var allocations : list[adcp.types.generated_poc.core.product_allocation.ProductAllocation]

var brief_alignment : str | None

var description : str | None

var expires_at : pydantic.types.AwareDatetime | None

var ext : adcp.types.generated_poc.core.ext.ExtensionObject | None

var forecast : adcp.types.generated_poc.core.delivery_forecast.DeliveryForecast | None

var insertion_order : adcp.types.generated_poc.core.insertion_order.InsertionOrder | None

var model_config

var name : str

var proposal_id : str

var proposal_status : adcp.types.generated_poc.enums.proposal_status.ProposalStatus | None

var total_budget_guidance : adcp.types.generated_poc.core.proposal.TotalBudgetGuidance | None

Inherited members

• AdCPBaseModel:
  • model_dump
  • model_dump_json
  • model_summary

class StateReader (*args, **kwargs)

Expand source code

@runtime_checkable
class StateReader(Protocol):
    """Sync reads of framework-owned in-flight workflow state.

    Platform methods read prior workflow context (recent media-buy
    transitions, related proposals, in-flight governance bindings)
    without re-querying their own DB. The framework owns the cache; the
    Protocol surface is purely read.

    Framework-supplied; never constructed by adopter code. The
    ``RequestContext.state`` field is populated by the dispatch
    hydration helper. Adopters substituting test doubles use
    :func:`dataclasses.replace` on the context, not direct
    construction.

    Mirrors the TS-side ``WorkflowStateReader`` interface in
    ``src/lib/server/decisioning/context.ts``. v6.0 ships the contract
    + the no-op stub; v6.1 lands the backing store.

    .. note::
       :class:`runtime_checkable` Protocols match by attribute *name*
       only — return types (including :data:`GovernanceContextJWS`,
       which is a :func:`typing.NewType` invisible at runtime) and
       method signatures are NOT enforced by ``isinstance``. A custom
       impl that returns ``int`` from ``governance_context()`` will
       pass the structural check; mypy is the only enforcement for
       return-type contracts. Coverage gap is acceptable for v6.0.
    """

    def find_by_object(
        self,
        object_type: WorkflowObjectType,
        object_id: str,
    ) -> Sequence[WorkflowStep]:
        """Return workflow steps that touched the given object,
        chronological. Used for "what's happened to this buy?" reads
        without a platform-side fetch."""
        ...

    def find_proposal_by_id(self, proposal_id: str) -> Proposal | None:
        """Resolve a ``proposal_id`` threaded across
        ``get_products → refine → create_media_buy`` without platform
        code. Returns ``None`` if the framework doesn't recognize the
        id."""
        ...

    def governance_context(self) -> GovernanceContextJWS | None:
        """Currently in-flight verified governance context (the JWS
        token). ``None`` for non-governance flows. Framework verifies
        before exposure; platform code can trust the value.

        Adopters claiming ``governance-*`` specialisms in
        ``capabilities.specialisms`` MUST set
        ``capabilities.governance_aware=True`` and wire a real
        ``StateReader`` that returns real JWS tokens. The default stub
        returns ``None``, which would silently skip the gate — server
        boot fails fast if a governance specialism is claimed without
        the opt-in. See
        ``docs/proposals/decisioning-platform-dispatch-design.md#d15``.
        """
        ...

    def workflow_steps(self) -> Sequence[WorkflowStep]:
        """All chronological steps for this request's account.
        Audit-read shape."""
        ...

Sync reads of framework-owned in-flight workflow state.

Platform methods read prior workflow context (recent media-buy transitions, related proposals, in-flight governance bindings) without re-querying their own DB. The framework owns the cache; the Protocol surface is purely read.

Framework-supplied; never constructed by adopter code. The RequestContext.state field is populated by the dispatch hydration helper. Adopters substituting test doubles use :func:dataclasses.replace on the context, not direct construction.

Mirrors the TS-side WorkflowStateReader interface in src/lib/server/decisioning/context.ts. v6.0 ships the contract + the no-op stub; v6.1 lands the backing store.

Note

:class:runtime_checkable Protocols match by attribute name only — return types (including :data:GovernanceContextJWS, which is a :func:typing.NewType invisible at runtime) and method signatures are NOT enforced by isinstance. A custom impl that returns int from governance_context() will pass the structural check; mypy is the only enforcement for return-type contracts. Coverage gap is acceptable for v6.0.

Ancestors

• typing.Protocol
• typing.Generic

Methods

def find_by_object(self,
object_type: WorkflowObjectType,
object_id: str) ‑> Sequence[WorkflowStep]

Expand source code

def find_by_object(
    self,
    object_type: WorkflowObjectType,
    object_id: str,
) -> Sequence[WorkflowStep]:
    """Return workflow steps that touched the given object,
    chronological. Used for "what's happened to this buy?" reads
    without a platform-side fetch."""
    ...

Return workflow steps that touched the given object, chronological. Used for "what's happened to this buy?" reads without a platform-side fetch.

def find_proposal_by_id(self, proposal_id: str) ‑> adcp.types.generated_poc.core.proposal.Proposal | None

Expand source code

def find_proposal_by_id(self, proposal_id: str) -> Proposal | None:
    """Resolve a ``proposal_id`` threaded across
    ``get_products → refine → create_media_buy`` without platform
    code. Returns ``None`` if the framework doesn't recognize the
    id."""
    ...

Resolve a proposal_id threaded across get_products → refine → create_media_buy without platform code. Returns None if the framework doesn't recognize the id.

def governance_context(self) ‑> GovernanceContextJWS | None

Expand source code

def governance_context(self) -> GovernanceContextJWS | None:
    """Currently in-flight verified governance context (the JWS
    token). ``None`` for non-governance flows. Framework verifies
    before exposure; platform code can trust the value.

    Adopters claiming ``governance-*`` specialisms in
    ``capabilities.specialisms`` MUST set
    ``capabilities.governance_aware=True`` and wire a real
    ``StateReader`` that returns real JWS tokens. The default stub
    returns ``None``, which would silently skip the gate — server
    boot fails fast if a governance specialism is claimed without
    the opt-in. See
    ``docs/proposals/decisioning-platform-dispatch-design.md#d15``.
    """
    ...

Currently in-flight verified governance context (the JWS token). None for non-governance flows. Framework verifies before exposure; platform code can trust the value.

Adopters claiming governance-* specialisms in capabilities.specialisms MUST set capabilities.governance_aware=True and wire a real StateReader that returns real JWS tokens. The default stub returns None, which would silently skip the gate — server boot fails fast if a governance specialism is claimed without the opt-in. See docs/proposals/decisioning-platform-dispatch-design.md#d15.

def workflow_steps(self) ‑> Sequence[WorkflowStep]

Expand source code

def workflow_steps(self) -> Sequence[WorkflowStep]:
    """All chronological steps for this request's account.
    Audit-read shape."""
    ...

All chronological steps for this request's account. Audit-read shape.

class WorkflowStep (id: str,
object_type: WorkflowObjectType,
object_id: str,
tool: str,
at: str,
actor: dict[str, str],
status: "Literal['submitted', 'completed', 'failed', 'progress']")

Expand source code

@dataclass(frozen=True)
class WorkflowStep:
    """A chronological event the framework recorded against an object.

    Frozen because the framework writes the step record once at the
    transition; platform code reads but does not mutate. The shape
    mirrors the TS-side ``WorkflowStep`` interface so cross-language
    adopters get the same fields.

    :param id: Stable step identifier (framework-allocated UUID).
    :param object_type: The object this step touched.
    :param object_id: Stable id of the touched object within
        :attr:`object_type`.
    :param tool: Wire verb that ran the step
        (e.g. ``'create_media_buy'``, ``'sync_creatives'``).
    :param at: ISO 8601 timestamp of the step.
    :param actor: Who initiated the step. ``agent_url`` for an agent
        principal, ``principal`` for a service-account principal,
        possibly both.
    :param status: Step outcome. ``'submitted'`` for a kicked-off task,
        ``'completed'``/``'failed'`` for terminal states,
        ``'progress'`` for a mid-flight update.
    """

    id: str
    object_type: WorkflowObjectType
    object_id: str
    tool: str
    at: str
    actor: dict[str, str]
    status: Literal["submitted", "completed", "failed", "progress"]

A chronological event the framework recorded against an object.

Frozen because the framework writes the step record once at the transition; platform code reads but does not mutate. The shape mirrors the TS-side WorkflowStep interface so cross-language adopters get the same fields.

:param id: Stable step identifier (framework-allocated UUID). :param object_type: The object this step touched. :param object_id: Stable id of the touched object within :attr:object_type. :param tool: Wire verb that ran the step (e.g. 'create_media_buy', 'sync_creatives'). :param at: ISO 8601 timestamp of the step. :param actor: Who initiated the step. agent_url for an agent principal, principal for a service-account principal, possibly both. :param status: Step outcome. 'submitted' for a kicked-off task, 'completed'/'failed' for terminal states, 'progress' for a mid-flight update.

Instance variables

var actor : dict[str, str]

var at : str

var id : str

var object_id : str

var object_type : Literal['media_buy', 'creative', 'product', 'plan', 'audience', 'rights_grant', 'task']

var status : Literal['submitted', 'completed', 'failed', 'progress']

var tool : str