From 3b31a5c565b157b709354802f0e6f9d29877c18a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 09:51:51 +0100 Subject: [PATCH 01/35] feat(share): create share sessions with ingest URL Update ShareNext to create sessions via `/api/opencode/session` and persist/use `ingestUrl` for syncing. Automatically create share sessions on `Session.Created` and route session sharing through `ShareNext.share()`. --- packages/opencode/src/session/index.ts | 2 +- packages/opencode/src/share/share-next.ts | 25 ++++++++++++++++------- 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts index b81a21a57b..ba95c07b74 100644 --- a/packages/opencode/src/session/index.ts +++ b/packages/opencode/src/session/index.ts @@ -254,7 +254,7 @@ export namespace Session { throw new Error("Sharing is disabled in configuration") } const { ShareNext } = await import("@/share/share-next") - const share = await ShareNext.create(id) + const share = await ShareNext.share(id) await update( id, (draft) => { diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index dddce95cb4..61f81a98ca 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -12,13 +12,16 @@ export namespace ShareNext { const log = Log.create({ service: "share-next" }) async function url() { - return Config.get().then((x) => x.enterprise?.url ?? "https://opncd.ai") + return Config.get().then((x) => x.enterprise?.url ?? "http://localhost:3000") } const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1" export async function init() { if (disabled) return + Bus.subscribe(Session.Event.Created, async (evt) => { + await create(evt.properties.info.id) + }) Bus.subscribe(Session.Event.Updated, async (evt) => { await sync(evt.properties.info.id, [ { @@ -66,9 +69,9 @@ export namespace ShareNext { } export async function create(sessionID: string) { - if (disabled) return { id: "", url: "", secret: "" } - log.info("creating share", { sessionID }) - const result = await fetch(`${await url()}/api/share`, { + if (disabled) return { id: "", ingestUrl: "", secret: "" } + log.info("creating session", { sessionID }) + const result = await fetch(`${await url()}/api/opencode/session`, { method: "POST", headers: { "Content-Type": "application/json", @@ -76,17 +79,25 @@ export namespace ShareNext { body: JSON.stringify({ sessionID: sessionID }), }) .then((x) => x.json()) - .then((x) => x as { id: string; url: string; secret: string }) + .then((x) => x as { id: string; ingestUrl: string; secret: string }) await Storage.write(["session_share", sessionID], result) fullSync(sessionID) return result } + export async function share(sessionID: string) { + if (disabled) return { url: "" } + log.info("creating share", { sessionID }) + + return { url: "" } + } + function get(sessionID: string) { return Storage.read<{ id: string secret: string - url: string + url?: string + ingestUrl: string }>(["session_share", sessionID]) } @@ -135,7 +146,7 @@ export namespace ShareNext { const share = await get(sessionID).catch(() => undefined) if (!share) return - await fetch(`${await url()}/api/share/${share.id}/sync`, { + await fetch(share.ingestUrl, { method: "POST", headers: { "Content-Type": "application/json", From 6233c0318647dd6a1893b6e94be78a84ccf8a196 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 12:56:27 +0100 Subject: [PATCH 02/35] feat(share): align share-next session API usage Update default enterprise URL port, switch session creation to `/api/session`, and drop secret handling from stored share data and sync payloads. --- packages/opencode/src/share/share-next.ts | 77 +++++++++++++---------- 1 file changed, 43 insertions(+), 34 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 61f81a98ca..a28a877fa3 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -12,7 +12,7 @@ export namespace ShareNext { const log = Log.create({ service: "share-next" }) async function url() { - return Config.get().then((x) => x.enterprise?.url ?? "http://localhost:3000") + return Config.get().then((x) => x.enterprise?.url ?? "http://localhost:8787") } const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1" @@ -68,37 +68,42 @@ export namespace ShareNext { }) } - export async function create(sessionID: string) { - if (disabled) return { id: "", ingestUrl: "", secret: "" } - log.info("creating session", { sessionID }) - const result = await fetch(`${await url()}/api/opencode/session`, { + export async function create(sessionId: string) { + if (disabled) return { id: "", ingestUrl: "" } + + log.info("creating session", { sessionId }) + + const result = await fetch(`${await url()}/api/session`, { method: "POST", headers: { "Content-Type": "application/json", }, - body: JSON.stringify({ sessionID: sessionID }), + body: JSON.stringify({ sessionId }), }) .then((x) => x.json()) - .then((x) => x as { id: string; ingestUrl: string; secret: string }) - await Storage.write(["session_share", sessionID], result) - fullSync(sessionID) + .then((x) => x as { id: string; ingestUrl: string }) + + await Storage.write(["session_share", sessionId], result) + + fullSync(sessionId) + return result } - export async function share(sessionID: string) { + export async function share(sessionId: string) { if (disabled) return { url: "" } - log.info("creating share", { sessionID }) + + log.info("creating share", { sessionId }) return { url: "" } } - function get(sessionID: string) { + function get(sessionId: string) { return Storage.read<{ id: string - secret: string url?: string ingestUrl: string - }>(["session_share", sessionID]) + }>(["session_share", sessionId]) } type Data = @@ -124,9 +129,9 @@ export namespace ShareNext { } const queue = new Map }>() - async function sync(sessionID: string, data: Data[]) { + async function sync(sessionId: string, data: Data[]) { if (disabled) return - const existing = queue.get(sessionID) + const existing = queue.get(sessionId) if (existing) { for (const item of data) { existing.data.set("id" in item ? (item.id as string) : ulid(), item) @@ -140,10 +145,12 @@ export namespace ShareNext { } const timeout = setTimeout(async () => { - const queued = queue.get(sessionID) + const queued = queue.get(sessionId) if (!queued) return - queue.delete(sessionID) - const share = await get(sessionID).catch(() => undefined) + + queue.delete(sessionId) + + const share = await get(sessionId).catch(() => undefined) if (!share) return await fetch(share.ingestUrl, { @@ -152,43 +159,45 @@ export namespace ShareNext { "Content-Type": "application/json", }, body: JSON.stringify({ - secret: share.secret, data: Array.from(queued.data.values()), }), }) }, 1000) - queue.set(sessionID, { timeout, data: dataMap }) + queue.set(sessionId, { timeout, data: dataMap }) } - export async function remove(sessionID: string) { + export async function remove(sessionId: string) { if (disabled) return - log.info("removing share", { sessionID }) - const share = await get(sessionID) + + log.info("removing share", { sessionId }) + + const share = await get(sessionId) if (!share) return + await fetch(`${await url()}/api/share/${share.id}`, { method: "DELETE", headers: { "Content-Type": "application/json", }, - body: JSON.stringify({ - secret: share.secret, - }), }) - await Storage.remove(["session_share", sessionID]) + + await Storage.remove(["session_share", sessionId]) } - async function fullSync(sessionID: string) { - log.info("full sync", { sessionID }) - const session = await Session.get(sessionID) - const diffs = await Session.diff(sessionID) - const messages = await Array.fromAsync(MessageV2.stream(sessionID)) + async function fullSync(sessionId: string) { + log.info("full sync", { sessionId }) + + const session = await Session.get(sessionId) + const diffs = await Session.diff(sessionId) + const messages = await Array.fromAsync(MessageV2.stream(sessionId)) const models = await Promise.all( messages .filter((m) => m.info.role === "user") .map((m) => (m.info as SDK.UserMessage).model) .map((m) => Provider.getModel(m.providerID, m.modelID).then((m) => m)), ) - await sync(sessionID, [ + + await sync(sessionId, [ { type: "session", data: session, From 7a6281aca988d914284ec8a5e4fddff3dfc7132d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 13:22:21 +0100 Subject: [PATCH 03/35] feat(share): authenticate share-next requests with kilocode token Add a reusable client that injects bearer auth headers using the Kilo provider configuration (kilocodeToken or apiKey fallback), and gate share init/create/sync/remove when sharing is disabled or no token is present. --- packages/opencode/src/share/share-next.ts | 88 ++++++++++++++++++----- 1 file changed, 69 insertions(+), 19 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index a28a877fa3..bc4b1a1aa2 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -15,13 +15,60 @@ export namespace ShareNext { return Config.get().then((x) => x.enterprise?.url ?? "http://localhost:8787") } + export async function kilocodeConfig() { + return Config.get().then((x) => x.provider?.["kilo"]) + } + + export async function kilocodeToken() { + const cfg = await kilocodeConfig() + const token = cfg?.options?.kilocodeToken + if (typeof token === "string" && token.length > 0) return token + const apiKey = cfg?.options?.apiKey + if (typeof apiKey === "string" && apiKey.length > 0) return apiKey + return undefined + } + + type Client = { + url: string + fetch: (input: RequestInfo | URL, init?: RequestInit) => Promise + } + + async function getClient(): Promise { + if (disabled) return undefined + const token = await kilocodeToken() + if (!token) return undefined + + const base = await url() + const baseHeaders: Record = { + "Content-Type": "application/json", + Authorization: `bearer ${token}`, + } + + const withHeaders = (init?: RequestInit) => { + const headers = new Headers(init?.headers) + for (const [k, v] of Object.entries(baseHeaders)) headers.set(k, v) + return { + ...init, + headers, + } satisfies RequestInit + } + + return { + url: base, + fetch: (input, init) => fetch(input, withHeaders(init)), + } + } + const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1" export async function init() { - if (disabled) return + const client = await getClient() + if (!client) return + Bus.subscribe(Session.Event.Created, async (evt) => { await create(evt.properties.info.id) }) + Bus.subscribe(Session.Event.Updated, async (evt) => { await sync(evt.properties.info.id, [ { @@ -30,6 +77,7 @@ export namespace ShareNext { }, ]) }) + Bus.subscribe(MessageV2.Event.Updated, async (evt) => { await sync(evt.properties.info.sessionID, [ { @@ -37,6 +85,7 @@ export namespace ShareNext { data: evt.properties.info, }, ]) + if (evt.properties.info.role === "user") { await sync(evt.properties.info.sessionID, [ { @@ -50,6 +99,7 @@ export namespace ShareNext { ]) } }) + Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => { await sync(evt.properties.part.sessionID, [ { @@ -58,6 +108,7 @@ export namespace ShareNext { }, ]) }) + Bus.subscribe(Session.Event.Diff, async (evt) => { await sync(evt.properties.sessionID, [ { @@ -69,17 +120,16 @@ export namespace ShareNext { } export async function create(sessionId: string) { - if (disabled) return { id: "", ingestUrl: "" } + const client = await getClient() + if (!client) return { id: "", ingestUrl: "" } log.info("creating session", { sessionId }) - const result = await fetch(`${await url()}/api/session`, { - method: "POST", - headers: { - "Content-Type": "application/json", - }, - body: JSON.stringify({ sessionId }), - }) + const result = await client + .fetch(`${client.url}/api/session`, { + method: "POST", + body: JSON.stringify({ sessionId }), + }) .then((x) => x.json()) .then((x) => x as { id: string; ingestUrl: string }) @@ -130,7 +180,9 @@ export namespace ShareNext { const queue = new Map }>() async function sync(sessionId: string, data: Data[]) { - if (disabled) return + const client = await getClient() + if (!client) return + const existing = queue.get(sessionId) if (existing) { for (const item of data) { @@ -153,11 +205,11 @@ export namespace ShareNext { const share = await get(sessionId).catch(() => undefined) if (!share) return - await fetch(share.ingestUrl, { + const client = await getClient() + if (!client) return + + await client.fetch(share.ingestUrl, { method: "POST", - headers: { - "Content-Type": "application/json", - }, body: JSON.stringify({ data: Array.from(queued.data.values()), }), @@ -167,18 +219,16 @@ export namespace ShareNext { } export async function remove(sessionId: string) { - if (disabled) return + const client = await getClient() + if (!client) return log.info("removing share", { sessionId }) const share = await get(sessionId) if (!share) return - await fetch(`${await url()}/api/share/${share.id}`, { + await client.fetch(`${client.url}/api/share/${share.id}`, { method: "DELETE", - headers: { - "Content-Type": "application/json", - }, }) await Storage.remove(["session_share", sessionId]) From 91bdc96896d2ab3940f62fe464b3969eae787c20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 15:52:19 +0100 Subject: [PATCH 04/35] refactor(share): use Auth for token and ingest path Replace kilocode token lookup via provider config with Auth.get("kilo") and switch session share ingestion from ingestUrl to ingestPath. --- packages/opencode/src/share/share-next.ts | 32 ++++++++--------------- 1 file changed, 11 insertions(+), 21 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index bc4b1a1aa2..1c17d3f683 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -1,3 +1,4 @@ +// kilocode_change pretty much completely refactored import { Bus } from "@/bus" import { Config } from "@/config/config" import { ulid } from "ulid" @@ -6,25 +7,17 @@ import { Session } from "@/session" import { MessageV2 } from "@/session/message-v2" import { Storage } from "@/storage/storage" import { Log } from "@/util/log" +import { Auth } from "@/auth" import type * as SDK from "@opencode-ai/sdk/v2" export namespace ShareNext { const log = Log.create({ service: "share-next" }) - async function url() { - return Config.get().then((x) => x.enterprise?.url ?? "http://localhost:8787") - } - - export async function kilocodeConfig() { - return Config.get().then((x) => x.provider?.["kilo"]) - } - export async function kilocodeToken() { - const cfg = await kilocodeConfig() - const token = cfg?.options?.kilocodeToken - if (typeof token === "string" && token.length > 0) return token - const apiKey = cfg?.options?.apiKey - if (typeof apiKey === "string" && apiKey.length > 0) return apiKey + const auth = await Auth.get("kilo") + if (auth?.type === "api" && auth.key.length > 0) return auth.key + if (auth?.type === "oauth" && auth.access.length > 0) return auth.access + if (auth?.type === "wellknown" && auth.token.length > 0) return auth.token return undefined } @@ -38,7 +31,7 @@ export namespace ShareNext { const token = await kilocodeToken() if (!token) return undefined - const base = await url() + const base = await Config.get().then((x) => x.enterprise?.url ?? "http://localhost:8787") const baseHeaders: Record = { "Content-Type": "application/json", Authorization: `bearer ${token}`, @@ -62,9 +55,6 @@ export namespace ShareNext { const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1" export async function init() { - const client = await getClient() - if (!client) return - Bus.subscribe(Session.Event.Created, async (evt) => { await create(evt.properties.info.id) }) @@ -121,7 +111,7 @@ export namespace ShareNext { export async function create(sessionId: string) { const client = await getClient() - if (!client) return { id: "", ingestUrl: "" } + if (!client) return { id: "", ingestPath: "" } log.info("creating session", { sessionId }) @@ -131,7 +121,7 @@ export namespace ShareNext { body: JSON.stringify({ sessionId }), }) .then((x) => x.json()) - .then((x) => x as { id: string; ingestUrl: string }) + .then((x) => x as { id: string; ingestPath: string }) await Storage.write(["session_share", sessionId], result) @@ -152,7 +142,7 @@ export namespace ShareNext { return Storage.read<{ id: string url?: string - ingestUrl: string + ingestPath: string }>(["session_share", sessionId]) } @@ -208,7 +198,7 @@ export namespace ShareNext { const client = await getClient() if (!client) return - await client.fetch(share.ingestUrl, { + await client.fetch(`${client.url}${share.ingestPath}`, { method: "POST", body: JSON.stringify({ data: Array.from(queued.data.values()), From 0dc0b4bec3c85e5e7053850819e294b3e8106be0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 19:58:53 +0100 Subject: [PATCH 05/35] feat(share): create share link via session api Implement share() by posting to the session share endpoint and returning the generated public URL, with an early return when no client is available. --- packages/opencode/src/share/share-next.ts | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 1c17d3f683..d387e8c84f 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -133,9 +133,20 @@ export namespace ShareNext { export async function share(sessionId: string) { if (disabled) return { url: "" } + const client = await getClient() + if (!client) return { url: "" } + log.info("creating share", { sessionId }) - return { url: "" } + const result = await client + .fetch(`${client.url}/api/session/${sessionId}/share`, { + method: "POST", + body: JSON.stringify({ sessionId }), + }) + .then((x) => x.json()) + .then((x) => x as { public_id: string }) + + return { url: `http://localhost:3000/s/${result.public_id}` } } function get(sessionId: string) { From 842d4f2c17bee95dd4411e7bf6adcdd24aa50a37 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 20:01:52 +0100 Subject: [PATCH 06/35] feat(share): persist generated share url to storage --- packages/opencode/src/share/share-next.ts | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index d387e8c84f..aad4319e4e 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -146,7 +146,15 @@ export namespace ShareNext { .then((x) => x.json()) .then((x) => x as { public_id: string }) - return { url: `http://localhost:3000/s/${result.public_id}` } + const current = (await Storage.read(["session_share", sessionId])) as Awaited> + const url = `http://localhost:3000/s/${result.public_id}` + + await Storage.write(["session_share", sessionId], { + ...current, + url, + }) + + return { url } } function get(sessionId: string) { From 5c88b996f1239aa8f4f93bdc424bbc075bd43bf3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 20:40:23 +0100 Subject: [PATCH 07/35] feat(share): add unshare flow via session api Add ShareNext.unshare to call the session unshare endpoint and clear the stored share url, and update Session.unshare to use the new method. --- packages/opencode/src/session/index.ts | 2 +- packages/opencode/src/share/share-next.ts | 22 +++++++++++++++++++++- 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts index ba95c07b74..eef0100717 100644 --- a/packages/opencode/src/session/index.ts +++ b/packages/opencode/src/session/index.ts @@ -270,7 +270,7 @@ export namespace Session { export const unshare = fn(Identifier.schema("session"), async (id) => { // Use ShareNext to remove the share (same as share function uses ShareNext to create) const { ShareNext } = await import("@/share/share-next") - await ShareNext.remove(id) + await ShareNext.unshare(id) await update( id, (draft) => { diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index aad4319e4e..44303f7b29 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -136,7 +136,7 @@ export namespace ShareNext { const client = await getClient() if (!client) return { url: "" } - log.info("creating share", { sessionId }) + log.info("sharing", { sessionId }) const result = await client .fetch(`${client.url}/api/session/${sessionId}/share`, { @@ -157,6 +157,26 @@ export namespace ShareNext { return { url } } + export async function unshare(sessionId: string) { + if (disabled) return + + const client = await getClient() + if (!client) return + + log.info("unsharing", { sessionId }) + + const result = await client.fetch(`${client.url}/api/session/${sessionId}/unshare`, { + method: "POST", + body: JSON.stringify({ sessionId }), + }) + + const current = (await Storage.read(["session_share", sessionId])) as Awaited> + + delete current.url + + await Storage.write(["session_share", sessionId], current) + } + function get(sessionId: string) { return Storage.read<{ id: string From 923cb4597678435998d2afce5ee93857c5b4fe11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Tue, 27 Jan 2026 22:27:05 +0100 Subject: [PATCH 08/35] fix(share): default to production ingest and app URLs --- packages/opencode/src/share/share-next.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 44303f7b29..35a9a4c38f 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -31,7 +31,7 @@ export namespace ShareNext { const token = await kilocodeToken() if (!token) return undefined - const base = await Config.get().then((x) => x.enterprise?.url ?? "http://localhost:8787") + const base = await Config.get().then((x) => x.enterprise?.url ?? "https://ingest.kilosessions.ai") const baseHeaders: Record = { "Content-Type": "application/json", Authorization: `bearer ${token}`, @@ -147,7 +147,7 @@ export namespace ShareNext { .then((x) => x as { public_id: string }) const current = (await Storage.read(["session_share", sessionId])) as Awaited> - const url = `http://localhost:3000/s/${result.public_id}` + const url = `https://app.kilo.ai/s/${result.public_id}` await Storage.write(["session_share", sessionId], { ...current, From 42658f710aa7ba1a3b67df7b57d7576dbb8b5b81 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 10:18:41 +0100 Subject: [PATCH 09/35] fix(cli): update import to use kilo share URLs Switch URL parsing and API fetch endpoint from opncd.ai to app.kilo.ai, and use the returned payload directly for import. --- packages/opencode/src/cli/cmd/import.ts | 21 ++++++--------------- 1 file changed, 6 insertions(+), 15 deletions(-) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index 9d7e8c5617..31033df3e5 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -31,15 +31,15 @@ export const ImportCommand = cmd({ const isUrl = args.file.startsWith("http://") || args.file.startsWith("https://") if (isUrl) { - const urlMatch = args.file.match(/https?:\/\/opncd\.ai\/share\/([a-zA-Z0-9_-]+)/) + const urlMatch = args.file.match(/https?:\/\/app\.kilo\.ai\/s\/([a-zA-Z0-9_-]+)/) if (!urlMatch) { - process.stdout.write(`Invalid URL format. Expected: https://opncd.ai/share/`) + process.stdout.write(`Invalid URL format. Expected: https://app.kilo.ai/s/`) process.stdout.write(EOL) return } - const slug = urlMatch[1] - const response = await fetch(`https://opncd.ai/api/share/${slug}`) + const id = urlMatch[1] + const response = await fetch(`https://app.kilo.ai/api/s/${id}`) if (!response.ok) { process.stdout.write(`Failed to fetch share data: ${response.statusText}`) @@ -50,21 +50,12 @@ export const ImportCommand = cmd({ const data = await response.json() if (!data.info || !data.messages || Object.keys(data.messages).length === 0) { - process.stdout.write(`Share not found: ${slug}`) + process.stdout.write(`Share not found: ${id}`) process.stdout.write(EOL) return } - exportData = { - info: data.info, - messages: Object.values(data.messages).map((msg: any) => { - const { parts, ...info } = msg - return { - info, - parts, - } - }), - } + exportData = data } else { const file = Bun.file(args.file) exportData = await file.json().catch(() => {}) From b5787409e1b6f7a56891eb5e8af03da7401649cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 10:44:49 +0100 Subject: [PATCH 10/35] fix(share): use proper Bearer auth header casing --- packages/opencode/src/share/share-next.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 35a9a4c38f..1b58fec23b 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -34,7 +34,7 @@ export namespace ShareNext { const base = await Config.get().then((x) => x.enterprise?.url ?? "https://ingest.kilosessions.ai") const baseHeaders: Record = { "Content-Type": "application/json", - Authorization: `bearer ${token}`, + Authorization: `Bearer ${token}`, } const withHeaders = (init?: RequestInit) => { From 8c9c38e2ea91bd4a564f111ee5646e47ebb799d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 11:02:39 +0100 Subject: [PATCH 11/35] fix(cli): update import to use ingest session endpoint --- packages/opencode/src/cli/cmd/import.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index 31033df3e5..50f6d815d9 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -39,7 +39,7 @@ export const ImportCommand = cmd({ } const id = urlMatch[1] - const response = await fetch(`https://app.kilo.ai/api/s/${id}`) + const response = await fetch(`https://ingest.kilosessions.ai/session/${id}`) if (!response.ok) { process.stdout.write(`Failed to fetch share data: ${response.statusText}`) From 574c6ba0c65e7028db5f218048986ad8a644dea2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 11:25:47 +0100 Subject: [PATCH 12/35] docs(cli): update import help text for app.kilo.ai URL --- packages/opencode/src/cli/cmd/import.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index 50f6d815d9..80f3de78cf 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -11,7 +11,7 @@ export const ImportCommand = cmd({ describe: "import session data from JSON file or URL", builder: (yargs: Argv) => { return yargs.positional("file", { - describe: "path to JSON file or opencode.ai share URL", + describe: "path to JSON file or app.kilo.ai share URL", type: "string", demandOption: true, }) From e877b984aeec6c8e82dcdd2791a5d9e594a22842 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 11:34:47 +0100 Subject: [PATCH 13/35] fix(share): delete sessions via session API endpoint --- packages/opencode/src/share/share-next.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 1b58fec23b..2667205e5d 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -256,7 +256,7 @@ export namespace ShareNext { const share = await get(sessionId) if (!share) return - await client.fetch(`${client.url}/api/share/${share.id}`, { + await client.fetch(`${client.url}/api/session/${share.id}`, { method: "DELETE", }) From 7082648ac59d624facf4a77e02faeb27a2710152 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 12:37:40 +0100 Subject: [PATCH 14/35] fix(session): use ShareNext when removing shared sessions --- packages/opencode/src/session/index.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts index eef0100717..ceee3ca5ab 100644 --- a/packages/opencode/src/session/index.ts +++ b/packages/opencode/src/session/index.ts @@ -22,6 +22,7 @@ import { Snapshot } from "@/snapshot" import type { Provider } from "@/provider/provider" import { PermissionNext } from "@/permission/next" import { Global } from "@/global" +import { ShareNext } from "@/share/share-next" export namespace Session { const log = Log.create({ service: "session" }) @@ -340,7 +341,7 @@ export namespace Session { for (const child of await children(sessionID)) { await remove(child.id) } - await unshare(sessionID).catch(() => {}) + await ShareNext.remove(sessionID).catch(() => {}) for (const msg of await Storage.list(["message", sessionID])) { for (const part of await Storage.list(["part", msg.at(-1)!])) { await Storage.remove(part) From c43b5b3a291de2669576eeb21252eae9936545fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 16:15:15 +0100 Subject: [PATCH 15/35] fix(session): lazy-load ShareNext during removal Avoid eager import of ShareNext by dynamically importing it only when removing a session, reducing initialization coupling and preventing import-time issues. --- packages/opencode/src/session/index.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts index 1d8a5088f4..1c87327f74 100644 --- a/packages/opencode/src/session/index.ts +++ b/packages/opencode/src/session/index.ts @@ -22,7 +22,6 @@ import { Snapshot } from "@/snapshot" import type { Provider } from "@/provider/provider" import { PermissionNext } from "@/permission/next" import { Global } from "@/global" -import { ShareNext } from "@/share/share-next" export namespace Session { const log = Log.create({ service: "session" }) @@ -341,6 +340,7 @@ export namespace Session { for (const child of await children(sessionID)) { await remove(child.id) } + const { ShareNext } = await import("@/share/share-next") await ShareNext.remove(sessionID).catch(() => {}) for (const msg of await Storage.list(["message", sessionID])) { for (const part of await Storage.list(["part", msg.at(-1)!])) { From 1dcbe284a63decc906443935bf185619c5a256a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 16:16:29 +0100 Subject: [PATCH 16/35] fix(share): surface share/unshare failures with errors Throw explicit errors when sharing is disabled or credentials are missing, validate session sync initialization, and check HTTP responses/public_id before persisting the share URL. --- packages/opencode/src/share/share-next.ts | 61 +++++++++++++++++------ 1 file changed, 45 insertions(+), 16 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index b06edb33dc..b742c3a4c5 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -131,23 +131,40 @@ export namespace ShareNext { } export async function share(sessionId: string) { - if (disabled) return { url: "" } + if (disabled) { + throw new Error("Sharing is disabled (OPENCODE_DISABLE_SHARE=1)") + } const client = await getClient() - if (!client) return { url: "" } + if (!client) { + throw new Error("Unable to share session: no Kilo credentials found. Run `kilo auth login`.") + } + + const current = (await get(sessionId).catch(() => undefined)) ?? (await create(sessionId)) + if (!current.id || !current.ingestPath) { + throw new Error(`Unable to share session ${sessionId}: failed to initialize session sync.`) + } log.info("sharing", { sessionId }) - const result = await client - .fetch(`${client.url}/api/session/${sessionId}/share`, { - method: "POST", - body: JSON.stringify({ sessionId }), - }) - .then((x) => x.json()) - .then((x) => x as { public_id: string }) + const response = await client.fetch(`${client.url}/api/session/${sessionId}/share`, { + method: "POST", + body: JSON.stringify({ sessionId }), + }) + + if (!response.ok) { + throw new Error(`Unable to share session ${sessionId}: ${response.status} ${response.statusText}`) + } + + const result = (await response.json()) as { public_id?: string } + if (!result.public_id) { + throw new Error(`Unable to share session ${sessionId}: server did not return a public id`) + } - const current = (await Storage.read(["session_share", sessionId])) as Awaited> const url = `https://app.kilo.ai/s/${result.public_id}` + if (!url) { + throw new Error(`Unable to share session ${sessionId}: generated share URL is empty`) + } await Storage.write(["session_share", sessionId], { ...current, @@ -158,23 +175,35 @@ export namespace ShareNext { } export async function unshare(sessionId: string) { - if (disabled) return + if (disabled) { + throw new Error("Unshare is disabled (OPENCODE_DISABLE_SHARE=1)") + } const client = await getClient() - if (!client) return + if (!client) { + throw new Error("Unable to unshare session: no Kilo credentials found. Run `opencode auth login`.") + } log.info("unsharing", { sessionId }) - const result = await client.fetch(`${client.url}/api/session/${sessionId}/unshare`, { + const response = await client.fetch(`${client.url}/api/session/${sessionId}/unshare`, { method: "POST", body: JSON.stringify({ sessionId }), }) - const current = (await Storage.read(["session_share", sessionId])) as Awaited> + if (!response.ok) { + throw new Error(`Unable to unshare session ${sessionId}: ${response.status} ${response.statusText}`) + } - delete current.url + const current = await get(sessionId).catch(() => undefined) + if (!current) return - await Storage.write(["session_share", sessionId], current) + const next = { + ...current, + } + delete next.url + + await Storage.write(["session_share", sessionId], next) } function get(sessionId: string) { From e76cc9a55d871b59c7e1e96177c0b1e356725acf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 16:17:18 +0100 Subject: [PATCH 17/35] chore(share): rename disable share env var Update share modules, CI workflow, and e2e scripts to use KILO_DISABLE_SHARE instead of OPENCODE_DISABLE_SHARE. --- .github/workflows/test.yml | 6 +++--- packages/app/script/e2e-local.ts | 2 +- packages/opencode/src/share/share-next.ts | 6 +++--- packages/opencode/src/share/share.ts | 2 +- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ba80a69540..35b13d268c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -67,7 +67,7 @@ jobs: working-directory: packages/opencode run: bun script/seed-e2e.ts env: - OPENCODE_DISABLE_SHARE: "true" + KILO_DISABLE_SHARE: "true" OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" @@ -86,7 +86,7 @@ jobs: working-directory: packages/opencode run: bun dev -- --print-logs --log-level WARN serve --port 4096 --hostname 127.0.0.1 & env: - OPENCODE_DISABLE_SHARE: "true" + KILO_DISABLE_SHARE: "true" OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" @@ -111,7 +111,7 @@ jobs: run: ${{ matrix.settings.command }} env: CI: true - OPENCODE_DISABLE_SHARE: "true" + KILO_DISABLE_SHARE: "true" OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" diff --git a/packages/app/script/e2e-local.ts b/packages/app/script/e2e-local.ts index 2c7be2ad95..0d381ab635 100644 --- a/packages/app/script/e2e-local.ts +++ b/packages/app/script/e2e-local.ts @@ -58,7 +58,7 @@ const sandbox = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-e2e-")) const serverEnv = { ...process.env, - OPENCODE_DISABLE_SHARE: "true", + KILO_DISABLE_SHARE: "true", OPENCODE_DISABLE_LSP_DOWNLOAD: "true", OPENCODE_DISABLE_DEFAULT_PLUGINS: "true", OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true", diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index b742c3a4c5..f5fda9dabc 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -52,7 +52,7 @@ export namespace ShareNext { } } - const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1" + const disabled = process.env["KILO_DISABLE_SHARE"] === "true" || process.env["KILO_DISABLE_SHARE"] === "1" export async function init() { Bus.subscribe(Session.Event.Created, async (evt) => { @@ -132,7 +132,7 @@ export namespace ShareNext { export async function share(sessionId: string) { if (disabled) { - throw new Error("Sharing is disabled (OPENCODE_DISABLE_SHARE=1)") + throw new Error("Sharing is disabled (KILO_DISABLE_SHARE=1)") } const client = await getClient() @@ -176,7 +176,7 @@ export namespace ShareNext { export async function unshare(sessionId: string) { if (disabled) { - throw new Error("Unshare is disabled (OPENCODE_DISABLE_SHARE=1)") + throw new Error("Unshare is disabled (KILO_DISABLE_SHARE=1)") } const client = await getClient() diff --git a/packages/opencode/src/share/share.ts b/packages/opencode/src/share/share.ts index f7bf4b3fa5..b4121e8d4b 100644 --- a/packages/opencode/src/share/share.ts +++ b/packages/opencode/src/share/share.ts @@ -70,7 +70,7 @@ export namespace Share { process.env["OPENCODE_API"] ?? (Installation.isPreview() || Installation.isLocal() ? "https://api.dev.opencode.ai" : "https://api.opencode.ai") - const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1" + const disabled = process.env["KILO_DISABLE_SHARE"] === "true" || process.env["KILO_DISABLE_SHARE"] === "1" export async function create(sessionID: string) { if (disabled) return { url: "", secret: "" } From fbd25d609fa922cf838249823aab5efcf78d86a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 16:25:17 +0100 Subject: [PATCH 18/35] fix(share): update auth login hint in unshare error --- packages/opencode/src/share/share-next.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index f5fda9dabc..b27022cbca 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -181,7 +181,7 @@ export namespace ShareNext { const client = await getClient() if (!client) { - throw new Error("Unable to unshare session: no Kilo credentials found. Run `opencode auth login`.") + throw new Error("Unable to unshare session: no Kilo credentials found. Run `kilo auth login`.") } log.info("unsharing", { sessionId }) From 722c97da923598866059759a186baff86ef93e33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 16:25:54 +0100 Subject: [PATCH 19/35] refactor(share): remove redundant share url empty check --- packages/opencode/src/share/share-next.ts | 3 --- 1 file changed, 3 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index b27022cbca..c9d1f1c7c9 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -162,9 +162,6 @@ export namespace ShareNext { } const url = `https://app.kilo.ai/s/${result.public_id}` - if (!url) { - throw new Error(`Unable to share session ${sessionId}: generated share URL is empty`) - } await Storage.write(["session_share", sessionId], { ...current, From 282e3ba4ae0c407c0b70a25405ca768356d8e0d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Wed, 28 Jan 2026 18:31:47 +0100 Subject: [PATCH 20/35] perf(share): debounce session sync and retry on failure Coalesce frequent session updates into a per-session debounce queue with stable entity keys to reduce redundant POSTs. Improve resilience by handling async init/full sync errors, validating responses, and requeueing failed batches without overwriting newer updates. --- packages/opencode/src/share/share-next.ts | 139 ++++++++++++++++++---- 1 file changed, 114 insertions(+), 25 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index c9d1f1c7c9..1b87dd1447 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -1,4 +1,4 @@ -// kilocode_change pretty much completely refactored +// kilocode_change pretty much completely refactored - @iscekic for conflicts import { Bus } from "@/bus" import { Config } from "@/config/config" import { ulid } from "ulid" @@ -10,6 +10,9 @@ import { Log } from "@/util/log" import { Auth } from "@/auth" import type * as SDK from "@kilocode/sdk/v2" // kilocode_change +/** + * Even though this is called "share-next", this is where we handle session stuff. + */ export namespace ShareNext { const log = Log.create({ service: "share-next" }) @@ -55,8 +58,9 @@ export namespace ShareNext { const disabled = process.env["KILO_DISABLE_SHARE"] === "true" || process.env["KILO_DISABLE_SHARE"] === "1" export async function init() { - Bus.subscribe(Session.Event.Created, async (evt) => { - await create(evt.properties.info.id) + Bus.subscribe(Session.Event.Created, (evt) => { + const sessionId = evt.properties.info.id + void create(sessionId).catch((error) => log.error("share init create failed", { sessionId, error })) }) Bus.subscribe(Session.Event.Updated, async (evt) => { @@ -125,7 +129,7 @@ export namespace ShareNext { await Storage.write(["session_share", sessionId], result) - fullSync(sessionId) + void fullSync(sessionId).catch((error) => log.error("share full sync failed", { sessionId, error })) return result } @@ -233,43 +237,128 @@ export namespace ShareNext { data: SDK.Model[] } + // Per-session debounce queue. + // + // Events fire frequently (message/part updates during streaming), so we coalesce many updates + // into at most one POST per ~1s per session. + // + // - Outer Map key: local session id + // - Inner Map key: stable entity key (message:, part:, etc.) so newer updates overwrite older + // within the same debounce window. const queue = new Map }>() + + function id(value: unknown) { + if (!value) return undefined + if (typeof value !== "object") return undefined + if (!("id" in value)) return undefined + const result = (value as { id?: unknown }).id + if (typeof result === "string" && result.length > 0) return result + return undefined + } + + function key(item: Data) { + // Stable keys are important so updates for the same entity collapse to a single queued item. + // If we can't derive a stable key, we fall back to a random key (ulid) so the item is still sent. + if (item.type === "session") return "session" + if (item.type === "session_diff") return "session_diff" + + if (item.type === "message") { + const value = id(item.data) + return value ? `message:${value}` : ulid() + } + + if (item.type === "part") { + const value = id(item.data) + return value ? `part:${value}` : ulid() + } + + const models = item.data + .map((m) => `${m.providerID}:${m.id}`) + .sort() + .join(",") + return models.length > 0 ? `model:${models}` : ulid() + } + + function flush(sessionId: string, timeout: NodeJS.Timeout) { + // Flush is scheduled by sync() and sends the currently queued payload. + // + // Note: we delete the queue entry before the network call so that new incoming events can start + // a fresh debounce window immediately. + void (async () => { + const queued = queue.get(sessionId) + if (!queued) return + + clearTimeout(timeout) + queue.delete(sessionId) + + try { + const share = await get(sessionId).catch(() => undefined) + if (!share) return + + const client = await getClient() + if (!client) return + + const response = await client.fetch(`${client.url}${share.ingestPath}`, { + method: "POST", + body: JSON.stringify({ + data: Array.from(queued.data.values()), + }), + }) + + if (!response.ok) { + throw new Error(`sync failed: ${response.status} ${response.statusText}`) + } + } catch (error) { + log.error("share sync failed", { sessionId, error }) + // Requeue without overwriting newer updates. + // If a new debounce window is already queued (due to fresh events while this flush was in-flight), + // only fill missing keys so we don't clobber newer data with stale data from the failed batch. + requeue(sessionId, Array.from(queued.data.values())) + } + })() + } + + function requeue(sessionId: string, items: Data[]) { + const existing = queue.get(sessionId) + if (existing) { + for (const item of items) { + const k = key(item) + if (existing.data.has(k)) continue + existing.data.set(k, item) + } + return + } + + const dataMap = new Map() + for (const item of items) { + dataMap.set(key(item), item) + } + + const timeout = setTimeout(() => flush(sessionId, timeout), 1000) + queue.set(sessionId, { timeout, data: dataMap }) + } + async function sync(sessionId: string, data: Data[]) { + // sync() is called by event handlers and is intentionally cheap: + // - If sharing isn't configured (no token / disabled), we skip queueing. + // - Otherwise, merge into the pending queue entry (if present) or start a new 1s timer. const client = await getClient() if (!client) return const existing = queue.get(sessionId) if (existing) { for (const item of data) { - existing.data.set("id" in item ? (item.id as string) : ulid(), item) + existing.data.set(key(item), item) } return } const dataMap = new Map() for (const item of data) { - dataMap.set("id" in item ? (item.id as string) : ulid(), item) + dataMap.set(key(item), item) } - const timeout = setTimeout(async () => { - const queued = queue.get(sessionId) - if (!queued) return - - queue.delete(sessionId) - - const share = await get(sessionId).catch(() => undefined) - if (!share) return - - const client = await getClient() - if (!client) return - - await client.fetch(`${client.url}${share.ingestPath}`, { - method: "POST", - body: JSON.stringify({ - data: Array.from(queued.data.values()), - }), - }) - }, 1000) + const timeout = setTimeout(() => flush(sessionId, timeout), 1000) queue.set(sessionId, { timeout, data: dataMap }) } From ed19d4872ccd199db4400eae6e1f25997c2fcf00 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 11:12:20 +0100 Subject: [PATCH 21/35] fix(share): skip session sync when kilo token is invalid Validate the kilo API token against the user endpoint before syncing and cache auth validity per token to avoid repeated checks. --- packages/opencode/src/share/share-next.ts | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 1b87dd1447..3414c74b94 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -16,6 +16,23 @@ import type * as SDK from "@kilocode/sdk/v2" // kilocode_change export namespace ShareNext { const log = Log.create({ service: "share-next" }) + const authCache = new Map() + + async function authValid(token: string) { + const cached = authCache.get(token) + if (cached) return cached.valid + + const response = await fetch("https://app.kilo.ai/api/user", { + headers: { + Authorization: `Bearer ${token}`, + }, + }).catch(() => undefined) + + const valid = response ? response.ok : false + authCache.set(token, { valid: !!response?.ok }) + return valid + } + export async function kilocodeToken() { const auth = await Auth.get("kilo") if (auth?.type === "api" && auth.key.length > 0) return auth.key @@ -34,6 +51,9 @@ export namespace ShareNext { const token = await kilocodeToken() if (!token) return undefined + const valid = await authValid(token) + if (!valid) return undefined + const base = await Config.get().then((x) => x.enterprise?.url ?? "https://ingest.kilosessions.ai") const baseHeaders: Record = { "Content-Type": "application/json", From 1b75836d8a6b18f463bf32fa9f00b7fbb63de28a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 11:15:39 +0100 Subject: [PATCH 22/35] fix(share): remove disabled guard from getClient --- packages/opencode/src/share/share-next.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 3414c74b94..72f65a9321 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -47,7 +47,6 @@ export namespace ShareNext { } async function getClient(): Promise { - if (disabled) return undefined const token = await kilocodeToken() if (!token) return undefined From 7fff59fa477e6dc903e17b41d568e7fd7934d4ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 11:23:20 +0100 Subject: [PATCH 23/35] fix(cli): validate and parse kilo session URLs safely --- packages/opencode/src/cli/cmd/import.ts | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index 80f3de78cf..22072612a5 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -31,14 +31,28 @@ export const ImportCommand = cmd({ const isUrl = args.file.startsWith("http://") || args.file.startsWith("https://") if (isUrl) { - const urlMatch = args.file.match(/https?:\/\/app\.kilo\.ai\/s\/([a-zA-Z0-9_-]+)/) - if (!urlMatch) { + const url = (() => { + try { + return new URL(args.file) + } catch { + return undefined + } + })() + + if (!url || url.hostname !== "app.kilo.ai") { + process.stdout.write(`Invalid URL format. Expected: https://app.kilo.ai/s/`) + process.stdout.write(EOL) + return + } + + const parts = url.pathname.split("/").filter(Boolean) + const id = parts.length >= 2 && parts[0] === "s" ? parts[1] : undefined + if (!id) { process.stdout.write(`Invalid URL format. Expected: https://app.kilo.ai/s/`) process.stdout.write(EOL) return } - const id = urlMatch[1] const response = await fetch(`https://ingest.kilosessions.ai/session/${id}`) if (!response.ok) { From 70675549784006bf7dc0b192e7551b21987899d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 11:27:03 +0100 Subject: [PATCH 24/35] feat(share): allow disabling session ingest Add `KILO_DISABLE_SESSION_INGEST` to skip session ingest initialization. Also cache the ingest client briefly to reduce repeated auth/token checks and surface clearer errors when session creation requests fail. --- packages/opencode/src/share/share-next.ts | 94 +++++++++++++++-------- 1 file changed, 60 insertions(+), 34 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 72f65a9321..c2fb417404 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -1,6 +1,5 @@ // kilocode_change pretty much completely refactored - @iscekic for conflicts import { Bus } from "@/bus" -import { Config } from "@/config/config" import { ulid } from "ulid" import { Provider } from "@/provider/provider" import { Session } from "@/session" @@ -8,7 +7,7 @@ import { MessageV2 } from "@/session/message-v2" import { Storage } from "@/storage/storage" import { Log } from "@/util/log" import { Auth } from "@/auth" -import type * as SDK from "@kilocode/sdk/v2" // kilocode_change +import type * as SDK from "@kilocode/sdk/v2" /** * Even though this is called "share-next", this is where we handle session stuff. @@ -46,37 +45,61 @@ export namespace ShareNext { fetch: (input: RequestInfo | URL, init?: RequestInit) => Promise } + const cache = { + at: 0, + value: undefined as Client | undefined, + inflight: undefined as Promise | undefined, + } + async function getClient(): Promise { - const token = await kilocodeToken() - if (!token) return undefined + const now = Date.now() + if (cache.value && now - cache.at < 5_000) return cache.value + if (cache.inflight && now - cache.at < 5_000) return cache.inflight - const valid = await authValid(token) - if (!valid) return undefined + cache.at = now + cache.inflight = (async () => { + const token = await kilocodeToken() + if (!token) return undefined - const base = await Config.get().then((x) => x.enterprise?.url ?? "https://ingest.kilosessions.ai") - const baseHeaders: Record = { - "Content-Type": "application/json", - Authorization: `Bearer ${token}`, - } + const valid = await authValid(token) + if (!valid) return undefined + + const base = "https://ingest.kilosessions.ai" + const baseHeaders: Record = { + "Content-Type": "application/json", + Authorization: `Bearer ${token}`, + } + + const withHeaders = (init?: RequestInit) => { + const headers = new Headers(init?.headers) + for (const [k, v] of Object.entries(baseHeaders)) headers.set(k, v) + return { + ...init, + headers, + } satisfies RequestInit + } - const withHeaders = (init?: RequestInit) => { - const headers = new Headers(init?.headers) - for (const [k, v] of Object.entries(baseHeaders)) headers.set(k, v) return { - ...init, - headers, - } satisfies RequestInit - } + url: base, + fetch: (input, init) => fetch(input, withHeaders(init)), + } + })() - return { - url: base, - fetch: (input, init) => fetch(input, withHeaders(init)), + try { + cache.value = await cache.inflight + return cache.value + } finally { + cache.inflight = undefined } } - const disabled = process.env["KILO_DISABLE_SHARE"] === "true" || process.env["KILO_DISABLE_SHARE"] === "1" + const shareDisabled = process.env["KILO_DISABLE_SHARE"] === "true" || process.env["KILO_DISABLE_SHARE"] === "1" + const ingestDisabled = + process.env["KILO_DISABLE_SESSION_INGEST"] === "true" || process.env["KILO_DISABLE_SESSION_INGEST"] === "1" export async function init() { + if (ingestDisabled) return + Bus.subscribe(Session.Event.Created, (evt) => { const sessionId = evt.properties.info.id void create(sessionId).catch((error) => log.error("share init create failed", { sessionId, error })) @@ -138,13 +161,16 @@ export namespace ShareNext { log.info("creating session", { sessionId }) - const result = await client - .fetch(`${client.url}/api/session`, { - method: "POST", - body: JSON.stringify({ sessionId }), - }) - .then((x) => x.json()) - .then((x) => x as { id: string; ingestPath: string }) + const response = await client.fetch(`${client.url}/api/session`, { + method: "POST", + body: JSON.stringify({ sessionId }), + }) + + if (!response.ok) { + throw new Error(`Unable to create session ${sessionId}: ${response.status} ${response.statusText}`) + } + + const result = (await response.json()) as { id: string; ingestPath: string } await Storage.write(["session_share", sessionId], result) @@ -154,7 +180,7 @@ export namespace ShareNext { } export async function share(sessionId: string) { - if (disabled) { + if (shareDisabled) { throw new Error("Sharing is disabled (KILO_DISABLE_SHARE=1)") } @@ -195,7 +221,7 @@ export namespace ShareNext { } export async function unshare(sessionId: string) { - if (disabled) { + if (shareDisabled) { throw new Error("Unshare is disabled (KILO_DISABLE_SHARE=1)") } @@ -361,9 +387,6 @@ export namespace ShareNext { // sync() is called by event handlers and is intentionally cheap: // - If sharing isn't configured (no token / disabled), we skip queueing. // - Otherwise, merge into the pending queue entry (if present) or start a new 1s timer. - const client = await getClient() - if (!client) return - const existing = queue.get(sessionId) if (existing) { for (const item of data) { @@ -372,6 +395,9 @@ export namespace ShareNext { return } + const client = await getClient() + if (!client) return + const dataMap = new Map() for (const item of data) { dataMap.set(key(item), item) From 96404c0f4419f041f7d3d3735473fa6215d6e28f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 11:46:50 +0100 Subject: [PATCH 25/35] chore(e2e): disable kilo session ingest in test runs --- .github/workflows/test.yml | 3 +++ packages/app/script/e2e-local.ts | 1 + 2 files changed, 4 insertions(+) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b6b1499921..b0b37264de 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -70,6 +70,7 @@ jobs: KILO_API_KEY: ${{ secrets.KILO_API_KEY }} KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }} KILO_DISABLE_SHARE: "true" + KILO_DISABLE_SESSION_INGEST: "true" OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" @@ -91,6 +92,7 @@ jobs: KILO_API_KEY: ${{ secrets.KILO_API_KEY }} KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }} KILO_DISABLE_SHARE: "true" + KILO_DISABLE_SESSION_INGEST: "true" OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" @@ -118,6 +120,7 @@ jobs: KILO_API_KEY: ${{ secrets.KILO_API_KEY }} KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }} KILO_DISABLE_SHARE: "true" + KILO_DISABLE_SESSION_INGEST: "true" OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" diff --git a/packages/app/script/e2e-local.ts b/packages/app/script/e2e-local.ts index 364d487b05..809e43f6db 100644 --- a/packages/app/script/e2e-local.ts +++ b/packages/app/script/e2e-local.ts @@ -59,6 +59,7 @@ const sandbox = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-e2e-")) const serverEnv = { ...process.env, KILO_DISABLE_SHARE: "true", + KILO_DISABLE_SESSION_INGEST: "true", OPENCODE_DISABLE_LSP_DOWNLOAD: "true", OPENCODE_DISABLE_DEFAULT_PLUGINS: "true", OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true", From b3d2673499f7597b58b56e6dd3dae56d47684a61 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 11:50:30 +0100 Subject: [PATCH 26/35] fix(share): avoid caching auth on network failures Skip caching when the auth request fails without a response so subsequent calls can retry instead of persisting an invalid state. --- packages/opencode/src/share/share-next.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index c2fb417404..8f85c2cca8 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -27,8 +27,11 @@ export namespace ShareNext { }, }).catch(() => undefined) - const valid = response ? response.ok : false - authCache.set(token, { valid: !!response?.ok }) + // Don't cache transient network failures; allow future calls to retry. + if (!response) return false + + const valid = response.ok + authCache.set(token, { valid }) return valid } From f3e5c7eebc2f4581b4bd9d326ddb81d3f977fa75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 12:08:07 +0100 Subject: [PATCH 27/35] fix(cli): encode session id in import fetch url --- packages/opencode/src/cli/cmd/import.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index 22072612a5..2ea9a8fac5 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -53,7 +53,7 @@ export const ImportCommand = cmd({ return } - const response = await fetch(`https://ingest.kilosessions.ai/session/${id}`) + const response = await fetch(`https://ingest.kilosessions.ai/session/${encodeURIComponent(id)}`) if (!response.ok) { process.stdout.write(`Failed to fetch share data: ${response.statusText}`) From b81323b062d9aacce5bf18d8168654b038692ce9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 12:08:28 +0100 Subject: [PATCH 28/35] fix(share): block share/unshare when ingest disabled Add ingest-disabled guardrails to share and unshare, and harden session removal by encoding the session id and logging network/HTTP failures instead of throwing. --- packages/opencode/src/share/share-next.ts | 30 ++++++++++++++++++++--- 1 file changed, 27 insertions(+), 3 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 8f85c2cca8..c594c9a791 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -183,6 +183,10 @@ export namespace ShareNext { } export async function share(sessionId: string) { + if (ingestDisabled) { + throw new Error("Session ingest is disabled (KILO_DISABLE_SESSION_INGEST=1)") + } + if (shareDisabled) { throw new Error("Sharing is disabled (KILO_DISABLE_SHARE=1)") } @@ -224,6 +228,10 @@ export namespace ShareNext { } export async function unshare(sessionId: string) { + if (ingestDisabled) { + throw new Error("Session ingest is disabled (KILO_DISABLE_SESSION_INGEST=1)") + } + if (shareDisabled) { throw new Error("Unshare is disabled (KILO_DISABLE_SHARE=1)") } @@ -419,9 +427,25 @@ export namespace ShareNext { const share = await get(sessionId) if (!share) return - await client.fetch(`${client.url}/api/session/${share.id}`, { - method: "DELETE", - }) + const response = await client + .fetch(`${client.url}/api/session/${encodeURIComponent(share.id)}`, { + method: "DELETE", + }) + .catch(() => undefined) + + if (!response) { + log.error("share remove failed", { sessionId, error: "network" }) + return + } + + if (!response.ok) { + log.error("share remove failed", { + sessionId, + status: response.status, + statusText: response.statusText, + }) + return + } await Storage.remove(["session_share", sessionId]) } From 81f69b4f04d27ed965f6c168f07993137bbfeff5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 12:26:53 +0100 Subject: [PATCH 29/35] fix(share): add retry/backoff to session sync flush Coalesce ingest updates with a per-session due time and schedule flushes to respect active backoff windows. Retry transient failures (network, 429, 5xx, etc.) with capped exponential backoff and a small retry budget, while avoiding overwriting newer queued updates during in-flight retries. --- packages/opencode/src/share/share-next.ts | 206 ++++++++++++++++------ 1 file changed, 149 insertions(+), 57 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index c594c9a791..6bd06ef561 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -293,15 +293,38 @@ export namespace ShareNext { data: SDK.Model[] } - // Per-session debounce queue. + // Per-session debounce/flush queue. // - // Events fire frequently (message/part updates during streaming), so we coalesce many updates - // into at most one POST per ~1s per session. + // The share ingest endpoint is updated very frequently (streaming message parts, diffs, etc.). + // To avoid spamming the server, we coalesce updates and flush at most once per ~1s per session. // - // - Outer Map key: local session id - // - Inner Map key: stable entity key (message:, part:, etc.) so newer updates overwrite older - // within the same debounce window. - const queue = new Map }>() + // `due` is the earliest time we should flush; it is also used to respect backoff when retries are + // active. A later `due` always wins over an earlier one. + const queue = new Map }>() + + // Per-session retry state. + // + // We keep retry logic intentionally simple and local: + // - Only retry a small set of transient errors (network, 429, 5xx, etc.) + // - Use exponential backoff with a small max budget to prevent infinite loops/log spam + // - Store `until` so sync() can avoid scheduling a flush before backoff expires + const retry = new Map() + + function retryable(status: number) { + // Retry only statuses that are likely transient. + if (status === 408) return true + if (status === 409) return true + if (status === 425) return true + if (status === 429) return true + if (status >= 500) return true + return false + } + + function backoff(count: number) { + // Exponential backoff capped to keep the system responsive. + const clamped = Math.min(count, 6) + return Math.min(60_000, 1_000 * 2 ** (clamped - 1)) + } function id(value: unknown) { if (!value) return undefined @@ -335,7 +358,43 @@ export namespace ShareNext { return models.length > 0 ? `model:${models}` : ulid() } - function flush(sessionId: string, timeout: NodeJS.Timeout) { + function schedule(sessionId: string, due: number, data: Map) { + const existing = queue.get(sessionId) + if (existing) { + // Don't reschedule if an earlier flush is already planned. + // We only move the flush later (e.g., to respect backoff). + if (existing.due >= due) return + clearTimeout(existing.timeout) + } + + const wait = Math.max(0, due - Date.now()) + const timeout = setTimeout(() => flush(sessionId), wait) + queue.set(sessionId, { timeout, due, data }) + } + + function enqueue(sessionId: string, items: Data[], mode: "overwrite" | "fill", due: number) { + const existing = queue.get(sessionId) + if (existing) { + for (const item of items) { + const k = key(item) + // overwrite: normal event updates (newer data should win) + // fill: retry requeue (never clobber newer updates that arrived while a flush was in-flight) + if (mode === "fill" && existing.data.has(k)) continue + existing.data.set(k, item) + } + schedule(sessionId, due, existing.data) + return + } + + const data = new Map() + for (const item of items) { + data.set(key(item), item) + } + + schedule(sessionId, due, data) + } + + function flush(sessionId: string) { // Flush is scheduled by sync() and sends the currently queued payload. // // Note: we delete the queue entry before the network call so that new incoming events can start @@ -344,9 +403,11 @@ export namespace ShareNext { const queued = queue.get(sessionId) if (!queued) return - clearTimeout(timeout) + clearTimeout(queued.timeout) queue.delete(sessionId) + const items = Array.from(queued.data.values()) + try { const share = await get(sessionId).catch(() => undefined) if (!share) return @@ -354,68 +415,99 @@ export namespace ShareNext { const client = await getClient() if (!client) return - const response = await client.fetch(`${client.url}${share.ingestPath}`, { - method: "POST", - body: JSON.stringify({ - data: Array.from(queued.data.values()), - }), - }) + const response = await client + .fetch(`${client.url}${share.ingestPath}`, { + method: "POST", + body: JSON.stringify({ + data: items, + }), + }) + .catch(() => undefined) - if (!response.ok) { - throw new Error(`sync failed: ${response.status} ${response.statusText}`) + if (!response) { + // Network failures are assumed transient; retry with backoff and a small budget. + const count = (retry.get(sessionId)?.count ?? 0) + 1 + if (count > 6) { + log.error("share sync failed", { sessionId, error: "retry budget exceeded" }) + retry.delete(sessionId) + return + } + + const delay = backoff(count) + retry.set(sessionId, { count, until: Date.now() + delay }) + log.error("share sync failed", { sessionId, error: "network", retryInMs: delay }) + enqueue(sessionId, items, "fill", Date.now() + delay) + return } + + if (response.ok) { + retry.delete(sessionId) + return + } + + if (response.status === 401 || response.status === 403) { + // Non-retryable until credentials are fixed. + // Clearing caches prevents repeated use of a now-invalid token/client. + authCache.clear() + cache.value = undefined + cache.inflight = undefined + cache.at = 0 + log.error("share sync failed", { + sessionId, + status: response.status, + statusText: response.statusText, + }) + retry.delete(sessionId) + return + } + + if (!retryable(response.status)) { + // Permanent-ish failures (eg. 404 due to bad ingestPath) should not loop forever. + log.error("share sync failed", { + sessionId, + status: response.status, + statusText: response.statusText, + }) + retry.delete(sessionId) + return + } + + const current = retry.get(sessionId) + const count = (current?.count ?? 0) + 1 + if (count > 6) { + log.error("share sync failed", { sessionId, error: "retry budget exceeded" }) + retry.delete(sessionId) + return + } + + const delay = backoff(count) + retry.set(sessionId, { count, until: Date.now() + delay }) + log.error("share sync failed", { + sessionId, + status: response.status, + statusText: response.statusText, + retryInMs: delay, + }) + enqueue(sessionId, items, "fill", Date.now() + delay) } catch (error) { log.error("share sync failed", { sessionId, error }) - // Requeue without overwriting newer updates. - // If a new debounce window is already queued (due to fresh events while this flush was in-flight), - // only fill missing keys so we don't clobber newer data with stale data from the failed batch. - requeue(sessionId, Array.from(queued.data.values())) } })() } - function requeue(sessionId: string, items: Data[]) { - const existing = queue.get(sessionId) - if (existing) { - for (const item of items) { - const k = key(item) - if (existing.data.has(k)) continue - existing.data.set(k, item) - } - return - } - - const dataMap = new Map() - for (const item of items) { - dataMap.set(key(item), item) - } - - const timeout = setTimeout(() => flush(sessionId, timeout), 1000) - queue.set(sessionId, { timeout, data: dataMap }) - } - async function sync(sessionId: string, data: Data[]) { // sync() is called by event handlers and is intentionally cheap: // - If sharing isn't configured (no token / disabled), we skip queueing. - // - Otherwise, merge into the pending queue entry (if present) or start a new 1s timer. - const existing = queue.get(sessionId) - if (existing) { - for (const item of data) { - existing.data.set(key(item), item) - } - return - } - + // - Otherwise, merge into the pending queue entry. + // The next flush is scheduled ~1s after the first queued event (throttled), but never earlier + // than the current backoff window (if retries are active). const client = await getClient() if (!client) return - const dataMap = new Map() - for (const item of data) { - dataMap.set(key(item), item) - } - - const timeout = setTimeout(() => flush(sessionId, timeout), 1000) - queue.set(sessionId, { timeout, data: dataMap }) + const until = retry.get(sessionId)?.until ?? 0 + const base = queue.get(sessionId)?.due ?? Date.now() + 1000 + const due = Math.max(base, until) + enqueue(sessionId, data, "overwrite", due) } export async function remove(sessionId: string) { From ecc2b29824f3888369869def833560a1a894c8fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 12:39:30 +0100 Subject: [PATCH 30/35] fix(cli): validate import response messages array --- packages/opencode/src/cli/cmd/import.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index 2ea9a8fac5..f79d3b5e2a 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -63,7 +63,7 @@ export const ImportCommand = cmd({ const data = await response.json() - if (!data.info || !data.messages || Object.keys(data.messages).length === 0) { + if (!data.info || !data.messages || !Array.isArray(data.messages) || data.messages.length === 0) { process.stdout.write(`Share not found: ${id}`) process.stdout.write(EOL) return From f5b7b82d6d1faf8d9e397e0986ab66e05eb84bda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 12:59:43 +0100 Subject: [PATCH 31/35] fix(share): encode session id in share/unshare urls --- packages/opencode/src/share/share-next.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 6bd06ef561..8f5260149c 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -203,7 +203,7 @@ export namespace ShareNext { log.info("sharing", { sessionId }) - const response = await client.fetch(`${client.url}/api/session/${sessionId}/share`, { + const response = await client.fetch(`${client.url}/api/session/${encodeURIComponent(sessionId)}/share`, { method: "POST", body: JSON.stringify({ sessionId }), }) @@ -243,7 +243,7 @@ export namespace ShareNext { log.info("unsharing", { sessionId }) - const response = await client.fetch(`${client.url}/api/session/${sessionId}/unshare`, { + const response = await client.fetch(`${client.url}/api/session/${encodeURIComponent(sessionId)}/unshare`, { method: "POST", body: JSON.stringify({ sessionId }), }) From 452a88ec8d5628b1051173fc372969ebc218b8cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 13:06:29 +0100 Subject: [PATCH 32/35] refactor(share): remove legacy Share sync module Drop Share.init() from bootstrap and delete the old Share implementation now superseded by ShareNext. --- packages/opencode/src/project/bootstrap.ts | 2 - packages/opencode/src/share/share.ts | 92 ---------------------- 2 files changed, 94 deletions(-) delete mode 100644 packages/opencode/src/share/share.ts diff --git a/packages/opencode/src/project/bootstrap.ts b/packages/opencode/src/project/bootstrap.ts index efdcaba990..a2be3733f8 100644 --- a/packages/opencode/src/project/bootstrap.ts +++ b/packages/opencode/src/project/bootstrap.ts @@ -1,5 +1,4 @@ import { Plugin } from "../plugin" -import { Share } from "../share/share" import { Format } from "../format" import { LSP } from "../lsp" import { FileWatcher } from "../file/watcher" @@ -17,7 +16,6 @@ import { Truncate } from "../tool/truncation" export async function InstanceBootstrap() { Log.Default.info("bootstrapping", { directory: Instance.directory }) await Plugin.init() - Share.init() ShareNext.init() Format.init() await LSP.init() diff --git a/packages/opencode/src/share/share.ts b/packages/opencode/src/share/share.ts deleted file mode 100644 index b4121e8d4b..0000000000 --- a/packages/opencode/src/share/share.ts +++ /dev/null @@ -1,92 +0,0 @@ -import { Bus } from "../bus" -import { Installation } from "../installation" -import { Session } from "../session" -import { MessageV2 } from "../session/message-v2" -import { Log } from "../util/log" - -export namespace Share { - const log = Log.create({ service: "share" }) - - let queue: Promise = Promise.resolve() - const pending = new Map() - - export async function sync(key: string, content: any) { - if (disabled) return - const [root, ...splits] = key.split("/") - if (root !== "session") return - const [sub, sessionID] = splits - if (sub === "share") return - const share = await Session.getShare(sessionID).catch(() => {}) - if (!share) return - const { secret } = share - pending.set(key, content) - queue = queue - .then(async () => { - const content = pending.get(key) - if (content === undefined) return - pending.delete(key) - - return fetch(`${URL}/share_sync`, { - method: "POST", - body: JSON.stringify({ - sessionID: sessionID, - secret, - key: key, - content, - }), - }) - }) - .then((x) => { - if (x) { - log.info("synced", { - key: key, - status: x.status, - }) - } - }) - } - - export function init() { - Bus.subscribe(Session.Event.Updated, async (evt) => { - await sync("session/info/" + evt.properties.info.id, evt.properties.info) - }) - Bus.subscribe(MessageV2.Event.Updated, async (evt) => { - await sync("session/message/" + evt.properties.info.sessionID + "/" + evt.properties.info.id, evt.properties.info) - }) - Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => { - await sync( - "session/part/" + - evt.properties.part.sessionID + - "/" + - evt.properties.part.messageID + - "/" + - evt.properties.part.id, - evt.properties.part, - ) - }) - } - - export const URL = - process.env["OPENCODE_API"] ?? - (Installation.isPreview() || Installation.isLocal() ? "https://api.dev.opencode.ai" : "https://api.opencode.ai") - - const disabled = process.env["KILO_DISABLE_SHARE"] === "true" || process.env["KILO_DISABLE_SHARE"] === "1" - - export async function create(sessionID: string) { - if (disabled) return { url: "", secret: "" } - return fetch(`${URL}/share_create`, { - method: "POST", - body: JSON.stringify({ sessionID: sessionID }), - }) - .then((x) => x.json()) - .then((x) => x as { url: string; secret: string }) - } - - export async function remove(sessionID: string, secret: string) { - if (disabled) return {} - return fetch(`${URL}/share_delete`, { - method: "POST", - body: JSON.stringify({ sessionID, secret }), - }).then((x) => x.json()) - } -} From a7910b4d926472e014cbd37d22554d44d5669d8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 13:17:09 +0100 Subject: [PATCH 33/35] fix(cli): allow empty messages array in import response --- packages/opencode/src/cli/cmd/import.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index f79d3b5e2a..d9b99ed366 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -63,7 +63,7 @@ export const ImportCommand = cmd({ const data = await response.json() - if (!data.info || !data.messages || !Array.isArray(data.messages) || data.messages.length === 0) { + if (!data.info || !data.messages || !Array.isArray(data.messages)) { process.stdout.write(`Share not found: ${id}`) process.stdout.write(EOL) return From 1bb41afc7278a6e3b27eb1ff16c0d7bf753967c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 14:10:10 +0100 Subject: [PATCH 34/35] refactor(share): extract ingest queue into module Move per-session debounce/retry ingest logic into a reusable IngestQueue helper and wire ShareNext to use it, including an auth error callback to clear cached credentials. Add test coverage for coalescing, throttling, retry/backoff, and non-retryable failures. --- packages/opencode/src/share/ingest-queue.ts | 273 ++++++++++++++++++ packages/opencode/src/share/share-next.ts | 267 ++--------------- .../opencode/test/share/ingest-queue.test.ts | 248 ++++++++++++++++ 3 files changed, 542 insertions(+), 246 deletions(-) create mode 100644 packages/opencode/src/share/ingest-queue.ts create mode 100644 packages/opencode/test/share/ingest-queue.test.ts diff --git a/packages/opencode/src/share/ingest-queue.ts b/packages/opencode/src/share/ingest-queue.ts new file mode 100644 index 0000000000..9f62880baf --- /dev/null +++ b/packages/opencode/src/share/ingest-queue.ts @@ -0,0 +1,273 @@ +import { ulid } from "ulid" +import type * as SDK from "@kilocode/sdk/v2" + +export namespace IngestQueue { + export type Client = { + url: string + fetch: (input: RequestInfo | URL, init?: RequestInit) => Promise + } + + export type Data = + | { + type: "session" + data: SDK.Session + } + | { + type: "message" + data: SDK.Message + } + | { + type: "part" + data: SDK.Part + } + | { + type: "session_diff" + data: SDK.FileDiff[] + } + | { + type: "model" + data: SDK.Model[] + } + + type Share = { + ingestPath: string + } + + type Timer = ReturnType + + export type Options = { + getShare: (sessionId: string) => Promise + getClient: () => Promise + onAuthError?: () => void + log: { + error: (message: string, data: Record) => void + } + now?: () => number + setTimeout?: (fn: () => void, ms: number) => Timer + clearTimeout?: (timer: Timer) => void + } + + export function create(options: Options) { + // Per-session debounce/flush queue. + // + // The share ingest endpoint is updated very frequently (streaming message parts, diffs, etc.). + // To avoid spamming the server, we coalesce updates and flush at most once per ~1s per session. + // + // `due` is the earliest time we should flush; it is also used to respect backoff when retries are + // active. A later `due` always wins over an earlier one. + const queue = new Map }>() + + // Per-session retry state. + // + // We keep retry logic intentionally simple and local: + // - Only retry a small set of transient errors (network, 429, 5xx, etc.) + // - Use exponential backoff with a small max budget to prevent infinite loops/log spam + // - Store `until` so sync() can avoid scheduling a flush before backoff expires + const retry = new Map() + + const now = options.now ?? (() => Date.now()) + const set = options.setTimeout ?? ((fn, ms) => setTimeout(fn, ms)) + const clear = options.clearTimeout ?? ((timer) => clearTimeout(timer)) + + function retryable(status: number) { + // Retry only statuses that are likely transient. + if (status === 408) return true + if (status === 409) return true + if (status === 425) return true + if (status === 429) return true + if (status >= 500) return true + return false + } + + function backoff(count: number) { + // Exponential backoff capped to keep the system responsive. + const clamped = Math.min(count, 6) + return Math.min(60_000, 1_000 * 2 ** (clamped - 1)) + } + + function id(value: unknown) { + if (!value) return undefined + if (typeof value !== "object") return undefined + if (!("id" in value)) return undefined + const result = (value as { id?: unknown }).id + if (typeof result === "string" && result.length > 0) return result + return undefined + } + + function key(item: Data) { + // Stable keys are important so updates for the same entity collapse to a single queued item. + // If we can't derive a stable key, we fall back to a random key (ulid) so the item is still sent. + if (item.type === "session") return "session" + if (item.type === "session_diff") return "session_diff" + + if (item.type === "message") { + const value = id(item.data) + return value ? `message:${value}` : ulid() + } + + if (item.type === "part") { + const value = id(item.data) + return value ? `part:${value}` : ulid() + } + + const models = item.data + .map((m) => `${m.providerID}:${m.id}`) + .sort() + .join(",") + return models.length > 0 ? `model:${models}` : ulid() + } + + function schedule(sessionId: string, due: number, data: Map) { + const existing = queue.get(sessionId) + if (existing) { + // Don't reschedule if an earlier flush is already planned. + // We only move the flush later (e.g., to respect backoff). + if (existing.due >= due) return + clear(existing.timeout) + } + + const wait = Math.max(0, due - now()) + const timeout = set(() => { + void flush(sessionId) + }, wait) + queue.set(sessionId, { timeout, due, data }) + } + + function enqueue(sessionId: string, items: Data[], mode: "overwrite" | "fill", due: number) { + const existing = queue.get(sessionId) + if (existing) { + for (const item of items) { + const k = key(item) + // overwrite: normal event updates (newer data should win) + // fill: retry requeue (never clobber newer updates that arrived while a flush was in-flight) + if (mode === "fill" && existing.data.has(k)) continue + existing.data.set(k, item) + } + schedule(sessionId, due, existing.data) + return + } + + const data = new Map() + for (const item of items) { + data.set(key(item), item) + } + + schedule(sessionId, due, data) + } + + async function flush(sessionId: string) { + // Flush is scheduled by sync() and sends the currently queued payload. + // + // Note: we delete the queue entry before the network call so that new incoming events can start + // a fresh debounce window immediately. + const queued = queue.get(sessionId) + if (!queued) return + + clear(queued.timeout) + queue.delete(sessionId) + + const items = Array.from(queued.data.values()) + + try { + const share = await options.getShare(sessionId).catch(() => undefined) + if (!share) return + + const client = await options.getClient() + if (!client) return + + const response = await client + .fetch(`${client.url}${share.ingestPath}`, { + method: "POST", + body: JSON.stringify({ + data: items, + }), + }) + .catch(() => undefined) + + if (!response) { + // Network failures are assumed transient; retry with backoff and a small budget. + const count = (retry.get(sessionId)?.count ?? 0) + 1 + if (count > 6) { + options.log.error("share sync failed", { sessionId, error: "retry budget exceeded" }) + retry.delete(sessionId) + return + } + + const delay = backoff(count) + retry.set(sessionId, { count, until: now() + delay }) + options.log.error("share sync failed", { sessionId, error: "network", retryInMs: delay }) + enqueue(sessionId, items, "fill", now() + delay) + return + } + + if (response.ok) { + retry.delete(sessionId) + return + } + + if (response.status === 401 || response.status === 403) { + // Non-retryable until credentials are fixed. + options.onAuthError?.() + options.log.error("share sync failed", { + sessionId, + status: response.status, + statusText: response.statusText, + }) + retry.delete(sessionId) + return + } + + if (!retryable(response.status)) { + // Permanent-ish failures (eg. 404 due to bad ingestPath) should not loop forever. + options.log.error("share sync failed", { + sessionId, + status: response.status, + statusText: response.statusText, + }) + retry.delete(sessionId) + return + } + + const current = retry.get(sessionId) + const count = (current?.count ?? 0) + 1 + if (count > 6) { + options.log.error("share sync failed", { sessionId, error: "retry budget exceeded" }) + retry.delete(sessionId) + return + } + + const delay = backoff(count) + retry.set(sessionId, { count, until: now() + delay }) + options.log.error("share sync failed", { + sessionId, + status: response.status, + statusText: response.statusText, + retryInMs: delay, + }) + enqueue(sessionId, items, "fill", now() + delay) + } catch (error) { + options.log.error("share sync failed", { sessionId, error }) + } + } + + async function sync(sessionId: string, data: Data[]) { + // sync() is called by event handlers and is intentionally cheap: + // - If sharing isn't configured (no token / disabled), we skip queueing. + // - Otherwise, merge into the pending queue entry. + // The next flush is scheduled ~1s after the first queued event (throttled), but never earlier + // than the current backoff window (if retries are active). + const client = await options.getClient() + if (!client) return + + const until = retry.get(sessionId)?.until ?? 0 + const base = queue.get(sessionId)?.due ?? now() + 1000 + const due = Math.max(base, until) + enqueue(sessionId, data, "overwrite", due) + } + + return { + sync, + flush, + } as const + } +} diff --git a/packages/opencode/src/share/share-next.ts b/packages/opencode/src/share/share-next.ts index 8f5260149c..f8f0a44ca6 100644 --- a/packages/opencode/src/share/share-next.ts +++ b/packages/opencode/src/share/share-next.ts @@ -1,12 +1,12 @@ // kilocode_change pretty much completely refactored - @iscekic for conflicts import { Bus } from "@/bus" -import { ulid } from "ulid" import { Provider } from "@/provider/provider" import { Session } from "@/session" import { MessageV2 } from "@/session/message-v2" import { Storage } from "@/storage/storage" import { Log } from "@/util/log" import { Auth } from "@/auth" +import { IngestQueue } from "@/share/ingest-queue" // kilocode_change import type * as SDK from "@kilocode/sdk/v2" /** @@ -96,6 +96,20 @@ export namespace ShareNext { } } + const ingest = IngestQueue.create({ + getShare: async (sessionId) => get(sessionId).catch(() => undefined), + getClient, + log, + onAuthError: () => { + // Non-retryable until credentials are fixed. + // Clearing caches prevents repeated use of a now-invalid token/client. + authCache.clear() + cache.value = undefined + cache.inflight = undefined + cache.at = 0 + }, + }) + const shareDisabled = process.env["KILO_DISABLE_SHARE"] === "true" || process.env["KILO_DISABLE_SHARE"] === "1" const ingestDisabled = process.env["KILO_DISABLE_SESSION_INGEST"] === "true" || process.env["KILO_DISABLE_SESSION_INGEST"] === "1" @@ -109,7 +123,7 @@ export namespace ShareNext { }) Bus.subscribe(Session.Event.Updated, async (evt) => { - await sync(evt.properties.info.id, [ + await ingest.sync(evt.properties.info.id, [ { type: "session", data: evt.properties.info, @@ -118,7 +132,7 @@ export namespace ShareNext { }) Bus.subscribe(MessageV2.Event.Updated, async (evt) => { - await sync(evt.properties.info.sessionID, [ + await ingest.sync(evt.properties.info.sessionID, [ { type: "message", data: evt.properties.info, @@ -126,7 +140,7 @@ export namespace ShareNext { ]) if (evt.properties.info.role === "user") { - await sync(evt.properties.info.sessionID, [ + await ingest.sync(evt.properties.info.sessionID, [ { type: "model", data: [ @@ -140,7 +154,7 @@ export namespace ShareNext { }) Bus.subscribe(MessageV2.Event.PartUpdated, async (evt) => { - await sync(evt.properties.part.sessionID, [ + await ingest.sync(evt.properties.part.sessionID, [ { type: "part", data: evt.properties.part, @@ -149,7 +163,7 @@ export namespace ShareNext { }) Bus.subscribe(Session.Event.Diff, async (evt) => { - await sync(evt.properties.sessionID, [ + await ingest.sync(evt.properties.sessionID, [ { type: "session_diff", data: evt.properties.diff, @@ -271,245 +285,6 @@ export namespace ShareNext { }>(["session_share", sessionId]) } - type Data = - | { - type: "session" - data: SDK.Session - } - | { - type: "message" - data: SDK.Message - } - | { - type: "part" - data: SDK.Part - } - | { - type: "session_diff" - data: SDK.FileDiff[] - } - | { - type: "model" - data: SDK.Model[] - } - - // Per-session debounce/flush queue. - // - // The share ingest endpoint is updated very frequently (streaming message parts, diffs, etc.). - // To avoid spamming the server, we coalesce updates and flush at most once per ~1s per session. - // - // `due` is the earliest time we should flush; it is also used to respect backoff when retries are - // active. A later `due` always wins over an earlier one. - const queue = new Map }>() - - // Per-session retry state. - // - // We keep retry logic intentionally simple and local: - // - Only retry a small set of transient errors (network, 429, 5xx, etc.) - // - Use exponential backoff with a small max budget to prevent infinite loops/log spam - // - Store `until` so sync() can avoid scheduling a flush before backoff expires - const retry = new Map() - - function retryable(status: number) { - // Retry only statuses that are likely transient. - if (status === 408) return true - if (status === 409) return true - if (status === 425) return true - if (status === 429) return true - if (status >= 500) return true - return false - } - - function backoff(count: number) { - // Exponential backoff capped to keep the system responsive. - const clamped = Math.min(count, 6) - return Math.min(60_000, 1_000 * 2 ** (clamped - 1)) - } - - function id(value: unknown) { - if (!value) return undefined - if (typeof value !== "object") return undefined - if (!("id" in value)) return undefined - const result = (value as { id?: unknown }).id - if (typeof result === "string" && result.length > 0) return result - return undefined - } - - function key(item: Data) { - // Stable keys are important so updates for the same entity collapse to a single queued item. - // If we can't derive a stable key, we fall back to a random key (ulid) so the item is still sent. - if (item.type === "session") return "session" - if (item.type === "session_diff") return "session_diff" - - if (item.type === "message") { - const value = id(item.data) - return value ? `message:${value}` : ulid() - } - - if (item.type === "part") { - const value = id(item.data) - return value ? `part:${value}` : ulid() - } - - const models = item.data - .map((m) => `${m.providerID}:${m.id}`) - .sort() - .join(",") - return models.length > 0 ? `model:${models}` : ulid() - } - - function schedule(sessionId: string, due: number, data: Map) { - const existing = queue.get(sessionId) - if (existing) { - // Don't reschedule if an earlier flush is already planned. - // We only move the flush later (e.g., to respect backoff). - if (existing.due >= due) return - clearTimeout(existing.timeout) - } - - const wait = Math.max(0, due - Date.now()) - const timeout = setTimeout(() => flush(sessionId), wait) - queue.set(sessionId, { timeout, due, data }) - } - - function enqueue(sessionId: string, items: Data[], mode: "overwrite" | "fill", due: number) { - const existing = queue.get(sessionId) - if (existing) { - for (const item of items) { - const k = key(item) - // overwrite: normal event updates (newer data should win) - // fill: retry requeue (never clobber newer updates that arrived while a flush was in-flight) - if (mode === "fill" && existing.data.has(k)) continue - existing.data.set(k, item) - } - schedule(sessionId, due, existing.data) - return - } - - const data = new Map() - for (const item of items) { - data.set(key(item), item) - } - - schedule(sessionId, due, data) - } - - function flush(sessionId: string) { - // Flush is scheduled by sync() and sends the currently queued payload. - // - // Note: we delete the queue entry before the network call so that new incoming events can start - // a fresh debounce window immediately. - void (async () => { - const queued = queue.get(sessionId) - if (!queued) return - - clearTimeout(queued.timeout) - queue.delete(sessionId) - - const items = Array.from(queued.data.values()) - - try { - const share = await get(sessionId).catch(() => undefined) - if (!share) return - - const client = await getClient() - if (!client) return - - const response = await client - .fetch(`${client.url}${share.ingestPath}`, { - method: "POST", - body: JSON.stringify({ - data: items, - }), - }) - .catch(() => undefined) - - if (!response) { - // Network failures are assumed transient; retry with backoff and a small budget. - const count = (retry.get(sessionId)?.count ?? 0) + 1 - if (count > 6) { - log.error("share sync failed", { sessionId, error: "retry budget exceeded" }) - retry.delete(sessionId) - return - } - - const delay = backoff(count) - retry.set(sessionId, { count, until: Date.now() + delay }) - log.error("share sync failed", { sessionId, error: "network", retryInMs: delay }) - enqueue(sessionId, items, "fill", Date.now() + delay) - return - } - - if (response.ok) { - retry.delete(sessionId) - return - } - - if (response.status === 401 || response.status === 403) { - // Non-retryable until credentials are fixed. - // Clearing caches prevents repeated use of a now-invalid token/client. - authCache.clear() - cache.value = undefined - cache.inflight = undefined - cache.at = 0 - log.error("share sync failed", { - sessionId, - status: response.status, - statusText: response.statusText, - }) - retry.delete(sessionId) - return - } - - if (!retryable(response.status)) { - // Permanent-ish failures (eg. 404 due to bad ingestPath) should not loop forever. - log.error("share sync failed", { - sessionId, - status: response.status, - statusText: response.statusText, - }) - retry.delete(sessionId) - return - } - - const current = retry.get(sessionId) - const count = (current?.count ?? 0) + 1 - if (count > 6) { - log.error("share sync failed", { sessionId, error: "retry budget exceeded" }) - retry.delete(sessionId) - return - } - - const delay = backoff(count) - retry.set(sessionId, { count, until: Date.now() + delay }) - log.error("share sync failed", { - sessionId, - status: response.status, - statusText: response.statusText, - retryInMs: delay, - }) - enqueue(sessionId, items, "fill", Date.now() + delay) - } catch (error) { - log.error("share sync failed", { sessionId, error }) - } - })() - } - - async function sync(sessionId: string, data: Data[]) { - // sync() is called by event handlers and is intentionally cheap: - // - If sharing isn't configured (no token / disabled), we skip queueing. - // - Otherwise, merge into the pending queue entry. - // The next flush is scheduled ~1s after the first queued event (throttled), but never earlier - // than the current backoff window (if retries are active). - const client = await getClient() - if (!client) return - - const until = retry.get(sessionId)?.until ?? 0 - const base = queue.get(sessionId)?.due ?? Date.now() + 1000 - const due = Math.max(base, until) - enqueue(sessionId, data, "overwrite", due) - } - export async function remove(sessionId: string) { const client = await getClient() if (!client) return @@ -555,7 +330,7 @@ export namespace ShareNext { .map((m) => Provider.getModel(m.providerID, m.modelID).then((m) => m)), ) - await sync(sessionId, [ + await ingest.sync(sessionId, [ { type: "session", data: session, diff --git a/packages/opencode/test/share/ingest-queue.test.ts b/packages/opencode/test/share/ingest-queue.test.ts new file mode 100644 index 0000000000..2e0a9470c5 --- /dev/null +++ b/packages/opencode/test/share/ingest-queue.test.ts @@ -0,0 +1,248 @@ +import { describe, expect, test, beforeEach } from "bun:test" +import { IngestQueue } from "../../src/share/ingest-queue" + +function scheduler(now: () => number) { + const tasks = new Map void }>() + let next = 1 + + const setTimeout = (fn: () => void, ms: number) => { + const id = next + next += 1 + tasks.set(id, { at: now() + ms, fn }) + return id as unknown as ReturnType + } + + const clearTimeout = (timer: ReturnType) => { + tasks.delete(timer as unknown as number) + } + + const run = () => { + const due = Array.from(tasks.entries()) + .filter(([, t]) => t.at <= now()) + .map(([id]) => id) + for (const id of due) { + const task = tasks.get(id) + tasks.delete(id) + task?.fn() + } + } + + const size = () => tasks.size + + const nextAt = () => { + const at = Array.from(tasks.values()) + .map((t) => t.at) + .sort((a, b) => a - b)[0] + return at + } + + return { + setTimeout, + clearTimeout, + run, + size, + nextAt, + } as const +} + +describe("share ingest queue", () => { + const clock = { + now: 0, + } + + beforeEach(() => { + clock.now = 0 + }) + + test("throttles flush scheduling: later sync does not reschedule", async () => { + const calls: unknown[] = [] + const sched = scheduler(() => clock.now) + + const q = IngestQueue.create({ + now: () => clock.now, + setTimeout: sched.setTimeout, + clearTimeout: sched.clearTimeout, + log: { error: () => {} }, + getShare: async () => ({ ingestPath: "/ingest" }), + getClient: async () => ({ + url: "https://ingest.test", + fetch: async (_input, init) => { + calls.push(JSON.parse((init?.body as string) ?? "{}")) + return new Response("{}", { status: 200 }) + }, + }), + }) + + await q.sync("s1", [{ type: "session", data: { id: "s1", v: 1 } as any }]) + expect(sched.size()).toBe(1) + + clock.now = 900 + await q.sync("s1", [{ type: "session", data: { id: "s1", v: 2 } as any }]) + expect(sched.size()).toBe(1) + + clock.now = 1000 + sched.run() + await Bun.sleep(0) + expect(calls.length).toBe(1) + expect((calls[0] as any).data[0].data.v).toBe(2) + }) + + test("coalesces same-key updates and sends latest", async () => { + const sent: unknown[] = [] + const sched = scheduler(() => clock.now) + + const q = IngestQueue.create({ + now: () => clock.now, + setTimeout: sched.setTimeout, + clearTimeout: sched.clearTimeout, + log: { error: () => {} }, + getShare: async () => ({ ingestPath: "/ingest" }), + getClient: async () => ({ + url: "https://ingest.test", + fetch: async (_input, init) => { + sent.push(JSON.parse((init?.body as string) ?? "{}")) + return new Response("{}", { status: 200 }) + }, + }), + }) + + await q.sync("s2", [{ type: "session", data: { id: "s2", v: 1 } as any }]) + clock.now = 100 + await q.sync("s2", [{ type: "session", data: { id: "s2", v: 2 } as any }]) + + clock.now = 1000 + sched.run() + await Bun.sleep(0) + expect(sent.length).toBe(1) + expect((sent[0] as any).data.length).toBe(1) + expect((sent[0] as any).data[0].data.v).toBe(2) + }) + + test("network failure retries and fill preserves newer updates", async () => { + const sent: unknown[] = [] + const sched = scheduler(() => clock.now) + let attempt = 0 + + const q = IngestQueue.create({ + now: () => clock.now, + setTimeout: sched.setTimeout, + clearTimeout: sched.clearTimeout, + log: { error: () => {} }, + getShare: async () => ({ ingestPath: "/ingest" }), + getClient: async () => ({ + url: "https://ingest.test", + fetch: async (_input, init) => { + attempt += 1 + if (attempt === 1) throw new Error("network") + sent.push(JSON.parse((init?.body as string) ?? "{}")) + return new Response("{}", { status: 200 }) + }, + }), + }) + + await q.sync("s3", [{ type: "session", data: { id: "s3", v: 1 } as any }]) + + clock.now = 1000 + sched.run() // attempt 1 -> network fail -> requeue due at 2000 + await Bun.sleep(0) + + clock.now = 1500 + await q.sync("s3", [{ type: "session", data: { id: "s3", v: 2 } as any }]) + + clock.now = 2000 + sched.run() // attempt 2 -> ok + await Bun.sleep(0) + expect(sent.length).toBe(1) + expect((sent[0] as any).data[0].data.v).toBe(2) + }) + + test("404 does not requeue", async () => { + const sched = scheduler(() => clock.now) + const q = IngestQueue.create({ + now: () => clock.now, + setTimeout: sched.setTimeout, + clearTimeout: sched.clearTimeout, + log: { error: () => {} }, + getShare: async () => ({ ingestPath: "/ingest" }), + getClient: async () => ({ + url: "https://ingest.test", + fetch: async () => new Response("{}", { status: 404 }), + }), + }) + + await q.sync("s4", [{ type: "session", data: { id: "s4" } as any }]) + clock.now = 1000 + sched.run() + await Bun.sleep(0) + expect(sched.size()).toBe(0) + }) + + test("401 triggers auth error handler and does not requeue", async () => { + const sched = scheduler(() => clock.now) + let cleared = false + + const q = IngestQueue.create({ + now: () => clock.now, + setTimeout: sched.setTimeout, + clearTimeout: sched.clearTimeout, + log: { error: () => {} }, + onAuthError: () => { + cleared = true + }, + getShare: async () => ({ ingestPath: "/ingest" }), + getClient: async () => ({ + url: "https://ingest.test", + fetch: async () => new Response("{}", { status: 401 }), + }), + }) + + await q.sync("s5", [{ type: "session", data: { id: "s5" } as any }]) + clock.now = 1000 + sched.run() + await Bun.sleep(0) + expect(cleared).toBe(true) + expect(sched.size()).toBe(0) + }) + + test("retry budget exceeded stops requeueing", async () => { + const errors: Record[] = [] + const sched = scheduler(() => clock.now) + let attempts = 0 + + const q = IngestQueue.create({ + now: () => clock.now, + setTimeout: sched.setTimeout, + clearTimeout: sched.clearTimeout, + log: { + error: (_message, data) => { + errors.push(data) + }, + }, + getShare: async () => ({ ingestPath: "/ingest" }), + getClient: async () => ({ + url: "https://ingest.test", + fetch: async () => { + attempts += 1 + throw new Error("network") + }, + }), + }) + + await q.sync("s6", [{ type: "session", data: { id: "s6" } as any }]) + expect(sched.size()).toBe(1) + + for (const n of [1, 2, 3, 4, 5, 6, 7]) { + const at = sched.nextAt() + expect(typeof at).toBe("number") + + clock.now = at ?? 0 + sched.run() + await Bun.sleep(0) + + expect(attempts).toBe(n) + expect(sched.size()).toBe(n < 7 ? 1 : 0) + } + + expect(errors.some((e) => e.error === "retry budget exceeded")).toBe(true) + }) +}) From e99573595e946a9361b9bbe81435adcb252223df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Thu, 29 Jan 2026 18:38:33 +0100 Subject: [PATCH 35/35] chore: annotate files with kilocode_change markers --- .github/workflows/test.yml | 12 ++++++------ packages/app/script/e2e-local.ts | 4 ++-- packages/opencode/src/cli/cmd/import.ts | 4 +++- packages/opencode/src/project/bootstrap.ts | 4 ++-- packages/opencode/src/session/index.ts | 6 +++--- packages/opencode/src/share/ingest-queue.ts | 1 + packages/opencode/test/share/ingest-queue.test.ts | 1 + 7 files changed, 18 insertions(+), 14 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b0b37264de..61d7d0a965 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -69,8 +69,8 @@ jobs: env: KILO_API_KEY: ${{ secrets.KILO_API_KEY }} KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }} - KILO_DISABLE_SHARE: "true" - KILO_DISABLE_SESSION_INGEST: "true" + KILO_DISABLE_SHARE: "true" # kilocode_change + KILO_DISABLE_SESSION_INGEST: "true" # kilocode_change OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" @@ -91,8 +91,8 @@ jobs: env: KILO_API_KEY: ${{ secrets.KILO_API_KEY }} KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }} - KILO_DISABLE_SHARE: "true" - KILO_DISABLE_SESSION_INGEST: "true" + KILO_DISABLE_SHARE: "true" # kilocode_change + KILO_DISABLE_SESSION_INGEST: "true" # kilocode_change OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" @@ -119,8 +119,8 @@ jobs: CI: true KILO_API_KEY: ${{ secrets.KILO_API_KEY }} KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }} - KILO_DISABLE_SHARE: "true" - KILO_DISABLE_SESSION_INGEST: "true" + KILO_DISABLE_SHARE: "true" # kilocode_change + KILO_DISABLE_SESSION_INGEST: "true" # kilocode_change OPENCODE_DISABLE_LSP_DOWNLOAD: "true" OPENCODE_DISABLE_DEFAULT_PLUGINS: "true" OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true" diff --git a/packages/app/script/e2e-local.ts b/packages/app/script/e2e-local.ts index 809e43f6db..9c6dd6c8d4 100644 --- a/packages/app/script/e2e-local.ts +++ b/packages/app/script/e2e-local.ts @@ -58,8 +58,8 @@ const sandbox = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-e2e-")) const serverEnv = { ...process.env, - KILO_DISABLE_SHARE: "true", - KILO_DISABLE_SESSION_INGEST: "true", + KILO_DISABLE_SHARE: "true", // kilocode_change + KILO_DISABLE_SESSION_INGEST: "true", // kilocode_change OPENCODE_DISABLE_LSP_DOWNLOAD: "true", OPENCODE_DISABLE_DEFAULT_PLUGINS: "true", OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: "true", diff --git a/packages/opencode/src/cli/cmd/import.ts b/packages/opencode/src/cli/cmd/import.ts index d9b99ed366..65c6295fb1 100644 --- a/packages/opencode/src/cli/cmd/import.ts +++ b/packages/opencode/src/cli/cmd/import.ts @@ -11,7 +11,7 @@ export const ImportCommand = cmd({ describe: "import session data from JSON file or URL", builder: (yargs: Argv) => { return yargs.positional("file", { - describe: "path to JSON file or app.kilo.ai share URL", + describe: "path to JSON file or app.kilo.ai share URL", // kilocode_change type: "string", demandOption: true, }) @@ -31,6 +31,7 @@ export const ImportCommand = cmd({ const isUrl = args.file.startsWith("http://") || args.file.startsWith("https://") if (isUrl) { + // kilocode_change start const url = (() => { try { return new URL(args.file) @@ -70,6 +71,7 @@ export const ImportCommand = cmd({ } exportData = data + // kilocode_change end } else { const file = Bun.file(args.file) exportData = await file.json().catch(() => {}) diff --git a/packages/opencode/src/project/bootstrap.ts b/packages/opencode/src/project/bootstrap.ts index a2be3733f8..a99ca6962b 100644 --- a/packages/opencode/src/project/bootstrap.ts +++ b/packages/opencode/src/project/bootstrap.ts @@ -9,14 +9,14 @@ import { Command } from "../command" import { Instance } from "./instance" import { Vcs } from "./vcs" import { Log } from "@/util/log" -import { ShareNext } from "@/share/share-next" +import { ShareNext } from "@/share/share-next" // kilocode_change import { Snapshot } from "../snapshot" import { Truncate } from "../tool/truncation" export async function InstanceBootstrap() { Log.Default.info("bootstrapping", { directory: Instance.directory }) await Plugin.init() - ShareNext.init() + ShareNext.init() // kilocode_change Format.init() await LSP.init() FileWatcher.init() diff --git a/packages/opencode/src/session/index.ts b/packages/opencode/src/session/index.ts index 1c87327f74..6572e20a20 100644 --- a/packages/opencode/src/session/index.ts +++ b/packages/opencode/src/session/index.ts @@ -254,7 +254,7 @@ export namespace Session { throw new Error("Sharing is disabled in configuration") } const { ShareNext } = await import("@/share/share-next") - const share = await ShareNext.share(id) + const share = await ShareNext.share(id) // kilocode_change await update( id, (draft) => { @@ -270,7 +270,7 @@ export namespace Session { export const unshare = fn(Identifier.schema("session"), async (id) => { // Use ShareNext to remove the share (same as share function uses ShareNext to create) const { ShareNext } = await import("@/share/share-next") - await ShareNext.unshare(id) + await ShareNext.unshare(id) // kilocode_change await update( id, (draft) => { @@ -341,7 +341,7 @@ export namespace Session { await remove(child.id) } const { ShareNext } = await import("@/share/share-next") - await ShareNext.remove(sessionID).catch(() => {}) + await ShareNext.remove(sessionID).catch(() => {}) // kilocode_change for (const msg of await Storage.list(["message", sessionID])) { for (const part of await Storage.list(["part", msg.at(-1)!])) { await Storage.remove(part) diff --git a/packages/opencode/src/share/ingest-queue.ts b/packages/opencode/src/share/ingest-queue.ts index 9f62880baf..61200db69c 100644 --- a/packages/opencode/src/share/ingest-queue.ts +++ b/packages/opencode/src/share/ingest-queue.ts @@ -1,3 +1,4 @@ +// kilocode_change - new file import { ulid } from "ulid" import type * as SDK from "@kilocode/sdk/v2" diff --git a/packages/opencode/test/share/ingest-queue.test.ts b/packages/opencode/test/share/ingest-queue.test.ts index 2e0a9470c5..5203f7dbc2 100644 --- a/packages/opencode/test/share/ingest-queue.test.ts +++ b/packages/opencode/test/share/ingest-queue.test.ts @@ -1,3 +1,4 @@ +// kilocode_change - new file import { describe, expect, test, beforeEach } from "bun:test" import { IngestQueue } from "../../src/share/ingest-queue"