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()