Let a command run in the background instead of being killed

An agent command is one blocking conn.run over a per-call connection, killed the
moment it hits its timeout -- so a ten-minute apt install is impossible, which is
exactly what a user hit. This is the substrate for running it detached instead:
the model can ask for background=true, or a command that outlasts its timeout is
kept running rather than killed, and either way the model gets tools to read and
stop it. Opt-in, off by default, under Admin -> Agents; off is byte-for-byte the
old behaviour.

The mechanism has to survive the connection closing (that is the whole premise
of the per-call model), so a job is a setsid-detached process on the far side,
redirected to a remote logfile and an exit-file; LLeMbas reconnects, as always,
to read it later. services/agent/jobs.py holds the wrappers.

Three things in those wrappers are load-bearing and each was got wrong in the
first sketch:

- The command never touches a quoted shell context. sh -c '<cmd>' shatters the
  instant the command contains a quote -- git commit -m 'fix', awk '{…}', sed
  's/…/…/' are the common case, and it is an injection hole besides. So the
  command is base64-encoded in Python and decoded on the far side into a script
  file; it is bytes, never shell syntax.
- The child records its own pid via $$ as its first act, under setsid where it
  is the session leader, so job_stop can kill the whole process group. echo $!
  from the launcher captures the wrong pid.
- The command's exit status comes from the exit-file, never the wrapper's own
  status -- which is ~0 from its trailing rm. Reading the wrapper's status would
  mark every job a success.

A command that finishes in time is indistinguishable from a foreground one --
same output, same wording; the difference shows only when it does not, where
instead of "stopped after Ns" it becomes a job id. Auto-convert is its own
sub-switch: with it off, a timeout stays a hard stop and nothing is left
running, because routing the plain case through the detached wrapper would leave
an orphan running past a stop an administrator asked for.

New agent tools job_output/job_list/job_stop, offered only when the feature is
on (the plan_submit gating pattern); job_stop is RISK_EXECUTE since it kills a
process. A job's files are namespaced by the calling chat's id and the wrappers
are always built from it, so a model in one chat cannot even name another's job.

Tested against a real local /bin/sh rather than the fake echo-the-command sshd
fixture, because the shell logic -- setsid, base64, the wait loop, the child
surviving the wait being cut off -- is the whole of the risk. The auto-wake that
prompts the model back when a job finishes is the next commit.

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

src/lembas/services/agent/jobs.py

