feat(reporting): persist scheduled provisional observations

.github/workflows/ci.yml

@@ -76,13 +76,11 @@ jobs:
     # ruff/mypy/mypy --strict/the type-ignore contract before pytest starts,
     # then post-job cleanup after it ends. On an ubuntu-latest runner that
     # pre-test work is ~1m30s and the plain suite is ~16m, but 3.12 adds
-    # coverage tracing on ~9.4k tests: at 30 minutes that leg was cancelled
-    # mid-run at 99% with no failing test, which reports as a red matrix and
-    # hides real signal. Bound the suite itself below (so a hang fails one
-    # named step with its own message instead of silently taking the job), and
-    # leave the job enough room that step bound plus setup and cleanup still
-    # fit with margin for runner variance.
-    timeout-minutes: 60
+    # coverage tracing on the full suite: even a 45-minute coverage step can
+    # time out near 99% on a loaded runner with no failing test. Bound the
+    # suite itself so a hang fails the named step. The 3.12 job keeps 15
+    # minutes beyond that step for setup and cleanup; other jobs keep 60.
+    timeout-minutes: ${{ matrix.python-version == '3.12' && 70 || 60 }}
     strategy:
       matrix:
         python-version: ["3.10", "3.11", "3.12", "3.13"]
@@ -118,30 +116,33 @@ jobs:
         if: matrix.python-version == '3.12'
         run: python scripts/check_type_ignore_contract.py
 
-      # Bounded well above the observed ~16m (plain) and ~30m (coverage) runs
-      # so ordinary variance never trips it, and well below the job ceiling so
-      # a genuinely stuck suite still fails *this* step with a timeout rather
-      # than being cancelled as a whole job.
+      # Bound the plain and coverage runs separately so a genuinely stuck
+      # suite fails its named step rather than cancelling the entire job.
       - name: Run tests
         if: matrix.python-version != '3.12'
         timeout-minutes: 45
         run: python scripts/reporting_test_harness.py pytest tests/ -v -ra
 
       - name: Run tests with coverage
         if: matrix.python-version == '3.12'
-        timeout-minutes: 45
+        timeout-minutes: 55
         run: python scripts/reporting_test_harness.py pytest tests/ -v -ra --cov=src/adcp --cov-report=term-missing
 
   pg-conformance:
     name: Postgres conformance tests (Postgres 16, ${{ matrix.lane }})
     runs-on: ubuntu-latest
-    timeout-minutes: 15
+    timeout-minutes: ${{ matrix.timeout_minutes }}
     permissions:
       contents: read
     strategy:
       fail-fast: false
       matrix:
-        lane: [core, process]
+        include:
+          # Core also builds and installs the reporting wheel/sdist fixtures.
+          - lane: core
+            timeout_minutes: 30
+          - lane: process
+            timeout_minutes: 15
     services:
       postgres:
         # CI-local ephemeral database. POSTGRES_HOST_AUTH_METHOD=trust
@@ -184,8 +185,8 @@ jobs:
           ADCP_PG_TEST_URL: postgresql://postgres@localhost:5432/adcp_test
           PG_LANE: ${{ matrix.lane }}
         run: |
-          # Keep every case and its deadline. Separate process-crash controls
-          # so setup and teardown also fit inside each unchanged job budget.
+          # Keep every case and its per-case deadline. The core lane includes
+          # installed artifact coverage and has a larger overall job budget.
           case "$PG_LANE" in
             core)
               python scripts/reporting_test_harness.py pytest tests/conformance/signing/test_pg_replay_store.py \
@@ -827,7 +828,7 @@ jobs:
     runs-on: ubuntu-latest
     permissions:
       contents: read
-    timeout-minutes: 40
+    timeout-minutes: 50
     strategy:
       fail-fast: false
       matrix:
@@ -857,7 +858,7 @@ jobs:
         run: pip install -e ".[dev,pg]"
       - name: Run installed source-equivalent production contract
         shell: bash
-        timeout-minutes: 35
+        timeout-minutes: 45
         env:
           ADCP_PG_TEST_URL: postgresql://postgres@localhost:5432/adcp_production_installed_test
           ADCP_PYTHON310: ${{ steps.production-python310.outputs.python-path }}

docs/reporting-provisional-observations.md

