From 6e4474cd2e0a9173cec02838e9e1c32940311093 Mon Sep 17 00:00:00 2001 From: Menghuan1918 Date: Wed, 12 Aug 2026 23:02:37 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BB=BB=E5=8A=A1=E8=BE=93=E5=87=BA?= =?UTF-8?q?=E5=9B=9E=E6=94=BE=E5=90=88=E5=B9=B6=E6=B4=BB=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E9=95=9C=E5=83=8F=EF=BC=8C=E4=BF=AE=E5=A4=8D=E9=87=8D=E5=90=AF?= =?UTF-8?q?=E5=90=8E=20store=20=E4=BC=9A=E8=AF=9D=E6=BB=9E=E5=90=8E?= =?UTF-8?q?=E5=AF=BC=E8=87=B4=E6=96=B0=E8=AF=BB=E5=8F=96=E4=B8=8D=E5=8F=AF?= =?UTF-8?q?=E8=A7=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - host 半挂 ctx.on('session/event')（api-proxy 同款 API），镜像 task_output 的 tool/call + tool/result 活事件（每会话 200 条有界环，无关结果不缓存） - tasks.output 回放合并「store 事件日志 + 活事件镜像」，按 seq 去重排序配对 - 零 DSH 写入：会话 store 在宿主重启后可能冻结在再水合边界（持久化文件 停在重启时刻），活事件只走 append 推送链；镜像绕开该滞后 - 测试：新增镜像用例（live 事件、跨会话隔离、种子+镜像合并不双计）， 全量 373 通过 --- src/context-types.ts | 9 ++ src/tasks-routes.ts | 180 ++++++++++++++++++++++++++++++------- tests/tasks-routes.spec.ts | 71 +++++++++++++++ 3 files changed, 226 insertions(+), 34 deletions(-) diff --git a/src/context-types.ts b/src/context-types.ts index d6df297..de751a4 100644 --- a/src/context-types.ts +++ b/src/context-types.ts @@ -388,6 +388,15 @@ declare module 'cordis' { * {@link ./client/index.tsx}); undefined on the host side. */ betterSidebar: BetterSidebarService + /** + * Subscribe to the session append feed (mirror of the cordis event API): + * the listener receives every appended session event with the LIVE + * Session instance that appended it. The api-proxy pushes the same feed + * to browsers; the sidebar uses it to mirror task_output events the + * session store's own log can lag behind (restart divergence). Returns + * the disposer. + */ + on(event: string, listener: (session: unknown, event: SidebarSessionEvent) => void): () => void /** * Register a lifecycle callback (DSH-vendored cordis): runs at plugin * activation; its returned cleanup runs at disposal. diff --git a/src/tasks-routes.ts b/src/tasks-routes.ts index 95b18ae..3fcf0ba 100644 --- a/src/tasks-routes.ts +++ b/src/tasks-routes.ts @@ -4,18 +4,21 @@ * harness's `session/tasks` push mirror (`tasksBySession` in the sessions * list feed). The routes: * - * - 'tasks.output' — REPLAYS the output the MODEL has read so far from the - * owner session's OWN event log: `tool/call` rows of `task_output` name - * the task via `arguments.task_id`, and the paired `tool/result` rows - * carry the finalized content the model actually received. This touches - * NO DSH source (no peek seam, no consuming read): the model's - * `task_output` cursor is untouched by construction, and the pane stays - * empty until the agent reads the task. + * - 'tasks.output' — REPLAYS the output the MODEL has read so far for one + * task. The source is the owner session's own event log: `tool/call` rows + * of `task_output` name the task via `arguments.task_id`, and the paired + * `tool/result` rows carry the finalized content the model received. + * Because the session store's in-memory log can lag the live append feed + * after a host restart (the store session stays frozen at its + * rehydration boundary), the plugin ALSO mirrors task_output events from + * the live `session/event` feed and merges both sources (deduped by seq). + * This touches NO DSH source: the model's `task_output` cursor is never + * consumed, and the pane stays empty until the agent reads the task. * - 'tasks.kill' — the registry's stock `kill` (a pristine DSH API), * fenced by the owning session via the live agent caller. Absent registry * → 503, mirroring the settings routes' optional-service downgrade. */ -import type { Context } from './context-types.ts' +import type { Context, SidebarSessionEvent } from './context-types.ts' import { requireString, SidebarError } from './wire.ts' /** The two background-task routes of the sidebar API. */ @@ -50,7 +53,7 @@ function resultText(message: ToolResultMessageLike): string | undefined { for (const block of message.content) { if (block === null || typeof block !== 'object') continue const candidate = block as ToolResultBlockLike - if (candidate.type !== 'tool-result' || candidate.isError === true) continue + if (candidate.type !== 'tool-result') continue const inner = candidate.content if (!Array.isArray(inner)) continue for (const item of inner) { @@ -64,16 +67,130 @@ function resultText(message: ToolResultMessageLike): string | undefined { return parts.length > 0 ? parts.join('\n') : undefined } +/** Whether a tool/result is an error result (the inner block's isError flag). */ +function resultIsError(message: ToolResultMessageLike): boolean { + if (!Array.isArray(message.content)) return false + return message.content.some((block) => { + if (block === null || typeof block !== 'object') return false + return (block as ToolResultBlockLike).type === 'tool-result' + && (block as ToolResultBlockLike).isError === true + }) +} + /** Whether a task_output result carries no new output — the controller's * model-facing "(no new output)" body, noise for the human pane. */ function isNoNewOutput(text: string): boolean { return text.startsWith('(no new output)') } +/** One compact task_output trace (a tool/call or its paired tool/result). */ +interface TaskOutputTrace { + seq: number + kind: 'call' | 'result' + /** The tool call identity pairing the two rows. */ + callId: string + /** tool/call: the task id parsed from the model arguments. */ + taskId?: string + /** tool/result: the finalized text the model received. */ + text?: string + /** tool/result: whether the result was an error (read counts, text skipped). */ + isError?: boolean +} + +/** Extract the task_output trace of one raw session event (undefined = unrelated). */ +function traceOf(event: SidebarSessionEvent): TaskOutputTrace | undefined { + if (event.type === 'tool/call') { + const data = event.data as { name?: unknown; callId?: unknown; arguments?: unknown } + if (data.name !== 'task_output' || typeof data.callId !== 'string') return undefined + let taskId: string | undefined + try { + const args = JSON.parse(typeof data.arguments === 'string' ? data.arguments : '') as { task_id?: unknown } + if (typeof args.task_id === 'string') taskId = args.task_id + } catch { + // Malformed model arguments: not a task_output pair. + } + if (taskId === undefined) return undefined + return { seq: event.seq, kind: 'call', callId: data.callId, taskId } + } + if (event.type === 'tool/result') { + const message = (event.data as { message?: unknown }).message as ToolResultMessageLike | undefined + if (message === undefined) return undefined + const callId = message.source?.callId + if (typeof callId !== 'string') return undefined + return { + seq: event.seq, + kind: 'result', + callId, + text: resultText(message), + isError: resultIsError(message), + } + } + return undefined +} + +/** Per-session cap of mirrored live traces (a bounded, lossy ring). */ +const MIRROR_MAX_ENTRIES = 200 + /** - * Build the tasks routes bound to the plugin context. `output` reads only - * the session store; `kill` reads the tasks/agents services lazily and - * degrades to a 503 when the deployment lacks the registry. + * The live task_output mirror: subscribes to the session append feed and + * caches the task_output traces the session store's own log can lag behind + * (after a host restart the store session stays frozen at its rehydration + * boundary, so `session.events` misses everything appended since — the very + * reads the pane exists to show). Zero DSH writes: the api-proxy pushes the + * same feed to browsers. + */ +function createTaskOutputMirror(ctx: Context): { entries(sessionId: string): readonly TaskOutputTrace[] } { + const perSession = new Map() + // tool/call identities per session, so unrelated tool/result rows are + // never cached (only task_output results pair with a cached call). + const callIds = new Map>() + if (typeof ctx.on !== 'function') { + // Test doubles without the event API degrade to seed-only replay. + return { entries: () => [] } + } + const dispose = ctx.on('session/event', (session, event) => { + const sessionId = (session as { id?: unknown } | null)?.id + if (typeof sessionId !== 'string') return + if (event.type === 'tool/call') { + const trace = traceOf(event) + if (trace?.kind !== 'call') return + let ids = callIds.get(sessionId) + if (ids === undefined) callIds.set(sessionId, ids = new Set()) + ids.add(trace.callId) + push(sessionId, trace) + } else if (event.type === 'tool/result') { + const trace = traceOf(event) + if (trace?.kind !== 'result') return + if (!callIds.get(sessionId)?.has(trace.callId)) return + push(sessionId, trace) + } + }) + ctx.effect(() => dispose, 'dsh-better-sidebar: task-output event mirror') + + const push = (sessionId: string, trace: TaskOutputTrace): void => { + let list = perSession.get(sessionId) + if (list === undefined) perSession.set(sessionId, list = []) + list.push(trace) + if (list.length > MIRROR_MAX_ENTRIES) { + const removed = list.splice(0, list.length - MIRROR_MAX_ENTRIES) + const ids = callIds.get(sessionId) + if (ids !== undefined) { + for (const entry of removed) { + if (entry.kind === 'call') ids.delete(entry.callId) + } + if (ids.size === 0) callIds.delete(sessionId) + } + } + } + + return { entries: (sessionId) => perSession.get(sessionId) ?? [] } +} + +/** + * Build the tasks routes bound to the plugin context. `output` merges the + * owner session's own event log with the live task_output mirror; `kill` + * reads the tasks/agents services lazily and degrades to a 503 when the + * deployment lacks the registry. * @param ctx - host plugin context. * @param outputLimit - response cap for one output replay in bytes; longer * texts are sliced and flagged `truncated` (mirrors the fs.read cap). @@ -81,6 +198,7 @@ function isNoNewOutput(text: string): boolean { export function buildTasksApi(ctx: Context, outputLimit: number): SidebarTasksRoutes { const tasks = ctx.get('tasks') const agents = ctx.get('agents') + const mirror = createTaskOutputMirror(ctx) /** The live caller whose session id the registry fence compares against. */ const callerOf = (sessionId: string) => agents?.get(sessionId) /** Registry refusals become a 404 task-error; unknown and foreign ids are indistinguishable. */ @@ -90,32 +208,26 @@ export function buildTasksApi(ctx: Context, outputLimit: number): SidebarTasksRo output(payload) { const sessionId = requireString(payload, 'sessionId') const id = requireString(payload, 'id') - // One pass over the owner session's event log: tool/call rows first - // map callId → task_id (a call always precedes its result), then each - // matching tool/result contributes the finalized text the model saw. - const session = ctx.sessions.get(sessionId) + // Merge the store's event log (durable seed + whatever it received) + // with the live mirror, deduped by seq — a trace never double-counts. + const bySeq = new Map() + for (const event of ctx.sessions.get(sessionId)?.events ?? []) { + const trace = traceOf(event) + if (trace !== undefined) bySeq.set(trace.seq, trace) + } + for (const trace of mirror.entries(sessionId)) bySeq.set(trace.seq, trace) + // Pair calls with results in seq order: the model's reads, oldest first. const taskOf = new Map() const parts: string[] = [] let read = false - for (const event of session?.events ?? []) { - if (event.type === 'tool/call') { - const data = event.data as { name?: unknown; callId?: unknown; arguments?: unknown } - if (data.name !== 'task_output' || typeof data.callId !== 'string') continue - try { - const args = JSON.parse(typeof data.arguments === 'string' ? data.arguments : '') as { task_id?: unknown } - if (typeof args.task_id === 'string') taskOf.set(data.callId, args.task_id) - } catch { - // Malformed model arguments: not a task_output pair. - } - } else if (event.type === 'tool/result') { - const message = (event.data as { message?: unknown }).message as ToolResultMessageLike | undefined - if (message === undefined) continue - const callId = message.source?.callId - if (typeof callId !== 'string') continue - if (taskOf.get(callId) !== id) continue + for (const trace of [...bySeq.values()].sort((left, right) => left.seq - right.seq)) { + if (trace.kind === 'call') { + if (trace.taskId !== undefined) taskOf.set(trace.callId, trace.taskId) + } else if (taskOf.get(trace.callId) === id) { read = true - const text = resultText(message) - if (text !== undefined && !isNoNewOutput(text)) parts.push(text) + if (trace.isError !== true && trace.text !== undefined && !isNoNewOutput(trace.text)) { + parts.push(trace.text) + } } } const text = parts.join('\n') diff --git a/tests/tasks-routes.spec.ts b/tests/tasks-routes.spec.ts index 211b64c..e1f67ee 100644 --- a/tests/tasks-routes.spec.ts +++ b/tests/tasks-routes.spec.ts @@ -19,6 +19,28 @@ function ctxWith(sessions: unknown, tasks: unknown, agents: unknown): Context { } as unknown as Context } +/** A context that additionally captures the session/event listener (live mirror). */ +function ctxWithFeed(sessions: unknown): { + ctx: Context + emit: (session: unknown, event: SidebarSessionEvent) => void +} { + let listener: ((session: unknown, event: SidebarSessionEvent) => void) | undefined + const base = ctxWith(sessions, undefined, undefined) as unknown as { + on: (event: string, fn: (session: unknown, event: SidebarSessionEvent) => void) => () => void + effect: (fn: () => void | (() => void)) => void + } + base.on = (_event: string, fn) => { + listener = fn + return () => { if (listener === fn) listener = undefined } + } + // The vendored cordis runs the registration effect immediately. + base.effect = (fn) => { fn() } + return { + ctx: base as unknown as Context, + emit: (session, event) => { listener?.(session, event) }, + } +} + /** A stub live agent (the fence compares `id` only). */ const agent = (id: string) => ({ id, session: { header: { cwd: '/p' } } }) @@ -97,6 +119,55 @@ describe('tasks.output route (event replay)', () => { expect(tasks.kill).not.toHaveBeenCalled() }) + it('mirrors live task_output events the store log misses (restart divergence)', () => { + // The store session is frozen at its rehydration boundary (no events); + // the read exists only on the live session/event feed. + const { ctx, emit } = ctxWithFeed({ get: () => session([]) }) + const api = buildTasksApi(ctx, 512 * 1024) + expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({ text: '', truncated: false, read: false }) + + // The live feed delivers the task_output call and its result. + emit({ id: 's1' }, taskOutputCall(100, 'c-live', 'bash-1')) + emit({ id: 's1' }, taskOutputResult(101, 'c-live', 'live line\n[status: completed, exit code: 0]')) + expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({ + text: 'live line\n[status: completed, exit code: 0]', + truncated: false, + read: true, + }) + + // Unrelated results are never cached (no task_output call paired them), + // and another session's feed does not leak into this one. + emit({ id: 's1' }, taskOutputResult(102, 'c-other', 'unpaired line')) + emit({ id: 's2' }, taskOutputCall(103, 'c-s2', 'bash-1')) + emit({ id: 's2' }, taskOutputResult(104, 'c-s2', 'other session line')) + expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({ + text: 'live line\n[status: completed, exit code: 0]', + truncated: false, + read: true, + }) + expect(api.output({ sessionId: 's2', id: 'bash-1' })).toEqual({ + text: 'other session line', + truncated: false, + read: true, + }) + }) + + it('merges store-log traces with live mirror traces without double-counting', () => { + // A seed read (seq 5) in the store log plus a live read (seq 106) that + // ALSO reached the store log would double-count — the seq dedupe keeps + // exactly one copy of each. + const events = [taskOutputCall(5, 'c-seed', 'bash-1'), taskOutputResult(6, 'c-seed', 'seed line')] + const { ctx, emit } = ctxWithFeed({ get: () => session(events) }) + const api = buildTasksApi(ctx, 512 * 1024) + emit({ id: 's1' }, taskOutputCall(106, 'c-live', 'bash-1')) + emit({ id: 's1' }, taskOutputResult(107, 'c-live', 'live line')) + expect(api.output({ sessionId: 's1', id: 'bash-1' })).toEqual({ + text: 'seed line\nlive line', + truncated: false, + read: true, + }) + }) + it('caps oversized replays with the truncated flag', () => { const events = [taskOutputCall(0, 'c1', 'bash-1'), taskOutputResult(1, 'c1', 'x'.repeat(10_000))] const api = buildTasksApi(ctxWith({ get: () => session(events) }, undefined, undefined), 100)