feat: execute automation run now

packages/opencode/src/automation/index.ts

@@ -7,6 +7,7 @@ import { ProjectID } from "@/project/schema"
 import { PermissionID } from "@/permission/schema"
 import { SessionID } from "@/session/schema"
 import { NotFoundError } from "@/storage/db"
+import type { AutomationRunAttendance, AutomationRunBlocker } from "./run-context"
 
 export const AutomationID = {
   Definition: {
@@ -232,8 +233,16 @@ export namespace Automation {
   type State = {
     definitions: Map<string, Definition>
     runs: Map<string, Run[]>
+    activeWriters: Set<string>
   }
-  const state = Instance.state<State>(() => ({ definitions: new Map(), runs: new Map() }))
+  const state = Instance.state<State>(() => ({ definitions: new Map(), runs: new Map(), activeWriters: new Set() }))
+
+  export type RunExecutor = (input: {
+    definition: Definition
+    run: Run
+    attendance: AutomationRunAttendance
+    signal: AbortSignal
+  }) => Promise<{ sessionID: SessionID; result: string | null; cost?: number | null }>
 
   const COMMON_CREATE_FIELDS = new Set([
     "kind",
@@ -416,6 +425,10 @@ export namespace Automation {
     return definition
   }
 
+  function getOptional(id: string): Definition | undefined {
+    return state().definitions.get(id)
+  }
+
   function isSameValue(left: unknown, right: unknown): boolean {
     if (Object.is(left, right)) return true
     if (Array.isArray(left) || Array.isArray(right)) {
@@ -460,6 +473,68 @@ export namespace Automation {
     return { id: previous.id, deleted: true, revision: previous.revision + 1 }
   }
 
+  function replaceRun(run: Run) {
+    const current = state().runs.get(run.automationID) ?? []
+    state().runs.set(
+      run.automationID,
+      current.map((item) => (item.id === run.id ? run : item)),
+    )
+    return run
+  }
+
+  function reviseRun(run: Run, patch: Record<string, unknown>): Run {
+    const next = {
+      ...run,
+      ...patch,
+      revision: run.revision + 1,
+    }
+    if (next.state !== "awaiting_input") delete (next as Record<string, unknown>).blocker
+    if (next.state !== "stopped") delete (next as Record<string, unknown>).stopReason
+    for (const [key, value] of Object.entries(next)) {
+      if (value === undefined) delete (next as Record<string, unknown>)[key]
+    }
+    return replaceRun(Run.parse(next))
+  }
+
+  export function markRunStarted(run: Run, sessionID: SessionID, options?: { now?: number }): Run {
+    return reviseRun(run, {
+      state: "running",
+      sessionID,
+      startedAt: options?.now ?? Date.now(),
+      completedAt: null,
+      result: null,
+      error: null,
+    })
+  }
+
+  function setDefinitionAutomationSession(definition: Definition, sessionID: SessionID) {
+    if (definition.automationSessionID === sessionID) return definition
+    const next = Definition.parse({
+      ...definition,
+      automationSessionID: sessionID,
+      revision: definition.revision + 1,
+      updatedAt: Date.now(),
+    })
+    state().definitions.set(definition.id, next)
+    return next
+  }
+
+  export function markRunBlocked(run: Run, blocker: AutomationRunBlocker): Run {
+    if (run.state !== "running" && run.state !== "awaiting_input") return run
+    return reviseRun(run, {
+      state: "awaiting_input",
+      blocker,
+    })
+  }
+
+  export function clearRunBlocker(run: Run): Run {
+    if (run.state !== "awaiting_input") return run
+    return reviseRun(run, {
+      state: "running",
+      blocker: undefined,
+    })
+  }
+
   export function runNow(id: string, options?: { now?: number }): Run {
     const definition = get(id)
     const current = state().runs.get(id) ?? []
@@ -481,6 +556,77 @@ export namespace Automation {
     return run
   }
 
+  export function runNowExecuting(
+    id: string,
+    options: { executor: RunExecutor; attendance?: AutomationRunAttendance; now?: number },
+  ): Run {
+    const initial = runNow(id, { now: options.now })
+    void executeRun(initial, options.executor, options.attendance ?? "attended")
+    return initial
+  }
+
+  async function executeRun(initial: Run, executor: RunExecutor, attendance: AutomationRunAttendance) {
+    const data = state()
+    const definition = get(initial.automationID)
+    const writerKey = definition.where.worktree ?? definition.where.projectID
+    if (data.activeWriters.has(writerKey)) {
+      const stopped = reviseRun(initial, {
+        state: "stopped",
+        completedAt: Date.now(),
+        stopReason: "previous_run_awaiting_input",
+      })
+      await publishRunUpdated(stopped)
+      return
+    }
+    data.activeWriters.add(writerKey)
+    const controller = new AbortController()
+    let current = initial
+    try {
+      const prepared = await executor({ definition, run: initial, attendance, signal: controller.signal })
+      const latest = state().runs.get(initial.automationID)?.find((item) => item.id === initial.id) ?? initial
+      const running = latest.state === "scheduled" ? markRunStarted(latest, prepared.sessionID) : latest
+      current = running
+      await publishRunUpdated(running)
+      const latestDefinition = getOptional(initial.automationID)
+      if (latestDefinition?.context === "continue") {
+        const updatedDefinition = setDefinitionAutomationSession(latestDefinition, prepared.sessionID)
+        if (updatedDefinition !== latestDefinition) await publishDefinitionUpdated(updatedDefinition)
+      }
+      const succeeded = reviseRun(running, {
+        state: "succeeded",
+        completedAt: Date.now(),
+        result: prepared.result,
+        error: null,
+        cost: prepared.cost ?? null,
+      })
+      current = succeeded
+      await publishRunUpdated(succeeded)
+    } catch (error) {
+      current = state().runs.get(initial.automationID)?.find((item) => item.id === initial.id) ?? current
+      if (current.state === "scheduled") {
+        const stopped = reviseRun(current, {
+          state: "stopped",
+          completedAt: Date.now(),
+          stopReason: "cancelled",
+        })
+        await publishRunUpdated(stopped)
+        return
+      }
+      const isStepCap = error instanceof globalThis.Error && error.name === "AutomationStepCapError"
+      const failed = reviseRun(current, {
+        state: "failed",
+        completedAt: Date.now(),
+        error: {
+          code: isStepCap ? "step_cap" : "execution_failed",
+          message: error instanceof globalThis.Error ? error.message : String(error),
+        },
+      })
+      await publishRunUpdated(failed)
+    } finally {
+      data.activeWriters.delete(writerKey)
+    }
+  }
+
   export function runs(input: { automationID: string; limit?: number; cursor?: string }) {
     get(input.automationID)
     const limit = Math.min(Math.max(input.limit ?? 50, 1), 100)

packages/opencode/src/automation/run-context.ts

@@ -0,0 +1,47 @@
+import { Context, Effect } from "effect"
+import type { Permission } from "@/permission"
+import type { PermissionID } from "@/permission/schema"
+
+export type AutomationRunAttendance = "attended" | "unattended"
+
+export type AutomationRunBlocker =
+  | { kind: "permission"; requestID: PermissionID }
+  | { kind: "question"; callID: string }
+
+export interface AutomationRunContext {
+  readonly attendance: AutomationRunAttendance
+  readonly stepCap?: number
+  readonly block: (blocker: AutomationRunBlocker) => Effect.Effect<void>
+  readonly clear: () => Effect.Effect<void>
+}
+
+export class AutomationStepCapError extends Error {
+  readonly _tag = "AutomationStepCapError"
+  constructor(readonly stepCap: number) {
+    super(`Automation run exceeded the hard step cap (${stepCap}).`)
+    this.name = "AutomationStepCapError"
+  }
+}
+
+export const AutomationRunContextService: Context.Reference<AutomationRunContext | undefined> = Context.Reference<
+  AutomationRunContext | undefined
+>("@opencode/AutomationRunContext", {
+  defaultValue: () => undefined,
+})
+
+export const AutomationRunContext = {
+  service: AutomationRunContextService,
+  current: AutomationRunContextService,
+  attended(input: Pick<AutomationRunContext, "block" | "clear" | "stepCap">): AutomationRunContext {
+    return { ...input, attendance: "attended" }
+  },
+  unattended(input: Pick<AutomationRunContext, "block" | "clear" | "stepCap">): AutomationRunContext {
+    return { ...input, attendance: "unattended" }
+  },
+  permissionOnPending(
+    context: AutomationRunContext | undefined,
+  ): ((request: Permission.Request) => Effect.Effect<void>) | undefined {
+    if (!context) return undefined
+    return (request) => context.block({ kind: "permission", requestID: request.id })
+  },
+}

packages/opencode/src/automation/runner.ts

@@ -0,0 +1,44 @@
+import { Effect } from "effect"
+import { Automation } from "."
+import { Session } from "@/session"
+import { SessionPrompt } from "@/session/prompt"
+import { AutomationRunContext, type AutomationRunBlocker } from "./run-context"
+
+export const sessionPromptExecutor: Automation.RunExecutor = async ({ definition, run, attendance, signal }) => {
+  const sessionID =
+    definition.context === "continue" && definition.automationSessionID
+      ? definition.automationSessionID
+      : (await Session.create({ title: `Automation: ${definition.title}` })).id
+  let currentRun = Automation.markRunStarted(run, sessionID)
+  await Automation.publishRunUpdated(currentRun)
+  const handlers = {
+    stepCap: 50,
+    block: (blocker: AutomationRunBlocker) =>
+      Effect.gen(function* () {
+        const previous = currentRun
+        currentRun = Automation.markRunBlocked(currentRun, blocker)
+        if (currentRun !== previous) yield* Effect.promise(() => Automation.publishRunUpdated(currentRun))
+      }),
+    clear: () =>
+      Effect.gen(function* () {
+        const previous = currentRun
+        currentRun = Automation.clearRunBlocker(currentRun)
+        if (currentRun !== previous) yield* Effect.promise(() => Automation.publishRunUpdated(currentRun))
+      }),
+  }
+  const scoped =
+    attendance === "attended" ? AutomationRunContext.attended(handlers) : AutomationRunContext.unattended(handlers)
+  const message = await SessionPrompt.promptWithAutomationContext(
+    {
+      sessionID,
+      parts: [{ type: "text", text: definition.prompt }],
+    },
+    scoped,
+  )
+  signal.throwIfAborted()
+  return {
+    sessionID,
+    result: message.parts.find((part) => part.type === "text")?.text ?? null,
+    cost: message.info.role === "assistant" ? message.info.cost : null,
+  }
+}

packages/opencode/src/permission/index.ts

@@ -117,8 +117,12 @@ export namespace Permission {
     message: z.string().optional(),
   })
 
+  export type AskOptions = z.infer<typeof AskInput> & {
+    onPending?: (request: Request) => Effect.Effect<void>
+  }
+
   export interface Interface {
-    readonly ask: (input: z.infer<typeof AskInput>) => Effect.Effect<void, Error>
+    readonly ask: (input: AskOptions) => Effect.Effect<void, Error>
     readonly reply: (input: z.infer<typeof ReplyInput>) => Effect.Effect<void>
     readonly clearSession: (
       sessionID: SessionID,
@@ -171,9 +175,9 @@ export namespace Permission {
         }),
       )
 
-      const ask = Effect.fn("Permission.ask")(function* (input: z.infer<typeof AskInput>) {
+      const ask = Effect.fn("Permission.ask")(function* (input: AskOptions) {
         const { approved, pending } = yield* InstanceState.get(state)
-        const { ruleset, ...request } = input
+        const { ruleset, onPending, ...request } = input
         let needsAsk = false
         const denied: Array<{ pattern: string; rule: Rule }> = []
 
@@ -222,6 +226,7 @@ export namespace Permission {
 
         const deferred = yield* Deferred.make<void, RejectedError | CorrectedError>()
         pending.set(id, { info, deferred })
+        if (onPending) yield* onPending(info)
         yield* bus.publish(Event.Asked, info)
         return yield* Effect.ensuring(
           Deferred.await(deferred),

packages/opencode/src/server/instance/automation.ts

@@ -3,6 +3,7 @@ import type { Context } from "hono"
 import { describeRoute, resolver, validator } from "hono-openapi"
 import z from "zod"
 import { Automation, AutomationID, ValidationError } from "@/automation"
+import { sessionPromptExecutor } from "@/automation/runner"
 import { errors } from "../error"
 
 function validationError(error: ValidationError) {
@@ -229,7 +230,7 @@ export const AutomationRoutes = (): Hono =>
       "/:automationID/run",
       describeRoute({
         summary: "Run automation now",
-        description: "Create a scheduled automation run record. Execution lands in a later PR.",
+        description: "Create a queued automation run, start execution in the background, and return the queued run immediately.",
         operationId: "automation.runNow",
         responses: {
           200: {
@@ -241,7 +242,9 @@ export const AutomationRoutes = (): Hono =>
       }),
       validator("param", z.object({ automationID: AutomationID.Definition.zod })),
       async (c) => {
-        const run = Automation.runNow(c.req.valid("param").automationID)
+        const run = Automation.runNowExecuting(c.req.valid("param").automationID, {
+          executor: sessionPromptExecutor,
+        })
         await Automation.publishRunUpdated(run)
         return c.json(run)
       },