"""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
import posixpath
from typing import Any

from lembas.services import plans
from lembas.services.agent import index, instructions, jobs, patch, policy
from lembas.services.agent.base import ExecError, ExecRequest, clean_output
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

# And how much of a diff. Same reasoning as the constant above and the same
# ceiling in spirit: a generated file's diff can be larger than the file, and
# this one is stored on the row forever and re-parsed on every page load.
MAX_DIFF_LINES = 200

_STRING = {"type": "string"}

# What the model says it is doing, offered on everything that changes something
# or that stops for approval. It is shown to the person -- above the command on
# an approval card, and beside the call in the transcript when nothing stopped
# for approval at all -- which is the only reason it exists: in Auto mode a
# reader otherwise watches a list of commands with no account of what they are
# for until the reply ends.
#
# Not on `file_read`, `file_list` or `file_search`. They are the hot path, their
# detail says everything ("Read src/main.py"), and a schema property costs
# tokens on every request whether or not it is filled in.
_WHY = {
    **_STRING,
    "description": (
        "One short line saying what you are doing this for, in plain language. "
        "It is shown to the person — beside the command when they are asked to "
        "approve it, and in the transcript when they are not."
    ),
}

# One line, and short. It goes in a summary line beside the command, and it is
# stored on the message row forever.
MAX_WHY_CHARS = 240


def why_of(args: dict[str, Any]) -> str:
    """What the model said this call is for, as one short line."""
    return " ".join(str(args.get("why") or "").split())[:MAX_WHY_CHARS]


def _explained(run):
    """Wrap a runner so whatever it returns carries the model's explanation.

    Applied at the `ToolDef`, next to the schema that declares `why`, so the two
    halves cannot drift apart -- a tool that offers the argument records it, and
    one that does not offer it never sees it.

    A wrapper rather than a parameter threaded through, because `shell_run`
    alone builds its outcome in five places -- foreground, convertible,
    launched, backgrounded and the shared formatter -- and none of them has any
    other reason to know this exists. `ToolOutcome.event` is a plain mutable
    dict, so every path through a runner is covered by one line here.
    """

    async def wrapped(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
        outcome = await run(context, args)
        if why := why_of(args):
            outcome.event["why"] = why
        return outcome

    return wrapped


def _event(name: str, context: AgentContext, summary: str, **extra: Any) -> dict[str, Any]:
    """One line in the transcript for one call.

    No `label`. What a tool is called is decided by `services/tool_labels.py`,
    for every tool at once -- this used to write the SSH profile's name here, so
    a bubble said "homeserver · ls -la" and named the machine rather than the
    thing that was done. The machine is a fact about *where*, so it belongs with
    the directory in `detail`, which the template already renders in the body.
    """
    where = context.label
    if context.project_dir:
        where = f"{where}:{context.project_dir}"
    return {
        "name": name,
        "kind": "agent",
        "query": summary,
        "detail": where,
        "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)

    cwd = str(args.get("cwd") or "").strip()
    timeout = _timeout(args.get("timeout"), agent)
    background = bool(args.get("background"))

    # The explicit choice: launch detached and return at once.
    if background and agent.background:
        return await _run_background(agent, command, cwd)

    # A timed-out command becomes a job only when the feature AND the auto-convert
    # are both on. Otherwise -- feature off, or auto-convert off -- the plain path
    # runs, which is byte-for-byte what shell_run always did: killed on timeout,
    # nothing left running. Routing the plain case through the detached wrapper
    # would leave an orphan running past a timeout an administrator said to kill.
    if agent.background and agent.background_on_timeout:
        return await _run_convertible(agent, command, cwd, timeout)

    return await _run_foreground(agent, command, cwd, timeout)


async def _run_foreground(
    agent: AgentContext, command: str, cwd: str, timeout: float
) -> ToolOutcome:
    """One command, run to completion or killed at the timeout. The original."""
    try:
        result = await agent.executor().run(
            ExecRequest(command=command, cwd=cwd, 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)
        )
    return _shell_outcome(
        agent,
        command,
        result.output.strip(),
        exit_status=result.exit_status,
        timed_out=result.timed_out,
        timeout=timeout,
    )


async def _run_convertible(
    agent: AgentContext, command: str, cwd: str, timeout: float
) -> ToolOutcome:
    """Launch detached and wait; if it outlasts the timeout, keep it as a job.

    While it finishes in time this is indistinguishable from `_run_foreground` --
    same output, same wording. The difference is only visible when it does not:
    instead of being killed, it is left running and handed back as a job id.
    """
    job_id = jobs.new_id()
    wrapper = jobs.launch_and_wait_command(agent.chat_id, job_id, command, agent.max_output)
    try:
        result = await agent.executor().run(
            ExecRequest(command=wrapper, cwd=cwd, 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)
        )

    if result.timed_out:
        job = jobs.JobState(id=job_id, chat_id=agent.chat_id, command=command)
        jobs.register(job)
        jobs.start_watch(agent, job)
        return _backgrounded(agent, command, job, converted=True, timeout=timeout)

    # It finished. The wrapper already tailed the log remotely; strip ANSI here.
    output, _ = clean_output(result.output or "", limit=agent.max_output)
    done = jobs.parse_completed(output, job_id)
    return _shell_outcome(
        agent, command, done.body, exit_status=done.exit_status, timed_out=False, timeout=timeout
    )


async def _run_background(agent: AgentContext, command: str, cwd: str) -> ToolOutcome:
    """The model asked to background it up front."""
    try:
        job = await jobs.launch(agent, command, cwd)
    except ExecError as exc:
        return ToolOutcome(
            exc.message, _event("shell_run", agent, command, status="error", error=exc.message)
        )
    jobs.start_watch(agent, job)
    return _backgrounded(agent, command, job, converted=False, timeout=0)


def _backgrounded(
    agent: AgentContext, command: str, job: jobs.JobState, *, converted: bool, timeout: float
) -> ToolOutcome:
    lead = (
        f"Still running after {timeout:g}s, so it was kept running in the background"
        if converted
        else "Started in the background"
    )
    tail = (
        " You will be told when it finishes."
        if agent.background_notify
        else f' Check on it with job_output("{job.id}").'
    )
    return ToolOutcome(
        f'{lead} as job {job.id}. It keeps running after this reply.{tail}',
        _event(
            "shell_run",
            agent,
            command,
            status="ok",
            text=f"job {job.id} — running in the background",
        ),
    )


def _shell_outcome(
    agent: AgentContext,
    command: str,
    body: str,
    *,
    exit_status: int | None,
    timed_out: bool,
    timeout: float,
) -> ToolOutcome:
    if timed_out:
        head = f"The command was stopped after {timeout:g}s."
    elif exit_status == 0:
        head = "" if body else "It ran, and printed nothing."
    elif exit_status is None:
        head = "It stopped before its exit status could be read."
    else:
        head = f"It exited {exit_status}."

    ok = exit_status == 0 and not timed_out
    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 ok else "error",
            error="" if 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 ---------------------------------------------------------------------
def _path_key(agent: AgentContext, path: str) -> str:
    """One name for one file, so `./a.py` and `a.py` are the same file.

    Relative paths are resolved against the project directory, which is what the
    executor does with them, so the two cannot disagree about what was read.
    """
    if not posixpath.isabs(path) and agent.project_dir:
        path = posixpath.join(agent.project_dir, path)
    return posixpath.normpath(path)


def _canvas(agent: AgentContext, path: str) -> dict[str, str]:
    """"This file should be on screen."

    Written onto the event because a runner cannot write the message row --
    `_persist` is the single writer -- so the generation loop carries it, in
    exactly the way it carries a merged plan.

    The key comes from `_path_key`, the same normaliser the read-path set uses,
    so a tab a model opened and a tab a person opened are one tab rather than
    two spellings of the same file.

    It never brings the tab to the front; see `canvas.open_tab`. This rides on
    calls the model was already making, so it costs no schema and no tokens.
    """
    return {
        "key": f"agent:{_path_key(agent, path)}",
        "title": posixpath.basename(path) or path,
        "source": "agent",
    }


def _forget_instructions(agent: AgentContext, path: str) -> None:
    """Drop the cached AGENTS.md when the thing just written *is* it.

    The one case its TTL cannot cover: this process changing the file it has
    been quoting into every request for the last five minutes.
    """
    if agent.profile_id and instructions.is_instruction_file(path, agent.project_dir):
        instructions.forget(agent.profile_id, agent.project_dir)


async def _current(agent: AgentContext, path: str) -> tuple[str, bool]:
    """What is in the file now, and whether it is safe to diff against.

    Best effort, and one extra SFTP round trip on every write -- see the note in
    `_run_write`. A file that cannot be read and a file that does not exist are
    the same thing over SFTP without a second trip for a stat, and both are
    shown as a new file, which is what git does and is honest enough here.

    Not diffable when the read came back at the ceiling: `read_file` truncates
    and says so in the text rather than in a flag, so a file at `max_output` is
    assumed truncated. Diffing a truncated original invents deletions of the
    tail, which is worse than showing no diff at all.
    """
    try:
        text = await agent.executor().read_file(path, max_bytes=agent.max_output)
    except ExecError:
        return "", True
    return text, len(text) < agent.max_output


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)
        )

    # What makes `file_edit` possible: a patch may only be applied to something
    # this reply has actually looked at.
    agent.read_paths.add(_path_key(agent, path))

    return ToolOutcome(
        text or "(the file is empty)",
        _event(
            "file_read",
            agent,
            path,
            status="ok",
            text=text[:MAX_EVENT_CHARS],
            canvas=_canvas(agent, path),
        ),
    )


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)

    # One extra SFTP round trip per write, on the hottest agent operation, and a
    # conscious trade. It buys the transcript a real diff instead of "1284
    # bytes" -- which is the difference between being able to see what an agent
    # did and having to go and look -- and it counts as having read the file, so
    # a write followed by an edit works in one reply.
    before, diffable = await _current(agent, path)

    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)
        )

    agent.read_paths.add(_path_key(agent, path))

    # The tree just changed, and this process is what changed it. The listing's
    # TTL is for drift nobody can see coming; leaving five more minutes of a
    # listing known to be wrong makes a model conclude the file it has just
    # written does not exist.
    if agent.profile_id:
        index.forget_dir(agent.profile_id, agent.project_dir)
        _forget_instructions(agent, path)

    event = _event(
        "file_write",
        agent,
        path,
        status="ok",
        text=f"{written} bytes",
        canvas=_canvas(agent, path),
    )
    if diffable and before != content:
        event["diff"] = patch.render(before, content, path, max_lines=MAX_DIFF_LINES)

    return ToolOutcome(f"Wrote {written} bytes to {path}.", event)


async def _run_edit(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
    """Change part of a file by applying a unified diff.

    The read-first requirement is the whole point. A patch written from memory
    against a file the model has not looked at either fails on context -- the
    good case -- or matches something it did not mean, and `file_write`'s
    failure mode is worse still: it silently drops everything the model did not
    happen to recall. Making the read compulsory turns "lost half the file" into
    "was told to read it first".
    """
    agent = _agent(context)
    path = str(args.get("path") or "").strip()
    if agent is None or not path:
        return _no_connection_or_path("file_edit", agent, path)

    if reason := _permitted(agent, "file_edit", RISK_WRITE):
        return _refused("file_edit", agent, path, reason)

    if _path_key(agent, path) not in agent.read_paths:
        return ToolOutcome(
            f"Read the file first! Nothing was written. Call file_read on {path} "
            f"in this reply, then send a patch that matches what came back.",
            _event("file_edit", agent, path, status="error", error="Not read yet."),
        )

    raw = args.get("patch")
    if not isinstance(raw, str) or not raw.strip():
        return ToolOutcome(
            "No patch was given. Send a unified diff: one or more "
            "`@@ -old,count +new,count @@` hunks.",
            _event("file_edit", agent, path, status="error", error="No patch."),
        )

    # Read here rather than through `_current`, which answers a different
    # question. `_current` exists for `file_write`, where a file that cannot be
    # read is a file about to be created and "" is the honest answer. Applying a
    # patch to that "" instead reported a context mismatch against
    # "(past the end of the file)" -- a model told the file is empty when it is
    # in fact unreadable retries the same patch, then rewrites the file whole,
    # which is how an unreadable file becomes a lost one.
    try:
        before = await agent.executor().read_file(path, max_bytes=agent.max_output)
    except ExecError as exc:
        return ToolOutcome(
            f"{exc.message} Nothing was written.",
            _event("file_edit", agent, path, status="error", error=exc.message),
        )

    # And a file too big to read whole may not be patched at all. `read_file`
    # truncates at the ceiling, so `after` would be the beginning of the file
    # with the patch applied -- and `write_file` replaces, so writing it back is
    # how the rest of the file is deleted. Silently, and reported as a success
    # with a byte count. This is the same rule Canvas follows for the same
    # reason: a truncated read opens read-only.
    if len(before) >= agent.max_output:
        return ToolOutcome(
            f"{path} is too large to patch: only the first {agent.max_output} bytes "
            f"can be read, and writing back what was read would delete the rest. "
            f"Nothing was written. Change it with a command instead — sed, or a "
            f"short script.",
            _event("file_edit", agent, path, status="error", error="Too large to patch."),
        )

    try:
        after = patch.apply(before, patch.parse(raw))
    except patch.PatchError as exc:
        # Returned, never raised: `run_tool`'s blanket catch would keep the
        # model going but lose the detail, and the detail is what it retries
        # from.
        return ToolOutcome(
            exc.message,
            _event("file_edit", agent, path, status="error", error=exc.message[:200]),
        )

    if after == before:
        return ToolOutcome(
            f"That patch changes nothing in {path}. It is already as you want it.",
            _event("file_edit", agent, path, status="ok", text="no change"),
        )

    try:
        written = await agent.executor().write_file(path, after)
    except ExecError as exc:
        return ToolOutcome(
            exc.message, _event("file_edit", agent, path, status="error", error=exc.message)
        )

    # Deliberately NOT index.forget_dir: an edit does not change the listing,
    # because the file was already there. Forgetting it would cost the next
    # reply either a wait on `INDEX_WAIT` or a turn with no listing at all, and
    # buy nothing. The instruction file is the opposite case -- the listing only
    # cares that it exists, that cache is a copy of what is in it.
    _forget_instructions(agent, path)

    event = _event(
        "file_edit",
        agent,
        path,
        status="ok",
        text=f"{written} bytes",
        canvas=_canvas(agent, path),
    )
    # Always diffable here: an unreadable original and a truncated one have both
    # already been refused above, which is the whole difference between this and
    # `file_write`'s use of `_current`.
    event["diff"] = patch.render(before, after, path, max_lines=MAX_DIFF_LINES)

    return ToolOutcome(f"Updated {path} ({written} bytes).", event)


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 `plan_final` and ends the reply
    there: a plan followed by three more rounds of the model changing its mind
    is not a plan.

    `steps` is still accepted alongside the structure. A small model sends it,
    `plans.normalise` turns it into one phase, and refusing would cost a whole
    round trip to say so.
    """
    agent = _agent(context)
    plan = plans.build(
        title=args.get("title"),
        summary=args.get("summary"),
        findings=args.get("findings"),
        objectives=args.get("objectives"),
        phases=args.get("phases"),
        steps=args.get("steps"),
    )

    if not plan or not plan["steps"]:
        return ToolOutcome(
            "A plan needs at least one task. Say what you would actually do, as "
            "phases of concrete tasks — or as a flat list of steps if there is "
            "only one phase of work.",
            {"name": "plan_submit", "kind": "plan", "status": "error",
             "error": "No tasks.", "results": []},
        )

    if agent is not None:
        agent.plan = plan

    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",
            "detail": agent.label if agent else "",
            "query": plan["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": plan,
            # Only `plan_submit` sets this, and it is what withdraws the tools
            # for the last round. `plan_update` is bookkeeping in the middle of
            # work and must not end the reply.
            "plan_final": True,
        },
    )


