"""The four things an agent chat can do to the machine it is pointed at. Two rules shape all of them. **The descriptions say nothing about where.** A tool description is schema, sent verbatim and deliberately not editable, and it states facts about what a runner does. Which machine, which directory and which mode is in force are facts about *this chat*, so they live in the harness fragment where they can change without the schema changing under a model mid-conversation. **Every runner re-checks the mode.** `_authorise` in the generation loop is the real gate and runs before any of this, but a backstop here means a future path that reaches `run_tool` directly -- a retry, a test, an admin re-run button -- cannot walk past it. That is the same instinct that closed the registry hole: the check belongs where the action is, not only where the action was decided. """ from __future__ import annotations import json import logging from typing import Any from lembas.services.agent import policy from lembas.services.agent.base import ExecError, ExecRequest from lembas.services.agent.session import AgentContext from lembas.services.tools import ( RISK_EXECUTE, RISK_READ, RISK_WRITE, ToolContext, ToolDef, ToolOutcome, ) log = logging.getLogger(__name__) FAMILY_AGENT = "agent" # How much of a command's output is kept on the message row for the transcript, # separately from what the model reads. `max_output` is spent once; this is # stored on every message forever. MAX_EVENT_CHARS = 4000 _STRING = {"type": "string"} def _event(name: str, context: AgentContext, summary: str, **extra: Any) -> dict[str, Any]: return { "name": name, "kind": "agent", "label": f"{context.label}", "query": summary, "detail": context.project_dir or "", "results": [], **extra, } def _refused(name: str, context: AgentContext, summary: str, reason: str) -> ToolOutcome: return ToolOutcome( f"That was not allowed: {reason}", _event(name, context, summary, status="error", error=reason), ) def _permitted(context: AgentContext, name: str, risk: str, command: str = "") -> str: """Empty when this call may proceed, else why not. The backstop. What it catches is a call arriving by a path that skipped `_authorise` -- a retry, a test, some future re-run button. A call a person has just allowed carries `approved` and goes straight through. Without that this would refuse the very thing that was approved: the mode says "ask", and asking is precisely what happened. """ if context.approved: return "" decision = policy.decide( mode=context.mode, risk=risk, tool_name=name, command=command, allow=context.allow, deny=context.deny, ) if decision.verdict == policy.ALLOW: return "" return decision.reason or "it needs to be approved first." def _agent(context: ToolContext) -> AgentContext | None: return getattr(context, "agent", None) # --- Running a command -------------------------------------------------------- async def _run_shell(context: ToolContext, args: dict[str, Any]) -> ToolOutcome: agent = _agent(context) command = str(args.get("command") or "").strip() if agent is None: return ToolOutcome( "This conversation is not connected to a machine, so nothing can be run.", {"name": "shell_run", "status": "error", "error": "No connection.", "results": []}, ) if not command: return _refused("shell_run", agent, "", "no command was given.") if reason := _permitted(agent, "shell_run", RISK_EXECUTE, command): return _refused("shell_run", agent, command, reason) timeout = _timeout(args.get("timeout"), agent) try: result = await agent.executor().run( ExecRequest( command=command, cwd=str(args.get("cwd") or "").strip(), timeout=timeout, max_bytes=agent.max_output, ) ) except ExecError as exc: return ToolOutcome( exc.message, _event("shell_run", agent, command, status="error", error=exc.message) ) body = result.output.strip() if result.timed_out: head = f"The command was stopped after {timeout:g}s." elif result.exit_status == 0: head = "" if body else "It ran, and printed nothing." else: head = f"It exited {result.exit_status}." content = f"{head}\n\n{body}".strip() if head else body return ToolOutcome( content or "It ran, and printed nothing.", _event( "shell_run", agent, command, status="ok" if result.ok else "error", error="" if result.ok else head, text=body[:MAX_EVENT_CHARS], ), ) def _timeout(raw: Any, agent: AgentContext) -> float: """What the model asked for, bounded by what an administrator allowed.""" try: wanted = float(raw) if raw is not None else agent.timeout except (TypeError, ValueError): wanted = agent.timeout return min(max(wanted, 1.0), agent.max_timeout) # --- Files --------------------------------------------------------------------- async def _run_read(context: ToolContext, args: dict[str, Any]) -> ToolOutcome: agent = _agent(context) path = str(args.get("path") or "").strip() if agent is None or not path: return _no_connection_or_path("file_read", agent, path) if reason := _permitted(agent, "file_read", RISK_READ): return _refused("file_read", agent, path, reason) try: text = await agent.executor().read_file(path, max_bytes=agent.max_output) except ExecError as exc: return ToolOutcome( exc.message, _event("file_read", agent, path, status="error", error=exc.message) ) return ToolOutcome( text or "(the file is empty)", _event("file_read", agent, path, status="ok", text=text[:MAX_EVENT_CHARS]), ) async def _run_write(context: ToolContext, args: dict[str, Any]) -> ToolOutcome: agent = _agent(context) path = str(args.get("path") or "").strip() if agent is None or not path: return _no_connection_or_path("file_write", agent, path) if reason := _permitted(agent, "file_write", RISK_WRITE): return _refused("file_write", agent, path, reason) content = args.get("content") if not isinstance(content, str): content = "" if content is None else json.dumps(content, ensure_ascii=False) try: written = await agent.executor().write_file(path, content) except ExecError as exc: return ToolOutcome( exc.message, _event("file_write", agent, path, status="error", error=exc.message) ) return ToolOutcome( f"Wrote {written} bytes to {path}.", _event("file_write", agent, path, status="ok", text=f"{written} bytes"), ) async def _run_list(context: ToolContext, args: dict[str, Any]) -> ToolOutcome: agent = _agent(context) if agent is None: return _no_connection_or_path("file_list", agent, "") path = str(args.get("path") or "").strip() if reason := _permitted(agent, "file_list", RISK_READ): return _refused("file_list", agent, path, reason) try: names = await agent.executor().list_dir(path) except ExecError as exc: return ToolOutcome( exc.message, _event("file_list", agent, path, status="error", error=exc.message) ) where = path or agent.project_dir or "." body = "\n".join(names) if names else "(empty)" return ToolOutcome( f"{where}:\n{body}", _event("file_list", agent, where, status="ok", text=body[:MAX_EVENT_CHARS]), ) def _no_connection_or_path(name: str, agent: AgentContext | None, path: str) -> ToolOutcome: if agent is None: return ToolOutcome( "This conversation is not connected to a machine.", {"name": name, "status": "error", "error": "No connection.", "results": []}, ) return _refused(name, agent, path, "no path was given.") # --- Proposing a plan ----------------------------------------------------------- MAX_STEPS = 20 async def _run_plan(context: ToolContext, args: dict[str, Any]) -> ToolOutcome: """Record a plan and stop. Writes nothing and runs nothing, which is why it is `RISK_READ` and works in Plan mode without asking. The loop notices the event and ends the reply there: a plan followed by three more rounds of the model changing its mind is not a plan. """ agent = _agent(context) title = str(args.get("title") or "").strip() or "A plan" steps = [str(s).strip() for s in (args.get("steps") or []) if str(s).strip()] steps = steps[:MAX_STEPS] if not steps: return ToolOutcome( "A plan needs at least one step. Say what you would actually do.", {"name": "plan_submit", "kind": "plan", "status": "error", "error": "No steps.", "results": []}, ) return ToolOutcome( "Plan recorded. Stop here — they will read it and decide whether to " "carry it out. Do not start doing it.", { "name": "plan_submit", "kind": "plan", "label": agent.label if agent else "", "query": title, "status": "ok", "results": [], # Read back by the loop, which puts it on the message so the # Execute button sends exactly what was proposed rather than an # approximation parsed out of the prose. "plan": {"title": title, "steps": steps}, }, ) # --- The definitions ----------------------------------------------------------- def tool_defs(context: AgentContext | None = None) -> list[ToolDef]: """The agent tools, bound to one chat's machine. `None` yields the same definitions unbound, which is what `tools.registry` needs: it maps an offered tool *name* back to its family and has no chat to resolve. Their runners still work -- they report that the conversation is not connected to a machine, which is true. `plan_submit` is offered in Plan mode and nowhere else. It ends the reply, and a model in Auto mode that proposed a plan instead of doing the work would be obeying the wrong instinct at exactly the wrong moment. """ defs = [ ToolDef( name="shell_run", family=FAMILY_AGENT, description=( "Run a shell command and read back everything it printed, stdout " "and stderr together. Each call is a fresh shell, so a `cd` in one " "does not carry into the next — pass `cwd` instead. Nothing can " "answer a prompt, so pass the flags that make a command " "non-interactive rather than waiting for it to ask." ), parameters={ "type": "object", "properties": { "command": {**_STRING, "description": "The command line to run."}, "cwd": { **_STRING, "description": "Where to run it. Defaults to the project directory.", }, "timeout": { "type": "number", "description": "Seconds to allow. Bounded by the instance settings.", }, }, "required": ["command"], }, run=_run_shell, risk=RISK_EXECUTE, ), ToolDef( name="file_read", family=FAMILY_AGENT, description=( "Read a text file. A relative path is taken from the project " "directory. Large files are cut off at the end rather than " "refused, and you are told when that happened." ), parameters={ "type": "object", "properties": {"path": {**_STRING, "description": "The file to read."}}, "required": ["path"], }, run=_run_read, risk=RISK_READ, ), ToolDef( name="file_write", family=FAMILY_AGENT, description=( "Write a text file, replacing it entirely if it already exists. " "A relative path is taken from the project directory. Read a file " "before rewriting it unless you are certain what is in it." ), parameters={ "type": "object", "properties": { "path": {**_STRING, "description": "The file to write."}, "content": {**_STRING, "description": "Its whole new contents."}, }, "required": ["path", "content"], }, run=_run_write, risk=RISK_WRITE, ), ToolDef( name="file_list", family=FAMILY_AGENT, description=( "List a directory. Defaults to the project directory. Use this " "before guessing at a path." ), parameters={ "type": "object", "properties": {"path": {**_STRING, "description": "The directory to list."}}, "required": [], }, run=_run_list, risk=RISK_READ, ), ToolDef( name="plan_submit", family=FAMILY_AGENT, description=( "Set out what you would do, as an ordered list of steps, and " "stop. Use this to finish when you have been asked to plan " "rather than to act: they will read it and decide whether to " "carry it out. Each step should be one thing, concrete enough " "to follow — name the files and the commands." ), parameters={ "type": "object", "properties": { "title": {**_STRING, "description": "What the plan achieves, in a line."}, "steps": { "type": "array", "items": _STRING, "description": "The steps, in order.", }, }, "required": ["title", "steps"], }, run=_run_plan, # It writes nothing and runs nothing, so it needs no approval -- # which is the point: Plan mode has to be able to finish. risk=RISK_READ, ), ] if context is not None and context.mode != policy.MODE_PLAN: return [tool for tool in defs if tool.name != "plan_submit"] return defs __all__ = ["FAMILY_AGENT", "MAX_EVENT_CHARS", "tool_defs"]