HarnessAgent

Production-grade, multi-tenant agent harness for building, running, observing, and self-improving AI agents. HarnessAgent wraps any agent framework (LangGraph, CrewAI, AutoGen) or can run native SQL/Code agents with a full production lifecycle: paged context management, semantic memory, tool safety, distributed tracing, and automated self-improvement.

What it gives you

Architecture

HarnessAgent is structured as a layered service. External requests enter through the FastAPI layer and pass through the orchestration layer to individual agents, which interact with memory, tools, and observability systems.

Quick Start

Install

pip install harnessagent[vector,observe,mcp]
# Or with all extras:
pip install harnessagent[all]bash

Run a single native agent

from harness.core.context import AgentContext
from harness.agents.sql_agent import SQLAgent
from harness.memory.manager import MemoryManager
from harness.llm.router import LLMRouter
from pathlib import Path

# Build components
memory = await MemoryManager.create(config)
llm    = LLMRouter()
llm.register(anthropic_provider, priority=0)

agent = SQLAgent(
    llm_router=llm,
    memory_manager=memory,
    tool_registry=tool_registry,
    safety_pipeline=None,
    step_tracer=None,
    mlflow_tracer=None,
    failure_tracker=failure_tracker,
    audit_logger=audit_logger,
    event_bus=event_bus,
    cost_tracker=cost_tracker,
    checkpoint_manager=checkpoint_manager,
    trace_recorder=trace_recorder,   # ← new: durable span tree
)

ctx = AgentContext.create(
    tenant_id="acme",
    agent_type="sql",
    task="List all tables and their row counts",
    memory=memory,
    workspace_path=Path("/workspaces/acme/run1"),
)

result = await agent.run(ctx)
print(result.output, result.cost_usd, result.steps)python

Wrap an existing LangGraph agent

import harness
from langgraph.graph import StateGraph

graph = StateGraph(...)  # your existing graph
adapter = harness.wrap(graph)
adapter.attach_harness(
    safety_pipeline=my_pipeline,
    cost_tracker=cost_tracker,
    audit_logger=audit_logger,
)

async for event in adapter.run_with_harness(ctx, {"input": "analyze sales data"}):
    print(event.event_type, event.payload)python

Start the API server

uvicorn harness.api.main:create_app --factory --host 0.0.0.0 --port 8000
# Or via Makefile:
make apibash

Configuration

All settings are loaded from environment variables or a .env file via harness.core.config.Settings (Pydantic BaseSettings). Access the singleton via get_config().

Core — Context & Events

Defined in harness/core/context.py. These are the fundamental data objects that flow through every part of the system.

AgentContext

Mutable per-run state shared across all components. Created once per run and passed everywhere.

StepEvent

Emitted at each significant moment in an agent run. Published to EventBus (Redis Pub/Sub) for SSE streaming.

Core — Protocols

Defined in harness/core/protocols.py. Structural Protocol ABCs that all pluggable components must satisfy. Use isinstance(x, LLMProvider) for runtime checking.

Core — Error Hierarchy

All harness errors extend HarnessError and carry a canonical FailureClass enum value. The failure class drives HTTP status codes, Prometheus labels, and Hermes sampling.

BaseAgent — Run Lifecycle

Defined in harness/agents/base.py. All concrete agents inherit from BaseAgent. The run(ctx) method orchestrates the full production lifecycle.

Constructor Parameters

Run loop steps

1. Open RUN trace span, emit started event, begin MLflow run
2. Resume from checkpoint if one exists (_maybe_resume_checkpoint)
3. Loop while ctx.is_budget_ok():
   1. Fit history to context window (compress if > 40 messages)
   2. GraphRAG/vector retrieve — build retrieval context string
   3. Build LLM messages + system prompt
   4. Open LLM trace span → call LLM → register token usage → close span
   5. Open GUARDRAIL span → run safety pipeline → close span
   6. If no tool calls → extract final answer → break
   7. HITL approval check (sequential)
   8. Execute all tool calls in parallel, each in a TOOL span
   9. Checkpoint every 10 steps
4. Close RUN span (OK or ERROR), emit completed/failed event
5. Return AgentResult

Methods to override in subclasses

CodeAgent

harness/agents/code_agent.py — Specialized for Python code writing, debugging, and iterative improvement.

