diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index bc7828dd854..fba52db7ecd 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -39,6 +39,11 @@ const failPrompt = process.env.T3_ACP_FAIL_PROMPT === "1"; const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1"; const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1"; const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT; +// Selects which provider-specific ACP model list this mock advertises. +// Defaults to the Grok list; `kimi` swaps in the Kimi models so the same +// mock agent backs both provider adapters. Additive — leaving it unset +// preserves the original Grok behaviour byte-for-byte. +const modelSet = process.env.T3_ACP_MODEL_SET; const promptDelayMs = Number(process.env.T3_ACP_PROMPT_DELAY_MS ?? "0"); const permissionOptionIds = { allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once", @@ -283,13 +288,21 @@ const grokAcpModels: ReadonlyArray = [ { modelId: "grok-mock-alt", name: "Grok Mock Alt" }, ]; +const kimiAcpModels: ReadonlyArray = [ + { modelId: "kimi-k3", name: "Kimi K3" }, + { modelId: "kimi-k2-thinking", name: "Kimi K2 Thinking" }, +]; + +const acpModels = modelSet === "kimi" ? kimiAcpModels : grokAcpModels; +const defaultAcpModelId = modelSet === "kimi" ? "kimi-k3" : "grok-build"; + function modelState(): AcpSchema.SessionModelState { - const modelId = grokAcpModels.some((model) => model.modelId === currentModelId) + const modelId = acpModels.some((model) => model.modelId === currentModelId) ? currentModelId - : "grok-build"; + : defaultAcpModelId; return { currentModelId: modelId, - availableModels: grokAcpModels, + availableModels: acpModels, }; } @@ -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 (!acpModels.some((model) => model.modelId === request.modelId)) { return yield* AcpError.AcpRequestError.invalidParams( `Unknown mock model id: ${request.modelId}`, { diff --git a/apps/server/src/provider/Drivers/KimiDriver.ts b/apps/server/src/provider/Drivers/KimiDriver.ts new file mode 100644 index 00000000000..6e86f7504a9 --- /dev/null +++ b/apps/server/src/provider/Drivers/KimiDriver.ts @@ -0,0 +1,168 @@ +import { KimiSettings, ProviderDriverKind, type ServerProvider } from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; +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 { 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 { + makeManualOnlyProviderMaintenanceCapabilities, + makeStaticProviderMaintenanceResolver, + resolveProviderMaintenanceCapabilitiesEffect, +} from "../providerMaintenance.ts"; +import { + haveProviderSnapshotSettingsChanged, + makeProviderSnapshotSettingsSource, + type ProviderSnapshotSettings, +} from "../providerUpdateSettings.ts"; +const decodeKimiSettings = Schema.decodeSync(KimiSettings); + +const DRIVER_KIND = ProviderDriverKind.make("kimi"); +const SNAPSHOT_REFRESH_INTERVAL = Duration.minutes(5); +const UPDATE = makeStaticProviderMaintenanceResolver( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: DRIVER_KIND, + packageName: null, + }), +); + +export type KimiDriverEnv = + | 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 fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + 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), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + ); + + 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, + }), + refreshInterval: SNAPSHOT_REFRESH_INTERVAL, + }).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/KimiAdapter.test.ts b/apps/server/src/provider/Layers/KimiAdapter.test.ts new file mode 100644 index 00000000000..080159e023b --- /dev/null +++ b/apps/server/src/provider/Layers/KimiAdapter.test.ts @@ -0,0 +1,1023 @@ +// @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 Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import { + ApprovalRequestId, + KimiSettings, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + TurnId, + type ProviderRuntimeEvent, +} from "@t3tools/contracts"; + +import { ServerConfig } from "../../config.ts"; +import { kimiPromptSettlementBelongsToContext, makeKimiAdapter } from "./KimiAdapter.ts"; +const decodeKimiSettings = Schema.decodeSync(KimiSettings); + +const __dirname = NodePath.dirname(NodeURL.fileURLToPath(import.meta.url)); +const mockAgentPath = NodePath.join(__dirname, "../../../scripts/acp-mock-agent.ts"); +const mockAgentCommand = process.execPath; + +async function makeMockKimiWrapper(extraEnv?: Record) { + const dir = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "kimi-acp-mock-")); + const wrapperPath = NodePath.join(dir, "fake-kimi.sh"); + const envExports = Object.entries({ T3_ACP_MODEL_SET: "kimi", ...extraEnv }) + .map(([key, value]) => `export ${key}=${JSON.stringify(value)}`) + .join("\n"); + const script = `#!/bin/sh +${envExports} +exec ${JSON.stringify(mockAgentCommand)} ${JSON.stringify(mockAgentPath)} "$@" +`; + await NodeFSP.writeFile(wrapperPath, script, "utf8"); + await NodeFSP.chmod(wrapperPath, 0o755); + return wrapperPath; +} + +function waitForFileContent( + filePath: string, + attempts = 40, + expectedContent?: string, +): Effect.Effect { + const readAttempt = (remainingAttempts: number): Effect.Effect => + Effect.gen(function* () { + if (remainingAttempts <= 0) { + return yield* Effect.die(new Error(`Timed out waiting for file content at ${filePath}`)); + } + const raw = yield* Effect.tryPromise(() => NodeFSP.readFile(filePath, "utf8")).pipe( + Effect.orElseSucceed(() => ""), + ); + if ( + raw.trim().length > 0 && + (expectedContent === undefined || raw.includes(expectedContent)) + ) { + return raw; + } + yield* Effect.sleep("25 millis"); + return yield* readAttempt(remainingAttempts - 1); + }); + return readAttempt(attempts); +} + +async function readJsonLines(filePath: string) { + const raw = await NodeFSP.readFile(filePath, "utf8"); + return raw + .split("\n") + .map((line) => line.trim()) + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as Record); +} + +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.isFalse( + kimiPromptSettlementBelongsToContext({ + liveAcpSessionId: "replacement-session", + expectedAcpSessionId: "stale-session", + liveActiveTurnId: staleTurnId, + liveSessionActiveTurnId: staleTurnId, + 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", + modelSelection: { + instanceId: ProviderInstanceId.make("kimi"), + model: "kimi-k2-thinking", + }, + }); + + assert.equal(session.provider, "kimi"); + assert.equal(session.model, "kimi-k2-thinking"); + assert.deepStrictEqual(session.resumeCursor, { + schemaVersion: 1, + sessionId: "mock-session-1", + }); + + yield* adapter.sendTurn({ + threadId, + input: "hello kimi", + attachments: [], + }); + + yield* Deferred.await(turnCompleted); + yield* Fiber.interrupt(runtimeEventsFiber); + const types = runtimeEvents.map((e) => e.type); + + assert.includeMembers(types, [ + "session.started", + "session.state.changed", + "thread.started", + "turn.started", + "item.started", + "content.delta", + "turn.completed", + ] as const); + + const delta = runtimeEvents.find((e) => e.type === "content.delta"); + assert.isDefined(delta); + if (delta?.type === "content.delta") { + assert.equal(delta.payload.delta, "hello from mock"); + } + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("switches the session model mid-session via session/set_model", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-mid-session-model-switch"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "kimi-set-model-")), + ); + const requestLogPath = NodePath.join(tempDir, "requests.ndjson"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ T3_ACP_REQUEST_LOG_PATH: requestLogPath }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + const turnCompleted = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + 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", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + assert.equal(session.model, "kimi-k3"); + + yield* adapter.sendTurn({ + threadId, + input: "switch models mid-session", + attachments: [], + modelSelection: { + instanceId: ProviderInstanceId.make("kimi"), + model: "kimi-k2-thinking", + }, + }); + + yield* Deferred.await(turnCompleted); + yield* Fiber.interrupt(runtimeEventsFiber); + + const sessions = yield* adapter.listSessions(); + const liveSession = sessions.find((candidate) => candidate.threadId === threadId); + assert.equal(liveSession?.model, "kimi-k2-thinking"); + + const requestLog = yield* Effect.promise(() => NodeFSP.readFile(requestLogPath, "utf8")); + const setModelIds = requestLog + .split("\n") + .filter((line) => line.trim().length > 0) + .map((line) => JSON.parse(line) as { method?: string; params?: { modelId?: string } }) + .filter((request) => request.method === "session/set_model") + .map((request) => request.params?.modelId); + assert.deepEqual(setModelIds, ["kimi-k2-thinking"]); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("keeps streaming events after the session-starting fiber exits", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-pump-survives-caller"); + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + + const turnCompleted = yield* Deferred.make(); + const sawContentDelta = yield* Ref.make(false); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.gen(function* () { + if (event.type === "content.delta") { + yield* Ref.set(sawContentDelta, true); + } + if (event.type === "turn.completed") { + yield* Deferred.succeed(turnCompleted, undefined); + } + }), + ).pipe(Effect.forkChild); + + // Run startSession in its own short-lived fiber and let that fiber exit + // before the turn runs: the notification pump must outlive its caller. + const startFiber = yield* adapter + .startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + }) + .pipe(Effect.forkChild); + yield* Fiber.join(startFiber); + + yield* adapter.sendTurn({ + threadId, + input: "turn after the starting fiber exited", + attachments: [], + }); + yield* Deferred.await(turnCompleted).pipe(Effect.timeout("5 seconds")); + yield* Fiber.interrupt(runtimeEventsFiber); + + assert.isTrue(yield* Ref.get(sawContentDelta)); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("closes the ACP child process when a session stops", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-stop-session-close"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "kimi-adapter-exit-log-")), + ); + const exitLogPath = NodePath.join(tempDir, "exit.log"); + + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_EXIT_LOG_PATH: exitLogPath, + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + + yield* adapter.stopSession(threadId); + + const exitLog = yield* waitForFileContent(exitLogPath); + assert.include(exitLog, "SIGTERM"); + }), + ); + + it.effect("reports a Kimi session running only while the prompt is in flight", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-session-ready-after-prompt"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_EMIT_TOOL_CALLS: "1", + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + const requestOpened = + yield* Deferred.make>(); + const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + event.type === "request.opened" + ? Deferred.succeed(requestOpened, event).pipe(Effect.ignore) + : Effect.void, + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "approval-required", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + + const sendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "check lifecycle", attachments: [] }) + .pipe(Effect.forkChild); + const requestOpenedEvent = yield* Deferred.await(requestOpened); + + const runningSessions = yield* adapter.listSessions(); + const runningSession = runningSessions.find((session) => session.threadId === threadId); + assert.equal(runningSession?.status, "running"); + assert.isDefined(runningSession?.activeTurnId); + + yield* adapter.respondToRequest( + threadId, + ApprovalRequestId.make(String(requestOpenedEvent.requestId)), + "accept", + ); + yield* Fiber.join(sendTurnFiber); + + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + yield* Fiber.interrupt(eventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("restores ready without completing an unstarted turn when preparation fails", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-preparation-failure-while-connecting"); + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + + const error = yield* Effect.flip( + adapter.sendTurn({ + threadId, + input: "prepare invalid attachment", + attachments: [ + { + type: "image", + id: "missing-image", + name: "missing.png", + mimeType: "image/png", + sizeBytes: 1, + }, + ], + }), + ); + for (let yieldAttempt = 0; yieldAttempt < 4; yieldAttempt += 1) { + yield* Effect.yieldNow; + } + + const turnCompletedEvent = runtimeEvents.find( + (event): event is Extract => + event.type === "turn.completed", + ); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + + assert.equal(error._tag, "ProviderAdapterRequestError"); + assert.isUndefined(turnCompletedEvent); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("lets Stop unblock a fully silent Kimi prompt and accept a follow-up turn", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-stop-after-full-silence"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1", + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + + yield* Effect.gen(function* () { + yield* Effect.sleep("500 millis"); + yield* adapter.interruptTurn(threadId); + }).pipe(Effect.forkChild({ startImmediately: true })); + + yield* adapter.sendTurn({ + threadId, + input: "hang forever", + attachments: [], + }); + for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { + yield* Effect.yieldNow; + } + + const cancelledEvents = runtimeEvents.filter( + (event): event is Extract => + event.type === "turn.completed" && String(event.threadId) === String(threadId), + ); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + + assert.lengthOf(cancelledEvents, 1); + assert.equal(cancelledEvents[0]?.payload.state, "cancelled"); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + const followUpEventsBefore = runtimeEvents.length; + yield* adapter.sendTurn({ + threadId, + input: "continue after stop", + attachments: [], + }); + for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { + yield* Effect.yieldNow; + } + + const followUpCompletedEvents = runtimeEvents + .slice(followUpEventsBefore) + .filter( + (event): event is Extract => + event.type === "turn.completed" && String(event.threadId) === String(threadId), + ); + assert.lengthOf(followUpCompletedEvents, 1); + assert.equal(followUpCompletedEvents[0]?.payload.state, "completed"); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }).pipe(TestClock.withLive), + ); + + it.effect("does not let a cancelled prompt settlement consume the follow-up prompt slot", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-cancelled-settlement-before-follow-up"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "kimi-acp-cancel-race-")), + ); + const requestLogPath = NodePath.join(tempDir, "requests.ndjson"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_HANG_FIRST_PROMPT_FOREVER: "1", + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const firstTurnStarted = yield* Deferred.make(); + const twoTurnsCompleted = yield* Deferred.make(); + const completedCountRef = yield* Ref.make(0); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.gen(function* () { + runtimeEvents.push(event); + if (String(event.threadId) !== String(threadId)) { + return; + } + if (event.type === "turn.started" && event.turnId !== undefined) { + yield* Deferred.succeed(firstTurnStarted, event.turnId).pipe(Effect.ignore); + return; + } + if (event.type !== "turn.completed") { + return; + } + const completedCount = yield* Ref.updateAndGet(completedCountRef, (count) => count + 1); + if (completedCount === 2) { + yield* Deferred.succeed(twoTurnsCompleted, undefined); + } + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + }); + + const firstSendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "cancel this prompt", attachments: [] }) + .pipe(Effect.forkChild); + const firstTurnId = yield* Deferred.await(firstTurnStarted).pipe(Effect.timeout("2 seconds")); + yield* waitForFileContent(requestLogPath, 80, '"method":"session/prompt"'); + + yield* adapter.interruptTurn(threadId, firstTurnId).pipe(Effect.timeout("2 seconds")); + const followUp = yield* adapter + .sendTurn({ threadId, input: "complete the follow-up", attachments: [] }) + .pipe(Effect.timeout("2 seconds")); + yield* Fiber.join(firstSendTurnFiber).pipe(Effect.timeout("2 seconds")); + yield* Deferred.await(twoTurnsCompleted).pipe(Effect.timeout("2 seconds")); + + const turnCompletedEvents = runtimeEvents.filter( + (event): event is Extract => + event.type === "turn.completed" && String(event.threadId) === String(threadId), + ); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + + assert.notEqual(String(followUp.turnId), String(firstTurnId)); + assert.deepEqual( + turnCompletedEvents.map((event) => [String(event.turnId), event.payload.state]), + [ + [String(firstTurnId), "cancelled"], + [String(followUp.turnId), "completed"], + ], + ); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }).pipe(TestClock.withLive), + ); + + it.effect("drops late ACP notifications after a turn is cancelled", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-drop-late-cancelled-notifications"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_HANG_PROMPT_FOREVER: "1", + T3_ACP_EMIT_LATE_UPDATE_AFTER_CANCEL: "1", + }), + ); + const lateNativeUpdate = yield* Deferred.make(); + const adapter = yield* makeTestAdapter(wrapperPath, { + nativeEventLogger: { + filePath: "memory://kimi-cancelled-native-events", + write: (record: unknown) => + JSON.stringify(record).includes("late after cancel") + ? Deferred.succeed(lateNativeUpdate, undefined).pipe(Effect.asVoid) + : Effect.void, + close: () => Effect.void, + }, + }); + + const runtimeEvents: ProviderRuntimeEvent[] = []; + const turnStarted = yield* Deferred.make(); + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }).pipe( + Effect.andThen( + event.type === "turn.started" && + event.turnId !== undefined && + String(event.threadId) === String(threadId) + ? Deferred.succeed(turnStarted, event.turnId).pipe(Effect.asVoid) + : Effect.void, + ), + ), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + }); + + const sendTurnFiber = yield* adapter + .sendTurn({ threadId, input: "cancel before the late update", attachments: [] }) + .pipe(Effect.forkChild); + const turnId = yield* Deferred.await(turnStarted).pipe(Effect.timeout("2 seconds")); + yield* adapter.interruptTurn(threadId, turnId).pipe(Effect.timeout("2 seconds")); + yield* Fiber.join(sendTurnFiber).pipe(Effect.timeout("2 seconds")); + yield* Deferred.await(lateNativeUpdate).pipe(Effect.timeout("2 seconds")); + for (let yieldAttempt = 0; yieldAttempt < 8; yieldAttempt += 1) { + yield* Effect.yieldNow; + } + + const cancelledIndex = runtimeEvents.findIndex( + (event) => + event.type === "turn.completed" && + String(event.threadId) === String(threadId) && + String(event.turnId) === String(turnId) && + event.payload.state === "cancelled", + ); + const turnOutputTypes = new Set([ + "content.delta", + "item.started", + "item.updated", + "item.completed", + "turn.plan.updated", + ]); + const outputAfterCancellation = runtimeEvents + .slice(cancelledIndex + 1) + .filter( + (event) => String(event.threadId) === String(threadId) && turnOutputTypes.has(event.type), + ); + + assert.isAtLeast(cancelledIndex, 0); + assert.deepEqual(outputAfterCancellation, []); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }).pipe(TestClock.withLive), + ); + + it.effect("settles the in-flight prompt before emitting completion", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-completion-before-next-turn"); + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + const completedCountRef = yield* Ref.make(0); + const secondTurnCompleted = yield* Deferred.make(); + + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => { + if (event.type !== "turn.completed" || String(event.threadId) !== String(threadId)) { + return Effect.void; + } + + return Ref.modify(completedCountRef, (count) => { + const nextCount = count + 1; + return [nextCount, nextCount] as const; + }).pipe( + Effect.flatMap((count) => { + if (count === 1) { + return adapter + .sendTurn({ + threadId, + input: "second turn after completion", + attachments: [], + }) + .pipe(Effect.forkChild, Effect.asVoid); + } + if (count === 2) { + return Deferred.succeed(secondTurnCompleted, undefined).pipe(Effect.asVoid); + } + return Effect.void; + }), + ); + }).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + + yield* adapter.sendTurn({ + threadId, + input: "first turn", + attachments: [], + }); + yield* Deferred.await(secondTurnCompleted); + + const completedCount = yield* Ref.get(completedCountRef); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + + assert.equal(completedCount, 2); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("restores a Kimi session to ready when the prompt RPC fails", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-prompt-failure-ready"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_FAIL_PROMPT: "1", + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + const runtimeEvents: ProviderRuntimeEvent[] = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + + const error = yield* Effect.flip( + adapter.sendTurn({ + threadId, + input: "fail prompt", + attachments: [], + }), + ); + const readySessions = yield* adapter.listSessions(); + const readySession = readySessions.find((session) => session.threadId === threadId); + const failedTurnCompleted = runtimeEvents.find( + (event) => event.type === "turn.completed" && event.threadId === threadId, + ); + + assert.equal(error._tag, "ProviderAdapterRequestError"); + assert.equal(readySession?.status, "ready"); + assert.isUndefined(readySession?.activeTurnId); + assert.equal(failedTurnCompleted?.type, "turn.completed"); + if (failedTurnCompleted?.type === "turn.completed") { + assert.equal(failedTurnCompleted.payload.state, "failed"); + assert.isString(failedTurnCompleted.payload.errorMessage); + } + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("ignores replayed session/load updates when resuming a Kimi session", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-load-replay-filter"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_EMIT_LOAD_REPLAY: "1", + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + const runtimeEvents: ProviderRuntimeEvent[] = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); + + const session = yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + resumeCursor: { schemaVersion: 1, sessionId: "mock-session-1" }, + }); + + yield* adapter.sendTurn({ + threadId, + input: "after resume", + attachments: [], + }); + + assert.deepStrictEqual(session.resumeCursor, { + schemaVersion: 1, + sessionId: "mock-session-1", + }); + assert.isFalse( + runtimeEvents.some( + (event) => event.type === "item.completed" && event.payload.title === "Replay tool", + ), + ); + assert.isFalse( + runtimeEvents.some( + (event) => + event.type === "content.delta" && event.payload.delta === "replayed assistant text", + ), + ); + + yield* Fiber.interrupt(runtimeEventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("rejects startSession when provider mismatches", () => + Effect.gen(function* () { + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + const threadId = ThreadId.make("kimi-provider-mismatch"); + + const error = yield* Effect.flip( + adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("cursor"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }), + ); + + assert.equal(error._tag, "ProviderAdapterValidationError"); + }), + ); + + it.effect("rejects sendTurn with empty input and no attachments", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-empty-turn"); + + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + modelSelection: { instanceId: ProviderInstanceId.make("kimi"), model: "kimi-k3" }, + }); + + const error = yield* Effect.flip( + adapter.sendTurn({ + threadId, + input: " ", + attachments: [], + }), + ); + + assert.equal(error._tag, "ProviderAdapterValidationError"); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("responds to ACP approvals using provider-supplied option ids", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-custom-approval-option-id"); + const tempDir = yield* Effect.promise(() => + NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "kimi-acp-")), + ); + const requestLogPath = NodePath.join(tempDir, "requests.ndjson"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ + T3_ACP_REQUEST_LOG_PATH: requestLogPath, + T3_ACP_EMIT_TOOL_CALLS: "1", + T3_ACP_ALLOW_ONCE_OPTION_ID: "agent-defined-approval-id", + }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + event.type === "request.opened" + ? adapter.respondToRequest( + threadId, + ApprovalRequestId.make(String(event.requestId)), + "accept", + ) + : Effect.void, + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "approval-required", + }); + yield* adapter.sendTurn({ threadId, input: "approve this", attachments: [] }); + + const requests = yield* Effect.promise(() => readJsonLines(requestLogPath)); + assert.isTrue( + requests.some( + (entry) => + !("method" in entry) && + typeof entry.result === "object" && + entry.result !== null && + "outcome" in entry.result && + typeof entry.result.outcome === "object" && + entry.result.outcome !== null && + "optionId" in entry.result.outcome && + entry.result.outcome.optionId === "agent-defined-approval-id", + ), + ); + + yield* Fiber.interrupt(eventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("does not report success for an approval already settled by interrupt", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-approval-after-interrupt"); + const wrapperPath = yield* Effect.promise(() => + makeMockKimiWrapper({ T3_ACP_EMIT_TOOL_CALLS: "1" }), + ); + const adapter = yield* makeTestAdapter(wrapperPath); + + const openedRequestId = yield* Deferred.make(); + const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + event.type === "request.opened" + ? Deferred.succeed(openedRequestId, ApprovalRequestId.make(String(event.requestId))) + : Effect.void, + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "approval-required", + }); + const turnFiber = yield* adapter + .sendTurn({ threadId, input: "ask for approval", attachments: [] }) + .pipe(Effect.exit, Effect.forkChild); + const requestId = yield* Deferred.await(openedRequestId).pipe(Effect.timeout("5 seconds")); + + yield* adapter.interruptTurn(threadId, undefined); + + // The interrupt settled the approval as cancelled: a late user decision + // must surface an error, never silently report success. + const error = yield* adapter + .respondToRequest(threadId, requestId, "accept") + .pipe(Effect.flip, Effect.timeout("5 seconds")); + assert.isTrue( + /already settled|Unknown pending approval/.test(String(error)), + `unexpected error: ${String(error)}`, + ); + + yield* Fiber.join(turnFiber); + yield* Fiber.interrupt(eventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("continues streaming events when native notification logging fails", () => + Effect.gen(function* () { + const threadId = ThreadId.make("kimi-native-log-failure"); + const wrapperPath = yield* Effect.promise(() => makeMockKimiWrapper()); + const adapter = yield* makeTestAdapter(wrapperPath, { + nativeEventLogger: { + filePath: "memory://kimi-native-events", + write: (record: unknown) => + typeof record === "object" && + record !== null && + "event" in record && + typeof record.event === "object" && + record.event !== null && + "kind" in record.event && + record.event.kind === "notification" + ? Effect.die(new Error("native log write failed")) + : Effect.void, + close: () => Effect.void, + }, + }); + const contentDelta = yield* Deferred.make(); + const eventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + event.type === "content.delta" ? Deferred.succeed(contentDelta, undefined) : Effect.void, + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("kimi"), + cwd: process.cwd(), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ threadId, input: "keep streaming", attachments: [] }); + yield* Deferred.await(contentDelta); + + yield* Fiber.interrupt(eventsFiber); + yield* adapter.stopSession(threadId); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/KimiAdapter.ts b/apps/server/src/provider/Layers/KimiAdapter.ts new file mode 100644 index 00000000000..f228cb39a90 --- /dev/null +++ b/apps/server/src/provider/Layers/KimiAdapter.ts @@ -0,0 +1,1414 @@ +import { + ApprovalRequestId, + type KimiSettings, + EventId, + type ProviderApprovalDecision, + type ProviderRuntimeEvent, + type ProviderSession, + 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 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; +} + +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; + 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 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 { + return response?.stopReason ?? null; +} + +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); + + // Reference-counted per-thread locks: each withThreadLock user holds a + // registry reference from acquire to release, and the entry is removed + // only when the LAST user releases. Long-lived servers therefore don't + // retain a semaphore per stopped thread, and — unlike deleting the entry + // inside stopSessionInternal, which runs while the lock is still held — + // no concurrent operation can ever mint a second semaphore for a thread + // that still has active holders or waiters. + const acquireThreadSemaphore = (threadId: string) => + SynchronizedRef.modifyEffect(threadLocksRef, (current) => { + const existing = current.get(threadId); + if (existing) { + const next = new Map(current); + next.set(threadId, { semaphore: existing.semaphore, refCount: existing.refCount + 1 }); + return Effect.succeed([existing.semaphore, next] as const); + } + return Semaphore.make(1).pipe( + Effect.map((semaphore) => { + const next = new Map(current); + next.set(threadId, { semaphore, refCount: 1 }); + return [semaphore, next] as const; + }), + ); + }); + + const releaseThreadSemaphore = (threadId: string) => + SynchronizedRef.update(threadLocksRef, (current) => { + const existing = current.get(threadId); + if (!existing) { + return current; + } + const next = new Map(current); + if (existing.refCount <= 1) { + next.delete(threadId); + } else { + next.set(threadId, { semaphore: existing.semaphore, refCount: existing.refCount - 1 }); + } + return next; + }); + + const withThreadLock = (threadId: string, effect: Effect.Effect) => + Effect.acquireUseRelease( + acquireThreadSemaphore(threadId), + (semaphore) => semaphore.withPermit(effect), + () => releaseThreadSemaphore(threadId), + ); + + 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); + 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 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), + // resolveKimiBinaryPath (its PATH check + ~/.kimi-code/bin fallback) + // needs FileSystem/Path; provide the ones captured above. + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + 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, + 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, + 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, + rawPayload: event.rawPayload, + }), + ); + return; + } + }), + ), + ).pipe( + Effect.catch((cause) => + Effect.logError("Failed to process Kimi runtime notification.", { cause }), + ), + // Fork into the session scope, not the calling fiber: startSession + // may run in a short-lived caller fiber, and the pump must keep + // forwarding ACP events until the session itself stops. + Effect.forkIn(sessionScope), + ); + + 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, + mapError: (cause) => + mapAcpToAdapterError(PROVIDER, input.threadId, "session/set_model", cause), + }); + // Track the switch immediately: if prompt validation below fails, + // the ACP session has still changed models, and a later turn + // requesting the same model must not skip session/set_model + // against a stale tracked id. Sync the session's reported model + // in the same step so listSessions never diverges from the ACP + // session either. + ctx.currentModelId = currentModelId; + const displayModel = currentModelId + ? resolveKimiAcpBaseModelId(currentModelId) + : undefined; + if (displayModel && ctx.session.model !== displayModel) { + ctx.session = { ...ctx.session, model: displayModel }; + } + + 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.", + }); + } + + 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, + 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); + // Plain mutable flag flipped synchronously with the append: the + // settlement path below yields between appending and marking + // promptSettled, and an interruption in that window must not let the + // ensuring finalizer append the same prompt/result pair again. + let promptResultAppended = 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) => + // Store the result before marking success: the finalizer treats + // promptRpcSucceeded with an empty promptResultRef as "nothing + // to settle", so an interruption between the two writes must + // never leave success marked without its result. + Effect.all([ + Ref.set(promptResultRef, promptResult), + Ref.set(promptRpcSucceeded, true), + ]), + ), + 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); + promptResultAppended = true; + ctx.session = { + ...ctx.session, + status: "running", + activeTurnId: prepared.turnId, + updatedAt: yield* nowIso, + ...(ctx.currentModelId + ? { model: resolveKimiAcpBaseModelId(ctx.currentModelId) } + : {}), + }; + if (ctx.interruptedTurnIds.has(prepared.turnId)) { + yield* Ref.set(promptSettled, true); + return { + threadId: input.threadId, + turnId: prepared.turnId, + resumeCursor: ctx.session.resumeCursor, + }; + } + + // Delegate slot decrement and terminal-event emission to + // settlePromptInFlight: only the last remaining prompt settles + // the merged turn (a steer-superseded prompt just releases its + // slot), and keeping promptsInFlight untouched until settlement + // actually runs means an interruption anywhere before it leaves + // state the ensuring finalizer can still settle from. + yield* settlePromptInFlight(input.threadId, prepared.turnId, prepared.acpSessionId, { + completedStopReason: completedStopReasonFromPromptResponse(result), + }); + ctx.interruptedTurnIds.delete(prepared.turnId); + 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; + } + if (!promptResultAppended) { + appendPromptResultToTurn( + ctx, + prepared.turnId, + prepared.promptParts, + promptResult, + ); + promptResultAppended = true; + } + 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* 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, + ) => + withThreadLock( + threadId, + 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}`, + }); + } + const settled = yield* Deferred.succeed(pending.decision, decision); + if (!settled) { + // Interrupt/stop already resolved this approval as cancelled; the + // caller's decision was not delivered, so don't report success. + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/request_permission", + detail: `Approval request already settled: ${requestId}`, + }); + } + ctx.pendingApprovals.delete(requestId); + }), + ); + + // Standard ACP has no user-input extension (unlike Grok's xAI + // ask_user_question), so no user-input request can ever be pending. + const respondToUserInput: KimiAdapterShape["respondToUserInput"] = (threadId, requestId) => + Effect.gen(function* () { + yield* requireSession(threadId); + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session/user_input", + detail: `Unknown pending user-input request: ${requestId}`, + }); + }); + + 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..8c74ce576ec --- /dev/null +++ b/apps/server/src/provider/Layers/KimiProvider.test.ts @@ -0,0 +1,138 @@ +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, + classifyKimiSessionModels, +} from "./KimiProvider.ts"; + +const decodeKimiSettings = Schema.decodeSync(KimiSettings); + +describe("classifyKimiSessionModels", () => { + it("treats a missing model option as an incompatible CLI, not a sign-in prompt", () => { + expect(classifyKimiSessionModels(null).kind).toBe("no-model-option"); + expect(classifyKimiSessionModels(undefined).kind).toBe("no-model-option"); + }); + + it("treats an EMPTY model list as signed-out (Kimi's logged-out signal)", () => { + expect(classifyKimiSessionModels({ availableModels: [] } as never).kind).toBe("signed-out"); + }); + + it("returns the deduped discovered models when the list is populated", () => { + const result = classifyKimiSessionModels({ + availableModels: [ + { modelId: "kimi-k3", name: "Kimi K3" }, + { modelId: "kimi-k3", name: "Kimi K3 (dupe)" }, + ], + } as never); + expect(result.kind).toBe("models"); + if (result.kind === "models") { + expect(result.models.map((model) => model.slug)).toEqual(["kimi-k3"]); + } + }); +}); + +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).toBeUndefined(); + }), + ); +}); + +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: "/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 = path.join(dir, "kimi"); + yield* fs.writeFileString( + kimiPath, + ["#!/bin/sh", `printf "%s\\n" "${secretStderr}" >&2`, "exit 2", ""].join("\n"), + ); + yield* fs.chmod(kimiPath, 0o755); + + 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 = path.join(dir, "kimi"); + yield* fs.writeFileString( + kimiPath, + ["#!/bin/sh", 'printf "kimi-cli 0.0.99\\n"', "exit 0", ""].join("\n"), + ); + yield* fs.chmod(kimiPath, 0o755); + + 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-k3"]); + 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..20ab77ae383 --- /dev/null +++ b/apps/server/src/provider/Layers/KimiProvider.ts @@ -0,0 +1,410 @@ +import { + type KimiSettings, + type ModelCapabilities, + ProviderDriverKind, + 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 FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +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 { + buildServerProvider, + isCommandMissingCause, + parseGenericCliVersion, + providerModelsFromSettings, + spawnAndCollect, + type ServerProviderDraft, +} from "../providerSnapshot.ts"; +import { + enrichProviderSnapshotWithVersionAdvisory, + type ProviderMaintenanceCapabilities, +} from "../providerMaintenance.ts"; +import { + makeKimiAcpRuntime, + resolveKimiAcpBaseModelId, + resolveKimiBinaryPath, +} from "../acp/KimiAcpSupport.ts"; + +const KIMI_PRESENTATION = { + displayName: "Kimi", + badgeLabel: "Early Access", + showInteractionModeToggle: false, +} as const; +const PROVIDER = ProviderDriverKind.make("kimi"); +const EMPTY_CAPABILITIES: ModelCapabilities = createModelCapabilities({ + optionDescriptors: [], +}); + +// The Kimi CLI is a large single binary; cold starts on Windows (first spawn after +// boot, antivirus scanning) can far exceed the ~1s warm-path latency, so a short +// probe timeout would spuriously report an installed CLI as missing. +const VERSION_PROBE_TIMEOUT_MS = 15_000; +const KIMI_ACP_MODEL_DISCOVERY_TIMEOUT_MS = 15_000; + +const KIMI_BUILT_IN_MODELS: ReadonlyArray = [ + { + slug: "kimi-k3", + name: "Kimi K3", + isCustom: false, + capabilities: EMPTY_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, + PROVIDER, + customModels ?? [], + EMPTY_CAPABILITIES, + ); +} + +// The result of ACP model discovery, distinguishing the two "no models" shapes: +// - "signed-out": the session carried a `model` config option but its list was +// EMPTY — Kimi's signal that no account is logged in. Surfaced as "run kimi login". +// - "no-model-option": the session carried NO `model` option at all, which points +// at an incompatible/malformed CLI, NOT an auth problem, so a login prompt would +// mislead the user. +export type KimiModelDiscovery = + | { readonly kind: "models"; readonly models: ReadonlyArray } + | { readonly kind: "signed-out" } + | { readonly kind: "no-model-option" }; + +export const KIMI_SIGNED_OUT_MESSAGE = + "Kimi Code CLI is installed but not signed in. Run `kimi login` and try again."; +export const KIMI_MODELS_UNAVAILABLE_MESSAGE = + "Kimi Code CLI is installed but returned no models. The installed CLI may be incompatible or misconfigured; check server logs for details."; + +export function classifyKimiSessionModels( + modelState: EffectAcpSchema.SessionModelState | null | undefined, +): KimiModelDiscovery { + if (!modelState) { + return { kind: "no-model-option" }; + } + if (modelState.availableModels.length === 0) { + return { kind: "signed-out" }; + } + const seen = new Set(); + const models = 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: EMPTY_CAPABILITIES, + }; + }) + .filter((model): model is ServerProviderModel => model !== undefined); + return { kind: "models", models }; +} + +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(); + return classifyKimiSessionModels(started.sessionSetupResult.models); + }).pipe(Effect.scoped); + +const runKimiVersionCommand = ( + kimiSettings: KimiSettings, + environment: NodeJS.ProcessEnv = process.env, +) => + Effect.gen(function* () { + const command = yield* resolveKimiBinaryPath(kimiSettings, environment); + 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 | FileSystem.FileSystem | Path.Path +> { + 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 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. Check server logs for details.", + }, + }); + } + 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 discovery = discoveryExit.value.value; + + // Signed out: the CLI is installed and healthy but has no account. Tell the user + // to log in rather than silently falling back to the built-in `kimi-k3` entry, + // which would then fail the moment they tried to use it. + if (discovery.kind === "signed-out") { + yield* Effect.logInfo("Kimi CLI is installed but signed out (empty ACP model list)."); + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models: fallbackModels, + probe: { + installed: true, + version, + status: "error", + auth: { status: "unauthenticated" }, + message: KIMI_SIGNED_OUT_MESSAGE, + }, + }); + } + + // No model option at all — an incompatible/malformed CLI, not an auth problem. + if (discovery.kind === "no-model-option") { + yield* Effect.logWarning("Kimi CLI ACP session carried no model config option."); + return buildServerProvider({ + presentation: KIMI_PRESENTATION, + enabled: kimiSettings.enabled, + checkedAt, + models: fallbackModels, + probe: { + installed: true, + version, + status: "error", + auth: { status: "unknown" }, + message: KIMI_MODELS_UNAVAILABLE_MESSAGE, + }, + }); + } + + const discoveredModels = discovery.models; + 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: "authenticated" }, + }, + }); +}); + +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/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index 73390450efa..4de4242042a 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -29,6 +29,7 @@ import { type CodexSettings, type CursorSettings, type GrokSettings, + type KimiSettings, type OpenCodeSettings, ProviderDriverKind, type ProviderInstanceConfigMap, @@ -44,6 +45,7 @@ import { ClaudeDriver } from "../Drivers/ClaudeDriver.ts"; import { CodexDriver } from "../Drivers/CodexDriver.ts"; import { CursorDriver } from "../Drivers/CursorDriver.ts"; import { GrokDriver } from "../Drivers/GrokDriver.ts"; +import { KimiDriver } from "../Drivers/KimiDriver.ts"; import { OpenCodeDriver } from "../Drivers/OpenCodeDriver.ts"; import { OpenCodeRuntimeLive } from "../opencodeRuntime.ts"; import { NoOpProviderEventLoggers, ProviderEventLoggers } from "./ProviderEventLoggers.ts"; @@ -89,6 +91,13 @@ const makeGrokConfig = (overrides: Partial): GrokSettings => ({ ...overrides, }); +const makeKimiConfig = (overrides: Partial): KimiSettings => ({ + enabled: false, + binaryPath: "kimi", + customModels: [], + ...overrides, +}); + const makeOpenCodeConfig = (overrides: Partial): OpenCodeSettings => ({ enabled: false, binaryPath: "opencode", @@ -257,12 +266,14 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { const claudeId = ProviderInstanceId.make("claude_default"); const cursorId = ProviderInstanceId.make("cursor_default"); const grokId = ProviderInstanceId.make("grok_default"); + const kimiId = ProviderInstanceId.make("kimi_default"); const openCodeId = ProviderInstanceId.make("opencode_default"); const codexDriverKind = ProviderDriverKind.make("codex"); const claudeDriverKind = ProviderDriverKind.make("claudeAgent"); const cursorDriverKind = ProviderDriverKind.make("cursor"); const grokDriverKind = ProviderDriverKind.make("grok"); + const kimiDriverKind = ProviderDriverKind.make("kimi"); const openCodeDriverKind = ProviderDriverKind.make("opencode"); const configMap: ProviderInstanceConfigMap = { @@ -293,6 +304,12 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { enabled: false, config: makeGrokConfig({}), }, + [kimiId]: { + driver: kimiDriverKind, + displayName: "Kimi", + enabled: false, + config: makeKimiConfig({}), + }, [openCodeId]: { driver: openCodeDriverKind, displayName: "OpenCode", @@ -302,7 +319,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { }; const { registry } = yield* makeProviderInstanceRegistry({ - drivers: [CodexDriver, ClaudeDriver, CursorDriver, GrokDriver, OpenCodeDriver], + drivers: [CodexDriver, ClaudeDriver, CursorDriver, GrokDriver, KimiDriver, OpenCodeDriver], configMap, }); @@ -312,9 +329,9 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { expect(unavailable).toEqual([]); const instances = yield* registry.listInstances; - expect(instances).toHaveLength(5); + expect(instances).toHaveLength(6); expect(instances.map((instance) => instance.instanceId).toSorted()).toEqual( - [codexId, claudeId, cursorId, grokId, openCodeId].toSorted(), + [codexId, claudeId, cursorId, grokId, kimiId, openCodeId].toSorted(), ); // Instance lookup by id resolves each instance to its own bundle — @@ -324,16 +341,19 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { const claude = yield* registry.getInstance(claudeId); const cursor = yield* registry.getInstance(cursorId); const grok = yield* registry.getInstance(grokId); + const kimi = yield* registry.getInstance(kimiId); const openCode = yield* registry.getInstance(openCodeId); expect(codex?.driverKind).toBe(codexDriverKind); expect(claude?.driverKind).toBe(claudeDriverKind); expect(cursor?.driverKind).toBe(cursorDriverKind); expect(grok?.driverKind).toBe(grokDriverKind); + expect(kimi?.driverKind).toBe(kimiDriverKind); expect(openCode?.driverKind).toBe(openCodeDriverKind); expect(codex?.displayName).toBe("Codex"); expect(claude?.displayName).toBe("Claude"); expect(cursor?.displayName).toBe("Cursor"); expect(grok?.displayName).toBe("Grok"); + expect(kimi?.displayName).toBe("Kimi"); expect(openCode?.displayName).toBe("OpenCode"); // Every instance owns its own set of closures — no sharing across @@ -346,6 +366,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { claude!.adapter, cursor!.adapter, grok!.adapter, + kimi!.adapter, openCode!.adapter, ]; expect(new Set(adapters).size).toBe(adapters.length); @@ -354,6 +375,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { claude!.textGeneration, cursor!.textGeneration, grok!.textGeneration, + kimi!.textGeneration, openCode!.textGeneration, ]; expect(new Set(textGenerations).size).toBe(textGenerations.length); @@ -362,6 +384,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { claude!.snapshot, cursor!.snapshot, grok!.snapshot, + kimi!.snapshot, openCode!.snapshot, ]; expect(new Set(snapshots).size).toBe(snapshots.length); @@ -398,6 +421,12 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { expect(grokSnapshot.enabled).toBe(false); expect(grokSnapshot.continuation?.groupKey).toBe(`${grokDriverKind}:instance:${grokId}`); + const kimiSnapshot = yield* kimi!.snapshot.getSnapshot; + expect(kimiSnapshot.instanceId).toBe(kimiId); + expect(kimiSnapshot.driver).toBe(kimiDriverKind); + expect(kimiSnapshot.enabled).toBe(false); + expect(kimiSnapshot.continuation?.groupKey).toBe(`${kimiDriverKind}:instance:${kimiId}`); + const openCodeSnapshot = yield* openCode!.snapshot.getSnapshot; expect(openCodeSnapshot.instanceId).toBe(openCodeId); expect(openCodeSnapshot.driver).toBe(openCodeDriverKind); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 5456a90bdf5..623090ae38c 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -1076,6 +1076,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te claudeAgent: { enabled: false }, cursor: { enabled: false }, grok: { enabled: false }, + kimi: { enabled: false }, opencode: { enabled: false }, }, // `providerInstances` keys are branded `ProviderInstanceId`; @@ -1187,6 +1188,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te claudeAgent: { enabled: false }, cursor: { enabled: false }, grok: { enabled: false }, + kimi: { enabled: false }, opencode: { enabled: false }, }, }), @@ -1300,6 +1302,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te claudeAgent: { enabled: false }, cursor: { enabled: false }, grok: { enabled: false }, + kimi: { enabled: false }, opencode: { enabled: false }, }, providerInstances: { @@ -1369,6 +1372,9 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te grok: { enabled: false, }, + kimi: { + enabled: false, + }, }, }), ), @@ -1437,6 +1443,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te "codex", "cursor", "grok", + "kimi", "opencode", ]); assert.strictEqual(cursorProvider?.enabled, false); diff --git a/apps/server/src/provider/Services/KimiAdapter.ts b/apps/server/src/provider/Services/KimiAdapter.ts new file mode 100644 index 00000000000..c8e15aba78d --- /dev/null +++ b/apps/server/src/provider/Services/KimiAdapter.ts @@ -0,0 +1,16 @@ +/** + * KimiAdapter — shape type for the Kimi 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/KimiAcpCliProbe.test.ts b/apps/server/src/provider/acp/KimiAcpCliProbe.test.ts new file mode 100644 index 00000000000..771ab5d7f69 --- /dev/null +++ b/apps/server/src/provider/acp/KimiAcpCliProbe.test.ts @@ -0,0 +1,68 @@ +/** + * Optional integration check against a real `kimi acp` install. + * Enable with: T3_KIMI_ACP_PROBE=1 bun run test KimiAcpCliProbe + * + * The probe assumes either `KIMI_API_KEY` is set in the environment or + * the user has previously run `kimi login`. Without credentials the + * agent's `authenticate` request will fail and the test will surface + * the error. + */ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import { ChildProcessSpawner } from "effect/unstable/process"; +import { describe, expect } from "vite-plus/test"; + +import { makeKimiAcpRuntime } from "./KimiAcpSupport.ts"; + +const makeProbeRuntime = Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + return yield* makeKimiAcpRuntime({ + kimiSettings: { binaryPath: "kimi" }, + environment: process.env, + childProcessSpawner, + cwd: process.cwd(), + clientInfo: { name: "t3-kimi-probe", version: "0.0.0" }, + }); +}); + +describe.runIf(process.env.T3_KIMI_ACP_PROBE === "1")("Kimi ACP CLI probe", () => { + it.effect("initialize and authenticate against real kimi acp", () => + Effect.gen(function* () { + const runtime = yield* makeProbeRuntime; + const started = yield* runtime.start(); + expect(started.initializeResult).toBeDefined(); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("session/new advertises typed SessionModelState with at least one model", () => + Effect.gen(function* () { + const runtime = yield* makeProbeRuntime; + const started = yield* runtime.start(); + const result = started.sessionSetupResult; + + expect(typeof started.sessionId).toBe("string"); + + // Kimi advertises models through the typed `SessionModelState` field. + // If this assertion fails the upstream surface has regressed. + const models = result.models; + expect(models).toBeDefined(); + expect(typeof models?.currentModelId).toBe("string"); + expect(models?.availableModels.length ?? 0).toBeGreaterThan(0); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); + + it.effect("session/set_model accepts a no-op switch to the current model", () => + Effect.gen(function* () { + const runtime = yield* makeProbeRuntime; + const started = yield* runtime.start(); + const currentModelId = started.sessionSetupResult.models?.currentModelId?.trim(); + expect(currentModelId).toBeDefined(); + if (!currentModelId) return; + + // No-op switch — selecting the model the session already runs on must + // succeed against every Kimi build that implements `session/set_model`. + yield* runtime.setSessionModel(currentModelId); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), + ); +}); 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..0562861ded7 --- /dev/null +++ b/apps/server/src/provider/acp/KimiAcpSupport.test.ts @@ -0,0 +1,169 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as NodeOS from "node:os"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as EffectAcpErrors from "effect-acp/errors"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; + +import { + applyKimiAcpModelSelection, + buildKimiAcpSpawnInput, + resolveKimiAcpBaseModelId, + resolveKimiBinaryPath, +} from "./KimiAcpSupport.ts"; + +describe("resolveKimiAcpBaseModelId", () => { + it("defaults empty ids and passes ACP-discovered ids through verbatim", () => { + expect(resolveKimiAcpBaseModelId(undefined)).toBe("kimi-k3"); + expect(resolveKimiAcpBaseModelId(" ")).toBe("kimi-k3"); + expect(resolveKimiAcpBaseModelId("k3")).toBe("k3"); + expect(resolveKimiAcpBaseModelId("k2-thinking")).toBe("k2-thinking"); + expect(resolveKimiAcpBaseModelId(" kimi-test-custom-model ")).toBe("kimi-test-custom-model"); + }); +}); + +describe("buildKimiAcpSpawnInput", () => { + it.effect("spawns `kimi acp` and passes the environment through unchanged", () => + Effect.gen(function* () { + const spawn = yield* buildKimiAcpSpawnInput( + { binaryPath: "/usr/local/bin/kimi" }, + "/tmp/project", + { KIMI_API_KEY: "secret" }, + ); + + expect(spawn).toEqual({ + command: "/usr/local/bin/kimi", + args: ["acp"], + cwd: "/tmp/project", + env: { + KIMI_API_KEY: "secret", + }, + }); + }).pipe(Effect.provide(NodeServices.layer)), + ); +}); + +describe("resolveKimiBinaryPath", () => { + it.effect("uses an explicit binaryPath verbatim without probing PATH", () => + Effect.gen(function* () { + const resolved = yield* resolveKimiBinaryPath({ binaryPath: "/opt/kimi/bin/kimi" }); + expect(resolved).toBe("/opt/kimi/bin/kimi"); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("falls back to the well-known install path when kimi is not on PATH", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const expected = path.join(NodeOS.homedir(), ".kimi-code", "bin", "kimi.exe"); + const fakeFileSystem = FileSystem.FileSystem.of({ + ...fileSystem, + exists: (candidate) => Effect.succeed(candidate === expected), + }); + + // Empty PATH so the bare `kimi` is not resolvable; win32 so the fallback + // targets kimi.exe. + const resolved = yield* resolveKimiBinaryPath( + { binaryPath: "kimi" }, + { PATH: "", PATHEXT: ".EXE" }, + ).pipe( + Effect.provideService(FileSystem.FileSystem, fakeFileSystem), + Effect.provideService(HostProcessPlatform, "win32"), + ); + + expect(resolved).toBe(expected); + }).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("returns the bare command when kimi is neither on PATH nor at the fallback", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const fakeFileSystem = FileSystem.FileSystem.of({ + ...fileSystem, + exists: () => Effect.succeed(false), + }); + + const resolved = yield* resolveKimiBinaryPath({ binaryPath: "kimi" }, { PATH: "" }).pipe( + Effect.provideService(FileSystem.FileSystem, fakeFileSystem), + Effect.provideService(HostProcessPlatform, "win32"), + ); + + expect(resolved).toBe("kimi"); + }).pipe(Effect.provide(NodeServices.layer)), + ); +}); + +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-k3", + requestedModelId: "kimi-k2-thinking", + mapError: (cause) => cause.message, + }); + expect(modelCalls).toEqual(["kimi-k2-thinking"]); + expect(result).toBe("kimi-k2-thinking"); + }), + ); + + it.effect("skips set_model when requested matches current", () => + Effect.gen(function* () { + const { runtime, modelCalls } = makeRecordingRuntime(); + const result = yield* applyKimiAcpModelSelection({ + runtime, + currentModelId: "kimi-k3", + requestedModelId: "kimi-k3", + mapError: (cause) => cause.message, + }); + expect(modelCalls).toEqual([]); + expect(result).toBe("kimi-k3"); + }), + ); + + it.effect("skips set_model when no model is requested", () => + Effect.gen(function* () { + const { runtime, modelCalls } = makeRecordingRuntime(); + const result = yield* applyKimiAcpModelSelection({ + runtime, + currentModelId: "kimi-k3", + requestedModelId: undefined, + mapError: (cause) => cause.message, + }); + expect(modelCalls).toEqual([]); + expect(result).toBe("kimi-k3"); + }), + ); + + 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-k3", + requestedModelId: "kimi-k2-thinking", + mapError: (cause) => cause.message, + }), + ); + expect(error).toBe(failure.message); + }), + ); +}); diff --git a/apps/server/src/provider/acp/KimiAcpSupport.ts b/apps/server/src/provider/acp/KimiAcpSupport.ts new file mode 100644 index 00000000000..43b56d3bc0b --- /dev/null +++ b/apps/server/src/provider/acp/KimiAcpSupport.ts @@ -0,0 +1,136 @@ +import { type KimiSettings, ProviderDriverKind } from "@t3tools/contracts"; +import * as NodeOS from "node:os"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +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 { normalizeModelSlug } from "@t3tools/shared/model"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { isCommandAvailable } from "@t3tools/shared/shell"; + +import * as AcpSessionRuntime from "./AcpSessionRuntime.ts"; + +const KIMI_AUTH_METHOD_LOGIN = "login"; +const KIMI_DRIVER_KIND = ProviderDriverKind.make("kimi"); + +type KimiAcpRuntimeKimiSettings = Pick; + +/** + * Resolve the `kimi` executable, falling back to the CLI's well-known install + * directory when it is not on `PATH`. + * + * An explicit `binaryPath` (anything other than the bare default `"kimi"`) is + * authoritative and used verbatim. Otherwise we prefer a `PATH`-resolvable `kimi`, + * and only if that is missing do we try `~/.kimi-code/bin/kimi[.exe]` — the Kimi + * installer does NOT add `kimi` to `PATH` on Windows, so a Windows user with no + * manual override would otherwise get a spurious "not installed". Threading the + * resolved concrete path through means downstream spawns don't re-scan `PATH`. + */ +export const resolveKimiBinaryPath = Effect.fn("resolveKimiBinaryPath")(function* ( + kimiSettings: KimiAcpRuntimeKimiSettings | null | undefined, + environment?: NodeJS.ProcessEnv, +) { + const configured = kimiSettings?.binaryPath?.trim(); + if (configured && configured !== "kimi") { + return configured; + } + const command = configured || "kimi"; + if (yield* isCommandAvailable(command, environment ? { env: environment } : {})) { + return command; + } + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const platform = yield* HostProcessPlatform; + const wellKnownPath = path.join( + NodeOS.homedir(), + ".kimi-code", + "bin", + platform === "win32" ? "kimi.exe" : "kimi", + ); + const exists = yield* fileSystem.exists(wellKnownPath).pipe(Effect.orElseSucceed(() => false)); + return exists ? wellKnownPath : command; +}); + +interface KimiAcpRuntimeInput extends Omit< + AcpSessionRuntime.AcpSessionRuntimeOptions, + "authMethodId" | "clientCapabilities" | "spawn" +> { + readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; + readonly kimiSettings: KimiAcpRuntimeKimiSettings | null | undefined; + readonly environment?: NodeJS.ProcessEnv; +} + +export const buildKimiAcpSpawnInput = Effect.fn("buildKimiAcpSpawnInput")(function* ( + kimiSettings: KimiAcpRuntimeKimiSettings | null | undefined, + cwd: string, + environment?: NodeJS.ProcessEnv, +) { + const command = yield* resolveKimiBinaryPath(kimiSettings, environment); + return { + command, + args: ["acp"], + cwd, + env: { ...environment }, + } satisfies AcpSessionRuntime.AcpSpawnInput; +}); + +export const makeKimiAcpRuntime = ( + input: KimiAcpRuntimeInput, +): Effect.Effect< + AcpSessionRuntime.AcpSessionRuntime["Service"], + EffectAcpErrors.AcpError, + Crypto.Crypto | Scope.Scope | FileSystem.FileSystem | Path.Path +> => + Effect.gen(function* () { + const spawn = yield* buildKimiAcpSpawnInput(input.kimiSettings, input.cwd, input.environment); + const acpContext = yield* Layer.build( + AcpSessionRuntime.layer({ + ...input, + spawn, + 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 base = trimmed && trimmed.length > 0 ? trimmed : "kimi-k3"; + return normalizeModelSlug(base, KIMI_DRIVER_KIND) ?? "kimi-k3"; +} + +export function currentKimiModelIdFromSessionSetup( + sessionSetupResult: + | EffectAcpSchema.LoadSessionResponse + | EffectAcpSchema.NewSessionResponse + | EffectAcpSchema.ResumeSessionResponse, +): string | undefined { + return sessionSetupResult.models?.currentModelId?.trim() || undefined; +} + +export function applyKimiAcpModelSelection(input: { + readonly runtime: Pick; + readonly currentModelId: string | undefined; + readonly requestedModelId: string | undefined; + readonly mapError: (cause: EffectAcpErrors.AcpError) => E; +}): Effect.Effect { + const shouldSwitchModel = + input.requestedModelId !== undefined && input.requestedModelId !== input.currentModelId; + if (!shouldSwitchModel) { + return Effect.succeed(input.currentModelId); + } + return input.runtime + .setSessionModel(input.requestedModelId) + .pipe(Effect.mapError(input.mapError), Effect.as(input.requestedModelId)); +} 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"); + const kimiPath = NodePath.join(binDir, "kimi"); + NodeFS.mkdirSync(binDir, { recursive: true }); + NodeFS.writeFileSync( + kimiPath, + [ + "#!/bin/sh", + "export T3_ACP_MODEL_SET=kimi", + ...Object.entries(env).map(([key, value]) => `export ${key}=${shellSingleQuote(value)}`), + 'if [ "$1" != "acp" ]; then', + ' printf "%s\\n" "unexpected args: $*" >&2', + " exit 11", + "fi", + `exec ${JSON.stringify(process.execPath)} ${JSON.stringify(mockAgentPath)}`, + "", + ].join("\n"), + "utf8", + ); + NodeFS.chmodSync(kimiPath, 0o755); + return kimiPath; +} + +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* () { + 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"), + "kimi-k2-thinking", + ), + }); + + 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 === "kimi-k2-thinking", + ), + ).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"), + "kimi-k2-thinking", + ), + }); + 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"), "kimi-k3"), + }), + ); + expect(error._tag).toBe("TextGenerationError"); + expect(error.detail).toMatch(/empty/i); + }), + ), + ); + + it.effect("decodes a structured PR title + body", () => + withFakeAcpKimi( + { + T3_ACP_PROMPT_RESPONSE_TEXT: JSON.stringify({ + title: "feat(kimi): wire up session/set_model", + body: "## Summary\n- Wire up the typed ACP `session/set_model`.\n- Translate model switch failures into a validation error.", + }), + }, + (textGeneration) => + Effect.gen(function* () { + const generated = yield* textGeneration.generatePrContent({ + cwd: process.cwd(), + baseBranch: "main", + headBranch: "feat/kimi-provider", + commitSummary: "feat: add kimi provider", + diffSummary: "M apps/server/src/provider/Drivers/KimiDriver.ts", + diffPatch: "diff --git a/.../KimiDriver.ts b/.../KimiDriver.ts", + modelSelection: createModelSelection(ProviderInstanceId.make("kimi"), "kimi-k3"), + }); + + expect(generated.title).toBe("feat(kimi): wire up session/set_model"); + expect(generated.body).toContain("Translate model switch failures"); + }), + ), + ); + + it.effect("fails with TextGenerationError when output is unparseable JSON", () => + withFakeAcpKimi( + { + T3_ACP_PROMPT_RESPONSE_TEXT: "totally not json output from a confused model", + }, + (textGeneration) => + Effect.gen(function* () { + const error = yield* Effect.flip( + textGeneration.generateThreadTitle({ + cwd: process.cwd(), + message: "anything", + modelSelection: createModelSelection(ProviderInstanceId.make("kimi"), "kimi-k3"), + }), + ); + expect(error._tag).toBe("TextGenerationError"); + expect(error.detail).toMatch(/invalid structured output/i); + }), + ), + ); +}); diff --git a/apps/server/src/textGeneration/KimiTextGeneration.ts b/apps/server/src/textGeneration/KimiTextGeneration.ts new file mode 100644 index 00000000000..882cb8a8d28 --- /dev/null +++ b/apps/server/src/textGeneration/KimiTextGeneration.ts @@ -0,0 +1,268 @@ +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +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; + // resolveKimiBinaryPath (PATH probe + ~/.kimi-code/bin fallback, reached via + // makeKimiAcpRuntime below) needs FileSystem/Path. Capture them at construction so + // each per-request runKimiJson can provide them — otherwise the requirement leaks + // past this service's declared R (never) and defects at runtime. + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + + 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), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + ); + + 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, + }); + + 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, + }); + + 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, + 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 e62a79afe78..7e61481c3f6 100644 --- a/apps/server/src/textGeneration/TextGeneration.ts +++ b/apps/server/src/textGeneration/TextGeneration.ts @@ -7,7 +7,13 @@ import { TextGenerationError } from "@t3tools/contracts"; import * as ProviderInstanceRegistry from "../provider/Services/ProviderInstanceRegistry.ts"; import type { ProviderInstance } from "../provider/ProviderDriver.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..93088a182a8 100644 --- a/apps/web/src/components/Icons.tsx +++ b/apps/web/src/components/Icons.tsx @@ -211,6 +211,17 @@ export const GrokIcon: Icon = ({ className, ...props }) => ( ); +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 f9e7a700716..9f7cc35bd2f 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/modelSelection.test.ts b/apps/web/src/modelSelection.test.ts index 3d973ccca74..1d7a7ccc588 100644 --- a/apps/web/src/modelSelection.test.ts +++ b/apps/web/src/modelSelection.test.ts @@ -122,6 +122,27 @@ describe("instance-scoped model selection", () => { ); }); + it("includes Kimi custom models from the selected provider instance", () => { + const providers = [provider({ provider: ProviderDriverKind.make("kimi"), instanceId: "kimi" })]; + const settings: UnifiedSettings = { + ...settingsWithProviderInstances(), + providerInstances: { + ...settingsWithProviderInstances().providerInstances, + [ProviderInstanceId.make("kimi")]: { + driver: ProviderDriverKind.make("kimi"), + config: { customModels: ["kimi-test-custom-model"] }, + }, + }, + }; + const kimi = deriveProviderInstanceEntries(providers).find( + (entry) => entry.instanceId === "kimi", + )!; + + expect(getAppModelOptionsForInstance(settings, kimi).map((option) => option.slug)).toContain( + "kimi-test-custom-model", + ); + }); + it("does not inject an unknown selected slug into the stock instance list", () => { const providers = [ provider({ 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/packages/contracts/src/model.ts b/packages/contracts/src/model.ts index dddf3f37459..f89f46eaa32 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.4"; @@ -141,6 +142,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 b05f397bf5c..689f0cb1370 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -305,6 +305,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( @@ -398,6 +422,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 @@ -493,6 +518,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), @@ -522,6 +553,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), }), ), diff --git a/packages/shared/src/model.test.ts b/packages/shared/src/model.test.ts index ee91e0b53f7..2b4572fd580 100644 --- a/packages/shared/src/model.test.ts +++ b/packages/shared/src/model.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vite-plus/test"; -import { ProviderInstanceId, type ModelCapabilities } from "@t3tools/contracts"; +import { ProviderDriverKind, ProviderInstanceId, type ModelCapabilities } from "@t3tools/contracts"; import { buildProviderOptionSelectionsFromDescriptors, @@ -10,6 +10,7 @@ import { getProviderOptionDescriptors, getProviderOptionBooleanSelectionValue, getProviderOptionStringSelectionValue, + normalizeModelSlug, } from "./model.ts"; const codexCaps: ModelCapabilities = createModelCapabilities({ @@ -59,6 +60,28 @@ const claudeCaps: ModelCapabilities = createModelCapabilities({ ], }); +describe("normalizeModelSlug — Kimi", () => { + const KIMI = ProviderDriverKind.make("kimi"); + + it("has no Kimi aliases — ACP-discovered ids must round-trip verbatim", () => { + expect(normalizeModelSlug("k3", KIMI)).toBe("k3"); + expect(normalizeModelSlug("k2-thinking", KIMI)).toBe("k2-thinking"); + expect(normalizeModelSlug("kimi-k2", KIMI)).toBe("kimi-k2"); + }); + + it("passes canonical and unknown Kimi slugs through unchanged", () => { + expect(normalizeModelSlug("kimi-k3", KIMI)).toBe("kimi-k3"); + expect(normalizeModelSlug("kimi-k2-thinking", KIMI)).toBe("kimi-k2-thinking"); + expect(normalizeModelSlug(" kimi-k3 ", KIMI)).toBe("kimi-k3"); + expect(normalizeModelSlug("some-future-model", KIMI)).toBe("some-future-model"); + }); + + it("returns null for empty or non-string input", () => { + expect(normalizeModelSlug(undefined, KIMI)).toBeNull(); + expect(normalizeModelSlug(" ", KIMI)).toBeNull(); + }); +}); + describe("descriptor helpers", () => { it("applies selection values to capability descriptors", () => { expect(