feat(session): centralize async session work through SessionRunner

edlsh · edlsh · commit 020cf5ce9886 · 2025-12-23T22:07:29.000-05:00
- Fix circular import by using regular function for runOnce
diff --git a/packages/opencode/src/server/server.ts b/packages/opencode/src/server/server.ts
@@ -32,6 +32,7 @@ import { ProjectRoute } from "./project"
 import { ToolRegistry } from "../tool/registry"
 import { zodToJsonSchema } from "zod-to-json-schema"
 import { SessionPrompt } from "../session/prompt"
+import { SessionRunner } from "../session/runner"
 import { SessionCompaction } from "../session/compaction"
 import { SessionRevert } from "../session/revert"
 import { lazy } from "../util/lazy"
@@ -1325,13 +1326,14 @@ export namespace Server {
         ),
         validator("json", SessionPrompt.PromptInput.omit({ sessionID: true })),
         async (c) => {
-          c.status(204)
-          c.header("Content-Type", "application/json")
-          return stream(c, async () => {
-            const sessionID = c.req.valid("param").sessionID
-            const body = c.req.valid("json")
-            SessionPrompt.prompt({ ...body, sessionID })
-          })
+          const sessionID = c.req.valid("param").sessionID
+          const body = c.req.valid("json")
+          void SessionRunner.promptBackground({
+            ...body,
+            sessionID,
+            kind: "session.prompt_async",
+          }).catch((err) => log.error("prompt_async failed", { sessionID, error: err }))
+          return c.body(null, 204)
         },
       )
       .post(
diff --git a/packages/opencode/src/session/runner.ts b/packages/opencode/src/session/runner.ts
@@ -377,20 +377,107 @@ export namespace SessionRunner {
     return Object.keys(state().abortById)
   }
 
-  export const runOnce = fn(SessionPrompt.PromptInput, async (input): Promise<MessageV2.WithParts> => {
+  export async function runOnce(input: SessionPrompt.PromptInput): Promise<MessageV2.WithParts> {
     log.info("runOnce", { sessionID: input.sessionID, agent: input.agent })
     return SessionPrompt.prompt(input)
+  }
+
+  export const LoopBackgroundInput = z.object({
+    sessionID: Identifier.schema("session"),
+    kind: JobKind.optional(),
+    parentSessionID: Identifier.schema("session").optional(),
+    toolCallID: z.string().optional(),
+    timeoutMs: z.number().optional(),
+    dedupeKey: z.string().optional(),
+  })
+  export type LoopBackgroundInput = z.infer<typeof LoopBackgroundInput>
+
+  export const loopBackground = fn(LoopBackgroundInput, async (input): Promise<Job> => {
+    log.info("loopBackground", { sessionID: input.sessionID })
+    return enqueue(
+      input.kind ?? "session.loop",
+      input.sessionID,
+      async () => {
+        await SessionPrompt.loop(input.sessionID)
+      },
+      {
+        parentSessionID: input.parentSessionID,
+        toolCallID: input.toolCallID,
+        timeoutMs: input.timeoutMs,
+        dedupeKey: input.dedupeKey,
+      },
+    )
   })
 
-  export function runBackground(_id: string, _options: Options): void {
-    throw new Error("SessionRunner.runBackground not yet implemented")
-  }
+  // Note: Defined inline to avoid circular dependency with SessionPrompt
+  // (prompt.ts -> task.ts -> runner.ts -> prompt.ts)
+  export const PromptBackgroundInput = z.object({
+    sessionID: Identifier.schema("session"),
+    messageID: Identifier.schema("message").optional(),
+    model: z
+      .object({
+        providerID: z.string(),
+        modelID: z.string(),
+      })
+      .optional(),
+    agent: z.string().optional(),
+    noReply: z.boolean().optional(),
+    tools: z.record(z.string(), z.boolean()).optional(),
+    system: z.string().optional(),
+    parts: z.array(
+      z.discriminatedUnion("type", [
+        MessageV2.TextPart.omit({ messageID: true, sessionID: true }).partial({ id: true }),
+        MessageV2.FilePart.omit({ messageID: true, sessionID: true }).partial({ id: true }),
+        MessageV2.AgentPart.omit({ messageID: true, sessionID: true }).partial({ id: true }),
+        MessageV2.SubtaskPart.omit({ messageID: true, sessionID: true }).partial({ id: true }),
+      ]),
+    ),
+    kind: JobKind.optional(),
+    parentSessionID: Identifier.schema("session").optional(),
+    toolCallID: z.string().optional(),
+    timeoutMs: z.number().optional(),
+    dedupeKey: z.string().optional(),
+  })
+  export type PromptBackgroundInput = z.infer<typeof PromptBackgroundInput>
+
+  export const PromptBackgroundResult = z.object({
+    job: Job,
+    message: MessageV2.WithParts,
+  })
+  export type PromptBackgroundResult = z.infer<typeof PromptBackgroundResult>
+
+  export const promptBackground = fn(PromptBackgroundInput, async (input): Promise<PromptBackgroundResult> => {
+    log.info("promptBackground", { sessionID: input.sessionID, agent: input.agent })
+    const message = await SessionPrompt.prompt({ ...input, noReply: true })
+    const job = await loopBackground({
+      sessionID: input.sessionID,
+      kind: input.kind ?? "session.prompt_async",
+      parentSessionID: input.parentSessionID,
+      toolCallID: input.toolCallID,
+      timeoutMs: input.timeoutMs,
+      dedupeKey: input.dedupeKey,
+    })
+    return { job, message }
+  })
 
-  export function cancelBackground(_id: string): boolean {
-    throw new Error("SessionRunner.cancelBackground not yet implemented")
+  export function cancelBySession(sessionID: string): boolean {
+    const s = state()
+    for (const job of Object.values(s.jobsById)) {
+      if (job.targetSessionID === sessionID && (job.status === "queued" || job.status === "running")) {
+        SessionPrompt.cancel(sessionID)
+        return cancel(job.id)
+      }
+    }
+    return false
   }
 
-  export async function waitFor(_id: string): Promise<RunResult> {
-    throw new Error("SessionRunner.waitFor not yet implemented")
+  export function getBySession(sessionID: string): Job | undefined {
+    const s = state()
+    for (const job of Object.values(s.jobsById)) {
+      if (job.targetSessionID === sessionID && (job.status === "queued" || job.status === "running")) {
+        return job
+      }
+    }
+    return undefined
   }
 }
diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts
@@ -7,8 +7,8 @@ import { MessageV2 } from "../session/message-v2"
 import { Identifier } from "../id/id"
 import { Agent } from "../agent/agent"
 import { SessionPrompt } from "../session/prompt"
+import { SessionRunner } from "../session/runner"
 import { iife } from "@/util/iife"
-import { defer } from "@/util/defer"
 import { Config } from "../config/config"
 
 export const TaskTool = Tool.define("task", async () => {
@@ -81,32 +81,62 @@ export const TaskTool = Tool.define("task", async () => {
         providerID: msg.info.providerID,
       }
 
-      function cancel() {
+      const promptParts = await SessionPrompt.resolvePromptParts(params.prompt)
+      const config = await Config.get()
+
+      const cancelChild = () => {
+        SessionRunner.cancelBySession(session.id)
         SessionPrompt.cancel(session.id)
       }
-      ctx.abort.addEventListener("abort", cancel)
-      using _ = defer(() => ctx.abort.removeEventListener("abort", cancel))
-      const promptParts = await SessionPrompt.resolvePromptParts(params.prompt)
+      ctx.abort.addEventListener("abort", cancelChild, { once: true })
 
-      const config = await Config.get()
-      const result = await SessionPrompt.prompt({
-        messageID,
-        sessionID: session.id,
-        model: {
-          modelID: model.modelID,
-          providerID: model.providerID,
+      let result: MessageV2.WithParts | undefined
+      const job = await SessionRunner.enqueue(
+        "task.child_session",
+        session.id,
+        async () => {
+          result = await SessionPrompt.prompt({
+            messageID,
+            sessionID: session.id,
+            model: {
+              modelID: model.modelID,
+              providerID: model.providerID,
+            },
+            agent: agent.name,
+            tools: {
+              todowrite: false,
+              todoread: false,
+              task: false,
+              ...Object.fromEntries((config.experimental?.primary_tools ?? []).map((t) => [t, false])),
+              ...agent.tools,
+            },
+            parts: promptParts,
+          })
         },
-        agent: agent.name,
-        tools: {
-          todowrite: false,
-          todoread: false,
-          task: false,
-          ...Object.fromEntries((config.experimental?.primary_tools ?? []).map((t) => [t, false])),
-          ...agent.tools,
+        {
+          parentSessionID: ctx.sessionID,
+          toolCallID: ctx.messageID,
         },
-        parts: promptParts,
-      })
+      )
       unsub()
+      ctx.abort.removeEventListener("abort", cancelChild)
+
+      if (job.status === "canceled" || ctx.abort.aborted) {
+        return {
+          title: params.description,
+          metadata: { summary: [], sessionId: session.id },
+          output: "Task was canceled",
+        }
+      }
+
+      if (job.status !== "completed" || !result) {
+        const error = job.error?.message ?? "Task failed"
+        return {
+          title: params.description,
+          metadata: { summary: [], sessionId: session.id },
+          output: `Task failed: ${error}`,
+        }
+      }
       const messages = await Session.messages({ sessionID: session.id })
       const summary = messages
         .filter((x) => x.info.role === "assistant")
diff --git a/packages/opencode/test/session/runner.test.ts b/packages/opencode/test/session/runner.test.ts
@@ -107,21 +107,118 @@ describe("SessionRunner", () => {
     })
   })
 
-  describe("stub methods", () => {
-    test("runBackground throws", () => {
-      const opts: SessionRunner.Options = {
+  describe("LoopBackgroundInput schema", () => {
+    test("validates minimal input", () => {
+      const input = { sessionID: "session_123" }
+      expect(SessionRunner.LoopBackgroundInput.safeParse(input).success).toBe(true)
+    })
+
+    test("validates full input", () => {
+      const input = {
+        sessionID: "session_123",
+        kind: "session.loop" as const,
+        parentSessionID: "session_parent",
+        toolCallID: "tool_123",
+        timeoutMs: 30000,
+        dedupeKey: "my-key",
+      }
+      expect(SessionRunner.LoopBackgroundInput.safeParse(input).success).toBe(true)
+    })
+  })
+
+  describe("PromptBackgroundInput schema", () => {
+    test("validates minimal input", () => {
+      const input = {
+        sessionID: "session_123",
+        parts: [{ type: "text" as const, text: "hello" }],
+      }
+      expect(SessionRunner.PromptBackgroundInput.safeParse(input).success).toBe(true)
+    })
+
+    test("validates full input with job options", () => {
+      const input = {
+        sessionID: "session_123",
+        parts: [{ type: "text" as const, text: "hello" }],
         model: { providerID: "anthropic", modelID: "claude-3-5-sonnet" },
-        agent: "code",
+        agent: "build",
+        kind: "session.prompt_async" as const,
+        parentSessionID: "session_parent",
+        timeoutMs: 60000,
       }
-      expect(() => SessionRunner.runBackground("session_123", opts)).toThrow("not yet implemented")
+      expect(SessionRunner.PromptBackgroundInput.safeParse(input).success).toBe(true)
+    })
+  })
+
+  describe("helper functions", () => {
+    test("cancelBySession returns false for unknown session", async () => {
+      await using tmp = await tmpdir({ git: true })
+      await Instance.provide({
+        directory: tmp.path,
+        fn: () => {
+          expect(SessionRunner.cancelBySession("unknown_session")).toBe(false)
+        },
+      })
+    })
+
+    test("getBySession returns undefined for unknown session", async () => {
+      await using tmp = await tmpdir({ git: true })
+      await Instance.provide({
+        directory: tmp.path,
+        fn: () => {
+          expect(SessionRunner.getBySession("unknown_session")).toBeUndefined()
+        },
+      })
     })
 
-    test("cancelBackground throws", () => {
-      expect(() => SessionRunner.cancelBackground("session_123")).toThrow("not yet implemented")
+    test("getBySession finds active job", async () => {
+      await using tmp = await tmpdir({ git: true })
+      await Instance.provide({
+        directory: tmp.path,
+        fn: async () => {
+          let resolve: () => void
+          const blocker = new Promise<void>((r) => {
+            resolve = r
+          })
+
+          const jobPromise = SessionRunner.enqueue("session.loop", "ses_target", async () => {
+            await blocker
+          })
+
+          await new Promise((r) => setTimeout(r, 10))
+
+          const found = SessionRunner.getBySession("ses_target")
+          expect(found).toBeDefined()
+          expect(found?.targetSessionID).toBe("ses_target")
+          expect(["queued", "running"]).toContain(found!.status)
+
+          resolve!()
+          await jobPromise
+        },
+      })
     })
 
-    test("waitFor throws", async () => {
-      await expect(SessionRunner.waitFor("session_123")).rejects.toThrow("not yet implemented")
+    test("cancelBySession cancels running job", async () => {
+      await using tmp = await tmpdir({ git: true })
+      await Instance.provide({
+        directory: tmp.path,
+        fn: async () => {
+          let started = false
+          const jobPromise = SessionRunner.enqueue("session.loop", "ses_cancel", async () => {
+            started = true
+            await new Promise((r) => setTimeout(r, 5000))
+          })
+
+          while (!started) {
+            await new Promise((r) => setTimeout(r, 5))
+          }
+
+          const cancelled = SessionRunner.cancelBySession("ses_cancel")
+          expect(cancelled).toBe(true)
+
+          const job = await jobPromise
+          expect(job.status).toBe("canceled")
+        },
+      })
     })
   })
 