agent_type: "code"

Workflow: Understand → Write clean Python with type hints → Lint with ruff → Run code → Iterate on failures.

Default tools: run_python, lint_code, read_file, write_file, apply_patch, list_workspace

SQLAgent

harness/agents/sql_agent.py — Specialized for SQL queries, schema exploration, and data analysis.

agent_type: "sql"

Schema pre-population: Before the main loop, _populate_schema(ctx) introspects the database and stores table/column nodes in the knowledge graph. This enables GraphRAG to surface schema context automatically in subsequent queries.

Default tools: list_tables, describe_table, sample_rows, execute_sql

Framework Adapters

harness/adapters/ — Wraps external agent frameworks with the harness production lifecycle.

import harness

# Auto-detect from object type
adapter = harness.wrap(my_langgraph_graph)   # → LangGraphAdapter
adapter = harness.wrap(my_crew)               # → CrewAIAdapter
adapter = harness.wrap(my_autogen_agent)      # → AutoGenAdapter

# Inject production components
adapter.attach_harness(safety_pipeline, cost_tracker, audit_logger)
adapter.attach_mcp(mcp_client, tool_names=["search", "execute"])python

MemoryManager

harness/memory/manager.py — Unified interface to all memory tiers. Injected into every agent via AgentContext.memory.

Memory tiers

Factory

memory = await MemoryManager.create(config, llm_provider=llm)
# Builds: ShortTermMemory + EmbeddingProvider + VectorStore
#         + GraphMemory + ContextWindowManager + ContextEnginepython

Key methods

ContextEngine

harness/memory/context_engine.py — Paged, skill-isolated context management with automatic offload and semantic retrieval.

Offload pipeline

1. Check hot window size (O(1) Redis LLEN)
2. If count < _HOT_MAX_MSGS (200), skip
3. Compute hot token usage (scan all raw items)
4. If < 80% of max_hot_tokens, skip
5. Walk from oldest to newest until ~2 000 tokens collected
6. Compress (LLM preferred, extractive fallback)
7. Embed summary → store in vector DB with metadata
8. Store ContextPage in Redis HASH + ZSET (scored by importance)
9. LTRIM hot list to remove offloaded messages

Importance scoring

Cold pages with higher importance are kept longer. Score starts at 0.5 and bumps for:

• +0.10 tool messages
• +0.15 error/failed content
• +0.08 result/found/success content

Action scoring formula

composite = 0.5 × goal_progress
          + 0.3 × tool_relevance
          + 0.2 × confidence

goal_progress  = min(1, keyword_overlap(response, goal) × 2)
tool_relevance = 0.9 (success + result > 20 chars)
               | 0.6 (success, no result)
               | 0.5 (no tool)
               | 0.1 (error)
confidence     = 0.6 − 0.08 per hedging phrase
               + 0.08 per affirming phraseformula

Key constants

GraphRAG Engine

harness/memory/graph_rag.py — Weighted multi-hop graph retrieval with vector bridging. Used by SQLAgent (schema) and any knowledge-intensive workload.

Edge weights

Retrieval strategy (priority order)

1. Extract entities from query (regex patterns for table/column names)
2. Find matching graph nodes (find_nodes(entities))
3. If no anchors found → vector-to-graph bridging: embed query, find vector hits, use as graph seeds
4. BFS from anchor nodes, score paths by cumulative edge weight
5. Render top-N paths as compact context: [SCHEMA], [JOINS], [PAST QUERIES], [PAST ERRORS]
6. Supplement with raw vector hits if graph coverage is thin

Semantic LLM Cache

harness/llm/cache.py — Cosine-similarity cache for LLM responses, keyed by message embedding.

Trace System — Overview

HarnessAgent produces a hierarchical span tree for every agent run. Spans are stored in Redis (48 h TTL) and appended to logs/runs/{run_id}/trace.jsonl.

Span hierarchy

The span stack is tracked automatically per run_id inside TraceRecorder._stacks. Callers never need to pass parent_span_id — it is always the top of the stack.

TraceSpan Schema

harness/observability/trace_schema.py

TraceRecorder

harness/observability/trace_recorder.py

Persistence sinks

Usage

recorder = TraceRecorder.create(redis_url=cfg.redis_url, log_dir="logs")

