Skip to main content
Adding a public workflow in Fleet Manager usually touches three layers:
  • The Temporal workflow and activities
  • The workflow-agent registration
  • The API model and seeded defaults
If the workflow is internal-only, implement the Temporal layer and set __api_exposed__ = False.
Before authoring or reviewing a workflow, use the repository’s temporal-workflows agent skill at .agents/skills/temporal-workflows/SKILL.md. It is the required checklist for activity boundaries, determinism and replay safety, failure tracking, idempotency, and the satellite no-ORM rule.

​
Workflow compatibility

Use the production data converter for workflow tests and replay recording. Preserve commands and optional-field behavior in existing histories. Use a patch guard when commands change. Replay existing histories and new histories that cover each changed decision. Deploy compatible readers and validation before writers produce new payloads. Confirm worker readiness before enabling new planning contracts. Preserve readers for stored payloads during rollback.

​
Files you usually change

  • src/fleet_manager/temporal/workflows/<name>.py
  • src/fleet_manager/temporal/activities/<name>.py
  • src/fleet_manager/temporal/workflows/__init__.py
  • src/fleet_manager/temporal/worker.py
  • src/fleet_manager/core/workflow/names.py
  • src/fleet_manager/agents/workflow/registry.py
  • src/fleet_manager/core/eligibility/policies.py
  • src/fleet_manager/api/core/models.py
  • src/fleet_manager/api/core/signals.py
  • src/fleet_manager/api/core/migrations/<new_migration>.py
  • Tests under tests/temporal/ and tests/api/core/
You usually do not need to edit src/fleet_manager/agents/workflow/submitter.py; it already handles request building and schema validation for registered workflows.

​
1. Pick the execution model

Decide this first. It drives the request type, registry entry, and progress handling.
  • Use ExecutionMode.EMBARRASSINGLY_PARALLEL for one workflow per device.
  • Use ExecutionMode.NATIVELY_BATCHED for one workflow handling many devices.
Request types:
  • Single-device: WorkflowRequest[ParamsT]
  • Batched: BatchWorkflowRequest[ParamsT]
For batched workflows:
  • Subclass NativelyBatchedWorkflow
  • Call _init_progress(req.metadata.devices) at the start of run(...)
  • Mark device completion or failure through the helper methods on NativelyBatchedWorkflow

​
2. Implement the Temporal workflow

Create the params model and workflow in src/fleet_manager/temporal/workflows/<name>.py. Rules:
  • Always use a typed Pydantic params model, even if it is empty.
  • Keep run(...) orchestration-only.
  • Put all external I/O in activities.
  • The workflow class name must match the enum value you add to TemporalWorkflowName.
  • Use sandboxed=False only when the workflow actually needs it.
Repository conventions:
  • Configure retries with RetryPolicy
  • Read retry limits from settings when operators may need to tune them
  • Re-raise ApplicationError causes from ActivityError
  • Wrap unexpected workflow exceptions in a non-retryable ApplicationError
For batched workflows, update device status before re-raising:
  • Mark devices FAILED on terminal errors
  • Mark devices CANCELLED on workflow cancellation
  • Mark devices COMPLETED only after the full workflow succeeds

​
3. Put external work in activities

Create activities in src/fleet_manager/temporal/activities/<name>.py. Expected pattern:
  • Build an activity class with injected dependencies
  • Instantiate it once in the worker
  • Register bound methods with Temporal
Activities should classify failures explicitly:
  • Fatal business errors should become non-retryable ApplicationErrors
  • Transient infrastructure errors should stay retryable
Use format_activity_error(...) and FatalError(...) for repo-standard error handling when appropriate. See src/fleet_manager/temporal/activities/quarantine.py for a concrete example of the expected pattern and imports. Activities must also be idempotent enough for Temporal retries:
  • Prefer naturally idempotent operations
  • Check current external state before one-way changes
  • Use stable identifiers so retries can treat “already applied” as success
  • Split non-idempotent work so retries happen before the irreversible step
If the workflow adds new Ironic, Placement, or Nova API calls, update docs/published/fleet-manager/deployment-guide/new-site-deployment/openstack-application-credentials.mdx.

​
4. Register it with Temporal

Update these files:
  • Export the workflow from src/fleet_manager/temporal/workflows/__init__.py
  • Add the workflow class to workflows=[...] in src/fleet_manager/temporal/worker.py
  • Instantiate the activity class in src/fleet_manager/temporal/worker.py
  • Add each bound activity method to activities=[...]
Do not split to a new worker image or task queue unless there is a real dependency boundary.

​
5. Register it with the workflow agent

