slack: add slack ingest pipeline

.envrc

@@ -1,3 +1,5 @@
 # shellcheck shell=bash
 use flake
-dotenv_if_exists
+
+# shellcheck disable=SC2154
+eval "$("$direnv" dotenv bash <(sops -d --output-type dotenv secrets.yaml))"

.gitignore

@@ -8,4 +8,11 @@ result-*
 
 # cache
 __pycache__
-cocoindex.db
+.pytest_cache/
+
+# runtime state: engine db, vector store
+var/
+
+# secrets: the encrypted secrets.yaml is committed, plaintext never is
+.env
+secrets.yaml.dec

.sops.yaml

@@ -0,0 +1,9 @@
+creation_rules:
+  - key_groups:
+      - age:
+          - age1730f3cxdyh56zw8xcvlmpa7u2x7353wu4u0e58kyx24rsefgp98sxehm6s
+    path_regex: ^secrets\.yaml$
+  - key_groups:
+      - age:
+          - age1730f3cxdyh56zw8xcvlmpa7u2x7353wu4u0e58kyx24rsefgp98sxehm6s
+    path_regex: ^evals/questions\.enc\.yaml$

evals/.gitignore

@@ -0,0 +1,5 @@
+# The labels quote channel content, so only the encrypted copy is committed.
+*
+!.gitignore
+!questions.example.yaml
+!questions.enc.yaml

evals/questions.example.yaml

@@ -0,0 +1,27 @@
+# Question set for `python -m slack_index.evals`. Copy to questions.yaml and fill
+# it with questions whose answers you already know; that file stays untracked
+# because its labels quote channel content.
+#
+#   - question: what someone would actually type, NOT the wording of the message
+#     expected: [source_id, ...]   # thread_ts or file id; any hit in top-k counts
+#     tags: [category, ...]        # scored per tag, so weak spots stay visible
+#
+# Paraphrase away from the message text: a question copied from the message lets
+# lexical retrieval win for free and stops measuring anything.
+#
+# Categories in use:
+#   concept  — about content, worded differently from the message
+#   lexical  — hinges on an exact identifier (part number, hostname, error string)
+#   decision — the answer is an agreement or conclusion, often a short ack
+#   temporal — scheduling, or what happened when
+#   file     — the answer lives in a shared file, not a message
+#   people   — who said or owns something
+- question: "did we settle on bumping that library"
+  expected: ["1700000000.000100"]
+  tags: [decision]
+- question: "what was the env var the deploy script reads"
+  expected: ["1700000100.000200"]
+  tags: [lexical]
+- question: "the design doc shared last week"
+  expected: ["F00EXAMPLE1", "1700000200.000300"]
+  tags: [file]

nix/devshell.nix

@@ -13,6 +13,7 @@ let
       git
       lmdb
       ruff
+      sops
       uv
     ];
     env = {

nix/treefmt.nix

@@ -24,5 +24,9 @@
   settings.formatter.ruff-check.priority = 1;
   settings.formatter.ruff-format.priority = 2;
 
-  settings.global.excludes = [ "**/.direnv/**" ];
+  settings.global.excludes = [
+    "**/.direnv/**"
+    "secrets.yaml"
+    "**/*.enc.yaml"
+  ];
 }

pyproject.toml

@@ -5,13 +5,20 @@ description = "CocoIndex playground"
 requires-python = ">=3.14"
 dependencies = [
   "aiohttp>=3.14.3",
+  "anthropic>=1.8.0",
   "cocoindex[lancedb,sentence-transformers]>=1.0.24",
   "numpy>=2.5.3",
+  "pyyaml>=6.0.3",
   "slack-sdk>=3.44.1",
 ]
 
 [dependency-groups]
-dev = ["mypy>=1.14", "ipykernel>=7.3.0", "pytest>=9.1.1"]
+dev = [
+  "mypy>=1.14",
+  "ipykernel>=7.3.0",
+  "pytest>=9.1.1",
+  "types-pyyaml>=6.0.12.20260906",
+]
 
 [tool.uv]
 package = false

