Coverage for little_loops/transport.py: 18%

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]``).

18Public exports:

19 Transport: runtime-checkable Protocol that any sink must satisfy

20 JsonlTransport: writes events to a JSONL file

21 UnixSocketTransport: streams events over an AF_UNIX socket

22 OTelTransport: exports loop traces via OTLP

23 WebhookTransport: POSTs batched events to an HTTP endpoint

24 wire_transports: register transports listed in `EventsConfig` on an `EventBus`

25"""

27from __future__ import annotations

29import json

30import logging

31import socket

32import threading

33import time

34from collections.abc import Callable

35from pathlib import Path

36from queue import Empty, Full, Queue

37from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable

39if TYPE_CHECKING:

40 from little_loops.config.features import EventsConfig

41 from little_loops.events import EventBus

43logger = logging.getLogger(__name__)

45_CLIENT_QUEUE_MAXSIZE = 1024

46_DROP_LOG_INTERVAL_SEC = 5.0

47_ACCEPT_THREAD_JOIN_TIMEOUT = 2.0

48_CLIENT_THREAD_JOIN_TIMEOUT = 1.0

49_CLOSE_TOTAL_TIMEOUT = 10.0

50_ACCEPT_POLL_TIMEOUT = 1.0

51_CLIENT_QUEUE_POLL_TIMEOUT = 0.5

53_WEBHOOK_BATCH_MS_DEFAULT = 1000

54_WEBHOOK_CLOSE_TIMEOUT = 10.0

55_WEBHOOK_RETRY_BASE_S = 0.5

56_WEBHOOK_RETRY_MAX_S = 8.0

59@runtime_checkable

60class Transport(Protocol):

61 """Protocol for an event sink registered on an `EventBus`.

63 A transport receives every event emitted on the bus (no filtering at the

64 transport layer; subscribe an observer with a filter for that). Implementations

65 must tolerate being called with arbitrary `dict[str, Any]` shapes — the bus

66 does not validate event contents.

67 """

69 def send(self, event: dict[str, Any]) -> None:

70 """Deliver a single event."""

71 ...

73 def close(self) -> None:

74 """Release any resources held by the transport. May be a no-op."""

75 ...

78class JsonlTransport:

79 """Append events to a JSONL file, one JSON object per line.

81 The parent directory is created at construction time so per-event writes do

82 not have to check it. `close()` is a no-op since each `send()` opens and

83 closes the file (matching the existing JSONL write pattern in this codebase).

84 """

86 def __init__(self, path: Path) -> None:

87 self._path = path

88 self._path.parent.mkdir(parents=True, exist_ok=True)

90 def send(self, event: dict[str, Any]) -> None:

91 with open(self._path, "a", encoding="utf-8") as f:

92 f.write(json.dumps(event) + "\n")

94 def close(self) -> None:

95 return None

98class _SocketClient:

99 """Per-client state: connection, outbound queue, write thread, drop counters."""

100

101 def __init__(self, conn: socket.socket, addr: str) -> None:

102 self.conn = conn

103 self.addr = addr

104 self.queue: Queue[bytes] = Queue(maxsize=_CLIENT_QUEUE_MAXSIZE)

105 self.thread: threading.Thread | None = None

106 self.dropped_total = 0

107 self.dropped_since_log = 0

108 self.last_drop_log_ts = 0.0

109 self.first_drop_logged = False

110

111

112class UnixSocketTransport:

113 """Stream events as newline-delimited JSON over an `AF_UNIX` socket.

114

115 On construction, binds the socket at ``path`` (after unlinking any stale

116 file), starts an accept thread, and accepts up to ``max_clients`` concurrent

117 consumers. Each accepted client gets its own daemon thread and bounded

118 outbound queue; ``send()`` enqueues the serialized event into every client

119 queue without blocking. A full queue causes the newest event to be dropped

120 (preserving causal order) and a rate-limited warning is logged.

121

122 A misbehaving / disconnected client is removed from the pool without

123 affecting other clients or the FSM thread.

124

125 Not available on platforms without ``AF_UNIX`` (e.g. Windows). The platform

126 check lives in :func:`wire_transports` so the user-facing error has clearer

127 placement; constructing this class on a platform without ``AF_UNIX`` will

128 raise an `OSError` from `socket.socket` directly.

