fix: 任务输出回放合并活事件镜像,修复重启后 store 会话滞后导致新读取不可见

- host 半挂 ctx.on('session/event')(api-proxy 同款 API),镜像 task_output
  的 tool/call + tool/result 活事件(每会话 200 条有界环,无关结果不缓存)
- tasks.output 回放合并「store 事件日志 + 活事件镜像」,按 seq 去重排序配对
- 零 DSH 写入:会话 store 在宿主重启后可能冻结在再水合边界(持久化文件
  停在重启时刻),活事件只走 append 推送链;镜像绕开该滞后
- 测试:新增镜像用例(live 事件、跨会话隔离、种子+镜像合并不双计),
  全量 373 通过
This commit is contained in:
Menghuan1918
2026-08-12 23:04:30 +08:00
parent 88f116338a
commit 6e4474cd2e
3 changed files with 226 additions and 34 deletions
+9
View File
@@ -388,6 +388,15 @@ declare module 'cordis' {
* {@link ./client/index.tsx}); undefined on the host side. * {@link ./client/index.tsx}); undefined on the host side.
*/ */
betterSidebar: BetterSidebarService 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 * Register a lifecycle callback (DSH-vendored cordis): runs at plugin
* activation; its returned cleanup runs at disposal. * activation; its returned cleanup runs at disposal.
+146 -34
View File
@@ -4,18 +4,21 @@
* harness's `session/tasks` push mirror (`tasksBySession` in the sessions * harness's `session/tasks` push mirror (`tasksBySession` in the sessions
* list feed). The routes: * list feed). The routes:
* *
* - 'tasks.output' — REPLAYS the output the MODEL has read so far from the * - 'tasks.output' — REPLAYS the output the MODEL has read so far for one
* owner session's OWN event log: `tool/call` rows of `task_output` name * task. The source is the owner session's own event log: `tool/call` rows
* the task via `arguments.task_id`, and the paired `tool/result` rows * of `task_output` name the task via `arguments.task_id`, and the paired
* carry the finalized content the model actually received. This touches * `tool/result` rows carry the finalized content the model received.
* NO DSH source (no peek seam, no consuming read): the model's * Because the session store's in-memory log can lag the live append feed
* `task_output` cursor is untouched by construction, and the pane stays * after a host restart (the store session stays frozen at its
* empty until the agent reads the task. * 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), * - 'tasks.kill' — the registry's stock `kill` (a pristine DSH API),
* fenced by the owning session via the live agent caller. Absent registry * fenced by the owning session via the live agent caller. Absent registry
* → 503, mirroring the settings routes' optional-service downgrade. * → 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' import { requireString, SidebarError } from './wire.ts'
/** The two background-task routes of the sidebar API. */ /** 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) { for (const block of message.content) {
if (block === null || typeof block !== 'object') continue if (block === null || typeof block !== 'object') continue
const candidate = block as ToolResultBlockLike const candidate = block as ToolResultBlockLike
if (candidate.type !== 'tool-result' || candidate.isError === true) continue if (candidate.type !== 'tool-result') continue
const inner = candidate.content const inner = candidate.content
if (!Array.isArray(inner)) continue if (!Array.isArray(inner)) continue
for (const item of inner) { for (const item of inner) {
@@ -64,16 +67,130 @@ function resultText(message: ToolResultMessageLike): string | undefined {
return parts.length > 0 ? parts.join('\n') : 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 /** Whether a task_output result carries no new output — the controller's
* model-facing "(no new output)" body, noise for the human pane. */ * model-facing "(no new output)" body, noise for the human pane. */
function isNoNewOutput(text: string): boolean { function isNoNewOutput(text: string): boolean {
return text.startsWith('(no new output)') 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 live task_output mirror: subscribes to the session append feed and
* the session store; `kill` reads the tasks/agents services lazily and * caches the task_output traces the session store's own log can lag behind
* degrades to a 503 when the deployment lacks the registry. * (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<string, TaskOutputTrace[]>()
// 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<string, Set<string>>()
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 ctx - host plugin context.
* @param outputLimit - response cap for one output replay in bytes; longer * @param outputLimit - response cap for one output replay in bytes; longer
* texts are sliced and flagged `truncated` (mirrors the fs.read cap). * 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 { export function buildTasksApi(ctx: Context, outputLimit: number): SidebarTasksRoutes {
const tasks = ctx.get('tasks') const tasks = ctx.get('tasks')
const agents = ctx.get('agents') const agents = ctx.get('agents')
const mirror = createTaskOutputMirror(ctx)
/** The live caller whose session id the registry fence compares against. */ /** The live caller whose session id the registry fence compares against. */
const callerOf = (sessionId: string) => agents?.get(sessionId) const callerOf = (sessionId: string) => agents?.get(sessionId)
/** Registry refusals become a 404 task-error; unknown and foreign ids are indistinguishable. */ /** 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) { output(payload) {
const sessionId = requireString(payload, 'sessionId') const sessionId = requireString(payload, 'sessionId')
const id = requireString(payload, 'id') const id = requireString(payload, 'id')
// One pass over the owner session's event log: tool/call rows first // Merge the store's event log (durable seed + whatever it received)
// map callId → task_id (a call always precedes its result), then each // with the live mirror, deduped by seq — a trace never double-counts.
// matching tool/result contributes the finalized text the model saw. const bySeq = new Map<number, TaskOutputTrace>()
const session = ctx.sessions.get(sessionId) 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<string, string>() const taskOf = new Map<string, string>()
const parts: string[] = [] const parts: string[] = []
let read = false let read = false
for (const event of session?.events ?? []) { for (const trace of [...bySeq.values()].sort((left, right) => left.seq - right.seq)) {
if (event.type === 'tool/call') { if (trace.kind === 'call') {
const data = event.data as { name?: unknown; callId?: unknown; arguments?: unknown } if (trace.taskId !== undefined) taskOf.set(trace.callId, trace.taskId)
if (data.name !== 'task_output' || typeof data.callId !== 'string') continue } else if (taskOf.get(trace.callId) === id) {
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
read = true read = true
const text = resultText(message) if (trace.isError !== true && trace.text !== undefined && !isNoNewOutput(trace.text)) {
if (text !== undefined && !isNoNewOutput(text)) parts.push(text) parts.push(trace.text)
}
} }
} }
const text = parts.join('\n') const text = parts.join('\n')
+71
View File
@@ -19,6 +19,28 @@ function ctxWith(sessions: unknown, tasks: unknown, agents: unknown): Context {
} as unknown as 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). */ /** A stub live agent (the fence compares `id` only). */
const agent = (id: string) => ({ id, session: { header: { cwd: '/p' } } }) const agent = (id: string) => ({ id, session: { header: { cwd: '/p' } } })
@@ -97,6 +119,55 @@ describe('tasks.output route (event replay)', () => {
expect(tasks.kill).not.toHaveBeenCalled() 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', () => { it('caps oversized replays with the truncated flag', () => {
const events = [taskOutputCall(0, 'c1', 'bash-1'), taskOutputResult(1, 'c1', 'x'.repeat(10_000))] const events = [taskOutputCall(0, 'c1', 'bash-1'), taskOutputResult(1, 'c1', 'x'.repeat(10_000))]
const api = buildTasksApi(ctxWith({ get: () => session(events) }, undefined, undefined), 100) const api = buildTasksApi(ctxWith({ get: () => session(events) }, undefined, undefined), 100)