diff --git a/AGENTS.md b/AGENTS.md index 4eb7fb9..1e31ed8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,16 +41,21 @@ subclasses the official `openai` client, pre-points it at the Grid, reads the ke environment, and adds Grid-specific conveniences (`online_models()`, and `client.grid` for the native SYNCHRONOUS media endpoints: `/v1/images/generations` (video / img2img / ControlNet / LoRAs / styles) and `/v1/videos/generations` — the call returns the finished result, no -submit/poll). The retired horde `/api/v2` async queue is NOT used. Its PyPI package name is +submit/poll). A separate candidate `client.grid.video_jobs` exposes Core's +default-off `/v1/videos/jobs` credit-backed submission/recovery contract; it +does not change those existing synchronous methods. The retired horde `/api/v2` async queue is NOT used. Its PyPI package name is `grid-sdk`; it is importable as `grid_sdk`. Co-canonical peer of the JS SDK `../grid-sdk-js` (npm `grid-ai`) — keep the two SDKs' surfaces aligned. ## Ownership - **`src/grid_sdk/`** — the package (clients + raw-Grid helpers). Owned in its own AGENTS.md. -- **`tests/`** — pytest suite (`test_client.py`, `test_grid.py`); uses `respx` to mock HTTP. +- **`tests/`** — pytest suite (client, legacy media and durable video jobs); uses `respx` to mock HTTP. Owned in its own AGENTS.md. - **`pyproject.toml`** — setuptools build, `src/` layout, deps, pytest config. +- **`qualification/`** - optional Core SQL/auth plus built-JS wire parity checks; + additional actual JS localhost HTTP submission/recovery, synthetic + storage/workers only, explicitly selected dependency environment. - **`.github/workflows/secret-scan.yml`, `.gitleaks.toml`, and `.gitleaksignore`** — checksum-verified complete-history secret scanning with exact historical fingerprints only. - **`.github/workflows/release.yml`** — release-published, tag-bound PyPI @@ -69,9 +74,23 @@ submit/poll). The retired horde `/api/v2` async queue is NOT used. Its PyPI pack recent OpenAI SDK releases may install `httpx2` instead of the `httpx` import used here. Do not add dependencies casually — thinness is the product. - **Public API is `__init__.__all__`:** `Grid`, `AsyncGrid`, `AIPG`, `AsyncAIPG`, - `DEFAULT_BASE_URL`. `AIPG`/`AsyncAIPG` are aliases of `Grid`/`AsyncGrid` and must stay so. + `DEFAULT_BASE_URL`, `VideoJobs`, `AsyncVideoJobs`, `VideoJob`, `VideoJobError`, + `VideoMedia`, `VideoResult`. `AIPG`/`AsyncAIPG` are aliases of `Grid`/`AsyncGrid` and must stay so. + Candidate payment exports add `VideoAuthorizationRequest`, + `SignedVideoAuthorization`, `VideoPaymentError`, `prepare_video_authorization`, + `sign_video_authorization` and `async_sign_video_authorization`. - **Keep `__version__` (in `__init__.py`) in sync with `version` in `pyproject.toml`.** - Package metadata, source SPDX headers, and `LICENSE` all use MIT. +- Candidate video helpers are independently implemented MIT clients for the + wire protocol, not copies of AGPL Core/Director implementation. They never + retry POST, download media, hold funds, or claim a closed job was refunded. + Payment helpers independently implement the public quote/EIP-712 wire contract + without adding dependencies. They perform no HTTP and never take keys or + send; actual preview quotes refuse signer callbacks. A returned signature + is not funding, account/contract verification or a payment rail. + Fresh signing requires binding-v3 upfront/EIP-3009 terms plus explicit + `accept_upfront_payment=True`; default/false refuses callbacks even when + enabled. Earlier quotes are historical-only. Keep this policy aligned with JS. - **Release identity:** publish only a non-prerelease GitHub Release whose `v` tag exactly matches `pyproject.toml` and `__version__`. The release workflow imports the wheel with runtime dependencies only, then runs @@ -93,3 +112,4 @@ submit/poll). The retired horde `/api/v2` async queue is NOT used. Its PyPI pack - [src/grid_sdk/AGENTS.md](src/grid_sdk/AGENTS.md) — the SDK package: clients + raw-Grid access. - [tests/AGENTS.md](tests/AGENTS.md) — mocked Python SDK contract tests. +- [qualification/AGENTS.md](qualification/AGENTS.md) - cross-repo video jobs qualification. diff --git a/README.md b/README.md index 3eed34b..c2c4e5f 100644 --- a/README.md +++ b/README.md @@ -104,8 +104,121 @@ result = client.grid.generate({"prompt": "...", "models": [...], "params": {...} ``` `client.grid` waits for the synchronous Grid response and returns the finished -OpenAI-shaped result. Use `timeout` for long media jobs; there is no SDK -submit/poll mode on this `/v1` client. +OpenAI-shaped result. Use `timeout` for long media jobs. These existing calls +do not use the separate video-jobs lifecycle below. + +## Candidate Durable Video Jobs + +This branch's `0.2.0a3` adds `client.grid.video_jobs` (and its `AsyncGrid` +twin). It requires the candidate Core `/v1/videos/jobs` contract, which is +default-off and **not a public production feature yet**. Installing this SDK +does not enable admission, qualify a worker or authorize spending. + +```python +from uuid import uuid4 + +jobs = client.grid.video_jobs +key = str(uuid4()) +body = { + "model": "LTX-2.5 Distilled BF16", + "prompt": "A person waves from inside a parked car", + "frames": 33, + "fps": 24, + "max_cost_micro_usd": 50_000, # $0.05 ceiling, not a quote + # Audio-to-video: reference_audio="data:audio/wav;base64,..." + # Video-to-video: also reference_video="data:video/mp4;base64,..." + # Video reference requires an explicit size, e.g. size="768x512". +} +# Durably save key + immutable body in your application's private journal FIRST. +# Do not generate a fresh key when recovering this operation. +receipt = jobs.submit(body, idempotency_key=key) +# Save the accepted receipt before waiting or downloading any result. +terminal = jobs.wait(receipt, timeout=600) +``` + +`submit` checks `/health` contract features without credentials, then sends +**one POST**. Core still owns worker availability, recipe/input validation, +live pricing and the atomic credit hold. There is no model fallback, automatic +POST retry, asset download or client-side billing counter. Costs are integer +micro-USD, limited to `2**53 - 1` for exact parity with JavaScript. + +If the reply is lost, malformed, cancelled or fails after POST begins, preserve +the saved key and use `jobs.recover(idempotency_key=key)`. Once you have a job +ID, use `jobs.recover(job_id=receipt["job_id"])` or `jobs.wait(receipt)`. +Recovery is **GET-only**, does not re-upload sources, and bypasses capability +preflight so accepted jobs remain recoverable after new admission is disabled. +404/409/5xx are surfaced; none triggers another submission. A `VideoJobError` +reports sanitized `kind`, `http_status`, and `submission_uncertain`; HTTP errors +after POST are conservatively uncertain, even if Core probably rejected them. +Cancellation also requires recovery; it does not prove the job was cancelled. + +`wait` returns `completed` or `closed_without_result`, validating stable job, +model, request hash, held amount and ceiling on every poll. The latter means +no recoverable output, **not proof of a refund**. Timeout does not release a +hold or start another render. The returned hashes are metadata commitments; +the SDK does not download, hash or quality-check the output. + +For delegated service calls, pass `user_token=...` to submit/recover/wait; +it travels only in `X-Grid-User-Token`, never a URL. Keep service credentials +server-side. The SDK refuses redirects on the new authenticated paths. + +Source inputs must be inline, server-validated WAV/MP4, not arbitrary remote +URLs. Audio uses stereo PCM16 WAV at 48 kHz, exactly `frames * 2000` samples +at 24 fps. Video uses frame-matched 24-fps H.264 MP4 plus a separate WAV and +explicit output size. Current LTX frame/recipe gates remain authoritative. +An `image` or timed `keyframes` pin constrains actual frames: **it is not an +appearance-only identity reference**. + +The separate, unpriced **`LTX-2.5 Dev BF16 Ingredients`** candidate accepts +`reference_image` as a 768x448 PNG, exact-duration `reference_audio`, and optional +frame-matched `reference_video`, at 121 frames/24 fps, 40 steps and CFG 2. +Use explicit `size="768x448"` and no `image`/`keyframes`. `reference_strength` +defaults to 1; with video, `guide_strength` defaults to 0.5 and +`motion_attention_strength` to 0.25. Optional `reference_attention_strength` +is independently bounded to (0, 1]; omission preserves the original appearance +guide, rather than silently choosing a new default. Core validates these controls and decodes +every source. The helper forwards them; it never executes an adapter or turns +the PNG into frame zero. Source-health features do not mean this recipe is +admitted, priced or visually qualified. Private native test-credit lifecycle +evidence exists; public/purchased-credit acceptance, measured quality and +model-use clearance remain gates. x402 video payments are a proposal, not an +implemented SDK payment rail. + +## Candidate Video Wallet Preparation + +`prepare_video_authorization` checks a trusted, authenticated Core quote against +your independently retained model, normalized intent hash, reviewed recipient, +network and exact spending ceiling. It returns unsigned EIP-712 data without +keys, HTTP or dependencies. +New binding-v3 quotes must disclose upfront settlement and EIP-3009. Earlier +quotes are historical-only, not fresh wallet requests. Upfront means payment +before rendering; explicitly accept it only after reviewing that policy. + +```python +from grid_sdk import prepare_video_authorization + +request = prepare_video_authorization( + quote, + payer=wallet_address, + expected_model=model, + expected_request_hash=saved_intent_hash, + expected_recipient=reviewed_recipient, + network="eip155:84532", + max_cost_micro_usd=100_000, + accept_upfront_payment=True, # Explicitly accept payment before rendering. +) +``` + +Quotes are private/default-off and return `payment_enabled: false`. +`sign_video_authorization` and its async twin block those before invoking a +wallet callback. Do not flip that field yourself or prompt a real wallet for +this candidate. Enabled quotes still refuse signing unless upfront acceptance +was explicitly true; omitted/false policies never invoke the wallet. No payment +endpoint or sender exists. Future enabled callbacks +sign only, with expiry checked afterwards; Core must verify signatures, live +accounts/tokens, nonce claims and actual funds. Never log or persist +`SignedVideoAuthorization.payload` as ordinary evidence. A signed authorization +is not a completed payment or generation. ## It's just OpenAI underneath @@ -114,6 +227,29 @@ tool calling, structured output, the full `.chat`/`.images`/`.models` surface works here too. You can also point existing OpenAI code at the Grid by setting `base_url="https://api.aipowergrid.io/v1"` if you'd rather not switch packages. +## Candidate Qualification + +Unit tests mock all HTTP (`pytest tests`). Optional cross-repo tests exercise +Core's real scoped authentication and SQL holds, but synthetic storage/workers, +including accepted-reply loss and recovery after admission rollback. Build the +JS peer first, then run from this SDK checkout with Core's locked test Python: + +```bash +PYTHONPATH=/absolute/path/to/candidate-core \ +AIPG_JS_VIDEO_QUALIFICATION=/absolute/path/to/js-peer/qualification/core-wire.mjs \ +/absolute/path/to/core-test-python -m pytest qualification/test_core_video_jobs.py +``` + +Use an isolated scratch `CREDITS_TEST_DB_URL` for PostgreSQL; otherwise its +cases skip and only SQLite is proved. Source decoder skips are not successful +qualification. This harness does not run GPUs, use R2 or spend live credits. +The JavaScript transport cases use unmocked fetch against a temporary IPv4 +loopback Core server, with fixture credentials supplied on stdin and no Core +lifespan/background tasks. They require one actual POST, GET-only request and +terminal recovery, plus real 401/403/404 ownership/scope checks. Local socket +permission is required. Reply loss is a post-commit 503, not a network-outage +test; the synthetic completion does not qualify actual video output. + ## Config | | | diff --git a/pyproject.toml b/pyproject.toml index 4f52758..ec14df6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "grid-sdk" -version = "0.1.2" +version = "0.2.0a3" description = "Python SDK for AI Power Grid open-model, OpenAI-compatible inference." readme = "README.md" requires-python = ">=3.9" @@ -30,3 +30,4 @@ where = ["src"] [tool.pytest.ini_options] asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/qualification/AGENTS.md b/qualification/AGENTS.md new file mode 100644 index 0000000..b816744 --- /dev/null +++ b/qualification/AGENTS.md @@ -0,0 +1,57 @@ +# SDK/Core Qualification + +## Purpose + +Optional cross-repo checks against an explicitly selected Core checkout and +built JavaScript SDK. Not part of runtime packages or ordinary SDK unit tests. + +## Contracts + +- Run with Core's locked Python dependencies and `PYTHONPATH` pointing to the + candidate Core root. The normal SDK test environment cannot run these checks. +- Import Core's scoped-auth/SQL fixtures, not production credentials or URLs. + Storage, workers, quotes and output bytes are synthetic; SQL/auth are real. +- `AIPG_JS_VIDEO_QUALIFICATION` names the JS peer's `qualification/core-wire.mjs`. + Build that peer first. Missing peer configuration fails instead of silently + skipping its parity check. +- Exercise generated/audio/video modes with/without lost accepted replies; + require one job/hold/payout and one submission per SDK. Recovery after + admission rollback must be GET-only. +- Also exercise separate Dev Ingredients audio/appearance and motion/appearance + profiles, with reference attention omitted or explicit, using decoded + synthetic PNG/WAV/MP4 and an explicit isolated fixture tariff. Require the + appearance worker feature, no opening pin, distinct private bindings and + 100,834-micro hold/charge. This does not price/admit the real candidate or + qualify native GPU/R2/identity, even when the SQL proof runs on Postgres. +- Real JavaScript transport cases additionally serve the isolated router on an + ephemeral IPv4 loopback socket and run the peer's `core-http.mjs`, without + mocking fetch. Supply fixture credentials over stdin only and clear the child + environment except PATH. These include both appearance modes/attention + choices, not just the ordinary source protocol. Require unauthenticated + health, one actual POST, the committed guide/control, owner/key GET recovery, + and real 401/403/404 isolation after settlement and + admission rollback. Stop the child and local server in every terminal path. + Lost replies here are a post-commit 503, not a physical network outage. +- Use real source decoders for synthetic MP4/WAV. Missing FFmpeg skips are not + successful media qualification. PostgreSQL runs require an isolated scratch + `CREDITS_TEST_DB_URL`; never use a live Grid DB. + +## Verification + +Run `pytest qualification/test_core_video_jobs.py` with those dependencies and +paths. These checks do not qualify GPU output, funded R2 lifecycle or production. + +`test_core_video_payments.py` uses real scoped quote HTTP/SQL, decoded +generated/audio/video fixtures and the standalone MIT payment module. Set +`AIPG_JS_VIDEO_PAYMENTS_QUALIFICATION` to the built peer's +`qualification/video-payments.mjs`; optionally choose its Node executable with +`AIPG_VIDEO_QUALIFICATION_NODE`. Missing peer configuration fails. Both clients +must block signing the actual preview quote; only a local future-enabled copy +exercises new unfunded signatures and Core's actual verifier. Private two-phase +IPC never carries keys/credentials; synthetic signatures are transient private +data, not logged evidence. Terminate children on every path and require +unchanged balance plus zero holds/jobs/ledgers. +Both SDKs must refuse callbacks on an enabled test copy when upfront consent is +omitted or false, then sign only with explicit acceptance of binding-v3 terms. +This is protocol/crypto proof, not wallet UI, live contract/facilitator, receipts, +replay exclusion or funding. diff --git a/qualification/test_core_video_jobs.py b/qualification/test_core_video_jobs.py new file mode 100644 index 0000000..6585108 --- /dev/null +++ b/qualification/test_core_video_jobs.py @@ -0,0 +1,543 @@ +# SPDX-License-Identifier: MIT +"""Optional Core SQL/auth qualification; never uses a live Grid or funds.""" + +import asyncio +import base64 +import contextlib +import importlib.util +import io +import json +import os +import shutil +import socket +import subprocess +import sys +from collections.abc import AsyncIterator +from pathlib import Path +from typing import Any, cast +from uuid import uuid4 + +import httpx +import pytest +from fastapi import FastAPI +from grid_api.routers import video_jobs, videos +from grid_api.routers.tests.test_video_jobs import stored +from grid_api.services import accounts, credits, media_jobs, pricing +from grid_api.services.tests.test_media_job_journal import ( + MODEL, + counts, + owner, +) +from grid_api.services.tests.test_video_sources import wav +from PIL import Image + +pytest_plugins = ( + "grid_api.routers.tests.test_video_jobs", + "grid_api.services.tests.test_video_sources", +) + +# Load the MIT wire module without importing the OpenAI convenience layer into +# Core's dependency environment. Normal SDK tests cover package/client wiring. +spec = importlib.util.spec_from_file_location( + "sdk_video_qualification", Path(__file__).parents[1] / "src/grid_sdk/video_jobs.py" +) +module = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = module +spec.loader.exec_module(module) + +MODES = [ + "generated", + "audio", + "video", + "appearance-audio", + "appearance-video", + "appearance-audio-attention", + "appearance-video-attention", +] + + +@pytest.fixture +def video_body(mode, api, mp4, tmp_path, monkeypatch): + appearance = mode.startswith("appearance-") + frames = 121 if appearance else 9 + model = "LTX-2.5 Dev BF16 Ingredients" if appearance else MODEL + if appearance: + api[-1].return_value = [model] + # Fixture tariff only; the actual Ingredients recipe stays dark/unpriced. + monkeypatch.setitem( + pricing.PRICING, + model.lower(), + pricing.ModelPrice(0, 0, video_per_second=0.02), + ) + body = { + "model": model, + "prompt": "A synthetic red ball rolls", + "frames": frames, + "fps": 24, + "size": "768x448" if appearance else "256x256", + "seed": 42, + "max_cost_micro_usd": 150_000 if appearance else 100_000, + } + if mode != "generated": + body["reference_audio"] = base64.b64encode(wav(frames)).decode("ascii") + if appearance: + image = io.BytesIO() + Image.new("RGB", (768, 448), "red").save(image, "PNG") + body["reference_image"] = base64.b64encode(image.getvalue()).decode("ascii") + body.update(steps=40, cfg_scale=2.0, reference_strength=1.0) + if mode.endswith("-attention"): + body["reference_attention_strength"] = 0.25 + if "video" in mode: + if appearance: + decoder = shutil.which("ffmpeg") + if not decoder or not shutil.which("ffprobe"): + pytest.skip("independent FFmpeg source-decoder gate unavailable") + path = tmp_path / "appearance-motion.mp4" + subprocess.run( + [ + decoder, + "-v", + "error", + "-nostdin", + "-f", + "lavfi", + "-i", + "testsrc2=size=768x448:rate=24", + "-frames:v", + str(frames), + "-an", + "-c:v", + "libx264", + "-threads", + "1", + "-pix_fmt", + "yuv420p", + str(path), + ], + capture_output=True, + timeout=15, + check=True, + env={"PATH": os.environ["PATH"]}, + ) + mp4 = path.read_bytes() + body["reference_video"] = base64.b64encode(mp4).decode("ascii") + return body + + +def assert_source_contract(row, body): + appearance = "reference_image" in body + mode = ( + "video" + if "reference_video" in body + else "audio" + if "reference_audio" in body + else "generated" + ) + inputs = row["payload"]["recipe_inputs"] + assert inputs.get("source_mode", "generated") == mode + if appearance: + assert row["payload"]["appearance_profile"] == "ingredients-dev-v1" + assert "source_image_url" not in row["payload"] + roles = ["reference_image", "reference_audio"] + if mode == "video": + roles.append("reference_video") + assert [item["role"] for item in row["payload"]["source_media"]] == roles + assert inputs.get("reference_attention_strength") == body.get( + "reference_attention_strength" + ) + guide = row["payload"]["recipe_spec"]["reference_guide"] + if "reference_attention_strength" in body: + assert guide["class_type"] == "LTXAddVideoICLoRAGuideAdvanced" + assert ( + guide["inputs"]["attention_strength"] + == body["reference_attention_strength"] + ) + else: + assert guide["class_type"] == "LTXAddVideoICLoRAGuide" + assert "attention_strength" not in guide["inputs"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("lost_reply", [False, True]) +async def test_sdk_core_auth_hold_recovery_and_js_parity( + api, video_body, monkeypatch, mode, lost_reply +): + client, aid, credential, settings, quota, _available = api + body = video_body + model, frames = body["model"], body["frames"] + app = client._transport.app + features = [module.FEATURE, module.SOURCE_FEATURE] + + @app.get("/health") + async def synthetic_health(): + return {"features": features} + + original_client = httpx.AsyncClient + monkeypatch.setattr( + module.httpx, + "AsyncClient", + lambda **kwargs: original_client( + transport=httpx.ASGITransport(app=app), **kwargs + ), + ) + monkeypatch.setattr( + videos.video_sources.storage, + "upload_private_source", + lambda raw, extension: ( + "https://" + + "a" * 32 + + ".r2.cloudflarestorage.com/private-synthetic-source/source/" + + "b" * 32 + + "." + + extension + + "?X-Amz-Signature=synthetic" + ), + ) + key = str(uuid4()) + if lost_reply: + reserve = media_jobs.reserve + + async def commit_then_lose(*args, **kwargs): + await reserve(*args, **kwargs) + raise ConnectionError("synthetic lost accepted response") + + monkeypatch.setattr(media_jobs, "reserve", commit_then_lose) + jobs = module.AsyncVideoJobs(credential, "http://test/v1") + if lost_reply: + with pytest.raises(module.VideoJobError) as error: + await jobs.submit(body, idempotency_key=key) + assert error.value.submission_uncertain and error.value.http_status == 503 + pending = await jobs.recover(idempotency_key=key) + else: + pending = await jobs.submit(body, idempotency_key=key) + assert pending["state"] == "pending_dispatch" + assert await counts() == (1, 1, 0) + quota.assert_awaited_once() + row = await stored(pending["job_id"]) + assert_source_contract(row, body) + if "reference_image" in body: + assert pending["reserved_micro"] == 100_834 + assert pending["request_hash"] == video_jobs.request_hash( + video_jobs.AsyncVideoRequest.model_validate(body) + ) + worker, token = uuid4(), str(uuid4()) + assert ( + await media_jobs.claim_execution( + job_id=row["job_id"], + payload=row["payload"], + route_models=row["route_models"], + preferred_worker="", + selected_model=model, + worker_id=worker, + job_type="video", + execution_token=token, + ) + == "claimed" + ) + media = { + "model": model, + "worker": "synthetic-worker", + "gen_time": 1.0, + "recipe_root": row["payload"]["recipe_root"], + "media": [ + { + "url": "https://media.invalid/video.mp4", + "key": "synthetic/video.mp4", + "sha256": "a" * 64, + "seed": 42, + } + ], + } + values = dict( + job_id=row["job_id"], + worker_id=str(worker), + wallet="", + model=model, + job_type="video", + den=1.0, + output_units=frames, + prompt_hash="b" * 64, + result_hash="c" * 64, + ) + assert ( + await credits.record_and_settle( + ledger_values=values, + exact=True, + media_result=media, + media_execution_token=token, + ) + == "settled" + ) + settings.media_async_enabled = False + features.clear() + terminal = await jobs.wait(pending) + assert ( + terminal["result"] == media + and terminal["actual_micro"] == pending["reserved_micro"] + ) + assert await counts() == (1, 1, 1) + peer = os.environ.get("AIPG_JS_VIDEO_QUALIFICATION", "") + assert peer and Path(peer).is_file(), ( + "explicit built JS peer qualification path required" + ) + node = subprocess.run( + ["node", peer], + input=json.dumps( + { + "body": body, + "key": key, + "pending": pending, + "terminal": terminal, + "lost": lost_reply, + } + ), + text=True, + capture_output=True, + timeout=15, + check=True, + ) + proof = json.loads(node.stdout) + assert proof["body"] == body and proof["post_count"] == 1 + assert proof["terminal"] == terminal + assert ( + video_jobs.request_hash( + video_jobs.AsyncVideoRequest.model_validate(proof["body"]) + ) + == pending["request_hash"] + ) + + +@contextlib.asynccontextmanager +async def loopback_core(app: FastAPI) -> AsyncIterator[str]: + """Serve only this test's router; no Core lifespan or production loops.""" + import uvicorn + + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener: + listener.bind(("127.0.0.1", 0)) + listener.listen(16) + listener.setblocking(False) + port = listener.getsockname()[1] + server = uvicorn.Server( + uvicorn.Config(app, lifespan="off", access_log=False, log_level="critical") + ) + serving = asyncio.create_task(server.serve(sockets=[listener])) + try: + + async def ready() -> None: + while not server.started: + if serving.done(): + await serving + raise RuntimeError("isolated Core server did not start") + await asyncio.sleep(0.01) + + await asyncio.wait_for(ready(), timeout=10) + yield f"http://127.0.0.1:{port}" + finally: + server.should_exit = True + await asyncio.wait_for(serving, timeout=10) + + +async def child_record(process: asyncio.subprocess.Process) -> dict[str, Any]: + assert process.stdout is not None + line = await asyncio.wait_for(process.stdout.readline(), timeout=15) + assert line and len(line) <= 128 * 1024, "bounded SDK qualification reply required" + value = json.loads(line) + assert value.get("stage") != "failed", ( + "real SDK/Core transport rejected the fixture" + ) + return cast(dict[str, Any], value) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("lost_reply", [False, True]) +async def test_js_sdk_real_http_admission_and_recovery( + api, video_body, monkeypatch, mode, lost_reply +): + client, aid, credential, settings, quota, _available = api + body = video_body + model, frames = body["model"], body["frames"] + app = client._transport.app + features = [module.FEATURE, module.SOURCE_FEATURE] + trace, submissions = [], [] + + @app.middleware("http") + async def observe_request(request, call_next): + trace.append( + ( + request.method, + request.url.path, + bool( + request.headers.get("apikey") + or request.headers.get("authorization") + ), + ) + ) + if request.method == "POST": + submissions.append(json.loads(await request.body())) + return await call_next(request) + + @app.get("/health") + async def synthetic_health(): + return {"features": features} + + monkeypatch.setattr( + videos.video_sources.storage, + "upload_private_source", + lambda _raw, extension: ( + "https://" + + "a" * 32 + + ".r2.cloudflarestorage.com/private-synthetic-source/source/" + + "b" * 32 + + "." + + extension + + "?X-Amz-Signature=synthetic" + ), + ) + if lost_reply: + reserve = media_jobs.reserve + + async def commit_then_lose(*args, **kwargs): + await reserve(*args, **kwargs) + raise ConnectionError("synthetic lost accepted response") + + monkeypatch.setattr(media_jobs, "reserve", commit_then_lose) + configured_peer = os.environ.get("AIPG_JS_VIDEO_QUALIFICATION", "") + assert configured_peer, "explicit built JS peer qualification path required" + peer = Path(configured_peer).with_name("core-http.mjs") + assert peer.is_file(), "explicit built JS peer qualification path required" + key = str(uuid4()) + async with loopback_core(app) as origin: + process = await asyncio.create_subprocess_exec( + "node", + str(peer), + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env={"PATH": os.environ["PATH"]}, + limit=128 * 1024, + ) + try: + process.stdin.write( + ( + json.dumps( + { + "origin": origin, + "credential": credential, + "body": body, + "key": key, + "lost": lost_reply, + } + ) + + "\n" + ).encode() + ) + await process.stdin.drain() + pending = (await child_record(process))["pending"] + assert pending["state"] == "pending_dispatch" + assert pending["request_hash"] == video_jobs.request_hash( + video_jobs.AsyncVideoRequest.model_validate(body) + ) + assert submissions == [body] and await counts() == (1, 1, 0) + quota.assert_awaited_once() + accepted_requests = len(trace) + row = await stored(pending["job_id"]) + assert_source_contract(row, body) + if "reference_image" in body: + assert pending["reserved_micro"] == 100_834 + worker, token = uuid4(), str(uuid4()) + assert ( + await media_jobs.claim_execution( + job_id=row["job_id"], + payload=row["payload"], + route_models=row["route_models"], + preferred_worker="", + selected_model=model, + worker_id=worker, + job_type="video", + execution_token=token, + ) + == "claimed" + ) + result = { + "model": model, + "worker": "synthetic-worker", + "gen_time": 1.0, + "recipe_root": row["payload"]["recipe_root"], + "media": [ + { + "url": "https://media.invalid/video.mp4", + "key": "synthetic/video.mp4", + "sha256": "a" * 64, + "seed": 42, + } + ], + } + assert ( + await credits.record_and_settle( + ledger_values={ + "job_id": row["job_id"], + "worker_id": str(worker), + "wallet": "", + "model": model, + "job_type": "video", + "den": 1.0, + "output_units": frames, + "prompt_hash": "b" * 64, + "result_hash": "c" * 64, + }, + exact=True, + media_result=result, + media_execution_token=token, + ) + == "settled" + ) + settings.media_async_enabled = False + features.clear() + foreign_key = await accounts.issue_key( + await owner(), scopes=["inference.submit"] + ) + observer = await accounts.issue_key(aid, scopes=["account.read"]) + process.stdin.write( + ( + json.dumps( + { + "denied": [ + ["invalid-fixture", 401], + [observer, 403], + [foreign_key, 404], + ], + } + ) + + "\n" + ).encode() + ) + await process.stdin.drain() + proof = await child_record(process) + assert proof["stage"] == "completed" and proof["denied"] == [401, 403, 404] + terminal = proof["terminal"] + assert terminal["state"] == "completed" and terminal["result"] == result + assert terminal["actual_micro"] == pending["reserved_micro"] + assert await counts() == (1, 1, 1) + assert sum(method == "POST" for method, *_ in trace) == 1 + assert [item for item in trace if item[1] == "/health"] == [ + ("GET", "/health", False) + ] + assert all(authorized for _, path, authorized in trace if path != "/health") + assert all(method == "GET" for method, *_ in trace[accepted_requests:]) + assert {path for _, path, _ in trace} == { + "/health", + "/v1/videos/jobs", + "/v1/videos/jobs/" + pending["job_id"], + } + await asyncio.wait_for(process.wait(), timeout=10) + assert process.returncode == 0 + finally: + if process.returncode is None: + process.terminate() + try: + await asyncio.wait_for(process.wait(), timeout=5) + except asyncio.TimeoutError: + process.kill() + await process.wait() diff --git a/qualification/test_core_video_payments.py b/qualification/test_core_video_payments.py new file mode 100644 index 0000000..7aa4963 --- /dev/null +++ b/qualification/test_core_video_payments.py @@ -0,0 +1,165 @@ +# SPDX-License-Identifier: MIT +"""Real quote HTTP/SQL plus SDK/crypto IPC; no chain, payment, upload or GPU.""" + +import asyncio +import base64 +import importlib.util +import json +import os +import sys +from pathlib import Path +from uuid import uuid4 + +import pytest +from eth_account import Account +from eth_account.messages import encode_typed_data +from grid_api.routers.tests import test_video_quotes as router_tests +from grid_api.services import credits, video_eip3009, video_quotes +from grid_api.services.tests.test_media_job_journal import OPENING, counts +from grid_api.services.tests.test_video_sources import wav + +pytest_plugins = ( + "grid_api.routers.tests.test_video_quotes", + "grid_api.services.tests.test_video_sources", +) + +# Keep the OpenAI convenience layer out of Core's dependency environment. +spec = importlib.util.spec_from_file_location( + "sdk_video_payment_qualification", + Path(__file__).parents[1] / "src/grid_sdk/video_payments.py", +) +sdk = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = sdk +spec.loader.exec_module(sdk) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["generated", "audio", "video"]) +@pytest.mark.parametrize("network", ["eip155:8453", "eip155:84532"]) +async def test_actual_quote_and_both_sdk_wallet_callbacks_match_core( + quote_api, mp4, mode, network +): + client, aid, credential, *_ = quote_api + quote_api[3].video_x402_quote_network = network + key = uuid4() + body = {"frames": 9, "size": "256x256", "seed": 42} + if mode != "generated": + body["reference_audio"] = base64.b64encode(wav()).decode("ascii") + if mode == "video": + body["reference_video"] = base64.b64encode(mp4).decode("ascii") + response = await router_tests.post(quote_api, key=key, body=body) + assert response.status_code == 201 + quote = response.json() + row = await video_quotes.lookup(aid, idempotency_key=str(key)) + wallet = Account.create() # Unfunded throwaway signer; never crosses IPC. + policy = { + "payer": wallet.address, + "expected_model": quote["model"], + "expected_request_hash": quote["request_hash"], + "expected_recipient": quote["payment"]["pay_to"], + "network": quote["payment"]["network"], + "max_cost_micro_usd": 100_000, + "accept_upfront_payment": True, + } + preview = sdk.prepare_video_authorization(quote, **policy) + called = [] + with pytest.raises(sdk.VideoPaymentError, match="payment_disabled"): + sdk.sign_video_authorization(preview, lambda data: called.append(data)) + assert not called + enabled_quote = {**quote, "payment_enabled": True} + no_consent = { + key: value for key, value in policy.items() if key != "accept_upfront_payment" + } + for consent_policy in (no_consent, {**no_consent, "accept_upfront_payment": False}): + unsigned_without_consent = sdk.prepare_video_authorization( + enabled_quote, **consent_policy + ) + with pytest.raises(sdk.VideoPaymentError, match="upfront_payment_not_accepted"): + sdk.sign_video_authorization( + unsigned_without_consent, lambda data: called.append(data) + ) + assert not called + unsigned = sdk.prepare_video_authorization(enabled_quote, **policy) + signed = sdk.sign_video_authorization( + unsigned, + lambda data: ( + "0x" + + wallet.sign_message(encode_typed_data(full_message=data)).signature.hex() + ), + ) + result = video_eip3009.verify(row, signed.payload, expected_payer=wallet.address) + assert result.job_id == quote["job_id"] and result.amount_micro == 7500 + assert signed.payload["signature"] not in repr(signed) + + script = os.environ.get("AIPG_JS_VIDEO_PAYMENTS_QUALIFICATION", "") + node = os.environ.get("AIPG_VIDEO_QUALIFICATION_NODE", "node") + if not script or not Path(script).is_file(): + pytest.fail("explicit built JS video-payments qualification required") + child = await asyncio.create_subprocess_exec( + node, + script, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env={"PATH": os.environ.get("PATH", "")}, + ) + try: + js_policy = { + "payer": wallet.address, + "expectedModel": policy["expected_model"], + "expectedRequestHash": policy["expected_request_hash"], + "expectedRecipient": policy["expected_recipient"], + "network": policy["network"], + "maxCostMicroUsd": policy["max_cost_micro_usd"], + "acceptUpfrontPayment": policy["accept_upfront_payment"], + } + child.stdin.write( + (json.dumps({"quote": quote, "policy": js_policy}) + "\n").encode() + ) + await child.stdin.drain() + first = await asyncio.wait_for(child.stdout.readline(), timeout=10) + assert 0 < len(first) <= 16 * 1024 + data = json.loads(first) + assert ( + data["phase"] == "unsigned" + and data["previewBlocked"] + and data["consentBlocked"] + and not data["called"] + ) + assert ( + data["authorization"] == unsigned.authorization + and data["typedData"] == unsigned.typed_data + ) + signature = ( + "0x" + + wallet.sign_message( + encode_typed_data(full_message=data["typedData"]) + ).signature.hex() + ) + child.stdin.write((json.dumps({"signature": signature}) + "\n").encode()) + await child.stdin.drain() + child.stdin.close() + second = await asyncio.wait_for(child.stdout.readline(), timeout=10) + assert 0 < len(second) <= 4096 + forwarded = json.loads(second) + assert forwarded["phase"] == "signed" and forwarded["hidden"] + assert ( + forwarded["quoteId"] == quote["quote_id"] + and forwarded["jobId"] == quote["job_id"] + ) + assert ( + video_eip3009.verify( + row, forwarded["payload"], expected_payer=wallet.address + ) + == result + ) + assert await asyncio.wait_for(child.wait(), timeout=10) == 0 + assert await child.stderr.read() == b"" + finally: + if child.returncode is None: + child.terminate() + await asyncio.wait_for(child.wait(), timeout=5) + assert await credits.get_balance(aid) == OPENING and await counts() == (0, 0, 0) + assert ( + await client.get(response.headers["location"], headers={"apikey": credential}) + ).json() == quote diff --git a/src/grid_sdk/AGENTS.md b/src/grid_sdk/AGENTS.md index e52afbc..a84d5c8 100644 --- a/src/grid_sdk/AGENTS.md +++ b/src/grid_sdk/AGENTS.md @@ -13,6 +13,10 @@ raw-Grid escape hatch (`grid.py`) for generation beyond the `/v1` surface. the `AIPG_API_KEY` env var, and `online_models()`. - **`grid.py`** — `GridRaw`/`AsyncGridRaw` (reached via `client.grid`) and `GridError`. POSTs synchronously to the `/v1` media endpoints with `httpx`. +- **`video_jobs.py`** — separate lazy `video_jobs` namespace, public typed + receipts and sanitized `VideoJobError`; synchronous and asynchronous twins. +- `video_payments.py` - dependency-free quote-policy checks, unsigned EIP-712 + preparation and explicit sync/async signer callbacks. No HTTP or sender. ## Local Contracts @@ -29,6 +33,40 @@ raw-Grid escape hatch (`grid.py`) for generation beyond the `/v1` surface. - `credits()` reads Core's canonical three-pocket account view; SDKs do not calculate allowances or maintain local free-use counters. - The sync and async classes are mirror twins — change both together. +- Candidate `video_jobs.submit(body, idempotency_key=...)` requires an explicit + model, prompt, positive exact integer micro-USD ceiling and caller-persisted + UUIDv4 key. It snapshots the body, checks credential-free `/health` features, + then sends one POST to `/videos/jobs`. Core owns pricing/admission/holds. +- `recover` requires exactly one canonical job UUID or header idempotency key. + Recovery/wait are GET-only, bypass capability preflight and never re-upload. + Wait bounds its deadline and checks stable job/model/request hash/hold/ceiling. + A closed job is not refund proof; completion metadata is not media QC. +- New video HTTP paths refuse redirects, bound JSON replies to 64 KiB, and + sanitize transport/HTTP errors. Delegated user tokens travel only in headers. + Request bodies are bounded and money is at most `2**53 - 1` for JS parity. +- `image`/`keyframes` are literal frame constraints, not character identity + references. Separate `reference_image` PNG plus explicit audio/size forwards + the dark Ingredients recipe contract without creating frame pins. Reject + mixing these references with `image`/`keyframes` before HTTP. Core owns + decoding, fixed profile/knob validation, worker capabilities and admission; + source-health flags alone do not qualify appearance or enable its price. + Optional `reference_attention_strength` passes through unchanged, separately + from denoising and motion attention. Do not insert a default or recompute + graph/intent authority in the client; Core owns those contracts. + No x402 collection or video execution is implemented by the SDK. +- Payment preparation requires a trusted Core binding-v3 upfront/EIP-3009 quote and independent + expected model/intent hash/recipient/network/ceiling. Check canonical USDC + domains, one committed nonce, exact integers and the five-minute window. + Do not copy the response hash as independent intent verification or claim + the hidden plan was proved. Client clocks are not chain/funding authority. +- Preview `payment_enabled:false` blocks callbacks. Enabled quotes also require + explicit `accept_upfront_payment=True`, never default/false/coerced consent. + Earlier binding revisions are historical-only, not fresh signing requests. + Enabled sync/async callbacks + receive fresh JSON-ready signing data; recheck expiry after wallet work, + sanitize failures and propagate cancellation. Byte format is not ECDSA or + live account verification. Signed payloads are private capabilities omitted + from repr, explicitly exported only. No auto-sign, HTTP, retries or raw keys. ## Work Guidance diff --git a/src/grid_sdk/__init__.py b/src/grid_sdk/__init__.py index 486763f..c0ea7c4 100644 --- a/src/grid_sdk/__init__.py +++ b/src/grid_sdk/__init__.py @@ -24,6 +24,40 @@ """ from .client import AIPG, AsyncAIPG, AsyncGrid, Grid, DEFAULT_BASE_URL +from .video_jobs import ( + AsyncVideoJobs, + VideoJob, + VideoJobError, + VideoJobs, + VideoMedia, + VideoResult, +) +from .video_payments import ( + SignedVideoAuthorization, + VideoAuthorizationRequest, + VideoPaymentError, + async_sign_video_authorization, + prepare_video_authorization, + sign_video_authorization, +) -__all__ = ["Grid", "AsyncGrid", "AIPG", "AsyncAIPG", "DEFAULT_BASE_URL"] -__version__ = "0.1.2" +__all__ = [ + "Grid", + "AsyncGrid", + "AIPG", + "AsyncAIPG", + "DEFAULT_BASE_URL", + "VideoJobs", + "AsyncVideoJobs", + "VideoJob", + "VideoJobError", + "VideoMedia", + "VideoResult", + "VideoAuthorizationRequest", + "SignedVideoAuthorization", + "VideoPaymentError", + "prepare_video_authorization", + "sign_video_authorization", + "async_sign_video_authorization", +] +__version__ = "0.2.0a3" diff --git a/src/grid_sdk/grid.py b/src/grid_sdk/grid.py index 4ddcf9f..ef2324a 100644 --- a/src/grid_sdk/grid.py +++ b/src/grid_sdk/grid.py @@ -26,6 +26,8 @@ import httpx +from .video_jobs import AsyncVideoJobs, VideoJobs + class GridError(RuntimeError): pass @@ -112,6 +114,12 @@ def __init__(self, api_key: str, base_url: str, timeout: float = 300.0): self._headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} self._base = _v1_base(base_url) self._timeout = timeout + self._api_key = api_key + + @property + def video_jobs(self) -> VideoJobs: + """Separate candidate async contract; existing generation is unchanged.""" + return VideoJobs(self._api_key, self._base) def _post(self, path: str, body: Dict[str, Any], timeout: float) -> Dict[str, Any]: with httpx.Client(timeout=timeout) as c: @@ -193,6 +201,11 @@ def __init__(self, api_key: str, base_url: str, timeout: float = 300.0): self._headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} self._base = _v1_base(base_url) self._timeout = timeout + self._api_key = api_key + + @property + def video_jobs(self) -> AsyncVideoJobs: + return AsyncVideoJobs(self._api_key, self._base) async def _post(self, path: str, body: Dict[str, Any], timeout: float) -> Dict[str, Any]: async with httpx.AsyncClient(timeout=timeout) as c: diff --git a/src/grid_sdk/video_jobs.py b/src/grid_sdk/video_jobs.py new file mode 100644 index 0000000..1c93e56 --- /dev/null +++ b/src/grid_sdk/video_jobs.py @@ -0,0 +1,574 @@ +# SPDX-License-Identifier: MIT +"""Candidate credit-backed video jobs. No POST retries or asset downloads.""" + +from __future__ import annotations + +import asyncio +import json +import math +import re +import time +from dataclasses import dataclass +from typing import Dict, List, Literal, Mapping, Optional, TypedDict, cast +from urllib.parse import urlsplit +from uuid import UUID + +import httpx + +SCHEMA = "aipg.video-job.v1" +FEATURE = "video-async-credit-jobs-v1" +SOURCE_FEATURE = "video-source-conditioning-v1" +MAX_JSON = 64 * 1024 +MAX_BODY = 64 * 1024 * 1024 + MAX_JSON +MAX_MONEY = 2**53 - 1 # Exact in both SDK languages. +STATES = { + "pending_dispatch", + "queued", + "execution_uncertain", + "completed", + "closed_without_result", +} +TERMINAL = {"completed", "closed_without_result"} + + +class VideoMedia(TypedDict): + url: str + key: str + sha256: str + seed: Optional[int] + + +class VideoResult(TypedDict): + model: str + worker: str + gen_time: float + recipe_root: Optional[str] + media: List[VideoMedia] + + +class VideoJob(TypedDict): + schema: str + job_id: str + model: str + state: Literal[ + "pending_dispatch", + "queued", + "execution_uncertain", + "completed", + "closed_without_result", + ] + request_hash: str + reserved_micro: int + max_cost_micro: int + actual_micro: Optional[int] + result: Optional[VideoResult] + + +class VideoJobError(RuntimeError): + """Sanitized failure. Preserve the key/receipt when submission is uncertain.""" + + def __init__( + self, + kind: str, + *, + http_status: Optional[int] = None, + submission_uncertain: bool = False, + ): + self.kind = kind + self.http_status = http_status + self.submission_uncertain = submission_uncertain + super().__init__( + f"Grid video job {kind}" + (f" (HTTP {http_status})" if http_status else "") + ) + + +def _positive(value: float) -> float: + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not 0 < value <= 2_147_483.647 + or not math.isfinite(value) + ): + raise ValueError("timeouts and poll intervals must be positive and finite") + return value + + +def _uuid(value: str, *, key: bool = False) -> str: + try: + parsed = UUID(value) + except (ValueError, TypeError, AttributeError): + raise ValueError( + "use a canonical UUIDv4 key or canonical UUID job id" + ) from None + if str(parsed) != value or (key and parsed.version != 4): + raise ValueError("use a canonical UUIDv4 key or canonical UUID job id") + return value + + +def _headers( + api_key: str, user_token: Optional[str], key: Optional[str] +) -> Dict[str, str]: + values = [api_key] + ([user_token] if user_token is not None else []) + if any( + not isinstance(value, str) + or not value + or any(not 32 <= ord(char) <= 126 for char in value) + for value in values + ): + raise ValueError("credentials must be nonempty printable ASCII header values") + result = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + if user_token is not None: + result["X-Grid-User-Token"] = user_token + if key is not None: + result["Idempotency-Key"] = _uuid(key, key=True) + return result + + +def _selector(job_id: Optional[str], key: Optional[str]) -> str: + if (job_id is None) == (key is None): + raise ValueError("provide exactly one of job_id and idempotency_key") + return "/videos/jobs/" + _uuid(job_id) if job_id is not None else "/videos/jobs" + + +def _money(value: object, *, zero: bool = False) -> bool: + return type(value) is int and (0 if zero else 1) <= value <= MAX_MONEY + + +def _status(value: object) -> VideoJob: + if not isinstance(value, dict): + raise VideoJobError("invalid_response") + try: + _uuid(value["job_id"]) + state = value["state"] + if ( + value["schema"] != SCHEMA + or state not in STATES + or not isinstance(value["model"], str) + or not 1 <= len(value["model"]) <= 255 + or not isinstance(value["request_hash"], str) + or not re.fullmatch("[0-9a-f]{64}", value["request_hash"]) + or not _money(value["reserved_micro"]) + or not _money(value["max_cost_micro"]) + or value["reserved_micro"] > value["max_cost_micro"] + ): + raise ValueError + actual, result = value["actual_micro"], value["result"] + if actual is not None and ( + not _money(actual, zero=True) or actual > value["max_cost_micro"] + ): + raise ValueError + if state not in TERMINAL and (actual is not None or result is not None): + raise ValueError + if state == "completed": + if ( + actual is None + or not isinstance(result, dict) + or result["model"] != value["model"] + ): + raise ValueError + if ( + not isinstance(result["worker"], str) + or len(result["worker"]) > 255 + or isinstance(result["gen_time"], bool) + or not isinstance(result["gen_time"], (int, float)) + or not math.isfinite(result["gen_time"]) + or result["gen_time"] < 0 + or not isinstance(result["media"], list) + or len(result["media"]) != 1 + or ( + result.get("recipe_root") is not None + and ( + not isinstance(result["recipe_root"], str) + or len(result["recipe_root"]) > 255 + ) + ) + ): + raise ValueError + item = result["media"][0] + url = urlsplit(item["url"]) + if ( + url.scheme not in {"http", "https"} + or not url.hostname + or url.username + or url.password + or not isinstance(item["key"], str) + or not 1 <= len(item["key"]) <= 2048 + or not re.fullmatch("[0-9a-f]{64}", item["sha256"]) + or ( + item.get("seed") is not None and not _money(item["seed"], zero=True) + ) + ): + raise ValueError + elif result is not None: + raise ValueError + except (KeyError, ValueError, TypeError, AttributeError, IndexError): + raise VideoJobError("invalid_response") from None + return cast(VideoJob, value) + + +def _bound(current: VideoJob, expected: Mapping[str, object]) -> VideoJob: + fields = cast(Mapping[str, object], current) + if any( + fields[field] != expected[field] + for field in ( + "job_id", + "model", + "request_hash", + "reserved_micro", + "max_cost_micro", + ) + ): + raise VideoJobError("changed_identity") + return current + + +@dataclass(frozen=True) +class _Submission: + encoded: bytes + model: str + ceiling: int + source: bool + + +def _submission(body: Mapping[str, object]) -> _Submission: + try: + encoded = json.dumps(dict(body), allow_nan=False, separators=(",", ":")).encode( + "utf-8" + ) + if len(encoded) > MAX_BODY: + raise ValueError + snapshot = json.loads(encoded) + if ( + not isinstance(snapshot.get("model"), str) + or not 1 <= len(snapshot["model"]) <= 255 + or not isinstance(snapshot.get("prompt"), str) + or not 1 <= len(snapshot["prompt"]) <= 8000 + or not _money(snapshot.get("max_cost_micro_usd")) + or type(snapshot.get("n", 1)) is not int + or snapshot.get("n", 1) != 1 + or (snapshot.get("image") and snapshot.get("keyframes")) + or ( + snapshot.get("reference_video") + and (not snapshot.get("reference_audio") or not snapshot.get("size")) + ) + or ( + snapshot.get("reference_image") + and ( + snapshot.get("image") + or snapshot.get("keyframes") + or not snapshot.get("reference_audio") + or not snapshot.get("size") + ) + ) + ): + raise ValueError + except (ValueError, TypeError, OverflowError, RecursionError): + raise ValueError( + "invalid video body; explicit model, prompt and positive exact cost ceiling required" + ) from None + return _Submission( + encoded, + snapshot["model"], + snapshot["max_cost_micro_usd"], + bool( + snapshot.get("reference_audio") + or snapshot.get("reference_video") + or snapshot.get("reference_image") + ), + ) + + +def _decode(raw: bytes) -> object: + try: + return json.loads(raw, parse_constant=_invalid_constant) + except (ValueError, UnicodeError, RecursionError): + raise VideoJobError("invalid_response") from None + + +def _invalid_constant(value: str) -> object: + raise ValueError("nonfinite JSON is invalid") + + +def _features(value: object) -> List[str]: + if ( + not isinstance(value, dict) + or not isinstance(value.get("features"), list) + or not all(isinstance(item, str) for item in value["features"]) + ): + raise VideoJobError("invalid_capabilities") + return cast(List[str], value["features"]) + + +def _accepted(value: object, code: int, submission: _Submission) -> VideoJob: + status = _status(value) + if ( + status["model"] != submission.model + or status["max_cost_micro"] != submission.ceiling + or code != (200 if status["state"] in TERMINAL else 202) + ): + raise VideoJobError("invalid_response") + return status + + +class _VideoBase: + def __init__(self, api_key: str, base_url: str): + url = urlsplit(base_url) + if ( + url.scheme not in {"http", "https"} + or not url.hostname + or url.username + or url.password + or url.query + or url.fragment + or not url.path.rstrip("/").endswith("/v1") + ): + raise ValueError( + "video jobs require an http(s) /v1 base without credentials, query or fragment" + ) + _headers(api_key, None, None) + self._key = api_key + self._base = base_url.rstrip("/") + self._health = self._base[:-3] + "/health" + + +class VideoJobs(_VideoBase): + """Persist key + body before submit; recover/wait only GET the existing job.""" + + def _request( + self, + method: str, + url: str, + headers: Dict[str, str], + timeout: float, + body: Optional[bytes] = None, + ) -> tuple[object, int]: + deadline = time.monotonic() + _positive(timeout) + try: + with httpx.Client(timeout=timeout, follow_redirects=False) as client: + with client.stream( + method, url, headers=headers, content=body + ) as response: + if response.status_code not in ( + {200, 202} if method == "POST" else {200} + ): + raise VideoJobError( + "http_error", http_status=response.status_code + ) + data = bytearray() + for chunk in response.iter_bytes(): + if time.monotonic() >= deadline: + raise VideoJobError("deadline") + if len(data) + len(chunk) > MAX_JSON: + raise VideoJobError("response_too_large") + data.extend(chunk) + return _decode(bytes(data)), response.status_code + except httpx.HTTPError: + raise VideoJobError("transport_error") from None + + def capabilities(self, *, timeout: float = 30.0) -> List[str]: + """Contract features, not permission, pricing or worker availability.""" + return _features(self._request("GET", self._health, {}, timeout)[0]) + + def submit( + self, + body: Mapping[str, object], + *, + idempotency_key: str, + user_token: Optional[str] = None, + timeout: float = 30.0, + ) -> VideoJob: + submission = _submission(body) + headers = _headers(self._key, user_token, idempotency_key) + features = self.capabilities(timeout=timeout) + if FEATURE not in features or ( + submission.source and SOURCE_FEATURE not in features + ): + raise VideoJobError("contract_unavailable") + try: + value, code = self._request( + "POST", + self._base + "/videos/jobs", + headers, + timeout, + submission.encoded, + ) + return _accepted(value, code, submission) + except VideoJobError as exc: + raise VideoJobError( + exc.kind, http_status=exc.http_status, submission_uncertain=True + ) from None + + def recover( + self, + *, + job_id: Optional[str] = None, + idempotency_key: Optional[str] = None, + user_token: Optional[str] = None, + timeout: float = 30.0, + ) -> VideoJob: + path = _selector(job_id, idempotency_key) + status = _status( + self._request( + "GET", + self._base + path, + _headers(self._key, user_token, idempotency_key), + timeout, + )[0] + ) + if job_id is not None and status["job_id"] != job_id: + raise VideoJobError("changed_identity") + return status + + def wait( + self, + receipt: VideoJob, + *, + user_token: Optional[str] = None, + timeout: float = 600.0, + poll_interval: float = 5.0, + request_timeout: float = 30.0, + ) -> VideoJob: + deadline = time.monotonic() + _positive(timeout) + _positive(poll_interval) + _positive(request_timeout) + anchor = dict(_status(receipt)) + status = _status(receipt) + while status["state"] not in TERMINAL: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise VideoJobError("deadline") + status = _bound( + self.recover( + job_id=status["job_id"], + user_token=user_token, + timeout=min(remaining, request_timeout), + ), + anchor, + ) + if status["state"] not in TERMINAL: + time.sleep(min(poll_interval, max(0.0, deadline - time.monotonic()))) + return status + + +class AsyncVideoJobs(_VideoBase): + """Async twin of VideoJobs; cancellation never triggers a replacement POST.""" + + async def _request( + self, + method: str, + url: str, + headers: Dict[str, str], + timeout: float, + body: Optional[bytes] = None, + ) -> tuple[object, int]: + deadline = time.monotonic() + _positive(timeout) + try: + async with httpx.AsyncClient( + timeout=timeout, follow_redirects=False + ) as client: + async with client.stream( + method, url, headers=headers, content=body + ) as response: + if response.status_code not in ( + {200, 202} if method == "POST" else {200} + ): + raise VideoJobError( + "http_error", http_status=response.status_code + ) + data = bytearray() + async for chunk in response.aiter_bytes(): + if time.monotonic() >= deadline: + raise VideoJobError("deadline") + if len(data) + len(chunk) > MAX_JSON: + raise VideoJobError("response_too_large") + data.extend(chunk) + return _decode(bytes(data)), response.status_code + except httpx.HTTPError: + raise VideoJobError("transport_error") from None + + async def capabilities(self, *, timeout: float = 30.0) -> List[str]: + return _features((await self._request("GET", self._health, {}, timeout))[0]) + + async def submit( + self, + body: Mapping[str, object], + *, + idempotency_key: str, + user_token: Optional[str] = None, + timeout: float = 30.0, + ) -> VideoJob: + submission = _submission(body) + headers = _headers(self._key, user_token, idempotency_key) + features = await self.capabilities(timeout=timeout) + if FEATURE not in features or ( + submission.source and SOURCE_FEATURE not in features + ): + raise VideoJobError("contract_unavailable") + try: + value, code = await self._request( + "POST", + self._base + "/videos/jobs", + headers, + timeout, + submission.encoded, + ) + return _accepted(value, code, submission) + except VideoJobError as exc: + raise VideoJobError( + exc.kind, http_status=exc.http_status, submission_uncertain=True + ) from None + + async def recover( + self, + *, + job_id: Optional[str] = None, + idempotency_key: Optional[str] = None, + user_token: Optional[str] = None, + timeout: float = 30.0, + ) -> VideoJob: + path = _selector(job_id, idempotency_key) + status = _status( + ( + await self._request( + "GET", + self._base + path, + _headers(self._key, user_token, idempotency_key), + timeout, + ) + )[0] + ) + if job_id is not None and status["job_id"] != job_id: + raise VideoJobError("changed_identity") + return status + + async def wait( + self, + receipt: VideoJob, + *, + user_token: Optional[str] = None, + timeout: float = 600.0, + poll_interval: float = 5.0, + request_timeout: float = 30.0, + ) -> VideoJob: + deadline = time.monotonic() + _positive(timeout) + _positive(poll_interval) + _positive(request_timeout) + anchor = dict(_status(receipt)) + status = _status(receipt) + while status["state"] not in TERMINAL: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise VideoJobError("deadline") + status = _bound( + await self.recover( + job_id=status["job_id"], + user_token=user_token, + timeout=min(remaining, request_timeout), + ), + anchor, + ) + if status["state"] not in TERMINAL: + await asyncio.sleep( + min(poll_interval, max(0.0, deadline - time.monotonic())) + ) + return status diff --git a/src/grid_sdk/video_payments.py b/src/grid_sdk/video_payments.py new file mode 100644 index 0000000..3f66a9c --- /dev/null +++ b/src/grid_sdk/video_payments.py @@ -0,0 +1,277 @@ +# SPDX-License-Identifier: MIT +"""Opt-in unsigned video payment preparation; no HTTP, key handling or sending.""" + +from __future__ import annotations + +import hashlib +import json +import re +import time +from dataclasses import dataclass, field +from typing import Awaitable, Callable, Dict, Mapping, cast +from uuid import UUID + +_MAX_MONEY = 2**53 - 1 +_DOMAINS = { + "eip155:8453": (8453, "USD Coin", "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913"), + "eip155:84532": (84532, "USDC", "0x036cbd53842c5426634e7929541ec2318f3dcf7e"), +} +_TYPES = { + "EIP712Domain": [ + {"name": name, "type": kind} + for name, kind in ( + ("name", "string"), + ("version", "string"), + ("chainId", "uint256"), + ("verifyingContract", "address"), + ) + ], + "TransferWithAuthorization": [ + {"name": name, "type": kind} + for name, kind in ( + ("from", "address"), + ("to", "address"), + ("value", "uint256"), + ("validAfter", "uint256"), + ("validBefore", "uint256"), + ("nonce", "bytes32"), + ) + ], +} + + +class VideoPaymentError(RuntimeError): + """Sanitized error, without quote values, signer messages or signatures.""" + + def __init__(self, kind: str): + self.kind = kind + super().__init__(f"Grid video payment {kind}") + + +def _address(value: object) -> str: + if ( + not isinstance(value, str) + or not re.fullmatch(r"0x[0-9a-fA-F]{40}", value) + or int(value[2:], 16) == 0 + ): + raise ValueError + return value.lower() + + +def _digest(value: object) -> str: + if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value): + raise ValueError + return value + + +def _uuid(value: object) -> str: + if ( + not isinstance(value, str) + or str(UUID(value)) != value + or UUID(value).version != 4 + ): + raise ValueError + return value + + +def _integer(value: object, *, positive: bool = True) -> int: + if type(value) is not int or not (1 if positive else 0) <= value <= _MAX_MONEY: + raise ValueError + return value + + +@dataclass(frozen=True) +class VideoAuthorizationRequest: + quote_id: str + job_id: str + request_hash: str + expires: int + payment_enabled: bool + _encoded: str = field(repr=False) + accepts_upfront_payment: bool = False + + @property + def authorization(self) -> Dict[str, str]: + return cast(Dict[str, str], json.loads(self._encoded)["authorization"]) + + @property + def typed_data(self) -> Dict[str, object]: + return cast(Dict[str, object], json.loads(self._encoded)["typed_data"]) + + def _signable(self) -> None: + if not self.payment_enabled: + raise VideoPaymentError("payment_disabled") + if self.accepts_upfront_payment is not True: + raise VideoPaymentError("upfront_payment_not_accepted") + if int(time.time()) >= self.expires: + raise VideoPaymentError("expired_quote") + + +@dataclass(frozen=True) +class SignedVideoAuthorization: + """A temporary spending capability; explicitly export, never log it.""" + + quote_id: str + job_id: str + _encoded: str = field(repr=False) + + @property + def payload(self) -> Dict[str, object]: + return cast(Dict[str, object], json.loads(self._encoded)) + + +def prepare_video_authorization( + quote: Mapping[str, object], + *, + payer: str, + expected_model: str, + expected_request_hash: str, + expected_recipient: str, + network: str, + max_cost_micro_usd: int, + accept_upfront_payment: bool = False, +) -> VideoAuthorizationRequest: + """Check a trusted Core quote against independently retained intent/policy. + + Preview quotes can be inspected, but signing requires payment_enabled=true. + This does not independently verify Core's hidden plan or an on-chain token. + """ + try: + encoded = json.dumps( + dict(quote), allow_nan=False, ensure_ascii=True, separators=(",", ":") + ) + if len(encoded) > 64 * 1024: + raise ValueError + value = json.loads(encoded) + _integer(max_cost_micro_usd) + if type(accept_upfront_payment) is not bool: + raise ValueError + intent = _digest(expected_request_hash) + payer, recipient = _address(payer), _address(expected_recipient) + chain, name, asset = _DOMAINS[network] + quote_id, job_id = _uuid(value["quote_id"]), _uuid(value["job_id"]) + created, expires = _integer(value["created"]), _integer(value["expires"]) + if ( + value["schema"] != "aipg.video-quote.v1" + or value["binding_schema"] != "aipg.video-quote-binding.v3" + or value["state"] != "quoted" + or type(value["payment_enabled"]) is not bool + or not isinstance(expected_model, str) + or not expected_model + or value["model"] != expected_model + or value["request_hash"] != intent + or value["max_cost_micro_usd"] != max_cost_micro_usd + or type(value["max_cost_micro_usd"]) is not int + or payer == recipient + or expires - created != 300 + or not created <= int(time.time()) < expires + ): + raise ValueError + _digest(value["plan_hash"]) + commitment = _digest(value["commitment"]) + sealed = json.dumps( + {"quote": commitment, "schema": "aipg.video-authorization.v1"}, + sort_keys=True, + separators=(",", ":"), + ) + nonce = "0x" + hashlib.sha256(sealed.encode("ascii")).hexdigest() + if value["authorization_nonce"] != nonce: + raise ValueError + payment = value["payment"] + amount = _integer(payment["amount_micro"]) + domain = { + "name": name, + "version": "2", + "chainId": chain, + "verifyingContract": asset, + } + if ( + payment["scheme"] != "exact" + or payment["payment_flow"] != "upfront" + or payment["asset_transfer_method"] != "eip3009" + or payment["network"] != network + or payment["asset"] != asset + or _address(payment["pay_to"]) != recipient + or payment["eip712_domain"] != domain + or type(payment["valid_after"]) is not int + or payment["valid_after"] != 0 + or type(payment["valid_before"]) is not int + or payment["valid_before"] != expires + or amount > max_cost_micro_usd + ): + raise ValueError + authorization = { + "from": payer, + "to": recipient, + "value": str(amount), + "validAfter": "0", + "validBefore": str(expires), + "nonce": nonce, + } + typed = { + "domain": domain, + "types": _TYPES, + "primaryType": "TransferWithAuthorization", + "message": dict(authorization), + } + frozen = json.dumps( + {"authorization": authorization, "typed_data": typed}, separators=(",", ":") + ) + return VideoAuthorizationRequest( + quote_id, + job_id, + intent, + expires, + value["payment_enabled"], + frozen, + accept_upfront_payment, + ) + except ( + ValueError, + KeyError, + TypeError, + AttributeError, + OverflowError, + RecursionError, + ): + raise VideoPaymentError("invalid_quote_or_policy") from None + + +def _signed( + request: VideoAuthorizationRequest, signature: str +) -> SignedVideoAuthorization: + request._signable() + if ( + not isinstance(signature, str) + or not re.fullmatch(r"0x[0-9a-fA-F]{130}", signature) + or int(signature[-2:], 16) not in {27, 28} + ): + raise VideoPaymentError("unsupported_signature") + payload = {"authorization": request.authorization, "signature": signature} + return SignedVideoAuthorization( + request.quote_id, request.job_id, json.dumps(payload, separators=(",", ":")) + ) + + +def sign_video_authorization( + request: VideoAuthorizationRequest, + signer: Callable[[Dict[str, object]], str], +) -> SignedVideoAuthorization: + request._signable() + try: + signature = signer(request.typed_data) + except Exception: + raise VideoPaymentError("signer_failed") from None + return _signed(request, signature) + + +async def async_sign_video_authorization( + request: VideoAuthorizationRequest, + signer: Callable[[Dict[str, object]], Awaitable[str]], +) -> SignedVideoAuthorization: + request._signable() + try: + signature = await signer(request.typed_data) + except Exception: + raise VideoPaymentError("signer_failed") from None + return _signed(request, signature) diff --git a/tests/AGENTS.md b/tests/AGENTS.md index b03a588..d9ad571 100644 --- a/tests/AGENTS.md +++ b/tests/AGENTS.md @@ -9,17 +9,34 @@ Grid media helper. - `test_client.py` - Grid/OpenAI client defaults, aliases, and public surface. - `test_grid.py` - `/v1` image/video request and response behavior. +- `test_video_jobs.py` - candidate async-job wire contract, source passthrough, + delegation, no POST retry, GET recovery, bounded polling and receipt identity. + Separate appearance tests preserve PNG/audio/optional video and strengths + through lost-reply recovery, rejecting opening-pin conflicts before HTTP. + Cover omitted/explicit reference attention in both audio and motion modes; + the retained request must be unchanged after an uncertain accepted reply. + Inline fixture strings are transport-only, not decoded media/native proof. +- `test_video_payments.py` - quote-policy, immutable views, disabled-before- + signer, expiry/failure/cancellation and hidden-signature controls. Mock + callbacks do not prove ECDSA, wallet UI, funds or HTTP payment. + Require binding-v3 upfront/EIP-3009 and explicit consent before sync/async + callbacks; default/false, coerced policy and legacy quotes reject signing. ## Local Contracts - Mock all HTTP. Never use production API keys or consume live credits. - Assert exact URL, headers, payload, sync behavior, async parity, and errors. - Keep public aliases/version/export behavior covered. +- Preserve legacy custom-base construction; only accessing the candidate + helper requires its `/v1` base. Never replace source/receipt validation with + live production requests in this suite. ## Work Guidance - Add a focused regression test with every public or wire-contract change. - Prefer respx assertions over implementation-specific internal mocks. +- Tests must collect on the SDK's Python 3.9 minimum; postpone annotations + rather than evaluating newer union-type operators during import. ## Verification diff --git a/tests/test_video_jobs.py b/tests/test_video_jobs.py new file mode 100644 index 0000000..756883b --- /dev/null +++ b/tests/test_video_jobs.py @@ -0,0 +1,407 @@ +# SPDX-License-Identifier: MIT +"""Wire contracts for the separate candidate /v1/videos/jobs client.""" + +from __future__ import annotations + +import asyncio +import copy +import json +from typing import Any + +import httpx +import pytest +import respx + +from grid_sdk import ( + AsyncGrid, + AsyncVideoJobs, + Grid, + VideoJobError, + VideoJobs, + video_jobs, +) +from grid_sdk.video_jobs import FEATURE, MAX_JSON, SOURCE_FEATURE + +BASE = "https://grid.example/v1" +KEY = "00000000-0000-4000-8000-0000000000aa" # Public request UUID, not a credential. +JOB = "62964058-56d2-4b84-9ba3-cd8a26ebf831" +MODEL = "LTX-2.5 Distilled BF16" +BODY = { + "model": MODEL, + "prompt": "a person waving in a car", + "frames": 33, + "fps": 24, + "max_cost_micro_usd": 50000, +} + + +def receipt(state: str = "queued", **changes: Any) -> dict: + value = { + "schema": "aipg.video-job.v1", + "job_id": JOB, + "model": MODEL, + "state": state, + "request_hash": "a" * 64, + "reserved_micro": 12000, + "max_cost_micro": 50000, + "actual_micro": None, + "result": None, + } + if state == "completed": + value.update( + actual_micro=12000, + result={ + "model": MODEL, + "worker": "fixture-worker", + "gen_time": 4.5, + "recipe_root": None, + "media": [ + { + "url": "https://media.example/test.mp4", + "key": "test.mp4", + "sha256": "b" * 64, + "seed": None, + } + ], + }, + ) + value.update(changes) + return value + + +def health(features: list = None) -> respx.Route: + return respx.get("https://grid.example/health").respond( + json={"features": [FEATURE, SOURCE_FEATURE] if features is None else features} + ) + + +def test_exports_and_lazy_wiring_preserve_legacy_custom_base() -> None: + assert isinstance(Grid(api_key="fixture", base_url=BASE).grid.video_jobs, VideoJobs) + assert isinstance( + AsyncGrid(api_key="fixture", base_url=BASE).grid.video_jobs, AsyncVideoJobs + ) + # Access to the new contract, not construction of the legacy client, validates /v1. + legacy = Grid(api_key="fixture", base_url="https://grid.example/custom").grid + with pytest.raises(ValueError): + _ = legacy.video_jobs + + +@respx.mock +@pytest.mark.parametrize("mode", ["generated", "audio", "video"]) +def test_submit_source_modes_and_service_delegation(mode: str) -> None: + public = health() + body = dict(BODY) + if mode != "generated": + body["reference_audio"] = "data:audio/wav;base64,fixture" + if mode == "video": + body.update(reference_video="data:video/mp4;base64,fixture", size="768x512") + submit = respx.post(BASE + "/videos/jobs").respond(202, json=receipt()) + assert ( + VideoJobs("fixture", BASE).submit( + body, idempotency_key=KEY, user_token="delegated-fixture" + )["job_id"] + == JOB + ) + assert submit.call_count == 1 + request = submit.calls.last.request + assert json.loads(request.content) == body + assert request.headers["Idempotency-Key"] == KEY + assert request.headers["X-Grid-User-Token"] == "delegated-fixture" + assert request.headers["Authorization"] == "Bearer fixture" + assert "authorization" not in public.calls.last.request.headers + + +@respx.mock +@pytest.mark.parametrize("features", [[], [SOURCE_FEATURE], [FEATURE]]) +def test_missing_feature_never_posts(features: list) -> None: + health(features) + body = {**BODY, "reference_audio": "fixture"} + with pytest.raises(VideoJobError, match="contract_unavailable") as exc: + VideoJobs("fixture", BASE).submit(body, idempotency_key=KEY) + assert not exc.value.submission_uncertain + assert all(call.request.method == "GET" for call in respx.calls) + + +@pytest.mark.parametrize( + "changes", + [ + {"max_cost_micro_usd": 0}, + {"max_cost_micro_usd": True}, + {"max_cost_micro_usd": 1.2}, + {"max_cost_micro_usd": 2**53}, + {"cfg_scale": float("nan")}, + {"model": ""}, + {"prompt": ""}, + {"n": True}, + {"n": 2}, + {"image": "x", "keyframes": [{"image": "y"}]}, + {"reference_video": "x"}, + {"reference_image": "x"}, + { + "reference_image": "x", + "reference_audio": "y", + "size": "768x448", + "image": "z", + }, + { + "reference_image": "x", + "reference_audio": "y", + "size": "768x448", + "keyframes": [{"image": "z"}], + }, + ], +) +def test_invalid_bodies_fail_before_network(changes: dict) -> None: + with respx.mock(assert_all_called=False) as mock: + with pytest.raises(ValueError): + VideoJobs("fixture", BASE).submit({**BODY, **changes}, idempotency_key=KEY) + assert not mock.calls + + +@respx.mock +def test_lost_reply_recovers_by_header_without_post_retry_or_health() -> None: + health() + submit = respx.post(BASE + "/videos/jobs").mock( + side_effect=httpx.ReadTimeout("fixture secret should not leak") + ) + jobs = VideoJobs("fixture", BASE) + with pytest.raises(VideoJobError) as exc: + jobs.submit(BODY, idempotency_key=KEY) + assert exc.value.submission_uncertain + assert "secret" not in str(exc.value) + recover = respx.get(BASE + "/videos/jobs").respond( + json=receipt("execution_uncertain") + ) + assert ( + jobs.recover(idempotency_key=KEY, user_token="fixture-token")["state"] + == "execution_uncertain" + ) + assert recover.calls.last.request.headers["Idempotency-Key"] == KEY + assert recover.calls.last.request.headers["X-Grid-User-Token"] == "fixture-token" + assert str(recover.calls.last.request.url) == BASE + "/videos/jobs" + assert submit.call_count == 1 + + +@respx.mock +@pytest.mark.asyncio +@pytest.mark.parametrize("motion", [False, True]) +@pytest.mark.parametrize("attention", [None, 0.25]) +async def test_appearance_sources_survive_lost_reply_without_becoming_opening_pins( + motion: bool, attention: float | None, +) -> None: + health() + model = "LTX-2.5 Dev BF16 Ingredients" + body = { + **BODY, + "model": model, + "size": "768x448", + "frames": 121, + "reference_image": "data:image/png;base64,fixture", + "reference_audio": "data:audio/wav;base64,fixture", + "reference_strength": 1.0, + } + if attention is not None: + body["reference_attention_strength"] = attention + if motion: + body.update( + reference_video="data:video/mp4;base64,fixture", + guide_strength=0.5, + motion_attention_strength=0.25, + ) + post = respx.post(BASE + "/videos/jobs").mock( + side_effect=httpx.ReadTimeout("private fixture") + ) + jobs = AsyncVideoJobs("fixture", BASE) + with pytest.raises(VideoJobError) as error: + await jobs.submit(body, idempotency_key=KEY) + assert error.value.submission_uncertain + wire = json.loads(post.calls.last.request.content) + assert wire == body and "image" not in wire and "keyframes" not in wire + respx.get(BASE + "/videos/jobs").respond(json=receipt(model=model)) + assert (await jobs.recover(idempotency_key=KEY))["model"] == model + assert post.call_count == 1 + + +@respx.mock +@pytest.mark.parametrize("code", [302, 402, 404, 409, 429, 500, 503]) +def test_submit_error_is_sanitized_uncertain_and_never_retried(code: int) -> None: + health() + submit = respx.post(BASE + "/videos/jobs").respond( + code, + headers={"Location": "https://foreign.example"}, + text="private prompt fixture", + ) + with pytest.raises(VideoJobError) as exc: + VideoJobs("fixture", BASE).submit(BODY, idempotency_key=KEY) + assert exc.value.http_status == code and exc.value.submission_uncertain + assert "private" not in str(exc.value) + assert submit.call_count == 1 + + +@respx.mock +def test_wait_gets_only_same_identity_and_completed_result() -> None: + recover = respx.get(BASE + "/videos/jobs/" + JOB).mock( + side_effect=[ + httpx.Response(200, json=receipt("execution_uncertain")), + httpx.Response(200, json=receipt("completed")), + ] + ) + assert ( + VideoJobs("fixture", BASE).wait(receipt(), poll_interval=0.001)["result"][ + "media" + ][0]["sha256"] + == "b" * 64 + ) + assert recover.call_count == 2 + assert all(call.request.method == "GET" for call in respx.calls) + + +@respx.mock +@pytest.mark.parametrize( + "field,value", + [ + ("job_id", KEY), + ("model", "other"), + ("request_hash", "c" * 64), + ("reserved_micro", 11000), + ("max_cost_micro", 50001), + ], +) +def test_wait_refuses_changed_identity(field: str, value: Any) -> None: + respx.get(BASE + "/videos/jobs/" + JOB).respond(json=receipt(**{field: value})) + with pytest.raises(VideoJobError, match="changed_identity"): + VideoJobs("fixture", BASE).wait(receipt()) + + +@respx.mock +@pytest.mark.parametrize( + "changes", + [ + {"schema": "wrong"}, + {"state": "paid"}, + {"actual_micro": 1}, + {"reserved_micro": True}, + {"reserved_micro": 50001}, + {"max_cost_micro": 2**53}, + {"request_hash": "bad"}, + {"result": {}}, + {"job_id": "../other"}, + ], +) +def test_invalid_status_fails_closed(changes: dict) -> None: + respx.get(BASE + "/videos/jobs/" + JOB).respond(json=receipt(**changes)) + with pytest.raises(VideoJobError, match="invalid_response"): + VideoJobs("fixture", BASE).recover(job_id=JOB) + + +@respx.mock +@pytest.mark.parametrize( + "payload", [b"not JSON", b"{" + b" " * MAX_JSON, b'{"state":NaN}'] +) +def test_invalid_or_oversized_json_never_succeeds(payload: bytes) -> None: + respx.get(BASE + "/videos/jobs/" + JOB).respond(content=payload) + with pytest.raises(VideoJobError): + VideoJobs("fixture", BASE).recover(job_id=JOB) + + +@respx.mock +@pytest.mark.parametrize("state,code", [("queued", 200), ("completed", 202)]) +def test_submit_http_state_and_cost_binding(state: str, code: int) -> None: + health() + respx.post(BASE + "/videos/jobs").respond(code, json=receipt(state)) + with pytest.raises(VideoJobError, match="invalid_response") as exc: + VideoJobs("fixture", BASE).submit(BODY, idempotency_key=KEY) + assert exc.value.submission_uncertain + + +@pytest.mark.parametrize( + "options", + [ + {}, + {"job_id": JOB, "idempotency_key": KEY}, + {"job_id": "../x"}, + {"idempotency_key": JOB.upper()}, + {"idempotency_key": "00000000-0000-1000-8000-000000000000"}, + ], +) +def test_selector_validation(options: dict) -> None: + with pytest.raises(ValueError): + VideoJobs("fixture", BASE).recover(**options) + + +@pytest.mark.parametrize( + "token", ["", "private\r\nInjected: x", "private\0", "private\u2603"] +) +def test_invalid_credentials_are_rejected_without_echo_or_network(token: str) -> None: + with respx.mock(assert_all_called=False) as mock: + with pytest.raises(ValueError) as error: + VideoJobs("fixture", BASE).recover(job_id=JOB, user_token=token) + assert "private" not in str(error.value) + assert not mock.calls + + +@pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf"), True, 10**300]) +def test_wait_timeout_validation(value: float) -> None: + with pytest.raises(ValueError): + VideoJobs("fixture", BASE).wait(receipt(), timeout=value) + + +@respx.mock +def test_deadline_does_not_create_new_job(monkeypatch: pytest.MonkeyPatch) -> None: + clock = [0.0] + monkeypatch.setattr(video_jobs.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr( + video_jobs.time, "sleep", lambda delay: clock.__setitem__(0, clock[0] + delay) + ) + recover = respx.get(BASE + "/videos/jobs/" + JOB).respond(json=receipt()) + with pytest.raises(VideoJobError, match="deadline"): + VideoJobs("fixture", BASE).wait(receipt(), timeout=0.025, poll_interval=0.01) + assert recover.call_count == 3 + assert all(call.request.method == "GET" for call in respx.calls) + + +def test_closed_without_result_is_not_reported_as_refunded() -> None: + result = receipt("closed_without_result", actual_micro=12000) + assert VideoJobs("fixture", BASE).wait(result) == result + assert "refunded" not in result + + +@respx.mock +@pytest.mark.asyncio +async def test_async_source_submission_lost_reply_and_wait() -> None: + health() + submit = respx.post(BASE + "/videos/jobs").mock( + side_effect=httpx.ReadTimeout("private") + ) + jobs = AsyncVideoJobs("fixture", BASE) + with pytest.raises(VideoJobError) as exc: + await jobs.submit({**BODY, "reference_audio": "fixture"}, idempotency_key=KEY) + assert exc.value.submission_uncertain + respx.get(BASE + "/videos/jobs").respond(json=receipt()) + recovered = await jobs.recover(idempotency_key=KEY) + respx.get(BASE + "/videos/jobs/" + JOB).respond(json=receipt("completed")) + assert (await jobs.wait(recovered))["state"] == "completed" + assert submit.call_count == 1 + + +@respx.mock +@pytest.mark.asyncio +async def test_async_wait_is_cancellable_without_resubmission() -> None: + respx.get(BASE + "/videos/jobs/" + JOB).respond(json=receipt()) + task = asyncio.create_task( + AsyncVideoJobs("fixture", BASE).wait(receipt(), poll_interval=30) + ) + await asyncio.sleep(0.05) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert all(call.request.method == "GET" for call in respx.calls) + + +@respx.mock +def test_malformed_completed_media_is_refused() -> None: + value = copy.deepcopy(receipt("completed")) + value["result"]["media"][0]["url"] = "file:///private/fixture" + respx.get(BASE + "/videos/jobs/" + JOB).respond(json=value) + with pytest.raises(VideoJobError, match="invalid_response"): + VideoJobs("fixture", BASE).recover(job_id=JOB) diff --git a/tests/test_video_payments.py b/tests/test_video_payments.py new file mode 100644 index 0000000..f801e84 --- /dev/null +++ b/tests/test_video_payments.py @@ -0,0 +1,295 @@ +# SPDX-License-Identifier: MIT +"""Hermetic wallet-policy controls, not real payments or ECDSA proof.""" + +import asyncio +import hashlib +import json +from unittest.mock import Mock + +import pytest + +from grid_sdk import ( + VideoPaymentError, + async_sign_video_authorization, + prepare_video_authorization, + sign_video_authorization, +) +from grid_sdk import video_payments as payments + +NOW = 2_000_000_000 +SIG = "0x" + "12" * 64 + "1b" + + +@pytest.fixture +def fixture(monkeypatch): + monkeypatch.setattr(payments.time, "time", lambda: NOW) + commitment = "c" * 64 + nonce = ( + "0x" + + hashlib.sha256( + json.dumps( + {"quote": commitment, "schema": "aipg.video-authorization.v1"}, + sort_keys=True, + separators=(",", ":"), + ).encode("ascii") + ).hexdigest() + ) + asset = "0x036cbd53842c5426634e7929541ec2318f3dcf7e" + quote = { + "schema": "aipg.video-quote.v1", + "binding_schema": "aipg.video-quote-binding.v3", + "quote_id": "11111111-1111-4111-8111-111111111111", + "job_id": "22222222-2222-4222-8222-222222222222", + "state": "quoted", + "payment_enabled": False, + "created": NOW, + "expires": NOW + 300, + "model": "LTX-2.5 Distilled BF16", + "request_hash": "a" * 64, + "plan_hash": "b" * 64, + "commitment": commitment, + "authorization_nonce": nonce, + "max_cost_micro_usd": 100_000, + "payment": { + "scheme": "exact", + "payment_flow": "upfront", + "asset_transfer_method": "eip3009", + "network": "eip155:84532", + "asset": asset, + "pay_to": "0x" + "1" * 40, + "amount_micro": 7500, + "valid_after": 0, + "valid_before": NOW + 300, + "eip712_domain": { + "name": "USDC", + "version": "2", + "chainId": 84532, + "verifyingContract": asset, + }, + }, + } + policy = { + "payer": "0x" + "2" * 40, + "expected_model": quote["model"], + "expected_request_hash": "a" * 64, + "expected_recipient": "0x" + "1" * 40, + "network": "eip155:84532", + "max_cost_micro_usd": 100_000, + "accept_upfront_payment": True, + } + return quote, policy + + +def test_preview_never_calls_wallet_and_request_views_are_immutable(fixture): + quote, policy = fixture + request = prepare_video_authorization(quote, **policy) + assert request.authorization["nonce"] == quote["authorization_nonce"] + assert request.typed_data["message"]["value"] == "7500" + assert "EIP712Domain" in request.typed_data["types"] + request.typed_data["domain"]["name"] = "changed" + quote["payment"]["amount_micro"] = 1 + assert ( + request.authorization["value"] == "7500" + and request.typed_data["domain"]["name"] == "USDC" + ) + signer = Mock(return_value=SIG) + with pytest.raises(VideoPaymentError, match="payment_disabled"): + sign_video_authorization(request, signer) + signer.assert_not_called() + + +@pytest.mark.parametrize( + "field,value", + [ + ("expected_model", "wrong"), + ("expected_request_hash", "b" * 64), + ("expected_recipient", "0x" + "3" * 40), + ("network", "eip155:8453"), + ("max_cost_micro_usd", 99_999), + ("max_cost_micro_usd", True), + ("payer", "0x" + "1" * 40), + ], +) +def test_independent_policy_mismatch_refuses_before_wallet(fixture, field, value): + quote, policy = fixture + with pytest.raises(VideoPaymentError): + prepare_video_authorization(quote, **{**policy, field: value}) + + +@pytest.mark.parametrize( + "field,value", + [ + ("binding_schema", "aipg.video-quote-binding.v1"), + ("payment_enabled", "false"), + ("authorization_nonce", "0x" + "0" * 64), + ("expires", NOW), + ("created", NOW + 1), + ("max_cost_micro_usd", True), + ("state", "funded"), + ], +) +def test_wrong_quote_identity_window_and_types_are_private_errors( + fixture, field, value +): + quote, policy = fixture + with pytest.raises(VideoPaymentError) as error: + prepare_video_authorization({**quote, field: value}, **policy) + assert value not in str(error.value) if isinstance(value, str) else True + + +@pytest.mark.parametrize( + "field,value", + [ + ("amount_micro", 100_001), + ("amount_micro", True), + ("asset", "0x" + "3" * 40), + ("valid_after", 1), + ("valid_before", NOW + 301), + ("scheme", "upto"), + ("network", "eip155:1"), + ], +) +def test_changed_payment_fields_are_rejected(fixture, field, value): + quote, policy = fixture + quote["payment"][field] = value + with pytest.raises(VideoPaymentError): + prepare_video_authorization(quote, **policy) + + +def test_domain_nonfinite_and_unbounded_data_are_rejected(fixture): + quote, policy = fixture + quote["payment"]["eip712_domain"]["chainId"] = 1 + with pytest.raises(VideoPaymentError): + prepare_video_authorization(quote, **policy) + quote["payment"]["eip712_domain"]["chainId"] = 84532 + for item in (float("nan"), "private-marker" * 6000): + with pytest.raises(VideoPaymentError) as error: + prepare_video_authorization({**quote, "unused": item}, **policy) + assert "private-marker" not in str(error.value) + + +def test_sign_callback_is_explicit_and_spending_capability_is_not_in_repr(fixture): + quote, policy = fixture + quote["payment_enabled"] = True # Synthetic future server; actual Core stays false. + request = prepare_video_authorization(quote, **policy) + signer = Mock(return_value=SIG) + signed = sign_video_authorization(request, signer) + signer.assert_called_once_with(request.typed_data) + assert signed.payload == {"authorization": request.authorization, "signature": SIG} + assert SIG not in repr(signed) + signed.payload["signature"] = "changed" + assert signed.payload["signature"] == SIG + + +@pytest.mark.parametrize("accept", [False, None]) +def test_enabled_quote_still_requires_explicit_upfront_acceptance(fixture, accept): + quote, policy = fixture + quote["payment_enabled"] = True + if accept is None: + policy.pop("accept_upfront_payment") + else: + policy["accept_upfront_payment"] = accept + request = prepare_video_authorization(quote, **policy) + signer = Mock(return_value=SIG) + with pytest.raises(VideoPaymentError, match="upfront_payment_not_accepted"): + sign_video_authorization(request, signer) + signer.assert_not_called() + + +@pytest.mark.parametrize( + "term,value", + [ + ("payment_flow", "authorization"), + ("payment_flow", "escrow"), + ("payment_flow", None), + ("asset_transfer_method", "permit2"), + ("asset_transfer_method", None), + ], +) +def test_wrong_or_implicit_upfront_terms_are_rejected(fixture, term, value): + quote, policy = fixture + if value is None: + quote["payment"].pop(term) + else: + quote["payment"][term] = value + with pytest.raises(VideoPaymentError): + prepare_video_authorization(quote, **policy) + + +@pytest.mark.parametrize("accept", ["true", 1, None]) +def test_upfront_policy_cannot_be_coerced(fixture, accept): + quote, policy = fixture + policy["accept_upfront_payment"] = accept + with pytest.raises(VideoPaymentError): + prepare_video_authorization(quote, **policy) + + +def test_legacy_v2_quote_cannot_gain_upfront_signing_terms(fixture): + quote, policy = fixture + quote["binding_schema"] = "aipg.video-quote-binding.v2" + with pytest.raises(VideoPaymentError): + prepare_video_authorization(quote, **policy) + + +@pytest.mark.asyncio +async def test_async_callback_requires_upfront_consent_before_invocation(fixture): + quote, policy = fixture + quote["payment_enabled"] = True + policy.pop("accept_upfront_payment") + called = [] + + async def signer(data): + called.append(data) + return SIG + + with pytest.raises(VideoPaymentError, match="upfront_payment_not_accepted"): + await async_sign_video_authorization( + prepare_video_authorization(quote, **policy), signer + ) + assert not called + + +def test_expiry_after_wallet_and_signer_failure_are_sanitized(fixture, monkeypatch): + quote, policy = fixture + quote["payment_enabled"] = True + request = prepare_video_authorization(quote, **policy) + + def slow(_data): + monkeypatch.setattr(payments.time, "time", lambda: NOW + 300) + return SIG + + with pytest.raises(VideoPaymentError, match="expired_quote"): + sign_video_authorization(request, slow) + monkeypatch.setattr(payments.time, "time", lambda: NOW) + with pytest.raises(VideoPaymentError) as error: + sign_video_authorization( + request, Mock(side_effect=RuntimeError("private-signer-marker")) + ) + assert "private-signer-marker" not in str(error.value) + with pytest.raises(VideoPaymentError, match="unsupported_signature"): + sign_video_authorization(request, Mock(return_value="private-signer-marker")) + + +@pytest.mark.asyncio +async def test_async_wallet_parity_preview_and_cancellation(fixture): + quote, policy = fixture + + async def signer(data): + assert data["message"]["nonce"] == quote["authorization_nonce"] + return SIG + + with pytest.raises(VideoPaymentError, match="payment_disabled"): + await async_sign_video_authorization( + prepare_video_authorization(quote, **policy), signer + ) + quote["payment_enabled"] = True + request = prepare_video_authorization(quote, **policy) + assert ( + await async_sign_video_authorization(request, signer) + ).payload == sign_video_authorization(request, lambda _data: SIG).payload + + async def cancel(_data): + raise asyncio.CancelledError + + with pytest.raises(asyncio.CancelledError): + await async_sign_video_authorization(request, cancel)