openapi: 3.0.3
info:
  title: faucet serve — HTTP control plane
  description: >
    The `faucet serve` REST API (#127). Submit pipeline configs over HTTP, poll
    run status, list/cancel/delete runs, and stream per-run logs over SSE, plus
    unauthenticated health and Prometheus endpoints. All `/v1/*` endpoints
    require a bearer token unless the server was started with `--no-auth`.

    This spec is kept in two-way sync with the live router by
    `cli/tests/serve_openapi.rs` (every documented path is routable; every route
    is documented).
  version: "1.0.0"
servers:
  - url: http://127.0.0.1:8080
    description: Default loopback bind
security:
  - bearerAuth: []
paths:
  /v1/runs:
    post:
      summary: Submit a pipeline run
      description: >
        Validate + interpolate the config synchronously (4xx on error), then
        queue the run and return immediately. Long-running execution is
        asynchronous — poll `GET /v1/runs/{id}`.
      operationId: submitRun
      requestBody:
        required: true
        content:
          application/json:
            schema: { $ref: "#/components/schemas/SubmitRequest" }
      responses:
        "202":
          description: Run queued (or an idempotency replay of an existing run).
          content:
            application/json:
              schema: { $ref: "#/components/schemas/SubmitResponse" }
        "400": { $ref: "#/components/responses/Error" }
        "401": { $ref: "#/components/responses/Error" }
        "409": { $ref: "#/components/responses/Error" }
        "413": { $ref: "#/components/responses/Error" }
        "422": { $ref: "#/components/responses/Error" }
        "429": { $ref: "#/components/responses/Error" }
        "503": { $ref: "#/components/responses/Error" }
    get:
      summary: List runs
      operationId: listRuns
      parameters:
        - name: status
          in: query
          schema: { type: string }
          description: >-
            Comma-separated status names to include (e.g. `running,failed`);
            absent = every status. Any unknown token is a 400. Valid names are
            the RunStatus values.
        - { name: name, in: query, schema: { type: string } }
        - { name: since, in: query, schema: { type: string, format: date-time } }
        - { name: until, in: query, schema: { type: string, format: date-time } }
        - { name: limit, in: query, schema: { type: integer, default: 50, maximum: 500 } }
        - { name: cursor, in: query, schema: { type: string }, description: "Last run_id from the previous page." }
      responses:
        "200":
          description: A page of runs, newest first.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/ListResponse" }
        "401": { $ref: "#/components/responses/Error" }
  /v1/runs/{id}:
    get:
      summary: Get a run record
      operationId: getRun
      parameters: [ { $ref: "#/components/parameters/RunId" } ]
      responses:
        "200":
          description: The run record.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/RunRecord" }
        "401": { $ref: "#/components/responses/Error" }
        "404": { $ref: "#/components/responses/Error" }
    delete:
      summary: Delete a terminal run from history
      description: Does not cancel a running run; returns 409 if it is still in flight.
      operationId: deleteRun
      parameters: [ { $ref: "#/components/parameters/RunId" } ]
      responses:
        "204": { description: Deleted. }
        "401": { $ref: "#/components/responses/Error" }
        "404": { $ref: "#/components/responses/Error" }
        "409": { $ref: "#/components/responses/Error" }
  /v1/runs/{id}/cancel:
    post:
      summary: Cancel a run
      description: 202 if cancellation was requested for an in-flight run; 200 (no-op) if already terminal.
      operationId: cancelRun
      parameters: [ { $ref: "#/components/parameters/RunId" } ]
      responses:
        "200": { description: Already terminal — no-op. }
        "202": { description: Cancellation requested. }
        "401": { $ref: "#/components/responses/Error" }
        "404": { $ref: "#/components/responses/Error" }
  /v1/runs/{id}/rollback:
    post:
      summary: Undo a finished run (#706)
      description: >
        Asks the run's sink to undo one invocation — delete the rows it
        appended (by run id), restore the journaled before-images of the keys
        it upserted, or swap back the kept previous table it overwrote — then
        rewinds the row's bookmark (and exactly-once watermark) so the next run
        re-reads what was undone. The config the run was made with is taken
        from the stored run record (cluster mode) or from `config` in the
        body. A key a later run changed since is a conflict: the response then
        carries `applied: false` and `conflicts > 0`, and nothing was changed
        (pass `force` to restore anyway). Requires the `Rollback` permission
        (admin); audited as `run.rollback`.
      operationId: rollbackRun
      security: [{ bearerAuth: [] }]
      parameters: [ { $ref: "#/components/parameters/RunId" } ]
      requestBody:
        required: false
        content:
          application/json:
            schema:
              type: object
              properties:
                invocation_id: { type: string, description: "One of the run's `invocations[].run_id`; optional when the run has exactly one invocation." }
                row: { type: string, description: "The row that invocation wrote (default: search every root's state)." }
                config: { type: string, description: "The pipeline config the run was made with; required unless the server stored it." }
                config_format: { type: string, enum: [yaml, json], default: yaml }
                dry_run: { type: boolean, default: false }
                force: { type: boolean, default: false }
      responses:
        "200":
          description: The rollback report (`applied`, `deleted`, `restored`, `conflicts`, `bookmark_rewound`, `token_rewound`).
          content: { application/json: { schema: { $ref: "#/components/schemas/RollbackReport" } } }
        "400": { description: Bad config, unknown invocation, or no undoable run with that id }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the `Rollback` permission }
        "404": { $ref: "#/components/responses/Error" }
        "409": { description: The run has not finished yet }
        "422": { description: The run's config is not stored and none was supplied, or it recorded no invocation ids }
  /v1/runs/{id}/logs:
    get:
      summary: Stream a run's logs (SSE), or read persisted logs
      description: >
        Default (no `format`): `text/event-stream` — replays the run's ring
        buffer then streams the live tail. Events: `log` (a line), `truncated`
        (the reader fell behind), `end` (the run finished; the stream closes).
        The SSE buffer is ephemeral (a short drain window after the run ends).


        With `format=jsonl` or `format=text` (#529): the **persisted** logs from
        the history backend — fetchable any time after the run ends, paginated
        with `after`/`limit`. `jsonl` returns `application/x-ndjson`, one
        `{seq,ts,level,line}` object per line (oldest-first), plus a trailing
        `{"truncated":true}` when earlier lines were dropped by the per-run cap.
        `text` returns `text/plain`. Requires a persistent `--history` backend
        with `--log-retention-secs > 0`; otherwise the persisted read is empty.
      operationId: streamRunLogs
      parameters:
        - { $ref: "#/components/parameters/RunId" }
        - name: format
          in: query
          required: false
          schema: { type: string, enum: [jsonl, text] }
          description: Persisted-log format. Omit for the SSE stream.
        - name: after
          in: query
          required: false
          schema: { type: integer, format: int64 }
          description: Return only lines with `seq` greater than this (pagination).
        - name: limit
          in: query
          required: false
          schema: { type: integer, default: 1000, maximum: 10000 }
          description: Max lines per page (persisted read).
      responses:
        "200":
          description: SSE stream (default), NDJSON (`format=jsonl`), or plain text (`format=text`).
          content:
            text/event-stream:
              schema: { type: string }
            application/x-ndjson:
              schema: { type: string }
            text/plain:
              schema: { type: string }
        "400": { $ref: "#/components/responses/Error" }
        "401": { $ref: "#/components/responses/Error" }
        "404": { $ref: "#/components/responses/Error" }
  /v1/schemas:
    get:
      summary: List compiled connector and transform schemas
      security: [{ bearerAuth: [] }]
      responses:
        "200":
          description: Catalog of sources, sinks, transforms, and state-store kinds
          content:
            application/json:
              schema:
                type: object
                properties:
                  sources: { type: array, items: { type: object, properties: { name: { type: string }, description: { type: string } } } }
                  sinks: { type: array, items: { type: object, properties: { name: { type: string }, description: { type: string } } } }
                  transforms: { type: array, items: { type: object, properties: { name: { type: string }, description: { type: string } } } }
                  state: { type: array, items: { type: string } }
                  blocks:
                    type: array
                    description: Config blocks a submitted pipeline can add; fetch each schema from /v1/schemas/block/{name}
                    items:
                      type: object
                      properties:
                        name: { type: string }
                        description: { type: string }
                        placement: { type: string, enum: [pipeline, top], description: "Whether the block lives under `pipeline:` or at the top level" }
        "401": { description: Missing or invalid bearer token }
  /v1/schemas/{kind}/{name}:
    get:
      summary: Get the JSON Schema for one connector, transform, or pipeline block
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: kind, in: path, required: true, schema: { type: string, enum: [source, sink, transform, block] } }
        - { name: name, in: path, required: true, schema: { type: string } }
      responses:
        "200":
          description: JSON Schema document
          content: { application/json: { schema: { type: object } } }
        "401": { description: Missing or invalid bearer token }
        "404": { description: Unknown kind or name }
  /v1/triggers/{name}:
    post:
      summary: Fire a named webhook trigger
      description: >
        Enqueues a pipeline run for the named `webhook` trigger defined in the
        `--triggers` file. Bearer-authenticated (same token as `/v1/runs`).
        Returns 202 on success or an idempotency replay. Returns 404 for an
        unknown trigger name or if the server was started without `--triggers`.
        Returns 400 when the HTTP method is not in the trigger's configured
        `methods` list. A fire coalesced by the trigger's leading-edge
        `debounce_secs` returns 200 with `{ "status": "coalesced" }` and enqueues
        no run. Requires the `triggers` Cargo feature.
      operationId: fireTrigger
      parameters:
        - { name: name, in: path, required: true, schema: { type: string }, description: "Trigger name from the triggers file." }
      requestBody:
        required: false
        content:
          application/json:
            schema:
              type: object
              description: Optional JSON body; available as ${trigger.body} in the pipeline config.
      responses:
        "200":
          description: Fire coalesced by debounce; no new run enqueued.
          content:
            application/json:
              schema:
                type: object
                properties:
                  status: { type: string, example: coalesced }
        "202":
          description: Run enqueued (or an idempotency replay).
          content:
            application/json:
              schema: { $ref: "#/components/schemas/SubmitResponse" }
        "400": { $ref: "#/components/responses/Error" }
        "401": { $ref: "#/components/responses/Error" }
        "404": { $ref: "#/components/responses/Error" }
        "429": { $ref: "#/components/responses/Error" }
    put:
      summary: Fire a named webhook trigger (PUT)
      description: >
        Identical to `POST /v1/triggers/{name}` — fires the named `webhook`
        trigger. Accepted only when the trigger's configured `methods` list
        includes `PUT`. Bearer-authenticated (same token as `/v1/runs`). Returns
        202 on success or an idempotency replay; 404 for an unknown trigger name
        or if the server was started without `--triggers`; 400 when `PUT` is not
        in the trigger's `methods` list. A fire coalesced by the trigger's
        leading-edge `debounce_secs` returns 200 with `{ "status": "coalesced" }`
        and enqueues no run. Requires the `triggers` Cargo feature.
      operationId: fireTriggerPut
      parameters:
        - { name: name, in: path, required: true, schema: { type: string }, description: "Trigger name from the triggers file." }
      requestBody:
        required: false
        content:
          application/json:
            schema:
              type: object
              description: Optional JSON body; available as ${trigger.body} in the pipeline config.
      responses:
        "200":
          description: Fire coalesced by debounce; no new run enqueued.
          content:
            application/json:
              schema:
                type: object
                properties:
                  status: { type: string, example: coalesced }
        "202":
          description: Run enqueued (or an idempotency replay).
          content:
            application/json:
              schema: { $ref: "#/components/schemas/SubmitResponse" }
        "400": { $ref: "#/components/responses/Error" }
        "401": { $ref: "#/components/responses/Error" }
        "404": { $ref: "#/components/responses/Error" }
        "429": { $ref: "#/components/responses/Error" }
  /v1/doctor:
    post:
      summary: Validate and probe a config without running it
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [config]
              properties:
                config: { type: string }
                config_format: { type: string, enum: [yaml, json], default: yaml }
      responses:
        "200":
          description: All probes passed; body is the doctor report
          content: { application/json: { schema: { type: object } } }
        "400": { description: Malformed config body }
        "401": { description: Missing or invalid bearer token }
        "422":
          description: One or more probes failed; report in body
          content: { application/json: { schema: { type: object } } }
        "403": { description: Principal's role lacks the required permission }
  /v1/verify:
    post:
      summary: Verify a destination matches its source by content (#701)
      description: >
        Compares one root row's destination to its source by key: key ranges
        are compared by digest (server-side when both backends share an
        algorithm, so matching ranges ship no rows), disagreeing ranges are
        bisected down to `leaf_rows`, and only those are fetched and diffed
        per key. With `repair` the differing keys are re-synced through the
        row's sink (`write_mode: upsert`; deletes only with `allow_delete`).
        A mismatch is a result, not an error: the 200 body lists the
        differences. Requires `RunWrite` (operator); audited as `verify`.
      operationId: verify
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [config]
              properties:
                config: { type: string }
                config_format: { type: string, enum: [yaml, json], default: yaml }
                row: { type: string }
                repair: { type: boolean, default: false }
                allow_delete: { type: boolean, default: false }
                dry_run: { type: boolean, default: false }
                max_differences: { type: integer }
      responses:
        "200":
          description: The verification outcome (`strategy`, `ranges_compared`, `differences[]`, `repaired_upserts`, …)
          content: { application/json: { schema: { type: object } } }
        "400": { description: Bad config, no key to match on, or a destination that cannot be read back }
        "401": { description: Missing or invalid bearer token }
        "403": { description: Principal's role lacks the required permission }
        "422": { description: Config failed validation / expansion }
  /v1/backfill:
    post:
      summary: Submit a windowed backfill as tracked runs (one per window unit)
      description: >
        Plans the `[from, to)` range into window units (chunked by `window`)
        and submits one tracked run per unit through the standard run path —
        each with its `${backfill.*}` tokens substituted, its pipeline `name`
        suffixed (`{name}-backfill-{unit}`, so unit state keys never touch the
        forward-sync bookmark), `delivery` forced to `at_least_once`, the
        `${now.*}` clock set to the unit's window start, and a deterministic
        idempotency key (`backfill:{hash}:{unit}`). Re-POSTing the same body
        is replay-safe: already-submitted units replay, the rest submit — the
        API-level resume. A config carrying `shard: { count }` makes each unit
        a sharded run tracked via shard progress. Requires the `RunWrite`
        permission (operator). Bookmark-range backfills are CLI-only
        (`faucet backfill --from-bookmark`).
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [config, from, to]
              properties:
                config: { type: string }
                config_format: { type: string, enum: [yaml, json], default: yaml }
                from: { type: string, description: "RFC3339 or YYYY-MM-DD (inclusive)" }
                to: { type: string, description: "RFC3339 or YYYY-MM-DD (exclusive)" }
                window: { type: string, description: "Chunk duration: 45s / 30m / 6h / 1d / 1w" }
                timezone: { type: string, description: "IANA timezone for date boundaries" }
                name: { type: string }
                labels: { type: object, additionalProperties: { type: string } }
                timeout_secs: { type: integer, description: "Per-unit run timeout" }
      responses:
        "202":
          description: Units planned; one run submitted per unit
          content:
            application/json:
              schema:
                type: object
                properties:
                  backfill: { type: string, description: "Stable range hash (the `backfill` label on every unit run)" }
                  descriptor: { type: string }
                  planned: { type: integer }
                  submitted: { type: integer }
                  units:
                    type: array
                    items:
                      type: object
                      properties:
                        unit: { type: string }
                        start: { type: string }
                        end: { type: string }
                        status: { type: string, enum: [submitted, not_submitted] }
                        run_id: { type: string }
                        error: { type: string }
        "400": { description: Malformed body / range / window / timezone, config invalid, or a root source is not window-scoped }
        "401": { description: Missing or invalid bearer token }
        "403": { description: Principal's role lacks the required permission }
  /v1/dlq/inspect:
    post:
      summary: Inspect a dead-letter-queue location (read-only)
      description: >
        Reads a server-local DLQ location (a `.jsonl` file, a directory of
        `*.jsonl` files, or a glob) back and returns a per-reason /
        per-error-kind breakdown plus a bounded sample. Requires the `DlqRead`
        permission (viewer).
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [location]
              properties:
                location: { type: string }
                reason: { type: string, enum: [partial, dlq_all, quality, schema_drift, contract] }
                limit: { type: integer, default: 5 }
                encryption_keys:
                  type: array
                  items: { type: string }
                  description: Keys for a DLQ sealed at rest by the jsonl sink's `encryption` block (first = current, rest = rotated). Requires a server built with the `encryption` feature.
      responses:
        "200":
          description: Grouped DLQ summary
          content: { application/json: { schema: { type: object } } }
        "400": { description: Bad location or reason filter }
        "401": { description: Missing or invalid bearer token }
        "403": { description: Principal's role lacks the required permission }
  /v1/dlq/replay:
    post:
      summary: Replay quarantined records through a pipeline config
      description: >
        Re-feeds the unwrapped original payloads from a DLQ location through the
        submitted config's transforms / quality / contract / sink. Rows that
        fail again land in a fresh DLQ (never the source). Requires the
        `DlqManage` permission (operator); audited as `dlq.replay`.
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [config, from]
              properties:
                config: { type: string }
                config_format: { type: string, enum: [yaml, json], default: yaml }
                from: { type: string }
                reason: { type: string, enum: [partial, dlq_all, quality, schema_drift, contract] }
                failed_dlq: { type: string }
                row: { type: string }
                dry_run: { type: boolean, default: false }
                encryption_keys:
                  type: array
                  items: { type: string }
                  description: Keys for a DLQ sealed at rest by the jsonl sink's `encryption` block (first = current, rest = rotated). Requires a server built with the `encryption` feature.
                  # When empty, the config's own dlq jsonl `encryption` block is used.
      responses:
        "200":
          description: Replay outcome (candidates, records_written, failed_dlq)
          content: { application/json: { schema: { type: object } } }
        "400": { description: Bad config, location, or reason filter }
        "401": { description: Missing or invalid bearer token }
        "403": { description: Principal's role lacks the required permission }
        "422": { description: Config failed validation / expansion }
  /v1/dlq/discard:
    post:
      summary: Discard (archive or delete) DLQ envelopes
      description: >
        Removes DLQ envelopes matching a reason and/or age filter from a
        server-local location — archived to a `<file>.archived.jsonl` sibling by
        default, or permanently deleted with `delete: true`. Requires the
        `DlqManage` permission (operator); audited as `dlq.discard`.
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [location]
              properties:
                location: { type: string }
                reason: { type: string, enum: [partial, dlq_all, quality, schema_drift, contract] }
                before_ms: { type: integer, description: Only discard envelopes with ts_ms strictly older than this }
                delete: { type: boolean, default: false }
                encryption_keys:
                  type: array
                  items: { type: string }
                  description: Keys for a DLQ sealed at rest by the jsonl sink's `encryption` block (first = current, rest = rotated). Requires a server built with the `encryption` feature.
      responses:
        "200":
          description: Discard outcome (discarded, files_rewritten, archived_to)
          content: { application/json: { schema: { type: object } } }
        "400": { description: Bad location or reason filter }
        "401": { description: Missing or invalid bearer token }
        "403": { description: Principal's role lacks the required permission }
  /v1/catalog/datasets:
    get:
      summary: List catalogued datasets (Data Movement Catalog)
      description: >
        Every dataset the server's pipelines have touched (source or sink),
        accumulated run over run into the `--history` backend, newest activity
        first. Requires the `CatalogRead` permission (viewer). Requires the
        `catalog` Cargo feature.
      operationId: listCatalogDatasets
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: kind, in: query, required: false, schema: { type: string }, description: "Exact connector-kind filter (csv, postgres, …)." }
        - { name: q, in: query, required: false, schema: { type: string }, description: "Case-insensitive substring match on the dataset URI." }
        - { name: limit, in: query, required: false, schema: { type: integer, default: 100, maximum: 1000 } }
        - { name: cursor, in: query, required: false, schema: { type: string }, description: "Last dataset id from the previous page." }
      responses:
        "200":
          description: A page of datasets, ordered (last_seen DESC, id DESC).
          content:
            application/json:
              schema:
                type: object
                required: [datasets]
                properties:
                  datasets:
                    type: array
                    items: { $ref: "#/components/schemas/CatalogDataset" }
                  next_cursor: { type: string, nullable: true }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
  /v1/catalog/datasets/{id}:
    get:
      summary: Get one dataset's catalog detail
      description: >
        Current schema, the deduplicated schema timeline (with per-version
        diffs), recent per-run volume points, and the dataset's upstream /
        downstream lineage edges. Requires the `CatalogRead` permission
        (viewer). Requires the `catalog` Cargo feature.
      operationId: getCatalogDataset
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string }, description: "Dataset id (16-hex sha256 prefix of the canonical URI)." }
      responses:
        "200":
          description: The dataset detail.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/CatalogDatasetDetail" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
  /v1/catalog/lineage:
    get:
      summary: Get the dataset lineage graph
      description: >
        The source→sink lineage edges accumulated across runs — the whole
        graph, or a depth-bounded slice around a root dataset. Requires the
        `CatalogRead` permission (viewer). Requires the `catalog` Cargo feature.
      operationId: getCatalogLineage
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: root, in: query, required: false, schema: { type: string }, description: "Dataset id to root the graph at; omitted = the whole graph." }
        - { name: depth, in: query, required: false, schema: { type: integer, default: 5, maximum: 32 }, description: "BFS hop bound around root (ignored without one)." }
      responses:
        "200":
          description: The lineage edges.
          content:
            application/json:
              schema:
                type: object
                required: [edges]
                properties:
                  edges:
                    type: array
                    items: { $ref: "#/components/schemas/CatalogLineageEdge" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
  /v1/local-outputs:
    get:
      summary: List local sink outputs
      description: >
        The local files this server's sinks wrote (jsonl / csv / parquet), each
        with its age, lifecycle state, and the retention window in force. The
        provenance the retention GC works from — it only ever deletes files
        listed here. `state` is `present` (on disk), `expired` (collected; the
        record is kept), `external` (faucet appended to a file it did not
        create, so it is never collected or previewed), or `replaced` (faucet
        truncated a file it did not create: previewable, never collected). Requires the `LocalOutputRead` permission
        (viewer). Requires the `catalog` Cargo feature.
      operationId: listLocalOutputs
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: dataset_id, in: query, required: false, schema: { type: string }, description: "Only this dataset's outputs." }
        - { name: pipeline, in: query, required: false, schema: { type: string }, description: "Only this pipeline's outputs." }
        - { name: include_expired, in: query, required: false, schema: { type: boolean, default: false }, description: "Include already-collected outputs." }
        - { name: limit, in: query, required: false, schema: { type: integer, default: 200, maximum: 2000 } }
      responses:
        "200":
          description: The tracked local outputs.
          content:
            application/json:
              schema:
                type: object
                required: [outputs, retention_days, gc_enabled, can_manage, preview_enabled, preview_default_rows, preview_max_rows]
                properties:
                  outputs:
                    type: array
                    items: { $ref: "#/components/schemas/LocalOutput" }
                  retention_days:
                    type: integer
                    description: The server's default retention window, in days.
                  gc_enabled:
                    type: boolean
                    description: Whether the background sweeper is running (retention_days > 0).
                  can_manage:
                    type: boolean
                    description: >
                      Whether the calling principal holds `LocalOutputManage`, so a
                      client can hide destructive controls instead of offering
                      buttons that can only 403.
                  preview_enabled:
                    type: boolean
                    description: >
                      Whether this server serves dataset previews of these files
                      (`--preview-local-outputs`, #586). Same reasoning as
                      `can_manage`: a client renders a Preview control only where
                      it can work.
                  preview_default_rows:
                    type: [integer, "null"]
                    description: >
                      Rows a preview loads when `row_count_to_load` is omitted —
                      the soft cap (`FAUCET_SERVE_PREVIEW_DEFAULT_ROWS`). `null`
                      = the whole dataset by default.
                  preview_max_rows:
                    type: [integer, "null"]
                    description: >
                      Hard ceiling on a preview's rows
                      (`FAUCET_SERVE_PREVIEW_MAX_ROWS`). A client should bound its
                      own input by this rather than letting a user type a number
                      that will be silently clamped. `null` = no ceiling, which is
                      what lets a client offer "load every row" as something that
                      will actually load every row.
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
  /v1/local-outputs/{id}:
    delete:
      summary: Delete one local sink output now
      description: >
        Deletes the single recorded file, then marks its ledger row `expired`.
        Run history, catalog entries, and lineage are untouched — this removes a
        data file only. A refusal is a `200` whose report carries
        `deleted: 0` and a `skipped` reason, not an error: a `pre_existing` file
        (one faucet wrote but did not create) is never deleted, and an output of
        a run still in flight is skipped rather than unlinked mid-write.
        Requires the `LocalOutputManage` permission (operator). Requires the
        `catalog` Cargo feature.
      operationId: deleteLocalOutput
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string }, description: "Output id from the list endpoint." }
      responses:
        "200":
          description: The sweep report for this one output.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/SweepReport" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { description: No such tracked output }
  /v1/local-outputs/{id}/preview:
    get:
      summary: Preview the rows a local sink output holds
      description: >
        Reads back the first N rows of a tracked local output (jsonl / csv /
        parquet) and returns them with their column names — "here are the
        records", next to the "N records written" the run report already gives.


        Implemented as a **source-backed capped read**: the server builds the
        matching *source* connector (csv → source-csv, parquet → source-parquet,
        jsonl → its JSON Lines reader) and stops as soon as it has enough rows,
        so a capped preview of a huge file reads only its first pages and never
        loads it fully.


        **Local testing only, and opt-in.** The endpoint is inert unless the
        server was started with `--preview-local-outputs`
        (`FAUCET_SERVE_PREVIEW_LOCAL_OUTPUTS`); without it every request is a
        `403` naming the flag. The request names a ledger `id`, never a path:
        only files the local-output ledger recorded as faucet's own sink outputs
        can be read, so there is no request field that could address another
        file. Requires the `LocalOutputRead` permission (viewer) and the
        `catalog` Cargo feature.


        Row bounds are the server's, and there is **no offset or cursor** — these
        sources are sequential streams with no row index, so "show me more" is
        spelled "raise the limit". `row_count_to_load` overrides the soft cap
        (`FAUCET_SERVE_PREVIEW_DEFAULT_ROWS`, default 500) and is clamped to the
        hard cap (`FAUCET_SERVE_PREVIEW_MAX_ROWS`, default 5000) — never honoured
        above it. Both are surfaced on `GET /v1/local-outputs` so a client can
        label its own control.


        `row_count_to_load=all` (or `0`) asks for the **whole dataset**. It is
        served in full only when the operator lifted the ceiling
        (`--preview-max-rows 0`, reported as `preview_max_rows: null`);
        otherwise it resolves to the ceiling, which is the point of having one.
        An unlimited read is unlimited in *rows* only — it is still paged, and
        still bounded by a response-size budget (64 MiB) and a 30s deadline, both
        of which are checked as pages arrive. A dataset that exceeds either comes
        back as a **partial answer whose `capped_by` says which bound stopped
        it**, never as an error and never as an unbounded buffer.


        A served preview writes a `local_output.preview` audit entry (principal,
        output, row count) — the one read on this control plane that returns
        pipeline data rather than metadata about a pipeline.
      operationId: previewLocalOutput
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string }, description: "Output id from the list endpoint." }
        - name: row_count_to_load
          in: query
          required: false
          schema: { type: string, pattern: "^([0-9]+|[Aa][Ll][Ll])$" }
          description: >
            Rows to load: a count, or `all` (or `0`) for the whole dataset.
            Omitted → the server's soft cap. Above the hard cap → clamped to it.
            The bound actually used comes back as `row_limit`. A value that is
            neither a number nor `all` is a `400` naming the parameter, never a
            silent fall back to the default.
      responses:
        "200":
          description: The capped page of rows.
          content:
            application/json:
              schema:
                type: object
                required: [output_id, path, kind, dataset_id, pipeline, run_id, rows, columns, row_count, row_limit, max_rows, truncated, elapsed_ms]
                properties:
                  output_id: { type: string, description: "The ledger id that was previewed." }
                  path: { type: string, description: "The file that was read." }
                  kind: { type: string, enum: [jsonl, csv, parquet], description: "Connector kind of the writing sink." }
                  dataset_id: { type: string }
                  pipeline: { type: string }
                  run_id: { type: string, description: "The run that most recently wrote this file." }
                  rows:
                    type: array
                    items: { type: object, additionalProperties: true }
                    description: >
                      The records, at most `row_limit` of them. Records that are
                      not JSON objects (a scalar or array per line is legal
                      NDJSON) come back as-is.
                  columns:
                    type: array
                    items: { type: string }
                    description: >
                      Column names across `rows`, in source order — the table
                      header. Empty when the records are not JSON objects. A
                      field only some records carry still appears, so a preview
                      never hides a written column.
                  row_count:
                    type: integer
                    description: >
                      Rows returned. Explicit so a client never has to decide
                      whether `rows.length` is the answer.
                  row_limit:
                    type: [integer, "null"]
                    description: >
                      The bound this request resolved to, after clamping. `null`
                      = unlimited (the caller asked for everything and this
                      server has no ceiling).
                  max_rows:
                    type: [integer, "null"]
                    description: The server's hard cap; `null` = no ceiling.
                  truncated:
                    type: boolean
                    description: >
                      More of the dataset exists beyond what was returned.
                      Observed (one row past the cap is read), not inferred from
                      `rows.length`.
                  capped_by:
                    type: string
                    enum: [rows, bytes, time]
                    description: >
                      Which bound stopped the read: the row limit, the
                      response-size budget, or the deadline. **Absent when the
                      response is the whole dataset** — a whole-dataset request
                      that came back partial must say why, or a clipped answer is
                      indistinguishable from a complete one.
                  elapsed_ms: { type: integer }
        "400":
          description: >
            The output's kind has no reader (e.g. a sink that is not one of
            jsonl / csv / parquet), this build of faucet lacks the source
            connector for it, or `row_count_to_load` was neither a number nor
            `all`.
        "401": { $ref: "#/components/responses/Error" }
        "403":
          description: >
            Previews are disabled on this server (`--preview-local-outputs` was
            not passed); the principal's role lacks `LocalOutputRead`; or the
            output is `external` — a file faucet appended to but did not create,
            whose contents are not faucet's to serve, for the same reason the
            retention GC refuses to delete it. (A `replaced` output — truncated
            by faucet — is previewable.)
        "404": { description: No such tracked output }
        "409":
          description: >
            The file is gone — collected by local-output retention (#587) or
            removed out of band. The ledger row and the run record are kept; the
            message says which. Reported as a conflict rather than a 500 from a
            failed open.
        "422":
          description: >
            The file is present but could not be parsed — e.g. a partially
            written last line from a run that died mid-flush. The message carries
            the connector's own diagnostic (line number, byte offset).
        "503":
          description: >
            The read was abandoned after the server's 60s hard timeout — a single
            page that never returned, rather than a verdict on the file's
            contents. (The ordinary 30s deadline yields a partial answer with
            `capped_by: time` instead of an error.)
  /v1/local-outputs/cleanup:
    post:
      summary: Bulk-clean local sink outputs
      description: >
        Deletes recorded local output files in bulk. Exactly one scope must be
        given — `older_than_days`, `expired` (each output's own window),
        `dataset_id`, `run_id`, or `all` — and combining them is a `400` rather
        than a guess. A scope that ignores retention windows (`all`, or
        `older_than_days: 0`) additionally requires `confirm: true`, the same gate
        the CLI spells `--yes`. `all` includes outputs still inside their retention window, so a
        client should confirm before sending it. Only files the ledger records as
        faucet's own sink outputs are ever deleted: never a glob, never a
        directory, and never a file faucet merely appended to. Requires the
        `LocalOutputManage` permission (operator). Requires the `catalog` Cargo
        feature.
      operationId: cleanupLocalOutputs
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              properties:
                older_than_days:
                  type: integer
                  description: Delete outputs older than this many days, ignoring per-pipeline overrides.
                expired:
                  type: boolean
                  description: Delete outputs whose own retention window has elapsed.
                dataset_id:
                  type: string
                  description: Delete this dataset's outputs.
                run_id:
                  type: string
                  description: >
                    Delete the outputs one run most recently wrote — "clean up
                    after that run". The run's history record is untouched.
                all:
                  type: boolean
                  description: Delete every tracked output, regardless of age.
                dry_run:
                  type: boolean
                  default: false
                  description: Report what would be deleted without touching anything.
                confirm:
                  type: boolean
                  default: false
                  description: >
                    Required for a scope that can delete outputs still inside their
                    retention window — `all`, and `older_than_days: 0` (which
                    matches everything). Without it such a request is a `400`.
                    Ignored for the bounded scopes and for a dry run.
      responses:
        "200":
          description: The sweep report.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/SweepReport" }
        "400": { $ref: "#/components/responses/Error" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
  /v1/templates:
    post:
      summary: Register a pipeline template
      description: >
        Stores a config document plus the typed `params:` it declares, so later
        runs are triggered by `{id, params}` instead of re-sending the whole
        config. Each register appends a **new version** and **does not move
        existing callers** — an unpinned run keeps resolving `stable` until a
        version is explicitly launched (pass `launch: true` here, or call
        `POST /v1/templates/{id}/launch`). The first registration therefore leaves
        the template a `draft`. The body is stored verbatim, so `${env:…}` /
        `${vault:…}` stay unresolved and are resolved at trigger time on the
        instance that runs the pipeline. Validation runs against a
        placeholder binding, so no secret is read and no connector is built.
        Requires the `TemplateAdmin` permission (admin) and the `templates`
        Cargo feature.
      operationId: registerTemplate
      security: [{ bearerAuth: [] }]
      requestBody:
        required: true
        content:
          application/json:
            schema: { $ref: "#/components/schemas/RegisterTemplateRequest" }
      responses:
        "201":
          description: The newly registered version's summary.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/TemplateSummary" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "422": { $ref: "#/components/responses/Error" }
    get:
      summary: List registered templates
      description: >
        The newest version of every registered template, newest-registered first,
        each carrying its `kind` and the template's release `state` (status,
        `stable` / `previous` / `newest`, channel pointers). Bodies are omitted —
        fetch one with `GET /v1/templates/{id}`. `?kind=` narrows to source
        templates, sink templates, or complete pipelines. Requires the
        `TemplateRead` permission (viewer).
      operationId: listTemplates
      security: [{ bearerAuth: [] }]
      parameters:
        - name: kind
          in: query
          required: false
          schema: { type: string, enum: [source-template, sink-template, deployment, pipeline] }
          description: Return only templates of this kind.
      responses:
        "200":
          description: The template summaries.
          content:
            application/json:
              schema:
                type: object
                required: [templates]
                properties:
                  templates:
                    type: array
                    items: { $ref: "#/components/schemas/TemplateSummary" }
                  sync:
                    description: >
                      Present only when the server was started with
                      `--templates-sync`: the remote origins it pulls templates
                      from (RFC 0006), so a client can offer "sync now" /
                      "publish" only where they apply.
                    type: object
                    required: [origins]
                    properties:
                      origins:
                        type: array
                        items:
                          type: object
                          required: [name, kind, prefix, launch, prune]
                          properties:
                            name: { type: string }
                            kind: { type: string, enum: [github, s3, gcs, azure_blob] }
                            prefix: { type: string, description: "Id namespace the origin owns." }
                            launch: { type: string, enum: [ignore, follow, always] }
                            prune: { type: string, enum: [keep, deprecate] }
                            interval_secs: { type: integer, description: "Periodic pull interval; absent = on start / on demand only." }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
  /v1/templates/matrix:
    get:
      summary: Source × sink compatibility matrix of the registered templates
      description: >
        Composes every registered `source-template` with every registered
        `sink-template` (each at its `stable` version when launched, else its
        newest) and reports, per pairing, whether every stream has a write mode
        the sink supports and which mode each stream resolves to. The same
        shape as a Template Hub catalog's `index.json`, with `command` set to
        the `faucet template run … --sink …` invocation. Static route — a
        template can never be addressed as `matrix`. Requires `TemplateRead`.
      operationId: templateMatrix
      security: [{ bearerAuth: [] }]
      responses:
        "200":
          description: Sources, sinks, and one cell per pairing.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/HubIndex" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
  /v1/templates/{id}:
    get:
      summary: Get one pipeline template
      description: >
        One version of a template — the launched (`stable`) one unless `version`
        is given — including the stored config body, the template's whole release
        state, and the launch log. Pass `version=newest` to open a `draft`
        template, which has no `stable` version yet. Requires the `TemplateRead`
        permission (viewer).
      operationId: getTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string }, description: "Template id (slug)." }
        - { name: version, in: query, required: false, schema: { $ref: "#/components/schemas/VersionSelector" }, description: "Version to return: `stable` (the default when omitted), another channel, or an exact version number." }
      responses:
        "200":
          description: The template version plus its version list.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/TemplateDetail" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
    delete:
      summary: Delete a pipeline template
      description: >
        Removes one version (`version` given) or every version of the template.
        Runs already produced by it are untouched. Requires the
        `TemplateAdmin` permission (admin).
      operationId: deleteTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
        - { name: version, in: query, required: false, schema: { $ref: "#/components/schemas/VersionSelector" }, description: "Delete just this version (a channel name or a number). Omitted = every version. Channels pointing at a deleted version, and its launch-log entries, are dropped with it." }
      responses:
        "204": { description: Deleted }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
  /v1/templates/{id}/tags:
    post:
      summary: Point a named channel at a template version
      description: >
        Points one of the assignable channels (`dev`, `test`, `staging`,
        `pre-prod`, `canary`, `prod`) at a version. Versions themselves are numeric
        and immutable; this is how `prod` advances from v3 to v4. `version` may be
        a number, or another channel whose current target should be copied
        (`{"tag":"prod","version":"stable"}`), defaulting to `stable` — resolved to
        a concrete version, so the pointer never silently follows future
        registrations. The derived channels (`stable`, `previous`, `newest`) cannot
        be assigned (`422`) — `stable` moves only via `launch`. Requires the
        `TemplateAdmin` permission (admin) and the `templates` Cargo feature.
      operationId: promoteTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [tag]
              properties:
                tag: { $ref: "#/components/schemas/VersionChannel" }
                version: { $ref: "#/components/schemas/VersionSelector" }
      responses:
        "200":
          description: The channel now points at this version.
          content:
            application/json:
              schema:
                type: object
                required: [id, tag, version]
                properties:
                  id: { type: string }
                  tag: { type: string }
                  version: { type: integer }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
        "422": { $ref: "#/components/responses/Error" }
  /v1/templates/{id}/launch:
    post:
      summary: Launch a template version (make it live)
      description: >
        The one operation that moves unpinned callers: it appends to the launch log
        so `stable` points at `version`, and the version it replaced becomes
        `previous`. Registering a build does **not** do this — that separation is
        what lets a nightly land without dragging anyone along. `version` defaults
        to `newest` (launching what you just registered is the common case) and may
        also be a channel whose current target should be blessed
        (`{"version":"pre-prod"}`). Re-launching the already-live version is a
        no-op (`already_launched: true`), keeping `previous` a real rollback
        target. A deprecated template refuses to launch — revive it first.
        Requires the `TemplateAdmin` permission (admin) and the `templates`
        Cargo feature.
      operationId: launchTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
      requestBody:
        required: false
        content:
          application/json:
            schema:
              type: object
              properties:
                version:
                  description: "Version to make live: a number, or a channel whose target to copy. Defaults to `newest`."
                  $ref: "#/components/schemas/VersionSelector"
      responses:
        "200":
          description: The version now live.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/LaunchResponse" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
        "422": { $ref: "#/components/responses/Error" }
  /v1/templates/{id}/rollback:
    post:
      summary: Roll a template back to its previous launch
      description: >
        Re-launches `previous` — the version that was live before the current one.
        Implemented as a normal launch, so the launch log keeps the full audit
        trail and `previous` becomes the version you just rolled off. A template
        with fewer than two launches has nothing to roll back to (`422`). Requires
        the `TemplateAdmin` permission (admin) and the `templates` Cargo
        feature.
      operationId: rollbackTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
      responses:
        "200":
          description: The version now live.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/LaunchResponse" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
        "422": { $ref: "#/components/responses/Error" }
  /v1/templates/{id}/deprecate:
    post:
      summary: Retire (or revive) a pipeline template
      description: >
        Deprecation is template-wide and deliberately soft: a deprecated template
        keeps serving callers who pin or ride `stable` — retiring must not
        hard-break them — but every trigger response carries a `deprecated` field
        and listings mark it. `DELETE` is the hard stop. Pass `undo: true` to
        revive. Requires the `TemplateAdmin` permission (admin) and the
        `templates` Cargo feature.
      operationId: deprecateTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
      requestBody:
        required: false
        content:
          application/json:
            schema:
              type: object
              properties:
                reason: { type: string, description: "Why it is being retired; surfaced to anyone who triggers it." }
                undo: { type: boolean, default: false, description: "Revive instead of retire." }
      responses:
        "200":
          description: The template's status after the change.
          content:
            application/json:
              schema:
                type: object
                required: [id, status]
                properties:
                  id: { type: string }
                  status: { $ref: "#/components/schemas/TemplateStatus" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
  /v1/templates/{id}/versions/{version}/deprecate:
    post:
      summary: Retire (or revive) one version of a pipeline template
      description: >
        A retired version keeps running when pinned, or when `stable` or a
        channel points at it, and every trigger of it carries a `deprecated`
        field. `newest` skips it and `launch` refuses it. Pass `undo: true` to
        revive. `version` is a number or a channel name. Requires the
        `TemplateAdmin` permission (admin) and the `templates` Cargo feature.
      operationId: deprecateTemplateVersion
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
        - { name: version, in: path, required: true, schema: { type: string }, description: "A version number or a channel name." }
      requestBody:
        required: false
        content:
          application/json:
            schema:
              type: object
              properties:
                reason: { type: string, description: "Why it is being retired; surfaced on every trigger of it." }
                undo: { type: boolean, default: false, description: "Revive instead of retire." }
      responses:
        "200":
          description: The version's state after the change.
          content:
            application/json:
              schema:
                type: object
                required: [id, version, deprecated]
                properties:
                  id: { type: string }
                  version: { type: integer }
                  deprecated: { type: boolean }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
        "422": { $ref: "#/components/responses/Error" }
  /v1/templates/sync:
    post:
      summary: Pull pipeline templates from the configured remote origins
      description: >
        Template hosting + sync (RFC 0006). Lists every origin in the server's
        `--templates-sync` file (or just `origin`), pairs each `<id>.yaml` /
        `.json` with its optional `<id>.faucet.yaml` sidecar, and diffs against
        the registry: a new or changed body **appends** a version, an unchanged
        one is a no-op (body hash), a template that vanished upstream is
        reported (`prune: keep`) or deprecated (`prune: deprecate`) — never
        deleted. `stable` moves only under `launch: follow` / `always`. Every
        register is attributed to the calling principal. `dry_run: true`
        returns the plan without applying it. One unreadable origin does not
        stop the others; it is listed under `origin_errors`. Requires the
        `TemplateAdmin` permission (admin) and the `templates-sync` Cargo
        feature; `422` when the server has no origins configured.
      operationId: syncTemplates
      security: [{ bearerAuth: [] }]
      requestBody:
        required: false
        content:
          application/json:
            schema:
              type: object
              properties:
                origin: { type: string, description: "Pull only this origin (by name). Default: all." }
                dry_run: { type: boolean, default: false }
      responses:
        "200":
          description: One report per origin pulled.
          content:
            application/json:
              schema:
                type: object
                required: [dry_run, reports, origin_errors]
                properties:
                  dry_run: { type: boolean }
                  reports:
                    type: array
                    items: { $ref: "#/components/schemas/TemplateSyncReport" }
                  origin_errors:
                    type: array
                    items:
                      type: object
                      required: [origin, error]
                      properties:
                        origin: { type: string }
                        error: { type: string }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "422": { $ref: "#/components/responses/Error" }
  /v1/templates/{id}/publish:
    post:
      summary: Write one registered version back to a remote origin
      description: >
        The manual reverse of `sync` (RFC 0006): writes the selected version's
        body to `origin` as `<id minus the origin's prefix>.<yaml|json>`. Never
        runs on its own — an operator-driven step, audited as
        `template.publish`. The id must carry the origin's `prefix`
        (otherwise the next pull would re-register it under a different id).
        Requires `TemplateAdmin` (admin) and the `templates-sync` Cargo feature.
      operationId: publishTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [origin]
              properties:
                origin: { type: string, description: "Origin name from the sync file." }
                version: { $ref: "#/components/schemas/VersionSelector" }
      responses:
        "200":
          description: Where the file landed.
          content:
            application/json:
              schema:
                type: object
                required: [id, version, origin, name, location]
                properties:
                  id: { type: string }
                  version: { type: integer }
                  origin: { type: string }
                  name: { type: string, description: "File name written under the origin's directory." }
                  location: { type: string, description: "Human-readable location (URL or store path)." }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
        "422": { $ref: "#/components/responses/Error" }
  /v1/templates/{id}/runs:
    post:
      summary: Trigger a run from a template (a source template composed with a sink, or a pipeline)
      description: >
        Materializes the template with the supplied `params` (and optional
        per-request `env` overrides), then submits it through the same path as
        `POST /v1/runs` — so idempotency keys, `doctor_first`, queue limits,
        cluster dispatch, metrics, and the audit log behave identically. The run
        is labelled `template` / `template_version` for provenance. Requires the
        `RunWrite` permission (operator).


        `version` defaults to `stable`, so a nightly registration never drags
        unpinned callers along. A `draft` template (nothing launched) refuses an
        unpinned request with a `422`; a `deprecated` one still runs but the
        response carries a `deprecated` field. A missing required param or a type
        mismatch is a `422` naming the param; an unknown template or pinned
        version is a `404`. On a clustered server a
        template declaring `secret: true` params is refused (`422`) because the
        materialized config is persisted for peer execution — reference the
        secret from the template body instead.
      operationId: triggerTemplate
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: id, in: path, required: true, schema: { type: string } }
      requestBody:
        required: false
        content:
          application/json:
            schema: { $ref: "#/components/schemas/TriggerTemplateRequest" }
      responses:
        "202":
          description: Run accepted.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/TriggerTemplateResponse" }
        "401": { $ref: "#/components/responses/Error" }
        "403": { description: Principal's role lacks the required permission }
        "404": { $ref: "#/components/responses/Error" }
        "422": { $ref: "#/components/responses/Error" }
        "429": { $ref: "#/components/responses/Error" }
  /v1/audit:
    get:
      summary: Read the control-plane audit log (RBAC, admin-only)
      description: >
        Returns recent audit records (newest first), each a mutating or denied
        control-plane action attributed to a principal. Requires the `AuditRead`
        permission (admin role). Under a single `--auth-token` or `--no-auth`
        the implicit principal is admin, so this endpoint is available.
      security: [{ bearerAuth: [] }]
      parameters:
        - { name: principal, in: query, required: false, schema: { type: string } }
        - { name: action, in: query, required: false, schema: { type: string } }
        - { name: since, in: query, required: false, schema: { type: string, format: date-time } }
        - { name: until, in: query, required: false, schema: { type: string, format: date-time } }
        - { name: limit, in: query, required: false, schema: { type: integer, default: 100, maximum: 1000 } }
      responses:
        "200":
          description: Audit records, newest first.
          content:
            application/json:
              schema:
                type: object
                properties:
                  entries:
                    type: array
                    items: { $ref: "#/components/schemas/AuditEntry" }
        "400": { description: Malformed `since` / `until` timestamp }
        "401": { description: Missing or invalid bearer token }
        "403": { description: Principal's role lacks the AuditRead permission }
  /v1/reload:
    post:
      summary: Hot-reload the server's --default-config merge base (admin-only)
      description: >
        Re-reads and re-validates the `--default-config` file and atomically
        swaps the in-memory merge base — without restarting or interrupting
        in-flight runs (which already captured their config). An invalid new
        config is rejected (422) and the previous base is kept. When the server
        was started without `--default-config`, this is a no-op (200,
        `reloaded: false`). Requires the `Reload` permission (admin role).
      security: [{ bearerAuth: [] }]
      responses:
        "200":
          description: Reloaded (or a no-op when no --default-config is set).
          content:
            application/json:
              schema:
                type: object
                properties:
                  reloaded: { type: boolean }
                  path: { type: string }
                  reason: { type: string }
        "401": { description: Missing or invalid bearer token }
        "403": { description: Principal's role lacks the Reload permission }
        "422": { description: New config invalid; previous config kept }
  /v1/whoami:
    get:
      summary: The caller's principal, role and permissions
      description: >
        Every role may call it. The web console uses it to hide controls the
        caller cannot use; each route still enforces its own permission.
        `--no-auth` and `--auth-token` report an implicit admin.
      operationId: whoami
      security: [{ bearerAuth: [] }]
      responses:
        "200":
          description: The resolved identity.
          content:
            application/json:
              schema:
                type: object
                required: [principal, role, permissions]
                properties:
                  principal: { type: string }
                  role: { type: string, enum: [viewer, operator, admin] }
                  permissions:
                    type: array
                    items:
                      type: string
                      enum: [run_read, run_write, schema_read, doctor, trigger_fire, dlq_read, dlq_manage, catalog_read, template_read, template_admin, local_output_read, local_output_manage, audit_read, reload, identity]
        "401": { description: Missing or invalid bearer token }
  /healthz:
    get:
      summary: Liveness probe (unauthenticated)
      operationId: healthz
      security: []
      responses:
        "200": { description: The process is alive. }
  /readyz:
    get:
      summary: Readiness probe (unauthenticated)
      description: >
        503 when the history backend is degraded or the run queue is full.
        Always returns a JSON body with `status`, `history_ok`, `queue_ok`, and
        a `cluster` object (`enabled` bool + `instances` live-member count).
      operationId: readyz
      security: []
      responses:
        "200":
          description: Ready to accept work.
          content:
            application/json:
              schema:
                type: object
                properties:
                  status: { type: string, example: ready }
                  history_ok: { type: boolean }
                  queue_ok: { type: boolean }
                  cluster:
                    type: object
                    properties:
                      enabled: { type: boolean, description: "true when --cluster was set." }
                      instances: { type: integer, description: "Live cluster member count (from last membership heartbeat)." }
                  triggers:
                    type: array
                    description: "Per-watcher health; present only when --triggers is active."
                    items:
                      type: object
                      properties:
                        name: { type: string }
                        healthy: { type: boolean, description: "false when the watcher is in error backoff." }
        "503":
          description: Not ready (history degraded or queue full).
          content:
            application/json:
              schema:
                type: object
                properties:
                  status: { type: string, example: not_ready }
                  history_ok: { type: boolean }
                  queue_ok: { type: boolean }
                  cluster:
                    type: object
                    properties:
                      enabled: { type: boolean }
                      instances: { type: integer }
                  triggers:
                    type: array
                    description: "Per-watcher health; present only when --triggers is active."
                    items:
                      type: object
                      properties:
                        name: { type: string }
                        healthy: { type: boolean }
  /metrics:
    get:
      summary: Prometheus metrics (unauthenticated)
      operationId: metrics
      security: []
      responses:
        "200":
          description: Prometheus text exposition.
          content:
            text/plain: { schema: { type: string } }
        "503": { description: Metrics recorder unavailable in this process. }
components:
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
  parameters:
    RunId:
      name: id
      in: path
      required: true
      schema: { type: string }
  responses:
    Error:
      description: Error envelope.
      content:
        application/json:
          schema: { $ref: "#/components/schemas/ApiError" }
  schemas:
    RunStatus:
      type: string
      enum: [pending, queued, running, sharded, completed, failed, cancelled]
      description: >
        `pending` — cluster mode only; stored in the shared DB awaiting a claim by any instance.
        `queued` — claimed/admitted to the local execution queue.
        `running` — actively executing.
        `sharded` — cluster Mode B only; the run was expanded into source shards
        (rows in `faucet_serve_shards`) that execute independently; finalized to a
        terminal state once every shard finishes.
        `completed` / `failed` / `cancelled` — terminal states.
    SubmitRequest:
      type: object
      required: [config]
      properties:
        config: { type: string, description: "YAML or JSON pipeline body." }
        config_format: { type: string, enum: [yaml, json], default: yaml }
        name: { type: string, description: "Metadata + state-key/metric identity. Reused name = shared bookmarks." }
        labels: { type: object, additionalProperties: { type: string } }
        timeout_secs: { type: integer, nullable: true }
        doctor_first: { type: boolean, default: false, description: "Run doctor probes first; 422 with report on failure." }
        idempotency_key: { type: string, nullable: true }
        clock: { type: string, format: date-time, description: "Override the ${now.*} clock (backfills)." }
        concurrency: { type: integer, minimum: 1, nullable: true, description: "Override this run's connector concurrency (source + sink connections/fetches), whatever the config says. Maps onto whichever knob the connector declares (max_connections / request_concurrency / partition_concurrency / shard_concurrency / concurrency); a connector with none ignores it. Does not change matrix parallelism or the server's --max-concurrent slots. Per-shard for a sharded run. 0 is rejected." }
        callback: { $ref: "#/components/schemas/CallbackSpec" }
      example:
        name: orders_to_warehouse
        config_format: yaml
        config: |
          version: 1
          pipeline:
            source:
              type: postgres
              config:
                connection_url: ${env:PG_URL}
                query: SELECT * FROM orders
            sink:
              type: bigquery
              config:
                dataset: analytics
                table: orders
                write_mode: upsert
                primary_key: [id]
        doctor_first: false
    CallbackSpec:
      type: object
      required: [url]
      description: >-
        Completion callback for this run (#481). Fired once when the run reaches a
        terminal state. Delivery is **at-most-once** and best-effort: it is not
        fired when a run is failed by lease-expiry recovery inside the history
        backend, so treat a missing callback as "unknown" and reconcile against
        GET /v1/runs/{id}, which is always authoritative. Refused with 422 on
        /v1/backfill (one POST fans out into N unit runs).
      properties:
        url:
          type: string
          description: >-
            Destination. http/https only. Link-local / cloud-metadata addresses
            are refused unless named in the server's --callback-allow-host
            allowlist; when that allowlist is non-empty, only its hosts are
            permitted.
        method: { type: string, default: POST }
        headers:
          type: object
          additionalProperties: { type: string }
          description: >-
            Extra request headers. Refused with 422 on a clustered server, which
            persists the run record (including this map) into the shared
            run-history database for a peer to execute.
        extra_fields:
          type: object
          additionalProperties: true
          description: >-
            Static fields merged into the callback body. A key colliding with a
            faucet-emitted field is refused with 422.
        on:
          type: array
          items: { $ref: "#/components/schemas/RunStatus" }
          description: >-
            Terminal statuses to fire on. Empty (default) = all of them. A
            callback subscribed only to `completed` will never fire for a
            cancelled or failed run.
      example:
        url: https://example.com/faucet
        method: example
        headers:
          key: value
        extra_fields:
          key: value
        'on':
        - pending
    SubmitResponse:
      type: object
      required: [run_id, status, submitted_at]
      properties:
        run_id: { type: string }
        status: { $ref: "#/components/schemas/RunStatus" }
        submitted_at: { type: string, format: date-time }
      example:
        run_id: run_01HZY8K3QK7
        status: pending
        submitted_at: '2026-01-01T00:00:00Z'
    RollbackReport:
      type: object
      properties:
        run_id: { type: string, description: "The invocation that was undone." }
        row: { type: string }
        sink_kind: { type: string }
        dataset: { type: string }
        mode: { type: string, enum: [append, upsert, overwrite] }
        dry_run: { type: boolean }
        deleted: { type: integer }
        restored: { type: integer }
        conflicts: { type: integer, description: "Keys a later run changed since; without `force` these block the rollback." }
        applied: { type: boolean }
        blocked: { type: boolean, description: "Refused because a later run changed the keys and `force` was not set (a dry run reports whether it would be)." }
        note: { type: string, nullable: true }
        bookmark_rewound: { type: boolean }
        token_rewound: { type: boolean }
    InvocationRecord:
      type: object
      properties:
        row_id: { type: string }
        parent_record_key: { type: string, nullable: true }
        run_id: { type: string, nullable: true, description: "The invocation's own run id — what `POST /v1/runs/{id}/rollback` undoes (#706)." }
        records_written: { type: integer }
        duration_ms: { type: integer, description: "Wall-clock duration of this invocation in milliseconds (#645)." }
        error: { type: string, nullable: true }
      example:
        row_id: orders_to_warehouse
        parent_record_key: example
        records_written: 1
        duration_ms: 1483
        error: example
    RunRecord:
      type: object
      required: [run_id, status, submitted_at, records_written, invocations, labels]
      properties:
        run_id: { type: string }
        name: { type: string, nullable: true }
        labels: { type: object, additionalProperties: { type: string } }
        status: { $ref: "#/components/schemas/RunStatus" }
        submitted_at: { type: string, format: date-time }
        started_at: { type: string, format: date-time, nullable: true }
        finished_at: { type: string, format: date-time, nullable: true }
        elapsed_secs: { type: number, nullable: true }
        records_written: { type: integer }
        invocations:
          type: array
          items: { $ref: "#/components/schemas/InvocationRecord" }
        error: { type: string, nullable: true }
        idempotency_key: { type: string, nullable: true }
        doctor_report: { type: object, nullable: true }
        callback: { $ref: "#/components/schemas/CallbackSpec" }
      example:
        run_id: run_01HZY8K3QK7
        name: orders_to_warehouse
        labels:
          key: value
        status: pending
        submitted_at: '2026-01-01T00:00:00Z'
        started_at: '2026-01-01T00:00:00Z'
        finished_at: '2026-01-01T00:00:00Z'
        elapsed_secs: 1.0
        records_written: 1
        invocations:
        - row_id: orders_to_warehouse
          parent_record_key: example
          records_written: 1
          error: example
        error: example
        idempotency_key: example
        doctor_report:
          key: value
        callback:
          url: https://example.com/faucet
          method: example
          headers:
            key: value
          extra_fields:
            key: value
          'on':
          - {}
    ListResponse:
      type: object
      required: [runs]
      properties:
        runs:
          type: array
          items: { $ref: "#/components/schemas/RunRecord" }
        next_cursor: { type: string, nullable: true }
      example:
        runs:
        - run_id: run_01HZY8K3QK7
          name: orders_to_warehouse
          labels:
            key: value
          status: {}
          submitted_at: '2026-01-01T00:00:00Z'
          started_at: '2026-01-01T00:00:00Z'
          finished_at: '2026-01-01T00:00:00Z'
          elapsed_secs: 1.0
          records_written: 1
          invocations:
          - {}
          error: example
          idempotency_key: example
          doctor_report:
            key: value
          callback: {}
        next_cursor: example
    AuditEntry:
      type: object
      required: [id, timestamp, principal, role, action, result]
      properties:
        id: { type: string, description: "Time-ordered UUIDv7 (also the sort key)." }
        timestamp: { type: string, format: date-time }
        principal: { type: string, description: "Principal name (anonymous / token / <name> / trigger:<name>)." }
        role: { type: string, enum: [viewer, operator, admin] }
        action: { type: string, description: "e.g. run.submit / run.cancel / run.delete / trigger.fire / <route> (denials)." }
        run_id: { type: string, nullable: true }
        config_fingerprint: { type: string, nullable: true, description: "sha256 of the merged config (submit actions)." }
        source_ip: { type: string, nullable: true }
        result: { type: string, enum: [ok, denied] }
      example:
        id: orders_to_warehouse
        timestamp: '2026-01-01T00:00:00Z'
        principal: example
        role: viewer
        action: example
        run_id: run_01HZY8K3QK7
        config_fingerprint: example
        source_ip: example
        result: ok
    CatalogDataset:
      type: object
      required: [id, uri, kind, roles, first_seen, last_seen, last_success, last_run_id, pipeline, last_records, total_records, runs, schema_versions]
      properties:
        id: { type: string, description: "16-hex sha256 prefix of the canonical URI." }
        uri: { type: string, description: "Canonical dataset URI — credential-redacted; ${now.*}-derived segments folded back to their tokens." }
        kind: { type: string, description: "Connector kind (csv, postgres, …)." }
        roles: { type: array, items: { type: string, enum: [source, sink] } }
        first_seen: { type: string, format: date-time }
        last_seen: { type: string, format: date-time }
        last_success: { type: string, format: date-time, description: "Last successful run that touched this dataset (freshness)." }
        last_run_id: { type: string }
        pipeline: { type: string, description: "Pipeline name of the most recent run." }
        last_records: { type: integer }
        total_records: { type: integer }
        runs: { type: integer }
        schema_versions: { type: integer }
        current_schema: { type: object, nullable: true, description: "Latest observed record schema (infer_schema shape)." }
        current_schema_hash: { type: string, nullable: true }
      example:
        id: orders_to_warehouse
        uri: example
        kind: postgres
        roles:
        - source
        first_seen: '2026-01-01T00:00:00Z'
        last_seen: '2026-01-01T00:00:00Z'
        last_success: '2026-01-01T00:00:00Z'
        last_run_id: run_01HZY8K3QK7
        pipeline: example
        last_records: 1
        total_records: 42
        runs: 1
        schema_versions: 3
        current_schema:
          key: value
        current_schema_hash: example
    CatalogSchemaVersion:
      type: object
      required: [dataset_id, version, recorded_at, run_id, schema, schema_hash]
      properties:
        dataset_id: { type: string }
        version: { type: integer, description: "1-based, appended only on content change." }
        recorded_at: { type: string, format: date-time }
        run_id: { type: string }
        schema: { type: object }
        schema_hash: { type: string }
        diff: { type: object, nullable: true, description: "Diff vs the previous version ({added, widened, changed, removed}); absent for version 1." }
      example:
        dataset_id: orders_to_warehouse
        version: 3
        recorded_at: '2026-01-01T00:00:00Z'
        run_id: run_01HZY8K3QK7
        schema:
          key: value
        schema_hash: example
        diff:
          key: value
    CatalogStatsPoint:
      type: object
      required: [recorded_at, run_id, records]
      properties:
        recorded_at: { type: string, format: date-time }
        run_id: { type: string }
        records: { type: integer }
      example:
        recorded_at: '2026-01-01T00:00:00Z'
        run_id: run_01HZY8K3QK7
        records: 1
    CatalogLineageEdge:
      type: object
      required: [src_id, dst_id, src_uri, dst_uri, pipeline, row, first_seen, last_seen, last_run_id, runs, last_records]
      properties:
        src_id: { type: string }
        dst_id: { type: string }
        src_uri: { type: string }
        dst_uri: { type: string }
        pipeline: { type: string }
        row: { type: string }
        first_seen: { type: string, format: date-time }
        last_seen: { type: string, format: date-time }
        last_run_id: { type: string }
        runs: { type: integer }
        last_records: { type: integer }
        column_lineage: { type: object, nullable: true, description: "Column-lineage facet from the most recent run ({fields: {out: [in, …]}})." }
      example:
        src_id: orders_to_warehouse
        dst_id: orders_to_warehouse
        src_uri: example
        dst_uri: example
        pipeline: example
        row: example
        first_seen: '2026-01-01T00:00:00Z'
        last_seen: '2026-01-01T00:00:00Z'
        last_run_id: run_01HZY8K3QK7
        runs: 1
        last_records: 1
        column_lineage:
          key: value
    LocalOutput:
      type: object
      description: >
        One recorded local sink output file. The `record` fields are the stored
        ledger row; `state`, `age_secs`, and `retention_days_effective` are
        derived server-side so clients cannot disagree about what "expired" means.
      required: [id, path, dataset_uri, dataset_id, kind, pipeline, row, run_id, pre_existing, first_written_at, last_written_at, state, age_secs]
      properties:
        id: { type: string, description: "Stable id (sha256 prefix of the path); used in the delete route." }
        path: { type: string, description: "The concrete file faucet wrote." }
        dataset_uri: { type: string, description: "Canonical dataset URI of the writing sink." }
        dataset_id: { type: string, description: "Catalog dataset id, so an output groups under its dataset." }
        kind: { type: string, description: "Connector kind (jsonl / csv / parquet)." }
        pipeline: { type: string }
        row: { type: string, description: "Matrix row id (default for non-matrix runs)." }
        run_id: { type: string, description: "The run that most recently wrote this file." }
        pre_existing:
          type: boolean
          description: >
            The file already existed when faucet first opened it — faucet wrote
            to a file it did not create. Never deleted by any cleanup scope.
        retention_days: { type: integer, nullable: true, description: "Per-pipeline override from `local_outputs.retention_days` (0 = keep forever)." }
        first_written_at: { type: string, format: date-time }
        last_written_at: { type: string, format: date-time, description: "The instant retention is measured from." }
        deleted_at: { type: string, format: date-time, nullable: true, description: "When the GC collected the file. The record is kept." }
        deleted_bytes: { type: integer, nullable: true, description: "Bytes reclaimed when it was collected." }
        state:
          type: string
          enum: [present, expired, external, replaced]
          description: "present = on disk; expired = collected (record kept); external = faucet appended to a file it did not create; replaced = faucet truncated a file it did not create (previewable, never collected)."
        age_secs: { type: integer, description: "Seconds since the last write." }
        retention_days_effective: { type: integer, nullable: true, description: "The window actually in force; null = never expires." }
      example:
        id: 9f2b1c4d5e6f7a8b
        path: /work/out/orders.jsonl
        dataset_uri: file:///work/out/orders.jsonl
        dataset_id: 4a1f0c8e2b7d6a95
        kind: jsonl
        pipeline: orders_to_warehouse
        row: default
        run_id: run_01HZY8K3QK7
        pre_existing: false
        first_written_at: '2026-01-01T00:00:00Z'
        last_written_at: '2026-01-01T00:00:00Z'
        state: present
        age_secs: 3600
        retention_days_effective: 7
    SweepOutcome:
      type: object
      required: [id, path, dataset_uri, deleted, bytes]
      properties:
        id: { type: string }
        path: { type: string }
        dataset_uri: { type: string }
        deleted: { type: boolean, description: "The file was deleted (or, in a dry run, would be)." }
        bytes: { type: integer, description: "Bytes reclaimed; 0 when nothing was deleted." }
        skipped:
          type: string
          nullable: true
          enum: [pre_existing, already_deleted, not_on_disk, in_flight, delete_failed]
          description: >
            Why the file was left alone. `pre_existing` = faucet did not create
            it; `not_on_disk` = already gone (a no-op, and the record is marked
            expired); `in_flight` = the file may still be being written — either
            its run is executing, or the file was touched inside the server's
            in-flight grace window — so it is retried later.
        error: { type: string, nullable: true, description: "Detail for a delete_failed outcome." }
    SweepReport:
      type: object
      required: [dry_run, scope, deleted, bytes, skipped, outputs]
      properties:
        dry_run: { type: boolean, description: "Nothing was deleted; the report is what would happen." }
        scope: { type: string, enum: [output, dataset, run, older_than, expired, all] }
        deleted: { type: integer, description: "Files deleted." }
        bytes: { type: integer, description: "Total bytes reclaimed." }
        skipped: { type: integer, description: "In-scope outputs left alone." }
        outputs:
          type: array
          items: { $ref: "#/components/schemas/SweepOutcome" }
      example:
        dry_run: false
        scope: expired
        deleted: 2
        bytes: 40960
        skipped: 1
        outputs:
          - { id: 9f2b1c4d5e6f7a8b, path: /work/out/orders.jsonl, dataset_uri: "file:///work/out/orders.jsonl", deleted: true, bytes: 40960 }
          - { id: 1a2b3c4d5e6f7081, path: /shared/theirs.csv, dataset_uri: "file:///shared/theirs.csv", deleted: false, bytes: 0, skipped: pre_existing }
    CatalogDatasetDetail:
      allOf:
        - { $ref: "#/components/schemas/CatalogDataset" }
        - type: object
          required: [schema_timeline, stats, upstream, downstream]
          properties:
            schema_timeline:
              type: array
              description: Oldest first.
              items: { $ref: "#/components/schemas/CatalogSchemaVersion" }
            stats:
              type: array
              description: Most recent volume points, newest first (bounded).
              items: { $ref: "#/components/schemas/CatalogStatsPoint" }
            upstream:
              type: array
              items: { $ref: "#/components/schemas/CatalogLineageEdge" }
            downstream:
              type: array
              items: { $ref: "#/components/schemas/CatalogLineageEdge" }
    ParamSpec:
      type: object
      description: One entry of a config's `params:` block — the trigger-time override surface.
      properties:
        type:
          type: string
          enum: [string, int, float, bool]
          default: string
        required: { type: boolean, default: false, description: "Caller must supply a value. Mutually exclusive with `default`." }
        default: { description: "Value used when the caller supplies none. Resolved like any other config scalar, so `${env:X}` works." }
        secret: { type: boolean, default: false, description: "Registered for log/response redaction the moment it is bound; never persisted." }
        description: { type: string }
      example:
        type: string
        required: false
        default: example
        secret: false
        description: Orders -> BigQuery, hourly
    TemplateSummary:
      type: object
      description: Body-free view of one registered template version.
      required: [id, version, params, created_at]
      properties:
        id: { type: string, description: "Registry id (slug)." }
        kind:
          type: string
          enum: [source-template, sink-template, deployment, pipeline]
          default: pipeline
          description: "What the document is. A source template runs composed with a sink template; a sink template is composed into a source template's run; a deployment overlays state / DLQ / notifications / SLA onto such a run; a pipeline runs alone."
        version: { type: integer, description: "The newest registered version (the build tip). `state.stable` is what an unpinned run uses." }
        name: { type: string, nullable: true, description: "The config's own `name:`." }
        description: { type: string, nullable: true }
        params:
          type: object
          additionalProperties: { $ref: "#/components/schemas/ParamSpec" }
        created_at: { type: string, format: date-time }
        created_by: { type: string, nullable: true, description: "Principal that registered this version." }
        state:
          allOf: [{ $ref: "#/components/schemas/TemplateState" }]
          description: "Release state of the template as a whole. Present on the read paths."
      example:
        id: orders_to_warehouse
        version: 3
        kind: pipeline
        name: orders_to_warehouse
        description: Orders -> BigQuery, hourly
        params:
          key: value
        created_at: '2026-01-01T00:00:00Z'
        created_by: example
        state: example
    HubIndex:
      type: object
      description: A Template Hub catalog index — the shape of `hub/index.json` and of `GET /v1/templates/matrix`.
      required: [version, sources, sinks, matrix]
      properties:
        version: { type: integer, enum: [1] }
        commit: { type: string, nullable: true, description: "The catalog commit the index describes (catalog-generated indexes only)." }
        sources:
          type: array
          items:
            type: object
            required: [name, streams, params]
            properties:
              id: { type: string, description: "Hub id — `owner/name` (the official set is `faucet-hq/name`)." }
              owner: { type: string, nullable: true, description: "Publisher namespace (GitHub user or org login); `faucet-hq` for an official template." }
              official: { type: boolean }
              newest: { type: integer, nullable: true, description: "Catalog version count (v1, v2, …); present when the index carries history." }
              stable: { type: integer, nullable: true, description: "The version an unpinned run uses." }
              versions:
                type: array
                items:
                  type: object
                  properties:
                    version: { type: integer }
                    commit: { type: string }
                    date: { type: string, nullable: true }
                    pr: { type: integer, nullable: true }
              name: { type: string }
              description: { type: string, nullable: true }
              tags: { type: array, items: { type: string } }
              docs: { type: string, nullable: true }
              source_type: { type: string, description: "Connector kind (rest, csv, …)." }
              file: { type: string, description: "Catalog path, or the registry id." }
              streams:
                type: array
                items:
                  type: object
                  properties:
                    name: { type: string }
                    description: { type: string, nullable: true }
                    write: { type: array, items: { type: string }, description: "Write preference, in order." }
                    primary_keys: { type: array, items: { type: string } }
              params: { type: array, items: { type: object } }
        sinks:
          type: array
          items:
            type: object
            required: [name, write_modes, params]
            properties:
              id: { type: string, description: "Hub id — `owner/name` (the official set is `faucet-hq/name`)." }
              owner: { type: string, nullable: true }
              official: { type: boolean }
              newest: { type: integer, nullable: true }
              stable: { type: integer, nullable: true }
              name: { type: string }
              description: { type: string, nullable: true }
              tags: { type: array, items: { type: string } }
              docs: { type: string, nullable: true }
              sink_type: { type: string }
              file: { type: string }
              write_modes: { type: array, items: { type: string }, description: "Modes the connector supports natively." }
              params: { type: array, items: { type: object } }
        matrix:
          type: array
          items:
            type: object
            required: [source, sink, compatible, streams]
            properties:
              source: { type: string }
              sink: { type: string }
              compatible: { type: boolean }
              streams:
                type: array
                items:
                  type: object
                  properties:
                    stream: { type: string }
                    write_mode: { type: string }
                    satisfies: { type: string, nullable: true, description: "The requested mode this one stands in for, via an alias." }
              incompatible:
                type: array
                items:
                  type: object
                  properties:
                    stream: { type: string }
                    reason: { type: string }
              command: { type: string, nullable: true, description: "Copy-paste run command for a compatible pairing." }
    TemplateSyncReport:
      type: object
      description: The plan for one origin and, unless `dry_run`, what applying it did.
      required: [origin, kind, dry_run, warnings, plan]
      properties:
        origin: { type: string }
        kind: { type: string, enum: [github, s3, gcs, azure_blob] }
        dry_run: { type: boolean }
        warnings:
          type: array
          items: { type: string }
          description: Remote files that were recognized but could not be used (broken sidecar, ambiguous stem, sidecar without a template).
        plan:
          type: array
          items:
            type: object
            required: [action]
            properties:
              action: { type: string, enum: [register, launch, revive, unchanged, orphaned, deprecate, skipped] }
              id: { type: string }
              version: { type: integer }
              launch: { type: boolean }
              replaces: { type: integer, description: "Version a `register` supersedes." }
              description: { type: string }
              tags: { type: array, items: { type: string } }
              name: { type: string, description: "Remote file stem of a `skipped` entry." }
              reason: { type: string }
        outcome:
          type: object
          description: Present unless `dry_run`.
          properties:
            registered:
              type: array
              items: { type: object, properties: { id: { type: string }, version: { type: integer }, launched: { type: boolean } } }
            launched:
              type: array
              items: { type: object, properties: { id: { type: string }, version: { type: integer }, launched: { type: boolean } } }
            revived: { type: array, items: { type: string } }
            deprecated: { type: array, items: { type: string } }
            unchanged: { type: integer }
            orphaned: { type: array, items: { type: string } }
            skipped:
              type: array
              items: { type: array, items: { type: string }, minItems: 2, maxItems: 2 }
            failed:
              type: array
              items: { type: object, required: [id, error], properties: { id: { type: string }, error: { type: string } } }
    TemplateStatus:
      type: string
      description: >
        Lifecycle state of the **template** (not of a version), derived so it can
        never disagree with the registry: `draft` until something is launched,
        `launched` afterwards, `deprecated` once explicitly retired.
      enum: [draft, launched, deprecated]
    TemplateState:
      type: object
      description: >
        The template's whole release state. Everything except `tags` and
        `deprecation` is derived from the append-only launch log, so a pointer can
        never outlive its target.
      required: [status, versions, tags]
      properties:
        status: { $ref: "#/components/schemas/TemplateStatus" }
        versions:
          type: array
          description: Every stored version, newest first.
          items: { type: integer }
        stable: { type: integer, nullable: true, description: "The launched version — what `stable` and an unpinned request resolve to. Null while the template is a draft." }
        previous: { type: integer, nullable: true, description: "The version launched before the current one; the rollback target." }
        newest: { type: integer, nullable: true, description: "Highest version number that is not deprecated, launched or not. Null when every version is deprecated." }
        tags:
          type: object
          description: "Assignable channel pointers ({channel: version}), excluding the derived `stable` / `previous` / `newest`."
          additionalProperties: { type: integer }
        deprecation:
          allOf: [{ $ref: "#/components/schemas/DeprecationRecord" }]
          description: "Present only when `status` is `deprecated`."
        deprecated_versions:
          type: array
          description: "Individually retired versions, newest first. Omitted when there are none."
          items:
            allOf:
              - { $ref: "#/components/schemas/DeprecationRecord" }
              - type: object
                required: [version]
                properties:
                  version: { type: integer }
      example:
        status: draft
        versions:
        - 1
        stable: 1
        previous: 1
        newest: 1
        tags:
          key: value
        deprecation: example
    DeprecationRecord:
      type: object
      required: [deprecated_at]
      properties:
        deprecated_at: { type: string, format: date-time }
        deprecated_by: { type: string, nullable: true }
        reason: { type: string, nullable: true }
      example:
        deprecated_at: '2026-01-01T00:00:00Z'
        deprecated_by: example
        reason: example
    LaunchRecord:
      type: object
      description: One entry of the append-only launch log — who blessed which build, and when.
      required: [seq, version, launched_at]
      properties:
        seq: { type: integer, description: "Monotonic per-template sequence, from 1." }
        version: { type: integer }
        launched_at: { type: string, format: date-time }
        launched_by: { type: string, nullable: true, description: "Principal that launched it; null for a CLI launch." }
      example:
        seq: 1
        version: 3
        launched_at: '2026-01-01T00:00:00Z'
        launched_by: example
    VersionChannel:
      type: string
      description: >
        A named version channel — a pointer at one numeric version. The set is
        **closed**: an unknown or mistyped name is rejected rather than creating a
        channel nobody watches. `stable` (the launched version — the default for
        an unpinned request), `previous`, and `newest` are **derived** and cannot
        be assigned; the rest are assigned with `POST /v1/templates/{id}/tags`.
        `latest` is deliberately **not** a channel: it reads as both "newest
        build" and "current stable", so it is rejected with a message naming both.
      enum: [stable, previous, newest, dev, test, staging, pre-prod, canary, prod]
    VersionSelector:
      description: >
        A template version selector: a `VersionChannel` name, a numeric string, or
        a bare number. Omitting it means `stable` — so a freshly registered build
        never moves existing callers. `0`, `latest`, and unknown channel names are
        rejected.
      oneOf:
        - { $ref: "#/components/schemas/VersionChannel" }
        - { type: integer, minimum: 1 }
    TemplateDetail:
      description: >
        One version, plus the template's whole release state (flattened) — so a
        client can pin, promote, launch, or roll back without a second request.
      allOf:
        - { $ref: "#/components/schemas/TemplateSummary" }
        - { $ref: "#/components/schemas/TemplateState" }
        - type: object
          required: [body, format, is_stable, launches]
          properties:
            body: { type: string, description: "The config document, stored verbatim (interpolation directives unresolved)." }
            format: { type: string, enum: [yaml, json] }
            is_stable: { type: boolean, description: "Whether the returned version is the currently launched one." }
            launches:
              type: array
              description: The launch log, newest first.
              items: { $ref: "#/components/schemas/LaunchRecord" }
    LaunchResponse:
      type: object
      description: Result of a launch or rollback.
      required: [id, version, already_launched, status]
      properties:
        id: { type: string }
        version: { type: integer, description: "The version now live." }
        replaced: { type: integer, description: "The version it replaced — the new `previous`. Absent on a first launch." }
        already_launched: { type: boolean, description: "True when the version was already live, so nothing changed." }
        status: { $ref: "#/components/schemas/TemplateStatus" }
      example:
        id: orders_to_warehouse
        version: 3
        replaced: 1
        already_launched: false
        status: draft
    RegisterTemplateRequest:
      type: object
      required: [config]
      properties:
        id: { type: string, description: "Registry id. Derived from the config's `name:` when omitted. A `source-template` / `sink-template` is always registered under its own hub id — `owner/name` (an explicit id must match it)." }
        config: { type: string, description: "The template document, stored verbatim. Its `kind:` decides how it is validated and run: `source-template` (a system and its streams), `sink-template` (a destination), or `pipeline` (a complete config). A document without `kind:` is registered as a pipeline and flagged as deprecated input." }
        config_format: { type: string, enum: [yaml, json], default: yaml }
        description: { type: string }
        tags:
          type: array
          description: "Assignable channels to point at the newly registered version. The version number always auto-increments; a derived channel (`stable` / `previous` / `newest`) is rejected."
          items: { $ref: "#/components/schemas/VersionChannel" }
        launch:
          type: boolean
          default: false
          description: >
            Launch the newly registered version immediately, making it `stable`.
            Off by default: registering a build must never move existing callers.
      example:
        id: orders_to_warehouse
        config_format: yaml
        config: |
          version: 1
          pipeline:
            source:
              type: postgres
              config: { connection_url: ${env:PG_URL}, query: SELECT * FROM orders }
            sink:
              type: bigquery
              config: { dataset: analytics, table: orders }
        description: Orders → BigQuery
        launch: true
    TriggerTemplateRequest:
      type: object
      description: >
        Everything after `params` / `env` / `version` / `sink` / `sink_version`
        mirrors `SubmitRunRequest`, because the run is submitted through the
        same path. For a `source-template` the trigger composes it with the
        named `sink` (a registered `sink-template`) and binds the merged
        params; a `pipeline` template takes no `sink`; a `sink-template` is
        never runnable on its own.
      properties:
        params:
          type: object
          description: "Values for the template's declared params (for a source template: the union of the source's and the sink's). Scalars; coerced to the declared type."
          additionalProperties: true
        sink:
          type: string
          nullable: true
          description: "For a `source-template`: the registered `sink-template` to compose with. Required for a source template (`422` without it); refused for a `pipeline`."
        sink_version:
          description: "Version of the sink template: `stable` (default), another channel, or an exact number."
          $ref: "#/components/schemas/VersionSelector"
        overlay:
          description: "Deployment overlay for a composed run (#679): a registered `kind: deployment` id, or an inline mapping of operational blocks (`state`, `dlq`, `notifications`, `sla`, `resilience`, `execution`, `delivery`, `schedule`, and per-stream `streams`). An inline overlay may omit `kind` and `name`. Refused for a `pipeline` template; anything that would change connectors or streams is a `422`."
          nullable: true
          oneOf:
            - { type: string }
            - { type: object, additionalProperties: true }
        overlay_version:
          description: "Version of a registered overlay: `stable` (default), another channel, or an exact number."
          $ref: "#/components/schemas/VersionSelector"
        env:
          type: object
          description: "Values that win over the server's environment for `${env:VAR}` during this materialization only."
          additionalProperties: { type: string }
        version:
          description: "Version to run: `stable` (the default when omitted), another channel, or an exact version number."
          $ref: "#/components/schemas/VersionSelector"
        name: { type: string, description: "Run name override (default: the template config's `name:`)." }
        labels:
          type: object
          additionalProperties: { type: string }
        timeout_secs: { type: integer }
        doctor_first: { type: boolean, default: false }
        idempotency_key: { type: string }
        clock: { type: string, description: "RFC3339 or YYYY-MM-DD override for `${now.*}`." }
        concurrency: { type: integer, minimum: 1, nullable: true, description: "Override this run's connector concurrency (source + sink connections/fetches), whatever the config says. Maps onto whichever knob the connector declares (max_connections / request_concurrency / partition_concurrency / shard_concurrency / concurrency); a connector with none ignores it. Does not change matrix parallelism or the server's --max-concurrent slots. Per-shard for a sharded run. 0 is rejected." }
      example:
        params:
          region: us-east-1
          since: "2026-01-01"
        env:
          PG_URL: postgres://user:pass@db:5432/app
        version: stable
        sink: bigquery
        sink_version: stable
        overlay: prod-ops
        idempotency_key: 2026-01-01-orders
    TriggerTemplateResponse:
      type: object
      required: [run_id, status, submitted_at, template_id, template_version, params, streams]
      properties:
        run_id: { type: string }
        status: { type: string, enum: [queued, pending] }
        submitted_at: { type: string, format: date-time }
        template_id: { type: string }
        template_version: { type: integer }
        sink_template: { type: string, nullable: true, description: "The sink template composed in (source-template runs only)." }
        sink_template_version: { type: integer, nullable: true }
        overlay: { type: string, nullable: true, description: "The deployment overlay applied: a registered id, or `inline`." }
        overlay_version: { type: integer, nullable: true, description: "Version of a registered overlay (absent for an inline one)." }
        overlay_contributes:
          type: array
          items: { type: string }
          description: "What the overlay set, as config paths (`pipeline.state`, `matrix.orders.dlq`, …)."
        warnings:
          type: array
          items: { type: string }
          description: "Warnings about the composed run — e.g. incremental streams with no state store."
        streams:
          type: array
          description: "Per-stream write-mode plan of a composed run (empty for a `pipeline` template)."
          items:
            type: object
            required: [stream, requested, chosen]
            properties:
              stream: { type: string, description: "Stream name — the matrix row id and destination table." }
              requested: { type: array, items: { type: string }, description: "The stream's write preference list." }
              chosen: { type: string, description: "The mode the sink runs it as (natively, or through a declared alias)." }
              satisfies: { type: string, nullable: true, description: "The requested mode `chosen` stands in for, when an alias was used." }
              key: { type: array, items: { type: string }, description: "The upsert key (the stream's `primary_keys`) for keyed modes." }
        params:
          type: object
          description: "The bound params, with every `secret: true` value replaced by `***`."
          additionalProperties: true
        deprecated:
          type: string
          description: >
            Present only when the template is deprecated — the run still started,
            but the caller should migrate. Carries the retirement reason when one
            was given.
      example:
        run_id: run_01HZY8K3QK7
        status: queued
        submitted_at: '2026-01-01T00:00:00Z'
        template_id: orders_to_warehouse
        template_version: 3
        sink_template: bigquery
        sink_template_version: 1
        overlay: prod-ops
        overlay_version: 2
        overlay_contributes: [pipeline.state, pipeline.dlq]
        streams:
          - { stream: orders, requested: [overwrite, upsert], chosen: overwrite, key: [id] }
        params:
          key: value
        deprecated: example
    ApiError:
      type: object
      required: [error]
      properties:
        error:
          type: object
          required: [code, message]
          properties:
            code: { type: string }
            message: { type: string }
            details: { type: object, nullable: true }

      example:
        error:
          code: invalid_config
          message: 'config failed validation: missing sink'
          details:
            key: value