#!/usr/bin/env python3 """higgs_vclaw_adapter — videoclaw-v3 `seedance-direct` provider adapter that renders on our FREE Higgsfield Seedance engine (seedance_mini_unlimited) instead of the paid suitui-ai / ark path. videoclaw-v3 resolves a per-route adapter command from an env var (VCLAW_SEEDANCE_DIRECT_ADAPTER) and drives it as a subprocess: it spawns `sh -lc ""`, writes ONE JSON object to the child's STDIN, and reads ONE JSON result from STDOUT. There are four actions, distinguished by the `action` field on stdin (SUBMIT carries no `action`): SUBMIT stdin = the raw VideoExecutionPayload (NO `action` field): { workspaceRoot, outputDir, executionProfile{aspectRatio, quality, resolution, generateAudio, ...}, tasks:[{sceneIndex, prompt, durationSeconds, referencePaths[], ...}], ... } stdout = { externalJobId, rawResult:{ externalJobId, submittedScenes:[{sceneIndex, taskId}], responses:[...] } } POLL stdin = { action:'poll', projectSlug, routeId, externalJobId, outputDir, workspaceRoot } stdout = VideoExecutionPollResult: { status:'pending'|'completed'|'failed', externalJobId, outputs:[{ id, kind:'video', path, sceneIndex, backend:'seedance-direct' }], issues:[], rawResult } CANCEL stdin = { action:'cancel', projectSlug, routeId, externalJobId, outputDir, workspaceRoot } stdout = VideoExecutionCancelResult: { status:'cancelled'|'unsupported', externalJobId, issues:[], rawResult } WALLET stdin = { action:'wallet' } stdout = the authenticated, read-only Higgsfield wallet object. The unattended free worker uses it to bracket completion. (These shapes were verified against videoclaw-v3 `src/video/types.ts`, `src/video/execution-runtime.ts` submit/poll/cancel spawn blocks, and the reference impl `src/video/native-seedance.ts`. See VCLAW_INTEGRATION.md.) JOB STATE — mirrors native-seedance.ts writeJobState/readJobState exactly: location: /.vclaw-jobs/.json shape: { externalJobId, routeId:'seedance-direct', outputDir, createdAt, scenes:[{ sceneIndex, prompt, taskId, outputPath, status:'submitted'|'completed'|'failed', error? }] } `taskId` = our Higgsfield job_set_id (per scene). SUBMIT submits-and-returns (does NOT block on render — single free slot + slow queue). POLL is the long-running resolver videoclaw calls on a schedule: it queries each job_set, confirms-rights on ip_detected, downloads each completed scene to /scene-.mp4 (only if not already present). FREE-SAFETY (critical): only the unlimited free path is ever used (use_unlim=true, use_free_gens=false). SUBMIT reads the wallet before and after the submit batch; if credits_balance drops, it aborts with an issue (a real spend must never occur). All renders are free; credits_balance must stay flat. PHASE-2 — IMAGE-TO-VIDEO (ordered image-reference upload): When a task carries `referencePaths`, the adapter uploads up to nine visual references to Higgsfield in stable array order and attaches them as image media. The first is the frame-zero/composition seed; later images reinforce identity, costume, location or props. Human prompts bind them as @image1, @image2, ...; the adapter compiles those slots to Higgsfield's <<>>, <<>>, ... grammar and appends any missing citation so attached media cannot ship silently. For each usable visual reference in `referencePaths`: - an image (.png/.jpg/.jpeg/.webp) is uploaded directly (proven 3-step flow); - a VIDEO (.mp4/.mov/.webm/...) has its LAST frame extracted with ffmpeg (`-sseof`) and that still is uploaded — this implements SEED-style tail-frame RELAY (videoclaw chains a scene from the previous scene's output video); - other image types are normalized to PNG via ffmpeg; audio/unknown refs are skipped as start-frame candidates. i2v renders route to the proven Unlimited I2V model; pure text-to-video (no usable reference) stays byte-identical to phase-1 on `seedance_mini_unlimited` with `medias:[]`. Both stay on the free path (`use_unlim:true`, `use_free_gens:false`) and under the same wallet hard-abort. When a reference is present but cannot be applied (missing file, unsupported type, upload/ffmpeg failure) the scene renders text-to-video and a non-fatal issue is surfaced — spend is never risked for it. - executionProfile.quality is informational only here (single free tier); aspectRatio/resolution/generateAudio ARE mapped. """ import fcntl import json import os import re import shutil import subprocess import sys import tempfile import time # --------------------------------------------------------------------------- # Constants (match the proven free flow in counsel_mini_clean.py / higgs_engine.py) # --------------------------------------------------------------------------- # THE API HOST IS LOAD-BEARING, not cosmetic. Read docs/PLAN-unblock-datadome.md. # # Higgsfield fronts its API with DataDome. The browser tag patches window.fetch # and attaches a rotating session header — but ONLY for hosts on its allowlist, # read live from the page on 2026-08-07: # # window.ddoptions.ajaxListenerPath -> [{"host": "fnf-api-gw.higgsfield.ai"}] # window.ddoptions.sessionByHeader -> true # # Exactly one host, and the engine's old one (fnf.higgsfield.ai) is not it. So # every request we made went out WITHOUT the DataDome session while every UI # request carried it. Under some volume the gateway tolerates that; ten submits # in seven minutes did not, and the block never lifted. # # Calling the allowlisted host means our in-page fetch runs through the same # patched fetch the UI uses and inherits the session automatically — including # the ROTATION (the server re-issues the token via x-set-cookie on nearly every # protected response). Hand-injecting a value from document.cookie would replay a # stale token, which is itself a bot signal. # # Both overridable: the /fnf mapping is observed for the endpoints that matter # (/fnf/workspaces/wallet, /fnf/jobs/v2/, /fnf/video[/[/upload]], # /fnf/jobs/) but is structural inference for /media/batch, /media/ and # /job-sets/. A 404 on one of those is then a one-env-var rollback. API_BASE = os.environ.get("HIGGS_VCLAW_API_BASE", "https://fnf-api-gw.higgsfield.ai") API_PREFIX = os.environ.get("HIGGS_VCLAW_API_PREFIX", "/fnf") APP_URL = "https://higgsfield.ai/ai/video" # Keep the authenticated profile beside this vendored engine so the transport is # relocatable across machines and worktrees. Operators may still pin a different # private profile explicitly. DEFAULT_PROFILE = os.environ.get( "HIGGS_VCLAW_PROFILE", os.path.join(os.path.dirname(os.path.abspath(__file__)), ".cloak-profile"), ) # Chromium allows ONE instance per persistent profile (ProcessSingleton). Two # concurrent vclaw renders share DEFAULT_PROFILE, so launches must be # serialized across processes or the second one aborts on SingletonLock. PROFILE_LOCK_TIMEOUT_S = int(os.environ.get("HIGGS_VCLAW_PROFILE_LOCK_TIMEOUT", "1800")) # Where drive_higgsfield.py sources the logged-in higgsfield.ai session from — # the user's real Chrome profile. Used by the automatic re-seed fallback when # the stealth profile's Clerk session has expired. CHROME_COOKIE_FILE = os.environ.get( "HIGGS_VCLAW_CHROME_COOKIES", os.path.expanduser("~/Library/Application Support/Google/Chrome/Profile 1/Cookies"), ) # The legacy free models now answer `unlimited_generation_not_allowed`; the # unlimited path moved onto `use_unlim` over a current model. Overridable so a # working legacy account can pin the old one. MODEL = os.environ.get("HIGGS_VCLAW_MODEL", "seedance_2_5") # Image-to-video routes to the model the Kurukshetra i2v batch proved free # (higgs_api.py / higgs_batch.py: 39/40 @ 8s, use_unlim). `seedance_mini_unlimited` # has no proven i2v/medias path; keep i2v on the proven model, T2V on Mini. MODEL_I2V = os.environ.get("HIGGS_VCLAW_I2V_MODEL", "seedance_2_5") # params.model is a MODE (default | video_edit | video_extension), distinct from # the MODEL name in the URL. Sending the model name here is HTTP 422. PARAMS_MODE = os.environ.get("HIGGS_VCLAW_PARAMS_MODE", "default") # Optional passthrough for job-set fields this adapter does not model itself # (genre, speedramp, prompt_language, multi_shots + multi_shot_mode + # multi_prompt, reference_elements, extension_mode). The shape is NOT guessed: # it mirrors a captured real UI submit, the same source higgs_engine.py copies # from — see shows/seematti-texture.json in the upstream engine, where # seedance_2_5 renders with: # {"genre":"auto","multi_shots":false,"multi_shot_mode":"custom", # "multi_prompt":[],"speedramp":"auto","reference_elements":[], # "prompt_language":"en","extension_mode":null,"medias":[]} # # Unset -> nothing is added and the submitted params are byte-identical, which # matters because 2.5 already renders fine without any of them. This exists so # the fields can be exercised without another adapter change, not because they # are required. _EXTRA_RAW = os.environ.get("HIGGS_VCLAW_EXTRA_PARAMS", "").strip() try: EXTRA_PARAMS = json.loads(_EXTRA_RAW) if _EXTRA_RAW else {} if not isinstance(EXTRA_PARAMS, dict): raise ValueError("HIGGS_VCLAW_EXTRA_PARAMS must be a JSON object") except (ValueError, TypeError) as exc: # Fail loudly at import: a malformed override silently ignored would submit # a different job than the operator asked for. raise SystemExit(f"HIGGS_VCLAW_EXTRA_PARAMS is not a valid JSON object: {exc}") # The mixed-role citation note is off by default: it applies to every voice # render, so on by default it would be pure noise. Set when debugging a voice # that did not take. VERBOSE_MEDIA_NOTE = os.environ.get("HIGGS_VCLAW_VERBOSE_MEDIA_NOTE") == "1" # Cloudflare Turnstile. A 10-scene batch on 2026-08-07 got EXACTLY 3 submits # through and then 403'd the remaining 7 with this marker — rapid sequential # submits trip the anti-bot gate. Spacing them out avoids it; the retry in # submit_scene recovers when it fires anyway. TURNSTILE_MARKER = "turnstile_required" TURNSTILE_MAX_RETRIES = int(os.environ.get("HIGGS_VCLAW_TURNSTILE_RETRIES", "12")) # Seconds to wait BETWEEN scene submits. 3 back-to-back tripped Turnstile, so # the default paces them; a single-scene submit pays nothing for this. SUBMIT_SPACING_S = int(os.environ.get("HIGGS_VCLAW_SUBMIT_SPACING_S", "10")) # Per-model duration windows. The production policy for authenticated Higgsfield # Unlimited uses the current 20s ceiling; the 2.0 family remains at 15s. # Out-of-range values # are a 422 on EVERY submit — the same failure shape the resolution clamp already # exists to prevent, which once killed a whole overnight batch on its first night. # Clamping is deterministic and visible in the submitted params, and a 20s render # beats no render. MODEL_DURATION_RANGE = { "seedance_2_5": (4, 20), "seedance_2_0": (4, 15), "seedance_2_0_fast": (4, 15), "seedance_2_0_mini": (4, 15), "seedance_unlimited": (4, 15), "seedance_mini_unlimited": (4, 15), } DEFAULT_DURATION_RANGE = (4, 15) # conservative for an unknown model def clamp_duration(seconds, model=None): """Clamp a requested duration into the model's accepted window. Pure. An unknown model gets the conservative window rather than the widest one: a clip shorter than asked for is a visible, recoverable disappointment, while a 422 is a dead batch. """ lo, hi = MODEL_DURATION_RANGE.get(model or "", DEFAULT_DURATION_RANGE) try: want = int(seconds) if seconds else 8 except (TypeError, ValueError): want = 8 return max(lo, min(hi, want)) # Video upload (captured 2026-08-07 — see docs/CAPTURED-VIDEO-UPLOAD.md). # The gateway path is "/fnf/video"; this adapter's host drops the "/fnf" prefix, # the same way its "/media/batch" is "/fnf/media/batch" there. Overridable because # that prefix mapping is the one part of the flow not observed on THIS host. VIDEO_PATH = os.environ.get("HIGGS_VCLAW_VIDEO_PATH", "/video") # Audio has its own endpoint family, discovered from its own 422s. It is the # RIGHT channel for a cloned-voice track: the black-frame-video hack exists only # because the old Runway lane had no audio field. Measured 2026-08-07 — a window # whose voice was rejected by moderation SIX times as a video media was accepted # on the first attempt as an audio media. AUDIO_PATH = os.environ.get("HIGGS_VCLAW_AUDIO_PATH", "/audio") # Higgsfield rejects a video media below this edge. Our black-frame voice clips # are 720x1280, so they clear it; a 512x512 one would be accepted at upload and # then fail opaquely at submit, which is the failure this makes legible. MIN_VIDEO_EDGE = int(os.environ.get("HIGGS_VCLAW_MIN_VIDEO_EDGE", "640")) # The ONLY models this engine may submit to. # # READ THIS BEFORE EDITING THE SET — the safety model changed on 2026-08-07 and is # now WEAKER than it was. # # It used to be categorical: every entry was a dedicated unlimited-tier model that # could not be billed whatever we sent. It is now CONDITIONAL. `seedance_2_5` is a # model that quotes ~32 credits for an 8s clip when "Unlimited mode" is off; it is # free only because we send `use_unlim: true` and the server honours it. The # legacy `*_unlimited` names now answer `unlimited_generation_not_allowed`, so # there is no categorical option left to choose. # # The evidence for `seedance_2_5` is measurement, not its name: two UI generations # plus one adapter submit, wallet 6005 credits before and 6005 after. But that is # n=3 on ONE account with an active Unlimited entitlement. If that entitlement # lapses, or the unlimited pool is exhausted mid-batch and the server bills instead # of refusing, this allowlist waves the submit through and only the per-scene # wallet delta catches it — one charged scene later. # # So: adding a model here is no longer "is it named *_unlimited". It is "have I # measured the wallet across a real submit on it". Do not add one from a # screenshot, a docs table, or a name. # # The allowlist still earns its place, because the wallet check is FORENSIC rather # than preventive: it reads the balance, submits every scene, reads it again, and # only then aborts — so without this gate a paid model would be noticed after all # 25 scenes were away. `HIGGS_VCLAW_MODEL` / `HIGGS_VCLAW_I2V_MODEL` are env vars, # so a typo or a well-meant "just point it at 2.0" is exactly how that happens. # `seedance_2_5` is here on MEASURED evidence, not on its name. The legacy # `*_unlimited` models now answer `unlimited_generation_not_allowed` on this # account — the unlimited mechanism moved off dedicated model names and onto the # `use_unlim` flag applied to a current model. Two UI generations on # seedance_2_5 with use_unlim:true were captured on 2026-08-07, and the wallet # read 6005 credits before them and 6005 after: zero movement. That is the only # reason it is on this list. The per-scene wallet guard below is still the # backstop if the unlimited pool is ever exhausted and it silently falls back to # credits. FREE_MODELS = frozenset({ "seedance_mini_unlimited", # legacy, currently refused by the API "seedance_unlimited", # legacy, currently refused by the API "seedance_2_5", # proven zero-cost with use_unlim (see above) }) # The submit error the API returns when the account holds no unlimited allowance. # Seen on every model once the entitlement lapsed; each scene costs a full retry # budget before failing, so N scenes waste N × that for one already-known answer. NO_UNLIM_MARKER = "unlimited_generation_not_allowed" # The only resolutions the free /jobs/v2 endpoints accept (anything else -> 422). FREE_RESOLUTIONS = ("480p", "720p") ROUTE_ID = "seedance-direct" BACKEND = "seedance-direct" # Reference-type routing for start-frame selection. Images in DIRECT_IMAGE_MIME are # uploaded byte-for-byte (the proven /media/batch flow); videos have their last frame # extracted (relay); other images are normalized to PNG via ffmpeg. DIRECT_IMAGE_MIME = { ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", } OTHER_IMAGE_EXTS = (".gif", ".bmp", ".tif", ".tiff", ".heic", ".heif") VIDEO_EXTS = (".mp4", ".mov", ".webm", ".mkv", ".m4v", ".avi", ".mpg", ".mpeg") # In-page fetch: mint a fresh Clerk token and call the fnf API with it. # Response headers are returned as well as the body. They used to be discarded, # which is precisely why DataDome's rotation (x-set-cookie / x-dd-b on protected # responses) was invisible to us for hours while we theorised about rate limits # and headless fingerprints. Only the bot-protection headers are kept — the whole # set would be noise, and x-set-cookie's VALUE is a session token, so only its # length is recorded. _FETCH_JS = ( 'async ({m,p,b})=>{const t=await window.Clerk.session.getToken();' 'const h={Authorization:"Bearer "+t};if(b)h["Content-Type"]="application/json";' 'const r=await fetch("' + API_BASE + API_PREFIX + '"+p,{method:m,headers:h,' 'body:b?JSON.stringify(b):undefined,credentials:"include"});' 'let j;try{j=await r.json()}catch{j=null}' 'const dh={};r.headers.forEach((v,k)=>{if(/^x-(datadome|set-cookie|dd-b)/i.test(k))' 'dh[k]=(k.toLowerCase()==="x-set-cookie"?("<"+v.length+" chars>"):v)});' 'return{status:r.status,body:j,ddHeaders:dh}}' ) # Terminal failure statuses returned by the job poll. FAIL_STATUSES = ("nsfw", "failed", "error", "canceled", "cancelled") # Resolution -> (width, height) for the two aspect ratios we support. Higgsfield's # free submit shape wants explicit width/height (the UI otherwise drifts to 480p). _DIMS = { "16:9": {"720p": (1280, 720), "1080p": (1920, 1080)}, "9:16": {"720p": (720, 1280), "1080p": (1080, 1920)}, "1:1": {"720p": (720, 720), "1080p": (1080, 1080)}, } # --------------------------------------------------------------------------- # Pure helpers (no network) — keep these testable without a browser. # --------------------------------------------------------------------------- def job_state_dir(output_dir): return os.path.join(output_dir, ".vclaw-jobs") def job_state_path(output_dir, external_job_id): return os.path.join(job_state_dir(output_dir), f"{external_job_id}.json") def write_job_state(state): """Persist job state. Mirrors native-seedance writeJobState (pretty JSON + \\n).""" d = job_state_dir(state["outputDir"]) os.makedirs(d, exist_ok=True) path = job_state_path(state["outputDir"], state["externalJobId"]) with open(path, "w") as f: f.write(json.dumps(state, indent=2) + "\n") def read_job_state(output_dir, external_job_id): path = job_state_path(output_dir, external_job_id) if not os.path.exists(path): raise RuntimeError(f"Seedance native job state not found for {external_job_id}.") with open(path) as f: return json.load(f) def scene_out_path(output_dir, scene_index): """videoclaw's per-scene output convention: outputDir/scene-.mp4.""" return os.path.join(output_dir, f"scene-{scene_index}.mp4") def _ffmpeg_bin(): """Path to ffmpeg, or None when it is not installed (relay/normalize degrade gracefully to text-to-video).""" return shutil.which("ffmpeg") def _run_ffmpeg(args, timeout=120): """Run ffmpeg quietly; return True on a clean exit that produced the output. The caller passes the output path last so we can size-check it.""" ff = _ffmpeg_bin() if not ff: return False out = args[-1] try: r = subprocess.run([ff, "-nostdin", "-y", *args], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=timeout) except Exception: return False return r.returncode == 0 and os.path.exists(out) and os.path.getsize(out) > 0 def extract_last_frame(video_path, tmp_dir): """Extract a video's LAST frame to a PNG (SEED-style tail-frame relay). `-sseof -0.3` seeks ~0.3s before EOF; a very short clip may have nothing past that, so we fall back to a plain single-frame decode. Returns the PNG path or None (ffmpeg missing / decode failed) — caller then renders text-to-video. """ out = os.path.join(tmp_dir, f"tail-{abs(hash(video_path)) & 0xffffffff:08x}.png") # -sseof before -i (fast tail seek); map the first video stream only. if _run_ffmpeg(["-sseof", "-0.3", "-i", video_path, "-map", "0:v:0", "-frames:v", "1", out]): return out if _run_ffmpeg(["-i", video_path, "-map", "0:v:0", "-frames:v", "1", out]): return out return None def normalize_image_to_png(image_path, tmp_dir): """Re-encode an image Higgsfield's direct upload doesn't take (gif/bmp/heic/...) to a PNG via ffmpeg. Returns the PNG path or None.""" out = os.path.join(tmp_dir, f"img-{abs(hash(image_path)) & 0xffffffff:08x}.png") return out if _run_ffmpeg(["-i", image_path, "-frames:v", "1", out]) else None def _extract_audio_track(path, tmp_dir): """Pull the audio out of a black-frame voice clip. Returns a path or None. The voice assets are mp4s whose only purpose is to carry a track (the video is a black rectangle). The audio channel wants the track itself. """ if not path: return None if os.path.splitext(path)[1].lower() in (".mp3", ".m4a", ".wav", ".aac", ".ogg"): return path if not shutil.which("ffmpeg"): return None out = os.path.join(tmp_dir, os.path.splitext(os.path.basename(path))[0] + ".mp3") try: subprocess.run(["ffmpeg", "-v", "error", "-i", path, "-vn", "-acodec", "libmp3lame", "-q:a", "2", "-y", out], check=True, capture_output=True, timeout=120) except Exception: return None return out if os.path.exists(out) and os.path.getsize(out) > 0 else None def _resolve_ref_path(raw, base_dirs=()): """Resolve one referencePath to an existing file, or None. Extracted from resolve_start_frame so the voice path gets the same handling. videoclaw asset manifests commonly store PROJECT-RELATIVE paths, and before this retry existed those silently degraded a scene to text-to-video (the Last Call absolute-path foot-gun). A voice ref would fail the same way, just more quietly — the clip renders, only mute. """ if not raw or not isinstance(raw, str): return None candidates = [raw] if os.path.isabs(raw) else [raw] + [ os.path.join(base, raw) for base in base_dirs if base ] for candidate in candidates: if os.path.exists(candidate): return candidate return None def resolve_start_frame(ref_paths, tmp_dir, base_dirs=()): """Pick one i2v image from a task's referencePaths (videoclaw order). Returns (image_path, mime, source) where source describes provenance, or (None, None, None) if no referencePath yields a usable still. Skips audio / unknown / missing refs. A video's LAST frame is extracted (relay); an unusual image type is normalized to PNG; a direct image is used as-is. A RELATIVE path is retried against each of `base_dirs` (the caller passes the project dir + workspace root): videoclaw asset manifests commonly store project-relative paths, and before this resolution step those silently degraded the scene to text-to-video (the Last Call absolute-path foot-gun). """ for raw in ref_paths or []: if not raw or not isinstance(raw, str): continue p = _resolve_ref_path(raw, base_dirs) if p is None: continue ext = os.path.splitext(p)[1].lower() if ext in DIRECT_IMAGE_MIME: return p, DIRECT_IMAGE_MIME[ext], {"from": p, "extractedFromVideo": False} if ext in VIDEO_EXTS: frame = extract_last_frame(p, tmp_dir) if frame: return frame, "image/png", {"from": p, "extractedFromVideo": True} continue # ffmpeg unavailable/failed — try the next reference if ext in OTHER_IMAGE_EXTS: frame = normalize_image_to_png(p, tmp_dir) if frame: return frame, "image/png", {"from": p, "extractedFromVideo": False} continue # audio / unknown extension — not a start-frame candidate return None, None, None def resolve_image_references(ref_paths, tmp_dir, base_dirs=(), max_images=9): """Resolve visual references in stable payload order. The first image is the frame-zero/composition anchor. Later images reinforce identity, costume, location or prop evidence and bind as @image2, @image3, etc. Every source keeps its original path for the execution receipt. """ resolved = [] for raw in ref_paths or []: image, mime, source = resolve_start_frame([raw], tmp_dir, base_dirs) if image: resolved.append((image, mime, source)) if len(resolved) >= max_images: break return resolved def build_params(profile, prompt, duration_seconds, medias=None, model=None): """Build the params dict for one scene's free Higgsfield submit. Maps videoclaw's executionProfile -> the Higgsfield free submit shape. `medias` defaults to [] and `model` to the free Mini tier — i.e. calling this with just (profile, prompt, duration) is byte-identical text-to-video. Phase-2 passes an uploaded start-frame media + MODEL_I2V for image-to-video. quality is intentionally NOT mapped (informational only). """ aspect = profile.get("aspectRatio") or "16:9" resolution = profile.get("resolution") or "720p" # The free endpoint accepts ONLY 480p/720p — any other value (e.g. a 1080p # executionProfile) fails EVERY submit with HTTP 422 literal_error, which is # exactly how the Last Call batch died on its first night. Clamp instead of # passing the rejection through: a 720p render beats no render, and the # clamp is deterministic + visible in the submitted params. if resolution not in FREE_RESOLUTIONS: resolution = "720p" width, height = _DIMS.get(aspect, _DIMS["16:9"]).get( resolution, _DIMS["16:9"]["720p"] ) params = { "model": model or MODEL, "prompt": prompt, "duration": clamp_duration(duration_seconds, model or MODEL), "aspect_ratio": aspect, "resolution": resolution, "bitrate_mode": "high", "generate_audio": bool(profile.get("generateAudio", True)), "width": width, "height": height, "medias": medias or [], } # Caller-supplied fields last so an explicit override wins, but never let # them clobber the identity of the submit (model/prompt/medias are ours). for key, value in EXTRA_PARAMS.items(): if key in ("model", "prompt", "medias"): continue params[key] = value return params def submit_body(params): """POST body for /jobs/v2/{MODEL} — free path only. `batch_size` and `use_unlim` are repeated INSIDE params to mirror the captured UI submit (docs/CAPTURED-VIDEO-UPLOAD.md), which is the only shape observed returning 200 on the current free path. The outer `use_unlim` is kept as well; the UI sends both. """ # `params.model` is a MODE, not a model name. The server is explicit about it: # # {"loc": ["model"], "msg": "Input should be 'default', 'video_edit' or # 'video_extension'", "input": "seedance_2_5"} (HTTP 422) # # The MODEL name selects the endpoint — /jobs/v2/ — and submit_scene has # already used params["model"] to build that path by the time we get here. So # the body carries the mode instead. This is exactly what the captured UI # submit sends ("model": "default" alongside /jobs/v2/seedance_2_5); mirroring # it was the fix. # # video_edit / video_extension are the v2v modes — not used here, noted so the # next person does not rediscover them from a 422. return {"params": {"batch_size": 1, "use_unlim": True, **params, "model": PARAMS_MODE}, "use_unlim": True, "use_free_gens": False} def url_of(job): """Result URL from a job-set job entry (raw preferred, min fallback).""" r = job.get("results") or {} return (r.get("raw") or {}).get("url") or (r.get("min") or {}).get("url") def _summarize_body(body, limit=400): """Render a server response body to a short single-line string for issues[]. The job-creation endpoint returns its rejection reason (moderation gate, param validation, quota, ...) in this body; SUBMIT used to discard it and report a bare "no job_set id returned". Keep it compact but faithful so the real reason reaches videoclaw's issues[] and the on-disk job state.""" if body is None: return "" try: s = json.dumps(body, ensure_ascii=False) except (TypeError, ValueError): s = str(body) s = " ".join(s.split()) return s if len(s) <= limit else s[:limit] + "…" # --------------------------------------------------------------------------- def _profile_seed(profile: str) -> int: """ONE PROFILE, ONE SEED. CloakBrowser builds a full device identity from the --fingerprint seed; without one, every launch presents the SAME cookies on a DIFFERENT machine — "something real users never do" (their own cheat sheet), and a plausible feeder of the Turnstile challenges this lane has eaten. The seed is minted once per profile and lives beside it, so profile and device stay paired for life; a new profile gets a new seed by construction.""" marker = os.path.join(profile, ".fingerprint-seed") try: return int(open(marker, encoding="utf-8").read().strip()) except (OSError, ValueError): pass seed = int.from_bytes(os.urandom(3), "big") % 900000 + 100000 os.makedirs(profile, exist_ok=True) tmp = marker + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: fh.write(str(seed)) os.replace(tmp, marker) return seed def _clear_stale_singleton(profile): """Remove a stale Chromium ProcessSingleton so the next launch_persistent_context doesn't hang the full timeout. A crashed CloakBrowser leaves `SingletonLock` -> "-" behind; we clear it ONLY when that pid is dead. A live holder (a non-flocked render still running) is left untouched, as is an unparseable/foreign lock. Call while holding the profile flock so no live HiggsSession races the removal.""" lock = os.path.join(profile, "SingletonLock") try: pid_s = os.readlink(lock).rpartition("-")[2] except OSError: return # no singleton (or not a symlink) — nothing to do try: os.kill(int(pid_s), 0) return # holder is alive — genuine, leave it be except (ValueError, PermissionError): return # unparseable, or a live process that isn't ours — don't touch except ProcessLookupError: pass # dead pid -> the lock is stale, fall through and clear it for name in ("SingletonLock", "SingletonSocket", "SingletonCookie"): try: os.unlink(os.path.join(profile, name)) except OSError: pass print(f"[higgs-adapter] cleared stale Chromium singleton (dead pid {pid_s}) in {profile}", file=sys.stderr) VIDEO_EXTS = (".mp4", ".mov", ".webm", ".m4v", ".mkv") # A voice reference may arrive as a bare audio file rather than the black-frame # mp4 wrapper. Both are voices and both belong on the audio channel. AUDIO_EXTS = (".mp3", ".m4a", ".wav", ".aac", ".ogg", ".flac") def split_references(ref_paths, reference_role=None): """Split refs into (start_frame_candidates, voice_videos). Pure. Both a cloned-voice clip and a chain-relay seed arrive as a bare `.mp4`, and they need OPPOSITE handling: * a relay seed is the previous scene's output — tail-frame it to a still and use it as the i2v start frame (existing behaviour, unchanged); * a voice clip is a black-frame video whose audio locks the voice — upload it whole and attach it as a `role: "video"` media. Tail-framing one yields a black rectangle and silently discards the voice. videoclaw already distinguishes them, so this reads its existing field rather than inventing a filename convention: `execution-runtime.ts` puts seedance-direct in VOICE_REF_ROUTES and tags voice-clone mp4s `referenceRole: 'character'`, while chain seeds carry `'keyframe'`. Anything that is not a video, and every video when the role is not 'character', falls through to the start-frame path exactly as before — so a task without `referenceRole` behaves byte-identically to today. AMBIGUITY. `referenceRole` is a PER-TASK scalar, but `referencePaths` is a MIXED list. videoclaw resolves the role as `resolvedAssetUris.length ? 'character' : chainSeed ? 'keyframe' : …` — identity refs WIN over the chain seed — while `resolveSceneReferencePaths` still returns `[chainSeed, ...identityRefs, ...voiceClips]`, keeping both. So a show-bible project running --auto-chain emits role='character' with TWO videos in the list: the previous scene's output AND the voice clip. Treating every video as a voice then uploads the chain seed as a voice reference and silently breaks continuity. One video is the proven, unambiguous case. Two or more, and the role scalar genuinely cannot tell us which is which — so this REFUSES to guess and says so, falling back to the pre-existing behaviour. A voice that is loudly not applied is recoverable; a chain seed silently re-cast as a voice is a continuity break nobody sees until the cut is assembled. Returns (starts, voices, warning) — warning is None when unambiguous. """ def _is_voice(x): return str(x).lower().endswith(VIDEO_EXTS + AUDIO_EXTS) videos = [p for p in (ref_paths or []) if _is_voice(p)] others = [p for p in (ref_paths or []) if not _is_voice(p)] if reference_role == "character" and len(videos) == 1: return others, videos, None if reference_role == "character" and len(videos) > 1: return list(ref_paths), [], ( f"{len(videos)} video referencePaths with referenceRole='character' — " "cannot tell the cloned-voice clip from a chain-relay seed, because " "referenceRole is per-task and referencePaths is mixed. NOT attaching a " "voice rather than risk uploading the chain seed as one (which would " "break continuity invisibly). Pass the voice as the only video, or have " "videoclaw emit per-path roles." ) # Not 'character': no voice is attached. Flag a video that may be a dropped # voice clone — the old code discarded it with no issue at all, which is the # invisible failure this module exists to prevent. if reference_role != "character" and len(videos) > 1: return list(ref_paths), [], ( f"{len(videos)} video referencePaths with referenceRole=" f"{reference_role!r} — if one is a cloned-voice clip it was NOT applied " "(only referenceRole='character' attaches a voice). Rendered without " "the voice lock." ) return list(ref_paths), [], None def build_media_entries(images=(), videos=()): """Build `params.medias` and the prompt citation tokens. Pure. Captured shape (2026-08-07, docs/CAPTURED-VIDEO-UPLOAD.md): "prompt": "<<>> in the concert singing <<>>", "medias": [{"role": "video", "data": {}}] `data` is the whole media object the upload returned — not a trimmed {id,url,type} — because that is what the UI sends and we do not know which fields the server actually reads. This is exactly the shape higgs-cloak's `medias_from_config` emitted and its PIPELINE.md documented (`@Video N` -> `<<>>`). That grammar was always right; the missing piece was the upload, which is why replaying a UI-made media id worked and nothing else did. INDEXING. Tokens are numbered per ROLE (`video_1` is the first video), which is unambiguous for our case — one voice clip per window. PIPELINE.md says "1-indexed into params.medias", which would be a GLOBAL index; the two agree whenever a submit carries a single role, which is every case observed. If a mixed image+video submit is ever needed, capture one before trusting either reading. """ medias, tokens = [], [] for i, m in enumerate(images or (), 1): medias.append({"role": "image", "data": m}) tokens.append(f"<<>>") n_audio = n_video = 0 for m in videos or (): # An audio media must ride the "audio" role. Sending it as "video" is how # the black-frame workaround expressed itself, and it is exactly the shape # moderation rejects. if (m or {}).get("type") == "audio_input": n_audio += 1 medias.append({"role": "audio", "data": m}) tokens.append(f"<<>>") else: n_video += 1 medias.append({"role": "video", "data": m}) tokens.append(f"<<>>") return medias, tokens def cite_in_prompt(prompt, tokens): """Append citation tokens that the prompt does not already carry. An operator-authored prompt may already place `<<>>` deliberately — mid-sentence placement reads differently to the model than a trailing tag, and the captured example puts it inline ("in the concert singing <<>>"). So only append what is genuinely absent, and never reorder what is there. A media present in `medias[]` but never cited is the silent-failure case: the server accepts the submit and ignores the clip, which looks exactly like a voice that did not take. """ text = prompt or "" missing = [t for t in tokens if t not in text] if not missing: return text return (text.rstrip() + " " + " ".join(missing)).strip() def compile_media_citations(prompt, tokens): """Compile human-facing @imageN/@videoN/@audioN slots to Higgsfield tokens. VideoClaw prompt artifacts use the portable, reviewable ``@image1`` syntax. Higgsfield's captured job contract uses ``<<>>``. Attaching a media object without compiling/citing that token leaves the reference implicit and can make Seedance ignore it. Replace only canonical numbered slots, then append any still-uncited provider tokens so every attached media is referenced. """ text = prompt or "" for token in tokens: match = re.fullmatch(r"<<<(image|video|audio)_(\d+)>>>", token) if not match: continue role, index = match.groups() text = re.sub( rf"(?>>") return tokens # Live Higgsfield session (opens CloakBrowser — consequential, network). # --------------------------------------------------------------------------- class HiggsSession: """Minimal CloakBrowser wrapper over the never-throwing in-page api().""" def __init__(self, profile=DEFAULT_PROFILE, headless=None): # HEADLESS IS NOW OVERRIDABLE, and that matters for Cloudflare Turnstile. # # Evidence 2026-08-07: the manual capture ran HEADED and submitted twice # with no challenge at all. The headless adapter first got 3 submits # through, then 403'd `turnstile_required` on every subsequent one — # including single, well-spaced submits. Reloading the page did not clear # it. A headless fingerprint is challenged far more aggressively, and once # challenged it stays challenged. # # Default stays headless (unattended renders must not spawn windows), but # HIGGS_VCLAW_HEADLESS=0 runs headed when Turnstile has locked us out. if headless is None: headless = os.environ.get("HIGGS_VCLAW_HEADLESS", "1") != "0" from cloakbrowser import launch_persistent_context self._lock_fd = self._acquire_profile_lock(profile) # We hold the flock now, so any SingletonLock left in the profile is either a # crashed render's stale lock (dead pid -> cleared) or a live non-flocked render # (left intact). Clearing the stale case here is what stops the launch below from # hanging the full 180s and false-reporting the keepalive as DOWN. _clear_stale_singleton(profile) deadline = time.time() + PROFILE_LOCK_TIMEOUT_S while True: try: self.ctx = launch_persistent_context( profile, headless=headless, viewport={"width": 1280, "height": 800}, args=[f"--fingerprint={_profile_seed(profile)}"] ) break except Exception as e: # A render started before this lock existed can still hold the # profile without holding the flock — queue behind it too. if "ProcessSingleton" not in str(e) or time.time() >= deadline: self._release_profile_lock() raise print( "[higgs-adapter] profile held by another CloakBrowser; retrying in 20s", file=sys.stderr, ) time.sleep(20) self.page = self.ctx.pages[0] if self.ctx.pages else self.ctx.new_page() # higgsfield.ai is occasionally slow to hit domcontentloaded; a single transient # goto timeout shouldn't fail the whole session (the keepalive would false-report # the auth as DEAD). Retry once before giving up. for _attempt in range(2): try: self.page.goto(APP_URL, wait_until="domcontentloaded", timeout=60000) break except Exception: if _attempt == 1: raise self.page.wait_for_timeout(3000) self.page.wait_for_timeout(5000) def _acquire_profile_lock(self, profile): """Cross-process flock so only one HiggsSession opens the profile at a time; released in close() and automatically on process death.""" lock_path = profile.rstrip("/") + ".lock" fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644) deadline = time.time() + PROFILE_LOCK_TIMEOUT_S while True: try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) return fd except OSError: if time.time() >= deadline: os.close(fd) raise RuntimeError( f"Higgsfield profile busy: another render holds {lock_path} " f"(waited {PROFILE_LOCK_TIMEOUT_S}s). Retry after it finishes." ) print( f"[higgs-adapter] profile busy, waiting on {lock_path} ...", file=sys.stderr, ) time.sleep(10) def _release_profile_lock(self): fd = getattr(self, "_lock_fd", None) if fd is None: return self._lock_fd = None try: fcntl.flock(fd, fcntl.LOCK_UN) except OSError: pass try: os.close(fd) except OSError: pass def api(self, method, path, body=None): """Never throws. Retries transient evaluate failures.""" for k in range(5): try: return self.page.evaluate(_FETCH_JS, {"m": method, "p": path, "b": body}) except Exception: self.page.wait_for_timeout(3000) return {"status": 0, "body": None} def wallet(self): w = self.api("GET", "/workspaces/wallet").get("body") or {} return {k: w.get(k) for k in ("credits_balance", "subscription_balance")} def reseed_from_chrome(self): """Self-heal an expired Clerk session: pull fresh higgsfield.ai cookies from the user's Chrome (same flow as drive_higgsfield.py) into the LIVE context, reload the app, and re-probe. Returns True when the session authenticates afterwards. Never raises — callers decide how to fail.""" try: import browser_cookie3 jar = browser_cookie3.chrome( cookie_file=CHROME_COOKIE_FILE, domain_name="higgsfield.ai" ) cookies = [] for c in jar: ck = { "name": c.name, "value": c.value, "domain": c.domain, "path": c.path or "/", "secure": bool(c.secure), } if c.expires: ck["expires"] = float(c.expires) cookies.append(ck) if not cookies: print("[higgs-adapter] re-seed: no higgsfield.ai cookies in Chrome", file=sys.stderr) return False self.ctx.add_cookies(cookies) self.page.goto(APP_URL, wait_until="domcontentloaded", timeout=60000) self.page.wait_for_timeout(5000) ok = self.api("GET", "/workspaces/wallet").get("status") != 0 print(f"[higgs-adapter] re-seed from Chrome: " f"{'session restored' if ok else 'still unauthenticated'}", file=sys.stderr) return ok except Exception as e: print(f"[higgs-adapter] re-seed from Chrome failed: {e}", file=sys.stderr) return False def submit_scene(self, params): """Submit one scene; return (job_set_id, error). On success -> (job_set_id, None). On hard fail -> (None, reason) where `reason` carries the server's actual rejection (HTTP status + response body) so SUBMIT can surface it in issues[] instead of a bare "no job_set id returned". The body is where Higgsfield reports a moderation gate / param-validation / quota rejection at job creation. 429/0 = single free slot busy / transient -> wait and retry (does NOT block on render, just on getting the job accepted into the queue). """ # T2V params carry MODEL (byte-identical endpoint); i2v carries MODEL_I2V. model = params.get("model") or MODEL # SPEND GUARD — before the body is built, before anything leaves the # process. The wallet check in action_submit runs after every scene has # already been submitted, so it can only report a charge, never prevent # one. This can. if model not in FREE_MODELS: return None, ( f"REFUSED: model {model!r} is not on the free allowlist " f"({', '.join(sorted(FREE_MODELS))}). This engine renders on the " "unlimited tier only and will not submit to a paid model. If the " "model is genuinely free now, add it to FREE_MODELS deliberately." ) body = submit_body(params) path = f"/jobs/v2/{model}" zero_streak = 0 turnstile_hits = 0 for a in range(40): cr = self.api("POST", path, body) st = cr["status"] if st == 429: # single free slot busy — stay patient zero_streak = 0 self.page.wait_for_timeout(20000) continue if st == 0: # token/fetch failed — likely dead session zero_streak += 1 if zero_streak >= 3: # fail fast instead of a ~24-min grind return None, ( "in-page fetch returned status 0 three times (the Clerk " "token could not be minted — the Higgsfield session is " "likely dead/expired); no HTTP response was received." ) self.page.wait_for_timeout(5000) continue if st == 403 and TURNSTILE_MARKER in str(cr.get("body")): # Cloudflare Turnstile. Observed 2026-08-07 on a 10-scene batch: # EXACTLY the first 3 submits succeeded and every one after failed # — a rate limit on rapid submits, not a dead session (the same # session had just submitted successfully). # # We are inside a real browser, and Turnstile normally auto-clears # for a legitimate fingerprint once the widget runs. So reload the # app page to let it execute, then retry. Treat as transient rather # than fatal: returning here would fail the remaining scenes of a # batch that is otherwise perfectly healthy. # MEASURED 2026-08-07: the WEB UI hits this exact 403 too, then # retries and gets 200 on its second attempt. So it is TRANSIENT, # not a lockout. The 403 carries a fresh DataDome token # (x-set-cookie, 208 chars vs a healthy 128); the tag writes it # back and the next request carries a valid one. # # We were reloading the page between attempts, which resets the # tag mid-recovery — that is why our retries never converged while # the UI sailed through. Do what the UI does: retry promptly, # same page, and let the tag settle. turnstile_hits += 1 if turnstile_hits > TURNSTILE_MAX_RETRIES: return None, ( f"Cloudflare Turnstile blocked this submit " f"{turnstile_hits} times (HTTP 403 {TURNSTILE_MARKER!r}). " "The account is fine and the session is live — this is " "anti-bot rate limiting. Submit fewer scenes at once, or " "raise HIGGS_VCLAW_SUBMIT_SPACING_S (currently " f"{SUBMIT_SPACING_S}s between scenes)." ) print(f"[higgs-adapter] human-check 403 ({turnstile_hits}/" f"{TURNSTILE_MAX_RETRIES}); retrying like the UI does", file=sys.stderr) # Short settle so the tag can process the rotated token, then go # again. NO reload. Only fall back to a reload once several # prompt retries have failed, in case the tag really is wedged. self.page.wait_for_timeout(3000) if turnstile_hits >= 6: try: self.page.reload(wait_until="domcontentloaded", timeout=60000) self.page.wait_for_timeout(8000) except Exception: pass continue if st not in (200, 201): # A real HTTP rejection — the body holds the reason (moderation / # validation / quota). Surface it verbatim rather than swallowing. return None, f"HTTP {st} from POST {path}: {_summarize_body(cr.get('body'))}" jsid = ((cr.get("body") or {}).get("job_sets") or [{}])[0].get("id") if jsid: return jsid, None # 2xx but no job_set id — the server accepted the request yet created # no job (also how a silent content gate can present). Keep the body. return None, ( f"HTTP {st} but response carried no job_sets[].id: " f"{_summarize_body(cr.get('body'))}" ) return None, ( "submit retry budget exhausted (40 attempts) while the free slot " "stayed busy (HTTP 429); job was never accepted into the queue." ) def upload_image(self, path, mime="image/png"): """Upload one still; return (media, reason). On success -> ({id,url}, None) usable as an i2v start-frame media. On any hard failure -> (None, reason) where `reason` carries the REAL cause so the caller surfaces it instead of a misleading downstream 404. The Higgsfield NSFW/IP gate rejects a moderated reference at the finalize step (HTTP 422) and drives the media to a terminal `nsfw`/`failed` status; a media that never reaches a usable/finished state must NOT be handed to job creation — doing so returns HTTP 404 "Media input not found" (the confusing symptom this replaces). The proven 3-step free flow (higgs_api.py / higgs_batch.py, the Kurukshetra i2v batch): presign via POST /media/batch -> PUT the bytes to the presigned url -> finalize via POST /media/{id}/upload -> wait for the input media's IP/NSFW check to finish before it can seed a render. higgs_batch.py returns the media ONLY once it is finished; this mirrors that contract and additionally reports why a rejection happened. Upload is free (no credits move; the wallet bracket around SUBMIT still guards spend). """ try: with open(path, "rb") as f: data = f.read() except OSError as e: return None, f"could not read start frame {path}: {e}" r1 = self.api("POST", "/media/batch", {"mimetypes": [mime], "source": "user_upload", "surface": "seedance_2", "force_ip_check": True}) item = ((r1.get("body") or [None]) or [None])[0] or {} mid, murl, up = item.get("id"), item.get("url"), item.get("upload_url") if not (mid and murl and up): return None, ( "presign failed (POST /media/batch -> " f"HTTP {r1.get('status')}: {_summarize_body(r1.get('body'))})" ) try: put = self.ctx.request.put(up, data=data, headers={"content-type": mime}) if put.status not in (200, 204): return None, f"S3 PUT of the reference bytes failed (status {put.status})" except Exception as e: return None, f"S3 PUT of the reference bytes raised: {e}" fin = self.api("POST", f"/media/{mid}/upload", {"filename": os.path.basename(path), "force_nsfw_check": True, "force_ip_check": True, "surface": "seedance_2"}) # Finalize is where a moderated reference is rejected up front (typically # HTTP 422 {"detail":{"error_type":"nsfw"|"ip",...}}). Surface it verbatim — # this is the true reason a later job-creation would 404 "Media input not # found"; do NOT proceed to hand a rejected media to the render. if fin.get("status") not in (200, 201): return None, ( "reference image rejected at upload finalize " f"(HTTP {fin.get('status')}: {_summarize_body(fin.get('body'))})" ) # Wait for the IP/NSFW check to FINISH — only a finished (or IP-gated but # confirmable) media is a usable job input (proven higgs_batch contract). # A terminal moderation status ends the wait with the reason. # 48 (~240s) was too short: the free IP/NSFW check routinely runs longer on this # account, and the caller then rendered TEXT-TO-VIDEO with the identity anchor # dropped — producing generic toddlers in a Krishna film. Overridable. for _ in range(int(os.environ.get("HIGGS_VCLAW_MEDIA_READY_TRIES", "180"))): # ~900s self.page.wait_for_timeout(5000) mb = self.api("GET", f"/media/{mid}").get("body") or {} st = mb.get("status") if st in ("nsfw", "failed", "error"): return None, ( f"reference image rejected by moderation after upload " f"(media status={st})" ) if (mb.get("ip_check_finished") is True or st in ("ready", "ip_clear", "ip_detected")): return {"id": mid, "url": murl}, None return None, ( "reference image not ready after 240s (its IP/NSFW check never " "finished); not attached as an i2v start frame" ) def upload_video(self, path, mime="video/mp4"): """Upload one video (e.g. a black-frame voice clip); return (media, reason). This is the flow higgs-cloak never had. Its provision_assets.py named it as the open question — "That exact request shape was never captured (the UI did it)" — and every working voice run there replayed media ids a human had produced by drag-and-drop, which does not scale past a handful of reusable character voices. Captured from the live UI on 2026-08-07; see docs/CAPTURED-VIDEO-UPLOAD.md for the raw exchanges. The reason it stayed unsolved is that video is NOT part of the /media/batch flow at all — it has its own endpoint family. The `surface` everyone was hunting turned out to be "seedance_2", exactly the value images already used. Three differences from upload_image, all captured rather than inferred: * endpoint /video not /media/batch * create body {"mimetype": …} SINGULAR, and no "source" key * response a DICT where /media/batch returns a LIST That last one matters: upload_image does `(body or [None])[0]`, which would raise on a dict. Handled explicitly below rather than shared. PATH NOTE. The capture ran against the web app's host (fnf-api-gw.higgsfield.ai) where the path is "/fnf/video", while this adapter talks to fnf.higgsfield.ai where the same endpoints appear without the "/fnf" prefix (the adapter's own "/media/batch" is "/fnf/media/batch" on the gateway). So "/video" here is a prefix mapping, not a fresh guess — but it is the one part of this flow NOT directly observed on this host, so a 404 is reported with that context instead of a bare failure. """ try: with open(path, "rb") as f: data = f.read() except OSError as e: return None, f"could not read video reference {path}: {e}" r1 = self.api("POST", VIDEO_PATH, {"mimetype": mime, "force_ip_check": True}) body = r1.get("body") # Defensive on shape: captured as a dict, but accept a single-item list so a # server-side change to match the image endpoint does not break us. item = (body[0] if isinstance(body, list) and body else body) or {} if not isinstance(item, dict): item = {} mid, murl, up = item.get("id"), item.get("url"), item.get("upload_url") if not (mid and murl and up): hint = "" if r1.get("status") == 404: hint = ( f" — {VIDEO_PATH!r} 404s on this host. The path was mapped from " "the gateway's '/fnf/video' by dropping the '/fnf' prefix; if " "that mapping is wrong, set HIGGS_VCLAW_VIDEO_PATH." ) return None, ( f"video presign failed (POST {VIDEO_PATH} -> HTTP {r1.get('status')}: " f"{_summarize_body(r1.get('body'))}){hint}" ) try: # The presigned URL signs content-type;host, so this header must be # exactly the mimetype presigned for or the signature check fails. put = self.ctx.request.put(up, data=data, headers={"content-type": mime}) if put.status not in (200, 204): return None, f"S3 PUT of the video bytes failed (status {put.status})" except Exception as e: return None, f"S3 PUT of the video bytes raised: {e}" fin = self.api("POST", f"{VIDEO_PATH}/{mid}/upload", {"filename": os.path.basename(path), "force_nsfw_check": True, "surface": "seedance_2", "force_ip_check": True}) if fin.get("status") not in (200, 201): return None, ( "video reference rejected at upload finalize " f"(HTTP {fin.get('status')}: {_summarize_body(fin.get('body'))})" ) # Finalize returns the server's own probe (width/height/duration/frames). # Use it for the >=640 floor rather than shelling out to ffprobe: it is the # dimension the SERVER measured, so it cannot disagree with the check. meta = fin.get("body") or {} # Coerce rather than isinstance-gate: the check used to require int, so a # server returning 720.0 or "720" would SKIP the floor silently. Same class # as the wallet fail-open — a guard that quietly stops guarding. def _num(v): try: return float(v) except (TypeError, ValueError): return None w, h = _num(meta.get("width")), _num(meta.get("height")) if w is not None and h is not None and (w < MIN_VIDEO_EDGE or h < MIN_VIDEO_EDGE): return None, ( f"video reference is {int(w)}x{int(h)}, below the {MIN_VIDEO_EDGE}px floor " "Higgsfield requires for a usable video media; it would be accepted " "here and then fail opaquely at submit." ) for _ in range(48): # ~240s ceiling, same budget as the image path self.page.wait_for_timeout(5000) mb = self.api("GET", f"{VIDEO_PATH}/{mid}").get("body") or {} st = mb.get("status") if st in ("nsfw", "failed", "error"): return None, ( f"video reference rejected by moderation after upload " f"(media status={st})" ) # "uploaded" must NOT be here. The finalize response returns # {"status": "uploaded", "ip_check_finished": false} — the bytes have # landed but the IP check has not run. Accepting it submits early and # the job is rejected with # HTTP 400 {"error_type":"other","text":"IP check not finished for # input media"} # upload_image never had this bug because its accept list is # ready|ip_clear|ip_detected only. Same contract here. if (mb.get("ip_check_finished") is True or st in ("ready", "ip_clear", "ip_detected")): # Return the FULL media object, not a trimmed {id,url,type}. # The captured submit puts this whole thing in medias[].data — # width/height/duration/frame_rate/startSec and all — so trimming # it here would mean guessing which fields the server needs. Merge # finalize (the probe) under poll (the fresher status), then force # id/url/type, which are the three we depend on downstream. media = dict(meta) media.update(mb if isinstance(mb, dict) else {}) media["id"] = mid media["url"] = murl media.setdefault("type", "video_input") return media, None return None, ( "video reference not ready after 240s (its IP/NSFW check never " "finished); not attached" ) def upload_audio(self, path, mime="audio/mpeg"): """Upload a voice track as an AUDIO media; return (media, reason). The correct channel for a cloned voice. `upload_video` wraps the track in a black-frame mp4 — a workaround inherited from a lane that had no audio field — and Higgsfield's moderation rejects some of those outright with no recourse: the prompt is innocent, the rejection is on the video media. Measured 2026-08-07: a window refused SIX times as a video media was accepted first time as audio. The contract differs from /video and was read off the server's own 422s — `name` and `extension` are REQUIRED here and absent from the video create body. Audio also has no IP check to wait on: finalize returns the usable media directly, typed "audio_input". """ try: with open(path, "rb") as f: data = f.read() except OSError as e: return None, f"could not read voice track {path}: {e}" stem = os.path.splitext(os.path.basename(path))[0] ext = (os.path.splitext(path)[1] or ".mp3").lstrip(".").lower() r1 = self.api("POST", AUDIO_PATH, {"name": stem, "extension": ext, "mimetype": mime, "force_ip_check": True}) body = r1.get("body") item = body if isinstance(body, dict) else {} mid, murl, up = item.get("id"), item.get("url"), item.get("upload_url") if not (mid and murl and up): return None, ( f"audio presign failed (POST {AUDIO_PATH} -> HTTP " f"{r1.get('status')}: {_summarize_body(r1.get('body'))})" ) try: put = self.ctx.request.put(up, data=data, headers={"content-type": mime}) if put.status not in (200, 204): return None, f"S3 PUT of the voice track failed (status {put.status})" except Exception as e: return None, f"S3 PUT of the voice track raised: {e}" fin = self.api("POST", f"{AUDIO_PATH}/{mid}/upload", {"filename": os.path.basename(path), "name": stem, "extension": ext, "force_nsfw_check": True, "surface": "seedance_2", "force_ip_check": True}) if fin.get("status") not in (200, 201): return None, ( "voice track rejected at upload finalize " f"(HTTP {fin.get('status')}: {_summarize_body(fin.get('body'))})" ) media = fin.get("body") or {} if not media.get("id"): return None, f"audio finalize returned no id: {_summarize_body(media)}" media.setdefault("type", "audio_input") return media, None def query_job(self, job_set_id): """Return the first job entry for a job_set, confirming rights if gated. On ip_detected (own-art gate) we fire the cracked confirm-rights PUT once — it un-gates viewing; download works regardless once the url exists. Returns (job_dict, raw_body). """ body = self.api("GET", f"/job-sets/{job_set_id}").get("body") or {} jobs = body.get("jobs") or [] job = jobs[0] if jobs else {} st = job.get("status") if (st == "ip_detected" or job.get("ip_detected")): self.api( "PUT", "/jobs/seedance/confirm-rights/batch", {"job_ids": [job.get("id")]}, ) return job, body def download(self, url, path): os.makedirs(os.path.dirname(path) or ".", exist_ok=True) tmp = path + ".tmp" try: resp = self.ctx.request.get(url) with open(tmp, "wb") as f: f.write(resp.body()) os.replace(tmp, path) return os.path.getsize(path) except Exception: try: os.remove(tmp) except OSError: pass raise def close(self): try: self.ctx.close() except Exception: pass self._release_profile_lock() class _TransportStubSession: """No-browser stub used ONLY when HIGGS_VCLAW_TEST_STUB=1, so the self-test can exercise the real stdin->stdout subprocess transport without launching CloakBrowser or rendering. Returns a flat wallet and a completed job with a fake url; download writes a placeholder file. Never used in production.""" def __init__(self): # Deliberately write noise to stdout to PROVE the stdout-isolation guard # in main() keeps it out of the JSON videoclaw parses (this mimics a # CloakBrowser/Chromium banner). A regression here corrupts stdout. print("[stub] launching fake browser session (this must NOT reach stdout JSON)") def wallet(self): return {"credits_balance": 100, "subscription_balance": 50} def upload_image(self, path, mime="image/png"): return {"id": "stub-media", "url": "https://example.invalid/stub-media.png"}, None def upload_video(self, path, mime="video/mp4"): return {"id": "stub-video", "url": "https://example.invalid/stub-video.mp4", "type": "video_input", "width": 720, "height": 1280, "duration": 8.0}, None class _Page: def wait_for_timeout(self, ms): pass def reload(self, **kw): pass page = _Page() def upload_audio(self, path, mime="audio/mpeg"): return {"id": "stub-audio", "url": "https://example.invalid/stub-audio.mp3", "type": "audio_input"}, None def submit_scene(self, params): return "stub-jobset", None def query_job(self, job_set_id): job = {"id": f"job-{job_set_id}", "status": "completed", "results": {"raw": {"url": f"https://example.invalid/{job_set_id}.mp4"}}} return job, {"jobs": [job]} def download(self, url, path): os.makedirs(os.path.dirname(path) or ".", exist_ok=True) with open(path, "wb") as f: f.write(b"FAKEMP4") return 7 def close(self): pass # Session factory — overridable in tests to inject a stub without CloakBrowser. def make_session(): if os.environ.get("HIGGS_VCLAW_TEST_STUB") == "1": return _TransportStubSession() return HiggsSession() # --------------------------------------------------------------------------- # Actions (each takes the parsed stdin payload, returns the stdout dict). # --------------------------------------------------------------------------- def ensure_session_auth(sess): """Preflight the wallet; on a dead session (status 0 — no Clerk token) try the automatic Chrome cookie re-seed before failing loud. No-op for the offline stub (no `.api`).""" if not hasattr(sess, "api"): return if sess.api("GET", "/workspaces/wallet").get("status") != 0: return print("[higgs-adapter] dead Higgsfield session; attempting Chrome cookie re-seed", file=sys.stderr) if getattr(sess, "reseed_from_chrome", None) and sess.reseed_from_chrome(): return raise RuntimeError( "Higgsfield session not authenticated: the in-page Clerk token " "could not be minted (wallet probe returned status 0), and the " "automatic Chrome cookie re-seed failed. Log into higgsfield.ai in " "Chrome, then run drive_higgsfield.py — and retry. No job was " "submitted; no credits were spent." ) def action_submit(payload, session_factory=make_session): """SUBMIT: submit each task to the free engine and persist job state. Submits-and-returns (does NOT poll to completion). Free-safety: aborts if the wallet's credits_balance drops across the submit batch. """ output_dir = payload["outputDir"] profile = payload.get("executionProfile") or {} tasks = payload.get("tasks") or [] # Bases for resolving RELATIVE referencePaths (project dir first — asset # manifests store project-relative paths — then the workspace root). workspace_root = payload.get("workspaceRoot") or "" project_slug = payload.get("projectSlug") or "" ref_base_dirs = tuple( d for d in ( os.path.join(workspace_root, "projects", project_slug) if workspace_root and project_slug else None, workspace_root or None, ) if d ) external_job_id = f"seedance-{int(time.time() * 1000)}" sess = session_factory() # Scratch dir for relay tail-frames / normalized stills extracted from # referencePaths; cleaned up after the session closes. tmp_dir = tempfile.mkdtemp(prefix="higgs-vclaw-i2v-") try: # Preflight: a dead/expired stealth session mints no Clerk token, so every # API call returns status 0 and each scene's submit would grind through its # full retry budget (~24 min) before failing. Detect it up front, self-heal # via Chrome cookie re-seed, and only then fail fast + loud (raise -> # non-zero exit + stderr, surfaced by videoclaw). ensure_session_auth(sess) wallet_before = sess.wallet() scenes = [] responses = [] issues = [] # Set by either per-scene guard below; stops the loop so the remaining # scenes are never submitted, and is raised AFTER job-state is persisted. abort_reason = None for _task_i, task in enumerate(tasks): if abort_reason: break # Pace the submits. Three back-to-back tripped Cloudflare Turnstile on # 2026-08-07 and failed the remaining 7 scenes of a 10-scene batch that # was otherwise healthy. The wait is skipped for the first scene, so a # single-scene submit is unchanged. if _task_i and SUBMIT_SPACING_S: sess.page.wait_for_timeout(SUBMIT_SPACING_S * 1000) scene_index = task["sceneIndex"] prompt = task.get("prompt") or "" ref_paths = task.get("referencePaths") or [] # Phase-2: resolve + upload a start frame -> image-to-video. Any failure # (no usable ref, ffmpeg/upload failure) degrades to text-to-video with a # non-fatal issue — a reference is never allowed to risk spend or block. medias = [] start_frame = None model = None # Voice clips are pulled out FIRST. They are videos, so the start-frame # resolver would otherwise tail-frame one into a still — which for a # black-frame voice clip is a black rectangle, and silently drops the # voice. See split_references for why referenceRole is the discriminator. ref_paths, voice_paths, split_warning = split_references( ref_paths, task.get("referenceRole")) if split_warning: issues.append(f"scene {scene_index}: {split_warning}") voice_medias = [] for vp in voice_paths: resolved = _resolve_ref_path(vp, ref_base_dirs) if not resolved: issues.append( f"scene {scene_index}: voice reference {vp!r} not found; " "rendered without the voice lock." ) continue # Send the voice on the AUDIO channel, not as a black-frame # video. The video route is a workaround from a lane with no # audio field, and Higgsfield's moderation rejects some of those # outright — six refusals on one window that went through first # time as audio. Extract the track and upload it properly; fall # back to the video route only if extraction is impossible, so a # missing ffmpeg degrades rather than blocks. track = _extract_audio_track(resolved, tmp_dir) if track: vm, v_reason = sess.upload_audio(track) if vm and vm.get("id"): vm.setdefault("type", "audio_input") else: vm, v_reason = sess.upload_video(resolved) if vm and vm.get("id"): voice_medias.append(vm) else: # Never fatal and never a spend risk — but say plainly that the # voice was dropped, because the clip still renders and the # failure is otherwise invisible in the output. issues.append( f"scene {scene_index}: voice NOT applied — " f"{v_reason or 'upload failed'} ({os.path.basename(str(vp))}); " "rendered without the voice lock." ) # THE MOTION DONOR (task.motionReferencePath): a muted clip whose movement the # take copies. It rides as a `video` media BEFORE the voice so it cites # <<>> and the voice keeps <<>>. It is never in # referencePaths (split_references would drop the voice for a second video). # Absent field = byte-identical params. A refused donor is a HARD FAIL: the # render would look valid and prove nothing about the moves. applied_motion = None motion_path = task.get("motionReferencePath") if motion_path: resolved_motion = _resolve_ref_path(motion_path, ref_base_dirs) mm, m_reason = (sess.upload_video(resolved_motion) if resolved_motion else (None, "not found")) if mm and mm.get("id"): mm.setdefault("type", "video_input") voice_medias.insert(0, mm) applied_motion = {"from": os.path.basename(str(motion_path)), "mediaId": mm.get("id"), "slot": "@video1"} elif os.environ.get("HIGGS_VCLAW_ALLOW_NO_MOTION") != "1": raise RuntimeError( f"scene {scene_index}: motion reference NOT applied — {m_reason or 'upload failed'} " f"({os.path.basename(str(motion_path))}) — REFUSING to render a take that would look " "valid and copy no moves. Set HIGGS_VCLAW_ALLOW_NO_MOTION=1 to render without it.") else: issues.append(f"scene {scene_index}: motion reference NOT applied — {m_reason or 'upload failed'}") applied_images = [] if ref_paths: resolved_images = resolve_image_references( ref_paths, tmp_dir, ref_base_dirs, max_images=9) if resolved_images: for image_index, (img, mime, source) in enumerate(resolved_images, 1): up, up_reason = sess.upload_image(img, mime) if up and up.get("id") and up.get("url"): medias.append({"role": "image", "data": {"type": "media_input", "url": up["url"], "id": up["id"]}}) applied = {**source, "slot": f"@image{image_index}", "mediaId": up["id"]} applied_images.append(applied) if image_index == 1: start_frame = applied continue # Surface the REAL reason (moderation rejection / not # ready / upload failure) rather than a bare "upload # failed" — the identity anchor was dropped, so the T2V # fallback will NOT match the reference. # HARD FAIL, do not degrade. A start frame was REQUESTED; without # it the render cannot match the locked characters, so submitting # anyway just burns a slot and produces a clip that looks valid to # every automated check while showing the wrong children. # Set HIGGS_VCLAW_ALLOW_T2V_FALLBACK=1 to restore the old behaviour. msg = (f"scene {scene_index}: @image{image_index} NOT applied — " f"{up_reason or 'upload failed'} ({source.get('from')})") if os.environ.get("HIGGS_VCLAW_ALLOW_T2V_FALLBACK") != "1": raise RuntimeError( msg + " — REFUSING to render with an incomplete positional " "reference contract (set HIGGS_VCLAW_ALLOW_T2V_FALLBACK=1 to allow)") issues.append(msg + "; rendered text-to-video (identity anchor dropped).") if applied_images: model = MODEL_I2V else: msg = (f"scene {scene_index}: {len(ref_paths)} referencePath(s) present " "but none usable as an i2v start frame (missing file / " "unsupported type / no ffmpeg)") if os.environ.get("HIGGS_VCLAW_ALLOW_T2V_FALLBACK") != "1": raise RuntimeError(msg + " — REFUSING to render text-to-video") issues.append(msg + "; rendered text-to-video.") # Merge the voice medias in and cite them. An attached-but-uncited # media is accepted and IGNORED by the server, which presents exactly # like a voice that did not take — so the citation is not optional. if voice_medias: voice_entries, tokens = build_media_entries(videos=voice_medias) medias = list(medias) + voice_entries prompt = cite_in_prompt(prompt, tokens) # INDEX AMBIGUITY, surfaced rather than assumed. We number tokens # PER ROLE (`video_1` = first video). PIPELINE.md reads the index as # a GLOBAL index into params.medias. The two agree only while a # submit carries a single role — true of every submit observed. Here # an i2v start frame sits at global index 1 and the voice at 2, # while the prompt says `video_1`. If the server reads globally the # citation resolves to the wrong media and the voice is silently # ignored — exactly the failure cite_in_prompt exists to close. Say # so; do not guess. One captured mixed submit settles it. # Measured 2026-08-07: a mixed submit (portrait start frame + # voice) scored 0.937 envelope correlation against the supplied # track, so the per-role reading resolves correctly in practice. # That is one data point, so the note stays — but it is a note, not # a warning. Every voice render is mixed (identity + voice), so # phrasing it as a problem would fire on 100% of renders and train # the operator to ignore issues[]. if any(m["role"] != "video" for m in medias) and VERBOSE_MEDIA_NOTE: issues.append( f"scene {scene_index}: note — {len(medias)} medias of mixed " "roles; `<<>>` is numbered per-role while " "PIPELINE.md reads it globally. Measured working " "(0.937 correlation) on this shape; if a voice ever fails " "to take, check this first." ) # A voice clip is a reference, so this is no longer plain t2v; keep # it on the model the reference path is proven on. model = model or MODEL_I2V # A media object in params.medias is not enough: the captured # Higgsfield contract requires the corresponding <<>> / # <<>> / <<>> token in the prompt. Human-authored # artifacts carry portable @imageN slots; compile them here and fail # closed by appending any missing provider token. prompt = compile_media_citations(prompt, citation_tokens_for_medias(medias)) params = build_params(profile, prompt, task.get("durationSeconds"), medias=medias, model=model) jsid, submit_err = sess.submit_scene(params) # SPEND GUARD, per scene. The end-of-run comparison cannot stop # anything — by the time it runs, every scene is submitted. Checking # here bounds the worst case to ONE charged scene instead of all of # them, which is the difference between a mistake and a bill. # # It records and BREAKS rather than raising: the abort path below # persists forensic job-state first, and that guarantee is load-bearing # (test_free_safety_abort asserts it). Raising here would skip it. # `wallet()` is built on api(), which NEVER throws — it returns # {"status": 0, "body": None} after its retries. So a wallet outage, an # endpoint rename, or a response-shape change yields credits_balance # None, and the old `if cb_start is not None` silently SKIPPED the whole # guard: full-speed submits with zero protection and no diagnostic. A # guard that quietly stops guarding is worse than no guard, because the # log looks identical to a healthy run. Refuse instead. cb_start = wallet_before.get("credits_balance") if not isinstance(cb_start, (int, float)) and abort_reason is None: abort_reason = ( "cannot read credits_balance from the wallet " f"(got {cb_start!r}) — refusing to submit blind. The spend " "guard depends on this number; without it a paid submit would " "go unnoticed. Set HIGGS_VCLAW_ALLOW_BLIND_WALLET=1 to override " "(only with use_unlim confirmed working)." ) if os.environ.get("HIGGS_VCLAW_ALLOW_BLIND_WALLET") != "1" else None if isinstance(cb_start, (int, float)) and abort_reason is None: cb_now = (sess.wallet() or {}).get("credits_balance") if isinstance(cb_now, (int, float)) and cb_now < cb_start: abort_reason = ( f"credits_balance dropped {cb_start} -> {cb_now} on scene " f"{scene_index} (model {params.get('model')!r})" ) # ENTITLEMENT SHORT-CIRCUIT. Without an unlimited allowance EVERY scene # fails identically, each burning its full retry budget. One scene is # enough to learn that; the rest is just latency on a known answer. # A refused model is the same class of already-known answer as a missing # entitlement: EVERY scene will refuse identically. Without this the # loop ran all N, action_submit returned NORMALLY with an externalJobId, # and an unattended batch "succeeded" with zero video. if abort_reason is None and submit_err and submit_err.startswith("REFUSED"): abort_reason = ( f"model refused by the free allowlist on scene {scene_index} — " f"{submit_err.split(chr(10))[0][:160]}" ) if abort_reason is None and submit_err and NO_UNLIM_MARKER in submit_err: abort_reason = ( f"the API answered {NO_UNLIM_MARKER!r} on scene {scene_index} — " "this account holds no unlimited allowance" ) if not jsid: # Surface the server's actual rejection reason (moderation gate, # param validation, quota, dead session) — not a bare # "no job_set id returned". This reaches videoclaw via the scene # `error` (re-surfaced on poll) and the rawResult issues[]. issues.append(f"scene {scene_index}: submit failed — {submit_err}") responses.append({"sceneIndex": scene_index, "jobSetId": jsid, "model": params["model"], "startFrame": start_frame, # Machine-verifiable positional receipt. The # unattended worker must prove every requested # visual reference was attached before it treats # the provider submission as valid. "appliedImages": applied_images, "appliedMotion": applied_motion, **({"submitError": submit_err} if submit_err else {})}) scenes.append({ "sceneIndex": scene_index, "prompt": prompt, "taskId": jsid or "", "outputPath": scene_out_path(output_dir, scene_index), "status": "submitted" if jsid else "failed", # ref_paths is POST-split, so on its own the record loses the fact # that a voice was even present — and `referenceApplied` used to # mean "start frame applied" but became true for a voice-only # scene. Both degrade the forensic record the abort path exists to # produce, and forensics are the only way to verify a voice # actually attached after the fact (the clip renders either way). "referencePaths": list(ref_paths) + list(voice_paths), "referenceApplied": bool(start_frame), "voicePaths": list(voice_paths), # the donor rides in voice_medias too — count the AUDIO entries, not the list "voiceApplied": any((m or {}).get("type") == "audio_input" for m in voice_medias), "startFrame": start_frame, "appliedImages": applied_images, "appliedMotion": applied_motion, "model": params["model"], **({"error": f"submit failed: {submit_err}" if submit_err else "submit failed (no job_set id returned)"} if not jsid else {}), }) wallet_after = sess.wallet() finally: sess.close() shutil.rmtree(tmp_dir, ignore_errors=True) # Persist job-state BEFORE the free-safety check so a forensic record exists # even when we abort: wallet snapshots + per-scene job_set ids are captured. cb_before = wallet_before.get("credits_balance") cb_after = wallet_after.get("credits_balance") spend_detected = ( cb_before is not None and cb_after is not None and cb_after < cb_before ) state = { "externalJobId": external_job_id, "routeId": ROUTE_ID, "outputDir": output_dir, "createdAt": _now_iso(), "walletBefore": wallet_before, "walletAfter": wallet_after, "spendDetected": spend_detected, "scenes": scenes, } write_job_state(state) # FREE-SAFETY: credits_balance must NEVER drop. videoclaw does not read # rawResult, so a buried issue would be invisible — instead HARD-ABORT # (raise -> non-zero exit + stderr) so the run fails loudly and visibly. # A per-scene guard tripped: the loop stopped early and the remaining scenes # were never submitted. Raised here, after job-state is on disk, so the # forensic record survives the abort exactly as the end-of-run check does. # # Checked BEFORE the end-of-run comparison because it is strictly more # informative: it names the scene, the cause, and how many were spared. The # check below can only say "a drop happened somewhere in this run". if abort_reason: raise RuntimeError( f"ABORT FREE-SAFETY: {abort_reason}. Stopped after {len(scenes)} of " f"{len(tasks)} scene(s) — the rest were never submitted. Job state " f"saved at {job_state_path(output_dir, external_job_id)}. The queue " "is resumable once the cause is fixed." ) # Fallback: a drop that only shows in the end-to-end comparison (e.g. billed # asynchronously after the last per-scene read). if spend_detected: raise RuntimeError( "ABORT FREE-SAFETY: credits_balance dropped " f"{cb_before} -> {cb_after} during submit (a paid spend occurred; " "the free Seedance path must never charge credits). Job state saved " f"at {job_state_path(output_dir, external_job_id)}." ) return { "externalJobId": external_job_id, "rawResult": { "externalJobId": external_job_id, "submittedScenes": [ {"sceneIndex": s["sceneIndex"], "taskId": s["taskId"]} for s in scenes ], "responses": responses, "walletBefore": wallet_before, "walletAfter": wallet_after, "issues": issues, }, } def action_poll(payload, session_factory=make_session): """POLL: resolve each submitted scene; download completed ones. Long-running resolver — videoclaw calls this on a schedule. Reads the job state written by SUBMIT, queries each job_set once, confirms rights on ip_detected, downloads completed scenes to outputDir/scene-.mp4 (only if not already present), and reports aggregate status. """ output_dir = payload["outputDir"] external_job_id = payload["externalJobId"] state = read_job_state(output_dir, external_job_id) outputs = [] issues = [] raw_results = [] any_pending = False any_failed = False # Re-surface any scene whose referencePaths were present but NOT applied as an # i2v start frame (missing file / unsupported / upload failed) — it rendered # text-to-video. Scenes that DID apply a start frame need no advisory. for scene in state["scenes"]: if scene.get("referencePaths") and not scene.get("referenceApplied"): issues.append( f"scene {scene['sceneIndex']}: referencePaths present but not applied " "as an i2v start frame; rendered text-to-video." ) sess = session_factory() try: # Soft auth preflight: jobs are already submitted, so a dead session must # NOT fail the poll (that would mark live renders failed). Try the Chrome # cookie re-seed; if it fails, continue anyway — queries return status 0, # scenes stay pending, and the next scheduled poll retries. if (hasattr(sess, "api") and sess.api("GET", "/workspaces/wallet").get("status") == 0): print("[higgs-adapter] dead session on poll; attempting Chrome cookie re-seed", file=sys.stderr) if getattr(sess, "reseed_from_chrome", None): sess.reseed_from_chrome() for scene in state["scenes"]: scene_index = scene["sceneIndex"] jsid = scene.get("taskId") if not jsid: scene["status"] = "failed" scene["error"] = scene.get("error") or "no job_set id (submit failed)" issues.append(f"scene {scene_index}: {scene['error']}") any_failed = True continue job, raw = sess.query_job(jsid) raw_results.append(raw) st = job.get("status") u = url_of(job) out_path = scene.get("outputPath") or scene_out_path(output_dir, scene_index) if u: if not os.path.exists(out_path): sess.download(u, out_path) scene["status"] = "completed" scene["outputPath"] = out_path outputs.append({ "id": f"generated-scene-{scene_index}", "kind": "video", "path": out_path, "sceneIndex": scene_index, "backend": BACKEND, }) elif st in FAIL_STATUSES: scene["status"] = "failed" scene["error"] = f"Higgsfield job_set {jsid} ended status={st}." issues.append(scene["error"]) any_failed = True else: # No url yet and not terminal -> still rendering (free queue is slow). if not st: issues.append( f"scene {scene_index}: job_set {jsid} poll had no status " f"(treating as pending): {json.dumps(raw)[:200]}" ) any_pending = True finally: sess.close() write_job_state(state) status = "failed" if any_failed else "pending" if any_pending else "completed" # videoclaw treats `completed` with an empty outputs[] as a FAILURE # (a completed candidate must carry at least one downloaded clip). If we # somehow reach "completed" with nothing to show, report it as failed with a # clear reason instead of letting videoclaw infer the failure. if status == "completed" and not outputs: status = "failed" issues.append( f"job {external_job_id} reported completed but produced no video " "outputs (treating as failed)." ) return { "status": status, "externalJobId": external_job_id, "outputs": outputs, "issues": issues, "rawResult": raw_results, } def action_cancel(payload, session_factory=make_session): """CANCEL: Higgsfield has no cancel — report 'unsupported'. The free Higgsfield queue exposes no reliable cancel endpoint, so we cannot stop an in-flight render. Per the videoclaw contract we return status='unsupported' (a valid VideoExecutionCancelResult status) rather than falsely claiming 'cancelled'. We do NOT open a browser or mutate job state: the job keeps running and remains pollable, which is the truthful outcome. """ external_job_id = payload.get("externalJobId") return { "status": "unsupported", "externalJobId": external_job_id, "issues": [ "Higgsfield free Seedance has no cancel endpoint; the in-flight " "render cannot be stopped and remains pollable." ], "rawResult": None, } def action_wallet(payload, session_factory=make_session): """Read-only authenticated wallet probe for long-running free campaigns. The ordinary submit path brackets each submission. An overnight driver also needs a completion-side reading so asynchronous billing cannot hide between processes. This action never submits media or jobs. """ sess = session_factory() try: ensure_session_auth(sess) return sess.wallet() finally: sess.close() def _now_iso(): return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()) + "Z" # --------------------------------------------------------------------------- # I/O contract — read ONE JSON object from stdin, write ONE from stdout. # --------------------------------------------------------------------------- def dispatch(payload, session_factory=make_session): """Route a parsed stdin payload to the right action by its `action` field. SUBMIT carries no `action` (videoclaw writes the raw VideoExecutionPayload); POLL/CANCEL set action='poll'/'cancel'. Mirrors provider-adapter.ts dispatch. """ action = payload.get("action") if action == "wallet": return action_wallet(payload, session_factory) if action == "poll": return action_poll(payload, session_factory) if action == "cancel": return action_cancel(payload, session_factory) return action_submit(payload, session_factory) def main(argv=None): # CONTRACT: STDOUT must be CLEAN JSON ONLY — videoclaw JSON.parses our stdout # and hard-fails on anything else. CloakBrowser/Chromium can print banners, # warnings, or DevTools noise to the OS-level fd 1, which would corrupt the # JSON. So we (1) dup the REAL stdout fd aside, (2) point both fd 1 and # Python's sys.stdout at stderr for the whole dispatch (any subprocess/library # write to "stdout" now lands on stderr), then (3) write ONLY the result JSON # back to the saved real stdout. CLI args (e.g. `--route seedance-direct`, # which videoclaw's _ADAPTER path does NOT append but the builtin does) are # accepted and ignored — the adapter is driven entirely by the stdin payload. raw = sys.stdin.read() payload = json.loads(raw) if raw.strip() else {} real_stdout_fd = os.dup(1) # save the true stdout os.dup2(2, 1) # redirect fd 1 -> stderr for dispatch saved_py_stdout = sys.stdout sys.stdout = sys.stderr # redirect Python-level prints too try: result = dispatch(payload) finally: sys.stdout = saved_py_stdout os.dup2(real_stdout_fd, 1) # restore the true stdout os.close(real_stdout_fd) sys.stdout.write(json.dumps(result) + "\n") sys.stdout.flush() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except SystemExit: raise except Exception as e: # noqa: BLE001 — adapter must exit non-zero with a message on stderr sys.stderr.write(f"{e}\n") raise SystemExit(1)