diff --git a/apps/server/package.json b/apps/server/package.json index d0903c77d75..40f8145be7d 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.11", "@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..b8696ac33ee --- /dev/null +++ b/apps/server/src/provider/Drivers/KiloDriver.ts @@ -0,0 +1,172 @@ +/** + * KiloDriver — `ProviderDriver` for the Kilo runtime. + * + * Mirrors OpenCode: a plain value whose `create()` bundles + * `snapshot` / `adapter` / `textGeneration` closures over the + * per-instance `KiloSettings`. + * + * @module provider/Drivers/KiloDriver + */ +import { KiloSettings, ProviderDriverKind, type ServerProvider } from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import { HttpClient } from "effect/unstable/http"; +import { ChildProcessSpawner } from "effect/unstable/process"; + +import { 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 { ProviderEventLoggers } from "../Layers/ProviderEventLoggers.ts"; +import { makeManagedServerProvider } from "../makeManagedServerProvider.ts"; +import { KiloRuntime } from "../kiloRuntime.ts"; +import { + defaultProviderContinuationIdentity, + type ProviderDriver, + type ProviderInstance, +} from "../ProviderDriver.ts"; +import type { ServerProviderDraft } from "../providerSnapshot.ts"; +import { mergeProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; +import { + enrichProviderSnapshotWithVersionAdvisory, + makeManualOnlyProviderMaintenanceCapabilities, + makeStaticProviderMaintenanceResolver, + resolveProviderMaintenanceCapabilitiesEffect, +} from "../providerMaintenance.ts"; +import { + haveProviderSnapshotSettingsChanged, + makeProviderSnapshotSettingsSource, + type ProviderSnapshotSettings, +} from "../providerUpdateSettings.ts"; + +const decodeKiloSettings = Schema.decodeSync(KiloSettings); + +const DRIVER_KIND = ProviderDriverKind.make("kilo"); +const SNAPSHOT_REFRESH_INTERVAL = Duration.minutes(5); +const UPDATE = makeStaticProviderMaintenanceResolver( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: DRIVER_KIND, + packageName: null, + }), +); + +export type KiloDriverEnv = + | ChildProcessSpawner.ChildProcessSpawner + | Crypto.Crypto + | FileSystem.FileSystem + | HttpClient.HttpClient + | KiloRuntime + | Path.Path + | ProviderEventLoggers + | ServerConfig + | ServerSettingsService; + +const withInstanceIdentity = + (input: { + readonly instanceId: ProviderInstance["instanceId"]; + readonly displayName: string | undefined; + readonly accentColor: string | undefined; + readonly continuationGroupKey: string; + }) => + (snapshot: ServerProviderDraft): ServerProvider => ({ + ...snapshot, + instanceId: input.instanceId, + driver: DRIVER_KIND, + ...(input.displayName ? { displayName: input.displayName } : {}), + ...(input.accentColor ? { accentColor: input.accentColor } : {}), + continuation: { groupKey: input.continuationGroupKey }, + }); + +export const KiloDriver: ProviderDriver = { + driverKind: DRIVER_KIND, + metadata: { + displayName: "Kilo", + supportsMultipleInstances: true, + }, + configSchema: KiloSettings, + defaultConfig: (): KiloSettings => decodeKiloSettings({}), + create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => + Effect.gen(function* () { + const kiloRuntime = yield* KiloRuntime; + const serverConfig = yield* ServerConfig; + const httpClient = yield* HttpClient.HttpClient; + const serverSettings = yield* ServerSettingsService; + const eventLoggers = yield* ProviderEventLoggers; + const processEnv = mergeProviderInstanceEnvironment(environment); + const continuationIdentity = defaultProviderContinuationIdentity({ + driverKind: DRIVER_KIND, + instanceId, + }); + const stampIdentity = withInstanceIdentity({ + instanceId, + displayName, + accentColor, + continuationGroupKey: continuationIdentity.continuationKey, + }); + const effectiveConfig = { ...config, enabled } satisfies KiloSettings; + const maintenanceCapabilities = yield* resolveProviderMaintenanceCapabilitiesEffect(UPDATE, { + binaryPath: effectiveConfig.binaryPath, + env: processEnv, + }); + + const adapter = yield* makeKiloAdapter(effectiveConfig, { + instanceId, + environment: processEnv, + ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}), + }); + const textGeneration = yield* makeKiloTextGeneration(effectiveConfig, processEnv); + + const checkProvider = checkKiloProviderStatus( + effectiveConfig, + serverConfig.cwd, + processEnv, + ).pipe(Effect.map(stampIdentity), Effect.provideService(KiloRuntime, kiloRuntime)); + + const snapshotSettings = makeProviderSnapshotSettingsSource(effectiveConfig, serverSettings); + const snapshot = yield* makeManagedServerProvider>({ + maintenanceCapabilities, + getSettings: snapshotSettings.getSettings, + streamSettings: snapshotSettings.streamSettings, + haveSettingsChanged: haveProviderSnapshotSettingsChanged, + initialSnapshot: (settings) => + makePendingKiloProvider(settings.provider).pipe(Effect.map(stampIdentity)), + checkProvider, + enrichSnapshot: ({ settings, snapshot, publishSnapshot }) => + enrichProviderSnapshotWithVersionAdvisory(snapshot, maintenanceCapabilities, { + enableProviderUpdateChecks: settings.enableProviderUpdateChecks, + }).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.flatMap((enrichedSnapshot) => publishSnapshot(enrichedSnapshot)), + ), + refreshInterval: SNAPSHOT_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..4b82e014697 --- /dev/null +++ b/apps/server/src/provider/Layers/KiloAdapter.test.ts @@ -0,0 +1,604 @@ +import * as NodeAssert from "node:assert/strict"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +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 Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { beforeEach } from "vite-plus/test"; + +import { + ApprovalRequestId, + KiloSettings, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + TurnId, +} from "@t3tools/contracts"; +import { createModelSelection } from "@t3tools/shared/model"; +import { ServerConfig } from "../../config.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import type { KiloAdapterShape } from "../Services/KiloAdapter.ts"; +import { KiloRuntime, KiloRuntimeError, type KiloRuntimeShape } from "../kiloRuntime.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[], + sessionCreateCalls: [] as Array, + authHeaders: [] as Array, + abortCalls: [] as string[], + closeCalls: [] as string[], + promptCalls: [] as Array, + permissionReplies: [] as Array, + promptAsyncError: null as Error | null, + messages: [] as Array, + subscribedEvents: [] as unknown[], + /** + * Optional gate the mocked `session.abort` RPC awaits before + * resolving. Tests install a gate so they can pin `interruptTurn` + * inside the in-flight abort RPC and then exercise concurrent + * `sendTurn`. While the gate is installed, all registered + * `abortEnteredCallbacks` are invoked synchronously when the abort + * RPC is entered so the test can wake up deterministically before + * allowing the RPC to make progress. + */ + abortGate: null as { readonly promise: Promise } | null, + abortEnteredCallbacks: [] as Array<() => void>, + }, + reset() { + this.state.startCalls.length = 0; + this.state.sessionCreateCalls.length = 0; + this.state.authHeaders.length = 0; + this.state.abortCalls.length = 0; + this.state.closeCalls.length = 0; + this.state.promptCalls.length = 0; + this.state.permissionReplies.length = 0; + this.state.promptAsyncError = null; + this.state.messages = []; + this.state.subscribedEvents = []; + this.state.abortGate = null; + this.state.abortEnteredCallbacks.length = 0; + }, +}; + +const KiloRuntimeTestDouble: KiloRuntimeShape = { + startKiloServerProcess: ({ binaryPath }) => + Effect.gen(function* () { + runtimeMock.state.startCalls.push(binaryPath); + const url = "http://127.0.0.1:4301"; + yield* Effect.addFinalizer(() => + Effect.sync(() => { + runtimeMock.state.closeCalls.push(url); + }), + ); + return { + url, + password: "secret-password", + exitCode: Effect.never, + }; + }), + connectToKiloServer: ({ binaryPath }) => + Effect.gen(function* () { + runtimeMock.state.startCalls.push(binaryPath); + const url = "http://127.0.0.1:4301"; + yield* Effect.addFinalizer(() => + Effect.sync(() => { + runtimeMock.state.closeCalls.push(url); + }), + ); + return { + url, + password: "secret-password", + exitCode: null, + external: false, + }; + }), + runKiloCommand: () => Effect.succeed({ stdout: "", stderr: "", code: 0 }), + createKiloSdkClient: ({ baseUrl, serverPassword }) => + ({ + session: { + create: async (input: unknown) => { + runtimeMock.state.sessionCreateCalls.push(input); + runtimeMock.state.authHeaders.push( + serverPassword ? `Basic ${btoa(`kilo:${serverPassword}`)}` : null, + ); + return { data: { id: `${baseUrl}/session` } }; + }, + abort: async ({ sessionID }: { sessionID: string }) => { + runtimeMock.state.abortCalls.push(sessionID); + const gate = runtimeMock.state.abortGate; + if (gate) { + // Fire any "abort entered" callbacks so the test fiber + // wakes up, THEN await the gate's promise. Resolving the + // entered callbacks here is safe — they schedule an Effect + // that resumes when the test fiber gets a chance to run. + while (runtimeMock.state.abortEnteredCallbacks.length > 0) { + const cb = runtimeMock.state.abortEnteredCallbacks.shift()!; + cb(); + } + await gate.promise; + } + }, + promptAsync: async (input: unknown) => { + runtimeMock.state.promptCalls.push(input); + if (runtimeMock.state.promptAsyncError) { + throw runtimeMock.state.promptAsyncError; + } + }, + messages: async () => ({ data: runtimeMock.state.messages }), + revert: async () => undefined, + }, + permission: { + reply: async (input: unknown) => { + runtimeMock.state.permissionReplies.push(input); + }, + }, + question: { + reply: async () => undefined, + }, + mcp: { + add: async () => undefined, + }, + event: { + subscribe: async () => ({ + stream: (async function* () { + for (const event of runtimeMock.state.subscribedEvents) { + yield event; + } + })(), + }), + }, + }) as unknown as ReturnType, + loadKiloInventory: () => + Effect.fail( + new KiloRuntimeError({ + operation: "loadKiloInventory", + detail: "KiloRuntimeTestDouble.loadKiloInventory not used in this test", + cause: null, + }), + ), +}; + +const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory, { + upsert: () => Effect.void, + getProvider: () => + Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")), + getBinding: () => Effect.succeed(Option.none()), + listThreadIds: () => Effect.succeed([]), + listBindings: () => Effect.succeed([]), +}); + +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(providerSessionDirectoryTestLayer), + Layer.provideMerge(NodeServices.layer), +); + +beforeEach(() => { + runtimeMock.reset(); +}); + +it.layer(KiloAdapterTestLayer)("KiloAdapter", (it) => { + it.effect("starts a managed Kilo server session with agent code", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + + const session = yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId: asThreadId("thread-kilo"), + runtimeMode: "full-access", + }); + + NodeAssert.equal(session.provider, "kilo"); + NodeAssert.equal(session.threadId, "thread-kilo"); + NodeAssert.deepEqual(runtimeMock.state.startCalls, ["fake-kilo"]); + NodeAssert.deepEqual(runtimeMock.state.authHeaders, [ + `Basic ${btoa("kilo:secret-password")}`, + ]); + const createCall = runtimeMock.state.sessionCreateCalls.at(-1) as { + agent?: string; + }; + NodeAssert.equal(createCall?.agent, "code"); + }), + ); + + it.effect("sends turns with model slug and default agent code", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId: asThreadId("thread-kilo-turn"), + runtimeMode: "full-access", + }); + + yield* adapter.sendTurn({ + threadId: asThreadId("thread-kilo-turn"), + input: "Hello", + modelSelection: createModelSelection( + ProviderInstanceId.make("kilo"), + "anthropic/claude-sonnet-4-5", + ), + }); + + NodeAssert.deepEqual(runtimeMock.state.promptCalls.at(-1), { + sessionID: "http://127.0.0.1:4301/session", + model: { + providerID: "anthropic", + modelID: "claude-sonnet-4-5", + }, + agent: "code", + parts: [{ type: "text", text: "Hello" }], + }); + }), + ); + + it.effect("uses agent plan when interactionMode is plan", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId: asThreadId("thread-kilo-plan"), + runtimeMode: "full-access", + }); + + yield* adapter.sendTurn({ + threadId: asThreadId("thread-kilo-plan"), + input: "Plan this", + interactionMode: "plan", + modelSelection: createModelSelection( + ProviderInstanceId.make("kilo"), + "anthropic/claude-sonnet-4-5", + ), + }); + + const prompt = runtimeMock.state.promptCalls.at(-1) as { agent?: string }; + NodeAssert.equal(prompt?.agent, "plan"); + }), + ); + + it.effect("interrupts an active turn via session.abort", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-interrupt"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId, + input: "long task", + modelSelection: createModelSelection( + ProviderInstanceId.make("kilo"), + "anthropic/claude-sonnet-4-5", + ), + }); + + yield* adapter.interruptTurn(threadId, turn.turnId); + NodeAssert.ok(runtimeMock.state.abortCalls.includes("http://127.0.0.1:4301/session")); + const sessions = yield* adapter.listSessions(); + NodeAssert.equal(sessions[0]?.status, "ready"); + NodeAssert.equal(sessions[0]?.activeTurnId, undefined); + }), + ); + + it.effect("steers concurrent sendTurn into one active turn id", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-steer"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + const modelSelection = createModelSelection( + ProviderInstanceId.make("kilo"), + "anthropic/claude-sonnet-4-5", + ); + const first = yield* adapter.sendTurn({ + threadId, + input: "first", + modelSelection, + }); + const second = yield* adapter.sendTurn({ + threadId, + input: "second", + modelSelection, + }); + + NodeAssert.equal(String(second.turnId), String(first.turnId)); + NodeAssert.equal(runtimeMock.state.promptCalls.length, 2); + }), + ); + + it.effect("maps permission decisions to once/always/reject", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-permission"); + const requestId = ApprovalRequestId.make("perm-1"); + + runtimeMock.state.subscribedEvents = [ + { + type: "permission.asked", + properties: { + id: requestId, + sessionID: "http://127.0.0.1:4301/session", + permission: "bash", + patterns: ["ls"], + metadata: {}, + always: [], + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second")); + + yield* adapter.respondToRequest(threadId, requestId, "accept"); + NodeAssert.deepEqual(runtimeMock.state.permissionReplies, [ + { requestID: requestId, reply: "once" }, + ]); + + yield* adapter.respondToRequest(threadId, requestId, "acceptForSession"); + NodeAssert.deepEqual(runtimeMock.state.permissionReplies.at(-1), { + requestID: requestId, + reply: "always", + }); + + yield* adapter.respondToRequest(threadId, requestId, "decline"); + NodeAssert.deepEqual(runtimeMock.state.permissionReplies.at(-1), { + requestID: requestId, + reply: "reject", + }); + }), + ); + + it.effect("rolls back session state when sendTurn fails", () => + Effect.gen(function* () { + const adapter = yield* KiloAdapter; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId: asThreadId("thread-send-turn-failure"), + runtimeMode: "full-access", + }); + + runtimeMock.state.promptAsyncError = new Error("prompt failed"); + const error = yield* adapter + .sendTurn({ + threadId: asThreadId("thread-send-turn-failure"), + input: "Fix it", + modelSelection: { + instanceId: ProviderInstanceId.make("kilo"), + model: "openai/gpt-5", + }, + }) + .pipe(Effect.flip); + const sessions = yield* adapter.listSessions(); + + NodeAssert.equal(error._tag, "ProviderAdapterRequestError"); + if (error._tag === "ProviderAdapterRequestError") { + NodeAssert.match(error.detail, /prompt failed/); + } + NodeAssert.equal(sessions[0]?.status, "ready"); + NodeAssert.equal(sessions[0]?.activeTurnId, undefined); + }), + ); + + it.effect( + "does not call session.abort when interrupting a stale turn id", + Effect.fn(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-stale-interrupt"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + // Open a turn so the adapter claims an activeTurnId. + const first = yield* adapter.sendTurn({ + threadId, + input: "first", + modelSelection: createModelSelection( + ProviderInstanceId.make("kilo"), + "anthropic/claude-sonnet-4-5", + ), + }); + + // Simulate that an interrupting client targets a turn id that no + // longer matches the active claim. The adapter must NOT issue + // session.abort because the SDK aborts every turn in the session, + // which would cancel the newer running turn. + runtimeMock.state.abortCalls.length = 0; + yield* adapter.interruptTurn(threadId, TurnId.make("stale-turn-id")); + NodeAssert.deepEqual(runtimeMock.state.abortCalls, []); + + // The next legitimate interrupt on the live turn id should still + // resolve correctly (no leftover state from the stale call). + runtimeMock.state.abortCalls.length = 0; + yield* adapter.interruptTurn(threadId, first.turnId); + NodeAssert.ok( + (runtimeMock.state.abortCalls as ReadonlyArray).includes( + "http://127.0.0.1:4301/session", + ), + ); + }), + ); + + it.effect( + "keeps the active turn claim during in-flight session.abort so a concurrent sendTurn steers instead of opening a new turn", + Effect.fn(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-abort-race"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + const modelSelection = createModelSelection( + ProviderInstanceId.make("kilo"), + "anthropic/claude-sonnet-4-5", + ); + const first = yield* adapter.sendTurn({ + threadId, + input: "first", + modelSelection, + }); + + // Install an "abort entered" callback so we can wait until + // `interruptTurn` has reached the in-flight abort RPC, then a + // gate that holds the RPC open until we are ready. + runtimeMock.state.abortCalls.length = 0; + let releaseAbort!: () => void; + let abortReached = false; + let abortEnteredDeferred!: (value: void | PromiseLike) => void; + const abortEnteredPromise = new Promise((resolve) => { + abortEnteredDeferred = resolve; + }); + runtimeMock.state.abortGate = { + promise: new Promise((resolve) => { + releaseAbort = resolve; + }), + }; + runtimeMock.state.abortEnteredCallbacks.push(() => { + abortReached = true; + abortEnteredDeferred(); + }); + + const interruptFiber = yield* adapter + .interruptTurn(threadId, first.turnId) + .pipe(Effect.forkChild); + + // Wait (with a generous timeout) until the abort RPC has been + // invoked, proving that interruptTurn is parked inside it. + yield* Effect.tryPromise({ + try: () => abortEnteredPromise, + catch: () => new Error("abort RPC was never entered"), + }).pipe(Effect.timeout("2 seconds")); + + // While the abort RPC is still in flight, run a concurrent + // sendTurn. With the fix in place, the adapter MUST keep + // `activeTurnId` set so concurrent `sendTurn` steers into the + // in-flight turn id rather than opening a new one that the + // session-wide `session.abort` would unintentionally cancel. + const second = yield* adapter.sendTurn({ + threadId, + input: "second", + modelSelection, + }); + NodeAssert.equal(String(second.turnId), String(first.turnId)); + NodeAssert.equal(runtimeMock.state.promptCalls.length, 2); + + // Release the abort RPC and let interruptTurn complete. + releaseAbort(); + yield* Fiber.join(interruptFiber); + + // Session should be ready with no active turn after the abort + // settles. + const sessions = yield* adapter.listSessions(); + NodeAssert.equal(sessions[0]?.status, "ready"); + NodeAssert.equal(sessions[0]?.activeTurnId, undefined); + NodeAssert.equal(abortReached, true); + }), + ); + + it.effect( + "idle handler does not clear a turn id that is no longer owned", + Effect.fn(function* () { + const adapter = yield* KiloAdapter; + const threadId = asThreadId("thread-kilo-idle-race"); + + // Pre-stage an idle event so the SSE stream yields it when the + // event pump subscribes. We send the idle event with a STALE turn + // id (different from the one sendTurn will claim) so the handler's + // compare-and-set must reject it. Because the mock SSE stream is + // consumed by the event pump before sendTurn runs, the observed + // activeTurnId at the top of the idle handler is undefined, so the + // `if (turnId)` guard short-circuits. To exercise the + // compare-and-set branch we instead drive the handler through a + // dedicated session that does not start any turn. + runtimeMock.state.subscribedEvents = [ + { + type: "session.status", + properties: { + sessionID: "http://127.0.0.1:4301/session", + status: { type: "idle" }, + }, + }, + ]; + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("kilo"), + threadId, + runtimeMode: "full-access", + }); + + // After consuming the idle event with no active turn, listSessions + // should still report a ready session with no activeTurnId. The + // session.started / thread.started events should still be emitted. + const sessions = yield* adapter.listSessions(); + NodeAssert.equal(sessions[0]?.status, "ready"); + NodeAssert.equal(sessions[0]?.activeTurnId, undefined); + }), + ); + + it.effect( + "surfaces unknown-session errors as typed failures, not defects", + Effect.fn(function* () { + const adapter = yield* KiloAdapter; + const missingThreadId = asThreadId("thread-kilo-missing"); + + const stopError = yield* adapter.stopSession(missingThreadId).pipe(Effect.flip); + NodeAssert.equal(stopError._tag, "ProviderAdapterSessionNotFoundError"); + + const interruptError = yield* adapter.interruptTurn(missingThreadId).pipe(Effect.flip); + NodeAssert.equal(interruptError._tag, "ProviderAdapterSessionNotFoundError"); + + const readError = yield* adapter.readThread(missingThreadId).pipe(Effect.flip); + NodeAssert.equal(readError._tag, "ProviderAdapterSessionNotFoundError"); + + const rollbackError = yield* adapter.rollbackThread(missingThreadId, 1).pipe(Effect.flip); + NodeAssert.equal(rollbackError._tag, "ProviderAdapterSessionNotFoundError"); + + const respondError = yield* adapter + .respondToRequest(missingThreadId, ApprovalRequestId.make("perm"), "accept") + .pipe(Effect.flip); + NodeAssert.equal(respondError._tag, "ProviderAdapterSessionNotFoundError"); + }), + ); +}); diff --git a/apps/server/src/provider/Layers/KiloAdapter.ts b/apps/server/src/provider/Layers/KiloAdapter.ts new file mode 100644 index 00000000000..bb23d792c96 --- /dev/null +++ b/apps/server/src/provider/Layers/KiloAdapter.ts @@ -0,0 +1,1645 @@ +import { + EventId, + type KiloSettings, + ProviderDriverKind, + ProviderInstanceId, + type ProviderRuntimeEvent, + type ProviderSession, + RuntimeItemId, + RuntimeRequestId, + ThreadId, + type ToolLifecycleItemType, + TurnId, + type UserInputQuestion, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Result from "effect/Result"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import type { KiloClient, Part, PermissionRequest, QuestionRequest } from "@kilocode/sdk/v2"; +import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; + +import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; +import { + ProviderAdapterProcessError, + ProviderAdapterRequestError, + ProviderAdapterSessionClosedError, + ProviderAdapterSessionNotFoundError, + ProviderAdapterValidationError, +} from "../Errors.ts"; +import { type KiloAdapterShape } from "../Services/KiloAdapter.ts"; +import { + buildKiloPermissionRules, + KiloRuntime, + KiloRuntimeError, + kiloQuestionId, + kiloRuntimeErrorDetail, + parseKiloModelSlug, + runKiloSdk, + toKiloFileParts, + toKiloPermissionReply, + toKiloQuestionAnswers, + resolveKiloAgent, + type KiloServerConnection, +} from "../kiloRuntime.ts"; +import * as Option from "effect/Option"; + +const PROVIDER = ProviderDriverKind.make("kilo"); + +interface KiloTurnSnapshot { + readonly id: TurnId; + readonly items: Array; +} + +type KiloSubscribedEvent = + Awaited> extends { + readonly stream: AsyncIterable; + } + ? TEvent + : never; + +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 turns: Array; + /** + * Active turn claim. Stored in a Ref so concurrent `sendTurn` calls can + * atomically steer (reuse) vs open a new turn without both emitting + * `turn.started`. + */ + readonly activeTurnId: Ref.Ref; + activeAgent: string | undefined; + activeVariant: string | undefined; + /** + * One-shot guard flipped by `stopKiloContext` / `emitUnexpectedExit`. + * The session lifecycle is owned by `sessionScope`; this Ref exists only + * so concurrent callers can race the transition safely via `getAndSet`. + */ + readonly stopped: Ref.Ref; + /** + * Sole lifecycle handle for the session. Closing this scope: + * - aborts the `AbortController` registered as a finalizer + * (cancels the in-flight `event.subscribe` fetch), + * - interrupts the event-pump and server-exit fibers forked + * via `Effect.forkIn(sessionScope)`, + * - tears down the Kilo server process for scope-owned servers. + */ + readonly sessionScope: Scope.Closeable; +} + +export interface KiloAdapterLiveOptions { + readonly instanceId?: ProviderInstanceId; + readonly environment?: NodeJS.ProcessEnv; + readonly nativeEventLogPath?: string; + readonly nativeEventLogger?: EventNdjsonLogger; +} + +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +/** + * Map a tagged KiloRuntimeError produced by {@link runKiloSdk} into + * the adapter-boundary `ProviderAdapterRequestError`. SDK-method-level call + * sites pipe through this in `Effect.mapError` so they never build the error + * shape by hand. + */ +const toRequestError = (cause: KiloRuntimeError): ProviderAdapterRequestError => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: cause.operation, + detail: cause.detail, + cause: cause.cause, + }); + +/** + * Map a `Cause.squash`-ed failure into a `ProviderAdapterProcessError`. The + * typed cause is usually an `KiloRuntimeError` (from {@link runKiloSdk}), + * in which case we preserve its `detail`; otherwise we fall back to + * {@link kiloRuntimeErrorDetail} for unknown causes (defects, etc.). + */ +const toProcessError = (threadId: ThreadId, cause: unknown): ProviderAdapterProcessError => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId, + detail: KiloRuntimeError.is(cause) ? cause.detail : kiloRuntimeErrorDetail(cause), + cause, + }); + +type EventBaseInput = { + readonly threadId: ThreadId; + readonly turnId?: TurnId | undefined; + readonly itemId?: string | undefined; + readonly requestId?: string | undefined; + readonly createdAt?: string | undefined; + readonly raw?: unknown; +}; + +function toToolLifecycleItemType(toolName: string): ToolLifecycleItemType { + const normalized = toolName.toLowerCase(); + if (normalized.includes("bash") || normalized.includes("command")) { + return "command_execution"; + } + if ( + normalized.includes("edit") || + normalized.includes("write") || + normalized.includes("patch") || + normalized.includes("multiedit") + ) { + return "file_change"; + } + if (normalized.includes("web")) { + return "web_search"; + } + if (normalized.includes("mcp")) { + return "mcp_tool_call"; + } + if (normalized.includes("image")) { + return "image_view"; + } + if ( + normalized.includes("task") || + normalized.includes("agent") || + normalized.includes("subtask") + ) { + return "collab_agent_tool_call"; + } + return "dynamic_tool_call"; +} + +function mapPermissionToRequestType( + permission: string, +): "command_execution_approval" | "file_read_approval" | "file_change_approval" | "unknown" { + switch (permission) { + case "bash": + return "command_execution_approval"; + case "read": + return "file_read_approval"; + case "edit": + return "file_change_approval"; + default: + return "unknown"; + } +} + +function mapPermissionDecision(reply: "once" | "always" | "reject"): string { + switch (reply) { + case "once": + return "accept"; + case "always": + return "acceptForSession"; + case "reject": + default: + return "decline"; + } +} + +function resolveTurnSnapshot(context: KiloSessionContext, turnId: TurnId): KiloTurnSnapshot { + const existing = context.turns.find((turn) => turn.id === turnId); + if (existing) { + return existing; + } + + const created: KiloTurnSnapshot = { id: turnId, items: [] }; + context.turns.push(created); + return created; +} + +function appendTurnItem( + context: KiloSessionContext, + turnId: TurnId | undefined, + item: unknown, +): void { + if (!turnId) { + return; + } + resolveTurnSnapshot(context, turnId).items.push(item); +} + +// Yields typed adapter failures on the Effect error channel so callers using +// `Effect.gen` see them as recoverable `ProviderAdapterError`s instead of +// defects. All call sites in this module invoke it via `yield*`. +const ensureSessionContext = Effect.fn("ensureSessionContext")(function* ( + sessions: ReadonlyMap, + threadId: ThreadId, +) { + const session = sessions.get(threadId); + if (!session) { + return yield* new ProviderAdapterSessionNotFoundError({ + provider: PROVIDER, + threadId, + }); + } + if (yield* Ref.get(session.stopped)) { + return yield* new ProviderAdapterSessionClosedError({ + provider: PROVIDER, + threadId, + }); + } + return session; +}); + +function normalizeQuestionRequest(request: QuestionRequest): ReadonlyArray { + return request.questions.map((question, index) => ({ + id: kiloQuestionId(index, question), + header: question.header, + question: question.question, + options: question.options.map((option) => ({ + label: option.label, + description: option.description, + })), + ...(question.multiple ? { multiSelect: true } : {}), + })); +} + +function resolveTextStreamKind(part: Part | undefined): "assistant_text" | "reasoning_text" { + return part?.type === "reasoning" ? "reasoning_text" : "assistant_text"; +} + +function textFromPart(part: Part): string | undefined { + switch (part.type) { + case "text": + case "reasoning": + return part.text; + default: + return undefined; + } +} + +function commonPrefixLength(left: string, right: string): number { + let index = 0; + while (index < left.length && index < right.length && left[index] === right[index]) { + index += 1; + } + return index; +} + +function resolveLatestAssistantText(previousText: string | undefined, nextText: string): string { + if (previousText && previousText.length > nextText.length && previousText.startsWith(nextText)) { + return previousText; + } + return nextText; +} + +export function mergeKiloAssistantText( + previousText: string | undefined, + nextText: string, +): { + readonly latestText: string; + readonly deltaToEmit: string; +} { + const latestText = resolveLatestAssistantText(previousText, nextText); + return { + latestText, + deltaToEmit: latestText.slice(commonPrefixLength(previousText ?? "", latestText)), + }; +} + +export function appendKiloAssistantTextDelta( + previousText: string, + delta: string, +): { + readonly nextText: string; + readonly deltaToEmit: string; +} { + return { + nextText: previousText + delta, + deltaToEmit: delta, + }; +} + +const isoFromEpochMs = (value: number) => + DateTime.make(value).pipe( + Option.match({ + onNone: () => undefined, + onSome: DateTime.formatIso, + }), + ); + +function messageRoleForPart( + context: KiloSessionContext, + part: Pick, +): "assistant" | "user" | undefined { + const known = context.messageRoleById.get(part.messageID); + if (known) { + return known; + } + return part.type === "tool" ? "assistant" : undefined; +} + +function detailFromToolPart(part: Extract): string | undefined { + switch (part.state.status) { + case "completed": + return part.state.output; + case "error": + return part.state.error; + case "running": + return part.state.title; + default: + return undefined; + } +} + +function toolStateCreatedAt(part: Extract): string | undefined { + switch (part.state.status) { + case "running": + return isoFromEpochMs(part.state.time.start); + case "completed": + case "error": + return isoFromEpochMs(part.state.time.end); + default: + return undefined; + } +} + +function sessionErrorMessage(error: unknown): string { + if (!error || typeof error !== "object") { + return "Kilo session failed."; + } + const data = "data" in error && error.data && typeof error.data === "object" ? error.data : null; + const message = data && "message" in data ? data.message : null; + return typeof message === "string" && message.trim().length > 0 + ? message + : "Kilo session failed."; +} + +function updateProviderSession( + context: KiloSessionContext, + patch: Partial, + options?: { + readonly clearActiveTurnId?: boolean; + readonly clearLastError?: boolean; + }, +): Effect.Effect { + return Effect.gen(function* () { + const updatedAt = yield* nowIso; + const nextSession = { + ...context.session, + ...patch, + updatedAt, + } as ProviderSession & Record; + const mutableSession = nextSession as Record; + if (options?.clearActiveTurnId) { + delete mutableSession.activeTurnId; + } + if (options?.clearLastError) { + delete mutableSession.lastError; + } + context.session = nextSession; + return nextSession; + }); +} + +const stopKiloContext = Effect.fn("stopKiloContext")(function* (context: KiloSessionContext) { + // Race-safe one-shot: first caller flips the flag, everyone else no-ops. + if (yield* Ref.getAndSet(context.stopped, true)) { + return false; + } + + // Best-effort remote abort. The scope close below tears down the local + // handles (event-pump fiber, server-exit fiber, event-subscribe fetch), + // but we still want to tell Kilo that this session is done. + yield* runKiloSdk("session.abort", () => + context.client.session.abort({ sessionID: context.kiloSessionId }), + ).pipe(Effect.ignore({ log: true })); + + // Closing the session scope interrupts every fiber forked into it and + // runs each finalizer we registered — the `AbortController.abort()` call, + // the child-process termination, etc. + yield* Scope.close(context.sessionScope, Exit.void); + return true; +}); + +export function makeKiloAdapter(kiloSettings: KiloSettings, options?: KiloAdapterLiveOptions) { + return Effect.gen(function* () { + const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("kilo"); + const serverConfig = yield* ServerConfig; + const kiloRuntime = yield* KiloRuntime; + const crypto = yield* Crypto.Crypto; + const nativeEventLogger = + options?.nativeEventLogger ?? + (options?.nativeEventLogPath !== undefined + ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { + stream: "native", + }) + : undefined); + // Only close loggers we created. If the caller passed one in via + // `options.nativeEventLogger`, they own its lifecycle. + const managedNativeEventLogger = + options?.nativeEventLogger === undefined ? nativeEventLogger : undefined; + const runtimeEvents = yield* Queue.unbounded(); + const sessions = new Map(); + const randomUUIDv4 = crypto.randomUUIDv4.pipe( + Effect.mapError( + (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "crypto/randomUUIDv4", + detail: "Failed to generate Kilo runtime identifier.", + cause, + }), + ), + ); + const buildEventBase = (input: EventBaseInput) => + Effect.all({ + eventId: randomUUIDv4.pipe(Effect.map(EventId.make)), + createdAt: input.createdAt === undefined ? nowIso : Effect.succeed(input.createdAt), + }).pipe( + Effect.map(({ eventId, createdAt }) => ({ + eventId, + provider: PROVIDER, + 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, + }, + } + : {}), + })), + ); + + // Layer-level finalizer: when the adapter layer shuts down, stop every + // session. Each session's `Scope.close` tears down its spawned Kilo + // server (via the `ChildProcessSpawner` finalizer installed in + // `startKiloServerProcess`) and interrupts the forked event/exit + // fibers. Consumers that can't reason about Effect scopes therefore + // cannot leak Kilo child processes by forgetting to call `stopAll`. + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + const contexts = [...sessions.values()]; + sessions.clear(); + // `ignoreCause` swallows both typed failures (none here) and defects + // from throwing scope finalizers so a sibling's death can't interrupt + // the remaining cleanups. + yield* Effect.forEach(contexts, (context) => Effect.ignoreCause(stopKiloContext(context)), { + concurrency: "unbounded", + discard: true, + }); + // Close the logger AFTER session teardown so any final lifecycle + // events emitted during shutdown still get written. `close` flushes + // the `Logger.batched` window and closes each per-thread + // `RotatingFileSink` handle owned by the logger's internal scope. + if (managedNativeEventLogger !== undefined) { + yield* managedNativeEventLogger.close(); + } + }).pipe(Effect.ensuring(Queue.shutdown(runtimeEvents))), + ); + + const emit = (event: ProviderRuntimeEvent) => + Queue.offer(runtimeEvents, event).pipe(Effect.asVoid); + const writeNativeEvent = ( + threadId: ThreadId, + event: { + readonly observedAt: string; + readonly event: Record; + }, + ) => (nativeEventLogger ? nativeEventLogger.write(event, threadId) : Effect.void); + const writeNativeEventBestEffort = ( + threadId: ThreadId, + event: { + readonly observedAt: string; + readonly event: Record; + }, + ) => writeNativeEvent(threadId, event).pipe(Effect.catchCause(() => Effect.void)); + + const emitUnexpectedExit = Effect.fn("emitUnexpectedExit")(function* ( + context: KiloSessionContext, + message: string, + ) { + // Atomic one-shot: two fibers can race here (the event-pump on stream + // failure and the server-exit watcher). `getAndSet` flips the flag in + // a single step so the loser observes `true` and returns; a plain + // `Ref.get` would let both racers slip past and emit duplicates. + if (yield* Ref.getAndSet(context.stopped, true)) { + return; + } + const turnId = yield* Ref.get(context.activeTurnId); + // Compare-and-delete: only evict our entry if we still own it. + // A concurrent `startSession` may have already replaced the entry; + // deleting by `threadId` would orphan that newly-started Kilo server. + if (sessions.get(context.session.threadId) === context) { + sessions.delete(context.session.threadId); + } + // Emit lifecycle events BEFORE tearing down the scope. Both call sites + // run this inside a fiber forked via `Effect.forkIn(context.sessionScope)`; + // closing that scope triggers the fiber-interrupt finalizer, so any + // subsequent yield point would unwind and silently drop these emits. + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + })), + type: "runtime.error", + payload: { + message, + class: "transport_error", + }, + }).pipe(Effect.ignore); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + })), + type: "session.exited", + payload: { + reason: message, + recoverable: false, + exitKind: "error", + }, + }).pipe(Effect.ignore); + // Inline the teardown that `stopKiloContext` would do; we can't + // delegate to it because our `getAndSet` above already flipped the + // one-shot guard, so the call would no-op. + yield* runKiloSdk("session.abort", () => + context.client.session.abort({ sessionID: context.kiloSessionId }), + ).pipe(Effect.ignore({ log: true })); + yield* Scope.close(context.sessionScope, Exit.void); + }); + + /** Emit content.delta and item.completed events for an assistant text part. */ + const emitAssistantTextDelta = Effect.fn("emitAssistantTextDelta")(function* ( + context: KiloSessionContext, + part: Part, + turnId: TurnId | undefined, + raw: unknown, + ) { + const text = textFromPart(part); + if (text === undefined) { + return; + } + const previousText = context.emittedTextByPartId.get(part.id); + const { latestText, deltaToEmit } = mergeKiloAssistantText(previousText, text); + context.emittedTextByPartId.set(part.id, latestText); + if (latestText !== text) { + context.partById.set( + part.id, + (part.type === "text" || part.type === "reasoning" + ? { ...part, text: latestText } + : part) satisfies Part, + ); + } + if (deltaToEmit.length > 0) { + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + itemId: part.id, + createdAt: + (part.type === "text" || part.type === "reasoning") && part.time !== undefined + ? isoFromEpochMs(part.time.start) + : undefined, + raw, + })), + type: "content.delta", + payload: { + streamKind: resolveTextStreamKind(part), + delta: deltaToEmit, + }, + }); + } + + if ( + part.type === "text" && + part.time?.end !== undefined && + !context.completedAssistantPartIds.has(part.id) + ) { + context.completedAssistantPartIds.add(part.id); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + itemId: part.id, + createdAt: isoFromEpochMs(part.time.end), + raw, + })), + type: "item.completed", + payload: { + itemType: "assistant_message", + status: "completed", + title: "Assistant message", + ...(latestText.length > 0 ? { detail: latestText } : {}), + }, + }); + } + }); + + const handleSubscribedEvent = Effect.fn("handleSubscribedEvent")(function* ( + context: KiloSessionContext, + event: KiloSubscribedEvent, + ) { + const payloadSessionId = + "properties" in event ? (event.properties as { sessionID?: unknown }).sessionID : undefined; + if (payloadSessionId !== context.kiloSessionId) { + return; + } + + const turnId = yield* Ref.get(context.activeTurnId); + yield* writeNativeEventBestEffort(context.session.threadId, { + observedAt: yield* nowIso, + event: { + provider: PROVIDER, + threadId: context.session.threadId, + providerThreadId: context.kiloSessionId, + type: event.type, + ...(turnId ? { turnId } : {}), + payload: event, + }, + }); + + switch (event.type) { + case "message.updated": { + context.messageRoleById.set(event.properties.info.id, event.properties.info.role); + if (event.properties.info.role === "assistant") { + for (const part of context.partById.values()) { + if (part.messageID !== event.properties.info.id) { + continue; + } + yield* emitAssistantTextDelta(context, part, turnId, event); + } + } + break; + } + + case "message.removed": { + context.messageRoleById.delete(event.properties.messageID); + break; + } + + case "message.part.delta": { + const existingPart = context.partById.get(event.properties.partID); + if (!existingPart) { + break; + } + const role = messageRoleForPart(context, existingPart); + if (role !== "assistant") { + break; + } + const streamKind = resolveTextStreamKind(existingPart); + const delta = event.properties.delta; + if (delta.length === 0) { + break; + } + const previousText = + context.emittedTextByPartId.get(event.properties.partID) ?? + textFromPart(existingPart) ?? + ""; + const { nextText, deltaToEmit } = appendKiloAssistantTextDelta(previousText, delta); + if (deltaToEmit.length === 0) { + break; + } + context.emittedTextByPartId.set(event.properties.partID, nextText); + if (existingPart.type === "text" || existingPart.type === "reasoning") { + context.partById.set(event.properties.partID, { + ...existingPart, + text: nextText, + }); + } + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + itemId: event.properties.partID, + raw: event, + })), + type: "content.delta", + payload: { + streamKind, + delta: deltaToEmit, + }, + }); + break; + } + + case "message.part.updated": { + const part = event.properties.part; + context.partById.set(part.id, part); + const messageRole = messageRoleForPart(context, part); + + if (messageRole === "assistant") { + yield* emitAssistantTextDelta(context, part, turnId, event); + } + + if (part.type === "tool") { + const itemType = toToolLifecycleItemType(part.tool); + const title = + part.state.status === "running" ? (part.state.title ?? part.tool) : part.tool; + const detail = detailFromToolPart(part); + const payload = { + itemType, + ...(part.state.status === "error" + ? { status: "failed" as const } + : part.state.status === "completed" + ? { status: "completed" as const } + : { status: "inProgress" as const }), + ...(title ? { title } : {}), + ...(detail ? { detail } : {}), + data: { + tool: part.tool, + state: part.state, + }, + }; + const runtimeEvent: ProviderRuntimeEvent = { + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + itemId: part.callID, + createdAt: toolStateCreatedAt(part), + raw: event, + })), + type: + part.state.status === "pending" + ? "item.started" + : part.state.status === "completed" || part.state.status === "error" + ? "item.completed" + : "item.updated", + payload, + }; + appendTurnItem(context, turnId, part); + yield* emit(runtimeEvent); + } + break; + } + + case "permission.asked": { + context.pendingPermissions.set(event.properties.id, event.properties); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + requestId: event.properties.id, + raw: event, + })), + type: "request.opened", + payload: { + requestType: mapPermissionToRequestType(event.properties.permission), + detail: + event.properties.patterns.length > 0 + ? 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* buildEventBase({ + threadId: context.session.threadId, + turnId, + requestId: event.properties.requestID, + raw: event, + })), + type: "request.resolved", + payload: { + requestType: "unknown", + decision: mapPermissionDecision(event.properties.reply), + }, + }); + break; + } + + case "question.asked": { + context.pendingQuestions.set(event.properties.id, event.properties); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + requestId: event.properties.id, + raw: event, + })), + type: "user-input.requested", + payload: { + questions: normalizeQuestionRequest(event.properties), + }, + }); + break; + } + + case "question.replied": { + const request = context.pendingQuestions.get(event.properties.requestID); + context.pendingQuestions.delete(event.properties.requestID); + const answers = Object.fromEntries( + (request?.questions ?? []).map((question, index) => [ + kiloQuestionId(index, question), + event.properties.answers[index]?.join(", ") ?? "", + ]), + ); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + requestId: event.properties.requestID, + raw: event, + })), + type: "user-input.resolved", + payload: { answers }, + }); + break; + } + + case "question.rejected": { + context.pendingQuestions.delete(event.properties.requestID); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + requestId: event.properties.requestID, + raw: event, + })), + type: "user-input.resolved", + payload: { answers: {} }, + }); + break; + } + + case "session.status": { + if (event.properties.status.type === "busy") { + // Read the CURRENT active turn id, not the snapshot taken at the + // top of this handler. `interruptTurn` may have already cleared + // the claim by the time this late `busy` event arrives; reusing + // the stale snapshot would resurrect an aborted turn and rewrite + // `activeTurnId` for an already-interrupted session. + const busyTurnId = yield* Ref.get(context.activeTurnId); + if (busyTurnId) { + yield* updateProviderSession(context, { + status: "running", + activeTurnId: busyTurnId, + }); + } + } + + if (event.properties.status.type === "retry") { + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + raw: event, + })), + type: "runtime.warning", + payload: { + message: event.properties.status.message, + detail: event.properties.status, + }, + }); + break; + } + + if (event.properties.status.type === "idle" && turnId) { + // Compare-and-set: only clear and emit `turn.completed` if we + // still own the turn id we observed at the top of this handler. + // A concurrent interrupt / steer may have already taken over the + // slot (or cleared it) by the time the SDK idle event arrives — + // unconditionally wiping the ref would orphan a still-running + // turn or drop a legitimate `turn.completed` for a newer turn. + const stillOwnsTurn = yield* Ref.modify(context.activeTurnId, (current) => { + if (current === turnId) { + return [true, undefined] as const; + } + return [false, current] as const; + }); + if (stillOwnsTurn) { + yield* updateProviderSession( + context, + { status: "ready" }, + { clearActiveTurnId: true }, + ); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + raw: event, + })), + type: "turn.completed", + payload: { + state: "completed", + }, + }); + } + } + break; + } + + case "session.error": { + const message = sessionErrorMessage(event.properties.error); + const activeTurnId = yield* Ref.getAndSet(context.activeTurnId, undefined); + yield* updateProviderSession( + context, + { + status: "error", + lastError: message, + }, + { clearActiveTurnId: true }, + ); + if (activeTurnId) { + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId: activeTurnId, + raw: event, + })), + type: "turn.completed", + payload: { + state: "failed", + errorMessage: message, + }, + }); + } + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + raw: event, + })), + type: "runtime.error", + payload: { + message, + class: "provider_error", + detail: event.properties.error, + }, + }); + break; + } + + default: + break; + } + }); + + const startEventPump = Effect.fn("startEventPump")(function* (context: KiloSessionContext) { + // One AbortController per session scope. The finalizer fires when + // the scope closes (explicit stop, unexpected exit, or layer + // shutdown) and cancels the in-flight `event.subscribe` fetch so + // the async iterable unwinds cleanly. + const eventsAbortController = new AbortController(); + yield* Scope.addFinalizer( + context.sessionScope, + Effect.sync(() => eventsAbortController.abort()), + ); + + // Fibers forked into `context.sessionScope` are interrupted + // automatically when the scope closes — no bookkeeping required. + yield* Effect.flatMap( + runKiloSdk("event.subscribe", () => + context.client.event.subscribe(undefined, { + signal: eventsAbortController.signal, + }), + ), + (subscription) => + Stream.fromAsyncIterable( + subscription.stream, + (cause) => + new KiloRuntimeError({ + operation: "event.subscribe", + detail: kiloRuntimeErrorDetail(cause), + cause, + }), + ).pipe(Stream.runForEach((event) => handleSubscribedEvent(context, event))), + ).pipe( + Effect.exit, + Effect.flatMap((exit) => + Effect.gen(function* () { + // Expected paths: caller aborted the fetch or the session + // has already been marked stopped. Treat as a clean exit. + // Clean stream completion is ignored here (same as OpenCode): + // production SSE is long-lived and only ends on abort/failure. + if (eventsAbortController.signal.aborted || (yield* Ref.get(context.stopped))) { + return; + } + if (Exit.isFailure(exit)) { + yield* emitUnexpectedExit(context, kiloRuntimeErrorDetail(Cause.squash(exit.cause))); + } + }), + ), + Effect.forkIn(context.sessionScope), + ); + + if (!context.server.external && context.server.exitCode !== null) { + yield* context.server.exitCode.pipe( + Effect.flatMap((code) => + Effect.gen(function* () { + if (yield* Ref.get(context.stopped)) { + return; + } + yield* emitUnexpectedExit(context, `Kilo server exited unexpectedly (${code}).`); + }), + ), + Effect.forkIn(context.sessionScope), + ); + } + }); + + const startSession: KiloAdapterShape["startSession"] = Effect.fn("startSession")( + function* (input) { + const binaryPath = kiloSettings.binaryPath; + const directory = input.cwd ?? serverConfig.cwd; + const existing = sessions.get(input.threadId); + if (existing) { + yield* stopKiloContext(existing); + // Only remove the map entry if it still points at the context we + // stopped — a concurrent startSession may already have replaced it. + if (sessions.get(input.threadId) === existing) { + sessions.delete(input.threadId); + } + } + + const started = yield* Effect.gen(function* () { + const sessionScope = yield* Scope.make(); + const startedExit = yield* Effect.exit( + Effect.gen(function* () { + // The runtime binds the server's lifetime to the Scope.Scope + // we provide below — closing `sessionScope` kills the child + // process automatically. No manual `server.close()` needed. + const server = yield* kiloRuntime.connectToKiloServer({ + binaryPath, + ...(options?.environment ? { environment: options.environment } : {}), + }); + const client = kiloRuntime.createKiloSdkClient({ + baseUrl: server.url, + directory, + serverPassword: server.password, + }); + const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); + if (mcpSession && !server.external) { + yield* runKiloSdk("mcp.add", () => + client.mcp.add({ + name: "t3-code", + config: { + type: "remote", + url: mcpSession.endpoint, + headers: { + Authorization: mcpSession.authorizationHeader, + }, + oauth: false, + }, + }), + ); + } + const kiloSession = yield* runKiloSdk("session.create", () => + client.session.create({ + title: `T3 Code ${input.threadId}`, + agent: "code", + permission: buildKiloPermissionRules(input.runtimeMode), + }), + ); + if (!kiloSession.data) { + return yield* new KiloRuntimeError({ + operation: "session.create", + detail: "Kilo session.create returned no session payload.", + }); + } + return { + sessionScope, + server, + client, + kiloSession: kiloSession.data, + }; + }).pipe(Effect.provideService(Scope.Scope, sessionScope)), + ); + if (Exit.isFailure(startedExit)) { + yield* Scope.close(sessionScope, Exit.void).pipe(Effect.ignore); + return yield* toProcessError(input.threadId, Cause.squash(startedExit.cause)); + } + return startedExit.value; + }); + + // Guard against a concurrent startSession call that may have raced + // and already inserted a session while we were awaiting async work. + // Compare-and-set: only the first writer owns the map entry. + const raceWinner = sessions.get(input.threadId); + if (raceWinner) { + // Another call won the race – clean up the session we just created + // (including the remote SDK session) and return the existing one. + yield* runKiloSdk("session.abort", () => + started.client.session.abort({ + sessionID: started.kiloSession.id, + }), + ).pipe(Effect.ignore); + yield* Scope.close(started.sessionScope, Exit.void).pipe(Effect.ignore); + return raceWinner.session; + } + + const createdAt = yield* nowIso; + const session: ProviderSession = { + provider: PROVIDER, + providerInstanceId: boundInstanceId, + status: "ready", + runtimeMode: input.runtimeMode, + cwd: directory, + ...(input.modelSelection ? { model: input.modelSelection.model } : {}), + threadId: input.threadId, + createdAt, + updatedAt: createdAt, + }; + + const context: KiloSessionContext = { + session, + client: started.client, + server: started.server, + directory, + kiloSessionId: started.kiloSession.id, + pendingPermissions: new Map(), + pendingQuestions: new Map(), + partById: new Map(), + emittedTextByPartId: new Map(), + messageRoleById: new Map(), + completedAssistantPartIds: new Set(), + turns: [], + activeTurnId: yield* Ref.make(undefined), + activeAgent: undefined, + activeVariant: undefined, + stopped: yield* Ref.make(false), + sessionScope: started.sessionScope, + }; + // Re-check immediately before insert so two post-await winners cannot + // both `set` and leak a managed server. + const lateRaceWinner = sessions.get(input.threadId); + if (lateRaceWinner) { + yield* runKiloSdk("session.abort", () => + started.client.session.abort({ + sessionID: started.kiloSession.id, + }), + ).pipe(Effect.ignore); + yield* Scope.close(started.sessionScope, Exit.void).pipe(Effect.ignore); + return lateRaceWinner.session; + } + sessions.set(input.threadId, context); + yield* startEventPump(context); + + // Event-pump / server-exit fibers can tear the session down before we + // resume. Do not advertise a started session that is already gone. + if ((yield* Ref.get(context.stopped)) || sessions.get(input.threadId) !== context) { + return yield* new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: "Kilo session exited during startup.", + }); + } + + yield* emit({ + ...(yield* buildEventBase({ threadId: input.threadId })), + type: "session.started", + payload: { + message: "Kilo session started", + }, + }); + yield* emit({ + ...(yield* buildEventBase({ threadId: input.threadId })), + type: "thread.started", + payload: { + providerThreadId: started.kiloSession.id, + }, + }); + + return session; + }, + ); + + const sendTurn: KiloAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) { + const context = yield* ensureSessionContext(sessions, input.threadId); + const modelSelection = + input.modelSelection ?? + (context.session.model + ? { instanceId: boundInstanceId, model: context.session.model } + : undefined); + if (modelSelection !== undefined && modelSelection.instanceId !== boundInstanceId) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: `Kilo model selection is bound to instance '${modelSelection?.instanceId}', expected '${boundInstanceId}'.`, + }); + } + const parsedModel = parseKiloModelSlug(modelSelection?.model); + if (!parsedModel) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Kilo model selection must use the 'provider/model' format.", + }); + } + + const text = input.input?.trim(); + const fileParts = toKiloFileParts({ + attachments: input.attachments, + resolveAttachmentPath: (attachment) => + resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }), + }); + if ((!text || text.length === 0) && fileParts.length === 0) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Kilo turns require text input or at least one attachment.", + }); + } + + const variant = getModelSelectionStringOptionValue(modelSelection, "variant"); + + // Atomic claim: concurrent sendTurn on an idle session must not both + // open new turns. First caller wins and emits turn.started; later + // callers steer into that turn id. + const candidateTurnId = TurnId.make(`kilo-turn-${yield* randomUUIDv4}`); + type TurnClaim = { readonly turnId: TurnId; readonly isSteer: boolean }; + const claimed = yield* Ref.modify(context.activeTurnId, (current): [TurnClaim, TurnId] => { + if (current !== undefined) { + return [{ turnId: current, isSteer: true }, current]; + } + return [{ turnId: candidateTurnId, isSteer: false }, candidateTurnId]; + }); + const turnId = claimed.turnId; + const isSteer = claimed.isSteer; + + context.activeAgent = resolveKiloAgent({ interactionMode: input.interactionMode }); + context.activeVariant = variant; + yield* updateProviderSession( + context, + { + status: "running", + activeTurnId: turnId, + model: modelSelection?.model ?? context.session.model, + }, + { clearLastError: true }, + ); + + if (!isSteer) { + yield* emit({ + ...(yield* buildEventBase({ threadId: input.threadId, turnId })), + type: "turn.started", + payload: { + model: modelSelection?.model ?? context.session.model, + ...(variant ? { effort: variant } : {}), + }, + }); + } + + yield* runKiloSdk("session.promptAsync", () => + context.client.session.promptAsync({ + sessionID: context.kiloSessionId, + model: parsedModel, + agent: context.activeAgent ?? "code", + ...(context.activeVariant ? { variant: context.activeVariant } : {}), + parts: [...(text ? [{ type: "text" as const, text }] : []), ...fileParts], + }), + ).pipe( + Effect.mapError(toRequestError), + // On failure of a fresh turn: clear active-turn state, flip the + // session back to ready with lastError set, emit turn.aborted, then + // let the typed error propagate. We don't need to rebuild the error + // here — `toRequestError` already produced the right shape. A failed + // steer leaves the still-running original turn untouched. + Effect.tapError((requestError) => + isSteer + ? Effect.void + : Effect.gen(function* () { + // Only clear when we still own the claim. A concurrent steer + // reuses this turn id; wiping it would abort still-running work. + const stillOwnsClaim = yield* Ref.modify(context.activeTurnId, (current) => { + if (current === turnId) { + return [true, undefined] as const; + } + return [false, current] as const; + }); + if (!stillOwnsClaim) { + return; + } + context.activeAgent = undefined; + context.activeVariant = undefined; + yield* updateProviderSession( + context, + { + status: "ready", + model: modelSelection?.model ?? context.session.model, + lastError: requestError.detail, + }, + { clearActiveTurnId: true }, + ); + // Emit turn.completed(failed) so orchestration ingestion clears + // the running turn (turn.aborted alone is not applied there). + yield* emit({ + ...(yield* buildEventBase({ + threadId: input.threadId, + turnId, + })), + type: "turn.completed", + payload: { + state: "failed", + errorMessage: requestError.detail, + }, + }); + }), + ), + ); + + return { + threadId: input.threadId, + turnId, + }; + }); + + const interruptTurn: KiloAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")( + function* (threadId, turnId) { + const context = yield* ensureSessionContext(sessions, threadId); + // Decide which turn id this call intends to abort. We only clear + // the active claim while owning it AND we observe current === target + // — a stale `turnId` (interrupted twice, or a newer turn already + // claimed) must NOT trigger `session.abort`, which cancels every + // turn for the SDK session including any unrelated newer one. + type InterruptClaim = { + readonly interruptedTurnId: TurnId | undefined; + readonly ownedClaim: boolean; + }; + const claim: InterruptClaim = yield* Ref.get(context.activeTurnId).pipe( + Effect.map((current): InterruptClaim => { + const target = turnId ?? current; + if (target === undefined) { + return { interruptedTurnId: undefined, ownedClaim: false }; + } + if (current === target) { + return { interruptedTurnId: target, ownedClaim: true }; + } + // Stale turnId: emit nothing for the wire but keep the live + // claim untouched. + return { interruptedTurnId: target, ownedClaim: false }; + }), + ); + const ownsClaimAndAborted = yield* claim.ownedClaim + ? Effect.gen(function* () { + // Keep `activeTurnId` set during the remote abort. The Kilo + // SDK's `session.abort` is session-wide: clearing the claim + // first opens a window where a concurrent `sendTurn` can + // claim a fresh turn id which the in-flight abort then + // unintentionally cancels. Leaving the claim in place makes + // concurrent `sendTurn` steer into the same (about-to-be + // aborted) turn instead of opening a new one. The claim is + // cleared below once the abort RPC has completed. + const abortResult = yield* runKiloSdk("session.abort", () => + context.client.session.abort({ sessionID: context.kiloSessionId }), + ).pipe(Effect.mapError(toRequestError), Effect.result); + if (Result.isSuccess(abortResult)) { + yield* Ref.set(context.activeTurnId, undefined); + return { aborted: true as const }; + } + return { + aborted: false as const, + abortError: abortResult.failure, + }; + }) + : Effect.succeed({ aborted: true as const }); + if (!ownsClaimAndAborted.aborted) { + // Abort failed: the claim was never cleared, so the session + // still reports `interruptTurn`'s target as the busy turn id + // (matching what a subsequent SDK idle/busy event will see). + // Emit turn.completed(failed) for the turn we tried to interrupt + // so orchestration can clear the half-interrupted running turn. + // Do NOT flip the session to `ready` — the turn is still + // considered busy from the orchestrator's perspective and the + // SDK may yet emit an idle/busy event that drives the state + // correctly. + const abortError: ProviderAdapterRequestError = ownsClaimAndAborted.abortError; + if (claim.interruptedTurnId) { + yield* emit({ + ...(yield* buildEventBase({ + threadId, + turnId: claim.interruptedTurnId, + })), + type: "turn.completed", + payload: { + state: "failed", + errorMessage: abortError.detail, + }, + }); + } + return yield* abortError; + } + // Abort succeeded and we just cleared the claim. Flip the session + // back to `ready` so the orchestration observes an idle thread. + // No concurrent sendTurn can race past our clear: any call that + // observed the set claim during the abort steered (or queued) + // into the (now aborted) turn, and clearing the ref after the + // abort returns ownership of subsequent turns to fresh sendTurn + // calls. + { + context.activeAgent = undefined; + context.activeVariant = undefined; + yield* updateProviderSession( + context, + { + status: "ready", + }, + { clearActiveTurnId: true }, + ); + } + if (claim.ownedClaim && claim.interruptedTurnId) { + yield* emit({ + ...(yield* buildEventBase({ + threadId, + turnId: claim.interruptedTurnId, + })), + type: "turn.aborted", + payload: { + reason: "Interrupted by user.", + }, + }); + yield* emit({ + ...(yield* buildEventBase({ + threadId, + turnId: claim.interruptedTurnId, + })), + type: "turn.completed", + payload: { + state: "interrupted", + }, + }); + } + }, + ); + + const respondToRequest: KiloAdapterShape["respondToRequest"] = Effect.fn("respondToRequest")( + function* (threadId, requestId, decision) { + const context = yield* ensureSessionContext(sessions, threadId); + if (!context.pendingPermissions.has(requestId)) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "permission.reply", + detail: `Unknown pending permission request: ${requestId}`, + }); + } + + yield* runKiloSdk("permission.reply", () => + context.client.permission.reply({ + requestID: requestId, + reply: toKiloPermissionReply(decision), + }), + ).pipe(Effect.mapError(toRequestError)); + }, + ); + + const respondToUserInput: KiloAdapterShape["respondToUserInput"] = Effect.fn( + "respondToUserInput", + )(function* (threadId, requestId, answers) { + const context = yield* ensureSessionContext(sessions, threadId); + const request = context.pendingQuestions.get(requestId); + if (!request) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "question.reply", + detail: `Unknown pending user-input request: ${requestId}`, + }); + } + + yield* runKiloSdk("question.reply", () => + context.client.question.reply({ + requestID: requestId, + answers: toKiloQuestionAnswers(request, answers), + }), + ).pipe(Effect.mapError(toRequestError)); + }); + + const stopSession: KiloAdapterShape["stopSession"] = Effect.fn("stopSession")( + function* (threadId) { + const context = sessions.get(threadId); + if (!context) { + return yield* new ProviderAdapterSessionNotFoundError({ + provider: PROVIDER, + threadId, + }); + } + const stopped = yield* stopKiloContext(context); + // Compare-and-delete: only evict our entry if we still own it. + // A concurrent `startSession` may have already replaced the entry; + // deleting by `threadId` would orphan that newly-started Kilo server. + if (sessions.get(threadId) === context) { + sessions.delete(threadId); + } + if (!stopped) { + return; + } + yield* emit({ + ...(yield* buildEventBase({ threadId })), + type: "session.exited", + payload: { + reason: "Session stopped.", + recoverable: false, + exitKind: "graceful", + }, + }); + }, + ); + + const listSessions: KiloAdapterShape["listSessions"] = () => + Effect.sync(() => [...sessions.values()].map((context) => context.session)); + + const hasSession: KiloAdapterShape["hasSession"] = (threadId) => + Effect.sync(() => sessions.has(threadId)); + + const readThread: KiloAdapterShape["readThread"] = Effect.fn("readThread")( + function* (threadId) { + const context = yield* ensureSessionContext(sessions, threadId); + const messages = yield* runKiloSdk("session.messages", () => + context.client.session.messages({ + sessionID: context.kiloSessionId, + }), + ).pipe(Effect.mapError(toRequestError)); + + const turns: Array = []; + for (const entry of messages.data ?? []) { + if (entry.info.role === "assistant") { + turns.push({ + id: TurnId.make(entry.info.id), + items: [entry.info, ...entry.parts], + }); + } + } + + return { + threadId, + turns, + }; + }, + ); + + const rollbackThread: KiloAdapterShape["rollbackThread"] = Effect.fn("rollbackThread")( + function* (threadId, numTurns) { + const context = yield* ensureSessionContext(sessions, threadId); + const messages = yield* runKiloSdk("session.messages", () => + context.client.session.messages({ + sessionID: context.kiloSessionId, + }), + ).pipe(Effect.mapError(toRequestError)); + + const assistantMessages = (messages.data ?? []).filter( + (entry) => entry.info.role === "assistant", + ); + const targetIndex = assistantMessages.length - numTurns - 1; + const target = targetIndex >= 0 ? assistantMessages[targetIndex] : null; + yield* runKiloSdk("session.revert", () => + context.client.session.revert({ + sessionID: context.kiloSessionId, + ...(target ? { messageID: target.info.id } : {}), + }), + ).pipe(Effect.mapError(toRequestError)); + + return yield* readThread(threadId); + }, + ); + + const stopAll: KiloAdapterShape["stopAll"] = () => + Effect.gen(function* () { + const contexts = [...sessions.values()]; + sessions.clear(); + // `stopKiloContext` is typed as never-failing — SDK aborts are + // already `Effect.ignore`'d inside it. `ignoreCause` here also + // swallows defects from throwing finalizers so one bad close can't + // interrupt the sibling fibers. Same pattern as the layer finalizer. + yield* Effect.forEach(contexts, (context) => Effect.ignoreCause(stopKiloContext(context)), { + concurrency: "unbounded", + discard: true, + }); + }); + + return { + provider: PROVIDER, + capabilities: { + sessionModelSwitch: "in-session", + }, + startSession, + sendTurn, + interruptTurn, + respondToRequest, + respondToUserInput, + stopSession, + listSessions, + hasSession, + readThread, + rollbackThread, + stopAll, + get streamEvents() { + return Stream.fromQueue(runtimeEvents); + }, + } 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..a996558feeb --- /dev/null +++ b/apps/server/src/provider/Layers/KiloProvider.test.ts @@ -0,0 +1,255 @@ +import * as NodeAssert from "node:assert/strict"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import { beforeEach, describe, it as vitestIt } from "vite-plus/test"; + +import { KiloSettings } from "@t3tools/contracts"; +import { ServerConfig } from "../../config.ts"; +import { + KiloRuntime, + KiloRuntimeError, + type KiloInventory, + type KiloRuntimeShape, +} from "../kiloRuntime.ts"; +import { checkKiloProviderStatus, flattenKiloModels } from "./KiloProvider.ts"; + +const decodeKiloSettings = Schema.decodeSync(KiloSettings); + +const DEFAULT_VERSION_STDOUT = "kilo 7.4.11\n"; + +const runtimeMock = { + state: { + runVersionError: null as Error | null, + versionStdout: DEFAULT_VERSION_STDOUT, + inventoryError: null as Error | null, + closeCalls: 0, + inventory: { + providerList: { connected: [] as string[], all: [] as unknown[], default: {}, failed: [] }, + agents: [] as unknown[], + } as unknown, + }, + reset() { + this.state.runVersionError = null; + this.state.versionStdout = DEFAULT_VERSION_STDOUT; + this.state.inventoryError = null; + this.state.closeCalls = 0; + this.state.inventory = { + providerList: { connected: [], all: [] as unknown[], default: {}, failed: [] }, + agents: [] as unknown[], + }; + }, +}; + +const KiloRuntimeTestDouble: KiloRuntimeShape = { + startKiloServerProcess: () => + Effect.succeed({ + url: "http://127.0.0.1:4301", + password: "test-password", + exitCode: Effect.never, + }), + connectToKiloServer: () => + Effect.gen(function* () { + yield* Effect.addFinalizer(() => + Effect.sync(() => { + runtimeMock.state.closeCalls += 1; + }), + ); + return { + url: "http://127.0.0.1:4301", + password: "test-password", + exitCode: null, + external: false, + }; + }), + runKiloCommand: () => + runtimeMock.state.runVersionError + ? Effect.fail( + new KiloRuntimeError({ + operation: "runKiloCommand", + detail: runtimeMock.state.runVersionError.message, + cause: runtimeMock.state.runVersionError, + }), + ) + : Effect.succeed({ stdout: runtimeMock.state.versionStdout, stderr: "", code: 0 }), + createKiloSdkClient: () => ({}) as unknown as ReturnType, + loadKiloInventory: () => + runtimeMock.state.inventoryError + ? Effect.fail( + new KiloRuntimeError({ + operation: "loadKiloInventory", + detail: runtimeMock.state.inventoryError.message, + cause: runtimeMock.state.inventoryError, + }), + ) + : Effect.succeed(runtimeMock.state.inventory as KiloInventory), +}; + +beforeEach(() => { + runtimeMock.reset(); +}); + +const testLayer = Layer.succeed(KiloRuntime, KiloRuntimeTestDouble).pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), +); + +const makeKiloSettings = (overrides?: Partial): KiloSettings => + decodeKiloSettings({ + enabled: true, + binaryPath: "kilo", + customModels: [], + ...overrides, + }); + +it.layer(testLayer)("checkKiloProviderStatus", (it) => { + it.effect("shows a clear missing binary message", () => + Effect.gen(function* () { + runtimeMock.state.runVersionError = new Error("spawn kilo ENOENT"); + const snapshot = yield* checkKiloProviderStatus(makeKiloSettings(), process.cwd()); + + NodeAssert.equal(snapshot.status, "error"); + NodeAssert.equal(snapshot.installed, false); + NodeAssert.equal(snapshot.message, "Kilo CLI (`kilo`) is not installed or not on PATH."); + }), + ); + + it.effect("returns disabled snapshot without probing", () => + Effect.gen(function* () { + const snapshot = yield* checkKiloProviderStatus( + makeKiloSettings({ enabled: false }), + process.cwd(), + ); + + NodeAssert.equal(snapshot.enabled, false); + NodeAssert.equal(snapshot.status, "disabled"); + NodeAssert.match(snapshot.message ?? "", /disabled/i); + NodeAssert.equal(runtimeMock.state.closeCalls, 0); + }), + ); + + it.effect("flattens connected upstream models as providerID/modelID", () => + Effect.gen(function* () { + runtimeMock.state.inventory = { + providerList: { + connected: ["anthropic", "openai"], + failed: [], + default: {}, + all: [ + { + id: "anthropic", + name: "Anthropic", + models: { + "claude-sonnet-4-5": { + id: "claude-sonnet-4-5", + name: "Claude Sonnet 4.5", + variants: {}, + }, + }, + }, + { + id: "openai", + name: "OpenAI", + models: { + "gpt-5": { id: "gpt-5", name: "GPT-5", variants: { medium: {}, high: {} } }, + }, + }, + { + id: "disconnected", + name: "Disconnected", + models: { + ignored: { id: "ignored", name: "Ignored", variants: {} }, + }, + }, + ], + }, + agents: [{ name: "code", mode: "primary", hidden: false }], + }; + + const snapshot = yield* checkKiloProviderStatus(makeKiloSettings(), process.cwd()); + NodeAssert.equal(snapshot.status, "ready"); + NodeAssert.equal(snapshot.installed, true); + const slugs = snapshot.models.map((model) => model.slug).toSorted(); + NodeAssert.deepEqual(slugs, ["anthropic/claude-sonnet-4-5", "openai/gpt-5"]); + const gpt = snapshot.models.find((model) => model.slug === "openai/gpt-5"); + NodeAssert.equal(gpt?.subProvider, "OpenAI"); + NodeAssert.ok( + !(gpt?.capabilities?.optionDescriptors ?? []).some( + (descriptor) => descriptor.id === "agent", + ), + ); + }), + ); + + it.effect("appends custom models", () => + Effect.gen(function* () { + runtimeMock.state.inventory = { + providerList: { + connected: ["anthropic"], + failed: [], + default: {}, + all: [ + { + id: "anthropic", + name: "Anthropic", + models: { + "claude-sonnet-4-5": { + id: "claude-sonnet-4-5", + name: "Claude Sonnet 4.5", + variants: {}, + }, + }, + }, + ], + }, + agents: [], + }; + + const snapshot = yield* checkKiloProviderStatus( + makeKiloSettings({ customModels: ["custom/my-model"] }), + process.cwd(), + ); + NodeAssert.ok(snapshot.models.some((model) => model.slug === "custom/my-model")); + }), + ); + + it.effect("warns when Kilo reports zero connected upstreams", () => + Effect.gen(function* () { + const snapshot = yield* checkKiloProviderStatus(makeKiloSettings(), process.cwd()); + NodeAssert.equal(snapshot.status, "warning"); + NodeAssert.match(snapshot.message ?? "", /did not report any connected upstream providers/); + }), + ); +}); + +describe("flattenKiloModels", () => { + vitestIt("skips disconnected providers", () => { + const models = flattenKiloModels({ + providerList: { + connected: ["a"], + failed: [], + default: {}, + all: [ + { + id: "a", + name: "A", + models: { m1: { id: "m1", name: "Model 1", variants: {} } }, + } as never, + { + id: "b", + name: "B", + models: { m2: { id: "m2", name: "Model 2", variants: {} } }, + } as never, + ], + }, + agents: [], + }); + NodeAssert.deepEqual( + models.map((model) => model.slug), + ["a/m1"], + ); + }); +}); diff --git a/apps/server/src/provider/Layers/KiloProvider.ts b/apps/server/src/provider/Layers/KiloProvider.ts new file mode 100644 index 00000000000..c846d92579c --- /dev/null +++ b/apps/server/src/provider/Layers/KiloProvider.ts @@ -0,0 +1,352 @@ +import { + type ModelCapabilities, + type KiloSettings, + type ServerProviderModel, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Data from "effect/Data"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { createModelCapabilities } from "@t3tools/shared/model"; +import { + buildServerProvider, + nonEmptyTrimmed, + parseGenericCliVersion, + providerModelsFromSettings, + type ServerProviderDraft, +} from "../providerSnapshot.ts"; +import { KiloRuntime, kiloRuntimeErrorDetail, type KiloInventory } from "../kiloRuntime.ts"; +import type { ProviderListResponse } from "@kilocode/sdk/v2"; + +const KILO_PRESENTATION = { + displayName: "Kilo", + showInteractionModeToggle: true, +} as const; + +class KiloProbeError extends Data.TaggedError("KiloProbeError")<{ + readonly cause: unknown; + readonly detail: string; +}> {} + +function normalizeProbeMessage(message: string): string | undefined { + const trimmed = message.trim(); + if (trimmed.length === 0) { + return undefined; + } + if ( + trimmed === "An error occurred in Effect.tryPromise" || + trimmed === "An error occurred in Effect.try" + ) { + return undefined; + } + return trimmed; +} + +function normalizedErrorMessage(cause: unknown): string | undefined { + if (cause instanceof KiloProbeError) { + return normalizeProbeMessage(cause.detail); + } + + if (!(cause instanceof Error)) { + return undefined; + } + + return normalizeProbeMessage(cause.message); +} + +function formatKiloProbeError(input: { readonly cause: unknown }): { + readonly installed: boolean; + readonly message: string; +} { + const detail = normalizedErrorMessage(input.cause); + const lower = detail?.toLowerCase() ?? ""; + + if (lower.includes("enoent") || lower.includes("notfound")) { + return { + installed: false, + message: "Kilo CLI (`kilo`) is not installed or not on PATH.", + }; + } + + if ( + lower.includes("401") || + lower.includes("403") || + lower.includes("unauthorized") || + lower.includes("forbidden") + ) { + return { + installed: true, + message: + "Kilo server rejected authentication. Restart T3 Code or check the Kilo CLI install.", + }; + } + + if (lower.includes("quarantine")) { + return { + installed: true, + message: + "macOS is blocking the Kilo binary (quarantine). Run `xattr -d com.apple.quarantine $(which kilo)` to fix this.", + }; + } + + if (lower.includes("invalid code signature") || lower.includes("corrupted")) { + return { + installed: true, + message: + "macOS killed the Kilo process due to an invalid code signature. The binary may be corrupted — try reinstalling Kilo.", + }; + } + + return { + installed: true, + message: detail + ? `Failed to execute Kilo CLI health check: ${detail}` + : "Failed to execute Kilo CLI health check.", + }; +} + +const DEFAULT_KILO_MODEL_CAPABILITIES: ModelCapabilities = createModelCapabilities({ + optionDescriptors: [], +}); + +function titleCaseSlug(value: string): string { + const segments: Array = []; + for (const segment of value.split(/[-_/]+/)) { + if (segment.length > 0) { + segments.push(segment.charAt(0).toUpperCase() + segment.slice(1)); + } + } + return segments.join(" "); +} + +function inferDefaultVariant( + providerID: string, + variants: ReadonlyArray, +): string | undefined { + if (variants.length === 1) { + return variants[0]; + } + if (providerID === "anthropic" || providerID.startsWith("google")) { + return variants.includes("high") ? "high" : undefined; + } + if (providerID === "openai" || providerID === "opencode" || providerID === "kilo") { + return variants.includes("medium") ? "medium" : variants.includes("high") ? "high" : undefined; + } + return undefined; +} + +function kiloCapabilitiesForModel(input: { + readonly providerID: string; + readonly model: ProviderListResponse["all"][number]["models"][string]; +}): ModelCapabilities { + const variantValues = Object.keys(input.model.variants ?? {}); + const defaultVariant = inferDefaultVariant(input.providerID, variantValues); + const variantOptions = variantValues.map((value) => + defaultVariant === value + ? { id: value, label: titleCaseSlug(value), isDefault: true as const } + : { id: value, label: titleCaseSlug(value) }, + ); + // v1: hide agents in model capabilities; default agent is fixed to `code`. + return createModelCapabilities({ + optionDescriptors: + variantOptions.length > 0 + ? [ + { + id: "variant", + label: "Variant", + type: "select" as const, + options: variantOptions, + ...(defaultVariant ? { currentValue: defaultVariant } : {}), + }, + ] + : [], + }); +} + +export function flattenKiloModels(input: KiloInventory): ReadonlyArray { + const connected = new Set(input.providerList.connected); + const models: Array = []; + + for (const provider of input.providerList.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: kiloCapabilitiesForModel({ + providerID: provider.id, + model, + }), + }); + } + } + + return models.toSorted((left, right) => left.name.localeCompare(right.name)); +} + +export const makePendingKiloProvider = ( + kiloSettings: KiloSettings, +): Effect.Effect => + Effect.gen(function* () { + const checkedAt = yield* Effect.map(DateTime.now, DateTime.formatIso); + const models = providerModelsFromSettings( + [], + kiloSettings.customModels, + DEFAULT_KILO_MODEL_CAPABILITIES, + ); + + if (!kiloSettings.enabled) { + return buildServerProvider({ + presentation: KILO_PRESENTATION, + enabled: false, + checkedAt, + models, + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Kilo is disabled in T3 Code settings.", + }, + }); + } + + return buildServerProvider({ + presentation: KILO_PRESENTATION, + enabled: true, + checkedAt, + models, + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Kilo provider status has not been checked in this session yet.", + }, + }); + }); + +export const checkKiloProviderStatus = Effect.fn("checkKiloProviderStatus")(function* ( + kiloSettings: KiloSettings, + cwd: string, + environment?: NodeJS.ProcessEnv, +): Effect.fn.Return { + const kiloRuntime = yield* KiloRuntime; + const resolvedEnvironment = environment ?? process.env; + const checkedAt = DateTime.formatIso(yield* DateTime.now); + const customModels = kiloSettings.customModels; + + const fallback = (cause: unknown, version: string | null = null) => { + const failure = formatKiloProbeError({ cause }); + return buildServerProvider({ + presentation: KILO_PRESENTATION, + enabled: kiloSettings.enabled, + checkedAt, + models: providerModelsFromSettings([], customModels, DEFAULT_KILO_MODEL_CAPABILITIES), + probe: { + installed: failure.installed, + version, + status: "error", + auth: { status: "unknown" }, + message: failure.message, + }, + }); + }; + + if (!kiloSettings.enabled) { + return buildServerProvider({ + presentation: KILO_PRESENTATION, + enabled: false, + checkedAt, + models: providerModelsFromSettings([], customModels, DEFAULT_KILO_MODEL_CAPABILITIES), + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown" }, + message: "Kilo is disabled in T3 Code settings.", + }, + }); + } + + const versionExit = yield* Effect.exit( + kiloRuntime + .runKiloCommand({ + binaryPath: kiloSettings.binaryPath, + args: ["--version"], + environment: resolvedEnvironment, + }) + .pipe( + Effect.mapError( + (cause) => new KiloProbeError({ cause, detail: kiloRuntimeErrorDetail(cause) }), + ), + ), + ); + 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* kiloRuntime.connectToKiloServer({ + binaryPath: kiloSettings.binaryPath, + environment: resolvedEnvironment, + }); + return yield* kiloRuntime.loadKiloInventory( + kiloRuntime.createKiloSdkClient({ + baseUrl: server.url, + directory: cwd, + serverPassword: server.password, + }), + ); + }).pipe( + Effect.mapError( + (cause) => new KiloProbeError({ cause, detail: kiloRuntimeErrorDetail(cause) }), + ), + ), + ), + ); + if (inventoryExit._tag === "Failure") { + return fallback(Cause.squash(inventoryExit.cause), version); + } + + const models = providerModelsFromSettings( + flattenKiloModels(inventoryExit.value), + customModels, + DEFAULT_KILO_MODEL_CAPABILITIES, + ); + const connectedCount = inventoryExit.value.providerList.connected.length; + return buildServerProvider({ + presentation: KILO_PRESENTATION, + enabled: true, + checkedAt, + models, + 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 384de852f9b..745b9c6fbe9 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -10,7 +10,7 @@ * * 2. **Many drivers, one registry** — the "all drivers slice" describe * block below configures one instance of every shipped driver - * (`codex`, `claudeAgent`, `cursor`, `grok`, `opencode`) in a single + * (`codex`, `claudeAgent`, `cursor`, `grok`, `opencode`, `kilo`) in a single * `ProviderInstanceConfigMap` and asserts the registry boots them all * without cross-contamination. This proves the driver SPI is uniform * across every provider — any driver plugs into the registry through @@ -18,7 +18,7 @@ * * Every instance in these tests is configured with `enabled: false` so the * provider-status checks short-circuit to pending/disabled snapshots - * without trying to spawn real `codex` / `claude` / `agent` / `grok` / `opencode` + * without trying to spawn real `codex` / `claude` / `agent` / `grok` / `opencode` / `kilo` * binaries. That keeps the assertions focused on registry routing * behaviour rather than the runtime details of each provider. */ @@ -29,6 +29,7 @@ import { type CodexSettings, type CursorSettings, type GrokSettings, + type KiloSettings, type OpenCodeSettings, ProviderDriverKind, type ProviderInstanceConfigMap, @@ -44,9 +45,13 @@ import { ClaudeDriver } from "../Drivers/ClaudeDriver.ts"; import { CodexDriver } from "../Drivers/CodexDriver.ts"; import { CursorDriver } from "../Drivers/CursorDriver.ts"; import { GrokDriver } from "../Drivers/GrokDriver.ts"; +import { KiloDriver } from "../Drivers/KiloDriver.ts"; import { OpenCodeDriver } from "../Drivers/OpenCodeDriver.ts"; +import { KiloRuntimeLive } from "../kiloRuntime.ts"; import { OpenCodeRuntimeLive } from "../opencodeRuntime.ts"; import { NoOpProviderEventLoggers, ProviderEventLoggers } from "./ProviderEventLoggers.ts"; +import type { BuiltInDriversEnv } from "../builtInDrivers.ts"; +import type { AnyProviderDriver } from "../ProviderDriver.ts"; import { makeProviderInstanceRegistry } from "./ProviderInstanceRegistryLive.ts"; const TestHttpClientLive = Layer.succeed( @@ -99,6 +104,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 @@ -231,18 +243,20 @@ describe("ProviderInstanceRegistryLive — multi-instance codex slice", () => { describe("ProviderInstanceRegistryLive — all drivers slice", () => { // All drivers need `NodeServices` (ChildProcessSpawner + FileSystem + - // Path). `OpenCodeDriver.create` additionally yields `OpenCodeRuntime` - // at construction time, so we wire `OpenCodeRuntimeLive` into the stack. - // `OpenCodeRuntimeLive` bundles its own `NetService.layer` via - // `Layer.provide`, so the only external requirement it still exposes is - // `ChildProcessSpawner` — resolved here by piping it through + // Path). `OpenCodeDriver.create` / `KiloDriver.create` additionally yield + // their runtime services at construction time, so we wire those Live + // layers into the stack. Each runtime bundles its own `NetService.layer` + // via `Layer.provide`, so the only external requirement still exposed is + // `ChildProcessSpawner` — resolved here by piping through // `provideMerge(NodeServices.layer)`. // // The nested `provideMerge`s read bottom-up: `NodeServices.layer` - // 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)); + // provides runtime 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 = Layer.mergeAll(OpenCodeRuntimeLive, KiloRuntimeLive).pipe( + Layer.provideMerge(NodeServices.layer), + ); const testLayer = ServerConfig.layerTest(process.cwd(), { prefix: "provider-instance-registry-all-drivers-test", }).pipe( @@ -259,12 +273,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]: { @@ -300,10 +316,23 @@ 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], + drivers: [ + CodexDriver, + ClaudeDriver, + CursorDriver, + GrokDriver, + OpenCodeDriver, + KiloDriver, + ] as ReadonlyArray>, configMap, }); @@ -313,9 +342,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 — @@ -326,16 +355,19 @@ 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"); expect(grok?.displayName).toBe("Grok"); expect(openCode?.displayName).toBe("OpenCode"); + expect(kilo?.displayName).toBe("Kilo"); // Every instance owns its own set of closures — no sharing across // drivers. `adapter` / `textGeneration` / `snapshot` are all @@ -348,6 +380,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { cursor!.adapter, grok!.adapter, openCode!.adapter, + kilo!.adapter, ]; expect(new Set(adapters).size).toBe(adapters.length); const textGenerations = [ @@ -356,6 +389,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { cursor!.textGeneration, grok!.textGeneration, openCode!.textGeneration, + kilo!.textGeneration, ]; expect(new Set(textGenerations).size).toBe(textGenerations.length); const snapshots = [ @@ -364,6 +398,7 @@ describe("ProviderInstanceRegistryLive — all drivers slice", () => { cursor!.snapshot, grok!.snapshot, openCode!.snapshot, + kilo!.snapshot, ]; expect(new Set(snapshots).size).toBe(snapshots.length); diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 159d853121c..2b17c2308bf 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -32,6 +32,7 @@ import { applyServerSettingsPatch } from "@t3tools/shared/serverSettings"; import { checkCodexProviderStatus, type CodexAppServerProviderSnapshot } from "./CodexProvider.ts"; import { checkClaudeProviderStatus } from "./ClaudeProvider.ts"; +import * as KiloRuntime from "../kiloRuntime.ts"; import * as OpenCodeRuntime from "../opencodeRuntime.ts"; import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import { ProviderInstanceRegistryHydrationLive } from "./ProviderInstanceRegistryHydration.ts"; @@ -1138,6 +1139,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 @@ -1205,6 +1207,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te claudeAgent: { enabled: false }, cursor: { enabled: false }, grok: { enabled: false }, + kilo: { enabled: false }, opencode: { enabled: false }, }, }), @@ -1230,6 +1233,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); @@ -1351,6 +1355,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( @@ -1412,6 +1417,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 === "cursor-agent") { @@ -1455,6 +1461,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te "codex", "cursor", "grok", + "kilo", "opencode", ]); assert.strictEqual(cursorProvider?.enabled, false); diff --git a/apps/server/src/provider/Services/KiloAdapter.ts b/apps/server/src/provider/Services/KiloAdapter.ts new file mode 100644 index 00000000000..fb3626ff86a --- /dev/null +++ b/apps/server/src/provider/Services/KiloAdapter.ts @@ -0,0 +1,13 @@ +/** + * KiloAdapter — shape type for the Kilo provider adapter. + * + * @module KiloAdapter + */ +import type { ProviderAdapterError } from "../Errors.ts"; +import type { ProviderAdapterShape } from "./ProviderAdapter.ts"; + +/** + * KiloAdapterShape — per-instance Kilo adapter contract. Carries + * a branded driver kind as the nominal discriminant. + */ +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 providerID/modelID with nested model paths", () => { + NodeAssert.deepEqual(parseKiloModelSlug("anthropic/claude-sonnet-4-5"), { + providerID: "anthropic", + modelID: "claude-sonnet-4-5", + }); + NodeAssert.deepEqual(parseKiloModelSlug("openai/gpt-5/mini"), { + providerID: "openai", + modelID: "gpt-5/mini", + }); + NodeAssert.equal(parseKiloModelSlug("no-slash"), null); + NodeAssert.equal(parseKiloModelSlug("/leading"), null); + NodeAssert.equal(parseKiloModelSlug("trailing/"), null); + NodeAssert.equal(parseKiloModelSlug(null), null); + }); + + it("maps T3 approval decisions to Kilo permission replies", () => { + NodeAssert.equal(toKiloPermissionReply("accept"), "once"); + NodeAssert.equal(toKiloPermissionReply("acceptForSession"), "always"); + NodeAssert.equal(toKiloPermissionReply("decline"), "reject"); + NodeAssert.equal(toKiloPermissionReply("cancel"), "reject"); + }); + + it("resolves plan interaction mode to agent plan, otherwise code", () => { + NodeAssert.equal(resolveKiloAgent({ interactionMode: "plan" }), "plan"); + NodeAssert.equal(resolveKiloAgent({ interactionMode: "default" }), "code"); + NodeAssert.equal(resolveKiloAgent({}), "code"); + NodeAssert.equal(resolveKiloAgent({ interactionMode: null }), "code"); + }); +}); diff --git a/apps/server/src/provider/kiloRuntime.ts b/apps/server/src/provider/kiloRuntime.ts new file mode 100644 index 00000000000..674ac0515f0 --- /dev/null +++ b/apps/server/src/provider/kiloRuntime.ts @@ -0,0 +1,574 @@ +import * as NodeURL from "node:url"; +import * as NodeCrypto from "node:crypto"; + +import type { ChatAttachment, ProviderApprovalDecision, RuntimeMode } from "@t3tools/contracts"; +import { + createKiloClient, + type Agent, + 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 Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; + +import { isWindowsCommandNotFound } from "../processRunner.ts"; +import { collectStreamAsString } from "./providerSnapshot.ts"; +import * as NetService from "@t3tools/shared/Net"; +import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; +import { resolveSpawnCommand } from "@t3tools/shared/shell"; + +const encodeUnknownJsonStringExit = Schema.encodeUnknownExit(Schema.UnknownFromJsonString); + +const KILO_SERVER_READY_PREFIX = "kilo server listening"; +const DEFAULT_KILO_SERVER_TIMEOUT_MS = 30_000; +const DEFAULT_HOSTNAME = "127.0.0.1"; + +export interface KiloServerProcess { + readonly url: string; + readonly password: string; + readonly exitCode: Effect.Effect; +} + +export interface KiloServerConnection { + readonly url: string; + readonly password: string; + readonly exitCode: Effect.Effect | null; + readonly external: boolean; +} + +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 = (u: unknown): u is KiloRuntimeError => P.isTagged(u, KILO_RUNTIME_ERROR_TAG); +} + +function encodeJsonStringForDiagnostics(input: unknown): string | undefined { + const result = encodeUnknownJsonStringExit(input); + return Exit.isSuccess(result) ? result.value : undefined; +} + +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(); + if (cause && typeof cause === "object") { + const anyCause = cause as Record; + const status = (anyCause.response as { status?: number } | undefined)?.status; + const body = anyCause.error ?? anyCause.data ?? anyCause.body; + const encodedBody = encodeJsonStringForDiagnostics(body ?? cause); + if (encodedBody) { + return `status=${status ?? "?"} body=${encodedBody}`; + } + } + return String(cause); +} + +export const runKiloSdk = ( + operation: string, + fn: () => Promise, +): Effect.Effect => + 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 KiloInventory { + readonly providerList: ProviderListResponse; + readonly agents: ReadonlyArray; +} + +export interface ParsedKiloModelSlug { + readonly providerID: string; + readonly modelID: string; +} + +export interface KiloRuntimeShape { + /** + * Spawns a local Kilo server process. Lifetime is bound to the caller's + * `Scope.Scope` — the child is killed when that scope closes. + */ + readonly startKiloServerProcess: (input: { + readonly binaryPath: string; + readonly environment?: NodeJS.ProcessEnv; + readonly port?: number; + readonly hostname?: string; + readonly timeoutMs?: number; + }) => Effect.Effect; + /** + * Always spawns a managed local server (v1 has no user-facing serverUrl). + * Lifetime is bound to the caller's scope. + */ + readonly connectToKiloServer: (input: { + readonly binaryPath: string; + readonly environment?: NodeJS.ProcessEnv; + readonly port?: number; + readonly hostname?: string; + readonly timeoutMs?: number; + }) => Effect.Effect; + readonly runKiloCommand: (input: { + readonly binaryPath: string; + readonly args: ReadonlyArray; + readonly environment?: NodeJS.ProcessEnv; + }) => Effect.Effect; + readonly createKiloSdkClient: (input: { + readonly baseUrl: string; + readonly directory: string; + readonly serverPassword: string; + }) => KiloClient; + readonly loadKiloInventory: ( + client: KiloClient, + ) => Effect.Effect; +} + +function parseServerUrlFromOutput(output: string): string | null { + for (const line of output.split("\n")) { + const trimmed = line.trim(); + if (!trimmed.toLowerCase().includes(KILO_SERVER_READY_PREFIX)) { + continue; + } + const match = trimmed.match(/on\s+(https?:\/\/[^\s]+)/i); + // Keep scanning: a log line may mention the ready phrase without a URL. + if (match?.[1]) { + return match[1]; + } + } + return null; +} + +export function parseKiloModelSlug(slug: string | null | undefined): ParsedKiloModelSlug | null { + 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 kiloQuestionId( + index: number, + question: QuestionRequest["questions"][number], +): string { + const header = question.header + .trim() + .toLowerCase() + .replace(/[^a-z0-9_-]+/g, "-"); + return header.length > 0 ? `question-${index}-${header}` : `question-${index}`; +} + +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 attachmentPath = input.resolveAttachmentPath(attachment); + if (!attachmentPath) { + continue; + } + + parts.push({ + type: "file", + mime: attachment.mimeType, + filename: attachment.name, + url: NodeURL.pathToFileURL(attachmentPath).href, + }); + } + + return parts; +} + +export function buildKiloPermissionRules(runtimeMode: RuntimeMode): PermissionRuleset { + if (runtimeMode === "full-access") { + return [{ permission: "*", pattern: "*", action: "allow" }]; + } + + return [ + { permission: "*", pattern: "*", action: "ask" }, + { permission: "bash", pattern: "*", action: "ask" }, + { permission: "edit", pattern: "*", action: "ask" }, + { permission: "webfetch", pattern: "*", action: "ask" }, + { permission: "websearch", pattern: "*", action: "ask" }, + { permission: "codesearch", pattern: "*", action: "ask" }, + { permission: "external_directory", pattern: "*", action: "ask" }, + { permission: "doom_loop", pattern: "*", action: "ask" }, + { permission: "question", pattern: "*", action: "allow" }, + ]; +} + +export function toKiloPermissionReply( + decision: ProviderApprovalDecision, +): "once" | "always" | "reject" { + switch (decision) { + case "accept": + return "once"; + case "acceptForSession": + return "always"; + case "decline": + case "cancel": + default: + return "reject"; + } +} + +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"); + } + if (typeof raw === "string") { + return raw.trim().length > 0 ? [raw] : []; + } + return []; + }); +} + +export function resolveKiloAgent(input: { + readonly interactionMode?: string | null | undefined; +}): "code" | "plan" { + return input.interactionMode === "plan" ? "plan" : "code"; +} + +function ensureRuntimeError( + operation: KiloRuntimeError["operation"], + detail: string, + cause: unknown, +): KiloRuntimeError { + return KiloRuntimeError.is(cause) ? cause : new KiloRuntimeError({ operation, detail, cause }); +} + +function generateServerPassword(): string { + return NodeCrypto.randomBytes(32).toString("hex"); +} + +const makeKiloRuntime = Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const netService = yield* NetService.NetService; + const hostPlatform = yield* HostProcessPlatform; + const resolveCommand = (command: string, args: ReadonlyArray, env?: NodeJS.ProcessEnv) => + resolveSpawnCommand(command, args, env ? { env } : {}); + + const runKiloCommand: KiloRuntimeShape["runKiloCommand"] = (input) => + Effect.gen(function* () { + const spawnCommand = yield* resolveCommand(input.binaryPath, input.args, input.environment); + const child = yield* spawner.spawn( + ChildProcess.make(spawnCommand.command, spawnCommand.args, { + shell: spawnCommand.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: "runKiloCommand", + detail: `spawn ${input.binaryPath} ENOENT`, + }); + } + return { + stdout, + stderr, + code: exitCode, + } satisfies KiloCommandResult; + }).pipe( + Effect.scoped, + Effect.mapError((cause) => + ensureRuntimeError( + "runKiloCommand", + `Failed to execute '${input.binaryPath} ${input.args.join(" ")}': ${kiloRuntimeErrorDetail(cause)}`, + cause, + ), + ), + ); + + const startKiloServerProcess: KiloRuntimeShape["startKiloServerProcess"] = (input) => + Effect.gen(function* () { + const runtimeScope = yield* Scope.Scope; + + const hostname = input.hostname ?? DEFAULT_HOSTNAME; + const port = + input.port ?? + (yield* netService.findAvailablePort(0).pipe( + Effect.mapError( + (cause) => + new KiloRuntimeError({ + operation: "startKiloServerProcess", + detail: `Failed to find available port: ${kiloRuntimeErrorDetail(cause)}`, + cause, + }), + ), + )); + const timeoutMs = input.timeoutMs ?? DEFAULT_KILO_SERVER_TIMEOUT_MS; + const password = generateServerPassword(); + const args = ["serve", `--hostname=${hostname}`, `--port=${port}`]; + const spawnCommand = yield* resolveCommand(input.binaryPath, args, input.environment); + + const child = yield* spawner + .spawn( + ChildProcess.make(spawnCommand.command, spawnCommand.args, { + detached: hostPlatform !== "win32", + shell: spawnCommand.shell, + env: { + ...input.environment, + KILO_SERVER_PASSWORD: password, + KILO_PARENT_PID: String(process.pid), + }, + extendEnv: input.environment === undefined, + }), + ) + .pipe( + Effect.provideService(Scope.Scope, runtimeScope), + Effect.mapError( + (cause) => + new KiloRuntimeError({ + operation: "startKiloServerProcess", + detail: `Failed to spawn Kilo server process: ${kiloRuntimeErrorDetail(cause)}`, + cause, + }), + ), + ); + + const killKiloProcessGroup = (signal: NodeJS.Signals) => + hostPlatform === "win32" + ? child.kill({ killSignal: signal, forceKillAfter: "1 second" }).pipe(Effect.asVoid) + : Effect.sync(() => { + try { + process.kill(-Number(child.pid), signal); + } catch { + // Best-effort process-group cleanup. + } + }); + const terminateChild = killKiloProcessGroup("SIGTERM").pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.andThen(killKiloProcessGroup("SIGKILL")), + Effect.ignore, + ); + yield* Scope.addFinalizer(runtimeScope, terminateChild); + + const stdoutRef = yield* Ref.make(""); + const stderrRef = yield* Ref.make(""); + // Capture output only until ready/failure diagnostics are resolved; after + // that keep draining pipes without retaining unbounded log buffers. + const captureOutputRef = yield* Ref.make(true); + const readyDeferred = yield* Deferred.make(); + + const setReadyFromStdoutChunk = (chunk: string) => + Effect.gen(function* () { + const capture = yield* Ref.get(captureOutputRef); + if (!capture) { + return; + } + const nextStdout = yield* Ref.updateAndGet(stdoutRef, (stdout) => `${stdout}${chunk}`); + const parsed = parseServerUrlFromOutput(nextStdout); + if (parsed) { + // Do not clear capture here — keep capturing until the success path + // after ready settles so concurrent stderr diagnostics are not lost. + yield* Deferred.succeed(readyDeferred, parsed).pipe(Effect.ignore); + } + }); + + const stdoutFiber = yield* child.stdout.pipe( + Stream.decodeText(), + Stream.runForEach(setReadyFromStdoutChunk), + Effect.ignore, + Effect.forkIn(runtimeScope), + ); + const stderrFiber = yield* child.stderr.pipe( + Stream.decodeText(), + Stream.runForEach((chunk) => + Effect.gen(function* () { + if (!(yield* Ref.get(captureOutputRef))) { + return; + } + yield* Ref.update(stderrRef, (stderr) => `${stderr}${chunk}`); + }), + ), + Effect.ignore, + Effect.forkIn(runtimeScope), + ); + + const exitFiber = yield* child.exitCode.pipe( + Effect.flatMap((code) => + Effect.gen(function* () { + const stdout = yield* Ref.get(stdoutRef); + const stderr = yield* Ref.get(stderrRef); + const exitCode = Number(code); + yield* Deferred.fail( + readyDeferred, + new KiloRuntimeError({ + operation: "startKiloServerProcess", + detail: [ + `Kilo server exited before startup completed (code: ${String(exitCode)}).`, + stdout.trim() ? `stdout:\n${stdout.trim()}` : null, + stderr.trim() ? `stderr:\n${stderr.trim()}` : null, + ] + .filter(Boolean) + .join("\n\n"), + cause: { exitCode, stdout, stderr }, + }), + ).pipe(Effect.ignore); + }), + ), + Effect.ignore, + Effect.forkIn(runtimeScope), + ); + + const readyExit = yield* Effect.exit( + Deferred.await(readyDeferred).pipe(Effect.timeoutOption(timeoutMs)), + ); + + if (Exit.isFailure(readyExit)) { + // Stop draining + kill the child immediately so a failed start does not + // leave a half-started server holding the port until the outer scope ends. + yield* Fiber.interrupt(stdoutFiber).pipe(Effect.ignore); + yield* Fiber.interrupt(stderrFiber).pipe(Effect.ignore); + yield* Fiber.interrupt(exitFiber).pipe(Effect.ignore); + yield* terminateChild; + const squashed = Cause.squash(readyExit.cause); + return yield* ensureRuntimeError( + "startKiloServerProcess", + `Failed while waiting for Kilo server startup: ${kiloRuntimeErrorDetail(squashed)}`, + squashed, + ); + } + + const readyOption = readyExit.value; + if (Option.isNone(readyOption)) { + yield* Fiber.interrupt(stdoutFiber).pipe(Effect.ignore); + yield* Fiber.interrupt(stderrFiber).pipe(Effect.ignore); + yield* Fiber.interrupt(exitFiber).pipe(Effect.ignore); + yield* terminateChild; + return yield* new KiloRuntimeError({ + operation: "startKiloServerProcess", + detail: `Timed out waiting for Kilo server start after ${timeoutMs}ms.`, + }); + } + + // Keep stdout/stderr drain fibers alive for the server lifetime so the + // OS pipe buffers cannot fill and block the child process. Stop + // retaining captured output once ready so long-lived sessions do not + // accumulate unbounded log buffers. + yield* Ref.set(captureOutputRef, false); + return { + url: readyOption.value, + password, + exitCode: child.exitCode.pipe( + Effect.map(Number), + Effect.orElseSucceed(() => 0), + ), + } satisfies KiloServerProcess; + }); + + const connectToKiloServer: KiloRuntimeShape["connectToKiloServer"] = (input) => + startKiloServerProcess({ + binaryPath: input.binaryPath, + ...(input.environment !== undefined ? { environment: input.environment } : {}), + ...(input.port !== undefined ? { port: input.port } : {}), + ...(input.hostname !== undefined ? { hostname: input.hostname } : {}), + ...(input.timeoutMs !== undefined ? { timeoutMs: input.timeoutMs } : {}), + }).pipe( + Effect.map((server) => ({ + url: server.url, + password: server.password, + exitCode: server.exitCode, + external: false, + })), + ); + + const createKiloSdkClient: KiloRuntimeShape["createKiloSdkClient"] = (input) => + createKiloClient({ + baseUrl: input.baseUrl, + directory: input.directory, + headers: { + Authorization: `Basic ${Buffer.from(`kilo:${input.serverPassword}`, "utf8").toString("base64")}`, + }, + throwOnError: true, + }); + + const loadProviders = (client: KiloClient) => + runKiloSdk("provider.list", () => client.provider.list()).pipe( + Effect.filterMapOrFail( + (list) => + list.data + ? Result.succeed(list.data) + : Result.fail( + new KiloRuntimeError({ + operation: "provider.list", + detail: "Kilo provider list was empty.", + }), + ), + (result) => result, + ), + ); + + const loadAgents = (client: KiloClient) => + runKiloSdk("app.agents", () => client.app.agents()).pipe( + Effect.map((result) => result.data ?? []), + ); + + const loadKiloInventory: KiloRuntimeShape["loadKiloInventory"] = (client) => + Effect.all([loadProviders(client), loadAgents(client)], { concurrency: "unbounded" }).pipe( + Effect.map(([providerList, agents]) => ({ providerList, agents })), + ); + + return { + startKiloServerProcess, + connectToKiloServer, + runKiloCommand, + createKiloSdkClient, + loadKiloInventory, + } 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/server.ts b/apps/server/src/server.ts index 0e6db87b109..b03d1f36731 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -24,6 +24,7 @@ import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRe import * as ProviderEventLoggers from "./provider/Layers/ProviderEventLoggers.ts"; import { ProviderServiceLive } from "./provider/Layers/ProviderService.ts"; import { ProviderSessionReaperLive } from "./provider/Layers/ProviderSessionReaper.ts"; +import * as KiloRuntime from "./provider/kiloRuntime.ts"; import * as OpenCodeRuntime from "./provider/opencodeRuntime.ts"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as CheckpointStore from "./checkpointing/CheckpointStore.ts"; @@ -312,7 +313,13 @@ 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), + // SDK-backed drivers yield their runtimes at create(); expose both at the + // runtime level so the instance registry can construct them without each + // driver layer re-providing the same Live. Merged into one provideMerge + // slot so the pipe arity stays within Effect's Layer helper limits. + Layer.provideMerge( + Layer.mergeAll(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.ts b/apps/server/src/textGeneration/KiloTextGeneration.ts new file mode 100644 index 00000000000..6a44362129f --- /dev/null +++ b/apps/server/src/textGeneration/KiloTextGeneration.ts @@ -0,0 +1,44 @@ +import * as Effect from "effect/Effect"; + +import { type KiloSettings, TextGenerationError } from "@t3tools/contracts"; +import type * as TextGeneration from "./TextGeneration.ts"; + +/** + * Kilo text generation is not implemented in v1. The driver still exposes a + * typed service so instance construction matches other providers; all ops + * fail with a clear unsupported message. + */ +export const makeKiloTextGeneration = ( + _kiloSettings: KiloSettings, + _environment?: NodeJS.ProcessEnv, +): Effect.Effect => + Effect.succeed({ + generateCommitMessage: () => + Effect.fail( + new TextGenerationError({ + operation: "generateCommitMessage", + detail: "Kilo does not support git/text generation in T3 Code yet.", + }), + ), + generatePrContent: () => + Effect.fail( + new TextGenerationError({ + operation: "generatePrContent", + detail: "Kilo does not support git/text generation in T3 Code yet.", + }), + ), + generateBranchName: () => + Effect.fail( + new TextGenerationError({ + operation: "generateBranchName", + detail: "Kilo does not support git/text generation in T3 Code yet.", + }), + ), + generateThreadTitle: () => + Effect.fail( + new TextGenerationError({ + operation: "generateThreadTitle", + detail: "Kilo does not support git/text generation in T3 Code yet.", + }), + ), + }); diff --git a/apps/server/src/textGeneration/TextGeneration.ts b/apps/server/src/textGeneration/TextGeneration.ts index e62a79afe78..100e155eccf 100644 --- a/apps/server/src/textGeneration/TextGeneration.ts +++ b/apps/server/src/textGeneration/TextGeneration.ts @@ -7,7 +7,13 @@ import { TextGenerationError } from "@t3tools/contracts"; import * as ProviderInstanceRegistry from "../provider/Services/ProviderInstanceRegistry.ts"; import type { ProviderInstance } from "../provider/ProviderDriver.ts"; -export type TextGenerationProvider = "codex" | "claudeAgent" | "cursor" | "grok" | "opencode"; +export type TextGenerationProvider = + | "codex" + | "claudeAgent" + | "cursor" + | "grok" + | "opencode" + | "kilo"; export interface CommitMessageGenerationInput { cwd: string; diff --git a/apps/web/src/components/Icons.tsx b/apps/web/src/components/Icons.tsx index 8ea38c51958..62c95b7e57d 100644 --- a/apps/web/src/components/Icons.tsx +++ b/apps/web/src/components/Icons.tsx @@ -663,6 +663,15 @@ 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..beccd4c250e 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, + type Icon, + KiloIcon, + OpenAI, + OpenCodeIcon, +} from "../Icons"; type ProviderSettingsSchema = { readonly fields: Readonly>; @@ -67,6 +76,12 @@ export const PROVIDER_CLIENT_DEFINITIONS: readonly ProviderClientDefinition[] = icon: OpenCodeIcon, settingsSchema: OpenCodeSettings, }, + { + value: ProviderDriverKind.make("kilo"), + label: "Kilo", + icon: KiloIcon, + settingsSchema: KiloSettings, + }, ]; export const PROVIDER_CLIENT_DEFINITION_BY_VALUE: Partial< diff --git a/apps/web/src/lib/contextWindow.ts b/apps/web/src/lib/contextWindow.ts index 80f7d31cf2f..8a3ab4ffd1f 100644 --- a/apps/web/src/lib/contextWindow.ts +++ b/apps/web/src/lib/contextWindow.ts @@ -38,6 +38,8 @@ export function formatProviderDisplayName(provider: string | null | undefined): return "Cursor"; case "opencode": return "OpenCode"; + case "kilo": + return "Kilo"; default: { // Title-case unknown driver kinds so they read reasonably. const trimmed = provider.replace(/Agent$/i, "").trim(); 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..f90152761d8 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.test.ts b/packages/contracts/src/settings.test.ts index 0f618729c43..f311cc3ada4 100644 --- a/packages/contracts/src/settings.test.ts +++ b/packages/contracts/src/settings.test.ts @@ -42,6 +42,8 @@ describe("ServerSettings.providerInstances (slice-2 invariant)", () => { // Legacy `providers` struct is still hydrated with its per-driver defaults // so existing call sites keep working through the migration. expect(decoded.providers.codex.enabled).toBe(true); + expect(decoded.providers.kilo.enabled).toBe(true); + expect(decoded.providers.kilo.binaryPath).toBe("kilo"); }); it("decodes a multi-instance map mixing first-party and fork drivers", () => { diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index fe593f49199..1a0a3caf1eb 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -362,6 +362,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(""))), @@ -406,6 +433,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 @@ -509,6 +537,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), @@ -531,6 +565,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 c58ccffc420..2bcf5c9b787 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -456,6 +456,9 @@ importers: '@ff-labs/fff-node': specifier: 0.9.4 version: 0.9.4(patch_hash=2b16019ce7ab61aec6478dd02f79ef468cc1d5c51e9d00764f7d2ab8167210c8) + '@kilocode/sdk': + specifier: ^7.4.11 + version: 7.4.11 '@opencode-ai/sdk': specifier: ^1.3.15 version: 1.15.13 @@ -2935,6 +2938,9 @@ packages: resolution: {integrity: sha512-hloP58zRVCRSpgDxmqCWJNlizAlUgJFqG2ypq79DCvyv9tHjRYMDOcPFjzfl/A1/YxDvRCZz8wvZvmapQnKwFQ==} engines: {node: '>=12'} + '@kilocode/sdk@7.4.11': + resolution: {integrity: sha512-SOPa4ozNzL4H51hByKbmCFhbFY41QjmZw1CGCd36/YueJ9N0KzxikgqaJjnSKD5BO57LLwqF6Yu/YH7D/FnLPg==} + '@legendapp/list@3.2.0': resolution: {integrity: sha512-bN+g/oQYjFz+UAyuBN4cmYJAwdJS1TdNcZZOVlh3+VwCQUWrsg0PH46Mvm76gdZSCYMfoFanPY4dKnILcYEzeg==} peerDependencies: @@ -12878,6 +12884,10 @@ snapshots: dependencies: jsbi: 4.3.2 + '@kilocode/sdk@7.4.11': + 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 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 22757702fc8..f3fe97fe1d8 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -58,6 +58,7 @@ minimumReleaseAgeExclude: - "@clerk/expo@3.7.2" - "@clerk/react@6.12.1" - "@clerk/shared@4.25.1" + - "@kilocode/sdk@7.4.11" overrides: "@clerk/backend": "catalog:"