# Context manager (recommended)
async with recorder.span(run_id, SpanKind.TOOL, "tool:execute_sql", ctx,
                          input_preview=str(args)) as span_id:
    result = await registry.execute(ctx, call)

# Manual control
span_id = await recorder.start_span(run_id, SpanKind.LLM, "llm:call", ctx)
recorder.set_llm_usage(span_id, input_tokens=200, output_tokens=100, cost_usd=0.003)
await recorder.end_span(run_id, span_id)

# Query
trace = await recorder.get_trace(run_id)   # → TraceView | None
span  = await recorder.get_span(span_id)   # → TraceSpan | Nonepython

Metrics & Audit

Prometheus metrics (harness/observability/metrics.py)

AuditLogger (harness/observability/audit.py)

Append-only log for compliance. Writes to {workspace_base}/audit/{tenant_id}/{date}.jsonl and Redis stream harness:audit.

Actions tracked: tool_call, memory_write/read, llm_call, agent_start/complete, hitl_request/resolve, safety_block, patch_apply

PII safety: payloads are SHA-256 hashed before storage — content is never stored, only fingerprints.

Event Bus (SSE)

harness/observability/event_bus.py — Redis Pub/Sub channels for real-time streaming via Server-Sent Events.

Channel: harness:events:{run_id}

# BaseAgent emits
await event_bus.publish(StepEvent.llm_called(ctx, response))

# Client subscribes (used by SSE route)
async for event in event_bus.subscribe(run_id):
    print(event.event_type, event.payload)

# From the browser:
const source = new EventSource(`/runs/${runId}/stream`)
source.onmessage = (e) => console.log(JSON.parse(e.data))python / js

ToolRegistry

harness/tools/registry.py — Central tool management with validation, safety, timeout, and audit.

Execution pipeline

1. Lookup — find tool by name; raise TOOL_NOT_FOUND if missing
2. Schema validation — validate args against input_schema; raise TOOL_SCHEMA_ERROR if invalid
3. Safety check — run through safety_pipeline; raise SAFETY_STEP if blocked
4. Timeout — wrap execution in asyncio.timeout(tool.timeout_seconds)
5. Execute — call tool.execute(ctx, args)
6. Audit log — record in AuditLogger
7. Metrics — increment tool_calls_total

Built-in tools

MCP Client

harness/tools/mcp_client.py — Connect to any MCP-compatible server (stdio or SSE) and expose its tools in the ToolRegistry.

from harness.tools.mcp_client import MCPToolAdapter, load_mcp_servers_from_config

# From YAML config
adapters = load_mcp_servers_from_config("mcp_servers.yaml")
for adapter in adapters:
    tools = await adapter.connect()
    for tool in tools:
        registry.register(tool)

# Manual
adapter = MCPToolAdapter(MCPServerConfig(
    name="search",
    transport="stdio",
    command=["npx", "-y", "@modelcontextprotocol/server-brave-search"],
    env={"BRAVE_API_KEY": "${BRAVE_API_KEY}"},
))
tools = await adapter.connect()python

YAML config format

servers:
  search:
    transport: stdio
    command: ["npx", "-y", "@modelcontextprotocol/server-brave-search"]
    env: {BRAVE_API_KEY: "${BRAVE_API_KEY}"}
  filesystem:
    transport: stdio
    command: ["npx", "-y", "@modelcontextprotocol/server-filesystem", "/data"]yaml

Skills

harness/tools/skills.py — Named, versioned, composable agent capabilities. Skills bundle a system prompt fragment, required tools, and usage examples.

from harness.tools.skills import Skill, SkillRegistry

registry = SkillRegistry()
registry.register(Skill(
    name="sql_analysis",
    version="1.0.0",
    description="Analyze SQL databases",
    system_prompt="You are an expert SQL analyst...",
    required_tools=["list_tables", "describe_table", "execute_sql"],
    tags=["sql", "analytics"],
))

# Compose multiple skills
combined = registry.compose("sql_analysis", "data_visualization")

# Filter by tag
analytics_skills = registry.list_for_tags(["analytics"])python

Execution Sandbox

harness/filesystem/sandbox.py — Isolated code execution. Two classes: DockerSandbox (one container per call) and SessionDockerSandbox (one container per run, zero cold-start). RestrictedPython is the fallback when Docker is unavailable.

