From cf2227e8a5222ad9bd1167b573b62599d0efd906 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Tue, 26 May 2026 14:56:32 +0800 Subject: [PATCH] fix: warn instead of error on newer wire protocol version (#49) * fix: warn instead of error on newer wire protocol version * fix * fix test --- .changeset/warn-newer-wire-version.md | 7 +++++++ apps/kimi-code/src/tui/kimi-tui.ts | 16 ++++++++++++++++ packages/agent-core/src/agent/index.ts | 5 +++-- packages/agent-core/src/agent/records/index.ts | 14 +++++++++++--- .../src/agent/records/migration/index.ts | 11 +++++------ packages/agent-core/src/rpc/core-impl.ts | 8 ++++++-- packages/agent-core/src/rpc/events.ts | 9 +++++++-- packages/agent-core/src/rpc/resumed.ts | 1 + packages/agent-core/src/session/index.ts | 10 ++++++++-- .../agent-core/test/agent/records/index.test.ts | 8 ++++---- packages/node-sdk/src/events.ts | 1 + packages/node-sdk/src/session.ts | 1 + packages/node-sdk/src/types.ts | 2 +- .../node-sdk/test/session-event-types.test.ts | 1 + 14 files changed, 72 insertions(+), 22 deletions(-) create mode 100644 .changeset/warn-newer-wire-version.md diff --git a/.changeset/warn-newer-wire-version.md b/.changeset/warn-newer-wire-version.md new file mode 100644 index 000000000..7988cda5b --- /dev/null +++ b/.changeset/warn-newer-wire-version.md @@ -0,0 +1,7 @@ +--- +"@moonshot-ai/agent-core": patch +"@moonshot-ai/kimi-code-sdk": patch +"@moonshot-ai/kimi-code": patch +--- + +Resume sessions with a newer wire protocol version instead of failing. A warning is now shown in the TUI and records are replayed without migration. diff --git a/apps/kimi-code/src/tui/kimi-tui.ts b/apps/kimi-code/src/tui/kimi-tui.ts index 08b1d8451..91273ec6f 100644 --- a/apps/kimi-code/src/tui/kimi-tui.ts +++ b/apps/kimi-code/src/tui/kimi-tui.ts @@ -88,6 +88,7 @@ import type { TurnStepCompletedEvent, TurnStepInterruptedEvent, TurnStepStartedEvent, + WarningEvent, } from '@moonshot-ai/kimi-code-sdk'; import chalk from 'chalk'; @@ -802,6 +803,10 @@ export class KimiTUI { this.requireSession(), ); } + const resumeState = this.session?.getResumeState(); + if (resumeState?.warning !== undefined) { + this.showStatus(`Warning: ${resumeState.warning}`, this.state.theme.colors.warning); + } if (this.session !== undefined) { this.startSessionEventSubscription(); } @@ -2102,6 +2107,10 @@ export class KimiTUI { } finally { this.startSessionEventSubscription(); } + const resumeState = session.getResumeState(); + if (resumeState?.warning !== undefined) { + this.showStatus(`Warning: ${resumeState.warning}`, this.state.theme.colors.warning); + } this.showStatus(statusMessage); } @@ -2260,6 +2269,9 @@ export class KimiTUI { case 'error': this.handleSessionError(event); break; + case 'warning': + this.handleSessionWarning(event); + break; case 'compaction.started': this.handleCompactionBegin(event); break; @@ -2680,6 +2692,10 @@ export class KimiTUI { } } + private handleSessionWarning(event: WarningEvent): void { + this.showStatus(`Warning: ${event.message}`, this.state.theme.colors.warning); + } + private renderMcpServerStatus(server: McpServerStatusSnapshot): void { const key = mcpServerStatusKey(server); if (this.state.renderedMcpServerStatusKeys.get(server.name) === key) return; diff --git a/packages/agent-core/src/agent/index.ts b/packages/agent-core/src/agent/index.ts index 5423ea560..6733fea36 100644 --- a/packages/agent-core/src/agent/index.ts +++ b/packages/agent-core/src/agent/index.ts @@ -234,11 +234,12 @@ export class Agent { this.tools.setActiveTools(profile.tools); } - async resume(): Promise { - await this.records.replay(); + async resume(): Promise<{ warning?: string }> { + const result = await this.records.replay(); await this.background.loadFromDisk(); await this.background.reconcile(); this.turn.finishResume(); + return result; } get rpcMethods(): PromisableMethods { diff --git a/packages/agent-core/src/agent/records/index.ts b/packages/agent-core/src/agent/records/index.ts index 6b133f7f7..9d2a263a7 100644 --- a/packages/agent-core/src/agent/records/index.ts +++ b/packages/agent-core/src/agent/records/index.ts @@ -1,6 +1,7 @@ import type { Agent } from '..'; import { AGENT_WIRE_PROTOCOL_VERSION, + isNewerWireVersion, migrateWireRecord, resolveWireMigrations, type WireMigration, @@ -140,11 +141,12 @@ export class AgentRecords { } } - async replay(): Promise { + async replay(): Promise<{ warning?: string }> { if (!this.persistence) throw new Error('No persistence provided for AgentRecords'); let migrations: readonly WireMigration[] = []; let hasMetadata = false; let shouldRewrite = false; + let warning: string | undefined; const replayedRecords: AgentRecord[] = []; for await (const record of this.persistence.read()) { if (!hasMetadata) { @@ -154,8 +156,13 @@ export class AgentRecords { hasMetadata = true; this.metadataInitialized = true; const readVersion = record.protocol_version; - migrations = resolveWireMigrations(readVersion); - shouldRewrite = readVersion !== AGENT_WIRE_PROTOCOL_VERSION; + if (isNewerWireVersion(readVersion)) { + warning = `Session wire protocol version ${readVersion} is newer than the current version ${AGENT_WIRE_PROTOCOL_VERSION}. Records will be replayed without migration.`; + shouldRewrite = false; + } else { + migrations = resolveWireMigrations(readVersion); + shouldRewrite = readVersion !== AGENT_WIRE_PROTOCOL_VERSION; + } } let migratedRecord = migrateWireRecord( record as WireMigrationRecord, @@ -174,6 +181,7 @@ export class AgentRecords { this.persistence.rewrite(replayedRecords); await this.persistence.flush(); } + return { warning }; } async flush(): Promise { diff --git a/packages/agent-core/src/agent/records/migration/index.ts b/packages/agent-core/src/agent/records/migration/index.ts index 06422999c..0c2691104 100644 --- a/packages/agent-core/src/agent/records/migration/index.ts +++ b/packages/agent-core/src/agent/records/migration/index.ts @@ -16,15 +16,14 @@ export interface WireMigration { const MIGRATIONS: readonly WireMigration[] = [migrateV1_0ToV1_1]; +export function isNewerWireVersion(readVersion: string): boolean { + return compareWireVersions(readVersion, AGENT_WIRE_PROTOCOL_VERSION) > 0; +} + export function resolveWireMigrations(readVersion: string): readonly WireMigration[] { - if (compareWireVersions(readVersion, AGENT_WIRE_PROTOCOL_VERSION) === 0) { + if (compareWireVersions(readVersion, AGENT_WIRE_PROTOCOL_VERSION) >= 0) { return []; } - if (compareWireVersions(readVersion, AGENT_WIRE_PROTOCOL_VERSION) > 0) { - throw new Error( - `Unsupported wire protocol version: ${readVersion} (current: ${AGENT_WIRE_PROTOCOL_VERSION})`, - ); - } const migrations: WireMigration[] = []; let version = readVersion; diff --git a/packages/agent-core/src/rpc/core-impl.ts b/packages/agent-core/src/rpc/core-impl.ts index 189ae276d..c25a1c759 100644 --- a/packages/agent-core/src/rpc/core-impl.ts +++ b/packages/agent-core/src/rpc/core-impl.ts @@ -252,8 +252,10 @@ export class KimiCore implements PromisableMethods { telemetry: withTelemetryContext(this.telemetry, { sessionId: summary.id }), initializeMainAgent: false, }); + let warning: string | undefined; try { - await session.resume(); + const resumeResult = await session.resume(); + warning = resumeResult.warning; await this.refreshSessionRuntimeConfig(session, config); } catch (error) { await session.close().catch(() => {}); @@ -263,7 +265,7 @@ export class KimiCore implements PromisableMethods { throw error; } this.sessions.set(summary.id, session); - return resumeSessionResult(summary, session); + return resumeSessionResult(summary, session, warning); } async forkSession(input: ForkSessionPayload): Promise { @@ -693,6 +695,7 @@ function telemetryErrorReason(error: unknown): string { async function resumeSessionResult( summary: SessionSummary, session: Session, + warning?: string, ): Promise { const api = new SessionAPIImpl(session); const agents: Record = {}; @@ -719,6 +722,7 @@ async function resumeSessionResult( ...summary, sessionMetadata: api.getSessionMetadata({}), agents, + warning, }; } diff --git a/packages/agent-core/src/rpc/events.ts b/packages/agent-core/src/rpc/events.ts index d387c49ae..40466a80a 100644 --- a/packages/agent-core/src/rpc/events.ts +++ b/packages/agent-core/src/rpc/events.ts @@ -69,8 +69,12 @@ export interface SkillActivatedEvent { export interface ErrorEvent extends KimiErrorPayload { readonly type: 'error'; - readonly sessionId?: string | undefined; - readonly agentId?: string | undefined; +} + +export interface WarningEvent { + readonly type: 'warning'; + readonly message: string; + readonly code?: string | undefined; } export interface TurnStartedEvent { @@ -265,6 +269,7 @@ export interface McpServerStatusPayload { export type AgentEvent = | ErrorEvent + | WarningEvent | AgentStatusUpdatedEvent | SessionMetaUpdatedEvent | SkillActivatedEvent diff --git a/packages/agent-core/src/rpc/resumed.ts b/packages/agent-core/src/rpc/resumed.ts index 61eb1dd67..211f2571c 100644 --- a/packages/agent-core/src/rpc/resumed.ts +++ b/packages/agent-core/src/rpc/resumed.ts @@ -36,4 +36,5 @@ export interface ResumedAgentState { export interface ResumeSessionResult extends SessionSummary { readonly sessionMetadata: SessionMeta; readonly agents: Readonly>; + readonly warning?: string | undefined; } diff --git a/packages/agent-core/src/session/index.ts b/packages/agent-core/src/session/index.ts index b4c0adba5..3fb9f778c 100644 --- a/packages/agent-core/src/session/index.ts +++ b/packages/agent-core/src/session/index.ts @@ -148,15 +148,20 @@ export class Session { return agent; } - async resume() { + async resume(): Promise<{ warning?: string }> { await this.skillsReady; const { agents } = await this.readMetadata(); this.agents.clear(); + let warning: string | undefined; const resumeTasks = Object.keys(agents).map(async (id) => { const agent = this.ensureResumeAgentInstantiated(id, agents); - await agent.resume(); + const result = await agent.resume(); + if (result.warning !== undefined && warning === undefined) { + warning = result.warning; + } }); await Promise.all(resumeTasks); + const resumeWarning = warning; // A session migrated from an external tool ships a wire without the // `config.update` bootstrap events a natively-created agent writes, so the // main agent comes back with an empty system prompt and no tools. Apply the @@ -168,6 +173,7 @@ export class Session { await this.bootstrapAgentProfile(main, profile); } await this.triggerSessionStart('resume'); + return { warning: resumeWarning }; } async close(): Promise { diff --git a/packages/agent-core/test/agent/records/index.test.ts b/packages/agent-core/test/agent/records/index.test.ts index 13513c37a..735b3fc66 100644 --- a/packages/agent-core/test/agent/records/index.test.ts +++ b/packages/agent-core/test/agent/records/index.test.ts @@ -157,7 +157,7 @@ describe('AgentRecords persistence metadata', () => { expect(migrated.message.toolCalls[0]?.['function']).toBeUndefined(); }); - it('rejects replaying records from a newer wire version', async () => { + it('warns but continues when replaying records from a newer wire version', async () => { const persistence = new InMemoryAgentRecordPersistence([ { type: 'metadata', @@ -167,9 +167,9 @@ describe('AgentRecords persistence metadata', () => { ]); const records = new AgentRecords(() => {}, persistence); - await expect(records.replay()).rejects.toThrow( - `Unsupported wire protocol version: 9.9 (current: ${AGENT_WIRE_PROTOCOL_VERSION})`, - ); + const result = await records.replay(); + expect(result.warning).toContain('9.9'); + expect(result.warning).toContain(AGENT_WIRE_PROTOCOL_VERSION); }); it('rejects replaying records without a registered migration path', async () => { diff --git a/packages/node-sdk/src/events.ts b/packages/node-sdk/src/events.ts index 014d81bb3..a20ec597b 100644 --- a/packages/node-sdk/src/events.ts +++ b/packages/node-sdk/src/events.ts @@ -16,6 +16,7 @@ export type { SessionMetaUpdatedEvent, SkillActivatedEvent, ErrorEvent, + WarningEvent, UsageStatus, } from '@moonshot-ai/agent-core'; diff --git a/packages/node-sdk/src/session.ts b/packages/node-sdk/src/session.ts index 766cae584..7728dfe2e 100644 --- a/packages/node-sdk/src/session.ts +++ b/packages/node-sdk/src/session.ts @@ -395,6 +395,7 @@ function resumeStateFromSummary( return { sessionMetadata: summary.sessionMetadata, agents: summary.agents, + warning: summary.warning, }; } diff --git a/packages/node-sdk/src/types.ts b/packages/node-sdk/src/types.ts index c6571786e..7e4944f6f 100644 --- a/packages/node-sdk/src/types.ts +++ b/packages/node-sdk/src/types.ts @@ -159,6 +159,6 @@ export interface SessionSummary { readonly metadata?: JsonObject | undefined; } -export type ResumedSessionState = Pick; +export type ResumedSessionState = Pick; export interface ResumedSessionSummary extends SessionSummary, ResumedSessionState {} diff --git a/packages/node-sdk/test/session-event-types.test.ts b/packages/node-sdk/test/session-event-types.test.ts index 24265cb7b..55710d9db 100644 --- a/packages/node-sdk/test/session-event-types.test.ts +++ b/packages/node-sdk/test/session-event-types.test.ts @@ -43,6 +43,7 @@ describe('Event public types', () => { case 'session.meta.updated': case 'skill.activated': case 'error': + case 'warning': case 'turn.started': case 'turn.ended': case 'turn.step.started':