The capstone assembles the pieces already studied into a service that can run, stop, preserve evidence, and explain its outcome. Its scope is a local simulated Bell experiment. Keeping that scope small makes it possible to inspect the complete path from configuration to measurement counts and final disposition.
Use the local Python project supplied with the lab, in its specified Python 3.12 environment. The dependency lock and project metadata identify the packages used by the reference implementation. The browser lab is an alternative execution environment for the same declared model; it does not start or host the Python service.
Run one complete request
From the installed project environment:
quantum-ops --db capstone.sqlite3 run --scenario baselineThe global --db option precedes the subcommand. This creates a local SQLite job record and returns JSON evidence. Copy the returned run identifier; the fixture labels shown in the book are stable publication labels, not identifiers automatically present in your database.
Replace RUN_ID with that actual identifier:
quantum-ops --db capstone.sqlite3 inspect RUN_ID
quantum-ops --db capstone.sqlite3 validate RUN_IDThe first command retrieves evidence. The second recomputes validation and exits unsuccessfully if the record fails its checks. Successful completion of a command is a software event; the scientific result still comes from the validated disposition.
Follow the implementation below. The runner creates the record before work starts, claims execution ownership, samples incremental batches, invokes the statistical analyzer, and saves complete checkpoints. Sampling belongs to the model module, and evidence checks belong to the contract module. This keeps the orchestration path readable.
"""One local execution path shared by the CLI and loopback HTTP service."""
import copy
import json
import os
import sqlite3
import uuid
from pathlib import Path
from .contract import (BASES, CHECKPOINTS, SCENARIOS, configuration, validate_config, artifact_hashes,
aggregate, analyze, finalize, findings)
from .model import engine_versions, sample
class ClosingConnection(sqlite3.Connection):
def __exit__(self,*args):
try:return super().__exit__(*args)
finally:self.close()
class Runner:
def __init__(self, database="quantum-runs.sqlite3"):
self.database=str(database)
with self.connect() as db:
db.execute("CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, status TEXT NOT NULL, owner_pid INTEGER, cancel_requested INTEGER NOT NULL DEFAULT 0, evidence TEXT NOT NULL, workflow_status TEXT NOT NULL DEFAULT 'single')")
if 'workflow_status' not in [r[1] for r in db.execute("PRAGMA table_info(jobs)")]: db.execute("ALTER TABLE jobs ADD COLUMN workflow_status TEXT NOT NULL DEFAULT 'single'")
db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_recovery ON jobs(json_extract(evidence, '$.recoveryOf'))")
def connect(self):
db=sqlite3.connect(self.database,timeout=10,factory=ClosingConnection)
db.execute("PRAGMA journal_mode=WAL")
return db
def submit(self, config=None, recovery_of=None):
config=copy.deepcopy(config or configuration());validate_config(config)
job_id=str(uuid.uuid4())
e={"schemaVersion":"1.0.0","id":job_id,"simulated":True,"config":config,"engine":engine_versions(),"artifactHashes":artifact_hashes(),"executionStatus":"queued","disposition":"inconclusive","batches":[],"history":[],"uncertaintyMethod":"hoeffding-3-bases-7-looks-alpha-0.05-v1","recoveryOf":recovery_of}
finalize(e)
with self.connect() as db: db.execute("INSERT INTO jobs (id,status,evidence,workflow_status) VALUES (?,?,?,?)",(job_id,"queued",json.dumps(e),"pending" if config["id"]=="drift" else "single"))
return job_id
def inspect(self, job_id):
with self.connect() as db: row=db.execute("SELECT evidence FROM jobs WHERE id=?",(job_id,)).fetchone()
if not row: raise KeyError("Unknown job")
return json.loads(row[0])
def save(self,e):
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
row=db.execute("SELECT status,owner_pid,cancel_requested FROM jobs WHERE id=?",(e["id"],)).fetchone()
if not row or row[0]!="running" or row[1]!=os.getpid(): raise ValueError("Execution ownership changed")
if row[2] and e["executionStatus"]!="failed":
e["executionStatus"]="cancelled";e["disposition"]="inconclusive"
finalize(e)
db.execute("UPDATE jobs SET status=?,evidence=? WHERE id=? AND status='running' AND owner_pid=?",(e["executionStatus"],json.dumps(e),e["id"],os.getpid()))
def cancel(self,job_id):
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
row=db.execute("SELECT status,evidence FROM jobs WHERE id=?",(job_id,)).fetchone()
if not row: raise KeyError("Unknown job")
if row[0] in ("completed","cancelled","failed"): return self.inspect(job_id)
db.execute("UPDATE jobs SET cancel_requested=1 WHERE id=?",(job_id,))
if row[0] in ("queued","interrupted"):
e=json.loads(row[1]);e["executionStatus"]="cancelled";e["disposition"]="inconclusive";finalize(e)
db.execute("UPDATE jobs SET status='cancelled',evidence=? WHERE id=?",(json.dumps(e),job_id))
return self.inspect(job_id)
def execute(self,job_id,*,resume=False,interrupt_after=None):
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
row=db.execute("SELECT status,owner_pid,evidence FROM jobs WHERE id=?",(job_id,)).fetchone()
if not row: raise KeyError("Unknown job")
status,pid,raw=row;e=json.loads(raw)
if status=="running":
alive=False
if pid:
try: os.kill(pid,0);alive=True
except ProcessLookupError: pass
if alive: raise ValueError("Job already has a live execution owner")
if not resume: raise ValueError("Interrupted owner; use resume")
elif status not in (("queued","interrupted") if resume else ("queued",)):
raise ValueError(f"Cannot execute job in {status} state")
if findings(e) or e["disposition"]=="invalid": raise ValueError("Cannot resume invalid evidence")
if e["engine"] != engine_versions(): raise ValueError("Engine versions changed; start a separate run")
e["executionStatus"]="running";finalize(e)
db.execute("UPDATE jobs SET status='running',owner_pid=?,evidence=? WHERE id=?",(os.getpid(),json.dumps(e),job_id))
try:
for i,n in enumerate(CHECKPOINTS):
if i<len(e["batches"]):continue
if n>e["config"]["budget"]:break
with self.connect() as db: cancel=db.execute("SELECT cancel_requested FROM jobs WHERE id=?",(job_id,)).fetchone()[0]
if cancel: e["executionStatus"]="cancelled";break
previous=e["batches"][-1]["checkpoint"] if e["batches"] else 0
seeds={b:(e["config"]["seed"]+i*3+j)%2**31 for j,b in enumerate(BASES)}
counts={b:sample(b,n-previous,e["config"]["p"],e["config"]["q"],seeds[b]) for b in BASES}
e["batches"].append({"checkpoint":n,"context":e["config"]["context"],"counts":counts,"seeds":seeds})
report=analyze(aggregate(e["batches"]),e["config"]["budget"]);e["history"].append(report)
with self.connect() as db: cancel=db.execute("SELECT cancel_requested FROM jobs WHERE id=?",(job_id,)).fetchone()[0]
if cancel: e["executionStatus"]="cancelled";e["disposition"]="inconclusive"
elif report["decision"]!="continue": e["executionStatus"]="completed";e["disposition"]=report["decision"]
elif interrupt_after and len(e["batches"])>=interrupt_after: e["executionStatus"]="interrupted"
if e["config"]["id"]=="invalid" and e["executionStatus"]=="completed": e["engine"].pop("model",None)
self.save(e)
if e["executionStatus"]!="running":return e
except (KeyboardInterrupt, SystemExit):
e["executionStatus"]="interrupted";e["disposition"]="inconclusive";self.save(e);raise
except Exception:
e["executionStatus"]="failed";e["disposition"]="invalid";self.save(e);raise
self.save(e)
return e
def workflow_status(self,job):
with self.connect() as db:row=db.execute("SELECT workflow_status,status FROM jobs WHERE id=?",(job,)).fetchone()
if not row:raise KeyError("Unknown job")
return row[1] if row[0]=='single' or row[1] in ('cancelled','failed') else row[0]
def set_workflow(self,job,status):
with self.connect() as db:db.execute("UPDATE jobs SET workflow_status=? WHERE id=?",(status,job))
def execute_scenario(self,job):
config=self.inspect(job)["config"]
scenario=next(s for s in SCENARIOS["scenarios"] if s["id"]==config["id"])
try:
evidence=self.execute(job,interrupt_after=scenario.get("interruptAfter"));runs=[evidence]
if scenario.get("recovery"):
if evidence["executionStatus"]=="completed":
self.set_workflow(job,"running")
recovered=configuration("baseline",seed=(config["seed"]+1000003)%2**31,budget=config["budget"],context="recovery-after-drift")
runs.append(self.execute(self.submit(recovered,recovery_of=job)))
self.set_workflow(job,"cancelled" if any(e["executionStatus"]=="cancelled" for e in runs) else "completed")
return runs
except BaseException:
if scenario.get("recovery"):self.set_workflow(job,"failed")
raise
def related_runs(self,job):
with self.connect() as db:
return [row[0] for row in db.execute("SELECT id FROM jobs WHERE json_extract(evidence,'$.recoveryOf')=?",(job,))]
def run_scenario(self,name,**overrides):
return self.execute_scenario(self.submit(configuration(name,**overrides)))
Add a local service boundary
The optional HTTP service uses the same runner:
quantum-ops --db capstone.sqlite3 serve --port 8765It listens on 127.0.0.1. A local client can submit with POST /runs, inspect with GET /runs/RUN_ID, retrieve the full record with GET /runs/RUN_ID/evidence, and request cancellation with POST /runs/RUN_ID/cancel. Submission uses a JSON configuration request, including the supported scenario selector. The handler validates it before admitting work.
A successful submission returns HTTP 202 with the identifier. It is not a completed quality result. The client subsequently inspects execution status, disposition, and validation. The service has two active worker slots and returns 429 when another task cannot be admitted. Resume is available through the local command line; this API does not advertise a resume route.
The public reader website does not proxy these requests. Its browser simulation and your loopback Python service have separate execution locations, even though their evidence follows the shared contract.
Worked example: recover the saved prefix
Run the interrupted scenario in the same database:
quantum-ops --db capstone.sqlite3 run --scenario interruptedInspect the returned record before resuming it. It should contain the first two completed checkpoints, representing 200 shots in each of three bases. Its execution status is interrupted and its disposition is inconclusive.
Resume using that record's identifier:
quantum-ops --db capstone.sqlite3 resume RUN_IDThe existing batches remain the prefix of the final record. The runner adds only the work needed for later checkpoints. This tests a concrete recovery property: continuation is based on persisted execution evidence, rather than recreating a new request and hoping its output looks similar.
Changing the implementation or declared operating context requires a separate compatibility decision. Inspectable evidence from another engine is not permission to continue it while retaining the original engine label.
Exercise and worked answer
A client receives HTTP 202, loses its connection, and cannot yet tell whether the job completed. It proposes issuing a second submission and combining whichever counts arrive first. What should it do?
Worked answer: Retain the returned identifier and inspect that job. A second submission creates another request; this service does not supply an idempotency key that would automatically merge the two. Combining their counts would also need a new statistical and provenance contract. If the original request's identifier was never received, the uncertainty is an application-level recovery problem, not proof that no work occurred.
The completed capstone should let you explain this distinction, locate the original evidence, and avoid an invented exactly-once guarantee. The next chapter, Validate and Operate the Service, tests the assembled service against the full set of intended outcomes.