diff --git a/docs/skills.md b/docs/skills.md index 4e67436..b63d5d9 100644 --- a/docs/skills.md +++ b/docs/skills.md @@ -74,7 +74,7 @@ ## Storage -Two tables (PostgreSQL, lazy DDL on first use): +Four tables (PostgreSQL, lazy DDL on first use): ```sql CREATE TABLE skills ( @@ -98,6 +98,17 @@ skill_id TEXT, error TEXT, created_at TIMESTAMPTZ, decided_at TIMESTAMPTZ, decided_by TEXT ); +CREATE TABLE protocol_runs ( + id TEXT PRIMARY KEY, + protocol_id TEXT, protocol_name TEXT, protocol_version INTEGER, + session_id TEXT, user_id TEXT, + status TEXT, -- 'running' | 'done' | 'failed' | 'cancelled' + steps JSONB, post_checks JSONB, -- snapshot at start + report TEXT, reason TEXT, + started_at TIMESTAMPTZ, finished_at TIMESTAMPTZ +); +CREATE UNIQUE INDEX idx_protocol_runs_one_active + ON protocol_runs (session_id) WHERE status = 'running'; ``` - **Ownership**: personal rows have `owner_id = `; shared rows have @@ -156,9 +167,43 @@ Pending proposals do not show up in search or `skills(op="load")` — a document the user has not agreed to must never reach another turn as a methodology. -## Agent tool +## Runs -`skills` — `search`, `load`, `create`, `update`, `propose`: +Loading a protocol only shows its text. Executing one is the `protocol` tool: + +| op | what it does | +|---|---| +| `start` | opens a run and splices the steps into the todo plan after the agent's current position | +| `finish` | closes the run as `done`; **requires a report** and refuses while any step of the run is open | +| `cancel` | closes it as `cancelled` with a reason; still-open run steps become `skipped` | +| `fail` | closes it as `failed` with a reason; a run with a failed step can only end this way | +| `status` | the active run and the state of its steps | + +A run row snapshots the protocol's `version`, `steps` and `post_checks` at +start, so editing the catalog entry mid-run cannot change what the running +agent owes the user; the log still says which version was executed. One run per +session is `running` at a time (a partial unique index enforces it). + +The steps are ordinary todo tasks, tagged with `source="protocol"`, +`protocol_run_id` and a stable `step_id`; `insert_steps` keeps the agent's own +progress and statuses intact. Marking one `done` still goes through +`todo(op="update")` and still needs a `validation`. + +While a run is `running`, every model turn gets a system reminder with the +protocol, its steps and the way to close it (`navi/core/agent.py`, next to the +todo reminder). This is what survives compression: a run started ten turns ago +is otherwise invisible to the model once the todo lines lose their context. The +final-turn intercept also fires when a run is open even though all its steps are +closed — a bare-text turn cannot leave it running. + +The log is a row per run with `report`/`reason`; there is no separate UI yet — +`protocol(op="status")` and the table are the current view. + +## Agent tools + +### `skills` + +`search`, `load`, `create`, `update`, `propose`: ```json {"op": "search", "query": "выключение", "area": "infrastructure"} @@ -172,7 +217,21 @@ post-checks and the three escape hatches; for a skill, the instructions. `create` and `update` accept skills only — a protocol document sent there is refused with the reason, and `update` of a protocol points at `propose` with `replaces_id`. -The tool is available to `assistant`, `developer`, `server_admin` and `discuss`. + +### `protocol` + +`start`, `finish`, `cancel`, `fail`, `status` — the run machinery described +above: + +```json +{"op": "start", "id": "9c4d2b7a1e6f8035"} +{"op": "finish", "report": ""} +{"op": "fail", "reason": ""} +``` + +Both tools are available to `assistant`, `developer`, `server_admin` and +`discuss`; sub-agents get neither, because a protocol belongs to the session's +plan, not to a delegated run. ## Settings tab @@ -184,7 +243,7 @@ ## Not implemented yet -- `protocol_start` / `protocol_finish` / `protocol_cancel` / `protocol_fail` and - run logging. - Hive synchronization of shared entries (schema is ready: `content_hash`, `version`). +- A run-history view in the webclient (`protocol_runs` is written, but the tab + shows the catalog only). diff --git a/docs/todo-store.md b/docs/todo-store.md index 5837e49..1b94c85 100644 --- a/docs/todo-store.md +++ b/docs/todo-store.md @@ -92,11 +92,20 @@ ## Helpers for protocols -`navi/tools/todo.py` exposes two internals that the protocol work builds on: +`navi/tools/todo.py` exposes the plan surgery the `protocol` tool builds on +(see [skills.md](skills.md#runs)): - `insert_steps(session_id, steps, after=None, source="protocol", protocol_run_id=…, step_ids=…)` splices a block into the existing plan after the last step the agent reached (or after an explicit index) and returns the new 1-based indexes. Existing steps and statuses are preserved — no re-issue of the whole plan via `set`. +- `protocol_step_states(session_id, protocol_run_id)` lists the steps a run + injected, with their 1-based plan indexes and statuses — what `finish` + validates against. +- `close_protocol_steps(session_id, protocol_run_id, note=…)` marks a run's + still-open steps `skipped` (used by `cancel`/`fail`) and returns their indexes. +- `todo_session_id(explicit=None)` is the public form of `_sid` — any tool that + touches the plan must resolve the session id the same way, or a sub-agent's + call would edit the parent's plan. - `_current_position(tasks)` finds that insertion point (last `done` / `in_progress` / `failed` step). diff --git a/docs/tools.md b/docs/tools.md index fa3fd3e..cffcf47 100644 --- a/docs/tools.md +++ b/docs/tools.md @@ -43,6 +43,8 @@ | `SshExecTool` | `ssh_exec` | SSH exec and SCP file transfer; connection pool keyed by session ID | | `ImageViewTool` | `image_view` | Load image from path/URL → resize to 1024px, convert to JPEG, return base64 for multimodal LLM | | `TodoTool` | `todo` | Per-session task checklist (set/update/read) | +| `SkillTool` | `skills` | Skills and protocols catalog: `search`/`load`, plus `create`/`update` (skills) and `propose` (protocols, user accepts) — see [`skills.md`](skills.md) | +| `ProtocolTool` | `protocol` | Run a catalog protocol: `start` splices its steps into the todo plan, `finish`/`cancel`/`fail` close the run with a report or reason; `status` reads it — see [`skills.md`](skills.md#runs) | | `ScratchpadTool` | `scratchpad` | Per-session named working notes (write/append/read/clear) | | `ReloadToolsTool` | `reload_tools` | Hot-reload user tools and context providers without server restart | | `ListToolsTool` | `list_tools` | Tools a profile can actually use, grouped by source (`native`, then one section per `mcp____`). Defaults to the profile the run executes as (`profile_id` overrides it); `scope` selects that profile's `agent` or `subagent` toolset. Names only by default; `query` narrows by substring over names and descriptions, `verbose` adds descriptions. Declared-but-unregistered tools (server not connected) are listed separately as "Not registered" instead of being advertised | diff --git a/manuals/protocol.md b/manuals/protocol.md new file mode 100644 index 0000000..cce0a91 --- /dev/null +++ b/manuals/protocol.md @@ -0,0 +1,59 @@ +# protocol — Manual + +## What it does +Runs a **protocol** from the catalog: a strict, pre-agreed order of actions. Loading one (`skills(op="load")`) only shows its text. `start` is the commitment — its steps are inserted into the todo plan after your current position, the run is logged, and every turn afterwards carries a reminder that it is still open. + +A protocol is not advice. Its steps are executed in order, each one checked before the next; deviating is an execution error to report, not a variation to choose. + +## Parameters + +| Parameter | Required | Description | +|-----------|----------|-------------| +| `op` | yes | `start` \| `finish` \| `cancel` \| `fail` \| `status`. | +| `id` | for `start` | Protocol id from `skills(op="search", kind="protocol")`. | +| `name` | for `start` | Exact protocol name, when you have no id. | +| `report` | for `finish` | **Mandatory.** What was done and how each post-check was verified. | +| `reason` | for `cancel`/`fail` | **Mandatory.** Which step stopped the run, why, and what state the system is in. | + +## Actions + +### `start` +```json +{"op": "start", "id": "9c4d2b7a1e6f8035"} +{"op": "start", "name": "Корректное выключение инфраструктуры"} +``` +The result names the plan positions the steps landed on, the preconditions, the post-checks, and the three escape hatches. **One protocol runs at a time per session** — close the active one before starting another. + +Start a protocol when the user asks for it by name, or when its `triggers` match what they asked for. Do not start one to "see what it says" — `load` does that without committing to anything. + +### `finish` +```json +{"op": "finish", "report": "Stopped apps, then databases; systemctl shows no app units; disk synced and unmounted."} +``` +Refused unless every injected step is `done` or `skipped`, and refused outright if any step is `failed` — that run ends with `fail`. Skipping a step is legitimate only through `on_already_done` (verify the result exists, then mark it done or skipped with validation). + +The report is the run's record: name the post-checks and what you saw for each. + +### `cancel` +```json +{"op": "cancel", "reason": "user changed their mind before step 2; nothing was stopped"} +``` +Use when the run cannot or should not continue. Still-open steps are marked `skipped` so the plan does not keep dangling work. Then tell the user what was and was not done. + +### `fail` +```json +{"op": "fail", "reason": "step 2: postgres did not stop within 60 s; apps are still down, databases up"} +``` +Use when reality diverged from the plan — a step failed, a precondition was false, the result is not what the protocol expected. Same step cleanup as `cancel`. This is the honest end of a run that did not go as agreed; never report a failed run as finished. + +### `status` +```json +{"op": "status"} +``` +Shows the active run and the state of its steps. Useful after a long tool chain or a compressed context. + +## Rules +- Mark each step `done` with `todo(op="update")` **only after its check passes**; the validation field is required. +- Do not reorder, merge, or silently skip steps. If a step is already done, follow `on_already_done`. +- Do not end the turn while a protocol is running: finish it, or cancel/fail it with the reason. +- The user reads the run's outcome as your report — not as a claim that the protocol "went fine". diff --git a/manuals/skills.md b/manuals/skills.md index f92b940..5258a43 100644 --- a/manuals/skills.md +++ b/manuals/skills.md @@ -66,6 +66,8 @@ Do not silently improvise around a step. A protocol exists because the user pre-agreed to the exact actions it names; deviating is an execution error to report, not a variation to choose. +To **run** a loaded protocol, use the `protocol` tool (`protocol(op="start", id=…)`), not this one — `tool_manual("protocol")` has the rules for starting, finishing and closing a run. + ## Format (for the catalog editor, not the tool) A document is a YAML front matter block plus a markdown body: diff --git a/navi/core/agent.py b/navi/core/agent.py index a1c49bf..3dccbff 100644 --- a/navi/core/agent.py +++ b/navi/core/agent.py @@ -143,10 +143,23 @@ return await get_progress_message(session_id, first_iteration=first_iteration) +async def _protocol_progress_message(session_id: str) -> "Message | None": + """System reminder for an active protocol run, or None. + + The run was started in some earlier turn; compression may drop that turn and + the todo lines no longer say which of them the user pre-agreed to. This + reminder keeps the commitment — and the way to close it — in view. + """ + from navi.tools.protocol import get_protocol_message + + return await get_protocol_message(session_id) + + # Final-turn intercept nudges — injected when the model ends a turn with bare -# text (no tool call) while the todo still has open steps, then the loop is -# continued instead of closed. Escalating tone: 1st is a soft pointer to the -# available recovery tools, 2nd is a hard "act now, no bare text" instruction. +# text (no tool call) while the todo still has open steps (or a protocol run is +# still open), then the loop is continued instead of closed. Escalating tone: +# 1st is a soft pointer to the available recovery tools, 2nd is a hard "act +# now, no bare text" instruction. # Indexed by turn_ctx.final_interceptions (clamped to the last/hardest). _FINAL_INTERCEPT_NUDGES = ( "[Final-turn check] You ended the turn with text only, but the todo list still has open steps " @@ -159,6 +172,14 @@ "text without a tool call.", ) +# Same intercept, different move: the todo shows no open work, but a protocol +# run is still `running` because it was never closed. +_PROTOCOL_FINAL_NUDGE = ( + "[Final-turn check] You ended the turn with text only while a protocol run is still " + "open. Close it now: protocol(op='finish', report=...) if its steps are done, or " + "protocol(op='cancel'/'fail', reason=...) if not. Do not leave a run running." +) + class Agent: def __init__( @@ -583,6 +604,10 @@ if todo_msg: built_ctx.append(todo_msg) + protocol_msg = await _protocol_progress_message(session_id) + if protocol_msg: + built_ctx.append(protocol_msg) + # Anti-stall / adaptive re-plan: inject intervention message if needed _stall_msg = await anti_stall.pre_turn(session_id, iteration) if _stall_msg: @@ -696,22 +721,30 @@ # Final-turn intercept: the model stopped with bare text, but the # anti-stall/adaptive nudges never fired (they live inside the # tool-loop; a bare-text turn closes the run before post_turn). - # If the todo still has open steps and we haven't hit the limit, - # queue a recovery nudge and continue the loop instead of closing. + # If the todo still has open steps — or a protocol run is open + # even though its steps are all closed — and we haven't hit the + # limit, queue a recovery nudge and continue the loop instead of + # closing. + from navi.tools.protocol import has_active_run from navi.tools.todo import has_open_steps - if ( + may_intercept = ( profile.final_intercept_enabled and turn_ctx.final_interceptions < profile.final_intercept_limit - and await has_open_steps(session_id) - ): + ) + todo_open = await has_open_steps(session_id) if may_intercept else False + protocol_open = await has_active_run(session_id) if may_intercept else False + if may_intercept and (todo_open or protocol_open): turn_ctx.final_interceptions += 1 - pending_final_nudge = Message( - role="system", - content=_FINAL_INTERCEPT_NUDGES[ + if protocol_open and not todo_open: + # A run whose steps are all closed needs a different + # closing move than "there is still work to do". + nudge = _PROTOCOL_FINAL_NUDGE + else: + nudge = _FINAL_INTERCEPT_NUDGES[ min(turn_ctx.final_interceptions, len(_FINAL_INTERCEPT_NUDGES)) - 1 - ], - ) + ] + pending_final_nudge = Message(role="system", content=nudge) await self._sessions.save(session) continue # No StreamEnd, no workers — the run continues. diff --git a/navi/core/container.py b/navi/core/container.py index ddc1ddf..30d66a3 100644 --- a/navi/core/container.py +++ b/navi/core/container.py @@ -19,7 +19,7 @@ from navi.db import Database from navi.memory import MemoryStore from navi.mcp import McpManager - from navi.skills import SkillStore + from navi.skills import ProtocolRunStore, SkillStore from navi.store import KvStore from navi.todo_store import PgTodoStore from navi.workers.base import Worker @@ -36,6 +36,7 @@ kv_store: "KvStore" = None # type: ignore[assignment] todo_store: "PgTodoStore" = None # type: ignore[assignment] skill_store: "SkillStore" = None # type: ignore[assignment] + protocol_store: "ProtocolRunStore" = None # type: ignore[assignment] scheduler: "RecallScheduler" = None # type: ignore[assignment] tool_registry: "ToolRegistry" = None # type: ignore[assignment] profile_registry: "ProfileRegistry" = None # type: ignore[assignment] @@ -109,7 +110,7 @@ from navi.memory import MemoryStore from navi.mcp import McpManager, load_mcp_servers from navi.mcp.tools import McpTool - from navi.skills import SkillStore + from navi.skills import ProtocolRunStore, SkillStore from navi.store import KvStore from navi.todo_store import PgTodoStore from navi.workers import build_default_workers @@ -125,6 +126,7 @@ kv_store = KvStore(pool) todo_store = PgTodoStore(pool) skill_store = SkillStore(pool) + protocol_store = ProtocolRunStore(pool) scheduler = RecallScheduler(pool) # Create MCP manager early so build_default_registries can wire it in @@ -149,6 +151,7 @@ terminal_manager=terminal_manager, todo_store=todo_store, skill_store=skill_store, + protocol_store=protocol_store, ) # Wire embedding backend into memory store @@ -238,6 +241,7 @@ kv_store=kv_store, todo_store=todo_store, skill_store=skill_store, + protocol_store=protocol_store, scheduler=scheduler, tool_registry=tool_registry, profile_registry=profile_registry, diff --git a/navi/core/registry.py b/navi/core/registry.py index a5c9bfc..21e6a09 100644 --- a/navi/core/registry.py +++ b/navi/core/registry.py @@ -209,6 +209,7 @@ terminal_manager=None, todo_store=None, skill_store=None, + protocol_store=None, ) -> tuple[ToolRegistry, ProfileRegistry, BackendRegistry, ContextProviderRegistry]: """Build and populate registries with all built-in components.""" from navi.core.ai_helper import AIHelper @@ -294,6 +295,10 @@ from navi.tools.skills import SkillTool builtins.append(SkillTool(skill_store)) + if protocol_store is not None: + from navi.tools.protocol import ProtocolTool + + builtins.append(ProtocolTool(skill_store, protocol_store)) for builtin in builtins: tools.register(builtin, builtin=True) diff --git a/navi/profiles/assistant/config.json b/navi/profiles/assistant/config.json index 72d5a7b..df0969e 100644 --- a/navi/profiles/assistant/config.json +++ b/navi/profiles/assistant/config.json @@ -38,6 +38,7 @@ "native": [ "todo", "skills", + "protocol", "scratchpad", "reflect", "plan", diff --git a/navi/profiles/developer/config.json b/navi/profiles/developer/config.json index f6858bc..dbf10be 100644 --- a/navi/profiles/developer/config.json +++ b/navi/profiles/developer/config.json @@ -40,6 +40,7 @@ "native": [ "todo", "skills", + "protocol", "scratchpad", "reflect", "plan", diff --git a/navi/profiles/discuss/config.json b/navi/profiles/discuss/config.json index 4264dcf..40f6610 100644 --- a/navi/profiles/discuss/config.json +++ b/navi/profiles/discuss/config.json @@ -38,6 +38,7 @@ "image_view", "todo", "skills", + "protocol", "get_current_datetime", "list_tools", "tool_manual", diff --git a/navi/profiles/server_admin/config.json b/navi/profiles/server_admin/config.json index c94ce6a..af058ae 100644 --- a/navi/profiles/server_admin/config.json +++ b/navi/profiles/server_admin/config.json @@ -40,6 +40,7 @@ "native": [ "todo", "skills", + "protocol", "scratchpad", "reflect", "plan", diff --git a/navi/skills/__init__.py b/navi/skills/__init__.py index 4a07cdd..e1d2970 100644 --- a/navi/skills/__init__.py +++ b/navi/skills/__init__.py @@ -14,12 +14,15 @@ parse_document, render_for_agent, ) +from navi.skills.runs import ProtocolRun, ProtocolRunStore from navi.skills.store import SkillEntry, SkillProposal, SkillRef, SkillStore __all__ = [ "AREAS", "KINDS", "VISIBILITIES", + "ProtocolRun", + "ProtocolRunStore", "SkillDoc", "SkillEntry", "SkillError", diff --git a/navi/skills/runs.py b/navi/skills/runs.py new file mode 100644 index 0000000..8bb19c2 --- /dev/null +++ b/navi/skills/runs.py @@ -0,0 +1,214 @@ +"""Protocol runs — the log of a protocol being executed. + +A protocol is not just a document: starting one commits the agent to a fixed +order of actions, and the run is what records that the commitment was taken and +how it ended. The row snapshots the steps and the catalog version at start, so +editing the protocol mid-run cannot quietly change what the running agent owes +the user, and the log keeps saying which version was actually executed. + +One run per session may be active. The steps themselves live in the session's +todo plan (``navi.tools.todo.insert_steps``), tagged with ``protocol_run_id``; +this table only holds the run's state and its closing report. +""" + +from __future__ import annotations + +import asyncio +import json +import uuid +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import asyncpg +import structlog + +log = structlog.get_logger() + +STATUSES = ("running", "done", "failed", "cancelled") + +_DDL = """ +CREATE TABLE IF NOT EXISTS protocol_runs ( + id TEXT PRIMARY KEY, + protocol_id TEXT NOT NULL DEFAULT '', + protocol_name TEXT NOT NULL DEFAULT '', + protocol_version INTEGER NOT NULL DEFAULT 0, + session_id TEXT NOT NULL DEFAULT '', + user_id TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'running' + CHECK (status IN ('running','done','failed','cancelled')), + steps JSONB NOT NULL DEFAULT '[]'::jsonb, + post_checks JSONB NOT NULL DEFAULT '[]'::jsonb, + report TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL DEFAULT '', + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + finished_at TIMESTAMPTZ +); +CREATE INDEX IF NOT EXISTS idx_protocol_runs_session + ON protocol_runs (session_id, started_at DESC); +CREATE UNIQUE INDEX IF NOT EXISTS idx_protocol_runs_one_active + ON protocol_runs (session_id) WHERE status = 'running'; +""" + +_SELECT = ( + "id, protocol_id, protocol_name, protocol_version, session_id, user_id, " + "status, steps, post_checks, report, reason, started_at, finished_at" +) + + +def _json_list(value: Any) -> list[str]: + if isinstance(value, str): + # asyncpg hands JSONB back as text unless a codec is registered. + try: + value = json.loads(value) + except ValueError: + return [] + return [str(v) for v in value] if isinstance(value, list) else [] + + +@dataclass +class ProtocolRun: + id: str + protocol_id: str + protocol_name: str + protocol_version: int + session_id: str + user_id: str + status: str + steps: list[str] = field(default_factory=list) + post_checks: list[str] = field(default_factory=list) + report: str = "" + reason: str = "" + started_at: datetime | None = None + finished_at: datetime | None = None + + def to_dict(self) -> dict: + return { + "id": self.id, + "protocol_id": self.protocol_id, + "protocol_name": self.protocol_name, + "protocol_version": self.protocol_version, + "session_id": self.session_id, + "user_id": self.user_id, + "status": self.status, + "steps": list(self.steps), + "post_checks": list(self.post_checks), + "report": self.report, + "reason": self.reason, + "started_at": self.started_at.isoformat() if self.started_at else None, + "finished_at": self.finished_at.isoformat() if self.finished_at else None, + } + + +def _row_to_run(row) -> ProtocolRun: + return ProtocolRun( + id=row["id"], + protocol_id=row["protocol_id"], + protocol_name=row["protocol_name"], + protocol_version=row["protocol_version"], + session_id=row["session_id"], + user_id=row["user_id"], + status=row["status"], + steps=_json_list(row["steps"]), + post_checks=_json_list(row["post_checks"]), + report=row["report"], + reason=row["reason"], + started_at=row["started_at"], + finished_at=row["finished_at"], + ) + + +class ProtocolRunStore: + """Runs in their own table; DDL runs once, lazily.""" + + def __init__(self, pool: asyncpg.Pool) -> None: + self._pool = pool + self._ready = False + self._lock = asyncio.Lock() + + async def _ensure_ready(self) -> None: + if self._ready: + return + async with self._lock: + if self._ready: + return + async with self._pool.acquire() as conn: + await conn.execute(_DDL) + self._ready = True + + async def start( + self, + *, + protocol_id: str, + protocol_name: str, + protocol_version: int, + session_id: str, + user_id: str, + steps: list[str], + post_checks: list[str] | None = None, + ) -> ProtocolRun: + await self._ensure_ready() + run_id = uuid.uuid4().hex + async with self._pool.acquire() as conn: + await conn.execute( + """ + INSERT INTO protocol_runs + (id, protocol_id, protocol_name, protocol_version, session_id, + user_id, status, steps, post_checks) + VALUES ($1, $2, $3, $4, $5, $6, 'running', $7::jsonb, $8::jsonb) + """, + run_id, protocol_id, protocol_name, int(protocol_version), session_id, + user_id, json.dumps(steps, ensure_ascii=False), + json.dumps(post_checks or [], ensure_ascii=False), + ) + run = await self.get(run_id) + assert run is not None # the INSERT just succeeded + return run + + async def get(self, run_id: str) -> ProtocolRun | None: + await self._ensure_ready() + async with self._pool.acquire() as conn: + row = await conn.fetchrow( + f"SELECT {_SELECT} FROM protocol_runs WHERE id = $1", run_id, + ) + return _row_to_run(row) if row else None + + async def active(self, session_id: str) -> ProtocolRun | None: + await self._ensure_ready() + async with self._pool.acquire() as conn: + row = await conn.fetchrow( + f"SELECT {_SELECT} FROM protocol_runs " + "WHERE session_id = $1 AND status = 'running' " + "ORDER BY started_at DESC LIMIT 1", + session_id, + ) + return _row_to_run(row) if row else None + + async def close( + self, run_id: str, *, status: str, report: str = "", reason: str = "" + ) -> ProtocolRun | None: + """Close a run. Only a ``running`` row is touched; an already closed one + is returned as it is, so a repeated call cannot rewrite history.""" + if status not in ("done", "failed", "cancelled"): + raise ValueError(f"cannot close a run as {status!r}") + await self._ensure_ready() + async with self._pool.acquire() as conn: + await conn.execute( + """ + UPDATE protocol_runs + SET status = $2, report = $3, reason = $4, finished_at = now() + WHERE id = $1 AND status = 'running' + """, + run_id, status, report, reason, + ) + return await self.get(run_id) + + async def list(self, session_id: str, limit: int = 20) -> list[ProtocolRun]: + await self._ensure_ready() + async with self._pool.acquire() as conn: + rows = await conn.fetch( + f"SELECT {_SELECT} FROM protocol_runs WHERE session_id = $1 " + "ORDER BY started_at DESC LIMIT $2", + session_id, max(1, min(int(limit), 200)), + ) + return [_row_to_run(r) for r in rows] diff --git a/navi/tools/protocol.py b/navi/tools/protocol.py new file mode 100644 index 0000000..3adb308 --- /dev/null +++ b/navi/tools/protocol.py @@ -0,0 +1,519 @@ +"""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)) diff --git a/navi/tools/todo.py b/navi/tools/todo.py index 474c825..1a23984 100644 --- a/navi/tools/todo.py +++ b/navi/tools/todo.py @@ -575,6 +575,57 @@ # ── Plan surgery (used by protocol_start) ─────────────────────────────────── +def todo_session_id(explicit: str | None = None) -> str: + """The session id the plan is stored under — public form of ``_sid``. + + Other tools that touch the plan (protocol runs) must resolve it the same + way, or a sub-agent's call would edit the parent session's plan. + """ + return _sid(explicit) + + +async def protocol_step_states( + session_id: str, protocol_run_id: str +) -> list[dict]: + """The plan steps a protocol run injected, with their current statuses. + + Returns dicts of ``index`` (1-based, as ``todo(op='update')`` expects it), + ``text``, ``status`` and ``validation``, in plan order. + """ + tasks = await _load_tasks(session_id) + return [ + { + "index": i, + "text": t.text, + "status": t.status, + "validation": t.validation, + "step_id": t.step_id, + } + for i, t in enumerate(tasks, 1) + if t.protocol_run_id == protocol_run_id + ] + + +async def close_protocol_steps( + session_id: str, protocol_run_id: str, *, note: str = "" +) -> list[int]: + """Mark a protocol run's still-open steps ``skipped``; returns their indexes. + + Used when the run ends without the steps being carried out (cancel/fail), so + the plan does not keep dangling steps that look like unfinished work. + """ + tasks = await _load_tasks(session_id) + closed: list[int] = [] + for i, t in enumerate(tasks, 1): + if t.protocol_run_id == protocol_run_id and t.status in ("pending", "in_progress"): + t.status = "skipped" + t.validation = note + closed.append(i) + if closed: + await _save_tasks(session_id, tasks) + return closed + + def _current_position(tasks: list[_Task]) -> int: """0-based index of the last step the agent has reached, or -1.""" for i in range(len(tasks) - 1, -1, -1): diff --git a/tests/unit/skills/test_protocol_runs.py b/tests/unit/skills/test_protocol_runs.py new file mode 100644 index 0000000..8a50d06 --- /dev/null +++ b/tests/unit/skills/test_protocol_runs.py @@ -0,0 +1,142 @@ +"""Tests for navi.skills.runs — the protocol run log against a fake pool.""" + +from datetime import datetime, timezone + +import pytest + +from navi.skills.runs import ProtocolRunStore +from tests.conftest_factory import FakeConnection, FakePool, FakeRecord + + +def _store(conn: FakeConnection | None = None) -> tuple[ProtocolRunStore, FakeConnection]: + conn = conn or FakeConnection() + store = ProtocolRunStore(FakePool(conn)) + # Skip the DDL round trip: it would eat a queued result. + store._ready = True + return store, conn + + +def _row(**overrides) -> FakeRecord: + base = { + "id": "run-1", + "protocol_id": "proto-1", + "protocol_name": "Выключение", + "protocol_version": 2, + "session_id": "s1", + "user_id": "u1", + "status": "running", + "steps": ["Раз", "Два"], + "post_checks": ["Готово"], + "report": "", + "reason": "", + "started_at": datetime(2026, 11, 7, 10, 0, tzinfo=timezone.utc), + "finished_at": None, + } + base.update(overrides) + return FakeRecord(**base) + + +async def test_start_writes_the_snapshot_and_reads_it_back(): + store, conn = _store() + conn.enqueue(None) # INSERT + conn.enqueue(_row()) # the SELECT that follows + run = await store.start( + protocol_id="proto-1", + protocol_name="Выключение", + protocol_version=2, + session_id="s1", + user_id="u1", + steps=["Раз", "Два"], + post_checks=["Готово"], + ) + + insert = conn.calls[0] + assert insert[0] == "execute" + assert "INSERT INTO protocol_runs" in insert[1] + assert "'running'" in insert[1] + assert '["Раз", "Два"]' in insert[2] + assert insert[2][1] == "proto-1" + assert insert[2][3] == 2 + + assert run.id == "run-1" + assert run.steps == ["Раз", "Два"] + assert run.status == "running" + + +async def test_jsonb_columns_come_back_as_text(): + store, conn = _store() + conn.enqueue(_row(steps='["Раз","Два"]', post_checks='["Готово"]')) + + run = await store.get("run-1") + + assert run.steps == ["Раз", "Два"] + assert run.post_checks == ["Готово"] + + +async def test_broken_json_is_an_empty_list(): + store, conn = _store() + conn.enqueue(_row(steps="{not json")) + + run = await store.get("run-1") + + assert run.steps == [] + + +async def test_active_returns_the_running_row(): + store, conn = _store() + conn.enqueue(_row()) + + run = await store.active("s1") + + assert run is not None + assert run.session_id == "s1" + assert "status = 'running'" in conn.calls[0][1] + + +async def test_active_without_a_run_is_none(): + store, conn = _store() + conn.enqueue(None) + assert await store.active("s1") is None + + +async def test_close_only_touches_a_running_row(): + store, conn = _store() + conn.enqueue("UPDATE 1") # the UPDATE + conn.enqueue(_row(status="done", report="готово")) # read back + + run = await store.close("run-1", status="done", report="готово") + + update = conn.calls[0] + assert "UPDATE protocol_runs" in update[1] + assert "status = 'running'" in update[1] + assert update[2] == ("run-1", "done", "готово", "") + assert run.status == "done" + assert run.report == "готово" + + +async def test_close_refuses_an_unknown_status(): + store, _ = _store() + with pytest.raises(ValueError): + await store.close("run-1", status="pending") + + +async def test_list_maps_rows_newest_first(): + store, conn = _store() + conn.enqueue([_row(id="a"), _row(id="b", status="cancelled")]) + + runs = await store.list("s1") + + assert [r.id for r in runs] == ["a", "b"] + assert runs[1].status == "cancelled" + assert "ORDER BY started_at DESC" in conn.calls[0][1] + + +async def test_to_dict_is_json_ready(): + store, conn = _store() + conn.enqueue(_row()) + + data = (await store.get("run-1")).to_dict() + + assert data["steps"] == ["Раз", "Два"] + assert data["started_at"].startswith("2026-11-07T10:00") + assert data["finished_at"] is None diff --git a/tests/unit/tools/test_protocol.py b/tests/unit/tools/test_protocol.py new file mode 100644 index 0000000..402e02c --- /dev/null +++ b/tests/unit/tools/test_protocol.py @@ -0,0 +1,444 @@ +"""Tests for the `protocol` tool — run start, closing and the per-turn reminder.""" + +from datetime import datetime, timezone + +import pytest + +from navi.skills.models import parse_document +from navi.skills.runs import ProtocolRun +from navi.skills.store import SkillEntry +from navi.tools import protocol as protocol_mod +from navi.tools._internal.base import ToolContext +from navi.tools.protocol import ProtocolTool + +PROTOCOL_DOC = """--- +kind: protocol +name: Выключение +description: Порядок выключения. +area: infrastructure +steps: [Остановить сервисы, Отмонтировать диски] +post_checks: [Юнитов не осталось] +on_deviation: Остановиться и доложить. +on_blocked: Прервать и доложить. +on_already_done: Проверить и пропустить. +--- +""" + + +def _entry(content: str = PROTOCOL_DOC, **kw) -> SkillEntry: + doc = parse_document(content) + base = dict( + id=kw.pop("id", "proto-1"), + kind=doc.kind, + name=doc.name, + description=doc.description, + triggers=doc.triggers, + tags=doc.tags, + area=doc.area, + visibility="personal", + owner_id="u1", + content=content, + version=kw.pop("version", 2), + content_hash="hash", + created_at=datetime.now(timezone.utc), + updated_at=datetime.now(timezone.utc), + ) + base.update(kw) + return SkillEntry(**base) + + +class FakeSkills: + def __init__(self, entries=None): + self.entries = entries if entries is not None else [_entry()] + self.calls = [] + + async def find(self, *, user_id, name=None, kind=None, skill_id=None): + self.calls.append(("find", skill_id, name, kind)) + for e in self.entries: + if skill_id and e.id != skill_id: + continue + if name and e.name.lower() != name.lower(): + continue + if kind and e.kind != kind: + continue + return e + return None + + +class FakeRuns: + def __init__(self, active=None): + self._active = active + self.started = [] + self.closed = [] + self.calls = [] + + async def active(self, session_id): + self.calls.append(("active", session_id)) + return self._active + + async def start(self, **kw): + self.started.append(kw) + run = ProtocolRun( + id="run-1", + protocol_id=kw["protocol_id"], + protocol_name=kw["protocol_name"], + protocol_version=kw["protocol_version"], + session_id=kw["session_id"], + user_id=kw["user_id"], + status="running", + steps=list(kw["steps"]), + post_checks=list(kw.get("post_checks") or []), + started_at=datetime(2026, 11, 7, 10, 0, tzinfo=timezone.utc), + ) + self._active = run + return run + + async def close(self, run_id, *, status, report="", reason=""): + self.closed.append((run_id, status, report, reason)) + run = self._active + if run is None: + return None + self._active = None + run.status = status + run.report = report + run.reason = reason + return run + + +def _run(**overrides) -> ProtocolRun: + base = dict( + id="run-1", + protocol_id="proto-1", + protocol_name="Выключение", + protocol_version=2, + session_id="s", + user_id="u1", + status="running", + steps=["Остановить сервисы", "Отмонтировать диски"], + post_checks=["Юнитов не осталось"], + started_at=datetime(2026, 11, 7, 10, 0, tzinfo=timezone.utc), + ) + base.update(overrides) + return ProtocolRun(**base) + + +def _states(*statuses: str) -> list[dict]: + return [ + {"index": i, "text": f"step {i}", "status": st, "validation": "", "step_id": f"s{i}"} + for i, st in enumerate(statuses, 1) + ] + + +@pytest.fixture +def ctx(): + return ToolContext(session_id="s", user_id="u1") + + +@pytest.fixture +def spliced(monkeypatch): + """Capture insert_steps calls and return fixed plan indexes.""" + calls = [] + + async def _insert(session_id, steps, **kw): + calls.append((session_id, list(steps), kw)) + return list(range(1, len(steps) + 1)) + + monkeypatch.setattr(protocol_mod, "insert_steps", _insert) + return calls + + +@pytest.fixture +def step_states(monkeypatch): + """Patch the plan read used by finish/status and let a test set the state.""" + holder = {"states": []} + + async def _states_of(session_id, run_id): + return holder["states"] + + monkeypatch.setattr(protocol_mod, "protocol_step_states", _states_of) + return holder + + +@pytest.fixture +def closed_steps(monkeypatch): + calls = [] + + async def _close(session_id, run_id, *, note=""): + calls.append((session_id, run_id, note)) + return [2] + + monkeypatch.setattr(protocol_mod, "close_protocol_steps", _close) + return calls + + +# ── start ─────────────────────────────────────────────────────────────────── + +async def test_start_splices_the_steps_and_names_their_positions(ctx, spliced): + runs = FakeRuns() + tool = ProtocolTool(FakeSkills(), runs) + + result = await tool.execute({"op": "start", "id": "proto-1"}, ctx) + + assert result.success + assert "PROTOCOL STARTED" in result.output + assert "Выключение" in result.output + assert "positions 1–2" in result.output + assert "Остановиться и доложить." in result.output + assert "protocol(op='finish'" in result.output + assert result.metadata["run_id"] == "run-1" + assert result.metadata["step_indexes"] == [1, 2] + + session_id, steps, kw = spliced[0] + assert session_id == "s" + assert steps == ["Остановить сервисы", "Отмонтировать диски"] + assert kw["source"] == "protocol" + assert kw["protocol_run_id"] == "run-1" + assert kw["step_ids"] == ["s1", "s2"] + + +async def test_start_records_a_snapshot_of_the_version(ctx, spliced): + runs = FakeRuns() + tool = ProtocolTool(FakeSkills(), runs) + + await tool.execute({"op": "start", "name": "Выключение"}, ctx) + + assert runs.started[0]["protocol_version"] == 2 + assert runs.started[0]["protocol_name"] == "Выключение" + assert runs.started[0]["post_checks"] == ["Юнитов не осталось"] + + +async def test_start_by_name(ctx, spliced): + tool = ProtocolTool(FakeSkills(), FakeRuns()) + result = await tool.execute({"op": "start", "name": "выключение"}, ctx) + assert result.success + + +async def test_start_refuses_while_another_run_is_open(ctx, spliced): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + + result = await tool.execute({"op": "start", "id": "proto-1"}, ctx) + + assert not result.success + assert "already running" in result.error + assert spliced == [] + + +async def test_start_reports_an_unknown_protocol(ctx, spliced): + tool = ProtocolTool(FakeSkills(entries=[]), FakeRuns()) + result = await tool.execute({"op": "start", "id": "nope"}, ctx) + assert not result.success + assert "No protocol" in result.error + + +async def test_start_needs_an_id_or_a_name(ctx, spliced): + tool = ProtocolTool(FakeSkills(), FakeRuns()) + result = await tool.execute({"op": "start"}, ctx) + assert not result.success + assert "'id' or 'name'" in result.error + + +# ── finish ────────────────────────────────────────────────────────────────── + +async def test_finish_requires_a_report(ctx, step_states): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + + result = await tool.execute({"op": "finish"}, ctx) + + assert not result.success + assert "report" in result.error + assert runs.closed == [] + + +async def test_finish_refuses_while_steps_are_open(ctx, step_states): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + step_states["states"] = _states("done", "in_progress") + + result = await tool.execute({"op": "finish", "report": "сделано"}, ctx) + + assert not result.success + assert "still open" in result.error + assert "2 in_progress" in result.error + assert runs.closed == [] + + +async def test_finish_refuses_a_failed_step_and_points_at_fail(ctx, step_states): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + step_states["states"] = _states("failed", "pending") + + result = await tool.execute({"op": "finish", "report": "сделано"}, ctx) + + assert not result.success + assert "marked failed" in result.error + assert "op='fail'" in result.error + assert runs.closed == [] + + +async def test_finish_refuses_when_the_plan_lost_the_steps(ctx, step_states): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + step_states["states"] = _states("done") # one of two + + result = await tool.execute({"op": "finish", "report": "сделано"}, ctx) + + assert not result.success + assert "no longer contains all" in result.error + assert runs.closed == [] + + +async def test_finish_closes_a_run_whose_steps_are_closed(ctx, step_states): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + step_states["states"] = _states("done", "skipped") + + result = await tool.execute({"op": "finish", "report": "всё проверено"}, ctx) + + assert result.success + assert "finished" in result.output + assert runs.closed == [("run-1", "done", "всё проверено", "")] + + +# ── cancel / fail ─────────────────────────────────────────────────────────── + +async def test_cancel_requires_a_reason_and_skips_open_steps(ctx, closed_steps): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + + refused = await tool.execute({"op": "cancel"}, ctx) + assert not refused.success + assert "reason" in refused.error + assert runs.closed == [] + + result = await tool.execute({"op": "cancel", "reason": "передумали"}, ctx) + assert result.success + assert "cancelled" in result.output + assert closed_steps[0][1] == "run-1" + assert closed_steps[0][2] == "protocol cancelled: передумали" + assert runs.closed == [("run-1", "cancelled", "", "передумали")] + + +async def test_fail_closes_the_run_as_failed(ctx, closed_steps): + runs = FakeRuns(active=_run()) + tool = ProtocolTool(FakeSkills(), runs) + + result = await tool.execute({"op": "fail", "reason": "сервис не остановился"}, ctx) + + assert result.success + assert "failed" in result.output + assert runs.closed == [("run-1", "failed", "", "сервис не остановился")] + + +async def test_closing_without_a_run_is_refused(ctx): + tool = ProtocolTool(FakeSkills(), FakeRuns()) + result = await tool.execute({"op": "finish", "report": "x"}, ctx) + assert not result.success + assert "No protocol is running" in result.error + + +# ── status ────────────────────────────────────────────────────────────────── + +async def test_status_without_a_run(ctx): + tool = ProtocolTool(FakeSkills(), FakeRuns()) + result = await tool.execute({"op": "status"}, ctx) + assert result.success + assert "No protocol is running" in result.output + + +async def test_status_lists_the_run_and_its_steps(ctx, step_states): + tool = ProtocolTool(FakeSkills(), FakeRuns(active=_run())) + step_states["states"] = _states("done", "in_progress") + + result = await tool.execute({"op": "status"}, ctx) + + assert "ACTIVE PROTOCOL" in result.output + assert "1. step 1 (done)" in result.output + assert "2. step 2 (in_progress)" in result.output + + +async def test_unknown_op_names_the_options(ctx): + tool = ProtocolTool(FakeSkills(), FakeRuns()) + result = await tool.execute({"op": "run"}, ctx) + assert not result.success + assert "start" in result.error and "fail" in result.error + + +async def test_a_broken_run_store_is_reported_not_raised(ctx, spliced): + class Broken: + async def active(self, session_id): + raise RuntimeError("db down") + + tool = ProtocolTool(FakeSkills(), Broken()) + + for op in ({"op": "start", "id": "proto-1"}, {"op": "status"}, {"op": "finish", "report": "x"}): + result = await tool.execute(op, ctx) + assert not result.success + assert "Cannot read the protocol run" in result.error + + +# ── per-turn reminder ─────────────────────────────────────────────────────── + +async def test_no_reminder_without_a_run(monkeypatch, step_states): + ProtocolTool(FakeSkills(), FakeRuns()) + assert await protocol_mod.get_protocol_message("s") is None + + +async def test_reminder_carries_the_open_steps(step_states): + ProtocolTool(FakeSkills(), FakeRuns(active=_run())) + step_states["states"] = _states("done", "pending") + + msg = await protocol_mod.get_protocol_message("s") + + assert msg.role == "system" + assert "PROTOCOL ACTIVE" in msg.content + assert "2 of 2 steps" in msg.content + assert "protocol(op='fail'" in msg.content + assert "do not end the turn" in msg.content + + +async def test_reminder_says_to_finish_when_all_steps_are_closed(step_states): + ProtocolTool(FakeSkills(), FakeRuns(active=_run())) + step_states["states"] = _states("done", "done") + + msg = await protocol_mod.get_protocol_message("s") + + assert "Every step is closed" in msg.content + assert "protocol(op='finish'" in msg.content + + +async def test_reminder_flags_a_plan_that_lost_the_steps(step_states): + ProtocolTool(FakeSkills(), FakeRuns(active=_run())) + step_states["states"] = [] + + msg = await protocol_mod.get_protocol_message("s") + + assert "no longer in the plan" in msg.content + assert "op='fail'" in msg.content + + +async def test_reminder_is_silent_when_the_store_breaks(monkeypatch): + class Broken: + async def active(self, session_id): + raise RuntimeError("db down") + + ProtocolTool(FakeSkills(), Broken()) + assert await protocol_mod.get_protocol_message("s") is None + + +async def test_has_active_run_gates_the_final_intercept(): + ProtocolTool(FakeSkills(), FakeRuns()) + assert await protocol_mod.has_active_run("s") is False + + ProtocolTool(FakeSkills(), FakeRuns(active=_run())) + assert await protocol_mod.has_active_run("s") is True + + +async def test_has_active_run_is_false_when_the_store_breaks(): + class Broken: + async def active(self, session_id): + raise RuntimeError("db down") + + ProtocolTool(FakeSkills(), Broken()) + assert await protocol_mod.has_active_run("s") is False diff --git a/tests/unit/tools/test_todo.py b/tests/unit/tools/test_todo.py index dbad00e..f69487b 100644 --- a/tests/unit/tools/test_todo.py +++ b/tests/unit/tools/test_todo.py @@ -693,6 +693,74 @@ @pytest.mark.asyncio +async def test_protocol_step_states_lists_only_that_runs_steps(_fake_kv): + from navi.tools import todo as _mod + + _mod._store = _FakeTodoStore(tasks=[ + {"text": "plan one", "status": "done"}, + {"text": "p1", "status": "done", "protocol_run_id": "run-1", "step_id": "s1"}, + {"text": "p2", "status": "in_progress", "protocol_run_id": "run-1", "step_id": "s2"}, + {"text": "other run", "status": "pending", "protocol_run_id": "run-2"}, + ]) + + states = await _mod.protocol_step_states("sess1", "run-1") + + # 1-based plan indexes, so the model can feed them straight to todo(op='update'). + assert [(s["index"], s["text"], s["status"]) for s in states] == [ + (2, "p1", "done"), + (3, "p2", "in_progress"), + ] + assert states[0]["step_id"] == "s1" + + +@pytest.mark.asyncio +async def test_close_protocol_steps_skips_only_open_steps_of_the_run(_fake_kv): + from navi.tools import todo as _mod + + store = _FakeTodoStore(tasks=[ + {"text": "plan one", "status": "in_progress"}, + {"text": "p1", "status": "done", "protocol_run_id": "run-1"}, + {"text": "p2", "status": "pending", "protocol_run_id": "run-1"}, + {"text": "p3", "status": "failed", "protocol_run_id": "run-2"}, + ]) + _mod._store = store + + closed = await _mod.close_protocol_steps("sess1", "run-1", note="protocol cancelled: no") + + assert closed == [3] + assert store.tasks[2]["status"] == "skipped" + assert store.tasks[2]["validation"] == "protocol cancelled: no" + assert store.tasks[1]["status"] == "done" # already closed: untouched + assert store.tasks[0]["status"] == "in_progress" # other work: untouched + assert store.tasks[3]["status"] == "failed" # other run: untouched + + +@pytest.mark.asyncio +async def test_close_protocol_steps_saves_nothing_when_all_are_closed(_fake_kv): + from navi.tools import todo as _mod + + store = _FakeTodoStore(tasks=[ + {"text": "p1", "status": "done", "protocol_run_id": "run-1"}, + ]) + _mod._store = store + + assert await _mod.close_protocol_steps("sess1", "run-1") == [] + assert store.saved is None + + +def test_todo_session_id_prefers_the_subagent_override(_fake_kv): + from navi.tools import todo as _mod + from navi.tools._internal.base import current_todo_session_id + + assert _mod.todo_session_id("parent") == "parent" + token = current_todo_session_id.set("ephemeral") + try: + assert _mod.todo_session_id("parent") == "ephemeral" + finally: + current_todo_session_id.reset(token) + + +@pytest.mark.asyncio async def test_restart_note_fires_once_for_a_stale_in_progress_step(_fake_kv): from datetime import timedelta