fix(mothership): address review round on content delivery

Committed writes answer saved-not-shown instead of a retryable error, keep
their panel refresh, and account per operation. Chat streams pin inline
bytes off, taint needs a registry, compile admits before its cache, owned
files bind to their context, and count/truncation oracles are closed for
Chat in knowledge, files search and row search. Table selection cells are
projected before escaping and coerced import cells keep their taint.

Author: waleedlatif1

.claude/rules/sim-content-delivery.md

@@ -1,12 +1,17 @@
 ---
 description: How stored content (workspace files, run files, knowledge, table rows) reaches Chat's model, and the per-path policy. Read before changing a use case, route or Chat surface that returns such content.
 paths:
-  - "apps/sim/lib/core/application/content-delivery.ts"
+  - "apps/sim/lib/core/application/**"
   - "apps/sim/lib/mothership/content-delivery/**"
-  - "apps/sim/lib/**/application/admit-*.ts"
+  - "apps/sim/lib/workspace-files/application/**"
+  - "apps/sim/lib/knowledge/application/**"
+  - "apps/sim/lib/table/application/**"
+  - "apps/sim/lib/workflows/application/**"
+  - "apps/sim/lib/mothership/chat/application/**"
   - "apps/sim/lib/**/application/operations.ts"
   - "apps/sim/lib/mothership/agent-cli/**"
-  - "apps/sim/lib/mothership/tools/sandbox-resource-transport.ts"
+  - "apps/sim/lib/mothership/tools/**"
+  - "apps/sim/app/api/v2/**"
 ---
 
 # Sim Content Delivery to Chat

.cursor/rules/sim-content-delivery.mdc

@@ -1,6 +1,6 @@
 ---
 description: "How stored content (workspace files, run files, knowledge, table rows) reaches Chat's model, and the per-path policy. Read before changing a use case, route or Chat surface that returns such content."
-globs: ["apps/sim/lib/core/application/content-delivery.ts","apps/sim/lib/mothership/content-delivery/**","apps/sim/lib/**/application/admit-*.ts","apps/sim/lib/**/application/operations.ts","apps/sim/lib/mothership/agent-cli/**","apps/sim/lib/mothership/tools/sandbox-resource-transport.ts"]
+globs: ["apps/sim/lib/core/application/**","apps/sim/lib/mothership/content-delivery/**","apps/sim/lib/workspace-files/application/**","apps/sim/lib/knowledge/application/**","apps/sim/lib/table/application/**","apps/sim/lib/workflows/application/**","apps/sim/lib/mothership/chat/application/**","apps/sim/lib/**/application/operations.ts","apps/sim/lib/mothership/agent-cli/**","apps/sim/lib/mothership/tools/**","apps/sim/app/api/v2/**"]
 ---
 
 <!-- Generated from .claude/rules/sim-content-delivery.md by `bun run skills:sync`. Edit the source, not this file. -->

apps/sim/lib/core/application/content-delivery.test.ts

@@ -208,7 +208,7 @@ describe('runDeclaredContentOperation', () => {
     expect(refusal).toMatchObject({ message: expect.stringContaining('The change was saved') })
   })
 
