"""Offline contracts for the dependency-free gateway web UI.
These tests intentionally inspect the static sources instead of starting the
gateway or a browser. They catch broken asset links and the most important
HTML/JavaScript integration seams while keeping the default test suite fully
offline and stdlib-only.
"""
from __future__ import annotations
import re
from collections import Counter
from html.parser import HTMLParser
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
WEBUI = ROOT / "openai4s" / "server" / "webui"
INDEX_PATH = WEBUI / "index.html"
APP_PATH = WEBUI / "app.js"
STYLE_PATH = WEBUI / "style.css"
SCIENTIFIC_RENDERERS_PATH = WEBUI / "scientific_renderers.js"
INDEX_HTML = INDEX_PATH.read_text(encoding="utf-8")
APP_JS = APP_PATH.read_text(encoding="utf-8")
STYLE_CSS = STYLE_PATH.read_text(encoding="utf-8")
SCIENTIFIC_RENDERERS_JS = SCIENTIFIC_RENDERERS_PATH.read_text(encoding="utf-8")
class _WebUIShellParser(HTMLParser):
"""Collect the small part of the HTML surface these contracts need."""
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.ids: list[str] = []
self.classes: set[str] = set()
self.data_icons: set[str] = set()
self.static_assets: set[str] = set()
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
values = {name: value for name, value in attrs}
if values.get("id"):
self.ids.append(values["id"] or "")
self.classes.update((values.get("class") or "").split())
if values.get("data-icon"):
self.data_icons.add(values["data-icon"] or "")
for attr in ("href", "src"):
value = values.get(attr) or ""
if value.startswith("/static/"):
self.static_assets.add(value.split("?", 1)[0].split("#", 1)[0])
SHELL = _WebUIShellParser()
SHELL.feed(INDEX_HTML)
def _extract_js_function(source: str, name: str) -> str:
"""Return a named classic JS function, balancing braces outside strings.
The web UI deliberately has no build tool or JavaScript parser dependency.
This tiny scanner is enough for its classic ``function name(...)`` forms
and is more stable than stopping at the first nested closing brace.
"""
match = re.search(
rf"\b(?:async\s+)?function\s+{re.escape(name)}\s*\([^)]*\)\s*\{{",
source,
)
assert match, f"app.js must define function {name}()"
start = match.end() - 1
depth = 0
quote: str | None = None
escaped = False
line_comment = False
block_comment = False
index = start
while index < len(source):
char = source[index]
nxt = source[index + 1] if index + 1 < len(source) else ""
if line_comment:
if char == "\n":
line_comment = False
elif block_comment:
if char == "*" and nxt == "/":
block_comment = False
index += 1
elif quote:
if escaped:
escaped = False
elif char == "\\":
escaped = True
elif char == quote:
quote = None
elif char == "/" and nxt == "/":
line_comment = True
index += 1
elif char == "/" and nxt == "*":
block_comment = True
index += 1
elif char in {'"', "'", "`"}:
quote = char
elif char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
return source[match.start() : index + 1]
index += 1
raise AssertionError(f"unterminated function {name}() in app.js")
def _icon_definitions() -> set[str]:
match = re.search(
r"\bconst\s+ICONS\s*=\s*\{(?P.*?)^\};",
APP_JS,
flags=re.MULTILINE | re.DOTALL,
)
assert match, "app.js must define the ICONS object"
return set(
re.findall(r"^\s*['\"]([^'\"]+)['\"]\s*:", match.group("body"), re.MULTILINE)
)
def test_every_referenced_static_asset_exists() -> None:
# Include JavaScript-loaded first-party assets (for example vendored
# 3Dmol), not only /", ""],
["//evil.example/path", ""],
["ftp://evil.example/path", ""],
["https:/missing-slash.example", ""],
[None, ""],
[42, ""],
]
script = (
source
+ "\nconst cases = "
+ json.dumps(cases)
+ ";\nconsole.log(JSON.stringify(cases.map(([input]) => "
+ "searchResultHttpUrl(input))));"
)
out = subprocess.run(
[NODE, "--input-type=module", "-e", script],
capture_output=True,
text=True,
timeout=60,
)
assert out.returncode == 0, out.stderr[:800]
assert json.loads(out.stdout) == [expected for _, expected in cases]
def _drive(program: str) -> dict:
"""Run the shipped ticket functions, lifted from app.js, under node.
Lifted rather than reimplemented, for the same reason the R producer tests
lift `.oai4s_cap_message`: a hand-written stand-in is the thing that drifts.
"""
sources = "\n".join(
_fn(name)
for name in (
"openTurnTicket",
"commitTurnTicket",
"closeTurnTicket",
"activateTurnTicket",
"ownsTurnTicket",
"acceptTurnTicket",
"retireTurnTicket",
"isStaleTurnEvent",
)
)
script = f"const S = {{ running: false }};\n{sources}\n{program}"
out = subprocess.run(
[NODE, "--input-type=module", "-e", script],
capture_output=True,
text=True,
timeout=60,
)
assert out.returncode == 0, out.stderr[:800]
return json.loads(out.stdout.strip().splitlines()[-1])
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_202_that_resolves_after_the_turn_failed_writes_nothing() -> None:
"""The race, in the order that actually happens.
Terminal WS first, 202 second. Before the guard the slot ended up holding a
finished turn's id, and the *next* turn quoted it -- sending an operator to
the wrong request, which is worse than showing none.
"""
state = _drive("""
S.running = true;
const ticket = openTurnTicket(); // send() takes it, then awaits
closeTurnTicket(); // terminal WS wins the race
S.running = false;
const wrote = commitTurnTicket(ticket, { request_id: "req-old" });
console.log(JSON.stringify({ wrote, pending: S.pendingRequestId }));
""")
assert state["wrote"] is False
assert state.get("pending") in (None, ""), state
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_closing_the_ticket_alone_is_enough_to_refuse_a_late_202() -> None:
"""Isolates the generation half.
`S.running` is left true, so only the generation moved. Without this the
test above passes on a `closeTurnTicket` that merely nulls the slot and
never advances the generation -- and then a late 202 writes the id straight
back in, which is the whole defect.
"""
state = _drive("""
S.running = true;
const ticket = openTurnTicket();
closeTurnTicket();
const wrote = commitTurnTicket(ticket, { request_id: "req-old" });
console.log(JSON.stringify({ wrote, pending: S.pendingRequestId }));
""")
assert state["wrote"] is False, "a closed ticket still accepted its 202"
assert state.get("pending") in (None, ""), state
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_turn_that_is_no_longer_running_refuses_its_own_202() -> None:
"""Isolates the running half.
The generation is untouched, so only `S.running` can refuse. `turnDone`
clears `S.running` at its first statement and closes the ticket near its
last, so this is the window between them -- narrow today, and the kind of
thing a later edit widens without noticing.
"""
state = _drive("""
S.running = true;
const ticket = openTurnTicket();
S.running = false; // turn ended, ticket not yet closed
const wrote = commitTurnTicket(ticket, { request_id: "req-old" });
console.log(JSON.stringify({ wrote, pending: S.pendingRequestId }));
""")
assert state["wrote"] is False, "an ended turn still stored its 202"
assert state.get("pending") in (None, ""), state
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_the_next_turn_does_not_inherit_the_previous_ticket() -> None:
"""A turn with no id of its own must quote nothing, not the last one."""
state = _drive("""
S.running = true;
const first = openTurnTicket();
commitTurnTicket(first, { request_id: "req-first" });
closeTurnTicket(); // first turn ends
S.running = true;
openTurnTicket(); // second turn starts
const late = commitTurnTicket(first, { request_id: "req-first" });
console.log(JSON.stringify({ late, pending: S.pendingRequestId }));
""")
assert state["late"] is False, "a stale 202 revived the previous turn's id"
assert state.get("pending") in (None, ""), state
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_the_ordinary_ordering_still_stores_the_ticket() -> None:
"""The guard must not make every turn id-less, or it is just a deletion."""
state = _drive("""
S.running = true;
const ticket = openTurnTicket();
const wrote = commitTurnTicket(ticket, { request_id: "req-live" });
console.log(JSON.stringify({ wrote, pending: S.pendingRequestId }));
""")
assert state["wrote"] is True
assert state["pending"] == "req-live"
# --- queued follow-ups ---------------------------------------------------------
def test_a_queued_send_takes_no_ticket() -> None:
"""A follow-up is not the running turn.
Its 202 resolves while the previous turn still owns the screen, so writing
its id into the slot makes the *active* turn's failure quote it -- an
operator sent to a request that had not started.
"""
body = _fn("send")
assert "const sawRunningAtDispatch = S.running;" in body, body[:900]
assert "sawRunningAtDispatch ? null : openTurnTicket()" in body
# The snapshot from the top of `send` may not decide this: by the time the
# POST goes out it is stale, and `S.running` has been set by this send.
assert "queueing ? null" not in body
assert (
"if (!acceptTurnTicket(turnTicket, accepted)) retireTurnTicket(turnTicket)"
in body
)
def test_a_rejected_send_only_tears_down_a_turn_it_owns() -> None:
"""`queueing` is a snapshot, and the catch runs long after it was taken."""
body = _fn("send")
assert 'if (ownsTurnTicket(turnTicket)) turnDone("failed")' in body, body[-2500:]
assert 'if (queueing) w.classList.add("cancelled");' not in body
def test_the_processing_event_hands_the_slot_over() -> None:
"""And unconditionally: `S.running` is already true when a queued turn starts."""
assert (
'if (m.status === "processing") activateTurnTicket(m.request_id, m.execution_id);'
in APP_JS
)
index = APP_JS.index('if (m.status === "processing") activateTurnTicket')
guarded = APP_JS.index('if (m.status === "processing" && !S.running)')
assert index < guarded, "the hand-off sits inside the not-running guard"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_only_queue_position_zero_proves_this_send_is_the_active_turn() -> None:
"""Every acceptance shape, through the shipped helper.
`> 0` is queued behind someone else. Absent means the snapshot could not be
taken -- typically a job that finished before it was read -- and an unknown
is not a yes. Both used to write, and both put a non-running turn's id in
the slot the running turn's failure quotes.
"""
state = _drive("""
const out = {};
const run = (label, accepted) => {
S.running = true;
closeTurnTicket();
const token = openTurnTicket();
commitTurnTicket(token, { request_id: "req-ACTIVE" });
out[label] = {
wrote: acceptTurnTicket(token, accepted),
pending: S.pendingRequestId,
};
};
run("zero", { request_id: "req-NEW", queue_position: 0 });
run("queued", { request_id: "req-NEW", queue_position: 3 });
run("absent", { request_id: "req-NEW" });
run("noId", { queue_position: 0 });
console.log(JSON.stringify(out));
""")
assert state["zero"] == {"wrote": True, "pending": "req-NEW"}
assert state["queued"] == {"wrote": False, "pending": "req-ACTIVE"}
assert state["absent"] == {"wrote": False, "pending": "req-ACTIVE"}
assert state["noId"] == {"wrote": False, "pending": "req-ACTIVE"}
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_stale_or_absent_token_can_never_claim_the_slot() -> None:
"""The two ways a send stops owning the turn: superseded, or never owned it.
`null` is the queued send, which took no ticket at all. Stale is the send
whose turn ended -- or whose slot another turn's `processing` claimed --
while its POST was still in flight.
"""
state = _drive("""
S.running = true;
const token = openTurnTicket();
commitTurnTicket(token, { request_id: "req-A" });
activateTurnTicket("req-B"); // another turn started
const stale = acceptTurnTicket(token, { request_id: "req-A", queue_position: 0 });
const queued = acceptTurnTicket(null, { request_id: "req-C", queue_position: 0 });
console.log(JSON.stringify({
stale, queued, owns: ownsTurnTicket(token), pending: S.pendingRequestId
}));
""")
assert state["stale"] is False, "a superseded send reclaimed the slot"
assert state["queued"] is False, "a queued send with no ticket claimed the slot"
assert state["owns"] is False
assert state["pending"] == "req-B"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_two_queued_202s_in_any_order_leave_the_active_id_alone() -> None:
"""Driven, not asserted from a comment.
Both follow-ups took no ticket and both are answered as queued, so each is
refused twice over. Order does not matter, which is the point: they resolve
whenever the network says.
"""
state = _drive("""
S.running = true;
const a = openTurnTicket();
commitTurnTicket(a, { request_id: "req-A" });
const generationBefore = S.turnTicket;
const wrote = [
{ request_id: "req-C", queue_position: 2 },
{ request_id: "req-B", queue_position: 1 },
].map(accepted => acceptTurnTicket(null, accepted));
console.log(JSON.stringify({
wrote,
sameGeneration: generationBefore === S.turnTicket,
pending: S.pendingRequestId,
stillLive: acceptTurnTicket(a, { request_id: "req-A", queue_position: 0 })
}));
""")
assert state["wrote"] == [False, False]
assert state["sameGeneration"] is True, "a queued send advanced the generation"
assert state["pending"] == "req-A"
assert state["stillLive"] is True, "the running turn's own ticket was invalidated"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_rejected_follow_up_does_not_tear_down_the_running_turn() -> None:
"""The `false -> true` window, which the snapshot gets wrong.
A send begins while the session is idle, so it takes a ticket. Before its
POST is answered, another turn starts and its `processing` claims the slot.
The POST then fails. Deciding from `queueing` (still false) tears down the
turn that is actually running; deciding from ownership does not.
"""
state = _drive("""
S.running = true;
const mine = openTurnTicket(); // began idle, took a ticket
activateTurnTicket("req-OTHER"); // someone else's turn started
console.log(JSON.stringify({
tearsDown: ownsTurnTicket(mine),
pending: S.pendingRequestId
}));
""")
assert (
state["tearsDown"] is False
), "a rejected follow-up would tear down a live turn"
assert state["pending"] == "req-OTHER"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_ownership_is_false_before_any_turn_has_started() -> None:
"""`undefined === undefined` is the trap.
`S.turnTicket` is unset until the first send, so a bare `token ===
S.turnTicket` answers *true* for a caller that passes nothing -- and the
catch would tear down a turn that does not exist.
"""
state = _drive("""
S.turnTicket = undefined;
console.log(JSON.stringify({
nothing: ownsTurnTicket(undefined),
queued: ownsTurnTicket(null),
first: (() => { const t = openTurnTicket(); return ownsTurnTicket(t); })()
}));
""")
assert state["nothing"] is False, "a caller with no ticket owned the turn"
assert state["queued"] is False
assert state["first"] is True
def test_no_correctness_branch_reads_the_dispatch_snapshot() -> None:
"""`queueing` may style the bubble; it may not decide who owns the turn.
Every later read of it is stale by construction -- and one of them, the
rebind prompt, is modal, so the window is however long the user takes.
"""
body = _fn("send")
for branch in (
"if (S.running && !queueing) turnDone",
'else if (S.running) turnDone("failed")',
'if (queueing) w.classList.add("cancelled")',
):
assert branch not in body, branch
assert body.count("ownsTurnTicket(turnTicket)") >= 2, body[-3000:]
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_ticket_retired_by_a_queued_answer_can_never_claim_the_slot() -> None:
"""`queue_position: 1` on a send that thought it was active.
The ticket was taken in good faith at dispatch, and the server's answer
says otherwise. Retiring it is what stops a later resolution -- a retry, a
duplicated promise -- from using it, and `pendingRequestId` is left alone
because it does not belong to this turn.
"""
state = _drive("""
S.running = false;
const token = openTurnTicket();
S.running = true; // this send locked the UI
const accepted = { request_id: "req-B", queue_position: 1 };
const wrote = acceptTurnTicket(token, accepted);
const retired = retireTurnTicket(token);
console.log(JSON.stringify({
wrote, retired,
ownsAfter: ownsTurnTicket(token),
reclaim: acceptTurnTicket(token, { request_id: "req-B", queue_position: 0 })
}));
""")
assert state["wrote"] is False
assert state["retired"] is True
assert state["ownsAfter"] is False
assert state["reclaim"] is False, "a retired ticket claimed the slot later"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_retiring_never_touches_a_turn_it_does_not_own() -> None:
state = _drive("""
S.running = true;
const mine = openTurnTicket();
activateTurnTicket("req-OTHER"); // another turn took over
const retired = retireTurnTicket(mine);
console.log(JSON.stringify({ retired, pending: S.pendingRequestId }));
""")
assert state["retired"] is False
assert state["pending"] == "req-OTHER", "retiring cleared another turn's id"
# --- a terminal event for a turn that is no longer on screen -------------------
def test_every_late_turn_event_goes_through_the_stale_filter() -> None:
"""Prose as well as the terminal.
A failure that arrives after the next turn started would otherwise wipe the
running turn's stream and print its predecessor's error into it.
"""
assert "if (mine(fid) && !isStaleTurnEvent(m)) startStream();" in APP_JS
assert "if (mine(fid) && !isStaleTurnEvent(m)) feed(" in APP_JS
assert "if (isStaleTurnEvent(m)) scheduleWorkbenchRefresh();" in APP_JS
assert "activateTurnTicket(m.request_id, m.execution_id)" in APP_JS
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_an_idless_processing_retires_tickets_without_forgetting_who_is_running():
"""An older daemon sends `processing` with no ids at all.
The generation must still advance -- another turn is running, so every
ticket in flight is stale. But clearing the identity there throws away what
the 202 already gave us, leaving the running turn's own failure with
nothing to quote and nothing to filter on.
"""
state = _drive("""
S.running = true;
const token = openTurnTicket();
commitTurnTicket(token, { request_id: "req-A", execution_id: "exec-A" });
const before = S.turnTicket;
activateTurnTicket(undefined, undefined);
console.log(JSON.stringify({
bumped: S.turnTicket === before + 1,
request: S.pendingRequestId,
execution: S.pendingExecutionId,
stale: ownsTurnTicket(token)
}));
""")
assert state["bumped"] is True, "an idless processing left stale tickets valid"
assert state["request"] == "req-A", "the running turn's id was forgotten"
assert state["execution"] == "exec-A"
assert state["stale"] is False
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_late_terminal_from_the_previous_execution_is_stale() -> None:
"""processing(A) -> processing(B) -> failed(A), the order that reproduces.
A fails inside the turn, persists its row, and finishes unwinding only
after B has been promoted out of the queue. Acting on A's terminal closes
B's turn and unlocks the composer under a turn that is still running.
"""
state = _drive("""
activateTurnTicket("req-A", "exec-A");
activateTurnTicket("req-B", "exec-B");
console.log(JSON.stringify({
lateA: isStaleTurnEvent({ request_id: "req-A", execution_id: "exec-A" }),
ownB: isStaleTurnEvent({ request_id: "req-B", execution_id: "exec-B" })
}));
""")
assert state["lateA"] is True, "A's terminal would have closed B's turn"
assert state["ownB"] is False, "B's own terminal was refused"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_reused_request_id_is_still_told_apart_by_execution() -> None:
"""Clients may reuse `X-Request-Id`, so A and B can share one.
A filter that compared only request ids would call A's late terminal
current, which is the whole reason the execution id is on the wire.
"""
state = _drive("""
activateTurnTicket("req-same", "exec-A");
activateTurnTicket("req-same", "exec-B");
console.log(JSON.stringify({
lateA: isStaleTurnEvent({ request_id: "req-same", execution_id: "exec-A" }),
ownB: isStaleTurnEvent({ request_id: "req-same", execution_id: "exec-B" })
}));
""")
assert state["lateA"] is True, "two turns sharing a request id were confused"
assert state["ownB"] is False
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_an_older_daemon_without_identities_still_closes_its_turns() -> None:
"""The filter must not strand a client talking to a server that predates it.
Neither side offering any identity is the pre-identity contract, and
treating that as stale would leave every turn open forever. One side silent
is the mixed case -- also not evidence of staleness.
"""
state = _drive("""
const out = {};
closeTurnTicket();
out.bothSilent = isStaleTurnEvent({});
activateTurnTicket("req-A", undefined); // daemon sends no exec id
out.reqOnlyMatch = isStaleTurnEvent({ request_id: "req-A" });
out.reqOnlyOther = isStaleTurnEvent({ request_id: "req-Z" });
out.execArrivesLater = isStaleTurnEvent({ execution_id: "exec-A" });
console.log(JSON.stringify(out));
""")
assert state["bothSilent"] is False
assert state["reqOnlyMatch"] is False
assert state["reqOnlyOther"] is True, "the request-id fallback stopped working"
assert state["execArrivesLater"] is False, "one side silent is not staleness"
# --- plan turns take a ticket too ----------------------------------------------
def test_the_three_plan_turns_share_one_generation_owned_dispatch() -> None:
"""Each used to lock the UI after its own 202, or without a ticket at all."""
for name in ("approvePlan", "resumePlan", "revisePlan"):
assert "dispatchPlanTurn(" in _fn(name), name
body = _fn("dispatchPlanTurn")
# The ticket and the lock come BEFORE the POST, or a terminal event that
# beats the 202 cannot invalidate anything.
assert body.index("openTurnTicket()") < body.index("await api("), body
assert body.index("S.running = true") < body.index("await api(")
assert "if (!ownsTurnTicket(token)) return true;" in body
assert "commitTurnTicket(token, accepted || {})" in body
assert 'if (ownsTurnTicket(token)) turnDone("failed");' in body
def test_approve_still_leaves_plan_mode_on_a_failed_dispatch() -> None:
"""The toggle follows acceptance; a refused approve is still a draft."""
body = _fn("approvePlan")
assert "if (await dispatchPlanTurn(" in body, body
assert "S.planMode = false" in body
# --- the plan dispatcher, actually executed -----------------------------------
#
# Lifting the ticket helpers and simulating what `dispatchPlanTurn` *would* do
# is not evidence about `dispatchPlanTurn`: those tests stay green while the
# shipped action is wrong. These run it, against a real pending promise.
def _drive_plan(program: str) -> dict:
"""Run the shipped `dispatchPlanTurn` under node with a deferred `api`."""
lifted = "\n".join(
_fn(name)
for name in (
"openTurnTicket",
"commitTurnTicket",
"closeTurnTicket",
"activateTurnTicket",
"ownsTurnTicket",
"retireTurnTicket",
"isStaleTurnEvent",
"dispatchPlanTurn",
)
)
harness = """
const S = { currentId: "frame-1", running: false, _openGen: 3 };
const calls = { api: 0, resume: 0, turnDone: 0, hint: 0 };
let settle = null;
const api = (_path, _opts) => {
calls.api += 1;
return new Promise((resolve, reject) => { settle = { resolve, reject }; });
};
const $ = () => ({ classList: { add() {}, remove() {} } });
const hint = () => { calls.hint += 1; };
const t = (key) => key;
const apiErrorText = (e) => String(e && e.message ? e.message : e);
const enableComposer = () => {};
const resumeWatch = () => { calls.resume += 1; };
const turnDone = (_status) => { calls.turnDone += 1; S.running = false; closeTurnTicket(); };
const tick = () => new Promise(r => setTimeout(r, 0));
"""
out = subprocess.run(
[NODE, "--input-type=module", "-e", harness + lifted + "\n" + program],
capture_output=True,
text=True,
timeout=60,
)
assert out.returncode == 0, out.stderr[:1000]
return json.loads(out.stdout.strip().splitlines()[-1])
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_plan_dispatch_while_a_turn_runs_is_refused_outright() -> None:
"""A double-click, or a plan control pressed during someone else's turn.
A second ticket makes the newer request the owner: a 409 then tears down
the turn that is actually running, and an acceptance replaces its identity
so its own terminal event is judged stale and never closes it.
"""
state = _drive_plan("""
activateTurnTicket("req-A", "exec-A");
S.running = true;
const generation = S.turnTicket;
const refused = await dispatchPlanTurn("/plan/approve", {}, "h", "k");
console.log(JSON.stringify({
refused, api: calls.api, turnDone: calls.turnDone,
sameGeneration: S.turnTicket === generation,
running: S.running,
pending: S.pendingRequestId, execution: S.pendingExecutionId
}));
""")
assert state["refused"] is False
assert state["api"] == 0, "a second plan request was sent during a running turn"
assert state["turnDone"] == 0
assert state["sameGeneration"] is True
assert state["running"] is True
assert state["pending"] == "req-A" and state["execution"] == "exec-A"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_second_plan_click_while_the_first_is_in_flight_sends_nothing() -> None:
"""The first POST is still pending; the button is pressed again."""
state = _drive_plan("""
const first = dispatchPlanTurn("/plan/approve", {}, "h", "k");
await tick();
const second = await dispatchPlanTurn("/plan/approve", {}, "h", "k");
settle.resolve({ request_id: "req-1", execution_id: "exec-1" });
const firstResult = await first;
console.log(JSON.stringify({
second, firstResult, api: calls.api,
pending: S.pendingRequestId, execution: S.pendingExecutionId
}));
""")
assert state["second"] is False
assert state["api"] == 1, "the second click sent its own request"
assert state["firstResult"] is True
assert state["pending"] == "req-1"
assert state["execution"] == "exec-1"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_terminal_before_the_plan_202_does_not_relock_the_composer() -> None:
"""The stuck-session case, run rather than simulated."""
state = _drive_plan("""
const pending = dispatchPlanTurn("/plan/approve", {}, "h", "k");
await tick();
turnDone("failed"); // the WS beat the 202
settle.resolve({ request_id: "req-late", execution_id: "exec-late" });
await pending;
console.log(JSON.stringify({
running: S.running, resume: calls.resume,
pending: S.pendingRequestId, execution: S.pendingExecutionId
}));
""")
assert state["running"] is False, "the late 202 locked the composer again"
assert state["resume"] == 0, "the watchdog was re-armed for a finished turn"
assert state.get("pending") in (None, "")
assert state.get("execution") in (None, "")
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_rejected_plan_post_does_not_end_the_turn_that_took_over() -> None:
state = _drive_plan("""
const pending = dispatchPlanTurn("/plan/approve", {}, "h", "k");
await tick();
activateTurnTicket("req-B", "exec-B"); // B took over mid-flight
settle.reject(new Error("409 plan_not_paused"));
await pending;
console.log(JSON.stringify({
turnDone: calls.turnDone, running: S.running,
pending: S.pendingRequestId, execution: S.pendingExecutionId
}));
""")
assert state["turnDone"] == 0, "a rejected plan POST ended another turn"
assert state["running"] is True
assert state["pending"] == "req-B" and state["execution"] == "exec-B"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_a_legacy_202_without_ids_still_leaves_the_turn_running() -> None:
"""An older daemon answers with neither id, and its turn must still work."""
state = _drive_plan("""
const pending = dispatchPlanTurn("/plan/approve", {}, "h", "k");
await tick();
settle.resolve({ status: "accepted", job_id: "j-1" });
await pending;
const before = { running: S.running, resume: calls.resume };
const stale = isStaleTurnEvent({ type: "frame_update", status: "failed" });
if (!stale) turnDone("failed");
console.log(JSON.stringify({ before, stale, running: S.running }));
""")
assert state["before"] == {"running": True, "resume": 1}
assert state["stale"] is False, "an idless terminal could never close the turn"
assert state["running"] is False
def test_action_timeline_ledger_row_reports_state_it_cannot_fabricate() -> None:
"""The row must not present unknown state as a known value.
Three of these were regressions against the card the ledger replaced:
status and the attempt error were reduced to an icon colour, an absent
usage row printed a fabricated ``0``, and a time brush resolved through the
paint model so an action with no execution attempt vanished from the list.
"""
row = _extract_js_function(APP_JS, "actionTimelineLedgerRow")
overlap = _extract_js_function(APP_JS, "actionTimelineSelectionOverlaps")
filtered = _extract_js_function(APP_JS, "filteredActionTimelineGroups")
epoch = _extract_js_function(APP_JS, "timelineEpochMs")
duration = _extract_js_function(APP_JS, "timelineDurationMs")
search_doc = _extract_js_function(APP_JS, "actionTimelineSearchDocument")
creator = _extract_js_function(APP_JS, "createActionTimelineView")
renderer = _extract_js_function(APP_JS, "renderActionTimeline")
history = _extract_js_function(APP_JS, "syncActionTimelineHistoryState")
# Unknown token usage stays unknown; a real 0 and an absent row differ.
assert 'group.usage ? String(timelineTokenTotal(group.usage)) : "—"' in row
# Status and the attempt error reach text, not just a colour class.
assert "timeline-ledger-status" in row and "statusNoteworthy" in row
assert "latest.error" in row or "rowError" in row
assert ".timeline-ledger-status" in STYLE_CSS
# A time filter asks when an action happened, not whether it was painted.
assert "group.created_at" in overlap and "if (!item) return false" not in overlap
assert "selection, group)" in filtered
# One timestamp parser, shared, and never one Date cannot represent.
assert "Date.parse" in epoch and "TIMELINE_MAX_EPOCH_MS" in epoch
assert "const parse = timelineEpochMs" in duration
# Search covers the Kind label the placeholder advertises.
assert "timelineKind(group" in search_doc
assert 't("timeline.kind." + kind)' in search_doc
assert "cached.lang === LANG" in _extract_js_function(
APP_JS, "syncActionTimelineSearchIndex"
)
# The CSS strips implicit table roles, so they are declared explicitly.
for role in ('"role", "table"', '"role", "rowgroup"', '"role", "columnheader"'):
assert role in creator
assert '"role", "row"' in row and '"role", "cell"' in row
# Any render that does not re-append the region must destroy the view.
assert "else {" in renderer and "destroyActionTimelineView();" in renderer
# The history slot is measured with its previous reservation still applied.
# Clearing it first reads 0 on an already-reserved slot, collapsing the band
# during a prepend and moving the compensated row -- browser_smoke's
# "history prepend N moved the visible anchor" case.
assert "const previousHeight = target.getBoundingClientRect().height" in history
assert 'target.replaceChildren(); target.style.minHeight = ""' in history
# --- persist-first candidate delivery ---------------------------------------
def test_candidate_events_are_bound_to_durable_message_rows() -> None:
events = _fn("onEvent")
auto_terminal = re.search(
r'else if \(m\.type === "auto_run_terminal"\)(?P.*?)\n else if',
events,
flags=re.DOTALL,
)
assert auto_terminal, "onEvent must keep the Auto Mode terminal branch"
assert "scheduleWorkbenchRefresh" in auto_terminal.group("body")
assert "setLiveReviewBadge" not in auto_terminal.group("body")
assert "applyCandidateResolution(m, fid)" in events
assert "applyFinalReviewStatus(m, fid)" in events
stored = _fn("renderStored")
selector = _fn("candidateMessageNode")
committed = _fn("candidateReplacementCommitted")
replacement = _fn("replaceMessageAnswer")
dedupe = _fn("storedCandidateOwnsChunk")
assert "rememberCandidateIdentity(w, m)" in stored
assert "identity.messageId" in selector and "dataset.messageId" in selector
assert "value.durable === true" in committed
assert "value.delivered === true" in committed
assert "value.replaced === true" in committed
assert "addMsgActions(node, text)" in replacement
assert "candidateMessageNode(value)" in dedupe
assert "discardDuplicateLiveCandidate(target, value)" in dedupe
def _drive_candidate_resolution(program: str) -> dict:
lifted = "\n".join(
_fn(name)
for name in (
"candidateIdentityText",
"candidateIdentity",
"rememberCandidateIdentity",
"candidateNodeMatches",
"candidateMessageNode",
"reviewStatusFrom",
"reviewTruthFrom",
"candidateReplacementText",
"candidateReplacementCommitted",
"applyCandidateResolution",
"applyFinalReviewStatus",
)
)
harness = """
const S = { stream: null };
const nodes = [];
const host = { querySelectorAll: () => nodes };
const $ = selector => selector === "#messages" ? host : null;
const calls = { replace: [], badge: [], resync: [] };
function node(messageId, turnId, executionId, status) {
return { dataset: {
...(messageId ? { messageId } : {}),
...(turnId ? { turnId } : {}),
...(executionId ? { executionId } : {}),
...(status ? { reviewStatus: status } : {}),
} };
}
function discardDuplicateLiveCandidate() {}
function replaceMessageAnswer(target, text) {
calls.replace.push({ id: target.dataset.messageId || "", text });
target.text = text;
return true;
}
function setMessageReviewBadge(target, status, truth) {
calls.badge.push({ id: target.dataset.messageId || "", status, truth });
target.dataset.reviewStatus = status;
return true;
}
function scheduleConversationResync(fid) { calls.resync.push(fid); }
"""
out = subprocess.run(
[NODE, "--input-type=module", "-e", harness + lifted + "\n" + program],
capture_output=True,
text=True,
timeout=60,
)
assert out.returncode == 0, out.stderr[:1200]
return json.loads(out.stdout.strip().splitlines()[-1])
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_candidate_replacement_requires_exact_durable_delivery() -> None:
state = _drive_candidate_resolution("""
const first = node("m-1", "turn-1", "exec-1", "candidate");
const exact = node("m-2", "turn-2", "exec-2", "candidate");
nodes.push(first, exact);
const result = applyCandidateResolution({
message_id: "m-2", turn_id: "turn-2", execution_id: "exec-2",
replaced: true, delivered: true, durable: true, text: "reviewed answer",
review_status: "verified", user_truth: "Verified"
}, "frame-1");
console.log(JSON.stringify({ result, calls, first, exact }));
""")
assert state["result"] == {
"targetFound": True,
"replacementApplied": True,
"badgeApplied": True,
}
assert state["calls"]["replace"] == [{"id": "m-2", "text": "reviewed answer"}]
assert state["calls"]["badge"] == [
{"id": "m-2", "status": "verified", "truth": "Verified"}
]
assert state["calls"]["resync"] == []
assert "text" not in state["first"]
assert state["exact"]["dataset"]["candidateResolved"] == "true"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_unchanged_durable_candidate_is_verified_without_rewriting_text() -> None:
state = _drive_candidate_resolution("""
const exact = node("m-1", "turn-1", "exec-1", "candidate");
exact.text = "reviewed unchanged answer";
nodes.push(exact);
const result = applyCandidateResolution({
message_id: "m-1", turn_id: "turn-1", execution_id: "exec-1",
replaced: false, delivered: true, durable: true,
review_status: "verified", user_truth: "Verified"
}, "frame-1");
console.log(JSON.stringify({ result, calls, exact }));
""")
assert state["result"] == {
"targetFound": True,
"replacementApplied": False,
"badgeApplied": True,
}
assert state["calls"]["replace"] == []
assert state["calls"]["badge"] == [
{"id": "m-1", "status": "verified", "truth": "Verified"}
]
assert state["exact"]["text"] == "reviewed unchanged answer"
assert state["exact"]["dataset"]["candidateResolved"] == "true"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_final_frame_cannot_verify_an_unresolved_candidate() -> None:
state = _drive_candidate_resolution("""
const exact = node("m-1", "turn-1", "exec-1", "candidate");
nodes.push(exact);
const refused = applyFinalReviewStatus({
message_id: "m-1", review_status: "verified"
}, "frame-1");
exact.dataset.candidateResolved = "true";
const accepted = applyFinalReviewStatus({
message_id: "m-1", review_status: "verified"
}, "frame-1");
console.log(JSON.stringify({ refused, accepted, calls, exact }));
""")
assert state["refused"] is False
assert state["accepted"] is True
assert state["calls"]["resync"] == ["frame-1"]
assert state["calls"]["badge"] == [{"id": "m-1", "status": "verified", "truth": ""}]
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
@pytest.mark.parametrize(
"receipt",
[
{"delivered": False, "durable": True},
{"delivered": True, "durable": False},
{"delivered": True},
],
)
def test_candidate_replacement_failure_keeps_provisional_text_and_refetches(
receipt: dict[str, bool],
) -> None:
state = _drive_candidate_resolution("""
const exact = node("m-1", "turn-1", "exec-1", "candidate");
exact.text = "provisional answer";
nodes.push(exact);
const receipt = %s;
const result = applyCandidateResolution({
message_id: "m-1", turn_id: "turn-1", execution_id: "exec-1",
replaced: true, text: "reviewed answer", review_status: "verified",
...receipt
}, "frame-1");
console.log(JSON.stringify({ result, calls, exact }));
""" % json.dumps(receipt))
assert state["result"]["replacementApplied"] is False
assert state["result"]["badgeApplied"] is False
assert state["calls"]["replace"] == []
assert state["calls"]["badge"] == []
assert state["calls"]["resync"], "failed promotion must request REST truth"
assert state["exact"]["text"] == "provisional answer"
assert state["exact"]["dataset"]["reviewStatus"] == "candidate"
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_missing_exact_candidate_row_never_falls_back_to_live_turn_identity() -> None:
state = _drive_candidate_resolution("""
const live = node("", "turn-1", "exec-1", "candidate");
live.text = "provisional answer";
nodes.push(live);
S.stream = { wrap: live };
const result = applyCandidateResolution({
message_id: "m-durable", turn_id: "turn-1", execution_id: "exec-1",
replaced: true, delivered: true, durable: true, text: "reviewed answer",
review_status: "verified"
}, "frame-1");
console.log(JSON.stringify({ result, calls, live }));
""")
assert state["result"]["targetFound"] is False
assert state["calls"]["replace"] == []
assert state["calls"]["badge"] == []
assert state["calls"]["resync"]
assert state["live"]["text"] == "provisional answer"
assert "messageId" not in state["live"]["dataset"]
def _drive_candidate_replay(program: str) -> dict:
lifted = "\n".join(
_fn(name)
for name in (
"candidateIdentityText",
"candidateIdentity",
"rememberCandidateIdentity",
"candidateNodeMatches",
"candidateMessageNode",
"reviewStatusFrom",
"discardDuplicateLiveCandidate",
"storedCandidateOwnsChunk",
)
)
harness = """
const S = { stream: null };
const nodes = [];
const host = { querySelectorAll: () => nodes };
const $ = selector => selector === "#messages" ? host : null;
const calls = { badges: 0, removed: 0 };
function stored(messageId, turnId, executionId, status) {
return { dataset: { messageId, turnId, executionId, reviewStatus: status } };
}
function live(turnId, executionId) {
return {
dataset: { turnId, executionId },
remove() { calls.removed += 1; }
};
}
function setMessageReviewBadge(target, status) {
calls.badges += 1; target.dataset.reviewStatus = status; return true;
}
"""
out = subprocess.run(
[NODE, "--input-type=module", "-e", harness + lifted + "\n" + program],
capture_output=True,
text=True,
timeout=60,
)
assert out.returncode == 0, out.stderr[:1200]
return json.loads(out.stdout.strip().splitlines()[-1])
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_replayed_provisional_chunks_defer_to_the_stored_candidate() -> None:
state = _drive_candidate_replay("""
const row = stored("m-1", "turn-1", "exec-1", "candidate");
const duplicate = live("turn-1", "exec-1");
nodes.push(row, duplicate); S.stream = { wrap: duplicate };
const owned = storedCandidateOwnsChunk({
turn_id: "turn-1", execution_id: "exec-1", provisional: true,
block_type: "text", chunk: "provisional answer"
});
console.log(JSON.stringify({ owned, calls, stream: S.stream }));
""")
assert state["owned"] is True
assert state["calls"]["removed"] == 1
assert state["stream"] is None
@pytest.mark.skipif(NODE is None, reason="no node on this machine")
def test_replay_dedupe_does_not_claim_an_unrelated_or_final_chunk() -> None:
state = _drive_candidate_replay("""
const row = stored("m-1", "turn-1", "exec-1", "verified");
nodes.push(row);
const otherTurn = storedCandidateOwnsChunk({
turn_id: "turn-2", execution_id: "exec-2", provisional: true,
block_type: "text", chunk: "different turn"
});
const finalChunk = storedCandidateOwnsChunk({
turn_id: "turn-1", execution_id: "exec-1", provisional: false,
block_type: "text", chunk: "ordinary final text"
});
const replay = storedCandidateOwnsChunk({
message_id: "m-1", turn_id: "turn-1", execution_id: "exec-1",
provisional: true, block_type: "text", chunk: "old candidate"
});
console.log(JSON.stringify({ otherTurn, finalChunk, replay, calls, row }));
""")
assert state["otherTurn"] is False
assert state["finalChunk"] is False
assert state["replay"] is True
assert state["calls"]["badges"] == 0, "replay must not demote a final REST row"
assert state["row"]["dataset"]["reviewStatus"] == "verified"