luna/oh-my-opencode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev' into fix/cli-run-premature-exit-with-background-tasks

Browse Source
This commit is contained in:
柯杨
2026-04-22 09:45:56 +08:00
parent dd95d59402 e0bcf3e2f9
commit eaa3837d0b
390 changed files with 49705 additions and 41270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/features/background-agent/AGENTS.md
+1 -1
View File
@@ -44,7 +44,7 @@ Both must agree before marking a task complete. Prevents premature completion on
## CONCURRENCY MODEL
- Key format: `{providerID}/{modelID}` (e.g., `anthropic/claude-opus-4-6`)
- Key format: `{providerID}/{modelID}` (e.g., `anthropic/claude-opus-4-7`)
- Default limit: 5 concurrent per key (configurable via `background_task` config)
- FIFO queue: tasks wait in order when slots full
- Slot released on: completion, error, cancellation
src/features/background-agent/compaction-aware-message-resolver.test.ts
+6 -6
View File
@@ -83,7 +83,7 @@ describe("findNearestMessageExcludingCompaction", () => {
// given
const message = {
agent: "sisyphus",
model: { providerID: "anthropic", modelID: "claude-opus-4-6" },
model: { providerID: "anthropic", modelID: "claude-opus-4-7" },
}
writeFileSync(join(tempDir, "001.json"), JSON.stringify(message))
@@ -94,18 +94,18 @@ describe("findNearestMessageExcludingCompaction", () => {
expect(result).not.toBeNull()
expect(result?.agent).toBe("sisyphus")
expect(result?.model?.providerID).toBe("anthropic")
expect(result?.model?.modelID).toBe("claude-opus-4-6")
expect(result?.model?.modelID).toBe("claude-opus-4-7")
})
test("skips compaction agent messages", () => {
// given
const compactionMessage = {
agent: "compaction",
model: { providerID: "anthropic", modelID: "claude-opus-4-6" },
model: { providerID: "anthropic", modelID: "claude-opus-4-7" },
}
const validMessage = {
agent: "sisyphus",
model: { providerID: "anthropic", modelID: "claude-opus-4-6" },
model: { providerID: "anthropic", modelID: "claude-opus-4-7" },
}
writeFileSync(join(tempDir, "002.json"), JSON.stringify(compactionMessage))
writeFileSync(join(tempDir, "001.json"), JSON.stringify(validMessage))
@@ -125,12 +125,12 @@ describe("findNearestMessageExcludingCompaction", () => {
writeFileSync(join(tempDir, "002.json"), JSON.stringify({
id: compactionMessageID,
agent: "atlas",
model: { providerID: "anthropic", modelID: "claude-opus-4-6" },
model: { providerID: "anthropic", modelID: "claude-opus-4-7" },
}))
writeFileSync(join(tempDir, "001.json"), JSON.stringify({
id: "msg_001",
agent: "sisyphus",
model: { providerID: "anthropic", modelID: "claude-opus-4-6" },
model: { providerID: "anthropic", modelID: "claude-opus-4-7" },
}))
mkdirSync(partDir, { recursive: true })
writeFileSync(join(partDir, "prt_0001.json"), JSON.stringify({ type: "compaction" }))
src/features/background-agent/concurrency.test.ts
+1 -1
View File
@@ -94,7 +94,7 @@ describe("ConcurrencyManager.getConcurrencyLimit", () => {
// when
const modelLimit = manager.getConcurrencyLimit("anthropic/claude-sonnet-4-6")
const providerLimit = manager.getConcurrencyLimit("anthropic/claude-opus-4-6")
const providerLimit = manager.getConcurrencyLimit("anthropic/claude-opus-4-7")
const defaultLimit = manager.getConcurrencyLimit("google/gemini-3.1-pro")
// then
src/features/background-agent/index.ts
+2
View File
@@ -1,2 +1,4 @@
export * from "./types"
export { BackgroundManager, type SubagentSessionCreatedEvent, type OnSubagentSessionCreated } from "./manager"
export { waitForTaskSessionID } from "./wait-for-task-session"
export type { WaitForTaskSessionIDOptions } from "./wait-for-task-session"
src/features/background-agent/manager.test.ts
+43 -48
View File
@@ -855,7 +855,7 @@ describe("BackgroundManager.notifyParentSession - dynamic message lookup", () =>
{
info: {
agent: "sisyphus",
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
},
},
{
@@ -890,7 +890,7 @@ describe("BackgroundManager.notifyParentSession - dynamic message lookup", () =>
//#then
expect(capturedBody?.agent).toBe("sisyphus")
expect(capturedBody?.model).toEqual({ providerID: "anthropic", modelID: "claude-opus-4.6" })
expect(capturedBody?.model).toEqual({ providerID: "anthropic", modelID: "claude-opus-4.7" })
manager.shutdown()
})
@@ -913,7 +913,7 @@ describe("BackgroundManager.notifyParentSession - dynamic message lookup", () =>
}
const currentMessage: CurrentMessage = {
agent: "sisyphus",
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
}
// when
@@ -921,7 +921,7 @@ describe("BackgroundManager.notifyParentSession - dynamic message lookup", () =>
// then - uses currentMessage values, not task.parentModel/parentAgent
expect(promptBody.agent).toBe("sisyphus")
expect(promptBody.model).toEqual({ providerID: "anthropic", modelID: "claude-opus-4.6" })
expect(promptBody.model).toEqual({ providerID: "anthropic", modelID: "claude-opus-4.7" })
})
test("should fallback to parentAgent when currentMessage.agent is undefined", async () => {
@@ -1155,7 +1155,7 @@ describe("BackgroundManager.notifyParentSession - notifications toggle", () => {
agent: "explore",
model: {
providerID: "anthropic",
modelID: "claude-opus-4.6",
modelID: "claude-opus-4.7",
variant: "high",
},
},
@@ -1211,7 +1211,7 @@ describe("BackgroundManager.notifyParentSession - variant propagation", () => {
agent: "explore",
model: {
providerID: "anthropic",
modelID: "claude-opus-4.6",
modelID: "claude-opus-4.7",
variant: "max",
},
},
@@ -1231,7 +1231,7 @@ describe("BackgroundManager.notifyParentSession - variant propagation", () => {
status: "completed",
startedAt: new Date(),
completedAt: new Date(),
model: { providerID: "anthropic", modelID: "claude-opus-4.6", variant: "high" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7", variant: "high" },
}
getPendingByParent(manager).set("session-parent", new Set([task.id]))
@@ -1272,7 +1272,7 @@ describe("BackgroundManager.notifyParentSession - variant propagation", () => {
status: "completed",
startedAt: new Date(),
completedAt: new Date(),
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
}
getPendingByParent(manager).set("session-parent", new Set([task.id]))
@@ -1349,7 +1349,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
test("should release concurrency and clear key on completion", async () => {
// given
const concurrencyKey = "anthropic/claude-opus-4.6"
const concurrencyKey = "anthropic/claude-opus-4.7"
const concurrencyManager = getConcurrencyManager(manager)
await concurrencyManager.acquire(concurrencyKey)
@@ -1378,7 +1378,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
test("should prevent double completion and double release", async () => {
// given
const concurrencyKey = "anthropic/claude-opus-4.6"
const concurrencyKey = "anthropic/claude-opus-4.7"
const concurrencyManager = getConcurrencyManager(manager)
await concurrencyManager.acquire(concurrencyKey)
@@ -1508,7 +1508,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
test("should release task concurrencyKey when startTask throws after assigning it", async () => {
// given
const concurrencyKey = "anthropic/claude-opus-4.6"
const concurrencyKey = "anthropic/claude-opus-4.7"
const concurrencyManager = getConcurrencyManager(manager)
const task = createMockTask({
@@ -1524,7 +1524,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
agent: task.agent,
parentSessionID: task.parentSessionID,
parentMessageID: task.parentMessageID,
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
}
getTaskMap(manager).set(task.id, task)
getQueuesByKey(manager).set(concurrencyKey, [{ task, input }])
@@ -1544,7 +1544,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
test("should mark task as error when startTask throws after session creation", async () => {
//#given - startTask creates session but fails before sending prompt
const concurrencyKey = "anthropic/claude-opus-4.6"
const concurrencyKey = "anthropic/claude-opus-4.7"
const task = createMockTask({
id: "task-zombie-session",
@@ -1561,7 +1561,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
agent: task.agent,
parentSessionID: task.parentSessionID,
parentMessageID: task.parentMessageID,
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
}
getTaskMap(manager).set(task.id, task)
getQueuesByKey(manager).set(concurrencyKey, [{ task, input }])
@@ -1585,7 +1585,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
test("should release queue slot when queued task is already interrupt", async () => {
// given
const concurrencyKey = "anthropic/claude-opus-4.6"
const concurrencyKey = "anthropic/claude-opus-4.7"
const concurrencyManager = getConcurrencyManager(manager)
const task = createMockTask({
@@ -1601,7 +1601,7 @@ describe("BackgroundManager.tryCompleteTask", () => {
agent: task.agent,
parentSessionID: task.parentSessionID,
parentMessageID: task.parentMessageID,
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
}
getTaskMap(manager).set(task.id, task)
getQueuesByKey(manager).set(concurrencyKey, [{ task, input }])
@@ -2104,7 +2104,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
agent: "test-agent",
parentSessionID: "parent-session",
parentMessageID: "parent-message",
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
}
const launchInputWithoutModel = {
description: "Test task without model",
@@ -2124,7 +2124,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
expect(taskWithModel.status).toBe("pending")
expect(taskWithoutModel.status).toBe("pending")
expect(promptBodies).toHaveLength(2)
expect(promptBodies[0].model).toEqual({ providerID: "anthropic", modelID: "claude-opus-4.6" })
expect(promptBodies[0].model).toEqual({ providerID: "anthropic", modelID: "claude-opus-4.7" })
expect(promptBodies[0].agent).toBe("test-agent")
expect(promptBodies[1].agent).toBe("test-agent")
expect("model" in promptBodies[1]).toBe(false)
@@ -2327,7 +2327,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
await expect(result).rejects.toThrow("background_task.maxDepth=3")
})
test("should block launches when maxDescendants is reached", async () => {
test("allows multiple descendants without a root spawn cap", async () => {
// given
manager.shutdown()
manager = new BackgroundManager(
@@ -2337,7 +2337,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}),
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 1 },
)
const input = {
@@ -2354,10 +2353,10 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
const result = manager.launch(input)
// then
await expect(result).rejects.toThrow("background_task.maxDescendants=1")
await expect(result).resolves.toBeDefined()
})
test("should consume descendant quota for reserved sync spawns", async () => {
test("allows spawn assertions after reserveSubagentSpawn without a root spawn cap", async () => {
// given
manager.shutdown()
manager = new BackgroundManager(
@@ -2367,7 +2366,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}),
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 1 },
)
await manager.reserveSubagentSpawn("session-root")
@@ -2376,7 +2374,10 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
const result = manager.assertCanSpawn("session-root")
// then
await expect(result).rejects.toThrow("background_task.maxDescendants=1")
await expect(result).resolves.toMatchObject({
rootSessionID: "session-root",
childDepth: 1,
})
})
test("should fail closed when session lineage lookup fails", async () => {
@@ -2392,7 +2393,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
),
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 1 },
)
const input = {
@@ -2407,10 +2407,10 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
const result = manager.launch(input)
// then
await expect(result).rejects.toThrow("background_task.maxDescendants cannot be enforced safely")
await expect(result).rejects.toThrow("background_task.maxDepth cannot be enforced safely")
})
test("should release descendant quota when queued task is cancelled before session starts", async () => {
test("allows replacement launch when a queued task is cancelled before session starts", async () => {
// given
manager.shutdown()
manager = new BackgroundManager(
@@ -2420,7 +2420,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}),
directory: tmpdir(),
} as unknown as PluginInput,
{ defaultConcurrency: 1, maxDescendants: 2 },
{ defaultConcurrency: 1 },
)
const input = {
@@ -2445,7 +2445,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
expect(replacementTask.status).toBe("pending")
})
test("should release descendant quota when session creation fails before session starts", async () => {
test("allows retry after session creation fails before session starts", async () => {
// given
let createAttempts = 0
manager.shutdown()
@@ -2472,7 +2472,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
},
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 1 },
)
const input = {
@@ -2887,7 +2886,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}
})
test("should release descendant quota when task completes", async () => {
test("allows relaunch after task completes", async () => {
manager.shutdown()
manager = new BackgroundManager(
{
@@ -2896,7 +2895,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}),
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 1 },
)
stubNotifyParentSession(manager)
@@ -2920,7 +2918,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
await expect(manager.launch(input)).resolves.toBeDefined()
})
test("should release descendant quota when running task is cancelled", async () => {
test("allows relaunch after running task is cancelled", async () => {
manager.shutdown()
manager = new BackgroundManager(
{
@@ -2929,7 +2927,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}),
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 1 },
)
const input = {
@@ -2950,7 +2947,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
await expect(manager.launch(input)).resolves.toBeDefined()
})
test("should release descendant quota when task errors", async () => {
test("allows relaunch after task errors", async () => {
manager.shutdown()
manager = new BackgroundManager(
{
@@ -2959,7 +2956,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}),
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 1 },
)
const input = {
@@ -2984,7 +2980,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
await expect(manager.launch(input)).resolves.toBeDefined()
})
test("should not double-decrement quota when pending task is cancelled", async () => {
test("allows repeated relaunch after pending tasks are cancelled", async () => {
manager.shutdown()
manager = new BackgroundManager(
{
@@ -2993,7 +2989,6 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
}),
directory: tmpdir(),
} as unknown as PluginInput,
{ maxDescendants: 2 },
)
const input = {
@@ -3250,7 +3245,7 @@ describe("BackgroundManager - Non-blocking Queue Integration", () => {
description: "Task 1",
prompt: "Do something",
agent: "test-agent",
model: { providerID: "anthropic", modelID: "claude-opus-4.6" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7" },
parentSessionID: "parent-session",
parentMessageID: "parent-message",
}
@@ -4230,7 +4225,7 @@ describe("BackgroundManager.handleEvent - session.deleted cascade", () => {
describe("BackgroundManager.handleEvent - session.error", () => {
const defaultRetryFallbackChain = [
{ providers: ["anthropic"], model: "claude-opus-4-6", variant: "max" },
{ providers: ["anthropic"], model: "claude-opus-4-7", variant: "max" },
{ providers: ["anthropic"], model: "gpt-5.3-codex", variant: "high" },
]
@@ -4254,7 +4249,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
agent: "sisyphus",
status: "running",
concurrencyKey: input.concurrencyKey,
model: { providerID: "anthropic", modelID: "claude-opus-4.6-thinking" },
model: { providerID: "anthropic", modelID: "claude-opus-4.7-thinking" },
fallbackChain: input.fallbackChain ?? defaultRetryFallbackChain,
attemptCount: 0,
})
@@ -4399,7 +4394,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
//#given
const manager = createBackgroundManager()
const concurrencyManager = getConcurrencyManager(manager)
const concurrencyKey = "anthropic/claude-opus-4.6-thinking"
const concurrencyKey = "anthropic/claude-opus-4.7-thinking"
await concurrencyManager.acquire(concurrencyKey)
stubProcessKey(manager)
@@ -4411,7 +4406,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
description: "task that should retry",
concurrencyKey,
fallbackChain: [
{ providers: ["anthropic"], model: "claude-opus-4-6", variant: "max" },
{ providers: ["anthropic"], model: "claude-opus-4-7", variant: "max" },
{ providers: ["anthropic"], model: "claude-opus-4-5", variant: "max" },
],
})
@@ -4425,7 +4420,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
name: "UnknownError",
data: {
message:
"Bad Gateway: {\"error\":{\"message\":\"unknown provider for model claude-opus-4.6-thinking\"}}",
"Bad Gateway: {\"error\":{\"message\":\"unknown provider for model claude-opus-4.7-thinking\"}}",
},
},
},
@@ -4436,7 +4431,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
expect(task.attemptCount).toBe(1)
expect(task.model).toEqual({
providerID: "anthropic",
modelID: "claude-opus-4.6",
modelID: "claude-opus-4.7",
variant: "max",
})
expect(task.concurrencyKey).toBeUndefined()
@@ -4474,7 +4469,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
expect(task.attemptCount).toBe(1)
expect(task.model).toEqual({
providerID: "anthropic",
modelID: "claude-opus-4.6",
modelID: "claude-opus-4.7",
variant: "max",
})
@@ -4502,7 +4497,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
name: "UnknownError",
data: {
message:
"Bad Gateway: {\"error\":{\"message\":\"unknown provider for model claude-opus-4.6-thinking\"}}",
"Bad Gateway: {\"error\":{\"message\":\"unknown provider for model claude-opus-4.7-thinking\"}}",
},
},
}
@@ -4519,7 +4514,7 @@ describe("BackgroundManager.handleEvent - session.error", () => {
expect(task.attemptCount).toBe(1)
expect(task.model).toEqual({
providerID: "anthropic",
modelID: "claude-opus-4.6",
modelID: "claude-opus-4.7",
variant: "max",
})
src/features/background-agent/manager.ts
-12
View File
@@ -73,8 +73,6 @@ import {
} from "./loop-detector"
import {
createSubagentDepthLimitError,
createSubagentDescendantLimitError,
getMaxRootSessionSpawnBudget,
getMaxSubagentDepth,
resolveSubagentSpawnContext,
type SubagentSpawnContext,
@@ -219,16 +217,6 @@ export class BackgroundManager {
})
}
const maxRootSessionSpawnBudget = getMaxRootSessionSpawnBudget(this.config)
const descendantCount = this.rootDescendantCounts.get(spawnContext.rootSessionID) ?? 0
if (descendantCount >= maxRootSessionSpawnBudget) {
throw createSubagentDescendantLimitError({
rootSessionID: spawnContext.rootSessionID,
descendantCount,
maxDescendants: maxRootSessionSpawnBudget,
})
}
return spawnContext
}
src/features/background-agent/process-cleanup.test-helpers.ts
+27
View File
@@ -0,0 +1,27 @@
type ProcessCleanupEvent =
| NodeJS.Signals
| "beforeExit"
| "exit"
| "uncaughtException"
| "unhandledRejection"
export function getNewListener(
signal: ProcessCleanupEvent,
existingListeners: Function[],
): () => void {
const listener = process
.listeners(signal)
.find((registeredListener) => !existingListeners.includes(registeredListener))
if (typeof listener !== "function") {
throw new Error(`Expected a ${signal} listener to be registered`)
}
return listener
}
export async function flushMicrotasks(): Promise<void> {
for (let iteration = 0; iteration < 10; iteration += 1) {
await Promise.resolve()
}
}
src/features/background-agent/process-cleanup.test.ts
+116 -37
View File
@@ -1,3 +1,5 @@
/// <reference types="bun-types" />
import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from "bun:test"
import {
@@ -5,42 +7,17 @@ import {
registerManagerForCleanup,
unregisterManagerForCleanup,
} from "./process-cleanup"
import { flushMicrotasks, getNewListener } from "./process-cleanup.test-helpers"
type CleanupManager = {
shutdown: () => void | Promise<void>
}
type ProcessCleanupEvent = NodeJS.Signals | "beforeExit" | "exit"
function getNewListener(
signal: ProcessCleanupEvent,
existingListeners: Function[],
): () => void {
const listener = process
.listeners(signal)
.find((registeredListener) => !existingListeners.includes(registeredListener))
expect(listener).toBeDefined()
if (typeof listener !== "function") {
throw new Error(`Expected a ${signal} listener to be registered`)
}
return listener
}
async function flushMicrotasks(): Promise<void> {
for (let iteration = 0; iteration < 10; iteration += 1) {
await Promise.resolve()
}
}
describe("#given process cleanup registration", () => {
const registeredManagers: CleanupManager[] = []
const originalExitCode = process.exitCode
beforeEach(() => {
process.exitCode = originalExitCode
process.exitCode = 0
registeredManagers.length = 0
_resetForTesting()
})
@@ -50,7 +27,7 @@ describe("#given process cleanup registration", () => {
unregisterManagerForCleanup(manager)
}
process.exitCode = originalExitCode
process.exitCode = 0
_resetForTesting()
})
@@ -92,13 +69,7 @@ describe("#given process cleanup registration", () => {
test("#when cleanup finishes after SIGINT #then the fallback exit timer is cleared", async () => {
const sigintListenersBefore = process.listeners("SIGINT")
const timeoutHandle = setTimeout(() => undefined, 0)
clearTimeout(timeoutHandle)
const setTimeoutImplementation: typeof setTimeout = () => timeoutHandle
const setTimeoutSpy = spyOn(globalThis, "setTimeout").mockImplementation(
setTimeoutImplementation,
)
const setTimeoutSpy = spyOn(globalThis, "setTimeout")
const clearTimeoutSpy = spyOn(globalThis, "clearTimeout")
try {
@@ -117,11 +88,10 @@ describe("#given process cleanup registration", () => {
await flushMicrotasks()
expect(setTimeoutSpy).toHaveBeenCalledTimes(1)
expect(clearTimeoutSpy).toHaveBeenCalledWith(timeoutHandle)
expect(clearTimeoutSpy).toHaveBeenCalledTimes(1)
} finally {
setTimeoutSpy.mockRestore()
clearTimeoutSpy.mockRestore()
clearTimeout(timeoutHandle)
}
})
})
@@ -163,6 +133,32 @@ describe("#given process cleanup registration", () => {
expect(process.listeners("SIGINT")).toHaveLength(sigintListenersAfterFirstRegistration)
})
test("#given two managers registered #when uncaughtException fires #then both shutdowns called", async () => {
const exitSpy = spyOn(process, "exit").mockImplementation((code?: number): never => {
throw new Error(`Unexpected process.exit(${String(code)})`)
})
const shutdownOne = mock(() => {})
const shutdownTwo = mock(() => {})
const managerOne = { shutdown: shutdownOne }
const managerTwo = { shutdown: shutdownTwo }
registeredManagers.push(managerOne, managerTwo)
try {
registerManagerForCleanup(managerOne)
registerManagerForCleanup(managerTwo)
process.emit("uncaughtException", new Error("boom"))
await flushMicrotasks()
expect(shutdownOne).toHaveBeenCalledTimes(1)
expect(shutdownTwo).toHaveBeenCalledTimes(1)
expect(process.exitCode).toBe(1)
expect(exitSpy).not.toHaveBeenCalled()
} finally {
exitSpy.mockRestore()
}
})
})
describe("#given cleanup managers are unregistered", () => {
@@ -202,5 +198,88 @@ describe("#given process cleanup registration", () => {
expect(remainingManagerShutdown).toHaveBeenCalledTimes(1)
expect(removedManagerShutdown).not.toHaveBeenCalled()
})
test("#given uncaughtException handler registered #when manager is unregistered via unregisterManagerForCleanup #then subsequent events do not invoke that manager", () => {
const uncaughtExceptionListenersBefore = process.listeners("uncaughtException")
const shutdown = mock(() => {})
const manager = { shutdown }
registeredManagers.push(manager)
registerManagerForCleanup(manager)
expect(process.listeners("uncaughtException")).toHaveLength(
uncaughtExceptionListenersBefore.length + 1,
)
unregisterManagerForCleanup(manager)
registeredManagers.length = 0
process.emit("uncaughtException", new Error("boom"))
expect(shutdown).not.toHaveBeenCalled()
})
})
describe("#given uncaught exception and rejection cleanup", () => {
test("#given manager registered AND process emits uncaughtException #when event fires #then manager.shutdown() called AND process.exitCode set to 1", async () => {
const exitSpy = spyOn(process, "exit").mockImplementation((code?: number): never => {
throw new Error(`Unexpected process.exit(${String(code)})`)
})
const shutdown = mock(() => {})
const manager = { shutdown }
registeredManagers.push(manager)
try {
registerManagerForCleanup(manager)
process.emit("uncaughtException", new Error("boom"))
await flushMicrotasks()
expect(shutdown).toHaveBeenCalledTimes(1)
expect(process.exitCode).toBe(1)
expect(exitSpy).not.toHaveBeenCalled()
} finally {
exitSpy.mockRestore()
}
})
test("#given manager registered AND process emits unhandledRejection #when event fires #then manager.shutdown() called AND process.exitCode set to 1", async () => {
const exitSpy = spyOn(process, "exit").mockImplementation((code?: number): never => {
throw new Error(`Unexpected process.exit(${String(code)})`)
})
const shutdown = mock(() => {})
const manager = { shutdown }
registeredManagers.push(manager)
try {
registerManagerForCleanup(manager)
process.emit("unhandledRejection", new Error("boom"), Promise.resolve())
await flushMicrotasks()
expect(shutdown).toHaveBeenCalledTimes(1)
expect(process.exitCode).toBe(1)
expect(exitSpy).not.toHaveBeenCalled()
} finally {
exitSpy.mockRestore()
}
})
test("#given _resetForTesting() called #when event fires #then no cleanup runs", () => {
const uncaughtExceptionListenersBefore = process.listeners("uncaughtException")
const shutdown = mock(() => {})
const manager = { shutdown }
registerManagerForCleanup(manager)
expect(process.listeners("uncaughtException")).toHaveLength(
uncaughtExceptionListenersBefore.length + 1,
)
_resetForTesting()
process.emit("uncaughtException", new Error("boom"))
expect(shutdown).not.toHaveBeenCalled()
expect(process.listeners("uncaughtException")).toHaveLength(
uncaughtExceptionListenersBefore.length,
)
})
})
})
src/features/background-agent/process-cleanup.ts
+42 -14
View File
@@ -1,33 +1,51 @@
import { log } from "../../shared"
type ProcessCleanupEvent = NodeJS.Signals | "beforeExit" | "exit"
type ProcessCleanupSignal = NodeJS.Signals | "beforeExit" | "exit"
type ProcessCleanupErrorEvent = "uncaughtException" | "unhandledRejection"
function scheduleForcedExit(cleanupResult: void | Promise<void>, exitCode: number): void {
process.exitCode = exitCode
const exitTimeout = setTimeout(() => process.exit(), 6000)
void Promise.resolve(cleanupResult).finally(() => {
clearTimeout(exitTimeout)
})
}
function registerProcessSignal(
signal: ProcessCleanupEvent,
signal: ProcessCleanupSignal,
handler: () => void | Promise<void>,
exitAfter: boolean
): () => void {
const listener = () => {
const cleanupResult = handler()
if (exitAfter) {
process.exitCode = 0
const exitTimeout = setTimeout(() => process.exit(), 6000)
void Promise.resolve(cleanupResult).finally(() => {
clearTimeout(exitTimeout)
})
scheduleForcedExit(cleanupResult, 0)
}
}
process.on(signal, listener)
return listener
}
function registerErrorEvent(
signal: ProcessCleanupErrorEvent,
handler: (error: unknown) => void | Promise<void>
): (error: unknown) => void {
const listener = (error: unknown) => {
log(`[background-agent] ${signal} received during shutdown cleanup:`, error)
scheduleForcedExit(handler(error), 1)
}
process.on(signal, listener)
return listener
}
interface CleanupTarget {
shutdown(): void | Promise<void>
}
const cleanupManagers = new Set<CleanupTarget>()
let cleanupRegistered = false
const cleanupHandlers = new Map<ProcessCleanupEvent, () => void>()
const cleanupSignalHandlers = new Map<ProcessCleanupSignal, () => void>()
const cleanupErrorHandlers = new Map<ProcessCleanupErrorEvent, (error: unknown) => void>()
export function registerManagerForCleanup(manager: CleanupTarget): void {
cleanupManagers.add(manager)
@@ -59,9 +77,9 @@ export function registerManagerForCleanup(manager: CleanupTarget): void {
return cleanupPromise
}
const registerSignal = (signal: ProcessCleanupEvent, exitAfter: boolean): void => {
const registerSignal = (signal: ProcessCleanupSignal, exitAfter: boolean): void => {
const listener = registerProcessSignal(signal, cleanupAll, exitAfter)
cleanupHandlers.set(signal, listener)
cleanupSignalHandlers.set(signal, listener)
}
registerSignal("SIGINT", true)
@@ -71,6 +89,8 @@ export function registerManagerForCleanup(manager: CleanupTarget): void {
}
registerSignal("beforeExit", false)
registerSignal("exit", false)
cleanupErrorHandlers.set("uncaughtException", registerErrorEvent("uncaughtException", cleanupAll))
cleanupErrorHandlers.set("unhandledRejection", registerErrorEvent("unhandledRejection", cleanupAll))
}
export function unregisterManagerForCleanup(manager: CleanupTarget): void {
@@ -78,10 +98,14 @@ export function unregisterManagerForCleanup(manager: CleanupTarget): void {
if (cleanupManagers.size > 0) return
for (const [signal, listener] of cleanupHandlers.entries()) {
for (const [signal, listener] of cleanupSignalHandlers.entries()) {
process.off(signal, listener)
}
cleanupHandlers.clear()
for (const [signal, listener] of cleanupErrorHandlers.entries()) {
process.off(signal, listener)
}
cleanupSignalHandlers.clear()
cleanupErrorHandlers.clear()
cleanupRegistered = false
}
@@ -90,9 +114,13 @@ export function _resetForTesting(): void {
for (const manager of [...cleanupManagers]) {
cleanupManagers.delete(manager)
}
for (const [signal, listener] of cleanupHandlers.entries()) {
for (const [signal, listener] of cleanupSignalHandlers.entries()) {
process.off(signal, listener)
}
cleanupHandlers.clear()
for (const [signal, listener] of cleanupErrorHandlers.entries()) {
process.off(signal, listener)
}
cleanupSignalHandlers.clear()
cleanupErrorHandlers.clear()
cleanupRegistered = false
}
src/features/background-agent/spawner.test.ts
+81
View File
@@ -577,3 +577,84 @@ describe("background-agent spawner fallback model promotion", () => {
expect(promptCalls[0]?.body?.agent).toBe("sisyphus-junior")
})
})
describe("background-agent spawner tmux callback ordering", () => {
test("fires promptAsync before tmux callback resolves (no blocking)", async () => {
//#given
const events: string[] = []
let resolveTmuxCallback: () => void = () => {}
const tmuxCallbackPromise = new Promise<void>((resolve) => {
resolveTmuxCallback = resolve
})
const client = {
session: {
get: async () => ({ data: { directory: "/tmp/test" } }),
create: async () => {
events.push("session.create")
return { data: { id: "ses_blocking_tmux" } }
},
promptAsync: async () => {
events.push("promptAsync")
return { data: {} }
},
},
} as any
const onSubagentSessionCreated = mock(async () => {
events.push("tmux.callback.start")
await tmuxCallbackPromise
events.push("tmux.callback.end")
})
const task = createTask({
description: "Blocking tmux test",
prompt: "Do work",
agent: "general",
parentSessionID: "ses_parent",
parentMessageID: "msg_parent",
})
const item = {
task,
input: {
description: task.description,
prompt: task.prompt,
agent: task.agent,
parentSessionID: task.parentSessionID,
parentMessageID: task.parentMessageID,
},
}
const ctx = {
client,
directory: "/tmp/test",
concurrencyManager: { release: () => {} },
tmuxEnabled: true,
onSubagentSessionCreated,
onTaskError: () => {},
}
const originalTmux = process.env.TMUX
process.env.TMUX = "/tmp/fake-tmux-socket"
try {
//#when
await startTask(item as any, ctx as any)
await new Promise((resolve) => setTimeout(resolve, 20))
//#then
expect(events).toContain("session.create")
expect(events).toContain("promptAsync")
expect(events).toContain("tmux.callback.start")
const promptIdx = events.indexOf("promptAsync")
const tmuxStartIdx = events.indexOf("tmux.callback.start")
expect(promptIdx < tmuxStartIdx).toBe(true)
expect(events).not.toContain("tmux.callback.end")
} finally {
resolveTmuxCallback()
if (originalTmux === undefined) delete process.env.TMUX
else process.env.TMUX = originalTmux
}
})
})
src/features/background-agent/spawner.ts
+25 -25
View File
@@ -1,6 +1,5 @@
import type { BackgroundTask, LaunchInput, ResumeInput } from "./types"
import type { OpencodeClient, OnSubagentSessionCreated, QueueItem } from "./constants"
import { TMUX_CALLBACK_DELAY_MS } from "./constants"
import { log, getAgentToolRestrictions, promptWithModelSuggestionRetry, createInternalAgentTextPart } from "../../shared"
import { applySessionPromptParams } from "../../shared/session-prompt-params-helpers"
import { subagentSessions } from "../claude-code-session-state"
@@ -115,29 +114,6 @@ export async function startTask(
const sessionID = createResult.data.id
subagentSessions.add(sessionID)
log("[background-agent] tmux callback check", {
hasCallback: !!onSubagentSessionCreated,
tmuxEnabled,
isInsideTmux: isInsideTmux(),
sessionID,
parentID: input.parentSessionID,
})
if (onSubagentSessionCreated && tmuxEnabled && isInsideTmux()) {
log("[background-agent] Invoking tmux callback NOW", { sessionID })
await onSubagentSessionCreated({
sessionID,
parentID: input.parentSessionID,
title: input.description,
}).catch((err) => {
log("[background-agent] Failed to spawn tmux pane:", err)
})
log("[background-agent] tmux callback completed, waiting")
await new Promise(r => setTimeout(r, TMUX_CALLBACK_DELAY_MS))
} else {
log("[background-agent] SKIP tmux callback - conditions not met")
}
task.status = "running"
task.startedAt = new Date()
task.sessionID = sessionID
@@ -188,7 +164,8 @@ export async function startTask(
parts: [createInternalAgentTextPart(input.prompt)],
}
promptWithModelSuggestionRetry(client, {
// Must fire BEFORE tmux callback: attach client needs session activity to render TUI.
const promptChain = promptWithModelSuggestionRetry(client, {
path: { id: sessionID },
body: promptBody,
}).catch(async (error) => {
@@ -214,6 +191,29 @@ export async function startTask(
log("[background-agent] promptAsync error:", error)
onTaskError(task, error instanceof Error ? error : new Error(String(error)))
})
void promptChain
log("[background-agent] tmux callback check", {
hasCallback: !!onSubagentSessionCreated,
tmuxEnabled,
isInsideTmux: isInsideTmux(),
sessionID,
parentID: input.parentSessionID,
})
if (onSubagentSessionCreated && tmuxEnabled && isInsideTmux()) {
log("[background-agent] Invoking tmux callback (fire-and-forget)", { sessionID })
void onSubagentSessionCreated({
sessionID,
parentID: input.parentSessionID,
title: input.description,
}).catch((err) => {
log("[background-agent] Failed to spawn tmux pane:", err)
})
} else {
log("[background-agent] SKIP tmux callback - conditions not met")
}
}
export async function resumeTask(
src/features/background-agent/subagent-spawn-limits.test.ts
+2 -34
View File
@@ -5,9 +5,6 @@ import {
getMaxSubagentDepth,
DEFAULT_MAX_SUBAGENT_DEPTH,
createSubagentDepthLimitError,
createSubagentDescendantLimitError,
getMaxRootSessionSpawnBudget,
DEFAULT_MAX_ROOT_SESSION_SPAWN_BUDGET,
} from "./subagent-spawn-limits"
function createMockClient(sessionGet: OpencodeClient["session"]["get"]): OpencodeClient {
@@ -62,7 +59,7 @@ describe("resolveSubagentSpawnContext", () => {
const result = resolveSubagentSpawnContext(client, "parent-session")
// then
await expect(result).rejects.toThrow(/background_task\.maxDescendants cannot be enforced safely.*lookup failed/)
await expect(result).rejects.toThrow(/background_task\.maxDepth cannot be enforced safely.*lookup failed/)
})
})
@@ -77,7 +74,7 @@ describe("resolveSubagentSpawnContext", () => {
const result = resolveSubagentSpawnContext(client, "parent-session")
// then
await expect(result).rejects.toThrow(/background_task\.maxDescendants cannot be enforced safely.*No session data returned/)
await expect(result).rejects.toThrow(/background_task\.maxDepth cannot be enforced safely.*No session data returned/)
})
})
@@ -209,20 +206,6 @@ describe("getMaxSubagentDepth", () => {
})
})
describe("getMaxRootSessionSpawnBudget", () => {
test("returns DEFAULT_MAX_ROOT_SESSION_SPAWN_BUDGET when no config", () => {
expect(getMaxRootSessionSpawnBudget()).toBe(DEFAULT_MAX_ROOT_SESSION_SPAWN_BUDGET)
})
test("returns config.maxDescendants when provided", () => {
expect(getMaxRootSessionSpawnBudget({ maxDescendants: 10 })).toBe(10)
})
test("default is 50", () => {
expect(DEFAULT_MAX_ROOT_SESSION_SPAWN_BUDGET).toBe(50)
})
})
describe("createSubagentDepthLimitError", () => {
test("includes childDepth, maxDepth, and session IDs in message", () => {
const error = createSubagentDepthLimitError({
@@ -239,18 +222,3 @@ describe("createSubagentDepthLimitError", () => {
expect(error.message).toContain("spawn blocked")
})
})
describe("createSubagentDescendantLimitError", () => {
test("includes descendant count, max, and root session ID", () => {
const error = createSubagentDescendantLimitError({
rootSessionID: "root-789",
descendantCount: 50,
maxDescendants: 50,
})
expect(error.message).toContain("root-789")
expect(error.message).toContain("50")
expect(error.message).toContain("maxDescendants=50")
expect(error.message).toContain("spawn blocked")
})
})
src/features/background-agent/subagent-spawn-limits.ts
+1 -17
View File
@@ -2,7 +2,6 @@ import type { BackgroundTaskConfig } from "../../config/schema"
import type { OpencodeClient } from "./constants"
export const DEFAULT_MAX_SUBAGENT_DEPTH = 3
export const DEFAULT_MAX_ROOT_SESSION_SPAWN_BUDGET = 50
export interface SubagentSpawnContext {
rootSessionID: string
@@ -14,10 +13,6 @@ export function getMaxSubagentDepth(config?: BackgroundTaskConfig): number {
return config?.maxDepth ?? DEFAULT_MAX_SUBAGENT_DEPTH
}
export function getMaxRootSessionSpawnBudget(config?: BackgroundTaskConfig): number {
return config?.maxDescendants ?? DEFAULT_MAX_ROOT_SESSION_SPAWN_BUDGET
}
export async function resolveSubagentSpawnContext(
client: OpencodeClient,
parentSessionID: string,
@@ -53,7 +48,7 @@ export async function resolveSubagentSpawnContext(
} catch (error) {
const reason = error instanceof Error ? error.message : String(error)
throw new Error(
`Subagent spawn blocked: failed to resolve session lineage for ${parentSessionID}, so background_task.maxDescendants cannot be enforced safely. ${reason}`
`Subagent spawn blocked: failed to resolve session lineage for ${parentSessionID}, so background_task.maxDepth cannot be enforced safely. ${reason}`
)
}
@@ -84,14 +79,3 @@ export function createSubagentDepthLimitError(input: {
`Subagent spawn blocked: child depth ${childDepth} exceeds background_task.maxDepth=${maxDepth}. Parent session: ${parentSessionID}. Root session: ${rootSessionID}. Continue in an existing subagent session instead of spawning another.`
)
}
export function createSubagentDescendantLimitError(input: {
rootSessionID: string
descendantCount: number
maxDescendants: number
}): Error {
const { rootSessionID, descendantCount, maxDescendants } = input
return new Error(
`Subagent spawn blocked: root session ${rootSessionID} already has ${descendantCount} descendants, which meets background_task.maxDescendants=${maxDescendants}. Reuse an existing session instead of spawning another.`
)
}
src/features/background-agent/task-poller.test.ts
+2 -2
View File
@@ -648,7 +648,7 @@ describe("checkAndInterruptStaleTasks", () => {
const task = createRunningTask({
startedAt: new Date(Date.now() - 15 * 60 * 1000),
progress: undefined,
concurrencyKey: "anthropic/claude-opus-4-6",
concurrencyKey: "anthropic/claude-opus-4-7",
})
//#when
@@ -661,7 +661,7 @@ describe("checkAndInterruptStaleTasks", () => {
})
//#then
expect(releaseMock).toHaveBeenCalledWith("anthropic/claude-opus-4-6")
expect(releaseMock).toHaveBeenCalledWith("anthropic/claude-opus-4-7")
expect(task.concurrencyKey).toBeUndefined()
})
src/features/background-agent/wait-for-task-session.test.ts
+95
View File
@@ -0,0 +1,95 @@
import { describe, expect, test } from "bun:test"
import type { BackgroundTaskStatus } from "./types"
import { waitForTaskSessionID } from "./wait-for-task-session"
interface TaskSnapshot {
sessionID?: string
status?: BackgroundTaskStatus
}
function createManager(responses: TaskSnapshot[]) {
let index = 0
return {
getTask(_taskID: string): TaskSnapshot {
const response = responses[Math.min(index, responses.length - 1)]
index += 1
return response
},
}
}
describe("waitForTaskSessionID", () => {
test("#given task already has a session id #when waiting #then it returns immediately", async () => {
// given
const manager = createManager([{ sessionID: "ses_ready_123", status: "running" }])
// when
const sessionID = await waitForTaskSessionID(manager, "bg_ready")
// then
expect(sessionID).toBe("ses_ready_123")
})
test("#given session appears later #when waiting #then it polls until resolved", async () => {
// given
const manager = createManager([
{ status: "running" },
{ status: "running" },
{ sessionID: "ses_late_123", status: "running" },
])
// when
const sessionID = await waitForTaskSessionID(manager, "bg_late", {
intervalMs: 1,
timeoutMs: 20,
})
// then
expect(sessionID).toBe("ses_late_123")
})
test("#given aborted signal #when waiting #then it returns undefined", async () => {
// given
const controller = new AbortController()
controller.abort()
const manager = createManager([{ status: "running" }])
// when
const sessionID = await waitForTaskSessionID(manager, "bg_abort", {
signal: controller.signal,
})
// then
expect(sessionID).toBeUndefined()
})
test("#given task never resolves #when waiting past timeout #then it returns undefined", async () => {
// given
const manager = createManager([{ status: "running" }, { status: "running" }, { status: "running" }])
// when
const sessionID = await waitForTaskSessionID(manager, "bg_timeout", {
intervalMs: 1,
timeoutMs: 3,
})
// then
expect(sessionID).toBeUndefined()
})
test.each(["error", "cancelled", "interrupt"] satisfies BackgroundTaskStatus[])(
"#given %s task state #when waiting #then it returns undefined",
async (status: BackgroundTaskStatus) => {
// given
const manager = createManager([{ status }])
// when
const sessionID = await waitForTaskSessionID(manager, `bg_${status}`)
// then
expect(sessionID).toBeUndefined()
}
)
})
src/features/background-agent/wait-for-task-session.ts
+68
View File
@@ -0,0 +1,68 @@
import { getTimingConfig } from "../../tools/delegate-task/timing"
import type { BackgroundTaskStatus } from "./types"
type SessionWaitTerminalStatus = Extract<BackgroundTaskStatus, "error" | "cancelled" | "interrupt">
type AbortSignalLike = { aborted: boolean }
interface TaskReader {
getTask(taskID: string): { sessionID?: string; status?: BackgroundTaskStatus } | undefined
}
export interface WaitForTaskSessionIDOptions {
timeoutMs?: number
intervalMs?: number
signal?: AbortSignalLike
}
function isTerminalStatus(status: BackgroundTaskStatus | undefined): status is SessionWaitTerminalStatus {
return status === "error" || status === "cancelled" || status === "interrupt"
}
function waitForInterval(intervalMs: number): Promise<void> {
return new Promise(resolve => {
const scheduler = globalThis as { setTimeout: (handler: () => void, timeout?: number) => unknown }
scheduler.setTimeout(resolve, intervalMs)
})
}
export async function waitForTaskSessionID(
manager: TaskReader,
taskID: string,
options: WaitForTaskSessionIDOptions = {}
): Promise<string | undefined> {
const timing = getTimingConfig()
const timeoutMs = options.timeoutMs ?? timing.WAIT_FOR_SESSION_TIMEOUT_MS
const intervalMs = options.intervalMs ?? timing.WAIT_FOR_SESSION_INTERVAL_MS
if (options.signal?.aborted) {
return undefined
}
const initialTask = manager.getTask(taskID)
if (initialTask?.sessionID) {
return initialTask.sessionID
}
if (isTerminalStatus(initialTask?.status)) {
return undefined
}
const deadline = Date.now() + timeoutMs
while (Date.now() < deadline) {
if (options.signal?.aborted) {
return undefined
}
await waitForInterval(intervalMs)
const task = manager.getTask(taskID)
if (task?.sessionID) {
return task.sessionID
}
if (isTerminalStatus(task?.status)) {
return undefined
}
}
return undefined
}