fix(cloud-agent): link attachment rows before dispatch and require the client message id

apps/extension/src/shared/extension-agent-session-manager.test.ts

@@ -1254,6 +1254,7 @@ describe('createExtensionAgentSessionManager', () => {
       const trpc = setup();
       await capturedConfig!.api.send({
         attachments: { files: [], path: '/f', type: 'file' } as never,
+        messageId: 'msg_123456789abc123456789ABC04',
         payload: { mode: 'code', model: 'gpt', prompt: 'hi', type: 'prompt' },
         sessionId: CLOUD_AGENT_ID as never,
       });
@@ -1325,6 +1326,7 @@ describe('createExtensionAgentSessionManager', () => {
     it('send calls org cloudAgentNext.sendMessage including organizationId', async () => {
       const trpc = setup();
       await capturedConfig!.api.send({
+        messageId: 'msg_123456789abc123456789ABC03',
         payload: { mode: 'code', model: 'gpt', prompt: 'hi', type: 'prompt' },
         sessionId: CLOUD_AGENT_ID as never,
       });

apps/extension/src/shared/extension-trpc-client.test.ts

@@ -10,14 +10,17 @@ const FAKE_BASE_URL = 'https://api.example.com';
 const FAKE_SESSION_ID = 'test-session-id';
 const PROMPT_PAYLOAD = {
   cloudAgentSessionId: FAKE_SESSION_ID,
+  messageId: 'msg_123456789abc123456789ABCDE',
   payload: { mode: 'plan' as const, model: 'test-model', prompt: 'hello', type: 'prompt' as const },
 };
 const PROMPT_FIRST = {
   cloudAgentSessionId: FAKE_SESSION_ID,
+  messageId: 'msg_123456789abc123456789ABC01',
   payload: { mode: 'plan' as const, model: 'test-model', prompt: 'first', type: 'prompt' as const },
 };
 const PROMPT_SECOND = {
   cloudAgentSessionId: FAKE_SESSION_ID,
+  messageId: 'msg_123456789abc123456789ABC02',
   payload: {
     mode: 'plan' as const,
     model: 'test-model',

apps/web/src/hooks/useCloudAgentAttachmentUpload.ts

@@ -492,11 +492,9 @@ export function useCloudAgentAttachmentUpload(
 
   /**
    * Build the attachment payload for the caller to submit. The pending-ledger
-   * rows are NOT linked here: the server links authoritatively only after the
-   * send or prepare succeeds (both the personal and organization routers call
-   * linkPendingUploads post-send). Linking client-side before the send would
-   * leave a failed or abandoned send with 'linked' rows that the reaper never
-   * deletes.
+   * rows are linked by the server as part of the send or prepare dispatch, so
+   * this side must not link them earlier: a row this side linked before a send
+   * that never happened would keep its object out of the reaper's reach.
    */
   const finalizeAttachments = useCallback(async (): Promise<CloudAgentAttachments | undefined> => {
     return getAttachmentsData();

apps/web/src/lib/cloud-agent-next/release-refused-send.test.ts

@@ -0,0 +1,172 @@
+import { beforeEach, describe, expect, it, jest } from '@jest/globals';
+import { TRPCClientError } from '@trpc/client';
+
+import {
+  CloudAgentBillingError,
+  InsufficientCreditsError,
+} from '@/lib/cloud-agent-next/cloud-agent-client';
+import { releaseRefusedSend, workerRefusalStatus } from './release-refused-send';
+
+const mockRelease = jest.fn(
+  async (_kiloUserId: string, _messageUuid: string, _objectKeys: string[]) => 1
+);
+
+const attachments = { path: 'f7c1f9a0-0000-4000-8000-000000000001', files: ['a.bin'] };
+const linkedObjectKeys = ['user-1/cloud-agent/f7c1f9a0-0000-4000-8000-000000000001/a.bin'];
+
+const refusal = (httpStatus: number) =>
+  new TRPCClientError(`worker answered ${String(httpStatus)}`, {
+    result: { error: { code: -32600, message: 'refused', data: { httpStatus } } },
+  });
+
+const dependencies = { release: mockRelease };
+
+describe('releaseRefusedSend', () => {
+  beforeEach(() => {
+    jest.restoreAllMocks();
+    jest.clearAllMocks();
+    jest.spyOn(console, 'error').mockImplementation(() => {});
+    jest.spyOn(console, 'info').mockImplementation(() => {});
+  });
+
+  it('returns the rows this dispatch linked when the worker refused the turn', async () => {
+    await releaseRefusedSend(
+      { attachments, error: refusal(400), kiloUserId: 'user-1', linkedObjectKeys },
+      dependencies
+    );
+
+    expect(mockRelease).toHaveBeenCalledWith(
+      'user-1',
+      'f7c1f9a0-0000-4000-8000-000000000001',
+      linkedObjectKeys
+    );
+  });
+
+  it('keeps the rows when the worker answered with a server-side status', async () => {
+    await releaseRefusedSend(
+      { attachments, error: refusal(503), kiloUserId: 'user-1', linkedObjectKeys },
+      dependencies
+    );
+
+    expect(mockRelease).not.toHaveBeenCalled();
+  });
+
+  it('returns the rows when the client rewrote a credits refusal', async () => {
+    await releaseRefusedSend(
+      {
+        attachments,
+        error: new InsufficientCreditsError(),
+        kiloUserId: 'user-1',
+        linkedObjectKeys,
+      },
+      dependencies
+    );
+
+    expect(mockRelease).toHaveBeenCalledTimes(1);
+  });
+
+  it('returns the rows for a billing refusal but keeps them for a billing outage', async () => {
+    await releaseRefusedSend(
+      {
+        attachments,
+        error: new CloudAgentBillingError({ kind: 'billing' } as never, 402),
+        kiloUserId: 'user-1',
+        linkedObjectKeys,
+      },
+      dependencies
+    );
+    expect(mockRelease).toHaveBeenCalledTimes(1);
+
+    await releaseRefusedSend(
+      {
+        attachments,
+        error: new CloudAgentBillingError({ kind: 'billing' } as never, 503),
+        kiloUserId: 'user-1',
+        linkedObjectKeys,
+      },
+      dependencies
+    );
+    expect(mockRelease).toHaveBeenCalledTimes(1);
+  });
+
+  it('returns the rows for a status the client kept only on the cause', async () => {
+    const normalized = new Error('Cloud agent returned a non-JSON error response', {
+      cause: refusal(413),
+    });
+
+    await releaseRefusedSend(
+      { attachments, error: normalized, kiloUserId: 'user-1', linkedObjectKeys },
+      dependencies
+    );
+
+    expect(mockRelease).toHaveBeenCalledTimes(1);
+  });
+
+  it('keeps the rows when the send failed without a worker answer', async () => {
+    await releaseRefusedSend(
+      {
+        attachments,
+        error: new TRPCClientError('fetch failed'),
+        kiloUserId: 'user-1',
+        linkedObjectKeys,
+      },
+      dependencies
+    );
+
+    expect(mockRelease).not.toHaveBeenCalled();
+  });
+
+  it('keeps the rows when this dispatch linked none of them', async () => {
+    await releaseRefusedSend(
+      { attachments, error: refusal(400), kiloUserId: 'user-1', linkedObjectKeys: [] },
+      dependencies
+    );
+
+    expect(mockRelease).not.toHaveBeenCalled();
+  });
+
+  it('does nothing for a send without attachments', async () => {
+    await releaseRefusedSend(
+      { attachments: undefined, error: refusal(400), kiloUserId: 'user-1', linkedObjectKeys },
+      dependencies
+    );
+
+    expect(mockRelease).not.toHaveBeenCalled();
+  });
+
+  it('stays silent when the ledger cannot return the rows', async () => {
+    const release = jest.fn(async () => {
+      throw new Error('database unavailable');
+    });
+
+    await expect(
+      releaseRefusedSend(
+        { attachments, error: refusal(400), kiloUserId: 'user-1', linkedObjectKeys },
+        { release }
+      )
+    ).resolves.toBeUndefined();
+
+    expect(release).toHaveBeenCalledTimes(1);
+    expect(jest.mocked(console.error)).toHaveBeenCalledWith(
+      '[cloud-agent] Could not return the attachments of a refused send',
+      expect.any(Error)
+    );
+  });
+});
+
+describe('workerRefusalStatus', () => {
+  it('reads the status the worker answered with, and nothing else', () => {
+    expect(workerRefusalStatus(refusal(400))).toBe(400);
+    expect(workerRefusalStatus(new InsufficientCreditsError())).toBe(402);
+    expect(workerRefusalStatus(new CloudAgentBillingError({ kind: 'billing' } as never, 409))).toBe(
+      409
+    );
+    expect(workerRefusalStatus(new Error('non-JSON', { cause: refusal(429) }))).toBe(429);
+    expect(
+      workerRefusalStatus(new Error('non-JSON', { cause: new TRPCClientError('fetch failed') }))
+    ).toBeUndefined();
+    expect(workerRefusalStatus(new TRPCClientError('fetch failed'))).toBeUndefined();
+    expect(workerRefusalStatus(new Error('boom'))).toBeUndefined();
+    expect(workerRefusalStatus(undefined)).toBeUndefined();
+  });
+});

apps/web/src/lib/cloud-agent-next/release-refused-send.ts

@@ -0,0 +1,82 @@
+import { TRPCClientError } from '@trpc/client';
+import {
+  releaseUnadmittedUploads,
+  type DispatchAttachments,
+} from '@/lib/r2/cloud-agent-pending-uploads';
+
+/**
+ * The http status a send failure carries, when the worker answered it.
+ *
+ * `client.sendMessage` rewrites several worker answers before this sees them: a
+ * credits refusal becomes `InsufficientCreditsError`, a billing refusal becomes
+ * `CloudAgentBillingError`, and a non-JSON answer becomes a plain error whose
+ * cause is the original. Those carry the status themselves or keep it on the
+ * cause, so the status is read along that chain as well as from the tRPC shape.
+ */
+export function workerRefusalStatus(error: unknown): number | undefined {
+  // The client wraps some answers and keeps the original as the cause, so the
+  // status may sit one or two links down the chain.
+  let current: unknown = error;
+  for (let depth = 0; depth < 4; depth += 1) {
+    if (current === null || typeof current !== 'object') return undefined;
+
+    const direct = (current as { httpStatus?: unknown }).httpStatus;
+    if (typeof direct === 'number') return direct;
+
+    if (current instanceof TRPCClientError) {
+      const data = current.data as { httpStatus?: unknown } | undefined;
+      const shapeData = current.shape?.data as { httpStatus?: unknown } | undefined;
+      const status = data?.httpStatus ?? shapeData?.httpStatus;
+      if (typeof status === 'number') return status;
+    }
+
+    current = (current as { cause?: unknown }).cause;
+  }
+  return undefined;
+}
+
+export type RefusedSend = {
+  attachments: DispatchAttachments | undefined;
+  error: unknown;
+  kiloUserId: string;
+  /** The object keys this dispatch's link moved to 'linked'. */
+  linkedObjectKeys: string[];
+};
+
+type ReleaseDependencies = {
+  release?: (
+    kiloUserId: string,
+    messageUuid: string,
+    linkedObjectKeys: string[]
+  ) => Promise<number>;
+};
+
+/**
+ * Give back the attachment rows of a send the worker refused.
+ *
+ * A refusal is a 4xx answer to this send: the worker refuses before it admits,
+ * and an admitted send answers with a queued result, so a refused turn never
+ * existed. A transport failure, a timeout or a server-side status proves
+ * nothing, so those keep the rows linked; the same holds for a send that linked
+ * no rows here, which is the retry of an attachment an earlier dispatch holds.
+ *
+ * The send's own error is what the caller reports; this helper never throws.
+ */
+export async function releaseRefusedSend(
+  input: RefusedSend,
+  { release = releaseUnadmittedUploads }: ReleaseDependencies = {}
+): Promise<void> {
+  const { attachments, error, kiloUserId, linkedObjectKeys } = input;
+  const status = workerRefusalStatus(error);
+  if (status === undefined || status < 400 || status >= 500) return;
+  if (!attachments || linkedObjectKeys.length === 0) return;
+
+  try {
+    const released = await release(kiloUserId, attachments.path, linkedObjectKeys);
+    if (released > 0) {
+      console.info('[cloud-agent] Returned the attachments of a refused send', { released });
+    }
+  } catch (error) {
+    console.error('[cloud-agent] Could not return the attachments of a refused send', error);
+  }
+}