File-Based IPC & Events

The Pattern

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.

Key Design Decisions

1. JSONL append-only event log + atomic state.json snapshot

The EventEmitter writes two complementary files:

• events.jsonl — append-only, one JSON object per line, flushed immediately so tail -f works
• state.json — atomic overwrite of current status, always a complete snapshot

def emit(self, event_type: str, *, phase=None, data=None) -> None:
    """Append a structured event to events.jsonl. Flushes immediately."""
    record = {
        "schema_version": SCHEMA_VERSION,
        "timestamp": datetime.now(UTC).isoformat(),
        "instance_id": self._instance_id,
        "event_type": event_type,
        "phase": phase,
        "data": data if data is not None else {},
    }
    line = json.dumps(record, separators=(",", ":")) + "\n"
    with self._lock, (self._work_dir / "events.jsonl").open("a") as fh:
        fh.write(line)
        fh.flush()                              # immediate for tail -f

The state snapshot uses atomic write:

def update_state(self, **kwargs) -> None:
    """Overwrite state.json atomically via os.replace."""
    snapshot = {
        "instance_id": self._instance_id,
        "updated_at": datetime.now(UTC).isoformat(),
        **kwargs,
    }
    target = self._work_dir / "state.json"
    fd, tmp_path = tempfile.mkstemp(dir=self._work_dir, prefix=".state-", suffix=".tmp")
    try:
        with os.fdopen(fd, "w") as fh:
            json.dump(snapshot, fh)
        Path(tmp_path).replace(target)          # atomic on POSIX
    except Exception:
        Path(tmp_path).unlink(missing_ok=True)
        raise

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

…