Skip to main content

P4.8 — KB ingestion connectors (R3.5) implementation plan

Date: 2026-07-10. Branch: feat/specialist-value-output-p2plus. Phase 4 (final program task).

Goal (R3.5)​

Scheduled sync from external sources (Notion / Drive / websites, via Nango) into K1/K2. Synced content enters the same curation / versioning / eval path as uploads — live external data never bypasses the KB store into retrieval. A failing connector degrades to stale-but-served KB, never blocks retrieval. Connector-synced docs are versioned and re-embedded on change.

Load-bearing safety invariant (why the shape is what it is)​

The "never bypasses the KB store" rule is enforced structurally by the allowedDocumentIds authorization gate (api/src/haystack/CLAUDE.md): a doc with an org_documents shadow row is governed fail-CLOSED (status/parseStatus/specialist/effectiveDate arms); a doc without a row rides fail-OPEN and bypasses all governance. Therefore connector content MUST land as an authorized org_documents shadow row via HaystackIngestionService.upload() — the exact path uploads, golden answers (P2.4), and URL fetches use. This is what puts external content under Org×Specialist isolation + status governance. It does NOT go into kb_documents.

Deterministic id haystackDocumentId(orgId, dataset, contentHash) = ${orgId}:${dataset}:${contentHash} gives versioning + change-detection + dedup for free.

DECISION POINT (可逆默认，用户暂离期做，可推翻)​

Connector docs land status="pending_review" (AM curates before retrieval), NOT status="approved" (auto-live). Rationale (first-principles):

  • R3.5 says "same curation path" — curation implies human-in-loop. External-source content (Notion/Drive) quality is uncontrolled; auto-injecting into Specialist retrieval risks polluting replies.
  • Platform core is always-HITL (autoRespondThreshold=101). Auto-live external content contradicts that.
  • Mirrors P4.2 (loop_proposal is a draft, human publishes) — consistent.
  • Review states (pending_review|approved|rejected|superseded) + AM review surface already exist (KbService approve/reject) — zero new cost.
  • Reversible: it is ONE field value on the upload() call. To make it auto-live, change status:"pending_review" → status:"approved". The seam is a single constant.

The stronger reading ("same path as uploads" — uploads go approved) would auto-live. Both are one field. Pinned here for the user to flip if they prefer auto-live.

Steps (mirror existing precedents exactly)​

Step 1 — Nango: add Notion + Drive providers​

Files: api/src/integrations/credentials/credentials.service.ts (NANGO_BACKED_TYPES:79, CHANNEL_REQUIRED_KEYS:62, validateNangoProvider:188 — widen union + error string), credentials.dto.ts (ChannelType union). NangoService/NangoClient unchanged — proxy() is provider-generic.

  • New vs extend: extend (code small). Nango dashboard provider config + sync scripts = new, out-of-repo (needs user).
  • Local-verifiable: unit (provider validation). Real OAuth: blocked on Nango creds.

Step 2 — Scheduled sync job (BullMQ, leader-gated, error-isolated)​

New api/src/kb/kb-connector-sync.service.ts + api/src/kb/connector-fetch.processor.ts. New KB_CONNECTOR_SYNC in scheduler/scheduler.constants.ts + scheduler.service.ts onModuleInit. Register queue in kb.module.ts + QueueMetricsCollector (common/).

  • Pattern = Gmail inbound poll (gmail-inbound-polling.service.ts): SchedulerService.scheduleRepeatableJob + stable jobId (NOT @Cron/setInterval — active-active), self-registered handler via setHandler, leader-gated (LeaderElectionService.acquireLeadership()), per-connector try/catch isolation (one broken connector must not block the rest — this IS the "failing connector degrades, never blocks" AC).
  • Worker RLS: wrap DB work in rlsContext.withOrgContext(orgId, ...) (workers run outside HTTP).
  • Payload self-contained: {orgId, specialistId, provider, connectionId, externalDocId, ...}.
  • Local-verifiable: yes — mock NangoService.proxy, assert fan-out + leader gate + isolation.

Step 3 — Land as authorized shadow row via upload()​

Reuse HaystackIngestionService.upload() as-is. New OrgDocumentSource enum value connector_sync (or per-provider) in entities.ts:1357 + migration for the source column default handling (expand-contract). Call upload(orgId, dataset, syntheticFile, { source:"connector_sync", status:"pending_review", contentHash, specialistId, metadata:{external_doc_id, provider} }).

  • Mirror golden-answer-sync.processor.ts (self-contained payload → upload()).
  • Local-verifiable: yes — upload() + allowedDocumentIds gate fully testable; mirror golden-answer-sync.processor.spec.

