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

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

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]: ...

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)

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,
)

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
        ...

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"]

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."""
        ...

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))

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)

# 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

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}

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