obsidian-plugin/murashit_codex-panel
1
0
Fork 0
mirror of https://github.com/murashit/codex-panel.git synced 2026-07-22 06:57:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Complete connection handler ownership

Browse source
This commit is contained in:
murashit 2026-06-15 09:30:30 +09:00
parent 6fe1a837c8
commit f56fae77cf
8 changed files with 120 additions and 125 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
src/app-server/connection/connection-manager.ts
View file

@ -32,7 +32,6 @@ export class ConnectionManager {
constructor(
private readonly codexPath: () => string,
private readonly cwd: string,
private readonly handlers: ConnectionManagerHandlers,
private readonly clientFactory: AppServerClientFactory = (codexPath, cwd, handlers) => new AppServerClient(codexPath, cwd, handlers),
) {}
@ -44,7 +43,7 @@ export class ConnectionManager {
return Boolean(this.currentClient());
}
async connect(): Promise<ServerInitialization> {
async connect(handlers: ConnectionManagerHandlers): Promise<ServerInitialization> {
const currentClient = this.currentClient();
if (currentClient) {
return appServerInitializationFromResponse(currentClient.initializeResponse);
@ -55,20 +54,20 @@ export class ConnectionManager {
const client = this.clientFactory(this.codexPath(), this.cwd, {
onNotification: (notification) => {
if (this.isStale(generation)) return;
this.handlers.onNotification(notification);
handlers.onNotification(notification);
},
onServerRequest: (request) => {
if (this.isStale(generation)) return;
this.handlers.onServerRequest(request);
handlers.onServerRequest(request);
},
onLog: (message) => {
if (this.isStale(generation)) return;
this.handlers.onLog(message);
handlers.onLog(message);
},
onExit: () => {
if (this.isStale(generation)) return;
this.state = { kind: "disconnected", generation };
this.handlers.onExit();
handlers.onExit();
},
});
const promise = client

21
src/features/chat/application/connection/connection-controller.ts
View file

@ -46,6 +46,20 @@ export interface ChatConnectionControllerHost {
notifyConnectionFailed: () => void;
}
type ChatConnectionExitHost = Pick<
ChatConnectionControllerHost,
"connectionWork" | "invalidateResumeWork" | "setStatus" | "stateStore" | "resetThreadTurnPresence" | "refreshLiveState"
>;
export function handleChatConnectionExit(host: ChatConnectionExitHost): void {
host.connectionWork.invalidate();
host.invalidateResumeWork();
host.setStatus(STATUS_CONNECTION_STOPPED, { kind: "disconnected", message: STATUS_CONNECTION_STOPPED });
host.stateStore.dispatch({ type: "connection/scoped-cleared" });
host.resetThreadTurnPresence(false);
host.refreshLiveState();
}
export class ChatConnectionController {
constructor(private readonly host: ChatConnectionControllerHost) {}
@ -72,12 +86,7 @@ export class ChatConnectionController {
}
handleExit(): void {
this.invalidate();
this.host.invalidateResumeWork();
this.host.setStatus(STATUS_CONNECTION_STOPPED, { kind: "disconnected", message: STATUS_CONNECTION_STOPPED });
this.host.stateStore.dispatch({ type: "connection/scoped-cleared" });
this.host.resetThreadTurnPresence(false);
this.host.refreshLiveState();
handleChatConnectionExit(this.host);
}
async refreshThreads(): Promise<void> {

110
src/features/chat/host/connection-bundle.ts
View file

@ -1,10 +1,9 @@
import { Notice } from "obsidian";
import type { AppServerClient } from "../../../app-server/connection/client";
import type { ConnectionManager, ConnectionManagerHandlers } from "../../../app-server/connection/connection-manager";
import { ConnectionManager } from "../../../app-server/connection/connection-manager";
import type { ThreadSurfaceBroadcaster } from "../application/ports/chat-host";
import type { ChatConnectionWorkTracker, ChatViewDeferredTasks } from "../application/lifecycle";
import { ChatConnectionController } from "../application/connection/connection-controller";
import { ChatConnectionController, handleChatConnectionExit } from "../application/connection/connection-controller";
import { createChatServerDiagnosticsActions, type ChatServerDiagnosticsActions } from "../app-server/actions/diagnostics";
import { createChatServerMetadataActions, type ChatServerMetadataActions } from "../app-server/actions/metadata";
import { createChatServerThreadActions, type ChatServerThreadActions } from "../app-server/actions/threads";
@ -18,7 +17,10 @@ import type { AutoTitleController } from "../application/threads/auto-title-cont
import type { MessageStreamNoticeSection } from "../domain/message-stream/items";
export interface ChatConnectionBundle {
connectionController: ChatConnectionController;
connection: {
manager: ConnectionManager;
controller: ChatConnectionController;
};
inboundController: ChatInboundController;
serverActions: {
threads: ChatServerThreadActions;
@ -27,10 +29,6 @@ export interface ChatConnectionBundle {
};
}
interface ChatConnectionClientPorts {
currentClient: () => AppServerClient | null;
}
interface ChatConnectionRefreshPorts {
refreshThreads: () => Promise<void>;
refreshSkills: (forceReload?: boolean) => Promise<void>;
@ -45,11 +43,10 @@ interface ChatConnectionBundleStatus {
export interface ChatConnectionBundleContext {
stateStore: ChatStateStore;
vaultPath: string;
connection: ConnectionManager;
codexPath: () => string;
connectionWork: ChatConnectionWorkTracker;
deferredTasks: ChatViewDeferredTasks;
threadSurfaces: ThreadSurfaceBroadcaster;
client: ChatConnectionClientPorts;
refresh: ChatConnectionRefreshPorts;
goals: GoalActions;
autoTitle: AutoTitleController;
@ -59,21 +56,18 @@ export interface ChatConnectionBundleContext {
refreshDeferredDiagnostics: () => Promise<void>;
refreshTabHeader: () => void;
refreshLiveState: () => void;
deferLiveStateRefresh: () => void;
configuredCommand: () => string;
}
export interface ChatConnectionEventTargets {
inbound: ChatInboundController;
connectionController: ChatConnectionController;
}
export function createChatConnectionBundle(context: ChatConnectionBundleContext): ChatConnectionBundle {
const { stateStore, vaultPath, connection, connectionWork, deferredTasks, threadSurfaces, client, refresh, goals, autoTitle, status } =
context;
const { stateStore, vaultPath, connectionWork, deferredTasks, threadSurfaces, refresh, goals, autoTitle, status } = context;
const connection = new ConnectionManager(context.codexPath, vaultPath);
const currentClient = () => connection.currentClient();
const serverMetadata = createChatServerMetadataActions({
stateStore,
vaultPath,
currentClient: client.currentClient,
currentClient,
publishAppServerMetadata: (metadata) => {
threadSurfaces.publishAppServerMetadata(metadata);
},
@ -81,7 +75,7 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
const serverDiagnostics = createChatServerDiagnosticsActions({
stateStore,
vaultPath,
currentClient: client.currentClient,
currentClient,
publishAppServerMetadata: (metadata) => {
threadSurfaces.publishAppServerMetadata(metadata);
},
@ -90,7 +84,7 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
const serverThreads = createChatServerThreadActions({
stateStore,
vaultPath,
currentClient: client.currentClient,
currentClient,
runtimeSnapshotForState: runtimeSnapshotForChatState,
publishThreadList: (threads) => {
threadSurfaces.applyThreadListSnapshot(threads);
@ -100,7 +94,7 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
},
});
const serverRequestHost = {
currentClient: client.currentClient,
currentClient,
};
const inboundController = new ChatInboundController(stateStore, {
refreshThreads: () => {
@ -128,10 +122,39 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
respondToServerRequest: (requestId, result) => respondToServerRequest(serverRequestHost, requestId, result),
rejectServerRequest: (requestId, code, message) => rejectServerRequest(serverRequestHost, requestId, code, message),
});
const connectionController = new ChatConnectionController({
const connectionExitHost = {
stateStore,
connection,
connectionWork,
invalidateResumeWork: context.invalidateResumeWork,
setStatus: status.set,
resetThreadTurnPresence: (hadTurns: boolean) => {
autoTitle.resetThreadTurnPresence(hadTurns);
},
refreshLiveState: context.refreshLiveState,
};
const connectionController = new ChatConnectionController({
...connectionExitHost,
connection: {
connect: () =>
connection.connect({
onNotification: (notification) => {
inboundController.handleNotification(notification);
context.deferLiveStateRefresh();
},
onServerRequest: (request) => {
inboundController.handleServerRequest(request);
context.deferLiveStateRefresh();
},
onLog: (message) => {
inboundController.handleAppServerLog(message);
},
onExit: () => {
handleChatConnectionExit(connectionExitHost);
},
}),
currentClient,
isConnected: () => connection.isConnected(),
},
metadata: {
refreshPublishedAppServerMetadata: () => serverMetadata.refreshPublishedAppServerMetadata(),
refreshPublishedSkills: (forceReload) => serverMetadata.refreshPublishedSkills(forceReload),
@ -139,7 +162,6 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
diagnostics: {
refreshPublishedDiagnosticProbes: () => serverDiagnostics.refreshPublishedDiagnosticProbes(),
},
invalidateResumeWork: context.invalidateResumeWork,
loadSharedThreadList: context.loadSharedThreadList,
scheduleDeferredDiagnostics: () => {
deferredTasks.scheduleDiagnostics(() => {
@ -150,9 +172,6 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
deferredTasks.clearDiagnostics();
},
refreshTabHeader: context.refreshTabHeader,
resetThreadTurnPresence: (hadTurns) => {
autoTitle.resetThreadTurnPresence(hadTurns);
},
setStatus: status.set,
addSystemMessage: status.addSystemMessage,
configuredCommand: context.configuredCommand,
@ -163,7 +182,10 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
});
return {
connectionController,
connection: {
manager: connection,
controller: connectionController,
},
inboundController,
serverActions: {
threads: serverThreads,
@ -172,35 +194,3 @@ export function createChatConnectionBundle(context: ChatConnectionBundleContext)
},
};
}
export function createChatConnectionEventRouter(callbacks: { deferLiveStateRefresh: () => void }): {
handlers: ConnectionManagerHandlers;
attach: (targets: ChatConnectionEventTargets) => void;
} {
let targets: ChatConnectionEventTargets | null = null;
const currentTargets = () => {
if (!targets) throw new Error("Codex app-server connection event received before chat session parts were initialized.");
return targets;
};
return {
handlers: {
onNotification: (notification) => {
currentTargets().inbound.handleNotification(notification);
callbacks.deferLiveStateRefresh();
},
onServerRequest: (request) => {
currentTargets().inbound.handleServerRequest(request);
callbacks.deferLiveStateRefresh();
},
onLog: (message) => {
currentTargets().inbound.handleAppServerLog(message);
},
onExit: () => {
currentTargets().connectionController.handleExit();
},
},
attach: (nextTargets) => {
targets = nextTargets;
},
};
}

35
src/features/chat/host/session.ts
View file

@ -1,6 +1,6 @@
import { Notice, type App, type Component, type EventRef } from "obsidian";
import { ConnectionManager } from "../../../app-server/connection/connection-manager";
import type { ConnectionManager } from "../../../app-server/connection/connection-manager";
import type { ModelMetadata } from "../../../domain/catalog/metadata";
import type { Thread } from "../../../domain/threads/model";
import { getThreadTitle } from "../../../domain/threads/model";
@ -41,7 +41,7 @@ import type { ThreadRenameEditorController } from "../application/threads/rename
import type { RestorationController } from "../application/threads/restoration-controller";
import type { ResumeController } from "../application/threads/resume-controller";
import { createThreadParts, createThreadSelectionActions } from "../application/threads/composition";
import { createChatConnectionBundle, createChatConnectionEventRouter, type ChatConnectionBundle } from "./connection-bundle";
import { createChatConnectionBundle, type ChatConnectionBundle } from "./connection-bundle";
import type { MessageStreamPresenter } from "../panel/surface/message-stream-presenter";
import { pendingRequestsSignature } from "../domain/pending-requests/signatures";
import { createChatPanelGoalSurface } from "../panel/surface/goal-surface";
@ -300,25 +300,14 @@ export class ChatPanelSession {
}
private createSessionParts(): ChatPanelSessionParts {
const connectionHandlers = createChatConnectionEventRouter({
deferLiveStateRefresh: () => {
this.deferLiveStateRefresh();
},
});
const connection = new ConnectionManager(
() => this.environment.plugin.settingsRef.settings.codexPath,
this.environment.plugin.settingsRef.vaultPath,
connectionHandlers.handlers,
);
const sideEffects = this.createSideEffects();
const currentClient = () => connection.currentClient();
const ensureConnected = () => this.parts.connection.controller.ensureConnected();
const refreshThreads = () => this.parts.connection.controller.refreshThreads();
const refreshSkills = (forceReload?: boolean) => this.parts.connection.controller.refreshSkills(forceReload);
const runtimeSettings = createChatRuntimeSettingsActions({
stateStore: this.stateStore,
currentClient,
currentClient: () => this.parts.connection.manager.currentClient(),
runtimeSnapshotForState: runtimeSnapshotForChatState,
collaborationModeLabel: () => this.collaborationModeLabel(),
addSystemMessage: sideEffects.status.addSystemMessage,
@ -340,7 +329,7 @@ export class ChatPanelSession {
getClosing: () => this.closing,
},
client: {
getClient: currentClient,
getClient: () => this.parts.connection.manager.currentClient(),
ensureConnected,
},
status: sideEffects.status,
@ -419,13 +408,10 @@ export class ChatPanelSession {
const serverParts = createChatConnectionBundle({
stateStore: this.stateStore,
vaultPath: this.environment.plugin.settingsRef.vaultPath,
connection,
codexPath: () => this.environment.plugin.settingsRef.settings.codexPath,
connectionWork: this.connectionWork,
deferredTasks: this.deferredTasks,
threadSurfaces: this.environment.plugin.threadSurfaces,
client: {
currentClient,
},
refresh: {
refreshThreads,
refreshSkills,
@ -444,10 +430,15 @@ export class ChatPanelSession {
refreshLiveState: () => {
this.refreshLiveState();
},
deferLiveStateRefresh: () => {
this.deferLiveStateRefresh();
},
configuredCommand: () => this.environment.plugin.settingsRef.settings.codexPath,
});
const { connectionController, inboundController } = serverParts;
connectionHandlers.attach({ inbound: inboundController, connectionController });
const {
connection: { controller: connectionController, manager: connection },
inboundController,
} = serverParts;
const { threads: serverThreads, diagnostics: serverDiagnostics } = serverParts.serverActions;
const toolbarActions = createChatPanelToolbarActions(
@ -534,7 +525,7 @@ export class ChatPanelSession {
},
},
client: {
getClient: currentClient,
getClient: () => this.parts.connection.manager.currentClient(),
ensureConnected,
},
status: sideEffects.status,

12
src/features/threads-view/session.ts
View file

@ -1,7 +1,7 @@
import { Notice } from "obsidian";
import type { AppServerClient } from "../../app-server/connection/client";
import { ConnectionManager, StaleConnectionError } from "../../app-server/connection/connection-manager";
import { ConnectionManager, type ConnectionManagerHandlers, StaleConnectionError } from "../../app-server/connection/connection-manager";
import { listThreads, readCompletedConversationSummariesPage } from "../../app-server/services/threads";
import type { Thread } from "../../domain/threads/model";
import type { CodexPanelSettings } from "../../settings/model";
@ -73,7 +73,11 @@ export class CodexThreadsSession {
constructor(private readonly environment: CodexThreadsSessionEnvironment) {
this.deferredTasks = createThreadsViewDeferredTasks(() => this.viewWindow());
this.connection = new ConnectionManager(() => this.host.settings.codexPath, this.host.vaultPath, {
this.connection = new ConnectionManager(() => this.host.settings.codexPath, this.host.vaultPath);
}
private connectionHandlers(): ConnectionManagerHandlers {
return {
onNotification: () => {
this.scheduleRefresh();
},
@ -91,7 +95,7 @@ export class CodexThreadsSession {
this.status = { kind: "error", message: "Codex app-server stopped." };
this.render();
},
});
};
}
open(): void {
@ -178,7 +182,7 @@ export class CodexThreadsSession {
const connection = this.beginConnectionWork();
const promise = this.connection
.connect()
.connect(this.connectionHandlers())
.then(() => {
if (this.isStaleConnectionWork(connection)) throw new StaleConnectionError();
this.client = this.connection.currentClient();

14
tests/app-server/connection-manager.test.ts
View file

@ -54,7 +54,6 @@ describe("ConnectionManager", () => {
const manager = new ConnectionManager(
() => "/bin/codex",
"/vault",
silentConnectionHandlers(),
(codexPath, cwd, handlers) =>
new AppServerClient(codexPath, cwd, handlers, 5, (transportHandlers) => {
transport = new SilentTransport(transportHandlers);
@ -62,7 +61,7 @@ describe("ConnectionManager", () => {
}),
);
await expect(manager.connect()).rejects.toThrow("Codex app-server request timed out: initialize");
await expect(manager.connect(silentConnectionHandlers())).rejects.toThrow("Codex app-server request timed out: initialize");
expect(transport.running).toBe(false);
expect(manager.currentClient()).toBeNull();
@ -73,7 +72,6 @@ describe("ConnectionManager", () => {
const manager = new ConnectionManager(
() => "/bin/codex",
"/vault",
silentConnectionHandlers(),
(codexPath, cwd, handlers) =>
new AppServerClient(codexPath, cwd, handlers, 500, (transportHandlers) => {
transport = new SilentTransport(transportHandlers);
@ -81,8 +79,8 @@ describe("ConnectionManager", () => {
}),
);
const first = manager.connect();
const second = manager.connect();
const first = manager.connect(silentConnectionHandlers());
const second = manager.connect(silentConnectionHandlers());
transport.emitLine({ id: 1, result: { codexHome: "/tmp/codex" } });
await expect(first).resolves.toMatchObject({ codexHome: "/tmp/codex" });
@ -97,7 +95,6 @@ describe("ConnectionManager", () => {
const manager = new ConnectionManager(
() => "/bin/codex",
"/vault",
{ ...silentConnectionHandlers(), onExit },
(codexPath, cwd, handlers) =>
new AppServerClient(codexPath, cwd, handlers, 500, (transportHandlers) => {
transport = new SilentTransport(transportHandlers);
@ -105,7 +102,7 @@ describe("ConnectionManager", () => {
}),
);
const connecting = manager.connect();
const connecting = manager.connect({ ...silentConnectionHandlers(), onExit });
transport.emitExit();
await expect(connecting).rejects.toThrow("Codex app-server exited: unknown");
@ -119,7 +116,6 @@ describe("ConnectionManager", () => {
const manager = new ConnectionManager(
() => "/bin/codex",
"/vault",
{ ...silentConnectionHandlers(), onExit },
(codexPath, cwd, handlers) =>
new AppServerClient(codexPath, cwd, handlers, 500, (transportHandlers) => {
transport = new SilentTransport(transportHandlers);
@ -127,7 +123,7 @@ describe("ConnectionManager", () => {
}),
);
const connecting = manager.connect();
const connecting = manager.connect({ ...silentConnectionHandlers(), onExit });
manager.disconnect();
transport.emitLine({ id: 1, result: { codexHome: "/tmp/codex" } });

37
tests/features/chat/view-connection.test.ts
View file

@ -40,21 +40,25 @@ vi.mock("../../../src/app-server/connection/connection-manager", () => {
class StaleConnectionError extends Error {}
class ConnectionManager {
constructor(
_codexPath: () => string,
_cwd: string,
private readonly handlers: {
onNotification: (notification: ServerNotification) => void;
onServerRequest: (request: unknown) => void;
onLog: (message: string) => void;
onExit: () => void;
},
) {
connectionMock.state.onNotification = handlers.onNotification;
connectionMock.state.onExit = handlers.onExit;
private handlers: {
onNotification: (notification: ServerNotification) => void;
onServerRequest: (request: unknown) => void;
onLog: (message: string) => void;
onExit: () => void;
} | null;
constructor(_codexPath: () => string, _cwd: string) {
this.handlers = null;
}
connect(): Promise<unknown> {
connect(handlers: {
onNotification: (notification: ServerNotification) => void;
onServerRequest: (request: unknown) => void;
onLog: (message: string) => void;
onExit: () => void;
}): Promise<unknown> {
this.handlers = handlers;
this.publishHandlers(handlers);
connectionMock.state.connectCalls += 1;
connectionMock.state.connected = true;
return Promise.resolve({
@ -83,7 +87,12 @@ vi.mock("../../../src/app-server/connection/connection-manager", () => {
exit(): void {
connectionMock.state.connected = false;
this.handlers.onExit();
this.handlers?.onExit();
}
private publishHandlers(handlers: NonNullable<ConnectionManager["handlers"]>): void {
connectionMock.state.onNotification = handlers.onNotification;
connectionMock.state.onExit = handlers.onExit;
}
}

5
tests/features/threads-view/view.test.ts
View file

@ -35,11 +35,8 @@ vi.mock("../../../src/app-server/connection/connection-manager", () => {
class StaleConnectionError extends Error {}
class ConnectionManager {
constructor(_codexPath: () => string, _cwd: string, handlers: { onExit: () => void }) {
connect(handlers: { onExit: () => void }): Promise<unknown> {
connectionMock.state.onExit = handlers.onExit;
}
connect(): Promise<unknown> {
connectionMock.state.connectCalls += 1;
connectionMock.state.connected = true;
return Promise.resolve({