tests/test_exec_panel_stream_browser.py
Rama master · réplica de lectura del repositorio original.
"""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"]