secrets.yaml

@@ -0,0 +1,31 @@
+#ENC[AES256_GCM,data:4nxLgDN+tVSdCkWVbnmeMI0Ro9eo/lSuKsxe6434HNEUnP/Yv6IDw8JWZbrvblc5sZWgnUqI4CStdJet2PXZa+HpfUCx6Sd5WA==,iv:d+1bGA71iMcYFtv5NDOT7Q/k9Vy1+LyOpDuUj6b2KjU=,tag:jCRYG0Wun7yPBaHBi+rlhw==,type:comment]
+#ENC[AES256_GCM,data:VpYCLgmUnaBx5AaY7Tj2E9Lh0umhGdUZxb9yZq+9n3pQo8AidJdYUK5KIB+V2AtdsHjOe/gGwso=,iv:fMfx7xlU92GE1NDR2aPmy/Mdbr0ItgMwUxPlRzU6K3s=,tag:BBWoyREI+8kLPCR245iVnw==,type:comment]
+SLACK_BOT_TOKEN: ENC[AES256_GCM,data:cVNAnDTK1hQamns/dPXY/zodL04dOvb+O7nW/eahdQ0Onf4TWvDmUe3rWVjguiK4KQ8o++v6VV6Vfw==,iv:uu/Ujpc7IObUJu/Xuu2tcl1P8EQNybKIgGEGTdH2Klw=,tag:/cc0FdN0Gp8EPcuR3XG7dQ==,type:str]
+#ENC[AES256_GCM,data:6bfENRrTG/d/4bd6/9AolE3fz6J/1RXqsxHaHFFhgvfLWcm6i8V8Vle9VLdzzmsGUfSMJS0E,iv:NUwJZuwfNaE9CaLcpSTPSyXj2yp6lp6ssKa26y8KLjI=,tag:c0QWWKxvI4BBhkBV896YHA==,type:comment]
+SLACK_CHANNEL_IDS: ENC[AES256_GCM,data:0vGH9saX+EGtCCs=,iv:gZ4bIMQr8OeANw8nMat79A4Uy245+TL8sqUf0jGfVuA=,tag:N654rfTrZJltvVs01C5dsw==,type:str]
+#ENC[AES256_GCM,data:/87rtbLj3SQpP+kasbwH05zqo30SHHaPtaEcG4eSKQgl9hTW7w9288bnq0pnThP50BnjYORMajyzugxWW+5Ru0zjUQ==,iv:4SIzwgrLP0PzAveX/C5PMbFbLwGNvQ8eb8U6tC3JJZ8=,tag:b7zCQ9NgVUz88FvYQhH15g==,type:comment]
+ANTHROPIC_API_KEY: ENC[AES256_GCM,data:LjEV90smqXsLU1HOGD3YShEEmGalXj5cS3A1DHGRingUZH0SKox7ceeOgXWME8ztrhwOOe7BUNmSS1wUiTOgQa9fD7ILhoADzCw6gMcGGAYunC8hrWU2JHhNnSYih1qbjOczssAkf38y6BhW,iv:Uf+MGNzy/PdTus4IloN5tH7ldl8P7VOmZoFU+fnnj6s=,tag:HB8csr+7I2x+9Uk3tK+xiA==,type:str]
+#ENC[AES256_GCM,data:DZfAI37opf/HUr0f6dRn/chYXPZAFPyd0AcjXwEUEcsYxTFDIgJgvCFqCl1cfqYUmgUxNHCB+4dhbc1eazsvLneUFyJvemulhk0scg==,iv:P6+Ek/oAJN20IllyUoJ1WGJ6E7jA1Nq0NbmWs3DLlpI=,tag:8Y8VDnwWbEgXYRkTwA/PYw==,type:comment]
+#ENC[AES256_GCM,data:HoSUNGy3R1zwmp2LvTHKM37P6U5aRe69qzqCtURXT6BAt1F/pCZrvIMyQXU=,iv:ZRziCRWcA+Hca5I7KY+ov0iBkai8UZXd8gjdPDZ1T/Q=,tag:rnnx7Chm2IRkIshKAWGd6g==,type:comment]
+#ENC[AES256_GCM,data:cI58KLZYzOIdZk+eFv9m/cbU1JJg2GOfAYedtkw+jNKbDhT+XV/AuhTh,iv:8NsIc4v4GB9qEBDHycmhu3j1zNSNVpSmtfDx0YbgfIs=,tag:DrCwC/do5WHbijZoztVttw==,type:comment]
+#ENC[AES256_GCM,data:0dOad7gf7q4AvBUIsVtvHR3Hx/9x8W9yQjXRKwO5J0bWVskKgiiDW971hA==,iv:mxMy+eo4DmqfoRN8I0ueMa82krSw5Kp+Fgkh3Evqf20=,tag:TlkZgkqY1BO3LQxTW701pA==,type:comment]
+#ENC[AES256_GCM,data:aCZNjkgqSQkX3WqYfPFcupWK5O6y+EXMJIbrxUoE,iv:4z+NqPAMJuhfylnubb5VDBg4pY4qV17O/EPWJPKU91o=,tag:7FDJsU1HF8clOQ0KMy+c1g==,type:comment]
+#ENC[AES256_GCM,data:J4krqZs26uZbYk9+cL/RbakKUjPMqAIQBsiQKhsY68lDXt9rM9k7znlNpQTxSzCrwWj15wv1GA==,iv:1J13sn/q0HZu4uptsyQ/jCLDKB9SsxyJPIKEtJmNdbg=,tag:24MiA+B0JW/4qq5AhO/btQ==,type:comment]
+#ENC[AES256_GCM,data:FqccKI0ZTaiYx/KWfp+J1l3i0EY5bKa08bHOV1Ut,iv:h8eBY6IRPJ4QfwM+k4jYEHxgTAbnEhBUZ20aN043MQQ=,tag:am/VRBvo7YO/CmTVdbB7Ww==,type:comment]
+#ENC[AES256_GCM,data:zLaA23xAd1sVwBsVXD+Dun1EX6bJbohsVXctjHHa,iv:7+p50PqD8qbz9RhuqcINFGZMHCqXeu4wC0ixY3SACE4=,tag:4YuSzP1jV1/DeDYHFFZiFw==,type:comment]
+#ENC[AES256_GCM,data:5VxRntT1v/dxvxPx49mhTZqjeYOpP/yAuxLilRQDsA2BHk6sdw==,iv:z8jxiqsBwvJATEBNX5jttIHIsKr/3kDJ1AMhIrAl4/Q=,tag:9Ao52YQL1WQA3pJG9mw+/Q==,type:comment]
+sops:
+    age:
+        - enc: |
+            -----BEGIN AGE ENCRYPTED FILE-----
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBuTGRTaDRKR0NOeTU1eEQr
+            NjNXZlB5bUNqR29qR2w5SjMwTjRPNUlpZkNRCmJsZWtObzcrcmx1YkRzc0ZWcjdr
+            ZkNOcXNON2IxdmtQaUtGVFEyM2dCK1EKLS0tIHFvajY3dVBzNEJBYW9oTkF4TnU4
+            di9rd0FJTGV1TUJPSHRpU0NHancwZFEK48gfvnkzQoPgD0kQGrwxzYFp+Agp4k7K
+            C1vBUNEgRXrv0zpnNL7CChXKAb2zzfgGN9x1NOkSQ8Yd7NGjz1cfJg==
+            -----END AGE ENCRYPTED FILE-----
+          recipient: age1730f3cxdyh56zw8xcvlmpa7u2x7353wu4u0e58kyx24rsefgp98sxehm6s
+    lastmodified: "2026-09-23T06:33:35Z"
+    mac: ENC[AES256_GCM,data:PoO2ZUovKek6lSkQhgkofnn4CEoyAL6q3knv1wjI461JaxJ9NYSmdKu9jYlyQy/pC3OzVKtsNLWG8n5Mk98H51gY9NtVFLLHLlrIqyEGMPvi25wfKql2ft4AJVlTd+U8qTcDBb9Dv6rLgMggFl9F4YuVN0lb4V25+AnC2EFiDdA=,iv:B3hOWQh1IKPx0xLUK5usF1Nqu5OJzraTklzukCa3zZs=,tag:idVox70tlxxGQVqcyNThDw==,type:str]
+    unencrypted_suffix: _unencrypted
+    version: 3.13.3

