{"id":2890,"date":"2026-06-14T04:18:00","date_gmt":"2026-06-14T04:18:00","guid":{"rendered":"https:\/\/tucumandevelopers.com\/index.php\/2026\/06\/14\/agent-series-20-harness-in-production-from-single-file-to-reusable-package\/"},"modified":"2026-06-14T04:18:00","modified_gmt":"2026-06-14T04:18:00","slug":"agent-series-20-harness-in-production-from-single-file-to-reusable-package","status":"publish","type":"post","link":"https:\/\/tucumandevelopers.com\/index.php\/2026\/06\/14\/agent-series-20-harness-in-production-from-single-file-to-reusable-package\/","title":{"rendered":"Agent Series (20): Harness in Production \u2014 From Single File to Reusable Package"},"content":{"rendered":"
\n
<\/div>\n
\n

<\/a> From Demo Code to a Reusable Package <\/h2>\n

Article 19 used a 900-line harness_full_demo.py<\/code> to demonstrate eight defense layers. That file is good for explaining concepts, but not for reuse \u2014 all layers are coupled together, nothing can be tested in isolation, and nothing can be imported by another project.<\/p>\n

A production-grade Agent project needs something you can actually import<\/code>: <\/p>\n

\n
harness\/ \u251c\u2500\u2500 __init__.py Public API exports \u251c\u2500\u2500 registry.py Layer 2: ActionRegistry + PermissionLevel \u251c\u2500\u2500 budget.py Layer 3: PermissionBudget (with refund()) \u251c\u2500\u2500 sandbox.py Layer 4: sanitise_input + sandboxed_eval \u251c\u2500\u2500 audit.py Layer 6: ImmutableAuditLog (hash-chained) \u251c\u2500\u2500 rollback.py Layer 7: RollbackCoordinator \u2514\u2500\u2500 harness.py Unified entry point: AgentHarness <\/code><\/pre>\n
\n<\/p><\/div>\n<\/p><\/div>\n
This article starts with package design, covers three key API decisions, and finishes with two integration styles: standalone Python and LangGraph graph embedding.<\/p>\n
\n
  <\/a> Module Design <\/h2>\n
  <\/a> registry.py \u2014 Layer 2 <\/h3>\n
\n
class<\/span> PermissionLevel<\/span>(<\/span>Enum<\/span>):<\/span> READ<\/span> =<\/span> 1<\/span> WRITE<\/span> =<\/span> 2<\/span> ADMIN<\/span> =<\/span> 3<\/span> IRREVERSIBLE<\/span> =<\/span> 4<\/span> @dataclass<\/span> class<\/span> RegisteredAction<\/span>:<\/span> name<\/span>:<\/span> str<\/span> level<\/span>:<\/span> PermissionLevel<\/span> budget_cost<\/span>:<\/span> int<\/span> description<\/span>:<\/span> \"<\/span>str<\/span>\"<\/span> handler<\/span>:<\/span> Any<\/span> # Callable or BaseTool <\/span> class<\/span> ActionRegistry<\/span>:<\/span> def<\/span> register<\/span>(<\/span>self<\/span>,<\/span> action<\/span>:<\/span> RegisteredAction<\/span>)<\/span> -><\/span> None<\/span>:<\/span> ...<\/span> def<\/span> get<\/span>(<\/span>self<\/span>,<\/span> name<\/span>:<\/span> str<\/span>)<\/span> -><\/span> RegisteredAction<\/span>:<\/span> ...<\/span> # not found \u2192 PermissionError <\/span> def<\/span> is_allowed<\/span>(<\/span>self<\/span>,<\/span> name<\/span>:<\/span> str<\/span>)<\/span> -><\/span> bool<\/span>:<\/span> ...<\/span> def<\/span> names<\/span>(<\/span>self<\/span>)<\/span> -><\/span> list<\/span>[<\/span>str<\/span>]:<\/span> ...<\/span> <\/code><\/pre>\n
\n<\/p><\/div>\n<\/p><\/div>\n
get()<\/code> rather than __getitem__<\/code>: raises a consistent PermissionError<\/code>, without leaking the internal KeyError<\/code> detail.<\/p>\n
\n
  <\/a> budget.py \u2014 Layer 3 <\/h3>\n
\n
class<\/span> PermissionBudget<\/span>:<\/span> def<\/span> spend<\/span>(<\/span>self<\/span>,<\/span> action_name<\/span>:<\/span> str<\/span>,<\/span> cost<\/span>:<\/span> int<\/span>)<\/span> -><\/span> None<\/span>:<\/span> if<\/span> self<\/span>.<\/span>remaining<\/span> <<\/span> cost<\/span>:<\/span> raise<\/span> BudgetExhaustedError<\/span>(...)<\/span> self<\/span>.<\/span>remaining<\/span> -=<\/span> cost<\/span> def<\/span> refund<\/span>(<\/span>self<\/span>,<\/span> action_name<\/span>:<\/span> str<\/span>,<\/span> cost<\/span>:<\/span> int<\/span>)<\/span> -><\/span> None<\/span>:<\/span> self<\/span>.<\/span>remaining<\/span> =<\/span> min<\/span>(<\/span>self<\/span>.<\/span>total<\/span>,<\/span> self<\/span>.<\/span>remaining<\/span> +<\/span> cost<\/span>)<\/span> <\/code><\/pre>\n
\n<\/p><\/div>\n<\/p><\/div>\n
The new refund()<\/code> method fixes a design flaw from Article 19: budget was deducted before approval, and never returned on rejection. The production package corrects this \u2014 when an IRREVERSIBLE action is intercepted, harness.py<\/code> proactively calls refund()<\/code> to keep budget accounting accurate.<\/p>\n
\n
  <\/a> sandbox.py \u2014 Layer 4 <\/h3>\n
\n
INJECTION_PATTERN<\/span> =<\/span> re<\/span>.<\/span>compile<\/span>(<\/span> r<\/span>\"<\/span>(ignore.*(previous|above|prior)|forget.*instruction|<\/span>\"<\/span> r<\/span>\"<\/span>you are now|act as|jailbreak|bypass|<\/span>\"<\/span> r<\/span>\"<\/span>override.*system|system.*override|<\/span>\"<\/span> # both word orders covered <\/span> r<\/span>\"<\/span><\/s>|\\n\\n###|###\\s*system|<\\|im_start\\|>|system prompt)<\/span>\"<\/span>,<\/span> re<\/span>.<\/span>IGNORECASE<\/span>,<\/span> )<\/span> <\/code><\/pre>\n
\n<\/p><\/div>\n<\/p><\/div>\n
Two subtle points:<\/p>\n
    \n
  1. Both SYSTEM OVERRIDE<\/code> (system first) and override.*system<\/code> (override first) are covered<\/li>\n
  2.  \\n\\n###<\/code> matches a real newline, not the literal string \\\\n\\\\n###<\/code> <\/li>\n<\/ol>\n
    Both bugs were discovered and fixed during the adversarial tests in Article 21.<\/p>\n
    \n
      <\/a> audit.py \u2014 Layer 6 <\/h3>\n
    \n
    class<\/span> ImmutableAuditLog<\/span>:<\/span> def<\/span> log<\/span>(<\/span>self<\/span>,<\/span> action<\/span>,<\/span> actor<\/span>,<\/span> target<\/span>,<\/span> result<\/span>,<\/span> metadata<\/span>=<\/span>None<\/span>)<\/span> -><\/span> str<\/span>:<\/span> entry<\/span> =<\/span> {...,<\/span> \"<\/span>prev_hash<\/span>\"<\/span>:<\/span> self<\/span>.<\/span>_last_hash<\/span>}<\/span> entry<\/span>[<\/span>\"<\/span>hash<\/span>\"<\/span>]<\/span> =<\/span> self<\/span>.<\/span>_hash<\/span>(<\/span>json<\/span>.<\/span>dumps<\/span>(<\/span>entry<\/span>,<\/span> sort_keys<\/span>=<\/span>True<\/span>)<\/span> +<\/span> self<\/span>.<\/span>_last_hash<\/span>)<\/span> with<\/span> self<\/span>.<\/span>_path<\/span>.<\/span>open<\/span>(<\/span>\"<\/span>a<\/span>\"<\/span>)<\/span> as<\/span> f<\/span>:<\/span> # append-only <\/span> f<\/span>.<\/span>write<\/span>(<\/span>json<\/span>.<\/span>dumps<\/span>(<\/span>entry<\/span>)<\/span> +<\/span> \"<\/span>\\n<\/span>\"<\/span>)<\/span> return<\/span> entry<\/span>[<\/span>\"<\/span>hash<\/span>\"<\/span>]<\/span> def<\/span> verify_integrity<\/span>(<\/span>self<\/span>)<\/span> -><\/span> bool<\/span>:<\/span> # Replays the hash chain; any modified field returns False <\/span> ...<\/span> <\/code><\/pre>\n
    \n<\/p><\/div>\n<\/p><\/div>\n
    The __len__()<\/code> helper lets tests use len(audit)<\/code> to check entry count directly.<\/p>\n
    \n
      <\/a> rollback.py \u2014 Layer 7 <\/h3>\n
    \n
    class<\/span> RollbackCoordinator<\/span>:<\/span> @contextmanager<\/span> def<\/span> transaction<\/span>(<\/span>self<\/span>,<\/span> state<\/span>:<\/span> dict<\/span>,<\/span> op_name<\/span>:<\/span> str<\/span>):<\/span> snapshot<\/span> =<\/span> copy<\/span>.<\/span>deepcopy<\/span>(<\/span>state<\/span>)<\/span> self<\/span>.<\/span>_snapshots<\/span>.<\/span>append<\/span>({<\/span>\"<\/span>op<\/span>\"<\/span>:<\/span> op_name<\/span>,<\/span> \"<\/span>snapshot<\/span>\"<\/span>:<\/span> snapshot<\/span>})<\/span> try<\/span>:<\/span> yield<\/span> state<\/span> except<\/span> Exception<\/span>:<\/span> state<\/span>.<\/span>clear<\/span>()<\/span> state<\/span>.<\/span>update<\/span>(<\/span>snapshot<\/span>)<\/span> self<\/span>.<\/span>_snapshots<\/span>.<\/span>pop<\/span>()<\/span> raise<\/span> def<\/span> rollback_last<\/span>(<\/span>self<\/span>,<\/span> state<\/span>:<\/span> dict<\/span>)<\/span> -><\/span> str<\/span> |<\/span> None<\/span>:<\/span> \"\"\"<\/span>Manual trigger: undo the most recent committed transaction.<\/span>\"\"\"<\/span> if<\/span> not<\/span> self<\/span>.<\/span>_snapshots<\/span>:<\/span> return<\/span> None<\/span> entry<\/span> =<\/span> self<\/span>.<\/span>_snapshots<\/span>.<\/span>pop<\/span>()<\/span> state<\/span>.<\/span>clear<\/span>()<\/span> state<\/span>.<\/span>update<\/span>(<\/span>entry<\/span>[<\/span>\"<\/span>snapshot<\/span>\"<\/span>])<\/span> return<\/span> entry<\/span>[<\/span>\"<\/span>op<\/span>\"<\/span>]<\/span> <\/code><\/pre>\n
    \n<\/p><\/div>\n<\/p><\/div>\n
    rollback_last()<\/code> enables manual rollback: after a transaction commits, the snapshot is retained until explicitly confirmed or cleared by the caller.<\/p>\n
    \n
      <\/a> Unified Entry Point: AgentHarness <\/h2>\n
    \n
    class<\/span> AgentHarness<\/span>:<\/span> def<\/span> __init__<\/span>(<\/span>self<\/span>,<\/span> budget<\/span>:<\/span> int<\/span> =<\/span> 100<\/span>,<\/span> log_path<\/span>:<\/span> str<\/span> =<\/span> ...):<\/span> self<\/span>.<\/span>registry<\/span> =<\/span> ActionRegistry<\/span>()<\/span> self<\/span>.<\/span>budget<\/span> =<\/span> PermissionBudget<\/span>(<\/span>total<\/span>=<\/span>budget<\/span>)<\/span> self<\/span>.<\/span>audit<\/span> =<\/span> ImmutableAuditLog<\/span>(<\/span>log_path<\/span>=<\/span>log_path<\/span>)<\/span> self<\/span>.<\/span>rollback<\/span> =<\/span> RollbackCoordinator<\/span>()<\/span> self<\/span>.<\/span>_state<\/span>:<\/span> dict<\/span> =<\/span> {}<\/span> def<\/span> execute<\/span>(<\/span>self<\/span>,<\/span> action_name<\/span>:<\/span> str<\/span>,<\/span> actor<\/span>:<\/span> str<\/span> =<\/span> \"<\/span>agent<\/span>\"<\/span>,<\/span> **<\/span>kwargs<\/span>)<\/span> -><\/span> Any<\/span>:<\/span> # Layer 4: sanitise string arguments <\/span> # Layer 2: registry check (missing \u2192 PermissionError) <\/span> # Layer 3: budget deduction (insufficient \u2192 BudgetExhaustedError) <\/span> # Layer 5: IRREVERSIBLE \u2192 refund budget + raise HumanApprovalRequired <\/span> # Layer 7: WRITE\/ADMIN wrapped in rollback.transaction <\/span> # Layer 6: audit record <\/span> ...<\/span> def<\/span> approve_and_execute<\/span>(<\/span>self<\/span>,<\/span> action_name<\/span>:<\/span> str<\/span>,<\/span> actor<\/span>:<\/span> str<\/span> =<\/span> \"<\/span>human<\/span>\"<\/span>,<\/span> **<\/span>kwargs<\/span>)<\/span> -><\/span> Any<\/span>:<\/span> \"\"\"<\/span>Call this after catching HumanApprovalRequired to complete execution.<\/span>\"\"\"<\/span> ...<\/span> <\/code><\/pre>\n
    \n<\/p><\/div>\n<\/p><\/div>\n
    Why the two methods are separate:<\/strong><\/p>\n