Skip to content

feat(decisioning): lifecycle observer hook on PgTaskRegistry / TaskRegistry for task metrics #1296

Closed

Description

@bokelley

Summary

On 8.0.0, PgTaskRegistry has no way to observe task lifecycle transitions. Its constructor (adcp/decisioning/pg/task_registry.py:135-163) accepts only pool, task_webhook_outbox, webhook_signing_scope_resolver and a private _table. issue, update_progress, complete and fail all return None. An adopter therefore cannot count tasks created, completed or failed, or measure time to terminal, without forking or subclassing the registry. PgBuyerAgentRegistry already has this kind of hook (add_mutation_observer, decisioning/pg/buyer_agent_registry.py:371-408), and the task registry has no equivalent.

Details

Subclassing and overriding the write methods doesn't produce correct metrics:

  • complete() returns None both when it performs the working -> completed transition and when it is an idempotent repeat with an equal result (task_registry.py:403-454; the repeat path is at 449-452). A wrapper can't tell the first terminal transition from a retry, so it double-counts.
  • update_progress() silently no-ops on unknown or terminal tasks, and doesn't report whether submitted -> working happened (task_registry.py:376-402).
  • issue() takes eight keyword arguments plus **_extra, five of them for webhook registration. A subclass has to mirror that signature and keep it in sync on every upgrade.

Only the registry knows, inside its transaction, whether a row actually changed state. That makes it the right place for the hook.

Why an adopter hits this

Embedded Sales Agent moved async get_products onto the SDK's PgTaskRegistry (core/decisioning/task_registry.py). Its earlier registry emitted get_products_async_tasks_created_total, ..._completed_total and ..._failed_total. Those series are now kept at zero on purpose, so existing dashboards don't show no-data alerts. src/core/metrics.py (~lines 87-106) states the reason: "The exact SDK PgTaskRegistry has no lifecycle-observer hook, so these counters intentionally remain at zero." Operators have lost visibility into async task throughput and failure rate. The only metric ESA still records, task expiry, comes from its own tasks_get shim, not the registry.

Ask

Add an observer seam that both InMemoryTaskRegistry and PgTaskRegistry honor. For example:

TaskTransition = Literal["issued", "working", "completed", "failed", "discarded"]

class TaskLifecycleObserver(Protocol):
    def __call__(self, event: TaskTransition, *, task_id: str, account_id: str,
                 task_type: str, created_at: float, updated_at: float) -> None: ...

PgTaskRegistry(pool=..., observers=[my_observer])   # or registry.add_lifecycle_observer(...)

The contract:

  • Fire only after commit.
  • Fire only when a row actually changed state, never on idempotent repeats or silent no-ops.
  • Log and swallow observer exceptions, as add_mutation_observer does.

A smaller fallback: have complete, fail and update_progress return a bool saying whether the transition was applied, so a thin subclass can count correctly.

Activity

  1. added
    enhancementNew feature or request
    claude-triagingTriage routine is actively working on this issue (1-3 min)
    on Oct 2, 2026
  2. bokelley commented on Oct 2, 2026

    @bokelley
    ContributorAuthor

    Triage

    Classification: Feature request
    Bucket(s): client (no matching repo label — gap noted)
    Status: ready-to-implement

    What the experts said:

    • code-reviewer: Non-breaking — add to concrete classes only, not to TaskRegistry Protocol (@runtime_checkable); three impl blockers: mutable observers=[] default (use None), Pg update_progress SQL needs RETURNING old state to distinguish submitted→working from idempotent, discard SQL needs RETURNING task_id for no-op detection; bool-return fallback is breaking — drop it.
    • dx-expert: Prefer add_lifecycle_observer()/remove_lifecycle_observer() (mirrors PgBuyerAgentRegistry) over constructor param; fire observers outside DB transaction and asyncio lock; rename "issued" → "submitted" in TaskTransition to match TaskState vocabulary.

    My take: Additive feature with direct codebase precedent. Experts converge on API shape and impl requirements with no unresolved disagreements.

    Implementation scope:

    • Add TaskLifecycleObserver Protocol + TaskTransition = Literal["submitted", "working", "completed", "failed", "discarded"] to task_registry.py; add add_lifecycle_observer/remove_lifecycle_observer to both concrete classes (not to TaskRegistry Protocol)
    • Fire callbacks after DB commit / after asyncio lock release; signature (event, *, task_id, account_id, task_type, created_at, updated_at); log-and-swallow (match _notify_mutation at pg/buyer_agent_registry.py:371)
    • Pg SQL: update_progress → RETURNING state to detect real submitted→working; discard → DELETE … RETURNING task_id; InMemoryTaskRegistry.discard → check pop() return
    • Required checks: pytest tests/test_decisioning_task_registry*, mypy src/adcp/decisioning/, ruff check . && black --check .

    Triaged by Claude Code. Session: https://claude.ai/code/session_011567bv6By9jafEm1bK8Qd7


    Generated by Claude Code

  3. added and removed
    claude-triagingTriage routine is actively working on this issue (1-3 min)
    on Oct 2, 2026
  4. bokelley commented on Oct 4, 2026

    @bokelley
    ContributorAuthor

    Fixed by #1323 (squash 9a70217): committed task lifecycle transitions now notify registered observers after commit, outside locks, without duplicate or no-op events.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions