#!/usr/bin/env python3 """Synthetic, credential-free workflow reference. No model calls or external effects. Run: python3 fde-agent-engineering-lab.py test python3 fde-agent-engineering-lab.py demo --db field-desk.sqlite python3 fde-agent-engineering-lab.py inspect --db field-desk.sqlite python3 fde-agent-engineering-lab.py review --db field-desk.sqlite --decision accepted Python 3.11+. SQLite reference only; not a distributed worker implementation. """ import argparse import hashlib import json import sqlite3 import tempfile import unittest from pathlib import Path class Workflow: def __init__(self, database): self.db = sqlite3.connect(database) self.db.row_factory = sqlite3.Row self.db.executescript(""" CREATE TABLE IF NOT EXISTS sources ( tenant TEXT NOT NULL, id TEXT NOT NULL, version INTEGER NOT NULL, body TEXT NOT NULL, active INTEGER NOT NULL DEFAULT 1, PRIMARY KEY (tenant, id)); CREATE TABLE IF NOT EXISTS runs ( tenant TEXT NOT NULL, request TEXT NOT NULL, source TEXT NOT NULL, version INTEGER NOT NULL, draft TEXT NOT NULL, digest TEXT NOT NULL, status TEXT NOT NULL, PRIMARY KEY (tenant, request)); """) with self.db: self.db.execute("INSERT OR IGNORE INTO sources VALUES (?, ?, ?, ?, ?)", ("north", "P1", 1, "Inspect the disconnected training unit before replacement.", 1)) self.db.execute("INSERT OR IGNORE INTO sources VALUES (?, ?, ?, ?, ?)", ("south", "P9", 1, "Synthetic other-tenant procedure.", 1)) def close(self): self.db.close() @staticmethod def digest(source, version, draft): return hashlib.sha256(json.dumps([source, version, draft], separators=(",", ":")).encode()).hexdigest() def start(self, tenant, request, source="P1"): # tenant is trusted caller context supplied by the host in a real service. # This CLI uses synthetic values; it is not an authentication service. if not tenant or not request or len(request) > 100: raise ValueError("invalid_identity") with self.db: prior = self.db.execute("SELECT * FROM runs WHERE tenant=? AND request=?", (tenant, request)).fetchone() if prior: if prior["source"] != source: raise ValueError("request_conflict") return dict(prior) row = self.db.execute("SELECT * FROM sources WHERE tenant=? AND id=? AND active=1", (tenant, source)).fetchone() if not row: raise PermissionError("source_unavailable") # Scripted proposal, not a model output or a sent message. draft = "Operator draft: " + row["body"] digest = self.digest(source, row["version"], draft) self.db.execute("INSERT INTO runs VALUES (?, ?, ?, ?, ?, ?, ?)", (tenant, request, source, row["version"], draft, digest, "awaiting_review")) return self.inspect(tenant, request) def inspect(self, tenant, request): row = self.db.execute("SELECT * FROM runs WHERE tenant=? AND request=?", (tenant, request)).fetchone() if not row: raise PermissionError("run_unavailable") return dict(row) def review(self, tenant, request, expected_digest, decision): if decision not in ("accepted", "rejected"): raise ValueError("invalid_decision") # The host must authorize the reviewer before calling this method. # BEGIN IMMEDIATE serializes this local read/check/update transaction. try: self.db.execute("BEGIN IMMEDIATE") run = self.inspect(tenant, request) if run["digest"] != expected_digest: raise ValueError("proposal_changed") if run["status"] in ("accepted", "rejected"): if run["status"] != decision: raise ValueError("decision_conflict") self.db.commit() return run if run["status"] != "awaiting_review": raise ValueError("review_not_pending") current = self.db.execute("SELECT version FROM sources WHERE tenant=? AND id=? AND active=1", (tenant, run["source"])).fetchone() if not current or current["version"] != run["version"]: self.db.execute("UPDATE runs SET status='stale' WHERE tenant=? AND request=?", (tenant, request)) self.db.commit() raise ValueError("source_changed") self.db.execute("UPDATE runs SET status=? WHERE tenant=? AND request=? AND status='awaiting_review'", (decision, tenant, request)) self.db.commit() return self.inspect(tenant, request) except Exception: self.db.rollback() raise class WorkflowTests(unittest.TestCase): def setUp(self): self.folder = tempfile.TemporaryDirectory() self.path = str(Path(self.folder.name) / "test.sqlite") self.flow = Workflow(self.path) def tearDown(self): self.flow.close() self.folder.cleanup() def test_useful_authorized_draft(self): run = self.flow.start("north", "R1") self.assertEqual(run["status"], "awaiting_review") self.assertIn("Inspect", run["draft"]) def test_foreign_source_denied_before_draft(self): with self.assertRaisesRegex(PermissionError, "source_unavailable"): self.flow.start("north", "R1", "P9") self.assertEqual(self.flow.db.execute("SELECT count(*) FROM runs").fetchone()[0], 0) def test_foreign_run_unavailable(self): self.flow.start("north", "R1") with self.assertRaisesRegex(PermissionError, "run_unavailable"): self.flow.inspect("south", "R1") def test_restart_and_idempotent_request(self): original = self.flow.start("north", "R1") self.flow.close() self.flow = Workflow(self.path) self.assertEqual(original, self.flow.start("north", "R1")) self.assertEqual(self.flow.db.execute("SELECT count(*) FROM runs").fetchone()[0], 1) def test_request_reuse_cannot_change_source(self): self.flow.start("north", "R1") with self.assertRaisesRegex(ValueError, "request_conflict"): self.flow.start("north", "R1", "P9") def test_review_is_replayable_but_not_changeable(self): run = self.flow.start("north", "R1") accepted = self.flow.review("north", "R1", run["digest"], "accepted") self.assertEqual(accepted, self.flow.review("north", "R1", run["digest"], "accepted")) with self.assertRaisesRegex(ValueError, "decision_conflict"): self.flow.review("north", "R1", run["digest"], "rejected") def test_digest_binds_exact_proposal(self): self.flow.start("north", "R1") with self.assertRaisesRegex(ValueError, "proposal_changed"): self.flow.review("north", "R1", "not-the-proposal", "accepted") self.assertEqual(self.flow.inspect("north", "R1")["status"], "awaiting_review") def test_changed_source_invalidates_pending_review(self): run = self.flow.start("north", "R1") with self.flow.db: self.flow.db.execute("UPDATE sources SET version=2 WHERE tenant='north' AND id='P1'") with self.assertRaisesRegex(ValueError, "source_changed"): self.flow.review("north", "R1", run["digest"], "accepted") self.assertEqual(self.flow.inspect("north", "R1")["status"], "stale") def test_withdrawn_source_invalidates_pending_review(self): run = self.flow.start("north", "R1") with self.flow.db: self.flow.db.execute("UPDATE sources SET active=0 WHERE tenant='north' AND id='P1'") with self.assertRaisesRegex(ValueError, "source_changed"): self.flow.review("north", "R1", run["digest"], "accepted") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=["test", "demo", "inspect", "review"]) parser.add_argument("--db", default="field-desk.sqlite") parser.add_argument("--tenant", default="north") parser.add_argument("--request", default="R1") parser.add_argument("--decision", choices=["accepted", "rejected"], default="accepted") args = parser.parse_args() if args.command == "test": suite = unittest.defaultTestLoader.loadTestsFromTestCase(WorkflowTests) return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1 flow = Workflow(args.db) try: if args.command == "demo": result = flow.start(args.tenant, args.request) elif args.command == "inspect": result = flow.inspect(args.tenant, args.request) else: shown = flow.inspect(args.tenant, args.request) print("Synthetic proposal only. This records a local draft decision; nothing is sent.") print(shown["draft"]) result = flow.review(args.tenant, args.request, shown["digest"], args.decision) print(json.dumps(result, indent=2)) return 0 except (PermissionError, ValueError) as error: print(json.dumps({"status": "refused", "reason": str(error)})) return 1 finally: flow.close() if __name__ == "__main__": raise SystemExit(main())