From 8e31454bcb6bbba27199dc73fe257b50f198dc28 Mon Sep 17 00:00:00 2001 From: Robin Newhouse Date: Fri, 26 Jun 2026 09:59:54 -0700 Subject: [PATCH] Simplify compaction sidecar kill switch --- .../runtime/interactive/session-runtime.ts | 39 +-- sdk/packages/core/src/cline-core/types.ts | 8 +- .../handlers/connector-handlers.test.ts | 3 +- .../core/src/hub/server/handlers/context.ts | 3 +- .../hub/server/handlers/run-handlers.test.ts | 3 +- .../hub/server/handlers/session-handlers.ts | 25 +- .../src/hub/server/hub-server-transport.ts | 10 +- sdk/packages/core/src/index.ts | 8 +- .../src/runtime/host/local-runtime-host.ts | 226 +++++++++--------- .../src/session/models/session-compaction.ts | 70 ++++++ 10 files changed, 243 insertions(+), 152 deletions(-) diff --git a/apps/cli/src/runtime/interactive/session-runtime.ts b/apps/cli/src/runtime/interactive/session-runtime.ts index 1b2ecb8280..29970863ad 100644 --- a/apps/cli/src/runtime/interactive/session-runtime.ts +++ b/apps/cli/src/runtime/interactive/session-runtime.ts @@ -3,6 +3,7 @@ import { type AgentHooks, type CheckpointEntry, createSessionCompactionState, + createSessionCompactionSidecarAccess, createSessionCompactionSidecarEnabledResolver, isSessionNotFoundError, type PendingPromptMutationResult, @@ -103,6 +104,9 @@ export function createInteractiveSessionRuntime(input: { const isCompactionSidecarEnabled = input.isCompactionSidecarEnabled ?? createSessionCompactionSidecarEnabledResolver(); + const compactionSidecar = createSessionCompactionSidecarAccess( + isCompactionSidecarEnabled, + ); let runtimeHooks: RuntimeHooks | undefined; let unsubscribeAgent = () => {}; let unsubscribePendingPrompts = () => {}; @@ -209,14 +213,17 @@ export function createInteractiveSessionRuntime(input: { ): Promise => { const generation = sessionStartGeneration; const manager = await ensureSessionManager(); + const sidecarInitialCompactionState = compactionSidecar.initialState( + initialCompactionState, + ); const started = await manager.start({ source: SessionSource.CLI, config: buildSessionConfig(), toolPolicies: input.config.toolPolicies, interactive: true, initialMessages: initial, - ...(isCompactionSidecarEnabled() && initialCompactionState - ? { initialCompactionState } + ...(sidecarInitialCompactionState + ? { initialCompactionState: sidecarInitialCompactionState } : {}), ...(sessionMetadata ? { sessionMetadata } : {}), localRuntime: { @@ -298,15 +305,14 @@ export function createInteractiveSessionRuntime(input: { const readCompactionState = async ( sessionId: string, ): Promise => { - if (!isCompactionSidecarEnabled()) { - return undefined; - } const manager = sessionManager; if (!manager) { return undefined; } try { - return await manager.readSessionCompactionState(sessionId); + return await compactionSidecar.read(() => + manager.readSessionCompactionState(sessionId), + ); } catch (error) { input.config.logger?.log?.("Failed to read session compaction state", { sessionId, @@ -634,25 +640,20 @@ export function createInteractiveSessionRuntime(input: { compacted: false, }; } - if (!isCompactionSidecarEnabled()) { - return { - messagesBefore, - messagesAfter: result.canonicalMessages.length, - workingContextMessagesAfter: result.compactionState.messages.length, - compacted: true, - }; - } - const updated = await manager.updateSessionCompactionState( - sourceSessionId, - result.compactionState, + const compactionState = result.compactionState; + const updated = await compactionSidecar.update(() => + manager.updateSessionCompactionState( + sourceSessionId, + compactionState, + ), ); - if (!updated.updated) { + if (!updated.updated && !updated.disabled) { throw new Error("Compaction could not be saved. Try again."); } return { messagesBefore, messagesAfter: result.canonicalMessages.length, - workingContextMessagesAfter: result.compactionState?.messages.length, + workingContextMessagesAfter: compactionState.messages.length, compacted: true, }; }; diff --git a/sdk/packages/core/src/cline-core/types.ts b/sdk/packages/core/src/cline-core/types.ts index 2f3c8866fc..5269d55e25 100644 --- a/sdk/packages/core/src/cline-core/types.ts +++ b/sdk/packages/core/src/cline-core/types.ts @@ -126,8 +126,10 @@ export interface ClineCoreAutomationApi { export type ClineCoreListHistoryOptions = SessionHistoryListOptions; -export interface ClineCoreStartInput - extends Omit { +export interface ClineCoreStartInput extends Omit< + StartSessionInput, + "config" | "localRuntime" +> { config: CoreSessionConfig; localRuntime?: LocalRuntimeStartOptions; } @@ -221,7 +223,7 @@ export interface ClineCoreOptions { */ featureFlags?: FeatureFlagsService; /** - * Overrides the compaction sidecar rollout decision. + * Gets whether persisted compaction sidecar reads and writes are enabled. * @internal */ isCompactionSidecarEnabled?: () => boolean; diff --git a/sdk/packages/core/src/hub/server/handlers/connector-handlers.test.ts b/sdk/packages/core/src/hub/server/handlers/connector-handlers.test.ts index 1b182fb06f..934c7518ab 100644 --- a/sdk/packages/core/src/hub/server/handlers/connector-handlers.test.ts +++ b/sdk/packages/core/src/hub/server/handlers/connector-handlers.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import type { HubCommandEnvelope } from "@cline/shared"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { createSessionCompactionSidecarAccess } from "../../../session/models/session-compaction"; import { __test__, handleConnectorCommand } from "./connector-handlers"; import type { HubTransportContext } from "./context"; @@ -34,7 +35,7 @@ describe("connector hub handlers", () => { pendingCapabilityRequests: new Map(), suppressNextTerminalEventBySession: new Map(), telemetry: telemetry as never, - isCompactionSidecarEnabled: () => true, + compactionSidecar: createSessionCompactionSidecarAccess(() => true), sessionHost: {} as never, publish: vi.fn(), buildEvent: vi.fn() as never, diff --git a/sdk/packages/core/src/hub/server/handlers/context.ts b/sdk/packages/core/src/hub/server/handlers/context.ts index 2cbbb1cca1..18e758a3a3 100644 --- a/sdk/packages/core/src/hub/server/handlers/context.ts +++ b/sdk/packages/core/src/hub/server/handlers/context.ts @@ -14,6 +14,7 @@ import type { RuntimeHost, SessionUsageRuntimeService, } from "../../../runtime/host/runtime-host"; +import type { SessionCompactionSidecarAccess } from "../../../session/models/session-compaction"; import { type CoreSessionSnapshot, createCoreSessionSnapshot, @@ -51,7 +52,7 @@ export interface HubTransportContext { readonly pendingCapabilityRequests: Map; readonly suppressNextTerminalEventBySession: Map; readonly telemetry?: ITelemetryService; - readonly isCompactionSidecarEnabled: () => boolean; + readonly compactionSidecar: SessionCompactionSidecarAccess; readonly sessionHost: RuntimeHost & Partial; publish(event: HubEventEnvelope): void; diff --git a/sdk/packages/core/src/hub/server/handlers/run-handlers.test.ts b/sdk/packages/core/src/hub/server/handlers/run-handlers.test.ts index 392f303c09..3606e63e19 100644 --- a/sdk/packages/core/src/hub/server/handlers/run-handlers.test.ts +++ b/sdk/packages/core/src/hub/server/handlers/run-handlers.test.ts @@ -1,6 +1,7 @@ import type { HubEventEnvelope } from "@cline/shared"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { RuntimeHost } from "../../../runtime/host/runtime-host"; +import { createSessionCompactionSidecarAccess } from "../../../session/models/session-compaction"; import { buildHubEvent, type HubTransportContext } from "./context"; import { handleRunAbort, handleSessionInput } from "./run-handlers"; @@ -14,7 +15,7 @@ function createContext( pendingApprovals: new Map(), pendingCapabilityRequests: new Map(), suppressNextTerminalEventBySession: new Map(), - isCompactionSidecarEnabled: () => true, + compactionSidecar: createSessionCompactionSidecarAccess(() => true), sessionHost: { startSession: vi.fn(), runTurn: vi.fn(), diff --git a/sdk/packages/core/src/hub/server/handlers/session-handlers.ts b/sdk/packages/core/src/hub/server/handlers/session-handlers.ts index a8816dadf5..3e31261e3d 100644 --- a/sdk/packages/core/src/hub/server/handlers/session-handlers.ts +++ b/sdk/packages/core/src/hub/server/handlers/session-handlers.ts @@ -111,9 +111,9 @@ export async function handleSessionCreate( payload.runtimeOptions && typeof payload.runtimeOptions === "object" ? (payload.runtimeOptions as Record) : {}; - const initialCompactionState = ctx.isCompactionSidecarEnabled() - ? parseSessionCompactionState(payload.initialCompactionState) - : undefined; + const initialCompactionState = ctx.compactionSidecar.initialState( + parseSessionCompactionState(payload.initialCompactionState), + ); if (typeof sessionConfig?.mode === "string") { metadata.mode = sessionConfig.mode; } else if (typeof runtimeOptions.mode === "string") { @@ -387,9 +387,9 @@ export async function handleSessionRestore( payload.runtimeOptions && typeof payload.runtimeOptions === "object" ? (payload.runtimeOptions as Record) : {}; - const initialCompactionState = ctx.isCompactionSidecarEnabled() - ? parseSessionCompactionState(payload.initialCompactionState) - : undefined; + const initialCompactionState = ctx.compactionSidecar.initialState( + parseSessionCompactionState(payload.initialCompactionState), + ); const metadata = payload.metadata && typeof payload.metadata === "object" ? JSON.parse(JSON.stringify(payload.metadata)) @@ -757,7 +757,7 @@ export async function handleSessionCompactionGet( `Unknown session: ${sessionId}`, ); } - if (!ctx.isCompactionSidecarEnabled()) { + if (!ctx.compactionSidecar.enabled) { return okReply(envelope, { sessionId, state: undefined, disabled: true }); } const clientId = envelope.clientId?.trim() || "hub-client"; @@ -770,7 +770,9 @@ export async function handleSessionCompactionGet( if (unauthorized) { return unauthorized; } - const state = await ctx.sessionHost.readSessionCompactionState(sessionId); + const state = await ctx.compactionSidecar.read(() => + ctx.sessionHost.readSessionCompactionState(sessionId), + ); return okReply(envelope, { sessionId, state }); } @@ -844,7 +846,7 @@ export async function handleSessionCompactionUpdate( `Unknown session: ${sessionId}`, ); } - if (!ctx.isCompactionSidecarEnabled()) { + if (!ctx.compactionSidecar.enabled) { return okReply(envelope, { sessionId, updated: false, disabled: true }); } const unauthorized = authorizeSessionCompactionAccess({ @@ -868,9 +870,8 @@ export async function handleSessionCompactionUpdate( "session.compaction.update requires a valid compaction state", ); } - const updated = await ctx.sessionHost.updateSessionCompactionState( - sessionId, - state, + const updated = await ctx.compactionSidecar.update(() => + ctx.sessionHost.updateSessionCompactionState(sessionId, state), ); const [updatedSession, snapshot] = updated.updated ? await Promise.all([ diff --git a/sdk/packages/core/src/hub/server/hub-server-transport.ts b/sdk/packages/core/src/hub/server/hub-server-transport.ts index d9c45d40b9..692c214018 100644 --- a/sdk/packages/core/src/hub/server/hub-server-transport.ts +++ b/sdk/packages/core/src/hub/server/hub-server-transport.ts @@ -15,7 +15,10 @@ import type { RuntimeHost, } from "../../runtime/host/runtime-host"; import { SqliteSessionStore } from "../../services/storage/sqlite-session-store"; -import { createSessionCompactionSidecarEnabledResolver } from "../../session/models/session-compaction"; +import { + createSessionCompactionSidecarAccess, + createSessionCompactionSidecarEnabledResolver, +} from "../../session/models/session-compaction"; import { CoreSessionService } from "../../session/services/session-service"; import { type CoreSettingsListInput, @@ -187,6 +190,9 @@ export class HubServerTransport implements NativeHubTransport { const isCompactionSidecarEnabled = options.isCompactionSidecarEnabled ?? createSessionCompactionSidecarEnabledResolver(); + const compactionSidecar = createSessionCompactionSidecarAccess( + isCompactionSidecarEnabled, + ); this.sessionHost = options.sessionHost ?? new LocalRuntimeHost({ @@ -203,7 +209,7 @@ export class HubServerTransport implements NativeHubTransport { suppressNextTerminalEventBySession: this.suppressNextTerminalEventBySession, telemetry: options.telemetry, - isCompactionSidecarEnabled, + compactionSidecar, sessionHost: this.sessionHost, publish: (event) => this.publish(event), buildEvent: buildHubEvent, diff --git a/sdk/packages/core/src/index.ts b/sdk/packages/core/src/index.ts index b20a904d6c..206b7d46b8 100644 --- a/sdk/packages/core/src/index.ts +++ b/sdk/packages/core/src/index.ts @@ -730,10 +730,7 @@ export type { CoreSettingsToggleInput, CoreSettingsType, } from "./settings"; -export { - CoreSettingsService, - createCoreSettingsService, -} from "./settings"; +export { CoreSettingsService, createCoreSettingsService } from "./settings"; export type { ChatMessage, ChatSessionConfig, @@ -879,10 +876,13 @@ export { type TelemetryServiceOptions, } from "./services/telemetry/TelemetryService"; export { + createSessionCompactionSidecarAccess, createSessionCompactionSidecarEnabledResolver, createSessionCompactionState, parseSessionCompactionState, projectSessionCompactionState, + type SessionCompactionSidecarAccess, + type SessionCompactionSidecarUpdateResult, type SessionCompactionState, } from "./session/models/session-compaction"; // Compatibility barrel (legacy imports). diff --git a/sdk/packages/core/src/runtime/host/local-runtime-host.ts b/sdk/packages/core/src/runtime/host/local-runtime-host.ts index e87188f3c1..fa70545a35 100644 --- a/sdk/packages/core/src/runtime/host/local-runtime-host.ts +++ b/sdk/packages/core/src/runtime/host/local-runtime-host.ts @@ -51,8 +51,10 @@ import { } from "../../services/usage"; import { enrichPromptWithMentions } from "../../services/workspace"; import { + createSessionCompactionSidecarAccess, createSessionCompactionSidecarEnabledResolver, projectSessionCompactionState, + type SessionCompactionSidecarAccess, type SessionCompactionState, } from "../../session/models/session-compaction"; import { @@ -224,7 +226,7 @@ export class LocalRuntimeHost implements RuntimeHost { private readonly oauthTokenManager: RuntimeOAuthTokenManager; private readonly defaultTelemetry?: ITelemetryService; private readonly defaultFetch?: typeof fetch; - private readonly isCompactionSidecarEnabled: () => boolean; + private readonly compactionSidecar: SessionCompactionSidecarAccess; private readonly events = new RuntimeHostEventBus(); private readonly sessions = new Map(); private readonly usageBySession = new Map(); @@ -261,9 +263,10 @@ export class LocalRuntimeHost implements RuntimeHost { this.defaultTelemetry = options.telemetry; this.defaultTelemetry?.setDistinctId(distinctId); this.defaultFetch = options.fetch; - this.isCompactionSidecarEnabled = + this.compactionSidecar = createSessionCompactionSidecarAccess( options.isCompactionSidecarEnabled ?? - createSessionCompactionSidecarEnabledResolver(); + createSessionCompactionSidecarEnabledResolver(), + ); this.pendingPromptsController = new PendingPromptsController({ getSession: (sid) => this.sessions.get(sid), @@ -381,18 +384,18 @@ export class LocalRuntimeHost implements RuntimeHost { ); if (existingManifest) { manifest = existingManifest; - resumedArtifacts = { - manifestPath, - messagesPath: existingManifest.messages_path || messagesPath, - compactionPath: existingManifest.compaction_path, - manifest: existingManifest, - }; - resumedCompactionState = this.isCompactionSidecarEnabled() - ? await this.invokeOptionalValue( - "readSessionCompactionState", - sessionId, - ) - : undefined; + resumedArtifacts = { + manifestPath, + messagesPath: existingManifest.messages_path || messagesPath, + compactionPath: existingManifest.compaction_path, + manifest: existingManifest, + }; + resumedCompactionState = await this.compactionSidecar.read(() => + this.invokeOptionalValue( + "readSessionCompactionState", + sessionId, + ), + ); } } const initialAggregateUsage = await this.seedAggregateUsageFromArtifacts({ @@ -492,64 +495,70 @@ export class LocalRuntimeHost implements RuntimeHost { const explicitInitialCompactionState = startInput.initialCompactionState; let activeSessionRef: ActiveSession | undefined; const compact = createContextCompactionPrepareTurn(configWithProvider); - const sidecarEnabled = this.isCompactionSidecarEnabled(); const rawInitialCompactionState = explicitInitialCompactionState ?? resumedCompactionState; - const initialCompactionState = - sidecarEnabled && compact && rawInitialCompactionState + const initialCompactionState = this.compactionSidecar.initialState( + compact && rawInitialCompactionState ? { ...rawInitialCompactionState, conversation_id: rawInitialCompactionState.conversation_id?.trim() || sessionId, } - : undefined; + : undefined, + ); const prepareTurn = compact ? createCompactionStateAwarePrepareTurn({ compact, getState: () => - sidecarEnabled ? activeSessionRef?.compactionState : undefined, + this.compactionSidecar.initialState( + activeSessionRef?.compactionState, + ), saveState: async (state) => { - if (!sidecarEnabled) return; - const activeSession = activeSessionRef; - if (!activeSession) return; - const stateForSession = { - ...state, - conversation_id: activeSession.sessionId, - }; - try { - const result = await this.persistActiveSessionCompactionState( - activeSession, - stateForSession, - ); - if (!result.updated) { - configWithProvider.logger?.debug?.( - "Skipped stale session compaction state", - { - sessionId: activeSession.sessionId, - sourceMessageCount: stateForSession.source_message_count, - }, + await this.compactionSidecar.update(async () => { + const activeSession = activeSessionRef; + if (!activeSession) return { updated: false }; + const stateForSession = { + ...state, + conversation_id: activeSession.sessionId, + }; + try { + const result = await this.persistActiveSessionCompactionState( + activeSession, + stateForSession, ); + if (!result.updated) { + configWithProvider.logger?.debug?.( + "Skipped stale session compaction state", + { + sessionId: activeSession.sessionId, + sourceMessageCount: + stateForSession.source_message_count, + }, + ); + } + return result; + } catch (error) { + configWithProvider.logger?.error?.( + "Failed to persist session compaction state", + { sessionId: activeSession.sessionId, error }, + ); + captureSdkError(configWithProvider.telemetry, { + component: "core", + operation: "session.persist_compaction_state", + severity: "warn", + handled: true, + error, + context: { + sessionId: activeSession.sessionId, + providerId: configWithProvider.providerId, + modelId: configWithProvider.modelId, + }, + }); + return { updated: false }; } - } catch (error) { - configWithProvider.logger?.error?.( - "Failed to persist session compaction state", - { sessionId: activeSession.sessionId, error }, - ); - captureSdkError(configWithProvider.telemetry, { - component: "core", - operation: "session.persist_compaction_state", - severity: "warn", - handled: true, - error, - context: { - sessionId: activeSession.sessionId, - providerId: configWithProvider.providerId, - modelId: configWithProvider.modelId, - }, - }); - } + }); }, - }) + }) : undefined; const agentConfig = { @@ -1057,66 +1066,65 @@ export class LocalRuntimeHost implements RuntimeHost { sessionId: string, state: SessionCompactionState, ): Promise<{ updated: boolean }> { - if (!this.isCompactionSidecarEnabled()) { - return { updated: false }; - } - const target = sessionId.trim(); - if (!target) return { updated: false }; - const activeSession = this.sessions.get(target); - const sessionRecord = activeSession - ? undefined - : await this.getSession(target); - const existing = activeSession ?? sessionRecord; - if (!existing) return { updated: false }; - if ( - !(await this.canPersistCompactionState( + const result = await this.compactionSidecar.update(async () => { + const target = sessionId.trim(); + if (!target) return { updated: false }; + const activeSession = this.sessions.get(target); + const sessionRecord = activeSession + ? undefined + : await this.getSession(target); + const existing = activeSession ?? sessionRecord; + if (!existing) return { updated: false }; + if ( + !(await this.canPersistCompactionState( + target, + state, + activeSession, + sessionRecord, + )) + ) { + return { updated: false }; + } + if (activeSession) { + return await this.persistActiveSessionCompactionState( + activeSession, + state, + ); + } + const current = await this.invokeOptionalValue( + "readSessionCompactionState", target, - state, - activeSession, - sessionRecord, - )) - ) { - return { updated: false }; - } - if (activeSession) { - return await this.persistActiveSessionCompactionState( - activeSession, - state, ); - } - const current = await this.invokeOptionalValue( - "readSessionCompactionState", - target, - ); - if (isIncomingCompactionStateStale(state, current)) { - return { updated: false }; - } - await this.invoke("persistSessionCompactionState", target, state); - return { updated: true }; + if (isIncomingCompactionStateStale(state, current)) { + return { updated: false }; + } + await this.invoke("persistSessionCompactionState", target, state); + return { updated: true }; + }); + return { updated: result.updated }; } async readSessionCompactionState( sessionId: string, ): Promise { - if (!this.isCompactionSidecarEnabled()) { - return undefined; - } - const target = sessionId.trim(); - if (!target) return undefined; - const activeSession = this.sessions.get(target); - if (activeSession) { - for (;;) { - const pendingWrite = activeSession.compactionStateWriteQueue; - if (!pendingWrite) { - return activeSession.compactionState; + return await this.compactionSidecar.read(async () => { + const target = sessionId.trim(); + if (!target) return undefined; + const activeSession = this.sessions.get(target); + if (activeSession) { + for (;;) { + const pendingWrite = activeSession.compactionStateWriteQueue; + if (!pendingWrite) { + return activeSession.compactionState; + } + await pendingWrite.catch(() => undefined); } - await pendingWrite.catch(() => undefined); } - } - return await this.invokeOptionalValue( - "readSessionCompactionState", - target, - ); + return await this.invokeOptionalValue( + "readSessionCompactionState", + target, + ); + }); } private isCompactionStateForSession( diff --git a/sdk/packages/core/src/session/models/session-compaction.ts b/sdk/packages/core/src/session/models/session-compaction.ts index 9b31f62501..b7c5b594e4 100644 --- a/sdk/packages/core/src/session/models/session-compaction.ts +++ b/sdk/packages/core/src/session/models/session-compaction.ts @@ -49,6 +49,76 @@ export function createSessionCompactionSidecarEnabledResolver( featureFlags?.getFlagPayload(FeatureFlag.COMPACTION_SIDECAR) !== false; } +export type SessionCompactionSidecarUpdateResult = { + updated: boolean; + disabled?: true; +}; + +export interface SessionCompactionSidecarAccess { + readonly enabled: boolean; + initialState( + state: SessionCompactionState | undefined, + ): SessionCompactionState | undefined; + read(read: () => Promise): Promise; + update( + update: () => Promise<{ updated: boolean }>, + ): Promise; +} + +type SessionCompactionSidecarRole = SessionCompactionSidecarAccess; + +class EnabledSessionCompactionSidecar implements SessionCompactionSidecarRole { + readonly enabled = true; + + initialState( + state: SessionCompactionState | undefined, + ): SessionCompactionState | undefined { + return state; + } + + async read(read: () => Promise): Promise { + return await read(); + } + + async update( + update: () => Promise<{ updated: boolean }>, + ): Promise { + return await update(); + } +} + +class DisabledSessionCompactionSidecar implements SessionCompactionSidecarRole { + readonly enabled = false; + + initialState(): undefined { + return undefined; + } + + async read(): Promise { + return undefined; + } + + async update(): Promise { + return { updated: false, disabled: true }; + } +} + +export function createSessionCompactionSidecarAccess( + isEnabled: () => boolean = createSessionCompactionSidecarEnabledResolver(), +): SessionCompactionSidecarAccess { + const enabled = new EnabledSessionCompactionSidecar(); + const disabled = new DisabledSessionCompactionSidecar(); + const current = () => (isEnabled() ? enabled : disabled); + return { + get enabled() { + return current().enabled; + }, + initialState: (state) => current().initialState(state), + read: (read) => current().read(read), + update: (update) => current().update(update), + }; +} + function cloneMessages( messages: readonly MessageWithMetadata[], ): MessageWithMetadata[] {