diff --git a/src/hmz/coganchor/agents/base.py b/src/hmz/coganchor/agents/base.py index 715eb67c6..80660a61a 100644 --- a/src/hmz/coganchor/agents/base.py +++ b/src/hmz/coganchor/agents/base.py @@ -484,6 +484,20 @@ class SessionBase(ABC): #: cannot otherwise tell that turn from one that has wedged. narrates: ClassVar[bool] = False + #: Whether a fork of this backend's conversation may be opened in another directory than + #: the one the conversation is in. Only where the CLI is told where the child works as it + #: cuts it -- a thread forked with a `cwd` of its own, a `fork --cwd` -- or where a store + #: kept per directory can be carried across first, which :meth:`_carry` is for. False for + #: every backend that has not been checked to: a fork its CLI looks for in the wrong + #: directory is a first turn that fails saying the conversation does not exist. + forks_elsewhere: ClassVar[bool] = False + + #: Whether :meth:`cut` puts down what carries a turn, for a backend an interrupt does not + #: reach while a turn is under way: one whose turns an app server or a daemon holds for + #: every session of the agent, and one whose runtime holds them where nothing reaches in. + #: A goal on those is no turn at all, and is reached the same way. + cuts_transport: ClassVar[bool] = False + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: @@ -2344,7 +2358,12 @@ def forks(self) -> bool: profile = named(self._agent.backend) return profile is not None and profile.forks - def fork(self) -> SessionBase: + def fork( + self, + *, + into: AgentBase | None = None, + cwd: str | os.PathLike[str] | None = None, + ) -> SessionBase: """A second conversation carrying this one's history, and its own from here on. Which is what a conversation that has got somewhere is worth: a flow that has spent @@ -2372,15 +2391,30 @@ def fork(self) -> SessionBase: after another turn has been sent here would branch from somewhere else, which is not what was asked for, and is refused rather than done quietly. + The child may belong to another agent of the same backend and account -- one set up + differently, at another rung or carrying other skills -- and may work in another + directory, where :attr:`forks_elsewhere` says the backend can be told so. Both are + what a caller holding an agent per conversation needs to branch one: the history is + the account's, so any agent signed in as it can carry it on. + + Args: + into: The agent the child is a conversation of, or None for this one's. It must be + of this backend, on the account this conversation is kept under, and working on + the same machine. + cwd: Where the child works, or None for where this conversation does. + Returns: The new session, unopened with the backend: it is the fork, and the fork happens where its first turn does. Which is what makes two forks of one conversation cost nothing until they are used. Raises: - NotImplementedError: If this backend has no fork to reach for. A flow handed back a - second handle on the one conversation would be two loops writing into one, so it - is refused where it is asked -- and :attr:`forks` is how a flow asks first. + NotImplementedError: If this backend has no fork to reach for, or none that can be + opened in another directory. A flow handed back a second handle on the one + conversation would be two loops writing into one, so it is refused where it is + asked -- and :attr:`forks` is how a flow asks first. + ValueError: If `into` is of another backend, on another account or on another + machine, none of which can read this conversation where it is kept. RuntimeError: If no turn has landed here yet, so there is no conversation to carry: a session that has got nowhere is one to open rather than one to fork. """ @@ -2389,8 +2423,28 @@ def fork(self) -> SessionBase: f"{self._agent.backend} has no way of carrying a conversation " "into a second one" ) + agent = self._agent if into is None else into + if agent is not self._agent and ( + type(agent) is not type(self._agent) + or agent.backend != self._agent.backend + or agent.config.provider != self._agent.config.provider + or agent.config.machine != self._agent.config.machine + ): + raise ValueError( + f"{self._agent.backend}: a conversation is carried on only by an agent of " + "the same backend, signed in as the same account, on the same machine" + ) + where = self._cwd + elsewhere = cwd is not None and os.path.abspath(cwd) != self.cwd # noqa: PTH100 + if elsewhere and not type(self).forks_elsewhere: + raise NotImplementedError( + f"{self._agent.backend} cannot carry a conversation into another directory" + ) seed = self.id # raises while nothing has landed, which is nothing to carry - made = self._agent._opens_at(self._cwd) + if elsewhere and cwd is not None: + where = os.fspath(cwd) + self._carry(where) + made = agent._opens_at(where) made._forked_from = seed # Where this conversation had got to when the fork was asked for, and a weak hold on # the conversation itself: the child checks both as it opens, so that a fork taken @@ -2409,6 +2463,38 @@ def fork(self) -> SessionBase: made.offers(self._tools) return made + def _carry(self, cwd: str) -> None: + """Puts this conversation where a fork of it opened in another directory will look. + + Nothing by default: a backend whose CLI is told where the child works as it cuts it + finds the conversation wherever it is kept. One that keeps its conversations per + directory says so here, by copying this one across before the child's first turn. + + Args: + cwd: The directory the child is to work in. + """ + del cwd + + def cut(self, *, why: str) -> None: + """Cuts the turn under way off, whatever is holding it, for a caller holding it alone. + + :meth:`interrupt`, and then -- on a backend whose turns are held somewhere every + session of the agent shares, an app server or a daemon, where an interrupt is heard + only at the next answer -- that transport put down, which is what ends the turn now. + It also reaches a goal, which such a backend runs outside a turn of its own. The + conversation is not ended by it: the next turn starts the transport again and resumes + this conversation by its id, as it does after a watchdog. + + For a caller that holds the agent for this one conversation. On one holding several, + putting the transport down ends the turns of all of them. + + Args: + why: What to say it was cut off for. + """ + self.interrupt(why=why) + if type(self).cuts_transport: + self._lets_go() + def _at_the_boundary(self) -> None: """Refuses a fork whose conversation has moved on since it was asked for. diff --git a/src/hmz/coganchor/agents/claude.py b/src/hmz/coganchor/agents/claude.py index ec4594a5a..ac36da4d2 100644 --- a/src/hmz/coganchor/agents/claude.py +++ b/src/hmz/coganchor/agents/claude.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os import uuid from collections import Counter from dataclasses import dataclass @@ -14,7 +15,6 @@ from .hooks import EVERYWHERE, SUBAGENTS, WAITING, Moment, about, arriving if TYPE_CHECKING: - import os from collections.abc import Iterator #: The tool Claude reaches for when it wants a person rather than a file. Its input is a list @@ -57,6 +57,55 @@ _ALLOWED_TOOLS_MAX = 32 +#: How long a directory spelled as a name may be before Claude cuts it short and tells it +#: apart from others cut the same way by a hash of the whole. +_PROJECT_MAX = 200 + + +def _project(where: str) -> str: + """A directory as Claude names the folder its conversations there are kept in. + + Every character that is not an ASCII letter or digit is a dash -- counted as Claude counts + them, in UTF-16 code units, so a character outside the basic plane is two -- and a name + longer than :data:`_PROJECT_MAX` is cut there and given the base-36 of a 32-bit string + hash of the whole path. Read off Claude Code 2.1.282, where it is the one function that + names a project directory. + + Args: + where: The directory, absolute. + + Returns: + The folder's name. + """ + units = where.encode("utf-16-le") + codes = [ + int.from_bytes(units[at : at + 2], "little") for at in range(0, len(units), 2) + ] + said = "".join( + chr(code) if chr(code).isascii() and chr(code).isalnum() else "-" + for code in codes + ) + if len(said) <= _PROJECT_MAX: + return said + hashed = 0 + for code in codes: + hashed = (hashed * 31 + code) & 0xFFFFFFFF + if hashed >= 1 << 31: + hashed -= 1 << 32 + return f"{said[:_PROJECT_MAX]}-{_base36(abs(hashed))}" + + +def _base36(number: int) -> str: + """A non-negative number written in base 36, as JavaScript's `toString(36)` writes it.""" + digits = "0123456789abcdefghijklmnopqrstuvwxyz" + said = "" + while True: + number, digit = divmod(number, 36) + said = digits[digit] + said + if not number: + return said + + _ALLOWED_TOOL_RULE_MAX_CHARS = 4096 #: Reasons that leave an answer unfinished even when a broken intermediary labels the result @@ -245,6 +294,11 @@ class ClaudeCodeSession(StreamSessionBase): #: reads them -- which is why it is declared here rather than in `hmz.coganchor.backends`. narrates: ClassVar[bool] = True + #: Claude keeps a conversation under the directory it was held in, and `--resume` looks + #: for it under the directory it is run in. A fork opened elsewhere is carried there first, + #: by :meth:`_carry`. + forks_elsewhere: ClassVar[bool] = True + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: @@ -1019,6 +1073,53 @@ def _reply(self, said: dict[str, Any], answer: dict[str, Any]) -> None: + "\n" ) + def _carry(self, cwd: str) -> None: + """Copies this conversation to where a Claude run in another directory looks for it. + + Claude keeps a conversation as `projects//.jsonl` + under its home, and `--resume --fork-session` reads it from under the directory + it is run in. So a fork opened elsewhere is given a copy there to be cut from; the + copy is this conversation as it stands, which is what the fork carries on from. + + Args: + cwd: The directory the fork is to work in. + + Raises: + NotImplementedError: For an agent whose turns land on another machine, whose Claude + keeps its conversations there rather than here. + RuntimeError: If this conversation cannot be found where Claude keeps it. + """ + import shutil + + from hmz.coganchor.backends import named + + if self._agent.config.machine is not None: + raise NotImplementedError( + "claude cannot carry a conversation into another directory on another machine" + ) + profile = named("claude") + assert profile is not None # noqa: S101 -- the backend this driver is for + projects = profile.directory(self._environ()) / "projects" + # Where this conversation is held first: an earlier fork carried elsewhere left a + # copy of it there, as it stood then, which is not where it stands now. + held = os.path.abspath(self.cwd) # noqa: PTH100 + found = next( + ( + kept + for spelled in dict.fromkeys((held, os.path.realpath(held))) + if (kept := projects / _project(spelled) / f"{self.id}.jsonl").is_file() + ), + None, + ) or next(projects.glob(f"*/{self.id}.jsonl"), None) + if found is None: + raise RuntimeError(f"claude: no conversation {self.id} under {projects}") + where = os.path.abspath(cwd) # noqa: PTH100 + for spelled in dict.fromkeys((where, os.path.realpath(where))): + there = projects / _project(spelled) + there.mkdir(parents=True, exist_ok=True) + if (there / found.name) != found: + shutil.copyfile(found, there / found.name) + def _pursue(self, objective: str) -> str: """Runs the turn as Claude Code's own ``/goal``, which print mode expands like any other. diff --git a/src/hmz/coganchor/agents/codex.py b/src/hmz/coganchor/agents/codex.py index 293977edf..9ab12c25e 100644 --- a/src/hmz/coganchor/agents/codex.py +++ b/src/hmz/coganchor/agents/codex.py @@ -106,6 +106,9 @@ #: same thing straight through. _ANSWERED = ("browser_use", "computer_use", "standalone_web_search", "web_search") +#: The approval policies a config may put in place of its rung's, "" being the rung's own. +_APPROVAL_POLICIES = ("", "untrusted", "on-failure", "on-request", "never") + #: What the server calls a turn stopping to ask its user something. Every other request it #: makes of a client is an approval, which an unattended flow does not stop for. _ASKS = "item/tool/requestUserInput" @@ -214,6 +217,55 @@ class _Running: _KINDS = {"input": "inputTokens", "output": "outputTokens"} +#: The keywords of a JSON Schema whose value maps names to schemas, rather than being one. +_NAMED_SCHEMAS = ("properties", "$defs", "definitions", "patternProperties") + + +def strict(schema: Mapping[str, Any]) -> dict[str, Any]: + """A JSON Schema as the structured outputs Codex's models are held to will take it. + + Those take an object only with every property required and no other property allowed, + and refuse the whole turn otherwise -- `additionalProperties is required to be supplied + and to be false` -- where a pydantic model leaves a field with a default out of `required` + and says nothing of other properties. So every object here is closed and has every + property it names required, and defaults are dropped: the model writes every field, and + what it writes is read back through the pydantic model the schema came from, defaults + and all. An object that names no properties -- a mapping of whatever keys -- is left as + it is, having nothing to require. + + Args: + schema: The schema, as `model_json_schema()` writes it. + + Returns: + The same schema, held to that. + """ + + def held(node: object) -> object: + if isinstance(node, list): + return [held(one) for one in cast("list[Any]", node)] + if not isinstance(node, dict): + return node + said: dict[str, Any] = {} + for key, value in cast("dict[str, Any]", node).items(): + if key == "default": + continue + if key in _NAMED_SCHEMAS and isinstance(value, dict): + # Names mapped to schemas, where a name may be spelled like a keyword. + said[key] = { + name: held(one) + for name, one in cast("dict[str, Any]", value).items() + } + else: + said[key] = held(value) + named = said.get("properties") + if isinstance(named, dict): + said["required"] = list(cast("dict[str, Any]", named)) + said["additionalProperties"] = False + return said + + return cast("dict[str, Any]", held(schema)) + + def unattended(permission: str, service_tier: str = "default") -> dict[str, Any]: """What a thread is opened with, at the rung the agent was configured for. @@ -619,7 +671,9 @@ def _called(self, method: str, params: dict[str, Any]) -> Any: pass # a read with no deadline ends in a message or in the server stopping return self._answer(message, "") - def pursue(self, params: dict[str, Any]) -> str: + def pursue( + self, params: dict[str, Any], spends: Callable[[Usage], None] | None = None + ) -> str: """Starts a turn on a thread that has a goal, and reads until the goal is done with it. A goal is as many turns of the model as the objective takes, and Codex starts each one @@ -636,6 +690,8 @@ def pursue(self, params: dict[str, Any]) -> str: Args: params: What to start the first turn with. + spends: Told what each request of the goal cost as the server says it, which a + goal spends as surely as a turn does, or None to tell nobody. Returns: The last thing the agent said, stripped. @@ -675,6 +731,10 @@ def pursue(self, params: dict[str, Any]) -> str: # only sign it is running -- said whole, as each message of # it lands, which is how a turn is watched here too. say(said, sys.stderr) + case "thread/tokenUsage/updated" if spends is not None: + risen = self._rose(thread, message.get("params") or {}) + if risen is not None: + spends(risen) case "thread/goal/updated": pursuing = message["params"]["goal"]["status"] == "active" case "thread/goal/cleared": @@ -816,35 +876,7 @@ def turn(self, params: dict[str, Any], running: _Running) -> Iterator[Event]: ], ) case "thread/tokenUsage/updated": - # Sent as the turn spends it. `total` is the thread, every turn of - # it; `last` is the one request that just came back. Cached input - # is counted inside the input rather than beside it, so the input - # the server states is the whole of what went in -- and the two - # kinds together are the whole of what crossed the wire. - counted: dict[str, Any] = told.get("tokenUsage") or {} - usage: dict[str, Any] = counted.get("total") or {} - held = Counter( - { - kind: int(usage.get(named) or 0) - for kind, named in _KINDS.items() - if usage.get(named) - } - ) - if sum(held.values()): - risen = Usage( - { - kind: tokens - for kind in set(held) | set(before) - if ( - tokens := held[kind] - - (self._counted.get(thread) or Counter())[ - kind - ] - ) - > 0 - } - ) - self._counted[thread] = held + if (risen := self._rose(thread, told)) is not None: costing = costing + risen if running.spends is not None: # As the turn spends it rather than once it is over: a @@ -889,6 +921,56 @@ def turn(self, params: dict[str, Any], running: _Running) -> Iterator[Event]: spent=costing, ) + def _rose(self, thread: str, told: Mapping[str, Any]) -> Usage | None: + """What one `thread/tokenUsage/updated` says the thread has spent since the last. + + Sent as a turn spends it. `total` is the thread, every turn of it; `last` is the one + request that just came back. Cached input is counted inside the input rather than + beside it, so the input the server states is the whole of what went in -- and the two + kinds together are the whole of what crossed the wire. + + Args: + thread: The thread it is about. + told: The notification's params. + + Returns: + The rise by kind, and None for a notification that states nothing. + """ + counted: dict[str, Any] = told.get("tokenUsage") or {} + usage: dict[str, Any] = counted.get("total") or {} + held = Counter( + { + kind: int(usage.get(named) or 0) + for kind, named in _KINDS.items() + if usage.get(named) + } + ) + if not sum(held.values()): + return None + before = self._counted.get(thread) + self._counted[thread] = held + if before is None: + # The first word of a thread this server did not start: one picked back up after + # a server was put down, or a fork. Its total is the thread's whole history, which + # the server it was picked up from counted already, so what this request cost is + # what the server says it cost -- `last` -- and the total is where the next rise + # is read from. + last: dict[str, Any] = counted.get("last") or {} + return Usage( + { + kind: float(tokens) + for kind, named in _KINDS.items() + if (tokens := int(last.get(named) or 0)) > 0 + } + ) + return Usage( + { + kind: tokens + for kind in set(held) | set(before) + if (tokens := held[kind] - before[kind]) > 0 + } + ) + def steer(self, thread: str, turn: str, text: str, ticket: str) -> None: """Says something to a turn that is already running. @@ -1342,6 +1424,12 @@ class CodexAgentConfig(AgentConfig): and the web families are :attr:`~hmz.coganchor.agents.config.AgentConfig.web_search` and :attr:`~hmz.coganchor.agents.config.AgentConfig.permission` -- see :data:`_ANSWERED`. + approvals: Codex's own approval policy, in place of the one the rung comes with, or "" + for the rung's. Asked for by a caller that wants every command put to it first while + the sandbox stays where the rung puts it -- `untrusted` at `bypass` is full access + with nothing but a known-safe read run unasked, and every request answered by this + client, which grants it unless a hook hung on `PERMISSION_REQUEST` refuses. A policy + said here is not stepped down from on an installation that refuses the rung's sandbox. strict_config: Whether the app server refuses a setting it does not recognise rather than passing over it -- `--strict-config`. False, as Codex's own default is. A flow that turns it on finds out at the first turn that a key it names has been renamed by @@ -1372,9 +1460,16 @@ class CodexAgentConfig(AgentConfig): overrides: tuple[tuple[str, str], ...] = () features: tuple[tuple[str, bool], ...] = () strict_config: bool = False + approvals: str = "" def __post_init__(self) -> None: super().__post_init__() + if self.approvals not in _APPROVAL_POLICIES: + raise ValueError( + "approvals must be one of " + f"{', '.join(repr(one) for one in _APPROVAL_POLICIES)}, " + f"not {self.approvals!r}" + ) object.__setattr__(self, "overrides", _overrides(self.overrides)) object.__setattr__(self, "features", _features(self.features)) @@ -1401,6 +1496,14 @@ class CodexSession(SessionBase): #: under way rather than answered as the next one -- which is what an app server buys. steers: ClassVar[bool] = True + #: `thread/fork` takes the `cwd` the child is opened at, and reads the thread it cuts from + #: out of the account's own store, which is not kept per directory. + forks_elsewhere: ClassVar[bool] = True + + #: The app server holds every thread of the agent, and hears an interrupt only once the + #: turn has answered; a goal is not a turn at all. + cuts_transport: ClassVar[bool] = True + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: @@ -1471,16 +1574,11 @@ def _stream( "model": self._agent.config.model, **_thinking(self.effort), **( - {"outputSchema": schema.model_json_schema()} + {"outputSchema": strict(schema.model_json_schema())} if schema is not None else {} ), - **turning( - server.permitted( - self._agent.config.permission, - self._agent.config.service_tier, - ) - ), + **turning(self._permitted(server)), }, self._running, ): @@ -1534,6 +1632,27 @@ def interject(self, text: str) -> None: self.took(ticket) raise + def _permitted(self, server: _AppServer) -> dict[str, Any]: + """What this session's calls tell the server the agent may do, and who is asked. + + The rung's own settings, as the server will take them, with the approval policy the + config says in place of the rung's where it says one: a sandbox is the rung's, and + whether the server asks before a command is something a flow may want a say in + without the sandbox moving. + + Args: + server: The server the call is made on. + + Returns: + The settings to send. + """ + config = self._agent.config + rung = server.permitted(config.permission, config.service_tier) + asked = str(getattr(config, "approvals", "")) + if asked and "approvalPolicy" in rung: + rung["approvalPolicy"] = asked + return rung + def _thread(self, server: _AppServer) -> str: """The thread this session is, started or picked back up as needed. @@ -1544,9 +1663,7 @@ def _thread(self, server: _AppServer) -> str: Returns: The thread's id, which is also the session's. """ - rung = server.permitted( - self._agent.config.permission, self._agent.config.service_tier - ) + rung = self._permitted(server) if (thread := self._id) is None: opened = { "cwd": self._workspace(), @@ -1617,10 +1734,9 @@ def _pursue(self, objective: str) -> str: "input": [{"type": "text", "text": objective}], "model": config.model, **_thinking(self.effort), - **turning( - server.permitted(config.permission, config.service_tier) - ), - } + **turning(self._permitted(server)), + }, + self._spends, ) finally: server.give() diff --git a/src/hmz/coganchor/agents/dsh.py b/src/hmz/coganchor/agents/dsh.py index f3181c693..521006231 100644 --- a/src/hmz/coganchor/agents/dsh.py +++ b/src/hmz/coganchor/agents/dsh.py @@ -306,6 +306,9 @@ def new(self, cwd: str | os.PathLike[str] | None = None) -> DshSession: class DshSession(SessionBase): """One durable DeepSeek Harness conversation.""" + #: The SDK's runtime holds the turn, and nothing reaches into it to stop one. + cuts_transport: ClassVar[bool] = True + def __init__( self, agent: DshAgent, cwd: str | os.PathLike[str] | None = None ) -> None: diff --git a/src/hmz/coganchor/agents/kimi.py b/src/hmz/coganchor/agents/kimi.py index 044e4e3cd..07fec490e 100644 --- a/src/hmz/coganchor/agents/kimi.py +++ b/src/hmz/coganchor/agents/kimi.py @@ -861,6 +861,14 @@ class KimiCodeCLISession(SessionBase): #: put in rather than one waiting. steers: ClassVar[bool] = True + #: `kimi fork --cwd` cuts the child where it is told to work, out of the account's own + #: store rather than one kept beside the directory the parent is in. + forks_elsewhere: ClassVar[bool] = True + + #: The daemon holds every conversation of the agent, and nothing reaches into one of its + #: turns to stop it. + cuts_transport: ClassVar[bool] = True + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: diff --git a/src/hmz/coganchor/agents/zcode.py b/src/hmz/coganchor/agents/zcode.py index 76889a000..6621c5d22 100644 --- a/src/hmz/coganchor/agents/zcode.py +++ b/src/hmz/coganchor/agents/zcode.py @@ -1430,6 +1430,14 @@ class ZcodeSession(SessionBase): _agent: ZcodeAgent # every turn is run on the app server this agent holds + #: `session/fork` is told the workspace the child works in, and cuts it from a session + #: the server picks back up by id wherever that one was working. + forks_elsewhere: ClassVar[bool] = True + + #: The app server holds every session of the agent, and nothing reaches into one of its + #: turns to stop it. + cuts_transport: ClassVar[bool] = True + def __init__( self, agent: AgentBase, cwd: str | os.PathLike[str] | None = None ) -> None: diff --git a/src/hmz/flows/agents.py b/src/hmz/flows/agents.py index c6b50aeea..757d720ed 100644 --- a/src/hmz/flows/agents.py +++ b/src/hmz/flows/agents.py @@ -650,7 +650,7 @@ class KimiCodeAgent( """Kimi Code, with everything it can do.""" -class GrokBuildAgent(Agent, PermissionRequestHookAgentMixin, Protocol): +class GrokBuildAgent(Agent, Protocol): """Grok Build, with everything it can do.""" diff --git a/src/hmz/runtime/flowing/harnesses.py b/src/hmz/runtime/flowing/harnesses.py index 32807fbfd..be2c247ad 100644 --- a/src/hmz/runtime/flowing/harnesses.py +++ b/src/hmz/runtime/flowing/harnesses.py @@ -1,24 +1,171 @@ """The agent drivers: one :class:`~hmz.runtime.flowing.spi.AgentDriver` per coding agent CLI. -Not written yet. :func:`open_agent` is the contract the ways in are written against, and -says what it must do; until then it raises `NotImplementedError`. +Each is coganchor's own driver for its CLI, held so that it keeps what +:mod:`hmz.runtime.flowing.spi` promises. What a harness can be asked for is +:data:`~hmz.runtime.flowing.spi.HARNESS_CAPABILITIES`; how a +:class:`~hmz.flows.Permission` reaches its CLI, and what a hook's answer does there, is +:mod:`hmz.runtime.flowing.harnessing`. + +A session is a coganchor agent of its own and one conversation of it. The agent is what +coganchor settles everything a session runs by on -- the rung, the web, the skills, the +machine, the hooks and the person a question goes to -- and two sessions of one flow agent may +differ in every one of those, so each has its own; they share the CLI, the account and the +model, which is the driver. + +A turn is a coganchor turn on a thread of its own, so the engine's loop is never held by one. +Around it, on the loop, the driver fires the hooks that bracket a turn -- SESSION_START before +the first, USER_PROMPT_SUBMIT before each, STOP after each, SESSION_END as the session closes +-- and takes the turn again where a STOP hook or a steer sends the agent on. The moments that +arrive from inside the CLI -- a tool, a permission, a notification, a subagent, a question -- +come on coganchor's threads and reach the engine's hooks through a +:class:`~hmz.runtime.flowing.spi.HookBridge`, which answers as if nothing were hung once a +hook has kept one waiting for :data:`HOOK_TIMEOUT` -- a question excepted, which waits for a +person. A hook hung while a turn is running reaches the moments of that turn that come later, +except the few that decide how a CLI is started, which take hold from the next turn: +PRE_TOOL_USE on a CLI that gates its tools with a hook table of its own, and the hooks +:mod:`~hmz.runtime.flowing.harnessing` starts a CLI asking for -- Codex's `untrusted` +approvals and its asking feature, whose app server is started again between two turns for +it, and Kimi Code's and ZCode's asking rung. + +What a hook answers is done where the CLI waits for it. A PRE_TOOL_USE hook refuses a tool on +a CLI that gates its tools (Claude Code, Qwen Code) and watches one already reached for on +the rest; a PERMISSION_REQUEST refusal is the tool refused, its reason told to the agent -- on +Codex, whose refusals carry none, as a steer into the turn; SUBAGENT_START and SUBAGENT_STOP +are told, no CLI waiting on either, so what their hooks answer changes nothing. + +A prompt that is `/goal ` is the harness's own goal, on a harness that has one; every +other prompt, `/loop` included, goes to the CLI as it is. A turn's +:class:`~hmz.runtime.flowing.spi.Limits` are held by the driver: what it spends is read as the +CLI reports it, priced with :func:`hmz.coganchor.prices.cost` -- a model nobody prices costs +nothing there. A graceful turn runs to its end and answers as usual, whatever it spends, the +engine refusing the next; one that is not graceful is cut off the moment a limit is reached -- +the CLI stops spending -- and raises the :class:`~hmz.flows.errors.BudgetExceeded` leaf. + +A fork is the CLI's own: Claude Code, Codex, Kimi Code and ZCode fork into another workdir, +every other harness that forks does so only into the workdir it is in, and cursor-agent, +Antigravity and dsh do not fork. A fork is cut where its first turn is taken, so it is refused +then if the session it came from has taken a turn since. + +:func:`open_outworlder` is the driver for whoever is outside the run. """ from __future__ import annotations -from typing import TYPE_CHECKING +import asyncio +import contextlib +import dataclasses +import datetime +import os +import threading +import time +from typing import TYPE_CHECKING, Any, Final + +from hmz.flows import ( + AskUserHookAgentMixin, + CostExceeded, + DurationExceeded, + GoalCommandAgentMixin, + HarnessKind, + HarnessNotInstalled, + HarnessUnrecoverable, + HookKind, + OutputTokensExceeded, + OutworlderAway, + PermissionRequestHookAgentMixin, + SessionError, + SessionStartHookResult, + SteeringAgentMixin, + StopHookResult, + SubagentStartHookAgentMixin, + UnsupportedOperation, + Usage, + UserPromptSubmitHookResult, +) + +from . import harnessing +from .spi import HARNESS_CAPABILITIES, HookBridge if TYPE_CHECKING: + from collections.abc import Callable, Mapping + + import pydantic + + from hmz.coganchor.agents import ( + AgentBase, + AgentConfig, + Event, + Hung, + Occasion, + Question, + SessionBase, + Verdict, + ) + from hmz.coganchor.backends import Profile + from hmz.coganchor.machines import MachineConfig + from hmz.flows import BudgetExceeded, HookResult, Permission + from .specs import AgentSpec - from .spi import AgentDriver + from .spi import ( + HookTable, + Limits, + OutworlderDriver, + Placement, + SessionHandle, + Skill, + TurnRequest, + UsageSink, + ) -__all__ = ["open_agent"] +__all__ = [ + "HarnessDriver", + "HarnessSession", + "HumanOutworlder", + "Listener", + "open_agent", + "open_outworlder", +] +#: What is told everything a session's CLI says, as coganchor says it: the agent, the +#: conversation, and the event. How a way in shows a run. +type Listener = Callable[[AgentBase, SessionBase | None, Event], None] -def open_agent(spec: AgentSpec) -> AgentDriver: +#: The prefix of a prompt that is the harness's own goal. +_GOAL: Final = "/goal " + +#: How often a turn's spending is read while nothing it says has been heard. +_POLL: Final = 1.0 + +#: How often a turn that was cut off is cut again while it has not ended. +_RECUT: Final = 1.0 + +#: How long a cancelled turn is waited for to end before `CancelledError` leaves it. +_DRAIN: Final = 30.0 + +#: The CLIs whose answer to a permission request carries no reason, which a refusal's reason +#: is steered into the turn for instead: Codex's approvals are a decision and nothing else. +_REASONLESS: Final = frozenset({HarnessKind.CODEX}) + +#: The moments whose hook decides whether a tool runs. +_GATES: Final = frozenset({HookKind.PRE_TOOL_USE, HookKind.PERMISSION_REQUEST}) + +#: The hooks whose being hung decides how a session's CLI is started or its rung. +_STARTING: Final = frozenset( + {HookKind.PRE_TOOL_USE, HookKind.PERMISSION_REQUEST, HookKind.ASK_USER} +) + +#: How long a moment from inside a CLI -- a tool, a permission, a notification, a subagent -- +#: waits on the flow's hook before it is answered as if nothing were hung: as long as a CLI's +#: own hook table waits on one, so that a hook that never answers cannot hold a turn for +#: good. A question to the user waits for as long as the loop runs; a person is its answer. +HOOK_TIMEOUT: float = 900.0 + + +def open_agent(spec: AgentSpec) -> HarnessDriver: """Makes the driver for one `-a`. - Starts nothing: the CLI is reached when the first session opens. + Starts nothing and looks for nothing: the CLI is reached when the first session opens, + which is also where one not installed where the session works is refused -- this does not + know yet which machine that is. Args: spec: The agent. @@ -28,6 +175,1225 @@ def open_agent(spec: AgentSpec) -> AgentDriver: :data:`~hmz.runtime.flowing.spi.HARNESS_CAPABILITIES`. Raises: - HarnessNotInstalled: If the CLI is known not to be installed here. + HarnessNotInstalled: For a CLI added by hand that nobody has added on this machine. + HarnessUnrecoverable: For a model, effort or account the CLI cannot be configured at. """ - raise NotImplementedError("the agent drivers are not written yet") + from hmz.coganchor import backends + from hmz.coganchor.agents import driver + from hmz.coganchor.agents.base import identifying + + try: + kind, made = driver(spec.cli) + except KeyError: + raise HarnessNotInstalled( + f"{spec.cli}: no such CLI has been added on this machine" + ) from None + profile = backends.named(spec.cli) + try: + config = made( + model=spec.model, + effort=spec.effort, + provider=spec.provider, + **identifying(made, spec.cli), + ) + # Built once and let go, which is where coganchor refuses an effort, a rung or a + # tier the CLI has no word for -- said here rather than at the first session. + kind(config) + except ValueError as refused: + raise HarnessUnrecoverable(f"{spec}: {refused}") from refused + return HarnessDriver(spec, kind, config, profile) + + +class HarnessDriver: + """The driver of one CLI at one account, model and effort; see the module docstring.""" + + __slots__ = ( + "_capabilities", + "_closed", + "_config", + "_installed", + "_kind", + "_listeners", + "_profile", + "_sessions", + "_spec", + ) + + def __init__( + self, + spec: AgentSpec, + kind: type[AgentBase], + config: AgentConfig, + profile: Profile | None, + ) -> None: + """Initializes a driver that has opened nothing. + + Args: + spec: The agent, as `-a` named it. + kind: coganchor's agent class for the CLI. + config: What every session's agent is configured with before its own settings. + profile: What coganchor knows of the CLI, or None for one it knows nothing of. + """ + self._spec = spec + self._kind = kind + self._config = config + self._profile = profile + self._capabilities = HARNESS_CAPABILITIES[spec.harness] + self._listeners: list[Listener] = [] + self._sessions: set[HarnessSession] = set() + self._closed = False + self._installed = False + + @property + def harness(self) -> HarnessKind: + """Which CLI it is.""" + return self._spec.harness + + @property + def model(self) -> str: + """The model its sessions run.""" + return self._spec.model + + @property + def effort(self) -> str: + """The effort, in the harness's own words; "" for the harness's default.""" + return self._spec.effort + + @property + def provider(self) -> str: + """The account turns run as, or "" for whoever this machine's CLI is logged in as.""" + return self._spec.provider + + @property + def capabilities(self) -> frozenset[type]: + """The agent mixins of :mod:`hmz.flows` this driver serves.""" + return self._capabilities + + @property + def spec(self) -> AgentSpec: + """The agent, as `-a` named it.""" + return self._spec + + @property + def tellable(self) -> bool: + """Whether the CLI can be told whether its agent may use the web.""" + return self._profile is not None and self._profile.searches + + def watch(self, listener: Listener) -> None: + """Has everything the CLI says in every session opened from now on reach `listener`. + + coganchor stops writing a watched agent's words to this process's own stdout and + stderr, and every session here is watched for what it spends -- so a way in that + shows a run shows it through this. + + Args: + listener: What to tell, from whichever thread the CLI is read on. + """ + self._listeners.append(listener) + + async def open( + self, + placement: Placement, + *, + permission: Permission, + skills: tuple[Skill, ...], + hooks: HookTable, + fork_of: SessionHandle | None = None, + ) -> HarnessSession: + """Opens a session, which starts no CLI until its first turn. + + Args: + placement: Where it works. + permission: What it may touch, as :mod:`hmz.runtime.flowing.harnessing` maps it. + skills: What skills it is given, mounted where its CLI reads them. + hooks: What is hung on the agent. + fork_of: A session of this driver to carry on from, or None for a fresh one. + + Returns: + The session. + + Raises: + SessionError: If the driver is closed, `fork_of` is not an open session of it that + has taken a turn, or the workdir is not a directory here. + UnsupportedOperation: If the harness cannot fork, or not into `placement`. + HarnessNotInstalled: If the CLI is not installed on this machine and the session + works here. + HarnessUnrecoverable: If the CLI cannot be configured as the session asks. + """ + if self._closed: + raise SessionError("the agent's driver is closed") + parent = self._parent(fork_of, placement) + cwd = self._where(placement) + if placement.machine is None: + self._check_installed() + hung = frozenset(kind for kind in _STARTING if kind in hooks) + config = self._configured(permission, placement, hung) + agent, session = await asyncio.to_thread( + self._made, config, skills, parent, cwd + ) + handle = HarnessSession( + self, + agent, + session, + hooks=hooks, + placement=placement, + permission=permission, + bridge=HookBridge.here(timeout=HOOK_TIMEOUT), + ) + if self._closed: + # Closed while the session was being made, so it is closed with the rest. + await handle.close() + raise SessionError("the agent's driver is closed") + self._sessions.add(handle) + return handle + + def _parent( + self, fork_of: SessionHandle | None, placement: Placement + ) -> HarnessSession | None: + """The session a fork is cut from, checked against where the fork is to work. + + Raises: + SessionError: If it is not an open session of this driver that has taken a turn. + UnsupportedOperation: If the harness cannot fork there. + """ + if fork_of is None: + return None + if not isinstance(fork_of, HarnessSession) or fork_of.driver is not self: + raise SessionError( + "a session is forked only by the agent it is a session of" + ) + if fork_of.closed: + raise SessionError("the session to fork is closed") + if fork_of.id is None: + raise SessionError("the session to fork has taken no turn to carry on from") + if self._profile is None or not self._profile.forks: + raise UnsupportedOperation(f"{self.harness} cannot fork a session") + was = fork_of.placement + if was.machine != placement.machine: + raise UnsupportedOperation( + f"{self.harness} cannot fork a session onto another machine" + ) + if ( + was.workdir != placement.workdir + and not type(fork_of.coganchor).forks_elsewhere + ): + raise UnsupportedOperation( + f"{self.harness} cannot fork a session into another workdir" + ) + return fork_of + + @staticmethod + def _where(placement: Placement) -> str | None: + """The directory a session's conversation is opened at, as coganchor takes it. + + Raises: + SessionError: If it is on this machine and is not a directory. + """ + workdir = str(placement.workdir) + if placement.machine is not None: + # A remote workdir under the login's home is the anchor's own workspace, which is + # where a session opened at no directory works. + return workdir if placement.workdir.is_absolute() else None + where = os.path.expanduser(workdir) # noqa: PTH111 -- a string path, as coganchor takes + if not os.path.isdir(where): # noqa: PTH112 + raise SessionError(f"{where} is not a directory to open a session in") + return where + + def _check_installed(self) -> None: + """Refuses a CLI that is not installed on this machine, looked for once. + + Raises: + HarnessNotInstalled: If it is not. + """ + if self._installed: + return + from hmz.coganchor import backends + + command = self._profile.runs() if self._profile is not None else self._spec.cli + if backends.program(command) is None: + raise HarnessNotInstalled( + f"{command} is not installed here: {backends.installing(self._spec.cli)}" + ) + self._installed = True + + def _configured( + self, permission: Permission, placement: Placement, hung: frozenset[HookKind] + ) -> AgentConfig: + """What a session's own agent is configured with. + + Raises: + HarnessUnrecoverable: If the CLI cannot be configured so. + """ + return settled( + self._config, + self.harness, + self._kind, + permission, + hung, + tellable=self.tellable, + machine=placement.machine, + ) + + def _made( + self, + config: AgentConfig, + skills: tuple[Skill, ...], + parent: HarnessSession | None, + cwd: str | None, + ) -> tuple[AgentBase, SessionBase]: + """Builds a session's own agent and its conversation, on a thread of its own. + + Raises: + HarnessUnrecoverable: If the agent cannot be built as configured. + UnsupportedOperation: If the CLI will not fork into `cwd`. + SessionError: If the session to fork cannot be carried on from. + """ + from hmz.coganchor.agents.skills import Loaded + + try: + agent = self._kind(config) + except ValueError as refused: + raise HarnessUnrecoverable(f"{self._spec}: {refused}") from refused + agent.loads(Loaded(name=one.name, at=one.at) for one in skills) + for listener in self._listeners: + agent.watch(listener) + if parent is None: + return agent, agent.new(cwd) + try: + return agent, parent.coganchor.fork(into=agent, cwd=cwd) + except NotImplementedError as refused: + raise UnsupportedOperation(str(refused)) from refused + except (RuntimeError, ValueError, OSError) as refused: + raise SessionError(f"the session cannot be forked: {refused}") from refused + + def forget(self, session: HarnessSession) -> None: + """Drops a session that has closed from those the driver closes.""" + self._sessions.discard(session) + + async def close(self) -> None: + """Closes every session still open, and opens no more. Idempotent.""" + self._closed = True + sessions, self._sessions = list(self._sessions), set() + await asyncio.gather(*(one.close() for one in sessions)) + + +def settled( + config: AgentConfig, + harness: HarnessKind, + kind: type[AgentBase], + permission: Permission, + hung: frozenset[HookKind], + *, + tellable: bool, + machine: MachineConfig | None = None, +) -> AgentConfig: + """A driver's config, set up for one session given the hooks hung on it now. + + Args: + config: The driver's own config. + harness: The CLI. + kind: coganchor's agent class for it, whose rungs the rung is one of. + permission: What the session may touch. + hung: The hooks among :data:`_STARTING` hung on the agent. + tellable: Whether the CLI can be told about the web. + machine: The machine its turns land on, or None for this one. + + Returns: + The config. + + Raises: + HarnessUnrecoverable: If the CLI cannot be configured so. + """ + from hmz.coganchor.agents import CodexAgentConfig + + changes: dict[str, Any] = { + "permission": harnessing.rung(harness, permission, rungs=kind.rungs, hung=hung), + "web_search": harnessing.searching(permission, tellable=tellable), + "machine": machine, + } + if isinstance(config, CodexAgentConfig): + feature = harnessing.ASKING_FEATURE + rest = tuple(one for one in config.features if one[0] != feature) + asks = ((feature, True),) if HookKind.ASK_USER in hung else () + changes["features"] = rest + asks + changes["approvals"] = harnessing.approvals(harness, hung) + try: + return dataclasses.replace(config, **changes) + except ValueError as refused: + raise HarnessUnrecoverable(str(refused)) from refused + + +class HarnessSession: + """One session of a :class:`HarnessDriver`: one conversation with its CLI. + + Every method may be called from the engine's loop, and :meth:`interrupt` from any thread. + """ + + __slots__ = ( + "_agent", + "_asking", + "_bridge", + "_closed", + "_cost", + "_driver", + "_duration", + "_flying", + "_hooks", + "_hung", + "_interrupted", + "_last", + "_limits", + "_lock", + "_loop", + "_over", + "_permission", + "_placement", + "_pre", + "_returned", + "_rose", + "_running", + "_seen", + "_session", + "_sink", + "_started", + "_steer_cut", + "_steers", + "_take", + "_timers", + "_tokens", + "_turn_cost", + "_turn_tokens", + "_written", + ) + + def __init__( + self, + driver: HarnessDriver, + agent: AgentBase, + session: SessionBase, + *, + hooks: HookTable, + placement: Placement, + permission: Permission, + bridge: HookBridge, + ) -> None: + """Initializes a session that has taken no turn. + + Args: + driver: The driver it is a session of. + agent: Its own coganchor agent. + session: Its conversation. + hooks: What is hung on the flow agent it belongs to. + placement: Where it works. + permission: What it may touch. + bridge: What carries moments from the CLI's threads to the engine's loop. + """ + self._driver = driver + self._loop = asyncio.get_running_loop() + self._agent = agent + self._session = session + self._hooks = hooks + self._placement = placement + self._permission = permission + self._bridge = bridge + # A question waits on a person, for as long as the loop runs. + self._asking = HookBridge(self._loop, timeout=None) + self._lock = threading.Lock() + # What the session has spent, and what of the CLI's own count has been read. + self._cost = 0.0 + self._tokens = 0 + self._duration = 0.0 + self._written = 0.0 + self._seen: dict[str, float] = {} + #: What the meter rose by in the coganchor turn now running, against which what its + #: answer says it spent is read. + self._rose: dict[str, float] = {} + # The turn in flight, if one is. + self._flying = False + self._sink: UsageSink | None = None + self._limits: Limits | None = None + self._last = 0.0 + self._turn_cost = 0.0 + self._turn_tokens = 0 + self._over: type[BudgetExceeded] | None = None + self._timers: list[asyncio.TimerHandle] = [] + self._interrupted = False + self._steers: list[str] = [] + self._steer_cut = False + # The coganchor call the turn is taking, and whether it is running. + self._take = 0 + self._running = False + self._returned = threading.Event() + self._started = False + self._closed = False + self._pre: Hung | None = None + self._hung = frozenset(kind for kind in _STARTING if kind in hooks) + self._listen() + + # ---------------------------------------------------------------------- what it is + + @property + def id(self) -> str | None: + """The CLI's own id for the conversation, or None before it has said one.""" + return self._session.named + + @property + def usage(self) -> Usage: + """Everything this session's turns have spent, up to the moment it is read.""" + with self._lock: + return Usage( + duration=datetime.timedelta(seconds=self._duration), + cost=self._cost, + output_tokens=self._tokens, + ) + + @property + def driver(self) -> HarnessDriver: + """The driver it is a session of.""" + return self._driver + + @property + def placement(self) -> Placement: + """Where it works.""" + return self._placement + + @property + def closed(self) -> bool: + """Whether it has been closed.""" + return self._closed + + @property + def coganchor(self) -> SessionBase: + """Its conversation, as coganchor holds it.""" + return self._session + + @property + def agent(self) -> AgentBase: + """Its own coganchor agent.""" + return self._agent + + # ------------------------------------------------------------------------- moments + + def _listen(self) -> None: + """Hangs what reaches the engine's hooks on the moments the CLI fires itself.""" + capabilities = self._driver.capabilities + allowed = { + HookKind.NOTIFICATION, + *( + (HookKind.PERMISSION_REQUEST,) + if PermissionRequestHookAgentMixin in capabilities + else () + ), + *( + (HookKind.SUBAGENT_START, HookKind.SUBAGENT_STOP) + if SubagentStartHookAgentMixin in capabilities + else () + ), + } + hooks = self._agent.hooks + for moment in hooks.moments: + kind = harnessing.MOMENTS.get(moment.value) + if kind in allowed: + hooks.on(moment, self._moment) + if AskUserHookAgentMixin in capabilities: + self._agent.ask = self._ask + self._agent.watch(self._heard) + self._settle_pre_tool_use() + + def _settle_pre_tool_use(self) -> None: + """Hangs PRE_TOOL_USE on the CLI only while a hook is hung on it. + + A CLI that gates its tools with a hook table of its own is started with one only + while something is hung there, since the table is a program it runs before every tool. + """ + from hmz.coganchor.agents import Moment + + wanted = HookKind.PRE_TOOL_USE in self._hooks + if wanted and self._pre is None: + self._pre = self._agent.hooks.on(Moment.PRE_TOOL_USE, self._moment) + elif not wanted and self._pre is not None: + self._pre.off() + self._pre = None + + def _moment(self, occasion: Occasion) -> Verdict | None: + """Carries a moment the CLI fired to the engine's hook for it, on the CLI's thread.""" + kind = harnessing.MOMENTS[occasion.moment.value] + if kind not in self._hooks: + return None + fields = harnessing.fields(kind, occasion) + result = self._bridge.call( + lambda: self._hooks.fire(kind, self, **fields), + default=_default(kind), + ) + said = harnessing.verdict(kind, result) + if said is None and kind in _GATES and (self._closed or self._cut_off()): + # The hook was given up on because the turn is being cut off: its tool is not + # to run meanwhile, whatever the hook would have said. + from hmz.coganchor.agents import Verdict + + return Verdict(refused=True, because="the turn was cut off") + if ( + said is not None + and kind is HookKind.PERMISSION_REQUEST + and self._driver.harness in _REASONLESS + ): + # The CLI takes a refusal with no reason, so the reason is put into the turn. + with contextlib.suppress(Exception): + self._session.interject( + f"Using {occasion.tool or 'that tool'} was refused: {said.because}" + ) + return said + + def _ask(self, question: Question) -> str | None: + """Carries a question the agent stopped to ask to the engine's ASK_USER hook.""" + if HookKind.ASK_USER not in self._hooks: + return None + result = self._asking.call( + lambda: self._hooks.fire( + HookKind.ASK_USER, + self, + question=question.text, + options=question.options, + ), + default=_default(HookKind.ASK_USER), + ) + return harnessing.answer(result) + + async def _fire(self, kind: HookKind, **fields: Any) -> HookResult | None: + """Fires a moment the driver brackets a turn with, on the loop, where one is hung.""" + if kind not in self._hooks: + return None + return await self._hooks.fire(kind, self, **fields) + + # -------------------------------------------------------------------------- a turn + + async def turn( + self, request: TurnRequest, sink: UsageSink + ) -> str | pydantic.BaseModel: + """Takes one turn; see :meth:`hmz.runtime.flowing.spi.SessionHandle.turn`.""" + self._check_open() + with self._lock: + if self._flying: + raise SessionError("a turn is already in flight in this session") + limits = request.limits + _refuse_spent(limits) + goal = ( + request.prompt.startswith(_GOAL) + and GoalCommandAgentMixin in self._driver.capabilities + ) + body = request.prompt.removeprefix(_GOAL) if goal else request.prompt + self._begin(sink, limits) + try: + if not self._started: + self._started = True + started = await self._fire(HookKind.SESSION_START) + if isinstance(started, SessionStartHookResult) and started.context: + body = f"{started.context}\n\n{body}" + submitted = await self._fire( + HookKind.USER_PROMPT_SUBMIT, prompt=request.prompt + ) + if isinstance(submitted, UserPromptSubmitHookResult): + if submitted.block: + raise SessionError(submitted.reason or "a hook refused the prompt") + if submitted.context: + body = f"{body}\n\n{submitted.context}" + again = 0 + while True: + said = await self._taken(body, request.output_schema, goal=goal) + if self._over is not None: + break + if self._interrupted: + raise SessionError("the turn was interrupted") + if not self._steers: + stopping = await self._fire(HookKind.STOP, said=said, again=again) + if ( + isinstance(stopping, StopHookResult) + and stopping.block + and stopping.reason + ): + body, goal, again = stopping.reason, False, again + 1 + continue + with self._lock: + steered, self._steers = self._steers, [] + if steered: + body, goal = "\n\n".join(steered), False + continue + break + finally: + self._end() + if self._over is not None and not limits.graceful: + raise self._exceeded(self._over, limits) + if self._interrupted: + # Interrupted between two of the CLI's turns -- while a STOP hook was deciding -- + # which is the turn interrupted all the same. + raise SessionError("the turn was interrupted") + if request.output_schema is not None: + return harnessing.read_shape(said, request.output_schema) + return said + + def _begin(self, sink: UsageSink, limits: Limits) -> None: + """Starts counting a turn: its sink, its limits and its clocks.""" + with self._lock: + self._flying = True + self._sink = sink + self._limits = limits + self._last = time.monotonic() + self._turn_cost = 0.0 + self._turn_tokens = 0 + self._over = None + self._interrupted = False + self._steers = [] + self._steer_cut = False + loop = self._loop + if limits.deadline is not None and not limits.graceful: + self._timers.append( + loop.call_at( + loop.time() + (limits.deadline - time.monotonic()), self._deadline + ) + ) + self._timers.append(loop.call_later(_POLL, self._poll)) + + def _end(self) -> None: + """Stops counting a turn: its clocks, and the last of what it spent.""" + for timer in self._timers: + timer.cancel() + self._timers = [] + self._account() + with self._lock: + sink, self._sink = self._sink, None + now = time.monotonic() + duration = now - self._last + self._last = now + self._duration += duration + self._flying = False + if sink is not None: + sink.add(cost=0.0, output_tokens=0, duration=duration) + + async def _taken( + self, body: str, schema: type[pydantic.BaseModel] | None, *, goal: bool + ) -> str: + """Takes one coganchor turn or goal on a thread of its own, and answers with its text. + + Returns: + What the agent said, or "" for a turn cut off by an interrupt, a limit or a steer. + + Raises: + SessionError: If the session was interrupted before the turn began. + HarnessError: The leaf for why the CLI could not take the turn. + asyncio.CancelledError: If the awaiting task was cancelled; the CLI has stopped. + """ + loop = self._loop + landed: asyncio.Future[str] = loop.create_future() + # Retrieved whatever happens to it, so that a turn given up on leaves no warning. + landed.add_done_callback(_retrieved) + settle = self._settling() + with self._lock: + if self._interrupted: + raise SessionError("the turn was interrupted") + if self._over is not None: + return "" + self._take += 1 + take = self._take + self._steer_cut = False + returned = self._returned = threading.Event() + steered, self._steers = self._steers, [] + if steered: + body = "\n\n".join([body, *steered]) + threading.Thread( + target=self._call, + args=(loop, landed, take, returned, settle, body, schema, goal), + name=f"{self._agent.id}-turn", + daemon=True, + ).start() + try: + return await asyncio.shield(landed) + except asyncio.CancelledError: + self.interrupt() + await asyncio.wait({landed}, timeout=_DRAIN) + raise + except Exception as error: + if self._cut_off(): + return "" + leaf = harnessing.harness_error(error, self._agent.backend) + if leaf is None: + raise + raise leaf from error + + def _call( + self, + loop: asyncio.AbstractEventLoop, + landed: asyncio.Future[str], + take: int, + returned: threading.Event, + settle: Callable[[], None] | None, + body: str, + schema: type[pydantic.BaseModel] | None, + goal: bool, # noqa: FBT001 -- positional, as a thread's arguments are + ) -> None: + """The coganchor call one turn is, on the thread it runs on.""" + try: + said = self._called(take, settle, body, schema, goal=goal) + except BaseException as why: # noqa: BLE001 -- carried to the loop, not handled + _posted(loop, _failed, landed, why) + else: + _posted(loop, _landed, landed, said) + finally: + with self._lock: + if self._take == take: + self._running = False + returned.set() + + def _called( + self, + take: int, + settle: Callable[[], None] | None, + body: str, + schema: type[pydantic.BaseModel] | None, + *, + goal: bool, + ) -> str: + """Takes the turn, unless it was cut off before it could begin.""" + if settle is not None: + settle() + with self._lock: + if self._take != take or self._cutting(): + return "" + self._running = True + self._rose = {} + if goal: + return self._session.pursue(body) + said = "" + for event in self._session.stream(body, schema=schema): + if event.kind == "result": + said = event.text + return said.strip() + + def _settling(self) -> Callable[[], None] | None: + """What to change about the CLI before the next turn, given the hooks hung now. + + Returns: + What sets it up again, run on the turn's thread, or None where nothing changes. + """ + self._settle_pre_tool_use() + hung = frozenset(kind for kind in _STARTING if kind in self._hooks) + if hung == self._hung: + return None + driver = self._driver + was = self._agent.config + now = settled( + was, + driver.harness, + type(self._agent), + self._permission, + hung, + tellable=driver.tellable, + machine=self._placement.machine, + ) + if now == was: + self._hung = hung + return None + restart = getattr(now, "features", ()) != getattr(was, "features", ()) + agent, session = self._agent, self._session + + def settle() -> None: + if restart: + # The feature is an argument of the server, which is put down to start again. + # Between two turns of this session, whose agent holds no other: nothing is + # in flight on the server, and the next turn picks the conversation back up + # on the one it starts. + session.cut(why="started again to be asked what it asks") + agent.reconfigure(now) + # Written down only once it has been done, so that a turn which never got as far + # as this leaves the next one to do it. + self._hung = hung + + return settle + + # --------------------------------------------------------------------- what it spends + + def _heard( + self, agent: AgentBase, session: SessionBase | None, event: Event + ) -> None: + """Reads what the turn has spent whenever the CLI says anything.""" + del agent, session + self._account(event.spent if event.kind == "result" else None) + + def _poll(self) -> None: + """Reads what the turn has spent, and again in a while, while one is in flight.""" + if not self._flying: + return + self._account() + self._timers.append(self._loop.call_later(_POLL, self._poll)) + + def _account(self, answered: Mapping[str, float] | None = None) -> None: + """Reports what the CLI has said was spent since the last time, and checks limits. + + Read off the conversation's meter, which most of coganchor's drivers feed as each + request lands, and off the answer a turn ends on, which states the whole turn: what + the answer states beyond what the meter rose by in that turn is counted then, which + is all of it for a driver that feeds no meter and nothing for one that feeds it all. + + Args: + answered: What the answer a turn just ended on says it spent, or None. + """ + from hmz.coganchor import prices + from hmz.coganchor.agents import Usage as Tokens + + spent = self._session.spent() + with self._lock: + sink, limits = self._sink, self._limits + if sink is None or limits is None: + return + rose = { + kind: tokens - self._seen.get(kind, 0.0) + for kind, tokens in spent.items() + if tokens > self._seen.get(kind, 0.0) + } + if rose: + self._seen = dict(spent) + for kind, tokens in rose.items(): + self._rose[kind] = self._rose.get(kind, 0.0) + tokens + if answered is not None: + for kind, tokens in answered.items(): + if (beyond := tokens - self._rose.get(kind, 0.0)) > 0: + rose[kind] = rose.get(kind, 0.0) + beyond + self._rose = {} + if not rose: + return + cost = prices.cost(Tokens(rose), self._agent.config.model) or 0.0 + written = ( + self._written + rose.get("output", 0.0) + rose.get("reasoning", 0.0) + ) + tokens = int(written) - int(self._written) + self._written = written + now = time.monotonic() + duration = now - self._last + self._last = now + self._cost += cost + self._tokens += tokens + self._duration += duration + self._turn_cost += cost + self._turn_tokens += tokens + over: type[BudgetExceeded] | None = None + if limits.graceful: + pass # the turn runs to its end, and the engine refuses the next one + elif limits.cost is not None and self._turn_cost >= limits.cost: + over = CostExceeded + elif ( + limits.output_tokens is not None + and self._turn_tokens >= limits.output_tokens + ): + over = OutputTokensExceeded + sink.add(cost=cost, output_tokens=tokens, duration=duration) + if over is not None: + self._overrun(over) + + def _deadline(self) -> None: + """The deadline of a turn that is not graceful, on the loop: ends it now.""" + if self._flying: + self._overrun(DurationExceeded) + + def _overrun(self, kind: type[BudgetExceeded]) -> None: + """Ends the turn for a limit it reached, from whichever thread noticed.""" + with self._lock: + if self._over is not None or not self._flying: + return + self._over = kind + self._cut("its budget is spent") + + def _exceeded(self, kind: type[BudgetExceeded], limits: Limits) -> BudgetExceeded: + """The error for a limit the turn reached.""" + if kind is CostExceeded: + return CostExceeded( + f"the turn spent ${self._turn_cost:.4f} of ${limits.cost or 0.0:.4f}" + ) + if kind is OutputTokensExceeded: + return OutputTokensExceeded( + f"the turn wrote {self._turn_tokens} of {limits.output_tokens or 0} " + "output tokens" + ) + return DurationExceeded("the turn ran past its deadline") + + # ------------------------------------------------------------ steering and stopping + + async def steer(self, prompt: str, *, queued: bool) -> None: + """Puts a prompt into the turn in flight; see the SPI's `SessionHandle.steer`.""" + self._check_open() + if SteeringAgentMixin not in self._driver.capabilities: + raise UnsupportedOperation(f"{self._driver.harness} cannot be steered") + with self._lock: + if not self._flying: + raise SessionError("no turn is in flight to steer") + running = self._running + if not (queued and running): + self._steers.append(prompt) + if not queued: + self._steer_cut = True + if not queued: + self._cut("steered") + return + if not running: + return + try: + await asyncio.to_thread(self._session.interject, prompt) + except Exception as missed: + # No turn of the CLI's to put it into after all: it is taken as the next, while + # this turn is still in flight to take one. + with self._lock: + if self._flying: + self._steers.append(prompt) + return + raise SessionError("the turn ended before it could be steered") from missed + + def interrupt(self) -> None: + """Stops the turn in flight, if any, from any thread. Idempotent.""" + with self._lock: + if not self._flying or self._interrupted: + return + self._interrupted = True + self._bridge.abandon() + self._asking.abandon() + self._cut("interrupted") + + def _cutting(self) -> bool: + """Whether the coganchor call now starting is to be cut off. Held with the lock.""" + return self._interrupted or self._over is not None or self._steer_cut + + def _cut_off(self) -> bool: + """Whether the coganchor call that just ended was cut off by the driver.""" + with self._lock: + return self._cutting() + + def _cut(self, why: str) -> None: + """Cuts the coganchor call off, and again until it has ended, from a thread.""" + with self._lock: + if not self._running: + return + returned = self._returned + threading.Thread(target=self._cutter, args=(returned, why), daemon=True).start() + + def _cutter(self, returned: threading.Event, why: str) -> None: + """Cuts one coganchor call off until it has ended, which one cut may not do. + + Held to the call rather than to the turn: a call a cancelled turn stopped waiting for + is cut again for as long as it runs, the turn after it having started behind it. + """ + while True: + with contextlib.suppress(Exception): + self._session.cut(why=why) + if returned.wait(_RECUT): + return + + # ----------------------------------------------------------------------- closing + + def _check_open(self) -> None: + """Refuses a session that has been closed. + + Raises: + SessionError: If it has. + """ + if self._closed: + raise SessionError("the session is closed") + + async def close(self) -> None: + """Ends the session, interrupting a turn in flight. Idempotent.""" + if self._closed: + return + self._closed = True + self._driver.forget(self) + self.interrupt() + if self._started: + with contextlib.suppress(Exception): + await self._fire(HookKind.SESSION_END) + self._bridge.close() + self._asking.close() + await asyncio.to_thread(self._shut) + + def _shut(self) -> None: + """Lets go of the conversation and of the CLI serving it, on a thread.""" + with contextlib.suppress(Exception): + self._session.close() + with contextlib.suppress(Exception): + self._agent.stop() + + +def _default(kind: HookKind) -> HookResult: + """What a moment comes to with no hook answering it.""" + from .spi import default_result + + return default_result(kind) + + +def _refuse_spent(limits: Limits) -> None: + """Refuses a turn given nothing to spend. + + Raises: + BudgetExceeded: The leaf for a limit already reached. + """ + if limits.deadline is not None and limits.deadline <= time.monotonic(): + raise DurationExceeded("the turn's deadline has passed") + if limits.cost is not None and limits.cost <= 0: + raise CostExceeded("the turn has no money left to spend") + if limits.output_tokens is not None and limits.output_tokens <= 0: + raise OutputTokensExceeded("the turn has no output tokens left to write") + + +def _posted( + loop: asyncio.AbstractEventLoop, what: Callable[..., None], *said: Any +) -> None: + """Calls something on the loop from a thread, unless the loop has gone.""" + with contextlib.suppress(RuntimeError): + loop.call_soon_threadsafe(what, *said) + + +def _landed[T](landed: asyncio.Future[T], answered: T) -> None: + """Hands a thread's answer to whoever awaits it, unless nobody does any more.""" + if not landed.done(): + landed.set_result(answered) + + +def _failed(landed: asyncio.Future[Any], why: BaseException) -> None: + """Raises a thread's failure where it is awaited, unless nobody awaits it any more.""" + if not landed.done(): + landed.set_exception(why) + + +def _retrieved(landed: asyncio.Future[Any]) -> None: + """Takes a finished future's exception, so that one nobody awaited says nothing.""" + if not landed.cancelled(): + landed.exception() + + +# ----------------------------------------------------------------------------- outworlder + + +def open_outworlder( + *, + ask: Callable[[Question], str | None] | None = None, + away: Callable[[], bool] | None = None, +) -> HumanOutworlder: + """Makes the driver for whoever is outside the run. + + Args: + ask: Asks the person one question and waits for the answer, on a thread of its own: + the question's text, and the answers it offers where it offers any. Answers None when + nobody answered -- they walked away, or the interface closed. A text turn is one + question; a turn asked for a schema is one question per field, as + :class:`hmz.coganchor.agents.HumanAgent` asks them. None for nobody at all, which is + `hmz exec`. + away: Whether nobody is there now -- `/afk` -- asked whenever it matters. None for + away exactly when there is nobody to `ask`. + + Returns: + The driver. + """ + return HumanOutworlder(ask=ask, away=away) + + +class HumanOutworlder: + """Whoever is outside the run, asked through the person-shaped agent coganchor has. + + Implements :class:`~hmz.runtime.flowing.spi.OutworlderDriver`. + """ + + __slots__ = ("_ask", "_away", "_person") + + def __init__( + self, + *, + ask: Callable[[Question], str | None] | None, + away: Callable[[], bool] | None, + ) -> None: + """Initializes the driver; see :func:`open_outworlder`.""" + from hmz.coganchor.agents import HumanAgent + + self._ask = ask + self._away = away + self._person = HumanAgent(name="outworlder") + self._person.ask = ask + + @property + def away(self) -> bool: + """Whether nobody is there to answer. It may change at any time.""" + if self._ask is None: + return True + return self._away() if self._away is not None else False + + async def run( + self, prompt: str, output_schema: type[pydantic.BaseModel] | None + ) -> str | pydantic.BaseModel: + """Asks, and waits for the answer. + + Returns: + The answer. An away outworlder answers at once: "" for text, and the schema built + with no arguments where every field has a default. + + Raises: + OutworlderAway: If it is away and the schema cannot be answered by default, or went + away -- answered nothing -- while this was waiting. + """ + if self.away: + return away_answer(output_schema) + said = await _threaded(lambda: self._asked(prompt, output_schema)) + if said is None: + raise OutworlderAway("nobody answered") + return said + + def _asked( + self, prompt: str, output_schema: type[pydantic.BaseModel] | None + ) -> str | pydantic.BaseModel | None: + """Puts the prompt to the person, on a thread of its own.""" + from hmz.coganchor.agents import Question + + if output_schema is None: + return self._person.asked(Question(text=prompt)) + return self._person.new()(prompt, suppress=True, schema=output_schema) + + +def away_answer( + output_schema: type[pydantic.BaseModel] | None, +) -> str | pydantic.BaseModel: + """What an outworlder that is away answers. + + Args: + output_schema: The model asked for, or None for text. + + Returns: + "" for text, and the model built with no arguments where every field has a default. + + Raises: + OutworlderAway: For a model some field of which has no default. + """ + if output_schema is None: + return "" + if any(field.is_required() for field in output_schema.model_fields.values()): + raise OutworlderAway( + f"nobody is there to answer a {output_schema.__name__}, and it has no default" + ) + return output_schema() + + +async def _threaded[T](work: Callable[[], T]) -> T: + """Runs blocking work on a thread of its own, and answers with its result.""" + loop = asyncio.get_running_loop() + landed: asyncio.Future[T] = loop.create_future() + landed.add_done_callback(_retrieved) + + def carry() -> None: + try: + answered = work() + except BaseException as why: # noqa: BLE001 -- carried to the loop, not handled + _posted(loop, _failed, landed, why) + else: + _posted(loop, _landed, landed, answered) + + threading.Thread(target=carry, name="outworlder", daemon=True).start() + return await landed + + +if TYPE_CHECKING: + from .spi import AgentDriver + + _: type[AgentDriver] = HarnessDriver + __: type[SessionHandle] = HarnessSession + ___: type[OutworlderDriver] = HumanOutworlder diff --git a/src/hmz/runtime/flowing/harnessing.py b/src/hmz/runtime/flowing/harnessing.py new file mode 100644 index 000000000..d1a520e3b --- /dev/null +++ b/src/hmz/runtime/flowing/harnessing.py @@ -0,0 +1,411 @@ +"""How the agent drivers read the flow API into coganchor's words, and coganchor's back. + +Everything here is a table or a pure function, so that what a harness driver decides can be +read -- and tested -- without starting a CLI: which rung of coganchor's ladder a +:class:`~hmz.flows.Permission` comes to, what a hook's result tells a turn, and which +:class:`~hmz.flows.errors.HarnessError` a failed turn is. The drivers themselves are +:mod:`hmz.runtime.flowing.harnesses`. + +**Permission.** A :class:`~hmz.flows.Permission` is four scopes, and coganchor has one +ladder of rungs -- `read-only`, `workspace-write`, `auto`, `bypass` -- that each of its +drivers maps onto its CLI's own sandbox and approval settings. Approvals are never put to +anybody: every session runs at its CLI's nothing-asked mode, which is what the flow API +calls BYPASS -- `danger-full-access` and `never` on Codex, and on Claude Code, whose +`bypassPermissions` a managed policy may forbid, `manual` with humanize answering every +request yes. `auto` is Claude Code's and cursor-agent's model-reviewed modes and is never +used for them; the one place it is used is the last column below, on the two CLIs where it +means the CLI asks and humanize answers. + +| `local` | every harness but dsh and acp | dsh, acp | +|--------------|-------------------------------|----------| +| READ or NONE | `read-only` | `bypass` | +| ALL | `bypass` | `bypass` | + +- `local` ALL fences nothing else. `user` and `system` are not held to READ or NONE: a + session that may write its workdir may write anywhere its user can. Two of these CLIs + have a sandbox that could fence it -- Codex's `workspace-write` and cursor-agent's + `--sandbox enabled` -- and neither is used, for two reasons. The flow API's own word for + Codex's BYPASS is `danger-full-access` with `never`. And the sandbox is bubblewrap, which + cannot start on a machine that gives it no user namespace -- the one this was written on + answers every command of a sandboxed Codex turn with `bwrap: loopback: Failed + RTM_NEWADDR: Operation not permitted` -- so a fence here would be a flow that loses its + shell wherever it runs in a container. Which harnesses fence is :data:`FENCED`, empty. +- `local` READ is the CLI's own read-only rung: Claude Code's `plan`, Codex's read-only + sandbox, a tool list with nothing that writes on the rest. It reads outside the workdir + too, which is wider than a `user` or `system` of NONE. `local` NONE is the same rung, + which reads the workdir. +- dsh and ACP CLIs can be held to nothing but `bypass`, which is wider than asked for any + permission below ALL. + +While a hook is hung on a moment only asking reaches, three CLIs are started so that they +ask, and humanize answers every request yes unless the hook says no -- never a model: + +- Codex, a PERMISSION_REQUEST hook: the rung's sandbox, and approval policy `untrusted`, so + every command but a known-safe read is asked about. An ASK_USER hook switches on Codex's + `default_mode_request_user_input` feature instead, without which its agent cannot ask its + user anything outside plan mode. +- Kimi Code and ZCode, either hook: coganchor's `auto` -- Kimi's `yolo`, ZCode's `build` -- + which asks about what the CLI deems risky, and on Kimi is the mode where the agent may + ask its user at all. + +`online` is the CLI's own web tools: on for ALL, off for NONE where the CLI can be told, and +left as the CLI has it where it cannot (cursor-agent, pi, agy, acp), which may be wider. A +shell command the agent runs reaches the network whatever this says. +""" + +from __future__ import annotations + +import json +import re +import subprocess +from types import MappingProxyType +from typing import TYPE_CHECKING, Any, Final + +import pydantic + +from hmz.flows import ( + AskUserHookResult, + HarnessContended, + HarnessDropped, + HarnessError, + HarnessKilled, + HarnessKind, + HarnessMissing, + HarnessNotInstalled, + HarnessRefused, + HarnessSandboxed, + HarnessThrottled, + HarnessUnrecoverable, + HookKind, + ModelUnavailable, + OutputSchemaError, + PermissionKind, + PermissionRequestHookResult, + PreToolUseHookResult, + SessionError, +) + +if TYPE_CHECKING: + from collections.abc import Mapping + + from hmz.coganchor.agents import Occasion, Verdict + from hmz.flows import HookResult, Permission + +__all__ = [ + "ASKING", + "ASKING_FEATURE", + "ASKS", + "BYPASS", + "FAULTS", + "FENCED", + "MOMENTS", + "READ_ONLY", + "UNTRUSTED", + "WORKSPACE_WRITE", + "answer", + "approvals", + "asking", + "fields", + "harness_error", + "read_shape", + "rung", + "searching", + "verdict", +] + +READ_ONLY: Final = "read-only" +WORKSPACE_WRITE: Final = "workspace-write" +#: coganchor's `auto`: the CLI asks, and what it asks is granted. Used for exactly the +#: harnesses in :data:`ASKS`, and only while a hook is hung that the asking is for. +ASKING: Final = "auto" +BYPASS: Final = "bypass" + +#: The harnesses whose `workspace-write` would be used to fence writes to the workdir, which +#: is none of them: see the module docstring. Adding a CLI here puts a session whose `local` +#: is ALL and whose `user` is READ or NONE in that CLI's `workspace-write` rung instead. +FENCED: frozenset[HarnessKind] = frozenset() + +#: The harnesses run at coganchor's `auto` while a hook is hung on the moments that come from +#: the asking, each with the moments it is for. +ASKS: Mapping[HarnessKind, frozenset[HookKind]] = MappingProxyType( + { + HarnessKind.KIMI: frozenset({HookKind.PERMISSION_REQUEST, HookKind.ASK_USER}), + HarnessKind.ZCODE: frozenset({HookKind.PERMISSION_REQUEST, HookKind.ASK_USER}), + } +) + +#: Codex's approval policy while a PERMISSION_REQUEST hook is hung: everything but a +#: known-safe read is asked about, over whatever sandbox the rung puts it in. +UNTRUSTED: Final = "untrusted" + +#: The feature Codex's agent needs to ask its user anything outside plan mode. +ASKING_FEATURE: Final = "default_mode_request_user_input" + + +def rung( + harness: HarnessKind, + permission: Permission, + *, + rungs: tuple[str, ...], + hung: frozenset[HookKind] = frozenset(), +) -> str: + """The rung of coganchor's ladder a session is run at, as the table above says. + + Args: + harness: The CLI. + permission: What the session may touch. + rungs: The rungs the CLI's coganchor driver takes. + hung: The hooks hung on the agent now. + + Returns: + One of `read-only`, `workspace-write`, `auto` and `bypass`, and never a rung outside + `rungs`. + """ + if permission.local <= PermissionKind.READ: + said = READ_ONLY + elif harness in FENCED and permission.user <= PermissionKind.READ: + said = WORKSPACE_WRITE + else: + said = BYPASS + if said != READ_ONLY and asking(harness, hung): + said = ASKING + return said if said in rungs else BYPASS + + +def approvals(harness: HarnessKind, hung: frozenset[HookKind]) -> str: + """The approval policy a Codex session is started with over its rung's, or "". + + Args: + harness: The CLI. + hung: The hooks hung on the agent now. + + Returns: + :data:`UNTRUSTED` for Codex while a PERMISSION_REQUEST hook is hung, and "" otherwise. + """ + if harness is HarnessKind.CODEX and HookKind.PERMISSION_REQUEST in hung: + return UNTRUSTED + return "" + + +def asking(harness: HarnessKind, hung: frozenset[HookKind]) -> bool: + """Whether a harness is to be run at its asking rung, given the hooks hung on it. + + Args: + harness: The CLI. + hung: The hooks hung on the agent now. + + Returns: + True where a hook is hung on a moment only that rung reaches. + """ + wanted = ASKS.get(harness) + return wanted is not None and not wanted.isdisjoint(hung) + + +def searching(permission: Permission, *, tellable: bool) -> bool | None: + """Whether a session may use its CLI's web tools. + + Args: + permission: What the session may touch. + tellable: Whether the CLI can be told at all. + + Returns: + True or False where the CLI can be told, and None -- the CLI's own default -- where it + cannot. + """ + if not tellable: + return None + return permission.online == PermissionKind.ALL + + +# ---------------------------------------------------------------------------------- hooks + +#: The moments of a turn that coganchor fires from a thread of the CLI's, each with the hook +#: kind it is. The rest -- SESSION_START, USER_PROMPT_SUBMIT, STOP, SESSION_END -- the drivers +#: fire themselves, on the engine's loop, around the turns they take. +MOMENTS: Mapping[str, HookKind] = MappingProxyType( + { + "PreToolUse": HookKind.PRE_TOOL_USE, + "PermissionRequest": HookKind.PERMISSION_REQUEST, + "Notification": HookKind.NOTIFICATION, + "SubagentStart": HookKind.SUBAGENT_START, + "SubagentStop": HookKind.SUBAGENT_STOP, + } +) + + +def fields(kind: HookKind, occasion: Occasion) -> dict[str, Any]: + """What a hook of one kind is told of a moment coganchor fired. + + Args: + kind: The hook kind the moment is. + occasion: The moment, as coganchor says it. + + Returns: + The params of the kind's :data:`hmz.flows.HOOK_TYPES` entry, less `ctx` and `session`. + A tool read off a CLI's stream rather than asked about carries only the line a + transcript shows of it; that line is the tool's `input` then, under `about`. + """ + match kind: + case HookKind.PRE_TOOL_USE | HookKind.PERMISSION_REQUEST: + given = dict(occasion.input) or ( + {"about": occasion.about} if occasion.about else {} + ) + return {"tool": occasion.tool, "input": given} + case HookKind.NOTIFICATION: + return {"message": occasion.said} + case HookKind.SUBAGENT_START: + return {"subagent": occasion.tool, "task": occasion.about} + case HookKind.SUBAGENT_STOP: + return {"subagent": occasion.tool, "said": occasion.said} + case _: + return {} + + +def verdict(kind: HookKind, result: HookResult) -> Verdict | None: + """What a hook's result tells coganchor to do about the moment it was hung on. + + Only a tool reached for and a tool asking to run can be refused; the other moments + coganchor fires are told, and what their hooks answer changes nothing there. + + Args: + kind: The moment. + result: What the hook answered. + + Returns: + The verdict, or None to let the moment go on as it would have. + """ + from hmz.coganchor.agents import Verdict + + if kind is HookKind.PRE_TOOL_USE and isinstance(result, PreToolUseHookResult): + if result.block: + return Verdict(refused=True, because=result.reason or "refused by a hook") + return None + if ( + kind is HookKind.PERMISSION_REQUEST + and isinstance(result, PermissionRequestHookResult) + and not result.allow + ): + return Verdict(refused=True, because=result.reason or "refused by a hook") + return None + + +def answer(result: HookResult) -> str | None: + """What an ASK_USER hook's result answers the agent with, or None for no answer.""" + return result.answer if isinstance(result, AskUserHookResult) else None + + +# --------------------------------------------------------------------------------- errors + +#: Which harness error each kind of failure coganchor names is. +FAULTS: Mapping[str, type[HarnessError]] = MappingProxyType( + { + "contended": HarnessContended, + "throttled": HarnessThrottled, + "refused": HarnessRefused, + "unlisted": ModelUnavailable, + "retired": ModelUnavailable, + "missing": HarnessMissing, + "sandboxed": HarnessSandboxed, + "killed": HarnessKilled, + "dropped": HarnessDropped, + } +) + +#: What a shell exits with for a command it could not find, which is a CLI not installed. +_NOTHING_TO_RUN = 127 + + +def harness_error(error: BaseException, backend: str) -> HarnessError | None: + """The harness error a turn that failed in coganchor comes to. + + Args: + error: What the turn raised. + backend: The CLI, as coganchor names it, for reading a failure it did not classify. + + Returns: + The leaf for it, carrying the failure's own words, or None for an exception that is + not a turn failing -- a bug, which stays the bug it was. + """ + from hmz.coganchor.agents import Failed, Stopped, Unrecoverable + from hmz.coganchor.anchor import NotInstalled + + said = str(error) or type(error).__name__ + if isinstance(error, NotInstalled): + return HarnessNotInstalled(said) + if isinstance(error, Stopped): + return SessionError(said) + if isinstance(error, subprocess.CalledProcessError): + fault = ( + error.fault if isinstance(error, Failed) else _classified(error, backend) + ) + if fault == "missing" and error.returncode == _NOTHING_TO_RUN: + return HarnessNotInstalled(said) + if isinstance(error, Unrecoverable) and not fault: + return HarnessUnrecoverable(said) + return FAULTS.get(fault, HarnessUnrecoverable)(said) + if isinstance(error, OSError): + return HarnessDropped(said) + if isinstance(error, RuntimeError): + return SessionError(said) + if isinstance(error, ValueError): + return HarnessUnrecoverable(said) + return None + + +def _classified(error: subprocess.CalledProcessError, backend: str) -> str: + """Which kind of failure coganchor reads a failed process it did not classify as.""" + from hmz.coganchor import backends + + return backends.trouble( + backend, + _text(error.stderr), + _text(error.output), + status=error.returncode, + ) + + +def _text(said: object) -> str: + """A stream a failed process wrote, as text.""" + if isinstance(said, bytes): + return said.decode("utf-8", "replace") + return said if isinstance(said, str) else "" + + +# -------------------------------------------------------------------------------- answers + +#: A block of an answer fenced as code, which is where a model that talked as well as +#: answered puts the object it was asked for. +_FENCED_JSON = re.compile(r"```(?:json)?\s*(.*?)```", re.DOTALL) + + +def read_shape[T: pydantic.BaseModel](said: str, schema: type[T]) -> T: + """Reads an answer as the model it was asked for. + + Tried as the whole answer, then as each fenced block in it, then as the span from its + first brace to its last -- a CLI held to a schema answers with the object alone, and one + asked for it in the prompt may talk around it. + + Args: + said: The answer. + schema: The model. + + Returns: + The answer, as that model. + + Raises: + OutputSchemaError: If no part of it is one. + """ + held = said.strip() + readings = [held, *(str(one).strip() for one in _FENCED_JSON.findall(held))] + first, last = held.find("{"), held.rfind("}") + if 0 <= first < last: + readings.append(held[first : last + 1]) + refused: pydantic.ValidationError | None = None + for reading in readings: + try: + return schema.model_validate_json(reading) + except pydantic.ValidationError as why: + refused = refused or why + raise OutputSchemaError( + f"the answer is not a {schema.__name__}: {json.dumps(held[:200])}" + ) from refused diff --git a/tests/flows/standins.py b/tests/flows/standins.py new file mode 100644 index 000000000..7d23bdf46 --- /dev/null +++ b/tests/flows/standins.py @@ -0,0 +1,435 @@ +"""Stand-in CLIs the harness drivers are held to their contract against, off the network. + +Three of them, one per way coganchor drives a CLI: `claude`, one process held open and spoken +to in stream JSON; `codex`, an app server spoken to in JSON-RPC; and `opencode`, one command +per turn. Each prints what its real CLI prints, opens with what that CLI's own parser refuses +(:func:`tests.agents.standins.refusing`), and answers what the checks in +:mod:`tests.flows.contracts` ask as a model would. A few prompts make it do what a model can be +asked to: + +- `Reply with the single word: ` answers the word, and a `JSON object whose "answer" is + "ok"` answers that object. +- A prompt starting `slow` keeps the turn going for half a minute, saying a word every tenth + of a second, and takes a word put in mid-turn as the end of it. +- `use a tool` reaches for `Bash` -- asking permission where the CLI asks -- and answers + `allowed` or `denied: `; `ask me` asks its user `Which way?` and answers what it was + told; `delegate` starts a subagent. +- `fail: ` fails the turn saying the words. + +Each writes one JSON line per call to `.log`: its argv and working directory where it +starts, and every thing said to it. +""" + +from __future__ import annotations + +import os +import sys +from typing import TYPE_CHECKING + +from tests.agents import standins + +if TYPE_CHECKING: + from pathlib import Path + +__all__ = ["CLAUDE", "CODEX", "OPENCODE", "install"] + +#: What every stand-in starts with: its log, its answers, and a reader of its stdin. +_COMMON = r""" +import json, os, pathlib, queue, re, shlex, subprocess, sys, threading, time, uuid + +LOG = pathlib.Path(sys.argv[0] + ".log") + + +def note(entry): + with LOG.open("a") as stream: + stream.write(json.dumps(entry) + "\n") + + +def answered(said, schema=False): + word = re.search(r"single word: (\w+)", said) + if schema or 'JSON object whose "answer"' in said: + return json.dumps({"answer": "ok"}) + if said.startswith("/goal "): + return "goal met: " + said[len("/goal "):] + return word.group(1) if word else said + + +def failing(said): + return said[len("fail: "):] if said.startswith("fail: ") else None +""" + +#: `claude --print --input-format stream-json`, held open for the session. It keeps each +#: conversation where the real one does -- `projects//.jsonl` +#: under `$CLAUDE_CONFIG_DIR` -- and refuses to resume one that is not there. +CLAUDE = ( + _COMMON + + r""" +argv = sys.argv[1:] +flags = {} +at = 0 +while at < len(argv): + word = argv[at] + if word.startswith("--") and at + 1 < len(argv) and not argv[at + 1].startswith("--"): + flags[word] = argv[at + 1] + at += 2 + else: + flags[word] = True + at += 1 +home = pathlib.Path(os.environ.get("CLAUDE_CONFIG_DIR") or pathlib.Path.home() / ".claude") +projects = home / "projects" / re.sub(r"[^a-zA-Z0-9]", "-", os.getcwd()) +model = flags.get("--model", "m") +schema = flags.get("--json-schema") +settings = json.loads(flags.get("--settings") or "{}") +gate = [ + hook["command"] + for table in settings.get("hooks", {}).get("PreToolUse", []) + for hook in table.get("hooks", []) +] + + +def out(said): + print(json.dumps(said), flush=True) + + +if "--resume" in flags: + was = flags["--resume"] + if not (projects / f"{was}.jsonl").exists(): + out({"type": "result", "subtype": "error_during_execution", "is_error": True, + "result": f"No conversation found with session ID: {was}"}) + sys.exit(1) + session = str(uuid.uuid4()) if flags.get("--fork-session") else was + if session != was: + (projects / f"{session}.jsonl").write_text((projects / f"{was}.jsonl").read_text()) +else: + session = flags["--session-id"] +projects.mkdir(parents=True, exist_ok=True) +transcript = projects / f"{session}.jsonl" +note({"argv": argv, "cwd": os.getcwd(), "session": session}) +out({"type": "system", "subtype": "init", "session_id": session}) + +lines = queue.Queue() + + +def reading(): + for line in sys.stdin: + lines.put(json.loads(line)) + lines.put(None) + + +threading.Thread(target=reading, daemon=True).start() +held = [] +spent = {"in": 0, "out": 0} + + +def take(timeout=None): + if held: + return held.pop(0) + try: + return lines.get(timeout=timeout) + except queue.Empty: + return False + + +def speak(text, tokens=3): + spent["in"] += 10 + spent["out"] += tokens + out({"type": "assistant", "message": {"id": str(uuid.uuid4()), "content": [ + {"type": "text", "text": text}], "usage": {"input_tokens": 10, "output_tokens": tokens}}}) + + +def result(text): + out({"type": "result", "subtype": "success", "is_error": False, "session_id": session, + "result": text, "modelUsage": {model: {"inputTokens": spent["in"], + "outputTokens": spent["out"]}}}) + + +def asked(request): + named = str(uuid.uuid4()) + out({"type": "control_request", "request_id": named, "request": request}) + while True: + said = take() + if said is None: + sys.exit(0) + if said.get("type") == "control_response" and said["response"]["request_id"] == named: + return said["response"]["response"] + held.append(said) + + +def gated(tool, called): + for command in gate: + ran = subprocess.run(shlex.split(command), input=json.dumps({ + "hook_event_name": "PreToolUse", "tool_name": tool, "tool_input": called, + "session_id": session}), capture_output=True, text=True, check=False) + told = json.loads(ran.stdout or "{}").get("hookSpecificOutput", {}) + if told.get("permissionDecision") == "deny": + return told.get("permissionDecisionReason", "denied") + return None + + +def turn(said): + with transcript.open("a") as stream: + stream.write(json.dumps({"user": said}) + "\n") + if (why := failing(said)) is not None: + out({"type": "result", "subtype": "error_during_execution", "is_error": True, + "session_id": session, "result": why}) + print(why, file=sys.stderr, flush=True) + sys.exit(1) + if said.startswith("slow"): + for step in range(300): + more = take(0.1) + if more is None: + sys.exit(0) + if more: + words = more["message"]["content"][0]["text"] + out({"type": "command_lifecycle", "state": "started", + "command_uuid": more.get("uuid", "")}) + result("put aside") + speak("heard " + words) + result(answered(words)) + return + speak(f"step {step}", 1) + result("slow done") + return + if said == "use a tool": + called = {"command": "echo hi"} + out({"type": "assistant", "message": {"id": str(uuid.uuid4()), "content": [ + {"type": "tool_use", "id": "toolu_1", "name": "Bash", "input": called}], + "usage": {"input_tokens": 10, "output_tokens": 2}}}) + if (why := gated("Bash", called)) is not None: + result("blocked: " + why) + return + told = asked({"subtype": "can_use_tool", "tool_name": "Bash", "input": called}) + if told.get("behavior") == "allow": + result("allowed") + else: + result("denied: " + told.get("message", "")) + return + if said == "ask me": + told = asked({"subtype": "can_use_tool", "tool_name": "AskUserQuestion", "input": { + "questions": [{"question": "Which way?", "header": "Way", + "options": [{"label": "left"}, {"label": "right"}]}]}}) + answers = told.get("updatedInput", {}).get("answers", {}) + result(answers.get("Which way?", "nobody")) + return + if said == "delegate": + out({"type": "assistant", "message": {"id": str(uuid.uuid4()), "content": [ + {"type": "tool_use", "id": "toolu_2", "name": "Task", + "input": {"description": "look around"}}], + "usage": {"input_tokens": 10, "output_tokens": 2}}}) + out({"type": "user", "message": {"content": [ + {"type": "tool_result", "tool_use_id": "toolu_2", "content": "looked"}]}}) + result("delegated") + return + speak("working") + result(answered(said, schema=bool(schema))) + + +while True: + said = take() + if said is None: + break + if said.get("type") != "user": + continue + text = said["message"]["content"][0]["text"] + note({"said": text, "session": session}) + turn(text) +""" +) + +#: `codex app-server`, spoken to in JSON-RPC. A turn runs on a thread of its own so that a +#: steer, an interrupt and the answers to what it asks can arrive while it does. +CODEX = ( + _COMMON + + r""" +ASKS = "default_mode_request_user_input" +writing = threading.Lock() +answers = {} +arrived = threading.Condition() +turns = {} +goals = {} + + +def send(message): + with writing: + sys.stdout.write(json.dumps(message) + "\n") + sys.stdout.flush() + + +def reply(ident, result): + send({"jsonrpc": "2.0", "id": ident, "result": result}) + + +def request(ident, method, params): + send({"jsonrpc": "2.0", "id": ident, "method": method, "params": params}) + with arrived: + arrived.wait_for(lambda: ident in answers) + return answers.pop(ident) + + +def notify(method, params): + send({"jsonrpc": "2.0", "method": method, "params": params}) + + +TOTALS = pathlib.Path(sys.argv[0] + ".totals.json") + + +def usage(thread, named, step): + # What the thread has spent all told, kept where a server started after this one finds + # it: a real one restores a thread's totals from its rollout when it picks it back up. + with writing: + totals = json.loads(TOTALS.read_text()) if TOTALS.exists() else {} + total = totals[thread] = totals.get(thread, 0) + step + TOTALS.write_text(json.dumps(totals)) + notify("thread/tokenUsage/updated", {"threadId": thread, "turnId": named, "tokenUsage": { + "last": {"inputTokens": 10 * step, "outputTokens": step}, + "total": {"inputTokens": 10 * total, "outputTokens": total}}}) + + +def run(thread, params, named): + said = params["input"][0]["text"] + state = turns[named] + notify("turn/started", {"threadId": thread, "turnId": named}) + if (why := failing(said)) is not None: + notify("turn/completed", {"threadId": thread, "turn": { + "id": named, "status": "failed", "error": {"message": why}}}) + return + if said.startswith("slow"): + for step in range(300): + if state["interrupted"]: + notify("turn/completed", {"threadId": thread, "turn": { + "id": named, "status": "interrupted"}}) + return + if state["steers"]: + words, client = state["steers"].pop(0) + notify("item/started", {"threadId": thread, "turnId": named, "item": { + "type": "userMessage", "clientId": client, "content": words}}) + said = words + break + usage(thread, named, 1) + time.sleep(0.1) + if said == "use a tool" and params.get("approvalPolicy") in (None, "never"): + said = "ran without asking" + elif said == "use a tool": + told = request("ok_" + named, "item/commandExecution/requestApproval", { + "itemId": "i", "threadId": thread, "turnId": named, "command": "echo hi"}) + said = "allowed" if told.get("decision") == "accept" else "denied" + elif said == "ask me" and ASKS not in sys.argv: + said = "nobody" + elif said == "ask me": + told = request("ask_" + named, "item/tool/requestUserInput", { + "itemId": "i", "threadId": thread, "turnId": named, "questions": [ + {"id": "way", "header": "Way", "question": "Which way?", + "options": [{"label": "left"}, {"label": "right"}]}]}) + picked = told.get("answers", {}).get("way", {}).get("answers", []) + said = picked[0] if picked else "nobody" + elif said == "delegate": + notify("item/started", {"threadId": thread, "turnId": named, "item": { + "id": "sub", "type": "collabAgentToolCall", "tool": "look around"}}) + notify("item/completed", {"threadId": thread, "turnId": named, "item": { + "id": "sub", "type": "collabAgentToolCall", "tool": "look around"}}) + said = "delegated" + else: + said = answered(said, schema="outputSchema" in params) + usage(thread, named, 2) + notify("item/completed", {"threadId": thread, "turnId": named, "item": { + "id": "msg", "type": "agentMessage", "text": said}}) + if thread in goals: + notify("thread/goal/updated", {"threadId": thread, "goal": {"status": "complete"}}) + del goals[thread] + notify("turn/completed", {"threadId": thread, "turn": {"id": named, "status": "completed"}}) + notify("thread/status/changed", {"threadId": thread, "status": {"type": "idle"}}) + + +note({"argv": sys.argv[1:], "cwd": os.getcwd()}) +for line in sys.stdin: + call = json.loads(line) + if "method" not in call: + with arrived: + answers[call["id"]] = call.get("result") or {} + arrived.notify_all() + continue + note({"method": call["method"], "params": call.get("params", {})}) + params = call.get("params", {}) + method = call["method"] + if method in ("thread/start", "thread/fork"): + thread = "thread_" + uuid.uuid4().hex[:8] + reply(call["id"], {"thread": {"id": thread}}) + notify("thread/status/changed", {"threadId": thread, "status": {"type": "idle"}}) + elif method == "turn/start": + named = "turn_" + uuid.uuid4().hex[:8] + turns[named] = {"interrupted": False, "steers": []} + reply(call["id"], {"turn": {"id": named}}) + threading.Thread(target=run, args=(params["threadId"], params, named), + daemon=True).start() + elif method == "turn/steer": + state = turns.get(params.get("expectedTurnId"), {"steers": []}) + state["steers"].append((params["input"][0]["text"], + params.get("clientUserMessageId", ""))) + reply(call["id"], {}) + elif method == "turn/interrupt": + turns.get(params.get("turnId"), {})["interrupted"] = True + reply(call["id"], {}) + elif method == "thread/goal/set": + goals[params["threadId"]] = params["objective"] + reply(call["id"], {}) + elif "id" in call: + reply(call["id"], {}) +""" +) + +#: `opencode run`, one process per turn, taking the prompt on stdin and answering in events. +OPENCODE = ( + _COMMON + + r""" +said = sys.stdin.read() +flags = dict(zip(sys.argv, sys.argv[1:])) +forked = "--fork" in sys.argv +session = ("ses_" + uuid.uuid4().hex[:8]) if forked or "--session" not in flags else ( + flags["--session"]) +note({"argv": sys.argv[1:], "cwd": os.getcwd(), "said": said, "session": session}) + + +def out(kind, part): + print(json.dumps({"type": kind, "sessionID": session, "part": part}), flush=True) + + +if (why := failing(said)) is not None: + print(json.dumps({"type": "error", "sessionID": session, + "error": {"name": "APIError", "data": {"message": why}}}), flush=True) + sys.exit(0) +out("step_start", {"id": "prt_start", "type": "step-start"}) +if said.startswith("slow"): + for step in range(300): + out("step_finish", {"id": f"prt_{step}", "type": "step-finish", "reason": "tool-calls", + "tokens": {"total": 2, "input": 1, "output": 1, "reasoning": 0, + "cache": {"read": 0, "write": 0}}}) + time.sleep(0.1) +out("text", {"id": "prt_text", "type": "text", "text": answered(said)}) +out("step_finish", {"id": "prt_two", "type": "step-finish", "reason": "stop", + "tokens": {"total": 7, "input": 5, "output": 2, "reasoning": 0, + "cache": {"read": 0, "write": 0}}}) +""" +) + + +def install(binaries: Path, named: str, script: str) -> Path: + """Puts one stand-in on PATH's first directory under the name its CLI is installed as. + + Args: + binaries: The directory, which the caller puts at the front of PATH. + named: The command. + script: What it runs: :data:`CLAUDE`, :data:`CODEX` or :data:`OPENCODE`. + + Returns: + The log it writes, one JSON object a line. + """ + binaries.mkdir(parents=True, exist_ok=True) + fake = binaries / named + fake.write_text(f"#!{sys.executable}\n{standins.refusing(named)}{script}") + fake.chmod(0o755) + return binaries / f"{named}.log" + + +def path_with(binaries: Path) -> str: + """PATH with a directory of stand-ins in front of it.""" + return f"{binaries}{os.pathsep}{os.environ.get('PATH', '')}" diff --git a/tests/integration/flows/test_harness_drivers.py b/tests/integration/flows/test_harness_drivers.py new file mode 100644 index 000000000..542fd9c3e --- /dev/null +++ b/tests/integration/flows/test_harness_drivers.py @@ -0,0 +1,852 @@ +"""The harness drivers, held to their contract against stand-in CLIs on PATH. + +Every driver is run through :func:`tests.flows.contracts.check_agent_driver` against a stand-in +for Claude Code (a process held open), Codex (an app server) and opencode (a command a turn), +and then through what the contract does not reach: hooks carried from the CLI's threads to the +loop and back, steering, forks across workdirs, a turn's limits, a failed turn's error, and the +outworlder. The stand-ins are :mod:`tests.flows.standins`. +""" + +from __future__ import annotations + +import asyncio +import json +import time +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Any + +import pydantic +import pytest + +from hmz.flows import ( + AskUserHookResult, + CostExceeded, + DurationExceeded, + EnvBackendKind, + HarnessKind, + HarnessNotInstalled, + HarnessRefused, + HookKind, + HookResult, + ModelUnavailable, + NotificationHookResult, + OutputSchemaError, + OutputTokensExceeded, + OutworlderAway, + Permission, + PermissionKind, + PermissionRequestHookResult, + PreToolUseHookResult, + SessionEndHookResult, + SessionError, + SessionStartHookResult, + StopHookResult, + SubagentStartHookResult, + SubagentStopHookResult, + UnsupportedOperation, + UserPromptSubmitHookResult, +) +from hmz.runtime.flowing import harnesses +from hmz.runtime.flowing.harnesses import HarnessSession, open_agent, open_outworlder +from hmz.runtime.flowing.specs import AgentSpec +from hmz.runtime.flowing.spi import HookTable, Limits, Placement, Skill, TurnRequest +from tests.flows import standins +from tests.flows.contracts import RecordingSink, check_agent_driver + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from hmz.coganchor.agents import Question + from hmz.runtime.flowing.harnesses import HarnessDriver + from hmz.runtime.flowing.spi import SessionHandle + +pytestmark = pytest.mark.timeout(120) + +#: What each stand-in is driven as, at a model the price list in `priced` names. +SPECS = { + HarnessKind.CLAUDE: AgentSpec( + "coder", HarnessKind.CLAUDE, "", "claude-haiku-4-5", "low", "claude" + ), + HarnessKind.CODEX: AgentSpec( + "coder", HarnessKind.CODEX, "", "gpt-5.5", "low", "codex" + ), + HarnessKind.OPENCODE: AgentSpec( + "coder", HarnessKind.OPENCODE, "", "opencode/big-pickle", "high", "opencode" + ), +} + + +class Logs: + """Where the stand-ins are, and what they were asked.""" + + def __init__(self, binaries: Path) -> None: + self.binaries = binaries + + def of(self, cli: str) -> list[dict[str, Any]]: + """Everything one stand-in wrote down, oldest first.""" + at = self.binaries / f"{cli}.log" + if not at.exists(): + return [] + return [json.loads(line) for line in at.read_text().splitlines()] + + +@pytest.fixture +def clis(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Logs: + """Puts the three stand-ins on PATH, with a home of their own.""" + binaries = tmp_path / "bin" + standins.install(binaries, "claude", standins.CLAUDE) + standins.install(binaries, "codex", standins.CODEX) + standins.install(binaries, "opencode", standins.OPENCODE) + monkeypatch.setenv("PATH", standins.path_with(binaries)) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path / "claude-home")) + return Logs(binaries) + + +def _placement(where: Path) -> Placement: + where.mkdir(parents=True, exist_ok=True) + return Placement(EnvBackendKind.LOCAL, "", PurePosixPath(str(where))) + + +@pytest.fixture +def work(tmp_path: Path) -> Placement: + """A workdir on this machine.""" + return _placement(tmp_path / "work") + + +def _hears( + table: HookTable, kind: HookKind, answer: HookResult +) -> list[dict[str, Any]]: + """Hangs a hook that answers `answer` and writes down what it was told.""" + heard: list[dict[str, Any]] = [] + + async def bound(handle: SessionHandle, fields: dict[str, Any]) -> HookResult: + del handle + heard.append(fields) + return answer + + table.set(kind, bound) + return heard + + +@pytest.fixture +async def claude(clis: Logs) -> AsyncIterator[HarnessDriver]: + """A driver of the Claude Code stand-in, closed afterwards.""" + del clis + driver = open_agent(SPECS[HarnessKind.CLAUDE]) + yield driver + await driver.close() + + +@pytest.fixture +async def codex(clis: Logs) -> AsyncIterator[HarnessDriver]: + """A driver of the Codex stand-in, closed afterwards.""" + del clis + driver = open_agent(SPECS[HarnessKind.CODEX]) + yield driver + await driver.close() + + +async def _open( + driver: HarnessDriver, + placement: Placement, + hooks: HookTable | None = None, + permission: Permission | None = None, +) -> HarnessSession: + return await driver.open( + placement, + permission=permission or Permission(), + skills=(), + hooks=hooks or HookTable(), + ) + + +# ------------------------------------------------------------------------------ the contract + + +@pytest.mark.parametrize("harness", sorted(SPECS), ids=str) +async def test_every_stand_in_keeps_the_driver_contract( + clis: Logs, work: Placement, harness: HarnessKind +) -> None: + del clis + await check_agent_driver( + open_agent(SPECS[harness]), work, slow_prompt="slow, please", settle=0.5 + ) + + +async def test_a_priced_model_reports_what_it_cost( + claude: HarnessDriver, work: Placement, priced: str +) -> None: + del priced + handle = await _open(claude, work) + sink = RecordingSink() + await handle.turn(TurnRequest("Reply with the single word: ok"), sink) + assert sink.cost > 0 + assert handle.usage.cost == pytest.approx(sink.cost) + assert handle.usage.output_tokens == sink.output_tokens > 0 + + +# --------------------------------------------------------------------------------- the hooks + + +async def test_moments_around_a_turn_reach_their_hooks_on_the_loop( + claude: HarnessDriver, work: Placement, clis: Logs +) -> None: + hooks = HookTable() + started = _hears( + hooks, HookKind.SESSION_START, SessionStartHookResult(context="be brief") + ) + submitted = _hears( + hooks, HookKind.USER_PROMPT_SUBMIT, UserPromptSubmitHookResult(context="thanks") + ) + ended = _hears(hooks, HookKind.SESSION_END, SessionEndHookResult()) + handle = await _open(claude, work, hooks) + said = await handle.turn( + TurnRequest("Reply with the single word: ok"), RecordingSink() + ) + assert said == "ok" + await handle.turn(TurnRequest("Reply with the single word: two"), RecordingSink()) + assert started == [{}] + assert [one["prompt"] for one in submitted] == [ + "Reply with the single word: ok", + "Reply with the single word: two", + ] + told = [one["said"] for one in clis.of("claude") if "said" in one] + assert told[0] == "be brief\n\nReply with the single word: ok\n\nthanks" + assert told[1] == "Reply with the single word: two\n\nthanks" + await handle.close() + assert ended == [{}] + + +async def test_a_blocked_prompt_is_refused_without_a_turn( + claude: HarnessDriver, work: Placement, clis: Logs +) -> None: + hooks = HookTable() + _hears( + hooks, + HookKind.USER_PROMPT_SUBMIT, + UserPromptSubmitHookResult(block=True, reason="not today"), + ) + handle = await _open(claude, work, hooks) + with pytest.raises(SessionError, match="not today"): + await handle.turn(TurnRequest("anything"), RecordingSink()) + assert clis.of("claude") == [] + + +async def test_a_stop_hook_sends_the_agent_on_until_it_lets_it_stop( + claude: HarnessDriver, work: Placement +) -> None: + hooks = HookTable() + stops: list[dict[str, Any]] = [] + + async def keep_going(handle: SessionHandle, fields: dict[str, Any]) -> HookResult: + del handle + stops.append(fields) + if fields["again"] < 2: + return StopHookResult(block=True, reason="Reply with the single word: more") + return StopHookResult() + + hooks.set(HookKind.STOP, keep_going) + handle = await _open(claude, work, hooks) + said = await handle.turn( + TurnRequest("Reply with the single word: first"), RecordingSink() + ) + assert said == "more" + assert [(one["said"], one["again"]) for one in stops] == [ + ("first", 0), + ("more", 1), + ("more", 2), + ] + + +@pytest.mark.parametrize( + ("answer", "said"), + [ + (PermissionRequestHookResult(), "allowed"), + ( + PermissionRequestHookResult(allow=False, reason="no shell"), + "denied: no shell", + ), + ], +) +async def test_claude_asks_a_permission_hook_and_does_what_it_says( + claude: HarnessDriver, + work: Placement, + answer: PermissionRequestHookResult, + said: str, +) -> None: + hooks = HookTable() + heard = _hears(hooks, HookKind.PERMISSION_REQUEST, answer) + handle = await _open(claude, work, hooks) + assert await handle.turn(TurnRequest("use a tool"), RecordingSink()) == said + assert heard == [{"tool": "Bash", "input": {"command": "echo hi"}}] + + +async def test_with_no_permission_hook_every_request_is_approved( + claude: HarnessDriver, work: Placement, clis: Logs +) -> None: + handle = await _open(claude, work) + assert await handle.turn(TurnRequest("use a tool"), RecordingSink()) == "allowed" + argv = clis.of("claude")[0]["argv"] + assert argv[argv.index("--permission-mode") + 1] == "manual" + assert "--permission-prompt-tool" in argv + + +async def test_a_pre_tool_use_hook_gates_the_tool_before_it_runs( + claude: HarnessDriver, work: Placement +) -> None: + hooks = HookTable() + heard = _hears( + hooks, + HookKind.PRE_TOOL_USE, + PreToolUseHookResult(block=True, reason="not that"), + ) + handle = await _open(claude, work, hooks) + assert await handle.turn(TurnRequest("use a tool"), RecordingSink()) == ( + "blocked: not that" + ) + assert heard == [{"tool": "Bash", "input": {"command": "echo hi"}}] + + +async def test_a_pre_tool_use_hook_hung_later_takes_hold_from_the_next_turn( + claude: HarnessDriver, work: Placement +) -> None: + hooks = HookTable() + handle = await _open(claude, work, hooks) + assert await handle.turn(TurnRequest("use a tool"), RecordingSink()) == "allowed" + _hears(hooks, HookKind.PRE_TOOL_USE, PreToolUseHookResult(block=True, reason="no")) + assert ( + await handle.turn(TurnRequest("use a tool"), RecordingSink()) == "blocked: no" + ) + + +async def test_a_question_goes_to_the_ask_user_hook_and_notification( + claude: HarnessDriver, work: Placement +) -> None: + hooks = HookTable() + asked = _hears(hooks, HookKind.ASK_USER, AskUserHookResult(answer="left")) + told = _hears(hooks, HookKind.NOTIFICATION, NotificationHookResult()) + handle = await _open(claude, work, hooks) + assert await handle.turn(TurnRequest("ask me"), RecordingSink()) == "left" + assert asked == [{"question": "Which way?", "options": ("left", "right")}] + assert told == [{"message": "Which way?"}] + + +async def test_with_no_ask_user_hook_the_question_goes_unanswered( + claude: HarnessDriver, work: Placement +) -> None: + handle = await _open(claude, work) + assert await handle.turn(TurnRequest("ask me"), RecordingSink()) == "nobody" + + +async def test_subagents_starting_and_stopping_reach_their_hooks( + claude: HarnessDriver, work: Placement +) -> None: + hooks = HookTable() + starts = _hears(hooks, HookKind.SUBAGENT_START, SubagentStartHookResult()) + stops = _hears(hooks, HookKind.SUBAGENT_STOP, SubagentStopHookResult()) + handle = await _open(claude, work, hooks) + assert await handle.turn(TurnRequest("delegate"), RecordingSink()) == "delegated" + assert [one["subagent"] for one in starts] == ["Task"] + assert [one["subagent"] for one in stops] == ["Task"] + + +async def test_codex_asks_permission_only_while_a_hook_is_hung_on_it( + codex: HarnessDriver, work: Placement, clis: Logs +) -> None: + hooks = HookTable() + handle = await _open(codex, work, hooks) + assert await handle.turn( + TurnRequest("Reply with the single word: a"), RecordingSink() + ) + heard = _hears( + hooks, + HookKind.PERMISSION_REQUEST, + PermissionRequestHookResult(allow=False, reason="not in this repo"), + ) + assert await handle.turn(TurnRequest("use a tool"), RecordingSink()) == "denied" + hooks.set(HookKind.PERMISSION_REQUEST, None) + await handle.turn(TurnRequest("use a tool"), RecordingSink()) + said = clis.of("codex") + started = [one["params"] for one in said if one.get("method") == "turn/start"] + # Every command asked about only while the hook is hung, and never by a model. + assert [one["approvalPolicy"] for one in started] == ["never", "untrusted", "never"] + assert [one["tool"] for one in heard] == ["commandExecution"] + # Codex takes a refusal with no reason, so the reason reaches the model as a steer. + steered = [ + one["params"]["input"][0]["text"] + for one in said + if one.get("method") == "turn/steer" + ] + assert steered == ["Using commandExecution was refused: not in this repo"] + + +async def test_a_hook_that_never_answers_is_given_up_on_and_the_tool_allowed( + clis: Logs, work: Placement, monkeypatch: pytest.MonkeyPatch +) -> None: + del clis + monkeypatch.setattr(harnesses, "HOOK_TIMEOUT", 0.5) + stuck = asyncio.Event() + + async def never(handle: SessionHandle, fields: dict[str, Any]) -> HookResult: + del handle, fields + await stuck.wait() + return PermissionRequestHookResult(allow=False) + + for harness in (HarnessKind.CLAUDE, HarnessKind.CODEX): + hooks = HookTable() + hooks.set(HookKind.PERMISSION_REQUEST, never) + driver = open_agent(SPECS[harness]) + try: + handle = await _open(driver, work, hooks) + said = await asyncio.wait_for( + handle.turn(TurnRequest("use a tool"), RecordingSink()), 30 + ) + finally: + await driver.close() + assert said == "allowed", harness + + +async def test_codex_is_started_able_to_ask_once_an_ask_user_hook_is_hung( + codex: HarnessDriver, work: Placement, clis: Logs +) -> None: + hooks = HookTable() + handle = await _open(codex, work, hooks) + assert await handle.turn(TurnRequest("ask me"), RecordingSink()) == "nobody" + _hears(hooks, HookKind.ASK_USER, AskUserHookResult(answer="right")) + assert await handle.turn(TurnRequest("ask me"), RecordingSink()) == "right" + said = clis.of("codex") + started = [one["argv"] for one in said if "argv" in one] + assert len(started) == 2 + assert "default_mode_request_user_input" not in started[0] + assert started[1][started[1].index("--enable") + 1] == ( + "default_mode_request_user_input" + ) + # Started again between two turns, with nothing in flight, and the conversation picked + # back up on the new server rather than begun again. + threads = { + one["params"]["threadId"] for one in said if one.get("method") == "turn/start" + } + resumed = [ + one["params"]["threadId"] + for one in said + if one.get("method") == "thread/resume" + ] + assert len(threads) == 1 + assert resumed == list(threads) + + +# ------------------------------------------------------------------------ steering and goals + + +@pytest.mark.parametrize("harness", [HarnessKind.CLAUDE, HarnessKind.CODEX], ids=str) +async def test_a_queued_steer_is_taken_into_the_turn_in_flight( + clis: Logs, work: Placement, harness: HarnessKind +) -> None: + del clis + driver = open_agent(SPECS[harness]) + try: + handle = await _open(driver, work) + turning = asyncio.create_task( + handle.turn(TurnRequest("slow, please"), RecordingSink()) + ) + await asyncio.sleep(0.5) + await handle.steer("Reply with the single word: turned", queued=True) + assert await asyncio.wait_for(turning, 30) == "turned" + finally: + await driver.close() + + +@pytest.mark.parametrize("harness", [HarnessKind.CLAUDE, HarnessKind.CODEX], ids=str) +async def test_a_goal_is_the_harness_own( + clis: Logs, work: Placement, harness: HarnessKind +) -> None: + driver = open_agent(SPECS[harness]) + try: + handle = await _open(driver, work) + said = await handle.turn( + TurnRequest("/goal Reply with the single word: done"), RecordingSink() + ) + finally: + await driver.close() + if harness is HarnessKind.CLAUDE: + assert said == "goal met: Reply with the single word: done" + else: + assert said == "done" + goals = [ + one for one in clis.of("codex") if one.get("method") == "thread/goal/set" + ] + assert [one["params"]["objective"] for one in goals] == [ + "Reply with the single word: done" + ] + + +async def test_a_session_that_cannot_steer_says_so(clis: Logs, work: Placement) -> None: + del clis + driver = open_agent(SPECS[HarnessKind.OPENCODE]) + try: + handle = await _open(driver, work) + with pytest.raises(UnsupportedOperation): + await handle.steer("hello", queued=True) + finally: + await driver.close() + + +# ------------------------------------------------------------------------------------ forks + + +@pytest.mark.parametrize("harness", [HarnessKind.CLAUDE, HarnessKind.CODEX], ids=str) +async def test_a_fork_carries_on_in_another_workdir( + clis: Logs, tmp_path: Path, harness: HarnessKind +) -> None: + driver = open_agent(SPECS[harness]) + here, there = _placement(tmp_path / "here"), _placement(tmp_path / "there") + try: + parent = await _open(driver, here) + await parent.turn( + TurnRequest("Reply with the single word: one"), RecordingSink() + ) + child = await driver.open( + there, + permission=Permission(), + skills=(), + hooks=HookTable(), + fork_of=parent, + ) + said = await child.turn( + TurnRequest("Reply with the single word: two"), RecordingSink() + ) + assert said == "two" + assert child.id not in (None, parent.id) + finally: + await driver.close() + if harness is HarnessKind.CLAUDE: + started = [one for one in clis.of("claude") if "argv" in one] + assert started[-1]["cwd"] == str(there.workdir) + assert "--fork-session" in started[-1]["argv"] + else: + forked = [one for one in clis.of("codex") if one.get("method") == "thread/fork"] + assert [one["params"]["cwd"] for one in forked] == [str(there.workdir)] + + +async def test_a_harness_that_forks_only_where_it_is_refuses_another_workdir( + clis: Logs, tmp_path: Path +) -> None: + del clis + driver = open_agent(SPECS[HarnessKind.OPENCODE]) + try: + parent = await _open(driver, _placement(tmp_path / "here")) + await parent.turn( + TurnRequest("Reply with the single word: one"), RecordingSink() + ) + with pytest.raises(UnsupportedOperation): + await driver.open( + _placement(tmp_path / "there"), + permission=Permission(), + skills=(), + hooks=HookTable(), + fork_of=parent, + ) + finally: + await driver.close() + + +async def test_a_second_fork_elsewhere_carries_the_conversation_as_it_is_now( + claude: HarnessDriver, tmp_path: Path +) -> None: + here = _placement(tmp_path / "here") + parent = await _open(claude, here) + await parent.turn(TurnRequest("Reply with the single word: one"), RecordingSink()) + await claude.open( + _placement(tmp_path / "b"), + permission=Permission(), + skills=(), + hooks=HookTable(), + fork_of=parent, + ) + await parent.turn(TurnRequest("Reply with the single word: two"), RecordingSink()) + later = await claude.open( + _placement(tmp_path / "c"), + permission=Permission(), + skills=(), + hooks=HookTable(), + fork_of=parent, + ) + await later.turn(TurnRequest("Reply with the single word: three"), RecordingSink()) + projects = tmp_path / "claude-home" / "projects" + (carried,) = projects.glob(f"*c/{later.id}.jsonl") + said = [json.loads(line)["user"] for line in carried.read_text().splitlines()] + assert said == [ + "Reply with the single word: one", + "Reply with the single word: two", + "Reply with the single word: three", + ] + + +async def test_codex_counts_a_thread_picked_up_on_a_new_server_from_where_it_was( + codex: HarnessDriver, work: Placement +) -> None: + handle = await _open(codex, work) + sink = RecordingSink() + await handle.turn(TurnRequest("Reply with the single word: one"), sink) + turning = asyncio.create_task(handle.turn(TurnRequest("slow, please"), sink)) + await asyncio.sleep(0.5) + handle.interrupt() + with pytest.raises(SessionError): + await turning + before = handle.usage.output_tokens + # The interrupt put the server down, so this is the thread picked up on a new one, whose + # first word states the thread's whole total: only what this turn cost is counted. + await handle.turn(TurnRequest("Reply with the single word: two"), sink) + assert handle.usage.output_tokens - before == 2 + + +async def test_a_session_with_no_turn_has_nothing_to_fork( + claude: HarnessDriver, work: Placement +) -> None: + parent = await _open(claude, work) + with pytest.raises(SessionError): + await claude.open( + work, permission=Permission(), skills=(), hooks=HookTable(), fork_of=parent + ) + + +async def test_a_fork_is_refused_once_its_parent_has_moved_on( + claude: HarnessDriver, work: Placement +) -> None: + parent = await _open(claude, work) + await parent.turn(TurnRequest("Reply with the single word: one"), RecordingSink()) + child = await claude.open( + work, permission=Permission(), skills=(), hooks=HookTable(), fork_of=parent + ) + await parent.turn(TurnRequest("Reply with the single word: two"), RecordingSink()) + with pytest.raises(SessionError): + await child.turn( + TurnRequest("Reply with the single word: three"), RecordingSink() + ) + + +# ------------------------------------------------------------------------ schemas and limits + + +class Counted(pydantic.BaseModel): + """A shape the stand-ins never answer in.""" + + number: int + + +async def test_an_answer_not_in_the_shape_is_an_output_schema_error( + claude: HarnessDriver, work: Placement +) -> None: + handle = await _open(claude, work) + with pytest.raises(OutputSchemaError) as raised: + await handle.turn(TurnRequest("count", Counted), RecordingSink()) + assert isinstance(raised.value, ValueError) + + +async def test_a_turn_past_its_output_tokens_is_stopped_and_raises( + claude: HarnessDriver, work: Placement, clis: Logs +) -> None: + handle = await _open(claude, work) + sink = RecordingSink() + with pytest.raises(OutputTokensExceeded): + await handle.turn( + TurnRequest("slow, please", limits=Limits(output_tokens=5, graceful=False)), + sink, + ) + assert sink.output_tokens >= 5 + said = await handle.turn(TurnRequest("Reply with the single word: next"), sink) + assert said == "next" + assert len([one for one in clis.of("claude") if "argv" in one]) == 2 + + +@pytest.mark.parametrize( + "limits", + [ + Limits(output_tokens=1, graceful=True), + Limits(cost=1e-9, graceful=True), + Limits(deadline=time.monotonic() + 3600, graceful=True), + ], +) +async def test_a_graceful_turn_runs_to_its_end_whatever_it_spends( + claude: HarnessDriver, work: Placement, priced: str, limits: Limits +) -> None: + del priced + handle = await _open(claude, work) + sink = RecordingSink() + said = await handle.turn( + TurnRequest("Reply with the single word: finished", limits=limits), sink + ) + assert said == "finished" + assert sink.output_tokens > 1 + + +async def test_a_graceful_deadline_lets_the_turn_finish_past_it( + codex: HarnessDriver, work: Placement +) -> None: + handle = await _open(codex, work) + turning = asyncio.create_task( + handle.turn( + TurnRequest( + "slow, please", + limits=Limits(deadline=time.monotonic() + 0.3, graceful=True), + ), + RecordingSink(), + ) + ) + await asyncio.sleep(1.0) + assert not turning.done(), "a graceful turn was stopped at its deadline" + await handle.steer("Reply with the single word: late", queued=True) + assert await asyncio.wait_for(turning, 30) == "late" + + +async def test_a_turn_past_its_deadline_is_stopped_and_raises( + codex: HarnessDriver, work: Placement +) -> None: + handle = await _open(codex, work) + with pytest.raises(DurationExceeded) as raised: + await handle.turn( + TurnRequest( + "slow, please", + limits=Limits(deadline=time.monotonic() + 0.5, graceful=False), + ), + RecordingSink(), + ) + assert isinstance(raised.value, TimeoutError) + + +async def test_a_turn_past_its_cost_is_stopped_and_raises( + claude: HarnessDriver, work: Placement, priced: str +) -> None: + del priced + handle = await _open(claude, work) + with pytest.raises(CostExceeded): + await handle.turn( + TurnRequest("slow, please", limits=Limits(cost=1e-5, graceful=False)), + RecordingSink(), + ) + + +async def test_a_turn_with_nothing_left_to_spend_starts_nothing( + claude: HarnessDriver, work: Placement, clis: Logs +) -> None: + handle = await _open(claude, work) + with pytest.raises(OutputTokensExceeded): + await handle.turn( + TurnRequest("anything", limits=Limits(output_tokens=0)), RecordingSink() + ) + assert clis.of("claude") == [] + + +# ----------------------------------------------------------------------------------- errors + + +@pytest.mark.parametrize( + ("words", "leaf"), + [ + ("401 Unauthorized: invalid api key", HarnessRefused), + ("model not found", ModelUnavailable), + ], +) +async def test_a_failed_turn_is_the_leaf_for_why( + claude: HarnessDriver, + work: Placement, + words: str, + leaf: type[Exception], +) -> None: + handle = await _open(claude, work) + with pytest.raises(leaf) as raised: + await handle.turn(TurnRequest(f"fail: {words}"), RecordingSink()) + assert raised.value.__cause__ is not None + + +async def test_a_cli_that_is_not_installed_is_refused_at_open( + work: Placement, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setenv("PATH", str(tmp_path / "nothing")) + monkeypatch.setenv("HOME", str(tmp_path / "home")) + driver = open_agent(SPECS[HarnessKind.CLAUDE]) + with pytest.raises(HarnessNotInstalled): + await _open(driver, work) + + +async def test_a_read_only_session_runs_at_the_read_only_rung( + claude: HarnessDriver, work: Placement, clis: Logs +) -> None: + handle = await _open( + claude, + work, + permission=Permission(local=PermissionKind.READ, user=PermissionKind.READ), + ) + await handle.turn(TurnRequest("Reply with the single word: ok"), RecordingSink()) + argv = clis.of("claude")[0]["argv"] + assert argv[argv.index("--permission-mode") + 1] == "plan" + assert argv[argv.index("--disallowedTools") + 1] == "WebSearch,WebFetch" + + +# ------------------------------------------------------------------------------ outworlder + + +class Choice(pydantic.BaseModel): + """What an outworlder is asked to fill in.""" + + way: str = pydantic.Field(description="Which way to go") + far: int = 1 + + +async def test_an_outworlder_answers_through_the_callback_it_was_given() -> None: + asked: list[Question] = [] + + def ask(question: Question) -> str | None: + asked.append(question) + return "3" if question.text.startswith("far") else "north" + + outworlder = open_outworlder(ask=ask) + assert not outworlder.away + assert await outworlder.run("Where now?", None) == "north" + chosen = await outworlder.run("Choose.", Choice) + assert chosen == Choice(way="north", far=3) + assert asked[0].text == "Where now?" + + +async def test_an_outworlder_nobody_answers_is_away() -> None: + outworlder = open_outworlder(ask=lambda question: None) + with pytest.raises(OutworlderAway): + await outworlder.run("Anyone?", None) + + +async def test_an_away_outworlder_answers_what_it_can_by_default() -> None: + outworlder = open_outworlder() + assert outworlder.away + assert await outworlder.run("Anyone?", None) == "" + + class Defaulted(pydantic.BaseModel): + far: int = 2 + + assert await outworlder.run("Anyone?", Defaulted) == Defaulted() + with pytest.raises(OutworlderAway): + await outworlder.run("Anyone?", Choice) + here = open_outworlder(ask=lambda question: "x", away=lambda: True) + assert here.away + assert await here.run("Anyone?", None) == "" + + +async def test_a_session_carries_its_skills_where_its_cli_reads_them( + claude: HarnessDriver, work: Placement, tmp_path: Path +) -> None: + kept = tmp_path / "skills" / "tidy" + kept.mkdir(parents=True) + (kept / "SKILL.md").write_text( + "---\nname: tidy\ndescription: Tidy up.\n---\nTidy.\n" + ) + handle = await claude.open( + work, + permission=Permission(), + skills=(Skill("tidy", kept),), + hooks=HookTable(), + ) + await handle.turn(TurnRequest("Reply with the single word: ok"), RecordingSink()) + mounted = Path(str(work.workdir)) / ".claude" / "skills" / "tidy" / "SKILL.md" + assert mounted.read_text().endswith("Tidy.\n") + await handle.close() + assert not mounted.exists() diff --git a/tests/system/agents/test_harness_drivers.py b/tests/system/agents/test_harness_drivers.py new file mode 100644 index 000000000..164f5cb3b --- /dev/null +++ b/tests/system/agents/test_harness_drivers.py @@ -0,0 +1,295 @@ +"""The harness drivers against the real CLIs installed here, as the account signed in on them. + +Each installed CLI is run through :func:`tests.flows.contracts.check_agent_driver` on its +cheapest model -- a plain turn, a schema turn, an interrupt, a cancel, a steer where it steers, +and a fork where it forks -- and then through what only the real thing can show: a `/goal` +where the harness has one, a fork into another workdir, a permission request and a question +reaching the flow's hooks. + +A CLI that is installed but refuses the account -- signed out, a model it may not name -- is +skipped, saying so, rather than failed: that is the account's, and the stand-ins in +`tests/integration/flows/test_harness_drivers.py` hold the driver itself. + +These cost tokens and need network access, so they only run with ``pytest --run-agents``. +""" + +from __future__ import annotations + +import json +import shutil +import uuid +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Any, cast + +import pytest + +from hmz.flows import ( + AskUserHookResult, + EnvBackendKind, + HarnessError, + HarnessKind, + HookKind, + HookResult, + Permission, + PermissionRequestHookAgentMixin, + PermissionRequestHookResult, + UnsupportedOperation, +) +from hmz.runtime.flowing.harnesses import open_agent +from hmz.runtime.flowing.specs import AgentSpec +from hmz.runtime.flowing.spi import ( + HARNESS_CAPABILITIES, + HookTable, + Placement, + TurnRequest, +) +from tests.flows.contracts import RecordingSink, check_agent_driver + +if TYPE_CHECKING: + from hmz.runtime.flowing.harnesses import HarnessDriver + from hmz.runtime.flowing.spi import SessionHandle + +pytestmark = [pytest.mark.agent, pytest.mark.timeout(900)] + +#: What each harness is run at: the model it is cheapest to ask, and the least effort it takes. +#: A harness whose CLI has said nothing about what it runs here is looked up in what it did +#: say, in `~/.humanize/models`. +CHEAPEST: dict[HarnessKind, tuple[str, str]] = { + HarnessKind.CLAUDE: ("claude-haiku-4-5-20251001", "low"), + HarnessKind.CODEX: ("gpt-5.5", "low"), + HarnessKind.CURSOR_AGENT: ("auto", ""), + HarnessKind.OPENCODE: ("opencode/nemotron-3.5-lightning-free", ""), + HarnessKind.MIMO: ("xiaomi/mimo-v2.5", "low"), + HarnessKind.QWEN: ("qwen3-coder-flash", "low"), + HarnessKind.GROK: ("grok-4.5", "low"), + HarnessKind.AGY: ("gemini-3.8-flash-low", "low"), + HarnessKind.DSH: ("deepseek-v4-flash", "off"), +} + +#: A prompt that keeps a turn going long enough to be interrupted, cancelled and steered. +SLOW = ( + "Count from 1 to 400, writing each number on its own line, and nothing else. Do not " + "use any tools." +) + +#: How long a real CLI is given to be under way before its turn is interrupted. +SETTLE = 4.0 + + +def _model(harness: HarnessKind) -> tuple[str, str]: + """The model and effort to run a harness at here, or a skip saying why there is none.""" + if harness in CHEAPEST: + return CHEAPEST[harness] + kept = Path.home() / ".humanize" / "models" / f"{harness.value}.json" + try: + said = json.loads(kept.read_text(encoding="utf-8")) + except (OSError, ValueError): + pytest.skip(f"nothing has asked {harness} what it runs on this machine") + models = cast( + "list[dict[str, Any]]", cast("dict[str, Any]", said).get("models") or [] + ) + if not models: + pytest.skip(f"{harness} has said nothing about what it runs on this machine") + efforts = cast("list[str]", models[0].get("efforts") or []) + return str(models[0]["name"]), efforts[-1] if efforts else "" + + +def _driver(harness: HarnessKind) -> HarnessDriver: + """A driver of the real CLI, or a skip where it is not installed here.""" + if harness is HarnessKind.DSH: + import importlib.util + + if importlib.util.find_spec("deepseek_harness") is None: + pytest.skip("the DeepSeek Harness SDK is not installed here") + elif shutil.which(harness.value) is None: + pytest.skip(f"{harness} is not installed here") + model, effort = _model(harness) + return open_agent(AgentSpec("coder", harness, "", model, effort, harness.value)) + + +def _placement(where: Path) -> Placement: + where.mkdir(parents=True, exist_ok=True) + return Placement(EnvBackendKind.LOCAL, "", PurePosixPath(str(where))) + + +async def _answers(driver: HarnessDriver, placement: Placement) -> None: + """Skips a harness whose account will not take a turn here at all.""" + handle = await driver.open( + placement, permission=Permission(), skills=(), hooks=HookTable() + ) + try: + await handle.turn( + TurnRequest("Reply with the single word: ok"), RecordingSink() + ) + except HarnessError as refused: + await driver.close() + pytest.skip(f"{driver.harness} will not take a turn here: {refused}") + finally: + await handle.close() + + +@pytest.mark.parametrize( + "harness", [one for one in HarnessKind if one is not HarnessKind.ACP], ids=str +) +async def test_the_real_cli_keeps_the_driver_contract( + harness: HarnessKind, tmp_path: Path +) -> None: + driver = _driver(harness) + placement = _placement(tmp_path / "work") + await _answers(driver, placement) + await check_agent_driver(driver, placement, slow_prompt=SLOW, settle=SETTLE) + + +@pytest.mark.parametrize( + "harness", + [HarnessKind.CLAUDE, HarnessKind.CODEX, HarnessKind.KIMI, HarnessKind.ZCODE], + ids=str, +) +async def test_a_goal_is_met_through_the_harness_own_goal( + harness: HarnessKind, tmp_path: Path +) -> None: + driver = _driver(harness) + placement = _placement(tmp_path / "work") + await _answers(driver, placement) + try: + handle = await driver.open( + placement, permission=Permission(), skills=(), hooks=HookTable() + ) + done = tmp_path / "work" / "DONE.txt" + said = await handle.turn( + TurnRequest(f"/goal Create the file {done} whose only content is: done"), + RecordingSink(), + ) + finally: + await driver.close() + assert isinstance(said, str) + assert done.read_text().strip() == "done" + + +@pytest.mark.parametrize( + "harness", + [HarnessKind.CLAUDE, HarnessKind.CODEX, HarnessKind.KIMI, HarnessKind.ZCODE], + ids=str, +) +async def test_a_fork_carries_on_in_another_workdir( + harness: HarnessKind, tmp_path: Path +) -> None: + driver = _driver(harness) + here, there = _placement(tmp_path / "here"), _placement(tmp_path / "there") + await _answers(driver, here) + word = uuid.uuid4().hex[:6] + try: + parent = await driver.open( + here, permission=Permission(), skills=(), hooks=HookTable() + ) + await parent.turn( + TurnRequest( + f"Remember the code word {word}. Reply with the single word: ok" + ), + RecordingSink(), + ) + try: + child = await driver.open( + there, + permission=Permission(), + skills=(), + hooks=HookTable(), + fork_of=parent, + ) + except UnsupportedOperation as refused: + pytest.fail(f"{harness} is said to fork into another workdir: {refused}") + said = await child.turn( + TurnRequest( + "What was the code word? Reply with it alone, then run `pwd` and reply " + "with its output on a second line." + ), + RecordingSink(), + ) + finally: + await driver.close() + assert isinstance(said, str) + assert word in said + assert str(there.workdir) in said + + +def _recording(table: HookTable, kind: HookKind, answer: HookResult) -> list[Any]: + heard: list[Any] = [] + + async def bound(handle: SessionHandle, fields: dict[str, Any]) -> HookResult: + del handle + heard.append(fields) + return answer + + table.set(kind, bound) + return heard + + +@pytest.mark.parametrize( + "harness", + [ + one + for one in HarnessKind + if PermissionRequestHookAgentMixin in HARNESS_CAPABILITIES[one] + ], + ids=str, +) +async def test_a_permission_request_reaches_the_hook_and_its_no_holds( + harness: HarnessKind, tmp_path: Path +) -> None: + driver = _driver(harness) + placement = _placement(tmp_path / "work") + await _answers(driver, placement) + hooks = HookTable() + heard = _recording( + hooks, + HookKind.PERMISSION_REQUEST, + PermissionRequestHookResult(allow=False, reason="not allowed in this test"), + ) + probe = Path.home() / f".hmz-harness-probe-{uuid.uuid4().hex[:8]}" + try: + handle = await driver.open( + placement, permission=Permission(), skills=(), hooks=hooks + ) + await handle.turn( + TurnRequest( + f"Use your shell tool to run exactly: touch {probe} -- and if you are not " + "allowed to, reply with the single word: refused" + ), + RecordingSink(), + ) + finally: + await driver.close() + existed = probe.exists() + probe.unlink(missing_ok=True) + assert heard, f"{harness} never asked the hook" + assert not existed, "the tool ran although the hook refused it" + + +@pytest.mark.parametrize( + "harness", [HarnessKind.CLAUDE, HarnessKind.CODEX, HarnessKind.KIMI], ids=str +) +async def test_a_question_reaches_the_ask_user_hook( + harness: HarnessKind, tmp_path: Path +) -> None: + driver = _driver(harness) + placement = _placement(tmp_path / "work") + await _answers(driver, placement) + hooks = HookTable() + heard = _recording(hooks, HookKind.ASK_USER, AskUserHookResult(answer="blue")) + try: + handle = await driver.open( + placement, permission=Permission(), skills=(), hooks=hooks + ) + said = await handle.turn( + TurnRequest( + "Use your tool for asking the user a question to ask me which colour I " + "prefer, offering exactly the options red and blue. Then reply with just " + "the colour I picked, one word." + ), + RecordingSink(), + ) + finally: + await driver.close() + assert heard, f"{harness} never asked the user" + assert isinstance(said, str) + assert "blue" in said.lower() diff --git a/tests/unit/flows/test_harness_mapping.py b/tests/unit/flows/test_harness_mapping.py new file mode 100644 index 000000000..0747fd253 --- /dev/null +++ b/tests/unit/flows/test_harness_mapping.py @@ -0,0 +1,562 @@ +"""What the harness drivers decide before any CLI is started. + +Which rung a permission comes to on each harness, what a hook's answer tells coganchor, which +error a failed turn is, what an answer is read as, and that every harness's driver serves +exactly the capabilities its protocol declares -- each of them read off the tables in +:mod:`hmz.runtime.flowing.harnessing` and the driver built by `open_agent`, with nothing spawned. +""" + +from __future__ import annotations + +import subprocess +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Any + +import pydantic +import pytest + +from hmz.coganchor import backends +from hmz.coganchor.agents import ( + AgentBase, + CodexAgent, + CodexAgentConfig, + Failed, + Moment, + Occasion, + Stopped, + Unrecoverable, + Verdict, + driver, +) +from hmz.coganchor.agents.claude import _project +from hmz.coganchor.agents.codex import strict +from hmz.coganchor.anchor import NotInstalled +from hmz.flows import ( + AskUserHookAgentMixin, + AskUserHookResult, + EnvBackendKind, + GoalCommandAgentMixin, + HarnessContended, + HarnessDropped, + HarnessKilled, + HarnessKind, + HarnessMissing, + HarnessNotInstalled, + HarnessRefused, + HarnessSandboxed, + HarnessThrottled, + HarnessUnrecoverable, + HookKind, + ModelUnavailable, + OutputSchemaError, + OutworlderAway, + Permission, + PermissionKind, + PermissionRequestHookAgentMixin, + PermissionRequestHookResult, + PreToolUseHookResult, + SessionError, + SteeringAgentMixin, + StopHookResult, + SubagentStartHookAgentMixin, + SubagentStopHookAgentMixin, + UnsupportedOperation, +) +from hmz.runtime.flowing import harnessing +from hmz.runtime.flowing.harnesses import away_answer, open_agent, settled +from hmz.runtime.flowing.specs import AgentSpec +from hmz.runtime.flowing.spi import HARNESS_CAPABILITIES, HookTable, Placement + +if TYPE_CHECKING: + from hmz.flows import HarnessError + +NONE, READ, ALL = PermissionKind.NONE, PermissionKind.READ, PermissionKind.ALL + +#: The permissions the table is read at: looking only, the workdir alone, the whole machine. +LOOKING = Permission(local=READ, user=READ, system=READ) +WORKDIR = Permission(local=ALL, user=READ, system=NONE) +EVERYTHING = Permission(local=ALL, user=ALL, system=ALL, online=ALL) + +#: The rung each harness runs at for each of them, with no hook hung. +RUNGS: dict[HarnessKind, tuple[str, str, str]] = { + HarnessKind.CLAUDE: ("read-only", "bypass", "bypass"), + HarnessKind.CODEX: ("read-only", "bypass", "bypass"), + HarnessKind.CURSOR_AGENT: ("read-only", "bypass", "bypass"), + HarnessKind.OPENCODE: ("read-only", "bypass", "bypass"), + HarnessKind.MIMO: ("read-only", "bypass", "bypass"), + HarnessKind.QWEN: ("read-only", "bypass", "bypass"), + HarnessKind.KIMI: ("read-only", "bypass", "bypass"), + HarnessKind.GROK: ("read-only", "bypass", "bypass"), + HarnessKind.PI: ("read-only", "bypass", "bypass"), + HarnessKind.ZCODE: ("read-only", "bypass", "bypass"), + HarnessKind.AGY: ("read-only", "bypass", "bypass"), + HarnessKind.DSH: ("bypass", "bypass", "bypass"), + HarnessKind.ACP: ("bypass", "bypass", "bypass"), +} + +#: The CLI an added ACP harness is, in these tests. +ADDED = "acme" + + +def _spec(harness: HarnessKind, model: str = "m", effort: str = "") -> AgentSpec: + cli = ADDED if harness is HarnessKind.ACP else harness.value + return AgentSpec("coder", harness, "", model, effort, cli) + + +@pytest.fixture(autouse=True) +def _acp_added() -> None: + """Adds a CLI spoken to over ACP, under the suite's own home.""" + backends.remember(ADDED, [ADDED, "--acp"]) + + +def _kind(harness: HarnessKind) -> type[AgentBase]: + return driver(ADDED if harness is HarnessKind.ACP else harness.value)[0] + + +# --------------------------------------------------------------------------------- rungs + + +@pytest.mark.parametrize("harness", sorted(RUNGS), ids=str) +def test_each_permission_comes_to_the_rung_the_table_says(harness: HarnessKind) -> None: + rungs = _kind(harness).rungs + said = tuple( + harnessing.rung(harness, permission, rungs=rungs) + for permission in (LOOKING, WORKDIR, EVERYTHING) + ) + assert said == RUNGS[harness] + + +@pytest.mark.parametrize("harness", sorted(RUNGS), ids=str) +def test_no_rung_is_ever_a_model_reviewing_itself(harness: HarnessKind) -> None: + rungs = _kind(harness).rungs + for permission in (LOOKING, WORKDIR, EVERYTHING): + for hung in (frozenset[HookKind](), frozenset(HookKind)): + said = harnessing.rung(harness, permission, rungs=rungs, hung=hung) + if said == harnessing.ASKING: + assert harness in harnessing.ASKS, (harness, permission, hung) + assert said in rungs + + +@pytest.mark.parametrize("harness", sorted(RUNGS), ids=str) +def test_a_session_that_may_write_its_workdir_is_fenced_nowhere( + harness: HarnessKind, +) -> None: + # Pinned on purpose: `user` and `system` are not held below ALL once `local` is, since + # the only fences there are -- Codex's and cursor-agent's sandboxes -- cannot start where + # there is no user namespace, and the flow API calls Codex's BYPASS full access. + assert frozenset() == harnessing.FENCED + rungs = _kind(harness).rungs + for user in (NONE, READ, ALL): + narrow = Permission(local=ALL, user=user, system=NONE) + assert harnessing.rung(harness, narrow, rungs=rungs) == "bypass" + + +@pytest.mark.parametrize( + ("harness", "hook", "rung"), + [ + (HarnessKind.CODEX, HookKind.PERMISSION_REQUEST, "bypass"), + (HarnessKind.CODEX, HookKind.ASK_USER, "bypass"), + (HarnessKind.KIMI, HookKind.PERMISSION_REQUEST, "auto"), + (HarnessKind.KIMI, HookKind.ASK_USER, "auto"), + (HarnessKind.ZCODE, HookKind.PERMISSION_REQUEST, "auto"), + (HarnessKind.ZCODE, HookKind.ASK_USER, "auto"), + (HarnessKind.CLAUDE, HookKind.PERMISSION_REQUEST, "bypass"), + (HarnessKind.GROK, HookKind.PERMISSION_REQUEST, "bypass"), + (HarnessKind.KIMI, HookKind.PRE_TOOL_USE, "bypass"), + ], +) +def test_a_harness_asks_only_while_a_hook_is_hung_on_what_it_asks( + harness: HarnessKind, hook: HookKind, rung: str +) -> None: + rungs = _kind(harness).rungs + assert harnessing.rung( + harness, EVERYTHING, rungs=rungs, hung=frozenset({hook}) + ) == (rung) + assert ( + harnessing.rung(harness, LOOKING, rungs=rungs, hung=frozenset({hook})) + == (RUNGS[harness][0]) + ) + + +def test_codex_asks_every_command_while_a_permission_hook_is_hung() -> None: + asking = frozenset({HookKind.PERMISSION_REQUEST}) + assert harnessing.approvals(HarnessKind.CODEX, asking) == "untrusted" + assert harnessing.approvals(HarnessKind.CODEX, frozenset()) == "" + assert harnessing.approvals(HarnessKind.CLAUDE, asking) == "" + + +def test_the_web_is_said_where_the_cli_can_be_told_and_left_where_it_cannot() -> None: + assert harnessing.searching(EVERYTHING, tellable=True) is True + assert harnessing.searching(WORKDIR, tellable=True) is False + assert harnessing.searching(WORKDIR, tellable=False) is None + + +def test_a_session_config_carries_the_rung_the_web_and_codex_asking() -> None: + config = CodexAgentConfig(model="m", effort="low", features=(("undo", True),)) + plain = settled( + config, HarnessKind.CODEX, CodexAgent, WORKDIR, frozenset(), tellable=True + ) + assert (plain.permission, plain.web_search) == ("bypass", False) + assert isinstance(plain, CodexAgentConfig) + assert plain.features == (("undo", True),) + assert plain.approvals == "" + asking = settled( + config, + HarnessKind.CODEX, + CodexAgent, + EVERYTHING, + frozenset({HookKind.ASK_USER, HookKind.PERMISSION_REQUEST}), + tellable=True, + ) + assert (asking.permission, asking.web_search) == ("bypass", True) + assert isinstance(asking, CodexAgentConfig) + assert asking.approvals == "untrusted" + assert asking.features == ( + ("undo", True), + ("default_mode_request_user_input", True), + ) + + +# ---------------------------------------------------------------------------- capabilities + + +@pytest.mark.parametrize("harness", sorted(HarnessKind), ids=str) +def test_every_driver_serves_exactly_its_harness_capabilities( + harness: HarnessKind, +) -> None: + made = open_agent(_spec(harness)) + assert made.harness is harness + assert made.capabilities == HARNESS_CAPABILITIES[harness] + assert (made.model, made.effort, made.provider) == ("m", "", "") + + +@pytest.mark.parametrize("harness", sorted(HarnessKind), ids=str) +def test_each_capability_is_one_the_coganchor_driver_has(harness: HarnessKind) -> None: + kind = _kind(harness) + capabilities = HARNESS_CAPABILITIES[harness] + moments = kind.moments + assert (GoalCommandAgentMixin in capabilities) == kind.pursues + assert (PermissionRequestHookAgentMixin in capabilities) <= ( + Moment.PERMISSION_REQUEST in moments + ) + assert (SubagentStartHookAgentMixin in capabilities) == ( + Moment.SUBAGENT_START in moments + ) + assert (SubagentStopHookAgentMixin in capabilities) == ( + Moment.SUBAGENT_STOP in moments + ) + + +@pytest.mark.parametrize( + ("harness", "steers", "asks"), + [ + (HarnessKind.CLAUDE, True, True), + (HarnessKind.CODEX, True, True), + (HarnessKind.KIMI, True, True), + (HarnessKind.PI, True, True), + (HarnessKind.ZCODE, False, True), + (HarnessKind.GROK, False, False), + (HarnessKind.QWEN, False, False), + (HarnessKind.CURSOR_AGENT, False, False), + ], +) +def test_steering_and_asking_are_where_the_coganchor_driver_has_them( + harness: HarnessKind, steers: bool, asks: bool +) -> None: + capabilities = HARNESS_CAPABILITIES[harness] + assert (SteeringAgentMixin in capabilities) == steers + assert (AskUserHookAgentMixin in capabilities) == asks + agent = open_agent(_spec(harness)) + kind = _kind(harness) + session = kind(agent._config).new() + assert type(session).steers == steers + + +def test_an_effort_the_cli_has_no_word_for_is_refused_where_the_agent_is_named() -> ( + None +): + with pytest.raises(HarnessUnrecoverable): + open_agent(_spec(HarnessKind.CLAUDE, effort="ludicrous")) + + +def test_a_cli_nobody_added_is_not_installed() -> None: + spec = AgentSpec("coder", HarnessKind.ACP, "", "m", "", "nobody-added-this") + with pytest.raises(HarnessNotInstalled): + open_agent(spec) + + +async def test_a_session_cannot_be_opened_in_a_workdir_that_is_not_there( + tmp_path: Path, +) -> None: + made = open_agent(_spec(HarnessKind.CLAUDE)) + made._installed = True + with pytest.raises(SessionError): + await made.open( + Placement(EnvBackendKind.LOCAL, "", PurePosixPath(str(tmp_path / "gone"))), + permission=Permission(), + skills=(), + hooks=HookTable(), + ) + + +async def test_a_session_opens_without_starting_its_cli(tmp_path: Path) -> None: + made = open_agent(_spec(HarnessKind.CURSOR_AGENT)) + made._installed = True + placement = Placement(EnvBackendKind.LOCAL, "", PurePosixPath(str(tmp_path))) + handle = await made.open( + placement, permission=WORKDIR, skills=(), hooks=HookTable() + ) + assert handle.id is None + assert handle.usage.output_tokens == 0 + assert handle.agent.config.permission == "bypass" + with pytest.raises(UnsupportedOperation): + await handle.steer("anything", queued=True) + handle.interrupt() + with pytest.raises(SessionError): + await made.open( + placement, permission=WORKDIR, skills=(), hooks=HookTable(), fork_of=handle + ) + await made.close() + assert handle.closed + with pytest.raises(SessionError): + await made.open(placement, permission=WORKDIR, skills=(), hooks=HookTable()) + + +# ---------------------------------------------------------------------------------- hooks + + +def _occasion(moment: Moment, **said: Any) -> Occasion: + return Occasion(moment=moment, agent="coder", **said) + + +def test_a_tool_is_told_with_what_it_was_called_with_or_the_line_it_was_shown_as() -> ( + None +): + called = _occasion(Moment.PRE_TOOL_USE, tool="Bash", input={"command": "ls"}) + shown = _occasion(Moment.PRE_TOOL_USE, tool="Bash", about="git push --force") + assert harnessing.fields(HookKind.PRE_TOOL_USE, called) == { + "tool": "Bash", + "input": {"command": "ls"}, + } + assert harnessing.fields(HookKind.PRE_TOOL_USE, shown) == { + "tool": "Bash", + "input": {"about": "git push --force"}, + } + assert harnessing.fields( + HookKind.NOTIFICATION, _occasion(Moment.NOTIFICATION, said="Which way?") + ) == {"message": "Which way?"} + assert harnessing.fields( + HookKind.SUBAGENT_START, + _occasion(Moment.SUBAGENT_START, tool="Task", about="look around"), + ) == {"subagent": "Task", "task": "look around"} + + +def test_a_refusing_hook_is_a_refusing_verdict_and_nothing_else_is() -> None: + assert harnessing.verdict( + HookKind.PRE_TOOL_USE, PreToolUseHookResult(block=True, reason="no") + ) == Verdict(refused=True, because="no") + assert harnessing.verdict(HookKind.PRE_TOOL_USE, PreToolUseHookResult()) is None + assert harnessing.verdict( + HookKind.PERMISSION_REQUEST, PermissionRequestHookResult(allow=False) + ) == Verdict(refused=True, because="refused by a hook") + assert ( + harnessing.verdict(HookKind.PERMISSION_REQUEST, PermissionRequestHookResult()) + is None + ) + assert harnessing.verdict(HookKind.STOP, StopHookResult(block=True)) is None + assert harnessing.answer(AskUserHookResult(answer="left")) == "left" + assert harnessing.answer(StopHookResult()) is None + + +# --------------------------------------------------------------------------------- errors + + +def _failed(fault: str = "", status: int = 1) -> Failed: + return Failed(status, ["cli"], "", "it went wrong", fault=fault) + + +@pytest.mark.parametrize( + ("error", "leaf"), + [ + (_failed("contended"), HarnessContended), + (_failed("throttled"), HarnessThrottled), + (_failed("refused"), HarnessRefused), + (_failed("unlisted"), ModelUnavailable), + (_failed("retired"), ModelUnavailable), + (_failed("missing"), HarnessMissing), + (_failed("missing", 127), HarnessNotInstalled), + (_failed("sandboxed"), HarnessSandboxed), + (_failed("killed"), HarnessKilled), + (_failed("dropped"), HarnessDropped), + (_failed(), HarnessUnrecoverable), + (Unrecoverable(1, ["cli"], "", "too long"), HarnessUnrecoverable), + (NotInstalled("claude is not on the target"), HarnessNotInstalled), + (Stopped("coder was stopped"), SessionError), + (ConnectionResetError("reset"), HarnessDropped), + (RuntimeError("forked from somewhere else"), SessionError), + (ValueError("no such provider"), HarnessUnrecoverable), + ( + subprocess.CalledProcessError(1, ["cli"], "", "429 Too Many Requests"), + HarnessThrottled, + ), + ], + ids=lambda one: type(one).__name__ if isinstance(one, type) else repr(one)[:40], +) +def test_a_failed_turn_is_the_leaf_for_why( + error: BaseException, leaf: type[HarnessError] +) -> None: + said = harnessing.harness_error(error, "claude") + assert type(said) is leaf + assert str(said) + + +def test_a_bug_is_not_a_failed_turn() -> None: + assert harnessing.harness_error(KeyError("oops"), "claude") is None + + +# -------------------------------------------------------------------------------- answers + + +class Verdicted(pydantic.BaseModel): + """A shape an answer is read as.""" + + done: bool + why: str = "" + + +@pytest.mark.parametrize( + "said", + [ + '{"done": true}', + 'Here it is:\n```json\n{"done": true}\n```', + 'Sure. {"done": true} -- hope that helps', + ], +) +def test_an_answer_is_read_as_its_shape_wherever_in_it_the_object_is(said: str) -> None: + assert harnessing.read_shape(said, Verdicted) == Verdicted(done=True) + + +def test_an_answer_that_is_not_its_shape_is_an_output_schema_error() -> None: + with pytest.raises(OutputSchemaError) as raised: + harnessing.read_shape('{"finished": true}', Verdicted) + assert isinstance(raised.value, ValueError) + assert isinstance(raised.value.__cause__, pydantic.ValidationError) + + +def test_an_away_outworlder_answers_what_nothing_has_to_be_asked_for() -> None: + class Loose(pydantic.BaseModel): + far: int = 2 + + assert away_answer(None) == "" + assert away_answer(Loose) == Loose() + with pytest.raises(OutworlderAway): + away_answer(Verdicted) + + +# -------------------------------------------------------------------- coganchor's own parts + + +def test_claude_names_a_project_directory_as_claude_code_does() -> None: + long = "/tmp/" + "a-very-long-directory-name/" * 9 + "ü\U0001f600end" + assert _project("/home/me/my repo") == "-home-me-my-repo" + assert _project("/tmp/x/ü\U0001f600end") == "-tmp-x----end" + # What Claude Code 2.1.282's own function answers for the same path. + assert _project(long) == ( + "-tmp" + "-a-very-long-directory-name" * 7 + "-a-very-qz2e4t" + ) + + +def test_codex_is_held_to_a_schema_its_models_will_take() -> None: + class Inner(pydantic.BaseModel): + default: int = 1 + + class Outer(pydantic.BaseModel): + inner: Inner + tags: list[str] = pydantic.Field(default_factory=list[str]) + + held = strict(Outer.model_json_schema()) + assert held["required"] == ["inner", "tags"] + assert held["additionalProperties"] is False + inner = held["$defs"]["Inner"] + assert inner["required"] == ["default"] + assert "default" in inner["properties"] + assert "default" not in inner["properties"]["default"] + + +def test_codex_takes_only_an_approval_policy_it_has() -> None: + assert CodexAgentConfig(model="m", effort="low", approvals="untrusted").approvals + with pytest.raises(ValueError, match="approvals"): + CodexAgentConfig(model="m", effort="low", approvals="auto-review") + + +def test_a_conversation_is_forked_only_where_its_backend_can_carry_it( + tmp_path: Path, +) -> None: + claude, _ = driver("claude") + opencode, _ = driver("opencode") + held = claude(open_agent(_spec(HarnessKind.CLAUDE))._config).new(tmp_path) + other = opencode(open_agent(_spec(HarnessKind.OPENCODE))._config) + with pytest.raises(ValueError, match="same backend"): + held.fork(into=other) + with pytest.raises(RuntimeError): + held.fork(cwd=tmp_path / "elsewhere") # nothing has landed to carry + stays = other.new(tmp_path) + with pytest.raises(NotImplementedError): + stays.fork(cwd=tmp_path / "elsewhere") + + +async def test_a_change_of_hooks_is_settled_until_a_turn_has_settled_it( + tmp_path: Path, +) -> None: + made = open_agent(_spec(HarnessKind.CODEX, model="gpt-5.5", effort="low")) + made._installed = True + hooks = HookTable() + handle = await made.open( + Placement(EnvBackendKind.LOCAL, "", PurePosixPath(str(tmp_path))), + permission=EVERYTHING, + skills=(), + hooks=hooks, + ) + + async def refuses(one: Any, fields: dict[str, Any]) -> Any: + del one, fields + return PermissionRequestHookResult(allow=False) + + hooks.set(HookKind.PERMISSION_REQUEST, refuses) + assert handle._settling() is not None + # A turn that never got as far as settling leaves the next one to settle it. + settle = handle._settling() + assert settle is not None + settle() + config = handle.agent.config + assert isinstance(config, CodexAgentConfig) + assert config.approvals == "untrusted" + assert handle._settling() is None + await made.close() + + +async def test_a_tool_asked_about_while_the_turn_is_cut_off_is_refused( + tmp_path: Path, +) -> None: + made = open_agent(_spec(HarnessKind.CLAUDE)) + made._installed = True + hooks = HookTable() + + async def allows(one: Any, fields: dict[str, Any]) -> Any: + del one, fields + return PermissionRequestHookResult() + + hooks.set(HookKind.PERMISSION_REQUEST, allows) + handle = await made.open( + Placement(EnvBackendKind.LOCAL, "", PurePosixPath(str(tmp_path))), + permission=EVERYTHING, + skills=(), + hooks=hooks, + ) + asked = _occasion(Moment.PERMISSION_REQUEST, tool="Bash") + handle._flying = True + handle._interrupted = True + said = handle._moment(asked) + assert said is not None + assert said.refused + await made.close() diff --git a/tests/unit/flows/test_harness_table.py b/tests/unit/flows/test_harness_table.py index 93ffccd0f..e964ed1d4 100644 --- a/tests/unit/flows/test_harness_table.py +++ b/tests/unit/flows/test_harness_table.py @@ -54,7 +54,7 @@ HarnessKind.CURSOR_AGENT: ("CursorAgent", SUBAGENTS), HarnessKind.KIMI: ("KimiCodeAgent", {GOAL, STEER, PERMISSION, ASK}), HarnessKind.ZCODE: ("ZCodeAgent", {GOAL, PERMISSION, ASK}), - HarnessKind.GROK: ("GrokBuildAgent", {PERMISSION}), + HarnessKind.GROK: ("GrokBuildAgent", set()), HarnessKind.PI: ("PiAgent", {STEER, ASK}), HarnessKind.DSH: ("DeepSeekHarnessAgent", {GOAL}), HarnessKind.OPENCODE: ("OpenCodeAgent", set()),