Watches and growing from a gap: trees that run when something speaks, and the branch a gap asks for drafted by rules; release 0.9.0
Claude · 2026-10-26 · 96f675e6ffef931661d8b8dd3ce6e9f0c7b1b610 · parents 032023b
A tree may carry [[watches]]: a clock (every = "15m"), a file (each time lines are appended), an MQTT topic (each message) or a serial line (each line) starts a run, with inputs rendered from the event, env and now. watch.py holds one thread per watch over a service's trees and runs them through the service's own runner, so gaps are counted and delegations route; the file is polled, the queue and the serial line are held open as streams (adapters.mqtt_stream with pings, adapters.serial_lines). The trace begins with a watch step, the way an asked run's begins with intent. Each watch remembers what it last did and the path the run took; when the path of a repeated run differs from the one before, the watch records that: a site going from up to erroring on a clock is learned before anyone asks. --watch on a service or hub, pyrtan watch DIR alone, /watches and pyrtan watches --at URL, and a Watches tab in the console. Watches are static and checked against the tree's lists by pyrtan check; a hosted service (--egress public) watches no file and no serial line. The watcher follows the files on disk: a tree that appears starts its watches, one that vanishes stops them.
pyrtan grow reads gaps (a running service or hub, a --gaps file, a /gaps answer, or the federation's reports for your catalog trees) and drafts the branch each asks for, by rules: { exists = false } for a missing value, the range of the hole a number fell into between the ranges the node already has, { type = ... } for a value a report could only carry as its type, { eq = value } otherwise, plus a fail node that says what was seen. --apply writes the branch before the node's else and the node at the end of the file; the file must check clean afterwards or it is put back. A person then says what the value means.
Unanswered sentences persist with --unanswered FILE ($PYRTAN_UNANSWERED) on a service or hub. The site-status and room-temperature examples carry a watch each. Tests: the checker's refusals, a file watch (append, rotation, path change), a clock watch (path change, sync), a queue watch against the fake broker, a serial watch over a pty, egress, /watches on a service and through a hub, the commands, persistence, every drafting rule, apply on multi-line and one-line lists with revert, the grow command from a file, a report and a running service.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YChr78vDHtQJ15HNbsaRLZ
Changes
CONTRIBUTING.md | 6 +- README.md | 66 ++++++- _meta/design.md | 21 +- _meta/io.md | 10 +- _meta/tree-format.md | 27 +++ deploy/openmediavault/docker-compose.yml | 2 + examples/room-temperature/room-temperature.toml | 7 + examples/site-status/site-status.toml | 7 + pyproject.toml | 2 +- src/pyrtan/__init__.py | 2 +- src/pyrtan/adapters.py | 122 ++++++++++++ src/pyrtan/catalog.py | 2 +- src/pyrtan/cli.py | 210 +++++++++++++++++++- src/pyrtan/grow.py | 208 +++++++++++++++++++- src/pyrtan/hub.py | 32 +++- src/pyrtan/js/pyrtan.js | 7 +- src/pyrtan/page.py | 23 +++ src/pyrtan/service.py | 31 ++- src/pyrtan/transport.py | 2 +- src/pyrtan/tree.py | 96 +++++++++- src/pyrtan/watch.py | 296 ++++++++++++++++++++++++++++ tests/conformance/vectors.json | 24 ++- tests/test_adapters.py | 11 ++ tests/test_watch_grow.py | 455 ++++++++++++++++++++++++++++++++++++++++++++ uv.lock | 2 +- 25 files changed, 1626 insertions(+), 45 deletions(-)
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 904012f..69321da 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md@@ -45,7 +45,11 @@ the same day. See [`catalog/README.md`](catalog/README.md) andOnce it is out there, `pyrtan reports <namespace>/my-tree` shows the gaps hubs that opted in have met with it: the node, the value no branch-covered, how often. That is the list of branches to write next.+covered, how often. That is the list of branches to write next, and+`pyrtan grow --reports --trees . --apply` drafts each one into the file+by rules (the range of the hole a number fell into, `{ eq = … }`+otherwise, a `fail` node to finish); you say what the value means and+bump the version.## Changing the runtime diff --git a/README.md b/README.md index 353753c..87fa026 100644 --- a/README.md +++ b/README.md@@ -331,6 +331,38 @@ tree and splits into many without the callers changing. A tree thatroutes it to whichever peer can; the peer's steps come back inside the caller's trace.+## Watches: trees that run on their own++A tree may carry `[[watches]]`: a clock, a file, a topic or a serial line+starts a run without anyone asking, with `inputs` rendered from what+spoke (`event`), `env` and `now`.++```toml+[[watches]]+name = "example.com"+every = "15m"+inputs = { host = "example.com" }++[[watches]]+file = "/var/log/syslog" # each time lines are appended+inputs = { text = "{{ event.body | strip }}" }++[[watches]]+queue = { url = "mqtt://broker.local", topic = "home/+/temperature" } # each message+```++`pyrtan serve trees --watch` (or `pyrtan hub --trees trees --watch`) runs+them beside the service; `pyrtan watch trees` runs them alone and prints+each run. A watched run's trace begins with a `watch` step, the way an+asked run's begins with `intent`. `/watches` and `pyrtan watches --at URL`+show each watch, what it last did, and, for a watch that runs the same+tree again and again, whether the **path changed** between two runs:+`site-status` going from `up` to `erroring` on a clock is how you learn a+site changed before anyone asks. Watches are static: the path, topic or+device is literal in the file and checked against the tree's lists by+`pyrtan check`, and a service started `--egress public` watches no file+and no serial line.+## When no service in the swarm answers A sentence nobody answers is not a dead end; it is the next tree. The hub@@ -357,6 +389,27 @@ directory picks the file up the moment it is saved. No model draftsanything: the proposal is derived from the words by rules, and the demo page shows the same proposal and the same command whenever nothing answers.+A **gap** gets the same treatment one level down. `pyrtan grow` reads+gaps (from a running service or hub, a `--gaps` file, or the federation's+reports for your catalog trees) and drafts, by rules, the branch each one+asks for: `{ exists = false }` for a missing value, the range of the hole+a number fell into between the ranges the node already has, `{ eq = … }`+otherwise, and a new `fail` node that says what was seen. `--apply`+writes it into the file, which must still check clean; the person then+says what the value means.++```+$ pyrtan grow --at http://127.0.0.1:8700 --trees trees --apply+# weather: in [nodes.sky], add to branches (before the else, if any):+ { when = { between = [68, 70] }, then = "sky-70" },+...+# why: the value falls in the hole between 67 and 71, which the ranges around it leave open+# written into trees/weather.toml; now edit [nodes.sky-70] to say what the value means+```++Unanswered sentences survive a restart too, with `--unanswered FILE` on a+service or hub.+## Sentences Pyrtan reads a sentence with rules, not a model, and the trace's first@@ -428,17 +481,18 @@ site/ pyrtan.com: the landing page, the seed list, the catalog## Status-0.8.0. The core, the service, the hub, the command, the console, the+0.9.0. The core, the service, the hub, the command, the console, thecatalog (namespaces, publishing by pull request through the gate, gap reports gossiped back to authors), the language rules, federation between hubs, the gossiped directory, members (a browser or a phone as part of the swarm), the blobs they keep and the trees they run-(`pyrtan.js`, conformant with the Python runtime), and six forms of I/O+(`pyrtan.js`, conformant with the Python runtime), six forms of I/O(HTTP, a browser, files, programs, serial lines, MQTT queues), each-behind an allow-list, are built and tested; a swarm can require a token,-keep its gaps, refuse private addresses and its own box, and follow-other swarms. Not yet built: persistent storage for unanswered-sentences, `pyrtan watch`, and hosting.+behind an allow-list, watches (a tree that runs when a clock, a file, a+topic or a serial line speaks), and `pyrtan grow` (the branch a gap asks+for, drafted by rules) are built and tested; a swarm can require a+token, keep its gaps and unanswered sentences, refuse private addresses+and its own box, and follow other swarms. Not yet built: hosting.See [`_meta/design.md`](_meta/design.md) for the runtime's design and [`_meta/platform.md`](_meta/platform.md) for pyrtan.com. diff --git a/_meta/design.md b/_meta/design.md index 5341a4b..494c4bb 100644 --- a/_meta/design.md +++ b/_meta/design.md@@ -274,10 +274,9 @@ Five trees, all against real public APIs, all tested offline:gaps persisted to a file per service, an opt-in egress policy, the console with a peers view off `/peers`, and the catalog (`pyrtan search`, `pull`, `index`) with `version` pinning a pull. Design in- [`platform.md`](platform.md). Still to come from the original v0.2: a- `pyrtan watch` that re-runs a set of sentences on a schedule and- reports when a path changes (the way to learn a site has changed before- a person does).+ [`platform.md`](platform.md). The `pyrtan watch` the original v0.2+ promised (re-run on a schedule, report when a path changes) came in+ v0.9, as `[[watches]]`.3. **v0.3, built: sentences read by rules, and hubs that follow each other.** The language rules and typed slots (§3, "Reading a sentence"), and federation in the shape of ActivityPub: a hub follows@@ -313,8 +312,18 @@ Five trees, all against real public APIs, all tested offline:(`--report-gaps`) carried as a federation topic like the directory, baked hourly into pyrtan.com and shown beside each catalog tree. Design in [`platform.md`](platform.md) §2.8.-9. **Beside the swarm, never inside it:** an authoring aid that reads gaps- and drafts the branch for a person to accept.+9. **v0.9, built: watches, and growing from a gap.** A tree carries+ `[[watches]]`: a clock, a file, a topic or a serial line starts a run+ (`watch.py`; `--watch` on a service or hub, `pyrtan watch` alone), the+ trace begins with a `watch` step, and `/watches` says what each last+ did and whether the path changed between runs. Beside the swarm, never+ inside it: `pyrtan grow` reads gaps (a service, a file, the+ federation's reports) and drafts the branch each asks for by rules+ (`grow.py`: a missing value, the hole between ranges, else the value),+ `--apply` writes it into the file, and a person says what it means.+ Unanswered sentences persist with `--unanswered FILE`.+10. **Later:** hosting (`platform.md` Phase 3); a serial line held open+ across the nodes of one run, for devices with a session.## 7. Open questions diff --git a/_meta/io.md b/_meta/io.md index 4e20b2e..a2aa15e 100644 --- a/_meta/io.md +++ b/_meta/io.md@@ -132,7 +132,9 @@ of them being present.- **v0.7, this:** the five sources, their lists, their fixtures, the examples, the JavaScript runtime's own HTML reader (so css extracts agree on a service, in a browser and in node).-- **Later:** a `watch` that runs a tree when a file, a topic or a serial- line speaks (the adapters are ready for it; the trigger is the design- still owed), and a serial line held open across nodes for devices with- a session.+- **v0.9, built:** `[[watches]]`, a tree that runs when a file, a topic+ or a serial line speaks, or on a clock (`watch.py`; the file is+ polled, the queue and the serial line are held open as streams,+ `adapters.mqtt_stream` and `adapters.serial_lines`).+- **Later:** a serial line held open across the nodes of one run, for+ devices with a session.diff --git a/_meta/tree-format.md b/_meta/tree-format.md index f92d95c..35b714a 100644 --- a/_meta/tree-format.md +++ b/_meta/tree-format.md@@ -19,6 +19,7 @@ A tree is one TOML file. This is the whole of it.| `license` | for a catalog | An SPDX identifier such as `MIT`, `Apache-2.0` or `CC0-1.0`: what others may do with the file. A tree without one runs anywhere but is left out of a catalog. | | `namespace` | for a catalog | Who publishes the tree (lowercase letters, digits, dashes). The catalog id is `namespace/name`; a hub reports a tree's gaps to its author only when it has one. | | `triggers` | no | Sentences the tree answers to (below). |+| `watches` | no | When the tree runs on its own: a clock, a file, a topic or a serial line (below). || `inputs` | no | What the tree needs (below). | | `nodes` | yes | The nodes (below). |@@ -103,6 +104,32 @@ word; a word close to two is left alone, because choosing would be a guess.The trace shows every correction (`typo: wether->weather`). A hub applies a tree's `[language]` when it routes, because `/capabilities` carries it.+## Watches++```toml+[[watches]]+name = "example.com" # optional; shown by pyrtan watches+every = "15m" # a duration, or seconds+inputs = { host = "example.com" } # templates may read event, env and now++[[watches]]+file = "/var/log/syslog" # fires with the lines appended since last time+inputs = { text = "{{ event.body | strip }}", n = "{{ event.lines | length }}" }++[[watches]]+queue = { url = "mqtt://broker.local", topic = "home/+/temperature" } # fires per message: event.topic, event.body, event.json++[[watches]]+serial = { device = "/dev/ttyUSB0", baud = 9600, read_until = "\n" } # fires per line: event.body+```++Exactly one of `every`, `file`, `queue`, `serial`. A watch is static:+no template reaches a path, a topic or a device, and `pyrtan check`+holds each against the tree's `paths`, `hosts` and `topics`, `devices`.+Every required input must be given or have a default. A watched run's+trace begins with a `watch` step. `--watch` on a service or hub runs+them; `pyrtan watch DIR` runs them alone.+## Inputs ```toml diff --git a/deploy/openmediavault/docker-compose.yml b/deploy/openmediavault/docker-compose.yml index ee04cda..6e3bc03 100644 --- a/deploy/openmediavault/docker-compose.yml +++ b/deploy/openmediavault/docker-compose.yml@@ -40,6 +40,8 @@ x-pyrtan: &pyrtan# PYRTAN_FOLLOW: "seed" # PYRTAN_LISTED: "true" # PYRTAN_REPORT_GAPS: "true" # opt in: the gaps your catalog trees meet (tree, node, value; never inputs) reach their authors via pyrtan.com+ # PYRTAN_WATCH: "true" # run the trees' [[watches]]: a clock, a file, a topic or a serial line starts a run+ PYRTAN_UNANSWERED: /state/unanswered.jsonl # sentences nobody answered, kept across restarts (the hub's; services with a hub leave the record to it)PYRTAN_STATE: /state/federation.json PYRTAN_STORE: /state/store # blobs: the hub's own and the ones it keeps for others PYRTAN_REPLICAS: "2" # how many members or hubs should keep each of its blobs diff --git a/examples/room-temperature/room-temperature.toml b/examples/room-temperature/room-temperature.toml index 1b5a2a6..98771dd 100644 --- a/examples/room-temperature/room-temperature.toml +++ b/examples/room-temperature/room-temperature.toml@@ -23,6 +23,13 @@ type = "str"default = "kitchen" pattern = "^[a-z0-9-]+$"+# Also on its own: every reading the kitchen's sensor publishes runs the+# tree (the run then takes the next message on the topic; a watch on the+# same topic as the tree reads is the simplest shape, not the only one).+[[watches]]+queue = { url = "mqtt://broker.local", topic = "home/kitchen/temperature" }+inputs = { room = "kitchen" }+[nodes.wait] kind = "observe" queue = { url = "mqtt://broker.local", topic = "home/{{ inputs.room }}/temperature" } diff --git a/examples/site-status/site-status.toml b/examples/site-status/site-status.toml index f3f4a1e..3caf8d5 100644 --- a/examples/site-status/site-status.toml +++ b/examples/site-status/site-status.toml@@ -23,6 +23,13 @@ type = "str"pattern = "(?i)^[a-z0-9.\\-]+(:[0-9]+)?$" description = "A host name, with no scheme"+# On its own, every quarter hour, for one host; `pyrtan watches` shows the+# last run and whether the path changed since the run before (up -> down).+[[watches]]+name = "example.com"+every = "15m"+inputs = { host = "example.com" }+[nodes.probe] kind = "observe" http = { method = "GET", url = "https://{{ inputs.host | lower }}/" } diff --git a/pyproject.toml b/pyproject.toml index 52fd365..165fe23 100644 --- a/pyproject.toml +++ b/pyproject.toml@@ -1,6 +1,6 @@[project] name = "pyrtan"-version = "0.8.0"+version = "0.9.0"description = "Open-source total automation: a core set of expandable services that act on all forms of I/O through hardcoded, inspectable decision trees." readme = "README.md" license = "AGPL-3.0-or-later" diff --git a/src/pyrtan/__init__.py b/src/pyrtan/__init__.py index 745db83..2d7a756 100644 --- a/src/pyrtan/__init__.py +++ b/src/pyrtan/__init__.py@@ -33,4 +33,4 @@ __all__ = ["Response", ]-__version__ = "0.8.0"+__version__ = "0.9.0"diff --git a/src/pyrtan/adapters.py b/src/pyrtan/adapters.py index a2ecf90..ef222a9 100644 --- a/src/pyrtan/adapters.py +++ b/src/pyrtan/adapters.py@@ -678,3 +678,125 @@ def _version() -> str:from . import __version__ return __version__+++# ---------------------------------------------------------------- streams, for watches+# A watch holds a line open and hands each message or line to a callback+# until told to stop. Errors are reported through the callback's twin and+# the caller reconnects; nothing here retries on its own.+++def mqtt_stream(spec: dict[str, Any], on_message: Callable[[dict[str, Any]], None], stop: threading.Event, egress: str = "any", timeout: float = 10.0) -> None:+ """Subscribe to ``spec['topic']`` at ``spec['url']`` and call+ ``on_message({"topic", "body", "json"})`` for each PUBLISH until+ ``stop`` is set. Pings the broker at half the keepalive. Raises+ :class:`TransportError` when the connection is lost."""+ url = str(spec.get("url", ""))+ topic = str(spec.get("topic", ""))+ try:+ scheme, host, port = broker(url)+ except ValueError as e:+ raise TransportError(str(e)) from None+ check_egress(f"http://{host}:{port}/", egress)+ keepalive = 60+ try:+ sock = socket.create_connection((host, port), timeout=timeout)+ except OSError as e:+ raise TransportError(f"{url}: {e.strerror or e}") from e+ try:+ if scheme == "mqtts":+ sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)+ client_id = str(spec.get("client_id") or f"pyrtan-watch-{os.getpid()}-{int(time.time() * 1000) % 100000}")+ user, password = spec.get("username"), spec.get("password")+ flags = 0x02 | (0x80 if user else 0) | (0x40 if password else 0)+ body = _mqtt_str("MQTT") + bytes([4, flags]) + struct.pack(">H", keepalive) + _mqtt_str(client_id)+ if user:+ body += _mqtt_str(str(user))+ if password:+ body += _mqtt_str(str(password))+ sock.sendall(_mqtt_packet(0x10, body))+ ack = _mqtt_read(sock, time.monotonic() + timeout)+ if ack is None or ack[0] != 2 or len(ack[2]) < 2 or ack[2][1] != 0:+ raise TransportError(f"{url}: the broker did not accept the connection")+ sock.sendall(_mqtt_packet(0x82, struct.pack(">H", 1) + _mqtt_str(topic) + b"\x00"))+ last_ping = time.monotonic()+ while not stop.is_set():+ pkt = _mqtt_read(sock, time.monotonic() + 1.0)+ if pkt is not None:+ ptype, pflags, pbody = pkt+ if ptype == 9 and len(pbody) >= 3 and pbody[2] == 0x80:+ raise TransportError(f"{url}: the broker refused the subscription to {topic!r}")+ if ptype == 3:+ tlen = struct.unpack(">H", pbody[:2])[0]+ mtopic = pbody[2:2 + tlen].decode("utf-8", errors="replace")+ rest = pbody[2 + tlen:]+ qos = (pflags >> 1) & 3+ if qos:+ pid, rest = rest[:2], rest[2:]+ if qos == 1:+ sock.sendall(_mqtt_packet(0x40, pid))+ text = rest.decode("utf-8", errors="replace")+ on_message({"topic": mtopic, "body": text, "json": _json(text)})+ if time.monotonic() - last_ping > keepalive / 2:+ sock.sendall(_mqtt_packet(0xC0, b""))+ last_ping = time.monotonic()+ try:+ sock.sendall(_mqtt_packet(0xE0, b""))+ except OSError:+ pass+ except OSError as e:+ raise TransportError(f"{url}: {e.strerror or e}") from e+ finally:+ try:+ sock.close()+ except OSError:+ pass+++def serial_lines(spec: dict[str, Any], on_line: Callable[[str], None], stop: threading.Event) -> None:+ """Hold ``spec['device']`` open and call ``on_line`` for each unit+ ending in ``read_until`` (default newline) until ``stop`` is set.+ Raises :class:`TransportError` when the device fails."""+ try:+ import termios+ except ImportError: # pragma: no cover+ raise TransportError("serial lines need a POSIX system") from None+ device = str(spec.get("device", ""))+ baud = int(spec.get("baud", 9600))+ terminator = str(spec.get("read_until", "\n")).encode("utf-8") or b"\n"+ speed = getattr(termios, f"B{baud}", None)+ if speed is None:+ raise TransportError(f"{device}: {baud} baud is not a rate this system knows")+ try:+ fd = os.open(device, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)+ except OSError as e:+ raise TransportError(f"{device}: {e.strerror or e}") from e+ buf = b""+ try:+ attrs = termios.tcgetattr(fd)+ attrs[0] &= ~(termios.IXON | termios.IXOFF | termios.ICRNL | termios.INLCR | termios.IGNCR | termios.ISTRIP)+ attrs[1] &= ~termios.OPOST+ attrs[2] = (attrs[2] & ~(termios.PARENB | termios.CSTOPB | termios.CSIZE)) | termios.CS8 | termios.CLOCAL | termios.CREAD+ attrs[3] &= ~(termios.ICANON | termios.ECHO | termios.ECHOE | termios.ISIG | termios.IEXTEN)+ attrs[4] = attrs[5] = speed+ termios.tcsetattr(fd, termios.TCSANOW, attrs)+ while not stop.is_set():+ ready, _, _ = select.select([fd], [], [], 0.5)+ if not ready:+ continue+ try:+ chunk = os.read(fd, 4096)+ except BlockingIOError:+ continue+ except OSError as e:+ raise TransportError(f"{device}: {e.strerror or e}") from e+ if not chunk:+ raise TransportError(f"{device}: closed")+ buf += chunk+ while terminator in buf:+ line, buf = buf.split(terminator, 1)+ on_line(line.decode("utf-8", errors="replace"))+ if len(buf) > MAX_READ:+ buf = b""+ finally:+ os.close(fd)diff --git a/src/pyrtan/catalog.py b/src/pyrtan/catalog.py index 2502f93..e0ff540 100644 --- a/src/pyrtan/catalog.py +++ b/src/pyrtan/catalog.py@@ -176,7 +176,7 @@ def _fetch(url: str, timeout: float = 20.0) -> bytes:raise CatalogError(f"{url}: {e}") from e if parts.scheme not in ("http", "https"): raise CatalogError(f"{url}: a catalog is reached over https (or a file: path)")- req = urllib.request.Request(url, headers={"User-Agent": "pyrtan-catalog/0.8", "Accept": "application/json, application/toml, text/plain, */*"})+ req = urllib.request.Request(url, headers={"User-Agent": "pyrtan-catalog/0.9", "Accept": "application/json, application/toml, text/plain, */*"})try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read() diff --git a/src/pyrtan/cli.py b/src/pyrtan/cli.py index c29bb68..c9e27fe 100644 --- a/src/pyrtan/cli.py +++ b/src/pyrtan/cli.py@@ -37,6 +37,7 @@ import argparseimport json import os import sys+import timefrom pathlib import Path from typing import Any@@ -50,7 +51,7 @@ from .runtime import Outcome, Runnerfrom .service import RemoteResolver, http_json, outcome_from_dict, serve from .adapters import LocalAdapters from .transport import EGRESS_POLICIES, RecordedTransport, RecordingTransport, UrllibTransport-from .tree import check, load_tree+from .tree import TreeError, check, load_tree@@ -103,10 +104,36 @@ def main(argv: list[str] | None = None) -> int:p.add_argument("--advertise", help="the URL the hub should reach this service at, if not host:port") p.add_argument("--env", action="append", default=[], metavar="K=V") p.add_argument("--gaps", metavar="FILE", help="keep gaps in this file (JSON lines) so they survive a restart")+ p.add_argument("--unanswered", metavar="FILE", default=os.environ.get("PYRTAN_UNANSWERED"), help="keep unanswered sentences in this file (JSON lines) so they survive a restart (default $PYRTAN_UNANSWERED)")+ p.add_argument("--watch", action="store_true", default=os.environ.get("PYRTAN_WATCH", "").lower() in ("1", "true", "yes"), help="run the trees' [[watches]]: a clock, a file, a topic or a serial line starts a run (default $PYRTAN_WATCH)")p.add_argument("--egress", choices=EGRESS_POLICIES, default=os.environ.get("PYRTAN_EGRESS", "any"), help="'public' refuses private, loopback and link-local addresses (default $PYRTAN_EGRESS or any)") _access_args(p) p.add_argument("--quiet", action="store_true")+ p = sub.add_parser("watch", help="run the trees' [[watches]] in this process, printing each run; nothing is served")+ p.add_argument("paths", nargs="*")+ p.add_argument("--env", action="append", default=[], metavar="K=V")+ p.add_argument("--gaps", metavar="FILE", help="keep gaps in this file")+ p.add_argument("--egress", choices=EGRESS_POLICIES, default=os.environ.get("PYRTAN_EGRESS", "any"))+ p.add_argument("--hub", help="delegate to this hub what the trees here do not hold")+ p.add_argument("--json", action="store_true", help="one JSON outcome per line instead of prose")+ _token_arg(p, "the hub's token, if any")++ p = sub.add_parser("watches", help="what a running service or hub is watching, and what each watch last did")+ p.add_argument("--at", default=os.environ.get("PYRTAN_AT", "http://127.0.0.1:8700"), help="a running service or hub (default $PYRTAN_AT or the local hub)")+ p.add_argument("--json", action="store_true")+ _token_arg(p, "the token the service or hub requires, if any")++ p = sub.add_parser("grow", help="draft the branch a gap asks for, by rules, for you to accept")+ p.add_argument("--trees", action="append", default=[], help="tree file or directory holding the trees the gaps name (repeatable; default $PYRTAN_TREES or ./trees)")+ p.add_argument("--at", default=os.environ.get("PYRTAN_AT"), help="read gaps from this running service or hub (default $PYRTAN_AT)")+ p.add_argument("--from", dest="source", metavar="FILE", help="read gaps from a file: a --gaps JSON-lines file, a /gaps answer, or a reports.json")+ p.add_argument("--reports", nargs="?", const="", metavar="HUB_OR_FILE", help="read the federation's gap reports for your catalog trees (from a hub's URL, a baked reports.json, or pyrtan.com)")+ p.add_argument("--tree", action="append", default=[], help="only gaps of this tree (repeatable)")+ p.add_argument("--apply", action="store_true", help="write each draft into its tree file (the file must load clean afterwards)")+ p.add_argument("--json", action="store_true")+ _token_arg(p, "the token the service or hub requires, if any")+p = sub.add_parser("hub", help="route between services") p.add_argument("--host", default="127.0.0.1") p.add_argument("--port", type=int, default=8700)@@ -115,6 +142,8 @@ def main(argv: list[str] | None = None) -> int:p.add_argument("--handout", action="append", default=[t for t in os.environ.get("PYRTAN_HANDOUT", "").split(os.pathsep) if t], metavar="PATH", help="trees for this hub's trusted members (a phone or a browser joined with the token) to run, not this hub (repeatable; default $PYRTAN_HANDOUT); its own --trees are handed out too") p.add_argument("--env", action="append", default=[], metavar="K=V", help="a value the hub's own trees may read as {{ env.K }}") p.add_argument("--gaps", metavar="FILE", help="keep the hub's own trees' gaps in this file")+ p.add_argument("--unanswered", metavar="FILE", default=os.environ.get("PYRTAN_UNANSWERED"), help="keep unanswered sentences in this file across restarts (default $PYRTAN_UNANSWERED)")+ p.add_argument("--watch", action="store_true", default=os.environ.get("PYRTAN_WATCH", "").lower() in ("1", "true", "yes"), help="run the hub's own trees' [[watches]] (default $PYRTAN_WATCH)")p.add_argument("--egress", choices=EGRESS_POLICIES, default=os.environ.get("PYRTAN_EGRESS", "any"), help="for the hub's own trees: 'public' refuses private addresses") p.add_argument("--peer", action="append", default=[], help="a service URL to start with (repeatable; services may also register themselves)") p.add_argument("--poll", type=float, default=10.0, help="seconds between health polls")@@ -218,6 +247,7 @@ def main(argv: list[str] | None = None) -> int:args = parser.parse_args(argv) commands = { "check": cmd_check, "list": cmd_list, "run": cmd_run, "match": cmd_match, "ask": cmd_ask, "serve": cmd_serve, "hub": cmd_hub, "gaps": cmd_gaps, "unanswered": cmd_unanswered, "new": cmd_new,+ "watch": cmd_watch, "watches": cmd_watches, "grow": cmd_grow,"search": cmd_search, "pull": cmd_pull, "index": cmd_index, "publish": cmd_publish, "console": cmd_console, "follow": cmd_follow, "unfollow": cmd_unfollow, "peers": cmd_peers, "directory": cmd_directory, "reports": cmd_reports, "keep": cmd_keep, "kept": cmd_kept, "fetch": cmd_fetch,@@ -430,8 +460,9 @@ def _report(outcome: Outcome, as_json: bool, with_trace: bool) -> int:def cmd_serve(args: argparse.Namespace) -> int: paths = _paths(args.paths) name = args.name or (paths[0].resolve().name if paths else "pyrtan")- service, server = serve(paths, args.host, args.port, name=name, hub=args.hub, env=_kv(args.env), advertise=args.advertise, quiet=args.quiet, token=args.token, cors=args.cors, gaps=args.gaps, egress=args.egress)- print(f"pyrtan service {name!r}: {len(service.registry)} tree(s) at {server.url}" + (f", registered with {args.hub}" if args.hub else "") + _access_note(args), flush=True)+ service, server = serve(paths, args.host, args.port, name=name, hub=args.hub, env=_kv(args.env), advertise=args.advertise, quiet=args.quiet, token=args.token, cors=args.cors, gaps=args.gaps, egress=args.egress, unanswered=args.unanswered, watch=args.watch)+ watching = f", watching {len(service.watcher)}" if service.watcher else ""+ print(f"pyrtan service {name!r}: {len(service.registry)} tree(s) at {server.url}{watching}" + (f", registered with {args.hub}" if args.hub else "") + _access_note(args), flush=True)for path, err in service.registry.errors.items(): print(f" not loaded: {path}: {err}", file=sys.stderr) try:@@ -439,10 +470,179 @@ def cmd_serve(args: argparse.Namespace) -> int:except KeyboardInterrupt: pass finally:+ service.stop()server.stop() return 0+def cmd_watch(args: argparse.Namespace) -> int:+ from .service import Service++ paths = _paths(args.paths)+ reg = Registry.from_paths(paths)+ for path, err in reg.errors.items():+ print(f"pyrtan: {path}: {err}", file=sys.stderr)+ service = Service(reg, name="watch", hub=args.hub, env=_kv(args.env), token=args.token, gaps=args.gaps, egress=args.egress)+ from .watch import Watcher++ def after(w, event, outcome):+ if args.json:+ print(json.dumps({"watch": w.name, "event": event, **outcome.to_dict(with_trace=False)}, ensure_ascii=False, default=str), flush=True)+ else:+ changed = " (the path changed)" if w.path_changed_at and w.last_at == w.path_changed_at else ""+ print(f"{time.strftime('%H:%M:%S')} {w.name}: {outcome.status}" + (f" ({outcome.reason})" if outcome.reason else "") + (f": {outcome.say}" if outcome.say else "") + changed, flush=True)++ watcher = Watcher(reg, service.run_watched, env=_kv(args.env), egress=args.egress, after=after, note=lambda text: print(f"pyrtan: {text}", file=sys.stderr, flush=True)).start()+ if not len(watcher):+ print("pyrtan watch: no tree here has [[watches]]; nothing to do", file=sys.stderr)+ watcher.stop()+ return 2+ for w in watcher.status()["watches"]:+ print(f"watching {w['name']}: {w['what']}", flush=True)+ try:+ while True:+ time.sleep(3600)+ except KeyboardInterrupt:+ pass+ finally:+ watcher.stop()+ return 0+++def cmd_watches(args: argparse.Namespace) -> int:+ try:+ status, body = http_json("GET", f"{args.at.rstrip('/')}/watches", token=args.token)+ except Exception as e:+ print(f"pyrtan: {args.at}: {e}", file=sys.stderr)+ return 3+ if status == 401:+ print(f"pyrtan: {args.at} needs a token: pass --token or set $PYRTAN_TOKEN", file=sys.stderr)+ return 3+ watches = (body or {}).get("watches", []) if isinstance(body, dict) else []+ if args.json:+ print(json.dumps({"watches": watches}, indent=2, ensure_ascii=False))+ return 0+ if not watches:+ print(f"{args.at} watches nothing (start it with --watch, and give a tree [[watches]])")+ return 0+ for w in watches:+ last = w.get("last") or {}+ when = time.strftime("%H:%M:%S", time.localtime(w["last_at"])) if w.get("last_at") else "never"+ line = f"{w['name']} {w['kind']} {w['what']} fired {w.get('fires', 0)}× (last {when})"+ if last:+ line += f": {last.get('status')}" + (f" ({last.get('reason')})" if last.get("reason") else "") + (f" {last.get('say')}" if last.get("say") else "")+ if w.get("path_changed_at"):+ line += f" [path changed at {time.strftime('%H:%M:%S', time.localtime(w['path_changed_at']))}: {' > '.join(w.get('previous_path') or [])} -> {' > '.join(w.get('path') or [])}]"+ if w.get("last_error"):+ line += f" error: {w['last_error']}"+ if w.get("peer") and w["peer"] != "local":+ line += f" ({w['peer']})"+ print(line)+ return 0+++def cmd_grow(args: argparse.Namespace) -> int:+ from .grow import apply_draft, draft_branch++ gaps: list[dict[str, Any]] = []+ if args.at:+ try:+ status, body = http_json("GET", f"{args.at.rstrip('/')}/gaps", token=args.token)+ except Exception as e:+ print(f"pyrtan: {args.at}: {e}", file=sys.stderr)+ return 3+ if status != 200 or not isinstance(body, dict):+ print(f"pyrtan: {args.at} answered {status}", file=sys.stderr)+ return 3+ gaps.extend(g for g in body.get("gaps", []) if isinstance(g, dict))+ if args.source:+ text = Path(args.source).read_text(encoding="utf-8")+ try:+ data = json.loads(text)+ if isinstance(data, dict) and ("gaps" in data or "reports" in data):+ rows = data.get("gaps") or data.get("reports") or [] # a /gaps answer, or a reports.json+ elif isinstance(data, dict):+ rows = [data] # one gap, one line+ else:+ rows = data+ except json.JSONDecodeError:+ rows = [json.loads(l) for l in text.splitlines() if l.strip()] # a --gaps file: JSON lines+ gaps.extend(_from_report(r) for r in rows if isinstance(r, dict))+ if args.reports is not None:+ given = args.reports+ source = f"{given.rstrip('/')}/reports" if given.startswith(("http://", "https://")) else given if given else (os.environ.get("PYRTAN_REPORTS") or "https://pyrtan.com/rendezvous/reports.json")+ try:+ if source.startswith(("http://", "https://")):+ status, body = http_json("GET", source)+ else:+ status, body = 200, json.loads(Path(source).read_text(encoding="utf-8"))+ except Exception as e:+ print(f"pyrtan: {source}: {e}", file=sys.stderr)+ return 3+ gaps.extend(_from_report(r) for r in (body or {}).get("reports", []) if isinstance(r, dict))+ if not args.at and not args.source and args.reports is None:+ print("pyrtan grow: say where the gaps are: --at URL, --from FILE or --reports", file=sys.stderr)+ return 2+ if args.tree:+ gaps = [g for g in gaps if g.get("tree") in args.tree]+ reg = Registry.from_paths(_paths(args.trees))+ for path, err in reg.errors.items():+ print(f"pyrtan: {path}: {err}", file=sys.stderr)+ drafts = []+ skipped: list[str] = []+ seen: set[str] = set()+ for g in gaps:+ tree = reg.get(str(g.get("tree", "")))+ if tree is None:+ skipped.append(f"{g.get('tree')}: not among the trees here")+ continue+ key = json.dumps([tree.name, g.get("node"), g.get("on"), g.get("observed")], sort_keys=True, default=str)+ if key in seen:+ continue+ seen.add(key)+ draft = draft_branch(tree, g)+ if draft is None:+ skipped.append(f"{tree.name}.{g.get('node')}: {g.get('on')} = {json.dumps(g.get('observed'), default=str)} is covered already, or names no decide node")+ continue+ drafts.append((tree, draft))+ if args.json:+ print(json.dumps({"drafts": [d.to_dict() | {"file": str(t.source)} for t, d in drafts], "skipped": skipped}, indent=2, ensure_ascii=False, default=str))+ return 0 if drafts else 1+ for note in skipped:+ print(f"skipped {note}")+ if not drafts:+ print("nothing to draft" + (": every gap has its branch" if gaps else ": no gaps"))+ return 1+ worst = 0+ for tree, draft in drafts:+ print(f"\n{draft.render()}")+ print(f"# why: {draft.reason}")+ if args.apply:+ if not tree.source:+ print(f"# cannot apply: {tree.name} has no file", file=sys.stderr)+ worst = 1+ continue+ try:+ apply_draft(tree.source, draft)+ print(f"# written into {tree.source}; now edit [nodes.{draft.then}] to say what the value means, then: pyrtan check {tree.source}")+ except TreeError as e:+ print(f"# {e}", file=sys.stderr)+ worst = 1+ reg.reload()+ if not args.apply:+ print("\n# nothing written; --apply puts each draft into its file for you to finish")+ return worst+++def _from_report(r: dict[str, Any]) -> dict[str, Any]:+ """A federation report (id namespace/name@version) as a gap (tree name)."""+ if "tree" in r:+ return r+ tree_id = str(r.get("id", ""))+ name = tree_id.split("/", 1)[-1].split("@", 1)[0]+ return {**r, "tree": name}++def cmd_hub(args: argparse.Namespace) -> int: from .hub import serve_hub@@ -460,10 +660,12 @@ def cmd_hub(args: argparse.Namespace) -> int:federate=args.federate or bool(args.advertise) or bool(follow) or args.listed, advertise=args.advertise, federation_secret=args.federation_secret, state=args.state, open_follows=not args.closed, follow=follow, listed=args.listed, trees=[Path(t) for t in args.trees], handout=[Path(t) for t in args.handout], env=_kv(args.env), gaps=args.gaps, egress=args.egress,- store=args.store, replicas=args.replicas, report_gaps=args.report_gaps,+ store=args.store, replicas=args.replicas, report_gaps=args.report_gaps, unanswered=args.unanswered, watch=args.watch,) fed = f", federated as {hub.federation.id}" + (f" following {len(hub.federation.following)}" if hub.federation.following else "") + (", listed" if hub.federation.listed else "") if hub.federation else "" local = f", holding {len(hub.local.trees)} tree(s)" if hub.local else ""+ if hub.local is not None and hub.local.service.watcher is not None:+ local += f", watching {len(hub.local.service.watcher)}"if hub.handout_registry is not None: local += f", handing out {len(hub.handout_registry.trees())} tree(s) to trusted members" print(f"pyrtan hub at {server.url}: {len(hub.peers)} peer(s){local}{fed}" + _access_note(args), flush=True) diff --git a/src/pyrtan/grow.py b/src/pyrtan/grow.py index 0dafaa2..cc31a09 100644 --- a/src/pyrtan/grow.py +++ b/src/pyrtan/grow.py@@ -22,14 +22,18 @@ observes and how it decides.from __future__ import annotations+import jsonimport re import shlex import threading import time from dataclasses import dataclass, field+from pathlib import Pathfrom typing import Any+from . import conditionsfrom .intent import Trigger, TriggerError, normalise+from .tree import Tree, TreeError, check, load_tree# Words that carry no meaning for a tree's name: dropped from the name, # kept in the trigger (where "what's the" is worth matching literally).@@ -233,26 +237,49 @@ class Unanswered:dropped. Lives in memory, like gaps; a restart forgets it. """- def __init__(self, limit: int = UNANSWERED_LIMIT) -> None:+ def __init__(self, limit: int = UNANSWERED_LIMIT, path: "str | Path | None" = None) -> None:self.limit = limit+ self.path = Path(path) if path else None+ self.error: str | None = Noneself._entries: dict[str, dict[str, Any]] = {} self._lock = threading.Lock()+ if self.path is not None and self.path.exists():+ try:+ for line in self.path.read_text(encoding="utf-8").splitlines():+ try:+ rec = json.loads(line)+ except json.JSONDecodeError:+ continue+ if isinstance(rec, dict) and isinstance(rec.get("text"), str):+ self._take(rec["text"], rec.get("peer"), int(rec.get("count", 1)), float(rec.get("last", time.time())), persist=False)+ except OSError as e:+ self.error = str(e)def record(self, text: str, peer: str | None = None) -> dict[str, Any]:+ return self._take(text, peer, 1, time.time(), persist=True)++ def _take(self, text: str, peer: str | None, count: int, when: float, *, persist: bool) -> dict[str, Any]:key = normalise(text).lower() or text- now = time.time()with self._lock: entry = self._entries.get(key) if entry is None:- entry = {"text": normalise(text) or text, "count": 0, "first": now, "last": now, "proposal": propose(text).to_dict()}+ entry = {"text": normalise(text) or text, "count": 0, "first": when, "last": when, "proposal": propose(text).to_dict()}if peer: entry["peer"] = peer self._entries[key] = entry- entry["count"] += 1- entry["last"] = now+ entry["count"] += count+ entry["last"] = max(entry["last"], when)while len(self._entries) > self.limit: oldest = min(self._entries, key=lambda k: self._entries[k]["last"]) del self._entries[oldest]+ if persist and self.path is not None:+ try:+ self.path.parent.mkdir(parents=True, exist_ok=True)+ with self.path.open("a", encoding="utf-8") as f:+ f.write(json.dumps({"text": entry["text"], "peer": peer, "count": 1, "last": when}, ensure_ascii=False) + "\n")+ self.error = None+ except OSError as e:+ self.error = str(e)return dict(entry) def to_list(self) -> list[dict[str, Any]]:@@ -261,3 +288,174 @@ class Unanswered:def __len__(self) -> int: return len(self._entries)+++# ---------------------------------------------------------------- growing a tree from a gap+++@dataclass+class Draft:+ """A branch a gap asks for, drafted by rules: the condition that would+ have taken the observed value, a new terminal node for it, and the+ text of both. A person reads it, changes the node's kind and words to+ what the value means, and keeps it. Nothing here decides anything."""++ tree: str+ node: str+ on: str+ observed: Any+ when: dict[str, Any]+ then: str+ branch: str+ node_text: str+ reason: str = ""++ def to_dict(self) -> dict[str, Any]:+ return {"tree": self.tree, "node": self.node, "on": self.on, "observed": self.observed, "when": self.when, "then": self.then, "branch": self.branch, "node_text": self.node_text, "reason": self.reason}++ def render(self) -> str:+ return f"# {self.tree}: in [nodes.{self.node}], add to branches (before the else, if any):\n {self.branch}\n\n{self.node_text}"+++def draft_branch(tree: Tree, gap: dict[str, Any]) -> Draft | None:+ """The branch a gap on ``tree`` asks for, or None when the gap names+ no decide node of this tree or a branch already covers the value+ (the file is newer than the gap).++ Rules: a missing value asks for ``{ exists = false }``; a number that+ falls in a hole between the ``between`` ranges or ``eq`` points the+ node already has asks for the range of that hole; anything else asks+ for ``{ eq = value }`` (``{ type = ... }`` for a value a report could+ only carry as its type). The new node is a ``fail`` that says what+ was seen; the author decides what it means."""+ node = tree.nodes.get(str(gap.get("node", "")))+ if node is None or node.kind != "decide":+ return None+ on = str(gap.get("on") or node.get("on") or "")+ observed = gap.get("observed")+ branches = [b for b in (node.get("branches") or []) if isinstance(b, dict)]+ if any("when" in b and conditions.holds(b["when"], observed) for b in branches):+ return None+ when, reason = _condition_for(observed, branches)+ then = _unique_node(tree, f"{node.id}-{_slug(observed)}")+ slug = _slug(observed)+ branch = "{ when = " + _toml_inline(when) + f', then = "{then}" }},'+ node_text = (+ f"# Drafted by pyrtan grow from a gap: {on} = {json.dumps(observed, ensure_ascii=False)} had no branch.\n"+ f"# Decide what that value means: keep this as a fail, or make it a done, a set, an observe...\n"+ f"[nodes.{then}]\n"+ f'kind = "fail"\n'+ f'reason = "{_slug(observed, 40) or "unhandled"}"\n'+ f'say = "{_toml_str(on)} was {_toml_str(json.dumps(observed, ensure_ascii=False))}; nobody has said what that means yet."\n'+ )+ return Draft(tree.name, node.id, on, observed, when, then, branch, node_text, reason)+++def _condition_for(observed: Any, branches: list[dict[str, Any]]) -> tuple[dict[str, Any], str]:+ if observed is None:+ return {"exists": False}, "the value was missing"+ if isinstance(observed, str) and observed.startswith("<") and observed.endswith(">") and observed[1:-1] in ("str", "dict", "list", "int", "float", "bool"):+ return {"type": observed[1:-1]}, "a report carries a long or structured value as its type"+ if isinstance(observed, (int, float)) and not isinstance(observed, bool):+ covered: list[tuple[float, float]] = []+ for b in branches:+ w = b.get("when") if isinstance(b, dict) else None+ if not isinstance(w, dict):+ continue+ if "between" in w and isinstance(w["between"], list) and len(w["between"]) == 2:+ try:+ covered.append((float(w["between"][0]), float(w["between"][1])))+ except (TypeError, ValueError):+ pass+ elif "eq" in w and isinstance(w["eq"], (int, float)) and not isinstance(w["eq"], bool):+ covered.append((float(w["eq"]), float(w["eq"])))+ if covered and isinstance(observed, int):+ below = [hi for lo, hi in covered if hi < observed]+ above = [lo for lo, hi in covered if lo > observed]+ if below and above and all(float(x).is_integer() for x in below + above):+ lo, hi = int(max(below)) + 1, int(min(above)) - 1+ if lo <= hi:+ return {"between": [lo, hi]}, f"the value falls in the hole between {int(max(below))} and {int(min(above))}, which the ranges around it leave open"+ return {"eq": observed}, "the value itself"+++def _unique_node(tree: Tree, base: str) -> str:+ name = base+ n = 2+ while name in tree.nodes:+ name = f"{base}-{n}"+ n += 1+ return name+++def _slug(value: Any, limit: int = 24) -> str:+ text = "missing" if value is None else str(value).lower()+ text = re.sub(r"[^a-z0-9]+", "-", text).strip("-")+ return (text[:limit].rstrip("-") or "value")+++def _toml_inline(cond: dict[str, Any]) -> str:+ return "{ " + ", ".join(f"{k} = {_toml_value(v)}" for k, v in cond.items()) + " }"+++def _toml_value(v: Any) -> str:+ if isinstance(v, bool):+ return "true" if v else "false"+ if isinstance(v, (int, float)):+ return repr(v)+ if isinstance(v, list):+ return "[" + ", ".join(_toml_value(x) for x in v) + "]"+ return '"' + _toml_str(str(v)) + '"'+++def _toml_str(s: str) -> str:+ return s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")+++def apply_draft(path: str | Path, draft: Draft) -> str:+ """Write a draft into the tree file: the branch into the node's+ ``branches`` list (before its ``else``), the node at the end. The file+ must load clean afterwards, else it is put back and :class:`TreeError`+ raised. Returns the new text."""+ path = Path(path)+ original = path.read_text(encoding="utf-8")+ lines = original.splitlines(keepends=True)+ header = f"[nodes.{draft.node}]"+ start = next((i for i, l in enumerate(lines) if l.strip() == header), None)+ if start is None:+ raise TreeError(f"{path}: no {header} table")+ end = next((i for i in range(start + 1, len(lines)) if lines[i].lstrip().startswith("[")), len(lines))+ b = next((i for i in range(start, end) if re.match(r"^\s*branches\s*=\s*\[", lines[i])), None)+ if b is None:+ raise TreeError(f"{path}: {header} has no branches list")+ indent = " "+ if "]" in lines[b].split("[", 1)[1]:+ # A one-line list.+ head, tail = lines[b].rsplit("]", 1)+ m = re.search(r",?\s*\{\s*else\s*=", head)+ if m:+ head = head[: m.start()] + ", " + draft.branch.rstrip(",") + head[m.start():]+ else:+ head = head.rstrip().rstrip(",") + ", " + draft.branch.rstrip(",")+ lines[b] = head + "]" + tail+ else:+ close = next(i for i in range(b + 1, end) if lines[i].strip().startswith("]"))+ else_line = next((i for i in range(b + 1, close) if re.match(r"^\s*\{\s*else\s*=", lines[i])), None)+ at = else_line if else_line is not None else close+ if at > b + 1:+ indent = re.match(r"^\s*", lines[at - 1 if else_line is None else at]).group(0) or indent+ lines.insert(at, f"{indent}{draft.branch}\n")+ text = "".join(lines)+ if not text.endswith("\n"):+ text += "\n"+ text += "\n" + draft.node_text+ path.write_text(text, encoding="utf-8")+ try:+ tree = load_tree(path)+ errors = [p for p in check(tree) if p.level == "error"]+ if errors:+ raise TreeError("; ".join(f"{p.where}: {p.message}" for p in errors))+ except TreeError as e:+ path.write_text(original, encoding="utf-8")+ raise TreeError(f"{path}: the draft did not fit the file ({e}); nothing changed") from e+ return textdiff --git a/src/pyrtan/hub.py b/src/pyrtan/hub.py index fbffcc9..3b72a85 100644 --- a/src/pyrtan/hub.py +++ b/src/pyrtan/hub.py@@ -119,7 +119,7 @@ class _HubResolver:class Hub:- def __init__(self, *, name: str = "hub", poll_every: float = 10.0, token: str | None = None, local: Service | None = None, handout: Registry | None = None) -> None:+ def __init__(self, *, name: str = "hub", poll_every: float = 10.0, token: str | None = None, local: Service | None = None, handout: Registry | None = None, unanswered: str | Path | None = None) -> None:self.name = name self.handout_registry = handout # trees for trusted members only: this hub hands them out and does not run them self.poll_every = poll_every@@ -129,7 +129,7 @@ class Hub:if local is not None: self.hold(local) self.federation: Federation | None = None- self.unanswered = Unanswered()+ self.unanswered = Unanswered(path=unanswered)self._reports: list[dict[str, Any]] = [] self._reports_at = 0.0 self.started = time.time()@@ -242,6 +242,24 @@ class Hub:def stop(self) -> None: self._stop.set()+ if self.local is not None:+ self.local.service.stop()++ def watches(self) -> dict[str, Any]:+ """This hub's own trees' watches, then each service's."""+ out: list[dict[str, Any]] = []+ if self.local is not None and self.local.service.watcher is not None:+ out.extend({**w, "peer": "local"} for w in self.local.service.watcher.status()["watches"])+ for peer in list(self.peers.values()):+ if peer.link == "down":+ continue+ try:+ status, body = http_json("GET", f"{peer.url}/watches", timeout=5.0, token=self.token)+ except Exception:+ continue+ if status == 200 and isinstance(body, dict):+ out.extend({**w, "peer": peer.url} for w in body.get("watches", []))+ return {"count": len(out), "watches": out}# ------------------------------------------------------------ lookup@@ -525,6 +543,8 @@ class Hub:return 200, self.gaps() if method == "GET" and path == "/unanswered": return 200, self.unanswered_all()+ if method == "GET" and path == "/watches":+ return 200, self.watches()if method == "GET" and path in ("/", "/index.html", "/console"): return 200, PAGE if method == "GET" and path == "/manifest.webmanifest":@@ -534,7 +554,7 @@ class Hub:if method == "GET" and path == "/pyrtan.js": return 200, ("text/javascript; charset=utf-8", PYRTAN_JS) if method == "GET" and path == "/api":- endpoints = ["/health", "/peers", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/register"]+ endpoints = ["/health", "/peers", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/watches", "/register"]if fed is not None: endpoints += ["/actor", "/inbox", "/mailbox", "/directory", "/reports", "/follow", "/unfollow", "/listed", "/following", "/followers", "/outbox", "/keep", "/fetch", "/kept/{key}"] return 200, {"pyrtan": "hub", "name": self.name, "version": __version__, "auth": "bearer" if self.token else "none", "federation": fed.id if fed else None, "endpoints": endpoints}@@ -584,9 +604,11 @@ def serve_hub(store: str | None = None, replicas: int = 2, report_gaps: bool = False,+ unanswered: str | Path | None = None,+ watch: bool = False,) -> tuple[Hub, Server]:- local = Service(Registry.from_paths(trees), name=f"{name}", env=env, gaps=gaps, egress=egress) if trees else None- hub = Hub(name=name, poll_every=poll_every, token=token, local=local, handout=Registry.from_paths(handout) if handout else None)+ local = Service(Registry.from_paths(trees), name=f"{name}", env=env, gaps=gaps, egress=egress, watch=watch) if trees else None+ hub = Hub(name=name, poll_every=poll_every, token=token, local=local, handout=Registry.from_paths(handout) if handout else None, unanswered=unanswered)server = Server(hub.route, host, port, quiet, token=token, cors=cors, federated=federate or bool(advertise)).start() if federate or advertise: hub.federate(advertise or server.url, secret=federation_secret, state=state, open_follows=open_follows, listed=listed, store=store or (str(Path(state).with_name("store")) if state else None), replicas=replicas, report_gaps=report_gaps) diff --git a/src/pyrtan/js/pyrtan.js b/src/pyrtan/js/pyrtan.js index ed71cb1..62b6f4a 100644 --- a/src/pyrtan/js/pyrtan.js +++ b/src/pyrtan/js/pyrtan.js@@ -612,7 +612,10 @@if (this.namespace) cap.namespace = this.namespace; Object.assign(cap, { description: this.description, tags: this.tags, hosts: this.hosts }); for (const k of LIST_NAMES) if (this[k].length) cap[k] = this[k];- return Object.assign(cap, { sources: this.sources(), license: this.license, language: this.language, triggers: this.triggers.map((t) => t.pattern), inputs });+ cap.sources = this.sources();+ const watches = Array.isArray(this.raw.watches) ? this.raw.watches : [];+ if (watches.length) cap.watches = watches.map((w) => { const kind = ["every", "file", "queue", "serial"].find((k) => k in w) || "?"; return { kind, what: kind === "every" ? String(w.every) : kind === "file" ? String(w.file) : kind === "queue" ? `${w.queue && w.queue.topic} at ${w.queue && w.queue.url}` : String(w.serial && w.serial.device) }; });+ return Object.assign(cap, { license: this.license, language: this.language, triggers: this.triggers.map((t) => t.pattern), inputs });} }@@ -908,5 +911,5 @@return outcome(d.status || "error", d.tree || fallbackTree, { say: d.say === undefined ? null : d.say, output: d.output === undefined ? null : d.output, reason: d.reason === undefined ? null : d.reason, question: d.question === undefined ? null : d.question, snapshot: d.snapshot === undefined ? null : d.snapshot, trace, inputs: d.inputs || {} }); }- return { MISSING, Tree, Runner, Trace, FetchTransport, RecordedTransport, Response, render, resolve, holds, extract, outcomeDict, fromDict, pyrepr, pyJson, html, TemplateError, TreeError, TransportError, ConditionError, SOURCES, adapterKey, version: "0.8.0" };+ return { MISSING, Tree, Runner, Trace, FetchTransport, RecordedTransport, Response, render, resolve, holds, extract, outcomeDict, fromDict, pyrepr, pyJson, html, TemplateError, TreeError, TransportError, ConditionError, SOURCES, adapterKey, version: "0.9.0" };}); diff --git a/src/pyrtan/page.py b/src/pyrtan/page.py index 51e9a43..5d33d40 100644 --- a/src/pyrtan/page.py +++ b/src/pyrtan/page.py@@ -102,6 +102,7 @@ PAGE = r"""<!doctype html><a href="#ask" data-tab="ask">Ask</a> <a href="#trees" data-tab="trees">Trees</a> <a href="#peers" data-tab="peers" id="peers-tab" hidden>Peers</a>+ <a href="#watches" data-tab="watches">Watches</a><a href="#gaps" data-tab="gaps">Gaps</a> <a href="#unanswered" data-tab="unanswered">Unanswered</a> <a href="#directory" data-tab="directory" id="directory-tab" hidden>Directory</a>@@ -138,6 +139,12 @@ PAGE = r"""<!doctype html></div> </section>+ <section id="tab-watches" hidden>+ <h2>Watches</h2>+ <p class="mute" style="font-size:14px;margin-top:0">Trees that run on their own: a clock, a file, a topic or a serial line starts them (<code>[[watches]]</code> in the tree, <code>--watch</code> on the service). Each row is what the watch last did; a path that changed between two runs of the same watch is how you learn a site changed before anyone asked.</p>+ <div class="card"><table id="watches"><thead><tr><th>Watch</th><th>On</th><th>Fired</th><th>Last run</th></tr></thead><tbody></tbody></table></div>+ </section>+<section id="tab-gaps" hidden> <h2>Gaps</h2> <p class="mute" style="font-size:14px;margin-top:0">A gap is a decision that met a value no branch covers. Each one is the next branch to write, in the tree and node named.</p>@@ -246,6 +253,7 @@ function showTab(name) {for (const s of document.querySelectorAll("main > section[id^=tab-]")) s.hidden = s.id !== "tab-" + name; if (name === "peers") loadPeers(); if (name === "gaps") loadGaps();+ if (name === "watches") loadWatches();if (name === "unanswered") loadUnanswered(); if (name === "directory") loadDirectory(); if (name === "member") member.render();@@ -362,6 +370,21 @@ async function loadPeers() {} }+async function loadWatches() {+ let w; try { w = await json("watches"); } catch (e) { problem(e.message, true); return; }+ const body = $("#watches tbody"); body.innerHTML = "";+ for (const x of w.watches || []) {+ const last = x.last || {};+ const when = x.last_at ? new Date(x.last_at * 1000).toLocaleTimeString() : "never";+ const changed = x.path_changed_at ? `<div class="mute">path changed ${esc(new Date(x.path_changed_at * 1000).toLocaleString())}: <code>${esc((x.previous_path||[]).join(" > "))}</code> → <code>${esc((x.path||[]).join(" > "))}</code></div>` : "";+ const err = x.last_error ? `<div class="mute">error: ${esc(x.last_error)}</div>` : "";+ const tr = document.createElement("tr");+ tr.innerHTML = `<td><code>${esc(x.name)}</code>${x.peer && x.peer !== "local" ? `<div class="mute">${esc(x.peer)}</div>` : ""}</td><td>${esc(x.kind)} <code>${esc(x.what)}</code></td><td>${esc(x.fires||0)}× <span class="mute">${esc(when)}</span></td><td>${last.status ? `<span class="status ${esc(last.status)}">${esc(last.status)}</span>${last.reason ? ` <span class="mute">(${esc(last.reason)})</span>` : ""} ${esc(last.say||"")}` : '<span class="mute">not yet</span>'}${changed}${err}</td>`;+ body.appendChild(tr);+ }+ if (!(w.watches || []).length) body.innerHTML = `<tr><td colspan="4" class="mute">Nothing watched. Give a tree <code>[[watches]]</code> and start the service with <code>--watch</code>.</td></tr>`;+}+async function loadGaps() { let g; try { g = await json("gaps"); } catch (e) { problem(e.message, true); return; } const body = $("#gaps tbody"); body.innerHTML = ""; diff --git a/src/pyrtan/service.py b/src/pyrtan/service.py index 82d90c8..cfdbb03 100644 --- a/src/pyrtan/service.py +++ b/src/pyrtan/service.py@@ -43,6 +43,7 @@ from __future__ import annotationsimport hmac import json+import sysimport threading import time import urllib.error@@ -58,6 +59,8 @@ from .grow import Unansweredfrom .registry import Registry from .runtime import Outcome, Runner from .trace import Trace+from .tree import Tree+from .watch import Watcherfrom .adapters import LocalAdapters from .transport import Transport, UrllibTransport@@ -146,6 +149,8 @@ class Service:token: str | None = None, gaps: str | Path | GapStore | None = None, egress: str = "any",+ unanswered: str | Path | None = None,+ watch: bool = False,) -> None: self.registry = registry self.name = name@@ -157,15 +162,29 @@ class Service:adapters = transport if transport is not None and hasattr(transport, "perform") else LocalAdapters(egress=egress) self.runner = Runner(registry, transport or UrllibTransport(egress=egress), resolver=RemoteResolver(self.hub, self.token) if self.hub else None, env=env or {}, adapters=adapters) self.gaps = gaps if isinstance(gaps, GapStore) else GapStore(gaps)- self.unanswered = Unanswered()+ self.unanswered = Unanswered(path=unanswered)self.runs = 0 self._lock = threading.Lock()+ self.watcher: Watcher | None = None+ if watch:+ self.watcher = Watcher(registry, self.run_watched, env=env or {}, egress=egress, after=None, note=lambda text: print(f"pyrtan: {text}", file=sys.stderr)).start()++ def run_watched(self, tree: Tree, inputs: dict[str, Any], trace: Trace) -> Outcome:+ """How a watch runs a tree: through the same runner and the same+ bookkeeping as a request, with the watch step already on the trace."""+ outcome = self.runner.run(tree, inputs, trace=trace)+ self._finish(outcome, {})+ return outcome++ def stop(self) -> None:+ if self.watcher is not None:+ self.watcher.stop()# ------------------------------------------------------------ handlers def health(self) -> dict[str, Any]: self.registry.reload()- health = {"status": "ok", "name": self.name, "trees": self.registry.names(), "errors": self.registry.errors, "runs": self.runs, "gaps": len(self.gaps), "unanswered": len(self.unanswered), "egress": self.egress, "uptime_s": int(time.time() - self.started)}+ health = {"status": "ok", "name": self.name, "trees": self.registry.names(), "errors": self.registry.errors, "runs": self.runs, "gaps": len(self.gaps), "unanswered": len(self.unanswered), "egress": self.egress, "uptime_s": int(time.time() - self.started), "watches": len(self.watcher) if self.watcher else 0}if self.gaps.error: health["gaps_file_error"] = self.gaps.error return health@@ -228,6 +247,8 @@ class Service:return 200, {"gaps": self.gaps.all()} if method == "GET" and path == "/unanswered": return 200, {"unanswered": self.unanswered.to_list()}+ if method == "GET" and path == "/watches":+ return 200, self.watcher.status() if self.watcher else {"count": 0, "watches": [], "watching": False}if method == "GET" and path in ("/", "/index.html", "/console"): return 200, PAGE if method == "GET" and path == "/manifest.webmanifest":@@ -237,7 +258,7 @@ class Service:if method == "GET" and path == "/pyrtan.js": return 200, ("text/javascript; charset=utf-8", PYRTAN_JS) if method == "GET" and path == "/api":- return 200, {"pyrtan": "service", "name": self.name, "version": __version__, "auth": "bearer" if self.token else "none", "endpoints": ["/health", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered"]}+ return 200, {"pyrtan": "service", "name": self.name, "version": __version__, "auth": "bearer" if self.token else "none", "endpoints": ["/health", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/watches"]}if method == "POST" and path == "/intent": return 200, self.intent(body) if method == "POST" and path == "/run":@@ -421,9 +442,11 @@ def serve(cors: list[str] | None = None, gaps: str | Path | None = None, egress: str = "any",+ unanswered: str | Path | None = None,+ watch: bool = False,) -> tuple[Service, Server]: registry = Registry.from_paths(paths)- service = Service(registry, name=name, hub=hub, env=env, advertise=advertise, token=token, gaps=gaps, egress=egress)+ service = Service(registry, name=name, hub=hub, env=env, advertise=advertise, token=token, gaps=gaps, egress=egress, unanswered=unanswered, watch=watch)server = Server(service.route, host, port, quiet, token=token, cors=cors).start() # The server is already answering when it registers: the hub checks a new # peer's /capabilities inside /register, and would otherwise wait on a diff --git a/src/pyrtan/transport.py b/src/pyrtan/transport.py index ab88770..2c7d17f 100644 --- a/src/pyrtan/transport.py +++ b/src/pyrtan/transport.py@@ -129,7 +129,7 @@ class UrllibTransport:``public`` (refuse private, loopback, link-local and metadata addresses, on the first URL and on every redirect)."""- def __init__(self, user_agent: str = "pyrtan/0.8 (+https://github.com/b-i-t-8/pyrtan)", egress: str = "any") -> None:+ def __init__(self, user_agent: str = "pyrtan/0.9 (+https://github.com/b-i-t-8/pyrtan)", egress: str = "any") -> None:if egress not in EGRESS_POLICIES: raise ValueError(f"egress policy must be one of {EGRESS_POLICIES}, got {egress!r}") self.user_agent = user_agent diff --git a/src/pyrtan/tree.py b/src/pyrtan/tree.py index ecd0414..53f1a59 100644 --- a/src/pyrtan/tree.py +++ b/src/pyrtan/tree.py@@ -164,6 +164,7 @@ class Tree:license: str = "" language: Language = field(default_factory=Language) namespace: str = "" # who published it: the catalog id is namespace/name+ watches: list[dict[str, Any]] = field(default_factory=list) # when this tree runs on its own: a clock, a file, a topic, a serial linepaths: list[str] = field(default_factory=list) # what ``file`` nodes may reach commands: list[str] = field(default_factory=list) # what ``shell`` nodes may run devices: list[str] = field(default_factory=list) # what ``serial`` nodes may open@@ -223,6 +224,9 @@ class Tree:return None if adapters.topic_allowed(self.topics, topic) else ("topic-not-allowed", f"{topic!r} is outside this tree's topics {self.topics}.") return None+ def watch_kinds(self) -> list[str]:+ return [next(k for k in WATCH_KINDS if k in w) for w in self.watches]+def sources(self) -> list[str]: """The forms of I/O this tree's nodes use.""" return sorted({s for n in self.nodes.values() if n.kind in ("observe", "act") for s in adapters.SOURCES if n.get(s) is not None})@@ -255,6 +259,7 @@ class Tree:"hosts": self.hosts, **{k: getattr(self, k) for k in ("paths", "commands", "devices", "topics") if getattr(self, k)}, "sources": self.sources(),+ **({"watches": [{"kind": k, "what": _watch_what(w)} for k, w in zip(self.watch_kinds(), self.watches)]} if self.watches else {}),"license": self.license, "language": self.language.to_dict(), "triggers": [t.pattern for t in self.triggers],@@ -310,6 +315,9 @@ def build(raw: dict[str, Any], source: str | None = None) -> Tree:namespace = raw.get("namespace", "") if not isinstance(namespace, str) or (namespace and not NAME.match(namespace)): raise TreeError(f"{name}: 'namespace' must be lowercase letters, digits and dashes (who publishes the tree), got {namespace!r}")+ watches = raw.get("watches", [])+ if not isinstance(watches, list) or not all(isinstance(w, dict) for w in watches):+ raise TreeError(f"{name}: 'watches' is a list of tables ([[watches]])")try: language = Language.from_raw(raw.get("language"), "language")@@ -370,7 +378,7 @@ def build(raw: dict[str, Any], source: str | None = None) -> Tree:hint = " (top-level keys such as 'start' must come before the first [table] in TOML)" if _buried(raw, "start") else "" raise TreeError(f"{name}: 'start' must name a node{hint}")- tree = Tree(name, description, version, hosts, triggers, inputs, start, nodes, tags, source, raw, license_.strip(), language, namespace, **lists)+ tree = Tree(name, description, version, hosts, triggers, inputs, start, nodes, tags, source, raw, license_.strip(), language, namespace, watches, **lists)errors = [p for p in check(tree) if p.level == "error"] if errors: raise TreeError(f"{name}: " + "; ".join(f"{p.where}: {p.message}" for p in errors))@@ -570,6 +578,9 @@ def check(tree: Tree) -> list[Problem]:elif head not in ("item", "index", "tree", "now", "env"): err(where, f"template root {head!r} is not one of inputs, state, item, index, tree, now, env")+ for i, w in enumerate(tree.watches):+ used.update(_check_watch(tree, i, w, err, warn))+for list_name, sources in _NEEDS.items(): have = getattr(tree, list_name) using = sorted(used & sources)@@ -587,6 +598,89 @@ def check(tree: Tree) -> list[Problem]:_NEEDS = {"hosts": {"http", "browser", "queue"}, "paths": {"file"}, "commands": {"shell"}, "devices": {"serial"}, "topics": {"queue"}}+WATCH_KINDS = ("every", "file", "queue", "serial")+WATCH_ROOTS = ("event", "env", "now")+++def _watch_what(w: dict[str, Any]) -> str:+ if "every" in w:+ return str(w["every"])+ if "file" in w:+ return str(w["file"])+ if "queue" in w and isinstance(w["queue"], dict):+ return f"{w['queue'].get('topic')} at {w['queue'].get('url')}"+ if "serial" in w and isinstance(w["serial"], dict):+ return str(w["serial"].get("device"))+ return "?"+++def _check_watch(tree: Tree, i: int, w: dict[str, Any], err, warn) -> set[str]:+ """A watch is static: no template reaches a path, a topic or a device,+ so every one is checked here against the tree's lists. Returns the+ sources it uses, for the allow-list bookkeeping."""+ from .nlp import ParseError, parse_duration++ where = f"watches[{i}]"+ kinds = [k for k in WATCH_KINDS if k in w]+ if len(kinds) != 1:+ err(where, f"a watch is exactly one of {', '.join(WATCH_KINDS)}, with optional 'inputs' and 'name'")+ return set()+ kind = kinds[0]+ used: set[str] = set()+ if kind == "every":+ value = w["every"]+ try:+ seconds = int(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else parse_duration(str(value))+ except (ParseError, ValueError):+ seconds = 0+ if seconds < 1:+ err(where, f"'every' is a duration such as \"5m\" or \"1 hour\", got {value!r}")+ elif kind == "file":+ path = w["file"]+ if not isinstance(path, str) or not path or PLACEHOLDER.search(path):+ err(where, "'file' is a literal path; a watch cannot be decided at run time")+ elif tree.paths and not adapters.path_allowed(tree.paths, path):+ err(where, f"{path!r} is not under this tree's paths")+ used.add("file")+ elif kind == "queue":+ spec = w["queue"]+ if not isinstance(spec, dict) or not isinstance(spec.get("url"), str) or not isinstance(spec.get("topic"), str) or PLACEHOLDER.search(spec["url"] + spec["topic"]):+ err(where, "'queue' is a table with literal 'url' and 'topic'")+ else:+ try:+ _, host, port = adapters.broker(spec["url"])+ except ValueError as e:+ err(where, str(e))+ else:+ if tree.hosts and not tree.host_allowed(f"mqtt://{host}:{port}/"):+ err(where, f"host {host!r} is not in this tree's hosts")+ if tree.topics and not adapters.topic_allowed(tree.topics, spec["topic"]):+ err(where, f"{spec['topic']!r} is not under this tree's topics")+ used.add("queue")+ elif kind == "serial":+ spec = w["serial"]+ if not isinstance(spec, dict) or not isinstance(spec.get("device"), str) or PLACEHOLDER.search(spec["device"]):+ err(where, "'serial' is a table with a literal 'device' and optionally 'baud', 'read_until'")+ elif tree.devices and not adapters.device_allowed(tree.devices, spec["device"]):+ err(where, f"{spec['device']!r} is not in this tree's devices")+ used.add("serial")+ inputs = w.get("inputs", {})+ if not isinstance(inputs, dict):+ err(where, "'inputs' is a table of input = value (templates may read event, env and now)")+ inputs = {}+ for k, v in inputs.items():+ if k not in tree.inputs:+ err(where, f"input {k!r} is not declared")+ for expr in placeholders(v):+ head = expr.split("|")[0].strip().split(".")[0].split("[")[0]+ if head not in WATCH_ROOTS:+ err(where, f"template root {head!r} is not one of {', '.join(WATCH_ROOTS)}")+ for k, spec in tree.inputs.items():+ if spec.required and spec.default is None and k not in inputs:+ err(where, f"input {k!r} is required and the watch does not give it")+ if w.get("name") is not None and not isinstance(w["name"], str):+ err(where, "'name' is a string")+ return useddef _literal(value: Any) -> str | None: diff --git a/src/pyrtan/watch.py b/src/pyrtan/watch.py new file mode 100644 index 0000000..35f0d37 --- /dev/null +++ b/src/pyrtan/watch.py@@ -0,0 +1,296 @@+"""Watches: a tree that runs when something speaks, without being asked.++A tree may carry ``[[watches]]``: a clock (``every = "5m"``), a file+(``file = "/var/log/syslog"``: each time lines are appended), a queue+(``queue = { url, topic }``: each message) or a serial line (``serial =+{ device, baud }``: each line). Each watch may give ``inputs``, templates+over ``event`` (what spoke), ``env`` and ``now``. A :class:`Watcher` holds+one thread per watch over a service's trees, runs the tree through the+service's own runner (so delegations reach the hub and gaps are kept),+and remembers for each watch when it last fired, what the outcome was,+and the *path* the run took: the sequence of nodes. When the path of a+scheduled run differs from the last one, the watch records that too.+That is how a person learns a site has changed before anyone asks.++The trace of a watched run begins with a ``watch`` step, as an asked+run's begins with ``intent``. Watches are static: a path, a topic, a+device is literal in the file and checked against the tree's lists by+``pyrtan check``, and a service started ``--egress public`` watches no+file and no serial line, as it reads none.++ pyrtan serve trees --watch # a service that also watches+ pyrtan hub --trees trees --watch # likewise, in a hub+ pyrtan watch trees # nothing served: just the watches, printing each run+"""++from __future__ import annotations++import datetime as _dt+import json+import os+import threading+import time+from typing import Any, Callable++from . import adapters+from .nlp import ParseError, parse_duration+from .runtime import Outcome+from .template import TemplateError, render+from .trace import Step, Trace+from .transport import EgressRefused, TransportError+from .tree import Tree++SYNC_EVERY = 10.0 # how often the watcher looks for trees that changed on disk+FILE_POLL = 2.0 # how often a file watch looks at the file+RETRY_AFTER = 5.0 # after a stream fails, before reconnecting+HISTORY = 20 # fires remembered per watch+++def every_seconds(value: Any) -> int:+ if isinstance(value, (int, float)) and not isinstance(value, bool):+ return int(value)+ try:+ return parse_duration(str(value))+ except ParseError:+ return 0+++class Watch:+ """One watch of one tree, on its own thread."""++ def __init__(self, watcher: "Watcher", tree_name: str, index: int, spec: dict[str, Any]) -> None:+ self.watcher = watcher+ self.tree_name = tree_name+ self.index = index+ self.spec = spec+ self.kind = next(k for k in ("every", "file", "queue", "serial") if k in spec)+ self.name = str(spec.get("name") or f"{tree_name}#{index}")+ self.key = json.dumps([tree_name, index, spec], sort_keys=True, default=str)+ self.stop = threading.Event()+ self.thread: threading.Thread | None = None+ self.fires = 0+ self.errors = 0+ self.last_error: str | None = None+ self.last_at: float | None = None+ self.last_outcome: dict[str, Any] | None = None+ self.last_path: list[str] | None = None+ self.path_changed_at: float | None = None+ self.previous_path: list[str] | None = None+ self.history: list[dict[str, Any]] = []+ self.lock = threading.Lock()++ def start(self) -> None:+ self.thread = threading.Thread(target=self._run, name=f"pyrtan-watch-{self.name}", daemon=True)+ self.thread.start()++ def _run(self) -> None:+ egress = self.watcher.egress+ if self.kind in ("file", "serial") and egress != "any":+ self._fail(f"{self.kind}: this service reaches only public addresses (--egress {egress}); it watches no file and no serial line")+ return+ while not self.stop.is_set():+ try:+ if self.kind == "every":+ self._every()+ elif self.kind == "file":+ self._file()+ elif self.kind == "queue":+ adapters.mqtt_stream(self.spec["queue"], lambda m: self.fire({"kind": "queue", **m}), self.stop, egress)+ elif self.kind == "serial":+ adapters.serial_lines(self.spec["serial"], lambda line: self.fire({"kind": "serial", "device": self.spec["serial"].get("device"), "body": line, "lines": [line]}), self.stop)+ return+ except EgressRefused as e:+ self._fail(str(e))+ return+ except TransportError as e:+ self._fail(str(e))+ except Exception as e: # a bug, not a line: keep the watch alive, say so+ self._fail(f"{type(e).__name__}: {e}")+ self.stop.wait(RETRY_AFTER)++ def _fail(self, text: str) -> None:+ with self.lock:+ self.errors += 1+ self.last_error = text+ self.watcher.note(f"watch {self.name}: {text}")++ def _every(self) -> None:+ seconds = max(1, every_seconds(self.spec["every"]))+ n = 0+ while not self.stop.is_set():+ n += 1+ self.fire({"kind": "every", "n": n, "at": _dt.datetime.now(_dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), "every": seconds})+ if self.stop.wait(seconds):+ return++ def _file(self) -> None:+ path = adapters.absolute_path(self.spec["file"])+ offset: int | None = None # None until the file is first seen: a watch starts at the end+ while not self.stop.is_set():+ try:+ st = os.stat(path)+ except OSError:+ offset = None+ if self.stop.wait(FILE_POLL):+ return+ continue+ if offset is None or st.st_size < offset:+ offset = st.st_size if offset is None else 0 # new file: start at its end; truncated: read it all+ if st.st_size < (offset or 0):+ offset = 0+ if st.st_size > offset:+ with open(path, "rb") as f:+ f.seek(offset)+ raw = f.read(adapters.MAX_READ)+ offset += len(raw)+ text = raw.decode("utf-8", errors="replace")+ lines = text.splitlines()+ if lines:+ self.fire({"kind": "file", "path": path, "body": text, "lines": lines, "size": st.st_size})+ if self.stop.wait(FILE_POLL):+ return++ def fire(self, event: dict[str, Any]) -> Outcome | None:+ """Run the tree for one event. Returns the outcome, or None when the+ tree is gone or its inputs do not render."""+ tree = self.watcher.registry.get(self.tree_name)+ if tree is None:+ self._fail(f"tree {self.tree_name!r} is gone")+ return None+ scope = {"event": event, "env": self.watcher.env, "now": _now_scope(self.watcher.clock())}+ try:+ inputs = {k: render(v, scope) for k, v in (self.spec.get("inputs") or {}).items()}+ except TemplateError as e:+ self._fail(f"inputs: {e}")+ return None+ trace = Trace(tree.name)+ trace.add(Step("watch", "watch", _summary(self.kind, event), observed=_observed(event), next=tree.start))+ outcome = self.watcher.run(tree, inputs, trace)+ path = [s.node for s in outcome.trace.steps] if outcome.trace else []+ now = time.time()+ with self.lock:+ self.fires += 1+ self.last_at = now+ self.last_outcome = {k: v for k, v in outcome.to_dict(with_trace=False).items() if k in ("status", "reason", "say", "output")}+ if self.last_path is not None and path != self.last_path:+ self.previous_path = self.last_path+ self.path_changed_at = now+ self.last_path = path+ self.history.append({"at": now, "status": outcome.status, "say": outcome.say, "path": path, "event": _observed(event)})+ del self.history[:-HISTORY]+ self.watcher.after(self, event, outcome)+ return outcome++ def to_dict(self) -> dict[str, Any]:+ with self.lock:+ return {+ "tree": self.tree_name, "index": self.index, "name": self.name, "kind": self.kind, "what": _what(self.kind, self.spec),+ "fires": self.fires, "errors": self.errors, "last_error": self.last_error, "last_at": self.last_at,+ "last": self.last_outcome, "path": self.last_path, "path_changed_at": self.path_changed_at, "previous_path": self.previous_path,+ "alive": bool(self.thread and self.thread.is_alive()),+ }+++class Watcher:+ """Every watch of every tree in a registry, kept in step with the files+ on disk. ``run(tree, inputs, trace)`` is how a run happens (a service+ gives its own, so gaps are counted and delegations route); ``after``+ is told about each fire."""++ def __init__(self, registry: Any, run: Callable[[Tree, dict[str, Any], Trace], Outcome], *, env: dict[str, Any] | None = None, egress: str = "any", clock: Callable[[], _dt.datetime] | None = None, after: Callable[["Watch", dict[str, Any], Outcome], None] | None = None, note: Callable[[str], None] | None = None) -> None:+ self.registry = registry+ self.run = run+ self.env = env or {}+ self.egress = egress+ self.clock = clock or (lambda: _dt.datetime.now(_dt.timezone.utc))+ self.after = after or (lambda w, e, o: None)+ self.note = note or (lambda text: None)+ self.watches: dict[str, Watch] = {}+ self._stop = threading.Event()+ self._thread: threading.Thread | None = None+ self._lock = threading.Lock()++ def sync(self) -> tuple[int, int]:+ """Start watches for trees that appeared or changed, stop the ones+ that vanished. Returns (started, stopped)."""+ self.registry.reload()+ wanted: dict[str, Watch] = {}+ for tree in self.registry.trees():+ for i, spec in enumerate(tree.watches):+ w = Watch(self, tree.name, i, spec)+ wanted[w.key] = w+ started = stopped = 0+ with self._lock:+ for key, w in list(self.watches.items()):+ if key not in wanted:+ w.stop.set()+ del self.watches[key]+ stopped += 1+ for key, w in wanted.items():+ if key not in self.watches:+ self.watches[key] = w+ w.start()+ started += 1+ return started, stopped++ def start(self) -> "Watcher":+ self.sync()++ def loop() -> None:+ while not self._stop.wait(SYNC_EVERY):+ try:+ self.sync()+ except Exception as e:+ self.note(f"watcher: {e}")++ self._thread = threading.Thread(target=loop, name="pyrtan-watcher", daemon=True)+ self._thread.start()+ return self++ def stop(self) -> None:+ self._stop.set()+ with self._lock:+ for w in self.watches.values():+ w.stop.set()++ def status(self) -> dict[str, Any]:+ with self._lock:+ watches = [w.to_dict() for w in self.watches.values()]+ watches.sort(key=lambda w: (w["tree"], w["index"]))+ return {"count": len(watches), "watches": watches}++ def __len__(self) -> int:+ return len(self.watches)+++def _what(kind: str, spec: dict[str, Any]) -> str:+ if kind == "every":+ return f"every {spec['every']}"+ if kind == "file":+ return str(spec["file"])+ if kind == "queue":+ return f"{spec['queue'].get('topic')} at {spec['queue'].get('url')}"+ return str(spec["serial"].get("device"))+++def _summary(kind: str, event: dict[str, Any]) -> str:+ if kind == "every":+ return f"every {event.get('every')} s, run {event.get('n')}"+ if kind == "file":+ return f"{event.get('path')}: {len(event.get('lines') or [])} new line(s)"+ if kind == "queue":+ return f"{event.get('topic')}: {len(str(event.get('body') or ''))} B"+ return f"{event.get('device')}: {str(event.get('body') or '')[:60]}"+++def _observed(event: dict[str, Any]) -> dict[str, Any]:+ out = {k: v for k, v in event.items() if k != "lines"}+ if isinstance(out.get("body"), str) and len(out["body"]) > 200:+ out["body"] = out["body"][:200] + "…"+ return out+++def _now_scope(now: _dt.datetime) -> dict[str, Any]:+ from .runtime import _now++ return _now(now)diff --git a/tests/conformance/vectors.json b/tests/conformance/vectors.json index 5c96d82..78a90ea 100644 --- a/tests/conformance/vectors.json +++ b/tests/conformance/vectors.json@@ -3194,7 +3194,18 @@"pattern": "how warm is the {room:word}" } ],- "version": 2+ "version": 2,+ "watches": [+ {+ "inputs": {+ "room": "kitchen"+ },+ "queue": {+ "topic": "home/kitchen/temperature",+ "url": "mqtt://broker.local"+ }+ }+ ]}, "site-status": { "description": "Whether a web site answers, and how quickly.",@@ -3308,7 +3319,16 @@"pattern": "[check] status of {host:host}" } ],- "version": 2+ "version": 2,+ "watches": [+ {+ "every": "15m",+ "inputs": {+ "host": "example.com"+ },+ "name": "example.com"+ }+ ]}, "weather": { "description": "Current weather for a place, from Open-Meteo.", diff --git a/tests/test_adapters.py b/tests/test_adapters.py index 418d635..255930b 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py@@ -318,6 +318,7 @@ class Broker:self.refuse = refuse self.published: list[tuple[str, bytes]] = [] self.subscribed: list[str] = []+ self.subscribers: list[tuple[socket.socket, str]] = []self.sock = socket.socket() self.sock.bind(("127.0.0.1", 0)) self.sock.listen(5)@@ -350,9 +351,12 @@ class Broker:tlen = struct.unpack(">H", body[2:4])[0] topic = body[4:4 + tlen].decode() self.subscribed.append(topic)+ self.subscribers.append((conn, topic))conn.sendall(_mqtt_packet(0x90, pid + b"\x00")) if self.message is not None: conn.sendall(_mqtt_packet(0x30, _mqtt_str(topic.replace("+", "kitchen")) + self.message))+ elif ptype == 12: # PINGREQ+ conn.sendall(_mqtt_packet(0xD0, b""))elif ptype == 3: tlen = struct.unpack(">H", body[:2])[0] self.published.append((body[2:2 + tlen].decode(), body[2 + tlen:]))@@ -361,6 +365,13 @@ class Broker:finally: conn.close()+ def publish_to_subscribers(self, topic: str, payload: bytes) -> None:+ for conn, _ in list(self.subscribers):+ try:+ conn.sendall(_mqtt_packet(0x30, _mqtt_str(topic) + payload))+ except OSError:+ pass+def close(self): self.sock.close() diff --git a/tests/test_watch_grow.py b/tests/test_watch_grow.py new file mode 100644 index 0000000..5b32c74 --- /dev/null +++ b/tests/test_watch_grow.py@@ -0,0 +1,455 @@+"""Watches (a tree that runs when something speaks) and growing a tree from+a gap (the branch drafted by rules for a person to accept)."""++from __future__ import annotations++import json+import os+import threading+import time+from pathlib import Path++import pytest++from pyrtan import Registry+from pyrtan.cli import main+from pyrtan.grow import Draft, Unanswered, apply_draft, draft_branch+from pyrtan.service import Server, Service, http_json+from pyrtan.tree import TreeError, check, load_tree, load_tree_text+from pyrtan.watch import Watcher+from test_adapters import Broker++LINES = '''+name = "lines"+description = "counts the lines a watch brought"+paths = ["{root}"]+start = "count"++[inputs.text]+type = "str"++[inputs.n]+type = "int"+default = 0++[[watches]]+name = "log"+file = "{log}"+inputs = {{ text = "{{{{ event.body | strip }}}}", n = "{{{{ event.lines | length }}}}" }}++[nodes.count]+kind = "decide"+on = "inputs.n"+branches = [+ {{ when = {{ eq = 1 }}, then = "one" }},+ {{ when = {{ gt = 1 }}, then = "many" }},+ {{ else = "none" }},+]++[nodes.one]+kind = "done"+say = "one line: {{{{ inputs.text }}}}"++[nodes.many]+kind = "done"+say = "{{{{ inputs.n }}}} lines"++[nodes.none]+kind = "fail"+reason = "nothing"+say = "nothing"+'''++TICK = '''+name = "tick"+description = "reads a file on a clock; the path changes when the file does"+paths = ["{root}"]+start = "read"++[[watches]]+name = "clock"+every = 1++[nodes.read]+kind = "observe"+file = {{ path = "{flag}" }}+into = "f"+next = "there"++[nodes.there]+kind = "decide"+on = "f.exists"+branches = [{{ when = {{ eq = true }}, then = "yes" }}, {{ else = "no" }}]++[nodes.yes]+kind = "done"+say = "there"++[nodes.no]+kind = "done"+say = "gone"+'''+++def wait_for(cond, timeout=10.0):+ deadline = time.time() + timeout+ while time.time() < deadline:+ if cond():+ return True+ time.sleep(0.05)+ return False+++def registry(tmp_path: Path, **trees: str) -> Registry:+ d = tmp_path / "trees"+ d.mkdir(parents=True, exist_ok=True)+ for name, text in trees.items():+ (d / f"{name}.toml").write_text(text)+ reg = Registry.from_paths([d])+ assert not reg.errors, reg.errors+ return reg+++# ---------------------------------------------------------------- watches+++def test_watches_are_checked():+ base = LINES.format(root="/tmp", log="/tmp/x.log")+ tree = load_tree_text(base)+ assert tree.watch_kinds() == ["file"] and tree.capability()["watches"] == [{"kind": "file", "what": "/tmp/x.log"}]+ with pytest.raises(TreeError, match="exactly one of every, file, queue, serial"):+ load_tree_text(base.replace('file = "/tmp/x.log"', 'file = "/tmp/x.log"\nevery = "5m"'))+ with pytest.raises(TreeError, match="not under this tree's paths"):+ load_tree_text(base.replace('file = "/tmp/x.log"', 'file = "/etc/x.log"'))+ with pytest.raises(TreeError, match="literal path"):+ load_tree_text(base.replace('file = "/tmp/x.log"', 'file = "/tmp/{{ env.x }}"'))+ with pytest.raises(TreeError, match="template root 'state' is not one of event, env, now"):+ load_tree_text(base.replace("{{ event.body | strip }}", "{{ state.body }}"))+ with pytest.raises(TreeError, match="input 'zzz' is not declared"):+ load_tree_text(base.replace("n = \"{{ event.lines | length }}\"", "n = \"{{ event.lines | length }}\", zzz = 1"))+ with pytest.raises(TreeError, match="input 'text' is required and the watch does not give it"):+ load_tree_text(base.replace('inputs = { text = "{{ event.body | strip }}", n = "{{ event.lines | length }}" }', 'inputs = { n = 1 }'))+ with pytest.raises(TreeError, match="'every' is a duration"):+ load_tree_text(base.replace('file = "/tmp/x.log"', 'every = "soon"').replace('paths = ["/tmp"]', ""))+ ok = load_tree_text(base.replace('file = "/tmp/x.log"', 'every = "1 hour and a half"').replace('paths = ["/tmp"]', ""))+ assert ok.capability()["watches"] == [{"kind": "every", "what": "1 hour and a half"}]+ with pytest.raises(TreeError, match="not under this tree's topics"):+ load_tree_text(base.replace('file = "/tmp/x.log"', 'queue = { url = "mqtt://b", topic = "x/y" }').replace('paths = ["/tmp"]', 'hosts = ["b"]\ntopics = ["a/#"]'))+ with pytest.raises(TreeError, match="host 'c' is not in this tree's hosts"):+ load_tree_text(base.replace('file = "/tmp/x.log"', 'queue = { url = "mqtt://c", topic = "a/y" }').replace('paths = ["/tmp"]', 'hosts = ["b"]\ntopics = ["a/#"]'))+ with pytest.raises(TreeError, match="not in this tree's devices"):+ load_tree_text(base.replace('file = "/tmp/x.log"', 'serial = { device = "/dev/ttyACM0" }').replace('paths = ["/tmp"]', 'devices = ["/dev/ttyUSB*"]'))+ # A watch on a source counts as using it: the list is required.+ with pytest.raises(TreeError, match="must list the paths"):+ load_tree_text(base.replace('paths = ["/tmp"]', ""))+++def test_file_watch_runs_the_tree_for_appended_lines(tmp_path):+ log = tmp_path / "app.log"+ log.write_text("old line\n")+ reg = registry(tmp_path, lines=LINES.format(root=str(tmp_path), log=str(log)))+ svc = Service(reg, name="w")+ fired: list = []+ watcher = Watcher(reg, svc.run_watched, after=lambda w, e, o: fired.append((w.name, e, o))).start()+ try:+ assert len(watcher) == 1 and watcher.status()["watches"][0]["what"] == str(log)+ time.sleep(0.3) # the watch starts at the end: the old line is not an event+ with log.open("a") as f:+ f.write("hello\n")+ assert wait_for(lambda: len(fired) == 1)+ name, event, out = fired[0]+ assert name == "log" and event["lines"] == ["hello"] and out.status == "done" and out.say == "one line: hello"+ assert out.trace.steps[0].kind == "watch" and out.trace.steps[0].node == "watch" and out.trace.steps[0].next == "count"+ assert "hello" in out.trace.steps[0].summary or "1 new line" in out.trace.steps[0].summary+ with log.open("a") as f:+ f.write("a\nb\nc\n")+ assert wait_for(lambda: len(fired) == 2) and fired[1][2].say == "3 lines"+ # Truncation (log rotation) is read from the start again.+ log.write_text("fresh\n")+ assert wait_for(lambda: len(fired) == 3) and fired[2][2].say == "one line: fresh"+ st = watcher.status()["watches"][0]+ assert st["fires"] == 3 and st["last"]["status"] == "done" and st["path"] == ["watch", "count", "one"] and st["alive"]+ # The path changed between runs 2 and 3 (many -> one), and the service counted the runs.+ assert st["path_changed_at"] is not None and st["previous_path"] == ["watch", "count", "many"]+ assert svc.runs == 3+ finally:+ watcher.stop()+++def test_clock_watch_notices_the_path_change_and_sync_follows_the_files(tmp_path):+ flag = tmp_path / "flag"+ reg = registry(tmp_path, tick=TICK.format(root=str(tmp_path), flag=str(flag)))+ svc = Service(reg, name="w")+ fired: list = []+ watcher = Watcher(reg, svc.run_watched, after=lambda w, e, o: fired.append(o)).start()+ try:+ assert wait_for(lambda: len(fired) >= 1) and fired[0].say == "gone" and fired[0].trace.steps[0].summary == "every 1 s, run 1"+ flag.write_text("x")+ assert wait_for(lambda: any(o.say == "there" for o in fired), 5)+ st = watcher.status()["watches"][0]+ assert st["path_changed_at"] and st["previous_path"] == ["watch", "read", "there", "no"] and st["path"] == ["watch", "read", "there", "yes"]+ # A tree that vanishes stops its watch; one that appears starts one.+ (tmp_path / "trees" / "tick.toml").unlink()+ assert watcher.sync() == (0, 1) and len(watcher) == 0+ (tmp_path / "trees" / "tick.toml").write_text(TICK.format(root=str(tmp_path), flag=str(flag)))+ assert watcher.sync() == (1, 0) and len(watcher) == 1+ finally:+ watcher.stop()+++def test_queue_watch_runs_per_message_and_serial_watch_per_line(tmp_path):+ b = Broker(message=None)+ tree = f'''+name = "sensor"+description = "x"+hosts = ["127.0.0.1"]+topics = ["home/#"]+start = "say"++[inputs.reading]+type = "str"++[[watches]]+queue = {{ url = "{b.url}", topic = "home/+/temperature" }}+inputs = {{ reading = "{{{{ event.topic }}}}={{{{ event.json.temperature }}}}" }}++[nodes.say]+kind = "done"+say = "{{{{ inputs.reading }}}}"+'''+ reg = registry(tmp_path, sensor=tree)+ svc = Service(reg, name="w")+ fired: list = []+ watcher = Watcher(reg, svc.run_watched, after=lambda w, e, o: fired.append(o)).start()+ try:+ assert wait_for(lambda: b.subscribed == ["home/+/temperature"])+ b.publish_to_subscribers("home/kitchen/temperature", b'{"temperature": 21}')+ assert wait_for(lambda: len(fired) == 1) and fired[0].say == "home/kitchen/temperature=21"+ b.publish_to_subscribers("home/attic/temperature", b'{"temperature": 30}')+ assert wait_for(lambda: len(fired) == 2) and fired[1].say == "home/attic/temperature=30"+ finally:+ watcher.stop()+ b.close()+ # Serial: a pty, each line an event.+ master, slave = os.openpty()+ device = os.ttyname(slave)+ stree = f'''+name = "modem"+description = "x"+devices = ["/dev/pts/*"]+start = "say"++[inputs.line]+type = "str"++[[watches]]+serial = {{ device = "{device}", baud = 9600, read_until = "\\n" }}+inputs = {{ line = "{{{{ event.body | strip }}}}" }}++[nodes.say]+kind = "done"+say = "line: {{{{ inputs.line }}}}"+'''+ reg2 = registry(tmp_path / "s", modem=stree)+ svc2 = Service(reg2, name="w2")+ fired2: list = []+ watcher2 = Watcher(reg2, svc2.run_watched, after=lambda w, e, o: fired2.append(o)).start()+ try:+ time.sleep(0.3)+ os.write(master, b"RING\r\nOK\r\n")+ assert wait_for(lambda: len(fired2) == 2) and [o.say for o in fired2] == ["line: RING", "line: OK"]+ finally:+ watcher2.stop()+ os.close(master)+ os.close(slave)+++def test_hosted_service_watches_no_file_and_errors_are_shown(tmp_path):+ log = tmp_path / "x.log"+ log.write_text("")+ reg = registry(tmp_path, lines=LINES.format(root=str(tmp_path), log=str(log)))+ svc = Service(reg, name="w", egress="public")+ notes: list[str] = []+ watcher = Watcher(reg, svc.run_watched, egress="public", note=notes.append).start()+ try:+ assert wait_for(lambda: notes)+ assert "watches no file" in notes[0]+ st = watcher.status()["watches"][0]+ assert st["errors"] == 1 and "watches no file" in st["last_error"]+ finally:+ watcher.stop()+++def test_watches_over_http_and_the_commands(tmp_path, capsys):+ log = tmp_path / "x.log"+ log.write_text("")+ (tmp_path / "t").mkdir()+ (tmp_path / "t" / "lines.toml").write_text(LINES.format(root=str(tmp_path), log=str(log)))+ from pyrtan.hub import serve_hub+ from pyrtan.service import serve++ hub, hub_srv = serve_hub(port=0, quiet=True, poll_every=60)+ svc, srv = serve([tmp_path / "t"], port=0, quiet=True, watch=True, hub=hub_srv.url)+ try:+ assert srv.url in {p.url for p in hub.peers.values()}+ with log.open("a") as f:+ f.write("ping\n")+ assert wait_for(lambda: http_json("GET", f"{srv.url}/watches")[1]["watches"][0]["fires"] == 1)+ _, w = http_json("GET", f"{srv.url}/watches")+ assert w["count"] == 1 and w["watches"][0]["last"]["say"] == "one line: ping"+ _, hw = http_json("GET", f"{hub_srv.url}/watches")+ assert hw["count"] == 1 and hw["watches"][0]["peer"] == srv.url+ assert "/watches" in http_json("GET", f"{srv.url}/api")[1]["endpoints"]+ assert main(["watches", "--at", hub_srv.url]) == 0+ out = capsys.readouterr().out+ assert "log file " in out and "fired 1×" in out and "one line: ping" in out+ assert main(["watches", "--at", hub_srv.url, "--json"]) == 0 and json.loads(capsys.readouterr().out)["watches"][0]["name"] == "log"+ # A service without --watch watches nothing, and says so.+ svc2, srv2 = serve([tmp_path / "t"], port=0, quiet=True)+ try:+ assert main(["watches", "--at", srv2.url]) == 0 and "watches nothing" in capsys.readouterr().out+ finally:+ srv2.stop()+ finally:+ svc.stop()+ srv.stop()+ hub.stop()+ hub_srv.stop()+++def test_unanswered_survive_a_restart(tmp_path):+ path = tmp_path / "unanswered.jsonl"+ u = Unanswered(path=path)+ u.record("top 3 stories on lobsters")+ u.record("top 3 stories on lobsters", peer="x")+ u.record("is the printer on")+ again = Unanswered(path=path)+ rows = again.to_list()+ assert [(r["text"], r["count"]) for r in rows] == [("top 3 stories on lobsters", 2), ("is the printer on", 1)]+ assert rows[0]["proposal"]["name"] == "top-stories-lobsters"+ (tmp_path / "none").mkdir()+ svc = Service(Registry.from_paths([tmp_path / "none"]), name="s", unanswered=path)+ assert len(svc.unanswered) == 2+++# ---------------------------------------------------------------- growing from a gap+++WEATHERISH = '''+name = "sky"+description = "a code into a word"+start = "pick"++[inputs.code]+type = "int"++[nodes.pick]+kind = "decide"+on = "inputs.code"+branches = [+ { when = { eq = 0 }, then = "clear" },+ { when = { between = [1, 3] }, then = "cloudy" },+ { when = { between = [51, 67] }, then = "rain" },+ { when = { between = [95, 99] }, then = "storm" },+]++[nodes.clear]+kind = "done"+say = "clear"++[nodes.cloudy]+kind = "done"+say = "cloudy"++[nodes.rain]+kind = "done"+say = "rain"++[nodes.storm]+kind = "done"+say = "storm"+'''+++def test_draft_branch_rules():+ tree = load_tree_text(WEATHERISH)+ # A hole between ranges: the range of the hole.+ d = draft_branch(tree, {"tree": "sky", "node": "pick", "on": "inputs.code", "observed": 71})+ assert d is not None and d.when == {"between": [68, 94]} and d.then == "pick-71" and "hole between 67 and 95" in d.reason+ assert d.branch == '{ when = { between = [68, 94] }, then = "pick-71" },'+ assert d.node_text.startswith("# Drafted by pyrtan grow") and '[nodes.pick-71]\nkind = "fail"\nreason = "71"' in d.node_text+ # Beyond the last range: just the value.+ assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": 120}).when == {"eq": 120}+ # Covered already: nothing to draft.+ assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": 2}) is None+ # Not a decide, or no such node: nothing.+ assert draft_branch(tree, {"node": "clear", "on": "x", "observed": 1}) is None and draft_branch(tree, {"node": "nope", "observed": 1}) is None+ # Missing, strings, types from a report.+ assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": None}).when == {"exists": False}+ s = draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": "N/A"})+ assert s.when == {"eq": "N/A"} and s.then == "pick-n-a" and s.branch == '{ when = { eq = "N/A" }, then = "pick-n-a" },'+ assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": "<dict>"}).when == {"type": "dict"}+ assert draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": True}).when == {"eq": True}+ # A node name already taken gets a suffix.+ taken = load_tree_text(WEATHERISH.replace('[nodes.storm]', '[nodes.pick-71]\nkind = "done"\nsay = "x"\n\n[nodes.storm]'))+ assert draft_branch(taken, {"node": "pick", "on": "inputs.code", "observed": 71}).then == "pick-71-2"+++def test_apply_draft_edits_the_file_and_reverts_when_it_does_not_fit(tmp_path):+ path = tmp_path / "sky.toml"+ path.write_text(WEATHERISH)+ tree = load_tree(path)+ d = draft_branch(tree, {"node": "pick", "on": "inputs.code", "observed": 71})+ text = apply_draft(path, d)+ grown = load_tree(path)+ assert "pick-71" in grown.nodes and grown.nodes["pick-71"].kind == "fail"+ assert [b.get("then") for b in grown.nodes["pick"].get("branches")] == ["clear", "cloudy", "rain", "storm", "pick-71"]+ assert " { when = { between = [68, 94] }, then = \"pick-71\" },\n]" in text+ assert not [p for p in check(grown) if p.level == "error"]+ # Before an else, and on a one-line list.+ one = tmp_path / "one.toml"+ one.write_text(WEATHERISH.replace(" { when = { between = [95, 99] }, then = \"storm\" },\n]", " { when = { between = [95, 99] }, then = \"storm\" },\n { else = \"storm\" },\n]"))+ apply_draft(one, draft_branch(load_tree(one), {"node": "pick", "on": "inputs.code", "observed": 71}))+ assert [b.get("then") or "else" for b in load_tree(one).nodes["pick"].get("branches")] == ["clear", "cloudy", "rain", "storm", "pick-71", "else"]+ flat = tmp_path / "flat.toml"+ flat.write_text('name = "f"\ndescription = "x"\nstart = "p"\n[inputs.v]\ntype = "str"\n[nodes.p]\nkind = "decide"\non = "inputs.v"\nbranches = [{ when = { eq = "a" }, then = "a" }, { else = "b" }]\n[nodes.a]\nkind = "done"\nsay = "a"\n[nodes.b]\nkind = "done"\nsay = "b"\n')+ apply_draft(flat, draft_branch(load_tree(flat), {"node": "p", "on": "inputs.v", "observed": "zz"}))+ assert [b.get("then") or "else" for b in load_tree(flat).nodes["p"].get("branches")] == ["a", "p-zz", "else"]+ # A draft that cannot fit (a node the file has no table for) leaves the file as it was.+ before = path.read_text()+ with pytest.raises(TreeError, match="no \\[nodes.nope\\] table"):+ apply_draft(path, Draft("sky", "nope", "x", 1, {"eq": 1}, "nope-1", '{ when = { eq = 1 }, then = "nope-1" },', '[nodes.nope-1]\nkind = "done"\n'))+ assert path.read_text() == before+ bad = Draft("sky", "pick", "inputs.code", 5, {"eq": 5}, "pick-5", '{ when = { eq = 5 }, then = "pick-5" },', '[nodes.pick-5]\nkind = "done"\nnext = "nowhere"\n')+ with pytest.raises(TreeError, match="did not fit the file"):+ apply_draft(path, bad)+ assert path.read_text() == before+++def test_grow_command(tmp_path, capsys):+ (tmp_path / "t").mkdir()+ path = tmp_path / "t" / "sky.toml"+ path.write_text(WEATHERISH)+ gaps = tmp_path / "gaps.jsonl"+ gaps.write_text(json.dumps({"tree": "sky", "node": "pick", "on": "inputs.code", "observed": 71, "count": 3}) + "\n" + json.dumps({"tree": "sky", "node": "pick", "on": "inputs.code", "observed": 2}) + "\n" + json.dumps({"tree": "other", "node": "x", "on": "y", "observed": 1}) + "\n")+ assert main(["grow", "--from", str(gaps), "--trees", str(tmp_path / "t")]) == 0+ out = capsys.readouterr().out+ assert "skipped other: not among the trees here" in out and "is covered already" in out and "[nodes.pick-71]" in out and "nothing written" in out+ assert load_tree(path).nodes.get("pick-71") is None+ assert main(["grow", "--from", str(gaps), "--trees", str(tmp_path / "t"), "--apply"]) == 0+ assert "written into" in capsys.readouterr().out and load_tree(path).nodes["pick-71"].kind == "fail"+ assert main(["grow", "--from", str(gaps), "--trees", str(tmp_path / "t")]) == 1 and "every gap has its branch" in capsys.readouterr().out+ # From a federation report: the id names the tree; from a running service: its /gaps.+ reports = tmp_path / "reports.json"+ reports.write_text(json.dumps({"reports": [{"id": "pyrtan/sky@1", "node": "pick", "on": "inputs.code", "observed": 120, "count": 2, "reporters": 1}]}))+ assert main(["grow", "--reports", str(reports), "--trees", str(tmp_path / "t"), "--json"]) == 0+ drafts = json.loads(capsys.readouterr().out)["drafts"]+ assert drafts[0]["when"] == {"eq": 120} and drafts[0]["file"].endswith("sky.toml")+ svc = Service(Registry.from_paths([tmp_path / "t"]), name="s")+ srv = Server(svc.route, quiet=True).start()+ try:+ assert svc.run({"tree": "sky", "inputs": {"code": 42}})["reason"] == "gap"+ assert main(["grow", "--at", srv.url, "--trees", str(tmp_path / "t")]) == 0 and "pick-42" in capsys.readouterr().out+ finally:+ srv.stop()+ assert main(["grow", "--trees", str(tmp_path / "t")]) == 2diff --git a/uv.lock b/uv.lock index 488a654..e0becf9 100644 --- a/uv.lock +++ b/uv.lock@@ -147,7 +147,7 @@ wheels = [[[package]] name = "pyrtan"-version = "0.8.0"+version = "0.9.0"source = { editable = "." } [package.optional-dependencies]