Skip to content

Knowledge Store: A Decoupled RAG Layer for SPOT

Design record of 2026-04-14. The design of the Knowledge Store: what was decided and why. The developer guide is docs/guides/KNOWLEDGE-STORE.md. Internal host names are replaced by placeholders. It lists no security finding.

Context

SPOT analyzers need access to company information to spot spear-phishing (org charts, employee directories, wikis, ticket systems, policy docs, past incident reports). Today analyzers operate in isolation: they only see the raw email plus whatever a per-stage context provider dumped into analysis_context. That forces three undesirable couplings:

  1. Analyzer ↔ provider name ; an analyzer that wants employee data must read ctx.provider("employee-dir"), hard-coding the plugin id.
  2. Provider ↔ consumer ; providers have to guess what shape analyzers want; every analyzer has to know which provider produced what.
  3. Data re-fetching ; company data (employees, wikis, policies) is static-ish but is re-queried per email because there's nowhere persistent to put it.

We want a Knowledge Store: a platform service where providers deposit tagged documents and analyzers fetch them on demand by semantic similarity + tag expression. Neither side knows the other exists. Embedding is centralised in the store so no plugin has to ship its own embedding model. Retrieval is lazy ; analyzers ask for exactly what they need at the moment they need it, through an SDK abstraction that hides all embedding/DB/vector mechanics.

Breaking backward compatibility is acceptable. Context providers keep their current per-email enrich hook for truly dynamic data; the Knowledge Store is the channel for all reference data.

