Build agent job submission and scheduling API

Introduction

Engineers often stand up a shared agent execution cluster and route all job submissions through a single unbounded queue, only to discover that one team's burst workload blocks another team's time-sensitive production agent for several minutes. Without a typed submission layer that validates requests at the boundary, enforces per-team backpressure, and carries priority signals into a fair-share scheduler, the platform offers no mechanism to distinguish a critical inference job from a low-priority batch sweep. By the end of this lesson, you will have built a job submission API that validates agent execution requests, applies backpressure when per-team pending queues run deep, and feeds a priority-weighted fair-share scheduler that maps jobs to sandboxed execution slots according to configurable priority classes.

Key Terminology

• Job Submission — The three-phase boundary enforced by AgentJobSubmissionService.submit_job: validate the incoming AgentJobRequest with Pydantic field validators, enforce per-team backpressure before any I/O, then persist the AgentJobRecord and enqueue it with a numeric priority weight derived from PRIORITY_WEIGHTS.
• Fair-Share Scheduling — A scheduling policy that uses numeric weights from PRIORITY_WEIGHTS to order competing jobs, so a CRITICAL job (weight 0) is always popped before DEFAULT (weight 50) or LOW (weight 100) workloads regardless of which team submitted first or when the job arrived.
• Priority Class — The PriorityClass enum (CRITICAL, HIGH, DEFAULT, LOW) attached to each AgentJobRequest; each value maps to a numeric weight in PRIORITY_WEIGHTS, and the scheduler pops the lowest-weight item first — making weight inversion the mechanism that translates urgency vocabulary into scheduling order.
• Execution Slot — A logical unit of sandboxed capacity, bounded by the ResourceRequirements fields (cpu_millicores, memory_mb, gpu_count, timeout_seconds), that the scheduler assigns to exactly one running agent job at a time.
• Job Lifecycle — The state machine an AgentJobRecord traverses from "PENDING" at creation through scheduling and execution to a terminal state, with timestamps created_at, scheduled_at, started_at, and completed_at marking each transition for scheduling-lag diagnostics.
• Backpressure — The per-team throttle that AgentJobSubmissionService.submit_job enforces as its first action: if a team's pending count reaches MAX_PENDING_PER_TEAM (50), a BackpressureError is raised immediately and the HTTP layer returns HTTP 429 with a Retry-After header, keeping the scheduler's working set bounded.

Concepts

The Cost of an Untyped Submission Boundary

A shared agent execution cluster without a structured submission layer treats every incoming job identically — first-in, first-out, no team attribution, no priority signal. The consequence is queue monopolization: one team's burst of low-priority batch sweeps occupies every pending slot and starves another team's time-sensitive production agent for minutes. The fix is not a faster scheduler; it is a typed boundary that captures all scheduling signals at submission time, before any storage or scheduling logic runs. AgentJobRequest is that boundary — it encodes the submitting team, the requested priority class, and the resource envelope in a validated schema, so every downstream decision can rely on structured, pre-validated data rather than opaque payloads.

Priority Classes and Numeric Weight Inversion

Not all urgency vocabularies map naturally to scheduler mechanics. The lesson bridges the two with a deliberate inversion: the PriorityClass enum gives operators a human-readable vocabulary (CRITICAL, HIGH, DEFAULT, LOW), while PRIORITY_WEIGHTS converts each class into a numeric weight where lower means more urgent — CRITICAL maps to 0, LOW to 100. The scheduler treats the queue as a min-heap: the item with the smallest weight is always popped first. This inversion is the central mental model. CRITICAL does not mean "try harder"; it means "this job carries weight 0 and will sort ahead of every non-zero entry already in the queue." Fair-share scheduling layers a second constraint on top: capacity is partitioned across teams so that a single team's CRITICAL burst cannot exhaust all execution slots, leaving other teams' equally urgent jobs waiting (see Code Walkthrough).

Backpressure as a Fail-Fast Contract

If the submission path accepts jobs without bound during a traffic spike, the pending queue grows until memory is exhausted, persistence slows, or scheduling latency becomes unpredictable. Backpressure prevents this by making the submission service fail fast at the boundary rather than absorbing and degrading. The check runs first in submit_job — before persistence, before idempotency lookup — so that an over-quota team receives a BackpressureError carrying retry_after_seconds without any storage side-effect. The HTTP layer translates this to a 429 response with a Retry-After header, giving well-behaved clients an exact signal to back off. The result is a bounded scheduler working set: regardless of how many jobs a team submits concurrently, no more than MAX_PENDING_PER_TEAM (50) can be in-flight at once.

Job Lifecycle and Idempotent Re-submission

A job does not simply appear in a running pod — it traverses a sequence of states, each timestamped on AgentJobRecord. The gap between created_at and scheduled_at quantifies scheduler backlog; the gap between scheduled_at and started_at quantifies slot-acquisition latency. These timestamps make scheduling lag observable without inspecting queue internals. The lifecycle also enables idempotent re-submission: when a client retries after a transient network failure, supplying the same idempotency_key causes submit_job to detect the already-persisted record and return it unchanged rather than inserting a second entry. Without this guard, retries under failure would spin up duplicate running pods for the same logical job — a correctness hazard the idempotency check eliminates before any enqueue occurs (see Code Walkthrough).

Code Walkthrough

Now that you understand Job Submission, Fair-Share Scheduling, Priority Class, Execution Slot, Job Lifecycle, and Backpressure as defined in the Concepts section, the implementation can make each term operational in code.

The submission path has three sequential responsibilities: validate the incoming request with typed Pydantic models, enforce per-team backpressure before touching durable storage, and enqueue the persisted record with a numeric priority weight derived from the job's PriorityClass. The scheduler loop on the other side of the queue reads those weights so CRITICAL jobs surface before DEFAULT or LOW workloads competing for the same execution slots.

