Idempotent AI Jobs: Reliable Python Schedules

A scheduled AI job can run more than once. Depending on the scheduler and its retry policy, a worker may be retried after exceeding an execution deadline. Overlapping submissions are also possible. If a database connection drops around a commit, the caller may be unable to determine whether the transaction succeeded. If every attempt generates a new identifier or blindly inserts output, you get duplicate reports, repeated notifications, and inconsistent state.

The fix is architectural. Give each logical run a stable ID. Separate collection, model work, and persistence. Then use database constraints and atomic writes to make retries safe.

Define idempotency for scheduled AI jobs (and its limits)

GitLab defines an idempotent worker as one that can run multiple times with the same arguments while producing side effects only once, or making subsequent executions have no additional effect. Its documentation also recommends that workers be both idempotent and transactional.

The database and execution patterns below are application-architecture recommendations derived from that principle. They are not GitLab requirements and apply independently of GitLab or Sidekiq.

For a scheduled AI job, focus on final state rather than identical model output.

Suppose your agent creates a daily competitor summary. The same logical run might execute again because:

• The worker exceeded a configured deadline and the scheduler was configured to retry.
• The scheduler submitted overlapping jobs.
• A deployment restarted the worker.
• A connection dropped around the final commit, leaving its outcome uncertain to the caller.
• An operator manually retried a failed run.

The model may produce different text on each call. You cannot assume that repeating inference returns the same bytes. Instead, define the invariant at the application boundary:

For the same logical run, the system stores one accepted result and performs each external side effect once.

Queue deduplication helps, but it is not sufficient. A queue can prevent duplicate messages while its deduplication key exists. It cannot roll back an email, API call, object upload, or database write that already happened.

Your application therefore needs its own durable identity and uniqueness rules. These rules apply whether you use cron, a managed scheduler, or a persistent worker. If you are still choosing a runtime, start with the tradeoffs between cron and daemon agents.

Create one stable run ID for each logical schedule

Do not create a random UUID inside the worker. A retry would receive a new UUID and look like unrelated work.

Derive the run ID from fields that define the logical execution:

• Job name
• Tenant, account, or subject
• Scheduled time window
• Job version

Normalize these fields before hashing them. In particular, reject naïve datetimes, convert timezone-aware timestamps to UTC, and use the start of the intended schedule window.

from datetime import datetime, timezone
from hashlib import sha256


def stable_run_id(
    job_name: str,
    subject_id: str,
    scheduled_at: datetime,
    version: str,
) -> str:
    if scheduled_at.tzinfo is None:
        raise ValueError("scheduled_at must be timezone-aware")

    scheduled_utc = scheduled_at.astimezone(timezone.utc)
    canonical_time = scheduled_utc.isoformat(timespec="seconds")

    identity = "|".join(
        [
            job_name.strip(),
            subject_id.strip(),
            canonical_time,
            version.strip(),
        ]
    )
    return sha256(identity.encode("utf-8")).hexdigest()

Call it with the timestamp supplied by the scheduler, not the worker’s current time:

scheduled_at = datetime.fromisoformat("2026-08-07T09:00:00+00:00")

run_id = stable_run_id(
    job_name="competitor-summary",
    subject_id="workspace-acme",
    scheduled_at=scheduled_at,
    version="v1",
)

print(run_id)

Every retry must receive or reconstruct the same inputs. Two workers handling the same scheduled occurrence will then address the same database row.

Version belongs in the identity because changing transformation semantics may justify a new result. Make that decision explicit. Do not silently generate a new identity whenever code changes. Otherwise, a deployment can bypass your duplicate protection.

Store the run ID before expensive model work. It should be part of your durable execution record, not just a log field. The persistent state pattern for serverless jobs shows how job IDs, progress, retries, and results fit into the same schema.

Separate collection, AI transformation, and persistence

Keep the job in three stages:

1. Collect source data and store the raw artifact.
2. Transform the stored input with the model.
3. Validate and persist the accepted output.

This separation limits ambiguous side effects. If collection succeeds but inference times out, the retry can reuse the collected artifact. If persistence succeeds but the worker loses its response, the retry can read the completed record rather than call the model again.

from dataclasses import dataclass
import json


@dataclass(frozen=True)
class CandidateResult:
    text: str
    model_name: str


