Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions src/CodexAcpClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import type {
GetAccountResponse,
ListMcpServerStatusResponse,
Model,
ReviewTarget,
SkillsListParams,
SkillsListResponse,
Thread,
Expand Down Expand Up @@ -274,6 +275,22 @@ export class CodexAcpClient {
await this.codexClient.threadArchive({threadId: sessionId});
}

async runReview(
sessionId: string,
target: ReviewTarget,
onTurnStarted?: (turnId: string) => void,
): Promise<TurnCompletedNotification> {
return await this.codexClient.runReview({
threadId: sessionId,
target,
delivery: "inline",
}, onTurnStarted);
}

async runCompact(sessionId: string): Promise<void> {
await this.codexClient.runCompact({threadId: sessionId});
}

async awaitMcpServerStartup(serverNames: Array<string>, afterVersion: number): Promise<McpStartupResult> {
return await this.codexClient.awaitMcpServerStartup(serverNames, afterVersion);
}
Expand Down
28 changes: 27 additions & 1 deletion src/CodexAcpServer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1259,8 +1259,34 @@ export class CodexAcpServer implements acp.Agent {
approvalHandler,
elicitationHandler);

if (await this.availableCommands.tryHandleCommand(params.prompt, sessionState)) {
const commandResult = await this.availableCommands.tryHandleCommand(params.prompt, sessionState);
if (commandResult.handled) {
logger.log("Prompt handled by a command");
await this.codexAcpClient.waitForSessionNotifications(params.sessionId);
if (commandResult.turnCompleted?.turn.status === "interrupted") {
if (!this.sessionIsClosing(params.sessionId) && this.sessions.has(params.sessionId)) {
await this.connection.sessionUpdate({
sessionId: params.sessionId,
update: {
sessionUpdate: "agent_message_chunk",
content: {
type: "text",
text: "*Conversation interrupted*"
}
}
});
}
return {
stopReason: "cancelled",
usage: this.buildPromptUsage(sessionState.lastTokenUsage),
_meta: this.buildQuotaMeta(sessionState),
};
}
const error = eventHandler.getFailure()
if (error) {
// noinspection ExceptionCaughtLocallyJS
throw error;
}
return {
stopReason: "end_turn",
usage: this.buildPromptUsage(sessionState.lastTokenUsage),
Expand Down
77 changes: 77 additions & 0 deletions src/CodexAppServerClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,15 @@ import type {
McpServerStatusUpdatedNotification,
ModelListParams,
ModelListResponse,
ReviewStartParams,
ReviewStartResponse,
SkillsExtraRootsSetParams,
SkillsListParams,
SkillsListResponse,
ThreadArchiveParams,
ThreadArchiveResponse,
ThreadCompactStartParams,
ThreadCompactStartResponse,
ThreadLoadedListParams,
ThreadLoadedListResponse,
ThreadListParams,
Expand All @@ -47,6 +51,7 @@ import type {
CommandExecutionRequestApprovalResponse,
FileChangeRequestApprovalParams,
FileChangeRequestApprovalResponse,
ItemCompletedNotification,
} from "./app-server/v2";

export interface ApprovalHandler {
Expand Down Expand Up @@ -99,6 +104,7 @@ export class CodexAppServerClient {
private readonly mcpServerStartupStates = new Map<string, McpServerStartupSnapshot>();
private readonly mcpServerStartupResolvers: Array<McpServerStartupResolver> = [];
private readonly pendingTurnCompletionResolvers = new Map<string, Map<string, (event: TurnCompletedNotification) => void>>();
private readonly pendingCompactionCompletionResolvers = new Map<string, Set<(event: CompactionCompletedNotification) => void>>();
private readonly turnCompletionCaptures = new Map<string, Set<(event: TurnCompletedNotification) => void>>();
private readonly staleTurnIds = new Map<string, Set<string>>();

Expand All @@ -118,6 +124,9 @@ export class CodexAppServerClient {
if (isTurnCompletedNotification(serverNotification)) {
this.recordTurnCompleted(serverNotification.params);
}
if (isCompactionCompletedNotification(serverNotification)) {
this.recordCompactionCompleted(serverNotification);
}
const routing = extractTurnRouting(serverNotification);
const staleTurnNotification = this.isStaleTurn(routing.threadId, routing.turnId);
if (staleTurnNotification) {
Expand Down Expand Up @@ -213,10 +222,40 @@ export class CodexAppServerClient {
}
}

async runReview(params: ReviewStartParams, onTurnStarted?: (turnId: string) => void): Promise<TurnCompletedNotification> {
const capturedCompletions: Array<TurnCompletedNotification> = [];
const releaseCapture = this.captureTurnCompletions(params.threadId, (event) => {
capturedCompletions.push(event);
});

try {
const reviewStarted = await this.reviewStart(params);
onTurnStarted?.(reviewStarted.turn.id);
const earlyCompletion = capturedCompletions.find(event => event.turn.id === reviewStarted.turn.id);
releaseCapture();
if (earlyCompletion) {
return earlyCompletion;
}
return await this.awaitTurnCompleted(reviewStarted.reviewThreadId, reviewStarted.turn.id);
} finally {
releaseCapture();
}
}

async runCompact(params: ThreadCompactStartParams): Promise<CompactionCompletedNotification> {
const compactionCompleted = this.awaitCompactionCompleted(params.threadId);
await this.threadCompactStart(params);
return await compactionCompleted;
}

async turnInterrupt(params: TurnInterruptParams): Promise<TurnInterruptResponse> {
return await this.sendRequest({ method: "turn/interrupt", params: params });
}

async reviewStart(params: ReviewStartParams): Promise<ReviewStartResponse> {
return await this.sendRequest({ method: "review/start", params: params });
}

markTurnStale(threadId: string, turnId: string): void {
const threadStaleTurns = this.staleTurnIds.get(threadId) ?? new Set<string>();
threadStaleTurns.add(turnId);
Expand Down Expand Up @@ -251,6 +290,10 @@ export class CodexAppServerClient {
return await this.sendRequest({ method: "thread/unsubscribe", params: params });
}

async threadCompactStart(params: ThreadCompactStartParams): Promise<ThreadCompactStartResponse> {
return await this.sendRequest({ method: "thread/compact/start", params: params });
}

async listMcpServerStatus(params: ListMcpServerStatusParams): Promise<ListMcpServerStatusResponse> {
return await this.sendRequest({ method: "mcpServerStatus/list", params });
}
Expand Down Expand Up @@ -303,6 +346,14 @@ export class CodexAppServerClient {
});
}

async awaitCompactionCompleted(threadId: string): Promise<CompactionCompletedNotification> {
return await new Promise((resolve) => {
const resolvers = this.pendingCompactionCompletionResolvers.get(threadId) ?? new Set();
resolvers.add(resolve);
this.pendingCompactionCompletionResolvers.set(threadId, resolvers);
});
}

resolveTurnInterrupted(threadId: string, turnId: string): void {
this.recordTurnCompleted({
threadId,
Expand Down Expand Up @@ -380,6 +431,21 @@ export class CodexAppServerClient {
}
}

private recordCompactionCompleted(event: CompactionCompletedNotification): void {
const threadId = extractThreadId(event);
if (threadId === null) {
return;
}
const resolvers = this.pendingCompactionCompletionResolvers.get(threadId);
if (!resolvers) {
return;
}
this.pendingCompactionCompletionResolvers.delete(threadId);
for (const resolve of resolvers) {
resolve(event);
}
}

private isStaleTurn(threadId: string | null, turnId: string | null): boolean {
if (threadId === null || turnId === null) {
return false;
Expand Down Expand Up @@ -493,6 +559,10 @@ export type CodexConnectionEvent =
| ({ eventType: "response" } & unknown)
| ({ eventType: "notification" } & ServerNotification);

export type CompactionCompletedNotification =
| { method: "thread/compacted", params: Extract<ServerNotification, { method: "thread/compacted" }>["params"] }
| { method: "item/completed", params: ItemCompletedNotification & { item: Extract<ItemCompletedNotification["item"], { type: "contextCompaction" }> } };

type CodexRequest = DistributiveOmit<ClientRequest, "id">

type DistributiveOmit<T, K extends keyof any> = T extends any
Expand Down Expand Up @@ -525,6 +595,13 @@ function isTurnCompletedNotification(notification: ServerNotification): notifica
return notification.method === "turn/completed";
}

function isCompactionCompletedNotification(notification: ServerNotification): notification is CompactionCompletedNotification {
if (notification.method === "thread/compacted") {
return true;
}
return notification.method === "item/completed" && notification.params.item.type === "contextCompaction";
}

function extractThreadId(notification: ServerNotification): string | null {
const params = notification.params as { threadId?: unknown } | undefined;
if (params && typeof params.threadId === "string") {
Expand Down
Loading
Loading