Custom agent loops - PandaProbe

The harness needs exactly three touchpoints with your loop, whatever it is built with: the system context in the prompt, the toolset among the tools, and a turn-end signal. No adapter required.

from pandaprobe_harness import Harness
from pandaprobe_harness.agent_tools.native import as_anthropic_tools

harness = Harness.create()

Harness.create() provisions the workspace, wires every component (hook, mailbox, journal, rules, eval set, toolset, restricted shell), and schedules the startup health check. It accepts an explicit HarnessConfig and, for the closed loop, your replay function:

from pandaprobe_harness import Harness, HarnessConfig

harness = Harness.create(
    HarnessConfig.from_env(capture_eval_cases=True),
    replay=my_replay_fn,
)

1. The system context

system_prompt = harness.system_context() + MY_PROMPT

Rebuild it each turn if you can — the mailbox banner and the retrieved rules are current-state. Pass a task hint to sharpen rule retrieval:

system_prompt = harness.system_context(task_hint=user_input) + MY_PROMPT

For static-prompt setups, the standing protocol’s “check your mailbox at the start of each turn” instruction is the trigger instead of the banner.

2. The tools

specs, dispatch = as_anthropic_tools(harness.toolset)
tools = my_tools + specs
# on a harness tool call: result = await dispatch(name, args)

as_langchain_tools and as_openai_function_tools cover the other native formats, and the pandaprobe-harness-agent companion CLI covers sandboxed shells — see the toolset.

3. The turn boundary

Three equivalent styles — pick whichever fits your loop:

# Async context manager (fires turn-end even on exceptions):
async with harness.turn(session_id):
    await my_agent_step(...)

# One-shot runner:
result = await harness.run_turn(session_id, my_agent_step, ...)

# Decorator:
step = harness.turn(session_id)(my_agent_step)

Or call the hook directly with a raw payload — the end_state you pass is what a captured eval case will carry as its replay input:

harness.on_turn_end({
    "session_id": session_id,
    "turn_index": turn_index,
    "end_state": {"task": user_input},
})

Use the same session id for the harness turn and the SDK trace context (pandaprobe.session(session_id)) — the evaluation scores whatever traces landed under that session.

Determinism helpers

Everything after on_turn_end is detached and non-blocking. When a test or script needs to observe results deterministically:

await harness.refresh(session_id)      # join this session's in-flight evaluation (bounded)
await harness.refresh_all()            # join every in-flight evaluation
await harness.drain_validation()       # join in-flight candidate validation

These are bounded waits (drain_timeout_s) for callers’ convenience — correctness never depends on them; each background task handles its own result.

Everything on the facade

Surface	Purpose
`harness.system_context(task_hint=None)`	The prompt preamble: rules + protocol + banner.
`harness.toolset` / `harness.shell`	The 14 operations; the restricted shell for sandboxed delivery.
`harness.turn()` / `run_turn()` / `on_turn_end()`	Turn boundaries.
`harness.refresh()` / `refresh_all()` / `drain_validation()`	Bounded joins.
`harness.run_regression(sample=None)`	Replay the eval set against the current rules.
`harness.validate_candidates()`	Run one candidate-validation round now.
`harness.mailbox` / `journal` / `rules` / `evalset` / `config`	Direct store access for inspection and scripting.
`harness.check_health()`	The memoized CLI/auth probe.

​1. The system context

​2. The tools

​3. The turn boundary

​Determinism helpers

​Everything on the facade

1. The system context

2. The tools

3. The turn boundary

Determinism helpers

Everything on the facade