Skip to main content

Slack Archive Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Periodic sync of a client's Slack channel history into a provider-generic archive envelope ledger (Postgres) + a dedicated slack-archive Haystack dataset, searchable by the Specialist through a governed, flag-gated read tool — fully separate from the KB.

Architecture: Own tables + own dataset + direct HaystackClient calls — the archive never creates org_documents rows, so it is structurally invisible to the KB list, the Tavus conversational-context builder, the draft KB retrieval path (source: "kb" → default-kb), and the FORCE-RLS ACL post-filter. Sync follows the P4.8 connector pattern (leader-gated scheduler + BullMQ fan-out). Envelopes (archive_records) are the replay substrate: reindexing never re-fetches Slack. Content lands raw ("author, date, text" — no scrubbing, explicit product decision). Default sync start: yesterday.

Tech Stack: NestJS 11, TypeORM + PG migration, BullMQ, Nango (slack_archive user-token integration: channels:read,channels:history,users:read), Haystack/pgvector via existing HaystackClient, tool gateway (AGENT_TOOL_REGISTRY → runtime-tool-executor), feature flags.

Branch: feat/slack-archive (off origin/dev).

Execution status: Tasks 1–5 assigned to subagent ArchivePipeline, Tasks 6–7 to subagent ArchiveTool (running in parallel; disjoint file ownership). Tasks 8–10 = integrator (main session).

File Structure​

api/migrations/1811200000000-CreateSlackArchive.ts        (create)
api/src/slack-archive/slack-archive.entities.ts           (create: 3 entities)
api/src/slack-archive/slack-archive-sync.service.ts       (create: leader-gated scheduler)
api/src/slack-archive/slack-archive-fetch.processor.ts    (create: fetch → envelopes → day-docs → ingest)
api/src/slack-archive/slack-archive-config.service.ts     (create)
api/src/slack-archive/slack-archive.controller.ts         (create: orgs/:orgId/slack-archive)
api/src/slack-archive/slack-archive.module.ts             (create)
api/src/haystack/archive-retrieval.service.ts             (create: dataset-scoped search)
api/src/haystack/haystack.types.ts                        (modify: widen HaystackDataset)
api/src/agent-api/tool-registry.ts                        (modify: slack_archive_search entry)
api/src/runtime-control-plane/runtime-tool-executor.service.ts (modify: handler 'slack_archive')
api/src/conversations/tool-resolution.util.ts             (modify: FLAG_GATED_TOOLS)
api/src/feature-flags/feature-flags.constants.ts          (modify: slack_archive_enabled)
api/src/integrations/credentials/{credentials.dto,credentials.service,sources.service}.ts (modify: provider slack_archive)
api/src/scheduler/scheduler.constants.ts                  (modify: SLACK_ARCHIVE_SYNC)
api/src/config/typeorm-data-source.config.ts + api/src/data-source.ts (modify: entity registration — BOTH)
api/src/app.module.ts                                     (modify: SlackArchiveModule — Task 8 only)

Locked contracts (all tasks MUST match)​

  • Dataset: "slack-archive" · Nango provider/integration type: slack_archive · Tool: slack_archive_search (kind bespoke, handler slack_archive) · Flag: slack_archive_enabled (default OFF) · Queue: slack-archive-fetch · Scheduler job: SLACK_ARCHIVE_SYNC, env SLACK_ARCHIVE_SYNC_CRON default */15 * * * *.
  • Envelope (user-specified, plus isolation fields): 
    archive_records(id uuid PK, org_id uuid NOT NULL, source_id uuid NOT NULL FK→slack_archive_sources ON DELETE CASCADE,
      provider text NOT NULL, doc_type text NOT NULL, doc_external_id text NOT NULL,
      raw_data text NOT NULL, sha text NOT NULL, meta jsonb NOT NULL DEFAULT '{}',
      created_at timestamptz DEFAULT now(), UNIQUE(org_id, provider, doc_external_id))
    provider ∈ 'slack'|'github'|'notion' (text, no DB enum). doc_type ∈ 'message'|'comment'|'issue'. For Slack: doc_external_id = "${channelId}:${message.ts}", raw_data = "[YYYY-MM-DD HH:MM] <author>: <text>", sha = sha256(raw_data), meta = {channel_id, channel_name, author, ts}.
  • Config: slack_archive_sources(id, org_id UNIQUE, credential_id, enabled bool DEFAULT false, channels jsonb '[]', include_future bool DEFAULT false, since_date date DEFAULT (CURRENT_DATE - 1), cursors jsonb '{}', last_synced_at, last_error, updated_by, created_at, updated_at).
  • Index state: archive_index_state(id, org_id, source_id, channel_id, day date, content_hash, haystack_doc_id, message_count int, last_ingested_at, created_at, updated_at, UNIQUE(org_id, channel_id, day)).
  • Day-doc: header #<channel_name> — <day> + one line per message in envelope raw_data order (by ts). Haystack doc id ${orgId}:slack-archive:${contentHash}; ingest meta {dataset:"slack-archive", org_id, specialist_id:null, audience:"ai_internal", channel_id, channel_name, day, provider:"slack"}.
  • ADR-020: every entity docstring names its row (Org-scoped; served to agents only through the org-filtered slack-archive dataset; never joined into KB surfaces). All queries filter org_id.

