Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199export * as SessionExecution from "./execution.js"
import { Cause, Context, Effect, Exit, Layer } from "effect"import { Bus } from "../bus.js"import { Database } from "../database/database.js"import { Job } from "../job.js"import { Instance } from "../instance/service.js"import { makeGlobalNode } from "@opencode/util/effect/app-node"import { SessionEvent } from "./event.js"import { SessionRunCoordinator } from "./run-coordinator.js"import { SessionRunner } from "./runner/index.js"import { SessionSchema } from "./schema.js"import { SessionStore } from "./store.js"import { toSessionError } from "./to-session-error.js"import { UserInterruptedError } from "./error.js"import { SessionInbox } from "./inbox.js"
export interface Interface { /** Snapshots active execution owned by this process. */ readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>> /** Checks process-local ownership, including interruption cleanup and terminal settlement. */ readonly isActive: (sessionID: SessionSchema.ID) => Effect.Effect<boolean> /** Starts execution while idle or joins the active execution. */ readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError> /** Registers newly recorded work. Repeated wakeups may coalesce. */ readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void> /** * Interrupt active work owned by this process. Idle interruption is a no-op. Resolves once * the interruption is accepted; cleanup settles asynchronously in the execution fiber. * Returns whether an active execution was interrupted. Compose with `awaitIdle` when * settlement matters. `awaitSettlement` waits only for the interrupted execution, * rather than fresh work admitted during its cleanup. */ readonly interrupt: ( sessionID: SessionSchema.ID, options?: { readonly resume?: boolean readonly reason?: "user" | "inactivity" readonly awaitSettlement?: boolean }, ) => Effect.Effect<boolean> /** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */ readonly awaitIdle: (sessionID: SessionSchema.ID) => Effect.Effect<void>}
/** Routes execution from a Session ID to its selected instance's runner. */export class Service extends Context.Service<Service, Interface>()("@opencode/SessionExecution") {}
type InterruptReason = "user" | "shutdown" | "inactivity"
export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?: InterruptReason) { if (Exit.isSuccess(exit)) return { type: "succeeded" as const } if (Cause.hasInterrupts(exit.cause)) return { type: "interrupted" as const, reason: reason ?? "shutdown" } const failure = Cause.squash(exit.cause) if (failure instanceof UserInterruptedError) return { type: "interrupted" as const, reason: "user" as const } return { type: "failed" as const, error: toSessionError(failure) }}
/** Process-local execution: drains run in this process using the selected instance. */export const layer = Layer.effect( Service, Effect.gen(function* () { const store = yield* SessionStore.Service const instances = yield* Instance.Service const bus = yield* Bus.Service const jobs = yield* Job.Service const db = (yield* Database.Service).db const reportLifecycle = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) => effect.pipe( Effect.tapCause((cause) => Cause.hasInterruptsOnly(cause) ? Effect.void : Effect.logError("Failed to publish Session execution lifecycle", cause).pipe( Effect.annotateLogs({ sessionID }), ), ), Effect.asVoid, ) // Write-ahead claim: starting records the durable intent that a turn is in flight, in the same // transaction as the started event. Terminals release it — except shutdown interruption, which // preserves the claim so the next server start resumes the turn. A claim that survives with no // terminal is the signature of a process that died without teardown (crash, SIGKILL, eviction); // recovery is a property of the database, never of a shutdown hook that may not run. const claimOnCommit = (sessionID: SessionSchema.ID) => ({ commit: () => store.claim(sessionID), }) const releaseOnCommit = (sessionID: SessionSchema.ID) => ({ commit: () => store.release(sessionID), }) const drain = Effect.fnUntraced(function* ( sessionID: SessionSchema.ID, force: boolean, continuation?: SessionRunner.Continuation, promotable: SessionInbox.Promotable = "input", ): Effect.fn.Return<void, SessionRunner.RunError> { const session = yield* store.get(sessionID) if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) const result = yield* SessionRunner.Service.use((runner) => runner.drain({ sessionID, force, continuation, promotable }), ).pipe( instances.provide(session), Effect.tapCause((cause) => Cause.hasInterruptsOnly(cause) ? Effect.void : Effect.logError("Failed to drain Session", cause).pipe(Effect.annotateLogs({ sessionID })), ), ) return yield* SessionRunner.DrainResult.$match(result, { Complete: () => Effect.void, Moved: (result) => drain(sessionID, false, result.continuation, promotable), Reloaded: (result) => drain(sessionID, result.force, result.continuation, promotable), }) }) const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({ started: (sessionID) => reportLifecycle( sessionID, bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)), ), drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable), // One terminal observation per busy period, covering every coalesced drain. settled: (sessionID, exit, reason) => reportLifecycle( sessionID, Effect.gen(function* () { const outcome = terminal(exit, reason) if (outcome.type === "succeeded") { yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID }, releaseOnCommit(sessionID)) return } if (outcome.type === "interrupted") { // Deliberate stops release the claim; shutdown keeps it for restart continuity. if (outcome.reason !== "shutdown") yield* jobs.cancel(sessionID) yield* bus.publish( SessionEvent.Execution.Interrupted, { sessionID, reason: outcome.reason }, outcome.reason === "shutdown" ? undefined : releaseOnCommit(sessionID), ) return } yield* bus.publish( SessionEvent.Execution.Failed, { sessionID, error: outcome.error, }, releaseOnCommit(sessionID), ) }), ), })
return Service.of({ active: coordinator.active, isActive: coordinator.isActive, interrupt: (sessionID, options) => Effect.gen(function* () { const interrupted = yield* coordinator.interrupt(sessionID, options?.reason ?? "user", options) if (!options?.resume) return interrupted // Resume steering input and between-turn control work from the interrupted // intent. Queued next-turn prompts stay parked: a steer-scoped drain never // promotes them, and a control item behind a queued prompt waits its turn. // Interruption acknowledges before cleanup settles, so this wake usually lands // on the stopping execution's doorbell and starts the successor at settle. // Reading the inbox concurrently with the dying drain is safe: delivery consumes // rows inside uninterruptible publications, so a steer row is either still // promotable here or was fully delivered and needs no resumption. const next = yield* SessionInbox.nextPromotable(db, sessionID, "input") if (next === undefined) return interrupted if (next.delivery === "steer" || next.type === "compaction" || next.type === "move") yield* coordinator.wake(sessionID, "steer") return interrupted }), resume: coordinator.run, wake: coordinator.wake, awaitIdle: coordinator.awaitIdle, }) }),)
export const node = makeGlobalNode({ service: Service, layer, deps: [SessionStore.node, Instance.node, Bus.node, Database.node, Job.node],})
/** Low-level compatibility layer for callers that only need durable Session recording. */export const noopLayer = Layer.succeed( Service, Service.of({ active: Effect.succeed(new Set()), isActive: () => Effect.succeed(false), resume: () => Effect.void, wake: () => Effect.void, interrupt: () => Effect.succeed(false), awaitIdle: () => Effect.void, }),)