async def _run_plan_update(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
    """Tick something off, or record something found.

    `RISK_READ`, and the reasoning is worth stating because it sits in tension
    with `notes_edit` being `RISK_WRITE`. Risk is about what a tool does to *the
    world*, and the world the four modes govern is the machine -- this cannot
    touch it. Practically, `RISK_WRITE` would put an approval card on screen
    every time a task was ticked off: four cards to carry out a four-task plan,
    each one approving a bookkeeping entry, which is exactly the interruption
    that batching approvals exists to prevent. The distinguishing line against
    `notes_edit` is that a note is a durable artefact of the reader's that
    outlives the chat, while this is the chat's own record of what it is doing.
    An administrator who disagrees puts `plan_update` in the deny list.

    It reads and writes `agent.plan` rather than the database, because a runner
    cannot write the message row -- and because two updates in one reply would
    otherwise both read the same stale plan and the second would lose the first.
    """
    agent = _agent(context)
    if agent is None or not agent.plan:
        return ToolOutcome(
            "There is no plan for this conversation yet, so there is nothing to "
            "update.",
            {"name": "plan_update", "kind": "plan", "status": "error",
             "error": "No plan.", "results": []},
        )

    plan, changed = plans.merge(agent.plan, args)
    if not changed:
        return ToolOutcome(
            "Nothing in the plan changed. Quote a task or objective id from the "
            "plan above — they look like t1 and o1.",
            {"name": "plan_update", "kind": "plan", "status": "error",
             "error": "Nothing matched.", "results": []},
        )

    agent.plan = plan
    return ToolOutcome(
        "Plan updated: " + ", ".join(changed) + ". Carry on with the work.",
        {
            "name": "plan_update",
            "kind": "plan",
            "detail": agent.label,
            "query": plan["title"],
            "status": "ok",
            "results": [],
            "plan": plan,
            "text": "\n".join(changed),
        },
    )


# --- Background jobs -----------------------------------------------------------
# A job's files are namespaced by the *calling* chat's id (see jobs.py), and the
# read/stop wrappers are always built from `agent.chat_id`, so a model in one
# chat cannot name another chat's job -- the path simply would not exist. The id
# is still validated as our own hex first, so a crafted id cannot walk out of the
# job directory.
async def _run_job_output(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
    agent = _agent(context)
    if agent is None:
        return _no_machine("job_output")
    job_id = str(args.get("id") or "").strip()
    if not jobs.valid_id(job_id):
        return ToolOutcome(
            "There is no job with that id.",
            _event("job_output", agent, job_id, status="error", error="Unknown job."),
        )
    if reason := _permitted(agent, "job_output", RISK_READ):
        return _refused("job_output", agent, job_id, reason)

    try:
        reading = await jobs.read(agent, job_id)
    except ExecError as exc:
        return ToolOutcome(
            exc.message, _event("job_output", agent, job_id, status="error", error=exc.message)
        )

    if reading.status == "running":
        head = "Still running."
    elif reading.status == "done":
        head = "Finished, and printed nothing." if not reading.body else (
            "Finished." if reading.exit_status == 0 else f"Finished, exit {reading.exit_status}."
        )
    else:
        head = "No longer running — no exit status was recorded (it may have been killed)."

    content = f"{head}\n\n{reading.body}".strip() if reading.body else head
    return ToolOutcome(
        content,
        _event(
            "job_output",
            agent,
            job_id,
            status="ok" if reading.status != "lost" else "error",
            text=reading.body[:MAX_EVENT_CHARS],
        ),
    )


async def _run_job_list(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
    agent = _agent(context)
    if agent is None:
        return _no_machine("job_list")
    running = jobs.for_chat(agent.chat_id)
    if not running:
        return ToolOutcome(
            "No background jobs in this conversation.",
            _event("job_list", agent, "", status="ok", text="none"),
        )
    lines = [
        f"{job.id}  [{job.status}]  {job.command}"
        + (f" (exit {job.exit_status})" if job.exit_status is not None else "")
        for job in running
    ]
    body = "\n".join(lines)
    return ToolOutcome(
        f"Background jobs:\n{body}",
        _event("job_list", agent, f"{len(running)} job(s)", status="ok", text=body),
    )


async def _run_job_stop(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
    agent = _agent(context)
    if agent is None:
        return _no_machine("job_stop")
    job_id = str(args.get("id") or "").strip()
    if not jobs.valid_id(job_id):
        return ToolOutcome(
            "There is no job with that id.",
            _event("job_stop", agent, job_id, status="error", error="Unknown job."),
        )
    if reason := _permitted(agent, "job_stop", RISK_EXECUTE):
        return _refused("job_stop", agent, job_id, reason)

    try:
        await jobs.stop(agent, job_id)
    except ExecError as exc:
        return ToolOutcome(
            exc.message, _event("job_stop", agent, job_id, status="error", error=exc.message)
        )
    return ToolOutcome(
        f"Stopped job {job_id}.",
        _event("job_stop", agent, job_id, status="ok", text="stopped"),
    )


def _no_machine(name: str) -> ToolOutcome:
    return ToolOutcome(
        "This conversation is not connected to a machine.",
        {"name": name, "status": "error", "error": "No connection.", "results": []},
    )


# --- The definitions -----------------------------------------------------------
def _shell_parameters(background_on: bool) -> dict[str, Any]:
    properties: dict[str, Any] = {
        "command": {**_STRING, "description": "The command line to run."},
        "why": _WHY,
        "cwd": {**_STRING, "description": "Where to run it. Defaults to the project directory."},
        "timeout": {
            "type": "number",
            "description": "Seconds to allow. Bounded by the instance settings.",
        },
    }
    if background_on:
        properties["background"] = {
            "type": "boolean",
            "description": (
                "Run it detached instead of waiting. It keeps running after this "
                "reply; check on it with job_output. Use it for a long install, "
                "build or download. A command left waiting is also kept running "
                "as a job rather than killed when it hits its timeout."
            ),
        }
    return {"type": "object", "properties": properties, "required": ["command"]}


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=_shell_parameters(bool(context and context.background)),
            run=_explained(_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=(
                "Create a text file, or replace an existing one entirely. A "
                "relative path is taken from the project directory. Use this for a "
                "new file, or when you are rewriting the whole thing. To change "
                "part of a file that already exists, use file_edit instead: it is "
                "cheaper, and it cannot silently lose the parts you did not mean "
                "to touch."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "path": {**_STRING, "description": "The file to write."},
                    "content": {**_STRING, "description": "Its whole new contents."},
                    "why": _WHY,
                },
                "required": ["path", "content"],
            },
            run=_explained(_run_write),
            risk=RISK_WRITE,
        ),
        ToolDef(
            name="file_edit",
            family=FAMILY_AGENT,
            description=(
                "Change part of a text file by applying a unified diff. You must "
                "have read the file with file_read in this same reply first, or "
                "this is refused — a patch written from memory is how a change "
                "quietly becomes a rewrite.\n"
                "\n"
                "Send an ordinary patch: one or more `@@ -old,count +new,count @@` "
                "hunks, each with about three unchanged lines of context on either "
                "side of the change, ' ' for context, '-' to remove and '+' to add. "
                "The line numbers may be approximate — the context lines must be "
                "exact. Nothing is written unless every hunk applies, and you are "
                "told which one failed and what the file has there instead."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "path": {**_STRING, "description": "The file to change."},
                    "patch": {
                        **_STRING,
                        "description": "The unified diff to apply.",
                    },
                    "why": _WHY,
                },
                "required": ["path", "patch"],
            },
            run=_explained(_run_edit),
            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, 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.\n"
                "\n"
                "Say what you FOUND while looking, what the work is FOR, and then "
                "the work itself as PHASES of concrete tasks. A task should be "
                "one thing, specific enough to follow — name the files and the "
                "commands. If the work is short enough that phases would be "
                "ceremony, send `steps` instead and it becomes one phase.\n"
                "\n"
                "Findings are the part people skip and the part that makes a plan "
                "worth reading: what is actually there, what surprised you, what "
                "the plan is working around."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "title": {**_STRING, "description": "What the plan achieves, in a line."},
                    "summary": {
                        **_STRING,
                        "description": "One line on the approach. Optional.",
                    },
                    "findings": {
                        "type": "array",
                        "items": _STRING,
                        "description": (
                            "What you established while looking: what is there, "
                            "what constrains the work, what you ruled out."
                        ),
                    },
                    "objectives": {
                        "type": "array",
                        "items": _STRING,
                        "description": "What this is for. What has to be true at the end.",
                    },
                    "phases": {
                        "type": "array",
                        "description": "The work, in order.",
                        "items": {
                            "type": "object",
                            "properties": {
                                "title": {**_STRING, "description": "What this phase does."},
                                "tasks": {
                                    "type": "array",
                                    "items": _STRING,
                                    "description": "One thing each, in order.",
                                },
                            },
                            "required": ["title", "tasks"],
                        },
                    },
                    "steps": {
                        "type": "array",
                        "items": _STRING,
                        "description": (
                            "Instead of phases, when the work is one phase. "
                            "Becomes a single phase."
                        ),
                    },
                },
                "required": ["title"],
            },
            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,
        ),
        ToolDef(
            name="plan_update",
            family=FAMILY_AGENT,
            description=(
                "Keep the plan current while you carry it out. The plan is what "
                "somebody reads to see where you are, so it has to be updated as "
                "you go and not written up at the end.\n"
                "\n"
                "Mark a task 'doing' when you start it and 'done' when you have "
                "checked it actually works — not when you have written the code "
                "for it. Use 'dropped' for a task that turned out to be "
                "unnecessary, and say why in its note. Add tasks the plan did "
                "not anticipate as you discover them.\n"
                "\n"
                "One call carries as many changes as you like: finishing one "
                "task and starting the next is a single call, not two. Use the "
                "ids exactly as they appear in the plan above — tasks are t1, "
                "t2 and so on, objectives o1, phases p1.\n"
                "\n"
                "This does not end your turn and is not a progress report to "
                "stop after. Carry straight on with the work."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "task_status": {
                        "type": "array",
                        "description": "Tasks whose state has changed.",
                        "items": {
                            "type": "object",
                            "properties": {
                                "id": {**_STRING, "description": "The task id, e.g. t3."},
                                "status": {
                                    **_STRING,
                                    "description": "todo, doing, done or dropped.",
                                },
                                "note": {
                                    **_STRING,
                                    "description": "A short note about it. Optional.",
                                },
                            },
                            "required": ["id", "status"],
                        },
                    },
                    "objective_status": {
                        "type": "array",
                        "description": "Objectives whose state has changed.",
                        "items": {
                            "type": "object",
                            "properties": {
                                "id": {**_STRING, "description": "The objective id, e.g. o1."},
                                "status": {
                                    **_STRING,
                                    "description": "open, done or dropped.",
                                },
                            },
                            "required": ["id", "status"],
                        },
                    },
                    "findings": {
                        "type": "array",
                        "items": _STRING,
                        "description": "Anything new you have established.",
                    },
                    "add_tasks": {
                        "type": "array",
                        "description": "Work the plan did not anticipate.",
                        "items": {
                            "type": "object",
                            "properties": {
                                "text": {**_STRING, "description": "The task."},
                                "phase": {
                                    **_STRING,
                                    "description": (
                                        "Which phase it belongs to, e.g. p2. "
                                        "Defaults to the one in progress."
                                    ),
                                },
                            },
                            "required": ["text"],
                        },
                    },
                    "summary": {**_STRING, "description": "Where things stand, in a line."},
                },
                "required": [],
            },
            run=_run_plan_update,
            # See `_run_plan_update`: it cannot touch the machine, and asking
            # about it would mean an approval card per ticked-off task.
            risk=RISK_READ,
        ),
        ToolDef(
            name="job_output",
            family=FAMILY_AGENT,
            description=(
                "Read what a background job has printed so far, and whether it is "
                "still running. Give the id shell_run returned when it was "
                "backgrounded."
            ),
            parameters={
                "type": "object",
                "properties": {"id": {**_STRING, "description": "The job id."}},
                "required": ["id"],
            },
            run=_run_job_output,
            risk=RISK_READ,
        ),
        ToolDef(
            name="job_list",
            family=FAMILY_AGENT,
            description="List the background jobs in this conversation and their state.",
            parameters={"type": "object", "properties": {}, "required": []},
            run=_run_job_list,
            risk=RISK_READ,
        ),
        ToolDef(
            name="job_stop",
            family=FAMILY_AGENT,
            description="Stop a background job, killing it and everything it started.",
            parameters={
                "type": "object",
                "properties": {
                    "id": {**_STRING, "description": "The job id."},
                    "why": _WHY,
                },
                "required": ["id"],
            },
            run=_explained(_run_job_stop),
            # It terminates a process on the machine, so it goes through the mode
            # table exactly as shell_run does.
            risk=RISK_EXECUTE,
        ),
    ]
    if context is None:
        return defs

    # `plan_submit` in Plan mode and nowhere else; `plan_update` everywhere
    # else, and only once there is a plan to update. Offering it with no plan
    # would be the skills asymmetry again -- a tool for changing something that
    # does not exist, which costs a round to find out.
    drop = {"plan_submit"} if context.mode != policy.MODE_PLAN else {"plan_update"}
    if not context.plan:
        drop.add("plan_update")
    # The job tools exist only when commands may run in the background. Offering
    # them otherwise is a tool for checking on something that can never exist.
    if not context.background:
        drop.update({"job_output", "job_list", "job_stop"})
    return [tool for tool in defs if tool.name not in drop]


__all__ = [
    "FAMILY_AGENT",
    "MAX_DIFF_LINES",
    "MAX_EVENT_CHARS",
    "MAX_WHY_CHARS",
    "tool_defs",
    "why_of",
]
