diff --git a/.changeset/report-cli-vscode-presence.md b/.changeset/report-cli-vscode-presence.md new file mode 100644 index 0000000000..c5229c240f --- /dev/null +++ b/.changeset/report-cli-vscode-presence.md @@ -0,0 +1,6 @@ +--- +"@kilocode/cli": minor +"kilo-code": minor +--- + +Report active CLI and VS Code app and session presence. diff --git a/packages/kilo-console/src/client.test.ts b/packages/kilo-console/src/client.test.ts index 4c94e5e909..4a964733d8 100644 --- a/packages/kilo-console/src/client.test.ts +++ b/packages/kilo-console/src/client.test.ts @@ -1,17 +1,21 @@ import { expect, test } from "bun:test" -function setup() { - const calls: Array<{ url: string; method: string; body: unknown }> = [] - const win = { - fetch: async (input: RequestInfo | URL, init?: RequestInit) => { - const req = input instanceof Request ? input : new Request(input, init) - calls.push({ url: req.url, method: req.method, body: await req.json() }) - return new Response(JSON.stringify({ permission: { edit: { "*": "allow" } } }), { - headers: { "content-type": "application/json" }, - }) - }, - } +// client.ts binds window.fetch once at import time, so every test must share the +// same window whose fetch writes into a swappable calls array. +let calls: Array<{ url: string; method: string; body: unknown }> = [] +const win = { + fetch: async (input: RequestInfo | URL, init?: RequestInit) => { + const req = input instanceof Request ? input : new Request(input, init) + calls.push({ url: req.url, method: req.method, body: await req.json() }) + return new Response(JSON.stringify({ permission: { edit: { "*": "allow" } } }), { + headers: { "content-type": "application/json" }, + }) + }, +} + +function setup() { + calls = [] Object.defineProperty(globalThis, "window", { value: win, configurable: true }) return calls } @@ -46,3 +50,24 @@ test("config writes include the selected directory", async () => { unset: [["indexing", "model"]], }) }) + +test("viewed snapshots post the presence payload against the selected directory", async () => { + const calls = setup() + const client = await import("./client") + const query = { url: "http://kilo:secret@127.0.0.1:4097", dir: "/tmp/project" } + const viewer = { id: "11111111-1111-4111-8111-111111111111", active: false } + + await client.viewProjectSessions(query, viewer, ["ses_selected", "ses_terminal"], []) + + expect(calls).toHaveLength(1) + + const viewed = calls[0] + expect(viewed.method).toBe("POST") + expect(new URL(viewed.url).pathname).toBe("/session/viewed") + expect(new URL(viewed.url).searchParams.get("directory")).toBe("/tmp/project") + expect(viewed.body).toEqual({ + viewer: { id: "11111111-1111-4111-8111-111111111111", active: false }, + attached: ["ses_selected", "ses_terminal"], + visible: [], + }) +}) diff --git a/packages/kilo-console/src/client.ts b/packages/kilo-console/src/client.ts index b1e74cadc1..9a3781c10a 100644 --- a/packages/kilo-console/src/client.ts +++ b/packages/kilo-console/src/client.ts @@ -628,9 +628,14 @@ export async function removeProjectPty(input: Query, pty: string) { return demand("Remove terminal", result) } -export async function viewProjectSessions(input: ProjectQuery, focused: string[], open: string[]) { +export async function viewProjectSessions( + input: ProjectQuery, + viewer: { id: string; active: boolean }, + attached: string[], + visible: string[], +) { const sdk = client(input) - const result = await sdk.session.viewed({ directory: input.dir, focused, open }) + const result = await sdk.session.viewed({ directory: input.dir, viewer, attached, visible }) return demand("Viewed sessions", result) } diff --git a/packages/kilo-console/src/routes/projects/ProjectConsoleRoute.tsx b/packages/kilo-console/src/routes/projects/ProjectConsoleRoute.tsx index 5363f208db..f76a2db973 100644 --- a/packages/kilo-console/src/routes/projects/ProjectConsoleRoute.tsx +++ b/packages/kilo-console/src/routes/projects/ProjectConsoleRoute.tsx @@ -49,6 +49,7 @@ import { normalizeConsoleDiffStyle, normalizeContextSidebarWidth, } from "../config/state/console" +import { sender } from "./project-console-presence-sender" import { GhosttyTerminal } from "./terminal/GhosttyTerminal" const ui = new Set(["3017", "3018"]) @@ -147,6 +148,7 @@ function terminalKey(url: string, item: ProjectTerminalItem) { export function ProjectConsoleRoute() { const loc = useLocation() const params = useParams() + const viewerId = crypto.randomUUID() const search = createMemo(() => new URLSearchParams(loc.search)) const fallback = () => base(search()) const [url, setUrl] = createSignal(fallback()) @@ -660,19 +662,37 @@ export function ProjectConsoleRoute() { if (item) clearUnread(item) }) - createEffect(() => { + let lastInput: { url: string; dir: string } | undefined + const queue = sender((err) => console.warn(`Viewed sessions: ${errMsg(err)}`)) + + function sendSnapshot(force = false) { const base = query() const data = snap() if (!base || !data) return - const focused = activeSessionID() - const open = terminals().flatMap((item) => { + const selected = activeSessionID() + const ids = new Set() + if (selected) ids.add(selected) + for (const item of terminals()) { const id = sessionID(item) - return id ? [id] : [] - }) - void viewProjectSessions({ url: base.url, dir: data.project.worktree }, focused ? [focused] : [], open).catch( - () => {}, + if (id) ids.add(id) + } + const input = { url: base.url, dir: data.project.worktree } + const key = input.url + "|" + input.dir + "|" + [...ids].sort().join(",") + lastInput = input + queue.push( + { + key, + run: async () => { + await viewProjectSessions(input, { id: viewerId, active: false }, [...ids], []) + }, + }, + force, ) - }) + } + + createEffect(() => sendSnapshot()) + + const checkin = window.setInterval(() => sendSnapshot(true), 60_000) createEffect(() => { const base = query() @@ -704,6 +724,19 @@ export function ProjectConsoleRoute() { onCleanup(() => { if (events.timer) window.clearTimeout(events.timer) if (resize.timer) window.clearTimeout(resize.timer) + window.clearInterval(checkin) + if (lastInput) { + const input = lastInput + queue.push( + { + key: input.url + "|" + input.dir + "|", + run: async () => { + await viewProjectSessions(input, { id: viewerId, active: false }, [], []) + }, + }, + true, + ) + } }) createEffect(() => { diff --git a/packages/kilo-console/src/routes/projects/project-console-presence-sender.test.ts b/packages/kilo-console/src/routes/projects/project-console-presence-sender.test.ts new file mode 100644 index 0000000000..c0fe5f855b --- /dev/null +++ b/packages/kilo-console/src/routes/projects/project-console-presence-sender.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, test } from "bun:test" +import { sender } from "./project-console-presence-sender" + +function deferred() { + const state: { resolve?: () => void } = {} + const promise = new Promise((resolve) => { + state.resolve = resolve + }) + return { promise, resolve: () => state.resolve?.() } +} + +async function drain() { + await Promise.resolve() + await Promise.resolve() +} + +describe("project console presence sender", () => { + test("sends snapshots in order", async () => { + const first = deferred() + const calls: string[] = [] + const queue = sender(() => {}) + + queue.push({ key: "first", run: () => (calls.push("first"), first.promise) }) + queue.push({ key: "second", run: async () => void calls.push("second") }) + + expect(calls).toEqual(["first"]) + first.resolve() + await drain() + expect(calls).toEqual(["first", "second"]) + }) + + test("deduplicates the last successful snapshot unless forced", async () => { + const calls: string[] = [] + const queue = sender(() => {}) + const item = { key: "same", run: async () => void calls.push("same") } + + queue.push(item) + await drain() + queue.push(item) + await drain() + expect(calls).toEqual(["same"]) + + queue.push(item, true) + await drain() + expect(calls).toEqual(["same", "same"]) + }) + + test("does not queue a reactive duplicate of an in-flight snapshot", async () => { + const wait = deferred() + const calls: string[] = [] + const queue = sender(() => {}) + const item = { key: "same", run: () => (calls.push("same"), wait.promise) } + + queue.push(item) + queue.push(item) + wait.resolve() + await drain() + + expect(calls).toEqual(["same"]) + }) + + test("retains a forced renewal while the same snapshot is in flight", async () => { + const wait = deferred() + const calls: string[] = [] + const queue = sender(() => {}) + const item = { key: "same", run: () => (calls.push("same"), calls.length === 1 ? wait.promise : Promise.resolve()) } + + queue.push(item) + queue.push(item, true) + expect(calls).toEqual(["same"]) + + wait.resolve() + await drain() + expect(calls).toEqual(["same", "same"]) + }) + + test("replaces an obsolete pending snapshot with the latest state", async () => { + const first = deferred() + const calls: string[] = [] + const queue = sender(() => {}) + + queue.push({ key: "first", run: () => (calls.push("first"), first.promise) }) + queue.push({ key: "second", run: async () => void calls.push("second") }) + queue.push({ key: "third", run: async () => void calls.push("third") }) + + first.resolve() + await drain() + expect(calls).toEqual(["first", "third"]) + }) +}) diff --git a/packages/kilo-console/src/routes/projects/project-console-presence-sender.ts b/packages/kilo-console/src/routes/projects/project-console-presence-sender.ts new file mode 100644 index 0000000000..e2363a4dbe --- /dev/null +++ b/packages/kilo-console/src/routes/projects/project-console-presence-sender.ts @@ -0,0 +1,34 @@ +type Snapshot = { + key: string + run: () => Promise +} + +export function sender(report: (err: unknown) => void) { + let current: Snapshot | undefined + let next: Snapshot | undefined + let last: string | undefined + + async function drain() { + const item = next + if (!item) return + next = undefined + current = item + try { + await item.run() + last = item.key + } catch (err) { + report(err) + } + current = undefined + if (next) void drain() + } + + return { + push(item: Snapshot, force = false) { + if (!force && item.key === last && !current) return + if (!force && item.key === current?.key && !next) return + next = item + if (!current) void drain() + }, + } +} diff --git a/packages/kilo-console/src/routes/projects/project-console-presence.test.ts b/packages/kilo-console/src/routes/projects/project-console-presence.test.ts new file mode 100644 index 0000000000..ccaef85a31 --- /dev/null +++ b/packages/kilo-console/src/routes/projects/project-console-presence.test.ts @@ -0,0 +1,69 @@ +/** + * Contract test for the presence snapshot logic in ProjectConsoleRoute.tsx. + * + * The route is a large Solid component that cannot be mounted in a unit test, + * so these source assertions pin the load-bearing presence behaviour instead: + * the console is a dashboard viewer (always inactive, never reports visible + * sessions), the attached union covers the selected session plus every terminal + * session, the sender serializes snapshots and forced check-ins, and cleanup + * reuses the exact url+dir the last regular snapshot used to queue a final empty + * snapshot. + */ + +import { describe, expect, test } from "bun:test" +import fs from "node:fs" +import path from "node:path" + +const ROUTE_FILE = path.resolve(import.meta.dir, "./ProjectConsoleRoute.tsx") + +/** Collapse whitespace so multi-line expressions match regardless of formatting. */ +function flat(source: string) { + return source.replace(/\s+/g, " ").replace(/\( /g, "(").replace(/ \)/g, ")").replace(/,\)/g, ")") +} + +describe("project console presence contract", () => { + test("snapshots always report an inactive viewer with no visible sessions", () => { + const content = fs.readFileSync(ROUTE_FILE, "utf-8") + expect(content).toContain("const viewerId = crypto.randomUUID()") + expect(flat(content)).toContain( + "run: async () => { await viewProjectSessions(input, { id: viewerId, active: false }, [...ids], []) }", + ) + expect(content).not.toContain("active: true") + }) + + test("attached union includes the selected session and every terminal session", () => { + const content = fs.readFileSync(ROUTE_FILE, "utf-8") + expect(content).toContain("const selected = activeSessionID()") + expect(content).toContain("const ids = new Set()") + expect(content).toContain("if (selected) ids.add(selected)") + expect(flat(content)).toContain( + "for (const item of terminals()) { const id = sessionID(item) if (id) ids.add(id) }", + ) + }) + + test("snapshots record the url+dir they were sent with", () => { + const content = fs.readFileSync(ROUTE_FILE, "utf-8") + expect(content).toContain("let lastInput: { url: string; dir: string } | undefined") + expect(content).toContain("const input = { url: base.url, dir: data.project.worktree }") + expect(content).toContain("lastInput = input") + }) + + test("routes reactive snapshots and forced check-ins through the serialized sender", () => { + const content = fs.readFileSync(ROUTE_FILE, "utf-8") + expect(content).toContain('import { sender } from "./project-console-presence-sender"') + expect(content).toContain("const queue = sender") + expect(content).toContain("function sendSnapshot(force = false)") + expect(content).toContain("queue.push(") + expect(content).toContain("createEffect(() => sendSnapshot())") + expect(content).toContain("const checkin = window.setInterval(() => sendSnapshot(true), 60_000)") + expect(content).toContain("window.clearInterval(checkin)") + }) + + test("cleanup queues a final empty snapshot using the last snapshot's url+dir", () => { + const content = flat(fs.readFileSync(ROUTE_FILE, "utf-8")) + expect(content).toContain( + 'window.clearInterval(checkin) if (lastInput) { const input = lastInput queue.push({ key: input.url + "|" + input.dir + "|", run: async () => { await viewProjectSessions(input, { id: viewerId, active: false }, [], []) }, }, true) }', + ) + expect(content).not.toContain("dir: base.dir") + }) +}) diff --git a/packages/kilo-vscode/src/KiloProvider.ts b/packages/kilo-vscode/src/KiloProvider.ts index 56f6beb127..7f8a2f0650 100644 --- a/packages/kilo-vscode/src/KiloProvider.ts +++ b/packages/kilo-vscode/src/KiloProvider.ts @@ -502,8 +502,21 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper } private focusSession(id?: string): void { this.streams.focus(id) - if (id) this.connectionService.registerFocused(this.instanceId, id) - else this.connectionService.unregisterFocused(this.instanceId) + this.registerPresence() + } + + /** + * Report presence for this provider: the focused session is visible, and + * open local tab sessions (plus the focused one) stay attached even while + * the view is hidden. + */ + private registerPresence(): void { + if (this.opts.disableViewedRegistration) return + const focused = this.streams.focused + this.connectionService.registerVisible(this.instanceId, focused ? [focused] : []) + const attached = new Set(this.openSessionIds) + if (focused) attached.add(focused) + this.connectionService.registerAttached(this.instanceId, [...attached]) } public setStreamVisibility(active: boolean): void { @@ -735,9 +748,12 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper this.setupWebviewMessageHandler(panel.webview) this.viewStateDisposable?.dispose() - this.viewStateDisposable = this.visibleTaskStreams.bindPanel(panel, () => - this.focusSession(panel.active ? this.currentSession?.id : undefined), - ) + this.viewStateDisposable = this.visibleTaskStreams.bindPanel(panel, () => { + if (this.opts.disableViewedRegistration) return + const id = this.contextSessionID + this.streams.focus(panel.visible ? id : undefined) + this.connectionService.registerVisible(this.instanceId, panel.visible && id ? [id] : []) + }) this.initializeConnection() } @@ -832,7 +848,7 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper for (const [key, session] of this.draftSessions) { if (next.has(session.sid) || session.expires <= now) this.draftSessions.delete(key) } - this.connectionService.registerOpen(this.instanceId, ids) + this.registerPresence() this.recoverPendingPrompts() } @@ -4401,8 +4417,9 @@ export class KiloProvider implements vscode.WebviewViewProvider, TelemetryProper */ dispose(): void { this.unsubscribeRemote?.() - this.focusSession() - this.connectionService.registerOpen(this.instanceId, []) + this.streams.focus(undefined) + this.connectionService.unregisterVisible(this.instanceId) + this.connectionService.unregisterAttached(this.instanceId) this.statsPoller?.stop() this.statsGitOps?.dispose() this.unsubscribeEvent?.() diff --git a/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts b/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts index a40840eea6..97351e1cdf 100644 --- a/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts +++ b/packages/kilo-vscode/src/agent-manager/AgentManagerProvider.ts @@ -27,6 +27,7 @@ import { startVscodeRunTask } from "./run/task" import { RunController } from "./run/controller" import { handleRunMessage } from "./run/message" import { forkSession } from "./fork-session" +import { AgentManagerVisiblePresence } from "./am-visible-presence" import { continueInWorktree } from "./continue-in-worktree" import { WorktreeDiffController } from "./worktree-diff-controller" import { WorktreeImporter } from "./worktree-importer" @@ -83,10 +84,13 @@ export class AgentManagerProvider implements Disposable { private closing: Promise | undefined private onVisibilityChange: ((visible: boolean) => void) | undefined - /** Session ID most recently loaded via a `loadMessages` message from the webview. - * Updated synchronously — unlike the session provider's currentSession which depends on - * an async `session.get` round-trip and can be stale during rapid tab switches. */ + /** Session ID most recently loaded via `loadMessages`; updated synchronously. */ private activeSessionId: string | undefined + private visiblePresence = new AgentManagerVisiblePresence( + (ids) => this.connectionService.registerVisible("agent-manager", ids), + () => this.panel?.visible ?? false, + (ids) => this.connectionService.registerAttached("agent-manager", ids), + ) constructor( private readonly host: Host, private readonly connectionService: KiloConnectionService, @@ -257,6 +261,7 @@ export class AgentManagerProvider implements Disposable { this.onVisibilityChange?.(ctx.visible) ctx.onDidChangeVisibility((visible) => { this.statsPoller.setVisible(visible) + this.visiblePresence.flush() }) ctx.sessions.onFollowupAdopted((session, directory) => { @@ -276,8 +281,7 @@ export class AgentManagerProvider implements Disposable { this.prBridge.poller.stop() this.diffs.stop() this.activeSessionId = undefined - this.connectionService.unregisterFocused("agent-manager") - this.connectionService.registerOpen("agent-manager", []) + this.visiblePresence.clear() this.panel = undefined this.onVisibilityChange?.(false) } @@ -509,7 +513,6 @@ export class AgentManagerProvider implements Disposable { if (m.type === "loadMessages") { this.activeSessionId = m.sessionID - this.connectionService.registerFocused("agent-manager", m.sessionID) this.terminalManager.syncOnSessionSwitch(m.sessionID) this.prBridge.poller.setActiveWorktreeId(this.state?.getSession(m.sessionID)?.worktreeId ?? undefined) return msg @@ -517,7 +520,7 @@ export class AgentManagerProvider implements Disposable { if (m.type === "clearSession") { this.activeSessionId = undefined - this.connectionService.unregisterFocused("agent-manager") + this.visiblePresence.setDisplayed(null) void Promise.resolve().then(() => { if (!this.panel || !this.state) return for (const id of this.state.worktreeSessionIds()) { @@ -535,8 +538,8 @@ export class AgentManagerProvider implements Disposable { return msg } - if (m.type === "agentManager.openSessions") { - this.connectionService.registerOpen("agent-manager", m.sessionIDs) + if (m.type === "agentManager.openSessions" || m.type === "agentManager.visibleSession") { + this.visiblePresence.handle(m) return null } } @@ -1475,10 +1478,6 @@ export class AgentManagerProvider implements Disposable { return null } - // --------------------------------------------------------------------------- - // Keybindings - // --------------------------------------------------------------------------- - private sendKeybindings(): void { const keybindings = this.host.extensionKeybindings() const bindings = buildKeybindingMap(keybindings, process.platform === "darwin") @@ -1982,8 +1981,7 @@ export class AgentManagerProvider implements Disposable { this.unsubTool?.() this.unsubStatus?.() this.unsubFont?.() - this.connectionService.unregisterFocused("agent-manager") - this.connectionService.registerOpen("agent-manager", []) + this.visiblePresence.clear() this.diffs.stop() this.naming.dispose() this.statsPoller.stop() diff --git a/packages/kilo-vscode/src/agent-manager/am-visible-presence.test.ts b/packages/kilo-vscode/src/agent-manager/am-visible-presence.test.ts new file mode 100644 index 0000000000..9f96bead35 --- /dev/null +++ b/packages/kilo-vscode/src/agent-manager/am-visible-presence.test.ts @@ -0,0 +1,132 @@ +import { describe, expect, test } from "bun:test" +import { AgentManagerVisiblePresence } from "./am-visible-presence" + +function setup(initialVisible = true) { + const calls: string[][] = [] + const attached: string[][] = [] + let visible = initialVisible + const presence = new AgentManagerVisiblePresence( + (ids) => calls.push(ids), + () => visible, + (ids) => attached.push(ids), + ) + return { + calls, + attached, + presence, + setVisible(value: boolean) { + visible = value + }, + } +} + +describe("AgentManagerVisiblePresence", () => { + test("registers the displayed id while the panel is visible", () => { + const { calls, presence } = setup(true) + + presence.setDisplayed("ses_1") + + expect(calls.at(-1)).toEqual(["ses_1"]) + }) + + test("flush registers empty when the panel is hidden", () => { + const { calls, presence, setVisible } = setup(true) + presence.setDisplayed("ses_1") + + setVisible(false) + presence.flush() + + expect(calls.at(-1)).toEqual([]) + }) + + test("flush clears attached when the panel is hidden", () => { + const { attached, presence, setVisible } = setup(true) + presence.handle({ type: "agentManager.openSessions", sessionIDs: ["ses_1", "ses_2"] }) + + setVisible(false) + presence.flush() + + expect(attached.at(-1)).toEqual([]) + }) + + test("flush re-registers attached when the panel becomes visible again", () => { + const { attached, presence, setVisible } = setup(true) + presence.handle({ type: "agentManager.openSessions", sessionIDs: ["ses_1", "ses_2"] }) + + setVisible(false) + presence.flush() + setVisible(true) + presence.flush() + + expect(attached.at(-1)).toEqual(["ses_1", "ses_2"]) + }) + + test("setDisplayed(null) registers empty even while visible", () => { + const { calls, presence } = setup(true) + presence.setDisplayed("ses_1") + + presence.setDisplayed(null) + + expect(calls.at(-1)).toEqual([]) + }) + + test("flush after visibility returns re-registers the retained id", () => { + const { calls, presence, setVisible } = setup(false) + presence.setDisplayed("ses_1") + expect(calls.at(-1)).toEqual([]) + + setVisible(true) + presence.flush() + + expect(calls.at(-1)).toEqual(["ses_1"]) + }) + + test("setDisplayed(null) prevents a stale id from re-registering on a later flush", () => { + const { calls, presence, setVisible } = setup(true) + presence.setDisplayed("ses_1") + + setVisible(false) + presence.setDisplayed(null) + setVisible(true) + presence.flush() + + expect(calls.at(-1)).toEqual([]) + }) + + test("handle routes openSessions to attached and visibleSession to visible", () => { + const { calls, attached, presence } = setup(true) + + presence.handle({ type: "agentManager.openSessions", sessionIDs: ["ses_1", "ses_2"] }) + presence.handle({ type: "agentManager.visibleSession", sessionID: "ses_1" }) + + expect(attached.at(-1)).toEqual(["ses_1", "ses_2"]) + expect(calls.at(-1)).toEqual(["ses_1"]) + }) + + test("handle while hidden stores state but registers empty", () => { + const { calls, attached, presence, setVisible } = setup(false) + + presence.handle({ type: "agentManager.openSessions", sessionIDs: ["ses_1"] }) + presence.handle({ type: "agentManager.visibleSession", sessionID: "ses_1" }) + + expect(calls.at(-1)).toEqual([]) + expect(attached.at(-1)).toEqual([]) + + setVisible(true) + presence.flush() + + expect(calls.at(-1)).toEqual(["ses_1"]) + expect(attached.at(-1)).toEqual(["ses_1"]) + }) + + test("clear empties both the visible and attached registrations", () => { + const { calls, attached, presence } = setup(true) + presence.setDisplayed("ses_1") + presence.handle({ type: "agentManager.openSessions", sessionIDs: ["ses_1"] }) + + presence.clear() + + expect(calls.at(-1)).toEqual([]) + expect(attached.at(-1)).toEqual([]) + }) +}) diff --git a/packages/kilo-vscode/src/agent-manager/am-visible-presence.ts b/packages/kilo-vscode/src/agent-manager/am-visible-presence.ts new file mode 100644 index 0000000000..935ab174ce --- /dev/null +++ b/packages/kilo-vscode/src/agent-manager/am-visible-presence.ts @@ -0,0 +1,51 @@ +/** Vscode-free presence state for the Agent Manager. + * + * Owns the displayed session id and the open-tab session set. Both are gated + * on panel visibility: when the panel is hidden (retainContextWhenHidden + * keeps the webview alive), flush() clears both registrations so the retained + * webview's reactive updates cannot keep stale sessions attached or visible. + * When the panel returns, flush() re-registers from stored state. */ + +type Register = (ids: string[]) => void + +type PresenceMessage = + | { type: "agentManager.openSessions"; sessionIDs: string[] } + | { type: "agentManager.visibleSession"; sessionID: string | null } + +export class AgentManagerVisiblePresence { + private id: string | null = null + private open: string[] = [] + constructor( + private readonly register: Register, + private readonly panelVisible: () => boolean, + private readonly registerAttached: Register, + ) {} + + setDisplayed(id: string | null): void { + this.id = id + this.flush() + } + + flush(): void { + if (this.panelVisible()) { + this.register(this.id ? [this.id] : []) + this.registerAttached(this.open) + } else { + this.register([]) + this.registerAttached([]) + } + } + + handle(m: PresenceMessage): void { + if (m.type === "agentManager.openSessions") this.open = m.sessionIDs + else this.id = m.sessionID + this.flush() + } + + clear(): void { + this.id = null + this.open = [] + this.register([]) + this.registerAttached([]) + } +} diff --git a/packages/kilo-vscode/src/agent-manager/types.ts b/packages/kilo-vscode/src/agent-manager/types.ts index 0476f4a290..528abb6491 100644 --- a/packages/kilo-vscode/src/agent-manager/types.ts +++ b/packages/kilo-vscode/src/agent-manager/types.ts @@ -564,6 +564,11 @@ interface OpenSessionsIn { sessionIDs: string[] } +interface VisibleSessionIn { + type: "agentManager.visibleSession" + sessionID: string | null +} + interface OpenFileIn { type: "agentManager.openFile" sessionId: string @@ -809,6 +814,7 @@ export type AgentManagerInMessage = | RefreshPRIn | OpenPRIn | OpenSessionsIn + | VisibleSessionIn | OpenFileIn | GenericOpenFileIn | PreviewImageIn diff --git a/packages/kilo-vscode/src/agent-manager/vscode-host.ts b/packages/kilo-vscode/src/agent-manager/vscode-host.ts index 24d9640877..9a2a3ef315 100644 --- a/packages/kilo-vscode/src/agent-manager/vscode-host.ts +++ b/packages/kilo-vscode/src/agent-manager/vscode-host.ts @@ -96,6 +96,7 @@ export class VscodeHost implements Host { snapshotInitialization: SNAPSHOT_INITIALIZATION, slimEditMetadata: true, worktreeDirectories: () => opts.worktreeDirectories?.() ?? [], + disableViewedRegistration: true, }) if (this.diffVirtual) { provider.setDiffVirtualProvider(this.diffVirtual) diff --git a/packages/kilo-vscode/src/kilo-provider/options.ts b/packages/kilo-vscode/src/kilo-provider/options.ts index 32f5d3793c..f32caa4736 100644 --- a/packages/kilo-vscode/src/kilo-provider/options.ts +++ b/packages/kilo-vscode/src/kilo-provider/options.ts @@ -5,4 +5,6 @@ export type KiloProviderOptions = { slimEditMetadata?: boolean tabTitle?: (title: string) => void worktreeDirectories?: () => string[] + /** Composite hosts (Agent Manager) own viewed/presence registration themselves. */ + disableViewedRegistration?: boolean } diff --git a/packages/kilo-vscode/src/services/cli-backend/connection-service.test.ts b/packages/kilo-vscode/src/services/cli-backend/connection-service.test.ts index afb1a3d2ec..17420f7bc6 100644 --- a/packages/kilo-vscode/src/services/cli-backend/connection-service.test.ts +++ b/packages/kilo-vscode/src/services/cli-backend/connection-service.test.ts @@ -40,9 +40,9 @@ describe("KiloConnectionService clients", () => { }) describe("KiloConnectionService viewed sessions", () => { - test("keeps Agent Manager sessions when sidebar focus changes during a flush", async () => { + test("keeps Agent Manager sessions when sidebar visibility changes during a flush", async () => { const service = new KiloConnectionService({} as any) - const calls: Array<{ focused: string[]; open?: string[] }> = [] + const calls: Array<{ viewer: { id: string; active: boolean }; attached: string[]; visible: string[] }> = [] let release!: () => void const gate = new Promise((resolve) => { release = resolve @@ -50,10 +50,9 @@ describe("KiloConnectionService viewed sessions", () => { let active = 0 let max = 0 - ;(service as any).remoteService = { getState: () => ({ enabled: true }) } ;(service as any).client = { session: { - viewed: async (input: { focused: string[]; open?: string[] }) => { + viewed: async (input: { viewer: { id: string; active: boolean }; attached: string[]; visible: string[] }) => { calls.push(input) active += 1 max = Math.max(max, active) @@ -63,23 +62,91 @@ describe("KiloConnectionService viewed sessions", () => { }, } - service.registerFocused("agent-manager", "am-1") - service.registerOpen("agent-manager", ["am-1", "am-2"]) + service.registerVisible("agent-manager", ["am-1"]) + service.registerAttached("agent-manager", ["am-1", "am-2"]) await Bun.sleep(175) - expect(calls).toEqual([{ focused: ["am-1"], open: ["am-2"] }]) + expect(calls).toHaveLength(1) + expect([...calls[0].visible].sort()).toEqual(["am-1"]) + expect([...calls[0].attached].sort()).toEqual(["am-1", "am-2"]) - service.registerFocused("sidebar", "side-1") + service.registerVisible("sidebar", ["side-1"]) await Bun.sleep(175) expect(calls).toHaveLength(1) release() await Bun.sleep(10) expect(max).toBe(1) - expect(calls[1]).toEqual({ focused: ["am-1", "side-1"], open: ["am-2"] }) + expect([...calls[1].visible].sort()).toEqual(["am-1", "side-1"]) + expect([...calls[1].attached].sort()).toEqual(["am-1", "am-2", "side-1"]) - service.unregisterFocused("sidebar") + service.registerVisible("sidebar", []) await Bun.sleep(175) - expect(calls[2]).toEqual({ focused: ["am-1"], open: ["am-2"] }) + expect([...calls[2].visible].sort()).toEqual(["am-1"]) + expect([...calls[2].attached].sort()).toEqual(["am-1", "am-2"]) + }) + + test("window focus gates viewer.active but not attachment", async () => { + const window = vscode.window as unknown as { + state: { focused: boolean } + onDidChangeWindowState: (listener: (ws: { focused: boolean }) => void) => { dispose(): void } + } + const original = window.onDidChangeWindowState + let listener: ((ws: { focused: boolean }) => void) | undefined + window.onDidChangeWindowState = (cb) => { + listener = cb + return { dispose: () => {} } + } + + try { + const service = new KiloConnectionService({} as any) + const calls: Array<{ viewer: { id: string; active: boolean }; attached: string[]; visible: string[] }> = [] + ;(service as any).client = { + session: { + viewed: async (input: (typeof calls)[number]) => { + calls.push(input) + }, + }, + } + + service.registerVisible("sidebar", ["ses-1"]) + service.registerAttached("sidebar", ["ses-1", "ses-2"]) + await Bun.sleep(175) + expect(calls).toHaveLength(1) + expect(calls[0].viewer.active).toBe(true) + + listener!({ focused: false }) + await Bun.sleep(175) + expect(calls).toHaveLength(2) + expect(calls[1].viewer.active).toBe(false) + expect([...calls[1].visible].sort()).toEqual(["ses-1"]) + expect([...calls[1].attached].sort()).toEqual(["ses-1", "ses-2"]) + } finally { + window.onDidChangeWindowState = original + } + }) + + test("sends snapshots while remote control is disabled", async () => { + const service = new KiloConnectionService({} as any) + const calls: Array<{ viewer: { id: string; active: boolean }; attached: string[]; visible: string[] }> = [] + ;(service as any).client = { + session: { + viewed: async (input: (typeof calls)[number]) => { + calls.push(input) + }, + }, + } + service.setRemoteService({ + getState: () => ({ enabled: false, connected: false }), + onChange: () => () => {}, + } as any) + + service.registerVisible("sidebar", ["ses-1"]) + service.registerAttached("agent-manager", ["ses-2"]) + await Bun.sleep(175) + + expect(calls).toHaveLength(1) + expect([...calls[0].visible].sort()).toEqual(["ses-1"]) + expect([...calls[0].attached].sort()).toEqual(["ses-1", "ses-2"]) }) }) diff --git a/packages/kilo-vscode/src/services/cli-backend/connection-service.ts b/packages/kilo-vscode/src/services/cli-backend/connection-service.ts index f5d97798b3..7a7e120965 100644 --- a/packages/kilo-vscode/src/services/cli-backend/connection-service.ts +++ b/packages/kilo-vscode/src/services/cli-backend/connection-service.ts @@ -57,6 +57,12 @@ function isNotFound(err: unknown) { return false } +function sameSet(a: Set, b: Set): boolean { + if (a.size !== b.size) return false + for (const id of a) if (!b.has(id)) return false + return true +} + // Poll /global/health every 10 seconds. // This provides a second detection channel for server death independent of the SSE heartbeat. const HEALTH_POLL_INTERVAL_MS = 10_000 @@ -111,10 +117,14 @@ export class KiloConnectionService { */ private readonly messageSessionIdsByMessageId: Map = new Map() - /** Provider key → single focused session ID. */ - private readonly focused: Map = new Map() - /** Provider key → all open (background) session IDs. */ - private readonly opened: Map = new Map() + private readonly viewerId = crypto.randomUUID() + private active = true + private windowStateDisposable: vscode.Disposable | null = null + private checkinTimer: ReturnType | null = null + /** Provider key → attached (retained for remote control) session IDs. */ + private readonly attached: Map> = new Map() + /** Provider key → visibly rendered session IDs. */ + private readonly visible: Map> = new Map() private debounceTimer: ReturnType | null = null private viewedSending = false private viewedDirty = false @@ -129,6 +139,11 @@ export class KiloConnectionService { } satisfies Pick) this.sandboxPreference = new SandboxPreference(state) this.serverManager = new ServerManager(context, (code) => this.handleServerExit(code)) + this.active = vscode.window.state.focused + this.windowStateDisposable = vscode.window.onDidChangeWindowState((ws) => { + this.active = ws.focused + this.flushViewed() + }) } /** @@ -293,14 +308,15 @@ export class KiloConnectionService { for (const [mid, sid] of this.messageSessionIdsByMessageId) { if (sid === sessionId) this.messageSessionIdsByMessageId.delete(mid) } - for (const [key, sid] of this.focused) { - if (sid === sessionId) this.focused.delete(key) + for (const [key, ids] of this.attached) { + if (!ids.has(sessionId)) continue + ids.delete(sessionId) + if (ids.size === 0) this.attached.delete(key) } - for (const [key, ids] of this.opened) { - if (!ids.includes(sessionId)) continue - const next = ids.filter((id) => id !== sessionId) - if (next.length === 0) this.opened.delete(key) - else this.opened.set(key, next) + for (const [key, ids] of this.visible) { + if (!ids.has(sessionId)) continue + ids.delete(sessionId) + if (ids.size === 0) this.visible.delete(key) } this.flushViewed() } @@ -584,38 +600,49 @@ export class KiloConnectionService { } /** - * Register the session a provider is actively viewing (focused). - * After any change the aggregated set is sent to the server (debounced). + * Register the sessions a provider retains for remote control (attached). + * Sent to the server (debounced) regardless of remote-control enablement. */ - registerFocused(key: string, sessionID: string): void { - if (this.focused.get(key) === sessionID) return - this.focused.set(key, sessionID) + registerAttached(key: string, ids: string[]): void { + const next = new Set(ids) + const prev = this.attached.get(key) + if (prev && sameSet(prev, next)) return + this.attached.set(key, next) this.flushViewed() } /** - * Unregister a provider's focused session (e.g. on dispose, hidden, or clearSession). + * Unregister a provider's attached sessions (e.g. on dispose or clear). */ - unregisterFocused(key: string): void { - if (!this.focused.has(key)) return - this.focused.delete(key) + unregisterAttached(key: string): void { + if (!this.attached.has(key)) return + this.attached.delete(key) this.flushViewed() } /** - * Register the open (background tab) session IDs for a provider. - * Sessions that appear in both focused and open are reported as focused only. + * Register the sessions a provider visibly renders (visible). + * Visible sessions are also reported as attached. */ - registerOpen(key: string, ids: string[]): void { - const prev = this.opened.get(key) - if (prev && prev.length === ids.length && prev.every((v, i) => v === ids[i])) return - this.opened.set(key, ids) + registerVisible(key: string, ids: string[]): void { + const next = new Set(ids) + const prev = this.visible.get(key) + if (prev && sameSet(prev, next)) return + this.visible.set(key, next) this.flushViewed() } - /** Debounced: send the aggregated focused + open session IDs to the server. */ + /** + * Unregister a provider's visible sessions (e.g. on hide, clear, or dispose). + */ + unregisterVisible(key: string): void { + if (!this.visible.has(key)) return + this.visible.delete(key) + this.flushViewed() + } + + /** Debounced: send the aggregated attached + visible snapshot to the server. Works even when remote control is disabled. */ flushViewed(): void { - if (!this.isRemoteEnabled()) return if (this.debounceTimer) clearTimeout(this.debounceTimer) this.debounceTimer = setTimeout(() => { this.debounceTimer = null @@ -624,28 +651,21 @@ export class KiloConnectionService { } private sendViewed(): void { - if (!this.isRemoteEnabled()) { - this.viewedDirty = false - return - } if (this.viewedSending) { this.viewedDirty = true return } if (!this.client) return - const focus = new Set(this.focused.values()) - const open = new Set() - for (const ids of this.opened.values()) { - for (const id of ids) { - if (!focus.has(id)) open.add(id) - } - } + const visible = new Set() + for (const ids of this.visible.values()) for (const id of ids) visible.add(id) + const attached = new Set(visible) + for (const ids of this.attached.values()) for (const id of ids) attached.add(id) this.viewedSending = true this.viewedDirty = false void this.client.session - .viewed({ focused: [...focus], open: [...open] }) + .viewed({ viewer: { id: this.viewerId, active: this.active }, attached: [...attached], visible: [...visible] }) .catch((err) => console.warn("[Kilo New] ConnectionService: viewed flush failed:", err)) .finally(() => { this.viewedSending = false @@ -674,12 +694,23 @@ export class KiloConnectionService { this.permissionDirectories.clear() this.questionDirectories.clear() this.questionRevision += 1 - this.focused.clear() - this.opened.clear() + if (this.client?.session?.viewed) { + void this.client.session + .viewed({ viewer: { id: this.viewerId, active: false }, attached: [], visible: [] }) + .catch(() => {}) + } + this.attached.clear() + this.visible.clear() if (this.debounceTimer) { clearTimeout(this.debounceTimer) this.debounceTimer = null } + if (this.checkinTimer) { + clearInterval(this.checkinTimer) + this.checkinTimer = null + } + this.windowStateDisposable?.dispose() + this.windowStateDisposable = null this.viewedDirty = false this.unsubRemote?.() this.unsubRemote = null @@ -747,6 +778,7 @@ export class KiloConnectionService { private resetConnection(): void { this.stopHealthPoll() + this.stopCheckin() const sse = this.sseClient this.sseClient = null sse?.disconnect() @@ -837,6 +869,7 @@ export class KiloConnectionService { resolveConnected?.() resolveConnected = null rejectConnected = null + this.flushViewed() return } @@ -851,10 +884,24 @@ export class KiloConnectionService { await connectedPromise + this.startCheckin() // Start the independent health poll once we are confirmed connected. this.startHealthPoll(config.baseUrl, config.password) } + private startCheckin(): void { + this.stopCheckin() + this.checkinTimer = setInterval(() => this.flushViewed(), 60_000) + this.checkinTimer.unref?.() + } + + private stopCheckin(): void { + if (this.checkinTimer) { + clearInterval(this.checkinTimer) + this.checkinTimer = null + } + } + private handlePermissionEvent(event: SSEPayload, directory?: string): void { if (event.type === "permission.asked" && directory) { this.recordPermissionDirectory(event.properties.id, directory) diff --git a/packages/kilo-vscode/tests/setup/vscode-mock.ts b/packages/kilo-vscode/tests/setup/vscode-mock.ts index d85485821f..a5914f7bff 100644 --- a/packages/kilo-vscode/tests/setup/vscode-mock.ts +++ b/packages/kilo-vscode/tests/setup/vscode-mock.ts @@ -80,6 +80,8 @@ const mockVscode = { }, window: { activeTextEditor: undefined, + state: { focused: true }, + onDidChangeWindowState: () => ({ dispose: noop }), activeNotebookEditor: undefined, visibleTextEditors: [], visibleNotebookEditors: [], diff --git a/packages/kilo-vscode/tests/unit/agent-manager-arch.test.ts b/packages/kilo-vscode/tests/unit/agent-manager-arch.test.ts index 574b345bfd..efce41c2ba 100644 --- a/packages/kilo-vscode/tests/unit/agent-manager-arch.test.ts +++ b/packages/kilo-vscode/tests/unit/agent-manager-arch.test.ts @@ -216,8 +216,12 @@ describe("Agent Manager Provider Messages", () => { it("clears remote session registrations when the panel closes", () => { const body = getMethodBody("attachPanel") - expect(body).toContain('this.connectionService.unregisterFocused("agent-manager")') - expect(body).toContain('this.connectionService.registerOpen("agent-manager", [])') + // Presence must be cleared via visiblePresence.clear() — a direct + // registerVisible("agent-manager", []) would leave a stale displayed id + // that re-registers on the next flush after the panel reopens. + expect(body).toContain("this.visiblePresence.clear()") + expect(body).not.toContain('this.connectionService.registerVisible("agent-manager"') + expect(body).not.toContain('this.connectionService.registerAttached("agent-manager"') expect(body).toContain("this.activeSessionId = undefined") }) diff --git a/packages/kilo-vscode/tests/unit/agent-manager-remote-sessions.test.ts b/packages/kilo-vscode/tests/unit/agent-manager-remote-sessions.test.ts new file mode 100644 index 0000000000..c31c333663 --- /dev/null +++ b/packages/kilo-vscode/tests/unit/agent-manager-remote-sessions.test.ts @@ -0,0 +1,28 @@ +import { expect, test } from "bun:test" +import fs from "node:fs" +import path from "node:path" +import { visible } from "../../webview-ui/agent-manager/remote-sessions" + +const APP = path.resolve(import.meta.dir, "../../webview-ui/agent-manager/AgentManagerApp.tsx") + +function flat(source: string) { + return source.replace(/\s+/g, " ") +} + +test("reports a real session only while its chat surface is displayed", () => { + expect(visible("ses_1", false)).toBe("ses_1") + expect(visible("ses_1", true)).toBeNull() +}) + +test("does not report synthetic pending or cloud preview IDs", () => { + expect(visible("pending:1", false)).toBeNull() + expect(visible("cloud:1", false)).toBeNull() +}) + +test("blocks visible presence while setup or an empty pane covers chat", () => { + const source = flat(fs.readFileSync(APP, "utf-8")) + expect(source).toContain( + "visible( session.currentSessionID(), !!terms.activeId() || reviewActive() || history() || !!overlay() || contextEmpty(), )", + ) + expect(source).toContain("") +}) diff --git a/packages/kilo-vscode/tests/unit/kilo-provider-load-messages.test.ts b/packages/kilo-vscode/tests/unit/kilo-provider-load-messages.test.ts index db6fbb75f5..e3f39c994e 100644 --- a/packages/kilo-vscode/tests/unit/kilo-provider-load-messages.test.ts +++ b/packages/kilo-vscode/tests/unit/kilo-provider-load-messages.test.ts @@ -199,9 +199,10 @@ function createConnection(client: ReturnType) { recordMessageSessionId: () => undefined, notifyNotificationDismissed: () => undefined, pruneSession: () => undefined, - registerFocused: () => undefined, - unregisterFocused: () => undefined, - registerOpen: () => undefined, + registerVisible: () => undefined, + unregisterVisible: () => undefined, + registerAttached: () => undefined, + unregisterAttached: () => undefined, } } diff --git a/packages/kilo-vscode/tests/unit/presence-registration-contract.test.ts b/packages/kilo-vscode/tests/unit/presence-registration-contract.test.ts new file mode 100644 index 0000000000..e97cf52c43 --- /dev/null +++ b/packages/kilo-vscode/tests/unit/presence-registration-contract.test.ts @@ -0,0 +1,180 @@ +/** + * Source contract tests for session-presence registration. + * + * Static analysis — reads KiloProvider.ts, AgentManagerProvider.ts, + * vscode-host.ts, and connection-service.ts and verifies the locked + * viewed/presence behavior from the presence plan: + * + * - Editor panels register visible keyed on `panel.visible` and the + * synchronous `contextSessionID`; attachment persists while hidden. + * - Embedded Agent Manager providers skip generic viewed registration + * (`disableViewedRegistration`) so sessions are not double-reported. + * - The connection service resends the full snapshot on backend reconnect. + * - Agent Manager visible presence is routed through + * AgentManagerVisiblePresence so cleanup cannot leave a stale displayed id. + * + * Protects against accidental removal during Kilo development. + */ + +import { describe, it, expect } from "bun:test" +import fs from "node:fs" +import path from "node:path" + +const ROOT = path.resolve(import.meta.dir, "../..") +const KILOPROVIDER_FILE = path.join(ROOT, "src/KiloProvider.ts") +const AGENT_MANAGER_PROVIDER_FILE = path.join(ROOT, "src/agent-manager/AgentManagerProvider.ts") +const VSCODE_HOST_FILE = path.join(ROOT, "src/agent-manager/vscode-host.ts") +const CONNECTION_SERVICE_FILE = path.join(ROOT, "src/services/cli-backend/connection-service.ts") + +function readFile(filePath: string): string { + return fs.readFileSync(filePath, "utf-8") +} + +describe("KiloProvider editor-panel visible registration contract", () => { + const source = readFile(KILOPROVIDER_FILE) + // The bindPanel callback installed in resolveWebviewPanel. + const match = source.match( + /this\.viewStateDisposable = this\.visibleTaskStreams\.bindPanel\(panel, \(\) => \{([\s\S]*?)\n {4}\}\)/, + ) + + it("binds a view-state callback on the panel", () => { + expect(match).not.toBeNull() + }) + + it("registers visible keyed on panel.visible and the synchronous contextSessionID", () => { + // `panel.active` would drop visible-but-inactive split editors, and + // `this.currentSession?.id` is populated asynchronously — rapid A→B + // navigation must report B without awaiting B's metadata fetch. + const body = match![1] + expect(body).toContain("this.contextSessionID") + expect(body).toContain("panel.visible") + expect(body).toContain("this.connectionService.registerVisible(this.instanceId,") + expect(body).not.toContain("panel.active") + expect(body).not.toContain("this.currentSession") + }) + + it("does not clear attachment when the panel is hidden", () => { + // Hidden editor tabs stay reachable for remote control: the panel + // view-state callback must never touch the attached registration + // (directly or via focusSession) — attachment is cleared only by the + // dispose/clear/delete paths. + const body = match![1] + expect(body).toContain("this.streams.focus(panel.visible ? id : undefined)") + expect(body).not.toContain("registerAttached") + expect(body).not.toContain("focusSession") + }) +}) + +describe("KiloProvider disableViewedRegistration contract", () => { + const kiloProvider = readFile(KILOPROVIDER_FILE) + const vscodeHost = readFile(VSCODE_HOST_FILE) + + it("registerPresence skips viewed registration when the option is set", () => { + const match = kiloProvider.match(/private registerPresence\(\): void \{([\s\S]*?)\n {2}\}/) + expect(match).not.toBeNull() + const body = match![1] + const guard = body.indexOf("if (this.opts.disableViewedRegistration) return") + const visible = body.indexOf("this.connectionService.registerVisible(this.instanceId,") + const attached = body.indexOf("this.connectionService.registerAttached(this.instanceId,") + expect(guard).toBeGreaterThanOrEqual(0) + expect(visible).toBeGreaterThan(guard) + expect(attached).toBeGreaterThan(guard) + }) + + it("focusSession and trackOpenSessions report through registerPresence", () => { + // Both the focused session (visible) and the open local tabs (attached) + // funnel into one snapshot so neither write can clobber the other. + const focus = kiloProvider.match(/private focusSession\(id\?: string\): void \{([\s\S]*?)\n {2}\}/) + expect(focus).not.toBeNull() + expect(focus![1]).toContain("this.registerPresence()") + const track = kiloProvider.match(/private trackOpenSessions\(ids: string\[\]\): void \{([\s\S]*?)\n {2}\}/) + expect(track).not.toBeNull() + expect(track![1]).toContain("this.registerPresence()") + }) + + it("registerPresence attaches the open local tabs plus the focused session", () => { + const match = kiloProvider.match(/private registerPresence\(\): void \{([\s\S]*?)\n {2}\}/) + expect(match).not.toBeNull() + const body = match![1] + expect(body).toContain("const attached = new Set(this.openSessionIds)") + expect(body).toContain("if (focused) attached.add(focused)") + }) + + it("the editor-panel view-state callback honors the same option", () => { + const match = kiloProvider.match( + /this\.viewStateDisposable = this\.visibleTaskStreams\.bindPanel\(panel, \(\) => \{([\s\S]*?)\n {4}\}\)/, + ) + expect(match).not.toBeNull() + const guard = match![1].indexOf("if (this.opts.disableViewedRegistration) return") + const visible = match![1].indexOf("registerVisible") + expect(guard).toBeGreaterThanOrEqual(0) + expect(visible).toBeGreaterThan(guard) + }) + + it("embedded Agent Manager providers disable generic viewed registration", () => { + // Each Agent Manager panel hosts a full KiloProvider; the "agent-manager" + // keys own presence there, so the embedded provider must not + // double-register under its own instanceId. + expect(vscodeHost).toContain("disableViewedRegistration: true") + }) +}) + +describe("KiloConnectionService connection snapshot contract", () => { + const source = readFile(CONNECTION_SERVICE_FILE) + + it("sends the accumulated snapshot on initial connection and reconnect", () => { + const start = source.indexOf('if (sseState === "connected")') + const end = source.indexOf('if (!didConnect && sseState === "disconnected")', start) + expect(start).toBeGreaterThan(-1) + expect(end).toBeGreaterThan(start) + const body = source.slice(start, end) + expect(body).toContain("this.flushViewed()") + expect(body).not.toContain("if (isReconnect)") + }) +}) + +describe("AgentManagerProvider visible-presence contract", () => { + const source = readFile(AGENT_MANAGER_PROVIDER_FILE) + + it("routes all agent-manager visible registration through AgentManagerVisiblePresence", () => { + // Exactly one direct registerVisible("agent-manager", ...) call site — the + // presence callback. Cleanup paths that bypassed it (registering [] without + // clearing the displayed id) let a stale id re-register on the next flush. + const sites = source.match(/registerVisible\("agent-manager"/g) ?? [] + expect(sites).toHaveLength(1) + expect(source).toMatch( + /new AgentManagerVisiblePresence\(\s*\(ids\) => this\.connectionService\.registerVisible\("agent-manager", ids\)/, + ) + }) + + it("async shutdown clears both the visible and attached registrations", () => { + // clear() resets the displayed id and empties the attached set, so a + // stale id cannot re-register on a later flush. + const match = source.match(/private async disposeAsync\(\): Promise \{([\s\S]*?)\n {2}\}/) + expect(match).not.toBeNull() + expect(match![1]).toContain("this.visiblePresence.clear()") + }) + + it("routes the webview presence messages to visiblePresence.handle", () => { + // The webview reports the open tab set (→ attached) and the actually + // displayed real session id (null for terminal/review/pending/empty + // tabs, → visible); both flow through the presence helper. + expect(source).toMatch( + /if \(m\.type === "agentManager\.openSessions" \|\| m\.type === "agentManager\.visibleSession"\) \{\s*this\.visiblePresence\.handle\(m\)/, + ) + }) + + it("does not let background message loads override webview visibility", () => { + const match = source.match(/if \(m\.type === "loadMessages"\) \{([\s\S]*?)\n {4}\}/) + expect(match).not.toBeNull() + expect(match![1]).not.toContain("visiblePresence.setDisplayed") + }) + + it("recomputes visible presence when panel visibility changes", () => { + // A hidden Agent Manager panel must drop its session from visible (while + // keeping it attached); reappearing must re-register the retained id. + const match = source.match(/ctx\.onDidChangeVisibility\(\(visible\) => \{([\s\S]*?)\n {4}\}\)/) + expect(match).not.toBeNull() + expect(match![1]).toContain("this.visiblePresence.flush()") + }) +}) diff --git a/packages/kilo-vscode/tests/unit/prompt-send-contract.test.ts b/packages/kilo-vscode/tests/unit/prompt-send-contract.test.ts index 53fbad2134..3d46126d76 100644 --- a/packages/kilo-vscode/tests/unit/prompt-send-contract.test.ts +++ b/packages/kilo-vscode/tests/unit/prompt-send-contract.test.ts @@ -194,9 +194,9 @@ describe("KiloProvider pruneDeletedSession contract", () => { }) it("unfocuses the streams when the deleted id matches the focused session", () => { - // Without this, connectionService.focused still reports the deleted id to - // the backend (viewed.focused), and focusSession() never calls - // unregisterFocused for this instance. + // Without this, connectionService still reports the deleted id to the + // backend as visible, and focusSession() never clears the visible + // registration for this instance. const match = source.match(/pruneDeletedSession\(sessionID: string\): void \{([\s\S]*?)\n \}/) expect(match).not.toBeNull() expect(match![1]).toMatch(/if \(this\.streams\.focused === sessionID\) this\.focusSession\(undefined\)/) @@ -542,15 +542,15 @@ describe("Cloud import parts cleanup contract", () => { describe("KiloConnectionService pruneSession contract", () => { const source = readFile(CONNECTION_SERVICE_FILE) - it("drops the deleted session from focused and opened Maps", () => { + it("drops the deleted session from attached and visible Maps", () => { // KiloProvider's pruneDeletedSession calls connectionService.pruneSession. - // Without clearing focused/opened entries whose value is the deleted id, - // the backend keeps receiving viewed.focused with the dead session id and - // any background tab opener stays registered for it. + // Without clearing attached/visible entries whose value is the deleted id, + // the backend keeps receiving the dead session id and any background tab + // opener stays registered for it. const match = source.match(/pruneSession\(sessionId: string\): void \{([\s\S]*?)\n \}/) expect(match).not.toBeNull() - expect(match![1]).toMatch(/this\.focused\.delete\(key\)/) - expect(match![1]).toMatch(/this\.opened\.(?:set|delete)/) + expect(match![1]).toMatch(/this\.attached\.(?:set|delete)/) + expect(match![1]).toMatch(/this\.visible\.(?:set|delete)/) expect(match![1]).toMatch(/this\.flushViewed\(\)/) }) }) diff --git a/packages/kilo-vscode/webview-ui/agent-manager/AgentManagerApp.tsx b/packages/kilo-vscode/webview-ui/agent-manager/AgentManagerApp.tsx index 467cb4fe3e..51efb0d51d 100644 --- a/packages/kilo-vscode/webview-ui/agent-manager/AgentManagerApp.tsx +++ b/packages/kilo-vscode/webview-ui/agent-manager/AgentManagerApp.tsx @@ -111,7 +111,7 @@ import { } from "../src/utils/draft-store" import { reorderTabs, applyTabOrder, firstOrderedTitle } from "./tab-order" import { createTabOrderSync } from "./tab-order-sync" -import { reportRemoteSessions } from "./remote-sessions" +import { reportRemoteSessions, reportVisibleSession, visible } from "./remote-sessions" import { ConstrainDragYAxis } from "../src/components/chat/TabDnd" import { isTerminalTabId, createTerminalState, createTerminalHandlers, createTerminalMessageHandler } from "./terminal" import { focusCurrentTab, renderTab, renderTerminalLayer, renderNewTabButton } from "./tab-rendering" @@ -768,6 +768,21 @@ const AgentManagerContent: Component = () => { return false }) + const overlay = createMemo((): SetupState | null => { + const state = setup() + const sel = selection() + if (state.active && (!state.worktreeId || sel === state.worktreeId)) return state + if (typeof sel !== "string" || sel === LOCAL) return null + const busy = busyWorktrees().get(sel) + if (busy?.reason !== "setting-up") return null + const tree = worktrees().find((item) => item.id === sel) + return { + active: true, + message: busy.message ?? "", + branch: busy.branch ?? tree?.branch, + } + }) + createEffect(() => { const sel = selection() if (sel === null) { @@ -796,7 +811,13 @@ const AgentManagerContent: Component = () => { if (reviewActive()) return REVIEW_TAB_ID return session.currentSessionID() ?? activePendingId() }) - + const visibleSession = createMemo(() => + visible( + session.currentSessionID(), + !!terms.activeId() || reviewActive() || history() || !!overlay() || contextEmpty(), + ), + ) + reportVisibleSession(vscode, visibleSession) const worktreeLabel = (wt: WorktreeState): string => { if (wt.label) return wt.label return firstOrderedTitle(sessionsForWorktree(wt.id), worktreeTabOrder()[wt.id], wt.branch) @@ -2914,54 +2935,29 @@ const AgentManagerContent: Component = () => { - {(() => { - // Show setup overlay: either the transient ready/error state for the selected worktree, - // or if the selected worktree is still being set up (from busyWorktrees map) - const overlayState = (): SetupState | null => { - const s = setup() - const sel = selection() - // Transient ready/error overlay for the selected worktree (or worktree-less setup) - if (s.active && (!s.worktreeId || sel === s.worktreeId)) return s - // Persistent setup-in-progress for the currently selected worktree - if (typeof sel === "string" && sel !== LOCAL) { - const busy = busyWorktrees().get(sel) - if (busy?.reason === "setting-up") { - const wt = worktrees().find((w) => w.id === sel) - return { - active: true, - message: busy.message ?? "", - branch: busy.branch ?? wt?.branch, - } satisfies SetupState - } - } - return null - } - return ( - - {(state) => ( -
-
- -
- {state().error ? t("agentManager.setup.failed") : t("agentManager.setup.settingUp")} -
- -
{state().branch}
-
-
- }> - - - - {state().errorCode ? t(`agentManager.setup.error.${state().errorCode}`) : state().message} - -
-
+ + {(state) => ( +
+
+ +
+ {state().error ? t("agentManager.setup.failed") : t("agentManager.setup.settingUp")}
- )} - - ) - })()} + +
{state().branch}
+
+
+ }> + + + + {state().errorCode ? t(`agentManager.setup.error.${state().errorCode}`) : state().message} + +
+
+
+ )} +
{ diff --git a/packages/kilo-vscode/webview-ui/agent-manager/remote-sessions.ts b/packages/kilo-vscode/webview-ui/agent-manager/remote-sessions.ts index cfe5d5cf2c..8dbae4d666 100644 --- a/packages/kilo-vscode/webview-ui/agent-manager/remote-sessions.ts +++ b/packages/kilo-vscode/webview-ui/agent-manager/remote-sessions.ts @@ -5,8 +5,17 @@ type Bridge = { postMessage(message: { type: "agentManager.openSessions"; sessionIDs: string[] }): void } +type VisibleBridge = { + postMessage(message: { type: "agentManager.visibleSession"; sessionID: string | null }): void +} + type Managed = { id: string; worktreeId: string | null } +export function visible(id: string | undefined, blocked: boolean): string | null { + if (blocked || !id?.startsWith("ses")) return null + return id +} + export function reportRemoteSessions( vscode: Bridge, local: Accessor, @@ -20,3 +29,12 @@ export function reportRemoteSessions( }) }) } + +// Report the actually displayed real session id, or null when a terminal, +// review, pending, or empty tab is shown. Drives only visible presence; +// retained attached tabs are unaffected. +export function reportVisibleSession(vscode: VisibleBridge, visible: Accessor): void { + createEffect(() => { + vscode.postMessage({ type: "agentManager.visibleSession", sessionID: visible() }) + }) +} diff --git a/packages/kilo-vscode/webview-ui/src/types/messages/webview-messages.ts b/packages/kilo-vscode/webview-ui/src/types/messages/webview-messages.ts index 326f271bd3..0dfb373ff4 100644 --- a/packages/kilo-vscode/webview-ui/src/types/messages/webview-messages.ts +++ b/packages/kilo-vscode/webview-ui/src/types/messages/webview-messages.ts @@ -980,6 +980,11 @@ export interface SidebarOpenSessionsMessage { sessionIDs: string[] } +export interface AgentManagerVisibleSessionMessage { + type: "agentManager.visibleSession" + sessionID: string | null +} + export interface RequestAutoApproveStateMessage { type: "requestAutoApproveState" } @@ -1366,6 +1371,7 @@ export type WebviewMessage = | SetDefaultBaseBranchRequest | AgentManagerOpenSessionsMessage | SidebarOpenSessionsMessage + | AgentManagerVisibleSessionMessage | RequestAutoApproveStateMessage | ToggleAutoApproveMessage | RequestSandboxStatusMessage diff --git a/packages/opencode/src/kilo-sessions/kilo-sessions.ts b/packages/opencode/src/kilo-sessions/kilo-sessions.ts index 277b1eaaec..016e5b9bc1 100644 --- a/packages/opencode/src/kilo-sessions/kilo-sessions.ts +++ b/packages/opencode/src/kilo-sessions/kilo-sessions.ts @@ -38,6 +38,12 @@ async function provide(input: { directory: string; fn: () => R }): Promise return provide(input) } +function same(a: Set, b: Set): boolean { + if (a.size !== b.size) return false + for (const id of a) if (!b.has(id)) return false + return true +} + export namespace KiloSessions { export const Event = { RemoteStatusChanged: BusEvent.define( @@ -203,8 +209,7 @@ export namespace KiloSessions { let remote: { conn: RemoteWS.Connection; sender: RemoteSender.Sender } | undefined let enabling: Promise | undefined let remoteSeq = 0 - const focused = new Set() - const opened = new Set() + const attached = new Set() const statusSyncs = new Map() const STATUS_TIMEOUT_MS = 3_000 @@ -416,8 +421,7 @@ export namespace KiloSessions { const statusMap = await AppRuntime.runPromise(SessionStatus.Service.use((svc) => svc.list())) const statuses: Record = Object.fromEntries(statusMap) const ids = new Set(Object.keys(statuses)) - for (const id of focused) ids.add(id) - for (const id of opened) ids.add(id) + for (const id of attached) ids.add(id) const results = await AppRuntime.runPromise( Session.Service.use((svc) => Effect.all( @@ -438,11 +442,7 @@ export namespace KiloSessions { ), ) const sessions = results.filter((r): r is NonNullable => !!r) - return { - sessions, - focused: focused.size > 0 ? [...focused] : undefined, - open: opened.size > 0 ? [...opened] : undefined, - } + return { sessions } } const conn = RemoteWS.connect({ @@ -515,15 +515,11 @@ export namespace KiloSessions { connected: remote?.conn.connected ?? false, } } - export function setViewedSessions(input: { focused: readonly string[]; open?: readonly string[] }) { - focused.clear() - opened.clear() - for (const id of input.focused) { - focused.add(id) - } - for (const id of input.open ?? []) { - opened.add(id) - } + export function setAttachedSessions(ids: readonly string[]) { + const next = new Set(ids) + if (same(next, attached)) return + attached.clear() + for (const id of next) attached.add(id) if (remote) void remote.conn.heartbeat().catch((err) => log.warn("heartbeat failed", { error: String(err) })) } diff --git a/packages/opencode/src/kilo-sessions/remote-protocol.ts b/packages/opencode/src/kilo-sessions/remote-protocol.ts index 6077e86844..d7280b9882 100644 --- a/packages/opencode/src/kilo-sessions/remote-protocol.ts +++ b/packages/opencode/src/kilo-sessions/remote-protocol.ts @@ -18,8 +18,6 @@ export namespace RemoteProtocol { export const Heartbeat = z.object({ type: z.literal("heartbeat"), sessions: z.array(SessionInfo), - focused: z.array(z.string()).optional(), - open: z.array(z.string()).optional(), protocolVersion: z.string().optional(), // lets relay detect CLI capabilities without probing commands }) export type Heartbeat = z.infer diff --git a/packages/opencode/src/kilo-sessions/remote-ws.ts b/packages/opencode/src/kilo-sessions/remote-ws.ts index 32c6b6688e..d755ea27f0 100644 --- a/packages/opencode/src/kilo-sessions/remote-ws.ts +++ b/packages/opencode/src/kilo-sessions/remote-ws.ts @@ -7,7 +7,7 @@ export namespace RemoteWS { export type Options = { url: string getToken: () => Promise - getSessions: () => Promise<{ sessions: SessionInfo[]; focused?: string[]; open?: string[] }> + getSessions: () => Promise<{ sessions: SessionInfo[] }> log: { info: (...args: any[]) => void error: (...args: any[]) => void diff --git a/packages/opencode/src/kilocode/claw/client.ts b/packages/opencode/src/kilocode/claw/client.ts index 06f63e8e33..14a9586208 100644 --- a/packages/opencode/src/kilocode/claw/client.ts +++ b/packages/opencode/src/kilocode/claw/client.ts @@ -18,6 +18,7 @@ import type { ChatToken, ContentBlock, ConversationActivityEvent, + ConversationLeftEvent, ConversationListItem, ConversationRenamedEvent, ConversationStatusEvent, @@ -29,7 +30,7 @@ import type { TypingMember, } from "./types" import { KiloChatClient } from "./kilo-chat-client" -import { EventServiceClient } from "./event-service-client" +import { EventServiceClient } from "@/kilocode/event-service/client" import * as Log from "@opencode-ai/core/util/log" const log = Log.create({ service: "claw-chat" }) @@ -282,7 +283,7 @@ export async function connect(input: ConnectInput): Promise { emit(conversationsListeners, conversations) }) - events.on("conversation.left", (ctx, e) => { + events.on("conversation.left", (ctx, e: ConversationLeftEvent) => { if (ctx !== sandboxCtx) return conversations = conversations.filter((c) => c.conversationId !== e.conversationId) emit(conversationsListeners, conversations) diff --git a/packages/opencode/src/kilocode/cli/cmd/tui/app.tsx b/packages/opencode/src/kilocode/cli/cmd/tui/app.tsx index 46b45902e2..fefa7985a9 100644 --- a/packages/opencode/src/kilocode/cli/cmd/tui/app.tsx +++ b/packages/opencode/src/kilocode/cli/cmd/tui/app.tsx @@ -5,8 +5,8 @@ * via thin integration points so the upstream diff stays minimal. */ -import { createEffect, on } from "solid-js" -import { useKeyboard } from "@opentui/solid" +import { createEffect, createMemo, on, onCleanup } from "solid-js" +import { useKeyboard, useRenderer } from "@opentui/solid" import { TextAttributes } from "@opentui/core" import * as Clipboard from "@tui/util/clipboard" import { useBindings } from "@tui/keymap" @@ -77,37 +77,72 @@ export function useSessionEffects(deps: { sync: ReturnType }) { const pty = process.env.KILO_PTY_ID - const state = { prev: "" } + const viewerId = crypto.randomUUID() + const renderer = useRenderer() + const session = createMemo(() => (deps.route.data.type === "session" ? deps.route.data.sessionID : undefined)) + let active = true + const meta = { prev: "" } + + function send() { + const id = session() + const ids = id ? [id] : [] + deps.sdk.client.session.viewed({ viewer: { id: viewerId, active }, attached: ids, visible: ids }).catch(() => {}) + } + + createEffect(() => send()) + + const onFocus = () => { + active = true + send() + } + const onBlur = () => { + active = false + send() + } + renderer.on("focus", onFocus) + renderer.on("blur", onBlur) + + // The server prepends `server.connected` to every SSE (re)connect; a restarted + // backend has an empty viewer map, so resend the snapshot immediately instead + // of waiting for the 60s check-in. + const offConnected = deps.sdk.event.on("event", (event) => { + if (event.payload.type === "server.connected") send() + }) + + const timer = setInterval(send, 60_000) - // Notify server which session the user is viewing createEffect(() => { - const sessionID = deps.route.data.type === "session" ? deps.route.data.sessionID : undefined - deps.sdk.client.session.viewed({ focused: sessionID ? [sessionID] : [] }).catch(() => {}) - + const sessionID = session() if (!pty) return - const session = sessionID ? deps.sync.session.get(sessionID) : undefined - const key = [sessionID ?? "", session?.title ?? ""].join("\n") - if (key === state.prev) return - state.prev = key - + const s = sessionID ? deps.sync.session.get(sessionID) : undefined + const key = [sessionID ?? "", s?.title ?? ""].join("\n") + if (key === meta.prev) return + meta.prev = key deps.sdk.client.pty .update({ ptyID: pty, sessionID: sessionID ?? null, - ...(session?.title ? { title: session.title } : {}), + ...(s?.title ? { title: s.title } : {}), }) .catch(() => {}) }) - // Evict per-session data from store when navigating away createEffect( - on( - () => (deps.route.data.type === "session" ? deps.route.data.sessionID : undefined), - (current, prev) => { - if (prev && prev !== current) deps.sync.session.evict(prev) - }, - ), + on(session, (current, prev) => { + if (prev && prev !== current) deps.sync.session.evict(prev) + }), ) + + onCleanup(() => { + renderer.off("focus", onFocus) + renderer.off("blur", onBlur) + offConnected() + clearInterval(timer) + active = false + deps.sdk.client.session + .viewed({ viewer: { id: viewerId, active: false }, attached: [], visible: [] }) + .catch(() => {}) + }) } // --------------------------------------------------------------------------- diff --git a/packages/opencode/src/kilocode/claw/event-service-client.ts b/packages/opencode/src/kilocode/event-service/client.ts similarity index 84% rename from packages/opencode/src/kilocode/claw/event-service-client.ts rename to packages/opencode/src/kilocode/event-service/client.ts index 5bf2b9d8fe..717b75f215 100644 --- a/packages/opencode/src/kilocode/claw/event-service-client.ts +++ b/packages/opencode/src/kilocode/event-service/client.ts @@ -1,9 +1,6 @@ -// kilocode_change - new file - /** - * Event Service WebSocket client for the TUI. + * Generic Event Service WebSocket client. * - * Minimal inline port of `@kilocode/event-service` (cloud monorepo). * Connects via a two-step ticket flow: * 1. POST `/connect-ticket` with `Authorization: Bearer ` to mint a * single-use ticket (30 s TTL). @@ -11,10 +8,15 @@ * `kilo.events.v1`. * * Uses the global `WebSocket` constructor (Bun, Node 22+, browsers). + * + * Disconnect invalidation: every `connect()` and `disconnect()` bumps a + * generation counter. `connectOnce()` captures the generation at entry and, + * after the ticket mint resolves, refuses to construct a socket if the + * generation changed or the client was disposed. `disconnect()` also aborts + * an in-flight ticket request and the pending handshake, so a ticket response + * arriving after disposal can never create a socket. */ -import type { KiloChatEventMap, KiloChatEventName } from "./types" - const WS_SUBPROTOCOL = "kilo.events.v1" const HANDSHAKE_TIMEOUT_MS = 10_000 const PING_INTERVAL_MS = 15_000 @@ -44,12 +46,9 @@ export class HandshakeTimeoutError extends Error { } } -// Close codes that signal the server rejected us for auth/policy reasons -// and reconnecting with the same token is pointless. Everything else -// (including 1006 "abnormal closure" from flaky networks) is transient. function isAuthCloseCode(code: number): boolean { - if (code === 1008) return true // Policy Violation - if (code === 4401 || code === 4403) return true // Custom auth rejection + if (code === 1008) return true + if (code === 4401 || code === 4403) return true return false } @@ -59,13 +58,10 @@ export type EventServiceConfig = { url: string getToken: () => Promise onUnauthorized?: () => void + onServerError?: (error: unknown) => void + handshakeTimeoutMs?: number } -/** - * The event-service base URL is configured as a WebSocket URL (`wss://…` / - * `ws://…`) but the connect-ticket endpoint is a plain HTTP request. Strip - * the trailing slash and swap the protocol so `fetch()` accepts the URL. - */ function toHttpBase(wsBase: string): string { const trimmed = wsBase.replace(/\/$/, "") if (trimmed.startsWith("wss://")) return "https://" + trimmed.slice(6) @@ -77,16 +73,20 @@ export class EventServiceClient { private readonly url: string private readonly getToken: () => Promise private readonly onUnauthorized: (() => void) | undefined + private readonly onServerError: ((error: unknown) => void) | undefined + private readonly handshakeTimeoutMs: number private ws: WebSocket | null = null private connected = false private destroyed = false + private generation = 0 private reconnectAttempts = 0 private hasConnectedBefore = false private reconnectTimer: ReturnType | null = null private pingTimer: ReturnType | null = null private handshakeTimer: ReturnType | null = null private abortHandshake: ((err: Error) => void) | null = null + private tickets = new Set() private eventHandlers = new Map>() private activeContexts = new Set() @@ -96,9 +96,12 @@ export class EventServiceClient { this.url = config.url this.getToken = config.getToken this.onUnauthorized = config.onUnauthorized + this.onServerError = config.onServerError + this.handshakeTimeoutMs = config.handshakeTimeoutMs ?? HANDSHAKE_TIMEOUT_MS } async connect(): Promise { + const gen = ++this.generation this.destroyed = false this.reconnectAttempts = 0 if (this.reconnectTimer !== null) { @@ -108,13 +111,17 @@ export class EventServiceClient { try { await this.connectOnce() } catch (err) { + if (this.destroyed || this.generation !== gen) return if (this.handleAuthFailure(err)) return if (!this.destroyed) this.scheduleReconnect() } } disconnect(): void { + this.generation++ this.destroyed = true + for (const ctrl of this.tickets) ctrl.abort() + this.tickets.clear() if (this.reconnectTimer !== null) { clearTimeout(this.reconnectTimer) this.reconnectTimer = null @@ -149,9 +156,9 @@ export class EventServiceClient { } } - on(event: N, handler: (ctx: string, payload: KiloChatEventMap[N]) => void): () => void { + on(event: string, handler: (context: string, payload: T) => void): () => void { const set = this.eventHandlers.get(event) ?? new Set() - const wrapped: EventHandler = (ctx, payload) => handler(ctx, payload as KiloChatEventMap[N]) + const wrapped: EventHandler = (ctx, payload) => handler(ctx, payload as T) set.add(wrapped) this.eventHandlers.set(event, set) return () => { @@ -181,6 +188,7 @@ export class EventServiceClient { } private async connectOnce(): Promise { + const gen = this.generation if (this.ws) { const old = this.ws this.ws = null @@ -188,7 +196,9 @@ export class EventServiceClient { } const token = await this.getToken() + if (this.destroyed || this.generation !== gen) return const ticket = await this.fetchTicket(token) + if (this.destroyed || this.generation !== gen) return return new Promise((resolve, reject) => { const ws = new WebSocket(`${this.url}/connect?ticket=${encodeURIComponent(ticket)}`, [WS_SUBPROTOCOL]) @@ -215,9 +225,10 @@ export class EventServiceClient { this.handshakeTimer = null if (this.ws === ws) ws.close(1000, "handshake-timeout") settleReject(new HandshakeTimeoutError()) - }, HANDSHAKE_TIMEOUT_MS) + }, this.handshakeTimeoutMs) ws.addEventListener("open", () => { + if (this.ws !== ws) return const isReconnect = this.hasConnectedBefore this.connected = true this.hasConnectedBefore = true @@ -231,6 +242,7 @@ export class EventServiceClient { }) ws.addEventListener("message", (event: MessageEvent) => { + if (this.ws !== ws) return this.handleMessage(String(event.data)) }) @@ -240,10 +252,6 @@ export class EventServiceClient { this.connected = false this.stopPing() this.clearHandshakeTimer() - // A handshake failure always fires `close` after `error`, so we - // settle here with a classification based on the close code: - // explicit auth/policy codes → fatal; anything else → transient - // and the caller (`connect`) will schedule a reconnect. if (!wasConnected) { if (isAuthCloseCode(event.code)) { settleReject(new WebSocketAuthError()) @@ -257,23 +265,13 @@ export class EventServiceClient { if (!this.destroyed) this.scheduleReconnect() }) - ws.addEventListener("error", () => { - // Swallowed: the `close` event fires right after and carries the - // close code we need to distinguish auth failures from network - // blips. Settling here loses that context. - }) + ws.addEventListener("error", () => {}) }) } - /** - * Mint a single-use connection ticket. The event-service issues a 30 s ticket - * scoped to the bearer JWT; the WebSocket upgrade then consumes it. - * - * `this.url` is the WebSocket base (`wss://…` or `ws://…`); `fetch()` only - * accepts `http(s)`, so we rewrite the protocol before the HTTP call. - */ private async fetchTicket(token: string): Promise { const ctrl = new AbortController() + this.tickets.add(ctrl) const timer = setTimeout(() => ctrl.abort(), TICKET_FETCH_TIMEOUT_MS) try { const res = await fetch(toHttpBase(this.url) + "/connect-ticket", { @@ -300,6 +298,7 @@ export class EventServiceClient { throw new WebSocketConnectError(`Event-service ticket request failed: ${(err as Error)?.message ?? err}`, 0) } finally { clearTimeout(timer) + this.tickets.delete(ctrl) } } @@ -335,6 +334,7 @@ export class EventServiceClient { } if (m.type === "error") { console.warn("[Kilo] event-service server error", m) + this.onServerError?.(m) } } @@ -370,7 +370,10 @@ export class EventServiceClient { this.reconnectAttempts++ this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null + if (this.destroyed) return + const gen = this.generation this.connectOnce().catch((err) => { + if (this.destroyed || this.generation !== gen) return if (this.handleAuthFailure(err)) return if (!this.destroyed) this.scheduleReconnect() }) diff --git a/packages/opencode/src/kilocode/presence/context.ts b/packages/opencode/src/kilocode/presence/context.ts new file mode 100644 index 0000000000..1c8e209c2f --- /dev/null +++ b/packages/opencode/src/kilocode/presence/context.ts @@ -0,0 +1,34 @@ +// Presence context strings mirror the (private) cloud Event Service context +// scheme. The cloud package is private, so the literals are duplicated here and +// guarded by contract tests; a full drift guard lands later. + +export const CONTEXT_PREFIX = "/presence/" +export const CLI_SESSION_PREFIX = "/presence/cli-session/" + +export type Platform = "cli" | "vscode" + +export function platformContext(platform: Platform): string { + return `${CONTEXT_PREFIX}${platform}` +} + +export function cliSessionContext(sessionId: string): string { + return `${CLI_SESSION_PREFIX}${sessionId}` +} + +// Event Service enforces a 256-char context limit. +export const MAX_CONTEXT_LENGTH = 256 +// CLI_SESSION_PREFIX is 22 chars, so a session id must be <= 234 to keep the +// full context within the 256-char limit. +export const MAX_SESSION_ID_LENGTH = MAX_CONTEXT_LENGTH - CLI_SESSION_PREFIX.length + +// Event Service socket limit is 200 contexts. +export const MAX_CONTEXTS = 200 +// Reserve one slot for the platform context; visible session contexts cap at 199. +export const MAX_VISIBLE_SESSIONS = MAX_CONTEXTS - 1 + +// Per-viewer rejection thresholds: the service rejects oversized snapshots. +export const MAX_ATTACHED_PER_VIEWER = 1000 +export const MAX_VISIBLE_PER_VIEWER = 199 + +// Viewer lease TTL: a viewer expires exactly 120s after its last update. +export const VIEWER_TTL_MS = 120_000 diff --git a/packages/opencode/src/kilocode/presence/policy.ts b/packages/opencode/src/kilocode/presence/policy.ts new file mode 100644 index 0000000000..f426ccc274 --- /dev/null +++ b/packages/opencode/src/kilocode/presence/policy.ts @@ -0,0 +1,151 @@ +import { + MAX_ATTACHED_PER_VIEWER, + MAX_SESSION_ID_LENGTH, + MAX_VISIBLE_PER_VIEWER, + MAX_VISIBLE_SESSIONS, + VIEWER_TTL_MS, + cliSessionContext, + platformContext, + type Platform, +} from "./context" + +export type ViewerSnapshot = { + viewer: { id: string; active: boolean } + attached: readonly string[] + visible: readonly string[] +} + +export type ViewerState = { + id: string + active: boolean + attached: string[] + visible: string[] + lastSeen: number +} + +export type ValidationError = + | { kind: "missing_viewer" } + | { kind: "bad_viewer_id" } + | { kind: "attached_too_many" } + | { kind: "visible_too_many" } + | { kind: "bad_session_id"; id: string } + +const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i + +function validSessionId(sid: unknown): boolean { + return typeof sid === "string" && sid.startsWith("ses") && sid.length <= MAX_SESSION_ID_LENGTH +} + +// Deduplicate preserving first-seen order. +export function dedupe(ids: readonly string[]): string[] { + const seen = new Set() + const out: string[] = [] + for (const id of ids) { + if (seen.has(id)) continue + seen.add(id) + out.push(id) + } + return out +} + +type ValidationResult = + | { ok: true; viewer: { id: string; active: boolean }; attached: string[]; visible: string[] } + | { ok: false; error: ValidationError } + +export function validateSnapshot(input: { + viewer?: { id?: unknown; active?: unknown } + attached?: unknown + visible?: unknown +}): ValidationResult { + const v = input.viewer + if (!v || typeof v !== "object") return { ok: false, error: { kind: "missing_viewer" } } + const id = v.id + if (typeof id !== "string" || !UUID_RE.test(id)) return { ok: false, error: { kind: "bad_viewer_id" } } + + const rawAttached: readonly unknown[] = Array.isArray(input.attached) ? input.attached : [] + const rawVisible: readonly unknown[] = Array.isArray(input.visible) ? input.visible : [] + + // Reject on raw input size first to bound dedupe work. + if (rawAttached.length > MAX_ATTACHED_PER_VIEWER) return { ok: false, error: { kind: "attached_too_many" } } + if (rawVisible.length > MAX_VISIBLE_PER_VIEWER) return { ok: false, error: { kind: "visible_too_many" } } + + for (const sid of rawAttached) { + if (!validSessionId(sid)) { + return { ok: false, error: { kind: "bad_session_id", id: typeof sid === "string" ? sid : String(sid) } } + } + } + for (const sid of rawVisible) { + if (!validSessionId(sid)) { + return { ok: false, error: { kind: "bad_session_id", id: typeof sid === "string" ? sid : String(sid) } } + } + } + + return { + ok: true, + viewer: { id, active: v.active === true }, + attached: dedupe(rawAttached as readonly string[]), + visible: dedupe(rawVisible as readonly string[]), + } +} + +// Union of every viewer's attached ids (retained regardless of active). +export function attachedUnion(viewers: readonly ViewerState[]): string[] { + const all: string[] = [] + for (const v of viewers) all.push(...v.attached) + return dedupe(all) +} + +// Union of ACTIVE viewers' visible ids, deduped and lexicographically capped at +// MAX_VISIBLE_SESSIONS. Returns the retained ids and how many were omitted. +export function visibleUnion(viewers: readonly ViewerState[]): { ids: string[]; omitted: number } { + const all: string[] = [] + for (const v of viewers) { + if (!v.active) continue + all.push(...v.visible) + } + const union = dedupe(all) + union.sort() + const retained = union.slice(0, MAX_VISIBLE_SESSIONS) + return { ids: retained, omitted: union.length - retained.length } +} + +// Viewer ids expired at `now` (now >= lastSeen + VIEWER_TTL_MS). +export function expiredViewerIds(viewers: readonly ViewerState[], now: number): string[] { + const out: string[] = [] + for (const v of viewers) { + if (now >= v.lastSeen + VIEWER_TTL_MS) out.push(v.id) + } + return out +} + +// Earliest upcoming expiry deadline strictly greater than `now`, or undefined. +export function nextExpiryDeadline(viewers: readonly ViewerState[], now: number): number | undefined { + let min: number | undefined + for (const v of viewers) { + const deadline = v.lastSeen + VIEWER_TTL_MS + if (deadline <= now) continue + if (min === undefined || deadline < min) min = deadline + } + return min +} + +// Reconcile desired Event Service contexts: removals first, then additions. +export function reconcileContexts( + prev: ReadonlySet, + next: ReadonlySet, +): { remove: string[]; add: string[] } { + const remove: string[] = [] + const add: string[] = [] + for (const c of prev) if (!next.has(c)) remove.push(c) + for (const c of next) if (!prev.has(c)) add.push(c) + return { remove, add } +} + +// Desired Event Service context set for a platform and the capped visible ids. +// The platform context is published only when at least one viewer is active. +export function desiredContexts(platform: Platform, active: boolean, visibleIds: readonly string[]): Set { + const out = new Set() + if (active) out.add(platformContext(platform)) + for (const id of visibleIds) out.add(cliSessionContext(id)) + return out +} diff --git a/packages/opencode/src/kilocode/presence/service.ts b/packages/opencode/src/kilocode/presence/service.ts new file mode 100644 index 0000000000..fbbfbc5c56 --- /dev/null +++ b/packages/opencode/src/kilocode/presence/service.ts @@ -0,0 +1,224 @@ +import { Auth } from "@/auth" +import { EventServiceClient } from "@/kilocode/event-service/client" +import { KILO_EVENT_SERVICE_URL } from "@kilocode/kilo-gateway" +import * as Log from "@opencode-ai/core/util/log" +import { Context, Effect, Layer } from "effect" +import type { Platform } from "./context" +import { + attachedUnion, + desiredContexts, + dedupe, + expiredViewerIds, + nextExpiryDeadline, + reconcileContexts, + validateSnapshot, + visibleUnion, + type ViewerSnapshot, + type ViewerState, +} from "./policy" + +const log = Log.create({ service: "kilo-viewers" }) + +function inferPlatform(): Platform | undefined { + const p = process.env.KILO_PLATFORM + if (p === "vscode") return "vscode" + if (p === "cli") return "cli" + if (p === undefined || p === "") return "cli" + return undefined +} + +function extract(auth: Auth.Info | undefined): { token: string | undefined; identity: string | undefined } { + const envKey = process.env.KILO_API_KEY?.trim() + if (auth?.type === "api" && auth.key.length > 0) return { token: auth.key, identity: "api" } + if (auth?.type === "oauth" && auth.access.length > 0) + return { token: auth.access, identity: `oauth:${auth.accountId ?? "no-acct"}` } + if (auth?.type === "wellknown" && auth.token.length > 0) return { token: auth.token, identity: "wellknown" } + if (envKey) return { token: envKey, identity: "env" } + return { token: undefined, identity: undefined } +} + +function sameArr(a: readonly string[], b: readonly string[]): boolean { + if (a.length !== b.length) return false + const set = new Set(a) + for (const id of b) if (!set.has(id)) return false + return true +} + +export namespace KiloViewers { + export interface Interface { + readonly update: (snapshot: ViewerSnapshot) => Effect.Effect + readonly invalidateAuth: () => Effect.Effect + } + + export class Service extends Context.Service()("@kilocode/KiloViewers") {} + + export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const auth = yield* Auth.Service + const KiloSessions = (yield* Effect.promise(() => import("@/kilo-sessions/kilo-sessions"))).KiloSessions + + const platform = inferPlatform() + const killSwitch = process.env.KILO_DISABLE_PRESENCE === "1" + // Same endpoint the server envelope hands KiloClaw; KILO_EVENT_SERVICE_URL + // is a presence-specific override on top of the gateway's EVENT_SERVICE_URL. + const url = process.env.KILO_EVENT_SERVICE_URL || KILO_EVENT_SERVICE_URL + + const s = { + viewers: new Map(), + prevAttached: [] as string[], + prevContexts: new Set(), + identity: undefined as string | undefined, + token: undefined as string | undefined, + client: undefined as EventServiceClient | undefined, + timer: null as ReturnType | null, + } + + function presenceEnabled(): boolean { + return !killSwitch && !!url && !!platform && !!s.token + } + + function disconnectClient() { + if (s.client) { + s.client.disconnect() + s.client = undefined + } + s.prevContexts = new Set() + } + + function rebuild() { + log.warn("rebuilding presence connection") + disconnectClient() + apply(Date.now()) + } + + function onServerError(err: unknown) { + const e = err as Record + const code = typeof e.code === "string" ? e.code : typeof e.error === "string" ? e.error : "" + if (code === "too_many_contexts") rebuild() + } + + function pruneExpired(now: number) { + const expired = expiredViewerIds([...s.viewers.values()], now) + for (const id of expired) s.viewers.delete(id) + } + + function pushAttached() { + const union = attachedUnion([...s.viewers.values()]) + if (!sameArr(union, s.prevAttached)) { + s.prevAttached = union + KiloSessions.setAttachedSessions(union) + } + } + + function reconcilePresence() { + if (!presenceEnabled()) { + if (s.client) disconnectClient() + return + } + const active = [...s.viewers.values()].some((v) => v.active) + const { ids, omitted } = visibleUnion([...s.viewers.values()]) + if (omitted > 0) log.warn("omitted visible session contexts", { omitted }) + const desired = desiredContexts(platform as Platform, active, ids) + if (!s.client) { + // Don't hold an idle socket open: connect only once there is a context + // to assert (inactive-only viewers keep attachment but publish nothing). + if (desired.size === 0) return + if (!s.token) return + s.client = new EventServiceClient({ + url: url as string, + getToken: () => Promise.resolve(s.token!), + onUnauthorized: () => disconnectClient(), + onServerError, + }) + s.client.subscribe([...desired]) + s.prevContexts = desired + void s.client.connect().catch((err) => log.warn("presence connect failed", { error: String(err) })) + return + } + if (desired.size === 0) { + disconnectClient() + return + } + const { remove, add } = reconcileContexts(s.prevContexts, desired) + if (remove.length) s.client.unsubscribe(remove) + if (add.length) s.client.subscribe(add) + s.prevContexts = desired + } + + function apply(now: number) { + pruneExpired(now) + pushAttached() + reconcilePresence() + rescheduleExpiry(now) + } + + function rescheduleExpiry(now: number) { + if (s.timer) { + clearTimeout(s.timer) + s.timer = null + } + const deadline = nextExpiryDeadline([...s.viewers.values()], now) + if (deadline === undefined) return + const delay = Math.max(deadline - now, 0) + s.timer = setTimeout(() => { + s.timer = null + apply(Date.now()) + }, delay) + } + + const readAuth = auth.get("kilo").pipe(Effect.orElseSucceed((): Auth.Info | undefined => undefined)) + + const update = Effect.fn("KiloViewers.update")(function* (snapshot: ViewerSnapshot) { + const info = yield* readAuth + const { token, identity } = extract(info) + if (identity !== s.identity) { + disconnectClient() + s.identity = identity + } + s.token = token + + const result = validateSnapshot(snapshot) + if (!result.ok) { + log.warn("rejected viewer snapshot", { error: result.error.kind }) + return + } + s.viewers.set(result.viewer.id, { + id: result.viewer.id, + active: result.viewer.active, + attached: dedupe(result.attached), + visible: dedupe(result.visible), + lastSeen: Date.now(), + }) + apply(Date.now()) + }) + + const invalidateAuth = Effect.fn("KiloViewers.invalidateAuth")(function* () { + disconnectClient() + s.identity = undefined + s.token = undefined + const info = yield* readAuth + const { token, identity } = extract(info) + s.token = token + s.identity = identity + apply(Date.now()) + }) + + yield* Effect.addFinalizer(() => + Effect.sync(() => { + if (s.timer) { + clearTimeout(s.timer) + s.timer = null + } + disconnectClient() + s.viewers.clear() + KiloSessions.setAttachedSessions([]) + }), + ) + + return Service.of({ update, invalidateAuth }) + }), + ) + + export const defaultLayer = layer.pipe(Layer.provide(Auth.defaultLayer)) +} diff --git a/packages/opencode/src/kilocode/server/httpapi/server.ts b/packages/opencode/src/kilocode/server/httpapi/server.ts index 765c9cb640..1e23654ad1 100644 --- a/packages/opencode/src/kilocode/server/httpapi/server.ts +++ b/packages/opencode/src/kilocode/server/httpapi/server.ts @@ -8,6 +8,7 @@ import { fenceLayer } from "@/server/routes/instance/httpapi/middleware/fence" import * as AnacondaDesktop from "@/kilocode/anaconda-desktop/service" import { BackgroundJob } from "@/background/job" +import { KiloViewers } from "@/kilocode/presence/service" // kilocode_change import { agentBuilderHandlers } from "./handlers/agent-builder" import { anacondaDesktopHandlers } from "./handlers/anaconda-desktop" import { backgroundProcessHandlers } from "./handlers/background-process" @@ -64,6 +65,7 @@ export function provideListener(opts?: CorsOptions) { corsVaryFix, fenceLayer, cors, + KiloViewers.defaultLayer, // kilocode_change FetchHttpClient.layer, HttpServer.layerServices, Layer.succeed(CorsConfig)(opts), diff --git a/packages/opencode/src/kilocode/server/provider-auth-lifecycle.ts b/packages/opencode/src/kilocode/server/provider-auth-lifecycle.ts index d3961975e9..be109981b6 100644 --- a/packages/opencode/src/kilocode/server/provider-auth-lifecycle.ts +++ b/packages/opencode/src/kilocode/server/provider-auth-lifecycle.ts @@ -1,5 +1,6 @@ import { InstanceStore } from "@/project/instance-store" import { ModelCache } from "@/provider/model-cache" +import { KiloViewers } from "@/kilocode/presence/service" // kilocode_change import { Effect } from "effect" export const disposeAllInstancesAfterProviderAuthCallback = Effect.fn( @@ -9,6 +10,13 @@ export const disposeAllInstancesAfterProviderAuthCallback = Effect.fn( yield* store.disposeAll() }) +// kilocode_change start - drop the old presence socket; callers invoke this for the "kilo" provider only +export const invalidatePresence = Effect.fn("KiloServer.invalidatePresence")(function* () { + const viewers = yield* KiloViewers.Service + yield* viewers.invalidateAuth() +}) +// kilocode_change end + export const invalidateAfterProviderAuthChange = Effect.fn("KiloServer.invalidateAfterProviderAuthChange")(function* ( providerID: string, ) { diff --git a/packages/opencode/src/server/routes/instance/httpapi/groups/session.ts b/packages/opencode/src/server/routes/instance/httpapi/groups/session.ts index ba0c742a97..02837849be 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/groups/session.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/groups/session.ts @@ -75,9 +75,16 @@ export const PermissionResponsePayload = Schema.Struct({ response: PermissionV1.Reply, }) // kilocode_change start +const PresenceSessionId = Schema.String.check(Schema.isStartsWith("ses"), Schema.isMaxLength(234)).pipe( + Schema.brand("SessionID"), +) export const ViewedPayload = Schema.Struct({ - focused: Schema.optional(Schema.Array(Schema.String)), - open: Schema.optional(Schema.Array(Schema.String)), + viewer: Schema.Struct({ + id: Schema.String.check(Schema.isUUID()), + active: Schema.Boolean, + }), + attached: Schema.Array(PresenceSessionId).check(Schema.isMaxLength(1000)), + visible: Schema.Array(PresenceSessionId).check(Schema.isMaxLength(199)), }) // kilocode_change end @@ -454,6 +461,7 @@ export const SessionApi = HttpApi.make("session") query: WorkspaceRoutingQuery, payload: ViewedPayload, success: described(Schema.Boolean, "Viewed sessions updated"), + error: HttpApiError.BadRequest, }).annotateMerge( OpenApi.annotations({ identifier: "session.viewed", diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/control.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/control.ts index 1146910c3e..cee0ec265f 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/control.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/control.ts @@ -1,5 +1,8 @@ import { Auth } from "@/auth" -import { invalidateAfterProviderAuthChange } from "@/kilocode/server/provider-auth-lifecycle" // kilocode_change +import { + invalidateAfterProviderAuthChange, + invalidatePresence, +} from "@/kilocode/server/provider-auth-lifecycle" // kilocode_change import * as Log from "@opencode-ai/core/util/log" import { Effect } from "effect" import { HttpApiBuilder } from "effect/unstable/httpapi" @@ -16,6 +19,9 @@ export const controlHandlers = HttpApiBuilder.group(RootHttpApi, "control", (han payload: Auth.Info }) { yield* auth.set(ctx.params.providerID, ctx.payload).pipe(Effect.orDie) + // kilocode_change start - drop old presence socket before instance disposal on Kilo auth changes + if (ctx.params.providerID === "kilo") yield* invalidatePresence() + // kilocode_change end yield* invalidateAfterProviderAuthChange(ctx.params.providerID) // kilocode_change return true }) @@ -24,6 +30,9 @@ export const controlHandlers = HttpApiBuilder.group(RootHttpApi, "control", (han params: { providerID: ProviderV2.ID } }) { yield* auth.remove(ctx.params.providerID).pipe(Effect.orDie) + // kilocode_change start - drop old presence socket before instance disposal on Kilo auth changes + if (ctx.params.providerID === "kilo") yield* invalidatePresence() + // kilocode_change end yield* invalidateAfterProviderAuthChange(ctx.params.providerID) // kilocode_change return true }) diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts index e31dd095f9..28b639624e 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts @@ -5,7 +5,10 @@ import { Provider } from "@/provider/provider" import { mapValues, pickBy } from "remeda" // kilocode_change import { ModelCache } from "@/provider/model-cache" // kilocode_change -import { disposeAllInstancesAfterProviderAuthCallback } from "@/kilocode/server/provider-auth-lifecycle" // kilocode_change +import { + disposeAllInstancesAfterProviderAuthCallback, + invalidatePresence, +} from "@/kilocode/server/provider-auth-lifecycle" // kilocode_change import { providerMetadata } from "@/kilocode/provider/metadata" // kilocode_change import { filterPromptTrainingModels } from "@/kilocode/provider/model-filter" // kilocode_change import { overlay as overlayAnacondaDesktop } from "@/kilocode/anaconda-desktop/provider" // kilocode_change @@ -126,6 +129,9 @@ export const providerHandlers = HttpApiBuilder.group(InstanceHttpApi, "provider" code: ctx.payload.code, }), ) + // kilocode_change start - drop old-user presence before instance disposal on Kilo OAuth callback + if (ctx.params.providerID === "kilo") yield* invalidatePresence() + // kilocode_change end yield* disposeAllInstancesAfterProviderAuthCallback() // kilocode_change return true }) diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts index 57c074e077..8f2c202921 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts @@ -2,6 +2,7 @@ import { Image } from "@/image/image" // kilocode_change - classify user image v import { KiloSessionHttpApi } from "@/kilocode/server/httpapi/session-fork" // kilocode_change import { BlockedError as AgentRequirementError } from "@/kilocode/agent-requirements" // kilocode_change import { PermissionV1 } from "@opencode-ai/core/v1/permission" +import { KiloViewers } from "@/kilocode/presence/service" // kilocode_change import { Agent } from "@/agent/agent" import { SessionV1 } from "@opencode-ai/core/v1/session" import { EventV2Bridge } from "@/event-v2-bridge" @@ -62,6 +63,7 @@ export const sessionHandlers = HttpApiBuilder.group(InstanceHttpApi, "session", const todoSvc = yield* Todo.Service const summary = yield* SessionSummary.Service const events = yield* EventV2Bridge.Service + const viewers = yield* KiloViewers.Service // kilocode_change const scope = yield* Scope.Scope const list = Effect.fn("SessionHttpApi.list")(function* (ctx: { query: typeof ListQuery.Type }) { @@ -421,8 +423,7 @@ export const sessionHandlers = HttpApiBuilder.group(InstanceHttpApi, "session", // kilocode_change start const viewed = Effect.fn("SessionHttpApi.viewed")(function* (ctx: { payload: typeof ViewedPayload.Type }) { - const { KiloSessions } = yield* Effect.promise(() => import("@/kilo-sessions/kilo-sessions")) - KiloSessions.setViewedSessions({ focused: ctx.payload.focused ?? [], open: ctx.payload.open ?? [] }) + yield* viewers.update(ctx.payload) return true }) // kilocode_change end diff --git a/packages/opencode/src/server/routes/instance/httpapi/server.ts b/packages/opencode/src/server/routes/instance/httpapi/server.ts index 120f74a940..9e940d2bca 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/server.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/server.ts @@ -38,6 +38,7 @@ import { Provider } from "@/provider/provider" import { PtyTicket } from "@opencode-ai/core/pty/ticket" import { Question } from "@/question" import { Notebook } from "@/kilocode/notebook/service" // kilocode_change +import { KiloViewers } from "@/kilocode/presence/service" // kilocode_change import { Session } from "@/session/session" import { SessionCompaction } from "@/session/compaction" import { LLM } from "@/session/llm" @@ -251,6 +252,7 @@ export function createRoutes( PtyTicket.defaultLayer, Question.defaultLayer, Notebook.defaultLayer, // kilocode_change + KiloViewers.defaultLayer, // kilocode_change Ripgrep.defaultLayer, RuntimeFlags.defaultLayer, Session.defaultLayer, diff --git a/packages/opencode/test/kilocode/event-service/client.test.ts b/packages/opencode/test/kilocode/event-service/client.test.ts new file mode 100644 index 0000000000..cfaac9ae34 --- /dev/null +++ b/packages/opencode/test/kilocode/event-service/client.test.ts @@ -0,0 +1,345 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { EventServiceClient } from "@/kilocode/event-service/client" + +const OriginalWebSocket = globalThis.WebSocket +const OriginalFetch = globalThis.fetch +const OriginalSetTimeout = globalThis.setTimeout +const OriginalClearTimeout = globalThis.clearTimeout + +type WsListener = (event: unknown) => void + +class FakeWebSocket { + static readonly OPEN = 1 + readonly url: string + readonly protocols: string | string[] | undefined + readyState = 0 + readonly sent: string[] = [] + closedWith: { code?: number; reason?: string } | null = null + private readonly listeners = new Map>() + + constructor(url: string, protocols?: string | string[]) { + this.url = url + this.protocols = protocols + sockets.push(this) + } + + addEventListener(type: string, listener: WsListener): void { + const set = this.listeners.get(type) ?? new Set() + set.add(listener) + this.listeners.set(type, set) + } + + send(data: string): void { + this.sent.push(data) + } + + close(code?: number, reason?: string): void { + if (this.readyState === 3) return + this.readyState = 3 + this.closedWith = { code, reason } + } + + emitOpen(): void { + this.readyState = 1 + for (const l of this.listeners.get("open") ?? []) l({}) + } + + emitMessage(data: string): void { + for (const l of this.listeners.get("message") ?? []) l({ data }) + } + + emitClose(code: number, reason = ""): void { + if (this.readyState !== 3) this.readyState = 3 + for (const l of this.listeners.get("close") ?? []) l({ code, reason }) + } + + emitError(): void { + for (const l of this.listeners.get("error") ?? []) l({}) + } +} + +const sockets: FakeWebSocket[] = [] +let client: EventServiceClient | undefined + +function useFakeWebSocket(): void { + Object.defineProperty(globalThis, "WebSocket", { value: FakeWebSocket, configurable: true, writable: true }) +} + +function ticketBody(ticket = "t"): Response { + return new Response(JSON.stringify({ ticket }), { status: 200, headers: { "content-type": "application/json" } }) +} + +function statusBody(status: number): Response { + return new Response("", { status }) +} + +function installFetch(handler: (url: string, init?: RequestInit) => Response | Promise): void { + globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { + const url = typeof input === "string" ? input : new Request(input).url + return Promise.resolve(handler(url, init)) + }) as unknown as typeof globalThis.fetch +} + +function installTimers() { + const callbacks = new Map void>() + const scheduled: { delay: number }[] = [] + let nextId = 1 + globalThis.setTimeout = ((cb: () => void, delay?: number) => { + const id = nextId++ + callbacks.set(id, cb) + scheduled.push({ delay: delay ?? 0 }) + return id + }) as unknown as typeof setTimeout + globalThis.clearTimeout = ((id?: unknown) => { + if (typeof id === "number") callbacks.delete(id) + }) as unknown as typeof clearTimeout + return { + flush() { + const cbs = [...callbacks.values()] + callbacks.clear() + for (const cb of cbs) cb() + }, + size() { + return callbacks.size + }, + scheduled, + } +} + +async function drain(n = 30): Promise { + for (let i = 0; i < n; i++) await Promise.resolve() +} + +afterEach(() => { + client?.disconnect() + client = undefined + sockets.length = 0 + Object.defineProperty(globalThis, "WebSocket", { value: OriginalWebSocket, configurable: true, writable: true }) + globalThis.fetch = OriginalFetch + globalThis.setTimeout = OriginalSetTimeout + globalThis.clearTimeout = OriginalClearTimeout +}) + +describe("EventServiceClient transport", () => { + test("401 ticket response is fatal and fires onUnauthorized", async () => { + useFakeWebSocket() + installFetch(() => statusBody(401)) + let unauthorized = false + client = new EventServiceClient({ + url: "wss://events.test", + getToken: async () => "tok", + onUnauthorized: () => (unauthorized = true), + }) + await client.connect() + expect(unauthorized).toBe(true) + expect(client.isConnected()).toBe(false) + expect(sockets.length).toBe(0) + }) + + test("403 ticket response is fatal and fires onUnauthorized", async () => { + useFakeWebSocket() + installFetch(() => statusBody(403)) + let unauthorized = false + client = new EventServiceClient({ + url: "wss://events.test", + getToken: async () => "tok", + onUnauthorized: () => (unauthorized = true), + }) + await client.connect() + expect(unauthorized).toBe(true) + expect(sockets.length).toBe(0) + }) + + test("handshake timeout closes the socket with handshake-timeout and is transient", async () => { + useFakeWebSocket() + installFetch(() => ticketBody()) + const timers = installTimers() + client = new EventServiceClient({ + url: "wss://events.test", + getToken: async () => "tok", + handshakeTimeoutMs: 40, + }) + void client.connect() + await drain() + expect(sockets.length).toBe(1) + expect(timers.scheduled.some((s) => s.delay === 40)).toBe(true) + timers.flush() + expect(sockets[0].closedWith?.code).toBe(1000) + expect(sockets[0].closedWith?.reason).toBe("handshake-timeout") + await drain() + expect(timers.size()).toBe(1) + }) + + test("transient close code schedules a reconnect that opens a new socket", async () => { + useFakeWebSocket() + installFetch(() => ticketBody()) + const timers = installTimers() + client = new EventServiceClient({ + url: "wss://events.test", + getToken: async () => "tok", + handshakeTimeoutMs: 5000, + }) + const p = client.connect() + await drain() + expect(sockets.length).toBe(1) + sockets[0].emitOpen() + await p + expect(client.isConnected()).toBe(true) + sockets[0].emitClose(1006) + expect(timers.size()).toBe(1) + timers.flush() + await drain() + expect(sockets.length).toBe(2) + sockets[1].emitOpen() + await drain() + expect(client.isConnected()).toBe(true) + }) + + test("reconnect replays active contexts and fires onReconnect", async () => { + useFakeWebSocket() + installFetch(() => ticketBody()) + const timers = installTimers() + client = new EventServiceClient({ + url: "wss://events.test", + getToken: async () => "tok", + handshakeTimeoutMs: 5000, + }) + let reconnectFired = false + client.onReconnect(() => (reconnectFired = true)) + const p = client.connect() + await drain() + sockets[0].emitOpen() + await p + expect(reconnectFired).toBe(false) + client.subscribe(["ctx-1"]) + expect(sockets[0].sent).toContain(JSON.stringify({ type: "context.subscribe", contexts: ["ctx-1"] })) + sockets[0].emitClose(1006) + timers.flush() + await drain() + expect(sockets.length).toBe(2) + sockets[1].emitOpen() + await drain() + expect(reconnectFired).toBe(true) + expect(JSON.parse(sockets[1].sent[0])).toEqual({ type: "context.subscribe", contexts: ["ctx-1"] }) + }) + + test("unsubscribe sends while connected and disconnect closes the socket", async () => { + useFakeWebSocket() + installFetch(() => ticketBody()) + client = new EventServiceClient({ url: "wss://events.test", getToken: async () => "tok" }) + const p = client.connect() + await drain() + sockets[0].emitOpen() + await p + client.subscribe(["ctx-1"]) + client.unsubscribe(["ctx-1"]) + expect(sockets[0].sent).toContain(JSON.stringify({ type: "context.unsubscribe", contexts: ["ctx-1"] })) + client.disconnect() + expect(client.isConnected()).toBe(false) + expect(sockets[0].closedWith).not.toBeNull() + }) + + test("disposal during token lookup does not start a ticket request", async () => { + useFakeWebSocket() + let resolveToken!: (token: string) => void + const token = new Promise((resolve) => (resolveToken = resolve)) + let requests = 0 + installFetch(() => { + requests++ + return ticketBody() + }) + client = new EventServiceClient({ url: "wss://events.test", getToken: () => token }) + + const pending = client.connect() + await drain() + client.disconnect() + resolveToken("tok") + await pending + + expect(requests).toBe(0) + expect(sockets.length).toBe(0) + }) + + test("disconnect aborts ticket requests from concurrent connect attempts", async () => { + useFakeWebSocket() + installTimers() + const signals: AbortSignal[] = [] + installFetch( + (_url, init) => + new Promise((_resolve, reject) => { + const signal = init?.signal + if (!signal) return + signals.push(signal) + signal.addEventListener("abort", () => reject(new DOMException("Aborted", "AbortError"))) + }), + ) + client = new EventServiceClient({ url: "wss://events.test", getToken: async () => "tok" }) + + void client.connect() + await drain() + void client.connect() + await drain() + expect(signals).toHaveLength(2) + + client.disconnect() + await drain() + expect(signals.every((signal) => signal.aborted)).toBe(true) + }) + + test("disposal during ticket minting never creates a socket", async () => { + useFakeWebSocket() + let resolveFetch!: (r: Response) => void + const fetchPromise = new Promise((r) => (resolveFetch = r)) + installFetch(() => fetchPromise) + client = new EventServiceClient({ url: "wss://events.test", getToken: async () => "tok" }) + const p = client.connect() + await drain() + expect(sockets.length).toBe(0) + client.disconnect() + resolveFetch(ticketBody()) + await p + expect(sockets.length).toBe(0) + expect(client.isConnected()).toBe(false) + }) + + test("disposal suppresses a late unauthorized ticket response", async () => { + useFakeWebSocket() + let resolveFetch!: (r: Response) => void + const fetchPromise = new Promise((r) => (resolveFetch = r)) + installFetch(() => fetchPromise) + let unauthorized = false + client = new EventServiceClient({ + url: "wss://events.test", + getToken: async () => "tok", + onUnauthorized: () => (unauthorized = true), + }) + const p = client.connect() + await drain() + client.disconnect() + resolveFetch(statusBody(401)) + await p + expect(unauthorized).toBe(false) + }) + + test("disposal during handshake closes the socket and ignores a late open", async () => { + useFakeWebSocket() + installFetch(() => ticketBody()) + client = new EventServiceClient({ + url: "wss://events.test", + getToken: async () => "tok", + handshakeTimeoutMs: 5000, + }) + const p = client.connect() + await drain() + expect(sockets.length).toBe(1) + expect(sockets[0].readyState).toBe(0) + client.disconnect() + await drain() + sockets[0].emitOpen() + await drain() + expect(client.isConnected()).toBe(false) + expect(sockets[0].closedWith).not.toBeNull() + expect(sockets.length).toBe(1) + await p + }) +}) diff --git a/packages/opencode/test/kilocode/presence/policy.test.ts b/packages/opencode/test/kilocode/presence/policy.test.ts new file mode 100644 index 0000000000..b9d3a6cc2f --- /dev/null +++ b/packages/opencode/test/kilocode/presence/policy.test.ts @@ -0,0 +1,211 @@ +import { describe, expect, test } from "bun:test" +import { + CLI_SESSION_PREFIX, + MAX_CONTEXT_LENGTH, + MAX_SESSION_ID_LENGTH, + cliSessionContext, + platformContext, +} from "../../../src/kilocode/presence/context" +import { + attachedUnion, + dedupe, + desiredContexts, + expiredViewerIds, + nextExpiryDeadline, + reconcileContexts, + validateSnapshot, + visibleUnion, + type ViewerState, +} from "../../../src/kilocode/presence/policy" + +const UUID = "00000000-0000-4000-8000-000000000000" + +describe("presence context builders", () => { + test("platformContext maps a platform to its presence context", () => { + expect(platformContext("cli")).toBe("/presence/cli") + expect(platformContext("vscode")).toBe("/presence/vscode") + }) + + test("cliSessionContext prefixes the session id", () => { + expect(cliSessionContext("ses_123")).toBe("/presence/cli-session/ses_123") + }) + + test("CLI_SESSION_PREFIX is 22 chars and the budget derivation holds", () => { + expect(CLI_SESSION_PREFIX.length).toBe(22) + expect(MAX_CONTEXT_LENGTH - CLI_SESSION_PREFIX.length).toBe(MAX_SESSION_ID_LENGTH) + expect(MAX_SESSION_ID_LENGTH).toBe(234) + }) + + test("session id budget lands exactly on the 256-char context limit", () => { + expect(cliSessionContext("x".repeat(234)).length).toBe(256) + expect(cliSessionContext("x".repeat(235)).length).toBe(257) + }) +}) + +describe("dedupe", () => { + test("preserves first-seen order and drops duplicates", () => { + expect(dedupe(["a", "b", "a", "c", "b"])).toEqual(["a", "b", "c"]) + expect(dedupe([])).toEqual([]) + expect(dedupe(["x", "x"])).toEqual(["x"]) + }) +}) + +describe("validateSnapshot", () => { + test("accepts a well-formed snapshot and dedupes arrays", () => { + const r = validateSnapshot({ + viewer: { id: UUID, active: true }, + attached: ["ses_s1", "ses_s1", "ses_s2"], + visible: ["ses_v1", "ses_v1"], + }) + expect(r).toEqual({ ok: true, viewer: { id: UUID, active: true }, attached: ["ses_s1", "ses_s2"], visible: ["ses_v1"] }) + }) + + test("missing viewer yields missing_viewer", () => { + expect(validateSnapshot({})).toEqual({ ok: false, error: { kind: "missing_viewer" } }) + expect(validateSnapshot({ viewer: undefined })).toEqual({ ok: false, error: { kind: "missing_viewer" } }) + }) + + test("non-UUID viewer id yields bad_viewer_id", () => { + expect(validateSnapshot({ viewer: { id: "not-a-uuid" } })).toEqual({ ok: false, error: { kind: "bad_viewer_id" } }) + expect(validateSnapshot({ viewer: { id: "" } })).toEqual({ ok: false, error: { kind: "bad_viewer_id" } }) + }) + + test("UUID with an invalid variant nibble yields bad_viewer_id", () => { + expect(validateSnapshot({ viewer: { id: "11111111-1111-1111-1111-111111111111" } })).toEqual({ + ok: false, + error: { kind: "bad_viewer_id" }, + }) + }) + + test("session id missing the ses prefix yields bad_session_id", () => { + expect(validateSnapshot({ viewer: { id: UUID }, attached: ["no-prefix"] })).toEqual({ + ok: false, + error: { kind: "bad_session_id", id: "no-prefix" }, + }) + }) + + test("attached over the per-viewer cap yields attached_too_many", () => { + const attached = Array.from({ length: 1001 }, (_, i) => `ses_${i}`) + expect(validateSnapshot({ viewer: { id: UUID }, attached })).toEqual({ + ok: false, + error: { kind: "attached_too_many" }, + }) + }) + + test("visible over the per-viewer cap yields visible_too_many", () => { + const visible = Array.from({ length: 200 }, (_, i) => `ses_${i}`) + expect(validateSnapshot({ viewer: { id: UUID }, visible })).toEqual({ + ok: false, + error: { kind: "visible_too_many" }, + }) + }) + + test("oversized session id yields bad_session_id with the offending id", () => { + const long = "ses_" + "x".repeat(231) + expect(validateSnapshot({ viewer: { id: UUID }, attached: [long] })).toEqual({ + ok: false, + error: { kind: "bad_session_id", id: long }, + }) + }) + + test("active is coerced strictly to a boolean", () => { + const t = validateSnapshot({ viewer: { id: UUID, active: true }, attached: [], visible: [] }) + expect(t.ok && t.viewer.active).toBe(true) + const str = validateSnapshot({ viewer: { id: UUID, active: "true" }, attached: [], visible: [] }) + expect(str.ok && str.viewer.active).toBe(false) + }) + + test("non-array attached and visible coerce to empty arrays", () => { + expect(validateSnapshot({ viewer: { id: UUID }, attached: null, visible: 42 })).toEqual({ + ok: true, + viewer: { id: UUID, active: false }, + attached: [], + visible: [], + }) + }) +}) + +describe("attachedUnion", () => { + test("unions attached across viewers including inactive ones", () => { + const viewers: ViewerState[] = [ + { id: "u1", active: true, attached: ["a", "b"], visible: [], lastSeen: 0 }, + { id: "u2", active: false, attached: ["b", "c"], visible: [], lastSeen: 0 }, + ] + expect(attachedUnion(viewers)).toEqual(["a", "b", "c"]) + }) +}) + +describe("visibleUnion", () => { + test("only active viewers contribute, deduped and capped at 199", () => { + const ids = Array.from({ length: 201 }, (_, i) => `s${String(i).padStart(4, "0")}`) + const viewers: ViewerState[] = [ + { id: "u1", active: true, attached: [], visible: ids, lastSeen: 0 }, + { id: "u2", active: false, attached: [], visible: ["z_hidden"], lastSeen: 0 }, + ] + const r = visibleUnion(viewers) + expect(r.ids.length).toBe(199) + expect(r.omitted).toBe(2) + expect(r.ids).not.toContain("z_hidden") + const sorted = [...ids].sort() + expect(r.ids).toEqual(sorted.slice(0, 199)) + }) +}) + +describe("expiredViewerIds", () => { + test("expires at exactly lastSeen + TTL and not one ms earlier", () => { + const now = 1_000_000 + const viewers: ViewerState[] = [ + { id: "expired", active: true, attached: [], visible: [], lastSeen: now - 120_000 }, + { id: "alive", active: true, attached: [], visible: [], lastSeen: now - 119_999 }, + ] + expect(expiredViewerIds(viewers, now)).toEqual(["expired"]) + }) +}) + +describe("nextExpiryDeadline", () => { + test("returns the earliest future deadline", () => { + const now = 1_000_000 + const viewers: ViewerState[] = [ + { id: "a", active: true, attached: [], visible: [], lastSeen: now - 50_000 }, + { id: "b", active: true, attached: [], visible: [], lastSeen: now - 10_000 }, + ] + expect(nextExpiryDeadline(viewers, now)).toBe(now + 70_000) + }) + + test("returns undefined when all viewers are expired", () => { + const now = 1_000_000 + const viewers: ViewerState[] = [ + { id: "a", active: true, attached: [], visible: [], lastSeen: now - 120_000 }, + ] + expect(nextExpiryDeadline(viewers, now)).toBeUndefined() + }) + + test("returns undefined for no viewers", () => { + expect(nextExpiryDeadline([], 0)).toBeUndefined() + }) +}) + +describe("reconcileContexts", () => { + test("removals are prev minus next, additions are next minus prev", () => { + const prev = new Set(["a", "b", "c"]) + const next = new Set(["b", "c", "d"]) + expect(reconcileContexts(prev, next)).toEqual({ remove: ["a"], add: ["d"] }) + }) +}) + +describe("desiredContexts", () => { + test("active includes platform context plus each visible session context", () => { + const ctx = desiredContexts("cli", true, ["s1", "s2"]) + expect(ctx.size).toBe(3) + expect(ctx.has("/presence/cli")).toBe(true) + expect(ctx.has("/presence/cli-session/s1")).toBe(true) + expect(ctx.has("/presence/cli-session/s2")).toBe(true) + }) + + test("inactive omits platform context but keeps session contexts", () => { + const ctx = desiredContexts("vscode", false, ["s1"]) + expect(ctx.size).toBe(1) + expect(ctx.has("/presence/vscode")).toBe(false) + expect(ctx.has("/presence/cli-session/s1")).toBe(true) + }) +}) diff --git a/packages/opencode/test/kilocode/presence/service-presence.test.ts b/packages/opencode/test/kilocode/presence/service-presence.test.ts new file mode 100644 index 0000000000..2bd0ccfb83 --- /dev/null +++ b/packages/opencode/test/kilocode/presence/service-presence.test.ts @@ -0,0 +1,267 @@ +import { describe, expect, mock, setSystemTime, test } from "bun:test" +import { Effect, Layer } from "effect" +import { Auth } from "@/auth" + +// Each KiloViewers.layer construction reads these env vars (post-refactor), so +// setting them here controls presence wiring per test. +process.env.KILO_EVENT_SERVICE_URL = "wss://test-presence" +process.env.KILO_PLATFORM = "cli" + +const attachedCalls: string[][] = [] + +type Call = { type: "subscribe" | "unsubscribe" | "connect" | "disconnect"; contexts: string[] } + +// Ordering log shared across FakeClient instances, so ordering can be asserted +// across a disconnect on one client and a connect on its replacement. +const sequence: string[] = [] +let clientSeq = 0 + +class FakeClient { + calls: Call[] = [] + id = ++clientSeq + constructor() {} + async connect(): Promise { + this.calls.push({ type: "connect", contexts: [] }) + sequence.push(`connect:${this.id}`) + } + disconnect(): void { + this.calls.push({ type: "disconnect", contexts: [] }) + sequence.push(`disconnect:${this.id}`) + } + subscribe(contexts: string[]): void { + this.calls.push({ type: "subscribe", contexts: [...contexts] }) + } + unsubscribe(contexts: string[]): void { + this.calls.push({ type: "unsubscribe", contexts: [...contexts] }) + } + onReconnect(): () => void { + return () => {} + } +} + +const realSessions = await import("@/kilo-sessions/kilo-sessions") +const realSetAttached = realSessions.KiloSessions.setAttachedSessions +mock.module("@/kilo-sessions/kilo-sessions", () => ({ + ...realSessions, + KiloSessions: { + ...realSessions.KiloSessions, + setAttachedSessions: (ids: readonly string[]) => { + attachedCalls.push([...ids]) + realSetAttached(ids) + }, + }, +})) + +let current = new FakeClient() +mock.module("@/kilocode/event-service/client", () => ({ + EventServiceClient: class { + constructor() { + // The service constructs one client per layer; expose it for assertions. + current = new FakeClient() + } + async connect() { + await current.connect() + } + disconnect() { + current.disconnect() + } + subscribe(c: string[]) { + current.subscribe(c) + } + unsubscribe(c: string[]) { + current.unsubscribe(c) + } + onReconnect() { + return current.onReconnect() + } + }, +})) + +const { KiloViewers } = await import("@/kilocode/presence/service") + +const authLayer = Layer.succeed( + Auth.Service, + Auth.Service.of({ + get: () => Effect.succeed({ type: "api", key: "tok" } as unknown as Auth.Info), + all: () => Effect.succeed({} as never), + set: () => Effect.void, + remove: () => Effect.void, + }), +) + +const layer = KiloViewers.layer.pipe(Layer.provide(authLayer)) + +const uid = "11111111-1111-4111-8111-111111111111" + +function run(body: (viewers: { + update: (s: { + viewer: { id: string; active: boolean } + attached: readonly string[] + visible: readonly string[] + }) => Effect.Effect + invalidateAuth: () => Effect.Effect +}) => Effect.Effect, l: typeof layer = layer) { + return Effect.gen(function* () { + const v = yield* KiloViewers.Service + yield* body(v) + }).pipe(Effect.provide(l), Effect.runPromise) +} + +function subscribeCalls(): string[][] { + return current.calls.filter((c) => c.type === "subscribe").map((c) => c.contexts) +} + +function unsubscribeCalls(): string[][] { + return current.calls.filter((c) => c.type === "unsubscribe").map((c) => c.contexts) +} + +describe("KiloViewers.Service presence contexts", () => { + test("active viewer subscribes platform plus its visible session context", async () => { + attachedCalls.length = 0 + current = new FakeClient() + await run((v) => + v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }), + ) + const subs = subscribeCalls() + expect(subs.length).toBe(1) + expect(subs[0]).toContain("/presence/cli") + expect(subs[0]).toContain("/presence/cli-session/ses_a") + expect(attachedCalls).toEqual([["ses_a"], []]) + }) + + test("inactive viewer opens no presence socket but keeps attachment", async () => { + attachedCalls.length = 0 + current = new FakeClient() + await run((v) => + v.update({ viewer: { id: uid, active: false }, attached: ["ses_a"], visible: ["ses_a"] }), + ) + expect(subscribeCalls().length).toBe(0) + expect(current.calls.some((c) => c.type === "connect")).toBe(false) + expect(attachedCalls).toEqual([["ses_a"], []]) + }) + + test("replacing the visible set unsubscribes old contexts before subscribing new ones", async () => { + attachedCalls.length = 0 + current = new FakeClient() + const olds = Array.from({ length: 199 }, (_, i) => `ses_old_${i}`) + const next = Array.from({ length: 199 }, (_, i) => `ses_new_${i}`) + await run((v) => + Effect.gen(function* () { + yield* v.update({ viewer: { id: uid, active: true }, attached: olds, visible: olds }) + yield* v.update({ viewer: { id: uid, active: true }, attached: next, visible: next }) + }), + ) + const order = current.calls.filter((c) => c.type === "subscribe" || c.type === "unsubscribe") + const firstUnsubIdx = order.findIndex((c) => c.type === "unsubscribe") + const lastSubIdx = order.map((c) => c.type).lastIndexOf("subscribe") + expect(firstUnsubIdx).toBeGreaterThan(-1) + expect(lastSubIdx).toBeGreaterThan(firstUnsubIdx) + const unsub = unsubscribeCalls().at(-1)! + const sub = subscribeCalls().at(-1)! + expect(unsub.length).toBe(199) + expect(sub.length).toBe(199) + expect(sub.every((c) => c.startsWith("/presence/cli-session/ses_new_"))).toBe(true) + expect(unsub.every((c) => c.startsWith("/presence/cli-session/ses_old_"))).toBe(true) + }) + + test("kill switch blocks the presence socket but attached union still reaches KiloSessions", async () => { + attachedCalls.length = 0 + current = new FakeClient() + const prev = process.env.KILO_DISABLE_PRESENCE + process.env.KILO_DISABLE_PRESENCE = "1" + try { + await run((v) => + v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }), + ) + expect(subscribeCalls().length).toBe(0) + expect(current.calls.some((c) => c.type === "connect")).toBe(false) + expect(attachedCalls).toEqual([["ses_a"], []]) + } finally { + if (prev === undefined) delete process.env.KILO_DISABLE_PRESENCE + else process.env.KILO_DISABLE_PRESENCE = prev + } + }) +}) + +const uidB = "22222222-2222-4222-8222-222222222222" + +describe("KiloViewers.Service viewer lifecycle", () => { + test("viewer expires at lastSeen + 120s", async () => { + attachedCalls.length = 0 + current = new FakeClient() + const base = 1_700_000_000_000 + try { + setSystemTime(base) + await run((v) => + Effect.gen(function* () { + yield* v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }) + setSystemTime(base + 119_999) + yield* v.update({ viewer: { id: uidB, active: true }, attached: ["ses_b"], visible: ["ses_b"] }) + setSystemTime(base + 120_000) + yield* v.update({ viewer: { id: uidB, active: true }, attached: ["ses_b"], visible: ["ses_b"] }) + }), + ) + expect(attachedCalls.length).toBe(4) + // One tick before the TTL the first viewer is still present. + expect(attachedCalls[1]).toContain("ses_a") + expect(attachedCalls[1]).toContain("ses_b") + // At exactly lastSeen + TTL it is pruned (boundary inclusive). + expect(attachedCalls[2]).toEqual(["ses_b"]) + expect(attachedCalls[3]).toEqual([]) + } finally { + setSystemTime() + } + }) + + test("account switch disconnects the old client before connecting the new one", async () => { + attachedCalls.length = 0 + sequence.length = 0 + current = new FakeClient() + let authInfo = { type: "api", key: "tok1" } as unknown as Auth.Info + const mutableAuthLayer = Layer.succeed( + Auth.Service, + Auth.Service.of({ + get: () => Effect.sync(() => authInfo), + all: () => Effect.succeed({} as never), + set: () => Effect.void, + remove: () => Effect.void, + }), + ) + let first: FakeClient | undefined + let second: FakeClient | undefined + await run( + (v) => + Effect.gen(function* () { + yield* v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }) + first = current + authInfo = { type: "wellknown", key: "wk", token: "tok2" } as unknown as Auth.Info + yield* v.invalidateAuth() + second = current + yield* v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }) + }), + KiloViewers.layer.pipe(Layer.provide(mutableAuthLayer)), + ) + expect(first).toBeDefined() + expect(second).toBeDefined() + expect(second).not.toBe(first) + const oldConnect = sequence.indexOf(`connect:${first!.id}`) + const oldDisconnect = sequence.indexOf(`disconnect:${first!.id}`) + const newConnect = sequence.indexOf(`connect:${second!.id}`) + expect(oldConnect).toBeGreaterThan(-1) + expect(oldDisconnect).toBeGreaterThan(oldConnect) + expect(newConnect).toBeGreaterThan(oldDisconnect) + }) + + test("scope disposal runs the finalizer", async () => { + attachedCalls.length = 0 + current = new FakeClient() + await run((v) => + v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }), + ) + const types = current.calls.map((c) => c.type) + expect(types).toContain("connect") + expect(types.at(-1)).toBe("disconnect") + expect(types.indexOf("disconnect")).toBeGreaterThan(types.indexOf("connect")) + expect(attachedCalls).toEqual([["ses_a"], []]) + }) +}) diff --git a/packages/opencode/test/kilocode/presence/service.test.ts b/packages/opencode/test/kilocode/presence/service.test.ts new file mode 100644 index 0000000000..02d580e33e --- /dev/null +++ b/packages/opencode/test/kilocode/presence/service.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, mock, test } from "bun:test" +import { Effect, Layer } from "effect" +import { Auth } from "@/auth" + +const attachedCalls: string[][] = [] + +const realSessions = await import("@/kilo-sessions/kilo-sessions") +const realSetAttached = realSessions.KiloSessions.setAttachedSessions +mock.module("@/kilo-sessions/kilo-sessions", () => ({ + ...realSessions, + KiloSessions: { + ...realSessions.KiloSessions, + setAttachedSessions: (ids: readonly string[]) => { + attachedCalls.push([...ids]) + realSetAttached(ids) + }, + }, +})) + +const { KiloViewers } = await import("@/kilocode/presence/service") + +const authLayer = Layer.succeed( + Auth.Service, + Auth.Service.of({ + get: () => Effect.succeed(undefined), + all: () => Effect.succeed({} as never), + set: () => Effect.void, + remove: () => Effect.void, + }), +) + +const layer = KiloViewers.layer.pipe(Layer.provide(authLayer)) + +const uid = "11111111-1111-4111-8111-111111111111" + +function run(body: (viewers: { + update: (s: { + viewer: { id: string; active: boolean } + attached: readonly string[] + visible: readonly string[] + }) => Effect.Effect + invalidateAuth: () => Effect.Effect +}) => Effect.Effect) { + return Effect.gen(function* () { + const v = yield* KiloViewers.Service + yield* body(v) + }).pipe(Effect.provide(layer), Effect.runPromise) +} + +describe("KiloViewers.Service", () => { + test("pushes the attached union to KiloSessions on change", async () => { + attachedCalls.length = 0 + await run((v) => v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] })) + expect(attachedCalls).toEqual([["ses_a"], []]) + }) + + test("does not re-push an unchanged attached union", async () => { + attachedCalls.length = 0 + await run((v) => + Effect.gen(function* () { + yield* v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }) + yield* v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }) + }), + ) + expect(attachedCalls).toEqual([["ses_a"], []]) + }) + + test("unions attached sessions across viewers", async () => { + attachedCalls.length = 0 + await run((v) => + Effect.gen(function* () { + yield* v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }) + yield* v.update({ viewer: { id: "22222222-2222-4222-8222-222222222222", active: false }, attached: ["ses_b"], visible: [] }) + }), + ) + expect(attachedCalls).toEqual([["ses_a"], ["ses_a", "ses_b"], []]) + }) + + test("invalidateAuth does not throw and clears presence state", async () => { + attachedCalls.length = 0 + await run((v) => + Effect.gen(function* () { + yield* v.update({ viewer: { id: uid, active: true }, attached: ["ses_a"], visible: ["ses_a"] }) + yield* v.invalidateAuth() + }), + ) + expect(attachedCalls.length).toBeGreaterThanOrEqual(1) + }) +}) diff --git a/packages/opencode/test/kilocode/server/httpapi-exercise-scenarios.ts b/packages/opencode/test/kilocode/server/httpapi-exercise-scenarios.ts index b772667d1d..8375ebf795 100644 --- a/packages/opencode/test/kilocode/server/httpapi-exercise-scenarios.ts +++ b/packages/opencode/test/kilocode/server/httpapi-exercise-scenarios.ts @@ -687,8 +687,56 @@ export const kiloScenarios: Scenario[] = [ .status(401), http.protected .post("/session/viewed", "session.viewed") - .at((ctx) => ({ path: "/session/viewed", headers: ctx.headers(), body: { focused: [], open: [] } })) + .at((ctx) => ({ + path: "/session/viewed", + headers: ctx.headers(), + body: { + viewer: { id: "11111111-1111-4111-8111-111111111111", active: true }, + attached: [], + visible: [], + }, + })) .json(200, (body) => check(body === true, "session viewed should return true")), + http.protected + .post("/session/viewed", "session.viewed") + .at((ctx) => ({ path: "/session/viewed", headers: ctx.headers(), body: { attached: [], visible: [] } })) + .status(400), + http.protected + .post("/session/viewed", "session.viewed") + .at((ctx) => ({ + path: "/session/viewed", + headers: ctx.headers(), + body: { + viewer: { id: "not-a-uuid", active: true }, + attached: [], + visible: [], + }, + })) + .status(400), + http.protected + .post("/session/viewed", "session.viewed") + .at((ctx) => ({ + path: "/session/viewed", + headers: ctx.headers(), + body: { + viewer: { id: "11111111-1111-4111-8111-111111111111", active: true }, + attached: ["ses_" + "x".repeat(231)], + visible: [], + }, + })) + .status(400), + http.protected + .post("/session/viewed", "session.viewed") + .at((ctx) => ({ + path: "/session/viewed", + headers: ctx.headers(), + body: { + viewer: { id: "11111111-1111-4111-8111-111111111111", active: true }, + attached: Array.from({ length: 1001 }, () => "ses_1"), + visible: [], + }, + })) + .status(400), http.protected .post("/telemetry/capture", "telemetry.capture") .at((ctx) => ({ diff --git a/packages/opencode/test/kilocode/sessions/remote-protocol.test.ts b/packages/opencode/test/kilocode/sessions/remote-protocol.test.ts index 833420debb..f2c8adfa5c 100644 --- a/packages/opencode/test/kilocode/sessions/remote-protocol.test.ts +++ b/packages/opencode/test/kilocode/sessions/remote-protocol.test.ts @@ -29,6 +29,16 @@ describe("RemoteProtocol", () => { } }) + test("heartbeat serializes sessions only", () => { + const msg = { type: "heartbeat", sessions: [{ id: "ses_1", status: "idle", title: "t" }] } + const result = RemoteProtocol.Heartbeat.safeParse(msg) + expect(result.success).toBe(true) + if (result.success) { + expect(result.data).not.toHaveProperty("focused") + expect(result.data).not.toHaveProperty("open") + } + }) + test("valid event parses", () => { const msg = { type: "event", diff --git a/packages/opencode/test/kilocode/tui-session-presence-contract.test.ts b/packages/opencode/test/kilocode/tui-session-presence-contract.test.ts new file mode 100644 index 0000000000..0678f9696e --- /dev/null +++ b/packages/opencode/test/kilocode/tui-session-presence-contract.test.ts @@ -0,0 +1,78 @@ +/** + * Contract test for the TUI presence snapshot in kilocode/cli/cmd/tui/app.tsx. + * + * `useSessionEffects` must run inside a SolidJS owner with the @opentui/solid + * renderer context, so mounting it in a unit test would require mocking the TUI + * framework internals. These source-contract assertions pin the load-bearing + * presence behaviour instead: the snapshot payload shape (route session as both + * attached and visible), focus/blur toggling only `viewer.active`, the 60s + * check-in, the backend-reconnect resend, and the cleanup path (listeners and + * timer removed, reconnect listener unsubscribed, final empty inactive snapshot). + */ + +import { describe, expect, test } from "bun:test" +import fs from "node:fs" +import path from "node:path" + +const APP_FILE = path.resolve(import.meta.dir, "../../src/kilocode/cli/cmd/tui/app.tsx") + +/** The useSessionEffects function body, so assertions don't match unrelated code. */ +function effects() { + const content = fs.readFileSync(APP_FILE, "utf-8") + const start = content.indexOf("export function useSessionEffects") + const end = content.indexOf("export function getTerminalTitle") + expect(start).toBeGreaterThan(-1) + expect(end).toBeGreaterThan(start) + return content.slice(start, end) +} + +/** The onCleanup block of useSessionEffects (it is the last statement of the hook). */ +function cleanup() { + const body = effects() + const start = body.indexOf("onCleanup(() => {") + expect(start).toBeGreaterThan(-1) + return body.slice(start) +} + +/** Collapse whitespace so multi-line expressions match regardless of formatting. */ +function flat(source: string) { + return source.replace(/\s+/g, " ").replace(/\( /g, "(").replace(/ \)/g, ")").replace(/,\)/g, ")") +} + +describe("TUI session presence contract", () => { + test("snapshot sends the route session as both attached and visible", () => { + const body = effects() + expect(body).toContain('deps.route.data.type === "session" ? deps.route.data.sessionID : undefined') + expect(body).toContain("const ids = id ? [id] : []") + expect(flat(body)).toContain( + "deps.sdk.client.session.viewed({ viewer: { id: viewerId, active }, attached: ids, visible: ids }).catch(() => {})", + ) + }) + + test("focus sets active=true, blur sets active=false, both resend the snapshot", () => { + const body = flat(effects()) + expect(body).toContain("const onFocus = () => { active = true send() }") + expect(body).toContain("const onBlur = () => { active = false send() }") + expect(body).toContain('renderer.on("focus", onFocus)') + expect(body).toContain('renderer.on("blur", onBlur)') + }) + + test("60s check-in interval exists and is cleared on cleanup", () => { + expect(effects()).toContain("const timer = setInterval(send, 60_000)") + expect(cleanup()).toContain("clearInterval(timer)") + }) + + test("server.connected resends the snapshot and is unsubscribed on cleanup", () => { + expect(flat(effects())).toContain( + 'const offConnected = deps.sdk.event.on("event", (event) => { if (event.payload.type === "server.connected") send() })', + ) + expect(cleanup()).toContain("offConnected()") + }) + + test("cleanup removes focus/blur listeners and sends a final inactive empty snapshot", () => { + const tail = cleanup() + expect(tail).toContain('renderer.off("focus", onFocus)') + expect(tail).toContain('renderer.off("blur", onBlur)') + expect(flat(tail)).toContain(".viewed({ viewer: { id: viewerId, active: false }, attached: [], visible: [] })") + }) +}) diff --git a/packages/opencode/test/server/httpapi-exercise/environment.ts b/packages/opencode/test/server/httpapi-exercise/environment.ts index 313818f91d..4594f5f329 100644 --- a/packages/opencode/test/server/httpapi-exercise/environment.ts +++ b/packages/opencode/test/server/httpapi-exercise/environment.ts @@ -12,6 +12,7 @@ process.env.XDG_STATE_HOME = path.join(exerciseGlobalRoot, "state") process.env.XDG_CACHE_HOME = path.join(exerciseGlobalRoot, "cache") process.env.KILO_DISABLE_SHARE = "true" process.env.KILO_DISABLE_SESSION_INGEST = "true" // kilocode_change - isolate the exerciser from async Kilo session sync +process.env.KILO_DISABLE_PRESENCE = "1" // kilocode_change - presence now has a default Event Service URL; never open real sockets from the exerciser export const exerciseConfigDirectory = path.join(exerciseGlobalRoot, "config", "opencode") export const exerciseDataDirectory = path.join(exerciseGlobalRoot, "data", "kilo") // kilocode_change diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index 2a06325263..086b49ae22 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -5066,8 +5066,12 @@ export class Session2 extends HeyApiClient { parameters?: { directory?: string workspace?: string - focused?: Array - open?: Array + viewer?: { + id: string + active: boolean + } + attached?: Array + visible?: Array }, options?: Options, ) { @@ -5078,8 +5082,9 @@ export class Session2 extends HeyApiClient { args: [ { in: "query", key: "directory" }, { in: "query", key: "workspace" }, - { in: "body", key: "focused" }, - { in: "body", key: "open" }, + { in: "body", key: "viewer" }, + { in: "body", key: "attached" }, + { in: "body", key: "visible" }, ], }, ], diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 01f3b2e53e..122d5e87e8 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -9804,8 +9804,12 @@ export type PartUpdateResponse = PartUpdateResponses[keyof PartUpdateResponses] export type SessionViewedData = { body?: { - focused?: Array - open?: Array + viewer: { + id: string + active: boolean + } + attached: Array + visible: Array } path?: never query?: { @@ -9817,9 +9821,9 @@ export type SessionViewedData = { export type SessionViewedErrors = { /** - * Bad request + * BadRequest | InvalidRequestError */ - 400: BadRequestError + 400: EffectHttpApiErrorBadRequest | InvalidRequestError } export type SessionViewedError = SessionViewedErrors[keyof SessionViewedErrors] diff --git a/packages/sdk/openapi.json b/packages/sdk/openapi.json index 553f2acb26..f049600ebe 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -8798,11 +8798,18 @@ } }, "400": { - "description": "Bad request", + "description": "BadRequest | InvalidRequestError", "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/BadRequestError" + "anyOf": [ + { + "$ref": "#/components/schemas/effect_HttpApiError_BadRequest" + }, + { + "$ref": "#/components/schemas/InvalidRequestError" + } + ] } } } @@ -8816,19 +8823,41 @@ "schema": { "type": "object", "properties": { - "focused": { - "type": "array", - "items": { - "type": "string" - } + "viewer": { + "type": "object", + "properties": { + "id": { + "type": "string", + "pattern": "^([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-8][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}|00000000-0000-0000-0000-000000000000)$", + "format": "uuid" + }, + "active": { + "type": "boolean" + } + }, + "required": ["id", "active"], + "additionalProperties": false }, - "open": { + "attached": { "type": "array", "items": { - "type": "string" - } + "type": "string", + "pattern": "^ses", + "maxLength": 234 + }, + "maxItems": 1000 + }, + "visible": { + "type": "array", + "items": { + "type": "string", + "pattern": "^ses", + "maxLength": 234 + }, + "maxItems": 199 } }, + "required": ["viewer", "attached", "visible"], "additionalProperties": false } }