129 """

130

131 def __init__(

132 self,

133 path: Path,

134 max_clients: int = 8,

135 on_connect: Callable[[_SocketClient], None] | None = None,

136 ) -> None:

137 if not hasattr(socket, "AF_UNIX"):

138 raise RuntimeError(

139 "UnixSocketTransport requires AF_UNIX, which is not available on this platform"

140 )

141

142 self._path = path

143 self._max_clients = max_clients

144 self._on_connect = on_connect

145 self._shutdown = threading.Event()

146 self._clients: list[_SocketClient] = []

147 self._clients_lock = threading.Lock()

148

149 self._path.parent.mkdir(parents=True, exist_ok=True)

150 self._path.unlink(missing_ok=True)

151

152 self._server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

153 try:

154 self._server.bind(str(self._path))

155 self._path.chmod(0o600)

156 self._server.listen(max_clients)

157 self._server.settimeout(_ACCEPT_POLL_TIMEOUT)

158 except Exception:

159 self._server.close()

160 self._path.unlink(missing_ok=True)

161 raise

162

163 self._accept_thread = threading.Thread(

164 target=self._accept_loop,

165 name="unix-socket-transport-accept",

166 daemon=True,

167 )

168 self._accept_thread.start()

169

170 def _accept_loop(self) -> None:

171 while not self._shutdown.is_set():

172 try:

173 conn, _ = self._server.accept()

174 except TimeoutError:

175 continue

176 except OSError:

177 # Server socket closed during shutdown — exit cleanly

178 return

179

180 with self._clients_lock:

181 if len(self._clients) >= self._max_clients:

182 logger.warning(

183 "UnixSocketTransport: rejecting client; max_clients=%d reached",

184 self._max_clients,

185 )

186 try:

187 conn.close()

188 except OSError:

189 pass

190 continue

191 client = _SocketClient(conn, addr=str(self._path))

192 client.thread = threading.Thread(

193 target=self._client_loop,

194 args=(client,),

195 name="unix-socket-transport-client",

196 daemon=True,

197 )

198 self._clients.append(client)

199 if self._on_connect is not None:

200 self._on_connect(client)

201 client.thread.start()

202

203 def _client_loop(self, client: _SocketClient) -> None:

204 try:

205 while not self._shutdown.is_set():

206 try:

207 payload = client.queue.get(timeout=_CLIENT_QUEUE_POLL_TIMEOUT)

208 except Empty:

209 continue

210 try:

211 client.conn.sendall(payload)

212 except OSError:

213 return

214 finally:

215 try:

216 client.conn.close()

217 except OSError:

218 pass

219 with self._clients_lock:

220 if client in self._clients:

221 self._clients.remove(client)

222

223 def send(self, event: dict[str, Any]) -> None:

224 payload = (json.dumps(event) + "\n").encode("utf-8")

225 with self._clients_lock:

226 snapshot = list(self._clients)

227 for client in snapshot:

228 try:

229 client.queue.put_nowait(payload)

230 except Full:

231 self._record_drop(client)

232

233 def _record_drop(self, client: _SocketClient) -> None:

234 client.dropped_total += 1

235 client.dropped_since_log += 1

236 now = time.monotonic()

237 if not client.first_drop_logged:

238 logger.warning(

239 "UnixSocketTransport: dropping events for slow client (queue full at %d)",

240 _CLIENT_QUEUE_MAXSIZE,

241 )

242 client.first_drop_logged = True

243 client.last_drop_log_ts = now

244 client.dropped_since_log = 0

245 return

246 if now - client.last_drop_log_ts >= _DROP_LOG_INTERVAL_SEC:

247 logger.warning(

248 "UnixSocketTransport: dropped %d events for slow client",

249 client.dropped_since_log,

250 )

251 client.last_drop_log_ts = now

252 client.dropped_since_log = 0

253

254 def close(self) -> None:

255 deadline = time.monotonic() + _CLOSE_TOTAL_TIMEOUT

256 self._shutdown.set()

257

258 accept_budget = min(_ACCEPT_THREAD_JOIN_TIMEOUT, max(0.0, deadline - time.monotonic()))

259 if self._accept_thread.is_alive():

260 self._accept_thread.join(timeout=accept_budget)

261 if self._accept_thread.is_alive():

262 logger.warning(

263 "UnixSocketTransport: accept thread did not exit within %.1fs",

264 accept_budget,

265 )

266

267 try:

268 self._server.close()

269 except OSError:

270 pass

271

272 with self._clients_lock:

273 snapshot = list(self._clients)

274 for client in snapshot:

275 try:

276 client.conn.shutdown(socket.SHUT_RDWR)

277 except OSError:

278 pass

279 t = client.thread

280 if t is not None and t.is_alive():

281 budget = min(_CLIENT_THREAD_JOIN_TIMEOUT, max(0.0, deadline - time.monotonic()))

282 t.join(timeout=budget)

283 if t.is_alive():

284 logger.warning(

285 "UnixSocketTransport: client thread did not exit within %.1fs",

286 budget,

287 )

288

289 self._path.unlink(missing_ok=True)

290

291

292_OTEL_EVENT_TYPES = frozenset(

293 {

294 "evaluate",

295 "route",

296 "retry_exhausted",

297 "cycle_detected",

298 "handoff_detected",

299 "handoff_spawned",

300 "action_output",

301 }

302)

303_OTEL_ERROR_OUTCOMES = frozenset({"error", "failed", "exhausted"})

304

305

306class OTelTransport:

307 """Map ll loop executions to OpenTelemetry traces and spans, exporting via OTLP.

