diff --git a/apps/server/src/assets/AttachmentUpload.test.ts b/apps/server/src/assets/AttachmentUpload.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..cb08d5e4b2f18a45c53065fff17cdd2fea7739f7 --- /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 0000000000000000000000000000000000000000..6142b69d7342ee413d6c1f0c7fe42f17ed49a59d --- /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 e21d9cf62cf50f3c435ef428315267519a2d8380..5e782e55407f1d820bc42a708ed920eadc5a45c9 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 3d5b531db217cd49451f59e7cba0333892573bdd..d0334bce09f36a040a53c90e6f05aae790efac85 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 70227cdd4ebf5f3cdac8e7606a1b59b295b97ce0..28ceac4cec9922dff69fa4b40358193c080b636d 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 e678264dde5f15cfd519466ff781f634e7345608..bdff19572fddd863aa8c9538102bc7714656aa0a 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 ee30d987591d1bf74f9309bf42af892fecc384ca..b9a50ca8335e4b26d807a730101e58aa556f7ccb 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 45dc0ee9cfd523aa2e19c7ee28dc32c56ad42e51..e55639ce659c295f25e059eb21bab8e83d13ad09 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 be133399e0f84622d5fa39744ba33ff7b8831d6d..c3104e7bc4209a07e58526880e226ce1598852bc 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 0000000000000000000000000000000000000000..27a35977ffca39b2e8cfc0ddb5d44bb4e512e92b --- /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 24c65900b296ca867107a3662dbdd7a69d0b5e41..bd6a8f242b8719bf504b51ed067f841fa2a0490d 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 04d54ea8effb0f964759c71191bd0fc1c2336696..f7147106c7a9bb8b2a0eac1aba6352375589c03a 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 02a367c08792ff1ef4a920e941ab150d79d6ae3d..5e4f19172eff07dd02a98f502023f02aaebe8663 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 3e41b4390f824ab5f6177a66f4906bcc9f32572d..0a31bf376dae6b657adcf424e59ab4fc397cf3dd 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 11c659e28a704e823d3ddf7ab88a083b8dc6f12e..55b0be07c667188e43c82474ca9cc7dd89e3c0f1 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 46ed051154a6bc2f225c2ed59b5df8ae74cb4511..cb1cf698535a1d32493a397ba378aeb023d9bce9 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 b4f3b5996818e1d0aded75d44a72f52ff6439aae..4555544046fab94e926b57f141f0df93b3af861f 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 971ead810f07b4baf7b8f8ea9932047b86d0d81b..7a80559d390dd84137dfcca1ee0de2a9c99336c2 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 e854498ada593bf439c2ec32581c9d8849ca88b0..1ec58e0de702937e0f8c028c16dba3a472af9ec1 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(); @@ -3106,9 +3149,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); @@ -3158,66 +3209,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 4ee61b556a03ec1eac8d8f2da68a32b556d5e245..08c38faafd5e402c208244c459b0ef4e529930ae 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 0287dde73aaf7c9869fc3188bc29acb15800bcc7..8eb7e9897b8e887a6b80d52963fdaa96bccfc667 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 6768d2dc61ef5ab1947c4a1a7a64f8af48d7db43..6047b8fc48dc7623b9cb3b597178ea7057263ad0 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 569b4be96e62fd68e8b1698a0903576e1ecb7e79..f40920779b0f80af0164aefc2c8a3113b107e7ee 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 0000000000000000000000000000000000000000..2b2b94431c80e7d68bed791ce6bec346c2d15a42 --- /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 0000000000000000000000000000000000000000..37eb924ca25688aa18ac598fb06f41be528685a5 --- /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 0000000000000000000000000000000000000000..3156d7200778fbc9f013b1b8df37ecce4f3d2e14 --- /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 0000000000000000000000000000000000000000..6ca2d2bc155cb614d621dd85a14f0c50ebc1cadd --- /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 0000000000000000000000000000000000000000..a9b8a357725e68a486e1ed1bb5559a7a4c46d802 --- /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 0000000000000000000000000000000000000000..8b600d6c004a446a810642bf384283c1637713fc --- /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 50a30d7f155a5dca703a8e429c651f37b23c300e..b7ef57a56015488ad57c7ac77d4b1a7fea2f91bc 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 0000000000000000000000000000000000000000..ce4214d300daacfb6695861f6b1c459abf8f31f4 --- /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 e3922073455d474503b1340560d1e9866d0d0851..bfc2c9472aaaddff25f967613bd8397b2ba91c91 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 3a4324625a009522829462e9a3ecf7e40b3a804e..455cc58f47d15946a6bed5081e9c6f32887cdc41 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 1777bcebc2f825fc9954a609eaa5b5232f54c248..1468fe9ef3d367321d8ee5583daee296e513afd9 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 3e1b9be0bba57411169a2f56dc4712fc82fb15c8..27bdecdda7a8d5ec08b596509e4a1a803b9d7e71 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 2535616d4944b39ee53b9dd6df4421545f6841e9..af4fefaccf592f063344bcb1ace2d7ce944104a1 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -863,7 +863,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 45bf581de084a665b3d90876b9b6737ccb97624e..14363cfedff9565827cdc4fecbafafb6c39ac350 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,