consume.py

"""Smoke test for the Python source-load plugin path (Phase 1C).

Same shape as `examples/python-plugin-consumer/consume.py` — the
manifest's `plugins` field handles loading. The only difference is the
plugin entry: `"."` (this directory) instead of a path to a `.so`.

The resolver sees `plugin.toml` here, parses `language = "python"`,
provisions a uv venv (no deps in this example), spawns the runner,
and registers `EchoPythonNode` into the executor's registry.

Prereqs:
  pip install ...     # standard remotemedia Python deps

Run:
  cd examples/python-source-plugin
  PYTHONPATH=<repo>/clients/python REMOTEMEDIA_PYTHON_SRC=<repo>/clients/python python consume.py
"""
from __future__ import annotations

import asyncio
import logging
import sys
from pathlib import Path

import remotemedia.runtime as rt

log = logging.getLogger("source_plugin_consumer")


async def main_async(manifest_path: Path):
    log.info("loading manifest: %s", manifest_path)
    manifest_json = manifest_path.read_text()

    log.info("creating streaming session (provisions venv on first run)…")
    session = await rt.create_streaming_session(manifest_json)
    log.info("session_id = %s", session.session_id)

    for utterance in ["hello from python source", "second message", "third and final"]:
        log.info("sending: %r", utterance)
        await session.send_input({"type": "text", "data": utterance})
        out = await session.recv_data()
        log.info("got back: %r", out)

    await session.signal_input_complete()
    await session.close()
    log.info("clean shutdown")


def main():
    here = Path(__file__).parent.resolve()
    manifest_path = here / "manifest.json"

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-5s %(name)s | %(message)s",
    )

    if not manifest_path.exists():
        sys.exit(f"manifest not found: {manifest_path}")

    try:
        asyncio.run(main_async(manifest_path))
    except KeyboardInterrupt:
        print("\nbye.", file=sys.stderr)


if __name__ == "__main__":
    main()