@@ -0,0 +1,361 @@
+"""Commands that outlive the reply that started them.
+
+An ordinary `shell_run` is one blocking `conn.run` over a per-call connection
+(`ssh.py`): when it hits its timeout the command is killed, so a ten-minute
+`apt install` is impossible. A background job is the same command launched
+*detached* on the far side -- `setsid`, redirected to a remote logfile and an
+exit-file -- so it survives the connection closing. LLeMbas reconnects (a fresh
+connection, as always) to read the log and the exit code later.
+
+This is the opposite of `terminal.py`, which survives by *holding* a connection
+open. Here we hold nothing: the whole point of `ssh.py`/`base.py` is that no live
+connection is kept, and a job that needed one would be a job that broke that.
+
+**The command never touches a quoted shell context.** `sh -c '<cmd>'` shatters
+the instant the command contains a `'` -- `git commit -m 'fix'`, `awk '{…}'`,
+`sed 's/…/…/'` are the common case, not an edge one, and would also be an
+injection hole. So the command is base64-encoded here in Python and decoded on
+the far side into a script file; it is bytes, never shell syntax. Only
+server-generated hex ids and a fixed root ever reach a path.
+
+Three things make the wrappers correct, and each was got wrong in an earlier
+sketch:
+
+* **The child records its own pid via `$$`**, as its first act, under `setsid`
+  where it is the session/group leader -- so `job_stop` can `kill -<pid>` the
+  whole process group. `echo $!` from the launcher captures the wrong pid.
+* **The exit-file is the primary signal.** An empty pid-file means "still
+  starting", not "dead"; reading liveness first would race the launch and report
+  a job lost the instant it began.
+* **The command's exit status comes from the exit-file, never from the wrapper's
+  own status** -- which is ~0 from the trailing `rm`. Reading the wrapper's
+  status would mark every job a success.
+"""
+
+from __future__ import annotations
+
+import base64
+import contextlib
+import re
+import time
+import uuid
+from dataclasses import dataclass, field
+
+from lembas.services.agent.base import ExecError, ExecRequest, clean_output
+
+# Where a job's files live on the far side. `${TMPDIR:-/tmp}` so a host that
+# puts scratch space elsewhere is honoured, and it clears on reboot -- a job
+# does not survive a reboot of its own host either. The chat id namespaces it,
+# which is also what makes cross-chat access structurally impossible: a path is
+# only ever built from the *calling* chat's id, so a model in one chat cannot
+# name another chat's files.
+JOB_ROOT = "${TMPDIR:-/tmp}/lembas-jobs"
+
+# A job id is our own short hex; anything else is refused before it reaches a
+# path, so `job_output("../../etc/passwd")` cannot walk out of the job root.
+_ID = re.compile(r"^[a-f0-9]{12}$")
+
+# How long the fire-and-return launcher waits for the shell to accept the
+# command. Not the command's own timeout -- it returns the moment the process is
+# detached, which is immediate.
+LAUNCH_GRACE = 10.0
+
+# In-process, keyed by job id, lost on restart -- the durable record is the `Job`
+# row (added with the watcher). This holds the metadata `job_list` shows within
+# a session and, later, the watcher task.
+_JOBS: dict[str, JobState] = {}
+
+
+def new_id() -> str:
+    return uuid.uuid4().hex[:12]
+
+
+@dataclass
+class JobState:
+    """What LLeMbas remembers about one background job, in this process."""
+
+    id: str
+    chat_id: str
+    command: str
+    status: str = "running"  # running | done | killed | lost
+    exit_status: int | None = None
+    started_at: float = field(default_factory=time.monotonic)
+    finished_at: float = 0.0
+
+
+# --- Paths and the wrappers ----------------------------------------------------
+def _dir(chat_id: str) -> str:
+    return f'"{JOB_ROOT}/{chat_id}"'
+
+
+def _file(chat_id: str, job_id: str, ext: str) -> str:
+    # Double-quoted so `${TMPDIR:-/tmp}` still expands while the whole path stays
+    # one word. The chat id and job id are hex, so nothing here needs escaping.
+    return f'"{JOB_ROOT}/{chat_id}/{job_id}.{ext}"'
+
+
+def _sentinel(job_id: str) -> str:
+    return f"__LEMBAS_{job_id}__"
+
+
+def _inner_script(chat_id: str, job_id: str, command: str) -> str:
+    """The detached program: record the pid, run the command, record the status.
+
+    base64-encoded before it leaves, so `command` is bytes and never shell
+    syntax. `$$` first, because it is the session leader's pid under setsid and
+    `job_stop` kills the group by it. `$?` last, capturing the command's status;
+    it is the file `run`'s own exit status must never be read in place of.
+    """
+    return (
+        f"echo $$ > {_file(chat_id, job_id, 'pid')}\n"
+        f"{command}\n"
+        f"echo $? > {_file(chat_id, job_id, 'exit')}\n"
+    )
+
+
+def _blob(chat_id: str, job_id: str, command: str) -> str:
+    raw = _inner_script(chat_id, job_id, command).encode("utf-8")
+    return base64.b64encode(raw).decode("ascii")
+
+
+def _launch_lines(chat_id: str, job_id: str, command: str) -> str:
+    """Create the job dir, drop the script, and detach it. No wait."""
+    blob = _blob(chat_id, job_id, command)
+    return (
+        f"mkdir -p {_dir(chat_id)} 2>/dev/null\n"
+        f"printf %s '{blob}' | base64 -d > {_file(chat_id, job_id, 'sh')}\n"
+        f"setsid sh {_file(chat_id, job_id, 'sh')} "
+        f"> {_file(chat_id, job_id, 'log')} 2>&1 < /dev/null &\n"
+    )
+
+
+def launch_command(chat_id: str, job_id: str, command: str) -> str:
+    """Fire-and-return: detach the command and stop. Run with a short timeout."""
+    return _launch_lines(chat_id, job_id, command) + "printf started\n"
+
+
+def launch_and_wait_command(chat_id: str, job_id: str, command: str, max_bytes: int) -> str:
+    """Detach the command AND wait up to the (asyncssh) timeout for it.
+
+    If it finishes, stdout is the log tail plus a sentinel line carrying the exit
+    code, and the files are removed. If asyncssh times out first the channel is
+    torn down before the `rm`, so the files survive for a later read and the
+    detached process -- new session, redirected, stdin from /dev/null -- keeps
+    running. That torn-down-mid-wait case is exactly "it became a background
+    job".
+    """
+    s = _sentinel(job_id)
+    pid = _file(chat_id, job_id, "pid")
+    exit_ = _file(chat_id, job_id, "exit")
+    log = _file(chat_id, job_id, "log")
+    return (
+        _launch_lines(chat_id, job_id, command)
+        + "while :; do\n"
+        f"  [ -f {exit_} ] && break\n"
+        f"  __p=$(cat {pid} 2>/dev/null)\n"
+        '  [ -n "$__p" ] && ! kill -0 "$__p" 2>/dev/null && break\n'
+        # 0.2s: with the feature on, every ordinary command waits one poll for
+        # the exit-file, so this is added latency on the hot path. Short enough
+        # not to be felt, long enough not to spin.
+        "  sleep 0.2\n"
+        "done\n"
+        f"tail -c {max_bytes} {log} 2>/dev/null\n"
+        f"printf '\\n{s}:'\n"
+        f"cat {exit_} 2>/dev/null || printf LOST\n"
+        f"rm -f {_file(chat_id, job_id, 'sh')} {pid} {log} {exit_}\n"
+    )
+
+
+def read_command(chat_id: str, job_id: str, max_bytes: int) -> str:
+    """The log so far, and whether the job is still running."""
+    s = _sentinel(job_id)
+    pid = _file(chat_id, job_id, "pid")
+    exit_ = _file(chat_id, job_id, "exit")
+    return (
+        f"tail -c {max_bytes} {_file(chat_id, job_id, 'log')} 2>/dev/null\n"
+        f"printf '\\n{s}:'\n"
+        f"if [ -f {exit_} ]; then printf 'done '; cat {exit_};\n"
+        f'elif __p=$(cat {pid} 2>/dev/null); [ -n "$__p" ] && kill -0 "$__p" 2>/dev/null;'
+        " then printf running;\n"
+        "else printf lost; fi\n"
+    )
+
+
+def stop_command(chat_id: str, job_id: str) -> str:
+    """Kill the whole process group, then record an exit so a reader is not told
+    the job is merely lost. A killed process never writes its own exit file."""
+    pid = _file(chat_id, job_id, "pid")
+    exit_ = _file(chat_id, job_id, "exit")
+    return (
+        f'__p=$(cat {pid} 2>/dev/null); [ -n "$__p" ] && kill -TERM -"$__p" 2>/dev/null\n'
+        "sleep 0.3\n"
+        f'[ -n "$__p" ] && kill -KILL -"$__p" 2>/dev/null\n'
+        f"[ -f {exit_} ] || echo 143 > {exit_}\n"
+        "printf stopped\n"
+    )
+
+
+def cleanup_command(chat_id: str, job_id: str) -> str:
+    return (
+        f"rm -f {_file(chat_id, job_id, 'sh')} {_file(chat_id, job_id, 'pid')} "
+        f"{_file(chat_id, job_id, 'log')} {_file(chat_id, job_id, 'exit')}\n"
+    )
+
+
+# --- Parsing what a wrapper printed --------------------------------------------
+@dataclass(frozen=True)
+class Completed:
+    body: str
+    exit_status: int | None  # None ⇒ the job was lost (killed without an exit)
+
+
+def parse_completed(output: str, job_id: str) -> Completed:
+    """Split a launch-and-wait result into the command's output and its status.
+
+    On the *last* sentinel, because the command's own output could contain a
+    line that looks like one; everything before it is the body, everything after
+    is the exit code the file held.
+    """
+    marker = f"\n{_sentinel(job_id)}:"
+    at = output.rfind(marker)
+    if at == -1:
+        return Completed(body=output.strip(), exit_status=None)
+    body = output[:at].strip()
+    tail = output[at + len(marker) :].strip()
+    if tail.upper() == "LOST" or not tail:
+        return Completed(body=body, exit_status=None)
+    try:
+        return Completed(body=body, exit_status=int(tail.split()[0]))
+    except (ValueError, IndexError):
+        return Completed(body=body, exit_status=None)
+
+
+@dataclass(frozen=True)
+class Reading:
+    body: str
+    status: str  # running | done | lost
+    exit_status: int | None
+
+
+def parse_reading(output: str, job_id: str) -> Reading:
+    marker = f"\n{_sentinel(job_id)}:"
+    at = output.rfind(marker)
+    if at == -1:
+        return Reading(body=output.strip(), status="lost", exit_status=None)
+    body = output[:at].strip()
+    tail = output[at + len(marker) :].strip()
+    if tail.startswith("done"):
+        parts = tail.split()
+        code = int(parts[1]) if len(parts) > 1 and parts[1].lstrip("-").isdigit() else None
+        return Reading(body=body, status="done", exit_status=code)
+    if tail == "running":
+        return Reading(body=body, status="running", exit_status=None)
+    return Reading(body=body, status="lost", exit_status=None)
+
+
+# --- Operations against the machine --------------------------------------------
+async def launch(agent, command: str, cwd: str = "") -> JobState:
+    """Detach a command and return immediately. Raises ExecError if it will not
+    even start."""
+    job_id = new_id()
+    result = await agent.executor().run(
+        ExecRequest(
+            command=launch_command(agent.chat_id, job_id, command),
+            cwd=cwd,
+            timeout=LAUNCH_GRACE,
+            max_bytes=agent.max_output,
+        )
+    )
+    if result.timed_out:
+        raise ExecError("The machine did not accept the command in time.")
+    job = JobState(id=job_id, chat_id=agent.chat_id, command=command)
+    _JOBS[job_id] = job
+    return job
+
+
+async def read(agent, job_id: str) -> Reading:
+    output, _ = _clean(
+        await agent.executor().run(
+            ExecRequest(
+                command=read_command(agent.chat_id, job_id, agent.max_output),
+                timeout=agent.timeout,
+                max_bytes=agent.max_output,
+            )
+        ),
+        agent.max_output,
+    )
+    reading = parse_reading(output, job_id)
+    _record(job_id, reading.status, reading.exit_status)
+    if reading.status in ("done", "lost"):
+        await _cleanup(agent, job_id)
+    return reading
+
+
+async def stop(agent, job_id: str) -> None:
+    await agent.executor().run(
+        ExecRequest(command=stop_command(agent.chat_id, job_id), timeout=agent.timeout)
+    )
+    _record(job_id, "killed", 143)
+
+
+async def _cleanup(agent, job_id: str) -> None:
+    with contextlib.suppress(ExecError):
+        await agent.executor().run(
+            ExecRequest(command=cleanup_command(agent.chat_id, job_id), timeout=agent.timeout)
+        )
+
+
+def _clean(result, limit: int) -> tuple[str, bool]:
+    if result.timed_out:
+        return result.output, False
+    return clean_output(result.output or "", limit=limit)
+
+
+# --- The registry --------------------------------------------------------------
+def register(job: JobState) -> None:
+    _JOBS[job.id] = job
+
+
+def get(job_id: str) -> JobState | None:
+    return _JOBS.get(job_id)
+
+
+def for_chat(chat_id: str) -> list[JobState]:
+    return [j for j in _JOBS.values() if j.chat_id == chat_id]
+
+
+def valid_id(job_id: str) -> bool:
+    return bool(_ID.match(job_id or ""))
+
+
+def _record(job_id: str, status: str, exit_status: int | None) -> None:
+    job = _JOBS.get(job_id)
+    if job is None:
+        return
+    if status in ("done", "lost", "killed") and job.status == "running":
+        job.status = status
+        job.exit_status = exit_status
+        job.finished_at = time.monotonic()
+
+
+def clear() -> None:
+    _JOBS.clear()
+
+
+__all__ = [
+    "JOB_ROOT",
+    "Completed",
+    "JobState",
+    "Reading",
+    "clear",
+    "for_chat",
+    "get",
+    "launch",
+    "launch_and_wait_command",
+    "new_id",
+    "parse_completed",
+    "read",
+    "register",
+    "stop",
+    "valid_id",
+]