308

309 Span hierarchy: loop = trace root, state = child span, action = grandchild.

310 Span events are added for evaluate, route, retry_exhausted, handoff_detected,

311 handoff_spawned, and action_output on the innermost open span.

312

313 Requires ``opentelemetry-sdk`` and ``opentelemetry-exporter-otlp-grpc``.

314 Install with: ``pip install 'little-loops[otel]'``

315

316 Sub-loop events (``depth > 0``) are no-ops with a single warning per session.

317 """

318

319 def __init__(

320 self,

321 endpoint: str = "http://localhost:4317",

322 service_name: str = "little-loops",

323 *,

324 _tracer_provider: Any | None = None,

325 ) -> None:

326 try:

327 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

328 from opentelemetry.sdk.resources import Resource

329 from opentelemetry.sdk.trace import TracerProvider

330 from opentelemetry.sdk.trace.export import BatchSpanProcessor

331 except ImportError as exc:

332 raise RuntimeError(

333 "OTelTransport requires the 'opentelemetry-sdk' and "

334 "'opentelemetry-exporter-otlp-grpc' packages. "

335 "Install with: pip install 'little-loops[otel]'"

336 ) from exc

337

338 if _tracer_provider is not None:

339 self._provider = _tracer_provider

340 else:

341 resource = Resource.create({"service.name": service_name})

342 provider = TracerProvider(resource=resource)

343 exporter = OTLPSpanExporter(endpoint=endpoint)

344 provider.add_span_processor(BatchSpanProcessor(exporter))

345 self._provider = provider

346

347 self._tracer = self._provider.get_tracer("little-loops")

348 self._loop_span: Any | None = None

349 self._state_span: Any | None = None

350 self._action_span: Any | None = None

351 self._subloop_warned = False

352

353 def send(self, event: dict[str, Any]) -> None:

354 depth = event.get("depth", 0)

355 if isinstance(depth, int) and depth > 0:

356 if not self._subloop_warned:

357 logger.warning(

358 "OTelTransport: sub-loop events (depth > 0) are not supported; "

359 "nested-trace support is deferred. Event type: %r",

360 event.get("event"),

361 )

362 self._subloop_warned = True

363 return

364

365 event_type = event.get("event", "")

366 if event_type == "loop_start":

367 self._handle_loop_start(event)

368 elif event_type == "loop_resume":

369 self._handle_loop_resume(event)

370 elif event_type == "state_enter":

371 self._handle_state_enter(event)

372 elif event_type == "action_start":

373 self._handle_action_start(event)

374 elif event_type == "action_complete":

375 self._handle_action_complete()

376 elif event_type == "loop_complete":

377 self._handle_loop_complete(event)

378 elif event_type in _OTEL_EVENT_TYPES:

379 self._add_span_event(event_type, event)

380

381 def close(self) -> None:

382 self._provider.force_flush()

383 self._provider.shutdown()

384

385 # ------------------------------------------------------------------

386 # Internal span machine

387 # ------------------------------------------------------------------

388

389 def _handle_loop_start(self, event: dict[str, Any]) -> None:

390 loop_name = str(event.get("loop_name", "ll-loop"))

391 self._loop_span = self._tracer.start_span(loop_name)

392

393 def _handle_loop_resume(self, event: dict[str, Any]) -> None:

394 self._close_state_and_action()

395 if self._loop_span is not None:

396 self._loop_span.end()

397 loop_name = str(event.get("loop_name", "ll-loop"))

398 self._loop_span = self._tracer.start_span(loop_name)

399

400 def _handle_state_enter(self, event: dict[str, Any]) -> None:

401 self._close_state_and_action()

402 if self._loop_span is None:

403 logger.warning(

404 "OTelTransport: state_enter received without a prior loop_start; skipping span"

405 )

406 return

407 from opentelemetry import trace

408

409 state_name = str(event.get("state", "unknown-state"))

410 ctx = trace.set_span_in_context(self._loop_span)

411 self._state_span = self._tracer.start_span(state_name, context=ctx)

412

413 def _handle_action_start(self, event: dict[str, Any]) -> None:

414 if self._state_span is None:

415 logger.warning(

416 "OTelTransport: action_start received without a prior state_enter; skipping span"

417 )

418 return

419 from opentelemetry import trace

420

421 action_name = str(event.get("action", "unknown-action"))

422 ctx = trace.set_span_in_context(self._state_span)

423 self._action_span = self._tracer.start_span(action_name, context=ctx)

424

425 def _handle_action_complete(self) -> None:

426 if self._action_span is not None:

427 self._action_span.end()

428 self._action_span = None

429

430 def _handle_loop_complete(self, event: dict[str, Any]) -> None:

431 from opentelemetry.trace import StatusCode

432

433 self._close_state_and_action()

434 if self._loop_span is None:

435 logger.warning(

436 "OTelTransport: loop_complete received without a prior loop_start; skipping"

437 )

438 return

439 outcome = str(event.get("outcome", ""))

440 if outcome in _OTEL_ERROR_OUTCOMES:

441 self._loop_span.set_status(StatusCode.ERROR, outcome)

442 else:

443 self._loop_span.set_status(StatusCode.OK)

444 self._loop_span.end()

445 self._loop_span = None

446

447 def _add_span_event(self, event_type: str, event: dict[str, Any]) -> None:

448 span = self._action_span or self._state_span or self._loop_span

449 if span is None:

450 return

451 attrs = {k: str(v) for k, v in event.items() if k != "event"}

452 span.add_event(event_type, attributes=attrs)

453

454 def _close_state_and_action(self) -> None:

455 if self._action_span is not None:

456 self._action_span.end()

457 self._action_span = None

458 if self._state_span is not None:

459 self._state_span.end()

460 self._state_span = None

461

462

463class WebhookTransport:

464 """POSTs batched FSM events to an HTTP endpoint.

