mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 15:41:15 +00:00
52cafa6f8e
- Nudge text now warns that repeated protocol violations will block the task and require manual intervention, so the model understands the consequence of ignoring the nudge. - Kanban docs restructured to clearly separate the two defense layers: agent-side prevention (nudge, from #64350) and dispatcher-side recovery (bounded retry, from this PR). - Two new integration tests verifying the nudge mentions blocking and that the agent-side and dispatcher-side budgets are independent.
108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
"""Turn-end guard for kanban workers.
|
|
|
|
Kanban workers must end with ``kanban_complete`` or ``kanban_block``. Models
|
|
(especially GLM / Qwen families) sometimes narrate the next step
|
|
("Let me write the report now") and stop with ``finish_reason=stop`` and no
|
|
tool calls. Hermes treats that as a clean exit → ``rc=0`` → dispatcher
|
|
``protocol_violation``.
|
|
|
|
This module is policy-only: when a kanban worker tries to finish without a
|
|
terminal board tool, return a bounded synthetic nudge so the conversation
|
|
loop continues instead of exiting.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any, Iterable, Optional
|
|
|
|
|
|
_TERMINAL_KANBAN_TOOLS = frozenset({"kanban_complete", "kanban_block"})
|
|
|
|
_DEFAULT_MAX_ATTEMPTS = 2
|
|
|
|
|
|
def kanban_stop_nudge_enabled() -> bool:
|
|
"""Return whether the kanban stop-guard is active for this process.
|
|
|
|
On when ``HERMES_KANBAN_TASK`` is set (dispatcher-spawned worker), unless
|
|
``HERMES_KANBAN_STOP_NUDGE`` explicitly disables it.
|
|
"""
|
|
env = os.environ.get("HERMES_KANBAN_STOP_NUDGE")
|
|
if env is not None and env.strip().lower() in {"0", "false", "no", "off"}:
|
|
return False
|
|
task = (os.environ.get("HERMES_KANBAN_TASK") or "").strip()
|
|
return bool(task)
|
|
|
|
|
|
def _tool_call_name(tc: Any) -> str:
|
|
if isinstance(tc, dict):
|
|
fn = tc.get("function")
|
|
if isinstance(fn, dict):
|
|
return str(fn.get("name") or "")
|
|
return str(tc.get("name") or "")
|
|
fn = getattr(tc, "function", None)
|
|
if fn is not None:
|
|
return str(getattr(fn, "name", "") or "")
|
|
return str(getattr(tc, "name", "") or "")
|
|
|
|
|
|
def session_called_kanban_terminal(messages: Iterable[dict] | None) -> bool:
|
|
"""True if this conversation already invoked a terminal kanban tool."""
|
|
if not messages:
|
|
return False
|
|
for msg in messages:
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
role = msg.get("role")
|
|
if role == "assistant":
|
|
for tc in msg.get("tool_calls") or []:
|
|
if _tool_call_name(tc) in _TERMINAL_KANBAN_TOOLS:
|
|
return True
|
|
elif role == "tool":
|
|
name = str(msg.get("name") or "")
|
|
if name in _TERMINAL_KANBAN_TOOLS:
|
|
return True
|
|
return False
|
|
|
|
|
|
def build_kanban_stop_nudge(
|
|
*,
|
|
messages: Iterable[dict] | None = None,
|
|
attempts: int = 0,
|
|
max_attempts: int = _DEFAULT_MAX_ATTEMPTS,
|
|
task_id: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""Return a synthetic follow-up when a kanban worker exits without a terminal tool.
|
|
|
|
Returns ``None`` when the guard should not fire (not a kanban worker,
|
|
already completed/blocked, or nudge budget exhausted).
|
|
"""
|
|
if not kanban_stop_nudge_enabled():
|
|
return None
|
|
if attempts >= max_attempts:
|
|
return None
|
|
if session_called_kanban_terminal(messages):
|
|
return None
|
|
|
|
tid = (task_id or os.environ.get("HERMES_KANBAN_TASK") or "").strip() or "this task"
|
|
return (
|
|
"[System: You are a Hermes kanban worker. A plain-text reply is NOT a "
|
|
"terminal state for the board.\n\n"
|
|
f"Task `{tid}` is still `running`. Ending now without a board tool "
|
|
"causes a protocol violation (clean exit with no "
|
|
"`kanban_complete` / `kanban_block`).\n\n"
|
|
"Do this immediately in your next response — do not narrate intent:\n"
|
|
"1. Finish any remaining deliverable (write the required file(s) now).\n"
|
|
"2. Call `kanban_complete(summary=..., artifacts=[...])` if the work "
|
|
"is done, OR `kanban_block(reason=...)` if you are blocked.\n\n"
|
|
"Never end a turn with only a promise of future action. Repeated "
|
|
"protocol violations will block this task and require manual intervention.]"
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"build_kanban_stop_nudge",
|
|
"kanban_stop_nudge_enabled",
|
|
"session_called_kanban_terminal",
|
|
]
|