fix(files): cancel resumed runs across replicas, unify anchored file edits, and harden workspace, integration, and presence paths

apps/docs/content/docs/integrations/file.mdx

@@ -198,7 +198,7 @@ Apply one precise edit to an existing text file without rewriting it. Use search
 | `anchor` | string | No | For insert_after, the complete line after which content is inserted. |
 | `startAnchor` | string | No | For delete_between, the complete first line to delete. The start anchor is removed. |
 | `endAnchor` | string | No | For delete_between, the complete ending boundary line. The end anchor remains in the file. |
-| `occurrence` | number | No | For anchored edits, which matching anchor occurrence to use, starting at 1. Defaults to 1. |
+| `occurrence` | number | No | For anchored edits, which start or insert anchor match to use, starting at 1. Defaults to 1. The end anchor is the first match after it. |
 | `expectedRevision` | string | No | Refuse the edit unless the file still holds the content this revision names, as returned by Get File or an earlier write. Use it so an edit computed from what you read cannot overwrite someone else’s change. |
 
 #### Output

apps/docs/content/docs/knowledgebase/connectors.mdx

@@ -200,7 +200,7 @@ Images smaller than **2 KiB**, SVG, HEIC, and AVIF files are not indexed through
 
 New knowledge-base Confluence connections enable image reading by default. Connections created before image support keep it off until you choose **Read with AI**; their next sync then indexes images.
 
-The default admission limit is **500 new or changed images per sync**. Remaining images wait for another sync. Self-hosted operators can configure `KB_IMAGE_EXTRACTIONS_PER_SYNC`. Unchanged images reuse their extraction, including during a full resync; transient or malformed responses can cause additional model requests.
+The default admission limit is **500 new or changed images per sync**. Remaining images wait for another sync. With **Member accounts**, each run that continues a sync can read up to 500. Self-hosted operators can configure `KB_IMAGE_EXTRACTIONS_PER_SYNC`. Unchanged images reuse their extraction, including during a full resync; transient or malformed responses can cause additional model requests.
 
 Image extraction follows [model billing](/platform/costs): BYOK usage is billed by the provider, while platform-key usage follows the deployment's Sim billing policy. Choose **Don't index** to stop image indexing; the next sync removes indexed images. Switching back processes them again and incurs the applicable model usage.
 

apps/docs/content/docs/platform/self-hosting/environment-variables.mdx

