Compare commits

..
40 changed files with 465 additions and 1411 deletions
@@ -18,7 +18,6 @@ const WebSocketResponseCreate = Schema.StructWithRest(Schema.Struct({ type: Sche
])
const decodeMessage = ProviderShared.validateWith(Schema.decodeUnknownEffect(WebSocketResponseCreate))
const encodeMessage = Schema.encodeSync(Schema.fromJsonString(WebSocketResponseCreate))
const decodeEvent = Schema.decodeUnknownEffect(OpenResponses.protocol.stream.event)
export interface Options {
readonly id: string
@@ -27,6 +26,7 @@ export interface Options {
readonly enabled?: (url: string) => boolean
readonly url?: (url: string) => string
readonly headers?: (headers: Headers.Headers) => Headers.Headers
readonly continuation?: OpenResponsesContinuation.Shape
}
export interface Prepared {
@@ -60,7 +60,7 @@ const driver = (options: Options, body: string): WebSocketChannelDriver => {
}),
observe: (_create, frame) =>
Effect.gen(function* () {
const event = yield* decodeEvent(frame).pipe(
const event = yield* OpenResponses.decodeChannelEvent(frame).pipe(
Effect.mapError((cause) =>
ProviderShared.eventError(options.id, `Invalid ${options.name} WebSocket event`, frame, cause),
),
@@ -163,6 +163,7 @@ export const transport = <Body>(options: Options): Transport<Body, Prepared, str
request: create.request,
message: create.message,
base,
continuation: options.continuation,
}),
}
})
@@ -6,7 +6,6 @@ import { OpenResponses } from "./open-responses.js"
const PROTOCOL = "open-responses.websocket.v1"
const VERSION = 1
const decodeEvent = Schema.decodeUnknownEffect(OpenResponses.protocol.stream.event)
interface CheckpointValue {
readonly version: typeof VERSION
@@ -15,12 +14,19 @@ interface CheckpointValue {
readonly output: ReadonlyArray<unknown>
}
/**
* Fields to send next to `previous_response_id` on an incremental step, or undefined to send the step in full.
* Whether omitted fields carry over from the continued response is provider behavior the route must know.
*/
export type Shape = (request: Readonly<Record<string, unknown>>) => Readonly<Record<string, unknown>> | undefined
export interface DriverInput {
readonly id: string
readonly name: string
readonly request: Readonly<Record<string, unknown>>
readonly message: string
readonly base: WebSocketChannelDriver
readonly continuation?: Shape
}
const checkpointValue = (checkpoint: ChannelCheckpoint | undefined): CheckpointValue | undefined => {
@@ -127,22 +133,26 @@ const rejected = (
export const driver = (input: DriverInput): WebSocketChannelDriver => {
const { previous_response_id: _previousResponseID, ...request } = input.request
const shape = input.continuation ?? ((fields: Readonly<Record<string, unknown>>) => fields)
let output: OpenResponses.StreamItem[] = []
return {
create: (checkpoint) =>
Effect.sync(() => {
output = []
const previous = checkpointValue(checkpoint)
const delta = previous ? incremental(request, previous) : undefined
if (!previous || !delta) return { message: ProviderShared.encodeJson(request), mode: "full" as const }
// Ask the route first: diffing the whole history is wasted when it declines the continuation.
const fields = previous ? shape(request) : undefined
const delta = previous && fields ? incremental(request, previous) : undefined
if (!previous || !fields || !delta)
return { message: ProviderShared.encodeJson(request), mode: "full" as const }
return {
message: ProviderShared.encodeJson({ ...request, input: delta, previous_response_id: previous.responseID }),
message: ProviderShared.encodeJson({ ...fields, input: delta, previous_response_id: previous.responseID }),
mode: "incremental" as const,
}
}),
observe: (create, frame) =>
Effect.gen(function* () {
const event = yield* decodeEvent(frame).pipe(
const event = yield* OpenResponses.decodeChannelEvent(frame).pipe(
Effect.mapError((cause) =>
ProviderShared.eventError(input.id, `Invalid ${input.name} WebSocket event`, frame, cause),
),
@@ -195,4 +205,4 @@ export const driver = (input: DriverInput): WebSocketChannelDriver => {
}
}
export const OpenResponsesContinuation = { driver } as const
export * as OpenResponsesContinuation from "./open-responses-continuation.js"
@@ -405,6 +405,24 @@ export const Event = Schema.StructWithRest(
export type Event = Schema.Schema.Type<typeof Event>
export type NormalizedEvent = Event & { readonly item?: OutputItem | null }
const decodeEventValue = Schema.decodeUnknownEffect(Event)
const decodeFrame = Schema.decodeUnknownEffect(ProviderShared.Json)
/**
* Decodes one WebSocket frame. xAI answers a rejected `response.create` with `{ "error": { "message", "type" } }` and no
* event type; that envelope reads as an error event so the failure classifies instead of failing decoding.
*/
export const decodeChannelEvent = (frame: string) =>
decodeFrame(frame).pipe(
Effect.flatMap((value) =>
decodeEventValue(
ProviderShared.isRecord(value) && value.type === undefined && ProviderShared.isRecord(value.error)
? { ...value, type: "error" }
: value,
),
),
)
export interface ProviderAdapter {
readonly id: string
readonly name: string
+4
View File
@@ -41,6 +41,10 @@ const responsesRoute = Route.make({
id: "openai-responses",
name: "xAI Responses",
rotateAfterMs: RESPONSES_WEBSOCKET_ROTATE_AFTER_MS,
// xAI continues a chain only from stored responses: with `store: false` (the route default) `previous_response_id`
// fails with "Response with id=… not found", so those steps are sent in full over the reused connection. It also
// rejects `instructions` next to `previous_response_id` and keeps the instructions of the response it continues.
continuation: ({ instructions: _instructions, ...request }) => (request.store === false ? undefined : request),
}),
defaults: { providerOptions: { store: false, include: ["reasoning.encrypted_content"] } },
})
@@ -90,7 +90,11 @@ const classifyingChannelDriver = (message: string): WebSocketChannelDriver => {
}
}
const continuationDriver = (request: Readonly<Record<string, unknown>>, base = baseChannelDriver) => {
const continuationDriver = (
request: Readonly<Record<string, unknown>>,
base = baseChannelDriver,
continuation?: OpenResponsesContinuation.Shape,
) => {
const message = ProviderShared.encodeJson(request)
return OpenResponsesContinuation.driver({
id: "openai-responses",
@@ -98,6 +102,7 @@ const continuationDriver = (request: Readonly<Record<string, unknown>>, base = b
request,
message,
base: base(message),
continuation,
})
}
@@ -921,6 +926,58 @@ describe("OpenAI Responses route", () => {
type: "provider-failure",
error: { reason: { _tag: "InvalidRequest", classification: "context-overflow" } },
})
// A retryable failure stays one: the runner retries it, and the transport has already dropped the
// checkpoint, so that retry is a full send. xAI reports every rejection this way.
const internal = ProviderShared.encodeJson({
type: "error",
error: { type: "api_error", message: "gRPC error: Response with id=resp_1 not found" },
})
expect(yield* second.observe(yield* second.create(saved), internal)).toMatchObject({
type: "provider-failure",
error: { reason: { _tag: "ProviderInternal" } },
})
}),
)
it.effect("shapes the incremental send with the route continuation", () =>
Effect.gen(function* () {
const firstRequest = {
type: "response.create",
model: "grok-4.6",
store: true,
instructions: "You are terse.",
input: [{ role: "user", content: [{ type: "input_text", text: "First" }] }],
}
const secondRequest = {
...firstRequest,
input: [...firstRequest.input, { role: "user", content: [{ type: "input_text", text: "Second" }] }],
}
const saved = checkpoint(
yield* continuationDriver(firstRequest).observe(
yield* continuationDriver(firstRequest).create(undefined),
ProviderShared.encodeJson({ type: "response.completed", response: { id: "resp_1" } }),
),
)
const trimmed = yield* continuationDriver(
secondRequest,
baseChannelDriver,
({ instructions: _, ...rest }) => rest,
).create(saved)
expect(trimmed.mode).toBe("incremental")
expect(JSON.parse(trimmed.message)).toEqual({
type: "response.create",
model: "grok-4.6",
store: true,
previous_response_id: "resp_1",
input: [{ role: "user", content: [{ type: "input_text", text: "Second" }] }],
})
// Declining the continuation sends the step in full and never sends a previous_response_id.
const declined = yield* continuationDriver(secondRequest, baseChannelDriver, () => undefined).create(saved)
expect(declined.mode).toBe("full")
expect(JSON.parse(declined.message)).toEqual(secondRequest)
}),
)
+110 -2
View File
@@ -1,11 +1,18 @@
import { describe, expect } from "bun:test"
import { Effect } from "effect"
import { Effect, Layer, Stream } from "effect"
import { LLM, LLMEvent, Message } from "../../src/index.js"
import { XAI } from "../../src/providers.js"
import { OpenResponses } from "../../src/protocols/open-responses.js"
import { OpenAIResponses } from "../../src/protocols/openai-responses.js"
import * as ProviderShared from "../../src/protocols/shared.js"
import { XAIResponses } from "../../src/protocols/xai-responses.js"
import { LLMClient } from "../../src/route.js"
import {
LLMClient,
RequestExecutor,
WebSocketTransport,
type ChannelCheckpoint,
type WebSocketChannelDriver,
} from "../../src/route.js"
import { compileRequest } from "../../src/route/client.js"
import { it } from "../lib/effect.js"
import { fixedResponse } from "../lib/http.js"
@@ -13,6 +20,35 @@ import { sseEvents } from "../lib/sse.js"
const model = XAI.configure({ apiKey: "test", baseURL: "https://api.x.ai/v1" }).responses("grok-4.6")
/** Runs a request through the WebSocket transport and hands back its channel driver; the HTTP fallback answers. */
const channelDriver = (request: ReturnType<typeof LLM.request>) =>
Effect.gen(function* () {
let driver: WebSocketChannelDriver | undefined
yield* LLMClient.generate(request, {
webSocket: {
execute: (exchange) =>
Effect.sync(() => {
driver = exchange.driver
return { frames: exchange.fallback(), complete: Effect.void }
}),
},
}).pipe(Effect.provide(fixedResponse(sseEvents({ type: "response.completed", response: { id: "http" } }))))
if (!driver) throw new Error("Expected a WebSocket channel driver")
return driver
})
const completed = (driver: WebSocketChannelDriver, id: string) =>
Effect.gen(function* () {
const create = yield* driver.create(undefined)
yield* driver.observe(create, ProviderShared.encodeJson({ type: "response.created", response: { id } }))
const observation = yield* driver.observe(
create,
ProviderShared.encodeJson({ type: "response.completed", response: { id } }),
)
if (observation.type !== "completed" || !observation.checkpoint) throw new Error("Expected a checkpoint")
return observation.checkpoint
})
describe("xAI Responses route", () => {
it.effect("composes the Open Responses baseline with xAI extensions", () =>
Effect.gen(function* () {
@@ -162,6 +198,78 @@ describe("xAI Responses route", () => {
}),
)
it.effect("classifies xAI's untyped WebSocket error envelope", () =>
Effect.gen(function* () {
// xAI answers a rejected response.create with an error envelope that carries no event type.
const envelope = ProviderShared.encodeJson({
error: {
message:
'Request validation error: {"code":"400","error":"Argument not supported: instructions and previous_response_id together"}',
type: "api_error",
},
})
const webSocket = WebSocketTransport.makeDirect({
open: () =>
Effect.succeed({ sendText: () => Effect.void, messages: Stream.make(envelope), close: Effect.void }),
})
const error = yield* LLMClient.generate(LLM.request({ model, prompt: "Hello" }), { webSocket }).pipe(
Effect.provide(
LLMClient.layer.pipe(
Layer.provide(
Layer.succeed(
RequestExecutor.Service,
RequestExecutor.Service.of({ execute: () => Effect.die("unexpected HTTP request") }),
),
),
),
),
Effect.flip,
)
expect(error.reason._tag).toBe("ProviderInternal")
expect(error.message).toContain("Argument not supported: instructions and previous_response_id together")
expect(error.reason.body).toBe(envelope)
}),
)
it.effect("continues stored responses without instructions and sends unstored steps in full", () =>
Effect.gen(function* () {
const step = (store: boolean, ...prompts: string[]) =>
LLM.request({
model,
system: "You are terse.",
messages: prompts.map((prompt) => Message.user(prompt)),
providerOptions: { store },
})
const send = (store: boolean, checkpoint: ChannelCheckpoint) =>
channelDriver(step(store, "First", "Second")).pipe(Effect.flatMap((driver) => driver.create(checkpoint)))
const stored = yield* send(true, yield* completed(yield* channelDriver(step(true, "First")), "resp_1"))
expect(stored.mode).toBe("incremental")
expect(JSON.parse(stored.message)).toEqual({
type: "response.create",
model: "grok-4.6",
store: true,
include: ["reasoning.encrypted_content"],
previous_response_id: "resp_1",
input: [{ role: "user", content: [{ type: "input_text", text: "Second" }] }],
})
// The connection cache only serves stored responses, so the default store: false never chains.
const unstored = yield* send(false, yield* completed(yield* channelDriver(step(false, "First")), "resp_1"))
expect(unstored.mode).toBe("full")
expect(JSON.parse(unstored.message)).toMatchObject({
instructions: "You are terse.",
store: false,
input: [
{ role: "user", content: [{ type: "input_text", text: "First" }] },
{ role: "user", content: [{ type: "input_text", text: "Second" }] },
],
})
expect(JSON.parse(unstored.message).previous_response_id).toBeUndefined()
}),
)
it.effect("parses xAI hosted tool items", () =>
Effect.gen(function* () {
const item = { type: "x_search_call", id: "x_search_1", status: "completed", action: { query: "news" } }
+5 -10
View File
@@ -126,15 +126,9 @@ export function createComposerSubmit(input: ComposerSubmitInput) {
if (command) {
clearSubmission(input, submission)
// Commands always steer: the server applies a command's configured
// agent and model immediately at admission, so queueing one would
// reconfigure the turn it is supposed to wait behind.
void sendCommand(
session,
{ ...value, delivery: "steer" },
command,
input.adapter.controls().model.selection.trackSessionCommit,
).catch((error) => failSubmission(input, session, "command", error, restore, value.id))
void sendCommand(session, value, command, input.adapter.controls().model.selection.trackSessionCommit).catch(
(error) => failSubmission(input, session, "command", error, restore, value.id),
)
return
}
} finally {
@@ -326,7 +320,8 @@ async function sendCommand(
track?: ModelSelection["trackSessionCommit"],
) {
const request = await buildSubmissionRequest(session, value)
await applySelection(session, value.selection, track)
// Like queued prompts, queued commands must not apply the composer's selection to active work.
if (value.delivery === "steer") await applySelection(session, value.selection, track)
await session.api.command({
sessionID: session.id,
command: command.command,
+1 -11
View File
@@ -17,7 +17,6 @@ import type { PromptInput } from "@opencode/schema/prompt-input"
import type { AgentAttachment } from "@opencode/schema/prompt"
import type { Skill } from "@opencode/schema/skill"
import type { Event } from "@opencode/schema/event"
import type { FileDiff } from "@opencode/schema/file-diff"
import type { InstructionEntry } from "@opencode/schema/instruction-entry"
import type { Schema } from "effect"
import type { EventLog } from "@opencode/schema/event-log"
@@ -37,6 +36,7 @@ import type { PtyTicket } from "@opencode/schema/pty-ticket"
import type { Reference } from "@opencode/schema/reference"
import type { Worktree } from "@opencode/schema/worktree"
import type { Vcs } from "@opencode/schema/vcs"
import type { FileDiff } from "@opencode/schema/file-diff"
import type { WebSearch } from "@opencode/schema/websearch"
import type { Config } from "@opencode/schema/config"
@@ -360,15 +360,6 @@ export type SessionContextInput = { readonly sessionID: Session.ID }
export type SessionContextOutput = ReadonlyArray<SessionMessage.Info>
export type SessionContextOperation<E = never> = (input: SessionContextInput) => Effect.Effect<SessionContextOutput, E>
export type SessionDiffInput = {
readonly sessionID: Session.ID
readonly messageID?: SessionMessage.ID | undefined
readonly to?: SessionMessage.ID | undefined
readonly context?: number | undefined
}
export type SessionDiffOutput = ReadonlyArray<FileDiff.Info>
export type SessionDiffOperation<E = never> = (input: SessionDiffInput) => Effect.Effect<SessionDiffOutput, E>
export type SessionInboxListInput = { readonly sessionID: Session.ID }
export type SessionInboxListOutput = ReadonlyArray<SessionInbox.Info>
export type SessionInboxListOperation<E = never> = (
@@ -1148,7 +1139,6 @@ export interface SessionApi<E = never> {
readonly commit: SessionRevertCommitOperation<E>
}
readonly context: SessionContextOperation<E>
readonly diff: SessionDiffOperation<E>
readonly inbox: {
readonly list: SessionInboxListOperation<E>
readonly cancel: SessionInboxCancelOperation<E>
@@ -68,8 +68,6 @@ import type {
SessionRevertCommitOutput,
SessionContextInput,
SessionContextOutput,
SessionDiffInput,
SessionDiffOutput,
SessionInboxListInput,
SessionInboxListOutput,
SessionInboxCancelInput,
@@ -594,17 +592,6 @@ const EndpointSessionContext = (raw: RawClient["server.session"]) => (input: Ses
),
)
const EndpointSessionDiff = (raw: RawClient["server.session"]) => (input: SessionDiffInput) =>
preserveEffect<SessionDiffOutput>()(
raw["session.diff"]({
params: { sessionID: input["sessionID"] },
query: { messageID: input["messageID"], to: input["to"], context: input["context"] },
}).pipe(
Effect.mapError(mapClientError),
Effect.map((value) => value.data),
),
)
const EndpointSessionInboxList = (raw: RawClient["server.session"]) => (input: SessionInboxListInput) =>
preserveEffect<SessionInboxListOutput>()(
raw["session.inbox.list"]({ params: { sessionID: input["sessionID"] } }).pipe(
@@ -744,7 +731,6 @@ const adaptGroupSession = (raw: RawClient["server.session"]) => ({
commit: EndpointSessionRevertCommit(raw),
},
context: EndpointSessionContext(raw),
diff: EndpointSessionDiff(raw),
inbox: {
list: EndpointSessionInboxList(raw),
cancel: EndpointSessionInboxCancel(raw),
@@ -62,8 +62,6 @@ import type {
SessionRevertCommitOutput,
SessionContextInput,
SessionContextOutput,
SessionDiffInput,
SessionDiffOutput,
SessionInboxListInput,
SessionInboxListOutput,
SessionInboxCancelInput,
@@ -844,18 +842,6 @@ export function make(options: ClientOptions) {
},
requestOptions,
).then((value) => value.data),
diff: (input: SessionDiffInput, requestOptions?: RequestOptions) =>
request<{ readonly data: SessionDiffOutput }>(
{
method: "GET",
path: `/api/session/${encodeURIComponent(input.sessionID)}/diff`,
query: { messageID: input["messageID"], to: input["to"], context: input["context"] },
successStatus: 200,
declaredStatuses: [400, 401, 404, 500],
empty: false,
},
requestOptions,
).then((value) => value.data),
inbox: {
list: (input: SessionInboxListInput, requestOptions?: RequestOptions) =>
request<{ readonly data: SessionInboxListOutput }>(
@@ -147,14 +147,6 @@ export type SessionProviderContextProvenance = {
endpoint: string
}
export type SessionMessageIdle = {
id: string
metadata?: { [x: string]: JsonValue }
time: { created: number }
type: "idle"
outcome: "succeeded" | "failed" | "interrupted"
}
export type SessionActive = { type: "running" }
export type SessionInboxDelivery = "steer" | "queue"
@@ -2202,7 +2194,6 @@ export type SessionMessageInfo =
| SessionMessageShell
| SessionMessageAssistant
| SessionMessageCompaction
| SessionMessageIdle
export type SessionMessageContentUpdated = {
id: string
@@ -3161,13 +3152,6 @@ export type SessionImportInput = {
}
}
)
| {
readonly id: string
readonly metadata?: { readonly [x: string]: JsonValue }
readonly time: { readonly created: number }
readonly type: "idle"
readonly outcome: "succeeded" | "failed" | "interrupted"
}
>
readonly location?: { readonly directory: string; readonly workspaceID?: string } | null
}["info"]
@@ -3473,13 +3457,6 @@ export type SessionImportInput = {
}
}
)
| {
readonly id: string
readonly metadata?: { readonly [x: string]: JsonValue }
readonly time: { readonly created: number }
readonly type: "idle"
readonly outcome: "succeeded" | "failed" | "interrupted"
}
>
readonly location?: { readonly directory: string; readonly workspaceID?: string } | null
}["messages"]
@@ -3785,13 +3762,6 @@ export type SessionImportInput = {
}
}
)
| {
readonly id: string
readonly metadata?: { readonly [x: string]: JsonValue }
readonly time: { readonly created: number }
readonly type: "idle"
readonly outcome: "succeeded" | "failed" | "interrupted"
}
>
readonly location?: { readonly directory: string; readonly workspaceID?: string } | null
}["location"]
@@ -4281,27 +4251,6 @@ export type SessionContextInput = { readonly sessionID: { readonly sessionID: st
export type SessionContextOutput = { data: Array<SessionMessageInfo> }["data"]
export type SessionDiffInput = {
readonly sessionID: { readonly sessionID: string }["sessionID"]
readonly messageID?: {
readonly messageID?: string | undefined
readonly to?: string | undefined
readonly context?: number | undefined
}["messageID"]
readonly to?: {
readonly messageID?: string | undefined
readonly to?: string | undefined
readonly context?: number | undefined
}["to"]
readonly context?: {
readonly messageID?: string | undefined
readonly to?: string | undefined
readonly context?: number | undefined
}["context"]
}
export type SessionDiffOutput = { data: Array<FileDiffInfo> }["data"]
export type SessionInboxListInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] }
export type SessionInboxListOutput = { data: Array<SessionInboxInfo> }["data"]
-12
View File
@@ -1024,18 +1024,6 @@ export function createData(config: CreateDataInput) {
if (currentAssistant) currentAssistant.retry = undefined
})
if (event.type === "session.execution.interrupted" && event.data.reason === "shutdown") return
// Mirror the projected idle marker so turn boundaries match before the next message read.
message.insert(event.data.sessionID, {
id: messageIDFromEvent(event.id),
type: "idle",
outcome:
event.type === "session.execution.succeeded"
? "succeeded"
: event.type === "session.execution.failed"
? "failed"
: "interrupted",
time: { created: event.created },
})
// An event can overtake the first read; queue a revalidation when that read is still active.
if (!store.session.info[event.data.sessionID] && !sync.has(`session:${event.data.sessionID}`)) return
result.session.invalidate(event.data.sessionID)
+19 -9
View File
@@ -1,6 +1,7 @@
export * as Generate from "./generate.js"
import { LLM, LLMClient, AIError } from "@opencode/ai"
import { SessionID } from "@opencode/schema/session-id"
import { Context, Effect, Layer, Schema } from "effect"
import { makeLocationNode } from "@opencode/util/effect/app-node"
import { llmClient } from "./effect/app-node-platform.js"
@@ -58,15 +59,24 @@ export const layer = Layer.effect(
? `Model unavailable: ${input.model.providerID}/${input.model.id}`
: "No model specified and no supported model is available",
})
const response = yield* llm.generate(LLM.request({ model: resolved.model, prompt: input.prompt })).pipe(
Effect.mapError(
(error: AIError) =>
new UnavailableError({
message: error.message,
service: resolved.ref.providerID,
}),
),
)
const response = yield* llm
.generate(
LLM.request({
model: resolved.model,
prompt: input.prompt,
// Gateways require session attribution even for a stateless call; no Session is stored.
http: { headers: { "x-opencode-session": SessionID.create() } },
}),
)
.pipe(
Effect.mapError(
(error: AIError) =>
new UnavailableError({
message: error.message,
service: resolved.ref.providerID,
}),
),
)
return response.text
})
+64 -75
View File
@@ -9,7 +9,6 @@ import { AppProcess } from "@opencode/util/process"
import { makeGlobalNode } from "@opencode/util/effect/app-node"
import { File } from "./file.js"
import { KeyedMutex } from "./effect/keyed-mutex.js"
import { VcsPatch } from "./vcs/patch.js"
export class Repository extends Schema.Class<Repository>("Git.Repository")({
worktree: AbsolutePath,
@@ -309,7 +308,7 @@ const layer = Layer.effect(
operationName: OperationError["operation"],
repository: Repository,
args: string[],
options?: { stdin?: string; env?: Record<string, string>; maxOutputBytes?: number },
options?: { stdin?: string; env?: Record<string, string> },
) {
const result = yield* proc
.run(
@@ -318,7 +317,7 @@ const layer = Layer.effect(
env: options?.env,
extendEnv: true,
}),
{ stdin: options?.stdin, maxOutputBytes: options?.maxOutputBytes },
{ stdin: options?.stdin },
)
.pipe(
Effect.mapError(
@@ -332,8 +331,7 @@ const layer = Layer.effect(
),
)
const text = result.stdout.toString("utf8")
if (result.exitCode === 0)
return { text, stderr: result.stderr.toString("utf8"), truncated: result.stdoutTruncated }
if (result.exitCode === 0) return { text, stderr: result.stderr.toString("utf8") }
return yield* new OperationError({
operation: operationName,
directory: repository.worktree,
@@ -387,7 +385,9 @@ const layer = Layer.effect(
maximumUntrackedFileBytes?: number
}) {
const list = (args: string[]) =>
repositoryOperation("refresh", input.repository, args).pipe(Effect.map((result) => nuls(result.text)))
repositoryOperation("refresh", input.repository, args).pipe(
Effect.map((result) => result.text.split("\0").filter(Boolean)),
)
const [tracked, untracked] = yield* Effect.all(
[
list(["diff-files", "--name-only", "-z", "--", input.scope]),
@@ -464,7 +464,13 @@ const layer = Layer.effect(
directory: input.repository.worktree,
message: result.stderr.toString("utf8").trim() || "Failed to check ignored paths",
})
return new Set(nuls(result.stdout.toString("utf8")).map((file) => RelativePath.make(file)))
return new Set(
result.stdout
.toString("utf8")
.split("\0")
.filter(Boolean)
.map((file) => RelativePath.make(file)),
)
})
const writeTree = Effect.fn("Git.tree.write")(function* (repository: Repository) {
@@ -493,23 +499,19 @@ const layer = Layer.effect(
to: TreeID
}) {
// Undo needs both paths of a rename, not only its destination.
return nuls(
(yield* repositoryOperation("list_files", input.repository, [
"diff",
"--name-only",
"--no-renames",
"-z",
input.from,
input.to,
])).text,
).map((file) => RelativePath.make(file))
return (yield* repositoryOperation("list_files", input.repository, [
"diff",
"--name-only",
"--no-renames",
"-z",
input.from,
input.to,
])).text
.split("\0")
.filter(Boolean)
.map((file) => RelativePath.make(file))
})
/**
* Three batched invocations over the tree pair instead of three per file. An
* explicit empty selection diffs nothing; an absent one diffs every changed path.
* Patch output is capped like VCS diffs: files past the cap get an empty patch.
*/
const treeDiff = Effect.fn("Git.tree.diff")(function* (input: {
repository: Repository
from: TreeID
@@ -517,57 +519,49 @@ const layer = Layer.effect(
context?: number
paths?: readonly RelativePath[]
}) {
if (input.paths?.length === 0) return []
const args = ["--no-renames", input.from, input.to, "--", ...(input.paths ?? [])]
// Patch headers have no -z form: unquoted paths keep chunksByFile matching non-ASCII names.
const [names, numbers, patch] = yield* Effect.all(
[
repositoryOperation("diff", input.repository, ["diff", "--name-status", "-z", ...args]),
repositoryOperation("diff", input.repository, ["diff", "--numstat", "-z", ...args]),
repositoryOperation(
const paths = input.paths ?? (yield* treeFiles(input))
return yield* Effect.forEach(paths, (file) =>
Effect.gen(function* () {
const statusText = (yield* repositoryOperation("diff", input.repository, [
"diff",
input.repository,
["-c", "core.quotepath=false", "diff", "--no-ext-diff", `--unified=${input.context ?? 3}`, ...args],
{ maxOutputBytes: VcsPatch.MAX_TOTAL_PATCH_BYTES },
),
],
{ concurrency: 3 },
)
const statuses = nuls(names.text)
const files = statuses.flatMap((code, index) => {
const file = statuses[index + 1]
if (index % 2 !== 0 || !file) return []
return [
{
file: RelativePath.make(file),
status: code.startsWith("A") ? "added" : code.startsWith("D") ? "deleted" : "modified",
} as const,
]
})
const stats = new Map(
nuls(numbers.text).flatMap((line) => {
const [additions, deletions, ...file] = line.split("\t")
if (!additions || !deletions || file.length === 0) return []
return [
[
file.join("\t"),
additions === "-" || deletions === "-"
? { binary: true, additions: 0, deletions: 0 }
: { binary: false, additions: Number(additions), deletions: Number(deletions) },
] as const,
]
"--name-status",
"--no-renames",
input.from,
input.to,
"--",
file,
])).text.trim()
const status = statusText.startsWith("A") ? "added" : statusText.startsWith("D") ? "deleted" : "modified"
const stats = (yield* repositoryOperation("diff", input.repository, [
"diff",
"--numstat",
"--no-renames",
input.from,
input.to,
"--",
file,
])).text.split("\t")
const binary = stats[0] === "-" || stats[1] === "-"
const patch = binary
? ""
: (yield* repositoryOperation("diff", input.repository, [
"diff",
`--unified=${input.context ?? 3}`,
"--no-renames",
input.from,
input.to,
"--",
file,
])).text
return {
file,
status,
additions: binary ? 0 : Number(stats[0] ?? 0),
deletions: binary ? 0 : Number(stats[1] ?? 0),
patch,
} satisfies File.Diff
}),
)
const patches = VcsPatch.chunksByFile(patch, (index) => files[index]?.file)
return files.map((entry) => {
const stat = stats.get(entry.file)
return {
...entry,
additions: stat?.additions ?? 0,
deletions: stat?.deletions ?? 0,
patch: stat?.binary ? "" : (patches.get(entry.file) ?? VcsPatch.emptyPatch(entry.file)),
} satisfies File.Diff
})
})
const hasEntry = Effect.fnUntraced(function* (repository: Repository, tree: TreeID, file: RelativePath) {
@@ -739,11 +733,6 @@ function execute(cwd: string, proc: AppProcess.Interface, args: string[]) {
)
}
/** Split NUL-terminated git output into its records. */
function nuls(text: string) {
return text.split("\0").filter(Boolean)
}
function resolvePath(cwd: string, value: string) {
const trimmed = value.replace(/[\r\n]+$/, "")
if (!trimmed) return cwd
+1
View File
@@ -101,6 +101,7 @@ export const XAIPlugin = define({
for (const model of provider.models.values()) {
catalog.model.update(providerID, model.id, (draft) => {
draft.capabilities.responsesWebsockets = true
draft.websocket = true
})
}
})
-24
View File
@@ -54,11 +54,8 @@ import { SessionModelTransport } from "./session/model-transport.js"
import { llmClient } from "./effect/app-node-platform.js"
import { Snapshot } from "./snapshot.js"
import { Session } from "./session/session.js"
import { SessionDiff, TurnRangeError } from "./session/diff.js"
import { LocationServiceMap } from "./location-service-map.js"
import { FSUtil } from "@opencode/util/fs-util"
import type { EventLog } from "@opencode/schema/event-log"
import type { FileDiff } from "@opencode/schema/file-diff"
import { Job } from "./job.js"
import type { Command } from "./command.js"
import { SessionEnvironment } from "./session/environment.js"
@@ -110,7 +107,6 @@ export {
type InboxItemRef = { readonly sessionID: SessionSchema.ID; readonly inboxID: SessionMessage.ID }
export { DestinationNotFoundError, DestinationNotDirectoryError, DestinationUnavailableError }
export { TurnRangeError }
export interface Interface {
readonly list: (input?: ListInput) => Effect.Effect<{
@@ -137,13 +133,6 @@ export interface Interface {
readonly context: (
sessionID: SessionSchema.ID,
) => Effect.Effect<SessionMessage.Info[], NotFoundError | MessageDecodeError>
/** Structured diffs of the files changed by a turn or range of turns; see `SessionDiff.turn`. */
readonly diff: (input: {
readonly sessionID: SessionSchema.ID
readonly messageID?: SessionMessage.ID
readonly to?: SessionMessage.ID
readonly context?: number
}) => Effect.Effect<readonly FileDiff.Info[], NotFoundError | MessageNotFoundError | TurnRangeError | Snapshot.Error>
/**
* Durable admitted session work not yet visible in projected history,
* ordered by admission. Includes unpromoted user and synthetic inputs and
@@ -232,7 +221,6 @@ const layer = Layer.effect(
const moves = yield* SessionMove.Service
const jobs = yield* Job.Service
const environments = yield* SessionEnvironment.Service
const locations = yield* LocationServiceMap.Service
const sessions = yield* Session.make()
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
@@ -364,17 +352,6 @@ const layer = Layer.effect(
yield* result.get(sessionID)
return yield* store.context(sessionID)
}),
diff: Effect.fn("Session.diff")(function* (input) {
const session = yield* result.get(input.sessionID)
const active = yield* execution.isActive(input.sessionID)
return yield* SessionDiff.turn(db, locations, {
session,
active,
messageID: input.messageID,
to: input.to,
context: input.context,
})
}),
inbox: (sessionID) => sessions.forSession(sessionID).inbox(),
cancelInbox: (input) => sessions.forSession(input.sessionID).cancelInbox(input.inboxID),
steerInbox: (input) => sessions.forSession(input.sessionID).steerInbox(input.inboxID),
@@ -463,7 +440,6 @@ export const node: LayerNode.Provider<Service, never, typeof Node.tags.values.gl
SessionInbox.node,
SessionMove.node,
SessionProjector.node,
LocationServiceMap.node,
FSUtil.node,
App.node,
],
-138
View File
@@ -1,138 +0,0 @@
export * as SessionDiff from "./diff.js"
import { and, asc, eq, gt, inArray, lt, or, sql } from "drizzle-orm"
import { Context, Effect, Schema } from "effect"
import { Location } from "@opencode/schema/location"
import { Database } from "../database/database.js"
import { LocationServiceMap } from "../location-service-map.js"
import { Snapshot } from "../snapshot.js"
import { PATCH_CONTEXT_LINES } from "../vcs/patch.js"
import { MessageNotFoundError } from "./error.js"
import { SessionMessage } from "./message.js"
import { SessionSchema } from "./schema.js"
import { SessionMessageTable } from "./sql.js"
export class TurnRangeError extends Schema.TaggedError<TurnRangeError>()("Session.TurnRangeError", {
sessionID: SessionSchema.ID,
field: Schema.Literals(["messageID", "to"]),
message: Schema.String,
}) {}
const decodeLocation = Schema.decodeUnknownSync(Schema.fromJsonString(Location.Ref))
/**
* Diff the files changed by the turn containing a user message. A turn runs from
* the first prompt after the Session was last idle until the next idle marker, so
* prompts steered in while it was busy belong to the same turn; `to` extends the
* range through the turn containing a later user message. Compares the range's
* first recorded start snapshot with its last recorded end snapshot; only a step
* still running in the active Session compares against the working copy. Like VCS
* diffs, an omitted `context` yields full-file patches.
*
* A Session without any idle marker predates them, so its prompts span until the
* next user message instead.
*
* Snapshot trees live in the repository of the Location that captured them, so a
* range spanning a location switch is rejected rather than diffed wrongly.
*/
export const turn = Effect.fn("SessionDiff.turn")(function* (
db: Database.Interface["db"],
locations: Context.Service.Shape<typeof LocationServiceMap.Service>,
input: {
readonly session: SessionSchema.Info
/** The process is currently executing this Session. */
readonly active: boolean
readonly messageID?: SessionMessage.ID
readonly to?: SessionMessage.ID
readonly context?: number
},
) {
const sessionID = input.session.id
const rows = yield* db
.select({ id: SessionMessageTable.id, type: SessionMessageTable.type, seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
or(
inArray(SessionMessageTable.type, ["user", "idle"]),
input.messageID ? eq(SessionMessageTable.id, input.messageID) : undefined,
input.to ? eq(SessionMessageTable.id, input.to) : undefined,
),
),
)
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
const users = rows.filter((row) => row.type === "user")
const markers = rows.filter((row) => row.type === "idle")
const resolve = Effect.fn(function* (field: "messageID" | "to", id: SessionMessage.ID) {
const row = rows.find((row) => row.id === id)
if (!row) return yield* new MessageNotFoundError({ sessionID, messageID: id })
if (row.type !== "user")
return yield* new TurnRangeError({ sessionID, field, message: `Message ${id} is not a user message` })
return row
})
const anchor = input.messageID ? yield* resolve("messageID", input.messageID) : users[users.length - 1]
if (!anchor) return []
const last = input.to ? yield* resolve("to", input.to) : anchor
if (last.seq < anchor.seq)
return yield* new TurnRangeError({ sessionID, field: "to", message: `Message ${last.id} precedes ${anchor.id}` })
// Without any marker, history predates idle markers and a prompt's turn ends at the next prompt.
const legacy = markers.length === 0
// The turn opens with the first prompt after the previous idle marker; the anchor itself is the latest candidate.
const opened = markers.findLast((row) => row.seq < anchor.seq)?.seq ?? -1
const start = legacy ? anchor.seq : (users.find((row) => row.seq > opened)?.seq ?? anchor.seq)
const end = legacy ? users.find((row) => row.seq > last.seq)?.seq : markers.find((row) => row.seq > last.seq)?.seq
const steps = yield* db
.select({
seq: SessionMessageTable.seq,
start: sql<string | null>`json_extract(${SessionMessageTable.data}, '$.snapshot.start')`,
end: sql<string | null>`json_extract(${SessionMessageTable.data}, '$.snapshot.end')`,
completed: sql<number | null>`json_extract(${SessionMessageTable.data}, '$.time.completed')`,
})
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, sessionID),
eq(SessionMessageTable.type, "assistant"),
gt(SessionMessageTable.seq, start),
end === undefined ? undefined : lt(SessionMessageTable.seq, end),
),
)
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
const first = steps[0]
const final = steps[steps.length - 1]
const from = steps.find((step) => step.start)?.start
if (!first || !final || !from) return []
const switches = yield* db
.select({
seq: SessionMessageTable.seq,
location: sql<string>`json_extract(${SessionMessageTable.data}, '$.location')`,
previous: sql<string | null>`json_extract(${SessionMessageTable.data}, '$.previous.location')`,
})
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "location-switched")))
.orderBy(asc(SessionMessageTable.seq))
.all()
.pipe(Effect.orDie)
if (switches.some((row) => row.seq > first.seq && row.seq < final.seq))
return yield* new TurnRangeError({ sessionID, field: "to", message: "Turn range spans a location change" })
const before = switches.findLast((row) => row.seq < first.seq)?.location
const after = switches.find((row) => row.seq > first.seq)?.previous
const location = before ? decodeLocation(before) : after ? decodeLocation(after) : input.session.location
const recorded = steps.findLast((step) => step.end)?.end
return yield* Effect.gen(function* () {
const snapshot = yield* Snapshot.Service
const running = input.active && final.completed === null
const to = running ? ((yield* snapshot.capture()) ?? recorded) : recorded
if (!to) return []
return yield* snapshot.diff({
from: Snapshot.ID.make(from),
to: Snapshot.ID.make(to),
context: input.context ?? PATCH_CONTEXT_LINES,
})
}).pipe(Effect.provide(locations.get(location)))
})
+3 -20
View File
@@ -60,21 +60,6 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
)
})
const idle = (outcome: SessionMessage.Idle["outcome"]) =>
clearCurrentRetry.pipe(
Effect.andThen(
adapter.appendMessage(
SessionMessage.Idle.make({
id: SessionMessage.ID.fromEvent(event.id),
type: "idle",
outcome,
metadata: event.metadata,
time: { created },
}),
),
),
)
const project = pipe(
Match.type<SessionEvent.DurableEvent>(),
Match.discriminatorsExhaustive("type")({
@@ -138,11 +123,9 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
"session.inbox.cancelled": () => Effect.void,
"session.inbox.delivery.changed": () => Effect.void,
"session.execution.started": () => Effect.void,
"session.execution.succeeded": () => idle("succeeded"),
"session.execution.failed": () => idle("failed"),
// Shutdown keeps the execution claim and the resumed drain continues the turn.
"session.execution.interrupted": (event) =>
event.data.reason === "shutdown" ? clearCurrentRetry : idle("interrupted"),
"session.execution.succeeded": () => clearCurrentRetry,
"session.execution.failed": () => clearCurrentRetry,
"session.execution.interrupted": () => clearCurrentRetry,
"session.instructions.updated": (event) => {
if (event.data.text === undefined) return Effect.void
return adapter.appendMessage(
@@ -226,7 +226,6 @@ function toLLMMessage(message: SessionMessage.Info, model: Model.Ref, providerMe
switch (message.type) {
case "agent-switched":
case "model-switched":
case "idle":
return []
case "location-switched":
return [
+16 -33
View File
@@ -131,55 +131,38 @@ const layer = Layer.effect(
)
})
const comparison = Effect.fnUntraced(function* (operation: "files" | "diff", input: CompareInput) {
const compare = Effect.fnUntraced(function* (operation: "files" | "diff", input: CompareInput) {
const repo = yield* repository.pipe(Effect.mapError((cause) => failure(operation, cause)))
return {
source: repo.source,
const comparison = {
repository: repo.snapshotRepository,
from: Git.TreeID.make(input.from),
to: Git.TreeID.make(input.to),
}
})
// Snapshots track every scoped file; the source repository's ignore rules decide what callers see.
const ignored = Effect.fnUntraced(function* (
operation: "files" | "diff",
source: Git.Repository,
paths: readonly RelativePath[],
) {
return yield* git.index
.ignored({ repository: source, paths })
const files = yield* git.tree.files(comparison).pipe(Effect.mapError((cause) => failure(operation, cause)))
const ignored = yield* git.index
.ignored({ repository: repo.source, paths: files })
.pipe(Effect.mapError((cause) => failure(operation, cause)))
return {
input: comparison,
files,
ignored,
}
})
const files = Effect.fn("Snapshot.files")(function* (input: CompareInput) {
const compared = yield* comparison("files", input)
const changed = yield* git.tree
.files({ repository: compared.repository, from: compared.from, to: compared.to })
.pipe(Effect.mapError((cause) => failure("files", cause)))
const skipped = yield* ignored("files", compared.source, changed)
return changed.filter((file) => !skipped.has(file))
const comparison = yield* compare("files", input)
return comparison.files.filter((file) => !comparison.ignored.has(file))
})
const diff = Effect.fn("Snapshot.diff")(function* (input: DiffInput) {
if (input.paths?.length === 0) return []
const compared = yield* comparison("diff", input)
// Only an explicit selection becomes a pathspec; ignored paths are dropped from the result instead.
const diffs = yield* git.tree
const comparison = yield* compare("diff", input)
return yield* git.tree
.diff({
repository: compared.repository,
from: compared.from,
to: compared.to,
...comparison.input,
context: input.context,
paths: input.paths,
paths: (input.paths ?? comparison.files).filter((file) => !comparison.ignored.has(file)),
})
.pipe(Effect.mapError((cause) => failure("diff", cause)))
const skipped = yield* ignored(
"diff",
compared.source,
diffs.map((file) => RelativePath.make(file.file)),
)
return diffs.filter((file) => !skipped.has(RelativePath.make(file.file)))
})
const plan = Effect.fnUntraced(function* (worktree: AbsolutePath, input: RestoreInput) {
+56 -1
View File
@@ -1,5 +1,6 @@
import { expect } from "bun:test"
import { LanguageModel } from "@opencode/ai"
import { LanguageModel, LLMClient } from "@opencode/ai"
import { RequestExecutor } from "@opencode/ai/route"
import { OpenAIChat } from "@opencode/ai/protocols"
import { TestLLM } from "@opencode/ai/testing"
import { AISDK } from "@opencode/core/aisdk"
@@ -11,6 +12,7 @@ import { ID, Info, Ref } from "@opencode/core/model"
import { Provider } from "@opencode/core/provider"
import { Npm } from "@opencode/util/npm"
import { Effect, Layer } from "effect"
import { HttpClient, HttpClientResponse } from "effect/unstable/http"
import { testEffect } from "./lib/effect"
const selected = Info.make({
@@ -98,3 +100,56 @@ resolverIt.effect("resolves dynamic models with their catalog metadata", () =>
})
}),
)
testEffect(Layer.empty).effect("attributes each stateless completion without creating a stored session", () =>
Effect.gen(function* () {
const sessions: string[] = []
const http = Layer.succeed(
HttpClient.HttpClient,
HttpClient.make((request) =>
Effect.sync(() => {
const session = request.headers["x-opencode-session"]
if (!session)
return HttpClientResponse.fromWeb(
request,
Response.json(
{
error: { type: "MissingSessionID", message: "Session ID is required" },
},
{ status: 400 },
),
)
sessions.push(session)
return HttpClientResponse.fromWeb(
request,
new Response(
`data: ${JSON.stringify({
id: "completion",
object: "chat.completion.chunk",
created: 1,
model: "gemini",
choices: [{ index: 0, delta: { content: "OK" }, finish_reason: "stop" }],
})}\n\ndata: [DONE]\n\n`,
{ headers: { "content-type": "text/event-stream" } },
),
)
}),
),
)
const native = LLMClient.layer.pipe(Layer.provide(RequestExecutor.layer.pipe(Layer.provide(http))))
yield* Effect.gen(function* () {
const generate = yield* Generate.Service
for (let index = 0; index < 2; index++) {
expect(
yield* generate.text({
prompt: "Return exactly OK",
model: Ref.make({ providerID: selected.providerID, id: selected.id }),
}),
).toBe("OK")
}
}).pipe(Effect.provide(Generate.layer.pipe(Layer.provide(Layer.merge(resolver, native)))))
expect(sessions).toHaveLength(2)
expect(sessions[0]).toStartWith("ses_")
expect(sessions[1]).not.toBe(sessions[0])
}),
)
-37
View File
@@ -6,7 +6,6 @@ import { Effect } from "effect"
import { LayerNode } from "@opencode/util/effect/layer-node"
import { Git } from "@opencode/core/git"
import { AbsolutePath, RelativePath } from "@opencode/core/schema"
import { VcsPatch } from "@opencode/core/vcs/patch"
import { branch, commit, initRepo, read, withRemote } from "./fixture/git"
import { tmpdir } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
@@ -197,42 +196,6 @@ describe("Git trees", () => {
}),
)
it.live("caps batched tree patches, keeps per-file stats past the cap, and matches non-ASCII names", () =>
Effect.gen(function* () {
const root = yield* Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
)
yield* Effect.promise(() => initRepo(root.path))
const git = yield* Git.Service
const repository = yield* git.repo.discover(AbsolutePath.make(root.path))
if (!repository) throw new Error("Repository not found")
const before = yield* git.tree.capture({ repository, scopes: [RelativePath.make(".")] })
const lines = Math.ceil(VcsPatch.MAX_TOTAL_PATCH_BYTES / 80) + 1
yield* Effect.promise(async () => {
await Bun.write(path.join(root.path, "a-small.txt"), "small\n")
await Bun.write(path.join(root.path, "b-large.txt"), `${"x".repeat(79)}\n`.repeat(lines))
await Bun.write(path.join(root.path, "c-binary.bin"), new Uint8Array([0, 1, 2, 3]))
await Bun.write(path.join(root.path, "a-caf\u00e9.txt"), "caf\u00e9\n")
})
const after = yield* git.tree.capture({ repository, scopes: [RelativePath.make(".")] })
const diffs = yield* git.tree.diff({ repository, from: before, to: after, context: 0 })
expect(diffs.map((item) => [item.file, item.status, item.additions, item.deletions])).toEqual([
["a-caf\u00e9.txt", "added", 1, 0],
["a-small.txt", "added", 1, 0],
["b-large.txt", "added", lines, 0],
["c-binary.bin", "added", 0, 0],
])
// Patch headers are not NUL-delimited; a quoted (octal-escaped) header would orphan this chunk.
expect(diffs[0]?.patch).toContain("+caf\u00e9\n")
expect(diffs[1]?.patch).toContain("+small\n")
expect(diffs[2]?.patch).toBe(VcsPatch.emptyPatch("b-large.txt"))
expect(diffs[3]?.patch).toBe("")
expect(yield* git.tree.diff({ repository, from: before, to: after, paths: [] })).toEqual([])
}),
)
it.live("captures, compares, previews, and restores scoped trees", () =>
Effect.gen(function* () {
const root = yield* Effect.acquireRelease(
@@ -69,7 +69,7 @@ describe("XAIPlugin", () => {
}),
)
it.effect("keeps xAI Responses WebSockets opt-in", () =>
it.effect("enables xAI Responses WebSockets", () =>
Effect.gen(function* () {
const catalog = yield* Catalog.Service
const providerID = Provider.ID.make("xai")
@@ -84,7 +84,7 @@ describe("XAIPlugin", () => {
const model = yield* catalog.model.get(providerID, Model.ID.make("grok-4.6"))
expect(model?.capabilities.responsesWebsockets).toBe(true)
expect(model?.websocket).toBeUndefined()
expect(model?.websocket).toBe(true)
}),
)
})
-198
View File
@@ -1,198 +0,0 @@
import { $ } from "bun"
import { describe, expect } from "bun:test"
import fs from "fs/promises"
import path from "path"
import { Effect } from "effect"
import { Agent } from "@opencode/core/agent"
import { Bus } from "@opencode/core/bus"
import { Database } from "@opencode/core/database/database"
import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
import { LocationServiceMap } from "@opencode/core/location-service-map"
import { Model } from "@opencode/core/model"
import { Plugin } from "@opencode/core/plugin"
import { Provider } from "@opencode/core/provider"
import { AbsolutePath } from "@opencode/core/schema"
import { Session } from "@opencode/core/session"
import { SessionDiff } from "@opencode/core/session/diff"
import { SessionEvent } from "@opencode/core/session/event"
import { SessionExecution } from "@opencode/core/session/execution"
import { SessionInbox } from "@opencode/core/session/inbox"
import { SessionMessage } from "@opencode/core/session/message"
import { SessionProjector } from "@opencode/core/session/projector"
import { Snapshot } from "@opencode/core/snapshot"
import { Money } from "@opencode/schema/money"
import { LayerNode } from "@opencode/util/effect/layer-node"
import { Global } from "@opencode/util/global"
import { tempGlobalLayer } from "./fixture/global"
import { offlineModels } from "./fixture/models"
import { tmpdirScoped } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, Session.node, LocationServiceMap.node]),
[Global.node.replace(tempGlobalLayer), SessionExecution.node.replace(SessionExecution.noopLayer), offlineModels],
),
)
const summarize = (file: { file: string; status: string; additions: number; deletions: number }) => [
file.file,
file.status,
file.additions,
file.deletions,
]
describe("Session.diff", () => {
it.live(
"diffs the busy period containing a user message and ranges across later turns",
() =>
Effect.gen(function* () {
const tmp = yield* tmpdirScoped()
const directory = path.join(tmp.path, "project")
const write = (name: string, content: string) => () => Bun.write(path.join(directory, name), content)
yield* Effect.promise(async () => {
await fs.mkdir(directory)
await write("first.txt", "first\n")()
await write("second.txt", "second\n")()
await write("manual.txt", "manual\n")()
await $`git init -q`.cwd(directory).quiet()
await $`git -c core.fsmonitor=false add .`.cwd(directory).quiet()
})
const sessions = yield* Session.Service
const database = yield* Database.Service
const bus = yield* Bus.Service
const locations = yield* LocationServiceMap.Service
const created = yield* sessions.create({ location: { directory: AbsolutePath.make(directory) } })
const diff = (input?: { messageID?: SessionMessage.ID; to?: SessionMessage.ID }) =>
sessions
.diff({ sessionID: created.id, context: 0, ...input })
.pipe(Effect.map((files) => files.map(summarize)))
expect(yield* diff()).toEqual([])
yield* Effect.gen(function* () {
const plugins = yield* Plugin.Service
yield* plugins.awaitActivation
const snapshot = yield* Snapshot.Service
const usage = {
cost: Money.USD.zero,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
}
const prompt = Effect.fn(function* (text: string) {
const admitted = yield* sessions.prompt({ sessionID: created.id, text, resume: false })
yield* SessionInbox.promote(database.db, bus, created.id, "steer")
return admitted.id
})
const step = Effect.fn(function* (edit: () => Promise<unknown>, end: "recorded" | "unrecorded" | "running") {
const before = yield* snapshot.capture()
if (!before) throw new Error("Start snapshot missing")
const assistantMessageID = SessionMessage.ID.create()
yield* bus.publish(SessionEvent.Step.Started, {
sessionID: created.id,
assistantMessageID,
agent: Agent.defaultID,
model: { id: Model.ID.make("test-model"), providerID: Provider.ID.make("test-provider") },
snapshot: before,
})
yield* Effect.promise(edit)
if (end === "running") return assistantMessageID
const after = end === "recorded" ? yield* snapshot.capture() : undefined
yield* bus.publish(SessionEvent.Step.Ended, {
sessionID: created.id,
assistantMessageID,
finish: "stop",
...usage,
snapshot: after,
files: after && before ? yield* snapshot.files({ from: before, to: after }) : undefined,
})
return assistantMessageID
})
const idle = (outcome: "succeeded" | "failed") =>
outcome === "succeeded"
? bus.publish(SessionEvent.Execution.Succeeded, { sessionID: created.id })
: bus.publish(SessionEvent.Execution.Failed, {
sessionID: created.id,
error: { type: "unknown", message: "failed" },
})
// Before any idle marker exists, a prompt's turn ends at the next prompt.
const first = yield* prompt("Edit the first file")
const firstStep = yield* step(write("first.txt", "first edited\n"), "recorded")
// Edits made while idle are not a turn's work, but a range spanning them still sees them.
yield* Effect.promise(write("manual.txt", "manual edited\n"))
const second = yield* prompt("Edit the second file")
yield* step(write("second.txt", "second edited\n"), "recorded")
expect(yield* diff()).toEqual([["second.txt", "modified", 1, 1]])
expect(yield* diff({ messageID: first })).toEqual([["first.txt", "modified", 1, 1]])
// Once markers exist, a turn spans a whole busy period, steers included; earlier history merges into the first one.
yield* idle("succeeded")
const third = yield* prompt("Add a third file")
yield* step(write("third.txt", "third\n"), "recorded")
const steer = yield* prompt("Also add a fourth file")
yield* step(write("fourth.txt", "fourth\n"), "recorded")
yield* idle("failed")
const busy = [
["fourth.txt", "added", 1, 0],
["third.txt", "added", 1, 0],
]
expect(yield* diff()).toEqual(busy)
expect(yield* diff({ messageID: steer })).toEqual(busy)
expect(yield* diff({ messageID: second })).toEqual([
["first.txt", "modified", 1, 1],
["manual.txt", "modified", 1, 1],
["second.txt", "modified", 1, 1],
])
expect(yield* diff({ messageID: first, to: third })).toEqual([
["first.txt", "modified", 1, 1],
["fourth.txt", "added", 1, 0],
["manual.txt", "modified", 1, 1],
["second.txt", "modified", 1, 1],
["third.txt", "added", 1, 0],
])
const full = yield* sessions.diff({ sessionID: created.id, messageID: first })
expect(full[0]?.patch).toContain("-first\n+first edited\n")
expect(yield* diff({ messageID: steer, to: second }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.TurnRangeError",
field: "to",
})
expect(yield* diff({ messageID: firstStep }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.TurnRangeError",
field: "messageID",
})
expect(yield* diff({ messageID: SessionMessage.ID.create() }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.MessageNotFoundError",
})
// A completed step without an end snapshot falls back to the last recorded end.
yield* prompt("Edit both files again")
yield* step(write("first.txt", "first edited twice\n"), "recorded")
yield* step(write("second.txt", "second edited twice\n"), "unrecorded")
yield* idle("succeeded")
expect(yield* diff()).toEqual([["first.txt", "modified", 1, 1]])
// Only a step still running in the active session compares against the working copy.
yield* prompt("Delete the manual file")
yield* step(() => fs.rm(path.join(directory, "manual.txt")), "running")
expect(yield* diff()).toEqual([])
const session = yield* sessions.get(created.id)
const live = yield* SessionDiff.turn(database.db, locations, { session, active: true, context: 0 })
expect(live.map(summarize)).toEqual([["manual.txt", "deleted", 0, 1]])
// Reverting removes later history, markers included; a fork keeps the copied turns.
yield* sessions.revert.stage({ sessionID: created.id, messageID: steer, files: false })
yield* sessions.revert.commit(created.id)
expect(yield* diff()).toEqual([["third.txt", "added", 1, 0]])
expect(yield* diff({ messageID: steer }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.MessageNotFoundError",
})
const forked = yield* sessions.fork({ sessionID: created.id, boundary: { type: "through" } })
expect((yield* sessions.diff({ sessionID: forked.id, context: 0 })).map(summarize)).toEqual([
["third.txt", "added", 1, 0],
])
}).pipe(Effect.provide(LocationServiceMap.Service.get(created.location)))
}),
// Real Location/plugin startup and Git snapshots can exceed five seconds under CI load.
{ timeout: 30_000 },
)
})
+2 -3
View File
@@ -561,9 +561,7 @@ describe("SessionRestart background recovery", () => {
expect(yield* restarted.pendingBackground).toEqual([])
expect(yield* SessionInbox.list(database.db, sessionID)).toHaveLength(delivered ? 0 : 1)
yield* SessionInbox.promote(database.db, bus, sessionID, "steer")
// Recovery ends a busy period, so an idle marker follows the notification.
const messages = (yield* sessions.messages({ sessionID })).filter((message) => message.type !== "idle")
expect(messages).toMatchObject([
expect(yield* sessions.messages({ sessionID })).toMatchObject([
{
id: background.notificationID,
type: "synthetic",
@@ -571,6 +569,7 @@ describe("SessionRestart background recovery", () => {
metadata: { state: "completed" },
},
])
expect(yield* sessions.messages({ sessionID })).toHaveLength(1)
}),
)
}
+10 -1
View File
@@ -1944,7 +1944,16 @@ describe("SessionRunnerLLM", () => {
expect(yield* entries.list(sessionID)).toEqual([{ key: "nullable", value: null }])
})
scenario("rejects API instruction entries larger than 8KB", function* () {
scenario("accepts API instruction entries up to 256 KiB", function* () {
const entries = yield* InstructionEntry.Service
const value = "x".repeat(InstructionEntry.MaxValueBytes - 2)
yield* entries.put({ sessionID, key: "large", value })
expect(yield* entries.list(sessionID)).toEqual([{ key: "large", value }])
})
scenario("rejects API instruction entries larger than 256 KiB", function* () {
const entries = yield* InstructionEntry.Service
const exit = yield* entries
-181
View File
@@ -3242,152 +3242,6 @@
"summary": "Get session context"
}
},
"/api/session/{sessionID}/diff": {
"get": {
"tags": ["session"],
"operationId": "v2.session.diff",
"parameters": [
{
"name": "sessionID",
"in": "path",
"schema": {
"type": "string",
"pattern": "^ses"
},
"required": true
},
{
"name": "messageID",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string",
"pattern": "^msg_"
},
{
"type": "null"
}
],
"description": "User message whose turn to diff. Defaults to the turn of the newest user message."
},
"required": false
},
{
"name": "to",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string",
"pattern": "^msg_"
},
{
"type": "null"
}
],
"description": "Later user message whose turn ends the range. Defaults to the turn of `messageID` alone."
},
"required": false
},
{
"name": "context",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"description": "Unchanged lines around each hunk. Omit for full-file patches."
},
"required": false
}
],
"security": [],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"data": {
"type": "array",
"items": {
"$ref": "#/components/schemas/FileDiff.Info"
}
}
},
"required": ["data"],
"additionalProperties": false
}
}
}
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
},
{
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
]
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
},
"404": {
"description": "MessageNotFoundError | SessionNotFoundError",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/MessageNotFoundErrorEncoded"
},
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
]
}
}
}
},
"500": {
"description": "UnknownError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnknownErrorEncoded"
}
}
}
}
},
"description": "Structured per-file diffs of the files a turn changed. A turn runs from the first prompt after the session was last idle until its next idle marker, so prompts steered in while it was busy belong to the same turn; `to` extends the range through a later turn. Compares the range's first recorded snapshot with its last; a step still running in the active session compares against the working copy. Ranges that span a location change are rejected. In sessions without any idle marker, a prompt's turn spans until the next user message.",
"summary": "Diff session turns"
}
},
"/api/session/{sessionID}/inbox": {
"get": {
"tags": ["session"],
@@ -18540,38 +18394,6 @@
"required": ["type", "id", "time", "status", "reason", "summary", "recent"],
"additionalProperties": false
},
"Session.Message.Idle": {
"type": "object",
"properties": {
"id": {
"type": "string",
"pattern": "^msg_"
},
"metadata": {
"type": "object"
},
"time": {
"type": "object",
"properties": {
"created": {
"type": "number"
}
},
"required": ["created"],
"additionalProperties": false
},
"type": {
"type": "string",
"enum": ["idle"]
},
"outcome": {
"type": "string",
"enum": ["succeeded", "failed", "interrupted"]
}
},
"required": ["id", "time", "type", "outcome"],
"additionalProperties": false
},
"Session.Message.Info": {
"anyOf": [
{
@@ -18603,9 +18425,6 @@
},
{
"$ref": "#/components/schemas/Session.Message.Compaction"
},
{
"$ref": "#/components/schemas/Session.Message.Idle"
}
]
},
-26
View File
@@ -30,7 +30,6 @@ import { Model } from "@opencode/schema/model"
import { Location } from "@opencode/schema/location"
import { SessionEvent } from "@opencode/schema/session-event"
import { EventLog } from "@opencode/schema/event-log"
import { FileDiff } from "@opencode/schema/file-diff"
const ParentIDFilter = Schema.Union([
Session.ID,
@@ -522,31 +521,6 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
}),
),
)
.add(
HttpApiEndpoint.get("session.diff", "/api/session/:sessionID/diff", {
params: { sessionID: Session.ID },
query: Schema.Struct({
messageID: Schema.optional(SessionMessage.ID).annotate({
description: "User message whose turn to diff. Defaults to the turn of the newest user message.",
}),
to: Schema.optional(SessionMessage.ID).annotate({
description: "Later user message whose turn ends the range. Defaults to the turn of `messageID` alone.",
}),
context: Schema.NumberFromString.pipe(Schema.decodeTo(NonNegativeInt), Schema.optional).annotate({
description: "Unchanged lines around each hunk. Omit for full-file patches.",
}),
}),
success: Schema.Struct({ data: Schema.Array(FileDiff.Info) }),
error: [InvalidRequestError, MessageNotFoundError, SessionNotFoundError, UnknownError],
}).annotateMerge(
OpenApi.annotations({
identifier: "v2.session.diff",
summary: "Diff session turns",
description:
"Structured per-file diffs of the files a turn changed. A turn runs from the first prompt after the session was last idle until its next idle marker, so prompts steered in while it was busy belong to the same turn; `to` extends the range through a later turn. Compares the range's first recorded snapshot with its last; a step still running in the active session compares against the working copy. Ranges that span a location change are rejected. In sessions without any idle marker, a prompt's turn spans until the next user message.",
}),
),
)
.add(
HttpApiEndpoint.get("session.inbox.list", "/api/session/:sessionID/inbox", {
params: { sessionID: Session.ID },
+1 -1
View File
@@ -27,7 +27,7 @@ export const Snapshot = Schema.Array(
).annotate({ identifier: "InstructionEntry.Snapshot" })
export type Snapshot = typeof Snapshot.Type
export const MaxValueBytes = 8 * 1024
export const MaxValueBytes = 256 * 1024
export class ValueTooLargeError extends Schema.TaggedError<ValueTooLargeError>()(
"InstructionEntryValueTooLargeError",
-14
View File
@@ -280,18 +280,6 @@ export const Compaction = Schema.Union([CompactionRunning, CompactionCompleted,
)
export type Compaction = CompactionRunning | CompactionCompleted | CompactionFailed
/**
* Marks the Session going idle: every step since the previous marker belongs to
* one turn, including prompts steered in while it was busy. A shutdown does not
* record one, since the resumed execution continues the same turn.
*/
export interface Idle extends Schema.Schema.Type<typeof Idle> {}
export const Idle = Schema.Struct({
...Base,
type: Schema.tag("idle"),
outcome: Schema.Literals(["succeeded", "failed", "interrupted"]),
}).annotate({ identifier: "Session.Message.Idle" })
export const Info = Schema.Union([
AgentSelected,
ModelSelected,
@@ -303,7 +291,6 @@ export const Info = Schema.Union([
Shell,
Assistant,
Compaction,
Idle,
]).annotate({ identifier: "Session.Message.Info" })
export type Info =
| AgentSelected
@@ -316,5 +303,4 @@ export type Info =
| Shell
| Assistant
| Compaction
| Idle
export type Type = Info["type"]
+1 -23
View File
@@ -1,6 +1,5 @@
import { Session } from "@opencode/core/session"
import type { Snapshot } from "@opencode/core/snapshot"
import { MessageNotFoundError, SessionNotFoundError, UnknownError } from "@opencode/protocol/errors"
import { SessionNotFoundError, UnknownError } from "@opencode/protocol/errors"
import { Effect } from "effect"
export function missingSession(error: Session.NotFoundError) {
@@ -10,14 +9,6 @@ export function missingSession(error: Session.NotFoundError) {
})
}
export function missingMessage(error: Session.MessageNotFoundError) {
return new MessageNotFoundError({
sessionID: error.sessionID,
messageID: error.messageID,
message: `Message not found: ${error.messageID}`,
})
}
export function failedMessageDecode(error: Session.MessageDecodeError) {
const ref = `err_${crypto.randomUUID().slice(0, 8)}`
return Effect.logError("failed to decode session message").pipe(
@@ -27,16 +18,3 @@ export function failedMessageDecode(error: Session.MessageDecodeError) {
),
)
}
/** Snapshot repositories are host state clients cannot repair, so surface only a log reference. */
export function failedSnapshot(operation: string, sessionID: Session.ID) {
return (error: Snapshot.Error) => {
const ref = `err_${crypto.randomUUID().slice(0, 8)}`
return Effect.logError(`failed to ${operation}`, { cause: error }).pipe(
Effect.annotateLogs({ ref, sessionID }),
Effect.andThen(
Effect.fail(new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref })),
),
)
}
}
+53 -32
View File
@@ -17,9 +17,10 @@ import {
ServiceUnavailableError,
SessionBusyError,
SkillNotFoundError,
UnknownError,
} from "@opencode/protocol/errors"
import { AbsolutePath } from "@opencode/core/schema"
import { failedMessageDecode, failedSnapshot, missingMessage, missingSession } from "./session-error"
import { failedMessageDecode, missingSession } from "./session-error"
const DefaultSessionsLimit = 50
@@ -211,7 +212,15 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
return {
data: yield* session.fork({ sessionID: ctx.params.sessionID, boundary: ctx.payload.boundary }).pipe(
Effect.catchTag("Session.NotFoundError", missingSession),
Effect.catchTag("Session.MessageNotFoundError", missingMessage),
Effect.catchTag(
"Session.MessageNotFoundError",
(error) =>
new MessageNotFoundError({
sessionID: error.sessionID,
messageID: error.messageID,
message: `Message not found: ${error.messageID}`,
}),
),
Effect.catchTag(
"Session.ForkEmptyError",
(error) => new InvalidRequestError({ message: error.message, kind: "empty_session" }),
@@ -439,14 +448,32 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
files: ctx.payload.files,
})
return {
data: yield* session.revert
.stage({ ...ctx.params, ...ctx.payload })
.pipe(
Effect.catchTag("Session.NotFoundError", missingSession),
Effect.catchTag("Session.MessageNotFoundError", missingMessage),
Effect.catchTag("Session.BusyError", busySession),
Effect.catchTag("Snapshot.Error", failedSnapshot("stage session revert", ctx.params.sessionID)),
data: yield* session.revert.stage({ ...ctx.params, ...ctx.payload }).pipe(
Effect.catchTag("Session.NotFoundError", missingSession),
Effect.catchTag(
"Session.MessageNotFoundError",
(error) =>
new MessageNotFoundError({
sessionID: error.sessionID,
messageID: error.messageID,
message: `Message not found: ${error.messageID}`,
}),
),
Effect.catchTag("Session.BusyError", busySession),
Effect.catchTag("Snapshot.Error", (error) => {
const ref = `err_${crypto.randomUUID().slice(0, 8)}`
return Effect.logError("failed to stage session revert", { cause: error }).pipe(
Effect.andThen(
Effect.fail(
new UnknownError({
message: "Unexpected server error. Check server logs for details.",
ref,
}),
),
),
)
}),
),
}
}),
)
@@ -454,13 +481,23 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
"session.revert.clear",
Effect.fn(function* (ctx) {
yield* Effect.log("session.revert.clear", { sessionID: ctx.params.sessionID })
yield* session.revert
.clear(ctx.params.sessionID)
.pipe(
Effect.catchTag("Session.NotFoundError", missingSession),
Effect.catchTag("Session.BusyError", busySession),
Effect.catchTag("Snapshot.Error", failedSnapshot("clear session revert", ctx.params.sessionID)),
)
yield* session.revert.clear(ctx.params.sessionID).pipe(
Effect.catchTag("Session.NotFoundError", missingSession),
Effect.catchTag("Session.BusyError", busySession),
Effect.catchTag("Snapshot.Error", (error) => {
const ref = `err_${crypto.randomUUID().slice(0, 8)}`
return Effect.logError("failed to clear session revert", { cause: error }).pipe(
Effect.andThen(
Effect.fail(
new UnknownError({
message: "Unexpected server error. Check server logs for details.",
ref,
}),
),
),
)
}),
)
return HttpApiSchema.NoContent.make()
}),
)
@@ -490,22 +527,6 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
}
}),
)
.handle(
"session.diff",
Effect.fn(function* (ctx) {
return {
data: yield* session.diff({ sessionID: ctx.params.sessionID, ...ctx.query }).pipe(
Effect.catchTag("Session.NotFoundError", missingSession),
Effect.catchTag("Session.MessageNotFoundError", missingMessage),
Effect.catchTag(
"Session.TurnRangeError",
(error) => new InvalidRequestError({ message: error.message, field: error.field }),
),
Effect.catchTag("Snapshot.Error", failedSnapshot("diff session turn", ctx.params.sessionID)),
),
}
}),
)
.handle(
"session.inbox.list",
Effect.fn(function* (ctx) {
-98
View File
@@ -1,98 +0,0 @@
import { expect, setDefaultTimeout } from "bun:test"
import { Agent } from "@opencode/core/agent"
import { Bus } from "@opencode/core/bus"
import { Model } from "@opencode/core/model"
import { Provider } from "@opencode/core/provider"
import { Session } from "@opencode/core/session"
import { SessionEvent } from "@opencode/core/session/event"
import { SessionExecution } from "@opencode/core/session/execution"
import { SessionMessage } from "@opencode/core/session/message"
import { Money } from "@opencode/schema/money"
import { makeGlobalNode } from "@opencode/util/effect/app-node"
import { Effect, Layer } from "effect"
import { tmpdir } from "../../core/test/fixture/tmpdir"
import { it } from "../../core/test/lib/effect"
import { ServerFetch } from "../src/fetch"
setDefaultTimeout(30_000)
it.live("serves turn diffs by user message with range validation", () =>
Effect.gen(function* () {
const tmp = yield* Effect.acquireDisposable(Effect.promise(() => tmpdir("opencode-session-diff-")))
const ids = { user: SessionMessage.ID.create(), assistant: SessionMessage.ID.create() }
// Deliver the prompt and one step the way the runner would, without a model.
const execution = Layer.effect(
SessionExecution.Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
return SessionExecution.Service.of({
active: Effect.succeed(new Set()),
isActive: () => Effect.succeed(false),
resume: () => Effect.void,
wake: (sessionID) =>
Effect.gen(function* () {
yield* bus.publish(SessionEvent.InboxDelivered, { sessionID, inboxID: ids.user })
yield* bus.publish(SessionEvent.Step.Started, {
sessionID,
assistantMessageID: ids.assistant,
agent: Agent.defaultID,
model: { id: Model.ID.make("model"), providerID: Provider.ID.make("provider") },
})
yield* bus.publish(SessionEvent.Step.Ended, {
sessionID,
assistantMessageID: ids.assistant,
finish: "stop",
cost: Money.USD.zero,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
})
}),
interrupt: () => Effect.succeed(false),
awaitIdle: () => Effect.void,
})
}),
)
const handler = yield* ServerFetch.make(
{
app: { version: "test-version" },
database: { path: ":memory:" },
fs: { filewatcher: false },
models: { fetch: false },
},
{
overrides: [
SessionExecution.node.replace(
makeGlobalNode({ service: SessionExecution.Service, layer: execution, deps: [Bus.node] }),
),
],
},
)
const request = (path: string, body?: unknown) =>
Effect.promise(async () => {
const response = await handler(
new Request(`http://opencode.local${path}`, {
method: body === undefined ? "GET" : "POST",
headers: body === undefined ? undefined : { "content-type": "application/json" },
body: body === undefined ? undefined : JSON.stringify(body),
}),
)
return { status: response.status, body: (await response.json()) as Record<string, unknown> }
})
const created = yield* request("/api/session", { location: { directory: tmp.path } })
const sessionID = Session.ID.make((created.body.data as { id: string }).id)
const diff = (query = "") => request(`/api/session/${sessionID}/diff${query}`)
expect(yield* diff()).toEqual({ status: 200, body: { data: [] } })
expect((yield* request(`/api/session/${sessionID}/prompt`, { id: ids.user, text: "prompt" })).status).toBe(200)
// Not a git repository, so steps record no snapshots and the turn has no diff.
expect(yield* diff(`?messageID=${ids.user}&context=3`)).toEqual({ status: 200, body: { data: [] } })
expect(yield* diff(`?messageID=${ids.assistant}`)).toMatchObject({
status: 400,
body: { _tag: "InvalidRequestError", field: "messageID" },
})
expect(yield* diff(`?messageID=${SessionMessage.ID.create()}`)).toMatchObject({
status: 404,
body: { _tag: "MessageNotFoundError" },
})
expect((yield* request(`/api/session/${Session.ID.create()}/diff`)).status).toBe(404)
}),
)
@@ -16,7 +16,7 @@ export { TimelineRow, type PartGroup, type PartRef, type TimelineRowMap }
export type ReasoningMode = "hidden" | "compact" | "full"
type Notice = Exclude<SessionMessageInfo, { type: "user" | "assistant" | "shell" | "idle" }>
type Notice = Exclude<SessionMessageInfo, { type: "user" | "assistant" | "shell" }>
type Entry = { type: "assistant"; message: SessionMessageAssistant } | { type: "notice"; message: Notice }
type Content = SessionMessageAssistant["content"][number]
type GroupRow = Extract<TimelineRow.TimelineRow, { _tag: "AssistantPart" }>
@@ -765,8 +765,7 @@ function record(value: unknown): value is Record<string, unknown> {
}
function isNotice(message: SessionMessageInfo): message is Notice {
if (message.type === "user" || message.type === "assistant" || message.type === "shell" || message.type === "idle")
return false
if (message.type === "user" || message.type === "assistant" || message.type === "shell") return false
if (message.type !== "synthetic") return true
return !!message.description?.trim() || timelineNoticeRequired(message)
}
+2 -1
View File
@@ -51,7 +51,8 @@ export function createClipboardAdapter(clipboard: CoreClipboardService): OwnedCl
throw new Error(`Unexpected clipboard MIME type: ${result.representation.mimeType}`)
},
async write(text) {
const result = await clipboard.writeText(text, {
// OpenTUI rejects NUL before any destination; host clipboard text cannot contain it.
const result = await clipboard.writeText(text.replaceAll("\0", ""), {
destination: "all-available",
selection: "clipboard",
})
-1
View File
@@ -305,7 +305,6 @@ export function reduceSessionRows(messages: SessionMessageInfo[], inputs = new S
...messages.filter(isInput),
].reduce<SessionRow[]>((rows, message) => {
if (message.type !== "assistant") {
if (message.type === "idle") return rows
if (message.type === "synthetic" && !message.description?.trim()) return rows
if (message.type === "compaction" && message.status === "completed" && usage) usage.previousTurnCache = undefined
if (!pending.has(message.id)) completePrevious(rows)
+13
View File
@@ -102,6 +102,19 @@ test("uses all available routes but skips the process host remotely", async () =
expect(writes).toEqual({ host: 0, terminal: 1 })
})
test("removes NUL characters before writing", async () => {
const writes: string[] = []
const clipboard = createClipboardAdapter(
coreClipboard({
onWrite: (text) => writes.push(text),
}),
)
expect(await clipboard.write("before\0after")).toBeUndefined()
expect(await clipboard.write("clean")).toBeUndefined()
expect(writes).toEqual(["beforeafter", "clean"])
})
test("rejects only when no clipboard route accepted the write", async () => {
const writes: [string, ClipboardWriteOptions][] = []
const failure = new Error("native clipboard failed")
-181
View File
@@ -3242,152 +3242,6 @@
"summary": "Get session context"
}
},
"/api/session/{sessionID}/diff": {
"get": {
"tags": ["session"],
"operationId": "v2.session.diff",
"parameters": [
{
"name": "sessionID",
"in": "path",
"schema": {
"type": "string",
"pattern": "^ses"
},
"required": true
},
{
"name": "messageID",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string",
"pattern": "^msg_"
},
{
"type": "null"
}
],
"description": "User message whose turn to diff. Defaults to the turn of the newest user message."
},
"required": false
},
{
"name": "to",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string",
"pattern": "^msg_"
},
{
"type": "null"
}
],
"description": "Later user message whose turn ends the range. Defaults to the turn of `messageID` alone."
},
"required": false
},
{
"name": "context",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"description": "Unchanged lines around each hunk. Omit for full-file patches."
},
"required": false
}
],
"security": [],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"data": {
"type": "array",
"items": {
"$ref": "#/components/schemas/FileDiff.Info"
}
}
},
"required": ["data"],
"additionalProperties": false
}
}
}
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
},
{
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
]
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
},
"404": {
"description": "MessageNotFoundError | SessionNotFoundError",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/MessageNotFoundErrorEncoded"
},
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
]
}
}
}
},
"500": {
"description": "UnknownError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnknownErrorEncoded"
}
}
}
}
},
"description": "Structured per-file diffs of the files a turn changed. A turn runs from the first prompt after the session was last idle until its next idle marker, so prompts steered in while it was busy belong to the same turn; `to` extends the range through a later turn. Compares the range's first recorded snapshot with its last; a step still running in the active session compares against the working copy. Ranges that span a location change are rejected. In sessions without any idle marker, a prompt's turn spans until the next user message.",
"summary": "Diff session turns"
}
},
"/api/session/{sessionID}/inbox": {
"get": {
"tags": ["session"],
@@ -18540,38 +18394,6 @@
"required": ["type", "id", "time", "status", "reason", "summary", "recent"],
"additionalProperties": false
},
"Session.Message.Idle": {
"type": "object",
"properties": {
"id": {
"type": "string",
"pattern": "^msg_"
},
"metadata": {
"type": "object"
},
"time": {
"type": "object",
"properties": {
"created": {
"type": "number"
}
},
"required": ["created"],
"additionalProperties": false
},
"type": {
"type": "string",
"enum": ["idle"]
},
"outcome": {
"type": "string",
"enum": ["succeeded", "failed", "interrupted"]
}
},
"required": ["id", "time", "type", "outcome"],
"additionalProperties": false
},
"Session.Message.Info": {
"anyOf": [
{
@@ -18603,9 +18425,6 @@
},
{
"$ref": "#/components/schemas/Session.Message.Compaction"
},
{
"$ref": "#/components/schemas/Session.Message.Idle"
}
]
},
-181
View File
@@ -3242,152 +3242,6 @@
"summary": "Get session context"
}
},
"/api/session/{sessionID}/diff": {
"get": {
"tags": ["session"],
"operationId": "v2.session.diff",
"parameters": [
{
"name": "sessionID",
"in": "path",
"schema": {
"type": "string",
"pattern": "^ses"
},
"required": true
},
{
"name": "messageID",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string",
"pattern": "^msg_"
},
{
"type": "null"
}
],
"description": "User message whose turn to diff. Defaults to the turn of the newest user message."
},
"required": false
},
{
"name": "to",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string",
"pattern": "^msg_"
},
{
"type": "null"
}
],
"description": "Later user message whose turn ends the range. Defaults to the turn of `messageID` alone."
},
"required": false
},
{
"name": "context",
"in": "query",
"schema": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"description": "Unchanged lines around each hunk. Omit for full-file patches."
},
"required": false
}
],
"security": [],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"data": {
"type": "array",
"items": {
"$ref": "#/components/schemas/FileDiff.Info"
}
}
},
"required": ["data"],
"additionalProperties": false
}
}
}
},
"400": {
"description": "InvalidRequestError",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
},
{
"$ref": "#/components/schemas/InvalidRequestErrorEncoded"
}
]
}
}
}
},
"401": {
"description": "UnauthorizedError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnauthorizedErrorEncoded"
}
}
}
},
"404": {
"description": "MessageNotFoundError | SessionNotFoundError",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/MessageNotFoundErrorEncoded"
},
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
]
}
}
}
},
"500": {
"description": "UnknownError",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/UnknownErrorEncoded"
}
}
}
}
},
"description": "Structured per-file diffs of the files a turn changed. A turn runs from the first prompt after the session was last idle until its next idle marker, so prompts steered in while it was busy belong to the same turn; `to` extends the range through a later turn. Compares the range's first recorded snapshot with its last; a step still running in the active session compares against the working copy. Ranges that span a location change are rejected. In sessions without any idle marker, a prompt's turn spans until the next user message.",
"summary": "Diff session turns"
}
},
"/api/session/{sessionID}/inbox": {
"get": {
"tags": ["session"],
@@ -18540,38 +18394,6 @@
"required": ["type", "id", "time", "status", "reason", "summary", "recent"],
"additionalProperties": false
},
"Session.Message.Idle": {
"type": "object",
"properties": {
"id": {
"type": "string",
"pattern": "^msg_"
},
"metadata": {
"type": "object"
},
"time": {
"type": "object",
"properties": {
"created": {
"type": "number"
}
},
"required": ["created"],
"additionalProperties": false
},
"type": {
"type": "string",
"enum": ["idle"]
},
"outcome": {
"type": "string",
"enum": ["succeeded", "failed", "interrupted"]
}
},
"required": ["id", "time", "type", "outcome"],
"additionalProperties": false
},
"Session.Message.Info": {
"anyOf": [
{
@@ -18603,9 +18425,6 @@
},
{
"$ref": "#/components/schemas/Session.Message.Compaction"
},
{
"$ref": "#/components/schemas/Session.Message.Idle"
}
]
},
+5 -4
View File
@@ -110,10 +110,11 @@ role for other Foundry models. If a request fails because the token belongs to a
## WebSocket transport
OpenAI and supported Azure Responses models keep one WebSocket connection open per session and send each step over it
instead of opening a new HTTP request. While the request prefix is unchanged, consecutive steps only transmit what was
added since the previous response, which cuts upload volume on long sessions. Provider compaction runs over the same
connection.
OpenAI, xAI, and supported Azure Responses models keep one WebSocket connection open per session and send each step
over it instead of opening a new HTTP request. While the request prefix is unchanged, consecutive steps only transmit
what was added since the previous response, which cuts upload volume on long sessions. OpenAI provider compaction runs
over the same connection. xAI continues a chain only from stored responses, so with its default `store: false` each step
is sent in full over the reused connection.
The connection is transparent. When the provider closes the socket, the next step reconnects; when a connection cannot
be opened at all, the session continues over HTTP. Plugins that register `http.request` or `http.response` hooks for a