src/lembas/services/agent/session.py

@@ -86,6 +86,16 @@ class AgentContext:
     # read the same stale plan from the database and the second would lose the
     # first. This snapshot is what they actually merge into.
     plan: dict[str, Any] = field(default_factory=dict)
+    # Whether commands may run detached. When off, `shell_run` is byte-for-byte
+    # what it always was and the `job_*` tools are not offered -- a command that
+    # times out is killed, as before. When on, a command can be launched in the
+    # background (or converted to one when it times out) and the model gets the
+    # tools to check on it. `on_timeout` is the sub-switch for the auto-convert.
+    background: bool = False
+    background_on_timeout: bool = True
+    # Whether a finished job wakes the model on its own, rather than only being
+    # seen when it next runs. Read by the wording here and by the watcher.
+    background_notify: bool = True
 
     def executor(self) -> Executor:
         return ssh_service.SshExecutor(self.spec, self.project_dir)
@@ -179,6 +189,9 @@ def resolve(db: DBSession, chat: Chat, user: User | None) -> AgentContext | None
         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),
+        background=bool(values.get("background_enabled")),
+        background_on_timeout=bool(values.get("background_on_timeout", True)),
+        background_notify=bool(values.get("background_notify", True)),
         spec=ssh_service.spec_from(profile),
     )
 