Task 1: Migration + entities (owner: ArchivePipeline)​

Files: Create api/migrations/1811200000000-CreateSlackArchive.ts, api/src/slack-archive/slack-archive.entities.ts; Modify both entity arrays.

  • Step 1: Read a recent migration (e.g. ls api/migrations | tail) and mirror its class shape. Write the migration creating the three tables from Locked contracts, plus indexes: archive_records(org_id), archive_index_state(org_id), slack_archive_sources(org_id) (UNIQUE).
  • Step 2: Entities:
/**
 * ADR-020: Org-scoped connector configuration (one row per org's Slack
 * workspace archive). Not row-1 PII in itself; the archived CONTENT is served
 * to agents only through the org-filtered "slack-archive" Haystack dataset —
 * never joined into KB surfaces (kb list, Tavus context, draft retrieval).
 */
@Entity("slack_archive_sources")
export class SlackArchiveSource {
  @PrimaryGeneratedColumn("uuid") id: string;
  @Column({ name: "org_id", type: "uuid" }) orgId: string;
  @Column({ name: "credential_id", type: "uuid" }) credentialId: string;
  @Column({ default: false }) enabled: boolean;
  @Column({ type: "jsonb", default: () => "'[]'" }) channels: Array<{ id: string; name: string }>;
  @Column({ name: "include_future", default: false }) includeFuture: boolean;
  @Column({ name: "since_date", type: "date" }) sinceDate: string;
  @Column({ type: "jsonb", default: () => "'{}'" }) cursors: Record<string, string>;
  @Column({ name: "last_synced_at", type: "timestamptz", nullable: true }) lastSyncedAt: Date | null;
  @Column({ name: "last_error", type: "text", nullable: true }) lastError: string | null;
  @Column({ name: "updated_by", type: "uuid", nullable: true }) updatedBy: string | null;
  @CreateDateColumn({ name: "created_at" }) createdAt: Date;
  @UpdateDateColumn({ name: "updated_at" }) updatedAt: Date;
}

