Merge pull request #11410 from Kilo-Org/jungle-cobbler

fix(cli): share session runtime with HTTP listeners

.changeset/share-server-session-runtime.md

@@ -0,0 +1,6 @@
+---
+"@kilocode/cli": patch
+"kilo-code": patch
+---
+
+Keep server controls and events connected to active sessions and subagents.

packages/opencode/src/kilocode/server/httpapi/server.ts

@@ -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"
@@ -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),
+  ])
+}

packages/opencode/src/kilocode/server/listener.ts

@@ -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)
+}

packages/opencode/src/server/routes/instance/httpapi/handlers/v2.ts

@@ -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
 )

packages/opencode/src/server/routes/instance/httpapi/server.ts

@@ -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"
@@ -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(() =>

packages/opencode/src/server/server.ts

@@ -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
@@ -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,
@@ -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(

packages/opencode/test/kilocode/server/listener-runtime.test.ts

@@ -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)

script/check-opencode-promise-facades.ts

@@ -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" },
 }