Abstract:
Moving from theoretical taxonomy to production authority requires deep practical fluency in harness source code. This article provides a production-grade reference implementation of an H2/H3 Agent Harness adhering to the ETCLOVG framework. It then conducts a deep source-code anatomy of four leading production harnesses: Anthropic's Claude Code & Claude Agent SDK, OpenAI's Codex App Server, All-Hands AI's OpenHands, and LangGraph / Deep Agents StateGraph patterns. Finally, it outlines the progressive enrichment pipeline from a Minimum-Viable Harness (MVH) to an industrial Software Factory.
Below is a complete, modular reference architecture in Python demonstrating how the seven layers of ETCLOVG integrate into a cohesive runtime substrate:
"""
reference_harness.py
Production-grade H2/H3 Reference Harness implementing ETCLOVG.
"""
from dataclasses import dataclass, field
from enum import Enum
import json
import os
import subprocess
import time
from typing import Any, Callable, Dict, List, Optional
# ============================================================================
# 1. PERMISSIONS & GOVERNANCE LAYER (G)
# ============================================================================
class PermissionDecision(Enum):
ALLOW = "ALLOW"
ASK = "ASK"
DENY = "DENY"
@dataclass
class GovernancePolicy:
max_budget_usd: float = 5.0
current_spend_usd: float = 0.0
entropy_churn_limit_pct: float = 0.15 # Max 15% repo edit per turn
banned_commands: List[str] = field(default_factory=lambda: [
"rm -rf /", "dd if=", ":(){ :|:& };:", "chmod -R 777"
])
allowed_tools: List[str] = field(default_factory=lambda: [
"read_file_slice", "run_ast_linter", "run_scoped_tests", "git_diff"
])
def evaluate_action(self, tool_name: str, payload: Dict[str, Any]) -> PermissionDecision:
if self.current_spend_usd >= self.max_budget_usd:
return PermissionDecision.DENY
if tool_name == "bash_exec":
cmd = payload.get("command", "")
if any(banned in cmd for banned in self.banned_commands):
return PermissionDecision.DENY
if "git push" in cmd or "npm publish" in cmd:
return PermissionDecision.ASK
return PermissionDecision.ALLOW
if tool_name in self.allowed_tools:
return PermissionDecision.ALLOW
return PermissionDecision.ASK
# ============================================================================
# 2. EXECUTION & SANDBOX LAYER (E)
# ============================================================================
class SandboxedRuntime:
def __init__(self, workspace_path: str, timeout_sec: int = 30):
self.workspace_path = workspace_path
self.timeout_sec = timeout_sec
def execute_supervised_command(self, cmd: str) -> Dict[str, Any]:
"""
Executes a shell command with non-interactive env guards,
pager suppression, and hard timeout deadlines.
"""
safe_env = os.environ.copy()
safe_env.update({
"PAGER": "cat",
"GIT_PAGER": "cat",
"CI": "true",
"TERM": "dumb",
"DEBIAN_FRONTEND": "noninteractive"
})
start_time = time.time()
try:
result = subprocess.run(
cmd,
shell=True,
cwd=self.workspace_path,
env=safe_env,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
timeout=self.timeout_sec
)
duration = time.time() - start_time
return {
"exit_code": result.returncode,
"stdout": result.stdout,
"duration_sec": duration,
"timed_out": False
}
except subprocess.TimeoutExpired:
return {
"exit_code": 124,
"stdout": f"Execution Timed Out ({self.timeout_sec}s threshold exceeded)",
"duration_sec": self.timeout_sec,
"timed_out": True
}
# ============================================================================
# 3. CONTEXT & MEMORY LAYER (C)
# ============================================================================
class ContextWindowManager:
def __init__(self, max_tokens: int = 64_000):
self.max_tokens = max_tokens
self.history: List[Dict[str, Any]] = []
def compile_window(self, system_anchor: str, current_diff: str) -> List[Dict[str, Any]]:
"""
Implements Agentic Context Management (ACM):
Compacts intermediate debugging turns, preserving only invariants & diffs.
"""
tokens_estimate = len(system_anchor) // 4 + len(current_diff) // 4
compacted_history = []
# Retain only the most recent 3 turns if approaching budget
for turn in reversed(self.history[-3:]):
turn_tokens = len(json.dumps(turn)) // 4
if tokens_estimate + turn_tokens > self.max_tokens:
break
compacted_history.insert(0, turn)
tokens_estimate += turn_tokens
return [
{"role": "system", "content": system_anchor},
{"role": "user", "content": f"Repository State Diff:\n{current_diff}"},
*compacted_history
]
# ============================================================================
# 4. SENSORS & VERIFICATION LAYER (V)
# ============================================================================
@dataclass
class SensorReport:
passed: bool
failing_sensor: Optional[str] = None
structured_diagnostic: Optional[Dict[str, Any]] = None
class VerificationSensorCascade:
def __init__(self, runtime: SandboxedRuntime):
self.runtime = runtime
def run_cascade(self, modified_files: List[str]) -> SensorReport:
# Tier 1: Sub-100ms AST / Syntax Check
linter_res = self.runtime.execute_supervised_command("ruff check . --output-format=json")
if linter_res["exit_code"] != 0:
return SensorReport(
passed=False,
failing_sensor="AST_Linter",
structured_diagnostic={"type": "LINT_ERROR", "output": linter_res["stdout"]}
)
# Tier 2: Sub-2s Type Verification
type_res = self.runtime.execute_supervised_command("mypy . --no-error-summary")
if type_res["exit_code"] != 0:
return SensorReport(
passed=False,
failing_sensor="Type_Checker",
structured_diagnostic={"type": "TYPE_ERROR", "output": type_res["stdout"]}
)
# Tier 3: Sub-5s Scoped Unit Tests (Test Impact Analysis)
test_targets = " ".join([f"tests/test_{os.path.basename(f)}" for f in modified_files if f.endswith(".py")])
if test_targets:
test_res = self.runtime.execute_supervised_command(f"pytest {test_targets} -q")
if test_res["exit_code"] != 0:
return SensorReport(
passed=False,
failing_sensor="Test_Impact_Analysis",
structured_diagnostic={"type": "TEST_FAILURE", "output": test_res["stdout"]}
)
return SensorReport(passed=True)
# ============================================================================
# 5. OBSERVABILITY, ATTRIBUTION & LIFECYCLE (L, O)
# ============================================================================
class EpisodePackage:
def __init__(self, task_id: str):
self.task_id = task_id
self.trace_id = f"trace-{int(time.time())}"
self.events: List[Dict[str, Any]] = []
def record_turn(self, turn_idx: int, model_output: str, sensor_report: SensorReport):
self.events.append({
"turn": turn_idx,
"timestamp": time.time(),
"model_action": model_output,
"sensor_outcome": {
"passed": sensor_report.passed,
"diagnostic": sensor_report.structured_diagnostic
}
})
def export_package(self) -> Dict[str, Any]:
return {
"task_id": self.task_id,
"trace_id": self.trace_id,
"total_turns": len(self.events),
"events": self.events,
"audit_hash": hash(json.dumps(self.events))
}
view_file with explicit StartLine and EndLine slices, preserving the optimal 32k–64k reasoning zone.PAGER=cat and CI=true across all child sub-processes to prevent hanging on interactive terminal prompts.In an enterprise harness, exposing 50 tools simultaneously overwhelms the model's tool-selection attention. Production harnesses implement Skills Registries:
SKILLS REGISTRY DISCOVERY PIPELINE
Task Objective ──► ┌────────────────────────────────────────────────────────┐
│ SKILLS REGISTRY GATEWAY │
│ • Evaluates task semantic similarity │
│ • Dynamically binds 2-4 relevant skills │
└──────────────────────────┬─────────────────────────────┘
│
▼ Injected Tools
┌───────────────────────────────┐
│ ACTIVE MODEL RUN │
│ • Core Tools (Read, Edit, Bash│
│ • Bound Skills (e.g., Docker, │
│ Kubernetes, Database-Admin) │
└───────────────────────────────┘
git-tools, db-migrate-tools, aws-infra-tools).We haven't launched yet. But leave your email to get notified about exciting things that we will launch.