mirror of
https://github.com/murashit/codex-panel.git
synced 2026-07-22 17:30:31 +00:00
Move connection orchestration into connection actions
This commit is contained in:
murashit
parent
03a026512e
commit
dccd63b75b
10 changed files with 138 additions and 156 deletions
|
|
@ -1,7 +1,6 @@
|
|||
import type { ServerInitialization } from "../../../../domain/server/initialization";
|
||||
import type { ChatConnectionPhase } from "../state/root-reducer";
|
||||
import type { ChatStateStore } from "../state/store";
|
||||
import type { ActiveConnectionWork, ConnectionWorkTracker } from "./connection-work";
|
||||
|
||||
const STATUS_CONNECTION_STOPPED = "Codex app-server stopped.";
|
||||
const STATUS_CONNECTION_STARTING = "Starting Codex app-server...";
|
||||
|
|
@ -24,7 +23,6 @@ export interface ChatConnectionDiagnosticsActions {
|
|||
export interface ChatConnectionActionsHost {
|
||||
stateStore: ChatStateStore;
|
||||
connection: ChatConnectionAdapter;
|
||||
connectionWork: ConnectionWorkTracker;
|
||||
metadata: ChatConnectionMetadataActions;
|
||||
diagnostics: ChatConnectionDiagnosticsActions;
|
||||
invalidateThreadWork: () => void;
|
||||
|
|
@ -44,11 +42,10 @@ export interface ChatConnectionActionsHost {
|
|||
|
||||
type ChatConnectionExitHost = Pick<
|
||||
ChatConnectionActionsHost,
|
||||
"connectionWork" | "invalidateThreadWork" | "setStatus" | "stateStore" | "resetThreadTurnPresence" | "refreshLiveState"
|
||||
"invalidateThreadWork" | "setStatus" | "stateStore" | "resetThreadTurnPresence" | "refreshLiveState"
|
||||
>;
|
||||
|
||||
export function handleChatConnectionExit(host: ChatConnectionExitHost): void {
|
||||
host.connectionWork.invalidate();
|
||||
function handleChatConnectionExit(host: ChatConnectionExitHost): void {
|
||||
host.invalidateThreadWork();
|
||||
host.setStatus(STATUS_CONNECTION_STOPPED, { kind: "disconnected", message: STATUS_CONNECTION_STOPPED });
|
||||
host.stateStore.dispatch({ type: "connection/scoped-cleared" });
|
||||
|
|
@ -66,12 +63,33 @@ export interface ChatConnectionActions {
|
|||
}
|
||||
|
||||
export function createChatConnectionActions(host: ChatConnectionActionsHost): ChatConnectionActions {
|
||||
let generation = 0;
|
||||
let activeConnection: { generation: number; promise: Promise<void> } | null = null;
|
||||
const invalidate = (): void => {
|
||||
generation += 1;
|
||||
activeConnection = null;
|
||||
};
|
||||
const isStale = (candidateGeneration: number): boolean => candidateGeneration !== generation;
|
||||
const actions: ChatConnectionActions = {
|
||||
ensureConnected: () => ensureConnected(host),
|
||||
invalidate: () => {
|
||||
host.connectionWork.invalidate();
|
||||
ensureConnected: async () => {
|
||||
if (activeConnection) return activeConnection.promise;
|
||||
if (host.connection.isConnected()) return;
|
||||
|
||||
const connectionGeneration = generation;
|
||||
const promise = initializeConnection(host, () => isStale(connectionGeneration));
|
||||
const active = { generation: connectionGeneration, promise };
|
||||
activeConnection = active;
|
||||
try {
|
||||
await promise;
|
||||
} finally {
|
||||
if (activeConnection === active) {
|
||||
activeConnection = null;
|
||||
}
|
||||
}
|
||||
},
|
||||
invalidate,
|
||||
handleExit: () => {
|
||||
invalidate();
|
||||
handleChatConnectionExit(host);
|
||||
},
|
||||
refreshActiveThreads: () => refreshActiveThreads(host),
|
||||
|
|
@ -81,24 +99,6 @@ export function createChatConnectionActions(host: ChatConnectionActionsHost): Ch
|
|||
return actions;
|
||||
}
|
||||
|
||||
async function ensureConnected(host: ChatConnectionActionsHost): Promise<void> {
|
||||
const connecting = host.connectionWork.active();
|
||||
if (connecting?.promise) return connecting.promise;
|
||||
|
||||
if (host.connection.isConnected()) {
|
||||
return;
|
||||
}
|
||||
|
||||
const connection = host.connectionWork.begin();
|
||||
const promise = initializeConnection(host, connection);
|
||||
connection.promise = promise;
|
||||
try {
|
||||
await promise;
|
||||
} finally {
|
||||
host.connectionWork.finish(connection, promise);
|
||||
}
|
||||
}
|
||||
|
||||
async function refreshActiveThreads(host: ChatConnectionActionsHost): Promise<void> {
|
||||
if (!host.connection.isConnected()) return;
|
||||
try {
|
||||
|
|
@ -131,17 +131,17 @@ async function refreshStatusPanel(
|
|||
await actions.refreshActiveThreads();
|
||||
}
|
||||
|
||||
async function initializeConnection(host: ChatConnectionActionsHost, connection: ActiveConnectionWork): Promise<void> {
|
||||
async function initializeConnection(host: ChatConnectionActionsHost, isStale: () => boolean): Promise<void> {
|
||||
host.setStatus(STATUS_CONNECTION_STARTING, { kind: "connecting" });
|
||||
try {
|
||||
const initialization = await host.connection.connect();
|
||||
if (host.connectionWork.isStale(connection)) return;
|
||||
if (isStale()) return;
|
||||
host.stateStore.dispatch({ type: "connection/initialized", initializeResponse: initialization });
|
||||
if (!host.connection.isConnected()) throw new Error("Codex app-server connection did not initialize.");
|
||||
host.refreshTabHeader();
|
||||
host.setStatus(STATUS_CONNECTED, { kind: "connected" });
|
||||
} catch (error) {
|
||||
if (host.connectionWork.isStale(connection)) return;
|
||||
if (isStale()) return;
|
||||
if (host.isStaleConnectionError(error)) return;
|
||||
if (host.isStaleSharedQueryError(error)) return;
|
||||
const message = connectionErrorMessage(error, host.configuredCommand());
|
||||
|
|
@ -151,25 +151,25 @@ async function initializeConnection(host: ChatConnectionActionsHost, connection:
|
|||
return;
|
||||
}
|
||||
|
||||
await hydrateConnectedResources(host, connection);
|
||||
await hydrateConnectedResources(host, isStale);
|
||||
}
|
||||
|
||||
async function hydrateConnectedResources(host: ChatConnectionActionsHost, connection: ActiveConnectionWork): Promise<void> {
|
||||
async function hydrateConnectedResources(host: ChatConnectionActionsHost, isStale: () => boolean): Promise<void> {
|
||||
try {
|
||||
await host.metadata.refreshAppServerMetadata();
|
||||
} catch (error) {
|
||||
if (host.connectionWork.isStale(connection) || host.isStaleSharedQueryError(error)) return;
|
||||
if (isStale() || host.isStaleSharedQueryError(error)) return;
|
||||
host.addSystemMessage(`Could not refresh Codex metadata: ${errorMessage(error)}`);
|
||||
}
|
||||
if (host.connectionWork.isStale(connection)) return;
|
||||
if (isStale()) return;
|
||||
|
||||
try {
|
||||
await host.refreshSharedThreads();
|
||||
} catch (error) {
|
||||
if (host.connectionWork.isStale(connection) || host.isStaleSharedQueryError(error)) return;
|
||||
if (isStale() || host.isStaleSharedQueryError(error)) return;
|
||||
host.addSystemMessage(`Could not refresh Codex threads: ${errorMessage(error)}`);
|
||||
}
|
||||
if (host.connectionWork.isStale(connection)) return;
|
||||
if (isStale()) return;
|
||||
host.scheduleDeferredDiagnostics();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,46 +0,0 @@
|
|||
export type ConnectionWorkLifecycleState = { kind: "idle" } | { kind: "connecting"; promise: Promise<void> | null };
|
||||
export type ActiveConnectionWork = Extract<ConnectionWorkLifecycleState, { kind: "connecting" }>;
|
||||
type ConnectionWorkLifecycleEvent =
|
||||
| { type: "started"; connection: ActiveConnectionWork }
|
||||
| { type: "finished"; connection: ActiveConnectionWork; promise: Promise<void> }
|
||||
| { type: "invalidated" };
|
||||
|
||||
function transitionConnectionWorkLifecycle(
|
||||
state: ConnectionWorkLifecycleState,
|
||||
event: ConnectionWorkLifecycleEvent,
|
||||
): ConnectionWorkLifecycleState {
|
||||
switch (event.type) {
|
||||
case "started":
|
||||
return event.connection;
|
||||
case "finished":
|
||||
return state === event.connection && state.promise === event.promise ? { kind: "idle" } : state;
|
||||
case "invalidated":
|
||||
return state.kind === "idle" ? state : { kind: "idle" };
|
||||
}
|
||||
}
|
||||
|
||||
export class ConnectionWorkTracker {
|
||||
private state: ConnectionWorkLifecycleState = { kind: "idle" };
|
||||
|
||||
active(): ActiveConnectionWork | null {
|
||||
return this.state.kind === "connecting" ? this.state : null;
|
||||
}
|
||||
|
||||
begin(): ActiveConnectionWork {
|
||||
const connection: ActiveConnectionWork = { kind: "connecting", promise: null };
|
||||
this.state = transitionConnectionWorkLifecycle(this.state, { type: "started", connection });
|
||||
return connection;
|
||||
}
|
||||
|
||||
finish(connection: ActiveConnectionWork, promise: Promise<void>): void {
|
||||
this.state = transitionConnectionWorkLifecycle(this.state, { type: "finished", connection, promise });
|
||||
}
|
||||
|
||||
invalidate(): void {
|
||||
this.state = transitionConnectionWorkLifecycle(this.state, { type: "invalidated" });
|
||||
}
|
||||
|
||||
isStale(connection: ActiveConnectionWork): boolean {
|
||||
return this.state !== connection;
|
||||
}
|
||||
}
|
||||
|
|
@ -6,12 +6,7 @@ import { isStaleAppServerSharedQueryContextError } from "../../../../app-server/
|
|||
import type { SharedServerMetadata } from "../../../../domain/server/metadata";
|
||||
import { type ChatInboundHandler, createChatInboundHandler } from "../../app-server/inbound/handler";
|
||||
import type { ChatAppServerGateway } from "../../app-server/session-gateway";
|
||||
import {
|
||||
type ChatConnectionActions,
|
||||
createChatConnectionActions,
|
||||
handleChatConnectionExit,
|
||||
} from "../../application/connection/connection-actions";
|
||||
import type { ConnectionWorkTracker } from "../../application/connection/connection-work";
|
||||
import { type ChatConnectionActions, createChatConnectionActions } from "../../application/connection/connection-actions";
|
||||
import { createServerDiagnosticsActions } from "../../application/connection/server-diagnostics-actions";
|
||||
import { createServerMetadataActions } from "../../application/connection/server-metadata-actions";
|
||||
import type { LocalIdSource } from "../../application/local-id-source";
|
||||
|
|
@ -40,7 +35,6 @@ interface ChatPanelConnectionBundleInput {
|
|||
interface ChatPanelConnectionBundleHost {
|
||||
environment: ChatPanelEnvironment;
|
||||
stateStore: ChatStateStore;
|
||||
connectionWork: ConnectionWorkTracker;
|
||||
deferredTasks: ChatViewDeferredTasks;
|
||||
invalidateThreadWork: () => void;
|
||||
deferLiveStateRefresh: () => void;
|
||||
|
|
@ -180,7 +174,6 @@ export function createConnectionBundle(
|
|||
);
|
||||
const connectionExitHost = {
|
||||
stateStore,
|
||||
connectionWork: host.connectionWork,
|
||||
invalidateThreadWork: () => {
|
||||
host.invalidateThreadWork();
|
||||
},
|
||||
|
|
@ -212,7 +205,7 @@ export function createConnectionBundle(
|
|||
onExit: () => {
|
||||
serverRequestResponders.clear();
|
||||
serverDiagnostics.invalidate();
|
||||
handleChatConnectionExit(connectionExitHost);
|
||||
connectionActions.handleExit();
|
||||
},
|
||||
}),
|
||||
isConnected: () => connection.isConnected(),
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ import { codexPanelAppServerInitializeParams } from "../../../app-server/connect
|
|||
import { ConnectionManager } from "../../../app-server/connection/connection-manager";
|
||||
import { isStaleAppServerSharedQueryContextError } from "../../../app-server/query/shared-queries";
|
||||
import { createChatAppServerGateway } from "../app-server/session-gateway";
|
||||
import type { ConnectionWorkTracker } from "../application/connection/connection-work";
|
||||
import { reconnectPanel } from "../application/connection/reconnect-actions";
|
||||
import { createLocalIdSource, type LocalIdSource } from "../application/local-id-source";
|
||||
import { runtimeSnapshotForChatState } from "../application/runtime/snapshot";
|
||||
|
|
@ -24,6 +23,7 @@ import { type ChatPanelShellBundle, createShellBundle } from "./bundles/shell-bu
|
|||
import { createThreadActionBundle, createThreadFoundation, createThreadLifecycleBundle } from "./bundles/thread-bundle";
|
||||
import { createTurnBundle } from "./bundles/turn-bundle";
|
||||
import type { ChatPanelEnvironment } from "./contracts";
|
||||
import { createConnectedClientResolver } from "./session/connected-client-resolver";
|
||||
import type { ChatViewDeferredTasks } from "./session/deferred-work";
|
||||
import { type ChatPanelSharedStateBinding, createChatPanelSharedStateBinding } from "./session/shared-state-binding";
|
||||
|
||||
|
|
@ -66,7 +66,6 @@ interface ChatPanelSessionGraphHost {
|
|||
stateStore: ChatStateStore;
|
||||
deferredTasks: ChatViewDeferredTasks;
|
||||
resumeWork: ChatResumeWorkTracker;
|
||||
connectionWork: ConnectionWorkTracker;
|
||||
threadStreamScrollBinding: ChatThreadStreamScrollBinding;
|
||||
getClosing: () => boolean;
|
||||
viewWindow: () => Window;
|
||||
|
|
@ -77,18 +76,12 @@ export function createChatPanelSessionGraph(host: ChatPanelSessionGraphHost): Ch
|
|||
const localItemIds = createLocalIdSource();
|
||||
const connection = createConnectionManager(environment);
|
||||
const currentClient = () => connection.currentClient();
|
||||
let ensureConnected: () => Promise<void> = async () => {
|
||||
throw new Error("Codex app-server connection actions are not initialized.");
|
||||
};
|
||||
const connectedClient = async () => {
|
||||
await ensureConnected();
|
||||
return currentClient();
|
||||
};
|
||||
const connectedClient = createConnectedClientResolver(currentClient);
|
||||
const appServer = createChatAppServerGateway({
|
||||
codexPath: () => environment.plugin.settingsRef.settings.codexPath(),
|
||||
vaultPath: environment.plugin.settingsRef.vaultPath,
|
||||
currentClient,
|
||||
connectedClient,
|
||||
connectedClient: () => connectedClient.resolve(),
|
||||
});
|
||||
const status = createSessionStatus(stateStore, localItemIds);
|
||||
const refreshTabHeader = () => {
|
||||
|
|
@ -116,7 +109,6 @@ export function createChatPanelSessionGraph(host: ChatPanelSessionGraphHost): Ch
|
|||
{
|
||||
environment,
|
||||
stateStore,
|
||||
connectionWork: host.connectionWork,
|
||||
deferredTasks: host.deferredTasks,
|
||||
invalidateThreadWork: () => {
|
||||
threadFoundation.invalidateThreadWork();
|
||||
|
|
@ -137,7 +129,8 @@ export function createChatPanelSessionGraph(host: ChatPanelSessionGraphHost): Ch
|
|||
connection: { actions: connectionActions },
|
||||
inboundHandler,
|
||||
} = connectionBundle;
|
||||
ensureConnected = () => connectionActions.ensureConnected();
|
||||
const ensureConnected = () => connectionActions.ensureConnected();
|
||||
connectedClient.bindEnsureConnected(ensureConnected);
|
||||
const refreshActiveThreads = () => connectionActions.refreshActiveThreads();
|
||||
const runtime = createRuntimeBundle(host, {
|
||||
connection,
|
||||
|
|
@ -182,7 +175,7 @@ export function createChatPanelSessionGraph(host: ChatPanelSessionGraphHost): Ch
|
|||
reconnectPanel({
|
||||
stateStore,
|
||||
invalidateConnectionWork: () => {
|
||||
host.connectionWork.invalidate();
|
||||
connectionActions.invalidate();
|
||||
},
|
||||
invalidateThreadWork: () => {
|
||||
threadFoundation.invalidateThreadWork();
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ import type { AppServerClient } from "../../../app-server/connection/client";
|
|||
import { type AppServerQueryContext, appServerQueryContextMatches, appServerQueryContextRawEquals } from "../../../app-server/query/keys";
|
||||
import { pendingRequestCountsFromQueues } from "../../../domain/pending-requests/aggregate";
|
||||
import { threadMeaningfulTitle, threadWindowTitle } from "../../../domain/threads/title";
|
||||
import { ConnectionWorkTracker } from "../application/connection/connection-work";
|
||||
import type { ChatState } from "../application/state/root-reducer";
|
||||
import { type ChatStateStore, createChatStateStore } from "../application/state/store";
|
||||
import { parseRestoredThreadState, type RestoredThreadPlaceholderState } from "../application/threads/restored-thread-lifecycle";
|
||||
|
|
@ -18,7 +17,6 @@ export class ChatPanelSession implements ChatPanelHandle {
|
|||
private readonly graph: ChatPanelSessionGraph;
|
||||
|
||||
private readonly deferredTasks: ChatViewDeferredTasks;
|
||||
private readonly connectionWork = new ConnectionWorkTracker();
|
||||
private readonly resumeWork = new ChatResumeWorkTracker();
|
||||
private readonly threadStreamScrollBinding: ChatThreadStreamScrollBinding = createChatThreadStreamScrollBinding();
|
||||
private observedAppServerContext: AppServerQueryContext;
|
||||
|
|
@ -156,7 +154,7 @@ export class ChatPanelSession implements ChatPanelHandle {
|
|||
close(): void {
|
||||
this.opened = false;
|
||||
this.closing = true;
|
||||
this.connectionWork.invalidate();
|
||||
this.graph.connection.actions.invalidate();
|
||||
this.graph.actions.invalidateThreadWork();
|
||||
this.deferredTasks.clearAll();
|
||||
this.graph.runtime.sharedState.unsubscribe();
|
||||
|
|
@ -248,7 +246,6 @@ export class ChatPanelSession implements ChatPanelHandle {
|
|||
stateStore: this.stateStore,
|
||||
deferredTasks: this.deferredTasks,
|
||||
resumeWork: this.resumeWork,
|
||||
connectionWork: this.connectionWork,
|
||||
threadStreamScrollBinding: this.threadStreamScrollBinding,
|
||||
getClosing: () => this.closing,
|
||||
viewWindow: () => this.viewWindow(),
|
||||
|
|
|
|||
22
src/features/chat/host/session/connected-client-resolver.ts
Normal file
22
src/features/chat/host/session/connected-client-resolver.ts
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
import type { AppServerClient } from "../../../../app-server/connection/client";
|
||||
|
||||
export interface ConnectedClientResolver {
|
||||
resolve(): Promise<AppServerClient | null>;
|
||||
bindEnsureConnected(ensureConnected: () => Promise<void>): void;
|
||||
}
|
||||
|
||||
export function createConnectedClientResolver(currentClient: () => AppServerClient | null): ConnectedClientResolver {
|
||||
let ensureConnected: (() => Promise<void>) | null = null;
|
||||
|
||||
return {
|
||||
resolve: async () => {
|
||||
if (!ensureConnected) throw new Error("Codex app-server connection actions are not initialized.");
|
||||
await ensureConnected();
|
||||
return currentClient();
|
||||
},
|
||||
bindEnsureConnected: (nextEnsureConnected) => {
|
||||
if (ensureConnected) throw new Error("Codex app-server connection actions are already initialized.");
|
||||
ensureConnected = nextEnsureConnected;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -8,9 +8,9 @@ import {
|
|||
type ChatConnectionMetadataActions,
|
||||
createChatConnectionActions,
|
||||
} from "../../../../../src/features/chat/application/connection/connection-actions";
|
||||
import { ConnectionWorkTracker } from "../../../../../src/features/chat/application/connection/connection-work";
|
||||
import { createChatState } from "../../../../../src/features/chat/application/state/root-reducer";
|
||||
import { createChatStateStore } from "../../../../../src/features/chat/application/state/store";
|
||||
import { deferred } from "../../../../support/async";
|
||||
|
||||
function createActionsHarness({ connected = false } = {}) {
|
||||
const stateStore = createChatStateStore(createChatState());
|
||||
|
|
@ -34,7 +34,6 @@ function createActionsHarness({ connected = false } = {}) {
|
|||
const host: ChatConnectionActionsHost = {
|
||||
stateStore,
|
||||
connection,
|
||||
connectionWork: new ConnectionWorkTracker(),
|
||||
metadata,
|
||||
diagnostics,
|
||||
invalidateThreadWork: vi.fn(),
|
||||
|
|
@ -57,11 +56,47 @@ function createActionsHarness({ connected = false } = {}) {
|
|||
host,
|
||||
refreshAppServerMetadata,
|
||||
refreshServerDiagnostics,
|
||||
setConnected: (value: boolean) => {
|
||||
isConnected = value;
|
||||
},
|
||||
stateStore,
|
||||
};
|
||||
}
|
||||
|
||||
describe("ChatConnectionActions", () => {
|
||||
it("coalesces concurrent connection attempts inside the actions boundary", async () => {
|
||||
const { actions, connect, setConnected } = createActionsHarness();
|
||||
const pending = deferred<Awaited<ReturnType<ChatConnectionAdapter["connect"]>>>();
|
||||
connect.mockImplementationOnce(() => pending.promise);
|
||||
|
||||
const first = actions.ensureConnected();
|
||||
const second = actions.ensureConnected();
|
||||
|
||||
expect(connect).toHaveBeenCalledOnce();
|
||||
setConnected(true);
|
||||
pending.resolve({ codexHome: "/codex", platformFamily: "unix", platformOs: "macos", userAgent: "test" });
|
||||
await Promise.all([first, second]);
|
||||
});
|
||||
|
||||
it("starts fresh work after invalidation and ignores the obsolete result", async () => {
|
||||
const { actions, connect, setConnected, stateStore } = createActionsHarness();
|
||||
const obsolete = deferred<Awaited<ReturnType<ChatConnectionAdapter["connect"]>>>();
|
||||
const current = deferred<Awaited<ReturnType<ChatConnectionAdapter["connect"]>>>();
|
||||
connect.mockImplementationOnce(() => obsolete.promise).mockImplementationOnce(() => current.promise);
|
||||
|
||||
const first = actions.ensureConnected();
|
||||
actions.invalidate();
|
||||
const second = actions.ensureConnected();
|
||||
setConnected(true);
|
||||
current.resolve({ codexHome: "/current", platformFamily: "unix", platformOs: "macos", userAgent: "test" });
|
||||
await second;
|
||||
obsolete.resolve({ codexHome: "/obsolete", platformFamily: "unix", platformOs: "macos", userAgent: "test" });
|
||||
await first;
|
||||
|
||||
expect(connect).toHaveBeenCalledTimes(2);
|
||||
expect(stateStore.getState().connection.initializeResponse?.codexHome).toBe("/current");
|
||||
});
|
||||
|
||||
it("connects once and publishes startup metadata", async () => {
|
||||
const { connect, actions, host, refreshAppServerMetadata, stateStore } = createActionsHarness();
|
||||
|
||||
|
|
|
|||
|
|
@ -1,39 +0,0 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { ConnectionWorkTracker } from "../../../../../src/features/chat/application/connection/connection-work";
|
||||
|
||||
describe("ConnectionWorkTracker", () => {
|
||||
it("tracks active connection work by identity", () => {
|
||||
const tracker = new ConnectionWorkTracker();
|
||||
const connection = tracker.begin();
|
||||
const stale = { kind: "connecting" as const, promise: Promise.resolve() };
|
||||
|
||||
expect(tracker.active()).toBe(connection);
|
||||
expect(tracker.isStale(connection)).toBe(false);
|
||||
expect(tracker.isStale(stale)).toBe(true);
|
||||
|
||||
const promise = Promise.resolve();
|
||||
connection.promise = promise;
|
||||
tracker.finish(connection, Promise.resolve());
|
||||
expect(tracker.active()).toBe(connection);
|
||||
tracker.finish(connection, promise);
|
||||
expect(tracker.active()).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps stale connection completions from clearing the active connection", () => {
|
||||
const firstPromise = Promise.resolve();
|
||||
const secondPromise = Promise.resolve();
|
||||
const tracker = new ConnectionWorkTracker();
|
||||
const first = tracker.begin();
|
||||
first.promise = firstPromise;
|
||||
const second = tracker.begin();
|
||||
second.promise = secondPromise;
|
||||
|
||||
tracker.finish(first, firstPromise);
|
||||
expect(tracker.active()).toBe(second);
|
||||
tracker.finish(second, firstPromise);
|
||||
expect(tracker.active()).toBe(second);
|
||||
tracker.finish(second, secondPromise);
|
||||
expect(tracker.active()).toBeNull();
|
||||
});
|
||||
});
|
||||
31
tests/features/chat/host/connected-client-resolver.test.ts
Normal file
31
tests/features/chat/host/connected-client-resolver.test.ts
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import type { AppServerClient } from "../../../../src/app-server/connection/client";
|
||||
import { createConnectedClientResolver } from "../../../../src/features/chat/host/session/connected-client-resolver";
|
||||
|
||||
describe("createConnectedClientResolver", () => {
|
||||
it("connects before resolving the current client", async () => {
|
||||
const client = {} as AppServerClient;
|
||||
const currentClient = vi.fn(() => client);
|
||||
const ensureConnected = vi.fn().mockResolvedValue(undefined);
|
||||
const resolver = createConnectedClientResolver(currentClient);
|
||||
resolver.bindEnsureConnected(ensureConnected);
|
||||
|
||||
await expect(resolver.resolve()).resolves.toBe(client);
|
||||
expect(ensureConnected).toHaveBeenCalledOnce();
|
||||
expect(currentClient).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("rejects use before the connection actions are bound", async () => {
|
||||
const resolver = createConnectedClientResolver(() => null);
|
||||
|
||||
await expect(resolver.resolve()).rejects.toThrow("connection actions are not initialized");
|
||||
});
|
||||
|
||||
it("rejects rebinding the connection actions", () => {
|
||||
const resolver = createConnectedClientResolver(() => null);
|
||||
resolver.bindEnsureConnected(vi.fn().mockResolvedValue(undefined));
|
||||
|
||||
expect(() => resolver.bindEnsureConnected(vi.fn().mockResolvedValue(undefined))).toThrow("connection actions are already initialized");
|
||||
});
|
||||
});
|
||||
|
|
@ -5,7 +5,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
|||
import { StaleAppServerSharedQueryContextError } from "../../../../src/app-server/query/shared-queries";
|
||||
import type { ModelMetadata } from "../../../../src/domain/catalog/metadata";
|
||||
import type { Thread } from "../../../../src/domain/threads/model";
|
||||
import { ConnectionWorkTracker } from "../../../../src/features/chat/application/connection/connection-work";
|
||||
import { type ChatStateStore, createChatStateStore } from "../../../../src/features/chat/application/state/store";
|
||||
import { HistoryController } from "../../../../src/features/chat/application/threads/history-controller";
|
||||
import { ChatResumeWorkTracker } from "../../../../src/features/chat/application/threads/resume-work";
|
||||
|
|
@ -201,7 +200,7 @@ describe("createChatPanelSessionGraph actions", () => {
|
|||
});
|
||||
|
||||
it("wires reconnect cleanup through the graph toolbar action", async () => {
|
||||
const { graph, stateStore, connectionWork, deferredTasks } = sessionGraphFixture();
|
||||
const { graph, stateStore, deferredTasks } = sessionGraphFixture();
|
||||
stateStore.dispatch({
|
||||
type: "active-thread/resumed",
|
||||
approvalPolicyKnown: true,
|
||||
|
|
@ -217,7 +216,7 @@ describe("createChatPanelSessionGraph actions", () => {
|
|||
serviceTier: null,
|
||||
approvalsReviewer: null,
|
||||
});
|
||||
const activeConnectionWork = connectionWork.begin();
|
||||
const invalidateConnection = vi.spyOn(graph.connection.actions, "invalidate");
|
||||
const clearDiagnostics = vi.spyOn(deferredTasks, "clearDiagnostics");
|
||||
const resetConnection = vi.spyOn(graph.connection.manager, "resetConnection");
|
||||
const ensureConnected = vi.spyOn(graph.connection.actions, "ensureConnected").mockResolvedValue(undefined);
|
||||
|
|
@ -228,7 +227,7 @@ describe("createChatPanelSessionGraph actions", () => {
|
|||
await waitForAsyncWork(() => {
|
||||
expect(resumeThread).toHaveBeenCalledWith("thread-1");
|
||||
});
|
||||
expect(connectionWork.isStale(activeConnectionWork)).toBe(true);
|
||||
expect(invalidateConnection).toHaveBeenCalledOnce();
|
||||
expect(clearDiagnostics).toHaveBeenCalledOnce();
|
||||
expect(resetConnection).toHaveBeenCalledOnce();
|
||||
expect(stateStore.getState().connection.statusText).toBe("Reconnecting...");
|
||||
|
|
@ -250,12 +249,10 @@ describe("createChatPanelSessionGraph actions", () => {
|
|||
graph: ReturnType<typeof createChatPanelSessionGraph>;
|
||||
stateStore: ChatStateStore;
|
||||
resumeWork: ChatResumeWorkTracker;
|
||||
connectionWork: ConnectionWorkTracker;
|
||||
deferredTasks: ReturnType<typeof createChatViewDeferredTasks>;
|
||||
} {
|
||||
const stateStore = createChatStateStore();
|
||||
const resumeWork = new ChatResumeWorkTracker();
|
||||
const connectionWork = new ConnectionWorkTracker();
|
||||
const deferredTasks = createChatViewDeferredTasks(() => window);
|
||||
const environment = chatPanelEnvironmentFixture(options.environment);
|
||||
const graph = createChatPanelSessionGraph({
|
||||
|
|
@ -263,12 +260,11 @@ describe("createChatPanelSessionGraph actions", () => {
|
|||
stateStore,
|
||||
deferredTasks,
|
||||
resumeWork,
|
||||
connectionWork,
|
||||
threadStreamScrollBinding: createChatThreadStreamScrollBinding(),
|
||||
getClosing: () => false,
|
||||
viewWindow: () => window,
|
||||
});
|
||||
return { graph, stateStore, resumeWork, connectionWork, deferredTasks };
|
||||
return { graph, stateStore, resumeWork, deferredTasks };
|
||||
}
|
||||
|
||||
interface PartialChatPanelEnvironment {
|
||||
|
|
|
|||
Loading…
Reference in a new issue