@@ -124,7 +124,7 @@ For Ollama on the Docker host, use `OLLAMA_URL=http://host.docker.internal:11434
 | Variable | Description |
 |----------|-------------|
 | `KB_IMAGE_EXTRACTION_MODEL` | Gemini model that reads Confluence image attachments into searchable text. Defaults to `gemini-3.8-flash`; a value that is not a Gemini 3 or later model in the catalog falls back to the default. Needs `GEMINI_API_KEY` (or a workspace Gemini key); without one, image attachments are left out of syncs |
-| `KB_IMAGE_EXTRACTIONS_PER_SYNC` | New or changed images one connector sync sends for reading, at least `1`. Defaults to `500`; the rest wait for the next sync, which reports as partial |
+| `KB_IMAGE_EXTRACTIONS_PER_SYNC` | New or changed images one connector sync run sends for reading, at least `1`. Defaults to `500`; the rest wait for the next sync, which reports as partial. With member accounts, each continuation run can read up to this many |
 | `KB_CONFIG_VISION_REQUESTS_PER_MINUTE` | Image reads per minute per Gemini key, shared by every sync on the deployment. Defaults to `300`; lower it to match a key with a smaller Gemini quota |
 | `KB_EMBEDDING_MODEL` | Embedding model for new knowledge bases. Defaults to `text-embedding-3-small`; use `ollama/<model>` for a model on your own Ollama. An unrecognised hosted model id falls back to the default, but an `ollama/` id is taken at face value — if that model is not on the server, knowledge-base creation fails rather than falling back |
 | `EMBEDDING_OUTPUT_DIMS` | Stored vector width: `384`, `768`, `1024`, `1536`, or `3072`. Hosted models default to `1536` when supported, otherwise their largest storable width. Invalid hosted widths fall back with a warning; Ollama can infer its actual width when unset |

apps/docs/openapi-v2-files-audit.json

@@ -6920,7 +6920,7 @@
                     "description": "Replacement text. An empty string clears the interior."
                   },
                   "occurrence": {
-                    "description": "Matching anchor occurrence, starting at 1. Defaults to 1.",
+                    "description": "Which beforeAnchor match to use, starting at 1. Defaults to 1. afterAnchor is the first match after it.",
                     "type": "integer",
                     "minimum": 1,
                     "maximum": 9007199254740991
@@ -6948,7 +6948,7 @@
                     "description": "Text to insert."
                   },
                   "occurrence": {
-                    "description": "Matching anchor occurrence, starting at 1. Defaults to 1.",
+                    "description": "Which anchor match to insert after, starting at 1. Defaults to 1.",
                     "type": "integer",
                     "minimum": 1,
                     "maximum": 9007199254740991
@@ -6976,7 +6976,7 @@
                     "description": "Ending boundary line. This end anchor remains."
                   },
                   "occurrence": {
-                    "description": "Matching anchor occurrence, starting at 1. Defaults to 1.",
+                    "description": "Which startAnchor match to use, starting at 1. Defaults to 1. endAnchor is the first match after it.",
                     "type": "integer",
                     "minimum": 1,
                     "maximum": 9007199254740991

apps/realtime/src/handlers/file-doc-presence.integration.ts

@@ -0,0 +1,198 @@
+import { mkdir, writeFile } from 'node:fs/promises'
+import { createServer, type Server as HttpServer } from 'node:http'
+import { dirname, resolve } from 'node:path'
+import {
+  FILE_DOC_EVENTS,
+  FILE_DOC_SCHEMA_VERSION,
+  FILE_DOC_SEED,
+  type FileDocPresence,
+  type JoinFileDocSuccess,
+} from '@sim/realtime-protocol/file-doc'
+import { getErrorMessage } from '@sim/utils/errors'
+import { generateId } from '@sim/utils/id'
+import { createAdapter } from '@socket.io/redis-adapter'
+import { createClient, type RedisClientType } from 'redis'
+import { Server } from 'socket.io'
+import { io as connect, type Socket } from 'socket.io-client'
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
+import * as Y from 'yjs'
+import type { AuthenticatedSocket } from '@/middleware/auth'
+
+const redisUrl = process.env.REDIS_URL
+if (!redisUrl) throw new Error('Integration setup must provide disposable Redis')
+const internalSecret = process.env.INTERNAL_API_SECRET
+
+const projectId = generateId()
+const fileId = generateId()
+const roomName = `project-file-doc:${projectId}/${fileId}`
+const docId = generateId()
+const seedDoc = new Y.Doc()
+seedDoc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+seedDoc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.docIdKey, docId)
+const seed = Buffer.from(Y.encodeStateAsUpdate(seedDoc)).toString('base64')
+seedDoc.destroy()
+
+interface Replica {
+  io: Server
+  url: string
+  pub: RedisClientType
+  sub: RedisClientType
+  shutdownStore: () => Promise<void>
+}
+
+const checks: { name: string; status: string; durationMs: number; error?: string }[] = []
+const clients: Socket[] = []
+const replicas: Replica[] = []
+let app: HttpServer
+
+async function listen(server: HttpServer) {
+  await new Promise<void>((done) => server.listen(0, '127.0.0.1', done))
+  const address = server.address()
+  if (!address || typeof address === 'string') throw new Error('Missing fixture port')
+  return `http://127.0.0.1:${address.port}`
+}
+
+/**
+ * One realtime replica: a fresh module graph (its own rooms, owners and document store, as in a
+ * separate process) behind its own Socket.IO server on the shared Redis adapter.
+ */
+async function startReplica(): Promise<Replica> {
+  vi.resetModules()
+  const fileDoc = await import('@/handlers/file-doc')
+  const fileDocStore = await import('@/handlers/file-doc-store')
+  const { MemoryRoomManager } = await import('@/rooms')
+  await fileDocStore.initFileDocStore(redisUrl)
+  const pub: RedisClientType = createClient({ url: redisUrl })
+  const sub: RedisClientType = createClient({ url: redisUrl })
+  await Promise.all([pub.connect(), sub.connect()])
+  const http = createServer()
+  const io = new Server(http)
+  io.adapter(createAdapter(pub, sub))
+  fileDoc.registerFileDocPresenceRelay(io)
+  const manager = new MemoryRoomManager(io)
+  io.on('connection', (socket) => {
+    const authed = socket as AuthenticatedSocket
+    authed.userId = socket.handshake.auth.actor
+    authed.userName = `Fixture ${socket.handshake.auth.actor}`
+    authed.userImage = 'https://fixture.invalid/avatar.png'
+    fileDoc.setupWorkspaceFileDocHandlers(authed, manager)
+    socket.on('disconnect', () => fileDoc.cleanupFileDocForSocket(socket.id, io, true))
+  })
+  const replica = {
+    io,
+    url: await listen(http),
+    pub,
+    sub,
+    shutdownStore: () => fileDocStore.getFileDocStore().shutdown(),
+  }
+  replicas.push(replica)
+  return replica
+}
+
+beforeAll(async () => {
+  app = createServer((request, response) => {
+    response.setHeader('Content-Type', 'application/json')
+    const prefix = `/api/internal/project-file-doc/${projectId}/${fileId}/`
+    if (request.headers['x-api-key'] !== internalSecret || !request.url?.startsWith(prefix)) {
+      response.writeHead(401).end('{}')
+      return
+    }
+    const action = request.url.slice(prefix.length)
+    if (action === 'access') {
+      response.end(JSON.stringify({ projectId, fileId, canRead: true, canWrite: true, docId }))
+    } else if (action === 'seed') {
+      response.end(JSON.stringify({ update: seed, version: 1 }))
+    } else if (action === 'persist') {
+      response.end(JSON.stringify({ status: 'persisted', version: 2 }))
+    } else {
+      response.writeHead(404).end('{}')
+    }
+  })
+  process.env.NEXT_PUBLIC_APP_URL = await listen(app)
+})
+
+/** Joins the shared file through `replica`, recording every roster the client receives. */
+async function joinThrough(replica: Replica, actor: string) {
+  const socket = connect(replica.url, { transports: ['websocket'], auth: { actor } })
+  clients.push(socket)
+  const rosters: string[][] = []
+  socket.on(FILE_DOC_EVENTS.PRESENCE, (payload: FileDocPresence) => {
+    if (payload.fileId === fileId) rosters.push(payload.users.map((user) => user.socketId).sort())
+  })
+  await new Promise<void>((done) => socket.once('connect', done))
+  const joined = new Promise<JoinFileDocSuccess>((done, fail) => {
+    socket.once(FILE_DOC_EVENTS.JOIN_SUCCESS, done)
+    socket.once(FILE_DOC_EVENTS.JOIN_ERROR, (error) => fail(new Error(error.code)))
+  })
+  const clientDoc = new Y.Doc()
+  const clientId = clientDoc.clientID
+  clientDoc.destroy()
+  socket.emit(FILE_DOC_EVENTS.JOIN, {
+    fileId,
+    projectId,
+    clientId,
+    schemaVersion: FILE_DOC_SCHEMA_VERSION,
+  })
+  await joined
+  const id = socket.id
+  if (!id) throw new Error('Connected socket has no id')
+  return { socket, id, latestRoster: () => rosters.at(-1) }
+}
+
+function check(name: string, run: () => Promise<void>) {
+  it(name, async () => {
+    const start = performance.now()
+    try {
+      await run()
+      checks.push({ name, status: 'passed', durationMs: performance.now() - start })
+    } catch (error) {
+      checks.push({
+        name,
+        status: 'failed',
+        durationMs: performance.now() - start,
+        error: getErrorMessage(error),
+      })
+      throw error
+    }
+  })
+}
+
+describe('File editor presence across replicas sharing the Redis adapter', () => {
+  check(
+    'each replica lists collaborators joined through the other, and drops a departed one',
+    async () => {
+      const first = await startReplica()
+      const second = await startReplica()
+      const alice = await joinThrough(first, 'fixture-alice')
+      const bob = await joinThrough(second, 'fixture-bob')
+      const both = [alice.id, bob.id].sort()
+
+      await expect.poll(alice.latestRoster, { timeout: 5_000 }).toEqual(both)
+      await expect.poll(bob.latestRoster, { timeout: 5_000 }).toEqual(both)
+
+      bob.socket.disconnect()
+      await expect.poll(alice.latestRoster, { timeout: 5_000 }).toEqual([alice.id])
+    }
+  )
+})
+
+afterAll(async () => {
+  for (const client of clients) client.disconnect()
+  for (const replica of replicas) {
+    await replica.io.close()
+    await replica.shutdownStore()
+  }
+  const cleanup = replicas[0]?.pub
+  if (cleanup?.isOpen) {
+    const keys = await cleanup.keys(`filedoc:*:${roomName}`)
+    if (keys.length) await cleanup.del(keys)
+  }
+  for (const replica of replicas) {
+    for (const client of [replica.pub, replica.sub]) if (client.isOpen) await client.quit()
+  }
+  await new Promise<void>((done) => app.close(() => done()))
+  const reportPath =
+    process.env.FILE_DOC_PRESENCE_REPORT_PATH ?? resolve('test-results/file-doc-presence.json')
+  await mkdir(dirname(reportPath), { recursive: true })
+  await writeFile(reportPath, JSON.stringify({ checks }, null, 2))
+})

