fix(delegate-task): fetch session result before honoring abort signal (#2702)
The sync poller returned 'Task aborted' immediately when the abort signal fired, even when a terminal assistant message had already arrived during the previous wait. Now attempts one final fetch and returns the completion result if available before emitting the abort message. 🤖 Generated with OhMyOpenCode assistance https://github.com/code-yeongyu/oh-my-opencode
This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
|
import { afterEach, beforeEach, describe, expect, test } from "bun:test"
|
||||||
declare const require: (name: string) => any
|
declare const require: (name: string) => any
|
||||||
const { describe, test, expect, beforeEach, afterEach } = require("bun:test")
|
|
||||||
import { __setTimingConfig, __resetTimingConfig } from "./timing"
|
import { __setTimingConfig, __resetTimingConfig } from "./timing"
|
||||||
|
|
||||||
function createMockCtx(aborted = false) {
|
function createMockCtx(aborted = false) {
|
||||||
@@ -33,7 +33,6 @@ describe("pollSyncSession", () => {
|
|||||||
// and the assistant id > user id (native opencode condition)
|
// and the assistant id > user id (native opencode condition)
|
||||||
const { pollSyncSession } = require("./sync-session-poller")
|
const { pollSyncSession } = require("./sync-session-poller")
|
||||||
|
|
||||||
let pollCount = 0
|
|
||||||
const mockClient = {
|
const mockClient = {
|
||||||
session: {
|
session: {
|
||||||
messages: async () => ({
|
messages: async () => ({
|
||||||
@@ -272,6 +271,55 @@ describe("pollSyncSession", () => {
|
|||||||
})
|
})
|
||||||
|
|
||||||
describe("abort handling", () => {
|
describe("abort handling", () => {
|
||||||
|
test("#given session completed AND abort fires #then returns completion result not abort", async () => {
|
||||||
|
//#given
|
||||||
|
const { pollSyncSession } = require("./sync-session-poller")
|
||||||
|
const controller = new AbortController()
|
||||||
|
controller.abort()
|
||||||
|
|
||||||
|
let abortCount = 0
|
||||||
|
let messageCallCount = 0
|
||||||
|
const mockClient = {
|
||||||
|
session: {
|
||||||
|
abort: async () => {
|
||||||
|
abortCount++
|
||||||
|
},
|
||||||
|
messages: async () => {
|
||||||
|
messageCallCount++
|
||||||
|
return {
|
||||||
|
data: [
|
||||||
|
{ info: { id: "msg_001", role: "user", time: { created: 1000 } } },
|
||||||
|
{
|
||||||
|
info: { id: "msg_002", role: "assistant", time: { created: 2000 }, finish: "stop" },
|
||||||
|
parts: [{ type: "text", text: "Done" }],
|
||||||
|
},
|
||||||
|
],
|
||||||
|
}
|
||||||
|
},
|
||||||
|
status: async () => ({ data: {} }),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
//#when
|
||||||
|
const result = await pollSyncSession({
|
||||||
|
sessionID: "parent-session",
|
||||||
|
messageID: "parent-message",
|
||||||
|
agent: "test-agent",
|
||||||
|
abort: controller.signal,
|
||||||
|
}, mockClient, {
|
||||||
|
sessionID: "ses_abort_complete",
|
||||||
|
agentToUse: "test-agent",
|
||||||
|
toastManager: { removeTask: () => {} },
|
||||||
|
taskId: "task_123",
|
||||||
|
anchorMessageCount: 1,
|
||||||
|
})
|
||||||
|
|
||||||
|
//#then
|
||||||
|
expect(result).toBeNull()
|
||||||
|
expect(messageCallCount).toBe(1)
|
||||||
|
expect(abortCount).toBe(0)
|
||||||
|
})
|
||||||
|
|
||||||
test("returns abort message when signal is aborted", async () => {
|
test("returns abort message when signal is aborted", async () => {
|
||||||
//#given
|
//#given
|
||||||
const { pollSyncSession } = require("./sync-session-poller")
|
const { pollSyncSession } = require("./sync-session-poller")
|
||||||
@@ -347,7 +395,7 @@ describe("pollSyncSession", () => {
|
|||||||
//#given
|
//#given
|
||||||
const { pollSyncSession } = require("./sync-session-poller")
|
const { pollSyncSession } = require("./sync-session-poller")
|
||||||
|
|
||||||
let statusCallCount = 0
|
let statusCallCount = 0
|
||||||
let messageCallCount = 0
|
let messageCallCount = 0
|
||||||
const mockClient = {
|
const mockClient = {
|
||||||
session: {
|
session: {
|
||||||
|
|||||||
@@ -23,6 +23,15 @@ function abortSyncSession(client: OpencodeClient, sessionID: string, reason: str
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function fetchSessionMessages(
|
||||||
|
client: OpencodeClient,
|
||||||
|
sessionID: string
|
||||||
|
): Promise<SessionMessage[]> {
|
||||||
|
const messagesResult = await client.session.messages({ path: { id: sessionID } })
|
||||||
|
const rawData = (messagesResult as { data?: unknown })?.data ?? messagesResult
|
||||||
|
return Array.isArray(rawData) ? (rawData as SessionMessage[]) : []
|
||||||
|
}
|
||||||
|
|
||||||
export function isSessionComplete(messages: SessionMessage[]): boolean {
|
export function isSessionComplete(messages: SessionMessage[]): boolean {
|
||||||
let lastUser: SessionMessage | undefined
|
let lastUser: SessionMessage | undefined
|
||||||
let lastAssistant: SessionMessage | undefined
|
let lastAssistant: SessionMessage | undefined
|
||||||
@@ -69,6 +78,21 @@ export async function pollSyncSession(
|
|||||||
|
|
||||||
while (Date.now() - pollStart < maxPollTimeMs) {
|
while (Date.now() - pollStart < maxPollTimeMs) {
|
||||||
if (ctx.abort?.aborted) {
|
if (ctx.abort?.aborted) {
|
||||||
|
try {
|
||||||
|
const messages = await fetchSessionMessages(client, input.sessionID)
|
||||||
|
const hasNewMessages =
|
||||||
|
input.anchorMessageCount === undefined || messages.length > input.anchorMessageCount
|
||||||
|
if (hasNewMessages && isSessionComplete(messages)) {
|
||||||
|
log("[task] Abort detected after session already completed", { sessionID: input.sessionID })
|
||||||
|
return null
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
log("[task] Final messages fetch failed after abort, continuing with abort", {
|
||||||
|
sessionID: input.sessionID,
|
||||||
|
error: String(error),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
log("[task] Aborted by user", { sessionID: input.sessionID })
|
log("[task] Aborted by user", { sessionID: input.sessionID })
|
||||||
abortSyncSession(client, input.sessionID, "parent_abort")
|
abortSyncSession(client, input.sessionID, "parent_abort")
|
||||||
if (input.toastManager && input.taskId) input.toastManager.removeTask(input.taskId)
|
if (input.toastManager && input.taskId) input.toastManager.removeTask(input.taskId)
|
||||||
@@ -101,27 +125,25 @@ export async function pollSyncSession(
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
let messagesResult: { data?: unknown } | SessionMessage[]
|
let messages: SessionMessage[]
|
||||||
try {
|
try {
|
||||||
messagesResult = await client.session.messages({ path: { id: input.sessionID } })
|
messages = await fetchSessionMessages(client, input.sessionID)
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
log("[task] Poll messages fetch failed, retrying", { sessionID: input.sessionID, error: String(error) })
|
log("[task] Poll messages fetch failed, retrying", { sessionID: input.sessionID, error: String(error) })
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
const rawData = (messagesResult as { data?: unknown })?.data ?? messagesResult
|
|
||||||
const msgs = Array.isArray(rawData) ? (rawData as SessionMessage[]) : []
|
|
||||||
|
|
||||||
if (input.anchorMessageCount !== undefined && msgs.length <= input.anchorMessageCount) {
|
if (input.anchorMessageCount !== undefined && messages.length <= input.anchorMessageCount) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
if (isSessionComplete(msgs)) {
|
if (isSessionComplete(messages)) {
|
||||||
log("[task] Poll complete - terminal finish detected", { sessionID: input.sessionID, pollCount })
|
log("[task] Poll complete - terminal finish detected", { sessionID: input.sessionID, pollCount })
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
// 计数新出现的 assistant 轮次,用于熔断无限循环
|
// 计数新出现的 assistant 轮次,用于熔断无限循环
|
||||||
const lastAssistant = [...msgs].reverse().find((m) => m.info?.role === "assistant")
|
const lastAssistant = [...messages].reverse().find((m) => m.info?.role === "assistant")
|
||||||
if (lastAssistant?.info?.id && lastAssistant.info.id !== lastSeenAssistantId) {
|
if (lastAssistant?.info?.id && lastAssistant.info.id !== lastSeenAssistantId) {
|
||||||
lastSeenAssistantId = lastAssistant.info.id
|
lastSeenAssistantId = lastAssistant.info.id
|
||||||
assistantTurnCount++
|
assistantTurnCount++
|
||||||
@@ -137,7 +159,7 @@ export async function pollSyncSession(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
const hasAssistantText = msgs.some((m) => {
|
const hasAssistantText = messages.some((m) => {
|
||||||
if (m.info?.role !== "assistant") return false
|
if (m.info?.role !== "assistant") return false
|
||||||
const parts = m.parts ?? []
|
const parts = m.parts ?? []
|
||||||
return parts.some((p) => {
|
return parts.some((p) => {
|
||||||
|
|||||||
Reference in New Issue
Block a user