feat(wire,client-core): chunked attachment upload and read protocol

apps/daemon/src/__tests__/attachment-store.test.ts

@@ -11,6 +11,7 @@ import {
   ConversationOperationSchema,
   ConversationTurnSchema,
   PromptRecordSchema,
+  SessionIdSchema,
   SessionRecordSchema,
   SessionResourceSchema,
   UploadIdSchema,
@@ -209,4 +210,74 @@ describe('SQLite attachment store', () => {
     expect(await store.getBlob(leasedOnly.blobId)).toBeUndefined();
     expect(await store.getBlob(viaResource.blobId)).toEqual(viaResource);
   });
+
+  it('reports reachability from a session prompt or resource, not another session', async () => {
+    const { database, path, store } = await fixture();
+    await store.commitAttachment({ blob: blob('prompt'), attachment: attachment('att-prompt') });
+    await store.commitAttachment({
+      blob: blob('resource'),
+      attachment: attachment('att-resource'),
+    });
+    await createConversationStore(database.client).persistTurnIntent({
+      turn: ConversationTurnSchema.parse({
+        turnId: 't-1',
+        sessionId: 's-1',
+        parentTurnId: null,
+        siblingOrdinal: 1,
+        input: { type: 'prompt', promptId: 'p-1' },
+        runId: 'run-1',
+        state: 'preparing',
+        createdAt: 1,
+      }),
+      prompt: PromptRecordSchema.parse({
+        promptId: 'p-1',
+        blocks: [{ type: 'attachment_ref', attachmentId: 'att-prompt' }],
+        contextAttachmentIds: [],
+        createdAt: 1,
+      }),
+      operation: ConversationOperationSchema.parse({
+        operationId: 'op-reach',
+        sessionId: 's-1',
+        kind: 'turn.submit',
+        state: 'open',
+        createdAt: 1,
+      }),
+    });
+    await createResourceStore(path).save(
+      SessionResourceSchema.parse({
+        resourceId: 'resource-1',
+        sessionId: 's-1',
+        direction: 'source',
+        name: 'brief.txt',
+        kind: 'file',
+        status: 'ready',
+        locator: { type: 'managed-file', path: '/state/blobs/x' },
+        attachmentId: 'att-resource',
+        createdAt: 1,
+        updatedAt: 1,
+      }),
+    );
+
+    expect(
+      await store.isReachable(SessionIdSchema.parse('s-1'), AttachmentIdSchema.parse('att-prompt')),
+    ).toBe(true);
+    expect(
+      await store.isReachable(
+        SessionIdSchema.parse('s-1'),
+        AttachmentIdSchema.parse('att-resource'),
+      ),
+    ).toBe(true);
+    expect(
+      await store.isReachable(
+        SessionIdSchema.parse('s-other'),
+        AttachmentIdSchema.parse('att-prompt'),
+      ),
+    ).toBe(false);
+    expect(
+      await store.isReachable(
+        SessionIdSchema.parse('s-1'),
+        AttachmentIdSchema.parse('att-missing'),
+      ),
+    ).toBe(false);
+  });
 });

apps/daemon/src/attachment-store.ts

@@ -4,7 +4,14 @@ import type {
   AttachmentSweepWindow,
   StoredAttachment,
 } from '@linkcode/engine';
