HarnessAgent (HaaS)

HarnessAgent is a 7-pillar production infrastructure layer for any AI agent framework. It does not replace your agent logic — it wraps it with everything production demands: resilient LLM routing, multi-tier context engineering, real-time feedback, online reinforcement learning, safety guardrails, full observability, and a REST API.

The Design Contract

The core contract is intentionally minimal. Your framework implements a single method; the Harness takes care of everything else.

class MyAgent(BaseHaasAgent):
    async def task_step(self, state: AgentState) -> StepResult:
        # Your agent logic here — one step at a time
        response = await self.llm.complete(state.messages)
        return StepResult(output=response.text, done=True)

# The harness wraps execution end-to-end
trace_view = await harness.run(agent, task)
# → TraceView: spans, scores, token usage, cost, feedback logpython

The harness calls task_step in a loop, injecting feedback, checking guardrails, recording spans, and computing rewards — all transparently.

7 Production Pillars

Supported Frameworks

Quick Start

Install

# Core + all production extras
pip install agent-haas

# With optional vector / observability extras
pip install "agent-haas[vector,observe]"

# Full install (includes all optional deps)
pip install "agent-haas[all]"bash

Start infrastructure

A Docker Compose file ships with the package to bring up Redis, Chroma, and the API server with a single command.

# Clone or copy the docker-compose.yml from the repo
docker compose up -d

# Verify all services are healthy
curl http://localhost:8000/health
# → {"status": "ok", "redis": "ok", "vector_store": "ok"}bash

4-line integration

The minimum viable integration — supply your agent class and a task string.

from haas import Harness, HaasConfig

harness = Harness(HaasConfig.from_env())          # reads .env / env vars
trace   = await harness.run(MyAgent(), task="analyse Q3 sales")
print(trace.output, trace.cost_usd, trace.pass_rate)python

First run with NexusSql (SQLAgent)

from haas import Harness, HaasConfig
from haas.agents.nexussql import NexusSqlAgent

# Build harness from environment
config  = HaasConfig.from_env()
harness = Harness(config)

# Instantiate the NexusSql agent (self-correcting, schema-aware)
agent = NexusSqlAgent.from_config(config, db_path="./northwind.db")

# Run and inspect trace
trace = await harness.run(agent, task="List top 10 customers by revenue in 2024")

print("SQL:",   trace.output)
print("Cost:",  trace.cost_usd)
print("Steps:", trace.step_count)
print("Score:", trace.rlvr_score)   # correctness / quality / safetypython

Wrapping an existing LangGraph agent

import haas
from langgraph.graph import StateGraph

graph   = StateGraph(...)           # your existing graph — unchanged
adapter = haas.wrap(graph)

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

LLM Router

The LLMRouter is the single point of contact between agent logic and LLM providers. It implements health-aware routing, circuit breaking, exponential backoff with jitter, and semantic response caching — all transparently to the caller.

Health-aware routing

Before each request the router checks a per-provider health cache with a 5-second TTL. If the cached status is UNHEALTHY the provider is skipped without a network call. Health probes are lightweight HEAD requests to the provider's status endpoint.

from haas.llm.router import LLMRouter
from haas.llm.providers import OpenAIProvider, AzureOpenAIProvider

router = LLMRouter()

# Register providers in priority order (0 = highest priority)
router.register(OpenAIProvider(model="gpt-4o"),         priority=0)
router.register(AzureOpenAIProvider.from_env(),         priority=1)
# local vLLM fallback
router.register(VLLMProvider(base_url="http://gpu01:8080"), priority=2)

response = await router.complete(messages, temperature=0.2)python

Exponential backoff

On a retryable error (rate-limit 429, server 5xx) the router applies exponential backoff with ±20% jitter before switching providers. The backoff sequence is:

• Attempt 1 → wait 1 s ± 0.2 s
• Attempt 2 → wait 2 s ± 0.4 s
• Attempt 3 → wait 4 s ± 0.8 s
• After max retries exhausted on a provider → fall through to next priority level

