Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .changeset/share-server-session-runtime.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
---
"@kilocode/cli": patch
"kilo-code": patch
---

Keep server controls and events connected to active sessions and subagents.
26 changes: 26 additions & 0 deletions packages/opencode/src/kilocode/server/httpapi/server.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,10 @@
import { Layer } from "effect"
import { FetchHttpClient, HttpMiddleware, HttpRouter, HttpServer } from "effect/unstable/http"
import { CorsConfig, isAllowedCorsOrigin, type CorsOptions } from "@/server/cors"
import { compressionLayer } from "@/server/routes/instance/httpapi/middleware/compression"
import { corsVaryFix } from "@/server/routes/instance/httpapi/middleware/cors-vary"
import { errorLayer } from "@/server/routes/instance/httpapi/middleware/error"
import { fenceLayer } from "@/server/routes/instance/httpapi/middleware/fence"

import { agentBuilderHandlers } from "./handlers/agent-builder"
import { backgroundProcessHandlers } from "./handlers/background-process"
Expand Down Expand Up @@ -29,3 +35,23 @@ export const provide = Layer.provide([
suggestionHandlers,
telemetryHandlers,
])

export function provideListener(opts?: CorsOptions) {
const cors = HttpRouter.middleware(
HttpMiddleware.cors({
allowedOrigins: (origin) => isAllowedCorsOrigin(origin, opts),
maxAge: 86_400,
}),
{ global: true },
)
return Layer.provide([
errorLayer,
compressionLayer,
corsVaryFix,
fenceLayer,
cors,
FetchHttpClient.layer,
HttpServer.layerServices,
Layer.succeed(CorsConfig)(opts),
])
}
8 changes: 8 additions & 0 deletions packages/opencode/src/kilocode/server/listener.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
import { AppLayer } from "@/effect/app-runtime"
import { memoMap } from "@opencode-ai/core/effect/memo-map"
import { Layer, Scope } from "effect"

export function build<A, E, R>(layer: Layer.Layer<A, E, R>, scope: Scope.Scope) {
// Keep listener transport state fresh while AppLayer reuses the process-wide services.
return Layer.buildWithMemoMap(Layer.fresh(layer).pipe(Layer.provide(AppLayer)), memoMap, scope)
}
Original file line number Diff line number Diff line change
Expand Up @@ -8,5 +8,5 @@ import { sessionHandlers } from "./v2/session"

export const v2Handlers = Layer.mergeAll(sessionHandlers, messageHandlers, modelHandlers, providerHandlers).pipe(
Layer.provide(v2LocationLayer),
Layer.provide(SessionV2.defaultLayer),
Layer.provide(SessionV2.layer), // kilocode_change - use the application EventV2Bridge
)
15 changes: 14 additions & 1 deletion packages/opencode/src/server/routes/instance/httpapi/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,12 @@ import { syncHandlers } from "./handlers/sync"
import { tuiHandlers } from "./handlers/tui"
import { v2Handlers } from "./handlers/v2"
import { workspaceHandlers } from "./handlers/workspace"
import { provide as provideKiloHttpApiHandlers } from "@/kilocode/server/httpapi/server" // kilocode_change
// kilocode_change start
import {
provide as provideKiloHttpApiHandlers,
provideListener as provideKiloListenerRoutes,
} from "@/kilocode/server/httpapi/server"
// kilocode_change end
import { instanceContextLayer, instanceRouterMiddleware } from "./middleware/instance-context"
import { workspaceRouterMiddleware, workspaceRoutingLayer } from "./middleware/workspace-routing"
import { disposeMiddleware } from "./lifecycle"
Expand Down Expand Up @@ -248,6 +253,14 @@ export function createRoutes(
)
}

// kilocode_change start - keep listener routes local while application services come from AppRuntime
export function createListenerRoutes(corsOptions?: CorsOptions) {
return Layer.mergeAll(rootApiRoutes, eventApiRoutes, instanceRoutes, docRoute, uiRoute).pipe(
provideKiloListenerRoutes(corsOptions),
)
}
// kilocode_change end

export const routes = createRoutes()

export const webHandler = lazy(() =>
Expand Down
5 changes: 3 additions & 2 deletions packages/opencode/src/server/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import { WebSocketTracker } from "./routes/instance/httpapi/websocket-tracker"
import { PublicApi } from "./routes/instance/httpapi/public"
import type { CorsOptions } from "./cors"
import { lazy } from "@/util/lazy"
import * as KiloListener from "@/kilocode/server/listener" // kilocode_change

// @ts-ignore This global is needed to prevent ai-sdk from logging warnings to stdout https://github.com/vercel/ai/blob/2dc67e0ef538307f21368db32d5a12345d98831b/packages/ai/src/logger/log-warnings.ts#L85
globalThis.AI_SDK_LOG_WARNINGS = false
Expand Down Expand Up @@ -101,7 +102,7 @@ const listenEffect: (opts: ListenOptions) => Effect.Effect<EffectListener, unkno
)

