Merge pull request #54 from Kilo-Org/add-session-sync

.github/workflows/test.yml

@@ -69,7 +69,8 @@ jobs:
         env:
           KILO_API_KEY: ${{ secrets.KILO_API_KEY }}
           KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }}
-          OPENCODE_DISABLE_SHARE: "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"
@@ -90,7 +91,8 @@ jobs:
         env:
           KILO_API_KEY: ${{ secrets.KILO_API_KEY }}
           KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }}
-          OPENCODE_DISABLE_SHARE: "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"
@@ -117,7 +119,8 @@ jobs:
           CI: true
           KILO_API_KEY: ${{ secrets.KILO_API_KEY }}
           KILO_ORG_ID: ${{ secrets.KILO_ORG_ID }}
-          OPENCODE_DISABLE_SHARE: "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"

packages/app/script/e2e-local.ts

@@ -58,7 +58,8 @@ const sandbox = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-e2e-"))
 
 const serverEnv = {
   ...process.env,
-  OPENCODE_DISABLE_SHARE: "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",

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", // kilocode_change
       type: "string",
       demandOption: true,
     })
@@ -31,15 +31,30 @@ 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_-]+)/)
-        if (!urlMatch) {
-          process.stdout.write(`Invalid URL format. Expected: https://opncd.ai/share/<slug>`)
+        // kilocode_change start
+        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/<id>`)
           process.stdout.write(EOL)
           return
         }
 
-        const slug = urlMatch[1]
-        const response = await fetch(`https://opncd.ai/api/share/${slug}`)
+        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/<id>`)
+          process.stdout.write(EOL)
+          return
+        }
+
+        const response = await fetch(`https://ingest.kilosessions.ai/session/${encodeURIComponent(id)}`)
 
         if (!response.ok) {
           process.stdout.write(`Failed to fetch share data: ${response.statusText}`)
@@ -49,22 +64,14 @@ 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}`)
+        if (!data.info || !data.messages || !Array.isArray(data.messages)) {
+          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
+        // kilocode_change end
       } else {
         const file = Bun.file(args.file)
         exportData = await file.json().catch(() => {})

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"
@@ -10,15 +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()
-  Share.init()
-  ShareNext.init()
+  ShareNext.init() // kilocode_change
   Format.init()
   await LSP.init()
   FileWatcher.init()

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) // 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.remove(id)
+    await ShareNext.unshare(id) // kilocode_change
     await update(
       id,
       (draft) => {
@@ -340,7 +340,8 @@ export namespace Session {
       for (const child of await children(sessionID)) {
         await remove(child.id)
       }
-      await unshare(sessionID).catch(() => {})
+      const { ShareNext } = await import("@/share/share-next")
+      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)

packages/opencode/src/share/ingest-queue.ts

@@ -0,0 +1,274 @@
+// kilocode_change - new file
+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<Response>
+  }
+
+  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<typeof setTimeout>
+
+  export type Options = {
+    getShare: (sessionId: string) => Promise<Share | undefined>
+    getClient: () => Promise<Client | undefined>
+    onAuthError?: () => void
+    log: {
+      error: (message: string, data: Record<string, unknown>) => 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<string, { timeout: Timer; due: number; data: Map<string, Data> }>()
+
+    // 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<string, { count: number; until: number }>()
+
+    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<string, Data>) {
+      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<string, Data>()
+      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
+  }
+}

packages/opencode/src/share/share-next.ts

@@ -1,41 +1,146 @@
+// 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"
 import { MessageV2 } from "@/session/message-v2"
 import { Storage } from "@/storage/storage"
 import { Log } from "@/util/log"
-import type * as SDK from "@kilocode/sdk/v2" // kilocode_change
+import { Auth } from "@/auth"
+import { IngestQueue } from "@/share/ingest-queue" // kilocode_change
+import type * as SDK from "@kilocode/sdk/v2"
 
+/**
+ * Even though this is called "share-next", this is where we handle session stuff.
+ */
 export namespace ShareNext {
   const log = Log.create({ service: "share-next" })
 
-  async function url() {
-    return Config.get().then((x) => x.enterprise?.url ?? "https://opncd.ai")
+  const authCache = new Map<string, { valid: boolean }>()
+
+  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)
+
+    // 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
   }
 
-  const disabled = process.env["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_DISABLE_SHARE"] === "1"
+  export async function kilocodeToken() {
+    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
+  }
+
+  type Client = {
+    url: string
+    fetch: (input: RequestInfo | URL, init?: RequestInit) => Promise<Response>
+  }
+
+  const cache = {
+    at: 0,
+    value: undefined as Client | undefined,
+    inflight: undefined as Promise<Client | undefined> | undefined,
+  }
+
+  async function getClient(): Promise<Client | 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
+
+    cache.at = now
+    cache.inflight = (async () => {
+      const token = await kilocodeToken()
+      if (!token) return undefined
+
+      const valid = await authValid(token)
+      if (!valid) return undefined
+
+      const base = "https://ingest.kilosessions.ai"
+      const baseHeaders: Record<string, string> = {
+        "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)),
+      }
+    })()
+
+    try {
+      cache.value = await cache.inflight
+      return cache.value
+    } finally {
+      cache.inflight = undefined
+    }
+  }
+
+  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"
 
   export async function init() {
-    if (disabled) return
+    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 }))
+    })
+
     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,
         },
       ])
     })
+
     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,
         },
       ])
+
       if (evt.properties.info.role === "user") {
-        await sync(evt.properties.info.sessionID, [
+        await ingest.sync(evt.properties.info.sessionID, [
           {
             type: "model",
             data: [
@@ -47,16 +152,18 @@ 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,
         },
       ])
     })
+
     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,
@@ -65,119 +172,165 @@ 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`, {
+  export async function create(sessionId: string) {
+    const client = await getClient()
+    if (!client) return { id: "", ingestPath: "" }
+
+    log.info("creating session", { sessionId })
+
+    const response = await client.fetch(`${client.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; url: string; secret: string })
-    await Storage.write(["session_share", sessionID], result)
-    fullSync(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)
+
+    void fullSync(sessionId).catch((error) => log.error("share full sync failed", { sessionId, error }))
+
     return result
   }
 
-  function get(sessionID: string) {
-    return Storage.read<{
-      id: string
-      secret: string
-      url: string
-    }>(["session_share", sessionID])
+  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)")
+    }
+
+    const client = await getClient()
+    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 response = await client.fetch(`${client.url}/api/session/${encodeURIComponent(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 url = `https://app.kilo.ai/s/${result.public_id}`
+
+    await Storage.write(["session_share", sessionId], {
+      ...current,
+      url,
+    })
+
+    return { url }
   }
 
-  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[]
-      }
+  export async function unshare(sessionId: string) {
+    if (ingestDisabled) {
+      throw new Error("Session ingest is disabled (KILO_DISABLE_SESSION_INGEST=1)")
+    }
 
-  const queue = new Map<string, { timeout: NodeJS.Timeout; data: Map<string, Data> }>()
-  async function sync(sessionID: string, data: Data[]) {
-    if (disabled) 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)
-      }
+    if (shareDisabled) {
+      throw new Error("Unshare is disabled (KILO_DISABLE_SHARE=1)")
+    }
+
+    const client = await getClient()
+    if (!client) {
+      throw new Error("Unable to unshare session: no Kilo credentials found. Run `kilo auth login`.")
+    }
+
+    log.info("unsharing", { sessionId })
+
+    const response = await client.fetch(`${client.url}/api/session/${encodeURIComponent(sessionId)}/unshare`, {
+      method: "POST",
+      body: JSON.stringify({ sessionId }),
+    })
+
+    if (!response.ok) {
+      throw new Error(`Unable to unshare session ${sessionId}: ${response.status} ${response.statusText}`)
+    }
+
+    const current = await get(sessionId).catch(() => undefined)
+    if (!current) return
+
+    const next = {
+      ...current,
+    }
+    delete next.url
+
+    await Storage.write(["session_share", sessionId], next)
+  }
+
+  function get(sessionId: string) {
+    return Storage.read<{
+      id: string
+      url?: string
+      ingestPath: string
+    }>(["session_share", sessionId])
+  }
+
+  export async function remove(sessionId: string) {
+    const client = await getClient()
+    if (!client) return
+
+    log.info("removing share", { sessionId })
+
+    const share = await get(sessionId)
+    if (!share) return
+
+    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
     }
 
-    const dataMap = new Map<string, Data>()
-    for (const item of data) {
-      dataMap.set("id" in item ? (item.id as string) : ulid(), item)
+    if (!response.ok) {
+      log.error("share remove failed", {
+        sessionId,
+        status: response.status,
+        statusText: response.statusText,
+      })
+      return
     }
 
-    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
-
-      await fetch(`${await url()}/api/share/${share.id}/sync`, {
-        method: "POST",
-        headers: {
-          "Content-Type": "application/json",
-        },
-        body: JSON.stringify({
-          secret: share.secret,
-          data: Array.from(queued.data.values()),
-        }),
-      })
-    }, 1000)
-    queue.set(sessionID, { timeout, data: dataMap })
+    await Storage.remove(["session_share", sessionId])
   }
 
-  export async function remove(sessionID: string) {
-    if (disabled) return
-    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])
-  }
+  async function fullSync(sessionId: string) {
+    log.info("full sync", { 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 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 ingest.sync(sessionId, [
       {
         type: "session",
         data: session,

packages/opencode/src/share/share.ts

@@ -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<void> = Promise.resolve()
-  const pending = new Map<string, any>()
-
-  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["OPENCODE_DISABLE_SHARE"] === "true" || process.env["OPENCODE_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())
-  }
-}

packages/opencode/test/share/ingest-queue.test.ts

@@ -0,0 +1,249 @@
+// kilocode_change - new file
+import { describe, expect, test, beforeEach } from "bun:test"
+import { IngestQueue } from "../../src/share/ingest-queue"
+
+function scheduler(now: () => number) {
+  const tasks = new Map<number, { at: number; fn: () => 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<typeof globalThis.setTimeout>
+  }
+
+  const clearTimeout = (timer: ReturnType<typeof globalThis.setTimeout>) => {
+    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<string, unknown>[] = []
+    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)
+  })
+})