secrets.yaml.example

@@ -0,0 +1,28 @@
+# Copy to secrets.yaml and encrypt it: `sops -e -i secrets.yaml`.
+# Afterwards edit in place with `sops secrets.yaml`; the encrypted file is
+# committed, the plaintext never is.
+#
+# Nothing reads this file at run time. The dev shell's .envrc decrypts it into
+# environment variables; in production the unit supplies the same variables from
+# its own credentials. The application only ever reads the environment.
+
+# Bot token with channels:history, groups:history, files:read, users:read.
+# The bot must be a member of every channel listed below.
+SLACK_BOT_TOKEN: xoxb-replace-me
+
+# Comma-separated channel ids, e.g. C0123ABCD,C0456EFGH
+SLACK_CHANNEL_IDS: ""
+
+# Distillation. An org-scoped key also needs ANTHROPIC_WORKSPACE_ID.
+ANTHROPIC_API_KEY: sk-ant-replace-me
+ANTHROPIC_WORKSPACE_ID: ""
+
+# Optional overrides; the committed defaults in slack_index/config.py are the
+# source of truth, these are for experiments.
+#SLACK_INDEX_EMBED_MODEL: nlpai-lab/KURE-v1
+#SLACK_INDEX_DISTILL_MODEL: claude-haiku-4-5
+#SLACK_INDEX_RERANK_DEVICE: mps
+# Unset or 0 indexes the channel from its first message.
+#SLACK_INDEX_LOOKBACK_DAYS: "0"
+#SLACK_INDEX_POLL_SECONDS: "60"
+#SLACK_INDEX_MAX_FILE_BYTES: "5242880"

