File Ipc Patterns Agent Skill

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

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

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

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)

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

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)

@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")

# events.py — JSONL event log + atomic state + Future-based IPC
import asyncio, contextlib, json, os, tempfile, threading
from datetime import UTC, datetime
from pathlib import Path

class EventEmitter:
    def __init__(self, work_dir: Path, instance_id: str = ""):
        self._work_dir = work_dir
        self._instance_id = instance_id
        self._lock = threading.Lock()
        self._pending: dict[str, asyncio.Future] = {}

    def emit(self, event_type: str, data: dict | None = None) -> None:
        record = {
            "ts": datetime.now(UTC).isoformat(),
            "id": self._instance_id,
            "type": event_type,
            "data": data or {},
        }
        line = json.dumps(record, separators=(",", ":")) + "\n"
        with self._lock, (self._work_dir / "events.jsonl").open("a") as fh:
            fh.write(line)
            fh.flush()

    def update_state(self, **kwargs) -> None:
        snapshot = {"id": self._instance_id, "updated_at": datetime.now(UTC).isoformat(), **kwargs}
        target = self._work_dir / "state.json"
        fd, tmp = tempfile.mkstemp(dir=self._work_dir, suffix=".tmp")
        try:
            with os.fdopen(fd, "w") as fh:
                json.dump(snapshot, fh)
            Path(tmp).replace(target)
        except BaseException:
            with contextlib.suppress(OSError):
                os.close(fd)
            Path(tmp).unlink(missing_ok=True)
            raise

    async def request_input(self, request_id: str, schema: dict) -> asyncio.Future:
        req_dir = self._work_dir / "input-requests"
        req_dir.mkdir(exist_ok=True)
        (req_dir / f"{request_id}.json").write_text(
            json.dumps({"request_id": request_id, "schema": schema, "response": None}))
        self.emit("input_requested", {"request_id": request_id})
        loop = asyncio.get_running_loop()
        future = loop.create_future()
        self._pending[request_id] = future
        return future

    def resolve_input(self, request_id: str, payload: dict) -> None:
        future = self._pending.pop(request_id, None)
        if future:
            future.get_loop().call_soon_threadsafe(future.set_result, payload)