diff --git a/src/chatgpt_web2api/api_server.py b/src/chatgpt_web2api/api_server.py index c0ce8894..391e86ce 100644 --- a/src/chatgpt_web2api/api_server.py +++ b/src/chatgpt_web2api/api_server.py @@ -253,6 +253,17 @@ async def _handle_chat(self, request: web.Request) -> web.Response: model = body.get("model", self._config.chatgpt.default_model) stream = body.get("stream", False) + apps = body.get("apps") + if apps is not None and ( + not isinstance(apps, list) + or any(not isinstance(name, str) or not name.strip() for name in apps) + ): + return web.json_response( + {"error": {"message": "apps must be a list of non-empty app names", "type": "invalid_request_error"}}, + status=400, + ) + if apps is not None: + apps = [name.strip() for name in apps] project_id = ( body.get("project_id") or body.get("gizmo_id") @@ -388,9 +399,13 @@ async def _handle_chat(self, request: web.Request) -> web.Response: self._last_project_id = project_id if stream: - return await self._stream_response(request, model_slug, full_text, timeout) + return await self._stream_response( + request, model_slug, full_text, timeout, **({"apps": apps} if apps else {}), + ) else: - return await self._full_response(request, model_slug, full_text, timeout) + return await self._full_response( + request, model_slug, full_text, timeout, **({"apps": apps} if apps else {}), + ) except Exception as e: logger.error("Chat error: %s", e, exc_info=True) @@ -521,7 +536,8 @@ def _error_response(self, exc: Exception) -> web.Response: # ── Response formatters ─────────────────────────────────── async def _full_response( - self, request: web.Request, model: str, text: str, timeout: float + self, request: web.Request, model: str, text: str, timeout: float, + apps: list[str] | None = None, ) -> web.Response: """Non-streaming: collect all chunks, return one JSON. @@ -539,6 +555,7 @@ async def _send_and_collect() -> str: collected = "" async for chunk in self._driver.send_and_stream( text, timeout=timeout, budgets=budgets, model=model, + **({"apps": apps} if apps else {}), ): collected += chunk.delta return collected @@ -568,7 +585,8 @@ async def _send_and_collect() -> str: ) async def _stream_response( - self, request: web.Request, model: str, text: str, timeout: float + self, request: web.Request, model: str, text: str, timeout: float, + apps: list[str] | None = None, ) -> web.Response: """Streaming: SSE chunks as they arrive. @@ -650,6 +668,7 @@ async def _preflight() -> None: try: async for chunk in self._driver.send_and_stream( text, timeout=timeout, budgets=budgets, model=model, + **({"apps": apps} if apps else {}), ): if chunk.delta: await self._send_sse( diff --git a/src/chatgpt_web2api/backend_client.py b/src/chatgpt_web2api/backend_client.py index 4ea582bb..3b0876e3 100644 --- a/src/chatgpt_web2api/backend_client.py +++ b/src/chatgpt_web2api/backend_client.py @@ -232,7 +232,9 @@ async def _conversation_id_from_url(self) -> str: return "" if not url or "/c/" not in url: return "" - return url.split("/c/")[1].split("/")[0].split("?")[0] + conversation_id = url.split("/c/")[1].split("/")[0].split("?")[0] + # New chats can briefly expose a frontend-only id before persistence. + return "" if conversation_id.startswith("WEB:") else conversation_id async def _get_live_conversation_id_best_effort(self) -> str: """Resolve the in-flight conversation id by cheapest available source. diff --git a/src/chatgpt_web2api/cdp_driver.py b/src/chatgpt_web2api/cdp_driver.py index 3bc3eae4..ac3cdd97 100644 --- a/src/chatgpt_web2api/cdp_driver.py +++ b/src/chatgpt_web2api/cdp_driver.py @@ -55,6 +55,7 @@ class StreamChunk: # (is_rate_limited_text is imported from cdp_driver by api_server, chatgpt_dom, # and tests). _RATE_LIMIT_PHRASES stays private to completion_detector. from .completion_detector import ( # noqa: E402,F401 + ASSISTANT_ROOT_SELECTOR, PHASE_STALL_SECONDS, is_rate_limited_text, ) @@ -72,6 +73,13 @@ class StreamChunk: # matches ``navigate_new_chat``. _CONNECT_READY_TIMEOUT = 10 +# Shared by the pre-send baseline and acknowledgment probe. querySelectorAll +# deduplicates a node even when it matches multiple selectors in this list. +_USER_MESSAGE_SELECTOR = ( + '[data-message-author-role="user"], ' + '[data-markdown-text-tone="user-message"], .rich-text-user-turn' +) + # ChatGPT composer / send-button selectors. # # Canonical home moved to chatgpt_dom.py in Phase 5 PR3; re-exported here for @@ -1458,6 +1466,10 @@ async def type_message(self, text: str) -> None: """ await self._dom.type_message(text) + async def type_message_with_apps(self, text: str, apps: list[str]) -> None: + """Select app mentions and append text without clearing their chips.""" + await self._dom.type_message_with_apps(text, apps) + async def _detect_select_all_modifier(self) -> int: """Return the CDP modifiers value for select-all on the live platform. @@ -1499,12 +1511,12 @@ async def _read_assistant_count_baseline(self) -> int: selector = ( "document.querySelectorAll(" - "'[data-message-author-role=\"assistant\"]'" + f"'{ASSISTANT_ROOT_SELECTOR}'" ").length" ) user_selector = ( "document.querySelectorAll(" - "'[data-message-author-role=\"user\"]'" + f"'{_USER_MESSAGE_SELECTOR}'" ").length" ) max_attempts = 3 @@ -1610,7 +1622,7 @@ async def _verify_send_acknowledged(self) -> bool | None: result = await self._js_strict( "(function() {" " var userMsgs = document.querySelectorAll(" - " '[data-message-author-role=\"user\"]').length;" + f" '{_USER_MESSAGE_SELECTOR}').length;" f" var composer = document.querySelector('{COMPOSER_SELECTOR}')" f" || document.querySelector('{COMPOSER_FALLBACK_SELECTOR}');" " var composerPresent = !!composer;" @@ -1724,6 +1736,7 @@ async def send_and_stream( *, budgets=None, model: str | None = None, + apps: list[str] | None = None, ) -> AsyncIterator[StreamChunk]: """Send a message and yield streaming response chunks. @@ -1742,7 +1755,7 @@ async def send_and_stream( terminal path — success, timeout, exception, cancellation). """ from .identity_listener import hash_sent_text - from .turn_anchor import TurnReconciliationError + from .turn_anchor import TurnReconciliationError, normalize_text # PR4 belt-and-suspenders: refuse to mutate the DOM in parallel mode. self._assert_owned_tab_required() @@ -1757,17 +1770,21 @@ async def send_and_stream( # A2 Step 3+4: arm capture scope + build fallback anchor. # The fallback anchor captures pre-send state (backend node-ids/times # or wall-clock) for dual-anchor correlation if UUID capture fails. + # Correlate logical text; the listener/matcher decode observed serialization. fallback_anchor = await self._capture_pre_send_fallback_anchor(text) if self._identity_listener is not None and self._identity_listener.is_alive(): capture_scope = self._identity_listener.arm_capture_scope( - expected_text_hash=hash_sent_text(text), + expected_text_hash=hash_sent_text(normalize_text(text)), conversation_id=self._current_conv_id, target_id=self._target_id, ) try: # Type and send. - await self.type_message(text) + if apps: + await self.type_message_with_apps(text, apps) + else: + await self.type_message(text) await self.click_send() # A2 Step 6: wait for the IdentityListener to capture the UUID. @@ -1826,13 +1843,8 @@ async def send_and_stream( # Wait for URL to become /c/{id} conv_id = "" for _ in range(30): - try: - url = await self._js_strict("window.location.href") - except CDPJSError: - await asyncio.sleep(0.5) - continue - if "/c/" in url: - conv_id = url.split("/c/")[1].split("/")[0].split("?")[0] + conv_id = await self._conversation_id_from_url() + if conv_id: break await asyncio.sleep(0.5) @@ -1852,10 +1864,14 @@ async def send_and_stream( last_status = result.status last_diagnostic = result.diagnostic or {} if result.status == "matched" and result.text: - if len(result.text) > len(last_dom_text): - yield StreamChunk(delta=result.text[len(last_dom_text):]) - last_dom_text = result.text - break + if result.text.startswith(last_dom_text): + if result.text != last_dom_text: + yield StreamChunk(delta=result.text[len(last_dom_text):]) + last_dom_text = result.text + break + # A rewritten/lagging backend text cannot extend the + # emitted prefix. Wait for reconciliation, never splice. + last_status = "text_mismatch" if result.status == "non_text": # P2.5 RCA fix: non_text is NOT terminal here. The backend # propagates intermediary nodes (reasoning_recap, thoughts, diff --git a/src/chatgpt_web2api/chatgpt_dom.py b/src/chatgpt_web2api/chatgpt_dom.py index fc76f168..3071a227 100644 --- a/src/chatgpt_web2api/chatgpt_dom.py +++ b/src/chatgpt_web2api/chatgpt_dom.py @@ -51,6 +51,7 @@ import asyncio import json import logging +import re import time import unicodedata @@ -104,6 +105,7 @@ # indefinitely on a genuinely broken composer. SEND_BUTTON_POLL_INTERVAL_S = 0.3 SEND_BUTTON_POLL_MAX_WAIT_S = 10.0 +APP_MENTION_MAX_WAIT_S = 8.0 class ChatGPTDom: @@ -303,6 +305,125 @@ async def type_message(self, text: str) -> None: ) logger.info("Typed: %s", text[:80]) + async def _type_app_query(self, query: str) -> None: + """Trigger mention handling with layout-independent keyboard text.""" + d = self._driver + for char in query: + await d._cdp("Input.dispatchKeyEvent", {"type": "keyDown", "key": char, "text": char}) + await d._cdp("Input.dispatchKeyEvent", {"type": "keyUp", "key": char}) + + async def type_message_with_apps(self, text: str, apps: list[str]) -> None: + """Rebuild the composer with ordered app chips, then append the prompt. + + Keep type_message's plain-text verification out of the populated + editor: mention nodes are structured UI metadata, not prompt text. + A click alone is insufficient; require a matching chip before sending. + """ + from .cdp_driver import CDPJSError, SendReadinessError + + d = self._driver + focus_end_js = ( + "(function(){" + f" const el = document.querySelector('{COMPOSER_SELECTOR}');" + " if (!el || !el.isContentEditable) return false;" + " el.focus();" + " const range = document.createRange();" + " range.selectNodeContents(el); range.collapse(false);" + " const selection = window.getSelection();" + " selection.removeAllRanges(); selection.addRange(range);" + " return true;" + "})()" + ) + app_name = apps[0] + try: + # Also clears stale chips/prompt after a rate-limit retry, using + # the existing clear + verification seam before adding any apps. + await d.type_message("") + for index, app_name in enumerate(apps): + if not await d._js_strict(focus_end_js): + raise SendReadinessError("No editable app composer found") + word = re.search(r"[^\W_]+", app_name) + query = (word.group() if word else app_name)[:3] + await self._type_app_query("@" + query) + clicked = False + deadline = time.monotonic() + APP_MENTION_MAX_WAIT_S + while time.monotonic() < deadline: + result = await d._js_strict( + "(function(){" + f" const composer = document.querySelector('{COMPOSER_SELECTOR}');" + " if (!composer) return 'missing composer';" + f" const name = {json.dumps(app_name)};" + " function exact(el, name) {" + " const text = (el.innerText || el.textContent || '').trim();" + " return text === name || text === '@' + name;" + " }" + " function visible(el) {" + " const r = el.getBoundingClientRect(), s = getComputedStyle(el);" + " return r.width > 0 && r.height > 0 && r.bottom > 0 && r.top < innerHeight" + " && s.visibility !== 'hidden' && s.display !== 'none' && s.opacity !== '0';" + " }" + # Confirm selection in the editor, not merely in the + # popup (which can retain the same display name). + f" if ({json.dumps(clicked)}) {{" + " const chips = Array.from(composer.querySelectorAll(" + " '[contenteditable=\"false\"], [data-type=\"mention\"]'));" + " const outer = chips.filter(el => visible(el)" + " && !chips.some(parent => parent !== el && parent.contains(el))" + " );" + f" const names = {json.dumps(apps[:index + 1])};" + " return names.every(name => outer.filter(el => exact(el, name)" + " || Array.from(el.querySelectorAll('*')).some(child => exact(child, name))).length" + " >= names.filter(n => n === name).length) ? 'selected' : 'waiting';" + " }" + " const c = composer.getBoundingClientRect();" + # Portalled popups need not be descendants of the + # form. Scope by popup semantics AND composer geometry. + " const popups = document.querySelectorAll(" + " '[role=\"listbox\"], [role=\"menu\"], [data-radix-popper-content-wrapper], .popover, .composer-home-top-menu, [data-composer-overlay-floating-ui=\"true\"]');" + " const candidates = new Set();" + " for (const popup of popups) {" + " const r = popup.getBoundingClientRect();" + " if (!visible(popup) || r.right < c.left || r.left > c.right" + " || Math.max(0, c.top - r.bottom, r.top - c.bottom) > 120) continue;" + " for (const el of popup.querySelectorAll('*')) {" + # Display names are exact; the internal prefix never + # decides which app is selected. + " if (!visible(el) || (el.innerText || '').trim() !== name) continue;" + " const interactive = el.closest('button, [role=\"option\"], [role=\"menuitem\"], [role=\"button\"], .__menu-item');" + # Live app suggestions can be plain clickable DIVs. + " const item = interactive && popup.contains(interactive) ? interactive : el;" + " if (!item || !popup.contains(item) || !visible(item) || item.disabled" + " || item.getAttribute('aria-disabled') === 'true' || composer.contains(item)" + " || item.closest('nav, aside, article, [data-message-author-role]')) continue;" + " candidates.add(item);" + " }" + " }" + " const items = Array.from(candidates).filter(item =>" + " !Array.from(candidates).some(child => child !== item && item.contains(child)));" + " if (items.length > 1) return 'ambiguous app suggestions';" + " if (!items.length) return 'waiting';" + " items[0].click();" + " return 'clicked';" + "})()" + ) + if result == "selected": + break + if result == "clicked": + clicked = True + elif result != "waiting": + raise SendReadinessError(f"App autocomplete: {result}") + await asyncio.sleep(0.1) + else: + raise SendReadinessError("App suggestion or selected chip not found") + if not await d._js_strict(focus_end_js): + raise SendReadinessError("No editable app composer found") + await d._cdp("Input.insertText", {"text": text}) + except (CDPJSError, SendReadinessError) as exc: + await d._capture_selector_diagnostic(f"app mention ({app_name})") + if d._breakers: + d._breakers.record_failure(BreakerKind.COMPOSER_SEND_READINESS) + raise SendReadinessError(f"Could not select app {app_name!r}: {exc}") from exc + async def _detect_select_all_modifier(self) -> int: """Return the CDP modifiers value for select-all on the live platform. diff --git a/src/chatgpt_web2api/completion_detector.py b/src/chatgpt_web2api/completion_detector.py index c5a47d8f..ec41ec3d 100644 --- a/src/chatgpt_web2api/completion_detector.py +++ b/src/chatgpt_web2api/completion_detector.py @@ -75,6 +75,13 @@ logger = logging.getLogger(__name__) +# Shared with the pre-send baseline; overlapping selectors count each root once. +ASSISTANT_ROOT_SELECTOR = ( + '[data-message-author-role="assistant"], ' + '[data-chatgpt-search-unit-key$=":assistant"], ' + '[data-content-search-unit-key$=":assistant"]' +) + # A generation is considered "stuck" (vs. merely slow) if no DOM progress # signal occurs within this window. Slow-but-progressing generations @@ -377,7 +384,7 @@ async def stream_until_complete( try: raw = await d._js_strict( - "document.querySelectorAll('[data-message-author-role=\"assistant\"]').length" + f"document.querySelectorAll('{ASSISTANT_ROOT_SELECTOR}').length" ) current_count = int(raw or 0) except CDPJSError: @@ -413,6 +420,7 @@ async def stream_until_complete( # (either .markdown text OR non-trivial HTML footprint). The threshold # (> 50 chars) prevents false 'done' from an empty/partial node. last_dom_text = "" + last_observed_text = "" last_html_len = 0 last_child_count = 0 had_non_text_content = False @@ -441,8 +449,8 @@ async def stream_until_complete( # generating — it is absent while the message is streaming or thinking. # Polling for that button on the NEW message is immune to the Stop # flicker and to the empty-.markdown-during-streaming quirk. Text is - # captured from the message's innerText (which IS populated during - # streaming) rather than .markdown textContent (which lags). + # captured only from answer markdown; an empty markdown container + # must not expose transient reasoning/tool UI through innerText. last_change_time = time.monotonic() deadline = time.monotonic() + timeout # P1: two-state phase-2 machine. phase_2_start tracks total observation @@ -474,22 +482,17 @@ async def stream_until_complete( try: result = await d._js_strict( "(function() {" - " var msgs = document.querySelectorAll('[data-message-author-role=\"assistant\"]');" - " if (!msgs.length) return JSON.stringify({text:'', md_text:'', html_len:0, child_count:0, has_action:false, is_thinking:false});" + f" var msgs = document.querySelectorAll('{ASSISTANT_ROOT_SELECTOR}');" + # Tool transitions can temporarily remove the new message. + # Never fall back to the previous turn's assistant node. + f" if (msgs.length <= {initial_count}) return JSON.stringify({{text:'', md_text:'', html_len:0, child_count:0, has_action:false, is_thinking:false}});" " var last = msgs[msgs.length - 1];" - # Text: the clean answer lives in ``.markdown`` textContent. - # It's empty during streaming and populates as the turn - # settles — so we ALSO capture ``innerText`` (populated - # during streaming) as a fallback. innerText includes the - # reasoning UI label ("Thinking.../Thought for N seconds"), - # so md_text is captured SEPARATELY and Python prefers it; - # the innerText fallback is trimmed of the leading label. - " var md = last.querySelector('.markdown');" - " var mdText = md ? (md.textContent || '') : '';" + # Only answer markdown is safe to stream. Raw innerText + # includes localized reasoning/status UI (live: aria-busy). + " var md = last.querySelector('.markdown, [data-markdown-text-style=\"assistant-message\"]');" + " var mdText = md ? (md.innerText || md.textContent || '') : '';" " var rawText = (last.innerText || '').trim();" - # Strip a leading "Thinking..." / "Thought for …" reasoning - # label so the innerText fallback can't leak it as a delta. - " var text = mdText || rawText.replace(/^Think(ing|\\s+for)[^\\n]*\\n?/i, '');" + " var text = mdText;" " var html_len = last.innerHTML.length;" " var child_count = last.children.length;" # has_action: the per-turn copy/feedback action row appears @@ -552,7 +555,7 @@ async def stream_until_complete( # Also recognize a plain "Thinking..." innerText placeholder # (some layouts show reasoning text without .result-thinking) # so the stall clock treats it as active generation, not a stall. - " var hasThinkingEl = !!last.querySelector('.result-thinking');" + " var hasThinkingEl = !!last.querySelector('.result-thinking, [aria-busy=\"true\"]');" " var visibleThinking = /^(thinking|reasoning)\\b/i.test(rawText.trim());" " var is_thinking = !has_action && (hasThinkingEl || (visibleThinking && !mdText));" " return JSON.stringify({text: text, md_text: mdText, html_len: html_len, child_count: child_count, has_action: has_action, is_thinking: is_thinking});" @@ -570,11 +573,7 @@ async def stream_until_complete( has_action = data.get("has_action", False) is_thinking = data.get("is_thinking", False) - # Streaming source: prefer the clean .markdown answer container - # over the innerText fallback (which carries the reasoning label). - # When md_text is empty (early streaming, before .markdown fills), - # the innerText fallback (with its leading label already stripped - # in JS) is what carries the streamed answer. + # Both fields come from answer markdown, never raw status UI. current = md_text or current # is_thinking means the model is actively reasoning — the DOM is @@ -602,7 +601,8 @@ async def stream_until_complete( # logging, but does NOT pause the stall clock (a stuck indicator # must not create an infinite hang). generation_active_signal = bool(is_thinking) - if current != last_dom_text: + if current != last_observed_text: + last_observed_text = current last_change_time = time.monotonic() # P1: first text content transitions us from awaiting_first_content # to streaming_after_first_content. The stall budget changes with @@ -613,11 +613,13 @@ async def stream_until_complete( if current and not first_content_seen: first_content_seen = True last_change_time = time.monotonic() # reset stream-idle clock - if len(current) > len(last_dom_text): + # Keep the emitted prefix across rewrites/temporary empty nodes. + # A longer string alone does not prove it extends the answer. + if current.startswith(last_dom_text) and current != last_dom_text: delta = current[len(last_dom_text) :] yield StreamChunk(delta=delta) - last_dom_text = current - self.last_dom_text = last_dom_text + last_dom_text = current + self.last_dom_text = last_dom_text # Non-text progress signals (images, tool-use, etc.) if html_len != last_html_len or child_count != last_child_count: diff --git a/src/chatgpt_web2api/identity_listener.py b/src/chatgpt_web2api/identity_listener.py index b71ee5ae..8a942371 100644 --- a/src/chatgpt_web2api/identity_listener.py +++ b/src/chatgpt_web2api/identity_listener.py @@ -36,6 +36,8 @@ import re from dataclasses import dataclass +from .turn_anchor import canonicalize_user_text + logger = logging.getLogger(__name__) @@ -347,7 +349,7 @@ async def _process_send_post(self, scope: CaptureScope, msg: dict, url: str) -> parts = (m0.get("content") or {}).get("parts") or [] body_text = "\n".join(str(p) for p in parts if isinstance(p, str)) if body_text: - body_hash = hashlib.sha256(body_text.encode("utf-8")).hexdigest() + body_hash = hash_sent_text(canonicalize_user_text(body_text)) if body_hash != scope.expected_text_hash: # Text doesn't match — could be a different send (retry, # regenerate). Don't resolve; leave scope open. @@ -392,8 +394,7 @@ async def _fetch_post_data(self, request_id: str | None) -> str | None: def hash_sent_text(text: str) -> str: """Stable SHA-256 hash of the sent prompt, for capture validation. - The hash is computed over the *raw* sent text (before any normalization) - because the POST body carries the raw text. Used by the capture scope to - validate a POST belongs to this send (failure-mode D). + Hashes the supplied text unchanged. Callers normalize logical text or + canonicalize observed POST text before hashing (failure-mode D). """ return hashlib.sha256(text.encode("utf-8")).hexdigest() diff --git a/src/chatgpt_web2api/mcp_server.py b/src/chatgpt_web2api/mcp_server.py index 8387f0c4..72040b3f 100644 --- a/src/chatgpt_web2api/mcp_server.py +++ b/src/chatgpt_web2api/mcp_server.py @@ -40,7 +40,7 @@ from mcp import types as mcp_types from mcp.server import Server from mcp.server.stdio import stdio_server -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator from .breakers import BreakerKind, BreakerRegistry, CircuitOpenError from .cdp_driver import ( @@ -82,6 +82,21 @@ class ChatCompletionInput(BaseModel): """Input schema for chat_completion tool.""" message: str = Field(description="The user message to send to ChatGPT") + apps: list[str] | None = Field( + default=None, + description="Installed ChatGPT app display names to select via @mention, in order.", + ) + + @field_validator("apps", mode="before") + @classmethod + def validate_apps(cls, value): + if value is not None and ( + not isinstance(value, list) + or any(not isinstance(name, str) or not name.strip() for name in value) + ): + raise ValueError("apps must be a list of non-empty app names") + return [name.strip() for name in value] if value is not None else None + system_prompt: str | None = Field( default=None, description=( @@ -789,6 +804,7 @@ async def do_chat_completion( ) async for chunk in driver.send_and_stream( full_text, timeout=120, budgets=_budgets, model=validated.model, + **({"apps": validated.apps} if validated.apps else {}), ): if chunk.delta: full_response += chunk.delta diff --git a/src/chatgpt_web2api/turn_anchor.py b/src/chatgpt_web2api/turn_anchor.py index e14164a0..f3cc85ed 100644 --- a/src/chatgpt_web2api/turn_anchor.py +++ b/src/chatgpt_web2api/turn_anchor.py @@ -42,6 +42,8 @@ # can still pass). For rapid same-text repeats, the selector returns ``ambiguous`` # or ``not_ready`` rather than silently picking. Env-overridable for canary tuning. import os as _os +import re +import string import unicodedata from dataclasses import dataclass, field, replace from typing import Literal @@ -207,6 +209,18 @@ def normalize_text(s: str) -> str: return s.strip() +def canonicalize_user_text(text: str) -> str: + """Recover logical text from ChatGPT's user-message Markdown for matching. + + Strip only leading app prompt-links, then undo one layer of Markdown + escapes of ASCII punctuation. Keep ordinary/embedded links and literal + backslashes before non-punctuation. Sent logical text must not be decoded. + """ + text = re.sub(r"\A(?:\[\$[^\]\r\n]+\]\(app://[^)\s]+\)[ \t\r\n]*)+", "", text) + text = re.sub(r"\\([" + re.escape(string.punctuation) + r"])", r"\1", text) + return normalize_text(text) + + def user_text_matches_sent(parent_text: str, sent_text: str) -> bool: """Does a backend user-node's text match the sent prompt? @@ -217,7 +231,7 @@ def user_text_matches_sent(parent_text: str, sent_text: str) -> bool: - No tiny prefixes — agent prompts share long boilerplate prefixes, so a short shared prefix is not distinctive enough to match on. """ - p = normalize_text(parent_text) + p = canonicalize_user_text(parent_text) s = normalize_text(sent_text) if not p or not s: return False diff --git a/stream-test.json b/stream-test.json new file mode 100644 index 00000000..7dbb87cf --- /dev/null +++ b/stream-test.json @@ -0,0 +1,14 @@ +{ + "messages": [ + { + "role": "user", + "content": "Verwende beide ausgewählten Apps.\n\n1. Rufe über Hermes Memory MCP NoAuth das Tool get_poc_memory auf.\n2. Lies über GitHub aus rosterbrink-awp/Keystone auf Branch master die Datei global.json und ermittle daraus die SDK-Version.\n\nAntworte ausschließlich in genau diesem Format:\nMEMORY=\nSDK=" + } + ], + "model": "auto", + "apps": [ + "GitHub", + "Hermes Memory MCP NoAuth" + ], + "stream": true +} diff --git a/tests/test_apps.py b/tests/test_apps.py new file mode 100644 index 00000000..75caf02d --- /dev/null +++ b/tests/test_apps.py @@ -0,0 +1,506 @@ +"""App contract and send regressions; no live ChatGPT account required.""" + +import asyncio +import json +import shutil +import subprocess +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +import websockets +from aiohttp import ClientSession +from aiohttp.test_utils import TestClient, TestServer +from pydantic import ValidationError + +from chatgpt_web2api.api_server import APIServer +from chatgpt_web2api.cdp_driver import CDPDriver, RateLimitError, SendReadinessError, StreamChunk +from chatgpt_web2api.config import Config +from chatgpt_web2api.identity_listener import IdentityListener, hash_sent_text +from chatgpt_web2api.mcp_server import ChatCompletionInput, do_chat_completion +from chatgpt_web2api.resilience import retry_on_rate_limit + + +@pytest.fixture(autouse=True) +def isolated_lock_files(monkeypatch, tmp_path): + monkeypatch.setattr("chatgpt_web2api.cross_process_lock.Path.home", lambda: tmp_path) + + +def send_driver(): + driver = CDPDriver() + driver._read_assistant_count_baseline = AsyncMock(return_value=0) + driver._verify_send_acknowledged = AsyncMock(return_value=True) + driver._js_strict = AsyncMock(return_value="https://chatgpt.com/c/test") + driver._fetch_text_for_turn = AsyncMock( + return_value=SimpleNamespace(status="matched", text="OK", diagnostic={}) + ) + driver.type_message = AsyncMock() + driver.click_send = AsyncMock() + driver._capture_selector_diagnostic = AsyncMock() + driver._completion.last_dom_text = "OK" + + async def complete(**kwargs): + yield StreamChunk(delta="OK") + + driver._completion.stream_until_complete = complete + return driver + + +@pytest.mark.parametrize("kwargs", [{}, {"apps": None}, {"apps": []}]) +async def test_no_apps_preserves_original_send_path(kwargs): + driver = send_driver() + driver.type_message_with_apps = AsyncMock() + chunks = [chunk async for chunk in driver.send_and_stream("prompt", **kwargs)] + driver.type_message.assert_awaited_once_with("prompt") + driver.type_message_with_apps.assert_not_awaited() + driver.click_send.assert_awaited_once() + assert chunks[-1].finish_reason == "stop" + + +@pytest.mark.parametrize("apps", [["GitHub"], ["GitHub", "Hermes Memory MCP NoAuth"]]) +async def test_apps_selected_in_order_before_prompt_and_send(apps): + driver = send_driver() + events = [] + + async def clear(text): + events.append(("clear", text)) + + async def cdp(method, params): + events.append((method, params)) + + results = iter([value for _ in apps for value in [True, "waiting", "clicked", "selected"]] + + [True, "https://chatgpt.com/c/test"]) + + async def js(expr): + result = next(results) + if result in ("clicked", "selected"): + events.append(("chip", result)) + return result + + async def click(): + events.append(("send", "")) + + driver.type_message = clear + driver._cdp = cdp + driver._js_strict = js + driver.click_send = click + _ = [chunk async for chunk in driver.send_and_stream("prompt", apps=apps)] + expected = [("clear", "")] + for app in apps: + for char in "@" + app[:3]: + expected.extend([ + ("Input.dispatchKeyEvent", {"type": "keyDown", "key": char, "text": char}), + ("Input.dispatchKeyEvent", {"type": "keyUp", "key": char}), + ]) + expected.extend([("chip", "clicked"), ("chip", "selected")]) + assert events == expected + [("Input.insertText", {"text": "prompt"}), ("send", "")] + + +@pytest.mark.parametrize("clicked", [False, True], ids=["missing-suggestion", "click-without-chip"]) +async def test_app_failure_prevents_prompt_and_send_and_closes_capture(monkeypatch, clicked): + import chatgpt_web2api.chatgpt_dom as dom + + monkeypatch.setattr(dom, "APP_MENTION_MAX_WAIT_S", 0.05) + driver = send_driver() + driver._cdp = AsyncMock() + driver._identity_listener = IdentityListener(driver) + driver._identity_listener._ready = True + values = iter([True, "clicked"] if clicked else [True]) + + async def js(expr): + return next(values, "waiting") + + driver._js_strict = js + with pytest.raises(SendReadinessError, match="Missing App"): + _ = [chunk async for chunk in driver.send_and_stream("prompt", apps=["Missing App"])] + driver.click_send.assert_not_awaited() + assert all(call.args[0] == "Input.dispatchKeyEvent" for call in driver._cdp.await_args_list) + assert "".join(call.args[1].get("text", "") for call in driver._cdp.await_args_list) == "@Mis" + driver._capture_selector_diagnostic.assert_awaited_once() + assert driver._identity_listener._active_scope is None + + +async def test_rate_limit_retry_rebuilds_apps_and_prompt(monkeypatch): + import chatgpt_web2api.resilience as resilience + + monkeypatch.setattr(resilience.asyncio, "sleep", AsyncMock()) + driver = send_driver() + events = [] + + async def clear(text): + events.append("clear") + + async def cdp(method, params): + if "text" in params: + events.append(params["text"]) + + driver.type_message = clear + driver._cdp = cdp + driver._js_strict = AsyncMock(side_effect=[True, "clicked", "selected", True] * 2 + + ["https://chatgpt.com/c/test"]) + driver.click_send = AsyncMock(side_effect=[RateLimitError(retry_after=1), None]) + driver.dismiss_rate_limit = AsyncMock(return_value=True) + + async def send(): + return [chunk async for chunk in driver.send_and_stream("prompt", apps=["GitHub"])] + + await retry_on_rate_limit(driver, send, max_attempts=2) + assert events == ["clear", "@", "G", "i", "t", "prompt"] * 2 + assert driver.click_send.await_count == 2 + + +async def test_app_query_keyboard_events_in_local_contenteditable(tmp_path): + """Exercise real CDP input on an isolated blank page, without a ChatGPT session.""" + chrome = shutil.which(Config().chrome.chrome_path) + if not chrome: + pytest.skip("Chrome is required for the local DOM regression") + profile = tmp_path / "chrome-profile" + driver = CDPDriver() + driver.reconnect = AsyncMock(side_effect=AssertionError("Local test must not reconnect")) + with (tmp_path / "chrome.log").open("w", encoding="utf-8") as log: + process = subprocess.Popen( + [chrome, "--headless", "--disable-gpu", "--disable-background-networking", + "--no-first-run", "--no-default-browser-check", "--remote-debugging-port=0", + f"--user-data-dir={profile}", "about:blank"], + stdout=log, stderr=subprocess.STDOUT, + ) + try: + async with asyncio.timeout(15): + port_file = profile / "DevToolsActivePort" + while not port_file.exists() or not port_file.read_text().strip(): + assert process.poll() is None, "Local Chrome exited before CDP was ready" + await asyncio.sleep(0.05) + port = int(port_file.read_text().splitlines()[0]) + async with ClientSession() as client: + async with client.get(f"http://127.0.0.1:{port}/json/list") as response: + targets = await response.json() + target = next(t for t in targets if t.get("type") == "page" and t["url"] == "about:blank") + driver._ws = await websockets.connect(target["webSocketDebuggerUrl"]) + driver._reader_task = asyncio.create_task(driver._reader_loop()) + await driver._js_strict(""" + document.body.innerHTML = '
'; + for (const type of ['keydown', 'keypress', 'keyup', 'beforeinput', 'input']) { + editor.addEventListener(type, event => window.events.push({ + type, key: event.key, data: event.data, trusted: event.isTrusted, + keyboard: event instanceof KeyboardEvent, + input: event instanceof InputEvent + })); + } + true; + """) + for query in ("@Git", "@Her", "@Äpp"): + await driver._js_strict("editor.textContent = ''; window.events = []; editor.focus(); true;") + await driver._dom._type_app_query(query) + observed = json.loads(await driver._js_strict( + "JSON.stringify({text: editor.textContent, events: window.events})" + )) + assert observed["text"] == query + events = observed["events"] + assert all(event["trusted"] for event in events) + assert [(event["type"], event["key"]) for event in events if event["keyboard"]] == [ + (kind, char) for char in query for kind in ("keydown", "keypress", "keyup") + ] + assert [(event["type"], event["data"]) for event in events if event["input"]] == [ + (kind, char) for char in query for kind in ("beforeinput", "input") + ] + finally: + if driver._reader_task: + driver._reader_task.cancel() + await asyncio.gather(driver._reader_task, return_exceptions=True) + if driver._ws: + await driver._ws.close() + process.terminate() + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=10) + + +def transport_driver(): + driver = MagicMock(spec=CDPDriver) + driver._current_conv_id = None + driver.navigate_new_chat = AsyncMock() + driver._js_strict = AsyncMock(return_value='{"text":""}') + driver.calls = [] + + async def stream(text, **kwargs): + driver.calls.append((text, kwargs)) + yield StreamChunk(delta="OK", finish_reason="stop") + + driver.send_and_stream = stream + return driver + + +APP_INPUTS = [ + {}, {"apps": None}, {"apps": []}, {"apps": ["GitHub", "Hermes Memory MCP NoAuth"]}, + {"apps": [" GitHub ", "Hermes Memory MCP NoAuth\t", "GitHub"]}, +] +INVALID_APPS = ["GitHub", {}, [""], [" \n"], [None], [1], [True], ["GitHub", {}]] + + +@pytest.mark.parametrize("fields", APP_INPUTS) +@pytest.mark.parametrize("stream", [False, True]) +async def test_rest_apps_reach_driver(fields, stream): + driver = transport_driver() + server = APIServer(Config(), driver) + async with TestClient(TestServer(server.app)) as client: + response = await client.post("/v1/chat/completions", json={ + "messages": [{"role": "user", "content": "prompt"}], "stream": stream, **fields, + }) + assert response.status == 200 + assert "OK" in await response.text() + assert len(driver.calls) == 1 + text, kwargs = driver.calls[0] + assert text == "[User]\nprompt" + if fields.get("apps"): + assert kwargs["apps"] == [name.strip() for name in fields["apps"]] + else: + assert "apps" not in kwargs # Preserve existing driver/test seams. + + +@pytest.mark.parametrize("apps", INVALID_APPS) +async def test_rest_rejects_invalid_apps_before_browser_mutation(apps): + driver = transport_driver() + server = APIServer(Config(), driver) + async with TestClient(TestServer(server.app)) as client: + response = await client.post("/v1/chat/completions", json={ + "messages": [{"role": "user", "content": "prompt"}], "apps": apps, + }) + assert response.status == 400 + assert (await response.json())["error"]["type"] == "invalid_request_error" + driver.navigate_new_chat.assert_not_awaited() + assert driver.calls == [] + + +@pytest.mark.parametrize("fields", APP_INPUTS) +async def test_mcp_apps_schema_and_forwarding(fields): + schema = ChatCompletionInput.model_json_schema() + assert "apps" in schema["properties"] + assert "apps" not in schema["required"] + driver = transport_driver() + result = await do_chat_completion(driver, {"message": "prompt", **fields}, Config()) + assert result["content"] == "OK" + assert len(driver.calls) == 1 + text, kwargs = driver.calls[0] + assert text == "prompt" + if fields.get("apps"): + assert kwargs["apps"] == [name.strip() for name in fields["apps"]] + else: + assert "apps" not in kwargs + + +@pytest.mark.parametrize("apps", INVALID_APPS) +def test_mcp_rejects_invalid_apps(apps): + with pytest.raises(ValidationError, match="apps"): + ChatCompletionInput(message="prompt", apps=apps) + + +async def test_app_metadata_does_not_change_identity_text_hash(): + """Synthetic payload contract, not evidence of ChatGPT's live serializer.""" + listener = IdentityListener(MagicMock()) + prompt = "original prompt\nwith a second line" + scope = listener.arm_capture_scope( + expected_text_hash=hash_sent_text(prompt), conversation_id=None, target_id="test", + ) + message_id = "11111111-1111-4111-8111-111111111111" + body = {"action": "next", "messages": [{ + "id": message_id, + "author": {"role": "user"}, + "content": {"parts": [prompt, {"type": "app", "name": "GitHub"}]}, + "metadata": {"apps": ["GitHub"]}, + }]} + url = "https://chatgpt.com/backend-api/f/conversation" + try: + await listener._process_send_post(scope, {"params": { + "requestId": "request-1", "request": {"url": url, "postData": json.dumps(body)}, + }}, url) + assert await listener.wait_for_captured_uuid(timeout=0.1) == message_id + finally: + scope.close() + + +APP_PROMPT = "[User]\nVerwende beide ausgewählten Apps.\n\n1. Test\nMEMORY=" +APP_SERIALIZED_PROMPT = "[User]\nVerwende beide ausgewählten Apps.\n\n1\\. Test\nMEMORY=\\" +APP_LINKS = ( + "[$github](app://connector_76869538009648d5b282a4bb21c3d157)" + "[$hermes-memory-mcp-noauth](app://asdk_app_6ab28256fde8819197fa5c529d311f41)" +) + + +@pytest.mark.parametrize("apps,prefix", [ + (None, ""), + ([], ""), + (["Hermes Memory MCP NoAuth"], + "[$hermes-memory-mcp-noauth](app://asdk_app_6ab28256fde8819197fa5c529d311f41)"), + (["GitHub", "Hermes Memory MCP NoAuth"], APP_LINKS), +]) +@pytest.mark.parametrize("logical,serialized", [ + (APP_PROMPT, APP_SERIALIZED_PROMPT), + ("Cafe\u0301\r\n1. Test ", "Café\n1\\. Test"), +]) +async def test_app_serialization_captures_uuid_with_logical_anchor(apps, prefix, logical, serialized): + """Observed app links/Markdown escapes must match the logical send text.""" + driver = send_driver() + driver.type_message_with_apps = AsyncMock() + listener = driver._identity_listener = IdentityListener(driver) + listener._ready = True + message_id = "11111111-1111-4111-8111-111111111111" + + async def click(): + body = {"action": "next", "messages": [{ + "id": message_id, "author": {"role": "user"}, + "content": {"content_type": "text", "parts": [prefix + serialized]}, + }]} + await listener._process_send_post(listener._active_scope, {"params": { + "request": {"postData": json.dumps(body)}, + }}, "https://chatgpt.com/backend-api/f/conversation") + + async def complete(**kwargs): + anchor = kwargs["turn_anchor"] + assert anchor.sent_text == logical + assert anchor.captured_user_message_id == message_id + yield StreamChunk(delta="OK") + + driver.click_send = click + driver._completion.stream_until_complete = complete + _ = [chunk async for chunk in driver.send_and_stream(logical, apps=apps)] + assert listener.capture_success_count == 1 + assert listener._active_scope is None + driver._verify_send_acknowledged.assert_not_awaited() + + +async def test_app_without_captured_uuid_reconciles_serialized_text(): + from chatgpt_web2api.turn_anchor import select_text_for_turn + + driver = send_driver() + driver._current_conv_id = "test" + driver.type_message_with_apps = AsyncMock() + listener = driver._identity_listener = IdentityListener(driver) + listener._ready = True + listener.wait_for_captured_uuid = AsyncMock(return_value=None) + apps = ["GitHub", "Hermes Memory MCP NoAuth"] + prompt = APP_PROMPT + wire_text = APP_LINKS + APP_SERIALIZED_PROMPT + marker = "POC0_MEMORY_MARKER_7F2A" + + def turn(suffix, timestamp, answer): + user, assistant = "u-" + suffix, "a-" + suffix + return { + user: {"id": user, "children": [assistant], "role": "user", + "create_time": timestamp, "content_type": "text", "text": wire_text}, + assistant: {"id": assistant, "parent": user, "children": [], "role": "assistant", + "end_turn": True, "create_time": timestamp + 1, + "content_type": "text", "text": answer}, + } + + old_nodes = turn("old", 100, "STALE") + projection = {"nodes": {**old_nodes, **turn("new", 200, marker)}} + driver._backend_client._fetch_recent_conversation_projection = AsyncMock( + return_value={"nodes": old_nodes} + ) + + async def fetch(conv_id, anchor): + assert conv_id == "test" + assert anchor.captured_user_message_id is None + assert anchor.mode == "existing_conversation" + assert anchor.sent_text == prompt + result = select_text_for_turn(projection, anchor) + assert result.status == "matched", result.diagnostic + assert result.text == marker + assert result.diagnostic["assistant_node"] == "a-new" + return result + + async def complete(**kwargs): + driver._completion.last_dom_text = "" + if False: + yield + + driver._fetch_text_for_turn = fetch + driver._completion.stream_until_complete = complete + chunks = [c async for c in driver.send_and_stream(prompt, apps=apps)] + assert "".join(c.delta for c in chunks) == marker + driver.type_message_with_apps.assert_awaited_once_with(prompt, apps) + driver.click_send.assert_awaited_once() + listener.wait_for_captured_uuid.assert_awaited_once() + assert listener._active_scope is None + + +@pytest.mark.parametrize("popup_attributes,suggestion_html,app_name,click_selector", [ + ('role="listbox"', 'Hermes Memory MCP NoAuth', 'Hermes Memory MCP NoAuth', '#suggestion'), + ('class="popover"', '
Hermes Memory MCP NoAuth' + 'Hermes Memory MCP NoAuth
', 'Hermes Memory MCP NoAuth', '.__menu-item'), + ('class="suggestionMenu-XYZ composer-home-top-menu"', + '', + 'GitHub', 'button'), + ('data-composer-overlay-floating-ui="true"', + '
', + 'GitHub', 'button'), +]) +async def test_plain_div_app_suggestion_in_real_dom( + tmp_path, popup_attributes, suggestion_html, app_name, click_selector, +): + """Run the generated selector in local Chrome; no ChatGPT session or network.""" + chrome = shutil.which(Config().chrome.chrome_path) + if not chrome: + pytest.skip("Chrome is required for the local DOM regression") + driver = send_driver() + driver._cdp = AsyncMock() + scripts = [] + responses = iter([True, "clicked", "selected", True]) + + async def capture_script(expr): + scripts.append(expr) + return next(responses) + + driver._js_strict = capture_script + await driver.type_message_with_apps("prompt", [app_name]) + calls = driver._cdp.await_args_list + assert all(call.args[0] == "Input.dispatchKeyEvent" for call in calls[:-1]) + assert "".join(call.args[1].get("text", "") for call in calls[:-1]) == "@" + app_name[:3] + assert calls[-1].args == ("Input.insertText", {"text": "prompt"}) + fixture = tmp_path / "app-suggestion.html" + fixture.write_text(""" +
+
+
""" + suggestion_html + """
+
+ + """, encoding="utf-8") + result = subprocess.run( + [chrome, "--headless", "--disable-gpu", "--disable-background-networking", + "--no-first-run", "--no-default-browser-check", "--window-size=1280,900", + f"--user-data-dir={tmp_path / 'chrome-profile'}", "--dump-dom", fixture.as_uri()], + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stderr + assert '
[true,"waiting",0,"clicked","selected",1,"waiting"]
' in result.stdout, ( + result.stdout + result.stderr + ) diff --git a/tests/test_completion_detector.py b/tests/test_completion_detector.py index d7a50cf2..359a2302 100644 --- a/tests/test_completion_detector.py +++ b/tests/test_completion_detector.py @@ -22,16 +22,23 @@ rule); error classes / StreamChunk are imported lazily inside the method. """ +import html import inspect +import json +import re +import shutil +import subprocess from unittest.mock import AsyncMock, MagicMock import pytest +from chatgpt_web2api.cdp_driver import CDPDriver from chatgpt_web2api.completion_detector import ( PHASE_STALL_SECONDS, CompletionDetector, ) -from chatgpt_web2api.turn_anchor import TurnAnchor, TurnEndResult +from chatgpt_web2api.config import Config +from chatgpt_web2api.turn_anchor import TurnAnchor, TurnEndResult, TurnTextResult def _make_detector(): @@ -237,3 +244,164 @@ def importlib_import_module(name): import importlib return importlib.import_module(name) + + +async def _stream_probes(monkeypatch, probes, backend_texts, *, initial_count=1, phase1_counts=(2,)): + """Exercise the detector and final reconciliation with a virtual clock.""" + driver = CDPDriver() + clock = [100.0] + monkeypatch.setattr("chatgpt_web2api.completion_detector.time.monotonic", lambda: clock[0]) + + async def sleep(seconds): + clock[0] += seconds + + monkeypatch.setattr("chatgpt_web2api.completion_detector.asyncio.sleep", sleep) + driver._read_assistant_count_baseline = AsyncMock(return_value=initial_count) + driver._verify_send_acknowledged = AsyncMock(return_value=True) + driver.type_message = AsyncMock() + driver.click_send = AsyncMock() + index = 0 + counts = iter(phase1_counts) + + async def js(expr): + nonlocal index + if "has_action" in expr: + probe = probes[min(index, len(probes) - 1)] + index += 1 + return json.dumps(probe) + if "body.innerText" in expr: + return '{"text":""}' + if "location.href" in expr: + return "https://chatgpt.com/c/real-conversation" + return next(counts) + + async def end_turn(*args, **kwargs): + return TurnEndResult(status="matched" if index >= len(probes) else "not_ready") + + driver._js_strict = js + driver._fetch_end_turn_for_turn = end_turn + driver._fetch_text_for_turn = AsyncMock(side_effect=[ + TurnTextResult(status="matched", text=text) for text in backend_texts + ]) + return [c.delta async for c in driver.send_and_stream("prompt") if c.delta] + + +@pytest.mark.parametrize("texts,expected", [ + (["ABC", "ABCDE", "ABCDEFG"], ["ABC", "DE", "FG"]), + (["ABC", "Unrelated longer rendering", "ABCDEFG"], ["ABC", "DEFG"]), + (["ABC", "", "A", "ABCDEFG"], ["ABC", "DEFG"]), +]) +async def test_deltas_only_extend_emitted_prefix(monkeypatch, texts, expected): + probes = [{"text": text, "md_text": text, "html_len": 60} for text in texts] + assert await _stream_probes(monkeypatch, probes, ["ABCDEFG"]) == expected + + +async def test_reconciliation_waits_for_matching_prefix(monkeypatch): + probes = [{"text": "ABC", "md_text": "ABC", "html_len": 60}] + assert await _stream_probes(monkeypatch, probes, ["Unrelated text", "ABCDEFG"]) == [ + "ABC", "DEFG", + ] + + +@pytest.mark.parametrize("root_attributes,markdown_attributes", [ + ('data-message-author-role="assistant"', 'class="markdown"'), + ('data-chatgpt-search-unit-key="fallback-turn-0:2:assistant"', + 'data-markdown-text-style="assistant-message"'), + ('data-content-search-unit-key="fallback-turn-0:2:assistant"', + 'data-markdown-text-style="assistant-message"'), + ('data-chatgpt-search-unit-key="fallback-turn-0:2:assistant" ' + 'data-content-search-unit-key="fallback-turn-0:2:assistant"', + 'data-markdown-text-style="assistant-message"'), +]) +@pytest.mark.parametrize("initial_count", [0, 1], ids=["fresh-chat", "existing-chat"]) +async def test_reasoning_dom_and_source_switch_emit_only_answer( + monkeypatch, tmp_path, root_attributes, markdown_attributes, initial_count, +): + """Execute the real probe JS against the observed German aria-busy DOM. + + Also cover a disappearing tool placeholder and raw-text -> markdown switch. + Reuse the local headless Chrome approach from test_apps; no live account. + """ + chrome = shutil.which(Config().chrome.chrome_path) + if not chrome: + pytest.skip("Chrome is required for the local DOM regression") + detector, driver = _make_detector() + scripts = {} + baseline_driver = CDPDriver() + baseline_driver._js_strict = AsyncMock(side_effect=[initial_count, 0]) + assert await baseline_driver._read_assistant_count_baseline() == initial_count + scripts["baseline"] = baseline_driver._js_strict.await_args_list[0].args[0] + + async def capture(expr): + if "has_action" in expr: + scripts["phase2"] = expr + return '{"text":"answer","md_text":"answer","has_action":true}' + if "body.innerText" in expr: + return '{"text":""}' + scripts["phase1"] = expr + return initial_count + 1 + + driver._js_strict = capture + _ = [c async for c in detector.stream_until_complete( + initial_count=initial_count, timeout=5, turn_anchor=TurnAnchor(sent_text="test", mode="fresh_chat"), + )] + marker = "MEMORY=POC0_MEMORY_MARKER_7F2A\nSDK=10.0.100" + fixture = tmp_path / "reasoning.html" + fixture.write_text(""" +

Heading alone is not a message root

+ +
+ """, encoding="utf-8") + result = subprocess.run( + [chrome, "--headless", "--disable-gpu", "--disable-background-networking", + "--no-first-run", "--no-default-browser-check", + f"--user-data-dir={tmp_path / 'chrome-profile'}", "--dump-dom", fixture.as_uri()], + capture_output=True, text=True, encoding="utf-8", timeout=30, + ) + assert result.returncode == 0, result.stderr + output = re.search(r'
(.*?)
', result.stdout, re.DOTALL) + assert output, result.stdout + result.stderr + observed = json.loads(html.unescape(output.group(1))) + assert observed["countsBefore"] == [initial_count, initial_count] + assert observed["countsAfter"] == [initial_count + 1, initial_count + 1] + probes = observed["results"] + assert probes[0]["is_thinking"] is True + assert [p["text"] for p in probes] == ["", "", "", "MEMORY=", marker] + assert observed["paragraphs"]["text"] == "First paragraph\n\nSecond paragraph\nNext line" + assert "".join(await _stream_probes( + monkeypatch, probes, [marker], initial_count=observed["countsBefore"][0], + phase1_counts=[observed["countsBefore"][1], observed["countsAfter"][1]], + )) == marker diff --git a/tests/test_identity_listener.py b/tests/test_identity_listener.py index 3c5d6f1c..6ba0000d 100644 --- a/tests/test_identity_listener.py +++ b/tests/test_identity_listener.py @@ -16,6 +16,7 @@ IdentityListener, hash_sent_text, ) +from chatgpt_web2api.turn_anchor import normalize_text def _make_driver(): @@ -124,6 +125,54 @@ async def test_wrong_text_hash_does_not_resolve(): scope.close() +@pytest.mark.parametrize("logical,serialized", [ + ("[User]\nVerwende beide ausgewählten Apps.\n\n1. Test\nMEMORY=", + "[$github](app://connector_76869538009648d5b282a4bb21c3d157)" + "[$hermes-memory-mcp-noauth](app://asdk_app_6ab28256fde8819197fa5c529d311f41)" + "[User]\nVerwende beide ausgewählten Apps.\n\n1\\. Test\nMEMORY=\\"), + ("Café\n1. Test", "[$app](app://any-id)Cafe\u0301\r\n1\\. Test"), + (r"Literal \* and C:\temp", r"Literal \\\* and C:\\temp"), +]) +async def test_serialized_user_text_captures_logical_prompt(logical, serialized): + listener = IdentityListener(_make_driver()) + scope = listener.arm_capture_scope( + expected_text_hash=hash_sent_text(normalize_text(logical)), + conversation_id="conv-1", target_id="tgt-1", + ) + try: + await listener._process_send_post( + scope, _make_send_event(text=serialized), + "https://chatgpt.com/backend-api/f/conversation", + ) + assert await listener.wait_for_captured_uuid(timeout=0.1) == "11111111-1111-4111-8111-111111111111" + finally: + scope.close() + + +@pytest.mark.parametrize("serialized", [ + "[$app](app://any-id)different prompt", + "[docs](https://example.org)our prompt", + "[$app](https://example.org)our prompt", + "prefix [$app](app://any-id)our prompt", + "[$app](app://any-id)", +]) +async def test_canonicalization_does_not_capture_foreign_prompt(serialized): + listener = IdentityListener(_make_driver()) + scope = listener.arm_capture_scope( + expected_text_hash=hash_sent_text("our prompt"), + conversation_id="conv-1", target_id="tgt-1", + ) + try: + await listener._process_send_post( + scope, _make_send_event(text=serialized), + "https://chatgpt.com/backend-api/f/conversation", + ) + assert not scope.future.done() + assert listener.capture_success_count == 0 + finally: + scope.close() + + @pytest.mark.asyncio async def test_wrong_conversation_id_does_not_resolve(): """A POST for a different conversation is ignored.""" diff --git a/tests/test_send_acknowledgment.py b/tests/test_send_acknowledgment.py index 9fdf70e2..0534a9d2 100644 --- a/tests/test_send_acknowledgment.py +++ b/tests/test_send_acknowledgment.py @@ -19,12 +19,17 @@ """ import asyncio +import html import json +import re +import shutil +import subprocess from unittest.mock import AsyncMock, MagicMock import pytest from chatgpt_web2api.cdp_driver import CDPDriver, SendReadinessError +from chatgpt_web2api.config import Config from chatgpt_web2api.turn_anchor import TurnReconciliationError, TurnTextResult @@ -202,3 +207,59 @@ async def fake_js_strict(expr, timeout=15): assert result is None, ( f"Missing composer should return None (inconclusive), got {result!r}" ) + + +@pytest.mark.parametrize("attributes", [ + 'data-message-author-role="user"', + 'data-markdown-text-tone="user-message"', + 'class="rich-text-user-turn"', + 'data-markdown-text-tone="user-message" class="other rich-text-user-turn"', + 'data-message-author-role="user" data-markdown-text-tone="user-message" class="rich-text-user-turn"', +]) +async def test_user_count_baseline_and_acknowledgment_in_real_dom(tmp_path, attributes): + """Execute both production probes in isolated local Chrome, without ChatGPT.""" + chrome = shutil.which(Config().chrome.chrome_path) + if not chrome: + pytest.skip("Chrome is required for the local DOM regression") + driver = _make_driver() + driver._js_strict = AsyncMock(side_effect=[0, 1]) + assert await driver._read_assistant_count_baseline() == 0 + assert driver._pre_send_user_count == 1 + baseline_js = driver._js_strict.await_args_list[1].args[0] + driver._js_strict = AsyncMock(return_value=json.dumps({ + "userCount": 2, "composerPresent": True, "composerEmpty": True, + })) + assert await driver._verify_send_acknowledged() is True + ack_js = driver._js_strict.await_args.args[0] + + fixture = tmp_path / "user-turn.html" + fixture.write_text(""" +
+
Existing user message
+ """, encoding="utf-8") + result = subprocess.run( + [chrome, "--headless", "--disable-gpu", "--disable-background-networking", + "--no-first-run", "--no-default-browser-check", + f"--user-data-dir={tmp_path / 'chrome-profile'}", "--dump-dom", fixture.as_uri()], + capture_output=True, text=True, timeout=30, + ) + assert result.returncode == 0, result.stderr + output = re.search(r'
(.*?)
', result.stdout, re.DOTALL) + assert output, result.stdout + result.stderr + before, after = json.loads(html.unescape(output.group(1))) + assert before[0] == before[1]["userCount"] == 1 + assert after[0] == after[1]["userCount"] == 2 # overlapping selectors count each node once + driver._pre_send_user_count = before[0] + driver._js_strict = AsyncMock(return_value=json.dumps(after[1])) + assert await driver._verify_send_acknowledged() is True diff --git a/tests/test_sse_end_turn_new_chat.py b/tests/test_sse_end_turn_new_chat.py index fd11bea2..ecfc4333 100644 --- a/tests/test_sse_end_turn_new_chat.py +++ b/tests/test_sse_end_turn_new_chat.py @@ -123,6 +123,50 @@ async def _fake_js2(expr, timeout=15): # ── 3. Loop resolves conv_id_for_check mid-loop ───────────────────────── +@pytest.mark.parametrize("path,expected", [ + ("/c/WEB:temporary-id", ""), + ("/g/project/c/WEB:temporary-id", ""), + ("/c/real-id?model=auto", "real-id"), + ("/g/project/c/real-id", "real-id"), +]) +async def test_conversation_url_ignores_only_provisional_ids(path, expected): + driver = _make_driver() + driver._js_strict = AsyncMock(return_value="https://chatgpt.com" + path) + assert await driver._conversation_id_from_url() == expected + + +@pytest.mark.parametrize("backend_completion", [False, True], ids=["final-resolver", "mid-loop"]) +async def test_new_chat_waits_past_web_id_before_backend(monkeypatch, backend_completion): + driver = _make_driver() + _install_virtual_clock(monkeypatch) + state = {"phase1": 0, "phase2": 0} + urls = iter(["https://chatgpt.com/c/WEB:temporary-id"] * 4) + probe = _phase1_then_phase2_js(state, phase2_factory=lambda n: { + "text": "Answer", "md_text": "Answer", "html_len": 60, + "has_action": not backend_completion, + }) + + async def js(expr, timeout=15): + if "location.href" in expr: + return next(urls, "https://chatgpt.com/c/persisted-id") + return await probe(expr, timeout) + + driver._js_strict = js + driver.type_message = AsyncMock() + driver.click_send = AsyncMock() + driver._verify_send_acknowledged = AsyncMock(return_value=True) + driver._fetch_end_turn_for_turn = AsyncMock(return_value=TurnEndResult(status="matched")) + driver._fetch_text_for_turn = AsyncMock(return_value=TurnTextResult(status="matched", text="Answer")) + chunks = [c async for c in driver.send_and_stream("prompt")] + assert "".join(c.delta for c in chunks) == "Answer" + assert driver._current_conv_id == "persisted-id" + calls = driver._fetch_end_turn_for_turn.await_args_list + driver._fetch_text_for_turn.await_args_list + assert calls + assert all(call.args[0] == "persisted-id" for call in calls) + if backend_completion: + driver._fetch_end_turn_for_turn.assert_awaited() + + @pytest.mark.asyncio async def test_loop_resolves_conv_id_mid_loop(monkeypatch): """The completion poll loop must resolve conv_id_for_check from the URL diff --git a/tests/test_turn_anchor.py b/tests/test_turn_anchor.py index dd4a5707..55c95ad6 100644 --- a/tests/test_turn_anchor.py +++ b/tests/test_turn_anchor.py @@ -5,11 +5,14 @@ """ from __future__ import annotations +import string + import pytest from chatgpt_web2api.turn_anchor import ( TurnAnchor, TurnEndResult, + canonicalize_user_text, collapse_to_end_turn_status, normalize_text, select_end_turn_for_turn, @@ -386,3 +389,43 @@ def test_zero_width_stripped(self): def test_crlf_to_lf(self): assert normalize_text("a\r\nb\rc") == "a\nb\nc" + + +class TestCanonicalizeUserText: + @pytest.mark.parametrize("separator", ["", "\r\n", " \n"]) + def test_app_links_and_markdown_escapes(self, separator): + serialized = separator.join([ + "[$github](app://connector_76869538009648d5b282a4bb21c3d157)", + "[$hermes-memory-mcp-noauth](app://asdk_app_6ab28256fde8819197fa5c529d311f41)", + "[User]\nVerwende beide ausgewählten Apps.\n\n1\\. Test\nMEMORY=\\", + ]) + logical = "[User]\nVerwende beide ausgewählten Apps.\n\n1. Test\nMEMORY=" + assert canonicalize_user_text(serialized) == logical + assert user_text_matches_sent(serialized, logical) + assert not user_text_matches_sent(serialized, "fremder Prompt") + + @pytest.mark.parametrize("text", [ + "[docs](https://example.org)prompt", + "[$label](https://example.org)prompt", + "[label](app://connector_test)prompt", + "before [$label](app://connector_test)prompt", + "[$label](app://connector_test)", + ]) + def test_only_leading_app_prompt_links_are_removed(self, text): + assert not user_text_matches_sent(text, "prompt") + if not text.startswith("[$label](app://"): + assert canonicalize_user_text(text) == text + + def test_escaped_literal_app_link_is_not_removed(self): + logical = "[$label](app://connector_test)prompt" + serialized = r"\[$label\]\(app://connector_test\)prompt" + assert canonicalize_user_text(serialized) == logical + + def test_one_layer_of_ascii_punctuation_escapes(self): + assert canonicalize_user_text("".join("\\" + c for c in string.punctuation)) == string.punctuation + assert canonicalize_user_text(r"C:\temp\file\n \é") == r"C:\temp\file\n \é" + assert user_text_matches_sent(r"\\\*", r"\*") + assert not user_text_matches_sent(r"\*", r"\*") # logical text must not be decoded + + def test_existing_normalization_is_preserved(self): + assert canonicalize_user_text("[$app](app://id)Cafe\u0301\r\n1\\. Test\u200b ") == "Café\n1. Test"