Checkpointing & Diffs

apps/server/src/codexAppServerManager.test.ts

@@ -43,6 +43,36 @@ function createSendTurnHarness() {
   return { manager, context, requireSession, sendRequest, updateSession };
 }
 
+function createThreadControlHarness() {
+  const manager = new CodexAppServerManager();
+  const context = {
+    session: {
+      sessionId: "sess_1",
+      provider: "codex",
+      status: "ready",
+      threadId: "thread_1",
+      createdAt: "2026-02-10T00:00:00.000Z",
+      updatedAt: "2026-02-10T00:00:00.000Z",
+    },
+  };
+
+  const requireSession = vi
+    .spyOn(
+      manager as unknown as { requireSession: (sessionId: string) => unknown },
+      "requireSession",
+    )
+    .mockReturnValue(context);
+  const sendRequest = vi.spyOn(
+    manager as unknown as { sendRequest: (...args: unknown[]) => Promise<unknown> },
+    "sendRequest",
+  );
+  const updateSession = vi
+    .spyOn(manager as unknown as { updateSession: (...args: unknown[]) => void }, "updateSession")
+    .mockImplementation(() => {});
+
+  return { manager, context, requireSession, sendRequest, updateSession };
+}
+
 describe("classifyCodexStderrLine", () => {
   it("ignores empty lines", () => {
     expect(classifyCodexStderrLine("   ")).toBeNull();
@@ -113,6 +143,40 @@ describe("isRecoverableThreadResumeError", () => {
   });
 });
 
+describe("startSession", () => {
+  it("emits session/startFailed when resolving cwd throws before process launch", async () => {
+    const manager = new CodexAppServerManager();
+    const events: Array<{ method: string; kind: string; message?: string }> = [];
+    manager.on("event", (event) => {
+      events.push({
+        method: event.method,
+        kind: event.kind,
+        ...(event.message ? { message: event.message } : {}),
+      });
+    });
+
+    const processCwd = vi.spyOn(process, "cwd").mockImplementation(() => {
+      throw new Error("cwd missing");
+    });
+    try {
+      await expect(
+        manager.startSession({
+          provider: "codex",
+        }),
+      ).rejects.toThrow("cwd missing");
+      expect(events).toHaveLength(1);
+      expect(events[0]).toEqual({
+        method: "session/startFailed",
+        kind: "error",
+        message: "cwd missing",
+      });
+    } finally {
+      processCwd.mockRestore();
+      manager.stopAll();
+    }
+  });
+});
+
 describe("sendTurn", () => {
   it("sends text and image user input items to turn/start", async () => {
     const { manager, context, requireSession, sendRequest, updateSession } =
@@ -198,3 +262,91 @@ describe("sendTurn", () => {
     ).rejects.toThrow("Turn input must include text or attachments.");
   });
 });
+
+describe("thread checkpoint control", () => {
+  it("reads thread turns from thread/read", async () => {
+    const { manager, context, requireSession, sendRequest } = createThreadControlHarness();
+    sendRequest.mockResolvedValue({
+      thread: {
+        id: "thread_1",
+        turns: [
+          {
+            id: "turn_1",
+            items: [{ type: "userMessage", content: [{ type: "text", text: "hello" }] }],
+          },
+        ],
+      },
+    });
+
+    const result = await manager.readThread("sess_1");
+
+    expect(requireSession).toHaveBeenCalledWith("sess_1");
+    expect(sendRequest).toHaveBeenCalledWith(context, "thread/read", {
+      threadId: "thread_1",
+      includeTurns: true,
+    });
+    expect(result).toEqual({
+      threadId: "thread_1",
+      turns: [
+        {
+          id: "turn_1",
+          items: [{ type: "userMessage", content: [{ type: "text", text: "hello" }] }],
+        },
+      ],
+    });
+  });
+
+  it("reads thread turns from flat thread/read responses", async () => {
+    const { manager, context, sendRequest } = createThreadControlHarness();
+    sendRequest.mockResolvedValue({
+      threadId: "thread_1",
+      turns: [
+        {
+          id: "turn_1",
+          items: [{ type: "userMessage", content: [{ type: "text", text: "hello" }] }],
+        },
+      ],
+    });
+
+    const result = await manager.readThread("sess_1");
+
+    expect(sendRequest).toHaveBeenCalledWith(context, "thread/read", {
+      threadId: "thread_1",
+      includeTurns: true,
+    });
+    expect(result).toEqual({
+      threadId: "thread_1",
+      turns: [
+        {
+          id: "turn_1",
+          items: [{ type: "userMessage", content: [{ type: "text", text: "hello" }] }],
+        },
+      ],
+    });
+  });
+
+  it("rolls back turns via thread/rollback and resets session running state", async () => {
+    const { manager, context, sendRequest, updateSession } = createThreadControlHarness();
+    sendRequest.mockResolvedValue({
+      thread: {
+        id: "thread_1",
+        turns: [],
+      },
+    });
+
+    const result = await manager.rollbackThread("sess_1", 2);
+
+    expect(sendRequest).toHaveBeenCalledWith(context, "thread/rollback", {
+      threadId: "thread_1",
+      numTurns: 2,
+    });
+    expect(updateSession).toHaveBeenCalledWith(context, {
+      status: "ready",
+      activeTurnId: undefined,
+    });
+    expect(result).toEqual({
+      threadId: "thread_1",
+      turns: [],
+    });
+  });
+});

apps/server/src/codexAppServerManager.ts

@@ -65,6 +65,16 @@ interface JsonRpcNotification {
   params?: unknown;
 }
 
+export interface CodexThreadTurnSnapshot {
+  id: string;
+  items: unknown[];
+}
+
+export interface CodexThreadSnapshot {
+  threadId: string;
+  turns: CodexThreadTurnSnapshot[];
+}
+
 const ANSI_ESCAPE_CHAR = String.fromCharCode(27);
 const ANSI_ESCAPE_REGEX = new RegExp(`${ANSI_ESCAPE_CHAR}\\[[0-9;]*m`, "g");
 const CODEX_STDERR_LOG_REGEX =
@@ -138,40 +148,43 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
   async startSession(input: ProviderSessionStartInput): Promise<ProviderSession> {
     const sessionId = randomUUID();
     const now = new Date().toISOString();
+    let context: CodexSessionContext | undefined;
 
-    const session: ProviderSession = {
-      sessionId,
-      provider: "codex",
-      status: "connecting",
-      model: normalizeCodexModelSlug(input.model),
-      cwd: input.cwd,
-      createdAt: now,
-      updatedAt: now,
-    };
+    try {
+      const resolvedCwd = input.cwd ?? process.cwd();
+
+      const session: ProviderSession = {
+        sessionId,
+        provider: "codex",
+        status: "connecting",
+        model: normalizeCodexModelSlug(input.model),
+        cwd: resolvedCwd,
+        createdAt: now,
+        updatedAt: now,
+      };
 
-    const child = spawn("codex", ["app-server"], {
-      cwd: input.cwd,
-      env: process.env,
-      stdio: ["pipe", "pipe", "pipe"],
-    });
-    const output = readline.createInterface({ input: child.stdout });
-
-    const context: CodexSessionContext = {
-      session,
-      child,
-      output,
-      pending: new Map(),
-      pendingApprovals: new Map(),
-      nextRequestId: 1,
-      stopping: false,
-    };
+      const child = spawn("codex", ["app-server"], {
+        cwd: resolvedCwd,
+        env: process.env,
+        stdio: ["pipe", "pipe", "pipe"],
+      });
+      const output = readline.createInterface({ input: child.stdout });
+
+      context = {
+        session,
+        child,
+        output,
+        pending: new Map(),
+        pendingApprovals: new Map(),
+        nextRequestId: 1,
+        stopping: false,
+      };
 
-    this.sessions.set(sessionId, context);
-    this.attachProcessListeners(context);
+      this.sessions.set(sessionId, context);
+      this.attachProcessListeners(context);
 
-    this.emitLifecycleEvent(context, "session/connecting", "Starting codex app-server");
+      this.emitLifecycleEvent(context, "session/connecting", "Starting codex app-server");
 
-    try {
       await this.sendRequest(context, "initialize", {
         clientInfo: {
           name: "t3code_desktop",
@@ -241,12 +254,24 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
       return { ...context.session };
     } catch (error) {
       const message = error instanceof Error ? error.message : "Failed to start Codex session.";
-      this.updateSession(context, {
-        status: "error",
-        lastError: message,
-      });
-      this.emitErrorEvent(context, "session/startFailed", message);
-      this.stopSession(sessionId);
+      if (context) {
+        this.updateSession(context, {
+          status: "error",
+          lastError: message,
+        });
+        this.emitErrorEvent(context, "session/startFailed", message);
+        this.stopSession(sessionId);
+      } else {
+        this.emitEvent({
+          id: randomUUID(),
+          kind: "error",
+          provider: "codex",
+          sessionId,
+          createdAt: new Date().toISOString(),
+          method: "session/startFailed",
+          message,
+        });
+      }
       throw new Error(message, { cause: error });
     }
   }
@@ -331,6 +356,41 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
     });
   }
 
+  async readThread(sessionId: string): Promise<CodexThreadSnapshot> {
+    const context = this.requireSession(sessionId);
+    const threadId = context.session.threadId;
+    if (!threadId) {
+      throw new Error("Session is missing a thread id.");
+    }
+
+    const response = await this.sendRequest(context, "thread/read", {
+      threadId,
+      includeTurns: true,
+    });
+    return this.parseThreadSnapshot("thread/read", response);
+  }
+
+  async rollbackThread(sessionId: string, numTurns: number): Promise<CodexThreadSnapshot> {
+    const context = this.requireSession(sessionId);
+    const threadId = context.session.threadId;
+    if (!threadId) {
+      throw new Error("Session is missing a thread id.");
+    }
+    if (!Number.isInteger(numTurns) || numTurns < 1) {
+      throw new Error("numTurns must be an integer >= 1.");
+    }
+
+    const response = await this.sendRequest(context, "thread/rollback", {
+      threadId,
+      numTurns,
+    });
+    this.updateSession(context, {
+      status: "ready",
+      activeTurnId: undefined,
+    });
+    return this.parseThreadSnapshot("thread/rollback", response);
+  }
+
   async respondToRequest(
     sessionId: string,
     requestId: string,
@@ -742,6 +802,31 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
     return undefined;
   }
 
+  private parseThreadSnapshot(method: string, response: unknown): CodexThreadSnapshot {
+    const responseRecord = this.readObject(response);
+    const thread = this.readObject(responseRecord, "thread");
+    const threadId = this.readString(thread, "id") ?? this.readString(responseRecord, "threadId");
+    if (!threadId) {
+      throw new Error(`${method} response did not include a thread id.`);
+    }
+
+    const turnsRaw = this.readArray(thread, "turns") ?? this.readArray(responseRecord, "turns") ?? [];
+    const turns = turnsRaw.map((turnValue, index) => {
+      const turn = this.readObject(turnValue);
+      const turnId = this.readString(turn, "id") ?? `${threadId}:turn:${index + 1}`;
+      const items = this.readArray(turn, "items") ?? [];
+      return {
+        id: turnId,
+        items,
+      };
+    });
+
+    return {
+      threadId,
+      turns,
+    };
+  }
+
   private isServerRequest(value: unknown): value is JsonRpcRequest {
     if (!value || typeof value !== "object") {
       return false;
@@ -823,6 +908,16 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
     return target as Record<string, unknown>;
   }
 
+  private readArray(value: unknown, key?: string): unknown[] | undefined {
+    const target =
+      key === undefined
+        ? value
+        : value && typeof value === "object"
+          ? (value as Record<string, unknown>)[key]
+          : undefined;
+    return Array.isArray(target) ? target : undefined;
+  }
+
   private readString(value: unknown, key: string): string | undefined {
     if (!value || typeof value !== "object") {
       return undefined;