feat: model Worker Console display states

This commit is contained in:
2026-09-29 05:48:05 +09:00
parent f33e737d5c
commit 26f1de976a
5 changed files with 762 additions and 12 deletions
@@ -0,0 +1,150 @@
<script lang="ts">
import type { ConsoleDisplayState } from "./console-display-state";
type Props = {
state: ConsoleDisplayState;
hasContent: boolean;
hasUnfilteredContent: boolean;
onRetry: () => void;
};
let { state, hasContent, hasUnfilteredContent, onRetry }: Props = $props();
</script>
{#if state.kind === "loading"}
<div class="console-display-state loading" role="status" aria-live="polite">
<span class="loading-spinner" aria-hidden="true"></span>
<span>
<strong>Loading conversation</strong>
{#if state.stage === "session"}
<small>Checking Worker Session availability…</small>
{:else}
<small>Waiting for the initial conversation snapshot…</small>
{/if}
</span>
</div>
{:else if state.kind === "unavailable" || state.kind === "failed"}
<div class="console-display-state unavailable" role="alert">
<span>
<strong>
{state.kind === "unavailable"
? "Conversation unavailable"
: "Unable to load conversation"}
</strong>
<small>{state.reason}</small>
</span>
<button type="button" onclick={onRetry}>Retry</button>
</div>
{:else}
{#if state.kind === "stale"}
<div class="console-display-state stale" role="status" aria-live="polite">
<span>
<strong>
{state.phase === "reconnecting"
? "Refreshing conversation"
: "Conversation updates are unavailable"}
</strong>
<small>{state.reason} Previously loaded conversation is preserved.</small>
</span>
<button type="button" onclick={onRetry}>Retry</button>
</div>
{:else if state.source === "retained"}
<div class="console-display-state retained" role="status">
<span>
<strong>Read-only retained conversation</strong>
<small>This saved Session snapshot cannot receive live updates.</small>
</span>
</div>
{/if}
{#if state.kind === "ready" && !hasContent}
<div class="console-display-state empty" role="status">
<span>
<strong>No conversation to display</strong>
<small>
{hasUnfilteredContent
? "No console items match the current display mode."
: "This Worker view has no conversation history yet."}
</small>
</span>
</div>
{/if}
{/if}
<style>
.console-display-state {
display: flex;
align-items: center;
justify-content: space-between;
gap: var(--space-3);
border: 1px solid var(--line);
border-radius: 12px;
padding: var(--space-4);
color: var(--text-muted);
background: color-mix(in srgb, var(--bg-raised) 78%, transparent);
}
.console-display-state > span {
display: grid;
gap: var(--space-1);
}
.console-display-state strong {
color: var(--text-strong);
}
.console-display-state small {
font-size: var(--font-size-compact);
line-height: var(--line-height-compact);
}
.console-display-state button {
flex: 0 0 auto;
border: 1px solid var(--line);
border-radius: 999px;
padding: 0.4rem 0.75rem;
background: var(--bg);
color: var(--text-strong);
cursor: pointer;
font: inherit;
font-size: var(--font-size-compact);
font-weight: 700;
}
.console-display-state.loading {
justify-content: flex-start;
}
.console-display-state.unavailable {
border-color: color-mix(in srgb, var(--danger) 55%, var(--line));
}
.console-display-state.stale {
border-color: color-mix(in srgb, var(--tui-yellow) 48%, var(--line));
}
.console-display-state.retained {
border-color: color-mix(in srgb, var(--tui-cyan) 42%, var(--line));
}
.loading-spinner {
width: 1rem;
height: 1rem;
flex: 0 0 auto;
border: 2px solid color-mix(in srgb, var(--accent) 30%, var(--line));
border-top-color: var(--accent);
border-radius: 999px;
}
@media (prefers-reduced-motion: no-preference) {
.loading-spinner {
animation: console-loading-spin 800ms linear infinite;
}
}
@keyframes console-loading-spin {
to {
transform: rotate(360deg);
}
}
</style>
@@ -0,0 +1,37 @@
export type ConsoleDisplaySource = "live" | "retained";
export type ConsoleDisplayState =
| { kind: "loading"; stage: "session" | "snapshot" }
| { kind: "ready"; source: ConsoleDisplaySource }
| { kind: "unavailable"; reason: string }
| { kind: "failed"; reason: string }
| {
kind: "stale";
source: ConsoleDisplaySource;
phase: "reconnecting" | "disconnected";
reason: string;
};
const MAX_CONSOLE_REASON_LENGTH = 240;
export function boundedConsoleReason(
reason: string | null | undefined,
fallback: string,
): string {
const normalized = reason?.replace(/\s+/g, " ").trim() || fallback;
return normalized.slice(0, MAX_CONSOLE_REASON_LENGTH);
}
export function displayedConsoleSource(
state: ConsoleDisplayState,
): ConsoleDisplaySource | null {
switch (state.kind) {
case "ready":
case "stale":
return state.source;
case "loading":
case "unavailable":
case "failed":
return null;
}
}
@@ -555,6 +555,56 @@ Deno.test("Worker Console uses protocol observation events without transcript fe
); );
}); });
Deno.test("Worker Console models availability separately from projected lines", async () => {
const consolePage = await Deno.readTextFile(
new URL(
"./../../../routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte",
import.meta.url,
),
);
const displayState = await Deno.readTextFile(
new URL("./ConsoleDisplayState.svelte", import.meta.url),
);
const displayModel = await Deno.readTextFile(
new URL("./console-display-state.ts", import.meta.url),
);
assert(
consolePage.includes("consoleDisplayState") &&
consolePage.includes("pendingInitialSnapshotApplication") &&
consolePage.includes('source: "retained"') &&
consolePage.includes('source: "live"') &&
consolePage.includes("completeInitialSnapshot(") &&
consolePage.includes("showConsoleUnavailable(") &&
consolePage.includes("retryConsoleLoad") &&
!consolePage.includes(
"{#if lines.length === 0}\n <p>No console output",
),
"Console availability must advance on snapshot projection rather than infer loading, failure, or empty history from line count",
);
assert(
displayModel.includes('kind: "loading"') &&
displayModel.includes('kind: "ready"') &&
displayModel.includes('kind: "unavailable"') &&
displayModel.includes('kind: "failed"') &&
displayModel.includes('kind: "stale"') &&
displayModel.includes("MAX_CONSOLE_REASON_LENGTH = 240"),
"Console display state must explicitly model loading, ready, unavailable, failed, and stale outcomes with bounded reasons",
);
assert(
displayState.includes('role="status"') &&
displayState.includes('role="alert"') &&
displayState.includes('aria-live="polite"') &&
displayState.includes("prefers-reduced-motion: no-preference") &&
displayState.includes("Read-only retained conversation") &&
displayState.includes(
"No console items match the current display mode.",
) &&
displayState.includes(">Retry</button>"),
"Console state presentation must expose accessible loading/errors, retained and filtered-empty copy, retry, and reduced-motion-safe animation",
);
});
Deno.test("Worker Console owns its narrower centered shell width", async () => { Deno.test("Worker Console owns its narrower centered shell width", async () => {
const page = await Deno.readTextFile( const page = await Deno.readTextFile(
new URL( new URL(
@@ -1,6 +1,7 @@
<script lang="ts"> <script lang="ts">
import { tick, untrack, type SvelteComponent } from "svelte"; import { tick, untrack, type SvelteComponent } from "svelte";
import ConsoleLineItem from "$lib/workspace/console/ConsoleLineItem.svelte"; import ConsoleLineItem from "$lib/workspace/console/ConsoleLineItem.svelte";
import ConsoleDisplayStateView from "$lib/workspace/console/ConsoleDisplayState.svelte";
import ConsoleTasks from "$lib/workspace/console/ConsoleTasks.svelte"; import ConsoleTasks from "$lib/workspace/console/ConsoleTasks.svelte";
import ConsoleTimeline from "$lib/workspace/console/ConsoleTimeline.svelte"; import ConsoleTimeline from "$lib/workspace/console/ConsoleTimeline.svelte";
import ComposerInput from "$lib/workspace/console/ComposerInput.svelte"; import ComposerInput from "$lib/workspace/console/ComposerInput.svelte";
@@ -50,6 +51,12 @@
type ComposerAttachment, type ComposerAttachment,
} from "$lib/workspace/console/composer-attachments"; } from "$lib/workspace/console/composer-attachments";
import { pushWorkspaceAlert } from "$lib/workspace/alerts/store"; import { pushWorkspaceAlert } from "$lib/workspace/alerts/store";
import {
boundedConsoleReason,
displayedConsoleSource,
type ConsoleDisplaySource,
type ConsoleDisplayState,
} from "$lib/workspace/console/console-display-state";
import { workspaceApiPath } from "$lib/workspace/api/http"; import { workspaceApiPath } from "$lib/workspace/api/http";
import { workspaceMultiplexer, type WorkspaceMultiplexerSubscription } from "$lib/workspace/multiplexer"; import { workspaceMultiplexer, type WorkspaceMultiplexerSubscription } from "$lib/workspace/multiplexer";
import { import {
@@ -165,6 +172,10 @@
let protocolState = $state<"connecting" | "open" | "closed" | "error">( let protocolState = $state<"connecting" | "open" | "closed" | "error">(
"connecting", "connecting",
); );
let consoleDisplayState = $state<ConsoleDisplayState>({
kind: "loading",
stage: "session",
});
let protocolSubscription: WorkspaceMultiplexerSubscription | null = null; let protocolSubscription: WorkspaceMultiplexerSubscription | null = null;
let pendingSubmissions = $state<PendingSubmissionsSnapshot>({ let pendingSubmissions = $state<PendingSubmissionsSnapshot>({
revision: 0, revision: 0,
@@ -209,6 +220,11 @@
consoleProjector.snapshot(), consoleProjector.snapshot(),
); );
let pendingObservationEvents: ConsoleEventInput[] = []; let pendingObservationEvents: ConsoleEventInput[] = [];
let pendingInitialSnapshotApplication: {
eventId: string;
token: number;
source: ConsoleDisplaySource;
} | null = null;
let protocolEventSequence = 0; let protocolEventSequence = 0;
let pendingStreamDiagnostics: Diagnostic[] = []; let pendingStreamDiagnostics: Diagnostic[] = [];
let observationFlushHandle: number | null = null; let observationFlushHandle: number | null = null;
@@ -319,6 +335,79 @@
return nextReloadToken; return nextReloadToken;
} }
function beginConsoleLoad(reason = "Waiting for a fresh conversation snapshot.") {
const source = displayedConsoleSource(consoleDisplayState);
consoleDisplayState = source
? {
kind: "stale",
source,
phase: "reconnecting",
reason: boundedConsoleReason(reason, "Reconnecting to live updates."),
}
: { kind: "loading", stage: "session" };
protocolState = "connecting";
}
function waitForInitialSnapshot() {
if (displayedConsoleSource(consoleDisplayState) === null) {
consoleDisplayState = { kind: "loading", stage: "snapshot" };
}
}
function completeInitialSnapshot(
token: number,
source: ConsoleDisplaySource,
) {
if (token !== reloadToken) return;
consoleDisplayState = { kind: "ready", source };
}
function showConsoleUnavailable(message: string) {
const reason = boundedConsoleReason(
message,
"The Worker Session is not available.",
);
const source = displayedConsoleSource(consoleDisplayState);
consoleDisplayState = source
? { kind: "stale", source, phase: "disconnected", reason }
: { kind: "unavailable", reason };
}
function showConsoleFailure(message?: string) {
const reason = boundedConsoleReason(
message,
"The Worker Session connection ended before a snapshot was received.",
);
const source = displayedConsoleSource(consoleDisplayState);
consoleDisplayState = source
? { kind: "stale", source, phase: "disconnected", reason }
: { kind: "failed", reason };
}
function showConsoleReconnecting(message?: string) {
const source = displayedConsoleSource(consoleDisplayState);
if (!source) return;
consoleDisplayState = {
kind: "stale",
source,
phase: "reconnecting",
reason: boundedConsoleReason(
message,
"Waiting for a fresh conversation snapshot.",
),
};
}
function retryConsoleLoad() {
if (displayedConsoleSource(consoleDisplayState) === null) {
resetObservedEvents();
}
const target = consoleTarget;
beginConsoleLoad();
const token = advanceReloadToken();
if (!worker) void loadWorker(target, token);
}
function advanceEventObservedAtVersion() { function advanceEventObservedAtVersion() {
nextEventObservedAtVersion += 1; nextEventObservedAtVersion += 1;
eventObservedAtVersion = nextEventObservedAtVersion; eventObservedAtVersion = nextEventObservedAtVersion;
@@ -337,6 +426,7 @@
observationFlushHandle = null; observationFlushHandle = null;
} }
pendingObservationEvents = []; pendingObservationEvents = [];
pendingInitialSnapshotApplication = null;
pendingStreamDiagnostics = []; pendingStreamDiagnostics = [];
} }
@@ -352,8 +442,10 @@
function flushObservationBatch() { function flushObservationBatch() {
observationFlushHandle = null; observationFlushHandle = null;
const eventBatch = pendingObservationEvents; const eventBatch = pendingObservationEvents;
const initialSnapshotApplication = pendingInitialSnapshotApplication;
const diagnosticBatch = pendingStreamDiagnostics; const diagnosticBatch = pendingStreamDiagnostics;
pendingObservationEvents = []; pendingObservationEvents = [];
pendingInitialSnapshotApplication = null;
pendingStreamDiagnostics = []; pendingStreamDiagnostics = [];
if (eventBatch.length > 0) { if (eventBatch.length > 0) {
@@ -362,6 +454,17 @@
? "shutdown" ? "shutdown"
: workerStateFromSnapshot(consoleProjection.workerState); : workerStateFromSnapshot(consoleProjection.workerState);
advanceEventObservedAtVersion(); advanceEventObservedAtVersion();
if (
initialSnapshotApplication &&
eventBatch.some(
(event) => event.eventId === initialSnapshotApplication.eventId,
)
) {
completeInitialSnapshot(
initialSnapshotApplication.token,
initialSnapshotApplication.source,
);
}
} }
if (diagnosticBatch.length > 0) { if (diagnosticBatch.length > 0) {
@@ -369,7 +472,10 @@
} }
} }
function handleIncomingProtocolEvent(payload: ProtocolEvent) { function handleIncomingProtocolEvent(
payload: ProtocolEvent,
initialSnapshot?: { token: number; source: ConsoleDisplaySource },
) {
handleProtocolCommandEvent(payload); handleProtocolCommandEvent(payload);
if (payload.event === "snapshot") { if (payload.event === "snapshot") {
pendingSubmissions = payload.data.session.pending_submissions; pendingSubmissions = payload.data.session.pending_submissions;
@@ -397,6 +503,13 @@
event: payload, event: payload,
observedAtMs, observedAtMs,
}); });
if (initialSnapshot && payload.event === "snapshot") {
pendingInitialSnapshotApplication = {
eventId,
token: initialSnapshot.token,
source: initialSnapshot.source,
};
}
scheduleObservationFlush(); scheduleObservationFlush();
} }
@@ -960,6 +1073,7 @@
const action = workerSessionAction(observation); const action = workerSessionAction(observation);
if (action.kind === "show_unavailable") { if (action.kind === "show_unavailable") {
protocolState = "closed"; protocolState = "closed";
showConsoleUnavailable(action.message);
streamDiagnostics = [ streamDiagnostics = [
...streamDiagnostics, ...streamDiagnostics,
{ {
@@ -976,18 +1090,21 @@
targetWorker.state === "stopped" targetWorker.state === "stopped"
) { ) {
protocolState = "closed"; protocolState = "closed";
const message =
"The Session is live but the Worker is stopped. Reload to refresh its state.";
showConsoleFailure(message);
streamDiagnostics = [ streamDiagnostics = [
...streamDiagnostics, ...streamDiagnostics,
{ {
code: "worker_session_state_changed", code: "worker_session_state_changed",
severity: "warning", severity: "warning",
message: message,
"The Session is live but the Worker is stopped. Reload to refresh its state.",
}, },
]; ];
return; return;
} }
if (action.kind === "subscribe_live") { if (action.kind === "subscribe_live") {
waitForInitialSnapshot();
closeLiveSubscription = connectLiveProtocolTransport(token, target); closeLiveSubscription = connectLiveProtocolTransport(token, target);
return; return;
} }
@@ -1009,7 +1126,7 @@
in_flight: { responses: [], commands: [] }, in_flight: { responses: [], commands: [] },
internal_workers: [], internal_workers: [],
}, },
} as ProtocolEvent); } as ProtocolEvent, { token, source: "retained" });
protocolState = "closed"; protocolState = "closed";
streamDiagnostics = [ streamDiagnostics = [
...streamDiagnostics, ...streamDiagnostics,
@@ -1023,12 +1140,14 @@
.catch((error) => { .catch((error) => {
if (token !== reloadToken || controller.signal.aborted) return; if (token !== reloadToken || controller.signal.aborted) return;
protocolState = "error"; protocolState = "error";
const message = error instanceof Error ? error.message : String(error);
showConsoleFailure(message);
streamDiagnostics = [ streamDiagnostics = [
...streamDiagnostics, ...streamDiagnostics,
{ {
code: "worker_session_observation_failed", code: "worker_session_observation_failed",
severity: "warning", severity: "warning",
message: error instanceof Error ? error.message : String(error), message,
}, },
]; ];
}); });
@@ -1055,8 +1174,22 @@
frame.message.result === "subscribed" && frame.message.result === "subscribed" &&
frame.message.payload.snapshot.topic === "worker_protocol" frame.message.payload.snapshot.topic === "worker_protocol"
) { ) {
let initialSnapshotQueued = false;
for (const event of frame.message.payload.snapshot.data.events) { for (const event of frame.message.payload.snapshot.data.events) {
handleIncomingProtocolEvent(event); if (!initialSnapshotQueued && event.event === "snapshot") {
handleIncomingProtocolEvent(event, {
token,
source: "live",
});
initialSnapshotQueued = true;
} else {
handleIncomingProtocolEvent(event);
}
}
if (!initialSnapshotQueued) {
throw new Error(
"The live subscription did not provide an initial snapshot.",
);
} }
protocolState = "open"; protocolState = "open";
} else if ( } else if (
@@ -1070,6 +1203,7 @@
frame.message.event === "subscription_closed" frame.message.event === "subscription_closed"
) { ) {
protocolState = "closed"; protocolState = "closed";
showConsoleFailure(frame.message.data.message);
rejectPendingCompletion(new Error(frame.message.data.message)); rejectPendingCompletion(new Error(frame.message.data.message));
} else if ( } else if (
frame.frame === "response" && frame.frame === "response" &&
@@ -1079,20 +1213,29 @@
throw new Error(frame.message.payload.message); throw new Error(frame.message.payload.message);
} }
} catch (error) { } catch (error) {
protocolState = "error";
const message = error instanceof Error
? error.message
: String(error);
showConsoleFailure(message);
streamDiagnostics = [ streamDiagnostics = [
...streamDiagnostics, ...streamDiagnostics,
{ {
code: "worker_protocol_frame_invalid", code: "worker_protocol_frame_invalid",
severity: "warning", severity: "warning",
message: error instanceof Error ? error.message : String(error), message,
}, },
]; ];
} }
}, },
onStatus: (status) => { onStatus: (status, message) => {
if (token !== reloadToken) return; if (token !== reloadToken) return;
protocolState = status === "open" ? "connecting" : status; protocolState = status === "open" ? "connecting" : status;
if (status === "connecting" || status === "open") {
showConsoleReconnecting(message);
}
if (status === "closed") { if (status === "closed") {
showConsoleFailure(message);
rejectPendingCompletion(new Error("Worker protocol WebSocket closed.")); rejectPendingCompletion(new Error("Worker protocol WebSocket closed."));
} }
}, },
@@ -1697,6 +1840,7 @@
liveWorkerState = targetWorker?.state ?? null; liveWorkerState = targetWorker?.state ?? null;
streamDiagnostics = []; streamDiagnostics = [];
protocolState = "connecting"; protocolState = "connecting";
consoleDisplayState = { kind: "loading", stage: "session" };
const token = advanceReloadToken(); const token = advanceReloadToken();
if (!targetWorker) void loadWorker(target, token); if (!targetWorker) void loadWorker(target, token);
}); });
@@ -1705,7 +1849,19 @@
return () => discardAllAttachments(); return () => discardAllAttachments();
}); });
$effect(() => connectProtocolTransport(worker, reloadToken, consoleTarget)); $effect(() => {
const target = consoleTarget;
const targetWorker = worker;
const token = reloadToken;
if (
targetWorker &&
(targetWorker.runtime_id !== target.runtimeId ||
targetWorker.worker_id !== target.workerId)
) {
return;
}
return connectProtocolTransport(targetWorker, token, target);
});
</script> </script>
<svelte:window onkeydown={handleWorkerControlShortcut} /> <svelte:window onkeydown={handleWorkerControlShortcut} />
@@ -1836,9 +1992,14 @@
<p class="error">{workerError}</p> <p class="error">{workerError}</p>
{/if} {/if}
{#if lines.length === 0} <ConsoleDisplayStateView
<p>No console output is available for this Worker yet.</p> state={consoleDisplayState}
{:else} hasContent={lines.length > 0}
hasUnfilteredContent={selectedConsoleProjection.lines.length > 0}
onRetry={retryConsoleLoad}
/>
{#if (consoleDisplayState.kind === "ready" || consoleDisplayState.kind === "stale") && lines.length > 0}
<ol class="console-log"> <ol class="console-log">
{#each lines as item (item.id)} {#each lines as item (item.id)}
<ConsoleLineItem {item} /> <ConsoleLineItem {item} />
@@ -2182,6 +2343,9 @@
} }
.console-card { .console-card {
display: grid;
align-content: start;
gap: var(--space-4);
min-height: 100%; min-height: 100%;
} }
@@ -0,0 +1,349 @@
// @vitest-environment happy-dom
import {
cleanup,
fireEvent,
render,
screen,
waitFor,
} from "@testing-library/svelte";
import { afterEach, beforeEach, expect, test, vi } from "vitest";
import type {
Event as ProtocolEvent,
SessionSnapshot,
} from "$lib/generated/protocol";
import type { Worker } from "$lib/workspace/sidebar/types";
import ConsolePage from "./+page.svelte";
const multiplexer = vi.hoisted(() => {
type Listener = {
onFrame(frame: unknown): void;
onStatus?(status: "connecting" | "open" | "closed", message?: string): void;
};
const listeners: Listener[] = [];
return {
listeners,
close: vi.fn(),
sendWorkerMethod: vi.fn(),
subscribe: vi.fn((_selector: unknown, listener: Listener) => {
listeners.push(listener);
listener.onStatus?.("connecting");
return {
close: multiplexer.close,
sendWorkerMethod: multiplexer.sendWorkerMethod,
};
}),
};
});
vi.mock("$lib/workspace/multiplexer", () => ({
workspaceMultiplexer: () => ({ subscribe: multiplexer.subscribe }),
}));
function deferred<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((done) => {
resolve = done;
});
return { promise, resolve };
}
function worker(workerId = "worker-a", runtimeId = "runtime-a"): Worker {
return {
worker_id: workerId,
runtime_id: runtimeId,
resource_key: `worker:${workerId}`,
display_name: workerId,
label: workerId,
state: "idle",
host_id: "host-a",
profile: "default",
tags: [],
diagnostics: [],
capabilities: {
can_stop: true,
can_spawn_followup: true,
},
workspace: { visibility: "workspace", identity: "workspace-a" },
implementation: { kind: "builtin", display_hint: "test" },
workdir_attachments: [],
} as unknown as Worker;
}
function pageData(workerId = "worker-a", runtimeId = "runtime-a") {
return {
workspaceId: "workspace-a",
runtimeId,
workerId,
worker: worker(workerId, runtimeId),
workerError: null,
};
}
function emptySession(): SessionSnapshot {
return {
pending_submissions: {
revision: 0,
notification_count: 0,
head_id: null,
submissions: [],
},
entries: [],
};
}
function sessionWithUserMessage(content: string): SessionSnapshot {
return {
...emptySession(),
entries: [{
kind: "user_input",
segments: [{ kind: "text", content }],
entry_id: `entry-${content}`,
timestamp: 1,
provenance: "human_input",
derived_from: [],
}],
};
}
function snapshotEvent(session = emptySession()): ProtocolEvent {
return {
event: "snapshot",
data: {
session,
greeting: {
worker_name: "worker-a",
cwd: "",
provider: "test",
model: "test",
scope_summary: "test",
tools: [],
context_window: 0,
context_tokens: 0,
},
state: { last_command_id: 0, state: { kind: "idle" } },
in_flight: { blocks: [], commands: [], compaction: null },
internal_workers: [],
},
};
}
function subscribedFrame(session = emptySession()) {
return {
frame: "response",
message: {
result: "subscribed",
payload: {
snapshot: {
topic: "worker_protocol",
data: { events: [snapshotEvent(session)] },
},
},
},
};
}
function latestListener() {
const listener = multiplexer.listeners.at(-1);
if (!listener) throw new Error("missing Worker protocol listener");
return listener;
}
async function settleMicrotasks() {
await Promise.resolve();
await Promise.resolve();
}
beforeEach(() => {
multiplexer.listeners.length = 0;
multiplexer.close.mockClear();
multiplexer.sendWorkerMethod.mockClear();
multiplexer.subscribe.mockClear();
});
afterEach(() => {
cleanup();
vi.unstubAllGlobals();
vi.restoreAllMocks();
});
test("keeps delayed Session and live initial snapshot work in loading state until projection applies", async () => {
const sessionResponse = deferred<Response>();
vi.stubGlobal("fetch", vi.fn(() => sessionResponse.promise));
render(ConsolePage, { data: pageData() });
expect(screen.getByRole("status").textContent).toContain(
"Loading conversation",
);
expect(screen.queryByText("No conversation to display")).toBeNull();
sessionResponse.resolve(Response.json({ availability: "live_protocol" }));
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce());
expect(screen.getByRole("status").textContent).toContain(
"Waiting for the initial conversation snapshot",
);
expect(screen.queryByText("No conversation to display")).toBeNull();
latestListener().onFrame(subscribedFrame(emptySession()));
expect(await screen.findByText("No conversation to display")).not.toBeNull();
expect(screen.getByText("This Worker view has no conversation history yet."))
.not
.toBeNull();
});
test("applies retained empty and nonempty snapshots before showing their read-only result", async () => {
const response = deferred<Response>();
const nextResponse = deferred<Response>();
vi.stubGlobal(
"fetch",
vi.fn((input: RequestInfo | URL) => {
const pending = String(input).includes("worker-b")
? nextResponse.promise
: response.promise;
return pending.then((value) => value.clone());
}),
);
const view = render(ConsolePage, { data: pageData() });
expect(screen.getByRole("status").textContent).toContain(
"Loading conversation",
);
expect(screen.queryByText("No conversation to display")).toBeNull();
response.resolve(Response.json({
availability: "retained_snapshot",
snapshot: emptySession(),
}));
expect(await screen.findByText("Read-only retained conversation")).not
.toBeNull();
expect(screen.getByText("No conversation to display")).not.toBeNull();
await view.rerender({ data: pageData("worker-b", "runtime-b") });
expect(screen.getByRole("status").textContent).toContain(
"Loading conversation",
);
expect(screen.queryByText("No conversation to display")).toBeNull();
nextResponse.resolve(Response.json({
availability: "retained_snapshot",
snapshot: sessionWithUserMessage("retained message"),
}));
expect(await screen.findByText("retained message")).not.toBeNull();
expect(screen.getByText("Read-only retained conversation")).not.toBeNull();
expect(screen.queryByText("No conversation to display")).toBeNull();
});
test("shows bounded initial failures and retries explicitly", async () => {
const fetchMock = vi.fn()
.mockResolvedValueOnce(new Response(null, { status: 503 }))
.mockResolvedValueOnce(Response.json({
availability: "retained_snapshot",
snapshot: emptySession(),
}));
vi.stubGlobal("fetch", fetchMock);
render(ConsolePage, { data: pageData() });
const alert = await screen.findByRole("alert");
expect(alert.textContent).toContain("Unable to load conversation");
expect(alert.textContent).toContain("failed to observe Worker Session (503)");
await fireEvent.click(screen.getByRole("button", { name: "Retry" }));
expect(await screen.findByText("No conversation to display")).not.toBeNull();
expect(fetchMock).toHaveBeenCalledTimes(2);
});
test("distinguishes typed unavailability and an initial live connection closure", async () => {
const fetchMock = vi.fn()
.mockResolvedValueOnce(Response.json({
availability: "unavailable",
message: "retained state expired",
}))
.mockResolvedValueOnce(Response.json({ availability: "live_protocol" }));
vi.stubGlobal("fetch", fetchMock);
render(ConsolePage, { data: pageData() });
expect((await screen.findByRole("alert")).textContent).toContain(
"Conversation unavailable retained state expired",
);
await fireEvent.click(screen.getByRole("button", { name: "Retry" }));
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce());
latestListener().onStatus?.("closed", "subscription ended before snapshot");
expect((await screen.findByRole("alert")).textContent).toContain(
"subscription ended before snapshot",
);
expect(screen.queryByText("No conversation to display")).toBeNull();
});
test("preserves rendered content while reconnecting", async () => {
vi.stubGlobal(
"fetch",
vi.fn(() =>
Promise.resolve(Response.json({ availability: "live_protocol" }))
),
);
render(ConsolePage, { data: pageData() });
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce());
latestListener().onFrame(
subscribedFrame(sessionWithUserMessage("kept message")),
);
expect(await screen.findByText("kept message")).not.toBeNull();
latestListener().onStatus?.("closed", "Workspace subscription disconnected");
expect(await screen.findByText("Conversation updates are unavailable")).not
.toBeNull();
expect(screen.getByText("kept message")).not.toBeNull();
expect(screen.queryByText("No conversation to display")).toBeNull();
latestListener().onStatus?.("connecting", "resubscribing");
expect(await screen.findByText("Refreshing conversation")).not.toBeNull();
expect(screen.getByText("kept message")).not.toBeNull();
});
test("route changes clear prior content and fence stale Session responses", async () => {
const staleReload = deferred<Response>();
const current = deferred<Response>();
let firstWorkerLoads = 0;
const fetchMock = vi.fn((input: RequestInfo | URL) => {
if (String(input).includes("worker-b")) {
return current.promise.then((value) => value.clone());
}
firstWorkerLoads += 1;
if (firstWorkerLoads === 1) {
return Promise.resolve(Response.json({ availability: "live_protocol" }));
}
return staleReload.promise.then((value) => value.clone());
});
vi.stubGlobal("fetch", fetchMock);
const view = render(ConsolePage, { data: pageData() });
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce());
latestListener().onFrame(
subscribedFrame(sessionWithUserMessage("previous worker content")),
);
expect(await screen.findByText("previous worker content")).not.toBeNull();
latestListener().onStatus?.("closed", "reload old worker");
await fireEvent.click(
await screen.findByRole("button", { name: "Retry" }),
);
await waitFor(() => expect(firstWorkerLoads).toBe(2));
await view.rerender({ data: pageData("worker-b", "runtime-b") });
expect(screen.queryByText("previous worker content")).toBeNull();
expect(screen.getByRole("status").textContent).toContain(
"Loading conversation",
);
current.resolve(Response.json({
availability: "retained_snapshot",
snapshot: sessionWithUserMessage("current worker content"),
}));
expect(await screen.findByText("current worker content")).not.toBeNull();
staleReload.resolve(Response.json({
availability: "unavailable",
message: "stale worker error",
}));
await settleMicrotasks();
expect(screen.queryByText("stale worker error")).toBeNull();
expect(screen.queryByText("previous worker content")).toBeNull();
expect(screen.getByText("current worker content")).not.toBeNull();
});