"""The `protocol` tool — start, run and close a protocol from the catalog.
Loading a protocol (`skills(op="load")`) only shows its text. Starting one is
the commitment: its steps are spliced into the session's todo plan after the
agent's current position, the run is logged in ``protocol_runs``, and every
model turn while it is active carries a system reminder that the run is open.
That reminder is the part that survives compression — a protocol the agent has
forgotten about mid-run is worse than no protocol at all.
The tool is deliberately strict about closing: `finish` refuses while any of
the run's steps is still open, and it requires a report. `cancel`/`fail`
require a reason and close the remaining steps so the plan does not keep
dangling work that was never done.
"""
from __future__ import annotations
import structlog
from navi.llm.base import Message
from navi.skills import ProtocolRunStore, SkillError, SkillStore
from navi.tools._internal.base import (
Tool,
ToolContext,
ToolResult,
current_session_id,
current_user_id,
)
from navi.tools.todo import (
close_protocol_steps,
insert_steps,
protocol_step_states,
todo_session_id,
)
log = structlog.get_logger()
_run_store: ProtocolRunStore | None = None
def set_run_store(store: ProtocolRunStore) -> None:
"""Inject the run store — called once at startup (container wiring)."""
global _run_store
_run_store = store
def _store() -> ProtocolRunStore | None:
return _run_store
async def _active(store: ProtocolRunStore, session_id: str):
"""``store.active`` with the database error turned into text for the model."""
try:
return await store.active(session_id), ""
except Exception as e:
log.warning("protocol.active_failed", error=str(e))
return None, f"Cannot read the protocol run: {e}"
class ProtocolTool(Tool):
name = "protocol"
description = (
"Run a protocol — a strict, pre-agreed order of actions from the catalog. "
"Loading one only shows its text; 'start' is what commits you: its steps are "
"inserted into the todo plan after your current position and must be executed in "
"order, each one checked before the next.\n\n"
"'start' with the protocol id or exact name once the user asks for it or its "
"triggers match. One protocol runs at a time per session.\n"
"'finish' only after every injected step is done or skipped — it requires a report "
"saying how each post-check was verified. If reality differs from the plan, 'fail' "
"with the reason and the state the system is left in. If the run cannot continue, "
"'cancel' with the reason. Never close a run as done to avoid a step, and never "
"improvise around one: verify it and mark it done, or report the deviation.\n\n"
"'status' shows the active run. While a run is open, every turn carries a reminder "
"of it — do not end the turn with a protocol still running."
)
parameters = {
"type": "object",
"properties": {
"op": {
"type": "string",
"enum": ["start", "finish", "cancel", "fail", "status"],
"description": (
"start — begin a protocol; finish — close it as completed (report "
"required); cancel — stop it without completing (reason required); "
"fail — reality diverged from the plan (reason required); status — "
"show the active run."
),
},
"id": {
"type": "string",
"description": "start: the protocol's catalog id from skills(op='search').",
},
"name": {
"type": "string",
"description": "start: the exact protocol name, when you have no id.",
},
"report": {
"type": "string",
"description": (
"finish: mandatory. What was done and how each post-check was verified. "
"The user reads this as the run's record."
),
},
"reason": {
"type": "string",
"description": (
"cancel/fail: mandatory. Which step stopped the run, why, and what "
"state the system is in now."
),
},
},
"required": ["op"],
}
def __init__(self, skill_store: SkillStore, run_store: ProtocolRunStore | None = None) -> None:
self._skills = skill_store
if run_store is not None:
set_run_store(run_store)
async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult:
op = params.get("op")
session_id = todo_session_id(
(ctx.session_id if ctx else None) or current_session_id.get(None)
)
user_id = current_user_id.get(None) or (ctx.user_id if ctx else None) or ""
store = _store()
if store is None:
return ToolResult(
success=False, output="",
error="Protocol runs are not available on this instance (no store wired).",
)
if op == "start":
return await self._start(params, store, session_id, user_id)
if op == "finish":
return await self._finish(params, store, session_id)
if op == "cancel":
return await self._abort(params, store, session_id, status="cancelled")
if op == "fail":
return await self._abort(params, store, session_id, status="failed")
if op == "status":
return await self._status(store, session_id)
return ToolResult(
success=False, output="",
error=(
f"Unknown op: {op!r}. Use 'start', 'finish', 'cancel', 'fail' or 'status'."
),
)
# ── start ───────────────────────────────────────────────────────────────
async def _start(
self, params: dict, store: ProtocolRunStore, session_id: str, user_id: str
) -> ToolResult:
entry_id = (params.get("id") or "").strip()
name = (params.get("name") or "").strip()
if not entry_id and not name:
return ToolResult(
success=False, output="",
error="'id' or 'name' is required for 'start'",
)
try:
entry = await self._skills.find(
user_id=user_id, skill_id=entry_id or None, name=name or None, kind="protocol",
)
except Exception as e:
log.warning("protocol.start_find_failed", error=str(e))
return ToolResult(success=False, output="", error=f"Cannot read the catalog: {e}")
if entry is None:
return ToolResult(
success=False, output="",
error=(
f"No protocol {entry_id or name!r}. Find it with "
"skills(op='search', kind='protocol') first."
),
)
try:
doc = entry.doc
except SkillError as e:
return ToolResult(
success=False, output="",
error=f"The stored protocol is invalid and cannot be run: {e}",
)
active, error = await _active(store, session_id)
if error:
return ToolResult(success=False, output="", error=error)
if active is not None:
return ToolResult(
success=False, output="",
error=(
f"Protocol {active.protocol_name!r} is already running "
f"(run={active.id[:8]}). Close it first — protocol(op='finish', "
"report=...) if its steps are done, or protocol(op='cancel'/'fail', "
"reason=...) — then start this one."
),
)
try:
run = await store.start(
protocol_id=entry.id,
protocol_name=entry.name,
protocol_version=entry.version,
session_id=session_id,
user_id=user_id,
steps=doc.steps,
post_checks=doc.post_checks,
)
except Exception as e:
# The partial unique index refuses a second `running` row — two
# starts racing (or a run the check above missed) land here.
log.warning("protocol.start_failed", error=str(e))
return ToolResult(
success=False, output="",
error=(
f"Could not open the protocol run: {e}. Check protocol(op='status') "
"for the active run."
),
)
indexes = await insert_steps(
session_id,
doc.steps,
source="protocol",
protocol_run_id=run.id,
step_ids=[f"s{i}" for i in range(1, len(doc.steps) + 1)],
)
if not indexes:
# Defensive: an empty step list would leave a run that can never be
# finished honestly and would block every later start.
await store.close(
run.id, status="failed",
reason="the protocol has no steps to add to the plan",
)
return ToolResult(
success=False, output="",
error=(
"The protocol has no steps, so the run was closed as failed. "
"Fix the document in the catalog and propose it again."
),
)
lines = [
f"PROTOCOL STARTED — {entry.name} (v{entry.version}, run={run.id[:8]})",
f"Steps added to the plan at positions {indexes[0]}–{indexes[-1]}.",
]
if doc.preconditions:
lines.append("")
lines.append(f"Preconditions — check these before step {indexes[0]}:")
lines.extend(f"- {p}" for p in doc.preconditions)
lines.append("")
lines.append(
"Steps — mark each done with todo(op='update') only after its check passes:"
)
check_hint = "after each step: " + "; ".join(doc.post_checks)
lines.extend(
f" {idx}. {text} [{check_hint}]"
for idx, text in zip(indexes, doc.steps, strict=False)
)
lines.append("")
lines.append("Post-checks the finish report must cover:")
lines.extend(f"- {c}" for c in doc.post_checks)
lines.append("")
lines.append("If the run goes off plan: " + doc.on_deviation)
lines.append("If the protocol cannot be completed: " + doc.on_blocked)
lines.append("If a step is already done: " + doc.on_already_done)
lines.append("")
lines.append(
"Close with protocol(op='finish', report=...) once every step above is done "
"or skipped. Do not narrate around a step: run it, verify it, record it."
)
return ToolResult(
success=True,
output="\n".join(lines),
metadata={
"run_id": run.id,
"protocol_id": entry.id,
"name": entry.name,
"version": entry.version,
"step_indexes": indexes,
},
)
# ── finish / cancel / fail / status ─────────────────────────────────────
async def _finish(
self, params: dict, store: ProtocolRunStore, session_id: str
) -> ToolResult:
run, error = await _active(store, session_id)
if error:
return ToolResult(success=False, output="", error=error)
if run is None:
return ToolResult(
success=False, output="",
error="No protocol is running in this session. Start one with op='start'.",
)
report = (params.get("report") or "").strip()
if not report:
return ToolResult(
success=False, output="",
error=(
"A protocol closes only with a report: pass 'report' describing what "
"was done and how each post-check was verified. It is the run's record "
"for the user."
),
)
states = await protocol_step_states(session_id, run.id)
missing = len(run.steps) - len(states)
if missing > 0:
return ToolResult(
success=False, output="",
error=(
f"The plan no longer contains all of this run's steps ({len(states)} of "
f"{len(run.steps)}). The plan was replaced or cleared mid-run. Restore "
"it, or close the run with protocol(op='fail', reason=...) saying so."
),
)
failed = [s for s in states if s["status"] == "failed"]
if failed:
listing = ", ".join(
f"{s['index']} ({s['text']})" for s in failed
)
return ToolResult(
success=False, output="",
error=(
f"Cannot finish: step {listing} is marked failed. A run with a failed "
"step ends with protocol(op='fail', reason=...) — say what broke and "
"what state the system is in."
),
)
open_steps = [s for s in states if s["status"] not in ("done", "skipped")]
if open_steps:
listing = "; ".join(
f"{s['index']} {s['status']} ({s['text']})" for s in open_steps
)
return ToolResult(
success=False, output="",
error=(
f"Cannot finish: these steps are still open — {listing}. Complete and "
"verify them first, or close the run with protocol(op='cancel'/'fail', "
"reason=...)."
),
)
closed = await store.close(run.id, status="done", report=report)
run = closed or run
return ToolResult(
success=True,
output=(
f"Protocol {run.protocol_name!r} finished (run={run.id[:8]}). "
"The report is stored with the run; give the user a short summary of it."
),
metadata={
"run_id": run.id,
"protocol_id": run.protocol_id,
"status": "done",
},
)
async def _abort(
self, params: dict, store: ProtocolRunStore, session_id: str, *, status: str
) -> ToolResult:
run, error = await _active(store, session_id)
if error:
return ToolResult(success=False, output="", error=error)
if run is None:
return ToolResult(
success=False, output="",
error="No protocol is running in this session.",
)
reason = (params.get("reason") or "").strip()
if not reason:
word = "cancel" if status == "cancelled" else "fail"
return ToolResult(
success=False, output="",
error=(
f"'{word}' requires a reason: which step stopped the run, why, and what "
"state the system is in now."
),
)
note = (
f"protocol cancelled: {reason}" if status == "cancelled"
else f"protocol failed: {reason}"
)
skipped = await close_protocol_steps(session_id, run.id, note=note)
closed = await store.close(
run.id,
status=status,
report="",
reason=reason,
)
run = closed or run
skipped_note = (
f" Open steps {', '.join(str(i) for i in skipped)} were marked skipped."
if skipped else ""
)
verb = "cancelled" if status == "cancelled" else "failed"
return ToolResult(
success=True,
output=(
f"Protocol {run.protocol_name!r} {verb} (run={run.id[:8]}).{skipped_note} "
"Report the outcome to the user now: what happened, at which step, and "
"what state things are left in."
),
metadata={
"run_id": run.id,
"protocol_id": run.protocol_id,
"status": status,
"skipped_indexes": skipped,
},
)
async def _status(self, store: ProtocolRunStore, session_id: str) -> ToolResult:
run, error = await _active(store, session_id)
if error:
return ToolResult(success=False, output="", error=error)
if run is None:
return ToolResult(success=True, output="No protocol is running in this session.")
states = await protocol_step_states(session_id, run.id)
started = (
f", started {run.started_at:%Y-%m-%d %H:%M} UTC" if run.started_at else ""
)
lines = [
f"ACTIVE PROTOCOL — {run.protocol_name} (v{run.protocol_version}, "
f"run={run.id[:8]}){started}.",
]
lines.extend(
f" {_STATUS_ICON.get(s['status'], '?')} {s['index']}. {s['text']} ({s['status']})"
for s in states
)
if not states:
lines.append(" (none of this run's steps remain in the plan)")
lines.append("Close with protocol(op='finish'/'cancel'/'fail').")
return ToolResult(
success=True,
output="\n".join(lines),
metadata={"run_id": run.id, "status": "running"},
)
_STATUS_ICON = {
"pending": "○",
"in_progress": "◎",
"done": "✓",
"failed": "✗",
"skipped": "—",
}
# ── Per-turn reminder ───────────────────────────────────────────────────────
async def has_active_run(session_id: str) -> bool:
"""True while a protocol run is open in this session (for the final-turn
intercept in ``navi.core.agent``: a run whose steps are all closed is still
unfinished until it is finished or aborted)."""
store = _store()
if store is None or not session_id:
return False
try:
return await store.active(session_id) is not None
except Exception:
return False
async def get_protocol_message(session_id: str) -> Message | None:
"""System reminder for an active protocol run, or None.
Injected on every model turn (``navi.core.agent``), next to the todo
reminder: compression can drop the turn where the protocol was started, and
the run's obligations must not disappear with it.
"""
store = _store()
if store is None or not session_id:
return None
try:
run = await store.active(session_id)
if run is None:
return None
states = await protocol_step_states(session_id, run.id)
except Exception as e: # a broken reminder must not break the turn
log.warning("protocol.reminder_failed", error=str(e))
return None
lines = [
f"PROTOCOL ACTIVE — {run.protocol_name} (v{run.protocol_version}, "
f"run={run.id[:8]}): {len(states)} of {len(run.steps)} steps in the plan.",
]
if not states:
lines.append(
"This run's steps are no longer in the plan (it was replaced or cleared). "
"Close the run: protocol(op='fail', reason=...) with what happened."
)
return Message(role="system", content="\n".join(lines))
open_steps = [s for s in states if s["status"] not in ("done", "skipped")]
lines.extend(
f" {_STATUS_ICON.get(s['status'], '?')} {s['index']}. {s['text']} ({s['status']})"
for s in states
)
if open_steps:
lines.append(
"Execute these in order; mark each done only after its check passes. "
"If reality differs from the plan, protocol(op='fail', reason=...); if the run "
"cannot continue, protocol(op='cancel', reason=...). Do not improvise around a "
"step and do not end the turn leaving the protocol running."
)
else:
lines.append(
"Every step is closed — call protocol(op='finish', report=...) now, with how "
"each post-check was verified."
)
if run.post_checks:
lines.append("Post-checks to cover in the report: " + "; ".join(run.post_checks))
return Message(role="system", content="\n".join(lines))
