diff --git a/plans/backend/00-umbrella-plan.md b/plans/backend/00-umbrella-plan.md index e6157f758..fce43f137 100644 --- a/plans/backend/00-umbrella-plan.md +++ b/plans/backend/00-umbrella-plan.md @@ -95,22 +95,24 @@ Split into two independently testable subplans: ### Phase 5 — Pipelines data model [`subplan: 05-pipelines-model.md`] -- New tables: `pipelines` (workspace_id, user_id, connector_id, name, config JSON, schedule/cron, `save_to_kb` bool, `destination_folder_id` nullable, enabled, next_scheduled_at) and `pipeline_runs` (pipeline_id, status, trigger = manual/scheduled/upload, timestamps, doc counts, error, optional raw-result blob ref). +- New tables: `pipelines` (workspace_id, `created_by_id` nullable SET-NULL [creator metadata, **not** owner], connector_id nullable, name, config JSONB, `schedule_cron` + `schedule_timezone` + enabled + next_scheduled_at, `save_to_kb` bool, `destination_folder_id` nullable) and `pipeline_runs` (pipeline_id, status, trigger = manual/scheduled/upload, timestamps, doc counts, `crawls_*`, `charged_micros`, error, optional raw-result blob ref). - Models + Pydantic schemas + Alembic migration + backend Zero publication entry. - Pipelines API routes: CRUD + manual run trigger + list runs. ### Phase 6 — Pipeline execution + scheduling [`subplan: 06-pipelines-exec.md`] -- Run engine: pipeline run -> invoke connector fetch (WebURL crawler for MVP) -> if `save_to_kb`, route through `IndexingPipelineService` into the destination folder -> write `PipelineRun` record. +- Run engine: pipeline run -> invoke connector fetch (WebURL crawler for MVP) -> if `save_to_kb`, persist to the destination folder (the crawler writes `Document`s directly via `index_crawled_urls`, extended with a `folder_id` param; it does **not** use `IndexingPipelineService`, which is the file/Composio path) -> write `PipelineRun` record. (06 verified the crawler indexer is its own KB-write path.) - **Crawl billing wiring (carry-over from `03c`):** `03c` meters crawls inside `webcrawler_indexer`. A pipeline run that crawls but has `save_to_kb=false` must NOT bypass billing — wire the pipeline fetch through the same `WebCrawlCreditService` (pre-check + charge on `crawls_succeeded`) regardless of the KB-save branch, ideally recording `charged_micros` on the `PipelineRun` for idempotency. Otherwise non-KB pipeline crawls are free by accident. -- Scheduling: reuse the Celery Beat meta-scheduler pattern (`schedule_checker_task.py`, `periodic_scheduler.py`) for cron + manual triggers. +- Scheduling: a Celery Beat tick over `pipelines.next_scheduled_at`, modeled on the **automations** cron **selector** (`automations/triggers/builtin/schedule/selector.py` — cron + `FOR UPDATE SKIP LOCKED` + self-heal, reusing the `croniter` util), which fits `schedule_cron` better than the connector `frequency_minutes` checker. Plus the de-dup guard (a pipeline over a connector disables that connector's own periodic indexing) so the two minute-level scans never double-crawl/bill. - When `save_to_kb` is off, persist the raw fetch result on the run (blob via `file_storage`) so it is retrievable without indexing. - Chat agent context: expose pipeline run history to the `multi_agent_chat` agent (read-only) — via a tool (e.g. `list_pipelines` / `get_pipeline_runs`) and/or a context middleware injection (similar to `KnowledgeTreeMiddleware`). Scope strictly to the active workspace. Gives the agent awareness of recent runs, statuses, schedules, and last-fetched timestamps. ### Phase 7 — File upload as a pipeline + KB-save-secondary [`subplan: 07-upload-pipeline-kb.md`] -- Wire file upload (`documents_routes.py` fileupload flow) to create/use an "Uploads" pipeline and register a `PipelineRun`; uploads always `save_to_kb = true`. -- Generalize KB saving to be opt-in for non-upload pipelines via `save_to_kb` + destination folder. +- Wire file upload (`documents_routes.py` `fileupload` + `folder-upload` flows) to lazily get-or-create a **singleton "Uploads" pipeline** per workspace (`connector_id = NULL`, `save_to_kb = true`) and register a `PipelineRun(trigger=upload)`; uploads always `save_to_kb = true`. +- **Key design fact:** uploads are **route-recorded, not engine-executed** — Phase 6's engine fails `connector_id IS NULL` runs, so the upload routes themselves write a terminal audit `PipelineRun` (the existing upload code stays the executor). The run is an upload-*event* record; per-file ETL outcomes stay on `Document.status` (accurate per-file roll-up needs the deferred `documents.pipeline_run_id` provenance column). +- A small migration adds a partial unique index `ON pipelines(workspace_id) WHERE connector_id IS NULL` (enforces the singleton + race-safe get-or-create). +- KB-save-secondary: the opt-in `save_to_kb` + destination folder for connector pipelines already shipped in Phases 5–6; Phase 7 only records the inverse invariant (uploads always KB) and guards the connector default stays `False`. ## Deferred — Frontend & client phases (separate umbrella, planned LATER) @@ -124,8 +126,8 @@ These are recorded for continuity but are NOT planned in this umbrella. They sta ## Open items to confirm during subplanning - ~~Rename transition: hard cutover vs temporary API aliases~~ RESOLVED: HARD CUTOVER (see resolved log + 02-rename-backend.md). The frontend is rebuilt against the corrected backend in its own umbrella; backend is verified via tests/OpenAPI, not the old UI. -- Whether existing connector periodic-indexing config is migrated into Pipelines or coexists during MVP. -- Chat agent run-history access: tool vs middleware injection vs both (default: tool). +- ~~Whether existing connector periodic-indexing config is migrated into Pipelines or coexists during MVP.~~ RESOLVED (Phase 5): COEXIST — connector periodic path stays untouched; pipelines add a sibling `next_scheduled_at` scan. Phase 6 owns the de-dup guard (a pipeline over a connector disables that connector's `periodic_indexing_enabled`) to avoid double crawl/bill. See `05-pipelines-model.md`. +- ~~Chat agent run-history access: tool vs middleware injection vs both (default: tool).~~ RESOLVED (Phase 6): **tool** — a read-only main-agent `get_pipeline_runs` tool (registry pattern), workspace-scoped via its build-time `search_space_id`, opening its own `shielded_async_session()` (like `KnowledgeTreeMiddleware`). Always-on `` middleware injection is deferred (per-turn token cost). See `06-pipelines-exec.md`. - ~~Type-2 MCP migration depth~~ RESOLVED (Phase 4): branded natives are tagged `MIGRATING` and turned OFF for MVP (not re-pointed to MCP yet); only the generic `MCP_CONNECTOR` is a functional Type-2. Real MCP re-pointing is post-MVP. ## Resolved decisions log @@ -143,6 +145,9 @@ These are recorded for continuity but are NOT planned in this umbrella. They sta - Billable unit: one unit per URL that returns usable extracted content, regardless of how many internal fallback tiers were attempted (not per HTTP fetch, not per URL-processed). - Captcha solving (captcha-tools): DEFERRED to the last Phase-3 subplan (`03d`); non-MVP-blocking. - Roadmap: WebURL Crawler & Crawl Billing inserted as the new Phase 3; connector two-type → Phase 4; pipelines → Phases 5/6/7. +- Phase 5 pipelines data model: two new tables `pipelines` (mutable) + `pipeline_runs` (append-only), modeled on `automations`/`automation_runs`; ORM lives in `db.py` next to connectors/folders. `connector_id` nullable (NULL = Phase-7 Uploads), eligibility enforced at create via 04a's `is_pipeline_eligible`. Schedule = `schedule_cron` + `schedule_timezone` (default UTC) + `next_scheduled_at` (cron, matching automations). `pipeline_runs` pre-includes `charged_micros`/`crawls_*`/`result_blob_key` so Phase 6 needs no extra migration. Both tables published to Zero **full-row** (like folders/connectors). Routes reuse `CONNECTORS_*` permissions. Phase 5 ships the data model + API surface only; the `/run` endpoint enqueues a Phase-6 task stub. +- Phase 7 uploads-as-pipeline: a **singleton "Uploads" pipeline** per workspace (`connector_id NULL`, `save_to_kb=true`), lazily get-or-created (race-safe via a partial unique index `ON pipelines(workspace_id) WHERE connector_id IS NULL`). Each `fileupload`/`folder-upload` request writes a **terminal audit `PipelineRun(trigger=upload, status=succeeded, documents_indexed=)`** — uploads are **route-recorded, not engine-executed** (Phase 6 fails NULL-connector runs by design; existing upload code stays the executor). Best-effort via an **inner** try/except (never 5xx the upload — the route's outer handler would otherwise 500 an already-committed upload). No crawl billing (uploads aren't crawls; `charged_micros` NULL). Per-file ETL truth stays on `Document.status`; accurate roll-up needs the deferred `documents.pipeline_run_id` provenance. Connector `save_to_kb` default stays `False` (opt-in for connectors, mandatory for uploads). Phase 7 also **guards Phase-5's generic CRUD** against the system Uploads pipeline: `POST /pipelines` rejects `connector_id=None` (supersedes Phase 5's permissive create), `/run` and schedule-`PUT` reject NULL-connector pipelines, and Phase 6's scheduler `_claim_due` filters `connector_id IS NOT NULL` as a backstop (so the Uploads pipeline can never be manually-run or scheduled into perpetually-failing runs). See `07-upload-pipeline-kb.md`. +- Phase 6 pipeline execution: run engine mirrors **automations** (thin Celery `run_pipeline(run_id)` → `execute_pipeline_run`; PENDING-gated, idempotent terminal no-op; `pending→running→succeeded/failed` with timing/counts/error). MVP executor = **WebURL crawler only** (other types fail cleanly). `save_to_kb=true` reuses `index_crawled_urls` extended with a `folder_id` param (lands in the destination folder); `save_to_kb=false` runs a **fetch-only** loop and persists one JSON blob via `file_storage` (`result_blob_key`). **Crawl billing is owned by the run engine** for the pipeline path (pre-check on `len(urls)` + charge `crawls_succeeded` + idempotent `charged_micros`), calling the crawler with a new `bill=False` seam (the connector `/index`+periodic paths keep `03c`'s in-indexer `bill=True`) — so non-KB runs are billed identically. Scheduler = a `pipeline_schedule_select` Beat tick modeled on the automations cron **selector** (cron + `FOR UPDATE SKIP LOCKED` + self-heal, using the existing `croniter` util), plus the **de-dup guard** (a pipeline over a connector disables that connector's `periodic_indexing_enabled`). Chat context = the `get_pipeline_runs` tool. Carries a small additive `05` amendment: a `schedule_timezone` column (cron util needs a tz). See `06-pipelines-exec.md`. ## Subplan index (backend) @@ -156,8 +161,8 @@ These are recorded for continuity but are NOT planned in this umbrella. They sta | 3 | `03d-captcha-solving.md` | drafted (deferred — last) | | 4 | `04a-connector-category.md` | drafted | | 4 | `04b-source-discovery.md` | drafted | -| 5 | `05-pipelines-model.md` | not started | -| 6 | `06-pipelines-exec.md` | not started | -| 7 | `07-upload-pipeline-kb.md` | not started | +| 5 | `05-pipelines-model.md` | drafted | +| 6 | `06-pipelines-exec.md` | drafted | +| 7 | `07-upload-pipeline-kb.md` | drafted | Frontend & client subplans will be added under a separate umbrella later (see "Deferred — Frontend & client phases"). diff --git a/plans/backend/01-rename-db.md b/plans/backend/01-rename-db.md index 0ea093374..2df625c7a 100644 --- a/plans/backend/01-rename-db.md +++ b/plans/backend/01-rename-db.md @@ -82,7 +82,7 @@ These reference `search_space_id` by COLUMN but the index/constraint NAME does n ### Conventions (from the repo) -- Revision ids are plain integers; head is `165` (`revision="165"`, `down_revision="164"`) — [surfsense_backend/alembic/versions/165_add_chunk_position.py](surfsense_backend/alembic/versions/165_add_chunk_position.py) lines 25-26. New file: `166_rename_searchspace_to_workspace.py` with `revision="166"`, `down_revision="165"`. +- Revision ids are plain integers; the new migration chains after the **then-current head** — verify with `alembic heads` at implementation time, since the head advances as other PRs merge. Today the head is `166` ([surfsense_backend/alembic/versions/166_add_pat_and_api_access.py](surfsense_backend/alembic/versions/166_add_pat_and_api_access.py)), so the new file would be `167_rename_searchspace_to_workspace.py` with `revision="167"`, `down_revision="166"`. (Note: Phase 4b and Phase 5 each add a migration too; whichever lands first takes the next integer — chain by actual head, not by these illustrative numbers.) - Migrations run one-per-transaction under an advisory lock — [surfsense_backend/alembic/env.py](surfsense_backend/alembic/env.py) lines 77, 80-95. So all renames below land atomically. - Raw `op.execute("ALTER TABLE ...")` is the house style (e.g. [165_add_chunk_position.py](surfsense_backend/alembic/versions/165_add_chunk_position.py) lines 39-49). diff --git a/plans/backend/02-rename-backend.md b/plans/backend/02-rename-backend.md index f36b60905..41f96028c 100644 --- a/plans/backend/02-rename-backend.md +++ b/plans/backend/02-rename-backend.md @@ -94,7 +94,7 @@ These do not move with a symbol rename; each is decided here. 6. Notification dedup/operation IDs embedding `{search_space_id}` (e.g. `doc_..._{search_space_id}_...`, `insufficient_credits_{search_space_id}_...`) and frontend deep-link strings like `/dashboard/{search_space_id}/buy-more`. DECISION: the ID is a numeric value, not the literal word — leave format strings as-is functionally; the embedded VALUE is unchanged. The `/dashboard/{id}/...` deep link points at a FRONTEND route still named `[search_space_id]` (deferred umbrella) — KEEP it until the frontend segment renames, else links 404. 7. Storage path builders — `documents/{search_space_id}/...` ([file_storage/keys.py](surfsense_backend/app/file_storage/keys.py) 20-26) and `podcasts/{search_space_id}/...` ([podcasts/storage.py](surfsense_backend/app/podcasts/storage.py) 22-25). The path segment is the numeric ID; the literal word `search_space` is NOT in stored object keys. DECISION: rename the param only; NO blob migration needed; existing objects keep resolving. 8. `SearchSourceConnector` — contains "search" but is the connectors table, a different concept (the word "search" here is unrelated to `SearchSpace`). OUT OF SCOPE: this rename does NOT touch it, and Phase 4 does not rename the class either — Phase 4 adds a Type-1/Type-2 taxonomy via a static `connector_type`→(category, availability) registry (no new column) and KEEPS `is_indexable` (`db.py:1868`). Leave `SearchSourceConnector` as-is. -9. Historical Alembic migrations (`surfsense_backend/alembic/versions/*`) — ~20 files embed `searchspaces` / `search_space_id` as raw-SQL string literals (e.g. `23_associate_connectors_with_search_spaces.py`, `41_backfill_rbac_for_existing_searchspaces.py`, `40_move_llm_preferences_to_searchspace.py`). DECISION: NEVER rewrite these. They are an immutable replay log that intentionally references the schema as it existed at that revision; rewriting them corrupts history and breaks a clean `alembic upgrade` from zero. Verified safe: no migration imports the ORM classes (only `from app.db import Base` in `env.py` / `0_initial_schema.py`), so the class rename does not touch them. Phase 1's migration 166 is the single transition point; migrations >166 use the new names. The scripted rename scope is `app/` + `tests/` ONLY. +9. Historical Alembic migrations (`surfsense_backend/alembic/versions/*`) — ~20 files embed `searchspaces` / `search_space_id` as raw-SQL string literals (e.g. `23_associate_connectors_with_search_spaces.py`, `41_backfill_rbac_for_existing_searchspaces.py`, `40_move_llm_preferences_to_searchspace.py`). DECISION: NEVER rewrite these. They are an immutable replay log that intentionally references the schema as it existed at that revision; rewriting them corrupts history and breaks a clean `alembic upgrade` from zero. Verified safe: no migration imports the ORM classes (only `from app.db import Base` in `env.py` / `0_initial_schema.py`), so the class rename does not touch them. Phase 1's rename migration (the next integer after the live head — `167` at time of writing, but chain by actual `alembic heads`) is the single transition point; migrations after it use the new names. The scripted rename scope is `app/` + `tests/` ONLY. 10. LangGraph persisted state channel key `search_space_id` — `input_state = {..., "search_space_id": search_space_id, ...}` ([tasks/chat/streaming/flows/new_chat/input_state.py](surfsense_backend/app/tasks/chat/streaming/flows/new_chat/input_state.py) 131) is a CHECKPOINTED state channel. The resume_chat flow reads persisted state (`agent.aget_state({"configurable": {"thread_id": ...}})`, [flows/resume_chat/resume_routing.py](surfsense_backend/app/tasks/chat/streaming/flows/resume_chat/resume_routing.py) 50) and HITL interrupts (`surfsense_resume_value`, `HumanReview`) can sit pending across a deploy. DECISION: rename the channel key to `workspace_id` for consistency, and ACCEPT that any chat thread paused at a HITL interrupt BEFORE the cutover must be restarted after (the old checkpoint carries `search_space_id`, the new graph reads `workspace_id`). Operationally: drain/resolve pending interrupts before deploy if practical. (Alternative — keep the channel key as a carve-out — rejected: leaves a lone `search_space_id` in otherwise-renamed agent state for a transient, conversation-scoped value. The `configurable` keys are `thread_id` / `surfsense_resume_value`, not `search_space_id`, so only this state channel is affected.) 11. User-facing default literal — `users.py` seeds the default workspace `name="My Search Space"` (line 158, persisted + shown to users) and logs "Created default search space" (207). DECISION: rename the default name to "My Workspace" (backend-created seed value, not caught by a symbol rename); log strings are cosmetic. Also update the Redis-key assertion in `tests/unit/services/test_ai_sort_task_dedupe.py:14` when literal 3 is renamed. diff --git a/plans/backend/03a-crawler-core.md b/plans/backend/03a-crawler-core.md index d563a2913..afa96e368 100644 --- a/plans/backend/03a-crawler-core.md +++ b/plans/backend/03a-crawler-core.md @@ -83,7 +83,7 @@ class CrawlOutcomeStatus(str, Enum): FAILED = "failed" # invalid URL or every tier errored ``` -`crawl_url()` returns a small dataclass `CrawlOutcome(status, result, error, tier)` (or, to minimize churn, keep the tuple and add a third element). Either way the **billable success predicate is single-sourced**: `status == SUCCESS`. +`crawl_url()` returns a small dataclass `CrawlOutcome(status, result, error, tier)` — **commit to the dataclass** (not a tuple): `03c` keys billing off `status == SUCCESS`, and Phase 6's fetch-only path (`06-pipelines-exec.md`) consumes `outcome.status` / `outcome.result` / `outcome.error` as attributes, so a tuple form would break that consumer. The **billable success predicate is single-sourced**: `status == CrawlOutcomeStatus.SUCCESS`. | Outcome | When | Billable (`03c`)? | Document status (indexer) | |---------|------|-------------------|---------------------------| @@ -95,7 +95,9 @@ class CrawlOutcomeStatus(str, Enum): ### Success counter for the indexer -Add an explicit `crawls_succeeded` counter in `index_crawled_urls()` incremented whenever `crawl_url` returns `SUCCESS` (right after line 297's call, before the dedupe/unchanged branches), and surface it in the task-success metadata (lines 455–466) and return value. This is the hand-off point `03c` meters against. +Add an explicit `crawls_succeeded` counter in `index_crawled_urls()` incremented whenever `crawl_url` returns `SUCCESS` (right after line 297's call, before the dedupe/unchanged branches), and surface it in the task-success metadata (lines 455–466). `03c` meters against this **in-function** counter (it charges inside the indexer — the count does not need to escape via the return). + +> **Do NOT widen the positional return tuple.** The shared `_run_indexing_with_notifications` wrapper unpacks every indexer's return **by length** (`search_source_connectors_routes.py:1499–1507`: `if len(result) == 3: a,b,c = result else: a,b = result`) — a 3-tuple would mislabel `crawls_succeeded` as `documents_skipped`, a 4-tuple would raise `ValueError`. Keep the existing `(total_processed, error)` shape. **Phase 6** (`06-pipelines-exec.md`) later exposes `crawls_succeeded`/`documents_indexed`/`crawls_attempted` to the pipeline run engine via an optional `stats` **out-param** (plus `folder_id`/`urls`/`bill`), not via the return — so `03a` only needs the counter + metadata here. ### Chat scrape tool @@ -105,7 +107,7 @@ Drop the Firecrawl-only `formats=["markdown"]` arg (markdown is already the Traf 1. **Rip out Firecrawl** across the surface table above (crawler, dep, validators, indexer, chat plumbing, automations, tests) — code-only, no env/docs change. 2. **Refactor `crawl_url`** to the 3-tier Scrapling ladder + explicit `CrawlOutcome`; enable `solve_cloudflare` on the stealthy tier. -3. **Add `crawls_succeeded`** counting in `webcrawler_indexer.py` + expose in result/metadata. +3. **Add `crawls_succeeded`** counting in `webcrawler_indexer.py` + expose in **task metadata only** (03c bills off the in-function counter). **Do not change the positional return tuple** (the shared wrapper unpacks by length; Phase 6 adds a `stats` out-param). 4. **Update both `scrape_webpage` tools** to drop `firecrawl_api_key` + `formats`. 5. **Tests:** unit tests for `crawl_url` outcomes (mock Scrapling fetchers → SUCCESS/EMPTY/FAILED); update `test_dependencies.py`; assert the indexer's `crawls_succeeded` count. diff --git a/plans/backend/03c-crawl-billing.md b/plans/backend/03c-crawl-billing.md index f6106209d..0ea6a01cd 100644 --- a/plans/backend/03c-crawl-billing.md +++ b/plans/backend/03c-crawl-billing.md @@ -127,6 +127,6 @@ One unit per `CrawlOutcomeStatus.SUCCESS` — a URL that yielded usable extracte ## Out of scope (hand-offs) -- Per-**pipeline** run accounting / `charged_micros` persistence for idempotency → Phases 5–7. +- Per-**pipeline** run accounting / `charged_micros` persistence for idempotency → Phases 5–7. Specifically, **Phase 6 moves pipeline-run billing out of the indexer to the run level**: it adds a `bill: bool = True` param to this indexer wiring and calls `index_crawled_urls(bill=False)` from the pipeline engine, which then does its own `check_credits` + `charge_credits(crawls_succeeded)` and stamps `PipelineRun.charged_micros` (idempotency). So this phase's in-indexer billing keeps serving the **connector `/index` + periodic** paths; the "pipeline crawls" half of the objective is refined by `06`. Structure the §2 wiring so it's easy to gate behind that flag. - Captcha-solver upstream cost pass-through → `03d` (deferred). - Surfacing crawl spend in the credit-status UI → frontend umbrella. diff --git a/plans/backend/05-pipelines-model.md b/plans/backend/05-pipelines-model.md new file mode 100644 index 000000000..bdb057b47 --- /dev/null +++ b/plans/backend/05-pipelines-model.md @@ -0,0 +1,318 @@ +# Phase 5 — Pipelines data model (tables, schemas, migration, Zero, CRUD + run routes) + +> Part of the **CI Pivot MVP**. See `00-umbrella-plan.md` (Phase 5). +> Precondition: Phases 1–2 (rename) live, Phase 4a (connector registry) live. Siblings ahead: `06-pipelines-exec.md` (run engine + scheduling), `07-upload-pipeline-kb.md` (uploads-as-pipeline). + +> **Implementation note (post-rename).** This phase adds **brand-new** tables, created *after* the rename, so they use the canonical names natively: table `workspaces`, column `workspace_id`, ORM class `Workspace`. Existing code cited below still shows the **pre-rename** names (`searchspaces`/`search_space_id`/`SearchSpace`) because those citations were taken against today's tree — Phases 1–2 rename them; map accordingly. Locate code by **symbol/grep**, not the absolute line numbers cited here (the rename + Phase 3/4 migrations shift them). + +## Objective + +Introduce two first-class tables — `pipelines` and `pipeline_runs` — plus their enums, Pydantic schemas, one Alembic migration, the backend Zero-publication entries, and the HTTP routes (CRUD + manual-run trigger + list/get runs). **This phase is data-model + API surface only.** The actual run **engine** (invoke connector fetch → optional KB save → write run record), scheduling tick, crawl-billing wiring, and chat-agent context exposure are **Phase 6**; wiring file upload onto an "Uploads" pipeline is **Phase 7**. The manual-run endpoint here **creates a `PipelineRun` row and enqueues the Phase-6 task** (which is a stub until Phase 6 lands), so the surface is testable end-to-end without the engine. + +A **Pipeline** = a saved, runnable fetch over a Type-1 data source: it references a connector + per-pipeline config + an optional cron schedule + a KB destination (`save_to_kb` + `destination_folder_id`). A **PipelineRun** = one immutable execution record (manual / scheduled / upload), carrying status, timing, doc counts, error, an optional raw-result blob ref, and a billing idempotency field. + +## Locked model (MVP) + +| Concept | Decision | +|---------|----------| +| Tables | `pipelines` (mutable definition) + `pipeline_runs` (append-only execution record). Mirrors the `automations` / `automation_runs` split. | +| Where the ORM lives | In **`app/db.py`** alongside `Folder`/`Document`/`SearchSourceConnector`/`Log` (the other core, Zero-published, cross-referenced entities), **not** a separate package. Rationale: needs `back_populates` on `Workspace`/`User`/`SearchSourceConnector`/`Folder`, a Zero entry, and a `schemas/` Pydantic mirror — all lowest-friction in `db.py`. (The `automations` package avoided `db.py` but pays for it with string-relationships + import-time model registration; pipelines are more core, like connectors.) | +| Pipeline → connector | `connector_id` **nullable**, FK `ON DELETE CASCADE`. A connector-backed pipeline points at a Type-1 source; **`NULL` = a non-connector pipeline** (the Phase-7 "Uploads" pipeline). Eligibility (`DATA_SOURCE` + `AVAILABLE`) is enforced at create via Phase 4a's `is_pipeline_eligible` (app-level, not a DB constraint — the registry is code) **and re-checked at run time in Phase 6** (the static registry can flip a type to `MIGRATING` on deploy, stranding existing pipelines). | +| Ownership | `created_by_id` FK `ON DELETE SET NULL`, **nullable** — creator metadata only (workspace-shared, like `Folder`/`Automation`), not an owner. Billing targets the workspace owner (03c). | +| Schedule representation | `schedule_cron VARCHAR NULL` (NULL = manual-only) + `schedule_timezone VARCHAR NOT NULL DEFAULT 'UTC'` + `enabled BOOLEAN` + `next_scheduled_at TIMESTAMPTZ NULL` (computed by Phase 6's tick). Chosen over a bare `frequency_minutes` int to match the **automations** schedule model and reuse its cron util (`automations/triggers/builtin/schedule/cron.py`), which takes **both** a cron and a timezone (`compute_next_fire_at(cron, timezone, …)`). | +| KB destination | `save_to_kb BOOLEAN` + `destination_folder_id` FK `ON DELETE SET NULL` (nullable). `save_to_kb=true` with NULL folder ⇒ index to workspace root (`documents.folder_id` is nullable). No DB CHECK couples them (an FK `SET NULL` would violate a conditional CHECK); the pairing is validated at the API only. | +| Run status enum | `pipeline_run_status`: `pending`, `running`, `succeeded`, `failed`, `cancelled` (mirrors `automation_run_status` minus `timed_out`). | +| Run trigger enum | `pipeline_run_trigger`: `manual`, `scheduled`, `upload`. | +| Periodic coexistence | **COEXIST, do not migrate** (resolves umbrella open item §2 — see "Periodic-indexing coexistence" below). | +| Zero publication | Publish **both tables full-row** (`None`), mirroring `folders` / `search_source_connectors`. Avoids the column-list `_0_version` seam entirely; neither table has a bulky column. | +| Naming | The CI **`Pipeline`** entity (a saved, runnable fetch) is distinct from the existing **`IndexingPipelineService`** (`app/indexing_pipeline/`, the KB-ingest service). They coexist but do **not** share a code path for the MVP: Phase 6's WebURL executor reuses the crawler's own KB-write path (`index_crawled_urls`, extended with `folder_id`), **not** `IndexingPipelineService` (which is the file/Composio ingest path — and is what Phase 7's uploads flow builds on). Name routes/services `pipeline_*` (not `indexing_pipeline_*`) to keep the two legible. | + +## Current state (cited) + +### The precedent: `automations` / `automation_runs` + +The cleanest existing analogue to model against. + +- Editable parent: `Automation` — `automations/persistence/models/automation.py:24`. Pattern: `BaseModel, TimestampMixin`; `search_space_id` FK CASCADE (`:27`), `created_by_user_id` FK SET NULL (`:34`), a native enum `status` with `values_callable=lambda x: [e.value for e in x]` (`:44–54`), a JSONB `definition` (`:56`), an explicit `updated_at` with `onupdate` (`:60–66`), and `runs = relationship(..., cascade="all, delete-orphan", passive_deletes=True)` (`:76–81`). +- Append-only child: `AutomationRun` — `automations/persistence/models/run.py:20`. `automation_id` FK CASCADE (`:23`), native enum `status` default `pending` (`:37–47`), JSONB `error` (`:60`), nullable `started_at`/`finished_at` (`:62–63`). +- Enums: `automations/persistence/enums/run_status.py:8` (`RunStatus(StrEnum)` = pending/running/succeeded/failed/cancelled/timed_out). +- Migration: `alembic/versions/144_add_automation_tables.py` — the **exact template** for this phase: idempotent `CREATE TYPE ... DO $$ IF NOT EXISTS` guards (`:31–76`), `CREATE TABLE IF NOT EXISTS` with inline FKs (`:79–96`), `CREATE INDEX IF NOT EXISTS` per FK/status/created_at (`:97–111`), a **partial "due" index** for the schedule scan (`ix_automation_triggers_due`, `:144–152`), and symmetric `downgrade()` dropping indexes→tables→types (`:190–213`). +- Run routes: `automations/api/run.py` — `GET /automations/{id}/runs` (list, paginated `limit/offset`, `:13–30`) + `GET /automations/{id}/runs/{run_id}` (detail, `:33–44`). The exact shape to mirror for pipeline runs. + +### The other core tables (relationship + column conventions) + +- Base: `TimestampMixin` (gives `created_at`, `db.py:487`) + `BaseModel` (`id SERIAL PK`, `db.py:498`). Tables wanting a mutable `updated_at` declare it explicitly (see `Folder.updated_at` `db.py:1333–1339`, with `onupdate` + `index=True`). +- `SearchSourceConnector` — `db.py:1829`. `connector_type` native enum (`:1867`), `is_indexable` (`:1868`), periodic fields `periodic_indexing_enabled`/`indexing_frequency_minutes`/`next_scheduled_at` (`:1880–1882`), `search_space_id` FK CASCADE (`:1884`), `user_id` FK CASCADE (`:1891`), `documents = relationship(...)` (`:1897`). **Pipelines FK this table** (`connector_id`). +- `Folder` — `db.py:1310`. `search_space_id` FK CASCADE (`:1321`), `created_by_id` FK SET NULL (`:1327`). **Pipelines FK this table** (`destination_folder_id`, SET NULL). +- `Document.folder_id` is nullable (root/unfiled docs exist) — confirmed by `DOCUMENT_COLS` including `folder_id` (`zero_publication.py:32`) and `Document.folder` `passive_deletes=True` (`db.py:1345`). Lets `save_to_kb` work with a NULL destination. +- `Permission` enum has `CONNECTORS_CREATE/READ/UPDATE/DELETE` (`db.py:345–348`); routes authz via `check_permission(session, auth, search_space_id, Permission.X.value, msg)` (`folders_routes.py:42–48`) with `auth = Depends(get_auth_context)` + `session = Depends(get_async_session)` (`folders_routes.py:36–37`). **Reuse the `CONNECTORS_*` permissions** for pipelines (no new permission needed for MVP). + +### Schema conventions + +- `schemas/base.py`: `IDModel` (`id:int`, `:11`) + `TimestampModel` (`created_at`, `:6`), both `ConfigDict(from_attributes=True)`. +- `schemas/search_source_connector.py`: `Base` → `Create(Base)` → `Update(BaseModel, all-Optional)` → `Read(Base, IDModel, TimestampModel)` with `model_config = ConfigDict(from_attributes=True)` and `field_validator`/`model_validator` for config consistency (`:24–56`). The shape to mirror for `schemas/pipeline.py`. + +### Zero publication mechanics + +- `zero_publication.py`: `ZERO_PUBLICATION` map (`:71–82`) is the single source of truth; `None` ⇒ publish full row (e.g. `folders`, `search_source_connectors`), a list ⇒ column subset (e.g. `AUTOMATION_RUN_COLS` `:44–53`). `apply_publication(conn)` (`:151`) reconciles via `ALTER PUBLICATION ... SET TABLE`; a migration just calls it (template: `159_publish_podcasts_to_zero.py:21–22`). `_format_table_entry` omits a table until it physically exists with all its canonical columns (`:113–140`), so the migration **must create the tables before** calling `apply_publication`. The `_0_version` allowlist (`{"documents","user","podcasts"}`, `:106`) applies **only to column-list tables** — irrelevant here since we publish full-row. + +### The meta-scheduler we coexist with (Phase-6 reuse target) + +- `tasks/celery_tasks/schedule_checker_task.py:17` — `check_periodic_schedules_task`, runs every minute, scans `SearchSourceConnector` where `periodic_indexing_enabled AND next_scheduled_at <= now` (`:33–39`) and dispatches per-type Celery tasks. This is the **connector-level** periodic path and is **untouched** by Phase 5. Phase 6 adds a sibling scan over `pipelines.next_scheduled_at`. +- `utils/periodic_scheduler.py:30` — `create_periodic_schedule` (first-run trigger helper). Reference for Phase 6's pipeline scheduler. + +### Alembic head + +Today's head is **`166`** (`alembic/versions/166_add_pat_and_api_access.py`; sequential integer-prefixed files). By the time Phase 5 is implemented, Phase 1 (rename) and Phase 4b (search-enum drop) have each added a migration ahead of `166`. **Set `down_revision` to the then-current head** — verify with `alembic heads`; do not hardcode `166`. + +## Target design + +### 1. Enums (`db.py`, near the other `StrEnum`s) + +```python +class PipelineRunStatus(StrEnum): + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + +class PipelineRunTrigger(StrEnum): + MANUAL = "manual" + SCHEDULED = "scheduled" + UPLOAD = "upload" +``` + +Mapped with native PG enum types `pipeline_run_status` / `pipeline_run_trigger` using the `values_callable=lambda x: [e.value for e in x]` convention (matches `Automation.status` `automation.py:46–49`, so the DB stores lowercase values, not the Python member names). + +### 2. `pipelines` table (`db.py`) + +```python +class Pipeline(BaseModel, TimestampMixin): + __tablename__ = "pipelines" + + name = Column(String(200), nullable=False, index=True) + config = Column(JSONB, nullable=False, default=dict, server_default=text("'{}'::jsonb")) # per-pipeline overrides (URL list, crawl opts, proxy override seam) + + save_to_kb = Column(Boolean, nullable=False, default=False, server_default="false") + + # Schedule (NULL cron = manual-only). next_scheduled_at is owned/written by Phase 6's tick. + schedule_cron = Column(String(120), nullable=True) + schedule_timezone = Column(String(64), nullable=False, default="UTC", server_default="UTC") # cron util needs a tz (compute_next_fire_at) + enabled = Column(Boolean, nullable=False, default=True, server_default="true") + next_scheduled_at = Column(TIMESTAMP(timezone=True), nullable=True) + + # NULL connector_id = non-connector pipeline (Phase-7 Uploads). + connector_id = Column(Integer, ForeignKey("search_source_connectors.id", ondelete="CASCADE"), nullable=True, index=True) + destination_folder_id = Column(Integer, ForeignKey("folders.id", ondelete="SET NULL"), nullable=True, index=True) + workspace_id = Column(Integer, ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False, index=True) + # Creator metadata, NOT owner: pipelines are workspace-shared (like folders/automations), + # so member deletion must NOT nuke them — SET NULL, nullable. Billing targets the workspace + # owner (03c), not this field. Phase 6 resolves the acting user as created_by ?? workspace owner. + created_by_id = Column(UUID(as_uuid=True), ForeignKey("user.id", ondelete="SET NULL"), nullable=True, index=True) + + updated_at = Column(TIMESTAMP(timezone=True), nullable=False, + default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC), index=True) + + connector = relationship("SearchSourceConnector", back_populates="pipelines") + destination_folder = relationship("Folder") + workspace = relationship("Workspace", back_populates="pipelines") + created_by = relationship("User", back_populates="pipelines") + runs = relationship("PipelineRun", back_populates="pipeline", cascade="all, delete-orphan", passive_deletes=True) +``` + +Add the inverse `pipelines = relationship("Pipeline", back_populates=...)` on **`Workspace`**, **`User`**, and **`SearchSourceConnector`** (mirror how `SearchSpace.search_source_connectors` / `User.search_source_connectors` are declared at `db.py:1887–1894`; the `created_by`/SET-NULL side mirrors `Folder.created_by` `db.py:1344` and `Automation.created_by` `automation.py:69`). `destination_folder` is intentionally one-directional (no back_populates on `Folder`) to keep the folder model lean. + +> **Ownership decision (review).** Earlier draft used connector-style `user_id NOT NULL CASCADE`. Corrected to `created_by_id … SET NULL` because pipelines are **workspace-shared** content (closer to `Folder`/`Automation` than to the per-user `SearchSourceConnector`): removing a member must not delete the workspace's pipelines or erase run-history audit. For *connector-backed* pipelines this also avoids surprising double-cascade (the connector's own `user_id CASCADE` would already drop them on member deletion via `connector_id`); for the Phase-7 Uploads pipeline (`connector_id` NULL) it's the only thing keeping the pipeline alive after its creator leaves. + +> **FK target naming:** `workspaces` / `workspace_id` assume Phases 1–2 are merged (the table is `workspaces` by then). If sequencing slips and Phase 5 lands before the rename, the FK target is `searchspaces` and the column `search_space_id` — but per the umbrella the rename is strictly first, so author against `workspaces`. + +### 3. `pipeline_runs` table (`db.py`) + +```python +class PipelineRun(BaseModel, TimestampMixin): + __tablename__ = "pipeline_runs" + + pipeline_id = Column(Integer, ForeignKey("pipelines.id", ondelete="CASCADE"), nullable=False, index=True) + + status = Column(SQLAlchemyEnum(PipelineRunStatus, name="pipeline_run_status", + values_callable=lambda x: [e.value for e in x]), + nullable=False, default=PipelineRunStatus.PENDING, + server_default=PipelineRunStatus.PENDING.value, index=True) + trigger = Column(SQLAlchemyEnum(PipelineRunTrigger, name="pipeline_run_trigger", + values_callable=lambda x: [e.value for e in x]), nullable=False, index=True) + + # Result accounting (written by Phase 6). + documents_indexed = Column(Integer, nullable=True) + crawls_attempted = Column(Integer, nullable=True) + crawls_succeeded = Column(Integer, nullable=True) + error = Column(JSONB, nullable=True) + + # Raw fetch blob ref for save_to_kb=false runs (Phase 6 writes via file_storage). Just a key/path string. + result_blob_key = Column(String, nullable=True) + + # Crawl-billing idempotency (carry-over from 03c / umbrella Phase 6 §105): micro-USD charged for this run. + charged_micros = Column(BigInteger, nullable=True) + + started_at = Column(TIMESTAMP(timezone=True), nullable=True) + finished_at = Column(TIMESTAMP(timezone=True), nullable=True) + + pipeline = relationship("Pipeline", back_populates="runs") +``` + +These extra columns (`crawls_*`, `result_blob_key`, `charged_micros`) are added **now** so Phase 6 needs **no second migration** — they sit unused until the engine writes them. (`charged_micros` directly satisfies umbrella Phase 6 line 105's "record `charged_micros` on the `PipelineRun` for idempotency".) + +> **Out of scope (provenance):** linking individual `documents` back to the `pipeline_run` that produced them (a `documents.pipeline_run_id` column) is **not** in this phase — it would touch the existing `documents` table. Defer to Phase 6/7 if run-level provenance is wanted. + +### 4. Migration (one file, `144`-shaped) + +Chain after the then-current head. In `upgrade()`: + +1. `CREATE TYPE pipeline_run_status` + `pipeline_run_trigger` behind `DO $$ IF NOT EXISTS` guards (copy `144:31–76`). +2. `CREATE TABLE IF NOT EXISTS pipelines (...)` with inline FKs: `workspace_id → workspaces(id) ON DELETE CASCADE` (NOT NULL), `connector_id → search_source_connectors(id) ON DELETE CASCADE` (nullable), `destination_folder_id → folders(id) ON DELETE SET NULL` (nullable), `created_by_id → "user"(id) ON DELETE SET NULL` (nullable; note the quoted `"user"` table). `id SERIAL PK`, `created_at`/`updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()`, `config JSONB NOT NULL DEFAULT '{}'::jsonb`, `schedule_timezone VARCHAR(64) NOT NULL DEFAULT 'UTC'`. +3. `CREATE TABLE IF NOT EXISTS pipeline_runs (...)`. +4. Indexes (`CREATE INDEX IF NOT EXISTS`): per FK (`workspace_id`, `created_by_id`, `connector_id`, `destination_folder_id`, `pipeline_id`), plus `status`, `created_at`, `updated_at`, `name`, `trigger`. Add the **partial "due" index** for Phase 6's tick: + +```sql +CREATE INDEX IF NOT EXISTS ix_pipelines_due + ON pipelines (next_scheduled_at) + WHERE enabled = true AND next_scheduled_at IS NOT NULL; +``` + +5. `from app.zero_publication import apply_publication; apply_publication(op.get_bind())` **after** the tables exist (so `_format_table_entry` includes them). Requires the `ZERO_PUBLICATION` code change (§6) to be in the tree. + +`downgrade()`: drop indexes → `pipeline_runs` → `pipelines` → both types (symmetric to `144:190–213`). (The publication reconcile is intentionally not reversed — historical publication shapes are immutable, per `159:25–27`.) + +### 5. Pydantic schemas (`app/schemas/pipeline.py`, exported via `app/schemas/__init__.py`) + +```python +class PipelineBase(BaseModel): + name: str + connector_id: int | None = None + config: dict[str, Any] = {} + save_to_kb: bool = False + destination_folder_id: int | None = None + schedule_cron: str | None = None + schedule_timezone: str = "UTC" + enabled: bool = True + +class PipelineCreate(PipelineBase): + workspace_id: int # required on create (path/body), like FolderCreate.search_space_id + +class PipelineUpdate(BaseModel): # all-Optional, like SearchSourceConnectorUpdate + name: str | None = None + config: dict[str, Any] | None = None + save_to_kb: bool | None = None + destination_folder_id: int | None = None + schedule_cron: str | None = None + schedule_timezone: str | None = None + enabled: bool | None = None + # connector_id is immutable after create (re-create to re-point); not in Update. + +class PipelineRead(PipelineBase, IDModel, TimestampModel): + workspace_id: int + created_by_id: uuid.UUID | None = None + next_scheduled_at: datetime | None = None + updated_at: datetime + model_config = ConfigDict(from_attributes=True) + +class PipelineRunRead(IDModel, TimestampModel): + pipeline_id: int + status: PipelineRunStatus + trigger: PipelineRunTrigger + documents_indexed: int | None = None + crawls_attempted: int | None = None + crawls_succeeded: int | None = None + error: dict[str, Any] | None = None + charged_micros: int | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + model_config = ConfigDict(from_attributes=True) + +class PipelineRunList(BaseModel): + items: list[PipelineRunRead] + total: int +``` + +**Validators:** + +- `schedule_cron` + `schedule_timezone` (when cron not None): validate together with the existing automations cron util `validate_cron(cron, timezone)` (`automations/triggers/builtin/schedule/cron.py:15`) so a bad expression **or** bad IANA timezone 422s at create, not at the Phase-6 tick. (`schedule_timezone` is consumed by Phase 6's `compute_next_fire_at`; default `"UTC"`.) +- The `save_to_kb` ⇄ `destination_folder_id` pairing and **connector eligibility** are validated in the **route** (they need a DB session: folder-in-workspace check + `is_pipeline_eligible(connector.connector_type)`), not in the pure schema validator — mirroring how `folders_routes` validates parent-in-search-space in the handler (`folders_routes.py:54–58`). + +### 6. Zero publication (`zero_publication.py`) + +Add to `ZERO_PUBLICATION` (`:71–82`), both **full-row**: + +```python + "pipelines": None, + "pipeline_runs": None, +``` + +Full-row (like `folders`/`search_source_connectors`) — no `_0_version` allowlist edit needed, no column-drift migrations later. (If a bulky column is ever added and must be excluded, switch that table to an explicit COLS list and handle the `_0_version` seam at `:106` then.) `verify_publication` will then expect both tables; the migration's `apply_publication` call (§4) reconciles them. + +> **Client sync is not live until the frontend lands (safe).** Publishing backend-side only makes the rows *available* to Zero; nothing syncs to a client until the (deferred) frontend Zero schema + permissions include these tables. Publishing now is harmless and matches the existing pattern (e.g. `automation_runs` was published by migration before/independent of its UI). This entry exists so the chat-agent run-history context (Phase 6) and a later Pipelines UI get push for free. + +### 7. Routes (`app/routes/pipelines_routes.py`, registered in `routes/__init__.py`) + +Follow `folders_routes` for authz (`check_permission` + `get_auth_context` + `get_async_session`) and `automations/api/run.py` for the run-list shape. Register the router in `routes/__init__.py` near `search_source_connectors_router` (`:105`). + +| Method + path | Behaviour | AuthZ | +|---------------|-----------|-------| +| `POST /pipelines` | Create. Validate: workspace access; if `connector_id`, load it, assert same workspace + `is_pipeline_eligible` (4xx otherwise); if `destination_folder_id`, assert folder in same workspace; `save_to_kb=true` allowed with NULL folder (→ root). Set `created_by_id = auth.user.id`. Persist; return `PipelineRead`. | `CONNECTORS_CREATE` | +| `GET /pipelines?workspace_id=` | List pipelines in a workspace. | `CONNECTORS_READ` | +| `GET /pipelines/{id}` | Read one. | `CONNECTORS_READ` | +| `PUT /pipelines/{id}` | Update (all-optional). Re-validate folder-in-workspace + cron if changed. | `CONNECTORS_UPDATE` | +| `DELETE /pipelines/{id}` | Delete (runs cascade). | `CONNECTORS_DELETE` | +| `POST /pipelines/{id}/run` | **Manual trigger.** Insert a `PipelineRun(status=pending, trigger=manual)`, enqueue the Phase-6 Celery task with the run id, return the run. Until Phase 6, the task is a stub (no-op/marks failed) — the row + enqueue path are what this phase delivers. (Concurrency control — block a second run while one is in flight — is deferred to Phase 6, reusing the connector indexing Redis locks `utils/indexing_locks.py`.) | `CONNECTORS_UPDATE` | +| `GET /pipelines/{id}/runs?limit=&offset=` | List runs newest-first, paginated (`limit` 1–200 default 50, `offset` ≥0) → `PipelineRunList`. Mirror `automations/api/run.py:13–30`. | `CONNECTORS_READ` | +| `GET /pipelines/{id}/runs/{run_id}` | Run detail. | `CONNECTORS_READ` | + +All handlers scope by the pipeline's `workspace_id` and 404 on cross-workspace access. A thin `PipelineService` (mirroring `automations` `RunService`) is optional; for MVP, inline session logic in the routes (as `folders_routes` does) is acceptable. + +## Periodic-indexing coexistence (resolves umbrella open item §2) + +**Decision: COEXIST for MVP — do NOT migrate connector periodic config into pipelines.** + +- The existing `SearchSourceConnector.periodic_indexing_enabled` / `next_scheduled_at` path and its meta-scheduler (`schedule_checker_task.py`) **stay fully functional and untouched** in Phases 5–6. Pipelines are **purely additive**: a new `pipelines.next_scheduled_at` scan (Phase 6) runs **alongside** the connector scan. +- **Rationale:** (a) lowest risk + smallest diff (no backfill of existing connector schedules into pipeline rows, no data migration of live schedules); (b) only the WebURL crawler is the MVP pipeline executor (Phase 6) — file sources keep using their connector-level periodic path; (c) consistent with the umbrella posture "DB migrations carry users; backend behaviour can change incrementally." +- **Known overlap (flagged for Phase 6, not solved here):** a single `WEBCRAWLER_CONNECTOR` could have BOTH connector-level periodic indexing AND a pipeline wrapping it → double crawl + **double bill**. The data model permits it; the *guard* is Phase 6's responsibility. **Recommendation for Phase 6:** when a pipeline is created/enabled over a connector, treat the **pipeline as authoritative** and set that connector's `periodic_indexing_enabled=False` (single scheduler owns each connector). Recorded here so Phase 6 implements the de-dup; Phase 5 only needs the columns to support either path. + +## Work items + +1. **Enums** `PipelineRunStatus` + `PipelineRunTrigger` in `db.py`. +2. **ORM** `Pipeline` + `PipelineRun` in `db.py`; inverse `pipelines` relationships on `Workspace`, `User`, `SearchSourceConnector`. +3. **Migration**: 2 types + 2 tables + indexes (incl. `ix_pipelines_due`) + `apply_publication`; symmetric downgrade. +4. **Schemas** `app/schemas/pipeline.py` (+ export in `schemas/__init__.py`); cron syntactic validator. +5. **Zero**: add `pipelines`/`pipeline_runs` (full-row) to `ZERO_PUBLICATION`. +6. **Routes** `pipelines_routes.py` (CRUD + `/run` + `/runs` list/detail); register in `routes/__init__.py`; reuse `CONNECTORS_*` permissions. +7. **Phase-6 task stub**: a named Celery task `run_pipeline(run_id)` in **`app/pipelines/tasks.py`** (the module Phase 6 fleshes out) that `/run` enqueues — no-op/marks the run `failed` with an "engine not implemented" error until Phase 6 fills it in (keeps the endpoint honest and testable). Register it so it's dispatchable: add `"app.pipelines.tasks"` to the Celery `include` list (`celery_app.py`). Phase 6 replaces the stub body in-place (same task name + module → `/run`'s import is stable) and adds the scheduler task + Beat entry + queue routing. +8. **Tests** (below). + +## Tests + +- **Migration round-trip**: upgrade creates both tables + both enum types + `ix_pipelines_due`; `zero_publication --verify` reports no mismatch (both tables published full-row); downgrade drops cleanly. +- **Create gating (cross-link to 04a)**: `POST /pipelines` with an `AVAILABLE` `DATA_SOURCE` connector (WebURL/GDrive/OneDrive/Dropbox) → 201; with a `MIGRATING`/`MCP_TOOL`/`DISABLED` connector → 4xx (`is_pipeline_eligible` false); with `connector_id=None` → allowed (Uploads case). **(Superseded in Phase 7: once the system-managed Uploads pipeline exists, user `POST` with `connector_id=None` is rejected 4xx and the Uploads row is auto-created on upload — see `07` §7. Update this test in Phase 7.)** +- **Folder validation**: `destination_folder_id` in another workspace → 4xx; `save_to_kb=true` with NULL folder → allowed. +- **Cron validation**: invalid `schedule_cron` → 422; NULL cron → allowed (manual-only). +- **Cascade**: deleting a connector deletes its pipelines + their runs; deleting a folder SET-NULLs `destination_folder_id` (pipeline survives); deleting a workspace removes pipelines + runs. +- **Manual run**: `POST /pipelines/{id}/run` inserts a `pending`/`manual` run and enqueues the task; `GET /pipelines/{id}/runs` returns it newest-first with correct `total`. +- **AuthZ**: cross-workspace access to any pipeline/run route → 404; missing `CONNECTORS_*` permission → 403. +- **Zero shape**: `expected_publication_shape` includes `pipelines`/`pipeline_runs` as full-row. + +## Risks / trade-offs + +- **`db.py` growth.** Adds two more models to an already-large module. Accepted for relationship/Zero/schema ergonomics + consistency with the other core tables; a later extraction to a `pipelines` package is additive. +- **Coexisting schedulers (double-bill window).** Until Phase 6 adds the de-dup guard, a connector with both its own periodic indexing and a wrapping pipeline can crawl/bill twice. Documented above; Phase 6 owns the fix. MVP exposure is small (operator-created overlap only). +- **Eligibility enforced in code, not SQL.** `is_pipeline_eligible` is a registry (04a) check at the route, so a direct DB insert could bypass it. Same trade-off 04a already accepted (no DB column for the taxonomy). +- **Pre-built unused columns.** `crawls_*` / `result_blob_key` / `charged_micros` ship empty in Phase 5. Deliberate — avoids a second migration in Phase 6. +- **`connector_id` CASCADE drops run history.** Deleting a connector erases its pipelines' audit trail. If audit retention becomes a requirement, switch to `SET NULL` + a discriminator to distinguish Uploads from orphaned (additive change). +- **`created_by_id` nullable shifts work to Phase 6.** SET-NULL preserves shared pipelines past member deletion (the goal) but means the run engine cannot assume a creator — it must resolve `created_by ?? workspace owner` for the indexing actor + billing target. Flagged in the Phase-6 handoff; harmless until then (no engine). +- **Static-registry strand.** A pipeline created over an `AVAILABLE` connector can later reference a type the registry flips to `MIGRATING` on a deploy (create-time check passed; the row persists). Phase 6 must re-check `is_pipeline_eligible` at run time and fail the run cleanly. The data model intentionally does not encode eligibility, so no migration is needed when the registry changes. + +## Out of scope (hand-offs) + +- Run **engine**, scheduling tick over `pipelines.next_scheduled_at`, crawl-billing wiring (write `charged_micros`/`crawls_*`), raw-blob persistence, and chat-agent run-history context → **Phase 6** (`06-pipelines-exec.md`). Phase 6 also owns three guards this model only *enables*: (a) **run-time eligibility re-check** (`is_pipeline_eligible` — a deploy can flip a connector type to `MIGRATING`, stranding existing pipelines → fail the run cleanly, don't crash); (b) **acting-user resolution** (`created_by_id` can be NULL after member deletion → resolve `created_by ?? workspace owner` for indexing + billing); (c) **concurrency** (Redis indexing lock per pipeline/connector). +- File upload creating/using an "Uploads" pipeline (`connector_id=NULL`, `trigger=upload`, always `save_to_kb`) + generalizing opt-in KB save → **Phase 7** (`07-upload-pipeline-kb.md`). +- Document→run provenance column → deferred (Phase 6/7 if needed). +- Frontend Pipelines UI (list/create/configure/run-history/manual run) → frontend umbrella. +- Public pay-as-you-go API over Type-1 pipelines → post-MVP (umbrella "Deferred"). diff --git a/plans/backend/06-pipelines-exec.md b/plans/backend/06-pipelines-exec.md new file mode 100644 index 000000000..c234ca9d0 --- /dev/null +++ b/plans/backend/06-pipelines-exec.md @@ -0,0 +1,347 @@ +# Phase 6 — Pipeline execution + scheduling (run engine, crawl billing, blob, chat context) + +> Part of the **CI Pivot MVP**. See `00-umbrella-plan.md` (Phase 6). +> Precondition: Phase 5 (`05-pipelines-model.md`) live — `pipelines` / `pipeline_runs` tables, schemas, Zero, CRUD + `/run` (currently enqueues a **stub** `run_pipeline(run_id)`). Depends on `03a-crawler-core.md` (`crawl_url` SUCCESS signal + `crawls_succeeded` counter) and `03c-crawl-billing.md` (`WebCrawlCreditService`). Sibling ahead: `07-upload-pipeline-kb.md` (uploads-as-pipeline). + +> **Implementation note (post-rename).** Citations use **today's** identifiers (`search_space_id` / `SearchSpace` / `searchspaces`). Phases 1–2 rename them to `workspace_id` / `Workspace` / `workspaces`; map accordingly. Locate code by **symbol/grep**, not the absolute line numbers cited here (the rename + Phases 3–5 shift them). + +## Objective + +Make pipelines actually **run**. Phase 5 shipped the data model + a stub task; this phase fills in: + +1. **Run engine** — turn a `PipelineRun` row into work: resolve the pipeline + connector, re-check eligibility, acquire a lock, crawl the source (WebURL crawler = the only MVP executor), optionally save to the KB destination folder, persist a raw-result blob otherwise, write the run record (status/timing/counts/error). +2. **Crawl billing wiring** (carry-over from `03c`) — bill `crawls_succeeded` to the **workspace owner** via `WebCrawlCreditService` for **both** KB-save branches (a `save_to_kb=false` run must not crawl for free); record `charged_micros` on the run for idempotency. +3. **Scheduling** — a Celery Beat tick that fires due `pipelines.next_scheduled_at` (cron), modeled on the **automations** schedule selector (cron + `FOR UPDATE SKIP LOCKED` + self-heal). Plus the **de-dup guard** Phase 5 deferred (a pipeline over a connector disables that connector's own periodic indexing). +4. **Chat-agent context** — a read-only main-agent **tool** (`get_pipeline_runs`) exposing recent run history scoped to the active workspace (umbrella default = tool). + +This is backend-only; the Pipelines UI is deferred to the frontend umbrella. + +## Locked decisions (MVP) + +| Concept | Decision | +|---------|----------| +| MVP executor | **WebURL crawler only.** `connector.connector_type == WEBCRAWLER_CONNECTOR`. Any other Type-1 type (GDrive/OneDrive/Dropbox) → run fails cleanly with `"executor not implemented for "` (their connector-level periodic path still works; pipelines over them are a Phase-7+ concern). `connector_id IS NULL` (Phase-7 Uploads) → not run by this engine (uploads register their own run in Phase 7). | +| Run-engine shape | Mirror **automations**: thin Celery task → `run_async_celery_task` → async `execute_pipeline_run(session, run_id)` (`automations/tasks/execute_run.py:19-33`). PENDING-gated + idempotent terminal no-op (`automations/runtime/executor.py:29-30`). Status lifecycle `pending → running → succeeded/failed` with `started_at`/`finished_at` (mirrors `repository.mark_running/mark_succeeded/mark_failed`). | +| KB-save path | **Reuse `index_crawled_urls`** (the battle-tested 2-phase indexer) for `save_to_kb=true`, extended with a `folder_id` param (set on created/updated `Document`s → destination folder). `save_to_kb=true` + NULL folder → root (existing behaviour; `documents.folder_id` nullable). | +| Non-KB path | `save_to_kb=false` → a **fetch-only** crawl loop (no `Document` rows): crawl each URL, collect `{url, content, metadata}`, persist as **one JSON blob** via `file_storage` backend, store `result_blob_key` on the run. (The 2-phase indexer is `Document`-centric and cannot "not persist"; a separate small loop is cleaner than bending it.) | +| Billing ownership | **Run engine owns billing for the pipeline path** (pre-check on `len(urls)` + charge on `crawls_succeeded` + record `charged_micros`), calling the crawler with **billing suppressed** (`bill=False`). The connector `/index` + connector-periodic paths keep `03c`'s in-indexer billing (`bill=True`). This keeps `charged_micros` run-idempotency clean and bills both KB branches identically. **(03c coordination — see below.)** | +| Scheduler | New `pipeline_schedule_select` Beat task modeled on `automations/triggers/builtin/schedule/selector.py` (cron, `FOR UPDATE SKIP LOCKED`, self-heal of NULL `next_scheduled_at`), **not** the simpler connector `frequency_minutes` checker — because Phase 5 chose `schedule_cron`. Reuses the existing `croniter` util (`automations/triggers/builtin/schedule/cron.py`). | +| Schedule timezone | Cron needs a timezone (`compute_next_fire_at(cron, timezone, …)`). Phase 5's model has only `schedule_cron`. **Add `schedule_timezone VARCHAR NOT NULL DEFAULT 'UTC'` to `pipelines`** (small additive amendment to `05` — see "Required 05 amendment"). Matches how automations store cron **and** timezone (`schedule/selector.py:86-89`). | +| Concurrency | Per-pipeline Redis lock + (when connector-backed) the **existing connector lock** `utils/indexing_locks.py` so a pipeline run and any residual connector index can't double-crawl the same connector. | +| Chat context | A main-agent **tool** `get_pipeline_runs` (registry pattern, `main_agent/tools/registry.py:85-102`), opening its own `shielded_async_session()` (like `KnowledgeTreeMiddleware`) and scoping to the build-time `search_space_id`. Always-on middleware injection is **deferred** (token cost; tool is on-demand). | +| Code location | New `app/pipelines/` package: `engine.py` (`execute_pipeline_run`), `tasks.py` (Celery `run_pipeline` + `pipeline_schedule_select`), `scheduler.py` (the tick), `storage.py` (blob key). The Phase-5 stub `run_pipeline` is **replaced** by the real task here; the `/run` route + scheduler enqueue it. (ORM stays in `db.py` per Phase 5; only the *engine* is a package, mirroring `app/automations/`.) | + +## Current state (cited) + +### Run-engine precedent (automations) + +- **Task wrapper**: `automations/tasks/execute_run.py:19-33` — `@celery_app.task(name=..., bind=True)` → `run_async_celery_task(lambda: _impl(run_id))`; `_impl` opens `get_celery_session_maker()()` and calls `execute_run(session, run_id)`, rolling back + re-raising on failure. +- **Launch**: `automations/dispatch/launch.py:43-60` — create the child row `status=PENDING`, `session.add` + `commit` + `refresh`, then `task.apply_async(args=[run.id], time_limit=…)`. (Phase 5's `/run` already does the create+enqueue; this phase makes the task real.) +- **Executor lifecycle**: `automations/runtime/executor.py:23-75` — `load run; if run.status != PENDING: return` (idempotent terminal no-op, `:29-30`); `mark_running` + `commit` (`:46-47`); on bad snapshot → `mark_failed` + commit (`:35-44`); terminal `mark_succeeded`/`mark_failed` each `commit`. + +### The MVP executor: `webcrawler_indexer.index_crawled_urls` + +- Signature `webcrawler_indexer.py:44-53`: `(session, connector_id, search_space_id, user_id, start_date=None, end_date=None, update_last_indexed=True, on_heartbeat_callback=None) -> (int, str|None)`. +- Reads URLs from `connector.config["INITIAL_URLS"]` via `parse_webcrawler_urls` (`:118-119`). +- **Creates `Document` rows with NO `folder_id`** (`:222-239`) → today everything lands at workspace root. Updates existing docs in place. Crawls via `crawler.crawl_url(url)` (`:297`; `03a` changes this signature/return). Tracks `documents_indexed` / `documents_updated` / counts; final commit at `:432`; returns `(total_processed, error)` (`:477`). +- **No billing today** — `03c` adds the owner pre-check + `charge_credits` inside this function (`03c` §2). **This phase adds a `bill: bool = True` seam** so the pipeline path can suppress it (plus `folder_id` / `urls` params + a `stats` out-param — see "Crawler changes"; the positional return is left unchanged so the shared wrapper keeps working). +- Connector entrypoint chain (the path the pipeline engine **bypasses**, because it's notification-centric, not run-centric): `connector_tasks.py:418-454` (`index_crawled_urls_task` → `_index_crawled_urls` → `run_web_page_indexing`) → a `_run_indexing_with_notifications(...)` **call site** (e.g. the WebCrawler one at `search_source_connectors_routes.py:~2524`). The wrapper itself is **defined at `:1296`** (acquires the connector lock at `:1334`, length-unpacks the indexer return at `:1503`). + +### KB-ingest service (folder support already exists there) + +- `IndexingPipelineService` (`indexing_pipeline/indexing_pipeline_service.py:82`) already threads a destination folder via `ConnectorDocument.folder_id` (`:273-274`, `:301-302`, `:326`). **But the webcrawler indexer does not use this service** — it writes `Document`s directly. So the umbrella's "route through `IndexingPipelineService` into the destination folder" is accurate for **file/Composio** sources, **not** the crawler. For the WebURL MVP, the minimal correct move is to thread `folder_id` into `index_crawled_urls` (not refactor the crawler onto `IndexingPipelineService`). Unifying them is a post-MVP cleanup. + +### Scheduler precedent (cron) + Beat registration + +- **Selector** `automations/triggers/builtin/schedule/selector.py`: tick (`:47-65`) → self-heal NULL `next_fire_at` (`:68-98`) → claim due rows `FOR UPDATE SKIP LOCKED` + advance via `compute_next_fire_at` + set `last_fired_at` (`:101-150`) → `_start_one` launches each (`:153-183`). `_TICK_BATCH=200` cap (`:35`). +- **Cron util** `automations/triggers/builtin/schedule/cron.py`: `validate_cron(cron, timezone)` (`:15`) and `compute_next_fire_at(cron, timezone, *, after)` (`:28`, returns UTC). **Requires a timezone** → drives the 05 amendment. +- **Beat source** `automations/triggers/builtin/schedule/source.py:14-20` — a `BEAT_SCHEDULE` dict (`crontab(minute="*")`) merged into the app at `celery_app.py:323` (`**SCHEDULE_BEAT_SCHEDULE`). The app's beat dict is `celery_app.py:268-324`; the worker `include` list is `:180-198`; task routing (`index_crawled_urls → CONNECTORS_QUEUE`) is `:237-255` (`:246`). +- **Connector meta-scheduler we coexist with**: `schedule_checker_task.py:17` scans `SearchSourceConnector` where `periodic_indexing_enabled AND next_scheduled_at <= now` (`:33-39`), with a Redis-lock guard `is_connector_indexing_locked` (`:93`). Untouched; the pipeline tick is a sibling. + +### Locks, storage, chat tooling + +- **Locks** `utils/indexing_locks.py`: `acquire_connector_indexing_lock(connector_id)->bool` (`:24`), `release_…` (`:37`), `is_…_locked` (`:43`), keyed `indexing:connector_lock:{id}`, TTL `CONNECTOR_INDEXING_LOCK_TTL_SECONDS`. +- **Storage** `file_storage/service.py:39,46`: `get_storage_backend()` (from `file_storage/factory.py`) → `backend.put(key, data, content_type=…)` / `open_stream(key)` / `delete(key)`. Keys are built per-purpose (`file_storage/keys.py:11`, `build_document_file_key`). For runs we add a sibling key builder (no `document_files` row — runs aren't documents). +- **Tool registry** `main_agent/tools/registry.py:85-102` — `_MAIN_AGENT_TOOL_FACTORIES: {name: (factory, required_dep_names)}`; `build_main_agent_tools(deps, …)` resolves deps + builds (`:105-147`). Display metadata lives in `shared/tools/catalog`; the main-agent name list in `main_agent/tools/index` (per the registry docstring `:1-14`). Read-only DB pattern for a chat surface: `KnowledgeTreeMiddleware` opens `shielded_async_session()` and filters by its build-time `search_space_id` (`knowledge_tree/middleware.py:199-206`). +- **Workspace owner** (billing target): `SearchSpace.user_id` (`03c` §2.1 resolves the owner with a direct `select SearchSpace.user_id where id == search_space_id`). + +## Target design + +### 1. `execute_pipeline_run(session, run_id)` — the engine (`app/pipelines/engine.py`) + +``` +load run (pipeline + connector eager); if run is None: raise +if run.status != PENDING: return # status gate; also no-ops acks_late redelivery (executor.py:29-30) +pipeline = run.pipeline +mark_running(run): status=RUNNING, started_at=now; commit +# ^ committed BEFORE any crawl/charge → a redelivered/duplicate task sees RUNNING and returns (executor.py:46-47) + +pipeline_lock = connector_lock = False +connector = None +try: + # --- guards: fail cleanly, never crash the worker (Phase-5 hand-offs (a)/(b)) --- + if pipeline.connector_id is None: # Uploads pipeline — Phase 7, not here + return fail(run, "uploads pipeline has no fetch executor") + connector = pipeline.connector + if not is_pipeline_eligible(connector.connector_type): # 04a runtime re-check (strand guard) + return fail(run, f"connector type {connector.connector_type} no longer eligible") + if connector.connector_type != WEBCRAWLER_CONNECTOR: + return fail(run, f"executor not implemented for {connector.connector_type}") + + owner_id = workspace_owner_id(session, pipeline.workspace_id) # select Workspace.user_id (03c §2.1) + actor_user_id = str(pipeline.created_by_id or owner_id) # Phase-5 hand-off (b) + urls = parse_webcrawler_urls(pipeline.config.get("INITIAL_URLS") or connector.config.get("INITIAL_URLS")) + if not urls: + return fail(run, "no URLs to crawl") + + # --- concurrency: take both; release only what we took (Phase-5 hand-off (c)) --- + pipeline_lock = acquire_pipeline_indexing_lock(pipeline.id) + if not pipeline_lock: + return fail(run, "a run for this pipeline is already in progress") + connector_lock = acquire_connector_indexing_lock(connector.id) + if not connector_lock: + return fail(run, "connector is currently indexing; retry later") + + # --- billing pre-flight (run-level; 03c statics). len(urls) is a safe upper bound. --- + svc = WebCrawlCreditService(session) + if WebCrawlCreditService.billing_enabled() and run.charged_micros is None: + await svc.check_credits(owner_id, len(urls)) # InsufficientCreditsError → except → fail("out of crawl credit") + + # --- fetch --- + if pipeline.save_to_kb: + stats = {} # out-param sink — NON-breaking (see §3) + await index_crawled_urls(session, connector.id, pipeline.workspace_id, actor_user_id, + urls=urls, folder_id=pipeline.destination_folder_id, + bill=False, stats=stats, update_last_indexed=True) + documents_indexed = stats["documents_indexed"] + crawls_attempted = stats["crawls_attempted"] + crawls_succeeded = stats["crawls_succeeded"] + result_blob_key = None + else: + outcomes = await crawl_urls_fetch_only(urls) # list[dict] from CrawlOutcome (03a) + crawls_attempted = len(urls) + crawls_succeeded = sum(1 for o in outcomes if o["status"] == "success") + documents_indexed = 0 + result_blob_key = await persist_run_blob(run, outcomes) + + # --- charge: stamp run + add audit, THEN charge (its single commit flushes all three atomically) --- + if WebCrawlCreditService.billing_enabled() and crawls_succeeded > 0 and run.charged_micros is None: + run.charged_micros = WebCrawlCreditService.successes_to_micros(crawls_succeeded) + record_token_usage(session, usage_type="web_crawl", workspace_id=pipeline.workspace_id, + user_id=owner_id, cost_micros=run.charged_micros, + call_details={"urls": len(urls), "successes": crawls_succeeded, + "pipeline_id": pipeline.id, "run_id": run.id}, + message_id=None) # add only — does not commit (03c §2.3) + await svc.charge_credits(owner_id, crawls_succeeded) # debits + COMMITS run+audit+balance (03c §2.4) + + mark_succeeded(run): status=SUCCEEDED, finished_at=now, + documents_indexed, crawls_attempted, crawls_succeeded, result_blob_key; commit +except InsufficientCreditsError: + return fail(run, "out of crawl credit") # run is RUNNING→FAILED; no debit happened +except Exception as e: + return fail(run, {"message": str(e), "type": type(e).__name__}) +finally: + if connector_lock: release_connector_indexing_lock(connector.id) + if pipeline_lock: release_pipeline_indexing_lock(pipeline.id) +``` + +`fail(run, err)` = set `status=FAILED`, `finished_at=now`, `error={...}`, `commit`, return — so the `finally` still runs and releases only the locks actually taken. `connector.id` in `finally` is reached only when `connector_lock is True`, which implies `connector` was resolved. **Idempotency rests on the status gate**, not `charged_micros`: `mark_running` commits `RUNNING` before any crawl/charge, so an `acks_late` redelivery (`celery_app.py:228-229`) or a manual re-enqueue sees `RUNNING`/terminal and returns. `charged_micros` is the belt-and-suspenders guard reserved for a future stuck-`RUNNING` re-driver (none in MVP). + +Notes: +- **`workspace_owner_id`** = `select Workspace.user_id where id == workspace_id` (`03c` §2.1 pattern). +- **`is_pipeline_eligible`** is the 04a registry helper — re-checked here at run time (a deploy can flip a type to `MIGRATING`; the create-time check in Phase 5 doesn't protect the row forever). +- **URL source**: per-pipeline `config["INITIAL_URLS"]` overrides the connector's (the `config` JSONB "per-pipeline overrides" Phase 5 reserved); fall back to the connector's `INITIAL_URLS`. + +### 2. Celery tasks (`app/pipelines/tasks.py`) + +```python +@celery_app.task(name="run_pipeline", bind=True) +def run_pipeline(self, run_id: int) -> None: + return run_async_celery_task(lambda: _run(run_id)) # execute_run.py:19-22 shape + +async def _run(run_id: int) -> None: + async with get_celery_session_maker()() as session: + try: + await execute_pipeline_run(session, run_id) + except Exception: + logger.exception("pipeline run %d failed unexpectedly", run_id) + await session.rollback(); raise +``` + +- `run_pipeline` **replaces** the Phase-5 stub of the same name; Phase 5's `/run` route already enqueues `run_pipeline(run.id)` (no route change needed). +- **Routing/registration**: ensure `"app.pipelines.tasks"` is in the `include` list (`celery_app.py:180-198`) — Phase 5 already added it for the stub; no-op if present — and route `"run_pipeline": {"queue": CONNECTORS_QUEUE}` (it crawls — same queue as `index_crawled_urls`, `celery_app.py:246`). Give the `apply_async`/`delay` a `time_limit` (the connectors queue already has an 8h hard cap, `celery_app.py:219`). + +### 3. Crawler changes — `index_crawled_urls` (coordination with `03a`/`03c`) + +Four additive params, **all default-compatible** with the existing connector path: + +1. **`folder_id: int | None = None`** — set on each created `Document` (`:222-239`) and on the existing-doc update branch, so KB-save lands in the destination folder. Default `None` = root (today's behaviour). +2. **`urls: list[str] | None = None`** — when provided, crawl this list instead of re-reading `connector.config["INITIAL_URLS"]` (`:118-119`). **Required** for the per-pipeline `config["INITIAL_URLS"]` override (a Phase-5 `config` capability) and to keep the engine's `len(urls)` billing pre-check consistent with what's actually crawled. Default `None` = read connector config (today's behaviour). +3. **`bill: bool = True`** — gate `03c`'s in-indexer `check_credits` + `charge_credits` + `record_token_usage`. The pipeline engine calls `bill=False` and bills at the run level; the connector `/index` + periodic paths keep `bill=True`. *(Small addition to `03c`'s §2 wiring — see "Cross-plan coordination".)* +4. **`stats: dict | None = None`** — an **out-param sink**, NOT a return-shape change. When provided, the indexer populates `stats["documents_indexed"|"crawls_attempted"|"crawls_succeeded"]`. **Do NOT widen the return tuple:** the shared `_run_indexing_with_notifications` wrapper unpacks every indexer's return **by length** (`search_source_connectors_routes.py:1499-1507`: `if len(result) == 3: a,b,c = result else: a,b = result`), so a 4-tuple would raise `ValueError` for *every* connector. The 2-/3-tuple return stays exactly as-is (wrapper untouched); the engine passes a fresh `stats` dict and reads it after the call (the wrapper passes none → `None` → no-op). `03a` work-item 3 already reflects this (it exposes `crawls_succeeded` via **task metadata** only, with this `stats` sink owned by Phase 6 — not via the positional return). + +The engine **bypasses** `_run_indexing_with_notifications` (it acquires the connector lock itself at `:1334` — going through the wrapper would self-deadlock on that same lock, and the wrapper's connector-`Notification` UI is the wrong surface for a pipeline run, which has its own `PipelineRun` record). It calls `index_crawled_urls` directly with no heartbeat callback. (Consequence: pipeline runs don't emit connector-indexing notifications and aren't touched by `cleanup_stale_indexing_notifications` — fine; run status lives on `pipeline_runs`.) + +### 4. Fetch-only path + blob (`app/pipelines/engine.py` + `app/pipelines/storage.py`) + +For `save_to_kb=false`: + +```python +async def crawl_urls_fetch_only(urls): + crawler = WebCrawlerConnector() # 03a: no firecrawl ctor arg + out = [] + for url in urls: + outcome = await crawler.crawl_url(url) # 03a: CrawlOutcome(status, result, error, tier) + out.append({ + "url": url, + "status": outcome.status.value, # CrawlOutcomeStatus: success | empty | failed + "content": (outcome.result or {}).get("content", ""), + "metadata": (outcome.result or {}).get("metadata", {}), + "error": outcome.error, + }) + return out +``` + +> **03a return shape.** `03a` is **committed to the `CrawlOutcome` dataclass** (`status`/`result`/`error`/`tier`) — the tuple option was dropped precisely because this fetch-only path consumes `.status`/`.result`/`.error` as attributes. So the unpack above is the firm contract; no adaptation needed. + +```python +# storage.py +def build_pipeline_run_result_key(*, workspace_id: int, run_id: int) -> str: + return f"pipeline_runs/{workspace_id}/{run_id}/result.json" + +async def persist_run_blob(run, outcomes) -> str: + key = build_pipeline_run_result_key(workspace_id=run.pipeline.workspace_id, run_id=run.id) + data = json.dumps({"runs": outcomes}).encode() + await get_storage_backend().put(key, data, content_type="application/json") + return key # → run.result_blob_key +``` + +- One blob per run (single configured backend, like document files). `result_blob_key` (Phase 5 column) stores the key; reads resolve via `get_storage_backend()`. +- **Billing still applies** here — `crawls_succeeded` is counted from `outcomes`, so non-KB runs cost exactly the same as KB runs (umbrella line 105). This is the whole reason the run engine, not the indexer, owns pipeline billing. + +### 5. Scheduler tick (`app/pipelines/scheduler.py` + `tasks.py`) + +A near-copy of the automations selector, retargeted at `pipelines`: + +```python +@celery_app.task(name="pipeline_schedule_select") +def pipeline_schedule_select() -> None: + return run_async_celery_task(_tick) + +async def _tick(): + async with get_celery_session_maker()() as session: + now = datetime.now(UTC) + await _self_heal_null_next(session, now) # enabled + cron + next_scheduled_at IS NULL → compute + claims = await _claim_due(session, now) # enabled + next_scheduled_at <= now, FOR UPDATE SKIP LOCKED, advance + for c in claims: + await _start_run(session, c) # create PipelineRun(trigger=scheduled, status=pending) + run_pipeline.apply_async +``` + +- **`_claim_due`**: `select Pipeline where enabled AND connector_id IS NOT NULL AND schedule_cron IS NOT NULL AND next_scheduled_at IS NOT NULL AND next_scheduled_at <= now ORDER BY next_scheduled_at LIMIT 200 FOR UPDATE SKIP LOCKED` (uses Phase 5's partial `ix_pipelines_due`). The `connector_id IS NOT NULL` filter is **defense-in-depth** so a non-connector pipeline (the Phase-7 Uploads pipeline) can **never** be scheduled into perpetually-failing runs even if a cron is somehow set on it — the engine would `fail` every such run (`§1`); this stops them at the selector. (Phase 7 also guards the API; this is the belt to that suspenders.) Advance `next_scheduled_at = compute_next_fire_at(schedule_cron, schedule_timezone, after=now)`; on `InvalidCronError` → `enabled=False` (self-disable, like `selector.py:90-96,131-137`). Commit, then create the run + enqueue (mirror `_start_one` reload-after-commit, `selector.py:153-183`). +- **`_self_heal_null_next`**: backfills `next_scheduled_at` for enabled cron pipelines missing it (fresh inserts, restored rows) — so Phase 5's create can leave it NULL and the tick fills it within ≤1 min. +- **Duplicate suppression**: in `_start_run`, skip creating a scheduled run if the pipeline already has a non-terminal run (`status IN (pending, running)`) — mirrors the connector checker's in-progress guard (`schedule_checker_task.py:100-116`). Without it, a pipeline whose run outlasts its cron interval piles up `scheduled` runs that immediately `fail` on the lock (noisy, though harmless). `next_scheduled_at` is still advanced (the fire isn't retried — `catchup=False`, like the automations selector). +- **Beat entry**: add `"pipeline-schedule-select": {"task": "pipeline_schedule_select", "schedule": crontab(minute="*"), "options": {"expires": 50}}` to `celery_app.py:268-324` (default fast queue, like `check_periodic_schedules`). (`"app.pipelines.tasks"` is already in `include` from Phase 5's stub, so `pipeline_schedule_select` in the same module is auto-registered.) +- **`schedule_cron` edits** (Phase 5 `PUT`): when `schedule_cron` (or timezone) changes, **null `next_scheduled_at`** so the tick recomputes (small `PUT`-handler addition — see Required 05 amendment). NULL cron ⇒ manual-only (tick ignores it). + +### 6. De-dup guard (resolves Phase-5 deferred double-bill) + +When a pipeline that **wraps a connector** is created or enabled (Phase 5 routes), set that connector's `periodic_indexing_enabled = False` (+ `next_scheduled_at = None`) so a single scheduler owns each connector (the **pipeline is authoritative**, per Phase 5's recommendation). Two touch points: + +- Phase 6 owns the rule; implement it in the Phase-5 `POST /pipelines` and `PUT /pipelines/{id}` handlers (when `connector_id` is set and `enabled`). This is a tiny handler addition (Phase 5 explicitly handed it here). +- Belt-and-suspenders: the run engine already takes the **connector lock** during a run, so even if both scan, only one crawl proceeds at a time (the connector checker skips on a held lock, `schedule_checker_task.py:93`). + +### 7. Chat-agent run-history tool (`main_agent/tools/pipeline_runs.py`) + +A read-only tool registered in the main-agent registry (umbrella default = tool): + +```python +def create_get_pipeline_runs_tool(*, search_space_id: int) -> BaseTool: + @tool + async def get_pipeline_runs(pipeline_id: int | None = None, limit: int = 20) -> str: + """Recent pipeline run history for THIS workspace: name, status, trigger, + schedule, counts, started/finished. Read-only.""" + async with shielded_async_session() as session: # knowledge_tree/middleware.py:199 pattern + q = (select(PipelineRun, Pipeline.name, Pipeline.schedule_cron) + .join(Pipeline, PipelineRun.pipeline_id == Pipeline.id) + .where(Pipeline.workspace_id == search_space_id)) + if pipeline_id is not None: + q = q.where(Pipeline.id == pipeline_id) + rows = (await session.execute( + q.order_by(PipelineRun.created_at.desc()).limit(min(limit, 100)))).all() + return render_compact(rows) # name • status • trigger • finished_at • docs/crawls + return get_pipeline_runs +``` + +- **Registration**: add `"get_pipeline_runs": (_build_get_pipeline_runs_tool, ("search_space_id",))` to `_MAIN_AGENT_TOOL_FACTORIES` (`registry.py:85-102`); add the name to `main_agent/tools/index`; add display metadata to `shared/tools/catalog`. +- **Workspace scope is structural** — the tool only ever sees its build-time `search_space_id`; no cross-workspace leakage (same guarantee `KnowledgeTreeMiddleware` relies on). +- **Anonymous / no-workspace turns**: exclude the tool when there's no `search_space_id` (anonymous chat has no workspace/pipelines), the same way `KnowledgeTreeMiddleware` is CLOUD-only (`knowledge_tree/middleware.py:130-131`). The registry factory requires `search_space_id`, so just don't enable it in the anonymous tool set. +- The tool reads `pipelines`/`pipeline_runs` directly (no Zero dependency), so it works server-side immediately even though client Zero sync is dormant until the frontend lands (Phase 5 note). +- **Deferred**: an always-on `` middleware injection (à la `KnowledgeTreeMiddleware`) — adds per-turn token cost; revisit if the agent under-uses the tool. + +### 8. (Optional) blob read endpoint + +Phase 5 stores `result_blob_key` but exposes no reader. Add `GET /pipelines/{id}/runs/{run_id}/result` → stream the blob via `get_storage_backend().open_stream(run.result_blob_key)` (404 if NULL), `CONNECTORS_READ`, workspace-scoped — mirroring `file_storage/api.py:69-90`. Lets API consumers + the chat tool fetch non-KB results. (Small; include if cheap, else defer to the public-API phase.) + +## Required `05` amendment (timezone + recompute) + +Small, additive. **Items 1–3 (the `schedule_timezone` column) are already folded into `05`** (model/migration/schema/validator) — listed here for traceability. **Item 4 is Phase-6 route logic** added on top of Phase 5's handlers (Work items 5–6): + +1. *(done in `05` §2)* **Model**: `schedule_timezone = Column(String(64), nullable=False, default="UTC", server_default="UTC")` on `Pipeline`. +2. *(done in `05` §4)* **Migration**: `schedule_timezone VARCHAR(64) NOT NULL DEFAULT 'UTC'` in `CREATE TABLE pipelines`. +3. *(done in `05` §5)* **Schema**: `schedule_timezone: str = "UTC"` on `PipelineBase` + `PipelineUpdate`; `(schedule_cron, schedule_timezone)` validated together via `validate_cron(cron, tz)` (`cron.py:15`). +4. **Routes** (Phase 6 adds to `05` §7 handlers): on `PUT` when `schedule_cron`/`schedule_timezone` changes, set `next_scheduled_at = None` (tick self-heals). On `POST`/`PUT` with `connector_id` + `enabled`, apply the §6 de-dup guard. + +(If we'd rather not touch `05`: hardcode `"UTC"` everywhere in Phase 6 and skip the column — but that loses operator-meaningful local schedules and diverges from automations. The column is the right call.) + +## Cross-plan coordination + +- **`03a`** — Phase 6 calls `WebCrawlerConnector.crawl_url(url)` directly in the fetch-only path and relies on `03a`'s `CrawlOutcome.status` (`SUCCESS`/`EMPTY`/`FAILED`, dataclass form — already locked in `03a`). `03a` work-item 3 is **already aligned**: it surfaces the `crawls_succeeded` counter via **task metadata** only (not by widening `index_crawled_urls`'s positional return — the shared `_run_indexing_with_notifications` wrapper unpacks by tuple length, `:1499-1507`). Phase 6 owns the `folder_id`/`urls`/`stats` params; `03a` owns the counter + `CrawlOutcome`. +- **`03c`** — add the **`bill: bool = True`** parameter to the in-indexer billing (so the pipeline path can suppress it). `03c`'s `WebCrawlCreditService` statics (`billing_enabled()`, `successes_to_micros(n)`, `check_credits`, `charge_credits`) are reused verbatim by the run engine. `03c`'s risk note already anticipated this: *"Phase 5's `PipelineRun` can persist `charged_micros` for stronger idempotency"* — Phase 6 is where that lands. +- **`05`** — the four amendments above; `run_pipeline` task name + `/run` enqueue already exist (stub → real); `charged_micros`/`crawls_*`/`result_blob_key` columns already exist (pre-built in Phase 5). +- **`04a`** — `is_pipeline_eligible(connector_type)` is imported for the run-time re-check. + +## Work items + +1. **`app/pipelines/` package**: `engine.py` (`execute_pipeline_run` + `mark_running/succeeded/fail` helpers + `workspace_owner_id` + `crawl_urls_fetch_only`), `tasks.py` (`run_pipeline`, `pipeline_schedule_select`), `scheduler.py` (`_tick`/`_self_heal_null_next`/`_claim_due`/`_start_run`), `storage.py` (blob key + `persist_run_blob`). +2. **Pipeline lock helpers**: add `acquire_pipeline_indexing_lock(id)` / `release_pipeline_indexing_lock(id)` to `utils/indexing_locks.py` (key `indexing:pipeline_lock:{id}`), mirroring the connector lock (`:19-40`). +3. **Crawler changes**: `index_crawled_urls` gains `folder_id` + `urls` + `bill` + a `stats` **out-param** (populates `documents_indexed`/`crawls_attempted`/`crawls_succeeded`). **No return-shape change → no caller changes** (the shared wrapper's 2-/3-tuple unpack stays valid). Coordinate with `03a`/`03c` (see "Cross-plan coordination"). +4. **Celery wiring** (`celery_app.py`): ensure `"app.pipelines.tasks"` is in `include` (added in Phase 5 for the stub); route `run_pipeline → CONNECTORS_QUEUE`; add the `pipeline-schedule-select` Beat entry. +5. **De-dup guard** in Phase-5 `POST`/`PUT /pipelines` (disable wrapped connector's periodic indexing). +6. **`05` amendment**: `schedule_timezone` column/migration/schema/validator + `next_scheduled_at` reset on cron edit. +7. **Chat tool** `get_pipeline_runs` + registry/index/catalog registration. +8. **(Optional)** `GET …/runs/{run_id}/result` blob reader. +9. **Tests** (below). + +## Tests + +- **Manual run, KB-save**: WebURL pipeline `save_to_kb=true` + `destination_folder_id` → run goes `pending→running→succeeded`, `Document`s created **in that folder**, `documents_indexed`/`crawls_*` set, `finished_at` set. +- **Manual run, non-KB**: `save_to_kb=false` → **no** `Document`s, `result_blob_key` set + blob readable, `crawls_*` set. Billing **still** charges `crawls_succeeded` (parity with KB run). +- **Billing**: enabled + sufficient → owner debited `crawls_succeeded * 1000`, `charged_micros` stamped, one `web_crawl` `TokenUsage`; **insufficient → run fails pre-crawl (FAILED), no debit**; disabled (self-hosted) → no charge, no `web_crawl` row, `charged_micros` stays NULL. +- **Idempotency / lifecycle**: re-enqueuing a `RUNNING` or terminal run → immediate no-op (`status != PENDING` gate) → **no second charge** (the primary guard; `mark_running` commits before any charge); engine never raises out of `fail()` paths (bad type / NULL connector / no URLs / ineligible / lock contention) — all land the run in `FAILED` and release locks. +- **Eligibility re-check**: pipeline over a connector whose type is now `MIGRATING` → run fails cleanly (not crash). +- **Scheduler**: a due cron pipeline → tick creates a `scheduled` run + advances `next_scheduled_at` to the next cron match (in `schedule_timezone`); NULL `next_scheduled_at` self-heals; invalid cron self-disables (`enabled=False`); manual-only (NULL cron) never fires. +- **De-dup**: creating/enabling a pipeline over a connector flips that connector's `periodic_indexing_enabled=False`; concurrent connector-checker tick skips on the held lock. +- **Concurrency**: a second run while one is in-flight → fails with "already in progress" (locks held). +- **Chat tool**: `get_pipeline_runs` returns this workspace's runs only; another workspace's runs never appear; respects `limit`. +- **Counts contract (regression-critical)**: `index_crawled_urls` keeps its existing 2-/3-tuple return (the shared `_run_indexing_with_notifications` length-unpack still works for *all* connectors); when a `stats` dict is passed it's populated with `documents_indexed`/`crawls_attempted`/`crawls_succeeded`; `bill=False` suppresses in-indexer billing; connector path (`bill=True`, no `stats`) behaves byte-for-byte as before. Add a wrapper test asserting webcrawler indexing through `_run_indexing_with_notifications` still unpacks cleanly. + +## Risks / trade-offs + +- **Billing split across two call sites** (indexer for connector path, run engine for pipeline path), unified only by `03c`'s shared statics. Accepted: it's what makes `charged_micros` run-idempotency clean and keeps both KB branches billed identically. The `bill` flag is the single seam. +- **Stuck `RUNNING` runs.** A crash *after* `mark_running` but *before* terminal leaves `status=RUNNING` forever (the `status != PENDING` gate then no-ops any redelivery — same property as automations). No double-charge (gate + `charged_micros`), but the run never completes and its locks rely on Redis TTL to free. A stale-run sweeper (à la `cleanup_stale_indexing_notifications`, `celery_app.py:279-285`) is deferred; MVP exposure is a rare worker crash. +- **Two crawl code paths** (`index_crawled_urls` for KB vs `crawl_urls_fetch_only` for non-KB). Minor duplication of the crawl loop; deliberate, since forcing the `Document`-centric 2-phase indexer into a "don't persist" mode is messier. Both share `WebCrawlerConnector.crawl_url` + the same billing math. +- **WebURL-only executor.** A pipeline over GDrive/OneDrive/Dropbox is creatable (Phase 5 lets `AVAILABLE DATA_SOURCE`) but fails at run time ("executor not implemented"). Acceptable for MVP (those keep their connector-periodic path); generalize the executor in Phase 7+. +- **De-dup guard is one-way.** Creating/enabling a pipeline disables the wrapped connector's `periodic_indexing_enabled`, but **disabling or deleting that pipeline does NOT restore it** — so a connector can end up indexed by *neither* path (data goes stale) until the user re-enables connector periodic indexing manually. MVP-acceptable (operator action); a symmetric restore-on-disable is a small follow-up. Flag in the (deferred) UI copy. +- **Blob backend assumption.** `result_blob_key` stores only the key (no backend name, unlike `document_files.storage_backend`). Fine for the single configured backend; if multi-backend ever lands, add a backend column to `pipeline_runs` (additive). +- **Non-KB blob built in memory.** `crawl_urls_fetch_only` accumulates every URL's extracted content into one JSON blob before upload — a large URL list (or huge pages) inflates worker memory. Acceptable for MVP URL counts; if it bites, stream to the backend or cap per-run URLs (a `config` knob). Also: non-KB runs have **no dedup** (every run re-crawls + re-bills all URLs) — that's inherent to "don't persist", and the user opted out of KB. +- **Lock lifetime vs run length.** The Redis locks use `CONNECTOR_INDEXING_LOCK_TTL_SECONDS`; a crawl run longer than the TTL could let a second run start. Pre-existing property of the connector lock (the connectors queue caps at 8h); revisit the TTL if pipeline runs routinely exceed it. +- **Two minute-level Beat scans** (connector checker + pipeline tick) now run every minute. Both are indexed (`ix_pipelines_due` partial index; connector scan on its own predicate) + batch-capped (200), so cost is negligible. + +## Out of scope (hand-offs) + +- **Uploads pipeline execution** (`connector_id=NULL`, `trigger=upload`, always `save_to_kb`) → **Phase 7** (`07-upload-pipeline-kb.md`). This engine explicitly fails NULL-connector runs. +- **Non-WebURL executors** (GDrive/OneDrive/Dropbox pipelines) → Phase 7+. +- **Always-on chat context middleware** (`` injection) → deferred (tool-only for MVP). +- **Document→run provenance** (`documents.pipeline_run_id`) → deferred (touches `documents`). +- **Public pay-as-you-go API over pipelines** + **public MCP KB server** → post-MVP (umbrella "Deferred"). +- **Frontend** Pipelines UI (run-history, manual run, schedule editor) → frontend umbrella. diff --git a/plans/backend/07-upload-pipeline-kb.md b/plans/backend/07-upload-pipeline-kb.md new file mode 100644 index 000000000..2685ea304 --- /dev/null +++ b/plans/backend/07-upload-pipeline-kb.md @@ -0,0 +1,192 @@ +# 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).** Citations use **today's** identifiers (`search_space_id` / `SearchSpace`). Phases 1–2 rename them to `workspace_id` / `Workspace`; map accordingly. 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 `Document`s + 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=, 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 `Document`s). 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` + +```python +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` + +```python +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: + +```python +if files_to_process: # only when something was actually queued + await record_upload_run(session, workspace_id=search_space_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`): + +```python +await record_upload_run(session, workspace_id=search_space_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 + +```sql +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=` 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.