# DD — Pipeline coordinator (8B)

Pipeline Coordination executes saved steps in order. It sends each step through
Execution Core's existing launch acceptance and planning path, then follows that
Run until the step's advancement condition is met.

## Flow

1. Claim due `pipeline_work` with a database-time lease. Load the Pipeline and its
   progress. Each step starts `waiting`.
2. Save the next step as `launching`, with a deterministic launch-request UUID,
   before calling Execution Core. The Pipeline snapshot, receipt, launch index
   and next work schedule commit together.
3. Submit the saved Workload ID, exact revision, member, Project and environment
   through Execution Core's internal accepted-Pipeline path. Pipeline Coordination's
   published query confirms the exact saved launching step, order and cancellation
   state. Execution Core records its Pipeline/step provenance and owner revision
   atomically with the launch. This continues work accepted at Pipeline start after
   personal suspension, including when the editable Workload has been revised or
   retired. Planning reads that exact revision and checks provisioned resource
   limits before creating the Run. Repository commits are resolved at this point.
4. Save the correlated Run ID, owner version and observed state. Recover an
   uncertain submission through the same launch ID. An already accepted launch
   stays with Execution Core, whose resource, binding and lifecycle checks remain
   in force. Automatic retries preserve the Run's pinned inputs.
5. For batch Runs, Execution Core commits final state and a typed `RunCompleted`
   event together. Its outbox relay publishes to RabbitMQ with publisher confirms.
   The Pipeline consumer correlates the fact through the Run's launch reference.
6. Commit the consumer receipt, step progress and pending coordinator work in one
   Pipeline transaction, then acknowledge the message. Success makes the next
   step eligible. A final failure or cancellation stops advancement. A failed
   Attempt that is still eligible for retry does not fail its Pipeline step.
7. For application and Hugging Face services, query Execution Core's saved Run.
   Advance on `healthy` with a validated HTTP(S) endpoint. This establishes service
   readiness while the service keeps running. The same owner query recovers batch
   completion when a coordinator was interrupted around submission or delivery.
8. Repeat for the next step. All advancement conditions satisfied means
   `succeeded`. Training commands can push model files before this transition.
   The next action is pinned, when it starts, to the commit its branch names then. A recorded
   Pipeline cancellation stops further launches and [cancels active Runs](pipeline-outputs.md).

```mermaid
flowchart TD
    P[(Pipeline PostgreSQL)] --> C[Pipeline coordinator claims work]
    C --> I[Save stable step launch identity]
    I --> A[Execution Core launch acceptance]
    A --> W[Launch worker checks and plans]
    W --> R[(Execution Core Run and Attempt)]
    R --> S[Run Supervisor and existing Site execution path]
    S --> F[Final Run plus completion outbox event]
    F --> Q[RabbitMQ]
    Q --> E[Pipeline completion consumer]
    E --> T[Commit receipt and progress and wake work]
    T --> P
    R --> O[Owner query for service health and recovery]
    O --> C
```

## Records and recovery

