opensource/cline
3
0
Fork 0
mirror of https://github.com/cline/cline.git synced 2026-09-19 02:05:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

Simplify compaction sidecar kill switch

Browse Source
This commit is contained in:
Robin Newhouse
2026-06-26 10:05:07 -07:00
parent d40334a757
commit 8e31454bcb
10 changed files with 243 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/cli/src/runtime/interactive/session-runtime.ts
+20 -19
View File
@@ -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<void> => {
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<SessionCompactionState | undefined> => {
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,
};
};
sdk/packages/core/src/cline-core/types.ts
+5 -3
View File
@@ -126,8 +126,10 @@ export interface ClineCoreAutomationApi {
export type ClineCoreListHistoryOptions = SessionHistoryListOptions;
export interface ClineCoreStartInput
extends Omit<StartSessionInput, "config" | "localRuntime"> {
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;
sdk/packages/core/src/hub/server/handlers/connector-handlers.test.ts
+2 -1
View File
@@ -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,
sdk/packages/core/src/hub/server/handlers/context.ts
+2 -1
View File
@@ -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<string, PendingCapabilityRequest>;
readonly suppressNextTerminalEventBySession: Map<string, string>;
readonly telemetry?: ITelemetryService;
readonly isCompactionSidecarEnabled: () => boolean;
readonly compactionSidecar: SessionCompactionSidecarAccess;
readonly sessionHost: RuntimeHost &
Partial<PendingPromptsRuntimeService & SessionUsageRuntimeService>;
publish(event: HubEventEnvelope): void;
sdk/packages/core/src/hub/server/handlers/run-handlers.test.ts
+2 -1
View File
@@ -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(),
sdk/packages/core/src/hub/server/handlers/session-handlers.ts
+13 -12
View File
@@ -111,9 +111,9 @@ export async function handleSessionCreate(
payload.runtimeOptions && typeof payload.runtimeOptions === "object"
? (payload.runtimeOptions as Record<string, unknown>)
: {};
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<string, unknown>)
: {};
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([
sdk/packages/core/src/hub/server/hub-server-transport.ts
+8 -2
View File
@@ -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,
sdk/packages/core/src/index.ts
+4 -4
View File
@@ -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).
sdk/packages/core/src/runtime/host/local-runtime-host.ts
+117 -109
View File
@@ -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<string, ActiveSession>();
private readonly usageBySession = new Map<string, SessionAccumulatedUsage>();
@@ -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<SessionCompactionState>(
"readSessionCompactionState",
sessionId,
)
: undefined;
resumedArtifacts = {
manifestPath,
messagesPath: existingManifest.messages_path || messagesPath,
compactionPath: existingManifest.compaction_path,
manifest: existingManifest,
};
resumedCompactionState = await this.compactionSidecar.read(() =>
this.invokeOptionalValue<SessionCompactionState>(
"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<SessionCompactionState>(
"readSessionCompactionState",
target,
state,
activeSession,
sessionRecord,
))
) {
return { updated: false };
}
if (activeSession) {
return await this.persistActiveSessionCompactionState(
activeSession,
state,
);
}
const current = await this.invokeOptionalValue<SessionCompactionState>(
"readSessionCompactionState",
target,
);
if (isIncomingCompactionStateStale(state, current)) {
return { updated: false };
}
await this.invoke<void>("persistSessionCompactionState", target, state);
return { updated: true };
if (isIncomingCompactionStateStale(state, current)) {
return { updated: false };
}
await this.invoke<void>("persistSessionCompactionState", target, state);
return { updated: true };
});
return { updated: result.updated };
}
async readSessionCompactionState(
sessionId: string,
): Promise<SessionCompactionState | undefined> {
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<SessionCompactionState>(
"readSessionCompactionState",
target,
);
return await this.invokeOptionalValue<SessionCompactionState>(
"readSessionCompactionState",
target,
);
});
}
private isCompactionStateForSession(
sdk/packages/core/src/session/models/session-compaction.ts
+70
View File
@@ -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<T>(read: () => Promise<T | undefined>): Promise<T | undefined>;
update(
update: () => Promise<{ updated: boolean }>,
): Promise<SessionCompactionSidecarUpdateResult>;
}
type SessionCompactionSidecarRole = SessionCompactionSidecarAccess;
class EnabledSessionCompactionSidecar implements SessionCompactionSidecarRole {
readonly enabled = true;
initialState(
state: SessionCompactionState | undefined,
): SessionCompactionState | undefined {
return state;
}
async read<T>(read: () => Promise<T | undefined>): Promise<T | undefined> {
return await read();
}
async update(
update: () => Promise<{ updated: boolean }>,
): Promise<SessionCompactionSidecarUpdateResult> {
return await update();
}
}
class DisabledSessionCompactionSidecar implements SessionCompactionSidecarRole {
readonly enabled = false;
initialState(): undefined {
return undefined;
}
async read(): Promise<undefined> {
return undefined;
}
async update(): Promise<SessionCompactionSidecarUpdateResult> {
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[] {