"""Send one prompt to the owner's ChatGPT session and return what comes back. Drives the chat web app rather than the API: the owner's subscription already pays for these turns, and image generation is a first-class part of that surface. Nothing here touches metered API billing. """ from __future__ import annotations import base64 import binascii import errno import fcntl import fnmatch import os import random import re import shutil import sys import tempfile import time from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path from urllib.parse import unquote, urlparse from browser import Marionette, MarionetteError, Session import registry STATE = Path.home() / ".overdeck" / "gptbridge" OUT_DIR = STATE / "out" COMPOSER = "#prompt-textarea" SEND = 'button[data-testid="send-button"]' STOP = 'button[data-testid="stop-button"]' UPLOAD = "input#upload-files" # The composer's settings button carries no test id; it is labelled with the model on # one line and the effort in force on the next, so it is matched by that trailing word. # What the effort menu actually offers, keyed by the argument the CLI accepts. EFFORT_LABELS = { "instant": "instant", "medium": "medium", "high": "high", "xhigh": "extra high", "pro": "pro", } EFFORTS = tuple(EFFORT_LABELS) # The composer's settings button is labelled with the effort in force, sometimes below # the model name. Both surfaces use vocabulary from this set, so it locates the button # without depending on a test id the app does not provide. SETTINGS_WORDS = ("instant", "light", "medium", "high", "extra high", "max", "ultra", "pro", "auto", "thinking") SETTINGS_BUTTON = r"(" + "|".join(SETTINGS_WORDS) + r")\s*$" class ChatError(RuntimeError): pass class ProtocolTurnError(ChatError): """Response completed with no `/c/` ever seen — the CLI cannot name the log or the resume command, so this must never be improvised around.""" THINKING_UNHARVESTABLE = "[thinking present but not harvested]" AUTH_SESSION = """ const done = arguments[arguments.length - 1]; fetch('/api/auth/session', {credentials: 'include'}) .then(response => response.ok ? response.json() : null) .then(done) .catch(() => done(null)); """ DOCUMENT_TITLE = "return document.title || '';" CONV_ID_RE = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.I) def conversation_id_from_url(url: str) -> str | None: parsed = urlparse(url or "") if parsed.scheme.lower() != "https" or parsed.netloc.lower() != "chatgpt.com": return None path = parsed.path if not path.startswith("/c/"): return None conversation_id = path.removeprefix("/c/") return conversation_id.lower() if CONV_ID_RE.fullmatch(conversation_id) else None @dataclass class TurnState: """The engine's one accumulating turn state; every TurnEvent is a notification of a change to this. Consumers project it, never reconstruct it independently.""" phase: str = "attaching" conversation_id: str = "" url: str = "" thinking: str = "" answer: str = "" thinking_expand_tried: bool = False uploads: list[str] = field(default_factory=list) artifacts: list[str] = field(default_factory=list) def partial(self) -> dict: return {"phase": self.phase, "thinking": self.thinking, "answer": self.answer, "saved_artifacts": list(self.artifacts)} def _emit(on_event, state: TurnState, event: dict) -> None: if on_event is None: return on_event({**event, "phase": state.phase}) def error_class(exc: Exception) -> str: if isinstance(exc, ProtocolTurnError): return "protocol" if isinstance(exc, MarionetteError): return "marionette" msg = str(exc) if "did not finish in time" in msg or "never accepted the prompt" in msg \ or "never appeared" in msg or "never finished uploading" in msg: return "timeout" return "chat" @dataclass class Reply: text: str images: list[Path] = field(default_factory=list) files: list[Path] = field(default_factory=list) url: str = "" thinking: str = "" conversation_id: str = "" account: str = "" title: str = "" MARK = """ document.querySelectorAll('[data-gptbridge]').forEach( e => e.removeAttribute('data-gptbridge')); const want = arguments[0].toLowerCase(); const scope = arguments[1] || 'button,[role=menuitem],[role=option],[role=menuitemradio]'; let best = null; for (const n of document.querySelectorAll(scope)) { const t = (n.innerText || '').trim().toLowerCase(); if (!t || n.getBoundingClientRect().width < 1) continue; if (t === want || t.startsWith(want) || t.includes(want)) { if (!best || t.length < best.t.length) best = {n: n, t: t}; } } if (!best) return null; best.n.setAttribute('data-gptbridge', '1'); return best.t.slice(0, 80); """ MARK_RE = """ document.querySelectorAll('[data-gptbridge]').forEach( e => e.removeAttribute('data-gptbridge')); const re = new RegExp(arguments[0], 'i'); for (const n of document.querySelectorAll(arguments[1])) { const t = (n.innerText || '').trim(); if (t && re.test(t) && n.getBoundingClientRect().width > 1) { n.setAttribute('data-gptbridge', '1'); return t.slice(0, 80); } } return null; """ MENU_OPTIONS = """ return Array.from(document.querySelectorAll( '[role=menuitem],[role=option],[role=menuitemradio]')) .map(e => (e.innerText || '').trim().split('\\n')[0]) .filter(t => t); """ def _click_marked(m: Marionette) -> bool: return m.js_click("[data-gptbridge]") def click_label(m: Marionette, label: str, scope: str | None = None) -> str | None: found = m.sync_script(MARK, [label, scope]) if not found or not _click_marked(m): return None return str(found) def wait_ready(m: Marionette, timeout: float = 120.0) -> None: """Wait for the composer, sitting out any bot-check interstitial.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: time.sleep(2) try: state = m.sync_script( "return {t: document.title, c: !!document.querySelector(arguments[0])};", [COMPOSER], ) except MarionetteError as exc: if "unload" not in str(exc) and "detach" not in str(exc): raise continue if state.get("c"): return raise ChatError("the chat composer never appeared") RESUME_ID_RE = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.I) def parse_resume_id(value: str) -> str: """A conversation UUID or a full chatgpt.com/c/ URL, either way normalised to the bare lowercase id `-r/--resume` and the daemon's `conversation` field use.""" value = value.strip() if value.startswith("http://") or value.startswith("https://"): found = conversation_id_from_url(value) if not found: raise ChatError(f"not a chatgpt conversation url: {value}") return found if not RESUME_ID_RE.match(value): raise ChatError(f"not a conversation id or chatgpt.com/c/ url: {value}") return value.lower() def authenticated_account(m: Marionette) -> str: """Authenticated account from ChatGPT's session endpoint, never local profile metadata.""" data = m.script(AUTH_SESSION) if not isinstance(data, dict): raise ChatError("could not determine authenticated ChatGPT account") user = data.get("user") email = user.get("email") if isinstance(user, dict) else data.get("email") if not isinstance(email, str) or not email.strip(): raise ChatError("could not determine authenticated ChatGPT account") return email.strip().lower() def _browser_conversation_title( m: Marionette, expected_conversation_id: str, ) -> str | None: if conversation_id_from_url(m.url() or "") != expected_conversation_id.lower(): return None if int(m.sync_script(TURN_COUNT, [TURN, USER]) or 0) < 1: return None title = str(m.sync_script(DOCUMENT_TITLE) or "").strip() title = re.sub(r"\s*[-|]\s*ChatGPT\s*$", "", title, flags=re.I).strip() if title and title.lower() not in {"chatgpt", "new chat", "chat"}: return title return None def conversation_title( m: Marionette, prompt: str, expected_conversation_id: str, fallback: str | None = None, *, attempts: int = 6, interval: float = 0.5, ) -> str: """Return a stable title from the loaded target chat, then the safe fallback.""" candidate = None for _ in range(attempts): time.sleep(interval) title = _browser_conversation_title(m, expected_conversation_id) if title is not None and title == candidate: return title candidate = title if fallback is not None: return fallback compact = " ".join(prompt.split()) return compact[:80].rstrip() or "New conversation" def open_resume(m: Marionette, conversation_id: str, expected_account: str) -> None: """Verify ownership before target navigation, then prove the conversation loaded.""" m.navigate("https://chatgpt.com/") wait_ready(m) current = authenticated_account(m) if current != expected_account.lower(): raise ChatError(f"conversation {conversation_id} expects account {expected_account}; current account {current}") m.navigate(f"https://chatgpt.com/c/{conversation_id}") wait_ready(m) use_chat_surface(m) # The composer renders before the conversation's own turns hydrate, so an # immediate check false-negatives on a real conversation; bounded retry until # a turn appears or the url proves this was a foreign-id redirect. for _ in range(30): if conversation_id_from_url(m.url() or "") != conversation_id.lower(): break turns = int(m.sync_script(TURN_COUNT, [TURN, USER]) or 0) users = int(m.sync_script(USER_COUNT, [USER]) or 0) if turns + users >= 1: return time.sleep(1) raise ChatError(f"conversation {conversation_id} not found in this account") HYDRATE = r""" const turns = Array.from(document.querySelectorAll(arguments[0])); const out = []; let pending = null; for (const t of turns) { const isUser = !!t.querySelector(arguments[1]); const text = (t.innerText || '').replace(/^ChatGPT said:\s*/, '').trim(); if (isUser) { pending = {prompt: text, answer: '', thinking: ''}; out.push(pending); } else if (pending) { pending.answer = text; } } return out; """ def hydrate_dom(m: Marionette) -> list[dict]: """Every user/assistant turn currently visible on the page, oldest first — the web UI's own record, read once on resume to reconcile against the local log.""" data = m.sync_script(HYDRATE, [TURN, USER]) return [item for item in (data or []) if isinstance(item, dict)] # The account also has a Work surface with its own, much smaller quota. Turns must go # to the Chat surface, and the app remembers whichever was used last. SURFACE = """ const hdr = document.querySelector('header') || document.body; const tabs = Array.from(hdr.querySelectorAll('button,[role=tab]')) .filter(b => ['chat', 'work'].includes((b.innerText || '').trim().toLowerCase())); const active = tabs.find(b => b.getAttribute('data-state') === 'on'); const chat = tabs.find(b => (b.innerText || '').trim().toLowerCase() === 'chat'); if (!chat) return {found: false}; if (active === chat) return {found: true, switched: false}; chat.setAttribute('data-gptbridge', '1'); return {found: true, switched: true}; """ def use_chat_surface(m: Marionette) -> bool: """Put the conversation on the Chat surface rather than Work.""" state = m.sync_script(SURFACE) if not state.get("found"): return False if not state.get("switched"): return True _click_marked(m) time.sleep(3) wait_ready(m) return True def set_effort(m: Marionette, effort: str) -> str: """Pick a reasoning effort from the composer's settings menu.""" if effort not in EFFORTS: raise ChatError(f"effort must be one of {EFFORTS}") if not m.sync_script(MARK_RE, [SETTINGS_BUTTON, "form button"]): raise ChatError("no effort control on the composer") if not _click_marked(m): raise ChatError("the composer settings menu would not open") time.sleep(2) if not click_label(m, "effort"): raise ChatError("the settings menu has no effort entry") time.sleep(2) wanted = EFFORT_LABELS[effort] chosen = click_label(m, wanted) if not chosen: offered = m.sync_script(MENU_OPTIONS) raise ChatError( f"effort {effort!r} is not offered here; this surface offers: " + ", ".join(offered) ) time.sleep(1.5) # The menu stays open over the composer and would swallow the send click. m.press() time.sleep(1) return chosen # A capped account states it in a notice beside the composer or in a dialog over it. # The owner's account was not capped when this was written, so there is no class or # test id to pin: the scan is anchored on the composer's own region plus the cap # vocabulary, which returns None on anything it does not recognise rather than # inventing a cap. CAP_WORDS = r"(hit|reached|reach)ed?\s+(your|the)\s+.{0,24}limit|" \ r"you've\s+(hit|reached)|usage\s+limit|message\s+limit|" \ r"limit\s+(reached|resets)|out of\s+(messages|credits)" CAP_SCAN = r""" const words = new RegExp(arguments[1], 'i'); const form = document.querySelector('form'); const roots = [form && form.parentElement, form, document.querySelector('[role=dialog]')].filter(Boolean); for (const root of roots) { for (const n of root.querySelectorAll('div,span,p')) { const t = (n.innerText || '').trim(); if (!t || t.length > 400 || n.children.length > 3) continue; if (words.test(t)) return {text: t}; } } return null; """ # "resets at 3:45 PM", "try again at 15:45" — the wording varies, the clock time does not. RESET_TIME = re.compile(r"\b(\d{1,2}):(\d{2})\s*(am|pm)?\b", re.I) def parse_reset(text: str, *, now: datetime | None = None) -> str | None: """The next occurrence of the clock time a cap notice names, as ISO 8601.""" match = RESET_TIME.search(text) if not match: return None hour, minute = int(match.group(1)), int(match.group(2)) meridiem = (match.group(3) or "").lower() if meridiem == "pm" and hour < 12: hour += 12 elif meridiem == "am" and hour == 12: hour = 0 if not (0 <= hour <= 23 and 0 <= minute <= 59): return None now = now or datetime.now().astimezone() resume = now.replace(hour=hour, minute=minute, second=0, microsecond=0) if resume <= now: resume += timedelta(days=1) return resume.isoformat() def detect_cap(m: Marionette) -> dict | None: """The usage-cap state if the session is capped, else None. Reads only.""" hit = m.sync_script(CAP_SCAN, [COMPOSER, CAP_WORDS]) if not hit: return None text = str(hit.get("text", "")) return {"kind": "usage-cap", "resume_at": parse_reset(text), "text": text} def attach(m: Marionette, paths: list[Path], *, state: TurnState | None = None, on_event=None) -> None: for path in paths: resolved = path.expanduser().resolve() if not resolved.is_file(): raise ChatError(f"attachment not found: {resolved}") if state is not None: state.uploads.append(resolved.name) _emit(on_event, state, {"type": "upload", "file": resolved.name, "state": "queued"}) if not m.send_keys(UPLOAD, str(resolved)): raise ChatError("the chat page exposes no file input") if state is not None: _emit(on_event, state, {"type": "upload", "file": resolved.name, "state": "uploading"}) if paths: wait_uploads(m, state=state, on_event=on_event) def wait_uploads(m: Marionette, timeout: float = 180.0, *, state: TurnState | None = None, on_event=None) -> None: """Uploads must finish before send, or the turn goes without them.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: time.sleep(2) busy = m.sync_script( "return document.querySelectorAll(" "'[data-testid*=attachment] [role=progressbar]," " form [role=progressbar]').length;" ) if not busy: if state is not None: for name in state.uploads: _emit(on_event, state, {"type": "upload", "file": name, "state": "attached"}) return raise ChatError("attachments never finished uploading") TYPE = """ const ed = document.querySelector(arguments[0]); if (!ed) return false; ed.focus(); document.execCommand('selectAll', false, null); document.execCommand('insertText', false, arguments[1]); return (ed.innerText || '').length > 0; """ def type_prompt(m: Marionette, prompt: str) -> None: # Keystroke-by-keystroke entry is far too slow for a multi-line prompt, and the # editor ignores a plain textContent assignment. if not m.sync_script(TYPE, [COMPOSER, prompt]): raise ChatError("could not enter the prompt into the composer") time.sleep(1) # Text replies carry data-message-author-role="assistant", but image replies do not — # they render as a bare conversation turn. Count turns and treat any turn without a # user message inside it as the model's. TURN = '[data-testid^="conversation-turn"]' USER = '[data-message-author-role="user"]' TURN_COUNT = """ let n = 0; for (const t of document.querySelectorAll(arguments[0])) { if (!t.querySelector(arguments[1])) n++; } return n; """ PROGRESS = r""" let n = 0, imgs = 0, text = 0; for (const t of document.querySelectorAll(arguments[0])) { if (t.querySelector(arguments[1])) continue; n++; imgs = t.querySelectorAll('img').length; text = (t.innerText || '').replace(/^ChatGPT said:\s*/, '').trim().length; } return {turns: n, imgs: imgs, text: text, stop: !!document.querySelector(arguments[2])}; """ USER_COUNT = "return document.querySelectorAll(arguments[0]).length;" # Rendered-turn counts are virtualization-unsound: ChatGPT unmounts older turns as # the view scrolls to a new one, so on a long/resumed conversation a live turn count # can drop below its pre-send baseline and never recover. Acceptance is proven # instead by whether the composer has actually been consumed. ACCEPT_STATE = r""" return {sendGone: !document.querySelector(arguments[0]), composerEmpty: !((document.querySelector(arguments[1])||{}).textContent||'').trim()}; """ # Live-only reads: the default (on_event=None) path never runs these, so the # byte-identical stdout golden path never changes JS call shape. LIVE_STATE = r""" const strip = (node) => { const msg = node.querySelector('[data-message-author-role="assistant"]'); return ((msg || node).innerText || '').replace(/^ChatGPT said:\s*/, '').trim(); }; let n = 0, text = ''; for (const t of document.querySelectorAll(arguments[0])) { if (t.querySelector(arguments[1])) continue; n++; text = strip(t); } return {url: location.href, text: text, turns: n}; """ # The reasoning summary is a collapsible region beside the reply, labelled by its # own elapsed-time text ("Thought for 12s", "Reasoning") rather than a stable test # id or class name — MARK-style vocabulary match, never volatile classes. THINKING_WORDS = r"(thought for|thought about|worked for|thinking|reasoned|reasoning)" THINKING = r""" const turns = Array.from(document.querySelectorAll(arguments[0])) .filter(t => !t.querySelector(arguments[1])); const last = turns[turns.length - 1]; if (!last) return {present: false}; const toggle = Array.from(last.querySelectorAll('button,[role=button]')) .find(b => new RegExp(arguments[2], 'i').test((b.innerText || '').trim())); if (!toggle) return {present: false}; if (toggle.getAttribute('aria-expanded') === 'false' && arguments[3] !== '1') { toggle.click(); return {present: true, expanding: true}; } const controls = toggle.getAttribute('aria-controls'); const region = controls ? document.getElementById(controls) : toggle.parentElement; return {present: true, expanding: false, text: region ? (region.innerText || '').trim() : ''}; """ def _poll_live(m: Marionette, state: TurnState, on_event) -> None: """One extra round of reads beyond PROGRESS, only ever taken when a caller wants events: meta (once), then thinking/answer deltas or a full `replace`.""" live = m.sync_script(LIVE_STATE, [TURN, USER]) url = str(live.get("url") or "") if not state.conversation_id: found = conversation_id_from_url(url) if found: state.conversation_id, state.url = found, url state.phase = "named" _emit(on_event, state, {"type": "meta", "conversation_id": found, "url": url}) text = str(live.get("text") or "") if text and text != state.answer: if state.phase in ("named", "streaming"): state.phase = "streaming" if text.startswith(state.answer): delta = text[len(state.answer):] state.answer = text _emit(on_event, state, {"type": "answer", "delta": delta}) else: state.answer = text _emit(on_event, state, {"type": "replace", "thinking": state.thinking, "answer": text}) think = m.sync_script( THINKING, [TURN, USER, THINKING_WORDS, "1" if state.thinking_expand_tried else ""]) if not think.get("present"): return if think.get("expanding"): state.thinking_expand_tried = True return ttext = str(think.get("text") or "") if ttext: if ttext == state.thinking: return if ttext.startswith(state.thinking): delta = ttext[len(state.thinking):] state.thinking = ttext _emit(on_event, state, {"type": "thinking", "delta": delta}) else: state.thinking = ttext _emit(on_event, state, {"type": "replace", "thinking": ttext, "answer": state.answer}) elif state.thinking_expand_tried and not state.thinking: state.thinking = THINKING_UNHARVESTABLE _emit(on_event, state, {"type": "thinking", "delta": THINKING_UNHARVESTABLE}) def _dispatch_gap() -> float: """Return this dispatch window's gap, with a deterministic test override.""" configured = os.environ.get("ASKGPT_DISPATCH_GAP") if configured is not None: try: return float(configured) except ValueError: pass return random.uniform(60, 75) SUBMISSION_TIMEOUT = 120.0 def _submission_timeout() -> ChatError: return ChatError("prompt submission deadline expired") def _rate_limit_dispatch(deadline: float | None = None, turn: "TurnState | None" = None, on_event=None) -> None: """Reserve the next global prompt-dispatch slot within one shared deadline.""" deadline = deadline or time.monotonic() + SUBMISSION_TIMEOUT STATE.mkdir(parents=True, exist_ok=True) stamp = STATE / "dispatch.stamp" lock = STATE / "dispatch.lock" while True: handle = lock.open("w") acquired = False try: try: fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) acquired = True except BlockingIOError: if time.monotonic() >= deadline: raise _submission_timeout() time.sleep(min(0.1, max(0.0, deadline - time.monotonic()))) continue try: last_dispatch = float(stamp.read_text().strip()) except (OSError, ValueError): last_dispatch = None now = time.time() wait_s = 0.0 if last_dispatch is None else last_dispatch + _dispatch_gap() - now if wait_s <= 0: stamp.write_text(f"{now:.9f}\n") return finally: if acquired: fcntl.flock(handle, fcntl.LOCK_UN) handle.close() remaining = max(0.0, deadline - time.monotonic()) print(f"rate-limit: waiting {wait_s:.1f}s before dispatch", file=sys.stderr) event = {"type": "ratelimit", "wait_s": wait_s} if turn is not None: _emit(on_event, turn, event) elif on_event is not None: on_event(event) if wait_s > remaining: raise _submission_timeout() time.sleep(wait_s) def submit(m: Marionette, timeout: float = 120.0, *, deadline: float | None = None, turn: "TurnState | None" = None, on_event=None) -> None: """Click send until the prompt is actually accepted. A click lands on an enabled send button and is still dropped while a large attachment is being processed, and the page looks identical either way. A rendered USER turn count is not proof either: on a long/resumed conversation ChatGPT virtualizes the DOM (older turns unmount as the view scrolls to the new one), so that count can sit below its pre-send baseline for the whole timeout and never prove acceptance. Acceptance is proven instead by the composer being consumed (send button gone, draft cleared), generation starting (the stop button appearing), or the bottom turn's content growing — all virtualization- immune since the bottom turn is never unmounted. """ deadline = deadline or time.monotonic() + min(timeout, SUBMISSION_TIMEOUT) _rate_limit_dispatch(deadline=deadline, turn=turn, on_event=on_event) if turn is not None: turn.phase = "submitting" _emit(on_event, turn, {"type": "submitting"}) text0 = None clicked = False def accepted() -> None: if turn is not None and turn.phase != "submitted": turn.phase = "submitted" _emit(on_event, turn, {"type": "submitted"}) while True: try: if m.click(SEND) or m.js_click(SEND): clicked = True elif not clicked: raise ChatError("the send button never became available") except MarionetteError as exc: # A full-screen overlay (cookie banner, upsell) intercepts the click # instead of reaching the send button; dismiss it and try again rather # than failing the whole turn on a transient overlay. if "intercept" not in str(exc).lower(): raise m.press() time.sleep(1) continue for _ in range(6): time.sleep(1) state = m.sync_script(ACCEPT_STATE, [SEND, COMPOSER]) or {} if state.get("sendGone") and state.get("composerEmpty"): accepted() return poll = m.sync_script(PROGRESS, [TURN, USER, STOP]) or {} text_len = int(poll.get("text", 0) or 0) if text0 is None: text0 = text_len if poll.get("stop") or text_len > text0: accepted() return if time.monotonic() >= deadline: break raise ChatError("the composer never accepted the prompt" f" after {timeout:.0f}s of clicking send{_timeout_shot(m)}") def send(m: Marionette, timeout: float = 900.0, *, submission_deadline: float | None = None, turn: TurnState | None = None, on_event=None) -> None: # One budget covers submission and the reply: how long the app takes to accept a # large attachment is the caller's wait, not a constant unrelated to it. deadline = time.monotonic() + timeout submit(m, timeout=timeout, deadline=submission_deadline or min( deadline, time.monotonic() + SUBMISSION_TIMEOUT), turn=turn, on_event=on_event) if turn is not None: turn.phase = "streaming" # A rendered turn count can shrink mid-turn on a virtualized (long/resumed) # conversation, so completion is read off the stop-button lifecycle and the # bottom turn's content instead — both immune, since the bottom turn is never # unmounted. The stop button also flickers away between streamed chunks, so # one quiet poll is not proof the turn ended. seen_stop = False quiet = 0 last_content: tuple | None = None poll: dict = {} while time.monotonic() < deadline: time.sleep(2.5) poll = m.sync_script(PROGRESS, [TURN, USER, STOP]) if turn is not None: _poll_live(m, turn, on_event) stop = bool(poll.get("stop")) seen_stop = seen_stop or stop # An image turn stays textless, so accept either kind of content. content = (poll.get("imgs", 0), poll.get("text", 0)) has_content = bool(poll.get("imgs") or poll.get("text")) if seen_stop: done = (not stop) and has_content else: # A fast reply can finish before any poll observes the stop button: # content present and unchanged since the last poll, with the composer # back to idle, is completion without ever seeing `stop`. done = False if has_content and content == last_content: state = m.sync_script(ACCEPT_STATE, [SEND, COMPOSER]) or {} done = not state.get("sendGone") last_content = content quiet = quiet + 1 if done else 0 if quiet >= 2: if turn is not None: turn.phase = "response_done" _emit(on_event, turn, {"type": "response_done", "thinking": turn.thinking, "answer": turn.answer}) return raise ChatError(f"the reply did not finish in time; {describe(poll)}" f"{_timeout_shot(m)}") def describe(state: dict) -> str: """What the page looked like on the last poll — a bare timeout names no cause.""" if not state: return "the page was never polled" return (f"reply turns {state.get('turns', 0)}, " f"{state.get('text', 0)} chars, {state.get('imgs', 0)} images, stop button " f"{'still up' if state.get('stop') else 'gone'}") def _timeout_shot(m: Marionette) -> str: try: return f"; page saved to {m.screenshot(STATE / 'timeout.png')}" except (MarionetteError, OSError): return "" DOWNLOAD_BUTTONS = r""" const gptbridgeDownloadKey = value => value.replace(/\s+/g, ' ').trim().toLowerCase(); const gptbridgeDownloadButtons = root => { const byLabel = new Map(); for (const button of root.querySelectorAll('button')) { const text = (button.innerText || '').trim(); if (!/^download\s+.+/i.test(text) && !(text && button.querySelector('p.not-prose.truncate'))) continue; byLabel.set(gptbridgeDownloadKey(text), button); } return Array.from(byLabel.values()); }; """ HARVEST = DOWNLOAD_BUTTONS + r""" const turns = Array.from(document.querySelectorAll(arguments[0])) .filter(t => !t.querySelector(arguments[1])); const last = turns[turns.length - 1]; if (!last) return null; const imgs = []; for (const img of last.querySelectorAll('img')) { const src = img.currentSrc || img.src || ''; const r = img.getBoundingClientRect(); if (src && !src.startsWith('data:') && r.width >= 96 && r.height >= 96) { if (!imgs.includes(src)) imgs.push(src); } } const files = []; for (const link of last.querySelectorAll('a[href]')) { const raw = link.getAttribute('href') || ''; const href = link.href || raw; const downloadable = link.hasAttribute('download') || raw.startsWith('sandbox:') || raw.startsWith('blob:') || href.includes('/backend-api/files/') || href.includes('files.oaiusercontent.com') || href.includes('/mnt/data/'); if (!downloadable || files.some(item => item.src === href)) continue; files.push({ src: href, name: link.getAttribute('download') || link.getAttribute('title') || link.getAttribute('aria-label') || (link.innerText || '').trim(), }); } const downloads = gptbridgeDownloadButtons(last) .map(button => (button.innerText || '').trim()); const msg = last.querySelector('[data-message-author-role="assistant"]'); const text = ((msg || last).innerText || '').replace(/^ChatGPT said:\s*/, '').trim(); return {text: text, images: imgs, files: files, downloads: downloads, url: location.href}; """ CLICK_DOWNLOAD = DOWNLOAD_BUTTONS + r""" const turns = Array.from(document.querySelectorAll(arguments[0])) .filter(t => !t.querySelector(arguments[1])); const last = turns[turns.length - 1]; if (!last) return false; const expected = gptbridgeDownloadKey(arguments[2] || ''); const button = gptbridgeDownloadButtons(last) .find(candidate => gptbridgeDownloadKey(candidate.innerText || '') === expected); if (!button) return false; button.click(); return true; """ CLICK_DRAWER_DOWNLOAD = r""" const expected = (arguments[0] || '').trim().toLowerCase(); const dialogs = Array.from( document.querySelectorAll('[role="dialog"][data-state="open"]') ).filter(dialog => (dialog.getAttribute('aria-label') || '').trim().toLowerCase() === expected); if (!dialogs.length) return 'absent'; if (dialogs.length !== 1) return 'ambiguous-dialog'; const buttons = Array.from(dialogs[0].querySelectorAll('button[aria-label="Download"]')); if (buttons.length !== 1) return 'ambiguous-button'; buttons[0].click(); return 'clicked'; """ # Fetching from the page keeps the session cookies on the request; the asset hosts # reject an unauthenticated download. FETCH = """ const done = arguments[arguments.length - 1]; fetch(arguments[0], {credentials: 'include'}) .then(r => { if (!r.ok) throw new Error('http ' + r.status); const type = r.headers.get('content-type') || ''; const disposition = r.headers.get('content-disposition') || ''; return r.blob().then(blob => ({blob, type, disposition})); }) .then(item => new Promise(res => { const fr = new FileReader(); fr.onload = () => res({ type: item.type || item.blob.type, disposition: item.disposition, data: fr.result, }); fr.readAsDataURL(item.blob); })) .then(done) .catch(e => done({error: String(e)})); """ EXT = { "image/png": ".png", "image/jpeg": ".jpg", "image/webp": ".webp", "image/gif": ".gif", } def slugify(text: str, limit: int = 40) -> str: slug = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") return (slug[:limit].rstrip("-")) or "reply" def _content_disposition_name(value: str) -> str: encoded = re.search(r"filename\*=UTF-8''([^;]+)", value, re.IGNORECASE) if encoded: return unquote(encoded.group(1)) plain = re.search(r'filename="?([^";]+)', value, re.IGNORECASE) return plain.group(1).strip() if plain else "" def _safe_file_name(candidate: str, src: str, disposition: str, index: int) -> str: disposition_name = _content_disposition_name(disposition) path_name = unquote(urlparse(src).path.rsplit("/", 1)[-1]) candidates = ([candidate] if Path(candidate).suffix else []) + [ disposition_name, path_name, candidate] for value in candidates: value = value.strip().splitlines()[0] if value.strip() else "" if not value or value in {".", ".."} or "/" in value or "\\" in value: continue if any(ord(char) < 32 for char in value): continue cleaned = re.sub(r'[<>:"|?*]', "_", value).strip(" .") if cleaned: return cleaned return f"generated-{index}" def _fetch_blob(m: Marionette, src: str, kind: str, index: int) -> dict: blob = m.script(FETCH, [src]) if not isinstance(blob, dict) or "data" not in blob: raise ChatError(f"could not download {kind} {index}: {blob}") return blob def _blob_bytes(blob: dict) -> bytes: _, separator, payload = str(blob["data"]).partition(",") if not separator: raise ChatError("download did not return a data URL") try: return base64.b64decode(payload, validate=True) except (binascii.Error, ValueError) as exc: raise ChatError("download returned invalid base64") from exc def _download_snapshot(m: Marionette) -> dict[Path, tuple[int, int]]: m.download_dir.mkdir(mode=0o700, parents=True, exist_ok=True) return { path: (path.stat().st_mtime_ns, path.stat().st_size) for path in m.download_dir.iterdir() if path.is_file() } def _download_button( m: Marionette, index: int, label: str, out_dir: Path, timeout: float = 300.0, *, turn: TurnState | None = None, on_event=None, ) -> Path: name_hint = re.sub(r"^download\s+", "", label, flags=re.IGNORECASE).strip() or label if turn is not None: _emit(on_event, turn, {"type": "download", "file": name_hint, "state": "started", "path": None}) before = _download_snapshot(m) clicked = m.sync_script(CLICK_DOWNLOAD, [TURN, USER, label]) if not clicked: raise ChatError(f"could not click generated file {index + 1}: {label}") deadline = time.monotonic() + timeout drawer_clicked = False drawer_state = None candidates: dict[Path, tuple[tuple[int, int], float]] = {} while time.monotonic() < deadline: time.sleep(0.5) now = time.monotonic() current = _download_snapshot(m) changed = [] for path, signature in current.items(): if path.suffix == ".part" or before.get(path) == signature: continue previous = candidates.get(path) if signature[1] > 0 and previous and previous[0] == signature \ and now - previous[1] >= 2.0: changed.append(path) elif not previous or previous[0] != signature: candidates[path] = (signature, now) candidates = {path: candidates[path] for path in current if path in candidates} if not changed: if not drawer_clicked: drawer_state = m.sync_script(CLICK_DRAWER_DOWNLOAD, [name_hint]) if drawer_state == "clicked": drawer_clicked = True elif drawer_state not in ( "absent", "ambiguous-dialog", "ambiguous-button", None, ): raise ChatError( f"generated file drawer failed for {label}: {drawer_state}") continue if len(changed) != 1: raise ChatError(f"generated file click produced multiple downloads: {changed}") downloaded = changed[0] name = _safe_file_name(name_hint, downloaded.as_uri(), "", index + 1) destination = out_dir / name if destination.exists(): raise ChatError(f"refusing to overwrite generated file: {destination}") shutil.move(downloaded, destination) if turn is not None: turn.artifacts.append(str(destination)) _emit(on_event, turn, {"type": "download", "file": name_hint, "state": "saved", "path": str(destination)}) return destination detail = (f" ({drawer_state})" if drawer_state not in (None, "absent", "clicked") else "") raise ChatError(f"generated file download timed out: {label}{detail}") # `innerText` returns the RENDERED markdown, not the source: a `# heading` renders as # an

