Continuous pressure signal for long or failure-heavy agent trajectories.
(messages: list[Any], step_risk: str)
| 1664 | |
| 1665 | |
| 1666 | def _agent_state_pressure(messages: list[Any], step_risk: str) -> tuple[int, float]: |
| 1667 | """Continuous pressure signal for long or failure-heavy agent trajectories.""" |
| 1668 | tool_steps = 0 |
| 1669 | failure_steps = 0 |
| 1670 | for msg in messages: |
| 1671 | if not isinstance(msg, dict): |
| 1672 | continue |
| 1673 | if msg.get("tool_calls"): |
| 1674 | tool_steps += 1 |
| 1675 | tool_text, tool_is_error = _latest_tool_result_signal([msg]) |
| 1676 | if msg.get("role") == "tool" or tool_text: |
| 1677 | tool_steps += 1 |
| 1678 | command = "" |
| 1679 | if tool_is_error or _contains_tool_failure_signal(tool_text, command): |
| 1680 | failure_steps += 1 |
| 1681 | |
| 1682 | step_component = min(0.45, max(0, tool_steps) / 24.0) |
| 1683 | failure_component = min(0.35, failure_steps / 6.0) |
| 1684 | risk_component = 0.20 if str(step_risk or "").lower() == "high" else 0.0 |
| 1685 | return tool_steps, min(1.0, step_component + failure_component + risk_component) |
| 1686 | |
| 1687 | |
| 1688 | def _extract_routing_features( |
no test coverage detected