DockerSandbox security controls

Workload profiles

Set SANDBOX_WORKLOAD in .env to match the memory needs of your workload.

OOM detection

Docker OOM-kills containers with exit code 137. All three execution paths (session, per-call Docker, subprocess fallback) detect this and return a clear ToolResult(error="OOM: container exceeded memory limit") instead of a silent failure. The SandboxResult.oom_killed property is True only when exit_code == 137 and not timed out — timeout kills also exit 137 but are a distinct case.

SessionDockerSandbox — zero cold-start

Every docker run costs 2–5 seconds of cold-start overhead. SessionDockerSandbox starts one container per run and uses docker exec for each run_code call. State (variables, pip-installed packages, written files) persists across all calls within the session.

# Enable session reuse for a run
ctx.metadata["sandbox_session"] = True   # per-run flag
# or globally: SANDBOX_SESSION_REUSE=true in .env

# SessionDockerSandbox lifecycle (BaseAgent manages automatically)
async with SessionDockerSandbox(workspace_path, memory_limit="512m") as sb:
    r1 = await sb.run_code("import pandas as pd; df = pd.read_csv('data.csv')")
    r2 = await sb.run_code("print(df.shape)")  # df is still in scopepython

gVisor / Kata Containers

Set SANDBOX_RUNTIME=runsc to use gVisor (syscall-level kernel isolation) or kata for Kata Containers (full VM). Requires the runtime to be installed on the Docker host. Falls back to runc if the runtime is unavailable.

SANDBOX_RUNTIME=runsc    # gVisor — intercepts all syscalls, stronger isolation
SANDBOX_RUNTIME=kata     # Kata Containers — full lightweight VM per sandboxbash

Skill Store

harness/tools/skill_store.py — Vector-indexed library of reusable artifacts. Agents retrieve relevant skills (code snippets, architectural approaches, monitoring patterns, full codebase patterns) and inject them as context instead of regenerating common work from scratch.

Skill types

SkillArtifact

The core data model. Every skill has a requirements dict ({"pandas": ">=2.0"}) that is validated against the live environment on update_validation(). Mismatches force the skill to BROKEN status.

from harness.tools.skill_store import SkillStore, SkillArtifact, SkillType

store = SkillStore(redis=redis_client, memory_manager=memory)

skill = SkillArtifact(
    skill_id="batch-insert-001",
    tenant_id="acme",
    skill_type=SkillType.CODE,
    title="Batch insert helper",
    description="Inserts rows in 1000-row batches to avoid memory pressure",
    content="def batch_insert(rows, conn):\n    for i in range(0, len(rows), 1000):\n        conn.executemany(sql, rows[i:i+1000])",
    language="python",
    requirements={"psycopg2": ">=2.9"},
    score=0.9,
)
await store.save(skill)python

Retrieval

Vector search first (via MemoryManager.recall()), falls back to relevance-sorted Redis index. Broken skills are excluded by default. Results are injected as a [Relevant skills from library] block in the agent's first user message.

# Wire into agent run — auto-retrieval happens each loop iteration
ctx.metadata["skill_store"] = store

# Manual retrieval
skills = await store.retrieve_relevant(
    query="batch insert postgres",
    tenant_id="acme",
    k=3,
    skill_types=[SkillType.CODE],
    min_score=0.5,
)python

SkillCapture — auto-save from successful runs

Two gates before a skill is saved automatically: quality (score ≥ min_score) and novelty (cosine similarity to nearest existing skill < threshold). If the vector store is unavailable, the novelty gate fails open — the skill is saved.

from harness.tools.skill_store import SkillCapture

capture = SkillCapture(store, min_score=0.8, novelty_threshold=0.9)
saved = await capture.capture(
    title="Cursor-based pagination",
    description="Use WHERE id > last_seen_id instead of OFFSET",
    content="Always use cursor pagination...",
    skill_type=SkillType.APPROACH,
    tenant_id="acme",
    score=0.92,
    run_id=ctx.run_id,
)python

Auto-trigger from BaseAgent (v0.3.2)

High-scoring runs are now captured automatically. Set these in ctx.metadata before the run:

from harness.tools.skill_store import SkillCapture, SkillType

