pyrtan / sourcev0.19.0 · main

pyrtan / src / pyrtan / runtime.py

21,314 bytes · last changed 2026-10-26 in Forms of I/O: file, shell, serial, queue and browser sources, each behind an allow-list; release 0.7.0

1"""Executing a tree.
2
3The :class:`Runner` walks a tree from ``start`` to a terminal node, keeping a
4state table, a trace of every step, and the tree's host allow-list in front
5of every request. It returns an :class:`Outcome`.
6
7An outcome's ``status`` is one of:
8
9``done``
10 A ``done`` node was reached; ``say`` and ``output`` are set.
11``failed``
12 A ``fail`` node was reached; ``reason`` names which, and ``say`` explains.
13``asked``
14 An ``ask`` node paused the run; ``question`` holds the prompt and
15 ``snapshot`` what :meth:`Runner.resume` needs to continue.
16``error``
17 The run could not proceed: a transport failure, an unhandled observation
18 (a ``decide`` with no matching branch, recorded as a gap), a host outside
19 the allow-list, a missing template value, or a limit reached. A
20 ``no-match`` error carries, in ``output.proposal``, the tree that would
21 answer the sentence, so a dead end always points at the next tree.
22
23Nothing here calls a model. The runner is the whole of Pyrtan's "thinking";
24the trace is the whole of its explanation.
25"""
26
27from __future__ import annotations
28
29import datetime as _dt
30import time
31from dataclasses import dataclass, field
32from typing import Any, Callable, Protocol
33
34from . import conditions
35from .adapters import SOURCES, LocalAdapters, brief_result, what
36from .extract import extract
37from .grow import no_match_outcome
38from .registry import Ambiguous, NoMatch, Registry
39from .template import MISSING, TemplateError, render, resolve
40from .trace import Step, Trace
41from .transport import EgressRefused, Transport, TransportError, UrllibTransport
42from .tree import Node, Tree, TreeError
43
44MAX_STEPS = 500
45MAX_DEPTH = 8
46DEFAULT_TIMEOUT = 10.0
47DEFAULT_RETRIES = 1
48
49
50class Resolver(Protocol):
51 """Where ``delegate`` and ``each ... tree`` go when a tree is not local."""
52
53 def run_tree(self, name: str, inputs: dict[str, Any], depth: int) -> "Outcome": ...
54
55 def run_intent(self, text: str, depth: int) -> "Outcome": ...
56
57
58@dataclass
59class Outcome:
60 status: str
61 tree: str
62 say: str | None = None
63 output: Any = None
64 reason: str | None = None
65 question: str | None = None
66 snapshot: dict[str, Any] | None = None
67 trace: Trace | None = None
68 inputs: dict[str, Any] = field(default_factory=dict)
69
70 @property
71 def ok(self) -> bool:
72 return self.status == "done"
73
74 def to_dict(self, *, with_trace: bool = True) -> dict[str, Any]:
75 d: dict[str, Any] = {"status": self.status, "tree": self.tree, "inputs": self.inputs}
76 if self.say is not None:
77 d["say"] = self.say
78 if self.output is not None:
79 d["output"] = self.output
80 if self.reason is not None:
81 d["reason"] = self.reason
82 if self.question is not None:
83 d["question"] = self.question
84 if self.snapshot is not None:
85 d["snapshot"] = self.snapshot
86 if with_trace and self.trace is not None:
87 d["trace"] = self.trace.to_dict()
88 return d
89
90
91class _Halt(Exception):
92 def __init__(self, outcome: Outcome) -> None:
93 self.outcome = outcome
94
95
96@dataclass
97class Runner:
98 registry: Registry
99 transport: Transport = field(default_factory=UrllibTransport)
100 resolver: Resolver | None = None
101 env: dict[str, Any] = field(default_factory=dict)
102 clock: Callable[[], _dt.datetime] = field(default=lambda: _dt.datetime.now(_dt.timezone.utc))
103 max_steps: int = MAX_STEPS
104 max_depth: int = MAX_DEPTH
105 emitted: list[dict[str, Any]] = field(default_factory=list)
106 adapters: Any = None # the other sources (file, shell, serial, queue, browser); a recorded transport answers for them too
107
108 def __post_init__(self) -> None:
109 if self.adapters is None:
110 self.adapters = self.transport if hasattr(self.transport, "perform") else LocalAdapters()
111
112 # ------------------------------------------------------------ entry points
113
114 def run_intent(self, text: str, depth: int = 0) -> Outcome:
115 """Read a sentence and run the tree it names. The trace begins with an
116 ``intent`` step: what the sentence was rewritten to, which rules
117 fired, and which trigger took it."""
118 reading = self.registry.read(text, self.clock())
119 if not reading.matches:
120 if self.resolver is not None:
121 return self.resolver.run_intent(text, depth)
122 out = Outcome("error", "", reason="no-match", **no_match_outcome(text, "Nothing here"))
123 out.output = {**(out.output or {}), "reading": {"cleaned": reading.cleaned, "steps": reading.steps, "considered": [m.trigger.pattern for m in reading.considered]}}
124 return out
125 if len({m.tree for m in reading.matches}) > 1:
126 trees = sorted({m.tree for m in reading.matches})
127 return Outcome("error", "", reason="ambiguous", say=f"{text!r} matches more than one tree equally well: {', '.join(trees)}", output={"candidates": trees, "reading": {"cleaned": reading.cleaned, "steps": reading.steps}})
128 match = reading.matches[0]
129 tree = self.registry.get(match.tree)
130 if tree is None:
131 return self.run_tree(match.tree, match.inputs, depth)
132 trace = Trace(tree.name)
133 step = reading.step()
134 trace.add(Step(step["node"], step["kind"], step["summary"], observed=step["observed"], next=tree.start, depth=depth))
135 return self.run(tree, match.inputs, depth=depth, trace=trace)
136
137 def run_tree(self, name: str, inputs: dict[str, Any] | None = None, depth: int = 0) -> Outcome:
138 tree = self.registry.get(name)
139 if tree is None:
140 if self.resolver is not None:
141 return self.resolver.run_tree(name, inputs or {}, depth)
142 return Outcome("error", name, reason="no-such-tree", say=f"No tree named {name!r}.")
143 return self.run(tree, inputs or {}, depth=depth)
144
145 def run(self, tree: Tree, inputs: dict[str, Any] | None = None, *, depth: int = 0, start: str | None = None, state: dict[str, Any] | None = None, scope_extra: dict[str, Any] | None = None, trace: Trace | None = None) -> Outcome:
146 trace = trace or Trace(tree.name)
147 if depth > self.max_depth:
148 return Outcome("error", tree.name, reason="too-deep", say=f"Delegation deeper than {self.max_depth} levels.", trace=trace)
149 try:
150 bound = tree.bind_inputs(inputs or {})
151 except TreeError as e:
152 return Outcome("error", tree.name, reason="bad-input", say=str(e), trace=trace)
153 state = dict(state) if state is not None else {}
154 scope: dict[str, Any] = {"inputs": bound, "state": state, "tree": {"name": tree.name, "version": tree.version}, "now": _now(self.clock()), "env": self.env}
155 if scope_extra:
156 scope.update(scope_extra)
157 try:
158 return self._walk(tree, start or tree.start, scope, trace, depth)
159 except _Halt as h:
160 h.outcome.inputs = bound
161 h.outcome.trace = h.outcome.trace or trace
162 return h.outcome
163
164 def resume(self, snapshot: dict[str, Any], answer: Any) -> Outcome:
165 """Continue a run paused by ``ask``. The snapshot is what the asked
166 outcome carried; the answer goes where the ask node said."""
167 tree = self.registry.get(snapshot["tree"])
168 if tree is None:
169 return Outcome("error", snapshot.get("tree", ""), reason="no-such-tree")
170 node = tree.node(snapshot["node"])
171 into: str = node.get("into")
172 inputs = dict(snapshot["inputs"])
173 state = dict(snapshot["state"])
174 if into.startswith("inputs."):
175 key = into[len("inputs."):]
176 try:
177 inputs[key] = tree.inputs[key].coerce(answer)
178 except TreeError as e:
179 return Outcome("error", tree.name, reason="bad-input", say=str(e))
180 else:
181 _put(state, into[len("state."):] if into.startswith("state.") else into, answer)
182 trace = Trace(tree.name)
183 trace.steps = [Step(**s) for s in snapshot.get("trace", [])]
184 trace.add(Step(node.id, "ask", f"answered: {_brief(answer)}", next=node.next))
185 return self.run(tree, inputs, start=node.next, state=state, trace=trace)
186
187 # ------------------------------------------------------------ the walk
188
189 def _walk(self, tree: Tree, start: str, scope: dict[str, Any], trace: Trace, depth: int) -> Outcome:
190 state: dict[str, Any] = scope["state"]
191 node_id = start
192 steps = 0
193 while True:
194 steps += 1
195 if steps > self.max_steps:
196 raise _Halt(Outcome("error", tree.name, reason="too-many-steps", say=f"More than {self.max_steps} steps.", trace=trace))
197 node = tree.node(node_id)
198 started = time.monotonic()
199 handler = getattr(self, f"_do_{node.kind}")
200 try:
201 result = handler(tree, node, scope, trace, depth)
202 except TemplateError as e:
203 trace.add(Step(node.id, node.kind, f"template: {e}", depth=depth))
204 raise _Halt(Outcome("error", tree.name, reason="template", say=f"{node.id}: {e}", trace=trace))
205 if isinstance(result, Outcome):
206 result.trace = trace
207 raise _Halt(result)
208 step, next_id = result
209 step.elapsed_ms = int((time.monotonic() - started) * 1000)
210 step.depth = depth
211 step.next = next_id
212 trace.add(step)
213 if next_id is None:
214 raise _Halt(Outcome("error", tree.name, reason="dead-end", say=f"{node.id} has nowhere to go.", trace=trace))
215 node_id = next_id
216
217 # ------------------------------------------------------------ node kinds
218
219 def _do_observe(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
220 return self._observe_or_act(tree, node, scope, trace, retry=True)
221
222 def _do_act(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
223 emit = node.get("emit")
224 if emit is not None:
225 event = {"event": render(emit["event"], scope), "data": render(emit.get("data", {}), scope), "tree": tree.name}
226 self.emitted.append(event)
227 _put(scope["state"], node.get("into", "last"), event)
228 return Step(node.id, "act", f"emitted {event['event']}", observed=event["data"]), node.next
229 return self._observe_or_act(tree, node, scope, trace, retry=False)
230
231 def _observe_or_act(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, *, retry: bool):
232 state = scope["state"]
233 into = node.get("into", "last")
234 source_name = next((s for s in SOURCES if node.get(s) is not None), None)
235 if source_name is not None:
236 spec = render(node.get(source_name), scope)
237 problem = tree.allowed(source_name, spec)
238 if problem is not None:
239 return Outcome("error", tree.name, reason=problem[0], say=f"{node.id}: {problem[1]}")
240 timeout = float(node.get("timeout", DEFAULT_TIMEOUT))
241 if source_name == "http":
242 method = str(spec.get("method", "GET")).upper()
243 url = str(spec["url"])
244 headers = {str(k): str(v) for k, v in (spec.get("headers") or {}).items()}
245 body = spec.get("body")
246 if body is not None and not isinstance(body, str):
247 import json as _json
248
249 body = _json.dumps(body)
250 headers.setdefault("Content-Type", "application/json")
251 label = f"{method} {url}"
252 call = lambda: self.transport.request(method, url, headers, body, timeout).to_state() # noqa: E731
253 else:
254 label = f"{source_name} {what(source_name, spec)}"
255 call = lambda: self.adapters.perform(source_name, spec, timeout, tree.host_allowed) # noqa: E731
256 attempts = 1 + int(node.get("retries", DEFAULT_RETRIES)) if retry else 1
257 last_error: str | None = None
258 observed = None
259 for attempt in range(attempts):
260 try:
261 observed = call()
262 break
263 except TransportError as e:
264 last_error = str(e)
265 if isinstance(e, EgressRefused):
266 break # a policy refusal does not change on retry
267 if attempt + 1 < attempts:
268 time.sleep(min(0.2 * (attempt + 1), 1.0))
269 if observed is None:
270 trace.add(Step(node.id, node.kind, f"{label} failed: {last_error}"))
271 return Outcome("error", tree.name, reason="transport", say=f"{node.id}: {last_error}")
272 _put(state, into, observed)
273 source = observed
274 summary = f"{label} -> {brief_result(source_name, observed)}"
275 else:
276 frm = node.get("from")
277 source_val = resolve(_state_path(frm), scope)
278 if source_val is MISSING:
279 source = {}
280 elif isinstance(source_val, dict):
281 source = source_val
282 else:
283 source = {"value": source_val, "body": source_val if isinstance(source_val, str) else None}
284 summary = f"from {frm}"
285 extracted: dict[str, Any] = {}
286 cache: dict[int, Any] = {}
287 for entry in node.get("extract", []) or []:
288 value = extract(entry, source, cache)
289 if value is MISSING:
290 extracted[entry["into"]] = None
291 else:
292 _put(state, entry["into"], value)
293 extracted[entry["into"]] = value
294 if extracted:
295 summary += "; " + ", ".join(f"{k}={_brief(v)}" for k, v in extracted.items())
296 return Step(node.id, node.kind, summary, observed=extracted or None), node.next
297
298 def _do_decide(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
299 on = _state_path(node.get("on"))
300 value = resolve(on, scope)
301 for i, br in enumerate(node.get("branches")):
302 if "else" in br:
303 return Step(node.id, "decide", f"{on} = {_brief(value)}", observed=_plain(value), chosen="else"), br["else"]
304 if conditions.holds(br["when"], value):
305 return Step(node.id, "decide", f"{on} = {_brief(value)}", observed=_plain(value), chosen=_cond_text(br["when"])), br["then"]
306 trace.gap(node.id, on, _plain(value))
307 trace.add(Step(node.id, "decide", f"{on} = {_brief(value)}: no branch", observed=_plain(value), depth=depth))
308 return Outcome("error", tree.name, reason="gap", say=f"{node.id}: no branch for {on} = {_brief(value)}.", output={"gap": trace.gaps[-1]})
309
310 def _do_set(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
311 state = scope["state"]
312 values = {}
313 for key, tmpl in node.get("set").items():
314 value = render(tmpl, scope)
315 _put(state, key, value)
316 values[key] = value
317 return Step(node.id, "set", ", ".join(f"{k}={_brief(v)}" for k, v in values.items()), observed=values), node.next
318
319 def _do_each(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
320 state = scope["state"]
321 items = resolve(_state_path(node.get("over")), scope)
322 if items is MISSING or items is None:
323 items = []
324 if not isinstance(items, list):
325 return Outcome("error", tree.name, reason="not-a-list", say=f"{node.id}: {node.get('over')} is not a list.")
326 limit = node.get("limit")
327 if limit is not None:
328 limit = int(render(limit, scope))
329 items = items[:limit]
330 results: list[Any] = []
331 failures = 0
332 for index, item in enumerate(items):
333 with_inputs = render(node.get("with", {}), {**scope, "item": item, "index": index})
334 if "tree" in node.spec:
335 sub = self.run_tree(node.get("tree"), with_inputs, depth + 1)
336 else:
337 sub = self.run(tree, scope["inputs"], depth=depth + 1, start=node.get("start"), state=state, scope_extra={"item": item, "index": index}, trace=trace)
338 if sub.status == "done":
339 results.append(sub.output if sub.output is not None else sub.say)
340 else:
341 failures += 1
342 results.append(None)
343 if node.get("stop_on_fail", False):
344 return Outcome("error", tree.name, reason="each-failed", say=f"{node.id}: item {index} failed: {sub.say or sub.reason}")
345 if sub.trace is not None and sub.trace is not trace:
346 for s in sub.trace.steps:
347 s.depth = depth + 1
348 trace.steps.append(s)
349 trace.gaps.extend(sub.trace.gaps)
350 _put(state, node.get("into"), results)
351 summary = f"{len(items)} item(s)" + (f", {failures} failed" if failures else "")
352 return Step(node.id, "each", summary), node.next
353
354 def _do_delegate(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
355 state = scope["state"]
356 with_inputs = render(node.get("with", {}), scope)
357 if "intent" in node.spec:
358 text = render(node.get("intent"), scope)
359 sub = self.run_intent(str(text), depth + 1)
360 summary = f"intent {text!r} -> {sub.tree or '?'}: {sub.status}"
361 else:
362 sub = self.run_tree(node.get("tree"), with_inputs, depth + 1)
363 summary = f"tree {node.get('tree')}: {sub.status}"
364 if sub.trace is not None and sub.trace is not trace:
365 for s in sub.trace.steps:
366 s.depth = depth + 1
367 trace.steps.append(s)
368 trace.gaps.extend(sub.trace.gaps)
369 _put(state, node.get("into"), {"status": sub.status, "say": sub.say, "output": sub.output, "reason": sub.reason, "tree": sub.tree})
370 if sub.status == "asked":
371 return Outcome("error", tree.name, reason="ask-in-delegate", say=f"{node.id}: {sub.tree} asked a question; delegated trees cannot pause.")
372 return Step(node.id, "delegate", summary), node.next
373
374 def _do_ask(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
375 if depth > 0:
376 return Outcome("error", tree.name, reason="ask-in-subflow", say=f"{node.id}: only a top-level run can ask.")
377 prompt = render(node.get("prompt"), scope)
378 trace.add(Step(node.id, "ask", f"asked: {prompt}", depth=depth))
379 snapshot = {"tree": tree.name, "node": node.id, "inputs": scope["inputs"], "state": scope["state"], "trace": [s.to_dict() for s in trace.steps]}
380 return Outcome("asked", tree.name, question=str(prompt), snapshot=snapshot)
381
382 def _do_done(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
383 say = node.get("say")
384 output = node.get("output")
385 say = render(say, scope, strict=False) if say is not None else None
386 output = render(output, scope, strict=False) if output is not None else None
387 trace.add(Step(node.id, "done", str(say) if say else "done", depth=depth))
388 return Outcome("done", tree.name, say=str(say) if say is not None else None, output=output)
389
390 def _do_fail(self, tree: Tree, node: Node, scope: dict[str, Any], trace: Trace, depth: int):
391 say = node.get("say")
392 say = render(say, scope, strict=False) if say is not None else None
393 reason = str(node.get("reason"))
394 trace.add(Step(node.id, "fail", f"{reason}: {say or ''}".rstrip(": "), depth=depth))
395 return Outcome("failed", tree.name, reason=reason, say=str(say) if say is not None else None, output=render(node.get("output"), scope, strict=False) if node.get("output") is not None else None)
396
397
398# ---------------------------------------------------------------- helpers
399
400
401def _state_path(path: str) -> str:
402 path = path.strip()
403 head = path.split(".")[0].split("[")[0]
404 if head in ("inputs", "state", "item", "index", "tree", "now", "env"):
405 return path
406 return "state." + path
407
408
409def _put(state: dict[str, Any], path: str, value: Any) -> None:
410 if path.startswith("state."):
411 path = path[len("state."):]
412 parts = path.split(".")
413 cur = state
414 for part in parts[:-1]:
415 nxt = cur.get(part)
416 if not isinstance(nxt, dict):
417 nxt = {}
418 cur[part] = nxt
419 cur = nxt
420 cur[parts[-1]] = value
421
422
423def _plain(value: Any) -> Any:
424 return None if value is MISSING else value
425
426
427def _brief(value: Any, limit: int = 80) -> str:
428 if value is MISSING:
429 return "<missing>"
430 text = repr(value) if not isinstance(value, str) else value
431 return text if len(text) <= limit else text[: limit - 1] + "…"
432
433
434def _cond_text(cond: dict[str, Any]) -> str:
435 return " and ".join(f"{k} {v!r}" for k, v in cond.items())
436
437
438def _now(now: _dt.datetime | None = None) -> dict[str, Any]:
439 now = now or _dt.datetime.now(_dt.timezone.utc)
440 return {"iso": now.isoformat(timespec="seconds"), "date": now.date().isoformat(), "epoch": int(now.timestamp()), "year": now.year}