slack_index/__init__.py

@@ -0,0 +1 @@
+"""Index Slack channel conversations and shared files into LanceDB."""

slack_index/app.py

@@ -0,0 +1,90 @@
+"""Pipeline entry point.
+
+cocoindex update slack_index/app.py        # one-shot catch-up
+cocoindex update -L slack_index/app.py     # live: re-scan every poll interval
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+
+import cocoindex as coco
+from anthropic import AsyncAnthropic
+from cocoindex.connectors import lancedb
+from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
+from cocoindex.resources.rate_limit import RateLimiter
+from slack_sdk.web.async_client import AsyncWebClient
+
+from slack_index import config
+from slack_index.context import DISTILLER, EMBEDDER, LANCE_DB, SLACK, SLACK_LIMIT
+from slack_index.distill import Distiller
+from slack_index.files import process_file
+from slack_index.models import SlackChunk
+from slack_index.source import SlackChannelFiles, SlackChannelThreads
+from slack_index.threads import process_thread
+
+_settings = config.Settings.from_env()
+
+
+@coco.lifespan
+async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
+    config.VAR_DIR.mkdir(parents=True, exist_ok=True)
+    builder.settings.db_path = config.DB_PATH
+    builder.provide(SLACK, AsyncWebClient(token=config.bot_token()))
+    builder.provide(
+        DISTILLER,
+        Distiller(
+            AsyncAnthropic(
+                api_key=config.anthropic_api_key(),
+                default_headers=config.anthropic_headers(),
+            ),
+            config.DISTILL_MODEL,
+        ),
+    )
+    builder.provide(SLACK_LIMIT, RateLimiter(config.SLACK_REQUESTS_PER_SECOND))
+    builder.provide(EMBEDDER, SentenceTransformerEmbedder(_settings.embed_model))
+    builder.provide(LANCE_DB, await lancedb.connect_async(str(config.LANCEDB_URI)))
+    yield
+
+
+@coco.fn
+async def app_main(settings: config.Settings) -> None:
+    table = await lancedb.mount_table_target(
+        LANCE_DB,
+        config.TABLE_NAME,
+        await lancedb.TableSchema.from_class(SlackChunk, primary_key=["id"]),
+    )
+    table.declare_vector_index(column="embedding")
+
+    client = coco.use_context(SLACK)
+    limiter = coco.use_context(SLACK_LIMIT)
+    for channel in settings.channel_ids:
+        threads = SlackChannelThreads(
+            client,
+            limiter,
+            channel,
+            lookback=settings.lookback,
+            poll_interval=settings.poll_interval,
+        )
+        files = SlackChannelFiles(
+            client,
+            limiter,
+            channel,
+            lookback=settings.lookback,
+            poll_interval=settings.poll_interval,
+        )
+        # The channel is part of the subpath so each channel keeps its own
+        # component subtree — and its own rows — across runs.
+        await coco.mount_each(
+            coco.ComponentSubpath("threads", channel), process_thread, threads, table
+        )
+        await coco.mount_each(
+            coco.ComponentSubpath("files", channel),
+            process_file,
+            files,
+            table,
+            settings.max_file_bytes,
+        )
+
+
+app = coco.App(coco.AppConfig(name="SlackIndex"), app_main, settings=_settings)

slack_index/chunking.py

@@ -0,0 +1,71 @@
+"""Text -> chunks -> embedded rows, shared by the thread and file pipelines."""
+
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass
+
+import cocoindex as coco
+from cocoindex.connectors import lancedb
+from cocoindex.ops.text import RecursiveSplitter
+from cocoindex.resources.chunk import Chunk
+from cocoindex.resources.id import IdGenerator
+
+from slack_index.config import CHUNK_OVERLAP, CHUNK_SIZE
+from slack_index.context import EMBEDDER
+from slack_index.models import SlackChunk
+
+_splitter = RecursiveSplitter()
+
+
+@dataclass(frozen=True, slots=True)
+class ChunkMeta:
+    """Row fields every chunk of one source shares."""
+
+    kind: str
+    channel: str
+    source_id: str
+    # Every source-side id this document answers for — a window covers several
+    # message timestamps, so a lookup by any one of them must find it.
+    covered: str
+    permalink: str
+    author: str
+    posted_at: datetime.datetime
+
+
+@coco.fn
+async def _declare_chunk(
+    chunk: Chunk,
+    meta: ChunkMeta,
+    id_gen: IdGenerator,
+    table: lancedb.TableTarget[SlackChunk],
+) -> None:
+    table.declare_row(
+        row=SlackChunk(
+            id=await id_gen.next_id(chunk.text),
+            kind=meta.kind,
+            channel=meta.channel,
+            source_id=meta.source_id,
+            covered=meta.covered,
+            permalink=meta.permalink,
+            author=meta.author,
+            posted_at=meta.posted_at,
+            text=chunk.text,
+            embedding=await coco.use_context(EMBEDDER).embed(chunk.text),
+        ),
+    )
+
+
+async def declare_chunks(
+    text: str,
+    meta: ChunkMeta,
+    table: lancedb.TableTarget[SlackChunk],
+) -> None:
+    chunks = _splitter.split(
+        text,
+        chunk_size=CHUNK_SIZE,
+        chunk_overlap=CHUNK_OVERLAP,
+        language="markdown",
+    )
+    id_gen = IdGenerator()
+    await coco.map(_declare_chunk, chunks, meta, id_gen, table)