Update all three of these:
  • src/fleet_manager/core/workflow/names.py Add a TemporalWorkflowName entry whose value exactly matches the workflow class name.
  • src/fleet_manager/agents/workflow/registry.py Add a WorkflowConfig with the right ExecutionMode.
  • src/fleet_manager/core/eligibility/policies.py Add an eligibility entry. This file is exhaustive.
For batched workflows, do not manually add DeviceProgressHandler; the registry injects it.

​
6. Expose it through the API

If the workflow should be selectable in FM:
  • Add it to Workflow.TemporalWorkflow in src/fleet_manager/api/core/models.py
  • Add a DEFAULT_WORKFLOWS entry in src/fleet_manager/api/core/signals.py
The DEFAULT_WORKFLOWS entry is used by the organisation post-save signal to create the workflow row for organisations created after the code ships. Existing organisations are backfilled by the migration in the next step. If the workflow is internal-only, skip this step and keep __api_exposed__ = False on the workflow. The default workflow entry should define:
  • temporal_workflow_name
  • version
  • parameters_schema
  • exclusive
  • User-facing name
  • User-facing description
Use Params.model_json_schema() here, but not inside migrations.

​
7. Add the migration

For a new public workflow, create one consolidated migration in src/fleet_manager/api/core/migrations/ that:
  • Updates Workflow.temporal_workflow_name choices
  • Updates JobPreset.temporal_workflow_name choices
  • Uses RunPython to update_or_create the actual Workflow row for every existing organisation
Important:
  • Do not create a separate migration per organisation or a second migration just for seeding
  • The migration backfills existing organisations; DEFAULT_WORKFLOWS handles future organisations
  • Freeze the JSON schema inside the migration file
  • Do not call model_json_schema() from the migration
  • Keep preset seeding separate, as described below

​
8. Add presets if the workflow uses them

Public workflows should usually ship with at least one preset. For parameterless workflows, {} is fine. Update:
  • src/fleet_manager/temporal/workflows/presets.yaml
  • src/fleet_manager/temporal/workflows/presets-dev.yaml if needed
  • src/fleet_manager/api/core/management/commands/seed_job_presets.py if the payload shape is new
Then seed the presets:
Notes:
  • presets.yaml is only seed input; the runtime source of truth is the database
  • Presets are not created by the workflow migration path
  • If you change preset payload shape, update seed_job_presets.py in the same PR
  • If old orgs have an outdated stored parameters_schema, add a data migration for that too

​
9. Test it in the same PR

Minimum expected coverage:
  • tests/temporal/workflows/test_<name>.py
  • tests/temporal/activities/test_<name>.py
  • tests/temporal/test_worker.py
  • tests/temporal/test_workflow_enum_sync.py
  • tests/api/core/test_signals.py
  • tests/api/core/workflows/test_workflow_validation.py
Also update:
  • tests/agents/workflow/test_submitter.py if batching or request payload behavior changes
  • tests/api/core/test_seed_job_presets.py if presets changed

​
10. Smoke test through Fleet Manager

Do not trigger the workflow directly from Temporal first. Create a Job through FM so you exercise the real path. Minimum smoke test:
  1. Start the API, workflow agent, Temporal worker, and supporting services.
  2. Seed presets for the target org if needed.
  3. Create a job through POST /api/v1/jobs/.
  4. Confirm the job leaves PENDING.
  5. Inspect execution in Temporal UI.
If the workflow depends on services not covered by the default local stack, use the appropriate shared or specialist dev environment.

​
11. Rollout caution

Temporal replays workflow history. Incompatible workflow-definition changes can break inflight executions. Safe default:
  • If there are no inflight executions of that workflow type, deploy normally.
  • If there may be inflight executions, treat the change as replay-sensitive and plan the rollout deliberately.

​
Common mistakes

  • Forgetting __init__.py or worker registration
  • Updating names.py but not registry.py or eligibility/policies.py
  • Using model_json_schema() in a migration
  • Adding the API model choice but not DEFAULT_WORKFLOWS
  • Adding presets to YAML but never running seed_job_presets
  • Changing preset payload shape without updating seed_job_presets.py
  • Making replay-incompatible workflow changes while executions are inflight
  • Adding OpenStack API calls without updating the application-credentials doc

​
Verification

​
Good references

  • src/fleet_manager/temporal/workflows/quarantine.py
  • src/fleet_manager/temporal/activities/quarantine.py
  • src/fleet_manager/api/core/signals.py
  • src/fleet_manager/api/core/migrations/0049_add_quarantine_release_workflows.py

​
Reference PRs

  • Single-device workflow example: #1170
  • Batched workflow runtime example: #1120
  • API registration and migration example: #1144