Build an AI Knowledge Agent Over Scattered Internal Docs: Connectors, Permissions, Freshness and Cost Per Question (2026)

type Cursor = { sourceId: string; token: string | null; lastEventAt: string | null };

type ChangeBatch = {
  upserts: { externalId: string; body: string; updatedAt: string; sharing: SharingModel }[];
  deletes: string[];
  nextToken: string | null;
  hasMore: boolean;
};

const STALENESS_LIMITS_MIN: Record<string, number> = {
  chat: 15, wiki: 120, drive: 120, code: 60, tickets: 60,
};

export async function syncSource(source: Source, store: Store, index: Index): Promise<void> {
  let cursor = await store.getCursor(source.id);

  do {
    const batch: ChangeBatch = await source.fetchChanges(cursor.token, { limit: 200 });

    for (const item of batch.upserts) {
      const doc = normalise(source.kind, item);          // source-specific chunking
      const principals = await resolvePrincipals(item.sharing); // the lossy, important part
      // Idempotent upsert keyed on (sourceId, externalId, chunkOrdinal).
      await index.upsertChunks(source.id, doc, principals, item.updatedAt);
    }

    // Deletions must be explicit. A source that does not emit them needs the
    // reconciliation pass below, or content stays answerable after removal.
    for (const externalId of batch.deletes) {
      await index.tombstone(source.id, externalId);
    }

    // Write side effects FIRST, then advance the cursor. A crash here
    // re-processes a batch, which is harmless because upserts are idempotent.
    cursor = { sourceId: source.id, token: batch.nextToken, lastEventAt: new Date().toISOString() };
    await store.putCursor(cursor);

    if (!batch.hasMore) break;
  } while (true);

  await checkStaleness(source, cursor);
}

/**
 * Permissions sync on a faster cadence than content. A stale document is an
 * accuracy problem; a stale permission is a security problem.
 */
export async function syncPermissions(source: Source, idp: Idp, index: Index): Promise<void> {
  const changes = await idp.membershipChangesSince(await index.lastAclSync(source.id));
  for (const change of changes) {
    await index.reprojectPrincipals(source.id, change.groupId, change.members);
  }
  await index.markAclSynced(source.id, new Date().toISOString());
}

/**
 * For sources with unreliable deletion events: list identifiers at the source
 * and tombstone anything in the index that no longer exists there.
 */
export async function reconcile(source: Source, index: Index): Promise<number> {
  const live = new Set(await source.listAllIds());
  const indexed = await index.listExternalIds(source.id);
  let removed = 0;
  for (const id of indexed) {
    if (!live.has(id)) { await index.tombstone(source.id, id); removed++; }
  }
  return removed;
}

async function checkStaleness(source: Source, cursor: Cursor): Promise<void> {
  const limit = STALENESS_LIMITS_MIN[source.kind] ?? 120;
  const last = cursor.lastEventAt ? Date.parse(cursor.lastEventAt) : 0;
  if (Date.now() - last > limit * 60_000) {
    // A silent connector looks identical to a quiet source. Alarm on it.
    await alert("connector_stale", { sourceId: source.id, limitMinutes: limit });
  }
}

type Principal = string; // "user:u_123" | "group:g_eng" | "org:acme"

export type Candidate = {
  chunkId: string;
  externalId: string;
  sourceId: string;
  text: string;
  updatedAt: string;
  score: number;
};

/**
 * EARLY BINDING. The access predicate is inside the query. The database
 * never returns a row this principal set cannot see, so there is nothing
 * to post-filter and nothing to leak through result counts or timing.
 */
export async function retrieve(
  db: Db,
  queryVector: number[],
  queryText: string,
  principals: Principal[],
  limit = 60,
): Promise<Candidate[]> {
  return db.query<Candidate>(
    "SELECT c.chunk_id, c.external_id, c.source_id, c.text, c.updated_at, " +
    "       (0.6 * (1 - (c.embedding <=> $1)) + 0.4 * ts_rank(c.tsv, plainto_tsquery($2))) AS score " +
    "  FROM chunks c " +
    " WHERE c.principals && $3::text[] " +   // <-- the predicate, not a filter
    "   AND c.tombstoned_at IS NULL " +
    " ORDER BY score DESC " +
    " LIMIT $4",
    [queryVector, queryText, principals, limit],
  );
}

/**
 * LATE BINDING. Re-check authorisation against the source of truth for only
 * the documents that will be cited. Closes the window between permission
 * syncs. Fails CLOSED: an unresolvable check drops the document.
 */
export async function reauthorise(
  candidates: Candidate[],
  userId: string,
  authz: Authz,
  timeoutMs = 120,
): Promise<{ allowed: Candidate[]; dropped: number }> {
  const settled = await Promise.allSettled(
    candidates.map((c) =>
      withTimeout(authz.canRead(userId, c.sourceId, c.externalId), timeoutMs),
    ),
  );

  const allowed: Candidate[] = [];
  let dropped = 0;
  settled.forEach((r, i) => {
    // Only an explicit true keeps the document. Rejections, timeouts and
    // undefined all drop it. Failing open here is how a revoked employee
    // gets a citation to a document they can no longer open.
    if (r.status === "fulfilled" && r.value === true) allowed.push(candidates[i]);
    else dropped++;
  });

  return { allowed, dropped };
}

