"""The sub-agent's turn loop. WHAT MAKES THIS DIFFERENT FROM THE ATTEMPT THAT WAS RETIRED The archived sub-agent orchestrator was torn down for accuracy, and its post-mortem names two mechanisms. Both are answered here, and neither is answered in the prompt - they are answered by things the model cannot talk its way past: * "hallucinated library APIs and wrote self-confirming tests" -> finish() is refused until every task is closed, and the verification command it supplies is RE-RUN by vmd afterwards. The exit code the user sees comes from that second run. * "agents allowed to shortcut their tools guessed instead" -> plan_tasks must come first and finish() is refused if nothing was ever run. There is no path from "start" to "done" that does not go through the machine. It also compacts rather than cutting the session short. The old loop forced a final answer at a fraction of num_ctx, which is why long jobs ended mid-thought. EVERY EXIT PRODUCES A RESULT A step cap, a deadline, a cancellation, an unreachable model, a wedged machine - each of them ends with something for the user to read. A session that returns nothing is the worst outcome available, because the work happened and the account of it was thrown away. """ from __future__ import annotations import contextlib import json import logging import os import time from dataclasses import dataclass, field from typing import Any import compact import events as events_mod import ollama import tools as tools_mod from client import Vmd, VmdError, VmdUnavailable log = logging.getLogger("vmagent.loop") # ONE BUDGET, AND IT IS THE GENEROUS ONE. # # Seeing the result of an action on screen is its own turn - look, wait, # click, look again - where a command returns its result in the same one. # A session that built a working 259-line application in nine steps then # used its remaining twenty looking at the screen and stopped one # screenshot short of proving it worked. Every session can now reach for # the screen, so every session gets the budget that allows for it. MAX_STEPS = int(os.environ.get("OWUITOOLS_VM_MAX_STEPS", "110")) SESSION_SECONDS = float(os.environ.get("OWUITOOLS_VM_SESSION_SECONDS", "7200")) # How many turns in a row the model may answer with prose and no tool # call before it is finished for it. Three is enough to recover from one # confused turn and not enough to spend a session talking. MAX_STALLS = 3 # Same command, same error, this many times -> say so, in tool output. REPEAT_LIMIT = 3 # How many steps may pass without the task list moving before the agent is # reminded that somebody is reading it. See the note at the use site. TASK_DRIFT_LIMIT = int(os.environ.get("OWUITOOLS_VM_TASK_DRIFT", "12")) TASK_VERBS = frozenset({"plan_tasks", "update_task", "add_task"}) @dataclass class Job: task: str os_name: str interface: str project: str | None chat_id: str = "" message_id: str = "" owner: str = "" answer: str = "" deadline: float = field(default_factory=lambda: time.monotonic() + SESSION_SECONDS) class Outcome: """What the loop hands back. Always populated, whatever happened.""" def __init__(self) -> None: self.report: str = "" self.state: str = "done" # done | question | unfreeze | cancelled | failed self.steps: int = 0 self.notes: list[str] = [] def run(job: Job) -> Outcome: outcome = Outcome() ui = events_mod.Events(job.chat_id, job.message_id, job.interface) ui.status("Starting a machine…", force=True) try: vmd = Vmd() vmd.connect() except VmdUnavailable as exc: outcome.state = "failed" outcome.report = ( f"FAILED: the VM service is not available ({exc}). No machine was started " "and nothing was run. This is a fault on this host, not a problem with the " "request." ) ui.status("Could not start a machine", done=True, force=True) return outcome try: # Does not wait for the boot: the planning turn below happens # while the machine comes up, which also gets the model loaded and # its prompt prefilled before there is anything to do with it. # The job text goes with it so vmd can label the repository from # the work rather than from a model's prose. vmd derives the one # line it uses; nothing here decides the wording. opened = vmd.call("session_open", os=job.os_name, project=job.project, owner=job.owner, task=job.task, wait=False, chat_id=job.chat_id) except (VmdError, VmdUnavailable) as exc: outcome.state = "failed" outcome.report = f"FAILED: could not start the machine. {exc}" ui.status("Could not start a machine", done=True, force=True) vmd.close() return outcome ui.status(f"Booting a {job.os_name} machine - planning while it starts", force=True) try: _converse(vmd, job, ui, outcome, opened) except Exception as exc: # noqa: BLE001 log.exception("the agent loop failed") outcome.state = "failed" outcome.notes.append(f"the agent loop stopped unexpectedly: {type(exc).__name__}: {exc}") finally: # ORDER MATTERS. session_close is what commits anything left # uncommitted and pushes to Gitea, so the report has to be built # after it or the link is never in it. The machine is destroyed # inside the same call, which is fine: the report is assembled # from vmd's own record of the session, not from the machine. try: vmd.call("session_close") except Exception as exc: # noqa: BLE001 log.warning("session_close failed: %s", exc) try: outcome.report = vmd.call("report")["text"] except Exception as exc: # noqa: BLE001 outcome.report = outcome.report or f"FAILED: could not assemble a report ({exc})." ui.clear_embeds() ui.status("Done" if outcome.state == "done" else f"Stopped: {outcome.state}", done=True, force=True) vmd.close() if outcome.notes: outcome.report += "\n\n" + "\n".join(f"_{n}_" for n in outcome.notes) return outcome def _opening_message(job: Job, opened: dict[str, Any]) -> str: machine = "Ubuntu" if job.os_name == "ubuntu" else "Windows" lines = [f"Task: {job.task}", ""] lines.append( f"A {machine} machine is booting for you right now. It is not ready yet, and you " "have no way to reach it until it is." ) lines.append( "Write your task list while it comes up. That is the whole of this turn: call " "plan_tasks and stop. You will be told the moment the machine is ready, and you " "can start then - nothing is lost by planning first, and the machine would " "otherwise be booting while you sat idle." ) lines.append("") lines.append( "You get its command line and its desktop. Most work is the command line; the " "screen is there when something has to be seen or clicked." ) if opened.get("share"): lines.append( f"The project folder is /mnt/project (project '{job.project}'). It is the only " "thing that survives this session." ) if opened.get("storage"): lines.append(opened["storage"]) else: lines.append("There is no project folder, so nothing will be kept when this ends.") if job.answer: lines += ["", f"The user answered your question: {job.answer}"] return "\n".join(lines) def _converse(vmd: Vmd, job: Job, ui: events_mod.Events, outcome: Outcome, opened: dict[str, Any]) -> None: system_prompt = _load_prompt(job.os_name) messages: list[dict[str, Any]] = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": _opening_message(job, opened)}, ] PROTECTED = 2 # never compacted: the prompt and the task # Phase one: the machine is booting and only the planning tools are # offered. Phase two starts when the agent has a list and the machine # answers - see _hand_over_the_machine. booting = True schemas = tools_mod.schemas(job.os_name, booting=True) stalls = 0 since_task = 0 repeats: dict[str, int] = {} for step in range(MAX_STEPS): outcome.steps = step + 1 # --- the things that end a session, checked every turn --------- if _cancelled(vmd): outcome.state = "cancelled" outcome.notes.append("the session was cancelled.") return if time.monotonic() > job.deadline: outcome.state = "failed" outcome.notes.append( f"the session hit its time limit after {step} steps and was stopped. " "What had been committed is kept." ) return # --- keep the context affordable -------------------------------- before = compact.estimate_tokens(messages) if before > compact.TARGET_TOKENS: messages, notes = compact.compact(messages, PROTECTED, _summariser()) if notes: log.info("compacted %d -> %d tokens: %s", before, compact.estimate_tokens(messages), "; ".join(notes)) ui.heartbeat() try: # The only streaming call. The compaction summariser below is # left alone: nobody is watching that, and it is the one call # whose output is machinery rather than the agent talking. reply = ollama.chat(messages, tools=schemas, on_text=ui.speech) except ollama.OllamaError as exc: outcome.state = "failed" outcome.notes.append(f"the model became unreachable: {exc}") return assistant: dict[str, Any] = {"role": "assistant", "content": reply["content"]} if reply["tool_calls"]: assistant["tool_calls"] = reply["tool_calls"] # INTO THE SESSION LOG, BESIDE WHAT IT THEN DID. # # vmd records every verb it serves, which is what the agent DID; # this is what it said it was doing. Interleaved they read as an # account of the session rather than a list of calls. Suppressed # rather than checked: a log is never worth a step. if reply["content"].strip(): with contextlib.suppress(Exception): vmd.call("note", turn=step + 1, text=reply["content"]) # The thinking is deliberately NOT kept. It is the largest thing # the model produces, it is not needed to continue, and carrying # it forward would fill the window with reasoning about steps # that are already done. messages.append(assistant) if not reply["tool_calls"]: stalls += 1 if stalls >= MAX_STALLS: outcome.state = "failed" outcome.notes.append( "the agent stopped calling tools without finishing. What it had done " "up to that point is below." ) return messages.append({"role": "user", "content": _nudge(vmd)}) continue stalls = 0 for call in reply["tool_calls"]: function = call.get("function") or {} name = str(function.get("name", "")) arguments = function.get("arguments") or {} if isinstance(arguments, str): try: arguments = json.loads(arguments) except ValueError: arguments = {} ui.status(_status_for(vmd, name, arguments), force=(name in ("run", "finish"))) _show(ui, _terminal_for, name, arguments) text, image = tools_mod.call(vmd, name, arguments) # A model that runs the same failing command over and over is # not going to stop being told so in the system prompt; it is # told here, where it reads instructions as fact. if name == "run": key = f"{arguments.get('command','')}|{text[:200]}" repeats[key] = repeats.get(key, 0) + 1 if repeats[key] >= REPEAT_LIMIT: text += ( f"\n\n[You have now run this command {repeats[key]} times and got the " "same result. It will not change. Either try something different, or " "mark the task blocked with this as the reason.]" ) # THE LIST IS THE ONLY THING THE WATCHER CAN READ. # # An agent that plans twelve tasks and then does the work in one # long stretch without touching the list leaves the person # watching on "1/12" for twenty minutes - measured - while the # file grows and the status line says nothing. The work was # fine; the report of it was a flat line. # # Said here rather than in the prompt, because the prompt # already asks for it and asking has not worked. It is a nudge, # not a refusal: being mid-task is legitimate, and a tool that # blocked on it would be wrong far more often than right. if name in TASK_VERBS: since_task = 0 elif (since_task := since_task + 1) >= TASK_DRIFT_LIMIT: since_task = 0 text += ( f"\n\n[{TASK_DRIFT_LIMIT} steps have passed without your task list " "moving. Whoever asked for this is watching that list and it still says " "you are where you started. Mark what you have finished, mark what you " "are on now, and add anything you have taken on that was not in the " "plan.]" ) _show(ui, _terminal_result, name, text) message: dict[str, Any] = {"role": "tool", "content": compact.cap_result(text), "_label": name} if image is not None: message["images"] = [_b64(image)] ui.screenshot(image, caption=_status_for(vmd, name, arguments)) messages.append(message) if name == "finish" and not text.startswith(("REFUSED", "FAILED")): outcome.state = "done" return # --- hand the machine over, once there is a plan to work to ----- if booting and _status(vmd).get("planned"): handover = _hand_over_the_machine(vmd, job, ui, outcome) if handover is None: return # the machine never came up messages.append(handover) booting = False schemas = tools_mod.schemas(job.os_name, booting=False) # --- a pending question outranks continuing --------------------- state = _status(vmd) if state.get("question"): outcome.state = "question" return if state.get("unfreeze_request"): outcome.state = "unfreeze" return if state.get("status_line"): ui.status(state["status_line"]) ui.tasks(state.get("tasks", "")) outcome.state = "failed" outcome.notes.append( f"the agent used all {MAX_STEPS} of its steps without finishing. What it did is below." ) def _hand_over_the_machine(vmd: Vmd, job: Job, ui: events_mod.Events, outcome: Outcome) -> dict[str, Any] | None: """Wait for the boot to finish, then tell the agent it may start. By the time this runs the agent has already spent a turn planning, so most or all of the boot has happened underneath it. What is left is usually a few seconds. The screenshot is taken here rather than left for the agent to ask for: a GUI session's first act is always to look at the screen, and doing it now saves a whole model turn. """ ui.status("Waiting for the machine to finish booting", force=True) try: # NOT desktop=True. The graphical session is waited for lazily, on # the first screen verb, so a job that never looks at the screen # never pays for GNOME finishing its paint. See vmd._require_desktop. ready = vmd.call("session_ready", timeout=300) except (VmdError, VmdUnavailable) as exc: outcome.state = "failed" outcome.notes.append(f"the machine never finished booting: {exc}") return None machine = "Ubuntu" if job.os_name == "ubuntu" else "Windows" took = ready.get("seconds", 0) lines = [ f"The {machine} machine is up and you can reach it now" + (f" (it took {took:.0f}s, most of which was while you were planning)." if took else "."), ] # What is actually in the machine, read from the machine. Cheaper than # the three turns an agent otherwise spends finding out whether python # is there, what the browser is called, and where the project folder # is - and it cannot drift from the image the way a prompt would. facts = (ready.get("facts") or "").strip() if facts: lines.append("What is in it:\n" + facts) lines.append("Work through your task list. Mark each one in_progress as you start it.") # NO SCREENSHOT IS TAKEN HERE ANY MORE. # # Handing one over unasked put a picture of an idle desktop in front of # the reader before the session had done anything, and opened the # preview pane for every job including the ones that never look at a # screen. The operator's words: pop up the preview only when the agent # takes a screenshot. So the first picture in the chat is now one the # agent chose to take, which also makes it worth looking at. message: dict[str, Any] = {"role": "user", "content": "\n".join(lines)} return message # --------------------------------------------------------------------- # helpers # --------------------------------------------------------------------- def _load_prompt(os_name: str) -> str: """The prompt for THIS machine, not for machines in general. TWO MACHINES, TWO PROMPTS. Ubuntu or Windows, and the differences are not decoration: the project folder is `/mnt/project` or `Z:\\`, and the shell is bash or PowerShell 5.1 - which has no `&&`. Sending one prompt for both made the agent reason about which half applied to it. Measured: a Windows session wrote itself the task "check /mnt/project accessibility", because the prompt said that was where work survives and nothing told it otherwise. It cost a step and would have cost the file if the path had been writable. The screen section is the same for both, because every session has one. WHERE THESE COME FROM, AND WHY THE FIRST ONE IS NOT OPTIONAL. They are bind-mounted from configs/agent rather than baked into the image, so rewording one is a restart instead of a rebuild. That is worth having and it also introduces a way for them to be MISSING that did not exist before - a mount that was never declared, or a path typed wrong - where previously only a broken build could do it. So the base prompt is required and a missing one raises. An agent running with no prompt at all still calls tools and still reports success, which is this subsystem's characteristic failure: it looks like it worked. Refusing to start is the cheaper end of that trade. The per-machine and screen fragments keep the old policy and are skipped with a warning, because a thinner prompt genuinely does still run a session. """ # Same resolution as tools.py: the env var in the container, where # configs/agent is bind-mounted at /configs/agent; otherwise the # tree's own copy, so the tests and anyone reading the repo need no # environment set up. here = (os.environ.get("OWUITOOLS_AGENT_CONFIG_DIR") or os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "configs", "agent")) try: with open(os.path.join(here, "prompt.md"), encoding="utf-8") as handle: parts = [handle.read().rstrip()] except OSError as exc: raise RuntimeError( f"the agent's base prompt is not readable at {here}/prompt.md ({exc}). " "configs/agent is mounted there by docker-compose.yml; without it the agent " "would run with no instructions and still report success." ) from exc for name in (f"prompt.{os_name}.md", "prompt.screen.md"): try: with open(os.path.join(here, name), encoding="utf-8") as handle: parts.append(handle.read().rstrip()) except OSError: log.warning("prompt fragment %s is missing; continuing without it", name) return "\n\n".join(parts) + "\n" def _summariser(): def summarise(head: list[dict[str, Any]]) -> str: try: reply = ollama.chat( [{"role": "system", "content": compact.SUMMARY_PROMPT}, {"role": "user", "content": _transcribe(head)}], tools=None, think=False, ) return reply["content"].strip() or "(the summary came back empty)" except ollama.OllamaError as exc: # Degrade to a mechanical list rather than losing the session. # Open WebUI's own compactor does the same when its summary # call fails, for the same reason. log.warning("summary call failed, falling back to a list: %s", exc) return _transcribe(head)[:4000] return summarise def _transcribe(messages: list[dict[str, Any]]) -> str: lines = [] for message in messages: role = message.get("role", "?") label = message.get("_label") content = (message.get("content") or "")[:1500] lines.append(f"[{role}{'/' + label if label else ''}] {content}") return "\n\n".join(lines) def _b64(data: bytes) -> str: import base64 return base64.b64encode(data).decode() def _cancelled(vmd: Vmd) -> bool: return bool(_status(vmd).get("cancelled")) def _status(vmd: Vmd) -> dict[str, Any]: try: return vmd.call("session_status") except (VmdError, VmdUnavailable): return {} def _nudge(vmd: Vmd) -> str: state = _status(vmd) return ( "You answered without calling a tool, and the session is not finished.\n\n" f"{state.get('tasks', '')}\n\n" "Call a tool to make progress, or call finish if the work is done - and if it is " "not done and cannot be, mark the remaining tasks blocked with the reason and " "finish anyway. Do not reply with prose again." ) def _show(ui: Any, render: Any, name: str, payload: Any) -> None: """Put a line on the watcher's screen, and never fail doing it. This is a picture of the session for somebody watching a chat. It has no bearing on whether the work succeeds, so it does not get a vote. Measured, and the reason this exists: `finish` arrived with its `verification` as a string where the schema says an object, the renderer called `.get` on it, and the AttributeError propagated out of the display path and killed the agent loop - discarding a session that had already completed all six of its tasks and verified them. One malformed argument from the model must cost a line of output, not the work. """ try: ui.terminal(*render(name, payload)) except Exception: # noqa: BLE001 - a view may never break a session log.debug("could not render %s for the live view", name, exc_info=True) def _terminal_for(name: str, arguments: Any) -> list[str]: """The command line, as a person would have typed it. A prompt and a command, not a description of one: the point of this view is that somebody watching can see the session the way they would see their own terminal, and recognise a mistake in it. NOTHING IN HERE MAY RAISE. These arguments come from the model, and a model that has been asked for an object will sometimes send a string. Measured, and the reason this note exists: `finish` was called with `verification` as a string, `.get` was called on it, and the AttributeError came out of the display code and killed the agent loop - throwing away a session that had finished all six of its tasks. A view of the work must never be able to destroy the work, so every read here is defensive and the whole call is wrapped by `_safe_terminal` below. """ if not isinstance(arguments, dict): return [] match name: case "run": return [f"$ {str(arguments.get('command', ''))[:400]}"] case "write_file": body = str(arguments.get("content", "")) return [f"$ cat > {arguments.get('path', '')} ({len(body)} bytes)"] case "append_file": body = str(arguments.get("content", "")) return [f"$ cat >> {arguments.get('path', '')} ({len(body)} bytes)"] case "edit_file": old_text = " ".join(str(arguments.get("old", "")).split())[:60] return [f"$ edit {arguments.get('path', '')} ({old_text!r})"] case "read_file": return [f"$ cat {arguments.get('path', '')}"] case "list_dir": return [f"$ ls -l {arguments.get('path', '.')}"] case "git_commit": return [f"$ git commit -m {str(arguments.get('message', ''))[:100]!r}"] case "check_internet": return ["$ curl -sS -o /dev/null -w '%{http_code}' https://1.1.1.1/"] case "reboot_vm": return ["$ systemctl reboot"] case "web_search": return [f"# searching the web: {str(arguments.get('query', ''))[:120]}"] case "read_page" | "open_page": return [f"# {name}: {str(arguments.get('url', ''))[:150]}"] case "wait": return [f"$ sleep {arguments.get('seconds', '')}"] # The desktop verbs. Without these a GUI session's command feed is # a box of blank rows: every tool it uses is one of these, and # none of them used to render. case "launch_app": return [f"$ start {arguments.get('path', '')} (on the desktop)"] case "screenshot": return ["# looking at the screen"] case "click": return [f"# click ({arguments.get('x', '?')}, {arguments.get('y', '?')})" + (f" {arguments['button']}" if arguments.get("button") else "")] case "move": return [f"# move to ({arguments.get('x', '?')}, {arguments.get('y', '?')})"] case "key": return [f"# key {str(arguments.get('name') or arguments.get('key', ''))[:40]}"] case "type": return [f"# type {str(arguments.get('text', ''))[:120]!r}"] case "scroll": return [f"# scroll {arguments.get('amount', '')} at " f"({arguments.get('x', '?')}, {arguments.get('y', '?')})"] case "finish": # The schema says an object with a `command`; a string is what # arrives often enough to matter, and either is worth showing. check = arguments.get("verification") if isinstance(check, dict): check = check.get("command", "") return [f"$ {str(check or '')[:300]}"] if check else [] case _: return [] def _terminal_result(name: str, text: str) -> list[str]: """What the machine said back, trimmed to what a screen would show. Capped here rather than in events.py so a command that prints a megabyte does not put a megabyte through the progress file on its way to being cut down to twenty lines. """ # Nothing whose "result" is bookkeeping rather than output. write_file # returns {"path": ..., "bytes": ...}, which is right for the model and # is a JSON object sitting in the middle of what is supposed to look # like a terminal - and the command line above it already says the # path and the size. if name in ("plan_tasks", "update_task", "add_task", "screenshot", "click", "move", "key", "type", "scroll", "write_file", "open_page", "wait"): return [] lines = [line for line in (text or "").splitlines() if line.strip()] # `run` comes back as "exit code: 0 (0.4s) / stdout: / ...", which is # the shape the MODEL needs and not the shape a terminal has. The # person watching is looking at something that should read like their # own screen, so the framing is unwrapped: output as output, and the # exit code only when it is news. if name == "run": code, out = None, [] section = None for line in lines: if line.startswith("exit code:"): code = line.split(":", 1)[1].strip().split()[0] elif line in ("stdout:", "stderr:"): section = line[:-1] elif line.startswith("[output was truncated"): out.append(line) else: out.append(line if section != "stderr" else line) if code not in (None, "0"): out.append(f"[exit {code}]") lines = out if len(lines) > 12: lines = lines[:12] + [f"... ({len(lines) - 12} more lines)"] return lines def _status_for(vmd: Vmd, name: str, arguments: dict[str, Any]) -> str: """The one line the watching user sees.""" line = _status(vmd).get("status_line", "") prefix = f"{line} - " if line else "" match name: case "run": return prefix + f"running `{str(arguments.get('command', ''))[:70]}`" case "write_file": return prefix + f"writing {arguments.get('path', '')}" case "read_file" | "list_dir": return prefix + f"reading {arguments.get('path', '')}" case "screenshot": return prefix + "looking at the screen" case "click" | "move": return prefix + f"clicking ({arguments.get('x')}, {arguments.get('y')})" case "type": return prefix + "typing" case "key": return prefix + f"pressing {arguments.get('keys', '')}" case "git_commit": return prefix + "committing" case "check_internet": return prefix + "checking the network" case "reboot_vm": return prefix + "rebooting the machine" case "finish": return prefix + "verifying the work" case "plan_tasks": return "writing the task list" case _: return prefix + name