SurfSense/plans/backend/07-upload-pipeline-kb.md

Phase 7 — File upload as a pipeline + KB-save-secondary

Part of the CI Pivot MVP. See 00-umbrella-plan.md (Phase 7) — the final backend phase. Precondition: Phase 5 (05-pipelines-model.md) live (pipelines/pipeline_runs tables, PipelineRunTrigger.UPLOAD, connector_id nullable) and Phase 6 (06-pipelines-exec.md) live (run engine, which explicitly fails connector_id IS NULL runs). No sibling ahead — this closes the backend umbrella.

Implementation note (post-rename). Phases 1–2 are SHIPPED, so the live code already uses workspace_id / Workspace. Citations below that still show search_space_id / SearchSpace are pre-rename — substitute the workspace_* equivalent and grep the new name. New code in this plan uses the canonical workspace_* names (snippets below updated). Locate code by symbol/grep, not the absolute line numbers cited here (the rename + Phases 3–6 shift them).

Objective

Close the loop on the umbrella's positioning — "User file uploads also do a pipeline entry and register a run that files were uploaded by user; only file uploads are always saved in the knowledge base" — by:

1. Uploads-as-pipeline — give every workspace a singleton "Uploads" pipeline (connector_id = NULL, save_to_kb = true), lazily created, and make each upload request register a PipelineRun(trigger=upload) so uploads show up in pipeline run history (and in the Phase-6 chat get_pipeline_runs tool) exactly like connector fetches.
2. KB-save-secondary (confirm + close) — the opt-in save_to_kb + destination_folder_id machinery for connector pipelines already shipped in Phases 5–6; this phase records the inverse invariant for uploads (uploads always save to KB) and verifies nothing in the pivot accidentally made KB-save the default for connector pipelines.

The core design fact (drives everything): uploads are NOT executed by the Phase-6 run engine. Phase 6's execute_pipeline_run fails any connector_id IS NULL run with "uploads pipeline has no fetch executor" (06 §1). The existing upload routes remain the executor (they already create Documents + dispatch ETL); Phase 7 only attaches an audit PipelineRun to that flow and ensures the Uploads pipeline row exists. No run_pipeline enqueue, no engine change.

Locked decisions (MVP)

Concept	Decision
Uploads pipeline	One singleton per workspace, connector_id = NULL, save_to_kb = true, schedule_cron = NULL (never scheduled), enabled = true, name = "Uploads". Lazily get-or-created on first upload (not at workspace creation — avoids backfilling every existing workspace).
Uniqueness / race-safety	A partial unique index ON pipelines (workspace_id) WHERE connector_id IS NULL makes "one Uploads pipeline per workspace" a DB invariant and lets get-or-create be race-safe via INSERT … ON CONFLICT DO NOTHING + re-SELECT. (Phase 5 already reserves connector_id IS NULL to mean "Uploads", so this index is exactly that semantic.) Small Phase-7 migration.
Run lifecycle	Audit record, created terminal at request time. The upload route inserts PipelineRun(trigger=upload, status=succeeded, documents_indexed=<accepted file count>, started_at=finished_at=now()). Per-file ETL outcomes continue to live on Document.status (pending→ready/failed, already Zero-synced). Rationale below.
Why not track per-file ETL into the run	The fileupload path fans out N independent per-file Celery tasks (task_dispatcher.py:44) with no batch barrier, and accurate "which docs belong to this run" needs the documents.pipeline_run_id provenance column that Phase 5 deferred (05 §3 "Out of scope (provenance)"). Building a barrier is out of MVP scope; the run is an upload event record, matching the umbrella's "register a run that files were uploaded".
Billing	No crawl billing for uploads — uploads aren't crawls. charged_micros / crawls_* stay NULL on upload runs. Existing ETL/embedding credit behaviour (if any) is unchanged and orthogonal (03c only governs web_crawl).
Folder placement	The Uploads pipeline's destination_folder_id stays NULL — it is a singleton and cannot encode per-upload folders. Folder placement remains owned by the existing upload code (fileupload → root; folder-upload → its created/declared folder). The run/pipeline is descriptive, not the folder driver (unlike connector pipelines, where the engine reads destination_folder_id).
Surfaces wired	POST /documents/fileupload and POST /documents/folder-upload (both create Documents). The desktop precheck route POST /documents/folder-mtime-check (documents_routes.py:1533) creates nothing and is not wired; folder-unlink / folder-sync-finalize are delete/sync lifecycle, also out.
Connector-pipeline KB default	Unchanged from Phase 5: save_to_kb defaults False (05 §2 server_default="false"). This phase only asserts uploads set it true; it does not flip the connector default.