apps/realtime/src/handlers/file-doc.redis-store.test.ts

@@ -160,7 +160,6 @@ function createIo(sockets: FakeSocket[]) {
     in: vi.fn(() => ({ socketsLeave: () => {} })),
     local: { to },
     sockets: { sockets: { get: (id: string) => sockets.find((socket) => socket.id === id) } },
-    on: vi.fn(),
     serverSideEmit: vi.fn(),
   } as unknown as IRoomManager['io']
 }

apps/realtime/src/handlers/file-doc.test.ts

@@ -61,7 +61,10 @@ interface SentMessage {
 }
 
 /** An `io` mock that records every server-originated emit with its target/except. */
-function createIo(deliver?: (message: SentMessage) => void) {
+function createIo(
+  deliver?: (message: SentMessage) => void,
+  fetchSockets = async () => [...testSockets.values()]
+) {
   const sent: SentMessage[] = []
   const emit = (message: SentMessage) => {
     sent.push(message)
@@ -72,7 +75,7 @@ function createIo(deliver?: (message: SentMessage) => void) {
   const channel = (target: string, excluded: string[] = []) => ({
     except: (exclude: string) => channel(target, [...excluded, exclude]),
     emit: (event: string, payload: unknown) => emit({ target, except: excluded, event, payload }),
-    fetchSockets: async () => [...testSockets.values()],
+    fetchSockets,
   })
   const to = vi.fn((target: string) => channel(target))
   const inFn = vi.fn((socketId: string) => ({
@@ -87,7 +90,6 @@ function createIo(deliver?: (message: SentMessage) => void) {
       in: inFn,
       local: { to },
       sockets: { sockets: testSockets },
-      on: vi.fn(),
       serverSideEmit: vi.fn(),
     } as unknown as IRoomManager['io'],
     sent,
@@ -1107,6 +1109,71 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(sent.some((m) => m.event === FILE_DOC_EVENTS.MESSAGE)).toBe(false)
   })
 
+  it('delivers a roster change that arrived while the last presence read was failing', async () => {
+    vi.useFakeTimers()
+    try {
+      let reads = 0
+      const { io, sent } = createIo(undefined, async () => {
+        reads += 1
+        if (reads === 3) await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 })
+        if (reads <= 3) throw new Error('Redis publish failed')
+        return [...testSockets.values()]
+      })
+      const a = setup('socket-a', io, { userId: 'user-a' })
+      const b = setup('socket-b', io, { userId: 'user-b' })
+
+      await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+      await vi.advanceTimersByTimeAsync(1_000)
+
+      const roster = sent.filter((m) => m.event === FILE_DOC_EVENTS.PRESENCE).at(-1)?.payload as
+        | { users: Array<{ socketId: string }> }
+        | undefined
+      expect(roster?.users.map((user) => user.socketId).sort()).toEqual(['socket-a', 'socket-b'])
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('re-reads the roster when membership changes during a successful presence read', async () => {
+    let reads = 0
+    const { io, sent } = createIo(undefined, async () => {
+      reads += 1
+      const snapshot = [...testSockets.values()]
+      if (reads === 1) await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 })
+      return snapshot
+    })
+    const a = setup('socket-a', io, { userId: 'user-a' })
+    const b = setup('socket-b', io, { userId: 'user-b' })
+
+    await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await vi.waitFor(() => expect(reads).toBe(2))
+
+    const roster = sent.filter((m) => m.event === FILE_DOC_EVENTS.PRESENCE).at(-1)?.payload as
+      | { users: Array<{ socketId: string }> }
+      | undefined
+    expect(roster?.users.map((user) => user.socketId).sort()).toEqual(['socket-a', 'socket-b'])
+  })
+
+  it('does not retry a presence read the adapter timed out', async () => {
+    vi.useFakeTimers()
+    try {
+      let reads = 0
+      const { io, sent } = createIo(undefined, async () => {
+        reads += 1
+        throw new Error('timeout reached while waiting for fetchSockets response')
+      })
+      const a = setup('socket-a', io, { userId: 'user-a' })
+
+      await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+      await vi.advanceTimersByTimeAsync(1_000)
+
+      expect(reads).toBe(1)
+      expect(sent.some((m) => m.event === FILE_DOC_EVENTS.PRESENCE)).toBe(false)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('broadcasts a server-authenticated presence roster on join, one entry per session', async () => {
     const { io, sent } = createIo()
     const a = setup('socket-a', io, { userId: 'user-a', userName: 'Ada', userImage: 'ada.png' })