pyrtan / sourcev0.19.0 · main

pyrtan / src / pyrtan / service.py

22,397 bytes · last changed 2026-10-28 in Agents beside the swarm: open-weights models that probe the trees, suggest, chat, remember and follow skills; release 0.15.0

1"""One member of the swarm: an HTTP service hosting a registry of trees.
2
3Endpoints, all JSON:
4
5``GET /health``
6 ``{"status": "ok", "name": ..., "trees": [...], "errors": {...}}``
7``GET /capabilities``
8 What this service can do: every tree's name, triggers and inputs.
9``POST /intent`` ``{"text": "..."}``
10 Match a sentence to a tree and run it.
11``POST /run`` ``{"tree": "...", "inputs": {...}}``
12 Run a named tree.
13``POST /resume`` ``{"snapshot": {...}, "answer": ...}``
14 Continue a run that asked a question.
15``GET /gaps``
16 Decisions that met a value no branch covered, since the service started.
17 Each is a branch waiting to be written.
18``GET /unanswered``
19 Sentences no tree answered, counted, each with a proposed tree. Each is
20 a tree waiting to be written; ``pyrtan unanswered --at URL --scaffold
21 DIR`` writes the files.
22``GET /suggestions`` and ``POST /suggestions`` ``{"suggestions": [...]}``
23 What agents beside the swarm suggest (:mod:`pyrtan.suggestions`): a
24 branch's meaning, a trigger, a tree. Kept, never applied;
25 ``pyrtan grow --suggestions`` writes them, by rules, for a person.
26``GET /`` and ``GET /console``
27 The console (:mod:`pyrtan.page`); ``GET /api`` is the JSON index.
28
29A service may be given a hub (``--hub``). Then it registers itself on start,
30and a ``delegate`` to a tree it does not hold is forwarded to the hub.
31
32**Access.** With no token, everything is open: the swarm is assumed to sit
33on a trusted network behind the operator's ingress. With a token
34(``--token`` or ``$PYRTAN_TOKEN``), every request except ``GET /health``,
35the console and ``/api`` must carry ``Authorization: Bearer <token>``;
36``/health`` then answers with status and counts only. A swarm shares one
37token: a service sends it to its hub when it registers and the hub sends it
38to peers when it forwards. **CORS** is off unless ``--cors`` names the
39origins (or ``*``) a browser may call from; the console at pyrtan.com is
40one such origin. The server does not do TLS; put it behind a proxy that
41does when it leaves the LAN.
42
43The server is the standard library's threading HTTP server.
44"""
45
46from __future__ import annotations
47
48import hmac
49import json
50import sys
51import threading
52import time
53import urllib.error
54import urllib.request
55from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
56from pathlib import Path
57from typing import Any, Callable
58
59from . import __version__
60from .gaps import GapStore
61from .page import MANIFEST, PAGE, PYRTAN_JS, SERVICE_WORKER
62from .grow import Unanswered
63from .suggestions import Suggestions
64from .registry import Registry
65from .runtime import Outcome, Runner
66from .trace import Trace
67from .tree import Tree
68from .watch import Watcher
69from .adapters import LocalAdapters
70from .transport import Transport, UrllibTransport
71
72MAX_BODY = 1 << 20
73
74# What answers without a token: the console, its index, liveness, and a pushed
75# identity document (which proves itself by its signature).
76PUBLIC_PATHS = frozenset({"/", "/index.html", "/console", "/api", "/health", "/actor", "/directory", "/reports", "/manifest.webmanifest", "/sw.js", "/pyrtan.js", "/identity"})
77
78
79def public_path(path: str) -> bool:
80 """What answers without a token: the paths above, and ``/.well-known/`` (identity documents, signed)."""
81 return path in PUBLIC_PATHS or path.startswith("/.well-known/")
82# On a federated hub, a request carrying a signature reaches the route,
83# which verifies it (see :mod:`pyrtan.federation`).
84SIGNED_PATHS = frozenset({"/inbox", "/run", "/intent", "/resume", "/capabilities", "/mailbox"})
85
86
87def signed_path(path: str) -> bool:
88 """Whether a signed (federated) request may reach this path at all."""
89 return path in SIGNED_PATHS or path.startswith("/kept/")
90# Headers a browser at another origin may send: federation's HMAC and an identity's key.
91SIGNATURE_HEADERS = ("X-Pyrtan-Actor", "X-Pyrtan-Date", "X-Pyrtan-Signature", "X-Pyrtan-Key", "X-Pyrtan-Key-Signature", "X-Pyrtan-Device")
92# What /health says to a caller without a token, when one is required.
93PUBLIC_HEALTH_KEYS = ("status", "name", "peers", "up", "stale", "down", "uptime_s")
94
95
96def http_json(method: str, url: str, payload: Any = None, timeout: float = 15.0, token: str | None = None, headers: dict[str, str] | None = None) -> tuple[int, Any]:
97 # Canonical JSON on the wire, so a signature over the body (federation)
98 # can be checked from the parsed body on the other side.
99 data = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") if payload is not None else None
100 hdrs = {"Content-Type": "application/json", "Accept": "application/json"}
101 if token:
102 hdrs["Authorization"] = f"Bearer {token}"
103 hdrs.update(headers or {})
104 req = urllib.request.Request(url, data=data, method=method, headers=hdrs)
105 try:
106 with urllib.request.urlopen(req, timeout=timeout) as resp:
107 return resp.status, json.loads(resp.read().decode("utf-8") or "null")
108 except urllib.error.HTTPError as e:
109 try:
110 return e.code, json.loads(e.read().decode("utf-8") or "null")
111 except Exception:
112 return e.code, None
113
114
115class RemoteResolver:
116 """Sends what a local runner cannot answer to a hub or another service."""
117
118 def __init__(self, base_url: str, token: str | None = None) -> None:
119 self.base_url = base_url.rstrip("/")
120 self.token = token
121
122 def run_tree(self, name: str, inputs: dict[str, Any], depth: int) -> Outcome:
123 status, body = http_json("POST", f"{self.base_url}/run", {"tree": name, "inputs": inputs, "depth": depth}, token=self.token)
124 return outcome_from_dict(body, name) if isinstance(body, dict) else Outcome("error", name, reason="remote", say=f"{self.base_url} answered {status}")
125
126 def run_intent(self, text: str, depth: int) -> Outcome:
127 status, body = http_json("POST", f"{self.base_url}/intent", {"text": text, "depth": depth}, token=self.token)
128 return outcome_from_dict(body, "") if isinstance(body, dict) else Outcome("error", "", reason="remote", say=f"{self.base_url} answered {status}")
129
130
131def outcome_from_dict(d: dict[str, Any], fallback_tree: str = "") -> Outcome:
132 trace = None
133 if isinstance(d.get("trace"), dict):
134 from .trace import Step
135
136 trace = Trace(d["trace"].get("tree", fallback_tree))
137 trace.steps = [Step(**{k: v for k, v in s.items() if k in Step.__dataclass_fields__}) for s in d["trace"].get("steps", [])]
138 trace.gaps = list(d["trace"].get("gaps", []))
139 return Outcome(
140 status=d.get("status", "error"),
141 tree=d.get("tree") or fallback_tree,
142 say=d.get("say"),
143 output=d.get("output"),
144 reason=d.get("reason"),
145 question=d.get("question"),
146 snapshot=d.get("snapshot"),
147 trace=trace,
148 inputs=d.get("inputs") or {},
149 )
150
151
152class Service:
153 def __init__(
154 self,
155 registry: Registry,
156 *,
157 name: str = "pyrtan",
158 transport: Transport | None = None,
159 hub: str | None = None,
160 env: dict[str, Any] | None = None,
161 advertise: str | None = None,
162 token: str | None = None,
163 gaps: str | Path | GapStore | None = None,
164 egress: str = "any",
165 unanswered: str | Path | None = None,
166 watch: bool = False,
167 suggestions: str | Path | None = None,
168 ) -> None:
169 self.registry = registry
170 self.name = name
171 self.hub = hub.rstrip("/") if hub else None
172 self.advertise = advertise
173 self.token = token or None
174 self.egress = egress
175 self.started = time.time()
176 adapters = transport if transport is not None and hasattr(transport, "perform") else LocalAdapters(egress=egress)
177 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)
178 self.gaps = gaps if isinstance(gaps, GapStore) else GapStore(gaps)
179 self.unanswered = Unanswered(path=unanswered)
180 self.suggestions = Suggestions(path=suggestions)
181 self.runs = 0
182 self._lock = threading.Lock()
183 self.watcher: Watcher | None = None
184 if watch:
185 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()
186
187 def run_watched(self, tree: Tree, inputs: dict[str, Any], trace: Trace) -> Outcome:
188 """How a watch runs a tree: through the same runner and the same
189 bookkeeping as a request, with the watch step already on the trace."""
190 outcome = self.runner.run(tree, inputs, trace=trace)
191 self._finish(outcome, {})
192 return outcome
193
194 def stop(self) -> None:
195 if self.watcher is not None:
196 self.watcher.stop()
197
198 # ------------------------------------------------------------ handlers
199
200 def health(self) -> dict[str, Any]:
201 self.registry.reload()
202 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), "suggestions": len(self.suggestions), "egress": self.egress, "uptime_s": int(time.time() - self.started), "watches": len(self.watcher) if self.watcher else 0}
203 if self.gaps.error:
204 health["gaps_file_error"] = self.gaps.error
205 return health
206
207 def capabilities(self) -> dict[str, Any]:
208 self.registry.reload()
209 return {"name": self.name, "trees": self.registry.capabilities()}
210
211 def intent(self, body: dict[str, Any]) -> dict[str, Any]:
212 text = body.get("text")
213 if not isinstance(text, str) or not text.strip():
214 return {"status": "error", "reason": "bad-request", "say": "Send {\"text\": \"...\"}."}
215 self.registry.reload()
216 outcome = self.runner.run_intent(text, int(body.get("depth", 0)))
217 if outcome.reason == "no-match" and self.hub is None:
218 # With a hub, the hub saw the sentence and keeps the record.
219 self.unanswered.record(text)
220 return self._finish(outcome, body)
221
222 def run(self, body: dict[str, Any]) -> dict[str, Any]:
223 tree = body.get("tree")
224 if not isinstance(tree, str):
225 return {"status": "error", "reason": "bad-request", "say": "Send {\"tree\": \"...\", \"inputs\": {...}}."}
226 self.registry.reload()
227 return self._finish(self.runner.run_tree(tree, body.get("inputs") or {}, int(body.get("depth", 0))), body)
228
229 def resume(self, body: dict[str, Any]) -> dict[str, Any]:
230 snapshot = body.get("snapshot")
231 if not isinstance(snapshot, dict) or "tree" not in snapshot:
232 return {"status": "error", "reason": "bad-request", "say": "Send {\"snapshot\": {...}, \"answer\": ...}."}
233 return self._finish(self.runner.resume(snapshot, body.get("answer")), body)
234
235 def _finish(self, outcome: Outcome, body: dict[str, Any]) -> dict[str, Any]:
236 with self._lock:
237 self.runs += 1
238 if outcome.trace is not None and outcome.trace.gaps:
239 self.gaps.add(outcome.trace.gaps)
240 return outcome.to_dict(with_trace=bool(body.get("trace", True)))
241
242 # ------------------------------------------------------------ hub
243
244 def register(self, my_url: str) -> bool:
245 if not self.hub:
246 return False
247 try:
248 status, _ = http_json("POST", f"{self.hub}/register", {"url": my_url, "name": self.name}, token=self.token)
249 return status == 200
250 except Exception:
251 return False
252
253 # ------------------------------------------------------------ routing
254
255 def route(self, method: str, path: str, body: dict[str, Any] | None, headers: dict[str, str] | None = None) -> tuple[int, Any]:
256 body = body or {}
257 if method == "GET" and path == "/health":
258 return 200, self.health()
259 if method == "GET" and path == "/capabilities":
260 return 200, self.capabilities()
261 if method == "GET" and path == "/gaps":
262 return 200, {"gaps": self.gaps.all()}
263 if method == "GET" and path == "/unanswered":
264 return 200, {"unanswered": self.unanswered.to_list()}
265 if method in ("GET", "POST") and path == "/suggestions":
266 return self.suggestions.route(method, body)
267 if method == "GET" and path == "/watches":
268 return 200, self.watcher.status() if self.watcher else {"count": 0, "watches": [], "watching": False}
269 if method == "GET" and path in ("/", "/index.html", "/console"):
270 return 200, PAGE
271 if method == "GET" and path == "/manifest.webmanifest":
272 return 200, ("application/manifest+json", MANIFEST)
273 if method == "GET" and path == "/sw.js":
274 return 200, ("text/javascript; charset=utf-8", SERVICE_WORKER)
275 if method == "GET" and path == "/pyrtan.js":
276 return 200, ("text/javascript; charset=utf-8", PYRTAN_JS)
277 if method == "GET" and path == "/api":
278 return 200, {"pyrtan": "service", "name": self.name, "version": __version__, "auth": "bearer" if self.token else "none", "endpoints": ["/health", "/capabilities", "/intent", "/run", "/resume", "/gaps", "/unanswered", "/suggestions", "/watches"]}
279 if method == "POST" and path == "/intent":
280 return 200, self.intent(body)
281 if method == "POST" and path == "/run":
282 return 200, self.run(body)
283 if method == "POST" and path == "/resume":
284 return 200, self.resume(body)
285 return 404, {"status": "error", "reason": "not-found", "say": f"No {method} {path} here."}
286
287
288Route = Callable[..., tuple[int, Any]] # (method, path, body, headers) -> (status, payload)
289
290
291def bearer_ok(header: str | None, token: str) -> bool:
292 """Whether an ``Authorization`` header carries ``token`` (compared in constant time)."""
293 if not header:
294 return False
295 scheme, _, value = header.strip().partition(" ")
296 return scheme.lower() == "bearer" and hmac.compare_digest(value.strip().encode("utf-8"), token.encode("utf-8"))
297
298
299def make_handler(route: Route, quiet: bool = False, token: str | None = None, cors: list[str] | None = None, federated: bool = False, public: Callable[[str], bool] | None = None, keyed: bool = False):
300 token = token or None
301 origins = [o.rstrip("/") for o in (cors or []) if o]
302 any_origin = "*" in origins
303
304 class Handler(BaseHTTPRequestHandler):
305 server_version = f"pyrtan/{__version__}"
306
307 def log_message(self, fmt: str, *args: Any) -> None:
308 if not quiet:
309 super().log_message(fmt, *args)
310
311 # -------------------------------------------------------- cors
312
313 def _origin_allowed(self) -> str | None:
314 origin = self.headers.get("Origin")
315 if not origin or not origins:
316 return None
317 return origin if any_origin or origin.rstrip("/") in origins else None
318
319 def _cors_headers(self) -> None:
320 origin = self._origin_allowed()
321 if origin is None:
322 return
323 self.send_header("Access-Control-Allow-Origin", origin)
324 self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
325 self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type, " + ", ".join(SIGNATURE_HEADERS))
326 self.send_header("Access-Control-Max-Age", "600")
327 self.send_header("Vary", "Origin")
328
329 def do_OPTIONS(self) -> None:
330 self.send_response(204)
331 self._cors_headers()
332 self.send_header("Content-Length", "0")
333 self.end_headers()
334
335 # -------------------------------------------------------- auth
336
337 def _authorised(self, path: str) -> bool:
338 """True if the request may proceed; otherwise a 401 has been sent."""
339 if token is None or public_path(path) or bearer_ok(self.headers.get("Authorization"), token) or (public is not None and public(path)):
340 return True
341 if federated and signed_path(path) and self.headers.get("X-Pyrtan-Signature"):
342 return True # the route verifies the signature
343 if keyed and self.headers.get("X-Pyrtan-Key-Signature"):
344 return True # the route verifies the key and asks its grants
345 self._send(401, {"status": "error", "reason": "unauthorised", "say": "This Pyrtan needs a token: send Authorization: Bearer <token>."}, extra={"WWW-Authenticate": 'Bearer realm="pyrtan"'})
346 return False
347
348 def _public_view(self, path: str, payload: Any) -> Any:
349 """Trim /health for a caller with no token, when one is required."""
350 if token is not None and path == "/health" and isinstance(payload, dict) and not bearer_ok(self.headers.get("Authorization"), token):
351 return {k: v for k, v in payload.items() if k in PUBLIC_HEALTH_KEYS}
352 return payload
353
354 # -------------------------------------------------------- i/o
355
356 def _send(self, status: int, payload: Any, extra: dict[str, str] | None = None) -> None:
357 if isinstance(payload, tuple) and len(payload) == 2 and isinstance(payload[0], str):
358 ctype, text = payload # (content type, text): the manifest, the service worker
359 data = text.encode("utf-8")
360 elif isinstance(payload, str):
361 data = payload.encode("utf-8")
362 ctype = "text/html; charset=utf-8"
363 else:
364 data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
365 ctype = "application/json; charset=utf-8"
366 self.send_response(status)
367 self.send_header("Content-Type", ctype)
368 self.send_header("Content-Length", str(len(data)))
369 self.send_header("X-Content-Type-Options", "nosniff")
370 for k, v in (extra or {}).items():
371 self.send_header(k, v)
372 self._cors_headers()
373 self.end_headers()
374 try:
375 self.wfile.write(data)
376 except (BrokenPipeError, ConnectionResetError):
377 pass # the caller went away; nothing to do about it
378
379 def do_GET(self) -> None:
380 path = self.path.split("?", 1)[0]
381 if not self._authorised(path):
382 return
383 status, payload = route("GET", path, None, dict(self.headers))
384 self._send(status, self._public_view(path, payload))
385
386 def do_POST(self) -> None:
387 path = self.path.split("?", 1)[0]
388 if not self._authorised(path):
389 return
390 length = int(self.headers.get("Content-Length") or 0)
391 if length > MAX_BODY:
392 self._send(413, {"status": "error", "reason": "too-large"})
393 return
394 raw = self.rfile.read(length) if length else b""
395 try:
396 body = json.loads(raw.decode("utf-8")) if raw else {}
397 except (json.JSONDecodeError, UnicodeDecodeError):
398 self._send(400, {"status": "error", "reason": "bad-json"})
399 return
400 if not isinstance(body, dict):
401 self._send(400, {"status": "error", "reason": "bad-json", "say": "The body must be a JSON object."})
402 return
403 status, payload = route("POST", path, body, dict(self.headers))
404 self._send(status, payload)
405
406 return Handler
407
408
409class Server:
410 """A running HTTP server around a route function, for a service or a hub.
411
412 ``token`` requires ``Authorization: Bearer`` on everything but the public
413 paths, and any path ``public`` says yes to (a hub's open lobby); ``cors``
414 lists the browser origins allowed to call (``*`` for any)."""
415
416 def __init__(self, route: Route, host: str = "127.0.0.1", port: int = 0, quiet: bool = False, *, token: str | None = None, cors: list[str] | None = None, federated: bool = False, public: Callable[[str], bool] | None = None, keyed: bool = False) -> None:
417 self.httpd = ThreadingHTTPServer((host, port), make_handler(route, quiet, token, cors, federated, public, keyed))
418 self.httpd.daemon_threads = True
419 self.thread: threading.Thread | None = None
420
421 @property
422 def port(self) -> int:
423 return self.httpd.server_address[1]
424
425 @property
426 def url(self) -> str:
427 host = self.httpd.server_address[0]
428 return f"http://{host}:{self.port}"
429
430 def start(self) -> "Server":
431 self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
432 self.thread.start()
433 return self
434
435 def serve_forever(self) -> None:
436 self.httpd.serve_forever()
437
438 def wait(self) -> None:
439 """Block until the server started with :meth:`start` stops."""
440 if self.thread is None:
441 self.serve_forever()
442 return
443 while self.thread.is_alive():
444 self.thread.join(0.5)
445
446 def stop(self) -> None:
447 self.httpd.shutdown()
448 self.httpd.server_close()
449
450
451def serve(
452 paths: list[str | Path],
453 host: str = "127.0.0.1",
454 port: int = 8701,
455 *,
456 name: str = "pyrtan",
457 hub: str | None = None,
458 env: dict[str, Any] | None = None,
459 advertise: str | None = None,
460 quiet: bool = False,
461 token: str | None = None,
462 cors: list[str] | None = None,
463 gaps: str | Path | None = None,
464 egress: str = "any",
465 unanswered: str | Path | None = None,
466 watch: bool = False,
467 suggestions: str | Path | None = None,
468) -> tuple[Service, Server]:
469 registry = Registry.from_paths(paths)
470 service = Service(registry, name=name, hub=hub, env=env, advertise=advertise, token=token, gaps=gaps, egress=egress, unanswered=unanswered, watch=watch, suggestions=suggestions)
471 server = Server(service.route, host, port, quiet, token=token, cors=cors).start()
472 # The server is already answering when it registers: the hub checks a new
473 # peer's /capabilities inside /register, and would otherwise wait on a
474 # service that is itself waiting for the hub's answer.
475 if hub:
476 service.register(advertise or server.url)
477 return service, server