Code snippetpython
1import uuid
2from datetime import datetime, timezone
3from enum import Enum
4from dataclasses import dataclass, field
5from typing import Optional
6from pydantic import BaseModel, Field, validator
7
8class PriorityClass(str, Enum):
9    CRITICAL = "critical"
10    HIGH = "high"
11    DEFAULT = "default"
12    LOW = "low"
13
14PRIORITY_WEIGHTS = {
15    PriorityClass.CRITICAL: 0,
16    PriorityClass.HIGH: 10,
17    PriorityClass.DEFAULT: 50,
18    PriorityClass.LOW: 100,
19}
20
21class ResourceRequirements(BaseModel):
22    cpu_millicores: int = Field(ge=100, le=8000, default=1000)
23    memory_mb: int = Field(ge=128, le=32768, default=2048)
24    gpu_count: int = Field(ge=0, le=4, default=0)
25    timeout_seconds: int = Field(ge=30, le=7200, default=600)
26
27class AgentJobRequest(BaseModel):
28    agent_image: str = Field(..., regex=r"^[\w.\-/]+:[\w.\-]+$")
29    team_id: str = Field(..., min_length=3, max_length=64)
30    priority: PriorityClass = PriorityClass.DEFAULT
31    resources: ResourceRequirements = ResourceRequirements()
32    environment: dict[str, str] = Field(default_factory=dict)
33    input_payload: dict = Field(default_factory=dict)
34    idempotency_key: Optional[str] = None
35
36    @validator("environment")
37    def block_reserved_env_vars(cls, v):
38        reserved = {"K8S_NAMESPACE", "POD_NAME", "NODE_NAME"}
39        conflicts = reserved & set(v.keys())
40        if conflicts:
41            raise ValueError(f"Reserved env vars: {conflicts}")
42        return v
43
44@dataclass
45class AgentJobRecord:
46    job_id: str
47    request: AgentJobRequest
48    status: str = "PENDING"
49    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
50    scheduled_at: Optional[datetime] = None
51    started_at: Optional[datetime] = None
52    completed_at: Optional[datetime] = None
53    k8s_job_name: Optional[str] = None
54
55class BackpressureError(Exception):
56    def __init__(self, team_id: str, pending: int, retry_after_seconds: int):
57        self.team_id = team_id
58        self.pending = pending
59        self.retry_after_seconds = retry_after_seconds
60        super().__init__(
61            f"Team {team_id} has {pending} pending jobs; retry after {retry_after_seconds}s"
62        )
63
64class AgentJobSubmissionService:
65    MAX_PENDING_PER_TEAM = 50
66
67    def __init__(self, job_store, scheduler_queue):
68        self._store = job_store
69        self._queue = scheduler_queue
70
71    async def submit_job(self, request: AgentJobRequest) -> AgentJobRecord:
72        pending_count = await self._store.count_pending(request.team_id)
73        if pending_count >= self.MAX_PENDING_PER_TEAM:
74            raise BackpressureError(
75                team_id=request.team_id,
76                pending=pending_count,
77                retry_after_seconds=30,
78            )
79        if request.idempotency_key:
80            existing = await self._store.find_by_idempotency_key(
81                request.team_id, request.idempotency_key
82            )
83            if existing is not None:
84                return existing
85        job = AgentJobRecord(job_id=str(uuid.uuid4()), request=request)
86        await self._store.persist(job)
87        await self._queue.enqueue(item=job, priority=PRIORITY_WEIGHTS[request.priority])
88        return job

AgentJobRequest uses Pydantic field validators to reject malformed image references and reserved environment variable names before any I/O occurs. submit_job enforces the backpressure threshold as its first check — if the team's pending count reaches MAX_PENDING_PER_TEAM, the call raises BackpressureError immediately, which the HTTP layer translates to a 429 with a Retry-After header. Idempotency key de-duplication runs second so that a client retrying after a transient error does not double-enqueue the same job. Once those guards pass, the record is persisted and enqueued with its numeric priority weight; the scheduler loop later pops the lowest-weight job first, giving CRITICAL workloads scheduling priority over jobs queued at DEFAULT or LOW.

Confirm that submitting a job with priority=PriorityClass.CRITICAL returns an AgentJobRecord in PENDING status with a non-null job_id, and that a second call carrying the same idempotency_key returns the original record unchanged rather than inserting a second entry.

Do's and Don'ts

Do's

1. ✓Do enforce idempotency keys on all submission endpoints — Agent orchestrators commonly retry on timeout, and without idempotency protection, each retry creates a duplicate job that wastes execution slots and produces confusing duplicate results.
2. ✓Do implement backpressure at the submission layer — Returning HTTP 429 with Retry-After headers is far more effective than letting the queue grow unbounded, because clients can implement exponential backoff while the queue drains.
3. ✓Do separate scheduling policy from pod lifecycle — The Kubernetes Job controller manages container execution; your scheduler manages fairness, priority, and team quotas. Mixing these concerns makes both harder to reason about and test.

Don'ts

1. ✗Don't use Kubernetes pod priority for agent scheduling — K8s PriorityClass controls preemption at the kubelet level, not fair-share across teams. Conflating the two causes unpredictable evictions of lower-priority system pods.
2. ✗Don't allow unbounded queue depth per team — Without the MAX_PENDING_PER_TEAM check, a single misconfigured agent orchestrator can flood the queue with thousands of jobs, starving every other team even with fair-share scheduling.
3. ✗Don't retry failed K8s job creation silently by re-enqueuing — If a job fails to schedule, marking it as "SCHEDULE_FAILED" and surfacing it through monitoring is safer than re-enqueuing it, which can create infinite scheduling loops if the failure is deterministic (e.g., invalid image reference).