Add conversation message branching

src/cli/daemon/agent/__tests__/claude.test.ts

@@ -72,6 +72,23 @@ describe("ClaudeBackend", () => {
     return currentMockProc!;
   }
 
+  it("passes --fork-session when resuming a branch", async () => {
+    const session = backend.execute("hello", {
+      cwd: "/tmp",
+      resumeSessionId: "parent_session",
+      forkSession: true,
+    });
+    const mock = getMock();
+
+    const spawnCall = (spawn as any).mock.calls[0];
+    expect(spawnCall[1]).toContain("--resume");
+    expect(spawnCall[1]).toContain("parent_session");
+    expect(spawnCall[1]).toContain("--fork-session");
+
+    mock.proc.emit("close", 0);
+    await session.result;
+  });
+
   it("emits MessageText for assistant text blocks", async () => {
     const session = backend.execute("hello", { cwd: "/tmp" });
     const mock = getMock();

src/cli/daemon/agent/__tests__/codex.test.ts

@@ -214,6 +214,40 @@ describe("CodexBackend", () => {
     await session.result;
   });
 
+  it("uses thread/fork instead of thread/resume when resuming a branch", async () => {
+    const session = backend.execute("branch prompt", {
+      cwd: "/tmp",
+      model: "gpt-4",
+      resumeSessionId: "parent_thread",
+      forkSession: true,
+    });
+    const mock = getMock();
+
+    await tick();
+    sendResponse(1, {});
+    await tick();
+
+    const forkWrite = mock.stdinWrites.find((w) => w.includes('"thread/fork"'));
+    expect(forkWrite).toBeDefined();
+    const parsedFork = JSON.parse(forkWrite!);
+    expect(parsedFork.params.threadId).toBe("parent_thread");
+    expect(parsedFork.params.model).toBe("gpt-4");
+    expect(mock.stdinWrites.some((w) => w.includes('"thread/resume"'))).toBe(false);
+
+    sendResponse(2, { thread: { id: "branch_thread" } });
+    await tick();
+
+    const turnWrite = mock.stdinWrites.find((w) => w.includes('"turn/start"'));
+    expect(turnWrite).toBeDefined();
+    const parsedTurn = JSON.parse(turnWrite!);
+    expect(parsedTurn.params.threadId).toBe("branch_thread");
+    expect(parsedTurn.params.input).toEqual([{ type: "text", text: "branch prompt" }]);
+
+    sendResponse(3, {});
+    mock.proc.emit("close", 0);
+    await session.result;
+  });
+
   it("extracts session ID from thread/start response", async () => {
     const session = backend.execute("hello", { cwd: "/tmp" });
     const mock = getMock();

src/cli/daemon/agent/claude.ts

@@ -28,6 +28,9 @@ export class ClaudeBackend implements AgentBackend {
     }
     if (options.resumeSessionId) {
       args.push("--resume", options.resumeSessionId);
+      if (options.forkSession) {
+        args.push("--fork-session");
+      }
     }
 
     const proc = spawn(this.cliPath, args, {

src/cli/daemon/agent/codex.ts

@@ -474,7 +474,13 @@ export class CodexBackend implements AgentBackend {
 
           // 3. Start or resume thread
           let threadResponse: unknown;
-          if (options.resumeSessionId) {
+          if (options.resumeSessionId && options.forkSession) {
+            threadResponse = await sendRpc("thread/fork", {
+              threadId: options.resumeSessionId,
+              ...(options.model ? { model: options.model } : {}),
+            });
+            sessionId = extractThreadID(threadResponse);
+          } else if (options.resumeSessionId) {
             // thread/resume reopens an existing thread by ID
             threadResponse = await sendRpc("thread/resume", {
               threadId: options.resumeSessionId,

src/cli/daemon/session-runner.test.ts

@@ -854,6 +854,101 @@ describe("session-runner runSession", () => {
     );
   });
 
+  it("forks from the pinned root-message parent session for branch tasks", async () => {
+    setupBackend([], {
+      status: "completed",
+      output: "Forked",
+      error: "",
+      durationMs: 100,
+      sessionId: "branch-session",
+    });
+
+    await runSession(
+      makeInput({
+        task: {
+          ...makeInput().task,
+          conversationId: "branch_c",
+          contextKey: "branch_c",
+          context: {
+            runtime_branch: {
+              parent_context_key: "parent_c",
+              parent_task_id: "root_task",
+              parent_session_id: "root-message-session",
+              root_message_id: "root_m",
+              provider: "claude",
+            },
+          },
+        },
+      }),
+    );
+
+    expect(mockFindResumableSessionByContextKey).not.toHaveBeenCalled();
+    expect(mockBackendExecute).toHaveBeenCalledWith(
+      expect.any(String),
+      expect.objectContaining({
+        resumeSessionId: "root-message-session",
+        forkSession: true,
+      }),
+    );
+  });
+
+  it("fails branch tasks before spawning when runtime provider does not match branch provider", async () => {
+    await runSession(
+      makeInput({
+        provider: "claude",
+        task: {
+          ...makeInput().task,
+          conversationId: "branch_c",
+          contextKey: "branch_c",
+          context: {
+            runtime_branch: {
+              parent_context_key: "parent_c",
+              parent_task_id: "root_task",
+              parent_session_id: "root-session",
+              root_message_id: "root_m",
+              provider: "codex",
+            },
+          },
+        },
+      }),
+    );
+
+    expect(mockFindResumableSessionByContextKey).not.toHaveBeenCalled();
+    expect(mockBackendExecute).not.toHaveBeenCalled();
+    expect(mockClientInstance.failTask).toHaveBeenCalledWith(
+      "test_token",
+      "t1",
+      expect.stringContaining("does not match runtime provider claude"),
+    );
+  });
+
+  it("fails branch tasks before spawning when no pinned parent session is provided", async () => {
+    await runSession(
+      makeInput({
+        task: {
+          ...makeInput().task,
+          conversationId: "branch_c",
+          contextKey: "branch_c",
+          context: {
+            runtime_branch: {
+              parent_context_key: "parent_c",
+              root_message_id: "root_m",
+              provider: "claude",
+            },
+          },
+        },
+      }),
+    );
+
+    expect(mockFindResumableSessionByContextKey).not.toHaveBeenCalled();
+    expect(mockBackendExecute).not.toHaveBeenCalled();
+    expect(mockClientInstance.failTask).toHaveBeenCalledWith(
+      "test_token",
+      "t1",
+      expect.stringContaining("pinned parent session is required"),
+    );
+  });
+
   it("session starts fresh when provider has no matching prior entry", async () => {
     mockFindResumableSessionByContextKey.mockReturnValueOnce(null);
 

src/cli/daemon/session-runner.ts

@@ -318,11 +318,83 @@ export async function runSession(input: SessionRunnerInput): Promise<void> {
 
   const prompt = input.promptOverride ?? buildPrompt(task, attachments);
 
-  const resumeSessionId = task.contextKey
-    ? findResumableSessionByContextKey(timelineDir, task.contextKey, provider) ?? undefined
-    : undefined;
+  async function failBeforeSpawn(errMsg: string) {
+    log.error(errMsg);
+    updateEntry(timelineDir, task.id, (entry) => {
+      entry.pid = null;
+      entry.status = "failed";
+      entry.errmsg = errMsg;
+    });
+    await reportToServer(
+      () => client.failTask(token, task.id, errMsg),
+      { taskId: task.id, type: "fail", payload: { error: errMsg }, token, serverURL, createdAt: new Date().toISOString() },
+      workspacesRoot,
+    );
+    process.removeListener("SIGTERM", onKill);
+    process.removeListener("SIGINT", onKill);
+  }
+
+  const runtimeBranch = task.context?.runtime_branch as
+    | {
+        parent_context_key?: unknown;
+        parent_task_id?: unknown;
+        parent_session_id?: unknown;
+        root_message_id?: unknown;
+        provider?: unknown;
+      }
+    | undefined;
+  const branchTask = Boolean(runtimeBranch);
+  const branchParentContextKey =
+    typeof runtimeBranch?.parent_context_key === "string"
+      ? runtimeBranch.parent_context_key
+      : null;
+  const branchParentTaskId =
+    typeof runtimeBranch?.parent_task_id === "string"
+      ? runtimeBranch.parent_task_id
+      : null;
+  const branchParentSessionId =
+    typeof runtimeBranch?.parent_session_id === "string"
+      ? runtimeBranch.parent_session_id
+      : null;
+  const branchRootMessageId =
+    typeof runtimeBranch?.root_message_id === "string"
+      ? runtimeBranch.root_message_id
+      : null;
+  const branchProvider =
+    typeof runtimeBranch?.provider === "string" ? runtimeBranch.provider : null;
+  const forkSession = branchTask;
+  if (forkSession && !branchParentContextKey) {
+    await failBeforeSpawn("cannot branch: parent context key is required");
+    return;
+  }
+  if (forkSession && !branchParentSessionId) {
+    await failBeforeSpawn(
+      `cannot branch: pinned parent session is required for root_message_id ${branchRootMessageId ?? "unknown"}`,
+    );
+    return;
+  }
+  if (forkSession && !branchProvider) {
+    await failBeforeSpawn("cannot branch: branch provider is required");
+    return;
+  }
+  if (forkSession && branchProvider !== provider) {
+    await failBeforeSpawn(
+      `cannot branch: branch provider ${branchProvider} does not match runtime provider ${provider}`,
+    );
+    return;
+  }
+  const resumeContextKey = forkSession ? branchParentContextKey : task.contextKey ?? null;
+  const resumeSessionId = forkSession
+    ? branchParentSessionId!
+    : resumeContextKey
+      ? findResumableSessionByContextKey(timelineDir, resumeContextKey, provider) ?? undefined
+      : undefined;
   if (resumeSessionId) {
-    log.info(`resuming session ${resumeSessionId} (context_key: ${task.contextKey})`);
+    log.info(
+      `${forkSession ? "forking" : "resuming"} session ${resumeSessionId} (context_key: ${resumeContextKey}${
+        branchParentTaskId ? `, parent_task_id: ${branchParentTaskId}` : ""
+      })`,
+    );
   }
 
   const session = backend.execute(prompt, {
@@ -331,6 +403,7 @@ export async function runSession(input: SessionRunnerInput): Promise<void> {
     env,
     timeout: agentTimeout,
     resumeSessionId,
+    forkSession,
   });
 
   // Capture agent PID so the handler can reap it. backend.execute() spawns

src/cli/daemon/types.ts

@@ -88,6 +88,7 @@ export interface ExecOptions {
   maxTurns?: number;
   timeout?: number;
   resumeSessionId?: string;
+  forkSession?: boolean;
 }
 
 /** Serialized input passed from daemon to the detached session-runner process. */

src/shared/src/api-types.ts

@@ -129,6 +129,7 @@ export interface UpdateAgentRequest {
 
 export interface SendMessageRequest {
   content: string;
+  metadata?: Record<string, unknown>;
 }
 
 export interface CreateMachineTokenRequest {

src/shared/src/constants.ts

@@ -48,6 +48,17 @@ export const TASK_TYPES = {
 
 export type TaskType = (typeof TASK_TYPES)[keyof typeof TASK_TYPES];
 
+export const CONVERSATION_TYPES = {
+  USER_DM_MESSAGE: TASK_TYPES.USER_DM_MESSAGE,
+  EMAIL_NOTIFICATION: TASK_TYPES.EMAIL_NOTIFICATION,
+  CALENDAR_EVENT: TASK_TYPES.CALENDAR_EVENT,
+  ISSUE_EVENT: TASK_TYPES.ISSUE_EVENT,
+  MESSAGE_BRANCH: "message_branch",
+} as const;
+
+export type ConversationType =
+  (typeof CONVERSATION_TYPES)[keyof typeof CONVERSATION_TYPES];
+
 export const IssueStatus = {
   TODO: "todo",
   IN_PROGRESS: "in_progress",

src/shared/src/db/queries-index.ts

@@ -4,6 +4,7 @@ export * as member from "./queries/member";
 export * as agent from "./queries/agent";
 export * as runtime from "./queries/runtime";
 export * as conversation from "./queries/conversation";
+export * as conversationBranch from "./queries/conversation-branch";
 export * as message from "./queries/message";
 export * as task from "./queries/task";
 export * as taskMessage from "./queries/task-message";