Agent Harness — Technical Documentation¶
The Agent Harness is SecondBrain's bounded chat-turn executor. It takes a user message, assembles context, runs a tool-calling loop against an LLM provider, and returns a structured TurnResult.
Documentation Map¶
| Document | What it covers |
|---|---|
| Overview & Architecture | Design philosophy, collaborator map, key invariants |
| Data Flow: A Turn End-to-End | Step-by-step walkthrough of every stage in run_turn() |
| AgentHarness Façade | Constructor parameters and public API reference |
| Comparison And Improvement Plan | Side-by-side concept map and prioritized improvement backlog |
| Collaborators | |
| TurnBudget & DeadlineToken | Resource envelope, deadline propagation, cooperative cancellation |
| TurnRuntimeState | Per-turn mutable state container, late-write gate |
| ToolOutcomes | Typed outcome union replacing ad-hoc dicts |
| ToolScheduler | Safe wave partitioning, parallel vs sequential |
| ReplayControl | Idempotency-aware deduplication |
| ArtifactStore | Session-scoped, TTL-aware oversized output storage |
| BoundedToolExecutor | Deadline-safe, late-write-blocked tool execution |
| TurnPreparer | Message assembly, per-turn render hash |
| ReflectionEngine | Budget-aware LLM reflection step |
| TurnFinalizer | Citations, memory, decision records, judge scheduling |
| ThinkingManager | Thinking levels and reflection scheduling |
| Testing Guide | Test coverage map, how to write new tests |
Quick Start¶
from brain.agent.harness import AgentHarness, TurnResult
from brain.agent.tools import AgentToolRegistry
from brain.providers import build_provider
from brain.state import EventLog
harness = AgentHarness(
provider=build_provider("anthropic"),
tools=AgentToolRegistry(...),
event_log=EventLog(...),
max_steps=6,
timeout_s=60,
)
result: TurnResult = harness.run_turn(
session_id="sess_abc123",
history=[],
user_message="Summarise my open loops",
top_k=8,
web_mode="off",
show_citations=True,
)
print(result.text)
File Locations¶
brain/agent/
harness.py ← Thin façade (public entry point)
turn_budget.py ← TurnBudget, DeadlineToken
turn_state.py ← TurnRuntimeState, CitationAccumulator, ToolTraceEntry
tool_outcomes.py ← Typed outcome union
tool_scheduler.py ← ToolScheduler, ToolWave
replay_control.py ← ReplayControl
artifact_store.py ← ArtifactStore, ArtifactRef
tool_executor_v2.py ← BoundedToolExecutor
turn_preparer.py ← TurnPreparer, TurnPreparation
reflection_engine.py ← ReflectionEngine
turn_finalizer.py ← TurnFinalizer
thinking.py ← ThinkLevel, ThinkingConfig, ThinkingManager
Key Properties¶
- Stateless after construction —
AgentHarnessholds no mutable per-turn state; every turn receives a freshTurnRuntimeState. - Deadline-safe — every tool call has an explicit per-tool deadline derived from the turn budget; late-completing threads are suppressed via the write gate.
- Parallel-safe —
ToolSchedulerensures onlyREAD_ONLYtools with non-overlapping resource keys run concurrently; writes are always sequential. - Idempotency-aware —
ReplayControlallows retries of failed idempotent tools and never suppresses non-idempotent tools. - Typed — all tool outcomes are typed dataclasses (
ToolOutcomeunion), not raw dicts.