Circuit Breaker

Each provider has an independent circuit breaker that prevents cascading failures from overwhelming a degraded upstream.

Semantic LLM Cache

Identical or semantically near-identical prompts are served from cache, eliminating redundant API calls and latency.

from haas.llm.cache import SemanticLLMCache

cache = SemanticLLMCache(
    similarity_threshold=0.97,   # cosine threshold
    scan_cap=200,                 # max entries scanned per lookup
    backend="redis",              # or "memory" for testing
)
router = LLMRouter(cache=cache)

# First call — cache miss, API call made
r1 = await router.complete([{"role": "user", "content": "What is 2+2?"}])

# Semantically equivalent — cache hit (no API call)
r2 = await router.complete([{"role": "user", "content": "What does 2+2 equal?"}])
assert r2.cache_hit == Truepython

Azure OpenAI Support

from haas.llm.providers import AzureOpenAIProvider

# From explicit config
az = AzureOpenAIProvider(
    endpoint="https://my-resource.openai.azure.com/",
    api_key="...",
    deployment_name="gpt-4o",
    api_version="2024-08-01-preview",
)

# Or auto-populate from environment variables
# AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, AZURE_OPENAI_DEPLOYMENT
az = AzureOpenAIProvider.from_env()python

Local inference (vLLM / SGLang / llama.cpp)

from haas.llm.providers import VLLMProvider, SgLangProvider, LlamaCppProvider

# vLLM — OpenAI-compatible endpoint
router.register(VLLMProvider(base_url="http://gpu01:8080", model="mistral-7b"), priority=1)

# SGLang
router.register(SgLangProvider(base_url="http://gpu02:30000"), priority=2)

# llama.cpp server
router.register(LlamaCppProvider(base_url="http://localhost:8080"), priority=3)python

Cost-Aware Routing

On top of health- and context-aware fallback, the router can choose a model by cost/capability tier per request. Each request is scored for complexity, mapped to a tier (cheap / standard / premium), and served by the tier's model — falling back through the priority chain if that tier is unavailable. The whole mechanism is opt-in: with no scorer or tier map configured, the router behaves exactly as a priority-ordered router.

Why a heuristic scorer (and not a model)

The most efficient router adds no extra LLM call. HeuristicComplexityScorer inspects signals already in the request — token estimate, tool use, code/SQL/reasoning keywords, turn count, required context — and returns a tier in microseconds at zero cost. Calling a model to decide which model to call would add latency and spend to every request, undermining the savings. The ComplexityScorer protocol lets you swap in a learned classifier later (a fastembed embedding head, or a small LLM like Haiku / gpt-4o-mini) without touching the router.

Multi-vendor providers

Tiers can mix vendors freely. Beyond Anthropic, Azure, and OpenAI, a declarative catalog wires every OpenAI-API-compatible vendor with no new adapter code — each is the existing OpenAI provider pointed at the vendor's base_url. Enable one by setting its key.

AWS Bedrock

Two adapters cover Bedrock's two request shapes — install with pip install agent-haas[bedrock,anthropic] and set BEDROCK_ENABLED=true (AWS credentials resolve from the standard chain):

Per-tenant model maps

Each tenant can be served by its own model per tier. The agent threads tenant_id into every LLM call; the router reads that tenant's tier→model map (keys are provider:model, across vendors) and prefers those providers. Internal work like history summarization is pinned to cheap so it never burns the premium model.

# .env — enable vendors + Bedrock, then map tiers per tenant
DEEPSEEK_API_KEY=sk-...
BEDROCK_ENABLED=true
BEDROCK_CLAUDE_MODELS=anthropic.claude-opus-4-7:premium,anthropic.claude-haiku-4-5:cheap
ROUTING_TENANT_TIERS={"acme":{"cheap":["deepseek:deepseek-chat"],
                     "premium":["bedrock:anthropic.claude-opus-4-7"]}}bash

# The agent passes tenant_id automatically; you can also pin a tier explicitly:
resp = await router.complete(messages, max_tokens=1024,
                             tenant_id=ctx.tenant_id)   # tier inferred by complexity
