diff --git a/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts b/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts index ab81b6396..c7a891a3d 100644 --- a/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts +++ b/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts @@ -1,5 +1,19 @@ +/** + * `contextProjector` domain (L4) — projects stored context history into the wire + * messages sent to the model, and surfaces every repair it had to apply. + * + * `AgentContextProjectorService` is the Agent-scope binding. The projection + * itself stays a pure transform over the history; repairs that keep the + * outgoing wire valid (a displaced result moved back to its call, a synthetic + * result invented for a lost one, an orphan/duplicate dropped, leading + * non-user messages dropped, consecutive assistants merged, blank text + * dropped) are reported through an optional sink and surfaced once here as a + * single deduped warning, so a silently-mangled history always leaves a trace. + */ + import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { ILogService } from '#/_base/log/log'; import { renderToolResultForModel } from '#/agent/contextMemory/toolResultRender'; import type { ContextMessage } from '#/agent/contextMemory/types'; import { ErrorCodes, KimiError } from '#/errors'; @@ -9,28 +23,136 @@ import { IAgentContextProjectorService } from './contextProjector'; export class AgentContextProjectorService implements IAgentContextProjectorService { declare readonly _serviceBrand: undefined; + // Signature of the last notable repair set that was logged. Lets a defect that + // recurs identically every send (e.g. a persistently lost result re-synthesized + // each turn) log once, not per step; reset to null on a clean projection so a + // later recurrence after a healthy stretch is surfaced again. + private lastRepairSignature: string | null = null; + + constructor(@ILogService private readonly log: ILogService) {} + project(messages: readonly ContextMessage[]): readonly Message[] { - return project(messages); + return this.projectWithTrace(messages, project); } projectStrict(messages: readonly ContextMessage[]): readonly Message[] { - return projectStrict(messages); + return this.projectWithTrace(messages, projectStrict); + } + + private projectWithTrace( + messages: readonly ContextMessage[], + fn: (history: readonly ContextMessage[], onAnomaly?: (anomaly: ProjectionAnomaly) => void) => Message[], + ): readonly Message[] { + const anomalies: ProjectionAnomaly[] = []; + const result = fn(messages, (anomaly) => anomalies.push(anomaly)); + this.reportProjectionRepairs(anomalies); + return result; + } + + // Surface the projector's wire-repairs so a silently-mangled history leaves a + // trace. Deduped by signature so a defect that recurs identically every send + // (e.g. a persistently lost result re-synthesized each turn) logs once, not per + // step. Trailing-tail synthesis is excluded — it is the expected close of an + // in-flight call, not a defect. + private reportProjectionRepairs(anomalies: readonly ProjectionAnomaly[]): void { + const notable = anomalies.filter( + (anomaly) => !(anomaly.kind === 'tool_result_synthesized' && anomaly.trailing), + ); + if (notable.length === 0) { + this.lastRepairSignature = null; + return; + } + const signature = notable + .map((anomaly) => ('toolCallId' in anomaly ? `${anomaly.kind}:${anomaly.toolCallId}` : anomaly.kind)) + .toSorted() + .join('|'); + if (signature === this.lastRepairSignature) return; + this.lastRepairSignature = signature; + + let reordered = 0; + let synthesized = 0; + let droppedOrphan = 0; + let duplicateCallsDropped = 0; + let duplicateResultsDropped = 0; + let leadingDropped = 0; + let assistantsMerged = 0; + let whitespaceDropped = 0; + for (const anomaly of notable) { + if (anomaly.kind === 'tool_result_reordered') reordered += 1; + else if (anomaly.kind === 'tool_result_synthesized') synthesized += 1; + else if (anomaly.kind === 'orphan_tool_result_dropped') droppedOrphan += 1; + else if (anomaly.kind === 'duplicate_tool_call_dropped') duplicateCallsDropped += 1; + else if (anomaly.kind === 'duplicate_tool_result_dropped') duplicateResultsDropped += 1; + else if (anomaly.kind === 'leading_non_user_dropped') leadingDropped += 1; + else if (anomaly.kind === 'consecutive_assistants_merged') assistantsMerged += 1; + else whitespaceDropped += 1; + } + const toolCallIds = [ + ...new Set( + notable.flatMap((anomaly) => ('toolCallId' in anomaly ? [anomaly.toolCallId] : [])), + ), + ].slice(0, 5); + this.log.warn('repaired the request to keep it wire-valid', { + reordered, + synthesized, + droppedOrphan, + duplicateCallsDropped, + duplicateResultsDropped, + leadingDropped, + assistantsMerged, + whitespaceDropped, + toolCallIds, + }); } } -function projectStrict(history: readonly ContextMessage[]): Message[] { - const projected = project(history); - return dropLeadingNonUserMessages(mergeConsecutiveAssistantMessages(dedupeDuplicateToolCalls(projected))); +/** + * A repair the projector applied to make the history wire-valid. Each one means + * the stored history was not directly sendable to a strict provider. + */ +type ProjectionAnomaly = + /** A recorded result was not adjacent to its call and had to be moved up. */ + | { readonly kind: 'tool_result_reordered'; readonly toolCallId: string } + /** + * No result existed for a call, so a placeholder was synthesized. `trailing` + * is true when it closed a still-open tail call (expected, not a defect), + * false when it closed a mid-history orphan whose result was lost. + */ + | { readonly kind: 'tool_result_synthesized'; readonly toolCallId: string; readonly trailing: boolean } + /** A result with no matching call anywhere was dropped. */ + | { readonly kind: 'orphan_tool_result_dropped'; readonly toolCallId: string } + /** A tool call whose id already appeared earlier was dropped (strict only). */ + | { readonly kind: 'duplicate_tool_call_dropped'; readonly toolCallId: string } + /** A second result for an already-answered id was dropped (strict only). */ + | { readonly kind: 'duplicate_tool_result_dropped'; readonly toolCallId: string } + /** A leading non-user message was dropped so the first turn is user (strict). */ + | { readonly kind: 'leading_non_user_dropped'; readonly role: string } + /** Two adjacent assistant turns were merged into one (strict). */ + | { readonly kind: 'consecutive_assistants_merged' } + /** A non-empty but all-whitespace text block was dropped. */ + | { readonly kind: 'whitespace_text_dropped'; readonly role: string }; + +type OnAnomaly = (anomaly: ProjectionAnomaly) => void; + +function projectStrict(history: readonly ContextMessage[], onAnomaly?: OnAnomaly): Message[] { + const projected = project(history, onAnomaly); + return dropLeadingNonUserMessages( + mergeConsecutiveAssistantMessages(dedupeDuplicateToolCalls(projected, onAnomaly), onAnomaly), + onAnomaly, + ); } -function dedupeDuplicateToolCalls(messages: readonly Message[]): Message[] { +function dedupeDuplicateToolCalls(messages: readonly Message[], onAnomaly?: OnAnomaly): Message[] { const seenToolCallIds = new Set(); const keptToolResultIndexes = new Map(); const out: Message[] = []; for (const message of messages) { if (message.role === 'assistant' && message.toolCalls.length > 0) { const kept = message.toolCalls.filter((toolCall) => { - if (seenToolCallIds.has(toolCall.id)) return false; + if (seenToolCallIds.has(toolCall.id)) { + onAnomaly?.({ kind: 'duplicate_tool_call_dropped', toolCallId: toolCall.id }); + return false; + } seenToolCallIds.add(toolCall.id); return true; }); @@ -46,6 +168,8 @@ function dedupeDuplicateToolCalls(messages: readonly Message[]): Message[] { if (previousIndex !== undefined) { if (isInterruptedToolResult(out[previousIndex]) && !isInterruptedToolResult(message)) { out[previousIndex] = message; + } else { + onAnomaly?.({ kind: 'duplicate_tool_result_dropped', toolCallId: message.toolCallId }); } continue; } @@ -56,7 +180,10 @@ function dedupeDuplicateToolCalls(messages: readonly Message[]): Message[] { return out; } -function mergeConsecutiveAssistantMessages(messages: readonly Message[]): Message[] { +function mergeConsecutiveAssistantMessages( + messages: readonly Message[], + onAnomaly?: OnAnomaly, +): Message[] { const out: Message[] = []; for (const message of messages) { const previous = out.at(-1); @@ -66,6 +193,7 @@ function mergeConsecutiveAssistantMessages(messages: readonly Message[]): Messag content: [...previous.content, ...message.content], toolCalls: [...previous.toolCalls, ...message.toolCalls], }; + onAnomaly?.({ kind: 'consecutive_assistants_merged' }); continue; } out.push(message); @@ -73,9 +201,12 @@ function mergeConsecutiveAssistantMessages(messages: readonly Message[]): Messag return out; } -function dropLeadingNonUserMessages(messages: readonly Message[]): Message[] { +function dropLeadingNonUserMessages(messages: readonly Message[], onAnomaly?: OnAnomaly): Message[] { let start = 0; - while (start < messages.length && messages[start]?.role !== 'user') start += 1; + while (start < messages.length && messages[start]?.role !== 'user') { + onAnomaly?.({ kind: 'leading_non_user_dropped', role: messages[start]!.role }); + start += 1; + } return start === 0 ? [...messages] : messages.slice(start); } @@ -96,17 +227,35 @@ function dropLeadingNonUserMessages(messages: readonly Message[]): Message[] { // adjacent user prompts (accumulated and materialized once per run), and // strips context-only metadata off the wire. // +// Every repair that changes what the model sees (a displaced result pulled up, +// a lost result synthesized, an orphan dropped, blank text dropped) is reported +// through `onAnomaly`; the projection stays a pure transform and the caller +// decides whether to surface the trace. +// // The projected messages share their content parts and tool calls with the // stored context (only the top-level wrapper is rebuilt); consumers must // treat the projection as read-only, which every provider conversion already // honors by building fresh structures. -function project(history: readonly ContextMessage[]): Message[] { +function project(history: readonly ContextMessage[], onAnomaly?: OnAnomaly): Message[] { const hasAssistant = history.some( (message) => message.partial !== true && message.role === 'assistant', ); + // Last history index that is a real, non-tool turn. A call still open at the + // end whose owning assistant sits at/after it closed a trailing, possibly + // in-flight call (expected); one whose owner precedes it lost its result + // mid-history (a defect). Mirrors the trailing/mid-history split used to keep + // the trace free of routine in-flight closes. + let lastNonToolIndex = history.length - 1; + while ( + lastNonToolIndex >= 0 && + (history[lastNonToolIndex]?.role === 'tool' || history[lastNonToolIndex]?.partial === true) + ) { + lastNonToolIndex -= 1; + } + const out: Message[] = []; - const openSlots = new Map(); + const openSlots = new Map(); let merge: MergeGroup | undefined; const flushMerge = (): void => { @@ -127,10 +276,19 @@ function project(history: readonly ContextMessage[]): Message[] { merge = undefined; }; + // A real (non-tool) message — or a result for an unknown call — landing while + // calls are still open means those calls' results were not adjacent in the + // stored history; pulling them up is a real repair worth tracing. + const markForeignBetween = (): void => { + for (const slot of openSlots.values()) slot.foreignBetween = true; + }; + const emit = (source: ContextMessage): void => { - const content = projectedContent(source); + const content = projectedContent(source, onAnomaly); if (content.length === 0 && source.toolCalls.length === 0) return; + if (openSlots.size > 0) markForeignBetween(); + if (canMergeUserMessage(source)) { if (merge === undefined) { out.push(toWireMessage(source, content)); @@ -148,7 +306,7 @@ function project(history: readonly ContextMessage[]): Message[] { out.push(toWireMessage(source, content)); }; - for (const message of history) { + for (const [index, message] of history.entries()) { if (message.partial === true) continue; if (message.role === 'tool') { if (!hasAssistant) { @@ -157,24 +315,51 @@ function project(history: readonly ContextMessage[]): Message[] { } if (message.toolCallId === undefined) continue; const slot = openSlots.get(message.toolCallId); - if (slot === undefined) continue; + if (slot === undefined) { + if (openSlots.size > 0) markForeignBetween(); + onAnomaly?.({ kind: 'orphan_tool_result_dropped', toolCallId: message.toolCallId }); + continue; + } openSlots.delete(message.toolCallId); - out[slot] = toWireMessage(message, projectedContent(message)); + if (slot.foreignBetween) { + onAnomaly?.({ kind: 'tool_result_reordered', toolCallId: message.toolCallId }); + } + out[slot.index] = toWireMessage(message, projectedContent(message, onAnomaly)); continue; } emit(message); for (const call of message.toolCalls) { const reopened = openSlots.get(call.id); - if (reopened !== undefined) out[reopened] = createInterruptedToolResult(call.id); - openSlots.set(call.id, out.length); + if (reopened !== undefined) { + out[reopened.index] = createInterruptedToolResult(call.id); + onAnomaly?.({ + kind: 'tool_result_synthesized', + toolCallId: call.id, + trailing: reopened.ownerIndex >= lastNonToolIndex, + }); + } + openSlots.set(call.id, { index: out.length, ownerIndex: index, foreignBetween: false }); out.push(TOOL_RESULT_SLOT); } } - for (const [id, slot] of openSlots) out[slot] = createInterruptedToolResult(id); + for (const [id, slot] of openSlots) { + out[slot.index] = createInterruptedToolResult(id); + onAnomaly?.({ + kind: 'tool_result_synthesized', + toolCallId: id, + trailing: slot.ownerIndex >= lastNonToolIndex, + }); + } flushMerge(); return out; } +interface OpenSlot { + index: number; + ownerIndex: number; + foreignBetween: boolean; +} + interface MergeGroup { index: number; singleContent: readonly ContentPart[] | undefined; @@ -193,7 +378,7 @@ function appendMergeContent(group: MergeGroup, content: readonly ContentPart[]): if (text.length > 0) group.texts.push(text); } -function projectedContent(source: ContextMessage): ContentPart[] { +function projectedContent(source: ContextMessage, onAnomaly?: OnAnomaly): ContentPart[] { const content = source.role === 'tool' ? renderToolResultForModel({ @@ -202,13 +387,32 @@ function projectedContent(source: ContextMessage): ContentPart[] { note: source.note, }) : source.content; - return cleanContent(source, content); + return cleanContent(source, content, onAnomaly); } -function cleanContent(source: ContextMessage, rawContent: readonly ContentPart[]): ContentPart[] { - const content = rawContent.some(isBlankText) - ? rawContent.filter((part) => !isBlankText(part)) - : rawContent; +function cleanContent( + source: ContextMessage, + rawContent: readonly ContentPart[], + onAnomaly?: OnAnomaly, +): ContentPart[] { + const hasBlank = rawContent.some(isBlankText); + let content: readonly ContentPart[] = rawContent; + if (hasBlank) { + const filtered: ContentPart[] = []; + for (const part of rawContent) { + if (isBlankText(part)) { + // Report only whitespace-only (non-empty) blocks: a truly empty `''` + // block is routine cleanup, whereas a block that is non-empty yet + // all-whitespace signals upstream fed blank content worth surfacing. + if (part.type === 'text' && part.text.length > 0) { + onAnomaly?.({ kind: 'whitespace_text_dropped', role: source.role }); + } + } else { + filtered.push(part); + } + } + content = filtered; + } if (source.role === 'tool' && content.length === 0) { throw new KimiError( ErrorCodes.REQUEST_INVALID, diff --git a/packages/agent-core-v2/test/contextProjector/projector-tool-exchanges.test.ts b/packages/agent-core-v2/test/contextProjector/projector-tool-exchanges.test.ts index 81dc95977..dc23ba004 100644 --- a/packages/agent-core-v2/test/contextProjector/projector-tool-exchanges.test.ts +++ b/packages/agent-core-v2/test/contextProjector/projector-tool-exchanges.test.ts @@ -3,12 +3,45 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; +import { ILogService, type ILogger } from '#/_base/log/log'; import type { ContextMessage } from '#/agent/contextMemory/types'; import { IAgentContextProjectorService } from '#/agent/contextProjector/contextProjector'; import { AgentContextProjectorService } from '#/agent/contextProjector/contextProjectorService'; import { toProtocolMessage } from '#/agent/contextMemory/messageProjection'; import type { Message } from '#/app/llmProtocol/message'; +const REPAIR_WARNING = 'repaired the request to keep it wire-valid'; + +interface WarningCall { + readonly message: string; + readonly payload: unknown; +} + +function createCapturingLog(warnings: WarningCall[]): ILogService { + const logger: ILogger = { + error: () => {}, + warn: (message, payload) => { + warnings.push({ message, payload }); + }, + info: () => {}, + debug: () => {}, + child: () => logger, + }; + return { + ...logger, + _serviceBrand: undefined, + level: 'warn', + setLevel: () => {}, + flush: () => Promise.resolve(), + }; +} + +function repairPayloads(warnings: WarningCall[]): Record[] { + return warnings + .filter((call) => call.message === REPAIR_WARNING) + .map((call) => call.payload as Record); +} + // Tests for how the projector normalizes tool exchanges: results are pulled up // right after their call, messages that landed between a call and its results // are deferred to after the exchange, unanswered calls are closed with a @@ -45,10 +78,13 @@ function toolResult(toolCallId: string, text: string): ContextMessage { describe('projector tool-exchange normalization', () => { let disposables: DisposableStore; let projector: IAgentContextProjectorService; + let warnings: WarningCall[]; beforeEach(() => { disposables = new DisposableStore(); + warnings = []; const ix = disposables.add(new TestInstantiationService()); + ix.set(ILogService, createCapturingLog(warnings)); ix.set(IAgentContextProjectorService, new SyncDescriptor(AgentContextProjectorService)); projector = ix.get(IAgentContextProjectorService); }); @@ -321,4 +357,68 @@ describe('projector tool-exchange normalization', () => { ]); }); + describe('surfaces repairs so a mangled history leaves a trace', () => { + it('stays silent for a well-formed projection', () => { + project([user('go'), assistant('', ['c1']), toolResult('c1', 'one'), user('next')]); + expect(repairPayloads(warnings)).toEqual([]); + }); + + it('reports a result pulled up to its call as reordered', () => { + project([ + assistant('', ['c1', 'c2']), + reminder('host note'), + toolResult('c1', 'one'), + toolResult('c2', 'two'), + ]); + expect(repairPayloads(warnings)).toEqual([ + expect.objectContaining({ + reordered: 2, + toolCallIds: expect.arrayContaining(['c1', 'c2']), + }), + ]); + }); + + it('reports a mid-history lost result but not a trailing in-flight close', () => { + project([user('go'), assistant('', ['c1']), user('keep going'), assistant('All done.')]); + expect(repairPayloads(warnings)).toEqual([ + expect.objectContaining({ synthesized: 1, toolCallIds: ['c1'] }), + ]); + + warnings.length = 0; + project([user('go'), assistant('', ['c1'])]); + expect(repairPayloads(warnings)).toEqual([]); + }); + + it('reports an orphan result whose call was never recorded', () => { + project([user('hi'), assistant('hello'), toolResult('ghost', 'orphaned')]); + expect(repairPayloads(warnings)).toEqual([ + expect.objectContaining({ droppedOrphan: 1, toolCallIds: ['ghost'] }), + ]); + }); + + it('logs a recurring defect once per signature and again after a clean projection', () => { + const broken = [user('go'), assistant('', ['c1']), user('keep going'), assistant('x')]; + project(broken); + project(broken); + expect(repairPayloads(warnings)).toHaveLength(1); + + project([user('go'), assistant('', ['c1']), toolResult('c1', 'one'), user('next')]); + project(broken); + expect(repairPayloads(warnings)).toHaveLength(2); + }); + + it('reports strict-mode leading-drop and orphan', () => { + projectStrict([assistant('stale'), toolResult('ghost', 'orphaned'), user('hi')]); + expect(repairPayloads(warnings).at(-1)).toEqual( + expect.objectContaining({ leadingDropped: 1, droppedOrphan: 1, toolCallIds: ['ghost'] }), + ); + }); + + it('reports strict-mode consecutive assistant merge', () => { + projectStrict([user('go'), assistant('one'), assistant('two')]); + expect(repairPayloads(warnings).at(-1)).toEqual( + expect.objectContaining({ assistantsMerged: 1 }), + ); + }); + }); }); diff --git a/packages/agent-core-v2/test/contextProjector/projector.bench.ts b/packages/agent-core-v2/test/contextProjector/projector.bench.ts index 165860a48..ef3e06f7a 100644 --- a/packages/agent-core-v2/test/contextProjector/projector.bench.ts +++ b/packages/agent-core-v2/test/contextProjector/projector.bench.ts @@ -16,12 +16,28 @@ import { bench, describe } from 'vitest'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; +import { ILogService, type ILogger } from '#/_base/log/log'; import type { ContextMessage } from '#/agent/contextMemory/types'; import { IAgentContextProjectorService } from '#/agent/contextProjector/contextProjector'; import { AgentContextProjectorService } from '#/agent/contextProjector/contextProjectorService'; import { ErrorCodes, KimiError } from '#/errors'; import type { ContentPart, Message, TextPart, ToolCall } from '#/app/llmProtocol/message'; +const noopLogger: ILogger = { + error: () => {}, + warn: () => {}, + info: () => {}, + debug: () => {}, + child: () => noopLogger, +}; +const noopLogService: ILogService = { + ...noopLogger, + _serviceBrand: undefined, + level: 'off', + setLevel: () => {}, + flush: () => Promise.resolve(), +}; + // --------------------------------------------------------------------------- // Legacy implementation (verbatim copy of the pre-rewrite `project`) // --------------------------------------------------------------------------- @@ -185,6 +201,7 @@ function makeMixedHistory(turns: number): ContextMessage[] { function createProjector(disposables: DisposableStore): IAgentContextProjectorService { const ix = disposables.add(new TestInstantiationService()); + ix.set(ILogService, noopLogService); ix.set(IAgentContextProjectorService, new SyncDescriptor(AgentContextProjectorService)); return ix.get(IAgentContextProjectorService); }