-import type { AttachmentId, BlobId, BlobRecord, UploadId, UploadLease } from '@linkcode/schema';
+import type {
+  AttachmentId,
+  BlobId,
+  BlobRecord,
+  SessionId,
+  UploadId,
+  UploadLease,
+} from '@linkcode/schema';
 import {
   AttachmentRecordSchema,
   BlobIdSchema,
@@ -18,6 +25,7 @@ import {
   attachmentBlobs,
   attachments,
   blobs,
+  conversationTurns,
   promptAttachmentRefs,
   sessionResources,
   uploadLeases,
@@ -90,6 +98,32 @@ export function createAttachmentStore(db: DaemonDatabaseClient): AttachmentStore
       return Promise.resolve();
     },
 
+    isReachable(sessionId: SessionId, attachmentId: AttachmentId): Promise<boolean> {
+      const fromPrompt = db
+        .select({ id: promptAttachmentRefs.attachmentId })
+        .from(promptAttachmentRefs)
+        .innerJoin(conversationTurns, eq(conversationTurns.promptId, promptAttachmentRefs.promptId))
+        .where(
+          and(
+            eq(conversationTurns.sessionId, sessionId),
+            eq(promptAttachmentRefs.attachmentId, attachmentId),
+          ),
+        )
+        .get();
+      if (fromPrompt) return Promise.resolve(true);
+      const fromResource = db
+        .select({ id: sessionResources.attachmentId })
+        .from(sessionResources)
+        .where(
+          and(
+            eq(sessionResources.sessionId, sessionId),
+            eq(sessionResources.attachmentId, attachmentId),
+          ),
+        )
+        .get();
+      return Promise.resolve(fromResource !== undefined);
+    },
+
     commitAttachment({ attachment, blob, uploadId }: AttachmentCommit): Promise<void> {
       db.transaction((tx) => {
         tx.insert(blobs).values(blob).onConflictDoNothing().run();

packages/client/core/src/client.ts

@@ -10,6 +10,7 @@ import type {
   AgentKind,
   AgentRuntimes,
   AgentStartCatalog,
+  AttachmentId,
   ContentBlock,
   CustomMcpServerPatchOp,
   CustomMcpServerPublic,
@@ -67,6 +68,7 @@ import type {
   StartOptions,
   TerminalMetadata,
   TerminalReplayEvent,
+  UploadId,
   WireMessage,
   WorkspaceFile,
   WorkspaceId,
@@ -75,6 +77,7 @@ import type {
   WorkspaceScript,
 } from '@linkcode/schema';
 import {
+  ATTACHMENT_STORE_WIRE_VERSION,
   CONVERSATION_GRAPH_WIRE_VERSION,
   MIN_COMPATIBLE_WIRE_VERSION,
   WIRE_PROTOCOL_VERSION,
@@ -85,6 +88,13 @@ import { extractErrorMessage, isErrorLikeObject } from 'foxts/extract-error-mess
 import { noop } from 'foxts/noop';
 import type { AgentLoginHandlers } from './client/agent-login-channel';
 import { AgentLoginChannel } from './client/agent-login-channel';
+import type {
+  AttachmentBeginInput,
+  AttachmentPutInput,
+  AttachmentReadBytes,
+} from './client/attachment-channel';
+import { AttachmentChannel } from './client/attachment-channel';
+import type { Sha256Hex } from './client/blob-cache';
 import type { BrowserCommandExecutor } from './client/browser-host-channel';
 import { BrowserHostChannel } from './client/browser-host-channel';
 import type {
@@ -111,6 +121,12 @@ import { PendingRegistry, resolveRandomUUID } from './client/pending-registry';
 import { TerminalChannel } from './client/terminal-channel';
 
 export type { AgentLoginHandlers, AgentLoginSettled } from './client/agent-login-channel';
+export type {
+  AttachmentBeginInput,
+  AttachmentPutInput,
+  AttachmentReadBytes,
+} from './client/attachment-channel';
+export type { Sha256Hex } from './client/blob-cache';
 export type { BrowserCommandExecutor } from './client/browser-host-channel';
 export type {
   ConversationReadClientOptions,
@@ -146,6 +162,8 @@ type TerminalReplayTruncatedCb = (truncated: boolean) => void;
 
 export interface LinkCodeClientOptions {
   randomUUID?: RandomUUID;
+  /** Attachment upload hashes its bytes; hosts without `crypto.subtle` must supply the digest. */
+  sha256Hex?: Sha256Hex;
 }
 
 export interface TerminalAttachResult {
@@ -240,6 +258,7 @@ export function isRequestFailureReportedInConversation(error: unknown): boolean
 export class LinkCodeClient {
   private readonly pending: PendingRegistry;
   private readonly control: ControlChannel;
+  private readonly attachments: AttachmentChannel;
   private readonly events = new EventBuffer();
   private readonly graphChanges = new ConversationGraphChanges();
   private readonly terminals: TerminalChannel;
@@ -277,6 +296,7 @@ export class LinkCodeClient {
     const randomUUID = resolveRandomUUID(options.randomUUID);
     this.pending = new PendingRegistry(randomUUID);
     this.control = new ControlChannel(transport, this.pending);
+    this.attachments = new AttachmentChannel(transport, this.pending, options.sha256Hex);
     this.terminals = new TerminalChannel(transport, this.pending, randomUUID);
     this.browserHost = new BrowserHostChannel(transport, this.pending, randomUUID);
     this.agentLogin = new AgentLoginChannel(transport, this.pending);
@@ -326,6 +346,11 @@ export class LinkCodeClient {
     return this.peerWire !== null && this.peerWire.version >= CONVERSATION_GRAPH_WIRE_VERSION;
   }
 
+  /** Whether the host serves chunked `attachment.upload.*` / `attachment.read`. */
+  get supportsAttachmentStore(): boolean {
+    return this.peerWire !== null && this.peerWire.version >= ATTACHMENT_STORE_WIRE_VERSION;
+  }
+
   private async handshake(): Promise<void> {
     let settled = false;
     let cancelTimer: () => void = noop;
@@ -607,6 +632,36 @@ export class LinkCodeClient {
       case 'resource.hosted':
         this.pending.resolve('resourceHost', p.replyTo, p.hosted);
         break;
+      case 'attachment.upload.begun':
+        this.pending.resolve('attachmentBegin', p.replyTo, {
+          uploadId: p.uploadId,
+          chunkBytes: p.chunkBytes,
+          state: p.state,
+        });
+        break;
+      case 'attachment.upload.chunk.acked':
+        this.pending.resolve('attachmentChunk', p.replyTo, {
+          uploadId: p.uploadId,
+          receivedBytes: p.receivedBytes,
+        });
+        break;
+      case 'attachment.upload.committed':
+        this.pending.resolve('attachmentCommit', p.replyTo, {
+          attachmentId: p.attachmentId,
+          blobId: p.blobId,
+        });
+        break;
+      case 'attachment.read.result':
+        this.pending.resolve('attachmentRead', p.replyTo, {
+          sessionId: p.sessionId,
+          attachmentId: p.attachmentId,
+          blobId: p.blobId,
+          offset: p.offset,
+          data: p.data,
+          sizeBytes: p.sizeBytes,
+          eof: p.eof,
+        });
+        break;
       case 'resource.changed':
         for (const cb of this.resourceEventSubs) cb({ type: 'changed', resource: p.resource });
         break;
@@ -1323,6 +1378,39 @@ export class LinkCodeClient {
   hostResource(resourceId: SessionResourceId): Promise<HostedSessionResource> {
     return this.control.hostResource(resourceId);
   }
+
+  beginAttachmentUpload(input: AttachmentBeginInput) {
+    return this.attachments.beginUpload(input);
+  }
+
+  sendAttachmentChunk(uploadId: UploadId, offset: number, data: string) {
+    return this.attachments.sendChunk(uploadId, offset, data);
+  }
+
+  commitAttachmentUpload(uploadId: UploadId) {
+    return this.attachments.commit(uploadId);
+  }
+
+  abortAttachmentUpload(uploadId: UploadId) {
+    return this.attachments.abort(uploadId);
+  }
+
+  readAttachment(sessionId: SessionId, attachmentId: AttachmentId, offset: number, length: number) {
+    return this.attachments.read(sessionId, attachmentId, offset, length);
+  }
+
+  /** Hash + windowed chunked upload. Identical bytes commit with no transfer. */
+  putAttachment(input: AttachmentPutInput) {
+    return this.attachments.put(input);
+  }
+
+  /** Read every byte of an attachment, cached by `blobId`. */
+  getAttachmentBytes(
+    sessionId: SessionId,
+    attachmentId: AttachmentId,
+  ): Promise<AttachmentReadBytes> {
+    return this.attachments.get(sessionId, attachmentId);
+  }
   subscribeResources(cb: ResourceEventCb): Unsubscribe {
     this.resourceEventSubs.add(cb);
     return () => this.resourceEventSubs.delete(cb);