ArchiveRecord and ArchiveIndexState follow the same style with the Locked-contract columns (snake_case column names, camelCase properties).

  • Step 3: Register all three entities in api/src/config/typeorm-data-source.config.ts AND api/src/data-source.ts (both — forFeature alone throws "No metadata" at runtime, #3189).
  • Step 4: cd api && npx tsc --noEmit for the touched files compile. Commit: feat(slack-archive): schema + entities.

Task 2: Nango provider slack_archive (owner: ArchivePipeline)​

Files: Modify api/src/integrations/credentials/credentials.dto.ts, credentials.service.ts, sources.service.ts.

  • Step 1: Mirror every slack_search touchpoint: ChannelType union + NANGO_CONNECT_PROVIDERS + NANGO_BACKED_TYPES + CHANNEL_REQUIRED_KEYS: { slack_archive: [] } + NANGO_USER_SCOPES: { slack_archive: "channels:read,channels:history,users:read" } + validateNangoProvider message + SOURCE_TOOL_BY_PROVIDER: { slack_archive: "slack_archive_search" } + display name "Slack History".
  • Step 2: Extend the existing Nango-provider spec pattern (credentials.slack-search-provider.spec.ts style) with a slack_archive case asserting the session carries user_scopes: "channels:read,channels:history,users:read". Run only that spec. Commit.

Task 3: Sync service (owner: ArchivePipeline)​

Files: Create api/src/slack-archive/slack-archive-sync.service.ts; Modify api/src/scheduler/scheduler.constants.ts.

  • Step 1: Clone the structure of api/src/kb/kb-connector-sync.service.ts: onModuleInit self-registration under new SCHEDULER_JOBS.SLACK_ARCHIVE_SYNC, cron from SLACK_ARCHIVE_SYNC_CRON (default */15 * * * *), body leader-gated via LeaderElectionService.acquireLeadership.
  • Step 2: Body: SELECT slack_archive_sources WHERE enabled = true, join credential (integration_credentials.integration_type = 'slack_archive', status active); per source inside try/catch (failure → log + last_error, continue siblings): enqueue { sourceId, orgId } on queue slack-archive-fetch with a stable jobId (slack-archive:${sourceId} — dedupes overlapping ticks).
  • Step 3: Spec api/test/slack-archive-sync.spec.ts: picks enabled+active; skips disabled/inactive; no-ops when leadership not acquired; one failing source doesn't block the next. Run only it. Commit.

Task 4: Fetch processor (owner: ArchivePipeline)​

Files: Create api/src/slack-archive/slack-archive-fetch.processor.ts (queue slack-archive-fetch, attempts 3, exponential backoff 5s).

  • Step 1: Resolve org slug + Nango connection id exactly as runtime-tool-executor.service.ts#executeSlackWorkspaceSearch does (stored connection id from the credential row; NangoService.httpProxy({ orgSlug, provider: 'slack_archive', method: 'GET', endpoint, connectionId, params })).
  • Step 2: Once per run: users.list → Map<userId, displayName> (fallback: raw id). If includeFuture: conversations.list (types=public_channel&exclude_archived=true, paginate) → append unseen channels to channels jsonb.
  • Step 3: Per configured channel (own try/catch): page conversations.history with oldest = cursors[channelId] ?? slackTs(sinceDate), limit=200; hard cap 5000 messages per source per run (stop paging; cursor persists; next tick resumes). Skip subtypes (channel_join, bot_message without text, etc.).
  • Step 4: Per message → envelope insert (NO scrubbing):
const line = `[${fmtUtc(msg.ts)}] ${authors.get(msg.user) ?? msg.user}: ${msg.text ?? ""}`;
await repo.createQueryBuilder().insert().into(ArchiveRecord).values({
  orgId, sourceId, provider: "slack", docType: "message",
  docExternalId: `${channelId}:${msg.ts}`,
  rawData: line, sha: sha256(line),
  meta: { channel_id: channelId, channel_name: channelName, author: authors.get(msg.user) ?? msg.user, ts: msg.ts },
}).orIgnore().execute();   // ON CONFLICT (org_id, provider, doc_external_id) DO NOTHING
  • Step 5: For each touched (channel, day): rebuild the day-doc from envelopes (ORDER BY meta->>'ts'), header + lines; hash = sha256(text); compare archive_index_state:
    • unchanged → skip;
    • new → haystackClient.ingest({ org_id, dataset: "slack-archive", filename: ${channelName}-${day}.txt, content_b64, mime_type: "text/plain", meta: <Locked contract meta> }), insert state row;
    • changed → evict previous doc first (use the same client eviction call HaystackIngestionService/golden-answer-sync use for evict-then-reindex — verify the exact method name in haystack.client.ts when implementing), then ingest + update state row (evict-then-reindex, P2.4 pattern).
  • Step 6: Advance cursors[channelId] (max ts) ONLY after that channel's envelope + state writes persisted; set last_synced_at. Spec api/test/slack-archive-fetch.spec.ts: envelope mapping + ON CONFLICT, exact day-doc format, unchanged-hash skips, changed-hash evicts-then-ingests, cursor-after-persist, 5000 cap. Run only it. Commit.

Task 5: Config API (owner: ArchivePipeline)​

Files: Create api/src/slack-archive/slack-archive-config.service.ts, slack-archive.controller.ts, slack-archive.module.ts.

  • Step 1: Controller under orgs/:orgId/slack-archive, guard pattern copied from sources.controller.ts (client admin roles + platform superadmin/AM):
    • GET / → config + status (enabled, channels, sinceDate, lastSyncedAt, lastError, per-channel cursor presence);
    • PUT / → { enabled, channels: [{id,name}], includeFuture, sinceDate } (validate shape; stamp updatedBy); creates the row on first call (default sinceDate = yesterday);
    • GET /channels → live conversations.list via Nango proxy (feeds the picker UI later);
    • POST /resync → reset cursors = {}, clear lastError (envelope dedup makes refetch idempotent).
  • Step 2: Module: BullMQ queue + forFeature entities + Nango/scheduler/leader imports mirroring kb.module.ts. Do NOT touch app.module.ts.
  • Step 3: Spec api/test/slack-archive-config.spec.ts: PUT validation, role guard (client admin + AM allowed, member rejected), resync resets cursors. Run only it. Commit.

Task 6: Dataset + ArchiveRetrievalService (owner: ArchiveTool)​

Files: Modify api/src/haystack/haystack.types.ts; Create api/src/haystack/archive-retrieval.service.ts; Modify api/src/haystack/haystack.module.ts.

  • Step 1: Widen HaystackDataset with "slack-archive". Verify resolveHaystackSource still defaults unknowns to default-kb (KB draft path unreachable for the new dataset) and kb.service.ts search source mapping never produces it. Check rag-side (rag/pipelines/*/pipeline_wrapper.py) passes arbitrary dataset strings; widen any trivial allowlist.
  • Step 2: Service — direct client, no org_documents ACL, fail-open:
async search(orgId: string, query: string, opts?: { topK?: number }) {
  if (await this.pauseState.isIntegrationPausedSafe(orgId, "ragflow")) return { results: [] };
  try {
    const resp = await this.client.retrieve({
      org_id: orgId, org_ids: [orgId], query,
      source: "slack-archive", top_k: Math.min(Math.max(opts?.topK ?? 8, 1), 25),
      specialist_id: null, specialist_ids: [], audiences: ["ai_internal"],
      jurisdiction: null, as_of: new Date().toISOString().slice(0, 10),
    });
    return { results: resp.chunks.map(c => ({
      channel: c.meta?.channel_name ?? c.meta?.channel_id ?? "unknown",
      day: c.meta?.day ?? null,
      snippet: (c.content ?? "").slice(0, 700),
      score: c.score ?? 0,
    })) };
  } catch (err) { this.logger.warn(`archive search failed org=${orgId}: ${err}`); return { results: [] }; }
}

Docstring: isolation = org_id/org_ids filter over an org-scoped dataset; no KB governance semantics (status/effective-date) apply; deliberately no FORCE-RLS post-filter.

  • Step 3: Provide + export from haystack.module.ts. Spec api/test/archive-retrieval.service.spec.ts. Run only it. Commit.

Task 7: Tool registry + executor + flag (owner: ArchiveTool)​

Files: Modify tool-registry.ts, runtime-tool-executor.service.ts (+ its module imports), tool-resolution.util.ts, feature-flags.constants.ts, feature-flag.service.spec.ts, tool-dispatch-wiring.spec.ts.

  • Step 1: Registry entry:
slack_archive_search: {
  name: 'slack_archive_search', kind: 'bespoke', handler: 'slack_archive',
  description: "Search the org's archived Slack workspace history (synced copy). Results are unvetted conversation history, NOT verified knowledge: attribute claims to their author and date; do not state them as confirmed fact.",
  integrationTypes: ['slack_archive'],
  actions: [{ name: 'search', description: 'Search archived Slack messages',
    parameters: { type: 'object', required: ['query'], properties: {
      query: { type: 'string', description: 'Search query over archived Slack messages' },
      top_k: { type: 'integer', default: 8, maximum: 25 } } } }],
},
  • Step 2: Executor: add handler branch 'slack_archive' → this.archiveRetrieval.search(input.orgId, params.query, { topK: params.top_k }), returned in the sibling handlers' { ok: true, ... } envelope. Import HaystackModule into the runtime-control-plane module if absent.
  • Step 3: Flag: slack_archive_enabled in feature-flags.constants.ts (both touchpoints, default OFF, comment) + FLAG_GATED_TOOLS['slack_archive_search'] = 'slack_archive_enabled' + update the flag-list spec.
  • Step 4: Extend tool-dispatch-wiring.spec.ts with advertisement ON/OFF cases mirroring slack_search. Run only the touched specs. Commit.

Task 8: Integration (owner: main session)​

  • Register SlackArchiveModule in api/src/app.module.ts.
  • Verify both entity arrays contain the three entities; verify no file conflicts between the two workers.
  • cd api && npx tsc --noEmit — zero errors.

Task 9: Verification (owner: main session)​

  • Run all new spec files together: cd api && npx jest slack-archive archive-retrieval tool-dispatch-wiring feature-flag.service --silent → all pass.
  • Negative checks (the reason the design exists): grep confirms api/src/slack-archive/ never imports HaystackIngestionService/OrgDocument; resolveHaystackSource unreachable for slack-archive from the draft path.
  • Update api/CLAUDE.md module index (+ 5-line api/src/slack-archive/CLAUDE.md). Commit.

Task 10: Ship checklist (config, not code)​

  • Nango dashboard: create Slack OAuth app + integration key slack_archive (user scopes channels:read,channels:history,users:read; leave bot scopes empty).
  • Run migration on target env (migrate.yml workflow).
  • Pilot client admin: Sources page → connect Slack History (Connect-session flow works as soon as the provider is registered).
  • PUT orgs/:orgId/slack-archive → { enabled: true, channels: [...], sinceDate: <yesterday default> } (engineer/AM via API until the picker UI lands).
  • Publish pilot Specialist Skill with slack_archive_search in tools[] + read-class binding; flip slack_archive_enabled (+ tool_dispatch_enabled) for the pilot org.
  • Smoke: wait one sync tick (or POST /resync) → ask the Specialist a question answerable only from Slack history → draft cites archived content; ToolCall ledger row present; flag-OFF org advertises nothing; KB tab shows no archive docs.

Explicitly deferred (named slots)​

Channel-picker drawer UI (config via API until then) · secret scrubbing (product decision to land raw; revisit before non-pilot orgs) · threads (conversations.replies) · private channels · GitHub/Notion providers (envelope already generic: provider/doc_type fields) · retention/erasure policy semantics (envelope table is the substrate) · budget metering via CostBudgetService (cap-per-run constant only in v0).

Last updated on by Degen