resp = await router.complete(messages, max_tokens=1024,
                             tier="premium")              # or force a tierpython

Routing precedence: explicit tier → complexity scorer → plain priority order. Tier maps fall back to each provider's tier tag when no per-tenant map is set. Cost-aware routing is enabled by default (ROUTING_COMPLEXITY_ENABLED=true).

Context Engineering (L1 / L2 / L3)

The ContextPipeline assembles context for each LLM call from three tiers with configurable token budget splits. Each tier is independent and can be swapped or disabled.

L1 — ChatCache (hot window)

Redis-backed conversation window scoped by run_id + skill_ns. When the window grows beyond 80% capacity, older messages are automatically offloaded to cold storage (paged LZ4-compressed JSONL) and the active window is trimmed. Retrieval is O(1) from Redis LRANGE.

L2 — VectorCache (semantic memory)

Long-term semantic memory backed by Chroma, Qdrant, or Weaviate. At query time the pipeline embeds the current query and retrieves the top-k most relevant past memories. Only entries with cosine similarity ≥ 0.70 are included in the assembled context.

L3 — KnowledgeStore

Structured knowledge combining a SQL schema registry and a graph traversal cache.

from haas.context.l3 import SchemaStore

store = SchemaStore()
await store.store_from_sqlite("./northwind.db")
schema = await store.get_relevant(
    query="top customers by revenue",
    max_tables=8,
)python

ContextPipeline — full assembly

from haas.context.pipeline import ContextPipeline
from haas.context.l1 import ChatCache
from haas.context.l2 import VectorCache
from haas.context.l3 import SchemaStore, KGCache, KnowledgeStore

pipeline = ContextPipeline(
    l1=ChatCache(redis_url="redis://localhost:6379"),
    l2=VectorCache(backend="chroma", threshold=0.70),
    l3=KnowledgeStore(schema=SchemaStore(), kg=KGCache()),
    budget_split=(0.60, 0.25, 0.15),   # L1 / L2 / L3
)

ctx = await pipeline.assemble(
    run_id="run-abc123",
    query="top customers by revenue in 2024",
    db_id="northwind",
    token_budget=60_000,
)

# Inspect per-tier stats
print(ctx.stats.l1_tokens, ctx.stats.l1_hit)    # 36000, True
print(ctx.stats.l2_tokens, ctx.stats.l2_hit)    # 15000, True
print(ctx.stats.l3_tokens, ctx.stats.l3_hit)    # 9000, True
print(ctx.stats.total_latency_ms)               # e.g. 12.4python

AssembledContext.stats fields

Real-time Feedback Channel

The FeedbackChannel lets operators — or automated supervisors — inject guidance into a running agent without stopping it. Events are delivered via Redis Streams and consumed by the agent before every LLM call.

FeedbackEvent types

Publishing feedback (operator / API client)

from haas.feedback import FeedbackChannel, FeedbackEvent

channel = FeedbackChannel(redis_url="redis://localhost:6379")

# Publish a correction mid-run
await channel.publish(
    run_id="run-abc123",
    event=FeedbackEvent(
        type="correction",
        content="The date filter should use ORDER_DATE not SHIP_DATE",
        author="ops-team",
    )
)

# Non-blocking poll — returns immediately if no events
events = await channel.poll(run_id="run-abc123", count=10)
# Uses XREAD with block=None (non-blocking), NOT block=0python

How the agent consumes feedback

Every BaseHaasAgent subclass calls _apply_feedback() at the top of its step loop before any LLM call. The method checks the should_inject predicate per event type and mutates the message list accordingly.

# Inside BaseHaasAgent.run() — simplified
async def _run_loop(self, state):
    while not state.done:
        # 1. Pull feedback before every LLM call
        events = await self.feedback_channel.poll(state.run_id)
        state   = await self._apply_feedback(state, events)

        if state.stop_requested:
            break

        # 2. Assemble context + call LLM
        result = await self.task_step(state)
        state  = state.advance(result)

