From 65e0d2df659b43a59e279e3e2cb12725efeeab86 Mon Sep 17 00:00:00 2001 From: testikun Date: Fri, 4 Sep 2026 16:33:15 +0800 Subject: [PATCH 1/4] feat(desktop): show Agent Graph output previews Project bounded child output and usage-derived throughput from existing RuntimeEvent facts into the Agent Graph read model, then render streaming and completed previews without changing graph execution semantics. Generated-by: OpenAI Codex --- .../main/__tests__/agent-graph-panel.test.ts | 87 ++++++++ .../src/renderer/agent-graph-panel.tsx | 25 +++ .../src/renderer/styles/agent-graph.css | 13 ++ .../stories/agent-graph-panel.stories.tsx | 58 ++++++ .../__tests__/agent-graph-coordinator.test.ts | 25 +++ .../__tests__/agent-graph-protocol.test.ts | 24 +++ .../runtime-host/src/protocol/agent-graph.ts | 76 ++++++- packages/runtime-host/src/protocol/index.ts | 4 +- .../src/server/agent-graph-coordinator.ts | 23 ++ .../__tests__/stream-graph-projection.test.ts | 60 ++++++ .../__tests__/stream-graph-read-model.test.ts | 113 ++++++++++ .../runtime/src/stream-graph-coordinator.ts | 17 +- .../runtime/src/stream-graph-projection.ts | 109 ++++++++++ .../runtime/src/stream-graph-read-model.ts | 197 ++++++++++++++++-- 14 files changed, 801 insertions(+), 30 deletions(-) diff --git a/apps/desktop/src/main/__tests__/agent-graph-panel.test.ts b/apps/desktop/src/main/__tests__/agent-graph-panel.test.ts index 18b1690317..9ac0f679f3 100644 --- a/apps/desktop/src/main/__tests__/agent-graph-panel.test.ts +++ b/apps/desktop/src/main/__tests__/agent-graph-panel.test.ts @@ -101,6 +101,30 @@ function snapshot( }; } +function operator( + overrides: Pick & + Partial, +): AgentGraphClientSnapshot['operators'][number] { + return { + childSessionId: `child-${overrides.operatorId}`, + provisionId: `provision-${overrides.operatorId}`, + agentId: 'reviewer', + provisionedAt: 1, + inboundEdgeIds: [], + outboundEdgeIds: [], + scheduledWorkIds: [], + readiness: [], + omitted: { + inboundEdgeIds: 0, + outboundEdgeIds: 0, + scheduledWorkIds: 0, + readiness: 0, + readinessWaits: 0, + }, + ...overrides, + }; +} + function installGraphRenderer( initial: AgentGraphClientSnapshot, historical: readonly AgentGraphClientSnapshot[] = [], @@ -308,6 +332,69 @@ async function renderPanel( } describe('AgentGraphPanel dismiss', () => { + it('renders bounded live and completed output facts without replacing child navigation', async () => { + const opened: string[] = []; + const source = snapshot({ + graphId: 'graph-output', + status: 'active', + operators: [ + operator({ + operatorId: 'operator-live', + status: 'running', + output: { + activationId: 'run-live', + preview: 'Inspecting the renderer projection', + previewTruncated: true, + phase: 'streaming', + previewUpdatedAt: 2_000, + sourceEventId: 'event-live', + messageId: 'message-live', + sampleStartedAt: 1_000, + outputTokens: 21, + sampleDurationMs: 1_000, + tokensPerSecond: 21, + }, + }), + operator({ + operatorId: 'operator-done', + status: 'completed', + output: { + activationId: 'run-done', + preview: 'Projection verified', + previewTruncated: false, + phase: 'completed', + previewUpdatedAt: 3_000, + sourceEventId: 'event-done', + sampleStartedAt: 2_000, + }, + }), + ], + }); + const harness = installGraphRenderer(source); + await act(async () => { + harness.root.render( + createElement(AgentGraphPanel, { + rootSessionId: 'session-1', + enabled: true, + locale: 'en', + onOpenSession: (sessionId: string) => opened.push(sessionId), + }), + ); + await Promise.resolve(); + }); + + assert.match(harness.container.textContent ?? '', /Live output · 21\.0 token\/s/); + assert.match(harness.container.textContent ?? '', /Inspecting the renderer projection…/); + assert.match(harness.container.textContent ?? '', /Result preview/); + const open = [...harness.container.querySelectorAll('button')].find((button) => + button.textContent?.includes('Open child task'), + ); + assert.ok(open); + await act(async () => (open as HTMLElement).click()); + assert.deepEqual(opened, ['child-operator-live']); + await act(async () => harness.root.unmount()); + }); + it('keeps the new session loading when a disposed read settles later', async () => { const sessionA = snapshot({ graphId: 'graph-a', status: 'active' }); const sessionB = snapshot({ diff --git a/apps/desktop/src/renderer/agent-graph-panel.tsx b/apps/desktop/src/renderer/agent-graph-panel.tsx index 48a3bbcda6..51fe366be1 100644 --- a/apps/desktop/src/renderer/agent-graph-panel.tsx +++ b/apps/desktop/src/renderer/agent-graph-panel.tsx @@ -64,6 +64,9 @@ type GraphPanelCopy = { openSession: string; operators: string; selectedResults: string; + liveOutput: string; + completedOutput: string; + throughput(tokensPerSecond: number): string; epoch: string; currentEpoch: string; historicalEpoch: string; @@ -92,6 +95,9 @@ export function getAgentGraphPanelCopy(locale: UiLocale): GraphPanelCopy { openSession: '打开子任务', operators: 'Operators', selectedResults: '已选择结果', + liveOutput: '实时输出', + completedOutput: '结果预览', + throughput: (tokensPerSecond) => `${tokensPerSecond.toFixed(1)} token/s`, epoch: 'Graph 运行轮次', currentEpoch: '当前', historicalEpoch: '历史记录（只读）', @@ -187,6 +193,9 @@ export function getAgentGraphPanelCopy(locale: UiLocale): GraphPanelCopy { openSession: 'Open child task', operators: 'Operators', selectedResults: 'Selected results', + liveOutput: 'Live output', + completedOutput: 'Result preview', + throughput: (tokensPerSecond) => `${tokensPerSecond.toFixed(1)} token/s`, epoch: 'Graph run', currentEpoch: 'Current', historicalEpoch: 'History (read-only)', @@ -537,6 +546,22 @@ export function AgentGraphPanel(props: { {operator.agentId} {work?.instructionPreview ?? operator.operatorId} + {operator.output ? ( + + + {operator.output.phase === 'streaming' + ? copy.liveOutput + : copy.completedOutput} + {operator.output.tokensPerSecond === undefined + ? null + : ` · ${copy.throughput(operator.output.tokensPerSecond)}`} + + + {operator.output.preview} + {operator.output.previewTruncated ? '…' : ''} + + + ) : null} {wait ? {wait} : null} diff --git a/apps/desktop/src/renderer/styles/agent-graph.css b/apps/desktop/src/renderer/styles/agent-graph.css index cd4e0640a9..9668401f3e 100644 --- a/apps/desktop/src/renderer/styles/agent-graph.css +++ b/apps/desktop/src/renderer/styles/agent-graph.css @@ -180,6 +180,19 @@ font: var(--maka-text-supporting); } +.maka-agent-graph-output { + display: grid; + gap: var(--space-0-5); +} + +.maka-agent-graph-output-meta { + color: var(--muted-foreground); +} + +.maka-agent-graph-output-preview { + color: var(--foreground); +} + .maka-agent-graph-wait { color: var(--warning-text); } diff --git a/apps/desktop/stories/agent-graph-panel.stories.tsx b/apps/desktop/stories/agent-graph-panel.stories.tsx index cb60e7647c..dcd3f276f4 100644 --- a/apps/desktop/stories/agent-graph-panel.stories.tsx +++ b/apps/desktop/stories/agent-graph-panel.stories.tsx @@ -237,6 +237,64 @@ export const BlockedOnUpstream: Story = { }, }; +// Real path: one child task is streaming while another has settled. The graph +// read model supplies bounded text and provider-reported throughput; the panel +// never reads either child Session independently. +export const OutputPreviews: Story = { + decorators: [ + withScopedMakaBridge( + graphBridge( + snapshot({ + status: 'active', + operators: [ + operator({ + operatorId: 'op-research', + status: 'running', + output: { + activationId: 'run-research', + preview: 'Comparing the three provider adapters and their retry boundaries…', + previewTruncated: false, + phase: 'streaming', + previewUpdatedAt: 2_000, + sourceEventId: 'event-research', + messageId: 'message-research', + sampleStartedAt: 1_000, + outputTokens: 42, + sampleDurationMs: 1_000, + tokensPerSecond: 42, + }, + }), + operator({ + operatorId: 'op-review', + status: 'completed', + output: { + activationId: 'run-review', + preview: 'Found one retry guard that needs a focused regression test.', + previewTruncated: false, + phase: 'completed', + previewUpdatedAt: 3_000, + sourceEventId: 'event-review', + sampleStartedAt: 2_000, + outputTokens: 18, + sampleDurationMs: 1_000, + tokensPerSecond: 18, + }, + }), + ], + }), + ), + ), + ], + render: panel, + play: async ({ canvasElement }) => { + await waitFor(() => { + expect(canvasElement.textContent).toContain(copy.liveOutput); + expect(canvasElement.textContent).toContain(copy.completedOutput); + expect(canvasElement.textContent).toContain(copy.throughput(42)); + }); + }, +}; + // Real path: a wide fan-out — many operators provisioned at once, the breadth a // small graph never shows. The read-model only elides operators past 256 (far // beyond a story's scale), so this shows a genuine many-operator list rather diff --git a/packages/runtime-host/src/__tests__/agent-graph-coordinator.test.ts b/packages/runtime-host/src/__tests__/agent-graph-coordinator.test.ts index b719b4533a..9199593416 100644 --- a/packages/runtime-host/src/__tests__/agent-graph-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/agent-graph-coordinator.test.ts @@ -343,6 +343,19 @@ describe('Host Agent Graph coordinator', () => { test('projects only allowlisted Runtime fields onto the wire', () => { const source = graphSnapshot(); Object.assign(source.operators[0]!, { privatePrompt: 'operator-secret' }); + source.operators[0]!.output = { + activationId: 'activation-0', + preview: 'Bounded output', + previewTruncated: false, + phase: 'streaming', + previewUpdatedAt: 10, + sourceEventId: 'event-0', + sampleStartedAt: 1, + outputTokens: 18, + sampleDurationMs: 9, + tokensPerSecond: 2_000, + }; + Object.assign(source.operators[0]!.output, { privateOutput: 'output-secret' }); Object.assign(source.operators[0]!.readiness[0]!, { policyKind: 'map', privatePolicy: 'readiness-secret', @@ -351,6 +364,18 @@ describe('Host Agent Graph coordinator', () => { const projected = projectAgentGraphClientSnapshot(source); assert.equal(JSON.stringify(projected).includes('secret'), false); + assert.deepEqual(projected.operators[0]?.output, { + activationId: 'activation-0', + preview: 'Bounded output', + previewTruncated: false, + phase: 'streaming', + previewUpdatedAt: 10, + sourceEventId: 'event-0', + sampleStartedAt: 1, + outputTokens: 18, + sampleDurationMs: 9, + tokensPerSecond: 2_000, + }); assert.doesNotThrow(() => decodeAgentGraphClientSnapshot(projected)); }); diff --git a/packages/runtime-host/src/__tests__/agent-graph-protocol.test.ts b/packages/runtime-host/src/__tests__/agent-graph-protocol.test.ts index a66a4bbfbf..3ff4b2391d 100644 --- a/packages/runtime-host/src/__tests__/agent-graph-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/agent-graph-protocol.test.ts @@ -142,6 +142,17 @@ describe('Agent Graph Client protocol', () => { operators: [{ ...snapshot.operators[0], privatePrompt: 'secret' }], }), ); + assertInvalid(() => + decodeAgentGraphClientSnapshot({ + ...snapshot, + operators: [ + { + ...snapshot.operators[0], + output: { ...snapshot.operators[0]!.output, tokensPerSecond: -1 }, + }, + ], + }), + ); assertInvalid(() => decodeAgentGraphClientSnapshot({ ...snapshot, @@ -267,6 +278,19 @@ function graphSnapshot(): AgentGraphClientSnapshot { lastEventTime: 10, run: { sessionId: 'child-1', agentRunId: 'run:1', turnId: 'turn:1' }, }, + output: { + activationId: 'activation:1', + preview: 'Reviewing the graph projection.', + previewTruncated: false, + phase: 'streaming', + previewUpdatedAt: 10, + sourceEventId: 'event:1', + messageId: 'message:1', + sampleStartedAt: 1, + outputTokens: 18, + sampleDurationMs: 9, + tokensPerSecond: 2_000, + }, }, ], edges: [{ edgeId: 'edge:1', fromOperatorId: 'operator:0', toOperatorId: 'operator:1' }], diff --git a/packages/runtime-host/src/protocol/agent-graph.ts b/packages/runtime-host/src/protocol/agent-graph.ts index 02c9842c00..26de6efb2a 100644 --- a/packages/runtime-host/src/protocol/agent-graph.ts +++ b/packages/runtime-host/src/protocol/agent-graph.ts @@ -54,6 +54,7 @@ export const AGENT_GRAPH_MAX_INSPECTION_RECORDS = 32; export const AGENT_GRAPH_EPOCH_PAGE_SIZE = 32; const AGENT_GRAPH_INSTRUCTION_PREVIEW_MAX_BYTES = 2 * 1024; +const AGENT_GRAPH_OUTPUT_PREVIEW_MAX_BYTES = 2 * 1024; const AGENT_GRAPH_REASON_MAX_BYTES = 12 * 1024; const AGENT_GRAPH_RECONCILIATION_FAILURE_REASON_MAX_BYTES = 4 * 1024; @@ -177,6 +178,21 @@ export interface AgentGraphClientOperator { readonly terminalRecordId?: string; readonly run: AgentGraphClientRunRef; }; + readonly output?: AgentGraphClientOperatorOutput; +} + +export interface AgentGraphClientOperatorOutput { + readonly activationId: string; + readonly preview: string; + readonly previewTruncated: boolean; + readonly phase: 'streaming' | 'completed'; + readonly previewUpdatedAt: number; + readonly sourceEventId: string; + readonly messageId?: string; + readonly sampleStartedAt: number; + readonly outputTokens?: number; + readonly sampleDurationMs?: number; + readonly tokensPerSecond?: number; } export interface AgentGraphClientEdge { @@ -759,7 +775,7 @@ function decodeOperator(value: unknown): AgentGraphClientOperator { 'readiness', 'omitted', ], - ['currentActivation'], + ['currentActivation', 'output'], ); return { operatorId: requireOpaqueIdentity(record.operatorId, 'operatorId'), @@ -793,6 +809,57 @@ function decodeOperator(value: unknown): AgentGraphClientOperator { ...(record.currentActivation === undefined ? {} : { currentActivation: decodeCurrentActivation(record.currentActivation) }), + ...(record.output === undefined ? {} : { output: decodeOperatorOutput(record.output) }), + }; +} + +function decodeOperatorOutput(value: unknown): AgentGraphClientOperatorOutput { + const record = requireShapedRecord( + value, + 'agent graph operator output', + [ + 'activationId', + 'preview', + 'previewTruncated', + 'phase', + 'previewUpdatedAt', + 'sourceEventId', + 'sampleStartedAt', + ], + ['messageId', 'outputTokens', 'sampleDurationMs', 'tokensPerSecond'], + ); + if (record.phase !== 'streaming' && record.phase !== 'completed') { + throw invalidProtocolFrame('Invalid agent graph operator output phase'); + } + return { + activationId: requireOpaqueIdentity(record.activationId, 'activationId'), + preview: requireUtf8String( + record.preview, + 'operator output preview', + AGENT_GRAPH_OUTPUT_PREVIEW_MAX_BYTES, + ), + previewTruncated: requireBoolean(record.previewTruncated, 'previewTruncated'), + phase: record.phase, + previewUpdatedAt: requireCount(record.previewUpdatedAt, 'previewUpdatedAt'), + sourceEventId: requireOpaqueIdentity(record.sourceEventId, 'sourceEventId'), + ...(record.messageId === undefined + ? {} + : { messageId: requireOpaqueIdentity(record.messageId, 'messageId') }), + sampleStartedAt: requireCount(record.sampleStartedAt, 'sampleStartedAt'), + ...(record.outputTokens === undefined + ? {} + : { outputTokens: requireCount(record.outputTokens, 'outputTokens') }), + ...(record.sampleDurationMs === undefined + ? {} + : { sampleDurationMs: requireCount(record.sampleDurationMs, 'sampleDurationMs') }), + ...(record.tokensPerSecond === undefined + ? {} + : { + tokensPerSecond: requireNonNegativeFiniteNumber( + record.tokensPerSecond, + 'tokensPerSecond', + ), + }), }; } @@ -1348,6 +1415,13 @@ function requireBoolean(value: unknown, label: string): boolean { return value; } +function requireNonNegativeFiniteNumber(value: unknown, label: string): number { + if (typeof value !== 'number' || !Number.isFinite(value) || value < 0) { + throw invalidProtocolFrame(`Invalid ${label}`); + } + return value; +} + function requireOpaqueIdentity(value: unknown, label: string): string { if ( typeof value !== 'string' || diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index b58467c6ee..108a4fcc92 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -101,7 +101,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 112 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 113 as const; +// 113: Agent Graph operator projections may carry bounded output previews and +// usage-derived throughput. Older peers reject the strict projection shape. // 112: Owners can query the Host execution environment through an extensible, // bounded resource-envelope contract. Older Hosts do not implement the query. // 111: Client Capability tool schemas may use draft-07 tuple additionalItems. diff --git a/packages/runtime-host/src/server/agent-graph-coordinator.ts b/packages/runtime-host/src/server/agent-graph-coordinator.ts index 6060b12631..3d311903e4 100644 --- a/packages/runtime-host/src/server/agent-graph-coordinator.ts +++ b/packages/runtime-host/src/server/agent-graph-coordinator.ts @@ -408,6 +408,29 @@ function projectOperator( }, } : {}), + ...(operator.output + ? { + output: { + activationId: operator.output.activationId, + preview: operator.output.preview, + previewTruncated: operator.output.previewTruncated, + phase: operator.output.phase, + previewUpdatedAt: operator.output.previewUpdatedAt, + sourceEventId: operator.output.sourceEventId, + sampleStartedAt: operator.output.sampleStartedAt, + ...(operator.output.messageId ? { messageId: operator.output.messageId } : {}), + ...(operator.output.outputTokens !== undefined + ? { outputTokens: operator.output.outputTokens } + : {}), + ...(operator.output.sampleDurationMs !== undefined + ? { sampleDurationMs: operator.output.sampleDurationMs } + : {}), + ...(operator.output.tokensPerSecond !== undefined + ? { tokensPerSecond: operator.output.tokensPerSecond } + : {}), + }, + } + : {}), }; } diff --git a/packages/runtime/src/__tests__/stream-graph-projection.test.ts b/packages/runtime/src/__tests__/stream-graph-projection.test.ts index 751e6f93dd..b1046f7a7d 100644 --- a/packages/runtime/src/__tests__/stream-graph-projection.test.ts +++ b/packages/runtime/src/__tests__/stream-graph-projection.test.ts @@ -210,6 +210,66 @@ describe('committed stream graph projection', () => { assert.doesNotMatch(JSON.stringify(projection.records), /mutable-stream-chunk/); }); + test('derives bounded operator output and actual throughput from immutable runtime facts', async () => { + const run = runInvocation({ + sessionId: 'child-output', + runId: 'run-output', + turnId: 'turn-output', + status: 'completed', + createdAt: baseTs, + }); + const text = `Summary ${'界'.repeat(400)}`; + const projection = await readCommittedAgentGraphProjection({ + graphId: 'graph-output', + operators: [{ operatorId: 'writer', sessionId: run.sessionId }], + runtimeEventStore: { + async listSessionInvocations() { + return [run]; + }, + async readImmutableRuntimeEvents() { + return [ + runtimeEvent(run, { + id: 'output-start', + ts: baseTs + 500, + partial: true, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'Summary ' }, + }), + runtimeEvent(run, { + id: 'output-text', + ts: baseTs + 1_000, + role: 'model', + author: 'agent', + content: { kind: 'text', text }, + }), + runtimeEvent(run, { + id: 'output-usage', + ts: baseTs + 5_000, + actions: { tokenUsage: { input: 100, output: 90 } }, + }), + runtimeEvent(run, { + id: 'output-complete', + ts: baseTs + 5_001, + status: 'completed', + actions: { endInvocation: true }, + }), + ]; + }, + }, + }); + + const output = projection.operatorOutputs?.[0]; + assert.ok(output); + assert.equal(Array.from(output.preview).length, 280); + assert.equal(output.previewTruncated, true); + assert.equal(output.phase, 'completed'); + assert.equal(output.outputTokens, 90); + assert.equal(output.sampleDurationMs, 4_500); + assert.equal(output.tokensPerSecond, 20); + assert.equal(output.sourceEventId, 'output-text'); + }); + test('replay is deterministic for reordered delivery and idempotent duplicates', () => { const run = runInvocation({ sessionId: 'child-a', diff --git a/packages/runtime/src/__tests__/stream-graph-read-model.test.ts b/packages/runtime/src/__tests__/stream-graph-read-model.test.ts index 3f37466605..202b9c5675 100644 --- a/packages/runtime/src/__tests__/stream-graph-read-model.test.ts +++ b/packages/runtime/src/__tests__/stream-graph-read-model.test.ts @@ -376,6 +376,7 @@ describe('agent graph client read model', () => { undefined, ); + assert.ok(forward.activity); const canonicalRecords = forward.snapshot.recentActivity.map((activity, index) => graphRecordFromActivity(graphId, activity, index), ); @@ -401,6 +402,9 @@ describe('agent graph client read model', () => { }; canonicalObservation.projection.state.operators[operatorId]!.currentActivationId = 'run-0'; canonicalObservation.projection.state.operators[operatorId]!.status = 'completed'; + const forwardOutput = forward.operator.operator.output; + assert.ok(forwardOutput); + canonicalObservation.projection.operatorOutputs = [{ operatorId, ...forwardOutput }]; const rebuilt = materializeAgentGraphClientProjection({ ...baseInput, observation: canonicalObservation, @@ -419,6 +423,92 @@ describe('agent graph client read model', () => { assert.equal(rebuilt.snapshot.snapshotVersion, forward.snapshot.snapshotVersion); assert.deepEqual(rebuilt.snapshot.recentActivity, forward.snapshot.recentActivity); }); + + test('advances a bounded streaming preview and provider-reported TPS without activity records', () => { + const graphId = 'graph-output'; + const operatorId = 'operator-output'; + const childSessionId = 'child-output'; + const initial = materializeAgentGraphClientProjection( + runningInput(graphId, operatorId, childSessionId), + ); + const first = advanceMaterializedAgentGraphClientProjection( + initial.snapshot, + initial.operators[0]!, + outputRuntimeEvent(graphId, operatorId, childSessionId, { + id: 'delta-1', + type: 'text_delta', + ts: 1_000, + messageId: 'message-1', + startOffset: 0, + text: 'Reviewing ', + }), + false, + )!; + assert.equal(first.activity, undefined); + assert.equal(first.operator.operator.output?.preview, 'Reviewing '); + + const replayed = advanceMaterializedAgentGraphClientProjection( + first.snapshot, + first.operator, + outputRuntimeEvent(graphId, operatorId, childSessionId, { + id: 'delta-replay', + type: 'text_delta', + ts: 1_500, + messageId: 'message-1', + startOffset: 0, + text: 'Reviewing ', + }), + false, + )!; + assert.equal(replayed.operator.operator.output?.preview, 'Reviewing '); + + const second = advanceMaterializedAgentGraphClientProjection( + replayed.snapshot, + replayed.operator, + outputRuntimeEvent(graphId, operatorId, childSessionId, { + id: 'delta-2', + type: 'text_delta', + ts: 2_000, + messageId: 'message-1', + startOffset: 10, + text: 'projection', + }), + false, + )!; + assert.equal(second.operator.operator.output?.preview, 'Reviewing projection'); + + const usage = advanceMaterializedAgentGraphClientProjection( + second.snapshot, + second.operator, + outputRuntimeEvent(graphId, operatorId, childSessionId, { + id: 'usage-1', + type: 'token_usage', + ts: 3_000, + input: 80, + output: 20, + }), + false, + )!; + assert.equal(usage.activity?.facets[0], 'usage'); + assert.equal(usage.operator.operator.output?.outputTokens, 20); + assert.equal(usage.operator.operator.output?.sampleDurationMs, 2_000); + assert.equal(usage.operator.operator.output?.tokensPerSecond, 10); + assert.equal(usage.operator.operator.output?.sourceEventId, 'delta-2'); + + const settled = advanceMaterializedAgentGraphClientProjection( + usage.snapshot, + usage.operator, + outputRuntimeEvent(graphId, operatorId, childSessionId, { + id: 'complete-1', + type: 'complete', + ts: 4_000, + stopReason: 'end_turn', + }), + false, + )!; + assert.equal(settled.operator.operator.output?.phase, 'completed'); + assert.equal(settled.snapshot.operators[0]?.output?.phase, 'completed'); + }); }); function emptyInput(graphId: string) { @@ -658,6 +748,29 @@ function supervisorRuntimeEvent(input: { } as unknown as AgentGraphSupervisorRuntimeEvent; } +function outputRuntimeEvent( + graphId: string, + operatorId: string, + childSessionId: string, + event: Record, +): AgentGraphSupervisorRuntimeEvent { + return { + intent: { graphId }, + claim: { + targetOperatorId: operatorId, + targetSessionId: childSessionId, + targetRunId: 'run-0', + targetTurnId: 'turn-0', + }, + event: { + sessionId: childSessionId, + runId: 'run-0', + turnId: 'turn-0', + ...event, + }, + } as unknown as AgentGraphSupervisorRuntimeEvent; +} + function graphRecordFromActivity( graphId: string, activity: AgentGraphClientActivity, diff --git a/packages/runtime/src/stream-graph-coordinator.ts b/packages/runtime/src/stream-graph-coordinator.ts index 8ebb0885ca..e9961ae682 100644 --- a/packages/runtime/src/stream-graph-coordinator.ts +++ b/packages/runtime/src/stream-graph-coordinator.ts @@ -1169,13 +1169,15 @@ export class AgentGraphCoordinator { // stop race). Only the authoritative RuntimeEvent fold populates the // immutable terminal-history table. terminalActivities: [], - activityRecords: [ - { - recordId: advanced.activity.recordId, - eventTime: advanced.activity.eventTime, - }, - ], - incrementalRecordId: advanced.activity.recordId, + activityRecords: advanced.activity + ? [ + { + recordId: advanced.activity.recordId, + eventTime: advanced.activity.eventTime, + }, + ] + : [], + ...(advanced.activity ? { incrementalRecordId: advanced.activity.recordId } : {}), }); if (committed.snapshotVersion === advanced.snapshot.snapshotVersion) { this.#notifyClientChanged(driver, 'runtime_activity'); @@ -1701,7 +1703,6 @@ function isMaterializedGraphClientEvent( type: AgentGraphSupervisorRuntimeEvent['event']['type'], ): boolean { return ![ - 'text_delta', 'thinking_delta', 'tool_output_delta', 'tool_progress', diff --git a/packages/runtime/src/stream-graph-projection.ts b/packages/runtime/src/stream-graph-projection.ts index f0f42f6acf..c18063bdbc 100644 --- a/packages/runtime/src/stream-graph-projection.ts +++ b/packages/runtime/src/stream-graph-projection.ts @@ -202,6 +202,27 @@ export interface AgentGraphProjection { records: AgentGraphRecord[]; supervisorMetaStream: AgentGraphSupervisorMetaRecord[]; state: AgentGraphReplayState; + /** + * Bounded presentation facts derived from the same immutable RuntimeEvents. + * RuntimeEvents remain authoritative; this is not part of scheduling or + * supervisor replay. + */ + operatorOutputs?: AgentGraphOperatorOutputProjection[]; +} + +export interface AgentGraphOperatorOutputProjection { + operatorId: string; + activationId: string; + preview: string; + previewTruncated: boolean; + phase: 'streaming' | 'completed'; + previewUpdatedAt: number; + sourceEventId: string; + messageId?: string; + sampleStartedAt: number; + outputTokens?: number; + sampleDurationMs?: number; + tokensPerSecond?: number; } export interface ReadCommittedAgentGraphProjectionInput { @@ -287,10 +308,98 @@ export async function readCommittedAgentGraphProjectionWithRuns( records: projected.records, supervisorMetaStream: projected.supervisorMetaStream, state, + operatorOutputs: projectOperatorOutputs(streams), }, }; } +const AGENT_GRAPH_OUTPUT_PREVIEW_MAX_CODE_POINTS = 280; + +function projectOperatorOutputs( + streams: readonly AgentGraphRunStream[], +): AgentGraphOperatorOutputProjection[] { + const latestByOperator = new Map(); + for (const stream of streams) { + const current = latestByOperator.get(stream.operator.operatorId); + if ( + !current || + stream.run.openedAt > current.run.openedAt || + (stream.run.openedAt === current.run.openedAt && + compareAgentGraphIdentity(stream.run.runId, current.run.runId) > 0) + ) { + latestByOperator.set(stream.operator.operatorId, stream); + } + } + return [...latestByOperator.values()] + .map(projectOperatorOutput) + .filter((output): output is AgentGraphOperatorOutputProjection => output !== undefined) + .sort((left, right) => compareAgentGraphIdentity(left.operatorId, right.operatorId)); +} + +function projectOperatorOutput( + stream: AgentGraphRunStream, +): AgentGraphOperatorOutputProjection | undefined { + const orderedEvents = stream.events + .slice() + .sort((left, right) => left.ts - right.ts || compareAgentGraphIdentity(left.id, right.id)); + const events = orderedEvents.filter((event) => !event.partial); + const textEvents = events.filter( + (event) => event.role === 'model' && event.content?.kind === 'text', + ); + const latestText = textEvents.at(-1); + if (!latestText || latestText.content?.kind !== 'text') return undefined; + const preview = boundOutputPreview(latestText.content.text, 'head'); + if (!preview.text) return undefined; + const usageEvents = events.filter((event) => event.actions?.tokenUsage !== undefined); + const latestUsage = usageEvents.at(-1); + const outputTokens = latestUsage?.actions?.tokenUsage?.output; + const usageEndedAt = latestUsage?.ts; + const sampleStartedAt = + orderedEvents.find((event) => event.role === 'model' && event.content?.kind === 'text')?.ts ?? + textEvents[0]!.ts; + const sampleDurationMs = + outputTokens !== undefined && outputTokens > 0 && usageEndedAt !== undefined + ? Math.max(0, usageEndedAt - sampleStartedAt) + : undefined; + const tokensPerSecond = + outputTokens !== undefined && sampleDurationMs !== undefined && sampleDurationMs > 0 + ? roundTokensPerSecond((outputTokens * 1_000) / sampleDurationMs) + : undefined; + return { + operatorId: stream.operator.operatorId, + activationId: stream.run.runId, + preview: preview.text, + previewTruncated: preview.truncated, + phase: runtimeInvocationOutcome(stream.run) ? 'completed' : 'streaming', + previewUpdatedAt: latestText.ts, + sourceEventId: latestText.id, + ...(latestText.refs?.providerEventId ? { messageId: latestText.refs.providerEventId } : {}), + sampleStartedAt, + ...(outputTokens !== undefined ? { outputTokens } : {}), + ...(sampleDurationMs !== undefined ? { sampleDurationMs } : {}), + ...(tokensPerSecond !== undefined ? { tokensPerSecond } : {}), + }; +} + +function boundOutputPreview( + text: string, + edge: 'head' | 'tail', +): { text: string; truncated: boolean } { + const codePoints = Array.from(text.trim()); + if (codePoints.length <= AGENT_GRAPH_OUTPUT_PREVIEW_MAX_CODE_POINTS) { + return { text: codePoints.join(''), truncated: false }; + } + const visible = + edge === 'head' + ? codePoints.slice(0, AGENT_GRAPH_OUTPUT_PREVIEW_MAX_CODE_POINTS) + : codePoints.slice(-AGENT_GRAPH_OUTPUT_PREVIEW_MAX_CODE_POINTS); + return { text: visible.join(''), truncated: true }; +} + +function roundTokensPerSecond(value: number): number { + return Math.round(value * 10) / 10; +} + export function projectAgentGraphRecords(input: ProjectAgentGraphRecordsInput): { ignoredPartialEvents: number; records: AgentGraphRecord[]; diff --git a/packages/runtime/src/stream-graph-read-model.ts b/packages/runtime/src/stream-graph-read-model.ts index 5510f927b7..b05c5a5699 100644 --- a/packages/runtime/src/stream-graph-read-model.ts +++ b/packages/runtime/src/stream-graph-read-model.ts @@ -29,6 +29,7 @@ import type { } from './stream-graph-dispatch.js'; import type { AgentGraphActivationStatus, + AgentGraphOperatorOutputProjection, AgentGraphRecord, AgentGraphRecordFacet, AgentGraphSupervisorSignal, @@ -65,6 +66,7 @@ const MAX_OPERATOR_INSPECTION_EDGES = 512; const MAX_OPERATOR_INSPECTION_WORK = 256; const MAX_OPERATOR_INSPECTION_CLAIMS = 256; const MAX_INSTRUCTION_PREVIEW_CHARS = 500; +const MAX_OUTPUT_PREVIEW_CODE_POINTS = 280; export type AgentGraphClientOperatorStatus = | 'not_started' @@ -121,6 +123,21 @@ export interface AgentGraphClientOperator { terminalRecordId?: string; run: AgentGraphClientRunRef; }; + output?: AgentGraphClientOperatorOutput; +} + +export interface AgentGraphClientOperatorOutput { + activationId: string; + preview: string; + previewTruncated: boolean; + phase: 'streaming' | 'completed'; + previewUpdatedAt: number; + sourceEventId: string; + messageId?: string; + sampleStartedAt: number; + outputTokens?: number; + sampleDurationMs?: number; + tokensPerSecond?: number; } export interface AgentGraphClientEdge { @@ -291,7 +308,7 @@ export interface AgentGraphClientMaterialization { export interface AdvancedAgentGraphClientProjection { snapshot: AgentGraphClientSnapshot; operator: AgentGraphOperatorInspection; - activity: AgentGraphClientActivity; + activity?: AgentGraphClientActivity; terminalActivity?: AgentGraphClientActivity; } @@ -315,8 +332,9 @@ interface BuiltReadModel { /** * Durable, bounded graph-facing projection for untrusted presentation clients. * - * Payloads remain in Session/Runtime stores. This surface carries only - * identities, lifecycle state, wait reasons, and bounded instruction previews. + * Full payloads remain in Session/Runtime stores. This surface carries only + * identities, lifecycle state, wait reasons, bounded instruction/output + * previews, and provider-reported usage-derived throughput. */ export function buildAgentGraphClientSnapshot( input: BuildAgentGraphClientReadModelInput, @@ -352,7 +370,13 @@ export function advanceMaterializedAgentGraphClientProjection( activationHadError: boolean, ): AdvancedAgentGraphClientProjection | undefined { const projected = projectClientSessionEvent(runtime.event, activationHadError); - if (!projected) return undefined; + const output = advanceClientOperatorOutput( + inspectionInput.operator.output, + runtime.event, + runtime.claim.targetRunId, + ['completed', 'failed', 'aborted', 'cancelled'].includes(inspectionInput.operator.status), + ); + if (!projected && !output) return undefined; if ( snapshotInput.graphId !== runtime.intent.graphId || inspectionInput.graphId !== runtime.intent.graphId || @@ -361,21 +385,36 @@ export function advanceMaterializedAgentGraphClientProjection( ) { throw new Error('Agent graph runtime activity does not match its materialized projection'); } - const activity: AgentGraphClientActivity = { - recordId: clientRuntimeRecordId(runtime), - operatorId: runtime.claim.targetOperatorId, - activationId: runtime.claim.targetRunId, - eventTime: runtime.event.ts, - facets: projected.facets, - signals: projected.signals, - run: { - sessionId: runtime.claim.targetSessionId, - agentRunId: runtime.claim.targetRunId, - turnId: runtime.claim.targetTurnId, - }, - }; + const activity: AgentGraphClientActivity | undefined = projected + ? { + recordId: clientRuntimeRecordId(runtime), + operatorId: runtime.claim.targetOperatorId, + activationId: runtime.claim.targetRunId, + eventTime: runtime.event.ts, + facets: projected.facets, + signals: projected.signals, + run: { + sessionId: runtime.claim.targetSessionId, + agentRunId: runtime.claim.targetRunId, + turnId: runtime.claim.targetTurnId, + }, + } + : undefined; const snapshot = structuredClone(snapshotInput); const inspection = structuredClone(inspectionInput); + if (output) inspection.operator.output = output; + const visibleOperatorIndex = snapshot.operators.findIndex( + (operator) => operator.operatorId === inspection.operator.operatorId, + ); + if (visibleOperatorIndex >= 0 && output) { + snapshot.operators[visibleOperatorIndex] = structuredClone(inspection.operator); + } + if (!projected || !activity) { + snapshot.latestEventTime = Math.max(snapshot.latestEventTime ?? 0, runtime.event.ts); + snapshot.snapshotVersion = clientSnapshotVersion(snapshot); + inspection.snapshotVersion = snapshot.snapshotVersion; + return { snapshot, operator: inspection }; + } if ( snapshot.recentActivity.some((record) => record.recordId === activity.recordId) || inspection.recentRecords.some((record) => record.recordId === activity.recordId) @@ -487,9 +526,6 @@ export function advanceMaterializedAgentGraphClientProjection( .slice(-MAX_OPERATOR_INSPECTION_RECORDS); inspection.omitted.records = Math.max(0, inspectionRecordCount - inspection.recentRecords.length); - const visibleOperatorIndex = snapshot.operators.findIndex( - (operator) => operator.operatorId === inspection.operator.operatorId, - ); if (visibleOperatorIndex >= 0) { snapshot.operators[visibleOperatorIndex] = structuredClone(inspection.operator); } @@ -665,6 +701,12 @@ function buildReadModel(input: BuildAgentGraphClientReadModelInput): BuiltReadMo .map(clientActivity) .sort(compareClientActivity); const work = schedule.work.map(clientWork); + const outputByOperator = new Map( + (input.observation.projection.operatorOutputs ?? []).map((output) => [ + output.operatorId, + output, + ]), + ); const operators = input.observation.projection.operators.map((binding) => { const provision = provisionByOperator.get(binding.operatorId); if (!provision || provision.targetSessionId !== binding.sessionId) { @@ -683,6 +725,7 @@ function buildReadModel(input: BuildAgentGraphClientReadModelInput): BuiltReadMo const readiness = allReadiness.slice(0, MAX_OPERATOR_READINESS); const state = input.observation.projection.state.operators[binding.operatorId]; const currentActivation = state?.activations[state.currentActivationId]; + const output = outputByOperator.get(binding.operatorId); const currentClaim = currentActivation ? claims.find( (claim) => @@ -750,6 +793,9 @@ function buildReadModel(input: BuildAgentGraphClientReadModelInput): BuiltReadMo }, } : {}), + ...(currentActivation && output?.activationId === currentActivation.activationId + ? { output: clientOperatorOutput(output) } + : {}), } satisfies AgentGraphClientOperator; }); const stoppedTargets = schedule.stoppedTargets.map(clientStoppedTarget); @@ -796,6 +842,117 @@ function buildReadModel(input: BuildAgentGraphClientReadModelInput): BuiltReadMo }; } +function clientOperatorOutput( + output: AgentGraphOperatorOutputProjection, +): AgentGraphClientOperatorOutput { + return { + activationId: output.activationId, + preview: output.preview, + previewTruncated: output.previewTruncated, + phase: output.phase, + previewUpdatedAt: output.previewUpdatedAt, + sourceEventId: output.sourceEventId, + ...(output.messageId ? { messageId: output.messageId } : {}), + sampleStartedAt: output.sampleStartedAt, + ...(output.outputTokens !== undefined ? { outputTokens: output.outputTokens } : {}), + ...(output.sampleDurationMs !== undefined ? { sampleDurationMs: output.sampleDurationMs } : {}), + ...(output.tokensPerSecond !== undefined ? { tokensPerSecond: output.tokensPerSecond } : {}), + }; +} + +function advanceClientOperatorOutput( + current: AgentGraphClientOperatorOutput | undefined, + event: SessionEvent, + activationId: string, + operatorSettled: boolean, +): AgentGraphClientOperatorOutput | undefined { + if (current?.activationId === activationId && current.sourceEventId === event.id) { + return undefined; + } + const existing = current?.activationId === activationId ? current : undefined; + if ( + existing && + (event.type === 'text_delta' || event.type === 'text_complete') && + event.ts < existing.previewUpdatedAt + ) { + return undefined; + } + if (event.type === 'text_delta' || event.type === 'text_complete') { + const sameMessage = existing?.messageId === event.messageId; + const append = event.type === 'text_delta' && sameMessage; + const text = append ? foldClientOutputDelta(existing, event) : event.text; + const preview = boundClientOutputPreview( + text, + event.type === 'text_complete' ? 'head' : 'tail', + ); + if (!preview.text) return undefined; + return { + activationId, + preview: preview.text, + previewTruncated: preview.truncated || (append && existing.previewTruncated), + phase: operatorSettled || existing?.phase === 'completed' ? 'completed' : 'streaming', + previewUpdatedAt: event.ts, + sourceEventId: event.id, + ...(event.messageId ? { messageId: event.messageId } : {}), + sampleStartedAt: existing?.sampleStartedAt ?? event.ts, + ...(existing?.outputTokens !== undefined ? { outputTokens: existing.outputTokens } : {}), + ...(existing?.sampleDurationMs !== undefined + ? { sampleDurationMs: existing.sampleDurationMs } + : {}), + ...(existing?.tokensPerSecond !== undefined + ? { tokensPerSecond: existing.tokensPerSecond } + : {}), + }; + } + if (event.type === 'token_usage' && existing) { + const sampleDurationMs = Math.max(0, event.ts - existing.sampleStartedAt); + return { + ...existing, + outputTokens: event.output, + sampleDurationMs, + ...(sampleDurationMs > 0 + ? { tokensPerSecond: roundClientTokensPerSecond((event.output * 1_000) / sampleDurationMs) } + : {}), + }; + } + if ((event.type === 'complete' || event.type === 'abort') && existing) { + return { ...existing, phase: 'completed' }; + } + return undefined; +} + +function foldClientOutputDelta( + existing: AgentGraphClientOperatorOutput, + event: Extract, +): string { + if (event.startOffset === undefined || existing.previewTruncated) { + return `${existing.preview}${event.text}`; + } + if (event.startOffset <= existing.preview.length) { + return `${existing.preview.slice(0, event.startOffset)}${event.text}`; + } + return event.text; +} + +function boundClientOutputPreview( + text: string, + edge: 'head' | 'tail', +): { text: string; truncated: boolean } { + const codePoints = Array.from(edge === 'head' ? text.trim() : text); + if (codePoints.length <= MAX_OUTPUT_PREVIEW_CODE_POINTS) { + return { text: codePoints.join(''), truncated: false }; + } + const visible = + edge === 'head' + ? codePoints.slice(0, MAX_OUTPUT_PREVIEW_CODE_POINTS) + : codePoints.slice(-MAX_OUTPUT_PREVIEW_CODE_POINTS); + return { text: visible.join(''), truncated: true }; +} + +function roundClientTokensPerSecond(value: number): number { + return Math.round(value * 10) / 10; +} + function operatorStatus( runtimeStatus: AgentGraphActivationStatus | undefined, readiness: AgentGraphClientOperator['readiness'], From 04ab597c3e292bfcb9d050bb5cf11c072c7004cb Mon Sep 17 00:00:00 2001 From: testikun Date: Fri, 4 Sep 2026 16:51:57 +0800 Subject: [PATCH 2/4] test(runtime): cover Agent Graph output invalidations Replace the obsolete no-delta-write assertion with bounded streaming projection and one-invalidation-per-commit coverage. Generated-by: OpenAI Codex --- .../stream-graph-coordinator.test.ts | 40 ++++++++++++++++--- 1 file changed, 35 insertions(+), 5 deletions(-) diff --git a/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts b/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts index 4201b6c807..84a935213e 100644 --- a/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts +++ b/packages/runtime/src/__tests__/stream-graph-coordinator.test.ts @@ -502,6 +502,11 @@ describe('host-managed agent graph coordinator', () => { snapshot.work[0]?.instructionPreview, 'Inspect the repository and report one concrete finding.', ); + const operatorOutput = snapshot.operators[0]?.output; + assert.ok(operatorOutput); + assert.equal(operatorOutput.phase, 'completed'); + assert.ok(Array.from(operatorOutput.preview).length <= 280); + assert.match(operatorOutput.preview, /Fake backend received:/); assert.match(snapshot.snapshotVersion, /^sha256:[a-f0-9]{64}$/); const readsBeforeInspection = delayedControlStore.projectionReadCounts(); const inspection = await coordinator.inspectOperator( @@ -541,15 +546,40 @@ describe('host-managed agent graph coordinator', () => { ); await new Promise((resolve) => setImmediate(resolve)); assert.ok(clientEvents.includes('reconciled')); - assert.equal( - clientEvents.filter((reason) => reason === 'runtime_activity').length, - 2, - 'partial text deltas must not cause projection writes or invalidations', - ); const incrementalProjectionCommits = delayedControlStore.projectionCommits.filter( (request) => request.incrementalRecordId !== undefined, ); assert.equal(incrementalProjectionCommits.length, 2); + const outputProjectionCommits = delayedControlStore.projectionCommits.filter( + (request) => + request.incrementalRecordId === undefined && + !request.replaceOperators && + request.activityRecords.length === 0 && + request.operators.some((candidate) => { + const payload = candidate.payload as { + operator?: { output?: { preview?: unknown } }; + }; + return typeof payload.operator?.output?.preview === 'string'; + }), + ); + assert.ok(outputProjectionCommits.length > 0); + assert.ok( + outputProjectionCommits.every((request) => + request.operators.every((candidate) => { + const payload = candidate.payload as { + operator?: { output?: { preview?: unknown } }; + }; + const preview = payload.operator?.output?.preview; + return typeof preview === 'string' && Array.from(preview).length <= 280; + }), + ), + 'streaming output commits must remain bounded', + ); + assert.equal( + clientEvents.filter((reason) => reason === 'runtime_activity').length, + incrementalProjectionCommits.length + outputProjectionCommits.length, + 'each materialized activity or bounded output update must invalidate exactly once', + ); assert.ok( incrementalProjectionCommits.every((request) => request.terminalActivities.length === 0), 'yielded SessionEvents must never write immutable terminal history', From 156bc6d94f3bb2b9e5ecbc15b2acbbb09a66b4ab Mon Sep 17 00:00:00 2001 From: admin Date: Sat, 5 Sep 2026 09:59:11 +0800 Subject: [PATCH 3/4] fix(desktop): complete Agent Graph locale copy --- apps/desktop/src/renderer/agent-graph-panel.tsx | 3 +++ 1 file changed, 3 insertions(+) diff --git a/apps/desktop/src/renderer/agent-graph-panel.tsx b/apps/desktop/src/renderer/agent-graph-panel.tsx index 51fe366be1..70b01fbb7a 100644 --- a/apps/desktop/src/renderer/agent-graph-panel.tsx +++ b/apps/desktop/src/renderer/agent-graph-panel.tsx @@ -146,6 +146,9 @@ export function getAgentGraphPanelCopy(locale: UiLocale): GraphPanelCopy { openSession: '開啟子任務', operators: 'Operators', selectedResults: '已選取結果', + liveOutput: '即時輸出', + completedOutput: '結果預覽', + throughput: (tokensPerSecond) => `${tokensPerSecond.toFixed(1)} token/s`, epoch: 'Graph 執行輪次', currentEpoch: '目前', historicalEpoch: '歷史記錄（唯讀）', From b799b9b383d5c1df2b7eb49cdc2e5457a0257bf0 Mon Sep 17 00:00:00 2001 From: faith_liu Date: Sat, 5 Sep 2026 10:40:43 +0800 Subject: [PATCH 4/4] ci: retry flaky peer-mesh check