Repository navigation
Expand file tree
/
Copy pathevent_bus.py
More file actions
411 lines (337 loc) · 14.5 KB
/
Copy pathevent_bus.py
File metadata and controls
411 lines (337 loc) · 14.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
"""External event ingress for the Textual app.
Exposes a small `EventSource` protocol plus a Unix-domain-socket implementation
that lets local processes push commands, prompts, and signals into a running
session over a newline-delimited JSON wire protocol.
!!! warning "Experimental"
The wire format and configuration env vars may change without semver
guarantees while this surface stabilizes.
"""
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import os
import stat
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Literal, Protocol, get_args
from deepagents_code.command_registry import BypassTier
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
logger = logging.getLogger(__name__)
ExternalEventKind = Literal["command", "prompt", "signal"]
"""Top-level event kinds carried by the wire protocol."""
ExternalSignal = Literal["interrupt", "force-clear"]
"""Closed vocabulary of `kind="signal"` payloads accepted by the listener."""
EventSink = "Callable[[ExternalEvent], Awaitable[None]]"
"""Type alias for the async callback that receives parsed events."""
_VALID_KINDS: frozenset[str] = frozenset(get_args(ExternalEventKind))
_VALID_SIGNALS: frozenset[str] = frozenset(get_args(ExternalSignal))
_ACK = b'{"ok":true}\n'
"""Wire-level acknowledgement returned for an accepted event."""
_MAX_LINE_BYTES = 64 * 1024
"""Per-line read limit; an oversized line is rejected with a NACK."""
_CLIENT_IDLE_TIMEOUT_SECONDS = 60.0
"""Maximum idle time on a client connection before the server closes it."""
@dataclass(frozen=True, slots=True, kw_only=True)
class ExternalEvent:
"""A transport-independent event delivered from outside the TUI."""
kind: ExternalEventKind
payload: str
source: str
bypass: BypassTier = BypassTier.QUEUED
correlation_id: str | None = None
def __post_init__(self) -> None:
"""Validate invariants for direct construction.
Raises:
ValueError: If `kind` is not a known kind, the payload is empty
or whitespace-only, or the kind is `"signal"` but the payload
is not a recognized signal name.
"""
if self.kind not in _VALID_KINDS:
msg = f"Unknown external event kind: {self.kind!r}"
raise ValueError(msg)
if not self.payload or not self.payload.strip():
msg = "External event payload must be a non-empty string"
raise ValueError(msg)
if self.kind == "signal" and self.payload.strip().lower() not in _VALID_SIGNALS:
msg = (
f"Unknown external signal: {self.payload!r}; "
f"expected one of {sorted(_VALID_SIGNALS)}"
)
raise ValueError(msg)
class EventSource(Protocol):
"""Source of external events for the Textual app.
Implementations must be safe to `stop()` even when `start()` failed
partway through; the app always invokes `stop()` from a `finally` block.
"""
async def start(
self,
sink: Callable[[ExternalEvent], Awaitable[None]],
) -> None:
"""Start forwarding events to `sink`.
Args:
sink: Async callback that receives parsed external events.
"""
async def serve_forever(self) -> None:
"""Park until the source is cancelled or its transport dies.
Implementations should re-raise `CancelledError` and propagate
unexpected transport-layer failures so the lifecycle owner can
notice and surface them.
"""
async def stop(self) -> None:
"""Stop forwarding events and release transport resources."""
class UnixSocketEventSource:
"""Line-delimited JSON event source over a local Unix domain socket.
The listener creates its parent directory with mode `0o700` and binds the
socket inside it under a transient `umask(0o077)` so the socket inherits
`0o600` from the moment of `bind()`. Stale sockets at the configured path
are removed on start, but only after a `stat` confirms the path is a
socket — a regular file or directory at that path is left untouched and
causes start to fail loudly.
"""
def __init__(self, path: Path | None = None) -> None:
"""Create a Unix-socket event source.
Args:
path: Socket path. When omitted, a per-process path under the
runtime or temp directory is used.
"""
self.path = path or default_unix_socket_path()
self._server: asyncio.AbstractServer | None = None
self._sink: Callable[[ExternalEvent], Awaitable[None]] | None = None
async def start(
self,
sink: Callable[[ExternalEvent], Awaitable[None]],
) -> None:
"""Start listening for newline-delimited JSON events.
Args:
sink: Async callback invoked with each decoded event.
Raises:
RuntimeError: If `start()` has already been called on this
instance without a subsequent `stop()`.
FileExistsError: If the socket path is occupied by a non-socket
filesystem entry.
OSError: If the socket cannot be bound (e.g. permission denied,
path too long).
""" # noqa: DOC502 # FileExistsError/OSError propagate from helpers
if self._server is not None:
msg = "UnixSocketEventSource is already started"
raise RuntimeError(msg)
self._sink = sink
self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
with contextlib.suppress(FileNotFoundError):
_unlink_existing_socket(self.path)
previous_umask = os.umask(0o077)
try:
self._server = await asyncio.start_unix_server(
self._handle_client,
path=str(self.path),
limit=_MAX_LINE_BYTES,
)
finally:
os.umask(previous_umask)
# Defense in depth: even if `start_unix_server` somehow ignored the
# umask (different libc, mocked socket layer), force-tighten the mode.
with contextlib.suppress(OSError):
self.path.chmod(0o600)
logger.debug("External event listener bound at %s", self.path)
async def serve_forever(self) -> None:
"""Park until the underlying server is cancelled or fails.
`asyncio.start_unix_server` already begins accepting connections,
so this delegates to the server's own `serve_forever`. A fatal
error inside the accept loop propagates here, letting the
lifecycle owner notice and react.
Raises:
RuntimeError: If invoked before `start()`.
"""
if self._server is None:
msg = "UnixSocketEventSource.serve_forever called before start()"
raise RuntimeError(msg)
await self._server.serve_forever()
async def stop(self) -> None:
"""Close the listener and remove the socket path.
Idempotent: safe to call after a failed or never-started `start()`.
"""
server = self._server
self._server = None
if server is not None:
server.close()
with contextlib.suppress(Exception):
await server.wait_closed()
try:
_unlink_existing_socket(self.path)
except FileNotFoundError:
pass
except FileExistsError as exc:
logger.warning(
"Leaving non-socket entry at %s during shutdown: %s",
self.path,
exc,
)
except OSError as exc:
logger.warning("Failed to unlink event socket %s: %s", self.path, exc)
async def _handle_client(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
r"""Read newline-delimited JSON envelopes from one client.
Each accepted line yields an event to the configured sink and is
acked with `{"ok":true}\n` (plus `correlation_id` when present).
Rejected lines (oversized, malformed, sink unconfigured, sink
raised) are answered with `{"ok":false,"error":...}\n` and the
loop continues so a single bad caller cannot kill the connection.
"""
peer = writer.get_extra_info("peername") or "<unknown>"
logger.debug("External event client connected: %s", peer)
try:
while True:
try:
line = await asyncio.wait_for(
reader.readline(),
timeout=_CLIENT_IDLE_TIMEOUT_SECONDS,
)
except TimeoutError:
logger.debug("Closing idle external event client %s", peer)
break
except (ValueError, asyncio.LimitOverrunError) as exc:
logger.warning("External event line exceeded read limit: %s", exc)
await _write_nack(writer, "line exceeds read limit", None)
break
if not line:
break
await self._handle_one_line(line, writer)
except (BrokenPipeError, ConnectionResetError) as exc:
logger.debug("External event client %s disconnected: %s", peer, exc)
finally:
writer.close()
with contextlib.suppress(Exception):
await writer.wait_closed()
logger.debug("External event client closed: %s", peer)
async def _handle_one_line(
self,
line: bytes,
writer: asyncio.StreamWriter,
) -> None:
"""Decode and dispatch one envelope, replying with ACK or NACK."""
correlation_id: str | None = None
try:
event = decode_external_event(line, source=f"unix:{self.path}")
except (ValueError, TypeError) as exc:
logger.warning("Rejected malformed external event: %s", exc)
with contextlib.suppress(ValueError, TypeError, json.JSONDecodeError):
parsed = json.loads(line)
if isinstance(parsed, dict):
candidate = parsed.get("correlation_id")
if isinstance(candidate, str):
correlation_id = candidate
await _write_nack(writer, str(exc), correlation_id)
return
correlation_id = event.correlation_id
if self._sink is None:
logger.warning("External event arrived before sink was set; dropping")
await _write_nack(writer, "listener not ready", correlation_id)
return
try:
await self._sink(event)
except Exception as exc:
logger.exception("External event sink raised")
await _write_nack(writer, f"sink failed: {exc}", correlation_id)
return
await _write_ack(writer, correlation_id)
async def _write_ack(
writer: asyncio.StreamWriter,
correlation_id: str | None,
) -> None:
"""Write the success acknowledgement frame, echoing `correlation_id`."""
if correlation_id is None:
writer.write(_ACK)
else:
body = json.dumps({"ok": True, "correlation_id": correlation_id})
writer.write(body.encode("utf-8") + b"\n")
with contextlib.suppress(BrokenPipeError, ConnectionResetError):
await writer.drain()
async def _write_nack(
writer: asyncio.StreamWriter,
error: str,
correlation_id: str | None,
) -> None:
"""Write a failure response frame; never raises on a closed socket."""
body: dict[str, object] = {"ok": False, "error": error}
if correlation_id is not None:
body["correlation_id"] = correlation_id
try:
writer.write(json.dumps(body).encode("utf-8") + b"\n")
await writer.drain()
except (BrokenPipeError, ConnectionResetError):
pass
def default_unix_socket_path() -> Path:
"""Return the default per-process Unix socket path.
Prefers `XDG_RUNTIME_DIR` (per-user, tmpfs-backed, auto-cleaned on
logout) and falls back to the system temp dir when the runtime
directory is unset.
"""
root = os.environ.get("XDG_RUNTIME_DIR")
base = Path(root) if root else Path(tempfile.gettempdir())
return base / "deepagents" / f"events-{os.getpid()}.sock"
def _unlink_existing_socket(path: Path) -> None:
"""Remove a stale Unix socket without touching other filesystem entries.
Args:
path: Candidate socket path to remove.
Raises:
FileNotFoundError: If `path` does not exist.
FileExistsError: If `path` exists but is not a Unix socket.
OSError: If the entry exists but cannot be removed.
""" # noqa: DOC502 # FileNotFoundError/OSError propagate from stat/unlink
info = path.stat(follow_symlinks=False)
if not stat.S_ISSOCK(info.st_mode):
msg = f"Refusing to remove non-socket external event path: {path}"
raise FileExistsError(msg)
path.unlink()
def decode_external_event(data: bytes, *, source: str) -> ExternalEvent:
"""Decode one newline-delimited JSON external event.
Args:
data: Raw JSON line.
source: Transport-specific source label attached to the event.
Returns:
Parsed external event.
Raises:
TypeError: If the envelope is not a JSON object.
ValueError: If any envelope field is missing, of the wrong type, or
otherwise invalid.
"""
try:
raw = json.loads(data)
except json.JSONDecodeError as exc:
msg = f"External event must be valid JSON: {exc.msg}"
raise ValueError(msg) from exc
if not isinstance(raw, dict):
msg = "External event must be a JSON object"
raise TypeError(msg)
kind = raw.get("kind")
if kind not in _VALID_KINDS:
msg = f"External event kind must be one of {sorted(_VALID_KINDS)}; got {kind!r}"
raise ValueError(msg)
payload = raw.get("payload")
if not isinstance(payload, str) or not payload.strip():
msg = "External event payload must be a non-empty string"
raise ValueError(msg)
bypass = raw.get("bypass", BypassTier.QUEUED.value)
try:
bypass_tier = BypassTier(bypass)
except ValueError as exc:
msg = f"External event bypass must be a valid bypass tier: {exc}"
raise ValueError(msg) from exc
correlation_id = raw.get("correlation_id")
if correlation_id is not None and not isinstance(correlation_id, str):
msg = "External event correlation_id must be a string when present"
raise ValueError(msg)
return ExternalEvent(
kind=kind,
payload=payload,
source=source,
bypass=bypass_tier,
correlation_id=correlation_id,
)