feat(agent): add authenticated durable uploads to backend and web
This commit is contained in:
@@ -0,0 +1,117 @@
|
||||
/** Agent descriptors are independent of ordinary chat's inline attachment schema. */
|
||||
export type AgentAttachment = {
|
||||
id: string; kind: "image" | "text" | "file"; filename: string; mimeType: string;
|
||||
sizeBytes: number; sha256: string; workspacePath?: string;
|
||||
};
|
||||
export type AgentAttachmentCapability = {
|
||||
version: 1; maxCount: number; maxTotalBytes: number; maxImageBytes: number; maxFileBytes: number;
|
||||
imageMimeTypes: string[]; quotaBytes: number; usedBytes: number; remainingBytes: number;
|
||||
maxConcurrentUploads: number; maxConcurrentDecodes: number;
|
||||
};
|
||||
export function parseAgentAttachment(value: unknown): AgentAttachment {
|
||||
const item = value as AgentAttachment | null;
|
||||
if (!item || !/^att_[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/.test(item.id)
|
||||
|| !["image", "text", "file"].includes(item.kind) || typeof item.filename !== "string" || !item.filename
|
||||
|| typeof item.mimeType !== "string" || !Number.isSafeInteger(item.sizeBytes) || item.sizeBytes < 0 || item.sizeBytes > 8 * 1024 * 1024
|
||||
|| typeof item.sha256 !== "string" || !/^[a-f0-9]{64}$/.test(item.sha256)
|
||||
|| (item.workspacePath !== undefined && (typeof item.workspacePath !== "string" || !item.workspacePath.startsWith("/"))))
|
||||
throw new Error("Agent returned an invalid attachment descriptor.");
|
||||
return { id: item.id, kind: item.kind, filename: item.filename, mimeType: item.mimeType, sizeBytes: item.sizeBytes, sha256: item.sha256,
|
||||
...(item.workspacePath ? { workspacePath: item.workspacePath } : {}) };
|
||||
}
|
||||
export function agentAttachments(metadata: unknown): AgentAttachment[] {
|
||||
const values = (metadata as { agentAttachments?: unknown } | null)?.agentAttachments;
|
||||
if (!Array.isArray(values) || values.length > 8) return [];
|
||||
try { return values.map(parseAgentAttachment); } catch { return []; }
|
||||
}
|
||||
export function attachmentCaption(content: string, files: AgentAttachment[]): string {
|
||||
return content.trim() ? content : `Attached: ${files.map(file => file.filename).join(", ")}.`;
|
||||
}
|
||||
export function attachmentSize(bytes: number): string { return bytes < 1024 ? `${bytes} B` : bytes < 1024 ** 2 ? `${Math.ceil(bytes / 1024)} KiB` : `${(bytes / 1024 ** 2).toFixed(1)} MiB`; }
|
||||
export const AGENT_ATTACHMENT_DRAFT_KEY = "sybil.agent.attachment-draft.v1";
|
||||
export type AgentUpload = {
|
||||
uploadId: string; filename: string; sizeBytes: number; mimeType: string; sha256?: string; file?: File;
|
||||
status: "waiting" | "uploading" | "ready" | "failed" | "reselect"; attachment?: AgentAttachment; error?: string;
|
||||
};
|
||||
export type AgentUploadStorage = { read(): unknown; write(rows: Omit<AgentUpload, "file">[]): void };
|
||||
|
||||
/** One active transfer. Durable identities/descriptors survive navigation and reload; bytes stay in memory. */
|
||||
export function createAgentUploadQueue(options: {
|
||||
capability(): AgentAttachmentCapability | undefined;
|
||||
upload(file: File, uploadId: string, signal: AbortSignal): Promise<AgentAttachment>;
|
||||
changed(): void; storage?: AgentUploadStorage; newId?: () => string;
|
||||
}) {
|
||||
const rows = new Map<string, AgentUpload>();
|
||||
let running: { id: string; controller: AbortController } | undefined, disposed = false;
|
||||
const persist = () => options.storage?.write([...rows.values()].map(({ file: _file, ...row }) => ({ ...row })));
|
||||
const publish = () => { if (!disposed) options.changed(); };
|
||||
const save = () => { if (disposed) return; try { persist(); } catch { /* Pre-transfer persist below is mandatory. Already-uploaded identities remain in memory. */ } publish(); };
|
||||
const restored = options.storage?.read();
|
||||
if (Array.isArray(restored)) for (const value of restored.slice(0, 8)) {
|
||||
if (!value || typeof value.uploadId !== "string" || !/^[a-f0-9-]{36}$/i.test(value.uploadId)
|
||||
|| typeof value.filename !== "string" || typeof value.mimeType !== "string" || !Number.isSafeInteger(value.sizeBytes) || value.sizeBytes < 0 || value.sizeBytes > 8 * 1024 * 1024) continue;
|
||||
try {
|
||||
const attachment = value.attachment ? parseAgentAttachment(value.attachment) : undefined;
|
||||
rows.set(value.uploadId, { uploadId: value.uploadId, filename: value.filename, sizeBytes: value.sizeBytes, mimeType: value.mimeType,
|
||||
...(typeof value.sha256 === "string" && /^[a-f0-9]{64}$/.test(value.sha256) ? { sha256: value.sha256 } : {}),
|
||||
...(attachment ? { attachment, status: "ready" } : { status: "reselect", error: "Reselect the same file to resume this upload. Its retry ID has been retained." }) });
|
||||
} catch { /* Ignore invalid storage, never turn it into a server path/reference. */ }
|
||||
}
|
||||
const digest = async (file: File) => [...new Uint8Array(await crypto.subtle.digest("SHA-256", await file.arrayBuffer()))].map(n => n.toString(16).padStart(2, "0")).join("");
|
||||
const pump = () => {
|
||||
if (disposed || running) return;
|
||||
const row = [...rows.values()].find(row => row.status === "waiting" && row.file);
|
||||
if (!row) return;
|
||||
const controller = new AbortController(); running = { id: row.uploadId, controller }; row.status = "uploading"; publish();
|
||||
void (async () => {
|
||||
row.sha256 ||= await digest(row.file!);
|
||||
if (rows.get(row.uploadId) !== row || controller.signal.aborted) return;
|
||||
// Persist the upload identity AND original byte hash before any network request.
|
||||
persist();
|
||||
const value = await options.upload(row.file!, row.uploadId, controller.signal);
|
||||
if (rows.get(row.uploadId) !== row || controller.signal.aborted) return;
|
||||
const attachment = parseAgentAttachment(value);
|
||||
if (attachment.sha256 !== row.sha256 || attachment.sizeBytes !== row.sizeBytes || attachment.filename !== row.filename.normalize("NFC"))
|
||||
throw new Error("The uploaded file receipt does not match this file. Retry to confirm.");
|
||||
row.attachment = attachment; row.status = "ready"; delete row.error;
|
||||
})().catch(error => {
|
||||
if (rows.get(row.uploadId) !== row || controller.signal.aborted) return;
|
||||
row.status = "failed"; row.error = error instanceof Error ? error.message : String(error);
|
||||
}).finally(() => { running = undefined; save(); pump(); });
|
||||
};
|
||||
return {
|
||||
add(files: Iterable<File>) {
|
||||
const cap = options.capability(); if (!cap || cap.version !== 1) throw new Error("Attachments are not available on this Agent.");
|
||||
const selected = Array.from(files);
|
||||
if (rows.size + selected.length > Math.min(8, cap.maxCount)) throw new Error("Attach at most 8 files per message.");
|
||||
if ([...rows.values()].reduce((n, r) => n + r.sizeBytes, 0) + selected.reduce((n, f) => n + f.size, 0) > Math.min(32 * 1024 * 1024, cap.maxTotalBytes))
|
||||
throw new Error("Attachments exceed 32 MiB per message.");
|
||||
for (const file of selected) {
|
||||
const image = cap.imageMimeTypes.includes(file.type);
|
||||
const limit = image ? Math.min(6 * 1024 * 1024, cap.maxImageBytes) : Math.min(8 * 1024 * 1024, cap.maxFileBytes);
|
||||
if (file.size > limit) throw new Error(`${file.name} exceeds the ${image ? "6" : "8"} MiB file limit.`);
|
||||
}
|
||||
const added = selected.map(file => ({ uploadId: (options.newId || (() => crypto.randomUUID()))(), file, filename: file.name, sizeBytes: file.size, mimeType: file.type, status: "waiting" as const }));
|
||||
for (const row of added) rows.set(row.uploadId, row);
|
||||
try { persist(); } catch (error) { for (const row of added) rows.delete(row.uploadId); throw error; }
|
||||
publish(); pump();
|
||||
},
|
||||
async reselect(id: string, file: File) {
|
||||
const row = rows.get(id); if (!row || row.status !== "reselect") return;
|
||||
if (file.name !== row.filename || file.size !== row.sizeBytes || file.type !== row.mimeType) throw new Error("Choose the same filename, type and size to resume this upload.");
|
||||
const hash = await digest(file);
|
||||
if (row.sha256 && row.sha256 !== hash) throw new Error("The selected file has different contents. Choose the original file or remove this attachment.");
|
||||
if (rows.get(id) !== row) return;
|
||||
row.sha256 = hash; row.file = file; row.status = "waiting"; delete row.error; persist(); publish(); pump();
|
||||
},
|
||||
remove(id: string) { const row = rows.get(id); rows.delete(id); try { persist(); } catch (error) { if (row) rows.set(id, row); throw error; } if (running?.id === id) running.controller.abort(); publish(); },
|
||||
retry(id: string) { const row = rows.get(id); if (row?.status === "failed") { row.status = row.file ? "waiting" : "reselect"; delete row.error; save(); pump(); } },
|
||||
snapshot(): AgentUpload[] { return [...rows.values()].map(row => ({ ...row })); },
|
||||
ready(): AgentAttachment[] {
|
||||
if ([...rows.values()].some(row => row.status !== "ready")) throw new Error("Wait for uploads, reselect or retry failed files, or remove them before sending.");
|
||||
return [...rows.values()].map(row => row.attachment!);
|
||||
},
|
||||
clear() { rows.clear(); running?.controller.abort(); save(); },
|
||||
dispose() { disposed = true; running?.controller.abort(); try { persist(); } catch { /* Existing stored retry IDs remain usable. */ } },
|
||||
};
|
||||
}
|
||||
@@ -1,10 +1,11 @@
|
||||
import type { Message } from "./api";
|
||||
import { agentAttachments, attachmentCaption, parseAgentAttachment, type AgentAttachment } from "./agent-attachments.ts";
|
||||
|
||||
export const AGENT_OUTBOX_STORAGE_KEY = "sybil.agent.optimistic-outbox.v1";
|
||||
export type AgentDeliveryStatus = "sending" | "queued" | "complete" | "failed" | "unconfirmed";
|
||||
export type AgentOutboxRow = {
|
||||
id: string; requestId: string; content: string; createdAt: string;
|
||||
status: AgentDeliveryStatus; error?: string; seenOnServer: boolean; confirmed: boolean;
|
||||
attachments?: AgentAttachment[]; status: AgentDeliveryStatus; error?: string; seenOnServer: boolean; confirmed: boolean;
|
||||
};
|
||||
export type AgentOutboxStorage = {
|
||||
read(): unknown; write(rows: AgentOutboxRow[]): void;
|
||||
@@ -20,7 +21,7 @@ const validRequestId = (value: unknown): value is string => typeof value === "st
|
||||
|
||||
/** Local rows use the server's identity from their first render through final echo. */
|
||||
export function createAgentOptimisticOutbox(options: {
|
||||
submit(content: string, requestId: string): Promise<Receipt>;
|
||||
submit(content: string, requestId: string, attachments?: AgentAttachment[]): Promise<Receipt>;
|
||||
storage: AgentOutboxStorage;
|
||||
changed(): void;
|
||||
newId?: () => string;
|
||||
@@ -40,12 +41,14 @@ export function createAgentOptimisticOutbox(options: {
|
||||
const restored = options.storage.read();
|
||||
if (Array.isArray(restored)) {
|
||||
for (const value of restored.slice(0, 100)) {
|
||||
if (!value || typeof value.content !== "string" || !value.content.trim() || !validRequestId(value.requestId)) continue;
|
||||
if (!value || typeof value.content !== "string" || (!value.content.trim() && !value.attachments?.length) || !validRequestId(value.requestId)) continue;
|
||||
let attachments: AgentAttachment[] = [];
|
||||
try { if (value.attachments !== undefined) { if (!Array.isArray(value.attachments) || value.attachments.length > 8) continue; attachments = value.attachments.map(parseAgentAttachment); } } catch { continue; }
|
||||
const time = typeof value.createdAt === "string" && Number.isFinite(Date.parse(value.createdAt)) ? value.createdAt : timestamp();
|
||||
latestTime = Math.max(latestTime, Date.parse(time));
|
||||
const status: AgentDeliveryStatus = ["failed", "queued", "complete", "unconfirmed"].includes(value.status) ? value.status : "unconfirmed";
|
||||
const row: AgentOutboxRow = { id: idFor(value.requestId), requestId: value.requestId, content: value.content, createdAt: time,
|
||||
status, seenOnServer: Boolean(value.seenOnServer), confirmed: false,
|
||||
...(attachments.length ? { attachments } : {}), status, seenOnServer: Boolean(value.seenOnServer), confirmed: false,
|
||||
...(typeof value.error === "string" && value.error ? { error: value.error.slice(0, 240) } : {}) };
|
||||
rows.set(row.id, row);
|
||||
}
|
||||
@@ -78,7 +81,8 @@ export function createAgentOptimisticOutbox(options: {
|
||||
};
|
||||
const saveBestEffort = () => { try { persist(); } catch { /* Existing receipt IDs stay in memory; acknowledged messages remain sent. */ } };
|
||||
const publish = () => options.changed();
|
||||
const dispatch = (row: AgentOutboxRow): Promise<void> => {
|
||||
const matches = (row: AgentOutboxRow, message: Message) => !row.attachments?.length || JSON.stringify(agentAttachments(message.metadata).map(file => file.id)) === JSON.stringify(row.attachments.map(file => file.id));
|
||||
const dispatch = (row: AgentOutboxRow, onPersisted?: () => void): Promise<void> => {
|
||||
const running = inFlight.get(row.id);
|
||||
if (running) return running;
|
||||
if (row.confirmed || row.seenOnServer) return Promise.resolve();
|
||||
@@ -92,9 +96,10 @@ export function createAgentOptimisticOutbox(options: {
|
||||
row.status = "failed"; row.error = errorText(error); publish();
|
||||
return Promise.reject(error);
|
||||
}
|
||||
onPersisted?.();
|
||||
// Both durable local storage and the visible row exist before POST begins.
|
||||
publish();
|
||||
const pending = Promise.resolve().then(() => options.submit(row.content, row.requestId)).then(receipt => {
|
||||
const pending = Promise.resolve().then(() => options.submit(row.content, row.requestId, row.attachments)).then(receipt => {
|
||||
if (receipt.messageId && receipt.messageId !== row.id) throw new Error("Agent returned a different message identity.");
|
||||
if (!row.confirmed) {
|
||||
row.seenOnServer = true;
|
||||
@@ -115,20 +120,21 @@ export function createAgentOptimisticOutbox(options: {
|
||||
// their newly assigned identity/time must survive the next reload too.
|
||||
saveBestEffort();
|
||||
return {
|
||||
send(content: string): Promise<void> {
|
||||
if (!content.trim()) return Promise.reject(new Error("Enter a message."));
|
||||
send(content: string, attachments: AgentAttachment[] = [], onPersisted?: () => void): Promise<void> {
|
||||
if (!content.trim() && !attachments.length) return Promise.reject(new Error("Enter a message."));
|
||||
const requestId = newId();
|
||||
const row: AgentOutboxRow = { id: idFor(requestId), requestId, content, createdAt: timestamp(), status: "sending", seenOnServer: false, confirmed: false };
|
||||
const row: AgentOutboxRow = { id: idFor(requestId), requestId, content, ...(attachments.length ? { attachments: attachments.map(parseAgentAttachment) } : {}), createdAt: timestamp(), status: "sending", seenOnServer: false, confirmed: false };
|
||||
rows.set(row.id, row);
|
||||
return dispatch(row);
|
||||
return dispatch(row, onPersisted);
|
||||
},
|
||||
retry(messageId: string): Promise<void> {
|
||||
const row = rows.get(messageId);
|
||||
return row ? dispatch(row) : Promise.reject(new Error("This message is no longer pending."));
|
||||
},
|
||||
reconcile(room: RoomMessages): void {
|
||||
const committed = new Map(room.messages.map(message => [message.id, message]));
|
||||
const queued = new Set((room.activity?.queuedMessages || []).map(message => message.id));
|
||||
const compatible = (message: Message) => { const row = rows.get(message.id); return !row || matches(row, message); };
|
||||
const committed = new Map(room.messages.filter(compatible).map(message => [message.id, message]));
|
||||
const queued = new Set((room.activity?.queuedMessages || []).filter(compatible).map(message => message.id));
|
||||
let changed = false;
|
||||
for (const row of rows.values()) {
|
||||
if (committed.has(row.id)) {
|
||||
@@ -143,7 +149,7 @@ export function createAgentOptimisticOutbox(options: {
|
||||
if (changed) { saveBestEffort(); publish(); }
|
||||
},
|
||||
display(serverMessages: readonly Message[]): Message[] {
|
||||
const displayed = new Map(serverMessages.map(message => {
|
||||
const displayed = new Map(serverMessages.filter(message => { const row = rows.get(message.id); return !row || matches(row, message); }).map(message => {
|
||||
const row = rows.get(message.id);
|
||||
if (!row) return [message.id, message] as const;
|
||||
const serverInfo = info(message);
|
||||
@@ -155,8 +161,8 @@ export function createAgentOptimisticOutbox(options: {
|
||||
const echo = confirmedEchoes.get(row.id);
|
||||
if (row.confirmed && !echo) continue;
|
||||
displayed.set(row.id, echo ? { ...echo, createdAt: row.createdAt } : {
|
||||
id: row.id, createdAt: row.createdAt, role: "user", content: row.content, name: "buzzert",
|
||||
metadata: { kind: "message", status: row.status, channel: "sybil", requestId: row.requestId, ...(row.error ? { error: row.error } : {}) },
|
||||
id: row.id, createdAt: row.createdAt, role: "user", content: row.attachments?.length ? attachmentCaption(row.content, row.attachments) : row.content, name: "buzzert",
|
||||
metadata: { kind: "message", status: row.status, channel: "sybil", requestId: row.requestId, ...(row.attachments?.length ? { agentAttachments: row.attachments } : {}), ...(row.error ? { error: row.error } : {}) },
|
||||
});
|
||||
}
|
||||
return [...displayed.values()].sort((a, b) => a.createdAt.localeCompare(b.createdAt));
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { Message } from "./api";
|
||||
import { parseAgentAttachment, type AgentAttachment, type AgentAttachmentCapability } from "./agent-attachments.ts";
|
||||
|
||||
export type AgentActivity = {
|
||||
state: "idle" | "thinking" | "tool" | "waiting" | "error";
|
||||
@@ -16,7 +17,7 @@ export type AgentMessagesPage = {
|
||||
};
|
||||
export type AgentDescriptor = {
|
||||
id: string; title: string; model: string; activity: AgentActivity;
|
||||
websocket: string; protocol: string;
|
||||
websocket: string; protocol: string; capabilities?: { attachments?: AgentAttachmentCapability };
|
||||
};
|
||||
export type AgentReadState = { cursor: string; updatedAt: string | null };
|
||||
export type AgentRoomState = {
|
||||
@@ -126,7 +127,7 @@ export function agentAuthFrame(token: string | null, cursor: string): string {
|
||||
return JSON.stringify({ type: "auth", token, after: normalizeAgentCursor(cursor) });
|
||||
}
|
||||
|
||||
export type AgentSubmission = { id: number; status: "queued" | "placed" | "done" | "unanswered"; messageId: string };
|
||||
export type AgentSubmission = { id: number; status: "queued" | "placed" | "done" | "unanswered"; messageId: string; attachmentIds?: string[] };
|
||||
export type AgentRoomClientOptions = { baseUrl?: string; token?: () => string | null; fetch?: typeof globalThis.fetch };
|
||||
export class AgentRoomClient {
|
||||
private readonly baseUrl: string;
|
||||
@@ -141,7 +142,7 @@ export class AgentRoomClient {
|
||||
const headers = new Headers(init.headers);
|
||||
const token = this.token();
|
||||
if (token) headers.set("Authorization", `Bearer ${token}`);
|
||||
if (init.body !== undefined) headers.set("Content-Type", "application/json");
|
||||
if (init.body !== undefined && !(init.body instanceof FormData)) headers.set("Content-Type", "application/json");
|
||||
const response = await this.fetcher(`${this.baseUrl}/v1/agent${path}`, { ...init, headers });
|
||||
let value: unknown;
|
||||
try { value = await response.json(); }
|
||||
@@ -160,15 +161,30 @@ export class AgentRoomClient {
|
||||
markRead(cursor: string, signal?: AbortSignal): Promise<AgentReadState> {
|
||||
return this.request("/read-state", { method: "PUT", body: JSON.stringify({ cursor: normalizeAgentCursor(cursor) }), signal });
|
||||
}
|
||||
async submit(content: string, requestId: string): Promise<AgentSubmission> {
|
||||
const receipt = object(await this.request<unknown>("/messages", { method: "POST", body: JSON.stringify({ content, requestId }) }));
|
||||
async submit(content: string, requestId: string, attachments: AgentAttachment[] = []): Promise<AgentSubmission> {
|
||||
const receipt = object(await this.request<unknown>("/messages", { method: "POST", body: JSON.stringify({ content, requestId, ...(attachments.length ? { attachments: attachments.map(({ id }) => ({ id })) } : {}) }) }));
|
||||
// An HTTP success alone does not acknowledge delivery. Keep the original
|
||||
// outbox receipt key retryable if a proxy or upstream returns malformed data.
|
||||
if (!receipt || typeof receipt.id !== "number" || !Number.isSafeInteger(receipt.id) || receipt.id <= 0
|
||||
|| typeof receipt.status !== "string" || !["queued", "placed", "done", "unanswered"].includes(receipt.status)
|
||||
|| receipt.messageId !== `inbound:sybil:${requestId}`) throw new Error("Agent API returned an invalid message receipt. Retry to confirm delivery.");
|
||||
|| receipt.messageId !== `inbound:sybil:${requestId}`
|
||||
|| (attachments.length && JSON.stringify(receipt.attachmentIds) !== JSON.stringify(attachments.map(file => file.id)))) throw new Error("Agent API returned an invalid message receipt. Retry to confirm delivery.");
|
||||
return receipt as AgentSubmission;
|
||||
}
|
||||
async upload(file: File, uploadId: string, signal?: AbortSignal): Promise<AgentAttachment> {
|
||||
const body = new FormData(); body.append("uploadId", uploadId); body.append("file", file, file.name);
|
||||
const value = await this.request<{ attachment: unknown }>("/attachments", { method: "POST", body, signal });
|
||||
return parseAgentAttachment(value.attachment);
|
||||
}
|
||||
async download(attachment: AgentAttachment, signal?: AbortSignal): Promise<Blob> {
|
||||
const file = parseAgentAttachment(attachment), headers = new Headers(), token = this.token();
|
||||
if (token) headers.set("Authorization", `Bearer ${token}`);
|
||||
const response = await this.fetcher(`${this.baseUrl}/v1/agent/attachments/${encodeURIComponent(file.id)}/content`, { headers, signal, cache: "no-store", redirect: "error" });
|
||||
if (!response.ok) { const error = await response.json().catch(() => null); throw new Error(error?.message || "Attachment download failed."); }
|
||||
const blob = await response.blob();
|
||||
if (blob.size !== file.sizeBytes || blob.size > 8 * 1024 * 1024) throw new Error("Attachment download was incomplete.");
|
||||
return blob;
|
||||
}
|
||||
websocketUrl(pageUrl: string): string {
|
||||
const url = new URL(`${this.baseUrl}/v1/agent/ws`, pageUrl);
|
||||
url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
|
||||
|
||||
Reference in New Issue
Block a user