Spaces:

xjy999
/

ele-test

Paused

Add privacy-safe preflight diagnostics
2d02cf5 verified 7 days ago
41.2 kB
import asyncio
import time
from collections import defaultdict

from .diagnostics import event, request_parameters, span
from .images import validate_reference_durations
from .models import PoolError, UpstreamError, VideoRequest, reserve_credits
from .security import atomic_write


def loggable(response):
    """Upstream payload for the plaintext event log: signed media URLs lose their query
    string (the sealed task result keeps the full links; /result refreshes them)."""
    if isinstance(response, dict):
        return {
            k: v.split("?", 1)[0] if k.endswith("_url") and isinstance(v, str) else loggable(v)
            for k, v in response.items()
        }
    if isinstance(response, list):
        return [loggable(v) for v in response]
    return response


# One neutral 2.0 s clip per output aspect ratio; anything else uses 16:9.
AUTO_REFERENCE_SIZES = {
    "16:9": "1280x720", "9:16": "720x1280", "1:1": "720x720",
    "4:3": "960x720", "3:4": "720x960", "21:9": "1680x720",
}
STALE_ASSET_CODES = {"upstream_500", "upstream_404", "upstream_422"}
# Three attempts total; the last delay only cools the account after task failure.
CAPTCHA_RETRY_DELAYS = (30, 60, 120)


def first_asset(references):
    """Task/quote asset_id: the first user reference, never the shared neutral clip."""
    return next((r["asset_id"] for r in references if not r.get("auto")), None)


