> ## Documentation Index
> Fetch the complete documentation index at: https://docs.causeloop.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Sources

> Tenant-facing source connector management: registering connectors, per-source upload, staff-seeded source kinds, and how sources feed ingest and training.

Sources are how a tenant gets its own data into Causeloop. `services/api/sources_api.py` owns registering connectors, testing/syncing them, and per-source file upload; every mutating route requires `sources.manage` or `ingest.run` (`require_permission`), and credentials are write-only — encrypted immediately on submission (`services/api/secrets.py`) and never returned in any response afterward, only a last-4-style `credential_hint` persists for display.

## Connector kinds

Four connector kinds are registered in `services/ingest/connectors/registry.py`'s `CONNECTOR_REGISTRY`, each implementing the same `test_connection()` / `pull()` / `pull_file()` contract (`services/ingest/connectors/base.py`) so the ingestion pipeline never needs to know which kind it's talking to:

| Kind          | Connector class       | Config highlights                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
| ------------- | --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `file_upload` | `FileUploadConnector` | No remote endpoint — `pull_file()` parses whatever bytes were uploaded (xlsx/csv/json/jsonl). Content-hash checkpoint, so re-uploading the same file is a safe no-op.                                                                                                                                                                                                                                                                                                    |
| `s3`          | `S3Connector`         | `bucket`, `prefix`, `endpoint_url` (set for MinIO/non-AWS S3-compatible endpoints), `region`, `file_pattern` glob, `batch_size`. Checkpoint is the newest processed object's `LastModified`.                                                                                                                                                                                                                                                                             |
| `database`    | `DatabaseConnector`   | `dialect` (`postgres`/`mysql`), `host`, `port`, `database`, `table`, `id_column`, `text_column`, `timestamp_column`, optional custom `query`. Read-only is enforced per dialect: Postgres via `has_table_privilege()`, MySQL via a real `INSERT` inside a transaction that is always rolled back (MySQL checks privileges before constraints, so a `1142` access-denied error means no write privilege; any other outcome, including success, means the role can write). |
| `kafka`       | `KafkaConnector`      | `bootstrap_servers`, `topic`, `consumer_group` (named `causeloop.<tenant>.<source>` by convention), `security_protocol`/`sasl_mechanism`, `id_field`/`text_field`/`timestamp_field`. Offsets are the checkpoint — commits only after a batch is durably turned into records, so a crash between poll and commit re-delivers rather than drops.                                                                                                                           |

`GET /sources/connectors` returns `{kind: schema}` for every registered kind, each schema being that connector's Pydantic `config_schema.model_json_schema()` — the intent (per the module's own docstring) is for a source-creation form to be generated directly from this schema rather than hand-built per kind.

<Note>
  `GET /sources/connectors`, `POST /sources/{source_id}/test-connection`, `POST /sources/{source_id}/sync`, `GET /sources/{source_id}/history`, and `GET /sources/jobs/{job_id}/events` are all implemented in `services/api/sources_api.py`, but `services/api/product_app.py`'s `SOURCE_OPERATIONS` allowlist — the actual deployed product surface — admits only `GET /sources`, `POST /sources`, and `POST /sources/{source_id}/upload`. The shipped console (`SourcesClient.tsx`) matches the allowlist exactly: it lists, creates, and uploads to sources, and never calls connector discovery, test-connection, sync, history, or the job event stream. Those remain router-level capabilities for a future connector-management UI, not stubs returning errors — they simply aren't wired to any deployed route or frontend surface today.
</Note>

## Source health

`Source.health_status` (`services/api/db/models.py`) defaults to `"unknown"` and is otherwise only ever written to `"healthy"` or `"error"` — by `test_source_connection()` (from the connector's own `test_connection()` result) or by `sync_source()` (`"healthy"` on a clean sync, `"error"` with the exception text in `health_detail` on failure). The frontend's `sourceStatusTone()` (`SourcesClient.tsx`) also maps a `"syncing"` status to an amber chip defensively, but nothing in the current backend ever sets that value — a source card in the console today only ever shows `unknown` (neutral), `healthy` (green), or `error` (red).

## Permissions summary

| Route                                                                        | Permission                    | CSRF     |
| ---------------------------------------------------------------------------- | ----------------------------- | -------- |
| `GET /sources`                                                               | any authenticated tenant user | —        |
| `POST /sources`                                                              | `sources.manage`              | required |
| `POST /sources/{id}/upload`                                                  | `ingest.run`                  | required |
| `POST /sources/{id}/test-connection` *(router-only)*                         | `sources.manage`              | required |
| `POST /sources/{id}/sync` *(router-only)*                                    | `ingest.run`                  | required |
| `GET /sources/{id}/history`, `GET /sources/jobs/{id}/events` *(router-only)* | any authenticated tenant user | —        |

## Registering and using a source

```mermaid theme={null}
flowchart LR
    A["GET /sources"] --> B["POST /sources<br/>kind + name + config (+ credentials)"]
    B --> C["POST /sources/{id}/upload<br/>(file_upload kind only)"]
    C --> D["Dataset + DatasetRecord rows<br/>(dedup by content_hash)"]
    D --> E["materialize_insights job enqueued"]
```

### `GET /sources`

Returns every `Source` row for the tenant as a `SourceSummary`: `id`, `kind`, `name`, `config`, `credential_hint`, `health_status`, `health_detail`, `last_synced_at`. No credentials, ever.

### `POST /sources`

`CreateSourceRequest` is `{kind, name, config, credentials?}`. The handler validates `kind` against `CONNECTOR_REGISTRY` (422 if unknown) and validates `config` against that kind's `config_schema` (422 with the Pydantic validation error if it doesn't fit). If `credentials` is supplied, it's JSON-encoded and passed through `encrypt_secret()`; the stored `credential_hint` is derived from `password`, then `secret_access_key`, then whatever the first credential value is — never the raw secret. Requires `sources.manage` + CSRF.

