> ## Documentation Index
> Fetch the complete documentation index at: https://docs.internal.nscale.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Add a New Workflow

> Short guide for adding and registering a new Temporal workflow in Fleet Manager.

Adding a public workflow in Fleet Manager usually touches three layers:

* The [Temporal](../../start-here/glossary#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`.

<Warning>
  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.
</Warning>

## 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](../../start-here/glossary#node).
* 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 `ApplicationError`s
* 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](../../start-here/glossary#ironic), [Placement](../../start-here/glossary#placement), or [Nova](../../start-here/glossary#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:

```bash theme={null}
uv run fleetctl infra api manage seed_job_presets --org-slug <org-slug>
```

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

```bash theme={null}
uv run pytest \
  tests/temporal/workflows/test_<workflow>.py \
  tests/temporal/activities/test_<workflow>.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

uv run fleetctl infra api manage seed_job_presets --org-slug <org-slug>
uv run pre-commit run openapi-generate --all-files --hook-stage manual
make pre-commit
```

## 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](https://github.com/nscaledev/fleet-manager/pull/1170)
* Batched workflow runtime example: [#1120](https://github.com/nscaledev/fleet-manager/pull/1120)
* API registration and migration example: [#1144](https://github.com/nscaledev/fleet-manager/pull/1144)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.