Models are getting smarter, yet production agents keep breaking — not because the AI failed, but because the infrastructure around it did. Harness engineering is the operational backbone that keeps autonomous agents bounded, cost-controlled, and context-aware over complex, long-horizon tasks. Simply put: you don’t need a better model to build reliable AI software, you need a better harness.
Let’s briefly understand how we arrived at Harness & Loop Engineering.
1 Large Language Models (LLMs): Raw Artisan Worker
Statistical next-token prediction based on vast weights.
You hire an incredibly intelligent, raw artisan worker. They know nearly everything about general history, code syntax, and logic, but they have complete amnesia between tasks and hold zero corporate context.
Limitation: Passing a short phrase like "Write code" produces an abstract, generic snippet. It lacks custom business logic, architecture patterns, or styling standards. The model has no mechanism to interact with external reality.
Solution: We realized we needed to give this artisan specific, contextual instructions for the task at hand.
2 Prompt Engineering: Instruction Manual
In-context learning via system instructions, Few-Shot examples, and Chain-of-Thought (CoT) structures.
Instead of telling the artisan "Build a widget," you hand them a highly explicit Instruction Manual. It specifies: "Act as a master carpenter. First, measure twice. Second, cut once. Here are two examples of perfect cuts."
Limitation: You cannot fit your entire repository, system logs, and corporate database into a single static instruction manual. As the manual grows longer, the artisan gets distracted, suffers from “attention dilution,” and forgets the middle instructions.
Solution: Instead of typing massive manuals, we tried altering the model’s internal brain directly via Fine-Tuning.
3 Fine-Tuning: Muscle Memory / Apprenticeship
Weight updates via supervised training on specific (Input, Output) dataset pairs.
You send the artisan to a 3-month intensive vocational school focused entirely on your company’s proprietary blueprint schemas. They return with natural muscle memory for your specific formats.
Limitation: Fine-tuning is static, slow to update, and highly expensive. It changes the worker’s instincts, but it does not give them real-time access to live, changing environment data (like current git branches, API failures, or a database). If an API schema changes tomorrow, the muscle memory becomes actively harmful.
Solution: We shifted from static generation to dynamic execution by giving models the ability to execute tools.
4 Agents: Giving the Artisan Hand Tools
Reasoning-Action (ReAct) cycle. The model generates a thought, invokes a tool call string, waits for a response, and continues.
You hand the artisan a set of physical tools (a hammer, a saw, a measuring tape). The artisan can now look at the blueprint, grab the hammer, smash a nail, look at the result, and decide what tool to use next.
Limitation: If the tool outputs a massive payload (like a 5,000-line error log), it gets dumped straight back into the conversation history. The artisan’s desk becomes completely buried under mountains of raw text, causing severe information overload (“Context Rot”).
Solution: We moved from basic tool execution to programmatic management of the model’s active memory layer.
5 Context Engineering: Workspace Workbench
Dynamic Retrieval-Augmented Generation (RAG), conversation history compaction, sliding context windows, and FIFO token trimmers.
You build an active Workbench Environment for the artisan. An automated desk assistant constantly cleans up wood shavings (summarizes old history), files away giant blueprints to a side drawer (offloads token-heavy data), and keeps only the relevant tools and pages directly in front of the artisan’s eyes.
Limitation: While the workspace is clean, the artisan is still structurally unconstrained. They can confidently loop forever trying to hit a screw with a hammer, run a destructive command that deletes the factory floor (rm -rf /), or quietly fake their evaluation checks out of self-evaluation bias.
Solution: This led directly to Harness Engineering.
6 Harness Engineering: Factory Machine Scaffold
The deterministic runtime framework wrapped completely around the LLM session. It handles schema validation, sandbox isolation, programmatic linters, and strict operational boundaries.
You place the artisan inside an engineered factory machine rig. The artisan does not touch raw materials directly; they manipulate control levers. If they try to execute a command that will destroy the factory floor, the safety cage mechanically locks down. If they generate code, automated laser sensors (linters/compilers) test it instantly. If it fails, the error feeds back into the rig’s display panel automatically before any human looks at it.
In 2026, we accept that Agent = Model + Harness. The raw model proposes an action, but the harness executes it, validates it, and governs it.
Operational Gap: The harness perfectly controls a single session for a single explicit task. But what happens when you need to run dozens of sessions concurrently across a 2-week development lifecycle without human management?
7 Loop Engineering: Automated Production Line
Asynchronous, multi-agent orchestration structures (like the Ralph Wiggum Loop or Generator-Evaluator Sprint Contracts) executing recursive multi-step goal paths.
This is the complete Automated Production Line Facility. It manages multiple separate harnessed machine rigs working sequentially.
Rig A (The Planner) maps out the build strategy onto an execution file on disk.
Rig B (The Generator) takes that plan and crafts components under strict harness constraints.
Rig C (The Independent Evaluator) pulls the components into an isolated testing chamber to verify execution against an unyielding “Definition of Done” contract.
Evolution journey at a glance
Every layer of this architecture does not replace its predecessor; it wraps it. We optimize prompts to guide internal reasoning; we use context engineering to feed high-fidelity inputs; we deploy a harness to keep execution safe and deterministic; and we engineer loops to drive complex initiatives to completion.
Why Harness Engineering Matters Now
Understanding why we have shifted to harness engineering requires confronting a fundamental reality of building production-grade AI systems: Models have gotten incredibly smart, but raw model intelligence does not translate to system reliability.
If you look under the hood of a cutting-edge autonomous engineering agent — like Anthropic’s Claude Code or Google’s Agents CLI — you will find an surprising ratio. One teardown of Claude Code revealed that roughly 98% of the codebase is the harness, while the underlying model handles only the core reasoning slices.
Transition to harness engineering is driven by 3 main factors:
- Ceiling of Prompting and “Spiky” Intelligence Problem
In the early days of LLMs, engineering teams treated the model as a deterministic function: change the input string (the prompt), expect a better output. However, frontier models possess what the industry calls spiky intelligence. A model might successfully execute a brilliant, complex 200-line algorithmic refactor on the first try, and then immediately fail on the next turn because it hallucinated a flag on a standard git command.
Core Limitation: When you attempt to solve these edge-case failures by writing longer, more complex system prompts, you encounter the Law of Diminishing Context Returns. As a prompt ballooned into a massive document of rules:
It crowded out the model’s active working memory (the context window).
The model suffered from attention dilution, frequently pattern-matching local solutions while dropping global constraints.
The rules rotted instantly as your real-world codebase, APIs, or internal tooling evolved.
Harness Realization: We cannot prompt away a model’s intrinsic next-token architectural variance. Instead of trying to make the model’s brain 100% perfect via text, we build a deterministic engineering scaffold around it. If a rule matters, we enforce it via a programmatic linter or structural constraint in the runtime, not a sentence in a prompt.
Formulation The modern consensus across teams at OpenAI, Anthropic, and LangChain is defined by a simple architectural equation:
Agent = Model + Harness
Model functions as the CPU, supplying raw mathematical reasoning. Harness acts as the Operating System Layer.
A raw model cannot maintain durable state, safely execute shell code, inspect complex repository trees, or recover gracefully from an unhandled API crash. The harness provides the environment that makes model intelligence useful and actionable.
To see the direct impact of this approach, look at LangChain’s optimization work on deep coding agents. By focusing entirely on refining the execution harness — improving verification loops, sharpening context assembly, and hardening tool boundaries — they advanced an agent from 30th to 5th place on the industry-standard Terminal Bench 2.0 leaderboard without changing a single line of the underlying model weights.
- Functional Paradigm Shift
Harness engineering shifts your daily responsibilities as an AI architect. The differences highlight how this approach changes development practices:
- Engineering Principle: Ratchet Principle
Core operating philosophy of harness engineering was crystallized by Mitchell Hashimoto (creator of Terraform):
“Every time you discover an agent has made a mistake, you take the time to engineer a solution so that it can never make that mistake again.”
This is known as the Ratchet Principle.
In standard software engineering, when a bug slips into production, you write a unit test to prevent regression. In agent architecture, when an agent fails a task, you do not rewrite the prompt. You add a middleware validation hook, an automated verification loop, or a structural constraint to the harness.
The harness acts as a mechanical ratchet: it tightens with every failure, permanently shrinking the agent’s failure surface area while expanding its autonomous execution capability.
Architectural Taxonomy
To build a clean harness, you must separate your system into 3 distinct runtime layers. If you conflate these layers, you end up with messy code where application logic, prompt templates, and security guardrails are tangled together.
Static and dynamic text templates that define the persona, behavioral rules, and reasoning strategies (e.g., Chain-of-Thought) of the LLM.
Printed orientation packet handed to the artisan worker when they walk onto the floor.
Harness Responsibility: Version controlling these templates and injecting dynamic runtime variables (like current date or target user ID) before calling the model.
Layer 2: Inference Context Layer ( Active Workspace Memory)
The active, stateful window managed by the harness. It controls exactly what tokens enter the model’s short-term memory (RAM) at turn N. This includes conversation history management, automated pruning, and context compaction.
The physical workbench surface. The harness acting as a desk assistant, constantly clearing out raw lumber scraps and sliding relevant technical spec sheets into view.
Harness Responsibility: Token-budget enforcement, summarizing old conversational turns, and stripping out large raw strings from tool outputs before they exhaust the model’s context window.
Layer 3: System Execution Environments (Sandbox & Tools)
The isolated, physical runtime where the agent’s requested actions actually execute. This is entirely decoupled from the LLM’s memory or prompt space. It includes Docker containers, WebAssembly sandboxes, and Model Context Protocol (MCP) servers.
The enclosed, automated machine cage where a robotic saw cuts wood. The artisan flips switches outside the cage; they never reach their own hands into the blades.
Harness Responsibility: Catching tool crashes, enforcing file-system read/write permissions, monitoring process timeouts, and containing destructive actions (like a rogue rm -rf).
Trap of Over-Engineering the control flow When senior engineers start building agents, their natural instinct is to write complex state machines using conditional code blocks:
This approach fails in production because it attempts to micromanage a reasoning engine using static code paths. It creates rigid graphs that break the moment the model encounters an unpredicted edge case. You end up building an extensive labyrinth of Python logic trying to account for every permutation of model behavior, which leaves you with a fragile system that hovers around an 80% success rate.
“Bitter Lesson” Applied to Harnesses
Rich Sutton’s famous essay “Bitter Lesson” points out that general methods that leverage computation are ultimately most effective. In agent architecture, this means: Stop writing complex, hand-coded control flows to guide the agent, and stop exposing cluttered, low-level tools.
Instead, maximize reliability by focusing on two harness design patterns:
Strip Hand-Coded Graphs: Let the model handle the routing and control flow natively via its own internal reasoning loop. Remove your nested if/else steps and allow the LLM to decide what to do next based on the environment's state.
Fragile, Chatty Approach (80% Success) Giving the agent tools like open_file(), read_line(line_num), insert_text(line_num, text), and save_file(). The model wastes enormous context tokens orchestrating these microscopic operations, frequently messing up index math and introducing syntax errors.
Cohesive, High-Level Approach (100% Success) Providing a single, robust tool: apply_patch(diff_string). The harness takes the model's generated diff, runs a deterministic patch engine behind the scenes, and automatically runs a syntax linter or compiler over the result. If the patch fails or the linter catches an error, the harness surfaces the error back to the model cleanly.
By moving structural complexity out of prompt logic and into high-level, self-validating tools, you reduce tool noise and allow the model’s reasoning capabilities to shine.
Ratchet Principle in Action
As discussed earlier, Ratchet Principle dictates that you never patch a recurring agent failure by tweaking a prompt. Instead, you permanently narrow the failure surface area by updating the harness.
Production Ex: Unhandled Flag Failure Imagine a development agent tasked with rolling back a change. It decides to execute a shell tool call: git rollback --hard HEAD~1
The underlying shell throws an error because --hard is not a valid flag for a non-existent rollback command; it should have used git reset --hard HEAD~1
Prompt Patching Anti-Pattern: You open system_prompt.txt and append: "NOTE: Do not use git rollback. Always use git reset --hard when you want to revert changes". This works temporarily, but three weeks later, when the context window fills up, the model drops this rule and hits the exact same error.
Harness Ratchet Pattern: You intercept the tool execution at Layer 3 (System Execution Environment). You write an explicit middleware or tool wrapper that intercepts git commands. If the model invokes an invalid subcommand, the harness intercepts it and returns a structured error: "Error: 'git rollback' is not a valid command. Did you mean 'git reset --hard'?"
The system has now mechanically ratcheted. You have changed the environment so that this specific class of failure safely self-corrects at runtime, regardless of prompt length or attention dilution.
Model-Harness Training Loop
The boundary between the model’s brain and the harness has been structurally blurred. Frontier models are no longer trained in a vacuum; they are post-trained (via RLHF and DPO) with specific harnesses directly inside the training loop. This shift completely changes how we choose, tune, and debug our stacks.
Co-Training When AI labs (like Anthropic or OpenAI) prepare a model like Claude 3.5 Sonnet or GPT-5.2-Codex for release, they don’t just teach it to output raw text. They want it to power their flagship agent products (Claude Code, Codex CLI). To do this, they place the model inside their proprietary harness during the reinforcement learning phase. The model learns to optimize its thinking specifically for the tools, file-system drivers, and back-pressure signals designed by those specific lab engineers.
Direct Consequence: Tool Overfitting Because of this co-training, the model develops an architectural coupling to its parent environment. Ex:
Claude 3.5 Sonnet inside Claude Code is trained to expect highly specific tool names like str_replace or view_file. It knows exactly how the harness will respond if a search fails, and its internal weights are optimized to parse that precise error format.
If you take that exact same model out of Claude Code and drop it into a custom corporate harness that uses apply_patch or a custom python grep() script, the model's performance can experience unexpected regressions. A purely general reasoning engine wouldn't care about the difference between str_replace and apply_patch; it would adapt immediately. But modern frontier models are highly overfitted to the behavioral grammar of the harnesses they grew up inside.
If you treat a harness as a simple, static configuration file or an interchangeable wrapper, your enterprise agents will stall at a performance ceiling. Understanding co-training changes your implementation strategy in three ways:
A. “Harness Swap” Paradox As proven by Endor Labs and Viv’s research on Terminal Bench 2.0, a model running in a native harness often scores lower than the same model running in a highly optimized custom harness.
Data: A raw model variant running in its native lab harness scored 61.5% on complex functionality tests. The exact same model, during the same week, dropped into a custom harness (with optimized local context middleware and strict verification loops) surged to 87.2%.
Takeaway: Do not assume the model provider’s CLI is the gold standard for your business logic. The provider’s harness is built for generalized tasks. By building a custom harness tailored to your specific codebase or problem domain, you can unlock latent reasoning capabilities that the native harness left on the floor.
B. Tooling Regressions are Often Silent Weight Mismatches When you modify a tool’s internal logic, rewrite a prompt template, or alter an error message in your harness, and the agent suddenly starts hallucinating or repeating actions, you haven’t necessarily broken your python code. Instead, you have altered the “sensor” or “driver” feedback that the model’s weights expect based on its post-training. If the model expects a tool to return a specific string structure on failure and you clean up that string to make it look “nicer”, you can break the model’s internal self-correction pathways.
C. Designing for “Context Anxiety” Anthropic’s engineering teams documented a phenomenon known as context anxiety in modern models. Because models are trained inside harnesses that monitor token budgets, the model begins to show behavioral shifts (such as prematurely wrapping up work, skipping edge-case checks, or hallucinating a “done” state) as the active history approaches its perceived limit. A professional harness must actively manage this behavior. It needs features like LocalContextMiddleware to handle context resets and clear out short-term memory before the model’s internal threshold triggers this performance drop.
To truly appreciate why elite engineering agents require hundreds of lines of harness code, we must construct a bare-metal execution loop. We will intentionally remove all abstractions, middleware, and safety guardrails, exposing a raw LLM straight to tools. Then, we will walk through the exact runtime friction points where this un-harnessed setup completely falls apart under enterprise conditions.
Below is a complete, production-grade implementation of a raw, bare-metal Reasoning-Action (ReAct) loop. It uses the google-genai SDK and directly mirrors the core loop execution mechanism behind lightweight CLIs—but with zero harness layers.
This script exposes two powerful low-level system tools to the model: read_file and run_bash_command.
import osimport subprocessfrom google import genaifrom google.genai import types# =====================================================================# LAYER 3: Bare-Metal Tool Drivers (Exposed directly with no isolation)# =====================================================================def read_file(path: str) -> str: """Reads the contents of a file in the workspace directory.""" print(f" [DRV] Executing read_file for: {path}") with open(path, 'r', encoding='utf-8') as f: return f.read()def run_bash_command(command: str) -> str: """Executes an unrestricted bash command directly on the host system system.""" print(f" [DRV] Executing bash shell: '{command}'") # CRITICAL INSECURITY: No container sandbox, no timeouts, no network gating result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=30 # Primitive hard timeout ) return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"# Mapping string keys to live functions for manual executionTOOL_MAP = { "read_file": read_file, "run_bash_command": run_bash_command}# =====================================================================# LAYER 1 & 2: Execution Init (Direct Context & Prompt Injection)# =====================================================================client = genai.Client()model_id = 'gemini-1.5-pro' # Or gemini-2.0-flash, or Claude equivalent via API# The objective requires multi-step investigation and file modificationconversation_history = [ { "role": "user", "parts": [{"text": "Fix all broken unit tests in the current workspace directory."}] }]# Explicitly mapping Python signatures to Gemini Function Declarationstools_spec = [ types.Tool(function_declarations=[ types.FunctionDeclaration( name="read_file", description="Reads the complete text contents of a specified file path.", parameters={ "type": "OBJECT", "properties": {"path": {"type": "STRING", "description": "Relative workspace path"}}, "required": ["path"] } ), types.Tool(function_declarations=[ types.FunctionDeclaration( name="run_bash_command", description="Runs an arbitrary bash string directly on the system console.", parameters={ "type": "OBJECT", "properties": {"command": {"type": "STRING", "description": "The shell command to run"}}, "required": ["command"] } ) ]) ])]# =====================================================================# THE UNBOUNDED LOOP PLANE (No Harness Enforcements)# =====================================================================print("\n🚀 Starting the Bare-Metal Agent Loop (No Harness Protection)...")step_count = 0while True: step_count += 1 print(f"\n--- [TURN {step_count}] Requesting Model Inference (CPU) ---") # Send the raw, growing interaction list directly to the model response = client.models.generate_content( model=model_id, contents=conversation_history, config=types.GenerateContentConfig(tools=tools_spec) ) # Cache the assistant's internal thoughts and tool calls in memory model_turn_content = response.candidates[0].content conversation_history.append({"role": "model", "parts": model_turn_content.parts}) # Inspect if the model's response contains tool invocation calls function_calls = response.candidates[0].function_calls if not function_calls: print("✅ The model has concluded operations (No tool calls requested).") if model_turn_content.parts: print(f"Final Model Response: {model_turn_content.parts[0].text}") break # Process requested tool calls sequentially for call in function_calls: print(f"🛠️ Model Requested Tool: {call.name}(args={call.args})") if call.name not in TOOL_MAP: # Primitive error feedback loop tool_output = f"Error: Tool '{call.name}' is not registered in this system." else: try: # Direct dynamic execution using extracted dictionary arguments tool_output = TOOL_MAP[call.name](**call.args) except Exception as e: tool_output = f"Runtime Tool Execution Exception: {str(e)}" # Append the raw tool return payload directly back into active RAM (Context Window) conversation_history.append({ "role": "tool", "parts": [types.Part.from_function_response( name=call.name, response={"result": tool_output} )] })
When you deploy the above loop into real-world software consulting contexts (such as multi-page site migrations or running continuous integration checks), it hits an operational wall. Let’s trace the four cascading failures that occur when this script runs without a harness.
- Silent Infinite Loop
The agent runs pytest. The tests fail because of a minor syntax discrepancy in a test file. The agent invokes read_file("tests/test_core.py"), analyzes it, notices an issue, uses run_bash_command("echo 'new code' > tests/test_core.py") to overwrite it, and runs pytest again. However, its bash command introduces a typo. The test continues to fail.
Because there are no loop constraints, the agent enters a recursive loop. Turn 4, Turn 5, Turn 50 look identical. The model repetitively reads the file, applies the exact same broken fix, and checks the results. It has no self-awareness that it is looping, and your API usage metrics spike without making any real progress.
- Tool-Call Crash / Fatal Halt
The model attempts to execute a complex find-and-replace using a python command via bash: python -c "import sys; ... ". It messes up the escape quotes inside the JSON tool arguments. The JSON parser in the native SDK fails to deserialize the arguments block, or the host shell throws an unhandled syntax exception.
In our bare-metal script, an unhandled tool exception or a malformed JSON argument block causes the Python script to crash with a traceback, terminating the entire process. Any progress the agent made over the last 20 turns is completely lost because the runtime lacked an error-interception wrapper.
- Context Window Exhaustion (RAM Overload)
To find out why a test is failing, the agent executes run_bash_command("cat logs/output.log"). The log file happens to be an un-rotated enterprise service trace containing 8,500 lines of JSON telemetry data (equivalent to ~180,000 tokens).
The raw tool driver blindly dumps all 180,000 tokens directly into conversation_history. On the very next iteration, this entire block is re-transmitted to the API.
The model’s attention dilutes, causing it to drop your original instruction to fix the unit tests.
You hit the strict token ceiling, causing the next API call to return a 429 Token Limit Exceeded error.
Your infrastructure costs spike because you are re-sending a massive log file back and forth across every remaining turn of the conversation.
- State Amnesia & “Cold Reboot” Problem
The agent has successfully navigated 45 turns, read 12 files, isolated a core database connection leak, and rewritten 3 modules. At turn 46, your local internet connection drops briefly, or the API provider experiences a temporary 502 Bad Gateway error.
Because the entire state of the interaction exists exclusively as a transient Python list in volatile memory (conversation_history), the crash wipes out the session. You cannot hit "resume." When you restart the script, the agent boots up with a blank slate, entirely unaware of what it altered across the directory on its previous run. It suffers from complete state amnesia.
7 Core Components of an Agent Harness
core components of modern harness engineering as of 2026.
- Context Engineering & Memory Primitives
Scope: Active token optimization and active workspace memory.
What breaks if missing: Attention Dilution Collapse. A tool dumps a multi-megabyte server log into the history. The agent instantly overflows its context window, incurs massive API costs, and forgets the original system instructions because the prompt gets crowded out of the model’s active focus.
To implement this: Build an execution middleware that monitors current token usage per turn. When utilization crosses a specific threshold (e.g., 80% of capacity), a Compaction Hook is triggered. This hook leaves the most recent 3 turns untouched, summarizes older turns into a high-level summary block, and offloads raw tool data directly to a local file system pointer.
This keeps the model’s active context short and clear. It prevents the model from processing unnecessary tokens, ensuring it focuses its attention on the current task variables rather than wading through raw historical data.
- Tool Orchestration & Execution Guardrails
Scope: Defining schemas, isolating the sandbox, and validating arguments.
What breaks if missing:Arbitrary Host Execution / Hallucinated Parameter Crashes. The model either generates a malformed JSON payload that crashes your orchestrator script, or it gets tricked by a prompt injection into executing a destructive command (rm -rf /) directly on your physical machine.
To implement this: Decouple tool calling from execution using a strict Propose-and-Execute pattern. The model outputs a structured tool call; the harness intercepts it, runs it through a JSON-Schema validator, checks the requested path against an allow-list, and runs the operation inside an ephemeral, network-isolated container (like a Docker or WebAssembly microVM).
[Model Proposes Call] ──► [Harness Schema & Path Check] ──► [Isolated Container Exec]
It establishes a clear boundary between the reasoning engine and actual execution. The model can propose any action it wants, but it can only interact with the real world through a strict, validated API driver managed by the harness.
- State & Memory Management
Scope: Maintaining a persistent file system, tracking execution progress, and handling checkpoint restores.
What breaks if missing: State Amnesia. At turn 45 of a complex task, a temporary network failure or API timeout occurs. Because all history is stored in volatile memory (like a Python list), the session crashes. When restarted, the agent has no memory of what it was doing, leaving your directory in a partially modified state.
To implement this: Implement Transactional Checkpointing after every complete turn. Write the current conversation history, tool statuses, and file system diffs to a persistent database or a structured local file (e.g., .agent_snapshot.json).
This separates the agent’s short-term execution state from volatile process memory. If the application crashes, the harness reads the last verified snapshot on boot, allowing the agent to pick up exactly where it left off without losing its place in the workflow.
- Verification & Safety (Quality Gates)
Scope: Code linting, compiling checks, and automated retry loops.
What breaks if missing: Illusion of Done. The agent encounters an error, writes a code fix that introduces a syntax error, and confidently reports: “I have fixed the issue.” The agent assumes it succeeded based purely on its own generation, without verifying if the code actually works.
To implement this: Integrate Automated Post-Execution Hooks. When an agent attempts to save a file or mark a task completed, the harness intercepts the action and automatically runs a deterministic validation tool (like a syntax linter, a compiler, or an integration test suite).
This forces the agent’s progress to ground itself in objective system realities rather than subjective text predictions. If a linter catches an error, the harness feeds the raw compiler error straight back into the agent’s execution loop, prompting it to self-correct before a human ever reviews the code.
- Human-in-the-Loop (HITL) Controls
Scope: Managing approval gates, checking permissions, and calibrating action risk.
What breaks if missing: Uncontrolled Side Effects. An autonomous agent decides to fix a deployment issue by forcefully deleting a production database table or pushing an unreviewed commit straight to the main branch.
To implement this: Classify every registered tool into a specific Risk Tier Matrix:
Tier 3 (External Write): git_push, drop_table→ Triggers a hard execution , posts a confirmation payload to a human dashboard, and waits for a signed approval token before proceeding.
This ensures safety without slowing down performance. The agent runs autonomously on low-risk tasks, but the harness acts as a hard break the moment it attempts an action with significant real-world consequences.
- Lifecycle Management & Fault Recovery
Scope: Managing step budgets, tracking wall-clock time, and gracefully recovering from crashes.
What breaks if missing: Infinite Token Melt. The model gets stuck in a recursive loop, executing the exact same failing bash command hundreds of times. This drains your API credits and burns system resources while making zero actual progress.
To implement this: Initialize every agent loop session with a strict Resource Budget Object:
If the loop iterations pass step 30, or the cumulative API cost exceeds $2.00, the harness breaks the execution loop, saves a crash snapshot, and surfaces an error log to the system administrator.
It treats model execution like an untrusted multi-threaded compute job. It enforces hard operational limits directly at the engine level, ensuring runaway loops are caught and shut down quickly.
- Observability & Hierarchical Tracing
Scope: OpenTelemetry integration, logging performance metrics, and capturing execution trajectories.
What breaks if missing: Black Box Problem An agent runs for 5 minutes and eventually fails. Without detailed tracing, you cannot tell why it failed. Was it a poorly worded prompt instruction, a bad tool schema description, a model hallucination, or a hidden API timeout?
To implement this: Wrap the execution loop in a hierarchical tracing framework (like OpenTelemetry or LangSmith). Generate a unique Root_Trace_ID for each goal, and emit distinct span logs for every individual step: Context_Assembly -> Model_Inference -> Tool_Execution_Payload -> Validation_Output
It provides clean visibility into complex, multi-step agent behaviors. It gives developers a clear look under the hood, turning vague agent trajectories into organized, searchable log data you can use to debug failures and improve your system.
We created a raw, un-harnessed ReAct loop and analyzed how a single massive tool output (like an un-rotated log file or a huge repository file) can instantly overwhelm the model’s active memory (RAM), dilute its attention, or crash the session via token exhaustion.
Now, we will refactor that baseline script. We are going to build a production-grade Context Engineering Engine directly on top of the raw Gemini API. We will implement three core architectural primitives:
Token-Counting Budget Guardrail to measure real-time context inflation.
Tool-Call Off & Truncation Engine to prevent massive file dumps from polluting memory.
Sliding History Compaction Layer that preserves critical system goals and recent execution steps while summarizing old conversational turns.
Here is our baseline script transformed with a dedicated, programmatic memory management layer.
import osimport subprocessfrom google import genaifrom google.genai import types# =====================================================================# HARNESS INNER COMPONENT: Token & Memory Management Engine# =====================================================================class ContextManager: def __init__(self, client: genai.Client, model_id: str, max_context_tokens: int = 30000): self.client = client self.model_id = model_id self.max_context_tokens = max_context_tokens # We explicitly lock down system instructions to protect them from compaction self.system_instruction = "You are an elite software engineering agent. Fix broken code methodically." self.history = [] def append_message(self, role: str, parts: list): """Appends a new turn to the history.""" self.history.append({"role": role, "parts": parts}) def count_tokens(self) -> int: """Calculates the exact total token usage of the current active context.""" # Wrap history into contents to check token counts accurately via the SDK response = self.client.models.count_tokens( model=self.model_id, contents=self.history ) return response.total_tokens def process_tool_output(self, tool_name: str, raw_output: str, max_lines: int = 40) -> dict: """ PRIMITIVE 1: Tool-Call Off / Edge Truncation Driver Prevents giant log dumps from hitting the context window directly. """ lines = raw_output.splitlines() if len(lines) <= max_lines: return {"result": raw_output} # Keep only the head and tail signals of the massive payload head = "\n".join(lines[:15]) tail = "\n".join(lines[-15:]) truncated_summary = ( f"[HARNESS TRUNCATION ALERT]: Output exceeded {max_lines} lines ({len(lines)} lines total).\n" f"--- FIRST 15 LINES ---\n{head}\n" f"... [TRUNCATED {len(lines) - 30} LINES OF RAW OUTPUT BY THE HARNESS] ...\n" f"--- LAST 15 LINES ---\n{tail}\n" f"[SYSTEM NOTE]: The full untruncated artifact has been offloaded to your workspace filesystem." ) return {"result": truncated_summary} def enforce_compaction(self): """ PRIMITIVE 2: Sliding Window History Compaction Layer Summarizes old turns when token counts threaten memory limits. """ current_tokens = self.count_tokens() print(f" [HARNESS] Active Memory Check: {current_tokens} / {self.max_context_tokens} tokens utilized.") if current_tokens < self.max_context_tokens: return # Context is within healthy operational bounds print(" ⚠️ [HARNESS CRITICAL] Context threshold breached! Invoking Compaction Engine...") # Guardrail: Ensure we have enough history depth to compact if len(self.history) < 6: return # We preserve the original User Goal (Index 0) and the last 2 turns (Active Buffer) user_goal = self.history[0] active_buffer = self.history[-3:] mid_history = self.history[1:-3] # Request a lightweight model invocation to compress the middle history segment compaction_prompt = ( "Summarize the technical tasks completed, tools executed, and errors encountered " f"in the following historical developer log segment into a concise bulleted list:\n\n{str(mid_history)}" ) print(" [HARNESS] Running out-of-band context condensation summary...") summary_response = self.client.models.generate_content( model='gemini-2.0-flash', # Use a fast, cost-effective model for harness tasks contents=compaction_prompt ) summary_text = summary_response.text # Re-assemble the context window compacted_history = [ user_goal, { "role": "user", "parts": [{"text": f"[HARNESS HISTORICAL ARCHIVE SUMMARY]:\n{summary_text}"}] } ] compacted_history.extend(active_buffer) self.history = compacted_history print(f" [HARNESS] Compaction successful. New token count: {self.count_tokens()}")# =====================================================================# SYSTEM EXECUTION & BARE-METAL TOOLS# =====================================================================def read_file(path: str) -> str: with open(path, 'r', encoding='utf-8') as f: return f.read()def run_bash_command(command: str) -> str: result = subprocess.run(command, shell=True, capture_output=True, text=True) return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"TOOL_MAP = {"read_file": read_file, "run_bash_command": run_bash_command}# =====================================================================# ORCHESTRATION PIPELINE# =====================================================================client = genai.Client()model_id = 'gemini-1.5-pro'# Instantiate our context harness engine with a low token limit to force compaction visibilityctx_harness = ContextManager(client=client, model_id=model_id, max_context_tokens=8000)ctx_harness.append_message( role="user", parts=[types.Part.from_text(text="Fix all broken unit tests in the current workspace directory.")])tools_spec = [ types.Tool(function_declarations=[ types.FunctionDeclaration(name="read_file", description="Reads a file path", parameters={...}), types.FunctionDeclaration(name="run_bash_command", description="Runs a bash command", parameters={...}) ])]print("\n🚀 Starting the Harnessed Agent Loop (Layer 1 Active)...")step_count = 0while step_count < 10: step_count += 1 # Run our active pruning/compaction check before every single model turn ctx_harness.enforce_compaction() print(f"\n--- [TURN {step_count}] Model Inference ---") response = client.models.generate_content( model=model_id, contents=ctx_harness.history, config=types.GenerateContentConfig( system_instruction=ctx_harness.system_instruction, tools=tools_spec ) ) model_turn = response.candidates[0].content ctx_harness.append_message(role="model", parts=model_turn.parts) function_calls = response.candidates[0].function_calls if not function_calls: print("✅ Tasks finished.") break for call in function_calls: if call.name in TOOL_MAP: # Execute the raw tool tool driver raw_output = TOOL_MAP[call.name](**call.args) # Use our harness to safely intercept, measure, and filter the output string processed_output = ctx_harness.process_tool_output(tool_name=call.name, raw_output=raw_output) ctx_harness.append_message( role="tool", parts=[types.Part.from_function_response(name=call.name, response=processed_output)] )
- Head/Tail Truncation Mechanism
When a tool like run_bash_command("cat logs.txt") or an active test runner outputs thousands of lines, sending that raw text directly to the model is inefficient.
Our process_tool_output hook slices out the center rows of long strings. It preserves the top lines (which contain the execution parameters and initial entry headers) and the bottom lines (where tracebacks and error summaries live).
We compress a 200,000-token text dump into a clean, 800-token payload. The model receives the necessary diagnostic signals without crowding out its working memory.
- Protected Scoping (User Goal Lock)
Standard FIFO (First-In, First-Out) memory buffers are naive. If you discard older messages sequentially, you will eventually delete the user’s initial prompt and the high-level system rules. This causes the agent to lose its original direction and drift off-task.
Our compaction algorithm splits the history into 3 distinct architectural zones:
Anchored Root Zone: history[0] (The original objective) is permanently locked and exempt from compaction.
Compression Zone: history[1:-3] (The intermediate turns) is continuously compressed into high-level summary bullets.
Active Working Memory: history[-3:] (The most recent conversation turns) is kept fully expanded so the model retains immediate context for its current task steps.
- Managing “Context Anxiety”
As models approach their token limits, they can display behavioral shifts — such as rushing to declare a task “done” prematurely or skipping critical verification checks to save remaining context space.
By running enforce_compaction() prior to every inference call, the harness keeps active token utilization safely below its threshold. The model operates within a clean, optimized memory space, which avoids the performance drops associated with near-capacity context windows.
Let’s move from memory management to System Execution Environment (Layer 3 of our Taxonomy). A production agent cannot operate directly on your local machine’s bare metal; it needs a physical workspace, a secure boundary to prevent catastrophic system commands, and a mechanism to survive host crashes.
We will refactor our script to implement an isolated workspace using a unique project subdirectory, introduce execution sandboxing simulation, and build a durable JSON checkpoint system for crash resilience.
The following script extends our ContextManager loop with an infrastructure layer that establishes an isolated workspace directory, tracks changes, and serializes state to disk on every turn.
import osimport sysimport shutilimport jsonimport subprocessfrom google import genaifrom google.genai import types# =====================================================================# HARNESS INNER COMPONENT: Workspace Infrastructure & Sandbox Simulation# =====================================================================class AgentWorkspace: def __init__(self, workspace_root: str = "./agent_sandbox"): """ PRIMITIVE 1: The Filesystem as a Workspace Primitive Isolates all agent file mutations to a controlled, trackable directory tree. """ self.root = os.path.abspath(workspace_root) self.checkpoint_file = os.path.join(self.root, ".harness_state.json") self._initialize_workspace() def _initialize_workspace(self): """Prepares a secure execution environment and mimics container provisioning.""" if not os.path.exists(self.root): os.makedirs(self.root) print(f" [INFRA] Provisioned Isolated Workspace Directory: {self.root}") # Seed a dummy file to simulate broken environment tests for the agent test_file = os.path.join(self.root, "test_core.py") if not os.path.exists(test_file): with open(test_file, "w") as f: f.write("def test_addition():\n assert (1 + 1) == 3 # Intentionally broken unit test\n") def secure_path(self, relative_path: str) -> str: """ PRIMITIVE 2: Sandbox Path Jail (Path Traversal Prevention) Ensures the agent cannot use relative path tricks (../../etc/passwd) to escape. """ target_path = os.path.abspath(os.path.join(self.root, relative_path)) if not target_path.startswith(self.root): raise PermissionError(f"Security Violation: Agent attempted to escape sandbox boundary: {relative_path}") return target_path def run_isolated_bash(self, command: str) -> str: """ PRIMITIVE 2 (Cont.): Execution Isolation Box Executes instructions securely inside the designated workspace. """ print(f" [SANDBOX] Executing Command inside Workspace: '{command}'") # Hard Security Constraint: Intercept forbidden shell commands deterministically forbidden_tokens = ["rm -rf /", "chmod", "shutdown", "wget", "curl"] if any(token in command for token in forbidden_tokens): return "HARNESS SECURITY ERROR: Command blocked. Unauthorized system execution detected." try: # Force the execution context into the workspace root result = subprocess.run( command, shell=True, cwd=self.root, capture_output=True, text=True, timeout=15 # Mitigates infinite execution hangups ) return f"CWD: {self.root}\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" except subprocess.TimeoutExpired: return "HARNESS TIMEOUT ERROR: Process terminated automatically after 15 seconds." def save_checkpoint(self, step: int, history: list): """ PRIMITIVE 3: State Durability & Checkpoint System Dumps active session details to the disk to protect against runtime crashes. """ payload = { "last_executed_step": step, "conversation_history": history } with open(self.checkpoint_file, "w") as f: json.dump(payload, f, indent=2) print(f" [INFRA] Transaction saved to disk: step_{step}") def load_checkpoint(self) -> tuple[int, list] | None: """Attempts a seamless resume sequence if a crash or timeout occurred.""" if os.path.exists(self.checkpoint_file): try: with open(self.checkpoint_file, "r") as f: payload = json.load(f) print(f" 🚀 [INFRA] Valid state detected. Resuming operation from Step {payload['last_executed_step']}") return payload["last_executed_step"], payload["conversation_history"] except Exception: print(" [INFRA] Checkpoint corrupted, initializing fresh loop.") return None# =====================================================================# SYSTEM EXECUTION LAYER WITH INTEGRATED HARNESS# =====================================================================client = genai.Client()model_id = 'gemini-1.5-pro'# Spin up infrastructureinfra = AgentWorkspace("./agent_sandbox")# State Initialization Sequence (Checkpoint Restore vs. Cold Boot)saved_state = infra.load_checkpoint()if saved_state: start_step, conversation_history = saved_stateelse: start_step = 0 conversation_history = [ {"role": "user", "parts": [{"text": "Fix the broken test file 'test_core.py' in your current workspace directory."}]} ]# Bind high-level tool execution drivers straight to the infrastructure instancedef read_file_harness(path: str) -> str: safe_path = infra.secure_path(path) with open(safe_path, 'r', encoding='utf-8') as f: return f.read()def run_bash_harness(command: str) -> str: return infra.run_isolated_bash(command)TOOL_MAP = {"read_file": read_file_harness, "run_bash_command": run_bash_harness}# Tool Specs mappingtools_spec = [ types.Tool(function_declarations=[ types.FunctionDeclaration(name="read_file", description="Reads file text inside workspace", parameters={...}), types.FunctionDeclaration(name="run_bash_command", description="Runs shell tools inside sandbox", parameters={...}) ])]print("\n🚀 Starting the Infrastructure-Protected Agent Loop...")step_count = start_stepwhile step_count < 5: step_count += 1 print(f"\n--- [TURN {step_count}] Execution Iteration ---") try: response = client.models.generate_content( model=model_id, contents=conversation_history, config=types.GenerateContentConfig(tools=tools_spec) ) except Exception as api_err: print(f"❌ [CRASH] Network/API Error: {api_err}. Loop safely d.") sys.exit(1) # Simulated network drop; the state remains safely stored on disk model_turn = response.candidates[0].content conversation_history.append({"role": "model", "parts": model_turn.parts}) function_calls = response.candidates[0].function_calls if not function_calls: print("✅ Completion state reached cleanly.") # Clear out the operational state snapshot file upon a successful run if os.path.exists(infra.checkpoint_file): os.remove(infra.checkpoint_file) break for call in function_calls: if call.name in TOOL_MAP: # Secure tool execution via the workspace harness wrapper tool_output = TOOL_MAP[call.name](**call.args) conversation_history.append({ "role": "tool", "parts": [types.Part.from_function_response(name=call.name, response={"result": tool_output})] }) # Commit transaction to disk at the close of every successful turn infra.save_checkpoint(step_count, conversation_history)
- Filesystem as a Workspace Primitive In a simple script, an agent views files via arbitrary absolute paths like /var/log/app.log or ./src. In a hardened harness, we transform the filesystem into a jailsafe workspace primitive.
Every workspace path translation is verified by standardizing paths using os.path.abspath.
If the agent passes an argument designed to break out of the directory tree, the harness catches the path traversal attempt before it hits the operating system, returning a security restriction message to the model itself.
- Sandbox Isolation & Virtual Tools When an agent needs to execute code or write scripts, the harness executes these commands inside a strict Sandbox Box.
Our harness overrides the working directory parameter (cwd=self.root), ensuring that any command the model issues runs inside the sandbox area.
It enforces hard Process Timeouts. If the agent issues a command that hangs indefinitely (like a bad test runner or a standard loop bug), the harness forces a process shutdown at 15 seconds, records the timeout exception, and surfaces it to the model to handle.
It blocks high-risk command structures via regex strings or specific token allow-lists directly inside the execution gateway.
- State Durability (“Resume from Saved” Capability) In an un-harnessed runtime, an unexpected API disconnect or container restart drops all conversation history, forcing you to re-run the entire pipeline from scratch. Our harness addresses this with State Durability:
At the close of each conversation turn, the harness records the active state (conversation_history) to a non-volatile location (.harness_state.json).
If a network timeout or connection drop occurs at step 4, the session script halts safely.
Upon restarting, the harness reads the snapshot file on boot, reconstructs the conversation state, and resumes execution seamlessly from the exact step where it was interrupted.
Lets focus on the Governance and Risk Management Plane (Layer 3: Middleware and Constraints). An agent operating inside a secure container can still exhaust your budget, execute low-risk tools recklessly, or generate syntactically broken code. We will refactor our harness to enforce strict resource budgets, implement a multi-tiered security model that requires human approval for high-risk actions, and insert automated post-execution hooks (such as Python syntax linters) to verify all file modifications before the model proceeds.
import osimport sysimport jsonimport subprocessfrom google import genaifrom google.genai import types# =====================================================================# HARNESS INNER COMPONENT: Governance, Risk Tiers, & Quality Control# =====================================================================class GovernanceEngine: def __init__(self, max_steps=5, max_cost_usd=0.05): """ PRIMITIVE 1: Hard Governance Guardrails Enforces resource tracking and execution boundaries at the engine level. """ self.max_steps = max_steps self.max_cost_usd = max_cost_usd self.accumulated_cost_usd = 0.0 # Approximate pricing structures per 1M tokens for tracking calculations self.PRICE_INPUT_PER_M = 1.25 self.PRICE_OUTPUT_PER_M = 5.00 def track_and_validate_cost(self, step: int, input_tokens: int, output_tokens: int): """Monitors step limits and cumulative token costs during execution.""" turn_cost = ((input_tokens / 1000000) * self.PRICE_INPUT_PER_M) + \ ((output_tokens / 1000000) * self.PRICE_OUTPUT_PER_M) self.accumulated_cost_usd += turn_cost print(f" [GOV] Step: {step}/{self.max_steps} | Turn Cost: ${turn_cost:.5f} | Total: ${self.accumulated_cost_usd:.5f}") if step > self.max_steps: raise RuntimeError(f"HARNESS BUDGET EXCEEDED: Step limit of {self.max_steps} reached.") if self.accumulated_cost_usd > self.max_cost_usd: raise RuntimeError(f"HARNESS BUDGET EXCEEDED: Cost limit of ${self.max_cost_usd:.4f} breached.") def evaluate_risk_tier(self, tool_name: str, arguments: dict) -> bool: """ PRIMITIVE 2: Risk Tiering & Human-in-the-Loop Gating Classifies requested tool actions and enforces manual gates for high-risk items. """ # Tier 1: Read-Only Actions (Fully Autonomous) if tool_name in ["read_file", "view_status"]: return True # Tier 2: Draft/Staging Actions (Fully Autonomous within sandbox boundaries) if tool_name == "write_file" and not arguments.get("path", "").endswith(".sh"): return True # Tier 3: External Write / Highly Destructive Actions (Requires Manual Verification Gate) if tool_name == "run_bash_command" or (tool_name == "write_file" and arguments.get("path", "").endswith(".sh")): print(f"\n 🛑 [RISK GATE] Tier 3 Authorization Requested by Agent!") print(f" [ACTION]: {tool_name}(args={json.dumps(arguments)})") user_input = input(" [HUMAN APPROVAL REQUIRED] Approve execution? (yes/no): ").strip().lower() if user_input == "yes" or user_input == "y": print(" [RISK GATE] Action cleared by operator.") return True else: print(" [RISK GATE] Action rejected by operator.") return False return False def execute_post_write_hook(self, target_filepath: str) -> str | None: """ PRIMITIVE 3: Post-Execution Hooks & Automated Quality Assurance Runs automated syntax validation against modified files before returning control. """ if not target_filepath.endswith(".py"): return None # Only evaluate Python modules for this hook print(f" [HOOK] Intercepting file mutation. Compiling syntax for: {target_filepath}") try: with open(target_filepath, 'r') as f: source_code = f.read() # Programmatically compile the Python code code to catch syntax issues early compile(source_code, target_filepath, 'exec') print(" [HOOK] Syntax validation passed successfully.") return None except SyntaxError as syntax_err: print(f" ⚠️ [HOOK ALERT] Syntax error caught by validation filter: {syntax_err}") return ( f"HARNESS POST-WRITE VALIDATION FAILURE: Your last modification introduced a compile error.\n" f"Line: {syntax_err.lineno}, Offset: {syntax_err.offset}\n" f"Error details: {str(syntax_err)}" )# =====================================================================# SYSTEM TOOLS LAYER# =====================================================================WORKSPACE_DIR = "./agent_sandbox"os.makedirs(WORKSPACE_DIR, exist_ok=True)def write_file(path: str, content: str) -> str: full_path = os.path.join(WORKSPACE_DIR, path) with open(full_path, "w", encoding="utf-8") as f: f.write(content) return f"Success: File written completely to {path}."def run_bash_command(command: str) -> str: result = subprocess.run(command, shell=True, cwd=WORKSPACE_DIR, capture_output=True, text=True) return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"TOOL_MAP = {"write_file": write_file, "run_bash_command": run_bash_command}# =====================================================================# SYSTEM EXECUTION ENGINE# =====================================================================client = genai.Client()model_id = 'gemini-1.5-pro'gov = GovernanceEngine(max_steps=4, max_cost_usd=0.02)conversation_history = [ {"role": "user", "parts": [{"text": "Create a python file named 'app.py' that calculates factorials, but leave a syntax error in it to see what happens."}]}]tools_spec = [ types.Tool(function_declarations=[ types.FunctionDeclaration(name="write_file", description="Writes content to a file path", parameters={ "type": "OBJECT", "properties": {"path": {"type": "STRING"}, "content": {"type": "STRING"}}, "required": ["path", "content"] }), types.FunctionDeclaration(name="run_bash_command", description="Executes shell operations", parameters={ "type": "OBJECT", "properties": {"command": {"type": "STRING"}}, "required": ["command"] }) ])]print("\n🚀 Initializing Governance and Risk-Tier Controlled Loop...")step_count = 0while True: step_count += 1 print(f"\n--- [TURN {step_count}] Model Execution Cycle ---") # Calculate token volumes for cost and budget tracking metrics count_resp = client.models.count_tokens(model=model_id, contents=conversation_history) current_input_tokens = count_resp.total_tokens try: response = client.models.generate_content( model=model_id, contents=conversation_history, config=types.GenerateContentConfig(tools=tools_spec) ) # Apply strict cost and resource limits directly to the step sequence gov.track_and_validate_cost(step_count, current_input_tokens, response.usage_metadata.candidates_token_count) except Exception as budget_or_api_error: print(f"🛑 [LOOP TERMINATED BY HARNESS]: {str(budget_or_api_error)}") break model_turn = response.candidates[0].content conversation_history.append({"role": "model", "parts": model_turn.parts}) function_calls = response.candidates[0].function_calls if not function_calls: print("✅ The agent completed its execution loop cleanly.") break for call in function_calls: # Check risk tiers before allowing any tool execution to proceed is_authorized = gov.evaluate_risk_tier(call.name, call.args) if not is_authorized: tool_output = "HARNESS SECURITY REFUSAL: The operator denied permission to execute this tool step." else: if call.name in TOOL_MAP: # Execute the authorized tool call step safely tool_output = TOOL_MAP[call.name](**call.args) # Execute post-execution quality gates for file modifications if call.name == "write_file": hook_error = gov.execute_post_write_hook(os.path.join(WORKSPACE_DIR, call.args["path"])) if hook_error: # If validation fails, intercept the tool response and replace it with the compilation error details tool_output = hook_error else: tool_output = f"Error: Tool '{call.name}' not registered." conversation_history.append({ "role": "tool", "parts": [types.Part.from_function_response(name=call.name, response={"result": tool_output})] })
- Hard Governance Guardrails In standard software architectures, compute limits are enforced via memory allocations or thread controls. In agent systems, we manage execution using Financial and Step-Count Budgets.
The harness maps token counts across every API interaction using precise usage data.
If a model becomes caught in an infinite loop, the cumulative cost or total step count breaches your predefined budget limits. The harness halts execution immediately, protecting you from unexpected cloud bills.
- Risk Tiering & Human-in-the-Loop (HITL) Gating Instead of choosing between complete automation or micromanaged approvals, the harness implements a Multi-Level Permission Model:
Tier 1 (Read-Only): Low-risk operations (like checking file directories or parsing read-only code logs) require no manual intervention and execute with full autonomy.
Tier 2 (Staging / Internal Draft): Changes restricted to the sandboxed workspace folder (like writing local python modules) execute automatically.
Tier 3 (External Write / System Access): High-impact actions (like executing shell commands or modifying deployment scripts) hit a mandatory permission interceptor. The harness s execution, alerts the developer, and waits for a manual validation signature before running the command.
- Post-Execution Validation Hooks To keep the agent from proceeding with broken code or syntax errors, the harness applies deterministic validation filters to all changes:
When the agent tries to update a system module, the harness catches the modification using a post-write interceptor.
It passes the updated file through a programmatic test engine, compiler, or language syntax check.
If a compilation issue or syntax error is found, the harness intercepts the success response from the tool and swaps it for the raw compilation trace. This surfaces the error back to the model immediately, prompting it to fix its changes before moving forward in the workflow.
Harness Layer 4 — Telemetry & Feedback Layer
Let’s focus on System Observability (Layer 4 of our Taxonomy). When an autonomous agent executes 50 turns over 10 minutes and fails, reading a flat text file of the terminal log becomes completely unmanageable. We need structured instrumentation. We will refactor our script to implement structured hierarchical tracing, build an analytical engine to track real-time metric drift, and formalize an automated pipeline that extracts production failure trajectories and structures them into fine-tuning datasets.
import osimport jsonimport uuidimport timefrom typing import List, Dict, Anyfrom google import genaifrom google.genai import types# =====================================================================# HARNESS INNER COMPONENT: Telemetry, Span Tracer, & Dataset Generator# =====================================================================class AgentTelemetryHarness: def __init__(self, trace_dir: str = "./agent_telemetry"): self.trace_dir = trace_dir os.makedirs(self.trace_dir, exist_ok=True) self.root_trace_id = str(uuid.uuid4()) self.spans: List[Dict[str, Any]] = [] # Performance Tracking State Variables self.trajectory_scores: List[float] = [] def start_span(self, name: str, parent_id: str = None) -> str: """Creates a structured execution span to track step performance.""" span_id = str(uuid.uuid4()) span = { "span_id": span_id, "parent_id": parent_id or self.root_trace_id, "name": name, "start_time": time.time(), "end_time": None, "metadata": {} } self.spans.append(span) return span_id def end_span(self, span_id: str, metadata: Dict[str, Any] = None): """Finalizes a span record and attaches performance metadata.""" for span in self.spans: if span["span_id"] == span_id: span["end_time"] = time.time() if metadata: span["metadata"].update(metadata) break def track_realtime_drift(self, turn_idx: int, response_text: str, function_calls: Any): """ PRIMITIVE 2: Real-Time Performance Drift Detection Calculates simple trajectory metrics to flag when performance begins to degrade. """ # Feature 1: Check for repetitive tool usage (indicates looping behaviors) repeated_action_penalty = 0.0 if function_calls and len(self.spans) > 2: last_actions = [s.get("metadata", {}).get("tool_requested") for s in self.spans if "tool" in s["name"]][-3:] if len(last_actions) >= 2 and len(set(last_actions)) == 1: repeated_action_penalty = 0.4 # Deduct points if the agent calls the same tool repeatedly # Feature 2: Check text output length relative to action density # Excessively long text combined with zero action signals a drop into conversational patterns word_count = len(response_text.split()) verbosity_score = 1.0 if word_count < 150 else max(0.2, 1.0 - (word_count / 500)) # Combine metrics into a composite Trajectory Score turn_fidelity = max(0.0, 1.0 - repeated_action_penalty) * verbosity_score self.trajectory_scores.append(turn_fidelity) # Calculate a rolling average across the last three execution turns rolling_avg = sum(self.trajectory_scores[-3:]) / len(self.trajectory_scores[-3:]) print(f" [TELEMETRY] Turn {turn_idx} Trajectory Score: {turn_fidelity:.2f} | Rolling Fidelity Avg: {rolling_avg:.2f}") if rolling_avg < 0.4: print(" ⚠️ [TELEMETRY DRIFT WARNING] Trajectory degradation detected. The model may be caught in a loop or losing task focus.") def export_trajectory_as_dataset(self, final_success: bool): """ PRIMITIVE 3: Trajectories as Fine-Tuning Datasets Transforms execution history into structured data objects for future tuning. """ export_file = os.path.join(self.trace_dir, f"trajectory_{self.root_trace_id}.json") # Re-structure spans into an orderly call-and-response execution flow dataset_entry = { "trace_id": self.root_trace_id, "timestamp": time.time(), "outcome_verified_success": final_success, "rolling_fidelity_trend": self.trajectory_scores, "tuning_conversational_format": [] } # Extract conversational milestones for fine-tuning formats for s in self.spans: if s["name"] == "model_inference" and "prompt_tokens" in s["metadata"]: dataset_entry["tuning_conversational_format"].append({ "prompt_snapshot": s["metadata"].get("input_snapshot"), "ideal_response_output": s["metadata"].get("output_snapshot") }) with open(export_file, "w") as f: json.dump(dataset_entry, f, indent=2) print(f" [TELEMETRY] Complete trajectory trace successfully saved to disk: {export_file}")# =====================================================================# SYSTEM EXECUTOR LAYER WITH INTEGRATED TELEMETRY HOOKS# =====================================================================client = genai.Client()model_id = 'gemini-1.5-pro'telemetry = AgentTelemetryHarness()# Build baseline interaction layoutconversation_history = [ {"role": "user", "parts": [{"text": "Inspect the project environment and verify if standard tool packages are ready."}]}]def mock_read_env_tool() -> str: time.sleep(0.1) # Simulate tool response time latency return "SUCCESS: Environment profile initialized. Packages locked."TOOL_MAP = {"check_env": mock_read_env_tool}tools_spec = [types.Tool(function_declarations=[ types.FunctionDeclaration(name="check_env", description="Checks workspace environment settings", parameters={"type": "OBJECT", "properties": {}})])]print("\n🚀 Starting the Telemetry-Instrumented Agent Loop...")step_count = 0is_completed_successfully = Falsewhile step_count < 3: step_count += 1 print(f"\n--- [TURN {step_count}] Traced Execution Iteration ---") # 1. Start tracking the Context Assembly span span_ctx = telemetry.start_span("context_assembly") active_history_snapshot = json.dumps(conversation_history) telemetry.end_span(span_ctx, {"history_size_bytes": len(active_history_snapshot)}) # 2. Start tracking the Model Inference span span_inf = telemetry.start_span("model_inference") response = client.models.generate_content( model=model_id, contents=conversation_history, config=types.GenerateContentConfig(tools=tools_spec) ) model_turn = response.candidates[0].content model_text = model_turn.parts[0].text if model_turn.parts and hasattr(model_turn.parts[0], 'text') else "" function_calls = response.candidates[0].function_calls # Attach precise metadata snapshots to the active inference span log telemetry.end_span(span_inf, { "prompt_tokens": response.usage_metadata.prompt_token_count, "candidates_tokens": response.usage_metadata.candidates_token_count, "input_snapshot": active_history_snapshot, "output_snapshot": json.dumps({"text": model_text, "calls": str(function_calls)}) }) conversation_history.append({"role": "model", "parts": model_turn.parts}) # Calculate performance metrics and check for rolling drift telemetry.track_realtime_drift(step_count, model_text, function_calls) if not function_calls: print("✅ Task finished.") is_completed_successfully = True break for call in function_calls: # 3. Start tracking the Tool Execution span span_tool = telemetry.start_span(f"tool_execution_{call.name}") if call.name in TOOL_MAP: tool_output = TOOL_MAP[call.name]() else: tool_output = f"Error: Tool '{call.name}' not found." telemetry.end_span(span_tool, {"tool_requested": call.name, "output_preview": tool_output[:60]}) conversation_history.append({ "role": "tool", "parts": [types.Part.from_function_response(name=call.name, response={"result": tool_output})] })# Commit full trajectory records to disk upon loop terminationtelemetry.export_trajectory_as_dataset(final_success=is_completed_successfully)
- Standardized Hierarchical Tracing In traditional systems engineering, we use flat application logs (info, error) to monitor health. For autonomous agents, we require Hierarchical Spans to map the step-by-step logic of a multi-turn run:
Context Assembly Span: Logs the exact composition of the conversation history, active prompt structures, and token footprints before the model is called.
Model Inference Span: Records latency, token usage costs, raw text thoughts, and any tool invocation requests.
Tool Execution Span: Tracks the argument payload parsing performance, backend infrastructure latency, and raw response strings.
By nesting these spans together under a unified root_trace_id, you can trace any system bug directly to its source (Ex: Turn 12 →Inference Step -> Shell Tool Command Exception).
- Real-Time Drift Detection Agents can drift into unhelpful behaviors before they hit hard limits like budget caps or timeouts. The telemetry layer captures these trends early using rolling metrics:
Repetition Penalty Tracker: If the harness catches the agent calling the exact same tool with matching arguments back-to-back, it flags a loop vulnerability.
Verbosities Engine: If the model’s text volume spikes while its actual tool invocations drop, it signals that the agent is stalling or falling into conversational patterns rather than executing the task.
The harness evaluates these behaviors to generate a live Trajectory Fidelity Score. If this score drops below your target threshold, the system flags the issue immediately, allowing you to intercept or reset the session before wasting further API spend.
- Trajectories as High-Value Datasets A common challenge in production AI engineering is sourcing high-quality training and evaluation datasets. The harness turns production telemetry into data infrastructure:
Every run maps its internal spans into structured interaction logs (trajectory_[UUID].json).
Successful runs are tagged as Perfect Gold Demonstrations, which can be fed directly into your evaluation suites as baseline benchmarks.
Failed runs are isolated and flagged for root-cause analysis. Once fixed via a harness ratchet or improved system instruction, the corrected trajectory is transformed into a high-value data entry for future model fine-tuning and reinforcement learning pipelines.
Loop Engineering (Multi-Session Factory Floor)
Till now, we built a comprehensive, single-session Agent Harness. We gave our agent an optimized workspace memory (Layer 1), a sandboxed filesystem environment (Layer 2), strict cost and human approval gates (Layer 3), and hierarchical span telemetry (Layer 4).
Now, we scale our architecture up to the Orchestration Plane: Loop Engineering. We will move beyond managing a single continuous chat window and look at how to coordinate multiple separate agent sessions over long-running engineering lifecycles.
Why Single-Session Harnesses Hit a Wall
Using our Automated Manufacturing Plant analogy, let’s look at what we have achieved so far and where the single-session model falls short.
What we achieved with Harness Engineering: We built an exceptionally reliable Machine Rig. The artisan worker (the model) stands inside a safe, instrumented cage. They cannot delete the factory floor, they cannot waste an infinite amount of material without a budget trigger, and automated sensors instantly check their work for structural defects.
Imagine assigning this single machine rig a massive, open-ended task, such as: “Refactor the inventory system, update all deprecated API endpoints across 50 modules, and verify everything passes production checks.” If you try to complete this entire multi-day project inside a single continuous agent session, the system encounters three critical failure modes:
Context Window Saturation Ceiling: No matter how efficient your compaction engine is, a 50-turn interaction eventually fills up with code snippets, diff arrays, and compiler outputs. The model’s reasoning performance drops, costs rise, and it struggles to maintain global task focus.
Self-Evaluation Blindspot: When the same artisan worker who wrote a piece of code is also asked, “Is this code correct?” they suffer from confirmation bias. The model will look at its own buggy code, fail to spot the logic flaw, and confidently declare the task complete.
Workspace Locking: A single agent session operates linearly. It cannot scale out to work on five independent sub-modules concurrently without overwriting its own files and running into severe state conflicts.
How Loop Engineering resolves these Issues
Loop Engineering solves these limitations by treating the agent session as a transient, short-lived utility. Instead of running one massive conversation, Loop Engineering spins up, tears down, and orchestrates dozens of distinct, highly specialized harnessed sessions over time.
It shifts the primary source of truth out of the model’s volatile chat history and moves it directly into the physical workspace — using tracking files (like PROGRESS.md), git commits, and isolated testing sandboxes as the persistent memory layer.
- Recursive Goal Primitive (/goal)
Traditional LLM interactions rely on command-driven execution: you provide an instruction, and the model attempts to generate a direct response. Loop Engineering shifts this dynamic to an objective-driven loop using a recursive goal primitive.
Instead of passing a literal execution script, the loop controller initializes a structured goal state on disk:
{ "goal_id": "goal_refactor_factorial_001", "objective": "Migrate the factorial calculator engine to use iterative memoization.", "status": "IN_PROGRESS", "stopping_conditions": [ "app.py compiles without syntax errors", "pytest tests/test_math.py returns exit code 0", "Execution time for factorial(1000) is under 2ms" ]}
The loop controller sets up an automated routine that runs continuously:
The agent is not allowed to simply say, “I am done.” The loop controller retains ownership of the session. It executes the agent, inspects the physical workspace against the hard stopping conditions, and recursively spins up follow-up sub-sessions until every single validation check returns a clean pass.
2 Maker/Checker Split (Planner vs. Evaluator)
To eliminate self-evaluation bias, Loop Engineering splits execution responsibilities across 2 entirely separate agent personas: Maker (Generator) and Checker (Evaluator).
Maker Agent: This session is provisioned with write tools (write_file, apply_patch, run_bash_command). Its sole responsibility is to modify code and get it compiling.
Checker Agent:This session is completely stateless and has zero write permissions. It is spun up in a fresh context window, entirely unaware of the Maker’s internal reasoning steps or intermediate failures. It is handed only two things: the original goal criteria and the resulting workspace files.
Because the Checker has no historical attachment to the modifications, it evaluates the workspace with a clean slate. If it finds a failing edge case, it writes a detailed bug report. The loop controller captures this report and feeds it back as an input trace to the Maker for the next iteration.
3 Ralph Loop & Multi-Step Orchestration
Named after the pattern of iterative self-correction, a Ralph Loop formalizes how multi-step orchestration handles task transitions without human oversight. When a large objective is initiated, the loop controller coordinates a sequence of distinct, short-lived sessions:
Step 1: Architect Turn:A specialized planning session reads the repository structure and generates a PLAN.md file. The session then terminates immediately, clearing its context window.
Step 2: Developer Execution: A fresh Maker session is initialized. It reads the PLAN.md, applies the code updates across the files, runs its local harness linters, logs its achievements in a PROGRESS.md file, and shuts down.
Step 3: Automated Evaluation:A separate Checker session boots up, reads the PROGRESS.md, runs the integration test suite, and verifies the requirements.
If the Checker detects a failure, the loop controller routes the workflow back to Step 2 with a fresh context window. The code architecture maintains continuous momentum because the state transitions are safely written to disk at the close of each step.
4 Sprint Contract & Worktrees
Before any code is altered, the loop controller establishes an explicit Sprint Contract between the components. Maker and Checker must programmatically agree on the “Definition of Done” criteria before execution starts. This contract is written to a shared config file within the environment.
Parallel Isolation via Git Worktrees When an enterprise loop needs to process multiple sub-tasks concurrently, it uses Git Worktrees to maintain clean isolation.
┌──► Worktree Alpha (/sandbox/task_a) ──► Maker/Checker Loop A │Main Repository ──┼──► Worktree Beta (/sandbox/task_b) ──► Maker/Checker Loop B │ └──► Worktree Gamma (/sandbox/task_c) ──► Maker/Checker Loop C
Instead of allowing multiple sub-agents to make conflicting changes within a single shared directory, the loop controller issues a system command:
git worktree add ../sandbox/task_feature_b feature_branch_b
The harness places each agent loop inside its own separate, physical filesystem directory. The agents run their test suites and modify their respective modules in complete isolation. Once a task loop achieves a verified pass state, its worktree branch is merged back into the main repository branch via a standard pull-request validation pipeline.
Production Implementation: Loop Controller
Below is a complete implementation of a Loop Engineering Control Plane. It wraps our previous harness concepts, coordinates a separate Maker/Checker workflow, and tracks progress across distinct execution windows.
import osimport jsonimport subprocessfrom google import genaifrom google.genai import typesWORKSPACE_DIR = "./loop_factory"os.makedirs(WORKSPACE_DIR, exist_ok=True)# Seed a sample application file and a test suite filewith open(f"{WORKSPACE_DIR}/app.py", "w") as f: f.write("def calculate_factorial(n):\n # Core Bug: returns incorrect calculation\n return 0\n")with open(f"{WORKSPACE_DIR}/test_app.py", "w") as f: f.write("from app import calculate_factorial\ndef test_factorial():\n assert calculate_factorial(3) == 6\n")# =====================================================================# LOOP CONTROL PLANE: Maker / Checker Orchestrator# =====================================================================class LoopOrchestrator: def __init__(self, workspace: str): self.workspace = workspace self.client = genai.Client() self.model_id = 'gemini-1.5-pro' # The Sprint Contract definition of done self.contract_criteria = "test_app.py must pass successfully via pytest execution." def run_local_tests(self) -> tuple[bool, str]: """Deterministic verification gate evaluating the Sprint Contract.""" result = subprocess.run( "pytest test_app.py", shell=True, cwd=self.workspace, capture_output=True, text=True ) passed = result.returncode == 0 output = f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" return passed, output def invoke_maker_session(self, feedback: str) -> str: """Spawns a transient Maker session with write access to fix issues.""" print(" [LOOP] Launching specialized MAKER agent session...") # Read the current file contents to populate the context window with open(f"{self.workspace}/app.py", "r") as f: current_code = f.read() prompt = ( f"OBECTIVE: Fix app.py to satisfy this contract: {self.contract_criteria}\n" f"CURRENT CODE:\n{current_code}\n\n" f"PREVIOUS EVALUATOR FEEDBACK:\n{feedback}\n\n" f"Task: Rewrite app.py completely to resolve the bug. Return ONLY valid Python code inside markdown blocks." ) response = self.client.models.generate_content(model=self.model_id, contents=prompt) return response.text def invoke_checker_session(self, test_logs: str) -> tuple[bool, str]: """Spawns an isolated Checker session with zero write access to verify changes.""" print(" [LOOP] Launching stateless CHECKER agent session...") with open(f"{self.workspace}/app.py", "r") as f: updated_code = f.read() prompt = ( f"You are an independent quality control evaluator. Verify if this code satisfies the criteria.\n" f"CRITERIA: {self.contract_criteria}\n" f"PROPOSED CODE:\n{updated_code}\n\n" f"EXECUTION TEST LOGS:\n{test_logs}\n\n" f"If the tests failed, analyze the logs and output detailed debugging hints for the developer.\n" f"If everything passed perfectly, output exactly one token: 'PASSED'." ) response = self.client.models.generate_content(model=self.model_id, contents=prompt) verdict = response.text.strip() if "PASSED" in verdict: return True, "Verification successful." return False, f"Evaluator Rejection Report:\n{verdict}" def execute_goal(self): """The Recursive Goal Loop Control Routine.""" print(f"🚀 Initializing Loop Control Plane for Objective Goal...") max_loops = 3 loop_count = 0 current_feedback = "Initial Run: Code contains a placeholder defect." while loop_count < max_loops: loop_count += 1 print(f"\n🔄 --- [LOOP RUN {loop_count}] ---") # Step 1: Execute the Maker session to generate code changes maker_output = self.invoke_maker_session(current_feedback) # Parse out the updated Python source code from the model's text response if "Real-World Case Studies: How the Giants build Harnesses
In production applications, prompt tuning is a minor detail. Top engineering teams focus on building highly specialized Agent Execution Harnesses that treat the model as a transient compute engine.
Here is how 4 leading engineering teams — Anthropic (Claude Code), OpenAI (Codex), Vercel and Manus — implement harness and loop engineering patterns in production.
1. Anthropic (Claude Code)
Anthropic’s recent engineering documentation, Effective Harnesses for Long-Running Agents, highlights a clear production lesson: minimize complex multi-agent overhead where a single-threaded loop can do the job.
Implementation Strategy: Claude Code operates directly within the developer’s local terminal interface. Instead of spinning up an intricate network of planner and worker agents, Anthropic builds a high-leverage, single-threaded execution loop.
Production Trick: They emphasize providing rich, machine-readable developer utilities directly into the agent’s context window (such as specialized grep, find, and standard git diff wrappers). If a task fails or encounters an exception, the harness does not intercept or hide the failure; it feeds the raw, unedited compiler error or stack trace directly back into the conversation history. This builds an immediate feedback loop that allows the model to spot its mistakes and self-correct on the next turn.
2. OpenAI (Codex)
In early 2026, OpenAI’s Codex engineering team published Harness Engineering: Leveraging Codex in an Agent-First World. They detailed how they used a rigid harness to ship over one million lines of production code, written entirely by autonomous agents, in just five months.
Implementation Strategy (App Server): OpenAI abstracts the entire Codex agent loop away from the presentation layers (Web, CLI, IDE extensions) using a dedicated background process called the Codex App Server. This server communicates with client interfaces over a bidirectional, long-lived JSON-RPC protocol.
Production Trick: OpenAI discovered that a monolithic documentation file (like a massive AGENTS.md) quickly rots and causes context flooding. They downsized their global instructions to a lean 100-line map that points to a structured docs/ directory. Crucially, they enforce a strict one-way layered architecture (Types → Config → Repo → Service → Runtime → UI). Custom linters mechanically check every agent edit. If an agent tries to import a component upward or violate the design layout, the linter blocks the write operation and injects a remediation instruction directly into the tool response.
3. Vercel
Vercel’s engineering team discovered that giving an agent large, open-ended tools (such as broad shell execution engines) frequently causes performance drift and unexpected errors.
Implementation Strategy: Within platforms like v0, Vercel structures interactions using highly specific, isolated micro-tools (such as a single component renderer, a dedicated package installer, or a strict CSS linter).
Production Trick (Build to Delete): Vercel famously removed nearly 80% of their complex agent tools and discovered that task success rates actually went up. They design their agent harnesses with a “Build to Delete” philosophy: they write lean, modular scaffolding that is easy to remove or replace. As the underlying foundation models get smarter, the harness code can be thinned out, shifting the operational weight back onto the model’s native capabilities without breaking the product architecture.
4. Manus
Manus engineered its framework to tackle complex, multi-hour developer trajectories without stalling out or falling into recursive loops.
Implementation Strategy: Manus structures long-running trajectories as versioned, first-class file artifacts on disk, separating code production from task planning.
Production Trick: Instead of relying on a static, upfront plan, Manus creates lightweight, ephemeral execution plans inside a dedicated workspace directory (docs/exec-plans/active/). The agent maintains progress logs and records active engineering assumptions directly within these files. When a context window fills up or an execution turn crashes, a fresh agent instance boots up, reads the markdown status files, and picks up exactly where the previous instance left off.
Thank you so much for reading all the way through! If this deep dive into Harness and Loop Engineering helped you level up your AI agent architecture, please drop a few claps 👏 and leave a comment below with your thoughts — I’d love to hear how you’re building in production. What topics or frameworks should we tackle next? Catch you in the next article — until then, keep building!