fix(feed): commit live versions in feed_seq order; document share grants

README.md

@@ -87,6 +87,8 @@ Returns a `Hono<TenantEnv>` sub-app. Mount it at `/api/tenants/:tenantId/memory`
 
 A missing principal answers `401` and a missing grant answers `403`. `callerResolver` is optional: it resolves a non-session caller to a `{ tenantId, principalId }` before the same grant checks run, and a malformed result answers `500`.
 
+Sharing on `add` needs grants the host may have to mint. `share.principals` writes each peer a `search` grant on `memory.doc:<documentId>` only when `grantStore` has a `putGrant` method; Interchange's stock grant store has none, so on a stock hub the host mints those grants. `share.tenant` never writes grants: the host grants `search` on `memory.tenant:<tenantId>` to whoever should see tenant-wide documents.
+
 ### `runMemoryMigrations(config, { schema, ftsLanguage })`
 
 Takes the same `DBConfig` and `schema` as Interchange's `runMigrations`. `schema` holds the host's `tenant` and `principal` tables; this package's tables always go in `memory`. `ftsLanguage` is fixed into the full-text index and must match `FTS_LANGUAGE` at runtime.

docs/FEED.md

@@ -26,6 +26,7 @@ memory.feed({ tenantId, principalId, after?, limit?, excludeGenerator? })
 | `excludeGenerator` | Skip versions with this `generator_agent_id` (loop-safe). |
 
 - Ordered by `feed_seq` ascending (Postgres `bigserial` on `memory.version`).
+- Live versions of a tenant commit in `feed_seq` order (a per-tenant transaction lock), so a cursor never passes a version still being written.
 - **Live generation only** — same rule as default search.
 - Capability: `memory` / `search` (same as list/retrieve).
 - Document access: grant-tag post-filter identical to search.

e2e/feed.test.ts

@@ -0,0 +1,86 @@
+import { afterAll, beforeAll, describe, expect, test } from "bun:test";
+import { createInMemoryGrantStore } from "@intx/authz";
+
+import type { Memory } from "../src/memory.js";
+import {
+  allow,
+  createTestDb,
+  createTestMemory,
+  seedPrincipal,
+  testDatabaseUrl,
+  type TestDb,
+} from "./helpers.js";
+
+const alice = { tenantId: "acme", principalId: "alice" };
+const settle = () => new Promise((resolve) => setTimeout(resolve, 500));
+
+describe.skipIf(testDatabaseUrl() === undefined)("feed", () => {
+  let db: TestDb;
+  let memory: Memory;
+
+  beforeAll(async () => {
+    db = await createTestDb();
+    await seedPrincipal(db, "acme", "alice");
+    memory = createTestMemory(
+      db,
+      createInMemoryGrantStore([
+        allow("alice", "add"),
+        allow("alice", "search"),
+      ]),
+    );
+  });
+
+  afterAll(async () => {
+    await memory?.close();
+    await db?.close();
+  });
+
+  // A capture that took a lower feed_seq but commits after a later one must
+  // not fall behind a cursor that already moved past it.
+  test("a version that commits out of feed_seq order is not skipped", async () => {
+    const { documentId } = await memory.add({
+      ...alice,
+      externalRef: "slow",
+      content: { title: "Slow", text: "first draft" },
+    });
+    const start = await memory.feed!({ ...alice });
+
+    // Holding the document row stalls the next capture of it after it has
+    // inserted its version.
+    let release = () => {};
+    const held = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const blocker = db.sql.begin(async (tx) => {
+      await tx`SELECT 1 FROM memory.document WHERE id = ${documentId} FOR UPDATE`;
+      await held;
+    });
+    await settle();
+    const slow = memory.add({
+      ...alice,
+      externalRef: "slow",
+      content: { title: "Slow", text: "second draft" },
+    });
+    await settle();
+    const fast = memory.add({
+      ...alice,
+      externalRef: "fast",
+      content: { title: "Fast", text: "quick note" },
+    });
+    await settle();
+
+    const seen = await memory.feed!({ ...alice, after: start.nextCursor ?? 0 });
+    release();
+    await blocker;
+    const [slowResult] = await Promise.all([slow, fast]);
+    const rest = await memory.feed!({
+      ...alice,
+      after: seen.nextCursor ?? start.nextCursor ?? 0,
+    });
+
+    const versionIds = [...seen.entries, ...rest.entries].map(
+      (e) => e.versionId,
+    );
+    expect(versionIds).toContain(slowResult.versionId);
+  }, 30_000);
+});

src/services/capture.ts

@@ -120,6 +120,9 @@ function parseOptionalDate(value: string | undefined): Date | null {
   return new Date(value);
 }
 
+// Namespaces the two-int4 advisory lock; deliberately arbitrary.
+const FEED_LOCK_NAMESPACE = 0x3e30_7a13;
+
 async function insertVersion(
   tx: Tx,
   input: CaptureInput,
@@ -135,6 +138,14 @@ async function insertVersion(
 ): Promise<string> {
   const versionId = newId("kver");
   const authoritySignals = deriveAuthoritySignals(plan);
+  if (opts.generation === LIVE_GENERATION) {
+    // feed_seq is taken at insert but becomes visible at commit. Holding a
+    // per-tenant lock from here to commit makes live versions commit in
+    // feed_seq order, so a feed cursor never passes a row still in flight.
+    await tx.execute(
+      sql`SELECT pg_advisory_xact_lock(${FEED_LOCK_NAMESPACE}, hashtext(${input.tenantId}))`,
+    );
+  }
   await tx.insert(memoryVersion).values({
     id: versionId,
     tenantId: input.tenantId,