feat(runtime-python): cross-tool bridge — typed returns + q-orca invocation

docs/cross-tool-invoke-and-returns.md

@@ -115,6 +115,14 @@ Both forms parse to the same `InvokeDef`.
 
 AST: `InvokeDef` gains `returns?: Record<string,string>` and `shots?: number`.
 
+**Binding semantics.** A bound value **overwrites** the parent field named on the
+binding's LHS (expected to be a declared parent context key). A return the child
+**did not produce** is **skipped** — the binding is *soft*, leaving the parent
+field at its current value rather than raising — so a child that, say, omits an
+optional aggregate does not abort the parent run. A genuinely wrong shape (a
+malformed result envelope, or a `protocol_version` the parent cannot read) is a
+*hard* `BridgeError`, distinct from a missing field.
+
 ### 3. Bridge implementation (the q-orca protocol)
 
 Implement orca's side of the three JSON envelopes from the q-orca spec:
@@ -266,6 +274,52 @@ and binds the synthesized `prob_bits_0` into the trainer's `prob`. (The q-orca
 side is the `composed_predictive_coder` fixture, which already runs in-tool via
 `q-orca run`.)
 
+## Multi-Runtime Adoption
+
+Orca ships four runtimes (`runtime-python`, `runtime-ts`, `runtime-go`,
+`runtime-rust`). The bridge is **runtime-agnostic by construction**: the entire
+contract is three JSON envelopes (descriptor / invocation / result) plus a
+`protocol_version` constant, carried over a **process boundary** — no shared AST,
+no FFI, no language-specific types. Every language can serialize JSON and spawn a
+subprocess, so any runtime can implement it; the protocol is not even
+orca-specific (any tool that speaks the envelopes can join).
+
+**What a runtime needs** (the same shape everywhere — mirroring `runtime-python`):
+
+1. **Parser** — recognize `## returns` and the invoke `returns:` / `shots:`
+   modifiers (each runtime has its own parser).
+2. **Envelopes** — serialize/deserialize the three shapes + a `protocol_version`
+   check (`encoding/json`, `serde_json`, `JSON.parse`, …).
+3. **`dispatch_foreign`** — spawn the child runner, pipe the invocation envelope
+   to stdin, read the result envelope from stdout (`os/exec`, `std::process`,
+   `child_process`).
+4. **One dispatch hook** — a foreign-runner registry consulted where the invoke
+   target is not a local sibling (in `runtime-python` this is the
+   `start_child_machine` foreign branch; every runtime has the equivalent).
+
+**Direction asymmetry.** *Outbound* (a runtime as the classical orchestrator
+invoking a quantum child) is cheap and clean in any language — and is the
+motivating case; the foreign child is essentially always a **q-orca (Python)**
+process via `q-orca run --bridge`, since q-orca is the only quantum side.
+*Inbound* (a runtime serving as the invoked child) is symmetric at the protocol
+level, but orca machines are **reactive/event-driven**, so "run to completion as
+a child" needs a per-runtime auto-driver — a real work item independent of
+language. (`runtime-python` therefore scopes its first cut to outbound.)
+
+**Why do it in more than one runtime.** The contract is identical, so a **shared
+conformance suite** — every runtime checked against the *same* fixture envelopes
+and the same `1.0` constant — is what keeps four independent implementations from
+silently drifting. Each runtime then gives its own users (Go/Rust/TS
+orchestrators) the same hybrid classical↔quantum capability.
+
+**When it's worth it.** For parity, or a specific deployment (a Go service
+dispatching quantum jobs; the Rust runtime's C/Fortran FFI callers reaching a
+quantum child transitively). It may *not* be urgent where the real demand is "an
+ML/training loop drives a quantum forward pass" — Python is the natural
+orchestration language there, so `runtime-python` likely covers most actual
+usage, and the others are completeness rather than need-driven until a concrete
+consumer appears.
+
 ## Open Questions
 
 1. **How does the verifier know a local `.q.orca.md` child is

