diff --git a/packages/client/src/contract.ts b/packages/client/src/contract.ts index 413fea9dc338..e758c60e3774 100644 --- a/packages/client/src/contract.ts +++ b/packages/client/src/contract.ts @@ -50,4 +50,4 @@ export const endpointNames = { "question.request.list": "listRequests", } as const -export const omitEndpoints = new Set(["fs.read", "pty.connect", "pty.connectToken"]) +export const omitEndpoints = new Set(["fs.read", "pty.connect", "pty.connectToken", "session.attachment"]) diff --git a/packages/client/src/generated/client.ts b/packages/client/src/generated/client.ts index 27ec3d81ba2c..1a5bffcf56a0 100644 --- a/packages/client/src/generated/client.ts +++ b/packages/client/src/generated/client.ts @@ -374,7 +374,7 @@ export function make(options: ClientOptions) { path: `/api/session/${encodeURIComponent(input.sessionID)}/prompt`, body: { id: input["id"], prompt: input["prompt"], delivery: input["delivery"], resume: input["resume"] }, successStatus: 200, - declaredStatuses: [409, 404, 400, 401], + declaredStatuses: [404, 409, 500, 400, 401], empty: false, }, requestOptions, diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 3b3188c8742a..716edab516a4 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -33,6 +33,15 @@ export type SessionNotFoundError = { export const isSessionNotFoundError = (value: unknown): value is SessionNotFoundError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "SessionNotFoundError" +export type AttachmentNotFoundError = { + readonly _tag: "AttachmentNotFoundError" + readonly sessionID: string + readonly attachmentID?: string | undefined + readonly message: string +} +export const isAttachmentNotFoundError = (value: unknown): value is AttachmentNotFoundError => + typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "AttachmentNotFoundError" + export type ConflictError = { readonly _tag: "ConflictError" readonly message: string @@ -41,6 +50,14 @@ export type ConflictError = { export const isConflictError = (value: unknown): value is ConflictError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "ConflictError" +export type UnknownError = { + readonly _tag: "UnknownError" + readonly message: string + readonly ref?: string | undefined +} +export const isUnknownError = (value: unknown): value is UnknownError => + typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "UnknownError" + export type ServiceUnavailableError = { readonly _tag: "ServiceUnavailableError" readonly message: string @@ -58,14 +75,6 @@ export type MessageNotFoundError = { export const isMessageNotFoundError = (value: unknown): value is MessageNotFoundError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "MessageNotFoundError" -export type UnknownError = { - readonly _tag: "UnknownError" - readonly message: string - readonly ref?: string | undefined -} -export const isUnknownError = (value: unknown): value is UnknownError => - typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "UnknownError" - export type ProviderNotFoundError = { readonly _tag: "ProviderNotFoundError" readonly providerID: string diff --git a/packages/core/src/attachment-store.ts b/packages/core/src/attachment-store.ts new file mode 100644 index 000000000000..6faaea3ba913 --- /dev/null +++ b/packages/core/src/attachment-store.ts @@ -0,0 +1,638 @@ +export * as AttachmentStore from "./attachment-store" + +import { createHash } from "crypto" +import path from "path" +import { Attachment } from "@opencode-ai/schema/attachment" +import { Context, Duration, Effect, FileSystem, Layer, Option, Schedule, Schema, Semaphore, Stream } from "effect" +import { Database } from "./database/database" +import { Node } from "./effect/app-node" +import { KeyedMutex } from "./effect/keyed-mutex" +import { FSUtil } from "./fs-util" +import { Global } from "./global" +import { SessionMessage } from "./session/message" +import { SessionSchema } from "./session/schema" +import { SessionTable } from "./session/sql" +import { NonNegativeInt } from "./schema" + +export const MANAGED_DIRECTORY = "attachments" +export const MAX_FILE_BYTES = Attachment.MAX_FILE_BYTES +export const MAX_SESSION_BYTES = 100 * 1024 * 1024 +export const MAX_GLOBAL_BYTES = 1024 * 1024 * 1024 +export const UNBOUND_RETENTION = Duration.hours(24) + +const MAX_NAME_BYTES = 180 +const metadataName = "metadata.json" +const uploadName = ".upload" +const metadataUploadName = ".metadata" +const internalNames = new Set([metadataName, uploadName, metadataUploadName]) + +const Metadata = Schema.Struct({ + id: Attachment.ID, + sessionID: SessionSchema.ID, + originalName: Schema.String, + storedName: Schema.String, + clientMime: Schema.String, + detectedMime: Schema.String, + size: NonNegativeInt, + sha256: Schema.String, + createdAt: NonNegativeInt, + boundMessageID: SessionMessage.ID.pipe(Schema.optional), + nativeMediaDeliveredAt: NonNegativeInt.pipe(Schema.optional), +}) +type Metadata = typeof Metadata.Type + +export class StorageError extends Schema.TaggedErrorClass()("AttachmentStore.StorageError", { + operation: Schema.Literals(["allocate", "scan", "read", "write", "rename", "remove"]), + cause: Schema.Defect(), +}) {} + +export class QuotaError extends Schema.TaggedErrorClass()("AttachmentStore.QuotaError", { + scope: Schema.Literals(["file", "session", "global"]), + maximumBytes: NonNegativeInt, +}) {} + +export class FilenameError extends Schema.TaggedErrorClass()("AttachmentStore.FilenameError", { + reason: Schema.Literal("nul"), +}) {} + +export class ReferenceError extends Schema.TaggedErrorClass()("AttachmentStore.ReferenceError", { + sessionID: SessionSchema.ID, + attachmentID: Attachment.ID.pipe(Schema.optional), +}) {} + +export type Error = StorageError | QuotaError | FilenameError | ReferenceError +export type UploadError = StorageError | QuotaError | FilenameError +export type Info = Attachment.Info + +export interface Resolved extends Attachment.Info { + readonly path: string + readonly nativeMediaDelivered: boolean +} + +export interface UploadInput { + readonly sessionID: SessionSchema.ID + readonly name: string + readonly contentType: string + readonly content: Stream.Stream +} + +export interface Interface { + readonly upload: (input: UploadInput) => Effect.Effect + readonly resolve: (input: { + readonly sessionID: SessionSchema.ID + readonly attachmentID: Attachment.ID + }) => Effect.Effect + readonly bind: (input: { + readonly sessionID: SessionSchema.ID + readonly attachmentID: Attachment.ID + readonly messageID: SessionMessage.ID + }) => Effect.Effect + readonly markNativeMediaDelivered: (input: { + readonly sessionID: SessionSchema.ID + readonly attachmentID: Attachment.ID + }) => Effect.Effect + readonly remove: (input: { + readonly sessionID: SessionSchema.ID + readonly attachmentID: Attachment.ID + }) => Effect.Effect + readonly cleanup: (sessions?: ReadonlySet) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/v2/AttachmentStore") {} + +export interface Limits { + readonly file: number + readonly session: number + readonly global: number +} + +interface Usage { + readonly sessions: Map + global: number +} + +interface Reservation { + readonly sessionID: SessionSchema.ID + bytes: number +} + +interface StoreState { + usage: Usage | undefined + reserved: number +} + +const defaults: Limits = { + file: MAX_FILE_BYTES, + session: MAX_SESSION_BYTES, + global: MAX_GLOBAL_BYTES, +} + +export const isManagedURI = (uri: string) => /^opencode:/i.test(uri) + +export const attachmentID = (uri: string) => { + const match = /^opencode:\/\/attachment\/(att_[0-9A-Za-z]+)$/.exec(uri) + return match ? Attachment.ID.make(match[1]) : undefined +} + +const controlRanges = [ + [0x00, 0x1f], + [0x7f, 0x9f], + [0x061c, 0x061c], + [0x200e, 0x200f], + [0x202a, 0x202e], + [0x2066, 0x2069], +] satisfies ReadonlyArray + +const safeCharacter = (char: string) => { + const code = char.codePointAt(0) ?? 0 + return !controlRanges.some(([start, end]) => code >= start && code <= end) +} + +const safeBasename = (input: string) => + Array.from(input.normalize("NFC").split(/[\\/]/).at(-1) ?? "") + .filter(safeCharacter) + .join("") + .replace(/[<>:"/\\|?*]/g, "_") + .replace(/[. ]+$/g, "") + +const usableName = (input: string) => (input === "" || input === "." || input === ".." ? "attachment" : input) + +const deviceStem = (name: string, stem: string) => { + const candidate = (name.split(".", 1)[0] ?? "").replace(/[. ]+$/g, "") + const device = /^(con|prn|aux|nul|clock\$|conin\$|conout\$|com[1-9¹²³]|lpt[1-9¹²³])$/i.test(candidate) + return device || internalNames.has(name.toLowerCase()) ? `_${stem}` : stem +} + +const takeBytes = (input: string, maximum: number) => { + const state = { value: "", bytes: 0 } + for (const char of input) { + const bytes = Buffer.byteLength(char) + if (state.bytes + bytes > maximum) break + state.value += char + state.bytes += bytes + } + return state.value +} + +export const sanitizeName = (input: string) => { + if (input.includes("\0")) return Effect.fail(new FilenameError({ reason: "nul" })) + const fallback = usableName(safeBasename(input)) + const dot = fallback.lastIndexOf(".") + const extension = dot > 0 ? takeBytes(fallback.slice(dot), 24) : "" + const stem = dot > 0 ? fallback.slice(0, dot) : fallback + const prefixed = deviceStem(fallback, stem) + const name = `${takeBytes(prefixed, MAX_NAME_BYTES - Buffer.byteLength(extension))}${extension}` + return Effect.succeed(name || "attachment") +} + +const sniff = (prefix: Uint8Array) => { + const text = Buffer.from(prefix) + if (text.subarray(0, 8).equals(Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]))) return "image/png" + if (text[0] === 0xff && text[1] === 0xd8 && text[2] === 0xff) return "image/jpeg" + if (text.subarray(0, 6).toString() === "GIF87a" || text.subarray(0, 6).toString() === "GIF89a") return "image/gif" + if (text.subarray(0, 4).toString() === "RIFF" && text.subarray(8, 12).toString() === "WEBP") return "image/webp" + if (text.subarray(0, 5).toString() === "%PDF-") return "application/pdf" + return "application/octet-stream" +} + +const sessionDirectory = (root: string, sessionID: SessionSchema.ID) => path.join(root, encodeURIComponent(sessionID)) +const attachmentDirectory = (root: string, sessionID: SessionSchema.ID, id: Attachment.ID) => + path.join(sessionDirectory(root, sessionID), id) + +const directoryEntry = (entry: FSUtil.DirEntry) => entry.type === "directory" +const attachmentEntry = (entry: FSUtil.DirEntry) => entry.type === "directory" && /^att_[0-9A-Za-z]+$/.test(entry.name) +const namedEntry = (name: string) => (entry: FSUtil.DirEntry) => entry.name === name +const defined = (value: A | undefined): value is A => value !== undefined + +const makeLayer = (options: { readonly limits?: Partial; readonly now?: () => number } = {}) => + Layer.effect( + Service, + Effect.gen(function* () { + const fs = yield* FSUtil.Service + const global = yield* Global.Service + const root = path.join(global.data, MANAGED_DIRECTORY) + const limits = { ...defaults, ...options.limits } + const now = options.now ?? Date.now + const locks = KeyedMutex.makeUnsafe() + const quota = Semaphore.makeUnsafe(1) + // TODO(review): Enforce the global quota across processes before multiple processes share one data directory. + const state: StoreState = { usage: undefined, reserved: 0 } + const decodeMetadata = Schema.decodeUnknownEffect(Metadata) + + const storage = (operation: StorageError["operation"], cause: unknown) => new StorageError({ operation, cause }) + const writeError = (cause: unknown) => storage("write", cause) + const renameError = (cause: unknown) => storage("rename", cause) + + const readStoredMetadata = Effect.fn("AttachmentStore.readMetadata")(function* (file: string) { + const input = yield* fs.readJson(file).pipe(Effect.mapError((cause) => storage("read", cause))) + return yield* decodeMetadata(input).pipe(Effect.mapError((cause) => storage("read", cause))) + }) + + function storedSize(session: FSUtil.DirEntry, entry: FSUtil.DirEntry) { + return readStoredMetadata(path.join(root, session.name, entry.name, metadataName)).pipe( + Effect.map((metadata) => metadata.size), + Effect.catch(() => Effect.succeed(0)), + ) + } + + function scanSession(session: FSUtil.DirEntry) { + return Effect.gen(function* () { + const decoded = Option.getOrUndefined(Option.liftThrowable(decodeURIComponent)(session.name)) + if (!decoded || !Schema.is(SessionSchema.ID)(decoded)) return undefined + const sessionID = SessionSchema.ID.make(decoded) + const entries = yield* fs + .readDirectoryEntries(path.join(root, session.name)) + .pipe(Effect.catch(() => Effect.succeed([]))) + const sizes = yield* Effect.forEach(entries.filter(attachmentEntry), (entry) => storedSize(session, entry)) + return [sessionID, sizes.reduce((sum, value) => sum + value, 0)] satisfies readonly [SessionSchema.ID, number] + }) + } + + const scan = Effect.fn("AttachmentStore.scan")(function* () { + const sessionEntries = yield* fs.readDirectoryEntries(root).pipe(Effect.catch(() => Effect.succeed([]))) + const sizes = yield* Effect.forEach(sessionEntries.filter(directoryEntry), scanSession) + const sessions = new Map(sizes.filter(defined)) + return { sessions, global: Array.from(sessions.values()).reduce((sum, value) => sum + value, 0) } + }) + + const usage = Effect.fn("AttachmentStore.usage")(function* () { + if (state.usage) return state.usage + state.usage = yield* scan() + return state.usage + }) + + const reserve = (reservation: Reservation, bytes: number) => + quota.withPermit( + Effect.gen(function* () { + const next = reservation.bytes + bytes + if (next > limits.file) return yield* new QuotaError({ scope: "file", maximumBytes: limits.file }) + const current = yield* usage() + if ((current.sessions.get(reservation.sessionID) ?? 0) + next > limits.session) + return yield* new QuotaError({ scope: "session", maximumBytes: limits.session }) + if (current.global + state.reserved + bytes > limits.global) + return yield* new QuotaError({ scope: "global", maximumBytes: limits.global }) + reservation.bytes = next + state.reserved += bytes + }), + ) + + const release = (reservation: Reservation) => + quota.withPermit( + Effect.sync(() => { + state.reserved -= reservation.bytes + reservation.bytes = 0 + }), + ) + + const commit = (reservation: Reservation) => + quota.withPermit( + Effect.gen(function* () { + const current = yield* usage() + state.reserved -= reservation.bytes + current.global += reservation.bytes + current.sessions.set( + reservation.sessionID, + (current.sessions.get(reservation.sessionID) ?? 0) + reservation.bytes, + ) + reservation.bytes = 0 + }), + ) + + const allocate: ( + sessionID: SessionSchema.ID, + ) => Effect.Effect<{ readonly id: Attachment.ID; readonly directory: string }, StorageError> = Effect.fn( + "AttachmentStore.allocate", + )(function* (sessionID: SessionSchema.ID) { + yield* fs + .makeDirectory(root, { recursive: true, mode: 0o700 }) + .pipe(Effect.mapError((cause) => storage("allocate", cause))) + const rootEntry = (yield* fs + .readDirectoryEntries(global.data) + .pipe(Effect.mapError((cause) => storage("scan", cause)))).find((entry) => entry.name === MANAGED_DIRECTORY) + if (rootEntry?.type !== "directory") return yield* new StorageError({ operation: "allocate", cause: "symlink" }) + yield* fs.chmod(root, 0o700).pipe(Effect.mapError((cause) => storage("allocate", cause))) + const session = sessionDirectory(root, sessionID) + yield* fs + .makeDirectory(session, { recursive: true, mode: 0o700 }) + .pipe(Effect.mapError((cause) => storage("allocate", cause))) + const entry = (yield* fs + .readDirectoryEntries(root) + .pipe(Effect.mapError((cause) => storage("scan", cause)))).find( + (entry) => entry.name === path.basename(session), + ) + if (entry?.type !== "directory") return yield* new StorageError({ operation: "allocate", cause: "symlink" }) + yield* fs.chmod(session, 0o700).pipe(Effect.mapError((cause) => storage("allocate", cause))) + const id = Attachment.ID.create() + const directory = attachmentDirectory(root, sessionID, id) + const created = yield* fs.makeDirectory(directory, { mode: 0o700 }).pipe( + Effect.as(true), + Effect.catchReason("PlatformError", "AlreadyExists", () => Effect.succeed(false)), + Effect.mapError((cause) => storage("allocate", cause)), + ) + if (!created) return yield* allocate(sessionID) + return { id, directory } + }) + + const read = Effect.fn("AttachmentStore.resolve")(function* (input: { + readonly sessionID: SessionSchema.ID + readonly attachmentID: Attachment.ID + }) { + const directory = attachmentDirectory(root, input.sessionID, input.attachmentID) + const entries = yield* fs + .readDirectoryEntries(directory) + .pipe(Effect.mapError(() => new ReferenceError({ ...input }))) + const metadataEntry = entries.find((entry) => entry.name === metadataName) + if (metadataEntry?.type !== "file") return yield* new ReferenceError({ ...input }) + const metadata = yield* readStoredMetadata(path.join(directory, metadataName)).pipe( + Effect.mapError(() => new ReferenceError({ ...input })), + ) + if (metadata.id !== input.attachmentID || metadata.sessionID !== input.sessionID) + return yield* new ReferenceError({ ...input }) + const fileEntry = entries.find((entry) => entry.name === metadata.storedName) + if (fileEntry?.type !== "file") return yield* new ReferenceError({ ...input }) + const file = path.join(directory, metadata.storedName) + const realDirectory = yield* fs + .realPath(directory) + .pipe(Effect.mapError(() => new ReferenceError({ ...input }))) + const real = yield* fs.realPath(file).pipe(Effect.mapError(() => new ReferenceError({ ...input }))) + if (!FSUtil.contains(realDirectory, real)) return yield* new ReferenceError({ ...input }) + return { + id: metadata.id, + uri: Attachment.URI.fromID(metadata.id), + name: metadata.storedName, + mime: metadata.detectedMime, + size: metadata.size, + path: real, + nativeMediaDelivered: metadata.nativeMediaDeliveredAt !== undefined, + } + }) + + const writeMetadata = Effect.fn("AttachmentStore.writeMetadata")(function* ( + directory: string, + metadata: Metadata, + ) { + const temp = path.join(directory, metadataUploadName) + yield* fs + .writeFileString(temp, JSON.stringify(metadata, null, 2), { flag: "wx", mode: 0o600 }) + .pipe(Effect.mapError((cause) => storage("write", cause))) + yield* fs + .rename(temp, path.join(directory, metadataName)) + .pipe(Effect.mapError((cause) => storage("rename", cause))) + }) + + function sameFile(left: FileSystem.File.Info, right: FileSystem.File.Info) { + const leftInode = Option.getOrUndefined(left.ino) + const rightInode = Option.getOrUndefined(right.ino) + return ( + left.type === "File" && + right.type === "File" && + left.dev === right.dev && + leftInode !== undefined && + leftInode === rightInode + ) + } + + const uploadUnlocked = (input: UploadInput): Effect.Effect => + Effect.gen(function* () { + const name = yield* sanitizeName(input.name) + const allocated = yield* allocate(input.sessionID) + const reservation: Reservation = { sessionID: input.sessionID, bytes: 0 } + const hash = createHash("sha256") + const prefix = Buffer.alloc(16) + const progress = { prefix: 0 } + function writeChunk(file: FileSystem.File, chunk: Uint8Array) { + return Effect.gen(function* () { + yield* reserve(reservation, chunk.byteLength) + hash.update(chunk) + const copied = Math.min(prefix.length - progress.prefix, chunk.byteLength) + if (copied > 0) prefix.set(chunk.subarray(0, copied), progress.prefix) + progress.prefix += copied + yield* file.writeAll(chunk).pipe(Effect.mapError(writeError)) + }) + } + function write() { + return Effect.scoped( + Effect.gen(function* () { + const source = path.join(allocated.directory, uploadName) + const file = yield* fs.open(source, { flag: "wx", mode: 0o600 }).pipe(Effect.mapError(writeError)) + function consume(chunk: Uint8Array) { + return writeChunk(file, chunk) + } + yield* Stream.runForEach(input.content, consume) + yield* file.sync.pipe(Effect.mapError(writeError)) + return yield* finish(file, source) + }), + ) + } + function finish(file: FileSystem.File, source: string) { + return Effect.gen(function* () { + const mime = sniff(prefix.subarray(0, progress.prefix)) + const size = reservation.bytes + const realRoot = yield* fs.realPath(root).pipe(Effect.mapError(renameError)) + const realDirectory = yield* fs.realPath(allocated.directory).pipe(Effect.mapError(renameError)) + const realSource = yield* fs.realPath(source).pipe(Effect.mapError(renameError)) + const target = path.join(realDirectory, name) + const entries = yield* fs.readDirectoryEntries(realDirectory).pipe(Effect.mapError(renameError)) + const sourceEntry = entries.find(namedEntry(uploadName)) + const descriptorInfo = yield* file.stat.pipe(Effect.mapError(renameError)) + const sourceInfo = yield* fs.stat(realSource).pipe(Effect.mapError(renameError)) + const sourceValid = ![ + sourceEntry?.type !== "file", + entries.some(namedEntry(name)), + !FSUtil.contains(realRoot, realDirectory), + !FSUtil.contains(realDirectory, realSource), + !FSUtil.contains(realDirectory, target), + !sameFile(descriptorInfo, sourceInfo), + ].includes(true) + if (!sourceValid) return yield* new StorageError({ operation: "rename", cause: "containment" }) + yield* fs.rename(realSource, target).pipe(Effect.mapError(renameError)) + const realTarget = yield* fs.realPath(target).pipe(Effect.mapError(renameError)) + const targetEntry = (yield* fs + .readDirectoryEntries(realDirectory) + .pipe(Effect.mapError(renameError))).find(namedEntry(name)) + const targetInfo = yield* fs.stat(realTarget).pipe(Effect.mapError(renameError)) + const targetValid = ![ + targetEntry?.type !== "file", + !FSUtil.contains(realDirectory, realTarget), + !sameFile(descriptorInfo, targetInfo), + ].includes(true) + if (!targetValid) return yield* new StorageError({ operation: "rename", cause: "containment" }) + yield* writeMetadata(realDirectory, { + id: allocated.id, + sessionID: input.sessionID, + originalName: input.name, + storedName: name, + clientMime: input.contentType, + detectedMime: mime, + size, + sha256: hash.digest("hex"), + createdAt: now(), + }) + yield* commit(reservation) + return Attachment.Info.make({ + id: allocated.id, + uri: Attachment.URI.fromID(allocated.id), + name, + mime, + size, + }) + }) + } + function discard() { + return fs.remove(allocated.directory, { recursive: true }).pipe(Effect.catch(() => Effect.void)) + } + return yield* write().pipe(Effect.onError(discard), Effect.ensuring(release(reservation))) + }) + + const upload: Interface["upload"] = (input) => locks.withLock(input.sessionID)(uploadUnlocked(input)) + + const resolve: Interface["resolve"] = read + + const updateMetadata = Effect.fn("AttachmentStore.updateMetadata")(function* ( + input: { readonly sessionID: SessionSchema.ID; readonly attachmentID: Attachment.ID }, + update: (metadata: Metadata) => Metadata, + ) { + const resolved = yield* read(input) + const directory = attachmentDirectory(root, input.sessionID, input.attachmentID) + const metadata = yield* readStoredMetadata(path.join(directory, metadataName)).pipe( + Effect.mapError(() => new ReferenceError({ ...input })), + ) + const next = update(metadata) + if (next === metadata) return resolved + yield* fs + .writeFileString(path.join(directory, metadataUploadName), JSON.stringify(next, null, 2), { + flag: "wx", + mode: 0o600, + }) + .pipe(Effect.mapError((cause) => storage("write", cause))) + yield* fs + .rename(path.join(directory, metadataUploadName), path.join(directory, metadataName)) + .pipe(Effect.mapError((cause) => storage("rename", cause))) + return { ...resolved, nativeMediaDelivered: next.nativeMediaDeliveredAt !== undefined } + }) + + const bind: Interface["bind"] = (input) => + locks.withLock(input.sessionID)( + updateMetadata(input, (metadata) => + metadata.boundMessageID ? metadata : { ...metadata, boundMessageID: input.messageID }, + ), + ) + + const markNativeMediaDelivered: Interface["markNativeMediaDelivered"] = (input) => + locks.withLock(input.sessionID)( + updateMetadata(input, (metadata) => + metadata.nativeMediaDeliveredAt === undefined + ? { ...metadata, nativeMediaDeliveredAt: now() } + : metadata, + ), + ) + + const remove: Interface["remove"] = (input) => + locks.withLock(input.sessionID)( + fs.remove(attachmentDirectory(root, input.sessionID, input.attachmentID), { recursive: true }).pipe( + Effect.catchReason("PlatformError", "NotFound", () => Effect.void), + Effect.mapError((cause) => storage("remove", cause)), + Effect.andThen(quota.withPermit(Effect.sync(() => (state.usage = undefined)))), + ), + ) + + function cleanupAttachment(directory: string, cutoff: number, attachment: FSUtil.DirEntry) { + const target = path.join(directory, attachment.name) + function remove() { + return fs.remove(target, { recursive: true }) + } + function stored(metadata: Metadata) { + return !metadata.boundMessageID && metadata.createdAt < cutoff ? remove() : Effect.void + } + function partial() { + function stale(info: FileSystem.File.Info) { + return Option.getOrElse(info.mtime, () => new Date(0)).getTime() < cutoff ? remove() : Effect.void + } + return fs.stat(target).pipe( + Effect.flatMap(stale), + Effect.catch(() => Effect.void), + ) + } + return readStoredMetadata(path.join(target, metadataName)).pipe(Effect.flatMap(stored), Effect.catch(partial)) + } + + function cleanupSession(input: { + readonly sessionID: SessionSchema.ID + readonly directory: string + readonly orphan: boolean + readonly cutoff: number + }) { + return locks.withLock(input.sessionID)( + Effect.gen(function* () { + if (input.orphan) { + const info = yield* fs.stat(input.directory).pipe(Effect.catch(() => Effect.void)) + const modified = info?.mtime.pipe( + Option.map((date) => date.getTime()), + Option.getOrElse(() => 0), + ) + if (modified !== undefined && modified < input.cutoff) + yield* fs.remove(input.directory, { recursive: true }).pipe(Effect.catch(() => Effect.void)) + return + } + const attachments = yield* fs + .readDirectoryEntries(input.directory) + .pipe(Effect.catch(() => Effect.succeed([]))) + yield* Effect.forEach(attachments.filter(directoryEntry), (attachment) => + cleanupAttachment(input.directory, input.cutoff, attachment), + ) + }), + ) + } + + const cleanup = Effect.fn("AttachmentStore.cleanup")(function* (sessions?: ReadonlySet) { + const cutoff = now() - Duration.toMillis(UNBOUND_RETENTION) + const roots = yield* fs.readDirectoryEntries(root).pipe(Effect.catch(() => Effect.succeed([]))) + yield* Effect.forEach( + roots.filter((entry) => entry.type === "directory"), + (entry) => { + const decoded = Option.getOrUndefined(Option.liftThrowable(decodeURIComponent)(entry.name)) + if (!decoded || !Schema.is(SessionSchema.ID)(decoded)) return Effect.void + const sessionID = SessionSchema.ID.make(decoded) + return cleanupSession({ + sessionID, + directory: path.join(root, entry.name), + orphan: sessions !== undefined && !sessions.has(sessionID), + cutoff, + }) + }, + ) + yield* quota.withPermit(Effect.sync(() => (state.usage = undefined))) + }) + + return Service.of({ upload, resolve, bind, markNativeMediaDelivered, remove, cleanup }) + }), + ) + +export const layerWith = (options: { readonly limits?: Partial; readonly now?: () => number } = {}) => + makeLayer(options) + +const layer = makeLayer() + +export const node = Node.tags.make("global")({ + service: Service, + layer, + deps: [FSUtil.node, Global.node], +}) + +const cleanupLayer = Layer.effectDiscard( + Effect.gen(function* () { + const store = yield* Service + const { db } = yield* Database.Service + const cleanup = Effect.gen(function* () { + const rows = yield* db.select({ id: SessionTable.id }).from(SessionTable).all().pipe(Effect.orDie) + yield* store.cleanup(new Set(rows.map((row) => SessionSchema.ID.make(row.id)))) + }) + yield* cleanup.pipe(Effect.repeat(Schedule.spaced(Duration.hours(1))), Effect.forkScoped) + }), +) + +export const cleanupNode = Node.tags.make("global")({ + name: "attachment-cleanup", + layer: cleanupLayer, + deps: [node, Database.node], +}) diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 2dabfb2d6fba..0425e5cc4bd3 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -9,7 +9,7 @@ import { WorkspaceV2 } from "./workspace" import { ModelV2 } from "./model" import { Location } from "./location" import { SessionMessage } from "./session/message" -import { Prompt } from "./session/prompt" +import { Prompt, type FileAttachment } from "./session/prompt" import { PromptInput } from "@opencode-ai/schema/prompt-input" import { EventV2 } from "./event" import { Database } from "./database/database" @@ -37,10 +37,27 @@ import { SessionRevert } from "./session/revert" import { Revert } from "@opencode-ai/schema/revert" import { FSUtil } from "./fs-util" import { SessionDurable } from "@opencode-ai/schema/durable-event-manifest" +import { AttachmentStore } from "./attachment-store" export const RevertState = Revert.State export type RevertState = Revert.State +const bindAttachments = Effect.fn("V2Session.bindAttachments")(function* ( + attachments: AttachmentStore.Interface, + sessionID: SessionSchema.ID, + messageID: SessionMessage.ID, + files: readonly FileAttachment[], +) { + yield* Effect.forEach( + files, + (file) => { + const attachmentID = AttachmentStore.attachmentID(file.uri) + return attachmentID ? attachments.bind({ sessionID, attachmentID, messageID }) : Effect.void + }, + { discard: true }, + ) +}) + // get project -> project.locations // // get all sessions @@ -108,7 +125,13 @@ export class PromptConflictError extends Schema.TaggedErrorClass Effect.Effect @@ -150,7 +173,10 @@ export interface Interface { prompt: PromptInput.Prompt delivery?: SessionInput.Delivery resume?: boolean - }) => Effect.Effect + }) => Effect.Effect< + SessionInput.Admitted, + NotFoundError | PromptConflictError | AttachmentStore.ReferenceError | AttachmentStore.StorageError + > readonly shell: (input: { id?: EventV2.ID sessionID: SessionSchema.ID @@ -191,6 +217,7 @@ const layer = Layer.effect( const execution = yield* SessionExecution.Service const store = yield* SessionStore.Service const locations = yield* LocationServiceMap.Service + const attachments = yield* AttachmentStore.Service const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message) const isDurableSessionEvent = Schema.is(SessionEvent.Durable) const decode = (row: typeof SessionMessageTable.$inferSelect) => @@ -361,8 +388,8 @@ const layer = Layer.effect( Effect.uninterruptible( Effect.gen(function* () { yield* result.get(input.sessionID) - const prompt = resolvePrompt(input.prompt) const messageID = input.id ?? SessionMessage.ID.create() + const prompt = yield* resolvePrompt(attachments, input.sessionID, input.prompt) const delivery = input.delivery ?? "steer" const expected = { sessionID: input.sessionID, messageID, prompt, delivery } const admitted = yield* SessionInput.admit(db, events, { @@ -379,6 +406,7 @@ const layer = Layer.effect( ) if (!SessionInput.equivalent(admitted, expected)) return yield* new PromptConflictError({ sessionID: input.sessionID, messageID }) + yield* bindAttachments(attachments, input.sessionID, messageID, prompt.files ?? []) if (input.resume !== false) yield* execution.wake(admitted.sessionID) return admitted }), @@ -457,19 +485,28 @@ const layer = Layer.effect( }), ) -const resolvePrompt = (input: PromptInput.Prompt) => - Prompt.make({ - text: input.text, - agents: input.agents, - files: input.files?.map((file) => { +const resolvePrompt = Effect.fn("V2Session.resolvePrompt")(function* ( + attachments: AttachmentStore.Interface, + sessionID: SessionSchema.ID, + input: PromptInput.Prompt, +) { + const files = yield* Effect.forEach(input.files ?? [], (file) => { + if (!AttachmentStore.isManagedURI(file.uri)) { const dataMime = file.uri.match(/^data:([^;,]+)[;,]/i)?.[1] const target = URL.canParse(file.uri) ? new URL(file.uri).pathname : (file.name ?? file.uri) - return { + return Effect.succeed({ ...file, mime: dataMime ?? (target.endsWith("/") ? "application/x-directory" : FSUtil.mimeType(target)), - } - }), + }) + } + const attachmentID = AttachmentStore.attachmentID(file.uri) + if (!attachmentID) return Effect.fail(new AttachmentStore.ReferenceError({ sessionID })) + return attachments + .resolve({ sessionID, attachmentID }) + .pipe(Effect.map((resolved) => ({ ...file, name: resolved.name, mime: resolved.mime }))) }) + return Prompt.make({ text: input.text, agents: input.agents, files: input.files ? files : undefined }) +}) export const node = makeGlobalNode({ service: Service, @@ -482,5 +519,6 @@ export const node = makeGlobalNode({ SessionStore.node, LocationServiceMap.node, SessionProjector.node, + AttachmentStore.node, ], }) diff --git a/packages/core/src/session/runner/attachment-materialization.ts b/packages/core/src/session/runner/attachment-materialization.ts new file mode 100644 index 000000000000..80fc229d0869 --- /dev/null +++ b/packages/core/src/session/runner/attachment-materialization.ts @@ -0,0 +1,98 @@ +import { type Model } from "@opencode-ai/llm" +import { Effect } from "effect" +import { AttachmentStore } from "../../attachment-store" +import { ModelV2 } from "../../model" +import { SessionMessage } from "../message" +import { SessionSchema } from "../schema" +import type { MaterializedAttachment, NativeAttachment } from "./to-llm-message" + +interface Candidate { + readonly file: NonNullable[number] + readonly current: boolean +} + +export interface Materialization { + readonly attachments: ReadonlyMap + readonly native: ReadonlyArray +} + +const readMedia = (attachment: AttachmentStore.Resolved) => + Effect.tryPromise({ + try: async () => { + const data = new Uint8Array(attachment.size) + const progress = { offset: 0 } + for await (const chunk of Bun.file(attachment.path).stream()) { + if (progress.offset + chunk.byteLength > data.byteLength) throw new Error("Attachment size changed") + data.set(chunk, progress.offset) + progress.offset += chunk.byteLength + } + if (progress.offset !== data.byteLength) throw new Error("Attachment size changed") + return data + }, + catch: () => undefined, + }).pipe(Effect.catch(() => Effect.succeed(undefined))) + +const nativeMedia = ( + candidate: Candidate, + attachment: AttachmentStore.Resolved, + model: Model, + inputCapabilities: ModelV2.Capabilities["input"], +): Effect.Effect => { + const mime = attachment.mime.toLowerCase() + if (!candidate.current || attachment.nativeMediaDelivered) return Effect.succeed(undefined) + if (candidate.file.mime.toLowerCase() !== mime) return Effect.succeed(undefined) + const admission = model.route.media({ mime, bytes: attachment.size }) + if (!admission || !inputCapabilities.includes(admission.capability)) return Effect.succeed(undefined) + return readMedia(attachment).pipe( + Effect.map((data) => (data ? { type: "media" as const, path: attachment.path, mime, data } : undefined)), + ) +} + +const resolveCandidate = Effect.fn("SessionRunner.resolveAttachment")(function* (input: { + readonly store: AttachmentStore.Interface + readonly sessionID: SessionSchema.ID + readonly model: Model + readonly inputCapabilities: ModelV2.Capabilities["input"] + readonly candidate: Candidate +}) { + const attachmentID = AttachmentStore.attachmentID(input.candidate.file.uri) + if (!attachmentID) return yield* new AttachmentStore.ReferenceError({ sessionID: input.sessionID }) + const attachment = yield* input.store.resolve({ sessionID: input.sessionID, attachmentID }) + const native = yield* nativeMedia(input.candidate, attachment, input.model, input.inputCapabilities) + return { + uri: input.candidate.file.uri, + materialized: native ?? { type: "path" as const, path: attachment.path }, + native: native ? attachment : undefined, + } +}) + +export const materializeAttachments = Effect.fn("SessionRunner.materializeAttachments")(function* (input: { + readonly store: AttachmentStore.Interface + readonly sessionID: SessionSchema.ID + readonly model: Model + readonly inputCapabilities: ModelV2.Capabilities["input"] + readonly context: readonly SessionMessage.Message[] +}) { + const lastAssistant = input.context.findLastIndex((message) => message.type === "assistant") + const candidates = input.context.flatMap((message, index): Candidate[] => + message.type === "user" + ? (message.files ?? []) + .filter((file) => AttachmentStore.isManagedURI(file.uri)) + .map((file) => ({ file, current: index > lastAssistant })) + : [], + ) + const unique = Array.from(new Map(candidates.map((candidate) => [candidate.file.uri, candidate])).values()) + const resolved = yield* Effect.forEach(unique, (candidate) => + resolveCandidate({ + store: input.store, + sessionID: input.sessionID, + model: input.model, + inputCapabilities: input.inputCapabilities, + candidate, + }), + ) + return { + attachments: new Map(resolved.map((item) => [item.uri, item.materialized])), + native: resolved.flatMap((item) => (item.native ? [item.native] : [])), + } satisfies Materialization +}) diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 634075dd91b2..acf50c3f637d 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -7,6 +7,7 @@ import type { ContextSnapshotDecodeError, MessageDecodeError } from "../error" import { SessionRunnerModel } from "./model" import type { SystemContext } from "../../system-context/index" import type { ToolOutputStore } from "../../tool-output-store" +import type { AttachmentStore } from "../../attachment-store" export type RunError = | LLMError @@ -15,6 +16,7 @@ export type RunError = | ContextSnapshotDecodeError | SystemContext.InitializationBlocked | ToolOutputStore.Error + | AttachmentStore.Error /** Runs one local continuation from already-recorded Session history. */ export interface Interface { diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 874086a06bdb..05f65a015fd7 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -39,6 +39,8 @@ import { MAX_STEPS_PROMPT } from "./max-steps" import { Snapshot } from "../../snapshot" import { makeLocationNode } from "../../effect/app-node" import { llmClient } from "../../effect/app-node-platform" +import { AttachmentStore } from "../../attachment-store" +import { materializeAttachments } from "./attachment-materialization" /** * Runs one durable coding-agent Session until it settles. @@ -105,6 +107,7 @@ const layer = Layer.effect( const referenceGuidance = yield* ReferenceGuidance.Service const config = yield* Config.Service const snapshots = yield* Snapshot.Service + const attachments = yield* AttachmentStore.Service const db = (yield* Database.Service).db const compaction = SessionCompaction.make({ events, llm, config: yield* config.entries() }) const getSession = Effect.fn("SessionRunner.getSession")(function* (sessionID: SessionSchema.ID) { @@ -196,9 +199,17 @@ const layer = Layer.effect( } const system = initialized ?? (yield* SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id)) - const model = yield* models.resolve(session) + const selection = yield* models.resolve(session) + const model = selection.model const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) const context = entries.map((entry) => entry.message) + const materialized = yield* materializeAttachments({ + store: attachments, + sessionID: session.id, + model, + inputCapabilities: selection.inputCapabilities, + context, + }) const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps const toolMaterialization = isLastStep ? undefined : yield* tools.materialize(agent.info?.permissions) const promptCacheKey = /^ses_[0-9a-f]{64}$/.test(session.id) ? session.id.slice(4) : session.id @@ -215,7 +226,10 @@ const layer = Layer.effect( system: [agent.info?.system, system.baseline] .filter((part): part is string => part !== undefined && part.length > 0) .map(SystemPart.make), - messages: [...toLLMMessages(context, model), ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : [])], + messages: [ + ...toLLMMessages(context, model, materialized.attachments), + ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : []), + ], tools: toolMaterialization?.definitions ?? [], toolChoice: isLastStep ? "none" : undefined, }) @@ -236,6 +250,10 @@ const layer = Layer.effect( const publish = (event: LLMEvent, outputPaths: ReadonlyArray = []) => withPublication(publisher.publish(event, outputPaths)) let overflowFailure: ProviderErrorEvent | undefined + // Mark immediately before provider I/O for at-most-once delivery. Attachment metadata is the smallest durable marker. + yield* Effect.forEach(materialized.native, (attachment) => + attachments.markNativeMediaDelivered({ sessionID: session.id, attachmentID: attachment.id }), + ) const providerStream = llm.stream(request).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -435,5 +453,6 @@ export const node = makeLocationNode({ Config.node, Snapshot.node, Database.node, + AttachmentStore.node, ], }) diff --git a/packages/core/src/session/runner/model.ts b/packages/core/src/session/runner/model.ts index 74e78120c20e..ab3d2ccbedae 100644 --- a/packages/core/src/session/runner/model.ts +++ b/packages/core/src/session/runner/model.ts @@ -71,8 +71,13 @@ export type Error = | UnsupportedApiError | Integration.AuthorizationError +export interface Selection { + readonly model: Model + readonly inputCapabilities: ModelV2.Capabilities["input"] +} + export interface Interface { - readonly resolve: (session: SessionSchema.Info) => Effect.Effect + readonly resolve: (session: SessionSchema.Info) => Effect.Effect } export class Service extends Context.Service()("@opencode/v2/SessionRunnerModel") {} @@ -131,7 +136,7 @@ const apiName = (model: ModelV2.Info) => export const fromCatalogModel = ( model: ModelV2.Info, credential?: Credential.Value, -): Effect.Effect => { +): Effect.Effect => { const resolved = credential?.type !== "key" || credential.metadata === undefined ? model @@ -140,25 +145,28 @@ export const fromCatalogModel = ( }) const key = apiKey(resolved, credential) if (resolved.api.type === "aisdk" && resolved.api.package === "@ai-sdk/openai") { - return Effect.succeed( - withDefaults(resolved, OpenAIResponses.route) + return Effect.succeed({ + model: withDefaults(resolved, OpenAIResponses.route) .with({ auth: key === undefined ? Auth.none : Auth.bearer(key) }) .model({ id: resolved.api.id }), - ) + inputCapabilities: resolved.capabilities.input, + }) } if (resolved.api.type === "aisdk" && resolved.api.package === "@ai-sdk/anthropic") { - return Effect.succeed( - withDefaults(resolved, AnthropicMessages.route) + return Effect.succeed({ + model: withDefaults(resolved, AnthropicMessages.route) .with({ auth: key === undefined ? Auth.none : Auth.header("x-api-key", key) }) .model({ id: resolved.api.id }), - ) + inputCapabilities: resolved.capabilities.input, + }) } if (resolved.api.type === "aisdk" && resolved.api.package === "@ai-sdk/openai-compatible" && resolved.api.url) { - return Effect.succeed( - withDefaults(resolved, OpenAICompatibleChat.route) + return Effect.succeed({ + model: withDefaults(resolved, OpenAICompatibleChat.route) .with({ auth: key === undefined ? Auth.none : Auth.bearer(key) }) .model({ id: resolved.api.id }), - ) + inputCapabilities: resolved.capabilities.input, + }) } return Effect.fail( new UnsupportedApiError({ diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index b2b1af5d30f1..2a07c1eb3457 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -10,6 +10,20 @@ import { import { SessionMessage } from "../message" import type { FileAttachment } from "../prompt" +export interface NativeAttachment { + readonly type: "media" + readonly path: string + readonly mime: string + readonly data: Uint8Array +} + +export interface PathAttachment { + readonly type: "path" + readonly path: string +} + +export type MaterializedAttachment = PathAttachment | NativeAttachment + const media = (file: FileAttachment): ContentPart => ({ type: "media", mediaType: file.mime, @@ -18,6 +32,22 @@ const media = (file: FileAttachment): ContentPart => ({ metadata: file.description === undefined ? undefined : { description: file.description }, }) +const userFile = (file: FileAttachment, attachments: ReadonlyMap): ContentPart => { + const attachment = attachments.get(file.uri) + if (!attachment) return media(file) + if (attachment.type === "media") + return { + type: "media", + mediaType: attachment.mime, + data: attachment.data, + filename: attachment.path, + } + return { + type: "text", + text: `Attached file: ${JSON.stringify({ name: file.name, path: attachment.path, mime: file.mime })}`, + } +} + const toolInput = (tool: SessionMessage.AssistantTool) => { if (tool.state.status !== "pending") return tool.state.input try { @@ -112,7 +142,11 @@ const assistant = (message: SessionMessage.Assistant, model: Model) => { ] } -function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] { +function toLLMMessage( + message: SessionMessage.Message, + model: Model, + attachments: ReadonlyMap, +): Message[] { switch (message.type) { case "agent-switched": case "model-switched": @@ -122,7 +156,10 @@ function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] Message.make({ id: message.id, role: "user", - content: [{ type: "text", text: message.text }, ...(message.files ?? []).map(media)], + content: [ + { type: "text", text: message.text }, + ...(message.files ?? []).map((file) => userFile(file, attachments)), + ], metadata: { ...message.metadata, ...(message.agents?.length ? { agents: message.agents } : {}), @@ -167,5 +204,8 @@ ${message.recent} } /** Translate projected V2 Session history into canonical @opencode-ai/llm context. */ -export const toLLMMessages = (messages: readonly SessionMessage.Message[], model: Model) => - messages.flatMap((message) => toLLMMessage(message, model)) +export const toLLMMessages = ( + messages: readonly SessionMessage.Message[], + model: Model, + attachments: ReadonlyMap = new Map(), +) => messages.flatMap((message) => toLLMMessage(message, model, attachments)) diff --git a/packages/core/test/attachment-store.test.ts b/packages/core/test/attachment-store.test.ts new file mode 100644 index 000000000000..573077db66e7 --- /dev/null +++ b/packages/core/test/attachment-store.test.ts @@ -0,0 +1,318 @@ +import { describe, expect } from "bun:test" +import fs from "fs/promises" +import path from "path" +import { Deferred, Effect, Exit, Fiber, Latch, Layer, Stream } from "effect" +import { AttachmentStore } from "@opencode-ai/core/attachment-store" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Global } from "@opencode-ai/core/global" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionV2 } from "@opencode-ai/core/session" +import { testEffect } from "./lib/effect" +import { tmpdir } from "./fixture/tmpdir" + +const first = SessionV2.ID.make("ses_attachment_first") +const second = SessionV2.ID.make("ses_attachment_second") +const bytes = (size: number, value = 1) => new Uint8Array(size).fill(value) +const stat = (target: string) => Effect.promise(() => fs.stat(target)) +const readDirectory = (target: string) => Effect.promise(() => fs.readdir(target)) +const readBytes = (target: string) => Effect.promise(() => Bun.file(target).bytes()) +const readJson = (target: string) => Effect.promise(() => Bun.file(target).json()) +const attachmentDirectoryName = (name: string) => name.startsWith("att_") +const sequence = (length: number, offset: number) => Array.from({ length }, (_, index) => index + offset) +const infoName = (info: AttachmentStore.Info) => info.name +const contentBytes = (content: Uint8Array) => Array.from(content) + +const withStore = ( + body: Effect.Effect, + options: Parameters[0] = {}, +) => + Effect.acquireUseRelease( + Effect.promise(() => tmpdir()), + (tmp) => { + const global = Global.layerWith({ data: tmp.path }) + const layer = AttachmentStore.layerWith(options).pipe( + Layer.provide(LayerNode.compile(FSUtil.node)), + Layer.provide(global), + ) + return body.pipe(Effect.provide(Layer.merge(layer, global))) + }, + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + +const upload = (store: AttachmentStore.Interface, sessionID: SessionV2.ID, name: string, content: Uint8Array[]) => + store.upload({ + sessionID, + name, + contentType: "application/example", + content: Stream.fromIterable(content), + }) + +const it = testEffect(Layer.empty) + +describe("AttachmentStore", () => { + it.live("sanitizes hostile filenames and rejects NUL", () => + Effect.gen(function* () { + expect(yield* AttachmentStore.sanitizeName("../../cafe\u0301.txt")).toBe("café.txt") + expect(yield* AttachmentStore.sanitizeName("photo\u202egnp.exe")).toBe("photognp.exe") + expect(yield* AttachmentStore.sanitizeName("photo\u061c\u200e\u200f.png")).toBe("photo.png") + expect(yield* AttachmentStore.sanitizeName("C:\\temp\\CON.txt. ")).toBe("_CON.txt") + expect(yield* AttachmentStore.sanitizeName("CON .txt")).toBe("_CON .txt") + expect(yield* AttachmentStore.sanitizeName("con.report.txt")).toBe("_con.report.txt") + expect(yield* AttachmentStore.sanitizeName("COM¹.txt")).toBe("_COM¹.txt") + expect(yield* AttachmentStore.sanitizeName("LPT³ .txt")).toBe("_LPT³ .txt") + expect(yield* AttachmentStore.sanitizeName("../.. ")).toBe("attachment") + expect(yield* AttachmentStore.sanitizeName("a".repeat(300) + ".txt")).toHaveLength(180) + expect((yield* AttachmentStore.sanitizeName("bad\0name").pipe(Effect.exit))._tag).toBe("Failure") + }), + ) + + it.live("writes the first chunk before requesting the rest of a 20 MiB stream", () => + withStore( + Effect.gen(function* () { + const root = (yield* Global.Service).data + const store = yield* AttachmentStore.Service + const requested = yield* Deferred.make() + const resume = yield* Latch.make() + function chunk(index: number) { + const value = bytes(20 * 1024, index % 251) + if (index === 0) value.set([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]) + return value + } + const content = Stream.concat( + Stream.make(chunk(0)), + Stream.concat( + Stream.fromEffect( + Deferred.succeed(requested, undefined).pipe(Effect.andThen(resume.await), Effect.as(chunk(1))), + ), + Stream.fromIterable(sequence(1022, 2), { chunkSize: 1 }).pipe(Stream.map(chunk)), + ), + ) + const fiber = yield* store + .upload({ sessionID: first, name: "../image.png", contentType: "application/example", content }) + .pipe(Effect.forkChild) + yield* Deferred.await(requested) + const session = path.join(root, "attachments", encodeURIComponent(first)) + const directory = (yield* readDirectory(session)).find(attachmentDirectoryName) + expect(directory).toBeDefined() + expect((yield* stat(path.join(session, directory!, ".upload"))).size).toBe(20 * 1024) + yield* resume.open + const info = yield* Fiber.join(fiber) + const resolved = yield* store.resolve({ sessionID: first, attachmentID: info.id }) + function readMetadata() { + return Bun.file(path.join(path.dirname(resolved.path), "metadata.json")).json() + } + const metadata = yield* Effect.promise(readMetadata) + + expect(info).toMatchObject({ name: "image.png", mime: "image/png", size: 20 * 1024 * 1024 }) + expect((yield* stat(resolved.path)).size).toBe(20 * 1024 * 1024) + expect(metadata).toMatchObject({ + originalName: "../image.png", + storedName: "image.png", + clientMime: "application/example", + detectedMime: "image/png", + size: 20 * 1024 * 1024, + }) + expect(metadata.sha256).toMatch(/^[0-9a-f]{64}$/) + if (process.platform !== "win32") { + expect((yield* stat(path.join(root, "attachments"))).mode & 0o777).toBe(0o700) + expect((yield* stat(path.dirname(resolved.path))).mode & 0o777).toBe(0o700) + expect((yield* stat(resolved.path)).mode & 0o777).toBe(0o600) + } + expect(resolved.path.startsWith(path.join(root, "attachments", encodeURIComponent(first)))).toBe(true) + }), + ), + ) + + it.live("keeps internal metadata names separate from uploaded content", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const inputs = ["metadata.json", ".metadata", ".upload", "METADATA.JSON"] + function put(name: string, index: number) { + return upload(store, first, name, [bytes(3, index)]) + } + const uploaded = yield* Effect.forEach(inputs, put) + expect(uploaded.map(infoName)).toEqual(["_metadata.json", "_.metadata", "_.upload", "_METADATA.JSON"]) + function read(info: AttachmentStore.Info) { + function content(resolved: AttachmentStore.Resolved) { + return readBytes(resolved.path) + } + return store.resolve({ sessionID: first, attachmentID: info.id }).pipe(Effect.flatMap(content)) + } + const contents = yield* Effect.forEach(uploaded, read) + expect(contents.map(contentBytes)).toEqual([ + [0, 0, 0], + [1, 1, 1], + [2, 2, 2], + [3, 3, 3], + ]) + }), + ), + ) + + it.live("refuses a directory symlink swap before final rename", () => + withStore( + Effect.gen(function* () { + const root = (yield* Global.Service).data + const store = yield* AttachmentStore.Service + const outside = path.join(root, "outside") + const session = path.join(root, "attachments", encodeURIComponent(first)) + async function swap() { + const entry = (await fs.readdir(session)).find(attachmentDirectoryName) + if (!entry) throw new Error("attachment directory was not allocated") + const directory = path.join(session, entry) + await fs.mkdir(outside) + await fs.writeFile(path.join(outside, ".upload"), "outside") + await fs.rename(directory, `${directory}.moved`) + await fs.symlink(outside, directory, "dir") + return bytes(1) + } + function readOutside() { + return fs.readFile(path.join(outside, ".upload"), "utf8") + } + function listOutside() { + return fs.readdir(outside) + } + const content = Stream.concat(Stream.make(bytes(1)), Stream.fromEffect(Effect.promise(swap))) + const result = yield* store + .upload({ sessionID: first, name: "payload.bin", contentType: "application/example", content }) + .pipe(Effect.exit) + + expect(Exit.isFailure(result) && result.cause.toString()).toContain("AttachmentStore.StorageError") + expect(yield* Effect.promise(readOutside)).toBe("outside") + expect(yield* Effect.promise(listOutside)).toEqual([".upload"]) + }), + ), + ) + + it.live("sniffs unknown content as octet-stream", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + expect(yield* upload(store, first, "notes.txt", [bytes(4)])).toMatchObject({ + mime: "application/octet-stream", + }) + }), + ), + ) + + it.live("enforces file, session, and global quotas", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const file = yield* upload(store, first, "file.bin", [bytes(11)]).pipe(Effect.exit) + expect(Exit.isFailure(file) && file.cause.toString()).toContain("AttachmentStore.QuotaError") + + yield* upload(store, first, "first.bin", [bytes(8)]) + const session = yield* upload(store, first, "second.bin", [bytes(5)]).pipe(Effect.exit) + expect(Exit.isFailure(session) && session.cause.toString()).toContain("AttachmentStore.QuotaError") + + yield* upload(store, second, "global.bin", [bytes(8)]) + const global = yield* upload(store, second, "overflow.bin", [bytes(1)]).pipe(Effect.exit) + expect(Exit.isFailure(global) && global.cause.toString()).toContain("AttachmentStore.QuotaError") + }), + { limits: { file: 10, session: 12, global: 16 } }, + ), + ) + + it.live("serializes concurrent uploads at a session quota boundary", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + function race(name: string) { + return upload(store, first, name, [bytes(6)]).pipe(Effect.exit) + } + const results = yield* Effect.all(["one.bin", "two.bin"].map(race), { concurrency: "unbounded" }) + expect(results.filter(Exit.isSuccess)).toHaveLength(1) + expect(results.filter(Exit.isFailure)).toHaveLength(1) + }), + { limits: { file: 10, session: 10, global: 20 } }, + ), + ) + + it.live("reserves the global quota across concurrent sessions", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const results = yield* Effect.all( + [ + upload(store, first, "first.bin", [bytes(6)]).pipe(Effect.exit), + upload(store, second, "second.bin", [bytes(6)]).pipe(Effect.exit), + ], + { concurrency: "unbounded" }, + ) + expect(results.filter(Exit.isSuccess)).toHaveLength(1) + expect(results.filter(Exit.isFailure)).toHaveLength(1) + }), + { limits: { file: 10, session: 10, global: 10 } }, + ), + ) + + it.live("rejects cross-session resolution", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const info = yield* upload(store, first, "private.bin", [bytes(1)]) + const result = yield* store.resolve({ sessionID: second, attachmentID: info.id }).pipe(Effect.exit) + expect(Exit.isFailure(result) && result.cause.toString()).toContain("AttachmentStore.ReferenceError") + }), + ), + ) + + it.live("persists native media delivery state", () => { + const clock = { now: 42 } + return withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const info = yield* upload(store, first, "image.png", [ + new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), + ]) + expect(yield* store.resolve({ sessionID: first, attachmentID: info.id })).toMatchObject({ + nativeMediaDelivered: false, + }) + + const marked = yield* store.markNativeMediaDelivered({ sessionID: first, attachmentID: info.id }) + expect(marked.nativeMediaDelivered).toBe(true) + expect(yield* store.resolve({ sessionID: first, attachmentID: info.id })).toMatchObject({ + nativeMediaDelivered: true, + }) + const metadata = yield* readJson(path.join(path.dirname(marked.path), "metadata.json")) + expect(metadata).toMatchObject({ + nativeMediaDeliveredAt: 42, + }) + }), + { now: () => clock.now }, + ) + }) + + it.live("removes expired unbound uploads and preserves bound uploads", () => { + const clock = { now: 0 } + return withStore( + Effect.gen(function* () { + const root = (yield* Global.Service).data + const store = yield* AttachmentStore.Service + const expired = yield* upload(store, first, "expired.bin", [bytes(1)]) + const bound = yield* upload(store, first, "bound.bin", [bytes(1)]) + yield* store.bind({ sessionID: first, attachmentID: bound.id, messageID: SessionMessage.ID.create() }) + clock.now = 25 * 60 * 60 * 1000 + yield* store.cleanup(new Set([first])) + + expect( + Exit.isFailure(yield* store.resolve({ sessionID: first, attachmentID: expired.id }).pipe(Effect.exit)), + ).toBe(true) + expect(yield* store.resolve({ sessionID: first, attachmentID: bound.id })).toMatchObject({ id: bound.id }) + + function expireRoot() { + return fs.utimes(path.join(root, "attachments", encodeURIComponent(first)), new Date(0), new Date(0)) + } + yield* Effect.promise(expireRoot) + yield* store.cleanup(new Set()) + expect( + Exit.isFailure(yield* store.resolve({ sessionID: first, attachmentID: bound.id }).pipe(Effect.exit)), + ).toBe(true) + }), + { now: () => clock.now }, + ) + }) +}) diff --git a/packages/core/test/session-runner-attachment-media.test.ts b/packages/core/test/session-runner-attachment-media.test.ts new file mode 100644 index 000000000000..00a4f176e383 --- /dev/null +++ b/packages/core/test/session-runner-attachment-media.test.ts @@ -0,0 +1,313 @@ +import { describe, expect } from "bun:test" +import { LLMClient, Model, type LLMRequest } from "@opencode-ai/llm" +import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat" +import * as OpenAIResponses from "@opencode-ai/llm/protocols/openai-responses" +import { AttachmentStore } from "@opencode-ai/core/attachment-store" +import { Config } from "@opencode-ai/core/config" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNodePlatform } from "@opencode-ai/core/effect/app-node-platform" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Global } from "@opencode-ai/core/global" +import { Location } from "@opencode-ai/core/location" +import { PermissionV2 } from "@opencode-ai/core/permission" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { ReferenceGuidance } from "@opencode-ai/core/reference/guidance" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { FileAttachment, Prompt } from "@opencode-ai/core/session/prompt" +import { SessionRunner } from "@opencode-ai/core/session/runner" +import { materializeAttachments } from "@opencode-ai/core/session/runner/attachment-materialization" +import * as SessionRunnerLLM from "@opencode-ai/core/session/runner/llm" +import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" +import { toLLMMessages } from "@opencode-ai/core/session/runner/to-llm-message" +import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SkillGuidance } from "@opencode-ai/core/skill/guidance" +import { Snapshot } from "@opencode-ai/core/snapshot" +import { SystemContext } from "@opencode-ai/core/system-context" +import { DateTime, Effect, Layer, Stream } from "effect" +import { eq } from "drizzle-orm" +import path from "path" +import { testEffect } from "./lib/effect" +import { tmpdir } from "./fixture/tmpdir" + +const sessionID = SessionV2.ID.make("ses_attachment_media") +const created = DateTime.makeUnsafe(0) +const png = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]) +const pdf = new TextEncoder().encode("%PDF-1.7\n") +const it = testEffect(Layer.empty) + +const model = (input: ReadonlyArray) => ({ + model: Model.make({ id: "model", provider: "provider", route: OpenAIChat.route }), + inputCapabilities: input, +}) +const responsesModel = (input: ReadonlyArray) => ({ + model: Model.make({ id: "model", provider: "provider", route: OpenAIResponses.route }), + inputCapabilities: input, +}) + +const withStore = (body: Effect.Effect) => + Effect.acquireUseRelease( + Effect.promise(() => tmpdir()), + (tmp) => + body.pipe( + Effect.provide( + AttachmentStore.layerWith().pipe( + Layer.provide(LayerNode.compile(FSUtil.node)), + Layer.provide(Global.layerWith({ data: tmp.path })), + ), + ), + ), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + +const upload = (store: AttachmentStore.Interface, name: string, content: Uint8Array) => + store.upload({ + sessionID, + name, + contentType: "application/octet-stream", + content: Stream.make(content), + }) + +const message = (uri: string, mime: string, name: string) => + SessionMessage.User.make({ + id: SessionMessage.ID.create(), + type: "user", + text: "Inspect the attachment", + files: [FileAttachment.make({ uri, mime, name })], + time: { created }, + }) + +const lower = ( + store: AttachmentStore.Interface, + selected: { readonly model: Model; readonly inputCapabilities: ReadonlyArray }, + context: readonly SessionMessage.Message[], +) => + materializeAttachments({ store, sessionID, ...selected, context }).pipe( + Effect.map((result) => ({ result, messages: toLLMMessages(context, selected.model, result.attachments) })), + ) + +const contentTypes = (messages: ReturnType) => messages.map((item) => item.content[1]?.type) + +const requests: LLMRequest[] = [] +const crash = { next: false } +const client = Layer.succeed( + LLMClient.Service, + LLMClient.Service.of({ + prepare: () => Effect.die("unused"), + stream: (request) => { + requests.push(request) + if (!crash.next) return Stream.empty + crash.next = false + throw new Error("simulated provider process crash") + }, + generate: () => Effect.die("unused"), + }), +) +const selection = responsesModel(["text", "image"]) +const models = SessionRunnerModel.layerWith(() => Effect.succeed(selection)) +const permission = Layer.mock(PermissionV2.Service, { + assert: () => Effect.die("unused"), + ask: () => Effect.die("unused"), + reply: () => Effect.die("unused"), + get: () => Effect.die("unused"), + forSession: () => Effect.die("unused"), + list: () => Effect.die("unused"), +}) +const skills = Layer.mock(SkillGuidance.Service, { load: () => Effect.succeed(SystemContext.empty) }) +const references = Layer.mock(ReferenceGuidance.Service, { load: () => Effect.succeed(SystemContext.empty) }) +const config = Layer.mock(Config.Service, { entries: () => Effect.succeed([]) }) +const execution = Layer.succeed( + SessionExecution.Service, + SessionExecution.Service.of({ + active: Effect.succeed(new Set()), + resume: () => Effect.void, + wake: () => Effect.void, + interrupt: () => Effect.void, + }), +) + +const runtime = (data: string) => + AppNodeBuilder.build(LayerNode.group([Database.node, AttachmentStore.node, SessionV2.node, SessionRunnerLLM.node]), [ + [Global.node, Global.layerWith({ data })], + [Database.node, Database.layerFromPath(path.join(data, "session.db"))], + [LayerNodePlatform.llmClient, client], + [PermissionV2.node, permission], + [SessionRunnerModel.node, models], + [Location.node, Location.boundNode({ directory: AbsolutePath.make("/project") })], + [SkillGuidance.node, skills], + [ReferenceGuidance.node, references], + [Snapshot.node, Snapshot.noopLayer], + [SessionExecution.node, execution], + [Config.node, config], + ]) + +const persistedHistory = Effect.gen(function* () { + const { db } = yield* Database.Service + return yield* db + .select({ type: SessionMessageTable.type, data: SessionMessageTable.data }) + .from(SessionMessageTable) + .where(eq(SessionMessageTable.session_id, sessionID)) + .all() + .pipe(Effect.orDie) +}) + +const runFirstProviderTurn = (data: string) => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: sessionID, + directory: "/project", + title: "attachment media", + version: "test", + }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + const store = yield* AttachmentStore.Service + const info = yield* upload(store, "image.png", png) + const session = yield* SessionV2.Service + yield* session.prompt({ + sessionID, + prompt: Prompt.make({ + text: "Inspect the attachment", + files: [FileAttachment.make({ uri: info.uri, mime: info.mime, name: info.name })], + }), + resume: false, + }) + const runner = yield* SessionRunner.Service + const exit = yield* runner.run({ sessionID, force: true }).pipe(Effect.exit) + expect(exit._tag).toBe("Failure") + const history = yield* persistedHistory + expect(JSON.stringify(history)).toContain(info.uri) + expect(JSON.stringify(history)).not.toContain(Buffer.from(png).toString("base64")) + return info + }).pipe(Effect.provide(runtime(data)), Effect.scoped) + +const runReplayProviderTurn = (data: string, info: AttachmentStore.Info) => + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + expect(yield* store.resolve({ sessionID, attachmentID: info.id })).toMatchObject({ + nativeMediaDelivered: true, + }) + const runner = yield* SessionRunner.Service + yield* runner.run({ sessionID, force: true }) + const history = yield* persistedHistory + expect(JSON.stringify(history)).toContain(info.uri) + expect(JSON.stringify(history)).not.toContain(Buffer.from(png).toString("base64")) + }).pipe(Effect.provide(runtime(data)), Effect.scoped) + +const atMostOnceAcrossRestart = Effect.acquireUseRelease( + Effect.promise(tmpdir), + (tmp) => + Effect.gen(function* () { + requests.length = 0 + crash.next = true + const info = yield* runFirstProviderTurn(tmp.path) + yield* runReplayProviderTurn(tmp.path, info) + + expect(requests).toHaveLength(2) + expect(requests[0]?.messages[0]?.content[1]).toMatchObject({ + type: "media", + mediaType: "image/png", + data: png, + }) + expect(requests[1]?.messages[0]?.content[1]).toMatchObject({ type: "text" }) + expect( + requests[1]?.messages[0]?.content[1]?.type === "text" && requests[1].messages[0].content[1].text, + ).toContain('"path":') + }), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), +) + +describe("managed attachment media", () => { + it.live("promotes an image only when the model accepts image input", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const info = yield* upload(store, "image.png", png) + const input = message(info.uri, info.mime, info.name) + const capable = yield* lower(store, model(["text", "image"]), [input]) + const incapable = yield* lower(store, model(["text"]), [input]) + + expect(capable.messages[0]?.content[1]).toMatchObject({ + type: "media", + mediaType: "image/png", + data: png, + }) + expect(capable.messages[0]?.content[1]?.type === "media" && capable.messages[0].content[1].filename).toBe( + (yield* store.resolve({ sessionID, attachmentID: info.id })).path, + ) + expect(incapable.messages[0]?.content[1]).toMatchObject({ type: "text" }) + }), + ), + ) + + it.live("degrades MIME mismatches and unknown content to paths", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const image = yield* upload(store, "image.png", png) + const unknown = yield* upload(store, "notes.bin", new Uint8Array([1, 2, 3])) + const mismatched = yield* lower(store, model(["image"]), [message(image.uri, "image/jpeg", image.name)]) + const opaque = yield* lower(store, model(["image", "pdf"]), [message(unknown.uri, unknown.mime, unknown.name)]) + + expect(mismatched.messages[0]?.content[1]).toMatchObject({ type: "text" }) + expect(opaque.messages[0]?.content[1]).toMatchObject({ type: "text" }) + }), + ), + ) + + it.live("applies image and PDF capabilities independently", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const image = yield* upload(store, "image.png", png) + const document = yield* upload(store, "document.pdf", pdf) + const context = [ + message(image.uri, image.mime, image.name), + message(document.uri, document.mime, document.name), + ] + const imageOnly = yield* lower(store, responsesModel(["image"]), context) + const pdfOnly = yield* lower(store, responsesModel(["pdf"]), context) + const unsafePdf = yield* lower(store, model(["pdf"]), [context[1]!]) + + expect(contentTypes(imageOnly.messages)).toEqual(["media", "text"]) + expect(contentTypes(pdfOnly.messages)).toEqual(["text", "media"]) + expect(pdfOnly.messages[1]?.content[1]).toMatchObject({ type: "media", mediaType: "application/pdf" }) + expect(unsafePdf.messages[0]?.content[1]).toMatchObject({ type: "text" }) + }), + ), + ) + + it.live("degrades media above the provider decoded limit", () => + withStore( + Effect.gen(function* () { + const store = yield* AttachmentStore.Service + const content = new Uint8Array(20 * 1024 * 1024 + 1) + content.set(png) + const info = yield* upload(store, "large.png", content) + const lowered = yield* lower(store, model(["image"]), [message(info.uri, info.mime, info.name)]) + + expect(lowered.messages[0]?.content[1]).toMatchObject({ type: "text" }) + expect(lowered.result.native).toEqual([]) + }), + ), + ) + + it.live("sends native media at most once across a store restart without persisting base64", atMostOnceAcrossRestart) +}) diff --git a/packages/core/test/session-runner-message.test.ts b/packages/core/test/session-runner-message.test.ts index 5798b665a86a..7f00c7af6da0 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -14,6 +14,38 @@ const id = (value: string) => SessionMessage.ID.make(`msg_${value}`) const model = Model.make({ id: "model", provider: "provider", route: OpenAIChat.route }) describe("toLLMMessages", () => { + test("lowers managed attachments to one absolute-path text part", () => { + const uri = "opencode://attachment/att_test" + const message = SessionMessage.User.make({ + id: id("managed"), + type: "user", + text: "Inspect this file", + files: [FileAttachment.make({ uri, mime: "application/octet-stream", name: "report.csv" })], + time: { created }, + }) + + expect( + toLLMMessages( + [message], + model, + new Map([[uri, { type: "path", path: "/managed/session/att_test/report.csv" }]]), + ), + ).toEqual([ + Message.make({ + id: id("managed"), + role: "user", + content: [ + { type: "text", text: "Inspect this file" }, + { + type: "text", + text: 'Attached file: {"name":"report.csv","path":"/managed/session/att_test/report.csv","mime":"application/octet-stream"}', + }, + ], + metadata: {}, + }), + ]) + }) + test("omits empty assistant turns", () => { const assistant = (value: string, content: SessionMessage.Assistant["content"]) => SessionMessage.Assistant.make({ diff --git a/packages/core/test/session-runner-model.test.ts b/packages/core/test/session-runner-model.test.ts index 49bbce95a381..21a69da53450 100644 --- a/packages/core/test/session-runner-model.test.ts +++ b/packages/core/test/session-runner-model.test.ts @@ -22,13 +22,13 @@ type Api = } | { readonly type: "native"; readonly url?: string; readonly settings: Record } -const model = (api: Api, variants: ModelV2.Info["variants"] = []) => +const model = (api: Api, variants: ModelV2.Info["variants"] = [], input: ReadonlyArray = ["text"]) => ModelV2.Info.make({ id: ModelV2.ID.make("test-model"), providerID: ProviderV2.ID.make("test-provider"), name: "Test model", api: { id: ModelV2.ID.make("api-test-model"), ...api }, - capabilities: { tools: true, input: ["text"], output: ["text"] }, + capabilities: { tools: true, input, output: ["text"] }, request: { headers: { "x-test": "header" }, body: { apiKey: "secret", custom_extension: { enabled: true } }, @@ -44,12 +44,19 @@ const model = (api: Api, variants: ModelV2.Info["variants"] = []) => describe("SessionRunnerModel", () => { it.effect("maps catalog OpenAI AI SDK models into native Responses routes", () => Effect.gen(function* () { - const resolved = yield* SessionRunnerModel.fromCatalogModel( - model({ type: "aisdk", package: "@ai-sdk/openai", url: "https://openai.example/v1" }), + const selection = yield* SessionRunnerModel.fromCatalogModel( + model( + { type: "aisdk", package: "@ai-sdk/openai", url: "https://openai.example/v1" }, + [], + ["text", "image", "pdf"], + ), ) - expect(resolved).toMatchObject({ id: "api-test-model", provider: "test-provider" }) - expect(resolved.route).toMatchObject({ + expect(selection).toMatchObject({ + model: { id: "api-test-model", provider: "test-provider" }, + inputCapabilities: ["text", "image", "pdf"], + }) + expect(selection.model.route).toMatchObject({ id: "openai-responses", endpoint: { baseURL: "https://openai.example/v1" }, defaults: { @@ -63,7 +70,7 @@ describe("SessionRunnerModel", () => { it.effect("keeps catalog apiKey credentials out of provider JSON", () => Effect.gen(function* () { - const resolved = yield* SessionRunnerModel.fromCatalogModel( + const { model: resolved } = yield* SessionRunnerModel.fromCatalogModel( model({ type: "aisdk", package: "@ai-sdk/openai", url: "https://openai.example/v1" }), ) const prepared = yield* LLMClient.prepare(LLM.request({ model: resolved, prompt: "Hello" })) @@ -75,7 +82,7 @@ describe("SessionRunnerModel", () => { it.effect("uses merged API settings for OpenAI-compatible auth and request defaults", () => Effect.gen(function* () { - const resolved = yield* SessionRunnerModel.fromCatalogModel( + const { model: resolved } = yield* SessionRunnerModel.fromCatalogModel( ModelV2.Info.make({ ...model({ type: "aisdk", @@ -129,7 +136,7 @@ describe("SessionRunnerModel", () => { location: { directory: AbsolutePath.make("/project") }, }) - const resolved = yield* SessionRunnerModel.resolve(session, catalog) + const { model: resolved } = yield* SessionRunnerModel.resolve(session, catalog) expect(resolved.route.defaults.headers).toMatchObject({ "x-test": "header", "x-variant": "high" }) expect(resolved.route.defaults.http?.body).toEqual({ @@ -165,7 +172,7 @@ describe("SessionRunnerModel", () => { location: { directory: AbsolutePath.make("/project") }, }) - const resolved = yield* SessionRunnerModel.resolve(session, catalog) + const { model: resolved } = yield* SessionRunnerModel.resolve(session, catalog) expect(resolved.route.defaults.http?.body).toEqual({ custom_extension: { enabled: true }, @@ -225,7 +232,7 @@ describe("SessionRunnerModel", () => { location: { directory: AbsolutePath.make("/project") }, }) - const resolved = yield* SessionRunnerModel.resolve(session, catalog) + const { model: resolved } = yield* SessionRunnerModel.resolve(session, catalog) expect(resolved.route.defaults.http?.body).toEqual({ custom_extension: { enabled: true }, @@ -236,7 +243,7 @@ describe("SessionRunnerModel", () => { it.effect("maps catalog Anthropic AI SDK models into native routes", () => Effect.gen(function* () { - const resolved = yield* SessionRunnerModel.fromCatalogModel( + const { model: resolved } = yield* SessionRunnerModel.fromCatalogModel( model({ type: "aisdk", package: "@ai-sdk/anthropic", url: "https://anthropic.example/v1" }), ) @@ -249,7 +256,7 @@ describe("SessionRunnerModel", () => { it.effect("uses resolved credentials for bearer auth", () => Effect.gen(function* () { - const resolved = yield* SessionRunnerModel.fromCatalogModel( + const { model: resolved } = yield* SessionRunnerModel.fromCatalogModel( ModelV2.Info.make({ ...model({ type: "aisdk", package: "@ai-sdk/openai", url: "https://openai.example/v1" }), request: { headers: {}, body: {} }, @@ -272,7 +279,7 @@ describe("SessionRunnerModel", () => { it.effect("prefers stored credentials over configured auth", () => Effect.gen(function* () { const credential = Credential.Key.make({ type: "key", key: "stored-secret", metadata: { tenant: "work" } }) - const resolved = yield* SessionRunnerModel.fromCatalogModel( + const { model: resolved } = yield* SessionRunnerModel.fromCatalogModel( ModelV2.Info.make({ ...model({ type: "aisdk", package: "@ai-sdk/openai", url: "https://openai.example/v1" }), request: { headers: {}, body: { apiKey: "configured-secret" } }, @@ -294,7 +301,7 @@ describe("SessionRunnerModel", () => { it.effect("does not project OAuth account metadata into the request body", () => Effect.gen(function* () { - const resolved = yield* SessionRunnerModel.fromCatalogModel( + const { model: resolved } = yield* SessionRunnerModel.fromCatalogModel( ModelV2.Info.make({ ...model({ type: "aisdk", package: "@ai-sdk/openai", url: "https://openai.example/v1" }), request: { headers: {}, body: {} }, diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index d45cc8c73411..5a6c481a08fb 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -67,7 +67,7 @@ const model = OpenAIChat.route generation: { maxTokens: 20, temperature: 0 }, }) .model({ id: "gpt-4o-mini" }) -const models = SessionRunnerModel.layerWith(() => Effect.succeed(model)) +const models = SessionRunnerModel.layerWith(() => Effect.succeed({ model, inputCapabilities: ["text"] })) const systemContext = AppNodeBuilder.build(SystemContextRegistry.node) const skillGuidance = Layer.mock(SkillGuidance.Service, { load: () => Effect.succeed(SystemContext.empty) }) const referenceGuidance = Layer.mock(ReferenceGuidance.Service, { load: () => Effect.succeed(SystemContext.empty) }) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index cc58b43b2957..bd2050b45ac1 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -155,7 +155,12 @@ const echoNode = makeLocationNode({ name: "test/session-runner-tools", layer: ec let modelResolveHook = Effect.void let currentModel = model const models = SessionRunnerModel.layerWith((session) => - modelResolveHook.pipe(Effect.as(session.model?.id === "replacement" ? replacementModel : currentModel)), + modelResolveHook.pipe( + Effect.as({ + model: session.model?.id === "replacement" ? replacementModel : currentModel, + inputCapabilities: ["text"], + }), + ), ) const systemContextKey = SystemContext.Key.make("test/context") let systemBaseline = "Initial context" diff --git a/packages/llm/src/protocols/anthropic-messages.ts b/packages/llm/src/protocols/anthropic-messages.ts index 1c0dcd32a433..2f3f9f68f83d 100644 --- a/packages/llm/src/protocols/anthropic-messages.ts +++ b/packages/llm/src/protocols/anthropic-messages.ts @@ -55,6 +55,17 @@ const AnthropicImageBlock = Schema.Struct({ }) type AnthropicImageBlock = Schema.Schema.Type +const AnthropicDocumentBlock = Schema.Struct({ + type: Schema.tag("document"), + source: Schema.Struct({ + type: Schema.tag("base64"), + media_type: Schema.Literal("application/pdf"), + data: Schema.String, + }), + cache_control: Schema.optional(AnthropicCacheControl), +}) +type AnthropicDocumentBlock = Schema.Schema.Type + const AnthropicThinkingBlock = Schema.Struct({ type: Schema.tag("thinking"), thinking: Schema.String, @@ -116,7 +127,12 @@ const AnthropicToolResultBlock = Schema.Struct({ cache_control: Schema.optional(AnthropicCacheControl), }) -const AnthropicUserBlock = Schema.Union([AnthropicTextBlock, AnthropicImageBlock, AnthropicToolResultBlock]) +const AnthropicUserBlock = Schema.Union([ + AnthropicTextBlock, + AnthropicImageBlock, + AnthropicDocumentBlock, + AnthropicToolResultBlock, +]) type AnthropicUserBlock = Schema.Schema.Type const AnthropicAssistantBlock = Schema.Union([ AnthropicTextBlock, @@ -320,6 +336,19 @@ const lowerImage = Effect.fn("AnthropicMessages.lowerImage")(function* (part: Me } satisfies AnthropicImageBlock }) +const lowerMedia = Effect.fn("AnthropicMessages.lowerMedia")(function* (part: MediaPart) { + if (part.mediaType.toLowerCase() !== "application/pdf") return yield* lowerImage(part) + const media = yield* ProviderShared.validateMedia( + "Anthropic Messages", + part, + new Set(ProviderShared.PDF_MIMES), + ) + return { + type: "document" as const, + source: { type: "base64" as const, media_type: "application/pdf" as const, data: media.base64 }, + } satisfies AnthropicDocumentBlock +}) + // Tool results may carry structured text/images. Keep media as provider-native // content instead of JSON-stringifying base64 into a prompt string. const lowerToolResultContentItem = Effect.fn("AnthropicMessages.lowerToolResultContentItem")(function* ( @@ -430,7 +459,7 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* ( continue } if (part.type === "media") { - content.push(yield* lowerImage(part)) + content.push(yield* lowerMedia(part)) continue } return yield* ProviderShared.unsupportedContent("Anthropic Messages", "user", ["text", "media"]) @@ -831,6 +860,7 @@ const step = (state: ParserState, event: AnthropicEvent) => { */ export const protocol = Protocol.make({ id: ADAPTER, + media: ProviderShared.mediaAdmission({ image: ProviderShared.IMAGE_MIMES, pdf: ProviderShared.PDF_MIMES }), body: { schema: AnthropicMessagesBody, from: fromRequest, diff --git a/packages/llm/src/protocols/gemini.ts b/packages/llm/src/protocols/gemini.ts index c4bb9476a49d..ac5d4b00b4cf 100644 --- a/packages/llm/src/protocols/gemini.ts +++ b/packages/llm/src/protocols/gemini.ts @@ -23,7 +23,7 @@ import { Lifecycle } from "./utils/lifecycle" import { ToolSchemaProjection } from "./utils/tool-schema" const ADAPTER = "gemini" -const MEDIA_MIMES = new Set(ProviderShared.MEDIA_MIMES) +const MEDIA_MIMES = new Set([...ProviderShared.MEDIA_MIMES, ...ProviderShared.PDF_MIMES]) export const DEFAULT_BASE_URL = "https://generativelanguage.googleapis.com/v1beta" // ============================================================================= @@ -485,6 +485,7 @@ const step = (state: ParserState, event: GeminiEvent) => { */ export const protocol = Protocol.make({ id: ADAPTER, + media: ProviderShared.mediaAdmission({ image: ProviderShared.IMAGE_MIMES, pdf: ProviderShared.PDF_MIMES }), body: { schema: GeminiBody, from: fromRequest, diff --git a/packages/llm/src/protocols/openai-chat.ts b/packages/llm/src/protocols/openai-chat.ts index 9ac85b07b139..1a89c4681003 100644 --- a/packages/llm/src/protocols/openai-chat.ts +++ b/packages/llm/src/protocols/openai-chat.ts @@ -480,6 +480,7 @@ const finishEvents = (state: ParserState): ReadonlyArray => { */ export const protocol = Protocol.make({ id: ADAPTER, + media: ProviderShared.mediaAdmission({ image: ProviderShared.IMAGE_MIMES }), body: { schema: OpenAIChatBody, from: fromRequest, diff --git a/packages/llm/src/protocols/openai-responses.ts b/packages/llm/src/protocols/openai-responses.ts index 4936d31c921b..be85b1178129 100644 --- a/packages/llm/src/protocols/openai-responses.ts +++ b/packages/llm/src/protocols/openai-responses.ts @@ -40,7 +40,16 @@ const OpenAIResponsesInputImage = Schema.Struct({ type: Schema.tag("input_image"), image_url: Schema.String, }) -const OpenAIResponsesInputContent = Schema.Union([OpenAIResponsesInputText, OpenAIResponsesInputImage]) +const OpenAIResponsesInputFile = Schema.Struct({ + type: Schema.tag("input_file"), + filename: Schema.String, + file_data: Schema.String, +}) +const OpenAIResponsesInputContent = Schema.Union([ + OpenAIResponsesInputText, + OpenAIResponsesInputImage, + OpenAIResponsesInputFile, +]) type OpenAIResponsesInputContent = Schema.Schema.Type const OpenAIResponsesOutputText = Schema.Struct({ @@ -310,6 +319,18 @@ const lowerUserContent = Effect.fn("OpenAIResponses.lowerUserContent")(function* ) { if (part.type === "text") return { type: "input_text" as const, text: part.text } if (part.type === "media") { + if (part.mediaType.toLowerCase() === "application/pdf") { + const media = yield* ProviderShared.validateMedia( + "OpenAI Responses", + part, + new Set(ProviderShared.PDF_MIMES), + ) + return { + type: "input_file" as const, + filename: part.filename ?? "attachment.pdf", + file_data: media.dataUrl, + } + } const media = yield* ProviderShared.validateMedia( "OpenAI Responses", part, @@ -958,6 +979,7 @@ const step = (state: ParserState, event: OpenAIResponsesEvent) => { */ export const protocol = Protocol.make({ id: ADAPTER, + media: ProviderShared.mediaAdmission({ image: ProviderShared.IMAGE_MIMES, pdf: ProviderShared.PDF_MIMES }), body: { schema: OpenAIResponsesBody, from: fromRequest, diff --git a/packages/llm/src/protocols/shared.ts b/packages/llm/src/protocols/shared.ts index 173dc511bb03..ebdefda78fcb 100644 --- a/packages/llm/src/protocols/shared.ts +++ b/packages/llm/src/protocols/shared.ts @@ -13,6 +13,7 @@ import { type TextPart, type ToolResultPart, } from "../schema" +import type { MediaAdmissionQuery, MediaInputCapability } from "../route/protocol" import { isRecord } from "../utils/record" export { isRecord } @@ -156,12 +157,28 @@ export const parseToolInput = (route: string, name: string, raw: string) => parseJson(route, raw || "{}", `Invalid JSON input for ${route} tool call ${name}`) export const IMAGE_MIMES = ["image/png", "image/jpeg", "image/gif", "image/webp"] as const +export const PDF_MIMES = ["application/pdf"] as const export const VIDEO_MIMES = ["video/mp4", "video/webm", "video/quicktime"] as const export const AUDIO_MIMES = ["audio/wav", "audio/mp3", "audio/aiff", "audio/aac", "audio/ogg", "audio/flac"] as const export const MEDIA_MIMES = [...IMAGE_MIMES, ...VIDEO_MIMES, ...AUDIO_MIMES] as const export const MAX_MEDIA_ENCODED_BYTES = 28 * 1024 * 1024 export const MAX_MEDIA_DECODED_BYTES = 20 * 1024 * 1024 +export const mediaAdmission = (input: { + readonly image?: ReadonlyArray + readonly pdf?: ReadonlyArray +}): MediaAdmissionQuery => { + const accepted = new Map([ + ...(input.image ?? []).map((mime) => [mime, "image"] as const), + ...(input.pdf ?? []).map((mime) => [mime, "pdf"] as const), + ]) + return (media) => { + if (media.bytes > MAX_MEDIA_DECODED_BYTES) return undefined + const capability = accepted.get(media.mime.toLowerCase()) + return capability ? { capability } : undefined + } +} + const base64Pattern = /^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/ export interface ValidatedMedia { diff --git a/packages/llm/src/route/client.ts b/packages/llm/src/route/client.ts index d3b41f5817f1..cf2c2772b3c6 100644 --- a/packages/llm/src/route/client.ts +++ b/packages/llm/src/route/client.ts @@ -7,7 +7,7 @@ import type { Framing } from "./framing" import { HttpTransport } from "./transport" import type { Transport, TransportRuntime } from "./transport" import { WebSocketExecutor } from "./transport" -import type { Protocol } from "./protocol" +import type { MediaAdmissionQuery, Protocol } from "./protocol" import { applyCachePolicy } from "../cache-policy" import * as ProviderShared from "../protocols/shared" import type { LLMError, LLMEvent, PreparedRequestOf, ProtocolID, ProviderOptions } from "../schema" @@ -42,6 +42,7 @@ export interface Route { readonly transport: Transport readonly defaults: RouteDefaults readonly body: RouteBody + readonly media: MediaAdmissionQuery readonly with: (patch: RoutePatch) => Route readonly model: (input: RouteMappedModelInput) => Model readonly prepareTransport: (body: Body, request: LLMRequest) => Effect.Effect @@ -254,6 +255,7 @@ function makeFromTransport( transport: routeInput.transport, defaults: routeInput.defaults ?? {}, body: protocol.body, + media: protocol.media ?? (() => undefined), with: (patch: RoutePatch) => { const { id, provider, auth, transport, endpoint, ...defaults } = patch return build({ diff --git a/packages/llm/src/route/protocol.ts b/packages/llm/src/route/protocol.ts index acb1e78c67bb..4ec8fbd6c5c0 100644 --- a/packages/llm/src/route/protocol.ts +++ b/packages/llm/src/route/protocol.ts @@ -38,10 +38,25 @@ export interface Protocol { readonly id: ProtocolID /** Request side: schema for the provider-native body and how to build it. */ readonly body: ProtocolBody + /** Whether this protocol can safely accept one bounded native media input. */ + readonly media?: MediaAdmissionQuery /** Response side: streaming state machine. */ readonly stream: ProtocolStream } +export type MediaInputCapability = "image" | "pdf" + +export interface MediaAdmissionInput { + readonly mime: string + readonly bytes: number +} + +export interface MediaAdmission { + readonly capability: MediaInputCapability +} + +export type MediaAdmissionQuery = (input: MediaAdmissionInput) => MediaAdmission | undefined + export interface ProtocolBody { /** Schema for the validated provider-native body sent as the JSON request. */ readonly schema: Schema.Codec diff --git a/packages/llm/test/provider/anthropic-messages.test.ts b/packages/llm/test/provider/anthropic-messages.test.ts index 898931295849..eaac021c8ed3 100644 --- a/packages/llm/test/provider/anthropic-messages.test.ts +++ b/packages/llm/test/provider/anthropic-messages.test.ts @@ -299,6 +299,31 @@ describe("Anthropic Messages route", () => { }), ) + it.effect("lowers PDF user content as a document block", () => + Effect.gen(function* () { + const prepared = yield* LLMClient.prepare( + LLM.request({ + id: "req_pdf", + model, + messages: [Message.user({ type: "media", mediaType: "application/pdf", data: "JVBERi0=" })], + cache: "none", + }), + ) + + expect(prepared.body.messages).toEqual([ + { + role: "user", + content: [ + { + type: "document", + source: { type: "base64", media_type: "application/pdf", data: "JVBERi0=" }, + }, + ], + }, + ]) + }), + ) + it.effect("prepares the composed native continuation request", () => Effect.gen(function* () { const prepared = yield* LLMClient.prepare( diff --git a/packages/llm/test/provider/gemini.test.ts b/packages/llm/test/provider/gemini.test.ts index 1dc253c0ea88..02c104cc251a 100644 --- a/packages/llm/test/provider/gemini.test.ts +++ b/packages/llm/test/provider/gemini.test.ts @@ -182,6 +182,24 @@ describe("Gemini route", () => { }), ) + it.effect("lowers PDF user content as inline data", () => + Effect.gen(function* () { + const prepared = yield* LLMClient.prepare( + LLM.request({ + model, + messages: [Message.user({ type: "media", mediaType: "application/pdf", data: "JVBERi0=" })], + }), + ) + + expect(prepared.body.contents).toEqual([ + { + role: "user", + parts: [{ inlineData: { mimeType: "application/pdf", data: "JVBERi0=" } }], + }, + ]) + }), + ) + for (const [name, media] of [ ["mismatched data URL MIME", { mediaType: "image/png", data: "data:image/jpeg;base64,/9j/" }], ["malformed base64", { mediaType: "image/png", data: "%%%=" }], diff --git a/packages/llm/test/provider/openai-responses.test.ts b/packages/llm/test/provider/openai-responses.test.ts index cd8bad51af47..955845679195 100644 --- a/packages/llm/test/provider/openai-responses.test.ts +++ b/packages/llm/test/provider/openai-responses.test.ts @@ -1315,17 +1315,35 @@ describe("OpenAI Responses route", () => { }), ) - it.effect("rejects unsupported user media content", () => + it.effect("lowers PDF user content as an input file", () => Effect.gen(function* () { - const error = yield* LLMClient.prepare( + const prepared = yield* LLMClient.prepare( LLM.request({ id: "req_media", model, - messages: [Message.user({ type: "media", mediaType: "application/pdf", data: "AAECAw==" })], + messages: [ + Message.user({ + type: "media", + mediaType: "application/pdf", + data: "JVBERi0=", + filename: "/managed/document.pdf", + }), + ], }), - ).pipe(Effect.flip) + ) - expect(error.message).toContain("OpenAI Responses does not support media type application/pdf") + expect(prepared.body.input).toEqual([ + { + role: "user", + content: [ + { + type: "input_file", + filename: "/managed/document.pdf", + file_data: "data:application/pdf;base64,JVBERi0=", + }, + ], + }, + ]) }), ) diff --git a/packages/opencode/src/server/routes/instance/httpapi/server.ts b/packages/opencode/src/server/routes/instance/httpapi/server.ts index fb9d2db65621..7633511b0b56 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/server.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/server.ts @@ -65,6 +65,7 @@ import { Ripgrep } from "@opencode-ai/core/ripgrep" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionV2 } from "@opencode-ai/core/session" import { SessionExecution } from "@opencode-ai/core/session/execution" +import { AttachmentStore } from "@opencode-ai/core/attachment-store" import * as SessionExecutionLocal from "@opencode-ai/core/session/execution/local" import { lazy } from "@/util/lazy" import { CorsConfig, isAllowedCorsOrigin, type CorsOptions } from "@opencode-ai/server/cors" @@ -213,6 +214,8 @@ const app = LayerNode.group([ Npm.node, FSUtil.node, Database.node, + AttachmentStore.node, + AttachmentStore.cleanupNode, Auth.node, Account.node, Config.node, diff --git a/packages/opencode/test/server/httpapi-v2-attachment.test.ts b/packages/opencode/test/server/httpapi-v2-attachment.test.ts new file mode 100644 index 000000000000..57865fb71fff --- /dev/null +++ b/packages/opencode/test/server/httpapi-v2-attachment.test.ts @@ -0,0 +1,202 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { Global } from "@opencode-ai/core/global" +import { Attachment } from "@opencode-ai/schema/attachment" +import { Session } from "@opencode-ai/schema/session" +import { Context, Schema } from "effect" +import fs from "fs/promises" +import path from "path" +import { HttpApiApp } from "../../src/server/routes/instance/httpapi/server" +import { resetDatabase } from "../fixture/db" +import { disposeAllInstances, tmpdir } from "../fixture/fixture" + +// SAFETY: The generated test handler accepts an erased runtime context and requires no contextual services here. +const context = Context.empty() as Context.Context +const SessionResponse = Schema.Struct({ data: Schema.Struct({ id: Session.ID }) }) +const AttachmentResponse = Schema.Struct({ data: Attachment.Info }) + +function request(route: string, directory: string, init: RequestInit = {}) { + const headers = new Headers(init.headers) + headers.set("x-opencode-directory", directory) + return HttpApiApp.webHandler().handler( + new Request(`http://localhost${route}`, { + ...init, + headers, + }), + context, + ) +} + +async function createSession(directory: string) { + const response = await request("/api/session", directory, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ location: { directory } }), + }) + expect(response.status).toBe(200) + return Schema.decodeUnknownSync(SessionResponse)(await response.json()) +} + +function form(bytes: BlobPart[], name: string, type = "application/octet-stream") { + const data = new FormData() + data.append("file", new File(bytes, name, { type })) + return data +} + +function streamForm(chunks: Uint8Array[], name: string, abort = false) { + const boundary = "opencode-attachment-test" + const encoder = new TextEncoder() + const parts = [ + encoder.encode( + `--${boundary}\r\nContent-Disposition: form-data; name="file"; filename="${name}"\r\nContent-Type: application/octet-stream\r\n\r\n`, + ), + ...chunks, + ...(abort ? [] : [encoder.encode(`\r\n--${boundary}--\r\n`)]), + ] + const state = { index: 0 } + return { + headers: { "content-type": `multipart/form-data; boundary=${boundary}` }, + body: new ReadableStream({ + pull(controller) { + const part = parts[state.index] + if (part) { + state.index += 1 + controller.enqueue(part) + return + } + if (abort) { + controller.error(new Error("client aborted upload")) + return + } + controller.close() + }, + }), + } +} + +afterEach(async () => { + await disposeAllInstances() + await resetDatabase() +}) + +describe("v2 attachment HttpApi", () => { + test("uploads, admits, and isolates a managed attachment", async () => { + await using tmp = await tmpdir({ git: true }) + const first = await createSession(tmp.path) + const second = await createSession(tmp.path) + const uploaded = await request(`/api/session/${first.data.id}/attachment`, tmp.path, { + method: "POST", + body: form([new Uint8Array([1, 2, 3])], "report.bin"), + }) + expect(uploaded.status).toBe(200) + const attachment = Schema.decodeUnknownSync(AttachmentResponse)(await uploaded.json()) + expect(attachment.data).toMatchObject({ + uri: `opencode://attachment/${attachment.data.id}`, + name: "report.bin", + mime: "application/octet-stream", + size: 3, + }) + const admitted = await request(`/api/session/${first.data.id}/prompt`, tmp.path, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ prompt: { text: "Inspect it", files: [{ uri: attachment.data.uri }] }, resume: false }), + }) + expect(admitted.status).toBe(200) + expect(await admitted.json()).toMatchObject({ + data: { + prompt: { + files: [{ uri: attachment.data.uri, name: "report.bin", mime: "application/octet-stream" }], + }, + }, + }) + + const rejected = await request(`/api/session/${second.data.id}/prompt`, tmp.path, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ prompt: { text: "Inspect it", files: [{ uri: attachment.data.uri }] }, resume: false }), + }) + expect(rejected.status).toBe(404) + expect(await rejected.json()).toMatchObject({ + _tag: "AttachmentNotFoundError", + sessionID: second.data.id, + attachmentID: attachment.data.id, + }) + }) + + test("returns a typed 413 and removes the partial upload", async () => { + await using tmp = await tmpdir({ git: true }) + const session = await createSession(tmp.path) + const directory = path.join(Global.Path.data, "attachments", encodeURIComponent(session.data.id)) + const response = await request(`/api/session/${session.data.id}/attachment`, tmp.path, { + method: "POST", + body: form([new Uint8Array(25 * 1024 * 1024 + 1)], "oversized.bin"), + }) + + expect(response.status).toBe(413) + expect(await response.json()).toMatchObject({ + _tag: "PayloadTooLargeError", + scope: "file", + maximumBytes: 25 * 1024 * 1024, + }) + expect(await fs.readdir(directory).catch(() => [])).toEqual([]) + }) + + test("streams a chunked multipart upload without Content-Length", async () => { + await using tmp = await tmpdir({ git: true }) + const session = await createSession(tmp.path) + const input = streamForm([new Uint8Array([1, 2]), new Uint8Array([3, 4, 5])], "chunked.bin") + expect(new Headers(input.headers).has("content-length")).toBe(false) + const response = await request(`/api/session/${session.data.id}/attachment`, tmp.path, { + method: "POST", + ...input, + }) + + expect(response.status).toBe(200) + expect(Schema.decodeUnknownSync(AttachmentResponse)(await response.json()).data).toMatchObject({ + name: "chunked.bin", + size: 5, + }) + }) + + test("removes a partial attachment when the request body aborts", async () => { + await using tmp = await tmpdir({ git: true }) + const session = await createSession(tmp.path) + const directory = path.join(Global.Path.data, "attachments", encodeURIComponent(session.data.id)) + const response = await request(`/api/session/${session.data.id}/attachment`, tmp.path, { + method: "POST", + ...streamForm([new Uint8Array([1, 2, 3])], "aborted.bin", true), + }).catch(() => undefined) + + expect(response?.status).not.toBe(200) + expect(await fs.readdir(directory).catch(() => [])).toEqual([]) + }) + + test("rejects non-canonical forms of the managed attachment scheme", async () => { + await using tmp = await tmpdir({ git: true }) + const session = await createSession(tmp.path) + const uploaded = await request(`/api/session/${session.data.id}/attachment`, tmp.path, { + method: "POST", + body: form([new Uint8Array([1])], "private.bin"), + }) + const attachment = Schema.decodeUnknownSync(AttachmentResponse)(await uploaded.json()).data + const responses = await Promise.all( + [ + attachment.uri.replace("opencode", "OPENCODE"), + attachment.uri.replace("attachment", "ATTACHMENT"), + `${attachment.uri}/extra`, + ].map((uri) => + request(`/api/session/${session.data.id}/prompt`, tmp.path, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ prompt: { text: "Inspect it", files: [{ uri }] }, resume: false }), + }), + ), + ) + + expect(responses.map((response) => response.status)).toEqual([404, 404, 404]) + expect(await Promise.all(responses.map((response) => response.json()))).toEqual( + expect.arrayContaining([ + expect.objectContaining({ _tag: "AttachmentNotFoundError", sessionID: session.data.id }), + ]), + ) + }) +}) diff --git a/packages/protocol/src/errors.ts b/packages/protocol/src/errors.ts index 3b1eced63a2c..b82c3f44a1e3 100644 --- a/packages/protocol/src/errors.ts +++ b/packages/protocol/src/errors.ts @@ -1,4 +1,7 @@ import { Schema } from "effect" +import { NonNegativeInt } from "@opencode-ai/schema/schema" +import { Attachment } from "@opencode-ai/schema/attachment" +import { Session } from "@opencode-ai/schema/session" export class InvalidRequestError extends Schema.TaggedErrorClass()( "InvalidRequestError", @@ -25,6 +28,26 @@ export class ConflictError extends Schema.TaggedErrorClass()( { httpApiStatus: 409 }, ) {} +export class PayloadTooLargeError extends Schema.TaggedErrorClass()( + "PayloadTooLargeError", + { + message: Schema.String, + scope: Schema.Literals(["file", "session", "global"]), + maximumBytes: NonNegativeInt, + }, + { httpApiStatus: 413 }, +) {} + +export class AttachmentNotFoundError extends Schema.TaggedErrorClass()( + "AttachmentNotFoundError", + { + sessionID: Session.ID, + attachmentID: Attachment.ID.pipe(Schema.optional), + message: Schema.String, + }, + { httpApiStatus: 404 }, +) {} + export class ServiceUnavailableError extends Schema.TaggedErrorClass()( "ServiceUnavailableError", { diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 8ce85ef79686..4c217bbea2c0 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -8,10 +8,12 @@ import { Workspace } from "@opencode-ai/schema/workspace" import { Context, Effect, Encoding, Result, Schema, Struct } from "effect" import { HttpApiEndpoint, HttpApiGroup, HttpApiMiddleware, HttpApiSchema, OpenApi } from "effect/unstable/httpapi" import { + AttachmentNotFoundError, ConflictError, InvalidCursorError, InvalidRequestError, MessageNotFoundError, + PayloadTooLargeError, ServiceUnavailableError, SessionNotFoundError, UnknownError, @@ -21,6 +23,7 @@ import { Model } from "@opencode-ai/schema/model" import { Location } from "@opencode-ai/schema/location" import { Revert } from "@opencode-ai/schema/revert" import { SessionEvent } from "@opencode-ai/schema/session-event" +import { Attachment } from "@opencode-ai/schema/attachment" const SessionsQueryFields = { workspace: Workspace.ID.pipe(Schema.optional), @@ -201,6 +204,24 @@ export const makeSessionGroup = (sessionLo }), ), ) + .add( + HttpApiEndpoint.post("session.attachment", "/api/session/:sessionID/attachment", { + params: { sessionID: Session.ID }, + payload: Schema.Struct({ file: Schema.Unknown }).pipe( + HttpApiSchema.asMultipartStream({ maxParts: 1, maxFileSize: Attachment.MAX_FILE_BYTES }), + ), + success: Schema.Struct({ data: Attachment.Info }), + error: [InvalidRequestError, PayloadTooLargeError, SessionNotFoundError, UnknownError], + }) + .middleware(sessionLocationMiddleware) + .annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.attachment", + summary: "Upload session attachment", + description: "Stream one file into managed storage for a later Session prompt.", + }), + ), + ) .add( HttpApiEndpoint.post("session.prompt", "/api/session/:sessionID/prompt", { params: { sessionID: Session.ID }, @@ -211,7 +232,7 @@ export const makeSessionGroup = (sessionLo resume: Schema.Boolean.pipe(Schema.optional), }), success: Schema.Struct({ data: SessionInput.Admitted }), - error: [ConflictError, SessionNotFoundError], + error: [AttachmentNotFoundError, ConflictError, SessionNotFoundError, UnknownError], }) .middleware(sessionLocationMiddleware) .annotateMerge( diff --git a/packages/schema/src/attachment.ts b/packages/schema/src/attachment.ts new file mode 100644 index 000000000000..c7945bfa4586 --- /dev/null +++ b/packages/schema/src/attachment.ts @@ -0,0 +1,30 @@ +export * as Attachment from "./attachment" + +import { Schema } from "effect" +import { ascending } from "./identifier" +import { NonNegativeInt, statics } from "./schema" + +export const MAX_FILE_BYTES = 25 * 1024 * 1024 + +export const ID = Schema.String.check(Schema.isPattern(/^att_[0-9A-Za-z]+$/)).pipe( + Schema.brand("Attachment.ID"), + statics((schema) => ({ create: () => schema.make("att_" + ascending()) })), +) +export type ID = typeof ID.Type + +export const URI = Schema.String.check(Schema.isPattern(/^opencode:\/\/attachment\/att_[0-9A-Za-z]+$/)).pipe( + Schema.brand("Attachment.URI"), + statics((schema) => ({ + fromID: (id: ID) => schema.make(`opencode://attachment/${id}`), + })), +) +export type URI = typeof URI.Type + +export interface Info extends Schema.Schema.Type {} +export const Info = Schema.Struct({ + id: ID, + uri: URI, + name: Schema.String, + mime: Schema.String, + size: NonNegativeInt, +}).annotate({ identifier: "Attachment.Info" }) diff --git a/packages/schema/src/index.ts b/packages/schema/src/index.ts index b7c8e5110f73..c51f047c4100 100644 --- a/packages/schema/src/index.ts +++ b/packages/schema/src/index.ts @@ -1,4 +1,5 @@ export { Agent } from "./agent" +export { Attachment } from "./attachment" export { Command } from "./command" export { Connection } from "./connection" export { Credential } from "./credential" diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index 5b7d354b04fc..969459216080 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -1,24 +1,106 @@ import { SessionV2 } from "@opencode-ai/core/session" import { DateTime, Effect, Stream } from "effect" +import { Multipart } from "effect/unstable/http" import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi" import { Api } from "../api" import { SessionsCursor } from "@opencode-ai/protocol/groups/session" import { + AttachmentNotFoundError, ConflictError, InvalidCursorError, + InvalidRequestError, MessageNotFoundError, + PayloadTooLargeError, ServiceUnavailableError, SessionNotFoundError, UnknownError, } from "@opencode-ai/protocol/errors" import { AbsolutePath } from "@opencode-ai/core/schema" +import { AttachmentStore } from "@opencode-ai/core/attachment-store" const DefaultSessionsLimit = 50 const DefaultSessionHistoryLimit = 50 +interface UploadState { + value?: AttachmentStore.Info +} + +const uploadError = ( + error: InvalidRequestError | AttachmentStore.UploadError | Multipart.MultipartError, +): InvalidRequestError | PayloadTooLargeError | UnknownError => { + if (error._tag === "InvalidRequestError") return error + if (error._tag === "AttachmentStore.QuotaError") + return new PayloadTooLargeError({ + message: `Attachment exceeds the ${error.scope} storage limit`, + scope: error.scope, + maximumBytes: error.maximumBytes, + }) + if (error._tag === "AttachmentStore.FilenameError") + return new InvalidRequestError({ message: "Attachment filename contains a NUL byte", field: "file" }) + if (error._tag === "AttachmentStore.StorageError") return new UnknownError({ message: "Failed to store attachment" }) + if (error.reason._tag !== "FileTooLarge" && error.reason._tag !== "BodyTooLarge") + return new InvalidRequestError({ message: "Invalid multipart attachment", field: "file" }) + return new PayloadTooLargeError({ + message: "Attachment exceeds the file storage limit", + scope: "file", + maximumBytes: AttachmentStore.MAX_FILE_BYTES, + }) +} + +const uploadAttachment = Effect.fn("SessionHandler.uploadAttachment")(function* ( + attachments: AttachmentStore.Interface, + sessionID: SessionV2.ID, + parts: Stream.Stream, +) { + const uploaded: UploadState = {} + function save(value: AttachmentStore.Info) { + uploaded.value = value + return Effect.void + } + return yield* Effect.gen(function* () { + yield* Stream.runForEach( + parts, + (part): Effect.Effect => { + if (!Multipart.isFile(part) || part.key !== "file" || uploaded.value) + return Effect.fail(new InvalidRequestError({ message: "Expected one multipart file field", field: "file" })) + return attachments + .upload({ + sessionID, + name: part.name, + contentType: part.contentType, + content: part.content, + }) + .pipe(Effect.tap(save), Effect.asVoid) + }, + ) + if (!uploaded.value) + return yield* new InvalidRequestError({ message: "Expected one multipart file field", field: "file" }) + return { data: uploaded.value } + }).pipe( + Effect.tapError(() => + uploaded.value + ? attachments.remove({ sessionID, attachmentID: uploaded.value.id }).pipe(Effect.catch(() => Effect.void)) + : Effect.void, + ), + Effect.mapError(uploadError), + ) +}) + +const attachmentNotFound = (error: AttachmentStore.ReferenceError) => + Effect.fail( + new AttachmentNotFoundError({ + sessionID: error.sessionID, + attachmentID: error.attachmentID, + message: "Attachment not found for this session", + }), + ) + +const attachmentStorageError = () => new UnknownError({ message: "Failed to bind attachment" }) + export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handlers) => Effect.gen(function* () { const session = yield* SessionV2.Service + const attachments = yield* AttachmentStore.Service return handlers .handle( @@ -136,6 +218,12 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl return HttpApiSchema.NoContent.make() }), ) + .handle( + "session.attachment", + Effect.fn(function* (ctx) { + return yield* uploadAttachment(attachments, ctx.params.sessionID, ctx.payload) + }), + ) .handle( "session.prompt", Effect.fn(function* (ctx) { @@ -165,6 +253,8 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl }), ), ), + Effect.catchTag("AttachmentStore.ReferenceError", attachmentNotFound), + Effect.catchTag("AttachmentStore.StorageError", attachmentStorageError), ), } }), diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index cc1b1ae6a55d..2fba30637706 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -11,6 +11,7 @@ import { SessionExecution } from "@opencode-ai/core/session/execution" import { LocationServiceMap } from "@opencode-ai/core/location-service-map" import { SessionExecutionLocal } from "@opencode-ai/core/session/execution/local" import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" +import { AttachmentStore } from "@opencode-ai/core/attachment-store" import { HttpRouter, HttpServer } from "effect/unstable/http" import { HttpApiBuilder } from "effect/unstable/httpapi" import { Layer, Option } from "effect" @@ -28,6 +29,8 @@ const applicationServices = LayerNode.group([ EventV2.node, httpClient, ToolOutputStore.cleanupNode, + AttachmentStore.node, + AttachmentStore.cleanupNode, SessionV2.node, PermissionSaved.node, PtyTicket.node,