From e38261515f0d59249bd95b68275add19ac9eb4fb Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 2 Oct 2026 17:56:31 +0000 Subject: [PATCH 1/3] feat(decisioning): observe committed task lifecycle transitions --- docs/task-lifecycle-observers.md | 37 ++++ src/adcp/decisioning/__init__.py | 4 + src/adcp/decisioning/pg/task_registry.py | 107 +++++++---- src/adcp/decisioning/task_registry.py | 90 +++++++++- tests/test_decisioning_task_lifecycle.py | 166 ++++++++++++++++++ tests/type_checks/task_lifecycle_observers.py | 39 ++++ 6 files changed, 404 insertions(+), 39 deletions(-) create mode 100644 docs/task-lifecycle-observers.md create mode 100644 tests/test_decisioning_task_lifecycle.py create mode 100644 tests/type_checks/task_lifecycle_observers.py diff --git a/docs/task-lifecycle-observers.md b/docs/task-lifecycle-observers.md new file mode 100644 index 000000000..66ed5b9c9 --- /dev/null +++ b/docs/task-lifecycle-observers.md @@ -0,0 +1,37 @@ +# Task lifecycle metrics + +`InMemoryTaskRegistry` and `PgTaskRegistry` offer additive lifecycle hooks. The +minimal `TaskRegistry` protocol and all existing write-method return values stay +unchanged. Register a synchronous callback with `add_lifecycle_observer` and +remove it with `remove_lifecycle_observer` (returns whether removal succeeded). +Registering the same callback twice is a no-op. + +```python +from adcp.decisioning import InMemoryTaskRegistry, TaskTransition + +registry = InMemoryTaskRegistry() + +def record_transition(event: TaskTransition, *, task_id: str, account_id: str, + task_type: str, created_at: float, updated_at: float) -> None: + print(event, task_type, updated_at - created_at) + +registry.add_lifecycle_observer(record_transition) +``` + +Events are `submitted`, the first `working` transition, `completed`, `failed`, +and `discarded`. Repeated progress writes, idempotent terminal retries, unknown +progress/discard requests and failed transactions do not emit events. Terminal +states cannot be changed into another terminal state; memory and PostgreSQL +registries enforce the same rule, including concurrent complete/fail calls. + +Callbacks run after the PostgreSQL transaction and connection contexts exit, or +after the memory registry lock is released. Each event contains a snapshot of +its transition metadata with epoch timestamps; discard uses deletion time for +`updated_at`. Registration is thread-safe, and delivery uses a snapshot of the +registered callbacks. Callback failures are logged and swallowed without +preventing remaining callbacks or changing the write result. + +These hooks are best-effort metrics, not durable events: process exit can lose a +notification, and concurrent writers may notify out of order after commit. Keep +callbacks short and synchronous. Use the durable task webhook outbox for +buyer-facing delivery. No database migration is needed. diff --git a/src/adcp/decisioning/__init__.py b/src/adcp/decisioning/__init__.py index 2ca3b9c2c..5bb339608 100644 --- a/src/adcp/decisioning/__init__.py +++ b/src/adcp/decisioning/__init__.py @@ -224,8 +224,10 @@ def create_media_buy( from adcp.decisioning.task_registry import ( InMemoryTaskRegistry, TaskHandoffContext, + TaskLifecycleObserver, TaskRegistry, TaskState, + TaskTransition, TaskWebhookAuthentication, ) from adcp.decisioning.tenant_store import create_tenant_store @@ -473,6 +475,8 @@ def __init__(self, *args: object, **kwargs: object) -> None: "TaskHandoff", "TaskHandoffContext", "TaskRegistry", + "TaskLifecycleObserver", + "TaskTransition", "TaskState", "TaskWebhookAuthentication", "WebhookSenderResolver", diff --git a/src/adcp/decisioning/pg/task_registry.py b/src/adcp/decisioning/pg/task_registry.py index 0e1129cdd..a9fe431a5 100644 --- a/src/adcp/decisioning/pg/task_registry.py +++ b/src/adcp/decisioning/pg/task_registry.py @@ -71,7 +71,7 @@ async def main(): from typing import TYPE_CHECKING, Any, ClassVar, TypeAlias from adcp.decisioning.account_projection import strip_credentials_from_wire_result -from adcp.decisioning.task_registry import TaskWebhookAuthentication +from adcp.decisioning.task_registry import TaskWebhookAuthentication, _TaskLifecycleObservers if TYPE_CHECKING: from psycopg_pool import AsyncConnectionPool @@ -102,7 +102,7 @@ async def main(): WebhookSigningScopeResolver: TypeAlias = Callable[["RequestContext[Any]"], str | Awaitable[str]] -class PgTaskRegistry: +class PgTaskRegistry(_TaskLifecycleObservers): """PostgreSQL-backed :class:`~adcp.decisioning.TaskRegistry` — v6.1. Durable counterpart to :class:`~adcp.decisioning.InMemoryTaskRegistry`. @@ -156,6 +156,7 @@ def __init__( "webhook_signing_scope_resolver is required exactly when the " "PgTaskWebhookOutbox uses sender_resolver" ) + self._init_lifecycle_observers() self._pool = pool self._table = _table self.task_webhook_outbox = task_webhook_outbox @@ -165,33 +166,36 @@ def __init__( # Pre-format queries at construction so the hot path avoids f-strings per call. # _table is whitelisted by _SAFE_IDENTIFIER_RE above. self._sql_insert = ( # noqa: S608 — table name is whitelisted - f"INSERT INTO {self._table}" + f"INSERT INTO {self._table}" # nosec B608 — table identifier is constructor-validated f" (task_id, account_id, state, task_type, request_context," f" webhook_registration, webhook_registration_nonce, created_at, updated_at)" f" VALUES (%s, %s, 'submitted', %s, %s::jsonb, %s, %s, %s, %s)" ) self._sql_update_progress = ( # noqa: S608 - f"UPDATE {self._table}" - f" SET state = CASE state WHEN 'submitted' THEN 'working' ELSE state END," - f" progress = %s::jsonb, updated_at = %s" - f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" + f"WITH previous AS (SELECT task_id, state FROM {self._table}" # nosec B608 — table identifier is constructor-validated + f" WHERE task_id = %s AND state NOT IN ('completed', 'failed') FOR UPDATE)" + f" UPDATE {self._table} AS task" + f" SET state = 'working', progress = %s::jsonb, updated_at = %s" + f" FROM previous WHERE task.task_id = previous.task_id" + f" RETURNING previous.state, task.task_id, task.account_id, task.task_type," + f" task.created_at, task.updated_at" ) self._sql_complete = ( # noqa: S608 - f"UPDATE {self._table}" + f"UPDATE {self._table}" # nosec B608 — table identifier is constructor-validated f" SET state = 'completed', result = %s::jsonb, updated_at = %s" f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" f" RETURNING task_id, account_id, task_type, webhook_registration," - f" webhook_registration_nonce" + f" webhook_registration_nonce, created_at, updated_at" ) self._sql_fail = ( # noqa: S608 - f"UPDATE {self._table}" + f"UPDATE {self._table}" # nosec B608 — table identifier is constructor-validated f" SET state = 'failed', error = %s::jsonb, updated_at = %s" f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')" f" RETURNING task_id, account_id, task_type, webhook_registration," - f" webhook_registration_nonce" + f" webhook_registration_nonce, created_at, updated_at" ) self._sql_clear_webhook_registration = ( # noqa: S608 - f"UPDATE {self._table} SET webhook_registration = NULL," + f"UPDATE {self._table} SET webhook_registration = NULL," # nosec B608 — table identifier is constructor-validated f" webhook_registration_nonce = NULL WHERE task_id = %s" ) # Explicit ``::text`` cast on the optional account-filter @@ -201,21 +205,21 @@ def __init__( # ``%s IS NULL`` predicate gives psycopg no type context # for the parameter and the query fails at prepare time. self._sql_get = ( # noqa: S608 - f"SELECT task_id, account_id, state, task_type," + f"SELECT task_id, account_id, state, task_type," # nosec B608 — table identifier is constructor-validated f" progress, result, error, request_context, created_at, updated_at" f" FROM {self._table}" f" WHERE task_id = %s AND (%s::text IS NULL OR account_id = %s)" ) self._sql_get_state_result = ( # noqa: S608 - f"SELECT state, result, task_type FROM {self._table} WHERE task_id = %s" - ) - self._sql_get_task_type = ( # noqa: S608 - f"SELECT task_type FROM {self._table} WHERE task_id = %s" + f"SELECT state, result, task_type FROM {self._table}" # nosec B608 — validated identifier + " WHERE task_id = %s" ) - self._sql_get_state_error = ( # noqa: S608 - f"SELECT state, error FROM {self._table} WHERE task_id = %s" + self._sql_get_task_type = f"SELECT task_type FROM {self._table} WHERE task_id = %s" # noqa: S608 # nosec B608 — table identifier is constructor-validated + self._sql_get_state_error = f"SELECT state, error FROM {self._table} WHERE task_id = %s" # noqa: S608 # nosec B608 — table identifier is constructor-validated + self._sql_discard = ( # noqa: S608 + f"DELETE FROM {self._table} WHERE task_id = %s" # nosec B608 — table identifier is constructor-validated + f" RETURNING task_id, account_id, task_type, created_at, updated_at" ) - self._sql_discard = f"DELETE FROM {self._table} WHERE task_id = %s" # noqa: S608 self._sql_ddl = ( # noqa: S608 f"CREATE TABLE IF NOT EXISTS {self._table} (" f' task_id TEXT COLLATE "C" NOT NULL PRIMARY KEY,' @@ -331,6 +335,16 @@ async def issue( now, ), ) + self._notify_lifecycle_observers( + "submitted", + { + "task_id": task_id, + "account_id": account_id, + "task_type": task_type, + "created_at": now, + "updated_at": now, + }, + ) return task_id async def resolve_webhook_signing_scope( @@ -390,21 +404,29 @@ async def update_progress( straggler progress event. """ async with self._pool.connection() as conn: - await conn.execute( + cur = await conn.execute( self._sql_update_progress, - (json.dumps(progress), time.time(), task_id), + (task_id, json.dumps(progress), time.time()), ) - # Zero rows updated means unknown task_id or terminal state — silent - # no-op per Protocol contract. The InMemoryTaskRegistry logs a - # WARNING on terminal-state drops; we omit the extra SELECT needed - # to distinguish the two cases since the dispatch wrapper swallows - # the result either way. + row = await cur.fetchone() + if row is not None and row[0] == "submitted": + self._notify_lifecycle_observers("working", self._lifecycle_record(row[1:])) - async def complete( + @staticmethod + def _lifecycle_record(row: tuple[Any, ...]) -> dict[str, Any]: + return dict(zip(("task_id", "account_id", "task_type", "created_at", "updated_at"), row)) + + async def complete(self, task_id: str, result: dict[str, Any]) -> None: + """Complete once; notify metrics only after the transaction commits.""" + row = await self._complete(task_id, result) + if row is not None: + self._notify_lifecycle_observers("completed", self._lifecycle_record(row[:3] + row[5:])) + + async def _complete( self, task_id: str, result: dict[str, Any], - ) -> None: + ) -> tuple[Any, ...] | None: """Mark the task ``completed`` with ``result`` as the terminal artifact. Idempotent on repeated calls with an equal ``result``; raises @@ -438,7 +460,7 @@ async def complete( status="completed", payload=safe_result, ) - return # updated successfully + return row # connection and transaction exit before the caller notifies # Zero rows in RETURNING — task is unknown or already terminal. cur2 = await conn.execute(self._sql_get_state_result, (task_id,)) @@ -449,15 +471,21 @@ async def complete( safe_result = strip_credentials_from_wire_result(task_type, result) if state == "completed": if existing_result == safe_result: - return # idempotent + return None # idempotent raise ValueError(f"Task {task_id!r} already completed with a different result") raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") - async def fail( + async def fail(self, task_id: str, error: dict[str, Any]) -> None: + """Fail once; notify metrics only after the transaction commits.""" + row = await self._fail(task_id, error) + if row is not None: + self._notify_lifecycle_observers("failed", self._lifecycle_record(row[:3] + row[5:])) + + async def _fail( self, task_id: str, error: dict[str, Any], - ) -> None: + ) -> tuple[Any, ...] | None: """Mark the task ``failed`` with ``error`` as the terminal payload. Idempotent on repeated calls with an equal ``error``; raises @@ -477,7 +505,7 @@ async def fail( status="failed", payload=error, ) - return # updated successfully + return row # notify only after connection/transaction exit # Zero rows in RETURNING — task is unknown or already terminal. cur2 = await conn.execute(self._sql_get_state_error, (task_id,)) @@ -487,7 +515,7 @@ async def fail( state, existing_error = row if state == "failed": if existing_error == error: - return # idempotent + return None # idempotent raise ValueError(f"Task {task_id!r} already failed with a different error") raise ValueError(f"Task {task_id!r} already in terminal state {state!r}") @@ -533,7 +561,7 @@ async def _enqueue_terminal_if_registered( payload: dict[str, Any], ) -> None: """Enqueue on ``conn`` so task state and webhook commit atomically.""" - task_id, account_id, task_type, encrypted_registration, registration_nonce = row + task_id, account_id, task_type, encrypted_registration, registration_nonce = row[:5] if encrypted_registration is None: return if self.task_webhook_outbox is None: @@ -576,7 +604,12 @@ async def discard(self, task_id: str) -> None: matching the :class:`~adcp.decisioning.InMemoryTaskRegistry` contract. """ async with self._pool.connection() as conn: - await conn.execute(self._sql_discard, (task_id,)) + cur = await conn.execute(self._sql_discard, (task_id,)) + row = await cur.fetchone() + if row is not None: + event_record = self._lifecycle_record(row) + event_record["updated_at"] = time.time() + self._notify_lifecycle_observers("discarded", event_record) #: Backwards-compat alias. Renamed to :class:`PgTaskRegistry` in diff --git a/src/adcp/decisioning/task_registry.py b/src/adcp/decisioning/task_registry.py index c4479b259..6d819beeb 100644 --- a/src/adcp/decisioning/task_registry.py +++ b/src/adcp/decisioning/task_registry.py @@ -37,6 +37,7 @@ import asyncio import logging +import threading import time import uuid from collections.abc import Awaitable, Callable @@ -51,6 +52,71 @@ TaskState = Literal["submitted", "working", "completed", "failed"] +TaskTransition = Literal["submitted", "working", "completed", "failed", "discarded"] + + +class TaskLifecycleObserver(Protocol): + """Synchronous best-effort metrics hook; no durable delivery guarantee.""" + + def __call__( + self, + event: TaskTransition, + *, + task_id: str, + account_id: str, + task_type: str, + created_at: float, + updated_at: float, + ) -> None: ... + + +class _TaskLifecycleObservers: + """Thread-safe observer registration, shared by SDK registry implementations.""" + + def _init_lifecycle_observers(self) -> None: + self._lifecycle_observers: list[TaskLifecycleObserver] = [] + self._lifecycle_observers_lock = threading.Lock() + + def add_lifecycle_observer(self, observer: TaskLifecycleObserver) -> None: + """Register a synchronous hook fired after commit/outside registry locks. + + Re-registering the same callback is a no-op. Notifications use a snapshot; + registration changes during delivery apply to the next notification. + Observer exceptions are logged and swallowed. Hooks are best-effort metrics, + not a durable event stream, and concurrent transitions may notify out of order. + """ + with self._lifecycle_observers_lock: + if observer not in self._lifecycle_observers: + self._lifecycle_observers.append(observer) + + def remove_lifecycle_observer(self, observer: TaskLifecycleObserver) -> bool: + """Remove a hook, returning whether it was registered.""" + with self._lifecycle_observers_lock: + try: + self._lifecycle_observers.remove(observer) + except ValueError: + return False + return True + + def _notify_lifecycle_observers(self, event: TaskTransition, record: dict[str, Any]) -> None: + with self._lifecycle_observers_lock: + observers = tuple(self._lifecycle_observers) + for observer in observers: + try: + observer( + event, + task_id=record["task_id"], + account_id=record["account_id"], + task_type=record["task_type"], + created_at=record["created_at"], + updated_at=record["updated_at"], + ) + except Exception: + logger.exception( + "Task lifecycle observer failed for %s task %s", event, record["task_id"] + ) + + @dataclass(frozen=True) class TaskWebhookAuthentication: """Buyer-selected authentication for a durable task webhook. @@ -388,7 +454,7 @@ async def discard(self, task_id: str) -> None: # --------------------------------------------------------------------------- -class InMemoryTaskRegistry: +class InMemoryTaskRegistry(_TaskLifecycleObservers): """Process-local task registry — v6.0 reference implementation. Storage is a plain ``dict[str, TaskRecord]`` guarded by an @@ -411,6 +477,7 @@ class InMemoryTaskRegistry: is_durable: ClassVar[bool] = False def __init__(self) -> None: + self._init_lifecycle_observers() self._records: dict[str, TaskRecord] = {} self._lock = asyncio.Lock() @@ -454,6 +521,8 @@ async def issue( task_type=task_type, request_context=(dict(request_context) if request_context is not None else None), ) + event_record = self._records[task_id].to_dict() + self._notify_lifecycle_observers("submitted", event_record) return task_id async def update_progress( @@ -483,10 +552,14 @@ async def update_progress( record.state, ) return + transitioned = record.state == "submitted" record.progress = dict(progress) if record.state == "submitted": record.state = "working" record.updated_at = time.time() + event_record = record.to_dict() + if transitioned: + self._notify_lifecycle_observers("working", event_record) async def complete( self, @@ -515,9 +588,13 @@ async def complete( if record.result == stripped: return # idempotent raise ValueError(f"Task {task_id!r} already completed with a different result") + if record.state == "failed": + raise ValueError(f"Task {task_id!r} already in terminal state {record.state!r}") record.state = "completed" record.result = dict(stripped) if isinstance(stripped, dict) else stripped record.updated_at = time.time() + event_record = record.to_dict() + self._notify_lifecycle_observers("completed", event_record) async def fail( self, @@ -532,9 +609,13 @@ async def fail( if record.error == error: return # idempotent raise ValueError(f"Task {task_id!r} already failed with a different error") + if record.state == "completed": + raise ValueError(f"Task {task_id!r} already in terminal state {record.state!r}") record.state = "failed" record.error = dict(error) record.updated_at = time.time() + event_record = record.to_dict() + self._notify_lifecycle_observers("failed", event_record) async def get( self, @@ -562,7 +643,12 @@ async def discard(self, task_id: str) -> None: # 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) + record = self._records.pop(task_id, None) + if record is not None: + event_record = record.to_dict() + event_record["updated_at"] = time.time() + if record is not None: + self._notify_lifecycle_observers("discarded", event_record) # --------------------------------------------------------------------------- diff --git a/tests/test_decisioning_task_lifecycle.py b/tests/test_decisioning_task_lifecycle.py new file mode 100644 index 000000000..aa1b5190d --- /dev/null +++ b/tests/test_decisioning_task_lifecycle.py @@ -0,0 +1,166 @@ +"""Best-effort metrics hooks count committed transitions, including races.""" + +from __future__ import annotations + +import asyncio +import os +import secrets +from collections.abc import AsyncIterator +from typing import Any + +import pytest + +from adcp.decisioning import InMemoryTaskRegistry, TaskTransition + + +@pytest.fixture(params=["memory", "postgres", "postgres-autocommit"]) +async def registry(request: pytest.FixtureRequest) -> AsyncIterator[Any]: + if request.param == "memory": + yield InMemoryTaskRegistry() + return + from psycopg_pool import AsyncConnectionPool + + from adcp.decisioning.pg import PgTaskRegistry + + url = os.environ.get("ADCP_PG_TEST_URL") + if not url: + pytest.skip("ADCP_PG_TEST_URL required") + table = f"test_lifecycle_{secrets.token_hex(6)}" + async with AsyncConnectionPool( + url, + open=False, + min_size=1, + max_size=4, + kwargs={"autocommit": request.param == "postgres-autocommit"}, + ) as pool: + reg = PgTaskRegistry(pool=pool, _table=table) + await reg.create_schema() + yield reg + async with pool.connection() as conn: + await conn.execute(f"DROP TABLE {table}") + + +async def test_observe_committed_transitions_and_suppress_noops(registry: Any) -> None: + events: list[tuple[str, dict[str, Any]]] = [] + + def observe(event: TaskTransition, **metadata: Any) -> None: + # A synchronous callback can inspect committed state on another connection. + # It cannot observe an uncommitted row or run under the memory registry lock. + if isinstance(registry, InMemoryTaskRegistry): + assert not registry._lock.locked() + record = registry._records.get(metadata["task_id"]) + assert record is None if event == "discarded" else record.state == event + else: + import psycopg + + with psycopg.connect(os.environ["ADCP_PG_TEST_URL"]) as conn: + row = conn.execute( + f"SELECT state FROM {registry._table} WHERE task_id = %s", + (metadata["task_id"],), + ).fetchone() + assert row is None if event == "discarded" else row[0] == event + events.append((event, metadata)) + + registry.add_lifecycle_observer(observe) + registry.add_lifecycle_observer(observe) + task = await registry.issue(account_id="acct_1", task_type="get_products") + await asyncio.gather(*(registry.update_progress(task, {"percentage": n}) for n in range(10))) + await asyncio.gather(*(registry.complete(task, {"products": []}) for _ in range(10))) + await registry.update_progress(task, {"percentage": 100}) + await registry.update_progress("absent", {}) + await registry.discard("absent") + await registry.discard(task) + await registry.discard(task) + assert [event for event, _ in events] == ["submitted", "working", "completed", "discarded"] + for _, metadata in events: + assert metadata["task_id"] == task + assert metadata["account_id"] == "acct_1" + assert metadata["task_type"] == "get_products" + assert metadata["created_at"] <= metadata["updated_at"] + assert registry.remove_lifecycle_observer(observe) is True + assert registry.remove_lifecycle_observer(observe) is False + await registry.issue(account_id="acct_1", task_type="get_products") + assert len(events) == 4 + + +async def test_racing_terminal_transitions_emit_once(registry: Any) -> None: + events = [] + + def observe(event: TaskTransition, **metadata: Any) -> None: + events.append(event) + + registry.add_lifecycle_observer(observe) + task = await registry.issue(account_id="acct", task_type="get_products") + results = await asyncio.gather( + *(registry.complete(task, {"products": []}) for _ in range(10)), + *(registry.fail(task, {"code": "INTERNAL_ERROR", "message": "Failed"}) for _ in range(10)), + return_exceptions=True, + ) + assert events[0] == "submitted" + assert len(events) == 2 and events[1] in {"completed", "failed"} + assert any(isinstance(result, ValueError) for result in results) + record = await registry.get(task) + assert record["state"] == events[1] + if record["state"] == "completed": + with pytest.raises(ValueError): + await registry.fail(task, {"code": "CONFLICT", "message": "Conflict"}) + else: + with pytest.raises(ValueError): + await registry.complete(task, {"products": []}) + assert len(events) == 2 + + +async def test_observer_failure_and_snapshot_removal( + registry: Any, caplog: pytest.LogCaptureFixture +) -> None: + events = [] + + def broken(event: TaskTransition, **metadata: Any) -> None: + raise RuntimeError("Metrics unavailable") + + def removed(event: TaskTransition, **metadata: Any) -> None: + events.append(event) + + def remover(event: TaskTransition, **metadata: Any) -> None: + registry.remove_lifecycle_observer(removed) + + registry.add_lifecycle_observer(broken) + registry.add_lifecycle_observer(remover) + registry.add_lifecycle_observer(removed) + task = await registry.issue(account_id="acct", task_type="get_products") + await registry.fail(task, {"code": "INTERNAL_ERROR", "message": "Failed"}) + await registry.fail(task, {"code": "INTERNAL_ERROR", "message": "Failed"}) + assert events == ["submitted"] + assert "Task lifecycle observer failed" in caplog.text + assert (await registry.get(task))["state"] == "failed" + + +async def test_failed_database_transaction_never_notifies(registry: Any) -> None: + if isinstance(registry, InMemoryTaskRegistry): + with pytest.raises(ValueError): + await registry.complete("missing", {}) + return + import psycopg + + events = [] + registry.add_lifecycle_observer(lambda event, **metadata: events.append(event)) + task = await registry.issue(account_id="acct", task_type="get_products") + function = f"{registry._table}_reject" + async with registry._pool.connection() as conn: + await conn.execute( + f"CREATE FUNCTION {function}() RETURNS trigger LANGUAGE plpgsql AS $$ " + "BEGIN RAISE EXCEPTION 'fail commit'; END $$" + ) + await conn.execute( + f"CREATE CONSTRAINT TRIGGER reject_transition AFTER UPDATE ON {registry._table}" + f" DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION {function}()" + ) + try: + with pytest.raises(psycopg.errors.RaiseException): + await registry.complete(task, {"products": []}) + assert events == ["submitted"] + assert (await registry.get(task))["state"] == "submitted" + finally: + async with registry._pool.connection() as conn: + await conn.execute(f"DROP TRIGGER reject_transition ON {registry._table}") + await conn.execute(f"DROP FUNCTION {function}()") diff --git a/tests/type_checks/task_lifecycle_observers.py b/tests/type_checks/task_lifecycle_observers.py new file mode 100644 index 000000000..f628d38c9 --- /dev/null +++ b/tests/type_checks/task_lifecycle_observers.py @@ -0,0 +1,39 @@ +"""Public metrics hook works without changing the minimal registry protocol.""" + +from typing import Any + +from adcp.decisioning import ( + InMemoryTaskRegistry, + TaskLifecycleObserver, + TaskRegistry, + TaskTransition, +) +from adcp.decisioning.pg import PgTaskRegistry + + +class Metrics: + def __call__( + self, + event: TaskTransition, + *, + task_id: str, + account_id: str, + task_type: str, + created_at: float, + updated_at: float, + ) -> None: + print(event, task_id, account_id, task_type, updated_at - created_at) + + +observer: TaskLifecycleObserver = Metrics() + + +async def instrument(memory: InMemoryTaskRegistry, postgres: PgTaskRegistry) -> None: + for registry in (memory, postgres): + registry.add_lifecycle_observer(observer) + minimal: TaskRegistry = registry + task = await minimal.issue(account_id="account", task_type="get_products") + result: dict[str, Any] = {"products": []} + await minimal.complete(task, result) + removed: bool = registry.remove_lifecycle_observer(observer) + assert removed From 37a69cd4b0dc9ce12a5f3bfd9d93cb91ad2eee15 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 2 Oct 2026 21:30:04 +0000 Subject: [PATCH 2/3] fix(pg): type terminal task rows without optional driver imports --- src/adcp/decisioning/pg/task_registry.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/adcp/decisioning/pg/task_registry.py b/src/adcp/decisioning/pg/task_registry.py index a9fe431a5..497b97655 100644 --- a/src/adcp/decisioning/pg/task_registry.py +++ b/src/adcp/decisioning/pg/task_registry.py @@ -452,7 +452,7 @@ async def _complete( self._sql_complete, (json.dumps(safe_result), time.time(), task_id), ) - row = await cur.fetchone() + row: tuple[Any, ...] | None = await cur.fetchone() if row is not None: await self._enqueue_terminal_if_registered( conn, @@ -497,7 +497,7 @@ async def _fail( self._sql_fail, (json.dumps(error), time.time(), task_id), ) - row = await cur.fetchone() + row: tuple[Any, ...] | None = await cur.fetchone() if row is not None: await self._enqueue_terminal_if_registered( conn, From 4b66cd58f564d83e5855a92f91c4226860e0fa08 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sat, 3 Oct 2026 02:39:58 +0000 Subject: [PATCH 3/3] test(tasks): configure PostgreSQL lifecycle coverage explicitly --- .github/workflows/ci.yml | 1 + tests/test_decisioning_task_lifecycle.py | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a72211d14..c6e3254a4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -195,6 +195,7 @@ jobs: tests/conformance/signing/test_pg_replay_store_e2e.py \ tests/conformance/decisioning/test_pg_buyer_agent_registry.py \ tests/conformance/decisioning/test_pg_idempotency_backend.py \ + tests/test_decisioning_task_lifecycle.py \ tests/conformance/decisioning/test_pg_task_webhook_outbox.py \ tests/conformance/decisioning/test_pg_reference_workflow_queue.py \ tests/conformance/reporting/ \ diff --git a/tests/test_decisioning_task_lifecycle.py b/tests/test_decisioning_task_lifecycle.py index aa1b5190d..26944ffdd 100644 --- a/tests/test_decisioning_task_lifecycle.py +++ b/tests/test_decisioning_task_lifecycle.py @@ -18,13 +18,13 @@ async def registry(request: pytest.FixtureRequest) -> AsyncIterator[Any]: if request.param == "memory": yield InMemoryTaskRegistry() return + url = os.environ.get("ADCP_PG_TEST_URL") + if not url: + pytest.skip("ADCP_PG_TEST_URL required") from psycopg_pool import AsyncConnectionPool from adcp.decisioning.pg import PgTaskRegistry - url = os.environ.get("ADCP_PG_TEST_URL") - if not url: - pytest.skip("ADCP_PG_TEST_URL required") table = f"test_lifecycle_{secrets.token_hex(6)}" async with AsyncConnectionPool( url,