The Pipeline remains the aggregate root. Its ordered `steps` preserve the accepted
immutable definition. The separate child `progress` array contains each step's
status, launch-request ID, Run ID, last observed Run version/state/finality and,
for a ready service, its endpoint. Start-command material excludes progress so
replaying a current definition preserves its recorded outcome. Unknown or
inconsistent saved progress fails restoration. The repository correction changes
earlier stored/wire shapes; see the [compatibility note](pipeline-outputs.md#saved-and-wire-compatibility).

| Pipeline database table | Purpose |
| --- | --- |
| `pharos.aggregates` | Pipeline definition and all child progress at one version |
| `pharos.command_receipts` | Original command outcomes and conflicting-retry detection |
| `pharos.consumer_receipts` | Completion-event identity, byte fingerprint and original outcome |
| `pharos.pipeline_work` | Due time, lease owner/token/generation, claimed root version and failures |
| `pharos.pipeline_launches` | Rebuildable correlation index from saved step launch to Pipeline and Run |

A step launch UUID is derived from the Pipeline and step UUIDs. Retrying a lost
response reuses the original accepted envelope. The coordinator never writes a
Run or reads Execution Core tables directly. Its outer bridge calls Execution
Core's launch and progress interfaces, whose adaptors use that context's database.
Each context commits its own aggregate transaction.

New step admission requires an available Pipeline owner and matching saved scope.
It refuses unknown steps, changed scope, out-of-order dispatch and cancellation.
Human launch requests cannot select this internal continuation path. Proven saved
launches can be recovered during an owner outage without renewing their evidence.
Older launch snapshots still restore, but a coordinator cannot adopt one without
Pipeline provenance as accepted continuation. Cancellation and cleanup remain
available for correctly correlated historical launches. Paused development
installations must recreate unproven Pipeline fixtures before resuming them.

Completion can arrive before the worker saves its Run reference. The launch
index already exists, so the consumer can adopt the correctly correlated Run.
Duplicate deliveries return the consumer receipt. A crash after commit and before
acknowledgement causes redelivery without a second advancement. Incoming events
and cancellation invalidate an older worker's lease. Expired token/generation or
root versions cannot commit. Root-before-work locking applies to all writers.

An already successful Pipeline keeps its result when a previously ready service
later terminates. While a Pipeline is still running, a final service failure
stops advancement. Execution Core continues owning that service's lifecycle.

Repository contents are fetched when each action prepares. Training scripts or
post-commands use Git/LFS to publish their outputs; later actions fetch the latest
selected branch. [Repository hand-offs, cancellation and remaining connections](pipeline-outputs.md).

## Process and deployment

`services/pipeline_coordinator.py` composes the coordinator loop, Execution Core
outbox relay and Pipeline completion consumer. The shared Metallo HTTP process
keeps its empty delivery registries. The dedicated process receives separate
restricted context database bindings and scoped broker credentials.

`contracts/subscriptions.json` registers
`pipeline_coordination.run-completed` for `execution_core.run.completed.v1`.
Administration provisions the exact durable queue, retry/dead queues and binding
`integration.execution_core.run.completed.v1`. Runtime consumers cannot create
topology; the Execution Core publisher can publish only its registered completion
contract. Reuse the existing versioned Protobuf envelope, codec and delivery code.
Service readiness uses the owner query and needs no additional event contract.

The GitOps base and overlays include the coordinator Deployment, ServiceAccount,
scoped Secret references, NetworkPolicies, probes, resource limits and graceful
shutdown. `create_app()` builds the owner adaptors in code: Workloads from Workload
Portfolio, accepted-step evidence from Pipeline Coordination, and current-person
authority through Beto's enrolled query interface and Execution Core's local view.
The overlays install zero replicas pending #771 qualification. `compose(...)`
takes the owner ports; the qualification
entry `tests/infrastructure/pipeline_native_coordinator.py` passes labelled doubles.
[Deployment configuration](../../../../../gitops/manifests/pharos/base/pipeline-coordinator/README.md)
describes enabling it. The execution worker and native Runtime retain their own
provisioning, credentials and provider-integration responsibilities.

## Verification

Verification is recorded separately for the coordinator's implementation, real
PostgreSQL and RabbitMQ, production Execution Core adaptors, and external owners.
The reproducible cases are:

- [Domain and application](../../../tests/pipeline_coordination/test_coordinator.py)
  and [Run-event rules](../../../tests/pipeline_coordination/test_run_events.py).
- [PostgreSQL recovery](../../../tests/infrastructure/pipeline_coordinator.py)
  and [rollback and fencing](../../../tests/infrastructure/pipeline_coordinator_failures.py).
- [Production acceptance and planning](../../../tests/infrastructure/pipeline_execution.py),
  including both service-kind readiness queries and permission revocation between steps.
- [Real RabbitMQ delivery](../../../tests/infrastructure/pipeline_delivery.py),
  including a lost acknowledgement after commit, redelivery, early completion
  and denied runtime topology/foreign publication.
- [Process lifecycle](../../../tests/test_pipeline_coordinator_service.py),
  including missing providers, worker startup, health and shutdown.

Workload, permission and provisioning authorities and native observations in these
coordinator tests are explicit test doubles. PostgreSQL storage, receipts, codecs,
Execution Core acceptance/planning and RabbitMQ are production implementations.
Repository hand-offs and native Kubernetes pipelines are recorded under
[8C](pipeline-outputs.md) and [8D](pipeline-verification.md). Real provider wiring,
Slurm/mixed pipelines and enabled permanent GitOps deployment remain open.

Completed on 20 October 2026:

| Evidence | Result |
| --- | --- |
| Pipeline policy and handlers | 364 domain/application cases passed |
| PostgreSQL and RabbitMQ | 72 combined Pipeline integration cases passed, including existing 8A cases, owner acceptance/planning, real completion delivery and the running coordinator process |
| Full verification | `pnpm verify` passed, including 1,671 Metallo tests, strict Mypy over 251 source files, Go checks, frontend/DX builds and deployment checks |
| Disposable repository integration | `pnpm test:integration` passed with real PostgreSQL, RabbitMQ, application images, sessions and the existing browser regressions |
| Generation | Pinned `pnpm generate:contracts` reproduced outputs without drift |
| Isolated dev-box PostgreSQL lane | 1,444 cases passed, including the coordinator and production launch/planning adaptors. The owned test namespace was deleted |

The dev-box lane uses real PostgreSQL and production owner adaptors. Its Workload,
permission and provisioning responses and native observations are test doubles.
RabbitMQ and the running coordinator process were exercised in the disposable
local integration lane. [8D records native Kubernetes pipeline verification](pipeline-verification.md).
Slurm/mixed pipelines and enabling the permanent GitOps process with real providers
remain open.