src/lembas/services/agent/tools.py

@@ -23,8 +23,8 @@ import posixpath
 from typing import Any
 
 from lembas.services import plans
-from lembas.services.agent import index, instructions, patch, policy
-from lembas.services.agent.base import ExecError, ExecRequest
+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,
@@ -126,29 +126,135 @@ async def _run_shell(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
     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=str(args.get("cwd") or "").strip(),
-                timeout=timeout,
-                max_bytes=agent.max_output,
-            )
+            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)
         )
 
-    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}."
+        job = jobs.JobState(id=job_id, chat_id=agent.chat_id, command=command)
+        jobs.register(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)
+        )
+    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.",
@@ -156,8 +262,8 @@ async def _run_shell(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
             "shell_run",
             agent,
             command,
-            status="ok" if result.ok else "error",
-            error="" if result.ok else head,
+            status="ok" if ok else "error",
+            error="" if ok else head,
             text=body[:MAX_EVENT_CHARS],
         ),
     )
@@ -501,7 +607,131 @@ async def _run_plan_update(context: ToolContext, args: dict[str, Any]) -> ToolOu
     )
 
 
+# --- 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."},
+        "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.
 
@@ -525,21 +755,7 @@ def tool_defs(context: AgentContext | None = None) -> list[ToolDef]:
                 "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"],
-            },
+            parameters=_shell_parameters(bool(context and context.background)),
             run=_run_shell,
             risk=RISK_EXECUTE,
         ),
@@ -779,6 +995,44 @@ def tool_defs(context: AgentContext | None = None) -> list[ToolDef]:
             # 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."}},
+                "required": ["id"],
+            },
+            run=_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
@@ -790,6 +1044,10 @@ def tool_defs(context: AgentContext | None = None) -> list[ToolDef]:
     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]