#!/usr/bin/env python3 """ newface.ai image generation (GPT Image 2) driven through your ALREADY-LOGGED-IN Chrome tab over the DevTools Protocol — the same in-page-fetch trick as tryinfer_client.py. Every /api call runs as fetch() INSIDE the real tab, so newface's session cookie rides along automatically. WHY THE BROWSER PATH AND NOT THE OPEN API newface has a documented Open API (api key + /open-api/...), and it WORKS — but it does NOT apply the promo pricing. Measured 2026-09-01: an identical 1k/low GPT Image 2 render billed credits through the Open API while the WEB endpoints report `isPromoFree: true, actualCost: 0`. The promo lives on the web path only, so that is the path we drive. THE FREE LANE IS EXACTLY ONE COMBINATION resolution=1k AND quality=low. Everything else costs (measured: 2k=3, 4k=10, medium=1, high=1 credits). `count` does NOT matter — the promo is per-REQUEST, so 1 and 100 both cost 0. Both knobs are therefore HARD-LOCKED here and re-checked against /api/cost/estimate before every submit: if the API ever stops reporting isPromoFree we refuse rather than silently spend. There is no cheaper failure mode available — nothing reports spend back after the fact. FLOW (all in-page, credentials: 'include') POST /api/cost/estimate {kind,model,resolution,quality,count} -> {actualCost,isPromoFree,promoApplied} POST /api/generations/images {model,prompt,images:[{url,refName,fileName}],ratio,resolution,quality,number} -> {mid, mids[]} — returns IMMEDIATELY, async POST /api/generations/images/batch-poll {mids:[...]} -> {mid: {item_status, progress, unlimit, ...}} References ride as PUBLIC URLs (our R2) — verified the renderer fetches them, so there is NO upload step. Submissions run in PARALLEL: three fired at once all finished together in 47s, queued=false throughout. Output has TWO urls per image: `resWatermarkImageUrl` (watermarked) and `unlimit` (the raw one). We return `unlimit`. Both are presigned and expire in 7 days, so the backend rehosts them. Two run modes: • --emit-json : machine mode for the desktop bridge — NDJSON events on stdout, human logs on stderr. • (default) : standalone test — human logs on stderr, final JSON on stdout. Deps (same Windows Python as the other bridge clients): websocket-client """ import argparse import json import sys import time import urllib.request try: import websocket # websocket-client except ImportError: sys.exit("Missing dep. Install with: pip install websocket-client") for _s in (sys.stdout, sys.stderr): try: _s.reconfigure(encoding="utf-8") except Exception: pass def log(*a): print(*a, file=sys.stderr, flush=True) _JSON_OUT = None def event(**fields): if _JSON_OUT is not None: _JSON_OUT.write(json.dumps(fields) + "\n") _JSON_OUT.flush() def http_json(url): with urllib.request.urlopen(url, timeout=10) as r: return json.loads(r.read().decode()) # ---- the free lane. Changing either of these spends real money; see the module docstring. ---- MODEL_ID = "mkf-img7" # GPT Image 2 FREE_RESOLUTION = "1k" FREE_QUALITY = "low" # newface CLAMPS a single submit to 8 images and says nothing about it: ask for 10 and you silently get 8. # Verified against /api/cost/estimate, which echoes count:8 for any count >= 8. Larger runs are therefore # split across requests — and since the promo is per-REQUEST, each chunk of 8 is free exactly like one of 1. MAX_PER_REQUEST = 8 # newface also fails INDIVIDUAL images with no reason given (item_status "failed", null url), so a batch # routinely comes back short. Those are re-submitted rather than handed back as a hole in the results. # Bounded, because a prompt the model simply refuses would otherwise retry forever. TOP_UP_ROUNDS = 3 TOP_UP_WAIT_S = 15 class NoMatchingTab(RuntimeError): """The debug port answered fine; there is just no tab on the site. A different fix from no-cdp.""" # ---- minimal synchronous CDP client (same shape as tryinfer_client.py) ---- class CDP: def __init__(self, ws_url): self.ws = websocket.create_connection(ws_url, max_size=None, timeout=60) self._id = 0 def call(self, method, params=None, session_id=None, timeout=120): self._id += 1 msg = {"id": self._id, "method": method, "params": params or {}} if session_id: msg["sessionId"] = session_id self.ws.send(json.dumps(msg)) deadline = time.time() + timeout while time.time() < deadline: self.ws.settimeout(max(0.1, deadline - time.time())) try: m = json.loads(self.ws.recv()) except websocket.WebSocketTimeoutException: break if m.get("id") == self._id: if "error" in m: raise RuntimeError(f"CDP error: {m['error']}") return m.get("result", {}) raise TimeoutError(f"timed out waiting for {method}") def close(self): try: self.ws.close() except Exception: pass def _post_expr(path, body): """POST as fetch() INSIDE the page so the live session cookie is used (never reconstructed here). The session JWT rides in an HttpOnly cookie that `credentials:'include'` sends for us, but newface also runs the double-submit CSRF pattern: the request is rejected with 401 ("not signed in", which reads like a logged-out browser and is not) unless the readable `csrf_access_token` cookie is ECHOED as x-csrf-token, alongside x-organization-oid from `organization_id`. Both are read in-page at call time, so nothing sensitive is ever copied out of the browser and a token refresh can't leave us holding a stale one. """ return ( "(async()=>{try{" "const ck=(n)=>{const m=document.cookie.match('(^|;)\\\\s*'+n+'\\\\s*=\\\\s*([^;]+)');" "return m?decodeURIComponent(m[2]):'';};" "const h={'content-type':'application/json'};" "const c=ck('csrf_access_token'); if(c) h['x-csrf-token']=c;" "const o=ck('organization_id'); if(o) h['x-organization-oid']=o;" f"const r=await fetch({json.dumps(path)},{{method:'POST',credentials:'include',headers:h," f"body:{json.dumps(json.dumps(body))}}});" "const t=await r.text();return JSON.stringify({status:r.status,body:t});" "}catch(e){return JSON.stringify({status:0,error:String((e&&e.message)||e)});}})()" ) class BrowserSession: def __init__(self, cdp_http, match): self.cdp_http = cdp_http self.match = match ver = http_json(f"{cdp_http}/json/version") log(f"Connected to {ver.get('Browser')}") self.browser_ws = ver["webSocketDebuggerUrl"] self.cdp = None self.session_id = None self._attach() def _attach(self): tabs = [t for t in http_json(f"{self.cdp_http}/json/list") if t.get("type") == "page" and self.match in (t.get("url") or "")] if not tabs: raise NoMatchingTab( f"no open tab matches '{self.match}' — open newface.ai in the debug Chrome and sign in") if self.cdp: self.cdp.close() self.cdp = CDP(self.browser_ws) self.session_id = self.cdp.call( "Target.attachToTarget", {"targetId": tabs[0]["targetId" if "targetId" in tabs[0] else "id"], "flatten": True})["sessionId"] log(f"Attached to {tabs[0].get('url')}") def refresh_login(self, wait_s=30): """Make the APP mint a fresh access token, by reloading its own page. newface's access token lives in a SHORT-LIVED HttpOnly cookie beside a 30-day refresh_token_cookie. Leave a tab idle and the access cookie is simply gone: every call then answers 401 "Not authenticated" even though the browser is still perfectly signed in — which is why the old error told people to sign in to a tab they were already signed into. The app refreshes lazily on load, so a reload is enough, and it means we never have to guess at (or hardcode) a refresh endpoint. """ log(" access token expired — reloading the tab to let newface refresh it…") self.cdp.call("Page.enable", {}, session_id=self.session_id) self.cdp.call("Page.reload", {}, session_id=self.session_id) deadline = time.time() + wait_s while time.time() < deadline: time.sleep(1.5) try: ck = self.cdp.call("Network.getCookies", {"urls": ["https://newface.ai/"]}, session_id=self.session_id) except Exception: continue if any(c.get("name") == "access_token_cookie" for c in ck.get("cookies", [])): log(" signed in again") return True return False def post(self, path, body, retries=3): last = None refreshed = False for attempt in range(retries): try: res = self.cdp.call("Runtime.evaluate", {"expression": _post_expr(path, body), "awaitPromise": True, "returnByValue": True}, session_id=self.session_id) except Exception as e: last = f"CDP evaluate error: {e}" log(f" tab unresponsive — re-attaching… ({e})") try: self._attach() except Exception as e2: last = f"re-attach failed: {e2}" time.sleep(2); continue val = (res.get("result") or {}).get("value") if val is None: last = f"evaluate returned nothing: {res.get('exceptionDetails')}" time.sleep(2); continue d = json.loads(val) if d.get("status") == 0: last = f"in-page fetch error: {d.get('error')}" time.sleep(2); continue if d.get("status") in (401, 403): # Almost always an EXPIRED access token on a still-signed-in browser, not a logged-out one. # Refresh once, then retry; only a genuinely signed-out tab reaches the error below. if not refreshed: refreshed = True if self.refresh_login(): continue raise RuntimeError("newface says not signed in (HTTP %s) — open newface.ai in that Chrome " "and sign in (a reload did not restore the session)" % d.get("status")) try: parsed = json.loads(d.get("body") or "{}") except Exception: raise RuntimeError(f"newface returned non-JSON (HTTP {d.get('status')}): {(d.get('body') or '')[:200]}") if d.get("status") >= 400: raise RuntimeError(f"newface {path} failed (HTTP {d['status']}): {json.dumps(parsed)[:300]}") return parsed raise RuntimeError(f"newface {path} failed after {retries} tries: {last}") def close(self): if self.cdp: self.cdp.close() def unwrap(reply): """Web replies are either {data:{content:…}} or {content:…}; return the content.""" d = reply.get("data", reply) return d.get("content", d) def assert_free(session, count): """Refuse to submit anything the API does not itself call free. This is the ONLY guard: newface reports no spend afterwards, so a wrong answer here is invisible until the dashboard.""" c = unwrap(session.post("/api/cost/estimate", { "kind": "image", "model": MODEL_ID, "resolution": FREE_RESOLUTION, "quality": FREE_QUALITY, "count": count})) if not c.get("isPromoFree") or (c.get("actualCost") or 0) > 0: raise RuntimeError( f"refusing to submit: newface priced this at {c.get('actualCost')} credits " f"(isPromoFree={c.get('isPromoFree')}). The free lane is {FREE_RESOLUTION}/{FREE_QUALITY} — " f"if that changed, the promo is gone and this needs a decision, not an automatic charge.") log(f"cost check: free (promo {c.get('promoApplied')}, count={count})") return c def submit(session, prompt, image_urls, ratio, count): images = [{"url": u, "refName": f"image{i+1}", "fileName": f"{i+1}.png"} for i, u in enumerate(image_urls)] body = {"model": MODEL_ID, "prompt": prompt, "images": images, "ratio": ratio, "resolution": FREE_RESOLUTION, "quality": FREE_QUALITY, "number": int(count)} c = unwrap(session.post("/api/generations/images", body)) mids = c.get("mids") or ([c["mid"]] if c.get("mid") else []) if not mids: raise RuntimeError(f"submit returned no mid: {json.dumps(c)[:300]}") return mids def submit_many(session, prompt, image_urls, ratio, count): """Submit `count` images across as many requests as newface's per-request ceiling needs.""" mids = [] remaining = int(count) while remaining > 0: n = min(remaining, MAX_PER_REQUEST) assert_free(session, n) # every chunk is priced before it is sent — never once for the lot got = submit(session, prompt, image_urls, ratio, n) if not got: break mids.extend(got) remaining -= len(got) # newface can return FEWER mids than asked even under the ceiling; trust what it gave us, and # stop rather than spinning if a request produced nothing usable. if len(got) < n and len(got) == 0: break return mids TERMINAL_OK = ("success", "succeeded", "completed") TERMINAL_BAD = ("failed", "error", "cancelled", "canceled") def poll(session, mids, timeout=1800, interval=5, prefix=()): """Poll every mid in ONE request until all are terminal. Returns {mid: item}. `prefix` is what earlier rounds already produced. It is prepended to every streamed progress event so the consumer sees ONE append-only list: the backend caches rehosted copies by index, and a list that restarted each round would make index N mean a different image. """ end = time.time() + timeout done, last_key = {}, None while time.time() < end: content = unwrap(session.post("/api/generations/images/batch-poll", {"mids": mids})) pend = 0 for m in mids: it = content.get(m) or {} st = (it.get("item_status") or it.get("status") or "").lower() if st in TERMINAL_OK or st in TERMINAL_BAD: done[m] = it else: pend += 1 pct = int(sum(int((content.get(m) or {}).get("progress") or 0) for m in mids) / max(1, len(mids))) # The dedupe key MUST include elapsed time. The hub gives a render 30 minutes and refreshes that # window on every progress event, so a key of only "done:pct" stops emitting the moment a batch # plateaus — and a slow batch sitting at 0/15 · 0% therefore starved the heartbeat and was killed as # "newface render timed out" while it was still legitimately rendering. creaa_client already carries # a per-minute component for exactly this reason; this is the same bug, one provider later. key = f"{len(done)}:{pct}:{int(time.time() - (end - timeout)) // 60}" if key != last_key: last_key = key # Ship the urls of everything finished SO FAR, not just a count. A batch finishes at wildly # different times, and the canvas can show image 1 while image 5 is still rendering. `done` is # append-only across polls, so the list only ever grows and index N keeps meaning the same image. partial, _ = urls_from(done) event(event="progress", phase=f"generating {len(prefix) + len(done)}/{len(prefix) + len(mids)} · {pct}%", done=len(done), total=len(mids), percent=pct, mids=mids, urls=[*prefix, *partial]) log(f" {len(done)}/{len(mids)} done · {pct}%") if not pend: return done time.sleep(interval) raise TimeoutError(f"newface: {len(done)}/{len(mids)} finished within {timeout}s") def urls_from(items): """Prefer `unlimit` (the un-watermarked _raw file) over the watermarked preview.""" out, errs = [], [] for mid, it in items.items(): st = (it.get("item_status") or it.get("status") or "").lower() if st in TERMINAL_BAD: errs.append(f"{mid}: {it.get('error') or it.get('failReason') or st}") continue u = it.get("unlimit") or it.get("resWatermarkImageUrl") if u: out.append(u) else: errs.append(f"{mid}: finished with no url") return out, errs def main(): ap = argparse.ArgumentParser() ap.add_argument("--prompt", default="") ap.add_argument("--image-url", action="append", default=[], help="PUBLIC reference url; repeat for more") ap.add_argument("--ratio", default="16:9", help="16:9 | 9:16 | 1:1 | 4:3 | 3:4 | …") ap.add_argument("--count", type=int, default=1, help="images per run (free at any count)") ap.add_argument("--match", default="newface", help="substring of the logged-in tab's URL") ap.add_argument("--cdp", default=None, help="debug Chrome endpoint, e.g. http://localhost:9222") ap.add_argument("--host", default="localhost") ap.add_argument("--port", type=int, default=9222) ap.add_argument("--resume-mid", action="append", default=[], help="skip submit; poll these mids") ap.add_argument("--wallet", action="store_true", help="print the credit wallet and exit") ap.add_argument("--emit-json", action="store_true") args = ap.parse_args() global _JSON_OUT if args.emit_json: _JSON_OUT = sys.stdout sys.stdout = sys.stderr # Every endpoint is built as f"{cdp_http}/json/...", so a trailing slash — which anyone pasting a URL # naturally includes — would produce "…:9222//json/version" and an opaque rejection. cdp_http = (args.cdp or f"http://{args.host}:{args.port}").rstrip("/") try: session = BrowserSession(cdp_http, args.match) except Exception as e: # Reporting a dead endpoint as "no-tab" sends people to open a tab they already have open. The # usual cause is the HOST SPELLING, not the port: Chrome's DevTools server answers `localhost` and # drops the connection for `127.0.0.1` (surfacing as "Remote end closed connection without # response") even while it listens on 0.0.0.0. The bridge probes for the spelling that works; # name the endpoint here so a direct invocation is diagnosable too. if isinstance(e, NoMatchingTab): reason, hint = "no-tab", "the debug port is working; only the tab is missing." else: reason = "no-cdp" hint = (f"could not reach a debug Chrome at {cdp_http}. Try http://localhost:{args.port} " f"instead of a numeric host — Chrome rejects 127.0.0.1 on the DevTools endpoint.") event(event="error", reason=reason, detail=f"{e} — {hint}", cdp=cdp_http, fatal=True) log(f"ERROR: {e}\n{hint}") return 2 try: if args.wallet: w = unwrap(session.post("/api/organization-wallet", {})) print(json.dumps(w.get("wallet", w)), file=sys.stderr if args.emit_json else sys.stdout) event(event="done", wallet=w.get("wallet", w)) return 0 want = max(1, int(args.count)) urls, all_mids, errs = [], [], [] if args.resume_mid: # resume attaches to a run newface already accepted — take it exactly as it is, never top it up all_mids = list(args.resume_mid) log(f"resuming {len(all_mids)} mid(s)") event(event="progress", phase="resuming", mids=all_mids) urls, errs = urls_from(poll(session, all_mids)) else: # Round 0 is the real request; later rounds only replace what newface FAILED. Each round is # priced and submitted like any other run, so a top-up is free for exactly the same reason the # first round is. `urls` carries across rounds so the streamed list only ever grows. for rnd in range(TOP_UP_ROUNDS + 1): need = want - len(urls) if need <= 0: break if rnd: log(f"topping up {need} failed image(s) — round {rnd} of {TOP_UP_ROUNDS}") event(event="progress", phase=f"retrying {need} failed image(s) in {TOP_UP_WAIT_S}s", urls=urls) time.sleep(TOP_UP_WAIT_S) event(event="progress", phase="submitting" if not rnd else f"resubmitting {need}", urls=urls) mids = submit_many(session, args.prompt, args.image_url, args.ratio, need) if not mids: break all_mids.extend(mids) log(f"submitted {len(mids)} mid(s)" + (f" (round {rnd})" if rnd else "")) # emit IMMEDIATELY so a restart can resume by mid instead of re-generating event(event="progress", phase="submitted", mids=all_mids, urls=urls) got, errs = urls_from(poll(session, mids, prefix=urls)) urls = [*urls, *got] if not errs: break # nothing failed — no reason for another round if not urls: raise RuntimeError("newface produced no images" + (f" — {'; '.join(errs)}" if errs else "")) # Only a SHORTFALL is worth reporting now: a failure that a later round replaced is not something # the user has to know about, but "you asked for 10 and 8 came back" very much is. short = max(0, want - len(urls)) failed = errs[:short] if short else [] if short: log(f"WARNING: {len(urls)} of {want} produced after {TOP_UP_ROUNDS} top-up round(s)") event(event="done", urls=urls, mids=all_mids, failed=failed, requested=want) log(f"done → {len(urls)} image(s)") if not args.emit_json: print(json.dumps({"urls": urls, "mids": all_mids, "failed": failed}, indent=2)) return 0 except Exception as e: event(event="error", reason="newface", detail=str(e), fatal=True) log(f"ERROR: {e}") return 1 finally: try: session.close() except Exception: pass if __name__ == "__main__": sys.exit(main())