# Wire once at run setup
ctx.metadata["skill_capture"] = SkillCapture(store, min_score=0.8)
ctx.metadata["skill_score"]   = 0.92          # from RLVR, eval scorer, or static
ctx.metadata["skill_title"]   = "Batch insert helper"
ctx.metadata["skill_type"]    = "code"         # code | approach | codebase | monitoring

# BaseAgent.finally calls _maybe_capture_skill() automatically —
# skill is saved if score >= min_score AND novelty gate passespython

Health report & red flags

report = await store.health_report(tenant_id="acme")
print(f"{report.total_skills} skills, {report.broken} broken")
for flag in report.red_flags:
    print(f"  [{flag.severity}] {flag.title}: {flag.detail}")

# Validate against live environment
await store.update_validation(
    skill_id="batch-insert-001",
    status=ValidationStatus.VALID,
    env_requirements={"psycopg2": "2.9.6"},
)python

Secret Provider

harness/security/secrets.py — Abstraction layer that keeps real credentials out of AgentContext, traces, checkpoints, and logs. Swap backends at startup without changing any callsite.

Providers

from harness.security import configure, VaultSecretProvider, CachedSecretProvider, get_secret

# Configure once at application startup
configure(CachedSecretProvider(
    VaultSecretProvider(url="https://vault:8200", token=os.environ["VAULT_TOKEN"]),
    ttl_seconds=300,
))

# Use anywhere — no import changes needed at callsites
key = await get_secret("anthropic_api_key")

# Per-tenant isolation
from harness.security import TenantSecretProvider, EnvSecretProvider
provider = TenantSecretProvider(EnvSecretProvider(), tenant_id="acme")
key = await provider.get("anthropic_api_key")
# → tries ACME/anthropic_api_key first, falls back to ANTHROPIC_API_KEYpython

Safe logging

from harness.security import mask, is_likely_secret

# mask() for safe logging — never log raw keys
logger.info("Using key: %s", mask(api_key))  # → "sk-ant-ap[MASKED]xyz4"

# is_likely_secret() — heuristic detection
is_likely_secret("sk-ant-api03-abc...")  # True
is_likely_secret("hello world")          # Falsepython

Secret Scanner

harness/security/scanner.py — Detect and redact leaked API keys in any string or dict. Called by _HardConstraintPipeline.check_output() and by BaseAgent before LLM responses enter conversation history.

Detected patterns

from harness.security import SecretScanner, redact, scan, has_secrets

scanner = SecretScanner()

# Redact in a string
safe = scanner.redact("use key sk-ant-api03-abc123xyz for auth")

# Scan a nested dict (e.g. tool result, trace payload)
matches = scanner.scan_dict({"config": {"api_key": "sk-ant-…"}})

# Redact an entire dict recursively
clean = scanner.redact_dict(tool_result_data)

# Quick check
if has_secrets(response_content):
    logger.warning("Secret detected in LLM output")python

AgentRunner

harness/orchestrator/runner.py — Creates, executes, and tracks agent runs. The main component called by API routes.

RunRecord states

Planner & Scheduler

Multi-agent DAG decomposition and parallel execution.

Planner (harness/orchestrator/planner.py)

