Robert Helewka
ea37ab38c1
Introduces `pallas.loop_guard` module that detects and halts agentic loops
where the same `(tool, args) → result` repeats consecutively, preventing
wasted LLM turns when upstream MCP servers return contradictory data.
- Add per-request `ToolRunnerHooks` tracking rolling tool-call signatures
- Halt loop after `loop_repeat_threshold` consecutive repeats (default 3)
- Collapse `max_iterations` on halt to terminate without further LLM call
- Append user-facing explanation to the turn with `stop_reason=endTurn`
- Expose `pallas_agent_loop_aborted_total{agent,reason}` counter
- Add per-agent `max_iterations` and `loop_repeat_threshold` config
- Document guard behavior, metric, and alerting query
56 lines
1.6 KiB
Python
56 lines
1.6 KiB
Python
"""Tests for ``pallas.log._JSONFormatter``.
|
|
|
|
The formatter must serialize caller-supplied ``extra={...}`` fields (the
|
|
loop guard and other diagnostics rely on this) while never emitting the
|
|
internal ``LogRecord`` bookkeeping attributes.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
|
|
from pallas.log import _JSONFormatter
|
|
|
|
|
|
def _format(msg: str, **extra) -> dict:
|
|
record = logging.LogRecord(
|
|
name="pallas.test",
|
|
level=logging.WARNING,
|
|
pathname=__file__,
|
|
lineno=1,
|
|
msg=msg,
|
|
args=(),
|
|
exc_info=None,
|
|
)
|
|
for key, value in extra.items():
|
|
setattr(record, key, value)
|
|
return json.loads(_JSONFormatter().format(record))
|
|
|
|
|
|
def test_extra_fields_are_serialized():
|
|
out = _format(
|
|
"agentic loop halted",
|
|
event="loop_halt",
|
|
tool="kairos-update_task",
|
|
repeat_count=3,
|
|
result_preview="COMPLETED but 0%",
|
|
)
|
|
assert out["message"] == "agentic loop halted"
|
|
assert out["event"] == "loop_halt"
|
|
assert out["tool"] == "kairos-update_task"
|
|
assert out["repeat_count"] == 3
|
|
assert out["result_preview"] == "COMPLETED but 0%"
|
|
|
|
|
|
def test_standard_attributes_are_not_leaked():
|
|
out = _format("plain message")
|
|
for noise in ("msg", "args", "levelno", "pathname", "lineno", "funcName"):
|
|
assert noise not in out
|
|
assert out["level"] == "WARNING"
|
|
assert out["logger"] == "pallas.test"
|
|
|
|
|
|
def test_non_serializable_extra_does_not_crash():
|
|
out = _format("with object", obj=object())
|
|
assert "obj" in out # coerced via default=str, not dropped or raised
|