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