def should_inject(event: FeedbackEvent) -> bool:
    if event.type in ("correction", "hint", "redirect", "stop"):
        return True
    if event.type == "score":
        return event.score < 0.40   # only inject low scores
    return Falsepython

REST API endpoints

# POST /runs/{id}/feedback
curl -X POST http://localhost:8000/runs/run-abc123/feedback \
  -H "Content-Type: application/json" \
  -d '{"type":"correction","content":"Use ORDER_DATE not SHIP_DATE"}'

# GET /runs/{id}/feedback
curl http://localhost:8000/runs/run-abc123/feedbackbash

RLVR — Reinforcement Learning from Verifiable Rewards

RLVR provides online, per-step reward signals using deterministic, domain-specific verifiers. Unlike the Hermes self-improvement loop (offline, batch, binary pass/fail), RLVR operates in real time and produces graded scores that feed directly into advantage estimation and few-shot reinforcement.

Deterministic verifiers

All verifiers run at temperature=0. Results are cached with SHA-256 on (agent_type, input, output) so re-verification of identical inputs is free. A rule-based fallback fires if the primary verifier raises an exception.

SQLVerifier — 5-step pipeline

CodeVerifier — 4-step pipeline

ReasoningVerifier — 3-step pipeline

Verifier factory

from haas.rlvr.verifiers import get_verifier

verifier = get_verifier("sql")       # → SQLVerifier
verifier = get_verifier("code")      # → CodeVerifier
verifier = get_verifier("reasoning") # → ReasoningVerifier

result = await verifier.verify(
    question="Top 10 customers by revenue",
    output="SELECT customer_id, SUM(amount) ...",
    db_path="./northwind.db",
)
print(result.scores)    # {"correctness": 0.9, "quality": 0.8, "safety": 1.0}
print(result.passed)    # True — all thresholds metpython

AgentScores — 3-dimension scoring

Every verification produces an AgentScores object. A run PASSES only when all three dimensions independently meet their thresholds.

AdvantageEstimator

from haas.rlvr.advantage import AdvantageEstimator

estimator = AdvantageEstimator(gamma=0.95)

# rewards: list of per-step scalar rewards for the episode
advantages = estimator.compute(rewards=[0.3, 0.7, 0.9])
# Returns: discounted returns - rolling baseline, normalised to mean=0 std=1
# Rolling baseline computed over last 50 episodes (StepRewardBuffer)python

StepRewardBuffer

Redis-backed per-episode reward buffer. Maintains a rolling baseline over the last 50 episodes used by the AdvantageEstimator. Keys are scoped per agent type to prevent cross-domain contamination.

RLVRLoop — high / low advantage routing

from haas.rlvr.loop import RLVRLoop

loop = RLVRLoop(
    verifier=get_verifier("sql"),
    advantage_threshold=0.5,   # above → few-shot reinforce
)

# After a step completes:
await loop.process_step(
    run_id=state.run_id,
    step_output=result.output,
    question=state.task,
)
# High advantage  → few-shot examples prepended to next LLM prompt
# Low advantage   → weighted Hermes patch queued for offline cyclepython

GEPA — Reflective Prompt Optimization

GEPA evolves an agent's prompt components (system prompt, planner prompt, context summary) through reflective mutation, scoring each candidate against a gold-labeled eval dataset and keeping what wins. It is the third member of the self-improvement family: where RLVR reinforces online per step and Hermes applies offline code patches, GEPA optimizes the prompts themselves offline. Optional dependency — pip install agent-haas[improvement]; when the gepa package is absent the loop falls back to the heuristic PatchGenerator automatically.

Public surface

Optimize prompts on a dataset

from harness.improvement.gepa import optimize_prompts_on_dataset

result = await optimize_prompts_on_dataset(
    eval_runner=eval_runner,          # its scorers define correctness
    dataset=trainset,                 # labeled EvalDataset (GEPA trainset)
    llm_provider=reflection_lm,        # the reflection / teacher LM
    seed_prompts={"system_prompt": current_prompt},
    valset=heldout,                    # held-out split for scoring
    budget=60,                         # max candidate evaluations
    pass_threshold=0.5,
    concurrency=3,
    use_mlflow=True,                   # log each candidate to MLflow
)

