From 89d2d6ebfd723f79f21fc8c9af2bb5f90ecd5272 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jaroslav=20Bene=C5=A1?= Date: Mon, 3 Aug 2026 14:15:49 +0200 Subject: [PATCH] Let a command run in the background instead of being killed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 '' 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) --- src/lembas/api/admin_agents.py | 8 + src/lembas/services/agent/jobs.py | 361 +++++++++++++++++++++ src/lembas/services/agent/session.py | 13 + src/lembas/services/agent/tools.py | 320 ++++++++++++++++-- src/lembas/services/settings_store.py | 16 + src/lembas/services/tool_labels.py | 8 + src/lembas/web/templates/admin/agents.html | 55 ++++ tests/test_agent_jobs.py | 283 ++++++++++++++++ 8 files changed, 1033 insertions(+), 31 deletions(-) create mode 100644 src/lembas/services/agent/jobs.py create mode 100644 tests/test_agent_jobs.py diff --git a/src/lembas/api/admin_agents.py b/src/lembas/api/admin_agents.py index a434912..ae9b5be 100644 --- a/src/lembas/api/admin_agents.py +++ b/src/lembas/api/admin_agents.py @@ -79,6 +79,10 @@ async def save_agents( instructions_enabled: bool = Form(False), instructions_chars: int = Form(4000), nudge_unfinished: bool = Form(False), + background_enabled: bool = Form(False), + background_on_timeout: bool = Form(False), + background_notify: bool = Form(False), + background_max_jobs: int = Form(5), ) -> Response: settings_store.update( db, @@ -112,6 +116,10 @@ async def save_agents( "instructions_enabled": instructions_enabled, "instructions_chars": min(max(instructions_chars, 0), 20_000), "nudge_unfinished": nudge_unfinished, + "background_enabled": background_enabled, + "background_on_timeout": background_on_timeout, + "background_notify": background_notify, + "background_max_jobs": min(max(background_max_jobs, 1), 100), }, key=settings_store.AGENTS, ) diff --git a/src/lembas/services/agent/jobs.py b/src/lembas/services/agent/jobs.py new file mode 100644 index 0000000..0919f44 --- /dev/null +++ b/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 ''` 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 -` 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", +] diff --git a/src/lembas/services/agent/session.py b/src/lembas/services/agent/session.py index b99754b..d6cb9e4 100644 --- a/src/lembas/services/agent/session.py +++ b/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), ) diff --git a/src/lembas/services/agent/tools.py b/src/lembas/services/agent/tools.py index 8ab0612..e1799ac 100644 --- a/src/lembas/services/agent/tools.py +++ b/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] diff --git a/src/lembas/services/settings_store.py b/src/lembas/services/settings_store.py index faf0733..643888c 100644 --- a/src/lembas/services/settings_store.py +++ b/src/lembas/services/settings_store.py @@ -137,6 +137,21 @@ def _agents_defaults() -> dict[str, Any]: # one thing there is to be objectively wrong about -- a model with no # plan that says it has finished is believed. "nudge_unfinished": True, + # Whether a command may run detached, keep running after the reply ends, + # and be checked on later. Off by default, and off means byte-for-byte + # the old behaviour: a command that times out is killed. See + # services/agent/jobs.py. + "background_enabled": False, + # The sub-switch: a timed-out command is left running as a job instead + # of killed. Off leaves the timeout a hard stop and offers only the + # model's explicit `background=true`. + "background_on_timeout": True, + # Whether the model is woken with the result when a job finishes, rather + # than only seeing it when it next runs of its own accord. + "background_notify": True, + # Most background jobs watched at once. Each is a periodic reconnect to + # the far side, so it is a real cost, not a scruple. + "background_max_jobs": 5, } @@ -323,4 +338,5 @@ def agents(db: DBSession) -> dict[str, Any]: values["max_completion_tokens"] = min( max(int(values.get("max_completion_tokens") or 0), 0), 5_000_000 ) + values["background_max_jobs"] = min(max(int(values.get("background_max_jobs") or 0), 1), 100) return values diff --git a/src/lembas/services/tool_labels.py b/src/lembas/services/tool_labels.py index 6f088ca..eb5d4e1 100644 --- a/src/lembas/services/tool_labels.py +++ b/src/lembas/services/tool_labels.py @@ -47,6 +47,9 @@ LABELS: dict[str, str] = { "file_list": "List", "plan_submit": "Plan", "plan_update": "Plan updated", + "job_output": "Job output", + "job_list": "Jobs", + "job_stop": "Job stopped", # The web. "web_search": "Web search", "fetch": "Fetch", @@ -77,6 +80,9 @@ ICONS: dict[str, str] = { "file_list": "folder", "plan_submit": "check", "plan_update": "check", + "job_output": "clock", + "job_list": "dots", + "job_stop": "stop-circle", "web_search": "globe", "fetch": "link", "knowledge_search": "archive", @@ -126,6 +132,7 @@ ACTIONS: dict[str, str] = { "skill_get": "Read a skill", "skill_create": "Write a skill", "skill_edit": "Change a skill", + "job_stop": "Stop a background job", } # Which argument is the thing being agreed to. Shown verbatim and escaped on the @@ -141,6 +148,7 @@ DETAIL_KEYS: dict[str, str] = { "web_search": "query", "knowledge_search": "query", "notes_search": "query", + "job_stop": "id", } diff --git a/src/lembas/web/templates/admin/agents.html b/src/lembas/web/templates/admin/agents.html index d2f967b..2079401 100644 --- a/src/lembas/web/templates/admin/agents.html +++ b/src/lembas/web/templates/admin/agents.html @@ -124,6 +124,61 @@ +
+

