fix(agent): make preview and browser output parsers linear (#7003)

Six regexes over tool output backtracked polynomially on long whitespace
runs or repeated openers: the browser snapshot generic-line filter and
post-interaction split, the task-row and skill-bullet parsers, the email
"no results" check, and the Google /sorry/ interstitial check.

Rewrite each without overlapping quantifiers, keeping the same matches for
the text these tools emit. Move the /sorry/ check into
is_google_sorry_url() so it can be tested.

src/browser_observation.py

@@ -25,7 +25,7 @@ def compact_browser_observation(value, budget=8000):
                 # Snapshot text already contains labels and refs in DOM order.
                 # The refs mapping repeats them and buries menus in raw JSON.
                 lines = [line for line in snapshot.splitlines()
-                         if not re.fullmatch(r'\s*-?\s*generic(?:\s+\[ref=e\d+\])?:?\s*', line)]
+                         if not re.fullmatch(r'\s*(?:-\s*)?generic(?:\s+\[ref=e\d+\])?:?\s*', line)]
                 pages.append(('URL: ' + str(url) + '\n' if url else '') + '\n'.join(lines))
             elif url:
                 notices.append('URL: ' + str(url))
@@ -37,7 +37,8 @@ def compact_browser_observation(value, budget=8000):
                 parsed = json.loads(item)
             except (ValueError, TypeError):
                 # CLI status text precedes the JSON post-interaction state.
-                parts = re.split(r'\n\n\[(?:post-[^\]]+|page state after failed [^\]]+)\]\n', item)
+                # Labels are one line; stopping at \n keeps the split linear.
+                parts = re.split(r'\n\n\[(?:post-[^\]\n]+|page state after failed [^\]\n]+)\]\n', item)
                 if len(parts) > 1:
                     for part in parts:
                         visit(part)

src/clean_agent_preview.py

@@ -1229,7 +1229,7 @@ def tasks_terminal_response(raw, *, user_text='', max_items=20):
     text = str(payload or '').strip()
     rows = []
     for line in text.splitlines():
-        match = re.match(r'^\s*\d+\.\s+(.*?)\s+\(([^)]+)\)\s+—\s+(.+)$', line)
+        match = re.match(r'^\s*\d+\.\s(.*?)\s\(([^()]+)\)\s+—\s+(.+)$', line)
         if match:
             details = match[3].strip()
             rows.append({
@@ -1346,7 +1346,7 @@ def skills_terminal_response(raw, *, user_text='', max_items=20):
             continue
         match = re.match(r'^\s*-\s+\*\*([^*]+)\*\*(?:\s+\(([^)]+)\))?', line)
         if not match:
-            match = re.match(r'^\s*-\s+([^:]+?)(?:\s+\(([^)]+)\))?(?:\s*:|$)', line)
+            match = re.match(r'^\s*-\s+([^\s:][^:]*?)(?:\s\(([^()]+)\)(?:\s*:|$)|:|$)', line)
         if match and status:
             rows.append((status, match.group(1).strip(), (match.group(2) or '').strip()))
     if not rows:
@@ -5065,7 +5065,7 @@ def email_search_result_empty(value):
         try:
             decoded = json.loads(value)
         except (TypeError, ValueError):
-            return bool(re.fullmatch(r'\s*No emails matched [^\n]+\.?\s*', value, re.I))
+            return bool(re.fullmatch(r'\s*No emails matched [^\n]+(?:\n\s*)?', value, re.I))
         if decoded == value:
             return False
         value = decoded
@@ -5339,6 +5339,16 @@ def private_browser_effective_url(result):
     return urls[-1] if urls else ''
 
 
+def is_google_sorry_url(url):
+    """Recognize Google's rate-limit interstitial."""
+    # Matching the host separately keeps this linear; one combined pattern
+    # rescans the host from every dot.
+    return any(
+        re.fullmatch(r'(?:[^/]+\.)?google\.[^/]+', host[1], re.I)
+        for host in re.finditer(r'https?://([^/]+)/sorry/', url, re.I)
+    )
+
+
 def browser_observation_page_missing(raw):
     """Recognize a rendered error page, rather than an article mentioning 404."""
     text = str(raw or '')
@@ -8213,11 +8223,7 @@ async def stream_preview(*, endpoint_url, model, messages, headers, turn_contrac
                         browser_search_url = requested_browser_url or effective_browser_url
                         is_search_engine_navigation = contains_search_engine_navigation(
                             browser_search_url
-                        ) or bool(re.search(
-                            r'https?://(?:[^/]+\.)?google\.[^/]+/sorry/',
-                            effective_browser_url,
-                            re.I,
-                        ))
+                        ) or is_google_sorry_url(effective_browser_url)
                         has_native_search = any(
                             canonical(schema['function']['name']) == 'web_search'
                             for schema in offered

tests/test_redos_agent_preview_parsers.py

@@ -0,0 +1,141 @@
+"""Regression tests for py/polynomial-redos sinks in the agent preview parsers.
+
+Each pattern had overlapping quantifiers (adjacent `\\s` runs, or a lazy group
+that rescanned the rest of the line from every candidate) that backtrack
+polynomially when the tail fails. The rewrites keep the same matches for the
+text these tools emit. The old patterns took 5-20s on the flood inputs below;
+the new ones finish in milliseconds against a 4s budget.
+"""
+
+import time
+
+import pytest
+
+from src.browser_observation import compact_browser_observation
+from src.clean_agent_preview import (
+    email_search_result_empty,
+    is_google_sorry_url,
+    skills_terminal_response,
+    tasks_terminal_response,
+)
+
+_BUDGET_S = 4.0
+_DASH = chr(0x2014)  # the separator src/tools/system.py writes into task rows
+
+
+def _timed(fn, *args):
+    start = time.perf_counter()
+    result = fn(*args)
+    return result, time.perf_counter() - start
+
+
+# Browser snapshot: bare `generic` container lines are dropped, others kept.
+
+def test_generic_snapshot_lines_are_dropped():
+    snapshot = '- generic [ref=e1]:\n  - generic\n-generic:\n- generic button\n- button "Go" [ref=e2]'
+    output = compact_browser_observation({'snapshot': snapshot})
+    assert output == '- generic button\n- button "Go" [ref=e2]'
+
+
+def test_generic_filter_whitespace_flood_is_fast():
+    (output, dt) = _timed(compact_browser_observation, {'snapshot': '\t' * 120000 + '!'})
+    assert dt < _BUDGET_S, f'generic filter took {dt:.2f}s'
+    assert output.endswith('!')
+
+
+# Browser CLI text: status text is split from the post-interaction state.
+
+@pytest.mark.parametrize('label', ['post-click page state', 'page state after failed click'])
+def test_post_interaction_state_is_split(label):
+    raw = f'Clicked\n\n[{label}]\n{{"snapshot": "- link \\"A\\" [ref=e3]"}}'
+    assert compact_browser_observation(raw) == 'Clicked\n\n- link "A" [ref=e3]'
+
+
+@pytest.mark.parametrize('opener', ['\n\n[post-', '\n\n[page state after failed '])
+def test_post_interaction_split_flood_is_fast(opener):
+    (_, dt) = _timed(compact_browser_observation, 'x' + opener * 15000)
+    assert dt < _BUDGET_S, f'post-interaction split took {dt:.2f}s'
+
+
+# Task rows: `N. name (id) <dash> detail`, where the name may contain parentheses.
+
+def test_task_rows_parse():
+    raw = (f'1. Backup (nightly) (abc123) {_DASH} active, daily at 02:00\n'
+           f'2.  (e2) {_DASH} paused\n'
+           f'not a task row')
+    assert tasks_terminal_response(raw) == (
+        f'Tasks (2):\n- [Backup (nightly)](#task-abc123) {_DASH} active\n'
+        f'- [](#task-e2) {_DASH} paused'
+    )
+
+
+@pytest.mark.parametrize('evil', ['0.' + '\t' * 3000 + '!', '0.\t\t(' + '\t((' * 20000])
+def test_task_row_flood_is_fast(evil):
+    (output, dt) = _timed(tasks_terminal_response, evil)
+    assert dt < _BUDGET_S, f'task row parse took {dt:.2f}s'
+    assert output == evil.strip()
+
+
+# Skill inventory: `- name (category): summary` under a status heading.
+
+def test_skill_rows_parse():
+    raw = '## Published\n- pdf-tools (documents): Read PDFs\n- web\n## Drafts\n- notes  (writing)'
+    assert skills_terminal_response(raw) == (
+        'Skills (3):\n\n**Published**\n- [pdf-tools](#skill-pdf-tools) (documents)\n'
+        '- [web](#skill-web)\n\n**Drafts**\n- [notes](#skill-notes) (writing)'
+    )
+
+
+def test_nameless_skill_bullet_is_not_a_row():
+    raw = '## Published\n-  : x'
+    assert skills_terminal_response(raw) == raw
+
+
+def test_skill_row_flood_is_fast():
+    (_, dt) = _timed(skills_terminal_response, '## Published\n- \t9' + '\t\t' * 30000 + '!')
+    assert dt < _BUDGET_S, f'skill row parse took {dt:.2f}s'
+
+
+# Email search: the plain-text "no results" message.
+
+@pytest.mark.parametrize('value,expected', [
+    ('No emails matched from:bob.', True),
+    ('  no emails matched "invoice"\n\n', True),
+    ('No emails matched ', False),
+    ('No emails matched x\nmore', False),
+    ('Found 2 emails', False),
+])
+def test_email_no_results_message(value, expected):
+    assert email_search_result_empty(value) is expected
+
+
+def test_email_no_results_flood_is_fast():
+    (result, dt) = _timed(email_search_result_empty, 'No emails matched x' + '\t' * 60000 + '\n!')
+    assert dt < _BUDGET_S, f'email no-results check took {dt:.2f}s'
+    assert result is False
+
+
+# Google rate-limit interstitial.
+
+@pytest.mark.parametrize('url,expected', [
+    ('https://www.google.com/sorry/index?continue=x', True),
+    ('http://google.co.uk/sorry/', True),
+    ('HTTPS://IPV4.GOOGLE.COM/SORRY/', True),
+    ('https://example.com/sorry/?next=https://www.google.de/sorry/', True),
+    ('https://example.com/sorry/', False),
+    ('https://notgoogle.com/sorry/', False),
+    ('https://www.google.com/search?q=sorry', False),
+    ('', False),
+])
+def test_is_google_sorry_url(url, expected):
+    assert is_google_sorry_url(url) is expected
+
+
+@pytest.mark.parametrize('evil', [
+    'http://google.' + '..google.' * 50000,
+    'http://' + 'x.' * 200000 + 'google./sorry/',  # reaches the host match
+])
+def test_google_sorry_flood_is_fast(evil):
+    (result, dt) = _timed(is_google_sorry_url, evil)
+    assert dt < _BUDGET_S, f'google sorry check took {dt:.2f}s'
+    assert result is False