def collect(subject_id: str) -> dict:
    # Replace with responsible API or source access.
    return {
        "subject_id": subject_id,
        "documents": [
            {"title": "Release notes", "body": "A documented product update."}
        ],
    }


def transform(raw_input: dict, model_client) -> CandidateResult:
    prompt = (
        "Summarize the supplied documents. Return concise factual text.\n"
        + json.dumps(raw_input, sort_keys=True)
    )
    response = model_client.generate(prompt)

    return CandidateResult(
        text=response.text.strip(),
        model_name=response.model,
    )


def validate(candidate: CandidateResult) -> None:
    if not candidate.text:
        raise ValueError("Model returned an empty result")

Immediately after collect() returns, persist the raw artifact with a guarded update:

UPDATE runs
SET raw_input = COALESCE(raw_input, %(raw_input)s::jsonb),
    updated_at = CURRENT_TIMESTAMP
WHERE run_id = %(run_id)s
  AND status <> 'completed'
RETURNING raw_input;

Use the raw_input returned by this statement for transformation rather than assuming that the current worker’s local value won a concurrent update. If no row is returned because the run is already complete, fetch and return the completed result. The run row itself must already have been registered before this update, as shown in the next section.

None of these stages should mark the final run as completed until validation and final persistence succeed. Collection stores only raw input. Transformation returns a candidate. The final persistence stage owns the transition to completed.

That distinction matters for non-deterministic output. If a run is already completed, a retry should return the stored result. It should not replace it with a new model response unless your product has an explicit revision policy.

Long-running jobs benefit from the same boundary design. You can persist progress between stages and resume within the next execution budget. See the guide to resumable, time-bounded AI agent stages.

Use an upsert and uniqueness constraint as the final guard

Use a database constraint as the permanent authority. Application-level checks such as “select, then insert” can race when workers run concurrently.

A PostgreSQL schema can make run_id unique while retaining the source identity:

CREATE TABLE runs (
    run_id TEXT PRIMARY KEY,
    job_name TEXT NOT NULL,
    subject_id TEXT NOT NULL,
    scheduled_at TIMESTAMPTZ NOT NULL,
    version TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('pending', 'running', 'completed', 'failed')
    ),
    raw_input JSONB,
    output JSONB,
    lock_owner TEXT,
    lock_expires_at TIMESTAMPTZ,
    last_error TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE (job_name, subject_id, scheduled_at, version)
);

Register the run before collection:

INSERT INTO runs (
    run_id, job_name, subject_id, scheduled_at, version, status
)
VALUES (
    %(run_id)s, %(job_name)s, %(subject_id)s,
    %(scheduled_at)s, %(version)s, 'pending'
)
ON CONFLICT (run_id) DO UPDATE
SET updated_at = CURRENT_TIMESTAMP
RETURNING status, raw_input, output;

The conflict path must not clear output or move a completed run back to pending.

Commit the accepted candidate with a guarded update:

UPDATE runs
SET status = 'completed',
    output = %(output)s::jsonb,
    lock_owner = NULL,
    lock_expires_at = NULL,
    last_error = NULL,
    updated_at = CURRENT_TIMESTAMP
WHERE run_id = %(run_id)s
  AND status <> 'completed'
  AND lock_owner = %(worker_id)s
RETURNING output;

If this statement returns no row, fetch the current record. Another attempt may have completed it first, or the worker may have lost ownership of the lease. A worker that no longer owns the lease must not commit its candidate.

Apply the same pattern to final side effects. For example, use an outbox table with a unique key:

CREATE TABLE notification_outbox (
    run_id TEXT NOT NULL REFERENCES runs(run_id),
    channel TEXT NOT NULL,
    payload JSONB NOT NULL,
    delivered_at TIMESTAMPTZ,
    PRIMARY KEY (run_id, channel)
);

Inserting (run_id, channel) again becomes a harmless conflict instead of creating a second outbox entry.

GitLab’s test pattern runs a job’s perform method twice and verifies that the final model state remains unchanged. Your Python tests should enforce the same behavior at your persistence boundary.

Prevent concurrent retries and handle deduplication TTLs

A uniqueness constraint prevents duplicate rows, but it does not stop two workers from calling the model simultaneously. The database lease below is an application-level architecture recommendation: claim the run atomically before doing expensive work.