and comes back with the '#' gone, a `* item` list marker disappears into a # bare
  • , `*emphasis*` loses both asterisks. ChatGPT's own per-message Copy button # writes the ORIGINAL markdown to the clipboard, so that is the authoritative text # source for a completed reply — HARVEST's `text` field is only ever the fallback. COPY_BUTTON = r""" document.querySelectorAll('[data-gptbridge]').forEach( e => e.removeAttribute('data-gptbridge')); const turns = Array.from(document.querySelectorAll(arguments[0])) .filter(t => !t.querySelector(arguments[1])); const last = turns[turns.length - 1]; if (!last) return false; const btn = Array.from(last.querySelectorAll('button,[role=button]')).find(b => { const testid = b.getAttribute('data-testid') || ''; const aria = b.getAttribute('aria-label') || ''; return /copy/i.test(testid) || /copy/i.test(aria); }); if (!btn) return false; btn.setAttribute('data-gptbridge', '1'); return true; """ CLIPBOARD_READ = r""" const done = arguments[arguments.length - 1]; const deadline = Date.now() + arguments[0]; (function poll() { navigator.clipboard.readText().then(text => { if (text || Date.now() >= deadline) { done(text || ''); } else { setTimeout(poll, 100); } }).catch(() => done('')); })(); """ def copy_reply_text(m: Marionette, *, timeout_ms: int = 2000) -> str: """The last assistant turn's true source markdown via the Copy button + clipboard. Returns '' on anything short of success (no button, click failed, clipboard empty/denied, older UI) — never raises, so a caller always has the innerText fallback available.""" if not (hasattr(m, "js_click") and hasattr(m, "script")): return "" try: if not m.sync_script(COPY_BUTTON, [TURN, USER]): return "" if not m.js_click("[data-gptbridge]"): return "" text = m.script(CLIPBOARD_READ, [timeout_ms]) except Exception: # Best-effort only: any failure here (missing button, denied permission, # an older UI, a test double that never scripted this call) falls back to # the innerText text harvest() already has — this path must never crash a # turn that otherwise completed successfully. return "" return str(text or "") def harvest(m: Marionette, out_dir: Path, stamp: str, slug: str, *, turn: TurnState | None = None, on_event=None) -> Reply: data = m.sync_script(HARVEST, [TURN, USER]) if not data: raise ChatError("no reply turn appeared in the conversation") text = copy_reply_text(m) if not text: print("gptbridge: copy-button/clipboard text unavailable, falling back to " "rendered innerText (markdown formatting characters will be lost)", file=sys.stderr) text = str(data.get("text", "")) reply = Reply(text=text, url=str(data.get("url", ""))) out_dir.mkdir(parents=True, exist_ok=True) for index, src in enumerate(data.get("images") or [], start=1): file_label = f"image-{index}" if turn is not None: _emit(on_event, turn, {"type": "download", "file": file_label, "state": "started", "path": None}) blob = _fetch_blob(m, str(src), "image", index) suffix = EXT.get(str(blob.get("type", "")).split(";")[0], ".png") name = f"{stamp}-{slug}{'' if index == 1 else f'-{index}'}{suffix}" dest = out_dir / name if dest.exists(): raise ChatError(f"refusing to overwrite generated image: {dest}") dest.write_bytes(_blob_bytes(blob)) reply.images.append(dest) if turn is not None: turn.artifacts.append(str(dest)) _emit(on_event, turn, {"type": "download", "file": file_label, "state": "saved", "path": str(dest)}) for index, item in enumerate(data.get("files") or [], start=1): if not isinstance(item, dict) or not item.get("src"): raise ChatError(f"invalid generated file metadata {index}: {item}") src = str(item["src"]) file_label = str(item.get("name") or f"file-{index}") if turn is not None: _emit(on_event, turn, {"type": "download", "file": file_label, "state": "started", "path": None}) blob = _fetch_blob(m, src, "file", index) name = _safe_file_name( str(item.get("name") or ""), src, str(blob.get("disposition") or ""), index) dest = out_dir / name if dest.exists(): raise ChatError(f"refusing to overwrite generated file: {dest}") dest.write_bytes(_blob_bytes(blob)) reply.files.append(dest) if turn is not None: turn.artifacts.append(str(dest)) _emit(on_event, turn, {"type": "download", "file": file_label, "state": "saved", "path": str(dest)}) for index, label in enumerate(data.get("downloads") or []): reply.files.append( _download_button(m, index, str(label), out_dir, turn=turn, on_event=on_event)) return reply # §4b attachment recovery: the same per-turn image/file/download-button rules as # HARVEST, applied to every assistant turn instead of only the last. HARVEST_ALL = DOWNLOAD_BUTTONS + r""" const turns = Array.from(document.querySelectorAll(arguments[0])) .filter(t => !t.querySelector(arguments[1])); return turns.map(last => { const imgs = []; for (const img of last.querySelectorAll('img')) { const src = img.currentSrc || img.src || ''; const r = img.getBoundingClientRect(); if (src && !src.startsWith('data:') && r.width >= 96 && r.height >= 96) { if (!imgs.includes(src)) imgs.push(src); } } const files = []; for (const link of last.querySelectorAll('a[href]')) { const raw = link.getAttribute('href') || ''; const href = link.href || raw; const downloadable = link.hasAttribute('download') || raw.startsWith('sandbox:') || raw.startsWith('blob:') || href.includes('/backend-api/files/') || href.includes('files.oaiusercontent.com') || href.includes('/mnt/data/'); if (!downloadable || files.some(item => item.src === href)) continue; files.push({ src: href, name: link.getAttribute('download') || link.getAttribute('title') || link.getAttribute('aria-label') || (link.innerText || '').trim(), }); } const downloads = gptbridgeDownloadButtons(last) .map(button => (button.innerText || '').trim()); return {images: imgs, files: files, downloads: downloads}; }); """ @dataclass class DownloadItem: index: int total: int turn: int kind: str label: str source: object def _printable_label(value: object, fallback: str) -> str: text = value if isinstance(value, str) else "" normalized = "".join(char if char.isprintable() else " " for char in text) return " ".join(normalized.split()) or fallback def _unique_name(name: str, used: set[str]) -> str: if name not in used: used.add(name) return name stem, dot, ext = name.rpartition(".") base, suffix = (stem, f".{ext}") if dot else (name, "") n = 2 while f"{base}-{n}{suffix}" in used: n += 1 candidate = f"{base}-{n}{suffix}" used.add(candidate) return candidate def _find_duplicate_content(out_dir: Path, data: bytes) -> Path | None: for path in out_dir.iterdir(): if path.is_file() and path.stat().st_size == len(data) and path.read_bytes() == data: return path return None def _write_blob_atomic(destination: Path, data: bytes) -> None: if destination.exists(): raise ChatError(f"refusing to overwrite generated file: {destination}") fd, temp_name = tempfile.mkstemp( dir=destination.parent, prefix=f".{destination.name}.", suffix=".tmp") temp_path = Path(temp_name) try: with os.fdopen(fd, "wb", buffering=0) as handle: fd = -1 written = handle.write(data) if written != len(data): raise OSError(errno.EIO, f"short write: {written}/{len(data)} bytes") handle.flush() try: os.link(temp_path, destination) except FileExistsError as exc: raise ChatError(f"refusing to overwrite generated file: {destination}") from exc finally: if fd >= 0: os.close(fd) temp_path.unlink(missing_ok=True) def discover_attachments( m: Marionette, ) -> tuple[list[DownloadItem], list[dict], list[dict]]: per_turn = m.sync_script(HARVEST_ALL, [TURN, USER]) if not per_turn: raise ChatError("no reply turns appeared in the conversation") if not isinstance(per_turn, list): raise ChatError("attachment scan returned invalid turn metadata") items: list[DownloadItem] = [] unavailable: list[dict] = [] invalid: list[dict] = [] last_turn_index = len(per_turn) for turn_index, entry in enumerate(per_turn, start=1): if not isinstance(entry, dict): invalid.append({"file": f"turn-{turn_index}", "reason": "invalid metadata", "kind": "turn"}) continue images = entry.get("images") or [] if not isinstance(images, list): invalid.append({"file": f"turn-{turn_index}-images", "reason": "invalid metadata", "kind": "image"}) images = [] for image_index, src in enumerate(images, start=1): label = f"turn-{turn_index}-image-{image_index}" if not isinstance(src, str) or not src: invalid.append({"file": label, "reason": "invalid metadata", "kind": "image"}) continue items.append(DownloadItem(0, 0, turn_index, "image", label, src)) files = entry.get("files") or [] if not isinstance(files, list): invalid.append({"file": f"turn-{turn_index}-files", "reason": "invalid metadata", "kind": "file"}) files = [] for file_index, source in enumerate(files, start=1): fallback = f"turn-{turn_index}-file-{file_index}" src = source.get("src") if isinstance(source, dict) else None if not isinstance(src, str) or not src.strip(): invalid.append({"file": fallback, "reason": "invalid metadata", "kind": "file"}) continue label = _printable_label(source.get("name"), fallback) items.append(DownloadItem(0, 0, turn_index, "file", label, source)) downloads = entry.get("downloads") or [] if not isinstance(downloads, list): invalid.append({"file": f"turn-{turn_index}-downloads", "reason": "invalid metadata", "kind": "button"}) downloads = [] for button_index, raw_label in enumerate(downloads): fallback = f"turn-{turn_index}-button-{button_index + 1}" if not isinstance(raw_label, str): invalid.append({"file": fallback, "reason": "invalid metadata", "kind": "button"}) continue label = _printable_label(raw_label, fallback) if turn_index != last_turn_index: unavailable.append({ "file": label, "kind": "button", "reason": "only latest-turn button downloads are recoverable", }) continue items.append(DownloadItem(0, 0, turn_index, "button", label, button_index)) total = len(items) for index, item in enumerate(items, start=1): item.index = index item.total = total return items, unavailable, invalid def _download_item( m: Marionette, item: DownloadItem, out_dir: Path, used: set[str] ) -> tuple[Path, bool]: if item.kind == "button": path = _download_button(m, int(item.source), item.label, out_dir) used.add(path.name) return path, False if item.kind == "image": src = str(item.source) blob = _fetch_blob(m, src, "image", item.index) data = _blob_bytes(blob) suffix = EXT.get(str(blob.get("type", "")).split(";")[0], ".png") candidate = f"generated-{item.index}{suffix}" else: source = item.source if not isinstance(source, dict): raise ChatError(f"invalid generated file metadata: {source!r}") src = str(source["src"]) blob = _fetch_blob(m, src, "file", item.index) data = _blob_bytes(blob) candidate = _safe_file_name( item.label, src, str(blob.get("disposition") or ""), item.index) duplicate = _find_duplicate_content(out_dir, data) if duplicate is not None: return duplicate, True destination = out_dir / _unique_name(candidate, used) _write_blob_atomic(destination, data) return destination, False def _download_event(item: DownloadItem, state: str, attempt: str, **fields) -> dict: return { "type": "download", "state": state, "attempt": attempt, "index": item.index, "total": item.total, "file": item.label, "kind": item.kind, **fields, } def collect_attachments( m: Marionette, out_dir: Path, *, turn: TurnState | None = None, on_event=None, items: list[DownloadItem] | None = None, attempt: str = "download-1", ) -> tuple[list[Path], list[dict]]: """Download a complete preflight plan sequentially; item failures stay isolated.""" unavailable: list[dict] = [] invalid: list[dict] = [] if items is None: items, unavailable, invalid = discover_attachments(m) if turn is not None: _emit(on_event, turn, { "type": "download_plan", "attempt": attempt, "items": [{"index": item.index, "total": item.total, "turn": item.turn, "kind": item.kind, "file": item.label} for item in items], "unavailable": unavailable, "invalid": invalid, }) out_dir.mkdir(parents=True, exist_ok=True) used = {path.name for path in out_dir.iterdir() if path.is_file()} saved: list[Path] = [] failures: list[dict] = [] for item in items: if turn is not None: _emit(on_event, turn, _download_event( item, "started", attempt, path=None)) try: destination, reused = _download_item(m, item, out_dir, used) except ChatError as exc: failure = {"file": item.label, "error": str(exc)} failures.append(failure) if turn is not None: _emit(on_event, turn, _download_event( item, "failed", attempt, error=str(exc), path=None)) continue except OSError as exc: if turn is not None: _emit(on_event, turn, _download_event( item, "failed", attempt, error=str(exc), path=None)) raise saved.append(destination) if turn is not None: turn.artifacts.append(str(destination)) _emit(on_event, turn, _download_event( item, "saved", attempt, path=str(destination), reused_existing=reused)) return saved, failures def download_attachments_on( m: Marionette, conversation_id: str, *, expected_account: str, out_dir: Path = OUT_DIR, attempt: str = "direct-1", on_event=None, turn: TurnState | None = None, conversation_opened: bool = False, ) -> dict: state = turn or TurnState(conversation_id=conversation_id) state.conversation_id = conversation_id reused = 0 def emit(event: dict) -> None: nonlocal reused if event.get("type") == "download" and event.get("state") == "saved" \ and event.get("reused_existing"): reused += 1 if on_event is not None: on_event(event) try: if state.phase != "opening": state.phase = "opening" _emit(emit, state, {"type": "download_phase", "state": "opening", "attempt": attempt, "conversation_id": conversation_id}) if not conversation_opened: open_resume(m, conversation_id, expected_account) state.url = m.url() _emit(emit, state, {"type": "meta", "attempt": attempt, "conversation_id": conversation_id, "url": state.url}) state.phase = "scanning" _emit(emit, state, {"type": "download_phase", "state": "scanning", "attempt": attempt, "conversation_id": conversation_id}) items, unavailable, invalid = discover_attachments(m) _emit(emit, state, { "type": "download_plan", "attempt": attempt, "items": [{"index": item.index, "total": item.total, "turn": item.turn, "kind": item.kind, "file": item.label} for item in items], "unavailable": unavailable, "invalid": invalid, }) state.phase = "collecting" saved, failures = collect_attachments( m, out_dir, turn=state, on_event=emit, items=items, attempt=attempt) reply = { "text": "", "thinking": "", "images": [], "files": [str(path) for path in saved], "failures": failures, "unavailable": unavailable, "invalid": invalid, "planned": len(items), "reused": reused, "url": state.url, "conversation_id": conversation_id, } state.phase = "done" _emit(emit, state, {"type": "done", "attempt": attempt, "reply": reply}) return reply except Exception as exc: state.phase = "error" _emit(emit, state, {"type": "error", "attempt": attempt, "class": error_class(exc), "detail": str(exc), "partial": state.partial()}) raise def download_attachments( conversation_id: str, *, expected_account: str, out_dir: Path = OUT_DIR, mode: str = "virtual", reseed: bool = False, on_event=None, attempt: str = "direct-1", ) -> tuple[list[Path], list[dict]]: """Recover every artifact from an existing conversation without sending a prompt.""" with Session(mode=mode, reseed=reseed) as m: reply = download_attachments_on( m, conversation_id, expected_account=expected_account, out_dir=out_dir, attempt=attempt, on_event=on_event) return [Path(path) for path in reply["files"]], list(reply["failures"]) # §4c thread listing: the sidebar is the only source of truth for conversations # this tool did not itself create. SIDEBAR_SCAN = r""" const items = Array.from(document.querySelectorAll('a[href^="/c/"]')); const seen = new Set(); const out = []; for (const a of items) { const href = a.getAttribute('href') || ''; const match = href.match(/\/c\/([0-9a-f-]{36})/i); if (!match || seen.has(match[1])) continue; seen.add(match[1]); out.push({id: match[1].toLowerCase(), title: (a.innerText || a.getAttribute('aria-label') || '').trim().split('\n')[0]}); } return out; """ SIDEBAR_SCROLL = r""" let target = null; for (const e of document.querySelectorAll('nav, aside')) { if (e.scrollHeight > e.clientHeight + 50) { target = e; break; } } if (!target) target = document.querySelector('nav') || document.body; target.scrollTop = target.scrollHeight; return target.scrollHeight; """ # On the automation profile the sidebar renders collapsed: zero /c/ links, not zero # conversations. The opener carries no stable test id across surfaces, so it is # matched by vocabulary against both aria-label and data-testid (hyphenated forms # like "open-sidebar-button" included). SIDEBAR_OPEN_WORDS = r"(open[ -]?sidebar|toggle[ -]?sidebar)" SIDEBAR_OPEN = r""" const re = new RegExp(arguments[0], 'i'); for (const b of document.querySelectorAll('button,[role=button]')) { const label = ((b.getAttribute('aria-label') || '') + ' ' + (b.getAttribute('data-testid') || '')).trim(); if (re.test(label)) { b.click(); return true; } } return false; """ def _sidebar_items(m: Marionette) -> list[dict]: return list(m.sync_script(SIDEBAR_SCAN) or []) def _ensure_sidebar_open(m: Marionette) -> list[dict]: """Zero /c/ links before any scan has happened means the sidebar is collapsed, not empty — open it once and rescan, never mid-scroll (a real zero-conversation account is impossible for this owner, so this only ever fires the once). The open animation's streaming render means a single fixed-delay rescan can land mid-render and cache a partial slice; poll instead, until two consecutive reads agree on a non-empty count (settled), or 12 tries run out.""" items = _sidebar_items(m) if items: return items if not m.sync_script(SIDEBAR_OPEN, [SIDEBAR_OPEN_WORDS]): return items previous = None for _ in range(12): time.sleep(1) items = _sidebar_items(m) if items and previous is not None and len(items) == len(previous): return items previous = items return items def scroll_sidebar(m: Marionette, *, limit: int | None = None, max_idle: int = 2) -> list[dict]: """The sidebar lazy-loads; two consecutive scrolls that add no new entries means it is exhausted. `limit` stops early once enough of the (already most-recent- first) list is visible — `--search`/`--list-all` pass `limit=None` to force a full scroll. Zero threads after exhaustion is always a harvest failure on a real account (closed sidebar, rotted selectors) — never legitimate data — so it is never returned; the cache-writing caller must never persist an empty result.""" items = _ensure_sidebar_open(m) idle = 0 while idle < max_idle and (limit is None or len(items) < limit): before = len(items) m.sync_script(SIDEBAR_SCROLL) time.sleep(1) items = _sidebar_items(m) idle = idle + 1 if len(items) <= before else 0 if not items: raise ChatError("the sidebar produced no conversations — sidebar closed or " "selectors rotted") return items[:limit] if limit is not None else items def compile_search(pattern: str) -> re.Pattern: """Case-insensitive regex first; shell wildcards on `re.error`; else an error — pinned one way so `deploy.*fix`, `*seats*`, and a plain substring all just work.""" try: return re.compile(pattern, re.I) except re.error: try: return re.compile(fnmatch.translate(pattern), re.I) except re.error as exc: raise ChatError(f"not a valid pattern or glob: {pattern!r} ({exc})") from exc def list_threads(m: Marionette, limit: int = 10) -> list[dict]: use_chat_surface(m) return scroll_sidebar(m, limit=limit) def search_threads(m: Marionette, pattern: str) -> list[dict]: use_chat_surface(m) regex = compile_search(pattern) return [item for item in scroll_sidebar(m, limit=None) if regex.search(item["title"])] def list_all_threads(m: Marionette) -> list[dict]: use_chat_surface(m) return scroll_sidebar(m, limit=None) def _open_for_listing(mode: str, reseed: bool) -> Session: session = Session(mode=mode, reseed=reseed) m = session.__enter__() m.navigate("https://chatgpt.com/") wait_ready(m) return session, m def list_threads_standalone(*, limit: int = 10, mode: str = "virtual", reseed: bool = False) -> list[dict]: with Session(mode=mode, reseed=reseed) as m: m.navigate("https://chatgpt.com/") wait_ready(m) return list_threads(m, limit=limit) def search_threads_standalone(pattern: str, *, mode: str = "virtual", reseed: bool = False) -> list[dict]: with Session(mode=mode, reseed=reseed) as m: m.navigate("https://chatgpt.com/") wait_ready(m) return search_threads(m, pattern) def list_all_threads_standalone(*, mode: str = "virtual", reseed: bool = False) -> list[dict]: with Session(mode=mode, reseed=reseed) as m: m.navigate("https://chatgpt.com/") wait_ready(m) return list_all_threads(m) def bootstrap_registry_standalone(*, mode: str = "virtual", reseed: bool = False) -> tuple[str, list[dict]]: """The explicit one-time sidebar migration path.""" with Session(mode=mode, reseed=reseed) as m: m.navigate("https://chatgpt.com/") wait_ready(m) account = authenticated_account(m) return account, list_all_threads(m) def ask_on( m: Marionette, prompt: str, *, effort: str | None = None, attachments: list[Path] | None = None, out_dir: Path = OUT_DIR, stamp: str, timeout: float = 900.0, submission_deadline: float | None = None, on_event=None, ) -> Reply: """Run one browser turn and persist its named conversation before completion.""" account = authenticated_account(m) if on_event is not None else "" turn = TurnState() if on_event is not None else None prior_title = None def event(event: dict) -> None: nonlocal prior_title if event.get("type") == "meta": conversation_id = str(event["conversation_id"]) existing = registry.lookup(conversation_id) if existing is None: registry.append(account, conversation_id, slugify(prompt), str(event.get("url") or "")) elif existing.account != account: raise registry.RegistryError( f"conversation {conversation_id} belongs to {existing.account}, not {account}") else: prior_title = existing.title if on_event is not None: on_event(event) callback = event if on_event is not None else None try: if effort: set_effort(m, effort) attach(m, attachments or [], state=turn, on_event=callback) if turn is not None: turn.phase = "submitted" type_prompt(m, prompt) send(m, timeout=timeout, submission_deadline=submission_deadline, turn=turn, on_event=callback) if turn is not None and not turn.conversation_id: raise ProtocolTurnError("the reply finished but no chatgpt.com/c/ url ever appeared") if turn is not None: turn.phase = "collecting" reply = harvest(m, out_dir, stamp, slugify(prompt), turn=turn, on_event=callback) if turn is not None: reply.thinking, reply.conversation_id = turn.thinking, turn.conversation_id reply.account, reply.title = account, conversation_title( m, prompt, reply.conversation_id, fallback=prior_title) if reply.conversation_id: registry.append(account, reply.conversation_id, reply.title, reply.url) except (ChatError, MarionetteError, registry.RegistryError) as exc: if turn is not None: _emit(callback, turn, {"type": "error", "class": error_class(exc), "detail": str(exc), "partial": turn.partial()}) raise if turn is not None: turn.phase = "done" _emit(callback, turn, {"type": "done", "reply": { "text": reply.text, "thinking": reply.thinking, "images": [str(p) for p in reply.images], "files": [str(p) for p in reply.files], "url": reply.url, "conversation_id": reply.conversation_id, "account": reply.account, "title": reply.title}}) return reply def ask( prompt: str, *, effort: str | None = None, attachments: list[Path] | None = None, out_dir: Path = OUT_DIR, stamp: str, mode: str = "virtual", reseed: bool = False, timeout: float = 900.0, resume: str | None = None, expected_account: str | None = None, on_event=None, on_resumed=None, ) -> Reply: with Session(mode=mode, reseed=reseed) as m: if resume: if not expected_account: raise ChatError("resume requires expected account") open_resume(m, resume, expected_account) if on_resumed is not None: on_resumed(m) else: m.navigate("https://chatgpt.com/") wait_ready(m) use_chat_surface(m) return ask_on(m, prompt, effort=effort, attachments=attachments, out_dir=out_dir, stamp=stamp, timeout=timeout, on_event=on_event)