feat: probe-first idempotency/revision/delivery concurrency subsystem (rebuild for #1546)

.duplication-baseline

@@ -1,5 +1,5 @@
 {
-  "src": 35,
-  "tests": 73,
+  "src": 33,
+  "tests": 72,
   "scripts": 0
 }

.env.template

@@ -88,6 +88,31 @@
 #   production: Lenient - ignores unknown fields for forward compatibility
 # ENVIRONMENT=production
 
+# Read and write idempotency admission limits can be tuned independently.
+# If unset, each class falls back to the legacy IDEMPOTENCY_* setting, then
+# to the built-in defaults (1000 active rows; 300 inserts per 10 seconds).
+# IDEMPOTENCY_READ_MAX_ACTIVE_ATTEMPTS_PER_SCOPE=1000
+# IDEMPOTENCY_READ_MAX_INSERTS_PER_WINDOW=300
+# IDEMPOTENCY_READ_INSERT_RATE_WINDOW_SECONDS=10
+# IDEMPOTENCY_WRITE_MAX_ACTIVE_ATTEMPTS_PER_SCOPE=1000
+# IDEMPOTENCY_WRITE_MAX_INSERTS_PER_WINDOW=300
+# IDEMPOTENCY_WRITE_INSERT_RATE_WINDOW_SECONDS=10
+
+# When DATABASE_URL uses PgBouncer transaction pooling, provider-operation
+# advisory locks require a second URL that connects directly to PostgreSQL.
+# COORDINATION_DATABASE_URL=postgresql://user:password@postgres:5432/database
+
+# Required when accepting push_notification_config without an explicit legacy
+# Bearer/HMAC authentication block. JSON-encode a private Ed25519 or ES256 JWK
+# whose public half is published at the agent's jwks_uri and whose
+# adcp_use is "webhook-signing".
+# ADCP_WEBHOOK_SIGNING_JWK={"kty":"OKP","crv":"Ed25519","alg":"EdDSA","kid":"...","adcp_use":"webhook-signing","x":"...","d":"..."}
+# ADCP_WEBHOOK_SIGNING_JWKS_URI=https://seller.example/.well-known/jwks.json
+# ADCP_BRAND_JSON_URL=https://seller.example/.well-known/brand.json
+# Optional deployment assertion; the advertised/wire algorithm is derived from
+# the JWK and startup validation rejects a mismatch.
+# ADCP_WEBHOOK_SIGNING_ALGORITHM=ed25519
+
 # Demo tenant creation (default: false)
 # When true, creates a "Default Publisher" tenant with mock adapter on first startup
 # Set to true for local development/evaluation, leave false for production deployments

.github/workflows/ci.yml

@@ -286,6 +286,14 @@ jobs:
           docker container prune -f 2>/dev/null || true
           docker network prune -f 2>/dev/null || true
           docker volume prune -f 2>/dev/null || true
+      # The e2e stack exposes its narrowly scoped setup controls only when both
+      # ADCP_TESTING=true and this per-run secret are present. run_all_tests.sh
+      # generates it for local runs; this job invokes pytest directly, so it has
+      # to generate its own — the compose stack (docker-compose.e2e.yml) and the
+      # tests both read it from the environment and must see the same value.
+      - name: Generate the per-run e2e control token
+        run: |
+          echo "ADCP_TEST_CONTROL_TOKEN=$(python3 -c 'import secrets; print(secrets.token_hex(32))')" >> "$GITHUB_ENV"
       # No coverage: the server under test runs inside Docker, so host-side
       # coverage would measure the client harness, not src/.
       - uses: ./.github/actions/_pytest
@@ -365,9 +373,30 @@ jobs:
           # No host uvx in this job (no _setup-env); the dedicated Security
           # Audit check owns the uv-secure scan.
           RUN_ALL_SKIP_AUDIT: 1
+          # Fixed rather than the default adcp-innet-$$ (PID-derived, unknown
+          # to any later step) so the log-capture step below can reach the
+          # same compose project after run_all_tests.sh exits.
+          COMPOSE_PROJECT_NAME: ci-bdd-e2e-rest
+      # Diagnostic only, no behavior change: this job has intermittently failed
+      # with unrelated scenarios reporting INTERNAL_ERROR / HTTP 500 from
+      # account resolution and create_media_buy, on three separate runs with
+      # three disjoint failing-test sets and 0/428 failures on an identical
+      # local run each time (see PR #1689 review-triage notes). The live
+      # server here runs the delivery-webhook and media-buy-status background
+      # schedulers concurrently with foreground requests (not disabled in this
+      # compose file), and any OperationalError/DisconnectionError anywhere
+      # trips a process-wide 10s "unhealthy" circuit breaker
+      # (src/core/database/database_session.py) that fails every OTHER request
+      # in that window — a plausible single root cause for disjoint collateral
+      # failures. Captures the server's own traceback so the next occurrence
+      # is diagnosed from evidence instead of another blind re-run.
+      - name: Dump server logs on failure
+        if: failure()
+        run: |
+          docker compose -f docker-compose.e2e.yml -p ci-bdd-e2e-rest logs --no-color --timestamps adcp-server postgres 2>&1 | tail -2000 || true
       - name: Cleanup Docker services
         if: always()
-        run: docker compose -f docker-compose.e2e.yml down -v 2>/dev/null || true
+        run: docker compose -f docker-compose.e2e.yml -p ci-bdd-e2e-rest down -v 2>/dev/null || true
 
   admin-ui-tests:
     name: Admin UI Tests

.mypy-untyped-defs-baseline

@@ -1 +1 @@
-227
+216

.ruff-complexity-baseline

@@ -1,5 +1,5 @@
 {
-  "C901": 183,
-  "PLR0912": 134,
+  "C901": 180,
+  "PLR0912": 133,
   "PLR0915": 108
 }

.type-ignore-baseline

@@ -1 +1 @@
-63
+60

CLAUDE.md

@@ -124,7 +124,7 @@ AST-scanning tests enforce architecture invariants on every `make quality` run.
 
 ## AdCP Spec Version
 
-This project targets AdCP spec **3.1.0-beta.3** via the `adcp==5.7.0` Python SDK. See
+This project targets AdCP spec **3.1.1** via the `adcp==6.6.0` Python SDK. See
 [docs/adcp-spec-version.md](docs/adcp-spec-version.md) for the version mapping
 and bump procedure. The CI guard at `tests/unit/test_adcp_spec_version.py`
 fails on pin drift.

README.md

@@ -4,8 +4,8 @@ An open-source implementation of the [Ad Context Protocol (AdCP)](https://adcont
 sales-agent role, maintained under Prebid.org. It lets AI agents discover and buy advertising
 inventory through standardized MCP and A2A interfaces.
 
-> **Status: alpha.** A pre-1.0 implementation of a pre-release protocol
-> (AdCP 3.1.0-beta.3), under active development. APIs still change between
+> **Status: alpha.** A pre-1.0 implementation of AdCP 3.1.1,
+> under active development. APIs still change between
 > releases — `v2.0.0` introduced breaking changes — so pin a version and expect
 > to adapt on upgrade. It is one of several AdCP sales-agent implementations,
 > not a sole or canonical reference. The codebase is substantial and functional
@@ -22,9 +22,8 @@ The Prebid Sales Agent is a server that:
 
 ## AdCP Compatibility
 
-This implementation targets **AdCP spec version 3.1.0-beta.3** via the `adcp==5.7.0`
-Python SDK. That spec version is a **beta** — request/response shapes are not yet
-frozen, and SDK bumps can change them. See
+This implementation targets **AdCP spec version 3.1.1** via the `adcp==6.6.0`
+Python SDK. SDK bumps can change request/response shapes. See
 [docs/adcp-spec-version.md](docs/adcp-spec-version.md) for the SDK-to-spec mapping
 and bump procedure. The pin is enforced by a CI guard
 (`tests/unit/test_adcp_spec_version.py`), which fails on drift.
@@ -75,8 +74,8 @@ Publishers deploy their own sales agent. Choose based on your needs:
 
 **Docker is the fastest** - it bundles PostgreSQL and just works. Cloud platforms require separate database setup.
 
-Because this is alpha software tracking a beta protocol, pin the version you deploy
-and re-test after every upgrade — minor releases can carry breaking changes.
+Because this implementation is alpha software, pin the version you deploy and
+re-test after every upgrade — releases can carry breaking changes.
 
 ### After Deployment
 
@@ -284,8 +283,10 @@ async with client:
     products = await client.tools.get_products(brief="video ads for sports content")
 
     # 2. Book a media buy. Each package references a product and one of its
-    #    pricing options. `idempotency_key` is REQUIRED (16-255 chars) — reusing
-    #    the same key makes retries safe (returns the original buy, no duplicate).
+    #    pricing options. `idempotency_key` is REQUIRED and shape-validated
+    #    (16-255 chars). Retry an ambiguous request with the SAME key: completed
+    #    requests replay for the advertised 24-hour window, while active work
+    #    returns IDEMPOTENCY_IN_FLIGHT. Never mint a fresh key for a retry.
     result = await client.tools.create_media_buy(
         brand="acme.com",                     # domain shorthand for a BrandReference
         start_time="2026-08-01T00:00:00Z",    # ISO 8601, or "asap"

alembic/versions/a4d7e8c91f20_complete_idempotency_guarantees.py

@@ -0,0 +1,93 @@
+"""complete idempotency read and downstream reconciliation guarantees
+
+Revision ID: a4d7e8c91f20
+Revises: f3a1c92b47de
+Create Date: 2026-07-24 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "a4d7e8c91f20"
+down_revision: str | Sequence[str] | None = "f3a1c92b47de"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Enable anonymous read scopes, split admission classes, and add claims."""
+    op.alter_column("idempotency_attempts", "principal_id", existing_type=sa.String(length=50), nullable=True)
+    op.add_column(
+        "idempotency_attempts",
+        sa.Column("operation_class", sa.String(length=8), nullable=False, server_default="write"),
+    )
+    op.create_check_constraint(
+        "ck_idempotency_attempt_operation_class",
+        "idempotency_attempts",
+        "operation_class IN ('read', 'write')",
+    )
+    op.create_table(
+        "downstream_mutation_claims",
+        sa.Column("claim_id", sa.String(length=50), nullable=False),
+        sa.Column("tenant_id", sa.String(length=50), nullable=False),
+        sa.Column("principal_id", sa.String(length=50), nullable=False),
+        sa.Column("account_id", sa.String(length=100), nullable=True),
+        sa.Column("idempotency_key", sa.String(length=255), nullable=False),
+        sa.Column("provider", sa.String(length=50), nullable=False),
+        sa.Column("operation_key", sa.String(length=255), nullable=False),
+        sa.Column("downstream_request_id", sa.String(length=64), nullable=False),
+        sa.Column("request_hash", sa.String(length=64), nullable=False),
+        sa.Column("status", sa.String(length=16), nullable=False, server_default="planned"),
+        sa.Column("result_metadata", sa.JSON(), nullable=True),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.ForeignKeyConstraint(
+            ["tenant_id"],
+            ["tenants.tenant_id"],
+            name="fk_downstream_mutation_claims_tenant",
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["tenant_id", "principal_id"],
+            ["principals.tenant_id", "principals.principal_id"],
+            name="fk_downstream_mutation_claims_principal",
+            ondelete="CASCADE",
+        ),
+        sa.CheckConstraint(
+            "status IN ('planned', 'invoked', 'applied', 'unknown')",
+            name="ck_downstream_mutation_claim_status",
+        ),
+        sa.PrimaryKeyConstraint("claim_id"),
+    )
+    op.create_index(
+        "idx_downstream_mutation_claims_scope",
+        "downstream_mutation_claims",
+        ["tenant_id", "principal_id", "account_id", "idempotency_key", "provider", "operation_key"],
+        unique=True,
+        postgresql_nulls_not_distinct=True,
+    )
+    op.create_index(
+        "idx_downstream_mutation_claims_expires_at",
+        "downstream_mutation_claims",
+        ["expires_at"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Remove reconciliation claims and restore authenticated-only attempts."""
+    op.drop_index("idx_downstream_mutation_claims_expires_at", table_name="downstream_mutation_claims")
+    op.drop_index("idx_downstream_mutation_claims_scope", table_name="downstream_mutation_claims")
+    op.drop_table("downstream_mutation_claims")
+    op.execute("DELETE FROM idempotency_attempts WHERE principal_id IS NULL")
+    op.drop_constraint(
+        "ck_idempotency_attempt_operation_class",
+        "idempotency_attempts",
+        type_="check",
+    )
+    op.drop_column("idempotency_attempts", "operation_class")
+    op.alter_column("idempotency_attempts", "principal_id", existing_type=sa.String(length=50), nullable=False)