From a3b8f4dd37d943b079b6e58b1e72aeab3238a02f Mon Sep 17 00:00:00 2001 From: Harshith Mullapudi Date: Sat, 24 Oct 2026 21:17:21 +0000 Subject: [PATCH] feat(host-service): sweep the attachment store by last use, and keep fileId on cloud copies Part of the attachments plan (PR 3: seams 8, 2, and the fileId carry-through of seam 3). The host's attachment store had no retention. Nothing deletes chat.db rows and a terminal launch has no session, so there is no session delete to hook. Instead, metadata.json records lastUsedAt. Upload and import set it, and every path that hands an attachment out sets it again: read, agents.run's attached files block, and the ACP resolver. A daily sweep (not in sandboxes) removes entries unused for ATTACHMENT_RETENTION_DAYS. An entry written before this change is stamped, not removed. The check and the rename that takes the entry out of the store run in the same tick, so a hand-out on this event loop cannot land between them. The slow delete runs async, off the loop. A box's cached copy now keeps the fileId it was imported from, in its metadata and in a small .by-file-id index, so the adapters (next PR) can find a cloud part's cached copy without a scan. The first prompt of a cloud launch spreads that metadata into its parts. Without fileId, the first attachments of a cloud session would be the one kind a reader without the box cannot render. MAX_INLINE_ATTACHMENT_BYTES moves to @superset/shared/attachment-limits with MAX_THUMBNAIL_BYTES and ATTACHMENT_RETENTION_DAYS. The host and the shared launch schema use MAX_ATTACHMENTS for their max(10). --- packages/chat/src/protocol/items.ts | 3 + packages/host-service/src/app.ts | 18 +++ .../host-service/src/chat-v3/acpHarnesses.ts | 4 +- .../src/trpc/router/agents/agents.test.ts | 113 ++++++++++++++++++ .../src/trpc/router/agents/agents.ts | 29 +++-- .../router/attachments/attachments.test.ts | 58 ++++++++- .../trpc/router/attachments/attachments.ts | 35 ++++-- .../src/trpc/router/attachments/constants.ts | 8 -- .../src/trpc/router/attachments/index.ts | 2 +- .../src/trpc/router/attachments/storage.ts | 71 ++++++++++- .../src/trpc/router/attachments/sweep.test.ts | 103 ++++++++++++++++ .../src/trpc/router/attachments/sweep.ts | 79 ++++++++++++ .../router/sessions/create-chat-session.ts | 3 +- packages/shared/src/attachment-limits.ts | 14 +++ packages/shared/src/cloud-agent-launch.ts | 3 +- 15 files changed, 510 insertions(+), 33 deletions(-) delete mode 100644 packages/host-service/src/trpc/router/attachments/constants.ts create mode 100644 packages/host-service/src/trpc/router/attachments/sweep.test.ts create mode 100644 packages/host-service/src/trpc/router/attachments/sweep.ts diff --git a/packages/chat/src/protocol/items.ts b/packages/chat/src/protocol/items.ts index 1a0210d97e3..b3f28c37043 100644 --- a/packages/chat/src/protocol/items.ts +++ b/packages/chat/src/protocol/items.ts @@ -28,8 +28,11 @@ export const userContentSchema = z.discriminatedUnion("type", [ z.looseObject({ type: z.literal("attachment"), attachmentId: z.string().min(1), + /** The storage object behind a cloud workspace's attachment; set, it is the identity. */ + fileId: z.string().uuid().optional(), name: z.string(), mimeType: z.string(), + sizeBytes: z.number().int().nonnegative().optional(), }), ]); export type UserContent = z.infer; diff --git a/packages/host-service/src/app.ts b/packages/host-service/src/app.ts index 5158acc6f72..bcfa1c33ef0 100644 --- a/packages/host-service/src/app.ts +++ b/packages/host-service/src/app.ts @@ -56,6 +56,10 @@ import { } from "./terminal-agents"; import { matchesAgentBinding } from "./terminal-agents/matches-agent-binding"; import { appRouter } from "./trpc/router"; +import { + ATTACHMENT_SWEEP_INTERVAL_MS, + sweepAttachmentStore, +} from "./trpc/router/attachments/sweep"; import { gitStatusStore } from "./trpc/router/git/utils/git-status-store"; import { disposeLocalMobileDevices, @@ -385,6 +389,18 @@ export function createApp(options: CreateAppOptions): CreateAppResult { ? null : setInterval(() => void runProjectPurge(), PROJECT_PURGE_INTERVAL_MS); projectPurgeTimer?.unref?.(); + const runAttachmentSweep = () => + sweepAttachmentStore().catch((err) => { + console.warn("[host-service] attachment sweep failed:", err); + }); + const attachmentSweepTimer = + process.env.SUPERSET_HOST_RUN_MODE === "sandbox" + ? null + : setInterval( + () => void runAttachmentSweep(), + ATTACHMENT_SWEEP_INTERVAL_MS, + ); + attachmentSweepTimer?.unref?.(); if (process.env.SUPERSET_HOST_RUN_MODE !== "sandbox") { startLocalMobileDeviceCleanup(db); } @@ -415,6 +431,7 @@ export function createApp(options: CreateAppOptions): CreateAppResult { console.warn("[host-service] archived-workspace reconcile failed:", err); }); await runProjectPurge(); + await runAttachmentSweep(); // Re-share the default account's Claude/Codex config into the selected // provider profiles. Last: it touches no host state the sweeps above // repair, and a slow filesystem must not delay them. @@ -546,6 +563,7 @@ export function createApp(options: CreateAppOptions): CreateAppResult { const ownsDb = options.db === undefined; const dispose = async (): Promise => { if (projectPurgeTimer) clearInterval(projectPurgeTimer); + if (attachmentSweepTimer) clearInterval(attachmentSweepTimer); await disposeWorkspaceTitleJobs(db); // Each step is best-effort and isolated: a throw in one cleanup must // not skip the others, otherwise a flaky `.stop()` could leak the diff --git a/packages/host-service/src/chat-v3/acpHarnesses.ts b/packages/host-service/src/chat-v3/acpHarnesses.ts index 4aa0407db8d..40d63122991 100644 --- a/packages/host-service/src/chat-v3/acpHarnesses.ts +++ b/packages/host-service/src/chat-v3/acpHarnesses.ts @@ -4,7 +4,7 @@ import { dirname, join } from "node:path"; import type { HarnessFactory } from "@superset/chat-runtime"; import { createAcpAdapter } from "@superset/chat-runtime"; import type { HostDb } from "../db"; -import { resolveAttachmentPath } from "../trpc/router/attachments/storage"; +import { handOutAttachment } from "../trpc/router/attachments/storage"; import { captureSessionAccount } from "../trpc/router/usage/session-account/session-account"; import { ACP_HARNESSES } from "./acpCatalogue"; import { resolveAgentCli } from "./agentCli"; @@ -13,7 +13,7 @@ import { storeAgentImage } from "./agentImages"; import type { ChatAgentBridge } from "./chatAgentBridge"; async function resolveAttachment(attachmentId: string) { - const resolved = resolveAttachmentPath(attachmentId); + const resolved = handOutAttachment(attachmentId); if (!resolved) return null; return { path: resolved.path, diff --git a/packages/host-service/src/trpc/router/agents/agents.test.ts b/packages/host-service/src/trpc/router/agents/agents.test.ts index d721232cf08..ecf3cd19356 100644 --- a/packages/host-service/src/trpc/router/agents/agents.test.ts +++ b/packages/host-service/src/trpc/router/agents/agents.test.ts @@ -11,6 +11,12 @@ import type { HostDb } from "../../../db"; import * as schema from "../../../db/schema"; import { TerminalAgentStore } from "../../../terminal-agents"; import { claudeProjectDirName } from "../../../terminal-agents/harness-sessions/claude"; +import { importCloudAttachments } from "../attachments/attachments"; +import { + type AttachmentMetadata, + readAttachmentMetadata, + writeAttachment, +} from "../attachments/storage"; import { setDefaultAccountSelection } from "../usage/default-account"; import { bindResumedSession, @@ -1159,3 +1165,110 @@ describe("continuationTarget", () => { ).toEqual({ terminalId, label: "Team Claude" }); }); }); + +describe("attachments handed to an agent launch", () => { + const workspaceId = "11111111-1111-1111-1111-111111111111"; + let manifestDir: string; + + beforeEach(() => { + manifestDir = mkdtempSync(join(tmpdir(), "superset-agent-attachments-")); + process.env.HOST_MANIFEST_DIR = manifestDir; + }); + + afterEach(() => { + rmSync(manifestDir, { recursive: true, force: true }); + delete process.env.HOST_MANIFEST_DIR; + }); + + function seedClaude(db: HostDb) { + db.insert(schema.hostAgentConfigs) + .values({ + id: "00000000-0000-0000-0000-00000000000a", + presetId: "claude", + label: "Claude", + command: "claude", + argsJson: "[]", + promptTransport: "argv", + promptArgsJson: "[]", + resumeArgsJson: "[]", + forkArgsJson: "[]", + envJson: "{}", + displayOrder: 0, + }) + .run(); + } + + it("a cloud launch's first-prompt part carries the fileId it was imported from", async () => { + const server = Bun.serve({ + port: 0, + fetch: () => new Response("png"), + }); + try { + const fileId = crypto.randomUUID(); + const [imported] = await importCloudAttachments( + { + api: { + attachment: { + resolve: { + mutate: async () => [ + { + fileId, + name: "shot.png", + contentType: "image/png", + sizeBytes: 3, + url: server.url.href, + }, + ], + }, + }, + }, + }, + [fileId], + ); + const db = createTestDb(); + seedClaude(db); + + const target = chatLaunchTarget(db, { + workspaceId, + agent: "claude", + prompt: "look", + surface: "chat", + attachmentIds: [imported?.attachmentId ?? ""], + }); + expect(target?.attachments).toEqual([ + { + attachmentId: imported?.attachmentId ?? "", + fileId, + name: "shot.png", + mimeType: "image/png", + sizeBytes: 3, + }, + ]); + } finally { + server.stop(true); + } + }); + + it("a terminal launch stamps the attachments it lists as used", () => { + const metadata: AttachmentMetadata = { + attachmentId: crypto.randomUUID(), + mediaType: "image/png", + sizeBytes: 1, + createdAt: 0, + lastUsedAt: 0, + }; + writeAttachment(new Uint8Array([1]), metadata); + const db = createTestDb(); + seedClaude(db); + + buildTerminalAgentLaunch(db, { + workspaceId, + agent: "claude", + prompt: "look", + attachmentIds: [metadata.attachmentId], + }); + expect( + readAttachmentMetadata(metadata.attachmentId)?.lastUsedAt, + ).toBeGreaterThan(0); + }); +}); diff --git a/packages/host-service/src/trpc/router/agents/agents.ts b/packages/host-service/src/trpc/router/agents/agents.ts index 549555e9f39..01d697658de 100644 --- a/packages/host-service/src/trpc/router/agents/agents.ts +++ b/packages/host-service/src/trpc/router/agents/agents.ts @@ -44,7 +44,10 @@ import { import { hasHarnessSession } from "../../../terminal-agents/harness-sessions"; import type { HostServiceContext } from "../../../types"; import { protectedProcedure, router } from "../../index"; -import { resolveAttachmentPath } from "../attachments/storage"; +import { + handOutAttachment, + resolveAttachmentPath, +} from "../attachments/storage"; import { createChatSession } from "../sessions/create-chat-session"; import { toTerminalSessionError } from "../terminal/errors"; import { seedAgentFolderTrust } from "../workspace-creation/shared/seed-agent-trust"; @@ -425,7 +428,7 @@ export function buildTerminalAgentLaunch( const resolvedAttachments: Array<{ attachmentId: string; path: string }> = []; for (const attachmentId of input.attachmentIds ?? []) { - const resolved = resolveAttachmentPath(attachmentId); + const resolved = handOutAttachment(attachmentId); if (!resolved) { throw new TRPCError({ code: "NOT_FOUND", @@ -635,13 +638,25 @@ function bindingRunsConfig( return binding.agentId === launchIdentity(config); } +/** + * An imported copy keeps its `fileId` next to the host id: the adapters read + * `attachmentId` today, and `fileId` is what renders the part without this box. + */ +export interface ChatLaunchAttachment { + attachmentId: string; + fileId?: string; + name: string; + mimeType: string; + sizeBytes: number; +} + export function chatLaunchTarget( db: HostDb, input: AgentRunInput, ): { harness: string; label: string; - attachments: Array<{ attachmentId: string; name: string; mimeType: string }>; + attachments: ChatLaunchAttachment[]; } | null { if ( input.surface !== "chat" || @@ -651,18 +666,16 @@ export function chatLaunchTarget( ) { return null; } - const attachments: Array<{ - attachmentId: string; - name: string; - mimeType: string; - }> = []; + const attachments: ChatLaunchAttachment[] = []; for (const attachmentId of input.attachmentIds ?? []) { const resolved = resolveAttachmentPath(attachmentId); if (!resolved) return null; attachments.push({ attachmentId, + ...(resolved.metadata.fileId ? { fileId: resolved.metadata.fileId } : {}), name: resolved.metadata.originalFilename ?? attachmentId, mimeType: resolved.metadata.mediaType, + sizeBytes: resolved.metadata.sizeBytes, }); } const config = resolveHostAgentConfig(db, input.agent); diff --git a/packages/host-service/src/trpc/router/attachments/attachments.test.ts b/packages/host-service/src/trpc/router/attachments/attachments.test.ts index 24ee009c828..5ed19d1ed9b 100644 --- a/packages/host-service/src/trpc/router/attachments/attachments.test.ts +++ b/packages/host-service/src/trpc/router/attachments/attachments.test.ts @@ -2,13 +2,16 @@ import { afterEach, beforeEach, describe, expect, it } from "bun:test"; import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; import { homedir, tmpdir } from "node:os"; import { join } from "node:path"; +import { MAX_INLINE_ATTACHMENT_BYTES } from "@superset/shared/attachment-limits"; import type { HostServiceContext } from "../../../types"; -import { attachmentsRouter } from "./attachments"; -import { MAX_INLINE_ATTACHMENT_BYTES } from "./constants"; +import { attachmentsRouter, importCloudAttachments } from "./attachments"; import { getAttachmentDir, getAttachmentFilePath, getAttachmentsRoot, + handOutAttachmentByFileId, + readAttachmentMetadata, + writeAttachmentMetadata, } from "./storage"; let tempBase: string; @@ -187,6 +190,20 @@ describe("attachmentsRouter.read", () => { }); }); + it("stamps the attachment as used", async () => { + const caller = createCaller(); + const { attachmentId } = await caller.upload({ + data: { kind: "base64", data: PNG_BASE64 }, + mediaType: "image/png", + }); + const metadata = readAttachmentMetadata(attachmentId); + if (!metadata) throw new Error("upload wrote no metadata"); + writeAttachmentMetadata({ ...metadata, lastUsedAt: 0 }); + + await caller.read({ attachmentId }); + expect(readAttachmentMetadata(attachmentId)?.lastUsedAt).toBeGreaterThan(0); + }); + it("reports a missing attachment as NOT_FOUND", async () => { await expect( createCaller().read({ @@ -196,6 +213,43 @@ describe("attachmentsRouter.read", () => { }); }); +describe("importCloudAttachments", () => { + it("records the fileId a copy came from, so a lookup by fileId finds it", async () => { + const server = Bun.serve({ + port: 0, + fetch: () => new Response(Buffer.from(PNG_BASE64, "base64")), + }); + try { + const fileId = crypto.randomUUID(); + const ctx = { + api: { + attachment: { + resolve: { + mutate: async () => [ + { + fileId, + name: "shot.png", + contentType: "image/png", + sizeBytes: 68, + url: server.url.href, + }, + ], + }, + }, + }, + }; + + const [imported] = await importCloudAttachments(ctx, [fileId]); + expect(imported?.fileId).toBe(fileId); + expect(handOutAttachmentByFileId(fileId)?.metadata.attachmentId).toBe( + imported?.attachmentId, + ); + } finally { + server.stop(true); + } + }); +}); + describe("attachmentsRouter.delete", () => { it("removes the attachment directory", async () => { const caller = createCaller(); diff --git a/packages/host-service/src/trpc/router/attachments/attachments.ts b/packages/host-service/src/trpc/router/attachments/attachments.ts index c13505ba6dc..2f094d4df6f 100644 --- a/packages/host-service/src/trpc/router/attachments/attachments.ts +++ b/packages/host-service/src/trpc/router/attachments/attachments.ts @@ -2,6 +2,10 @@ import { randomUUID } from "node:crypto"; import { closeSync, mkdirSync, openSync } from "node:fs"; import { readFile } from "node:fs/promises"; import { join } from "node:path"; +import { + MAX_ATTACHMENTS, + MAX_INLINE_ATTACHMENT_BYTES, +} from "@superset/shared/attachment-limits"; import { assignAttachmentFileName, attachmentNameWithSuffix, @@ -12,13 +16,13 @@ import mimeTypes from "mime-types"; import { z } from "zod"; import { protectedProcedure, router } from "../../index"; import { resolveWorktreePath } from "../git/utils/resolve-worktree"; -import { MAX_INLINE_ATTACHMENT_BYTES } from "./constants"; import { type CloudAttachment, downloadAttachment } from "./download"; import { type AttachmentMetadata, deleteAttachment, + handOutAttachment, + indexAttachmentFileId, prepareAttachmentTarget, - resolveAttachmentPath, writeAttachment, writeAttachmentMetadata, } from "./storage"; @@ -128,12 +132,14 @@ export const attachmentsRouter = router({ }); } + const now = Date.now(); const metadata: AttachmentMetadata = { attachmentId: randomUUID(), mediaType, originalFilename: input.originalFilename, sizeBytes: bytes.length, - createdAt: Date.now(), + createdAt: now, + lastUsedAt: now, }; // Buffer already extends Uint8Array; no need to wrap. @@ -155,7 +161,11 @@ export const attachmentsRouter = router({ * an attachment can legitimately be. */ importFromCloud: protectedProcedure - .input(z.object({ fileIds: z.array(z.string().uuid()).min(1).max(10) })) + .input( + z.object({ + fileIds: z.array(z.string().uuid()).min(1).max(MAX_ATTACHMENTS), + }), + ) .mutation(({ ctx, input }) => importCloudAttachments(ctx, input.fileIds)), /** @@ -167,7 +177,7 @@ export const attachmentsRouter = router({ .input( z.object({ workspaceId: z.string(), - fileIds: z.array(z.string().uuid()).min(1).max(10), + fileIds: z.array(z.string().uuid()).min(1).max(MAX_ATTACHMENTS), }), ) .mutation(async ({ ctx, input }) => { @@ -201,7 +211,7 @@ export const attachmentsRouter = router({ read: protectedProcedure .input(z.object({ attachmentId: z.string().uuid() })) .query(async ({ input }) => { - const resolved = resolveAttachmentPath(input.attachmentId); + const resolved = handOutAttachment(input.attachmentId); if (!resolved) { throw new TRPCError({ code: "NOT_FOUND", @@ -236,6 +246,8 @@ export const attachmentsRouter = router({ export type AttachmentUploadResult = { attachmentId: string; + /** Set when the bytes came from cloud storage: the identity a reader without this host resolves. */ + fileId?: string; originalFilename?: string; mediaType: string; sizeBytes: number; @@ -261,17 +273,22 @@ export async function importCloudAttachments( const resolved = await ctx.api.attachment.resolve.mutate({ fileIds }); const results: AttachmentUploadResult[] = []; for (const attachment of resolved) { - const metadata: AttachmentMetadata = { + const now = Date.now(); + const metadata = { attachmentId: randomUUID(), + fileId: attachment.fileId, mediaType: mediaTypeOf(attachment.contentType), originalFilename: attachment.name, sizeBytes: attachment.sizeBytes, - createdAt: Date.now(), - }; + createdAt: now, + lastUsedAt: now, + } satisfies AttachmentMetadata; await downloadAttachment(attachment, prepareAttachmentTarget(metadata)); writeAttachmentMetadata(metadata); + indexAttachmentFileId(metadata); results.push({ attachmentId: metadata.attachmentId, + fileId: metadata.fileId, originalFilename: metadata.originalFilename, mediaType: metadata.mediaType, sizeBytes: metadata.sizeBytes, diff --git a/packages/host-service/src/trpc/router/attachments/constants.ts b/packages/host-service/src/trpc/router/attachments/constants.ts deleted file mode 100644 index 7e0f02a29c2..00000000000 --- a/packages/host-service/src/trpc/router/attachments/constants.ts +++ /dev/null @@ -1,8 +0,0 @@ -/** - * Per-file cap on the inline upload — bytes sent base64 in the request body, - * so the ceiling is really the transport's rather than a product decision. - * `importFromCloud` has no equivalent limit here: those bytes arrive from - * cloud storage, and what a composer may send is decided by - * `@superset/shared/attachment-limits`. - */ -export const MAX_INLINE_ATTACHMENT_BYTES = 25 * 1024 * 1024; diff --git a/packages/host-service/src/trpc/router/attachments/index.ts b/packages/host-service/src/trpc/router/attachments/index.ts index 26fbefbf2f2..9cccede2c59 100644 --- a/packages/host-service/src/trpc/router/attachments/index.ts +++ b/packages/host-service/src/trpc/router/attachments/index.ts @@ -1,3 +1,3 @@ +export { MAX_INLINE_ATTACHMENT_BYTES } from "@superset/shared/attachment-limits"; export type { AttachmentUploadResult } from "./attachments"; export { attachmentsRouter, claimFileName } from "./attachments"; -export { MAX_INLINE_ATTACHMENT_BYTES } from "./constants"; diff --git a/packages/host-service/src/trpc/router/attachments/storage.ts b/packages/host-service/src/trpc/router/attachments/storage.ts index 6a66941e43d..b85ea45b68f 100644 --- a/packages/host-service/src/trpc/router/attachments/storage.ts +++ b/packages/host-service/src/trpc/router/attachments/storage.ts @@ -3,10 +3,11 @@ import { mkdirSync, readFileSync, rmSync, + unlinkSync, writeFileSync, } from "node:fs"; import { homedir } from "node:os"; -import { join } from "node:path"; +import { dirname, join } from "node:path"; import mimeTypes from "mime-types"; export interface AttachmentMetadata { @@ -15,8 +16,15 @@ export interface AttachmentMetadata { originalFilename?: string; sizeBytes: number; createdAt: number; + /** The storage object a box's cached copy was imported from. */ + fileId?: string; + /** Absent on an entry written before the store was swept. */ + lastUsedAt?: number; } +/** Inside the store root, so a sweep must skip it. */ +export const FILE_ID_INDEX_DIR = ".by-file-id"; + /** * Resolves the per-org attachment storage root. Honors * `HOST_MANIFEST_DIR` (set by host-service-coordinator with the active @@ -113,6 +121,8 @@ export function deleteAttachment( attachmentId: string, baseDirOverride?: string, ): void { + const metadata = readAttachmentMetadata(attachmentId, baseDirOverride); + if (metadata) unindexAttachmentFileId(metadata, baseDirOverride); const dir = getAttachmentDir(attachmentId, baseDirOverride); rmSync(dir, { recursive: true, force: true }); } @@ -148,3 +158,62 @@ export function resolveAttachmentPath( ); return existsSync(path) ? { path, metadata } : null; } + +/** Every path that hands an attachment out goes through here, so the sweep sees it as used. */ +export function handOutAttachment( + attachmentId: string, + baseDirOverride?: string, +): { path: string; metadata: AttachmentMetadata } | null { + const resolved = resolveAttachmentPath(attachmentId, baseDirOverride); + if (!resolved) return null; + const metadata = { ...resolved.metadata, lastUsedAt: Date.now() }; + writeAttachmentMetadata(metadata, baseDirOverride); + return { path: resolved.path, metadata }; +} + +const FILE_ID_PATTERN = /^[0-9a-f-]{36}$/i; + +function fileIdIndexPath(fileId: string, baseDirOverride?: string): string { + return join(getAttachmentsRoot(baseDirOverride), FILE_ID_INDEX_DIR, fileId); +} + +export function indexAttachmentFileId( + metadata: AttachmentMetadata & { fileId: string }, + baseDirOverride?: string, +): void { + const path = fileIdIndexPath(metadata.fileId, baseDirOverride); + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + writeFileSync(path, metadata.attachmentId, { mode: 0o600 }); +} + +export function unindexAttachmentFileId( + metadata: AttachmentMetadata, + baseDirOverride?: string, +): void { + if (!metadata.fileId) return; + const path = fileIdIndexPath(metadata.fileId, baseDirOverride); + if (readIndexedAttachmentId(path) === metadata.attachmentId) unlinkSync(path); +} + +function readIndexedAttachmentId(path: string): string | null { + try { + return readFileSync(path, "utf-8"); + } catch { + return null; + } +} + +/** The cached copy of a storage object, as `handOutAttachment` hands it out. */ +export function handOutAttachmentByFileId( + fileId: string, + baseDirOverride?: string, +): { path: string; metadata: AttachmentMetadata } | null { + if (!FILE_ID_PATTERN.test(fileId)) return null; + const attachmentId = readIndexedAttachmentId( + fileIdIndexPath(fileId, baseDirOverride), + ); + if (!attachmentId) return null; + const metadata = readAttachmentMetadata(attachmentId, baseDirOverride); + if (metadata?.fileId !== fileId) return null; + return handOutAttachment(attachmentId, baseDirOverride); +} diff --git a/packages/host-service/src/trpc/router/attachments/sweep.test.ts b/packages/host-service/src/trpc/router/attachments/sweep.test.ts new file mode 100644 index 00000000000..628a1e94d88 --- /dev/null +++ b/packages/host-service/src/trpc/router/attachments/sweep.test.ts @@ -0,0 +1,103 @@ +import { afterEach, beforeEach, describe, expect, it } from "bun:test"; +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + type AttachmentMetadata, + getAttachmentDir, + handOutAttachment, + handOutAttachmentByFileId, + indexAttachmentFileId, + readAttachmentMetadata, + writeAttachment, +} from "./storage"; +import { sweepAttachmentStore } from "./sweep"; + +const DAY_MS = 24 * 60 * 60 * 1000; +const NOW = Date.now() - 365 * DAY_MS; + +let base: string; + +beforeEach(() => { + base = mkdtempSync(join(tmpdir(), "superset-attachment-sweep-")); +}); + +afterEach(() => { + rmSync(base, { recursive: true, force: true }); +}); + +function seed(fields: Partial = {}): AttachmentMetadata { + const metadata: AttachmentMetadata = { + attachmentId: crypto.randomUUID(), + mediaType: "image/png", + sizeBytes: 1, + createdAt: NOW - 90 * DAY_MS, + ...fields, + }; + writeAttachment(new Uint8Array([1]), metadata, base); + return metadata; +} + +const sweep = () => + sweepAttachmentStore({ now: NOW, retentionDays: 30, baseDirOverride: base }); + +describe("sweepAttachmentStore", () => { + it("removes an entry unused past retention and keeps a recent one", async () => { + const stale = seed({ lastUsedAt: NOW - 31 * DAY_MS }); + const recent = seed({ lastUsedAt: NOW - 29 * DAY_MS }); + + expect((await sweep()).removed).toEqual([stale.attachmentId]); + expect(existsSync(getAttachmentDir(stale.attachmentId, base))).toBe(false); + expect(existsSync(getAttachmentDir(recent.attachmentId, base))).toBe(true); + }); + + it("stamps an entry from before lastUsedAt instead of removing it", async () => { + const unstamped = seed(); + + expect((await sweep()).removed).toEqual([]); + expect( + readAttachmentMetadata(unstamped.attachmentId, base)?.lastUsedAt, + ).toBe(NOW); + }); + + it("keeps an entry handed out after the pass started", async () => { + const stale = seed({ lastUsedAt: NOW - 31 * DAY_MS }); + + const pass = sweep(); + expect(handOutAttachment(stale.attachmentId, base)).not.toBeNull(); + + expect((await pass).removed).toEqual([]); + expect(existsSync(getAttachmentDir(stale.attachmentId, base))).toBe(true); + }); + + it("drops the fileId lookup with the entry it pointed to", async () => { + const fileId = crypto.randomUUID(); + const stale = seed({ fileId, lastUsedAt: NOW - 31 * DAY_MS }); + indexAttachmentFileId({ ...stale, fileId }, base); + + await sweep(); + expect(handOutAttachmentByFileId(fileId, base)).toBeNull(); + }); +}); + +describe("handOutAttachment", () => { + it("stamps the entry it hands out", () => { + const entry = seed({ lastUsedAt: NOW - 31 * DAY_MS }); + const handed = handOutAttachment(entry.attachmentId, base); + expect(handed?.metadata.lastUsedAt).toBeGreaterThan(NOW); + expect( + readAttachmentMetadata(entry.attachmentId, base)?.lastUsedAt, + ).toBeGreaterThan(NOW); + }); + + it("finds a cloud copy by its fileId and stamps it", () => { + const fileId = crypto.randomUUID(); + const entry = seed({ fileId, lastUsedAt: NOW - 31 * DAY_MS }); + indexAttachmentFileId({ ...entry, fileId }, base); + + const handed = handOutAttachmentByFileId(fileId, base); + expect(handed?.metadata.attachmentId).toBe(entry.attachmentId); + expect(handed?.metadata.lastUsedAt).toBeGreaterThan(NOW); + expect(handOutAttachmentByFileId(crypto.randomUUID(), base)).toBeNull(); + }); +}); diff --git a/packages/host-service/src/trpc/router/attachments/sweep.ts b/packages/host-service/src/trpc/router/attachments/sweep.ts new file mode 100644 index 00000000000..946934e1eac --- /dev/null +++ b/packages/host-service/src/trpc/router/attachments/sweep.ts @@ -0,0 +1,79 @@ +import { renameSync } from "node:fs"; +import { readdir, rm } from "node:fs/promises"; +import { join } from "node:path"; +import { ATTACHMENT_RETENTION_DAYS } from "@superset/shared/attachment-limits"; +import { + FILE_ID_INDEX_DIR, + getAttachmentsRoot, + readAttachmentMetadata, + unindexAttachmentFileId, + writeAttachmentMetadata, +} from "./storage"; + +const DAY_MS = 24 * 60 * 60 * 1000; +const SWEEPING_PREFIX = ".sweeping-"; + +export const ATTACHMENT_SWEEP_INTERVAL_MS = DAY_MS; + +/** + * Removes store entries nobody has used for `retentionDays`. An entry from + * before `lastUsedAt` existed is stamped now instead, so it gets the full + * retention from the first sweep. + */ +export async function sweepAttachmentStore({ + now = Date.now(), + retentionDays = ATTACHMENT_RETENTION_DAYS, + baseDirOverride, +}: { + now?: number; + retentionDays?: number; + baseDirOverride?: string; +} = {}): Promise<{ removed: string[] }> { + const root = getAttachmentsRoot(baseDirOverride); + const entries = await readdir(root).catch(() => [] as string[]); + const removed: string[] = []; + for (const entry of entries) { + if (entry.startsWith(SWEEPING_PREFIX)) { + await rm(join(root, entry), { recursive: true, force: true }); + continue; + } + if (entry === FILE_ID_INDEX_DIR) continue; + const doomed = claimIfUnused(entry, { + cutoff: now - retentionDays * DAY_MS, + now, + root, + baseDirOverride, + }); + if (!doomed) continue; + await rm(doomed, { recursive: true, force: true }); + removed.push(entry); + } + return { removed }; +} + +/** + * Synchronous from the metadata read to the rename, so a hand-out on this + * event loop can't run between the check and the removal. Once renamed, the + * entry is unreachable by id, and the slow delete can run async. + */ +function claimIfUnused( + attachmentId: string, + { + cutoff, + now, + root, + baseDirOverride, + }: { cutoff: number; now: number; root: string; baseDirOverride?: string }, +): string | null { + const metadata = readAttachmentMetadata(attachmentId, baseDirOverride); + if (!metadata) return null; + if (metadata.lastUsedAt === undefined) { + writeAttachmentMetadata({ ...metadata, lastUsedAt: now }, baseDirOverride); + return null; + } + if (metadata.lastUsedAt >= cutoff) return null; + unindexAttachmentFileId(metadata, baseDirOverride); + const doomed = join(root, `${SWEEPING_PREFIX}${attachmentId}`); + renameSync(join(root, attachmentId), doomed); + return doomed; +} diff --git a/packages/host-service/src/trpc/router/sessions/create-chat-session.ts b/packages/host-service/src/trpc/router/sessions/create-chat-session.ts index cbe0d58e036..ff73061bc0f 100644 --- a/packages/host-service/src/trpc/router/sessions/create-chat-session.ts +++ b/packages/host-service/src/trpc/router/sessions/create-chat-session.ts @@ -1,4 +1,5 @@ import type { ChatRuntime } from "@superset/chat-runtime"; +import type { ChatLaunchAttachment } from "../agents/agents"; import type { ChatSessionResult, CreateChatSessionInput, @@ -7,7 +8,7 @@ import type { type ChatTarget = { harness: string; label: string; - attachments: Array<{ attachmentId: string; name: string; mimeType: string }>; + attachments: ChatLaunchAttachment[]; }; export function createChatSession( diff --git a/packages/shared/src/attachment-limits.ts b/packages/shared/src/attachment-limits.ts index c5e18dad369..f24fa3c47d0 100644 --- a/packages/shared/src/attachment-limits.ts +++ b/packages/shared/src/attachment-limits.ts @@ -16,3 +16,17 @@ export const MAX_CHAT_IMAGE_BYTES = 50 * 1024 * 1024; /** Per send, matching what the desktop terminal composer accepts. */ export const MAX_ATTACHMENTS = 10; + +/** + * Per file, for bytes sent base64 in a request body to a host, and read back + * the same way. The ceiling is the transport's: the host holds the body in + * memory. A workspace on a machine someone owns refuses above it, because its + * attachments must not go through our storage. + */ +export const MAX_INLINE_ATTACHMENT_BYTES = 25 * 1024 * 1024; + +/** Largest image a transcript renders inline; anything larger is a name badge. */ +export const MAX_THUMBNAIL_BYTES = 5 * 1024 * 1024; + +/** A host's attachment store removes an entry nobody has used for this long. */ +export const ATTACHMENT_RETENTION_DAYS = 30; diff --git a/packages/shared/src/cloud-agent-launch.ts b/packages/shared/src/cloud-agent-launch.ts index 7509de7ed75..f08814eff86 100644 --- a/packages/shared/src/cloud-agent-launch.ts +++ b/packages/shared/src/cloud-agent-launch.ts @@ -1,4 +1,5 @@ import { z } from "zod"; +import { MAX_ATTACHMENTS } from "./attachment-limits"; import { HOST_AGENT_PRESETS } from "./host-agent-presets"; /** @@ -37,7 +38,7 @@ export const cloudAgentLaunchSchema = z.object({ * them to at create time, so the ids travel and the box pulls the bytes * itself once host-service is up. */ - attachmentFileIds: z.array(z.string().uuid()).max(10).optional(), + attachmentFileIds: z.array(z.string().uuid()).max(MAX_ATTACHMENTS).optional(), surface: z.enum(["terminal", "chat"]).optional(), /** * Set only by cloud automation dispatch. A manual create must not name a