-  it('answers with a gate failure during a write only after its effects completed', async () => {
+  it('answers a write whose gate failed as saved-not-shown, after its effects completed', async () => {
     const failure = new Error('history unavailable')
     const failing: ContentDeliveryGate = {
       async admit() {
@@ -231,7 +231,37 @@ describe('runDeclaredContentOperation', () => {
       ).catch((error: unknown) => error)
     )
     expect(events).toEqual(['withheld=1', 'audit recorded'])
-    expect(outcome).toBe(failure)
+    /** Not the gate's error: that would invite a retry of a write that already committed. */
+    expect(outcome).toBeInstanceOf(ContentRefusedError)
+    expect(outcome).toMatchObject({ message: expect.stringContaining('The change was saved') })
+    /** The request's transport still sees the failure, to answer it itself (the proxy's 502). */
+    const record = await observeContentDelivery(failing, () =>
+      trackContentDelivery(() =>
+        runDeclaredContentOperation(SOURCED, CHAT, async () => {
+          await admitWrittenContent(CHAT, [source('written')])
+          return 'saved row'
+        }).catch(() => 'answered')
+      )
+    )
+    expect(record.gateFailure).toBe(failure)
+  })
+
+  it('does not let a nested use case account for what its enclosing operation returns', async () => {
+    const admitAll: ContentDeliveryGate = {
+      async admit(sources) {
+        return sources.map(() => true)
+      },
+    }
+    const outcome = await observeContentDelivery(admitAll, () =>
+      runDeclaredContentOperation(SOURCED, CHAT, async () => {
+        await runDeclaredContentOperation(SOURCED, CHAT, async () => {
+          await admitContent(CHAT, [source('nested')])
+          return 'nested'
+        })
+        return 'outer content never classified'
+      }).catch((error: unknown) => error)
+    )
+    expect(outcome).toBeInstanceOf(ContentDeliveryUnaccountedError)
   })
 
   it('rethrows a gate failure at once when no operation is open to answer with it', async () => {
@@ -265,7 +295,12 @@ describe('trackContentDelivery', () => {
       noteContentDeliveryOperation(SOURCED, CHAT)
       return 'response'
     })
-    expect(record).toEqual({ result: 'response', unaccounted: true, admitted: [] })
+    expect(record).toEqual({
+      result: 'response',
+      unaccounted: true,
+      committedWrite: false,
+      admitted: [],
+    })
   })
 
   it('does not report a none operation, or a sourced one another caller ran', async () => {

apps/sim/lib/core/application/content-delivery.ts

@@ -64,8 +64,6 @@ interface ContentLedger {
   admitted: ContentProvenance[]
   /** A committed write's returned item was refused; the operation answers with a refusal. */
   withheldAfterWrite?: ContentSourceKind
-  /** A committed write's gate failed; the operation completes, then answers with this error. */
-  failedAfterWrite?: unknown
   /** The gate itself failed while this ledger was open, whatever the route then answered. */
   gateFailure?: unknown
 }
@@ -185,8 +183,24 @@ export class ContentAdmission {
   }
 }
 
+/**
+ * Accounts for the operation running now and the requests around it. An enclosing operation is
+ * not marked: a nested use case's admission covers what it returned, not what the outer one adds.
+ */
 function markAccounted(): void {
-  for (let ledger = ledgers.getStore(); ledger; ledger = ledger.parent) ledger.accounted = true
+  const current = ledgers.getStore()
+  if (!current) return
+  current.accounted = true
+  for (let ledger = current.parent; ledger; ledger = ledger.parent) {
+    if (ledger.declaration === undefined) ledger.accounted = true
+  }
+}
+
+/** A committed write returned an item of `kind` its gate refused or could not decide. */
+function withholdAfterWrite(kind: ContentSourceKind): void {
+  for (let ledger = ledgers.getStore(); ledger; ledger = ledger.parent) {
+    ledger.withheldAfterWrite ??= kind
+  }
 }
 
 /**
@@ -254,25 +268,20 @@ export async function admitWrittenContent(
     admission = await admitContent(principal, sources)
   } catch (error) {
     /**
-     * A gate that cannot decide (its record could not be written) must not undo the write's
-     * audit and effects either: every item is withheld, and the operation answers with the error.
+     * A gate that cannot decide (its record could not be written) must not undo the write's audit
+     * and effects either: every item is withheld, and the operation answers "saved, not shown".
+     * A transport that must answer the failure itself reads it from the request's ledger.
      */
-    const ledger = ledgers.getStore()
-    if (!ledger) throw error
-    for (let open: ContentLedger | undefined = ledger; open; open = open.parent) {
-      open.failedAfterWrite ??= error
-    }
+    const [first] = sources
+    if (!ledgers.getStore() || !first) throw error
+    withholdAfterWrite(first.kind)
     return new ContentAdmission(
       sources,
       sources.map(() => false)
     )
   }
   const refused = admission.refused[0]
-  if (refused) {
-    for (let ledger = ledgers.getStore(); ledger; ledger = ledger.parent) {
-      ledger.withheldAfterWrite ??= refused.kind
-    }
-  }
+  if (refused) withholdAfterWrite(refused.kind)
   return admission
 }
 
@@ -285,14 +294,28 @@ export async function admitEach<T>(
   principal: Principal,
   items: readonly T[],
   classify: (items: readonly T[]) => Promise<readonly ContentSource[]>,
-  options: { written?: boolean } = {}
+  options: { written?: ContentSourceKind } = {}
 ): Promise<T[]> {
   if (items.length === 0 || !mustClassifyContent(principal)) {
     await admitContent(principal, [])
     return [...items]
   }
-  const sources = await classify(items)
-  const admission = await (options.written ? admitWrittenContent : admitContent)(principal, sources)
+  if (!options.written) {
+    const sources = await classify(items)
+    const admission = await admitContent(principal, sources)
+    return items.filter((_, index) => admission.admits(sources[index]))
+  }
+  /** A committed write whose items cannot even be classified is saved, then withheld whole. */
+  let sources: readonly ContentSource[]
+  try {
+    sources = await classify(items)
+  } catch (error) {
+    if (!ledgers.getStore()) throw error
+    markAccounted()
+    withholdAfterWrite(options.written)
+    return []
+  }
+  const admission = await admitWrittenContent(principal, sources)
   return items.filter((_, index) => admission.admits(sources[index]))
 }
 
@@ -308,13 +331,26 @@ export async function requireAdmittedContent(
   if (!admission.admits(source)) throw new ContentRefusedError(source.kind)
 }
 
+/** The gate failed during a dispatched request, which may have committed a write first. */
+export class ContentGateFailure extends Error {
+  constructor(
+    readonly gateFailure: unknown,
+    readonly committedWrite: boolean
+  ) {
+    super('The content gate failed during the request', { cause: gateFailure })
+    this.name = 'ContentGateFailure'
+  }
+}
+
 /** What one dispatched request delivered, as the transport that dispatched it needs to know. */
 export interface ContentDeliveryRecord<T> {
   result: T
   /** Chat ran a `sourced` operation whose use case never accounted for its content. */
   unaccounted: boolean
   /** The gate failed to decide during the request; see {@link ContentLedger.gateFailure}. */
   gateFailure?: unknown
+  /** A write committed, then its returned content was withheld: the change happened. */
+  committedWrite: boolean
   /** Provenance of every admitted item, in admission order. */
   admitted: readonly ContentProvenance[]
 }
@@ -333,10 +369,18 @@ export async function trackContentDelivery<T>(
     accounted: false,
     admitted: [],
   }
-  const result = await ledgers.run(ledger, execute)
+  let result: T
+  try {
+    result = await ledgers.run(ledger, execute)
+  } catch (error) {
+    /** The request threw because its gate failed; its transport answers that, not the error. */
+    if (ledger.gateFailure === undefined) throw error
+    throw new ContentGateFailure(ledger.gateFailure, ledger.withheldAfterWrite !== undefined)
+  }
   return {
     result,
     unaccounted: ledger.sourcedOperationAdmitted && !ledger.accounted,
+    committedWrite: ledger.withheldAfterWrite !== undefined,
     admitted: ledger.admitted,
     ...(ledger.gateFailure === undefined ? {} : { gateFailure: ledger.gateFailure }),
   }
@@ -381,7 +425,6 @@ export async function runDeclaredContentOperation<T>(
   }
   const result = await ledgers.run(ledger, execute)
   await complete?.(result)
-  if (ledger.failedAfterWrite) throw ledger.failedAfterWrite
   if (operation.contentDelivery === 'sourced' && !ledger.accounted) {
     throw new ContentDeliveryUnaccountedError(operation.id)
   }

apps/sim/lib/knowledge/__integration__/execution-archive-provenance.integration.ts

@@ -208,6 +208,12 @@ async function assertBlockedConsumers(
   } else {
     expect(imported).toMatchObject({ failed: [] })
     expect(imported.added).toHaveLength(1)
+    /** The accepted document queued its processing; that work is removed with the fixture. */
+    const events = await db
+      .select({ id: outboxEvent.id })
+      .from(outboxEvent)
+      .where(sql`${outboxEvent.payload}::jsonb ->> 'documentId' = ${imported.added[0].documentId}`)
+    trackedEventIds.push(...events.map((event) => event.id))
   }
 }
 

apps/sim/lib/knowledge/application/admit-knowledge-content.ts

@@ -1,5 +1,5 @@
 import type { Principal } from '@sim/auth/principal'
-import { admitEach } from '@/lib/core/application/content-delivery'
+import { admitEach, type ContentSourceKind } from '@/lib/core/application/content-delivery'
 import {
   classifyKnowledgeChunks,
   classifyKnowledgeDocumentMetadata,
@@ -10,6 +10,9 @@ interface KnowledgeAdmissionOptions {
   written?: boolean
 }
 
+const writtenAs = (options: KnowledgeAdmissionOptions, kind: ContentSourceKind) =>
+  options.written ? { written: kind } : {}
+
 /**
  * Admits each document whose rendered metadata a result returns, on its own, and returns the ids
  * admitted. `value` is what the result shows of the document, so only secrets it contains are
@@ -32,7 +35,7 @@ export async function admitKnowledgeDocuments(
         view: { derived: item.value },
       }))
     },
-    options
+    writtenAs(options, 'knowledge-document')
   )
   return new Set(admitted.map((item) => item.id))
 }
@@ -59,7 +62,7 @@ export async function admitKnowledgeChunks(
         view: chunk.value === undefined ? ('complete' as const) : { derived: chunk.value },
       }))
     },
-    options
+    writtenAs(options, 'knowledge-chunk')
   )
   return new Set(admitted.map((chunk) => chunk.id))
 }

apps/sim/lib/knowledge/application/chunks.test.ts

@@ -1,4 +1,8 @@
-import { createSessionPrincipal } from '@sim/testing/factories/principal.factory'
+import type { Principal } from '@sim/auth/principal'
+import {
+  createDelegatedPrincipal,
+  createSessionPrincipal,
+} from '@sim/testing/factories/principal.factory'
 import {
   knowledgeContextsMock,
   knowledgeContextsMockFns,
@@ -8,6 +12,7 @@ import { workspaceAuthzMock, workspaceAuthzMockFns } from '@sim/testing/mocks/wo
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 const mocks = vi.hoisted(() => ({
+  admitKnowledgeChunks: vi.fn(),
   queryChunks: vi.fn(),
   batchChunkOperation: vi.fn(),
   createChunk: vi.fn(),
@@ -36,6 +41,11 @@ vi.mock('@/lib/knowledge/model-input-provenance', () => ({
 
 vi.mock('@/providers/utils', () => providersUtilsMock)
 
+vi.mock('@/lib/knowledge/application/admit-knowledge-content', () => ({
+  admitKnowledgeChunks: mocks.admitKnowledgeChunks,
+}))
+
+import { admitContent } from '@/lib/core/application/content-delivery'
 import { ForbiddenOperationError } from '@/lib/core/application/forbidden'
 import { WORKSPACE_ACCESS_SCOPE } from '@/lib/knowledge/access/scope'
 import { KnowledgeDocumentNotReadyError } from '@/lib/knowledge/application/chunk-errors'
@@ -57,6 +67,65 @@ describe('knowledge chunk application use cases', () => {
     })
   })
 
+  /** Stands in for the gate's verdicts; accounting is the real primitive's, as the helper's is. */
+  function admitOnly(ids: string[]) {
+    mocks.admitKnowledgeChunks.mockImplementation(async (principal: Principal) => {
+      await admitContent(principal, [])
+      return new Set(ids)
+    })
+  }
+
+  describe('a searched chunk listing', () => {
+    const page = {
+      chunks: [
+        { id: 'chunk-1', content: 'first' },
+        { id: 'chunk-2', content: 'second' },
+      ],
+      pagination: { total: 7, limit: 2, offset: 4, hasMore: true },
+    }
+
+    beforeEach(() => {
+      knowledgeContextsMockFns.mockResolveCanonicalActiveKnowledgeDocumentContext.mockResolvedValue(
+        {
+          workspaceId: 'workspace-1',
+          workspaceOrganizationId: null,
+          allowPersonalApiKeys: true,
+          billedAccountUserId: 'billing-owner-1',
+          access: { get: async () => WORKSPACE_ACCESS_SCOPE },
+          knowledgeBaseId: 'knowledge-1',
+          knowledgeBase: { id: 'knowledge-1' },
+          documentId: 'document-1',
+          document: { id: 'document-1', processingStatus: 'completed', enabled: true },
+        }
+      )
+      mocks.queryChunks.mockResolvedValue(page)
+      admitOnly(['chunk-1'])
+    })
+
+    it('reports Chat a total of the matches it was given, not of withheld ones', async () => {
+      const result = await listKnowledgeChunks.execute({
+        principal: createDelegatedPrincipal({
+          workspaceId: 'workspace-1',
+          audience: 'sim:knowledge',
+        }),
+        input: { knowledgeBaseId: 'knowledge-1', documentId: 'document-1', search: 'token' },
+      })
+
+      expect(result.chunks.map((chunk) => chunk.id)).toEqual(['chunk-1'])
+      expect(result.pagination).toEqual({ total: 5, limit: 2, offset: 4, hasMore: true })
+    })
+
+    it('keeps the full total for a person', async () => {
+      admitOnly(['chunk-1', 'chunk-2'])
+      const result = await listKnowledgeChunks.execute({
+        principal: createSessionPrincipal(),
+        input: { knowledgeBaseId: 'knowledge-1', documentId: 'document-1', search: 'token' },
+      })
+
+      expect(result.pagination).toEqual(page.pagination)
+    })
+  })
+
   it('returns a typed transient failure before querying chunks for a processing document', async () => {
     const promise = listKnowledgeChunks.execute({
       principal: createSessionPrincipal(),