465

466 Events are enqueued non-blocking in ``send()`` and flushed by a daemon

467 thread on a configurable interval. Failed POSTs are retried with

468 exponential backoff; after ``max_retries`` the batch is dropped with a

469 warning rather than raising to the caller.

470

471 Requires ``httpx``: ``pip install little-loops[webhooks]``.

472 """

473

474 def __init__(

475 self,

476 url: str,

477 batch_ms: int = _WEBHOOK_BATCH_MS_DEFAULT,

478 headers: dict[str, str] | None = None,

479 max_retries: int = 3,

480 ) -> None:

481 try:

482 import httpx as _httpx

483 except ImportError as exc:

484 raise RuntimeError(

485 "WebhookTransport requires httpx: pip install little-loops[webhooks]"

486 ) from exc

487 self._httpx = _httpx

488 self._url = url

489 self._batch_ms = batch_ms

490 self._headers = dict(headers) if headers else {}

491 self._max_retries = max_retries

492 self._queue: Queue[dict[str, Any]] = Queue()

493 self._shutdown = threading.Event()

494 self._thread = threading.Thread(target=self._batch_loop, daemon=True, name="webhook-batch")

495 self._thread.start()

496

497 def send(self, event: dict[str, Any]) -> None:

498 """Enqueue an event for the next batch flush (non-blocking)."""

499 if not self._shutdown.is_set():

500 self._queue.put(event)

501

502 def close(self) -> None:

503 """Signal shutdown, drain the queue with one final flush, and join the thread."""

504 self._shutdown.set()

505 self._thread.join(timeout=_WEBHOOK_CLOSE_TIMEOUT)

506

507 def _batch_loop(self) -> None:

508 while not self._shutdown.is_set():

509 self._shutdown.wait(timeout=self._batch_ms / 1000.0)

510 self._flush()

511 # One final drain after shutdown signal

512 self._flush()

513

514 def _flush(self) -> None:

515 events: list[dict[str, Any]] = []

516 while True:

517 try:

518 events.append(self._queue.get_nowait())

519 except Empty:

520 break

521 if not events:

522 return

523 self._post_with_retry(events)

524

525 def _post_with_retry(self, events: list[dict[str, Any]]) -> None:

526 payload = json.dumps(events).encode()

527 headers = {"Content-Type": "application/json", **self._headers}

528 backoff = _WEBHOOK_RETRY_BASE_S

529 for attempt in range(self._max_retries + 1):

530 try:

531 resp = self._httpx.post(self._url, content=payload, headers=headers, timeout=10.0)

532 if resp.status_code < 500:

533 return

534 except Exception:

535 pass

536 if attempt < self._max_retries:

537 time.sleep(backoff)

538 backoff = min(backoff * 2, _WEBHOOK_RETRY_MAX_S)

539 logger.warning(

540 "WebhookTransport: giving up after %d retries posting to %r",

541 self._max_retries,

542 self._url,

543 )

544

545

546def _make_seed_callback() -> Callable[[_SocketClient], None]:

547 """Return an on_connect callback that seeds a new client with current running loop state."""

548 from little_loops.fsm.persistence import list_running_loops

549

550 def _seed(client: _SocketClient) -> None:

551 for state in list_running_loops(Path(".loops")):

552 event = {"event": "state_change", **state.to_dict()}

553 payload = (json.dumps(event) + "\n").encode("utf-8")

554 try:

555 client.queue.put_nowait(payload)

556 except Full:

557 pass

558

559 return _seed

560

561

562_TRANSPORT_REGISTRY: dict[str, str] = {

563 "jsonl": "jsonl",

564 "otel": "otel",

565 "socket": "socket",

566 "webhook": "webhook",

567}

568

569

570def wire_transports(

571 bus: EventBus,

572 config: EventsConfig,

573 log_dir: Path | None = None,

574) -> None:

575 """Register transports named in `config.transports` on `bus`.