print(result.improved)            # True if any component changed vs seed
print(result.best_score)          # aggregate validation score of the winner
print(result.components)          # evolved component name -> textpython

EvalOptimizationResult fields

Benchmark CLI

The offline experiment driver scripts/gepa_optimize.py loads a benchmark, builds the production agent stack, seeds the requested components, runs GEPA against the gold-labeled data, and reports baseline vs optimized score on a held-out split.

# GSM8K (default) — gold = final number, exact-match is a true correctness signal
python scripts/gepa_optimize.py --benchmark gsm8k --data-dir /data/gsm8k.jsonl \
    --n-samples 40 --budget 80

# HumanEval — pass@1 execution scoring (runs generated code in a sandbox)
python scripts/gepa_optimize.py --benchmark humaneval --data-dir /data/humaneval.jsonl \
    --components system_prompt,context_summary --mlflowbash

Key CLI flags: --budget (max candidate evaluations), --components (comma-separated prompt parts to evolve), --scorer, --val-frac, --concurrency, --reflection-model, --mlflow. This is an offline job requiring a real agent runtime (LLM credentials, Redis) — not the request path.

Secret Vault & Scanner

API keys never appear in agent context, traces, or checkpoints. The SecretProvider abstraction keeps credentials out of AgentContext, and SecretScanner redacts any that leak into LLM responses before they enter history.

# Dev — zero migration, reads from os.environ / .env
from harness.security import get_secret
key = await get_secret("anthropic_api_key")

# Production — swap backend without changing any callsite
from harness.security import configure, VaultSecretProvider, CachedSecretProvider
configure(CachedSecretProvider(
    VaultSecretProvider(url="https://vault:8200", token=os.environ["VAULT_TOKEN"]),
    ttl_seconds=300,
))python

The scanner detects Anthropic sk-ant-, OpenAI sk-, GitHub ghp_/github_pat_, Slack xoxb-/xoxp-, GitLab glpat-, JWTs, bearer tokens, and URL-embedded credentials — redacting them before they touch history, memory, or the trace store. TenantSecretProvider adds per-tenant isolation with global fallback.

Session Sandbox & Workload Profiles

Every run_python call previously started a fresh Docker container (~2–5 s each) — 20–50 s of overhead for a 10-iteration debug loop. Session reuse runs one container for the whole run: it starts at run begin, docker execs each call, and stops on exit. Variables, pip-installed packages, and written files persist between calls.

# Enable session reuse — one container for the entire run
ctx.metadata["sandbox_session"] = True
# or: SANDBOX_SESSION_REUSE=true in .envpython

Workload profiles & isolation

OOM kills surface a clear OOM: container exceeded memory limit error (instead of opaque exit code 137) across all three execution paths — session, per-call Docker, and subprocess fallback. Container death is detected and reported as SandboxError("Session container died").

Execution providers — local or third-party

Code execution runs self-hosted by default, but the session sandbox is pluggable via SANDBOX_PROVIDER. Each provider implements the same surface (async context manager + run_code(code, timeout) → SandboxResult + is_available()) and is stored under the same metadata key, so the agent's RunCodeTool is unchanged regardless of backend.

# Offload execution to E2B's cloud sandbox
SANDBOX_PROVIDER=e2b
E2B_API_KEY=e2b_xxxx

# …or Modal serverless containers
SANDBOX_PROVIDER=modal
MODAL_TOKEN_ID=ak-xxxx
MODAL_TOKEN_SECRET=as-xxxxbash

A provider falls back gracefully: if its SDK or credentials are missing, is_available() returns false and the run proceeds without a persistent session. Adding another backend (Daytona, Firecracker, …) is just implementing the three-method surface and registering a selector branch.

Skill Store