class Engine:
    def __init__(self, store, settings, upstream, browsers):
        self.store, self.settings = store, settings
        self.upstream, self.browsers = upstream, browsers
        self.selection_lock = asyncio.Lock()
        self.account_locks = defaultdict(asyncio.Lock)
        self.submission_locks = defaultdict(asyncio.Lock)
        self.running = {}
        self.loops = []
        self.stopping = False

    async def probe(self, account_id, *, clear_attention=False):
        try:
            balance = self.store.bind_workspace(
                account_id,
                await self.upstream.workspace(account_id),
                clear_attention=clear_attention,
            )
            self.store.reconcile_usage(balance["id"])
            return self.store.workspace(balance["id"])
        except PoolError as exc:
            state = "needs_login" if exc.code == "needs_login" else "unavailable"
            self.store.db.execute(
                "UPDATE accounts SET state=?,error=? WHERE id=?", (state, exc.code, account_id)
            )
            raise

    async def asset(self, task_id, account_id, request):
        existing = self.store.one(
            "SELECT asset_id FROM assets WHERE task_id=? AND account_id=?", (task_id, account_id)
        )
        if existing:
            return existing["asset_id"]
        asset_id = await self.upstream.import_image(account_id, request.image_references()[0])
        self.ensure_unique_asset(task_id, asset_id)
        self.store.db.execute("INSERT INTO assets VALUES(?,?,?)", (task_id, account_id, asset_id))
        self.store.event(
            task_id, "asset_imported", {"account_id": account_id, "asset_id": asset_id}
        )
        return asset_id

    def ensure_unique_asset(self, task_id, asset_id):
        if self.store.one(
            "SELECT task_id FROM assets WHERE asset_id=? AND task_id<>?", (asset_id, task_id)
        ) or self.store.one(
            # The neutral auto reference is shared across tasks by design.
            "SELECT task_id FROM reference_assets WHERE asset_id=? AND task_id<>? AND kind<>'auto_video'",
            (asset_id, task_id),
        ):
            raise PoolError(
                "image_asset_reused", "Upstream reused another task's reconciliation asset", 502
            )

    async def references(self, request, account_id, task_id=None):
        if request.legacy_single_image():
            with span("reference_import", account_id=account_id,
                      reference_kind="image", reference_index=0):
                asset_id = (
                    await self.asset(task_id, account_id, request)
                    if task_id
                    else await self.upstream.import_image(account_id, request.image_references()[0])
                )
            references = [{"kind": "image", "index": 0, "asset_id": asset_id}]
        else:
            references = await self.user_references(request, account_id, task_id)
        with span("automatic_reference", account_id=account_id):
            auto = await self.auto_reference(request, account_id, task_id)
        references += [auto] if auto else []
        validate_reference_durations(references, request.limits()["reference_seconds"])
        return references

    async def user_references(self, request, account_id, task_id):
        references = []
        for kind, index, url in request.reference_urls():
            row = (
                self.store.one(
                    "SELECT metadata FROM reference_assets WHERE task_id=? AND account_id=? AND kind=? AND ordinal=?",
                    (task_id, account_id, kind, index),
                )
                if task_id
                else None
            )
            if row:
                reference = self.store.vault.open(row["metadata"])
            else:
                media_kind = "image" if kind in {"start_frame", "end_frame"} else kind
                with span("reference_import", account_id=account_id,
                          reference_kind=kind, reference_index=index):
                    reference = await self.upstream.import_media(account_id, url, media_kind)
                reference["kind"] = kind
                reference["index"] = index
                if task_id:
                    self.ensure_unique_asset(task_id, reference["asset_id"])
                    self.store.db.execute(
                        "INSERT INTO reference_assets VALUES(?,?,?,?,?,?)",
                        (
                            task_id,
                            account_id,
                            kind,
                            index,
                            reference["asset_id"],
                            self.store.vault.seal(reference),
                        ),
                    )
                    self.store.event(
                        task_id,
                        "reference_imported",
                        {
                            "account_id": account_id,
                            "kind": kind,
                            "index": index,
                            "asset_id": reference["asset_id"],
                        },
                    )
            references.append(reference)
        return references

    def auto_reference_name(self, request):
        limits = request.limits()
        if not (
            self.settings.auto_reference
            and limits["videos"]
            and not request.video_urls
            and not (request.start_frame_url or request.end_frame_url)
            and len(request.reference_urls()) < limits["total"]
        ):
            return None
        size = AUTO_REFERENCE_SIZES.get(request.aspect_ratio, AUTO_REFERENCE_SIZES["16:9"])
        return f"neutral-2.0s-{size}.mp4"

    async def auto_reference(self, request, account_id, task_id=None):
        """Neutral 2 s video for eligible requests: imported once per account and reused.
        Each prepare attempt rewrites the task's own copy, so submit always matches the quote."""
        if task_id:
            self.store.db.execute(
                "DELETE FROM reference_assets WHERE task_id=? AND account_id=? AND kind='auto_video'",
                (task_id, account_id),
            )
        name = self.auto_reference_name(request)
        if not name:
            return None
        row = self.store.one(
            "SELECT metadata FROM auto_reference_assets WHERE account_id=? AND name=?",
            (account_id, name),
        )
        if row:
            reference = self.store.vault.open(row["metadata"])
        else:
            url = self.settings.auto_reference_base_url.rstrip("/") + "/" + name
            reference = await self.upstream.import_media(account_id, url, "video")
            reference.update(kind="video", auto=name)
            self.store.db.execute(
                "INSERT OR REPLACE INTO auto_reference_assets VALUES(?,?,?,?,?)",
                (account_id, name, reference["asset_id"], self.store.vault.seal(reference), time.time()),
            )
        # The user has no videos on this path, so the neutral clip is always @Video1.
        reference["index"] = len(request.video_urls)
        if task_id:
            self.store.db.execute(
                "INSERT INTO reference_assets VALUES(?,?,'auto_video',0,?,?)",
                (task_id, account_id, reference["asset_id"], self.store.vault.seal(reference)),
            )
            self.store.event(task_id, "auto_reference", {
                "account_id": account_id, "file": name, "asset_id": reference["asset_id"],
                "cached": bool(row),
                "aspect_ratio_submitted": request.parameters([reference]).get("aspect_ratio"),
            })
        return reference

    def drop_auto_asset(self, account_id, name, exc):
        """A cached neutral asset can vanish upstream (quote returns 500/404/422)."""
        if not name or not isinstance(exc, UpstreamError) or exc.code not in STALE_ASSET_CODES:
            return False
        self.store.db.execute(
            "DELETE FROM auto_reference_assets WHERE account_id=? AND name=?", (account_id, name)
        )
        return True

    async def quoted(self, request, account_id, task_id=None):
        with span("reference_preparation", account_id=account_id):
            references = await self.references(request, account_id, task_id)
        try:
            with span("quote", account_id=account_id, attempt=1):
                return references, await self.upstream.quote(account_id, request, references)
        except UpstreamError as exc:
            auto = next((r["auto"] for r in references if r.get("auto")), None)
            if not self.drop_auto_asset(account_id, auto, exc):
                raise
        # Quotes are free: re-import the neutral clip once and quote again.
        with span("reference_preparation", account_id=account_id, attempt=2):
            references = await self.references(request, account_id, task_id)
        with span("quote", account_id=account_id, attempt=2):
            return references, await self.upstream.quote(account_id, request, references)

    def task_references(self, task, request):
        if request.legacy_single_image():
            references = [{"kind": "image", "index": 0, "asset_id": task["asset_id"]}]
        else:
            references = self.task_user_references(task, request)
        row = self.store.one(
            "SELECT metadata FROM reference_assets WHERE task_id=? AND account_id=? AND kind='auto_video'",
            (task["id"], task["account_id"]),
        )
        references += [self.store.vault.open(row["metadata"])] if row else []
        validate_reference_durations(references, request.limits()["reference_seconds"])
        return references

    def task_user_references(self, task, request):
        references = []
        for kind, index, _ in request.reference_urls():
            row = self.store.one(
                "SELECT metadata FROM reference_assets WHERE task_id=? AND account_id=? AND kind=? AND ordinal=?",
                (task["id"], task["account_id"], kind, index),
            )
            if not row:
                raise PoolError(
                    "media_manifest_missing", "Persisted reference mapping is incomplete", 409
                )
            references.append(self.store.vault.open(row["metadata"]))
        return references

    async def quote(self, request, account_id=None):
        request_parameters(request)
        candidates = [self.store.account(account_id)] if account_id else self.store.candidates()
        event("candidates", candidate_count=len(candidates))
        if not candidates:
            raise PoolError("no_ready_account", "Add and probe an account first", 409)
        account_id = candidates[0]["id"]
        with span("account_check", account_id=account_id, attempt=1):
            with span("balance", account_id=account_id):
                balance = await self.probe(account_id)
            # Preview imports are intentionally separate from task reconciliation assets.
            references, credits = await self.quoted(request, account_id)
        reservation = reserve_credits(credits)
        return {
            "account_id": account_id,
            "asset_id": first_asset(references),
            "references": references,
            "asset_transport": "url_import",
            "quoted_credits": credits,
            "reserve_credits": reservation,
            "balance": balance,
            "sufficient": balance["available_credits"]
            >= reservation + self.settings.quota_safety_margin,
            "expires": "Advisory only; task admission obtains a fresh quote and balance",
            "mode": self.settings.mode,
        }

    def receipt_path(self, task_id):
        return self.settings.data_dir / "receipts" / (task_id + ".enc")

    async def create_task(self, idem, request):
        request_parameters(request)
        with span("selection_lock_wait"):
            await self.selection_lock.acquire()
        try:
            # Replays (including conflicting payloads) keep the original contract,
            # even if the pool has since run out of credits or ready accounts.
            if self.store.one("SELECT id FROM tasks WHERE idem=?", (idem,)):
                return self.store.create_task(idem, request)
            candidates = self.store.candidates()
            event("candidates", candidate_count=len(candidates))
            if not candidates:
                raise PoolError("no_ready_account", "No ready account is available", 409)
            last_error = None
            for attempt, account in enumerate(candidates, 1):
                try:
                    with span("account_check", account_id=account["id"], attempt=attempt):
                        with span("balance", account_id=account["id"]):
                            balance = await self.probe(account["id"])
                        if balance["available_credits"] <= self.settings.quota_safety_margin:
                            continue
                        # Capacity can recover when a running job ends; spent credits
                        # cannot. Check the price even when all slots are occupied.
                        _, quote = await self.quoted(request, account["id"])
                        current = self.store.account(account["id"])
                        if (
                            not current["enabled"] or current["state"] != "ready"
                            or current["cooldown"] > time.time()
                        ):
                            last_error = PoolError(
                                "no_ready_account", "No ready account is available", 409
                            )
                            continue
                        balance = self.store.workspace(balance["id"])
                        if balance["available_credits"] >= (
                            reserve_credits(quote) + self.settings.quota_safety_margin
                        ):
                            return self.store.create_task(idem, request)
                except PoolError as exc:
                    if exc.code.startswith(
                        ("image_", "invalid_image", "unsafe_image", "media_", "invalid_media")
                    ):
                        raise
                    last_error = exc
            # A failed probe/quote is not evidence of an exhausted balance.
            if last_error:
                raise last_error
            raise PoolError(
                "insufficient_credits", "No ready account has enough credits for this request", 409
            )
        finally:
            self.selection_lock.release()

    async def start(self):
        self.store.recover()
        if self.settings.headless:
            for task in self.store.many(
                "SELECT t.* FROM tasks t JOIN accounts a ON a.id=t.account_id "
                "WHERE t.status='waiting_captcha' AND t.slot=0 AND t.error='needs_attention'"
            ):
                if self.captcha_retry_allowed(task):
                    self.retry_captcha(task, PoolError("needs_attention", "Legacy verification wait"))
        for task in self.store.many("SELECT * FROM tasks WHERE status='submission_unknown'"):
            path = self.receipt_path(task["id"])
            if path.exists():
                receipt = self.store.vault.open(path.read_text())
                if receipt["task_id"] == task["id"] and receipt["account_id"] == task["account_id"]:
                    self.store.change(
                        task["id"],
                        upstream_id=receipt["upstream_id"],
                        status="processing",
                        error=None,
                    )
        self.loops = [asyncio.create_task(self._dispatch()), asyncio.create_task(self._refresh()),
                      asyncio.create_task(self._reap_browsers())]

    async def reap_browsers(self):
        if not self.settings.browser_idle_seconds:
            return []
        protected = {row["id"] for row in self.store.many(
            "SELECT id FROM accounts WHERE state IN ('needs_login','needs_attention')"
        )}
        protected.update(row["account_id"] for row in self.store.many(
            "SELECT DISTINCT account_id FROM tasks WHERE slot=1 "
            "AND status IN ('preparing','waiting_captcha','submitting','submission_unknown')"
        ))
        for locks in (self.account_locks, self.submission_locks):
            protected.update(account_id for account_id, lock in locks.items() if lock.locked())
        return await self.browsers.reap_idle(protected=protected)

    async def _reap_browsers(self):
        while not self.stopping:
            try:
                await self.reap_browsers()
            except Exception:
                # Browser cleanup must not stop submission/status processing.
                pass
            timeout = self.settings.browser_idle_seconds
            await asyncio.sleep(min(5, max(1, timeout / 2)) if timeout else 5)

    async def _refresh(self):
        while not self.stopping:
            for account in self.store.many(
                "SELECT id,state FROM accounts WHERE credentials IS NOT NULL"
            ):
                try:
                    await self.upstream.token(account["id"])
                    # Preserve CAPTCHA attention until explicitly synced by the operator.
                    if account["state"] != "needs_attention":
                        await self.probe(account["id"])
                except PoolError as exc:
                    if exc.code == "needs_login":
                        self.store.db.execute(
                            "UPDATE accounts SET state='needs_login',error=? WHERE id=?",
                            (exc.code, account["id"]),
                        )
                except Exception:
                    self.store.db.execute(
                        "UPDATE accounts SET error='refresh_failed' WHERE id=?", (account["id"],)
                    )
            for task in self.store.many(
                "SELECT id FROM tasks WHERE hold=1 AND status IN ('completed','failed')"
            ):
                try:
                    await self.settle_if_observed(task["id"])
                except Exception:
                    pass
            await asyncio.sleep(60)

    async def _dispatch(self):
        while not self.stopping:
            for task_id, job in list(self.running.items()):
                if job.done():
                    # Consume exceptions without logging upstream payloads or credentials.
                    try:
                        job.result()
                    except Exception:
                        task = self.store.task(task_id)
                        if task["status"] == "submitting":
                            self.store.change(
                                task_id,
                                status="submission_unknown",
                                error="internal_submit_uncertainty",
                            )
                        else:
                            self.store.change(
                                task_id, error="internal_worker_error", next_check=time.time() + 30
                            )
                    del self.running[task_id]
            rows = self.store.many(
                "SELECT id,status FROM tasks WHERE status IN ('queued','processing','submission_unknown') AND next_check<=? ORDER BY created",
                (time.time(),),
            )
            for row in rows:
                if len(self.running) >= self.settings.worker_count:
                    break
                if row["id"] not in self.running:
                    self.running[row["id"]] = asyncio.create_task(self.step(row["id"]))
            await asyncio.sleep(0.1)

    async def step(self, task_id):
        task = self.store.task(task_id)
        if task["status"] == "queued":
            await self.prepare(task_id)
        elif task["status"] == "processing":
            await self.poll(task_id)
        elif task["status"] == "submission_unknown":
            await self.reconcile(task_id)

    async def prepare(self, task_id):
        task = self.store.task(task_id)
        if task["status"] != "queued":
            return
        if time.time() - task["created"] > self.settings.queue_timeout_seconds:
            self.store.finish_task(task_id, status="failed", error="queue_timeout")
            return
        if self.settings.mode == "live" and not self.settings.generation_enabled:
            self.store.change(task_id, error="generation_disabled", next_check=time.time() + 30)
            return
        self.store.change(task_id, status="preparing", error=None)
        request = VideoRequest.model_validate(self.store.vault.open(task["request"]))
        selected = False
        try:
            async with self.selection_lock:
                candidates = self.store.candidates()
                retryable = not candidates
                for account in candidates:
                    try:
                        balance = await self.probe(account["id"])
                        if balance["available_credits"] <= self.settings.quota_safety_margin:
                            continue
                        shared = self.store.one(
                            "SELECT COUNT(*) AS n FROM tasks WHERE workspace_id=? AND slot=1",
                            (balance["id"],),
                        )["n"]
                        if (
                            account["active"] >= account["max_concurrent"]
                            or shared >= self.settings.workspace_concurrency
                            or self.store.account(account["id"])["state"] != "ready"
                        ):
                            retryable = True
                            continue
                        references, quote = await self.quoted(request, account["id"], task_id)
                        if self.store.workspace(balance["id"])["available_credits"] < (
                            reserve_credits(quote) + self.settings.quota_safety_margin
                        ):
                            continue
                        selected = self.store.reserve(
                            task_id,
                            account["id"],
                            first_asset(references),
                            quote,
                            reserve_credits(quote),
                            self.settings.workspace_concurrency,
                            self.settings.quota_safety_margin,
                        )
                        if selected:
                            break
                        retryable = True
                    except PoolError as exc:
                        if exc.code.startswith(
                            ("image_", "invalid_image", "unsafe_image", "media_", "invalid_media")
                        ):
                            raise
                        retryable = True
                        self.store.db.execute(
                            "UPDATE accounts SET error=?,cooldown=? WHERE id=?",
                            (exc.code, time.time() + 30, account["id"]),
                        )
                if not selected:
                    if not retryable:
                        raise PoolError(
                            "insufficient_credits",
                            "No ready account has enough credits for this request",
                            409,
                        )
                    self.store.change(
                        task_id,
                        status="queued",
                        error="no_eligible_capacity_or_balance",
                        next_check=time.time() + self.settings.queue_retry_seconds,
                    )
                    return
            task = self.store.task(task_id)
            account_id = task["account_id"]
            self.store.change(task_id, status="waiting_captcha")
            # Consume each widget token before another task changes the account page.
            # This lock ends after POST, not after video completion.
            async with self.submission_locks[account_id]:
                await self.submit_reserved(task_id, request)
        except asyncio.CancelledError:
            # Shutdown leaves the durable state for recover() to classify.
            raise
        except Exception as exc:
            current = self.store.task(task_id)
            code = exc.code if isinstance(exc, PoolError) else "internal_error"
            auto = current["account_id"] and self.store.one(
                "SELECT metadata FROM reference_assets WHERE task_id=? AND account_id=? AND kind='auto_video'",
                (task_id, current["account_id"]),
            )
            if auto:
                # Never retry a paid POST; only make the next task re-import the clip.
                self.drop_auto_asset(
                    current["account_id"], self.store.vault.open(auto["metadata"])["auto"], exc
                )
            ambiguous = current["status"] == "submitting" and (
                not isinstance(exc, UpstreamError) or exc.ambiguous
            )
            if ambiguous:
                self.store.change(
                    task_id,
                    status="submission_unknown",
                    error=code,
                    next_check=time.time() + self.settings.poll_seconds,
                )
                self.store.event(
                    task_id,
                    "submission_unknown",
                    {"code": code, "response": getattr(exc, "body", None)},
                )
            else:
                if code in {"needs_attention", "captcha_account_cooling_down"} and self.captcha_retry_allowed(current):
                    if code == "captcha_account_cooling_down":
                        cooldown = self.store.account(current["account_id"])["cooldown"]
                        retry_at = time.time() + max(1, min(30, cooldown - time.time()))
                        self.store.change(task_id, status="queued", hold=0, slot=0,
                                          error="captcha_retry_scheduled", next_check=retry_at)
                        self.store.event(task_id, "captcha_account_cooldown", {"retry_at": retry_at})
                    else:
                        self.retry_captcha(current, exc)
                    return
                if code == "needs_attention" and current["account_id"] and current["status"] == "waiting_captcha":
                    self.store.db.execute(
                        "UPDATE accounts SET state='needs_attention',error=? WHERE id=?",
                        (code, current["account_id"]),
                    )
                    self.store.change(task_id, status="waiting_captcha", hold=0, slot=0, error=code)
                    self.store.event(
                        task_id,
                        "waiting_for_operator",
                        {"cause": repr(exc.__cause__ or exc)[:300]},
                    )
                    return
                # A known rejected POST still does not retry automatically.
                self.store.finish_task(task_id, status="failed", error=code)
                self.store.event(
                    task_id,
                    "failed_before_acceptance",
                    {
                        "code": code,
                        "response": getattr(exc, "body", None),
                        # Unexpected local exceptions are otherwise invisible to the operator.
                        "cause": repr(exc)[:300] if code == "internal_error" else None,
                    },
                )

    def captcha_retry_allowed(self, task):
        return (
            self.settings.headless
            and task["status"] == "waiting_captcha"
            and task["account_id"]
            and not self.browsers.is_open(task["account_id"])
            and not task["upstream_id"]
            and not self.store.one(
                "SELECT id FROM events WHERE task_id=? AND kind='submit_intent' LIMIT 1", (task["id"],)
            )
        )

    def retry_captcha(self, task, exc):
        """Only called before any paid submit; persist the budget and retry together."""
        now = time.time()
        with self.store.transaction():
            attempt = 1 + self.store.one(
                "SELECT COUNT(*) AS n FROM events WHERE task_id=? AND kind='captcha_attempt_failed'",
                (task["id"],),
            )["n"]
            delay = CAPTCHA_RETRY_DELAYS[min(attempt, len(CAPTCHA_RETRY_DELAYS)) - 1]
            detail = {"account_id": task["account_id"], "attempt": attempt,
                      "max_attempts": len(CAPTCHA_RETRY_DELAYS),
                      "stage": getattr(exc, "stage", "verification"),
                      "diagnostics": getattr(exc, "diagnostics", {"error_type": type(exc).__name__})}
            self.store.event(task["id"], "captcha_attempt_failed", detail)
            self.store.db.execute(
                "UPDATE accounts SET state='ready',error='captcha_unavailable',cooldown=MAX(cooldown,?) "
                "WHERE id=? AND state IN ('ready','needs_attention')",
                (now + delay, task["account_id"]),
            )
            exhausted = attempt >= len(CAPTCHA_RETRY_DELAYS)
            expired = now - task["created"] > self.settings.queue_timeout_seconds
            if exhausted or expired:
                code = "queue_timeout" if expired else "captcha_retry_exhausted"
                self.store.change(task["id"], status="failed", error=code)
                self.store.release_terminal_hold(self.store.task(task["id"]))
                self.store.event(task["id"], "failed_before_acceptance", {"code": code, **detail})
            else:
                self.store.change(task["id"], status="queued", hold=0, slot=0, billing_pending=0,
                                  error="captcha_retry_scheduled", next_check=now + delay)
                self.store.event(task["id"], "captcha_retry_scheduled",
                                 {"attempt": attempt, "retry_in_seconds": delay})

    async def submit_reserved(self, task_id, request):
        task = self.store.task(task_id)
        account_id = task["account_id"]
        account = self.store.account(account_id)
        if not account["enabled"]:
            raise PoolError("account_disabled", "Account disabled before submission", 409)
        if account["state"] == "needs_attention":
            raise PoolError("needs_attention", "Account browser needs attention", 409)
        if account["error"] == "captcha_unavailable" and account["cooldown"] > time.time():
            raise PoolError("captcha_account_cooling_down", "Account verification is cooling down", 409)
        references = self.task_references(task, request)
        captcha = await self.browsers.token(account_id)
        balance = await self.probe(account_id)
        # Include this task's own hold when checking external spending during CAPTCHA.
        if (
            balance["quota"] - balance["used"]
            < balance["reserved_credits"] + self.settings.quota_safety_margin
        ):
            raise PoolError("balance_changed", "Balance changed before submission", 409)
        if not self.store.account(account_id)["enabled"]:
            raise PoolError("account_disabled", "Account disabled before submission", 409)
        # No await between the durable submit intent and the one and only paid POST.
        self.store.begin_submission(task_id)
        generation_id, response = await self.upstream.submit(
            account_id, request, references, captcha
        )
        captcha = None
        atomic_write(
            self.receipt_path(task_id),
            self.store.vault.seal(
                {"task_id": task_id, "account_id": account_id, "upstream_id": generation_id}
            ).encode(),
        )
        self.store.change(
            task_id, status="processing", upstream_id=generation_id, error=None, next_check=0
        )
        self.store.event(
            task_id, "submitted", {"upstream_id": generation_id, "response": loggable(response)}
        )

    async def poll(self, task_id):
        task = self.store.task(task_id)
        try:
            result = await self.upstream.status(task["account_id"], task["upstream_id"])
            status = result.get("status")
            if status in {"completed", "failed", "cancelled"}:
                finish = self.store.change if status == "completed" else self.store.finish_task
                finish(
                    task_id,
                    status="completed" if status == "completed" else "failed",
                    slot=0,
                    result=self.store.vault.seal(result),
                    error=None
                    if status == "completed"
                    else result.get("failure_reason") or "upstream_generation_failed",
                )
                submitted = self.store.one(
                    "SELECT time FROM events WHERE task_id=? AND kind='submitted' "
                    "ORDER BY id DESC LIMIT 1",
                    (task_id,),
                )
                now = time.time()
                self.store.event(
                    task_id,
                    "terminal",
                    {
                        "status": status,
                        "elapsed_total_secs": round(now - task["created"], 1),
                        "elapsed_generation_secs": round(now - submitted["time"], 1)
                        if submitted
                        else None,
                        "response": loggable(result),
                    },
                )
                try:
                    await self.settle_if_observed(task_id)
                except Exception as exc:
                    # A failed balance refresh must not undo a terminal result or
                    # keep its local reservation alive.
                    self.store.event(task_id, "balance_refresh_failed", {
                        "code": exc.code if isinstance(exc, PoolError) else "balance_refresh_failed"
                    })
            else:
                self.store.change(
                    task_id, next_check=time.time() + self.settings.poll_seconds, error=None
                )
                # Prepaid usage can become visible while the video is still generating.
                # Monetary holds and upstream generation slots have independent lifetimes.
                await self.probe(task["account_id"])
        except Exception as exc:
            code = exc.code if isinstance(exc, PoolError) else "status_unavailable"
            self.store.change(
                task_id, error=code, next_check=time.time() + self.settings.poll_seconds
            )

    async def settle_if_observed(self, task_id):
        task = self.store.task(task_id)
        await self.probe(task["account_id"])

    def matches(self, task, generation):
        request = VideoRequest.model_validate(self.store.vault.open(task["request"]))
        params = generation.get("model_parameters", {})
        if not isinstance(params, dict):
            return False
        expected = request.parameters(self.task_references(task, request))
        reference_keys = {"reference_images", "reference_videos", "reference_audios"}
        for key in reference_keys:
            actual = params.get(key) or []
            if not isinstance(actual, list) or any(not isinstance(ref, dict) for ref in actual):
                return False
            # API history can serialize optional reference fields as null.
            actual = [{k: v for k, v in ref.items() if v is not None} for ref in actual]
            if actual != expected.get(key, []):
                return False
        return (
            generation.get("workspace_id") == task["workspace_id"]
            and generation.get("model_id") == request.model
            and generation.get("prompt") == request.prompt
            and all(params.get(k) == v for k, v in expected.items()
                    if k not in reference_keys | {"start_frame", "end_frame"})
            and all(
                {k: v for k, v in (params.get(key) or {}).items() if v is not None}
                == expected.get(key, {})
                for key in ("start_frame", "end_frame")
                if isinstance(params.get(key) or {}, dict)
            )
            and all(isinstance(params.get(key) or {}, dict) for key in ("start_frame", "end_frame"))
            and generation.get("created_at_unix", 0) >= task["created"] - 300
        )

    async def reconcile(self, task_id, upstream_id=None):
        task = self.store.task(task_id)
        if task["status"] != "submission_unknown":
            return self.store.public_task(task_id)
        request = VideoRequest.model_validate(self.store.vault.open(task["request"]))
        # Text-only requests have no unique imported asset to distinguish an older
        # identical generation or another client's submission. Require an explicit ID.
        if not request.reference_urls() and not upstream_id:
            self.store.change(task_id, error="manual_reconciliation_required", next_check=time.time()+30)
            return self.store.public_task(task_id)
        try:
            history = await self.upstream.history(task["account_id"])
            matches = [g for g in history if (not upstream_id or g["id"] == upstream_id)
                       and self.matches(task, g) and not self.store.one(
                           "SELECT id FROM tasks WHERE upstream_id=? AND id<>?",
                           (g["id"], task_id))]
            if len(matches) == 1 and (upstream_id is None or matches[0]["id"] == upstream_id):
                self.store.change(
                    task_id,
                    upstream_id=matches[0]["id"],
                    status="processing",
                    error=None,
                    next_check=0,
                )
                self.store.event(task_id, "reconciled", {"upstream_id": matches[0]["id"]})
            else:
                self.store.change(
                    task_id,
                    error="manual_reconciliation_required",
                    next_check=time.time() + max(30, self.settings.poll_seconds),
                )
        except Exception:
            self.store.change(
                task_id, error="reconciliation_unavailable", next_check=time.time() + 30
            )
        return self.store.public_task(task_id)

    async def close(self):
        self.stopping = True
        jobs = self.loops + list(self.running.values())
        for job in jobs:
            job.cancel()
        await asyncio.gather(*jobs, return_exceptions=True)
        await self.browsers.close()
        await self.upstream.close()