refactor: remove 16 orphaned files from background-agent
This commit is contained in:
YeonGyu-Kim
parent
d3b79064c6
commit
1fb6a7cc80
@ -1,168 +0,0 @@
|
||||
import { log } from "../../shared"
|
||||
import type { BackgroundTask } from "./types"
|
||||
import { cleanupTaskAfterSessionEnds } from "./session-task-cleanup"
|
||||
import { handleSessionIdleBackgroundEvent } from "./session-idle-event-handler"
|
||||
|
||||
type Event = { type: string; properties?: Record<string, unknown> }
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null
|
||||
}
|
||||
|
||||
function getString(obj: Record<string, unknown>, key: string): string | undefined {
|
||||
const value = obj[key]
|
||||
return typeof value === "string" ? value : undefined
|
||||
}
|
||||
|
||||
export function handleBackgroundEvent(args: {
|
||||
event: Event
|
||||
findBySession: (sessionID: string) => BackgroundTask | undefined
|
||||
getAllDescendantTasks: (sessionID: string) => BackgroundTask[]
|
||||
releaseConcurrencyKey?: (key: string) => void
|
||||
cancelTask: (
|
||||
taskId: string,
|
||||
options: { source: string; reason: string; skipNotification: true }
|
||||
) => Promise<boolean>
|
||||
tryCompleteTask: (task: BackgroundTask, source: string) => Promise<boolean>
|
||||
validateSessionHasOutput: (sessionID: string) => Promise<boolean>
|
||||
checkSessionTodos: (sessionID: string) => Promise<boolean>
|
||||
idleDeferralTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
completionTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
tasks: Map<string, BackgroundTask>
|
||||
cleanupPendingByParent: (task: BackgroundTask) => void
|
||||
clearNotificationsForTask: (taskId: string) => void
|
||||
emitIdleEvent: (sessionID: string) => void
|
||||
}): void {
|
||||
const {
|
||||
event,
|
||||
findBySession,
|
||||
getAllDescendantTasks,
|
||||
releaseConcurrencyKey,
|
||||
cancelTask,
|
||||
tryCompleteTask,
|
||||
validateSessionHasOutput,
|
||||
checkSessionTodos,
|
||||
idleDeferralTimers,
|
||||
completionTimers,
|
||||
tasks,
|
||||
cleanupPendingByParent,
|
||||
clearNotificationsForTask,
|
||||
emitIdleEvent,
|
||||
} = args
|
||||
|
||||
const props = event.properties
|
||||
|
||||
if (event.type === "message.part.updated" || event.type === "message.part.delta") {
|
||||
if (!props || !isRecord(props)) return
|
||||
const sessionID = getString(props, "sessionID")
|
||||
if (!sessionID) return
|
||||
|
||||
const task = findBySession(sessionID)
|
||||
if (!task) return
|
||||
|
||||
const existingTimer = idleDeferralTimers.get(task.id)
|
||||
if (existingTimer) {
|
||||
clearTimeout(existingTimer)
|
||||
idleDeferralTimers.delete(task.id)
|
||||
}
|
||||
|
||||
const type = getString(props, "type")
|
||||
const tool = getString(props, "tool")
|
||||
|
||||
if (!task.progress) {
|
||||
task.progress = { toolCalls: 0, lastUpdate: new Date() }
|
||||
}
|
||||
task.progress.lastUpdate = new Date()
|
||||
|
||||
if (type === "tool" || tool) {
|
||||
task.progress.toolCalls += 1
|
||||
task.progress.lastTool = tool
|
||||
}
|
||||
}
|
||||
|
||||
if (event.type === "session.idle") {
|
||||
if (!props || !isRecord(props)) return
|
||||
handleSessionIdleBackgroundEvent({
|
||||
properties: props,
|
||||
findBySession,
|
||||
idleDeferralTimers,
|
||||
validateSessionHasOutput,
|
||||
checkSessionTodos,
|
||||
tryCompleteTask,
|
||||
emitIdleEvent,
|
||||
})
|
||||
}
|
||||
|
||||
if (event.type === "session.error") {
|
||||
if (!props || !isRecord(props)) return
|
||||
const sessionID = getString(props, "sessionID")
|
||||
if (!sessionID) return
|
||||
|
||||
const task = findBySession(sessionID)
|
||||
if (!task || task.status !== "running") return
|
||||
|
||||
const errorRaw = props["error"]
|
||||
const dataRaw = isRecord(errorRaw) ? errorRaw["data"] : undefined
|
||||
const message =
|
||||
(isRecord(dataRaw) ? getString(dataRaw, "message") : undefined) ??
|
||||
(isRecord(errorRaw) ? getString(errorRaw, "message") : undefined) ??
|
||||
"Session error"
|
||||
|
||||
task.status = "error"
|
||||
task.error = message
|
||||
task.completedAt = new Date()
|
||||
|
||||
cleanupTaskAfterSessionEnds({
|
||||
task,
|
||||
tasks,
|
||||
idleDeferralTimers,
|
||||
completionTimers,
|
||||
cleanupPendingByParent,
|
||||
clearNotificationsForTask,
|
||||
releaseConcurrencyKey,
|
||||
})
|
||||
}
|
||||
|
||||
if (event.type === "session.deleted") {
|
||||
if (!props || !isRecord(props)) return
|
||||
const infoRaw = props["info"]
|
||||
if (!isRecord(infoRaw)) return
|
||||
const sessionID = getString(infoRaw, "id")
|
||||
if (!sessionID) return
|
||||
|
||||
const tasksToCancel = new Map<string, BackgroundTask>()
|
||||
const directTask = findBySession(sessionID)
|
||||
if (directTask) {
|
||||
tasksToCancel.set(directTask.id, directTask)
|
||||
}
|
||||
for (const descendant of getAllDescendantTasks(sessionID)) {
|
||||
tasksToCancel.set(descendant.id, descendant)
|
||||
}
|
||||
if (tasksToCancel.size === 0) return
|
||||
|
||||
for (const task of tasksToCancel.values()) {
|
||||
if (task.status === "running" || task.status === "pending") {
|
||||
void cancelTask(task.id, {
|
||||
source: "session.deleted",
|
||||
reason: "Session deleted",
|
||||
skipNotification: true,
|
||||
}).catch((err) => {
|
||||
log("[background-agent] Failed to cancel task on session.deleted:", {
|
||||
taskId: task.id,
|
||||
error: err,
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
cleanupTaskAfterSessionEnds({
|
||||
task,
|
||||
tasks,
|
||||
idleDeferralTimers,
|
||||
completionTimers,
|
||||
cleanupPendingByParent,
|
||||
clearNotificationsForTask,
|
||||
releaseConcurrencyKey,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,82 +0,0 @@
|
||||
import { log } from "../../shared"
|
||||
|
||||
import type { BackgroundTask, LaunchInput } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
import type { PluginInput } from "@opencode-ai/plugin"
|
||||
|
||||
type QueueItem = { task: BackgroundTask; input: LaunchInput }
|
||||
|
||||
export function shutdownBackgroundManager(args: {
|
||||
shutdownTriggered: { value: boolean }
|
||||
stopPolling: () => void
|
||||
tasks: Map<string, BackgroundTask>
|
||||
client: PluginInput["client"]
|
||||
onShutdown?: () => void
|
||||
concurrencyManager: ConcurrencyManager
|
||||
completionTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
idleDeferralTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
notifications: Map<string, BackgroundTask[]>
|
||||
pendingByParent: Map<string, Set<string>>
|
||||
queuesByKey: Map<string, QueueItem[]>
|
||||
processingKeys: Set<string>
|
||||
unregisterProcessCleanup: () => void
|
||||
}): void {
|
||||
const {
|
||||
shutdownTriggered,
|
||||
stopPolling,
|
||||
tasks,
|
||||
client,
|
||||
onShutdown,
|
||||
concurrencyManager,
|
||||
completionTimers,
|
||||
idleDeferralTimers,
|
||||
notifications,
|
||||
pendingByParent,
|
||||
queuesByKey,
|
||||
processingKeys,
|
||||
unregisterProcessCleanup,
|
||||
} = args
|
||||
|
||||
if (shutdownTriggered.value) return
|
||||
shutdownTriggered.value = true
|
||||
|
||||
log("[background-agent] Shutting down BackgroundManager")
|
||||
stopPolling()
|
||||
|
||||
for (const task of tasks.values()) {
|
||||
if (task.status === "running" && task.sessionID) {
|
||||
client.session.abort({ path: { id: task.sessionID } }).catch(() => {})
|
||||
}
|
||||
}
|
||||
|
||||
if (onShutdown) {
|
||||
try {
|
||||
onShutdown()
|
||||
} catch (error) {
|
||||
log("[background-agent] Error in onShutdown callback:", error)
|
||||
}
|
||||
}
|
||||
|
||||
for (const task of tasks.values()) {
|
||||
if (task.concurrencyKey) {
|
||||
concurrencyManager.release(task.concurrencyKey)
|
||||
task.concurrencyKey = undefined
|
||||
}
|
||||
}
|
||||
|
||||
for (const timer of completionTimers.values()) clearTimeout(timer)
|
||||
completionTimers.clear()
|
||||
|
||||
for (const timer of idleDeferralTimers.values()) clearTimeout(timer)
|
||||
idleDeferralTimers.clear()
|
||||
|
||||
concurrencyManager.clear()
|
||||
tasks.clear()
|
||||
notifications.clear()
|
||||
pendingByParent.clear()
|
||||
queuesByKey.clear()
|
||||
processingKeys.clear()
|
||||
unregisterProcessCleanup()
|
||||
|
||||
log("[background-agent] Shutdown complete")
|
||||
}
|
||||
@ -1,52 +0,0 @@
|
||||
import type { BackgroundTask } from "./types"
|
||||
|
||||
export function markForNotification(
|
||||
notifications: Map<string, BackgroundTask[]>,
|
||||
task: BackgroundTask
|
||||
): void {
|
||||
const queue = notifications.get(task.parentSessionID) ?? []
|
||||
queue.push(task)
|
||||
notifications.set(task.parentSessionID, queue)
|
||||
}
|
||||
|
||||
export function getPendingNotifications(
|
||||
notifications: Map<string, BackgroundTask[]>,
|
||||
sessionID: string
|
||||
): BackgroundTask[] {
|
||||
return notifications.get(sessionID) ?? []
|
||||
}
|
||||
|
||||
export function clearNotifications(
|
||||
notifications: Map<string, BackgroundTask[]>,
|
||||
sessionID: string
|
||||
): void {
|
||||
notifications.delete(sessionID)
|
||||
}
|
||||
|
||||
export function clearNotificationsForTask(
|
||||
notifications: Map<string, BackgroundTask[]>,
|
||||
taskId: string
|
||||
): void {
|
||||
for (const [sessionID, tasks] of notifications.entries()) {
|
||||
const filtered = tasks.filter((t) => t.id !== taskId)
|
||||
if (filtered.length === 0) {
|
||||
notifications.delete(sessionID)
|
||||
} else {
|
||||
notifications.set(sessionID, filtered)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function cleanupPendingByParent(
|
||||
pendingByParent: Map<string, Set<string>>,
|
||||
task: BackgroundTask
|
||||
): void {
|
||||
if (!task.parentSessionID) return
|
||||
const pending = pendingByParent.get(task.parentSessionID)
|
||||
if (!pending) return
|
||||
|
||||
pending.delete(task.id)
|
||||
if (pending.size === 0) {
|
||||
pendingByParent.delete(task.parentSessionID)
|
||||
}
|
||||
}
|
||||
@ -1,193 +0,0 @@
|
||||
import { log, normalizeSDKResponse } from "../../shared"
|
||||
|
||||
import { findNearestMessageWithFields } from "../hook-message-injector"
|
||||
import { getTaskToastManager } from "../task-toast-manager"
|
||||
|
||||
import { TASK_CLEANUP_DELAY_MS } from "./constants"
|
||||
import { formatDuration } from "./format-duration"
|
||||
import { isAbortedSessionError } from "./error-classifier"
|
||||
import { getMessageDir } from "./message-dir"
|
||||
import { buildBackgroundTaskNotificationText } from "./notification-builder"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { OpencodeClient } from "./opencode-client"
|
||||
|
||||
type AgentModel = { providerID: string; modelID: string }
|
||||
|
||||
type MessageInfo = {
|
||||
agent?: string
|
||||
model?: AgentModel
|
||||
providerID?: string
|
||||
modelID?: string
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null
|
||||
}
|
||||
|
||||
function extractMessageInfo(message: unknown): MessageInfo {
|
||||
if (!isRecord(message)) return {}
|
||||
const info = message["info"]
|
||||
if (!isRecord(info)) return {}
|
||||
|
||||
const agent = typeof info["agent"] === "string" ? info["agent"] : undefined
|
||||
const modelObj = info["model"]
|
||||
if (isRecord(modelObj)) {
|
||||
const providerID = modelObj["providerID"]
|
||||
const modelID = modelObj["modelID"]
|
||||
if (typeof providerID === "string" && typeof modelID === "string") {
|
||||
return { agent, model: { providerID, modelID } }
|
||||
}
|
||||
}
|
||||
|
||||
const providerID = info["providerID"]
|
||||
const modelID = info["modelID"]
|
||||
if (typeof providerID === "string" && typeof modelID === "string") {
|
||||
return { agent, model: { providerID, modelID } }
|
||||
}
|
||||
|
||||
return { agent }
|
||||
}
|
||||
|
||||
export async function notifyParentSession(args: {
|
||||
task: BackgroundTask
|
||||
tasks: Map<string, BackgroundTask>
|
||||
pendingByParent: Map<string, Set<string>>
|
||||
completionTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
clearNotificationsForTask: (taskId: string) => void
|
||||
client: OpencodeClient
|
||||
}): Promise<void> {
|
||||
const { task, tasks, pendingByParent, completionTimers, clearNotificationsForTask, client } = args
|
||||
|
||||
const duration = formatDuration(task.startedAt ?? new Date(), task.completedAt)
|
||||
log("[background-agent] notifyParentSession called for task:", task.id)
|
||||
|
||||
const toastManager = getTaskToastManager()
|
||||
if (toastManager) {
|
||||
toastManager.showCompletionToast({
|
||||
id: task.id,
|
||||
description: task.description,
|
||||
duration,
|
||||
})
|
||||
}
|
||||
|
||||
const pendingSet = pendingByParent.get(task.parentSessionID)
|
||||
if (pendingSet) {
|
||||
pendingSet.delete(task.id)
|
||||
if (pendingSet.size === 0) {
|
||||
pendingByParent.delete(task.parentSessionID)
|
||||
}
|
||||
}
|
||||
|
||||
const allComplete = !pendingSet || pendingSet.size === 0
|
||||
const remainingCount = pendingSet?.size ?? 0
|
||||
|
||||
const completedTasks = allComplete
|
||||
? Array.from(tasks.values()).filter(
|
||||
(t) =>
|
||||
t.parentSessionID === task.parentSessionID &&
|
||||
t.status !== "running" &&
|
||||
t.status !== "pending"
|
||||
)
|
||||
: []
|
||||
|
||||
const notification = buildBackgroundTaskNotificationText({
|
||||
task,
|
||||
duration,
|
||||
allComplete,
|
||||
remainingCount,
|
||||
completedTasks,
|
||||
})
|
||||
|
||||
let agent: string | undefined = task.parentAgent
|
||||
let model: AgentModel | undefined
|
||||
|
||||
try {
|
||||
const messagesResp = await client.session.messages({
|
||||
path: { id: task.parentSessionID },
|
||||
})
|
||||
const raw = normalizeSDKResponse(messagesResp, [] as unknown[])
|
||||
const messages = Array.isArray(raw) ? raw : []
|
||||
|
||||
for (let i = messages.length - 1; i >= 0; i--) {
|
||||
const extracted = extractMessageInfo(messages[i])
|
||||
if (extracted.agent || extracted.model) {
|
||||
agent = extracted.agent ?? task.parentAgent
|
||||
model = extracted.model
|
||||
break
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
if (isAbortedSessionError(error)) {
|
||||
log("[background-agent] Parent session aborted, skipping notification:", {
|
||||
taskId: task.id,
|
||||
parentSessionID: task.parentSessionID,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const messageDir = getMessageDir(task.parentSessionID)
|
||||
const currentMessage = messageDir ? findNearestMessageWithFields(messageDir) : null
|
||||
agent = currentMessage?.agent ?? task.parentAgent
|
||||
model =
|
||||
currentMessage?.model?.providerID && currentMessage?.model?.modelID
|
||||
? { providerID: currentMessage.model.providerID, modelID: currentMessage.model.modelID }
|
||||
: undefined
|
||||
}
|
||||
|
||||
log("[background-agent] notifyParentSession context:", {
|
||||
taskId: task.id,
|
||||
resolvedAgent: agent,
|
||||
resolvedModel: model,
|
||||
})
|
||||
|
||||
try {
|
||||
await client.session.promptAsync({
|
||||
path: { id: task.parentSessionID },
|
||||
body: {
|
||||
noReply: !allComplete,
|
||||
...(agent !== undefined ? { agent } : {}),
|
||||
...(model !== undefined ? { model } : {}),
|
||||
...(task.parentTools ? { tools: task.parentTools } : {}),
|
||||
parts: [{ type: "text", text: notification }],
|
||||
},
|
||||
})
|
||||
|
||||
log("[background-agent] Sent notification to parent session:", {
|
||||
taskId: task.id,
|
||||
allComplete,
|
||||
noReply: !allComplete,
|
||||
})
|
||||
} catch (error) {
|
||||
if (isAbortedSessionError(error)) {
|
||||
log("[background-agent] Parent session aborted, skipping notification:", {
|
||||
taskId: task.id,
|
||||
parentSessionID: task.parentSessionID,
|
||||
})
|
||||
return
|
||||
}
|
||||
log("[background-agent] Failed to send notification:", error)
|
||||
}
|
||||
|
||||
if (!allComplete) return
|
||||
|
||||
for (const completedTask of completedTasks) {
|
||||
const taskId = completedTask.id
|
||||
const existingTimer = completionTimers.get(taskId)
|
||||
if (existingTimer) {
|
||||
clearTimeout(existingTimer)
|
||||
completionTimers.delete(taskId)
|
||||
}
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
completionTimers.delete(taskId)
|
||||
if (tasks.has(taskId)) {
|
||||
clearNotificationsForTask(taskId)
|
||||
tasks.delete(taskId)
|
||||
log("[background-agent] Removed completed task from memory:", taskId)
|
||||
}
|
||||
}, TASK_CLEANUP_DELAY_MS)
|
||||
|
||||
completionTimers.set(taskId, timer)
|
||||
}
|
||||
}
|
||||
@ -1,181 +0,0 @@
|
||||
import { log, normalizeSDKResponse } from "../../shared"
|
||||
|
||||
import {
|
||||
MIN_STABILITY_TIME_MS,
|
||||
} from "./constants"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { OpencodeClient } from "./opencode-client"
|
||||
|
||||
type SessionStatusMap = Record<string, { type: string }>
|
||||
|
||||
type MessagePart = {
|
||||
type?: string
|
||||
tool?: string
|
||||
name?: string
|
||||
text?: string
|
||||
}
|
||||
|
||||
type SessionMessage = {
|
||||
info?: { role?: string }
|
||||
parts?: MessagePart[]
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null
|
||||
}
|
||||
|
||||
function asSessionMessages(value: unknown): SessionMessage[] {
|
||||
if (!Array.isArray(value)) return []
|
||||
return value.filter(isRecord) as SessionMessage[]
|
||||
}
|
||||
|
||||
export async function pollRunningTasks(args: {
|
||||
tasks: Iterable<BackgroundTask>
|
||||
client: OpencodeClient
|
||||
pruneStaleTasksAndNotifications: () => void
|
||||
checkAndInterruptStaleTasks: (statuses: Record<string, { type: string }>) => Promise<void>
|
||||
validateSessionHasOutput: (sessionID: string) => Promise<boolean>
|
||||
checkSessionTodos: (sessionID: string) => Promise<boolean>
|
||||
tryCompleteTask: (task: BackgroundTask, source: string) => Promise<boolean>
|
||||
hasRunningTasks: () => boolean
|
||||
stopPolling: () => void
|
||||
}): Promise<void> {
|
||||
const {
|
||||
tasks,
|
||||
client,
|
||||
pruneStaleTasksAndNotifications,
|
||||
checkAndInterruptStaleTasks,
|
||||
validateSessionHasOutput,
|
||||
checkSessionTodos,
|
||||
tryCompleteTask,
|
||||
hasRunningTasks,
|
||||
stopPolling,
|
||||
} = args
|
||||
|
||||
pruneStaleTasksAndNotifications()
|
||||
|
||||
const statusResult = await client.session.status()
|
||||
const allStatuses = normalizeSDKResponse(statusResult, {} as SessionStatusMap)
|
||||
|
||||
await checkAndInterruptStaleTasks(allStatuses)
|
||||
|
||||
for (const task of tasks) {
|
||||
if (task.status !== "running") continue
|
||||
|
||||
const sessionID = task.sessionID
|
||||
if (!sessionID) continue
|
||||
|
||||
try {
|
||||
const sessionStatus = allStatuses[sessionID]
|
||||
if (sessionStatus?.type === "idle") {
|
||||
const hasValidOutput = await validateSessionHasOutput(sessionID)
|
||||
if (!hasValidOutput) {
|
||||
log("[background-agent] Polling idle but no valid output yet, waiting:", task.id)
|
||||
continue
|
||||
}
|
||||
|
||||
if (task.status !== "running") continue
|
||||
|
||||
const hasIncompleteTodos = await checkSessionTodos(sessionID)
|
||||
if (hasIncompleteTodos) {
|
||||
log("[background-agent] Task has incomplete todos via polling, waiting:", task.id)
|
||||
continue
|
||||
}
|
||||
|
||||
await tryCompleteTask(task, "polling (idle status)")
|
||||
continue
|
||||
}
|
||||
|
||||
const messagesResult = await client.session.messages({
|
||||
path: { id: sessionID },
|
||||
})
|
||||
|
||||
if ((messagesResult as { error?: unknown }).error) {
|
||||
continue
|
||||
}
|
||||
|
||||
const messages = asSessionMessages(normalizeSDKResponse(messagesResult, [] as SessionMessage[], {
|
||||
preferResponseOnMissingData: true,
|
||||
}))
|
||||
const assistantMsgs = messages.filter((m) => m.info?.role === "assistant")
|
||||
|
||||
let toolCalls = 0
|
||||
let lastTool: string | undefined
|
||||
let lastMessage: string | undefined
|
||||
|
||||
for (const msg of assistantMsgs) {
|
||||
const parts = msg.parts ?? []
|
||||
for (const part of parts) {
|
||||
if (part.type === "tool_use" || part.tool) {
|
||||
toolCalls += 1
|
||||
lastTool = part.tool || part.name || "unknown"
|
||||
}
|
||||
if (part.type === "text" && part.text) {
|
||||
lastMessage = part.text
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!task.progress) {
|
||||
task.progress = { toolCalls: 0, lastUpdate: new Date() }
|
||||
}
|
||||
task.progress.toolCalls = toolCalls
|
||||
task.progress.lastTool = lastTool
|
||||
task.progress.lastUpdate = new Date()
|
||||
if (lastMessage) {
|
||||
task.progress.lastMessage = lastMessage
|
||||
task.progress.lastMessageAt = new Date()
|
||||
}
|
||||
|
||||
const currentMsgCount = messages.length
|
||||
const startedAt = task.startedAt
|
||||
if (!startedAt) continue
|
||||
|
||||
const elapsedMs = Date.now() - startedAt.getTime()
|
||||
if (elapsedMs >= MIN_STABILITY_TIME_MS) {
|
||||
if (task.lastMsgCount === currentMsgCount) {
|
||||
task.stablePolls = (task.stablePolls ?? 0) + 1
|
||||
if (task.stablePolls >= 3) {
|
||||
const recheckStatus = await client.session.status()
|
||||
const recheckData = normalizeSDKResponse(recheckStatus, {} as SessionStatusMap)
|
||||
const currentStatus = recheckData[sessionID]
|
||||
|
||||
if (currentStatus?.type !== "idle") {
|
||||
log("[background-agent] Stability reached but session not idle, resetting:", {
|
||||
taskId: task.id,
|
||||
sessionStatus: currentStatus?.type ?? "not_in_status",
|
||||
})
|
||||
task.stablePolls = 0
|
||||
continue
|
||||
}
|
||||
|
||||
const hasValidOutput = await validateSessionHasOutput(sessionID)
|
||||
if (!hasValidOutput) {
|
||||
log("[background-agent] Stability reached but no valid output, waiting:", task.id)
|
||||
continue
|
||||
}
|
||||
|
||||
if (task.status !== "running") continue
|
||||
|
||||
const hasIncompleteTodos = await checkSessionTodos(sessionID)
|
||||
if (!hasIncompleteTodos) {
|
||||
await tryCompleteTask(task, "stability detection")
|
||||
continue
|
||||
}
|
||||
}
|
||||
} else {
|
||||
task.stablePolls = 0
|
||||
}
|
||||
}
|
||||
|
||||
task.lastMsgCount = currentMsgCount
|
||||
} catch (error) {
|
||||
log("[background-agent] Poll error for task:", { taskId: task.id, error })
|
||||
}
|
||||
}
|
||||
|
||||
if (!hasRunningTasks()) {
|
||||
stopPolling()
|
||||
}
|
||||
}
|
||||
@ -1,19 +0,0 @@
|
||||
export type ProcessCleanupEvent = NodeJS.Signals | "beforeExit" | "exit"
|
||||
|
||||
export function registerProcessSignal(
|
||||
signal: ProcessCleanupEvent,
|
||||
handler: () => void,
|
||||
exitAfter: boolean
|
||||
): () => void {
|
||||
const listener = () => {
|
||||
handler()
|
||||
if (exitAfter) {
|
||||
// Set exitCode and schedule exit after delay to allow other handlers to complete async cleanup
|
||||
// Use 6s delay to accommodate LSP cleanup (5s timeout + 1s SIGKILL wait)
|
||||
process.exitCode = 0
|
||||
setTimeout(() => process.exit(), 6000)
|
||||
}
|
||||
}
|
||||
process.on(signal, listener)
|
||||
return listener
|
||||
}
|
||||
@ -1,114 +0,0 @@
|
||||
import { log, normalizeSDKResponse } from "../../shared"
|
||||
|
||||
import type { OpencodeClient } from "./opencode-client"
|
||||
|
||||
type Todo = {
|
||||
content: string
|
||||
status: string
|
||||
priority: string
|
||||
id: string
|
||||
}
|
||||
|
||||
type SessionMessage = {
|
||||
info?: { role?: string }
|
||||
parts?: unknown
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null
|
||||
}
|
||||
|
||||
function asSessionMessages(value: unknown): SessionMessage[] {
|
||||
if (!Array.isArray(value)) return []
|
||||
return value as SessionMessage[]
|
||||
}
|
||||
|
||||
function asParts(value: unknown): Array<Record<string, unknown>> {
|
||||
if (!Array.isArray(value)) return []
|
||||
return value.filter(isRecord)
|
||||
}
|
||||
|
||||
function hasNonEmptyText(value: unknown): boolean {
|
||||
return typeof value === "string" && value.trim().length > 0
|
||||
}
|
||||
|
||||
function isToolResultContentNonEmpty(content: unknown): boolean {
|
||||
if (typeof content === "string") return content.trim().length > 0
|
||||
if (Array.isArray(content)) return content.length > 0
|
||||
return false
|
||||
}
|
||||
|
||||
/**
|
||||
* Validates that a session has actual assistant/tool output before marking complete.
|
||||
* Prevents premature completion when session.idle fires before agent responds.
|
||||
*/
|
||||
export async function validateSessionHasOutput(
|
||||
client: OpencodeClient,
|
||||
sessionID: string
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const response = await client.session.messages({
|
||||
path: { id: sessionID },
|
||||
})
|
||||
|
||||
const messages = asSessionMessages(normalizeSDKResponse(response, [] as SessionMessage[], {
|
||||
preferResponseOnMissingData: true,
|
||||
}))
|
||||
|
||||
const hasAssistantOrToolMessage = messages.some(
|
||||
(m) => m.info?.role === "assistant" || m.info?.role === "tool"
|
||||
)
|
||||
if (!hasAssistantOrToolMessage) {
|
||||
log("[background-agent] No assistant/tool messages found in session:", sessionID)
|
||||
return false
|
||||
}
|
||||
|
||||
const hasContent = messages.some((m) => {
|
||||
if (m.info?.role !== "assistant" && m.info?.role !== "tool") return false
|
||||
|
||||
const parts = asParts(m.parts)
|
||||
return parts.some((part) => {
|
||||
const type = part.type
|
||||
if (type === "tool") return true
|
||||
if (type === "text" && hasNonEmptyText(part.text)) return true
|
||||
if (type === "reasoning" && hasNonEmptyText(part.text)) return true
|
||||
if (type === "tool_result" && isToolResultContentNonEmpty(part.content)) return true
|
||||
return false
|
||||
})
|
||||
})
|
||||
|
||||
if (!hasContent) {
|
||||
log("[background-agent] Messages exist but no content found in session:", sessionID)
|
||||
return false
|
||||
}
|
||||
|
||||
return true
|
||||
} catch (error) {
|
||||
log("[background-agent] Error validating session output:", error)
|
||||
// On error, allow completion to proceed (don't block indefinitely)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
export async function checkSessionTodos(
|
||||
client: OpencodeClient,
|
||||
sessionID: string
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const response = await client.session.todo({
|
||||
path: { id: sessionID },
|
||||
})
|
||||
|
||||
const todos = normalizeSDKResponse(response, [] as Todo[], {
|
||||
preferResponseOnMissingData: true,
|
||||
})
|
||||
if (todos.length === 0) return false
|
||||
|
||||
const incomplete = todos.filter(
|
||||
(t) => t.status !== "completed" && t.status !== "cancelled"
|
||||
)
|
||||
return incomplete.length > 0
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
@ -1,77 +0,0 @@
|
||||
import { log } from "../../shared"
|
||||
|
||||
import { TASK_TTL_MS } from "./constants"
|
||||
import { subagentSessions } from "../claude-code-session-state"
|
||||
import { pruneStaleTasksAndNotifications } from "./task-poller"
|
||||
|
||||
import type { BackgroundTask, LaunchInput } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
|
||||
type QueueItem = { task: BackgroundTask; input: LaunchInput }
|
||||
|
||||
export function pruneStaleState(args: {
|
||||
tasks: Map<string, BackgroundTask>
|
||||
notifications: Map<string, BackgroundTask[]>
|
||||
queuesByKey: Map<string, QueueItem[]>
|
||||
concurrencyManager: ConcurrencyManager
|
||||
cleanupPendingByParent: (task: BackgroundTask) => void
|
||||
clearNotificationsForTask: (taskId: string) => void
|
||||
}): void {
|
||||
const {
|
||||
tasks,
|
||||
notifications,
|
||||
queuesByKey,
|
||||
concurrencyManager,
|
||||
cleanupPendingByParent,
|
||||
clearNotificationsForTask,
|
||||
} = args
|
||||
|
||||
pruneStaleTasksAndNotifications({
|
||||
tasks,
|
||||
notifications,
|
||||
onTaskPruned: (taskId, task, errorMessage) => {
|
||||
const wasPending = task.status === "pending"
|
||||
const now = Date.now()
|
||||
const timestamp = task.status === "pending"
|
||||
? task.queuedAt?.getTime()
|
||||
: task.startedAt?.getTime()
|
||||
const age = timestamp ? now - timestamp : TASK_TTL_MS
|
||||
|
||||
log("[background-agent] Pruning stale task:", {
|
||||
taskId,
|
||||
status: task.status,
|
||||
age: Math.round(age / 1000) + "s",
|
||||
})
|
||||
|
||||
task.status = "error"
|
||||
task.error = errorMessage
|
||||
task.completedAt = new Date()
|
||||
if (task.concurrencyKey) {
|
||||
concurrencyManager.release(task.concurrencyKey)
|
||||
task.concurrencyKey = undefined
|
||||
}
|
||||
|
||||
cleanupPendingByParent(task)
|
||||
if (wasPending) {
|
||||
const key = task.model
|
||||
? `${task.model.providerID}/${task.model.modelID}`
|
||||
: task.agent
|
||||
const queue = queuesByKey.get(key)
|
||||
if (queue) {
|
||||
const index = queue.findIndex((item) => item.task.id === taskId)
|
||||
if (index !== -1) {
|
||||
queue.splice(index, 1)
|
||||
if (queue.length === 0) {
|
||||
queuesByKey.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
clearNotificationsForTask(taskId)
|
||||
tasks.delete(taskId)
|
||||
if (task.sessionID) {
|
||||
subagentSessions.delete(task.sessionID)
|
||||
}
|
||||
},
|
||||
})
|
||||
}
|
||||
@ -1,117 +0,0 @@
|
||||
import { log } from "../../shared"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { LaunchInput } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
import type { OpencodeClient } from "./opencode-client"
|
||||
|
||||
type QueueItem = { task: BackgroundTask; input: LaunchInput }
|
||||
|
||||
export async function cancelBackgroundTask(args: {
|
||||
taskId: string
|
||||
options?: {
|
||||
source?: string
|
||||
reason?: string
|
||||
abortSession?: boolean
|
||||
skipNotification?: boolean
|
||||
}
|
||||
tasks: Map<string, BackgroundTask>
|
||||
queuesByKey: Map<string, QueueItem[]>
|
||||
completionTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
idleDeferralTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
concurrencyManager: ConcurrencyManager
|
||||
client: OpencodeClient
|
||||
cleanupPendingByParent: (task: BackgroundTask) => void
|
||||
markForNotification: (task: BackgroundTask) => void
|
||||
notifyParentSession: (task: BackgroundTask) => Promise<void>
|
||||
}): Promise<boolean> {
|
||||
const {
|
||||
taskId,
|
||||
options,
|
||||
tasks,
|
||||
queuesByKey,
|
||||
completionTimers,
|
||||
idleDeferralTimers,
|
||||
concurrencyManager,
|
||||
client,
|
||||
cleanupPendingByParent,
|
||||
markForNotification,
|
||||
notifyParentSession,
|
||||
} = args
|
||||
|
||||
const task = tasks.get(taskId)
|
||||
if (!task || (task.status !== "running" && task.status !== "pending")) {
|
||||
return false
|
||||
}
|
||||
|
||||
const source = options?.source ?? "cancel"
|
||||
const abortSession = options?.abortSession !== false
|
||||
const reason = options?.reason
|
||||
|
||||
if (task.status === "pending") {
|
||||
const key = task.model
|
||||
? `${task.model.providerID}/${task.model.modelID}`
|
||||
: task.agent
|
||||
const queue = queuesByKey.get(key)
|
||||
if (queue) {
|
||||
const index = queue.findIndex((item) => item.task.id === taskId)
|
||||
if (index !== -1) {
|
||||
queue.splice(index, 1)
|
||||
if (queue.length === 0) {
|
||||
queuesByKey.delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
log("[background-agent] Cancelled pending task:", { taskId, key })
|
||||
}
|
||||
|
||||
task.status = "cancelled"
|
||||
task.completedAt = new Date()
|
||||
if (reason) {
|
||||
task.error = reason
|
||||
}
|
||||
|
||||
if (task.concurrencyKey) {
|
||||
concurrencyManager.release(task.concurrencyKey)
|
||||
task.concurrencyKey = undefined
|
||||
}
|
||||
|
||||
const completionTimer = completionTimers.get(task.id)
|
||||
if (completionTimer) {
|
||||
clearTimeout(completionTimer)
|
||||
completionTimers.delete(task.id)
|
||||
}
|
||||
|
||||
const idleTimer = idleDeferralTimers.get(task.id)
|
||||
if (idleTimer) {
|
||||
clearTimeout(idleTimer)
|
||||
idleDeferralTimers.delete(task.id)
|
||||
}
|
||||
|
||||
cleanupPendingByParent(task)
|
||||
|
||||
if (abortSession && task.sessionID) {
|
||||
client.session.abort({
|
||||
path: { id: task.sessionID },
|
||||
}).catch(() => {})
|
||||
}
|
||||
|
||||
if (options?.skipNotification) {
|
||||
log(`[background-agent] Task cancelled via ${source} (notification skipped):`, task.id)
|
||||
return true
|
||||
}
|
||||
|
||||
markForNotification(task)
|
||||
|
||||
try {
|
||||
await notifyParentSession(task)
|
||||
log(`[background-agent] Task cancelled via ${source}:`, task.id)
|
||||
} catch (err) {
|
||||
log("[background-agent] Error in notifyParentSession for cancelled task:", {
|
||||
taskId: task.id,
|
||||
error: err,
|
||||
})
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
@ -1,68 +0,0 @@
|
||||
import { log } from "../../shared"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
import type { OpencodeClient } from "./opencode-client"
|
||||
|
||||
export async function tryCompleteBackgroundTask(args: {
|
||||
task: BackgroundTask
|
||||
source: string
|
||||
concurrencyManager: ConcurrencyManager
|
||||
idleDeferralTimers: Map<string, ReturnType<typeof setTimeout>>
|
||||
client: OpencodeClient
|
||||
markForNotification: (task: BackgroundTask) => void
|
||||
cleanupPendingByParent: (task: BackgroundTask) => void
|
||||
notifyParentSession: (task: BackgroundTask) => Promise<void>
|
||||
}): Promise<boolean> {
|
||||
const {
|
||||
task,
|
||||
source,
|
||||
concurrencyManager,
|
||||
idleDeferralTimers,
|
||||
client,
|
||||
markForNotification,
|
||||
cleanupPendingByParent,
|
||||
notifyParentSession,
|
||||
} = args
|
||||
|
||||
if (task.status !== "running") {
|
||||
log("[background-agent] Task already completed, skipping:", {
|
||||
taskId: task.id,
|
||||
status: task.status,
|
||||
source,
|
||||
})
|
||||
return false
|
||||
}
|
||||
|
||||
task.status = "completed"
|
||||
task.completedAt = new Date()
|
||||
|
||||
if (task.concurrencyKey) {
|
||||
concurrencyManager.release(task.concurrencyKey)
|
||||
task.concurrencyKey = undefined
|
||||
}
|
||||
|
||||
markForNotification(task)
|
||||
cleanupPendingByParent(task)
|
||||
|
||||
const idleTimer = idleDeferralTimers.get(task.id)
|
||||
if (idleTimer) {
|
||||
clearTimeout(idleTimer)
|
||||
idleDeferralTimers.delete(task.id)
|
||||
}
|
||||
|
||||
if (task.sessionID) {
|
||||
client.session.abort({
|
||||
path: { id: task.sessionID },
|
||||
}).catch(() => {})
|
||||
}
|
||||
|
||||
try {
|
||||
await notifyParentSession(task)
|
||||
log(`[background-agent] Task completed via ${source}:`, task.id)
|
||||
} catch (err) {
|
||||
log("[background-agent] Error in notifyParentSession:", { taskId: task.id, error: err })
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
@ -1,77 +0,0 @@
|
||||
import { getTaskToastManager } from "../task-toast-manager"
|
||||
import { log } from "../../shared"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { LaunchInput } from "./types"
|
||||
|
||||
type QueueItem = {
|
||||
task: BackgroundTask
|
||||
input: LaunchInput
|
||||
}
|
||||
|
||||
export function launchBackgroundTask(args: {
|
||||
input: LaunchInput
|
||||
tasks: Map<string, BackgroundTask>
|
||||
pendingByParent: Map<string, Set<string>>
|
||||
queuesByKey: Map<string, QueueItem[]>
|
||||
getConcurrencyKeyFromInput: (input: LaunchInput) => string
|
||||
processKey: (key: string) => void
|
||||
}): BackgroundTask {
|
||||
const { input, tasks, pendingByParent, queuesByKey, getConcurrencyKeyFromInput, processKey } = args
|
||||
|
||||
log("[background-agent] launch() called with:", {
|
||||
agent: input.agent,
|
||||
model: input.model,
|
||||
description: input.description,
|
||||
parentSessionID: input.parentSessionID,
|
||||
})
|
||||
|
||||
if (!input.agent || input.agent.trim() === "") {
|
||||
throw new Error("Agent parameter is required")
|
||||
}
|
||||
|
||||
const task: BackgroundTask = {
|
||||
id: `bg_${crypto.randomUUID().slice(0, 8)}`,
|
||||
status: "pending",
|
||||
queuedAt: new Date(),
|
||||
description: input.description,
|
||||
prompt: input.prompt,
|
||||
agent: input.agent,
|
||||
parentSessionID: input.parentSessionID,
|
||||
parentMessageID: input.parentMessageID,
|
||||
parentModel: input.parentModel,
|
||||
parentAgent: input.parentAgent,
|
||||
model: input.model,
|
||||
category: input.category,
|
||||
}
|
||||
|
||||
tasks.set(task.id, task)
|
||||
|
||||
if (input.parentSessionID) {
|
||||
const pending = pendingByParent.get(input.parentSessionID) ?? new Set<string>()
|
||||
pending.add(task.id)
|
||||
pendingByParent.set(input.parentSessionID, pending)
|
||||
}
|
||||
|
||||
const key = getConcurrencyKeyFromInput(input)
|
||||
const queue = queuesByKey.get(key) ?? []
|
||||
queue.push({ task, input })
|
||||
queuesByKey.set(key, queue)
|
||||
|
||||
log("[background-agent] Task queued:", { taskId: task.id, key, queueLength: queue.length })
|
||||
|
||||
const toastManager = getTaskToastManager()
|
||||
if (toastManager) {
|
||||
toastManager.addTask({
|
||||
id: task.id,
|
||||
description: input.description,
|
||||
agent: input.agent,
|
||||
isBackground: true,
|
||||
status: "queued",
|
||||
skills: input.skills,
|
||||
})
|
||||
}
|
||||
|
||||
processKey(key)
|
||||
return task
|
||||
}
|
||||
@ -1,56 +0,0 @@
|
||||
import type { BackgroundTask } from "./types"
|
||||
|
||||
export function getTasksByParentSession(
|
||||
tasks: Iterable<BackgroundTask>,
|
||||
sessionID: string
|
||||
): BackgroundTask[] {
|
||||
const result: BackgroundTask[] = []
|
||||
for (const task of tasks) {
|
||||
if (task.parentSessionID === sessionID) {
|
||||
result.push(task)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
export function getAllDescendantTasks(
|
||||
tasksByParent: (sessionID: string) => BackgroundTask[],
|
||||
sessionID: string
|
||||
): BackgroundTask[] {
|
||||
const result: BackgroundTask[] = []
|
||||
const directChildren = tasksByParent(sessionID)
|
||||
|
||||
for (const child of directChildren) {
|
||||
result.push(child)
|
||||
if (child.sessionID) {
|
||||
result.push(...getAllDescendantTasks(tasksByParent, child.sessionID))
|
||||
}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
export function findTaskBySession(
|
||||
tasks: Iterable<BackgroundTask>,
|
||||
sessionID: string
|
||||
): BackgroundTask | undefined {
|
||||
for (const task of tasks) {
|
||||
if (task.sessionID === sessionID) return task
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
export function getRunningTasks(tasks: Iterable<BackgroundTask>): BackgroundTask[] {
|
||||
return Array.from(tasks).filter((t) => t.status === "running")
|
||||
}
|
||||
|
||||
export function getNonRunningTasks(tasks: Iterable<BackgroundTask>): BackgroundTask[] {
|
||||
return Array.from(tasks).filter((t) => t.status !== "running")
|
||||
}
|
||||
|
||||
export function hasRunningTasks(tasks: Iterable<BackgroundTask>): boolean {
|
||||
for (const task of tasks) {
|
||||
if (task.status === "running") return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
@ -1,52 +0,0 @@
|
||||
import { log } from "../../shared"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
|
||||
type QueueItem = {
|
||||
task: BackgroundTask
|
||||
input: import("./types").LaunchInput
|
||||
}
|
||||
|
||||
export async function processConcurrencyKeyQueue(args: {
|
||||
key: string
|
||||
queuesByKey: Map<string, QueueItem[]>
|
||||
processingKeys: Set<string>
|
||||
concurrencyManager: ConcurrencyManager
|
||||
startTask: (item: QueueItem) => Promise<void>
|
||||
}): Promise<void> {
|
||||
const { key, queuesByKey, processingKeys, concurrencyManager, startTask } = args
|
||||
|
||||
if (processingKeys.has(key)) return
|
||||
processingKeys.add(key)
|
||||
|
||||
try {
|
||||
const queue = queuesByKey.get(key)
|
||||
while (queue && queue.length > 0) {
|
||||
const item = queue[0]
|
||||
|
||||
await concurrencyManager.acquire(key)
|
||||
|
||||
if (item.task.status === "cancelled" || item.task.status === "error") {
|
||||
concurrencyManager.release(key)
|
||||
queue.shift()
|
||||
continue
|
||||
}
|
||||
|
||||
try {
|
||||
await startTask(item)
|
||||
} catch (error) {
|
||||
log("[background-agent] Error starting task:", error)
|
||||
// Release concurrency slot if startTask failed and didn't release it itself
|
||||
// This prevents slot leaks when errors occur after acquire but before task.concurrencyKey is set
|
||||
if (!item.task.concurrencyKey) {
|
||||
concurrencyManager.release(key)
|
||||
}
|
||||
}
|
||||
|
||||
queue.shift()
|
||||
}
|
||||
} finally {
|
||||
processingKeys.delete(key)
|
||||
}
|
||||
}
|
||||
@ -1,148 +0,0 @@
|
||||
import { log, getAgentToolRestrictions } from "../../shared"
|
||||
import { subagentSessions } from "../claude-code-session-state"
|
||||
import { getTaskToastManager } from "../task-toast-manager"
|
||||
|
||||
import type { BackgroundTask, ResumeInput } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
import type { OpencodeClient } from "./opencode-client"
|
||||
|
||||
type ModelRef = { providerID: string; modelID: string }
|
||||
|
||||
export async function resumeBackgroundTask(args: {
|
||||
input: ResumeInput
|
||||
findBySession: (sessionID: string) => BackgroundTask | undefined
|
||||
client: OpencodeClient
|
||||
concurrencyManager: ConcurrencyManager
|
||||
pendingByParent: Map<string, Set<string>>
|
||||
startPolling: () => void
|
||||
markForNotification: (task: BackgroundTask) => void
|
||||
cleanupPendingByParent: (task: BackgroundTask) => void
|
||||
notifyParentSession: (task: BackgroundTask) => Promise<void>
|
||||
}): Promise<BackgroundTask> {
|
||||
const {
|
||||
input,
|
||||
findBySession,
|
||||
client,
|
||||
concurrencyManager,
|
||||
pendingByParent,
|
||||
startPolling,
|
||||
markForNotification,
|
||||
cleanupPendingByParent,
|
||||
notifyParentSession,
|
||||
} = args
|
||||
|
||||
const existingTask = findBySession(input.sessionId)
|
||||
if (!existingTask) {
|
||||
throw new Error(`Task not found for session: ${input.sessionId}`)
|
||||
}
|
||||
|
||||
if (!existingTask.sessionID) {
|
||||
throw new Error(`Task has no sessionID: ${existingTask.id}`)
|
||||
}
|
||||
|
||||
if (existingTask.status === "running") {
|
||||
log("[background-agent] Resume skipped - task already running:", {
|
||||
taskId: existingTask.id,
|
||||
sessionID: existingTask.sessionID,
|
||||
})
|
||||
return existingTask
|
||||
}
|
||||
|
||||
const concurrencyKey =
|
||||
existingTask.concurrencyGroup ??
|
||||
(existingTask.model
|
||||
? `${existingTask.model.providerID}/${existingTask.model.modelID}`
|
||||
: existingTask.agent)
|
||||
await concurrencyManager.acquire(concurrencyKey)
|
||||
existingTask.concurrencyKey = concurrencyKey
|
||||
existingTask.concurrencyGroup = concurrencyKey
|
||||
|
||||
existingTask.status = "running"
|
||||
existingTask.completedAt = undefined
|
||||
existingTask.error = undefined
|
||||
existingTask.parentSessionID = input.parentSessionID
|
||||
existingTask.parentMessageID = input.parentMessageID
|
||||
existingTask.parentModel = input.parentModel
|
||||
existingTask.parentAgent = input.parentAgent
|
||||
existingTask.startedAt = new Date()
|
||||
|
||||
existingTask.progress = {
|
||||
toolCalls: existingTask.progress?.toolCalls ?? 0,
|
||||
lastUpdate: new Date(),
|
||||
}
|
||||
|
||||
startPolling()
|
||||
if (existingTask.sessionID) {
|
||||
subagentSessions.add(existingTask.sessionID)
|
||||
}
|
||||
|
||||
if (input.parentSessionID) {
|
||||
const pending = pendingByParent.get(input.parentSessionID) ?? new Set<string>()
|
||||
pending.add(existingTask.id)
|
||||
pendingByParent.set(input.parentSessionID, pending)
|
||||
}
|
||||
|
||||
const toastManager = getTaskToastManager()
|
||||
if (toastManager) {
|
||||
toastManager.addTask({
|
||||
id: existingTask.id,
|
||||
description: existingTask.description,
|
||||
agent: existingTask.agent,
|
||||
isBackground: true,
|
||||
})
|
||||
}
|
||||
|
||||
log("[background-agent] Resuming task:", { taskId: existingTask.id, sessionID: existingTask.sessionID })
|
||||
log("[background-agent] Resuming task - calling prompt (fire-and-forget) with:", {
|
||||
sessionID: existingTask.sessionID,
|
||||
agent: existingTask.agent,
|
||||
model: existingTask.model,
|
||||
promptLength: input.prompt.length,
|
||||
})
|
||||
|
||||
const resumeModel: ModelRef | undefined = existingTask.model
|
||||
? { providerID: existingTask.model.providerID, modelID: existingTask.model.modelID }
|
||||
: undefined
|
||||
const resumeVariant = existingTask.model?.variant
|
||||
|
||||
client.session.promptAsync({
|
||||
path: { id: existingTask.sessionID },
|
||||
body: {
|
||||
agent: existingTask.agent,
|
||||
...(resumeModel ? { model: resumeModel } : {}),
|
||||
...(resumeVariant ? { variant: resumeVariant } : {}),
|
||||
tools: {
|
||||
...getAgentToolRestrictions(existingTask.agent),
|
||||
task: false,
|
||||
call_omo_agent: true,
|
||||
question: false,
|
||||
},
|
||||
parts: [{ type: "text", text: input.prompt }],
|
||||
},
|
||||
}).catch((error) => {
|
||||
log("[background-agent] resume prompt error:", error)
|
||||
existingTask.status = "interrupt"
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
existingTask.error = errorMessage
|
||||
existingTask.completedAt = new Date()
|
||||
|
||||
if (existingTask.concurrencyKey) {
|
||||
concurrencyManager.release(existingTask.concurrencyKey)
|
||||
existingTask.concurrencyKey = undefined
|
||||
}
|
||||
|
||||
if (existingTask.sessionID) {
|
||||
client.session.abort({
|
||||
path: { id: existingTask.sessionID },
|
||||
}).catch(() => {})
|
||||
}
|
||||
|
||||
markForNotification(existingTask)
|
||||
cleanupPendingByParent(existingTask)
|
||||
notifyParentSession(existingTask).catch((err) => {
|
||||
log("[background-agent] Failed to notify on resume error:", err)
|
||||
})
|
||||
})
|
||||
|
||||
return existingTask
|
||||
}
|
||||
@ -1,190 +0,0 @@
|
||||
import { log, getAgentToolRestrictions, promptWithModelSuggestionRetry } from "../../shared"
|
||||
import { isInsideTmux } from "../../shared/tmux"
|
||||
|
||||
import { subagentSessions } from "../claude-code-session-state"
|
||||
import { getTaskToastManager } from "../task-toast-manager"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { LaunchInput } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
import type { OpencodeClient } from "./opencode-client"
|
||||
|
||||
type QueueItem = {
|
||||
task: BackgroundTask
|
||||
input: LaunchInput
|
||||
}
|
||||
|
||||
type ModelRef = { providerID: string; modelID: string }
|
||||
|
||||
export async function startQueuedTask(args: {
|
||||
item: QueueItem
|
||||
client: OpencodeClient
|
||||
defaultDirectory: string
|
||||
tmuxEnabled: boolean
|
||||
onSubagentSessionCreated?: (event: { sessionID: string; parentID: string; title: string }) => Promise<void>
|
||||
startPolling: () => void
|
||||
getConcurrencyKeyFromInput: (input: LaunchInput) => string
|
||||
concurrencyManager: ConcurrencyManager
|
||||
findBySession: (sessionID: string) => BackgroundTask | undefined
|
||||
markForNotification: (task: BackgroundTask) => void
|
||||
cleanupPendingByParent: (task: BackgroundTask) => void
|
||||
notifyParentSession: (task: BackgroundTask) => Promise<void>
|
||||
}): Promise<void> {
|
||||
const {
|
||||
item,
|
||||
client,
|
||||
defaultDirectory,
|
||||
tmuxEnabled,
|
||||
onSubagentSessionCreated,
|
||||
startPolling,
|
||||
getConcurrencyKeyFromInput,
|
||||
concurrencyManager,
|
||||
findBySession,
|
||||
markForNotification,
|
||||
cleanupPendingByParent,
|
||||
notifyParentSession,
|
||||
} = args
|
||||
|
||||
const { task, input } = item
|
||||
|
||||
log("[background-agent] Starting task:", {
|
||||
taskId: task.id,
|
||||
agent: input.agent,
|
||||
model: input.model,
|
||||
})
|
||||
|
||||
const concurrencyKey = getConcurrencyKeyFromInput(input)
|
||||
|
||||
const parentSession = await client.session.get({
|
||||
path: { id: input.parentSessionID },
|
||||
}).catch((err) => {
|
||||
log(`[background-agent] Failed to get parent session: ${err}`)
|
||||
return null
|
||||
})
|
||||
|
||||
const parentDirectory = parentSession?.data?.directory ?? defaultDirectory
|
||||
log(`[background-agent] Parent dir: ${parentSession?.data?.directory}, using: ${parentDirectory}`)
|
||||
|
||||
const createResult = await client.session.create({
|
||||
body: {
|
||||
parentID: input.parentSessionID,
|
||||
title: `${input.description} (@${input.agent} subagent)`,
|
||||
} as any,
|
||||
query: {
|
||||
directory: parentDirectory,
|
||||
},
|
||||
})
|
||||
|
||||
if (createResult.error) {
|
||||
throw new Error(`Failed to create background session: ${createResult.error}`)
|
||||
}
|
||||
|
||||
if (!createResult.data?.id) {
|
||||
throw new Error("Failed to create background session: API returned no session ID")
|
||||
}
|
||||
|
||||
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 200ms")
|
||||
await new Promise<void>((resolve) => {
|
||||
setTimeout(() => resolve(), 200)
|
||||
})
|
||||
} else {
|
||||
log("[background-agent] SKIP tmux callback - conditions not met")
|
||||
}
|
||||
|
||||
task.status = "running"
|
||||
task.startedAt = new Date()
|
||||
task.sessionID = sessionID
|
||||
task.progress = {
|
||||
toolCalls: 0,
|
||||
lastUpdate: new Date(),
|
||||
}
|
||||
task.concurrencyKey = concurrencyKey
|
||||
task.concurrencyGroup = concurrencyKey
|
||||
|
||||
startPolling()
|
||||
|
||||
log("[background-agent] Launching task:", { taskId: task.id, sessionID, agent: input.agent })
|
||||
|
||||
const toastManager = getTaskToastManager()
|
||||
if (toastManager) {
|
||||
toastManager.updateTask(task.id, "running")
|
||||
}
|
||||
|
||||
log("[background-agent] Calling prompt (fire-and-forget) for launch with:", {
|
||||
sessionID,
|
||||
agent: input.agent,
|
||||
model: input.model,
|
||||
hasSkillContent: !!input.skillContent,
|
||||
promptLength: input.prompt.length,
|
||||
})
|
||||
|
||||
const launchModel: ModelRef | undefined = input.model
|
||||
? { providerID: input.model.providerID, modelID: input.model.modelID }
|
||||
: undefined
|
||||
const launchVariant = input.model?.variant
|
||||
|
||||
promptWithModelSuggestionRetry(client, {
|
||||
path: { id: sessionID },
|
||||
body: {
|
||||
agent: input.agent,
|
||||
...(launchModel ? { model: launchModel } : {}),
|
||||
...(launchVariant ? { variant: launchVariant } : {}),
|
||||
system: input.skillContent,
|
||||
tools: {
|
||||
...getAgentToolRestrictions(input.agent),
|
||||
task: false,
|
||||
call_omo_agent: true,
|
||||
question: false,
|
||||
},
|
||||
parts: [{ type: "text", text: input.prompt }],
|
||||
},
|
||||
}).catch((error) => {
|
||||
log("[background-agent] promptAsync error:", error)
|
||||
const existingTask = findBySession(sessionID)
|
||||
if (!existingTask) return
|
||||
|
||||
existingTask.status = "interrupt"
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
if (errorMessage.includes("agent.name") || errorMessage.includes("undefined")) {
|
||||
existingTask.error = `Agent "${input.agent}" not found. Make sure the agent is registered in your opencode.json or provided by a plugin.`
|
||||
} else {
|
||||
existingTask.error = errorMessage
|
||||
}
|
||||
existingTask.completedAt = new Date()
|
||||
|
||||
if (existingTask.concurrencyKey) {
|
||||
concurrencyManager.release(existingTask.concurrencyKey)
|
||||
existingTask.concurrencyKey = undefined
|
||||
}
|
||||
|
||||
client.session.abort({
|
||||
path: { id: sessionID },
|
||||
}).catch(() => {})
|
||||
|
||||
markForNotification(existingTask)
|
||||
cleanupPendingByParent(existingTask)
|
||||
notifyParentSession(existingTask).catch((err) => {
|
||||
log("[background-agent] Failed to notify on error:", err)
|
||||
})
|
||||
})
|
||||
}
|
||||
@ -1,97 +0,0 @@
|
||||
import { log } from "../../shared"
|
||||
import { subagentSessions } from "../claude-code-session-state"
|
||||
|
||||
import type { BackgroundTask } from "./types"
|
||||
import type { ConcurrencyManager } from "./concurrency"
|
||||
|
||||
export async function trackExternalTask(args: {
|
||||
input: {
|
||||
taskId: string
|
||||
sessionID: string
|
||||
parentSessionID: string
|
||||
description: string
|
||||
agent?: string
|
||||
parentAgent?: string
|
||||
concurrencyKey?: string
|
||||
}
|
||||
tasks: Map<string, BackgroundTask>
|
||||
pendingByParent: Map<string, Set<string>>
|
||||
concurrencyManager: ConcurrencyManager
|
||||
startPolling: () => void
|
||||
cleanupPendingByParent: (task: BackgroundTask) => void
|
||||
}): Promise<BackgroundTask> {
|
||||
const { input, tasks, pendingByParent, concurrencyManager, startPolling, cleanupPendingByParent } = args
|
||||
|
||||
const existingTask = tasks.get(input.taskId)
|
||||
if (existingTask) {
|
||||
const parentChanged = input.parentSessionID !== existingTask.parentSessionID
|
||||
if (parentChanged) {
|
||||
cleanupPendingByParent(existingTask)
|
||||
existingTask.parentSessionID = input.parentSessionID
|
||||
}
|
||||
if (input.parentAgent !== undefined) {
|
||||
existingTask.parentAgent = input.parentAgent
|
||||
}
|
||||
if (!existingTask.concurrencyGroup) {
|
||||
existingTask.concurrencyGroup = input.concurrencyKey ?? existingTask.agent
|
||||
}
|
||||
|
||||
if (existingTask.sessionID) {
|
||||
subagentSessions.add(existingTask.sessionID)
|
||||
}
|
||||
startPolling()
|
||||
|
||||
if (existingTask.status === "pending" || existingTask.status === "running") {
|
||||
const pending = pendingByParent.get(input.parentSessionID) ?? new Set<string>()
|
||||
pending.add(existingTask.id)
|
||||
pendingByParent.set(input.parentSessionID, pending)
|
||||
} else if (!parentChanged) {
|
||||
cleanupPendingByParent(existingTask)
|
||||
}
|
||||
|
||||
log("[background-agent] External task already registered:", {
|
||||
taskId: existingTask.id,
|
||||
sessionID: existingTask.sessionID,
|
||||
status: existingTask.status,
|
||||
})
|
||||
|
||||
return existingTask
|
||||
}
|
||||
|
||||
const concurrencyGroup = input.concurrencyKey ?? input.agent ?? "task"
|
||||
if (input.concurrencyKey) {
|
||||
await concurrencyManager.acquire(input.concurrencyKey)
|
||||
}
|
||||
|
||||
const task: BackgroundTask = {
|
||||
id: input.taskId,
|
||||
sessionID: input.sessionID,
|
||||
parentSessionID: input.parentSessionID,
|
||||
parentMessageID: "",
|
||||
description: input.description,
|
||||
prompt: "",
|
||||
agent: input.agent || "task",
|
||||
status: "running",
|
||||
startedAt: new Date(),
|
||||
progress: {
|
||||
toolCalls: 0,
|
||||
lastUpdate: new Date(),
|
||||
},
|
||||
parentAgent: input.parentAgent,
|
||||
concurrencyKey: input.concurrencyKey,
|
||||
concurrencyGroup,
|
||||
}
|
||||
|
||||
tasks.set(task.id, task)
|
||||
subagentSessions.add(input.sessionID)
|
||||
startPolling()
|
||||
|
||||
if (input.parentSessionID) {
|
||||
const pending = pendingByParent.get(input.parentSessionID) ?? new Set<string>()
|
||||
pending.add(task.id)
|
||||
pendingByParent.set(input.parentSessionID, pending)
|
||||
}
|
||||
|
||||
log("[background-agent] Registered external task:", { taskId: task.id, sessionID: input.sessionID })
|
||||
return task
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user