Ingestion, Data Safety & Retrieval

Technical Architecture

End to end: how a file or event becomes governed memory, and how an agent pulls it back out. Inbound APIs (blue) accept data; the governance gate (amber) applies before storage is retrievable; outbound APIs (purple) are where agents call in for inference.

Note: retrieval ranking inside the Policy-Aware Retrieval Gateway currently falls back to keyword-overlap scoring rather than true vector cosine similarity, despite pgvector being provisioned — see the Retrieval section for detail.

Capabilities

What's actually implemented today, in plain terms:

Inbound APIs — data going into KrisVen

Full detail: Ingestion.

Outbound APIs — memory coming out of KrisVen

Full detail: Retrieval.

Ingestion Overview

Three ways to get data into KrisVen today. All require a Bearer token and are served under http://localhost:8000 in local dev.

CSV File Upload REAL

The only endpoint that accepts a real file today. Each row becomes one ingestion event. The file runs through the file-safety pipeline (size limit, content-type check, optional malware scan) before parsing.

curl -X POST "http://localhost:8000/v1/sources/csv/upload?tenant_id=acme&source_system=manual_upload&source_object=accounts&id_column=account_id" \
  -H "Authorization: Bearer mock-admin-token" \
  -F "file=@accounts.csv"

Response:

{
  "run_id": "...",
  "tenant_id": "acme",
  "source_system": "manual_upload",
  "rows_received": 240,
  "rows_queued": 240,
  "status": "queued"
}

Check progress:

curl "http://localhost:8000/v1/ingestion-runs/{run_id}?tenant_id=acme" \
  -H "Authorization: Bearer mock-admin-token"

Document Metadata REAL, NO OCR

curl -X POST "http://localhost:8000/v1/sources/documents/metadata" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer mock-admin-token" \
  -d '{
    "tenant_id": "acme",
    "document_id": "DOC-4471",
    "title": "Q3 Financial Statement — Acme Corp",
    "mime_type": "application/pdf",
    "size_bytes": 184320,
    "extracted_text_stub": "Q3 revenue was $12.4M, up 8% YoY...",
    "actor": {"type": "user", "id": "analyst_1"}
  }'

Single JSON Event REAL

For structured data you already have as JSON — a webhook payload, a row from your own database.

curl -X POST "http://localhost:8000/v1/sources/events" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer mock-admin-token" \
  -d '{
    "tenant_id": "acme",
    "source_system": "internal_crm",
    "source_object": "account",
    "source_record_id": "ACC-1001",
    "operation": "UPSERT",
    "payload": {"name": "Acme Corp", "industry": "Manufacturing"},
    "source_timestamp": "2026-07-14T10:00:00Z",
    "actor": {"type": "user", "id": "analyst_1"}
  }'

Connector Inventory

12 connector classes exist under krisven/connectors/. Only 3 do real work, and none has dedicated test coverage yet. The connectors folder's own base.py says it plainly: "MVP connectors are mocks/stubs. Real connectors extend these interfaces post-MVP."

Plan accordingly — if a demo or pilot needs Salesforce/SAP/Oracle/core-banking data today, it has to go through the CSV or JSON event route as an export, not a live connector.

Authentication

Every route needs Authorization: Bearer <token>. In local dev with MOCK_AUTH=true (the default), the token mock-admin-token is treated as a system/admin actor with access to every tenant. See the Mock Auth warning before exposing any of this outside local dev.

Data Safety: Keeping Unsafe Uploads Out of Memory

Any file a user uploads is untrusted until proven otherwise. krisven/security/file_safety.py implements three checks, run in order, cheapest first, stopping at the first failure:

uploaded bytes
      │
      ▼
1. validate_size        reject if over the configured byte limit
      │
      ▼
2. validate_mime_type    sniff actual content, reject if it doesn't match
      │                  what the route expects (catches a renamed .exe)
      ▼
3. scan_for_malware      ClamAV scan via clamd, if enabled
      │
      ▼
   parser (CSVConnector, etc.)

Wired into POST /v1/sources/csv/upload today. If a future route adds raw document/PDF upload, wire check_upload() in before the bytes reach a parser there too — the function is generic, not CSV-specific.

Configuration

Why malware scanning defaults to off: ClamAV needs a running daemon with a several-hundred-MB virus database — a real infra dependency, not something to force on every dev machine. Size and content-type checks still run regardless of this setting.

Running ClamAV Locally

Defined in docker-compose.yml under an opt-in profile (first boot downloads the virus database, which takes a few minutes):

docker compose --profile clamav up -d clamav

Then in your .env:

CLAMAV_ENABLED=true
CLAMAV_HOST=clamav
CLAMAV_PORT=3310

Production containers install libmagic1 via infra/docker/api.Dockerfile for content-type sniffing. On a bare dev machine without libmagic, validate_mime_type detects the missing library, logs a warning, and skips that one check rather than crashing — size limits and malware scanning still run.

