feat(core): interrupt v2 session execution

AGENTS.md

@@ -143,7 +143,7 @@ const table = sqliteTable("session", {
 
 - Keep durable prompt admission separate from model execution. `SessionV2.prompt(...)` admits one durable `session_input` row before scheduling advisory `SessionExecution.wake(sessionID)` unless `resume: false` requests admit-only behavior. The serialized runner promotes admitted inputs into visible user messages at safe boundaries.
 - Reusing a Session ID adopts the existing Session. Reusing a prompt message ID reconciles an exact retry only when Session, prompt, and delivery mode match; conflicting reuse fails. Historical projected prompts lazily synthesize promoted inbox records during exact retry.
-- Keep `SessionExecution` process-global and Session-ID based. It discovers placement through the read-side `SessionStore` and `LocationServiceMap.get(session.location)`; no layer should take a Session ID.
+- Keep `SessionExecution` process-global and Session-ID based. Its local implementation owns the process-local Session coordinator and discovers placement through `SessionStore` plus `LocationServiceMap.get(session.location)` only when a drain starts; no layer should take a Session ID. V2 interruption targets the active process-local ownership chain for that Session; idle or missing interruption is a no-op.
 - Keep `SessionRunner`, model resolution, tool registry, permissions, and filesystem Location-scoped. Omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics.
 - Preserve one explicit `llm.stream(request)` call per provider turn and reload projected history before durable continuation. Do not bridge through legacy `SessionPrompt.loop(...)` or delegate orchestration to an in-memory tool loop.
 - Keep local Session drains process-local until clustering is implemented. `SessionRunCoordinator` joins explicit same-Session resumes, coalesces prompt wakeups, and allows different Sessions to run concurrently. Advisory wakes drain eligible durable inbox rows only; post-crash activity recovery requires a separate explicit design before it may retry provider work.

packages/core/src/location-layer.ts

@@ -40,7 +40,6 @@ import { LLMClient } from "@opencode-ai/llm"
 import { RequestExecutor } from "@opencode-ai/llm/route"
 import * as SessionRunnerLLM from "./session/runner/llm"
 import { SessionRunnerModel } from "./session/runner/model"
-import { SessionRunCoordinator } from "./session/run-coordinator"
 import { SystemContextBuiltIns } from "./system-context/builtins"
 import { FetchHttpClient } from "effect/unstable/http"
 
@@ -87,7 +86,6 @@ export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("
       Layer.provide(model),
       Layer.provide(skillGuidance),
     )
-    const coordinator = SessionRunCoordinator.layer.pipe(Layer.provide(runner))
     return Layer.mergeAll(
       services,
       commits,
@@ -97,7 +95,6 @@ export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("
       questions,
       model,
       runner,
-      coordinator,
       builtInTools,
     ).pipe(Layer.fresh)
   },

packages/core/src/public/opencode.ts

@@ -51,6 +51,7 @@ export const layer = Layer.effect(
           }),
         get: sessions.get,
         list: sessions.list,
+        interrupt: sessions.interrupt,
         prompt: (input) =>
           sessions.prompt({
             id: input.id,

packages/core/src/public/session.ts

@@ -84,6 +84,8 @@ export interface Interface {
   readonly get: (sessionID: ID) => Effect.Effect<Info, NotFoundError>
   readonly list: (input?: ListInput) => Effect.Effect<Info[]>
   readonly prompt: (input: PromptInput) => Effect.Effect<Admission, NotFoundError | PromptConflictError>
+  /** Interrupt the active V2 execution chain for one Session on this process. Interrupting an idle or missing Session is a no-op. */
+  readonly interrupt: (sessionID: ID) => Effect.Effect<void>
   readonly messages: (input: MessagesInput) => Effect.Effect<Message[], NotFoundError | MessageDecodeError>
   readonly message: (input: MessageInput) => Effect.Effect<Message | undefined>
   readonly context: (sessionID: ID) => Effect.Effect<Message[], NotFoundError | MessageDecodeError>

packages/core/src/session.ts

@@ -1,7 +1,7 @@
 export * as SessionV2 from "./session"
 export * from "./session/schema"
 
-import { Cause, Effect, Layer, Schema, Context, Stream } from "effect"
+import { Cause, DateTime, Effect, Layer, Schema, Context, Stream } from "effect"
 import { and, asc, desc, eq, gt, like, lt, or, type SQL } from "drizzle-orm"
 import { ProjectV2 } from "./project"
 import { WorkspaceV2 } from "./workspace"
@@ -155,6 +155,7 @@ export interface Interface {
   readonly compact: (input: CompactInput) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
   readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
+  readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Session") {}
@@ -171,13 +172,13 @@ export const layer = Layer.effect(
     const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
     const scope = yield* Effect.scope
 
-    const enqueueWake = (sessionID: SessionSchema.ID) =>
-      execution.wake(sessionID).pipe(
+    const enqueueWake = (admitted: SessionInput.Admitted) =>
+      execution.wake(admitted.sessionID, admitted.admittedSeq).pipe(
         Effect.tapCause((cause) =>
           Cause.hasInterruptsOnly(cause)
             ? Effect.void
             : Effect.logError("Failed to wake Session").pipe(
-                Effect.annotateLogs("sessionID", sessionID),
+                Effect.annotateLogs("sessionID", admitted.sessionID),
                 Effect.annotateLogs("cause", cause),
               ),
         ),
@@ -351,7 +352,7 @@ export const layer = Layer.effect(
           Effect.gen(function* () {
             yield* result.get(input.sessionID)
             const returnPrompt = Effect.fnUntraced(function* (admitted: SessionInput.Admitted) {
-              if (input.resume !== false) yield* enqueueWake(input.sessionID)
+              if (input.resume !== false) yield* enqueueWake(admitted)
               return admitted
             }, Effect.uninterruptible)
             const messageID = input.id ?? SessionMessage.ID.create()
@@ -399,6 +400,20 @@ export const layer = Layer.effect(
         yield* result.get(sessionID)
         yield* execution.resume(sessionID)
       }),
+      interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
+        Effect.uninterruptible(
+          Effect.gen(function* () {
+            const session = yield* store.get(sessionID)
+            if (!session) return yield* execution.interrupt(sessionID)
+            const event = yield* events.publish(SessionEvent.InterruptRequested, {
+              sessionID,
+              timestamp: yield* DateTime.now,
+            })
+            if (event.seq === undefined) return yield* Effect.die("Interrupt request event is missing aggregate sequence")
+            yield* execution.interrupt(sessionID, event.seq)
+          }),
+        ),
+      ),
     })
 
     return result

packages/core/src/session/event.ts

@@ -119,6 +119,13 @@ export namespace PromptLifecycle {
   export type Promoted = typeof Promoted.Type
 }
 
+export const InterruptRequested = EventV2.define({
+  type: "session.next.interrupt.requested",
+  ...options,
+  schema: Base,
+})
+export type InterruptRequested = typeof InterruptRequested.Type
+
 export const ContextUpdated = EventV2.define({
   type: "session.next.context.updated",
   ...options,
@@ -455,6 +462,7 @@ const DurableDefinitions = [
   Prompted,
   PromptLifecycle.Admitted,
   PromptLifecycle.Promoted,
+  InterruptRequested,
   ContextUpdated,
   Synthetic,
   Shell.Started,

packages/core/src/session/execution.ts

@@ -8,11 +8,16 @@ export interface Interface {
   /** Explicitly drain one Session, making at least one provider attempt. */
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
   /** Schedule a drain after durable work is recorded. Repeated wakeups may coalesce. */
-  readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
+  readonly wake: (sessionID: SessionSchema.ID, seq?: number) => Effect.Effect<void, SessionRunner.RunError>
+  /** Interrupt active work owned by this process. Idle interruption is a no-op. */
+  readonly interrupt: (sessionID: SessionSchema.ID, seq?: number) => Effect.Effect<void>
 }
 
 /** Routes execution from a Session ID to the runner owned by that Session's Location. */
 export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionExecution") {}
 
 /** Low-level compatibility layer for callers that only need durable Session recording. */
-export const noopLayer = Layer.succeed(Service, Service.of({ resume: () => Effect.void, wake: () => Effect.void }))
+export const noopLayer = Layer.succeed(
+  Service,
+  Service.of({ resume: () => Effect.void, wake: () => Effect.void, interrupt: () => Effect.void }),
+)

packages/core/src/session/execution/local.ts

@@ -1,6 +1,7 @@
 import { Effect, Layer } from "effect"
 import { LocationServiceMap } from "../../location-layer"
 import { SessionRunCoordinator } from "../run-coordinator"
+import { SessionRunner } from "../runner"
 import { SessionSchema } from "../schema"
 import { SessionStore } from "../store"
 import { SessionExecution } from "../execution"
@@ -11,25 +12,25 @@ export const layer = Layer.effect(
   Effect.gen(function* () {
     const store = yield* SessionStore.Service
     const locations = yield* LocationServiceMap
-    const scope = yield* Effect.scope
-    const withCoordinator = Effect.fnUntraced(function* <A, E>(
-      sessionID: SessionSchema.ID,
-      use: (coordinator: SessionRunCoordinator.Interface) => Effect.Effect<A, E>,
-    ) {
-      const session = yield* store.get(sessionID)
-      if (!session) return yield* Effect.die(`Session not found: ${sessionID}`)
-      return yield* SessionRunCoordinator.Service.use(use).pipe(Effect.provide(locations.get(session.location)))
+    const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, void, SessionRunner.RunError>({
+      drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, mode) {
+        const session = yield* store.get(sessionID)
+        if (!session) return yield* Effect.die(`Session not found: ${sessionID}`)
+        return yield* SessionRunner.Service.use((runner) => runner.run({ sessionID, force: mode === "run" })).pipe(
+          Effect.provide(locations.get(session.location)),
+        )
+      }),
+      onFailure: (sessionID, cause) =>
+        Effect.logError("Failed to drain Session").pipe(
+          Effect.annotateLogs("sessionID", sessionID),
+          Effect.annotateLogs("cause", cause),
+        ),
     })
 
     return SessionExecution.Service.of({
-      resume: Effect.fn("SessionExecution.resume")(function* (sessionID) {
-        return yield* withCoordinator(sessionID, (coordinator) => coordinator.run(sessionID))
-      }),
-      wake: Effect.fn("SessionExecution.wake")(function* (sessionID) {
-        yield* withCoordinator(sessionID, (coordinator) =>
-          coordinator.wake(sessionID).pipe(Effect.andThen(coordinator.awaitIdle(sessionID))),
-        ).pipe(Effect.forkIn(scope), Effect.asVoid)
-      }),
+      interrupt: coordinator.interrupt,
+      resume: coordinator.run,
+      wake: coordinator.wake,
     })
   }),
 )

packages/core/src/session/message-updater.ts

@@ -159,6 +159,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
       },
       "session.next.prompt.admitted": () => Effect.void,
       "session.next.prompt.promoted": () => Effect.void,
+      "session.next.interrupt.requested": () => Effect.void,
       "session.next.context.updated": (event) =>
         adapter.appendMessage(
           new SessionMessage.System({

packages/core/src/session/projector.ts

@@ -428,6 +428,7 @@ export const layer = Layer.effectDiscard(
         )
       }),
     )
+    yield* events.project(SessionEvent.InterruptRequested, () => Effect.void)
     yield* events.project(SessionEvent.ContextUpdated, (event) => {
       if (!event.replay || event.seq === undefined) return run(db, event)
       return run(db, event).pipe(