Coverage for little_loops / transport.py: 18%
377 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-26 17:38 -0500
1"""Transport abstraction for the little-loops EventBus.
3A `Transport` is an additive sink for events emitted by `EventBus`. The Protocol
4is intentionally minimal — `send(event)` for delivery and `close()` for cleanup —
5so that new sinks can be added without modifying `EventBus` itself.
7Built-in implementations:
8 JsonlTransport: appends each event as a JSON line to a file.
9 UnixSocketTransport: streams newline-delimited JSON to AF_UNIX socket clients
10 for sub-second-latency local consumers (TUIs, log tailers, dashboards).
11 OTelTransport: maps loop executions to OpenTelemetry traces/spans, exporting
12 via OTLP to Grafana, Jaeger, Datadog, etc. Requires the optional
13 ``opentelemetry-sdk`` and ``opentelemetry-exporter-otlp-grpc`` packages.
14 WebhookTransport: POSTs batched events to an HTTP endpoint for remote
15 dashboards, Slack bots, and CI systems. Requires the optional ``httpx``
16 package (``pip install little-loops[webhooks]``).
17 SQLiteTransport: records FSM loop events into the per-project session
18 database (``.ll/history.db``) for indexed cross-cutting queries.
20Public exports:
21 Transport: runtime-checkable Protocol that any sink must satisfy
22 JsonlTransport: writes events to a JSONL file
23 UnixSocketTransport: streams events over an AF_UNIX socket
24 OTelTransport: exports loop traces via OTLP
25 WebhookTransport: POSTs batched events to an HTTP endpoint
26 wire_transports: register transports listed in `EventsConfig` on an `EventBus`
27"""
29from __future__ import annotations
31import json
32import logging
33import socket
34import threading
35import time
36from collections.abc import Callable
37from pathlib import Path
38from queue import Empty, Full, Queue
39from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
41if TYPE_CHECKING:
42 from little_loops.config.features import EventsConfig
43 from little_loops.events import EventBus
45logger = logging.getLogger(__name__)
47_CLIENT_QUEUE_MAXSIZE = 1024
48_DROP_LOG_INTERVAL_SEC = 5.0
49_REJECT_LOG_INTERVAL_SEC = 5.0
50_ACCEPT_THREAD_JOIN_TIMEOUT = 2.0
51_CLIENT_THREAD_JOIN_TIMEOUT = 1.0
52_CLOSE_TOTAL_TIMEOUT = 10.0
53_ACCEPT_POLL_TIMEOUT = 1.0
54_CLIENT_QUEUE_POLL_TIMEOUT = 0.5
56_WEBHOOK_BATCH_MS_DEFAULT = 1000
57_WEBHOOK_CLOSE_TIMEOUT = 10.0
58_WEBHOOK_RETRY_BASE_S = 0.5
59_WEBHOOK_RETRY_MAX_S = 8.0
62@runtime_checkable
63class Transport(Protocol):
64 """Protocol for an event sink registered on an `EventBus`.
66 A transport receives every event emitted on the bus (no filtering at the
67 transport layer; subscribe an observer with a filter for that). Implementations
68 must tolerate being called with arbitrary `dict[str, Any]` shapes — the bus
69 does not validate event contents.
70 """
72 def send(self, event: dict[str, Any]) -> None:
73 """Deliver a single event."""
74 ...
76 def close(self) -> None:
77 """Release any resources held by the transport. May be a no-op."""
78 ...
81class JsonlTransport:
82 """Append events to a JSONL file, one JSON object per line.
84 The parent directory is created at construction time so per-event writes do
85 not have to check it. `close()` is a no-op since each `send()` opens and
86 closes the file (matching the existing JSONL write pattern in this codebase).
87 """
89 def __init__(self, path: Path) -> None:
90 self._path = path
91 self._path.parent.mkdir(parents=True, exist_ok=True)
93 def send(self, event: dict[str, Any]) -> None:
94 with open(self._path, "a", encoding="utf-8") as f:
95 f.write(json.dumps(event) + "\n")
97 def close(self) -> None:
98 return None
101class _SocketClient:
102 """Per-client state: connection, outbound queue, write thread, drop counters."""
104 def __init__(self, conn: socket.socket, addr: str) -> None:
105 self.conn = conn
106 self.addr = addr
107 self.queue: Queue[bytes] = Queue(maxsize=_CLIENT_QUEUE_MAXSIZE)
108 self.thread: threading.Thread | None = None
109 self.dropped_total = 0
110 self.dropped_since_log = 0
111 self.last_drop_log_ts = 0.0
112 self.first_drop_logged = False
115class UnixSocketTransport:
116 """Stream events as newline-delimited JSON over an `AF_UNIX` socket.
118 On construction, binds the socket at ``path`` (after unlinking any stale
119 file), starts an accept thread, and accepts up to ``max_clients`` concurrent
120 consumers. Each accepted client gets its own daemon thread and bounded
121 outbound queue; ``send()`` enqueues the serialized event into every client
122 queue without blocking. A full queue causes the newest event to be dropped
123 (preserving causal order) and a rate-limited warning is logged.
125 A misbehaving / disconnected client is removed from the pool without
126 affecting other clients or the FSM thread.
128 Not available on platforms without ``AF_UNIX`` (e.g. Windows). The platform
129 check lives in :func:`wire_transports` so the user-facing error has clearer
130 placement; constructing this class on a platform without ``AF_UNIX`` will
131 raise an `OSError` from `socket.socket` directly.
132 """
134 def __init__(
135 self,
136 path: Path,
137 max_clients: int = 32,
138 on_connect: Callable[[_SocketClient], None] | None = None,
139 ) -> None:
140 if not hasattr(socket, "AF_UNIX"):
141 raise RuntimeError(
142 "UnixSocketTransport requires AF_UNIX, which is not available on this platform"
143 )
145 self._path = path
146 self._max_clients = max_clients
147 self._on_connect = on_connect
148 self._shutdown = threading.Event()
149 self._clients: list[_SocketClient] = []
150 self._clients_lock = threading.Lock()
151 self._rejections_total: int = 0
152 self._rejections_since_log: int = 0
153 self._last_reject_log_ts: float = 0.0
154 self._first_reject_logged: bool = False
156 self._path.parent.mkdir(parents=True, exist_ok=True)
157 self._path.unlink(missing_ok=True)
159 self._server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
160 try:
161 self._server.bind(str(self._path))
162 self._path.chmod(0o600)
163 self._server.listen(max_clients)
164 self._server.settimeout(_ACCEPT_POLL_TIMEOUT)
165 except Exception:
166 self._server.close()
167 self._path.unlink(missing_ok=True)
168 raise
170 self._accept_thread = threading.Thread(
171 target=self._accept_loop,
172 name="unix-socket-transport-accept",
173 daemon=True,
174 )
175 self._accept_thread.start()
177 def _accept_loop(self) -> None:
178 while not self._shutdown.is_set():
179 try:
180 conn, _ = self._server.accept()
181 except TimeoutError:
182 continue
183 except OSError:
184 # Server socket closed during shutdown — exit cleanly
185 return
187 with self._clients_lock:
188 if len(self._clients) >= self._max_clients:
189 self._record_rejection()
190 try:
191 conn.close()
192 except OSError:
193 pass
194 continue
195 client = _SocketClient(conn, addr=str(self._path))
196 client.thread = threading.Thread(
197 target=self._client_loop,
198 args=(client,),
199 name="unix-socket-transport-client",
200 daemon=True,
201 )
202 self._clients.append(client)
203 if self._on_connect is not None:
204 self._on_connect(client)
205 client.thread.start()
207 def _client_loop(self, client: _SocketClient) -> None:
208 try:
209 while not self._shutdown.is_set():
210 try:
211 payload = client.queue.get(timeout=_CLIENT_QUEUE_POLL_TIMEOUT)
212 except Empty:
213 continue
214 try:
215 client.conn.sendall(payload)
216 except OSError:
217 return
218 finally:
219 try:
220 client.conn.close()
221 except OSError:
222 pass
223 with self._clients_lock:
224 if client in self._clients:
225 self._clients.remove(client)
227 def send(self, event: dict[str, Any]) -> None:
228 payload = (json.dumps(event) + "\n").encode("utf-8")
229 with self._clients_lock:
230 snapshot = list(self._clients)
231 for client in snapshot:
232 try:
233 client.queue.put_nowait(payload)
234 except Full:
235 self._record_drop(client)
237 def _record_drop(self, client: _SocketClient) -> None:
238 client.dropped_total += 1
239 client.dropped_since_log += 1
240 now = time.monotonic()
241 if not client.first_drop_logged:
242 logger.warning(
243 "UnixSocketTransport: dropping events for slow client (queue full at %d)",
244 _CLIENT_QUEUE_MAXSIZE,
245 )
246 client.first_drop_logged = True
247 client.last_drop_log_ts = now
248 client.dropped_since_log = 0
249 return
250 if now - client.last_drop_log_ts >= _DROP_LOG_INTERVAL_SEC:
251 logger.warning(
252 "UnixSocketTransport: dropped %d events for slow client",
253 client.dropped_since_log,
254 )
255 client.last_drop_log_ts = now
256 client.dropped_since_log = 0
258 def _record_rejection(self) -> None:
259 """Rate-limited log for client rejections. Must be called under _clients_lock."""
260 self._rejections_total += 1
261 self._rejections_since_log += 1
262 now = time.monotonic()
263 if not self._first_reject_logged:
264 logger.warning(
265 "UnixSocketTransport: rejecting client; max_clients=%d reached",
266 self._max_clients,
267 )
268 self._first_reject_logged = True
269 self._last_reject_log_ts = now
270 self._rejections_since_log = 0
271 return
272 if now - self._last_reject_log_ts >= _REJECT_LOG_INTERVAL_SEC:
273 logger.warning(
274 "UnixSocketTransport: rejected %d clients (max_clients=%d reached)",
275 self._rejections_since_log,
276 self._max_clients,
277 )
278 self._last_reject_log_ts = now
279 self._rejections_since_log = 0
281 def get_stats(self) -> dict[str, int]:
282 """Return transport-level statistics."""
283 return {"client_rejections": self._rejections_total}
285 def close(self) -> None:
286 deadline = time.monotonic() + _CLOSE_TOTAL_TIMEOUT
287 self._shutdown.set()
289 accept_budget = min(_ACCEPT_THREAD_JOIN_TIMEOUT, max(0.0, deadline - time.monotonic()))
290 if self._accept_thread.is_alive():
291 self._accept_thread.join(timeout=accept_budget)
292 if self._accept_thread.is_alive():
293 logger.warning(
294 "UnixSocketTransport: accept thread did not exit within %.1fs",
295 accept_budget,
296 )
298 try:
299 self._server.close()
300 except OSError:
301 pass
303 with self._clients_lock:
304 snapshot = list(self._clients)
305 for client in snapshot:
306 try:
307 client.conn.shutdown(socket.SHUT_RDWR)
308 except OSError:
309 pass
310 t = client.thread
311 if t is not None and t.is_alive():
312 budget = min(_CLIENT_THREAD_JOIN_TIMEOUT, max(0.0, deadline - time.monotonic()))
313 t.join(timeout=budget)
314 if t.is_alive():
315 logger.warning(
316 "UnixSocketTransport: client thread did not exit within %.1fs",
317 budget,
318 )
320 self._path.unlink(missing_ok=True)
323_OTEL_EVENT_TYPES = frozenset(
324 {
325 "evaluate",
326 "route",
327 "retry_exhausted",
328 "cycle_detected",
329 "stall_detected",
330 "handoff_detected",
331 "handoff_spawned",
332 "action_output",
333 }
334)
335_OTEL_ERROR_OUTCOMES = frozenset({"error", "failed", "exhausted"})
338class OTelTransport:
339 """Map ll loop executions to OpenTelemetry traces and spans, exporting via OTLP.
341 Span hierarchy: loop = trace root, state = child span, action = grandchild.
342 Span events are added for evaluate, route, retry_exhausted, handoff_detected,
343 handoff_spawned, and action_output on the innermost open span.
345 Requires ``opentelemetry-sdk`` and ``opentelemetry-exporter-otlp-grpc``.
346 Install with: ``pip install 'little-loops[otel]'``
348 Sub-loop events (``depth > 0``) are no-ops with a single warning per session.
349 """
351 def __init__(
352 self,
353 endpoint: str = "http://localhost:4317",
354 service_name: str = "little-loops",
355 *,
356 _tracer_provider: Any | None = None,
357 ) -> None:
358 try:
359 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
360 from opentelemetry.sdk.resources import Resource
361 from opentelemetry.sdk.trace import TracerProvider
362 from opentelemetry.sdk.trace.export import BatchSpanProcessor
363 except ImportError as exc:
364 raise RuntimeError(
365 "OTelTransport requires the 'opentelemetry-sdk' and "
366 "'opentelemetry-exporter-otlp-grpc' packages. "
367 "Install with: pip install 'little-loops[otel]'"
368 ) from exc
370 if _tracer_provider is not None:
371 self._provider = _tracer_provider
372 else:
373 resource = Resource.create({"service.name": service_name})
374 provider = TracerProvider(resource=resource)
375 exporter = OTLPSpanExporter(endpoint=endpoint)
376 provider.add_span_processor(BatchSpanProcessor(exporter))
377 self._provider = provider
379 self._tracer = self._provider.get_tracer("little-loops")
380 self._loop_span: Any | None = None
381 self._state_span: Any | None = None
382 self._action_span: Any | None = None
383 self._subloop_warned = False
385 def send(self, event: dict[str, Any]) -> None:
386 depth = event.get("depth", 0)
387 if isinstance(depth, int) and depth > 0:
388 if not self._subloop_warned:
389 logger.warning(
390 "OTelTransport: sub-loop events (depth > 0) are not supported; "
391 "nested-trace support is deferred. Event type: %r",
392 event.get("event"),
393 )
394 self._subloop_warned = True
395 return
397 event_type = event.get("event", "")
398 if event_type == "loop_start":
399 self._handle_loop_start(event)
400 elif event_type == "loop_resume":
401 self._handle_loop_resume(event)
402 elif event_type == "state_enter":
403 self._handle_state_enter(event)
404 elif event_type == "action_start":
405 self._handle_action_start(event)
406 elif event_type == "action_complete":
407 self._handle_action_complete()
408 elif event_type == "loop_complete":
409 self._handle_loop_complete(event)
410 elif event_type in _OTEL_EVENT_TYPES:
411 self._add_span_event(event_type, event)
413 def close(self) -> None:
414 self._provider.force_flush()
415 self._provider.shutdown()
417 # ------------------------------------------------------------------
418 # Internal span machine
419 # ------------------------------------------------------------------
421 def _handle_loop_start(self, event: dict[str, Any]) -> None:
422 loop_name = str(event.get("loop_name", "ll-loop"))
423 self._loop_span = self._tracer.start_span(loop_name)
425 def _handle_loop_resume(self, event: dict[str, Any]) -> None:
426 self._close_state_and_action()
427 if self._loop_span is not None:
428 self._loop_span.end()
429 loop_name = str(event.get("loop_name", "ll-loop"))
430 self._loop_span = self._tracer.start_span(loop_name)
432 def _handle_state_enter(self, event: dict[str, Any]) -> None:
433 self._close_state_and_action()
434 if self._loop_span is None:
435 logger.warning(
436 "OTelTransport: state_enter received without a prior loop_start; skipping span"
437 )
438 return
439 from opentelemetry import trace
441 state_name = str(event.get("state", "unknown-state"))
442 ctx = trace.set_span_in_context(self._loop_span)
443 self._state_span = self._tracer.start_span(state_name, context=ctx)
445 def _handle_action_start(self, event: dict[str, Any]) -> None:
446 if self._state_span is None:
447 logger.warning(
448 "OTelTransport: action_start received without a prior state_enter; skipping span"
449 )
450 return
451 from opentelemetry import trace
453 action_name = str(event.get("action", "unknown-action"))
454 ctx = trace.set_span_in_context(self._state_span)
455 self._action_span = self._tracer.start_span(action_name, context=ctx)
457 def _handle_action_complete(self) -> None:
458 if self._action_span is not None:
459 self._action_span.end()
460 self._action_span = None
462 def _handle_loop_complete(self, event: dict[str, Any]) -> None:
463 from opentelemetry.trace import StatusCode
465 self._close_state_and_action()
466 if self._loop_span is None:
467 logger.warning(
468 "OTelTransport: loop_complete received without a prior loop_start; skipping"
469 )
470 return
471 outcome = str(event.get("outcome", ""))
472 if outcome in _OTEL_ERROR_OUTCOMES:
473 self._loop_span.set_status(StatusCode.ERROR, outcome)
474 else:
475 self._loop_span.set_status(StatusCode.OK)
476 self._loop_span.end()
477 self._loop_span = None
479 def _add_span_event(self, event_type: str, event: dict[str, Any]) -> None:
480 span = self._action_span or self._state_span or self._loop_span
481 if span is None:
482 return
483 attrs = {k: str(v) for k, v in event.items() if k != "event"}
484 span.add_event(event_type, attributes=attrs)
486 def _close_state_and_action(self) -> None:
487 if self._action_span is not None:
488 self._action_span.end()
489 self._action_span = None
490 if self._state_span is not None:
491 self._state_span.end()
492 self._state_span = None
495class WebhookTransport:
496 """POSTs batched FSM events to an HTTP endpoint.
498 Events are enqueued non-blocking in ``send()`` and flushed by a daemon
499 thread on a configurable interval. Failed POSTs are retried with
500 exponential backoff; after ``max_retries`` the batch is dropped with a
501 warning rather than raising to the caller.
503 Requires ``httpx``: ``pip install little-loops[webhooks]``.
504 """
506 def __init__(
507 self,
508 url: str,
509 batch_ms: int = _WEBHOOK_BATCH_MS_DEFAULT,
510 headers: dict[str, str] | None = None,
511 max_retries: int = 3,
512 ) -> None:
513 try:
514 import httpx as _httpx
515 except ImportError as exc:
516 raise RuntimeError(
517 "WebhookTransport requires httpx: pip install little-loops[webhooks]"
518 ) from exc
519 self._httpx = _httpx
520 self._url = url
521 self._batch_ms = batch_ms
522 self._headers = dict(headers) if headers else {}
523 self._max_retries = max_retries
524 self._queue: Queue[dict[str, Any]] = Queue()
525 self._shutdown = threading.Event()
526 self._thread = threading.Thread(target=self._batch_loop, daemon=True, name="webhook-batch")
527 self._thread.start()
529 def send(self, event: dict[str, Any]) -> None:
530 """Enqueue an event for the next batch flush (non-blocking)."""
531 if not self._shutdown.is_set():
532 self._queue.put(event)
534 def close(self) -> None:
535 """Signal shutdown, drain the queue with one final flush, and join the thread."""
536 self._shutdown.set()
537 self._thread.join(timeout=_WEBHOOK_CLOSE_TIMEOUT)
539 def _batch_loop(self) -> None:
540 while not self._shutdown.is_set():
541 self._shutdown.wait(timeout=self._batch_ms / 1000.0)
542 self._flush()
543 # One final drain after shutdown signal
544 self._flush()
546 def _flush(self) -> None:
547 events: list[dict[str, Any]] = []
548 while True:
549 try:
550 events.append(self._queue.get_nowait())
551 except Empty:
552 break
553 if not events:
554 return
555 self._post_with_retry(events)
557 def _post_with_retry(self, events: list[dict[str, Any]]) -> None:
558 payload = json.dumps(events).encode()
559 headers = {"Content-Type": "application/json", **self._headers}
560 backoff = _WEBHOOK_RETRY_BASE_S
561 for attempt in range(self._max_retries + 1):
562 try:
563 resp = self._httpx.post(self._url, content=payload, headers=headers, timeout=10.0)
564 if resp.status_code < 500:
565 return
566 except Exception:
567 pass
568 if attempt < self._max_retries:
569 time.sleep(backoff)
570 backoff = min(backoff * 2, _WEBHOOK_RETRY_MAX_S)
571 logger.warning(
572 "WebhookTransport: giving up after %d retries posting to %r",
573 self._max_retries,
574 self._url,
575 )
578def _make_seed_callback() -> Callable[[_SocketClient], None]:
579 """Return an on_connect callback that seeds a new client with current running loop state."""
580 from little_loops.fsm.persistence import list_running_loops
582 def _seed(client: _SocketClient) -> None:
583 for state in list_running_loops(Path(".loops")):
584 event = {"event": "state_change", **state.to_dict()}
585 payload = (json.dumps(event) + "\n").encode("utf-8")
586 try:
587 client.queue.put_nowait(payload)
588 except Full:
589 pass
591 return _seed
594_TRANSPORT_REGISTRY: dict[str, str] = {
595 "jsonl": "jsonl",
596 "otel": "otel",
597 "socket": "socket",
598 "sqlite": "sqlite",
599 "webhook": "webhook",
600}
603def wire_transports(
604 bus: EventBus,
605 config: EventsConfig,
606 log_dir: Path | None = None,
607) -> None:
608 """Register transports named in `config.transports` on `bus`.
610 Unknown transport names log a warning and are skipped (rather than raising)
611 so that a typo in user config does not prevent the loop from starting. The
612 one exception is the ``socket`` transport on platforms without ``AF_UNIX``,
613 which raises a `RuntimeError` so the user is told why their config does
614 not work — silently dropping the requested transport on Windows would be a
615 more confusing failure mode.
617 Args:
618 bus: EventBus to register transports on
619 config: EventsConfig holding the list of transport names to wire up
620 log_dir: Directory under which built-in transports place their log files.
621 Defaults to ``.ll`` under the current working directory.
622 """
623 base = log_dir if log_dir is not None else Path(".ll")
624 for name in config.transports:
625 if name not in _TRANSPORT_REGISTRY:
626 logger.warning("Unknown transport %r; skipping", name)
627 continue
628 if name == "jsonl":
629 bus.add_transport(JsonlTransport(base / "events.jsonl"))
630 elif name == "otel":
631 bus.add_transport(
632 OTelTransport(
633 endpoint=config.otel.endpoint,
634 service_name=config.otel.service_name,
635 )
636 )
637 elif name == "socket":
638 if not hasattr(socket, "AF_UNIX"):
639 raise RuntimeError(
640 "UnixSocketTransport requires AF_UNIX, which is not available on this "
641 'platform (e.g. Windows). Remove "socket" from events.transports or '
642 'use a different transport such as "jsonl".'
643 )
644 resolved = _resolve_socket_path(config.socket.path, base)
645 bus.add_transport(
646 UnixSocketTransport(
647 resolved,
648 config.socket.max_clients,
649 on_connect=_make_seed_callback(),
650 )
651 )
652 elif name == "sqlite":
653 from little_loops.session_store import SQLiteTransport
655 bus.add_transport(SQLiteTransport(base / "history.db"))
656 elif name == "webhook":
657 if config.webhook.url is None:
658 logger.warning("WebhookTransport: events.webhook.url is None; skipping")
659 continue
660 bus.add_transport(
661 WebhookTransport(
662 url=config.webhook.url,
663 batch_ms=config.webhook.batch_ms,
664 headers=config.webhook.headers,
665 max_retries=3,
666 )
667 )
670def _resolve_socket_path(configured: str, base: Path) -> Path:
671 """Resolve a configured socket path against the per-call log_dir.
673 The default config value is ``.ll/events.sock``. When `wire_transports` is
674 given a custom ``log_dir`` (e.g. a tmp dir in tests), the socket should land
675 inside that directory rather than literally at ``.ll/events.sock`` on disk.
676 Mirroring the JsonlTransport behaviour: strip the ``.ll/`` prefix and treat
677 the remainder as relative to ``base``. Absolute paths are honored as-is.
678 """
679 p = Path(configured)
680 if p.is_absolute():
681 return p
682 if p.parts and p.parts[0] == ".ll":
683 return base.joinpath(*p.parts[1:])
684 return base / p