Current state (cited)

The upload routes (the executors we attach to)

• POST /documents/fileupload — documents_routes.py:126-332. Permission DOCUMENTS_CREATE (:163-169). Phase 1: create Document(search_space_id, document_type=FILE, status=pending, created_by_id, …) per file, dedup by unique_identifier_hash (:217-266), commit (:278). Phase 1.5: store_document_file(...) per doc (:289-297). Phase 2: dispatcher.dispatch_file_processing(document_id=…) per file (:307-316). Returns document_ids / skipped_duplicates (:318-325). Creates no folder (root), no run record today. files_to_process (:213) = the accepted (newly-queued) set; skipped_duplicates (:214) = already-READY dupes.
• POST /documents/folder-upload — documents_routes.py:1598-1748. Permission DOCUMENTS_CREATE (:1623-1629). Creates/reuses a root Folder (:1673-1704), writes temps, then dispatches one batch task index_uploaded_folder_files_task.delay(...) (document_tasks.py:1403) which has its own notification + heartbeat (:1446-1457) — i.e. a real completion barrier exists on this path (unlike fileupload). Returns root_folder_id + file_count (:1743-1748).
• Dispatcher — task_dispatcher.py:12-52: dispatch_file_processing(...) → process_file_upload_with_document_task.delay(document_id=…), one task per file, fire-and-forget (no barrier).

Phase 5/6 hand-offs this phase consumes