Step 4 — Change detection + re-embed​

Content-hash change detection via org_documents.contentHash (compute sha256 of pulled content, like url-fetch). Unchanged content → same haystack_doc_id → upsert in place (no re-embed). Changed content → evict-then-reindex (P2.4 pattern, golden-answer-sync.processor.ts:44-79): deleteByMeta(orgId, {external_doc_id}) FIRST then upload() — two sequential awaits in one job (idempotent under retry). OR use HaystackIngestionService.replaceDocument() (per-row delete-previous + re-ingest).

  • Requires adding external_doc_id to the rag meta allowlist (rag/pipelines/retrieval/pipeline_wrapper.py _serialize_document) if deleting by that meta key.
  • Local-verifiable: delete+upload sequencing + idempotency yes (mock Haystack). rag allowlist round-trip: needs rag deploy.

Step 5 — Failure isolation​

Per-connector try/catch in poll loop (Gmail precedent) + best-effort swallow in processor (golden-answer precedent) + never touch retrieval path (a failed sync leaves prior rows = stale-but-served; retrieval fail-opens on Haystack outage anyway).

  • Local-verifiable: yes — inject a throwing connector, assert others still sync + retrieval unaffected.

What's locally verifiable vs blocked on user​

Locally verifiable (mock Nango + Haystack client): the entire orchestration — scheduler registration, leader gating, per-connector fan-out + error isolation, upload()→authorized shadow row, allowedDocumentIds fail-closed governance, evict-then-reindex idempotency, content-hash dedup, best-effort fail-open. Specs mirror golden-answer-sync.processor.spec / gmail-inbound-polling.service.spec / url-fetch.processor.spec. Use real uuid literals in seed data (PG-shard CI trap, api/CLAUDE.md).

Blocked on real credentials (HANDOFF to user): actual Nango OAuth apps for Notion + Google Drive, NANGO_SERVER_URL/NANGO_SECRET_KEY, per-org admin OAuth authorization (browser Connect flow via createNangoSession/confirmNangoConnection), Nango-side sync scripts that list/fetch Notion pages / Drive files. This is the same class of block as P4.7's Hermes-container: logic fully built + tested with mocks, real external-service wiring left as a documented handoff.

⚠️ Verify BEFORE flipping connectors to approved (auto-live) — the rag /admin/delete-by-meta route (P4.8 code-review, Important): HaystackIngestionService.deleteByMeta (the evict half of evict-then-reindex) POSTs /admin/delete-by-meta on the rag service, but that route is NOT defined in the tracked rag/app.py (which only wraps hayhooks.create_app()). This is pre-existing and platform-wide — the shipped P2.4 golden-answer processor uses the identical path — so P4.8 inherits, not introduces, it. external_doc_id IS persisted into rag base_meta at ingestion (survives on every chunk), so IF the route exists in the deployed rag image (outside this repo) it works for connectors identically to golden answers. But if the route is absent in prod, deleteByMeta has been silently no-op'ing (best-effort try/catch swallows the 404) for golden answers too — meaning an EDITED connector doc could leave its prior version's chunks reachable after re-embed. pending_review masks this today (un-curated chunks aren't retrievable anyway). Action for the handoff: confirm the deployed rag exposes POST /admin/delete-by-meta reading caller filters → store.delete_by_filter (AND-ed with org_id), OR fix rag to add it, before connectors go auto-live. Verified locally: the change-detection READ (metadata ->> 'external_doc_id') runs correctly on real PG16 — it's only the rag-side DELETE that needs prod confirmation.

Isolation matrix docstring (mandatory)​

New org_documents source value + connector service: ADR-020 row-1 (Customer PII / KB, Org × Specialist). upload() already persists org_id + specialist_id on every row. The connector service query paths (if any read org_documents) must filter both.

TDD order​

  1. Scheduler constant + registration (spec: job registered, leader-gated).
  2. Connector-fetch processor (spec: mock Nango.proxy + Haystack, assert upload() called with pending_review + content-hash + external_doc_id metadata; changed content → evict-then-reindex; unchanged → upsert-in-place; throwing connector isolated).
  3. Nango provider widening (spec: Notion/Drive validate).
  4. OrgDocumentSource enum + migration.
  5. Module wiring + QueueMetricsCollector.
  6. Verify: api tsc/lint, new specs green; document the OAuth handoff.
Last updated on by Degen