UPDATE runs
SET status = 'running',
    lock_owner = %(worker_id)s,
    lock_expires_at = (
        CURRENT_TIMESTAMP
        + (%(lease_seconds)s * INTERVAL '1 second')
    ),
    updated_at = CURRENT_TIMESTAMP
WHERE run_id = %(run_id)s
  AND status <> 'completed'
  AND (
      lock_owner IS NULL
      OR lock_expires_at < CURRENT_TIMESTAMP
      OR lock_owner = %(worker_id)s
  )
RETURNING run_id, status, raw_input;

If no row is returned, another worker owns the lease or the run is complete. Exit without performing inference.

The database clock should be authoritative for both lease acquisition and expiration; workers should provide a duration rather than an absolute expiration calculated from their local clocks. Every renewal, release, failure update, and final write should also verify lock_owner = %(worker_id)s. A worker must never clear or overwrite another worker’s active lease without an atomic ownership-and-expiry check.

A lease needs explicit ownership and expiry. Clear it after completion. On failure, record the error and release it only if the failing worker still owns it and retry is safe. Under this design, if a process stops renewing its lease, another worker may claim the run after the database clock considers the lease expired. Expiry does not prove that the original process has stopped, so ownership checks on later writes are still required to prevent a stale worker from committing over the new owner.

Queue-level locks remain useful as an optimization. GitLab documents two deduplication strategies in its idempotent jobs documentation:

• until_executing removes the lock before execution starts.
• until_executed retains the lock through job completion and can prevent simultaneous execution.

According to GitLab’s documented semantics, the second strategy offers stronger overlap protection within that implementation, but neither strategy replaces application-level database uniqueness.

Scheduled messages need special attention. GitLab does not deduplicate future scheduled jobs by default; its including_scheduled: true option enables that behavior. The documented default lifetime of 10 minutes applies specifically to GitLab’s Redis deduplication key. It is not a general TTL for every queue or scheduler. Once that GitLab key expires, duplicate jobs can be created, and GitLab recommends shortening its lifetime only when a job can tolerate duplication.

Treat queue TTLs as temporary admission control whose exact behavior depends on the queue implementation and configuration. Keep your application-level run ID and database constraint durable.

Test the same AI run twice before shipping

Test idempotency by executing the same logical run twice. Mock collection and inference so the second model response is visibly different. The stored completed output must remain unchanged.

def test_same_run_is_idempotent(repository, fake_model):
    scheduled_at = datetime.fromisoformat("2026-08-07T09:00:00+00:00")
    run_id = stable_run_id(
        "competitor-summary",
        "workspace-acme",
        scheduled_at,
        "v1",
    )

    fake_model.responses = ["first accepted summary", "different retry output"]

    run_job(
        repository=repository,
        model_client=fake_model,
        run_id=run_id,
        subject_id="workspace-acme",
        scheduled_at=scheduled_at,
    )
    run_job(
        repository=repository,
        model_client=fake_model,
        run_id=run_id,
        subject_id="workspace-acme",
        scheduled_at=scheduled_at,
    )

    rows = repository.find_runs(run_id)
    assert len(rows) == 1
    assert rows[0].status == "completed"
    assert rows[0].output["text"] == "first accepted summary"
    assert repository.count_outbox_entries(run_id) == 1

Also cover the failure paths your scheduler will trigger:

• Two workers try to claim the same run concurrently.
• A timeout occurs after collection.
• A timeout occurs after the final database commit.
• A worker dies while holding a lease.
• An expired lease is reclaimed.
• A stale worker tries to write after another worker acquires the lease.
• A failed run retries without duplicating downstream writes.
• A completed run is invoked with the same arguments.
• Queue deduplication expires while the durable run record remains.

Run these tests against the database semantics you use in production. A mocked repository will not expose transaction or uniqueness races. Combine them with the broader production-readiness testing workflow for Python AI agents.

Conclusion

Reliable idempotent AI jobs do not depend on deterministic model output. They depend on deterministic run identity and controlled side effects.

Build the run ID from the logical schedule. Persist it before inference. Separate collection, transformation, and writing. Protect final state with an upsert and uniqueness constraint. Add an atomic, ownership-checked lease for concurrent workers. Then execute the same run repeatedly in tests and verify that the accepted result does not change.

The durable contract is simple: one logical schedule maps to one run record, one accepted result, and guarded side effects. Keeping that contract in the schema and code makes retries routine instead of risky.