Testing Safely with EICAR

Don't use a real virus to test this. Use the industry-standard EICAR test string — every antivirus engine, including ClamAV, is specifically designed to flag this exact harmless byte sequence as "malware" for testing purposes:

X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*

Upload that with ClamAV enabled and you should get a 422 FILE_SAFETY_ERROR — malware detected (Eicar-Signature). The same string backs the automated test suite (tests/security/test_file_safety.py), so detection is covered without a real malicious sample.

A Related Gap: MOCK_AUTH

What This Does Not Cover

• Structured-payload validation (e.g. a prompt-injection payload embedded in a CSV text field) — a separate concern, handled downstream by the memory compiler's governance tagging, not this pipeline.
• No virus-definition auto-update monitoring or alerting.
• No rate limiting or per-tenant upload quotas.

Retrieval: Getting Qualified Memory for LLM Inference

Every endpoint below is real, live, and covered by tests — not aspirational. A planned SDK is described at the end, clearly marked as not yet built.

All four route through the same PolicyAwareRetrievalGateway — a 21-stage pipeline enforcing tenant isolation, memory lifecycle state, sensitivity clearance, and trust thresholds. Only ACTIVE-lifecycle memory is ever retrievable.

POST /v1/governed/agent-context

curl -X POST "http://localhost:8000/v1/governed/agent-context" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer mock-admin-token" \
  -d '{
    "purpose": "credit_review",
    "query": "What prior decisions affect Acme Corp'"'"'s current loan application?",
    "subject_refs": [{"subject_type": "borrower", "subject_id": "BORR-99881"}],
    "memory_types": ["decision", "temporal", "semantic"],
    "max_tokens": 4000,
    "min_trust_score": 0.60,
    "clearance_level": "internal"
  }'

tenant_id and agent_id are not request fields — resolved server-side from your JWT, so a caller can never claim to act as a different tenant by changing the request body.

Response:

{
  "assembly_id": "...",
  "tenant_id": "acme",
  "agent_id": "credit_agent",
  "purpose": "credit_review",
  "query_hash": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
  "items": [
    {
      "memory_id": "...",
      "memory_type": "decision",
      "trust_score": 0.87,
      "content_structured": { "...": "..." },
      "subject_id": "BORR-99881",
      "sensitivity": "confidential",
      "estimated_tokens": 312,
      "is_truncated": false
    }
  ],
  "token_budget": 4000,
  "tokens_used": 1840,
  "truncated_count": 0,
  "warnings": [],
  "assembled_at": "2026-07-14T10:00:00Z",
  "policy_decision": "allow"
}

A policy denial returns 403 — no content leaks in the response.

POST /v1/governed/retrieve and /context-pack

Lower-level access to the same pipeline, for the memory list itself rather than a pre-assembled context pack.

curl -X POST "http://localhost:8000/v1/governed/retrieve" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer mock-admin-token" \
  -d '{
    "tenant_id": "acme",
    "actor": {"actor_id": "analyst_1", "tenant_id": "acme", "roles": ["credit_officer"]},
    "purpose": "credit_review",
    "query": "Acme Corp payment history",
    "subject_refs": [{"subject_type": "borrower", "subject_id": "BORR-99881"}],
    "max_results": 10,
    "minimum_confidence": 0.70
  }'

Unlike /agent-context, this takes tenant_id and actor explicitly rather than deriving purely from the JWT — useful for service-to-service calls. Response is a ContextPack: memories[], warnings[], policy_decision, total_candidates, filtered_count, and stages_run[] (which of the 21 pipeline stages actually ran).

/context-pack is the same pipeline framed for LLM injection — prefer /agent-context for new integrations since it also handles token budgeting.

Authentication

Same as ingestion — Authorization: Bearer <token>. See the Mock Auth warning before exposing these endpoints outside local dev.

Planned: KrisVenMemoryClient SDK PLANNED

# PLANNED — not yet implemented. Use the curl examples above today.
from krisven.sdk import KrisVenMemoryClient  # does not exist yet

client = KrisVenMemoryClient(tenant_id="acme", auth_token=token)

context = await client.retrieve_memory(
    purpose="credit_review",
    query="What prior decisions affect Acme Corp's current loan application?",
    subject_refs=[{"subject_type": "borrower", "subject_id": "BORR-99881"}],
    memory_types=["decision", "temporal", "semantic"],
    max_tokens=4000,
)

# Planned: closing the loop for the trust-score/conflict-detection feedback path
await client.submit_answer_audit(
    assembly_id=context.assembly_id,
    used_memory_ids=[i.memory_id for i in context.items],
)

If/when built, this would wrap /v1/governed/agent-context plus a new audit-submission endpoint that doesn't exist yet either. Until then, the curl/HTTP examples above are the actual integration path — including for agent-facing connectors like an MCP server or a Custom GPT Action, which would call these same endpoints directly.