import { z } from "zod";

export const AnswerSchema = z.object({
  markdown: z.string().min(1),
  claims: z.array(z.object({
    text: z.string(),
    chunk_id: z.string(),
    quoted_span: z.string().min(15),
  })).min(1),
  conflicts: z.array(z.object({
    subject: z.string(),
    positions: z.array(z.object({ value: z.string(), chunk_id: z.string() })).min(2),
  })).default([]),
});

export type Answer = z.infer<typeof AnswerSchema>;

type Chunk = { id: string; text: string; updatedAt: string; sourceKind: string };

const DISCLOSE_AFTER_DAYS: Record<string, number> = {
  handbook: 540, adr: 730, wiki: 270, chat: 90, ticket: 180, code: 365, drive: 365,
};

export type GateResult =
  | { decision: "answer"; staleNoticeDays: number | null; conflicts: number }
  | { decision: "refuse"; reason: string; detail?: string };

export async function answerGate(
  answer: Answer,
  chunks: Map<string, Chunk>,
  allowedChunkIds: Set<string>,   // survivors of the late-binding ACL re-check
  verifyGrounded: (a: Answer, c: Map<string, Chunk>) => Promise<number>,
  threshold = 0.75,
): Promise<GateResult> {
  let newest = 0;
  let oldestKindOverThreshold: number | null = null;

  for (const claim of answer.claims) {
    const chunk = chunks.get(claim.chunk_id);

    // 1. free: the cited chunk must exist and the span must really be in it
    if (!chunk) return { decision: "refuse", reason: "citation_unknown_chunk" };
    if (!norm(chunk.text).includes(norm(claim.quoted_span))) {
      return { decision: "refuse", reason: "citation_span_not_found" };
    }

    // 2. free: never cite a document this person cannot open. A citation to an
    // inaccessible doc discloses its existence, which is usually the point.
    if (!allowedChunkIds.has(claim.chunk_id)) {
      return { decision: "refuse", reason: "cited_inaccessible_document" };
    }

    // 3. free: freshness is date arithmetic, per source kind
    const ageDays = Math.floor((Date.now() - Date.parse(chunk.updatedAt)) / 86_400_000);
    newest = Math.max(newest, Date.parse(chunk.updatedAt));
    const limit = DISCLOSE_AFTER_DAYS[chunk.sourceKind] ?? 365;
    if (ageDays > limit) {
      oldestKindOverThreshold = Math.max(oldestKindOverThreshold ?? 0, ageDays);
    }
  }

  // 4. the only model call: a cheap verifier, seeing the answer cold
  const score = await verifyGrounded(answer, chunks);
  if (score < threshold) {
    return { decision: "refuse", reason: "below_grounding_threshold" };
  }

  // Stale evidence is not a refusal. It is an answer with a mandatory notice.
  const staleNoticeDays =
    oldestKindOverThreshold !== null
      ? Math.floor((Date.now() - newest) / 86_400_000)
      : null;

  return { decision: "answer", staleNoticeDays, conflicts: answer.conflicts.length };
}

function norm(s: string): string {
  return s.toLowerCase().replace(/\s+/g, " ").trim();
}

1. Week 1
   Document model, index schema, journal

   One document model across sources, the chunk table with its principal array and tombstone column, the query journal, cost attribution. Everything downstream assumes this shape.

2. Weeks 2‑3
   Connector one, done properly

   Cursor with durable checkpoint, backfill, tombstones, reconciliation, staleness alarm, rate-limit budget. This is the template five more will be copied from, so shortcuts here compound.

3. Week 4
   ACL resolver and identity integration

   Group expansion with cache invalidation on membership change, per-source sharing model projection, and the rule that it never fails open. The security core of the system.

4. Weeks 5‑6
   Connectors two and three

   Faster because the pattern exists, but each source has its own permission quirks — link sharing, inherited folders, contractor groups nested inside team groups.

5. Week 7
   Normalisation and chunking

   Headings for wikis, threads for chat, description-plus-review for pull requests, problem-plus-resolution for tickets. Uniform token chunking retrieves fragments, and fragments produce useless answers.

6. Week 8
   Retrieval, re-check, freshness ranker

   ACL predicate inside the query, late-binding re-check on the final documents, per-source recency decay, canonical-source boost. Configuration, not machine learning.

7. Weeks 9‑10
   Synthesis, verification, refusals

   Cited spans, the four-check gate, staleness disclosure, conflict as an explicit output type, and refusals specific enough to be useful rather than discouraging.

8. Weeks 11‑14
   Evals, dashboards, rollout

   Fifty hand-labelled questions with known-correct sources, the six operational signals, the documentation-gap report, and one department for three weeks before anyone else sees it.