Background commands

+

+ A command that would outlast its timeout can be left running instead of + killed — detached on the far side, checked on later. It is how a long + install, build or download becomes possible at all. +

+
+ +

+ Off means byte-for-byte the old behaviour: a command that hits its + timeout is killed. On, a command can be launched detached (or kept + running when it times out), writing to a file under + /tmp on the machine, and the model gets tools to read and + stop it. A detached command's log can grow without bound on the host — + that is the host's to contain, as with everything an agent runs there. +

+
+
+ +

+ Off leaves the timeout a hard stop; the model can still choose to + background a command up front. +

+
+
+ +

+ On, a finished job starts (or joins) a reply carrying its result. Off, + the model only sees it the next time it runs of its own accord. +

+
+
+ + +

+ Each is a periodic reconnect to the machine. Jobs past this still run; + they are simply not watched, and the model is not woken for them. +

+
+
+

What one reply may spend

diff --git a/tests/test_agent_jobs.py b/tests/test_agent_jobs.py new file mode 100644 index 0000000..5695c8f --- /dev/null +++ b/tests/test_agent_jobs.py @@ -0,0 +1,283 @@ +"""Background jobs, exercised against a real local shell. + +The sshd fixture elsewhere is a *fake* shell (it echoes the command), which +cannot run `setsid`, `base64`, a wait loop, or an exit-file. So these run the +wrappers through the machine's own `/bin/sh` with the same contract +`SshExecutor.run` has — cd-prefix, interleaved output, a timeout that leaves the +detached child alive — because the shell logic is the whole of the risk here. +""" + +from __future__ import annotations + +import asyncio +import shutil +import time + +import pytest + +from lembas.services import tools as tools_service +from lembas.services.agent import jobs, policy +from lembas.services.agent import ssh as ssh_service +from lembas.services.agent.base import ExecRequest, ExecResult, clean_output +from lembas.services.agent.session import AgentContext +from lembas.services.agent.tools import _run_shell + +pytestmark = pytest.mark.skipif( + shutil.which("setsid") is None or shutil.which("base64") is None, + reason="needs setsid and base64 (Linux)", +) + + +class LocalExecutor: + """`SshExecutor.run`'s contract, run against the local shell. + + On timeout it kills only the outer shell -- exactly what an SSH channel + teardown does -- so a `setsid`-detached child survives, which is the whole + behaviour a background job depends on. + """ + + def __init__(self, project_dir: str) -> None: + self.project_dir = project_dir + + async def run(self, request: ExecRequest) -> ExecResult: + directory = request.cwd or self.project_dir + command = request.command + if directory: + command = f"cd {ssh_service._quote(directory)} && {command}" + started = time.monotonic() + proc = await asyncio.create_subprocess_shell( + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + stdin=asyncio.subprocess.DEVNULL, + ) + try: + out, _ = await asyncio.wait_for(proc.communicate(), timeout=request.timeout) + except TimeoutError: + proc.kill() + await proc.wait() + return ExecResult( + exit_status=-1, + output=f"The command was still running after {request.timeout:g}s.", + timed_out=True, + duration_ms=int((time.monotonic() - started) * 1000), + ) + output, truncated = clean_output(out.decode("utf-8", "replace"), limit=request.max_bytes) + return ExecResult( + exit_status=proc.returncode if proc.returncode is not None else -1, + output=output, + truncated=truncated, + duration_ms=int((time.monotonic() - started) * 1000), + ) + + +@pytest.fixture(autouse=True) +def _clear_jobs(): + jobs.clear() + yield + jobs.clear() + + +def _agent(tmp_path, **over) -> AgentContext: + fields = { + "chat_id": "chat0001", + "label": "Box", + "project_dir": str(tmp_path), + "mode": policy.MODE_AUTO, + "max_output": 65536, + "timeout": 60.0, + "max_timeout": 600.0, + "background": True, + "background_on_timeout": True, + "background_notify": True, + } + fields.update(over) + ctx = AgentContext(**fields) + ctx.executor = lambda: LocalExecutor(fields["project_dir"]) # type: ignore[method-assign] + return ctx + + +def _context(agent): + return tools_service.ToolContext(owner_id="u", agent=agent) + + +# --- The wrappers, as strings -------------------------------------------------- +def test_the_command_is_never_in_a_quoted_context(): + """git commit -m 'fix' would shatter sh -c ''. It is base64'd instead.""" + import base64 + + cmd = "git commit -m 'fix: it'" + wrapper = jobs.launch_and_wait_command("chatx", "abc123abc123", cmd, 4096) + assert cmd not in wrapper, "the command leaked into the wrapper as shell text" + blob = wrapper.split("printf %s '", 1)[1].split("'", 1)[0] + inner = base64.b64decode(blob).decode() + assert cmd in inner, "the command was lost in the base64 round-trip" + + +def test_parse_reads_the_last_sentinel(): + out = ("line one\n__LEMBAS_jobjobjob01__:0\nmore\n__LEMBAS_jobjobjob01__:0") + done = jobs.parse_completed(out, "jobjobjob01") + assert done.exit_status == 0 + + +def test_valid_id_refuses_a_path(): + assert jobs.valid_id("deadbeef0000") + assert not jobs.valid_id("../../etc/passwd") + assert not jobs.valid_id("") + + +# --- Real shell: launch-and-wait ----------------------------------------------- +async def test_a_fast_command_completes_like_a_foreground_one(tmp_path): + agent = _agent(tmp_path) + outcome = await _run_shell(_context(agent), {"command": "echo hello"}) + + assert "hello" in outcome.content + assert outcome.event["status"] == "ok" + assert not jobs.for_chat(agent.chat_id), "a finished command is not a job" + + +async def test_a_nonzero_exit_is_read_from_the_exit_file_not_the_wrapper(tmp_path): + """The wrapper's own status is ~0 from its trailing rm; the command's real + status is in the exit-file.""" + agent = _agent(tmp_path) + outcome = await _run_shell(_context(agent), {"command": "sh -c 'exit 3'"}) + + assert "exited 3" in outcome.content + assert outcome.event["status"] == "error" + + +async def test_a_command_with_single_quotes_runs(tmp_path): + agent = _agent(tmp_path) + outcome = await _run_shell(_context(agent), {"command": "echo 'a'\\''b'"}) + + assert "a'b" in outcome.content + + +async def test_output_is_captured(tmp_path): + agent = _agent(tmp_path) + outcome = await _run_shell(_context(agent), {"command": "printf 'one\\ntwo\\n'"}) + assert "one" in outcome.content and "two" in outcome.content + + +# --- Real shell: auto-background on timeout ------------------------------------- +async def test_a_slow_command_is_kept_running_when_it_times_out(tmp_path): + """The apt-install case. It is not killed; it becomes a job that finishes + on its own, and its result is readable afterwards.""" + agent = _agent(tmp_path) + outcome = await _run_shell( + _context(agent), {"command": "sleep 2; echo done-late", "timeout": 1} + ) + + assert "background" in outcome.content.lower() + running = jobs.for_chat(agent.chat_id) + assert len(running) == 1 + job_id = running[0].id + + # The detached child survived the wait being cut off; give it time to finish. + for _ in range(40): + reading = await jobs.read(agent, job_id) + if reading.status == "done": + break + await asyncio.sleep(0.2) + assert reading.status == "done" + assert "done-late" in reading.body + + +async def test_off_timeout_keeps_the_hard_stop(tmp_path): + """With the auto-convert off, a timed-out command is killed as before and no + job is left running.""" + agent = _agent(tmp_path, background_on_timeout=False) + outcome = await _run_shell( + _context(agent), {"command": "sleep 2; echo late", "timeout": 1} + ) + + assert "stopped after" in outcome.content + assert not jobs.for_chat(agent.chat_id) + + +async def test_feature_off_is_the_plain_path(tmp_path): + """No wrapper, no job, byte-for-byte the old wording.""" + agent = _agent(tmp_path, background=False) + outcome = await _run_shell(_context(agent), {"command": "echo hi"}) + + assert "hi" in outcome.content + assert not jobs.for_chat(agent.chat_id) + + +# --- Explicit background=true -------------------------------------------------- +async def test_background_true_returns_at_once_with_a_job(tmp_path): + agent = _agent(tmp_path) + started = time.monotonic() + outcome = await _run_shell( + _context(agent), {"command": "sleep 5", "background": True} + ) + assert time.monotonic() - started < 3, "it should not wait for the command" + + running = jobs.for_chat(agent.chat_id) + assert len(running) == 1 + assert "job " in outcome.content + await jobs.stop(agent, running[0].id) + + +# --- The job tools ------------------------------------------------------------- +async def test_job_output_reads_a_running_job(tmp_path): + from lembas.services.agent.tools import _run_job_output + + agent = _agent(tmp_path) + job = await jobs.launch(agent, "sleep 5") + outcome = await _run_job_output(_context(agent), {"id": job.id}) + assert "running" in outcome.content.lower() + await jobs.stop(agent, job.id) + + +async def test_job_stop_ends_it(tmp_path): + from lembas.services.agent.tools import _run_job_output, _run_job_stop + + agent = _agent(tmp_path) + job = await jobs.launch(agent, "sleep 30") + stopped = await _run_job_stop(_context(agent), {"id": job.id}) + assert "Stopped" in stopped.content + + reading = await _run_job_output(_context(agent), {"id": job.id}) + assert "running" not in reading.content.lower() + + +async def test_job_output_refuses_a_bad_id(tmp_path): + from lembas.services.agent.tools import _run_job_output + + agent = _agent(tmp_path) + outcome = await _run_job_output(_context(agent), {"id": "../../etc/passwd"}) + assert outcome.event["status"] == "error" + + +async def test_a_job_is_only_visible_to_its_own_chat(tmp_path): + """for_chat is keyed on chat id; the file paths are too, so a model in + another chat cannot even name it.""" + agent_a = _agent(tmp_path, chat_id="chataaaa") + await jobs.launch(agent_a, "sleep 5") + assert jobs.for_chat("chatbbbb") == [] + for job in jobs.for_chat("chataaaa"): + await jobs.stop(agent_a, job.id) + + +# --- Gating -------------------------------------------------------------------- +def test_job_tools_appear_only_when_enabled(tmp_path): + from lembas.services.agent.tools import tool_defs + + on = {t.name for t in tool_defs(_agent(tmp_path, background=True))} + off = {t.name for t in tool_defs(_agent(tmp_path, background=False))} + + assert {"job_output", "job_list", "job_stop"} <= on + assert not ({"job_output", "job_list", "job_stop"} & off) + + +def test_background_param_appears_only_when_enabled(tmp_path): + from lembas.services.agent.tools import tool_defs + + def shell(agent): + return next(t for t in tool_defs(agent) if t.name == "shell_run") + + on = shell(_agent(tmp_path, background=True)).parameters["properties"] + off = shell(_agent(tmp_path, background=False)).parameters["properties"] + assert "background" in on + assert "background" not in off