Agentic Engineering Lab

Build it in stages

Run the in-memory transactional seed and explain why repeated job delivery does not add a second effect.
Replace the seed store with a durable local or managed database and add job versions, leases, and fencing checks.
Create authenticated FastAPI submit and status endpoints that persist accepted work before returning a job ID.
Add a real queue and workers with bounded concurrency, idempotent operations, and graceful shutdown.
Package API and workers with Docker, pin reviewed dependency inputs, and add CI tests plus a fixed evaluation gate.
Run duplicate-delivery, worker-stop, provider-throttle, stale-token, and checkpoint-version drills; write observed outcomes and recovery times.

Your acceptance criteria

Use these as your project review. Record commands, outputs, and failure cases in your repository.

A repeated submission key returns the same job identity without additional publication.
Worker interruption after each documented step leaves a recoverable checkpoint or explicit uncertain outcome.
Stale fencing tokens cannot update authoritative job state.
The service exposes queue age, task outcome, cost, and latency without logging secrets.
Deployment and rollback instructions cover checkpoint compatibility and in-flight jobs.

A working starting point

The seed runs as supplied. Extend it to satisfy the full brief. It is a teaching starting point, not a finished portfolio submission.

main.py

python

import sqlite3

def setup():
    database = sqlite3.connect(":memory:")
    database.executescript(
        "CREATE TABLE jobs (id TEXT PRIMARY KEY, status TEXT NOT NULL);"
        "CREATE TABLE effects (job_id TEXT PRIMARY KEY, receipt TEXT NOT NULL);"
    )
    return database

def submit(database, job_id):
    with database:
        database.execute("INSERT OR IGNORE INTO jobs VALUES (?, ?)",
                         (job_id, "queued"))

def process(database, job_id, stop_before_commit=False):
    with database:
        row = database.execute("SELECT status FROM jobs WHERE id = ?",
                               (job_id,)).fetchone()
        if row is None:
            raise ValueError("unknown job")
        if row[0] == "done":
            return "already done"
        receipt = "receipt-" + job_id
        database.execute("INSERT OR IGNORE INTO effects VALUES (?, ?)",
                         (job_id, receipt))
        if stop_before_commit:
            raise RuntimeError("simulated stop before transaction commit")
        database.execute("UPDATE jobs SET status = ? WHERE id = ?",
                         ("done", job_id))
    return receipt

def main():
    database = setup()
    submit(database, "j1")
    submit(database, "j1")
    try:
        process(database, "j1", stop_before_commit=True)
    except RuntimeError:
        print("interrupted transaction rolled back")
    print("retry:", process(database, "j1"))
    print("redelivery:", process(database, "j1"))
    count = database.execute("SELECT COUNT(*) FROM effects").fetchone()[0]
    print("confirmed effects:", count)
    database.close()

if __name__ == "__main__":
    main()

Push it further

Add per-tenant fairness and autoscaling from queue-age metrics, then measure behavior under a mixed workload and a constrained provider quota.