diff --git a/apps/server/src/assets/AttachmentUpload.test.ts b/apps/server/src/assets/AttachmentUpload.test.ts new file mode 100644 index 000000000000..cb08d5e4b2f1 --- /dev/null +++ b/apps/server/src/assets/AttachmentUpload.test.ts @@ -0,0 +1,128 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as TestClock from "effect/testing/TestClock"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import { parseThreadSegmentFromAttachmentId } from "../attachmentStore.ts"; +import { + ATTACHMENT_UPLOAD_ROUTE_PREFIX, + deletePendingAttachment, + issueAttachmentUploadUrl, + storeAttachmentUpload, + validateAttachmentUploadToken, +} from "./AttachmentUpload.ts"; + +const testLayer = ServerSecretStore.layer.pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-attachment-upload-" })), + Layer.provideMerge(NodeServices.layer), +); + +const uploadInput = { + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 6, +} as const; + +describe("AttachmentUpload", () => { + it.effect("signs the attachment metadata and validates the upload token", () => + Effect.gen(function* () { + const issued = yield* issueAttachmentUploadUrl(uploadInput); + expect(parseThreadSegmentFromAttachmentId(issued.attachmentId)).toBe("pending"); + + const token = issued.relativeUrl.slice(`${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/`.length); + expect(yield* validateAttachmentUploadToken(token)).toMatchObject({ + kind: "attachment-upload", + attachmentId: issued.attachmentId, + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 6, + }); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("rejects tampered and malformed upload tokens", () => + Effect.gen(function* () { + const issued = yield* issueAttachmentUploadUrl(uploadInput); + const token = issued.relativeUrl.slice(`${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/`.length); + const [payload, signature] = token.split("."); + + expect(yield* validateAttachmentUploadToken(`${payload}x.${signature}`)).toBeNull(); + expect(yield* validateAttachmentUploadToken(`${token}.extra`)).toBeNull(); + expect(yield* validateAttachmentUploadToken("garbage")).toBeNull(); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("rejects expired upload tokens", () => + Effect.gen(function* () { + const issued = yield* issueAttachmentUploadUrl(uploadInput); + const token = issued.relativeUrl.slice(`${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/`.length); + + yield* TestClock.adjust("11 minutes"); + expect(yield* validateAttachmentUploadToken(token)).toBeNull(); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("removes expired pending uploads while issuing a new upload URL", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const staleId = "pending-00000000-0000-4000-8000-0000000000cc"; + const stalePath = NodePath.join(config.attachmentsDir, `${staleId}.png`); + NodeFS.writeFileSync(stalePath, Buffer.from("pixels")); + NodeFS.utimesSync(stalePath, 0, 0); + + yield* TestClock.adjust("25 hours"); + yield* issueAttachmentUploadUrl(uploadInput); + + expect(NodeFS.existsSync(stalePath)).toBe(false); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("stores the expected bytes without leaving temporary files", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const issued = yield* issueAttachmentUploadUrl(uploadInput); + const token = issued.relativeUrl.slice(`${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/`.length); + const claims = yield* validateAttachmentUploadToken(token); + if (!claims) { + throw new Error("Expected valid upload claims."); + } + + expect(yield* storeAttachmentUpload(claims, new Uint8Array([1, 2, 3]))).toMatchObject({ + ok: false, + status: 400, + }); + expect(yield* storeAttachmentUpload(claims, new Uint8Array(6))).toEqual({ ok: true }); + expect( + NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${issued.attachmentId}.png`)), + ).toBe(true); + expect( + NodeFS.readdirSync(config.attachmentsDir).filter((entry) => entry.endsWith(".part")), + ).toEqual([]); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("deletes pending uploads without deleting thread-owned copies", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const uuid = "00000000-0000-4000-8000-0000000000dd"; + const pendingPath = NodePath.join(config.attachmentsDir, `pending-${uuid}.png`); + const claimedPath = NodePath.join(config.attachmentsDir, `thread-1-${uuid}.png`); + NodeFS.writeFileSync(pendingPath, Buffer.from("pixels")); + NodeFS.writeFileSync(claimedPath, Buffer.from("pixels")); + + yield* deletePendingAttachment(`pending-${uuid}`); + yield* deletePendingAttachment(`pending-${uuid}`); + yield* deletePendingAttachment(`thread-1-${uuid}`); + + expect(NodeFS.existsSync(pendingPath)).toBe(false); + expect(NodeFS.existsSync(claimedPath)).toBe(true); + }).pipe(Effect.provide(testLayer)), + ); +}); diff --git a/apps/server/src/assets/AttachmentUpload.ts b/apps/server/src/assets/AttachmentUpload.ts new file mode 100644 index 000000000000..6142b69d7342 --- /dev/null +++ b/apps/server/src/assets/AttachmentUpload.ts @@ -0,0 +1,214 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeCrypto from "node:crypto"; + +import { + ATTACHMENT_UPLOAD_URL_TTL_MS, + type AttachmentCreateUploadUrlInput, + AttachmentUploadSigningKeyError, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; + +import { + createPendingAttachmentId, + parseThreadSegmentFromAttachmentId, + PENDING_ATTACHMENT_THREAD_SEGMENT, + resolveAttachmentPathById, + sweepStalePendingAttachments, +} from "../attachmentStore.ts"; +import { resolveAttachmentRelativePath } from "../attachmentPaths.ts"; +import { + base64UrlDecodeUtf8, + base64UrlEncode, + signPayload, + timingSafeEqualBase64Url, +} from "../auth/utils.ts"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import { inferImageExtension } from "../imageMime.ts"; + +export const ATTACHMENT_UPLOAD_ROUTE_PREFIX = "/api/attachments/upload"; + +// Asset download tokens share this key, but their signed claim kind is different. +const SIGNING_SECRET_NAME = "asset-access-signing-key"; +const PENDING_ATTACHMENT_SWEEP_INTERVAL_MS = 15 * 60_000; +const lastPendingSweepByDirectory = new Map(); + +const AttachmentUploadClaims = Schema.Struct({ + version: Schema.Literal(1), + kind: Schema.Literal("attachment-upload"), + attachmentId: Schema.String, + name: Schema.String, + mimeType: Schema.String, + sizeBytes: Schema.Number, + expiresAt: Schema.Number, +}); +export type AttachmentUploadClaims = typeof AttachmentUploadClaims.Type; + +const attachmentUploadClaimsJson = Schema.fromJsonString(AttachmentUploadClaims); +const decodeAttachmentUploadClaims = Schema.decodeUnknownOption(attachmentUploadClaimsJson); +const encodeAttachmentUploadClaims = Schema.encodeSync(attachmentUploadClaimsJson); + +function decodeClaims(encodedPayload: string): AttachmentUploadClaims | null { + try { + return Option.getOrNull(decodeAttachmentUploadClaims(base64UrlDecodeUtf8(encodedPayload))); + } catch { + return null; + } +} + +const loadSigningSecret = Effect.gen(function* () { + const secretStore = yield* ServerSecretStore.ServerSecretStore; + return yield* secretStore.getOrCreateRandom(SIGNING_SECRET_NAME, 32); +}); + +export const issueAttachmentUploadUrl = Effect.fn("AttachmentUpload.issueUrl")(function* ( + input: AttachmentCreateUploadUrlInput, +) { + const secret = yield* loadSigningSecret.pipe( + Effect.mapError((cause) => new AttachmentUploadSigningKeyError({ cause })), + ); + const config = yield* ServerConfig.ServerConfig; + const nowMs = yield* Clock.currentTimeMillis; + const previousSweep = lastPendingSweepByDirectory.get(config.attachmentsDir); + if ( + previousSweep === undefined || + nowMs - previousSweep >= PENDING_ATTACHMENT_SWEEP_INTERVAL_MS + ) { + lastPendingSweepByDirectory.set(config.attachmentsDir, nowMs); + const swept = sweepStalePendingAttachments({ + attachmentsDir: config.attachmentsDir, + nowMs, + }); + if (swept.deleted > 0) { + yield* Effect.logInfo("Removed expired attachment uploads.", { deleted: swept.deleted }); + } + } + + const attachmentId = createPendingAttachmentId(); + const expiresAt = nowMs + ATTACHMENT_UPLOAD_URL_TTL_MS; + const encodedPayload = base64UrlEncode( + encodeAttachmentUploadClaims({ + version: 1, + kind: "attachment-upload", + attachmentId, + name: input.name, + mimeType: input.mimeType, + sizeBytes: input.sizeBytes, + expiresAt, + }), + ); + + return { + attachmentId, + relativeUrl: `${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/${encodedPayload}.${signPayload(encodedPayload, secret)}`, + expiresAt, + }; +}); + +export const validateAttachmentUploadToken = Effect.fn("AttachmentUpload.validateToken")(function* ( + token: string, +) { + const [encodedPayload, signature, unexpectedSegment] = token.split("."); + if (!encodedPayload || !signature || unexpectedSegment) { + return null; + } + + const secret = yield* loadSigningSecret.pipe( + Effect.tapError((cause) => + Effect.logError("Failed to load the attachment upload signing key.", { cause }), + ), + Effect.orElseSucceed(() => null), + ); + if (!secret || !timingSafeEqualBase64Url(signature, signPayload(encodedPayload, secret))) { + return null; + } + + const claims = decodeClaims(encodedPayload); + if (!claims || claims.expiresAt <= (yield* Clock.currentTimeMillis)) { + return null; + } + return claims; +}); + +export type StoreAttachmentUploadResult = + | { readonly ok: true } + | { readonly ok: false; readonly status: number; readonly detail: string }; + +export const storeAttachmentUpload = Effect.fn("AttachmentUpload.store")(function* ( + claims: AttachmentUploadClaims, + bytes: Uint8Array, +) { + if (bytes.byteLength !== claims.sizeBytes) { + return { + ok: false, + status: 400, + detail: `Body was ${bytes.byteLength} bytes, expected ${claims.sizeBytes}.`, + } satisfies StoreAttachmentUploadResult; + } + + const config = yield* ServerConfig.ServerConfig; + const extension = inferImageExtension({ mimeType: claims.mimeType, fileName: claims.name }); + const relativePath = `${claims.attachmentId}${extension}`; + const finalPath = resolveAttachmentRelativePath({ + attachmentsDir: config.attachmentsDir, + relativePath, + }); + const partPath = resolveAttachmentRelativePath({ + attachmentsDir: config.attachmentsDir, + relativePath: `${relativePath}.${NodeCrypto.randomUUID()}.part`, + }); + if (!finalPath || !partPath) { + return { ok: false, status: 500, detail: "Failed to resolve attachment path." }; + } + + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + return yield* Effect.gen(function* () { + yield* fileSystem.makeDirectory(path.dirname(finalPath), { recursive: true }); + yield* fileSystem.writeFile(partPath, bytes); + yield* fileSystem.rename(partPath, finalPath); + return { ok: true } satisfies StoreAttachmentUploadResult; + }).pipe( + Effect.catch((cause) => + fileSystem.remove(partPath, { force: true }).pipe( + Effect.orElseSucceed(() => undefined), + Effect.andThen( + Effect.logError("Failed to persist attachment upload.", { + attachmentId: claims.attachmentId, + cause, + }), + ), + Effect.as({ + ok: false, + status: 500, + detail: "Failed to persist upload.", + } satisfies StoreAttachmentUploadResult), + ), + ), + ); +}); + +export const deletePendingAttachment = Effect.fn("AttachmentUpload.deletePending")(function* ( + attachmentId: string, +) { + if (parseThreadSegmentFromAttachmentId(attachmentId) !== PENDING_ATTACHMENT_THREAD_SEGMENT) { + return; + } + + const config = yield* ServerConfig.ServerConfig; + const attachmentPath = resolveAttachmentPathById({ + attachmentsDir: config.attachmentsDir, + attachmentId, + }); + if (!attachmentPath) { + return; + } + + const fileSystem = yield* FileSystem.FileSystem; + yield* fileSystem.remove(attachmentPath, { force: true }).pipe(Effect.orElseSucceed(() => {})); +}); diff --git a/apps/server/src/attachmentStore.test.ts b/apps/server/src/attachmentStore.test.ts index e21d9cf62cf5..5e782e55407f 100644 --- a/apps/server/src/attachmentStore.test.ts +++ b/apps/server/src/attachmentStore.test.ts @@ -7,8 +7,12 @@ import { describe, expect, it } from "vite-plus/test"; import { createAttachmentId, + createPendingAttachmentId, + parseAttachmentUuid, + planAttachmentClaim, parseThreadSegmentFromAttachmentId, resolveAttachmentPathById, + sweepStalePendingAttachments, } from "./attachmentStore.ts"; describe("attachmentStore", () => { @@ -44,6 +48,16 @@ describe("attachmentStore", () => { expect(parseThreadSegmentFromAttachmentId(attachmentId)).toBe("thread-foo"); }); + it("reserves the pending attachment segment", () => { + const pendingId = createPendingAttachmentId(); + expect(parseThreadSegmentFromAttachmentId(pendingId)).toBe("pending"); + expect(parseAttachmentUuid(pendingId)).toMatch(/^[a-f0-9-]{36}$/); + expect(parseThreadSegmentFromAttachmentId(createAttachmentId("pending")!)).toBe("_pending"); + expect(parseThreadSegmentFromAttachmentId(createAttachmentId("pending_thread")!)).toBe( + "pending_thread", + ); + }); + it("resolves attachment path by id using the extension that exists on disk", () => { const attachmentsDir = NodeFS.mkdtempSync( NodePath.join(NodeOS.tmpdir(), "t3code-attachment-store-"), @@ -77,4 +91,75 @@ describe("attachmentStore", () => { NodeFS.rmSync(attachmentsDir, { recursive: true, force: true }); } }); + + it("plans pending attachment claims with direct filename lookups", () => { + const attachmentsDir = NodeFS.mkdtempSync( + NodePath.join(NodeOS.tmpdir(), "t3code-attachment-claim-"), + ); + try { + const uuid = "00000000-0000-4000-8000-000000000001"; + const pendingPath = NodePath.join(attachmentsDir, `pending-${uuid}.png`); + NodeFS.writeFileSync(pendingPath, Buffer.from("pixels")); + + const claim = planAttachmentClaim({ + attachmentsDir, + threadId: "thread-1", + attachmentId: `pending-${uuid}`, + }); + expect(claim).toMatchObject({ + ok: true, + currentPath: pendingPath, + }); + if (!claim.ok) { + return; + } + expect(parseThreadSegmentFromAttachmentId(claim.finalId)).toBe("thread-1"); + expect(parseAttachmentUuid(claim.finalId)).not.toBe(uuid); + expect(claim.finalPath).toBe(NodePath.join(attachmentsDir, `${claim.finalId}.png`)); + } finally { + NodeFS.rmSync(attachmentsDir, { recursive: true, force: true }); + } + }); + + it("rejects thread-owned attachments even when thread segments collide", () => { + const attachmentsDir = NodeFS.mkdtempSync( + NodePath.join(NodeOS.tmpdir(), "t3code-attachment-ownership-"), + ); + try { + const attachmentId = "a-b-00000000-0000-4000-8000-000000000003"; + NodeFS.writeFileSync(NodePath.join(attachmentsDir, `${attachmentId}.png`), "pixels"); + + expect(planAttachmentClaim({ attachmentsDir, threadId: "a b", attachmentId })).toEqual({ + ok: false, + reason: "attachment must be a pending upload", + }); + } finally { + NodeFS.rmSync(attachmentsDir, { recursive: true, force: true }); + } + }); + + it("removes expired pending and partial files without touching thread attachments", () => { + const attachmentsDir = NodeFS.mkdtempSync( + NodePath.join(NodeOS.tmpdir(), "t3code-attachment-sweep-"), + ); + try { + const now = 1_800_000_000_000; + const oldTimeSeconds = (now - 2 * 24 * 60 * 60 * 1000) / 1000; + const uuid = "00000000-0000-4000-8000-000000000002"; + const pendingPath = NodePath.join(attachmentsDir, `pending-${uuid}.png`); + const threadPath = NodePath.join(attachmentsDir, `thread-1-${uuid}.png`); + const partialPath = NodePath.join(attachmentsDir, `${uuid}.part`); + for (const filePath of [pendingPath, threadPath, partialPath]) { + NodeFS.writeFileSync(filePath, Buffer.from("pixels")); + NodeFS.utimesSync(filePath, oldTimeSeconds, oldTimeSeconds); + } + + expect(sweepStalePendingAttachments({ attachmentsDir, nowMs: now })).toEqual({ deleted: 2 }); + expect(NodeFS.existsSync(pendingPath)).toBe(false); + expect(NodeFS.existsSync(partialPath)).toBe(false); + expect(NodeFS.existsSync(threadPath)).toBe(true); + } finally { + NodeFS.rmSync(attachmentsDir, { recursive: true, force: true }); + } + }); }); diff --git a/apps/server/src/attachmentStore.ts b/apps/server/src/attachmentStore.ts index 3d5b531db217..d0334bce09f3 100644 --- a/apps/server/src/attachmentStore.ts +++ b/apps/server/src/attachmentStore.ts @@ -1,6 +1,7 @@ // @effect-diagnostics nodeBuiltinImport:off import * as NodeCrypto from "node:crypto"; import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; import type { ChatAttachment } from "@t3tools/contracts"; @@ -19,6 +20,10 @@ const ATTACHMENT_ID_PATTERN = new RegExp( "i", ); +export const PENDING_ATTACHMENT_THREAD_SEGMENT = "pending"; +export const PENDING_ATTACHMENT_MAX_AGE_MS = 24 * 60 * 60 * 1000; +const PARTIAL_UPLOAD_MAX_AGE_MS = 60 * 60 * 1000; + export function toSafeThreadAttachmentSegment(threadId: string): string | null { const segment = threadId .trim() @@ -31,7 +36,19 @@ export function toSafeThreadAttachmentSegment(threadId: string): string | null { if (segment.length === 0) { return null; } - return segment; + return segment === PENDING_ATTACHMENT_THREAD_SEGMENT ? "_pending" : segment; +} + +export function createPendingAttachmentId(): string { + return `${PENDING_ATTACHMENT_THREAD_SEGMENT}-${NodeCrypto.randomUUID()}`; +} + +export function parseAttachmentUuid(attachmentId: string): string | null { + const normalizedId = normalizeAttachmentRelativePath(attachmentId); + if (!normalizedId || normalizedId.includes("/") || normalizedId.includes(".")) { + return null; + } + return normalizedId.match(ATTACHMENT_ID_PATTERN)?.[2]?.toLowerCase() ?? null; } export function createAttachmentId(threadId: string): string | null { @@ -96,6 +113,105 @@ export function resolveAttachmentPathById(input: { return null; } +export type AttachmentClaimPlan = + | { + readonly ok: true; + readonly finalId: string; + readonly currentPath: string; + readonly finalPath: string; + } + | { readonly ok: false; readonly reason: string }; + +export function planAttachmentClaim(input: { + readonly attachmentsDir: string; + readonly threadId: string; + readonly attachmentId: string; +}): AttachmentClaimPlan { + const uuid = parseAttachmentUuid(input.attachmentId); + const requestedSegment = parseThreadSegmentFromAttachmentId(input.attachmentId); + if (!uuid || !requestedSegment) { + return { ok: false, reason: "invalid attachment id" }; + } + + if (!toSafeThreadAttachmentSegment(input.threadId)) { + return { ok: false, reason: "invalid thread id" }; + } + if (requestedSegment !== PENDING_ATTACHMENT_THREAD_SEGMENT) { + return { ok: false, reason: "attachment must be a pending upload" }; + } + + const currentPath = resolveAttachmentPathById({ + attachmentsDir: input.attachmentsDir, + attachmentId: input.attachmentId, + }); + if (!currentPath) { + return { ok: false, reason: "attachment not found (removed or expired)" }; + } + const finalId = createAttachmentId(input.threadId); + if (!finalId) { + return { ok: false, reason: "failed to create attachment id" }; + } + + const expectedFinalPath = resolveAttachmentRelativePath({ + attachmentsDir: input.attachmentsDir, + relativePath: `${finalId}${NodePath.extname(currentPath)}`, + }); + if (!expectedFinalPath) { + return { ok: false, reason: "failed to resolve attachment path" }; + } + return { + ok: true, + finalId, + currentPath, + finalPath: expectedFinalPath, + }; +} + +export function sweepStalePendingAttachments(input: { + readonly attachmentsDir: string; + readonly nowMs: number; +}): { readonly deleted: number } { + let entries: string[]; + try { + entries = NodeFS.readdirSync(input.attachmentsDir); + } catch { + return { deleted: 0 }; + } + + let deleted = 0; + for (const entry of entries) { + const isPartial = entry.endsWith(".part"); + if (!isPartial) { + const attachmentId = parseAttachmentIdFromRelativePath(entry); + if ( + !attachmentId || + parseThreadSegmentFromAttachmentId(attachmentId) !== PENDING_ATTACHMENT_THREAD_SEGMENT + ) { + continue; + } + } + + const resolved = resolveAttachmentRelativePath({ + attachmentsDir: input.attachmentsDir, + relativePath: entry, + }); + if (!resolved) { + continue; + } + try { + const maxAgeMs = isPartial ? PARTIAL_UPLOAD_MAX_AGE_MS : PENDING_ATTACHMENT_MAX_AGE_MS; + if (input.nowMs - NodeFS.statSync(resolved).mtimeMs > maxAgeMs) { + NodeFS.unlinkSync(resolved); + deleted += 1; + } + } catch { + continue; + } + } + + return { deleted }; +} + export function parseAttachmentIdFromRelativePath(relativePath: string): string | null { const normalized = normalizeAttachmentRelativePath(relativePath); if (!normalized || normalized.includes("/")) { diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 70227cdd4ebf..28ceac4cec99 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -84,6 +84,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.shellOpenInEditor]: AuthOrchestrationOperateScope, [WS_METHODS.filesystemBrowse]: AuthOrchestrationReadScope, [WS_METHODS.assetsCreateUrl]: AuthOrchestrationReadScope, + [WS_METHODS.attachmentsCreateUploadUrl]: AuthOrchestrationOperateScope, + [WS_METHODS.attachmentsDelete]: AuthOrchestrationOperateScope, [WS_METHODS.providerUploadFeedback]: AuthOrchestrationOperateScope, [WS_METHODS.subscribeVcsStatus]: AuthOrchestrationReadScope, [WS_METHODS.subscribeResourceTelemetry]: AuthOrchestrationReadScope, diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index e678264dde5f..bdff19572fdd 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -7,6 +7,7 @@ * @module ServerConfig */ import * as Context from "effect/Context"; +import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; @@ -14,6 +15,8 @@ import * as LogLevel from "effect/LogLevel"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import { sweepStalePendingAttachments } from "./attachmentStore.ts"; + export const DEFAULT_PORT = 3773; export const RuntimeMode = Schema.Literals(["web", "desktop"]); @@ -152,6 +155,14 @@ export const ensureServerDirectories = Effect.fn(function* (derivedPaths: Server ], { concurrency: "unbounded" }, ); + + const swept = sweepStalePendingAttachments({ + attachmentsDir: derivedPaths.attachmentsDir, + nowMs: yield* Clock.currentTimeMillis, + }); + if (swept.deleted > 0) { + yield* Effect.logInfo("Removed expired attachment uploads.", { deleted: swept.deleted }); + } }); const makeTest = Effect.fn("ServerConfig.makeTest")(function* ( diff --git a/apps/server/src/environment/ServerEnvironment.test.ts b/apps/server/src/environment/ServerEnvironment.test.ts index ee30d987591d..b9a50ca8335e 100644 --- a/apps/server/src/environment/ServerEnvironment.test.ts +++ b/apps/server/src/environment/ServerEnvironment.test.ts @@ -90,6 +90,7 @@ it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => { expect(first.environmentId).toBe(second.environmentId); expect(second.capabilities.repositoryIdentity).toBe(true); expect(second.capabilities.connectionProbe).toBe(true); + expect(second.capabilities.attachmentUploads).toBe(true); expect(second.capabilities.pullRequests).toBe(true); expect(second.capabilities.threadTitleRegeneration).toBe(true); expect(second.capabilities.agentActivityPublishing).toBe(false); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 45dc0ee9cfd5..e55639ce659c 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -146,6 +146,7 @@ export const make = Effect.gen(function* () { capabilities: { repositoryIdentity: true, connectionProbe: true, + attachmentUploads: true, pullRequests: true, threadSettlement: true, threadSnooze: true, diff --git a/apps/server/src/http.ts b/apps/server/src/http.ts index be133399e0f8..c3104e7bc420 100644 --- a/apps/server/src/http.ts +++ b/apps/server/src/http.ts @@ -28,6 +28,11 @@ import { OtlpTracer } from "effect/unstable/observability"; import * as ServerConfig from "./config.ts"; import { ASSET_ROUTE_PREFIX, resolveAsset } from "./assets/AssetAccess.ts"; +import { + ATTACHMENT_UPLOAD_ROUTE_PREFIX, + storeAttachmentUpload, + validateAttachmentUploadToken, +} from "./assets/AttachmentUpload.ts"; import * as BrowserTraceCollector from "./observability/BrowserTraceCollector.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import { traceRelayRequest } from "./cloud/traceRelayRequest.ts"; @@ -230,6 +235,51 @@ export const assetRouteLayer = HttpRouter.add( }), ); +export const attachmentUploadRouteLayer = HttpRouter.add( + "POST", + `${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/*`, + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const url = HttpServerRequest.toURL(request); + if (Option.isNone(url)) { + return HttpServerResponse.text("Bad Request", { status: 400 }); + } + + const token = url.value.pathname.slice(`${ATTACHMENT_UPLOAD_ROUTE_PREFIX}/`.length); + if (!token) { + return HttpServerResponse.text("Not Found", { status: 404 }); + } + const claims = yield* validateAttachmentUploadToken(token); + if (!claims) { + return HttpServerResponse.text("Not Found", { status: 404 }); + } + + const contentLengthHeader = request.headers["content-length"]; + if ( + contentLengthHeader !== undefined && + (!Number.isInteger(Number(contentLengthHeader)) || + Number(contentLengthHeader) !== claims.sizeBytes) + ) { + return HttpServerResponse.text("Content-Length must match the upload size.", { + status: 400, + }); + } + + const body = yield* request.arrayBuffer.pipe( + Effect.provideService(HttpServerRequest.MaxBodySize, FileSystem.Size(claims.sizeBytes)), + Effect.orElseSucceed(() => null), + ); + if (body === null) { + return HttpServerResponse.text("Failed to read the upload body.", { status: 400 }); + } + + const stored = yield* storeAttachmentUpload(claims, new Uint8Array(body)); + return stored.ok + ? HttpServerResponse.empty({ status: 204 }) + : HttpServerResponse.text(stored.detail, { status: stored.status }); + }), +); + export const staticAndDevRouteLayer = HttpRouter.add( "GET", "*", diff --git a/apps/server/src/orchestration/Normalizer.attachments.test.ts b/apps/server/src/orchestration/Normalizer.attachments.test.ts new file mode 100644 index 000000000000..27a35977ffca --- /dev/null +++ b/apps/server/src/orchestration/Normalizer.attachments.test.ts @@ -0,0 +1,318 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; + +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { describe, expect, it } from "@effect/vitest"; +import { + type ClientOrchestrationCommand, + CommandId, + MessageId, + ThreadId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import * as ServerConfig from "../config.ts"; +import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; +import { cleanupFailedUploadedAttachments, normalizeDispatchCommand } from "./Normalizer.ts"; + +const testLayer = Layer.mergeAll( + WorkspacePaths.layer, + ServerConfig.layerTest(process.cwd(), { prefix: "t3-normalizer-attachments-" }), +).pipe(Layer.provideMerge(NodeServices.layer)); + +const attachmentUuid = "00000000-0000-4000-8000-0000000000aa"; + +function turnStartCommand(input: { + readonly threadId?: string; + readonly attachments: ReadonlyArray< + | { readonly id: string; readonly sizeBytes: number } + | { readonly dataUrl: string; readonly sizeBytes: number } + >; +}): ClientOrchestrationCommand { + return { + type: "thread.turn.start", + commandId: CommandId.make("command-1"), + threadId: ThreadId.make(input.threadId ?? "thread-1"), + message: { + messageId: MessageId.make("message-1"), + role: "user", + text: "look at this", + attachments: input.attachments.map((attachment) => ({ + type: "image" as const, + name: "screenshot.png", + mimeType: "image/png", + ...attachment, + })), + }, + runtimeMode: "full-access", + interactionMode: "default", + createdAt: "2026-08-01T00:00:00.000Z", + }; +} + +describe("normalizeDispatchCommand attachments", () => { + it.effect("preserves inline image attachments from existing mobile clients", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const normalized = yield* normalizeDispatchCommand( + turnStartCommand({ + attachments: [{ dataUrl: "data:image/png;base64,cGl4ZWxz", sizeBytes: 6 }], + }), + ); + if (normalized.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + + const attachment = normalized.message.attachments[0]!; + expect(attachment.id.startsWith("thread-1-")).toBe(true); + expect( + NodeFS.readFileSync(NodePath.join(config.attachmentsDir, `${attachment.id}.png`)), + ).toEqual(Buffer.from("pixels")); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("claims uploaded attachments while retaining a retryable pending copy", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const bytes = Buffer.from("pixels"); + const pendingPath = NodePath.join(config.attachmentsDir, `pending-${attachmentUuid}.png`); + NodeFS.writeFileSync(pendingPath, bytes); + + const normalized = yield* normalizeDispatchCommand( + turnStartCommand({ + attachments: [{ id: `pending-${attachmentUuid}`, sizeBytes: bytes.byteLength }], + }), + ); + if (normalized.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + + const attachmentId = normalized.message.attachments[0]!.id; + expect(attachmentId.startsWith("thread-1-")).toBe(true); + expect(attachmentId).not.toBe(`thread-1-${attachmentUuid}`); + expect(NodeFS.existsSync(pendingPath)).toBe(true); + expect(NodeFS.existsSync(NodePath.join(config.attachmentsDir, `${attachmentId}.png`))).toBe( + true, + ); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("normalizes inline and uploaded attachments in the same turn", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + NodeFS.writeFileSync( + NodePath.join(config.attachmentsDir, `pending-${attachmentUuid}.png`), + Buffer.from("pixels"), + ); + + const normalized = yield* normalizeDispatchCommand( + turnStartCommand({ + attachments: [ + { dataUrl: "data:image/png;base64,cGl4ZWxz", sizeBytes: 6 }, + { id: `pending-${attachmentUuid}`, sizeBytes: 6 }, + ], + }), + ); + if (normalized.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + + expect(normalized.message.attachments).toHaveLength(2); + expect(normalized.message.attachments[1]?.id.startsWith("thread-1-")).toBe(true); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("retries a failed bootstrap with a fresh thread id", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const bytes = Buffer.from("pixels"); + NodeFS.writeFileSync( + NodePath.join(config.attachmentsDir, `pending-${attachmentUuid}.png`), + bytes, + ); + + const first = yield* normalizeDispatchCommand( + turnStartCommand({ + attachments: [{ id: `pending-${attachmentUuid}`, sizeBytes: bytes.byteLength }], + }), + ); + if (first.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + NodeFS.rmSync( + NodePath.join(config.attachmentsDir, `${first.message.attachments[0]!.id}.png`), + ); + + const retried = yield* normalizeDispatchCommand( + turnStartCommand({ + threadId: "thread-retry", + attachments: [{ id: `pending-${attachmentUuid}`, sizeBytes: bytes.byteLength }], + }), + ); + if (retried.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + expect(retried.message.attachments[0]?.id.startsWith("thread-retry-")).toBe(true); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("removes failed attachment claims without deleting their pending uploads", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const pendingPath = NodePath.join(config.attachmentsDir, `pending-${attachmentUuid}.png`); + NodeFS.writeFileSync(pendingPath, Buffer.from("pixels")); + const command = turnStartCommand({ + attachments: [ + { dataUrl: "data:image/png;base64,cGl4ZWxz", sizeBytes: 6 }, + { id: `pending-${attachmentUuid}`, sizeBytes: 6 }, + ], + }); + const normalized = yield* normalizeDispatchCommand(command); + if (normalized.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + + const inlinePath = NodePath.join( + config.attachmentsDir, + `${normalized.message.attachments[0]!.id}.png`, + ); + const claimedPath = NodePath.join( + config.attachmentsDir, + `${normalized.message.attachments[1]!.id}.png`, + ); + yield* cleanupFailedUploadedAttachments(command, normalized); + + expect(NodeFS.existsSync(pendingPath)).toBe(true); + expect(NodeFS.existsSync(claimedPath)).toBe(false); + expect(NodeFS.existsSync(inlinePath)).toBe(true); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("removes a failed claimed copy after its pending original was removed", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const pendingPath = NodePath.join(config.attachmentsDir, `pending-${attachmentUuid}.png`); + NodeFS.writeFileSync(pendingPath, Buffer.from("pixels")); + const command = turnStartCommand({ + attachments: [{ id: `pending-${attachmentUuid}`, sizeBytes: 6 }], + }); + const normalized = yield* normalizeDispatchCommand(command); + if (normalized.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + + const claimedPath = NodePath.join( + config.attachmentsDir, + `${normalized.message.attachments[0]!.id}.png`, + ); + NodeFS.rmSync(pendingPath); + + yield* cleanupFailedUploadedAttachments(command, normalized); + + expect(NodeFS.existsSync(claimedPath)).toBe(false); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("keeps concurrent claims independent when one dispatch fails", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const pendingPath = NodePath.join(config.attachmentsDir, `pending-${attachmentUuid}.png`); + NodeFS.writeFileSync(pendingPath, Buffer.from("pixels")); + const command = turnStartCommand({ + attachments: [{ id: `pending-${attachmentUuid}`, sizeBytes: 6 }], + }); + + const [failed, succeeded] = yield* Effect.all( + [normalizeDispatchCommand(command), normalizeDispatchCommand(command)], + { concurrency: 2 }, + ); + if (failed.type !== "thread.turn.start" || succeeded.type !== "thread.turn.start") { + throw new Error("Expected thread.turn.start commands."); + } + + const failedPath = NodePath.join( + config.attachmentsDir, + `${failed.message.attachments[0]!.id}.png`, + ); + const succeededPath = NodePath.join( + config.attachmentsDir, + `${succeeded.message.attachments[0]!.id}.png`, + ); + expect(failedPath).not.toBe(succeededPath); + + yield* cleanupFailedUploadedAttachments(command, failed); + + expect(NodeFS.existsSync(pendingPath)).toBe(true); + expect(NodeFS.existsSync(failedPath)).toBe(false); + expect(NodeFS.existsSync(succeededPath)).toBe(true); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("removes earlier claimed copies when a later attachment cannot be normalized", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const pendingId = `pending-${attachmentUuid}`; + const pendingPath = NodePath.join(config.attachmentsDir, `${pendingId}.png`); + NodeFS.writeFileSync(pendingPath, Buffer.from("pixels")); + + const failure = yield* normalizeDispatchCommand( + turnStartCommand({ + attachments: [ + { id: pendingId, sizeBytes: 6 }, + { + id: "pending-00000000-0000-4000-8000-0000000000ff", + sizeBytes: 6, + }, + ], + }), + ).pipe(Effect.flip); + + expect(failure.message).toContain("not found"); + expect(NodeFS.readdirSync(config.attachmentsDir)).toEqual([`${pendingId}.png`]); + }).pipe(Effect.provide(testLayer)), + ); + + it.effect("rejects uploaded attachments with the wrong size or thread", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + NodeFS.writeFileSync( + NodePath.join(config.attachmentsDir, `pending-${attachmentUuid}.png`), + Buffer.from("pixels"), + ); + + const wrongSize = yield* normalizeDispatchCommand( + turnStartCommand({ + attachments: [{ id: `pending-${attachmentUuid}`, sizeBytes: 999 }], + }), + ).pipe(Effect.flip); + expect(wrongSize.message).toContain("size"); + + const wrongThread = yield* normalizeDispatchCommand( + turnStartCommand({ + attachments: [{ id: `another-thread-${attachmentUuid}`, sizeBytes: 6 }], + }), + ).pipe(Effect.flip); + expect(wrongThread.message).toContain("pending upload"); + + const mismatchedTypeCommand = turnStartCommand({ + attachments: [{ id: `pending-${attachmentUuid}`, sizeBytes: 6 }], + }); + if (mismatchedTypeCommand.type !== "thread.turn.start") { + throw new Error("Expected a thread.turn.start command."); + } + const mismatchedType = yield* normalizeDispatchCommand({ + ...mismatchedTypeCommand, + message: { + ...mismatchedTypeCommand.message, + attachments: mismatchedTypeCommand.message.attachments.map((attachment) => ({ + ...attachment, + mimeType: "image/jpeg", + })), + }, + }).pipe(Effect.flip); + expect(mismatchedType.message).toContain("image type"); + }).pipe(Effect.provide(testLayer)), + ); +}); diff --git a/apps/server/src/orchestration/Normalizer.ts b/apps/server/src/orchestration/Normalizer.ts index 24c65900b296..bd6a8f242b87 100644 --- a/apps/server/src/orchestration/Normalizer.ts +++ b/apps/server/src/orchestration/Normalizer.ts @@ -10,7 +10,13 @@ import { PROVIDER_SEND_TURN_MAX_IMAGE_BYTES, } from "@t3tools/contracts"; -import { createAttachmentId, resolveAttachmentPath } from "../attachmentStore.ts"; +import { + createAttachmentId, + planAttachmentClaim, + PENDING_ATTACHMENT_THREAD_SEGMENT, + parseThreadSegmentFromAttachmentId, + resolveAttachmentPath, +} from "../attachmentStore.ts"; import { ServerConfig } from "../config.ts"; import { parseBase64DataUrl } from "../imageMime.ts"; import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; @@ -43,6 +49,29 @@ export const canonicalizeClientCommandTimestamps = ( }; }; +const removeClaimedAttachmentPaths = Effect.fn("Normalizer.removeClaimedAttachmentPaths")( + function* (attachmentPaths: ReadonlyArray) { + if (attachmentPaths.length === 0) { + return; + } + const fileSystem = yield* FileSystem.FileSystem; + yield* Effect.forEach( + attachmentPaths, + (attachmentPath) => + fileSystem.remove(attachmentPath, { force: true }).pipe( + Effect.tapError((cause) => + Effect.logWarning("Failed to remove an unclaimed attachment copy.", { + attachmentPath, + cause, + }), + ), + Effect.orElseSucceed(() => undefined), + ), + { concurrency: 1 }, + ); + }, +); + export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) => Effect.gen(function* () { const receivedAt = DateTime.formatIso(yield* DateTime.now); @@ -104,10 +133,69 @@ export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) => return canonicalCommand as OrchestrationCommand; } + const claimedAttachmentPaths: string[] = []; const normalizedAttachments = yield* Effect.forEach( canonicalCommand.message.attachments, (attachment) => Effect.gen(function* () { + if (!("dataUrl" in attachment)) { + const claim = planAttachmentClaim({ + attachmentsDir: serverConfig.attachmentsDir, + threadId: canonicalCommand.threadId, + attachmentId: attachment.id, + }); + if (!claim.ok) { + return yield* new OrchestrationDispatchCommandError({ + message: `Attachment '${attachment.name}' cannot be sent: ${claim.reason}.`, + }); + } + + const info = yield* fileSystem.stat(claim.currentPath).pipe( + Effect.mapError( + (cause) => + new OrchestrationDispatchCommandError({ + message: `Attachment '${attachment.name}' cannot be sent: attachment not found.`, + cause, + }), + ), + ); + if (Number(info.size) !== attachment.sizeBytes) { + return yield* new OrchestrationDispatchCommandError({ + message: `Attachment '${attachment.name}' cannot be sent: stored size does not match.`, + }); + } + + const normalizedAttachment = { + ...attachment, + id: claim.finalId, + mimeType: attachment.mimeType.toLowerCase(), + }; + const expectedPath = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment: normalizedAttachment, + }); + if (expectedPath !== claim.finalPath) { + return yield* new OrchestrationDispatchCommandError({ + message: `Attachment '${attachment.name}' cannot be sent: image type does not match the upload.`, + }); + } + + // Keep the pending copy until the turn succeeds. A failed thread + // bootstrap can then retry with a fresh thread id. + yield* fileSystem.copyFile(claim.currentPath, claim.finalPath).pipe( + Effect.mapError( + (cause) => + new OrchestrationDispatchCommandError({ + message: `Failed to claim attachment '${attachment.name}' for this thread.`, + cause, + }), + ), + ); + claimedAttachmentPaths.push(claim.finalPath); + + return normalizedAttachment; + } + const parsed = parseBase64DataUrl(attachment.dataUrl); if (!parsed || !parsed.mimeType.startsWith("image/")) { return yield* new OrchestrationDispatchCommandError({ @@ -167,7 +255,7 @@ export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) => return persistedAttachment; }), { concurrency: 1 }, - ); + ).pipe(Effect.tapError(() => removeClaimedAttachmentPaths(claimedAttachmentPaths))); return { ...canonicalCommand, @@ -177,3 +265,33 @@ export const normalizeDispatchCommand = (command: ClientOrchestrationCommand) => }, } satisfies OrchestrationCommand; }); + +export const cleanupFailedUploadedAttachments = Effect.fn( + "Normalizer.cleanupFailedUploadedAttachments", +)(function* (command: ClientOrchestrationCommand, normalizedCommand: OrchestrationCommand) { + if (command.type !== "thread.turn.start" || normalizedCommand.type !== "thread.turn.start") { + return; + } + + const serverConfig = yield* ServerConfig; + const claimedPaths: string[] = []; + for (const [index, attachment] of normalizedCommand.message.attachments.entries()) { + const original = command.message.attachments[index]; + if ( + !original || + "dataUrl" in original || + parseThreadSegmentFromAttachmentId(original.id) !== PENDING_ATTACHMENT_THREAD_SEGMENT + ) { + continue; + } + + const claimedPath = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }); + if (claimedPath) { + claimedPaths.push(claimedPath); + } + } + yield* removeClaimedAttachmentPaths(claimedPaths); +}); diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 04d54ea8effb..f7147106c7a9 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -8,7 +8,7 @@ import * as Option from "effect/Option"; import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; import { projectThreadDetailSnapshot } from "./ActivityPayloadProjection.ts"; -import { normalizeDispatchCommand } from "./Normalizer.ts"; +import { cleanupFailedUploadedAttachments, normalizeDispatchCommand } from "./Normalizer.ts"; import { annotateEnvironmentRequest, failEnvironmentInternal, @@ -96,13 +96,14 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( const normalizedCommand = yield* normalizeDispatchCommand(args.payload).pipe( Effect.catch(() => failEnvironmentInvalidRequest("invalid_command")), ); - return yield* orchestrationEngine - .dispatch(normalizedCommand) - .pipe( - Effect.catch((cause) => - failEnvironmentInternal("orchestration_dispatch_failed", cause), - ), - ); + return yield* orchestrationEngine.dispatch(normalizedCommand).pipe( + Effect.tapError(() => + cleanupFailedUploadedAttachments(args.payload, normalizedCommand), + ), + Effect.catch((cause) => + failEnvironmentInternal("orchestration_dispatch_failed", cause), + ), + ); }), ); }), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 02a367c08792..5e4f19172eff 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -4486,6 +4486,55 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("uploads image bytes through a signed URL issued by websocket rpc", () => + Effect.gen(function* () { + const config = yield* buildAppUnderTest(); + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const wsUrl = yield* getWsServerUrl("/ws"); + + yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + const issued = yield* client[WS_METHODS.attachmentsCreateUploadUrl]({ + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 6, + }); + const rejected = yield* HttpClient.post(issued.relativeUrl, { + body: HttpBody.uint8Array(new Uint8Array([1, 2, 3]), "image/png"), + }); + assert.equal(rejected.status, 400); + + const response = yield* HttpClient.post(issued.relativeUrl, { + headers: { origin: crossOriginClientOrigin }, + body: HttpBody.uint8Array(new Uint8Array([1, 2, 3, 4, 5, 6]), "image/png"), + }); + assert.equal(response.status, 204); + assertBrowserApiCorsResponseHeaders(response.headers); + + const attachmentPath = path.join(config.attachmentsDir, `${issued.attachmentId}.png`); + assert.isTrue(yield* fileSystem.exists(attachmentPath)); + + yield* client[WS_METHODS.attachmentsDelete]({ attachmentId: issued.attachmentId }); + assert.isFalse(yield* fileSystem.exists(attachmentPath)); + + const streamed = yield* client[WS_METHODS.attachmentsCreateUploadUrl]({ + name: "streamed.png", + mimeType: "image/png", + sizeBytes: 6, + }); + const streamedResponse = yield* HttpClient.post(streamed.relativeUrl, { + body: HttpBody.stream(Stream.make(new Uint8Array([1, 2, 3, 4, 5, 6])), "image/png"), + }); + assert.equal(streamedResponse.status, 204); + yield* client[WS_METHODS.attachmentsDelete]({ attachmentId: streamed.attachmentId }); + }), + ), + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("keeps feedback errors structured across websocket rpc", () => Effect.gen(function* () { const threadId = ThreadId.make("thread-feedback-failure"); @@ -7993,12 +8042,14 @@ it.layer(NodeServices.layer)("server router seam", (it) => { it.effect("cleans up created bootstrap threads when worktree creation defects", () => Effect.gen(function* () { const dispatchedCommands: Array = []; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const createWorktree = vi.fn( (_: Parameters[0]) => Effect.die(new Error("worktree exploded")), ); - yield* buildAppUnderTest({ + const config = yield* buildAppUnderTest({ layers: { gitVcsDriver: { createWorktree, @@ -8016,40 +8067,62 @@ it.layer(NodeServices.layer)("server router seam", (it) => { const createdAt = "2026-01-01T00:00:00.000Z"; const wsUrl = yield* getWsServerUrl("/ws"); + let pendingAttachmentId: string | undefined; const result = yield* Effect.scoped( withWsRpcClient(wsUrl, (client) => - client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ - type: "thread.turn.start", - commandId: CommandId.make("cmd-bootstrap-turn-start-defect"), - threadId: ThreadId.make("thread-bootstrap-defect"), - message: { - messageId: MessageId.make("msg-bootstrap-defect"), - role: "user", - text: "hello", - attachments: [], - }, - modelSelection: defaultModelSelection, - runtimeMode: "full-access", - interactionMode: "default", - bootstrap: { - createThread: { - projectId: defaultProjectId, - title: "Bootstrap Thread", - modelSelection: defaultModelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: "main", - worktreePath: null, - createdAt, + Effect.gen(function* () { + const upload = yield* client[WS_METHODS.attachmentsCreateUploadUrl]({ + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 6, + }); + pendingAttachmentId = upload.attachmentId; + const uploadResponse = yield* HttpClient.post(upload.relativeUrl, { + body: HttpBody.uint8Array(new Uint8Array([1, 2, 3, 4, 5, 6]), "image/png"), + }); + assert.equal(uploadResponse.status, 204); + + return yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-bootstrap-turn-start-defect"), + threadId: ThreadId.make("thread-bootstrap-defect"), + message: { + messageId: MessageId.make("msg-bootstrap-defect"), + role: "user", + text: "hello", + attachments: [ + { + type: "image", + id: upload.attachmentId, + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 6, + }, + ], }, - prepareWorktree: { - projectCwd: "/tmp/project", - baseBranch: "main", - branch: "t3code/bootstrap-refName", + modelSelection: defaultModelSelection, + runtimeMode: "full-access", + interactionMode: "default", + bootstrap: { + createThread: { + projectId: defaultProjectId, + title: "Bootstrap Thread", + modelSelection: defaultModelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: "main", + worktreePath: null, + createdAt, + }, + prepareWorktree: { + projectCwd: "/tmp/project", + baseBranch: "main", + branch: "t3code/bootstrap-refName", + }, + runSetupScript: false, }, - runSetupScript: false, - }, - createdAt, + createdAt, + }); }), ).pipe(Effect.result), ); @@ -8062,6 +8135,13 @@ it.layer(NodeServices.layer)("server router seam", (it) => { dispatchedCommands.map((command) => command.type), ["thread.create", "thread.delete"], ); + assert.isDefined(pendingAttachmentId); + assert.isTrue( + yield* fileSystem.exists(path.join(config.attachmentsDir, `${pendingAttachmentId}.png`)), + ); + assert.deepEqual(yield* fileSystem.readDirectory(config.attachmentsDir), [ + `${pendingAttachmentId}.png`, + ]); }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 3e41b4390f82..0a31bf376dae 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -13,6 +13,7 @@ import * as ServerConfig from "./config.ts"; import { otlpTracesProxyRouteLayer, assetRouteLayer, + attachmentUploadRouteLayer, serverEnvironmentHttpApiLayer, staticAndDevRouteLayer, browserApiCorsLayer, @@ -456,6 +457,7 @@ export const makeRoutesLayer = Layer.mergeAll( ), otlpTracesProxyRouteLayer, assetRouteLayer, + attachmentUploadRouteLayer, staticAndDevRouteLayer, websocketRpcRouteLayer, ), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 11c659e28a70..55b0be07c667 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -73,7 +73,10 @@ import { projectActivityEvent, projectThreadDetailSnapshot, } from "./orchestration/ActivityPayloadProjection.ts"; -import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; +import { + cleanupFailedUploadedAttachments, + normalizeDispatchCommand, +} from "./orchestration/Normalizer.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { @@ -92,6 +95,7 @@ import * as TerminalManager from "./terminal/Manager.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as PreviewManager from "./preview/Manager.ts"; import { issueAssetUrl } from "./assets/AssetAccess.ts"; +import { deletePendingAttachment, issueAttachmentUploadUrl } from "./assets/AttachmentUpload.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts"; import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts"; @@ -1159,7 +1163,9 @@ const makeWsRpcLayer = ( ), ) : false; - const result = yield* dispatchNormalizedCommand(normalizedCommand); + const result = yield* dispatchNormalizedCommand(normalizedCommand).pipe( + Effect.tapError(() => cleanupFailedUploadedAttachments(command, normalizedCommand)), + ); yield* recordClientCommandAnalytics(normalizedCommand); if (parkingCommand) { const parkingKind = parkingCommand.type === "thread.archive" ? "archive" : "settle"; @@ -1951,6 +1957,16 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "workspace" }, ), + [WS_METHODS.attachmentsCreateUploadUrl]: (input) => + observeRpcEffect(WS_METHODS.attachmentsCreateUploadUrl, issueAttachmentUploadUrl(input), { + "rpc.aggregate": "workspace", + }), + [WS_METHODS.attachmentsDelete]: (input) => + observeRpcEffect( + WS_METHODS.attachmentsDelete, + deletePendingAttachment(input.attachmentId), + { "rpc.aggregate": "workspace" }, + ), [WS_METHODS.assetsCreateUrl]: (input) => observeRpcEffect( WS_METHODS.assetsCreateUrl, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 46ed051154a6..cb1cf698535a 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -352,6 +352,12 @@ import { import type { ThreadSyncPhase } from "../threadSync"; import { useLocalStorage } from "~/hooks/useLocalStorage"; import { useComposerHandleContext } from "../composerHandleContext"; +import { + awaitAttachmentUploads, + getUploadedAttachments, + releaseAttachmentUploads, + startAttachmentUpload, +} from "../lib/attachmentUploadQueue"; import { sanitizeThreadErrorMessage } from "~/rpc/transportError"; import { RightPanelSheet } from "./RightPanelSheet"; import { previewEnvironment } from "../state/preview"; @@ -2094,6 +2100,10 @@ function ChatViewContent(props: ChatViewProps) { : (primaryEnvironment?.serverConfig ?? null); const pullRequestsCapabilityKnown = serverConfig !== null; const supportsPullRequests = serverConfig?.environment.capabilities.pullRequests === true; + const attachmentEnvironmentConfig = environmentById.get(environmentId)?.serverConfig ?? null; + const attachmentUploadsCapabilityKnown = attachmentEnvironmentConfig !== null; + const supportsAttachmentUploads = + attachmentEnvironmentConfig?.environment.capabilities.attachmentUploads === true; const versionMismatch = resolveServerConfigVersionMismatch(serverConfig); const versionMismatchDismissKey = versionMismatch && activeThread @@ -5406,9 +5416,21 @@ function ChatViewContent(props: ChatViewProps) { return; } + sendInFlightRef.current = true; + if (supportsAttachmentUploads && composerImagesSnapshot.length > 0) { + for (const image of composerImagesSnapshot) { + startAttachmentUpload({ environmentId, image }); + } + await awaitAttachmentUploads(composerImagesSnapshot.map((image) => image.id)); + if (getUploadedAttachments({ environmentId, images: composerImagesSnapshot }) === null) { + sendInFlightRef.current = false; + setThreadError(threadIdForSend, "Retry or remove failed image uploads before sending."); + return; + } + } + const resolvedSubmissionIntent = submissionIntent === "background" && isLocalDraftThread ? "background" : "foreground"; - sendInFlightRef.current = true; if ( shouldDockDraftHeroForSubmission({ isDraftHeroState, @@ -5439,13 +5461,22 @@ function ChatViewContent(props: ChatViewProps) { const messageIdForSend = newMessageId(); const messageCreatedAt = new Date().toISOString(); const turnAttachmentsPromise = Promise.all( - composerImagesSnapshot.map(async (image) => ({ - type: "image" as const, - name: image.name, - mimeType: image.mimeType, - sizeBytes: image.sizeBytes, - dataUrl: await readFileAsDataUrl(image.file), - })), + composerImagesSnapshot.map(async (image) => { + if (supportsAttachmentUploads) { + const uploaded = getUploadedAttachments({ environmentId, images: [image] })?.[0]; + if (!uploaded) { + throw new Error(`Image '${image.name}' did not finish uploading.`); + } + return uploaded; + } + return { + type: "image" as const, + name: image.name, + mimeType: image.mimeType, + sizeBytes: image.sizeBytes, + dataUrl: await readFileAsDataUrl(image.file), + }; + }), ); const optimisticAttachments = composerImagesSnapshot.map((image) => ({ type: "image" as const, @@ -5632,6 +5663,9 @@ function ChatViewContent(props: ChatViewProps) { failure = startResult; } else { turnStartSucceeded = true; + if (supportsAttachmentUploads) { + releaseAttachmentUploads(composerImagesSnapshot); + } acknowledgeActiveThreadWoke(); if (backgroundThreadRef) { markPromotedDraftThreadByRef(backgroundThreadRef); @@ -6774,6 +6808,8 @@ function ChatViewContent(props: ChatViewProps) { composerRef={composerRef} composerDraftTarget={composerDraftTarget} environmentId={environmentId} + attachmentUploadsCapabilityKnown={attachmentUploadsCapabilityKnown} + supportsAttachmentUploads={supportsAttachmentUploads} routeKind={routeKind} routeThreadRef={routeThreadRef} draftId={draftId} diff --git a/apps/web/src/components/LegacySidebar.tsx b/apps/web/src/components/LegacySidebar.tsx index b4f3b5996818..4555544046fa 100644 --- a/apps/web/src/components/LegacySidebar.tsx +++ b/apps/web/src/components/LegacySidebar.tsx @@ -75,6 +75,7 @@ import { useDesktopLocalBootstraps } from "../connection/useDesktopLocalBootstra import { isElectron } from "../env"; import { useTerminalFocus } from "../hooks/useTerminalFocus"; import { useOpenPrLink } from "../lib/openPullRequestLink"; +import { releaseProjectDraftUploads } from "../lib/composerDraftUploads"; import { isTerminalFocused } from "../lib/terminalFocus"; import { isMacPlatform } from "../lib/utils"; import { @@ -1462,6 +1463,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec return result; } const draftStore = useComposerDraftStore.getState(); + releaseProjectDraftUploads(memberProjectRef); const projectDraftThread = draftStore.getDraftThreadByProjectRef(memberProjectRef); if (projectDraftThread) { draftStore.clearDraftThread(projectDraftThread.draftId); diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 971ead810f07..7a80559d390d 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -89,6 +89,7 @@ import { isModelPickerOpen } from "../modelPickerVisibility"; import { selectThreadTerminalUiState, useTerminalUiStateStore } from "../terminalUiStateStore"; import { isMacPlatform } from "~/lib/utils"; import { useOpenPrLink } from "../lib/openPullRequestLink"; +import { releaseComposerDraftUploads } from "../lib/composerDraftUploads"; import { readLocalApi } from "../localApi"; import { getProjectOrderKey, selectProjectGroupingSettings } from "../logicalProject"; import { @@ -654,6 +655,7 @@ const SidebarDraftBlock = memo(function SidebarDraftBlock(props: { // The /draft/$draftId route redirects home on its own when the draft // it renders disappears, so discarding the open draft needs no // special-casing here. + releaseComposerDraftUploads(draftId); clearDraftThread(draftId); }, [clearDraftThread], diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index f29d6c2b4f6a..929a8c1c03fa 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -74,6 +74,16 @@ import { type ComposerTasksProgress, } from "./ComposerTasksBadge"; import { compressImageForStash, compressImageToByteLimit } from "../../lib/imageCompression"; +import { + releaseAttachmentUpload, + retryAttachmentUpload, + startAttachmentUpload, + useAttachmentUploadStore, +} from "../../lib/attachmentUploadQueue"; +import { + attachmentUploadBlockReason, + formatAttachmentUploadProgress, +} from "../../lib/attachmentUploadState"; import { isCommandPaletteOpen } from "../../commandPaletteBus"; import { getTerminalFocusOwner } from "../../lib/terminalFocus"; import { resolveShortcutCommand } from "../../keybindings"; @@ -232,6 +242,7 @@ import { LockIcon, LockOpenIcon, PenLineIcon, + RotateCcwIcon, SparklesIcon, XIcon, } from "lucide-react"; @@ -533,6 +544,8 @@ export interface ChatComposerHandle { export interface ChatComposerProps { composerDraftTarget: ScopedThreadRef | DraftId; environmentId: EnvironmentId; + attachmentUploadsCapabilityKnown: boolean; + supportsAttachmentUploads: boolean; routeKind: "server" | "draft"; routeThreadRef: ScopedThreadRef; draftId: DraftId | null; @@ -647,6 +660,8 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const { composerDraftTarget, environmentId, + attachmentUploadsCapabilityKnown, + supportsAttachmentUploads, routeKind, routeThreadRef, draftId, @@ -660,7 +675,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) phase, isConnecting, isSendBusy, - sendDisabledReason, + sendDisabledReason: externalSendDisabledReason, isPreparingWorktree, environmentUnavailable, activePendingApproval, @@ -711,8 +726,6 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) setThreadError, onExpandImage, } = props; - const isSendDisabled = sendDisabledReason !== null; - // ------------------------------------------------------------------ // Store subscriptions (prompt / images / terminal contexts) // ------------------------------------------------------------------ @@ -724,6 +737,17 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const composerPreviewAnnotations = composerDraft.previewAnnotations; const composerReviewComments = composerDraft.reviewComments; const nonPersistedComposerImageIds = composerDraft.nonPersistedImageIds; + const uploadsByImageId = useAttachmentUploadStore((state) => state.uploadsByImageId); + const attachmentBlockReason = supportsAttachmentUploads + ? attachmentUploadBlockReason({ + imageIds: composerImages.map((image) => image.id), + uploadsByImageId, + environmentId, + }) + : null; + const sendDisabledReason = + externalSendDisabledReason ?? (activePendingProgress ? null : attachmentBlockReason); + const isSendDisabled = sendDisabledReason !== null; const setComposerDraftPrompt = useComposerDraftStore((store) => store.setPrompt); const addComposerDraftImage = useComposerDraftStore((store) => store.addImage); @@ -758,6 +782,21 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) ); const getComposerDraft = useComposerDraftStore((store) => store.getComposerDraft); + useEffect(() => { + if (!attachmentUploadsCapabilityKnown) { + return; + } + if (!supportsAttachmentUploads) { + for (const image of composerImages) { + releaseAttachmentUpload(image.id); + } + return; + } + for (const image of composerImages) { + startAttachmentUpload({ environmentId, image }); + } + }, [attachmentUploadsCapabilityKnown, composerImages, environmentId, supportsAttachmentUploads]); + // ------------------------------------------------------------------ // Model state // ------------------------------------------------------------------ @@ -1337,6 +1376,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) const removeComposerImageFromDraft = useCallback( (imageId: string) => { + releaseAttachmentUpload(imageId); removeComposerDraftImage(composerDraftTarget, imageId); }, [composerDraftTarget, removeComposerDraftImage], @@ -2244,6 +2284,9 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) // destroying them here would be unrecoverable. promptRef.current = ""; clearComposerDraftPromptAndImages(stashTarget); + for (const image of images) { + releaseAttachmentUpload(image.id); + } setComposerCursor(0); setComposerTrigger(null); pulseStashBadge(); @@ -3104,9 +3147,17 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) - removeComposerDraftPreviewAnnotation(composerDraftTarget, annotationId) - } + {...(supportsAttachmentUploads + ? { + uploadsByImageId, + onRetryUpload: (image: ComposerImageAttachment) => + retryAttachmentUpload({ environmentId, image }), + } + : {})} + onRemove={(annotationId) => { + releaseAttachmentUpload(annotationId); + removeComposerDraftPreviewAnnotation(composerDraftTarget, annotationId); + }} onExpandImage={(imageId) => { const preview = buildExpandedImagePreview(composerImages, imageId); if (preview) onExpandImage(preview); @@ -3156,66 +3207,104 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) (annotation) => annotation.id === image.id, ), ) - .map((image) => ( -
- {image.previewUrl ? ( - - ) : ( -
- {image.name} -
- )} - {nonPersistedComposerImageIdSet.has(image.id) && ( - - - - - } - /> - - Draft attachment could not be saved locally and may be lost on - navigation. - - - )} - -
- ))} + {image.previewUrl ? ( + + ) : ( +
+ {image.name} +
+ )} + {nonPersistedComposerImageIdSet.has(image.id) && ( + + + + + } + /> + + Draft attachment could not be saved locally and may be lost on + navigation. + + + )} + {upload?.status === "uploading" && ( + + {formatAttachmentUploadProgress(upload.progress)} + + )} + {upload?.status === "failed" && ( + + + retryAttachmentUpload({ environmentId, image }) + } + aria-label={`Retry upload for ${image.name}`} + /> + } + > + + + + {upload.reason} + + + )} + + + ); + })} )} diff --git a/apps/web/src/components/chat/ComposerPreviewAnnotationCards.test.tsx b/apps/web/src/components/chat/ComposerPreviewAnnotationCards.test.tsx index 4ee61b556a03..08c38faafd5e 100644 --- a/apps/web/src/components/chat/ComposerPreviewAnnotationCards.test.tsx +++ b/apps/web/src/components/chat/ComposerPreviewAnnotationCards.test.tsx @@ -1,4 +1,4 @@ -import type { PreviewAnnotationPayload } from "@t3tools/contracts"; +import { EnvironmentId, type PreviewAnnotationPayload } from "@t3tools/contracts"; import { renderToStaticMarkup } from "react-dom/server"; import { describe, expect, it, vi } from "vite-plus/test"; @@ -58,4 +58,34 @@ describe("ComposerPreviewAnnotationCards", () => { expect(markup).toContain('aria-label="Remove preview annotation"'); expect(markup).toContain('data-slot="button"'); }); + + it("shows a retry action for a failed screenshot upload", () => { + const image = { + type: "image" as const, + id: annotation.id, + name: "annotation.png", + mimeType: "image/png", + sizeBytes: 3, + previewUrl: "blob:annotation", + file: new File([new Uint8Array([1, 2, 3])], "annotation.png", { type: "image/png" }), + }; + const markup = renderToStaticMarkup( + , + ); + + expect(markup).toContain('aria-label="Retry upload for annotation.png"'); + }); }); diff --git a/apps/web/src/components/chat/ComposerPreviewAnnotationCards.tsx b/apps/web/src/components/chat/ComposerPreviewAnnotationCards.tsx index 0287dde73aaf..8eb7e9897b8e 100644 --- a/apps/web/src/components/chat/ComposerPreviewAnnotationCards.tsx +++ b/apps/web/src/components/chat/ComposerPreviewAnnotationCards.tsx @@ -1,9 +1,13 @@ import type { PreviewAnnotationPayload } from "@t3tools/contracts"; -import { Frame, MousePointerClick, Paintbrush, PenLine, X } from "lucide-react"; +import { Frame, MousePointerClick, Paintbrush, PenLine, RotateCcw, X } from "lucide-react"; import type { ReactNode } from "react"; import type { ComposerImageAttachment } from "~/composerDraftStore"; import { formatElementContextLabel, normalizeElementContextSelection } from "~/lib/elementContext"; +import { + formatAttachmentUploadProgress, + type AttachmentUploadState, +} from "~/lib/attachmentUploadState"; import { cn } from "~/lib/utils"; import { Button } from "../ui/button"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; @@ -13,6 +17,8 @@ interface ComposerPreviewAnnotationCardsProps { images: ReadonlyArray; onRemove: (annotationId: string) => void; onExpandImage: (imageId: string) => void; + uploadsByImageId?: Readonly>; + onRetryUpload?: (image: ComposerImageAttachment) => void; className?: string; } @@ -38,6 +44,8 @@ export function ComposerPreviewAnnotationCards({ images, onRemove, onExpandImage, + uploadsByImageId, + onRetryUpload, className, }: ComposerPreviewAnnotationCardsProps) { if (annotations.length === 0) return null; @@ -47,6 +55,7 @@ export function ComposerPreviewAnnotationCards({
{annotations.map((annotation) => { const image = imagesById.get(annotation.id); + const upload = image ? uploadsByImageId?.[image.id] : undefined; const elementLabels = annotation.elements.flatMap((target) => { const context = normalizeElementContextSelection(target.element); return context ? [{ id: target.id, label: formatElementContextLabel(context) }] : []; @@ -132,6 +141,28 @@ export function ComposerPreviewAnnotationCards({ label="style change" /> ) : null} + {upload?.status === "uploading" ? ( + + {formatAttachmentUploadProgress(upload.progress)} + + ) : null} + {upload?.status === "failed" && image && onRetryUpload ? ( + + onRetryUpload(image)} + /> + } + > + + + {upload.reason} + + ) : null}
diff --git a/apps/web/src/components/settings/ProjectSettingsPanel.tsx b/apps/web/src/components/settings/ProjectSettingsPanel.tsx index 6768d2dc61ef..6047b8fc48dc 100644 --- a/apps/web/src/components/settings/ProjectSettingsPanel.tsx +++ b/apps/web/src/components/settings/ProjectSettingsPanel.tsx @@ -46,6 +46,7 @@ import { useCopyToClipboard } from "../../hooks/useCopyToClipboard"; import { useT3ProjectFileState } from "../../hooks/useT3ProjectFileScripts"; import { shortcutLabelForCommand } from "../../keybindings"; import { keybindingValueForCommand } from "../../lib/projectScriptKeybindings"; +import { releaseProjectDraftUploads } from "../../lib/composerDraftUploads"; import { readLocalApi } from "../../localApi"; import { buildProjectScript, @@ -722,6 +723,7 @@ function ProjectDetail({ group }: { group: SidebarProjectSnapshot }) { return; } const projectRef = scopeProjectRef(member.environmentId, member.id); + releaseProjectDraftUploads(projectRef); const projectDraftThread = draftStore.getDraftThreadByProjectRef(projectRef); if (projectDraftThread) { draftStore.clearDraftThread(projectDraftThread.draftId); diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 569b4be96e62..f40920779b0f 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -20,6 +20,7 @@ import { threadEnvironment } from "../state/threads"; import { vcsEnvironment } from "../state/vcs"; import { useNewThreadHandler } from "./useHandleNewThread"; import { refreshArchivedThreadsForEnvironment } from "../lib/archivedThreadsState"; +import { releaseComposerDraftUploads } from "../lib/composerDraftUploads"; import { readLocalApi } from "../localApi"; import { readEnvironmentSupportsPinning, @@ -364,6 +365,7 @@ export function useThreadActions() { return deleteResult; } refreshArchivedThreadsForEnvironment(threadRef.environmentId); + releaseComposerDraftUploads(threadRef); clearComposerDraftForThread(threadRef); clearProjectDraftThreadById( scopeProjectRef(threadRef.environmentId, thread.projectId), diff --git a/apps/web/src/lib/attachmentUploadQueue.test.ts b/apps/web/src/lib/attachmentUploadQueue.test.ts new file mode 100644 index 000000000000..2b2b94431c80 --- /dev/null +++ b/apps/web/src/lib/attachmentUploadQueue.test.ts @@ -0,0 +1,306 @@ +import { EnvironmentId } from "@t3tools/contracts"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +import type { ComposerImageAttachment } from "../composerDraftStore"; + +const mocks = vi.hoisted(() => ({ + createUploadUrl: Symbol("create-upload-url"), + removeUpload: Symbol("remove-upload"), + runAtomCommand: vi.fn(), + readPreparedConnection: vi.fn(), +})); + +vi.mock("@t3tools/client-runtime/state/runtime", () => ({ + runAtomCommand: mocks.runAtomCommand, +})); + +vi.mock("../rpc/atomRegistry", () => ({ appAtomRegistry: {} })); + +vi.mock("../state/attachments", () => ({ + attachmentEnvironment: { + createUploadUrl: mocks.createUploadUrl, + remove: mocks.removeUpload, + }, +})); + +vi.mock("../state/session", () => ({ + readPreparedConnection: mocks.readPreparedConnection, +})); + +import { + awaitAttachmentUploads, + getUploadedAttachments, + readAttachmentUpload, + releaseAttachmentUpload, + releaseAttachmentUploads, + retryAttachmentUpload, + startAttachmentUpload, + useAttachmentUploadStore, +} from "./attachmentUploadQueue"; + +type ProgressListener = (event: { + readonly lengthComputable: boolean; + readonly loaded: number; + readonly total: number; +}) => void; + +class TestXmlHttpRequest { + static requests: TestXmlHttpRequest[] = []; + + status = 0; + timeout = 0; + method: string | null = null; + url: string | null = null; + readonly headers = new Map(); + readonly listeners = new Map void>(); + progressListener: ProgressListener | null = null; + + readonly upload = { + addEventListener: (_event: string, listener: ProgressListener) => { + this.progressListener = listener; + }, + }; + + constructor() { + TestXmlHttpRequest.requests.push(this); + } + + open(method: string, url: string): void { + this.method = method; + this.url = url; + } + + setRequestHeader(name: string, value: string): void { + this.headers.set(name, value); + } + + addEventListener(event: string, listener: () => void): void { + this.listeners.set(event, listener); + } + + send(): void {} + + abort(): void { + this.listeners.get("abort")?.(); + } + + progress(loaded: number, total: number): void { + this.progressListener?.({ lengthComputable: true, loaded, total }); + } + + complete(status = 204): void { + this.status = status; + this.listeners.get("load")?.(); + } +} + +const firstEnvironment = EnvironmentId.make("environment-1"); +const secondEnvironment = EnvironmentId.make("environment-2"); + +function makeImage(id: string): ComposerImageAttachment { + const file = new File([new Uint8Array([1, 2, 3])], `${id}.png`, { type: "image/png" }); + return { + type: "image", + id, + name: file.name, + mimeType: file.type, + sizeBytes: file.size, + previewUrl: `blob:${id}`, + file, + }; +} + +describe("attachmentUploadQueue", () => { + beforeEach(() => { + TestXmlHttpRequest.requests = []; + mocks.runAtomCommand.mockReset(); + mocks.readPreparedConnection.mockReset(); + mocks.readPreparedConnection.mockReturnValue({ httpBaseUrl: "https://environment.test/" }); + mocks.runAtomCommand.mockImplementation( + async ( + _registry: unknown, + command: unknown, + target: { + readonly environmentId: EnvironmentId; + readonly input: { readonly name?: string }; + }, + ) => { + if (command === mocks.createUploadUrl) { + const attachmentId = `pending-${target.environmentId}-${target.input.name}`; + return { + _tag: "Success", + value: { + attachmentId, + relativeUrl: `/api/attachments/upload/${attachmentId}`, + expiresAt: 1, + }, + }; + } + return { _tag: "Success", value: undefined }; + }, + ); + vi.stubGlobal("XMLHttpRequest", TestXmlHttpRequest); + }); + + afterEach(() => { + for (const imageId of Object.keys(useAttachmentUploadStore.getState().uploadsByImageId)) { + releaseAttachmentUpload(imageId); + } + vi.unstubAllGlobals(); + }); + + it("uploads images immediately and sends attachment references", async () => { + const image = makeImage("image-1"); + startAttachmentUpload({ environmentId: firstEnvironment, image }); + await Promise.resolve(); + + const request = TestXmlHttpRequest.requests[0]!; + expect(request.method).toBe("POST"); + expect(request.url).toBe( + "https://environment.test/api/attachments/upload/pending-environment-1-image-1.png", + ); + request.progress(1, 3); + expect(readAttachmentUpload(image.id)).toMatchObject({ status: "uploading", progress: 1 / 3 }); + + const settled = awaitAttachmentUploads([image.id]); + request.complete(); + await settled; + + expect(getUploadedAttachments({ environmentId: firstEnvironment, images: [image] })).toEqual([ + { + type: "image", + id: "pending-environment-1-image-1.png", + name: "image-1.png", + mimeType: "image/png", + sizeBytes: 3, + }, + ]); + + releaseAttachmentUploads([image]); + expect(readAttachmentUpload(image.id)).toBeUndefined(); + expect(mocks.runAtomCommand).toHaveBeenCalledWith( + expect.anything(), + mocks.removeUpload, + { + environmentId: firstEnvironment, + input: { attachmentId: "pending-environment-1-image-1.png" }, + }, + expect.anything(), + ); + }); + + it("retries rejected uploads", async () => { + const image = makeImage("image-retry"); + startAttachmentUpload({ environmentId: firstEnvironment, image }); + await Promise.resolve(); + + let settled = awaitAttachmentUploads([image.id]); + TestXmlHttpRequest.requests[0]!.complete(500); + await settled; + expect(readAttachmentUpload(image.id)).toMatchObject({ + status: "failed", + reason: "Upload rejected (500)", + }); + + retryAttachmentUpload({ environmentId: firstEnvironment, image }); + await Promise.resolve(); + settled = awaitAttachmentUploads([image.id]); + TestXmlHttpRequest.requests[1]!.complete(); + await settled; + + expect(readAttachmentUpload(image.id)).toMatchObject({ status: "ready" }); + }); + + it("releases an upload URL that resolves after its image was removed", async () => { + const image = makeImage("image-cancelled"); + const minted = { + _tag: "Success" as const, + value: { + attachmentId: "pending-environment-1-image-cancelled.png", + relativeUrl: "/api/attachments/upload/cancelled", + expiresAt: 1, + }, + }; + let resolveMint: (result: typeof minted) => void = () => {}; + const pendingMint = new Promise((resolve) => { + resolveMint = resolve; + }); + let resolveDelete: () => void = () => {}; + const deleted = new Promise((resolve) => { + resolveDelete = resolve; + }); + mocks.runAtomCommand.mockImplementation((_registry: unknown, command: unknown) => { + if (command === mocks.createUploadUrl) { + return pendingMint; + } + resolveDelete(); + return Promise.resolve({ _tag: "Success", value: undefined }); + }); + + startAttachmentUpload({ environmentId: firstEnvironment, image }); + releaseAttachmentUpload(image.id); + resolveMint(minted); + await deleted; + + expect(TestXmlHttpRequest.requests).toEqual([]); + expect(readAttachmentUpload(image.id)).toBeUndefined(); + expect(mocks.runAtomCommand).toHaveBeenCalledWith( + expect.anything(), + mocks.removeUpload, + { + environmentId: firstEnvironment, + input: { attachmentId: minted.value.attachmentId }, + }, + expect.anything(), + ); + }); + + it("restores the previous environment after a replacement upload fails", async () => { + const image = makeImage("image-move"); + startAttachmentUpload({ environmentId: firstEnvironment, image }); + await Promise.resolve(); + let settled = awaitAttachmentUploads([image.id]); + TestXmlHttpRequest.requests[0]!.complete(); + await settled; + + startAttachmentUpload({ environmentId: secondEnvironment, image }); + await Promise.resolve(); + settled = awaitAttachmentUploads([image.id]); + TestXmlHttpRequest.requests[1]!.complete(500); + await settled; + + startAttachmentUpload({ environmentId: firstEnvironment, image }); + expect(readAttachmentUpload(image.id)).toMatchObject({ + status: "ready", + environmentId: firstEnvironment, + attachmentId: "pending-environment-1-image-move.png", + }); + }); + + it("does not let stalled uploads block another environment", async () => { + const images = ["image-a", "image-b", "image-c", "image-d"].map(makeImage); + for (const image of images) { + startAttachmentUpload({ environmentId: firstEnvironment, image }); + } + const otherEnvironmentImage = makeImage("image-other"); + startAttachmentUpload({ environmentId: secondEnvironment, image: otherEnvironmentImage }); + await Promise.resolve(); + + expect(TestXmlHttpRequest.requests).toHaveLength(4); + const otherRequest = TestXmlHttpRequest.requests.find((request) => + request.url?.includes("environment-2"), + ); + expect(otherRequest).toBeDefined(); + + for (const request of TestXmlHttpRequest.requests) { + request.complete(); + } + await Promise.all([ + ...images.slice(0, 3).map((image) => awaitAttachmentUploads([image.id])), + awaitAttachmentUploads([otherEnvironmentImage.id]), + ]); + await Promise.resolve(); + TestXmlHttpRequest.requests[4]!.complete(); + await awaitAttachmentUploads([images[3]!.id]); + }); +}); diff --git a/apps/web/src/lib/attachmentUploadQueue.ts b/apps/web/src/lib/attachmentUploadQueue.ts new file mode 100644 index 000000000000..37eb924ca256 --- /dev/null +++ b/apps/web/src/lib/attachmentUploadQueue.ts @@ -0,0 +1,389 @@ +import { + PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES, + type ChatAttachment, + type EnvironmentId, +} from "@t3tools/contracts"; +import { resolveAssetUrl } from "@t3tools/client-runtime/state/assets"; +import { runAtomCommand } from "@t3tools/client-runtime/state/runtime"; +import { create } from "zustand"; + +import type { ComposerImageAttachment } from "../composerDraftStore"; +import { appAtomRegistry } from "../rpc/atomRegistry"; +import { attachmentEnvironment } from "../state/attachments"; +import { readPreparedConnection } from "../state/session"; +import type { AttachmentUploadState, ReadyAttachmentUpload } from "./attachmentUploadState"; + +const MAX_UPLOADS_PER_ENVIRONMENT = 3; +const UPLOAD_TIMEOUT_MS = 5 * 60_000; + +interface AttachmentUploadStore { + readonly uploadsByImageId: Readonly>; +} + +export const useAttachmentUploadStore = create(() => ({ + uploadsByImageId: {}, +})); + +interface UploadJob { + readonly image: ComposerImageAttachment; + readonly environmentId: EnvironmentId; + readonly previous?: ReadyAttachmentUpload; + readonly settled: Promise; + resolveSettled: () => void; + attachmentId: string | null; + cancelled: boolean; + abort: (() => void) | null; +} + +const jobsByImageId = new Map(); +const queue: UploadJob[] = []; +const activeUploadsByEnvironment = new Map(); + +function setUploadState(imageId: string, upload: AttachmentUploadState): void { + useAttachmentUploadStore.setState((state) => ({ + uploadsByImageId: { ...state.uploadsByImageId, [imageId]: upload }, + })); +} + +function clearUploadState(imageId: string): void { + useAttachmentUploadStore.setState((state) => { + if (!(imageId in state.uploadsByImageId)) { + return state; + } + const uploadsByImageId = { ...state.uploadsByImageId }; + delete uploadsByImageId[imageId]; + return { uploadsByImageId }; + }); +} + +export function readAttachmentUpload(imageId: string): AttachmentUploadState | undefined { + return useAttachmentUploadStore.getState().uploadsByImageId[imageId]; +} + +function deletePendingUpload(environmentId: EnvironmentId, attachmentId: string): void { + void runAtomCommand( + appAtomRegistry, + attachmentEnvironment.remove, + { environmentId, input: { attachmentId } }, + { reportFailure: false, reportDefect: false }, + ); +} + +function uploadBytes(input: { + readonly url: string; + readonly file: File; + readonly onProgress: (progress: number) => void; +}): { readonly done: Promise; readonly abort: () => void } { + const xhr = new XMLHttpRequest(); + const done = new Promise((resolve, reject) => { + xhr.open("POST", input.url, true); + xhr.timeout = UPLOAD_TIMEOUT_MS; + xhr.setRequestHeader("Content-Type", input.file.type); + xhr.upload.addEventListener("progress", (event) => { + if (event.lengthComputable && event.total > 0) { + input.onProgress(event.loaded / event.total); + } + }); + xhr.addEventListener("load", () => { + if (xhr.status >= 200 && xhr.status < 300) { + resolve(); + } else { + reject(new Error(`Upload rejected (${xhr.status})`)); + } + }); + xhr.addEventListener("error", () => reject(new Error("Upload failed"))); + xhr.addEventListener("timeout", () => reject(new Error("Upload timed out"))); + xhr.addEventListener("abort", () => reject(new Error("Upload cancelled"))); + xhr.send(input.file); + }); + + return { done, abort: () => xhr.abort() }; +} + +async function runUpload(job: UploadJob): Promise { + const mimeType = PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES.find( + (supportedMimeType) => supportedMimeType === job.image.mimeType.toLowerCase(), + ); + if (!mimeType) { + setUploadState(job.image.id, { + status: "failed", + environmentId: job.environmentId, + reason: "Unsupported image type", + ...(job.previous ? { previous: job.previous } : {}), + }); + return; + } + + const minted = await runAtomCommand( + appAtomRegistry, + attachmentEnvironment.createUploadUrl, + { + environmentId: job.environmentId, + input: { + name: job.image.name, + mimeType, + sizeBytes: job.image.file.size, + }, + }, + { reportFailure: false }, + ); + if (job.cancelled) { + if (minted._tag === "Success") { + deletePendingUpload(job.environmentId, minted.value.attachmentId); + } + return; + } + if (minted._tag !== "Success") { + setUploadState(job.image.id, { + status: "failed", + environmentId: job.environmentId, + reason: "Upload could not start", + ...(job.previous ? { previous: job.previous } : {}), + }); + return; + } + job.attachmentId = minted.value.attachmentId; + + const connection = readPreparedConnection(job.environmentId); + const url = connection ? resolveAssetUrl(connection.httpBaseUrl, minted.value.relativeUrl) : null; + if (!url) { + setUploadState(job.image.id, { + status: "failed", + environmentId: job.environmentId, + reason: "Not connected", + attachmentId: minted.value.attachmentId, + ...(job.previous ? { previous: job.previous } : {}), + }); + return; + } + + let lastStep = -1; + const upload = uploadBytes({ + url, + file: job.image.file, + onProgress: (progress) => { + const step = Math.floor(progress * 20); + if (step === lastStep || job.cancelled) { + return; + } + lastStep = step; + setUploadState(job.image.id, { + status: "uploading", + environmentId: job.environmentId, + progress, + ...(job.previous ? { previous: job.previous } : {}), + }); + }, + }); + job.abort = upload.abort; + + try { + await upload.done; + if (job.cancelled) { + return; + } + setUploadState(job.image.id, { + status: "ready", + environmentId: job.environmentId, + attachmentId: minted.value.attachmentId, + }); + if (job.previous) { + deletePendingUpload(job.previous.environmentId, job.previous.attachmentId); + } + } catch (error) { + if (job.cancelled) { + return; + } + setUploadState(job.image.id, { + status: "failed", + environmentId: job.environmentId, + reason: error instanceof Error ? error.message : "Upload failed", + attachmentId: minted.value.attachmentId, + ...(job.previous ? { previous: job.previous } : {}), + }); + } finally { + job.abort = null; + } +} + +function pumpUploads(): void { + for (let index = 0; index < queue.length; ) { + const job = queue[index]!; + const active = activeUploadsByEnvironment.get(job.environmentId) ?? 0; + if (active >= MAX_UPLOADS_PER_ENVIRONMENT) { + index += 1; + continue; + } + + queue.splice(index, 1); + if (job.cancelled) { + continue; + } + activeUploadsByEnvironment.set(job.environmentId, active + 1); + void runUpload(job) + .catch(() => { + if (!job.cancelled) { + setUploadState(job.image.id, { + status: "failed", + environmentId: job.environmentId, + reason: "Upload failed", + ...(job.previous ? { previous: job.previous } : {}), + }); + } + }) + .finally(() => { + if (jobsByImageId.get(job.image.id) === job) { + jobsByImageId.delete(job.image.id); + } + const remaining = (activeUploadsByEnvironment.get(job.environmentId) ?? 1) - 1; + if (remaining > 0) { + activeUploadsByEnvironment.set(job.environmentId, remaining); + } else { + activeUploadsByEnvironment.delete(job.environmentId); + } + job.resolveSettled(); + pumpUploads(); + }); + } +} + +export function startAttachmentUpload(input: { + readonly environmentId: EnvironmentId; + readonly image: ComposerImageAttachment; +}): void { + const existingJob = jobsByImageId.get(input.image.id); + if (existingJob?.environmentId === input.environmentId) { + return; + } + + const existing = readAttachmentUpload(input.image.id); + if (existing?.status === "ready" && existing.environmentId === input.environmentId) { + return; + } + if (existing?.status === "failed" && existing.environmentId === input.environmentId) { + return; + } + if ( + existing && + "previous" in existing && + existing.previous?.environmentId === input.environmentId + ) { + cancelAttachmentUpload(input.image.id); + if (existing.status === "failed" && existing.attachmentId) { + deletePendingUpload(existing.environmentId, existing.attachmentId); + } + setUploadState(input.image.id, existing.previous); + return; + } + + if (existingJob) { + cancelAttachmentUpload(input.image.id); + } + const previous = existing?.status === "ready" ? existing : existing?.previous; + let resolveSettled: () => void = () => {}; + const settled = new Promise((resolve) => { + resolveSettled = resolve; + }); + const job: UploadJob = { + image: input.image, + environmentId: input.environmentId, + ...(previous ? { previous } : {}), + settled, + resolveSettled, + attachmentId: null, + cancelled: false, + abort: null, + }; + + jobsByImageId.set(input.image.id, job); + queue.push(job); + setUploadState(input.image.id, { + status: "uploading", + environmentId: input.environmentId, + progress: 0, + ...(previous ? { previous } : {}), + }); + pumpUploads(); +} + +export function cancelAttachmentUpload(imageId: string): void { + const job = jobsByImageId.get(imageId); + if (!job) { + return; + } + job.cancelled = true; + jobsByImageId.delete(imageId); + const queuedIndex = queue.indexOf(job); + if (queuedIndex !== -1) { + queue.splice(queuedIndex, 1); + } + job.abort?.(); + if (job.attachmentId) { + deletePendingUpload(job.environmentId, job.attachmentId); + } + job.resolveSettled(); +} + +export function releaseAttachmentUpload(imageId: string): void { + const upload = readAttachmentUpload(imageId); + cancelAttachmentUpload(imageId); + if (upload?.status === "ready") { + deletePendingUpload(upload.environmentId, upload.attachmentId); + } else if (upload) { + if (upload.status === "failed" && upload.attachmentId) { + deletePendingUpload(upload.environmentId, upload.attachmentId); + } + if (upload.previous) { + deletePendingUpload(upload.previous.environmentId, upload.previous.attachmentId); + } + } + clearUploadState(imageId); +} + +export function retryAttachmentUpload(input: { + readonly environmentId: EnvironmentId; + readonly image: ComposerImageAttachment; +}): void { + const previous = readAttachmentUpload(input.image.id); + cancelAttachmentUpload(input.image.id); + if (previous?.status === "failed" && previous.attachmentId) { + deletePendingUpload(previous.environmentId, previous.attachmentId); + } + if (previous && "previous" in previous && previous.previous) { + setUploadState(input.image.id, previous.previous); + } else { + clearUploadState(input.image.id); + } + startAttachmentUpload(input); +} + +export async function awaitAttachmentUploads(imageIds: ReadonlyArray): Promise { + await Promise.all(imageIds.map((imageId) => jobsByImageId.get(imageId)?.settled)); +} + +export function getUploadedAttachments(input: { + readonly environmentId: EnvironmentId; + readonly images: ReadonlyArray; +}): ChatAttachment[] | null { + const attachments: ChatAttachment[] = []; + for (const image of input.images) { + const upload = readAttachmentUpload(image.id); + if (upload?.status !== "ready" || upload.environmentId !== input.environmentId) { + return null; + } + attachments.push({ + type: "image", + id: upload.attachmentId, + name: image.name, + mimeType: image.mimeType, + sizeBytes: image.sizeBytes, + }); + } + return attachments; +} + +export function releaseAttachmentUploads(images: ReadonlyArray): void { + for (const image of images) { + releaseAttachmentUpload(image.id); + } +} diff --git a/apps/web/src/lib/attachmentUploadState.test.ts b/apps/web/src/lib/attachmentUploadState.test.ts new file mode 100644 index 000000000000..3156d7200778 --- /dev/null +++ b/apps/web/src/lib/attachmentUploadState.test.ts @@ -0,0 +1,75 @@ +import { EnvironmentId } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { + attachmentUploadBlockReason, + formatAttachmentUploadProgress, +} from "./attachmentUploadState"; + +const environmentId = EnvironmentId.make("environment-1"); + +describe("attachmentUploadBlockReason", () => { + it("allows uploaded images from the active environment", () => { + expect( + attachmentUploadBlockReason({ + imageIds: ["image-1"], + environmentId, + uploadsByImageId: { + "image-1": { + status: "ready", + environmentId, + attachmentId: "pending-1", + }, + }, + }), + ).toBeNull(); + }); + + it("blocks images that are missing or uploading", () => { + expect( + attachmentUploadBlockReason({ + imageIds: ["image-1", "image-2"], + environmentId, + uploadsByImageId: { + "image-1": { status: "uploading", environmentId, progress: 0.5 }, + }, + }), + ).toBe("Images still uploading"); + }); + + it("asks the user to retry or remove failed uploads", () => { + expect( + attachmentUploadBlockReason({ + imageIds: ["image-1"], + environmentId, + uploadsByImageId: { + "image-1": { status: "failed", environmentId, reason: "Upload failed" }, + }, + }), + ).toBe("Retry or remove the failed image"); + }); + + it("does not accept an upload from another environment", () => { + expect( + attachmentUploadBlockReason({ + imageIds: ["image-1"], + environmentId, + uploadsByImageId: { + "image-1": { + status: "ready", + environmentId: EnvironmentId.make("environment-2"), + attachmentId: "pending-1", + }, + }, + }), + ).toBe("Image still uploading"); + }); +}); + +describe("formatAttachmentUploadProgress", () => { + it("formats bounded whole percentages", () => { + expect(formatAttachmentUploadProgress(0.429)).toBe("42%"); + expect(formatAttachmentUploadProgress(2)).toBe("100%"); + expect(formatAttachmentUploadProgress(Number.NaN)).toBe("0%"); + }); +}); diff --git a/apps/web/src/lib/attachmentUploadState.ts b/apps/web/src/lib/attachmentUploadState.ts new file mode 100644 index 000000000000..6ca2d2bc155c --- /dev/null +++ b/apps/web/src/lib/attachmentUploadState.ts @@ -0,0 +1,54 @@ +import type { EnvironmentId } from "@t3tools/contracts"; + +export type ReadyAttachmentUpload = { + readonly status: "ready"; + readonly environmentId: EnvironmentId; + readonly attachmentId: string; +}; + +export type AttachmentUploadState = + | { + readonly status: "uploading"; + readonly environmentId: EnvironmentId; + readonly progress: number; + readonly previous?: ReadyAttachmentUpload; + } + | ReadyAttachmentUpload + | { + readonly status: "failed"; + readonly environmentId: EnvironmentId; + readonly reason: string; + readonly attachmentId?: string; + readonly previous?: ReadyAttachmentUpload; + }; + +export function attachmentUploadBlockReason(input: { + readonly imageIds: ReadonlyArray; + readonly uploadsByImageId: Readonly>; + readonly environmentId: EnvironmentId; +}): string | null { + let pending = 0; + let failed = 0; + + for (const imageId of input.imageIds) { + const upload = input.uploadsByImageId[imageId]; + if (upload?.status === "failed" && upload.environmentId === input.environmentId) { + failed += 1; + } else if (upload?.status !== "ready" || upload.environmentId !== input.environmentId) { + pending += 1; + } + } + + if (failed > 0) { + return failed === 1 ? "Retry or remove the failed image" : "Retry or remove the failed images"; + } + if (pending > 0) { + return pending === 1 ? "Image still uploading" : "Images still uploading"; + } + return null; +} + +export function formatAttachmentUploadProgress(progress: number): string { + const bounded = Math.max(0, Math.min(1, Number.isFinite(progress) ? progress : 0)); + return `${Math.floor(bounded * 100)}%`; +} diff --git a/apps/web/src/lib/composerDraftUploads.ts b/apps/web/src/lib/composerDraftUploads.ts new file mode 100644 index 000000000000..a9b8a357725e --- /dev/null +++ b/apps/web/src/lib/composerDraftUploads.ts @@ -0,0 +1,23 @@ +import type { ScopedProjectRef, ScopedThreadRef } from "@t3tools/contracts"; + +import { type DraftId, useComposerDraftStore } from "../composerDraftStore"; +import { releaseAttachmentUploads } from "./attachmentUploadQueue"; + +export function releaseComposerDraftUploads(target: ScopedThreadRef | DraftId): void { + const draft = useComposerDraftStore.getState().getComposerDraft(target); + if (draft) { + releaseAttachmentUploads(draft.images); + } +} + +export function releaseProjectDraftUploads(projectRef: ScopedProjectRef): void { + const store = useComposerDraftStore.getState(); + for (const [draftKey, session] of Object.entries(store.draftThreadsByThreadKey)) { + if ( + session.environmentId === projectRef.environmentId && + session.projectId === projectRef.projectId + ) { + releaseAttachmentUploads(store.draftsByThreadKey[draftKey]?.images ?? []); + } + } +} diff --git a/apps/web/src/state/attachments.ts b/apps/web/src/state/attachments.ts new file mode 100644 index 000000000000..8b600d6c004a --- /dev/null +++ b/apps/web/src/state/attachments.ts @@ -0,0 +1,15 @@ +import { WS_METHODS } from "@t3tools/contracts"; +import { createEnvironmentRpcCommand } from "@t3tools/client-runtime/state/runtime"; + +import { connectionAtomRuntime } from "../connection/runtime"; + +export const attachmentEnvironment = { + createUploadUrl: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-command:attachments:create-upload-url", + tag: WS_METHODS.attachmentsCreateUploadUrl, + }), + remove: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-command:attachments:delete", + tag: WS_METHODS.attachmentsDelete, + }), +}; diff --git a/docs/user/composer.md b/docs/user/composer.md index 50a30d7f155a..b7ef57a56015 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -4,6 +4,9 @@ Messages can contain up to 120,000 characters. If a draft is longer, T3 Code kee composer and shows how many characters need to be removed. Shorten the draft or split it into multiple messages, then send again in the same thread. +On servers that support direct uploads, images upload as soon as you add them. The send button +becomes available after every upload finishes. Failed uploads can be retried or removed. + ## Commands and skills Type `/` to open the command menu. Type `$` to find and add a skill. Skill rows show their source, diff --git a/packages/contracts/src/assets.test.ts b/packages/contracts/src/assets.test.ts new file mode 100644 index 000000000000..ce4214d300da --- /dev/null +++ b/packages/contracts/src/assets.test.ts @@ -0,0 +1,30 @@ +import * as Schema from "effect/Schema"; +import { describe, expect, it } from "vite-plus/test"; + +import { AttachmentCreateUploadUrlInput } from "./assets.ts"; +import { PROVIDER_SEND_TURN_MAX_IMAGE_BYTES } from "./orchestration.ts"; + +const isUploadInput = Schema.is(AttachmentCreateUploadUrlInput); + +const uploadInput = { + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 3, +} as const; + +describe("AttachmentCreateUploadUrlInput", () => { + it("accepts supported image attachments", () => { + expect(isUploadInput(uploadInput)).toBe(true); + }); + + it("rejects image types that providers do not support", () => { + expect(isUploadInput({ ...uploadInput, mimeType: "image/svg+xml" })).toBe(false); + }); + + it("rejects empty and oversized uploads", () => { + expect(isUploadInput({ ...uploadInput, sizeBytes: 0 })).toBe(false); + expect( + isUploadInput({ ...uploadInput, sizeBytes: PROVIDER_SEND_TURN_MAX_IMAGE_BYTES + 1 }), + ).toBe(false); + }); +}); diff --git a/packages/contracts/src/assets.ts b/packages/contracts/src/assets.ts index e3922073455d..bfc2c9472aaa 100644 --- a/packages/contracts/src/assets.ts +++ b/packages/contracts/src/assets.ts @@ -1,7 +1,11 @@ import * as Schema from "effect/Schema"; -import { ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; -import { ProjectFaviconPath } from "./orchestration.ts"; +import { NonNegativeInt, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { + PROVIDER_SEND_TURN_MAX_IMAGE_BYTES, + PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES, + ProjectFaviconPath, +} from "./orchestration.ts"; const ASSET_PATH_MAX_LENGTH = 1024; @@ -36,6 +40,41 @@ export const AssetCreateUrlResult = Schema.Struct({ }); export type AssetCreateUrlResult = typeof AssetCreateUrlResult.Type; +export const ATTACHMENT_UPLOAD_URL_TTL_MS = 10 * 60_000; + +export const AttachmentCreateUploadUrlInput = Schema.Struct({ + name: TrimmedNonEmptyString.check(Schema.isMaxLength(255)), + mimeType: Schema.Literals(PROVIDER_SEND_TURN_SUPPORTED_IMAGE_MIME_TYPES), + sizeBytes: NonNegativeInt.check( + Schema.isGreaterThanOrEqualTo(1), + Schema.isLessThanOrEqualTo(PROVIDER_SEND_TURN_MAX_IMAGE_BYTES), + ), +}); +export type AttachmentCreateUploadUrlInput = typeof AttachmentCreateUploadUrlInput.Type; + +export const AttachmentCreateUploadUrlResult = Schema.Struct({ + attachmentId: TrimmedNonEmptyString.check(Schema.isMaxLength(256)), + relativeUrl: TrimmedNonEmptyString.check(Schema.isMaxLength(4096)), + expiresAt: Schema.Number, +}); +export type AttachmentCreateUploadUrlResult = typeof AttachmentCreateUploadUrlResult.Type; + +export const AttachmentDeleteInput = Schema.Struct({ + attachmentId: TrimmedNonEmptyString.check(Schema.isMaxLength(256)), +}); +export type AttachmentDeleteInput = typeof AttachmentDeleteInput.Type; + +export class AttachmentUploadSigningKeyError extends Schema.TaggedErrorClass()( + "AttachmentUploadSigningKeyError", + { + cause: Schema.Defect(), + }, +) { + override get message(): string { + return "Failed to load the attachment upload signing key."; + } +} + export class AssetWorkspaceContextNotFoundError extends Schema.TaggedErrorClass()( "AssetWorkspaceContextNotFoundError", { diff --git a/packages/contracts/src/environment.test.ts b/packages/contracts/src/environment.test.ts index 3a4324625a00..455cc58f47d1 100644 --- a/packages/contracts/src/environment.test.ts +++ b/packages/contracts/src/environment.test.ts @@ -26,4 +26,17 @@ describe("ExecutionEnvironmentDescriptor", () => { }).capabilities.pullRequests, ).toBe(true); }); + + it("treats a missing attachment upload capability as unsupported", () => { + expect(decodeDescriptor(descriptor).capabilities.attachmentUploads).toBeUndefined(); + }); + + it("preserves an advertised attachment upload capability", () => { + expect( + decodeDescriptor({ + ...descriptor, + capabilities: { ...descriptor.capabilities, attachmentUploads: true }, + }).capabilities.attachmentUploads, + ).toBe(true); + }); }); diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 1777bcebc2f8..1468fe9ef3d3 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -48,6 +48,8 @@ export type ServerSelfUpdateCapability = typeof ServerSelfUpdateCapability.Type; export const ExecutionEnvironmentCapabilities = Schema.Struct({ repositoryIdentity: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), connectionProbe: Schema.optionalKey(Schema.Boolean), + /** Missing on older servers, which still accept inline image attachments. */ + attachmentUploads: Schema.optionalKey(Schema.Boolean), /** Server exposes the pull-request list, detail, activity, diff, and mutation APIs. Absent on servers from before the pull-request workspace shipped, so clients must not probe them. */ pullRequests: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestration.test.ts b/packages/contracts/src/orchestration.test.ts index 3e1b9be0bba5..27bdecdda7a8 100644 --- a/packages/contracts/src/orchestration.test.ts +++ b/packages/contracts/src/orchestration.test.ts @@ -5,6 +5,7 @@ import * as Schema from "effect/Schema"; import { DEFAULT_PROVIDER_INTERACTION_MODE, DEFAULT_RUNTIME_MODE, + ClientOrchestrationCommand, ModelSelection, OrchestrationCommand, OrchestrationDispatchCommandError, @@ -35,6 +36,7 @@ const decodeProjectCreateCommand = Schema.decodeUnknownEffect(ProjectCreateComma const decodeProjectCreatedPayload = Schema.decodeUnknownEffect(ProjectCreatedPayload); const decodeProjectMetaUpdatedPayload = Schema.decodeUnknownEffect(ProjectMetaUpdatedPayload); const decodeThreadTurnStartCommand = Schema.decodeUnknownEffect(ThreadTurnStartCommand); +const decodeClientOrchestrationCommand = Schema.decodeUnknownEffect(ClientOrchestrationCommand); const decodeThreadTurnStartRequestedPayload = Schema.decodeUnknownEffect( ThreadTurnStartRequestedPayload, ); @@ -241,6 +243,47 @@ it.effect("decodes thread.turn.start defaults for provider and runtime mode", () }), ); +it.effect("accepts both inline and uploaded image attachments from clients", () => + Effect.gen(function* () { + const command = yield* decodeClientOrchestrationCommand({ + type: "thread.turn.start", + commandId: "cmd-turn-attachments", + threadId: "thread-1", + message: { + messageId: "msg-attachments", + role: "user", + text: "hello", + attachments: [ + { + type: "image", + name: "legacy.png", + mimeType: "image/png", + sizeBytes: 3, + dataUrl: "data:image/png;base64,YWJj", + }, + { + type: "image", + id: "pending-00000000-0000-4000-8000-000000000001", + name: "uploaded.png", + mimeType: "image/png", + sizeBytes: 3, + }, + ], + }, + runtimeMode: "full-access", + interactionMode: "default", + createdAt: "2026-01-01T00:00:00.000Z", + }); + + if (command.type !== "thread.turn.start") { + assert.fail(`Expected thread.turn.start, received ${command.type}.`); + } + assert.strictEqual(command.message.attachments.length, 2); + assert.strictEqual("dataUrl" in command.message.attachments[0]!, true); + assert.strictEqual("id" in command.message.attachments[1]!, true); + }), +); + it.effect("preserves explicit provider and runtime mode in thread.turn.start", () => Effect.gen(function* () { const parsed = yield* decodeThreadTurnStartCommand({ diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 1c27e6d3c6b4..991730661d2d 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -852,7 +852,7 @@ const ClientThreadTurnStartCommand = Schema.Struct({ messageId: MessageId, role: Schema.Literal("user"), text: Schema.String, - attachments: Schema.Array(UploadChatAttachment), + attachments: Schema.Array(Schema.Union([UploadChatAttachment, ChatAttachment])), }), modelSelection: Schema.optional(ModelSelection), titleSeed: Schema.optional(TrimmedNonEmptyString), diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 45bf581de084..14363cfedff9 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -18,7 +18,15 @@ import { FilesystemBrowseResult, FilesystemBrowseError, } from "./filesystem.ts"; -import { AssetAccessError, AssetCreateUrlInput, AssetCreateUrlResult } from "./assets.ts"; +import { + AssetAccessError, + AssetCreateUrlInput, + AssetCreateUrlResult, + AttachmentCreateUploadUrlInput, + AttachmentCreateUploadUrlResult, + AttachmentDeleteInput, + AttachmentUploadSigningKeyError, +} from "./assets.ts"; import { GitActionProgressEvent, VcsSwitchRefInput, @@ -215,6 +223,8 @@ export const WS_METHODS = { // Filesystem methods filesystemBrowse: "filesystem.browse", assetsCreateUrl: "assets.createUrl", + attachmentsCreateUploadUrl: "attachments.createUploadUrl", + attachmentsDelete: "attachments.delete", // Provider methods providerUploadFeedback: "provider.uploadFeedback", @@ -673,6 +683,17 @@ export const WsAssetsCreateUrlRpc = Rpc.make(WS_METHODS.assetsCreateUrl, { error: Schema.Union([AssetAccessError, EnvironmentAuthorizationError]), }); +export const WsAttachmentsCreateUploadUrlRpc = Rpc.make(WS_METHODS.attachmentsCreateUploadUrl, { + payload: AttachmentCreateUploadUrlInput, + success: AttachmentCreateUploadUrlResult, + error: Schema.Union([AttachmentUploadSigningKeyError, EnvironmentAuthorizationError]), +}); + +export const WsAttachmentsDeleteRpc = Rpc.make(WS_METHODS.attachmentsDelete, { + payload: AttachmentDeleteInput, + error: EnvironmentAuthorizationError, +}); + export const WsProviderUploadFeedbackRpc = Rpc.make(WS_METHODS.providerUploadFeedback, { payload: ProviderUploadFeedbackInput, success: ProviderUploadFeedbackResult, @@ -1048,6 +1069,8 @@ export const WsRpcGroup = RpcGroup.make( WsShellOpenInEditorRpc, WsFilesystemBrowseRpc, WsAssetsCreateUrlRpc, + WsAttachmentsCreateUploadUrlRpc, + WsAttachmentsDeleteRpc, WsProviderUploadFeedbackRpc, WsSubscribeVcsStatusRpc, WsVcsPullRpc,