diff --git a/apps/server/package.json b/apps/server/package.json index d0903c77d75..956420e8151 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -28,6 +28,7 @@ "@effect/platform-node-shared": "catalog:", "@effect/sql-sqlite-bun": "catalog:", "@ff-labs/fff-node": "0.9.4", + "@kilocode/sdk": "^7.4.5", "@opencode-ai/sdk": "^1.3.15", "@pierre/diffs": "catalog:", "effect": "catalog:", diff --git a/apps/server/src/provider/Drivers/KiloDriver.ts b/apps/server/src/provider/Drivers/KiloDriver.ts new file mode 100644 index 00000000000..9b3da0d5ffc --- /dev/null +++ b/apps/server/src/provider/Drivers/KiloDriver.ts @@ -0,0 +1,129 @@ +import { KiloSettings, ProviderDriverKind, type ServerProvider } from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Duration from "effect/Duration"; +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 { makeKiloTextGeneration } from "../../textGeneration/KiloTextGeneration.ts"; +import { ServerConfig } from "../../config.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import { ProviderDriverError } from "../Errors.ts"; +import { makeKiloAdapter } from "../Layers/KiloAdapter.ts"; +import { checkKiloProviderStatus, makePendingKiloProvider } from "../Layers/KiloProvider.ts"; +import { makeManagedServerProvider } from "../makeManagedServerProvider.ts"; +import { mergeProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; +import { + defaultProviderContinuationIdentity, + type ProviderDriver, + type ProviderInstance, +} from "../ProviderDriver.ts"; +import type { ServerProviderDraft } from "../providerSnapshot.ts"; +import { makeManualOnlyProviderMaintenanceCapabilities } from "../providerMaintenance.ts"; +import { + haveProviderSnapshotSettingsChanged, + makeProviderSnapshotSettingsSource, + type ProviderSnapshotSettings, +} from "../providerUpdateSettings.ts"; +import { KiloRuntime } from "../kiloRuntime.ts"; + +const DRIVER_KIND = ProviderDriverKind.make("kilo"); +const REFRESH_INTERVAL = Duration.minutes(5); +const decodeSettings = Schema.decodeSync(KiloSettings); + +export type KiloDriverEnv = + | Crypto.Crypto + | FileSystem.FileSystem + | HttpClient.HttpClient + | KiloRuntime + | Path.Path + | ServerConfig + | ServerSettingsService; + +const withIdentity = + (input: { + readonly instanceId: ProviderInstance["instanceId"]; + readonly displayName: string | undefined; + readonly accentColor: string | undefined; + readonly continuationKey: string; + }) => + (snapshot: ServerProviderDraft): ServerProvider => ({ + ...snapshot, + instanceId: input.instanceId, + driver: DRIVER_KIND, + ...(input.displayName ? { displayName: input.displayName } : {}), + ...(input.accentColor ? { accentColor: input.accentColor } : {}), + continuation: { groupKey: input.continuationKey }, + }); + +export const KiloDriver: ProviderDriver = { + driverKind: DRIVER_KIND, + metadata: { displayName: "Kilo", supportsMultipleInstances: true }, + configSchema: KiloSettings, + defaultConfig: () => decodeSettings({}), + create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => + Effect.gen(function* () { + const runtime = yield* KiloRuntime; + const serverConfig = yield* ServerConfig; + const settingsService = yield* ServerSettingsService; + const processEnvironment = mergeProviderInstanceEnvironment(environment); + const effectiveConfig = { ...config, enabled } satisfies KiloSettings; + const continuationIdentity = defaultProviderContinuationIdentity({ + driverKind: DRIVER_KIND, + instanceId, + }); + const stamp = withIdentity({ + instanceId, + displayName, + accentColor, + continuationKey: continuationIdentity.continuationKey, + }); + const adapter = yield* makeKiloAdapter(effectiveConfig, { + instanceId, + environment: processEnvironment, + }); + const textGeneration = yield* makeKiloTextGeneration(effectiveConfig, processEnvironment); + const checkProvider = checkKiloProviderStatus( + effectiveConfig, + serverConfig.cwd, + processEnvironment, + ).pipe(Effect.map(stamp), Effect.provideService(KiloRuntime, runtime)); + const snapshotSettings = makeProviderSnapshotSettingsSource(effectiveConfig, settingsService); + const snapshot = yield* makeManagedServerProvider>({ + maintenanceCapabilities: makeManualOnlyProviderMaintenanceCapabilities({ + provider: DRIVER_KIND, + packageName: "@kilocode/cli", + }), + getSettings: snapshotSettings.getSettings, + streamSettings: snapshotSettings.streamSettings, + haveSettingsChanged: haveProviderSnapshotSettingsChanged, + initialSnapshot: (settings) => + makePendingKiloProvider(settings.provider).pipe(Effect.map(stamp)), + checkProvider, + refreshInterval: REFRESH_INTERVAL, + }).pipe( + Effect.mapError( + (cause) => + new ProviderDriverError({ + driver: DRIVER_KIND, + instanceId, + detail: `Failed to build Kilo 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/KiloAdapter.test.ts b/apps/server/src/provider/Layers/KiloAdapter.test.ts new file mode 100644 index 00000000000..f027b1e3705 --- /dev/null +++ b/apps/server/src/provider/Layers/KiloAdapter.test.ts @@ -0,0 +1,510 @@ +import * as NodeAssert from "node:assert/strict"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { + KiloSettings, + ProviderDriverKind, + type ProviderRuntimeEvent, + ThreadId, +} from "@t3tools/contracts"; +import { it } from "@effect/vitest"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { beforeEach } from "vite-plus/test"; + +import { ServerConfig } from "../../config.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import { KiloRuntime, KiloRuntimeError, type KiloRuntimeShape } from "../kiloRuntime.ts"; +import type { KiloAdapterShape } from "../Services/KiloAdapter.ts"; +import { makeKiloAdapter } from "./KiloAdapter.ts"; + +class KiloAdapter extends Context.Service()( + "t3/provider/Layers/KiloAdapter.test/KiloAdapter", +) {} + +const asThreadId = (value: string): ThreadId => ThreadId.make(value); + +const runtimeMock = { + state: { + startCalls: [] as string[], + sessionCreates: 0, + promptCalls: [] as Array, + abortCalls: [] as string[], + subscribedEvents: [] as unknown[], + }, + reset() { + this.state.startCalls.length = 0; + this.state.sessionCreates = 0; + this.state.promptCalls.length = 0; + this.state.abortCalls.length = 0; + this.state.subscribedEvents.length = 0; + }, +}; + +const KiloRuntimeTestDouble: KiloRuntimeShape = { + startServer: ({ binaryPath }) => + Effect.gen(function* () { + runtimeMock.state.startCalls.push(binaryPath); + return { + url: "http://127.0.0.1:4301", + external: false as const, + exitCode: Effect.never, + }; + }), + runCommand: () => Effect.succeed({ stdout: "", stderr: "", code: 0 }), + createClient: () => + ({ + session: { + create: async () => { + runtimeMock.state.sessionCreates += 1; + return { data: { id: "kilo-session-1" } }; + }, + abort: async ({ sessionID }: { sessionID: string }) => { + runtimeMock.state.abortCalls.push(sessionID); + }, + promptAsync: async (input: unknown) => { + runtimeMock.state.promptCalls.push(input); + }, + messages: async () => ({ data: [] }), + revert: async () => {}, + }, + event: { + subscribe: async () => ({ + stream: (async function* () { + for (const event of runtimeMock.state.subscribedEvents) { + yield event; + } + })(), + }), + }, + permission: { + reply: async () => {}, + }, + question: { + reply: async () => {}, + }, + }) as unknown as ReturnType, + loadInventory: () => + Effect.fail( + new KiloRuntimeError({ + operation: "loadInventory", + detail: "KiloRuntimeTestDouble.loadInventory not used in this test", + cause: null, + }), + ), +}; + +const kiloAdapterTestSettings = Schema.decodeSync(KiloSettings)({ + binaryPath: "fake-kilo", +}); + +const KiloAdapterTestLayer = Layer.effect( + KiloAdapter, + makeKiloAdapter(kiloAdapterTestSettings), +).pipe( + Layer.provideMerge(Layer.succeed(KiloRuntime, KiloRuntimeTestDouble)), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(ServerSettingsService.layerTest()), + Layer.provideMerge(NodeServices.layer), +); + +beforeEach(() => { + runtimeMock.reset(); +}); + +const advanceTestClock = (ms: number) => + TestClock.adjust(`${ms} millis`).pipe(Effect.andThen(Effect.yieldNow)); + +const collectThreadEvents = (adapter: KiloAdapterShape, threadId: ThreadId) => + Effect.gen(function* () { + const collected: Array = []; + const drain = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.runForEach((event) => + Effect.sync(() => { + collected.push(event); + }), + ), + Effect.forkChild, + ); + return { collected, drain }; + }); + +const stopDrain = (drain: Fiber.Fiber) => Fiber.interrupt(drain); + +it.layer(KiloAdapterTestLayer)("KiloAdapterLive", (it) => { + it.effect( + "suppresses Kilo SDK synthetic snapshot-progress parts from the runtime event stream", + () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-snapshot-progress"); + const assistantMessageId = "msg-kilo-snapshot-progress"; + const snapshotPartId = "part-kilo-snapshot-progress"; + + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "kilo-session-1", + info: { id: assistantMessageId, role: "assistant" }, + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "kilo-session-1", + time: 1, + part: { + id: snapshotPartId, + sessionID: "kilo-session-1", + messageID: assistantMessageId, + type: "text", + text: "⠋ Initializing snapshot…", + }, + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "kilo-session-1", + time: 2, + part: { + id: snapshotPartId, + sessionID: "kilo-session-1", + messageID: assistantMessageId, + type: "text", + text: "⠙ Initializing snapshot…", + }, + }, + }, + ]; + + const { collected, drain } = yield* collectThreadEvents(adapter, threadId); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + yield* advanceTestClock(50); + + const assistantTextDeltas = collected.filter( + (event) => + event.type === "content.delta" && event.payload.streamKind === "assistant_text", + ) as Array>; + + NodeAssert.deepEqual( + assistantTextDeltas.map((event) => event.payload.delta), + [], + "snapshot-progress text must not produce assistant_text content deltas", + ); + + NodeAssert.equal( + collected.some( + (event) => + event.type === "item.completed" && + event.payload.itemType === "assistant_message" && + typeof event.payload.detail === "string" && + event.payload.detail.includes("Initializing snapshot"), + ), + false, + "snapshot-progress text must not produce assistant_message item.completed events", + ); + + yield* stopDrain(drain); + }), + ); + + it.effect("still surfaces real assistant text from the same assistant message", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-real-text"); + const assistantMessageId = "msg-kilo-real-text"; + const realPartId = "part-kilo-real-text"; + + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "kilo-session-1", + info: { id: assistantMessageId, role: "assistant" }, + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "kilo-session-1", + time: 1, + part: { + id: realPartId, + sessionID: "kilo-session-1", + messageID: assistantMessageId, + type: "text", + text: "Hello there", + }, + }, + }, + ]; + + const { collected, drain } = yield* collectThreadEvents(adapter, threadId); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + yield* advanceTestClock(50); + + const assistantTextDeltas = collected.filter( + (event) => event.type === "content.delta" && event.payload.streamKind === "assistant_text", + ) as Array>; + + NodeAssert.deepEqual( + assistantTextDeltas.map((event) => event.payload.delta), + ["Hello there"], + ); + + yield* stopDrain(drain); + }), + ); + + it.effect("clears local per-part bookkeeping when the SDK removes a part", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-part-removed"); + const assistantMessageId = "msg-kilo-part-removed"; + const partId = "part-kilo-part-removed"; + + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "kilo-session-1", + info: { id: assistantMessageId, role: "assistant" }, + }, + }, + { + type: "message.part.updated", + properties: { + sessionID: "kilo-session-1", + time: 1, + part: { + id: partId, + sessionID: "kilo-session-1", + messageID: assistantMessageId, + type: "text", + text: "Hi", + time: { start: 1, end: 2 }, + }, + }, + }, + { + type: "message.part.removed", + properties: { + sessionID: "kilo-session-1", + messageID: assistantMessageId, + partID: partId, + }, + }, + ]; + + const { collected, drain } = yield* collectThreadEvents(adapter, threadId); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + yield* advanceTestClock(50); + + // Removal should be a quiet, local-only cleanup: the pre-removal text + // update still produces its normal completion event, but no extra + // runtime event is emitted for the removal itself. + NodeAssert.equal( + collected.some( + (event) => + event.type === "item.completed" && + event.payload.itemType === "assistant_message" && + event.payload.detail === "Hi", + ), + true, + "the pre-removal text update must still produce an assistant_message item.completed", + ); + NodeAssert.deepEqual( + collected.map((event) => event.type), + ["session.started", "thread.started", "content.delta", "item.completed"], + "the removal must not produce any extra runtime events", + ); + + yield* stopDrain(drain); + }), + ); + + it.effect( + "backfills assistant text when message.part.updated arrives before message.updated", + () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-backfill"); + const assistantMessageId = "msg-kilo-backfill"; + const backfillPartId = "part-kilo-backfill"; + + runtimeMock.state.subscribedEvents = [ + { + // The part update arrives first, before the SDK has told us the + // role of the message it belongs to. + type: "message.part.updated", + properties: { + sessionID: "kilo-session-1", + time: 1, + part: { + id: backfillPartId, + sessionID: "kilo-session-1", + messageID: assistantMessageId, + type: "text", + text: "Hello there", + time: { start: 1, end: 2 }, + }, + }, + }, + { + type: "message.updated", + properties: { + sessionID: "kilo-session-1", + info: { id: assistantMessageId, role: "assistant" }, + }, + }, + ]; + + const { collected, drain } = yield* collectThreadEvents(adapter, threadId); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + yield* advanceTestClock(50); + + const assistantTextDeltas = collected.filter( + (event) => + event.type === "content.delta" && event.payload.streamKind === "assistant_text", + ) as Array>; + + NodeAssert.deepEqual( + assistantTextDeltas.map((event) => event.payload.delta), + ["Hello there"], + "the assistant text part must be surfaced once via backfill even when the role was unknown at update time", + ); + + const completed = collected.find( + (event) => + event.type === "item.completed" && event.payload.itemType === "assistant_message", + ); + NodeAssert.equal(completed?.type, "item.completed"); + if (completed?.type === "item.completed") { + NodeAssert.equal(completed.payload.detail, "Hello there"); + } + + yield* stopDrain(drain); + }), + ); + + it.effect("backfill never re-emits a stale snapshot that already streamed past it", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-stale-backfill"); + const assistantMessageId = "msg-kilo-stale-backfill"; + const stalePartId = "part-kilo-stale-backfill"; + + // Sequence: a message.part.updated captures a stale "Hello" snapshot + // (no role yet — emitText is skipped). Later deltas stream past + // (" world" then "!"). When the role finally arrives via + // message.updated, the backfill walks partById and re-invokes + // emitText with the stale snapshot. Without the staleness guard + // emitText would fall back to `delta = text` and re-emit "Hello", + // garbling the already-streamed output. With the guard, the stale + // snapshot is dropped and only the deltas are surfaced. + runtimeMock.state.subscribedEvents = [ + { + type: "message.part.updated", + properties: { + sessionID: "kilo-session-1", + time: 1, + part: { + id: stalePartId, + sessionID: "kilo-session-1", + messageID: assistantMessageId, + type: "text", + text: "Hello", + }, + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "kilo-session-1", + messageID: assistantMessageId, + partID: stalePartId, + field: "text", + delta: " world", + }, + }, + { + type: "message.part.delta", + properties: { + sessionID: "kilo-session-1", + messageID: assistantMessageId, + partID: stalePartId, + field: "text", + delta: "!", + }, + }, + { + type: "message.updated", + properties: { + sessionID: "kilo-session-1", + info: { id: assistantMessageId, role: "assistant" }, + }, + }, + ]; + + const { collected, drain } = yield* collectThreadEvents(adapter, threadId); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + yield* advanceTestClock(50); + + const assistantTextDeltas = collected.filter( + (event) => event.type === "content.delta" && event.payload.streamKind === "assistant_text", + ) as Array>; + + // The streamed output must NOT contain a duplicate "Hello" prefix + // re-emitted from the stale snapshot. + NodeAssert.equal( + assistantTextDeltas.some((event) => event.payload.delta === "Hello"), + false, + "the stale 'Hello' snapshot must not be re-emitted over already-streamed text", + ); + NodeAssert.deepEqual( + assistantTextDeltas.map((event) => event.payload.delta), + [" world", "!"], + "only the deltas after the stale snapshot must be surfaced", + ); + + yield* stopDrain(drain); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/KiloAdapter.ts b/apps/server/src/provider/Layers/KiloAdapter.ts new file mode 100644 index 00000000000..ee41775ab2f --- /dev/null +++ b/apps/server/src/provider/Layers/KiloAdapter.ts @@ -0,0 +1,1054 @@ +import { + EventId, + type KiloSettings, + ProviderDriverKind, + ProviderInstanceId, + type ProviderRuntimeEvent, + type ProviderSession, + RuntimeItemId, + RuntimeRequestId, + ThreadId, + type ToolLifecycleItemType, + TurnId, + type UserInputQuestion, +} from "@t3tools/contracts"; +import type { + GlobalEvent, + KiloClient, + Part, + PermissionRequest, + QuestionRequest, +} from "@kilocode/sdk/v2"; +import * as Cause from "effect/Cause"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +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 { resolveAttachmentPath } from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; +import { + ProviderAdapterProcessError, + ProviderAdapterRequestError, + ProviderAdapterSessionNotFoundError, + ProviderAdapterValidationError, +} from "../Errors.ts"; +import { + buildKiloPermissionRules, + kiloQuestionId, + KiloRuntime, + KiloRuntimeError, + kiloRuntimeErrorDetail, + parseKiloModelSlug, + runKiloSdk, + toKiloFileParts, + toKiloPermissionReply, + toKiloQuestionAnswers, + type KiloServerConnection, +} from "../kiloRuntime.ts"; +import { isKiloSnapshotProgressText } from "../kiloSnapshotProgressFilter.ts"; +import type { KiloAdapterShape } from "../Services/KiloAdapter.ts"; + +const PROVIDER = ProviderDriverKind.make("kilo"); +const FIXED_AGENT = "code"; +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +type KiloEvent = GlobalEvent["payload"]; + +interface KiloTurnSnapshot { + readonly id: TurnId; + readonly items: Array; +} + +interface KiloSessionContext { + session: ProviderSession; + readonly client: KiloClient; + readonly server: KiloServerConnection; + readonly directory: string; + readonly kiloSessionId: string; + readonly pendingPermissions: Map; + readonly pendingQuestions: Map; + readonly messageRoleById: Map; + readonly partById: Map; + readonly emittedTextByPartId: Map; + readonly completedAssistantPartIds: Set; + readonly interruptedTurnIds: Set; + readonly turns: Array; + activeTurnId: TurnId | undefined; + readonly stopped: Ref.Ref; + readonly scope: Scope.Closeable; +} + +export interface KiloAdapterOptions { + readonly instanceId?: ProviderInstanceId; + readonly environment?: NodeJS.ProcessEnv; +} + +function ensureContext( + sessions: ReadonlyMap, + threadId: ThreadId, +): KiloSessionContext { + const context = sessions.get(threadId); + if (!context) throw new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId }); + return context; +} + +function lookupContext( + sessions: ReadonlyMap, + threadId: ThreadId, +): Effect.Effect { + const context = sessions.get(threadId); + return context + ? Effect.succeed(context) + : Effect.fail(new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId })); +} + +function textFromPart(part: Part): string | undefined { + return part.type === "text" || part.type === "reasoning" ? part.text : undefined; +} + +function mapToolType(tool: string): ToolLifecycleItemType { + const value = tool.toLowerCase(); + if (value.includes("bash") || value.includes("shell") || value.includes("command")) + return "command_execution"; + if (value.includes("write") || value.includes("edit") || value.includes("patch")) + return "file_change"; + return "dynamic_tool_call"; +} + +function mapPermissionType(permission: string) { + const value = permission.toLowerCase(); + if (value.includes("bash") || value.includes("shell") || value.includes("command")) + return "command_execution_approval" as const; + if (value.includes("read")) return "file_read_approval" as const; + if (value.includes("write") || value.includes("edit")) return "file_change_approval" as const; + return "unknown" as const; +} + +function normalizeQuestions(request: QuestionRequest): ReadonlyArray { + return request.questions.map((question, index) => ({ + id: kiloQuestionId(index, question), + header: question.header, + question: question.question, + options: question.options ?? [], + ...(question.multiple ? { multiSelect: true } : {}), + })); +} + +function sessionErrorMessage(error: unknown): string { + if ( + error && + typeof error === "object" && + "message" in error && + typeof error.message === "string" + ) { + return error.message; + } + return "Kilo session failed."; +} + +export function makeKiloAdapter(settings: KiloSettings, options?: KiloAdapterOptions) { + return Effect.gen(function* () { + const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("kilo"); + const serverConfig = yield* ServerConfig; + const runtime = yield* KiloRuntime; + const crypto = yield* Crypto.Crypto; + const events = yield* Queue.unbounded(); + const sessions = new Map(); + const threadLocksRef = yield* SynchronizedRef.make(new Map()); + const getThreadSemaphore = (threadId: ThreadId) => + SynchronizedRef.modifyEffect(threadLocksRef, (current) => { + const existing = Option.fromNullishOr(current.get(threadId)); + return Option.match(existing, { + onNone: () => + Semaphore.make(1).pipe( + Effect.map((semaphore) => { + const next = new Map(current); + next.set(threadId, semaphore); + return [semaphore, next] as const; + }), + ), + onSome: (semaphore) => Effect.succeed([semaphore, current] as const), + }); + }); + const withThreadLock = (threadId: ThreadId, effect: Effect.Effect) => + Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect)); + const randomId = crypto.randomUUIDv4.pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "crypto/randomUUIDv4", + detail: "Failed to generate Kilo runtime identifier.", + cause, + }), + ), + ); + const eventBase = (input: { + readonly threadId: ThreadId; + readonly turnId?: TurnId; + readonly itemId?: string; + readonly requestId?: string; + readonly raw?: unknown; + }) => + Effect.all({ eventId: randomId.pipe(Effect.map(EventId.make)), createdAt: nowIso }).pipe( + Effect.map(({ eventId, createdAt }) => ({ + eventId, + provider: PROVIDER, + providerInstanceId: boundInstanceId, + threadId: input.threadId, + createdAt, + ...(input.turnId ? { turnId: input.turnId } : {}), + ...(input.itemId ? { itemId: RuntimeItemId.make(input.itemId) } : {}), + ...(input.requestId ? { requestId: RuntimeRequestId.make(input.requestId) } : {}), + ...(input.raw !== undefined + ? { raw: { source: "kilo.sdk.event" as const, payload: input.raw } } + : {}), + })), + ); + const emit = (event: ProviderRuntimeEvent) => Queue.offer(events, event).pipe(Effect.asVoid); + + const updateSession = ( + context: KiloSessionContext, + patch: Partial, + clearActive = false, + ) => + nowIso.pipe( + Effect.map((updatedAt) => { + context.session = { + ...context.session, + ...patch, + ...(clearActive ? { activeTurnId: undefined } : {}), + updatedAt, + }; + return context.session; + }), + ); + + const stopContext = Effect.fn("stopKiloContext")(function* (context: KiloSessionContext) { + if (yield* Ref.getAndSet(context.stopped, true)) return false; + yield* runKiloSdk("session.abort", () => + context.client.session.abort({ sessionID: context.kiloSessionId }, { throwOnError: true }), + ).pipe(Effect.ignore); + yield* Scope.close(context.scope, Exit.void).pipe(Effect.ignore); + return true; + }); + + yield* Effect.addFinalizer(() => + Effect.forEach( + [...sessions.values()], + (context) => Effect.ignoreCause(stopContext(context)), + { + concurrency: "unbounded", + discard: true, + }, + ).pipe(Effect.ensuring(Queue.shutdown(events))), + ); + + const emitText = Effect.fn("emitKiloText")(function* ( + context: KiloSessionContext, + part: Part, + raw: unknown, + ) { + const text = textFromPart(part); + if (text === undefined) return; + const previous = context.emittedTextByPartId.get(part.id) ?? ""; + // The Kilo SDK emits synthetic spinner text parts while it initializes + // its file snapshot — suppress them so they never reach the ingestion + // layer or the chat UI, and do NOT record them in the user-visible + // accumulator (the user never saw them). See + // `kiloSnapshotProgressFilter.ts`. Do not add the part id to + // `completedAssistantPartIds` so a real completion is still surfaced. + if (isKiloSnapshotProgressText(text)) return; + // Pick the next content.delta: + // - If the snapshot extends the previously-emitted text, emit the + // new tail and update the accumulator. + // - If nothing has been emitted to the user yet (or the prior + // content was suppressed as snapshot progress), emit the full + // text as the initial visible chunk and update the accumulator. + // - If the snapshot is a full replacement that is neither a strict + // extension nor a stale snapshot, emit the full text as a + // replacement and update the accumulator. + // - If the snapshot is shorter than what we already streamed, it is + // stale relative to the user-visible stream — drop it so we never + // garble or duplicate the already-emitted output. + let delta = ""; + if (text.length < previous.length) { + delta = ""; + } else if (text.startsWith(previous) && text.length > previous.length) { + delta = text.slice(previous.length); + } else { + delta = text; + } + context.emittedTextByPartId.set(part.id, text); + if (delta) { + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(context.activeTurnId ? { turnId: context.activeTurnId } : {}), + itemId: part.id, + raw, + })), + type: "content.delta", + payload: { + streamKind: part.type === "reasoning" ? "reasoning_text" : "assistant_text", + delta, + }, + }); + } + if ( + part.type === "text" && + part.time?.end !== undefined && + !context.completedAssistantPartIds.has(part.id) + ) { + context.completedAssistantPartIds.add(part.id); + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(context.activeTurnId ? { turnId: context.activeTurnId } : {}), + itemId: part.id, + raw, + })), + type: "item.completed", + payload: { + itemType: "assistant_message", + status: "completed", + title: "Assistant message", + ...(text ? { detail: text } : {}), + }, + }); + } + }); + + const handleEvent = Effect.fn("handleKiloEvent")(function* ( + context: KiloSessionContext, + event: KiloEvent, + ) { + if (!("properties" in event) || !("sessionID" in event.properties)) return; + if (event.properties.sessionID !== context.kiloSessionId) return; + const turnId = context.activeTurnId; + switch (event.type) { + case "message.updated": { + const messageId = event.properties.info.id; + const role = event.properties.info.role; + const learnedRole = !context.messageRoleById.has(messageId); + context.messageRoleById.set(messageId, role); + // Backfill any text/reasoning parts whose message role was still + // unknown when their initial message.part.updated arrived. By the + // time the role arrives, deltas may have already streamed into + // `emittedTextByPartId`, so a stale `partById[id].text` snapshot + // must NOT replace that accumulated output. Only emit a + // content.delta when nothing has been streamed yet; in either case + // still surface a missing item.completed so the chat UI knows the + // turn has a finished assistant message. + if (role === "assistant" && learnedRole) { + for (const part of context.partById.values()) { + if (part.messageID !== messageId) continue; + if (part.type !== "text" && part.type !== "reasoning") continue; + yield* emitText(context, part, event); + } + } + break; + } + case "message.part.delta": { + if (event.properties.field !== "text" || !event.properties.delta) break; + const part = context.partById.get(event.properties.partID); + if (!part || (part.type !== "text" && part.type !== "reasoning")) break; + const previous = context.emittedTextByPartId.get(part.id) ?? ""; + const accumulated = previous + event.properties.delta; + // Suppress Kilo SDK synthetic snapshot-progress deltas. The + // accumulator is only updated when the new text is also emitted to + // the user, so subsequent edits compute deltas against the + // user-visible stream rather than the raw SDK truth. See + // `kiloSnapshotProgressFilter.ts`. + if (isKiloSnapshotProgressText(accumulated)) break; + context.emittedTextByPartId.set(part.id, accumulated); + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + itemId: part.id, + raw: event, + })), + type: "content.delta", + payload: { + streamKind: part.type === "reasoning" ? "reasoning_text" : "assistant_text", + delta: event.properties.delta, + }, + }); + break; + } + case "message.part.removed": { + // The SDK removes parts when they are deleted from the message + // (e.g. when the snapshot-progress spinner finishes initializing). + // Drop our per-part bookkeeping so we don't leak entries over time. + // No runtime event is emitted here — removal is purely local state + // cleanup, matching how other adapters handle deleted parts. + context.partById.delete(event.properties.partID); + context.emittedTextByPartId.delete(event.properties.partID); + context.completedAssistantPartIds.delete(event.properties.partID); + break; + } + case "message.part.updated": { + const part = event.properties.part; + context.partById.set(part.id, part); + const role = context.messageRoleById.get(part.messageID); + if (role === "assistant") yield* emitText(context, part, event); + if (part.type === "tool") { + const status = + part.state.status === "error" + ? "failed" + : part.state.status === "completed" + ? "completed" + : "inProgress"; + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + itemId: part.callID, + raw: event, + })), + type: + part.state.status === "pending" + ? "item.started" + : part.state.status === "completed" || part.state.status === "error" + ? "item.completed" + : "item.updated", + payload: { + itemType: mapToolType(part.tool), + status, + title: + part.state.status === "running" ? (part.state.title ?? part.tool) : part.tool, + data: { tool: part.tool, state: part.state }, + }, + }); + } + break; + } + case "permission.asked": + context.pendingPermissions.set(event.properties.id, event.properties); + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + requestId: event.properties.id, + raw: event, + })), + type: "request.opened", + payload: { + requestType: mapPermissionType(event.properties.permission), + detail: event.properties.patterns.join("\n") || event.properties.permission, + args: event.properties.metadata, + }, + }); + break; + case "permission.replied": + context.pendingPermissions.delete(event.properties.requestID); + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + requestId: event.properties.requestID, + raw: event, + })), + type: "request.resolved", + payload: { requestType: "unknown", decision: event.properties.reply }, + }); + break; + case "question.asked": + context.pendingQuestions.set(event.properties.id, event.properties); + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + requestId: event.properties.id, + raw: event, + })), + type: "user-input.requested", + payload: { questions: normalizeQuestions(event.properties) }, + }); + break; + case "question.replied": + case "question.rejected": { + const request = context.pendingQuestions.get(event.properties.requestID); + context.pendingQuestions.delete(event.properties.requestID); + const answers = + event.type === "question.replied" + ? Object.fromEntries( + (request?.questions ?? []).map((question, index) => [ + kiloQuestionId(index, question), + event.properties.answers[index]?.join(", ") ?? "", + ]), + ) + : {}; + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + requestId: event.properties.requestID, + raw: event, + })), + type: "user-input.resolved", + payload: { answers }, + }); + break; + } + case "todo.updated": + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + raw: event, + })), + type: "turn.plan.updated", + payload: { + plan: event.properties.todos.map((todo) => ({ + step: todo.content, + status: + todo.status === "completed" + ? "completed" + : todo.status === "in_progress" + ? "inProgress" + : "pending", + })), + }, + }); + break; + case "session.status": + if (event.properties.status.type === "busy") { + yield* updateSession(context, { status: "running", activeTurnId: turnId }); + } else if (event.properties.status.type === "retry") { + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + raw: event, + })), + type: "runtime.warning", + payload: { + message: event.properties.status.message, + detail: event.properties.status, + }, + }); + } else if (turnId) { + const state = context.interruptedTurnIds.delete(turnId) ? "interrupted" : "completed"; + context.activeTurnId = undefined; + yield* updateSession(context, { status: "ready" }, true); + yield* emit({ + ...(yield* eventBase({ threadId: context.session.threadId, turnId, raw: event })), + type: "turn.completed", + payload: { state }, + }); + } + break; + case "session.error": { + const message = sessionErrorMessage(event.properties.error); + context.activeTurnId = undefined; + yield* updateSession(context, { status: "error", lastError: message }, true); + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + ...(turnId ? { turnId } : {}), + raw: event, + })), + type: "runtime.error", + payload: { message, class: "provider_error", detail: event.properties.error }, + }); + if (turnId) { + yield* emit({ + ...(yield* eventBase({ threadId: context.session.threadId, turnId, raw: event })), + type: "turn.completed", + payload: { state: "failed", errorMessage: message }, + }); + } + break; + } + default: + break; + } + }); + + const startPump = Effect.fn("startKiloPump")(function* (context: KiloSessionContext) { + const abort = new AbortController(); + yield* Scope.addFinalizer( + context.scope, + Effect.sync(() => abort.abort()), + ); + yield* runKiloSdk("event.subscribe", () => + context.client.event.subscribe(undefined, { signal: abort.signal, throwOnError: true }), + ).pipe( + Effect.flatMap((subscription) => + Stream.fromAsyncIterable( + subscription.stream, + (cause) => + new KiloRuntimeError({ + operation: "event.subscribe", + detail: kiloRuntimeErrorDetail(cause), + cause, + }), + ).pipe(Stream.runForEach((event) => handleEvent(context, event))), + ), + Effect.catchCause((cause) => + abort.signal.aborted + ? Effect.void + : Effect.gen(function* () { + const message = kiloRuntimeErrorDetail(Cause.squash(cause)); + const activeTurnId = context.activeTurnId; + context.activeTurnId = undefined; + yield* Ref.set(context.stopped, true); + yield* updateSession(context, { status: "error", lastError: message }, true); + sessions.delete(context.session.threadId); + if (activeTurnId) { + yield* emit({ + ...(yield* eventBase({ + threadId: context.session.threadId, + turnId: activeTurnId, + })), + type: "turn.completed", + payload: { state: "failed", errorMessage: message }, + }); + } + yield* emit({ + ...(yield* eventBase({ threadId: context.session.threadId })), + type: "runtime.error", + payload: { message, class: "transport_error" }, + }); + yield* emit({ + ...(yield* eventBase({ threadId: context.session.threadId })), + type: "session.exited", + payload: { + reason: message, + recoverable: true, + exitKind: "error", + }, + }); + yield* Scope.close(context.scope, Exit.void).pipe(Effect.ignore, Effect.forkDetach); + }), + ), + Effect.forkIn(context.scope), + ); + }); + + const startSession: KiloAdapterShape["startSession"] = Effect.fn("startKiloSession")( + function* (input) { + return yield* withThreadLock( + input.threadId, + Effect.gen(function* () { + const existing = sessions.get(input.threadId); + if (existing) { + yield* stopContext(existing); + sessions.delete(input.threadId); + } + const directory = input.cwd ?? serverConfig.cwd; + const scope = yield* Scope.make(); + const started = yield* Effect.exit( + Effect.gen(function* () { + const server = yield* runtime.startServer({ + binaryPath: settings.binaryPath, + ...(options?.environment ? { environment: options.environment } : {}), + }); + const client = runtime.createClient({ baseUrl: server.url, directory }); + const model = parseKiloModelSlug(input.modelSelection?.model); + const created = yield* runKiloSdk("session.create", () => + client.session.create( + { + title: `T3 Code ${input.threadId}`, + agent: FIXED_AGENT, + ...(model + ? { model: { id: model.modelID, providerID: model.providerID } } + : {}), + permission: buildKiloPermissionRules(input.runtimeMode), + platform: "t3code", + }, + { throwOnError: true }, + ), + ); + if (!created.data) { + return yield* new KiloRuntimeError({ + operation: "session.create", + detail: "Kilo session.create returned no session payload.", + }); + } + return { server, client, session: created.data }; + }).pipe(Effect.provideService(Scope.Scope, scope)), + ); + if (Exit.isFailure(started)) { + yield* Scope.close(scope, Exit.void).pipe(Effect.ignore); + const cause = Cause.squash(started.cause); + return yield* new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: kiloRuntimeErrorDetail(cause), + cause, + }); + } + const createdAt = yield* nowIso; + const session: ProviderSession = { + provider: PROVIDER, + providerInstanceId: boundInstanceId, + threadId: input.threadId, + status: "ready", + runtimeMode: input.runtimeMode, + cwd: directory, + ...(input.modelSelection ? { model: input.modelSelection.model } : {}), + createdAt, + updatedAt: createdAt, + }; + const context: KiloSessionContext = { + session, + client: started.value.client, + server: started.value.server, + directory, + kiloSessionId: started.value.session.id, + pendingPermissions: new Map(), + pendingQuestions: new Map(), + messageRoleById: new Map(), + partById: new Map(), + emittedTextByPartId: new Map(), + completedAssistantPartIds: new Set(), + interruptedTurnIds: new Set(), + turns: [], + activeTurnId: undefined, + stopped: yield* Ref.make(false), + scope, + }; + sessions.set(input.threadId, context); + yield* startPump(context); + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId })), + type: "session.started", + payload: { message: "Kilo session started" }, + }); + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId })), + type: "thread.started", + payload: { providerThreadId: started.value.session.id }, + }); + return session; + }), + ); + }, + ); + + const sendTurn: KiloAdapterShape["sendTurn"] = Effect.fn("sendKiloTurn")(function* (input) { + const context = ensureContext(sessions, input.threadId); + const turnId = context.activeTurnId ?? TurnId.make(`kilo-turn-${yield* randomId}`); + const selection = + input.modelSelection ?? + (context.session.model + ? { instanceId: boundInstanceId, model: context.session.model } + : undefined); + if (!selection || selection.instanceId !== boundInstanceId) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: `Kilo model selection must target instance '${boundInstanceId}'.`, + }); + } + const model = parseKiloModelSlug(selection.model); + if (!model) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Kilo model selection must use the 'provider/model' format.", + }); + } + const text = input.input?.trim(); + const files = toKiloFileParts({ + attachments: input.attachments, + resolveAttachmentPath: (attachment) => + resolveAttachmentPath({ attachmentsDir: serverConfig.attachmentsDir, attachment }), + }); + if (files.length !== (input.attachments?.length ?? 0)) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Kilo could not resolve one or more attachments.", + }); + } + if (!text && files.length === 0) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Kilo turns require text input or at least one attachment.", + }); + } + const freshTurn = context.activeTurnId === undefined; + context.activeTurnId = turnId; + yield* updateSession(context, { + status: "running", + activeTurnId: turnId, + model: selection.model, + }); + if (freshTurn) { + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId })), + type: "turn.started", + payload: { model: selection.model }, + }); + } + yield* runKiloSdk("session.promptAsync", () => + context.client.session.promptAsync( + { + sessionID: context.kiloSessionId, + model, + agent: input.interactionMode === "plan" ? "plan" : FIXED_AGENT, + parts: [...(text ? [{ type: "text" as const, text }] : []), ...files], + }, + { throwOnError: true }, + ), + ).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: cause.operation, + detail: cause.detail, + cause, + }), + ), + Effect.tapError((requestError) => + freshTurn + ? Effect.gen(function* () { + context.activeTurnId = undefined; + yield* updateSession( + context, + { status: "ready", lastError: requestError.detail }, + true, + ); + yield* emit({ + ...(yield* eventBase({ + threadId: input.threadId, + turnId, + raw: requestError, + })), + type: "turn.completed", + payload: { state: "failed", errorMessage: requestError.detail }, + }); + }) + : Effect.void, + ), + ); + return { threadId: input.threadId, turnId }; + }); + + const interruptTurn: KiloAdapterShape["interruptTurn"] = (threadId) => + Effect.gen(function* () { + const context = yield* lookupContext(sessions, threadId); + const turnId = context.activeTurnId; + if (turnId) context.interruptedTurnIds.add(turnId); + return yield* runKiloSdk("session.abort", () => + context.client.session.abort( + { sessionID: context.kiloSessionId }, + { throwOnError: true }, + ), + ).pipe( + Effect.tapError(() => + Effect.sync(() => turnId && context.interruptedTurnIds.delete(turnId)), + ), + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: cause.operation, + detail: cause.detail, + cause, + }), + ), + ); + }); + + const respondToRequest: KiloAdapterShape["respondToRequest"] = ( + threadId, + requestId, + decision, + ) => + Effect.gen(function* () { + const context = yield* lookupContext(sessions, threadId); + if (!context.pendingPermissions.has(requestId)) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "permission.reply", + detail: `Unknown pending permission request: ${requestId}`, + }); + } + return yield* runKiloSdk("permission.reply", () => + context.client.permission.reply( + { requestID: requestId, reply: toKiloPermissionReply(decision) }, + { throwOnError: true }, + ), + ).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: cause.operation, + detail: cause.detail, + cause, + }), + ), + Effect.asVoid, + ); + }); + + const respondToUserInput: KiloAdapterShape["respondToUserInput"] = ( + threadId, + requestId, + answers, + ) => + Effect.gen(function* () { + const context = yield* lookupContext(sessions, threadId); + const request = context.pendingQuestions.get(requestId); + if (!request) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "question.reply", + detail: `Unknown pending question request: ${requestId}`, + }); + } + return yield* runKiloSdk("question.reply", () => + context.client.question.reply( + { requestID: requestId, answers: toKiloQuestionAnswers(request, answers) }, + { throwOnError: true }, + ), + ).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: cause.operation, + detail: cause.detail, + cause, + }), + ), + Effect.asVoid, + ); + }); + + const stopSession: KiloAdapterShape["stopSession"] = Effect.fn("stopKiloSession")( + function* (threadId) { + return yield* withThreadLock( + threadId, + Effect.gen(function* () { + const context = sessions.get(threadId); + if (!context) return; + const stopped = yield* stopContext(context); + sessions.delete(threadId); + if (stopped) { + yield* emit({ + ...(yield* eventBase({ threadId })), + type: "session.exited", + payload: { + reason: "Session stopped.", + recoverable: false, + exitKind: "graceful", + }, + }); + } + }), + ); + }, + ); + + const readThread: KiloAdapterShape["readThread"] = Effect.fn("readKiloThread")( + function* (threadId) { + const context = ensureContext(sessions, threadId); + const response = yield* runKiloSdk("session.messages", () => + context.client.session.messages( + { sessionID: context.kiloSessionId }, + { throwOnError: true }, + ), + ).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: cause.operation, + detail: cause.detail, + cause, + }), + ), + ); + const turns: Array = []; + for (const entry of response.data ?? []) { + if (entry.info.role !== "assistant") continue; + turns.push({ id: TurnId.make(entry.info.id), items: [entry.info, ...entry.parts] }); + } + return { threadId, turns }; + }, + ); + + const rollbackThread: KiloAdapterShape["rollbackThread"] = Effect.fn("rollbackKiloThread")( + function* (threadId, numTurns) { + const context = ensureContext(sessions, threadId); + if (!Number.isInteger(numTurns) || numTurns < 1) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThread", + issue: "numTurns must be an integer >= 1.", + }); + } + const snapshot = yield* readThread(threadId); + if (numTurns > snapshot.turns.length) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "rollbackThread", + issue: "numTurns must not exceed the number of turns.", + }); + } + const target = snapshot.turns[snapshot.turns.length - numTurns - 1]; + yield* runKiloSdk("session.revert", () => + context.client.session.revert( + { sessionID: context.kiloSessionId, ...(target ? { messageID: target.id } : {}) }, + { throwOnError: true }, + ), + ).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: cause.operation, + detail: cause.detail, + cause, + }), + ), + ); + return yield* readThread(threadId); + }, + ); + + return { + provider: PROVIDER, + capabilities: { sessionModelSwitch: "in-session" }, + startSession, + sendTurn, + interruptTurn, + respondToRequest, + respondToUserInput, + stopSession, + listSessions: () => Effect.sync(() => [...sessions.values()].map((value) => value.session)), + hasSession: (threadId) => Effect.sync(() => sessions.has(threadId)), + readThread, + rollbackThread, + stopAll: () => + Effect.forEach( + [...sessions.values()], + (context) => Effect.ignoreCause(stopContext(context)), + { + concurrency: "unbounded", + discard: true, + }, + ).pipe(Effect.tap(() => Effect.sync(() => sessions.clear()))), + get streamEvents() { + return Stream.fromQueue(events); + }, + } satisfies KiloAdapterShape; + }); +} diff --git a/apps/server/src/provider/Layers/KiloProvider.test.ts b/apps/server/src/provider/Layers/KiloProvider.test.ts new file mode 100644 index 00000000000..22f2830432e --- /dev/null +++ b/apps/server/src/provider/Layers/KiloProvider.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from "@effect/vitest"; +import type { KiloSettings } from "@t3tools/contracts"; +import type { ProviderListResponse } from "@kilocode/sdk/v2"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { KiloRuntime, KiloRuntimeError, type KiloRuntimeShape } from "../kiloRuntime.ts"; +import { checkKiloProviderStatus, flattenKiloModels } from "./KiloProvider.ts"; + +const settings: KiloSettings = { + enabled: true, + binaryPath: "kilo", + customModels: [], +}; + +const inventory: ProviderListResponse = { + all: [ + { + id: "anthropic", + name: "Anthropic", + source: "api", + env: [], + options: {}, + models: { + sonnet: { + id: "claude-sonnet", + providerID: "anthropic", + api: { id: "claude-sonnet", url: "https://example.test", npm: "@ai-sdk/anthropic" }, + name: "Claude Sonnet", + capabilities: { + temperature: true, + reasoning: true, + attachment: true, + toolcall: true, + input: { text: true, audio: false, image: true, video: false, pdf: true }, + output: { text: true, audio: false, image: false, video: false, pdf: false }, + interleaved: false, + }, + cost: { input: 0, output: 0, cache: { read: 0, write: 0 } }, + limit: { context: 200_000, output: 16_000 }, + status: "active", + options: {}, + headers: {}, + release_date: "2026-01-01", + }, + }, + }, + { + id: "disconnected", + name: "Disconnected", + source: "api", + env: [], + options: {}, + models: {}, + }, + ], + default: {}, + connected: ["anthropic"], + failed: [], +}; + +describe("KiloProvider", () => { + it("flattens only connected upstream models into canonical provider/model slugs", () => { + expect(flattenKiloModels(inventory)).toEqual([ + expect.objectContaining({ + slug: "anthropic/claude-sonnet", + name: "Claude Sonnet", + subProvider: "Anthropic", + }), + ]); + }); + + it.effect("reports dynamic inventory and version through the Kilo snapshot", () => { + const runtime: KiloRuntimeShape = { + runCommand: () => Effect.succeed({ stdout: "7.4.5\n", stderr: "", code: 0 }), + startServer: () => + Effect.succeed({ + url: "http://127.0.0.1:4096", + external: false, + exitCode: Effect.succeed(0), + }), + createClient: () => ({}) as never, + loadInventory: () => Effect.succeed(inventory), + }; + return checkKiloProviderStatus(settings, process.cwd()).pipe( + Effect.provide(Layer.succeed(KiloRuntime, runtime)), + Effect.tap((snapshot) => + Effect.sync(() => { + expect(snapshot.installed).toBe(true); + expect(snapshot.version).toBe("7.4.5"); + expect(snapshot.status).toBe("ready"); + expect(snapshot.showInteractionModeToggle).toBe(true); + expect(snapshot.models.map((model) => model.slug)).toEqual(["anthropic/claude-sonnet"]); + }), + ), + ); + }); + + it.effect("distinguishes a missing Kilo binary", () => { + const runtime: KiloRuntimeShape = { + runCommand: () => + Effect.fail( + new KiloRuntimeError({ + operation: "runCommand", + detail: "spawn kilo ENOENT", + }), + ), + startServer: () => + Effect.fail(new KiloRuntimeError({ operation: "startServer", detail: "unused" })), + createClient: () => ({}) as never, + loadInventory: () => Effect.succeed(inventory), + }; + return checkKiloProviderStatus(settings, process.cwd()).pipe( + Effect.provide(Layer.succeed(KiloRuntime, runtime)), + Effect.tap((snapshot) => + Effect.sync(() => { + expect(snapshot.installed).toBe(false); + expect(snapshot.status).toBe("error"); + expect(snapshot.message).toContain("not installed"); + }), + ), + ); + }); +}); diff --git a/apps/server/src/provider/Layers/KiloProvider.ts b/apps/server/src/provider/Layers/KiloProvider.ts new file mode 100644 index 00000000000..faa82f28be6 --- /dev/null +++ b/apps/server/src/provider/Layers/KiloProvider.ts @@ -0,0 +1,154 @@ +import { + ProviderDriverKind, + type KiloSettings, + type ModelCapabilities, + type ServerProviderModel, +} from "@t3tools/contracts"; +import type { ProviderListResponse } from "@kilocode/sdk/v2"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { + buildServerProvider, + nonEmptyTrimmed, + parseGenericCliVersion, + providerModelsFromSettings, + type ServerProviderDraft, +} from "../providerSnapshot.ts"; +import { KiloRuntime, kiloRuntimeErrorDetail } from "../kiloRuntime.ts"; + +const PROVIDER = ProviderDriverKind.make("kilo"); +const PRESENTATION = { displayName: "Kilo", showInteractionModeToggle: true } as const; +const DEFAULT_CAPABILITIES: ModelCapabilities = { optionDescriptors: [] }; + +export function flattenKiloModels( + inventory: ProviderListResponse, +): ReadonlyArray { + const connected = new Set(inventory.connected); + const models: Array = []; + for (const provider of inventory.all) { + if (!connected.has(provider.id)) continue; + for (const model of Object.values(provider.models)) { + const name = nonEmptyTrimmed(model.name); + if (!name) continue; + const subProvider = nonEmptyTrimmed(provider.name); + models.push({ + slug: `${provider.id}/${model.id}`, + name, + ...(subProvider ? { subProvider } : {}), + isCustom: false, + capabilities: DEFAULT_CAPABILITIES, + }); + } + } + return models.toSorted((left, right) => left.name.localeCompare(right.name)); +} + +function modelsFromSettings(settings: KiloSettings, live: ReadonlyArray = []) { + return providerModelsFromSettings(live, PROVIDER, settings.customModels, DEFAULT_CAPABILITIES); +} + +export const makePendingKiloProvider = ( + settings: KiloSettings, +): Effect.Effect => + Effect.gen(function* () { + const checkedAt = DateTime.formatIso(yield* DateTime.now); + return buildServerProvider({ + presentation: PRESENTATION, + enabled: settings.enabled, + checkedAt, + models: modelsFromSettings(settings), + probe: settings.enabled + ? { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Kilo provider status has not been checked in this session yet.", + } + : { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Kilo is disabled in T3 Code settings.", + }, + }); + }); + +export const checkKiloProviderStatus = Effect.fn("checkKiloProviderStatus")(function* ( + settings: KiloSettings, + cwd: string, + environment?: NodeJS.ProcessEnv, +): Effect.fn.Return { + const runtime = yield* KiloRuntime; + const checkedAt = DateTime.formatIso(yield* DateTime.now); + const fallback = (cause: unknown, version: string | null = null) => { + const detail = kiloRuntimeErrorDetail(cause); + const missing = detail.toLowerCase().includes("enoent"); + return buildServerProvider({ + presentation: PRESENTATION, + enabled: settings.enabled, + checkedAt, + models: modelsFromSettings(settings), + probe: { + installed: !missing, + version, + status: "error", + auth: { status: "unknown" }, + message: missing + ? "Kilo CLI (`kilo`) is not installed or not on PATH." + : `Failed to initialize Kilo: ${detail}`, + }, + }); + }; + + if (!settings.enabled) return yield* makePendingKiloProvider(settings); + + const versionExit = yield* Effect.exit( + runtime.runCommand({ + binaryPath: settings.binaryPath, + args: ["--version"], + ...(environment ? { environment } : {}), + }), + ); + if (versionExit._tag === "Failure") return fallback(Cause.squash(versionExit.cause)); + const version = parseGenericCliVersion(versionExit.value.stdout) ?? null; + + const inventoryExit = yield* Effect.exit( + Effect.scoped( + Effect.gen(function* () { + const server = yield* runtime.startServer({ + binaryPath: settings.binaryPath, + ...(environment ? { environment } : {}), + }); + return yield* runtime.loadInventory( + runtime.createClient({ baseUrl: server.url, directory: cwd }), + ); + }), + ), + ); + if (inventoryExit._tag === "Failure") return fallback(Cause.squash(inventoryExit.cause), version); + + const connectedCount = inventoryExit.value.connected.length; + return buildServerProvider({ + presentation: PRESENTATION, + enabled: true, + checkedAt, + models: modelsFromSettings(settings, flattenKiloModels(inventoryExit.value)), + probe: { + installed: true, + version, + status: connectedCount > 0 ? "ready" : "warning", + auth: { + status: connectedCount > 0 ? "authenticated" : "unknown", + type: "kilo", + }, + message: + connectedCount > 0 + ? `${connectedCount} upstream provider${connectedCount === 1 ? "" : "s"} connected through Kilo.` + : "Kilo is available, but it did not report any connected upstream providers.", + }, + }); +}); diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index dbfa7faffea..886ba7a2813 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 KiloSettings, type OpenCodeSettings, ProviderDriverKind, type ProviderInstanceConfigMap, @@ -45,7 +46,10 @@ import { CodexDriver } from "../Drivers/CodexDriver.ts"; import { CursorDriver } from "../Drivers/CursorDriver.ts"; import { GrokDriver } from "../Drivers/GrokDriver.ts"; import { OpenCodeDriver } from "../Drivers/OpenCodeDriver.ts"; +import { KiloDriver } from "../Drivers/KiloDriver.ts"; import { OpenCodeRuntimeLive } from "../opencodeRuntime.ts"; +import type { BuiltInDriversEnv } from "../builtInDrivers.ts"; +import { KiloRuntimeLive } from "../kiloRuntime.ts"; import { NoOpProviderEventLoggers, ProviderEventLoggers } from "./ProviderEventLoggers.ts"; import { makeProviderInstanceRegistry } from "./ProviderInstanceRegistryLive.ts"; @@ -98,6 +102,13 @@ const makeOpenCodeConfig = (overrides: Partial): OpenCodeSetti ...overrides, }); +const makeKiloConfig = (overrides: Partial): KiloSettings => ({ + enabled: false, + binaryPath: "kilo", + customModels: [], + ...overrides, +}); + describe("ProviderInstanceRegistryLive — multi-instance codex slice", () => { // `ServerConfig.layerTest` needs `FileSystem` to materialize its scratch // directory. `Layer.merge` just unions requirements, so we have to push @@ -241,7 +252,9 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { // provides `OpenCodeRuntimeLive`'s deps while keeping its own outputs // surfaced; that merged layer then provides `ServerConfig.layerTest`'s // `FileSystem` dep while keeping everything else surfaced to the test. - const infraLayer = OpenCodeRuntimeLive.pipe(Layer.provideMerge(NodeServices.layer)); + const infraLayer = Layer.merge(OpenCodeRuntimeLive, KiloRuntimeLive).pipe( + Layer.provideMerge(NodeServices.layer), + ); const testLayer = ServerConfig.layerTest(process.cwd(), { prefix: "provider-instance-registry-all-drivers-test", }).pipe( @@ -258,12 +271,14 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { const cursorId = ProviderInstanceId.make("cursor_default"); const grokId = ProviderInstanceId.make("grok_default"); const openCodeId = ProviderInstanceId.make("opencode_default"); + const kiloId = ProviderInstanceId.make("kilo_default"); const codexDriverKind = ProviderDriverKind.make("codex"); const claudeDriverKind = ProviderDriverKind.make("claudeAgent"); const cursorDriverKind = ProviderDriverKind.make("cursor"); const grokDriverKind = ProviderDriverKind.make("grok"); const openCodeDriverKind = ProviderDriverKind.make("opencode"); + const kiloDriverKind = ProviderDriverKind.make("kilo"); const configMap: ProviderInstanceConfigMap = { [codexId]: { @@ -299,10 +314,16 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { enabled: false, config: makeOpenCodeConfig({}), }, + [kiloId]: { + driver: kiloDriverKind, + displayName: "Kilo", + enabled: false, + config: makeKiloConfig({}), + }, }; - const { registry } = yield* makeProviderInstanceRegistry({ - drivers: [CodexDriver, ClaudeDriver, CursorDriver, GrokDriver, OpenCodeDriver], + const { registry } = yield* makeProviderInstanceRegistry({ + drivers: [CodexDriver, ClaudeDriver, CursorDriver, GrokDriver, OpenCodeDriver, KiloDriver], configMap, }); @@ -312,9 +333,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, openCodeId, kiloId].toSorted(), ); // Instance lookup by id resolves each instance to its own bundle — @@ -325,11 +346,13 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { const cursor = yield* registry.getInstance(cursorId); const grok = yield* registry.getInstance(grokId); const openCode = yield* registry.getInstance(openCodeId); + const kilo = yield* registry.getInstance(kiloId); expect(codex?.driverKind).toBe(codexDriverKind); expect(claude?.driverKind).toBe(claudeDriverKind); expect(cursor?.driverKind).toBe(cursorDriverKind); expect(grok?.driverKind).toBe(grokDriverKind); expect(openCode?.driverKind).toBe(openCodeDriverKind); + expect(kilo?.driverKind).toBe(kiloDriverKind); expect(codex?.displayName).toBe("Codex"); expect(claude?.displayName).toBe("Claude"); expect(cursor?.displayName).toBe("Cursor"); @@ -405,6 +428,12 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { expect(openCodeSnapshot.continuation?.groupKey).toBe( `${openCodeDriverKind}:instance:${openCodeId}`, ); + + const kiloSnapshot = yield* kilo!.snapshot.getSnapshot; + expect(kiloSnapshot.instanceId).toBe(kiloId); + expect(kiloSnapshot.driver).toBe(kiloDriverKind); + expect(kiloSnapshot.enabled).toBe(false); + expect(kiloSnapshot.continuation?.groupKey).toBe(`${kiloDriverKind}:instance:${kiloId}`); }).pipe(Effect.provide(testLayer)), ); }); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index b3ab1145495..952f48b1495 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -33,6 +33,7 @@ import { applyServerSettingsPatch } from "@t3tools/shared/serverSettings"; import { checkCodexProviderStatus, type CodexAppServerProviderSnapshot } from "./CodexProvider.ts"; import { checkClaudeProviderStatus } from "./ClaudeProvider.ts"; import * as OpenCodeRuntime from "../opencodeRuntime.ts"; +import * as KiloRuntime from "../kiloRuntime.ts"; import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import { ProviderInstanceRegistryHydrationLive } from "./ProviderInstanceRegistryHydration.ts"; import { @@ -1120,6 +1121,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te ), ), Layer.provideMerge(OpenCodeRuntime.OpenCodeRuntimeLive), + Layer.provideMerge(KiloRuntime.KiloRuntimeLive), // NO spawner mock — `ChildProcessSpawner` is supplied by the // outer `NodeServices.layer` on `it.layer(...)` and will // genuinely spawn a subprocess. The missing-binary ENOENT is @@ -1212,6 +1214,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te ), ), Layer.provideMerge(OpenCodeRuntime.OpenCodeRuntimeLive), + Layer.provideMerge(KiloRuntime.KiloRuntimeLive), Layer.updateService(ChildProcessSpawner.ChildProcessSpawner, (spawner) => ChildProcessSpawner.make((command) => { spawnedCommands.push((command as { readonly command: string }).command); @@ -1333,6 +1336,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te ), ), Layer.provideMerge(OpenCodeRuntime.OpenCodeRuntimeLive), + Layer.provideMerge(KiloRuntime.KiloRuntimeLive), Layer.provideMerge(NodeServices.layer), ); const runtimeServices = yield* Layer.build(providerRegistryLayer).pipe( @@ -1394,6 +1398,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te ), ), Layer.provideMerge(OpenCodeRuntime.OpenCodeRuntimeLive), + Layer.provideMerge(KiloRuntime.KiloRuntimeLive), Layer.provideMerge( mockCommandSpawnerLayer((command, args) => { if (command === "agent") { diff --git a/apps/server/src/provider/Services/KiloAdapter.ts b/apps/server/src/provider/Services/KiloAdapter.ts new file mode 100644 index 00000000000..7aa0fc1aad5 --- /dev/null +++ b/apps/server/src/provider/Services/KiloAdapter.ts @@ -0,0 +1,4 @@ +import type { ProviderAdapterError } from "../Errors.ts"; +import type { ProviderAdapterShape } from "./ProviderAdapter.ts"; + +export interface KiloAdapterShape extends ProviderAdapterShape {} diff --git a/apps/server/src/provider/builtInDrivers.ts b/apps/server/src/provider/builtInDrivers.ts index 791a96e1da3..4677179f27f 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 { KiloDriver, type KiloDriverEnv } from "./Drivers/KiloDriver.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 + | KiloDriverEnv | OpenCodeDriverEnv; /** @@ -50,4 +52,5 @@ export const BUILT_IN_DRIVERS: ReadonlyArray { + it("parses canonical upstream provider/model selections", () => { + expect(parseKiloModelSlug("anthropic/claude-sonnet")).toEqual({ + providerID: "anthropic", + modelID: "claude-sonnet", + }); + expect(parseKiloModelSlug("invalid")).toBeNull(); + expect(parseKiloModelSlug("/missing-provider")).toBeNull(); + }); + + it("maps approval decisions without broadening access", () => { + expect(toKiloPermissionReply("accept")).toBe("once"); + expect(toKiloPermissionReply("acceptForSession")).toBe("always"); + expect(toKiloPermissionReply("decline")).toBe("reject"); + expect(toKiloPermissionReply("cancel")).toBe("reject"); + }); + + it("uses explicit ask rules outside full-access mode", () => { + expect(buildKiloPermissionRules("full-access")).toEqual([ + { permission: "*", pattern: "*", action: "allow" }, + ]); + expect(buildKiloPermissionRules("approval-required")).toEqual([ + { permission: "*", pattern: "*", action: "ask" }, + { permission: "question", pattern: "*", action: "allow" }, + ]); + expect(buildKiloPermissionRules("auto-accept-edits")).toEqual([ + { permission: "*", pattern: "*", action: "ask" }, + { permission: "edit", pattern: "*", action: "allow" }, + { permission: "write", pattern: "*", action: "allow" }, + { permission: "patch", pattern: "*", action: "allow" }, + { permission: "question", pattern: "*", action: "allow" }, + ]); + }); + + it("keeps Kilo as one top-level instance while model slugs remain upstream-qualified", () => { + const selection = { + instanceId: ProviderInstanceId.make("kilo"), + model: "openai/gpt-5", + }; + expect(selection.instanceId).toBe("kilo"); + expect(parseKiloModelSlug(selection.model)).toEqual({ providerID: "openai", modelID: "gpt-5" }); + }); +}); diff --git a/apps/server/src/provider/kiloRuntime.ts b/apps/server/src/provider/kiloRuntime.ts new file mode 100644 index 00000000000..9b12304886b --- /dev/null +++ b/apps/server/src/provider/kiloRuntime.ts @@ -0,0 +1,386 @@ +import * as NodeURL from "node:url"; + +import type { ChatAttachment, ProviderApprovalDecision, RuntimeMode } from "@t3tools/contracts"; +import { + createKiloClient, + type FilePartInput, + type KiloClient, + type PermissionRuleset, + type ProviderListResponse, + type QuestionAnswer, + type QuestionRequest, +} from "@kilocode/sdk/v2"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Data from "effect/Data"; +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 Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as P from "effect/Predicate"; +import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; + +import * as NetService from "@t3tools/shared/Net"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import { isWindowsCommandNotFound } from "../processRunner.ts"; +import { collectStreamAsString } from "./providerSnapshot.ts"; + +const KILO_SERVER_READY_PREFIX = "kilo server listening"; +const DEFAULT_KILO_SERVER_TIMEOUT_MS = 10_000; +const DEFAULT_HOSTNAME = "127.0.0.1"; +const KILO_EMPTY_CONFIG_CONTENT = "{}"; + +export interface KiloServerProcess { + readonly url: string; + readonly exitCode: Effect.Effect; +} + +export interface KiloServerConnection extends KiloServerProcess { + readonly external: false; +} + +const KILO_RUNTIME_ERROR_TAG = "KiloRuntimeError"; +export class KiloRuntimeError extends Data.TaggedError(KILO_RUNTIME_ERROR_TAG)<{ + readonly operation: string; + readonly cause?: unknown; + readonly detail: string; +}> { + static readonly is = (value: unknown): value is KiloRuntimeError => + P.isTagged(value, KILO_RUNTIME_ERROR_TAG); +} + +export function kiloRuntimeErrorDetail(cause: unknown): string { + if (KiloRuntimeError.is(cause)) return cause.detail; + if (cause instanceof Error && cause.message.trim().length > 0) return cause.message.trim(); + return String(cause); +} + +export const runKiloSdk = (operation: string, fn: () => Promise) => + Effect.tryPromise({ + try: fn, + catch: (cause) => + new KiloRuntimeError({ operation, detail: kiloRuntimeErrorDetail(cause), cause }), + }).pipe(Effect.withSpan(`kilo.${operation}`)); + +export interface KiloCommandResult { + readonly stdout: string; + readonly stderr: string; + readonly code: number; +} + +export interface KiloRuntimeShape { + readonly startServer: (input: { + readonly binaryPath: string; + readonly environment?: NodeJS.ProcessEnv; + readonly timeoutMs?: number; + }) => Effect.Effect; + readonly runCommand: (input: { + readonly binaryPath: string; + readonly args: ReadonlyArray; + readonly environment?: NodeJS.ProcessEnv; + }) => Effect.Effect; + readonly createClient: (input: { + readonly baseUrl: string; + readonly directory: string; + }) => KiloClient; + readonly loadInventory: ( + client: KiloClient, + ) => Effect.Effect; +} + +function parseServerUrl(output: string): string | null { + for (const line of output.split("\n")) { + if (!line.startsWith(KILO_SERVER_READY_PREFIX)) continue; + return line.match(/on\s+(https?:\/\/[^\s]+)/)?.[1] ?? null; + } + return null; +} + +export function parseKiloModelSlug(slug: string | null | undefined) { + if (typeof slug !== "string") return null; + const trimmed = slug.trim(); + const separator = trimmed.indexOf("/"); + if (separator <= 0 || separator === trimmed.length - 1) return null; + return { providerID: trimmed.slice(0, separator), modelID: trimmed.slice(separator + 1) }; +} + +export function toKiloFileParts(input: { + readonly attachments: ReadonlyArray | undefined; + readonly resolveAttachmentPath: (attachment: ChatAttachment) => string | null; +}): Array { + const parts: Array = []; + for (const attachment of input.attachments ?? []) { + const path = input.resolveAttachmentPath(attachment); + if (!path) continue; + parts.push({ + type: "file", + mime: attachment.mimeType, + filename: attachment.name, + url: NodeURL.pathToFileURL(path).href, + }); + } + return parts; +} + +export function buildKiloPermissionRules(runtimeMode: RuntimeMode): PermissionRuleset { + if (runtimeMode === "full-access") { + return [{ permission: "*", pattern: "*", action: "allow" }]; + } + if (runtimeMode === "auto-accept-edits") { + return [ + { permission: "*", pattern: "*", action: "ask" }, + { permission: "edit", pattern: "*", action: "allow" }, + { permission: "write", pattern: "*", action: "allow" }, + { permission: "patch", pattern: "*", action: "allow" }, + { permission: "question", pattern: "*", action: "allow" }, + ]; + } + return [ + { permission: "*", pattern: "*", action: "ask" }, + { permission: "question", pattern: "*", action: "allow" }, + ]; +} + +export function toKiloPermissionReply( + decision: ProviderApprovalDecision, +): "once" | "always" | "reject" { + if (decision === "accept") return "once"; + if (decision === "acceptForSession") return "always"; + return "reject"; +} + +export function kiloQuestionId(index: number, question: QuestionRequest["questions"][number]) { + const header = question.header + .trim() + .toLowerCase() + .replace(/[^a-z0-9_-]+/g, "-"); + return header ? `question-${index}-${header}` : `question-${index}`; +} + +export function toKiloQuestionAnswers( + request: QuestionRequest, + answers: Record, +): Array { + return request.questions.map((question, index) => { + const raw = + answers[kiloQuestionId(index, question)] ?? + answers[question.header] ?? + answers[question.question]; + if (Array.isArray(raw)) + return raw.filter((value): value is string => typeof value === "string"); + return typeof raw === "string" && raw.trim() ? [raw] : []; + }); +} + +const makeKiloRuntime = Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const net = yield* NetService.NetService; + const platform = yield* HostProcessPlatform; + + const runCommand: KiloRuntimeShape["runCommand"] = (input) => + Effect.gen(function* () { + const spawn = yield* resolveSpawnCommand( + input.binaryPath, + input.args, + input.environment ? { env: input.environment } : {}, + ); + const child = yield* spawner.spawn( + ChildProcess.make(spawn.command, spawn.args, { + shell: spawn.shell, + ...(input.environment ? { env: input.environment } : { extendEnv: true }), + }), + ); + const [stdout, stderr, code] = yield* Effect.all( + [collectStreamAsString(child.stdout), collectStreamAsString(child.stderr), child.exitCode], + { concurrency: "unbounded" }, + ); + const exitCode = Number(code); + if (yield* isWindowsCommandNotFound(exitCode, stderr)) { + return yield* new KiloRuntimeError({ + operation: "runCommand", + detail: `spawn ${input.binaryPath} ENOENT`, + }); + } + return { stdout, stderr, code: exitCode }; + }).pipe( + Effect.scoped, + Effect.mapError((cause) => + KiloRuntimeError.is(cause) + ? cause + : new KiloRuntimeError({ + operation: "runCommand", + detail: `Failed to execute Kilo: ${kiloRuntimeErrorDetail(cause)}`, + cause, + }), + ), + ); + + const startServer: KiloRuntimeShape["startServer"] = (input) => + Effect.gen(function* () { + const scope = yield* Scope.Scope; + const hostname = DEFAULT_HOSTNAME; + const port = yield* net.findAvailablePort(0).pipe( + Effect.mapError( + (cause) => + new KiloRuntimeError({ + operation: "startServer", + detail: `Failed to find available port: ${kiloRuntimeErrorDetail(cause)}`, + cause, + }), + ), + ); + const args = ["serve", `--hostname=${hostname}`, `--port=${port}`]; + const spawn = yield* resolveSpawnCommand( + input.binaryPath, + args, + input.environment ? { env: input.environment } : {}, + ); + const child = yield* spawner + .spawn( + ChildProcess.make(spawn.command, spawn.args, { + detached: platform !== "win32", + shell: spawn.shell, + env: { ...input.environment, KILO_CONFIG_CONTENT: KILO_EMPTY_CONFIG_CONTENT }, + extendEnv: input.environment === undefined, + }), + ) + .pipe( + Effect.provideService(Scope.Scope, scope), + Effect.mapError( + (cause) => + new KiloRuntimeError({ + operation: "startServer", + detail: `Failed to spawn Kilo server: ${kiloRuntimeErrorDetail(cause)}`, + cause, + }), + ), + ); + + const killGroup = (signal: NodeJS.Signals) => + platform === "win32" + ? child.kill({ killSignal: signal, forceKillAfter: "1 second" }).pipe(Effect.asVoid) + : Effect.sync(() => { + try { + process.kill(-Number(child.pid), signal); + } catch { + // Process already exited. + } + }); + yield* Scope.addFinalizer( + scope, + killGroup("SIGTERM").pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.andThen(killGroup("SIGKILL")), + Effect.ignore, + ), + ); + + const stdout = yield* Ref.make(""); + const stderr = yield* Ref.make(""); + const ready = yield* Deferred.make(); + yield* child.stdout.pipe( + Stream.decodeText(), + Stream.runForEach((chunk) => + Ref.updateAndGet(stdout, (value) => value + chunk).pipe( + Effect.flatMap((value) => { + const url = parseServerUrl(value); + return url ? Deferred.succeed(ready, url).pipe(Effect.ignore) : Effect.void; + }), + ), + ), + Effect.ignore, + Effect.forkIn(scope), + ); + yield* child.stderr.pipe( + Stream.decodeText(), + Stream.runForEach((chunk) => Ref.update(stderr, (value) => value + chunk)), + Effect.ignore, + Effect.forkIn(scope), + ); + const exitFiber = yield* child.exitCode.pipe( + Effect.flatMap((code) => + Effect.all({ stdout: Ref.get(stdout), stderr: Ref.get(stderr) }).pipe( + Effect.flatMap((output) => + Deferred.fail( + ready, + new KiloRuntimeError({ + operation: "startServer", + detail: `Kilo server exited before startup completed (code ${Number(code)}).\n${output.stderr || output.stdout}`, + }), + ).pipe(Effect.ignore), + ), + ), + ), + Effect.ignore, + Effect.forkIn(scope), + ); + + const result = yield* Effect.exit( + Deferred.await(ready).pipe( + Effect.timeoutOption(input.timeoutMs ?? DEFAULT_KILO_SERVER_TIMEOUT_MS), + ), + ); + if (Exit.isFailure(result)) { + yield* Fiber.interrupt(exitFiber).pipe(Effect.ignore); + const cause = Cause.squash(result.cause); + return yield* KiloRuntimeError.is(cause) + ? cause + : new KiloRuntimeError({ + operation: "startServer", + detail: kiloRuntimeErrorDetail(cause), + cause, + }); + } + if (Option.isNone(result.value)) { + yield* Fiber.interrupt(exitFiber).pipe(Effect.ignore); + return yield* new KiloRuntimeError({ + operation: "startServer", + detail: `Timed out waiting for Kilo server after ${input.timeoutMs ?? DEFAULT_KILO_SERVER_TIMEOUT_MS}ms.`, + }); + } + return { + url: result.value.value, + external: false as const, + exitCode: child.exitCode.pipe( + Effect.map(Number), + Effect.orElseSucceed(() => 0), + ), + }; + }); + + const createClient: KiloRuntimeShape["createClient"] = ({ baseUrl, directory }) => + createKiloClient({ baseUrl, directory, throwOnError: true }); + + const loadInventory: KiloRuntimeShape["loadInventory"] = (client) => + runKiloSdk("provider.list", () => client.provider.list(undefined, { throwOnError: true })).pipe( + Effect.filterMapOrFail( + (result) => + result.data + ? Result.succeed(result.data) + : Result.fail( + new KiloRuntimeError({ + operation: "provider.list", + detail: "Kilo provider list was empty.", + }), + ), + (result) => result, + ), + ); + + return { startServer, runCommand, createClient, loadInventory } satisfies KiloRuntimeShape; +}); + +export class KiloRuntime extends Context.Service()( + "t3/provider/kiloRuntime", +) {} + +export const KiloRuntimeLive = Layer.effect(KiloRuntime, makeKiloRuntime).pipe( + Layer.provide(NetService.layer), +); diff --git a/apps/server/src/provider/kiloSnapshotProgressFilter.test.ts b/apps/server/src/provider/kiloSnapshotProgressFilter.test.ts new file mode 100644 index 00000000000..52b1c0bf801 --- /dev/null +++ b/apps/server/src/provider/kiloSnapshotProgressFilter.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { isKiloSnapshotProgressText } from "./kiloSnapshotProgressFilter.ts"; + +describe("isKiloSnapshotProgressText", () => { + it("matches single-frame spinner progress with U+2026 ellipsis", () => { + expect(isKiloSnapshotProgressText("⠋ Initializing snapshot…")).toBe(true); + expect(isKiloSnapshotProgressText("⠹ Initializing snapshot…")).toBe(true); + }); + + it("matches progress with ASCII three-dot ellipsis", () => { + expect(isKiloSnapshotProgressText(" Initializing snapshot...")).toBe(true); + expect(isKiloSnapshotProgressText("⠋ Initializing snapshot.")).toBe(true); + }); + + it("matches progress with multiple braille frames", () => { + expect(isKiloSnapshotProgressText("⠋⠙ Initializing snapshot…")).toBe(true); + }); + + it("matches progress with leading or trailing whitespace", () => { + expect(isKiloSnapshotProgressText(" ⠋ Initializing snapshot… ")).toBe(true); + }); + + it("does not match plain assistant text", () => { + expect(isKiloSnapshotProgressText("Hi")).toBe(false); + }); + + it("does not match when the phrase appears mid-sentence", () => { + expect(isKiloSnapshotProgressText("Initializing snapshot… done")).toBe(false); + expect(isKiloSnapshotProgressText("Step 1: Initializing snapshot…")).toBe(false); + }); + + it("does not match when the ellipsis suffix is missing", () => { + expect(isKiloSnapshotProgressText("⠋ Initializing snapshot")).toBe(false); + expect(isKiloSnapshotProgressText("Initializing snapshot")).toBe(false); + }); + + it("does not match spinner characters outside the braille block", () => { + // U+2026 (ellipsis) used as the spinner; the spinner block must be U+2800..U+28FF. + expect(isKiloSnapshotProgressText("… Initializing snapshot…")).toBe(false); + }); +}); diff --git a/apps/server/src/provider/kiloSnapshotProgressFilter.ts b/apps/server/src/provider/kiloSnapshotProgressFilter.ts new file mode 100644 index 00000000000..0fa66e233df --- /dev/null +++ b/apps/server/src/provider/kiloSnapshotProgressFilter.ts @@ -0,0 +1,22 @@ +// Matches the synthetic "Initializing snapshot…" progress parts published by the +// Kilo SDK while it initializes the file snapshot used for undo/redo. The SDK +// (see `PROGRESS_INITIALIZING` in `@kilocode/sdk`) emits an ordinary text part +// whose full content is `{spinner} Initializing snapshot…`, where the spinner +// is a braille pattern character from the block U+2800..U+28FF. The frame is +// replaced in-place as the snapshot job progresses, so each frame arrives as a +// full replacement of the previous text rather than a delta. +// +// We tolerate: +// - zero or more braille frames (so empty spinner placeholders also match), +// - trailing whitespace, +// - either `…` (U+2026) or one or more `.` characters as the trailing +// ellipsis, since the UI font sometimes renders the ellipsis as three dots. +// +// The pattern is anchored (`^…$`) so it only matches when the entire text is +// the progress marker — a sentence that happens to contain the phrase +// "Initializing snapshot…" is unaffected. +const KILO_SNAPSHOT_PROGRESS_PATTERN = /^\s*[\u2800-\u28FF]*\s*Initializing snapshot[….]+\s*$/u; + +export function isKiloSnapshotProgressText(text: string): boolean { + return KILO_SNAPSHOT_PROGRESS_PATTERN.test(text); +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 0c632d8486c..b0a0d47c318 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -25,6 +25,7 @@ import * as ProviderEventLoggers from "./provider/Layers/ProviderEventLoggers.ts import { ProviderServiceLive } from "./provider/Layers/ProviderService.ts"; import { ProviderSessionReaperLive } from "./provider/Layers/ProviderSessionReaper.ts"; import * as OpenCodeRuntime from "./provider/opencodeRuntime.ts"; +import * as KiloRuntime from "./provider/kiloRuntime.ts"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as CheckpointStore from "./checkpointing/CheckpointStore.ts"; import * as AzureDevOpsCli from "./sourceControl/AzureDevOpsCli.ts"; @@ -312,7 +313,7 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( // the rewritten registry reads snapshots off the instance registry and // no longer transitively provides it. Exposing it at the runtime level // keeps a single Live for all opencode consumers. - Layer.provideMerge(OpenCodeRuntime.OpenCodeRuntimeLive), + Layer.provideMerge(Layer.merge(OpenCodeRuntime.OpenCodeRuntimeLive, KiloRuntime.KiloRuntimeLive)), Layer.provideMerge(ServerSettings.layer.pipe(Layer.provide(ServerSecretStore.layer))), Layer.provideMerge(WorkspaceLayerLive), Layer.provideMerge(ProjectFaviconResolverLayerLive), diff --git a/apps/server/src/textGeneration/KiloTextGeneration.test.ts b/apps/server/src/textGeneration/KiloTextGeneration.test.ts new file mode 100644 index 00000000000..2031ce37c82 --- /dev/null +++ b/apps/server/src/textGeneration/KiloTextGeneration.test.ts @@ -0,0 +1,125 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { KiloSettings, ProviderInstanceId, TextGenerationError } from "@t3tools/contracts"; +import { createModelSelection } from "@t3tools/shared/model"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import { beforeEach } from "vite-plus/test"; + +import { ServerConfig } from "../config.ts"; +import { KiloRuntime, KiloRuntimeError, type KiloRuntimeShape } from "../provider/kiloRuntime.ts"; +import * as TextGeneration from "./TextGeneration.ts"; +import { makeKiloTextGeneration } from "./KiloTextGeneration.ts"; + +const isTextGenerationError = Schema.is(TextGenerationError); + +const runtimeMock = { + state: { + sessionResult: undefined as { data?: { id: string } } | undefined, + promptResult: undefined as + | { data?: { info?: { error?: unknown }; parts?: Array } } + | undefined, + failSessionCreateWith: undefined as Error | undefined, + }, + reset() { + this.state.sessionResult = undefined; + this.state.promptResult = undefined; + this.state.failSessionCreateWith = undefined; + }, +}; + +const KiloRuntimeTestDouble: KiloRuntimeShape = { + startServer: () => + Effect.gen(function* () { + yield* Effect.addFinalizer(() => Effect.void); + return { url: "http://127.0.0.1:4301", external: false as const, exitCode: Effect.never }; + }), + runCommand: () => Effect.succeed({ stdout: "", stderr: "", code: 0 }), + createClient: () => + ({ + session: { + create: async () => { + if (runtimeMock.state.failSessionCreateWith) { + throw runtimeMock.state.failSessionCreateWith; + } + return runtimeMock.state.sessionResult ?? { data: { id: "session-id" } }; + }, + prompt: async () => runtimeMock.state.promptResult ?? { data: { parts: [] } }, + }, + }) as unknown as ReturnType, + loadInventory: () => + Effect.die(new KiloRuntimeError({ operation: "loadInventory", detail: "not used" })), +}; + +const TEST_SETTINGS = Schema.decodeSync(KiloSettings)({ binaryPath: "fake-kilo" }); +const TEST_MODEL = createModelSelection( + ProviderInstanceId.make("kilo"), + "anthropic/claude-sonnet-4-5", +); + +const kiloTextGenerationTestLayer = Layer.effect( + TextGeneration.TextGeneration, + makeKiloTextGeneration(TEST_SETTINGS), +).pipe( + Layer.provideMerge(Layer.succeed(KiloRuntime, KiloRuntimeTestDouble)), + Layer.provideMerge(ServerConfig.layerTest("/tmp/kilo-tg", "/tmp")), + Layer.provideMerge(NodeServices.layer), +); + +it.layer(kiloTextGenerationTestLayer)("KiloTextGeneration error mapping", (it) => { + beforeEach(() => { + runtimeMock.reset(); + }); + + it.effect( + "maps KiloRuntimeError to a TextGenerationError with structural detail, not cause.message", + () => + Effect.gen(function* () { + runtimeMock.state.failSessionCreateWith = new KiloRuntimeError({ + operation: "session.create", + detail: "credential material that must remain in the cause chain", + }); + const tg = yield* TextGeneration.TextGeneration; + const error = yield* tg + .generateBranchName({ + cwd: "/tmp/kilo-tg", + message: "noop", + modelSelection: TEST_MODEL, + }) + .pipe(Effect.flip); + + expect(isTextGenerationError(error)).toBe(true); + if (!isTextGenerationError(error)) return; + expect(error.operation).toBe("generateBranchName"); + expect(error.detail).toBe("Kilo text generation request failed (session.create)."); + // The wrapper's detail must NOT be derived from cause.message — this is + // the Macroscope review's central complaint about the old Exit/Cause wrap. + expect(error.detail.includes("credential material")).toBe(false); + expect(error.cause !== undefined).toBe(true); + }), + ); + + it.effect( + "maps KiloTextGenerationSessionPayloadError to TextGenerationError with structural detail", + () => + Effect.gen(function* () { + runtimeMock.state.sessionResult = { data: undefined } as unknown as { + data: { id: string }; + }; + const tg = yield* TextGeneration.TextGeneration; + const error = yield* tg + .generateBranchName({ + cwd: "/tmp/kilo-tg", + message: "noop", + modelSelection: TEST_MODEL, + }) + .pipe(Effect.flip); + + expect(isTextGenerationError(error)).toBe(true); + if (!isTextGenerationError(error)) return; + expect(error.operation).toBe("generateBranchName"); + expect(error.detail).toBe("Kilo session.create returned no session payload."); + }), + ); +}); diff --git a/apps/server/src/textGeneration/KiloTextGeneration.ts b/apps/server/src/textGeneration/KiloTextGeneration.ts new file mode 100644 index 00000000000..b8ff2f2cab9 --- /dev/null +++ b/apps/server/src/textGeneration/KiloTextGeneration.ts @@ -0,0 +1,248 @@ +import { + NonNegativeInt, + TextGenerationError, + type ChatAttachment, + type KiloSettings, + type ModelSelection, +} from "@t3tools/contracts"; +import { sanitizeBranchFragment, sanitizeFeatureBranchName } from "@t3tools/shared/git"; +import { extractJsonObject } from "@t3tools/shared/schemaJson"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +import { resolveAttachmentPath } from "../attachmentStore.ts"; +import * as ServerConfig from "../config.ts"; +import { + KiloRuntime, + KiloRuntimeError, + parseKiloModelSlug, + runKiloSdk, + toKiloFileParts, +} from "../provider/kiloRuntime.ts"; +import { sanitizeThreadTitle } from "./TextGenerationUtils.ts"; +import * as TextGeneration from "./TextGeneration.ts"; + +const FIXED_AGENT = "code"; +const kiloTextGenerationErrorContext = { + operation: Schema.String, + cwd: Schema.String, +}; + +export class KiloTextGenerationSessionPayloadError extends Schema.TaggedErrorClass()( + "KiloTextGenerationSessionPayloadError", + kiloTextGenerationErrorContext, +) { + override get message(): string { + return `Kilo session.create returned no session payload for ${this.operation} in ${this.cwd}.`; + } +} + +export class KiloTextGenerationEmptyOutputError extends Schema.TaggedErrorClass()( + "KiloTextGenerationEmptyOutputError", + { + ...kiloTextGenerationErrorContext, + sessionId: Schema.String, + responsePartCount: NonNegativeInt, + textPartCount: NonNegativeInt, + }, +) { + override get message(): string { + return `Kilo returned empty output for ${this.operation} in ${this.cwd} (session ${this.sessionId}, ${this.responsePartCount} response parts, ${this.textPartCount} text parts).`; + } +} + +function textFromParts(parts: ReadonlyArray): string { + return parts + .flatMap((part) => + part && + typeof part === "object" && + "type" in part && + part.type === "text" && + "text" in part && + typeof part.text === "string" + ? [part.text] + : [], + ) + .join("\n") + .trim(); +} +const CommitOutput = Schema.Struct({ + subject: Schema.String, + body: Schema.String, + branch: Schema.optional(Schema.String), +}); +const PrOutput = Schema.Struct({ title: Schema.String, body: Schema.String }); +const decodeCommitOutput = Schema.decodeEffect(Schema.fromJsonString(CommitOutput)); +const decodePrOutput = Schema.decodeEffect(Schema.fromJsonString(PrOutput)); + +export const makeKiloTextGeneration = Effect.fn("makeKiloTextGeneration")(function* ( + settings: KiloSettings, + environment?: NodeJS.ProcessEnv, +) { + const runtime = yield* KiloRuntime; + const config = yield* ServerConfig.ServerConfig; + + const run = Effect.fn("runKiloTextGeneration")(function* (input: { + readonly operation: string; + readonly cwd: string; + readonly prompt: string; + readonly modelSelection: ModelSelection; + readonly attachments?: ReadonlyArray; + }) { + const model = parseKiloModelSlug(input.modelSelection.model); + if (!model) { + return yield* new TextGenerationError({ + operation: input.operation, + detail: "Kilo model selection must use the 'provider/model' format.", + }); + } + return yield* Effect.scoped( + Effect.gen(function* () { + const server = yield* runtime.startServer({ + binaryPath: settings.binaryPath, + ...(environment ? { environment } : {}), + }); + const client = runtime.createClient({ baseUrl: server.url, directory: input.cwd }); + const session = yield* runKiloSdk("session.create", () => + client.session.create( + { + title: `T3 Code ${input.operation}`, + agent: FIXED_AGENT, + model: { id: model.modelID, providerID: model.providerID }, + permission: [{ permission: "*", pattern: "*", action: "deny" }], + platform: "t3code", + }, + { throwOnError: true }, + ), + ); + if (!session.data) { + return yield* new KiloTextGenerationSessionPayloadError({ + operation: input.operation, + cwd: input.cwd, + }); + } + const files = toKiloFileParts({ + attachments: input.attachments, + resolveAttachmentPath: (attachment) => + resolveAttachmentPath({ attachmentsDir: config.attachmentsDir, attachment }), + }); + const response = yield* runKiloSdk("session.prompt", () => + client.session.prompt( + { + sessionID: session.data.id, + model, + agent: FIXED_AGENT, + parts: [{ type: "text", text: input.prompt }, ...files], + }, + { throwOnError: true }, + ), + ); + const responseParts = response.data?.parts ?? []; + const text = textFromParts(responseParts); + if (!text) { + return yield* new KiloTextGenerationEmptyOutputError({ + operation: input.operation, + cwd: input.cwd, + sessionId: session.data.id, + responsePartCount: responseParts.length, + textPartCount: responseParts.filter((part) => part.type === "text").length, + }); + } + return text; + }), + ).pipe( + Effect.catchTags({ + KiloRuntimeError: (cause) => + Effect.fail( + new TextGenerationError({ + operation: input.operation, + detail: `Kilo text generation request failed (${cause.operation}).`, + cause, + }), + ), + KiloTextGenerationSessionPayloadError: (cause) => + Effect.fail( + new TextGenerationError({ + operation: cause.operation, + detail: "Kilo session.create returned no session payload.", + cause, + }), + ), + KiloTextGenerationEmptyOutputError: (cause) => + Effect.fail( + new TextGenerationError({ + operation: cause.operation, + detail: "Kilo returned empty output.", + cause, + }), + ), + }), + ); + }); + + return TextGeneration.TextGeneration.of({ + generateCommitMessage: (input) => + run({ + operation: "generateCommitMessage", + cwd: input.cwd, + modelSelection: input.modelSelection, + prompt: `Return JSON with ${input.includeBranch ? "subject, body, and a concise semantic branch name" : "subject and body"} for this staged change.\nBranch: ${input.branch ?? "unknown"}\nSummary:\n${input.stagedSummary}\nPatch:\n${input.stagedPatch}`, + }).pipe( + Effect.flatMap((text) => + decodeCommitOutput(extractJsonObject(text)).pipe( + Effect.map((output) => ({ + subject: output.subject, + body: output.body, + ...(input.includeBranch && output.branch + ? { branch: sanitizeFeatureBranchName(output.branch) } + : {}), + })), + Effect.mapError( + (cause) => + new TextGenerationError({ + operation: "generateCommitMessage", + detail: "Kilo returned invalid commit JSON.", + cause, + }), + ), + ), + ), + ), + generatePrContent: (input) => + run({ + operation: "generatePrContent", + cwd: input.cwd, + modelSelection: input.modelSelection, + prompt: `Return JSON with title and body for a pull request.\nBase: ${input.baseBranch}\nHead: ${input.headBranch}\nCommits:\n${input.commitSummary}\nDiff:\n${input.diffSummary}\nPatch:\n${input.diffPatch}`, + }).pipe( + Effect.flatMap((text) => + decodePrOutput(extractJsonObject(text)).pipe( + Effect.mapError( + (cause) => + new TextGenerationError({ + operation: "generatePrContent", + detail: "Kilo returned invalid pull request JSON.", + cause, + }), + ), + ), + ), + ), + generateBranchName: (input) => + run({ + operation: "generateBranchName", + cwd: input.cwd, + modelSelection: input.modelSelection, + ...(input.attachments ? { attachments: input.attachments } : {}), + prompt: `Return only a concise lowercase kebab-case git branch name for: ${input.message}`, + }).pipe(Effect.map((branch) => ({ branch: sanitizeBranchFragment(branch) }))), + generateThreadTitle: (input) => + run({ + operation: "generateThreadTitle", + cwd: input.cwd, + modelSelection: input.modelSelection, + ...(input.attachments ? { attachments: input.attachments } : {}), + prompt: `Return only a concise thread title for: ${input.message}`, + }).pipe(Effect.map((title) => ({ title: sanitizeThreadTitle(title) }))), + }); +}); diff --git a/apps/web/src/components/Icons.tsx b/apps/web/src/components/Icons.tsx index 8ea38c51958..e271f3ddbd4 100644 --- a/apps/web/src/components/Icons.tsx +++ b/apps/web/src/components/Icons.tsx @@ -663,6 +663,12 @@ export const OpenCodeIcon: Icon = (props) => ( ); +export const KiloIcon: Icon = (props) => ( + + + +); + export const GithubCopilotIcon: Icon = ({ className, ...props }) => ( > = { @@ -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("kilo")]: KiloIcon, }; 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..eede76bf8be 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, + KiloSettings, 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, + KiloIcon, + type Icon, + OpenAI, + OpenCodeIcon, +} from "../Icons"; type ProviderSettingsSchema = { readonly fields: Readonly>; @@ -67,6 +76,13 @@ export const PROVIDER_CLIENT_DEFINITIONS: readonly ProviderClientDefinition[] = icon: OpenCodeIcon, settingsSchema: OpenCodeSettings, }, + { + value: ProviderDriverKind.make("kilo"), + label: "Kilo", + icon: KiloIcon, + badgeLabel: "Early Access", + settingsSchema: KiloSettings, + }, ]; export const PROVIDER_CLIENT_DEFINITION_BY_VALUE: Partial< diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 5d5051f748e..bfc75fa0c93 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("kilo"), + label: "Kilo", + available: true, + pickerSidebarBadge: "new", + }, ]; export type WorkLogToolLifecycleStatus = diff --git a/packages/contracts/src/model.ts b/packages/contracts/src/model.ts index dddf3f37459..506c668b78c 100644 --- a/packages/contracts/src/model.ts +++ b/packages/contracts/src/model.ts @@ -132,6 +132,7 @@ const CLAUDE_DRIVER_KIND = ProviderDriverKind.make("claudeAgent"); const CURSOR_DRIVER_KIND = ProviderDriverKind.make("cursor"); const GROK_DRIVER_KIND = ProviderDriverKind.make("grok"); const OPENCODE_DRIVER_KIND = ProviderDriverKind.make("opencode"); +const KILO_DRIVER_KIND = ProviderDriverKind.make("kilo"); export const DEFAULT_MODEL = "gpt-5.4"; export const DEFAULT_GIT_TEXT_GENERATION_MODEL = "gpt-5.4-mini"; @@ -142,6 +143,7 @@ export const DEFAULT_MODEL_BY_PROVIDER: Partial> [CURSOR_DRIVER_KIND]: "Cursor", [GROK_DRIVER_KIND]: "Grok", [OPENCODE_DRIVER_KIND]: "OpenCode", + [KILO_DRIVER_KIND]: "Kilo", }; diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index eb2563eff00..d578553ec1a 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -26,6 +26,7 @@ const RuntimeEventRawSource = Schema.Union([ Schema.Literal("claude.sdk.permission"), Schema.Literal("codex.sdk.thread-event"), Schema.Literal("opencode.sdk.event"), + Schema.Literal("kilo.sdk.event"), Schema.Literal("acp.jsonrpc"), Schema.TemplateLiteral(["acp.", Schema.String, ".extension"]), ]); diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 6ccd65533dd..a5c9ef21a57 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -355,6 +355,33 @@ export const OpenCodeSettings = makeProviderSettingsSchema( ); export type OpenCodeSettings = typeof OpenCodeSettings.Type; +export const KiloSettings = makeProviderSettingsSchema( + { + enabled: Schema.Boolean.pipe( + Schema.withDecodingDefault(Effect.succeed(true)), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + binaryPath: makeBinaryPathSetting("kilo").pipe( + Schema.annotateKey({ + title: "Binary path", + description: "Path to the Kilo CLI binary.", + providerSettingsForm: { + placeholder: "kilo", + clearWhenEmpty: "omit", + }, + }), + ), + customModels: Schema.Array(Schema.String).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + }, + { + order: ["binaryPath"], + }, +); +export type KiloSettings = typeof KiloSettings.Type; + export const ObservabilitySettings = Schema.Struct({ otlpTracesUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), otlpMetricsUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), @@ -399,6 +426,7 @@ export const ServerSettings = Schema.Struct({ cursor: CursorSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), grok: GrokSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), opencode: OpenCodeSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), + kilo: KiloSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), }).pipe(Schema.withDecodingDefault(Effect.succeed({}))), // New driver-agnostic instance map. Keyed by `ProviderInstanceId`; values // are `ProviderInstanceConfig` envelopes. The driver-specific config blob @@ -501,6 +529,12 @@ const OpenCodeSettingsPatch = Schema.Struct({ customModels: Schema.optionalKey(Schema.Array(Schema.String)), }); +const KiloSettingsPatch = Schema.Struct({ + enabled: Schema.optionalKey(Schema.Boolean), + binaryPath: Schema.optionalKey(TrimmedString), + customModels: Schema.optionalKey(Schema.Array(Schema.String)), +}); + export const ServerSettingsPatch = Schema.Struct({ // Server settings enableAssistantStreaming: Schema.optionalKey(Schema.Boolean), @@ -523,6 +557,7 @@ export const ServerSettingsPatch = Schema.Struct({ cursor: Schema.optionalKey(CursorSettingsPatch), grok: Schema.optionalKey(GrokSettingsPatch), opencode: Schema.optionalKey(OpenCodeSettingsPatch), + kilo: Schema.optionalKey(KiloSettingsPatch), }), ), // Whole-map replacement for the new instance config. Patching individual diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ed67594e846..dd6d6f5ffa7 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -442,6 +442,9 @@ importers: '@ff-labs/fff-node': specifier: 0.9.4 version: 0.9.4(patch_hash=2b16019ce7ab61aec6478dd02f79ef468cc1d5c51e9d00764f7d2ab8167210c8) + '@kilocode/sdk': + specifier: ^7.4.5 + version: 7.4.5 '@opencode-ai/sdk': specifier: ^1.3.15 version: 1.15.13 @@ -2910,6 +2913,9 @@ packages: resolution: {integrity: sha512-hloP58zRVCRSpgDxmqCWJNlizAlUgJFqG2ypq79DCvyv9tHjRYMDOcPFjzfl/A1/YxDvRCZz8wvZvmapQnKwFQ==} engines: {node: '>=12'} + '@kilocode/sdk@7.4.5': + resolution: {integrity: sha512-sJurT4oQ7aXP6eHcYhhYCm50WghkEhdZgXjTeSZfRSMJkTlf7c+kz1wkWImv7J8lCJIwPwrE3KNBKwtOMfQtyg==} + '@legendapp/list@3.2.0': resolution: {integrity: sha512-bN+g/oQYjFz+UAyuBN4cmYJAwdJS1TdNcZZOVlh3+VwCQUWrsg0PH46Mvm76gdZSCYMfoFanPY4dKnILcYEzeg==} peerDependencies: @@ -12804,6 +12810,10 @@ snapshots: dependencies: jsbi: 4.3.2 + '@kilocode/sdk@7.4.5': + dependencies: + cross-spawn: 7.0.6 + '@legendapp/list@3.2.0(patch_hash=45e4cbcbeeca1b628b8480869374d7cbc77e3fbfa97ee0107a0d72c3c4a5d7f3)(react-dom@19.2.3(react@19.2.3))(react-native@0.85.3(@babel/core@7.29.7)(@react-native/metro-config@0.85.3(@babel/core@7.29.7)(bufferutil@4.1.0)(utf-8-validate@6.0.6))(@types/react@19.2.16)(bufferutil@4.1.0)(react@19.2.3)(utf-8-validate@6.0.6))(react@19.2.3)': dependencies: react: 19.2.3