From 7dc3901de8c7b5a6770140dc9e5072a65a4213c0 Mon Sep 17 00:00:00 2001 From: Safzan Pirani Date: Sat, 29 Aug 2026 22:39:31 +0530 Subject: [PATCH] feat(core): materialize managed attachments --- packages/client/src/contract.ts | 2 +- packages/client/src/generated/client.ts | 2 +- packages/client/src/generated/types.ts | 25 +- packages/core/src/attachment-store.ts | 615 ++++++++++++++++++ packages/core/src/session.ts | 62 +- packages/core/src/session/runner/index.ts | 2 + packages/core/src/session/runner/llm.ts | 28 +- .../core/src/session/runner/to-llm-message.ts | 27 +- packages/core/test/attachment-store.test.ts | 291 +++++++++ .../core/test/session-runner-message.test.ts | 26 + .../server/routes/instance/httpapi/server.ts | 3 + .../test/server/httpapi-v2-attachment.test.ts | 202 ++++++ packages/protocol/src/errors.ts | 23 + packages/protocol/src/groups/session.ts | 23 +- packages/schema/src/attachment.ts | 30 + packages/schema/src/index.ts | 1 + packages/server/src/handlers/session.ts | 90 +++ packages/server/src/routes.ts | 3 + 18 files changed, 1427 insertions(+), 28 deletions(-) create mode 100644 packages/core/src/attachment-store.ts create mode 100644 packages/core/test/attachment-store.test.ts create mode 100644 packages/opencode/test/server/httpapi-v2-attachment.test.ts create mode 100644 packages/schema/src/attachment.ts 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..a49feb1a181e --- /dev/null +++ b/packages/core/src/attachment-store.ts @@ -0,0 +1,615 @@ +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), +}) +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 +} + +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 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, + } + }) + + 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 bind: Interface["bind"] = (input) => + locks.withLock(input.sessionID)( + Effect.gen(function* () { + 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 })), + ) + if (metadata.boundMessageID) return resolved + yield* fs + .writeFileString( + path.join(directory, metadataUploadName), + JSON.stringify({ ...metadata, boundMessageID: input.messageID }, 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 + }), + ) + + 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, 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/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..4cd854ba5870 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -29,6 +29,7 @@ import { SessionCompaction } from "../compaction" import { SessionEvent } from "../event" import { SessionHistory } from "../history" import { SessionInput } from "../input" +import { SessionMessage } from "../message" import { SessionSchema } from "../schema" import { SessionStore } from "../store" import { type RunError, Service } from "./index" @@ -39,6 +40,25 @@ 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" + +const resolveAttachmentPaths = Effect.fn("SessionRunner.resolveAttachmentPaths")(function* ( + attachments: AttachmentStore.Interface, + sessionID: SessionSchema.ID, + context: readonly SessionMessage.Message[], +) { + const managed = context.flatMap((message) => + message.type === "user" ? (message.files ?? []).filter((file) => AttachmentStore.isManagedURI(file.uri)) : [], + ) + const resolved = yield* Effect.forEach(managed, (file) => { + const attachmentID = AttachmentStore.attachmentID(file.uri) + if (!attachmentID) return Effect.fail(new AttachmentStore.ReferenceError({ sessionID })) + return attachments + .resolve({ sessionID, attachmentID }) + .pipe(Effect.map((attachment) => [file.uri, attachment.path] satisfies readonly [string, string])) + }) + return new Map(resolved) +}) /** * Runs one durable coding-agent Session until it settles. @@ -105,6 +125,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) { @@ -199,6 +220,7 @@ const layer = Layer.effect( const model = yield* models.resolve(session) const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) const context = entries.map((entry) => entry.message) + const attachmentPaths = yield* resolveAttachmentPaths(attachments, session.id, 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 +237,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, attachmentPaths), + ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : []), + ], tools: toolMaterialization?.definitions ?? [], toolChoice: isLastStep ? "none" : undefined, }) @@ -435,5 +460,6 @@ export const node = makeLocationNode({ Config.node, Snapshot.node, Database.node, + AttachmentStore.node, ], }) diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index b2b1af5d30f1..a83725ca51f2 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -18,6 +18,15 @@ const media = (file: FileAttachment): ContentPart => ({ metadata: file.description === undefined ? undefined : { description: file.description }, }) +const userFile = (file: FileAttachment, paths: ReadonlyMap): ContentPart => { + const path = paths.get(file.uri) + if (!path) return media(file) + return { + type: "text", + text: `Attached file: ${JSON.stringify({ name: file.name, path, mime: file.mime })}`, + } +} + const toolInput = (tool: SessionMessage.AssistantTool) => { if (tool.state.status !== "pending") return tool.state.input try { @@ -112,7 +121,11 @@ const assistant = (message: SessionMessage.Assistant, model: Model) => { ] } -function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] { +function toLLMMessage( + message: SessionMessage.Message, + model: Model, + attachmentPaths: ReadonlyMap, +): Message[] { switch (message.type) { case "agent-switched": case "model-switched": @@ -122,7 +135,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, attachmentPaths)), + ], metadata: { ...message.metadata, ...(message.agents?.length ? { agents: message.agents } : {}), @@ -167,5 +183,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, + attachmentPaths: ReadonlyMap = new Map(), +) => messages.flatMap((message) => toLLMMessage(message, model, attachmentPaths)) diff --git a/packages/core/test/attachment-store.test.ts b/packages/core/test/attachment-store.test.ts new file mode 100644 index 000000000000..776a806106cd --- /dev/null +++ b/packages/core/test/attachment-store.test.ts @@ -0,0 +1,291 @@ +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 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("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-message.test.ts b/packages/core/test/session-runner-message.test.ts index 5798b665a86a..f4be2e27016a 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -14,6 +14,32 @@ 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, "/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/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,