Design principles

  • Single mediating abstraction ; the Knowledge Store is the only thing providers and analyzers share. They don't reference each other by name.
  • Embedding lives in the platform ; not in any plugin. Plugins handle content; the store handles vectors. Analyzers never see a vector and never run an embedding model.
  • Single categorisation axis: tags ; every document has a flat tags: list[str]. No separate type field ; tags does both jobs. The SDK ships a KnowledgeTag module with well-known constants (EMPLOYEE, WIKI_PAGE, POLICY, ORG_CHART, INCIDENT_REPORT, TICKET, THREAT_REPORT, SENDER_HISTORY, EXECUTIVE, FINANCE, ...) for interoperability; tags themselves are plain strings so external providers can introduce private labels (acme:jira_issue) without a platform change.
  • Tag expressions: + AND, | OR ; analyzer queries use a tiny expression language: "employee+executive" (both), "wiki_page|policy" (either), "employee+executive|director" — + binds tighter than |, no parentheses in v1 (precedence is Python-like). Empty/None means "no tag filter".
  • On-demand retrieval, not pre-fetch ; analyzers fetch what they need, when they need it, via the SDK. The orchestrator does not pre-fetch a stage-wide bundle. This avoids both "wasted retrievals" (top-K returns docs nobody uses) and the empty-bundle problem (post-filtering returns nothing useful for one analyzer's needs).
  • No god-class analyzers ; analyzers ask for documents through a high-level SDK abstraction (await email.kb.fetch(tags=..., top_k=...)). The SDK does the HTTP round-trip; the Knowledge Store does the embedding + vector search + filter. The analyzer is a pure consumer.
  • Two phases, cleanly separated:
  • Phase A ; Ingestion (out-of-band, scheduled): platform cron → provider /internal/sync → provider pulls from its source system → provider pushes tagged documents to the Knowledge Store. Not tied to per-email latency.
  • Phase B ; Consumption (per-email, on demand): analyzer calls await email.kb.fetch(...) whenever it needs context. The store embeds (cached by content-hash), runs a tag-filtered vector search, returns the documents.
  • No coupling, no names ; plugins only know the well-known platform URL (injected via env var at install time, like SPOT_CONFIG_URL already is).
  • Operator policy via stage limits ; the workflow stage may declare retrieval_limits (cap on top_k, floor on min_score). These are passed to the analyzer in the request payload; the SDK enforces them transparently. Limits are caps only ; they shrink what an analyzer asks for, never expand it.
  • Pluggable embedder ; the service depends on an Embedder protocol, not on Ollama directly. Default implementation uses Ollama; the protocol supports batch embedding so ingestion can scale. Future backends (sentence-transformers local, OpenAI/Cohere API) are a drop-in swap.
  • Chunk long content ; ingestion-time chunking is a first-class concern, not an afterthought. Long wikis / policies / reports are split into chunks, each stored as its own document with a parent_id in metadata. Embeddings are better per-chunk; retrieval is finer-grained.
  • Documents can expire ; expires_at is a first-class field, with a periodic cleanup job that deletes stale docs. Freshness is non-negotiable for phishing detection (employees leave, roles change).
  • Security is lightweight but present ; write endpoints require the existing SPOT_INTERNAL_API_KEY; the source field on every document records who wrote it (future: enforce ownership on overwrite). Query is inside the SPOT network.
  • Observability is baked in, not bolted on ; structured query/embed logs, Prometheus metrics for latency / cache hit rate / ingestion volume, slow-query ring buffer for debugging.

Architecture

                    ┌──────────────────────────────────────┐
                    │  core/services/knowledge             │
                    │  (new in-core microservice ;         │
                    │   same layout as api-gateway,        │
                    │   analyzer-orchestrator, etc.)       │
                    │                                      │
                    │  POST /api/v1/knowledge/upsert       │
                    │  POST /api/v1/knowledge/query        │
                    │  DELETE /api/v1/knowledge/{id}       │
                    │                                      │
                    │  uses shared Postgres (new table     │
                    │  knowledge_documents + vector ext)   │
                    │  uses shared Redis (cache)           │
                    │  uses existing Ollama (embeddings)   │
                    └──────────────────────────────────────┘
                          ▲                    │
                    upsert│                    │query
                          │                    ▼
   ┌────────────────────────┐         ┌────────────────────────────┐
   │  Context Provider      │         │  Analyzer                  │
   │  (e.g. provider-ldap)  │         │  (analyzer-llm, -rules)    │
   │                        │         │                            │
   │  POST /internal/sync   │         │  POST /internal/analyze    │
   │  pulls LDAP → upserts  │         │   gets email + retrieval   │
   │  tagged documents      │         │   limits in payload;       │
   │  (tags=["employee",    │         │   on demand calls          │
   │   "executive"])        │         │   await email.kb.fetch(    │
   │                        │         │     tags="employee+exec",  │
   │  bulk_upsert via       │         │     top_k=3)               │
   │  KnowledgeClient       │         │  (SDK does HTTP; store     │
   │                        │         │   does embed + search)     │
   └────────────────────────┘         └────────────────────────────┘
              ▲
              │triggered by
              │
   ┌──────────┴─────────────┐
   │  Platform cron         │
   │  (api-gateway job)     │
   │  reads sync_schedule   │
   │  from spot.yaml        │
   └────────────────────────┘

Answers to the four questions

1. How to structure the retrieved information?

Two things in the SDK:

KnowledgeDocument ; the Pydantic model:

class KnowledgeDocument(BaseModel):
    id: str                         # stable unique id (e.g. "employee:alice@co.com"
                                    # or "wiki:page-123#chunk-4" for chunks)
    content: str                    # text payload an LLM or analyzer reads
                                    # (should be chunk-sized, not a whole doc)
    tags: list[str] = []            # the only categorisation axis; AND/OR
                                    # filterable at query time
    metadata: dict[str, Any] = {}   # structured fields (role, email, url, ...)
                                    # reserved key: "parent_id" for chunks
    source: str = ""                # which provider produced it (audit + future
                                    # ownership check on overwrite)
    updated_at: datetime | None = None
    expires_at: datetime | None = None   # optional TTL; cleanup job deletes
                                         # rows where expires_at < now()
    score: float | None = None      # populated on query, 0.0 – 1.0

There is no separate type field. tags does both the "what kind" and "what subset" jobs. A document is described by its full tag set; a query selects via a tag expression.

Stable IDs & deduplication. Providers MUST use stable, deterministic IDs (e.g. employee:<email>, wiki:<page_id>#chunk-<n>). Upsert is idempotent ; re-syncing the same data replaces rather than duplicates. Computing id from a content hash is also acceptable when no natural key exists.

Chunking convention (for long content). When a provider ingests a wiki page or policy document, it splits the content into chunks of ~1500–2000 chars (with ~200 char overlap) and upserts each chunk as a separate document:

for i, chunk in enumerate(chunk_text(page.body, max_chars=2000, overlap=200)):
    docs.append(KnowledgeDocument(
        id=f"wiki:{page.id}#chunk-{i}",
        content=chunk,
        tags=[KnowledgeTag.WIKI_PAGE, *page.labels],
        metadata={"parent_id": f"wiki:{page.id}", "title": page.title, "chunk_index": i},
        source="provider-wiki",
    ))

The SDK ships spot_sdk.knowledge.chunk_text(text, max_chars, overlap) so providers don't each reinvent chunking.

KnowledgeTag ; well-known shared-vocabulary constants:

# spot_sdk/knowledge_tags.py
class KnowledgeTag:
    # "kind" tags
    EMPLOYEE         = "employee"
    WIKI_PAGE        = "wiki_page"
    POLICY           = "policy"
    ORG_CHART        = "org_chart"
    INCIDENT_REPORT  = "incident_report"
    TICKET           = "ticket"
    THREAT_REPORT    = "threat_report"
    SENDER_HISTORY   = "sender_history"
    # "subset" tags
    EXECUTIVE        = "executive"
    FINANCE          = "finance"
    ENGINEERING      = "engineering"
    HR               = "hr"
    PUBLIC           = "public"
    INTERNAL         = "internal"

Tags themselves are plain strs so external developers can introduce custom tags (e.g. "acme:jira_issue", "acme:vendor_contract") without changing the SDK. The constants simply encode the recommended shared vocabulary. metadata is for structured fields the analyzer might want to inspect (e.g. metadata["email"] on an employee doc). source is audit-only.

Tag expressions ; analyzers and the API filter on tag expressions:

Expression Semantics
"employee" has tag employee
"employee+executive" has BOTH employee AND executive
"wiki_page\|policy" has EITHER wiki_page OR policy
"employee+executive\|director" (employee AND executive) OR director (AND binds tighter than OR)
"" / None no tag filter

Precedence: + binds tighter than |, no parentheses in v1. Trivial parser, trivial SQL: split on | to get OR groups, split each group on + to get AND terms, translate to (tags @> ARRAY[...]) OR (tags @> ARRAY[...]).

2. How to store the retrieved information?

A new knowledge_documents table in the existing SPOT Postgres database ; same connection pool, same backup, same Alembic migration chain. Only the vector extension is new.

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE knowledge_documents (
    id          TEXT PRIMARY KEY,
    content     TEXT NOT NULL,
    tags        TEXT[] NOT NULL DEFAULT '{}',
    metadata    JSONB NOT NULL DEFAULT '{}',
    source      TEXT NOT NULL DEFAULT '',
    embedding   vector(1024) NOT NULL,     -- bge-m3 dimension
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    expires_at  TIMESTAMPTZ NULL
);

CREATE INDEX ON knowledge_documents USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON knowledge_documents USING gin (tags);
CREATE INDEX ON knowledge_documents (expires_at) WHERE expires_at IS NOT NULL;

The GIN index on tags makes both AND (@> containment) and OR (&& overlap) queries fast. The partial index on expires_at makes the periodic cleanup scan tight.

Migration added under core/shared/database/alembic/versions/ alongside the existing ones.

Embedding is generated server-side at upsert time. Query time, the store embeds the incoming text once and runs a filtered kNN. Redis (the existing SPOT Redis, namespaced prefix spot:knowledge:*) caches both the embedding (by content-hash) and the query result (by full-params hash) for 5 min ; handles retries and spam-campaign near-duplicates.

3. How do context providers access the store to add to it?

Via a thin KnowledgeClient in the SDK (httpx-wrapper, same style as the existing ConfigClient at sdk/python/spot_sdk/config_client.py):

class KnowledgeClient:
    # --- write side: providers ---
    async def upsert(self, doc: KnowledgeDocument) -> None: ...
    async def bulk_upsert(self, docs: list[KnowledgeDocument]) -> None: ...
    async def delete(self, doc_id: str) -> None: ...

    # --- read side: analyzers ---
    async def fetch(
        self,
        tags: str | None = None,            # tag expression: "a+b|c"
        text: str | None = None,            # if None, uses the analysis email body
        top_k: int = 5,
        min_score: float = 0.0,
        strategy: str = "similarity",       # "similarity" | "diverse"
    ) -> list[KnowledgeDocument]:
        """Semantic vector search with tag-expression filter.
        The store embeds the text once (cached by content hash),
        runs the tag filter, returns top-k by similarity (or
        re-ranked for diversity if strategy='diverse').
        retrieval_limits from the workflow stage are enforced
        transparently here ; top_k is capped, min_score is floored."""

The SDK also ships: - spot_sdk.knowledge.chunk_text(text, max_chars, overlap) for providers - KnowledgeClient.for_analysis(email) factory used by analyzers (reads SPOT_KNOWLEDGE_URL from env, picks up retrieval_limits from the email payload) - FakeKnowledgeClient for plugin unit tests

The URL is injected as an env var at install time. In services/api-gateway/src/services/installer.py:168-185 the installer already sets SPOT_CONFIG_URL; we add one more line:

env["SPOT_KNOWLEDGE_URL"] = "http://knowledge:8000/api/v1/knowledge"

(Hostname is the compose service name of the new in-core microservice.)

Provider code becomes:

# provider-employee-dir/src/app.py
from spot_sdk.knowledge import KnowledgeClient, KnowledgeDocument, KnowledgeTag
kb = KnowledgeClient(url=os.environ["SPOT_KNOWLEDGE_URL"])

@app.post("/internal/sync")
async def sync() -> dict[str, int]:
    employees = fetch_from_ldap()
    docs = [
        KnowledgeDocument(
            id=f"employee:{e.email}",
            content=f"{e.name}, {e.title}, {e.department}. Email: {e.email}.",
            tags=[
                KnowledgeTag.EMPLOYEE,
                *([KnowledgeTag.EXECUTIVE] if e.is_exec else []),
                e.department.lower(),  # e.g. "finance"
            ],
            metadata={"email": e.email, "title": e.title, "dept": e.department},
            source="provider-employee-dir",
        )
        for e in employees
    ]
    await kb.bulk_upsert(docs)
    return {"upserted": len(docs)}

The provider doesn't know about pgvector, embeddings, or any analyzer.

4. How do analyzers access the store to fetch company information?

On demand, through the SDK's KnowledgeClient.fetch() ; the analyzer asks for exactly what it needs, when it needs it. The orchestrator constructs a per-analysis client by passing SPOT_KNOWLEDGE_URL (env) and retrieval_limits (request payload) to the analyzer; the SDK factory KnowledgeClient.for_analysis(email) wraps the construction so analyzer code stays one-line clean.

# analyzer-llm ; LLM-based analyzer
from spot_sdk.knowledge import KnowledgeClient

@app.post("/internal/analyze")
async def analyze(email: Email) -> AnalysisResult:
    kb = KnowledgeClient.for_analysis(email)
    employees = await kb.fetch(tags="employee", top_k=3)
    policies  = await kb.fetch(tags="policy",   top_k=2)
    wikis     = await kb.fetch(tags="wiki_page", top_k=5)
    prompt = build_prompt(email, employees, wikis, policies)
    # Each retrieved doc can be folded into AnalysisIndicator.evidence
    # so the analyst UI can explain "why".
    return llm.classify(prompt)

# analyzer-rules ; deterministic, no LLM
@app.post("/internal/analyze")
async def analyze(email: Email) -> AnalysisResult:
    kb = KnowledgeClient.for_analysis(email)
    sender = email.headers.sender
    # "Does this email claim to be from an executive or director?"
    hits = await kb.fetch(
        tags="employee+executive|director",
        text=sender,
        top_k=1,
    )
    if hits and hits[0].metadata.get("email") != sender:
        return AnalysisResult(is_phishing=True, confidence=0.9, ...)
    return AnalysisResult(is_phishing=False, confidence=0.6, ...)

Every analyzer uses the same fetch() method with the same tag expression syntax. Neither the analyzer nor the provider knows about the other ; the shared tag vocabulary is their only contract.

Workflow-level retrieval limits. The operator can declare per-stage caps in spot.yaml:

stages:
  - name: enrichment
    retrieval_limits:        # optional; per-stage operator policy
      max_top_k: 10          # cap top_k requested by any analyzer here
      min_score_floor: 0.3   # raise the floor even if analyzer asked
                             # for a more permissive value
    analyzers:
      - id: analyzer-llm
      - id: analyzer-rules

Mechanics: - The orchestrator includes retrieval_limits in the request body to the analyzer (alongside the email and analysis_context). - KnowledgeClient.for_analysis(email) reads them and applies them transparently when the analyzer calls fetch(). If the analyzer asks for top_k=20 but max_top_k=10, it gets 10. If min_score=0.0 but min_score_floor=0.3, it gets 0.3. - Caps only ; never raises requested values. Limits absent → analyzer values used as-is.

This gives operators a cost/policy lever without coupling them to the analyzer's internal retrieval logic.

Components (files to create / modify)

New in-core microservice: core/services/knowledge/

New service directory in the core repo, same layout as the existing api-gateway, analyzer-orchestrator, mail-orchestrator.

  • Dockerfile ; inherits from the shared base image (services/shared/Dockerfile.base) like the other services
  • pyproject.toml ; adds fastapi, pgvector (SQLAlchemy integration), httpx
  • src/main.py ; FastAPI app entrypoint
  • src/api/knowledge.py ; endpoints: POST /upsert, POST /bulk-upsert, POST /query (the analyzer's fetch calls this), DELETE /{id}, plus /health, /version, /metrics. Internal API key middleware (SPOT_INTERNAL_API_KEY) required on write endpoints; /query is open to the in-network namespace. Structured logging on every call: logger.info("query", text_hash=..., tags=..., top_k=..., latency_ms=..., hits=N, cache_hit=bool).
  • src/service/store.py ; KnowledgeStore class wrapping Postgres+pgvector, using the shared session factory from shared/database/session.py. Tag-expression-aware vector search.
  • src/service/tag_expr.py ; tiny parser/translator for the + / | tag mini-language. ~20 lines: split on | for OR groups, split each on + for AND terms, translate to (tags @> ARRAY[...]) OR (tags @> ARRAY[...]). Validates input (alphanumeric + _-:) to prevent injection.
  • src/service/embedder.py ; Embedder protocol + concrete OllamaEmbedder default implementation. Protocol requires embed(text) -> list[float] and embed_batch(texts) -> list[...]. Bulk upserts go through embed_batch so ingestion isn't N round-trips. Content-hash Redis cache for per-text embeddings.
  • src/service/cache.py ; Redis wrapper for embedding + query caches, uses the existing SPOT Redis with prefix spot:knowledge:*.
  • src/service/cleanup.py ; background task that runs every hour and executes DELETE FROM knowledge_documents WHERE expires_at < now() using the partial expires_at index.
  • src/service/metrics.py ; Prometheus metrics:
  • knowledge_query_duration_seconds (histogram, labels: cache_hit, has_tag_filter)
  • knowledge_embed_duration_seconds (histogram, labels: backend)
  • knowledge_embed_cache_hits_total / _misses_total
  • knowledge_upserts_total (counter, labels: source)
  • knowledge_docs_total (gauge) ; from periodic COUNT
  • knowledge_cleanup_deleted_total (counter)
  • src/service_config.py ; settings (DATABASE_URL, REDIS_URL, OLLAMA_URL, EMBEDDING_MODEL default bge-m3, EMBEDDING_BATCH_SIZE default 16, CLEANUP_INTERVAL_SECONDS default 3600, INTERNAL_API_KEY) following the pattern of analyzer-orchestrator/src/service_config.py.
  • tests/ ; unit tests for store, tag_expr parser, embedder, cache, cleanup, API.

New shared DB model + migration

  • core/shared/database/models.py ; add KnowledgeDocument ORM model alongside the existing User, Email, AnalysisResult models.
  • core/shared/database/alembic/versions/NNN_knowledge_documents.py — Alembic migration enabling CREATE EXTENSION vector and creating the table + indexes.

SDK changes (sdk/sdk/python/spot_sdk/)

  • New: knowledge.py
  • KnowledgeDocument Pydantic model (no type; with tags, expires_at)
  • KnowledgeClient:
    • write side (providers): upsert, bulk_upsert, delete
    • read side (analyzers): fetch(tags, text, top_k, min_score, strategy)
    • constructor: KnowledgeClient(url, api_key=None, retrieval_limits=None)
    • factory: KnowledgeClient.for_analysis(email) reads SPOT_KNOWLEDGE_URL from env and email.retrieval_limits from the orchestrator-passed payload
  • chunk_text(text, max_chars=2000, overlap=200) -> list[str] helper so every provider chunks consistently
  • Optional KnowledgeCache in-process LRU for hot queries inside one analyzer process
  • New: knowledge_tags.py ; the KnowledgeTag constants (EMPLOYEE, WIKI_PAGE, POLICY, ORG_CHART, INCIDENT_REPORT, TICKET, THREAT_REPORT, SENDER_HISTORY, EXECUTIVE, FINANCE, ENGINEERING, HR, PUBLIC, INTERNAL)
  • Modified: email.py ; add an optional retrieval_limits: RetrievalLimits | None field on Email that the orchestrator populates on the per-analysis HTTP request
  • Modified: workflow.py ; add RetrievalLimits model and a retrieval_limits: RetrievalLimits | None field on WorkflowStage
  • Modified: __init__.py ; export KnowledgeDocument, KnowledgeClient, KnowledgeTag, chunk_text, RetrievalLimits
  • New: testing/fake_knowledge_client.py ; in-memory implementation for plugin unit tests
  • Bump: pyproject.toml → 2.20.0

Core platform changes

  • Modify core/services/api-gateway/src/services/installer.py:168-185 ; add SPOT_KNOWLEDGE_URL and SPOT_INTERNAL_API_KEY to every plugin container's env when it's installed. Include them as SPOT-managed env vars alongside SPOT_CONFIG_URL, SPOT_PLUGIN_ID, etc.

  • Modify core/services/analyzer-orchestrator/src/engine/executor.py ; when invoking an analyzer, include the stage's retrieval_limits in the request payload alongside the email and analysis_context. No pre-fetching of documents ; the analyzer fetches on demand. Mechanically: read WorkflowStage.retrieval_limits, attach to the analyzer-call payload as retrieval_limits: {max_top_k, min_score_floor}. Roughly 5 lines of code.

  • Modify core/shared/config/workflows.py (or wherever WorkflowStage is defined in core) ; accept the new retrieval_limits field that mirrors the SDK model.

  • New: core/services/api-gateway/src/services/knowledge_scheduler.py ; an asyncio background task that reads sync schedules from spot.yaml and calls each provider's /internal/sync on cadence. Started in lifecycle.py next to the existing service initializers (around line 196). Honours skip_if_running and exposes last-run metadata for the dashboard.

Isolation: the scheduler is a self-contained module with a tight interface (start(), stop(), trigger_sync(provider_id), status()). Its state (last-run timestamps, last-count, last-error) lives in Redis under spot:knowledge-scheduler:*, not in memory. This means the scheduler can be extracted into its own service later (e.g. a scheduler microservice) without any data migration or cross-module refactor ; it's already behind an interface and its state is already externalised.

  • Modify core/shared/config/settings.py (the ContextProviderSettings model) ; add optional sync_schedule: str | None (cron-style) and sync_timeout_ms: int = 60000 so operators can declare per-provider ingestion cadence in spot.yaml:
plugins:
  context_providers:
    employee-dir:
      enabled: true
      url: http://provider-employee-dir:8000
      sync_schedule: "0 */6 * * *"   # every 6 hours
      sync_timeout_ms: 120000
  • New: core/services/api-gateway/src/api/knowledge_admin.py ; admin endpoints for on-demand sync / inspection:
  • POST /api/v1/plugins/context_provider/{id}/sync ; trigger sync()
  • GET /api/v1/plugins/context_provider/{id}/sync ; last-run info

  • GET /api/v1/knowledge/stats ; total doc count + counts per common tag (proxied from the knowledge service)

  • Modify core/docker-compose.yml ; add a knowledge service on the spot-network with depends_on=[postgres, redis, ollama]. It connects to the same Postgres DB as the rest of SPOT (no new DB). Health check, restart policy, and YAML anchor usage follow the existing in-core service pattern.

  • Modify core/services/api-gateway/src/rbac.py ; add RBAC for the new admin endpoints.

Security model

  • Write endpoints (POST /upsert, POST /bulk-upsert, DELETE /{id}) on the knowledge service require the existing SPOT_INTERNAL_API_KEY header. Plugins receive this key the same way they receive SPOT_CONFIG_URL ; via env var injected by the installer. External callers cannot write.
  • Query endpoint (POST /query) is network-internal only — no auth required, the SPOT private network is the trust boundary. (If multi-tenancy lands later, query gets scoped by tenant claim.)
  • Source ownership ; every document records its source at upsert time. Today this is informational; a follow-up can enforce "a write may only overwrite docs whose existing source matches". The schema already supports it.

Observability

Metrics, logs, and a debug surface that scale beyond the first day:

  • Prometheus /metrics on the knowledge service (see metrics list above). Scraped by the existing SPOT Prometheus job.
  • Structured logs on every API call: query and upsert log text-hash, tag expression, top_k, latency, hit count, cache_hit. Failures log the exception with the same correlation context.
  • Slow-query buffer: in-memory ring buffer (last 50) of queries > 500 ms, exposed at GET /debug/slow-queries (admin only) for live troubleshooting.
  • Per-tag doc counts exposed at GET /api/v1/knowledge/stats — total + counts for the well-known tags. Surfaced on the Web dashboard so operators can see at a glance whether a provider has stopped syncing.

Web dashboard changes

  • Small additions to /plugins detail page for a context-provider kind: show sync_schedule, last_sync_at, last_sync_count, with a "Sync now" button → POST /api/v1/plugins/context_provider/{id}/sync.
  • No new pages or tabs ; the existing Plugins UI is enough.

Example reference plugin (separate follow-up repo)

provider-employee-dir as a reference implementation showing the ingestion pattern: implements /internal/sync that fetches from LDAP and upserts employee-tagged documents. Lives outside this branch.

Documentation

  • core/docs/guides/KNOWLEDGE-STORE.md ; developer guide:
  • What the store is, the tag vocabulary and tag-expression syntax
  • How to implement an ingestion provider (sync pattern, chunking)
  • How to query the store from an analyzer (KnowledgeClient.for_analysis(email) + fetch(tags=...))
  • How retrieval_limits work and when to use them
  • Recommended embedding model + performance notes
  • sdk/sdk/python/README.md ; document KnowledgeClient, KnowledgeDocument, KnowledgeTag, tag-expression syntax, for_analysis factory
  • Project CLAUDE.md ; one-paragraph update on the new store, the two phases (ingestion + on-demand consumption), and the tag vocabulary

Implementation sequence (suggested order)

  1. [DONE] SDK ; KnowledgeDocument (no type, with tags and expires_at), KnowledgeTag constants, KnowledgeClient (upsert, bulk_upsert, delete, fetch, for_analysis(email) factory), chunk_text helper, RetrievalLimits model, FakeKnowledgeClient. Bump SDK to 2.20.0. Ship wheel. Committed: sdk@8047ca9 ; 38 tests pass, wheel installed in devtools.
  2. [DONE] Shared DB model + Alembic migration ; add KnowledgeDocument ORM model + migration enabling vector, creating the table with the expires_at column and all indexes (HNSW on embedding, GIN on tags, partial expires_at). Committed: core@0c12e49 ; migration applies, pgvector image pinned, timescaledb dropped (unused). Compose postgres now uses pgvector/pgvector:pg15.
  3. [DONE] core/services/knowledge ; the in-core microservice: KnowledgeStore, tag_expr parser, Embedder protocol + OllamaEmbedder (with batch + cache), cleanup task, metrics, structured logging, internal-API-key middleware, API + tests. Committed: core@a9c67d5 ; 27 tests pass (tag-expr unit tests, store integration against real pgvector, API through httpx.ASGITransport).
  4. [DONE] Compose wiring ; add knowledge service to core/docker-compose.yml. Committed: core@02e454f ; compose config validates, knowledge builds via the shared Dockerfile with SERVICE_NAME=knowledge and depends on migration+redis. .env.example documents the new SPOT_INTERNAL_API_KEY variable.
  5. [DONE] Installer env vars ; SPOT_KNOWLEDGE_URL and SPOT_INTERNAL_API_KEY on every installed plugin. Committed: core@4dba802 ; InstallerService stamps both on the plugin env. Knowledge router mounted under /api/v1/knowledge. Migration short-circuits on SQLite so api-gateway unit tests stay green (239 passing).
  6. [DONE] Orchestrator change ; pass retrieval_limits from the workflow stage into the analyzer's request payload (no pre-fetching). Committed: core@a3a15f5 ; engine computes the caps dict per-stage from WorkflowStage.retrieval_limits and threads it through the parallel/sequential/conditional executor paths. 33 tests pass.
  7. [DONE] Scheduler ; knowledge_scheduler.py in api-gateway + sync_schedule field in ContextProviderSettings. Redis-backed state for future extraction. Committed: core@ec36c14 ; 5-field cron parser with @hourly/@daily aliases, one task per provider, Redis-backed last-run state. 17 new unit tests cover cron math + trigger I/O; all 256 api-gateway tests still pass.
  8. [DONE] Admin endpoints ; manual sync + last-run info + /stats proxy. Committed: core@501a7cc ; three endpoints (POST/GET sync by provider id, GET /api/v1/knowledge/stats proxy). RBAC rules added. 6 new tests, 262 api-gateway tests passing.
  9. [DONE] Dashboard ; show sync info + per-tag doc counts on the context provider detail page. Committed: web-dashboard@400fdcb ; detail page shows sync schedule + last-run state + "Sync now" button; listing page adds a Knowledge docs summary card. All reads best-effort.
  10. Reference provider ; provider-employee-dir in a separate repo, demonstrating both bulk_upsert and the chunking pattern. Deferred ; ships later as a separate follow-up repo.
  11. [DONE] Docs. Committed: core@bb2920a (docs/guides/KNOWLEDGE-STORE.md developer guide), sdk@872fb64 (README section), CLAUDE.md updated in-place.

Verification

End-to-end smoke test:

  1. Start the stack: spot-cli service start --dev.
  2. Verify the knowledge service comes up healthy: docker exec spot-api-gateway /app/.venv/bin/python -c 'import httpx; print(httpx.get("http://knowledge:8000/health").json())'
  3. Insert a few documents manually (bypassing a provider) by calling POST /api/v1/knowledge/upsert from devtools: a few employee docs (tags=["employee", "executive"], tags=["employee", "finance"]), a chunked wiki_page (with parent_id in metadata on each chunk), and a policy. Verify the rows + their embeddings exist in pgvector.
  4. Test tag expressions via POST /api/v1/knowledge/query:
  5. tags="employee" returns all employee docs
  6. tags="employee+executive" returns only execs
  7. tags="wiki_page|policy" returns wiki + policy docs
  8. tags="employee+executive|director" exercises mixed precedence
  9. tags="" (or omitted) returns all by similarity
  10. Wire a fake provider in spot.yaml with sync_schedule: "* * * * *", confirm the scheduler calls /internal/sync and new documents appear. Stop the provider, set its previous docs to expires_at in the past, run the cleanup task, confirm they are deleted.
  11. Hit GET /metrics on the knowledge service and confirm knowledge_query_duration_seconds and the cache-hit counters are populated.
  12. Modify analyzer-nlp (or a throwaway analyzer) to call KnowledgeClient.for_analysis(email).fetch(tags="employee", top_k=3) during analysis; submit an email; confirm retrieved documents appear in its decision logic via logs.
  13. Add retrieval_limits: {max_top_k: 1} to the workflow stage and confirm the analyzer receives at most 1 doc even when it asked for 3.
  14. Retest: analyzers that don't call the SDK are unaffected; context-provider-based workflows continue to work as before (email.ctx.provider(...) still functions for the existing per-email enrich use case).

What is explicitly NOT in scope

  • Multi-tenancy / per-org isolation ; store is global for now.
  • Analysis-result persistence of retrieved documents ; analyzers that care about "why" should emit indicators themselves (existing mechanism).
  • Moving existing context-provider use cases into the store — context providers keep their current per-email role for dynamic data (e.g. live URL reputation).
  • An LLM-driven agent that chooses what to retrieve adaptively ; the analyzer decides what to query.
  • Replacing pgvector with Qdrant/Weaviate ; the KnowledgeStore class keeps the backend abstract enough to swap later if needed; not now.