diff --git a/adcp/v3/legacypurchase/README.md b/adcp/v3/legacypurchase/README.md new file mode 100644 index 0000000..0e4f5a5 --- /dev/null +++ b/adcp/v3/legacypurchase/README.md @@ -0,0 +1,143 @@ +# adcp/v3/legacypurchase + +Durable coordinator for redeeming a deprecated AdCP 3.2 `products_available` +`legacy_create` purchase continuation — the Go SDK-local equivalent of the +protocol's `continueLegacyPurchase(CompatibilityPurchaseCoordinatorInput)`. + +Spec: [`specs/legacy-compact-lifecycle-compatibility.md`](https://github.com/adcontextprotocol/adcp/blob/main/specs/legacy-compact-lifecycle-compatibility.md) +("Products-only brief compatibility" / `legacy_create`), added by +[adcp#6733](https://github.com/adcontextprotocol/adcp/pull/6733) and shipped +in the `3.2.0-beta.9` schema bundle this SDK pins at +[`adcp/v3/schemas/VERSION`](../schemas/VERSION). Tracks issue +[adcp-go#466](https://github.com/adcontextprotocol/adcp-go/issues/466). + +## What this is for + +AdCP 3.2 splits legacy `get_products`-with-a-brief compound purchase flows +into compact lifecycle tasks. When an application's compatibility +projection layer can only offer the caller a `products_available` outcome +with a `legacy_create` continuation (no truthful account-scoped feed/pricing +fence is available), that continuation is single-use, principal- and +account-bound, and short-lived. This package is the durable coordinator +that: + +- records the continuation's bound facts the moment it is offered + (`Store.RegisterContinuation`); +- validates a later redemption attempt against every one of those facts, + atomically claims the continuation exactly once, invokes your legacy + `create_media_buy` call, and durably records the terminal outcome so an + exact `idempotency_key` retry returns the same result instead of buying + twice (`Store.ContinueLegacyPurchase`). + +```go +store := legacypurchase.New(legacypurchase.Options{ + Backend: legacypurchase.NewMemoryBackend(time.Minute, 24*time.Hour), +}) + +// At the moment your compatibility projection decides to offer +// products_available with a legacy_create continuation: +err := store.RegisterContinuation(ctx, &legacypurchase.Continuation{ + Token: continuationToken, + Principal: authenticatedPrincipalID, + Account: account, + SourceADCPVersion: "3.0", + ExpiresAt: time.Now().Add(5 * time.Minute), + ProductIDs: []string{"seller-product-1"}, + Losses: []string{legacypurchase.LossFeedVersionNotAtomic, legacypurchase.LossPricingVersionNotAtomic}, + ObservedPayload: observedProductsJSON, // the compact_projection.products payload actually returned +}) + +// Later, when the caller redeems it: +ctx = idempotency.WithPrincipal(ctx, authenticatedPrincipalID) +result, err := store.ContinueLegacyPurchase(ctx, input, func(ctx context.Context, legacyReq json.RawMessage) ([]byte, error) { + return callLegacySellerCreateMediaBuy(ctx, legacyReq) // exactly-once +}) +``` + +## Scope + +Implemented and tested: + +- `Store.RegisterContinuation` / `Store.ContinueLegacyPurchase` — the + coordinator API. +- `Backend` — the pluggable durable-store interface. +- `MemoryBackend` — a fully concurrency-safe in-process reference + implementation. +- Every binding check the spec states: principal, account (including the + `legacy_create_request`-carried account field cross-check, with AdCP + 2.5's no-wire-account-field carve-out), expiry, exact loss-set + acceptance, selected-product-ID subset-and-equality against the + request's explicit packages. +- Atomic single-use claim, proven under `-race` with concurrent distinct + `idempotency_key`s racing the same token (`store_race_test.go`). +- Deterministic replay: an exact retry (same `idempotency_key`, same + payload) after success or terminal failure returns the recorded result, + never a fresh `Executor` call. +- Fail-closed crash reconciliation: a claim left `StatePending` past + `Options.PendingLeaseTimeout` returns `AmbiguousClaimError` with recovery + guidance rather than being silently retried or silently expiring. +- The `products-only-brief-compatibility` vectors from the AdCP 3.2 schema + bundle (`testdata/products-only-brief-compatibility/`, see its + `PROVENANCE.md`), run end to end in `vectors_test.go`, plus the negative + cases (product substitution, package-selection drift, incomplete/stale + loss consent, wrong account, expiry) the vector bundle's own README + documents as SDK-suite-constructed. + +Deliberately deferred — disclosed, not silently dropped: + +1. **A persistent (e.g. Postgres) `Backend`.** This package ships the + interface and `MemoryBackend` only, matching how + [`adcp/v3/idempotency`](../idempotency)'s Postgres adapter and + [`adcp/v3/signing/pgreplay`](../signing) shipped as their own follow-on + work. Tracked in + [adcp-go#482](https://github.com/adcontextprotocol/adcp-go/issues/482). +2. **Full per-source-version (2.5/3.0/3.1) `create_media_buy` request + schema validation.** This package enforces the structural rules tied + directly to atomicity and single-use-claim safety (explicit-package + mode, exact package-product-ID match, the account cross-check), not a + complete replica of each legacy version's request schema — `adcp/v3` is + an AdCP 3.x-only module and does not vendor those schemas. **Your + `Executor` remains responsible for full legacy-request validation** + before or as part of calling the real legacy seller. +3. **The reverse compact-seller → legacy-buyer server-side facade** (the + spec's "Established buyers against a compact-backed seller" section, + and `vectors.json`'s `reverse_compatibility_cases`). Materially separate + scope from this buyer-side coordinator — tracked in the same + [adcp-go#482](https://github.com/adcontextprotocol/adcp-go/issues/482). + +Not a gap: `listed_purchase` continuations pass seller-issued, +account-scoped feed/pricing values straight into ordinary `buy_products` +per the spec — there is no durable continuation state for this coordinator +to claim on that path. + +## Migration guidance — what remains application-owned + +- **Minting the continuation.** This package does not decide *when* to + offer `products_available`/`legacy_create` versus a native + `request_proposals` result, or compute `ObservedPayload` — that is your + compatibility-projection logic, per the spec's classification rules. + `RegisterContinuation` only requires the result to be complete + (non-empty `ObservedPayload`, both required atomic-fence losses present). +- **Legacy request construction and validation.** Building + `legacy_create_request` for the caller's selected products, and + validating it in full against the source version's actual + `create_media_buy` schema, stays application-owned (see deferred item 2 + above). This package only enforces explicit-package mode and product-ID + agreement before calling your `Executor`. +- **The actual legacy seller call.** `Executor` is where you place the real + HTTP/MCP call to the legacy seller (or your own legacy facade). This + package guarantees it runs at most once per claimed continuation; it does + not implement retries, timeouts, or transport concerns for that call — + bring your own `http.Client` / MCP client conventions. +- **Crash reconciliation.** `AmbiguousClaimError` tells you a claim's + outcome is unknown; *how* to reconcile it (e.g. querying the legacy + seller's own idempotent `get_media_buys` for a buyer-ref/idempotency + marker your `Executor` embedded) is necessarily seller-specific and stays + application-owned. Embed a stable, discoverable marker in your + `legacy_create_request` (e.g. `buyer_ref`) specifically so this + reconciliation is possible. +- **Backend durability and retention.** `MemoryBackend` is process-local — + a restart loses in-flight and recent continuation state. Production + deployments needing cross-restart or cross-instance durability need a + persistent `Backend` (deferred item 1 above) or their own implementation + of the `Backend` interface. diff --git a/adcp/v3/legacypurchase/backend.go b/adcp/v3/legacypurchase/backend.go new file mode 100644 index 0000000..692edf8 --- /dev/null +++ b/adcp/v3/legacypurchase/backend.go @@ -0,0 +1,54 @@ +package legacypurchase + +import ( + "context" + "time" +) + +// Backend is the pluggable durable-store surface for legacy_create +// continuation coordination. Implementations MUST be safe for concurrent +// use, and MUST implement PutContinuation, ClaimPending, CompletePending, +// and FailPending as atomic operations against their storage engine (e.g. a +// unique constraint on Token plus an `UPDATE ... WHERE token = $1 AND +// state = 'offered'` for ClaimPending, mirroring +// adcp/v3/idempotency.Backend's PutIfAbsent contract). +// +// This is a three-state claim/commit/fail FSM rather than idempotency's +// single PutIfAbsent because redeeming a legacy_create continuation is a +// two-phase operation: claim the token, then call an external legacy +// seller whose outcome is not known at claim time. The atomic-claim and +// record-outcome steps cannot be the same write — a crash between them +// must be observable as StatePending, not silently lost (double-purchase +// risk) or silently fabricated as a result. +type Backend interface { + // PutContinuation durably records a newly offered continuation. + // Returns *DuplicateTokenError if rec.Token already exists. + PutContinuation(ctx context.Context, rec *ContinuationRecord) error + + // GetContinuation returns the current record for token, or (nil, nil) + // on miss. + GetContinuation(ctx context.Context, token string) (*ContinuationRecord, error) + + // ClaimPending atomically transitions token from StateOffered to + // StatePending, recording claimantKey (the idempotency_key redeeming + // it) and requestHash (a canonical hash of the full redemption input, + // for retry-conflict detection). claimed=false means the token was not + // in StateOffered at the time of the attempt — it does not exist, or + // it has already been claimed by this or another idempotency_key. The + // returned rec is the current record either way, so the caller can + // proceed without a second round trip. + ClaimPending(ctx context.Context, token, claimantKey, requestHash string, claimedAt time.Time) (rec *ContinuationRecord, claimed bool, err error) + + // CompletePending atomically transitions a StatePending record claimed + // by claimantKey to StateCommitted, storing result. ok=false if the + // record is not currently StatePending and claimed by claimantKey + // (defensive — should not happen absent a caller bug or a second + // coordinator instance racing on the same token, which ClaimPending's + // atomicity already prevents). + CompletePending(ctx context.Context, token, claimantKey string, result []byte, completedAt time.Time) (rec *ContinuationRecord, ok bool, err error) + + // FailPending is CompletePending's terminal-failure counterpart: + // StatePending -> StateFailed, recording errCode/errMessage as + // recovery guidance for the caller. + FailPending(ctx context.Context, token, claimantKey, errCode, errMessage string, failedAt time.Time) (rec *ContinuationRecord, ok bool, err error) +} diff --git a/adcp/v3/legacypurchase/doc.go b/adcp/v3/legacypurchase/doc.go new file mode 100644 index 0000000..df93dfb --- /dev/null +++ b/adcp/v3/legacypurchase/doc.go @@ -0,0 +1,99 @@ +// Package legacypurchase implements the SDK-local durable coordinator for +// redeeming a deprecated AdCP 3.2 "products_available" legacy_create +// purchase continuation — the Go SDK-local equivalent of the protocol +// contract's continueLegacyPurchase(CompatibilityPurchaseCoordinatorInput). +// +// # Source of the contract +// +// This package implements the normative rules in +// specs/legacy-compact-lifecycle-compatibility.md ("Products-only brief +// compatibility" / "legacy_create") from adcontextprotocol/adcp, added by +// https://github.com/adcontextprotocol/adcp/pull/6733 (merged 2026-08-20, +// resolving https://github.com/adcontextprotocol/adcp/issues/6716) and +// shipped in the AdCP 3.2.0-beta.9 schema bundle this SDK pins at +// adcp/v3/schemas/VERSION. CompatibilityPurchaseCoordinatorInput's field +// shape mirrors media-buy/legacy-purchase-continuation-input.json from that +// bundle exactly. +// +// The Go type is hand-written rather than generated: the schema is marked +// x-adcp-sdk-local (it is explicitly not an AdCP wire tool — "the input +// object MUST NOT be sent to the seller") and is not reachable by $ref from +// any tool request/response schema, so adcp/v3/schemas/generate.py's +// auto-discovery does not produce a Go type for it. Verified by downloading +// the pinned 3.2.0-beta.9 bundle and regenerating adcp/v3/types_gen.go +// locally: no CompatibilityPurchaseCoordinatorInput or PurchaseContinuation +// type is emitted, confirming this is by design rather than a generator +// gap. +// +// # Design precedent +// +// This coordinator follows the same claim-once, pluggable-Backend shape as +// adcp/v3/idempotency (PutIfAbsent-based replay) and +// adcp/v3/signing/pgreplay (atomic-insert replay-cap enforcement) — the two +// existing "durable, pluggable, atomic-claim" packages in this codebase. +// It widens that shape from a single PutIfAbsent into an explicit +// three-state FSM (StateOffered -> StatePending -> StateCommitted or +// StateFailed) because a legacy purchase continuation is a genuinely +// two-phase operation: claim the token, then execute an external legacy +// create_media_buy call whose outcome is not yet known at claim time. A +// crash between those two steps must be *observable* as StatePending +// rather than silently vanishing (double-purchase risk) or silently +// resolving on its own (a fabricated result) — see AmbiguousClaimError. +// +// # Scope of this package +// +// Implemented and tested here: +// - Store.RegisterContinuation: durably records a legacy_create +// continuation's bound facts (principal, account, source version, +// expiry, product IDs, required loss set, and the observed +// product/pricing payload it was minted against) at the moment an +// application's compatibility-projection layer decides to offer one. +// - Store.ContinueLegacyPurchase: validates a +// CompatibilityPurchaseCoordinatorInput against every binding rule the +// spec states (principal, account, expiry, exact loss-set acceptance, +// selected-product-ID subset-and-equality against the legacy_create +// request's explicit packages), atomically claims the token exactly +// once, invokes a caller-supplied Executor, and durably records the +// terminal result so an exact idempotency_key retry returns the +// deterministic prior result rather than re-invoking Executor or the +// legacy seller. +// - Backend: the pluggable durable-store interface, plus MemoryBackend, +// a fully concurrency-safe in-process reference implementation. +// +// Deliberately deferred, disclosed rather than silently omitted: +// +// 1. A persistent (e.g. Postgres) Backend implementation. This package +// ships the interface and a well-tested in-memory reference backend +// only, mirroring how adcp/v3/idempotency's Postgres adapter and +// adcp/v3/signing/pgreplay shipped as follow-on work after their +// respective in-memory/interface PRs. A distributed Backend is real, +// separable work (schema design, connection lifecycle, a real +// concurrency proof against a live database) — tracked in +// https://github.com/adcontextprotocol/adcp-go/issues/482. +// +// 2. Full per-source-version (AdCP 2.5 / 3.0 / 3.1) JSON Schema validation +// of legacy_create_request against each version's actual +// create-media-buy-request schema. This package enforces the +// structural invariants the spec ties directly to atomicity and +// single-use-claim safety — explicit-package mode, and that the +// request's distinct package product IDs exactly equal +// selected_product_ids — plus the account cross-check when the source +// version's request schema carries an account field. It does not +// replicate each legacy version's full request schema, which this SDK +// (adcp/v3, an AdCP 3.x-only module) does not vendor. Application code +// remains responsible for full legacy-version request validation +// before or inside its Executor. +// +// 3. The reverse compact-seller -> legacy-buyer server-side facade (the +// spec's "Established buyers against a compact-backed seller" +// section). This is seller-side adapter work — preserving a 3.2 +// seller's deprecated get_products/create_media_buy facades for +// established buyers — a materially different and separable scope from +// the buyer-side coordinator this package implements. Tracked in the +// same https://github.com/adcontextprotocol/adcp-go/issues/482. +// +// listed_purchase is not a gap: per the spec, a listed_purchase +// continuation carries a seller-issued, account-scoped feed/pricing fence +// straight into ordinary buy_products, with no durable continuation state +// to claim. There is nothing for this coordinator to do on that path. +package legacypurchase diff --git a/adcp/v3/legacypurchase/errors.go b/adcp/v3/legacypurchase/errors.go new file mode 100644 index 0000000..a138ada --- /dev/null +++ b/adcp/v3/legacypurchase/errors.go @@ -0,0 +1,258 @@ +package legacypurchase + +import "fmt" + +// Error codes drawn from AdCP's shared enums/error-code.json vocabulary +// where a direct match exists (this package is SDK-local — its errors are +// never sent over the wire, but reusing the shared vocabulary keeps +// application-level error mapping consistent with the rest of the SDK). +const ( + CodeInvalidRequest = "INVALID_REQUEST" + CodeValidationError = "VALIDATION_ERROR" + CodeConflict = "CONFLICT" + CodeIdempotencyConflict = "IDEMPOTENCY_CONFLICT" + CodeIdempotencyInFlight = "IDEMPOTENCY_IN_FLIGHT" +) + +// InvalidInputError is returned when a CompatibilityPurchaseCoordinatorInput +// or a Continuation registration fails structural validation (missing +// required field, malformed idempotency_key, legacy_create_request not +// valid JSON or not explicit-package mode, etc). +type InvalidInputError struct { + Field string + Reason string +} + +func (e *InvalidInputError) Error() string { + if e.Field == "" { + return "legacypurchase: invalid input: " + e.Reason + } + return fmt.Sprintf("legacypurchase: invalid input at %s: %s", e.Field, e.Reason) +} + +// Code returns the protocol error code. +func (*InvalidInputError) Code() string { return CodeInvalidRequest } + +// DuplicateTokenError is returned by Backend.PutContinuation when the token +// already has a durable record. +type DuplicateTokenError struct{ Token string } + +func (e *DuplicateTokenError) Error() string { + return "legacypurchase: continuation token already registered: " + logToken(e.Token) +} + +// Code returns the protocol error code. +func (*DuplicateTokenError) Code() string { return CodeConflict } + +// NotFoundError is returned when continuation_token has no durable record — +// unknown, or already swept past its retention window. +type NotFoundError struct{ Token string } + +func (e *NotFoundError) Error() string { + return "legacypurchase: unknown continuation token: " + logToken(e.Token) +} + +// Code returns the protocol error code. +func (*NotFoundError) Code() string { return CodeInvalidRequest } + +// ExpiredError is returned when a redemption attempt arrives after the +// continuation's continuation_expires_at. +type ExpiredError struct { + Token string + ExpiresAt string +} + +func (e *ExpiredError) Error() string { + return fmt.Sprintf("legacypurchase: continuation %s expired at %s", logToken(e.Token), e.ExpiresAt) +} + +// Code returns the protocol error code. +func (*ExpiredError) Code() string { return CodeValidationError } + +// PrincipalMismatchError is returned when the authenticated principal +// resolving the token differs from the principal it was bound to at +// registration — the spec's confused-deputy guard. +type PrincipalMismatchError struct{ Token string } + +func (e *PrincipalMismatchError) Error() string { + return "legacypurchase: continuation " + logToken(e.Token) + " is not bound to the authenticated principal" +} + +// Code returns the protocol error code. +func (*PrincipalMismatchError) Code() string { return CodeValidationError } + +// AccountMismatchError is returned when the input's account (or, when the +// source version's create_media_buy request carries one, the request's own +// account field) does not equal the token-bound account. +type AccountMismatchError struct { + Token string + Reason string +} + +func (e *AccountMismatchError) Error() string { + return "legacypurchase: continuation " + logToken(e.Token) + " account mismatch: " + e.Reason +} + +// Code returns the protocol error code. +func (*AccountMismatchError) Code() string { return CodeValidationError } + +// ProductSelectionError is returned when selected_product_ids is not a +// non-empty subset of the token-bound product IDs, or does not equal the +// distinct explicit-package product IDs in legacy_create_request. +type ProductSelectionError struct { + Token string + Reason string +} + +func (e *ProductSelectionError) Error() string { + return "legacypurchase: continuation " + logToken(e.Token) + " product selection invalid: " + e.Reason +} + +// Code returns the protocol error code. +func (*ProductSelectionError) Code() string { return CodeValidationError } + +// PricingSelectionError is returned when a legacy_create_request package +// names a pricing_option_id not present among the continuation's observed +// pricing options for that product — a substituted price the seller never +// actually offered against this continuation. +type PricingSelectionError struct { + Token string + Reason string +} + +func (e *PricingSelectionError) Error() string { + return "legacypurchase: continuation " + logToken(e.Token) + " pricing selection invalid: " + e.Reason +} + +// Code returns the protocol error code. +func (*PricingSelectionError) Code() string { return CodeValidationError } + +// LossAcceptanceError is returned when accepted_losses is not exactly equal +// to the continuation's declared loss set — missing, extra, or stale +// consent. +type LossAcceptanceError struct { + Token string + Required []string + Accepted []string +} + +func (e *LossAcceptanceError) Error() string { + return fmt.Sprintf("legacypurchase: continuation %s loss acceptance mismatch: required %v, got %v", + logToken(e.Token), e.Required, e.Accepted) +} + +// Code returns the protocol error code. +func (*LossAcceptanceError) Code() string { return CodeValidationError } + +// ExplicitPackageModeError is returned when legacy_create_request does not +// use explicit-package mode (a non-empty packages[] array where every +// package names a product_id). +type ExplicitPackageModeError struct{ Reason string } + +func (e *ExplicitPackageModeError) Error() string { + return "legacypurchase: legacy_create_request must use explicit-package mode: " + e.Reason +} + +// Code returns the protocol error code. +func (*ExplicitPackageModeError) Code() string { return CodeValidationError } + +// AlreadyClaimedError is returned when a continuation is no longer +// StateOffered and the redeeming idempotency_key does not match the +// claimant on record — the single-use guard. It is also returned for a +// terminal (StateCommitted/StateFailed) record reused with a different +// idempotency_key, and carries the claim's terminal state for the caller's +// diagnostics. +type AlreadyClaimedError struct { + Token string + State ContinuationState +} + +func (e *AlreadyClaimedError) Error() string { + return fmt.Sprintf("legacypurchase: continuation %s already claimed (state=%s)", logToken(e.Token), e.State) +} + +// Code returns the protocol error code. +func (*AlreadyClaimedError) Code() string { return CodeIdempotencyConflict } + +// RequestConflictError is returned when idempotency_key is reused with a +// different canonicalized redemption payload — the same class of conflict +// adcp/v3/idempotency.ConflictError reports for mutating tool calls. +type RequestConflictError struct { + Token string + IdempotencyKey string +} + +func (e *RequestConflictError) Error() string { + return "legacypurchase: idempotency_key " + logToken(e.IdempotencyKey) + " reused with a different payload for continuation " + logToken(e.Token) +} + +// Code returns the protocol error code. +func (*RequestConflictError) Code() string { return CodeIdempotencyConflict } + +// InFlightError is returned when the same idempotency_key's earlier claim +// is still within its pending lease window — the Executor call is very +// likely still running in another goroutine or process. Distinct from +// AmbiguousClaimError: this is an ordinary "retry shortly" condition, not a +// crash-recovery one. +type InFlightError struct { + Token string + ClaimedAt string + RetryAfter string +} + +func (e *InFlightError) Error() string { + return fmt.Sprintf("legacypurchase: continuation %s claim is in flight (claimed_at=%s); retry after %s", + logToken(e.Token), e.ClaimedAt, e.RetryAfter) +} + +// Code returns the protocol error code. +func (*InFlightError) Code() string { return CodeIdempotencyInFlight } + +// AmbiguousClaimError is returned when a continuation's claim has been +// StatePending for longer than the configured pending-lease timeout: the +// process that claimed it most likely crashed between claiming the token +// and recording a terminal outcome, so whether the legacy seller actually +// received the create_media_buy call is unknown. Per spec, this MUST fail +// closed rather than be silently retried — Guidance names the concrete +// recovery step. +type AmbiguousClaimError struct { + Token string + ClaimedAt string + Guidance string +} + +func (e *AmbiguousClaimError) Error() string { + return fmt.Sprintf("legacypurchase: continuation %s has an ambiguous crash-suspected claim from %s: %s", + logToken(e.Token), e.ClaimedAt, e.Guidance) +} + +// Code returns the protocol error code. +func (*AmbiguousClaimError) Code() string { return CodeConflict } + +// TerminalFailureError is returned on an exact retry of a redemption whose +// Executor call previously failed terminally (StateFailed). It carries the +// recorded failure so the caller does not need a side channel to learn why. +type TerminalFailureError struct { + Token string + ErrCode string + Message string +} + +func (e *TerminalFailureError) Error() string { + return fmt.Sprintf("legacypurchase: continuation %s previously failed terminally [%s]: %s", + logToken(e.Token), e.ErrCode, e.Message) +} + +// Code returns the protocol error code. +func (e *TerminalFailureError) Code() string { return e.ErrCode } + +// logToken returns a prefix-truncated form of a token/key safe for default +// logging, mirroring adcp/v3/idempotency.LogKey — full tokens are replay +// oracles. +func logToken(s string) string { + const n = 8 + if len(s) <= n { + return "****" + } + return s[:n] + "…" +} diff --git a/adcp/v3/legacypurchase/memory.go b/adcp/v3/legacypurchase/memory.go new file mode 100644 index 0000000..4de4caa --- /dev/null +++ b/adcp/v3/legacypurchase/memory.go @@ -0,0 +1,194 @@ +package legacypurchase + +import ( + "context" + "sync" + "time" +) + +// MemoryBackend is an in-process Backend suitable for tests and reference +// servers, mirroring adcp/v3/idempotency.MemoryBackend's shape and +// concurrency guarantees. All state transitions are guarded by one mutex, +// so ClaimPending/CompletePending/FailPending are trivially atomic; a +// distributed Backend (e.g. Postgres, deliberately not shipped in this PR — +// see doc.go) would use row-level locking or a conditional UPDATE to get +// the same guarantee across processes. +// +// A background sweeper removes StateOffered records past ExpiresAt and +// terminal (StateCommitted/StateFailed) records past a configurable +// retention window. StatePending records are never swept — an ambiguous, +// crash-suspected claim must be reconciled, not silently vanish. +type MemoryBackend struct { + mu sync.Mutex + records map[string]*ContinuationRecord + clock func() time.Time + + terminalRetention time.Duration + + stop chan struct{} + stopped chan struct{} +} + +// NewMemoryBackend returns a MemoryBackend. sweepInterval controls how +// often expired StateOffered records and terminal records older than +// terminalRetention are removed; zero disables the background sweeper +// (expired StateOffered records still fail closed via ExpiredError at +// redemption time — the sweeper only reclaims memory). terminalRetention +// of zero keeps terminal records forever (until the sweeper is disabled or +// Close is called). +func NewMemoryBackend(sweepInterval, terminalRetention time.Duration) *MemoryBackend { + return newMemoryBackend(sweepInterval, terminalRetention, time.Now) +} + +func newMemoryBackend(sweepInterval, terminalRetention time.Duration, clock func() time.Time) *MemoryBackend { + b := &MemoryBackend{ + records: map[string]*ContinuationRecord{}, + clock: clock, + terminalRetention: terminalRetention, + stop: make(chan struct{}), + stopped: make(chan struct{}), + } + if sweepInterval > 0 { + go b.sweepLoop(sweepInterval) + } else { + close(b.stopped) + } + return b +} + +// Close stops the background sweeper. +func (b *MemoryBackend) Close() { + select { + case <-b.stop: + return + default: + close(b.stop) + } + <-b.stopped +} + +func (b *MemoryBackend) sweepLoop(interval time.Duration) { + defer close(b.stopped) + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-b.stop: + return + case <-t.C: + b.sweep() + } + } +} + +func (b *MemoryBackend) sweep() { + now := b.clock() + b.mu.Lock() + defer b.mu.Unlock() + for token, rec := range b.records { + switch rec.State { + case StateOffered: + if now.After(rec.ExpiresAt) { + delete(b.records, token) + } + case StateCommitted, StateFailed: + if b.terminalRetention > 0 && now.After(rec.CompletedAt.Add(b.terminalRetention)) { + delete(b.records, token) + } + case StatePending: + // Never swept — see type doc. + } + } +} + +// cloneRecord returns a deep copy of rec: besides copying the struct itself, +// it copies the backing arrays of every mutable slice/byte-slice field +// (ProductIDs, Losses, ObservedPayload, Result) so the clone shares no +// memory with rec. Every accessor below returns a clone rather than a +// pointer sharing rec's slices — otherwise a caller mutating a byte in a +// returned record's Result (say) would silently corrupt what this backend +// has stored, or a caller later mutating a slice it handed to +// PutContinuation would silently corrupt state already durably recorded. +func cloneRecord(rec *ContinuationRecord) *ContinuationRecord { + cp := *rec + cp.ProductIDs = append([]string(nil), rec.ProductIDs...) + cp.Losses = append([]string(nil), rec.Losses...) + cp.ObservedPayload = append([]byte(nil), rec.ObservedPayload...) + cp.Result = append([]byte(nil), rec.Result...) + return &cp +} + +// PutContinuation implements Backend. +func (b *MemoryBackend) PutContinuation(_ context.Context, rec *ContinuationRecord) error { + b.mu.Lock() + defer b.mu.Unlock() + if _, exists := b.records[rec.Token]; exists { + return &DuplicateTokenError{Token: rec.Token} + } + b.records[rec.Token] = cloneRecord(rec) + return nil +} + +// GetContinuation implements Backend. +func (b *MemoryBackend) GetContinuation(_ context.Context, token string) (*ContinuationRecord, error) { + b.mu.Lock() + defer b.mu.Unlock() + rec, ok := b.records[token] + if !ok { + return nil, nil + } + return cloneRecord(rec), nil +} + +// ClaimPending implements Backend. +func (b *MemoryBackend) ClaimPending(_ context.Context, token, claimantKey, requestHash string, claimedAt time.Time) (*ContinuationRecord, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + rec, ok := b.records[token] + if !ok { + return nil, false, nil + } + if rec.State != StateOffered { + return cloneRecord(rec), false, nil + } + rec.State = StatePending + rec.ClaimantKey = claimantKey + rec.RequestHash = requestHash + rec.ClaimedAt = claimedAt + return cloneRecord(rec), true, nil +} + +// CompletePending implements Backend. +func (b *MemoryBackend) CompletePending(_ context.Context, token, claimantKey string, result []byte, completedAt time.Time) (*ContinuationRecord, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + rec, ok := b.records[token] + if !ok || rec.State != StatePending || rec.ClaimantKey != claimantKey { + if ok { + return cloneRecord(rec), false, nil + } + return nil, false, nil + } + rec.State = StateCommitted + rec.Result = append([]byte(nil), result...) + rec.CompletedAt = completedAt + return cloneRecord(rec), true, nil +} + +// FailPending implements Backend. +func (b *MemoryBackend) FailPending(_ context.Context, token, claimantKey, errCode, errMessage string, failedAt time.Time) (*ContinuationRecord, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + rec, ok := b.records[token] + if !ok || rec.State != StatePending || rec.ClaimantKey != claimantKey { + if ok { + return cloneRecord(rec), false, nil + } + return nil, false, nil + } + rec.State = StateFailed + rec.ErrorCode = errCode + rec.ErrorMessage = errMessage + rec.CompletedAt = failedAt + return cloneRecord(rec), true, nil +} diff --git a/adcp/v3/legacypurchase/memory_test.go b/adcp/v3/legacypurchase/memory_test.go new file mode 100644 index 0000000..78fc735 --- /dev/null +++ b/adcp/v3/legacypurchase/memory_test.go @@ -0,0 +1,59 @@ +package legacypurchase + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestMemoryBackend_ReturnedRecordMutationDoesNotCorruptStore proves every +// Backend accessor returns a record isolated from MemoryBackend's stored +// copy: mutating a byte in a caller's returned record — Result or +// ObservedPayload — must not change what a later call observes. +func TestMemoryBackend_ReturnedRecordMutationDoesNotCorruptStore(t *testing.T) { + now := time.Now().UTC() + b := newMemoryBackend(0, 0, func() time.Time { return now }) + ctx := context.Background() + + rec := &ContinuationRecord{ + Continuation: Continuation{ + Token: "continuation-token-0123456789", + Principal: "buyer-agent-1", + ObservedPayload: []byte(`[{"product_id":"prod-a"}]`), + }, + State: StateOffered, + } + require.NoError(t, b.PutContinuation(ctx, rec)) + + // Mutating the caller's own rec after Put must not reach the store. + rec.ObservedPayload[0] = 'X' + + got, err := b.GetContinuation(ctx, rec.Token) + require.NoError(t, err) + assert.JSONEq(t, `[{"product_id":"prod-a"}]`, string(got.ObservedPayload), "PutContinuation must not alias the caller's backing array") + + // Mutating a byte in a returned record must not corrupt the store. + got.ObservedPayload[0] = 'Y' + got2, err := b.GetContinuation(ctx, rec.Token) + require.NoError(t, err) + assert.JSONEq(t, `[{"product_id":"prod-a"}]`, string(got2.ObservedPayload), "GetContinuation must not alias its stored backing array") + + claimed, won, err := b.ClaimPending(ctx, rec.Token, "idem-1", "hash-1", now) + require.NoError(t, err) + require.True(t, won) + claimed.Result = []byte("mutate-me") + + completed, ok, err := b.CompletePending(ctx, rec.Token, "idem-1", []byte(`{"media_buy_id":"mb-1"}`), now) + require.NoError(t, err) + require.True(t, ok) + + // Changing a byte in the first replay response must not change what the + // next retry sees — the exact bug reported against this package. + completed.Result[0] = 'Z' + replayed, err := b.GetContinuation(ctx, rec.Token) + require.NoError(t, err) + assert.JSONEq(t, `{"media_buy_id":"mb-1"}`, string(replayed.Result), "a caller mutating a returned replay response must not change the next retry's response") +} diff --git a/adcp/v3/legacypurchase/store.go b/adcp/v3/legacypurchase/store.go new file mode 100644 index 0000000..3d77419 --- /dev/null +++ b/adcp/v3/legacypurchase/store.go @@ -0,0 +1,556 @@ +package legacypurchase + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "regexp" + "sort" + "time" + + adcp "github.com/adcontextprotocol/adcp-go/adcp/v3" + "github.com/adcontextprotocol/adcp-go/adcp/v3/idempotency" +) + +// DefaultPendingLeaseTimeout bounds how long a StatePending claim is +// treated as merely "in flight" (InFlightError, retry shortly) before it is +// instead treated as crash-suspected (AmbiguousClaimError, fail closed). +// Chosen generously relative to a typical legacy create_media_buy round +// trip; applications with slower legacy sellers should widen it via +// Options.PendingLeaseTimeout. +const DefaultPendingLeaseTimeout = 2 * time.Minute + +// Options configures a Store. +type Options struct { + // Backend stores continuation records. Required. + Backend Backend + + // PendingLeaseTimeout is the boundary between InFlightError and + // AmbiguousClaimError for a StatePending claim under the same + // idempotency_key. Defaults to DefaultPendingLeaseTimeout. + PendingLeaseTimeout time.Duration + + // Clock is injectable for tests. Defaults to time.Now.UTC. + Clock func() time.Time +} + +// Store coordinates durable legacy_create purchase continuations. +type Store struct { + opts Options +} + +// New returns a Store. Panics on misconfiguration, matching +// adcp/v3/idempotency.New's fail-fast convention: a coordinator that starts +// in a state where claims silently can't be recorded is worse than one that +// never starts. +func New(opts Options) *Store { + if opts.Backend == nil { + panic("legacypurchase: Options.Backend is required") + } + if opts.PendingLeaseTimeout < 0 { + panic("legacypurchase: Options.PendingLeaseTimeout must be non-negative") + } + if opts.PendingLeaseTimeout == 0 { + opts.PendingLeaseTimeout = DefaultPendingLeaseTimeout + } + if opts.Clock == nil { + opts.Clock = func() time.Time { return time.Now().UTC() } + } + return &Store{opts: opts} +} + +// RegisterContinuation durably records a newly minted legacy_create +// continuation. Call this from the application's compatibility-projection +// layer at the exact moment it decides to offer +// purchase_continuation.kind == "legacy_create" to a caller — before that +// outcome is returned over the wire, so a later ContinueLegacyPurchase call +// always has a durable record to resolve the token against. +func (s *Store) RegisterContinuation(ctx context.Context, c *Continuation) error { + if c == nil { + return &InvalidInputError{Reason: "continuation is required"} + } + if len(c.Token) < 16 { + return &InvalidInputError{Field: "token", Reason: "must be at least 16 characters"} + } + if c.Principal == "" { + return &InvalidInputError{Field: "principal", Reason: "is required"} + } + if !hasAccountIdentity(c.Account) { + return &InvalidInputError{Field: "account", Reason: "must set account_id, or brand+operator"} + } + if c.SourceADCPVersion == "" { + return &InvalidInputError{Field: "source_adcp_version", Reason: "is required"} + } + if c.ExpiresAt.IsZero() { + return &InvalidInputError{Field: "expires_at", Reason: "is required"} + } + if len(c.ProductIDs) == 0 { + return &InvalidInputError{Field: "product_ids", Reason: "must not be empty"} + } + if !containsAll(c.Losses, LossFeedVersionNotAtomic, LossPricingVersionNotAtomic) { + return &InvalidInputError{Field: "losses", Reason: "must include feed_version_not_atomic and pricing_version_not_atomic"} + } + if c.SourceADCPVersion == "2.5" && !containsAll(c.Losses, LossMutationIdempotencyNotGuaranteed) { + return &InvalidInputError{Field: "losses", Reason: "an AdCP 2.5 source continuation must also include mutation_idempotency_not_guaranteed — 2.5 has no mutation replay contract"} + } + if len(c.ObservedPayload) == 0 { + return &InvalidInputError{Field: "observed_payload", Reason: "must be non-empty — a continuation cannot be minted without a complete observed product/pricing payload"} + } + + rec := &ContinuationRecord{ + Continuation: *c, + State: StateOffered, + RegisteredAt: s.opts.Clock(), + } + return s.opts.Backend.PutContinuation(ctx, rec) +} + +// ContinueLegacyPurchase redeems a legacy_create continuation: validates +// input against every binding rule in +// specs/legacy-compact-lifecycle-compatibility.md, atomically claims the +// token exactly once, invokes exec, and durably records the terminal result +// so an exact idempotency_key retry returns the deterministic prior result +// instead of re-invoking exec or the legacy seller. +// +// The caller's context must carry a principal via +// idempotency.WithPrincipal — the same context key +// adcp/v3/idempotency.Store's middleware uses, so a seller that already +// wraps its handlers with idempotency gets principal binding here for free. +func (s *Store) ContinueLegacyPurchase(ctx context.Context, input *CompatibilityPurchaseCoordinatorInput, exec Executor) (*Result, error) { + if input == nil { + return nil, &InvalidInputError{Reason: "input is required"} + } + if exec == nil { + return nil, &InvalidInputError{Reason: "exec is required"} + } + if err := validateInputStructure(input); err != nil { + return nil, err + } + + principal := idempotency.PrincipalFromContext(ctx) + if principal == "" { + return nil, &InvalidInputError{Reason: "principal missing from context; call idempotency.WithPrincipal before invoking ContinueLegacyPurchase"} + } + + pkgProductIDs, err := explicitPackageProductIDs(input.LegacyCreateRequest) + if err != nil { + return nil, err + } + if !stringSetEqual(pkgProductIDs, input.SelectedProductIDs) { + return nil, &ProductSelectionError{ + Token: input.ContinuationToken, + Reason: fmt.Sprintf("selected_product_ids %v does not equal legacy_create_request's explicit package product IDs %v", sortedCopy(input.SelectedProductIDs), sortedCopy(pkgProductIDs)), + } + } + + reqHash, err := requestHash(input) + if err != nil { + return nil, err + } + + rec, err := s.opts.Backend.GetContinuation(ctx, input.ContinuationToken) + if err != nil { + return nil, err + } + if rec == nil { + return nil, &NotFoundError{Token: input.ContinuationToken} + } + + now := s.opts.Clock() + + switch rec.State { + case StateOffered: + if err := s.validateBinding(rec, input, principal, now); err != nil { + return nil, err + } + claimed, wonClaim, err := s.opts.Backend.ClaimPending(ctx, input.ContinuationToken, input.IdempotencyKey, reqHash, now) + if err != nil { + return nil, err + } + if !wonClaim { + // Lost the race to another concurrent redemption attempt. + // Re-evaluate against whatever the winner left behind, exactly + // as if this call had observed that state from the start. + return s.resolveNonOffered(ctx, claimed, input, principal, reqHash, now) + } + return s.execute(ctx, claimed, input, exec) + + default: + return s.resolveNonOffered(ctx, rec, input, principal, reqHash, now) + } +} + +// validateBinding runs every check the spec requires before a StateOffered +// continuation may be claimed. +func (s *Store) validateBinding(rec *ContinuationRecord, input *CompatibilityPurchaseCoordinatorInput, principal string, now time.Time) error { + if now.After(rec.ExpiresAt) { + return &ExpiredError{Token: rec.Token, ExpiresAt: rec.ExpiresAt.Format(time.RFC3339)} + } + if rec.Principal != principal { + return &PrincipalMismatchError{Token: rec.Token} + } + if !accountsEqual(rec.Account, input.Account) { + return &AccountMismatchError{Token: rec.Token, Reason: "input.account does not equal the token-bound account"} + } + if !stringSetEqual(rec.Losses, input.AcceptedLosses) { + return &LossAcceptanceError{Token: rec.Token, Required: sortedCopy(rec.Losses), Accepted: sortedCopy(input.AcceptedLosses)} + } + if !stringSetSubset(input.SelectedProductIDs, rec.ProductIDs) { + return &ProductSelectionError{ + Token: rec.Token, + Reason: fmt.Sprintf("selected_product_ids %v is not a subset of the token-bound product IDs %v", sortedCopy(input.SelectedProductIDs), sortedCopy(rec.ProductIDs)), + } + } + if err := validatePricingSelection(rec.Token, rec.ObservedPayload, input.LegacyCreateRequest); err != nil { + return err + } + if reqAccount, ok, err := requestAccount(input.LegacyCreateRequest); err != nil { + return err + } else if ok && !accountsEqual(reqAccount, rec.Account) { + // Source versions whose create_media_buy request carries an + // account field must agree with the token-bound account. AdCP 2.5 + // has no such field, so ok is false there and this check is + // skipped per the spec's explicit carve-out. + return &AccountMismatchError{Token: rec.Token, Reason: "legacy_create_request.account does not equal the token-bound account"} + } + return nil +} + +// resolveNonOffered handles a redemption attempt against a continuation +// that is StatePending, StateCommitted, or StateFailed — i.e. every case +// other than a fresh, winning claim. +func (s *Store) resolveNonOffered(ctx context.Context, rec *ContinuationRecord, input *CompatibilityPurchaseCoordinatorInput, principal, reqHash string, now time.Time) (*Result, error) { + if rec.Principal != principal { + // Reject before ClaimantKey/RequestHash so a token bound to another + // principal never leaks its pending/terminal state or replays a + // committed result to a caller it isn't bound to. + return nil, &PrincipalMismatchError{Token: rec.Token} + } + if rec.ClaimantKey != input.IdempotencyKey { + return nil, &AlreadyClaimedError{Token: rec.Token, State: rec.State} + } + // Same idempotency_key as the claimant — this is meant to be an exact + // retry. It is only exact if the payload matches too. + if rec.RequestHash != reqHash { + return nil, &RequestConflictError{Token: rec.Token, IdempotencyKey: input.IdempotencyKey} + } + + switch rec.State { + case StatePending: + if now.Sub(rec.ClaimedAt) > s.opts.PendingLeaseTimeout { + return nil, &AmbiguousClaimError{ + Token: rec.Token, + ClaimedAt: rec.ClaimedAt.Format(time.RFC3339), + Guidance: "the claiming process did not record a terminal outcome within the pending lease window; reconcile directly against the legacy seller (e.g. list media buys for this account/idempotency context) before treating this continuation as available again — it MUST NOT be retried automatically", + } + } + return nil, &InFlightError{ + Token: rec.Token, + ClaimedAt: rec.ClaimedAt.Format(time.RFC3339), + RetryAfter: s.opts.PendingLeaseTimeout.String(), + } + case StateCommitted: + return &Result{Response: rec.Result, Replayed: true}, nil + case StateFailed: + return nil, &TerminalFailureError{Token: rec.Token, ErrCode: rec.ErrorCode, Message: rec.ErrorMessage} + default: + // Unreachable: StateOffered is handled by the caller before + // resolveNonOffered is invoked, and these are the only states. + return nil, &AmbiguousClaimError{Token: rec.Token, Guidance: fmt.Sprintf("unrecognized continuation state %q", rec.State)} + } +} + +// execute runs exec exactly once for a freshly won claim and durably +// records its terminal outcome. +func (s *Store) execute(ctx context.Context, claimedRec *ContinuationRecord, input *CompatibilityPurchaseCoordinatorInput, exec Executor) (*Result, error) { + resp, execErr := exec(ctx, input.LegacyCreateRequest) + completedAt := s.opts.Clock() + if execErr != nil { + code, msg := classifyExecError(execErr) + if _, ok, err := s.opts.Backend.FailPending(ctx, claimedRec.Token, input.IdempotencyKey, code, msg, completedAt); err != nil { + return nil, err + } else if !ok { + // Could not record the failure — surface as ambiguous rather + // than pretending the failure was cleanly recorded. + return nil, &AmbiguousClaimError{ + Token: claimedRec.Token, + ClaimedAt: claimedRec.ClaimedAt.Format(time.RFC3339), + Guidance: "exec failed and the failure could not be durably recorded; reconcile directly before retrying", + } + } + return nil, execErr + } + if _, ok, err := s.opts.Backend.CompletePending(ctx, claimedRec.Token, input.IdempotencyKey, resp, completedAt); err != nil { + return nil, err + } else if !ok { + return nil, &AmbiguousClaimError{ + Token: claimedRec.Token, + ClaimedAt: claimedRec.ClaimedAt.Format(time.RFC3339), + Guidance: "exec succeeded but the result could not be durably recorded; reconcile directly against the legacy seller before treating this operation as unresolved", + } + } + return &Result{Response: resp, Replayed: false}, nil +} + +// classifyExecError extracts a (code, message) pair for FailPending's +// recovery guidance. Errors implementing an interface with a Code() string +// method (the convention every typed error in this SDK follows) contribute +// their code; everything else is recorded as a generic execution failure. +func classifyExecError(err error) (code, message string) { + type coder interface{ Code() string } + if c, ok := err.(coder); ok { + return c.Code(), err.Error() + } + return "LEGACY_CREATE_FAILED", err.Error() +} + +// ---- validation helpers ---- + +var uuidPattern = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`) + +func validateInputStructure(input *CompatibilityPurchaseCoordinatorInput) error { + if !uuidPattern.MatchString(input.IdempotencyKey) { + return &InvalidInputError{Field: "idempotency_key", Reason: "must be a UUID (schema: format uuid)"} + } + if len(input.ContinuationToken) < 16 { + return &InvalidInputError{Field: "continuation_token", Reason: "must be at least 16 characters"} + } + if !hasAccountIdentity(input.Account) { + return &InvalidInputError{Field: "account", Reason: "must set account_id, or brand+operator"} + } + if len(input.SelectedProductIDs) == 0 { + return &InvalidInputError{Field: "selected_product_ids", Reason: "must not be empty"} + } + if hasDuplicates(input.SelectedProductIDs) { + return &InvalidInputError{Field: "selected_product_ids", Reason: "must not contain duplicates"} + } + if len(input.AcceptedLosses) < 2 { + return &InvalidInputError{Field: "accepted_losses", Reason: "must contain at least 2 entries"} + } + if hasDuplicates(input.AcceptedLosses) { + return &InvalidInputError{Field: "accepted_losses", Reason: "must not contain duplicates"} + } + if !containsAll(input.AcceptedLosses, LossFeedVersionNotAtomic, LossPricingVersionNotAtomic) { + return &InvalidInputError{Field: "accepted_losses", Reason: "must contain feed_version_not_atomic and pricing_version_not_atomic (schema allOf/contains constraint)"} + } + for _, l := range input.AcceptedLosses { + if l != LossFeedVersionNotAtomic && l != LossPricingVersionNotAtomic && l != LossMutationIdempotencyNotGuaranteed { + return &InvalidInputError{Field: "accepted_losses", Reason: "unknown loss value: " + l} + } + } + if len(input.LegacyCreateRequest) == 0 || bytes.Equal(bytes.TrimSpace(input.LegacyCreateRequest), []byte("{}")) { + return &InvalidInputError{Field: "legacy_create_request", Reason: "must have at least one property (schema: minProperties 1)"} + } + return nil +} + +func hasAccountIdentity(a adcp.AccountReference) bool { + if a.AccountID != "" { + return true + } + return a.Brand != nil && a.Brand.Domain != "" && a.Operator != "" +} + +func containsAll(set []string, want ...string) bool { + m := make(map[string]bool, len(set)) + for _, s := range set { + m[s] = true + } + for _, w := range want { + if !m[w] { + return false + } + } + return true +} + +func hasDuplicates(items []string) bool { + seen := make(map[string]bool, len(items)) + for _, it := range items { + if seen[it] { + return true + } + seen[it] = true + } + return false +} + +func stringSetEqual(a, b []string) bool { + if len(a) != len(b) { + return false + } + sa, sb := sortedCopy(a), sortedCopy(b) + for i := range sa { + if sa[i] != sb[i] { + return false + } + } + return true +} + +func stringSetSubset(subset, superset []string) bool { + if len(subset) == 0 { + return false + } + m := make(map[string]bool, len(superset)) + for _, s := range superset { + m[s] = true + } + for _, s := range subset { + if !m[s] { + return false + } + } + return true +} + +func sortedCopy(items []string) []string { + out := append([]string(nil), items...) + sort.Strings(out) + return out +} + +func accountsEqual(a, b adcp.AccountReference) bool { + ab, errA := json.Marshal(a) + bb, errB := json.Marshal(b) + if errA != nil || errB != nil { + return false + } + return bytes.Equal(ab, bb) +} + +// explicitPackageProductIDs parses legacyCreateRequest and requires +// explicit-package mode: a non-empty top-level "packages" array where every +// entry has a non-empty string "product_id". Returns the distinct product +// IDs found, in the order first seen. +func explicitPackageProductIDs(legacyCreateRequest json.RawMessage) ([]string, error) { + var probe struct { + Packages []struct { + ProductID string `json:"product_id"` + } `json:"packages"` + } + if err := json.Unmarshal(legacyCreateRequest, &probe); err != nil { + return nil, &InvalidInputError{Field: "legacy_create_request", Reason: "not valid JSON: " + err.Error()} + } + if len(probe.Packages) == 0 { + return nil, &ExplicitPackageModeError{Reason: "legacy_create_request.packages must be a non-empty array"} + } + seen := map[string]bool{} + var ids []string + for i, p := range probe.Packages { + if p.ProductID == "" { + return nil, &ExplicitPackageModeError{Reason: fmt.Sprintf("packages[%d].product_id is required for explicit-package mode", i)} + } + if !seen[p.ProductID] { + seen[p.ProductID] = true + ids = append(ids, p.ProductID) + } + } + return ids, nil +} + +// validatePricingSelection checks every legacy_create_request package that +// names a pricing_option_id against the continuation's observed +// product/pricing payload — the spec's binding on the "complete observed +// product/pricing payload", separate from and in addition to the JSON +// structural checks validateInputStructure/explicitPackageProductIDs run. +// ObservedPayload is durably fixed at registration time, so this rejects a +// redemption that selects a pricing option the seller never actually +// offered for that product, even though the request's JSON shape is +// otherwise well-formed. Packages that omit pricing_option_id are not +// checked here — nothing was selected to compare. +func validatePricingSelection(token string, observedPayload, legacyCreateRequest json.RawMessage) error { + var req struct { + Packages []struct { + ProductID string `json:"product_id"` + PricingOptionID string `json:"pricing_option_id"` + } `json:"packages"` + } + if err := json.Unmarshal(legacyCreateRequest, &req); err != nil { + return &InvalidInputError{Field: "legacy_create_request", Reason: "not valid JSON: " + err.Error()} + } + + var observed []struct { + ProductID string `json:"product_id"` + PricingOptions []struct { + PricingOptionID string `json:"pricing_option_id"` + } `json:"pricing_options"` + } + if err := json.Unmarshal(observedPayload, &observed); err != nil { + return fmt.Errorf("legacypurchase: continuation %s observed payload: %w", logToken(token), err) + } + optionsByProduct := make(map[string]map[string]bool, len(observed)) + for _, p := range observed { + options := make(map[string]bool, len(p.PricingOptions)) + for _, po := range p.PricingOptions { + options[po.PricingOptionID] = true + } + optionsByProduct[p.ProductID] = options + } + + for _, pkg := range req.Packages { + if pkg.PricingOptionID == "" { + continue + } + if !optionsByProduct[pkg.ProductID][pkg.PricingOptionID] { + return &PricingSelectionError{ + Token: token, + Reason: fmt.Sprintf("pricing_option_id %q for product %q is not present in the continuation's observed product/pricing payload", pkg.PricingOptionID, pkg.ProductID), + } + } + } + return nil +} + +// requestAccount extracts a top-level "account" field from +// legacyCreateRequest, when present. AdCP 2.5's create_media_buy has no +// wire account field, so ok is false there by design — the spec's explicit +// carve-out that a 2.5 adapter relies on the token-bound session instead. +func requestAccount(legacyCreateRequest json.RawMessage) (adcp.AccountReference, bool, error) { + var probe struct { + Account *adcp.AccountReference `json:"account"` + } + if err := json.Unmarshal(legacyCreateRequest, &probe); err != nil { + return adcp.AccountReference{}, false, &InvalidInputError{Field: "legacy_create_request", Reason: "not valid JSON: " + err.Error()} + } + if probe.Account == nil { + return adcp.AccountReference{}, false, nil + } + return *probe.Account, true, nil +} + +// requestHash returns a stable, canonicalized hash of the logical +// redemption request, used to detect an idempotency_key reused with a +// different payload. selected_product_ids and accepted_losses are sorted +// before hashing since they are logically sets; legacy_create_request is +// run through idempotency.Canonicalize (JCS) so byte-level reordering of an +// equivalent JSON object does not register as a conflict. +func requestHash(input *CompatibilityPurchaseCoordinatorInput) (string, error) { + canon, err := idempotency.Canonicalize(input.LegacyCreateRequest) + if err != nil { + return "", &InvalidInputError{Field: "legacy_create_request", Reason: "not valid JSON: " + err.Error()} + } + acctBytes, err := json.Marshal(input.Account) + if err != nil { + return "", &InvalidInputError{Field: "account", Reason: "could not encode: " + err.Error()} + } + + h := sha256.New() + h.Write(acctBytes) + h.Write([]byte{0}) + for _, id := range sortedCopy(input.SelectedProductIDs) { + h.Write([]byte(id)) + h.Write([]byte{0}) + } + h.Write([]byte{0}) + for _, l := range sortedCopy(input.AcceptedLosses) { + h.Write([]byte(l)) + h.Write([]byte{0}) + } + h.Write([]byte{0}) + h.Write(canon) + return hex.EncodeToString(h.Sum(nil)), nil +} diff --git a/adcp/v3/legacypurchase/store_race_test.go b/adcp/v3/legacypurchase/store_race_test.go new file mode 100644 index 0000000..273d83f --- /dev/null +++ b/adcp/v3/legacypurchase/store_race_test.go @@ -0,0 +1,143 @@ +package legacypurchase + +import ( + "context" + "encoding/json" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/adcontextprotocol/adcp-go/adcp/v3/idempotency" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestContinueLegacyPurchase_ConcurrentDistinctKeysCannotPurchaseTwice is +// the sharpest proof of the first acceptance criterion: N goroutines, each +// with its own distinct idempotency_key (simulating N independent, +// concurrent redemption attempts — the scenario a single-use claim exists +// to prevent), race to redeem the *same* continuation token. Under -race, +// exactly one must win the claim and call Executor; every other goroutine +// must be rejected before Executor runs. If two goroutines ever both +// observed a successful claim, the underlying product would be purchased +// twice. +func TestContinueLegacyPurchase_ConcurrentDistinctKeysCannotPurchaseTwice(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, base := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + + var execCalls int32 + exec := func(context.Context, json.RawMessage) ([]byte, error) { + atomic.AddInt32(&execCalls, 1) + return []byte(`{"media_buy_id":"mb-race"}`), nil + } + + const n = 64 + inputs := make([]*CompatibilityPurchaseCoordinatorInput, n) + for i := range inputs { + in := *base + in.IdempotencyKey = idempotency.Generate() + inputs[i] = &in + } + + var wg sync.WaitGroup + wg.Add(n) + results := make([]*Result, n) + errs := make([]error, n) + start := make(chan struct{}) + ctx := ctxWithPrincipal() + for i := range n { + go func(i int) { + defer wg.Done() + <-start + results[i], errs[i] = s.ContinueLegacyPurchase(ctx, inputs[i], exec) + }(i) + } + close(start) + wg.Wait() + + winners := 0 + rejected := 0 + for i := range n { + if errs[i] == nil { + winners++ + assert.False(t, results[i].Replayed, "the single winner's own call must not be a replay") + continue + } + var ac *AlreadyClaimedError + assert.True(t, errors.As(errs[i], &ac), "loser must fail with AlreadyClaimedError, got %v", errs[i]) + rejected++ + } + assert.Equal(t, 1, winners, "exactly one distinct-key claim must win") + assert.Equal(t, n-1, rejected) + assert.Equal(t, int32(1), atomic.LoadInt32(&execCalls), "exec must run exactly once no matter how many goroutines raced for the token") +} + +// TestContinueLegacyPurchase_ConcurrentSameKeyExecutesOnceAndAgreesOnResult +// races many goroutines using the *same* idempotency_key against a fresh +// continuation (the realistic retry-storm case — a client that times out +// and retries with the same key while its first attempt is still in +// flight). Exactly one call may observe a fresh Executor run; every other +// call must either see the deterministic replayed result or a +// same-key-in-flight signal — never a distinct result, and Executor must +// never run more than once. +func TestContinueLegacyPurchase_ConcurrentSameKeyExecutesOnceAndAgreesOnResult(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + s.opts.PendingLeaseTimeout = time.Minute + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + + var execCalls int32 + release := make(chan struct{}) + exec := func(context.Context, json.RawMessage) ([]byte, error) { + atomic.AddInt32(&execCalls, 1) + <-release // hold the claim pending long enough for others to race in + return []byte(`{"media_buy_id":"mb-same-key"}`), nil + } + + const n = 16 + var wg sync.WaitGroup + wg.Add(n) + results := make([]*Result, n) + errs := make([]error, n) + start := make(chan struct{}) + ctx := ctxWithPrincipal() + for i := range n { + go func(i int) { + defer wg.Done() + <-start + results[i], errs[i] = s.ContinueLegacyPurchase(ctx, input, exec) + }(i) + } + close(start) + // Give every goroutine a chance to reach the backend before the winner + // finishes executing. + time.Sleep(20 * time.Millisecond) + close(release) + wg.Wait() + + freshOrReplayed := 0 + for i := range n { + if errs[i] == nil { + freshOrReplayed++ + assert.JSONEq(t, `{"media_buy_id":"mb-same-key"}`, string(results[i].Response)) + continue + } + var inFlight *InFlightError + assert.True(t, errors.As(errs[i], &inFlight), "a same-key loser must see InFlightError while pending, got %v", errs[i]) + } + assert.GreaterOrEqual(t, freshOrReplayed, 1, "at least the winner must succeed") + assert.Equal(t, int32(1), atomic.LoadInt32(&execCalls), "exec must run exactly once for one idempotency_key regardless of concurrent retries") + + // A final retry after everything has settled must see the same + // deterministic result, not a distinct one. + final, err := s.ContinueLegacyPurchase(ctx, input, exec) + require.NoError(t, err) + assert.True(t, final.Replayed) + assert.JSONEq(t, `{"media_buy_id":"mb-same-key"}`, string(final.Response)) + assert.Equal(t, int32(1), atomic.LoadInt32(&execCalls)) +} diff --git a/adcp/v3/legacypurchase/store_test.go b/adcp/v3/legacypurchase/store_test.go new file mode 100644 index 0000000..27b45bb --- /dev/null +++ b/adcp/v3/legacypurchase/store_test.go @@ -0,0 +1,481 @@ +package legacypurchase + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + adcp "github.com/adcontextprotocol/adcp-go/adcp/v3" + "github.com/adcontextprotocol/adcp-go/adcp/v3/idempotency" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testPrincipal = "buyer-agent-1" + +func mustJSON(t *testing.T, v any) json.RawMessage { + t.Helper() + b, err := json.Marshal(v) + require.NoError(t, err) + return b +} + +// validContinuation returns a fresh, valid Continuation bound to +// testPrincipal, plus a matching CompatibilityPurchaseCoordinatorInput that +// satisfies every one of its binding rules. Tests mutate copies of these to +// exercise individual failure modes. +func validFixture(t *testing.T, now time.Time) (*Continuation, *CompatibilityPurchaseCoordinatorInput) { + t.Helper() + token := "continuation-token-0123456789" + c := &Continuation{ + Token: token, + Principal: testPrincipal, + Account: adcp.AccountReference{AccountID: "account-acme"}, + SourceADCPVersion: "3.0", + ExpiresAt: now.Add(time.Hour), + ProductIDs: []string{"prod-a", "prod-b"}, + Losses: []string{LossFeedVersionNotAtomic, LossPricingVersionNotAtomic}, + // Mirrors the real compact_projection.products shape (see + // vectors_test.go): a bare array of product objects, each with the + // pricing_options a redemption's pricing_option_id is validated + // against. + ObservedPayload: mustJSON(t, []map[string]any{ + {"product_id": "prod-a", "pricing_options": []map[string]any{{"pricing_option_id": "fixed-cpm"}}}, + {"product_id": "prod-b", "pricing_options": []map[string]any{{"pricing_option_id": "fixed-cpm"}}}, + }), + } + input := &CompatibilityPurchaseCoordinatorInput{ + IdempotencyKey: idempotency.Generate(), + ContinuationToken: token, + Account: adcp.AccountReference{AccountID: "account-acme"}, + SelectedProductIDs: []string{"prod-a"}, + AcceptedLosses: []string{LossFeedVersionNotAtomic, LossPricingVersionNotAtomic}, + LegacyCreateRequest: mustJSON(t, map[string]any{ + "packages": []map[string]any{{"product_id": "prod-a", "budget": 1000, "pricing_option_id": "fixed-cpm"}}, + }), + } + return c, input +} + +func newTestStore(now func() time.Time) (*Store, *MemoryBackend) { + b := newMemoryBackend(0, 0, now) + s := New(Options{Backend: b, Clock: now}) + return s, b +} + +func ctxWithPrincipal() context.Context { + return idempotency.WithPrincipal(context.Background(), testPrincipal) +} + +func countingExecutor(t *testing.T, resp []byte) (Executor, *int) { + t.Helper() + calls := 0 + return func(_ context.Context, req json.RawMessage) ([]byte, error) { + calls++ + return resp, nil + }, &calls +} + +// ---- happy path ---- + +func TestContinueLegacyPurchase_Success(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + + exec, calls := countingExecutor(t, []byte(`{"media_buy_id":"mb-1"}`)) + res, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + require.NoError(t, err) + assert.False(t, res.Replayed) + assert.JSONEq(t, `{"media_buy_id":"mb-1"}`, string(res.Response)) + assert.Equal(t, 1, *calls) + + // The Executor must observe exactly legacy_create_request. + rec, err := s.opts.Backend.GetContinuation(context.Background(), c.Token) + require.NoError(t, err) + assert.Equal(t, StateCommitted, rec.State) +} + +// TestContinueLegacyPurchase_RetryReturnsDeterministicPriorResult proves the +// second acceptance criterion literally: an exact retry (same +// idempotency_key, same payload) after success returns the recorded result +// rather than re-invoking exec. +func TestContinueLegacyPurchase_RetryReturnsDeterministicPriorResult(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + + exec, calls := countingExecutor(t, []byte(`{"media_buy_id":"mb-1"}`)) + ctx := ctxWithPrincipal() + first, err := s.ContinueLegacyPurchase(ctx, input, exec) + require.NoError(t, err) + require.False(t, first.Replayed) + + second, err := s.ContinueLegacyPurchase(ctx, input, exec) + require.NoError(t, err) + assert.True(t, second.Replayed) + assert.Equal(t, first.Response, second.Response) + assert.Equal(t, 1, *calls, "exec must not run again on an exact retry") +} + +// TestContinueLegacyPurchase_RetryAfterFailureReturnsTerminalFailure proves +// the failure-path analogue: retrying an idempotency_key whose Executor +// call failed terminally surfaces the recorded failure, not a fresh +// attempt. +func TestContinueLegacyPurchase_RetryAfterFailureReturnsTerminalFailure(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + + calls := 0 + exec := func(context.Context, json.RawMessage) ([]byte, error) { + calls++ + return nil, &InvalidInputError{Reason: "legacy seller rejected: PRODUCT_UNAVAILABLE"} + } + ctx := ctxWithPrincipal() + _, err := s.ContinueLegacyPurchase(ctx, input, exec) + require.Error(t, err) + + _, err = s.ContinueLegacyPurchase(ctx, input, exec) + var tf *TerminalFailureError + require.True(t, errors.As(err, &tf)) + assert.Equal(t, 1, calls, "exec must not run again once a claim has failed terminally") +} + +// ---- single-use claim ---- + +func TestContinueLegacyPurchase_AlreadyClaimedByDifferentKey(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + exec, _ := countingExecutor(t, []byte(`{"media_buy_id":"mb-1"}`)) + ctx := ctxWithPrincipal() + _, err := s.ContinueLegacyPurchase(ctx, input, exec) + require.NoError(t, err) + + second := *input + second.IdempotencyKey = idempotency.Generate() + _, err = s.ContinueLegacyPurchase(ctx, &second, exec) + var ac *AlreadyClaimedError + require.True(t, errors.As(err, &ac)) + assert.Equal(t, StateCommitted, ac.State) +} + +func TestContinueLegacyPurchase_ExactRetryDifferentPayloadConflicts(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + exec, _ := countingExecutor(t, []byte(`{"media_buy_id":"mb-1"}`)) + ctx := ctxWithPrincipal() + _, err := s.ContinueLegacyPurchase(ctx, input, exec) + require.NoError(t, err) + + mutated := *input + mutated.LegacyCreateRequest = mustJSON(t, map[string]any{ + "packages": []map[string]any{{"product_id": "prod-a", "budget": 5000}}, + }) + _, err = s.ContinueLegacyPurchase(ctx, &mutated, exec) + var rc *RequestConflictError + assert.True(t, errors.As(err, &rc)) +} + +// ---- binding mismatches ---- + +func TestContinueLegacyPurchase_ExpiredContinuation(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + c.ExpiresAt = now.Add(-time.Minute) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var ee *ExpiredError + assert.True(t, errors.As(err, &ee)) +} + +func TestContinueLegacyPurchase_WrongPrincipal(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + exec, _ := countingExecutor(t, nil) + ctx := idempotency.WithPrincipal(context.Background(), "someone-else") + _, err := s.ContinueLegacyPurchase(ctx, input, exec) + var pm *PrincipalMismatchError + assert.True(t, errors.As(err, &pm)) +} + +// TestContinueLegacyPurchase_ReplayRejectedForDifferentPrincipal proves +// resolveNonOffered checks the redeeming principal against the token-bound +// one, the same as the fresh-claim path (validateBinding) already does. +// Without that check, a continuation claimed and completed under one +// principal would replay its committed result — or leak its +// pending/terminal state — to a second, unrelated principal who happens to +// reuse the same idempotency_key. +func TestContinueLegacyPurchase_ReplayRejectedForDifferentPrincipal(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + + exec, calls := countingExecutor(t, []byte(`{"media_buy_id":"mb-1"}`)) + first, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + require.NoError(t, err) + require.False(t, first.Replayed) + + // Same continuation, same idempotency_key and payload, but a different + // authenticated principal — must be rejected, not replayed. + otherCtx := idempotency.WithPrincipal(context.Background(), "someone-else") + _, err = s.ContinueLegacyPurchase(otherCtx, input, exec) + var pm *PrincipalMismatchError + require.True(t, errors.As(err, &pm)) + assert.Equal(t, 1, *calls, "exec must not run again, and the committed result must not leak to another principal") +} + +// TestContinueLegacyPurchase_PricingSubstitutionRejected proves a redemption +// cannot select a pricing_option_id the seller never offered for that +// product in the continuation's observed product/pricing payload — the +// spec's binding on the "complete observed product/pricing payload", +// separate from the JSON-structure and product-ID checks. +func TestContinueLegacyPurchase_PricingSubstitutionRejected(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + input.LegacyCreateRequest = mustJSON(t, map[string]any{ + "packages": []map[string]any{{"product_id": "prod-a", "budget": 1000, "pricing_option_id": "premium-cpm-not-offered"}}, + }) + exec, calls := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var pse *PricingSelectionError + require.True(t, errors.As(err, &pse)) + assert.Equal(t, 0, *calls, "exec must never run for a substituted pricing option") +} + +func TestContinueLegacyPurchase_WrongAccount(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + input.Account = adcp.AccountReference{AccountID: "account-other"} + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var am *AccountMismatchError + assert.True(t, errors.As(err, &am)) +} + +func TestContinueLegacyPurchase_RequestAccountMismatchRejected(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + // legacy_create_request carries its own account field (as a 3.0/3.1 + // source would) that disagrees with the token-bound account. + input.LegacyCreateRequest = mustJSON(t, map[string]any{ + "account": map[string]any{"account_id": "account-other"}, + "packages": []map[string]any{{"product_id": "prod-a", "budget": 1000}}, + }) + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var am *AccountMismatchError + assert.True(t, errors.As(err, &am)) +} + +func TestContinueLegacyPurchase_ProductSubstitutionRejected(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + // selected_product_ids names a product not bound into the token at all. + input.SelectedProductIDs = []string{"prod-not-offered"} + input.LegacyCreateRequest = mustJSON(t, map[string]any{ + "packages": []map[string]any{{"product_id": "prod-not-offered", "budget": 1000}}, + }) + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var ps *ProductSelectionError + assert.True(t, errors.As(err, &ps)) +} + +func TestContinueLegacyPurchase_PackageSelectionDriftRejected(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + // selected_product_ids says prod-a, but the actual legacy_create_request + // packages a different (still token-bound) product. + input.LegacyCreateRequest = mustJSON(t, map[string]any{ + "packages": []map[string]any{{"product_id": "prod-b", "budget": 1000}}, + }) + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var ps *ProductSelectionError + assert.True(t, errors.As(err, &ps)) +} + +func TestContinueLegacyPurchase_IncompleteLossConsentRejected(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + input.AcceptedLosses = []string{LossFeedVersionNotAtomic, LossPricingVersionNotAtomic, LossMutationIdempotencyNotGuaranteed} + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var la *LossAcceptanceError + assert.True(t, errors.As(err, &la)) +} + +func TestContinueLegacyPurchase_MissingRequiredLossRejectedAtStructuralValidation(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + // Schema requires accepted_losses to always contain both atomic-fence + // losses; this must fail before even reaching the token lookup. + input.AcceptedLosses = []string{LossFeedVersionNotAtomic} + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var ie *InvalidInputError + assert.True(t, errors.As(err, &ie)) +} + +func TestContinueLegacyPurchase_NotExplicitPackageModeRejected(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + input.LegacyCreateRequest = mustJSON(t, map[string]any{"budget": 1000}) + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var em *ExplicitPackageModeError + assert.True(t, errors.As(err, &em)) +} + +func TestContinueLegacyPurchase_UnknownToken(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + _, input := validFixture(t, now) + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var nf *NotFoundError + assert.True(t, errors.As(err, &nf)) +} + +func TestContinueLegacyPurchase_MissingPrincipalInContext(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + exec, _ := countingExecutor(t, nil) + _, err := s.ContinueLegacyPurchase(context.Background(), input, exec) + var ie *InvalidInputError + assert.True(t, errors.As(err, &ie)) +} + +// ---- crash reconciliation ---- + +func TestContinueLegacyPurchase_AmbiguousClaimAfterLeaseExpiry(t *testing.T) { + now := time.Now().UTC() + clock := &now + s, backend := newTestStore(func() time.Time { return *clock }) + s.opts.PendingLeaseTimeout = time.Minute + c, input := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + + // Simulate a claim that never reached a terminal state (process crash + // between ClaimPending and CompletePending/FailPending). + _, claimed, err := backend.ClaimPending(context.Background(), c.Token, input.IdempotencyKey, mustRequestHash(t, input), now) + require.NoError(t, err) + require.True(t, claimed) + + // Within the lease window: an exact retry is told to retry shortly, not + // treated as ambiguous. + exec, calls := countingExecutor(t, nil) + _, err = s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var inFlight *InFlightError + require.True(t, errors.As(err, &inFlight)) + assert.Equal(t, 0, *calls) + + // Past the lease window: fail closed with recovery guidance, per the + // acceptance criterion "ambiguous/crashed claims fail closed and expose + // recovery guidance." + *clock = now.Add(2 * time.Minute) + _, err = s.ContinueLegacyPurchase(ctxWithPrincipal(), input, exec) + var amb *AmbiguousClaimError + require.True(t, errors.As(err, &amb)) + assert.NotEmpty(t, amb.Guidance) + assert.Equal(t, 0, *calls, "exec must never run for an ambiguous claim") + + // A different idempotency_key must also be rejected — the token is + // already claimed, ambiguous or not; it is never re-offered. + other := *input + other.IdempotencyKey = idempotency.Generate() + _, err = s.ContinueLegacyPurchase(ctxWithPrincipal(), &other, exec) + var ac *AlreadyClaimedError + assert.True(t, errors.As(err, &ac)) +} + +func mustRequestHash(t *testing.T, input *CompatibilityPurchaseCoordinatorInput) string { + t.Helper() + h, err := requestHash(input) + require.NoError(t, err) + return h +} + +// ---- RegisterContinuation validation ---- + +func TestRegisterContinuation_RejectsIncompletePayload(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, _ := validFixture(t, now) + c.ObservedPayload = nil + err := s.RegisterContinuation(context.Background(), c) + var ie *InvalidInputError + assert.True(t, errors.As(err, &ie)) +} + +func TestRegisterContinuation_RejectsMissingRequiredLosses(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, _ := validFixture(t, now) + c.Losses = []string{LossMutationIdempotencyNotGuaranteed} + err := s.RegisterContinuation(context.Background(), c) + var ie *InvalidInputError + assert.True(t, errors.As(err, &ie)) +} + +// TestRegisterContinuation_Rejects25WithoutMutationIdempotencyLoss proves an +// AdCP 2.5-sourced continuation must declare +// mutation_idempotency_not_guaranteed — 2.5 has no mutation replay contract, +// per specs/legacy-compact-lifecycle-compatibility.md — rather than being +// registerable with only the two losses every source declares. +func TestRegisterContinuation_Rejects25WithoutMutationIdempotencyLoss(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, _ := validFixture(t, now) + c.SourceADCPVersion = "2.5" + c.Losses = []string{LossFeedVersionNotAtomic, LossPricingVersionNotAtomic} + err := s.RegisterContinuation(context.Background(), c) + var ie *InvalidInputError + require.True(t, errors.As(err, &ie)) + + c.Losses = []string{LossFeedVersionNotAtomic, LossPricingVersionNotAtomic, LossMutationIdempotencyNotGuaranteed} + assert.NoError(t, s.RegisterContinuation(context.Background(), c)) +} + +func TestRegisterContinuation_DuplicateTokenRejected(t *testing.T) { + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + c, _ := validFixture(t, now) + require.NoError(t, s.RegisterContinuation(context.Background(), c)) + err := s.RegisterContinuation(context.Background(), c) + var dt *DuplicateTokenError + assert.True(t, errors.As(err, &dt)) +} diff --git a/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/PROVENANCE.md b/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/PROVENANCE.md new file mode 100644 index 0000000..6481a1d --- /dev/null +++ b/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/PROVENANCE.md @@ -0,0 +1,52 @@ +# Provenance + +`vectors.json` and `UPSTREAM_README.md` (renamed from the upstream `README.md` +to avoid colliding with this package's own `testdata` conventions) are a +byte-for-byte copy of +`static/compliance/source/test-vectors/products-only-brief-compatibility/` +from `adcontextprotocol/adcp`, added by +[adcp#6733](https://github.com/adcontextprotocol/adcp/pull/6733) ("Define +legacy and compact lifecycle compatibility", merged 2026-08-20, commit +`e72ff10b5bdd1da1491918120656395b5d395a7e`) and shipped in the +`3.2.0-beta.9` schema bundle this repo pins at `adcp/v3/schemas/VERSION` +(confirmed identical: `diff`'d against the locally-downloaded pinned bundle's +copy of the same file with zero content differences). + +Do not hand-edit `vectors.json`. Re-sync it from the `adcp` repo (or a future +signed bundle release that vendors compliance vectors alongside schemas) +when the upstream vectors change. + +## What this package exercises against the vectors + +- `cases[]` (`legacy_create` continuations for AdCP 2.5.3 / 3.0.18 / 3.1.15 + sources) — exercised end-to-end in `vectors_test.go`: register the + continuation from `compact_projection`, redeem it via + `Store.ContinueLegacyPurchase` using `continuation_input` verbatim, and + assert the `Executor` receives exactly `legacy_create_request` and a + retry with the same `idempotency_key` returns the deterministic prior + result. Negative variants (product substitution, package-selection drift, + stale/incomplete loss consent, wrong account, wrong principal) are + constructed in `vectors_test.go` and `store_test.go` by mutating these + same fixtures, per this vector set's own `UPSTREAM_README.md` ("SDK suites + consume the same vectors to exercise expiry, principal/account binding, + atomic token claim, exact retry, and crash reconciliation against their + durable coordinator implementations") — the upstream bundle does not ship + the negative cases as separate JSON fixtures. + +- `listed_purchase_cases[]` — intentionally **not** exercised against + `Store`. Per + `specs/legacy-compact-lifecycle-compatibility.md#listed_purchase`, a + `listed_purchase` continuation carries a seller-issued, account-scoped + feed/pricing fence straight into ordinary `buy_products` — "The + coordinator passes those seller-issued values to ordinary buy_products." + There is no durable continuation state to claim for this branch; it is + structurally out of scope for a claim coordinator, not a deferred gap. + +- `reverse_compatibility_cases[]` — **not** exercised. This is the seller + side of the compatibility contract (an AdCP 3.2 seller preserving its + deprecated `get_products`/`create_media_buy` facades for established + buyers, described in the spec's "Established buyers against a + compact-backed seller" section). It is a materially separate, + substantial scope of server-side adapter work from the buyer-side + coordinator this package implements, and is out of scope for this PR — see + the linked follow-up issue in the package README. diff --git a/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/UPSTREAM_README.md b/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/UPSTREAM_README.md new file mode 100644 index 0000000..9d8c173 --- /dev/null +++ b/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/UPSTREAM_README.md @@ -0,0 +1,25 @@ +# Products-only brief compatibility vectors + +These vectors pin the valid established flow for AdCP 2.5, 3.0, and 3.1: + +1. `get_products` brief discovery returns usable products and no proposal. +2. The compact projection returns `products_available` without inventing a + proposal, commercial terms, terms digest, or feed/pricing fence. +3. The fail-closed `legacy_create` continuation names both missing atomic + fences and is redeemed through the typed SDK-local + `continueLegacyPurchase` input before routing the selected product to that + version's `create_media_buy` request. AdCP 2.5 also names + `mutation_idempotency_not_guaranteed`, since that release has no mutation + replay contract; the follow-up must accept every returned loss. +4. The seller-fenced `listed_purchase` branch carries real account-scoped feed + and pricing versions unchanged into `buy_products`. +5. A 3.2 seller's established facades preserve the reverse direction: + products-only legacy discovery remains executable through legacy create. + +The negative assertions pin product substitution, package-selection drift, +and incomplete loss consent. SDK suites consume the same vectors to exercise +expiry, principal/account binding, atomic token claim, exact retry, and crash +reconciliation against their durable coordinator implementations. + +The `legacy_create` continuation is deprecated compatibility behavior for the +AdCP 3.x window and is removable in AdCP 4.0. diff --git a/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/vectors.json b/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/vectors.json new file mode 100644 index 0000000..38775c8 --- /dev/null +++ b/adcp/v3/legacypurchase/testdata/products-only-brief-compatibility/vectors.json @@ -0,0 +1,292 @@ +{ + "description": "Established products-only brief discovery followed by its explicit AdCP 3.2 compatibility projection and legacy purchase continuation.", + "cases": [ + { + "source_version": "2.5.3", + "legacy_request": { + "brief": "A premium display campaign for Acme.", + "brand_manifest": { "name": "Acme" } + }, + "legacy_response": { + "products": [{ + "product_id": "brief-display-25", + "name": "Brief display 2.5", + "description": "A product composed for this brief.", + "publisher_properties": [{ "publisher_domain": "publisher.example", "selection_type": "all" }], + "format_ids": [{ "agent_url": "https://creative.example", "id": "display-300x250" }], + "delivery_type": "guaranteed", + "delivery_measurement": { "provider": "Publisher" }, + "pricing_options": [{ + "pricing_option_id": "fixed-cpm", + "pricing_model": "cpm", + "rate": 12, + "currency": "USD", + "is_fixed": true + }], + "reporting_capabilities": { + "available_reporting_frequencies": ["daily"], + "expected_delay_minutes": 240, + "timezone": "UTC", + "supports_webhooks": false, + "available_metrics": ["impressions"] + } + }] + }, + "compact_projection": { + "outcome": "products_available", + "products": [{ + "product_id": "brief-display-25", + "name": "Brief display 2.5", + "description": "A product composed for this brief.", + "pricing_options": [{ + "pricing_option_id": "fixed-cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_price": 12 + }] + }], + "purchase_continuation": { + "kind": "legacy_create", + "continuation_token": "products-only-brief-25-token", + "continuation_expires_at": "2099-01-01T00:05:00Z", + "product_ids": ["brief-display-25"], + "source_adcp_version": "2.5", + "losses": ["feed_version_not_atomic", "pricing_version_not_atomic", "mutation_idempotency_not_guaranteed"], + "requires_explicit_acceptance": true + } + }, + "continuation_input": { + "idempotency_key": "71a26db8-69fe-49d5-80a6-a4a39c71f625", + "continuation_token": "products-only-brief-25-token", + "account": { "account_id": "account-acme" }, + "selected_product_ids": ["brief-display-25"], + "accepted_losses": ["feed_version_not_atomic", "pricing_version_not_atomic", "mutation_idempotency_not_guaranteed"], + "legacy_create_request": { + "buyer_ref": "products-only-25", + "packages": [{ + "buyer_ref": "line-25", + "product_id": "brief-display-25", + "budget": 1000, + "pricing_option_id": "fixed-cpm" + }], + "brand_manifest": { "name": "Acme" }, + "start_time": "2099-01-01T00:00:00Z", + "end_time": "2099-02-01T00:00:00Z" + } + } + }, + { + "source_version": "3.0.18", + "legacy_request": { + "buying_mode": "brief", + "brief": "A premium display campaign for Acme.", + "brand": { "domain": "acme.example" }, + "account": { "account_id": "account-acme" } + }, + "legacy_response": { + "products": [{ + "product_id": "brief-display-30", + "name": "Brief display 3.0", + "description": "A product composed for this brief.", + "publisher_properties": [{ "publisher_domain": "publisher.example", "selection_type": "all" }], + "format_ids": [{ "agent_url": "https://creative.example", "id": "display-300x250" }], + "delivery_type": "guaranteed", + "pricing_options": [{ + "pricing_option_id": "fixed-cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_price": 12 + }], + "reporting_capabilities": { + "available_reporting_frequencies": ["daily"], + "expected_delay_minutes": 240, + "timezone": "UTC", + "supports_webhooks": false, + "available_metrics": ["impressions"], + "date_range_support": "date_range" + } + }], + "incomplete": [{ + "scope": "proposals", + "description": "The seller returned usable products without constructing a proposal." + }] + }, + "compact_projection": { + "outcome": "products_available", + "products": [{ + "product_id": "brief-display-30", + "name": "Brief display 3.0", + "description": "A product composed for this brief.", + "pricing_options": [{ + "pricing_option_id": "fixed-cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_price": 12 + }] + }], + "incomplete": [{ + "scope": "proposals", + "description": "The seller returned usable products without constructing a proposal." + }], + "purchase_continuation": { + "kind": "legacy_create", + "continuation_token": "products-only-brief-30-token", + "continuation_expires_at": "2099-01-01T00:05:00Z", + "product_ids": ["brief-display-30"], + "source_adcp_version": "3.0", + "losses": ["feed_version_not_atomic", "pricing_version_not_atomic"], + "requires_explicit_acceptance": true + } + }, + "continuation_input": { + "idempotency_key": "5f203b58-9ae7-4d02-b05f-eb239a47f065", + "continuation_token": "products-only-brief-30-token", + "account": { "account_id": "account-acme" }, + "selected_product_ids": ["brief-display-30"], + "accepted_losses": ["feed_version_not_atomic", "pricing_version_not_atomic"], + "legacy_create_request": { + "idempotency_key": "products-only-brief-30-0001", + "account": { "account_id": "account-acme" }, + "brand": { "domain": "acme.example" }, + "packages": [{ + "product_id": "brief-display-30", + "budget": 1000, + "pricing_option_id": "fixed-cpm" + }], + "start_time": "2099-01-01T00:00:00Z", + "end_time": "2099-02-01T00:00:00Z" + } + } + }, + { + "source_version": "3.1.15", + "legacy_request": { + "buying_mode": "brief", + "brief": "A premium display campaign for Acme.", + "brand": { "domain": "acme.example" }, + "account": { "account_id": "account-acme" } + }, + "legacy_response": { + "status": "completed", + "cache_scope": "account", + "products": [{ + "product_id": "brief-display-31", + "name": "Brief display 3.1", + "description": "A product composed for this brief.", + "publisher_properties": [{ "publisher_domain": "publisher.example", "selection_type": "all" }], + "format_ids": [{ "agent_url": "https://creative.example", "id": "display-300x250" }], + "delivery_type": "guaranteed", + "pricing_options": [{ + "pricing_option_id": "fixed-cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_price": 12 + }], + "reporting_capabilities": { + "available_reporting_frequencies": ["daily"], + "expected_delay_minutes": 240, + "timezone": "UTC", + "supports_webhooks": false, + "available_metrics": ["impressions"], + "date_range_support": "date_range" + } + }], + "incomplete": [{ + "scope": "proposals", + "description": "The seller returned usable products without constructing a proposal." + }] + }, + "compact_projection": { + "outcome": "products_available", + "products": [{ + "product_id": "brief-display-31", + "name": "Brief display 3.1", + "description": "A product composed for this brief.", + "pricing_options": [{ + "pricing_option_id": "fixed-cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_price": 12 + }] + }], + "incomplete": [{ + "scope": "proposals", + "description": "The seller returned usable products without constructing a proposal." + }], + "purchase_continuation": { + "kind": "legacy_create", + "continuation_token": "products-only-brief-31-token", + "continuation_expires_at": "2099-01-01T00:05:00Z", + "product_ids": ["brief-display-31"], + "source_adcp_version": "3.1", + "losses": ["feed_version_not_atomic", "pricing_version_not_atomic"], + "requires_explicit_acceptance": true + } + }, + "continuation_input": { + "idempotency_key": "a15ac836-a49e-4e59-bb49-df24dc2cc339", + "continuation_token": "products-only-brief-31-token", + "account": { "account_id": "account-acme" }, + "selected_product_ids": ["brief-display-31"], + "accepted_losses": ["feed_version_not_atomic", "pricing_version_not_atomic"], + "legacy_create_request": { + "idempotency_key": "products-only-brief-31-0001", + "account": { "account_id": "account-acme" }, + "brand": { "domain": "acme.example" }, + "packages": [{ + "product_id": "brief-display-31", + "budget": 1000, + "pricing_option_id": "fixed-cpm" + }], + "start_time": "2099-01-01T00:00:00Z", + "end_time": "2099-02-01T00:00:00Z" + } + } + } + ], + "listed_purchase_cases": [ + { + "compact_projection": { + "outcome": "products_available", + "products": [{ + "product_id": "account-listed-display", + "name": "Account-listed display", + "pricing_options": [{ + "pricing_option_id": "account-fixed-cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_price": 10 + }] + }], + "purchase_continuation": { + "kind": "listed_purchase", + "product_ids": ["account-listed-display"], + "cache_scope": "account", + "feed_version": "seller-feed-account-42", + "pricing_version": "seller-pricing-account-17" + } + }, + "buy_products_request": { + "idempotency_key": "listed-purchase-vector-0001", + "account": { "account_id": "account-acme" }, + "brand": { "domain": "acme.example" }, + "feed_version": "seller-feed-account-42", + "pricing_version": "seller-pricing-account-17", + "purchases": [{ + "product_id": "account-listed-display", + "pricing_option_id": "account-fixed-cpm", + "budget": 1000 + }], + "start_time": "2099-01-01T00:00:00Z", + "end_time": "2099-02-01T00:00:00Z" + } + } + ], + "reverse_compatibility_cases": [ + { + "description": "An established buyer uses the 3.2 seller's legacy get_products and create_media_buy facades without crossing into products_available.", + "seller_version": "3.2", + "source_case_index": 2 + } + ] +} diff --git a/adcp/v3/legacypurchase/types.go b/adcp/v3/legacypurchase/types.go new file mode 100644 index 0000000..5ca1170 --- /dev/null +++ b/adcp/v3/legacypurchase/types.go @@ -0,0 +1,193 @@ +package legacypurchase + +import ( + "context" + "encoding/json" + "time" + + adcp "github.com/adcontextprotocol/adcp-go/adcp/v3" +) + +// Continuation kinds from purchase_continuation.kind. Only KindLegacyCreate +// has durable state this package claims — see the package doc. +const ( + KindListedPurchase = "listed_purchase" + KindLegacyCreate = "legacy_create" +) + +// Loss names media-buy/legacy-purchase-continuation-input.json's +// accepted_losses enum allows. LossFeedVersionNotAtomic and +// LossPricingVersionNotAtomic are required on every legacy_create +// continuation per the schema's own allOf/contains constraint; +// LossMutationIdempotencyNotGuaranteed is required only for an AdCP 2.5 +// source (2.5 has no mutation replay contract) or when the actual 3.0/3.1 +// peer does not provide one. +const ( + LossFeedVersionNotAtomic = "feed_version_not_atomic" + LossPricingVersionNotAtomic = "pricing_version_not_atomic" + LossMutationIdempotencyNotGuaranteed = "mutation_idempotency_not_guaranteed" +) + +// ContinuationState is the lifecycle state of one durable legacy_create +// continuation record. +type ContinuationState string + +const ( + // StateOffered: registered, not yet redeemed. Single-use — the only + // state ClaimPending may transition out of. + StateOffered ContinuationState = "offered" + // StatePending: atomically claimed by one idempotency_key; the + // coordinator's Executor call is in flight or the process that claimed + // it crashed before recording a terminal outcome. See + // AmbiguousClaimError and InFlightError. + StatePending ContinuationState = "pending" + // StateCommitted: Executor succeeded; Result holds its response. + StateCommitted ContinuationState = "committed" + // StateFailed: Executor (or the legacy seller behind it) failed + // terminally; ErrorCode/ErrorMessage hold recovery guidance. A failed + // continuation is still single-use-spent — it is never re-offered. + StateFailed ContinuationState = "failed" +) + +// Continuation is the set of seller-issued facts an application's +// compatibility-projection layer binds into a legacy_create +// purchase_continuation at the moment it decides to offer one to a caller, +// per specs/legacy-compact-lifecycle-compatibility.md's "legacy_create" +// section. RegisterContinuation persists these facts; ContinueLegacyPurchase +// verifies every field against a later CompatibilityPurchaseCoordinatorInput +// before allowing a single atomic claim. +type Continuation struct { + // Token is the opaque continuation_token surfaced to the caller in + // purchase_continuation.continuation_token. Must be at least 16 + // characters (matches the schema's continuation_token minLength) and + // unique — RegisterContinuation fails closed on collision. + Token string + + // Principal is the authenticated identity the token is bound to (the + // same principal concept as adcp/v3/idempotency.WithPrincipal — this + // package reads it from the same context key via + // idempotency.PrincipalFromContext, so a seller that already wraps its + // handlers with idempotency.Store gets principal binding for free). + // ContinueLegacyPurchase rejects a redemption attempt from a different + // principal — the spec's confused-deputy guard. + Principal string + + // Account is the account identity bound into the token. AdCP 2.5 has + // no wire account field on its legacy request, so its adapter must + // still supply the same token-bound client/account session here — the + // spec's "MUST NOT retarget the seller connection" rule. + Account adcp.AccountReference + + // SourceADCPVersion is the legacy protocol version the continuation + // targets (e.g. "2.5", "3.0", "3.1"). + SourceADCPVersion string + + // ExpiresAt is the continuation's absolute deadline + // (purchase_continuation.continuation_expires_at). A redemption attempt + // after this time fails closed with ExpiredError. + ExpiresAt time.Time + + // ProductIDs is the complete, seller-issued set of product IDs bound + // into the continuation (purchase_continuation.product_ids). A + // redemption's selected_product_ids must be a non-empty subset of this + // set. + ProductIDs []string + + // Losses is the complete, exact loss set the continuation declares + // (purchase_continuation.losses). A redemption's accepted_losses must + // equal this set exactly — not a subset, not a superset. + Losses []string + + // ObservedPayload is the canonical bytes of the products/pricing + // payload actually observed when this continuation was minted (the + // compact_projection.products the caller saw). It is bound durably at + // registration time and never accepted from a later redemption + // request, which is what makes it a payload-substitution guard rather + // than a self-reported claim. Must be non-empty — RegisterContinuation + // rejects an incomplete payload rather than minting a continuation it + // cannot stand behind. + ObservedPayload []byte +} + +// ContinuationRecord is the durable record a Backend stores: the bound +// Continuation facts plus its claim/completion state. +type ContinuationRecord struct { + Continuation + + State ContinuationState + + RegisteredAt time.Time + + // ClaimantKey is the idempotency_key of the redemption call that + // claimed this continuation. Empty while State == StateOffered. + ClaimantKey string + // RequestHash is a canonical hash of the full redemption input claimed + // under ClaimantKey, used to detect an idempotency_key reused with a + // different payload (RequestConflictError) versus an exact retry. + RequestHash string + ClaimedAt time.Time + + // Result holds the Executor's response once State == StateCommitted. + Result []byte + // ErrorCode / ErrorMessage hold recovery guidance once + // State == StateFailed. + ErrorCode string + ErrorMessage string + CompletedAt time.Time +} + +// CompatibilityPurchaseCoordinatorInput is the SDK-local input for +// redeeming a legacy_create continuation. Its field shape mirrors +// media-buy/legacy-purchase-continuation-input.json from the AdCP +// 3.2.0-beta.9 schema bundle exactly (see doc.go for why it is hand-written +// rather than generated). Per the schema: "This object is consumed by the +// compatibility coordinator and MUST NOT be sent as an AdCP tool payload." +type CompatibilityPurchaseCoordinatorInput struct { + // IdempotencyKey is the replay identity for this logical coordinator + // operation (schema: format uuid). Exact retries resume the durable + // operation record instead of redeeming the continuation again. + IdempotencyKey string `json:"idempotency_key"` + + // ContinuationToken is the opaque token from + // products_available.purchase_continuation.continuation_token (schema: + // minLength 16). + ContinuationToken string `json:"continuation_token"` + + // Account must match the account bound into the continuation token. + Account adcp.AccountReference `json:"account"` + + // SelectedProductIDs is a non-empty subset of the product IDs bound + // into the continuation, and must equal the distinct explicit-package + // product IDs in LegacyCreateRequest. + SelectedProductIDs []string `json:"selected_product_ids"` + + // AcceptedLosses must equal the continuation's exact loss set. Per + // schema it must always contain at least + // feed_version_not_atomic and pricing_version_not_atomic. + AcceptedLosses []string `json:"accepted_losses"` + + // LegacyCreateRequest is the proposed create_media_buy payload for + // SourceADCPVersion. Validated structurally by this package + // (explicit-package mode, package product IDs) — see doc.go's scope + // note on what full per-version schema validation remains + // application-owned. + LegacyCreateRequest json.RawMessage `json:"legacy_create_request"` +} + +// Result is the outcome of a successful ContinueLegacyPurchase call. +type Result struct { + // Response is the Executor's response bytes — either freshly produced + // (Replayed == false) or the durably recorded prior result of an exact + // idempotency_key retry (Replayed == true). + Response []byte + Replayed bool +} + +// Executor performs the actual legacy create_media_buy call — against the +// real legacy seller, or an application's own legacy facade — and is +// invoked by ContinueLegacyPurchase at most once per distinct continuation +// claim. legacyCreateRequest is exactly +// CompatibilityPurchaseCoordinatorInput.LegacyCreateRequest, already +// structurally validated (explicit-package mode, product IDs) before +// Executor is called. +type Executor func(ctx context.Context, legacyCreateRequest json.RawMessage) (response []byte, err error) diff --git a/adcp/v3/legacypurchase/vectors_test.go b/adcp/v3/legacypurchase/vectors_test.go new file mode 100644 index 0000000..a5e187c --- /dev/null +++ b/adcp/v3/legacypurchase/vectors_test.go @@ -0,0 +1,252 @@ +package legacypurchase + +import ( + "context" + "encoding/json" + "os" + "testing" + "time" + + "github.com/adcontextprotocol/adcp-go/adcp/v3/idempotency" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// See testdata/products-only-brief-compatibility/PROVENANCE.md for exactly +// where this fixture came from and which of its sections this file +// exercises against Store. + +const vectorsPath = "testdata/products-only-brief-compatibility/vectors.json" + +type vectorFile struct { + Description string `json:"description"` + Cases []vectorCase `json:"cases"` + ListedPurchaseCases []json.RawMessage `json:"listed_purchase_cases"` + ReverseCompatibilityCases []json.RawMessage `json:"reverse_compatibility_cases"` +} + +type vectorCase struct { + SourceVersion string `json:"source_version"` + CompactProjection vectorProjection `json:"compact_projection"` + ContinuationInput json.RawMessage `json:"continuation_input"` +} + +type vectorProjection struct { + Outcome string `json:"outcome"` + Products json.RawMessage `json:"products"` + PurchaseContinuation vectorPurchaseContinuation `json:"purchase_continuation"` +} + +type vectorPurchaseContinuation struct { + Kind string `json:"kind"` + ContinuationToken string `json:"continuation_token"` + ContinuationExpiresAt string `json:"continuation_expires_at"` + ProductIDs []string `json:"product_ids"` + SourceADCPVersion string `json:"source_adcp_version"` + Losses []string `json:"losses"` +} + +func loadVectors(t *testing.T) *vectorFile { + t.Helper() + b, err := os.ReadFile(vectorsPath) + require.NoError(t, err, "products-only-brief-compatibility vectors must be present at %s — see testdata PROVENANCE.md", vectorsPath) + var vf vectorFile + require.NoError(t, json.Unmarshal(b, &vf)) + require.NotEmpty(t, vf.Cases, "vectors.json cases[] must not be empty") + return &vf +} + +const vectorPrincipal = "compatibility-vector-principal" + +// registerVectorContinuation registers the Continuation described by a +// vectors.json case, bound to vectorPrincipal. +func registerVectorContinuation(t *testing.T, s *Store, tc vectorCase, account CompatibilityPurchaseCoordinatorInput) { + t.Helper() + pc := tc.CompactProjection.PurchaseContinuation + expiresAt, err := time.Parse(time.RFC3339, pc.ContinuationExpiresAt) + require.NoError(t, err) + c := &Continuation{ + Token: pc.ContinuationToken, + Principal: vectorPrincipal, + Account: account.Account, // token-bound account: the fixture is self-consistent by construction + SourceADCPVersion: pc.SourceADCPVersion, + ExpiresAt: expiresAt, + ProductIDs: pc.ProductIDs, + Losses: pc.Losses, + ObservedPayload: tc.CompactProjection.Products, + } + require.NoError(t, s.RegisterContinuation(context.Background(), c)) +} + +// TestProductsOnlyBriefCompatibilityVectors runs every vectors.json +// cases[] entry — the AdCP 2.5 / 3.0 / 3.1 legacy_create continuations — +// end to end: register the continuation exactly as +// compact_projection.purchase_continuation describes it, redeem it via +// Store.ContinueLegacyPurchase using continuation_input verbatim from the +// fixture, and confirm the Executor receives exactly legacy_create_request +// and that a same-key retry returns the deterministic prior result. +func TestProductsOnlyBriefCompatibilityVectors(t *testing.T) { + vf := loadVectors(t) + for _, tc := range vf.Cases { + tc := tc + t.Run(tc.SourceVersion, func(t *testing.T) { + pc := tc.CompactProjection.PurchaseContinuation + require.Equal(t, KindLegacyCreate, pc.Kind, "this fixture's cases[] are documented to be legacy_create continuations") + + var input CompatibilityPurchaseCoordinatorInput + require.NoError(t, json.Unmarshal(tc.ContinuationInput, &input)) + + now := time.Now().UTC() + s, _ := newTestStore(func() time.Time { return now }) + registerVectorContinuation(t, s, tc, input) + + var observedReq json.RawMessage + calls := 0 + exec := func(_ context.Context, req json.RawMessage) ([]byte, error) { + calls++ + observedReq = req + return []byte(`{"media_buy_id":"mb-vector"}`), nil + } + + ctx := idempotency.WithPrincipal(context.Background(), vectorPrincipal) + res, err := s.ContinueLegacyPurchase(ctx, &input, exec) + require.NoError(t, err) + assert.False(t, res.Replayed) + assert.Equal(t, 1, calls) + assert.JSONEq(t, string(input.LegacyCreateRequest), string(observedReq), + "Executor must receive exactly legacy_create_request, unmodified") + + // Retry after success returns the deterministic prior result. + replay, err := s.ContinueLegacyPurchase(ctx, &input, exec) + require.NoError(t, err) + assert.True(t, replay.Replayed) + assert.Equal(t, res.Response, replay.Response) + assert.Equal(t, 1, calls, "exec must not run again on retry") + }) + } +} + +// TestProductsOnlyBriefCompatibilityVectors_NegativeMutations constructs +// the negative assertions the vector bundle's own README says SDK suites +// are expected to build from these fixtures ("product substitution, +// package-selection drift, and incomplete loss consent") — the upstream +// bundle does not ship separate negative-case JSON. +func TestProductsOnlyBriefCompatibilityVectors_NegativeMutations(t *testing.T) { + vf := loadVectors(t) + exec := func(context.Context, json.RawMessage) ([]byte, error) { + t.Fatal("exec must not run for a rejected redemption") + return nil, nil + } + ctx := idempotency.WithPrincipal(context.Background(), vectorPrincipal) + + setup := func(t *testing.T, tc vectorCase, input CompatibilityPurchaseCoordinatorInput) *Store { + s, _ := newTestStore(func() time.Time { return time.Now().UTC() }) + registerVectorContinuation(t, s, tc, input) + return s + } + + // 3.0 case: exactly one product, one required loss pair, no third loss. + tc30 := vf.Cases[1] + var input30 CompatibilityPurchaseCoordinatorInput + require.NoError(t, json.Unmarshal(tc30.ContinuationInput, &input30)) + + t.Run("product substitution", func(t *testing.T) { + s := setup(t, tc30, input30) + in := input30 + in.SelectedProductIDs = []string{"a-product-never-offered"} + in.LegacyCreateRequest = mustJSON(t, map[string]any{ + "packages": []map[string]any{{"product_id": "a-product-never-offered", "budget": 1000}}, + }) + _, err := s.ContinueLegacyPurchase(ctx, &in, exec) + var ps *ProductSelectionError + assert.ErrorAs(t, err, &ps) + }) + + t.Run("package selection drift", func(t *testing.T) { + s := setup(t, tc30, input30) + in := input30 + // selected_product_ids still names the real, token-bound product, + // but the actual legacy_create_request packages a different one. + in.LegacyCreateRequest = mustJSON(t, map[string]any{ + "packages": []map[string]any{{"product_id": "a-different-product", "budget": 1000}}, + }) + _, err := s.ContinueLegacyPurchase(ctx, &in, exec) + var ps *ProductSelectionError + assert.ErrorAs(t, err, &ps) + }) + + t.Run("stale loss consent - superset", func(t *testing.T) { + s := setup(t, tc30, input30) + in := input30 + // The 3.0 token requires exactly the two atomic-fence losses; add + // the 2.5-only mutation-idempotency loss, which this token never + // declared. + in.AcceptedLosses = append(append([]string{}, input30.AcceptedLosses...), LossMutationIdempotencyNotGuaranteed) + _, err := s.ContinueLegacyPurchase(ctx, &in, exec) + var la *LossAcceptanceError + assert.ErrorAs(t, err, &la) + }) + + // 2.5 case: has the extra mutation-idempotency loss, so dropping one + // element yields a genuine incomplete-consent case. + tc25 := vf.Cases[0] + var input25 CompatibilityPurchaseCoordinatorInput + require.NoError(t, json.Unmarshal(tc25.ContinuationInput, &input25)) + require.Len(t, input25.AcceptedLosses, 3, "the 2.5 vector is expected to declare 3 losses") + + t.Run("incomplete loss consent", func(t *testing.T) { + s := setup(t, tc25, input25) + in := input25 + in.AcceptedLosses = []string{LossFeedVersionNotAtomic, LossPricingVersionNotAtomic} // drops mutation_idempotency_not_guaranteed + _, err := s.ContinueLegacyPurchase(ctx, &in, exec) + var la *LossAcceptanceError + assert.ErrorAs(t, err, &la) + }) + + t.Run("wrong account", func(t *testing.T) { + s := setup(t, tc30, input30) + in := input30 + in.Account.AccountID = "account-not-bound-to-this-token" + _, err := s.ContinueLegacyPurchase(ctx, &in, exec) + var am *AccountMismatchError + assert.ErrorAs(t, err, &am) + }) + + t.Run("expired continuation", func(t *testing.T) { + pc := tc30.CompactProjection.PurchaseContinuation + expiresAt, err := time.Parse(time.RFC3339, pc.ContinuationExpiresAt) + require.NoError(t, err) + // Register and redeem against a clock fixed one hour past the + // vector's own continuation_expires_at (which is 2099-dated, so an + // unmodified "now" would never expire it). + s := New(Options{ + Backend: newMemoryBackend(0, 0, func() time.Time { return expiresAt.Add(time.Hour) }), + Clock: func() time.Time { return expiresAt.Add(time.Hour) }, + }) + require.NoError(t, s.RegisterContinuation(context.Background(), &Continuation{ + Token: pc.ContinuationToken, + Principal: vectorPrincipal, + Account: input30.Account, + SourceADCPVersion: pc.SourceADCPVersion, + ExpiresAt: expiresAt, + ProductIDs: pc.ProductIDs, + Losses: pc.Losses, + ObservedPayload: tc30.CompactProjection.Products, + })) + _, err = s.ContinueLegacyPurchase(ctx, &input30, exec) + var ee *ExpiredError + assert.ErrorAs(t, err, &ee) + }) +} + +// TestProductsOnlyBriefCompatibilityVectors_OutOfScopeSectionsParse proves +// vectors.json's listed_purchase_cases and reverse_compatibility_cases +// sections are read successfully (so a stale/misread fixture would be +// caught here) without claiming this package exercises them — see +// testdata/PROVENANCE.md for why they are structurally out of scope for a +// buyer-side claim coordinator. +func TestProductsOnlyBriefCompatibilityVectors_OutOfScopeSectionsParse(t *testing.T) { + vf := loadVectors(t) + assert.NotEmpty(t, vf.ListedPurchaseCases, "listed_purchase_cases should be present in the fixture even though this package does not exercise it") + assert.NotEmpty(t, vf.ReverseCompatibilityCases, "reverse_compatibility_cases should be present in the fixture even though this package does not exercise it") +}