From 989ef3a5f7af44bd1cf3011225fd5b16735f9e3a Mon Sep 17 00:00:00 2001 From: maiconfontana Date: Sun, 2 Aug 2026 16:04:28 -0300 Subject: [PATCH 1/3] feat(server): add Kimi Code CLI provider over ACP Bring Kimi into the built-in provider set with full ACP session support, prefixed model catalog (kimi-code/*), thinking effort options, thought-chunk worklog projection, and assistant streaming. Wire settings, web picker/icon, docs, and focused tests. --- AGENTS.md | 4 +- README.md | 5 +- .../Layers/ProviderRuntimeIngestion.ts | 173 ++ .../server/src/provider/Drivers/KimiDriver.ts | 168 ++ .../src/provider/Layers/CursorAdapter.ts | 1 + .../server/src/provider/Layers/GrokAdapter.ts | 1 + .../src/provider/Layers/KimiAdapter.test.ts | 182 +++ .../server/src/provider/Layers/KimiAdapter.ts | 1406 +++++++++++++++++ .../src/provider/Layers/KimiProvider.test.ts | 167 ++ .../src/provider/Layers/KimiProvider.ts | 474 ++++++ .../src/provider/Services/KimiAdapter.ts | 16 + .../src/provider/acp/AcpCoreRuntimeEvents.ts | 3 +- .../src/provider/acp/AcpRuntimeModel.test.ts | 29 + .../src/provider/acp/AcpRuntimeModel.ts | 14 + .../src/provider/acp/AcpSessionRuntime.ts | 8 + .../src/provider/acp/KimiAcpSupport.test.ts | 125 ++ .../server/src/provider/acp/KimiAcpSupport.ts | 208 +++ apps/server/src/provider/builtInDrivers.ts | 3 + .../textGeneration/KimiTextGeneration.test.ts | 222 +++ .../src/textGeneration/KimiTextGeneration.ts | 260 +++ .../src/textGeneration/TextGeneration.ts | 8 +- apps/web/src/components/Icons.tsx | 12 + .../src/components/chat/providerIconUtils.ts | 3 +- .../components/settings/providerDriverMeta.ts | 18 +- apps/web/src/session-logic.ts | 6 + docs/internals/glossary.md | 2 +- docs/internals/overview.md | 6 +- docs/internals/providers.md | 4 +- docs/user/install.md | 15 +- packages/contracts/src/model.ts | 11 + packages/contracts/src/settings.ts | 32 + 31 files changed, 3566 insertions(+), 20 deletions(-) create mode 100644 apps/server/src/provider/Drivers/KimiDriver.ts create mode 100644 apps/server/src/provider/Layers/KimiAdapter.test.ts create mode 100644 apps/server/src/provider/Layers/KimiAdapter.ts create mode 100644 apps/server/src/provider/Layers/KimiProvider.test.ts create mode 100644 apps/server/src/provider/Layers/KimiProvider.ts create mode 100644 apps/server/src/provider/Services/KimiAdapter.ts create mode 100644 apps/server/src/provider/acp/KimiAcpSupport.test.ts create mode 100644 apps/server/src/provider/acp/KimiAcpSupport.ts create mode 100644 apps/server/src/textGeneration/KimiTextGeneration.test.ts create mode 100644 apps/server/src/textGeneration/KimiTextGeneration.ts diff --git a/AGENTS.md b/AGENTS.md index c3a7fe92bf4..7293998042a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,6 +1,6 @@ # T3 Code -T3 Code is a minimal GUI for coding agents. A Node WebSocket server wraps provider CLIs (Codex, Claude Code, Cursor, Grok, OpenCode) and serves web, desktop, and mobile clients. +T3 Code is a minimal GUI for coding agents. A Node WebSocket server wraps provider CLIs (Codex, Claude Code, Cursor, Grok, Kimi, OpenCode) and serves web, desktop, and mobile clients. You can think of T3 Code as an open source "bring-your-own-subscription" alternative to apps like Claude Desktop, Codex App, Cursor Glass and Conductor. @@ -68,7 +68,7 @@ The most common defect in this repo is a change that works on the path you teste - **Entry points.** A behavior reachable from the chat view is usually also reachable from Settings, the command palette, and a keybinding. Fixing one is not fixing the feature. - **Clients.** Web, desktop (wraps web, adds Electron shell/IPC), and mobile (React Native, separate navigation). Shared logic lives in `packages/client-runtime` -- **Providers.** Codex, Claude, Cursor, Grok, and OpenCode each have an adapter. Provider-shaped features need a decision per adapter, even if the decision is "not supported here". +- **Providers.** Codex, Claude, Cursor, Grok, Kimi, and OpenCode each have an adapter. Provider-shaped features need a decision per adapter, even if the decision is "not supported here". - **Contracts.** Anything crossing the wire is typed in `packages/contracts`. Change the schema and the server, web, mobile, and desktop all follow. - **Reverse states.** If you added a way in, add the way out and the way to see it. Snooze needs unsnooze. Close needs reopen. A one-way door is a bug. - **Connection modes.** Local, remote/relay, and tunnel behave differently. Multi-device and multi-environment cases are real. diff --git a/README.md b/README.md index 1e9b0517945..008036e9964 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ T3 Code is an "agent harness control surface". It enables control of the agents on your machine with a best-in-class mobile app ([iOS](https://apps.apple.com/us/app/t3-code-remote-claude-more/id6787819824), [Android](https://play.google.com/store/apps/details?id=com.t3tools.t3code)), [web app](https://app.t3.codes) and [Electron-based desktop app](https://t3.codes). -Works with your subscriptions on Claude Code, Codex, Cursor, Grok Build, and OpenCode. If they're set up on your computer, T3 Code can control them. +Works with your subscriptions on Claude Code, Codex, Cursor, Grok Build, Kimi Code CLI, and OpenCode. If they're set up on your computer, T3 Code can control them. ## "Wait, what are you selling me?" @@ -13,12 +13,13 @@ We wanted something performant, remote-ready, and truly open. If we ever go the ## Installation > [!WARNING] -> T3 Code currently supports Codex, Claude, Cursor, Grok Build and OpenCode. Install and authenticate at least one provider before use: +> T3 Code currently supports Codex, Claude, Cursor, Grok Build, Kimi Code CLI, and OpenCode. Install and authenticate at least one provider before use: > > - Codex: install [Codex CLI](https://developers.openai.com/codex/cli) and run `codex login` > - Claude: install [Claude Code](https://claude.com/product/claude-code) and run `claude auth login` > - Cursor: install [Cursor CLI](https://cursor.com/cli) and run `agent login` > - Grok Build: install [Grok Build CLI](https://x.ai/cli) and run `grok login` +> - Kimi: install [Kimi Code CLI](https://moonshotai.github.io/kimi-code/en/) and run `kimi` then `/login` > - OpenCode: install [OpenCode](https://opencode.ai) and run `opencode auth login` ### Try it out (install-free) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index c8d619270d3..27dbe5a8224 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2,6 +2,7 @@ import { ApprovalRequestId, type AssistantDeliveryMode, CommandId, + EventId, MessageId, type OrchestrationEvent, type OrchestrationMessage, @@ -726,6 +727,17 @@ const make = Effect.gen(function* () { lookup: () => Effect.succeed({ text: "", createdAt: "" }), }); + // Streaming reasoning/thought chunks (ACP agent_thought_chunk, Claude thinking, …) + // are buffered per turn and projected as thinking worklog activity. + const bufferedReasoningByTurnKey = yield* Cache.make< + string, + { text: string; started: boolean; lastPublishedLength: number } + >({ + capacity: TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY, + timeToLive: TURN_MESSAGE_IDS_BY_TURN_TTL, + lookup: () => Effect.succeed({ text: "", started: false, lastPublishedLength: 0 }), + }); + // Task names arrive on task.started/task.progress but not on task.completed, // so remember them per task to title the completion activity. const taskDescriptionByTaskKey = yield* Cache.make({ @@ -737,6 +749,135 @@ const make = Effect.gen(function* () { const rememberTaskDescription = (threadId: ThreadId, taskId: string, description: string) => Cache.set(taskDescriptionByTaskKey, providerTaskKey(threadId, taskId), description); + const reasoningTaskIdForTurn = (turnId: TurnId) => `reasoning:${turnId}`; + const REASONING_PROGRESS_PUBLISH_CHARS = 280; + + const appendReasoningDelta = (input: { + readonly event: ProviderRuntimeEvent; + readonly threadId: ThreadId; + readonly turnId: TurnId; + readonly delta: string; + readonly createdAt: string; + }) => + Effect.gen(function* () { + const key = providerTurnKey(input.threadId, input.turnId); + const existing = yield* Cache.getOption(bufferedReasoningByTurnKey, key).pipe( + Effect.map((option) => + Option.getOrElse(option, () => ({ text: "", started: false, lastPublishedLength: 0 })), + ), + ); + const nextText = `${existing.text}${input.delta}`; + const taskId = reasoningTaskIdForTurn(input.turnId); + const activities: Array = []; + + if (!existing.started) { + activities.push({ + id: EventId.make(`${input.event.eventId}:reasoning-started`), + createdAt: input.createdAt, + tone: "info", + kind: "task.started", + summary: "Thinking", + payload: { + taskId, + taskType: "reasoning", + detail: "Thinking…", + }, + turnId: input.turnId, + }); + } + + const shouldPublishProgress = + !existing.started || + nextText.length - existing.lastPublishedLength >= REASONING_PROGRESS_PUBLISH_CHARS; + if (shouldPublishProgress && nextText.trim().length > 0) { + activities.push({ + id: EventId.make(`${input.event.eventId}:reasoning-progress`), + createdAt: input.createdAt, + tone: "info", + kind: "task.progress", + summary: "Thinking", + payload: { + taskId, + title: "Thinking", + detail: truncateDetail(nextText), + summary: truncateDetail(nextText, 160), + }, + turnId: input.turnId, + }); + } + + yield* Cache.set(bufferedReasoningByTurnKey, key, { + text: nextText, + started: true, + lastPublishedLength: shouldPublishProgress ? nextText.length : existing.lastPublishedLength, + }); + yield* rememberTaskDescription(input.threadId, taskId, "Thinking"); + + yield* Effect.forEach( + activities, + (activity) => + providerCommandId(input.event, "reasoning-activity").pipe( + Effect.flatMap((commandId) => + orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId, + threadId: input.threadId, + activity, + createdAt: activity.createdAt, + }), + ), + ), + { concurrency: 1 }, + ); + }); + + const completeReasoningForTurn = (input: { + readonly event: ProviderRuntimeEvent; + readonly threadId: ThreadId; + readonly turnId: TurnId; + readonly createdAt: string; + }) => + Effect.gen(function* () { + const key = providerTurnKey(input.threadId, input.turnId); + const existing = yield* Cache.getOption(bufferedReasoningByTurnKey, key).pipe( + Effect.map(Option.getOrUndefined), + ); + if (!existing?.started) { + return; + } + const taskId = reasoningTaskIdForTurn(input.turnId); + const detail = truncateDetail(existing.text.trim()); + yield* Cache.invalidate(bufferedReasoningByTurnKey, key); + if (detail.length === 0) { + return; + } + yield* providerCommandId(input.event, "reasoning-complete").pipe( + Effect.flatMap((commandId) => + orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId, + threadId: input.threadId, + activity: { + id: EventId.make(`${input.event.eventId}:reasoning-completed`), + createdAt: input.createdAt, + tone: "info", + kind: "task.completed", + summary: "Thinking complete", + payload: { + taskId, + status: "completed", + title: "Thinking", + summary: detail, + detail, + }, + turnId: input.turnId, + }, + createdAt: input.createdAt, + }), + ), + ); + }); + // Entries are left in place after completion so replayed or duplicate // terminal events stay titled; TTL, capacity, and the session-exit sweep // bound the cache. @@ -1458,11 +1599,37 @@ const make = Effect.gen(function* () { event.type === "content.delta" && event.payload.streamKind === "assistant_text" ? event.payload.delta : undefined; + const reasoningDelta = + event.type === "content.delta" && event.payload.streamKind === "reasoning_text" + ? event.payload.delta + : undefined; const proposedPlanDelta = event.type === "turn.proposed.delta" ? event.payload.delta : undefined; + if (reasoningDelta && reasoningDelta.length > 0) { + const turnId = toTurnId(event.turnId); + if (turnId) { + yield* appendReasoningDelta({ + event, + threadId: thread.id, + turnId, + delta: reasoningDelta, + createdAt: now, + }); + } + } + if (assistantDelta && assistantDelta.length > 0) { const turnId = toTurnId(event.turnId); + // First visible answer token closes the thinking worklog for this turn. + if (turnId) { + yield* completeReasoningForTurn({ + event, + threadId: thread.id, + turnId, + createdAt: now, + }); + } const assistantMessageId = yield* getOrCreateAssistantMessageId({ threadId: thread.id, event, @@ -1641,6 +1808,12 @@ const make = Effect.gen(function* () { const proposedPlans = detailedThread?.proposedPlans ?? []; const turnId = toTurnId(event.turnId); if (turnId) { + yield* completeReasoningForTurn({ + event, + threadId: thread.id, + turnId, + createdAt: now, + }); const assistantMessageIds = yield* getAssistantMessageIdsForTurn(thread.id, turnId); yield* Effect.forEach( assistantMessageIds, diff --git a/apps/server/src/provider/Drivers/KimiDriver.ts b/apps/server/src/provider/Drivers/KimiDriver.ts new file mode 100644 index 00000000000..bb9c36967c8 --- /dev/null +++ b/apps/server/src/provider/Drivers/KimiDriver.ts @@ -0,0 +1,168 @@ +import { KimiSettings, ProviderDriverKind, type ServerProvider } from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import { HttpClient } from "effect/unstable/http"; +import { ChildProcessSpawner } from "effect/unstable/process"; + +import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts"; +import { ServerConfig } from "../../config.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import { makeKimiTextGeneration } from "../../textGeneration/KimiTextGeneration.ts"; +import { ProviderDriverError } from "../Errors.ts"; +import { makeKimiAdapter } from "../Layers/KimiAdapter.ts"; +import { + buildInitialKimiProviderSnapshot, + checkKimiProviderStatus, + enrichKimiSnapshot, +} from "../Layers/KimiProvider.ts"; +import { ProviderEventLoggers } from "../Layers/ProviderEventLoggers.ts"; +import { makeManagedServerProvider } from "../makeManagedServerProvider.ts"; +import { + defaultProviderContinuationIdentity, + type ProviderDriver, + type ProviderInstance, +} from "../ProviderDriver.ts"; +import type { ServerProviderDraft } from "../providerSnapshot.ts"; +import { mergeProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; +import { + makeProviderMaintenanceCapabilities, + type ProviderMaintenanceCapabilitiesResolver, + resolveProviderMaintenanceCapabilitiesEffect, +} from "../providerMaintenance.ts"; +import { + haveProviderSnapshotSettingsChanged, + makeProviderSnapshotSettingsSource, + type ProviderSnapshotSettings, +} from "../providerUpdateSettings.ts"; + +const decodeKimiSettings = Schema.decodeSync(KimiSettings); + +const DRIVER_KIND = ProviderDriverKind.make("kimi"); +const UPDATE: ProviderMaintenanceCapabilitiesResolver = { + resolve: (options) => + makeProviderMaintenanceCapabilities({ + provider: DRIVER_KIND, + packageName: "@moonshot-ai/kimi-code", + updateExecutable: options?.binaryPath?.trim() || "kimi", + updateArgs: ["upgrade"], + updateLockKey: "kimi-code", + }), +}; + +export type KimiDriverEnv = + | BackgroundPolicy.BackgroundPolicy + | ChildProcessSpawner.ChildProcessSpawner + | Crypto.Crypto + | FileSystem.FileSystem + | HttpClient.HttpClient + | Path.Path + | ProviderEventLoggers + | ServerConfig + | ServerSettingsService; + +const withInstanceIdentity = + (input: { + readonly instanceId: ProviderInstance["instanceId"]; + readonly displayName: string | undefined; + readonly accentColor: string | undefined; + readonly continuationGroupKey: string; + }) => + (snapshot: ServerProviderDraft): ServerProvider => ({ + ...snapshot, + instanceId: input.instanceId, + driver: DRIVER_KIND, + ...(input.displayName ? { displayName: input.displayName } : {}), + ...(input.accentColor ? { accentColor: input.accentColor } : {}), + continuation: { groupKey: input.continuationGroupKey }, + }); + +export const KimiDriver: ProviderDriver = { + driverKind: DRIVER_KIND, + metadata: { + displayName: "Kimi", + supportsMultipleInstances: true, + }, + configSchema: KimiSettings, + defaultConfig: (): KimiSettings => decodeKimiSettings({}), + create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => + Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const httpClient = yield* HttpClient.HttpClient; + const serverSettings = yield* ServerSettingsService; + const eventLoggers = yield* ProviderEventLoggers; + const processEnv = mergeProviderInstanceEnvironment(environment); + const continuationIdentity = defaultProviderContinuationIdentity({ + driverKind: DRIVER_KIND, + instanceId, + }); + const stampIdentity = withInstanceIdentity({ + instanceId, + displayName, + accentColor, + continuationGroupKey: continuationIdentity.continuationKey, + }); + const effectiveConfig = { ...config, enabled } satisfies KimiSettings; + const maintenanceCapabilities = yield* resolveProviderMaintenanceCapabilitiesEffect(UPDATE, { + binaryPath: effectiveConfig.binaryPath, + env: processEnv, + }); + + const adapter = yield* makeKimiAdapter(effectiveConfig, { + environment: processEnv, + ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}), + instanceId, + }); + const textGeneration = yield* makeKimiTextGeneration(effectiveConfig, processEnv); + + const checkProvider = checkKimiProviderStatus(effectiveConfig, processEnv).pipe( + Effect.map(stampIdentity), + Effect.provideService(Crypto.Crypto, crypto), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ); + + const snapshotSettings = makeProviderSnapshotSettingsSource(effectiveConfig, serverSettings); + const snapshot = yield* makeManagedServerProvider>({ + maintenanceCapabilities, + getSettings: snapshotSettings.getSettings, + streamSettings: snapshotSettings.streamSettings, + haveSettingsChanged: haveProviderSnapshotSettingsChanged, + initialSnapshot: (settings) => + buildInitialKimiProviderSnapshot(settings.provider).pipe(Effect.map(stampIdentity)), + checkProvider, + enrichSnapshot: ({ settings, snapshot: currentSnapshot, publishSnapshot }) => + enrichKimiSnapshot({ + snapshot: currentSnapshot, + maintenanceCapabilities, + enableProviderUpdateChecks: settings.enableProviderUpdateChecks, + publishSnapshot, + httpClient, + }), + }).pipe( + Effect.mapError( + (cause) => + new ProviderDriverError({ + driver: DRIVER_KIND, + instanceId, + detail: `Failed to build Kimi snapshot: ${cause.message ?? String(cause)}`, + cause, + }), + ), + ); + + return { + instanceId, + driverKind: DRIVER_KIND, + continuationIdentity, + displayName, + accentColor, + enabled, + snapshot, + adapter, + textGeneration, + } satisfies ProviderInstance; + }), +}; diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index 4dd38519c5a..f6138400d9a 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -863,6 +863,7 @@ export function makeCursorAdapter( turnId: ctx.activeTurnId, ...(event.itemId ? { itemId: event.itemId } : {}), text: event.text, + streamKind: event.streamKind, rawPayload: event.rawPayload, }), ); diff --git a/apps/server/src/provider/Layers/GrokAdapter.ts b/apps/server/src/provider/Layers/GrokAdapter.ts index d8c288a8292..74f79e76454 100644 --- a/apps/server/src/provider/Layers/GrokAdapter.ts +++ b/apps/server/src/provider/Layers/GrokAdapter.ts @@ -865,6 +865,7 @@ export function makeGrokAdapter(grokSettings: GrokSettings, options?: GrokAdapte turnId: notificationTurnId, ...(event.itemId ? { itemId: event.itemId } : {}), text: event.text, + streamKind: event.streamKind, rawPayload: event.rawPayload, }), ); diff --git a/apps/server/src/provider/Layers/KimiAdapter.test.ts b/apps/server/src/provider/Layers/KimiAdapter.test.ts new file mode 100644 index 00000000000..f1500bdb8b4 --- /dev/null +++ b/apps/server/src/provider/Layers/KimiAdapter.test.ts @@ -0,0 +1,182 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodePath from "node:path"; +import * as NodeOS from "node:os"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeURL from "node:url"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import { + KimiSettings, + ProviderDriverKind, + ThreadId, + TurnId, + type ProviderRuntimeEvent, +} from "@t3tools/contracts"; + +import { ServerConfig } from "../../config.ts"; +import { kimiPromptSettlementBelongsToContext, makeKimiAdapter } from "./KimiAdapter.ts"; + +const decodeKimiSettings = Schema.decodeSync(KimiSettings); +const isWin = process.platform === "win32"; + +const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); +const mockAgentPath = NodePath.join(__dirname, "../../../scripts/acp-mock-agent.ts"); + +async function makeMockKimiWrapper(extraEnv?: Record) { + const dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "kimi-acp-mock-")); + const jsPath = NodePath.join(dir, "kimi.mjs"); + const envJson = JSON.stringify(extraEnv ?? {}); + await NodeFSP.writeFile( + jsPath, + [ + "import { spawnSync } from 'node:child_process';", + `const env = ${envJson};`, + "const args = process.argv.slice(2);", + 'if (args[0] !== "acp") {', + " process.stderr.write(`unexpected args: ${args.join(' ')}\\n`);", + " process.exit(11);", + "}", + `const result = spawnSync(${JSON.stringify(process.execPath)}, [${JSON.stringify(mockAgentPath)}], {`, + " stdio: 'inherit',", + " env: { ...process.env, ...env },", + "});", + "process.exit(result.status ?? 1);", + "", + ].join("\n"), + "utf8", + ); + + if (isWin) { + const cmdPath = NodePath.join(dir, "kimi.cmd"); + await NodeFSP.writeFile( + cmdPath, + ["@echo off", `node "${jsPath.replaceAll("/", "\\")}" %*`, ""].join("\r\n"), + "utf8", + ); + return cmdPath; + } + + const shPath = NodePath.join(dir, "fake-kimi.sh"); + await NodeFSP.writeFile( + shPath, + [ + "#!/bin/sh", + `exec ${JSON.stringify(process.execPath)} ${JSON.stringify(jsPath)} "$@"`, + "", + ].join("\n"), + "utf8", + ); + await NodeFSP.chmod(shPath, 0o755); + return shPath; +} + +const kimiAdapterTestLayer = ServerConfig.layerTest(process.cwd(), { + prefix: "t3code-kimi-adapter-test-", +}).pipe(Layer.provideMerge(NodeServices.layer)); + +const makeTestAdapter = (binaryPath: string, options?: Parameters[1]) => + makeKimiAdapter(decodeKimiSettings({ binaryPath }), options).pipe(Effect.orDie); + +it("requires a settlement to match the live Kimi turn", () => { + const staleTurnId = TurnId.make("stale-turn"); + const replacementTurnId = TurnId.make("replacement-turn"); + + assert.isFalse( + kimiPromptSettlementBelongsToContext({ + liveAcpSessionId: "session-1", + expectedAcpSessionId: "session-1", + liveActiveTurnId: replacementTurnId, + liveSessionActiveTurnId: replacementTurnId, + turnId: staleTurnId, + }), + ); + assert.isTrue( + kimiPromptSettlementBelongsToContext({ + liveAcpSessionId: "session-1", + expectedAcpSessionId: "session-1", + liveActiveTurnId: staleTurnId, + liveSessionActiveTurnId: staleTurnId, + turnId: staleTurnId, + }), + ); +}); + +it.layer(kimiAdapterTestLayer)("KimiAdapterLive", (it) => { + it.effect("starts a session and maps mock ACP prompt flow to runtime events", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-mock-thread"); + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const turnCompleted = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }).pipe( + Effect.andThen( + event.type === "turn.completed" + ? Deferred.succeed(turnCompleted, undefined) + : Effect.void, + ), + ), + ).pipe(Effect.forkChild); + + const session = yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + }); + + assert.equal(session.provider, ProviderDriverKind.make("kimi")); + assert.equal(session.status, "ready"); + assert.isTrue(yield* adapter.hasSession(threadId)); + + const turn = yield* adapter.sendTurn({ + threadId, + input: "hello from kimi adapter test", + }); + assert.equal(turn.threadId, threadId); + + yield* Deferred.await(turnCompleted); + yield* Fiber.interrupt(runtimeEventsFiber); + + const eventTypes = runtimeEvents.map((event) => event.type); + assert.includeMembers(eventTypes, [ + "session.started", + "session.state.changed", + "thread.started", + "turn.started", + "turn.completed", + ]); + + yield* adapter.stopSession(threadId); + assert.isFalse(yield* adapter.hasSession(threadId)); + }), + ); + + it.effect("rejects startSession without a cwd", () => + Effect.gen(function* () { + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + const error = yield* Effect.flip( + adapter.startSession({ + threadId: ThreadId.make("kimi-no-cwd"), + provider: ProviderDriverKind.make("kimi"), + cwd: " ", + runtimeMode: "full-access", + }), + ); + assert.equal(error._tag, "ProviderAdapterValidationError"); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/KimiAdapter.ts b/apps/server/src/provider/Layers/KimiAdapter.ts new file mode 100644 index 00000000000..b562719497a --- /dev/null +++ b/apps/server/src/provider/Layers/KimiAdapter.ts @@ -0,0 +1,1406 @@ +import { + ApprovalRequestId, + type KimiSettings, + EventId, + type ProviderApprovalDecision, + type ProviderRuntimeEvent, + type ProviderSession, + type ProviderUserInputAnswers, + ProviderDriverKind, + ProviderInstanceId, + RuntimeRequestId, + type ThreadId, + TurnId, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; +import * as SynchronizedRef from "effect/SynchronizedRef"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; +import * as EffectAcpErrors from "effect-acp/errors"; +import type * as EffectAcpSchema from "effect-acp/schema"; + +import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import { + ProviderAdapterProcessError, + ProviderAdapterRequestError, + ProviderAdapterSessionNotFoundError, + ProviderAdapterValidationError, +} from "../Errors.ts"; +import { mapAcpToAdapterError } from "../acp/AcpAdapterSupport.ts"; +import type * as AcpSessionRuntime from "../acp/AcpSessionRuntime.ts"; +import { + makeAcpAssistantItemEvent, + makeAcpContentDeltaEvent, + makeAcpPlanUpdatedEvent, + makeAcpRequestOpenedEvent, + makeAcpRequestResolvedEvent, + makeAcpToolCallEvent, +} from "../acp/AcpCoreRuntimeEvents.ts"; +import { parsePermissionRequest } from "../acp/AcpRuntimeModel.ts"; +import { makeAcpNativeLoggerFactory } from "../acp/AcpNativeLogging.ts"; +import { + applyKimiAcpModelSelection, + currentKimiModelIdFromSessionSetup, + makeKimiAcpRuntime, + resolveKimiAcpBaseModelId, +} from "../acp/KimiAcpSupport.ts"; +import { type KimiAdapterShape } from "../Services/KimiAdapter.ts"; +import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; + +const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.UnknownFromJsonString); + +const PROVIDER = ProviderDriverKind.make("kimi"); +const KIMI_RESUME_VERSION = 1 as const; + +function encodeJsonStringForDiagnostics(input: unknown): string | undefined { + const result = encodeUnknownJsonStringExit(input); + return Exit.isSuccess(result) ? result.value : undefined; +} + +export interface KimiAdapterLiveOptions { + readonly environment?: NodeJS.ProcessEnv; + readonly nativeEventLogPath?: string; + readonly nativeEventLogger?: EventNdjsonLogger; + readonly instanceId?: ProviderInstanceId; +} + +interface PendingApproval { + readonly decision: Deferred.Deferred; +} + +type PendingUserInputResolution = + | { readonly _tag: "answered"; readonly answers: ProviderUserInputAnswers } + | { readonly _tag: "cancelled" }; + +interface PendingUserInput { + readonly resolution: Deferred.Deferred; +} + +interface KimiSessionContext { + readonly threadId: ThreadId; + readonly acpSessionId: string; + session: ProviderSession; + readonly scope: Scope.Closeable; + readonly acp: AcpSessionRuntime.AcpSessionRuntime["Service"]; + notificationFiber: Fiber.Fiber | undefined; + readonly pendingApprovals: Map; + readonly pendingUserInputs: Map; + turns: Array<{ id: TurnId; items: Array }>; + lastPlanFingerprint: string | undefined; + activeTurnId: TurnId | undefined; + /** Turns already interrupted; late prompt RPCs must not resurrect them. */ + interruptedTurnIds: Set; + /** Number of sendTurn prompts currently in flight or being prepared. + * >0 means a turn is actively running, so a new sendTurn is a steer that + * continues it, and only the last remaining prompt settles the turn. */ + promptsInFlight: number; + currentModelId: string | undefined; + stopped: boolean; +} + +function settlePendingApprovalsAsCancelled( + pendingApprovals: ReadonlyMap, +): Effect.Effect { + return Effect.forEach( + Array.from(pendingApprovals.values()), + (pending) => Deferred.succeed(pending.decision, "cancel").pipe(Effect.ignore), + { discard: true }, + ); +} + +function settlePendingUserInputsAsCancelled( + pendingUserInputs: ReadonlyMap, +): Effect.Effect { + return Effect.forEach( + Array.from(pendingUserInputs.values()), + (pending) => Deferred.succeed(pending.resolution, { _tag: "cancelled" }).pipe(Effect.ignore), + { discard: true }, + ); +} + +function appendPromptResultToTurn( + ctx: KimiSessionContext, + turnId: TurnId, + promptParts: ReadonlyArray, + result: EffectAcpSchema.PromptResponse, +): void { + const existingTurnRecord = ctx.turns.find((turn) => turn.id === turnId); + ctx.turns = existingTurnRecord + ? ctx.turns.map((turn) => + turn.id === turnId + ? { ...turn, items: [...turn.items, { prompt: promptParts, result }] } + : turn, + ) + : [...ctx.turns, { id: turnId, items: [{ prompt: promptParts, result }] }]; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +const resolveNotificationTurnId = (ctx: KimiSessionContext): TurnId | undefined => ctx.activeTurnId; + +const resolveCallbackTurnId = (ctx: KimiSessionContext): TurnId | undefined => ctx.activeTurnId; + +const resolveSessionCallbackTurnId = ( + sessions: ReadonlyMap, + threadId: ThreadId, +): TurnId | undefined => { + const ctx = sessions.get(threadId); + return ctx ? resolveCallbackTurnId(ctx) : undefined; +}; + +function parseKimiResume(raw: unknown): { sessionId: string } | undefined { + if (!isRecord(raw)) return undefined; + if (raw.schemaVersion !== KIMI_RESUME_VERSION) return undefined; + if (typeof raw.sessionId !== "string" || !raw.sessionId.trim()) return undefined; + return { sessionId: raw.sessionId.trim() }; +} + +function selectPermissionOptionId( + request: EffectAcpSchema.RequestPermissionRequest, + decision: Exclude, +): string | undefined { + const kind = + decision === "acceptForSession" + ? "allow_always" + : decision === "accept" + ? "allow_once" + : "reject_once"; + const option = request.options.find((entry) => entry.kind === kind); + return option?.optionId.trim() || undefined; +} + +function selectAutoApprovedPermissionOption( + request: EffectAcpSchema.RequestPermissionRequest, +): string | undefined { + return ( + selectPermissionOptionId(request, "acceptForSession") ?? + selectPermissionOptionId(request, "accept") + ); +} + +function completedStopReasonFromPromptResponse( + response: EffectAcpSchema.PromptResponse | undefined, +): EffectAcpSchema.StopReason | null { + if (response === undefined) { + return null; + } + return response.stopReason; +} + +export function kimiPromptSettlementBelongsToContext(input: { + readonly liveAcpSessionId: string; + readonly expectedAcpSessionId: string; + readonly liveActiveTurnId: TurnId | undefined; + readonly liveSessionActiveTurnId: TurnId | undefined; + readonly turnId: TurnId; +}): boolean { + return ( + input.liveAcpSessionId === input.expectedAcpSessionId && + (input.liveActiveTurnId === input.turnId || input.liveSessionActiveTurnId === input.turnId) + ); +} + +export function makeKimiAdapter(kimiSettings: KimiSettings, options?: KimiAdapterLiveOptions) { + return Effect.gen(function* () { + const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("kimi"); + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const serverConfig = yield* Effect.service(ServerConfig); + const crypto = yield* Crypto.Crypto; + const nativeEventLogger = + options?.nativeEventLogger ?? + (options?.nativeEventLogPath !== undefined + ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { stream: "native" }) + : undefined); + const managedNativeEventLogger = + options?.nativeEventLogger === undefined ? nativeEventLogger : undefined; + const makeAcpNativeLoggers = yield* makeAcpNativeLoggerFactory(); + + const sessions = new Map(); + const threadLocksRef = yield* SynchronizedRef.make(new Map()); + const runtimeEventPubSub = yield* PubSub.unbounded(); + + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + const randomUUIDv4 = crypto.randomUUIDv4.pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "crypto/randomUUIDv4", + detail: "Failed to generate Kimi runtime identifier.", + cause, + }), + ), + ); + const nextEventId = Effect.map(randomUUIDv4, (id) => EventId.make(id)); + const makeEventStamp = () => Effect.all({ eventId: nextEventId, createdAt: nowIso }); + const mapAcpCallbackFailure = (effect: Effect.Effect) => + effect.pipe( + Effect.mapError( + (cause) => + new EffectAcpErrors.AcpTransportError({ + detail: "Failed to process Kimi ACP callback.", + cause, + }), + ), + ); + + const offerRuntimeEvent = (event: ProviderRuntimeEvent) => + PubSub.publish(runtimeEventPubSub, event).pipe(Effect.asVoid); + + const getThreadSemaphore = (threadId: string) => + SynchronizedRef.modifyEffect(threadLocksRef, (current) => { + const existing: Option.Option = Option.fromNullishOr( + current.get(threadId), + ); + return Option.match(existing, { + onNone: () => + Semaphore.make(1).pipe( + Effect.map((semaphore) => { + const next = new Map(current); + next.set(threadId, semaphore); + return [semaphore, next] as const; + }), + ), + onSome: (semaphore) => Effect.succeed([semaphore, current] as const), + }); + }); + + const withThreadLock = (threadId: string, effect: Effect.Effect) => + Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect)); + + const settlePromptInFlight = ( + threadId: ThreadId, + turnId: TurnId, + expectedAcpSessionId: string, + options?: { + readonly errorMessage?: string; + readonly completedStopReason?: EffectAcpSchema.StopReason | null; + readonly emitTurnCompletion?: boolean; + /** Interrupt/cancel: drop every outstanding prompt slot and settle once. */ + readonly settleAllPrompts?: boolean; + }, + ) => + Effect.gen(function* () { + const liveCtx = sessions.get(threadId); + if (!liveCtx) { + return; + } + const settlementBelongsToLiveContext = kimiPromptSettlementBelongsToContext({ + liveAcpSessionId: liveCtx.acpSessionId, + expectedAcpSessionId, + liveActiveTurnId: liveCtx.activeTurnId, + liveSessionActiveTurnId: liveCtx.session.activeTurnId, + turnId, + }); + if (!settlementBelongsToLiveContext) { + // interruptTurn already consumed every prompt slot for this turn. A + // late prompt result must neither emit a second terminal event nor + // consume a slot belonging to a newer turn on the same ACP session. + if ( + liveCtx.acpSessionId !== expectedAcpSessionId || + liveCtx.interruptedTurnIds.has(turnId) + ) { + return; + } + if (options?.emitTurnCompletion !== false) { + if (options?.errorMessage !== undefined) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId, + payload: { + state: "failed", + errorMessage: options.errorMessage, + }, + }); + } else if (options?.completedStopReason !== undefined) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId, + payload: { + state: options.completedStopReason === "cancelled" ? "cancelled" : "completed", + stopReason: options.completedStopReason ?? null, + }, + }); + } + } + return; + } + let settleTurnId = turnId; + if (options?.settleAllPrompts) { + liveCtx.promptsInFlight = 0; + if (liveCtx.activeTurnId !== turnId && liveCtx.session.activeTurnId !== turnId) { + const fallbackTurnId = liveCtx.activeTurnId ?? liveCtx.session.activeTurnId; + if (!fallbackTurnId) { + if (liveCtx.session.status === "running" || liveCtx.session.status === "connecting") { + const updatedAt = yield* nowIso; + const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session; + liveCtx.activeTurnId = undefined; + liveCtx.session = { + ...readySession, + status: "ready", + updatedAt, + }; + } + return; + } + settleTurnId = fallbackTurnId; + } + } else { + const remainingPrompts = Math.max(0, liveCtx.promptsInFlight - 1); + if ( + remainingPrompts > 0 || + liveCtx.activeTurnId !== settleTurnId || + liveCtx.session.activeTurnId !== settleTurnId + ) { + liveCtx.promptsInFlight = remainingPrompts; + return; + } + liveCtx.promptsInFlight = remainingPrompts; + } + const updatedAt = yield* nowIso; + const canEmitTurnCompletion = + liveCtx.session.status === "running" || liveCtx.session.status === "connecting"; + const shouldEmitFailedTurn = options?.errorMessage !== undefined && canEmitTurnCompletion; + const shouldEmitCompletedTurn = + options?.completedStopReason !== undefined && canEmitTurnCompletion; + const { activeTurnId: _activeTurnId, ...readySession } = liveCtx.session; + liveCtx.activeTurnId = undefined; + liveCtx.session = { + ...readySession, + status: "ready", + updatedAt, + }; + if (options?.emitTurnCompletion === false) { + return; + } + if (shouldEmitFailedTurn) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId: settleTurnId, + payload: { + state: "failed", + errorMessage: options.errorMessage, + }, + }); + } else if (shouldEmitCompletedTurn) { + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId, + turnId: settleTurnId, + payload: { + state: options.completedStopReason === "cancelled" ? "cancelled" : "completed", + stopReason: options.completedStopReason ?? null, + }, + }); + } + }); + + const logNative = (threadId: ThreadId, method: string, payload: unknown) => + Effect.gen(function* () { + if (!nativeEventLogger) return; + const observedAt = yield* nowIso; + yield* nativeEventLogger.write( + { + observedAt, + event: { + id: yield* randomUUIDv4, + kind: "notification", + provider: PROVIDER, + createdAt: observedAt, + method, + threadId, + payload, + }, + }, + threadId, + ); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to write native Kimi notification log.", { + cause, + threadId, + method, + }), + ), + ); + + const emitPlanUpdate = ( + ctx: KimiSessionContext, + turnId: TurnId | undefined, + stamp: { readonly eventId: EventId; readonly createdAt: string }, + payload: { + readonly explanation?: string | null; + readonly plan: ReadonlyArray<{ + readonly step: string; + readonly status: "pending" | "inProgress" | "completed"; + }>; + }, + rawPayload: unknown, + method: string, + ) => + Effect.gen(function* () { + const fingerprint = `${turnId ?? "no-turn"}:${encodeJsonStringForDiagnostics(payload) ?? "[unserializable payload]"}`; + if (ctx.lastPlanFingerprint === fingerprint) { + return; + } + ctx.lastPlanFingerprint = fingerprint; + yield* offerRuntimeEvent( + makeAcpPlanUpdatedEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId, + payload, + source: "acp.jsonrpc", + method, + rawPayload, + }), + ); + }); + + const requireSession = ( + threadId: ThreadId, + ): Effect.Effect => { + const ctx = sessions.get(threadId); + if (!ctx || ctx.stopped) { + return Effect.fail( + new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId }), + ); + } + return Effect.succeed(ctx); + }; + + const stopSessionInternal = (ctx: KimiSessionContext) => + Effect.gen(function* () { + if (ctx.stopped) return; + ctx.stopped = true; + yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals); + yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs); + if (ctx.notificationFiber) { + yield* Fiber.interrupt(ctx.notificationFiber); + } + yield* Effect.ignore(Scope.close(ctx.scope, Exit.void)); + sessions.delete(ctx.threadId); + yield* offerRuntimeEvent({ + type: "session.exited", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: ctx.threadId, + payload: { exitKind: "graceful" }, + }); + }); + + const startSession: KimiAdapterShape["startSession"] = (input) => + withThreadLock( + input.threadId, + Effect.gen(function* () { + if (input.provider !== undefined && input.provider !== PROVIDER) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, + }); + } + if (!input.cwd?.trim()) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: "cwd is required and must be non-empty.", + }); + } + + const cwd = path.resolve(input.cwd.trim()); + const kimiModelSelection = + input.modelSelection?.instanceId === boundInstanceId ? input.modelSelection : undefined; + const existing = sessions.get(input.threadId); + if (existing && !existing.stopped) { + yield* stopSessionInternal(existing); + } + + const pendingApprovals = new Map(); + const pendingUserInputs = new Map(); + const sessionScope = yield* Scope.make("sequential"); + let sessionScopeTransferred = false; + yield* Effect.addFinalizer(() => + sessionScopeTransferred ? Effect.void : Scope.close(sessionScope, Exit.void), + ); + + const resumeSessionId = parseKimiResume(input.resumeCursor)?.sessionId; + const acpNativeLoggers = makeAcpNativeLoggers({ + nativeEventLogger, + provider: PROVIDER, + threadId: input.threadId, + }); + + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); + const acp = yield* makeKimiAcpRuntime({ + kimiSettings, + ...(options?.environment ? { environment: options.environment } : {}), + childProcessSpawner, + cwd, + ...(resumeSessionId ? { resumeSessionId } : {}), + clientInfo: { name: "t3-code", version: "0.0.0" }, + ...(mcpSession + ? { + mcpServers: [ + { + type: "http" as const, + name: "t3-code", + url: mcpSession.endpoint, + headers: [ + { + name: "Authorization", + value: mcpSession.authorizationHeader, + }, + ], + }, + ], + } + : {}), + ...acpNativeLoggers, + }).pipe( + Effect.provideService(Crypto.Crypto, crypto), + Effect.provideService(Scope.Scope, sessionScope), + Effect.mapError( + (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: cause.message, + cause, + }), + ), + ); + const started = yield* Effect.gen(function* () { + yield* acp.handleRequestPermission((params) => + mapAcpCallbackFailure( + Effect.gen(function* () { + yield* logNative(input.threadId, "session/request_permission", params); + if (input.runtimeMode === "full-access") { + const autoApprovedOptionId = selectAutoApprovedPermissionOption(params); + if (autoApprovedOptionId !== undefined) { + return { + outcome: { + outcome: "selected" as const, + optionId: autoApprovedOptionId, + }, + }; + } + } + const permissionRequest = parsePermissionRequest(params); + const requestId = ApprovalRequestId.make(yield* randomUUIDv4); + const runtimeRequestId = RuntimeRequestId.make(requestId); + const decision = yield* Deferred.make(); + const turnId = resolveSessionCallbackTurnId(sessions, input.threadId); + pendingApprovals.set(requestId, { decision }); + yield* offerRuntimeEvent( + makeAcpRequestOpenedEvent({ + stamp: yield* makeEventStamp(), + provider: PROVIDER, + threadId: input.threadId, + turnId, + requestId: runtimeRequestId, + permissionRequest, + detail: + permissionRequest.detail ?? + encodeJsonStringForDiagnostics(params)?.slice(0, 2000) ?? + "[unserializable params]", + args: params, + source: "acp.jsonrpc", + method: "session/request_permission", + rawPayload: params, + }), + ); + const resolved = yield* Deferred.await(decision); + pendingApprovals.delete(requestId); + yield* offerRuntimeEvent( + makeAcpRequestResolvedEvent({ + stamp: yield* makeEventStamp(), + provider: PROVIDER, + threadId: input.threadId, + turnId, + requestId: runtimeRequestId, + permissionRequest, + decision: resolved, + }), + ); + const selectedOptionId = + resolved === "cancel" ? undefined : selectPermissionOptionId(params, resolved); + return { + outcome: selectedOptionId + ? { + outcome: "selected" as const, + optionId: selectedOptionId, + } + : ({ outcome: "cancelled" } as const), + }; + }), + ), + ); + return yield* acp.start(); + }).pipe( + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/start", error), + ), + ); + + const requestedStartModelId = kimiModelSelection?.model + ? resolveKimiAcpBaseModelId(kimiModelSelection.model) + : undefined; + const boundModelId = yield* applyKimiAcpModelSelection({ + runtime: acp, + currentModelId: currentKimiModelIdFromSessionSetup(started.sessionSetupResult), + requestedModelId: requestedStartModelId, + selections: kimiModelSelection?.options, + mapError: (cause) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), + }); + + const now = yield* nowIso; + const session: ProviderSession = { + provider: PROVIDER, + providerInstanceId: boundInstanceId, + status: "ready", + runtimeMode: input.runtimeMode, + cwd, + ...(boundModelId ? { model: resolveKimiAcpBaseModelId(boundModelId) } : {}), + threadId: input.threadId, + resumeCursor: { + schemaVersion: KIMI_RESUME_VERSION, + sessionId: started.sessionId, + }, + createdAt: now, + updatedAt: now, + }; + + const ctx: KimiSessionContext = { + threadId: input.threadId, + acpSessionId: started.sessionId, + session, + scope: sessionScope, + acp, + notificationFiber: undefined, + pendingApprovals, + pendingUserInputs, + turns: [], + lastPlanFingerprint: undefined, + activeTurnId: undefined, + interruptedTurnIds: new Set(), + promptsInFlight: 0, + currentModelId: boundModelId, + stopped: false, + }; + + const nf = yield* Stream.runDrain( + Stream.mapEffect(acp.getEvents(), (event) => + Effect.gen(function* () { + if (event._tag === "EventStreamBarrier") { + yield* Deferred.succeed(event.acknowledge, undefined); + return; + } + if ( + event._tag === "PlanUpdated" || + event._tag === "ToolCallUpdated" || + event._tag === "ContentDelta" + ) { + yield* logNative(ctx.threadId, "session/update", event.rawPayload); + } + + if (event._tag === "ModeChanged") { + return; + } + + const notificationTurnId = resolveNotificationTurnId(ctx); + if ( + notificationTurnId === undefined || + ctx.interruptedTurnIds.has(notificationTurnId) + ) { + return; + } + const stamp = yield* makeEventStamp(); + + switch (event._tag) { + case "AssistantItemStarted": + yield* offerRuntimeEvent( + makeAcpAssistantItemEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + itemId: event.itemId, + lifecycle: "item.started", + }), + ); + return; + case "AssistantItemCompleted": + yield* offerRuntimeEvent( + makeAcpAssistantItemEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + itemId: event.itemId, + lifecycle: "item.completed", + }), + ); + return; + case "PlanUpdated": + yield* emitPlanUpdate( + ctx, + notificationTurnId, + stamp, + event.payload, + event.rawPayload, + "session/update", + ); + return; + case "ToolCallUpdated": + yield* offerRuntimeEvent( + makeAcpToolCallEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + toolCall: event.toolCall, + rawPayload: event.rawPayload, + }), + ); + return; + case "ContentDelta": + yield* offerRuntimeEvent( + makeAcpContentDeltaEvent({ + stamp, + provider: PROVIDER, + threadId: ctx.threadId, + turnId: notificationTurnId, + ...(event.itemId ? { itemId: event.itemId } : {}), + text: event.text, + streamKind: event.streamKind, + rawPayload: event.rawPayload, + }), + ); + return; + } + }), + ), + ).pipe( + Effect.catch((cause) => + Effect.logError("Failed to process Kimi runtime notification.", { cause }), + ), + Effect.forkChild, + ); + + ctx.notificationFiber = nf; + sessions.set(input.threadId, ctx); + sessionScopeTransferred = true; + + yield* offerRuntimeEvent({ + type: "session.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { resume: started.initializeResult }, + }); + yield* offerRuntimeEvent({ + type: "session.state.changed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { state: "ready", reason: "Kimi ACP session ready" }, + }); + yield* offerRuntimeEvent({ + type: "thread.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + payload: { providerThreadId: started.sessionId }, + }); + + return session; + }).pipe(Effect.scoped), + ); + + const sendTurn: KimiAdapterShape["sendTurn"] = (input) => + Effect.gen(function* () { + const prepared = yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(input.threadId); + // A sendTurn while a prompt is in flight is a steer: the agent + // folds the new prompt into the ongoing work, so the active turn + // id is reused instead of opening a new turn. + const steeringTurnId = ctx.promptsInFlight > 0 ? ctx.activeTurnId : undefined; + const turnId = steeringTurnId ?? TurnId.make(yield* randomUUIDv4); + // Count this prompt immediately so a superseded in-flight prompt + // resolving from here on does not settle the turn; decremented on + // preparation failure here, and after the prompt below otherwise. + ctx.promptsInFlight += 1; + // Bind the turn id before cooperative yields so interruptTurn can + // settle this prompt even if stop arrives during preparation. + ctx.activeTurnId = turnId; + ctx.session = { + ...ctx.session, + status: steeringTurnId === undefined ? "connecting" : "running", + activeTurnId: turnId, + updatedAt: yield* nowIso, + }; + + return yield* Effect.gen(function* () { + const turnModelSelection = + input.modelSelection?.instanceId === boundInstanceId + ? input.modelSelection + : undefined; + const requestedTurnModelId = turnModelSelection?.model + ? resolveKimiAcpBaseModelId(turnModelSelection.model) + : undefined; + const currentModelId = yield* applyKimiAcpModelSelection({ + runtime: ctx.acp, + currentModelId: ctx.currentModelId, + requestedModelId: requestedTurnModelId, + selections: turnModelSelection?.options, + mapError: (cause) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), + }); + + const text = input.input?.trim(); + const imagePromptParts = yield* Effect.forEach( + input.attachments ?? [], + (attachment) => + Effect.gen(function* () { + const attachmentPath = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }); + if (!attachmentPath) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: `Invalid attachment id '${attachment.id}'.`, + }); + } + const bytes = yield* fileSystem.readFile(attachmentPath).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: cause.message, + cause, + }), + ), + ); + return { + type: "image", + data: Buffer.from(bytes).toString("base64"), + mimeType: attachment.mimeType, + } satisfies EffectAcpSchema.ContentBlock; + }), + ); + const promptParts: Array = [ + ...(text ? [{ type: "text" as const, text }] : []), + ...imagePromptParts, + ]; + + if (promptParts.length === 0) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Turn requires non-empty text or attachments.", + }); + } + + ctx.currentModelId = currentModelId; + const displayModel = currentModelId + ? resolveKimiAcpBaseModelId(currentModelId) + : undefined; + for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { + yield* Effect.yieldNow; + } + if (ctx.interruptedTurnIds.has(turnId)) { + yield* settlePromptInFlight(input.threadId, turnId, ctx.acpSessionId, { + completedStopReason: "cancelled", + emitTurnCompletion: false, + settleAllPrompts: true, + }); + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: "Kimi prompt was interrupted during preparation.", + }); + } + if (steeringTurnId === undefined) { + ctx.lastPlanFingerprint = undefined; + } + ctx.session = { + ...ctx.session, + status: "running", + activeTurnId: turnId, + updatedAt: yield* nowIso, + ...(displayModel ? { model: displayModel } : {}), + }; + + if (steeringTurnId === undefined) { + yield* offerRuntimeEvent({ + type: "turn.started", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId, + payload: displayModel ? { model: displayModel } : {}, + }); + } + + return { + acp: ctx.acp, + acpSessionId: ctx.acpSessionId, + displayModel, + promptParts, + turnId, + }; + }).pipe( + Effect.tapCause(() => + Effect.gen(function* () { + const liveCtx = sessions.get(input.threadId); + if (!liveCtx) { + return; + } + yield* settlePromptInFlight(input.threadId, turnId, liveCtx.acpSessionId, { + errorMessage: "Kimi prompt preparation failed.", + emitTurnCompletion: false, + }); + }), + ), + ); + }), + ); + const promptSettled = yield* Ref.make(false); + const promptRpcSucceeded = yield* Ref.make(false); + const promptResultRef = yield* Ref.make( + undefined, + ); + + const promptFailureMessageRef = yield* Ref.make(undefined); + + return yield* Effect.gen(function* () { + const result = yield* prepared.acp + .prompt({ + prompt: prepared.promptParts, + }) + .pipe( + Effect.tap((promptResult) => + Effect.all([ + Ref.set(promptRpcSucceeded, true), + Ref.set(promptResultRef, promptResult), + ]), + ), + Effect.tapError((error) => + Ref.set( + promptFailureMessageRef, + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error).message, + ).pipe(Effect.andThen(prepared.acp.drainEvents)), + ), + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error), + ), + ); + + return yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(input.threadId); + if (ctx.acpSessionId !== prepared.acpSessionId) { + yield* settlePromptInFlight( + input.threadId, + prepared.turnId, + prepared.acpSessionId, + { + errorMessage: "Kimi session changed before the turn completed.", + settleAllPrompts: true, + }, + ); + yield* Ref.set(promptSettled, true); + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/prompt", + detail: "Kimi session changed before the turn completed.", + }); + } + // Keep prompt settlement atomic with respect to Stop and steering. + // interruptTurn marks its target before waiting for this lock, so + // cancellation can still win while queued ACP events are drained. + for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { + yield* Effect.yieldNow; + } + yield* prepared.acp.drainEvents; + if (ctx.interruptedTurnIds.has(prepared.turnId)) { + yield* Ref.set(promptSettled, true); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + } + + if ( + ctx.promptsInFlight <= 0 || + ctx.activeTurnId !== prepared.turnId || + ctx.session.activeTurnId !== prepared.turnId + ) { + yield* Ref.set(promptSettled, true); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + } + + appendPromptResultToTurn(ctx, prepared.turnId, prepared.promptParts, result); + ctx.session = { + ...ctx.session, + status: "running", + activeTurnId: prepared.turnId, + updatedAt: yield* nowIso, + ...(prepared.displayModel ? { model: prepared.displayModel } : {}), + }; + const remainingPrompts = Math.max(0, ctx.promptsInFlight - 1); + ctx.promptsInFlight = remainingPrompts; + + // Only the last remaining prompt settles the turn. A steer- + // superseded prompt resolving while another is in flight or + // pending must leave the merged turn running. + if ( + remainingPrompts === 0 && + ctx.activeTurnId === prepared.turnId && + ctx.session.activeTurnId === prepared.turnId + ) { + if (ctx.interruptedTurnIds.has(prepared.turnId)) { + yield* Ref.set(promptSettled, true); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + } + const completedAt = yield* nowIso; + const { activeTurnId: _completedTurnId, ...readySession } = ctx.session; + ctx.activeTurnId = undefined; + ctx.session = { + ...readySession, + status: "ready", + updatedAt: completedAt, + ...(prepared.displayModel ? { model: prepared.displayModel } : {}), + }; + const completedStopReason = completedStopReasonFromPromptResponse(result); + yield* offerRuntimeEvent({ + type: "turn.completed", + ...(yield* makeEventStamp()), + provider: PROVIDER, + threadId: input.threadId, + turnId: prepared.turnId, + payload: { + state: result.stopReason === "cancelled" ? "cancelled" : "completed", + stopReason: completedStopReason, + }, + }); + ctx.interruptedTurnIds.delete(prepared.turnId); + yield* Ref.set(promptSettled, true); + } else if (remainingPrompts > 0) { + yield* Ref.set(promptSettled, true); + } + + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + }), + ); + }).pipe( + Effect.ensuring( + Effect.gen(function* () { + if (yield* Ref.get(promptSettled)) { + return; + } + + if (yield* Ref.get(promptRpcSucceeded)) { + const promptResult = yield* Ref.get(promptResultRef); + if (promptResult === undefined) { + return; + } + yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(input.threadId); + if (ctx.acpSessionId !== prepared.acpSessionId) { + yield* settlePromptInFlight( + input.threadId, + prepared.turnId, + prepared.acpSessionId, + { + errorMessage: "Kimi session changed before the turn completed.", + settleAllPrompts: true, + }, + ); + return; + } + if (ctx.interruptedTurnIds.has(prepared.turnId)) { + return; + } + if ( + ctx.promptsInFlight <= 0 || + ctx.activeTurnId !== prepared.turnId || + ctx.session.activeTurnId !== prepared.turnId + ) { + return; + } + appendPromptResultToTurn( + ctx, + prepared.turnId, + prepared.promptParts, + promptResult, + ); + yield* settlePromptInFlight( + input.threadId, + prepared.turnId, + prepared.acpSessionId, + { + completedStopReason: completedStopReasonFromPromptResponse(promptResult), + }, + ); + }), + ); + return; + } + + const errorMessage = yield* Ref.get(promptFailureMessageRef); + yield* withThreadLock( + input.threadId, + settlePromptInFlight(input.threadId, prepared.turnId, prepared.acpSessionId, { + errorMessage: errorMessage ?? "Kimi prompt request failed.", + }), + ); + }).pipe(Effect.catch(() => Effect.void)), + ), + ); + }); + + const interruptTurn: KimiAdapterShape["interruptTurn"] = (threadId, turnId) => + Effect.gen(function* () { + const observed = yield* Effect.sync(() => { + const ctx = sessions.get(threadId); + if (!ctx || ctx.stopped) { + return { + _tag: "Proceed" as const, + acpSessionId: undefined, + interruptedTurnId: turnId, + }; + } + const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId; + if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) { + return { _tag: "Ignore" as const }; + } + const interruptedTurnId = turnId ?? activeTurnId; + if (interruptedTurnId !== undefined) { + ctx.interruptedTurnIds.add(interruptedTurnId); + } + return { + _tag: "Proceed" as const, + acpSessionId: ctx.acpSessionId, + interruptedTurnId, + }; + }); + if (observed._tag === "Ignore") { + return; + } + + yield* withThreadLock( + threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + if (observed.acpSessionId !== undefined && ctx.acpSessionId !== observed.acpSessionId) { + return; + } + const activeTurnId = ctx.activeTurnId ?? ctx.session.activeTurnId; + if (turnId !== undefined && activeTurnId !== undefined && activeTurnId !== turnId) { + return; + } + if ( + observed.interruptedTurnId !== undefined && + activeTurnId !== undefined && + activeTurnId !== observed.interruptedTurnId + ) { + return; + } + const interruptedTurnId = + observed.interruptedTurnId ?? turnId ?? activeTurnId ?? ctx.session.activeTurnId; + yield* settlePendingApprovalsAsCancelled(ctx.pendingApprovals); + yield* settlePendingUserInputsAsCancelled(ctx.pendingUserInputs); + yield* Effect.ignore( + ctx.acp.cancel.pipe( + Effect.mapError((error) => + mapAcpToAdapterError(PROVIDER, threadId, "session/cancel", error), + ), + ), + ); + if (interruptedTurnId) { + ctx.interruptedTurnIds.add(interruptedTurnId); + yield* settlePromptInFlight(threadId, interruptedTurnId, ctx.acpSessionId, { + completedStopReason: "cancelled", + settleAllPrompts: true, + }); + } else if ( + ctx.promptsInFlight > 0 || + ctx.session.status === "running" || + ctx.session.status === "connecting" + ) { + const updatedAt = yield* nowIso; + ctx.promptsInFlight = 0; + ctx.activeTurnId = undefined; + const { activeTurnId: _activeTurnId, ...readySession } = ctx.session; + ctx.session = { + ...readySession, + status: "ready", + updatedAt, + }; + } + }), + ); + }); + + const respondToRequest: KimiAdapterShape["respondToRequest"] = ( + threadId, + requestId, + decision, + ) => + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + const pending = ctx.pendingApprovals.get(requestId); + if (!pending) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/request_permission", + detail: `Unknown pending approval request: ${requestId}`, + }); + } + yield* Deferred.succeed(pending.decision, decision); + }); + + const respondToUserInput: KimiAdapterShape["respondToUserInput"] = ( + threadId, + requestId, + answers, + ) => + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + const pending = ctx.pendingUserInputs.get(requestId); + if (!pending) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/request_permission", + detail: `Unknown pending user-input request: ${requestId}`, + }); + } + yield* Deferred.succeed(pending.resolution, { _tag: "answered", answers }); + }); + + const readThread: KimiAdapterShape["readThread"] = (threadId) => + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + return { threadId, turns: ctx.turns }; + }); + + const rollbackThread: KimiAdapterShape["rollbackThread"] = (threadId, numTurns) => + Effect.gen(function* () { + yield* requireSession(threadId); + if (!Number.isInteger(numTurns) || numTurns < 1) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThread", + issue: "numTurns must be an integer >= 1.", + }); + } + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "thread/rollback", + detail: "Kimi ACP sessions do not support provider-side rollback yet.", + }); + }); + + const stopSession: KimiAdapterShape["stopSession"] = (threadId) => + withThreadLock( + threadId, + Effect.gen(function* () { + const ctx = yield* requireSession(threadId); + yield* stopSessionInternal(ctx); + }), + ); + + const listSessions: KimiAdapterShape["listSessions"] = () => + Effect.sync(() => Array.from(sessions.values(), (c) => ({ ...c.session }))); + + const hasSession: KimiAdapterShape["hasSession"] = (threadId) => + Effect.sync(() => { + const c = sessions.get(threadId); + return c !== undefined && !c.stopped; + }); + + const stopAll: KimiAdapterShape["stopAll"] = () => + Effect.forEach(Array.from(sessions.values()), stopSessionInternal, { discard: true }); + + yield* Effect.addFinalizer(() => + Effect.ignore(stopAll()).pipe( + Effect.tap(() => PubSub.shutdown(runtimeEventPubSub)), + Effect.tap(() => managedNativeEventLogger?.close() ?? Effect.void), + ), + ); + + const streamEvents = Stream.fromPubSub(runtimeEventPubSub); + + return { + provider: PROVIDER, + capabilities: { sessionModelSwitch: "in-session" }, + startSession, + sendTurn, + interruptTurn, + readThread, + rollbackThread, + respondToRequest, + respondToUserInput, + stopSession, + listSessions, + hasSession, + stopAll, + streamEvents, + } satisfies KimiAdapterShape; + }); +} diff --git a/apps/server/src/provider/Layers/KimiProvider.test.ts b/apps/server/src/provider/Layers/KimiProvider.test.ts new file mode 100644 index 00000000000..d02e94cf978 --- /dev/null +++ b/apps/server/src/provider/Layers/KimiProvider.test.ts @@ -0,0 +1,167 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodePath from "node:path"; +import * as NodeOS from "node:os"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import { KimiSettings } from "@t3tools/contracts"; + +import { buildInitialKimiProviderSnapshot, checkKimiProviderStatus } from "./KimiProvider.ts"; + +const decodeKimiSettings = Schema.decodeSync(KimiSettings); +const isWin = process.platform === "win32"; + +function writeFakeKimiBinary( + fs: FileSystem.FileSystem["Service"], + dir: string, + path: Path.Path["Service"], + scriptBody: string, +): Effect.Effect { + return Effect.gen(function* () { + const jsPath = path.join(dir, "kimi.mjs"); + yield* fs.writeFileString(jsPath, scriptBody); + if (isWin) { + const cmdPath = path.join(dir, "kimi.cmd"); + yield* fs.writeFileString( + cmdPath, + ["@echo off", `node "${jsPath.replaceAll("/", "\\")}" %*`, ""].join("\r\n"), + ); + return cmdPath; + } + const shPath = path.join(dir, "kimi"); + yield* fs.writeFileString( + shPath, + ["#!/bin/sh", `exec "${process.execPath}" "${jsPath}" "$@"`, ""].join("\n"), + ); + yield* fs.chmod(shPath, 0o755); + return shPath; + }); +} + +describe("buildInitialKimiProviderSnapshot", () => { + it.effect("returns a disabled snapshot when settings.enabled is false", () => + Effect.gen(function* () { + const snapshot = yield* buildInitialKimiProviderSnapshot( + decodeKimiSettings({ enabled: false }), + ); + expect(snapshot.enabled).toBe(false); + expect(snapshot.status).toBe("disabled"); + expect(snapshot.installed).toBe(false); + expect(snapshot.message).toContain("disabled"); + }), + ); + + it.effect("returns a pending snapshot by default", () => + Effect.gen(function* () { + const snapshot = yield* buildInitialKimiProviderSnapshot(decodeKimiSettings({})); + expect(snapshot.enabled).toBe(true); + expect(snapshot.installed).toBe(true); + expect(snapshot.status).toBe("warning"); + expect(snapshot.version).toBeNull(); + expect(snapshot.message).toContain("Checking Kimi"); + expect(snapshot.requiresNewThreadForModelChange).toBe(true); + expect(snapshot.models.map((model) => model.slug)).toEqual([ + "kimi-code/k3", + "kimi-code/k3-256k", + "kimi-code/kimi-for-coding", + "kimi-code/kimi-for-coding-highspeed", + ]); + }), + ); +}); + +it.layer(NodeServices.layer)("checkKimiProviderStatus", (it) => { + it.effect("reports the binary as missing when the binary path does not resolve", () => + Effect.gen(function* () { + const snapshot = yield* checkKimiProviderStatus( + decodeKimiSettings({ + enabled: true, + binaryPath: NodePath.join(NodeOS.tmpdir(), "definitely-not-installed-kimi-binary"), + }), + ); + expect(snapshot.enabled).toBe(true); + expect(snapshot.installed).toBe(false); + expect(snapshot.status).toBe("error"); + expect(snapshot.message).toMatch(/not installed|not on PATH|Failed to execute/); + }), + ); + + it.effect("reports an installed CLI as unhealthy when --version exits non-zero", () => + Effect.gen(function* () { + const secretStderr = "broken kimi install: secret-token-value"; + const snapshot = yield* Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* fs.makeTempDirectoryScoped({ prefix: "t3code-kimi-version-" }); + const kimiPath = yield* writeFakeKimiBinary( + fs, + dir, + path, + [ + "const args = process.argv.slice(2);", + `process.stderr.write(${JSON.stringify(secretStderr + "\\n")});`, + "process.exit(2);", + "", + ].join("\n"), + ); + + return yield* checkKimiProviderStatus( + decodeKimiSettings({ enabled: true, binaryPath: kimiPath }), + ); + }), + ); + + expect(snapshot.enabled).toBe(true); + expect(snapshot.installed).toBe(true); + expect(snapshot.status).toBe("error"); + expect(snapshot.message).toBe("Kimi CLI is installed but failed to run."); + expect(snapshot.message).not.toContain(secretStderr); + }), + ); + + it.effect("reports an error when ACP model discovery is unavailable", () => + Effect.gen(function* () { + const snapshot = yield* Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const dir = yield* fs.makeTempDirectoryScoped({ prefix: "t3code-kimi-success-" }); + const kimiPath = yield* writeFakeKimiBinary( + fs, + dir, + path, + [ + "const args = process.argv.slice(2);", + 'if (args[0] === "--version") {', + ' process.stdout.write("kimi-cli 0.0.99\\n");', + " process.exit(0);", + "}", + 'process.stderr.write("not an acp agent\\n");', + "process.exit(1);", + "", + ].join("\n"), + ); + + return yield* checkKimiProviderStatus( + decodeKimiSettings({ enabled: true, binaryPath: kimiPath }), + ); + }), + ); + + expect(snapshot.status).toBe("error"); + expect(snapshot.installed).toBe(true); + expect(snapshot.models.map((model) => model.slug)).toEqual([ + "kimi-code/k3", + "kimi-code/k3-256k", + "kimi-code/kimi-for-coding", + "kimi-code/kimi-for-coding-highspeed", + ]); + expect(snapshot.message).toContain("ACP startup failed"); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/KimiProvider.ts b/apps/server/src/provider/Layers/KimiProvider.ts new file mode 100644 index 00000000000..3be83fab685 --- /dev/null +++ b/apps/server/src/provider/Layers/KimiProvider.ts @@ -0,0 +1,474 @@ +import { + type KimiSettings, + type ModelCapabilities, + type ServerProvider, + type ServerProviderModel, +} from "@t3tools/contracts"; +import type * as EffectAcpSchema from "effect-acp/schema"; +import { causeErrorTag } from "@t3tools/shared/observability"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import * as Result from "effect/Result"; +import { HttpClient } from "effect/unstable/http"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; +import { createModelCapabilities } from "@t3tools/shared/model"; +import { resolveSpawnCommand } from "@t3tools/shared/shell"; + +import { + buildSelectOptionDescriptor, + buildServerProvider, + isCommandMissingCause, + parseGenericCliVersion, + providerModelsFromSettings, + spawnAndCollect, + type ServerProviderDraft, +} from "../providerSnapshot.ts"; +import { + enrichProviderSnapshotWithVersionAdvisory, + type ProviderMaintenanceCapabilities, +} from "../providerMaintenance.ts"; +import { makeKimiAcpRuntime, resolveKimiAcpBaseModelId } from "../acp/KimiAcpSupport.ts"; +import { collectSessionConfigOptionValues } from "../acp/AcpRuntimeModel.ts"; + +const KIMI_PRESENTATION = { + displayName: "Kimi", + badgeLabel: "Early Access", + showInteractionModeToggle: false, + requiresNewThreadForModelChange: true, +} as const; +const EMPTY_CAPABILITIES: ModelCapabilities = createModelCapabilities({ + optionDescriptors: [], +}); + +/** Default Thinking levels when ACP discovery is unavailable (Kimi K3 family). */ +const DEFAULT_THINKING_CAPABILITIES: ModelCapabilities = createModelCapabilities({ + optionDescriptors: [ + buildSelectOptionDescriptor({ + id: "reasoning", + label: "Thinking", + options: [ + { value: "low", label: "Low" }, + { value: "high", label: "High", isDefault: true }, + { value: "max", label: "Max" }, + ], + }), + ], +}); + +const VERSION_PROBE_TIMEOUT_MS = 4_000; +const KIMI_ACP_MODEL_DISCOVERY_TIMEOUT_MS = 15_000; + +/** + * Catalog fallback when ACP discovery is unavailable. + * Wire IDs must use the `kimi-code/` provider prefix that `kimi acp` advertises + * on `session/set_model` / `session/set_config_option` (bare ids return Internal error). + */ +const KIMI_BUILT_IN_MODELS: ReadonlyArray = [ + { + slug: "kimi-code/k3", + name: "Kimi K3", + isCustom: false, + capabilities: DEFAULT_THINKING_CAPABILITIES, + }, + { + slug: "kimi-code/k3-256k", + name: "Kimi K3 256K", + isCustom: false, + capabilities: DEFAULT_THINKING_CAPABILITIES, + }, + { + slug: "kimi-code/kimi-for-coding", + name: "Kimi K2.7 Code", + isCustom: false, + capabilities: DEFAULT_THINKING_CAPABILITIES, + }, + { + slug: "kimi-code/kimi-for-coding-highspeed", + name: "Kimi K2.7 Code HighSpeed", + isCustom: false, + capabilities: DEFAULT_THINKING_CAPABILITIES, + }, +]; + +export function buildInitialKimiProviderSnapshot( + kimiSettings: KimiSettings, +): Effect.Effect { + return Effect.gen(function* () { + const checkedAt = yield* Effect.map(DateTime.now, DateTime.formatIso); + const models = kimiModelsFromSettings(kimiSettings.customModels); + + if (!kimiSettings.enabled) { + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: false, + checkedAt, + models, + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Kimi is disabled in T3 Code settings.", + }, + }); + } + + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: true, + checkedAt, + models, + probe: { + installed: true, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Checking Kimi CLI availability...", + }, + }); + }); +} + +function kimiModelsFromSettings( + customModels: ReadonlyArray | undefined, + builtInModels: ReadonlyArray = KIMI_BUILT_IN_MODELS, +): ReadonlyArray { + return providerModelsFromSettings( + builtInModels, + customModels ?? [], + DEFAULT_THINKING_CAPABILITIES, + ); +} + +function buildKimiThinkingCapabilitiesFromConfigOptions( + configOptions: ReadonlyArray | null | undefined, +): ModelCapabilities { + if (!configOptions || configOptions.length === 0) { + return DEFAULT_THINKING_CAPABILITIES; + } + const thinkingOption = configOptions.find( + (option) => + option.id === "thinking" || + option.category === "thought_level" || + option.name?.trim().toLowerCase() === "thinking", + ); + if (!thinkingOption || thinkingOption.type !== "select") { + return DEFAULT_THINKING_CAPABILITIES; + } + const values = collectSessionConfigOptionValues(thinkingOption); + if (values.length === 0) { + return DEFAULT_THINKING_CAPABILITIES; + } + const current = + typeof thinkingOption.currentValue === "string" ? thinkingOption.currentValue.trim() : ""; + const labelsByValue = new Map(); + for (const entry of thinkingOption.options) { + if ("value" in entry && typeof entry.value === "string") { + labelsByValue.set(entry.value, entry.name?.trim() || entry.value); + } else if ("options" in entry) { + for (const nested of entry.options) { + labelsByValue.set(nested.value, nested.name?.trim() || nested.value); + } + } + } + return createModelCapabilities({ + optionDescriptors: [ + buildSelectOptionDescriptor({ + id: "reasoning", + label: thinkingOption.name?.trim() || "Thinking", + options: values.map((value) => ({ + value, + label: labelsByValue.get(value) ?? value, + ...(current === value ? { isDefault: true } : {}), + })), + }), + ], + }); +} + +function buildKimiDiscoveredModelsFromConfigOptions( + configOptions: ReadonlyArray | null | undefined, +): ReadonlyArray { + if (!configOptions || configOptions.length === 0) { + return []; + } + const modelOption = configOptions.find( + (option) => option.id === "model" || option.category === "model", + ); + if (!modelOption || modelOption.type !== "select") { + return []; + } + const capabilities = buildKimiThinkingCapabilitiesFromConfigOptions(configOptions); + const seen = new Set(); + return modelOption.options.flatMap((entry): Array => { + if (!("value" in entry) || typeof entry.value !== "string") { + return []; + } + const slug = resolveKimiAcpBaseModelId(entry.value); + if (!slug || seen.has(slug)) { + return []; + } + seen.add(slug); + const name = + "name" in entry && typeof entry.name === "string" && entry.name.trim().length > 0 + ? entry.name.trim() + : slug; + return [ + { + slug, + name, + isCustom: false, + capabilities, + }, + ]; + }); +} + +function buildKimiDiscoveredModelsFromSessionModelState( + modelState: EffectAcpSchema.SessionModelState | null | undefined, + capabilities: ModelCapabilities = DEFAULT_THINKING_CAPABILITIES, +): ReadonlyArray { + if (!modelState || modelState.availableModels.length === 0) { + return []; + } + const seen = new Set(); + return modelState.availableModels + .map((model): ServerProviderModel | undefined => { + const slug = resolveKimiAcpBaseModelId(model.modelId); + if (!slug || seen.has(slug)) { + return undefined; + } + seen.add(slug); + return { + slug, + name: model.name.trim() || slug, + isCustom: false, + capabilities, + }; + }) + .filter((model): model is ServerProviderModel => model !== undefined); +} + +const discoverKimiModelsViaAcp = ( + kimiSettings: KimiSettings, + environment: NodeJS.ProcessEnv = process.env, +) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const acp = yield* makeKimiAcpRuntime({ + kimiSettings, + environment, + childProcessSpawner, + cwd: process.cwd(), + clientInfo: { name: "t3-code-provider-probe", version: "0.0.0" }, + }); + const started = yield* acp.start(); + const configOptions = started.sessionSetupResult.configOptions ?? []; + const capabilities = buildKimiThinkingCapabilitiesFromConfigOptions(configOptions); + const fromConfig = buildKimiDiscoveredModelsFromConfigOptions(configOptions); + if (fromConfig.length > 0) { + return fromConfig; + } + return buildKimiDiscoveredModelsFromSessionModelState( + started.sessionSetupResult.models, + capabilities, + ); + }).pipe(Effect.scoped); + +const runKimiVersionCommand = ( + kimiSettings: KimiSettings, + environment: NodeJS.ProcessEnv = process.env, +) => + Effect.gen(function* () { + const command = kimiSettings.binaryPath || "kimi"; + const spawnCommand = yield* resolveSpawnCommand(command, ["--version"], { + env: environment, + }); + return yield* spawnAndCollect( + command, + ChildProcess.make(spawnCommand.command, spawnCommand.args, { + env: environment, + shell: spawnCommand.shell, + }), + ); + }); + +export const checkKimiProviderStatus = Effect.fn("checkKimiProviderStatus")(function* ( + kimiSettings: KimiSettings, + environment: NodeJS.ProcessEnv = process.env, +): Effect.fn.Return< + ServerProviderDraft, + never, + ChildProcessSpawner.ChildProcessSpawner | Crypto.Crypto +> { + const checkedAt = DateTime.formatIso(yield* DateTime.now); + const fallbackModels = kimiModelsFromSettings(kimiSettings.customModels); + + if (!kimiSettings.enabled) { + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: false, + checkedAt, + models: fallbackModels, + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Kimi is disabled in T3 Code settings.", + }, + }); + } + + const versionResult = yield* runKimiVersionCommand(kimiSettings, environment).pipe( + Effect.timeoutOption(VERSION_PROBE_TIMEOUT_MS), + Effect.result, + ); + + if (Result.isFailure(versionResult)) { + const error = versionResult.failure; + yield* Effect.logWarning("Kimi CLI health check failed.", { + errorTag: error._tag, + }); + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models: fallbackModels, + probe: { + installed: !isCommandMissingCause(error), + version: null, + status: "error", + auth: { status: "unknown" }, + message: isCommandMissingCause(error) + ? "Kimi Code CLI (`kimi`) is not installed or not on PATH." + : "Failed to execute Kimi CLI health check.", + }, + }); + } + + if (Option.isNone(versionResult.success)) { + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models: fallbackModels, + probe: { + installed: true, + version: null, + status: "error", + auth: { status: "unknown" }, + message: "Kimi CLI is installed but timed out while running `kimi --version`.", + }, + }); + } + + const versionOutput = versionResult.success.value; + const version = parseGenericCliVersion(`${versionOutput.stdout}\n${versionOutput.stderr}`); + if (versionOutput.code !== 0) { + yield* Effect.logWarning("Kimi CLI version probe exited with a non-zero status.", { + exitCode: versionOutput.code, + stdoutLength: versionOutput.stdout.length, + stderrLength: versionOutput.stderr.length, + }); + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models: fallbackModels, + probe: { + installed: true, + version, + status: "error", + auth: { status: "unknown" }, + message: "Kimi CLI is installed but failed to run.", + }, + }); + } + + const discoveryExit = yield* discoverKimiModelsViaAcp(kimiSettings, environment).pipe( + Effect.timeoutOption(KIMI_ACP_MODEL_DISCOVERY_TIMEOUT_MS), + Effect.exit, + ); + if (Exit.isFailure(discoveryExit)) { + yield* Effect.logWarning("Kimi ACP model discovery failed", { + errorTag: causeErrorTag(discoveryExit.cause), + }); + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models: fallbackModels, + probe: { + installed: true, + version, + status: "error", + auth: { status: "unknown" }, + message: + "Kimi CLI is installed but ACP startup failed. Run `kimi` and `/login` once, then retry.", + }, + }); + } + if (Option.isNone(discoveryExit.value)) { + yield* Effect.logWarning( + `Kimi ACP model discovery timed out after ${KIMI_ACP_MODEL_DISCOVERY_TIMEOUT_MS}ms.`, + ); + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models: fallbackModels, + probe: { + installed: true, + version, + status: "error", + auth: { status: "unknown" }, + message: `Kimi CLI is installed but ACP startup timed out after ${KIMI_ACP_MODEL_DISCOVERY_TIMEOUT_MS}ms.`, + }, + }); + } + const discoveredModels = discoveryExit.value.value; + const models = + discoveredModels.length > 0 + ? kimiModelsFromSettings(kimiSettings.customModels, discoveredModels) + : fallbackModels; + + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models, + probe: { + installed: true, + version, + status: "ready", + auth: { status: "unknown" }, + }, + }); +}); + +export const enrichKimiSnapshot = (input: { + readonly snapshot: ServerProvider; + readonly maintenanceCapabilities: ProviderMaintenanceCapabilities; + readonly enableProviderUpdateChecks?: boolean; + readonly publishSnapshot: (snapshot: ServerProvider) => Effect.Effect; + readonly httpClient: HttpClient.HttpClient; +}): Effect.Effect => { + const { snapshot, publishSnapshot } = input; + + return enrichProviderSnapshotWithVersionAdvisory(snapshot, input.maintenanceCapabilities, { + enableProviderUpdateChecks: input.enableProviderUpdateChecks, + }).pipe( + Effect.provideService(HttpClient.HttpClient, input.httpClient), + Effect.flatMap((enrichedSnapshot) => publishSnapshot(enrichedSnapshot)), + Effect.catchCause((cause) => + Effect.logWarning("Kimi version advisory enrichment failed", { + errorTag: causeErrorTag(cause), + }), + ), + Effect.asVoid, + ); +}; diff --git a/apps/server/src/provider/Services/KimiAdapter.ts b/apps/server/src/provider/Services/KimiAdapter.ts new file mode 100644 index 00000000000..a7e5a6c7be4 --- /dev/null +++ b/apps/server/src/provider/Services/KimiAdapter.ts @@ -0,0 +1,16 @@ +/** + * KimiAdapter — shape type for the Kimi Code CLI provider adapter. + * + * The driver model ({@link ../Drivers/KimiDriver}) bundles one adapter per + * instance as a captured closure, so this module only retains the shape + * interface as a naming anchor for the driver bundle. + * + * @module KimiAdapter + */ +import type { ProviderAdapterError } from "../Errors.ts"; +import type { ProviderAdapterShape } from "./ProviderAdapter.ts"; + +/** + * KimiAdapterShape — per-instance Kimi adapter contract. + */ +export interface KimiAdapterShape extends ProviderAdapterShape {} diff --git a/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts b/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts index c93e61dc37b..ab9fd6ff5d9 100644 --- a/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts +++ b/apps/server/src/provider/acp/AcpCoreRuntimeEvents.ts @@ -220,6 +220,7 @@ export function makeAcpContentDeltaEvent(input: { readonly turnId: TurnId | undefined; readonly itemId?: string; readonly text: string; + readonly streamKind?: "assistant_text" | "reasoning_text"; readonly rawPayload: unknown; }): ProviderRuntimeEvent { return { @@ -230,7 +231,7 @@ export function makeAcpContentDeltaEvent(input: { turnId: input.turnId, ...(input.itemId ? { itemId: RuntimeItemId.make(input.itemId) } : {}), payload: { - streamKind: "assistant_text", + streamKind: input.streamKind ?? "assistant_text", delta: input.text, }, raw: { diff --git a/apps/server/src/provider/acp/AcpRuntimeModel.test.ts b/apps/server/src/provider/acp/AcpRuntimeModel.test.ts index 7682c5f5f9c..6871b983f7d 100644 --- a/apps/server/src/provider/acp/AcpRuntimeModel.test.ts +++ b/apps/server/src/provider/acp/AcpRuntimeModel.test.ts @@ -322,6 +322,7 @@ describe("AcpRuntimeModel", () => { { _tag: "ContentDelta", text: "hello from acp", + streamKind: "assistant_text", rawPayload: { sessionId: "session-1", update: { @@ -334,6 +335,34 @@ describe("AcpRuntimeModel", () => { }, }, ]); + + const thoughtResult = parseSessionUpdateEvent({ + sessionId: "session-1", + update: { + sessionUpdate: "agent_thought_chunk", + content: { + type: "text", + text: "thinking out loud", + }, + }, + } satisfies EffectAcpSchema.SessionNotification); + expect(thoughtResult.events).toEqual([ + { + _tag: "ContentDelta", + text: "thinking out loud", + streamKind: "reasoning_text", + rawPayload: { + sessionId: "session-1", + update: { + sessionUpdate: "agent_thought_chunk", + content: { + type: "text", + text: "thinking out loud", + }, + }, + }, + }, + ]); }); it("keeps permission request parsing compatible with loose extension payloads", () => { diff --git a/apps/server/src/provider/acp/AcpRuntimeModel.ts b/apps/server/src/provider/acp/AcpRuntimeModel.ts index e6bfc127e6e..907e0834913 100644 --- a/apps/server/src/provider/acp/AcpRuntimeModel.ts +++ b/apps/server/src/provider/acp/AcpRuntimeModel.ts @@ -107,6 +107,8 @@ export type AcpParsedSessionEvent = readonly _tag: "ContentDelta"; readonly itemId?: string; readonly text: string; + /** ACP assistant vs thought streams. Defaults to assistant_text when omitted. */ + readonly streamKind: "assistant_text" | "reasoning_text"; readonly rawPayload: unknown; }; @@ -569,6 +571,18 @@ export function parseSessionUpdateEvent(params: EffectAcpSchema.SessionNotificat events.push({ _tag: "ContentDelta", text: upd.content.text, + streamKind: "assistant_text", + rawPayload: params, + }); + } + break; + } + case "agent_thought_chunk": { + if (upd.content.type === "text" && upd.content.text.length > 0) { + events.push({ + _tag: "ContentDelta", + text: upd.content.text, + streamKind: "reasoning_text", rawPayload: params, }); } diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index 09fce6d56f9..38345e6bb71 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -891,12 +891,20 @@ const handleSessionUpdate = ({ continue; } if (event._tag === "ContentDelta") { + // Thought streams must not open assistant message segments — they are + // projected separately as reasoning/worklog activity. + if (event.streamKind === "reasoning_text") { + yield* Queue.offer(queue, event); + continue; + } if (event.text.trim().length === 0) { const assistantSegmentState = yield* Ref.get(assistantSegmentRef); if (!assistantSegmentState.activeItemId) { continue; } } + // Thought → answer boundary: close any open assistant segment first so + // a new answer segment starts cleanly after tools/thought. const itemId = yield* ensureActiveAssistantSegment({ queue, assistantSegmentRef, diff --git a/apps/server/src/provider/acp/KimiAcpSupport.test.ts b/apps/server/src/provider/acp/KimiAcpSupport.test.ts new file mode 100644 index 00000000000..e176d39689c --- /dev/null +++ b/apps/server/src/provider/acp/KimiAcpSupport.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as EffectAcpErrors from "effect-acp/errors"; + +import { + applyKimiAcpModelSelection, + buildKimiAcpSpawnInput, + resolveKimiAcpBaseModelId, +} from "./KimiAcpSupport.ts"; + +describe("resolveKimiAcpBaseModelId", () => { + it("normalizes empty, short, and prefixed Kimi model ids", () => { + expect(resolveKimiAcpBaseModelId(undefined)).toBe("kimi-code/kimi-for-coding"); + expect(resolveKimiAcpBaseModelId(" ")).toBe("kimi-code/kimi-for-coding"); + expect(resolveKimiAcpBaseModelId("k3")).toBe("kimi-code/k3"); + expect(resolveKimiAcpBaseModelId("k3-256k")).toBe("kimi-code/k3-256k"); + expect(resolveKimiAcpBaseModelId("kimi-for-coding")).toBe("kimi-code/kimi-for-coding"); + expect(resolveKimiAcpBaseModelId("kimi-for-coding-highspeed")).toBe( + "kimi-code/kimi-for-coding-highspeed", + ); + expect(resolveKimiAcpBaseModelId("kimi-code/k3")).toBe("kimi-code/k3"); + expect(resolveKimiAcpBaseModelId(" custom-provider/my-model ")).toBe( + "custom-provider/my-model", + ); + }); +}); + +describe("buildKimiAcpSpawnInput", () => { + it("spawns `kimi acp` with optional binary path and environment", () => { + const spawn = buildKimiAcpSpawnInput({ binaryPath: "/usr/local/bin/kimi" }, "/tmp/project", { + KIMI_CODE_HOME: "/tmp/kimi-home", + }); + + expect(spawn).toEqual({ + command: "/usr/local/bin/kimi", + args: ["acp"], + cwd: "/tmp/project", + env: { + KIMI_CODE_HOME: "/tmp/kimi-home", + }, + }); + }); + + it("defaults the command to `kimi` when no binary path is set", () => { + const spawn = buildKimiAcpSpawnInput(undefined, "/work"); + expect(spawn).toEqual({ + command: "kimi", + args: ["acp"], + cwd: "/work", + }); + }); +}); + +describe("applyKimiAcpModelSelection", () => { + const makeRecordingRuntime = (failure?: EffectAcpErrors.AcpError) => { + const modelCalls: Array = []; + const runtime = { + setSessionModel: (modelId: string) => + Effect.gen(function* () { + modelCalls.push(modelId); + if (failure) return yield* failure; + return {}; + }), + }; + return { runtime, modelCalls }; + }; + + it.effect("calls session/set_model when the requested model differs from current", () => + Effect.gen(function* () { + const { runtime, modelCalls } = makeRecordingRuntime(); + const result = yield* applyKimiAcpModelSelection({ + runtime, + currentModelId: "kimi-code/kimi-for-coding", + requestedModelId: "kimi-code/k3", + mapError: (cause) => cause.message, + }); + expect(modelCalls).toEqual(["kimi-code/k3"]); + expect(result).toBe("kimi-code/k3"); + }), + ); + + it.effect("skips set_model when requested matches current", () => + Effect.gen(function* () { + const { runtime, modelCalls } = makeRecordingRuntime(); + const result = yield* applyKimiAcpModelSelection({ + runtime, + currentModelId: "kimi-code/kimi-for-coding", + requestedModelId: "kimi-code/kimi-for-coding", + mapError: (cause) => cause.message, + }); + expect(modelCalls).toEqual([]); + expect(result).toBe("kimi-code/kimi-for-coding"); + }), + ); + + it.effect("skips set_model when no model is requested", () => + Effect.gen(function* () { + const { runtime, modelCalls } = makeRecordingRuntime(); + const result = yield* applyKimiAcpModelSelection({ + runtime, + currentModelId: "kimi-code/kimi-for-coding", + requestedModelId: undefined, + mapError: (cause) => cause.message, + }); + expect(modelCalls).toEqual([]); + expect(result).toBe("kimi-code/kimi-for-coding"); + }), + ); + + it.effect("propagates session/set_model failures via mapError", () => + Effect.gen(function* () { + const failure = EffectAcpErrors.AcpRequestError.invalidParams("session id not known"); + const { runtime } = makeRecordingRuntime(failure); + const error = yield* Effect.flip( + applyKimiAcpModelSelection({ + runtime, + currentModelId: "kimi-code/kimi-for-coding", + requestedModelId: "other", + mapError: (cause) => cause.message, + }), + ); + expect(error).toContain("session id not known"); + }), + ); +}); diff --git a/apps/server/src/provider/acp/KimiAcpSupport.ts b/apps/server/src/provider/acp/KimiAcpSupport.ts new file mode 100644 index 00000000000..08a3f37f487 --- /dev/null +++ b/apps/server/src/provider/acp/KimiAcpSupport.ts @@ -0,0 +1,208 @@ +import { + type KimiSettings, + type ProviderOptionSelection, + ProviderDriverKind, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Scope from "effect/Scope"; +import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner"; +import * as EffectAcpErrors from "effect-acp/errors"; +import type * as EffectAcpSchema from "effect-acp/schema"; +import { getProviderOptionStringSelectionValue, normalizeModelSlug } from "@t3tools/shared/model"; + +import * as AcpSessionRuntime from "./AcpSessionRuntime.ts"; +import { collectSessionConfigOptionValues, findSessionConfigOption } from "./AcpRuntimeModel.ts"; + +const KIMI_DRIVER_KIND = ProviderDriverKind.make("kimi"); +/** Auth method advertised by `kimi acp` (see Moonshot Kimi Code CLI ACP docs). */ +const KIMI_AUTH_METHOD_LOGIN = "login"; +/** + * Available to all Kimi Code members; used when the UI has no selection yet. + * Must be the full ACP model id (provider prefix + alias). + */ +const DEFAULT_KIMI_MODEL = "kimi-code/kimi-for-coding"; +const KIMI_MODEL_PROVIDER_PREFIX = "kimi-code/"; + +/** Short catalog aliases → full ACP model ids advertised by `kimi acp`. */ +const KIMI_MODEL_ALIASES: Readonly> = { + k3: "kimi-code/k3", + "k3-256k": "kimi-code/k3-256k", + "kimi-for-coding": "kimi-code/kimi-for-coding", + "kimi-for-coding-highspeed": "kimi-code/kimi-for-coding-highspeed", + // legacy fallback slug used before the Kimi Code model catalog was wired + "kimi-k2.5": "kimi-code/kimi-for-coding", +}; + +type KimiAcpRuntimeKimiSettings = Pick; + +interface KimiAcpRuntimeInput extends Omit< + AcpSessionRuntime.AcpSessionRuntimeOptions, + "authMethodId" | "clientCapabilities" | "spawn" +> { + readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; + readonly kimiSettings: KimiAcpRuntimeKimiSettings | null | undefined; + readonly environment?: NodeJS.ProcessEnv; +} + +export interface KimiAcpModelSelectionErrorContext { + readonly cause: EffectAcpErrors.AcpError; + readonly step: "set-config-option" | "set-model"; + readonly configId?: string; +} + +export function buildKimiAcpSpawnInput( + kimiSettings: KimiAcpRuntimeKimiSettings | null | undefined, + cwd: string, + environment?: NodeJS.ProcessEnv, +): AcpSessionRuntime.AcpSpawnInput { + return { + command: kimiSettings?.binaryPath || "kimi", + args: ["acp"], + cwd, + ...(environment ? { env: environment } : {}), + }; +} + +export const makeKimiAcpRuntime = ( + input: KimiAcpRuntimeInput, +): Effect.Effect< + AcpSessionRuntime.AcpSessionRuntime["Service"], + EffectAcpErrors.AcpError, + Crypto.Crypto | Scope.Scope +> => + Effect.gen(function* () { + const acpContext = yield* Layer.build( + AcpSessionRuntime.layer({ + ...input, + spawn: buildKimiAcpSpawnInput(input.kimiSettings, input.cwd, input.environment), + authMethodId: KIMI_AUTH_METHOD_LOGIN, + }).pipe( + Layer.provide( + Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, input.childProcessSpawner), + ), + ), + ); + return yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe( + Effect.provide(acpContext), + ); + }); + +export function resolveKimiAcpBaseModelId(model: string | null | undefined): string { + const trimmed = model?.trim(); + const raw = trimmed && trimmed.length > 0 ? trimmed : DEFAULT_KIMI_MODEL; + const aliased = KIMI_MODEL_ALIASES[raw] ?? raw; + const withPrefix = + aliased.includes("/") || aliased.startsWith(KIMI_MODEL_PROVIDER_PREFIX) + ? aliased + : `${KIMI_MODEL_PROVIDER_PREFIX}${aliased}`; + return normalizeModelSlug(withPrefix, KIMI_DRIVER_KIND) ?? DEFAULT_KIMI_MODEL; +} + +export function currentKimiModelIdFromSessionSetup( + sessionSetupResult: + | EffectAcpSchema.LoadSessionResponse + | EffectAcpSchema.NewSessionResponse + | EffectAcpSchema.ResumeSessionResponse, +): string | undefined { + const fromModels = sessionSetupResult.models?.currentModelId?.trim(); + if (fromModels) { + return resolveKimiAcpBaseModelId(fromModels); + } + // Kimi advertises the live model on the `model` config option, not SessionModelState. + const modelOption = findSessionConfigOption(sessionSetupResult.configOptions ?? [], "model"); + if (modelOption?.type === "select" && typeof modelOption.currentValue === "string") { + const value = modelOption.currentValue.trim(); + return value ? resolveKimiAcpBaseModelId(value) : undefined; + } + return undefined; +} + +function resolveKimiConfigUpdates( + configOptions: ReadonlyArray, + selections: ReadonlyArray | null | undefined, +): ReadonlyArray<{ configId: string; value: string }> { + const updates: Array<{ configId: string; value: string }> = []; + + const thinkingOption = configOptions.find( + (option) => + option.id === "thinking" || + option.category === "thought_level" || + option.name?.trim().toLowerCase() === "thinking", + ); + const requestedThinking = + getProviderOptionStringSelectionValue(selections, "reasoning") ?? + getProviderOptionStringSelectionValue(selections, "thinking"); + if (thinkingOption && requestedThinking) { + const allowed = collectSessionConfigOptionValues(thinkingOption); + if (allowed.includes(requestedThinking)) { + updates.push({ configId: thinkingOption.id, value: requestedThinking }); + } + } + + const modeOption = configOptions.find( + (option) => option.id === "mode" || option.category === "mode", + ); + const requestedMode = getProviderOptionStringSelectionValue(selections, "mode"); + if (modeOption && requestedMode) { + const allowed = collectSessionConfigOptionValues(modeOption); + if (allowed.includes(requestedMode)) { + updates.push({ configId: modeOption.id, value: requestedMode }); + } + } + + return updates; +} + +/** + * Apply model + optional thinking/mode selections to a live Kimi ACP session. + * + * When `selections` is omitted, only `session/set_model` is considered (text + * generation / simple probes). Full turn starts pass selections so Thinking + * effort is wired through `session/set_config_option`. + */ +export function applyKimiAcpModelSelection(input: { + readonly runtime: { + readonly setSessionModel: (modelId: string) => Effect.Effect; + readonly getConfigOptions?: Effect.Effect>; + readonly setConfigOption?: ( + configId: string, + value: string | boolean, + ) => Effect.Effect; + }; + readonly currentModelId: string | undefined; + readonly requestedModelId: string | undefined; + readonly selections?: ReadonlyArray | null | undefined; + readonly mapError: (cause: EffectAcpErrors.AcpError) => E; +}): Effect.Effect { + return Effect.gen(function* () { + const requested = input.requestedModelId + ? resolveKimiAcpBaseModelId(input.requestedModelId) + : undefined; + const current = input.currentModelId + ? resolveKimiAcpBaseModelId(input.currentModelId) + : undefined; + const shouldSwitchModel = requested !== undefined && requested !== current; + + if (shouldSwitchModel) { + yield* input.runtime.setSessionModel(requested).pipe(Effect.mapError(input.mapError)); + } + + if ( + input.selections && + input.selections.length > 0 && + input.runtime.getConfigOptions && + input.runtime.setConfigOption + ) { + const options = yield* input.runtime.getConfigOptions; + for (const update of resolveKimiConfigUpdates(options, input.selections)) { + yield* input.runtime + .setConfigOption(update.configId, update.value) + .pipe(Effect.mapError(input.mapError)); + } + } + + return shouldSwitchModel ? requested : current; + }); +} diff --git a/apps/server/src/provider/builtInDrivers.ts b/apps/server/src/provider/builtInDrivers.ts index 791a96e1da3..622b48944bd 100644 --- a/apps/server/src/provider/builtInDrivers.ts +++ b/apps/server/src/provider/builtInDrivers.ts @@ -24,6 +24,7 @@ import { ClaudeDriver, type ClaudeDriverEnv } from "./Drivers/ClaudeDriver.ts"; import { CodexDriver, type CodexDriverEnv } from "./Drivers/CodexDriver.ts"; import { CursorDriver, type CursorDriverEnv } from "./Drivers/CursorDriver.ts"; import { GrokDriver, type GrokDriverEnv } from "./Drivers/GrokDriver.ts"; +import { KimiDriver, type KimiDriverEnv } from "./Drivers/KimiDriver.ts"; import { OpenCodeDriver, type OpenCodeDriverEnv } from "./Drivers/OpenCodeDriver.ts"; import type { AnyProviderDriver } from "./ProviderDriver.ts"; @@ -37,6 +38,7 @@ export type BuiltInDriversEnv = | CodexDriverEnv | CursorDriverEnv | GrokDriverEnv + | KimiDriverEnv | OpenCodeDriverEnv; /** @@ -49,5 +51,6 @@ export const BUILT_IN_DRIVERS: ReadonlyArray): string { + const binDir = NodePath.join(dir, "bin"); + NodeFS.mkdirSync(binDir, { recursive: true }); + const jsPath = NodePath.join(binDir, "kimi.mjs"); + const envJson = JSON.stringify(env); + NodeFS.writeFileSync( + jsPath, + [ + "import { spawnSync } from 'node:child_process';", + `const env = ${envJson};`, + "const args = process.argv.slice(2);", + 'if (args[0] !== "acp") {', + " process.stderr.write(`unexpected args: ${args.join(' ')}\\n`);", + " process.exit(11);", + "}", + `const result = spawnSync(${JSON.stringify(process.execPath)}, [${JSON.stringify(mockAgentPath)}], {`, + " stdio: 'inherit',", + " env: { ...process.env, ...env },", + "});", + "process.exit(result.status ?? 1);", + "", + ].join("\n"), + "utf8", + ); + + if (isWin) { + const cmdPath = NodePath.join(binDir, "kimi.cmd"); + NodeFS.writeFileSync( + cmdPath, + ["@echo off", `node "${jsPath.replaceAll("/", "\\")}" %*`, ""].join("\r\n"), + "utf8", + ); + return cmdPath; + } + + const shPath = NodePath.join(binDir, "kimi"); + NodeFS.writeFileSync( + shPath, + [ + "#!/bin/sh", + `exec ${JSON.stringify(process.execPath)} ${JSON.stringify(jsPath)} "$@"`, + "", + ].join("\n"), + "utf8", + ); + NodeFS.chmodSync(shPath, 0o755); + return shPath; +} + +function withFakeAcpKimi( + env: Record, + effectFn: (textGeneration: TextGeneration.TextGeneration["Service"]) => Effect.Effect, +) { + return Effect.gen(function* () { + const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3code-kimi-text-acp-")); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + NodeFS.rmSync(tempDir, { recursive: true, force: true }); + }), + ); + const binaryPath = makeAcpKimiWrapper(tempDir, env); + const config = decodeKimiSettings({ binaryPath }); + const textGeneration = yield* makeKimiTextGeneration(config); + return yield* effectFn(textGeneration); + }).pipe(Effect.scoped); +} + +function readJsonRpcRequests( + filePath: string, +): ReadonlyArray<{ readonly method?: string; readonly params?: Record }> { + return NodeFS.readFileSync(filePath, "utf8") + .trim() + .split("\n") + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as { method?: string; params?: Record }); +} + +it.layer(KimiTextGenerationTestLayer)("KimiTextGeneration", (it) => { + it.effect("uses ACP with disabled tool capabilities and forwards the requested model id", () => { + const requestLogDir = NodeFS.mkdtempSync( + NodePath.join(NodeOS.tmpdir(), "t3code-kimi-text-log-"), + ); + const requestLogPath = NodePath.join(requestLogDir, "requests.ndjson"); + + return withFakeAcpKimi( + { + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + T3_ACP_PROMPT_RESPONSE_TEXT: JSON.stringify({ + subject: "Add Kimi provider", + body: "Wire up the ACP runtime and headless text generation path.", + }), + }, + (textGeneration) => + Effect.gen(function* () { + // Mock agent only knows grok-build / grok-mock-alt model ids. + const generated = yield* textGeneration.generateCommitMessage({ + cwd: process.cwd(), + branch: "feature/kimi", + stagedSummary: "M apps/server/src/provider/Drivers/KimiDriver.ts", + stagedPatch: "diff --git a/.../KimiDriver.ts b/.../KimiDriver.ts", + modelSelection: createModelSelection(ProviderInstanceId.make("kimi"), "grok-mock-alt"), + }); + + expect(generated.subject).toBe("Add Kimi provider"); + expect(generated.body).toBe("Wire up the ACP runtime and headless text generation path."); + + const requests = readJsonRpcRequests(requestLogPath); + expect( + requests.find((request) => request.method === "initialize")?.params?.clientCapabilities, + ).toMatchObject({ + fs: { readTextFile: false, writeTextFile: false }, + terminal: false, + }); + expect( + requests.some( + (request) => + request.method === "session/set_model" && + request.params?.modelId === "grok-mock-alt", + ), + ).toBe(true); + expect( + requests.some( + (request) => + request.method === "authenticate" && request.params?.methodId === "login", + ), + ).toBe(true); + }), + ); + }); + + it.effect("extracts the JSON object when Kimi wraps it in conversational text", () => + withFakeAcpKimi( + { + T3_ACP_PROMPT_RESPONSE_TEXT: + "Sure! Here's a thread title:\n\n" + + JSON.stringify({ title: "Investigate failing CI" }) + + "\n\nLet me know if you need anything else.", + }, + (textGeneration) => + Effect.gen(function* () { + const generated = yield* textGeneration.generateThreadTitle({ + cwd: process.cwd(), + message: "the lint job is red", + modelSelection: createModelSelection(ProviderInstanceId.make("kimi"), "grok-build"), + }); + expect(generated.title).toBe("Investigate failing CI"); + }), + ), + ); + + it.effect("surfaces ACP request failures as text generation errors", () => + withFakeAcpKimi( + { + T3_ACP_PROMPT_RESPONSE_TEXT: JSON.stringify({ branch: "unreachable" }), + }, + (textGeneration) => + Effect.gen(function* () { + const error = yield* Effect.flip( + textGeneration.generateBranchName({ + cwd: process.cwd(), + message: "wire up kimi", + modelSelection: createModelSelection( + ProviderInstanceId.make("kimi"), + "missing-kimi-model", + ), + }), + ); + expect(error._tag).toBe("TextGenerationError"); + expect(error.detail).toContain("Kimi ACP base model"); + }), + ), + ); + + it.effect("fails with TextGenerationError when output is empty", () => + withFakeAcpKimi( + { + T3_ACP_PROMPT_RESPONSE_TEXT: " \n ", + }, + (textGeneration) => + Effect.gen(function* () { + const error = yield* Effect.flip( + textGeneration.generateThreadTitle({ + cwd: process.cwd(), + message: "anything", + modelSelection: createModelSelection(ProviderInstanceId.make("kimi"), "grok-build"), + }), + ); + expect(error._tag).toBe("TextGenerationError"); + expect(error.detail).toMatch(/empty/i); + }), + ), + ); +}); diff --git a/apps/server/src/textGeneration/KimiTextGeneration.ts b/apps/server/src/textGeneration/KimiTextGeneration.ts new file mode 100644 index 00000000000..771c5c0d1ab --- /dev/null +++ b/apps/server/src/textGeneration/KimiTextGeneration.ts @@ -0,0 +1,260 @@ +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import { ChildProcessSpawner } from "effect/unstable/process"; +import type * as EffectAcpErrors from "effect-acp/errors"; + +import { type KimiSettings, type ModelSelection } from "@t3tools/contracts"; +import { sanitizeBranchFragment, sanitizeFeatureBranchName } from "@t3tools/shared/git"; +import { extractJsonObject } from "@t3tools/shared/schemaJson"; + +import { TextGenerationError } from "@t3tools/contracts"; +import * as TextGeneration from "./TextGeneration.ts"; +import { + buildBranchNamePrompt, + buildCommitMessagePrompt, + buildPrContentPrompt, + buildThreadTitlePrompt, +} from "./TextGenerationPrompts.ts"; +import { + sanitizeCommitSubject, + sanitizePrTitle, + sanitizeThreadTitle, +} from "./TextGenerationUtils.ts"; +import { + applyKimiAcpModelSelection, + currentKimiModelIdFromSessionSetup, + makeKimiAcpRuntime, + resolveKimiAcpBaseModelId, +} from "../provider/acp/KimiAcpSupport.ts"; + +const KIMI_TIMEOUT_MS = 180_000; + +const isTextGenerationError = Schema.is(TextGenerationError); + +export const makeKimiTextGeneration = Effect.fn("makeKimiTextGeneration")(function* ( + kimiSettings: KimiSettings, + environment: NodeJS.ProcessEnv = process.env, +) { + const crypto = yield* Crypto.Crypto; + const commandSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + + const runKimiJson = ({ + operation, + cwd, + prompt, + outputSchemaJson, + modelSelection, + }: { + operation: + | "generateCommitMessage" + | "generatePrContent" + | "generateBranchName" + | "generateThreadTitle"; + cwd: string; + prompt: string; + outputSchemaJson: S; + modelSelection: ModelSelection; + }): Effect.Effect => + Effect.gen(function* () { + const resolvedModel = resolveKimiAcpBaseModelId(modelSelection.model); + const outputRef = yield* Ref.make(""); + const runtime = yield* makeKimiAcpRuntime({ + kimiSettings, + environment, + childProcessSpawner: commandSpawner, + cwd, + clientInfo: { name: "t3-code-git-text", version: "0.0.0" }, + }).pipe(Effect.provideService(Crypto.Crypto, crypto)); + + yield* runtime.handleSessionUpdate((notification) => { + const update = notification.update; + if (update.sessionUpdate !== "agent_message_chunk") { + return Effect.void; + } + const content = update.content; + if (content.type !== "text") { + return Effect.void; + } + return Ref.update(outputRef, (current) => current + content.text); + }); + + const promptResult = yield* Effect.gen(function* () { + const started = yield* runtime.start(); + yield* applyKimiAcpModelSelection({ + runtime, + currentModelId: currentKimiModelIdFromSessionSetup(started.sessionSetupResult), + requestedModelId: resolvedModel, + mapError: (cause) => + new TextGenerationError({ + operation, + detail: "Failed to set Kimi ACP base model for text generation.", + cause, + }), + }); + + return yield* runtime.prompt({ + prompt: [{ type: "text", text: prompt }], + }); + }).pipe( + Effect.timeoutOption(KIMI_TIMEOUT_MS), + Effect.flatMap( + Option.match({ + onNone: () => + Effect.fail( + new TextGenerationError({ operation, detail: "Kimi ACP request timed out." }), + ), + onSome: (value) => Effect.succeed(value), + }), + ), + Effect.mapError((cause: EffectAcpErrors.AcpError | TextGenerationError) => + isTextGenerationError(cause) + ? cause + : new TextGenerationError({ + operation, + detail: "Kimi ACP request failed.", + cause, + }), + ), + ); + + const trimmed = (yield* Ref.get(outputRef)).trim(); + if (!trimmed) { + return yield* new TextGenerationError({ + operation, + detail: + promptResult.stopReason === "cancelled" + ? "Kimi ACP request was cancelled." + : "Kimi Agent returned empty output.", + }); + } + + const decodeOutput = Schema.decodeEffect(Schema.fromJsonString(outputSchemaJson)); + return yield* decodeOutput(extractJsonObject(trimmed)).pipe( + Effect.catchTags({ + SchemaError: (cause) => + Effect.fail( + new TextGenerationError({ + operation, + detail: "Kimi Agent returned invalid structured output.", + cause, + }), + ), + }), + ); + }).pipe( + Effect.mapError((cause) => + isTextGenerationError(cause) + ? cause + : new TextGenerationError({ + operation, + detail: "Kimi ACP text generation failed.", + cause, + }), + ), + Effect.scoped, + ); + + const generateCommitMessage: TextGeneration.TextGeneration["Service"]["generateCommitMessage"] = + Effect.fn("KimiTextGeneration.generateCommitMessage")(function* (input) { + const { prompt, outputSchema } = buildCommitMessagePrompt({ + branch: input.branch, + stagedSummary: input.stagedSummary, + stagedPatch: input.stagedPatch, + includeBranch: input.includeBranch === true, + policy: input.policy, + }); + + const generated = yield* runKimiJson({ + operation: "generateCommitMessage", + cwd: input.cwd, + prompt, + outputSchemaJson: outputSchema, + modelSelection: input.modelSelection, + }); + + return { + subject: sanitizeCommitSubject(generated.subject), + body: generated.body.trim(), + ...("branch" in generated && typeof generated.branch === "string" + ? { branch: sanitizeFeatureBranchName(generated.branch) } + : {}), + }; + }); + + const generatePrContent: TextGeneration.TextGeneration["Service"]["generatePrContent"] = + Effect.fn("KimiTextGeneration.generatePrContent")(function* (input) { + const { prompt, outputSchema } = buildPrContentPrompt({ + baseBranch: input.baseBranch, + headBranch: input.headBranch, + commitSummary: input.commitSummary, + diffSummary: input.diffSummary, + diffPatch: input.diffPatch, + policy: input.policy, + changeRequestTemplate: input.changeRequestTemplate, + }); + + const generated = yield* runKimiJson({ + operation: "generatePrContent", + cwd: input.cwd, + prompt, + outputSchemaJson: outputSchema, + modelSelection: input.modelSelection, + }); + + return { + title: sanitizePrTitle(generated.title), + body: generated.body.trim(), + }; + }); + + const generateBranchName: TextGeneration.TextGeneration["Service"]["generateBranchName"] = + Effect.fn("KimiTextGeneration.generateBranchName")(function* (input) { + const { prompt, outputSchema } = buildBranchNamePrompt({ + message: input.message, + attachments: input.attachments, + }); + + const generated = yield* runKimiJson({ + operation: "generateBranchName", + cwd: input.cwd, + prompt, + outputSchemaJson: outputSchema, + modelSelection: input.modelSelection, + }); + + return { + branch: sanitizeBranchFragment(generated.branch), + }; + }); + + const generateThreadTitle: TextGeneration.TextGeneration["Service"]["generateThreadTitle"] = + Effect.fn("KimiTextGeneration.generateThreadTitle")(function* (input) { + const { prompt, outputSchema } = buildThreadTitlePrompt({ + message: input.message, + previousTitle: input.previousTitle, + attachments: input.attachments, + }); + + const generated = yield* runKimiJson({ + operation: "generateThreadTitle", + cwd: input.cwd, + prompt, + outputSchemaJson: outputSchema, + modelSelection: input.modelSelection, + }); + + return { + title: sanitizeThreadTitle(generated.title), + } satisfies TextGeneration.ThreadTitleGenerationResult; + }); + + return { + generateCommitMessage, + generatePrContent, + generateBranchName, + generateThreadTitle, + } satisfies TextGeneration.TextGeneration["Service"]; +}); diff --git a/apps/server/src/textGeneration/TextGeneration.ts b/apps/server/src/textGeneration/TextGeneration.ts index 66b7ccd465f..35c7f03180c 100644 --- a/apps/server/src/textGeneration/TextGeneration.ts +++ b/apps/server/src/textGeneration/TextGeneration.ts @@ -8,7 +8,13 @@ import * as ProviderInstanceRegistry from "../provider/Services/ProviderInstance import type { ProviderInstance } from "../provider/ProviderDriver.ts"; import type { TextGenerationPolicy } from "./TextGenerationPolicy.ts"; -export type TextGenerationProvider = "codex" | "claudeAgent" | "cursor" | "grok" | "opencode"; +export type TextGenerationProvider = + | "codex" + | "claudeAgent" + | "cursor" + | "grok" + | "kimi" + | "opencode"; export interface CommitMessageGenerationInput { cwd: string; diff --git a/apps/web/src/components/Icons.tsx b/apps/web/src/components/Icons.tsx index 8ea38c51958..36acc033c5a 100644 --- a/apps/web/src/components/Icons.tsx +++ b/apps/web/src/components/Icons.tsx @@ -211,6 +211,18 @@ export const GrokIcon: Icon = ({ className, ...props }) => ( ); +/** Moonshot Kimi mark — simplified “K” monogram for provider picker chips. */ +export const KimiIcon: Icon = ({ className, ...props }) => ( + + + +); + export const TraeIcon: Icon = (props) => ( {/* Back rectangle: left strip + bottom strip drawn separately — empty bottom-left corner is the gap between them */} diff --git a/apps/web/src/components/chat/providerIconUtils.ts b/apps/web/src/components/chat/providerIconUtils.ts index 842c616fe1f..504c80be3cc 100644 --- a/apps/web/src/components/chat/providerIconUtils.ts +++ b/apps/web/src/components/chat/providerIconUtils.ts @@ -1,5 +1,5 @@ import { ProviderDriverKind } from "@t3tools/contracts"; -import { ClaudeAI, CursorIcon, GrokIcon, Icon, OpenAI, OpenCodeIcon } from "../Icons"; +import { ClaudeAI, CursorIcon, GrokIcon, Icon, KimiIcon, OpenAI, OpenCodeIcon } from "../Icons"; import { PROVIDER_OPTIONS } from "../../session-logic"; export const PROVIDER_ICON_BY_PROVIDER: Partial> = { @@ -8,6 +8,7 @@ export const PROVIDER_ICON_BY_PROVIDER: Partial [ProviderDriverKind.make("opencode")]: OpenCodeIcon, [ProviderDriverKind.make("cursor")]: CursorIcon, [ProviderDriverKind.make("grok")]: GrokIcon, + [ProviderDriverKind.make("kimi")]: KimiIcon, }; function isAvailableProviderOption(option: (typeof PROVIDER_OPTIONS)[number]): option is { diff --git a/apps/web/src/components/settings/providerDriverMeta.ts b/apps/web/src/components/settings/providerDriverMeta.ts index bfee6a8d680..10cbecb3dcd 100644 --- a/apps/web/src/components/settings/providerDriverMeta.ts +++ b/apps/web/src/components/settings/providerDriverMeta.ts @@ -3,11 +3,20 @@ import { CodexSettings, CursorSettings, GrokSettings, + KimiSettings, OpenCodeSettings, ProviderDriverKind, } from "@t3tools/contracts"; import type * as Schema from "effect/Schema"; -import { ClaudeAI, CursorIcon, GrokIcon, type Icon, OpenAI, OpenCodeIcon } from "../Icons"; +import { + ClaudeAI, + CursorIcon, + GrokIcon, + type Icon, + KimiIcon, + OpenAI, + OpenCodeIcon, +} from "../Icons"; type ProviderSettingsSchema = { readonly fields: Readonly>; @@ -61,6 +70,13 @@ export const PROVIDER_CLIENT_DEFINITIONS: readonly ProviderClientDefinition[] = badgeLabel: "Early Access", settingsSchema: GrokSettings, }, + { + value: ProviderDriverKind.make("kimi"), + label: "Kimi", + icon: KimiIcon, + badgeLabel: "Early Access", + settingsSchema: KimiSettings, + }, { value: ProviderDriverKind.make("opencode"), label: "OpenCode", diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 5d5051f748e..7acc41c3865 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -51,6 +51,12 @@ export const PROVIDER_OPTIONS: Array<{ available: true, pickerSidebarBadge: "new", }, + { + value: ProviderDriverKind.make("kimi"), + label: "Kimi", + available: true, + pickerSidebarBadge: "new", + }, ]; export type WorkLogToolLifecycleStatus = diff --git a/docs/internals/glossary.md b/docs/internals/glossary.md index da16f74d339..62c00933caf 100644 --- a/docs/internals/glossary.md +++ b/docs/internals/glossary.md @@ -94,7 +94,7 @@ The live backend agent implementation and its event stream. The main service is #### Provider -The backend agent runtime that actually performs work. Five drivers ship built in: Codex, Claude, Cursor, Grok, and OpenCode. See [ProviderService.ts][14], [ProviderAdapter.ts][15], and [CodexAdapter.ts][17] as a representative adapter. +The backend agent runtime that actually performs work. Six drivers ship built in: Codex, Claude, Cursor, Grok, Kimi, and OpenCode. See [ProviderService.ts][14], [ProviderAdapter.ts][15], and [CodexAdapter.ts][17] as a representative adapter. #### Session diff --git a/docs/internals/overview.md b/docs/internals/overview.md index b9454f7b58d..5ea20098690 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -23,7 +23,7 @@ there, never in the client. └──────────────────┬─────────────────────────────┘ │ per-driver transport ┌──────────────────▼─────────────────────────────┐ -│ Agent CLIs: Codex, Claude, Cursor, Grok, │ +│ Agent CLIs: Codex, Claude, Cursor, Grok, Kimi, │ │ OpenCode │ └────────────────────────────────────────────────┘ ``` @@ -106,8 +106,8 @@ build production behavior on receipts. ## Provider drivers -Five drivers ship built in, registered in [`builtInDrivers.ts`][drivers] as `BUILT_IN_DRIVERS`: -Codex, Claude, Cursor, Grok, and OpenCode. A driver declares its kind and config schema and creates a +Six drivers ship built in, registered in [`builtInDrivers.ts`][drivers] as `BUILT_IN_DRIVERS`: +Codex, Claude, Cursor, Grok, Kimi, and OpenCode. A driver declares its kind and config schema and creates a scoped adapter; `ProviderInstanceRegistry` owns live instances and `ProviderAdapterRegistry` resolves an instance to its adapter, so `ProviderService` routes session and turn operations without knowing which agent is behind them. See [providers.md](./providers.md). diff --git a/docs/internals/providers.md b/docs/internals/providers.md index a309d70f03d..804a0b9e842 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -7,7 +7,7 @@ orchestration layer does not know which one is behind a thread. ## Built-in drivers -[`builtInDrivers.ts`][drivers] exports `BUILT_IN_DRIVERS` with five entries: +[`builtInDrivers.ts`][drivers] exports `BUILT_IN_DRIVERS` with six entries: | Driver kind | Driver source | | ------------- | --------------------------------------- | @@ -15,6 +15,7 @@ orchestration layer does not know which one is behind a thread. | `claudeAgent` | [`Drivers/ClaudeDriver.ts`][claude] | | `cursor` | [`Drivers/CursorDriver.ts`][cursor] | | `grok` | [`Drivers/GrokDriver.ts`][grok] | +| `kimi` | [`Drivers/KimiDriver.ts`][kimi] | | `opencode` | [`Drivers/OpenCodeDriver.ts`][opencode] | Each driver declares its `driverKind`, a `configSchema`, and a `create` function that builds an @@ -80,6 +81,7 @@ when a request opens (approval) or user input is requested, via [claude]: ../../apps/server/src/provider/Drivers/ClaudeDriver.ts [cursor]: ../../apps/server/src/provider/Drivers/CursorDriver.ts [grok]: ../../apps/server/src/provider/Drivers/GrokDriver.ts +[kimi]: ../../apps/server/src/provider/Drivers/KimiDriver.ts [opencode]: ../../apps/server/src/provider/Drivers/OpenCodeDriver.ts [adapter]: ../../apps/server/src/provider/Services/ProviderAdapter.ts [instances]: ../../apps/server/src/provider/Services/ProviderInstanceRegistry.ts diff --git a/docs/user/install.md b/docs/user/install.md index fe0b418ca1e..ada8305fdb1 100644 --- a/docs/user/install.md +++ b/docs/user/install.md @@ -46,13 +46,14 @@ yay -S t3code-bin T3 Code drives provider CLIs; it does not ship them. Install the CLI for each provider you want to use, then authenticate it. -| Provider | CLI | Default binary | Log in with | -| ---------- | ----------------------------------------------------- | -------------- | --------------------- | -| Codex | [Codex CLI](https://developers.openai.com/codex/cli) | `codex` | `codex login` | -| Claude | [Claude Code](https://claude.com/product/claude-code) | `claude` | `claude auth login` | -| Cursor | [Cursor CLI](https://cursor.com/cli) | `cursor-agent` | `agent login` | -| Grok Build | [Grok Build CLI](https://x.ai/cli) | `grok` | `grok login` | -| OpenCode | [OpenCode](https://opencode.ai) | `opencode` | `opencode auth login` | +| Provider | CLI | Default binary | Log in with | +| ---------- | ----------------------------------------------------------- | -------------- | --------------------- | +| Codex | [Codex CLI](https://developers.openai.com/codex/cli) | `codex` | `codex login` | +| Claude | [Claude Code](https://claude.com/product/claude-code) | `claude` | `claude auth login` | +| Cursor | [Cursor CLI](https://cursor.com/cli) | `cursor-agent` | `agent login` | +| Grok Build | [Grok Build CLI](https://x.ai/cli) | `grok` | `grok login` | +| Kimi | [Kimi Code CLI](https://moonshotai.github.io/kimi-code/en/) | `kimi` | `kimi` then `/login` | +| OpenCode | [OpenCode](https://opencode.ai) | `opencode` | `opencode auth login` | Cursor is the one to watch: install Cursor CLI, which provides the `cursor-agent` binary that T3 Code looks for, but authenticate with `agent login`, not `cursor-agent login`. diff --git a/packages/contracts/src/model.ts b/packages/contracts/src/model.ts index 46067917f13..35c22d41e0b 100644 --- a/packages/contracts/src/model.ts +++ b/packages/contracts/src/model.ts @@ -131,6 +131,7 @@ const CODEX_DRIVER_KIND = ProviderDriverKind.make("codex"); const CLAUDE_DRIVER_KIND = ProviderDriverKind.make("claudeAgent"); const CURSOR_DRIVER_KIND = ProviderDriverKind.make("cursor"); const GROK_DRIVER_KIND = ProviderDriverKind.make("grok"); +const KIMI_DRIVER_KIND = ProviderDriverKind.make("kimi"); const OPENCODE_DRIVER_KIND = ProviderDriverKind.make("opencode"); export const DEFAULT_MODEL = "gpt-5.6-sol"; @@ -151,6 +152,7 @@ export const DEFAULT_MODEL_BY_PROVIDER: Partial> [CLAUDE_DRIVER_KIND]: "Claude", [CURSOR_DRIVER_KIND]: "Cursor", [GROK_DRIVER_KIND]: "Grok", + [KIMI_DRIVER_KIND]: "Kimi", [OPENCODE_DRIVER_KIND]: "OpenCode", }; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 7edda2e52e5..585b82292bc 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -351,6 +351,30 @@ export const GrokSettings = makeProviderSettingsSchema( ); export type GrokSettings = typeof GrokSettings.Type; +export const KimiSettings = makeProviderSettingsSchema( + { + enabled: Schema.Boolean.pipe( + Schema.withDecodingDefault(Effect.succeed(true)), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + binaryPath: makeBinaryPathSetting("kimi").pipe( + Schema.annotateKey({ + title: "Binary path", + description: "Path to the Kimi Code CLI binary.", + providerSettingsForm: { placeholder: "kimi", clearWhenEmpty: "omit" }, + }), + ), + customModels: Schema.Array(Schema.String).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + }, + { + order: ["binaryPath"], + }, +); +export type KimiSettings = typeof KimiSettings.Type; + export const OpenCodeSettings = makeProviderSettingsSchema( { enabled: Schema.Boolean.pipe( @@ -519,6 +543,7 @@ export const ServerSettings = Schema.Struct({ claudeAgent: ClaudeSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), cursor: CursorSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), grok: GrokSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), + kimi: KimiSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), opencode: OpenCodeSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), }).pipe(Schema.withDecodingDefault(Effect.succeed({}))), // New driver-agnostic instance map. Keyed by `ProviderInstanceId`; values @@ -615,6 +640,12 @@ const GrokSettingsPatch = Schema.Struct({ customModels: Schema.optionalKey(Schema.Array(Schema.String)), }); +const KimiSettingsPatch = Schema.Struct({ + enabled: Schema.optionalKey(Schema.Boolean), + binaryPath: Schema.optionalKey(TrimmedString), + customModels: Schema.optionalKey(Schema.Array(Schema.String)), +}); + const OpenCodeSettingsPatch = Schema.Struct({ enabled: Schema.optionalKey(Schema.Boolean), binaryPath: Schema.optionalKey(TrimmedString), @@ -662,6 +693,7 @@ export const ServerSettingsPatch = Schema.Struct({ claudeAgent: Schema.optionalKey(ClaudeSettingsPatch), cursor: Schema.optionalKey(CursorSettingsPatch), grok: Schema.optionalKey(GrokSettingsPatch), + kimi: Schema.optionalKey(KimiSettingsPatch), opencode: Schema.optionalKey(OpenCodeSettingsPatch), }), ), From c8a863902aef334a800f418c4d37925d94f3d343 Mon Sep 17 00:00:00 2001 From: maiconfontana Date: Sun, 2 Aug 2026 17:54:44 -0300 Subject: [PATCH 2/3] fix(server): address Kimi PR review (reasoning buffer, models, policy) Cap and clear per-turn reasoning buffers, only open Thinking tasks on visible text, flatten grouped ACP model select entries, forward text-generation policy for branch/title, and fix the built-in driver count in overview docs. Made with Grok Build. --- .../Layers/ProviderRuntimeIngestion.ts | 50 ++++++++++++++----- .../src/provider/Layers/KimiProvider.ts | 7 ++- .../src/textGeneration/KimiTextGeneration.ts | 2 + docs/internals/overview.md | 2 +- 4 files changed, 46 insertions(+), 15 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 27dbe5a8224..50e5ce3535c 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -94,6 +94,8 @@ const BUFFERED_PROPOSED_PLAN_BY_ID_TTL = Duration.minutes(120); const TASK_DESCRIPTION_BY_TASK_CACHE_CAPACITY = 10_000; const TASK_DESCRIPTION_BY_TASK_TTL = Duration.minutes(120); const MAX_BUFFERED_ASSISTANT_CHARS = 24_000; +/** Cap for per-turn reasoning/thought buffers (same order as assistant text). */ +const MAX_BUFFERED_REASONING_CHARS = 24_000; const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0"; type TurnStartRequestedDomainEvent = Extract< @@ -766,11 +768,23 @@ const make = Effect.gen(function* () { Option.getOrElse(option, () => ({ text: "", started: false, lastPublishedLength: 0 })), ), ); - const nextText = `${existing.text}${input.delta}`; + // Cap buffer growth so long thought streams cannot exhaust server memory. + let nextText = `${existing.text}${input.delta}`; + if (nextText.length > MAX_BUFFERED_REASONING_CHARS) { + nextText = nextText.slice(0, MAX_BUFFERED_REASONING_CHARS); + } const taskId = reasoningTaskIdForTurn(input.turnId); const activities: Array = []; + const hasVisibleText = nextText.trim().length > 0; + // Only open a Thinking task once there is non-whitespace content so a + // whitespace-only stream never leaves a dangling "Thinking" entry. + const shouldStart = !existing.started && hasVisibleText; + const shouldPublishProgress = + hasVisibleText && + (shouldStart || + nextText.length - existing.lastPublishedLength >= REASONING_PROGRESS_PUBLISH_CHARS); - if (!existing.started) { + if (shouldStart) { activities.push({ id: EventId.make(`${input.event.eventId}:reasoning-started`), createdAt: input.createdAt, @@ -786,10 +800,7 @@ const make = Effect.gen(function* () { }); } - const shouldPublishProgress = - !existing.started || - nextText.length - existing.lastPublishedLength >= REASONING_PROGRESS_PUBLISH_CHARS; - if (shouldPublishProgress && nextText.trim().length > 0) { + if (shouldPublishProgress) { activities.push({ id: EventId.make(`${input.event.eventId}:reasoning-progress`), createdAt: input.createdAt, @@ -806,12 +817,16 @@ const make = Effect.gen(function* () { }); } + const started = existing.started || shouldStart; yield* Cache.set(bufferedReasoningByTurnKey, key, { text: nextText, - started: true, + started, + // Only advance when progress was actually published (not on whitespace skips). lastPublishedLength: shouldPublishProgress ? nextText.length : existing.lastPublishedLength, }); - yield* rememberTaskDescription(input.threadId, taskId, "Thinking"); + if (started) { + yield* rememberTaskDescription(input.threadId, taskId, "Thinking"); + } yield* Effect.forEach( activities, @@ -843,14 +858,16 @@ const make = Effect.gen(function* () { Effect.map(Option.getOrUndefined), ); if (!existing?.started) { + // Drop any whitespace-only buffer that never opened a UI task. + if (existing) { + yield* Cache.invalidate(bufferedReasoningByTurnKey, key); + } return; } const taskId = reasoningTaskIdForTurn(input.turnId); const detail = truncateDetail(existing.text.trim()); yield* Cache.invalidate(bufferedReasoningByTurnKey, key); - if (detail.length === 0) { - return; - } + // Always close the task if we opened it, even when the visible detail is empty. yield* providerCommandId(input.event, "reasoning-complete").pipe( Effect.flatMap((commandId) => orchestrationEngine.dispatch({ @@ -867,8 +884,8 @@ const make = Effect.gen(function* () { taskId, status: "completed", title: "Thinking", - summary: detail, - detail, + summary: detail.length > 0 ? detail : "Thinking complete", + detail: detail.length > 0 ? detail : "Thinking complete", }, turnId: input.turnId, }, @@ -1313,6 +1330,7 @@ const make = Effect.gen(function* () { const assistantSegmentKeys = Array.from(yield* Cache.keys(assistantSegmentStateByTurnKey)); const proposedPlanKeys = Array.from(yield* Cache.keys(bufferedProposedPlanById)); const taskDescriptionKeys = Array.from(yield* Cache.keys(taskDescriptionByTaskKey)); + const reasoningKeys = Array.from(yield* Cache.keys(bufferedReasoningByTurnKey)); yield* Effect.forEach( turnKeys, (key) => @@ -1354,6 +1372,12 @@ const make = Effect.gen(function* () { key.startsWith(prefix) ? Cache.invalidate(taskDescriptionByTaskKey, key) : Effect.void, { concurrency: 1 }, ).pipe(Effect.asVoid); + yield* Effect.forEach( + reasoningKeys, + (key) => + key.startsWith(prefix) ? Cache.invalidate(bufferedReasoningByTurnKey, key) : Effect.void, + { concurrency: 1 }, + ).pipe(Effect.asVoid); }); const getSourceProposedPlanReferenceForPendingTurnStart = Effect.fn( diff --git a/apps/server/src/provider/Layers/KimiProvider.ts b/apps/server/src/provider/Layers/KimiProvider.ts index 3be83fab685..52531b7a594 100644 --- a/apps/server/src/provider/Layers/KimiProvider.ts +++ b/apps/server/src/provider/Layers/KimiProvider.ts @@ -203,7 +203,12 @@ function buildKimiDiscoveredModelsFromConfigOptions( } const capabilities = buildKimiThinkingCapabilitiesFromConfigOptions(configOptions); const seen = new Set(); - return modelOption.options.flatMap((entry): Array => { + // ACP select options may be flat `{ value }` entries or grouped + // `{ options: [...] }` entries — flatten groups the same way thinking does. + const flatEntries = modelOption.options.flatMap((entry) => + "options" in entry && Array.isArray(entry.options) ? entry.options : [entry], + ); + return flatEntries.flatMap((entry): Array => { if (!("value" in entry) || typeof entry.value !== "string") { return []; } diff --git a/apps/server/src/textGeneration/KimiTextGeneration.ts b/apps/server/src/textGeneration/KimiTextGeneration.ts index 771c5c0d1ab..85bedb26074 100644 --- a/apps/server/src/textGeneration/KimiTextGeneration.ts +++ b/apps/server/src/textGeneration/KimiTextGeneration.ts @@ -215,6 +215,7 @@ export const makeKimiTextGeneration = Effect.fn("makeKimiTextGeneration")(functi const { prompt, outputSchema } = buildBranchNamePrompt({ message: input.message, attachments: input.attachments, + policy: input.policy, }); const generated = yield* runKimiJson({ @@ -236,6 +237,7 @@ export const makeKimiTextGeneration = Effect.fn("makeKimiTextGeneration")(functi message: input.message, previousTitle: input.previousTitle, attachments: input.attachments, + policy: input.policy, }); const generated = yield* runKimiJson({ diff --git a/docs/internals/overview.md b/docs/internals/overview.md index 5ea20098690..e9826dd26a2 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -18,7 +18,7 @@ there, never in the client. ┌──────────────────▼─────────────────────────────┐ │ apps/server │ │ orchestration engine (event-sourced) │ -│ provider driver registry (5 built-in drivers) │ +│ provider driver registry (6 built-in drivers) │ │ checkpointing, VCS, terminals, filesystem │ └──────────────────┬─────────────────────────────┘ │ per-driver transport From 7eb874d5620a5937f084e366d2ad27f6d849423e Mon Sep 17 00:00:00 2001 From: maiconfontana Date: Sun, 2 Aug 2026 18:04:36 -0300 Subject: [PATCH 3/3] fix(server): finish Kimi PR review fixes and test mock models Forward modelSelection options into applyKimiAcpModelSelection for headless text gen, type the ACP runtime param from AcpSessionRuntime Service, teach the ACP mock agent kimi-code/* model ids so Kimi text-gen tests match resolveKimiAcpBaseModelId, and drop the incorrect branch/title policy wiring (those service inputs have no policy field). Made with Grok Build. --- apps/server/scripts/acp-mock-agent.ts | 19 ++++++++++++++++--- .../server/src/provider/acp/KimiAcpSupport.ts | 12 ++++-------- .../textGeneration/KimiTextGeneration.test.ts | 4 ++-- .../src/textGeneration/KimiTextGeneration.ts | 3 +-- 4 files changed, 23 insertions(+), 15 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index bc7828dd854..7327521bf37 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -283,13 +283,26 @@ const grokAcpModels: ReadonlyArray = [ { modelId: "grok-mock-alt", name: "Grok Mock Alt" }, ]; +/** Prefixed ids used by Kimi ACP resolution (`resolveKimiAcpBaseModelId`). */ +const kimiAcpModels: ReadonlyArray = [ + { modelId: "kimi-code/kimi-for-coding", name: "Kimi for Coding" }, + { modelId: "kimi-code/kimi-for-coding-highspeed", name: "Kimi for Coding Highspeed" }, + { modelId: "kimi-code/k3", name: "K3" }, + { modelId: "kimi-code/k3-256k", name: "K3 256k" }, + // Allows tests that pass short Grok mock ids to still succeed after Kimi prefixing. + { modelId: "kimi-code/grok-mock-alt", name: "Grok Mock Alt (Kimi-prefixed)" }, + { modelId: "kimi-code/grok-build", name: "Grok Build (Kimi-prefixed)" }, +]; + +const knownAcpModels: ReadonlyArray = [...grokAcpModels, ...kimiAcpModels]; + function modelState(): AcpSchema.SessionModelState { - const modelId = grokAcpModels.some((model) => model.modelId === currentModelId) + const modelId = knownAcpModels.some((model) => model.modelId === currentModelId) ? currentModelId : "grok-build"; return { currentModelId: modelId, - availableModels: grokAcpModels, + availableModels: knownAcpModels, }; } @@ -382,7 +395,7 @@ const program = Effect.gen(function* () { yield* agent.handleSetSessionModel((request) => Effect.gen(function* () { - if (!grokAcpModels.some((model) => model.modelId === request.modelId)) { + if (!knownAcpModels.some((model) => model.modelId === request.modelId)) { return yield* AcpError.AcpRequestError.invalidParams( `Unknown mock model id: ${request.modelId}`, { diff --git a/apps/server/src/provider/acp/KimiAcpSupport.ts b/apps/server/src/provider/acp/KimiAcpSupport.ts index 08a3f37f487..023609feb22 100644 --- a/apps/server/src/provider/acp/KimiAcpSupport.ts +++ b/apps/server/src/provider/acp/KimiAcpSupport.ts @@ -163,14 +163,10 @@ function resolveKimiConfigUpdates( * effort is wired through `session/set_config_option`. */ export function applyKimiAcpModelSelection(input: { - readonly runtime: { - readonly setSessionModel: (modelId: string) => Effect.Effect; - readonly getConfigOptions?: Effect.Effect>; - readonly setConfigOption?: ( - configId: string, - value: string | boolean, - ) => Effect.Effect; - }; + readonly runtime: Pick & + Partial< + Pick + >; readonly currentModelId: string | undefined; readonly requestedModelId: string | undefined; readonly selections?: ReadonlyArray | null | undefined; diff --git a/apps/server/src/textGeneration/KimiTextGeneration.test.ts b/apps/server/src/textGeneration/KimiTextGeneration.test.ts index 6d053fe3684..36954c3cb40 100644 --- a/apps/server/src/textGeneration/KimiTextGeneration.test.ts +++ b/apps/server/src/textGeneration/KimiTextGeneration.test.ts @@ -121,7 +121,7 @@ it.layer(KimiTextGenerationTestLayer)("KimiTextGeneration", (it) => { }, (textGeneration) => Effect.gen(function* () { - // Mock agent only knows grok-build / grok-mock-alt model ids. + // Short ids are resolved to kimi-code/* by resolveKimiAcpBaseModelId. const generated = yield* textGeneration.generateCommitMessage({ cwd: process.cwd(), branch: "feature/kimi", @@ -144,7 +144,7 @@ it.layer(KimiTextGenerationTestLayer)("KimiTextGeneration", (it) => { requests.some( (request) => request.method === "session/set_model" && - request.params?.modelId === "grok-mock-alt", + request.params?.modelId === "kimi-code/grok-mock-alt", ), ).toBe(true); expect( diff --git a/apps/server/src/textGeneration/KimiTextGeneration.ts b/apps/server/src/textGeneration/KimiTextGeneration.ts index 85bedb26074..5eca497a4cd 100644 --- a/apps/server/src/textGeneration/KimiTextGeneration.ts +++ b/apps/server/src/textGeneration/KimiTextGeneration.ts @@ -87,6 +87,7 @@ export const makeKimiTextGeneration = Effect.fn("makeKimiTextGeneration")(functi runtime, currentModelId: currentKimiModelIdFromSessionSetup(started.sessionSetupResult), requestedModelId: resolvedModel, + selections: modelSelection.options, mapError: (cause) => new TextGenerationError({ operation, @@ -215,7 +216,6 @@ export const makeKimiTextGeneration = Effect.fn("makeKimiTextGeneration")(functi const { prompt, outputSchema } = buildBranchNamePrompt({ message: input.message, attachments: input.attachments, - policy: input.policy, }); const generated = yield* runKimiJson({ @@ -237,7 +237,6 @@ export const makeKimiTextGeneration = Effect.fn("makeKimiTextGeneration")(functi message: input.message, previousTitle: input.previousTitle, attachments: input.attachments, - policy: input.policy, }); const generated = yield* runKimiJson({