Check in missing files for command parsing, and fixes for some approval flows

Author: achimala

apps/server/src/agents/adapters/codex-agent.ts

@@ -11,12 +11,16 @@ import {
 } from "@farfield/api";
 import {
   ProtocolValidationError,
+  parseCommandExecutionRequestApprovalResponse,
+  parseFileChangeRequestApprovalResponse,
+  parseToolRequestUserInputResponsePayload,
   parseThreadConversationState,
   parseThreadStreamStateChangedBroadcast,
   parseUserInputResponsePayload,
   type IpcFrame,
   type IpcRequestFrame,
   type IpcResponseFrame,
+  type ThreadConversationRequest,
   type ThreadConversationState,
   type ThreadStreamStateChangedBroadcast,
   type UserInputRequestId,
@@ -593,21 +597,121 @@ export class CodexAgentAdapter implements AgentAdapter {
     input: AgentSubmitUserInputInput,
   ): Promise<{ ownerClientId: string; requestId: UserInputRequestId }> {
     this.ensureCodexAvailable();
-    this.ensureIpcReady();
+    const parsedResponse = parseUserInputResponsePayload(input.response);
+    const ownerClientIdForResult = (() => {
+      const mapped = this.threadOwnerById.get(input.threadId);
+      if (mapped && mapped.trim().length > 0) {
+        return mapped.trim();
+      }
+      if (input.ownerClientId && input.ownerClientId.trim().length > 0) {
+        return input.ownerClientId.trim();
+      }
+      if (this.lastKnownOwnerClientId && this.lastKnownOwnerClientId.trim()) {
+        return this.lastKnownOwnerClientId.trim();
+      }
+      return "app-server";
+    })();
+
+    const threadForRouting = await this.runThreadOperationWithResumeRetry(
+      input.threadId,
+      () => this.appClient.readThread(input.threadId, false),
+    );
+    const parsedRoutingThread = parseThreadConversationState(threadForRouting.thread);
+    const routingPendingRequest = findPendingRequestWithId(
+      parsedRoutingThread,
+      input.requestId,
+    );
+
+    if (routingPendingRequest) {
+      await this.runAppServerCall(() =>
+        this.appClient.submitUserInput(input.requestId, parsedResponse),
+      );
+
+      const refreshedThread = await this.runThreadOperationWithResumeRetry(
+        input.threadId,
+        () => this.appClient.readThread(input.threadId, true),
+      );
+      const parsedThread = parseThreadConversationState(refreshedThread.thread);
+      this.streamSnapshotByThreadId.set(input.threadId, parsedThread);
+      this.streamSnapshotOriginByThreadId.set(input.threadId, "readThreadWithTurns");
+      this.setThreadTitle(input.threadId, parsedThread.title);
 
+      const currentEvents = this.streamEventsByThreadId.get(input.threadId) ?? [];
+      currentEvents.push(
+        buildSyntheticSnapshotEvent(input.threadId, ownerClientIdForResult, parsedThread),
+      );
+      if (currentEvents.length > 400) {
+        currentEvents.splice(0, currentEvents.length - 400);
+      }
+      this.streamEventsByThreadId.set(input.threadId, currentEvents);
+
+      return {
+        ownerClientId: ownerClientIdForResult,
+        requestId: input.requestId,
+      };
+    }
+
+    this.ensureIpcReady();
     const ownerClientId = resolveOwnerClientId(
       this.threadOwnerById,
       input.threadId,
       input.ownerClientId,
       this.lastKnownOwnerClientId ?? undefined,
     );
+    this.threadOwnerById.set(input.threadId, ownerClientId);
 
-    await this.service.submitUserInput({
-      threadId: input.threadId,
-      ownerClientId,
-      requestId: input.requestId,
-      response: parseUserInputResponsePayload(input.response),
-    });
+    const pendingIpcRequest = await this.resolvePendingIpcRequest(
+      input.threadId,
+      input.requestId,
+    );
+    switch (pendingIpcRequest.method) {
+      case "item/commandExecution/requestApproval": {
+        const commandResponse =
+          parseCommandExecutionRequestApprovalResponse(parsedResponse);
+        await this.service.submitCommandApprovalDecision({
+          threadId: input.threadId,
+          ownerClientId,
+          requestId: input.requestId,
+          response: commandResponse,
+        });
+        break;
+      }
+      case "item/fileChange/requestApproval": {
+        const fileResponse = parseFileChangeRequestApprovalResponse(
+          parsedResponse,
+        );
+        await this.service.submitFileApprovalDecision({
+          threadId: input.threadId,
+          ownerClientId,
+          requestId: input.requestId,
+          response: fileResponse,
+        });
+        break;
+      }
+      case "item/tool/requestUserInput": {
+        const toolResponse = parseToolRequestUserInputResponsePayload(
+          parsedResponse,
+        );
+        await this.service.submitUserInput({
+          threadId: input.threadId,
+          ownerClientId,
+          requestId: input.requestId,
+          response: toolResponse,
+        });
+        break;
+      }
+      case "execCommandApproval":
+      case "applyPatchApproval":
+        throw new Error(
+          `Legacy approval request method ${pendingIpcRequest.method} is not supported over desktop IPC for thread ${input.threadId}`,
+        );
+      case "account/chatgptAuthTokens/refresh":
+      case "item/tool/call":
+      case "item/plan/requestImplementation":
+        throw new Error(
+          `Unsupported pending request method ${pendingIpcRequest.method} for submitUserInput on thread ${input.threadId}`,
+        );
+    }
 
     return {
       ownerClientId,
@@ -1045,6 +1149,34 @@ export class CodexAgentAdapter implements AgentAdapter {
     return this.runAppServerCall(operation);
   }
 
+  private async resolvePendingIpcRequest(
+    threadId: string,
+    requestId: UserInputRequestId,
+  ): Promise<ThreadConversationRequest> {
+    const cachedSnapshot = this.streamSnapshotByThreadId.get(threadId);
+    if (cachedSnapshot) {
+      const pending = findPendingRequestWithId(cachedSnapshot, requestId);
+      if (pending) {
+        return pending;
+      }
+    }
+
+    const liveState = await this.readLiveState(threadId);
+    if (liveState.conversationState) {
+      const pending = findPendingRequestWithId(
+        liveState.conversationState,
+        requestId,
+      );
+      if (pending) {
+        return pending;
+      }
+    }
+
+    throw new Error(
+      `Unable to find pending request ${String(requestId)} in live state for thread ${threadId}`,
+    );
+  }
+
   private resolveThreadTitle(
     threadId: string,
     directTitle: string | null | undefined,
@@ -1229,6 +1361,28 @@ function deriveThreadWaitingState(
   };
 }
 
+function requestIdsMatch(
+  left: UserInputRequestId,
+  right: UserInputRequestId,
+): boolean {
+  return `${left}` === `${right}`;
+}
+
+function findPendingRequestWithId(
+  state: ThreadConversationState,
+  requestId: UserInputRequestId,
+): ThreadConversationRequest | null {
+  for (const request of state.requests) {
+    if (request.completed === true) {
+      continue;
+    }
+    if (requestIdsMatch(request.id, requestId)) {
+      return request;
+    }
+  }
+  return null;
+}
+
 function buildSyntheticSnapshotEvent(
   threadId: string,
   sourceClientId: string,

apps/server/src/unified/adapter.ts

@@ -1137,6 +1137,13 @@ function mapTurnItem(
         review: item.review,
       };
 
+    case "remoteTaskCreated":
+      return {
+        id: item.id,
+        type: "remoteTaskCreated",
+        taskId: item.taskId,
+      };
+
     case "modelChanged":
       return {
         id: item.id,

apps/server/test/unified-adapter.test.ts

@@ -570,4 +570,49 @@ describe("unified provider adapters", () => {
     expect(result.data[0]?.waitingOnApproval).toBe(true);
     expect(result.data[0]?.waitingOnUserInput).toBe(true);
   });
+
+  it("maps remoteTaskCreated turn items into unified items", async () => {
+    const adapter = createCodexAdapter();
+    adapter.readThread = async () => ({
+      thread: {
+        ...SAMPLE_THREAD,
+        turns: [
+          {
+            id: "turn-1",
+            status: "inProgress",
+            items: [
+              {
+                id: "item-remote-task",
+                type: "remoteTaskCreated",
+                taskId: "task-123",
+              },
+            ],
+          },
+        ],
+      },
+    });
+    const unified = new AgentUnifiedProviderAdapter("codex", adapter);
+
+    const result = await unified.execute(
+      UnifiedCommandSchema.parse({
+        kind: "readThread",
+        provider: "codex",
+        threadId: SAMPLE_THREAD.id,
+        includeTurns: true,
+      }),
+    );
+
+    expect(result.kind).toBe("readThread");
+    if (result.kind !== "readThread") {
+      return;
+    }
+
+    const remoteTaskItem = result.thread.turns[0]?.items[0];
+    expect(remoteTaskItem?.type).toBe("remoteTaskCreated");
+    expect(
+      remoteTaskItem && remoteTaskItem.type === "remoteTaskCreated"
+        ? remoteTaskItem.taskId
+        : null,
+    ).toBe("task-123");
+  });
 });

apps/web/src/App.tsx

@@ -1176,7 +1176,7 @@ export function App(): React.JSX.Element {
       return readConversationState;
     }
 
-    return readConversationState;
+    return liveConversationState;
   }, [liveState?.conversationState, readThreadState?.thread]);
 
   const pendingRequests = useMemo(() => {

apps/web/src/components/ConversationItem.tsx

@@ -26,6 +26,7 @@ const TOOL_BLOCK_TYPES: readonly UnifiedItem["type"][] = [
   "webSearch",
   "mcpToolCall",
   "collabAgentToolCall",
+  "remoteTaskCreated",
   "forkedFromConversation",
 ];
 
@@ -313,6 +314,19 @@ const ITEM_RENDERERS = {
     </div>
   ),
 
+  remoteTaskCreated: ({ item, toolSpacing }) => (
+    <div
+      className={`${toolSpacing} rounded-lg border border-border bg-muted/20 px-3 py-2 text-xs text-muted-foreground`}
+    >
+      <div className="text-[10px] text-muted-foreground font-mono mb-1 uppercase tracking-wider">
+        Remote task
+      </div>
+      <div className="text-xs text-foreground/90 whitespace-pre-wrap break-all">
+        Created task: {item.taskId}
+      </div>
+    </div>
+  ),
+
   modelChanged: (_args) => (
     <div className="rounded-lg border border-border bg-muted/20 px-3 py-2 text-xs text-muted-foreground">
       Model changed
@@ -388,6 +402,8 @@ function renderItem(
       return ITEM_RENDERERS.enteredReviewMode({ item, ...context });
     case "exitedReviewMode":
       return ITEM_RENDERERS.exitedReviewMode({ item, ...context });
+    case "remoteTaskCreated":
+      return ITEM_RENDERERS.remoteTaskCreated({ item, ...context });
     case "modelChanged":
       return ITEM_RENDERERS.modelChanged({ item, ...context });
     case "forkedFromConversation":

apps/web/src/lib/code-language.ts

@@ -0,0 +1,53 @@
+const EXTENSION_LANGUAGE_MAP: Record<string, string> = {
+  cjs: "javascript",
+  cpp: "cpp",
+  cs: "csharp",
+  css: "css",
+  go: "go",
+  h: "c",
+  htm: "markup",
+  html: "markup",
+  java: "java",
+  js: "javascript",
+  json: "json",
+  jsx: "jsx",
+  kt: "kotlin",
+  md: "markdown",
+  mjs: "javascript",
+  py: "python",
+  rb: "ruby",
+  rs: "rust",
+  scss: "scss",
+  sh: "bash",
+  sql: "sql",
+  svg: "markup",
+  swift: "swift",
+  toml: "toml",
+  ts: "typescript",
+  tsx: "tsx",
+  txt: "text",
+  xml: "markup",
+  yaml: "yaml",
+  yml: "yaml",
+  zsh: "bash",
+};
+
+export function languageFromPath(path: string): string {
+  const trimmedPath = path.trim();
+  if (trimmedPath.length === 0) {
+    return "text";
+  }
+
+  const fileName = trimmedPath.split("/").pop()?.toLowerCase() ?? "";
+  if (fileName === "dockerfile") {
+    return "docker";
+  }
+
+  const extensionIndex = fileName.lastIndexOf(".");
+  if (extensionIndex < 0 || extensionIndex === fileName.length - 1) {
+    return "text";
+  }
+
+  const extension = fileName.slice(extensionIndex + 1);
+  return EXTENSION_LANGUAGE_MAP[extension] ?? "text";
+}