Agent chats run commands, and stop to ask first

The four tools an agent chat has -- shell_run, file_read, file_write,
file_list -- and the mode table wired into the loop that decides which of
them stop for approval. Verified end to end against a real Kali container
over SSH: the card shows the command, allowing it runs it there, and the
file it writes is visible from outside.

The mode is enforced in `_authorise`, in the generation loop, server-side,
keyed on each tool's declared risk. Not in the prompt: a model is told
which mode it is in so it behaves sensibly, but everything it reads -- a
web page, a README, the output of the last command -- is untrusted, and a
rule written only into a system message is one a poisoned file can argue
with. Within an agent chat every call goes through the table, including
the built-in ones, because notes_edit writes and Plan mode meaning "look
but do not touch" has to mean that too.

Two things this turned up.

The runners re-check the mode as a backstop, and that backstop refused the
very thing a person had just approved -- the mode says "ask", and asking
was exactly what happened. Approval is now threaded per call, on a copy of
the context, because a round runs its calls together and only some of them
were allowed.

And the harness said nothing at all, because `registry` maps an offered
tool *name* back to a family and did not know the agent tools existed. So
shell_run resolved to no family and the fragment naming the machine, the
directory and the mode was never admitted. The same omission cost custom
tools their guidance once already; there is a test for it now.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/lembas/api/chats.py

@@ -14,7 +14,15 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session as DBSession
 
 from lembas.api.deps import Db, RequiredUser, require_permission
-from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Message, User
+from lembas.db.models import (
+    KIND_AGENT,
+    KIND_CHAT,
+    ROLE_ASSISTANT,
+    ROLE_USER,
+    Chat,
+    Message,
+    User,
+)
 from lembas.db.session import session_scope
 from lembas.security import permissions
 from lembas.services import audio as audio_service
@@ -24,8 +32,9 @@ from lembas.services import files as files_service
 from lembas.services import generation as generation_service
 from lembas.services import metrics as metrics_service
 from lembas.services import prompts as prompts_service
-from lembas.services import sse
+from lembas.services import settings_store, sse
 from lembas.services import tools as tools_service
+from lembas.services.agent import policy as agent_policy
 from lembas.services.markdown import escape_text, render_markdown
 from lembas.web.templating import render, templates
 
@@ -54,8 +63,16 @@ def _new_chat(
     folder_id: str = "",
     model_id: str = "",
     temporary: bool = False,
+    kind: str = KIND_CHAT,
+    ssh_profile_id: str = "",
+    project_dir: str = "",
 ) -> Chat:
