# prefect-server roadmap

comparison with prefect server (python). checkmarks indicate implemented.

## api endpoints

### core (required for basic flow execution)
- [x] GET /api/health
- [x] GET /api/csrf-token
- [x] GET /api/admin/version
- [x] POST /api/flows/
- [x] GET /api/flows/{id}
- [x] POST /api/flows/filter
- [x] PATCH /api/flows/{id}
- [x] DELETE /api/flows/{id}
- [x] POST /api/flow_runs/
- [x] GET /api/flow_runs/{id}
- [x] POST /api/flow_runs/{id}/set_state
- [x] POST /api/flow_runs/filter
- [x] PATCH /api/flow_runs/{id}
- [x] DELETE /api/flow_runs/{id}
- [x] POST /api/logs/
- [x] POST /api/logs/filter
- [x] WS /api/events/in

### task runs
- [x] POST /api/task_runs/
- [x] GET /api/task_runs/{id}
- [x] POST /api/task_runs/{id}/set_state
- [x] POST /api/task_runs/filter

### state endpoints
- [x] GET /api/flow_run_states/{id}
- [x] GET /api/flow_run_states/?flow_run_id=...
- [x] GET /api/task_run_states/{id}
- [x] GET /api/task_run_states/?task_run_id=...

### deployments
- [x] POST /api/deployments/
- [x] GET /api/deployments/{id}
- [x] GET /api/deployments/name/{flow_name}/{deployment_name}
- [x] POST /api/deployments/filter
- [x] POST /api/deployments/count
- [x] PATCH /api/deployments/{id}
- [x] DELETE /api/deployments/{id}
- [x] POST /api/deployments/{id}/create_flow_run
- [x] POST /api/deployments/{id}/pause_deployment
- [x] POST /api/deployments/{id}/resume_deployment
- [x] GET /api/deployments/{id}/schedules
- [x] POST /api/deployments/{id}/schedules
- [x] PATCH /api/deployments/{id}/schedules/{schedule_id}
- [x] DELETE /api/deployments/{id}/schedules/{schedule_id}
- [x] POST /api/deployments/get_scheduled_flow_runs

### work pools & workers
- [x] POST /api/work_pools/
- [x] GET /api/work_pools/{name}
- [x] PATCH /api/work_pools/{name}
- [x] DELETE /api/work_pools/{name}
- [x] POST /api/work_pools/filter
- [x] POST /api/work_pools/{name}/queues/
- [x] GET /api/work_pools/{name}/queues/{queue_name}
- [x] PATCH /api/work_pools/{name}/queues/{queue_name}
- [x] DELETE /api/work_pools/{name}/queues/{queue_name}
- [x] POST /api/work_pools/{name}/queues/filter
- [x] POST /api/work_pools/{name}/workers/heartbeat
- [x] POST /api/work_pools/{name}/workers/filter
- [x] DELETE /api/work_pools/{name}/workers/{worker_name}
- [x] POST /api/work_pools/{name}/get_scheduled_flow_runs

### blocks
- [x] POST /api/block_types/
- [x] POST /api/block_types/filter
- [x] GET /api/block_types/slug/{slug}
- [x] PATCH /api/block_types/{id}
- [x] GET /api/block_types/slug/{slug}/block_documents
- [x] GET /api/block_types/slug/{slug}/block_documents/name/{name}
- [x] POST /api/block_schemas/
- [x] POST /api/block_schemas/filter
- [x] GET /api/block_schemas/{id}
- [x] GET /api/block_schemas/checksum/{checksum}
- [x] POST /api/block_documents/
- [x] POST /api/block_documents/filter
- [x] GET /api/block_documents/{id}
- [x] PATCH /api/block_documents/{id}
- [x] DELETE /api/block_documents/{id}
- [ ] GET /api/block_capabilities/ (optional - discovery only, not used by client)

### concurrency (v2 only - skip v1 API)
- [x] POST /api/v2/concurrency_limits/
- [x] POST /api/v2/concurrency_limits/filter
- [x] GET /api/v2/concurrency_limits/{id}
- [x] DELETE /api/v2/concurrency_limits/{id}
- [x] POST /api/v2/concurrency_limits/increment (acquire slots, returns lease_id)
- [x] POST /api/v2/concurrency_limits/decrement (release slots by limit names)
- [x] lease storage (memory default, redis for HA - matches Python)
- [x] lease_cleanup background service (expires stale leases)
- note: v1 tag-based concurrency was reconstituted to use v2 in python prefect.
  we should only implement the v2 API to avoid the complexity of supporting both.

### artifacts
- [ ] POST /api/artifacts/
- [ ] POST /api/artifacts/filter

### automations
- [ ] POST /api/automations/
- [ ] POST /api/automations/filter

### variables
- [x] POST /api/variables/
- [x] POST /api/variables/filter
- [x] POST /api/variables/count
- [x] GET /api/variables/{id}
- [x] GET /api/variables/name/{name}
- [x] PATCH /api/variables/{id}
- [x] PATCH /api/variables/name/{name}
- [x] DELETE /api/variables/{id}
- [x] DELETE /api/variables/name/{name}

### events
- [x] POST /api/events/filter
- [x] GET /api/events/count
- [x] WS /api/events/out (subscribe)

### ui endpoints
- [ ] POST /api/ui/flows/
- [ ] POST /api/ui/flow_runs/
- [ ] POST /api/ui/task_runs/
- [ ] GET /api/ui/schemas/

## orchestration

