diff --git a/.changeset/scoped-interactive-terminal-state.md b/.changeset/scoped-interactive-terminal-state.md new file mode 100644 index 00000000000..c41d6db24ff --- /dev/null +++ b/.changeset/scoped-interactive-terminal-state.md @@ -0,0 +1,5 @@ +--- +"kilo-code": patch +--- + +Ensure interactive terminal state is isolated per directory and cleaned up when its scope closes. diff --git a/packages/opencode/src/kilocode/interactive-terminal/index.ts b/packages/opencode/src/kilocode/interactive-terminal/index.ts index ce38fefab07..2ece2bd90a3 100644 --- a/packages/opencode/src/kilocode/interactive-terminal/index.ts +++ b/packages/opencode/src/kilocode/interactive-terminal/index.ts @@ -1,18 +1,21 @@ import { Bus } from "@/bus" import { BusEvent } from "@/bus/bus-event" -import { InstanceState } from "@/effect/instance-state" +import { InstanceRef } from "@/effect/instance-ref" +import { registerDisposer } from "@/effect/instance-registry" import { makeRuntime } from "@/effect/run-service" +import { KiloShutdown } from "@/kilocode/cli/shutdown" import { appendTerminalOutput } from "@/kilocode/interactive-terminal/output" import { model as modelEnv } from "@/kilocode/process/env" import { Identifier } from "@/id/id" -import { Instance, type InstanceContext } from "@/kilocode/instance" +import { capture, Instance, type InstanceContext } from "@/kilocode/instance" import { SessionID } from "@/session/schema" import { Shell } from "@opencode-ai/core/shell" import { NonNegativeInt, PositiveInt, optionalOmitUndefined, withStatics } from "@opencode-ai/core/schema" import { zod, ZodOverride } from "@opencode-ai/core/effect-zod" import * as Log from "@opencode-ai/core/util/log" import type { Disp, Proc } from "@opencode-ai/core/pty/driver" -import { Context, Effect, Layer, Schema, Types } from "effect" +import { Context, Effect, Layer, LayerMap, Schema, Types } from "effect" +import * as Scope from "effect/Scope" import path from "path" import stripAnsi from "strip-ansi" import z from "zod" @@ -135,9 +138,12 @@ export namespace InteractiveTerminal { terminals: Map } - class StateService extends Context.Service Effect.Effect }>()( - "@kilocode/InteractiveTerminal.State", - ) {} + class StateService extends Context.Service()("@kilocode/InteractiveTerminal.State") {} + + class RuntimeService extends Context.Service< + RuntimeService, + { readonly get: (ctx: InstanceContext) => Effect.Effect } + >()("@kilocode/InteractiveTerminal.Runtime") {} function clone(info: Info): Info { return { ...info, time: { ...info.time } } @@ -358,32 +364,52 @@ export namespace InteractiveTerminal { } const stateLayer = Layer.effect( - StateService, + RuntimeService, Effect.gen(function* () { - const ref = yield* InstanceState.make( - Effect.fn("InteractiveTerminal.state")(function* (ctx) { - const state: State = { ctx, dir: ctx.directory, terminals: new Map() } - yield* Effect.addFinalizer(() => - Effect.promise(async () => { - await Promise.all( - Array.from(state.terminals.values()).map((active) => - finish(state, active, { closedBy: "abort", kill: true, silent: true }), - ), + const states = yield* LayerMap.make( + (_dir: string) => + Layer.effect( + StateService, + Effect.gen(function* () { + const ctx = yield* InstanceRef + if (!ctx) return yield* Effect.die(new Error("Instance context not available")) + const state: State = { ctx, dir: ctx.directory, terminals: new Map() } + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + await Promise.all( + Array.from(state.terminals.values()).map((active) => + finish(state, active, { closedBy: "abort", kill: true, silent: true }), + ), + ) + state.terminals.clear() + }), ) - state.terminals.clear() + return state }), - ) - return state - }), + ), + { idleTimeToLive: Number.POSITIVE_INFINITY }, ) - return StateService.of({ get: () => InstanceState.get(ref) }) + + const off = registerDisposer((dir) => Effect.runPromise(states.invalidate(dir))) + yield* Effect.addFinalizer(() => Effect.sync(off)) + + return RuntimeService.of({ + get: (ctx) => + states.contextEffect(ctx.directory).pipe( + Effect.provideService(InstanceRef, ctx), + Effect.map((value) => Context.get(value, StateService)), + ), + }) }), ) - const runtime = makeRuntime(StateService, stateLayer) + const runtime = makeRuntime(RuntimeService, stateLayer) + KiloShutdown.register(() => runtime.dispose()) function state() { - return runtime.runPromise((service) => service.get()) + const ctx = capture() + if (!ctx) return Promise.reject(new Error("Instance context not available")) + return runtime.runPromise((service) => service.get(ctx).pipe(Effect.scoped)) } export async function run(input: RunInput) { diff --git a/packages/opencode/test/kilocode/interactive-terminal.test.ts b/packages/opencode/test/kilocode/interactive-terminal.test.ts index 03c62fd71f3..b33e3f2157e 100644 --- a/packages/opencode/test/kilocode/interactive-terminal.test.ts +++ b/packages/opencode/test/kilocode/interactive-terminal.test.ts @@ -3,6 +3,7 @@ import { Bus } from "@/bus" import { Agent } from "@/agent/agent" import { Config } from "@/config/config" import { RuntimeFlags } from "@/effect/runtime-flags" +import { InstanceStore } from "@/project/instance-store" import { VtScreen } from "@/kilocode/cli/cmd/tui/vt/vt-screen" import { InteractiveTerminal } from "@/kilocode/interactive-terminal" import { Instance, capture, type InstanceContext } from "@/kilocode/instance" @@ -19,7 +20,7 @@ import { describe, expect } from "bun:test" import { Cause, Effect, Exit, Layer } from "effect" import path from "path" import { TestInstance, tmpdirScoped } from "../fixture/fixture" -import { it, testEffect } from "../lib/effect" +import { it, pollWithTimeout, testEffect } from "../lib/effect" const toolLayer = Layer.mergeAll( AppNodeBuilder.build(CrossSpawnSpawner.node), @@ -47,20 +48,22 @@ async function script(dir: string, name: string, source: string) { return `${bin} ${arg}` } -function started(sessionID: SessionID) { +function started(sessionID: SessionID, cwd?: string, ctx?: InstanceContext) { const state: { off?: () => void; timer?: ReturnType } = {} const promise = new Promise((resolve, reject) => { state.timer = setTimeout(() => { state.off?.() reject(new Error("timed out waiting for interactive terminal")) }, 5_000) - state.off = Bus.subscribe(InteractiveTerminal.Event.Updated, (event) => { - const info = event.properties.info - if (info.sessionID !== sessionID || info.status !== "running") return - state.off?.() - if (state.timer) clearTimeout(state.timer) - resolve(info) - }) + state.off = Instance.restore(ctx ?? capture()!, () => + Bus.subscribe(InteractiveTerminal.Event.Updated, (event) => { + const info = event.properties.info + if (info.sessionID !== sessionID || info.status !== "running" || (cwd && info.cwd !== cwd)) return + state.off?.() + if (state.timer) clearTimeout(state.timer) + resolve(info) + }), + ) }) return { promise, @@ -163,8 +166,7 @@ describe("InteractiveTerminal", () => { ).toMatchObject({ message: err.message }) const ext = requests.find((item) => item.permission === "external_directory") const bash = requests.find((item) => item.permission === "bash") - const want = - process.platform === "win32" ? FSUtil.normalizePathPattern(path.join(tmp, "*")) : path.join(tmp, "*") + const want = process.platform === "win32" ? FSUtil.normalizePathPattern(path.join(tmp, "*")) : path.join(tmp, "*") expect(ext?.patterns).toContain(want) expect(bash?.patterns).toContain(`cat ${quote(file)}`) }), @@ -321,4 +323,88 @@ process.stdin.once("data", () => process.exit(0)) } }), ) + + it.instance("coalesces concurrent first state initialization", () => + Effect.gen(function* () { + const first = SessionID.descending() + const second = SessionID.descending() + const command = `setInterval(() => {}, 1_000)` + const pending = [ + run({ sessionID: first, command, cwd: capture()!.directory }), + run({ sessionID: second, command, cwd: capture()!.directory }), + ] + const list = yield* pollWithTimeout( + Effect.promise(() => InteractiveTerminal.list()).pipe( + Effect.map((items) => (items.length === 2 ? items : undefined)), + ), + "timed out waiting for concurrent terminals", + ) + expect(list.map((item) => item.sessionID).sort()).toEqual([first, second].sort()) + yield* Effect.promise(() => Promise.all(list.map((item) => InteractiveTerminal.close(item.id)))) + expect((yield* Effect.promise(() => Promise.all(pending))).map((item) => item.closedBy)).toEqual(["user", "user"]) + }), + ) + + toolIt.instance("isolates terminals by directory even for the same session", () => + Effect.gen(function* () { + const store = yield* InstanceStore.Service + const one = yield* tmpdirScoped() + const two = yield* tmpdirScoped() + const first = yield* store.load({ directory: one }) + const second = yield* store.load({ directory: two }) + const sessionID = SessionID.descending() + const command = `setInterval(() => {}, 1_000)` + const pendingOne = Instance.restore(first, () => run({ sessionID, command, cwd: one })) + const pendingTwo = Instance.restore(second, () => run({ sessionID, command, cwd: two })) + try { + const [infoOne, infoTwo] = yield* Effect.all( + [first, second].map((ctx) => + pollWithTimeout( + Effect.promise(() => Instance.restore(ctx, () => InteractiveTerminal.list({ sessionID }))).pipe( + Effect.map((items) => items[0]), + ), + "timed out waiting for isolated terminal", + ), + ), + { concurrency: "unbounded" }, + ) + if (!infoOne || !infoTwo) throw new Error("isolated terminals did not start") + expect(infoOne.id).not.toBe(infoTwo.id) + yield* Effect.promise(() => + Promise.all([ + Instance.restore(first, () => InteractiveTerminal.close(infoOne.id)), + Instance.restore(second, () => InteractiveTerminal.close(infoTwo.id)), + ]), + ) + expect( + (yield* Effect.promise(() => Promise.all([pendingOne, pendingTwo]))).map((item) => item.closedBy), + ).toEqual(["user", "user"]) + } finally { + yield* store.dispose(first) + yield* store.dispose(second) + } + }), + ) + + toolIt.instance("closes all terminals when the instance scope is disposed", () => + Effect.gen(function* () { + const store = yield* InstanceStore.Service + const ctx = capture()! + const sessionID = SessionID.descending() + const command = `setInterval(() => {}, 1_000)` + const ready = started(sessionID) + const pending = run({ sessionID, command, cwd: ctx.directory }) + try { + yield* Effect.promise(() => ready.promise) + yield* store.dispose(ctx) + expect((yield* Effect.promise(() => pending)).closedBy).toBe("abort") + expect( + yield* Effect.promise(() => Instance.restore(ctx, () => InteractiveTerminal.list({ sessionID }))), + ).toEqual([]) + } finally { + ready.dispose() + yield* store.dispose(ctx) + } + }), + ) }) diff --git a/script/architecture-allowlist.json b/script/architecture-allowlist.json index 024a5c6a04c..dabd6a107c6 100644 --- a/script/architecture-allowlist.json +++ b/script/architecture-allowlist.json @@ -9,7 +9,6 @@ "packages/opencode/src/kilo-sessions/kilo-sessions.ts": { "count": 1, "owner": "session-runtime", "reason": "Kilo session coordination state" }, "packages/opencode/src/kilocode/agent-manager/service.ts": { "count": 1, "owner": "agent-manager", "reason": "Agent Manager multi-project worktree state" }, "packages/opencode/src/kilocode/background-process/index.ts": { "count": 1, "owner": "process-runtime", "reason": "Directory-keyed background process registry" }, - "packages/opencode/src/kilocode/interactive-terminal/index.ts": { "count": 1, "owner": "terminal-runtime", "reason": "Interactive terminal manager state" }, "packages/opencode/src/kilocode/notebook/service.ts": { "count": 1, "owner": "notebook-runtime", "reason": "Notebook cell execution service state" }, "packages/opencode/src/kilocode/watcher.ts": { "count": 1, "owner": "watcher-runtime", "reason": "Eager location watcher subscription" } }