576

577 Unknown transport names log a warning and are skipped (rather than raising)

578 so that a typo in user config does not prevent the loop from starting. The

579 one exception is the ``socket`` transport on platforms without ``AF_UNIX``,

580 which raises a `RuntimeError` so the user is told why their config does

581 not work — silently dropping the requested transport on Windows would be a

582 more confusing failure mode.

583

584 Args:

585 bus: EventBus to register transports on

586 config: EventsConfig holding the list of transport names to wire up

587 log_dir: Directory under which built-in transports place their log files.

588 Defaults to ``.ll`` under the current working directory.

589 """

590 base = log_dir if log_dir is not None else Path(".ll")

591 for name in config.transports:

592 if name not in _TRANSPORT_REGISTRY:

593 logger.warning("Unknown transport %r; skipping", name)

594 continue

595 if name == "jsonl":

596 bus.add_transport(JsonlTransport(base / "events.jsonl"))

597 elif name == "otel":

598 bus.add_transport(

599 OTelTransport(

600 endpoint=config.otel.endpoint,

601 service_name=config.otel.service_name,

602 )

603 )

604 elif name == "socket":

605 if not hasattr(socket, "AF_UNIX"):

606 raise RuntimeError(

607 "UnixSocketTransport requires AF_UNIX, which is not available on this "

608 'platform (e.g. Windows). Remove "socket" from events.transports or '

609 'use a different transport such as "jsonl".'

610 )

611 resolved = _resolve_socket_path(config.socket.path, base)

612 bus.add_transport(

613 UnixSocketTransport(

614 resolved,

615 config.socket.max_clients,

616 on_connect=_make_seed_callback(),

617 )

618 )

619 elif name == "webhook":

620 if config.webhook.url is None:

621 logger.warning("WebhookTransport: events.webhook.url is None; skipping")

622 continue

623 bus.add_transport(

624 WebhookTransport(

625 url=config.webhook.url,

626 batch_ms=config.webhook.batch_ms,

627 headers=config.webhook.headers,

628 max_retries=3,

629 )

630 )

631

632

633def _resolve_socket_path(configured: str, base: Path) -> Path:

634 """Resolve a configured socket path against the per-call log_dir.

635

636 The default config value is ``.ll/events.sock``. When `wire_transports` is

637 given a custom ``log_dir`` (e.g. a tmp dir in tests), the socket should land

638 inside that directory rather than literally at ``.ll/events.sock`` on disk.

639 Mirroring the JsonlTransport behaviour: strip the ``.ll/`` prefix and treat

640 the remainder as relative to ``base``. Absolute paths are honored as-is.

641 """

642 p = Path(configured)

643 if p.is_absolute():

644 return p

645 if p.parts and p.parts[0] == ".ll":

646 return base.joinpath(*p.parts[1:])

647 return base / p

Coverage for little_loops / transport.py: 18%

353 statements