refactor(core): remove the unused foreground path from SubagentJob (#63223)

This commit is contained in:
James Long
2026-11-11 13:52:20 -06:00
committed by GitHub
Unverified
parent 66e2fb94b1
commit f5f5190dd0
14 changed files with 22 additions and 81 deletions
+1 -1
View File
@@ -93,7 +93,7 @@ export const Plugin = define({
agent: selected.id,
description: command.description ?? name,
}
yield* subagents.start(recovery, { background: true })
yield* subagents.start(recovery)
return
}
if (agent !== undefined) {
-7
View File
@@ -24,11 +24,6 @@ export interface Interface {
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>
/**
* The durable sequence where the Session's current busy period started, if one is bound.
* Sequence-bounded lookups use it to find only work created by this period.
*/
readonly periodStartSeq: (sessionID: SessionSchema.ID) => Effect.Effect<number | undefined>
/**
* 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.
@@ -182,7 +177,6 @@ export const layer = Layer.effect(
return Service.of({
active: coordinator.active,
isActive: coordinator.isActive,
periodStartSeq: subagentReport.periodStartSeq,
interrupt: (sessionID, options) =>
Effect.gen(function* () {
const interrupted = yield* coordinator.interrupt(sessionID, "user")
@@ -236,7 +230,6 @@ export const noopLayer = Layer.succeed(
isActive: () => Effect.succeed(false),
resume: () => Effect.void,
wake: () => Effect.void,
periodStartSeq: () => Effect.succeed(undefined),
interrupt: () => Effect.succeed(false),
awaitIdle: () => Effect.void,
}),
@@ -53,8 +53,6 @@ export const make = (deps: Deps) => {
periods.delete(sessionID)
}
const periodStartSeq = (sessionID: SessionSchema.ID) => Effect.succeed(periods.get(sessionID)?.startedSeq)
const resolveBody = Effect.fnUntraced(function* (
sessionID: SessionSchema.ID,
outcome: Exclude<Outcome, { readonly type: "interrupted"; readonly reason: "shutdown" }>,
@@ -207,5 +205,5 @@ export const make = (deps: Deps) => {
return { commit, deliver }
})
return { started, clear, periodStartSeq, prepare }
return { started, clear, prepare }
}
+17 -53
View File
@@ -3,45 +3,26 @@ export * as SubagentJob from "./subagent-job.js"
import { Context, Effect, Layer } from "effect"
import { Node } from "@opencode/util/effect/app-node"
import type { LayerNode } from "@opencode/util/effect/layer-node"
import { Bus } from "../bus.js"
import { Database } from "../database/database.js"
import { Job } from "../job.js"
import { SessionBackground } from "./background.js"
import { SessionExecution } from "./execution.js"
import type { SessionSchema } from "./schema.js"
import { SessionStore } from "./store.js"
import { SubagentCompletion } from "./subagent-completion.js"
type Recovery = Extract<Job.Recovery, { kind: "subagent" }>
export type StartOptions = {
readonly background?: boolean
}
export interface Interface {
readonly outstanding: (sessionID: SessionSchema.ID) => Effect.Effect<readonly SessionBackground.OutstandingItem[]>
readonly start: (recovery: Recovery, options?: StartOptions) => Effect.Effect<Job.Info>
readonly start: (recovery: Recovery) => Effect.Effect<Job.Info>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SubagentJob") {}
const make = Effect.gen(function* () {
const execution = yield* SessionExecution.Service
const store = yield* SessionStore.Service
const jobs = yield* Job.Service
const { db } = yield* Database.Service
// Jobs outlive the plugin that launched them. Fork with the global context too, rather
// than retaining the caller's location.
const context = yield* Effect.context<never>()
const outstanding = (sessionID: SessionSchema.ID) =>
Effect.gen(function* () {
return yield* SessionBackground.outstanding(jobs, store, sessionID, yield* execution.active)
})
return Service.of({
outstanding,
start: (recovery, options) =>
start: (recovery) =>
jobs
.start({
id: recovery.childSessionID,
@@ -49,38 +30,21 @@ const make = Effect.gen(function* () {
title: recovery.description,
metadata: {},
recovery,
background: options?.background,
run: Effect.gen(function* () {
const resume = execution.resume(recovery.childSessionID).pipe(
// Only an explicit caller cancel releases the child's claim. A server shutdown
// interrupts this job but must leave the child resumable on the next start.
Effect.onInterrupt(() =>
jobs
.isCancelled(recovery.childSessionID)
.pipe(
Effect.flatMap((cancelled) =>
cancelled ? execution.interrupt(recovery.childSessionID) : Effect.void,
),
background: true,
run: execution.resume(recovery.childSessionID).pipe(
// Only an explicit caller cancel releases the child's claim. A server shutdown
// interrupts this job but must leave the child resumable on the next start.
Effect.onInterrupt(() =>
jobs
.isCancelled(recovery.childSessionID)
.pipe(
Effect.flatMap((cancelled) =>
cancelled ? execution.interrupt(recovery.childSessionID) : Effect.void,
),
),
)
if (options?.background) return yield* resume.pipe(Effect.as(""))
// A joining period already owns its boundary; otherwise the child was admitted but not
// resumed, so the latest durable sequence precedes everything this period will write.
const startedSeq =
(yield* execution.periodStartSeq(recovery.childSessionID)) ??
(yield* Bus.latestSequence(db, recovery.childSessionID))
yield* resume
const assistant = yield* store.lastAssistantAfter({
sessionID: recovery.childSessionID,
afterSeq: startedSeq,
})
return SubagentCompletion.formatBody(
SubagentCompletion.text(assistant),
recovery.childSessionID,
yield* outstanding(recovery.childSessionID),
)
}),
),
),
Effect.as(""),
),
})
.pipe(Effect.setContext(context)),
})
@@ -91,5 +55,5 @@ export const layer = Layer.effect(Service, make)
export const node: LayerNode.Provider<Service, never, typeof Node.tags.values.global> = Node.makeGlobalNode({
service: Service,
layer,
deps: [SessionExecution.node, SessionStore.node, Database.node, Job.node],
deps: [SessionExecution.node, Job.node],
})
+1 -1
View File
@@ -234,7 +234,7 @@ export const Plugin = {
agent: agent.name,
description: input.description,
}
yield* subagents.start(recovery, { background: true })
yield* subagents.start(recovery)
return backgroundResult(child.id, notify)
}).pipe(
Effect.map((output) => ({
-1
View File
@@ -63,7 +63,6 @@ const execution = Layer.succeed(
isActive: () => Effect.succeed(false),
resume: () => Effect.void,
interrupt: () => Effect.succeed(false),
periodStartSeq: () => Effect.succeed(undefined),
wake: (sessionID) =>
Effect.suspend(() => {
wakes.push(sessionID)
-2
View File
@@ -105,7 +105,6 @@ const setup = Effect.fnUntraced(function* (options?: {
}),
awaitIdle: () => Effect.void,
interrupt: () => Effect.succeed(false),
periodStartSeq: () => Effect.succeed(undefined),
wake: (id) =>
Effect.gen(function* () {
const pending = yield* SessionInbox.list(database.db, id)
@@ -638,7 +637,6 @@ describe("Session-owned handles", () => {
}),
wake: coordinator.wake,
awaitIdle: coordinator.awaitIdle,
periodStartSeq: () => Effect.succeed(undefined),
interrupt: (id, options) =>
Effect.sync(() => void interrupts.push({ sessionID: id, options })).pipe(
Effect.andThen(coordinator.interrupt(id)),
@@ -75,7 +75,6 @@ const execution = Layer.succeed(
wakeCalls.push(sessionID)
}),
awaitIdle: () => Effect.void,
periodStartSeq: () => Effect.succeed(undefined),
}),
)
const mediaDirectory = "/media"
@@ -125,7 +125,6 @@ const execution = (llmClient: Layer.Layer<LLMClientService>) =>
wake: coordinator.wake,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
periodStartSeq: () => Effect.succeed(undefined),
})
}),
).pipe(Layer.provide(runnerLayer(llmClient)), Layer.orDie)
@@ -469,7 +469,6 @@ const layer = Layer.unwrap(
wake: coordinator.wake,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
periodStartSeq: () => Effect.succeed(undefined),
})
}),
).pipe(
-1
View File
@@ -50,7 +50,6 @@ const executionLayer = Layer.effect(
resume: coordinator.run,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
periodStartSeq: () => Effect.succeed(undefined),
wake: (sessionID) =>
Effect.sync(() => control.wakes.push(sessionID)).pipe(Effect.andThen(coordinator.wake(sessionID))),
})
-1
View File
@@ -127,7 +127,6 @@ const executionNode = makeGlobalNode({
resume: complete,
wake: () => Effect.void,
interrupt: () => Effect.succeed(false),
periodStartSeq: () => Effect.succeed(undefined),
awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid),
})
}),
+2 -7
View File
@@ -109,7 +109,6 @@ const executionNode = makeGlobalNode({
resume: complete,
wake: () => Effect.void,
interrupt: () => Effect.succeed(false),
periodStartSeq: () => Effect.succeed(undefined),
awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
})
}),
@@ -1077,7 +1076,6 @@ describe("SubagentJob cancellation", () => {
it.live("interrupts the child only for an explicit cancel and preserves shutdown claims", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const store = yield* SessionStore.Service
const scope = yield* Scope.make()
const cancelStarted = yield* Deferred.make<void>()
const shutdownStarted = yield* Deferred.make<void>()
@@ -1100,7 +1098,6 @@ describe("SubagentJob cancellation", () => {
Effect.onInterrupt(() => Effect.sync(() => interrupted.add(sessionID))),
),
wake: () => Effect.void,
periodStartSeq: () => Effect.succeed(0),
interrupt: (sessionID) =>
Effect.sync(() => {
interrupts.push(sessionID)
@@ -1113,9 +1110,7 @@ describe("SubagentJob cancellation", () => {
const context = yield* Layer.buildWithScope(
SubagentJob.layer.pipe(
Layer.provide(Layer.succeed(SessionExecution.Service, execution)),
Layer.provide(Layer.succeed(SessionStore.Service, store)),
Layer.provide(Layer.succeed(Job.Service, jobs)),
Layer.provide(Layer.succeed(Database.Service, database)),
),
scope,
)
@@ -1128,7 +1123,7 @@ describe("SubagentJob cancellation", () => {
description: "review",
})
yield* subagents.start(recovery("ses_job_cancel_parent", cancelChild), { background: true })
yield* subagents.start(recovery("ses_job_cancel_parent", cancelChild))
yield* Deferred.await(cancelStarted)
yield* jobs.cancel(cancelChild)
expect(interrupted.has(cancelChild)).toBe(true)
@@ -1136,7 +1131,7 @@ describe("SubagentJob cancellation", () => {
expect(claimed.has(cancelChild)).toBe(false)
// A server shutdown interrupts the job fiber but must not release the child's claim.
yield* subagents.start(recovery("ses_job_shutdown_parent", shutdownChild), { background: true })
yield* subagents.start(recovery("ses_job_shutdown_parent", shutdownChild))
yield* Deferred.await(shutdownStarted)
yield* Scope.close(scope, Exit.void)
expect(interrupted.has(shutdownChild)).toBe(true)
@@ -49,7 +49,6 @@ it.live("serves turn diffs by user message with range validation", () =>
}),
interrupt: () => Effect.succeed(false),
awaitIdle: () => Effect.void,
periodStartSeq: () => Effect.succeed(undefined),
})
}),
)