Project: build a recoverable evidence-to-draft pipeline

Can a learner close their phone halfway through a task and return to the same saved draft? Build an evidence-to-draft service whose state survives a restart and whose request identity cannot silently change.

Before you begin: Complete orchestration, durable loops, human review, and agent evaluation. You need Python 3 with its standard library.

This project brings together orchestration, durable state, evidence checks, and evaluation. Start with deterministic workers so you can verify the control flow before making generation stochastic.

Define a small contract

A request has a task ID, owner, source version, and desired outcome. A successful result is a draft with the supported date and an explicit note when the time is missing. Publishing is outside the task.

Store attempted work and completed work distinctly. The same task ID must not silently be reused by a different user or for different source data. A database uniqueness constraint is useful here because an in-memory list disappears after a restart.

Run the persistent reference

Save the following as draft_pipeline.py in a temporary project directory and run it twice. It creates a local SQLite file named workshop_tasks.sqlite3. This is a complete deterministic reference program, not an LLM agent or a publishing service.

# Runnable: Python 3, standard library.
import json
import sqlite3

def create_draft(db, task_id, owner, record):
    source = json.dumps(record, sort_keys=True)
    db.execute(
        "CREATE TABLE IF NOT EXISTS tasks "
        "(id TEXT PRIMARY KEY, owner TEXT, source TEXT, draft TEXT)"
    )
    with db:
        existing = db.execute(
            "SELECT owner, source, draft FROM tasks WHERE id = ?",
            (task_id,),
        ).fetchone()
        if existing:
            if existing[:2] != (owner, source):
                raise ValueError("Task ID belongs to another request")
            return existing[2]

        time = record.get("time")
        ending = "Time has not been announced." if time is None else time
        draft = record["name"] + ": " + record["day"] + ". " + ending
        db.execute(
            "INSERT INTO tasks VALUES (?, ?, ?, ?)",
            (task_id, owner, source, draft),
        )
        return draft

if __name__ == "__main__":
    record = {"name": "Pottery", "day": "Saturday", "time": None}
    with sqlite3.connect("workshop_tasks.sqlite3") as db:
        first = create_draft(db, "task-1", "learner-a", record)
        second = create_draft(db, "task-1", "learner-a", record)
        assert first == second
        assert "not been announced" in first
        print(first)

The second call returns the saved draft. Closing and reopening the program preserves it. Parameterized SQL keeps values separate from SQL syntax.

This reference uses one local process and a short transaction. It is not a complete concurrent worker implementation. A production queue needs atomic claiming, leases or another ownership mechanism, bounded retries, and reconciliation of uncertain external operations.

Prove restoration across a real boundary

Calling create_draft twice on one connection tests reuse, but does not by itself demonstrate restoration after the process exits. Run the saved file, close the process, and run it again in the same directory. Inspect the database: there should still be one row for task-1 and the same draft.

Next, keep task-1 but change the owner to learner-b. Predict the result before running it. Restore the owner, change Saturday to Sunday, and try again. A task ID identifies a particular request, not a mutable slot for whatever data arrived most recently.

Add model work at a clear boundary

Replace the deterministic draft builder only after the storage behavior passes. Give a model the permitted record and an output contract: draft text plus source IDs. Validate required fields and unsupported claims before saving a completed result.

Do not keep a database write transaction open during a slow model call. Persist a running state and operation identity, perform the bounded call, then atomically store the validated result. If the worker crashes between stages, a recovery worker inspects that state instead of inventing a new task.

If separate agents retrieve and write, keep their artifacts distinct. The writer receives verified evidence, not an unstructured mixture of sources, guesses, and instructions.

Build the failure exercise

Run four tests: restart after saving; reuse the task ID with a different owner; provide a record with no time; and force a worker failure before a valid draft exists.

Expected outcomes are restoration of the same draft, rejection of the identity mismatch, preservation of the unknown time, and a recoverable failed or pending state. Add a fifth test asserting that no publication tool is called.

On a phone, the interface should show the saved task status after reconnecting, preserve the question, and let the user inspect the draft before a separate publishing decision.

Completion evidence

Submit the program, fixture records, observed test results, and a short explanation of the next production boundary you would implement. Report partial results honestly. Adding several agent roles does not compensate for a task that loses its state after a refresh.

The LangGraph persistence guide provides a framework approach to durable execution. Python's SQLite documentation documents the local storage API used here.

Continue: Prompt injection defense, which protects these boundaries when retrieved evidence contains instructions.