app.orchestrator

Job → Task DAG construction + cache-key computation.

A Job is the user-facing intent (one HTTP call). It owns a DAG of Task`s. Each Task has a `cache_key derived from (kind, inputs_hash, params_hash, runtime_version_id); if a Task with that cache_key has already produced output, the DAG short-circuits to the existing result without enqueuing.

This module builds DAGs deterministically from input specs. It does not talk to the DB or the queue — those are scheduler concerns.

class app.orchestrator.dag.TaskNode(task_id: 'str', kind: 'str', inputs_hash: 'str', params_hash: 'str', depends_on: 'list[str]' = <factory>, gpu_required: 'bool' = True, metadata: 'dict' = <factory>)[source]

Bases: object

Parameters:

  • task_id (str)

  • kind (str)

  • inputs_hash (str)

  • params_hash (str)

  • depends_on (list[str])

  • gpu_required (bool)

  • metadata (dict)

task_id: str
kind: str
inputs_hash: str
params_hash: str
depends_on: list[str]
gpu_required: bool = True
metadata: dict
cache_key(runtime_version_id)[source]
Parameters:

runtime_version_id (str)

Return type:

str

class app.orchestrator.dag.JobDag(nodes: 'list[TaskNode]')[source]

Bases: object

Parameters:

nodes (list[TaskNode])

nodes: list[TaskNode]
topo_order()[source]
Return type:

list[TaskNode]

app.orchestrator.dag.hash_params(spec)[source]
Parameters:

spec (dict)

Return type:

str

app.orchestrator.dag.hash_inputs(refs)[source]
Parameters:

refs (dict)

Return type:

str

Scheduler — submit job DAG and enqueue tasks for execution.

Persists Job + Task rows and hands each non-cached task off to the configured queue (see app.orchestrator.queue.get_queue). Whether the queue is ARQ-backed or in-process is a runtime decision driven by settings.queue_backend / settings.inline_tasks.

async app.orchestrator.scheduler.submit_job_dag(session, *, tenant_id, project_id, recipe, spec, nodes, inline=False)[source]

Persist Job + Task rows and return them. inline=True forces the InlineQueue regardless of settings (used by tests). ARQ enqueue failures (e.g. Redis absent in dev) are suppressed — the tasks remain pending and will be picked up on the next worker boot.

Parameters:

  • session (AsyncSession)

  • tenant_id (str)

  • project_id (str)

  • recipe (str)

  • spec (dict | None)

  • nodes (list[TaskNode])

  • inline (bool)

Return type:

tuple[str, list[Task]]

Lease helpers — works on SQLite and Postgres.

Postgres has SELECT … FOR UPDATE SKIP LOCKED; SQLite does not. We use the portable subset: an UPDATE that increments lease only when the existing lease has expired, then check rowcount. This works on both engines without dialect branches.

app.orchestrator.lease.now_utc()[source]
Return type:

datetime

async app.orchestrator.lease.try_acquire_lease(session, *, table, pk_col, lease_col, worker_col, pk_value, worker_id, ttl_seconds)[source]
Parameters:

  • session (AsyncSession)

  • pk_value (str)

  • worker_id (str)

  • ttl_seconds (int)

Return type:

bool

async app.orchestrator.lease.refresh_lease(session, *, table, pk_col, lease_col, worker_col, pk_value, worker_id, ttl_seconds)[source]
Parameters:

  • session (AsyncSession)

  • pk_value (str)

  • worker_id (str)

  • ttl_seconds (int)

Return type:

bool

Fair-share scheduler.

Picks the next ready Task across all tenants, biased toward tenants with fewer recent admissions. The interleaving rule:

Pick the tenant with the smallest count of currently-running tasks; break ties by the tenant whose last-admitted task is oldest. Within a tenant, pick by Task.created_at (FIFO).

A simpler invariant we expose for testing:

max_consecutive_jobs_per_tenant: never admit more than N consecutive Tasks from the same tenant when other tenants have ready work.

This is intentionally a picker — it does not enqueue. The supervisor calls pick_next_task(…) between subprocess slots; in tests we use the picker directly to validate fairness.

class app.orchestrator.fair_share.FairShareState(max_consecutive_per_tenant: 'int' = 2, recent_tenants: 'deque[str]' = <factory>)[source]

Bases: object

Parameters:

  • max_consecutive_per_tenant (int)

  • recent_tenants (deque[str])

max_consecutive_per_tenant: int = 2
recent_tenants: deque[str]
admit(tenant_id)[source]
Parameters:

tenant_id (str)

Return type:

None

consecutive_for(tenant_id)[source]
Parameters:

tenant_id (str)

Return type:

int

async app.orchestrator.fair_share.pick_next_task(session, *, state)[source]

Find the highest-priority ready Task that doesn’t violate max_consecutive_per_tenant. Ready = pending + deps satisfied.

For Phase 5 v1 we treat ‘ready’ as pending with no remaining deps; dep satisfaction is a precondition the orchestrator already enforces when emitting tasks. Tenants are weighted by current running count.

Parameters:

  • session (AsyncSession)

  • state (FairShareState)

Return type:

Task | None

Resume a previously failed/cancelled Job.

The resume contract:

  • Tasks in (failed, cancelled, cancelled_dirty) are reset to pending. Their cache_key is preserved so the same upstream cache hits replay.

  • Tasks in succeeded are kept as-is (free cache).

  • The Job itself transitions back to pending, cancel flags cleared.

  • If inline=True (or SFMAPI_INLINE_TASKS=true), reset tasks are re-run synchronously here. Otherwise they’re re-enqueued via ARQ.

  • For mapping tasks specifically, the worker reads jobs/{job_id}/checkpoints/ (see app.storage.mapping_input) to resume from the latest MappingInput.save write, avoiding a full restart.

async app.orchestrator.resume.resume_job(session, *, tenant_id, job_id, inline=None)[source]
Parameters:

  • session (AsyncSession)

  • tenant_id (str)

  • job_id (str)

  • inline (bool | None)

Return type:

Job