@@ -0,0 +1,163 @@
+# Provisional observations and upgrade status
+
+The producer records every successful scheduled provisional read as a new
+immutable revision, including a read whose rows are unchanged. A durable
+reservation freezes the acquisition before source work starts. Its successful
+observation, revision rows, ledger change, notifications and next checkpoint
+commit together. A retry of the same reserved acquisition returns the existing
+revision instead of appending another observation.
+
+This is a producer and persistence change. Complete service rollout qualification
+is still pending, including activation of previously retired work, source-object
+reuse, installed-package checks and independent integration review. Follow the
+[production guide](reporting-production.md) and
+[migration guide](reporting-ledger-migration.md) for the surrounding deployment gates.
+
+## Scheduling
+
+- An offering without `restatement_window` uses an SDK fallback of 72 elapsed
+  hours after the reporting period ends. An explicit offering window continues
+  to apply.
+- The first reserved acquisition freezes the window and cadence. Cadence is the
+  explicit `restatement_cadence`, or the greater of the reporting-period duration
+  and `fastest_safe_cadence`. Later successful reads retain that policy.
+- A successful read anchors the next due time at its checked/acquired instant
+  plus cadence, capped at the resolved provisional boundary. After downtime,
+  one due read runs; missed intervals do not create a burst of catch-up reads.
+- The adapter may return typed `InlineFetchResult.provisional_until` evidence to
+  shorten or extend the boundary. It must be timezone-aware and at least the
+  source observation time. Invalid evidence commits no revision or checkpoint.
+  If evidence is absent, the frozen offering window or SDK fallback applies.
+- Snapshot-only reporting retains one final inclusive read at the boundary,
+  including after downtime. Valid new source evidence may extend that boundary.
+  Expiry does not create an official revision.
+- Existing explicitly configured official-close behavior remains in force.
+  `official_close_lag` still requires an explicitly declared `restatement_window`
+  and an official offering. The SDK fallback does not opt an adapter into
+  automatic official publication.
+
+The durable acquisition freezes the source request, predecessor revision and
+observation ordinal. A retry renews only `deadline_at`, the execution budget for
+that attempt. It retains the execution key, run ID, cutoff, scope and remaining
+request fields even after the original deadline expires. Once reserved, a
+successful retry completes that observation before a later acquisition starts.
+
+## Persistence and compatibility
+
+`PgReportingLedgerStore.create_schema()` installs two additive private tables:
+`reporting_provisional_acquisitions` and `reporting_provisional_observations`.
+They retain immutable reservation and observation metadata. The existing
+restatement checkpoint remains the progress cursor. The in-memory store has the
+same atomic operations. Custom producer stores must implement
+`ProvisionalObservationStore` from `adcp.reporting.ledger.provisional` together
+with `RestatementCheckpointStore`.
+
+The packaged `reporting/ledger/required_provisional_schema.json` describes all
+27 objects added by this extension. Bootstrap and each reservation, observation
+read, and atomic observation commit validate those objects, including immutable
+triggers and their function definitions. Validation also runs with notifications
+disabled. Missing, disabled or changed objects fail closed with
+`PROVISIONAL_SCHEMA_UNREADY`; reinstall or repair the schema before resuming
+production. Notification readiness alone does not establish observation readiness:
+historical notification-only schemas remain valid for their original operations.
+
+Existing revision tables, exact-read row layout and pinned migration manifests
+are unchanged. Each observation has its own revision ID and revision-specific
+wire content hash. Unchanged source content retains its source content identity;
+it does not mean that the revision ID or wire hash is reused. Historical readers
+can still read rows directly from `reporting_revision_rows` for every revision.
+
+This change retains per-revision row payloads. Content-addressed reuse of staged
+source objects is a separate pending change; complete payload deduplication has
+not been accepted. Do not interpret immutable observation support as completion
+of the full no-duplicate-payload requirement.
+
+Stop old producer writers before switching scheduling behavior. Old binaries do
+not write the new observation metadata, even though they can still read the
+existing revision rows. Historical-binary rolling tests and full service
+qualification must pass before claiming an upgrade is supported.
+
+## Custom stores and publication decorators
+
+Declare the checkpoint and observation methods explicitly on the store class.
+Dynamic `__getattr__` forwarding does not enable these optional capabilities.
+The producer checks declarations consistently across supported Python versions;
+an incomplete adapter raises `RESTATEMENT_CHECKPOINTS_NOT_SUPPORTED` or
+`PROVISIONAL_OBSERVATIONS_NOT_SUPPORTED` before source acquisition. It does not
+fall back to separate revision and checkpoint writes.
+
+A decorator that prepares revisions in `commit_revision` must also prepare them
+in `commit_provisional_observation`. Explicitly compose all three observation
+methods (`reserve_provisional_acquisition`, `get_provisional_observation`,
+`commit_provisional_observation`) and both checkpoint methods
+(`get_restatement_checkpoint`, `record_restatement_checkpoint`). Reuse the same
+revision preparation, validation and fault boundaries around the delegated atomic
+commit. Calling ordinary `commit_revision` first would break the observation's
+transaction boundary; forwarding the new method without preparation would bypass
+the decorator's publication rules.
+
+An `OBSERVATION_CONFLICT` fails only the affected slice for that worker turn. Its
+pending work remains eligible for a later retry while sibling slices continue.
+Schema or capability failures stop the turn so the operator can repair the store.
+
+## Pending scheduling qualification
+
+The production progress queue still selects bounded pending work before checking
+its observation due time. During the new provisional window, future-due work can
+therefore consume turn capacity that would otherwise acquire newly closed periods.
+Durable due-time selection and its fairness tests remain a service activation
+requirement; the default window is not a claim of completed production rollout.
+
+Successful observations retain their next cadence checkpoint. Failed or not-ready
+attempts retain their acquisition identity, but do not yet persist an independent
+retry time. Repeated worker turns can retry that acquisition before its source's
+safe cadence. Restart-safe retry scheduling remains required before full rollout;
+an in-process delay alone would not complete that requirement.
+
+## Previously retired work: activation requirement remains open
+
+A deployment that already processed a snapshot with no declared window may have
+retired that obligation under the old one-shot policy. Installing this schema or
+restarting a producer does **not** re-enroll it. Until the activation migration
+below is implemented and verified, the new fallback applies to eligible pending
+work; it is not a complete upgrade path for existing deployments.
+
+The affected progress APIs are `ReportingProducerProgress.next_producer_obligations`,
+`finish_producer_acquisition`, `commit_producer_period` and
+`producer_closed_through`. PostgreSQL retains retired rows as `state='settled'`
+in `reporting_production_source_work`; `reporting_production_source_progress`
+retains the generation's `closed_through` and acquisition turn. The in-memory
+equivalents are `_production_source_work`, `_production_closed` and
+`_production_source_turns`. Selection currently considers only pending work.
+
+Existing configuration enrollment installs generation/source/destination
+bindings, without changing a retained work item's state. Repeated production
+activation returns without re-enrollment. Creating a new configuration generation
+would change the frozen reporting identity and is not a repair for an old
+obligation. Rewinding `closed_through` is also unsupported.
+
+The activation owner must provide an explicit disposition for each affected
+deployment. A candidate bounded migration starts with an admitted account and
+configuration generation, selects settled snapshot obligations in a recorded
+period-end range, excludes official publications, and resumes eligible work
+without changing the obligation or its retained policy evidence. It must also
+define the treatment of already-expired obligations and source overrides; a
+simple three-day filter alone cannot establish eligibility for every legacy
+record. Missing policy evidence needs an explicit upgrade decision.
+
+An indexed selection can use
+`(account_id, delivery_config_id, delivery_config_version, period_end,
+reporting_obligation_id) WHERE state='settled'`, a bounded period range and
+keyset batches. The current pending-only index does not cover this selection.
+Any index and re-enrollment operation belong in an additive migration, under the
+existing account/generation lock, with idempotent progress and preserved
+acquisition turns. Do not scan all history on each worker turn or change the
+separate delivery queues to trigger acquisition.
+
+The required regression starts with an old writer completing a no-window
+snapshot, leaving its work row settled. Upgrade while the intended provisional
+boundary is still open, activate/configure using the same frozen generation,
+advance one cadence, and require a second immutable snapshot with the original
+revision still readable. Current activation does not satisfy that regression.
+The completed migration must also cover official, expired, parked, foreign-account
+and interrupted/resumed batches without changing unrelated work.

pyproject.toml

@@ -194,6 +194,7 @@ adcp = [
     "signing/pg/*.sql",
     "decisioning/pg/*.sql",
     "reporting/ledger/*.sql",
+    "reporting/ledger/*.json",
     "reporting/outbox/*.json",
     "reporting/materializer/assets/*.json",
     "reporting/materializer/*.json",