diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index cb45acafb897..a582a0618610 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -234,6 +234,8 @@ import type { ExperimentalPersistentPtyRemoveOutput, ExperimentalPersistentPtyConnectTokenInput, ExperimentalPersistentPtyConnectTokenOutput, + ExperimentalMediaGetInput, + ExperimentalMediaGetOutput, ShellListInput, ShellListOutput, ShellCreateInput, @@ -2024,6 +2026,20 @@ export function make(options: ClientOptions) { requestOptions, ).then((value) => value.data), }, + media: { + get: (input: ExperimentalMediaGetInput, requestOptions?: RequestOptions) => + request( + { + method: "GET", + path: `/api/experimental/media/${encodeURIComponent(input.hash)}`, + successStatus: 200, + declaredStatuses: [400, 401, 404], + empty: false, + binary: true, + }, + requestOptions, + ), + }, }, shell: { list: (input?: ShellListInput, requestOptions?: RequestOptions) => diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 303f8e07afa2..bee7df75b10c 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -5832,6 +5832,10 @@ export type ExperimentalPersistentPtyConnectTokenInput = { export type ExperimentalPersistentPtyConnectTokenOutput = { data: PtyTicketConnectToken }["data"] +export type ExperimentalMediaGetInput = { readonly hash: { readonly hash: string }["hash"] } + +export type ExperimentalMediaGetOutput = globalThis.Uint8Array + export type ShellListInput = { readonly location?: { readonly location?: { readonly directory?: string | undefined } | undefined }["location"] } diff --git a/packages/client/src/promise/index.ts b/packages/client/src/promise/index.ts index 5b2561d6d440..04d10d1eef5b 100644 --- a/packages/client/src/promise/index.ts +++ b/packages/client/src/promise/index.ts @@ -21,4 +21,8 @@ export type { SkillApi, } from "./api.js" export type { EventSubscribeOutput as OpenCodeEvent } from "./generated/types.js" +export type { + MediaFileAttachment as PromptMediaFileAttachment, + Attachment as PromptAttachment, +} from "@opencode/schema/prompt" export type OpenCodeClient = ReturnType diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 2a975d57ffc5..78c1b0644558 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -53,7 +53,8 @@ test("exposes every standard HTTP API group", () => { expect(Object.keys(client.vcs)).toEqual(["init", "get", "base", "status", "branch", "diff"]) expect(Object.keys(client.pty)).toEqual(["list", "create", "get", "update", "remove", "connect"]) expect(Object.keys(client.pty.connect)).toEqual(["token"]) - expect(Object.keys(client.experimental)).toEqual(["persistentPty"]) + expect(Object.keys(client.experimental).sort()).toEqual(["media", "persistentPty"]) + expect(Object.keys(client.experimental.media)).toEqual(["get"]) expect(client.experimental.persistentPty.read).toBeFunction() expect(Object.keys(client.shell)).toEqual(["list", "create", "get", "output", "remove"]) expect(Object.keys(client.project)).toEqual(["list", "update"]) diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index ee29a7409302..73135b5f99b3 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -1,7 +1,7 @@ export * as Session from "./session.js" export * from "./session/schema.js" -import { DateTime, Effect, Layer, Context, Stream } from "effect" +import { DateTime, Effect, Layer, Context, Option, Stream } from "effect" import { LLMClient } from "@opencode/ai/effect" import { ListAnchor, ListCursor, ListFilter } from "@opencode/schema/session" import { and, desc, eq } from "drizzle-orm" @@ -16,6 +16,7 @@ import { Instance } from "./instance/service.js" import { Database } from "./database/database.js" import { SessionProjector } from "./session/projector.js" import { SessionMessageTable } from "./session/sql.js" +import { SessionBlob } from "./session/blob.js" import { SessionSchema } from "./session/schema.js" import { RelativePath } from "./schema.js" import { Agent } from "@opencode/schema/agent" @@ -147,6 +148,15 @@ export interface Interface { sessionID: SessionSchema.ID messageID: SessionMessage.ID }) => Effect.Effect + /** Experimental: {@link Interface.messages} with files stored in `blob` left as `media:` refs when the session's location enables `experimental.media_blob`. */ + readonly messagesWithRefs: ( + input: SessionStore.MessagesInput, + ) => Effect.Effect + /** Experimental: {@link Interface.message} with files stored in `blob` left as `media:` refs when the session's location enables `experimental.media_blob`. */ + readonly messageWithRefs: (input: { + sessionID: SessionSchema.ID + messageID: SessionMessage.ID + }) => Effect.Effect readonly context: ( sessionID: SessionSchema.ID, ) => Effect.Effect @@ -166,6 +176,8 @@ export interface Interface { * unhandled compaction barriers. */ readonly inbox: (sessionID: SessionSchema.ID) => Effect.Effect + /** Experimental: {@link Interface.inbox} with files stored in `blob` left as `media:` refs when the session's location enables `experimental.media_blob`. */ + readonly inboxWithRefs: (sessionID: SessionSchema.ID) => Effect.Effect readonly cancelInbox: (input: InboxItemRef) => Effect.Effect readonly steerInbox: (input: InboxItemRef) => Effect.Effect readonly queueInbox: (input: InboxItemRef) => Effect.Effect @@ -268,6 +280,16 @@ const layer = Layer.effect( const environments = yield* SessionEnvironment.Service const locations = yield* LocationServiceMap.Service const sessions = yield* Session.make() + // `experimental.media_blob` as the session's own location configures it, where its media is written. + const keepsMediaRefs = (session: SessionSchema.Info) => + SessionBlob.enabled.pipe( + instances.provide(session), + Effect.catch((cause) => + Effect.logWarning("could not resolve experimental.media_blob; serving media inline", { cause }).pipe( + Effect.as(false), + ), + ), + ) const isDurableSessionEvent = (item: Bus.LogItem): item is SessionEvent.DurableEvent => Object.hasOwn(SessionEvent.Durable.cases, item.type) @@ -413,6 +435,18 @@ const layer = Layer.effect( return yield* store.messages(input) }), message: (input) => sessions.forSession(input.sessionID).message(input.messageID), + messagesWithRefs: Effect.fn("Session.messagesWithRefs")(function* (input) { + const session = yield* result.get(input.sessionID) + if (!(yield* keepsMediaRefs(session))) return yield* store.messages(input) + return yield* store.messagesWithRefs(input) + }), + messageWithRefs: Effect.fn("Session.messageWithRefs")(function* (input) { + const session = yield* result.get(input.sessionID).pipe(Effect.option) + if (Option.isNone(session)) return undefined + if (!(yield* keepsMediaRefs(session.value))) return yield* result.message(input) + const stored = yield* store.messageWithRefs(input.messageID) + return stored?.sessionID === input.sessionID ? stored.message : undefined + }), context: Effect.fn("Session.context")(function* (sessionID) { yield* result.get(sessionID) return yield* store.context(sessionID) @@ -429,6 +463,11 @@ const layer = Layer.effect( }) }), inbox: (sessionID) => sessions.forSession(sessionID).inbox(), + inboxWithRefs: Effect.fn("Session.inboxWithRefs")(function* (sessionID) { + const session = yield* result.get(sessionID) + if (!(yield* keepsMediaRefs(session))) return yield* result.inbox(sessionID) + return yield* SessionInbox.listWithRefs(db, sessionID) + }), cancelInbox: (input) => sessions.forSession(input.sessionID).cancelInbox(input.inboxID), steerInbox: (input) => sessions.forSession(input.sessionID).steerInbox(input.inboxID), queueInbox: (input) => sessions.forSession(input.sessionID).queueInbox(input.inboxID), diff --git a/packages/core/src/session/blob.ts b/packages/core/src/session/blob.ts index 9208b799d5f4..72639427dfb1 100644 --- a/packages/core/src/session/blob.ts +++ b/packages/core/src/session/blob.ts @@ -1,10 +1,10 @@ export * as SessionBlob from "./blob.js" import { createHash } from "node:crypto" -import { and, eq, gte, inArray, notExists, notInArray, type SQL, sql } from "drizzle-orm" +import { and, eq, exists, gte, inArray, notExists, notInArray, type SQL, sql } from "drizzle-orm" import { Effect, Option, Schema } from "effect" import { map } from "effect/Array" -import type { FileAttachment } from "@opencode/schema/prompt" +import type { Attachment, FileAttachment } from "@opencode/schema/prompt" import type { Tool } from "@opencode/schema/tool" import { Config } from "../config.js" import type { Database } from "../database/database.js" @@ -40,27 +40,59 @@ function collect(data: MessageRow["data"] | InboxRow["payload"], hashes: Set `[media unavailable: ${name === undefined ? mime : `${name} (${mime})`}]` -function inlineFile(file: SessionStored.File, blobs: Blobs): FileAttachment { +function unavailableFile(file: SessionStored.FileRef): FileAttachment { + const { uri: _, ...rest } = file + return { ...rest, data: Buffer.from(unavailable(file.mime, file.name)).toString("base64"), mime: "text/plain" } +} + +const unavailableToolItem = (item: Tool.Content & { readonly type: "file" }): Tool.Content => ({ + type: "text", + text: unavailable(item.mime, item.name), +}) + +function inlineFile(file: SessionStored.File, blobs: Blobs): SessionStored.File { if (!isFileRef(file)) return file - const { uri, ...rest } = file - const data = blobs.get(SessionStored.refHash(uri)) - if (data === undefined) - return { ...rest, data: Buffer.from(unavailable(file.mime, file.name)).toString("base64"), mime: "text/plain" } + const data = blobs.get(SessionStored.refHash(file.uri)) + if (data === undefined) return unavailableFile(file) + const { uri: _, ...rest } = file return { ...rest, data } } function inlineToolItem(item: Tool.Content, blobs: Blobs): Tool.Content { if (!isToolFileRef(item)) return item const data = blobs.get(SessionStored.refHash(item.uri)) - if (data === undefined) return { type: "text", text: unavailable(item.mime, item.name) } + if (data === undefined) return unavailableToolItem(item) return { ...item, uri: `data:${item.mime};base64,${data}` } } -function inlinePart(part: Part, blobs: Blobs): Part { - if (part.type !== "tool" || (part.state.status !== "completed" && part.state.status !== "error")) return part - const content = part.state.content - if (!content?.some(isToolFileRef)) return part - return { ...part, state: { ...part.state, content: map(content, (item) => inlineToolItem(item, blobs)) } } +/** Rewrites the ref files in a message row, returning the row itself when it names none. */ +function rewriteMessage( + row: MessageRow, + file: (file: SessionStored.File) => SessionStored.File, + item: (item: Tool.Content) => Tool.Content, +): MessageRow { + const data = row.data + if ("files" in data && data.files?.some(isFileRef)) return { ...row, data: { ...data, files: data.files.map(file) } } + if ("content" in data && data.content.some((part) => toolItems(part).some(isToolFileRef))) + return { + ...row, + data: { + ...data, + content: data.content.map((part): Part => { + if (part.type !== "tool" || (part.state.status !== "completed" && part.state.status !== "error")) return part + const content = part.state.content + if (!content?.some(isToolFileRef)) return part + return { ...part, state: { ...part.state, content: map(content, item) } } + }), + }, + } + return row +} + +function rewriteInbox(row: InboxRow, file: (file: SessionStored.File) => SessionStored.File): InboxRow { + const payload = row.payload + if (!("files" in payload) || !payload.files?.some(isFileRef)) return row + return { ...row, payload: { ...payload, files: payload.files.map(file) } } } const load = Effect.fnUntraced(function* (db: DatabaseService, hashes: ReadonlySet) { @@ -87,14 +119,13 @@ export const hydrate = Effect.fnUntraced(function* (db: DatabaseService, rows: R for (const row of rows) collect(row.data, hashes) if (hashes.size === 0) return rows const blobs = yield* load(db, hashes) - return rows.map((row): MessageRow => { - const data = row.data - if ("files" in data && data.files?.some(isFileRef)) - return { ...row, data: { ...data, files: data.files.map((file) => inlineFile(file, blobs)) } } - if ("content" in data && data.content.some((part) => toolItems(part).some(isToolFileRef))) - return { ...row, data: { ...data, content: data.content.map((part) => inlinePart(part, blobs)) } } - return row - }) + return rows.map((row) => + rewriteMessage( + row, + (file) => inlineFile(file, blobs), + (item) => inlineToolItem(item, blobs), + ), + ) }) /** Restores inline media for the refs in `session_inbox` rows. Rows without refs are returned untouched. */ @@ -103,11 +134,107 @@ export const hydrateInbox = Effect.fnUntraced(function* (db: DatabaseService, ro for (const row of rows) collect(row.payload, hashes) if (hashes.size === 0) return rows const blobs = yield* load(db, hashes) - return rows.map((row): InboxRow => { - const payload = row.payload - if (!("files" in payload) || !payload.files?.some(isFileRef)) return row - return { ...row, payload: { ...payload, files: payload.files.map((file) => inlineFile(file, blobs)) } } + return rows.map((row) => rewriteInbox(row, (file) => inlineFile(file, blobs))) +}) + +/** What a client that fetches bytes itself learns about a blob. */ +export interface Meta { + readonly bytes: number + readonly width: number | null + readonly height: number | null +} + +/** + * For clients that fetch media themselves: refs stay in `rows`, except those whose blob is gone or not referenced (the media route only serves referenced blobs), which become the + * same unavailable placeholder {@link hydrate} gives. Returns the size and dimensions of every blob still named. + */ +export const keepRefs = Effect.fnUntraced(function* (db: DatabaseService, rows: ReadonlyArray) { + const hashes = new Set() + for (const row of rows) collect(row.data, hashes) + const meta = yield* loadMeta(db, hashes) + return { rows: rows.map((row) => rewriteMessage(row, keepFile(meta), keepToolItem(meta))), meta } +}) + +/** {@link keepRefs} for inbox rows. */ +export const keepInboxRefs = Effect.fnUntraced(function* (db: DatabaseService, rows: ReadonlyArray) { + const hashes = new Set() + for (const row of rows) collect(row.payload, hashes) + const meta = yield* loadMeta(db, hashes) + return { rows: rows.map((row) => rewriteInbox(row, keepFile(meta))), meta } +}) + +const loadMeta = Effect.fnUntraced(function* (db: DatabaseService, hashes: ReadonlySet) { + const meta = new Map() + const all = [...hashes] + for (let offset = 0; offset < all.length; offset += 500) { + const found = yield* db + .select({ hash: BlobTable.hash, bytes: BlobTable.bytes, width: BlobTable.width, height: BlobTable.height }) + .from(BlobTable) + .where( + and( + inArray(BlobTable.hash, all.slice(offset, offset + 500)), + exists( + db + .select({ one: sql`1` }) + .from(BlobRefTable) + .where(eq(BlobRefTable.hash, BlobTable.hash)), + ), + ), + ) + .all() + .pipe(Effect.orDie) + for (const { hash, ...rest } of found) meta.set(hash, rest) + } + return meta +}) + +const keepFile = + (meta: ReadonlyMap) => + (file: SessionStored.File): SessionStored.File => + isFileRef(file) && !meta.has(SessionStored.refHash(file.uri)) ? unavailableFile(file) : file + +const keepToolItem = + (meta: ReadonlyMap) => + (item: Tool.Content): Tool.Content => + isToolFileRef(item) && !meta.has(SessionStored.refHash(item.uri)) ? unavailableToolItem(item) : item + +/** Adds each kept ref's blob size and dimensions to files decoded from {@link keepRefs} or {@link keepInboxRefs}. */ +export function describe(files: ReadonlyArray, meta: ReadonlyMap): Attachment[] { + return files.map((file) => { + if (!("uri" in file)) return file + const blob = meta.get(SessionStored.refHash(file.uri)) + if (!blob) return file + return { + ...file, + bytes: blob.bytes, + ...(blob.width === null ? {} : { width: blob.width }), + ...(blob.height === null ? {} : { height: blob.height }), + } }) +} + +/** + * One blob's type and bytes, for the experimental media route and `media:` prompt attachments. Only a blob some + * message or inbox item still references is readable; an unreferenced one awaiting its sweep reads as missing. + */ +export const read = Effect.fnUntraced(function* (db: DatabaseService, hash: string) { + const blob = yield* db + .select({ mime: BlobTable.mime, data: BlobTable.data }) + .from(BlobTable) + .where( + and( + eq(BlobTable.hash, hash), + exists( + db + .select({ one: sql`1` }) + .from(BlobRefTable) + .where(eq(BlobRefTable.hash, BlobTable.hash)), + ), + ), + ) + .get() + .pipe(Effect.orDie) + return blob && { mime: blob.mime, data: view(blob.data) } }) /** A blob's hash is the lowercase hex SHA-256 of its raw bytes. */ @@ -124,8 +251,8 @@ export interface Owner { type Pending = Map type AssistantPart = SessionMessage.Assistant["content"][number] -/** Reads `experimental.media_blob` from the Config in scope on every write; without one, media stays inline. */ -const writeEnabled = Effect.gen(function* () { +/** Reads `experimental.media_blob` from the Config in scope on every call; without one, it is off. */ +export const enabled = Effect.gen(function* () { const config = yield* Effect.serviceOption(Config.Service) if (Option.isNone(config)) return false return Config.latest(yield* config.value.entries(), "experimental")?.media_blob === true @@ -172,7 +299,7 @@ const save = Effect.fnUntraced(function* (db: DatabaseService, pending: Pending) export const externalizePayload = Effect.fnUntraced(function* < T extends { readonly files?: ReadonlyArray }, >(db: DatabaseService, input: T): Effect.fn.Return { - if (input.files === undefined || !(yield* writeEnabled)) return input + if (input.files === undefined || !(yield* enabled)) return input const pending: Pending = new Map() const files = input.files.map((file) => fileRef(pending, file)) if (pending.size === 0) return input @@ -189,7 +316,7 @@ export const externalize = Effect.fnUntraced(function* ( message: SessionMessage.Info | SessionStored.User, ) { if (message.type === "user") return yield* externalizePayload(db, message) - if (message.type !== "assistant" || !(yield* writeEnabled)) return message + if (message.type !== "assistant" || !(yield* enabled)) return message const pending: Pending = new Map() const content = message.content.map((part) => toolPartRefs(pending, part)) if (pending.size === 0) return message @@ -203,6 +330,29 @@ const hashesOf = (data: MessageRow["data"] | InboxRow["payload"]) => { return hashes } +function storedPartRefs(pending: Pending, part: Part): Part { + if (part.type !== "tool" || (part.state.status !== "completed" && part.state.status !== "error")) return part + const content = part.state.content + if (!content?.some((item) => item.type === "file" && DATA_URI.test(item.uri))) return part + return { ...part, state: { ...part.state, content: map(content, (item) => toolRef(pending, item)) } } +} + +/** + * Moves a stored row's inline media into blobs whatever the flag says, for offline backfills. Returns the row to + * store (the same object when nothing moved), the blobs to insert, and every hash the row then names. + */ +export function split(data: MessageRow["data"] | InboxRow["payload"]) { + const pending: Pending = new Map() + const stored = + "files" in data && data.files?.some((file) => !isFileRef(file)) + ? { ...data, files: data.files.map((file) => fileRef(pending, file)) } + : "content" in data + ? { ...data, content: data.content.map((part) => storedPartRefs(pending, part)) } + : data + const result = pending.size > 0 ? stored : data + return { data: result, moved: pending.size > 0, blobs: [...pending.values()], hashes: hashesOf(result) } +} + /** Records that `owner` references each blob its stored row names, so release keeps them until it lets go. */ export const recordRefs = Effect.fnUntraced(function* ( db: DatabaseService, diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index ea7a69ea56d8..19f0ce716584 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -19,6 +19,7 @@ type DatabaseService = Database.Interface["db"] const decode = Schema.decodeUnknownEffect(SessionMessage.Info) const decodeSync = Schema.decodeUnknownSync(SessionMessage.Info) const decodeStoredUser = Schema.decodeUnknownEffect(SessionStored.User) +const decodeMedia = Schema.decodeUnknownEffect(SessionMessage.MediaInfo) /** * Which completed compactions bound a history read. Local summaries always do. Native @@ -115,6 +116,27 @@ export const decodeMessageRows = Effect.fnUntraced(function* ( return yield* Effect.forEach(yield* SessionBlob.hydrate(db, rows), decodeMessageRow, { concurrency: 1 }) }) +/** Decodes rows for clients that fetch media themselves, so files stored in `blob` stay `media:` refs. */ +export const decodeMessageRowsWithRefs = Effect.fnUntraced(function* ( + db: DatabaseService, + rows: ReadonlyArray, +) { + const kept = yield* SessionBlob.keepRefs(db, rows) + return yield* Effect.forEach( + kept.rows, + (row) => + decodeMedia(rowInput(row)).pipe( + Effect.mapError(decodeError(row)), + Effect.map((message) => + message.type === "user" && message.files + ? { ...message, files: SessionBlob.describe(message.files, kept.meta) } + : message, + ), + ), + { concurrency: 1 }, + ) +}) + const messageEntries = Effect.fnUntraced(function* ( db: DatabaseService, sessionID: SessionSchema.ID, diff --git a/packages/core/src/session/inbox.ts b/packages/core/src/session/inbox.ts index 9be6743c35e9..bbdef5def1ec 100644 --- a/packages/core/src/session/inbox.ts +++ b/packages/core/src/session/inbox.ts @@ -9,6 +9,9 @@ import { Delivery, Info, Item, + MediaInfo, + MediaUser, + MediaUserPayload, Move, MovePayload, Synthetic, @@ -35,6 +38,9 @@ export { Delivery, Info, Item, + MediaInfo, + MediaUser, + MediaUserPayload, Move, MovePayload, Synthetic, @@ -51,6 +57,7 @@ export { export type Promotable = "input" | "steer" const decodeUser = Schema.decodeUnknownSync(UserPayload) +const decodeMediaUser = Schema.decodeUnknownSync(MediaUserPayload) const decodeStoredUser = Schema.decodeUnknownSync(SessionStored.UserPayload) const encodeStoredUser = Schema.encodeSync(SessionStored.UserPayload) const decodeSynthetic = Schema.decodeUnknownSync(SyntheticPayload) @@ -417,15 +424,37 @@ export const projectDeliveryChanged = Effect.fn("SessionInbox.projectDeliveryCha }), ) -export const list = Effect.fn("SessionInbox.list")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - const rows = yield* db +const listRows = (db: DatabaseService, sessionID: SessionSchema.ID) => + db .select() .from(SessionInboxTable) .where(eq(SessionInboxTable.session_id, sessionID)) .orderBy(asc(SessionInboxTable.enqueued_seq)) .all() .pipe(Effect.orDie) - return (yield* SessionBlob.hydrateInbox(db, rows)).map(fromRow) + +export const list = Effect.fn("SessionInbox.list")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { + return (yield* SessionBlob.hydrateInbox(db, yield* listRows(db, sessionID))).map(fromRow) +}) + +/** {@link list} for clients that fetch media themselves: files stored in `blob` stay `media:` refs. */ +export const listWithRefs = Effect.fn("SessionInbox.listWithRefs")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, +) { + const kept = yield* SessionBlob.keepInboxRefs(db, yield* listRows(db, sessionID)) + return kept.rows.map((row): MediaInfo => { + if (row.type !== "user") return fromRow(row) + const payload = decodeMediaUser(row.payload) + return MediaUser.make({ + id: row.id, + sessionID: row.session_id, + time: { created: DateTime.makeUnsafe(row.time_created) }, + type: "user", + payload: payload.files ? { ...payload, files: SessionBlob.describe(payload.files, kept.meta) } : payload, + delivery: row.delivery, + }) + }) }) export const moveIDs = Effect.fn("SessionInbox.moveIDs")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { @@ -460,7 +489,12 @@ export const nextPromotable = Effect.fn("SessionInbox.nextPromotable")(function* /** What a boundary decides on: prompts are promoted by id, so only a move's payload is read. */ export type Promotion = | { readonly id: SessionMessage.ID; readonly type: "user" | "synthetic" | "compaction"; readonly delivery: Delivery } - | { readonly id: SessionMessage.ID; readonly type: "move"; readonly delivery: Delivery; readonly payload: MovePayload } + | { + readonly id: SessionMessage.ID + readonly type: "move" + readonly delivery: Delivery + readonly payload: MovePayload + } const promotion = (row: typeof SessionInboxTable.$inferSelect): Promotion => row.type === "move" diff --git a/packages/core/src/session/prompt.ts b/packages/core/src/session/prompt.ts index 20246af780de..a44ccaace330 100644 --- a/packages/core/src/session/prompt.ts +++ b/packages/core/src/session/prompt.ts @@ -6,15 +6,18 @@ import { SessionInbox } from "@opencode/schema/session-inbox" import type { Session } from "@opencode/schema/session" import type { SessionMessage } from "@opencode/schema/session-message" import { FSUtil } from "@opencode/util/fs-util" -import { Effect } from "effect" +import { Effect, Option, Schema } from "effect" import path from "path" import { fileURLToPath } from "url" +import { Database } from "../database/database.js" import { Image } from "../image.js" import { Mime } from "../mime.js" import { Plugin } from "../plugin/service.js" import { PluginHooks } from "../plugin/hooks.js" import { Skill } from "../skill.js" +import { SessionBlob } from "./blob.js" import { AttachmentError, SkillNotFoundError } from "./error.js" +import { SessionStored } from "./stored.js" export type Input = { text: string @@ -86,6 +89,19 @@ export const prepare = Effect.fn("SessionPrompt.prepare")(function* (request: { const materializeAttachment = Effect.fn("SessionPrompt.materializeAttachment")(function* ( input: PromptInput.FileAttachment, ) { + // Stored blob bytes were normalized when first admitted, so a `media:` ref re-admits them unchanged. + const ref = Schema.decodeUnknownOption(SessionStored.Ref)(input.uri) + if (Option.isSome(ref)) { + const blob = yield* readBlobAttachment(ref.value) + return FileAttachment.create({ + data: Base64.make(Buffer.from(blob.data).toString("base64")), + mime: blob.mime, + source: { type: "inline" }, + name: input.name, + description: input.description, + mention: input.mention, + }) + } const label = attachmentLabel(input) const resolved = input.uri.startsWith("data:") ? { @@ -141,6 +157,14 @@ const normalizeImageAttachment = Effect.fn("SessionPrompt.normalizeImageAttachme return { data: Base64.make(normalized.content), mime: normalized.mime } }) +/** A `media:` ref a client received from an experimental `experimental-media-uri` read, sent back as an attachment. */ +const readBlobAttachment = Effect.fn("SessionPrompt.readBlobAttachment")(function* (uri: SessionStored.Ref) { + const { db } = yield* Database.Service + const blob = yield* SessionBlob.read(db, SessionStored.refHash(uri)) + if (!blob) return yield* new AttachmentError({ uri, message: `Attachment is no longer stored: ${uri}` }) + return blob +}) + const readFileAttachment = Effect.fn("SessionPrompt.readFileAttachment")(function* (uri: string) { const fs = yield* FSUtil.Service const url = yield* Effect.try({ diff --git a/packages/core/src/session/session.ts b/packages/core/src/session/session.ts index d7153134faee..cec095ad1590 100644 --- a/packages/core/src/session/session.ts +++ b/packages/core/src/session/session.ts @@ -176,7 +176,11 @@ export const make = Effect.fn("Session.make")(function* () { sessionID: session.id, item, }) - }).pipe(instances.provide(session), Effect.provideService(FSUtil.Service, fs)) + }).pipe( + instances.provide(session), + Effect.provideService(Database.Service, database), + Effect.provideService(FSUtil.Service, fs), + ) }).pipe( Effect.catchTag("SessionInbox.LifecycleConflict", () => new PromptConflictError({ sessionID, messageID })), ) diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index 00a7eeb01049..a46f5e04816d 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -59,6 +59,12 @@ export interface Interface { readonly message: ( messageID: SessionMessage.ID, ) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.Info } | undefined> + /** {@link Interface.messages} for clients that fetch media themselves: files stored in `blob` stay `media:` refs. */ + readonly messagesWithRefs: (input: MessagesInput) => Effect.Effect + /** {@link Interface.message} for clients that fetch media themselves. */ + readonly messageWithRefs: ( + messageID: SessionMessage.ID, + ) => Effect.Effect<{ readonly sessionID: Session.ID; readonly message: SessionMessage.MediaInfo } | undefined> /** * Top-level Sessions holding an execution claim. Recoverable background * children are resumed separately through their durable Job records. @@ -92,6 +98,45 @@ const layer = Layer.effect( Effect.gen(function* () { const { db } = yield* Database.Service + /** One page of a session's message rows, oldest first within the page when paging backwards. */ + const messageRows = Effect.fnUntraced(function* (input: MessagesInput) { + const direction = input.cursor?.direction ?? "next" + const requestedOrder = input.order ?? "desc" + const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder + const anchor = input.cursor + ? yield* db + .select({ seq: SessionMessageTable.seq }) + .from(SessionMessageTable) + .where( + and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.cursor.id)), + ) + .get() + .pipe(Effect.orDie) + : undefined + if (input.cursor && !anchor) return [] + const boundary = anchor + ? order === "asc" + ? gt(SessionMessageTable.seq, anchor.seq) + : lt(SessionMessageTable.seq, anchor.seq) + : undefined + const query = db + .select() + .from(SessionMessageTable) + .where( + and( + eq(SessionMessageTable.session_id, input.sessionID), + boundary, + input.type === undefined ? undefined : eq(SessionMessageTable.type, input.type), + ), + ) + .orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq)) + const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(Effect.orDie) + return direction === "previous" ? rows.toReversed() : rows + }) + + const messageRow = (messageID: SessionMessage.ID) => + db.select().from(SessionMessageTable).where(eq(SessionMessageTable.id, messageID)).get().pipe(Effect.orDie) + return Service.of({ get: Effect.fnUntraced(function* (sessionID) { const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get().pipe(Effect.orDie) @@ -141,53 +186,24 @@ const layer = Layer.effect( return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row)) }), messages: Effect.fn("SessionStore.messages")(function* (input) { - const direction = input.cursor?.direction ?? "next" - const requestedOrder = input.order ?? "desc" - const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder - const anchor = input.cursor - ? yield* db - .select({ seq: SessionMessageTable.seq }) - .from(SessionMessageTable) - .where( - and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.cursor.id)), - ) - .get() - .pipe(Effect.orDie) - : undefined - if (input.cursor && !anchor) return [] - const boundary = anchor - ? order === "asc" - ? gt(SessionMessageTable.seq, anchor.seq) - : lt(SessionMessageTable.seq, anchor.seq) - : undefined - const query = db - .select() - .from(SessionMessageTable) - .where( - and( - eq(SessionMessageTable.session_id, input.sessionID), - boundary, - input.type === undefined ? undefined : eq(SessionMessageTable.type, input.type), - ), - ) - .orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq)) - const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( - Effect.orDie, - ) - return yield* SessionHistory.decodeMessageRows(db, direction === "previous" ? rows.toReversed() : rows) + return yield* SessionHistory.decodeMessageRows(db, yield* messageRows(input)) + }), + messagesWithRefs: Effect.fn("SessionStore.messagesWithRefs")(function* (input) { + return yield* SessionHistory.decodeMessageRowsWithRefs(db, yield* messageRows(input)) }), context: Effect.fn("SessionStore.context")((sessionID) => SessionHistory.load(db, sessionID, "latest")), message: Effect.fn("SessionStore.message")(function* (messageID) { - const row = yield* db - .select() - .from(SessionMessageTable) - .where(eq(SessionMessageTable.id, messageID)) - .get() - .pipe(Effect.orDie) + const row = yield* messageRow(messageID) if (!row) return undefined const [message] = yield* SessionHistory.decodeMessageRows(db, [row]).pipe(Effect.orDie) return message && { sessionID: row.session_id, message } }), + messageWithRefs: Effect.fn("SessionStore.messageWithRefs")(function* (messageID) { + const row = yield* messageRow(messageID) + if (!row) return undefined + const [message] = yield* SessionHistory.decodeMessageRowsWithRefs(db, [row]).pipe(Effect.orDie) + return message && { sessionID: row.session_id, message } + }), listSuspended: Effect.fn("SessionStore.listSuspended")(function* () { return yield* db .select({ sessionID: SessionTable.id }) diff --git a/packages/core/src/session/stored.ts b/packages/core/src/session/stored.ts index f040801d6dcd..eb076bd1e9c8 100644 --- a/packages/core/src/session/stored.ts +++ b/packages/core/src/session/stored.ts @@ -1,7 +1,7 @@ export * as SessionStored from "./stored.js" import { Schema } from "effect" -import { FileAttachment } from "@opencode/schema/prompt" +import { FileAttachment, MediaRef } from "@opencode/schema/prompt" import { optional } from "@opencode/schema/schema" import { UserPayload as InboxUserPayload } from "@opencode/schema/session-inbox" import { Tool } from "@opencode/schema/tool" @@ -11,8 +11,8 @@ import { SessionMessage } from "./message.js" * Stored rows may reference a `blob` row as `media:` and the lowercase hex SHA-256 of its raw bytes. The blob * row holds the bytes and size; the ref names no server route. */ -export const Ref = Schema.String.check(Schema.isPattern(/^media:[0-9a-f]{64}$/)) -export type Ref = typeof Ref.Type +export const Ref = MediaRef +export type Ref = MediaRef /** The `blob` hash a {@link Ref} names. */ export const refHash = (ref: Ref) => ref.slice("media:".length) diff --git a/packages/core/test/session-owned.test.ts b/packages/core/test/session-owned.test.ts index e1322cfb6f4d..8d68b5b49cfb 100644 --- a/packages/core/test/session-owned.test.ts +++ b/packages/core/test/session-owned.test.ts @@ -752,7 +752,7 @@ describe("SessionPrompt preparation", () => { } const items = yield* Effect.forEach([0, 1], () => SessionPrompt.prepare(request)).pipe( fixture.instances.provide(request.session), - Effect.satisfiesServicesType(), + Effect.satisfiesServicesType(), ) expect(items[0]).toEqual(items[1]) diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index 48964f54d3e4..1fcb90d8d2fb 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -11867,6 +11867,68 @@ "x-websocket": true } }, + "/api/experimental/media/{hash}": { + "get": { + "tags": ["experimental"], + "operationId": "experimental.media.get", + "parameters": [ + { + "name": "hash", + "in": "path", + "schema": { + "$ref": "#/components/schemas/MediaHash" + }, + "required": true + } + ], + "security": [], + "responses": { + "200": { + "description": "Success", + "content": { + "application/octet-stream": { + "schema": { + "type": "string", + "format": "binary" + } + } + } + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedErrorEncoded" + } + } + } + }, + "404": { + "description": "FileNotFoundError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/FileNotFoundErrorEncoded" + } + } + } + } + }, + "description": "Read a content-addressed media blob by SHA-256 hash with immutable Cache-Control, ETag, and 304 Not Modified support. Only blobs a stored message or inbox item still references are served. Experimental: may change without compatibility guarantees.", + "summary": "Read media blob" + } + }, "/api/shell": { "get": { "tags": ["shell"], @@ -17580,6 +17642,10 @@ "required": ["_tag", "server", "message"], "additionalProperties": false }, + "MediaHash": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + }, "MessageNotFoundErrorEncoded": { "type": "object", "properties": { diff --git a/packages/protocol/src/api.ts b/packages/protocol/src/api.ts index 555700c499f4..4aa9e16bb72c 100644 --- a/packages/protocol/src/api.ts +++ b/packages/protocol/src/api.ts @@ -2,6 +2,7 @@ import { Context } from "effect" import { HttpApi, HttpApiGroup, HttpApiMiddleware, OpenApi } from "effect/http-api" import { SchemaErrorMiddleware } from "./middleware/schema-error.js" import { GenerateGroup } from "./groups/generate.js" +import { ExperimentalGroup } from "./groups/experimental.js" import { VoiceGroup } from "./groups/voice.js" import { MessageGroup } from "./groups/message.js" import { ModelGroup } from "./groups/model.js" @@ -20,7 +21,6 @@ import { PluginGroup } from "./groups/plugin.js" import { ServerGroup } from "./groups/server.js" import { DebugGroup } from "./groups/debug.js" import { makePtyGroup } from "./groups/pty.js" -import { PersistentPtyGroup } from "./groups/persistent-pty.js" import { ShellGroup } from "./groups/shell.js" import { ReferenceGroup } from "./groups/reference.js" import { Authorization } from "./middleware/authorization.js" @@ -100,7 +100,7 @@ type ApiGroups< | typeof WorktreeGroup | typeof GenerateGroup | typeof VoiceGroup - | typeof PersistentPtyGroup + | typeof ExperimentalGroup | typeof CredentialGroup | typeof ProjectGroup | LocationGroups @@ -187,7 +187,7 @@ const makeApiFromGroup = < .add(RpcGroup.middleware(locationMiddleware)) .add(eventGroup) .add(makePtyGroup(locationMiddleware)) - .add(PersistentPtyGroup) + .add(ExperimentalGroup) .add(ShellGroup.middleware(locationMiddleware)) .add(ReferenceGroup.middleware(locationMiddleware)) .add(WorktreeGroup) diff --git a/packages/protocol/src/client.ts b/packages/protocol/src/client.ts index 57f1cb6c7c26..8d557798ce51 100644 --- a/packages/protocol/src/client.ts +++ b/packages/protocol/src/client.ts @@ -72,6 +72,7 @@ export const promiseOmitEndpoints = new Set(["pty.connect", "persistentPty.conne export const effectOmitEndpoints = new Set([ "fs.read", "fs.write", + "media.get", "pty.connect", "persistentPty.connect", "mcp.client", diff --git a/packages/protocol/src/groups/experimental.ts b/packages/protocol/src/groups/experimental.ts new file mode 100644 index 000000000000..c282c7917e74 --- /dev/null +++ b/packages/protocol/src/groups/experimental.ts @@ -0,0 +1,5 @@ +import { MediaGetEndpoint } from "./media.js" +import { PersistentPtyGroup } from "./persistent-pty.js" + +/** `server.experimental`: routes under /api/experimental, without compatibility guarantees. */ +export const ExperimentalGroup = PersistentPtyGroup.add(MediaGetEndpoint) diff --git a/packages/protocol/src/groups/media.ts b/packages/protocol/src/groups/media.ts new file mode 100644 index 000000000000..3e5054d1a3b4 --- /dev/null +++ b/packages/protocol/src/groups/media.ts @@ -0,0 +1,23 @@ +import { Schema } from "effect" +import { HttpApiEndpoint, HttpApiSchema, OpenApi } from "effect/http-api" +import { FileNotFoundError, InvalidRequestError } from "../errors.js" + +export const MediaHash = Schema.String.check(Schema.isPattern(/^[0-9a-f]{64}$/u)).annotate({ + identifier: "MediaHash", +}) + +/** Lives in the `server.experimental` group (see `groups/experimental.ts`). */ +export const MediaGetEndpoint = HttpApiEndpoint.get("media.get", "/api/experimental/media/:hash", { + params: { hash: MediaHash }, + success: Schema.Uint8Array.pipe(HttpApiSchema.asUint8Array()), + error: [FileNotFoundError, InvalidRequestError], +}).annotateMerge( + OpenApi.annotations({ + identifier: "experimental.media.get", + summary: "Read media blob", + description: + "Read a content-addressed media blob by SHA-256 hash with immutable Cache-Control, ETag, and 304 Not Modified support. Only blobs a stored message or inbox item still references are served. Experimental: may change without compatibility guarantees.", + // The route shares the `server.experimental` group with the persistent PTY prototype; tag it on its own. + transform: (operation) => ({ ...operation, tags: ["experimental"] }), + }), +) diff --git a/packages/protocol/src/groups/message.ts b/packages/protocol/src/groups/message.ts index 7260b3c87da2..a50080af0956 100644 --- a/packages/protocol/src/groups/message.ts +++ b/packages/protocol/src/groups/message.ts @@ -65,6 +65,38 @@ export const PublicSessionMessage = Schema.Union([ SessionMessage.Idle, ]).annotate({ identifier: "Session.Message.Info" }) +/** + * Experimental: {@link PublicSessionMessage} as sent to clients that opt in with `X-OpenCode-Capabilities: + * experimental-media-uri` while `experimental.media_blob` is on. User files stored in `blob` stay `media:` refs. + */ +export const PublicMediaSessionMessage = Schema.Union([ + SessionMessage.AgentSelected, + SessionMessage.ModelSelected, + PublicLocationSwitched, + SessionMessage.MediaUser, + SessionMessage.Synthetic, + SessionMessage.System, + SessionMessage.Skill, + SessionMessage.Shell, + SessionMessage.Assistant, + SessionMessage.Compaction, + SessionMessage.Idle, +]).annotate({ identifier: "Session.Message.MediaInfo" }) + +const MessagesCursor = Schema.Struct({ + previous: Schema.String.pipe(Schema.optional), + next: Schema.String.pipe(Schema.optional), +}) + +/** Experimental: the `session.messages` response for clients that opt into `experimental-media-uri`. */ +export const SessionMediaMessagesResponse = Schema.Struct({ + data: Schema.Array(PublicMediaSessionMessage), + cursor: MessagesCursor, +}) + +/** Experimental: the `session.message` response for clients that opt into `experimental-media-uri`. */ +export const SessionMediaMessageResponse = Schema.Struct({ data: PublicMediaSessionMessage }) + export const MessageGroup = HttpApiGroup.make("server.message") .add( HttpApiEndpoint.get("session.messages", "/api/session/:sessionID/message", { @@ -72,10 +104,7 @@ export const MessageGroup = HttpApiGroup.make("server.message") query: SessionMessagesQuery, success: Schema.Struct({ data: Schema.Array(PublicSessionMessage), - cursor: Schema.Struct({ - previous: Schema.String.pipe(Schema.optional), - next: Schema.String.pipe(Schema.optional), - }), + cursor: MessagesCursor, }).annotate({ identifier: "SessionMessagesResponse" }), error: [InvalidCursorError, SessionNotFoundError, UnknownError], }).annotateMerge( diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 405f1aec9487..47ac7cd47248 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -69,6 +69,15 @@ const PublicInboxInfo = Schema.Union([ PublicMove, ]).annotate({ identifier: "Session.Inbox.Info" }) +/** Experimental: the `session.inbox.list` response for clients that opt into `experimental-media-uri`. */ +export const SessionMediaInboxResponse = Schema.Struct({ + data: Schema.Array( + Schema.Union([SessionInbox.MediaUser, SessionInbox.Synthetic, SessionInbox.Compaction, PublicMove]).annotate({ + identifier: "Session.Inbox.MediaInfo", + }), + ), +}) + // Transcripts exported before v2.1 carry older names for native data. The wire stays plain JSON so any export file // is accepted; decoding upgrades those names and then validates the current message shape. const ImportedMessages = Schema.Array(Schema.Json).pipe( diff --git a/packages/schema/src/index.ts b/packages/schema/src/index.ts index 8025a185f03a..c7718c9f5689 100644 --- a/packages/schema/src/index.ts +++ b/packages/schema/src/index.ts @@ -36,6 +36,15 @@ export { PersistentPty } from "./persistent-pty.js" export { PtyTicket } from "./pty-ticket.js" export { Question } from "./question.js" export { Workspace } from "./workspace.js" -export { Prompt, PromptMention, FileSource, FileAttachment, AgentAttachment } from "./prompt.js" +export { + Prompt, + PromptMention, + FileSource, + FileAttachment, + MediaRef, + MediaFileAttachment, + Attachment, + AgentAttachment, +} from "./prompt.js" export { PromptInput } from "./prompt-input.js" export * from "./schema.js" diff --git a/packages/schema/src/prompt.ts b/packages/schema/src/prompt.ts index 0086ca11356d..666be1fa7c23 100644 --- a/packages/schema/src/prompt.ts +++ b/packages/schema/src/prompt.ts @@ -51,6 +51,32 @@ export const FileAttachment = Schema.Struct({ })), ) +/** `media:` and the lowercase hex SHA-256 of a stored blob's bytes. Names no server route. */ +export const MediaRef = Schema.String.check(Schema.isPattern(/^media:[0-9a-f]{64}$/)).annotate({ + identifier: "Prompt.MediaRef", +}) +export type MediaRef = typeof MediaRef.Type + +const { data: _, ...fileAttachmentFields } = FileAttachment.fields + +/** + * An attachment whose bytes stay on the server, sent to clients that opt in with the experimental `experimental-media-uri` + * capability. Fetch the bytes from `GET /api/experimental/media/:hash`. + */ +export interface MediaFileAttachment extends Schema.Schema.Type {} +export const MediaFileAttachment = Schema.Struct({ + ...fileAttachmentFields, + uri: MediaRef, + bytes: Schema.Int.pipe(optional), + width: Schema.Int.pipe(optional), + height: Schema.Int.pipe(optional), +}).annotate({ identifier: "Prompt.MediaFileAttachment" }) + +export const Attachment = Schema.Union([FileAttachment, MediaFileAttachment]).annotate({ + identifier: "Prompt.Attachment", +}) +export type Attachment = typeof Attachment.Type + export interface AgentAttachment extends Schema.Schema.Type {} export const AgentAttachment = Schema.Struct({ name: Schema.String, diff --git a/packages/schema/src/session-inbox.ts b/packages/schema/src/session-inbox.ts index 361c0c492ce8..16e900189369 100644 --- a/packages/schema/src/session-inbox.ts +++ b/packages/schema/src/session-inbox.ts @@ -3,7 +3,7 @@ export * as SessionInbox from "./session-inbox.js" import { Schema } from "effect" import { Location } from "./location.js" import { Project } from "./project.js" -import { Prompt } from "./prompt.js" +import { Attachment, Prompt } from "./prompt.js" import { DateTimeUtcFromMillis, optional, RelativePath } from "./schema.js" import { SessionID } from "./session-id.js" import { SessionMessage } from "./session-message.js" @@ -17,6 +17,13 @@ export const UserPayload = Schema.Struct({ metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional), }).annotate({ identifier: "Session.Inbox.UserPayload" }) +/** A {@link UserPayload} whose files may stay on the server, for clients that opt into `experimental-media-uri`. */ +export interface MediaUserPayload extends Schema.Schema.Type {} +export const MediaUserPayload = Schema.Struct({ + ...UserPayload.fields, + files: Schema.Array(Attachment).pipe(optional), +}).annotate({ identifier: "Session.Inbox.MediaUserPayload" }) + export interface SyntheticPayload extends Schema.Schema.Type {} export const SyntheticPayload = Schema.Struct({ text: Schema.String, @@ -58,6 +65,12 @@ const Enqueued = { export interface User extends Schema.Schema.Type {} export const User = Schema.Struct({ ...Enqueued, ...UserItem.fields }).annotate({ identifier: "Session.Inbox.User" }) +/** A {@link User} whose files may stay on the server, for clients that opt into `experimental-media-uri`. */ +export interface MediaUser extends Schema.Schema.Type {} +export const MediaUser = Schema.Struct({ ...User.fields, payload: MediaUserPayload }).annotate({ + identifier: "Session.Inbox.MediaUser", +}) + export interface Synthetic extends Schema.Schema.Type {} export const Synthetic = Schema.Struct({ ...Enqueued, ...SyntheticItem.fields }).annotate({ identifier: "Session.Inbox.Synthetic", @@ -76,3 +89,10 @@ export const Info = Schema.Union([User, Synthetic, Compaction, Move]).pipe( Schema.annotate({ identifier: "Session.Inbox.Info" }), ) export type Info = typeof Info.Type + +/** {@link Info} with {@link MediaUser} for clients that opt into `experimental-media-uri`. */ +export const MediaInfo = Schema.Union([MediaUser, Synthetic, Compaction, Move]).pipe( + Schema.toTaggedUnion("type"), + Schema.annotate({ identifier: "Session.Inbox.MediaInfo" }), +) +export type MediaInfo = typeof MediaInfo.Type diff --git a/packages/schema/src/session-message.ts b/packages/schema/src/session-message.ts index b3fbaa385761..639ad0472eaf 100644 --- a/packages/schema/src/session-message.ts +++ b/packages/schema/src/session-message.ts @@ -6,7 +6,7 @@ import { Content } from "./tool.js" import { Location } from "./location.js" import { Model } from "./model.js" import { Project } from "./project.js" -import { Prompt } from "./prompt.js" +import { Attachment, Prompt } from "./prompt.js" import { DateTimeUtcFromMillis, PositiveInt, RelativePath, statics } from "./schema.js" import { ascending } from "./identifier.js" import { Shell as ShellSchema } from "./shell.js" @@ -83,6 +83,13 @@ export const User = Schema.Struct({ type: Schema.tag("user"), }).annotate({ identifier: "Session.Message.User" }) +/** A {@link User} whose files may stay on the server, for clients that opt into `experimental-media-uri`. */ +export interface MediaUser extends Schema.Schema.Type {} +export const MediaUser = Schema.Struct({ + ...User.fields, + files: Schema.Array(Attachment).pipe(optional), +}).annotate({ identifier: "Session.Message.MediaUser" }) + export interface Synthetic extends Schema.Schema.Type {} export const Synthetic = Schema.Struct({ ...Base, @@ -320,3 +327,19 @@ export type Info = | Compaction | Idle export type Type = Info["type"] + +/** {@link Info} with {@link MediaUser} for clients that opt into `experimental-media-uri`. */ +export const MediaInfo = Schema.Union([ + AgentSelected, + ModelSelected, + LocationSwitched, + MediaUser, + Synthetic, + System, + Skill, + Shell, + Assistant, + Compaction, + Idle, +]).annotate({ identifier: "Session.Message.MediaInfo" }) +export type MediaInfo = Exclude | MediaUser diff --git a/packages/server/src/handlers/media.ts b/packages/server/src/handlers/media.ts new file mode 100644 index 000000000000..009a700a85f3 --- /dev/null +++ b/packages/server/src/handlers/media.ts @@ -0,0 +1,46 @@ +import { Database } from "@opencode/core/database/database" +import { SessionBlob } from "@opencode/core/session/blob" +import { FileNotFoundError } from "@opencode/protocol/errors" +import { Effect } from "effect" +import { type HttpServerRequest, HttpServerResponse } from "effect/http" + +const INLINE_IMAGE_MIMES = new Set(["image/png", "image/jpeg", "image/gif", "image/webp"]) +const CACHE_CONTROL = "private, max-age=31536000, immutable" +const SECURITY_HEADERS = { + "x-content-type-options": "nosniff", + "content-security-policy": "default-src 'none'; sandbox", +} + +/** `media.get` in the `server.experimental` group; serves only blobs a message or inbox item still references. */ +export const makeMediaGet = Effect.gen(function* () { + const { db } = yield* Database.Service + return (hash: string, request: HttpServerRequest.HttpServerRequest) => + Effect.gen(function* () { + const path = `/api/experimental/media/${hash}` + const blob = yield* SessionBlob.read(db, hash) + if (!blob) return yield* new FileNotFoundError({ path, message: `Media blob not found: ${hash}` }) + const etag = `"${hash}"` + const inline = INLINE_IMAGE_MIMES.has(blob.mime) + const headers = { + "cache-control": CACHE_CONTROL, + etag, + ...SECURITY_HEADERS, + ...(inline ? {} : { "content-disposition": "attachment" }), + } + if (matchesIfNoneMatch(request.headers["if-none-match"], etag)) + return HttpServerResponse.empty({ status: 304, headers }) + return HttpServerResponse.uint8Array(blob.data, { + status: 200, + contentType: inline ? blob.mime : "application/octet-stream", + headers, + }) + }) +}) + +function matchesIfNoneMatch(ifNoneMatch: string | undefined, etag: string) { + if (ifNoneMatch === undefined) return false + return ifNoneMatch.split(",").some((tag) => { + const trimmed = tag.trim() + return trimmed === "*" || trimmed.replace(/^w\//i, "") === etag + }) +} diff --git a/packages/server/src/handlers/message.ts b/packages/server/src/handlers/message.ts index 5328c7a98b08..5956a8f1512e 100644 --- a/packages/server/src/handlers/message.ts +++ b/packages/server/src/handlers/message.ts @@ -1,9 +1,12 @@ import { SessionMessage } from "@opencode/core/session/message" import { Session } from "@opencode/core/session" import { Effect, Option, Schema } from "effect" +import { HttpServerResponse } from "effect/http" import { HttpApiBuilder } from "effect/http-api" import { Api } from "../api" +import { wantsMediaRefs } from "../media-uri" import { InvalidCursorError } from "@opencode/protocol/errors" +import { SessionMediaMessagesResponse } from "@opencode/protocol/groups/message" import { failedMessageDecode, missingSession } from "./session-error" const DefaultMessagesLimit = 50 @@ -17,7 +20,7 @@ const Cursor = Schema.Struct({ const decodeCursor = Schema.decodeUnknownOption(Schema.fromJsonString(Cursor)) const cursor = { - encode(message: SessionMessage.Info, order: "asc" | "desc", direction: "previous" | "next") { + encode(message: { readonly id: SessionMessage.ID }, order: "asc" | "desc", direction: "previous" | "next") { return Buffer.from(JSON.stringify({ id: message.id, order, direction })).toString("base64url") }, decode(input: string) { @@ -35,31 +38,44 @@ export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handl if (ctx.query.cursor && ctx.query.order !== undefined) return yield* new InvalidCursorError({ message: "Cursor cannot be combined with order" }) const decoded = ctx.query.cursor - ? Option.getOrUndefined(cursor.decode(ctx.query.cursor)) ?? - (yield* new InvalidCursorError({ message: "Invalid cursor" })) + ? (Option.getOrUndefined(cursor.decode(ctx.query.cursor)) ?? + (yield* new InvalidCursorError({ message: "Invalid cursor" }))) : undefined const order = decoded?.order ?? ctx.query.order ?? "desc" + const input = { + sessionID: ctx.params.sessionID, + limit: ctx.query.limit ?? DefaultMessagesLimit, + order, + type: ctx.query.type, + cursor: decoded ? { id: decoded.id, direction: decoded.direction } : undefined, + } + const page = (messages: ReadonlyArray) => { + const first = messages[0] + const last = messages.at(-1) + return { + data: messages, + cursor: { + previous: first ? cursor.encode(first, order, "previous") : undefined, + next: last ? cursor.encode(last, order, "next") : undefined, + }, + } + } + if (wantsMediaRefs(ctx.request)) { + const messages = yield* session + .messagesWithRefs(input) + .pipe( + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.catchTag("Session.MessageDecodeError", failedMessageDecode), + ) + return yield* HttpServerResponse.schemaJson(SessionMediaMessagesResponse)(page(messages)).pipe(Effect.orDie) + } const messages = yield* session - .messages({ - sessionID: ctx.params.sessionID, - limit: ctx.query.limit ?? DefaultMessagesLimit, - order, - type: ctx.query.type, - cursor: decoded ? { id: decoded.id, direction: decoded.direction } : undefined, - }) + .messages(input) .pipe( Effect.catchTag("Session.NotFoundError", missingSession), Effect.catchTag("Session.MessageDecodeError", failedMessageDecode), ) - const first = messages[0] - const last = messages.at(-1) - return { - data: messages, - cursor: { - previous: first ? cursor.encode(first, order, "previous") : undefined, - next: last ? cursor.encode(last, order, "next") : undefined, - }, - } + return page(messages) }), ) }), diff --git a/packages/server/src/handlers/persistent-pty.ts b/packages/server/src/handlers/persistent-pty.ts index cb452e96e4e2..7f874bfd9c5f 100644 --- a/packages/server/src/handlers/persistent-pty.ts +++ b/packages/server/src/handlers/persistent-pty.ts @@ -12,6 +12,7 @@ import { HttpApiBuilder, HttpApiSchema } from "effect/http-api" import { Socket } from "effect/socket" import { Api } from "../api" import { CorsConfig, isAllowedRequestOrigin } from "../cors" +import { makeMediaGet } from "./media" import { type Outbound, runPtySocket } from "./pty-socket" export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experimental", (handlers) => @@ -19,8 +20,10 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experiment const tickets = yield* PtyTicket.Service const cors = yield* CorsConfig const pty = yield* PersistentPty.Service + const mediaGet = yield* makeMediaGet return handlers + .handleRaw("media.get", (ctx) => mediaGet(ctx.params.hash, ctx.request)) .handle( "persistentPty.read", Effect.fn(function* (ctx) { diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index af7ef13780e6..2b7f8b990337 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -6,8 +6,12 @@ import { SessionTransfer } from "@opencode/core/session/transfer" import { InstructionEntry } from "@opencode/core/session/instruction-entry" import { Form } from "@opencode/core/form" import { DateTime, Effect, Stream } from "effect" +import { HttpServerResponse } from "effect/http" import { HttpApiBuilder, HttpApiSchema } from "effect/http-api" +import { SessionMediaMessageResponse } from "@opencode/protocol/groups/message" +import { SessionMediaInboxResponse } from "@opencode/protocol/groups/session" import { Api } from "../api" +import { wantsMediaRefs } from "../media-uri" import { ConflictError, CommandExecutionError, @@ -519,6 +523,12 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.inbox.list", Effect.fn(function* (ctx) { + if (wantsMediaRefs(ctx.request)) + return yield* HttpServerResponse.schemaJson(SessionMediaInboxResponse)({ + data: yield* session + .inboxWithRefs(ctx.params.sessionID) + .pipe(Effect.catchTag("Session.NotFoundError", missingSession)), + }).pipe(Effect.orDie) return { data: yield* session .inbox(ctx.params.sessionID) @@ -612,8 +622,16 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl "session.message", Effect.fn(function* (ctx) { yield* session.get(ctx.params.sessionID).pipe(Effect.catchTag("Session.NotFoundError", missingSession)) - const message = yield* session.message(ctx.params) - if (message) return { data: message } + if (wantsMediaRefs(ctx.request)) { + const message = yield* session.messageWithRefs(ctx.params) + if (message) + return yield* HttpServerResponse.schemaJson(SessionMediaMessageResponse)({ data: message }).pipe( + Effect.orDie, + ) + } else { + const message = yield* session.message(ctx.params) + if (message) return { data: message } + } return yield* new MessageNotFoundError({ sessionID: ctx.params.sessionID, messageID: ctx.params.messageID, diff --git a/packages/server/src/media-uri.ts b/packages/server/src/media-uri.ts new file mode 100644 index 000000000000..e22c0fbed015 --- /dev/null +++ b/packages/server/src/media-uri.ts @@ -0,0 +1,13 @@ +import type { HttpServerRequest } from "effect/http" + +const CAPABILITIES_HEADER = "x-opencode-capabilities" +const MEDIA_URI_CAPABILITY = "experimental-media-uri" + +/** + * Whether the client opted into `media:` refs. The session's own location decides whether its stored media is + * actually left as refs (`experimental.media_blob`), so this only picks which read to call. + */ +export const wantsMediaRefs = (request: HttpServerRequest.HttpServerRequest) => + request.headers[CAPABILITIES_HEADER]?.split(",").some( + (token) => token.trim().toLowerCase() === MEDIA_URI_CAPABILITY, + ) ?? false diff --git a/packages/server/test/media.test.ts b/packages/server/test/media.test.ts new file mode 100644 index 000000000000..83f19306c361 --- /dev/null +++ b/packages/server/test/media.test.ts @@ -0,0 +1,370 @@ +import { Database as Sqlite } from "bun:sqlite" +import { createHash } from "node:crypto" +import fs from "node:fs/promises" +import path from "node:path" +import { expect } from "bun:test" +import { OpenCode, type SessionMessageInfo } from "@opencode/client" +import { SessionBlob } from "@opencode/core/session/blob" +import { Session } from "@opencode/schema/session" +import { Effect } from "effect" +import { tmpdirScoped } from "../../core/test/fixture/tmpdir" +import { it } from "../../core/test/lib/effect" +import { ServerFetch } from "../src/fetch" + +const png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=" +const text = Buffer.from("workspace file contents").toString("base64") +const svg = Buffer.from('').toString("base64") +const html = Buffer.from("").toString("base64") +const hash = (base64: string) => createHash("sha256").update(Buffer.from(base64, "base64")).digest("hex") +const size = (base64: string) => Buffer.from(base64, "base64").byteLength + +const user: SessionMessageInfo = { + id: "msg_media_user", + type: "user", + text: "See attached", + files: [ + { data: png, mime: "image/png", source: { type: "inline" }, name: "shot.png" }, + { + data: text, + mime: "text/plain", + source: { type: "uri", uri: "file:///workspace/readme.txt" }, + name: "readme.txt", + }, + { data: svg, mime: "image/svg+xml", source: { type: "inline" }, name: "xss.svg" }, + { data: html, mime: "text/html", source: { type: "inline" }, name: "xss.html" }, + ], + time: { created: 100 }, +} + +const assistant: SessionMessageInfo = { + id: "msg_media_assistant", + type: "assistant", + agent: "build", + model: { providerID: "test", id: "test" }, + content: [ + { + type: "tool", + id: "call_read_img", + name: "read", + executed: true, + state: { + status: "completed", + input: { path: "shot.png" }, + content: [{ type: "file", uri: `data:image/png;base64,${png}`, mime: "image/png", name: "shot.png" }], + }, + time: { created: 110, ran: 111, completed: 112 }, + }, + ], + finish: "stop", + time: { created: 110, completed: 120 }, +} + +const serve = (database: string, mediaBlob: boolean) => + ServerFetch.make({ + app: { version: "test" }, + database: { path: database }, + config: { project: false, content: JSON.stringify(mediaBlob ? { experimental: { media_blob: true } } : {}) }, + models: { fetch: false }, + fs: { filewatcher: false }, + }) + +const client = (handler: (request: Request) => Promise, capabilities?: string) => + OpenCode.make({ + baseUrl: "http://opencode.local", + ...(capabilities === undefined ? {} : { headers: { "X-OpenCode-Capabilities": capabilities } }), + fetch: Object.assign((input: string | URL | Request, init?: RequestInit) => handler(new Request(input, init)), { + preconnect: fetch.preconnect, + }), + }) + +// Import stores media inline, so move the imported rows' media into blobs the way the live write path stores them. +const storeAsBlobs = (database: string) => { + const db = new Sqlite(database) + const rows = db + .query< + { id: string; session_id: string; seq: number; data: string }, + [] + >(`SELECT id, session_id, seq, data FROM session_message`) + .all() + for (const row of rows) { + const split = SessionBlob.split(JSON.parse(row.data)) + if (!split.moved) continue + db.run(`UPDATE session_message SET data = ? WHERE id = ?`, [JSON.stringify(split.data), row.id]) + for (const blob of split.blobs) + db.run(`INSERT OR IGNORE INTO blob (hash, mime, bytes, data, time_created) VALUES (?, ?, ?, ?, ?)`, [ + blob.hash, + blob.mime, + blob.bytes, + blob.data, + Date.now(), + ]) + for (const hash of split.hashes) + db.run(`INSERT OR IGNORE INTO blob_ref (session_id, message_id, hash, seq) VALUES (?, ?, ?, ?)`, [ + row.session_id, + row.id, + hash, + row.seq, + ]) + } + db.close() +} + +const media = (base64: string) => `media:${hash(base64)}` +const route = (base64: string) => `http://opencode.local/api/experimental/media/${hash(base64)}` + +it.live("serves stored media as refs only to clients that opt in while experimental.media_blob is on", () => + Effect.gen(function* () { + // Acquired first so it is removed after the servers below are released; Windows can't delete an open db. + const dir = (yield* tmpdirScoped("opencode-media-wire-")).path + yield* Effect.gen(function* () { + const database = path.join(dir, "opencode.db") + const handler = yield* serve(database, true) + const plain = client(handler) + const opted = client(handler, "other, Experimental-Media-URI") + const sessionID = Session.ID.create() + + yield* Effect.promise(async () => { + const template = await plain.session.create({ title: "Media wire" }) + await plain.session.import({ info: { ...template, id: sessionID }, messages: [user, assistant] }) + storeAsBlobs(database) + + // Without the capability, reads stay byte-for-byte what was imported. + expect((await plain.message.list({ sessionID, order: "asc" })).data).toEqual([user, assistant]) + + const refs: unknown = (await opted.message.list({ sessionID, order: "asc" })).data + expect(refs).toEqual([ + { + ...user, + files: [ + { + uri: media(png), + mime: "image/png", + source: { type: "inline" }, + name: "shot.png", + bytes: size(png), + }, + { + uri: media(text), + mime: "text/plain", + source: { type: "uri", uri: "file:///workspace/readme.txt" }, + name: "readme.txt", + bytes: size(text), + }, + { + uri: media(svg), + mime: "image/svg+xml", + source: { type: "inline" }, + name: "xss.svg", + bytes: size(svg), + }, + { + uri: media(html), + mime: "text/html", + source: { type: "inline" }, + name: "xss.html", + bytes: size(html), + }, + ], + }, + { + ...assistant, + content: [ + { + type: "tool", + id: "call_read_img", + name: "read", + executed: true, + state: { + status: "completed", + input: { path: "shot.png" }, + content: [{ type: "file", uri: media(png), mime: "image/png", name: "shot.png" }], + }, + time: { created: 110, ran: 111, completed: 112 }, + }, + ], + }, + ]) + const single: unknown = await opted.session.message.get({ sessionID, messageID: "msg_media_user" }) + expect(single).toMatchObject({ files: [{ uri: media(png) }, { uri: media(text) }, {}, {}] }) + expect(await plain.session.message.get({ sessionID, messageID: "msg_media_user" })).toEqual(user) + expect((await opted.session.export({ sessionID })).messages).toEqual([user, assistant]) + + // The route serves raw bytes, immutable caching and a 304, and never renders active content inline. + const image = await handler(new Request(route(png))) + expect(image.status).toBe(200) + expect(image.headers.get("content-type")).toBe("image/png") + expect(image.headers.get("content-disposition")).toBeNull() + expect(image.headers.get("cache-control")).toBe("private, max-age=31536000, immutable") + expect(image.headers.get("etag")).toBe(`"${hash(png)}"`) + expect(image.headers.get("x-content-type-options")).toBe("nosniff") + expect(image.headers.get("content-security-policy")).toBe("default-src 'none'; sandbox") + expect(Buffer.from(await image.arrayBuffer()).toString("base64")).toBe(png) + for (const unsafe of [svg, html]) { + const response = await handler(new Request(route(unsafe))) + expect(response.headers.get("content-type")).toBe("application/octet-stream") + expect(response.headers.get("content-disposition")).toBe("attachment") + expect(response.headers.get("content-security-policy")).toBe("default-src 'none'; sandbox") + expect(Buffer.from(await response.arrayBuffer()).toString("base64")).toBe(unsafe) + } + const cached = await handler(new Request(route(png), { headers: { "If-None-Match": `W/"${hash(png)}"` } })) + expect(cached.status).toBe(304) + expect(cached.headers.get("etag")).toBe(`"${hash(png)}"`) + expect((await cached.arrayBuffer()).byteLength).toBe(0) + for (const invalid of ["deadbeef", "A".repeat(64), "g".repeat(64)]) + expect((await handler(new Request(`http://opencode.local/api/experimental/media/${invalid}`))).status).toBe( + 400, + ) + expect( + (await handler(new Request(`http://opencode.local/api/experimental/media/${"0".repeat(64)}`))).status, + ).toBe(404) + expect(Buffer.from(await opted.experimental.media.get({ hash: hash(png) })).toString("base64")).toBe(png) + + // A `media:` ref a client received can be sent back as an attachment; pending input keeps it a ref. + const queued = await opted.session.prompt({ + sessionID, + text: "Again", + files: [{ uri: media(png), name: "again.png" }], + delivery: "queue", + resume: false, + }) + const pending: unknown = await opted.session.inbox.list({ sessionID }) + expect(pending).toMatchObject([ + { + id: queued.id, + payload: { + files: [{ uri: media(png), mime: "image/png", name: "again.png", bytes: size(png) }], + }, + }, + ]) + expect(await plain.session.inbox.list({ sessionID })).toMatchObject([ + { id: queued.id, payload: { files: [{ data: png, mime: "image/png", name: "again.png" }] } }, + ]) + await opted.session.inbox.cancel({ sessionID, inboxID: queued.id }) + + // A ref whose blob is gone reaches opted-in clients as the same placeholder inline readers get. + const gone = "e".repeat(64) + const orphan = "d".repeat(64) + const db = new Sqlite(database) + db.run("INSERT INTO blob (hash, mime, bytes, data, time_created) VALUES (?, 'image/png', 3, ?, 1)", [ + orphan, + Buffer.from("abc"), + ]) + db.run( + "INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES (?, ?, 'user', 1001, 1001, 1001, ?)", + [ + "msg_orphan_media", + sessionID, + JSON.stringify({ + text: "orphan", + files: [{ uri: `media:${orphan}`, mime: "image/png", name: "gone.png", source: { type: "inline" } }], + time: { created: 1001 }, + }), + ], + ) + db.run( + "INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES (?, ?, 'user', 1000, 1000, 1000, ?)", + [ + "msg_gone_media", + sessionID, + JSON.stringify({ + text: "gone", + files: [{ uri: `media:${gone}`, mime: "image/png", name: "gone.png", source: { type: "inline" } }], + time: { created: 1000 }, + }), + ], + ) + db.close() + const placeholder = { + data: Buffer.from("[media unavailable: gone.png (image/png)]").toString("base64"), + mime: "text/plain", + source: { type: "inline" }, + name: "gone.png", + } + expect(await opted.session.message.get({ sessionID, messageID: "msg_gone_media" })).toMatchObject({ + files: [placeholder], + }) + expect(await plain.session.message.get({ sessionID, messageID: "msg_gone_media" })).toMatchObject({ + files: [placeholder], + }) + // A stored blob nothing references is not served by the route, so it is not offered as a ref either. + expect((await handler(new Request(route(orphan)))).status).toBe(404) + expect(await opted.session.message.get({ sessionID, messageID: "msg_orphan_media" })).toMatchObject({ + files: [placeholder], + }) + }) + + // Another server over the same database without the flag: opted-in clients get bytes, and the route still + // serves what stored messages reference. + const disabled = yield* serve(database, false) + yield* Effect.promise(async () => { + expect((await disabled(new Request(route(png)))).status).toBe(200) + expect( + (await client(disabled, "experimental-media-uri").message.list({ sessionID, order: "asc" })).data.slice(0, 2), + ).toEqual([user, assistant]) + expect((await handler(new Request(route(png)))).status).toBe(200) + }) + }) + }).pipe(Effect.scoped), +) + +it.live("follows experimental.media_blob from the session's project config and serves only referenced blobs", () => + Effect.gen(function* () { + const dir = (yield* tmpdirScoped("opencode-media-project-")).path + yield* Effect.gen(function* () { + const database = path.join(dir, "opencode.db") + const global = path.join(dir, "global") + const project = path.join(dir, "project") + yield* Effect.promise(async () => { + await fs.mkdir(global, { recursive: true }) + await fs.mkdir(project, { recursive: true }) + await fs.writeFile(path.join(project, "opencode.json"), JSON.stringify({ experimental: { media_blob: true } })) + }) + // The flag lives only in the project's config; the server's cwd and global config leave it off. + const handler = yield* ServerFetch.make({ + app: { version: "test" }, + database: { path: database }, + config: { directory: global, project: true, claude: false }, + models: { fetch: false }, + fs: { filewatcher: false }, + }) + const plain = client(handler) + const opted = client(handler, "experimental-media-uri") + + yield* Effect.promise(async () => { + const session = await plain.session.create({ location: { directory: project } }) + const queued = await opted.session.prompt({ + sessionID: session.id, + text: "Look", + files: [{ uri: `data:image/png;base64,${png}`, name: "shot.png" }], + delivery: "queue", + resume: false, + }) + expect(await opted.session.inbox.list({ sessionID: session.id })).toMatchObject([ + { id: queued.id, payload: { files: [{ uri: media(png), mime: "image/png", name: "shot.png" }] } }, + ]) + expect(await plain.session.inbox.list({ sessionID: session.id })).toMatchObject([ + { id: queued.id, payload: { files: [{ data: png, mime: "image/png", name: "shot.png" }] } }, + ]) + expect((await handler(new Request(route(png)))).status).toBe(200) + + // A blob no message or inbox item references is unreadable, by route or as a prompt attachment. + const db = new Sqlite(database) + db.run("INSERT INTO blob (hash, mime, bytes, data, time_created) VALUES (?, 'text/plain', ?, ?, 1)", [ + hash(text), + size(text), + Buffer.from(text, "base64"), + ]) + db.close() + expect((await handler(new Request(route(text)))).status).toBe(404) + const attached = await opted.session + .prompt({ sessionID: session.id, text: "Steal", files: [{ uri: media(text) }], resume: false }) + .then( + () => undefined, + (error: unknown) => error, + ) + expect(attached).toBeDefined() + expect(await plain.session.inbox.list({ sessionID: session.id })).toHaveLength(1) + }) + }) + }).pipe(Effect.scoped), +)