Takes a complex task, calls an LLM to decompose it into a JSON array of SubTask objects with dependency declarations. Validates the DAG (cycle detection via Kahn's algorithm).

planner = Planner(llm_provider=llm)
plan = await planner.plan(
    task="Build a sales dashboard: fetch data, analyze it, generate report",
    available_agents=["sql", "code", "research"],
)
# Returns TaskPlan with SubTasks:
# {id: "fetch", agent_type: "sql",  depends_on: []}
# {id: "analyze", agent_type: "code", depends_on: ["fetch"]}
# {id: "report",  agent_type: "code", depends_on: ["analyze"]}python

Scheduler (harness/orchestrator/scheduler.py)

Executes the plan by repeatedly finding ready tasks (deps satisfied) and running them in parallel with asyncio.gather + semaphore back-pressure.

• Parallelism: max_concurrent=10 semaphore
• Retry: max_retries=1 per subtask, exponential backoff
• Handoff enrichment: predecessor outputs injected into dependent task context
• Failure propagation: tasks with failed deps are skipped and marked failed

Human-in-the-Loop (HITL)

harness/orchestrator/hitl.py — Pause agent execution and wait for human approval before executing sensitive tool calls.

# Wire manager with optional policy learning + memory feedback (v0.3.2)
hitl_manager = HITLManager(
    redis=redis_client,
    policy_store=policy_store,         # rejection → adds tool to blocked_tools
    learn_from_rejections=True,
    memory_manager=memory_manager,     # NEW: approval/rejection stored in memory
)

ctx.metadata["hitl_manager"] = hitl_manager
ctx.metadata["policy"] = policy  # policy.requires_hitl(tool_name) → bool

# BaseAgent calls internally:
request = await hitl_manager.request_approval(run_id, tenant_id, tool_name, tool_args)
decision = await hitl_manager.await_decision(request.request_id, timeout=3600.0)
# "approved" → continue; "rejected"/"expired" → raise HITLRejectedpython

Pending requests are stored in Redis and exposed via GET /hitl/pending.

Inter-Agent Messaging

harness/messaging/bus.py — Redis Streams-based messaging between agents in multi-agent systems.

LLM Router

harness/llm/router.py — Health-aware, context-window-aware routing across multiple LLM providers with per-provider circuit breaking.

Routing algorithm

1. Sort registered providers by priority (lower = higher priority)
2. Skip providers where context_window < required_context
3. Skip providers with open circuit breakers
4. Try each provider in priority order
5. On retryable error (LLM_RATE_LIMIT, LLM_TIMEOUT, LLM_ERROR): try next provider
6. Return response from first successful provider

Circuit breaker

Registering providers

from harness.llm.router import LLMRouter
from harness.llm.anthropic import AnthropicProvider
from harness.llm.openai_provider import OpenAIProvider

router = LLMRouter()
router.register(AnthropicProvider(api_key=cfg.anthropic_api_key),
                priority=0, context_window=200_000)
router.register(OpenAIProvider(api_key=cfg.openai_api_key, model="gpt-4o-mini"),
                priority=10, context_window=128_000)
# Falls back to OpenAI if Anthropic is unavailablepython

Hermes — Self-Improvement Loop

Hermes is the self-healing system that automatically detects failure patterns, generates prompt patches, evaluates them, and optionally applies them — with regression detection and auto-rollback.

Scoring formula

score = success_rate
      − 0.01 × avg_steps_delta
      − 0.001 × avg_tokens_deltaformula

Regression detection (OnlineLearningMonitor)

After every run, metrics are recorded per prompt version. If the post-patch error count exceeds baseline × (1 + regression_threshold), the patch is automatically rolled back to the previous version.

Evaluation Framework

harness/eval/runner.py and harness/eval/diagnostics.py

from harness.eval.runner import EvalRunner
from harness.eval.datasets import EvalDataset, EvalCase

dataset = EvalDataset(
    name="sql_smoke",
    agent_type="sql",
    cases=[
        EvalCase(case_id="c1", task="List all tables", expected_output="users, orders, products"),
        EvalCase(case_id="c2", task="Count users", expected_output="42"),
    ],
)

runner = EvalRunner(agent_runner=agent_runner)
report = await runner.run(dataset, concurrency=3)
print(report.success_rate, report.avg_cost_usd)
print(report.to_markdown())python

EvalDiagnostics

Each EvalReport includes rich per-case diagnostics:

• failure_stage — where the agent failed: llm | tool | safety | timeout | budget | quality
• Recommendations — auto-generated hints based on failure patterns
• By-agent aggregates — token usage, cost, tool errors, guardrail hits per agent type

REST API Reference

Runs

Traces

Evaluations

Improvement

Authentication

Pass an API key in the X-API-Key header. In dev mode, anonymous requests default to tenant "default". JWT bearer tokens are also supported.

Run response schema

{
  "run_id":       "3f2a8c1e4d...",
  "tenant_id":    "acme",
  "agent_type":   "sql",
  "task":         "List all tables",
  "status":       "completed",
  "created_at":   "2024-01-01T00:00:00Z",
  "completed_at": "2024-01-01T00:00:12Z",
  "result": {
    "output":          "Found 7 tables: users, orders...",
    "steps":           4,
    "tokens":          1250,
    "cost_usd":        0.00031,
    "success":         true,
    "elapsed_seconds": 11.8
  }
}json

Trace response schema

{
  "trace_id":             "ddda858ebe8f...",
  "run_id":               "3f2a8c1e4d...",
  "agent_type":           "sql",
  "status":               "ok",
  "duration_ms":          11840,
  "total_input_tokens":   980,
  "total_output_tokens":  270,
  "total_cost_usd":       0.00031,
  "span_count":           6,
  "spans": [
    {
      "span_id":        "01f04e9413851d7f",
      "parent_span_id": null,
      "kind":           "run",
      "name":           "run:sql",
      "status":         "ok",
      "duration_ms":    11840,
      "input_preview":  "List all tables",
      "output_preview": "Found 7 tables..."
    },
    ...
  ]
}json

PEV Verifier

harness/verification/verifier.py — The missing third leg of the agent control loop. Without a verifier, agents terminate on implicit convergence — when they stop requesting tool calls — not when their output is objectively correct.

Verifier protocol

from harness.verification import CodeExitCodeVerifier, NoopVerifier, ExpectedOutputVerifier

# Wire into agent run — zero overhead, checks exit code from last run_python call
ctx.metadata["verifier"] = CodeExitCodeVerifier()

# Known-answer verifier for evals
ctx.metadata["verifier"] = ExpectedOutputVerifier(expected="42", exact=True)

# No-op when you have no objective oracle
ctx.metadata["verifier"] = NoopVerifier()python

How it works

After each no-tool-call LLM response, BaseAgent._verify_output() calls the verifier. If it fails, the feedback is injected as a user message and the loop continues — up to _MAX_VERIFICATION_ATTEMPTS = 3. The verifier never crashes a run: exceptions fail-open (treated as skipped).

CodeExitCodeVerifier failure triggers

Agent Blackboard

harness/orchestrator/blackboard.py — Per-plan Redis-backed shared artifact store. Instead of receiving only text summaries of predecessor outputs, downstream agents can query the exact typed artifact they need.

Artifact types

from harness.orchestrator import AgentBlackboard

# Scheduler creates one blackboard per plan when redis= is provided
scheduler = Scheduler(agent_runner=runner, redis=redis_client)

# Agents publish typed artifacts via result metadata
result.metadata["blackboard_sql"]         = "SELECT COUNT(*) FROM orders"
result.metadata["blackboard_test_result"] = "passed: 42 rows"

# Direct API
bb = AgentBlackboard(redis_client, plan_id="plan-123")
await bb.write("sql_agent", "sql", "SELECT * FROM users")
entry = await bb.get("sql_agent", "sql")

# Downstream agent reads predecessor artifacts
context = await bb.format_for_context(subtask_ids=["sql_agent"])
# → "[Shared plan artifacts]\n--- sql_agent / sql ---\nSELECT * FROM users"

# Query by type
sql_entries = await bb.read(artifact_types=["sql", "test_result"])python

Scheduler integration

When redis= is provided to Scheduler, each subtask's task description is automatically prefixed with a [Shared plan artifacts] block containing all predecessor artifacts. Backward compatible — omit redis= for the existing text-only handoff.

Conflict detection (v0.3.2)

Two agents writing to the same (subtask_id, artifact_type) key use last-write-wins semantics. The previous writer's identity is logged as a warning and the written_by field in metadata is always set, providing a full audit trail of who wrote what and when.

Failure Attribution

harness/eval/failure_taxonomy.py — Maps AgentResult.failure_class to the harness layer that caused the failure. Separates content failures (wrong SQL, hallucination) from infrastructure failures (tool timeout, budget exceeded).

HarnessComponent

from harness.eval.failure_taxonomy import attribute_to_component, HarnessComponent
from harness.eval.agent_scorer import AgentScores

# Pass failure_class through from AgentResult
scores = AgentScores(..., harness_failure_class=result.failure_class)

# EvalReport.component_attribution() counts failures per layer
report = AgentEvalReport("my-eval", scores_list)
attr = report.component_attribution()
# → {"tool": 6, "budget": 3, "safety": 1}

# In markdown report — Harness Component Attribution table is auto-includedpython

Extended Hermes patches

PatchGenerator now generates patches beyond prompts based on failure class:

Redis Schema

Dependencies

Core — v0.3.2 (~50 MB, no torch)

Optional extras — pick what you need