"""Static smoke guards for the shared interactive pane module. ``turnstone/shared_static/interactive.js`` is the per-workstream conversational ``Pane`` lifted out of ``ui/static/app.js`` (L-shell step 5a) so BOTH the standalone ``turnstone-server`` UI and the console L-shell can mount it. The load-bearing invariants of that extraction are pinned here — like the rest of the WebUI, the module has no JS test framework, so these are Python-side string-presence assertions that catch the silent one-line regression. """ from __future__ import annotations import re from pathlib import Path import pytest from tests._js_harness_helpers import strip_js_comments as _strip_comments _ROOT = Path(__file__).resolve().parent.parent _INTERACTIVE = _ROOT / "turnstone/shared_static/interactive.js" _COMPOSER = _ROOT / "turnstone/shared_static/composer.js" _AUTH = _ROOT / "turnstone/shared_static/auth.js" _APP = _ROOT / "turnstone/ui/static/app.js" _UI_INDEX = _ROOT / "turnstone/ui/static/index.html" def test_interactive_is_esm_imported_by_the_shell() -> None: """Real ES module: it ``export``s the factory the shell imports in BOTH deployments. Step 6 retired the window bridge (no window.InteractivePane) and the standalone HTML no longer script-tags interactive.js — shell.js pulls it via ``import``.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "export { Pane as InteractivePane, createInteractivePane };" in body assert "window.InteractivePane = Pane" not in body, ( "the window bridge is retired — the shell imports the factory (ESM)." ) html = _UI_INDEX.read_text(encoding="utf-8") assert "/shared/interactive.js" not in html, ( "the standalone HTML must NOT script-tag interactive.js — shell.js imports it." ) def test_pane_constructor_takes_transport_and_host_seam() -> None: """The constructor takes the ``(wsId, opts)`` seam: a transport ``base`` (the node-proxy prefix) and a ``host`` adapter for the few things only the surrounding shell knows. The old ``embedded`` flag is gone — every pane is L-shell-hosted since the step-6 fork collapse.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "constructor(wsId, opts) {" in body for field in ( "this._base = opts.base", "this._host = opts.host || INTERACTIVE_DEFAULT_HOST", ): assert field in body, f"missing constructor seam: {field!r}" assert "opts.embedded" not in body, ( "the embedded flag is retired — every pane is L-shell-hosted." ) def test_transport_urls_are_base_prefixed() -> None: """Every per-ws request is prefixed with ``this._base`` so a console pane proxies through ``/node/{id}`` (the LOCALITY invariant: an interactive session lives on a cluster node). A bare ``/v1/api/workstreams/`` URL would hit the console instead of the node and silently 404 / cross-talk.""" body = _strip_comments(_INTERACTIVE.read_text(encoding="utf-8")) # Collapse whitespace so a prettier line-wrap (``this._base +`` on the line # ABOVE the URL string) doesn't read as a bare URL: every workstream URL # must be preceded by ``this._base +``. collapsed = re.sub(r"\s+", " ", body) bad = re.findall(r'(? None: """The standalone split-pane chrome is GONE, not gated: no pane header (name / persona / state live in the tab + rail; the conversation owns the full pane height), no focus tracking, no split/close buttons. The dead ``!this._embedded`` branches referenced shell globals (setFocusedPane, splitPane, splitRoot…) that no longer exist anywhere — reaching them was a guaranteed ReferenceError, so their removal is a bugfix too.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "_embedded" not in body, "the embedded gate is retired (always-on)" assert 'className = "pane pane--embedded"' in body, ( "the pane root must carry pane--embedded unconditionally — " "interactive.css scopes the slim-chrome layout to it" ) for gone in ( "setFocusedPane", "showPaneContextMenu", "splitPane(", "splitRoot", "this.headerEl", '"pane-header"', '"pane-action-btn"', "updateWsName", ): assert gone not in body, f"retired split-pane symbol {gone!r} resurfaced" # The persona tag stays gone (the rail's INT/COORD vocabulary shows it). assert '"pane-persona-tag"' not in body assert '"INTERACTIVE"' not in body def test_factory_returns_lifecycle_over_node_proxy() -> None: """``createInteractivePane`` is the console factory (mirrors ``createCoordinatorPane``): it derives the node-proxy base from ``nodeId`` and returns the lifecycle controller the shell drives.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "function createInteractivePane(root, wsId, opts) {" in body assert '"/node/" + encodeURIComponent(opts.nodeId)' in body for hook in ("connect()", "deactivate()", "onLogin()", "destroy()"): assert hook in body, f"factory controller missing lifecycle hook {hook!r}" # Teardown must close the stream so a backgrounded pane can't leak an # upstream node connection. assert "pane.disconnectSSE();" in body def test_host_seam_routes_shell_couplings() -> None: """Every coupling to the surrounding shell goes through ``this._host`` — so the same Pane works standalone (real adapter) and console-embedded (no-op / Tier-1 adapter). No direct ``focusedPaneId`` / ``workstreams`` / consent badge reference survives in the module.""" body = _INTERACTIVE.read_text(encoding="utf-8") for call in ( "this._host.isFocused(this)", "this._host.onStreamError(this)", "this._host.warningTarget(this)", "this._host.onConsentDetected(", ): assert call in body, f"missing host seam call {call!r}" assert "getWsName" not in body, ( "getWsName left the host seam with the pane header — the tab + rail " "own the workstream name now." ) code = _strip_comments(body) # The classic split-pane shell globals must not leak into the module as # bare code references (URL path strings excepted, handled above). assert not re.search(r"(? None: """Step 6 retired the standalone's local split-pane construction: app.js no longer builds panes via window.InteractivePane / STANDALONE_HOST. Sessions open through the shared shell's PaneManager — openSessionPane delegates to openPane('interactive', wsId).""" app = _APP.read_text(encoding="utf-8") assert "STANDALONE_HOST" not in app, "the standalone host adapter is retired." assert "new window.InteractivePane(" not in app, ( "the standalone no longer constructs panes locally." ) start = app.index("function openSessionPane(wsId)") fn = app[start : start + 300] assert 'openPane("interactive", wsId)' in fn, ( "openSessionPane must open the session as a pane via the shell PaneManager." ) def test_approval_keyboard_shortcuts_wired() -> None: """The converged card advertises y/n/a (+Enter/Esc) kbd hints, so the pane must route those keys to resolveApproval when a pending approval is up — pane-owned on this.el (the fork collapse retired the old app.js global handler + getFocusedPane). Guards against the chips over-promising.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "if (!this.pendingApproval || !this.approvalBlockEl) return;" in body, ( "approval keydown must early-return unless a pending approval is up" ) assert "e.key.toLowerCase()" in body, "the y/n/a shortcut branch" assert ".conv-feedback" in body, ( "the feedback field uses the converged .conv-feedback, not the retired " ".ts-approval-feedback" ) def test_media_playback_lifted_and_pane_owned() -> None: """The media Play affordance is rendered by the pane (buildPlayButton / buildMediaEmbed), so its activation must live in the pane too — the old standalone wired a DOCUMENT-level click/keydown listener in app.js, which the console host never loaded (so the button was dead in console-hosted panes). The fix mirrors the approval-keydown pattern: a pane-owned listener on this.el, root-scoped via closest(".media-play-btn"). Pin both the lifted helpers and the pane wiring so the document-level regression can't silently come back.""" body = _INTERACTIVE.read_text(encoding="utf-8") # The lifted activation machinery now lives in the shared module. for fn in ( "function _loadHls(", "function _isHlsUrl(", "function _activatePlayer(", "function activateMediaPlayButton(", ): assert fn in body, f"media player helper must be lifted into the pane: {fn}" # The HLS vendor is fetched by absolute /shared/ URL (resolves in BOTH the # standalone server and the console, where /shared is mounted at the root). assert 'script.src = "/shared/hls-1.6.17/hls.min.js";' in body # Pane-owned + root-scoped — NOT a document-level delegated listener. assert 'this.el.addEventListener("click"' in body, ( "media play must be wired on this.el (pane-owned), not document" ) assert 'e.target.closest(".media-play-btn")' in body, ( "the play handler must be root-scoped via closest, not a document-wide id" ) assert "activateMediaPlayButton(btn)" in body collapsed = _strip_comments(body) assert 'document.addEventListener("click"' not in collapsed, ( "the pane must not register a document-level click delegate — that is " "the standalone regression that left console panes dead" ) def test_controller_terminal_dead_state() -> None: """Lifecycle round 2: the console controller must STOP reconnect-polling a session that is gone (closed / evicted / node restarted) — three consecutive CLOSED recovery beats → give up: stream closed, status bar terminal, ``opts.onDead()`` fired once. A successful stream open resets the counter (the new host.onStreamOpen seam). ``isDead()`` / ``markDead()`` / ``base`` are the shell's revive surface; a dead controller also ignores the login re-arm (recovery may need a DIFFERENT node — the shell's revive owns it).""" body = _INTERACTIVE.read_text(encoding="utf-8") # The give-up ladder. assert "let dead = false;" in body and "let failCount = 0;" in body assert "const giveUp = function () {" in body assert "failCount += 1;" in body and "if (failCount >= 3) giveUp();" in body assert 'pane._sbTokens.textContent = "Disconnected"' in body, ( "the terminal state must be worded distinctly from the transient Reconnecting…" ) assert "opts.onDead" in body, "the shell must hear about the give-up" # The reset seam: Pane.connectSSE onopen → host.onStreamOpen → failCount = 0. assert "this._host.onStreamOpen(this)" in body assert "onStreamOpen() {}" in body, "the default host must carry the no-op" # The shell-facing surface. assert "isDead()" in body and "markDead: giveUp," in body assert "base: base," in body, "the controller must expose its transport base" # Dead controllers don't reconnect on re-auth. assert "if (connected && !dead) pane._loadHistoryThenConnect(wsId);" in body def test_stream_pipeline_is_wedge_proof() -> None: """Long-session hardening (perf audit P0): the SSE pipeline must not be able to permanently wedge the pane. ``onmessage`` guards BOTH the ``JSON.parse`` and the ``handleEvent`` dispatch (an exception escaping it doesn't close the EventSource, so an unguarded throw left the streaming refs poisoned for the rest of the session), and ``stream_end`` resets the segment refs BEFORE the finalize render, with a plain-text fallback — with the old order a finalize throw skipped the clears and every later delta painted into the dead segment.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "dropping malformed SSE frame" in body assert "handleEvent failed for" in body case = body.index('case "stream_end"') seg = body[case : body.index("break;", case)] clears = seg.index("this.currentAssistantBodyEl = null;") finalize = seg.index("streamingRenderFinalize(") assert clears < finalize, ( "stream_end must clear segment refs BEFORE finalize — the old " "finalize-first order wedged all later assistant output on a throw." ) assert "doneBodyEl.textContent = doneBuffer;" in seg def test_rebuild_quiesces_live_events_and_releases_agent_tracking() -> None: """clear_ui re-render race (perf audit P0): live SSE events painted between the history snapshot and ``replaceChildren()`` were wiped with no redelivery, and streaming refs kept pointing at detached nodes. Pinned: the quiesce queue sits on the handleEvent hot path, the live-stream re-render trigger (clear_ui) arms it, ``replayHistory`` resets the streaming refs and clears the agent-card/orphan maps (the detached-DOM retention leak), and the mid-stream guard covers the reasoning bubble. (replay_truncated no longer arms the quiesce — it tears the stream down and runs the full fresh-connect flow instead; see test_truncated_resync_is_full_fresh_connect.)""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "this._replayQueue.events.push(evt);" in body assert body.count("this._beginReplayQuiesce(") >= 1, ( "clear_ui must arm the quiesce (it re-renders on a LIVE stream)" ) assert "!this.currentAssistantEl && !this.currentReasoningEl" in body replay = body.index("replayHistory(messages) {") seg = body[replay : replay + 2600] for line in ( "this._resetStreamingRefs();", "this._clearAgentTracking();", ): assert line in seg, f"replayHistory must reset: {line!r}" assert "this._agentCards.clear();" in body # Review-hardened lifecycle: the card entry SURVIVES the terminal # tool_result (a late child event finding no Map entry would rebuild a # duplicate empty card beside the finished one), and transport-only # reconnects preserve the maps + any armed quiesce queue — clearing them # in disconnectSSE duplicated cards and dropped buffered orphan steps on # every transient stream blip. Full-reload cleanup lives in # _loadHistoryThenConnect; terminal cleanup in the factory's destroy(). assert "this._agentCards.delete(callId);" not in body disc = body.index("disconnectSSE() {") disc_seg = body[ disc : body.index("_loadHistoryThenConnect(wsId, manualAttempt = false) {", disc) ] assert "this._clearAgentTracking();" not in disc_seg assert "this._replayQueue = null;" not in disc_seg load = body.index("_loadHistoryThenConnect(wsId, manualAttempt = false) {") load_seg = body[load : body.index("async _refetchHistory(", load)] assert "this._clearAgentTracking();" in load_seg assert "this._replayQueue = null;" in load_seg # A mid-stream replay_truncated DEFERS the re-sync (flag consumed on the # idle edge) instead of dropping it — skipping left the lost-event gap # unrepaired for the rest of the session. assert "this._pendingTruncatedResync = true;" in body # The refetch FAILURE branch is a DOM/ref no-op (#890): the full # guard-before-wipe contract — failure branch, clear_ui, the # resumability-gated reload reset, and the affordance gates — is # pinned in test_interactive_refetch_failure_preserves_the_pane. def test_interactive_refetch_failure_preserves_the_pane() -> None: """A FAILED /history fetch must leave the transcript, streaming refs, and repair-intent state untouched on EVERY re-render route (#890 — the interactive mirror of coord's #882 G3 guard-before-wipe, pinned there by test_coordinator_refetch_failure_preserves_the_pane). The wipe + resets live in replayHistory, reached only on success: - the clear_ui case must NOT pre-wipe (pre-#890 a failed fetch during a rewind blanked the highest-traffic pane on a live stream); - the _refetchHistory failure branch must be empty except for the quiesce release (no showEmptyState — the stray hint below stale content was the resync-route wart; no ref reset); - the failed-first-paint placeholder survives ONLY via the factory connect() pre-seed, which must stay ahead of the load call; - _loadHistoryThenConnect must reset streaming refs UNLESS a truncation resync is armed (the armed route's cursor replay resumes the mid-jitter bubble; every non-resumable flavor — ws switch, first paint, idle edge, unarmed re-auth reload — resets, or a stale ref would concatenate the next turn into the old bubble); - the row-level mutating affordances (rewind / edit / edit-and-resend) must gate on the _historyStale latch alongside busy — the latch spans clear_ui arrival through the next SUCCESSFUL render, covering the fetch window AND the failed-fetch aftermath (a quiesce-based gate reopened on the failure exit and let a second rewind over-rewind off the stale DOM); - the latch heals TRANSPORT-FREE: one turn-free bounded retry from the clear_ui .then, and a quiesced same-token refetch at organic idle edges as the double-failure backstop. Neither may touch the stream — a reload's fresh reconnect draws the server's synthetic state_change:idle back into the backstop's own trigger (the round-5 zero-backoff reconnect storm). """ body = _INTERACTIVE.read_text(encoding="utf-8") # clear_ui: no pre-wipe before the fetch, and the quiesce must be # armed BEFORE the fetch (queued live events land in the rebuilt — # or, on failure, the stale-but-real — pane, never the void). cl = body.index('case "clear_ui":') cl_seg = body[cl : body.index("break;", cl)] assert "this._refetchHistory(this.wsId, token)" in cl_seg assert "this.messagesEl.replaceChildren();" not in cl_seg, ( "clear_ui must not pre-wipe the transcript (#890)" ) assert "this._resetStreamingRefs();" not in cl_seg, ( "clear_ui must not pre-reset streaming refs (#890)" ) assert cl_seg.index("this._beginReplayQuiesce(token);") < cl_seg.index( "this._refetchHistory(" ), "clear_ui must arm the quiesce BEFORE the fetch" # Render-time cursor-safety gate (#900): a SEEDLESS render commits # /history's as-of-now truth without advancing _lastEventId, so # committing it while the transport is down strands the cursor below # the rows just painted and the next connect's replay paints them # again. The gate must be seedless-SCOPED — _loadHistoryThenConnect # disconnects first, so its evtSource is null for its whole fetch and # gating it would break every first paint, ws switch and resync. ref = body.index("async _refetchHistory(") ref_seg = body[ref : body.index("\n _beginReplayQuiesce(token) {", ref)] gate = ref_seg.index("const cursorSafe =") gate_seg = ref_seg[gate : ref_seg.index(";", gate)] assert "seedCursor ||" in gate_seg, ( "the render-time gate must exempt seeded (disconnect-first) loads (#900)" ) assert "this.evtSource.readyState === EventSource.OPEN" in gate_seg # Connection-generation term (#900 r2): readyState alone cannot see a # transport that DROPPED and finished RE-ESTABLISHING inside the await — # it reads OPEN either way, while the redial re-presented the frozen # cursor and the quiesce buffered the replay the render would duplicate. # Object identity is not a substitute: a NATIVE reconnect reuses the same # EventSource, so only a counter can see it. assert "this._connectEpoch === epoch" in gate_seg, ( "the seedless render must require an unchanged stream generation (#900)" ) assert "if (data && cursorSafe) {" in ref_seg, ( "the seedless render must be gated on a live cursor (#900)" ) # Captured at DISPATCH, not read live at render time — a live read would # compare the epoch against itself and the term would be vacuous. assert "const epoch = this._connectEpoch;" in ref_seg, ( "the gate must compare against the generation captured at dispatch (#900)" ) assert ref_seg.index("const epoch = this._connectEpoch;") < ref_seg.index("await authFetch("), ( "the generation must be captured BEFORE the fetch await (#900)" ) # The bump belongs to onopen and NOWHERE else. A native auto-reconnect # calls neither connectSSE nor disconnectSSE, so those two are blind to # the very case the term exists for; connectSSE would additionally # FALSE-bump on its document.hidden early return, which establishes no # stream. Absence from both is therefore load-bearing, not incidental. assert "this._connectEpoch = 0;" in body, ( "the generation must be initialised — undefined makes every compare " "NaN-false and silently declines every seedless render (#900)" ) onopen = body.index("this.evtSource.onopen = () => {") onopen_seg = body[onopen : body.index("\n };", onopen)] assert "this._connectEpoch += 1;" in onopen_seg, ( "the stream generation must be bumped in onopen — the only site that " "fires for a NATIVE auto-reconnect (#900)" ) conn = body.index("connectSSE(wsId) {") assert "_connectEpoch" not in _strip_comments(body[conn:onopen]), ( "connectSSE must NOT bump the generation — it would false-bump on the " "document.hidden early return, which establishes no stream (#900)" ) dis = re.search(r"\n disconnectSSE\(\) \{(.*?)\n \}\n", body, re.S) assert dis is not None, "disconnectSSE not found" # Comment-stripped: the method's inventory comment NAMES both fields as # deliberately-not-cleared, which is the ruling — assert on code only. dis_code = _strip_comments(dis.group(1)) assert "_connectEpoch" not in dis_code, ( "disconnectSSE must NOT bump the generation — a teardown is decided " "by the presence term, and a re-establish by the next onopen (#900)" ) # THE load-bearing negative: disconnectSSE deliberately does NOT cancel # the clear_ui retry (transport-only redials keep the pending heal # intent). That is exactly why the retry can fire against a dead # transport, which is what its OPEN fire-guard term exists to handle — # so a "symmetry" cleanup adding the cancel here would silently make # that guard unreachable and E5's detector vacuous, with nothing else # failing. Coord pins the same invariant (test_coordinator_page.py). assert "_staleRetryTimer" not in dis_code, ( "disconnectSSE must NOT cancel the clear_ui failure retry — " "transport-only reconnects keep the pending heal intent (#890/#900)" ) # Failure branch: quiesce release only. fail = body.index("Failed fetch = DOM + ref + repair-intent no-op") fail_seg = body[fail : fail + 1100] assert "this._endReplayQuiesce(token);" in fail_seg assert "this.showEmptyState();" not in fail_seg, ( "a failed fetch must not append an empty-state hint (#890)" ) assert "this._resetStreamingRefs();" not in fail_seg # The failure branch must RESOLVE, never throw/reject: the clear_ui # .then dispatches the queued edit-and-resend after a failed fetch # too (the rewind already committed server-side) — a throw here # would route to .catch and strand the resend. assert "throw " not in fail_seg assert "Promise.reject" not in fail_seg # Success path still owns the wipe + resets. replay = body.index("replayHistory(messages) {") replay_seg = body[replay : replay + 2600] assert "this.messagesEl.replaceChildren();" in replay_seg assert "this._resetStreamingRefs();" in replay_seg # Factory pre-seed ahead of the load — the failed-first-paint # placeholder's only remaining producer. conn = body.index("Load-bearing pre-seed (#890)") conn_seg = body[conn : conn + 600] seed = conn_seg.index("pane.showEmptyState();") load = conn_seg.index("pane._loadHistoryThenConnect(wsId);") assert seed < load, "connect() must seed the empty-state BEFORE the fetch" # Resumability-gated ref reset in _loadHistoryThenConnect: refs # survive a reload only when an armed truncation cursor lets the # reconnect resume into them; every other flavor (ws switch, # unarmed re-auth reload) resets. lh = body.index("_loadHistoryThenConnect(wsId, manualAttempt = false) {") lh_seg = body[lh : body.index("async _refetchHistory(", lh)] assert "if (this._truncatedFromCursor == null) this._resetStreamingRefs();" in lh_seg, ( "reload must reset streaming refs unless a truncation resync is armed (#890)" ) # The mutating row affordances gate on the staleness latch alongside # busy — scoped per method so a gate migrating off one affordance # cannot hide inside a file-wide occurrence count. for sig in ( "_rewindToMessage(msgEl) {", "_startEdit(msgEl, originalText) {", "_editAndResend(msgEl, newText) {", ): m = body.index(sig) assert "if (this.busy || this._historyStale) return;" in body[m : m + 400], ( f"missing busy || _historyStale gate in {sig!r} (#890)" ) # Latch lifecycle: set at clear_ui arrival BEFORE the quiesce is # armed (so no event can interleave between restructure-signal and # gate-close); cleared ONLY in replayHistory's render, which also # cancels the pending failure retry; the retry is scheduled from # the clear_ui .then (bounded by construction — a retry's own # failure cannot re-schedule) and fire-time-gated turn-free; the # idle edge backstops a double failure via else-if BEHIND the # truncated branch (whose own render heals the latch too). assert cl_seg.index("this._historyStale = true;") < cl_seg.index( "this._beginReplayQuiesce(token);" ), "the staleness latch must be set before the quiesce is armed" assert "this._historyStale = false;" in replay_seg, ( "replayHistory must clear the staleness latch (its only clear site)" ) assert "clearTimeout(this._staleRetryTimer);" in replay_seg assert "this._staleRetryTimer = setTimeout(" in cl_seg assert "!this.currentAssistantEl &&" in cl_seg, ( "the clear_ui failure retry must be turn-free-gated" ) assert "} else if (this._historyStale && !this._replayQueue) {" in body, ( "the idle edge must backstop the latch behind the truncated branch, " "skipping edges with a quiesced fetch already in flight" ) # The backstop must defer to the current event-backlog tail, then remain # TRANSPORT-FREE: a quiesced same-token refetch, never # _loadHistoryThenConnect. A synchronous refetch here lets replay_ok's # leading synthetic idle split the canonical backlog around a /history # repaint; a transport reload draws another synthetic idle into this # branch's own trigger (the round-5 storm). backstop = body.index("} else if (this._historyStale && !this._replayQueue) {") backstop_seg = body[backstop : body.index("// Only steal focus", backstop)] assert "this._deferStaleHistoryBackstop();" in backstop_seg, ( "the idle edge must defer its stale heal to the event-backlog tail" ) assert "this._loadHistoryThenConnect(" not in backstop_seg, ( "the staleness backstop must never touch the transport (#890 r5)" ) deferred = body.index("_deferStaleHistoryBackstop() {") deferred_seg = body[deferred : body.index("\n _clearAgentTracking() {", deferred)] assert "queueMicrotask(() => {" in deferred_seg assert "this._beginReplayQuiesce(staleToken);" in deferred_seg assert "this._refetchHistory(staleWs, staleToken);" in deferred_seg, ( "the deferred staleness backstop must heal via a quiesced REST refetch" ) assert "this._loadHistoryThenConnect(" not in deferred_seg, ( "the deferred staleness backstop must remain transport-free (#890 r5)" ) # The retry yields to an in-flight quiesce (no same-token stomp). retry = cl_seg.index("this._staleRetryTimer = setTimeout(") assert "!this._replayQueue &&" in cl_seg[retry : retry + 700], ( "the clear_ui retry must yield to an in-flight quiesced fetch" ) # ...and must not fire against a DOWN transport (#900): disconnectSSE # deliberately keeps the timer armed, so a hidden tab / degraded # cooldown / native redial can hold the fire. A seedless refetch then # paints rows the frozen _lastEventId still sits below, and the next # connect replays that slice on top. OPEN, not merely present — a # CONNECTING source has a frozen cursor with a replay pending. assert "this.evtSource.readyState === EventSource.OPEN" in cl_seg[retry:], ( "the clear_ui retry must require a live stream at fire time (#900)" ) # The delay is floor + spread, both from the SHARED module: one clear_ui # reaches every listener on the ws, so an un-spread retry re-fetches in # lockstep across tabs, and the floor is what the e2e non-occurrence # windows size themselves on. Coord carries the identical expression — # this pin is what keeps the two from drifting. assert "STALE_RETRY_BASE_MS + Math.random() * STALE_RETRY_JITTER_MS" in cl_seg[retry:], ( "the clear_ui retry must keep the shared floor + jitter (#900)" ) # Terminal teardown must invalidate in-flight loads AND cancel the # failure retry. The token bump (#900) is the chokepoint: without it # destroy() left _loadHistoryThenConnect's .finally free to reopen an # EventSource on the detached pane (re-registering the document-level # visibilitychange listener destroy just removed), a settling # _refetchHistory free to replayHistory into detached DOM, and the # clear_ui .then free to RE-ARM the timer destroy had cancelled. The # clearTimeout stays because the timer may already be armed and a # timer into a destroyed pane must be dead, not merely inert. # (disconnectSSE deliberately does NOT cancel it — transport-only # reconnects keep the pending heal intent.) dest = body.index("destroy() {") dest_seg = body[dest : body.index("\n },", dest)] assert "clearTimeout(pane._staleRetryTimer);" in dest_seg, ( "destroy() must cancel the clear_ui failure retry (#890)" ) assert "pane._historyLoadToken = (pane._historyLoadToken || 0) + 1;" in dest_seg, ( "destroy() must bump the load token to invalidate in-flight loads (#900)" ) # Same rule on the other terminal path: giveUp() already bumped the # token, so the timer is inert there — but inert is not dead. give = body.index("const giveUp = function () {") give_seg = body[give : body.index("\n };", give)] assert "pane._historyLoadToken = (pane._historyLoadToken || 0) + 1;" in give_seg assert "clearTimeout(pane._staleRetryTimer);" in give_seg, ( "giveUp() must cancel the clear_ui failure retry too (#900 symmetry)" ) def test_per_token_hot_path_avoids_container_scans() -> None: """P1 (perf audit): per-token work must stay O(1) in transcript length. The thinking indicator is an instance ref (the class-selector miss walked the whole transcript on EVERY content/reasoning delta); near-bottom state comes from the passive scroll listener instead of a forced-layout geometry read per event; the scroll pin is rAF-coalesced; per-tool row/stream lookups resolve through the self-healing caches.""" body = _INTERACTIVE.read_text(encoding="utf-8") stripped = _strip_comments(body) assert 'querySelector(".thinking-indicator")' not in stripped, ( "thinking indicator must use the instance ref, not a container scan" ) assert "this._thinkingEl" in body near = body.index("isNearBottom() {") assert "return this._nearBottom;" in body[near : near + 700] assert "passive: true" in body # The rAF pin re-checks the flag AT FIRE TIME (a user scroll landing in # the schedule→rAF window must win over a stale pin), with force # requests latched across the coalescing window; resizes re-derive the # flag via ResizeObserver since they move the bottom without a scroll. assert "this._scrollPinForce = false;" in body assert "ResizeObserver" in body for helper in ("_toolRow(callId) {", "_streamEl(callId) {"): assert helper in body, f"missing lookup-cache helper: {helper!r}" # -- Shared-workstream cross-user send gate ----------------------------------- # # The UX complement to the server-side CrossUserInterjectionError (a 409): while # another participant's turn is in flight, this viewer's send button is disabled # so they can't interject under the initiator's credentials / be misattributed. # The wiring spans three modules; these string-presence guards catch the silent # one-line regression the way the rest of this file does (no JS test framework). def test_composer_exposes_hard_send_block() -> None: """The composer has an independent hard-block axis, reconciled with busy, so a caller can disable send even in queueWhileBusy (queue) mode.""" body = _COMPOSER.read_text(encoding="utf-8") assert "Composer.prototype.setSendBlocked = function" in body assert "Composer.prototype._reconcileDisabled = function" in body assert "this._sendBlocked = false;" in body # setBusy must route the disabled write through the reconciler (not clobber # the block with a direct sendBtn.disabled assignment). stripped = _strip_comments(body) setbusy = stripped.index("Composer.prototype.setBusy = function") setbusy_end = stripped.index("Composer.prototype._reconcileDisabled") assert "this._reconcileDisabled();" in stripped[setbusy:setbusy_end] assert "this.sendBtn.disabled =" not in stripped[setbusy:setbusy_end], ( "setBusy must not write sendBtn.disabled directly — reconcile owns it" ) def test_auth_retains_user_id_for_gate() -> None: """whoami's opaque user_id is retained (separately from the display username) so the pane can compare it against the acting-user id.""" body = _AUTH.read_text(encoding="utf-8") assert 'sessionStorage.setItem("ts.user_id", data.user_id);' in body assert 'sessionStorage.removeItem("ts.user_id");' in body def test_pane_gates_send_on_cross_user_busy() -> None: """The pane tracks the acting user from state_change, compares it against the viewer's own id, and blocks send while another participant is busy.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "_reconcileSendBlock() {" in body # tracks the acting user from the state_change event... assert "this._actingUserId = evt.acting_user_id;" in body assert "this._actingUserId = null;" in body # cleared when the turn settles # ...compares against the viewer's own id from /whoami... assert 'sessionStorage.getItem("ts.user_id")' in body assert "this._actingUserId !== me" in body # ...and drives the composer's hard block, re-run on every busy edge. assert "this.composer.setSendBlocked(" in body stripped = _strip_comments(body) # The shared stripper is offset-preserving (comments become spaces), so # slice to the method's real closing brace instead of a fixed byte # window a comment edit could silently outgrow. setbusy = stripped.index("setBusy(b, source) {") setbusy_end = stripped.index("\n }", setbusy) assert "this._reconcileSendBlock();" in stripped[setbusy:setbusy_end] def test_pane_handles_cross_user_409() -> None: """The reactive fallback: a 409 (button not yet disabled) surfaces a clean message, not the generic 'Connection error' catch. Both the fetch-stage conversion and the status ARM now live in the shared helper (composer_queue.postAndSettleSend / settleSendResponse), so the pane owns only the request — every send flow it has reaches the conversion by construction instead of re-deriving it (the edit-and-resend flow used to lack it and reported a refused resend as a connection error).""" helper = (_ROOT / "turnstone/shared_static/composer_queue.js").read_text(encoding="utf-8") assert "response.status === 409" in helper assert 'status: "cross_user_interjection"' in helper assert 'status === "cross_user_interjection"' in helper body = _INTERACTIVE.read_text(encoding="utf-8") assert "cross_user_interjection" not in body, ( "the 409 conversion must not be re-derived per pane" ) assert body.count("postAndSettleSend(") == 2, "composer send + edit-and-resend" def test_sync_approval_state_prunes_orphan_cycles() -> None: """``_syncApprovalState`` prunes cycles whose block elements are no longer in the living DOM (``.isConnected === false``). This covers the rare case where an ``approve_request`` event is processed between a DOM wipe (``clear_ui`` / ``replay_truncated`` / ``replaceChildren``) and the refetch-restore — the cycle card lives in a detached subtree, the matching ``approval_resolved`` never arrives, and the send button stays disabled forever without this guard. The pin guards against a future refactor that drops the orphan prune but doesn't otherwise break ``_syncApprovalState``.""" body = _INTERACTIVE.read_text(encoding="utf-8") fn_start = body.index("_syncApprovalState() {") assert "entry.blockEls && !entry.blockEls.some((el) => el.isConnected)" in body, ( "orphan pruning must check .isConnected on block elements" ) tail = body[fn_start : body.index("_oldestCycleId()", fn_start)] assert "this.approvalCycles.delete(cid);" in tail, ( "orphan pruning must delete the cycle from the Map" ) # --------------------------------------------------------------------------- # SSE overflow recovery + close-on-hide (fast-stream corruption fixes) # --------------------------------------------------------------------------- def test_stream_overflow_case_counts_and_rate_limits() -> None: """The server closes an overflowed stream after an id-less ``stream_overflow`` frame; the pane must count it (field instrumentation for the drop-vs-render-wedge diagnosis) and route it through the reconnect limiter so a persistently slow consumer trips the degraded catch-up instead of churning reconnect/replay cycles.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert 'case "stream_overflow":' in body assert "this._noteStreamOverflow();" in body # The health object carries all four field-forensics counters; the # truncated-resync counter distinguishes the replay-window class from # the dropped-events class (overflows). health = re.search(r"_streamHealth = \{(.*?)\};", body, re.S) assert health is not None, "_streamHealth initializer not found" for field in ("overflows: 0", "renderThrows: 0", "malformedFrames: 0", "truncatedGaps: 0"): assert field in health.group(1), f"_streamHealth must init {field!r}" # Both wedge-class catch sites increment the render-throw counter, # and the malformed-frame drop counts too — the C-OVERDETERMINED # instrumentation that tells drops apart from wedges in the field. assert body.count("this._streamHealth.renderThrows += 1;") == 2 assert "this._streamHealth.malformedFrames += 1;" in body assert "this._streamHealth.overflows += 1;" in body def test_truncated_resync_is_full_fresh_connect_with_churn_limit() -> None: """replay_truncated = the stream admitted losing events past recovery. The pane must treat the connection as DEAD: run the full fresh-connect flow (``_loadHistoryThenConnect`` — disconnect first, REST /history, adopt the resume cursor, reconnect) rather than an in-place refetch. The in-place shape discarded the /history cursor while /history TRIMS the trailing in-flight turn whenever it returns one — a mid-run truncation wiped the executing turn (task cards included) with no redelivery, and sub-agent children then escaped to top-level rows after the orphan grace (the 2026-07 field reports). Pinned: (1) the immediate branch routes through ``_noteTruncatedResync()`` and SKIPS the resync when the limiter just tripped — the cooldown disconnected the stream, and the flow's ``.finally`` reconnect would defeat it; on no-trip it SCHEDULES the fresh connect behind the herd-spreading jitter rather than starting it inline (a node restart makes every stale-cursor tab resync inside the EventSource retry window, and the per-tab limiter cannot see a cross-tab herd); (2) the mid-stream guard still DEFERS (detachable bubble); (3) the idle-edge consumption runs the same fresh-connect flow, unjittered (staggered by turn-settle timing); (4) the limiter feeds the SAME churn window as overflow closes and enters degraded catch-up on trip; (5) neither truncated branch arms the quiesce or calls the in-place ``_refetchHistory`` — the old shape must not come back; (6) every ``truncatedGaps`` bump routes through the one increment+log step (``_recordTruncatedGap``), whose class-of-event wording never asserts a resync a trip may skip; (7) the scheduler nulls its handle before loading and ``disconnectSSE`` cancels a pending one, so a torn-down stream's resync can't fire against the pane's next workstream; (8) the gap-repair guarantee is interleaving-proof via the connect chokepoint; (9) a same-ws load supersession must not strand a queued edit-and-resend.""" body = _INTERACTIVE.read_text(encoding="utf-8") trunc = re.search(r'case "replay_truncated":(.*?)break;', body, re.S) assert trunc is not None, "replay_truncated case not found" t = trunc.group(1) # (1) immediate branch: limiter check gates the JITTERED fresh connect. assert "if (!this._noteTruncatedResync())" in t assert "this._scheduleTruncatedResync();" in t # (2) mid-stream defer unchanged. assert "!this.currentAssistantEl && !this.currentReasoningEl" in t assert "this._pendingTruncatedResync = true;" in t # (5) the old in-place shape must not come back. assert "_beginReplayQuiesce" not in t assert "_refetchHistory" not in t # (3) idle-edge consumption: consume the latch, record, fresh-connect. idle = re.search( r"if \(this\._pendingTruncatedResync\) \{(.*?)\n \}", body, re.S, ) assert idle is not None, "idle-edge truncated consumption not found" i = idle.group(1) assert "this._pendingTruncatedResync = false;" in i assert "this._recordTruncatedGap();" in i assert "this._loadHistoryThenConnect(this.wsId);" in i assert "_refetchHistory" not in i # (4) the limiter method: records the gap, then delegates churn # accounting to the ONE shared trip step (also used by # _noteStreamOverflow) so the trip parameters cannot silently diverge # between the overflow and truncated classes. note = re.search(r"_noteTruncatedResync\(\)\s*\{(.*?)\n \}", body, re.S) assert note is not None, "_noteTruncatedResync method not found" n = note.group(1) assert "this._recordTruncatedGap();" in n assert "return this._recordChurnAndMaybeTrip();" in n churn = re.search(r"_recordChurnAndMaybeTrip\(\)\s*\{(.*?)\n \}", body, re.S) assert churn is not None, "_recordChurnAndMaybeTrip method not found" c = churn.group(1) assert "this._overflowTimes.push(now);" in c assert "overflowWindowTripped(" in c assert "this._enterDegradedCatchup();" in c assert "return true;" in c assert "return false;" in c over = re.search(r"_noteStreamOverflow\(\)\s*\{(.*?)\n \}", body, re.S) assert over is not None, "_noteStreamOverflow method not found" assert "this._recordChurnAndMaybeTrip();" in over.group(1), ( "overflow closes must feed the same shared churn step" ) # (6) the one increment+log step: every bump carries the running count, # class-of-event wording (no action a trip may skip); the counter is # named for what it counts — gap detections, not resyncs performed. rec = re.search(r"_recordTruncatedGap\(\)\s*\{(.*?)\n \}", body, re.S) assert rec is not None, "_recordTruncatedGap method not found" r = rec.group(1) assert "this._streamHealth.truncatedGaps += 1;" in r assert "console.warn(" in r assert "resyncing history from REST" not in r, ( "the shared log line must not assert a resync the trip branch skips" ) assert body.count("this._streamHealth.truncatedGaps += 1;") == 1, ( "all truncatedGaps bumps must route through _recordTruncatedGap " "so the running-count console invariant holds" ) # (7) jittered scheduler: dedup guard, null-before-load, cancel on # disconnect. sched = re.search(r"_scheduleTruncatedResync\(\)\s*\{(.*?)\n \}", body, re.S) assert sched is not None, "_scheduleTruncatedResync method not found" s = sched.group(1) assert "if (this._resyncTimer != null) return;" in s assert "Math.random() * TRUNCATED_RESYNC_JITTER_MS" in s null_then_load = s.index("this._resyncTimer = null;") load = s.index("this._loadHistoryThenConnect(this.wsId);") assert null_then_load < load, ( "the firing path must null the handle BEFORE loading, or " "disconnectSSE (called inside the load) would cancel the work it " "is part of" ) dis = re.search(r"disconnectSSE\(\)\s*\{(.*?)\n \}", body, re.S) assert dis is not None, "disconnectSSE not found" assert "clearTimeout(this._resyncTimer)" in dis.group(1) # (8) interleaving-proof gap repair: the truncation-time cursor is a # single record — captured keep-oldest at the envelope, cleared by # any full committed-history render (replayHistory) or a ws switch — # and the CONNECT CHOKEPOINT consumes it: while set, every manual # (re)connect presents it instead of the since-advanced live cursor, # so the server re-answers replay_truncated and the resync re-arms # no matter which teardown (hide/show, degraded cooldown, recover # beat, failed /history fetch) cancelled the pending jittered timer. # A cancellable timer alone silently lost the repair when a # hide/show cycle landed inside the jitter window. assert re.search( r"if \(this\._truncatedFromCursor == null\) \{\s*" r"this\._truncatedFromCursor = this\._lastEventId;", t, ), "the truncated case must record the truncation-time cursor keep-oldest" load = body.index("_loadHistoryThenConnect(wsId, manualAttempt = false) {") load_seg = body[load : body.index("async _refetchHistory(", load)] assert "if (this.wsId !== wsId) this._truncatedFromCursor = null;" in load_seg, ( "a ws switch must drop the old ws's truncation record" ) replay_fn = body.index("replayHistory(messages) {") replay_head = body[replay_fn : replay_fn + 1200] assert "this._truncatedFromCursor = null;" in replay_head, ( "a successful full-history render must clear the truncation record" ) # (8b) ...and supersede ALL pending repair intent in the same breath — # the deferred mid-turn latch and any pending jittered timer. Without # these, a clear_ui heal left them armed and the next idle edge fired a # phantom _loadHistoryThenConnect against the repaired gap (false # truncatedGaps bump; on its failed-fetch leg, a cursorless reconnect # with nothing armed). Every _loadHistoryThenConnect flavor # clears both BEFORE its fetch, so these are no-ops on the load paths — # the clear_ui heal is the path they exist for. Mirrors the # coordinator's refetchHistory supersession. assert "this._pendingTruncatedResync = false;" in replay_head, ( "replayHistory must clear the deferred-resync latch — a latch " "surviving a heal fires a phantom resync at the next idle edge" ) assert "clearTimeout(this._resyncTimer)" in replay_head, ( "replayHistory must cancel a pending jittered resync — the render " "just repaired the gap it was scheduled for" ) conn = body.index("connectSSE(wsId) {") conn_seg = body[conn : body.index("this.evtSource = new EventSource", conn)] assert "this._truncatedFromCursor != null" in conn_seg, ( "connectSSE must consult the truncation record" ) assert "encodeURIComponent(connectCursor)" in conn_seg # (9) same-ws supersession must not strand the queued edit-and-resend # (the jittered resync bumps _historyLoadToken through # _loadHistoryThenConnect while a clear_ui edit flow is in flight); # only a ws SWITCH discards the resend, and then it recovers the # composer instead of leaving the latch armed for the next ws. clear = re.search(r'case "clear_ui": \{(.*?)break;', body, re.S) assert clear is not None, "clear_ui case not found" cl = clear.group(1) assert "const editWs = this.wsId;" in cl assert "if (token !== this._historyLoadToken && this.wsId !== editWs)" in cl, ( "only a cross-ws supersession may discard the edit-and-resend" ) assert "this._pendingEditSend = null;" in cl assert "this.setBusy(false);" in cl # Supersession is centralized at the render (replayHistory) — clear_ui # must not grow a path-local resync cancel of its own. assert "clearTimeout(this._resyncTimer)" not in cl def test_degraded_catchup_stops_live_stream_and_retries() -> None: """Degraded catch-up contract: close the stream FIRST (which also clears any earlier degraded timer — disconnectSSE owns that), show a plain-language status, then arm the retry timer with a doubling cooldown. The retry must defer to the show edge when the tab is hidden (reopening into a throttled tab would overflow again).""" body = _INTERACTIVE.read_text(encoding="utf-8") m = re.search(r"_enterDegradedCatchup\(\)\s*\{(.*?)\n \}", body, re.S) assert m is not None, "_enterDegradedCatchup method not found" method = m.group(1) # Order matters: disconnect before arming the timer, or the fresh # timer would be cancelled by its own disconnect. assert method.index("this.disconnectSSE()") < method.index("this._degradedTimer = setTimeout") assert "Connection is slow" in method, "degraded state must use plain language" assert "DEGRADED_COOLDOWN_MAX_MS" in method assert "document.hidden" in method # disconnectSSE owns the timer teardown (ws-switch / giveUp / destroy # all supersede a pending degraded retry through it). dis = re.search(r"disconnectSSE\(\)\s*\{(.*?)\n \}", body, re.S) assert dis is not None assert "clearTimeout(this._degradedTimer)" in dis.group(1) def test_visibilitychange_closes_on_hide_reconnects_on_show() -> None: """Close-on-hide / replay-on-show: a hidden tab's throttled drain is the likeliest slow consumer behind server-side overflow (the old "PR-G closes those connections on hide" comment described a handler that never existed). The pane installs one visibilitychange listener, marks ITS OWN hide-closes via ``_hiddenDisconnect`` so a show edge never resurrects a deliberately-closed stream, and the factory's destroy removes the listener (it strongly references the pane).""" body = _INTERACTIVE.read_text(encoding="utf-8") assert 'document.addEventListener("visibilitychange", this._visHandler);' in body assert 'document.removeEventListener("visibilitychange", this._visHandler);' in body vis = re.search(r"_onVisibilityChange\(\)\s*\{(.*?)\n \}", body, re.S) assert vis is not None, "_onVisibilityChange method not found" method = vis.group(1) assert "this.disconnectSSE();" in method assert "this._hiddenDisconnect = true;" in method assert "this.connectSSE(this.wsId);" in method # Reconnect only consumes OUR hide-close marker. assert "else if (this._hiddenDisconnect)" in method # Teardown: the factory controller removes the listener on destroy. assert "pane._removeVisibilityHandler();" in body # The streaming buffers survive a hide-close: disconnectSSE stays # transport-only (no contentBuffer wipe) so the visible tail is # intact when the tab returns. dis = re.search(r"disconnectSSE\(\)\s*\{(.*?)\n \}", body, re.S) assert dis is not None assert "contentBuffer" not in dis.group(1) def test_no_global_sse_gap_detector() -> None: """Live event ids are NOT strictly monotonic across concurrent tool+content emit (the fan-out runs outside the listeners lock), so a naive ``id !== lastEventId + 1`` gap check would false-positive. Recovery is server-signalled (``stream_overflow``) + reconnect replay instead. This tripwire pins the absence of the naive arithmetic — if gap detection is ever added, it must be scoped to the content stream only (content-vs-content never reorders).""" code = _strip_comments(_INTERACTIVE.read_text(encoding="utf-8")) assert not re.search(r"_lastEventId\s*[+\-]\s*1", code), ( "found lastEventId +/- 1 arithmetic — a global gap detector " "false-positives on legal concurrent tool/content id inversion" ) def test_overflow_helpers_extracted_to_shared_module() -> None: """The storm-guard constants + the two pure helpers were extracted to the shared ``sse_overflow.js`` module (its own runtime probes live in ``test_sse_overflow_js.py``) so the interactive and coordinator panes can't drift. Pin that the pane IMPORTS them rather than re-declaring a local copy: a stray local ``function overflowWindowTripped`` / ``const OVERFLOW_TRIP_COUNT`` would silently fork the trip math again.""" body = _INTERACTIVE.read_text(encoding="utf-8") m = re.search( r"import \{([^}]*)\} from \"\./sse_overflow\.js\";", body, re.S, ) assert m is not None, "interactive pane must import the shared overflow helpers" imported = m.group(1) for name in ( "OVERFLOW_TRIP_COUNT", "OVERFLOW_TRIP_WINDOW_MS", "DEGRADED_COOLDOWN_BASE_MS", "DEGRADED_COOLDOWN_MAX_MS", "DEGRADED_COOLDOWN_RESET_MS", "overflowWindowTripped", "degradedCooldownStep", ): assert name in imported, f"{name} must be imported from sse_overflow.js" # No local fork of the extracted definitions. assert not re.search(r"^function overflowWindowTripped\(", body, re.M), ( "overflowWindowTripped must be imported, not re-declared locally" ) assert not re.search(r"^function degradedCooldownStep\(", body, re.M), ( "degradedCooldownStep must be imported, not re-declared locally" ) assert not re.search(r"^const OVERFLOW_TRIP_COUNT\s*=", body, re.M), ( "the trip constants must be imported, not re-declared locally" ) def test_note_stream_overflow_does_not_reset_cooldown() -> None: """The exact finding [0] bug shape must not regress: _noteStreamOverflow only counts + trips; it must NOT touch _degradedCooldownMs (the reset that defeated the ladder lived here). The ladder decision lives solely in _enterDegradedCatchup, keyed off _lastDegradedAt via degradedCooldownStep.""" body = _INTERACTIVE.read_text(encoding="utf-8") note = re.search(r"_noteStreamOverflow\(\)\s*\{(.*?)\n \}", body, re.S) assert note is not None, "_noteStreamOverflow not found" assert "_degradedCooldownMs" not in note.group(1), ( "_noteStreamOverflow must not write _degradedCooldownMs — that reset " "was the bug that stopped the ladder escalating" ) enter = re.search(r"_enterDegradedCatchup\(\)\s*\{(.*?)\n \}", body, re.S) assert enter is not None assert "degradedCooldownStep(" in enter.group(1) assert "this._lastDegradedAt = now" in enter.group(1) def test_recover_beat_defers_reconnect_when_tab_hidden() -> None: """Review round-2 finding [1]: the factory's transient-error recovery beat (recoverTimer) must NOT reopen an EventSource into a hidden tab — that re-creates the throttled slow-consumer overflow that close-on-hide exists to prevent. It guards on document.hidden and defers to the visibilitychange show edge (marking _hiddenDisconnect).""" body = _INTERACTIVE.read_text(encoding="utf-8") beat = re.search(r"recoverTimer = setTimeout\(\(\) => \{(.*?)\n \}, 5000\);", body, re.S) assert beat is not None, "recoverTimer setTimeout body not found" b = beat.group(1) assert "document.hidden" in b, "recovery beat must guard on document.hidden" assert "pane._hiddenDisconnect = true" in b, ( "recovery beat must defer to the show edge when hidden" ) # The hidden guard must precede the reconnect (connectSSE) so it can't fall # through to reopening the stream. assert b.index("document.hidden") < b.index("pane.connectSSE(pane.wsId)") def test_giveup_removes_visibility_handler() -> None: """Review round-2 finding [3]: giveUp() (markDead) must detach the visibility handler and clear _hiddenDisconnect, or a tab hidden before the give-up resurrects the dead controller's stream on return (the show edge would connectSSE the closed ws and 404-reconnect it forever).""" body = _INTERACTIVE.read_text(encoding="utf-8") give = re.search(r"const giveUp = function \(\) \{(.*?)\n \};", body, re.S) assert give is not None, "giveUp function body not found" g = give.group(1) assert "pane._removeVisibilityHandler();" in g, ( "giveUp must remove the visibility handler so a show edge can't resurrect a dead controller" ) # _removeVisibilityHandler also clears _hiddenDisconnect (pinned in its body). rvh = re.search(r"_removeVisibilityHandler\(\)\s*\{(.*?)\n \}", body, re.S) assert rvh is not None assert "this._hiddenDisconnect = false" in rvh.group(1) def test_connectsse_defers_open_when_tab_hidden() -> None: """PR #805 review (Copilot + R3): connectSSE is the single connect chokepoint and must not open an EventSource into a hidden tab. The fresh-connect path (_loadHistoryThenConnect) has no timer guard, so a first load in a background tab would otherwise open a throttled stream — the slow-consumer overflow this PR exists to prevent. The guard sits AFTER the visibilitychange-handler install (so the show edge can reconnect) and AFTER the wsId assignment (so it targets the right ws), and BEFORE `new EventSource` (so nothing opens).""" body = _INTERACTIVE.read_text(encoding="utf-8") start = body.index("connectSSE(wsId) {") open_at = body.index("new EventSource(evtUrl)", start) head = body[start:open_at] # connectSSE up to the EventSource open assert "if (document.hidden) {" in head, ( "connectSSE must guard on document.hidden BEFORE opening the stream" ) assert "this._hiddenDisconnect = true;" in head, ( "the deferred connect must mark _hiddenDisconnect so the show edge reconnects" ) assert head.index("this.wsId = wsId;") < head.index("if (document.hidden) {") assert head.index('addEventListener("visibilitychange"') < head.index("if (document.hidden) {") def test_send_post_abort_machinery_is_gone() -> None: """The parked-POST era's client abort machinery must stay deleted in BOTH panes: sends during a command window are answered "queued" immediately (server-side defer-and-drain), so there is no long-lived POST for a compaction-aware bound (``sendAbortMs``) to protect, and dismissal is bind() → server-confirmed DELETE — never a POST abort (``_sendAbort``), which fired on the interjection path too and dispatched "dismissed" messages anyway. Reintroducing either hook means re-parking the POST; that design deterministically dropped messages from every timeout-bounded caller (coordinator client and console proxy at 30s, SDKs, stock proxies).""" interactive = _INTERACTIVE.read_text(encoding="utf-8") coordinator = (_ROOT / "turnstone/console/static/coordinator/coordinator.js").read_text( encoding="utf-8" ) conversation = (_ROOT / "turnstone/shared_static/conversation.js").read_text(encoding="utf-8") composer_queue = (_ROOT / "turnstone/shared_static/composer_queue.js").read_text( encoding="utf-8" ) for name, src in ( ("interactive.js", interactive), ("coordinator.js", coordinator), ("conversation.js", conversation), ("composer_queue.js", composer_queue), ): assert "sendAbortMs" not in src, f"{name}: the compaction-aware abort bound is dead" assert "_sendAbort" not in src, f"{name}: dismiss must be bind() → DELETE, not a POST abort" # The flat wedged-node bound stands in both panes: every /send answers # within RTT now (dispatched / queued / deferred-with-msg_id). assert "sendCtrl.abort(), 15000" in interactive assert "sendCtrl.abort(), 15000" in coordinator # The deferred-attachment count rides bind()'s documented options seam # (controller dataset) — the per-pane element expando is dead. for name, src in ( ("interactive.js", interactive), ("coordinator.js", coordinator), ("composer_queue.js", composer_queue), ): assert "_deferredAttachments" not in src, ( f"{name}: deferred state must ride bind(el, msgId, opts), not an expando" ) def test_deferred_send_settle_protocol_pins() -> None: """The deferred-chip settle protocol (round 7, C4): a deferred send's queued chip keeps its retraction affordance exactly until the message truly leaves the parked list. Pins the controller's contract and both panes' wiring — losing any of these silently re-promotes parked messages to "sent" while the server still honors DELETE (loss disguised as delivery on a node restart).""" interactive = _INTERACTIVE.read_text(encoding="utf-8") coordinator = (_ROOT / "turnstone/console/static/coordinator/coordinator.js").read_text( encoding="utf-8" ) composer_queue = (_ROOT / "turnstone/shared_static/composer_queue.js").read_text( encoding="utf-8" ) # Controller: bind() stores the options on its own dataset state... assert "function bind(el, msgId, opts)" in composer_queue assert 'el.dataset.deferred = "1"' in composer_queue assert "el.dataset.attachedCount = String(opts.attachedCount)" in composer_queue # ...the idle sweep skips deferred AND unbound chips (the "idle ⇒ # drained" invariant is untrue for both)... assert "if (el.dataset.deferred) return;" in composer_queue assert "if (!el.dataset.msgId) return;" in composer_queue # ...and settleDeferred branches on the fold-in arm: clear the flag # only (DELETE still genuinely removes a folded message until the seam # drains), promote only on the fresh-spawn arm. assert "function settleDeferred(msgId, folded)" in composer_queue assert "delete target.dataset.deferred;" in composer_queue assert "settleDeferred: settleDeferred" in composer_queue # A barrier-deferred entry can dispatch within milliseconds of its ack, # so the SSE settle can beat the POST response's bind(): the controller # parks chip-absent settles and bind() reconciles them — without this a # raced chip stays flagged deferred and the idle sweep skips it forever. # Expiry is TTL-based: a size cap evicted exactly this tab's raced # settle when a window closed with a burst of deferred sends (ours # parks FIRST, the foreign settles behind it overflow the cap). assert "_preBindSettles" in composer_queue assert "_preBindSettles.has(msgId)" in composer_queue assert "PRE_BIND_SETTLE_TTL_MS" in composer_queue assert "_preBindSettles.size" not in composer_queue, "size-cap eviction must stay dead" # The full send-response settle matrix lives ONCE, in the shared # helper — retro-convert (a parked, still-retractable message must not # render as a sent bubble), the deferred busy-undo, and the queue_full # idle-pane cleanup (bubble removed + busy restored: the refusal can # now fire with no worker and no drain alive, so no state event would # ever unstick the composer). assert "export function settleSendResponse(queue, data, ctx)" in composer_queue assert "!queuedEl && data.deferred" in composer_queue assert "deferred: !!data.deferred" in composer_queue assert "attachedCount: (data.attached_ids || []).length" in composer_queue assert "ctx.busyIsOptimistic()" in composer_queue assert composer_queue.count("ctx.optimisticEl.remove()") >= 3, ( "retro-convert, queue_full, and attachments_busy must clear false optimistic bubbles" ) # The missed-edge settle: a non-deferred chip binding onto an # already-idle pane missed its only sweep — the post-bind promote # (keyed on POST-bind chip state, honoring the aria-busy # dismiss-in-flight discipline) is what settles it. assert "!ctx.paneIsBusy()" in composer_queue assert 'queuedEl.hasAttribute("aria-busy")' in composer_queue # Both panes route their parsed /send response through the helper and # consume the pane-tier settle event; the busy stamp is centralized in # each pane's setBusy (source defaults to "server" — only the send # flow's optimistic flip may ever be undone). # postAndSettleSend wraps settleSendResponse with the fetch stage (rejected # -body normalization, the 409 conversion, the accepted-guarded transport # catch), so a pane reaching the settle matrix at all now proves it reached # the whole choreography — the panes must not call settleSendResponse # directly, which is how the edit-and-resend flows drifted. assert "export function postAndSettleSend(queue, sendRequest, ctx)" in composer_queue for name, src in (("interactive.js", interactive), ("coordinator.js", coordinator)): assert "postAndSettleSend(" in src, f"{name}: settle matrix must be the shared helper" assert "settleSendResponse(" not in src, f"{name}: must not bypass the fetch stage" assert "busyIsOptimistic" in src, name assert "paneIsBusy" in src, f"{name}: the missed-edge settle needs the live flag" assert "mergeRejectedComposerText" in src, f"{name}: refused text must be restored" assert src.count("restoreInput:") == 2, ( f"{name}: composer send and edit-resend both need refusal restoration" ) assert 'setBusy(true, "optimistic")' in src, f"{name}: optimistic flip must stamp" assert "parsePriority(" in src, f"{name}: shared !!! parse" assert 'case "message_dispatched"' in src, f"{name}: settle event not consumed" assert "settleDeferred(" in src, name # /command's degraded outcomes are ALL surfaced: busy, running (the # backstop answer), error (503 — the worker never spawned), the # status-less non-2xx arm (404 / proxy 502), and the transport catch — # silence at any of them reads as success. assert 'body.status === "running"' in interactive assert 'body.status === "error"' in interactive assert "Command failed (HTTP " in interactive assert '"Command failed: " + err.message' in interactive def test_settle_send_response_missed_edge_behavior(tmp_path) -> None: """Execute the shared settle helper under node and pin the missed-edge matrix behaviorally (not just textually): a non-deferred chip binding onto an idle pane promotes; a busy pane, a deferred chip, and a dismiss-in-flight chip do not.""" import shutil import subprocess if shutil.which("node") is None: pytest.skip("node binary not available on PATH") helper = _ROOT / "turnstone/shared_static/composer_queue.js" script = tmp_path / "settle_harness.mjs" script.write_text( f'const {{ settleSendResponse }} = await import("file://{helper}");\n' + """ function makeEl(over) { const el = { isConnected: true, classList: { contains: (c) => c === "msg-queued" }, dataset: {}, hasAttribute: () => false, }; return Object.assign(el, over || {}); } function run(queuedEl, paneBusy, data) { const calls = []; const queue = { bind: (el, id, opts) => { calls.push("bind"); // Mirror the real bind: stamp the deferred flag from opts. if (opts && opts.deferred) el.dataset.deferred = "1"; }, promote: () => calls.push("promote"), remove: () => calls.push("remove"), addQueuedMessage: () => makeEl(), }; settleSendResponse(queue, data, { queuedEl, optimisticEl: null, isBusy: true, displayText: "t", priority: "notice", setBusy: () => {}, busyIsOptimistic: () => false, paneIsBusy: () => paneBusy, renderError: () => {}, consumeAttachments: () => {}, }); return calls; } const queued = { status: "queued", msg_id: "m1" }; let c = run(makeEl(), false, queued); if (!(c.includes("bind") && c.includes("promote"))) throw new Error("missed-edge chip must promote: " + c); c = run(makeEl(), true, queued); if (c.includes("promote")) throw new Error("busy pane must not promote: " + c); c = run(makeEl(), false, { status: "queued", msg_id: "m1", deferred: true }); if (c.includes("promote")) throw new Error("deferred chip is message_dispatched's: " + c); c = run(makeEl({ hasAttribute: (a) => a === "aria-busy" }), false, queued); if (c.includes("promote")) throw new Error("dismiss-in-flight chip must be left to its DELETE verdict: " + c); // Null / non-object 2xx body (a misbehaving proxy answering `200 null`): the // helper normalizes it to {} so neither call site guards — it must fall through // to the unknown/"ok" arm and SETTLE the optimistic chip (promote), never throw // and strand a delivered message as a connection error. (The no-op // consumeAttachments stub cannot prevent this: data.attached_ids is evaluated to // build the :639 call args, so against unfixed code this line throws and crashes // the harness.) c = run(makeEl(), false, null); if (!c.includes("promote")) throw new Error("null body must settle via unknown-ok, not throw: " + c); console.log("settle matrix OK"); """, encoding="utf-8", ) proc = subprocess.run( ["node", str(script)], capture_output=True, text=True, timeout=15, ) assert proc.returncode == 0, f"settle harness failed:\n{proc.stderr}\n{proc.stdout}" def test_stale_idle_refusals_restore_input(tmp_path) -> None: """A stale local idle state must not render either busy refusal as delivered or discard its companion text. Text entered during the POST is retained after the rejected text, and an SSE idle that already arrived prevents the old optimistic busy state from being reasserted.""" import shutil import subprocess if shutil.which("node") is None: pytest.skip("node binary not available on PATH") helper = _ROOT / "turnstone/shared_static/composer_queue.js" script = tmp_path / "attachments_busy_harness.mjs" script.write_text( rf"""const {{ mergeRejectedComposerText, settleSendResponse }} = await import("file://{helper}"); function run(status, optimisticBusy) {{ const calls = []; let composerValue = "typed during request"; const optimisticEl = {{ isConnected: true, dataset: {{}}, remove: () => calls.push("remove-optimistic"), }}; settleSendResponse( {{ remove: () => calls.push("remove-queued") }}, {{ status }}, {{ queuedEl: null, optimisticEl, isBusy: false, setBusy: (value) => calls.push("busy:" + value), busyIsOptimistic: () => optimisticBusy, paneIsBusy: () => optimisticBusy, restoreInput: () => {{ composerValue = mergeRejectedComposerText("rejected", composerValue); calls.push("restore"); }}, renderError: () => calls.push("error"), consumeAttachments: () => calls.push("consume"), }}, ); return {{ calls, composerValue }}; }} for (const [status, expectedBusy] of [ ["attachments_busy", "busy:true"], ["cross_user_interjection", "busy:false"], ["queue_full", "busy:false"], ]) {{ let result = run(status, true); if (result.composerValue !== "rejected\ntyped during request") throw new Error(status + " companion/current text merge drifted: " + result.composerValue); for (const call of ["remove-optimistic", "restore", expectedBusy, "error"]) {{ if (!result.calls.includes(call)) throw new Error(status + " missing stale-idle settlement " + call + ": " + result.calls); }} if (result.calls.includes("consume") || result.calls.includes("remove-queued")) throw new Error(status + " attachments/chip state was consumed: " + result.calls); result = run(status, false); if (result.calls.some((call) => call.startsWith("busy:"))) throw new Error(status + " overwrote a raced SSE state: " + result.calls); }} if ( mergeRejectedComposerText("rejected", "rejected\nlater") !== "rejected\nrejected\nlater" ) throw new Error("independently typed matching text was discarded"); if (mergeRejectedComposerText("rejected", "") !== "rejected") throw new Error("empty composer did not restore rejected text"); """, encoding="utf-8", ) proc = subprocess.run( ["node", str(script)], capture_output=True, text=True, timeout=15, ) assert proc.returncode == 0, f"attachments_busy harness failed:\n{proc.stderr}\n{proc.stdout}" def test_accepted_tool_event_recorded_only_when_painted() -> None: """An unpainted accepted tool_result must stay replayable. appendToolOutput returns false on every no-target path (transcript wiped by clear_ui with the refetch in flight, a fresh mid-turn join); recording the event id anyway would dedupe the ring's later replay and permanently lose the tool's final guarded output. """ body = _INTERACTIVE.read_text(encoding="utf-8") case_start = body.index('case "tool_result"') case = body[case_start : body.index("case ", case_start + 20)] gate = case.index("if (\n this.appendToolOutput(") record = case.index("recordAcceptedToolEvent(this._renderedToolEventIds, evt)") assert gate < record def test_replay_system_rows_do_not_terminate_the_tool_batch_window() -> None: """A system row inside a tool batch is not a turn boundary: nulling ``lastToolBlock`` in the ``role === "system"`` replay branch made every tool result AFTER an interleaved row (mid-turn operator context, a second writer's append) silently vanish from this pane while the coordinator — whose indexHistoryToolOutcomes skips non-turn rows — rendered the identical history correctly. Only the user/assistant branches may reset the anchor.""" body = _strip_comments(_INTERACTIVE.read_text(encoding="utf-8")) start = body.index('(msg.role === "system")') end = body.index("for (const leftovers of pendingAssessments.values())", start) system_branch = body[start:end] assert "lastToolBlock = null" not in system_branch, ( "the system replay branch terminates the batch result window — " "interleaved-row tool results are dropped again" ) # Mutation control: the anchor resets still exist in the turn branches # (user, nudge-marker, assistant content/reasoning/pending arms). loop = body[body.index("let lastToolBlock = null") : end] assert loop.count("lastToolBlock = null") >= 4 def test_orphan_tool_result_does_not_mark_a_batch_failed() -> None: """Keeping the batch anchor live across non-turn rows means a tool row that names a call_id this batch never issued (a result for an earlier batch, a second writer's append) can reach the error stamp. It must not mark an all-succeeded batch as failed — the shared outcome index skips unmatched occurrences for exactly this reason. A row with NO call_id is the legacy positional case and must still stamp, so the guard keys on 'named a call_id we could not resolve', not on the absence of a resolved target.""" body = _strip_comments(_INTERACTIVE.read_text(encoding="utf-8")) assert "const isOrphanResult = !!msg.tool_call_id && !resultTarget;" in body, ( "the orphan discriminator must distinguish an unresolvable call_id " "from a legacy row that carries none" ) stamp = body.index("appendToolErrorBadge(lastToolBlock)") guard = body.rindex("if (", 0, stamp) assert "!isOrphanResult" in body[guard:stamp], ( "an orphan result can stamp conv-batch--error on a batch whose own calls all succeeded" )