From 1a4086a212556b371055bf5805bac9e8436a3075 Mon Sep 17 00:00:00 2001 From: Zijian Zhang <35801754+futrime@users.noreply.github.com> Date: Mon, 12 Oct 2026 23:01:47 +0800 Subject: [PATCH] feat(web): per-machine notices from host status A host's status now says the questions it waits on (`pending`), how the last run to end ended (`ended`), and the run's `epic` and `task`. `hmz web` polls every workspace's host status every 2 s, without attaching, and turns what changed into notices (done, failed, stopped, asked, listening), served at `GET /api/notices?after=` and as `notice` events on `GET /api/notices/stream`. Co-Authored-By: Claude Opus 5.5 --- docs/reference/daemon.md | 3 + docs/reference/web.md | 24 ++ src/hmz/runtime/doing/hosting.py | 36 ++- src/hmz/web/__init__.py | 8 +- src/hmz/web/notices.py | 279 +++++++++++++++++++++++ src/hmz/web/places.py | 43 +++- src/hmz/web/site.py | 9 +- tests/integration/test_web_notices.py | 146 ++++++++++++ tests/unit/runtime/doing/test_hosting.py | 51 +++++ tests/unit/web/test_notices.py | 274 ++++++++++++++++++++++ 10 files changed, 860 insertions(+), 13 deletions(-) create mode 100644 src/hmz/web/notices.py create mode 100644 tests/integration/test_web_notices.py create mode 100644 tests/unit/web/test_notices.py diff --git a/docs/reference/daemon.md b/docs/reference/daemon.md index fd903fb07..643c8c838 100644 --- a/docs/reference/daemon.md +++ b/docs/reference/daemon.md @@ -247,6 +247,9 @@ the daemon answers it ([Routing](#routing)). `Daemon.asked(said)` sends one with | `state` | string | `running`, `stopping` or `idle`. | | `run` | int | The run going, else the one stopping, else the last; `0` for none. | | `flow` | string | That run's flow, or `""`. | +| `epic`, `task` | string | That run's epic (its directory's name, `""` before it has one) and task, or `""`. | +| `pending` | list | `[{question, role, text, mode}]`: the questions waited on, as the [`pending` snapshot](#snapshots) says them. | +| `ended` | object \| null | How the last run to end here ended: `how` (`done`, `stopped`, `refused`, `budget`, `failed` or `crashed`), `why`, `epic`, `at` (Unix seconds), `run`, `flow`, `task`; `null` before one has. | | `budget`, `usage` | object \| null | As JSON, for the run going; `null` while stopping or idle. An infinite cost is the JSON string `"Infinity"`. | | `flows` † | list[string] | Canonical refs of the flow calls in progress. | | `calls` † | list | `[{ref, name, depth, seconds, id, parent}]`, oldest-started first across every run in the process; `seconds` rounded to 0.001; `parent` the index of the calling entry, or `null`. | diff --git a/docs/reference/web.md b/docs/reference/web.md index e4311393c..f5fbb4563 100644 --- a/docs/reference/web.md +++ b/docs/reference/web.md @@ -277,6 +277,30 @@ Every `error` but the last is the runtime's own sentence, e.g. An account is listed by the names of the variables it sets (`sets`), never by a value. +### Notices {#api-notices} + +What happened to this machine's runs worth telling somebody about. They are worked out of each +workspace's host [`status`](/reference/daemon#status), asked every **2 s** without attaching to +any host, so a host goes as it would with nobody reading. Anything already standing when +`hmz web` starts is not news. + +| Method and path | Takes | Answers | +| --- | --- | --- | +| `GET /api/notices` | `after` (an `id`; `0`, the default, for every one kept) | `notices` (those kept after it, oldest first; the last **200**), `last` (the newest `id`, `0` for none); `400` where `after` is not a whole number | +| `GET /api/notices/stream` | `Last-Event-ID` header, or `?last=` | `text/event-stream`: each `notice` event's `id` is its `id` and its data the notice, from after the one named (without one, from now on), then each as it comes; `: quiet` after **15 s** of nothing | + +A notice is `id` (a whole number, larger than any before it, on this machine too after +`hmz web` starts again), `kind`, `workspace`, `epic`, `flow`, `task`, `role`, `text` and `at` +(Unix seconds): + +| `kind` | When | `role`, `text` | +| --- | --- | --- | +| `done` | An epic ended done | `""`, `""` | +| `failed` | An epic ended failed, crashed, refused or over its budget | `""`, why | +| `stopped` | An epic was stopped | `""`, `""` | +| `asked` | A flow asks a person a question (mode `ask`) | The role asking, the question | +| `listening` | A flow waits for what to say next: a chat between turns (mode `listen`) | The role asking, what it asks | + ## The stream {#stream} `GET /api/held/stream` answers `text/event-stream`, starting with `retry: 2000`. Each event is diff --git a/src/hmz/runtime/doing/hosting.py b/src/hmz/runtime/doing/hosting.py index 8b1126544..a8fdda1e3 100644 --- a/src/hmz/runtime/doing/hosting.py +++ b/src/hmz/runtime/doing/hosting.py @@ -182,6 +182,12 @@ def _json(model: Any) -> Any: return json.loads(model.model_dump_json()) +def _epic_of(shown: _Run | None) -> str: + """The name of the epic a run is written into, or "" before it has one.""" + epic = shown.run.epic if shown is not None else None + return epic.name if epic is not None else "" + + def _spoken(given: Mapping[str, Any]) -> dict[str, str]: """What each role was given, as the words `-a` or `-e` would spell it.""" return { @@ -347,6 +353,8 @@ def __init__(self, hmz: Hmz) -> None: self._current: _Run | None = None self._stopping: _Run | None = None self._last: _Run | None = None + #: How the last run to end here ended, for whoever asks from outside the runs. + self._ended: dict[str, Any] | None = None self._runs = 0 self._starting = False self._questions: list[_Question] = [] @@ -569,14 +577,25 @@ def status(self) -> dict[str, Any]: Returns: How many frontends are attached and who, the state and number of the run, the - flow, what it may spend and has spent -- both None with nothing running -- and the - flows running, as refs and as calls. + flow, its epic and task, what it may spend and has spent -- both None with nothing + running -- the questions it waits on, how the last run to end ended, and the flows + running, as refs and as calls. """ with self._lock: running = self._current shown = running or self._stopping or self._last clients = self._clients_of() state = self._state() + pending = [ + { + "question": one.id, + "role": one.role, + "text": one.text, + "mode": one.mode, + } + for one in self._questions + ] + ended = dict(self._ended) if self._ended is not None else None calls = self._calls(now=time.monotonic()) return { "attached": len(clients), @@ -584,6 +603,10 @@ def status(self) -> dict[str, Any]: "state": state, "run": shown.number if shown is not None else 0, "flow": shown.run.flow if shown is not None else "", + "epic": _epic_of(shown), + "task": shown.started.get("task", "") if shown is not None else "", + "pending": pending, + "ended": ended, "budget": _json(running.run.budget) if running is not None else None, "usage": _json(running.run.usage) if running is not None else None, "flows": [one["ref"] for one in calls], @@ -1030,6 +1053,15 @@ def _drives(self, current: _Run) -> None: # arrive is the one it ended for. if not self._frontends and not current.stopped_by: self._keeps = True + self._ended = { + "how": how, + "why": why, + "epic": _epic_of(current), + "at": time.time(), + "run": current.number, + "flow": current.run.flow, + "task": current.started.get("task", ""), + } self._record( { "type": "ended", diff --git a/src/hmz/web/__init__.py b/src/hmz/web/__init__.py index 5c6d71970..9bddeef2b 100644 --- a/src/hmz/web/__init__.py +++ b/src/hmz/web/__init__.py @@ -59,6 +59,7 @@ def listening(*, port: int = 0, apart: bool = True) -> Generator[Site]: from hmz.coganchor import releases + from .notices import Notices from .places import Places from .runs import Runs from .site import Site @@ -67,7 +68,8 @@ def listening(*, port: int = 0, apart: bool = True) -> Generator[Site]: with contextlib.ExitStack() as held_open: places = Places(Path.cwd(), apart=apart) held_open.callback(places.close) - site = Site(port, places=places, runs=Runs(places)) + notices = Notices(places.statuses) + site = Site(port, places=places, runs=Runs(places), notices=notices) # What each index lists now, fetched once in the background as the interface starts. fetches(places.hmz) # And which release of humanize is the newest, where what is kept has got old. @@ -75,6 +77,10 @@ def listening(*, port: int = 0, apart: bool = True) -> Generator[Site]: held_open.callback(site.server_close) # Set first as the block ends, which is what ends every stream being sent. held_open.callback(site.going.set) + # And before it, what ends the notices' streams at once rather than once they wake. + held_open.callback(notices.close) + # What stands as it starts to answer is where the notices start from. + notices.start() yield site diff --git a/src/hmz/web/notices.py b/src/hmz/web/notices.py new file mode 100644 index 000000000..af4988c8c --- /dev/null +++ b/src/hmz/web/notices.py @@ -0,0 +1,279 @@ +"""What happened to the runs of this machine worth telling somebody about, as notices. + +An epic ending, a question for a person, a chat waiting for its next turn: each is a notice, +worked out of how every workspace's runs stand -- asked of each host every couple of seconds, +as `Places` asks what runs, and never by attaching to it, which would keep a host that would +otherwise go. What stood when this started is where it starts from rather than news. The last +few are kept, each numbered after the one before and after any this machine numbered before it +was started again, so that a page -- or another machine's interface -- carries on from the last +one it heard. +""" + +from __future__ import annotations + +import collections +import contextlib +import threading +import time +from typing import TYPE_CHECKING, Any + +from .routing import Refusal, Stream, routes + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + from .routing import Asked + +__all__ = ["ROUTES", "Notices"] + +#: How often how the runs stand is asked, in seconds. +_EVERY = 2.0 + +#: How many notices are kept. +_KEPT = 200 + +#: How long a stream says nothing before it says it is still there, in seconds. +_QUIET = 15.0 + +#: The kind of notice each way an epic can end is. +_ENDED = { + "done": "done", + "stopped": "stopped", + "failed": "failed", + "crashed": "failed", + "budget": "failed", + "refused": "failed", +} + +#: The kind of notice a question waited on is, by its mode. +_ASKED = {"ask": "asked", "listen": "listening"} + + +class Notices: + """The notices of this machine's runs, worked out of how they stand as that changes. + + Args: + statuses: How the runs of each workspace stand, by its path, as a host's `status` + says: asked once a look, and never of a host it would keep. + every: How often to look once started, in seconds. + """ + + def __init__( + self, + statuses: Callable[[], dict[str, dict[str, Any]]], + *, + every: float = _EVERY, + ) -> None: + self._statuses = statuses + self._every = every + self._changed = threading.Condition() + self._kept: collections.deque[dict[str, Any]] = collections.deque( + maxlen=_KEPT + ) + self._last = 0 + self._listeners: list[Callable[[dict[str, Any]], None]] = [] + #: How each workspace last stood, where its host said: what is news is against this. + self._stood: dict[str, dict[str, Any]] | None = None + #: When it first looked: an epic that ended before then is no news, wherever it is. + self._began = 0.0 + self._closed = threading.Event() + + @property + def last(self) -> int: + """The id of the newest notice, or 0 where there is none.""" + with self._changed: + return self._last + + def start(self) -> None: + """Looks every so often, on a thread of its own, until closed.""" + threading.Thread(target=self._looking, daemon=True, name="hmz-notices").start() + + def look(self) -> list[dict[str, Any]]: + """Asks how the runs stand, once, and tells what is new since the last look. + + The first look is where it starts from: nothing that stood then is news. + + Returns: + The notices that look made, oldest first. + """ + statuses = self._statuses() + made: list[dict[str, Any]] = [] + with self._changed: + first = self._stood is None + if first: + self._began = time.time() + stood = self._stood if self._stood is not None else {} + for workspace, status in statuses.items(): + # A host that would not answer says neither: unknown, rather than nothing. + if "pending" not in status or "ended" not in status: + continue + if not first: + was = stood.get(workspace, {}) + made += _noticed(workspace, was, status, self._began) + stood[workspace] = { + # Which host it was, apart: another one numbers its questions afresh. + "host": (status.get("pid"), status.get("started")), + "ended": status.get("ended"), + "pending": list(status.get("pending") or []), + } + self._stood = stood + for one in made: + self._last = one["id"] = max(self._last + 1, int(time.time() * 1000)) + self._kept.append(one) + if made: + self._changed.notify_all() + listeners = list(self._listeners) + for one in made: + for listener in listeners: + with contextlib.suppress(Exception): + listener(dict(one)) + return made + + def since(self, after: int | str = 0) -> list[dict[str, Any]]: + """The notices kept after one, oldest first. + + Args: + after: The id of the last one heard, or 0 for every one kept. + + Raises: + Refusal: If `after` is not a whole number. + """ + try: + heard = int(after or 0) + except ValueError as why: + raise Refusal("Say after as a whole number.") from why + with self._changed: + return [dict(one) for one in self._kept if one["id"] > heard] + + def heard(self, listener: Callable[[dict[str, Any]], None]) -> Callable[[], None]: + """Tells a listener each notice as it is made, on the thread that looks. + + Args: + listener: Told each notice; what it raises is not this one's. + + Returns: + What stops telling it. + """ + with self._changed: + self._listeners.append(listener) + + def unheard() -> None: + with self._changed, contextlib.suppress(ValueError): + self._listeners.remove(listener) + + return unheard + + def stream(self) -> Stream: + """Every notice after the last one a page heard, then each as it comes. + + A page that heard none hears only those made from now on. + """ + return Stream(self._events) + + def close(self) -> None: + """Stops looking, and ends every stream.""" + self._closed.set() + with self._changed: + self._changed.notify_all() + + def _events( + self, last: str, going: threading.Event + ) -> Iterator[tuple[str, str, Any] | None]: + # Where it starts from is as it is asked for, not as it is first read. + return self._after(int(last) if last.isdigit() else self.last, going) + + def _after( + self, after: int, going: threading.Event + ) -> Iterator[tuple[str, str, Any] | None]: + while not (going.is_set() or self._closed.is_set()): + with self._changed: + fresh = [one for one in self._kept if one["id"] > after] + if not fresh: + self._changed.wait(_QUIET) + fresh = [one for one in self._kept if one["id"] > after] + if not fresh: + yield None + continue + for one in fresh: + after = one["id"] + yield str(after), "notice", dict(one) + + def _looking(self) -> None: + while True: + with contextlib.suppress(Exception): + self.look() + if self._closed.wait(self._every): + return + + +def _noticed( + workspace: str, was: dict[str, Any], now: dict[str, Any], began: float +) -> list[dict[str, Any]]: + """What is news in how one workspace's runs stand now against how they stood. + + Args: + workspace: Whose runs. + was: How they last stood, or nothing where they were not seen before. + now: How they stand now. + began: When the first look was: an epic that ended before is no news. + """ + made: list[dict[str, Any]] = [] + ended = now.get("ended") + if ( + ended + and ended != was.get("ended") + and float(ended.get("at") or began) >= began + ): + why = str(ended.get("why") or "").strip() + made.append( + { + "kind": _ENDED.get(str(ended.get("how")), "failed"), + "workspace": workspace, + "epic": ended.get("epic") or "", + "flow": ended.get("flow") or now.get("flow") or "", + "task": ended.get("task") or now.get("task") or "", + "role": "", + # Why in a line: of a crash, the last of its traceback. + "text": why.splitlines()[-1] if why else "", + "at": ended.get("at") or time.time(), + } + ) + same = was.get("host") == (now.get("pid"), now.get("started")) + stood: list[dict[str, Any]] = (was.get("pending") or []) if same else [] + pending: list[dict[str, Any]] = now.get("pending") or [] + asked = {one.get("question") for one in stood} + for one in pending: + kind = _ASKED.get(one.get("mode", "")) + if kind is None or one.get("question") in asked: + continue + made.append( + { + "kind": kind, + "workspace": workspace, + "epic": now.get("epic") or "", + "flow": now.get("flow") or "", + "task": now.get("task") or "", + "role": one.get("role") or "", + "text": one.get("text") or "", + "at": time.time(), + } + ) + return made + + +def _notices(asked: Asked) -> dict[str, Any]: + notices: Notices = asked.site.notices + found = notices.since(asked.query.get("after", "0")) + return {"notices": found, "last": notices.last} + + +def _stream(asked: Asked) -> Stream: + notices: Notices = asked.site.notices + return notices.stream() + + +#: What a page reads of the notices. +ROUTES = routes( + ("GET", "/api/notices", _notices), + ("GET", "/api/notices/stream", _stream), +) diff --git a/src/hmz/web/places.py b/src/hmz/web/places.py index 129f2b39d..05df3a475 100644 --- a/src/hmz/web/places.py +++ b/src/hmz/web/places.py @@ -249,18 +249,43 @@ def _looking(self) -> None: if self._apart: self._asks_the_daemon() - def _asks_the_daemon(self) -> None: - """What the daemon says runs in each workspace no link here follows.""" - running: dict[str, dict[str, Any]] = {} + def statuses(self) -> dict[str, dict[str, Any]]: + """How the runs of every workspace held on this machine stand, attaching to none. + + Apart, what each host the daemon holds says of itself; otherwise, what the runs held + in this process say, of each workspace a page has reached. + + Returns: + Each host's status, by the whole path of its workspace. + """ + said: dict[str, dict[str, Any]] = {} + if not self._apart: + with self._lock: + hosts = [ + (str(one.workspace), one.hosting) + for one in self._places.values() + if one.hosting is not None + ] + for workspace, hosting in hosts: + with contextlib.suppress(Exception): + said[workspace] = hosting.status() + return said with contextlib.suppress(Exception): for one in self._daemon.daemons(): with contextlib.suppress(Exception): - status = one.status() - running[str(Path(one.workspace).resolve())] = { - "state": status.get("state") or "idle", - "run": status.get("run") or 0, - "flow": status.get("flow") or "", - } + said[str(Path(one.workspace).resolve())] = one.status() + return said + + def _asks_the_daemon(self) -> None: + """What the daemon says runs in each workspace no link here follows.""" + running = { + workspace: { + "state": status.get("state") or "idle", + "run": status.get("run") or 0, + "flow": status.get("flow") or "", + } + for workspace, status in self.statuses().items() + } with self._lock: self._running = running diff --git a/src/hmz/web/site.py b/src/hmz/web/site.py index e6204457b..ee4160750 100644 --- a/src/hmz/web/site.py +++ b/src/hmz/web/site.py @@ -42,6 +42,7 @@ fallbacks, flows, held, + notices, paths, places, runs, @@ -54,6 +55,7 @@ if TYPE_CHECKING: from collections.abc import Generator + from .notices import Notices from .places import Places from .runs import Runs @@ -71,6 +73,7 @@ *paths.ROUTES, *places.ROUTES, *verses.ROUTES, + *notices.ROUTES, ) #: How many ports any would do for are tried before one free on both loopbacks is given up on. @@ -131,12 +134,15 @@ class Site(ThreadingHTTPServer): port: The port to listen on, or 0 for any free one on both. places: Every workspace on this machine, and the runs held for each as this follows them. runs: The runs written down, of every workspace. + notices: What happened to the runs worth telling somebody about. """ daemon_threads = True allow_reuse_address = True - def __init__(self, port: int, *, places: Places, runs: Runs) -> None: + def __init__( + self, port: int, *, places: Places, runs: Runs, notices: Notices + ) -> None: self._beside: _Beside | None = None for left in reversed(range(_PORTS)): super().__init__(("127.0.0.1", port), _Handler) @@ -152,6 +158,7 @@ def __init__(self, port: int, *, places: Places, runs: Runs) -> None: self.places = places self.hmz: Any = places.hmz self.runs = runs + self.notices = notices #: Set as the server stops, which is what ends every stream it is sending. self.going = threading.Event() self._key = secrets.token_urlsafe(32) diff --git a/tests/integration/test_web_notices.py b/tests/integration/test_web_notices.py new file mode 100644 index 000000000..ad1cf26e2 --- /dev/null +++ b/tests/integration/test_web_notices.py @@ -0,0 +1,146 @@ +"""`hmz web`: an epic ending, and a chat waiting on its next turn, said as notices.""" + +from __future__ import annotations + +import contextlib +import http.client +import json +import time +from typing import TYPE_CHECKING, Any + +import pytest + +from tests.integration.doubles_core import AGENT, install, write_flow +from tests.integration.doubles_web import PATIENCE, Browser, served + +if TYPE_CHECKING: + from collections.abc import Callable, Generator + from pathlib import Path + +#: A flow of one turn of the stand-in, which ends done. +_ONCE = ''' +from hmz.flows import ( + Agent, AgentCollection, EnvCollection, FlowContext, FlowParams, LocalEnv, flow, +) + + +class Agents(AgentCollection): + """The one.""" + + solo: Agent + + +class Envs(EnvCollection): + """Where it works.""" + + workspace: LocalEnv + + +@flow(agents=Agents, envs=Envs, params=FlowParams) +async def once( + task: str, *, agents: Agents, envs: Envs, params: FlowParams, ctx: FlowContext +) -> str: + """One turn, and done.""" + one = agents["solo"] + return await one.run(task, session=await one.spawn(), env=envs["workspace"]) +''' + + +@pytest.fixture +def project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """The stand-in `claude` alone on PATH, and a project to serve the runs of.""" + return install(tmp_path, monkeypatch) + + +@contextlib.contextmanager +def _follows( + browser: Browser, last: str = "" +) -> Generator[Callable[[], tuple[str, dict[str, Any]]]]: + """Listens to the notices' stream as a page does: each notice, by its id.""" + connection = http.client.HTTPConnection( + "127.0.0.1", browser.site.port, timeout=PATIENCE + ) + headers = {"Host": f"127.0.0.1:{browser.site.port}", "Cookie": browser.cookie} + if last: + headers["Last-Event-ID"] = last + connection.request("GET", "/api/notices/stream", headers=headers) + stream = connection.getresponse() + + def next_notice() -> tuple[str, dict[str, Any]]: + ident = event = "" + while True: + line = stream.readline().decode().rstrip("\n") + if line.startswith("id: "): + ident = line[len("id: ") :] + elif line.startswith("event: "): + event = line[len("event: ") :] + elif line.startswith("data: ") and event == "notice": + return ident, json.loads(line[len("data: ") :]) + + try: + yield next_notice + finally: + connection.close() + + +def _until( + next_notice: Callable[[], tuple[str, dict[str, Any]]], kind: str +) -> tuple[str, dict[str, Any]]: + ending = time.monotonic() + PATIENCE + while time.monotonic() < ending: + ident, notice = next_notice() + if notice["kind"] == kind: + return ident, notice + raise AssertionError(f"no {kind} notice came") + + +def test_a_chat_waiting_and_an_epic_ending_are_notices(project: Path) -> None: + write_flow(project / ".hmz" / "flows", "once", _ONCE) + with served() as site: + browser = Browser(site) + browser.signs_in() + assert browser.get("/api/notices").json() == {"notices": [], "last": 0} + + with _follows(browser) as heard: + chat = {"flow": "chat", "task": "hello there", "agents": {"assistant": AGENT}} + assert browser.post("/api/held/start", chat).status == 200 + _, listening = _until(heard, "listening") + assert (listening["workspace"], listening["flow"]) == (str(project), "chat") + assert (listening["task"], listening["role"]) == ("hello there", "human") + assert listening["epic"] + + assert browser.post("/api/held/stop").status == 200 + stopped_at, stopped = _until(heard, "stopped") + assert stopped["epic"] == listening["epic"] + + started = browser.post( + "/api/held/start", + { + "flow": "@local/once", + "task": "do it", + "agents": {"solo": AGENT}, + "budget": {"cost": 5}, + }, + ) + assert started.status == 200, started.json() + _, done = _until(heard, "done") + assert (done["flow"], done["task"], done["text"]) == ( + "@local/once", + "do it", + "", + ) + assert done["epic"] != listening["epic"] + + kept = browser.get("/api/notices").json() + assert [one["kind"] for one in kept["notices"]] == [ + "listening", + "stopped", + "done", + ] + assert kept["last"] == kept["notices"][-1]["id"] + after = browser.get(f"/api/notices?after={stopped_at}").json()["notices"] + assert [one["kind"] for one in after] == ["done"] + assert browser.get("/api/notices?after=soon").status == 400 + # Carried on from the last one heard, a stream hears what came after it. + with _follows(browser, stopped_at) as heard: + assert heard()[1]["kind"] == "done" diff --git a/tests/unit/runtime/doing/test_hosting.py b/tests/unit/runtime/doing/test_hosting.py index 16b0843c0..0b12a5132 100644 --- a/tests/unit/runtime/doing/test_hosting.py +++ b/tests/unit/runtime/doing/test_hosting.py @@ -1039,6 +1039,57 @@ def test_a_question_is_put_to_everybody_and_answered_by_number( run.released.set() +def test_the_status_says_the_questions_waited_on_and_how_the_last_run_ended( + host: Host, where: Workspace, outworlders: Outworlders, tmp_path: Path +) -> None: + alice = Told(host, "alice") + where.next_run = FakeRun(epic=tmp_path / "20260101-ralph") + run = _started(alice, where) + assert outworlders.ask is not None + said = host.status() + assert (said["epic"], said["task"], said["pending"], said["ended"]) == ( + "20260101-ralph", + "fix it", + [], + None, + ) + + asking = Asking()(outworlders.ask, Question("which?", ("red", "blue"))) + asked = alice.told("asked", run=1) + + assert host.status()["pending"] == [ + { + "question": asked["question"], + "role": "outworlder", + "text": "which?", + "mode": "ask", + } + ] + alice.asks("answer", question=asked["question"], text="1") + assert asking.done.wait(PATIENCE) + run.raises = BudgetExceeded("spent") + run.released.set() + alice.told("ended", run=1) + said = host.status() + assert said["pending"] == [] + ended = said["ended"] + assert ( + ended["how"], + ended["epic"], + ended["run"], + ended["flow"], + ended["task"], + ) == ( + "budget", + "20260101-ralph", + 1, + "ralph", + "fix it", + ) + assert "spent" in ended["why"] + assert ended["at"] > 0 + + def test_a_question_of_a_claimed_role_is_its_claimants( host: Host, where: Workspace, outworlders: Outworlders ) -> None: diff --git a/tests/unit/web/test_notices.py b/tests/unit/web/test_notices.py new file mode 100644 index 000000000..7d28dba53 --- /dev/null +++ b/tests/unit/web/test_notices.py @@ -0,0 +1,274 @@ +"""`hmz.web.notices`: what is news in how the runs stand, against how they stood.""" + +from __future__ import annotations + +import threading +import time +from typing import Any + +import pytest + +from hmz.web.notices import Notices +from hmz.web.routing import Refusal + + +class Statuses: + """How the runs stand, as each look is told: changed by the test between looks.""" + + def __init__(self) -> None: + self.now: dict[str, dict[str, Any]] = {} + + def __call__(self) -> dict[str, dict[str, Any]]: + return {where: dict(status) for where, status in self.now.items()} + + def stands( + self, + workspace: str = "/w", + *, + pending: list[dict[str, Any]] | None = None, + ended: dict[str, Any] | None = None, + ) -> None: + self.now[workspace] = { + "state": "running", + "flow": "ralph", + "epic": "e1", + "task": "fix it", + "pending": pending or [], + "ended": ended, + } + + +def _ended(how: str, at: float | None = None, why: str = "") -> dict[str, Any]: + """An epic that ended as `how`, at `at` -- or now.""" + return { + "how": how, + "why": why, + "epic": "e1", + "at": time.time() if at is None else at, + "flow": "ralph", + "task": "t", + } + + +def _asked(question: str, mode: str = "ask") -> dict[str, Any]: + return {"question": question, "role": "outworlder", "text": "which?", "mode": mode} + + +@pytest.fixture +def statuses() -> Statuses: + return Statuses() + + +@pytest.fixture +def notices(statuses: Statuses) -> Notices: + return Notices(statuses) + + +def test_what_stood_at_the_start_is_no_news( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands(pending=[_asked("q1")], ended=_ended("done")) + + assert notices.look() == [] + assert notices.look() == [] + assert notices.since() == [] + assert notices.last == 0 + + +@pytest.mark.parametrize( + ("how", "kind"), + [ + ("done", "done"), + ("stopped", "stopped"), + ("failed", "failed"), + ("crashed", "failed"), + ("budget", "failed"), + ("refused", "failed"), + ], +) +def test_an_epic_ending_is_a_notice_of_how( + how: str, kind: str, statuses: Statuses, notices: Notices +) -> None: + statuses.stands() + notices.look() + statuses.stands(ended=_ended(how, at=time.time() + 1, why="because")) + + (made,) = notices.look() + + assert made["kind"] == kind + assert (made["workspace"], made["epic"], made["flow"], made["task"]) == ( + "/w", + "e1", + "ralph", + "t", + ) + assert (made["role"], made["text"]) == ("", "because") + assert made["at"] > time.time() + assert made["id"] == notices.last + assert notices.look() == [] + + +def test_another_epic_ending_after_is_another_notice( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands(ended=_ended("done")) + notices.look() + statuses.stands(ended=_ended("failed", at=time.time() + 1)) + + assert [one["kind"] for one in notices.look()] == ["failed"] + + +def test_a_question_is_asked_and_a_chat_waiting_is_listening( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands(pending=[_asked("q1")]) + notices.look() + statuses.stands( + pending=[_asked("q1"), _asked("q2"), _asked("q3", "listen"), _asked("q4", "x")] + ) + + made = notices.look() + + assert [one["kind"] for one in made] == ["asked", "listening"] + assert (made[0]["role"], made[0]["text"], made[0]["epic"]) == ( + "outworlder", + "which?", + "e1", + ) + + +def test_a_workspace_first_seen_later_is_news( + statuses: Statuses, notices: Notices +) -> None: + notices.look() + statuses.stands("/v", ended=_ended("done")) + + assert [one["workspace"] for one in notices.look()] == ["/v"] + + +def test_an_epic_that_ended_before_the_start_is_no_news_wherever_it_is_seen( + statuses: Statuses, notices: Notices +) -> None: + notices.look() + statuses.stands("/v", ended=_ended("done", at=1.0)) + + assert notices.look() == [] + + +def test_a_crash_is_said_in_the_last_line_of_why( + statuses: Statuses, notices: Notices +) -> None: + notices.look() + statuses.stands(ended=_ended("crashed", why="Traceback:\n here\nValueError: no\n")) + + assert notices.look()[0]["text"] == "ValueError: no" + + +def test_another_host_asking_a_question_of_the_same_id_is_news( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands(pending=[_asked("q1")]) + statuses.now["/w"] |= {"pid": 1, "started": "then"} + notices.look() + statuses.stands(pending=[_asked("q1")]) + statuses.now["/w"] |= {"pid": 2, "started": "now"} + + assert [one["kind"] for one in notices.look()] == ["asked"] + + +def test_a_host_that_would_not_answer_is_not_taken_for_one_that_went( + statuses: Statuses, notices: Notices +) -> None: + ended = _ended("done", at=time.time() + 1) + statuses.stands() + notices.look() + statuses.stands(ended=ended) + notices.look() + statuses.now["/w"] = {"attached": 0, "flows": [], "calls": []} + notices.look() + statuses.stands(ended=ended) + + assert notices.look() == [] + + +def test_notices_are_numbered_and_read_after_the_last_heard( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands() + notices.look() + statuses.stands(pending=[_asked("q1")]) + (first,) = notices.look() + statuses.stands(pending=[_asked("q2")]) + (second,) = notices.look() + + assert second["id"] > first["id"] + assert [one["id"] for one in notices.since()] == [first["id"], second["id"]] + assert notices.since(str(first["id"])) == [second] + assert notices.since(second["id"]) == [] + with pytest.raises(Refusal): + notices.since("soon") + + +def _fails(_one: dict[str, Any]) -> None: + raise RuntimeError + + +def test_a_listener_hears_each_notice_until_it_stops( + statuses: Statuses, notices: Notices +) -> None: + heard: list[dict[str, Any]] = [] + statuses.stands() + notices.look() + unheard = notices.heard(heard.append) + notices.heard(_fails) + statuses.stands(pending=[_asked("q1")]) + notices.look() + unheard() + statuses.stands(pending=[_asked("q2")]) + notices.look() + + assert [one["kind"] for one in heard] == ["asked"] + + +def test_only_the_last_two_hundred_are_kept( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands() + notices.look() + for one in range(205): + statuses.stands(pending=[_asked(f"q{one}")]) + notices.look() + + assert len(notices.since()) == 200 + + +def test_a_stream_carries_on_after_the_last_heard( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands() + notices.look() + statuses.stands(pending=[_asked("q1")]) + (first,) = notices.look() + going = threading.Event() + events = notices.stream().events("0", going) + + assert next(events) == (str(first["id"]), "notice", first) + statuses.stands(pending=[_asked("q2")]) + (second,) = notices.look() + assert next(events) == (str(second["id"]), "notice", second) + notices.close() + assert list(events) == [] + + +def test_a_stream_heard_from_nothing_starts_from_now( + statuses: Statuses, notices: Notices +) -> None: + statuses.stands() + notices.look() + statuses.stands(pending=[_asked("q1")]) + notices.look() + events = notices.stream().events("", threading.Event()) + statuses.stands(pending=[_asked("q2")]) + (second,) = notices.look() + + assert next(events) == (str(second["id"]), "notice", second)