From fe01f53e2bddbadc51736ff81dcc2e022fe6f27f Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Wed, 22 Jul 2026 15:48:29 +0200 Subject: [PATCH] Cloud Agent - Add kilo cloud command for running asynchronous cloud agent tasks (#11849) --- .changeset/run-cloud-agent.md | 5 + packages/kilo-docs/lychee.toml | 1 + .../markdoc/partials/cli-commands-table.md | 1 + .../code-with-ai/platforms/cli-reference.md | 70 +++ packages/kilo-docs/source-links.md | 3 + .../opencode/src/kilocode/cli/cmd/cloud.ts | 134 +++++ packages/opencode/src/kilocode/cli/setup.ts | 2 + packages/opencode/src/kilocode/cloud/auth.ts | 66 +++ .../opencode/src/kilocode/cloud/catalog.ts | 173 ++++++ .../opencode/src/kilocode/cloud/commands.ts | 295 ++++++++++ .../opencode/src/kilocode/cloud/contracts.ts | 194 +++++++ .../opencode/src/kilocode/cloud/defaults.ts | 138 +++++ .../opencode/src/kilocode/cloud/errors.ts | 24 + packages/opencode/src/kilocode/cloud/http.ts | 67 +++ .../opencode/src/kilocode/cloud/message-id.ts | 11 + .../opencode/src/kilocode/cloud/origin.ts | 70 +++ .../opencode/src/kilocode/cloud/repository.ts | 265 +++++++++ .../src/kilocode/cloud/response-json.ts | 45 ++ .../src/kilocode/cloud/stream-ticket.ts | 96 ++++ packages/opencode/src/kilocode/cloud/trpc.ts | 158 +++++ .../src/kilocode/cloud/websocket-stream.ts | 194 +++++++ packages/opencode/src/kilocode/commands.ts | 6 +- .../test/kilocode/cli-shutdown.test.ts | 2 + .../test/kilocode/cloud/command.test.ts | 540 ++++++++++++++++++ .../test/kilocode/cloud/defaults.test.ts | 436 ++++++++++++++ .../test/kilocode/cloud/repository.test.ts | 119 ++++ .../test/kilocode/cloud/stream-ticket.test.ts | 134 +++++ .../test/kilocode/cloud/transport.test.ts | 260 +++++++++ .../kilocode/cloud/websocket-stream.test.ts | 281 +++++++++ packages/opencode/test/kilocode/help.test.ts | 26 + 30 files changed, 3814 insertions(+), 2 deletions(-) create mode 100644 .changeset/run-cloud-agent.md create mode 100644 packages/opencode/src/kilocode/cli/cmd/cloud.ts create mode 100644 packages/opencode/src/kilocode/cloud/auth.ts create mode 100644 packages/opencode/src/kilocode/cloud/catalog.ts create mode 100644 packages/opencode/src/kilocode/cloud/commands.ts create mode 100644 packages/opencode/src/kilocode/cloud/contracts.ts create mode 100644 packages/opencode/src/kilocode/cloud/defaults.ts create mode 100644 packages/opencode/src/kilocode/cloud/errors.ts create mode 100644 packages/opencode/src/kilocode/cloud/http.ts create mode 100644 packages/opencode/src/kilocode/cloud/message-id.ts create mode 100644 packages/opencode/src/kilocode/cloud/origin.ts create mode 100644 packages/opencode/src/kilocode/cloud/repository.ts create mode 100644 packages/opencode/src/kilocode/cloud/response-json.ts create mode 100644 packages/opencode/src/kilocode/cloud/stream-ticket.ts create mode 100644 packages/opencode/src/kilocode/cloud/trpc.ts create mode 100644 packages/opencode/src/kilocode/cloud/websocket-stream.ts create mode 100644 packages/opencode/test/kilocode/cloud/command.test.ts create mode 100644 packages/opencode/test/kilocode/cloud/defaults.test.ts create mode 100644 packages/opencode/test/kilocode/cloud/repository.test.ts create mode 100644 packages/opencode/test/kilocode/cloud/stream-ticket.test.ts create mode 100644 packages/opencode/test/kilocode/cloud/transport.test.ts create mode 100644 packages/opencode/test/kilocode/cloud/websocket-stream.test.ts diff --git a/.changeset/run-cloud-agent.md b/.changeset/run-cloud-agent.md new file mode 100644 index 0000000000..4453c853d7 --- /dev/null +++ b/.changeset/run-cloud-agent.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": minor +--- + +Run asynchronous Cloud Agent tasks with repository, model, mode, and organization defaults through `kilo cloud`. Add `--stream` to `kilo cloud start` to print admission output and then stream WebSocket events as JSONL until completion or inactivity ends the stream. diff --git a/packages/kilo-docs/lychee.toml b/packages/kilo-docs/lychee.toml index 4b6d5113d2..73dc8a11fb 100644 --- a/packages/kilo-docs/lychee.toml +++ b/packages/kilo-docs/lychee.toml @@ -50,6 +50,7 @@ exclude = [ '^https?://vercel\.link/', # API base URL, returns 404 when fetched directly '^https?://api\.apertis\.ai/v1/?$', + '^https?://cloud-agent-next\.kilosessions\.ai/?$', # Redirects to authenticated Google Cloud console '^https?://console\.cloud\.google\.com', # Google AI Studio API keys page redirects to Google sign-in diff --git a/packages/kilo-docs/markdoc/partials/cli-commands-table.md b/packages/kilo-docs/markdoc/partials/cli-commands-table.md index 65bfbedcf2..e8e42e3921 100644 --- a/packages/kilo-docs/markdoc/partials/cli-commands-table.md +++ b/packages/kilo-docs/markdoc/partials/cli-commands-table.md @@ -26,6 +26,7 @@ | `kilo remote` | enable remote connection for real-time session relay | | `kilo daemon` | manage the local kilo daemon | | `kilo console` | open or stop the local Kilo Console | +| `kilo cloud` | run Cloud Agent tasks | | `kilo db` | database tools | | `kilo config` | configuration tools | | `kilo plugin ` | install plugin and update config | diff --git a/packages/kilo-docs/pages/code-with-ai/platforms/cli-reference.md b/packages/kilo-docs/pages/code-with-ai/platforms/cli-reference.md index 8a2d4178c2..144f545ca7 100644 --- a/packages/kilo-docs/pages/code-with-ai/platforms/cli-reference.md +++ b/packages/kilo-docs/pages/code-with-ai/platforms/cli-reference.md @@ -968,6 +968,76 @@ Options: --json print daemon details as JSON [boolean] ``` +## kilo cloud + +``` +run Cloud Agent tasks + +Commands: + kilo cloud start start a Cloud Agent task + kilo cloud send send a follow-up prompt to a Cloud Agent task + kilo cloud status show Cloud Agent task status + kilo cloud result show a Cloud Agent task result + +Options: + --help Show help [boolean] + --version Show version number [boolean] +``` + +### kilo cloud start + +``` +start a Cloud Agent task + +Options: + --help Show help [boolean] + --version Show version number [boolean] + --prompt prompt for the Cloud Agent [string] [required] + --repo repository shorthand or URL [string] + --repo-type repository provider type [string] [choices: "github", "gitlab", "git"] + --branch repository branch [string] + --model Cloud Agent model [string] + --mode Cloud Agent mode [string] + --org-id Kilo organization ID [string] + --stream connect to the WebSocket stream and print events as JSONL [boolean] +``` + +### kilo cloud send + +``` +send a follow-up prompt to a Cloud Agent task + +Options: + --help Show help [boolean] + --version Show version number [boolean] + --session-id Cloud Agent session ID [string] [required] + --prompt follow-up prompt for the Cloud Agent [string] [required] +``` + +### kilo cloud status + +``` +show Cloud Agent task status + +Options: + --help Show help [boolean] + --version Show version number [boolean] + --session-id Cloud Agent session ID [string] [required] + --message-id Cloud Agent message ID [string] [required] +``` + +### kilo cloud result + +``` +show a Cloud Agent task result + +Options: + --help Show help [boolean] + --version Show version number [boolean] + --session-id Cloud Agent session ID [string] [required] + --message-id Cloud Agent message ID [string] [required] +``` + ## kilo db ``` diff --git a/packages/kilo-docs/source-links.md b/packages/kilo-docs/source-links.md index 74e3dcec06..e9b3199d5e 100644 --- a/packages/kilo-docs/source-links.md +++ b/packages/kilo-docs/source-links.md @@ -50,6 +50,8 @@ - +- + - - @@ -106,6 +108,7 @@ - + - diff --git a/packages/opencode/src/kilocode/cli/cmd/cloud.ts b/packages/opencode/src/kilocode/cli/cmd/cloud.ts new file mode 100644 index 0000000000..61b4e75bae --- /dev/null +++ b/packages/opencode/src/kilocode/cli/cmd/cloud.ts @@ -0,0 +1,134 @@ +import type { Argv } from "yargs" +import { Effect } from "effect" +import { cmd } from "@/cli/cmd/cmd" +import { effectCmd } from "@/cli/effect-cmd" +import { CloudCommands } from "@/kilocode/cloud/commands" + +export const CloudStartCommand = effectCmd({ + command: "start", + describe: "start a Cloud Agent task", + builder: (yargs) => + yargs + .option("prompt", { + type: "string", + demandOption: true, + describe: "prompt for the Cloud Agent", + }) + .option("repo", { + type: "string", + describe: "repository shorthand or URL", + }) + .option("repo-type", { + type: "string", + choices: ["github", "gitlab", "git"] as const, + describe: "repository provider type", + }) + .option("branch", { + type: "string", + describe: "repository branch", + }) + .option("model", { + type: "string", + describe: "Cloud Agent model", + }) + .option("mode", { + type: "string", + describe: "Cloud Agent mode", + }) + .option("org-id", { + type: "string", + describe: "Kilo organization ID", + }) + .option("stream", { + type: "boolean", + describe: "connect to the WebSocket stream and print events as JSONL", + }), + handler: Effect.fn("Cli.cloud.start")(function* (args) { + yield* CloudCommands.start({ + prompt: args.prompt, + ...(args.repo === undefined ? {} : { repo: args.repo }), + ...(args.repoType === undefined ? {} : { repoType: args.repoType }), + ...(args.branch === undefined ? {} : { branch: args.branch }), + ...(args.model === undefined ? {} : { model: args.model }), + ...(args.mode === undefined ? {} : { mode: args.mode }), + ...(args.orgId === undefined ? {} : { orgID: args.orgId }), + ...(args.stream === undefined ? {} : { stream: args.stream }), + }) + }), +}) + +export const CloudSendCommand = effectCmd({ + command: "send", + describe: "send a follow-up prompt to a Cloud Agent task", + instance: false, + builder: (yargs) => + yargs + .option("session-id", { + type: "string", + demandOption: true, + describe: "Cloud Agent session ID", + }) + .option("prompt", { + type: "string", + demandOption: true, + describe: "follow-up prompt for the Cloud Agent", + }), + handler: Effect.fn("Cli.cloud.send")(function* (args) { + yield* CloudCommands.send({ sessionID: args.sessionId, prompt: args.prompt }) + }), +}) + +export const CloudStatusCommand = effectCmd({ + command: "status", + describe: "show Cloud Agent task status", + instance: false, + builder: (yargs) => + yargs + .option("session-id", { + type: "string", + demandOption: true, + describe: "Cloud Agent session ID", + }) + .option("message-id", { + type: "string", + demandOption: true, + describe: "Cloud Agent message ID", + }), + handler: Effect.fn("Cli.cloud.status")(function* (args) { + yield* CloudCommands.status({ sessionID: args.sessionId, messageID: args.messageId }) + }), +}) + +export const CloudResultCommand = effectCmd({ + command: "result", + describe: "show a Cloud Agent task result", + instance: false, + builder: (yargs) => + yargs + .option("session-id", { + type: "string", + demandOption: true, + describe: "Cloud Agent session ID", + }) + .option("message-id", { + type: "string", + demandOption: true, + describe: "Cloud Agent message ID", + }), + handler: Effect.fn("Cli.cloud.result")(function* (args) { + yield* CloudCommands.result({ sessionID: args.sessionId, messageID: args.messageId }) + }), +}) + +export const CloudCommand = cmd({ + command: "cloud", + describe: "run Cloud Agent tasks", + builder: (yargs: Argv) => + yargs + .command(CloudStartCommand) + .command(CloudSendCommand) + .command(CloudStatusCommand) + .command(CloudResultCommand) + .demandCommand(), + async handler() {}, +}) diff --git a/packages/opencode/src/kilocode/cli/setup.ts b/packages/opencode/src/kilocode/cli/setup.ts index 65fa60fe0c..2e41e1190f 100644 --- a/packages/opencode/src/kilocode/cli/setup.ts +++ b/packages/opencode/src/kilocode/cli/setup.ts @@ -12,6 +12,7 @@ import { SessionExport } from "@/kilocode/session-export" import { KiloShutdown } from "@/kilocode/cli/shutdown" import { createHelpCommand } from "@/kilocode/help-command" import { KiloConsoleCommand } from "@/kilocode/cli/cmd/console" +import { CloudCommand } from "@/kilocode/cli/cmd/cloud" import { RollCallCommand } from "@/kilocode/cli/cmd/roll-call" import { ProfileCommand } from "@/kilocode/cli/cmd/profile" import { DaemonCommand } from "@/kilocode/cli/cmd/daemon" @@ -32,6 +33,7 @@ export namespace KiloCli { export function register(cli: Argv): Argv { cli .command(KiloConsoleCommand) + .command(CloudCommand) .command(RollCallCommand) .command(ProfileCommand) .command(RemoteCommand) diff --git a/packages/opencode/src/kilocode/cloud/auth.ts b/packages/opencode/src/kilocode/cloud/auth.ts new file mode 100644 index 0000000000..442725e0a2 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/auth.ts @@ -0,0 +1,66 @@ +import { Auth } from "@/auth" +import { Effect, Redacted, Schema } from "effect" +import z from "zod" + +export namespace CloudAuth { + export type Environment = Readonly> + + export interface Input { + readonly orgID?: string + readonly env?: Environment + } + + export interface Resolved { + readonly token: Redacted.Redacted + readonly organizationID?: string + } + + export class ResolutionError extends Schema.TaggedErrorClass()("CloudAuthResolutionError", { + kind: Schema.Literals(["missing", "organization"]), + message: Schema.String, + }) {} + + const Uuid = z.uuid() + + const credentials = Effect.fn("CloudAuth.credentials")(function* (env: Environment) { + const service = yield* Auth.Service + const info = yield* service.get("kilo") + const stored = info?.type === "api" ? info.key.trim() : info?.type === "oauth" ? info.access.trim() : undefined + const fallback = env.KILO_API_KEY?.trim() + const value = stored || fallback + if (!value) { + return yield* Effect.fail( + new ResolutionError({ + kind: "missing", + message: "Kilo credentials are required; run `kilo auth login`", + }), + ) + } + return { token: Redacted.make(value), accountID: info?.type === "oauth" ? info.accountId : undefined } + }) + + export const token = Effect.fn("CloudAuth.token")(function* (env: Environment = process.env) { + return (yield* credentials(env)).token + }) + + export const resolve = Effect.fn("CloudAuth.resolve")(function* (input: Input = {}) { + const env = input.env ?? process.env + const auth = yield* credentials(env) + const explicit = input.orgID + const setting = env.KILO_ORG_ID?.trim() + const organizationID = explicit !== undefined ? explicit : setting || auth.accountID + if (organizationID !== undefined && !Uuid.safeParse(organizationID).success) { + return yield* Effect.fail( + new ResolutionError({ + kind: "organization", + message: "Kilo organization ID must be a valid UUID", + }), + ) + } + + return { + token: auth.token, + ...(organizationID ? { organizationID } : {}), + } satisfies Resolved + }) +} diff --git a/packages/opencode/src/kilocode/cloud/catalog.ts b/packages/opencode/src/kilocode/cloud/catalog.ts new file mode 100644 index 0000000000..7cb75d1ad9 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/catalog.ts @@ -0,0 +1,173 @@ +import { + buildKiloHeaders, + DEFAULT_KILO_API_URL, + getDefaultHeaders, + getKiloUrlFromToken, + resolveKiloOpenRouterBaseUrl, +} from "@kilocode/kilo-gateway" +import { Context, Effect, Layer, Redacted, Schema } from "effect" +import z from "zod" +import type { CloudAuth } from "./auth" +import { parseServiceOrigin } from "./origin" +import { readBoundedJson } from "./response-json" + +export namespace CloudCatalog { + const TIMEOUT = 10_000 + + const Models = z.object({ + data: z.array( + z.object({ + id: z.string().min(1).max(255), + architecture: z + .object({ + output_modalities: z.array(z.string()).nullish(), + }) + .optional(), + supported_parameters: z.array(z.string()).optional(), + }), + ), + }) + const Defaults = z.object({ + defaultModel: z.string().min(1).max(255), + }) + + export type Environment = Readonly> + export type Fetch = (request: Request) => Promise + + export interface Options { + readonly env?: Environment + readonly fetch?: Fetch + } + + export interface Input extends CloudAuth.Resolved {} + + export class CatalogError extends Schema.TaggedErrorClass()("CloudCatalogError", { + kind: Schema.Literals(["auth", "network", "schema", "http"]), + status: Schema.optional(Schema.Number), + message: Schema.String, + }) {} + + export interface Interface { + readonly models: (input: Input) => Effect.Effect + readonly defaultModel: (input: Input) => Effect.Effect + } + + export class Service extends Context.Service()("@kilocode/CloudCatalog") {} + + export const layer = (options: Options = {}) => { + const fetcher = options.fetch ?? ((request: Request) => globalThis.fetch(request)) + const env = options.env ?? process.env + + const request = Effect.fn("CloudCatalog.request")(function* (url: string, input: Input, schema: z.ZodType) { + const req = yield* Effect.try({ + try: () => + new Request(url, { + headers: { + ...getDefaultHeaders(), + ...buildKiloHeaders( + undefined, + input.organizationID ? { kilocodeOrganizationId: input.organizationID } : undefined, + ), + "X-KILOCODE-FEATURE": "kilo-cli", + Authorization: `Bearer ${Redacted.value(input.token)}`, + }, + redirect: "error", + signal: AbortSignal.timeout(TIMEOUT), + }), + catch: () => + new CatalogError({ + kind: "schema", + message: "Kilo catalog URL is invalid", + }), + }) + const response = yield* Effect.tryPromise({ + try: () => fetcher(req), + catch: () => + new CatalogError({ + kind: "network", + message: "Unable to reach the Kilo model catalog", + }), + }) + if (!response.ok) { + const kind = response.status === 401 || response.status === 403 ? "auth" : "http" + return yield* Effect.fail( + new CatalogError({ + kind, + status: response.status, + message: + kind === "auth" + ? "Kilo credentials or organization were rejected by the model catalog" + : "The Kilo model catalog is unavailable", + }), + ) + } + + const body = yield* Effect.tryPromise({ + try: () => readBoundedJson(response), + catch: () => + new CatalogError({ + kind: "schema", + message: "The Kilo model catalog returned an invalid response", + }), + }) + const parsed = schema.safeParse(body) + if (!parsed.success) { + return yield* Effect.fail( + new CatalogError({ + kind: "schema", + message: "The Kilo model catalog returned an invalid response", + }), + ) + } + return parsed.data + }) + + const base = Effect.fn("CloudCatalog.base")(function* (input: Input) { + const raw = env.KILO_API_URL?.trim() + const fallback = raw || DEFAULT_KILO_API_URL + const value = getKiloUrlFromToken(fallback, Redacted.value(input.token)) + return yield* Effect.try({ + try: () => { + const url = new URL(resolveKiloOpenRouterBaseUrl({ baseURL: value })) + parseServiceOrigin(url.origin, { allowHttpLoopback: !!raw || value !== fallback }) + if (url.username !== "" || url.password !== "") throw new Error("Catalog URL credentials are not allowed") + return url + }, + catch: () => + new CatalogError({ + kind: "schema", + message: "Kilo catalog URL must be secure", + }), + }) + }) + + const models = Effect.fn("CloudCatalog.models")(function* (input: Input) { + const root = yield* base(input) + const path = input.organizationID + ? `../organizations/${encodeURIComponent(input.organizationID)}/models` + : "models" + const result = yield* request(new URL(path, root).toString(), input, Models) + return [ + ...new Set( + result.data + .filter( + (model) => + !model.architecture?.output_modalities?.includes("image") && + model.supported_parameters?.includes("tools"), + ) + .map((model) => model.id), + ), + ] + }) + + const defaultModel = Effect.fn("CloudCatalog.defaultModel")(function* (input: Input) { + const root = yield* base(input) + const path = input.organizationID + ? `../organizations/${encodeURIComponent(input.organizationID)}/defaults` + : "../defaults" + return (yield* request(new URL(path, root).toString(), input, Defaults)).defaultModel + }) + + return Layer.succeed(Service, Service.of({ models, defaultModel })) + } +} diff --git a/packages/opencode/src/kilocode/cloud/commands.ts b/packages/opencode/src/kilocode/cloud/commands.ts new file mode 100644 index 0000000000..4f81664bfe --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/commands.ts @@ -0,0 +1,295 @@ +import { CliError, fail } from "@/cli/effect-cmd" +import { Effect, Layer, Redacted } from "effect" +import { CloudAuth } from "./auth" +import { CloudCatalog } from "./catalog" +import { + CloudAgentSessionIdSchema, + MessageIdSchema, + PromptSchema, + projectStatus, + resultExitCode, + type AgentResultExitCode, + type AgentStartRequest, + type AgentStartResponse, + type GetMessageResultInput, +} from "./contracts" +import { CloudDefaults } from "./defaults" +import { CloudError } from "./errors" +import { resolveCloudAgentOrigin, resolveWebAppOrigin, type ServiceOrigin } from "./origin" +import { CloudRepository } from "./repository" +import { createStreamTicketClient, type StreamTicketClient } from "./stream-ticket" +import { createCloudAgentClient, type AgentClient } from "./trpc" +import { streamAgentEvents, type StreamAgentEventsOptions } from "./websocket-stream" + +export namespace CloudCommands { + export interface StartInput { + readonly cwd?: string + readonly prompt: string + readonly repo?: string + readonly repoType?: CloudRepository.RepositoryType + readonly branch?: string + readonly model?: string + readonly mode?: string + readonly orgID?: string + readonly stream?: boolean + } + + export interface SendInput { + readonly sessionID: string + readonly prompt: string + } + + export interface LookupInput { + readonly sessionID: string + readonly messageID: string + } + + export interface ClientOptions { + readonly origin: ServiceOrigin + readonly apiKey: string + } + + export type ClientFactory = (options: ClientOptions) => AgentClient + + export interface Deps { + readonly env?: CloudAuth.Environment + readonly make?: ClientFactory + readonly write?: (text: string) => unknown + readonly exit?: (code: AgentResultExitCode) => unknown + readonly createStreamTicketClient?: (options: { + readonly origin: ServiceOrigin + readonly apiKey: string + }) => StreamTicketClient + readonly streamAgentEvents?: (options: StreamAgentEventsOptions) => Promise + } + + const factory: ClientFactory = (options) => createCloudAgentClient(options) + + function diagnostic(error: unknown) { + if (error instanceof CliError) return error.message + if (error instanceof CloudError) return error.message + if (error instanceof CloudAuth.ResolutionError) return error.message + if (error instanceof CloudCatalog.CatalogError) return error.message + if (error instanceof CloudDefaults.ResolutionError) return error.message + if (error instanceof CloudRepository.InvalidRepositoryError) return error.message + if (error instanceof CloudRepository.InvalidBranchError) return error.message + if (error instanceof CloudRepository.NotWorktreeError) return error.message + if (error instanceof CloudRepository.NoRemoteError) return error.message + if (error instanceof CloudRepository.AmbiguousRemoteError) return error.message + if (error instanceof CloudRepository.DiscoveryError) return error.message + return "Cloud Agent command failed" + } + + function clean(effect: Effect.Effect): Effect.Effect { + return effect.pipe(Effect.catch((error) => fail(diagnostic(error)))) + } + + function print(value: unknown, deps: Deps, admission?: "start" | "send") { + const text = JSON.stringify(value) + "\n" + const message = admission + ? `Cloud Agent ${admission} was admitted but output could not be written; do not retry automatically` + : "Unable to write Cloud Agent output" + const write = deps.write + if (write) { + return Effect.tryPromise({ + try: async () => { + await write(text) + }, + catch: () => new CloudError(message), + }) + } + return Effect.tryPromise({ + try: () => + new Promise((resolve, reject) => { + process.stdout.write(text, (error) => (error ? reject(error) : resolve())) + }), + catch: () => new CloudError(message), + }) + } + + async function writeLine(text: string, deps: Deps) { + const line = `${text}\n` + const write = deps.write + if (write) { + await write(line) + return + } + await new Promise((resolve, reject) => { + process.stdout.write(line, (error) => (error ? reject(error) : resolve())) + }) + } + + function notice(error: unknown, deps: Deps) { + return print({ streamEventType: "error", data: { message: diagnostic(error) } }, deps).pipe( + Effect.catch(() => Effect.void), + ) + } + + function attempt(run: () => Promise) { + return Effect.tryPromise({ + try: run, + catch: (error) => (error instanceof CloudError ? error : new CloudError("Cloud Agent request failed")), + }) + } + + const auth = Effect.fn("CloudCommands.auth")(function* (deps: Deps) { + const env = deps.env ?? process.env + return yield* CloudAuth.token({ KILO_API_KEY: env.KILO_API_KEY }) + }) + + function client(token: CloudAuth.Resolved["token"], deps: Deps) { + return Effect.try({ + try: () => { + const env = deps.env ?? process.env + const origin = resolveCloudAgentOrigin(env) + return (deps.make ?? factory)({ origin, apiKey: Redacted.value(token) }) + }, + catch: (error) => (error instanceof CloudError ? error : new CloudError("Cloud Agent client setup failed")), + }) + } + + const lookup = Effect.fn("CloudCommands.lookup")(function* (input: LookupInput, deps: Deps) { + if (!CloudAgentSessionIdSchema.safeParse(input.sessionID).success) { + return yield* fail("Cloud Agent session ID is invalid") + } + if (!MessageIdSchema.safeParse(input.messageID).success) { + return yield* fail("Cloud Agent message ID is invalid") + } + + const token = yield* auth(deps) + const agent = yield* client(token, deps) + const request = { + cloudAgentSessionId: input.sessionID, + messageId: input.messageID, + } satisfies GetMessageResultInput + return yield* attempt(() => agent.getMessageResult(request)) + }) + + export const start = Effect.fn("CloudCommands.start")(function* (input: StartInput, deps: Deps = {}) { + return yield* clean( + Effect.gen(function* () { + if (!PromptSchema.safeParse(input.prompt).success) return yield* fail("Cloud Agent prompt is invalid") + + const env = deps.env ?? process.env + const defaults = yield* CloudDefaults.resolve({ + env, + ...(input.mode === undefined ? {} : { mode: input.mode }), + ...(input.model === undefined ? {} : { model: input.model }), + ...(input.orgID === undefined ? {} : { orgID: input.orgID }), + }).pipe(Effect.provide(Layer.mergeAll(CloudDefaults.modelStateLayer, CloudCatalog.layer({ env })))) + const repository = yield* CloudRepository.resolve({ + cwd: input.cwd ?? process.cwd(), + ...(input.repo === undefined ? {} : { repo: input.repo }), + ...(input.repoType === undefined ? {} : { type: input.repoType }), + ...(input.branch === undefined ? {} : { branch: input.branch }), + }) + const agent = yield* client(defaults.token, deps) + const request = { + message: { prompt: input.prompt }, + agent: { mode: defaults.mode, model: defaults.model }, + repository, + options: { + createdOnPlatform: "kilo-cli", + ...(defaults.organizationID ? { kilocodeOrganizationId: defaults.organizationID } : {}), + }, + } satisfies AgentStartRequest + const result = yield* attempt(() => agent.start(request)) + yield* print({ ...result, streamUrl: undefined }, deps, "start") + if (input.stream) { + yield* Effect.gen(function* () { + const streamUrl = yield* resolveStreamUrl(result, defaults, env, deps) + yield* Effect.tryPromise({ + try: () => + (deps.streamAgentEvents ?? streamAgentEvents)({ + streamUrl, + origin: resolveCloudAgentOrigin(env), + writeLine: (line) => writeLine(line, deps), + WebSocket: globalThis.WebSocket, + }), + catch: (error) => (error instanceof CloudError ? error : new CloudError("Cloud Agent stream failed")), + }) + }).pipe(Effect.catch((error) => notice(error, deps))) + } + return result + }), + ) + }) + + export const send = Effect.fn("CloudCommands.send")(function* (input: SendInput, deps: Deps = {}) { + return yield* clean( + Effect.gen(function* () { + if (!CloudAgentSessionIdSchema.safeParse(input.sessionID).success) { + return yield* fail("Cloud Agent session ID is invalid") + } + if (!PromptSchema.safeParse(input.prompt).success) return yield* fail("Cloud Agent prompt is invalid") + + const token = yield* auth(deps) + const agent = yield* client(token, deps) + const result = yield* attempt(() => + agent.send({ + cloudAgentSessionId: input.sessionID, + message: { prompt: input.prompt }, + }), + ) + yield* print({ ...result, streamUrl: undefined }, deps, "send") + return result + }), + ) + }) + + export const status = Effect.fn("CloudCommands.status")(function* (input: LookupInput, deps: Deps = {}) { + return yield* clean( + Effect.gen(function* () { + const result = projectStatus(yield* lookup(input, deps)) + yield* print(result, deps) + return result + }), + ) + }) + + export const result = Effect.fn("CloudCommands.result")(function* (input: LookupInput, deps: Deps = {}) { + return yield* clean( + Effect.gen(function* () { + const result = yield* lookup(input, deps) + const code = resultExitCode(result.status) + yield* print(result, deps) + yield* Effect.sync(() => (deps.exit ?? ((value: AgentResultExitCode) => (process.exitCode = value)))(code)) + return result + }), + ) + }) + + function resolveStreamUrl( + response: AgentStartResponse, + defaults: CloudDefaults.Resolved, + env: CloudAuth.Environment, + deps: Deps, + ): Effect.Effect { + if (response.streamUrl !== undefined) { + return Effect.succeed(response.streamUrl) + } + + const origin = resolveWebAppOrigin(env) + const ticketClient = (deps.createStreamTicketClient ?? createStreamTicketClient)({ + origin, + apiKey: Redacted.value(defaults.token), + }) + + return Effect.tryPromise({ + try: () => + ticketClient.fetchTicket({ + cloudAgentSessionId: response.cloudAgentSessionId, + ...(defaults.organizationID ? { organizationId: defaults.organizationID } : {}), + }), + catch: (error) => (error instanceof CloudError ? error : new CloudError("Unable to obtain stream ticket")), + }).pipe( + Effect.map((ticket) => { + const params = new URLSearchParams({ + cloudAgentSessionId: response.cloudAgentSessionId, + ticket: ticket.ticket, + }) + return `/stream?${params.toString()}` + }), + ) + } +} diff --git a/packages/opencode/src/kilocode/cloud/contracts.ts b/packages/opencode/src/kilocode/cloud/contracts.ts new file mode 100644 index 0000000000..a4716e547e --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/contracts.ts @@ -0,0 +1,194 @@ +import z from "zod" + +export const MessageIdSchema = z.string().regex(/^msg_[0-9a-f]{12}[0-9A-Za-z]{14}$/) +export const CloudAgentSessionIdSchema = z + .string() + .regex(/^agent_[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i) + +const BranchSchema = z + .string() + .min(1) + .max(255) + .regex(/^[a-zA-Z0-9._\-/]+$/) +export const ModelSchema = z + .string() + .min(1) + .max(255) + .regex(/^[a-zA-Z0-9._\-/:]+$/) +export const ModeSchema = z + .string() + .min(1) + .max(50) + .regex(/^[a-z][a-z0-9-]*$/) +const GithubRepoSchema = z + .string() + .regex(/^[a-zA-Z0-9_.-]+\/[a-zA-Z0-9_.-]+$/) + .refine((value) => value.split("/").every((part) => part !== "." && part !== "..")) +const GitUrlSchema = z + .string() + .url() + .refine((url) => url.startsWith("https://"), "Only HTTPS URLs are supported") + +export const RepositoryInputSchema = z.discriminatedUnion("type", [ + z.object({ type: z.literal("github"), repo: GithubRepoSchema, branch: BranchSchema.optional() }).strict(), + z.object({ type: z.literal("gitlab"), url: GitUrlSchema, branch: BranchSchema.optional() }).strict(), + z + .object({ + type: z.literal("git"), + url: GitUrlSchema, + token: z.string().optional(), + branch: BranchSchema.optional(), + }) + .strict(), +]) + +export const PromptSchema = z.string().min(1).max(100_000) +const StartMessageSchema = z.object({ prompt: PromptSchema, id: MessageIdSchema.optional() }).strict() +const SendMessageSchema = z.object({ prompt: PromptSchema, id: MessageIdSchema.nullish() }).strict() +const AgentSchema = z + .object({ + mode: ModeSchema, + model: ModelSchema, + variant: z + .string() + .max(50) + .regex(/^[a-zA-Z]+$/) + .optional(), + }) + .strict() + +export const AgentStartRequestSchema = z + .object({ + message: StartMessageSchema, + agent: AgentSchema, + repository: RepositoryInputSchema, + options: z + .object({ + createdOnPlatform: z.literal("kilo-cli"), + kilocodeOrganizationId: z.string().uuid().optional(), + }) + .strict(), + }) + .strict() + +export const AgentSendRequestSchema = z + .object({ cloudAgentSessionId: CloudAgentSessionIdSchema, message: SendMessageSchema }) + .strict() + +export const GetMessageResultInputSchema = z + .object({ cloudAgentSessionId: CloudAgentSessionIdSchema, messageId: MessageIdSchema }) + .strict() + +export const AgentStartResponseSchema = z.object({ + cloudAgentSessionId: CloudAgentSessionIdSchema, + kiloSessionId: z.string(), + messageId: MessageIdSchema, + delivery: z.string().min(1).max(50), + streamUrl: z.string().min(1).optional(), + wrapperRunId: z.string().optional(), +}) + +export const AgentSendResponseSchema = z.object({ + cloudAgentSessionId: CloudAgentSessionIdSchema, + status: z.literal("started"), + streamUrl: z.string().min(1), + messageId: MessageIdSchema, + delivery: z.string().min(1).max(50), + wrapperRunId: z.string().optional(), +}) + +const FailureStageSchema = z.string().min(1).max(100) +const FailureCodeSchema = z.string().min(1).max(100) +const FailureSubtypeSchema = z.string().min(1).max(100) + +export const SafeFailureSchema = z + .object({ + stage: FailureStageSchema.optional(), + code: FailureCodeSchema.optional(), + subtype: FailureSubtypeSchema.optional(), + attempts: z.number().int().nonnegative().optional(), + message: z.string().min(1).max(4_096).optional(), + retryable: z.boolean(), + }) + .refine((failure) => failure.subtype === undefined || failure.code === "workspace_setup_failed", { + message: "Workspace failure subtype requires workspace_setup_failed failure code", + path: ["subtype"], + }) + +export const GetMessageResultOutputSchema = z + .object({ + cloudAgentSessionId: CloudAgentSessionIdSchema, + messageId: MessageIdSchema, + status: z.enum(["queued", "running", "completed", "failed", "interrupted"]), + createdAt: z.number(), + queuedAt: z.number().optional(), + acceptedAt: z.number().optional(), + terminalAt: z.number().optional(), + completionSource: z.string().min(1).max(100).optional(), + failure: SafeFailureSchema.optional(), + gateResult: z.string().min(1).max(50).optional(), + assistant: z.object({ messageId: z.string(), text: z.string().optional() }).optional(), + }) + .superRefine((result, ctx) => { + const terminal = result.status === "completed" || result.status === "failed" || result.status === "interrupted" + if (result.status === "queued" && result.acceptedAt !== undefined) { + ctx.addIssue({ code: "custom", message: "Queued results cannot include acceptedAt", path: ["acceptedAt"] }) + } + if (!terminal && result.terminalAt !== undefined) { + ctx.addIssue({ code: "custom", message: "Active results cannot include terminalAt", path: ["terminalAt"] }) + } + if (!terminal && result.completionSource !== undefined) { + ctx.addIssue({ + code: "custom", + message: "Active results cannot include completionSource", + path: ["completionSource"], + }) + } + if (result.status !== "failed" && result.status !== "interrupted" && result.failure !== undefined) { + ctx.addIssue({ + code: "custom", + message: "Only failed or interrupted results can include failure details", + path: ["failure"], + }) + } + if (result.status !== "completed" && result.gateResult !== undefined) { + ctx.addIssue({ code: "custom", message: "Only completed results can include gateResult", path: ["gateResult"] }) + } + if (result.status !== "completed" && result.assistant !== undefined) { + ctx.addIssue({ + code: "custom", + message: "Only completed results can include an assistant response", + path: ["assistant"], + }) + } + }) + +export type RepositoryInput = z.infer +export type AgentStartRequest = z.infer +export type AgentSendRequest = z.infer +export type GetMessageResultInput = z.infer +export type AgentStartResponse = z.infer +export type AgentSendResponse = z.infer +export type MessageResult = z.infer +export type AgentStatus = Omit +export type AgentResultExitCode = 0 | 2 | 3 | 4 + +export interface Decoder { + parse(value: unknown): T +} + +export function projectStatus(result: MessageResult): AgentStatus { + const { assistant: _, ...status } = result + return status +} + +export function resultExitCode(status: MessageResult["status"]): AgentResultExitCode { + const codes = { + completed: 0, + queued: 2, + running: 2, + failed: 3, + interrupted: 4, + } as const + return codes[status] +} diff --git a/packages/opencode/src/kilocode/cloud/defaults.ts b/packages/opencode/src/kilocode/cloud/defaults.ts new file mode 100644 index 0000000000..514489fefc --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/defaults.ts @@ -0,0 +1,138 @@ +import { Agent } from "@/agent/agent" +import { Config } from "@/config/config" +import { KilocodeModelState } from "@/kilocode/config/model-state" +import { Context, Effect, Layer, Schema } from "effect" +import { CloudAuth } from "./auth" +import { CloudCatalog } from "./catalog" +import { ModelSchema, ModeSchema } from "./contracts" + +export namespace CloudDefaults { + const COMPATIBLE = new Set(["code", "plan", "debug", "orchestrator", "ask", "build", "architect"]) + + export type ModelStateInfo = KilocodeModelState.State + + export interface ModelStateInterface { + readonly get: () => Effect.Effect + } + + export class ModelState extends Context.Service()("@kilocode/CloudModelState") {} + + export const modelStateLayer = Layer.succeed( + ModelState, + ModelState.of({ get: () => Effect.promise(() => KilocodeModelState.get()) }), + ) + + export interface Input { + readonly env?: CloudAuth.Environment + readonly mode?: string + readonly model?: string + readonly orgID?: string + } + + export interface Resolved extends CloudAuth.Resolved { + readonly mode: string + readonly model: string + } + + export class ResolutionError extends Schema.TaggedErrorClass()("CloudDefaultsResolutionError", { + kind: Schema.Literals(["mode", "model"]), + message: Schema.String, + }) {} + + const mode = Effect.fn("CloudDefaults.mode")(function* (value: string | undefined, fallback: string | undefined) { + const service = yield* Agent.Service + if (value !== undefined) { + if (!ModeSchema.safeParse(value).success || !COMPATIBLE.has(value)) { + return yield* Effect.fail( + new ResolutionError({ + kind: "mode", + message: `Cloud Agent mode is unavailable: ${value}`, + }), + ) + } + const info = yield* service.get(value) + if (info && (info.mode === "subagent" || info.hidden === true)) { + return yield* Effect.fail( + new ResolutionError({ + kind: "mode", + message: `Cloud Agent mode is unavailable: ${value}`, + }), + ) + } + return { name: value, info } + } + + const info = yield* service.get(fallback ?? "code") + if (info && COMPATIBLE.has(info.name) && info.mode !== "subagent" && info.hidden !== true) { + return { name: info.name, info } + } + const base = yield* service.get("code") + if (base && (base.mode === "subagent" || base.hidden === true)) { + return yield* Effect.fail( + new ResolutionError({ + kind: "mode", + message: "Cloud Agent mode is unavailable: code", + }), + ) + } + return { name: "code", info: base } + }) + + function ref(value: { readonly providerID: string; readonly modelID: string } | undefined) { + return value ? `${value.providerID}/${value.modelID}` : undefined + } + + function normalize(value: string) { + return value.startsWith("kilo/") ? value.slice("kilo/".length) : value + } + + export const resolve = Effect.fn("CloudDefaults.resolve")(function* (input: Input = {}) { + const auth = yield* CloudAuth.resolve({ orgID: input.orgID, env: input.env }) + const config = yield* Config.Service + const cfg = yield* config.get() + const selected = yield* mode(input.mode, cfg.default_agent ?? undefined) + const states = yield* ModelState + const saved = yield* states.get() + const catalog = yield* CloudCatalog.Service + const available = new Set(yield* catalog.models(auth)) + + if (input.model !== undefined) { + const explicit = normalize(input.model) + if (!ModelSchema.safeParse(explicit).success || !available.has(explicit)) { + return yield* Effect.fail( + new ResolutionError({ + kind: "model", + message: `Cloud Agent model is unavailable: ${input.model}`, + }), + ) + } + return { ...auth, mode: selected.name, model: explicit } satisfies Resolved + } + + const candidates = [ + ref(selected.info?.model), + ref(saved.model[selected.name]), + cfg.model ?? undefined, + ...saved.recent.map(ref), + ] + for (const value of candidates) { + if (!value) continue + const candidate = normalize(value) + if (!ModelSchema.safeParse(candidate).success) continue + if (!available.has(candidate)) continue + return { ...auth, mode: selected.name, model: candidate } satisfies Resolved + } + + const fallback = normalize(yield* catalog.defaultModel(auth)) + if (ModelSchema.safeParse(fallback).success && available.has(fallback)) { + return { ...auth, mode: selected.name, model: fallback } satisfies Resolved + } + + return yield* Effect.fail( + new ResolutionError({ + kind: "model", + message: "The Kilo model catalog has no available default model", + }), + ) + }) +} diff --git a/packages/opencode/src/kilocode/cloud/errors.ts b/packages/opencode/src/kilocode/cloud/errors.ts new file mode 100644 index 0000000000..8045d533f0 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/errors.ts @@ -0,0 +1,24 @@ +export class CloudError extends Error { + constructor(message: string) { + super(message) + this.name = "CloudError" + } +} + +export class ServiceTransportError extends CloudError { + constructor() { + super("Unable to reach configured service") + this.name = "ServiceTransportError" + } +} + +export class ServiceRedirectError extends CloudError { + constructor() { + super("Service redirects are not allowed") + this.name = "ServiceRedirectError" + } +} + +export function ambiguousAdmissionError(procedure: "start" | "send") { + return new CloudError(`Cloud Agent ${procedure} outcome is unknown; do not retry automatically`) +} diff --git a/packages/opencode/src/kilocode/cloud/http.ts b/packages/opencode/src/kilocode/cloud/http.ts new file mode 100644 index 0000000000..db41099d67 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/http.ts @@ -0,0 +1,67 @@ +import { CloudError, ServiceRedirectError, ServiceTransportError } from "./errors" +import type { ServiceOrigin } from "./origin" + +export const DEFAULT_HTTP_TIMEOUT_MS = 30_000 + +export interface BearerRequest { + readonly method: "GET" | "POST" + readonly path: `/${string}` + readonly headers?: Readonly> + readonly body?: BodyInit +} + +export interface BearerHttpClient { + request(request: BearerRequest): Promise +} + +export interface BearerHttpClientOptions { + readonly origin: ServiceOrigin + readonly apiKey: string + readonly fetch?: typeof globalThis.fetch + readonly timeoutMs?: number +} + +export function createBearerHttpClient(options: BearerHttpClientOptions): BearerHttpClient { + const fetcher = options.fetch ?? globalThis.fetch + + return { + async request(request) { + const url = resolve(options.origin, request.path) + const headers = new Headers(request.headers) + if (headers.has("authorization")) { + throw new CloudError("Authorization headers are managed by Kilo") + } + headers.set("authorization", `Bearer ${options.apiKey}`) + + const response = await fetcher(url, { + method: request.method, + headers, + ...(request.body === undefined ? {} : { body: request.body }), + redirect: "manual", + signal: AbortSignal.timeout(options.timeoutMs ?? DEFAULT_HTTP_TIMEOUT_MS), + }).catch(() => { + throw new ServiceTransportError() + }) + + if ((response.status >= 300 && response.status < 400) || response.redirected) { + await discard(response) + throw new ServiceRedirectError() + } + return response + }, + } +} + +async function discard(response: Response) { + await response.body?.cancel().catch(() => undefined) +} + +function resolve(origin: ServiceOrigin, path: `/${string}`) { + if (!path.startsWith("/") || path.startsWith("//") || path.includes("\\") || path.includes("#")) { + throw new CloudError("Unsafe service request path") + } + + const url = new URL(path, origin) + if (url.origin !== origin) throw new CloudError("Unsafe service request path") + return url +} diff --git a/packages/opencode/src/kilocode/cloud/message-id.ts b/packages/opencode/src/kilocode/cloud/message-id.ts new file mode 100644 index 0000000000..b337981a44 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/message-id.ts @@ -0,0 +1,11 @@ +const ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789" +const LENGTH = 14 + +export function createMessageId(now = Date.now()) { + const prefix = BigInt(now).toString(16).padStart(12, "0").slice(-12) + const suffix = Array.from( + crypto.getRandomValues(new Uint8Array(LENGTH)), + (byte) => ALPHABET[byte % ALPHABET.length], + ).join("") + return `msg_${prefix}${suffix}` +} diff --git a/packages/opencode/src/kilocode/cloud/origin.ts b/packages/opencode/src/kilocode/cloud/origin.ts new file mode 100644 index 0000000000..1db03bb029 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/origin.ts @@ -0,0 +1,70 @@ +import { CloudError } from "./errors" + +export const DEFAULT_CLOUD_AGENT_ORIGIN = "https://cloud-agent-next.kilosessions.ai" +export const DEFAULT_WEB_APP_ORIGIN = "https://kilo.ai" + +export type ServiceOrigin = string +export type CloudEnvironment = Readonly> + +export interface ParseServiceOriginOptions { + readonly allowHttpLoopback?: boolean +} + +export function parseServiceOrigin(value: string, options: ParseServiceOriginOptions = {}): ServiceOrigin { + if (!/^[a-zA-Z][a-zA-Z0-9+.-]*:\/\/[^/?#\\]+\/?$/.test(value) || value.includes("%") || /\p{Cc}/u.test(value)) { + throw new CloudError("Service URL must contain only an origin") + } + + const url = parse(value) + if ( + url.username !== "" || + url.password !== "" || + url.hostname.endsWith(".") || + (url.pathname !== "" && url.pathname !== "/") || + url.search !== "" || + url.hash !== "" + ) { + throw new CloudError("Service URL must contain only an origin") + } + + const secure = url.protocol === "https:" + const loopback = options.allowHttpLoopback === true && url.protocol === "http:" && isLoopback(url.hostname) + if (!secure && !loopback) { + throw new CloudError("Service URL must use HTTPS unless it is an explicit loopback override") + } + + return url.origin +} + +export function resolveCloudAgentOrigin(env: CloudEnvironment = process.env) { + const value = env.CLOUD_AGENT_NEXT_BASE_URL + return parseServiceOrigin(value ?? DEFAULT_CLOUD_AGENT_ORIGIN, { + allowHttpLoopback: value !== undefined, + }) +} + +export function resolveWebAppOrigin(env: CloudEnvironment = process.env) { + const value = env.KILO_WEB_APP_URL + return parseServiceOrigin(value ?? DEFAULT_WEB_APP_ORIGIN, { + allowHttpLoopback: value !== undefined, + }) +} + +function parse(value: string) { + try { + return new URL(value) + } catch { + throw new CloudError("Service URL must be a valid origin") + } +} + +function isLoopback(host: string) { + if (host === "localhost" || host === "[::1]") return true + const parts = host.split(".") + return parts.length === 4 && parts[0] === "127" && parts.every(isPart) +} + +function isPart(part: string) { + if (!/^(?:0|[1-9][0-9]{0,2})$/.test(part)) return false + return Number(part) <= 255 +} diff --git a/packages/opencode/src/kilocode/cloud/repository.ts b/packages/opencode/src/kilocode/cloud/repository.ts new file mode 100644 index 0000000000..9829a8ecc5 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/repository.ts @@ -0,0 +1,265 @@ +import { Effect, Schema } from "effect" +import { Git } from "@/git" + +export namespace CloudRepository { + export const RepositoryType = Schema.Literals(["github", "gitlab", "git"]) + export type RepositoryType = Schema.Schema.Type + + const GitHub = Schema.Struct({ + type: Schema.Literal("github"), + repo: Schema.String, + branch: Schema.optional(Schema.String), + }) + + const GitLab = Schema.Struct({ + type: Schema.Literal("gitlab"), + url: Schema.String, + branch: Schema.optional(Schema.String), + }) + + const GitRepository = Schema.Struct({ + type: Schema.Literal("git"), + url: Schema.String, + branch: Schema.optional(Schema.String), + }) + + export const Output = Schema.Union([GitHub, GitLab, GitRepository]) + export type Output = Schema.Schema.Type + + export type Input = { + readonly cwd: string + readonly repo?: string + readonly type?: RepositoryType + readonly branch?: string + } + + export type ParseInput = { + readonly repo: string + readonly type?: RepositoryType + readonly branch?: string + } + + export class InvalidRepositoryError extends Schema.TaggedErrorClass()( + "CloudRepositoryInvalidRepositoryError", + { message: Schema.String }, + ) {} + + export class InvalidBranchError extends Schema.TaggedErrorClass()( + "CloudRepositoryInvalidBranchError", + { message: Schema.String }, + ) {} + + export class NotWorktreeError extends Schema.TaggedErrorClass()("CloudRepositoryNotWorktreeError", { + message: Schema.String, + }) {} + + export class NoRemoteError extends Schema.TaggedErrorClass()("CloudRepositoryNoRemoteError", { + message: Schema.String, + }) {} + + export class AmbiguousRemoteError extends Schema.TaggedErrorClass()( + "CloudRepositoryAmbiguousRemoteError", + { message: Schema.String }, + ) {} + + export class DiscoveryError extends Schema.TaggedErrorClass()("CloudRepositoryDiscoveryError", { + message: Schema.String, + }) {} + + export type ParseError = InvalidRepositoryError | InvalidBranchError + export type Error = ParseError | NotWorktreeError | NoRemoteError | AmbiguousRemoteError | DiscoveryError + + const shorthand = /^[a-zA-Z0-9_.-]+\/[a-zA-Z0-9_.-]+$/ + const branch = /^[a-zA-Z0-9._\-/]+$/ + + function invalid(message: string): never { + throw new InvalidRepositoryError({ message }) + } + + function validate(value: string | undefined): string | undefined { + if (value === undefined) return undefined + if ( + value.length === 0 || + value.length > 255 || + !branch.test(value) || + value.startsWith("-") || + value.startsWith("/") || + value.endsWith("/") || + value.endsWith(".") || + value.includes("//") || + value.includes("..") || + value.split("/").some((part) => part.startsWith(".") || part.endsWith(".lock")) + ) { + throw new InvalidBranchError({ message: "Repository branch must be a valid Git branch name" }) + } + return value + } + + function github(path: string) { + const clean = path.startsWith("/") ? path.slice(1) : path + const trimmed = clean.endsWith("/") ? clean.slice(0, -1) : clean + const parts = trimmed.split("/") + if (parts.length !== 2) return invalid("GitHub repository URL must contain exactly owner/repository") + const owner = parts[0] + const raw = parts[1] + if (!owner || !raw) return invalid("GitHub repository URL must contain exactly owner/repository") + const repo = raw.endsWith(".git") ? raw.slice(0, -4) : raw + const value = `${owner}/${repo}` + if (!shorthand.test(value) || [owner, repo].some((part) => part === "." || part === "..")) { + return invalid("GitHub repository URL must contain a safe owner/repository value") + } + return value + } + + function ssh(value: string): string | undefined { + if (value.startsWith("ssh://")) { + if (!URL.canParse(value)) return invalid("Repository SSH URL is invalid") + const url = new URL(value) + if ( + url.protocol !== "ssh:" || + url.hostname !== "github.com" || + url.hostname.endsWith(".") || + url.username !== "git" || + url.password !== "" || + url.port !== "" + ) { + return invalid("Only standard GitHub SSH repository URLs are supported") + } + return github(url.pathname) + } + + const match = value.match(/^([^@/\s]+)@([^:/\s]+):(.+)$/) + if (!match) return undefined + const host = match[2].toLowerCase() + if (match[1] !== "git" || host !== "github.com" || host.endsWith(".")) { + return invalid("Only standard GitHub SCP repository URLs are supported") + } + return github(match[3]) + } + + function https(value: string) { + if (!value.startsWith("https://") || !URL.canParse(value)) { + return invalid("Repository URL must be a valid HTTPS URL") + } + const url = new URL(value) + if ( + url.protocol !== "https:" || + url.username !== "" || + url.password !== "" || + url.hostname.endsWith(".") || + url.search !== "" || + url.hash !== "" + ) { + return invalid("Repository URL must not include credentials, query strings, or fragments") + } + return url + } + + function parsed(input: ParseInput) { + return Effect.try({ + try: () => parse(input), + catch: (error) => { + if (error instanceof InvalidRepositoryError || error instanceof InvalidBranchError) return error + return new InvalidRepositoryError({ message: "Repository is invalid" }) + }, + }) + } + + export function parse(input: ParseInput): Output { + const ref = input.repo + const name = validate(input.branch) + const short = shorthand.test(ref) ? ref : undefined + if (short) { + const parts = short.split("/") + if (parts.some((part) => part === "." || part === "..")) { + return invalid("GitHub shorthand must contain a safe owner/repository value") + } + if (input.type !== undefined && input.type !== "github") { + return invalid("GitHub shorthand is only compatible with repository type github") + } + return { type: "github", repo: short, ...(name === undefined ? {} : { branch: name }) } + } + + const secure = ssh(ref) + if (secure !== undefined) { + if (input.type !== undefined && input.type !== "github") { + return invalid("GitHub SSH repositories are only compatible with repository type github") + } + return { type: "github", repo: secure, ...(name === undefined ? {} : { branch: name }) } + } + + const url = https(ref) + const host = url.hostname.toLowerCase() + if (host === "github.com") { + if (input.type !== undefined && input.type !== "github") { + return invalid("github.com URLs are only compatible with repository type github") + } + if (url.port !== "") return invalid("Repository type github requires a standard github.com URL") + return { type: "github", repo: github(url.pathname), ...(name === undefined ? {} : { branch: name }) } + } + if (input.type === "github") return invalid("Repository type github requires a standard github.com URL") + const type = input.type ?? (host === "gitlab.com" ? "gitlab" : "git") + return { type, url: url.toString(), ...(name === undefined ? {} : { branch: name }) } + } + + export const resolve = Effect.fn("CloudRepository.resolve")(function* (input: Input) { + if (input.repo !== undefined) { + return yield* parsed({ + repo: input.repo, + ...(input.type === undefined ? {} : { type: input.type }), + ...(input.branch === undefined ? {} : { branch: input.branch }), + }) + } + + const git = yield* Git.Service + const tree = yield* git.run(["rev-parse", "--is-inside-work-tree"], { cwd: input.cwd }) + if (tree.exitCode !== 0 || tree.text().trim() !== "true") { + return yield* Effect.fail(new NotWorktreeError({ message: "Current directory is not inside a Git worktree" })) + } + + const listed = yield* git.run(["remote"], { cwd: input.cwd }) + if (listed.exitCode !== 0) { + return yield* Effect.fail(new DiscoveryError({ message: "Unable to inspect Git remotes" })) + } + const remotes = listed + .text() + .split(/\r?\n/) + .map((item) => item.trim()) + .filter(Boolean) + if (remotes.length === 0) { + return yield* Effect.fail(new NoRemoteError({ message: "Current Git worktree has no remotes" })) + } + + const current = yield* git.branch(input.cwd) + const tracking = yield* Effect.gen(function* () { + if (!current) return undefined + const result = yield* git.run(["config", "--get", `branch.${current}.remote`], { cwd: input.cwd }) + if (result.exitCode !== 0) return undefined + const name = result.text().trim() + if (!remotes.includes(name)) return undefined + return name + }) + const remote = tracking ?? (remotes.includes("origin") ? "origin" : remotes.length === 1 ? remotes[0] : undefined) + if (!remote) { + return yield* Effect.fail( + new AmbiguousRemoteError({ message: "Current Git worktree has multiple remotes and none can be selected" }), + ) + } + + const fetched = yield* git.run(["remote", "get-url", remote], { cwd: input.cwd }) + const text = fetched.text() + const ref = text.endsWith("\r\n") ? text.slice(0, -2) : text.endsWith("\n") ? text.slice(0, -1) : text + if (fetched.exitCode !== 0 || !ref) { + return yield* Effect.fail(new DiscoveryError({ message: "Unable to read the selected Git remote fetch URL" })) + } + if (shorthand.test(ref)) { + return yield* Effect.fail(new InvalidRepositoryError({ message: "Local repository remotes are not supported" })) + } + + return yield* parsed({ + repo: ref, + ...(input.type === undefined ? {} : { type: input.type }), + ...(input.branch === undefined ? {} : { branch: input.branch }), + }) + }) +} diff --git a/packages/opencode/src/kilocode/cloud/response-json.ts b/packages/opencode/src/kilocode/cloud/response-json.ts new file mode 100644 index 0000000000..e59b178ab4 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/response-json.ts @@ -0,0 +1,45 @@ +export const MAX_CLOUD_AGENT_RESPONSE_BYTES = 5 * 1024 * 1024 + +export class ResponseJsonError extends Error { + constructor(message: string) { + super(message) + this.name = "ResponseJsonError" + } +} + +export async function readBoundedJson(response: Response, max = MAX_CLOUD_AGENT_RESPONSE_BYTES): Promise { + const length = response.headers.get("content-length") + if (length !== null && /^\d+$/.test(length) && Number(length) > max) { + await response.body?.cancel().catch(() => undefined) + throw new ResponseJsonError("JSON response exceeds the configured limit") + } + if (response.body === null) throw new ResponseJsonError("JSON response body is missing") + + const reader = response.body.getReader() + const chunks: Uint8Array[] = [] + let size = 0 + while (true) { + const chunk = await reader.read() + if (chunk.done) break + size += chunk.value.byteLength + if (size > max) { + await reader.cancel().catch(() => undefined) + throw new ResponseJsonError("JSON response exceeds the configured limit") + } + chunks.push(chunk.value) + } + + const bytes = new Uint8Array(size) + let offset = 0 + for (const chunk of chunks) { + bytes.set(chunk, offset) + offset += chunk.byteLength + } + + try { + const text = new TextDecoder("utf-8", { fatal: true }).decode(bytes) + return JSON.parse(text) as unknown + } catch { + throw new ResponseJsonError("JSON response is invalid") + } +} diff --git a/packages/opencode/src/kilocode/cloud/stream-ticket.ts b/packages/opencode/src/kilocode/cloud/stream-ticket.ts new file mode 100644 index 0000000000..573d5202f6 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/stream-ticket.ts @@ -0,0 +1,96 @@ +import z from "zod" +import { CloudError } from "./errors" +import { type ServiceOrigin } from "./origin" +import { readBoundedJson } from "./response-json" + +const MAX_STREAM_TICKET_RESPONSE_BYTES = 64 * 1024 +const STREAM_TICKET_TIMEOUT_MS = 30_000 +const TICKET_RETRY_ATTEMPTS = 10 +const TICKET_RETRY_DELAY_MS = 1000 + +const StreamTicketResponseSchema = z.object({ + ticket: z.string().min(1), + expiresAt: z.number(), +}) + +export interface StreamTicket { + readonly ticket: string + readonly expiresAt: number +} + +export interface StreamTicketClient { + fetchTicket(input: { readonly cloudAgentSessionId: string; readonly organizationId?: string }): Promise +} + +export interface CreateStreamTicketClientOptions { + readonly origin: ServiceOrigin + readonly apiKey: string + readonly fetch?: typeof globalThis.fetch +} + +export function createStreamTicketClient(options: CreateStreamTicketClientOptions): StreamTicketClient { + const fetcher = options.fetch ?? globalThis.fetch + + return { + async fetchTicket(input) { + const url = new URL("/api/cloud-agent-next/sessions/stream-ticket", options.origin) + const body = JSON.stringify({ + cloudAgentSessionId: input.cloudAgentSessionId, + ...(input.organizationId === undefined ? {} : { organizationId: input.organizationId }), + }) + + let last: Error | undefined + for (let attempt = 0; attempt < TICKET_RETRY_ATTEMPTS; attempt++) { + if (attempt > 0) await delay(TICKET_RETRY_DELAY_MS) + + let response: Response + try { + response = await fetcher(url, { + method: "POST", + redirect: "error", + headers: { + "content-type": "application/json", + authorization: `Bearer ${options.apiKey}`, + }, + body, + signal: AbortSignal.timeout(STREAM_TICKET_TIMEOUT_MS), + }) + } catch { + throw new CloudError("Unable to reach Web App stream ticket endpoint") + } + + const payload = await readBoundedJson(response, MAX_STREAM_TICKET_RESPONSE_BYTES).catch(() => { + if (response.status === 403 || response.status === 404) return undefined + throw new CloudError("Web App returned an invalid stream ticket response") + }) + + if (response.ok) { + const parsed = StreamTicketResponseSchema.safeParse(payload) + if (!parsed.success) throw new CloudError("Web App returned an invalid stream ticket response") + return { ticket: parsed.data.ticket, expiresAt: parsed.data.expiresAt } + } + + last = new CloudError(messageForStatus(response.status, payload)) + if (response.status !== 403 && response.status !== 404) throw last + } + + throw last ?? new CloudError("Unable to obtain stream ticket") + }, + } +} + +function messageForStatus(status: number, payload: unknown): string { + const server = + typeof payload === "object" && payload !== null && "error" in payload && typeof payload.error === "string" + ? payload.error + : undefined + + if (status === 401) return server ?? "Web App rejected authentication" + if (status === 403) return server ?? "Web App denied stream ticket access" + if (status === 404) return server ?? "Web App session was not found" + return server ?? `Web App rejected the stream ticket request with status ${status}` +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)) +} diff --git a/packages/opencode/src/kilocode/cloud/trpc.ts b/packages/opencode/src/kilocode/cloud/trpc.ts new file mode 100644 index 0000000000..9d22f24d46 --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/trpc.ts @@ -0,0 +1,158 @@ +import z from "zod" +import { + AgentSendRequestSchema, + AgentSendResponseSchema, + AgentStartRequestSchema, + AgentStartResponseSchema, + GetMessageResultInputSchema, + GetMessageResultOutputSchema, + type AgentSendRequest, + type AgentSendResponse, + type AgentStartRequest, + type AgentStartResponse, + type Decoder, + type GetMessageResultInput, + type MessageResult, +} from "./contracts" +import { ambiguousAdmissionError, CloudError, ServiceRedirectError, ServiceTransportError } from "./errors" +import { createBearerHttpClient, type BearerHttpClient, type BearerHttpClientOptions, type BearerRequest } from "./http" +import { createMessageId } from "./message-id" +import { readBoundedJson } from "./response-json" + +const TrpcSuccessEnvelopeSchema = z.object({ result: z.object({ data: z.unknown() }).strict() }).strict() + +type Admission = "start" | "send" + +export interface TrpcClient { + query(procedure: "getMessageResult", input: unknown, decoder: Decoder): Promise + mutation(procedure: Admission, input: unknown, decoder: Decoder): Promise +} + +export interface AgentClient { + start(input: AgentStartRequest): Promise + send(input: AgentSendRequest): Promise + getMessageResult(input: GetMessageResultInput): Promise +} + +export interface AgentClientOptions { + readonly id?: () => string +} + +export interface CloudAgentClientOptions extends BearerHttpClientOptions, AgentClientOptions {} + +export function createTrpcClient(options: { readonly http: BearerHttpClient }): TrpcClient { + return { + query(procedure, input, decoder) { + return request( + options.http, + { + method: "GET", + path: `/trpc/${procedure}?input=${encodeURIComponent(JSON.stringify(input))}`, + }, + decoder, + ) + }, + mutation(procedure, input, decoder) { + return request( + options.http, + { + method: "POST", + path: `/trpc/${procedure}`, + headers: { "content-type": "application/json" }, + body: JSON.stringify(input), + }, + decoder, + procedure, + ) + }, + } +} + +export function createAgentClient(trpc: TrpcClient, options: AgentClientOptions = {}): AgentClient { + const create = options.id ?? createMessageId + return { + async start(input) { + const parsed = decode(AgentStartRequestSchema, input, "Cloud Agent start request is invalid") + const id = parsed.message.id ?? create() + const body = { ...parsed, message: { ...parsed.message, id } } + const result = await trpc.mutation("start", body, AgentStartResponseSchema) + if (result.messageId !== id) throw ambiguousAdmissionError("start") + return result + }, + async send(input) { + const parsed = decode(AgentSendRequestSchema, input, "Cloud Agent send request is invalid") + const id = parsed.message.id ?? create() + const body = { ...parsed, message: { ...parsed.message, id } } + const result = await trpc.mutation("send", body, AgentSendResponseSchema) + if (result.cloudAgentSessionId !== parsed.cloudAgentSessionId || result.messageId !== id) { + throw ambiguousAdmissionError("send") + } + return result + }, + async getMessageResult(input) { + const parsed = decode(GetMessageResultInputSchema, input, "Cloud Agent lookup request is invalid") + const result = await trpc.query("getMessageResult", parsed, GetMessageResultOutputSchema) + if (result.cloudAgentSessionId !== parsed.cloudAgentSessionId || result.messageId !== parsed.messageId) { + throw new CloudError("Cloud Agent returned an invalid response") + } + return result + }, + } +} + +export function createCloudAgentClient(options: CloudAgentClientOptions) { + const http = createBearerHttpClient(options) + return createAgentClient(createTrpcClient({ http }), options) +} + +function decode(schema: z.ZodType, input: unknown, message: string) { + const result = schema.safeParse(input) + if (!result.success) throw new CloudError(message) + return result.data +} + +async function request( + http: BearerHttpClient, + req: BearerRequest, + decoder: Decoder, + admission?: Admission, +): Promise { + const response = await http.request(req).catch((error: unknown) => { + if (admission !== undefined && (error instanceof ServiceTransportError || error instanceof ServiceRedirectError)) { + throw ambiguousAdmissionError(admission) + } + throw error + }) + + if (!response.ok) { + await discard(response) + if (admission !== undefined && response.status >= 500) throw ambiguousAdmissionError(admission) + throw new CloudError(errorMessageForStatus(response.status)) + } + + const payload = await readBoundedJson(response).catch(() => { + if (admission !== undefined) throw ambiguousAdmissionError(admission) + throw new CloudError("Cloud Agent returned an invalid response") + }) + + try { + const envelope = TrpcSuccessEnvelopeSchema.parse(payload) + return decoder.parse(envelope.result.data) + } catch { + if (admission !== undefined) throw ambiguousAdmissionError(admission) + throw new CloudError("Cloud Agent returned an invalid response") + } +} + +async function discard(response: Response) { + await response.body?.cancel().catch(() => undefined) +} + +function errorMessageForStatus(status: number) { + if (status === 401) return "Cloud Agent rejected authentication" + if (status === 402) return "Cloud Agent requires additional balance" + if (status === 403) return "Cloud Agent denied the request" + if (status === 404) return "Cloud Agent session or message was not found" + if (status >= 500) return "Cloud Agent is temporarily unavailable" + return `Cloud Agent rejected the request with status ${status}` +} diff --git a/packages/opencode/src/kilocode/cloud/websocket-stream.ts b/packages/opencode/src/kilocode/cloud/websocket-stream.ts new file mode 100644 index 0000000000..56f84326cf --- /dev/null +++ b/packages/opencode/src/kilocode/cloud/websocket-stream.ts @@ -0,0 +1,194 @@ +import { CloudError } from "./errors" + +const COMPLETE_GRACE_PERIOD_MS = 3000 +const DEFAULT_STREAM_TIMEOUT_MS = 30_000 +const CLOSE_TIMEOUT_MS = 1000 +const DRAIN_TIMEOUT_MS = 1000 +// Measured via text.length (UTF-16 units), so queued string memory can reach +// roughly twice this for non-Latin-1 payloads; the goal is boundedness, not precision. +const MAX_QUEUED_BYTES = 8 * 1024 * 1024 + +export interface StreamAgentEventsOptions { + readonly streamUrl: string + readonly origin: string + readonly writeLine: (line: string) => void | Promise + readonly WebSocket?: typeof WebSocket | undefined + readonly timeoutMs?: number +} + +export function streamAgentEvents(options: StreamAgentEventsOptions): Promise { + const resolved = (() => { + try { + return { url: resolveWebSocketUrl(options.streamUrl, options.origin) } + } catch (error) { + return { error } + } + })() + if ("error" in resolved) return Promise.reject(resolved.error) + const url = resolved.url + const WebSocketImpl = options.WebSocket ?? globalThis.WebSocket + + return new Promise((resolve, reject) => { + const socket = new WebSocketImpl(url) + let settled = false + let completeTimer: ReturnType | undefined + let closeTimer: ReturnType | undefined + let idleTimer: ReturnType | undefined + let pending: Promise = Promise.resolve() + let queued = 0 + let writeError = false + let aborted = false + + function clear() { + if (completeTimer !== undefined) { + clearTimeout(completeTimer) + completeTimer = undefined + } + if (closeTimer !== undefined) { + clearTimeout(closeTimer) + closeTimer = undefined + } + if (idleTimer !== undefined) { + clearTimeout(idleTimer) + idleTimer = undefined + } + } + + function finish() { + clear() + if (settled) return + settled = true + void pending.then(() => { + if (writeError) { + reject(new CloudError("WebSocket stream output failed")) + return + } + resolve() + }) + } + + function fail(message: string) { + clear() + if (settled) return + settled = true + socket.close() + const timeout = Math.min(options.timeoutMs ?? DRAIN_TIMEOUT_MS, DRAIN_TIMEOUT_MS) + const timer = setTimeout(() => { + aborted = true + reject(new CloudError(writeError ? "WebSocket stream output failed" : message)) + }, timeout) + void pending.then(() => { + clearTimeout(timer) + reject(new CloudError(writeError ? "WebSocket stream output failed" : message)) + }) + } + + function abort(message: string) { + aborted = true + clear() + if (settled) return + settled = true + socket.close() + reject(new CloudError(message)) + } + + function arm() { + if (idleTimer !== undefined) clearTimeout(idleTimer) + idleTimer = setTimeout(() => fail("WebSocket stream timed out"), options.timeoutMs ?? DEFAULT_STREAM_TIMEOUT_MS) + } + + function initiateClose() { + if (settled) return + try { + socket.close(1000) + } catch { + finish() + return + } + closeTimer = setTimeout(finish, CLOSE_TIMEOUT_MS) + } + + arm() + + socket.onmessage = (event: MessageEvent) => { + arm() + if (settled) return + const text = normalizeMessageData(event.data) + if (queued >= MAX_QUEUED_BYTES) { + abort("WebSocket stream output consumer is too slow") + return + } + queued += text.length + pending = pending + .then(() => { + if (aborted) return + return options.writeLine(text) + }) + .catch(() => { + writeError = true + abort("WebSocket stream output failed") + }) + .finally(() => { + queued -= text.length + }) + if (isCompleteEvent(text) && completeTimer === undefined) { + completeTimer = setTimeout(initiateClose, COMPLETE_GRACE_PERIOD_MS) + } + } + + socket.onerror = () => { + fail("WebSocket stream connection failed") + } + + socket.onclose = (event) => { + if (event.code === 1000) return finish() + fail(`WebSocket stream closed unexpectedly (${event.code})`) + } + }) +} + +function resolveWebSocketUrl(streamUrl: string, origin: string): string { + let url: URL + try { + url = /^(?:wss?|https?):\/\//i.test(streamUrl) ? new URL(streamUrl) : new URL(streamUrl, origin) + } catch { + throw new CloudError("Invalid stream URL") + } + + if (url.protocol === "http:") { + url.protocol = "ws:" + } else if (url.protocol === "https:") { + url.protocol = "wss:" + } + + if (url.protocol !== "ws:" && url.protocol !== "wss:") { + throw new CloudError("Invalid stream URL protocol") + } + + const expected = new URL(origin) + if (expected.protocol === "http:") { + expected.protocol = "ws:" + } else if (expected.protocol === "https:") { + expected.protocol = "wss:" + } + if (url.origin !== expected.origin) { + throw new CloudError("Invalid stream URL origin") + } + + return url.toString() +} + +function isCompleteEvent(text: string): boolean { + try { + const parsed = JSON.parse(text) + return typeof parsed === "object" && parsed !== null && parsed.streamEventType === "complete" + } catch { + return false + } +} + +function normalizeMessageData(data: unknown): string { + if (typeof data === "string") return data + if (data instanceof ArrayBuffer || ArrayBuffer.isView(data)) return new TextDecoder().decode(data) + return String(data) +} diff --git a/packages/opencode/src/kilocode/commands.ts b/packages/opencode/src/kilocode/commands.ts index e462359eab..27f6316c5e 100644 --- a/packages/opencode/src/kilocode/commands.ts +++ b/packages/opencode/src/kilocode/commands.ts @@ -8,7 +8,7 @@ import { AttachCommand } from "../cli/cmd/attach" import { RunCommand } from "../cli/cmd/run" import { GenerateCommand } from "../cli/cmd/generate" import { DebugCommand } from "../cli/cmd/debug" -import { ProvidersCommand } from "../cli/cmd/providers" // kilocode_change — upstream renamed auth → providers +import { ProvidersCommand } from "../cli/cmd/providers" import { AgentCommand } from "../cli/cmd/agent" import { UpgradeCommand } from "../cli/cmd/upgrade" import { UninstallCommand } from "../cli/cmd/uninstall" @@ -30,6 +30,7 @@ import { RollCallCommand } from "./cli/cmd/roll-call" import { ProfileCommand } from "./cli/cmd/profile" import { DaemonCommand } from "./cli/cmd/daemon" import { KiloConsoleCommand } from "./cli/cmd/console" +import { CloudCommand } from "./cli/cmd/cloud" import { HelpCommand } from "./help-command" import { InstallationBuildKind } from "@opencode-ai/core/installation/version" @@ -55,7 +56,7 @@ export const commands = [ RunCommand, GenerateCommand, DebugCommand, - ProvidersCommand, // kilocode_change — upstream renamed AuthCommand → ProvidersCommand + ProvidersCommand, AgentCommand, UpgradeCommand, UninstallCommand, @@ -73,6 +74,7 @@ export const commands = [ RemoteCommand, DaemonCommand, KiloConsoleCommand, + CloudCommand, DbCommand, ConfigCLICommand, ...dev, diff --git a/packages/opencode/test/kilocode/cli-shutdown.test.ts b/packages/opencode/test/kilocode/cli-shutdown.test.ts index 3cbc92b118..e6fea808e8 100644 --- a/packages/opencode/test/kilocode/cli-shutdown.test.ts +++ b/packages/opencode/test/kilocode/cli-shutdown.test.ts @@ -73,6 +73,7 @@ mock.module("@/kilocode/help-command", () => ({ for (const path of [ "@/kilocode/cli/cmd/console", + "@/kilocode/cli/cmd/cloud", "@/kilocode/cli/cmd/roll-call", "@/kilocode/cli/cmd/profile", "@/kilocode/cli/cmd/daemon", @@ -82,6 +83,7 @@ for (const path of [ ]) { mock.module(path, () => ({ KiloConsoleCommand: { command: "console", handler() {} }, + CloudCommand: { command: "cloud", handler() {} }, RollCallCommand: { command: "roll-call", handler() {} }, ProfileCommand: { command: "profile", handler() {} }, DaemonCommand: { command: "daemon", handler() {} }, diff --git a/packages/opencode/test/kilocode/cloud/command.test.ts b/packages/opencode/test/kilocode/cloud/command.test.ts new file mode 100644 index 0000000000..4f18a6ce4b --- /dev/null +++ b/packages/opencode/test/kilocode/cloud/command.test.ts @@ -0,0 +1,540 @@ +import { expect } from "bun:test" +import { Agent } from "@/agent/agent" +import { Auth } from "@/auth" +import { Config } from "@/config/config" +import type { AgentSendRequest, AgentStartRequest, MessageResult } from "@/kilocode/cloud/contracts" +import { CloudCommands } from "@/kilocode/cloud/commands" +import { CloudError } from "@/kilocode/cloud/errors" +import { Git } from "@/git" +import { Effect, Layer } from "effect" +import { TestInstance } from "../../fixture/fixture" +import { testEffect } from "../../lib/effect" + +const SESSION = "agent_12345678-1234-1234-1234-123456789abc" +const MESSAGE = "msg_018f1e2d3c4bAbCdEfGhIjKlMn" +const TOKEN = "command-test-token" +const ORG = "11111111-1111-4111-8111-111111111111" + +const auth = Layer.mock(Auth.Service)({ + get: (id) => + Effect.succeed( + id === "kilo" + ? new Auth.Oauth({ + type: "oauth", + access: TOKEN, + refresh: "test-refresh", + expires: Date.now() + 60_000, + accountId: ORG, + }) + : undefined, + ), +}) + +const it = testEffect(Layer.mergeAll(Agent.defaultLayer, Config.defaultLayer, Git.defaultLayer, auth)) + +const run = Effect.fn("CloudCommandTest.git")(function* (cwd: string, ...args: string[]) { + const git = yield* Git.Service + const result = yield* git.run(args, { cwd }) + if (result.exitCode === 0) return + yield* Effect.die(new Error(result.stderr.toString("utf8"))) +}) + +it.instance( + "assembles the default start request from Kilo state and the current repository", + () => + Effect.acquireUseRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + const url = new URL(request.url) + if (url.pathname.endsWith("/models")) { + return Response.json({ data: [{ id: "anthropic/command-model", supported_parameters: ["tools"] }] }) + } + if (url.pathname.endsWith("/defaults")) { + return Response.json({ defaultModel: "anthropic/command-model" }) + } + return new Response(null, { status: 404 }) + }, + }), + ), + (server) => + Effect.gen(function* () { + const test = yield* TestInstance + yield* run(test.directory, "remote", "add", "origin", "git@github.com:Kilo-Org/kilocode.git") + + const requests: AgentStartRequest[] = [] + const keys: string[] = [] + const output: string[] = [] + const response = { + cloudAgentSessionId: SESSION, + kiloSessionId: "ses_command_test", + messageId: MESSAGE, + delivery: "queued" as const, + } + + const result = yield* CloudCommands.start( + { + cwd: test.directory, + prompt: "Inspect the current repository", + }, + { + env: { KILO_API_URL: server.url.origin }, + make: (options) => { + keys.push(options.apiKey) + return { + async start(input) { + requests.push(input) + return response + }, + async send() { + throw new Error("unused send") + }, + async getMessageResult() { + throw new Error("unused result") + }, + } + }, + write: (text) => output.push(text), + }, + ) + + expect(result).toEqual(response) + expect(keys).toEqual([TOKEN]) + expect(requests).toHaveLength(1) + expect(requests[0]).toEqual({ + message: { prompt: "Inspect the current repository" }, + agent: { mode: "plan", model: "anthropic/command-model" }, + repository: { type: "github", repo: "Kilo-Org/kilocode" }, + options: { + createdOnPlatform: "kilo-cli", + kilocodeOrganizationId: ORG, + }, + }) + expect(requests[0]?.repository).not.toHaveProperty("branch") + expect(output).toEqual([JSON.stringify(response) + "\n"]) + }), + (server) => Effect.promise(() => server.stop(true)), + ), + { + git: true, + config: { + default_agent: "plan", + agent: { plan: { model: "kilo/anthropic/command-model" } }, + }, + }, +) + +it.instance( + "streams WebSocket events when --stream is passed", + () => + Effect.acquireUseRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + const url = new URL(request.url) + if (url.pathname.endsWith("/models")) { + return Response.json({ data: [{ id: "anthropic/command-model", supported_parameters: ["tools"] }] }) + } + if (url.pathname.endsWith("/defaults")) { + return Response.json({ defaultModel: "anthropic/command-model" }) + } + return new Response(null, { status: 404 }) + }, + }), + ), + (server) => + Effect.gen(function* () { + const test = yield* TestInstance + const output: string[] = [] + const ticketCalls: { cloudAgentSessionId: string; organizationId?: string }[] = [] + const streamCalls: string[] = [] + const response = { + cloudAgentSessionId: SESSION, + kiloSessionId: "ses_stream_test", + messageId: MESSAGE, + delivery: "queued" as const, + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=inlined", + } + + const result = yield* CloudCommands.start( + { + cwd: test.directory, + prompt: "Inspect the repository", + repo: "Kilo-Org/kilocode", + stream: true, + }, + { + env: { KILO_API_URL: server.url.origin }, + make: () => ({ + async start() { + return response + }, + async send() { + throw new Error("unused") + }, + async getMessageResult() { + throw new Error("unused") + }, + }), + createStreamTicketClient: () => ({ + async fetchTicket(input) { + ticketCalls.push(input) + return { ticket: "should-not-be-used", expiresAt: 0 } + }, + }), + streamAgentEvents: async (options) => { + streamCalls.push(options.streamUrl) + await options.writeLine('{"event":"one"}') + await options.writeLine('{"streamEventType":"complete","data":{"exitCode":0}}') + }, + write: (text) => output.push(text), + }, + ) + + expect(result).toEqual(response) + expect(ticketCalls).toEqual([]) + expect(streamCalls).toEqual(["/stream?cloudAgentSessionId=agent_123&ticket=inlined"]) + expect(output).toEqual([ + JSON.stringify({ ...response, streamUrl: undefined }) + "\n", + '{"event":"one"}\n', + '{"streamEventType":"complete","data":{"exitCode":0}}\n', + ]) + }), + (server) => Effect.promise(() => server.stop(true)), + ), + { + git: true, + config: { + default_agent: "plan", + agent: { plan: { model: "kilo/anthropic/command-model" } }, + }, + }, +) + +it.instance( + "fetches a stream ticket when the response omits streamUrl", + () => + Effect.acquireUseRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + const url = new URL(request.url) + if (url.pathname.endsWith("/models")) { + return Response.json({ data: [{ id: "anthropic/command-model", supported_parameters: ["tools"] }] }) + } + if (url.pathname.endsWith("/defaults")) { + return Response.json({ defaultModel: "anthropic/command-model" }) + } + return new Response(null, { status: 404 }) + }, + }), + ), + (server) => + Effect.gen(function* () { + const test = yield* TestInstance + const output: string[] = [] + const ticketCalls: { cloudAgentSessionId: string; organizationId?: string }[] = [] + const streamCalls: string[] = [] + const response = { + cloudAgentSessionId: SESSION, + kiloSessionId: "ses_stream_test", + messageId: MESSAGE, + delivery: "queued" as const, + } + + yield* CloudCommands.start( + { + cwd: test.directory, + prompt: "Inspect the repository", + repo: "Kilo-Org/kilocode", + orgID: ORG, + stream: true, + }, + { + env: { KILO_API_URL: server.url.origin }, + make: () => ({ + async start() { + return response + }, + async send() { + throw new Error("unused") + }, + async getMessageResult() { + throw new Error("unused") + }, + }), + createStreamTicketClient: () => ({ + async fetchTicket(input) { + ticketCalls.push(input) + return { ticket: "derived-tok", expiresAt: 1234567890 } + }, + }), + streamAgentEvents: async (options) => { + streamCalls.push(options.streamUrl) + await options.writeLine('{"event":"derived"}') + }, + write: (text) => output.push(text), + }, + ) + + expect(ticketCalls).toEqual([{ cloudAgentSessionId: SESSION, organizationId: ORG }]) + expect(streamCalls).toEqual([`/stream?cloudAgentSessionId=${SESSION}&ticket=derived-tok`]) + expect(output).toEqual([JSON.stringify(response) + "\n", '{"event":"derived"}\n']) + }), + (server) => Effect.promise(() => server.stop(true)), + ), + { + git: true, + config: { + default_agent: "plan", + agent: { plan: { model: "kilo/anthropic/command-model" } }, + }, + }, +) + +it.instance( + "keeps admission successful when stream ticket acquisition fails", + () => + Effect.acquireUseRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + const url = new URL(request.url) + if (url.pathname.endsWith("/models")) { + return Response.json({ data: [{ id: "anthropic/command-model", supported_parameters: ["tools"] }] }) + } + if (url.pathname.endsWith("/defaults")) { + return Response.json({ defaultModel: "anthropic/command-model" }) + } + return new Response(null, { status: 404 }) + }, + }), + ), + (server) => + Effect.gen(function* () { + const test = yield* TestInstance + const output: string[] = [] + const response = { + cloudAgentSessionId: SESSION, + kiloSessionId: "ses_stream_test", + messageId: MESSAGE, + delivery: "queued" as const, + } + + const result = yield* CloudCommands.start( + { + cwd: test.directory, + prompt: "Inspect the repository", + repo: "Kilo-Org/kilocode", + stream: true, + }, + { + env: { KILO_API_URL: server.url.origin }, + make: () => ({ + async start() { + return response + }, + async send() { + throw new Error("unused") + }, + async getMessageResult() { + throw new Error("unused") + }, + }), + createStreamTicketClient: () => ({ + async fetchTicket() { + throw new CloudError("Unable to obtain stream ticket") + }, + }), + streamAgentEvents: async () => { + throw new Error("unused") + }, + write: (text) => output.push(text), + }, + ) + + expect(result).toEqual(response) + expect(output).toEqual([ + JSON.stringify(response) + "\n", + JSON.stringify({ streamEventType: "error", data: { message: "Unable to obtain stream ticket" } }) + "\n", + ]) + }), + (server) => Effect.promise(() => server.stop(true)), + ), + { + git: true, + config: { + default_agent: "plan", + agent: { plan: { model: "kilo/anthropic/command-model" } }, + }, + }, +) + +it.instance( + "keeps a successful admission successful when the follow-up stream fails", + () => + Effect.acquireUseRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + const url = new URL(request.url) + if (url.pathname.endsWith("/models")) { + return Response.json({ data: [{ id: "anthropic/command-model", supported_parameters: ["tools"] }] }) + } + if (url.pathname.endsWith("/defaults")) { + return Response.json({ defaultModel: "anthropic/command-model" }) + } + return new Response(null, { status: 404 }) + }, + }), + ), + (server) => + Effect.gen(function* () { + const test = yield* TestInstance + const output: string[] = [] + const response = { + cloudAgentSessionId: SESSION, + kiloSessionId: "ses_stream_test", + messageId: MESSAGE, + delivery: "queued" as const, + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=inlined", + } + + const result = yield* CloudCommands.start( + { + cwd: test.directory, + prompt: "Inspect the repository", + repo: "Kilo-Org/kilocode", + stream: true, + }, + { + env: { KILO_API_URL: server.url.origin }, + make: () => ({ + async start() { + return response + }, + async send() { + throw new Error("unused") + }, + async getMessageResult() { + throw new Error("unused") + }, + }), + streamAgentEvents: async () => { + throw new Error("stream failed after admission") + }, + write: (text) => output.push(text), + }, + ) + + expect(result).toEqual(response) + expect(output).toEqual([ + JSON.stringify({ ...response, streamUrl: undefined }) + "\n", + JSON.stringify({ streamEventType: "error", data: { message: "Cloud Agent stream failed" } }) + "\n", + ]) + }), + (server) => Effect.promise(() => server.stop(true)), + ), + { + git: true, + config: { + default_agent: "plan", + agent: { plan: { model: "kilo/anthropic/command-model" } }, + }, + }, +) + +it.instance("sends follow-ups, prints status without assistant content, and applies the result exit code", () => + Effect.gen(function* () { + const output: string[] = [] + const exits: number[] = [] + const sends: AgentSendRequest[] = [] + const sent = { + cloudAgentSessionId: SESSION, + status: "started" as const, + streamUrl: "wss://cloud-agent.example/stream", + messageId: MESSAGE, + delivery: "queued" as const, + } + const results: MessageResult[] = [ + { + cloudAgentSessionId: SESSION, + messageId: MESSAGE, + status: "completed", + createdAt: 1, + terminalAt: 2, + assistant: { messageId: "assistant_1", text: "done" }, + }, + { + cloudAgentSessionId: SESSION, + messageId: MESSAGE, + status: "failed", + createdAt: 1, + terminalAt: 2, + failure: { retryable: false }, + }, + ] + const deps = { + env: { KILO_ORG_ID: "not-relevant-to-existing-sessions" }, + make: () => ({ + async start() { + throw new Error("unused start") + }, + async send(input: AgentSendRequest) { + sends.push(input) + return sent + }, + async getMessageResult() { + const result = results.shift() + if (!result) throw new Error("missing test result") + return result + }, + }), + write: (text: string) => output.push(text), + exit: (code: number) => exits.push(code), + } + + yield* CloudCommands.send({ sessionID: SESSION, prompt: "Continue" }, deps) + yield* CloudCommands.status({ sessionID: SESSION, messageID: MESSAGE }, deps) + yield* CloudCommands.result({ sessionID: SESSION, messageID: MESSAGE }, deps) + + expect(sends).toEqual([{ cloudAgentSessionId: SESSION, message: { prompt: "Continue" } }]) + expect(output).toEqual([ + JSON.stringify({ ...sent, streamUrl: undefined }) + "\n", + JSON.stringify({ + cloudAgentSessionId: SESSION, + messageId: MESSAGE, + status: "completed", + createdAt: 1, + terminalAt: 2, + }) + "\n", + JSON.stringify({ + cloudAgentSessionId: SESSION, + messageId: MESSAGE, + status: "failed", + createdAt: 1, + terminalAt: 2, + failure: { retryable: false }, + }) + "\n", + ]) + expect(exits).toEqual([3]) + + const error = yield* CloudCommands.send( + { sessionID: SESSION, prompt: "Do not duplicate" }, + { ...deps, write: () => Promise.reject(new Error("closed output")) }, + ).pipe(Effect.flip) + expect(error.message).toBe( + "Cloud Agent send was admitted but output could not be written; do not retry automatically", + ) + }), +) diff --git a/packages/opencode/test/kilocode/cloud/defaults.test.ts b/packages/opencode/test/kilocode/cloud/defaults.test.ts new file mode 100644 index 0000000000..016705c7e7 --- /dev/null +++ b/packages/opencode/test/kilocode/cloud/defaults.test.ts @@ -0,0 +1,436 @@ +import { expect } from "bun:test" +import { Effect, Layer, Redacted, Ref } from "effect" +import { Agent } from "@/agent/agent" +import { Auth } from "@/auth" +import { Config } from "@/config/config" +import { CloudAuth } from "@/kilocode/cloud/auth" +import { CloudCatalog } from "@/kilocode/cloud/catalog" +import { CloudDefaults } from "@/kilocode/cloud/defaults" +import { MAX_CLOUD_AGENT_RESPONSE_BYTES } from "@/kilocode/cloud/response-json" +import { testEffect } from "../../lib/effect" + +const it = testEffect(Layer.mergeAll(Agent.defaultLayer, Config.defaultLayer)) + +type RequestInfo = { + readonly authorization: string | null + readonly feature: string | null + readonly organization: string | null + readonly path: string +} + +const oauth = (token: string, organizationID: string) => + new Auth.Oauth({ + type: "oauth", + access: token, + refresh: "test-refresh", + expires: Date.now() + 60_000, + accountId: organizationID, + }) + +const authLayer = (info: Auth.Info) => + Layer.mock(Auth.Service)({ + get: (id) => Effect.succeed(id === "kilo" ? info : undefined), + }) + +const stateLayer = (state: CloudDefaults.ModelStateInfo) => + Layer.mock(CloudDefaults.ModelState)({ + get: () => Effect.succeed(state), + }) + +const state = (input: Partial = {}): CloudDefaults.ModelStateInfo => ({ + model: {}, + recent: [], + favorite: [], + variant: {}, + ...input, +}) + +function withCatalog( + models: readonly string[], + defaultModel: string, + use: (url: URL, requests: RequestInfo[]) => Effect.Effect, +) { + const requests: RequestInfo[] = [] + return Effect.acquireUseRelease( + Effect.sync(() => + Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + const url = new URL(request.url) + requests.push({ + authorization: request.headers.get("authorization"), + feature: request.headers.get("x-kilocode-feature"), + organization: request.headers.get("x-kilocode-organizationid"), + path: url.pathname, + }) + if (url.pathname.endsWith("/models")) { + return Response.json({ data: models.map((id) => ({ id, supported_parameters: ["tools"] })) }) + } + if (url.pathname.endsWith("/defaults")) return Response.json({ defaultModel }) + return new Response(null, { status: 404 }) + }, + }), + ), + (server) => use(server.url, requests), + (server) => Effect.promise(() => server.stop(true)), + ) +} + +it.instance("routes URL-scoped credentials to their catalog origin", () => { + const token = "https://catalog.example.test:scoped-token" + const requests: RequestInfo[] = [] + return Effect.gen(function* () { + const catalog = yield* CloudCatalog.Service + + expect(yield* catalog.models({ token: Redacted.make(token) })).toEqual(["anthropic/scoped"]) + expect(requests).toEqual([ + { + authorization: `Bearer ${token}`, + feature: "kilo-cli", + organization: null, + path: "/api/openrouter/models", + }, + ]) + }).pipe( + Effect.provide( + CloudCatalog.layer({ + env: {}, + fetch: async (request) => { + const url = new URL(request.url) + expect(url.origin).toBe("https://catalog.example.test") + requests.push({ + authorization: request.headers.get("authorization"), + feature: request.headers.get("x-kilocode-feature"), + organization: request.headers.get("x-kilocode-organizationid"), + path: url.pathname, + }) + return Response.json({ data: [{ id: "anthropic/scoped", supported_parameters: ["tools"] }] }) + }, + }), + ), + ) +}) + +it.instance("returns only tool-capable text-output models", () => + Effect.gen(function* () { + const catalog = yield* CloudCatalog.Service + const models = yield* catalog.models({ token: Redacted.make("stored-token") }) + expect(models).toEqual(["anthropic/code"]) + }).pipe( + Effect.provide( + CloudCatalog.layer({ + fetch: async () => + Response.json({ + data: [ + { + id: "anthropic/code", + architecture: { output_modalities: ["text"] }, + supported_parameters: ["tools"], + }, + { + id: "image/generator", + architecture: { output_modalities: ["image"] }, + supported_parameters: ["tools"], + }, + { + id: "anthropic/chat", + architecture: { output_modalities: ["text"] }, + supported_parameters: ["temperature"], + }, + { id: "anthropic/unknown" }, + ], + }), + }), + ), + ), +) + +it.instance("rejects oversized catalog responses", () => + Effect.gen(function* () { + const catalog = yield* CloudCatalog.Service + const error = yield* catalog.models({ token: Redacted.make("stored-token") }).pipe(Effect.flip) + expect(error).toMatchObject({ _tag: "CloudCatalogError", kind: "schema" }) + }).pipe( + Effect.provide( + CloudCatalog.layer({ + fetch: async () => + Response.json({ data: [] }, { headers: { "content-length": String(MAX_CLOUD_AGENT_RESPONSE_BYTES + 1) } }), + }), + ), + ), +) + +it.instance( + "explicit overrides beat environment and saved defaults without persisting", + () => + withCatalog(["anthropic/explicit", "anthropic/default"], "anthropic/default", (url, requests) => + Effect.gen(function* () { + const savedID = "22222222-2222-4222-8222-222222222222" + const envID = "33333333-3333-4333-8333-333333333333" + const explicitID = "44444444-4444-4444-8444-444444444444" + const stored = oauth("stored-token", savedID) + const current = yield* Ref.make(stored) + const auth = Layer.mock(Auth.Service)({ + get: (id) => (id === "kilo" ? Ref.get(current) : Effect.succeed(undefined)), + set: (id, info) => (id === "kilo" ? Ref.set(current, info) : Effect.void), + }) + + const resolved = yield* Effect.gen(function* () { + const result = yield* CloudDefaults.resolve({ + mode: "debug", + model: "kilo/anthropic/explicit", + orgID: explicitID, + env: { + KILO_API_KEY: "ignored-env-token", + KILO_ORG_ID: envID, + }, + }) + const service = yield* Auth.Service + expect(yield* service.get("kilo")).toEqual(stored) + return result + }).pipe( + Effect.provide( + Layer.mergeAll( + auth, + stateLayer( + state({ + model: { debug: { providerID: "kilo", modelID: "anthropic/saved" } }, + }), + ), + CloudCatalog.layer({ env: { KILO_API_URL: url.origin } }), + ), + ), + ) + + expect(resolved).toMatchObject({ + mode: "debug", + model: "anthropic/explicit", + organizationID: explicitID, + }) + expect(requests.map((request) => request.path)).toEqual([`/api/organizations/${explicitID}/models`]) + expect( + requests.every( + (request) => + request.path.startsWith(`/api/organizations/${explicitID}/`) && + request.authorization === "Bearer stored-token" && + request.feature === "kilo-cli" && + request.organization === explicitID, + ), + ).toBe(true) + }), + ), + { + config: { + default_agent: "plan", + model: "kilo/anthropic/repository", + agent: { plan: { model: "kilo/anthropic/mode" } }, + }, + }, +) + +it.instance( + "skips a stale saved model and uses the available repository model", + () => + withCatalog( + ["anthropic/repository", "anthropic/recent", "anthropic/default"], + "anthropic/default", + (url, requests) => + Effect.gen(function* () { + const resolved = yield* CloudDefaults.resolve() + expect(resolved.mode).toBe("code") + expect(resolved.model).toBe("anthropic/repository") + expect(requests.map((request) => request.path)).toEqual(["/api/openrouter/models"]) + expect(requests.every((request) => request.organization === null)).toBe(true) + }).pipe( + Effect.provide( + Layer.mergeAll( + authLayer(new Auth.Api({ type: "api", key: "stored-api-token" })), + stateLayer( + state({ + model: { code: { providerID: "kilo", modelID: "anthropic/stale" } }, + recent: [{ providerID: "kilo", modelID: "anthropic/recent" }], + }), + ), + CloudCatalog.layer({ env: { KILO_API_URL: url.origin } }), + ), + ), + ), + ), + { + config: { + model: "kilo/anthropic/repository", + agent: { code: { model: null } }, + }, + }, +) + +it.instance( + "uses the saved model for the resolved mode when it remains available", + () => + withCatalog(["anthropic/saved", "anthropic/default"], "anthropic/default", (url) => + CloudDefaults.resolve().pipe( + Effect.tap((resolved) => + Effect.sync(() => { + expect(resolved.mode).toBe("code") + expect(resolved.model).toBe("anthropic/saved") + }), + ), + Effect.provide( + Layer.mergeAll( + authLayer(new Auth.Api({ type: "api", key: "stored-api-token" })), + stateLayer( + state({ + model: { code: { providerID: "kilo", modelID: "anthropic/saved" } }, + }), + ), + CloudCatalog.layer({ env: { KILO_API_URL: url.origin } }), + ), + ), + ), + ), + { + config: { agent: { code: { model: null } } }, + }, +) + +it.instance( + "fetches the catalog default only when configured and saved candidates are unavailable", + () => + withCatalog(["anthropic/default"], "anthropic/default", (url, requests) => + CloudDefaults.resolve().pipe( + Effect.tap((resolved) => + Effect.sync(() => { + expect(resolved.model).toBe("anthropic/default") + expect(requests.map((request) => request.path)).toEqual(["/api/openrouter/models", "/api/defaults"]) + }), + ), + Effect.provide( + Layer.mergeAll( + authLayer(new Auth.Api({ type: "api", key: "stored-api-token" })), + stateLayer(state()), + CloudCatalog.layer({ env: { KILO_API_URL: url.origin } }), + ), + ), + ), + ), + { + config: { agent: { code: { model: null } } }, + }, +) + +it.instance( + "falls back from an inferred custom mode but rejects an explicit custom mode", + () => + withCatalog(["anthropic/code", "anthropic/custom", "anthropic/default"], "anthropic/default", (url, requests) => + Effect.gen(function* () { + const inferred = yield* CloudDefaults.resolve() + expect(inferred.mode).toBe("code") + expect(inferred.model).toBe("anthropic/code") + + const error = yield* CloudDefaults.resolve({ mode: "custom" }).pipe(Effect.flip) + expect(error).toMatchObject({ + _tag: "CloudDefaultsResolutionError", + kind: "mode", + }) + expect(requests).toHaveLength(1) + }).pipe( + Effect.provide( + Layer.mergeAll( + authLayer(new Auth.Api({ type: "api", key: "stored-api-token" })), + stateLayer(state()), + CloudCatalog.layer({ env: { KILO_API_URL: url.origin } }), + ), + ), + ), + ), + { + config: { + default_agent: "custom", + agent: { + code: { model: "kilo/anthropic/code" }, + custom: { mode: "primary", model: "kilo/anthropic/custom" }, + }, + }, + }, +) + +it.instance("rejects invalid persisted organization state and insecure catalog origins", () => + Effect.gen(function* () { + const invalid = authLayer(oauth("stored-token", "not-a-uuid")) + const token = yield* CloudAuth.token().pipe(Effect.provide(invalid)) + expect(Redacted.value(token)).toBe("stored-token") + + const org = yield* CloudAuth.resolve().pipe(Effect.provide(invalid), Effect.flip) + expect(org).toMatchObject({ + _tag: "CloudAuthResolutionError", + kind: "organization", + }) + + const catalog = yield* CloudDefaults.resolve({ env: { KILO_API_URL: "http://example.com" } }).pipe( + Effect.provide( + Layer.mergeAll( + authLayer(new Auth.Api({ type: "api", key: "stored-api-token" })), + stateLayer(state()), + CloudCatalog.layer({ + env: { KILO_API_URL: "http://example.com" }, + fetch: () => Promise.reject(new Error("insecure catalog request must not run")), + }), + ), + ), + Effect.flip, + ) + expect(catalog).toMatchObject({ + _tag: "CloudCatalogError", + kind: "schema", + }) + }), +) + +it.instance( + "uses stored auth and the resolved mode model before lower-precedence defaults", + () => + withCatalog( + ["anthropic/mode", "anthropic/saved", "anthropic/repository", "anthropic/recent", "anthropic/default"], + "anthropic/default", + (url, requests) => + Effect.gen(function* () { + const organizationID = "11111111-1111-4111-8111-111111111111" + const resolved = yield* CloudDefaults.resolve({ + env: { KILO_API_KEY: "ignored-env-token" }, + }).pipe( + Effect.provide( + Layer.mergeAll( + authLayer(oauth("stored-token", organizationID)), + stateLayer( + state({ + model: { plan: { providerID: "kilo", modelID: "anthropic/saved" } }, + recent: [{ providerID: "kilo", modelID: "anthropic/recent" }], + }), + ), + CloudCatalog.layer({ env: { KILO_API_URL: url.origin } }), + ), + ), + ) + + expect(resolved).toMatchObject({ + mode: "plan", + model: "anthropic/mode", + organizationID, + }) + expect(requests.map((request) => request.path)).toEqual([`/api/organizations/${organizationID}/models`]) + expect( + requests.every( + (request) => request.authorization === "Bearer stored-token" && request.organization === organizationID, + ), + ).toBe(true) + }), + ), + { + config: { + default_agent: "plan", + model: "kilo/anthropic/repository", + agent: { plan: { model: "kilo/anthropic/mode" } }, + }, + }, +) diff --git a/packages/opencode/test/kilocode/cloud/repository.test.ts b/packages/opencode/test/kilocode/cloud/repository.test.ts new file mode 100644 index 0000000000..0f38f92170 --- /dev/null +++ b/packages/opencode/test/kilocode/cloud/repository.test.ts @@ -0,0 +1,119 @@ +import { describe, expect } from "bun:test" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { Effect, Layer } from "effect" +import { Git } from "../../../src/git" +import { CloudRepository } from "../../../src/kilocode/cloud/repository" +import { tmpdirScoped } from "../../fixture/fixture" +import { testEffect } from "../../lib/effect" + +const it = testEffect(Layer.mergeAll(Git.defaultLayer, CrossSpawnSpawner.defaultLayer)) + +const run = Effect.fn("CloudRepositoryTest.git")(function* (cwd: string, ...args: string[]) { + const git = yield* Git.Service + const result = yield* git.run(args, { cwd }) + if (result.exitCode === 0) return result.text().trim() + return yield* Effect.die(new Error(result.stderr.toString("utf8"))) +}) + +describe("CloudRepository", () => { + it.live("resolves and validates an explicit repository branch outside Git", () => + Effect.gen(function* () { + const cwd = yield* tmpdirScoped() + const result = yield* CloudRepository.resolve({ + cwd, + repo: "kilo-org/kilo", + branch: "feature/cloud-start", + }) + + expect(result).toEqual({ + type: "github", + repo: "kilo-org/kilo", + branch: "feature/cloud-start", + }) + + const error = yield* CloudRepository.resolve({ cwd, repo: "kilo-org/kilo", branch: "feature.lock" }).pipe( + Effect.flip, + ) + expect(error).toBeInstanceOf(CloudRepository.InvalidBranchError) + + const type = yield* CloudRepository.resolve({ + cwd, + repo: "https://github.com/kilo-org/kilo.git", + type: "gitlab", + }).pipe(Effect.flip) + expect(type).toBeInstanceOf(CloudRepository.InvalidRepositoryError) + }), + ) + + it.live("rejects GitHub repositories that become dot segments after trimming .git", () => + Effect.gen(function* () { + const cwd = yield* tmpdirScoped() + + for (const repo of ["https://github.com/kilo-org/...git", "git@github.com:kilo-org/...git"]) { + const error = yield* CloudRepository.resolve({ cwd, repo }).pipe(Effect.flip) + expect(error).toBeInstanceOf(CloudRepository.InvalidRepositoryError) + } + }), + ) + + it.live("normalizes an inferred GitHub SCP remote without adding the current branch", () => + Effect.gen(function* () { + const cwd = yield* tmpdirScoped({ git: true }) + yield* run(cwd, "remote", "add", "origin", "git@github.com:kilo-org/ssh-repo.git") + yield* run(cwd, "checkout", "-b", "feature/not-in-output") + + const result = yield* CloudRepository.resolve({ cwd }) + + expect(result).toEqual({ type: "github", repo: "kilo-org/ssh-repo" }) + }), + ) + + it.live("rejects inferred local remotes that resemble GitHub shorthand", () => + Effect.gen(function* () { + const cwd = yield* tmpdirScoped({ git: true }) + yield* run(cwd, "remote", "add", "origin", "kilo-org/local-repo") + + const error = yield* CloudRepository.resolve({ cwd }).pipe(Effect.flip) + + expect(error).toBeInstanceOf(CloudRepository.InvalidRepositoryError) + }), + ) + + it.live("prefers the tracking remote, then origin, then a sole remote fetch URL", () => + Effect.gen(function* () { + const cwd = yield* tmpdirScoped({ git: true }) + const git = yield* Git.Service + const branch = yield* git.branch(cwd) + if (!branch) yield* Effect.die(new Error("temporary repository has no current branch")) + + yield* run(cwd, "remote", "add", "origin", "https://github.com/kilo-org/origin.git") + yield* run(cwd, "remote", "add", "tracked", "https://github.com/kilo-org/tracked.git") + yield* run(cwd, "remote", "set-url", "--add", "--push", "tracked", "https://github.com/kilo-org/push.git") + yield* run(cwd, "config", `branch.${branch}.remote`, "tracked") + + expect(yield* CloudRepository.resolve({ cwd })).toEqual({ type: "github", repo: "kilo-org/tracked" }) + + yield* run(cwd, "config", `branch.${branch}.remote`, "missing") + expect(yield* CloudRepository.resolve({ cwd })).toEqual({ type: "github", repo: "kilo-org/origin" }) + + yield* run(cwd, "remote", "remove", "origin") + expect(yield* CloudRepository.resolve({ cwd })).toEqual({ type: "github", repo: "kilo-org/tracked" }) + }), + ) + + it.live("returns typed errors when no remote can be selected", () => + Effect.gen(function* () { + const cwd = yield* tmpdirScoped({ git: true }) + + const none = yield* CloudRepository.resolve({ cwd }).pipe(Effect.flip) + expect(none).toBeInstanceOf(CloudRepository.NoRemoteError) + + yield* run(cwd, "remote", "add", "alpha", "https://github.com/kilo-org/alpha.git") + yield* run(cwd, "remote", "add", "beta", "https://github.com/kilo-org/beta.git") + + const ambiguous = yield* CloudRepository.resolve({ cwd }).pipe(Effect.flip) + expect(ambiguous).toBeInstanceOf(CloudRepository.AmbiguousRemoteError) + }), + ) + +}) diff --git a/packages/opencode/test/kilocode/cloud/stream-ticket.test.ts b/packages/opencode/test/kilocode/cloud/stream-ticket.test.ts new file mode 100644 index 0000000000..7b3ac2d655 --- /dev/null +++ b/packages/opencode/test/kilocode/cloud/stream-ticket.test.ts @@ -0,0 +1,134 @@ +import { describe, expect, test } from "bun:test" +import { createStreamTicketClient, type StreamTicketClient } from "@/kilocode/cloud/stream-ticket" +import { parseServiceOrigin } from "@/kilocode/cloud/origin" + +function jsonResponse(body: unknown, status = 200) { + return new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json" }, + }) +} + +function client(options: { fetch: ReturnType }): StreamTicketClient { + return createStreamTicketClient({ + origin: parseServiceOrigin("https://app.example"), + apiKey: "key", + fetch: options.fetch.fetch, + }) +} + +describe("createStreamTicketClient", () => { + test("fetches a stream ticket from the web app", async () => { + const fetch = mockFetch().resolved(jsonResponse({ ticket: "tok", expiresAt: 1234567890 })) + const result = await client({ fetch }).fetchTicket({ cloudAgentSessionId: "agent_123" }) + + expect(result).toEqual({ ticket: "tok", expiresAt: 1234567890 }) + expect(fetch.calls).toHaveLength(1) + const [url, init] = fetch.calls[0]! + expect(url.toString()).toBe("https://app.example/api/cloud-agent-next/sessions/stream-ticket") + expect(init).toMatchObject({ + method: "POST", + redirect: "error", + headers: expect.objectContaining({ + authorization: "Bearer key", + "content-type": "application/json", + }), + body: JSON.stringify({ cloudAgentSessionId: "agent_123" }), + }) + expect(init?.signal).toBeInstanceOf(AbortSignal) + }) + + test("includes organizationId when provided", async () => { + const fetch = mockFetch().resolved(jsonResponse({ ticket: "tok", expiresAt: 1234567890 })) + await client({ fetch }).fetchTicket({ + cloudAgentSessionId: "agent_123", + organizationId: "123e4567-e89b-12d3-a456-426614174000", + }) + + expect(fetch.calls[0]![1]).toMatchObject({ + body: JSON.stringify({ + cloudAgentSessionId: "agent_123", + organizationId: "123e4567-e89b-12d3-a456-426614174000", + }), + }) + }) + + test("throws on transport failure", async () => { + const fetch = mockFetch().rejected(new Error("network error")) + await expect(client({ fetch }).fetchTicket({ cloudAgentSessionId: "agent_123" })).rejects.toThrow( + "Unable to reach Web App stream ticket endpoint", + ) + }) + + test("retries on 403/404 and succeeds once the session becomes visible", async () => { + const fetch = mockFetch() + .resolved(jsonResponse({ error: "Organization does not own this session" }, 403)) + .resolved(jsonResponse({ error: "Organization does not own this session" }, 403)) + .resolved(jsonResponse({ ticket: "tok", expiresAt: 1234567890 })) + const result = await client({ fetch }).fetchTicket({ cloudAgentSessionId: "agent_123" }) + + expect(result).toEqual({ ticket: "tok", expiresAt: 1234567890 }) + expect(fetch.calls).toHaveLength(3) + }) + + test("retries when a 404 response has an invalid body", async () => { + const fetch = mockFetch() + .resolved(new Response("not json", { status: 404 })) + .resolved(jsonResponse({ ticket: "tok", expiresAt: 1234567890 })) + + const result = await client({ fetch }).fetchTicket({ cloudAgentSessionId: "agent_123" }) + + expect(result).toEqual({ ticket: "tok", expiresAt: 1234567890 }) + expect(fetch.calls).toHaveLength(2) + }) + + test("gives up after repeated 403 responses", async () => { + const fetch = mockFetch().repeated(() => jsonResponse({ error: "Denied" }, 403)) + await expect(client({ fetch }).fetchTicket({ cloudAgentSessionId: "agent_123" })).rejects.toThrow("Denied") + expect(fetch.calls).toHaveLength(10) + }, 15_000) + + test("throws on authentication failure", async () => { + const fetch = mockFetch().resolved(jsonResponse({ error: "Unauthorized" }, 401)) + await expect(client({ fetch }).fetchTicket({ cloudAgentSessionId: "agent_123" })).rejects.toThrow("Unauthorized") + }) + + test("throws on invalid response", async () => { + const fetch = mockFetch().resolved(jsonResponse({ missing: "fields" })) + await expect(client({ fetch }).fetchTicket({ cloudAgentSessionId: "agent_123" })).rejects.toThrow( + "Web App returned an invalid stream ticket response", + ) + }) +}) + +function mockFetch() { + const calls: [URL | RequestInfo, RequestInit | undefined][] = [] + const sequence: (() => Response | Promise)[] = [] + let fallback: (() => Response | Promise) | undefined + + const self = { + calls, + resolved(value: Response) { + sequence.push(() => value) + return self + }, + repeated(factory: () => Response) { + fallback = factory + return self + }, + rejected(error: unknown) { + sequence.push(() => Promise.reject(error)) + return self + }, + get fetch(): typeof fetch { + return ((input: URL | RequestInfo, init?: RequestInit) => { + calls.push([input, init]) + const next = sequence.shift() ?? fallback + if (!next) throw new Error("unexpected fetch call") + return Promise.resolve(next()) + }) as typeof fetch + }, + } + + return self +} diff --git a/packages/opencode/test/kilocode/cloud/transport.test.ts b/packages/opencode/test/kilocode/cloud/transport.test.ts new file mode 100644 index 0000000000..0e1c560207 --- /dev/null +++ b/packages/opencode/test/kilocode/cloud/transport.test.ts @@ -0,0 +1,260 @@ +import { describe, expect, test } from "bun:test" +import { + AgentSendRequestSchema, + AgentStartRequestSchema, + GetMessageResultInputSchema, + MessageIdSchema, +} from "../../../src/kilocode/cloud/contracts" +import { parseServiceOrigin } from "../../../src/kilocode/cloud/origin" +import { MAX_CLOUD_AGENT_RESPONSE_BYTES } from "../../../src/kilocode/cloud/response-json" +import { createCloudAgentClient } from "../../../src/kilocode/cloud/trpc" + +const SESSION = "agent_12345678-1234-1234-1234-123456789abc" +const OTHER_SESSION = "agent_abcdefab-cdef-4abc-8def-abcdefabcdef" +const MESSAGE = "msg_018f1e2d3c4bAbCdEfGhIjKlMn" +const OTHER_MESSAGE = "msg_018f1e2d3c4bZyXwVuTsRqPoNm" +const SESSION_MESSAGE = "msg_018f1e2d3c4bQrStUvWxYzAbCd" +const TOKEN = "secret-bearer-value" + +type Seen = { + readonly url: string + readonly auth: string | null + readonly body: string +} + +function success(data: unknown) { + return Response.json({ result: { data } }) +} + +describe("Cloud Agent transport", () => { + test("places bearer auth only in headers and correlates generated message identities", async () => { + const seen: Seen[] = [] + const ids: string[] = [] + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + const url = new URL(request.url) + const body = request.method === "POST" ? await request.text() : "" + seen.push({ url: url.toString(), auth: request.headers.get("authorization"), body }) + + if (url.pathname === "/trpc/start") { + const input = AgentStartRequestSchema.parse(JSON.parse(body) as unknown) + const id = MessageIdSchema.parse(input.message.id) + ids.push(id) + return success({ + cloudAgentSessionId: input.message.prompt === "invalid-session" ? "invalid" : SESSION, + kiloSessionId: "ses_123", + messageId: input.message.prompt === "mismatch" ? OTHER_MESSAGE : id, + delivery: "queued", + }) + } + + if (url.pathname === "/trpc/send") { + const input = AgentSendRequestSchema.parse(JSON.parse(body) as unknown) + const id = MessageIdSchema.parse(input.message.id) + ids.push(id) + return success({ + cloudAgentSessionId: + input.message.prompt === "mismatch-session" ? OTHER_SESSION : input.cloudAgentSessionId, + status: "started", + streamUrl: "wss://cloud-agent.example/stream", + messageId: input.message.prompt === "mismatch-send" ? OTHER_MESSAGE : id, + delivery: "queued", + }) + } + + if (url.pathname === "/trpc/getMessageResult") { + const raw = url.searchParams.get("input") + const input = GetMessageResultInputSchema.parse(JSON.parse(raw ?? "null") as unknown) + return success({ + cloudAgentSessionId: input.messageId === SESSION_MESSAGE ? OTHER_SESSION : input.cloudAgentSessionId, + messageId: input.messageId === OTHER_MESSAGE ? MESSAGE : input.messageId, + status: "failed", + createdAt: 1, + terminalAt: 2, + completionSource: "delivery_failure", + failure: { + stage: "pre_dispatch", + code: "workspace_setup_failed", + subtype: "git_clone_timeout", + attempts: 1, + message: "Repository clone timed out", + retryable: true, + }, + }) + } + + return new Response(null, { status: 404 }) + }, + }) + + try { + const agent = createCloudAgentClient({ + origin: parseServiceOrigin(server.url.origin, { allowHttpLoopback: true }), + apiKey: TOKEN, + }) + const start = await agent.start({ + message: { prompt: "Inspect the repository" }, + agent: { mode: "code", model: "anthropic/claude-sonnet-4" }, + repository: { type: "github", repo: "Kilo-Org/kilocode" }, + options: { createdOnPlatform: "kilo-cli" }, + }) + const send = await agent.send({ + cloudAgentSessionId: start.cloudAgentSessionId, + message: { prompt: "Continue" }, + }) + const result = await agent.getMessageResult({ + cloudAgentSessionId: send.cloudAgentSessionId, + messageId: send.messageId, + }) + + expect(MessageIdSchema.safeParse(start.messageId).success).toBe(true) + expect(MessageIdSchema.safeParse(send.messageId).success).toBe(true) + expect(ids).toEqual([start.messageId, send.messageId]) + expect(result.failure).toEqual({ + stage: "pre_dispatch", + code: "workspace_setup_failed", + subtype: "git_clone_timeout", + attempts: 1, + message: "Repository clone timed out", + retryable: true, + }) + expect(seen).toHaveLength(3) + expect(seen.every((request) => request.auth === `Bearer ${TOKEN}`)).toBe(true) + expect(seen.every((request) => !request.url.includes(TOKEN) && !request.body.includes(TOKEN))).toBe(true) + + const error = await agent + .start({ + message: { prompt: "mismatch" }, + agent: { mode: "code", model: "anthropic/claude-sonnet-4" }, + repository: { type: "github", repo: "Kilo-Org/kilocode" }, + options: { createdOnPlatform: "kilo-cli" }, + }) + .then( + () => new Error("Expected start correlation to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(error.message).toBe("Cloud Agent start outcome is unknown; do not retry automatically") + expect(error.message).not.toContain(TOKEN) + + const malformed = await agent + .start({ + message: { prompt: "invalid-session" }, + agent: { mode: "code", model: "anthropic/claude-sonnet-4" }, + repository: { type: "github", repo: "Kilo-Org/kilocode" }, + options: { createdOnPlatform: "kilo-cli" }, + }) + .then( + () => new Error("Expected invalid session ID to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(malformed.message).toBe("Cloud Agent start outcome is unknown; do not retry automatically") + + const sendError = await agent.send({ cloudAgentSessionId: SESSION, message: { prompt: "mismatch-send" } }).then( + () => new Error("Expected send correlation to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(sendError.message).toBe("Cloud Agent send outcome is unknown; do not retry automatically") + + const resultError = await agent.getMessageResult({ cloudAgentSessionId: SESSION, messageId: OTHER_MESSAGE }).then( + () => new Error("Expected result correlation to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(resultError.message).toBe("Cloud Agent returned an invalid response") + + const sendSessionError = await agent + .send({ cloudAgentSessionId: SESSION, message: { prompt: "mismatch-session" } }) + .then( + () => new Error("Expected send session correlation to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(sendSessionError.message).toBe("Cloud Agent send outcome is unknown; do not retry automatically") + + const resultSessionError = await agent + .getMessageResult({ cloudAgentSessionId: SESSION, messageId: SESSION_MESSAGE }) + .then( + () => new Error("Expected result session correlation to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(resultSessionError.message).toBe("Cloud Agent returned an invalid response") + } finally { + await server.stop(true) + } + }) + + test("rejects malformed send responses before they cross the client boundary", async () => { + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch() { + return success({ + cloudAgentSessionId: "invalid", + status: "started", + streamUrl: "", + messageId: MESSAGE, + delivery: "queued", + }) + }, + }) + + try { + const agent = createCloudAgentClient({ + origin: parseServiceOrigin(server.url.origin, { allowHttpLoopback: true }), + apiKey: TOKEN, + id: () => MESSAGE, + }) + const error = await agent.send({ cloudAgentSessionId: SESSION, message: { prompt: "Continue" } }).then( + () => new Error("Expected send to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(error.message).toBe("Cloud Agent send outcome is unknown; do not retry automatically") + } finally { + await server.stop(true) + } + }) + + test("treats redirects, oversized bodies, and malformed envelopes as unknown start outcomes", async () => { + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + const input = AgentStartRequestSchema.parse((await request.json()) as unknown) + if (input.message.prompt === "redirect") { + return new Response(null, { status: 302, headers: { location: "/elsewhere" } }) + } + if (input.message.prompt === "oversized") { + return new Response(null, { + headers: { "content-length": String(MAX_CLOUD_AGENT_RESPONSE_BYTES + 1) }, + }) + } + if (input.message.prompt === "unavailable") return new Response(null, { status: 503 }) + return Response.json({ invalid: true }) + }, + }) + + try { + const agent = createCloudAgentClient({ + origin: parseServiceOrigin(server.url.origin, { allowHttpLoopback: true }), + apiKey: TOKEN, + }) + const start = (prompt: string) => + agent.start({ + message: { prompt }, + agent: { mode: "code", model: "anthropic/claude-sonnet-4" }, + repository: { type: "github", repo: "Kilo-Org/kilocode" }, + options: { createdOnPlatform: "kilo-cli" }, + }) + + for (const prompt of ["redirect", "oversized", "malformed", "unavailable"]) { + const error = await start(prompt).then( + () => new Error("Expected start to fail"), + (cause: unknown) => (cause instanceof Error ? cause : new Error("Non-error rejection")), + ) + expect(error.message).toBe("Cloud Agent start outcome is unknown; do not retry automatically") + } + } finally { + await server.stop(true) + } + }) +}) diff --git a/packages/opencode/test/kilocode/cloud/websocket-stream.test.ts b/packages/opencode/test/kilocode/cloud/websocket-stream.test.ts new file mode 100644 index 0000000000..5835986acc --- /dev/null +++ b/packages/opencode/test/kilocode/cloud/websocket-stream.test.ts @@ -0,0 +1,281 @@ +import { describe, expect, test } from "bun:test" +import { streamAgentEvents } from "@/kilocode/cloud/websocket-stream" + +function mockWebSocket( + events: ReadonlyArray< + { type: "message"; data: string | ArrayBuffer } | { type: "error" } | { type: "close"; code?: number } + >, + options?: { onClose?: (code?: number) => void; triggerOnCloseOnClose?: boolean }, +) { + return class MockWebSocket { + onmessage: ((event: MessageEvent) => void) | null = null + onerror: (() => void) | null = null + onclose: ((event: CloseEvent) => void) | null = null + + constructor(_url: string) { + queueMicrotask(() => { + for (const event of events) { + if (event.type === "message") { + this.onmessage?.(new MessageEvent("message", { data: event.data })) + } else if (event.type === "error") { + this.onerror?.() + return + } else if (event.type === "close") { + this.onclose?.({ code: event.code ?? 1000 } as CloseEvent) + return + } + } + }) + } + + close(code?: number) { + options?.onClose?.(code) + if (options?.triggerOnCloseOnClose) { + queueMicrotask(() => { + this.onclose?.({ code: code ?? 1000 } as CloseEvent) + }) + } + } + } +} + +describe("streamAgentEvents", () => { + test("writes WebSocket text messages as lines", async () => { + const lines: string[] = [] + const Socket = mockWebSocket([ + { type: "message", data: '{"event":"one"}' }, + { type: "message", data: '{"event":"two"}' }, + { type: "close" }, + ]) + + await streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: (line) => { + lines.push(line) + }, + WebSocket: Socket as unknown as typeof WebSocket, + }) + + expect(lines).toEqual(['{"event":"one"}', '{"event":"two"}']) + }) + + test("resolves an absolute wss URL", async () => { + const Socket = mockWebSocket([{ type: "close" }]) + const connectUrl: string[] = [] + + class Tracked extends Socket { + constructor(url: string) { + super(url) + connectUrl.push(url) + } + } + + await streamAgentEvents({ + streamUrl: "wss://agent.example/stream?ticket=tok", + origin: "https://agent.example", + writeLine: () => {}, + WebSocket: Tracked as unknown as typeof WebSocket, + }) + + expect(connectUrl).toEqual(["wss://agent.example/stream?ticket=tok"]) + }) + + test("rejects an absolute stream URL on another origin", async () => { + const Socket = mockWebSocket([{ type: "close" }]) + + await expect( + streamAgentEvents({ + streamUrl: "wss://other.example/stream?ticket=tok", + origin: "https://agent.example", + writeLine: () => {}, + WebSocket: Socket as unknown as typeof WebSocket, + }), + ).rejects.toThrow("Invalid stream URL origin") + }) + + test("converts a relative URL to an absolute wss URL", async () => { + const Socket = mockWebSocket([{ type: "close" }]) + const connectUrl: string[] = [] + + class Tracked extends Socket { + constructor(url: string) { + super(url) + connectUrl.push(url) + } + } + + await streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: () => {}, + WebSocket: Tracked as unknown as typeof WebSocket, + }) + + expect(connectUrl).toEqual(["wss://agent.example/stream?cloudAgentSessionId=agent_123&ticket=tok"]) + }) + + test("rejects when the WebSocket errors", async () => { + const Socket = mockWebSocket([{ type: "error" }]) + + await expect( + streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: () => {}, + WebSocket: Socket as unknown as typeof WebSocket, + }), + ).rejects.toThrow("WebSocket stream connection failed") + }) + + test("rejects when the WebSocket closes abnormally", async () => { + const Socket = mockWebSocket([{ type: "close", code: 1011 }]) + + await expect( + streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: () => {}, + WebSocket: Socket as unknown as typeof WebSocket, + }), + ).rejects.toThrow("WebSocket stream closed unexpectedly (1011)") + }) + + test("rejects when the WebSocket stream stalls", async () => { + const Socket = mockWebSocket([]) + + await expect( + streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: () => {}, + WebSocket: Socket as unknown as typeof WebSocket, + timeoutMs: 1, + }), + ).rejects.toThrow("WebSocket stream timed out") + }) + + test("resolves 3 seconds after receiving a complete event", async () => { + const lines: string[] = [] + const codes: Array = [] + const Socket = mockWebSocket( + [ + { type: "message", data: '{"event":"running"}' }, + { type: "message", data: '{"streamEventType":"complete","data":{"exitCode":0}}' }, + ], + { onClose: (code) => codes.push(code), triggerOnCloseOnClose: true }, + ) + + const start = Date.now() + await streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: (line) => { + lines.push(line) + }, + WebSocket: Socket as unknown as typeof WebSocket, + }) + + expect(Date.now() - start).toBeGreaterThanOrEqual(3000) + expect(codes).toEqual([1000]) + expect(lines).toEqual(['{"event":"running"}', '{"streamEventType":"complete","data":{"exitCode":0}}']) + }, 10_000) + + test("flushes slow writes in order before resolving", async () => { + const lines: string[] = [] + const Socket = mockWebSocket([ + { type: "message", data: '{"event":"one"}' }, + { type: "message", data: '{"event":"two"}' }, + { type: "close" }, + ]) + + await streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: async (line) => { + await new Promise((resolve) => setTimeout(resolve, 10)) + lines.push(line) + }, + WebSocket: Socket as unknown as typeof WebSocket, + }) + + expect(lines).toEqual(['{"event":"one"}', '{"event":"two"}']) + }) + + test("flushes slow writes in order before rejecting a transport failure", async () => { + const lines: string[] = [] + const Socket = mockWebSocket([ + { type: "message", data: '{"event":"one"}' }, + { type: "message", data: '{"event":"two"}' }, + { type: "close", code: 1011 }, + ]) + + await expect( + streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: async (line) => { + await new Promise((resolve) => setTimeout(resolve, 10)) + lines.push(line) + }, + WebSocket: Socket as unknown as typeof WebSocket, + }), + ).rejects.toThrow("WebSocket stream closed unexpectedly (1011)") + expect(lines).toEqual(['{"event":"one"}', '{"event":"two"}']) + }) + + test("bounds transport failure draining when an output write stalls", async () => { + const Socket = mockWebSocket([ + { type: "message", data: '{"event":"one"}' }, + { type: "close", code: 1011 }, + ]) + + await expect( + streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: () => new Promise(() => {}), + WebSocket: Socket as unknown as typeof WebSocket, + timeoutMs: 10, + }), + ).rejects.toThrow("WebSocket stream closed unexpectedly (1011)") + }) + + test("rejects when a stream output write fails", async () => { + const lines: string[] = [] + const Socket = mockWebSocket([ + { type: "message", data: '{"event":"one"}' }, + { type: "message", data: '{"event":"two"}' }, + { type: "close" }, + ]) + + await expect( + streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: (line) => { + if (lines.length > 0) throw new Error("EPIPE") + lines.push(line) + }, + WebSocket: Socket as unknown as typeof WebSocket, + }), + ).rejects.toThrow("WebSocket stream output failed") + expect(lines).toEqual(['{"event":"one"}']) + }) + + test("rejects when queued stream output exceeds the memory bound", async () => { + const line = "x".repeat(1024) + const Socket = mockWebSocket( + Array.from({ length: 9000 }, () => ({ type: "message" as const, data: line })), + ) + + await expect( + streamAgentEvents({ + streamUrl: "/stream?cloudAgentSessionId=agent_123&ticket=tok", + origin: "https://agent.example", + writeLine: () => new Promise(() => {}), + WebSocket: Socket as unknown as typeof WebSocket, + }), + ).rejects.toThrow("WebSocket stream output consumer is too slow") + }) +}) diff --git a/packages/opencode/test/kilocode/help.test.ts b/packages/opencode/test/kilocode/help.test.ts index fb94213b71..bb1179b38a 100644 --- a/packages/opencode/test/kilocode/help.test.ts +++ b/packages/opencode/test/kilocode/help.test.ts @@ -1,5 +1,6 @@ import { describe, test, expect } from "bun:test" import path from "path" +import yargs from "yargs" import { generateHelp, generateCommandTable } from "../../src/kilocode/help" import { AcpCommand } from "../../src/cli/cmd/acp" import { McpCommand } from "../../src/cli/cmd/mcp" @@ -26,6 +27,7 @@ import { HelpCommand } from "../../src/kilocode/help-command" import { ProfileCommand } from "../../src/kilocode/cli/cmd/profile" import { DaemonCommand } from "../../src/kilocode/cli/cmd/daemon" import { KiloConsoleCommand } from "../../src/kilocode/cli/cmd/console" +import { CloudCommand } from "../../src/kilocode/cli/cmd/cloud" // Stand-in for TuiThreadCommand — the real one imports @opentui/solid which // doesn't resolve in the test environment. Only command/describe matter here. @@ -76,6 +78,7 @@ const commands = [ ProfileCommand, DaemonCommand, KiloConsoleCommand, + CloudCommand, HelpCommand, CompletionStub, ] as any[] @@ -140,6 +143,29 @@ describe("kilo help ", () => { }) }) +describe("kilo cloud help", () => { + async function parser() { + const cli = yargs([]) + .scriptName("kilo cloud") + .exitProcess(false) + .help() + .fail((msg, err) => { + throw err ?? new Error(msg) + }) + if (typeof CloudCommand.builder !== "function") throw new Error("cloud command builder is missing") + return await CloudCommand.builder(cli) + } + + test("requires a subcommand and exposes only the public Cloud Agent operations", async () => { + const bare = await parser() + await expect(Promise.resolve().then(() => bare.parseAsync([]))).rejects.toThrow() + + const help = await (await parser()).getHelp() + const names = [...help.matchAll(/^\s*kilo cloud ([a-z][a-z-]*)\b/gm)].map((match) => match[1]) + expect([...new Set(names)].sort()).toEqual(["result", "send", "start", "status"]) + }) +}) + describe("edge cases", () => { test("output contains no ANSI escape sequences", async () => { const output = await generateHelp({ all: true, format: "md", commands })