packages/runtime-python/orca_runtime_python/__init__.py

@@ -10,6 +10,8 @@
     GuardDef,
     ActionSignature,
     EffectDef,
+    ReturnDef,
+    InvokeDef,
     MachineDef,
     StateValue,
     Context,
@@ -29,6 +31,17 @@
 
 from .parser import parse_orca_md, parse_orca_auto
 
+from .bridge import (
+    BRIDGE_PROTOCOL_VERSION,
+    BridgeError,
+    descriptor_for,
+    build_invocation,
+    make_result,
+    parse_result,
+    parse_invocation,
+    dispatch_foreign,
+)
+
 from .persistence import PersistenceAdapter, AsyncPersistenceAdapter, FilePersistence
 
 from .logging import LogSink, FileSink, ConsoleSink, MultiSink
@@ -42,6 +55,8 @@
     "GuardDef",
     "ActionSignature",
     "EffectDef",
+    "ReturnDef",
+    "InvokeDef",
     "MachineDef",
     "StateValue",
     "Context",
@@ -60,6 +75,15 @@
     # Parser
     "parse_orca_md",
     "parse_orca_auto",
+    # Bridge (cross-tool composition)
+    "BRIDGE_PROTOCOL_VERSION",
+    "BridgeError",
+    "descriptor_for",
+    "build_invocation",
+    "make_result",
+    "parse_result",
+    "parse_invocation",
+    "dispatch_foreign",
     # Persistence
     "PersistenceAdapter",
     "AsyncPersistenceAdapter",

packages/runtime-python/orca_runtime_python/bridge.py

