Problem: You have multiple processes (a web server and a container worker, or a host process and a spawned subprocess) that need to exchange messages and stream events. They don't share memory, and you don't want the complexity of a message broker.
Approach: Use the filesystem as the message bus. JSON files as request/response pairs, JSONL append-only logs as event streams, and atomic state.json snapshots for current status. Bridge async code with asyncio.Future objects that resolve when response files appear.
Pattern proven in production across multiple Python CLI tools and web services.
Why two files: events.jsonl is the complete history (for replay, debugging, SSE streaming). state.json is the current status (for quick reads without scanning the entire log).
2. JSON files as cross-process messages (request + response pair)
Input requests use a file-per-request convention:
async def request_input(self, request_id, schema):
"""Write a human-input request file and return a Future for the response."""
# Validate request_id to prevent path traversal
safe_name = Path(request_id).name
if not safe_name or safe_name != request_id:
raise ValueError(f"Invalid request_id: {request_id!r}")
input_requests_dir = self._work_dir / "input-requests"
input_requests_dir.mkdir(parents=True, exist_ok=True)
request_data = {
"request_id": request_id,
"schema": schema,
"requested_at": datetime.now(UTC).isoformat(),
}
request_file = input_requests_dir / f"{safe_name}.json"
request_file.write_text(json.dumps(request_data, indent=2))
# Create an asyncio.Future that will be resolved when the response arrives
loop = asyncio.get_running_loop()
future = loop.create_future()
self._pending_futures[request_id] = future
return future
3. asyncio.Future + file watcher for async request/response
The request creates a Future. When the response arrives (via an API call from the UI), the Future is resolved thread-safely:
def resolve_input_request(self, request_id, payload):
"""Resolve the pending Future for the given request_id."""
future = self._pending_futures.pop(request_id, None)
if future is None:
return
# Thread-safe: schedule resolution on the Future's event loop
loop = future.get_loop()
loop.call_soon_threadsafe(future.set_result, payload)
The key detail: call_soon_threadsafe is used because the resolve call may come from a different thread (e.g., a sync API handler in a threadpool worker). This schedules the set_result on the correct event loop.
4. In-memory pub/sub EventBus for SSE delivery
The EventBus routes events to per-instance subscriber queues:
class EventBus:
def __init__(self):
self._subscribers: dict[str, list[asyncio.Queue]] = {}
def subscribe(self, instance_id: str) -> asyncio.Queue:
queue = asyncio.Queue()
self._subscribers.setdefault(instance_id, []).append(queue)
return queue
def publish(self, instance_id: str, event: dict | None) -> None:
for queue in list(self._subscribers.get(instance_id, [])):
try:
queue.put_nowait(event)
except asyncio.QueueFull:
pass # drop event rather than block
The monitor loop mirrors events from a container's filesystem to the host, then publishes to the EventBus:
async def _mirror_events(source_path: Path, dest_path: Path, offset: int) -> int:
"""Mirror new lines from source to dest. Returns new offset."""
with open(source_path) as f:
f.seek(offset)
new_lines = f.readlines()
if new_lines:
with open(dest_path, "a") as f:
f.writelines(new_lines)
return offset + sum(len(l) for l in new_lines)
The SSE endpoint then reads from the host-side file:
@router.get("/instances/{instance_id}/events")
async def stream_events(request, instance_id):
async def _generate():
# Wait for file to appear (up to 60s)
# Incremental read with file position tracking
# Detect terminal status → close stream
...
return StreamingResponse(_generate(), media_type="text/event-stream")
_pending_futures is not protected by a lock — intentionally.dict.pop() is effectively atomic under CPython's GIL. This is a deliberate trade-off for simplicity. Warning: Python 3.13+ free-threaded mode removes this guarantee. If you adopt no-GIL Python, add a lock.
Path traversal on request IDs. The request_input method validates that the request_id is a safe filename. Without this, a malicious request_id like ../../etc/evil could write files outside the intended directory. Always validate IDs used in file path construction.
fh.flush() is necessary for tail -f to work. Python's file I/O buffers writes. Without explicit flush(), events appear in batches rather than in real-time. The EventEmitter flushes after every line.
Compact JSON for JSONL, pretty JSON for request files. Events use separators=(",", ":") for minimal wire size in high-volume streams. Request files use indent=2 for human readability during debugging. Match the format to the audience.
The None sentinel for stream completion. The EventBus publishes None when a stream is done. Subscribers check for None to close their SSE connections cleanly. Without a sentinel, subscribers would hang indefinitely waiting for more events from a completed instance.
