mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 01:11:59 +00:00
修三处订阅缺陷:审批恢复后事件重复、订阅泄漏、取消误杀
UI 重构前置。这三条都是新界面会放大的问题——时间线要常驻显示事件流, 重复累积和泄漏在旧界面里藏得住,在新界面里就是明显的错。 ## 审批恢复后事件重复收一遍 streamRun 里 `let lastSeq = 0` 是每次调用新建的闭包局部量,注释写着"重连时 带上已收到的最大 seq 所以不会重复"——那只在同一次调用内的断线重连成立。 审批放行会调 attachRun 重新订阅,lastSeq 归零,后端按 `seq > after` 把整条 历史再推一遍,而 applyEvent 是无条件 `[...events, event]`。 实测:一个带审批的运行中断时有 6 条事件,放行后共 15 条。修复前 attachRun 用 after=0 会把那 6 条再收一遍;现在传 max(seq)=6,只收 9 条新的。 把 after 提成 streamRun 的参数比在客户端做去重干净——省掉服务端的重发本身。 ## chat store 的订阅泄漏 cancel 是 store 上的单值,下一轮 ask 直接覆盖,上一轮的 WebSocket 再也关不掉。 句柄下放到 turn 级,clear 时逐个关。 ## stop() 误杀等待审批的历史轮次 原来把所有非终态 turn 一律标成"已取消"。正等人工介入的轮次是等着用户回来 处理的,不是卡住的——把它们标成已取消,用户回来就找不到那个审批了。 现在只停最后一轮,且跳过 waiting。 ## ApprovalCard 解除对 studio store 的硬依赖 它被对话页和运行页复用,而"恢复后重新接上事件流"默认走 studio.attachRun—— 在对话页用就是同时开两条订阅(chat 一条、studio 一条),两边各收一份事件。 加 onResolved 回调,谁用谁负责重新接流。
This commit is contained in:
@@ -274,17 +274,22 @@ export function streamCopilot(
|
||||
/**
|
||||
* 订阅一次运行的事件流。
|
||||
*
|
||||
* 先补历史再接实时由后端保证,这里只负责断线重连 —— 重连时带上已收到的最大
|
||||
* seq,所以不会重复也不会丢事件。
|
||||
* 先补历史再接实时由后端保证,这里负责断线重连——重连时带上已收到的最大 seq。
|
||||
*
|
||||
* `after` 必须由调用方给:lastSeq 是这次调用的闭包局部量,跨调用会归零。
|
||||
* 审批恢复时会重新 attachRun,不传 after 就是 after=0,后端按 `seq > after`
|
||||
* 把整条历史再推一遍,而消费方多半是无条件追加——事件就这么重复累积了。
|
||||
* 已经有事件在手的场景(重新接上一条运行)应当传当前最大 seq。
|
||||
*/
|
||||
export function streamRun(
|
||||
runId: string,
|
||||
onEvent: (event: RunEvent) => void,
|
||||
onClose?: () => void,
|
||||
after = 0,
|
||||
): () => void {
|
||||
let socket: WebSocket | null = null
|
||||
let closed = false
|
||||
let lastSeq = 0
|
||||
let lastSeq = after
|
||||
let retry = 0
|
||||
|
||||
const connect = () => {
|
||||
|
||||
@@ -186,15 +186,12 @@ function Approvals({ turn }: { turn: ChatTurn }) {
|
||||
const reattach = useChat((s) => s.reattach)
|
||||
const pending = approvals.filter((a) => a.run_id === turn.run?.id && a.status === 'pending')
|
||||
|
||||
// 审批处理完后运行会继续,重新接上事件流才能看到后续
|
||||
useEffect(() => {
|
||||
if (turn.phase === 'waiting' && !pending.length) void reattach(turn.id)
|
||||
}, [pending.length, turn.phase, turn.id, reattach])
|
||||
|
||||
if (!pending.length) return null
|
||||
return (
|
||||
<div className="mt-2 overflow-hidden rounded border" style={{ borderColor: 'var(--warn)' }}>
|
||||
{pending.map((a) => <ApprovalCard key={a.id} approval={a} />)}
|
||||
{pending.map((a) => (
|
||||
<ApprovalCard key={a.id} approval={a} onResolved={() => reattach(turn.id)} />
|
||||
))}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
@@ -172,7 +172,18 @@ function RunLauncher() {
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
export function ApprovalCard({ approval }: { approval: any }) {
|
||||
/**
|
||||
* 人工介入卡片。
|
||||
*
|
||||
* onResolved 是给非画布场景用的:这个组件被对话页和运行页复用,而默认的
|
||||
* "恢复后重新接上事件流"走的是 studio store——在对话页里用就会同时开两条
|
||||
* 订阅(一条 chat 的、一条 studio 的),两边各收一份事件。谁用谁负责重新
|
||||
* 接流,是唯一不会打架的分工。
|
||||
*/
|
||||
export function ApprovalCard({ approval, onResolved }: {
|
||||
approval: any
|
||||
onResolved?: (runId: string) => void | Promise<void>
|
||||
}) {
|
||||
const toast = useToast()
|
||||
const refreshApprovals = useCatalog((s) => s.refreshApprovals)
|
||||
const attachRun = useStudio((s) => s.attachRun)
|
||||
@@ -192,7 +203,9 @@ export function ApprovalCard({ approval }: { approval: any }) {
|
||||
...(approval.mode !== 'approve' ? { value } : {}),
|
||||
})
|
||||
await refreshApprovals()
|
||||
await attachRun(approval.run_id) // 恢复后重新接上事件流
|
||||
// 恢复后重新接上事件流:调用方给了 onResolved 就听它的
|
||||
if (onResolved) await onResolved(approval.run_id)
|
||||
else await attachRun(approval.run_id)
|
||||
// 驳回在"补充输入/编辑草稿"模式下会终止整个运行(图上没有 rejected 那条
|
||||
// 出口边),说清楚比笼统一句"已驳回"诚实
|
||||
toast(
|
||||
|
||||
@@ -41,6 +41,11 @@ export interface ChatTurn {
|
||||
output: Record<string, any> | null
|
||||
error: string
|
||||
startedAt: number
|
||||
/** 已收到的最大事件 seq。重新接流时要带上,否则后端把历史整条重推 */
|
||||
lastSeq: number
|
||||
/** 这一轮自己的取消句柄。放 turn 上而不是 store 上:store 单值会被
|
||||
* 下一轮 ask 覆盖,上一轮的 WebSocket 就再也关不掉了 */
|
||||
cancel: (() => void) | null
|
||||
}
|
||||
|
||||
interface ChatState {
|
||||
@@ -88,6 +93,7 @@ export const useChat = create<ChatState>((set, get) => ({
|
||||
id, question, phase: 'planning', status: PHASE_TEXT.planning,
|
||||
thinking: '', graph: null, explanation: '', run: null,
|
||||
steps: [], output: null, error: '', startedAt: Date.now(),
|
||||
lastSeq: 0, cancel: null,
|
||||
},
|
||||
],
|
||||
}))
|
||||
@@ -148,22 +154,31 @@ export const useChat = create<ChatState>((set, get) => ({
|
||||
set({ busy: false, cancel: null })
|
||||
},
|
||||
)
|
||||
patch(() => ({ cancel: cancelCopilot }))
|
||||
set({ cancel: cancelCopilot })
|
||||
},
|
||||
|
||||
stop: () => {
|
||||
// 只停最后一轮。把所有非终态 turn 一律标成"已取消"会误杀正等人工介入的
|
||||
// 历史轮次——那些是等着用户回来处理的,不是卡住的。
|
||||
const turns = get().turns
|
||||
const last = turns[turns.length - 1]
|
||||
last?.cancel?.()
|
||||
get().cancel?.()
|
||||
set((s) => ({
|
||||
busy: false,
|
||||
cancel: null,
|
||||
turns: s.turns.map((t) =>
|
||||
t.phase === 'done' || t.phase === 'error' ? t : { ...t, phase: 'error', error: '已取消' },
|
||||
t.id === last?.id && !['done', 'error', 'waiting'].includes(t.phase)
|
||||
? { ...t, phase: 'error', error: '已取消', cancel: null }
|
||||
: t,
|
||||
),
|
||||
}))
|
||||
},
|
||||
|
||||
clear: () => {
|
||||
get().cancel?.()
|
||||
get().turns.forEach((t) => t.cancel?.()) // 每轮各有各的订阅,逐个关
|
||||
set({ turns: [], busy: false, cancel: null })
|
||||
},
|
||||
|
||||
@@ -174,7 +189,7 @@ export const useChat = create<ChatState>((set, get) => ({
|
||||
set((s) => ({ turns: s.turns.map((t) => (t.id === turnId ? { ...t, ...fn(t) } : t)) }))
|
||||
patch(() => ({ phase: 'running', status: PHASE_TEXT.running }))
|
||||
set({ busy: true })
|
||||
watch(turn.run.id, turnId, patch, set)
|
||||
watch(turn.run.id, turnId, patch, set, turn.lastSeq)
|
||||
},
|
||||
}))
|
||||
|
||||
@@ -202,9 +217,12 @@ function watch(
|
||||
turnId: string,
|
||||
patch: (fn: (t: ChatTurn) => Partial<ChatTurn>) => void,
|
||||
set: any,
|
||||
after = 0,
|
||||
) {
|
||||
const stop = streamRun(runId, (event: RunEvent) => {
|
||||
const d: any = event.data ?? {}
|
||||
// 记下进度:审批恢复要重新接流,带上它才不会把历史再收一遍
|
||||
patch((t) => ({ lastSeq: Math.max(t.lastSeq ?? 0, event.seq ?? 0) }))
|
||||
switch (event.type) {
|
||||
case 'tool.start': {
|
||||
const tool = String(d.tool ?? '')
|
||||
@@ -261,7 +279,10 @@ function watch(
|
||||
void finish(runId, turnId, patch, set)
|
||||
break
|
||||
}
|
||||
})
|
||||
}, undefined, after)
|
||||
// 句柄同时挂在 turn 和 store 上:turn 上的用于精确关闭这一轮,
|
||||
// store 上的是"当前活跃流"的快捷方式
|
||||
patch(() => ({ cancel: stop }))
|
||||
set({ cancel: stop })
|
||||
}
|
||||
|
||||
|
||||
@@ -502,10 +502,17 @@ export const useStudio = create<StudioState>((set, get) => ({
|
||||
|
||||
attachRun: async (runId) => {
|
||||
get().unsubscribe?.()
|
||||
// 手上已有的事件不要再收一遍。审批恢复会走到这里,而 applyEvent 是无条件
|
||||
// 追加——不传 after 的话后端把整条历史重推一遍,时间线上每条都出现两次。
|
||||
const current = get()
|
||||
const after = current.run?.id === runId
|
||||
? current.events.reduce((max, e) => Math.max(max, e.seq ?? 0), 0)
|
||||
: 0
|
||||
const stop = streamRun(
|
||||
runId,
|
||||
(event) => get().applyEvent(event),
|
||||
() => set({ streaming: false }),
|
||||
after,
|
||||
)
|
||||
set({ unsubscribe: stop, streaming: true })
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user