@@ -0,0 +1,158 @@
+"""Cross-tool composition bridge — orca's Python side.
+
+Mirrors the q-orca ``bridge-protocol`` contract so a classical orca orchestrator
+can invoke a q-orca quantum child (and orca can be invoked by q-orca). Three
+versioned JSON envelopes; transport is process + JSON over each tool's ``run``
+entry point. See orca-lang/docs/cross-tool-invoke-and-returns.md and q-orca's
+openspec ``bridge-protocol`` spec.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from typing import Any
+
+from .types import MachineDef
+
+# Must match q-orca's BRIDGE_PROTOCOL_VERSION exactly.
+BRIDGE_PROTOCOL_VERSION = "1.0"
+
+_DEFAULT_TIMEOUT_S = 30
+
+
+class BridgeError(Exception):
+    """A bridge/transport failure (distinct from a child error in the envelope).
+
+    Raised for: unlaunchable runner, timeout, non-JSON output, or an
+    unsupported ``protocol_version``.
+    """
+
+    code = "BRIDGE_ERROR"
+
+
+def _wire_type_from_value(value: Any) -> str:
+    """Infer a wire type from a context default value.
+
+    orca's MachineDef.context stores ``{name: default}`` without the declared
+    type, so the descriptor infers it from the default (``any`` when unknown).
+    """
+    if isinstance(value, bool):
+        return "bool"
+    if isinstance(value, int):
+        return "int"
+    if isinstance(value, float):
+        return "float"
+    if isinstance(value, str):
+        return "string"
+    if isinstance(value, list):
+        return "list"
+    return "any"
+
+
+def descriptor_for(machine: MachineDef) -> dict:
+    """Emit a machine descriptor. orca machines are classical → measurement_bearing False."""
+    return {
+        "protocol_version": BRIDGE_PROTOCOL_VERSION,
+        "name": machine.name,
+        "params": [
+            {"name": name, "type": _wire_type_from_value(default)}
+            for name, default in machine.context.items()
+        ],
+        "returns": [
+            {"name": r.name, "type": r.type, "statistics": list(r.statistics)}
+            for r in machine.returns
+        ],
+        "measurement_bearing": False,
+    }
+
+
+def build_invocation(child: str, args: dict, shots: int | None, return_bindings: dict) -> dict:
+    return {
+        "protocol_version": BRIDGE_PROTOCOL_VERSION,
+        "child": child,
+        "args": dict(args),
+        "shots": shots,
+        "return_bindings": dict(return_bindings),
+    }
+
+
+def make_result(final_state: str, returns: dict, error: dict | None = None) -> dict:
+    envelope = {
+        "protocol_version": BRIDGE_PROTOCOL_VERSION,
+        "final_state": final_state,
+        "returns": dict(returns),
+    }
+    if error is not None:
+        envelope["error"] = error
+    return envelope
+
+
+def _check_version(envelope: dict, kind: str) -> None:
+    version = envelope.get("protocol_version")
+    if version != BRIDGE_PROTOCOL_VERSION:
+        raise BridgeError(
+            f"unsupported bridge {kind} protocol_version {version!r} "
+            f"(this tool speaks {BRIDGE_PROTOCOL_VERSION!r})"
+        )
+
+
+def _load(data: Any, kind: str) -> dict:
+    if isinstance(data, (str, bytes)):
+        try:
+            data = json.loads(data)
+        except (ValueError, TypeError) as exc:
+            raise BridgeError(f"{kind} envelope is not valid JSON: {exc}") from exc
+    if not isinstance(data, dict):
+        raise BridgeError(f"{kind} envelope must be a JSON object")
+    _check_version(data, kind)
+    return data
+
+
+def parse_result(data: Any) -> dict:
+    env = _load(data, "result")
+    return {
+        "final_state": env.get("final_state", ""),
+        "returns": env.get("returns") or {},
+        "error": env.get("error"),
+    }
+
+
+def parse_invocation(data: Any) -> dict:
+    env = _load(data, "invocation")
+    if "child" not in env:
+        raise BridgeError("invocation envelope is missing 'child'")
+    return {
+        "child": env["child"],
+        "args": env.get("args") or {},
+        "shots": env.get("shots"),
+        "return_bindings": env.get("return_bindings") or {},
+    }
+
+
+def dispatch_foreign(runner_argv: list[str], invocation: dict, timeout: float = _DEFAULT_TIMEOUT_S) -> dict:
+    """Run a foreign child via ``runner_argv``: invocation envelope on stdin,
+    result envelope on stdout. Raises ``BridgeError`` on any transport failure."""
+    payload = json.dumps(invocation)
+    try:
+        proc = subprocess.run(
+            runner_argv, input=payload, capture_output=True, text=True, timeout=timeout
+        )
+    except FileNotFoundError as exc:
+        raise BridgeError(f"foreign runner not found: {runner_argv[0]!r}") from exc
+    except subprocess.TimeoutExpired as exc:
+        raise BridgeError(f"foreign runner timed out after {timeout}s") from exc
+
+    try:
+        return parse_result(proc.stdout)
+    except BridgeError as exc:
+        stderr = proc.stderr.strip()[:300]
+        if proc.returncode != 0:
+            raise BridgeError(
+                f"foreign runner exited {proc.returncode}"
+                + (f": {stderr}" if stderr else "")
+            ) from exc
+        # Exit 0 but unusable output — surface stderr to aid debugging.
+        if stderr:
+            raise BridgeError(f"{exc} (runner stderr: {stderr})") from exc
+        raise

packages/runtime-python/orca_runtime_python/machine.py

@@ -35,6 +35,7 @@
 )
 from .bus import EventBus, Event, EventType, get_event_bus
 from .persistence import PersistenceAdapter, AsyncPersistenceAdapter
+from .bridge import BridgeError, build_invocation, dispatch_foreign
 
 
 # Type alias for transition callback
@@ -114,6 +115,9 @@ def __init__(
         # Child machine management
         self._child_machines: dict[str, OrcaMachine] = {}
         self._sibling_machines: dict[str, MachineDef] | None = None
+        # Cross-tool bridge: child name -> runner argv for a foreign (other-tool)
+        # child dispatched over the bridge instead of run as a local machine.
+        self._foreign_runners: dict[str, list[str]] = {}
         self._active_invoke: str | None = None
 
     def _get_initial_state(self) -> str:
@@ -240,15 +244,26 @@ async def resume(self, snap: dict[str, Any]) -> None:
         for leaf in self._state.leaves():
             self._start_timeout_for_state(leaf)
 
+    def register_foreign_runner(self, machine_name: str, runner_argv: list[str]) -> None:
+        """Register a foreign (other-tool) child, invoked over the bridge.
+
+        When an invoke targets `machine_name` and it is not a local sibling, the
+        runtime dispatches it via `runner_argv` (e.g. ``["q-orca", "run",
+        "forward.q.orca.md", "--bridge"]``) instead of starting a local machine.
+        """
+        self._foreign_runners[machine_name] = list(runner_argv)
+
     def register_machines(self, machines: dict[str, MachineDef]) -> None:
         """Register sibling machines for invocation."""
         self._sibling_machines = machines
 
     async def start_child_machine(self, state_name: str, invoke_def: InvokeDef) -> None:
         """Start a child machine as part of an invoke state."""
-        if self._sibling_machines is None:
-            return
-        if invoke_def.machine not in self._sibling_machines:
+        siblings = self._sibling_machines or {}
+        if invoke_def.machine not in siblings:
+            # Foreign (other-tool) child → dispatch over the bridge.
+            if invoke_def.machine in self._foreign_runners:
+                await self._invoke_foreign(invoke_def)
             return
 
         child_def = self._sibling_machines[invoke_def.machine]
@@ -291,6 +306,54 @@ async def on_transition_handler(old: StateValue, new: StateValue) -> None:
         child.on_transition = on_transition_handler
         await child.start()
 
+    async def _invoke_foreign(self, invoke_def: InvokeDef) -> None:
+        """Dispatch an invoke to a foreign (other-tool) child over the bridge.
+
+        Builds the invocation envelope from the parent context (via `input`),
+        runs the foreign runner off the event loop, binds the child's declared
+        returns into the parent context, and emits `on_done` / `on_error`.
+        """
+        args: dict[str, Any] = {}
+        if invoke_def.input:
+            for child_param, parent_expr in invoke_def.input.items():
+                field_name = parent_expr.replace("ctx.", "")
+                args[child_param] = self.context.get(field_name)
+        envelope = build_invocation(
+            invoke_def.machine, args, invoke_def.shots, invoke_def.returns or {}
+        )
+        runner = self._foreign_runners[invoke_def.machine]
+
+        async def _emit_error(error: dict[str, Any]) -> None:
+            if invoke_def.on_error:
+                await self.send(invoke_def.on_error, {"child": invoke_def.machine, "error": error})
+
+        loop = asyncio.get_event_loop()
+        try:
+            result = await loop.run_in_executor(
+                None, dispatch_foreign, list(runner), envelope
+            )
+        except BridgeError as exc:
+            await _emit_error({"code": exc.code, "message": str(exc)})
+            return
+        if result.get("error"):
+            await _emit_error(result["error"])
+            return
+
+        # Bind returns into the parent context. Precedence: a bound value
+        # overwrites the parent field named on the binding's LHS (the field is
+        # expected to be a declared parent context key). A return the child did
+        # not produce is skipped, leaving the parent field at its current value.
+        if invoke_def.returns:
+            for parent_field, child_return in invoke_def.returns.items():
+                if child_return in result["returns"]:
+                    self.context[parent_field] = result["returns"][child_return]
+        if invoke_def.on_done:
+            await self.send(invoke_def.on_done, {
+                "child": invoke_def.machine,
+                "final_state": result["final_state"],
+                "returns": result["returns"],
+            })
+
     async def stop_child_machine(self, state_name: str) -> None:
         """Stop a child machine associated with a state."""
         if self._active_invoke == state_name: