"""The Exec panel's agent transcript, in WebKit (Wai's phone engine). The panel is what /cc's page was until 2026-09-29, so this file carries that page's streaming tests too. Every boundary is mocked at the network edge -- no sidecar, no subscription run, no LLM: - an Exec card tool leaves a one-line receipt; any other tool is ONE line with its output folded under it, and an empty or missing result says so; - a reply that resumes after a tool call opens its own bubble UNDER it; - the prompt sent is Wai's words alone -- the server adds the board; - the cursor blinks while a turn is in flight and is gone once it ends; - sending again interrupts the run in flight; - a busy sidecar says so; /help answers without a query. """ import json import pytest pytest.importorskip("playwright.sync_api") pytestmark = pytest.mark.browser _STUB_AUDIO = ("window.HosakaAudio={createPlayer:()=>({" "unlock(){},speak(){return Promise.resolve()}," "flush(){},setVolume(){},elapsed:()=>0," "audioDuration:()=>0,isUnlocked:()=>true," "gestureUnlocked:()=>true})};") def _frames(*events) -> str: return "".join(f"data: {json.dumps(e)}\n\n" for e in events) def _json(payload): return lambda r: r.fulfill(status=200, content_type="application/json", body=json.dumps(payload)) _CARD_TURN = _frames( {"type": "session", "sessionId": "s"}, {"type": "text", "text": "Checking."}, {"type": "tool", "name": "mcp__exec__create_card", "input": {"title": "Buy chalk"}}, {"type": "tool_result", "isError": False, "text": '{"ok":true,"title":"Buy chalk"}'}, {"type": "text", "text": "Added it."}, {"type": "done"}, ) _TOOL_TURN = _frames( {"type": "text", "text": "Getting real numbers."}, {"type": "tool", "name": "WebFetch", "input": {"url": "https://x/pricing"}}, {"type": "tool_result", "text": ""}, {"type": "text", "text": "**HG group coaching:** 90 min weekly."}, {"type": "tool", "name": "WebFetch", "input": {"url": "https://x/other"}}, {"type": "done", "turns": 2, "ms": 10}, ) _DONE = _frames({"type": "text", "text": "done thinking"}, {"type": "done", "turns": 1, "ms": 10}) @pytest.fixture def panel(browser, base_url, admin_headers): """/rd with the panel open and every sidecar-backed route mocked. `replies` is the list of SSE bodies, one per query; None HOLDS that query forever (a run in flight).""" contexts = [] def _open(*replies, history=()): ctx = browser.new_context(extra_http_headers=admin_headers) contexts.append(ctx) pg = ctx.new_page() # Voice OFF: the audio stub's clock never advances, so a reveal paced to # it would wait forever. Narration has its own suite # (test_exec_voice_browser.py); this one is about layout. pg.add_init_script("try { localStorage.setItem('exec.voice', '0'); } catch (e) {}") pg.route("**/vendor/marked-*.min.js", lambda r: r.fulfill( status=200, content_type="application/javascript", body="window.marked={use(){},parse:s=>s,Renderer:function(){}};")) pg.route("**/hosaka-audio.js*", lambda r: r.fulfill( status=200, content_type="application/javascript", body=_STUB_AUDIO)) pg.route("**/api/cc/exec-history", _json({"messages": list(history), "monitorTotal": 0})) pg.route("**/api/cc/health*", _json({"ok": True, "busy": False, "authed": True})) pg.route("**/api/cc/title*", _json({"sessionId": "t", "title": "test"})) pg.route("**/api/cc/limits*", _json({"ok": False})) pg.route("**/api/cc/sessions*", _json({"current": "t", "sessions": []})) sent = [] # Mocked like every other sidecar route: unmocked, a test's interrupt # would reach the LIVE sidecar and stop whatever Wai has running. pg.route("**/api/cc/interrupt", lambda r: ( sent.append({"interrupt": True}), r.fulfill(status=200, content_type="application/json", body='{"ok": true, "stopped": true}'))) def _query(route): sent.append(json.loads(route.request.post_data)) n = sum(1 for x in sent if "prompt" in x) body = replies[n - 1] if n <= len(replies) else _DONE if body is not None: route.fulfill(status=200, content_type="text/event-stream", body=body) pg.route("**/api/cc/query", _query) pg.goto(f"{base_url}/rd", wait_until="domcontentloaded") pg.wait_for_selector("#exec-bubble", timeout=5000) pg.click("#exec-bubble") pg.wait_for_selector("#exec-panel.open", timeout=4000) return pg, sent yield _open for c in contexts: try: c.close() except Exception: pass def _send(pg, text): pg.locator("#exec-minput").fill(text) pg.keyboard.press("Control+Enter") # bare Enter is a newline; 3 in a row send def _texts(pg, sel): return pg.eval_on_selector_all(sel, "els => els.map(e => e.textContent.trim())") def _idle(pg): """The turn is over. The user line lands before the turn starts (and the turn starts in the same tick), so a mocked turn that finishes instantly is still caught -- waiting for `execStreaming === true` first would miss it.""" pg.wait_for_function( "() => document.querySelector('#exec-term .msg.user') && execStreaming === false" " && !document.querySelector('#exec-term #exec-bc')", timeout=15000) def test_card_tool_is_a_receipt_and_the_prompt_is_only_her_words(panel): pg, sent = panel(_CARD_TURN) _send(pg, "buy chalk") _idle(pg) assert _texts(pg, "#exec-term .msg.sys") == ["[ card added: Buy chalk ]"] assert _texts(pg, "#exec-term .msg.tool") == [] bodies = _texts(pg, "#exec-term .msg.assistant") assert any("Checking." in b for b in bodies) and any("Added it." in b for b in bodies) assert sent == [{"prompt": "buy chalk", "images": [], "files": []}] def test_actions_are_one_line_each_and_no_turn_count(panel): """A turn lists what it DID, one dense line per action; the old "[ 2 turns ]" footnote counted rounds and said nothing about them.""" pg, _ = panel(_CARD_TURN) _send(pg, "buy chalk") _idle(pg) assert _texts(pg, "#exec-term .msg.act") == ["[ card added: Buy chalk ]"] assert not any("turns" in t for t in _texts(pg, "#exec-term .exec-receipt")) one_line = pg.eval_on_selector( "#exec-term .msg.act .msg-body", "e => e.getBoundingClientRect().height <= parseFloat(getComputedStyle(e).lineHeight) + 1") assert one_line def test_replay_shows_each_action(panel): """A reload keeps the actions: the sidecar's /history carries `action` rows (exec_panel passes them through), rendered as they were live.""" created = json.dumps({"ok": True, "id": "card-1", "title": "Drive Kayla to airport"}) pg, _ = panel(history=[ {"role": "user", "text": "drive kayla airport", "ts": "2026-10-07T21:00:00Z"}, {"role": "action", "name": "mcp__exec__create_card", "input": {"title": "x"}, "result": created, "isError": False, "ts": "2026-10-07T21:00:01Z"}, {"role": "action", "name": "Bash", "input": {"command": "ls"}, "result": "a\nb", "isError": False, "ts": "2026-10-07T21:00:02Z"}, {"role": "assistant", "text": "Logged.", "ts": "2026-10-07T21:00:03Z"}, ]) pg.wait_for_selector("#exec-term .msg.act", timeout=5000) assert _texts(pg, "#exec-term .msg.act") == [ "[ card added: Drive Kayla to airport ]", "Bash ls"] assert pg.eval_on_selector("#exec-term .msg.out", "e => e.hidden") def test_text_after_a_tool_call_opens_its_own_bubble(panel): """Appending into the same bubble ran two messages together and put the continuation ABOVE the tool line it came after.""" pg, _ = panel(_TOOL_TURN) _send(pg, "prices?") _idle(pg) got = pg.evaluate("""() => { const tools = [...document.querySelectorAll('#exec-term .msg.tool')]; const order = [...document.querySelectorAll('#exec-term .msg')] .map(n => n.className.replace(' open', '')) .filter(c => /assistant|tool|out/.test(c)); const foldable = tools.map(t => t.classList.contains('exec-fold')); tools.forEach(t => t.click()); return {order, foldable, outs: [...document.querySelectorAll('#exec-term .msg.out')].map(o => o.textContent.trim())}; }""") assert got["order"][:4] == ["msg assistant", "msg tool act exec-fold", "msg out", "msg assistant"] bodies = _texts(pg, "#exec-term .msg.assistant") assert bodies[0].startswith("Getting real numbers.") and "HG group" not in bodies[0] assert all(got["foldable"]) assert got["outs"] == ["[ no output ]", "[ no result returned ]"] def test_cursor_blinks_while_a_turn_is_in_flight(panel): pg, _ = panel(None) _send(pg, "hello") pg.wait_for_selector("#exec-term #exec-bc", timeout=5000) # Scrub the animation to both halves of its period: phase-exact, where # wall-clock samples on a throttled background page all land in one phase. blink = pg.evaluate("""() => { const el = document.querySelector('#exec-term #exec-bc'); const a = (el.getAnimations ? el.getAnimations() : [])[0]; if (!a) return { supported: false }; const before = a.playState; a.pause(); const at = (t) => { a.currentTime = t; return getComputedStyle(el).opacity; }; return { supported: true, before, lit: at(0), dark: at(750) }; }""") assert blink["supported"] and blink["before"] == "running", blink assert blink["lit"] == "1" and blink["dark"] == "0", blink def test_sending_again_interrupts_and_the_cursor_goes(panel): pg, sent = panel(None, _DONE) _send(pg, "first") pg.wait_for_selector("#exec-term #exec-bc", timeout=5000) _send(pg, "second") pg.wait_for_function( "() => [...document.querySelectorAll('#exec-term .msg.sys')]" ".some(e => e.textContent.includes('interrupted'))", timeout=8000) _idle(pg) assert any("second" in u for u in _texts(pg, "#exec-term .msg.user")) # Hanging up no longer stops a run (exec-resync.js), so the interrupt is # asked for by name, between the two queries. assert sent[0]["prompt"] == "first" and sent[1] == {"interrupt": True} assert sent[2]["prompt"] == "second" and len(sent) == 3 def _replay(pg, *messages): """What the transcript holds once the run is done: history replays it.""" pg.route("**/api/cc/exec-history", _json({"messages": list(messages), "monitorTotal": 0})) def _has(pg, sel, text): return (f"() => [...document.querySelectorAll('{sel}')]" f".some(e => e.textContent.includes({json.dumps(text)}))") def test_a_cut_stream_is_picked_back_up_from_the_transcript(panel): """The phone backgrounding the page cuts the stream, and the sidecar's run goes on (2026-10-14). No error: the page waits for the run to end and replays the whole reply from the transcript.""" pg, _ = panel(_frames({"type": "session", "sessionId": "s"}, {"type": "text", "text": "Half an ans"})) _replay(pg, {"role": "user", "text": "hello", "ts": "2026-10-14T15:00:00Z"}, {"role": "assistant", "text": "Half an answer, then the rest.", "ts": "2026-10-14T15:00:05Z"}) _send(pg, "hello") pg.wait_for_function(_has(pg, "#exec-term .msg.assistant", "then the rest"), timeout=8000) sys = _texts(pg, "#exec-term .msg.sys") assert not any("cut" in t or "interrupted" in t or "dropped" in t for t in sys), sys def test_a_stream_cut_before_the_sidecar_took_it_says_send_again(panel): """Cut while still queued for the slot, nothing runs: the message stays on screen and the line says to send it again.""" pg, _ = panel(_frames({"type": "waiting"})) _send(pg, "hello") pg.wait_for_function(_has(pg, "#exec-term .msg.sys", "send again"), timeout=8000) assert any("hello" in u for u in _texts(pg, "#exec-term .msg.user")) def test_reopened_mid_answer_the_reply_lands_when_the_run_ends(panel): """Closed and reopened while Exec was still answering: the page says so, then replays once the sidecar's slot frees.""" pg, _ = panel() calls = [] def _health(route): calls.append(1) busy = len(calls) == 1 route.fulfill(status=200, content_type="application/json", body=json.dumps({"ok": True, "busy": busy, "authed": True})) pg.route("**/api/cc/health*", _health) _replay(pg, {"role": "assistant", "text": "The finished reply.", "ts": "2026-10-14T15:00:05Z"}) pg.evaluate("execFollowRun()") pg.wait_for_function(_has(pg, "#exec-term .msg.assistant", "finished reply"), timeout=8000) def test_a_dead_stream_is_given_up_on_when_the_page_comes_back(panel): """iOS can suspend a page without the socket ever erroring. Back in the foreground, a sidecar that is free while the stream is still open means the stream is dead: the page stops waiting on it.""" pg, _ = panel(None) _send(pg, "hello") pg.wait_for_selector("#exec-term #exec-bc", timeout=5000) pg.evaluate("document.dispatchEvent(new Event('visibilitychange'))") pg.wait_for_function(_has(pg, "#exec-term .msg.sys", "connection dropped"), timeout=10000) _idle(pg) def test_busy_sidecar_is_said_out_loud(panel): pg, _ = panel(_frames({"type": "busy", "detail": "a run is already in flight"})) _send(pg, "hi") pg.wait_for_function( "() => [...document.querySelectorAll('#exec-term .msg.sys')].some(d => /busy/.test(d.textContent))", timeout=8000) def test_help_answers_without_a_query(panel): pg, sent = panel() _send(pg, "/help") pg.wait_for_selector("#exec-term .exec-help-row", timeout=4000) assert sent == [] def test_a_dropped_file_waits_as_a_chip_and_is_sent(panel): """Drag a file anywhere: the panel takes it (the browser must not navigate to it), it waits as a named chip, and the send carries it as {name, data} beside Wai's words -- the sidecar writes it into the sandbox.""" pg, sent = panel(_DONE) pg.evaluate("""() => { const dt = new DataTransfer(); dt.items.add(new File(['hello'], 'notes.txt', {type: 'text/plain'})); for (const t of ['dragover', 'drop']) window.dispatchEvent(new DragEvent(t, {dataTransfer: dt, bubbles: true, cancelable: true})); }""") pg.wait_for_selector("#exec-thumbs .exec-thumb-file", timeout=4000) _send(pg, "read this") _idle(pg) assert sent[0]["prompt"] == "read this" assert sent[0]["files"] == [{"name": "notes.txt", "data": "aGVsbG8="}] assert _texts(pg, "#exec-term .msg.user .exec-file") == ["notes.txt"]