### global transforms (bookkeeping)
- [x] SetStartTime - set start_time when first entering RUNNING
- [x] SetEndTime - set end_time when entering terminal state
- [x] IncrementRunTime - accumulate total_run_time when exiting RUNNING
- [x] IncrementRunCount - increment run_count when entering RUNNING

### orchestration rules framework
- [x] ResponseStatus enum (ACCEPT, REJECT, WAIT, ABORT)
- [x] OrchestrationRule abstraction (from_states, to_states, before_transition)
- [x] Policy composition (CoreFlowPolicy as ordered list of rules)
- [x] applyPolicy function to run rules in order

### flow run rules (CoreFlowPolicy)
- [x] PreventPendingTransitions - reject PENDING/RUNNING/CANCELLING/CANCELLED → PENDING
- [x] CopyScheduledTime - copy scheduled_time when SCHEDULED → PENDING
- [x] WaitForScheduledTime - delay transition if scheduled_time in future
- [x] RetryFailedFlows - schedule retry when RUNNING → FAILED if retries available
- [x] PreventDuplicateTransitions - idempotency via transition_id
- [ ] HandleFlowTerminalStateTransitions - prevent leaving completed with persisted data
- [ ] HandlePausingFlows - govern RUNNING → PAUSED
- [ ] HandleResumingPausedFlows - govern PAUSED → RUNNING

### task run rules (CoreTaskPolicy)
- [ ] CacheRetrieval - check for cached results before execution
- [ ] CacheInsertion - write completed results to cache
- [ ] PreventRunningTasksFromStoppedFlows - tasks can't run if flow isn't running
- [ ] RetryFailedTasks - govern retry transitions

### other
- [x] concurrency limits (v2 only)

## background services

- [x] event_persister (batched event writes, deduplication, retention trimming)
- [x] event_broadcaster (websocket fan-out to /api/events/out subscribers)
- [x] scheduler (create flow runs from deployment schedules - interval + cron)
- [x] lease_cleanup (expire stale concurrency leases)
- [x] late_runs (mark runs as late)
- [ ] foreman (infrastructure management)
- [ ] cancellation_cleanup (clean up cancelled runs)
- [ ] pause_expirations (expire paused runs)
- [ ] task_run_recorder (record task run events)
- [ ] telemetry

## message broker

- [x] memory broker (growable queue, in-process)
- [x] redis broker (redis streams: XADD/XREADGROUP/XACK)
- [x] broker abstraction layer (broker/core.zig)
- [x] fan-out to consumer groups

## database

- [x] flow table
- [x] flow_run table
- [x] flow_run_state table
- [x] events table
- [x] task_run table
- [x] task_run_state table
- [x] deployment table
- [x] deployment_schedule table
- [x] work_pool table
- [x] work_queue table
- [x] worker table
- [x] block_type table
- [x] block_document table
- [x] block_schema table
- [x] concurrency_limit table
- [ ] artifact table
- [ ] automation table
- [x] variable table
- [x] log table

### database backends
- [x] sqlite (zqlite)
- [x] postgres (pg.zig)
- [x] backend abstraction layer (db/backend.zig)

## infrastructure

- [x] http server (zap/facil.io)
- [x] websocket support
- [x] sqlite database (zqlite)
- [x] postgres database (pg.zig)
- [x] structured logging
- [x] service lifecycle management
- [x] postgres connection pooling (pg.zig built-in)
- [ ] sqlite connection pooling
- [x] migrations
- [x] graceful shutdown
- [ ] configuration file support

## notes

### implementation order for worker-based execution

workers are the primitive - they poll for work and execute runs. order matters:

1. ~~**work_pool** - infrastructure configuration container~~ ✓
   - table + CRUD API
   - types: process, docker, kubernetes, etc.

2. ~~**work_queue** - routes runs to workers~~ ✓
   - table + CRUD API
   - each pool has a default queue
   - queues have priority

3. ~~**worker heartbeat** - workers register and poll~~ ✓
   - `POST /work_pools/{name}/workers/heartbeat`
   - tracks worker health and last seen

4. ~~**deployment** - flow + schedule + work pool binding~~ ✓
   - table + CRUD API
   - links flow_id → work_pool → work_queue
   - `.serve()` support via `POST /deployments/get_scheduled_flow_runs`

5. ~~**deployment_schedule** - cron/interval/rrule schedules~~ ✓
   - table, linked to deployment
   - CRUD API for managing schedules

6. ~~**scheduler service** - creates runs from schedules~~ ✓
   - background service
   - queries deployments needing runs
   - creates flow_runs in SCHEDULED state
   - supports interval + cron schedules
   - parameter merging (schedule overrides deployment)
   - idempotent via idempotency_key

7. ~~**get_scheduled_flow_runs** - workers poll for work~~ ✓
   - `POST /work_pools/{name}/get_scheduled_flow_runs`
   - returns runs ready to execute
   - updates pool/deployment status to READY

### short-term plan
1. ~~**log persistence**~~ ✓ - logs now persisted to database
2. **automations** - event-driven triggers, core feature for production use

### what's working (~5x faster than python)
- flow/flow_run/task_run lifecycle
- blocks (types, schemas, documents)
- variables (full CRUD)
- work pools, work queues, workers (full CRUD + heartbeat)
- deployments + schedules (full CRUD, `.serve()` support)
- scheduler service (interval + cron, idempotent, parameter merging)
- late_runs service (marks overdue scheduled runs as Late)
- get_scheduled_flow_runs (worker polling)
- events (ingest via websocket, persist, broadcast with filtered backfill)
- concurrency limits v2 (CRUD, lease-based slots, memory/redis storage)
- dual database backends (sqlite/postgres)
- dual message brokers (memory/redis)