Agents retrieve relevant skills — code snippets, architectural approaches, monitoring patterns — from a vector-indexed library instead of regenerating common work from scratch. Token savings compound for patterns used repeatedly.

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

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

# Auto-capture from a successful run (score gate + novelty gate)
capture = SkillCapture(store, min_score=0.8)
await capture.capture(
    title="Batch insert helper", description="...", content="...",
    skill_type=SkillType.CODE, tenant_id="acme", score=0.92,
    run_id=ctx.run_id,
)

# Wire into a run — skills are auto-retrieved and injected into context
ctx.metadata["skill_store"] = storepython

Skills declare requirements (e.g. {"pandas": ">=2.0"}); update_validation() checks them against the live environment and marks a skill BROKEN when no longer satisfied. health_report() surfaces dashboard red flags:

Checkpoints, Policy & Output Caps

Reliable checkpoints

• Full history saved — CheckpointManager.save(ctx, history) serializes the complete conversation alongside step/token counts
• Correct resume — load(run_id, tenant_id) restores both counters and history; the loop resumes from the exact message where it stopped
• Always-on — saved in finally on every exit path: clean completion, budget exceeded, exception, and CancelledError

Policy enforcement

Per-tenant policies enforce blocked_tools, allow_code_execution, and allow_file_write at tool dispatch time — before HITL approval — raising SafetyViolation(SAFETY_STEP) immediately.

from harness.safety.policies import HarnessPolicy

ctx.metadata["policy"] = HarnessPolicy(
    tenant_id="acme",
    blocked_tools=["drop_table", "delete_database"],
    allow_code_execution=False,   # blocks run_python, exec_*, run_* tools
    allow_file_write=False,       # blocks write_file, apply_patch, write_* tools
)python

Tool result size cap

Large tool outputs (full-table SQL dumps, verbose file reads) are capped at 8,000 chars before entering agent history, preventing a single noisy call from consuming the whole context window. The truncation suffix shows the original byte count, and result.metadata carries {"truncated": True, "original_chars": N}. The FailureCategory.OUTPUT_TRUNCATED category surfaces this automatically during eval.

Budgets, rate limits & cancellation

Three independent guards bound what a run is allowed to consume, enforced inside the agent loop and at the API edge:

• Per-tenant cost cap — before every LLM call the loop checks the tenant's accumulated monthly spend against COST_BUDGET_USD_PER_TENANT. Over budget, the run stops immediately with failure class BUDGET_COST (HTTP 429) before incurring further spend. A Redis/infra error in the check fails open so a transient outage never kills a healthy run. Toggle with ENFORCE_COST_BUDGET.
• Step / token / time budgets — ctx.tick() raises BudgetExceeded with BUDGET_STEPS, BUDGET_TOKENS, or BUDGET_TIME the moment a per-run limit is crossed.
• API rate limiting — a Redis sliding-window limiter enforces RATE_LIMIT_RPM requests/minute per tenant via middleware. Health and docs paths are exempt; when Redis is unavailable the limiter fails open (requests pass) rather than returning 500s. Responses carry X-RateLimit-Remaining / X-RateLimit-Reset; a denied request returns 429 with Retry-After. Toggle with RATE_LIMIT_ENABLED.
• Operator cancellation — DELETE /runs/{id} flips the persisted run status to cancelled; the loop polls that status between steps and stops a running agent with failure class CANCELLED (not just pending runs).

NexusSql Agent

NexusSql is HaaS's built-in self-correcting SQL agent optimised for large databases (100+ tables). It combines L3 SchemaStore routing, SQLVerifier feedback loops, and optional harness wrapping through NexusSqlAgent.

Schema-aware routing

For queries that touch large databases, NexusSql uses the L3 SchemaStore to identify which tables are relevant before sending anything to the LLM. When there are more than 12 candidate tables, keyword-based routing narrows the field so the LLM prompt contains only the relevant schema fragments — not the entire 100-table schema.

Self-correction loop

Standalone usage (no harness)

from haas.agents.nexussql import NexusSql

