Merge pull request #4228 from code-yeongyu/fix/delegate-stale-activity
Fix delegate stale timeout activity checks
This commit is contained in:
@@ -0,0 +1,154 @@
|
|||||||
|
import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"
|
||||||
|
import { tmpdir } from "node:os"
|
||||||
|
import type { PluginInput } from "@opencode-ai/plugin"
|
||||||
|
import { unsafeTestValue } from "../../../test-support/unsafe-test-value"
|
||||||
|
import { BackgroundManager } from "./manager"
|
||||||
|
import type { BackgroundTask } from "./types"
|
||||||
|
|
||||||
|
type PollingManager = {
|
||||||
|
readonly pollRunningTasks: () => Promise<void>
|
||||||
|
readonly tasks: Map<string, BackgroundTask>
|
||||||
|
}
|
||||||
|
|
||||||
|
function createPluginContext(client: unknown): PluginInput {
|
||||||
|
const directory = tmpdir()
|
||||||
|
return unsafeTestValue<PluginInput>({
|
||||||
|
project: {
|
||||||
|
id: "test-project",
|
||||||
|
worktree: directory,
|
||||||
|
time: { created: Date.now() },
|
||||||
|
},
|
||||||
|
directory,
|
||||||
|
worktree: directory,
|
||||||
|
serverUrl: new URL("http://localhost:4096"),
|
||||||
|
$: {},
|
||||||
|
client,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
function createRunningTask(overrides: Partial<BackgroundTask> = {}): BackgroundTask {
|
||||||
|
return {
|
||||||
|
id: "bg_test_session_activity",
|
||||||
|
sessionId: "ses-active",
|
||||||
|
parentSessionId: "parent-session",
|
||||||
|
parentMessageId: "parent-message",
|
||||||
|
description: "test task",
|
||||||
|
prompt: "test prompt",
|
||||||
|
agent: "explore",
|
||||||
|
status: "running",
|
||||||
|
startedAt: new Date(Date.now() - 120_000),
|
||||||
|
progress: { toolCalls: 0, lastUpdate: new Date() },
|
||||||
|
...overrides,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("BackgroundManager persisted session activity stale checks", () => {
|
||||||
|
const originalDateNow = Date.now
|
||||||
|
const fixedTime = new Date("2026-05-21T03:00:00.000Z").getTime()
|
||||||
|
|
||||||
|
afterEach(() => {
|
||||||
|
Date.now = originalDateNow
|
||||||
|
mock.restore()
|
||||||
|
})
|
||||||
|
|
||||||
|
test("keeps a busy task running when session.get reports fresh activity", async () => {
|
||||||
|
//#given - live event progress is stale, but OpenCode session metadata was updated recently
|
||||||
|
spyOn(globalThis.Date, "now").mockReturnValue(fixedTime)
|
||||||
|
let abortCallCount = 0
|
||||||
|
const sessionGet = mock(async () => ({
|
||||||
|
data: {
|
||||||
|
id: "ses-active",
|
||||||
|
time: { updated: fixedTime - 10_000 },
|
||||||
|
},
|
||||||
|
}))
|
||||||
|
const client = {
|
||||||
|
session: {
|
||||||
|
status: async () => ({ data: { "ses-active": { type: "busy" } } }),
|
||||||
|
get: sessionGet,
|
||||||
|
prompt: async () => ({}),
|
||||||
|
promptAsync: async () => ({}),
|
||||||
|
abort: async () => {
|
||||||
|
abortCallCount += 1
|
||||||
|
return {}
|
||||||
|
},
|
||||||
|
todo: async () => ({ data: [] }),
|
||||||
|
messages: async () => ({ data: [] }),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
const manager = new BackgroundManager({
|
||||||
|
pluginContext: createPluginContext(client),
|
||||||
|
config: { staleTimeoutMs: 180_000 },
|
||||||
|
enableParentSessionNotifications: false,
|
||||||
|
})
|
||||||
|
const task = createRunningTask({
|
||||||
|
startedAt: new Date(Date.now() - 45 * 60 * 1000),
|
||||||
|
progress: {
|
||||||
|
toolCalls: 3,
|
||||||
|
lastUpdate: new Date(Date.now() - 45 * 60 * 1000),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
const pollingManager = unsafeTestValue<PollingManager>(manager)
|
||||||
|
pollingManager.tasks.set(task.id, task)
|
||||||
|
|
||||||
|
//#when - polling reaches stale confirmation for the active child session
|
||||||
|
await pollingManager.pollRunningTasks()
|
||||||
|
|
||||||
|
//#then - persisted activity refreshes the task instead of aborting it
|
||||||
|
expect(task.status).toBe("running")
|
||||||
|
expect(task.error).toBeUndefined()
|
||||||
|
expect(task.progress?.lastUpdate).toEqual(new Date(fixedTime - 10_000))
|
||||||
|
expect(abortCallCount).toBe(0)
|
||||||
|
expect(sessionGet).toHaveBeenCalledTimes(1)
|
||||||
|
|
||||||
|
await manager.shutdown()
|
||||||
|
})
|
||||||
|
|
||||||
|
test("keeps a busy task running when session.get returns an error response", async () => {
|
||||||
|
//#given - live event progress is stale and OpenCode session lookup fails without throwing
|
||||||
|
spyOn(globalThis.Date, "now").mockReturnValue(fixedTime)
|
||||||
|
let abortCallCount = 0
|
||||||
|
const sessionGet = mock(async () => ({
|
||||||
|
error: "lookup failed",
|
||||||
|
data: undefined,
|
||||||
|
}))
|
||||||
|
const client = {
|
||||||
|
session: {
|
||||||
|
status: async () => ({ data: { "ses-active": { type: "busy" } } }),
|
||||||
|
get: sessionGet,
|
||||||
|
prompt: async () => ({}),
|
||||||
|
promptAsync: async () => ({}),
|
||||||
|
abort: async () => {
|
||||||
|
abortCallCount += 1
|
||||||
|
return {}
|
||||||
|
},
|
||||||
|
todo: async () => ({ data: [] }),
|
||||||
|
messages: async () => ({ data: [] }),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
const manager = new BackgroundManager({
|
||||||
|
pluginContext: createPluginContext(client),
|
||||||
|
config: { staleTimeoutMs: 180_000 },
|
||||||
|
enableParentSessionNotifications: false,
|
||||||
|
})
|
||||||
|
const task = createRunningTask({
|
||||||
|
startedAt: new Date(Date.now() - 45 * 60 * 1000),
|
||||||
|
progress: {
|
||||||
|
toolCalls: 3,
|
||||||
|
lastUpdate: new Date(Date.now() - 45 * 60 * 1000),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
const pollingManager = unsafeTestValue<PollingManager>(manager)
|
||||||
|
pollingManager.tasks.set(task.id, task)
|
||||||
|
|
||||||
|
//#when - polling tries to confirm stale activity through the SDK response
|
||||||
|
await pollingManager.pollRunningTasks()
|
||||||
|
|
||||||
|
//#then - the lookup failure defers cancellation for the active child session
|
||||||
|
expect(task.status).toBe("running")
|
||||||
|
expect(task.error).toBeUndefined()
|
||||||
|
expect(abortCallCount).toBe(0)
|
||||||
|
expect(sessionGet).toHaveBeenCalledTimes(1)
|
||||||
|
|
||||||
|
await manager.shutdown()
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -0,0 +1,56 @@
|
|||||||
|
import { isRecord, log } from "../../shared"
|
||||||
|
import type { OpencodeClient } from "./opencode-client"
|
||||||
|
|
||||||
|
export type SessionActivityLookup =
|
||||||
|
| { readonly type: "activity"; readonly activity: Date }
|
||||||
|
| { readonly type: "missing" }
|
||||||
|
| { readonly type: "unavailable" }
|
||||||
|
|
||||||
|
export type SessionActivityResolver = (sessionID: string) => Promise<SessionActivityLookup>
|
||||||
|
|
||||||
|
function dateFromMillis(value: unknown): Date | undefined {
|
||||||
|
if (typeof value !== "number") return undefined
|
||||||
|
if (!Number.isFinite(value) || value < 0) return undefined
|
||||||
|
return new Date(value)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function extractSessionActivityDate(sessionInfo: unknown): Date | undefined {
|
||||||
|
if (!isRecord(sessionInfo)) return undefined
|
||||||
|
const time = isRecord(sessionInfo.time) ? sessionInfo.time : undefined
|
||||||
|
return dateFromMillis(time?.updated) ?? dateFromMillis(sessionInfo.time_updated)
|
||||||
|
}
|
||||||
|
|
||||||
|
function sessionActivityLookupFromInfo(sessionInfo: unknown): SessionActivityLookup {
|
||||||
|
const activity = extractSessionActivityDate(sessionInfo)
|
||||||
|
return activity ? { type: "activity", activity } : { type: "missing" }
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function getSessionActivityFromClient(
|
||||||
|
client: OpencodeClient,
|
||||||
|
sessionID: string,
|
||||||
|
directory?: string,
|
||||||
|
): Promise<SessionActivityLookup> {
|
||||||
|
const sessionGet = client.session.get
|
||||||
|
if (typeof sessionGet !== "function") return { type: "missing" }
|
||||||
|
|
||||||
|
try {
|
||||||
|
const response = await sessionGet({
|
||||||
|
path: { id: sessionID },
|
||||||
|
...(directory ? { query: { directory } } : {}),
|
||||||
|
})
|
||||||
|
if (isRecord(response) && response.error !== undefined && response.error !== null) {
|
||||||
|
log("[background-agent] Failed to read session activity:", { sessionID, error: response.error })
|
||||||
|
return { type: "unavailable" }
|
||||||
|
}
|
||||||
|
|
||||||
|
const sessionInfo = isRecord(response) && "data" in response ? response.data : response
|
||||||
|
return sessionActivityLookupFromInfo(sessionInfo)
|
||||||
|
} catch (error) {
|
||||||
|
if (error instanceof Error) {
|
||||||
|
log("[background-agent] Failed to read session activity:", { sessionID, error: error.message })
|
||||||
|
return { type: "unavailable" }
|
||||||
|
}
|
||||||
|
log("[background-agent] Failed to read session activity:", { sessionID, error })
|
||||||
|
return { type: "unavailable" }
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,49 @@
|
|||||||
|
import { log } from "../../shared"
|
||||||
|
import type { SessionActivityLookup, SessionActivityResolver } from "./session-activity"
|
||||||
|
import type { BackgroundTask } from "./types"
|
||||||
|
|
||||||
|
export type TaskActivityRefreshResult =
|
||||||
|
| { readonly type: "activity"; readonly activityTime: number }
|
||||||
|
| { readonly type: "missing" }
|
||||||
|
| { readonly type: "unavailable" }
|
||||||
|
|
||||||
|
function updateTaskActivityFromLookup(
|
||||||
|
task: BackgroundTask,
|
||||||
|
lookup: SessionActivityLookup,
|
||||||
|
): TaskActivityRefreshResult {
|
||||||
|
if (lookup.type !== "activity") return lookup
|
||||||
|
|
||||||
|
const activityTime = lookup.activity.getTime()
|
||||||
|
if (!Number.isFinite(activityTime)) return { type: "missing" }
|
||||||
|
|
||||||
|
const baseline = task.progress?.lastUpdate.getTime() ?? task.startedAt?.getTime()
|
||||||
|
if (baseline !== undefined && activityTime <= baseline) return { type: "activity", activityTime }
|
||||||
|
|
||||||
|
if (!task.progress) {
|
||||||
|
task.progress = { toolCalls: 0, lastUpdate: new Date(activityTime) }
|
||||||
|
} else {
|
||||||
|
task.progress.lastUpdate = new Date(activityTime)
|
||||||
|
}
|
||||||
|
return { type: "activity", activityTime }
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function refreshTaskActivityFromSession(
|
||||||
|
task: BackgroundTask,
|
||||||
|
getSessionActivity: SessionActivityResolver,
|
||||||
|
): Promise<TaskActivityRefreshResult> {
|
||||||
|
if (!task.sessionId) return { type: "missing" }
|
||||||
|
|
||||||
|
let lookup: SessionActivityLookup
|
||||||
|
try {
|
||||||
|
lookup = await getSessionActivity(task.sessionId)
|
||||||
|
} catch (error) {
|
||||||
|
if (error instanceof Error) {
|
||||||
|
log("[background-agent] Error refreshing task session activity:", { taskId: task.id, error: error.message })
|
||||||
|
return { type: "unavailable" }
|
||||||
|
}
|
||||||
|
log("[background-agent] Error refreshing task session activity:", { taskId: task.id, error })
|
||||||
|
return { type: "unavailable" }
|
||||||
|
}
|
||||||
|
|
||||||
|
return updateTaskActivityFromLookup(task, lookup)
|
||||||
|
}
|
||||||
@@ -0,0 +1,160 @@
|
|||||||
|
import { afterEach, describe, expect, mock, spyOn, test } from "bun:test"
|
||||||
|
import { unsafeTestValue } from "../../../test-support/unsafe-test-value"
|
||||||
|
import { checkAndInterruptStaleTasks } from "./task-poller"
|
||||||
|
import type { BackgroundTask } from "./types"
|
||||||
|
|
||||||
|
function createRunningTask(overrides: Partial<BackgroundTask> = {}): BackgroundTask {
|
||||||
|
return {
|
||||||
|
id: "task-1",
|
||||||
|
sessionId: "ses-1",
|
||||||
|
parentSessionId: "parent-ses-1",
|
||||||
|
parentMessageId: "msg-1",
|
||||||
|
description: "test",
|
||||||
|
prompt: "test",
|
||||||
|
agent: "explore",
|
||||||
|
status: "running",
|
||||||
|
startedAt: new Date(Date.now() - 120_000),
|
||||||
|
...overrides,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("checkAndInterruptStaleTasks persisted session activity", () => {
|
||||||
|
const mockClient = unsafeTestValue<Parameters<typeof checkAndInterruptStaleTasks>[0]["client"]>({
|
||||||
|
session: {
|
||||||
|
abort: mock(() => Promise.resolve()),
|
||||||
|
get: mock(() => Promise.resolve({ data: { id: "ses-1" } })),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
const mockConcurrencyManager = unsafeTestValue<Parameters<typeof checkAndInterruptStaleTasks>[0]["concurrencyManager"]>({
|
||||||
|
release: mock(() => {}),
|
||||||
|
})
|
||||||
|
const mockNotify = mock(() => Promise.resolve())
|
||||||
|
|
||||||
|
const originalDateNow = Date.now
|
||||||
|
const fixedTime = new Date("2026-05-21T03:00:00.000Z").getTime()
|
||||||
|
|
||||||
|
afterEach(() => {
|
||||||
|
Date.now = originalDateNow
|
||||||
|
mockClient.session.abort.mockClear()
|
||||||
|
mockConcurrencyManager.release.mockClear()
|
||||||
|
mockNotify.mockClear()
|
||||||
|
})
|
||||||
|
|
||||||
|
test("keeps a busy task running when persisted session activity is fresh", async () => {
|
||||||
|
//#given - in-memory progress is stale, but OpenCode storage shows recent child-session activity
|
||||||
|
spyOn(globalThis.Date, "now").mockReturnValue(fixedTime)
|
||||||
|
const staleActivity = new Date(Date.now() - 45 * 60 * 1000)
|
||||||
|
const freshActivity = new Date(Date.now() - 10_000)
|
||||||
|
const task = createRunningTask({
|
||||||
|
startedAt: staleActivity,
|
||||||
|
progress: {
|
||||||
|
toolCalls: 2,
|
||||||
|
lastUpdate: staleActivity,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
//#when - stale checking can refresh activity from the persisted session timestamp
|
||||||
|
await checkAndInterruptStaleTasks({
|
||||||
|
tasks: [task],
|
||||||
|
client: mockClient,
|
||||||
|
config: { staleTimeoutMs: 180_000 },
|
||||||
|
concurrencyManager: mockConcurrencyManager,
|
||||||
|
notifyParentSession: mockNotify,
|
||||||
|
sessionStatuses: { "ses-1": { type: "busy" } },
|
||||||
|
getSessionActivity: async () => ({ type: "activity", activity: freshActivity }),
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then - task stays running and future stale checks use the persisted activity timestamp
|
||||||
|
expect(task.status).toBe("running")
|
||||||
|
expect(task.progress?.lastUpdate).toEqual(freshActivity)
|
||||||
|
expect(mockNotify).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
|
test("keeps a busy task with no local progress running when persisted session activity is fresh", async () => {
|
||||||
|
//#given - no progress event reached the manager, but OpenCode storage shows recent activity
|
||||||
|
spyOn(globalThis.Date, "now").mockReturnValue(fixedTime)
|
||||||
|
const staleStart = new Date(Date.now() - 15 * 60 * 1000)
|
||||||
|
const freshActivity = new Date(Date.now() - 10_000)
|
||||||
|
const task = createRunningTask({
|
||||||
|
startedAt: staleStart,
|
||||||
|
progress: undefined,
|
||||||
|
})
|
||||||
|
|
||||||
|
//#when - message staleness confirmation refreshes from persisted session metadata
|
||||||
|
await checkAndInterruptStaleTasks({
|
||||||
|
tasks: [task],
|
||||||
|
client: mockClient,
|
||||||
|
config: { messageStalenessTimeoutMs: 180_000 },
|
||||||
|
concurrencyManager: mockConcurrencyManager,
|
||||||
|
notifyParentSession: mockNotify,
|
||||||
|
sessionStatuses: { "ses-1": { type: "busy" } },
|
||||||
|
getSessionActivity: async () => ({ type: "activity", activity: freshActivity }),
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then - the task stays running and receives a progress timestamp for future stale checks
|
||||||
|
expect(task.status).toBe("running")
|
||||||
|
expect(task.progress?.lastUpdate).toEqual(freshActivity)
|
||||||
|
expect(mockNotify).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
|
test("cancels a busy task when persisted session activity is also stale", async () => {
|
||||||
|
//#given - local progress is older than persisted activity, but both are outside the stale window
|
||||||
|
spyOn(globalThis.Date, "now").mockReturnValue(fixedTime)
|
||||||
|
const localActivity = new Date(Date.now() - 45 * 60 * 1000)
|
||||||
|
const stalePersistedActivity = new Date(Date.now() - 10 * 60 * 1000)
|
||||||
|
const task = createRunningTask({
|
||||||
|
startedAt: localActivity,
|
||||||
|
progress: {
|
||||||
|
toolCalls: 2,
|
||||||
|
lastUpdate: localActivity,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
//#when - the persisted timestamp confirms the child session is truly stale
|
||||||
|
await checkAndInterruptStaleTasks({
|
||||||
|
tasks: [task],
|
||||||
|
client: mockClient,
|
||||||
|
config: { staleTimeoutMs: 180_000 },
|
||||||
|
concurrencyManager: mockConcurrencyManager,
|
||||||
|
notifyParentSession: mockNotify,
|
||||||
|
sessionStatuses: { "ses-1": { type: "busy" } },
|
||||||
|
getSessionActivity: async () => ({ type: "activity", activity: stalePersistedActivity }),
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then - cancellation still happens, but the stale age reflects persisted activity
|
||||||
|
expect(task.status).toBe("cancelled")
|
||||||
|
expect(task.error).toContain("Stale timeout")
|
||||||
|
expect(task.error).toContain("10min")
|
||||||
|
expect(task.progress?.lastUpdate).toEqual(stalePersistedActivity)
|
||||||
|
expect(mockNotify).toHaveBeenCalledWith(task)
|
||||||
|
})
|
||||||
|
|
||||||
|
test("keeps a busy task running when persisted session activity lookup is unavailable", async () => {
|
||||||
|
//#given - the child session is active, but session.get returned an error response during confirmation
|
||||||
|
spyOn(globalThis.Date, "now").mockReturnValue(fixedTime)
|
||||||
|
const staleActivity = new Date(Date.now() - 45 * 60 * 1000)
|
||||||
|
const task = createRunningTask({
|
||||||
|
startedAt: staleActivity,
|
||||||
|
progress: {
|
||||||
|
toolCalls: 2,
|
||||||
|
lastUpdate: staleActivity,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
//#when - stale checking cannot verify persisted activity for this poll
|
||||||
|
await checkAndInterruptStaleTasks({
|
||||||
|
tasks: [task],
|
||||||
|
client: mockClient,
|
||||||
|
config: { staleTimeoutMs: 180_000 },
|
||||||
|
concurrencyManager: mockConcurrencyManager,
|
||||||
|
notifyParentSession: mockNotify,
|
||||||
|
sessionStatuses: { "ses-1": { type: "busy" } },
|
||||||
|
getSessionActivity: async () => ({ type: "unavailable" }),
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then - active-session cancellation is deferred instead of treating lookup failure as inactivity
|
||||||
|
expect(task.status).toBe("running")
|
||||||
|
expect(task.progress?.lastUpdate).toEqual(staleActivity)
|
||||||
|
expect(mockNotify).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -19,6 +19,8 @@ import { removeTaskToastTracking } from "./remove-task-toast-tracking"
|
|||||||
import { MIN_SESSION_GONE_POLLS, verifySessionExists } from "./session-existence"
|
import { MIN_SESSION_GONE_POLLS, verifySessionExists } from "./session-existence"
|
||||||
|
|
||||||
import { isActiveSessionStatus } from "./session-status-classifier"
|
import { isActiveSessionStatus } from "./session-status-classifier"
|
||||||
|
import { getSessionActivityFromClient, type SessionActivityResolver } from "./session-activity"
|
||||||
|
import { refreshTaskActivityFromSession } from "./task-activity-refresh"
|
||||||
|
|
||||||
const TERMINAL_TASK_STATUSES = new Set<BackgroundTask["status"]>([
|
const TERMINAL_TASK_STATUSES = new Set<BackgroundTask["status"]>([
|
||||||
"completed",
|
"completed",
|
||||||
@@ -120,6 +122,7 @@ export async function checkAndInterruptStaleTasks(args: {
|
|||||||
notifyParentSession: (task: BackgroundTask) => Promise<void>
|
notifyParentSession: (task: BackgroundTask) => Promise<void>
|
||||||
sessionStatuses?: SessionStatusMap
|
sessionStatuses?: SessionStatusMap
|
||||||
onTaskInterrupted?: (task: BackgroundTask) => void
|
onTaskInterrupted?: (task: BackgroundTask) => void
|
||||||
|
getSessionActivity?: SessionActivityResolver
|
||||||
}): Promise<void> {
|
}): Promise<void> {
|
||||||
const {
|
const {
|
||||||
tasks,
|
tasks,
|
||||||
@@ -137,6 +140,8 @@ export async function checkAndInterruptStaleTasks(args: {
|
|||||||
const abortPromises: Array<Promise<unknown>> = []
|
const abortPromises: Array<Promise<unknown>> = []
|
||||||
|
|
||||||
const messageStalenessMs = config?.messageStalenessTimeoutMs ?? DEFAULT_MESSAGE_STALENESS_TIMEOUT_MS
|
const messageStalenessMs = config?.messageStalenessTimeoutMs ?? DEFAULT_MESSAGE_STALENESS_TIMEOUT_MS
|
||||||
|
const getSessionActivity = args.getSessionActivity
|
||||||
|
?? ((id: string) => getSessionActivityFromClient(client, id, directory))
|
||||||
|
|
||||||
for (const task of tasks) {
|
for (const task of tasks) {
|
||||||
if (task.status !== "running") continue
|
if (task.status !== "running") continue
|
||||||
@@ -157,6 +162,9 @@ export async function checkAndInterruptStaleTasks(args: {
|
|||||||
|
|
||||||
const sessionGone = sessionMissing && (task.consecutiveMissedPolls ?? 0) >= MIN_SESSION_GONE_POLLS
|
const sessionGone = sessionMissing && (task.consecutiveMissedPolls ?? 0) >= MIN_SESSION_GONE_POLLS
|
||||||
const shouldSkipInactivityTimeout = task.teamRunId !== undefined && !sessionGone
|
const shouldSkipInactivityTimeout = task.teamRunId !== undefined && !sessionGone
|
||||||
|
const shouldRefreshFromSessionActivity = !sessionGone
|
||||||
|
&& sessionStatus !== undefined
|
||||||
|
&& isActiveSessionStatus(sessionStatus)
|
||||||
|
|
||||||
if (!task.progress?.lastUpdate) {
|
if (!task.progress?.lastUpdate) {
|
||||||
if (shouldSkipInactivityTimeout) continue
|
if (shouldSkipInactivityTimeout) continue
|
||||||
@@ -164,6 +172,12 @@ export async function checkAndInterruptStaleTasks(args: {
|
|||||||
const effectiveTimeout = sessionGone ? sessionGoneTimeoutMs : messageStalenessMs
|
const effectiveTimeout = sessionGone ? sessionGoneTimeoutMs : messageStalenessMs
|
||||||
if (runtime <= effectiveTimeout) continue
|
if (runtime <= effectiveTimeout) continue
|
||||||
|
|
||||||
|
if (shouldRefreshFromSessionActivity) {
|
||||||
|
const activityRefresh = await refreshTaskActivityFromSession(task, getSessionActivity)
|
||||||
|
if (activityRefresh.type === "unavailable") continue
|
||||||
|
if (activityRefresh.type === "activity" && now - activityRefresh.activityTime <= effectiveTimeout) continue
|
||||||
|
}
|
||||||
|
|
||||||
if (sessionGone && await verifySessionExists(client, sessionID, directory)) {
|
if (sessionGone && await verifySessionExists(client, sessionID, directory)) {
|
||||||
task.consecutiveMissedPolls = 0
|
task.consecutiveMissedPolls = 0
|
||||||
continue
|
continue
|
||||||
@@ -197,9 +211,21 @@ export async function checkAndInterruptStaleTasks(args: {
|
|||||||
|
|
||||||
if (runtime < MIN_RUNTIME_BEFORE_STALE_MS) continue
|
if (runtime < MIN_RUNTIME_BEFORE_STALE_MS) continue
|
||||||
|
|
||||||
const timeSinceLastUpdate = now - task.progress.lastUpdate.getTime()
|
let timeSinceLastUpdate = now - task.progress.lastUpdate.getTime()
|
||||||
const effectiveStaleTimeout = sessionGone ? sessionGoneTimeoutMs : staleTimeoutMs
|
const effectiveStaleTimeout = sessionGone ? sessionGoneTimeoutMs : staleTimeoutMs
|
||||||
if (timeSinceLastUpdate <= effectiveStaleTimeout) continue
|
if (timeSinceLastUpdate <= effectiveStaleTimeout) continue
|
||||||
|
|
||||||
|
if (shouldRefreshFromSessionActivity) {
|
||||||
|
const activityRefresh = await refreshTaskActivityFromSession(task, getSessionActivity)
|
||||||
|
if (activityRefresh.type === "unavailable") continue
|
||||||
|
const refreshedLastUpdate = task.progress?.lastUpdate.getTime()
|
||||||
|
?? (activityRefresh.type === "activity" ? activityRefresh.activityTime : undefined)
|
||||||
|
if (refreshedLastUpdate !== undefined && now - refreshedLastUpdate <= effectiveStaleTimeout) continue
|
||||||
|
if (refreshedLastUpdate !== undefined) {
|
||||||
|
timeSinceLastUpdate = now - refreshedLastUpdate
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (task.status !== "running") continue
|
if (task.status !== "running") continue
|
||||||
|
|
||||||
if (sessionGone && await verifySessionExists(client, sessionID, directory)) {
|
if (sessionGone && await verifySessionExists(client, sessionID, directory)) {
|
||||||
|
|||||||
@@ -191,9 +191,10 @@ describe("remapAgentKeysToDisplayNames", () => {
|
|||||||
const result = remapAgentKeysToDisplayNames(agents)
|
const result = remapAgentKeysToDisplayNames(agents)
|
||||||
|
|
||||||
// then exactly one row is emitted under the clean literal display name
|
// then exactly one row is emitted under the clean literal display name
|
||||||
expect(Object.keys(result)).toEqual(["Sisyphus - Ultraworker"])
|
const displayName = getAgentListDisplayName("sisyphus")
|
||||||
expect(result["Sisyphus - Ultraworker"]).toEqual({
|
expect(Object.keys(result)).toEqual([displayName])
|
||||||
name: "Sisyphus - Ultraworker",
|
expect(result[displayName]).toEqual({
|
||||||
|
name: displayName,
|
||||||
foo: "bar",
|
foo: "bar",
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ Original error: ${createResult.error}`
|
|||||||
log(`[look_at] Created session: ${sessionID}`)
|
log(`[look_at] Created session: ${sessionID}`)
|
||||||
|
|
||||||
log(`[look_at] Sending prompt with ${isBase64Input ? "base64 image" : "file"} to session ${sessionID}`)
|
log(`[look_at] Sending prompt with ${isBase64Input ? "base64 image" : "file"} to session ${sessionID}`)
|
||||||
|
let promptFailed = false
|
||||||
try {
|
try {
|
||||||
await promptSyncWithModelSuggestionRetry(ctx.client, {
|
await promptSyncWithModelSuggestionRetry(ctx.client, {
|
||||||
path: { id: sessionID },
|
path: { id: sessionID },
|
||||||
@@ -83,27 +84,39 @@ Original error: ${createResult.error}`
|
|||||||
queueBehavior: "defer",
|
queueBehavior: "defer",
|
||||||
})
|
})
|
||||||
} catch (promptError) {
|
} catch (promptError) {
|
||||||
|
promptFailed = true
|
||||||
log("[look_at] Prompt error (ignored, will still fetch messages):", promptError)
|
log("[look_at] Prompt error (ignored, will still fetch messages):", promptError)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let observedMessages: unknown[] | undefined
|
||||||
|
let observedText: string | undefined
|
||||||
if (typeof ctx.client.session.status === "function") {
|
if (typeof ctx.client.session.status === "function") {
|
||||||
await waitForLookAtSessionResult(ctx.client, sessionID)
|
const waitResult = await waitForLookAtSessionResult(ctx.client, sessionID, {
|
||||||
|
allowStableIdleWithoutActivity: true,
|
||||||
|
allowEmptyStableIdleWithoutActivity: promptFailed,
|
||||||
|
})
|
||||||
|
observedText = waitResult.outcome.text ?? undefined
|
||||||
|
if (observedText) {
|
||||||
|
observedMessages = waitResult.messages
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
log(`[look_at] Fetching messages from session ${sessionID}...`)
|
let messages = observedMessages
|
||||||
const messagesResult = await ctx.client.session.messages({
|
if (!messages) {
|
||||||
path: { id: sessionID },
|
log(`[look_at] Fetching messages from session ${sessionID}...`)
|
||||||
})
|
const messagesResult = await ctx.client.session.messages({
|
||||||
|
path: { id: sessionID },
|
||||||
|
})
|
||||||
|
|
||||||
if (messagesResult.error) {
|
if (messagesResult.error) {
|
||||||
log("[look_at] Messages error:", messagesResult.error)
|
log("[look_at] Messages error:", messagesResult.error)
|
||||||
return `Error: Failed to get messages: ${messagesResult.error}`
|
return `Error: Failed to get messages: ${messagesResult.error}`
|
||||||
|
}
|
||||||
|
messages = messagesResult.data
|
||||||
}
|
}
|
||||||
|
|
||||||
const messages = messagesResult.data
|
|
||||||
log(`[look_at] Got ${messages.length} messages`)
|
log(`[look_at] Got ${messages.length} messages`)
|
||||||
|
|
||||||
const responseText = extractLatestAssistantText(messages)
|
const responseText = observedText ?? extractLatestAssistantText(messages)
|
||||||
if (!responseText) {
|
if (!responseText) {
|
||||||
log("[look_at] No assistant message found")
|
log("[look_at] No assistant message found")
|
||||||
return "Error: No response from multimodal-looker agent"
|
return "Error: No response from multimodal-looker agent"
|
||||||
|
|||||||
@@ -78,6 +78,18 @@ describe("waitForLookAtSessionResult", () => {
|
|||||||
expect(result.outcome.text).toBe("done")
|
expect(result.outcome.text).toBe("done")
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test("#given session is absent and has no assistant output #when stable idle is allowed #then keeps polling", async () => {
|
||||||
|
const client = createMockClient([{ data: {} }], [])
|
||||||
|
|
||||||
|
await expect(
|
||||||
|
waitForLookAtSessionResult(unsafeTestValue(client), "ses_test", {
|
||||||
|
pollIntervalMs: 10,
|
||||||
|
timeoutMs: 50,
|
||||||
|
allowStableIdleWithoutActivity: true,
|
||||||
|
}),
|
||||||
|
).rejects.toThrow("timed out")
|
||||||
|
})
|
||||||
|
|
||||||
test("#given session never becomes idle #when polling exceeds timeout #then rejects", async () => {
|
test("#given session never becomes idle #when polling exceeds timeout #then rejects", async () => {
|
||||||
const client = createMockClient(
|
const client = createMockClient(
|
||||||
[{ data: { ses_test: { type: "busy" } } }],
|
[{ data: { ses_test: { type: "busy" } } }],
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ export interface PollOptions {
|
|||||||
timeoutMs?: number
|
timeoutMs?: number
|
||||||
abortSignal?: AbortSignal
|
abortSignal?: AbortSignal
|
||||||
allowStableIdleWithoutActivity?: boolean
|
allowStableIdleWithoutActivity?: boolean
|
||||||
|
allowEmptyStableIdleWithoutActivity?: boolean
|
||||||
}
|
}
|
||||||
|
|
||||||
const DEFAULT_POLL_INTERVAL_MS = 1000
|
const DEFAULT_POLL_INTERVAL_MS = 1000
|
||||||
@@ -136,7 +137,10 @@ export async function waitForLookAtSessionResult(
|
|||||||
const canConcludeIdle =
|
const canConcludeIdle =
|
||||||
sawNonIdleStatus ||
|
sawNonIdleStatus ||
|
||||||
!status.supported ||
|
!status.supported ||
|
||||||
Boolean(options?.allowStableIdleWithoutActivity)
|
(
|
||||||
|
Boolean(options?.allowStableIdleWithoutActivity)
|
||||||
|
&& (outcome.hasAssistant || Boolean(options?.allowEmptyStableIdleWithoutActivity))
|
||||||
|
)
|
||||||
|
|
||||||
if (canConcludeIdle && stableIdlePolls >= IDLE_STABILITY_POLLS_REQUIRED) {
|
if (canConcludeIdle && stableIdlePolls >= IDLE_STABILITY_POLLS_REQUIRED) {
|
||||||
return { messages, outcome, statusType }
|
return { messages, outcome, statusType }
|
||||||
|
|||||||
Reference in New Issue
Block a user