Repository navigation
feat(reporting): notify the adopter when a reporting issue opens, escalates or resolves #1258
Description
Summary
On 8.0.0b18, a seller has no way to learn that a reporting issue opened. This matters most for CONSUMER_STATUS_MISMATCH, which only the submitting caller can see.
- Where issues open:
ConsumerStatusIngest._open_issue_on_first_observationopens the occurrence when async_reporting_statusstatement lands. Theget_reporting_statuslifecycle intents open it on a read. Both go throughstore.ensure_issue_opened. - What the SDK exposes: neither path calls back into the adopter, and
ReliableReportingService.__init__has no hook for it. The renderedReportingIssue, withresponsible_party,recommended_action, severity andopened_at, only exists inside that caller's status response.
So a mismatch the seller must act on (responsible_party: seller or provider) is invisible to the seller's operators unless the buyer contacts them.
What ESA does today
ESA scans every get_reporting_status response it serves for open CONSUMER_STATUS_MISMATCH issues attributed to the seller or provider. It logs them and sends a Sentry event, de-duplicated per issue_id per process. That has two problems:
- It depends on the buyer. A buyer that files a statement and never reads status again is never alerted on.
_open_issue_on_first_observationexists precisely to handle that unattended case. - It is tied to reads. It fires on a read, not when the occurrence opens. Escalations past
consumer_mismatch_escalation_secondsare only seen if someone polls, and the de-duplication has to be rebuilt in every process.
Ask
An adopter hook on the service, called once per lifecycle transition, after the store commits:
async def on_issue(event: ReportingIssueEvent) -> None: ...
ReliableReportingService(..., issue_observer=on_issue)where ReportingIssueEvent carries:
kind:opened,escalated,acknowledged,waivedorresolved- the rendered
ReportingIssue account_idandconsumer_id(for caller-scoped issues)- the causing
reporting_status_id
Requirements:
- Exactly once per transition across replicas.
reporting_issue_lifecyclealready makes opening idempotent (reporting_issue_lifecycle_one_live), so the transition that inserts the row can fire the event. - Fires on ingest. The event should fire from
sync_reporting_status, not only fromget_reporting_status, so an unattended statement still reaches the seller. - Observer failures don't fail the request. An exception in the observer is reported through
worker_error_handleror logged, and never fails the buyer's call. - Escalation fires on time.
escalatedfires when the escalation boundary passes. This needs the worker to evaluate open occurrences on its turn, because no read may happen.
ESA would then swap its response scan for this hook.
Refs #1240.
Activity
bokelley commentedon Sep 30, 2026
Triage
Classification: Feature request
Bucket(s): service layer (
src/adcp/reporting/service.py,ledger/consumer_status.py)Status: ready-to-implement
What the experts said:
ensure_issue_openedhas no "was_new" discriminator — firing the observer unconditionally would break the exactly-once guarantee across replicas. Escalation also uses a separate time-based worker path, not the ingest path.*_handlerconvention (reporting_issue_handler);production=guard is the key blocker for ESA —issue_observermust be explicitly carved out of it or threaded throughReportingProductionSupport.My take: Well-scoped, clearly specified, non-breaking. The two design decisions to settle before coding: (1) does
reporting_issue_handlerwork alongsideproduction=, and if so, thread it throughReportingProductionSupport._compose; (2)ensure_issue_openedneeds awas_opened: booldiscriminator so the observer fires only on the actual INSERT, not on read-path idempotency hits.Implementation scope:
src/adcp/reporting/ledger/models.py— addReportingIssueEventas@dataclass(frozen=True)withkind: Literal["opened", "escalated", "acknowledged", "waived", "resolved"],issue: ReportingIssue,account_id: str,consumer_id: str | None,reporting_status_id: strsrc/adcp/reporting/ledger/store.py+pg.py— extendensure_issue_openedreturn to carrywas_opened: bool(True only from the INSERT branch in_ensure_issue_opened_on); no change to thestatus.py/status_snapshot.pyread-path callsites (they must not fire the observer)src/adcp/reporting/ledger/consumer_status.py— addissue_observerfield onConsumerStatusIngest; fire afterensure_issue_openedat line ~299 only whenwas_openedis True; wrap in try/except, route errors viaworker_error_handlersrc/adcp/reporting/service.py— addreporting_issue_handler: Callable[[ReportingIssueEvent], Any | Awaitable[Any]] | None = Noneparam; type aliasReportingIssueHandler; decideproduction=guard (carve out or add toReportingProductionOptions); export in__all__kind="escalated"; this is a separate code path from ingestpyteston reporting ledger/service tests,mypy src/,ruff check .Nonedefault; new frozen dataclass; new try/except-guarded callbacksTriaged by Claude Code. Session: https://claude.ai/code/session_01YCsVHyCBbgzLqR1fQvoz1j
Generated by Claude Code
feat(reporting): notify the adopter when a reporting issue opens, esc…
feat(reporting): notify the adopter when a reporting issue opens, esc…
feat(reporting): notify the adopter when a reporting issue opens, esc…
bokelley commentedon Nov 9, 2026
Fixed by #1875 and #1874 on
next. The remainingproduction=gap is tracked in #1873. It didn't auto-close becausemainis the default branch.