# Simple one-shot SQL generation with self-correction
sql = await NexusSql.generate_sql(
    question="What are the top 5 products by units sold in Q1 2024?",
    db_path="./sales.db",
)
print(sql)
# SELECT product_name, SUM(quantity) AS units_sold
# FROM order_items JOIN orders ON ...
# WHERE order_date BETWEEN '2024-01-01' AND '2024-03-31'
# GROUP BY product_name ORDER BY units_sold DESC LIMIT 5python

Harness agent usage

from haas.agents.nexussql import NexusSqlAgent
from haas import Harness, HaasConfig

config  = HaasConfig.from_env()
harness = Harness(config)

# from_config() picks up OPENAI_API_KEY / AZURE_* from .env automatically
agent = NexusSqlAgent.from_config(config, db_path="./sales.db")

trace = await harness.run(
    agent,
    task="Monthly revenue trend for the last 12 months"
)
print(trace.output)       # generated + verified SQL
print(trace.step_count)   # 1–3 depending on corrections
print(trace.rlvr_score)   # AgentScores objectpython

NexusSqlAgent — class hierarchy

BaseHaasAgent
  └─ SQLAgent            # base SQL agent (schema, execution)
       └─ NexusSqlAgent  # + L3 routing, self-correction, RLVRtext

from_config() factory

NexusSqlAgent.from_config(config, db_path) wires up all dependencies from environment variables:

• LLM provider from OPENAI_API_KEY or AZURE_OPENAI_*
• SchemaStore auto-indexed from db_path
• FeedbackChannel from REDIS_URL
• SQLVerifier with SHA-256 cache enabled
• RLVRLoop with advantage_threshold=0.5

General Benchmark (bench_agent.py)

The bench_agent.py CLI evaluates any HaaS-compatible agent against standard datasets. It auto-selects the correct domain verifier from the agent type, estimates cost before running, and emits a structured BenchReport.

Supported domains

CLI usage

# SQL benchmark against BIRD dev set (cost estimate shown first)
python bench_agent.py \
  --domain sql \
  --dataset bird \
  --agent NexusSqlAgent \
  --db-path ./bird_dev_dbs/ \
  --n 100 \
  --estimate-cost

# GSM8K reasoning benchmark, 20 samples
python bench_agent.py \
  --domain base \
  --dataset gsm8k \
  --n 20 \
  --model gpt-4o

# Custom JSONL dataset
python bench_agent.py \
  --domain base \
  --dataset ./my_evals.jsonl \
  --n 50bash

BenchReport fields

from haas.bench import run_benchmark

report = await run_benchmark(
    agent_class=NexusSqlAgent,
    domain="sql",
    dataset="bird",
    n=100,
    db_path="./bird_dev_dbs/",
)

print(f"Pass rate: {report.pass_rate:.1%}")
print(f"By hardness: {report.by_hardness}")
print(f"Step failures: {report.failure_distribution}")python

Benchmark Results

System benchmarks (GraphRAG, semantic cache, circuit breaker, span overhead, Hermes loop) are deterministic and reproducible offline with PYTHONPATH=src python benchmarks/run_all.py — no API keys or datasets, Redis is faked in-process; they regenerate benchmarks/results/REPORT.md. Task / accuracy benchmarks (BIRD, GSM8K, HumanEval) were measured with NexusSql on GPT-5.5; because they need a live LLM and the dataset download, the numbers depend on the model + dataset version — reproduce them with bench_agent.py using your own keys. The Hermes figure is a synthetic loop-mechanics demo (mock LLM modeling a clearly-good fix), not a measured LLM gain.

Summary

† Task/accuracy benchmark — measured with NexusSql on GPT-5.5; reproduce with your own keys + dataset via bench_agent.py --agent sql --dataset bird (results depend on the model and dataset version). ‡ The Hermes self-improvement benchmark uses a mock LLM and a modeled golden fix to exercise the loop mechanics end-to-end; real self-improvement requires bench_hermes_real.py with a live LLM.

Full results table

Safety & HITL

