Agent Series (20): Harness in Production — From Single File to Reusable Package

A developer refactored a 900-line demonstration file into a reusable Python package called 'harness', which includes modules for action registration, permission budgeting, input sanitization, audit logging, and rollback coordination. The package introduces a refund() method to correct a budget accounting flaw and supports integration as standalone Python or as a LangGraph graph embedding.

Article 19 used a 900-line harness full demo.py to demonstrate eight defense layers. That file is good for explaining concepts, but not for reuse — all layers are coupled together, nothing can be tested in isolation, and nothing can be imported by another project. A production-grade Agent project needs something you can actually import : harness/ ├── init .py Public API exports ├── registry.py Layer 2: ActionRegistry + PermissionLevel ├── budget.py Layer 3: PermissionBudget with refund ├── sandbox.py Layer 4: sanitise input + sandboxed eval ├── audit.py Layer 6: ImmutableAuditLog hash-chained ├── rollback.py Layer 7: RollbackCoordinator └── harness.py Unified entry point: AgentHarness This article starts with package design, covers three key API decisions, and finishes with two integration styles: standalone Python and LangGraph graph embedding. class PermissionLevel Enum : READ = 1 WRITE = 2 ADMIN = 3 IRREVERSIBLE = 4 @dataclass class RegisteredAction: name: str level: PermissionLevel budget cost: int description: "str" handler: Any Callable or BaseTool class ActionRegistry: def register self, action: RegisteredAction - None: ... def get self, name: str - RegisteredAction: ... not found → PermissionError def is allowed self, name: str - bool: ... def names self - list str : ... get rather than getitem : raises a consistent PermissionError , without leaking the internal KeyError detail. php class PermissionBudget: def spend self, action name: str, cost: int - None: if self.remaining < cost: raise BudgetExhaustedError ... self.remaining -= cost def refund self, action name: str, cost: int - None: self.remaining = min self.total, self.remaining + cost The new refund method fixes a design flaw from Article 19: budget was deducted before approval, and never returned on rejection. The production package corrects this — when an IRREVERSIBLE action is intercepted, harness.py proactively calls refund to keep budget accounting accurate. INJECTION PATTERN = re.compile r" ignore. previous|above|prior |forget. instruction|" r"you are now|act as|jailbreak|bypass|" r"override. system|system. override|" both word orders covered r"</s |\n\n | \s system|<\|im start\| |system prompt ", re.IGNORECASE, Two subtle points: SYSTEM OVERRIDE system first and override. system override first are covered \n\n matches a real newline, not the literal string \\n\\n Both bugs were discovered and fixed during the adversarial tests in Article 21. python class ImmutableAuditLog: def log self, action, actor, target, result, metadata=None - str: entry = {..., "prev hash": self. last hash} entry "hash" = self. hash json.dumps entry, sort keys=True + self. last hash with self. path.open "a" as f: append-only f.write json.dumps entry + "\n" return entry "hash" def verify integrity self - bool: Replays the hash chain; any modified field returns False ... The len helper lets tests use len audit to check entry count directly. python class RollbackCoordinator: @contextmanager def transaction self, state: dict, op name: str : snapshot = copy.deepcopy state self. snapshots.append {"op": op name, "snapshot": snapshot} try: yield state except Exception: state.clear state.update snapshot self. snapshots.pop raise def rollback last self, state: dict - str | None: """Manual trigger: undo the most recent committed transaction.""" if not self. snapshots: return None entry = self. snapshots.pop state.clear state.update entry "snapshot" return entry "op" rollback last enables manual rollback: after a transaction commits, the snapshot is retained until explicitly confirmed or cleared by the caller. python class AgentHarness: def init self, budget: int = 100, log path: str = ... : self.registry = ActionRegistry self.budget = PermissionBudget total=budget self.audit = ImmutableAuditLog log path=log path self.rollback = RollbackCoordinator self. state: dict = {} def execute self, action name: str, actor: str = "agent", kwargs - Any: Layer 4: sanitise string arguments Layer 2: registry check missing → PermissionError Layer 3: budget deduction insufficient → BudgetExhaustedError Layer 5: IRREVERSIBLE → refund budget + raise HumanApprovalRequired Layer 7: WRITE/ADMIN wrapped in rollback.transaction Layer 6: audit record ... def approve and execute self, action name: str, actor: str = "human", kwargs - Any: """Call this after catching HumanApprovalRequired to complete execution.""" ... Why the two methods are separate: execute is the automated path: all checks pass, execute immediately approve and execute is the human path: the caller explicitly signals "this has been approved"Merging them e.g., with an approved=False parameter makes intent ambiguous and harder to test. harness = AgentHarness budget=50 Register actions harness.registry.register RegisteredAction "read ticket", PermissionLevel.READ, 1, "Read Jira ticket", handler fn harness.registry.register RegisteredAction "write draft", PermissionLevel.WRITE, 3, "Write draft fix", handler fn harness.registry.register RegisteredAction "create pr", PermissionLevel.ADMIN, 8, "Open pull request", handler fn harness.registry.register RegisteredAction "merge to main", PermissionLevel.IRREVERSIBLE, 20, "Merge to main", handler fn READ → WRITE → ADMIN normal flow: r1 = harness.execute "read ticket", ticket id="BUG-101" r2 = harness.execute "write draft", ticket id="BUG-101", patch="fix: add null check" r3 = harness.execute "create pr", ticket id="BUG-101", title="fix: BUG-101" read=1 + write=3 + admin=8 = 12 spent, 38 remaining try: harness.execute "delete all data" except PermissionError as e: "Action 'delete all data' not in registry. Execution blocked." ... try: harness.execute "merge to main", pr id=1 except HumanApprovalRequired as e: print e.action name "merge to main" print e.action args {"pr id": 1} After human review: result = harness.approve and execute "merge to main", pr id=1 Key point : when execute intercepts an IRREVERSIBLE action, it calls budget.refund first. The net budget cost is zero. Only approve and execute actually charges the budget. budget=5, write cost=3 h = AgentHarness budget=5 h.execute "write draft", ... OK, 2 remaining h.execute "write draft", ... BudgetExhaustedError: need 3, remaining 2 Embedding the harness inside LangGraph's tools node : php def tools node state: HState - dict: last = state "messages" -1 results = for tc in last.tool calls: name, args = tc "name" , tc "args" try: reg = harness.registry.get name Layer 2 harness.budget.spend name, reg.budget cost Layer 3 if reg.level == PermissionLevel.IRREVERSIBLE: decision = interrupt {...} Layer 5: LangGraph primitive if decision = "approved": harness.budget.refund name, reg.budget cost harness.audit.log name, "checkpoint", ..., "HUMAN REJECTED" results.append ToolMessage content="rejected", ... continue if reg.level in WRITE, ADMIN : with harness.rollback.transaction harness. state, name : Layer 7 output = TOOL MAP name .invoke args else: output = TOOL MAP name .invoke args harness.audit.log name, "agent", ..., "EXECUTED" Layer 6 results.append ToolMessage content=str output , ... except PermissionError as e: harness.audit.log name, "registry", ..., "BLOCKED" results.append ToolMessage content=str e , ... except BudgetExhaustedError as e: results.append ToolMessage content=str e , ... return {"messages": results} tools node is the harness's natural insertion point: it intercepts before tool execution without touching any agent node reasoning layer logic. This package's behavior is fully verified by Article 21's test suite: Functional Layer 1–7 basic behaviour ████████████████████████████████ 19/19 PASS Adversarial injection / escalation ████████████████████████████████ 17/17 PASS Chaos fault injection / partial ████████████████████████████████ 9/ 9 PASS Total 45/ 45 tests passed Two real bugs found by the tests: INJECTION PATTERN only matched override. system , missing SYSTEM OVERRIDE reversed word order \\n\\n matched the literal string \n , not a real newline — jailbreak pattern System: slipped throughBoth fixed in sandbox.py with a one-line regex adjustment. Package Structure init .py exports only the public API; internal classes stay private AgentHarness acts as Facade; callers don't reach into subsystems directly API Design execute is the automated path covering the full Layer 2→7 chain approve and execute is the human path; the caller signals "approved" refund when IRREVERSIBLE is intercepted, keeping accounting accurate PermissionError / BudgetExhaustedError / HumanApprovalRequired exported from init .py Sandbox \n is a real newline character, not the literal \\n LangGraph Integration tools node , not in agent node interrupt , not a Python exceptionFive core conclusions: execute and approve and execute makes intent explicit tools node is the harness's natural slot Check out PrimeSkills — a curated marketplace of AI agents and skills that have been validated in real-world, enterprise-grade workflows. No fluff, just what actually works. Find more useful knowledge and interesting products on my Homepage