### `POST /sources/{source_id}/upload`

The route the shipped console's "Upload and ingest" button calls. `file_upload`-kind sources have no remote endpoint to sync, so this pushes the uploaded bytes through the same connector interface (`FileUploadConnector.pull_file()`) and the same shared pipeline write path every other connector's sync uses — producing a real dataset version and `ingest_jobs` row exactly like a scheduled `s3`/`database`/`kafka` sync would. Returns 409 if the target source isn't `kind == "file_upload"`, 413 if the upload exceeds the object-store size limit, 422 if the file fails to parse. Requires `ingest.run` + CSRF.

The frontend's `SourcesClient.tsx` doesn't make the user create a source first: `handleUpload()` looks for an existing `file_upload`-kind source and creates one named "File uploads" on the fly if none exists, then uploads to it.

### Router-level (not deployed): connector discovery, test-connection, sync, history, job events

For completeness — these exist in `services/api/sources_api.py` and are exercised by tests, but are not part of the current product allowlist:

* **`GET /sources/connectors`** — the schema-driven form data described above.
* **`POST /sources/{source_id}/test-connection`** — calls the connector's `test_connection()` and writes the result back to `health_status`/`health_detail`.
* **`POST /sources/{source_id}/sync`** — pulls new records since the source's last checkpoint (any non-`file_upload` kind) through the same unified pipeline as upload.
* **`GET /sources/{source_id}/history`** — every `IngestJob` for that source, newest first (`status`, `stage`, `progress`, `dataset_id`, `error_detail`).
* **`GET /sources/jobs/{job_id}/events`** — a Server-Sent Events stream that polls the durable `ingest_jobs` row (roughly once per second, up to \~5 minutes) and pushes one `data:` line per change, closing once the job reaches a terminal status (`succeeded`/`failed`/`cancelled`). Polling the durable Postgres record deliberately avoids standing up a second pub/sub authority just for this.

## Seeded source kinds (staff onboarding)

`POST /admin/tenants/{tenant_id}/sources/seed` (`services/api/onboarding_api.py`, staff/employee-authenticated, part of `ONBOARDING_OPERATIONS`) pre-registers placeholder `Source` rows — kind plus a default name, no credentials — for the onboarding wizard's "Sources" step, where a client's admin picks which connector kinds they expect to use on day one:

```python theme={null}
_SEED_SOURCE_NAMES = {
    "file_upload": "Historical issue register (file upload)",
    "s3": "Audit exports (S3)",
    "database": "GRC system (database)",
    "kafka": "Issue events (Kafka)",
}
```

Requesting an unknown kind returns 422 up front so a typo doesn't silently create nothing. This only saves the client's admin from re-picking the kind and typing a name for each source they already told onboarding they wanted — actual source *registration* (the client-facing add-source flow, credentials included) is exactly `POST /sources` above; there is no second source store. See [Client onboarding pipeline](/features/client-onboarding-pipeline) for where this step sits in the overall onboarding flow.

## How sources feed ingest and training

Every successful upload or sync writes a new `Dataset` row (`tenant_id`, `source_id`, `version` — versions are strictly increasing per tenant, locked via `lock_tenant_dataset_versions()` to avoid a race between concurrent syncs) and a batch of `DatasetRecord` rows, deduplicated on `UNIQUE(tenant_id, dataset_id, content_hash)` so re-ingesting the same content is a no-op rather than a duplicate. The write path then atomically enqueues a `materialize_insights` platform job for that new `(dataset_id, version)` pair — see [Insights console](/features/insights-console) for exactly what that job computes and when the resulting snapshot becomes visible to the tenant console. A source's `checkpoint` column (opaque, per-connector-kind — a content hash for `file_upload`, an S3 object timestamp, a Kafka consumer-group offset) is what lets a subsequent sync pull only what's new.

Training a model checkpoint against a dataset version is a separate, staff-driven step (`POST /admin/tenants/{tenant_id}/train`) — sources and ingest only get data *into* the tenant schema as dataset versions; they don't themselves trigger model training.

## Related

* [Insights console](/features/insights-console)
* [Client onboarding pipeline](/features/client-onboarding-pipeline)
* [Workers and jobs](/architecture/workers-and-jobs)
* [Tenancy and data model](/architecture/tenancy-and-data-model)