function listenerLayer(opts: ListenOptions, port: number) {
return HttpRouter.serve(HttpApiApp.createRoutes(opts), {
return HttpRouter.serve(HttpApiApp.createListenerRoutes(opts), { // kilocode_change
middleware: disposeMiddleware,
disableLogger: true,
disableListenLog: true,
Expand All @@ -126,7 +127,7 @@ function startWithPortFallback(opts: ListenOptions) {

function startListener(opts: ListenOptions, port: number) {
const scope = Scope.makeUnsafe()
return Layer.buildWithMemoMap(listenerLayer(opts, port), Layer.makeMemoMapUnsafe(), scope).pipe(
return KiloListener.build(listenerLayer(opts, port), scope).pipe( // kilocode_change
Effect.provide(HttpApiApp.context),
Effect.onError(() => Scope.close(scope, Exit.void).pipe(Effect.ignore)),
Effect.map(
Expand Down
67 changes: 67 additions & 0 deletions packages/opencode/test/kilocode/server/listener-runtime.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
import { afterEach, expect, test } from "bun:test"
import { Flag } from "@opencode-ai/core/flag/flag"
import * as Log from "@opencode-ai/core/util/log"
import { Effect } from "effect"
import { AppRuntime } from "../../../src/effect/app-runtime"
import { InstanceRef } from "../../../src/effect/instance-ref"
import { Server } from "../../../src/server/server"
import { SessionPaths } from "../../../src/server/routes/instance/httpapi/groups/session"
import { SessionRunState } from "../../../src/session/run-state"
import { SessionID } from "../../../src/session/schema"
import { withTimeout } from "../../../src/util/timeout"
import { resetDatabase } from "../../fixture/db"
import { disposeAllInstances, reloadTestInstance, tmpdir } from "../../fixture/fixture"

void Log.init({ print: false })

const previous = {
flag: Flag.KILO_SERVER_PASSWORD,
env: process.env.KILO_SERVER_PASSWORD,
}

afterEach(async () => {
Flag.KILO_SERVER_PASSWORD = previous.flag
if (previous.env === undefined) delete process.env.KILO_SERVER_PASSWORD
else process.env.KILO_SERVER_PASSWORD = previous.env
await disposeAllInstances()
await resetDatabase()
})

test("listener aborts shared session runners", async () => {
Flag.KILO_SERVER_PASSWORD = undefined
delete process.env.KILO_SERVER_PASSWORD
await using tmp = await tmpdir({ git: true, config: { formatter: false, lsp: false } })
const ctx = await reloadTestInstance({ directory: tmp.path })
const sessionID = SessionID.descending()
const started = Promise.withResolvers<void>()
const stopped = Promise.withResolvers<void>()
const running = AppRuntime.runPromise(
SessionRunState.Service.use((state) =>
state.ensureRunning(
sessionID,
Effect.interrupt,
Effect.sync(started.resolve).pipe(Effect.andThen(Effect.never), Effect.ensuring(Effect.sync(stopped.resolve))),
),
).pipe(Effect.provideService(InstanceRef, ctx)),
).catch(() => undefined)

try {
await withTimeout(started.promise, 5_000, "timed out waiting for shared session")
const listener = await Server.listen({ hostname: "127.0.0.1", port: 0 })
try {
const response = await fetch(new URL(SessionPaths.abort.replace(":sessionID", sessionID), listener.url), {
method: "POST",
headers: { "x-kilo-directory": tmp.path },
})
expect(response.status).toBe(200)
await withTimeout(stopped.promise, 5_000, "listener did not interrupt the shared session")
} finally {
await withTimeout(listener.stop(true), 10_000, "timed out cleaning up shared-runtime listener")
}
} finally {
await AppRuntime.runPromise(
SessionRunState.Service.use((state) => state.cancel(sessionID)).pipe(Effect.provideService(InstanceRef, ctx)),
).catch(() => undefined)
await running
}
}, 20_000)
1 change: 1 addition & 0 deletions script/check-opencode-promise-facades.ts
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ const testAllow: Record<string, { count: number; reason: string }> = {
"provider/provider.test.ts": { count: 3, reason: "existing runtime integration test" },
"server/experimental-session-list.test.ts": { count: 2, reason: "Kilo session list integration test" },
"server/httpapi-event.test.ts": { count: 6, reason: "event stream integration test" },
"kilocode/server/listener-runtime.test.ts": { count: 3, reason: "listener and AppRuntime integration test" },
"session/llm.test.ts": { count: 2, reason: "existing runtime integration test" },
"tool/recall.test.ts": { count: 11, reason: "existing runtime integration test" },
}
Expand Down
Loading