HaaS enforces safety at three points in the execution lifecycle through a staged guardrail pipeline, and provides a human-in-the-loop approval gate for high-risk decisions.

3-stage guardrail pipeline

from haas.safety import SafetyPipeline, GuardrailPolicy

policy = GuardrailPolicy(
    tenant_id="acme",
    blocked_tools=["DROP", "DELETE"],
    pii_redaction=True,
    max_cost_usd=2.00,
)

pipeline = SafetyPipeline(policy=policy)
harness  = Harness(config, safety=pipeline)python

HITLManager

When an agent reaches a checkpoint that requires human approval (e.g. destructive database mutations, exceeding cost budget), the run is paused and an approval request is created. The operator receives a Server-Sent Event push and approves or rejects via the REST API.

from haas.safety.hitl import HITLManager

hitl = HITLManager(redis_url="redis://localhost:6379")

# Inside an agent step — pause for human review
request_id = await hitl.request_approval(
    run_id=state.run_id,
    action="ALTER TABLE orders ADD COLUMN discount DECIMAL",
    metadata={"reason": "schema migration required"},
)

decision = await hitl.await_decision(request_id, timeout=300)

if decision.approved:
    await execute_migration()
else:
    raise HITLRejected(decision.reason)python

Per-tenant policies

Policies are stored in Redis keyed by tenant_id and loaded at the start of each run. They control: allowed tools, blocked SQL keywords, PII redaction rules, max cost per run, and HITL trigger conditions. Policies persist without expiry — a tenant's restrictions never silently lapse back to the permissive default — and can be updated at runtime without restarting the server.

Observability

Span trace tree — 7 SpanKind types

Every run produces a hierarchical span tree stored in Redis (live) and JSONL (durable). The root span is always RUN; all other spans are children.

Prometheus metrics (13 pre-defined)

MLflow experiment tracking

from haas.observability.eval_log import log_all

# Log a BenchReport to all configured backends
log_all(
    report,
    mlflow=True,      # logs metrics + artifacts to MLflow
    wandb=False,       # optional W&B integration
    langsmith=False,   # optional LangSmith dataset logging
)python

Audit log

Every run produces an append-only audit entry in audit.jsonl. PII fields (email, phone, SSN patterns) are SHA-256 hashed before writing. The log captures: run_id, tenant_id, agent_type, task hash, tool calls, guardrail decisions, and final outcome.

# Example audit entry (PII hashed)
{
  "run_id": "run-abc123",
  "tenant_id": "acme",
  "agent_type": "sql",
  "task_hash": "sha256:a3f1...",
  "tool_calls": ["schema_lookup", "sql_execute"],
  "guardrail_result": "PASS",
  "outcome": "PASS",
  "cost_usd": 0.0042,
  "ts": "2026-05-17T10:23:44Z"
}json

REST API Reference

All endpoints require Authorization: Bearer <JWT> in production. The JWT must contain tenant_id. In development mode (ENVIRONMENT=dev) auth is bypassed.

Runs

Feedback

Evals

Health

Request / response examples

# POST /runs — start a NexusSql run
curl -X POST http://localhost:8000/runs \
  -H "Authorization: Bearer $JWT" \
  -H "Content-Type: application/json" \
  -d '{"agent_type":"sql","task":"Top 10 customers by revenue 2024"}'
# → {"run_id":"run-abc123","status":"running"}

# GET /runs/{id}/stream — SSE stream
curl -N http://localhost:8000/runs/run-abc123/stream \
  -H "Authorization: Bearer $JWT"
# → data: {"event":"step_start","step":1}
# → data: {"event":"llm_call","tokens":1240,"provider":"openai"}
# → data: {"event":"run_end","output":"SELECT ...","cost_usd":0.004}

# GET /runs/{id}/trace — full span tree
curl http://localhost:8000/runs/run-abc123/trace \
  -H "Authorization: Bearer $JWT"

# GET /health
curl http://localhost:8000/health
# → {"status":"ok","redis":"ok","vector_store":"ok","llm_router":"ok"}bash