• Pipeline.connector_id nullable, NULL = "non-connector pipeline (the Phase-7 Uploads pipeline)" (05 §"Pipeline → connector"; model 05 §2 :105-106). PipelineRunTrigger.UPLOAD already exists (05 §1). PipelineRun has trigger, status, documents_indexed, started_at, finished_at, charged_micros (nullable) (05 §3).
• Phase 6 engine refuses uploads: if pipeline.connector_id is None: return fail(run, "uploads pipeline has no fetch executor") (06 §1) — confirms uploads must be recorded by the route, not enqueued to run_pipeline.
• Chat tool get_pipeline_runs (06 §7) selects PipelineRun joined to Pipeline by workspace_id — so upload runs surface to the agent for free once they exist.
• De-dup guard (06 §6) only touches connector-backed pipelines (when connector_id is set); the Uploads pipeline (connector_id NULL) is untouched by it. The scheduler (06 §5 _claim_due) ignores the Uploads pipeline two ways — schedule_cron IS NOT NULL (NULL cron) and connector_id IS NOT NULL (the defense-in-depth filter 06 added at Phase 7's request). Both confirmed safe.

Document model

• Document — db.py:1348: document_type (:1352, DocumentType.FILE), unique_identifier_hash (:1365), folder_id nullable (:1389, root/unfiled), created_by_id (:1397), plus search_space_id/status. KB membership = a Document row exists; uploads always create one ⇒ uploads are always "in the KB" by construction.

Target design

1. get_or_create_uploads_pipeline(session, workspace_id, created_by_id) — app/pipelines/uploads.py

async def get_or_create_uploads_pipeline(session, *, workspace_id: int, created_by_id) -> Pipeline:
    # fast path
    existing = (await session.execute(
        select(Pipeline).where(Pipeline.workspace_id == workspace_id,
                               Pipeline.connector_id.is_(None)))).scalar_one_or_none()
    if existing:
        return existing
    # race-safe create (partial unique index ux_pipelines_uploads_per_workspace)
    stmt = (pg_insert(Pipeline.__table__)
            .values(workspace_id=workspace_id, connector_id=None, name="Uploads",
                    save_to_kb=True, enabled=True, schedule_cron=None,
                    created_by_id=created_by_id, config={})
            .on_conflict_do_nothing(index_elements=[Pipeline.workspace_id],
                                    index_where=Pipeline.connector_id.is_(None)))
    await session.execute(stmt)
    await session.flush()
    return (await session.execute(
        select(Pipeline).where(Pipeline.workspace_id == workspace_id,
                               Pipeline.connector_id.is_(None)))).scalar_one()

• Lazy (first upload), idempotent, race-safe. created_by_id = the uploading user (becomes the pipeline's creator metadata; SET NULL on member deletion per Phase 5).
• Lives in the app/pipelines/ package (created in Phase 6) so the upload route imports it without a route↔route dependency.
• Implementation gotcha — core insert bypasses ORM defaults. Pipeline.updated_at/created_at use Python-side default=/onupdate= callables (05 §2 :114-115), which a core pg_insert does not fire. The .values(...) must set created_at/updated_at explicitly (e.g. now) — or implement get-or-create as ORM-insert + flush inside a session.begin_nested() SAVEPOINT, catching IntegrityError → rollback savepoint → re-SELECT (avoids the core/ORM default mismatch entirely; the SAVEPOINT keeps the route's outer transaction intact). Either is fine; the SAVEPOINT form is closer to the ORM conventions used elsewhere.

2. Register an upload run (shared helper) — app/pipelines/uploads.py

async def record_upload_run(session, *, workspace_id, created_by_id, documents_indexed: int) -> PipelineRun:
    pipeline = await get_or_create_uploads_pipeline(session, workspace_id=workspace_id,
                                                    created_by_id=created_by_id)
    now = datetime.now(UTC)
    run = PipelineRun(pipeline_id=pipeline.id, trigger=PipelineRunTrigger.UPLOAD,
                      status=PipelineRunStatus.SUCCEEDED,
                      documents_indexed=documents_indexed, started_at=now, finished_at=now)
    session.add(run)
    return run   # caller commits (folds into the route's existing commit)

• Terminal-at-creation (locked decision): records the upload event with the accepted file count. crawls_* / charged_micros left NULL (not a crawl).
• Caller does not await any Celery task — the run is purely the route's own DB write, folded into the route's existing transaction (no extra round-trips, no new failure mode that could block the upload).

3. Wire POST /documents/fileupload

After Phase 2 dispatch (documents_routes.py:307-316), before building the response, add:

if files_to_process:                              # only when something was actually queued
    await record_upload_run(session, workspace_id=workspace_id,
                            created_by_id=str(user.id),
                            documents_indexed=len(files_to_process))
    await session.commit()

• Count = len(files_to_process) (the newly-created/re-queued docs), excluding skipped_duplicates (already-READY no-ops). An all-duplicate upload records no run (nothing ingested).
• Best-effort, non-blocking — and the try/except is mandatory, not cosmetic. The route has an outer except Exception → session.rollback() → HTTP 500 (documents_routes.py:328-332). If record_upload_run raised unguarded, it would hit that handler and 500 an upload whose documents are already committed + dispatched. So the call MUST be wrapped in its own inner try/except that swallows: on failure await session.rollback() (discards only the run insert — documents were committed at :278/:304) + logger.warning(...), then fall through to the response. A run-record write must never fail the upload. (Mirrors the store_document_file best-effort pattern at :298-303.)

4. Wire POST /documents/folder-upload

Folder-upload dispatches one batch task, so record the run at request time with the file count (documents_routes.py before the return at :1743):

await record_upload_run(session, workspace_id=workspace_id,
                        created_by_id=str(user.id), documents_indexed=len(files))
await session.commit()   # best-effort / logged, same as §3

• Uses len(files) (folder-upload doesn't pre-dedup in the route; dedup happens inside the batch task).
• Optional richer accounting (deferred): because this path has a barrier (index_uploaded_folder_files_task), a later enhancement could create the run pending here and flip it succeeded/failed from inside the task (it already manages a notification lifecycle). MVP keeps it uniform with fileupload (terminal-at-request). Flagged, not built.

5. Migration (one small file) — partial unique index

CREATE UNIQUE INDEX IF NOT EXISTS ux_pipelines_uploads_per_workspace
    ON pipelines (workspace_id) WHERE connector_id IS NULL;

• Chain after the then-current head (verify alembic heads). Enforces the singleton + powers ON CONFLICT. downgrade() drops it.
• Pre-flight (safety): if any workspace somehow already has >1 connector_id IS NULL pipeline (shouldn't, pre-Phase-7), the CREATE UNIQUE INDEX fails — the migration docstring notes a dedup step (keep the oldest, delete extras) just in case. Expectation: none exist (uploads pipelines are introduced here).

6. KB-save-secondary verification (no new code, a guard)

• Uploads pipeline hardcodes save_to_kb=True (§1) — uploads always land in the KB (a Document row), matching the umbrella invariant.
• Confirm Phase 5's connector-pipeline default is still save_to_kb=False (05 §2) — i.e. KB-save is opt-in for connectors, mandatory for uploads. Add a test asserting both defaults so a future change can't silently make connector KB-save the default.

7. Guard the generic Phase-5 CRUD against the system Uploads pipeline (the gap)

The Uploads pipeline is a system-managed row (connector_id IS NULL, auto-created by §1). Phase 5's generic /pipelines CRUD doesn't know it's special, which opens three footguns — Phase 7 closes them:

Phase-5 endpoint	Problem if left generic	Phase-7 guard
POST /pipelines	Phase 5 allows connector_id=None (05 §7 create + test). A user could hand-create a rogue/duplicate non-connector pipeline.	Reject connector_id is None with 4xx — non-connector pipelines are system-managed (auto-created on upload), not user-created. This supersedes Phase 5's "connector_id=None → allowed" create test (update it). The partial unique index (§5) is the DB backstop.
POST /pipelines/{id}/run	Manual-running the Uploads pipeline enqueues run_pipeline, which the Phase-6 engine fails ("uploads pipeline has no fetch executor") → a confusing FAILED manual run every click.	Reject connector_id IS NULL with 4xx ("uploads are triggered by uploading files, not a manual run") — before creating the run row / enqueuing.
PUT /pipelines/{id}	Setting schedule_cron on the Uploads pipeline makes the scheduler eligible-by-cron → perpetually-failing scheduled runs (engine refuses null-connector). Editing save_to_kb/destination_folder_id is silently ineffective (uploads are route-driven, not engine-driven).	Reject schedule edits (schedule_cron/schedule_timezone) on connector_id IS NULL pipelines with 4xx. (connector_id is already immutable per Phase 5.) Optionally reject all PUTs to the Uploads pipeline; minimally block the scheduling fields. Belt-and-suspenders: Phase 6's _claim_due also filters connector_id IS NOT NULL (06 §5).
DELETE /pipelines/{id}	Deleting the Uploads pipeline cascades its run history.	Allowed (MVP): it is lazily re-created on the next upload (§1); only past upload-run history is lost. Note in UI copy later; an optional hard block is a small follow-up.

Net: after Phase 7, the only non-connector pipeline that can exist is the system Uploads pipeline, it is never user-created / manually-run / scheduled, and the scheduler ignores it two ways (NULL cron and the connector_id IS NOT NULL filter).

Work items

1. app/pipelines/uploads.py: get_or_create_uploads_pipeline (race-safe via ON CONFLICT) + record_upload_run (terminal upload run).
2. Migration: partial unique index ux_pipelines_uploads_per_workspace ON pipelines(workspace_id) WHERE connector_id IS NULL; docstring dedup pre-flight note; downgrade drops it.
3. Wire fileupload (documents_routes.py:126-332): call record_upload_run(documents_indexed=len(files_to_process)) after dispatch, in an inner best-effort try/except (must not reach the route's outer 500 handler), only when files_to_process non-empty.
4. Wire folder-upload (documents_routes.py:1598-1748): call record_upload_run(documents_indexed=len(files)) before the response, same inner best-effort guard.
5. Guard the generic CRUD (§7): POST /pipelines rejects connector_id is None; /run rejects connector_id IS NULL; PUT rejects schedule edits on connector_id IS NULL. Update Phase 5's "connector_id=None → allowed" create test accordingly. (Phase 6's _claim_due connector_id IS NOT NULL filter is the scheduler backstop.)
6. Tests (below).

Tests

• First upload creates the pipeline: fileupload into a fresh workspace → exactly one connector_id IS NULL pipeline named "Uploads" with save_to_kb=True; a PipelineRun(trigger=upload, status=succeeded, documents_indexed=N).
• Subsequent uploads reuse it: a second fileupload → still one Uploads pipeline, two upload runs. Concurrency: two simultaneous first-uploads (race) → still exactly one pipeline (partial-unique + ON CONFLICT), two runs.
• Duplicate-only upload: re-uploading already-READY files (skipped_duplicates == len(files), files_to_process == []) → no run recorded.
• folder-upload: records one upload run with documents_indexed=len(files); the existing batch task/folder behaviour is unchanged.
• No crawl billing: upload runs have charged_micros IS NULL, crawls_* IS NULL; no web_crawl TokenUsage row is written.
• Engine still refuses uploads: enqueuing run_pipeline for an Uploads pipeline's run (or a NULL-connector pipeline) → Phase-6 engine fails it cleanly (regression guard for 06 §1). Confirms uploads are never engine-executed.
• Scheduler ignores uploads (two ways): the Phase-6 tick never selects the Uploads pipeline — both because schedule_cron IS NULL and because _claim_due filters connector_id IS NOT NULL. Add a case: even with a cron forced onto a NULL-connector row (direct DB), _claim_due still skips it.
• CRUD guards (§7): POST /pipelines with connector_id=None → 4xx; POST /pipelines/{uploads_id}/run → 4xx (no FAILED run created); PUT setting schedule_cron on the Uploads pipeline → 4xx; DELETE of the Uploads pipeline → allowed, and the next upload re-creates it.
• Chat tool surfaces uploads: get_pipeline_runs returns the upload runs for the workspace (trigger=upload), scoped correctly.
• KB defaults guard: Uploads pipeline save_to_kb=True; a connector pipeline created without save_to_kb defaults False (§6).
• Best-effort isolation: a forced failure inside record_upload_run does not fail the upload request (documents still created + dispatched, 2xx returned).

Risks / trade-offs

• Run is an upload event, not per-file ETL truth. status=succeeded + documents_indexed=<accepted count> is stamped at request time; individual files may still fail ETL afterward (visible on Document.status, not reflected back onto the run). Accepted for MVP — accurate per-file roll-up needs the deferred documents.pipeline_run_id provenance column. Documented; the folder-upload barrier is the natural place to add real status later (§4).
• Singleton Uploads pipeline ⇒ no per-upload folder on the pipeline. destination_folder_id stays NULL; folder placement stays in the upload code. Fine because the upload run is descriptive, not engine-driven. If product later wants "uploads to folder X" as a first-class pipeline, that's a connector-less pipeline variant (additive).
• Partial-unique index assumes connector_id IS NULL ⇔ Uploads. True for MVP (Phase 5 reserved NULL for Uploads, and Phase 6 fails NULL-connector runs). If a second kind of connector-less pipeline is ever introduced, replace the index predicate with an explicit kind/discriminator column (additive migration) — flagged so the NULL-overload doesn't silently block it.
• Best-effort run write can be skipped on error. If record_upload_run raises (e.g. transient DB error) the upload still succeeds but no run is logged — an audit gap, never a data-loss or user-facing failure. Acceptable: the Document rows (the actual KB content) are the source of truth; the run is supplementary history.
• No backfill for historical uploads. Pre-Phase-7 uploads have no upload runs (pipeline is created lazily on the next upload). Accepted — run history starts accumulating from rollout; no migration backfills synthetic runs.
• Upload run doesn't capture the uploader. PipelineRun has no user column (Phase 5), so the run records what/when/how-many, not who — the per-document uploader is still on Document.created_by_id, and the pipeline's created_by is just whoever triggered the first-ever upload. If per-run attribution is wanted, add a nullable triggered_by_id to pipeline_runs (additive Phase-5 model change); deferred.
• enabled=True on a never-scheduled pipeline. The Uploads pipeline is enabled=True with NULL cron — harmless (scheduler ignores it via NULL cron and the connector_id IS NOT NULL filter); enabled only ever gates scheduling, which uploads never use. Left True so a future UI can show it as an active pipeline.
• CRUD-guard coupling to Phase 5. Phase 7 tightens Phase 5's create contract (connector_id=None now rejected) and adds /run + PUT guards. This is a documented evolution (Phase 7 ships after 5/6); the Phase-5 create test is updated here, not left contradictory.

Out of scope (hand-offs)

• documents.pipeline_run_id provenance (per-file → run linkage, accurate ETL roll-up, folder-upload barrier-driven status) → post-MVP (touches documents).
• Engine execution of uploads — intentionally never; uploads are route-recorded (06 fails NULL-connector runs by design).
• Migrating connector periodic indexing into pipelines — still COEXIST (Phase 5 decision); not revisited here.
• Public pay-as-you-go API / MCP KB server → post-MVP (umbrella "Deferred").
• Frontend — surfacing the Uploads pipeline + upload runs in the Pipelines UI, "uploads always saved to KB" copy → frontend umbrella.