diff --git a/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts b/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts index 43becbb33..c0f50e302 100644 --- a/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts +++ b/packages/agent-core-v2/src/agent/contextProjector/contextProjectorService.ts @@ -1,46 +1,43 @@ /** - * `contextProjector` domain — projects stored context history into the wire - * messages sent to the model, and surfaces every repair it had to apply. + * `contextProjector` domain — `IAgentContextProjectorService` implementation. * - * `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, wholly-vacuous messages — nothing sendable was recorded, e.g. an - * assistant step that kept only an empty thinking part — dropped whole) are - * reported through an optional sink and surfaced once here as a - * single deduped warning plus a `context_projection_repaired` telemetry event, - * so a silently-mangled history always leaves a trace. The mutable - * repair-dedup signature (`lastRepairSignature`) is registered into - * `agentState` (`IAgentStateService`) and read/written through it. - * - * `policy.media` selects the fallback projections for the two deterministic - * provider rejections: `'degraded'` (all but the most recent media replaced - * by text markers) resends after an HTTP 413 body-size rejection; - * `{ strip }` replaces only the snapshotted media identities present when - * degraded media is still too large or an image format is rejected, so a - * newly generated recovery image remains visible on later steps. Both are - * read-side only — the history keeps its media. + * Projects stored context history into the wire messages sent to the model, + * applies the read-side media fallbacks selected by `policy.media`, and + * surfaces every repair the projection had to apply: the repairs are + * summarized once per distinct signature into a single deduped warning + * (through `log`) plus a `context_projection_repaired` telemetry event + * (through `telemetry`), so a silently-mangled history always leaves a + * trace. The mutable repair-dedup signature (`lastRepairSignature`) is + * registered into `agentState` (`IAgentStateService`) and read/written + * through it. Bound at Agent scope. */ -import { createHash } from 'node:crypto'; import { LifecycleScope } from '#/app/scopes'; import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; import { ILogService } from '#/_base/log/log'; import { defineState } from '#/state/state'; -import { renderToolResultForModel } from '#/agent/contextMemory/toolResultRender'; import type { ContextMessage } from '#/agent/contextMemory/types'; -import { isVacuousContentPart } from '#/agent/contextMemory/vacuousContent'; import { IAgentStateService } from '#/agent/state/agentState'; -import { ErrorCodes, Error2 } from '#/errors'; -import type { ContentPart, Message } from '#/kosong/contract/message'; +import type { Message } from '#/kosong/contract/message'; import { ITelemetryService } from '#/app/telemetry/telemetry'; import { IAgentContextProjectorService, type MediaStripSnapshot, type ProjectionPolicy, } from './contextProjector'; +import { + MEDIA_DEGRADE_KEEP_RECENT, + captureMediaStripSnapshot, + degradeOlderMediaParts, + stripMediaPartsBySnapshot, +} from './mediaProjection'; +import { + project, + projectStrict, + summarizeProjectionRepairs, + type OnAnomaly, + type ProjectionAnomaly, +} from './projection'; export const contextProjectorLastRepairSignatureKey = defineState( 'contextProjector.lastRepairSignature', @@ -86,7 +83,7 @@ export class AgentContextProjectorService implements IAgentContextProjectorServi private projectWithTrace( messages: readonly ContextMessage[], - fn: (history: readonly ContextMessage[], onAnomaly?: (anomaly: ProjectionAnomaly) => void) => Message[], + fn: (history: readonly ContextMessage[], onAnomaly?: OnAnomaly) => Message[], ): readonly Message[] { const anomalies: ProjectionAnomaly[] = []; const result = fn(messages, (anomaly) => anomalies.push(anomaly)); @@ -151,499 +148,6 @@ export class AgentContextProjectorService implements IAgentContextProjectorServi } } -type ProjectionAnomaly = - | { readonly kind: 'tool_result_reordered'; readonly toolCallId: string } - | { readonly kind: 'tool_result_synthesized'; readonly toolCallId: string; readonly trailing: boolean } - | { readonly kind: 'orphan_tool_result_dropped'; readonly toolCallId: string } - | { readonly kind: 'duplicate_tool_call_dropped'; readonly toolCallId: string } - | { readonly kind: 'duplicate_tool_result_dropped'; readonly toolCallId: string } - | { readonly kind: 'leading_non_user_dropped'; readonly role: string } - | { readonly kind: 'consecutive_assistants_merged' } - | { readonly kind: 'whitespace_text_dropped'; readonly role: string } - | { readonly kind: 'vacuous_message_dropped'; readonly role: string }; - -interface ProjectionRepairSummary { - readonly reordered: number; - readonly synthesized: number; - readonly droppedOrphan: number; - readonly duplicateCallsDropped: number; - readonly duplicateResultsDropped: number; - readonly leadingDropped: number; - readonly assistantsMerged: number; - readonly whitespaceDropped: number; - readonly vacuousDropped: number; -} - -function summarizeProjectionRepairs( - anomalies: readonly ProjectionAnomaly[], -): ProjectionRepairSummary { - const summary = { - reordered: 0, - synthesized: 0, - droppedOrphan: 0, - duplicateCallsDropped: 0, - duplicateResultsDropped: 0, - leadingDropped: 0, - assistantsMerged: 0, - whitespaceDropped: 0, - vacuousDropped: 0, - }; - for (const anomaly of anomalies) { - if (anomaly.kind === 'tool_result_reordered') summary.reordered += 1; - else if (anomaly.kind === 'tool_result_synthesized') summary.synthesized += 1; - else if (anomaly.kind === 'orphan_tool_result_dropped') summary.droppedOrphan += 1; - else if (anomaly.kind === 'duplicate_tool_call_dropped') summary.duplicateCallsDropped += 1; - else if (anomaly.kind === 'duplicate_tool_result_dropped') summary.duplicateResultsDropped += 1; - else if (anomaly.kind === 'leading_non_user_dropped') summary.leadingDropped += 1; - else if (anomaly.kind === 'consecutive_assistants_merged') summary.assistantsMerged += 1; - else if (anomaly.kind === 'vacuous_message_dropped') summary.vacuousDropped += 1; - else summary.whitespaceDropped += 1; - } - return summary; -} - -type OnAnomaly = (anomaly: ProjectionAnomaly) => void; - -export const MEDIA_DEGRADE_KEEP_RECENT = 2; - -const MEDIA_DEGRADED_PLACEHOLDERS = { - image_url: - '[image omitted: dropped to fit the provider request size limit; re-read the file to view it]', - audio_url: - '[audio omitted: dropped to fit the provider request size limit; re-read the file to hear it]', - video_url: - '[video omitted: dropped to fit the provider request size limit; re-read the file to view it]', -} as const; - -export const MEDIA_STRIPPED_PLACEHOLDERS = { - image_url: - '[image omitted for provider compatibility; re-read the file to view it or get conversion guidance]', - audio_url: - '[audio omitted for provider compatibility; re-read the file to hear it]', - video_url: - '[video omitted for provider compatibility; re-read the file to view it]', -} as const; - -type MediaPlaceholderSet = typeof MEDIA_DEGRADED_PLACEHOLDERS | typeof MEDIA_STRIPPED_PLACEHOLDERS; - -type DegradableMediaPart = Extract< - ContentPart, - { readonly type: keyof MediaPlaceholderSet } ->; - -interface MediaContainer { - readonly url: string; - readonly id?: string; -} - -interface MediaStripSnapshotData { - readonly keys: ReadonlySet; -} - -type MediaContainerKeyCache = Partial>; - -const MEDIA_CONTAINER_KEY_CACHE = new WeakMap(); - -function isDegradableMediaPart( - part: ContentPart, -): part is DegradableMediaPart { - return part.type in MEDIA_DEGRADED_PLACEHOLDERS; -} - -function mediaContainer(part: DegradableMediaPart): MediaContainer { - if (part.type === 'image_url') return part.imageUrl; - if (part.type === 'audio_url') return part.audioUrl; - return part.videoUrl; -} - -function mediaStripKey(part: DegradableMediaPart): string { - const container = mediaContainer(part); - let cache = MEDIA_CONTAINER_KEY_CACHE.get(container); - const cached = cache?.[part.type]; - if (cached !== undefined) return cached; - - const key = createHash('sha256') - .update(part.type) - .update('\0') - .update(container.id ?? '') - .update('\0') - .update(container.url) - .digest('hex'); - if (cache === undefined) { - cache = {}; - MEDIA_CONTAINER_KEY_CACHE.set(container, cache); - } - cache[part.type] = key; - return key; -} - -function mediaStripSnapshotKeys(snapshot: MediaStripSnapshot): ReadonlySet { - return (snapshot as unknown as MediaStripSnapshotData).keys; -} - -export function captureMediaStripSnapshot( - messages: readonly Message[], -): MediaStripSnapshot { - const keys = new Set(); - for (const message of messages) { - for (const part of message.content) { - if (isDegradableMediaPart(part)) keys.add(mediaStripKey(part)); - } - } - return Object.freeze({ keys }) as unknown as MediaStripSnapshot; -} - -export function stripMediaPartsBySnapshot( - messages: readonly Message[], - snapshot: MediaStripSnapshot, -): readonly Message[] { - const keys = mediaStripSnapshotKeys(snapshot); - let changed = false; - const result = messages.map((message) => { - let messageChanged = false; - const content = message.content.map((part): ContentPart => { - if (!isDegradableMediaPart(part) || !keys.has(mediaStripKey(part))) return part; - changed = true; - messageChanged = true; - return { type: 'text', text: MEDIA_STRIPPED_PLACEHOLDERS[part.type] }; - }); - return messageChanged ? { ...message, content } : message; - }); - return changed ? result : messages; -} - -export function degradeOlderMediaParts( - messages: readonly Message[], - keepRecent: number, - placeholders: MediaPlaceholderSet = MEDIA_DEGRADED_PLACEHOLDERS, -): readonly Message[] { - const mediaCount = messages.reduce( - (count, message) => count + message.content.filter(isDegradableMediaPart).length, - 0, - ); - let toDegrade = Math.max(0, mediaCount - keepRecent); - if (toDegrade === 0) return messages; - - return messages.map((message) => { - if (toDegrade === 0 || !message.content.some(isDegradableMediaPart)) return message; - const content = message.content.map((part): ContentPart => { - if (toDegrade === 0 || !isDegradableMediaPart(part)) return part; - toDegrade -= 1; - return { type: 'text', text: placeholders[part.type] }; - }); - return { ...message, content }; - }); -} - -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[], 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)) { - onAnomaly?.({ kind: 'duplicate_tool_call_dropped', toolCallId: toolCall.id }); - return false; - } - seenToolCallIds.add(toolCall.id); - return true; - }); - if (kept.length === message.toolCalls.length) { - out.push(message); - } else if (kept.length > 0 || !message.content.every(isVacuousContentPart)) { - out.push({ ...message, toolCalls: kept }); - } else if (message.content.length > 0) { - onAnomaly?.({ kind: 'vacuous_message_dropped', role: message.role }); - } - continue; - } - if (message.role === 'tool' && message.toolCallId !== undefined) { - const previousIndex = keptToolResultIndexes.get(message.toolCallId); - if (previousIndex !== undefined) { - if (isInterruptedToolResult(out[previousIndex]) && !isInterruptedToolResult(message)) { - out[previousIndex] = message; - } else { - onAnomaly?.({ kind: 'duplicate_tool_result_dropped', toolCallId: message.toolCallId }); - } - continue; - } - keptToolResultIndexes.set(message.toolCallId, out.length); - } - out.push(message); - } - return out; -} - -function mergeConsecutiveAssistantMessages( - messages: readonly Message[], - onAnomaly?: OnAnomaly, -): Message[] { - const out: Message[] = []; - for (const message of messages) { - const previous = out.at(-1); - if (previous !== undefined && previous.role === 'assistant' && message.role === 'assistant') { - out[out.length - 1] = { - ...previous, - content: [...previous.content, ...message.content], - toolCalls: [...previous.toolCalls, ...message.toolCalls], - }; - onAnomaly?.({ kind: 'consecutive_assistants_merged' }); - continue; - } - out.push(message); - } - return out; -} - -function dropLeadingNonUserMessages(messages: readonly Message[], onAnomaly?: OnAnomaly): Message[] { - let start = 0; - 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); -} - -function project(history: readonly ContextMessage[], onAnomaly?: OnAnomaly): Message[] { - const hasAssistant = history.some( - (message) => message.partial !== true && message.role === 'assistant', - ); - - 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(); - let merge: MergeGroup | undefined; - - const flushMerge = (): void => { - if (merge === undefined) return; - if (merge.singleContent === undefined) { - const text = merge.texts.join('\n\n'); - const content: ContentPart[] = text === '' ? [] : [{ type: 'text', text }]; - content.push(...merge.parts); - out[merge.index] = { - role: 'user', - name: undefined, - content, - toolCalls: [], - toolCallId: undefined, - partial: undefined, - }; - } - merge = undefined; - }; - - const markForeignBetween = (): void => { - for (const slot of openSlots.values()) slot.foreignBetween = true; - }; - - const emit = (source: ContextMessage): void => { - const content = projectedContent(source, onAnomaly); - if (source.toolCalls.length === 0 && !hasDeclaredTools(source)) { - if (content.length === 0) return; - if (content.every(isVacuousContentPart)) { - onAnomaly?.({ kind: 'vacuous_message_dropped', role: source.role }); - return; - } - } - - if (openSlots.size > 0) markForeignBetween(); - - if (canMergeUserMessage(source)) { - if (merge === undefined) { - out.push(toWireMessage(source, content)); - merge = { index: out.length - 1, singleContent: content, texts: [], parts: [] }; - } else { - if (merge.singleContent !== undefined) { - appendMergeContent(merge, merge.singleContent); - merge.singleContent = undefined; - } - appendMergeContent(merge, content); - } - return; - } - flushMerge(); - out.push(toWireMessage(source, content)); - }; - - for (const [index, message] of history.entries()) { - if (message.partial === true) continue; - if (message.role === 'tool') { - if (!hasAssistant) { - emit(message); - continue; - } - if (message.toolCallId === undefined) continue; - const slot = openSlots.get(message.toolCallId); - if (slot === undefined) { - if (openSlots.size > 0) markForeignBetween(); - onAnomaly?.({ kind: 'orphan_tool_result_dropped', toolCallId: message.toolCallId }); - continue; - } - openSlots.delete(message.toolCallId); - 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.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.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; - texts: string[]; - parts: ContentPart[]; -} - -function appendMergeContent(group: MergeGroup, content: readonly ContentPart[]): void { - let text = ''; - for (const part of content) { - if (part.type === 'text') text += part.text; - else group.parts.push(part); - } - if (text.length > 0) group.texts.push(text); -} - -function projectedContent(source: ContextMessage, onAnomaly?: OnAnomaly): ContentPart[] { - const content = - source.role === 'tool' - ? renderToolResultForModel({ - output: outputFromToolContent(source.content), - isError: source.isError, - note: source.note, - }) - : source.content; - return cleanContent(source, content, onAnomaly); -} - -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)) { - 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 Error2( - ErrorCodes.REQUEST_INVALID, - 'Tool result message content cannot be empty after removing empty text blocks.', - { details: { toolCallId: source.toolCallId } }, - ); - } - return [...content]; -} - -function outputFromToolContent(content: readonly ContentPart[]): string | readonly ContentPart[] { - const only = content[0]; - return content.length === 1 && only?.type === 'text' ? only.text : content; -} - -const TOOL_INTERRUPTED_TEXT = - 'Tool result is not available in the current context. Do not assume the tool completed successfully.'; - -const TOOL_RESULT_SLOT: Message = createInterruptedToolResult(''); - -function createInterruptedToolResult(toolCallId: string): Message { - return { - role: 'tool', - name: undefined, - content: [{ type: 'text', text: TOOL_INTERRUPTED_TEXT }], - toolCalls: [], - toolCallId, - partial: undefined, - }; -} - -function isInterruptedToolResult(message: Message | undefined): boolean { - if (message?.role !== 'tool') return false; - const [part] = message.content; - return part?.type === 'text' && part.text === TOOL_INTERRUPTED_TEXT; -} - -function isBlankText(part: ContentPart): boolean { - return part.type === 'text' && part.text.trim().length === 0; -} - -function canMergeUserMessage(message: ContextMessage): boolean { - return message.role === 'user' && message.origin?.kind === 'user'; -} - -function hasDeclaredTools(message: ContextMessage): boolean { - return message.tools !== undefined && message.tools.length > 0; -} - -function toWireMessage(message: ContextMessage, content: ContentPart[]): Message { - return { - role: message.role, - name: message.name, - content, - toolCalls: message.toolCalls, - toolCallId: message.toolCallId, - partial: message.partial, - tools: message.tools, - }; -} - registerScopedService( LifecycleScope.Agent, IAgentContextProjectorService, diff --git a/packages/agent-core-v2/src/agent/contextProjector/mediaProjection.ts b/packages/agent-core-v2/src/agent/contextProjector/mediaProjection.ts new file mode 100644 index 000000000..0c81f03c5 --- /dev/null +++ b/packages/agent-core-v2/src/agent/contextProjector/mediaProjection.ts @@ -0,0 +1,148 @@ +/** + * `contextProjector` domain — read-side media fallbacks for the two + * deterministic provider rejections. + * + * The degraded projection replaces all but the most recent media parts with + * text markers after an HTTP 413 body-size rejection; the strip projection + * replaces exactly the snapshotted media identities after a rejected-format + * or still-too-large resend, so a newly generated recovery image stays + * visible on later steps. Both rewrite only the projected wire messages — + * the stored history keeps its media. + */ + +import { createHash } from 'node:crypto'; + +import type { ContentPart, Message } from '#/kosong/contract/message'; + +import type { MediaStripSnapshot } from './contextProjector'; + +export const MEDIA_DEGRADE_KEEP_RECENT = 2; + +const MEDIA_DEGRADED_PLACEHOLDERS = { + image_url: + '[image omitted: dropped to fit the provider request size limit; re-read the file to view it]', + audio_url: + '[audio omitted: dropped to fit the provider request size limit; re-read the file to hear it]', + video_url: + '[video omitted: dropped to fit the provider request size limit; re-read the file to view it]', +} as const; + +export const MEDIA_STRIPPED_PLACEHOLDERS = { + image_url: + '[image omitted for provider compatibility; re-read the file to view it or get conversion guidance]', + audio_url: + '[audio omitted for provider compatibility; re-read the file to hear it]', + video_url: + '[video omitted for provider compatibility; re-read the file to view it]', +} as const; + +type MediaPlaceholderSet = typeof MEDIA_DEGRADED_PLACEHOLDERS | typeof MEDIA_STRIPPED_PLACEHOLDERS; + +type DegradableMediaPart = Extract< + ContentPart, + { readonly type: keyof MediaPlaceholderSet } +>; + +interface MediaContainer { + readonly url: string; + readonly id?: string; +} + +interface MediaStripSnapshotData { + readonly keys: ReadonlySet; +} + +type MediaContainerKeyCache = Partial>; + +const MEDIA_CONTAINER_KEY_CACHE = new WeakMap(); + +function isDegradableMediaPart( + part: ContentPart, +): part is DegradableMediaPart { + return part.type in MEDIA_DEGRADED_PLACEHOLDERS; +} + +function mediaContainer(part: DegradableMediaPart): MediaContainer { + if (part.type === 'image_url') return part.imageUrl; + if (part.type === 'audio_url') return part.audioUrl; + return part.videoUrl; +} + +function mediaStripKey(part: DegradableMediaPart): string { + const container = mediaContainer(part); + let cache = MEDIA_CONTAINER_KEY_CACHE.get(container); + const cached = cache?.[part.type]; + if (cached !== undefined) return cached; + + const key = createHash('sha256') + .update(part.type) + .update('\0') + .update(container.id ?? '') + .update('\0') + .update(container.url) + .digest('hex'); + if (cache === undefined) { + cache = {}; + MEDIA_CONTAINER_KEY_CACHE.set(container, cache); + } + cache[part.type] = key; + return key; +} + +function mediaStripSnapshotKeys(snapshot: MediaStripSnapshot): ReadonlySet { + return (snapshot as unknown as MediaStripSnapshotData).keys; +} + +export function captureMediaStripSnapshot( + messages: readonly Message[], +): MediaStripSnapshot { + const keys = new Set(); + for (const message of messages) { + for (const part of message.content) { + if (isDegradableMediaPart(part)) keys.add(mediaStripKey(part)); + } + } + return Object.freeze({ keys }) as unknown as MediaStripSnapshot; +} + +export function stripMediaPartsBySnapshot( + messages: readonly Message[], + snapshot: MediaStripSnapshot, +): readonly Message[] { + const keys = mediaStripSnapshotKeys(snapshot); + let changed = false; + const result = messages.map((message) => { + let messageChanged = false; + const content = message.content.map((part): ContentPart => { + if (!isDegradableMediaPart(part) || !keys.has(mediaStripKey(part))) return part; + changed = true; + messageChanged = true; + return { type: 'text', text: MEDIA_STRIPPED_PLACEHOLDERS[part.type] }; + }); + return messageChanged ? { ...message, content } : message; + }); + return changed ? result : messages; +} + +export function degradeOlderMediaParts( + messages: readonly Message[], + keepRecent: number, + placeholders: MediaPlaceholderSet = MEDIA_DEGRADED_PLACEHOLDERS, +): readonly Message[] { + const mediaCount = messages.reduce( + (count, message) => count + message.content.filter(isDegradableMediaPart).length, + 0, + ); + let toDegrade = Math.max(0, mediaCount - keepRecent); + if (toDegrade === 0) return messages; + + return messages.map((message) => { + if (toDegrade === 0 || !message.content.some(isDegradableMediaPart)) return message; + const content = message.content.map((part): ContentPart => { + if (toDegrade === 0 || !isDegradableMediaPart(part)) return part; + toDegrade -= 1; + return { type: 'text', text: placeholders[part.type] }; + }); + return { ...message, content }; + }); +} diff --git a/packages/agent-core-v2/src/agent/contextProjector/projection.ts b/packages/agent-core-v2/src/agent/contextProjector/projection.ts new file mode 100644 index 000000000..3b25495cc --- /dev/null +++ b/packages/agent-core-v2/src/agent/contextProjector/projection.ts @@ -0,0 +1,447 @@ +/** + * `contextProjector` domain — rebuilds stored context history into + * provider-valid wire messages and reports every repair through an anomaly + * sink. + * + * The default projection pairs tool calls with their results (a displaced + * result returns to its call, an orphan is dropped, a call left open is + * closed with a synthetic interrupted result), renders stored tool-result + * facts for the model, drops blank text and wholly-vacuous messages, skips + * partial messages, and merges consecutive user prompts. The strict + * projection adds the repairs strict providers need: duplicate tool calls + * dropped, consecutive assistants merged, leading non-user messages dropped. + * + * A history slice without any assistant message is a sizing slice (used to + * size tool results): tool messages project like any other message instead + * of pairing into exchanges. A synthesized close counts as `trailing` — an + * expected in-flight close rather than a defect — exactly when no non-tool, + * non-partial message follows the owning message in the slice. + */ + +import { ErrorCodes, Error2 } from '#/errors'; +import { renderToolResultForModel } from '#/agent/contextMemory/toolResultRender'; +import type { ContextMessage } from '#/agent/contextMemory/types'; +import { isVacuousContentPart } from '#/agent/contextMemory/vacuousContent'; +import type { ContentPart, Message } from '#/kosong/contract/message'; + +export type ProjectionAnomaly = + | { readonly kind: 'tool_result_reordered'; readonly toolCallId: string } + | { readonly kind: 'tool_result_synthesized'; readonly toolCallId: string; readonly trailing: boolean } + | { readonly kind: 'orphan_tool_result_dropped'; readonly toolCallId: string } + | { readonly kind: 'duplicate_tool_call_dropped'; readonly toolCallId: string } + | { readonly kind: 'duplicate_tool_result_dropped'; readonly toolCallId: string } + | { readonly kind: 'leading_non_user_dropped'; readonly role: string } + | { readonly kind: 'consecutive_assistants_merged' } + | { readonly kind: 'whitespace_text_dropped'; readonly role: string } + | { readonly kind: 'vacuous_message_dropped'; readonly role: string }; + +export type OnAnomaly = (anomaly: ProjectionAnomaly) => void; + +export interface ProjectionRepairSummary { + readonly reordered: number; + readonly synthesized: number; + readonly droppedOrphan: number; + readonly duplicateCallsDropped: number; + readonly duplicateResultsDropped: number; + readonly leadingDropped: number; + readonly assistantsMerged: number; + readonly whitespaceDropped: number; + readonly vacuousDropped: number; +} + +export function summarizeProjectionRepairs( + anomalies: readonly ProjectionAnomaly[], +): ProjectionRepairSummary { + const summary = { + reordered: 0, + synthesized: 0, + droppedOrphan: 0, + duplicateCallsDropped: 0, + duplicateResultsDropped: 0, + leadingDropped: 0, + assistantsMerged: 0, + whitespaceDropped: 0, + vacuousDropped: 0, + }; + for (const anomaly of anomalies) { + if (anomaly.kind === 'tool_result_reordered') summary.reordered += 1; + else if (anomaly.kind === 'tool_result_synthesized') summary.synthesized += 1; + else if (anomaly.kind === 'orphan_tool_result_dropped') summary.droppedOrphan += 1; + else if (anomaly.kind === 'duplicate_tool_call_dropped') summary.duplicateCallsDropped += 1; + else if (anomaly.kind === 'duplicate_tool_result_dropped') summary.duplicateResultsDropped += 1; + else if (anomaly.kind === 'leading_non_user_dropped') summary.leadingDropped += 1; + else if (anomaly.kind === 'consecutive_assistants_merged') summary.assistantsMerged += 1; + else if (anomaly.kind === 'vacuous_message_dropped') summary.vacuousDropped += 1; + else summary.whitespaceDropped += 1; + } + return summary; +} + +export function project(history: readonly ContextMessage[], onAnomaly?: OnAnomaly): Message[] { + const layout = sliceLayout(history); + return flattenBlocks(pairBlocks(history, layout, onAnomaly), layout, onAnomaly); +} + +export function projectStrict( + history: readonly ContextMessage[], + onAnomaly?: OnAnomaly, +): Message[] { + const projected = project(history, onAnomaly); + return dropLeadingNonUserMessages( + mergeConsecutiveAssistantMessages(dedupeDuplicateToolCalls(projected, onAnomaly), onAnomaly), + onAnomaly, + ); +} + +interface SliceLayout { + readonly sizing: boolean; + readonly lastNonToolIndex: number; +} + +function sliceLayout(history: readonly ContextMessage[]): SliceLayout { + let sizing = true; + let lastNonToolIndex = -1; + for (const [index, message] of history.entries()) { + if (message.partial === true || message.role === 'tool') continue; + lastNonToolIndex = index; + if (message.role === 'assistant') sizing = false; + } + return { sizing, lastNonToolIndex }; +} + +interface AttachedResult { + readonly source: ContextMessage; + readonly content: ContentPart[]; +} + +const INTERRUPTED_RESULT = Symbol('interruptedResult'); + +interface PendingCall { + readonly callId: string; + result: AttachedResult | typeof INTERRUPTED_RESULT | undefined; + foreignBetween: boolean; +} + +interface Exchange { + readonly source: ContextMessage; + readonly content: ContentPart[]; + readonly ownerIndex: number; + readonly pending: PendingCall[]; +} + +type Block = + | { + readonly kind: 'message'; + readonly source: ContextMessage; + readonly content: ContentPart[]; + } + | { readonly kind: 'exchange'; readonly exchange: Exchange }; + +function pairBlocks( + history: readonly ContextMessage[], + layout: SliceLayout, + onAnomaly?: OnAnomaly, +): Block[] { + const blocks: Block[] = []; + const openCalls = new Map(); + + const markForeignBetween = (): void => { + for (const { pending } of openCalls.values()) pending.foreignBetween = true; + }; + + for (const [index, message] of history.entries()) { + if (message.partial === true) continue; + if (message.role === 'tool' && !layout.sizing) { + if (message.toolCallId === undefined) continue; + const open = openCalls.get(message.toolCallId); + if (open === undefined) { + markForeignBetween(); + onAnomaly?.({ kind: 'orphan_tool_result_dropped', toolCallId: message.toolCallId }); + continue; + } + openCalls.delete(message.toolCallId); + open.pending.result = { source: message, content: projectedContent(message, onAnomaly) }; + if (open.pending.foreignBetween) { + onAnomaly?.({ kind: 'tool_result_reordered', toolCallId: message.toolCallId }); + } + continue; + } + + const content = projectedContent(message, onAnomaly); + if (message.toolCalls.length === 0 && !hasDeclaredTools(message)) { + if (content.length === 0) continue; + if (content.every(isVacuousContentPart)) { + onAnomaly?.({ kind: 'vacuous_message_dropped', role: message.role }); + continue; + } + } + markForeignBetween(); + if (message.toolCalls.length === 0) { + blocks.push({ kind: 'message', source: message, content }); + continue; + } + + const exchange: Exchange = { source: message, content, ownerIndex: index, pending: [] }; + blocks.push({ kind: 'exchange', exchange }); + for (const call of message.toolCalls) { + const superseded = openCalls.get(call.id); + if (superseded !== undefined) { + superseded.pending.result = INTERRUPTED_RESULT; + onAnomaly?.({ + kind: 'tool_result_synthesized', + toolCallId: call.id, + trailing: superseded.exchange.ownerIndex >= layout.lastNonToolIndex, + }); + } + const pending: PendingCall = { callId: call.id, result: undefined, foreignBetween: false }; + exchange.pending.push(pending); + openCalls.set(call.id, { exchange, pending }); + } + } + return blocks; +} + +interface MergeState { + single: { readonly source: ContextMessage; readonly content: ContentPart[] } | undefined; + readonly texts: string[]; + readonly parts: ContentPart[]; +} + +function flattenBlocks( + blocks: readonly Block[], + layout: SliceLayout, + onAnomaly?: OnAnomaly, +): Message[] { + const out: Message[] = []; + let merge: MergeState | undefined; + + const flushMerge = (): void => { + if (merge === undefined) return; + if (merge.single !== undefined) { + out.push(toWireMessage(merge.single.source, merge.single.content)); + } else { + const text = merge.texts.join('\n\n'); + const content: ContentPart[] = text === '' ? [] : [{ type: 'text', text }]; + content.push(...merge.parts); + out.push({ + role: 'user', + name: undefined, + content, + toolCalls: [], + toolCallId: undefined, + partial: undefined, + }); + } + merge = undefined; + }; + + for (const block of blocks) { + if (block.kind === 'message') { + if (canMergeUserMessage(block.source)) { + if (merge === undefined) { + merge = { single: block, texts: [], parts: [] }; + } else { + if (merge.single !== undefined) { + appendMergeContent(merge, merge.single.content); + merge.single = undefined; + } + appendMergeContent(merge, block.content); + } + continue; + } + flushMerge(); + out.push(toWireMessage(block.source, block.content)); + continue; + } + + flushMerge(); + const { exchange } = block; + out.push(toWireMessage(exchange.source, exchange.content)); + for (const pending of exchange.pending) { + if (pending.result === undefined) { + out.push(createInterruptedToolResult(pending.callId)); + onAnomaly?.({ + kind: 'tool_result_synthesized', + toolCallId: pending.callId, + trailing: exchange.ownerIndex >= layout.lastNonToolIndex, + }); + } else if (pending.result === INTERRUPTED_RESULT) { + out.push(createInterruptedToolResult(pending.callId)); + } else { + out.push(toWireMessage(pending.result.source, pending.result.content)); + } + } + } + flushMerge(); + return out; +} + +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)) { + onAnomaly?.({ kind: 'duplicate_tool_call_dropped', toolCallId: toolCall.id }); + return false; + } + seenToolCallIds.add(toolCall.id); + return true; + }); + if (kept.length === message.toolCalls.length) { + out.push(message); + } else if (kept.length > 0 || !message.content.every(isVacuousContentPart)) { + out.push({ ...message, toolCalls: kept }); + } else if (message.content.length > 0) { + onAnomaly?.({ kind: 'vacuous_message_dropped', role: message.role }); + } + continue; + } + if (message.role === 'tool' && message.toolCallId !== undefined) { + const previousIndex = keptToolResultIndexes.get(message.toolCallId); + if (previousIndex !== undefined) { + if (isInterruptedToolResult(out[previousIndex]) && !isInterruptedToolResult(message)) { + out[previousIndex] = message; + } else { + onAnomaly?.({ kind: 'duplicate_tool_result_dropped', toolCallId: message.toolCallId }); + } + continue; + } + keptToolResultIndexes.set(message.toolCallId, out.length); + } + out.push(message); + } + return out; +} + +function mergeConsecutiveAssistantMessages( + messages: readonly Message[], + onAnomaly?: OnAnomaly, +): Message[] { + const out: Message[] = []; + for (const message of messages) { + const previous = out.at(-1); + if (previous !== undefined && previous.role === 'assistant' && message.role === 'assistant') { + out[out.length - 1] = { + ...previous, + content: [...previous.content, ...message.content], + toolCalls: [...previous.toolCalls, ...message.toolCalls], + }; + onAnomaly?.({ kind: 'consecutive_assistants_merged' }); + continue; + } + out.push(message); + } + return out; +} + +function dropLeadingNonUserMessages(messages: readonly Message[], onAnomaly?: OnAnomaly): Message[] { + let start = 0; + 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); +} + +function appendMergeContent(group: MergeState, content: readonly ContentPart[]): void { + let text = ''; + for (const part of content) { + if (part.type === 'text') text += part.text; + else group.parts.push(part); + } + if (text.length > 0) group.texts.push(text); +} + +function projectedContent(source: ContextMessage, onAnomaly?: OnAnomaly): ContentPart[] { + const content = + source.role === 'tool' + ? renderToolResultForModel({ + output: outputFromToolContent(source.content), + isError: source.isError, + note: source.note, + }) + : source.content; + return cleanContent(source, content, onAnomaly); +} + +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)) { + 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 Error2( + ErrorCodes.REQUEST_INVALID, + 'Tool result message content cannot be empty after removing empty text blocks.', + { details: { toolCallId: source.toolCallId } }, + ); + } + return [...content]; +} + +function outputFromToolContent(content: readonly ContentPart[]): string | readonly ContentPart[] { + const only = content[0]; + return content.length === 1 && only?.type === 'text' ? only.text : content; +} + +const TOOL_INTERRUPTED_TEXT = + 'Tool result is not available in the current context. Do not assume the tool completed successfully.'; + +function createInterruptedToolResult(toolCallId: string): Message { + return { + role: 'tool', + name: undefined, + content: [{ type: 'text', text: TOOL_INTERRUPTED_TEXT }], + toolCalls: [], + toolCallId, + partial: undefined, + }; +} + +function isInterruptedToolResult(message: Message | undefined): boolean { + if (message?.role !== 'tool') return false; + const [part] = message.content; + return part?.type === 'text' && part.text === TOOL_INTERRUPTED_TEXT; +} + +function isBlankText(part: ContentPart): boolean { + return part.type === 'text' && part.text.trim().length === 0; +} + +function canMergeUserMessage(message: ContextMessage): boolean { + return message.role === 'user' && message.origin?.kind === 'user'; +} + +function hasDeclaredTools(message: ContextMessage): boolean { + return message.tools !== undefined && message.tools.length > 0; +} + +function toWireMessage(message: ContextMessage, content: ContentPart[]): Message { + return { + role: message.role, + name: message.name, + content, + toolCalls: message.toolCalls, + toolCallId: message.toolCallId, + partial: message.partial, + tools: message.tools, + }; +} diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index 7dffef792..853e9d9b5 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -604,6 +604,7 @@ export * from '#/features/dateChange/dateChangeService'; import '#/features/dateChange/dateChangeFeature'; export * from '#/agent/contextProjector/contextProjector'; export * from '#/agent/contextProjector/contextProjectorService'; +export * from '#/agent/contextProjector/mediaProjection'; export * from '#/agent/tokenCounting/tokenCounting'; export * from '#/agent/tokenCounting/tokenCountingOps'; export * from '#/agent/tokenCounting/tokenCountingService';