-    """Create a chat row, resolving which model it should use."""
+    """Create a chat row, resolving which model it should use.
+
+    An agent chat's connection is settled here and never again. That is the
+    lock: the harness, the tools offered and the approval loop all differ, so a
+    conversation whose earlier turns ran somewhere else is not one conversation.
+    """
     chosen = None
     if model_id:
         match = next(
@@ -66,12 +83,16 @@ def _new_chat(
     if chosen is None:
         chosen = chat_service.default_model(db, user)
 
+    profile = _agent_target(db, user, kind, ssh_profile_id)
     chat = Chat(
         user_id=user.id,
         folder_id=folder_id or None,
         model_id=chosen[0] if chosen else "",
         connection_id=chosen[1] if chosen else None,
         temporary=temporary,
+        kind=KIND_AGENT if profile is not None else KIND_CHAT,
+        ssh_profile_id=profile.id if profile is not None else None,
+        project_dir=(project_dir.strip() or profile.default_dir) if profile is not None else "",
     )
     db.add(chat)
     db.commit()
@@ -87,6 +108,9 @@ async def start_chat(
     folder_id: str = Form(""),
     model_id: str = Form(""),
     temporary: bool = Form(False),
+    kind: str = Form(KIND_CHAT),
+    ssh_profile_id: str = Form(""),
+    project_dir: str = Form(""),
 ) -> Response:
     """Create a chat from its first message.
 
@@ -100,7 +124,14 @@ async def start_chat(
         return Response(status_code=status.HTTP_204_NO_CONTENT)
 
     chat = _new_chat(
-        db, user, folder_id=folder_id, model_id=model_id, temporary=temporary
+        db,
+        user,
+        folder_id=folder_id,
+        model_id=model_id,
+        temporary=temporary,
+        kind=kind,
+        ssh_profile_id=ssh_profile_id,
+        project_dir=project_dir,
     )
 
     user_message = chat_service.create_message(db, chat, ROLE_USER, content)
@@ -116,6 +147,32 @@ async def start_chat(
     return response
 
 
+def _agent_target(db: DBSession, user: User, kind: str, profile_id: str):
+    """The connection an agent chat is being pointed at, or None.
+
+    Every "no" collapses to None and the chat is an ordinary one: not asked
+    for, no permission, the feature off, or a profile that is not this person's.
+    Refusing outright would be worse -- somebody whose permission was withdrawn
+    between opening the composer and sending would lose the message.
+    """
+    from lembas.db.models import SshProfile
+    from lembas.security import permissions
+
+    if kind != KIND_AGENT or not profile_id:
+        return None
+    if not permissions.has(db, user, "tools.agent"):
+        return None
+    if not settings_store.agents(db).get("enabled"):
+        return None
+
+    profile = db.get(SshProfile, profile_id)
+    # Ownership re-checked rather than trusted from the form: an id in a POST is
+    # not an authorisation, and these are credentials to somebody's machine.
+    if profile is None or profile.owner_id != user.id or not profile.enabled:
+        return None
+    return profile
+
+
 # There is deliberately no route that creates an empty chat. Starting one is
 # navigation to /chat (optionally ?model=...), and the row is written by
 # /start when the first message is actually sent.
@@ -751,6 +808,25 @@ async def update_chat(request: Request, db: Db, user: RequiredUser, chat_id: str
     if "folder_id" in form:
         chat.folder_id = str(form["folder_id"]) or None
 
+    # The mode is the one agent field that changes mid-chat: it decides what
+    # gets asked about, not what the conversation is.
+    if "agent_mode" in form:
+        wanted = str(form["agent_mode"]).strip()
+        if wanted in agent_policy.MODES:
+            chat.agent_mode = wanted
+
+    # And these are the ones that never do. Refused rather than ignored: a form
+    # that quietly did nothing would look like a bug from the outside, and
+    # without the refusal a crafted POST would repoint a conversation at another
+    # machine halfway through.
+    for locked in ("kind", "ssh_profile_id", "project_dir"):
+        if locked in form:
+            raise HTTPException(
+                status.HTTP_409_CONFLICT,
+                "A chat's connection is fixed when it is created. Start a new "
+                "chat to work somewhere else.",
+            )
+
     model_id = str(form.get("model_id", "")).strip()
 
     if model_id:

src/lembas/api/pages.py

@@ -58,10 +58,48 @@ def _chat_context(db: DBSession, user: User, chat: Chat | None) -> dict:
             else []
         ),
         "attached_base_ids": [base.id for base in chat.knowledge_bases] if chat else [],
+        **_agent_context(db, user, chat),
         **audio_service.template_flags(db, user),
     }
 
 
+def _agent_context(db: DBSession, user: User, chat: Chat | None) -> dict:
+    """What the composer and the chat header need to know about agent chats.
+
+    `agent_profiles` is empty unless every one of the conditions holds -- the
+    feature is on, the reader may run commands, and they have a usable
+    connection -- which is what makes the picker appear only when choosing it
+    would lead anywhere.
+    """
+    from lembas.db.models import SshProfile
+    from lembas.services.agent import policy as agent_policy
+
+    profiles: list[SshProfile] = []
+    if settings_store.agents(db).get("enabled") and permissions.has(db, user, "tools.agent"):
+        profiles = list(
+            db.scalars(
+                select(SshProfile)
+                .where(SshProfile.owner_id == user.id, SshProfile.enabled.is_(True))
+                .order_by(SshProfile.name)
+            )
+        )
+
+    current = None
+    if chat is not None and chat.ssh_profile_id:
+        current = db.get(SshProfile, chat.ssh_profile_id)
+        if current is not None and current.owner_id != user.id:
+            current = None
+
+    return {
+        "agent_profiles": profiles,
+        "agent_profile": current,
+        "agent_modes": [
+            (m, agent_policy.MODE_LABELS[m], agent_policy.MODE_HINTS[m])
+            for m in agent_policy.MODES
+        ],
+    }
+
+
 def sidebar_context(db: DBSession, user: User) -> dict:
     """Folder tree plus the chats that belong to no folder.
 

src/lembas/services/agent/policy.py

@@ -41,6 +41,35 @@ MODE_HINTS = {
     MODE_PLAN: "Reads freely, changes nothing, and finishes by proposing a plan.",
 }
 
+# What the *model* is told about the mode it is in. Different words from
+# MODE_HINTS, which describes it to a person: this is about how to behave, and
+# says the one thing that changes what a competent model does -- that being
+# stopped for approval is normal and worth batching for.
+MODE_GUIDANCE = {
+    MODE_MANUAL: (
+        "You are in **Manual** mode: everything you do is shown to them for "
+        "approval first. Expect to be interrupted, and say what you are about "
+        "to do before you do it."
+    ),
+    MODE_EDIT: (
+        "You are in **Edit** mode: you may read and write files freely, but "
+        "every command is shown to them for approval first. Prefer reading and "
+        "writing files over shelling out where both would work."
+    ),
+    MODE_AUTO: (
+        "You are in **Auto** mode: nothing is shown to them first. That is trust "
+        "rather than permission — be as careful as you would be if each step "
+        "were being watched, and stop to say so if you find yourself about to "
+        "do something you could not undo."
+    ),
+    MODE_PLAN: (
+        "You are in **Plan** mode: read and explore freely, but change nothing. "
+        "Anything that writes or runs will be stopped for approval, so do not "
+        "rely on it. Finish by setting out what you would do, as steps, so it "
+        "can be carried out afterwards."
+    ),
+}
+
 ALLOW = "allow"
 ASK = "ask"
 
@@ -172,6 +201,7 @@ __all__ = [
     "MODES",
     "MODE_AUTO",
     "MODE_EDIT",
+    "MODE_GUIDANCE",
     "MODE_HINTS",
     "MODE_LABELS",
     "MODE_MANUAL",

src/lembas/services/agent/session.py

@@ -0,0 +1,130 @@
+"""What one agent chat is pointed at, resolved while a session is open.
+
+Everything a runner needs travels in `AgentContext`: the machine, the decrypted
+credential, the mode in force, and the two lists that adjust it. Nothing is
+looked up later, for the reason `Endpoint` is a frozen copy of a `Connection`
+and `ToolContext` carries an owner id rather than a `User` -- a generation
+outlives the request that started it, and a detached instance is a trap.
+
+The mode is read **once, at the start of the reply**, and deliberately does not
+change under a reply already in flight. Somebody switching to Auto halfway
+through must not retroactively approve what is already queued.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field, replace
+from typing import Any
+
+from sqlalchemy.orm import Session as DBSession
+
+from lembas.db.models import KIND_AGENT, Chat, SshProfile, User
+from lembas.services import settings_store
+from lembas.services.agent import policy
+from lembas.services.agent import ssh as ssh_service
+from lembas.services.agent.base import Executor
+from lembas.services.agent.policy import Limits
+
+log = logging.getLogger(__name__)
+
+
+@dataclass
+class AgentContext:
+    """The machine an agent chat acts on, and what it may do there."""
+
+    chat_id: str
+    label: str
+    project_dir: str
+    mode: str = policy.MODE_MANUAL
+    allow: tuple[str, ...] = ()
+    deny: tuple[str, ...] = ()
+    limits: Limits = field(default_factory=Limits)
+    # Per-command bounds, from the instance settings.
+    timeout: float = 60.0
+    max_timeout: float = 600.0
+    max_output: int = 64 * 1024
+    # The decrypted credential. Held here and nowhere else, and cleared by
+    # `generation` when the reply ends -- a finished Generation lingers five
+    # minutes so late followers get the final frames, and a private key should
+    # not linger with it.
+    spec: dict[str, Any] = field(default_factory=dict)
+    # Set only on the per-call copy handed to a runner whose call a person has
+    # just allowed. The runners re-check the mode as a backstop, and without
+    # this they would refuse the very thing that was approved -- the mode says
+    # "ask", and asking is exactly what happened.
+    approved: bool = False
+
+    def executor(self) -> Executor:
+        return ssh_service.SshExecutor(self.spec, self.project_dir)
+
+    def clear(self) -> None:
+        self.spec = {}
+
+    def as_approved(self) -> AgentContext:
+        """A copy of this context for one call a person has allowed."""
+        return replace(self, approved=True)
+
+
+def profile_for(db: DBSession, chat: Chat, user: User | None) -> SshProfile | None:
+    """The connection this chat is pointed at, if it is still usable.
+
+    Ownership is re-checked here rather than trusted from when the chat was
+    created: a profile can be deleted, disabled, or moved to a host whose key
+    has not been confirmed since, and any of those should stop the chat acting
+    rather than be discovered at the first command.
+    """
+    if chat is None or chat.kind != KIND_AGENT or not chat.ssh_profile_id:
+        return None
+
+    profile = db.get(SshProfile, chat.ssh_profile_id)
+    if profile is None or not profile.enabled:
+        return None
+    if user is not None and profile.owner_id != user.id:
+        return None
+    return profile
+
+
+def resolve(db: DBSession, chat: Chat, user: User | None) -> AgentContext | None:
+    """This chat's agent setup, or None if it has none it can use.
+
+    None is the answer to every "no": not an agent chat, the feature switched
+    off, the connection gone or disabled, SSH not installed. Each of those means
+    the agent tools are not offered at all, which is better than offering a tool
+    that fails on its first call.
+
+    A profile whose host key has never been confirmed is deliberately *not* one
+    of them. The tools are offered and the failure is explicit, because "check
+    the connection and accept its fingerprint" is a thing the reader can act on,
+    while a silently missing tool is not.
+    """
+    profile = profile_for(db, chat, user)
+    if profile is None:
+        return None
+
+    values = settings_store.agents(db)
+    if not values.get("enabled"):
+        return None
+    if ssh_service.available():
+        return None
+
+    return AgentContext(
+        chat_id=chat.id,
+        label=profile.label,
+        project_dir=chat.project_dir or profile.default_dir or "",
+        mode=chat.agent_mode if chat.agent_mode in policy.MODES else policy.MODE_MANUAL,
+        allow=tuple(values.get("allow_default") or ()),
+        deny=tuple(values.get("deny_default") or ()),
+        limits=Limits(
+            steps=int(values.get("max_steps") or 40),
+            wall_seconds=float(values.get("max_wall_seconds") or 900),
+            output_bytes=int(values.get("max_total_output_bytes") or 1024 * 1024),
+        ),
+        timeout=float(values.get("default_timeout") or 60),
+        max_timeout=float(values.get("max_timeout") or 600),
+        max_output=int(values.get("max_output_bytes") or 64 * 1024),
+        spec=ssh_service.spec_from(profile),
+    )
+
+
+__all__ = ["AgentContext", "profile_for", "resolve"]

src/lembas/services/agent/tools.py

@@ -0,0 +1,331 @@
+"""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.")
+
+
+# --- 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.
+    """
+    return [
+        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,
+        ),
+    ]
+
+
+__all__ = ["FAMILY_AGENT", "MAX_EVENT_CHARS", "tool_defs"]

src/lembas/services/generation.py

@@ -21,7 +21,7 @@ import json
 import logging
 import time
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from datetime import UTC, datetime, timedelta
 
 from sqlalchemy import select
@@ -34,6 +34,7 @@ from lembas.services import interaction, tokens
 from lembas.services import metrics as metrics_service
 from lembas.services import prompts as prompts_service
 from lembas.services import tools as tools_service
+from lembas.services.agent import policy as agent_policy
 from lembas.services.llm.openai_client import (
     LLMError,
     chunk_usage,
@@ -120,6 +121,10 @@ class Generation:
     # Seconds spent waiting for a person, cumulative. Taken off the wall-clock
     # budget so that thinking time is the model's and not the reader's.
     waited: float = 0.0
+    # How much tool output this reply has handed back, against the agent budget.
+    # A model that fills its own context with build logs has no room left to
+    # answer with.
+    output_bytes: int = 0
 
     def touch(self) -> None:
         self.version += 1
@@ -273,6 +278,9 @@ async def _run(generation: Generation) -> None:
     endpoint = model_id = None
     needs_title = False
     title_prompt = ""
+    # Bound before the try, because the finally clears the credential on it and
+    # a chat that has been deleted returns before it would otherwise be set.
+    tool_context = None
 
     try:
         # Before the request is assembled, so build_request is called once and
@@ -313,8 +321,25 @@ async def _run(generation: Generation) -> None:
 
         generation.prompt_estimate = tokens.estimate_request(payload)
 
-        for round_number in range(tools_service.MAX_ROUNDS + 1):
+        limits = tool_context.agent.limits if tool_context.agent else None
+        budget = limits.steps if limits else tools_service.MAX_ROUNDS
+
+        for round_number in range(budget + 1):
             generation.rounds = round_number + 1
+
+            # Checked between rounds, never mid-stream: cutting a reply off in
+            # the middle of a sentence to enforce a budget produces garbage, and
+            # Stop already covers the mid-stream case. Time spent waiting for a
+            # person is subtracted -- somebody who thinks for ten minutes about
+            # one command should not thereby spend the whole allowance.
+            if limits is not None and round_number:
+                spent = (time.monotonic() - started) - generation.waited
+                if spent > limits.wall_seconds:
+                    _gave_up(generation, f"after {spent / 60:.0f} minutes")
+                    break
+                if generation.output_bytes > limits.output_bytes:
+                    _gave_up(generation, "with too much output to read")
+                    break
             accumulator = tools_service.ToolCallAccumulator()
             # Text the model produced in *this* round, needed separately from
             # generation.content when echoing the assistant turn back.
@@ -380,8 +405,8 @@ async def _run(generation: Generation) -> None:
                         "name": calls[0]["name"],
                         "status": "error",
                         "error": (
-                            f"Stopped after {tools_service.MAX_ROUNDS} rounds of tool "
-                            f"calls without an answer."
+                            f"Stopped after {budget} rounds of tool calls "
+                            f"without an answer."
                         ),
                     }
                 )
@@ -397,20 +422,23 @@ async def _run(generation: Generation) -> None:
             # together under a semaphore, and four people-shaped pauses inside
             # that gather would queue behind each other invisibly -- see
             # services/interaction.py.
-            decided = await _authorise(generation, tool_context, calls)
+            decided, allowed = await _authorise(generation, tool_context, calls)
             if generation.stopped:
                 break
 
             generation.status = _tool_status(calls)
             generation.touch()
             try:
-                outcomes = await _run_calls(tool_context, calls, decided=decided)
+                outcomes = await _run_calls(
+                    tool_context, calls, decided=decided, allowed=allowed
+                )
             finally:
                 generation.status = ""
                 generation.touch()
 
             for call, outcome in zip(calls, outcomes, strict=True):
                 generation.tool_events.append(outcome.event)
+                generation.output_bytes += len(outcome.content)
                 messages.append(tools_service.tool_turn(call, outcome.content))
             generation.touch()
 
@@ -463,6 +491,14 @@ async def _run(generation: Generation) -> None:
                     )
                 title = title or chat_service.fallback_title(question)
 
+        # The decrypted SSH credential dies with the reply rather than with the
+        # object holding it. A finished Generation lingers KEEP_FINISHED so a
+        # follower arriving at the last moment still gets the final frames, and
+        # a private key should not sit in memory for five minutes waiting on
+        # that.
+        if tool_context is not None and getattr(tool_context, "agent", None) is not None:
+            tool_context.agent.clear()
+
         # Written *before* `done`, because `_follow` breaks out of its loop the
         # moment it sees that flag and immediately re-renders the bubble from
         # the row. The other order left a window in which the finished frame
@@ -539,6 +575,25 @@ async def _maybe_compact(generation: Generation) -> None:
 MAX_PARALLEL_TOOLS = 4
 
 
+def _gave_up(generation, why: str) -> None:
+    """Stop, and leave something in the transcript saying why.
+
+    A reply that simply stopped would look like the model losing interest. The
+    event is the same shape the out-of-rounds branch uses, so it renders with
+    everything else.
+    """
+    generation.tool_events.append(
+        {
+            "name": "budget",
+            "kind": "agent",
+            "status": "error",
+            "results": [],
+            "error": f"Stopped {why}. Ask again to carry on from here.",
+        }
+    )
+    generation.touch()
+
+
 def _tool_status(calls: list[dict]) -> str:
     """What to show while tools run.
 
@@ -550,6 +605,81 @@ def _tool_status(calls: list[dict]) -> str:
     return f"Running {len(calls)} tools…"
 
 
+def _arguments_of(call: dict) -> dict:
+    try:
+        args = json.loads(call["arguments"] or "{}")
+    except json.JSONDecodeError:
+        return {}
+    return args if isinstance(args, dict) else {}
+
+
+def _describe(name: str, args: dict) -> tuple[str, str]:
+    """What an approval card says about one call: a title, and the detail.
+
+    The detail is the thing being agreed to -- the command line, the path -- and
+    is shown verbatim and escaped. A summary that paraphrased it would be a card
+    approving something other than what runs.
+    """
+    if name == "shell_run":
+        return "Run a command", str(args.get("command") or "")
+    if name == "file_write":
+        return "Write a file", str(args.get("path") or "")
+    if name == "file_read":
+        return "Read a file", str(args.get("path") or "")
+    if name == "file_list":
+        return "List a directory", str(args.get("path") or "")
+    detail = ", ".join(f"{k}={v!r}" for k, v in list(args.items())[:4])
+    return f"Use {name}", detail[:400]
+
+
+def _approvals(context, calls: list[dict]) -> list[interaction.Item]:
+    """The calls in this round that a person has to allow before they run.
+
+    Only in an agent chat: `context.agent` is None everywhere else, and an
+    ordinary conversation behaves exactly as it did. Within one, *every* call
+    goes through the table, including the built-in ones -- `notes_edit` writes,
+    and Plan mode meaning "look but do not touch" has to mean that too.
+    """
+    agent = getattr(context, "agent", None)
+    if agent is None:
+        return []
+
+    book = context.tools if context.tools is not None else tools_service.REGISTRY
+    items: list[interaction.Item] = []
+
+    for index, call in enumerate(calls):
+        tool = book.get(call["name"])
+        if tool is None or tool.risk == tools_service.RISK_ASK:
+            continue  # unknown names are refused by run_tool; questions are their own card
+
+        args = _arguments_of(call)
+        command = str(args.get("command") or "") if call["name"] == "shell_run" else ""
+        decision = agent_policy.decide(
+            mode=agent.mode,
+            risk=tool.risk,
+            tool_name=call["name"],
+            command=command,
+            allow=agent.allow,
+            deny=agent.deny,
+        )
+        if decision.verdict == agent_policy.ALLOW:
+            continue
+
+        title, detail = _describe(call["name"], args)
+        items.append(
+            interaction.Item(
+                index=index,
+                key=f"a{index}",
+                kind=interaction.KIND_APPROVAL,
+                tool_name=call["name"],
+                title=f"{title} on {agent.label}",
+                detail=detail,
+                reason=decision.reason,
+            )
+        )
+    return items
+
+
 def _ask_items(context, calls: list[dict]) -> list[interaction.Item]:
     """Which of this round's calls need a person, and what to show about each.
 
@@ -565,12 +695,7 @@ def _ask_items(context, calls: list[dict]) -> list[interaction.Item]:
         tool = book.get(call["name"])
         if tool is None or tool.risk != tools_service.RISK_ASK:
             continue
-        try:
-            args = json.loads(call["arguments"] or "{}")
-        except json.JSONDecodeError:
-            args = {}
-        if not isinstance(args, dict):
-            args = {}
+        args = _arguments_of(call)
 
         for asked in _questions_in(args):
             options = [str(o).strip() for o in (asked.get("options") or []) if str(o).strip()]
@@ -617,7 +742,9 @@ def _questions_in(args: dict) -> list[dict]:
     return out
 
 
-async def _authorise(generation, context, calls: list[dict]) -> dict[int, ToolOutcome]:
+async def _authorise(
+    generation, context, calls: list[dict]
+) -> tuple[dict[int, ToolOutcome], set[int]]:
     """Which of this round's calls may run, and what the others answer instead.
 
     Returns outcomes keyed by the call's index. Every index the caller does not
@@ -625,10 +752,17 @@ async def _authorise(generation, context, calls: list[dict]) -> dict[int, ToolOu
     the runner being reached at all. That is what keeps
     `zip(calls, outcomes, strict=True)` aligned -- an endpoint matching on
     `tool_call_id` pairs the wrong content with the right id otherwise.
+
+    Also returns the indices a person explicitly allowed, so the runners can be
+    told. They re-check the mode as a backstop and would otherwise refuse the
+    very thing that was just approved -- the mode says "ask", and asking is what
+    happened.
     """
-    items = _ask_items(context, calls)
+    questions = _ask_items(context, calls)
+    approvals = _approvals(context, calls)
+    items = [*approvals, *questions]
     if not items:
-        return {}
+        return {}, set()
 
     timeout = float(context.interaction_timeout or 900)
     pause = interaction.build(uuid.uuid4().hex, items, timeout=timeout)
@@ -638,14 +772,56 @@ async def _authorise(generation, context, calls: list[dict]) -> dict[int, ToolOu
 
     if reply.ended:
         generation.stopped = True
-        return {}
+        return {}, set()
 
-    # Grouped back by call, because one `ask_user` call may have carried several
-    # questions and the endpoint expects exactly one tool turn per call.
+    decided: dict[int, ToolOutcome] = {}
+    allowed: set[int] = set()
+
+    # An approval that came back as a refusal answers its call without the
+    # runner being reached; one that came back allowed is simply left out, which
+    # is how `_run_calls` is told to go ahead.
+    for item in approvals:
+        if reply.permitted:
+            allowed.add(item.index)
+            continue
+        decided[item.index] = _not_allowed(item, reply)
+
+    # Questions are grouped back by call, because one `ask_user` call may have
+    # carried several and the endpoint expects exactly one tool turn per call.
     grouped: dict[int, list[interaction.Item]] = {}
-    for item in items:
+    for item in questions:
         grouped.setdefault(item.index, []).append(item)
-    return {index: _answered(asked, reply) for index, asked in grouped.items()}
+    for index, asked in grouped.items():
+        decided[index] = _answered(asked, reply)
+
+    return decided, allowed
+
+
+def _not_allowed(item: interaction.Item, reply: interaction.Reply) -> ToolOutcome:
+    """What the model is told when a person declined, or never answered.
+
+    Told plainly, and told to stop rather than to try again: a model that reads
+    "not allowed" as "not allowed *that way*" will spend the rest of the reply
+    looking for a way round, which is the opposite of what the refusal meant.
+    """
+    event = {
+        "name": item.tool_name,
+        "kind": "agent",
+        "label": item.title,
+        "query": item.detail,
+        "results": [],
+    }
+    if reply.outcome == interaction.EXPIRED:
+        return ToolOutcome(
+            "Nobody answered, so this was not run. Stop and say what you were "
+            "about to do and why.",
+            {**event, "status": "error", "error": "Not answered."},
+        )
+    return ToolOutcome(
+        "They declined this. Do not try it another way — say what you were "
+        "going to do and ask what they would prefer.",
+        {**event, "status": "error", "error": "Declined."},
+    )
 
 
 def _answered(items: list[interaction.Item], reply: interaction.Reply) -> ToolOutcome:
@@ -688,7 +864,11 @@ def _answered(items: list[interaction.Item], reply: interaction.Reply) -> ToolOu
 
 
 async def _run_calls(
-    context, calls: list[dict], *, decided: dict[int, ToolOutcome] | None = None
+    context,
+    calls: list[dict],
+    *,
+    decided: dict[int, ToolOutcome] | None = None,
+    allowed: set[int] | None = None,
 ) -> list:
     """Run one round's calls together, results in call order.
 
@@ -713,8 +893,17 @@ async def _run_calls(
         # occupies its index, because the tool turns have to line up.
         if decided and index in decided:
             return decided[index]
+
+        # A per-call copy for anything a person allowed, so the runner's own
+        # check does not undo their decision. A copy rather than a flag on the
+        # shared context, because a round runs its calls together and only some
+        # of them were approved.
+        ctx = context
+        if allowed and index in allowed and getattr(context, "agent", None) is not None:
+            ctx = replace(context, agent=context.agent.as_approved())
+
         async with limit:
-            return await tools_service.run_tool(context, call["name"], call["arguments"])
+            return await tools_service.run_tool(ctx, call["name"], call["arguments"])
 
     return list(await asyncio.gather(*(one(i, c) for i, c in enumerate(calls))))
 

src/lembas/services/harness.py

@@ -131,6 +131,10 @@ def context_variables(
         "skills": skills_service.index_block(db, user) if "skills" in families else "",
         "knowledge_bases": "",
         "document_names": "",
+        "agent_target": "",
+        "agent_dir": "",
+        "agent_mode": "",
+        "agent_rewound": "",
     }
 
     if chat is not None:
@@ -145,9 +149,38 @@ def context_variables(
             values["knowledge_bases"] = ", ".join(base.name for base in chat.knowledge_bases)
         values["document_names"] = _document_names(db, chat)
 
+        # The one thing a tool description cannot carry, because a description
+        # is schema: which machine, which directory, and what this chat's mode
+        # currently permits. `max_rounds` is corrected here too, or an agent
+        # chat with forty rounds is told it has three.
+        if "agent" in families:
+            values.update(_agent_values(db, chat, user))
+
     return values
 
 
+def _agent_values(db: DBSession, chat, user) -> dict[str, str]:
+    """What an agent chat's harness needs to say about where it is."""
+    from lembas.services.agent import policy
+    from lembas.services.agent import session as agent_session
+
+    context = agent_session.resolve(db, chat, user)
+    if context is None:
+        return {}
+
+    rewound = ""
+    if getattr(chat, "rewound_at", None) is not None:
+        rewound = chat.rewound_at.strftime("on %-d %B at %H:%M")
+
+    return {
+        "agent_target": context.label,
+        "agent_dir": context.project_dir or "the login directory",
+        "agent_mode": policy.MODE_GUIDANCE.get(context.mode, ""),
+        "agent_rewound": rewound,
+        "max_rounds": str(context.limits.steps),
+    }
+
+
 def limit_for(db: DBSession) -> int:
     """The ceiling on the assembled block."""
     stored = settings_store.get(db, "max_harness_chars", key=settings_store.PROMPTS)

src/lembas/services/prompts.py

@@ -134,6 +134,27 @@ VARIABLES: tuple[Variable, ...] = (
         "The character limit on a single remembered fact.",
     ),
     Variable("tool_names", "Tool names", "The tools offered on this request, comma separated."),
+    Variable(
+        "agent_target",
+        "Agent machine",
+        "The connection an agent chat acts on. Empty in an ordinary chat.",
+    ),
+    Variable(
+        "agent_dir",
+        "Project directory",
+        "Where commands start on that machine, and what relative paths mean.",
+    ),
+    Variable(
+        "agent_mode",
+        "Agent mode",
+        "Which of Manual, Edit, Auto or Plan is in force, and what it permits.",
+    ),
+    Variable(
+        "agent_rewound",
+        "Rewound at",
+        "When an agent chat was last edited or regenerated. Empty otherwise, "
+        "which is what keeps the note about it out of every other reply.",
+    ),
     Variable(
         "memories",
         "Memories",
@@ -754,6 +775,55 @@ BUILTIN: tuple[Fragment, ...] = (
             "Read one with skill_get before following it."
         ),
     ),
+    Fragment(
+        key="tool.agent",
+        label="Acting on a machine",
+        group=GROUP_TOOLS,
+        order=250,
+        families=("agent",),
+        variables=("agent_target", "agent_dir", "agent_mode"),
+        requires=("agent_target",),
+        hint="Appears in an agent chat. Says which machine, which directory and "
+        "what the mode permits -- none of which can go in a tool description, "
+        "because those are schema and cannot change per chat.",
+        default=(
+            "### Acting on {{agent_target}}\n"
+            "\n"
+            "- You are working on **{{agent_target}}**, in `{{agent_dir}}`. That is "
+            "where commands start and what a relative path is measured from. "
+            "Nothing you do reaches the machine LLeMbas itself runs on.\n"
+            "- **Each command is a fresh shell.** A `cd` in one call is gone by the "
+            "next, so pass `cwd` instead of chaining directory changes.\n"
+            "- Nothing can answer a prompt. Pass the flags that make a command "
+            "non-interactive — `-y`, `--no-input`, `--yes` — rather than waiting "
+            "for it to ask. On a Debian-derived system `apt-get install` needs an "
+            "`apt-get update` first or it reports the package as missing.\n"
+            "- Look before you write. Read a file before replacing it, and list a "
+            "directory before guessing at a path.\n"
+            "- {{agent_mode}}\n"
+            "- If something is refused, say what you were going to do and ask. Do "
+            "not look for another way round it."
+        ),
+    ),
+    Fragment(
+        key="tool.agent_rewound",
+        label="After a rewind",
+        group=GROUP_CONTEXT,
+        order=330,
+        families=("agent",),
+        requires=("agent_rewound",),
+        variables=("agent_rewound", "agent_target"),
+        hint="Only after a turn in an agent chat was edited or regenerated. The "
+        "transcript rewinds; the machine does not.",
+        default=(
+            "### This conversation was rewound\n"
+            "\n"
+            "Turns were edited or regenerated {{agent_rewound}}, but "
+            "{{agent_target}} was not. Files created or changed by steps no longer "
+            "in the transcript are still there. Check before assuming anything is "
+            "unmade."
+        ),
+    ),
     # --- Tasks ---------------------------------------------------------------
     Fragment(
         key="task.title",

src/lembas/services/tools.py

@@ -73,6 +73,11 @@ FAMILY_MCP = "mcp"
 # is the only tool the model cannot resolve by itself.
 FAMILY_ASK = "ask"
 
+# Acting on the machine an agent chat is pointed at. Offered only when the chat
+# is one, has a usable connection, and the feature is switched on -- see
+# services/agent/session.py:resolve, which answers all three at once.
+FAMILY_AGENT = "agent"
+
 # The built-in families, in the order they are offered.
 FAMILIES = (
     FAMILY_SEARCH,
@@ -81,6 +86,7 @@ FAMILIES = (
     FAMILY_MEMORY,
     FAMILY_SKILLS,
     FAMILY_ASK,
+    FAMILY_AGENT,
 )
 
 GATES = (*FAMILIES, FAMILY_CUSTOM, FAMILY_MCP)
@@ -128,6 +134,10 @@ class ToolContext:
     # something. Read from the instance settings while the session was open,
     # like everything else here.
     interaction_timeout: float = 900.0
+    # Set only for an agent chat: the machine to act on, the mode in force, and
+    # the decrypted credential. None everywhere else, which is what every agent
+    # runner checks first. `generation` clears it when the reply ends.
+    agent: Any = None
 
 
 @dataclass
@@ -825,7 +835,7 @@ def _family_allowed(
             and config.get("enabled")
             and not search_service.availability(str(config.get("provider") or "ddgs"))
         )
-    if gate in (FAMILY_CUSTOM, FAMILY_MCP, FAMILY_ASK):
+    if gate in (FAMILY_CUSTOM, FAMILY_MCP, FAMILY_ASK, FAMILY_AGENT):
         # Deliberately without `library.use`: an HTTP endpoint an administrator
         # wrote has nothing to do with this person's own documents and notes,
         # and requiring the library permission for it would be a coincidence of
@@ -852,6 +862,23 @@ def _row_defs(db: DBSession, user: User | None, *, everything: bool = False) ->
     return [*custom, *mcp_registry.tool_defs(db, user, everything=everything, taken=taken)]
 
 
+def _agent_defs(db: DBSession, chat: Chat | None, user: User | None) -> list[ToolDef]:
+    """The agent tools, when this chat is pointed at a machine it can use.
+
+    Everything that would make them useless -- not an agent chat, the feature
+    switched off, the connection deleted or disabled, SSH not installed -- comes
+    back as an empty list, because offering a tool that fails on its first call
+    is worse than not offering it.
+    """
+    from lembas.services.agent import session as agent_session
+    from lembas.services.agent import tools as agent_tools
+
+    context = agent_session.resolve(db, chat, user) if chat is not None else None
+    if context is None:
+        return []
+    return agent_tools.tool_defs(context)
+
+
 def _book(defs: list[ToolDef]) -> dict[str, ToolDef]:
     """Keyed by name, first claim winning.
 
@@ -872,8 +899,16 @@ def registry(db: DBSession) -> dict[str, ToolDef]:
     an administrator-defined tool is a row. Callers that only need to map a name
     back to a family use this; callers deciding what to *offer* use
     `resolve_tools`, which applies the gates as well.
+
+    The agent tools are listed here **unbound to any chat**. Mapping a name back
+    to its family is exactly what the harness does to decide whether a
+    fragment applies, and without them `shell_run` would resolve to no family at
+    all -- so an agent chat would be told nothing about the machine it is
+    working on. The same omission cost custom tools their guidance once already.
     """
-    return _book(_row_defs(db, None, everything=True))
+    from lembas.services.agent import tools as agent_tools
+
+    return _book([*_row_defs(db, None, everything=True), *agent_tools.tool_defs()])
 
 
 def families(db: DBSession) -> tuple[str, ...]:
@@ -900,7 +935,7 @@ def resolve_tools(db: DBSession, chat: Chat, user: User | None) -> ToolSet:
 
     # Resolved against what this reader may see, not against everything that
     # exists: a tool restricted to a group is not offered outside it.
-    book = _book(_row_defs(db, user))
+    book = _book([*_row_defs(db, user), *_agent_defs(db, chat, user)])
     return ToolSet(
         tuple(
             tool
@@ -929,7 +964,10 @@ def context_for(
     tools: ToolSet | None = None,
 ) -> ToolContext:
     """The snapshot a running tool needs, taken while the session is open."""
+    from lembas.services.agent import session as agent_session
+
     return ToolContext(
+        agent=agent_session.resolve(db, chat, user) if chat is not None else None,
         owner_id=user.id if user else "",
         search_config=settings_store.search(db),
         base_ids=[base.id for base in chat.knowledge_bases] if chat is not None else [],
@@ -1105,6 +1143,7 @@ def _row_source(db: DBSession):
 
 __all__ = [
     "FAMILIES",
+    "FAMILY_AGENT",
     "MAX_ROUNDS",
     "REGISTRY",
     "ToolCallAccumulator",

src/lembas/web/static/css/chat.css

@@ -893,3 +893,31 @@
   box-shadow: 0 0 0 2px color-mix(in srgb, var(--success) 25%, transparent);
 }
 .unread-dot[hidden] { display: none; }
+
+/* --- Agent chats ----------------------------------------------------------- */
+/* In the header, not the settings panel: the mode is the difference between
+   being interrupted and not, and it is looked at constantly. */
+.agent-bar { display: flex; align-items: center; gap: var(--sp-2); }
+.agent-bar__where {
+  display: inline-flex;
+  align-items: center;
+  gap: var(--sp-1);
+  color: var(--ink-muted);
+  font-size: var(--text-xs);
+  white-space: nowrap;
+  max-width: 14rem;
+  overflow: hidden;
+  text-overflow: ellipsis;
+}
+
+/* Chat or Agent, on the new-chat composer. */
+.composer__kind {
+  display: flex;
+  flex-wrap: wrap;
+  align-items: center;
+  gap: var(--sp-2);
+  padding: 0 var(--sp-2) var(--sp-2);
+}
+.composer__kind-agent { display: flex; gap: var(--sp-2); flex: 1 1 18rem; min-width: 0; }
+.composer__kind-agent .input { flex: 1; min-width: 0; }
+.select--sm, .input--sm { height: calc(var(--control-h) - 0.35rem); font-size: var(--text-xs); }

src/lembas/web/static/js/ui.js

@@ -477,3 +477,52 @@ document.addEventListener("lembas:notify", function (event) {
   document.addEventListener("DOMContentLoaded", watch);
   document.body && document.body.addEventListener("htmx:afterSettle", sync);
 })();
+
+/*
+  Chat or Agent, on the new-chat composer.
+
+  Two radios rather than a checkbox because they are two kinds of conversation,
+  not a setting on one -- and the choice is permanent, so it should read as a
+  fork. Picking Agent reveals the connection and directory; picking Chat hides
+  them and sets the hidden `kind` back, so a form submitted either way carries
+  exactly what it means.
+*/
+(function () {
+  function wire(root) {
+    var kind = root.querySelector("#chat-kind") ||
+               root.parentNode.querySelector("#chat-kind");
+    var extra = root.querySelector(".composer__kind-agent");
+    var picker = root.querySelector('select[name="ssh_profile_id"]');
+    var dir = root.querySelector('input[name="project_dir"]');
+    if (!kind || !extra) return;
+
+    function sync() {
+      var chosen = root.querySelector('input[name="kind_choice"]:checked');
+      var agent = chosen && chosen.value === "agent";
+      kind.value = agent ? "agent" : "chat";
+      extra.hidden = !agent;
+    }
+
+    root.addEventListener("change", function (event) {
+      if (event.target.name === "kind_choice") sync();
+      // Following the profile's own directory is a convenience, not a rule:
+      // once someone has typed their own it is left alone.
+      if (event.target === picker && dir && !dir.dataset.touched) {
+        var option = picker.options[picker.selectedIndex];
+        dir.value = (option && option.dataset.dir) || "";
+      }
+    });
+    if (dir) dir.addEventListener("input", function () { dir.dataset.touched = "1"; });
+    sync();
+  }
+
+  function scan() {
+    document.querySelectorAll("[data-agent-picker]").forEach(function (el) {
+      if (!el.dataset.wired) { el.dataset.wired = "1"; wire(el); }
+    });
+  }
+
+  document.addEventListener("DOMContentLoaded", scan);
+  document.body && scan();
+  document.addEventListener("htmx:afterSettle", scan);
+})();

src/lembas/web/templates/chat/_composer.html

@@ -54,6 +54,38 @@
       <input type="hidden" name="temporary" value="true">
       {% endif %}
 
+      {# Chat or Agent, chosen once. There is no switching afterwards: the
+         tools offered, the harness and the approval loop all differ, so a
+         conversation whose earlier turns ran somewhere else is not one
+         conversation. Only shown when picking Agent would lead anywhere. #}
+      {% if not chat and agent_profiles %}
+      <input type="hidden" name="kind" value="chat" id="chat-kind">
+      <div class="composer__kind" data-agent-picker>
+        <label class="chip">
+          <input type="radio" name="kind_choice" value="chat" checked>
+          <span>{{ icon("chat", "icon--sm") }} Chat</span>
+        </label>
+        <label class="chip">
+          <input type="radio" name="kind_choice" value="agent">
+          <span>{{ icon("server", "icon--sm") }} Agent</span>
+        </label>
+
+        <span class="composer__kind-agent" hidden>
+          <select class="select select--sm" name="ssh_profile_id" aria-label="Connection">
+            {% for profile in agent_profiles %}
+            <option value="{{ profile.id }}" data-dir="{{ profile.default_dir }}"
+                    {{ 'disabled' if not profile.verified }}>
+              {{ profile.name }}{{ ' — not checked' if not profile.verified }}
+            </option>
+            {% endfor %}
+          </select>
+          <input class="input input--sm input--mono" name="project_dir"
+                 value="{{ agent_profiles[0].default_dir }}"
+                 aria-label="Project directory" placeholder="/project">
+        </span>
+      </div>
+      {% endif %}
+
       <div class="composer__row">
         {% if can.get("files.upload") %}
         {# A menu rather than the file picker straight away: there are four ways

src/lembas/web/templates/chat/index.html

@@ -24,6 +24,26 @@
         <span id="chat-title">{{ chat.title if chat else "New chat" }}</span>
       </h1>
 
+      {# The mode is the one agent setting that changes mid-chat: it decides
+         what gets asked about, not what the conversation is. In the header
+         rather than the settings panel because it is looked at constantly --
+         it is the difference between being interrupted and not. #}
+      {% if chat and chat.kind == "agent" %}
+      <form class="agent-bar" hx-post="/api/chats/{{ chat.id }}" hx-swap="none"
+            hx-trigger="change">
+        <span class="agent-bar__where" title="{{ chat.project_dir }}">
+          {{ icon("server", "icon--sm") }}
+          {{ agent_profile.name if agent_profile else "connection missing" }}
+        </span>
+        <select class="select select--sm" name="agent_mode" aria-label="Mode">
+          {% for value, label, hint in agent_modes %}
+          <option value="{{ value }}" title="{{ hint }}"
+                  {{ 'selected' if value == chat.agent_mode }}>{{ label }}</option>
+          {% endfor %}
+        </select>
+      </form>
+      {% endif %}
+
       <div class="topbar__actions">
         {#
           A link, not a script: the flag lives in the URL, so it survives a

tests/test_agent_tools.py

@@ -0,0 +1,485 @@
+"""The agent tools: when they are offered, and what stops them running.
+
+Everything here goes through a real SSH server on 127.0.0.1, so the gate and the
+approval loop are exercised against something that genuinely executes rather
+than a stub that always agrees.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from lembas.db.models import (
+    KIND_AGENT,
+    ROLE_ASSISTANT,
+    Chat,
+    Connection,
+    Message,
+    Model,
+    SshProfile,
+    User,
+)
+from lembas.services import generation as generation_service
+from lembas.services import interaction, settings_store
+from lembas.services import tools as tools_service
+from lembas.services.agent import policy, session
+from lembas.services.agent import ssh as ssh_service
+
+asyncssh = pytest.importorskip("asyncssh")
+
+
+# --- A machine to act on --------------------------------------------------------
+class _Server(asyncssh.SSHServer):
+    def begin_auth(self, username: str) -> bool:
+        return False
+
+
+async def _handler(process):
+    command = process.command or ""
+    process.stdout.write(f"ran: {command}\n")
+    process.exit(0)
+
+
+@pytest.fixture
+async def machine(tmp_path):
+    """A real sshd, and a profile pointing at it with its key already pinned."""
+    project = tmp_path / "project"
+    project.mkdir()
+
+    server = await asyncssh.create_server(
+        _Server,
+        "127.0.0.1",
+        0,
+        server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
+        process_factory=_handler,
+        sftp_factory=True,
+    )
+    port = next(iter(server.sockets)).getsockname()[1]
+    line, fingerprint = await ssh_service.capture_host_key("127.0.0.1", port)
+    try:
+        yield {"port": port, "host_key": line, "fingerprint": fingerprint, "dir": str(project)}
+    finally:
+        server.close()
+        await server.wait_closed()
+
+
+def _setup(db, user_id, machine, *, mode=policy.MODE_MANUAL, enabled=True, kind=KIND_AGENT):
+    """An agent chat pointed at the machine, with the feature switched on."""
+    settings_store.update(db, {"enabled": enabled}, key=settings_store.AGENTS)
+
+    profile = SshProfile(
+        owner_id=user_id,
+        name="Test box",
+        host="127.0.0.1",
+        port=machine["port"],
+        username="tester",
+        host_key=machine["host_key"],
+        host_fingerprint=machine["fingerprint"],
+        default_dir=machine["dir"],
+    )
+    db.add(profile)
+    db.commit()
+
+    connection = Connection(name="c", base_url="http://127.0.0.1:1", api_key_encrypted="")
+    db.add(connection)
+    db.commit()
+    db.add(
+        Model(
+            connection_id=connection.id,
+            model_id="m",
+            capabilities_json={"tools": True},
+        )
+    )
+    db.commit()
+
+    chat = Chat(
+        user_id=user_id,
+        model_id="m",
+        connection_id=connection.id,
+        kind=kind,
+        ssh_profile_id=profile.id,
+        project_dir=machine["dir"],
+        agent_mode=mode,
+    )
+    db.add(chat)
+    db.commit()
+    return chat, profile
+
+
+def _offered(db, chat, user) -> set[str]:
+    return set(tools_service.resolve_tools(db, chat, user).by_name)
+
+
+# --- The gate --------------------------------------------------------------------
+async def test_the_agent_tools_are_offered_to_an_agent_chat(db, user_id, machine):
+    chat, _profile = _setup(db, user_id, machine)
+    names = _offered(db, chat, db.get(User, user_id))
+
+    assert {"shell_run", "file_read", "file_write", "file_list"} <= names
+
+
+async def test_an_ordinary_chat_gets_none_of_them(db, user_id, machine):
+    """A plain conversation can never shell out, whatever it is asked."""
+    chat, _profile = _setup(db, user_id, machine, kind="chat")
+    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))
+
+
+async def test_nothing_is_offered_while_the_feature_is_off(db, user_id, machine):
+    chat, _profile = _setup(db, user_id, machine, enabled=False)
+    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))
+
+
+async def test_nothing_is_offered_without_the_permission(db, user_id, machine):
+    chat, _profile = _setup(db, user_id, machine)
+    user = db.get(User, user_id)
+    user.role = "user"  # administrators pass everything
+    settings_store.update(db, {"default_permissions": {"tools.agent": False}})
+    db.commit()
+
+    assert "shell_run" not in _offered(db, chat, user)
+
+
+async def test_nothing_is_offered_without_the_model_capability(db, user_id, machine):
+    chat, _profile = _setup(db, user_id, machine)
+    model = db.scalar(tools_service.select(Model))
+    model.capabilities_json = {"tools": True, "tool_agent": False}
+    db.commit()
+
+    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))
+
+
+async def test_a_disabled_connection_takes_the_tools_away(db, user_id, machine):
+    """Offering a tool that fails on its first call is worse than not offering
+    it, so every "no" collapses to an empty list."""
+    chat, profile = _setup(db, user_id, machine)
+    profile.enabled = False
+    db.commit()
+
+    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))
+
+
+async def test_somebody_elses_connection_is_not_reachable(db, user_id, machine):
+    from lembas.security.passwords import hash_password
+
+    chat, profile = _setup(db, user_id, machine)
+    intruder = User(name="Sam", email="s@shire.test", password_hash=hash_password("x"))
+    intruder.role = "user"
+    db.add(intruder)
+    db.commit()
+
+    assert session.resolve(db, chat, intruder) is None
+
+
+# --- Running something -----------------------------------------------------------
+async def test_a_command_runs_on_the_machine(db, user_id, machine):
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
+    user = db.get(User, user_id)
+    resolved = tools_service.resolve_tools(db, chat, user)
+    context = tools_service.context_for(db, user, chat, tools=resolved)
+
+    outcome = await tools_service.run_tool(
+        context, "shell_run", '{"command": "echo hello"}'
+    )
+    assert outcome.event["status"] == "ok"
+    assert "echo hello" in outcome.content
+
+
+async def test_files_are_written_and_read_back(db, user_id, machine, tmp_path):
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
+    user = db.get(User, user_id)
+    resolved = tools_service.resolve_tools(db, chat, user)
+    context = tools_service.context_for(db, user, chat, tools=resolved)
+
+    await tools_service.run_tool(
+        context, "file_write", '{"path": "note.txt", "content": "a mallorn tree"}'
+    )
+    assert (tmp_path / "project" / "note.txt").read_text() == "a mallorn tree"
+
+    read = await tools_service.run_tool(context, "file_read", '{"path": "note.txt"}')
+    assert "mallorn" in read.content
+
+    listed = await tools_service.run_tool(context, "file_list", "{}")
+    assert "note.txt" in listed.content
+
+
+# --- The runner backstop ----------------------------------------------------------
+async def test_a_runner_refuses_what_the_mode_forbids(db, user_id, machine):
+    """`_authorise` is the real gate and runs first. This is the belt to that
+    brace: a call arriving by some other path -- a retry, a re-run button --
+    must not walk past it."""
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_PLAN)
+    user = db.get(User, user_id)
+    resolved = tools_service.resolve_tools(db, chat, user)
+    context = tools_service.context_for(db, user, chat, tools=resolved)
+
+    outcome = await tools_service.run_tool(context, "shell_run", '{"command": "rm -rf /"}')
+    assert outcome.event["status"] == "error"
+    assert "not allowed" in outcome.content.lower()
+
+
+async def test_plan_mode_still_reads(db, user_id, machine, tmp_path):
+    (tmp_path / "project" / "readme.txt").write_text("contents")
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_PLAN)
+    user = db.get(User, user_id)
+    resolved = tools_service.resolve_tools(db, chat, user)
+    context = tools_service.context_for(db, user, chat, tools=resolved)
+
+    outcome = await tools_service.run_tool(context, "file_read", '{"path": "readme.txt"}')
+    assert outcome.event["status"] == "ok"
+    assert "contents" in outcome.content
+
+
+async def test_edit_mode_writes_but_will_not_run(db, user_id, machine):
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_EDIT)
+    user = db.get(User, user_id)
+    resolved = tools_service.resolve_tools(db, chat, user)
+    context = tools_service.context_for(db, user, chat, tools=resolved)
+
+    wrote = await tools_service.run_tool(
+        context, "file_write", '{"path": "x.txt", "content": "hi"}'
+    )
+    assert wrote.event["status"] == "ok"
+
+    ran = await tools_service.run_tool(context, "shell_run", '{"command": "ls"}')
+    assert ran.event["status"] == "error"
+
+
+# --- The approval card in the loop -------------------------------------------------
+def _chunk(name: str, arguments: str, *, call_id="c1", index=0):
+    return {
+        "choices": [
+            {
+                "delta": {
+                    "tool_calls": [
+                        {
+                            "index": index,
+                            "id": call_id,
+                            "function": {"name": name, "arguments": arguments},
+                        }
+                    ]
+                }
+            }
+        ]
+    }
+
+
+def _text(text: str) -> dict:
+    return {"choices": [{"delta": {"content": text}}]}
+
+
+def _stub_stream(rounds, seen):
+    async def stream_chat(_endpoint, payload):
+        seen.append(payload)
+        for chunk in rounds[min(len(seen) - 1, len(rounds) - 1)]:
+            yield chunk
+
+    return stream_chat
+
+
+async def _until_paused(generation, *, timeout: float = 3.0):
+    deadline = asyncio.get_running_loop().time() + timeout
+    while asyncio.get_running_loop().time() < deadline:
+        if generation.pending is not None:
+            return generation.pending
+        await asyncio.sleep(0.01)
+    raise AssertionError("the reply never paused for approval")
+
+
+def _pending_reply(db, chat):
+    db.add(Message(chat_id=chat.id, role="user", content="do it", complete=True))
+    db.commit()
+    assistant = Message(chat_id=chat.id, role=ROLE_ASSISTANT, content="", complete=False)
+    db.add(assistant)
+    db.commit()
+    return assistant.id
+
+
+async def test_a_command_waits_for_approval_and_the_card_shows_it(
+    db, user_id, machine, monkeypatch
+):
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
+    message_id = _pending_reply(db, chat)
+    monkeypatch.setattr(
+        generation_service,
+        "stream_chat",
+        _stub_stream(
+            [[_chunk("shell_run", '{"command": "rm -rf /tmp/x"}')], [_text("Done.")]], []
+        ),
+    )
+
+    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
+    task = asyncio.create_task(generation_service._run(generation))
+    pending = await _until_paused(generation)
+
+    assert pending.kind == interaction.KIND_APPROVAL
+    item = pending.items[0]
+    # The exact command, verbatim. A card that paraphrased it would be
+    # approving something other than what runs.
+    assert item.detail == "rm -rf /tmp/x"
+    assert "Test box" in item.title
+    assert "Manual" in item.reason
+
+    pending.resolve(interaction.DENY)
+    await task
+
+
+async def test_denying_reaches_the_model_as_words_and_runs_nothing(
+    db, user_id, machine, monkeypatch, tmp_path
+):
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
+    message_id = _pending_reply(db, chat)
+    payloads: list[dict] = []
+    monkeypatch.setattr(
+        generation_service,
+        "stream_chat",
+        _stub_stream(
+            [
+                [_chunk("file_write", '{"path": "never.txt", "content": "nope"}')],
+                [_text("Understood.")],
+            ],
+            payloads,
+        ),
+    )
+
+    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
+    task = asyncio.create_task(generation_service._run(generation))
+    pending = await _until_paused(generation)
+    pending.resolve(interaction.DENY)
+    await task
+
+    turns = [m for m in payloads[1]["messages"] if m.get("role") == "tool"]
+    assert "declined" in turns[0]["content"].lower()
+    # And it says not to look for another way round, because a model reads
+    # "not allowed" as "not allowed like that" otherwise.
+    assert "another way" in turns[0]["content"]
+    assert not (tmp_path / "project" / "never.txt").exists()
+
+
+async def test_allowing_runs_it(db, user_id, machine, monkeypatch, tmp_path):
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
+    message_id = _pending_reply(db, chat)
+    payloads: list[dict] = []
+    monkeypatch.setattr(
+        generation_service,
+        "stream_chat",
+        _stub_stream(
+            [
+                [_chunk("file_write", '{"path": "yes.txt", "content": "written"}')],
+                [_text("Done.")],
+            ],
+            payloads,
+        ),
+    )
+
+    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
+    task = asyncio.create_task(generation_service._run(generation))
+    pending = await _until_paused(generation)
+    pending.resolve(interaction.ALLOW)
+    await task
+
+    assert (tmp_path / "project" / "yes.txt").read_text() == "written"
+
+
+async def test_auto_mode_never_pauses(db, user_id, machine, monkeypatch, tmp_path):
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
+    message_id = _pending_reply(db, chat)
+    monkeypatch.setattr(
+        generation_service,
+        "stream_chat",
+        _stub_stream(
+            [
+                [_chunk("file_write", '{"path": "auto.txt", "content": "no asking"}')],
+                [_text("Done.")],
+            ],
+            [],
+        ),
+    )
+
+    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
+    await asyncio.wait_for(generation_service._run(generation), timeout=10)
+
+    assert generation.pending is None
+    assert (tmp_path / "project" / "auto.txt").read_text() == "no asking"
+
+
+async def test_the_credential_is_cleared_when_the_reply_ends(db, user_id, machine, monkeypatch):
+    """A finished Generation lingers five minutes so late followers get the
+    final frames. A private key should not linger with it."""
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
+    message_id = _pending_reply(db, chat)
+
+    captured = {}
+    original = tools_service.context_for
+
+    def capture(*args, **kwargs):
+        context = original(*args, **kwargs)
+        captured["context"] = context
+        return context
+
+    monkeypatch.setattr(tools_service, "context_for", capture)
+    monkeypatch.setattr(
+        generation_service, "stream_chat", _stub_stream([[_text("Nothing to do.")]], [])
+    )
+
+    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
+    await generation_service._run(generation)
+
+    assert captured["context"].agent is not None
+    assert captured["context"].agent.spec == {}, "the decrypted credential is dropped"
+
+
+# --- The harness has to be able to name the machine ---------------------------
+def test_the_registry_maps_the_agent_tools_to_their_family(db):
+    """`harness._families` maps an offered tool *name* back to a family to
+    decide which fragments apply, and it has no chat to resolve against. Without
+    the agent tools listed here, `shell_run` resolves to no family and an agent
+    chat is told nothing about the machine it is working on -- the same omission
+    that cost custom tools their guidance once already."""
+    book = tools_service.registry(db)
+    for name in ("shell_run", "file_read", "file_write", "file_list"):
+        assert name in book, name
+        assert book[name].family == "agent"
+
+
+async def test_the_harness_says_where_and_under_what_rules(db, user_id, machine):
+    from lembas.services import harness
+
+    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
+    user = db.get(User, user_id)
+    offered = tools_service.resolve_tools(db, chat, user).schemas
+    text = harness.compose(db, user, offered, chat)
+
+    assert "Test box" in text, "which machine"
+    assert machine["dir"] in text, "which directory"
+    assert "Manual" in text, "what the mode permits"
+    # The single most likely cause of "the agent seems stupid": `cd build`
+    # followed by `make` fails silently otherwise.
+    assert "fresh shell" in text
+
+
+async def test_an_ordinary_chat_is_told_none_of_it(db, user_id, machine):
+    from lembas.services import harness
+
+    chat, _profile = _setup(db, user_id, machine, kind="chat")
+    user = db.get(User, user_id)
+    offered = tools_service.resolve_tools(db, chat, user).schemas
+    text = harness.compose(db, user, offered, chat)
+
+    assert "Test box" not in text
+    assert "fresh shell" not in text
+
+
+async def test_an_agent_chat_is_told_its_real_round_budget(db, user_id, machine):
+    """MAX_ROUNDS is three. An agent chat gets forty, and telling it three would
+    be a false fact about its own budget on every turn."""
+    from lembas.services import harness
+
+    settings_store.update(db, {"max_steps": 25}, key=settings_store.AGENTS)
+    chat, _profile = _setup(db, user_id, machine)
+    user = db.get(User, user_id)
+    values = harness.context_variables(
+        db, user, tools_service.resolve_tools(db, chat, user).schemas, chat
+    )
+    assert values["max_rounds"] == "25"