Buffering, flushing and asking questions

Lesson 20 called flush without explaining it. This lesson explains it, and then asks the log the questions you would actually ask it.

The recorder batches writes. It holds them in memory and commits when the buffer fills or enough time passes, which makes a busy agent much cheaper to record.

import os, tempfile
from agent_control_plane import AgentKernel, PolicyEngine, FlightRecorder

recorder = FlightRecorder(db_path=os.path.join(tempfile.mkdtemp(), "a.db"))
policy = PolicyEngine(); policy.add_constraint("support", ["lookup_order"])
kernel = AgentKernel(policy_engine=policy, audit_logger=recorder)

kernel.intercept_tool_execution("support", "lookup_order", {})
print("before flush:", len(recorder.query_logs()))
recorder.flush()
print("after flush: ", len(recorder.query_logs()))

The call happened, the decision was made, and for a moment the log did not know about it. A test that checks the log immediately after the call fails for this reason and looks like a bug in the recorder.

The default buffer is a hundred entries and five seconds. Pass enable_batching=False when you would rather every write hit the disk at once, which is the right choice when the log is the thing you are being judged on.

recorder = FlightRecorder(db_path=os.path.join(tempfile.mkdtemp(), "b.db"),
                          enable_batching=False)
kernel = AgentKernel(policy_engine=policy, audit_logger=recorder)
kernel.intercept_tool_execution("support", "lookup_order", {})
print("no batching:", len(recorder.query_logs()))

Filtering

query_logs takes an agent, a verdict and a time range. The verdict strings are the ones the recorder writes: allowed, blocked, shadow, error and pending.

for tool in ["lookup_order", "delete_order", "delete_order"]:
    kernel.intercept_tool_execution("support", tool, {})
recorder.flush()

print("all:    ", len(recorder.query_logs()))
print("blocked:", len(recorder.query_logs(policy_verdict="blocked")))
print("other:  ", len(recorder.query_logs(agent_id="nobody")))

Filtering by an agent that made no calls returns nothing rather than raising, which is the right shape for a dashboard and a trap for a test that forgets to assert on the count.

The summary

stats = recorder.get_statistics()
print("total:", stats["total_actions"])
for verdict in sorted(stats["by_verdict"]):
    print(f"  {verdict:8} {stats['by_verdict'][verdict]}")

One refusal in three calls is the number worth watching. A governed agent whose blocked count is always zero is either well behaved or governed by a policy that forbids nothing, and the summary is where you notice which.

The statistics dictionary is sorted here for the same reason as the rows in lesson 20: nothing promises the order of its keys.