"""Local agent-engineering lab. Python 3.10+, standard library, no model/network calls. Run: python3 agent_lab.py The tests demonstrate boundaries with synthetic data, not model quality or production readiness. """ import hashlib import hmac import json import sqlite3 import tempfile import time import unittest from pathlib import Path def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() def proposal(ticket_id, label, version): if type(ticket_id) is not str or ticket_id != "ticket-1": raise ValueError("unauthorized ticket") if label not in ("billing", "technical", "other"): raise ValueError("unsupported label") if type(version) is not int or version < 0: raise ValueError("invalid version") return {"ticket_id": ticket_id, "label": label, "version": version} def approval_digest(change): return hashlib.sha256(canonical(change)).hexdigest() def open_store(path): db = sqlite3.connect(path) db.executescript(""" CREATE TABLE IF NOT EXISTS tickets(id TEXT PRIMARY KEY, label TEXT, version INTEGER); CREATE TABLE IF NOT EXISTS operations(id TEXT PRIMARY KEY, digest TEXT NOT NULL); CREATE TABLE IF NOT EXISTS memory(owner TEXT, key TEXT, value TEXT, expires INTEGER, PRIMARY KEY(owner, key)); INSERT OR IGNORE INTO tickets VALUES ('ticket-1', 'other', 0); """) return db def execute_change(db, operation_id, change, approved_digest): # A real service must obtain identity/approval from its authenticated boundary. # Here the caller supplies a synthetic approval to make the test inspectable. if type(operation_id) is not str or not operation_id: raise ValueError("missing operation id") if set(change) != {"ticket_id", "label", "version"}: raise ValueError("unexpected arguments") checked = proposal(**change) digest = approval_digest(checked) if not hmac.compare_digest(digest, approved_digest): raise ValueError("approval does not match proposal") with db: db.execute("BEGIN IMMEDIATE") prior = db.execute("SELECT digest FROM operations WHERE id=?", (operation_id,)).fetchone() if prior: if prior[0] != digest: raise ValueError("idempotency key reused for different action") return "already_applied" cursor = db.execute("UPDATE tickets SET label=?, version=version+1 WHERE id=? AND version=?", (checked["label"], checked["ticket_id"], checked["version"])) if cursor.rowcount != 1: raise ValueError("stale record; new approval required") db.execute("INSERT INTO operations VALUES (?, ?)", (operation_id, digest)) return "applied" def retrieve(query, documents, allowed_ids): # Deterministic baseline, not semantic retrieval. Permission filter precedes ranking. words = set(query.lower().split()) ranked = [] for doc in documents: if doc["id"] not in allowed_ids: continue score = len(words & set(doc["text"].lower().split())) if score: ranked.append((score, doc["id"], doc)) return [doc for _, _, doc in sorted(ranked, key=lambda x: (-x[0], x[1]))] def remember(db, owner, key, value, expires): with db: db.execute("INSERT INTO memory VALUES (?, ?, ?, ?) ON CONFLICT(owner,key) " "DO UPDATE SET value=excluded.value, expires=excluded.expires", (owner, key, value, expires)) def recall(db, owner, key, now): row = db.execute("SELECT value FROM memory WHERE owner=? AND key=? AND expires>?", (owner, key, now)).fetchone() return row[0] if row else None def verify_webhook(raw_body, timestamp, signature, secret, now): # Illustrative signed format; production must use the actual provider's specification. if abs(now - timestamp) > 300: raise ValueError("expired timestamp") signed = str(timestamp).encode() + b"." + raw_body expected = hmac.new(secret, signed, hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature): raise ValueError("invalid signature") event = json.loads(raw_body) if not isinstance(event, dict) or not isinstance(event.get("id"), str) or not event["id"]: raise ValueError("invalid event") return event def bounded_plan(steps, tools, limit=3): results = [] if len(steps) > limit: raise ValueError("plan exceeds step budget") for step in steps: if step not in tools: raise ValueError("unknown tool") result = tools[step]() results.append({"tool": step, "result": result}) if result.get("status") != "ok": break return results class LabTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = Path(self.tmp.name) / "lab.sqlite" self.db = open_store(self.path) def tearDown(self): self.db.close() self.tmp.cleanup() def test_tool_arguments_and_authorization(self): with self.assertRaises(ValueError): proposal("someone-elses-ticket", "billing", 0) with self.assertRaises(ValueError): proposal("ticket-1", "delete", 0) with self.assertRaises(ValueError): proposal("ticket-1", "billing", True) def test_approval_is_bound_to_exact_change(self): change = proposal("ticket-1", "billing", 0) with self.assertRaises(ValueError): execute_change(self.db, "op1", {**change, "label": "technical"}, approval_digest(change)) self.assertEqual(self.db.execute("SELECT version FROM tickets").fetchone()[0], 0) def test_retry_and_crash_recovery(self): change = proposal("ticket-1", "billing", 0) self.assertEqual(execute_change(self.db, "op1", change, approval_digest(change)), "applied") self.db.close() self.db = open_store(self.path) self.assertEqual(execute_change(self.db, "op1", change, approval_digest(change)), "already_applied") self.assertEqual(self.db.execute("SELECT version FROM tickets").fetchone()[0], 1) different = proposal("ticket-1", "technical", 1) with self.assertRaises(ValueError): execute_change(self.db, "op1", different, approval_digest(different)) def test_stale_approval_and_rollback(self): change = proposal("ticket-1", "billing", 0) execute_change(self.db, "op1", change, approval_digest(change)) stale = proposal("ticket-1", "technical", 0) with self.assertRaises(ValueError): execute_change(self.db, "op2", stale, approval_digest(stale)) self.assertEqual(self.db.execute("SELECT count(*) FROM operations").fetchone()[0], 1) self.assertEqual(self.db.execute("SELECT label FROM tickets").fetchone()[0], "billing") def test_retrieval_permissions_and_no_evidence(self): docs = [{"id": "public", "text": "refund policy", "version": 1}, {"id": "private", "text": "refund policy secret", "version": 2}] self.assertEqual([d["id"] for d in retrieve("refund", docs, {"public"})], ["public"]) self.assertEqual(retrieve("unrelated", docs, {"public"}), []) def test_memory_scope_expiry_and_persistence(self): remember(self.db, "alice", "preference", "short answers", 100) self.db.close() self.db = open_store(self.path) self.assertEqual(recall(self.db, "alice", "preference", 99), "short answers") self.assertIsNone(recall(self.db, "bob", "preference", 99)) self.assertIsNone(recall(self.db, "alice", "preference", 100)) def test_webhook_integrity_and_freshness(self): raw = b'{"id":"event-1","ticket":"ticket-1"}' secret, timestamp = b"synthetic-lab-secret", int(time.time()) signature = hmac.new(secret, str(timestamp).encode() + b"." + raw, hashlib.sha256).hexdigest() self.assertEqual(verify_webhook(raw, timestamp, signature, secret, timestamp)["id"], "event-1") with self.assertRaises(ValueError): verify_webhook(raw + b" ", timestamp, signature, secret, timestamp) with self.assertRaises(ValueError): verify_webhook(raw, timestamp, signature, secret, timestamp + 301) def test_planning_stops_on_error_and_bounds_work(self): called = [] def fail(): called.append("lookup") return {"status": "error", "reason": "unavailable"} result = bounded_plan(["lookup", "write"], {"lookup": fail, "write": lambda: called.append("write")}) self.assertEqual(len(result), 1) self.assertEqual(called, ["lookup"]) with self.assertRaises(ValueError): bounded_plan(["unknown"], {}) with self.assertRaises(ValueError): bounded_plan(["lookup"] * 4, {"lookup": fail}) if __name__ == "__main__": unittest.main(verbosity=2)