feat: add remote runtime state bridge

README.md

@@ -84,7 +84,7 @@ Skills: `coral-quickstart` (install → setup → `.coral_workspace/`), `setting
 | [Kiro](https://kiro.dev) | `kiro` |
 | [OpenCode](https://github.com/opencode-ai/opencode) | `opencode` |
 
-Each agent must be installed and authenticated separately. Per-runtime config — including the [LiteLLM gateway](https://docs.coralxyz.com/guides/gateway) for custom models — is documented at [Agent Runtimes](https://docs.coralxyz.com/guides/agent-runtimes).
+Each agent must be installed and authenticated separately. Per-runtime config — including the [LiteLLM gateway](https://docs.coralxyz.com/guides/gateway) for custom models and the opt-in remote runtime state bridge — is documented at [Agent Runtimes](https://docs.coralxyz.com/guides/agent-runtimes).
 
 ### How It Works
 

coral/agent/__init__.py

@@ -5,6 +5,7 @@
 from coral.agent.builtin.opencode import OpenCodeRuntime
 from coral.agent.manager import AgentManager
 from coral.agent.registry import get_runtime, register_runtime
+from coral.agent.remote import RemoteAgentHandle, RemoteAgentSpec, RemoteAgentState, RemoteRuntime
 from coral.agent.runtime import AgentRuntime
 
 __all__ = [
@@ -13,6 +14,10 @@
     "ClaudeCodeRuntime",
     "CodexRuntime",
     "OpenCodeRuntime",
+    "RemoteAgentHandle",
+    "RemoteAgentSpec",
+    "RemoteAgentState",
+    "RemoteRuntime",
     "get_runtime",
     "register_runtime",
 ]

coral/agent/manager.py

@@ -37,6 +37,7 @@
     choose_roster_balanced_subset,
 )
 from coral.agent.registry import get_runtime
+from coral.agent.remote import RemoteStateBridge, load_remote_runtime
 from coral.agent.runtime import AgentHandle, AgentRuntime
 from coral.agent.state import (
     AgentRuntimeState,
@@ -157,6 +158,8 @@ def __init__(
         self._pending_restart_after_pause: set[str] = set()
         self._gateway: Any | None = None
         self._gateway_keys: dict[str, str] = {}  # agent_id -> proxy key
+        self._remote_state_bridge: RemoteStateBridge | None = None
+        self._last_remote_state_sync = 0.0
         self._grader_proc: multiprocessing.Process | None = None
         self._grader_stop_event: Any | None = None  # multiprocessing.Event
         # Island migration. Only meaningful with >=2 islands and migration
@@ -211,6 +214,11 @@ def start_all(self) -> list[AgentHandle]:
         logger.info(f"  coral_dir: {self.paths.coral_dir}")
         logger.info(f"  repo_dir:  {self.paths.repo_dir}")
 
+        # Optional remote runtime state bridge. This is additive: local
+        # subprocess agents keep their current worktree lifecycle.
+        self._start_remote_state_bridge_if_configured()
+        self._sync_remote_state(force=True)
+
         # 1b. Start gateway if configured
         self._start_gateway_if_enabled()
 
@@ -398,6 +406,35 @@ def _start_gateway_if_enabled(self) -> None:
         self._gateway = gateway
         logger.info(f"Gateway running at {gateway.url}")
 
+    def _start_remote_state_bridge_if_configured(self) -> None:
+        """Initialize the remote state bridge when agents.remote_runtime is configured."""
+        assert self.paths is not None
+        remote_cfg = self.config.agents.remote_runtime
+        if not remote_cfg.class_path:
+            return
+
+        runtime = load_remote_runtime(remote_cfg.class_path, remote_cfg.config)
+        self._remote_state_bridge = RemoteStateBridge(
+            runtime=runtime,
+            state_dir=self.paths.coral_dir / "public" / "remote_state",
+        )
+        logger.info("Remote runtime state bridge enabled: %s", remote_cfg.class_path)
+
+    def _sync_remote_state(self, force: bool = False) -> None:
+        """Best-effort sync from the configured remote runtime into public state."""
+        if self._remote_state_bridge is None:
+            return
+        interval = self.config.agents.remote_runtime.sync_interval_seconds
+        now = time.time()
+        if not force and now - self._last_remote_state_sync < interval:
+            return
+        try:
+            states = self._remote_state_bridge.sync_once()
+            self._last_remote_state_sync = now
+            logger.debug("Synced %d remote agent state snapshot(s)", len(states))
+        except Exception as e:
+            logger.warning("Remote runtime state sync failed: %s", e)
+
     def _run_warmstart_research(
         self,
         warmstart: WarmStartRunner,
@@ -2093,6 +2130,8 @@ def _signal_handler(sig: int, frame: Any) -> None:
         logger.info(f"Monitoring {len(self.handles)} agent(s) (check every {check_interval}s)...")
 
         while self._running:
+            self._sync_remote_state()
+
             # Check for new attempts
             current_attempts = self._get_seen_attempts()
             new_attempts = current_attempts - seen_attempts

coral/agent/remote.py

@@ -0,0 +1,160 @@
+"""Remote agent runtime contracts and state bridge helpers."""
+
+from __future__ import annotations
+
+import importlib
+import json
+import os
+from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Protocol, runtime_checkable
+from urllib.parse import quote
+
+
+def utc_now_iso() -> str:
+    """Return the current UTC timestamp in ISO-8601 format."""
+    return datetime.now(UTC).isoformat()
+
+
+@dataclass(frozen=True)
+class RemoteAgentSpec:
+    """Description of an agent CORAL wants a remote runtime to deploy."""
+
+    agent_id: str
+    name: str
+    code_path: Path | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["code_path"] = str(self.code_path) if self.code_path else None
+        return data
+
+
+@dataclass(frozen=True)
+class RemoteAgentHandle:
+    """Stable pointer to an agent managed by a remote runtime."""
+
+    agent_id: str
+    runtime_type: str
+    runtime_id: str
+    endpoint: str | None = None
+    status: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class RemoteAgentState:
+    """Normalized state collected from a remote runtime."""
+
+    agent_id: str
+    runtime_type: str
+    runtime_id: str
+    status: str
+    collected_at: str = field(default_factory=utc_now_iso)
+    metrics: dict[str, Any] = field(default_factory=dict)
+    artifacts: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@runtime_checkable
+class RemoteRuntime(Protocol):
+    """Protocol for agents that run outside CORAL-managed local worktrees."""
+
+    runtime_type: str
+
+    def deploy(self, agent_spec: RemoteAgentSpec) -> RemoteAgentHandle:
+        """Deploy or bind a remote agent and return its handle."""
+
+    def invoke(self, handle: RemoteAgentHandle, payload: dict[str, Any]) -> dict[str, Any]:
+        """Invoke a remote agent."""
+
+    def list_agents(self) -> list[RemoteAgentHandle]:
+        """Return known remote agents."""
+
+    def collect_metrics(self) -> list[RemoteAgentState]:
+        """Return normalized state for known remote agents."""
+
+
+def load_remote_runtime(class_path: str, config: dict[str, Any] | None = None) -> RemoteRuntime:
+    """Load a remote runtime adapter from ``module:Class`` notation."""
+    if class_path.count(":") != 1:
+        raise ValueError(
+            f"Remote runtime class must be 'module.path:ClassName', got {class_path!r}"
+        )
+    module_name, class_name = class_path.split(":", 1)
+    if not module_name or not class_name:
+        raise ValueError(
+            f"Remote runtime class must be 'module.path:ClassName', got {class_path!r}"
+        )
+
+    module = importlib.import_module(module_name)
+    runtime_class = getattr(module, class_name)
+    runtime_config = config or {}
+    from_config = getattr(runtime_class, "from_config", None)
+    runtime = (
+        from_config(runtime_config) if callable(from_config) else runtime_class(**runtime_config)
+    )
+    if not isinstance(runtime, RemoteRuntime):
+        raise TypeError(
+            f"{class_path} does not satisfy the RemoteRuntime protocol "
+            "(see coral/agent/remote.py for the required methods)."
+        )
+    return runtime
+
+
+class RemoteStateBridge:
+    """Persist remote runtime state under ``.coral/public/remote_state``."""
+
+    def __init__(self, runtime: RemoteRuntime, state_dir: Path) -> None:
+        self.runtime = runtime
+        self.state_dir = state_dir
+
+    def sync_once(self) -> list[RemoteAgentState]:
+        """Collect metrics once and atomically write state files."""
+        states = self.runtime.collect_metrics()
+        serialized = [state.to_dict() for state in states]
+        self.state_dir.mkdir(parents=True, exist_ok=True)
+
+        for state in serialized:
+            self._write_json(self.state_dir / self._state_filename(state["agent_id"]), state)
+
+        self._write_json(
+            self.state_dir / "index.json",
+            {
+                "runtime_type": self.runtime.runtime_type,
+                "collected_at": utc_now_iso(),
+                "agents": serialized,
+            },
+        )
+        return states
+
+    @staticmethod
+    def _write_json(path: Path, payload: dict[str, Any]) -> None:
+        tmp_path = path.with_suffix(path.suffix + ".tmp")
+        tmp_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+        os.replace(tmp_path, path)
+
+    @staticmethod
+    def _state_filename(agent_id: Any) -> str:
+        encoded = quote(str(agent_id), safe="")
+        return f"{encoded or '_'}.json"
+
+
+def read_remote_state(coral_dir: str | Path) -> dict[str, Any] | None:
+    """Best-effort read of the remote state index for CLI and web status."""
+    path = Path(coral_dir) / "public" / "remote_state" / "index.json"
+    if not path.exists():
+        return None
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    return payload if isinstance(payload, dict) else None

coral/cli/start.py

@@ -838,6 +838,7 @@ def cmd_stop(args: argparse.Namespace) -> None:
 
 def cmd_status(args: argparse.Namespace) -> None:
     """Show agent status and leaderboard."""
+    from coral.agent.remote import read_remote_state
     from coral.hub.attempts import (
         format_leaderboard,
         format_status_summary,
@@ -990,6 +991,27 @@ def cmd_status(args: argparse.Namespace) -> None:
                         f"|  grader-error rate: {rate_str}"
                     )
 
+    remote_state = read_remote_state(coral_dir)
+    if remote_state:
+        remote_agents = remote_state.get("agents", [])
+        if isinstance(remote_agents, list):
+            print(f"\nRemote agents: {len(remote_agents)}")
+            for agent in remote_agents:
+                if not isinstance(agent, dict):
+                    continue
+                metrics = agent.get("metrics") if isinstance(agent.get("metrics"), dict) else {}
+                metric_text = ""
+                if metrics:
+                    metric_text = "  |  " + ", ".join(
+                        f"{key}={value}" for key, value in sorted(metrics.items())
+                    )
+                print(
+                    f"  {agent.get('agent_id', 'unknown')}: "
+                    f"{agent.get('status', 'unknown')}  |  "
+                    f"{agent.get('runtime_type', remote_state.get('runtime_type', 'remote'))}"
+                    f"{metric_text}"
+                )
+
     direction = read_direction(coral_dir)
     print()
     show_all = getattr(args, "all", False)

coral/config.py

@@ -112,6 +112,22 @@ class WarmStartConfig:
     enabled: bool = False
 
 
+@dataclass
+class RemoteRuntimeConfig:
+    """Optional adapter for state from agents running in a remote runtime."""
+
+    class_path: str = ""  # YAML key: agents.remote_runtime.class
+    config: dict[str, Any] = field(default_factory=dict)
+    sync_interval_seconds: int = 30
+
+    def __post_init__(self) -> None:
+        if self.sync_interval_seconds < 1:
+            raise ValueError(
+                "agents.remote_runtime.sync_interval_seconds must be >= 1, "
+                f"got {self.sync_interval_seconds}"
+            )
+
+
 @dataclass
 class AgentAssignmentConfig:
     """Per-assignment override of runtime/model for mix-and-match multi-agent runs.
@@ -145,6 +161,7 @@ class AgentConfig:
     model: str = "sonnet"
     gateway: GatewayConfig = field(default_factory=GatewayConfig)
     warmstart: WarmStartConfig = field(default_factory=WarmStartConfig)
+    remote_runtime: RemoteRuntimeConfig = field(default_factory=RemoteRuntimeConfig)
     runtime_options: dict[str, Any] = field(default_factory=dict)
     # OS-user isolation: when set (e.g. "agent"), the agent subprocess is run as
     # this unprivileged user while the manager/grader stay root. The agent's
@@ -200,6 +217,8 @@ class AgentConfig:
     min_clean_runtime_seconds: int = 60
 
     def __post_init__(self) -> None:
+        if isinstance(self.remote_runtime, dict):
+            self.remote_runtime = RemoteRuntimeConfig(**self.remote_runtime)
         # Reject negative values for the new reliability knobs;
         # 0 is treated as "disabled" for the same fields where it makes sense.
         for field_name in (
@@ -389,6 +408,11 @@ def to_dict(self) -> dict[str, Any]:
         container: dict[str, Any] = OmegaConf.to_container(sc, resolve=True)  # type: ignore[assignment]
         # Remove internal-only fields
         container.pop("task_dir", None)
+        remote_runtime = container.get("agents", {}).get("remote_runtime")
+        if isinstance(remote_runtime, dict):
+            class_path = remote_runtime.pop("class_path", "")
+            if class_path:
+                remote_runtime["class"] = class_path
         # Serialize heartbeat is_global as "global" for YAML compat
         for h in container.get("agents", {}).get("heartbeat", []):
             h["global"] = h.pop("is_global", False)
@@ -571,6 +595,13 @@ def _preprocess(data: dict[str, Any]) -> dict[str, Any]:
     # existing runtime/model/assignment system.
     _expand_bindings(agents_data)
 
+    remote_runtime = agents_data.get("remote_runtime")
+    if isinstance(remote_runtime, dict):
+        remote_runtime = dict(remote_runtime)
+        if "class" in remote_runtime:
+            remote_runtime["class_path"] = remote_runtime.pop("class")
+        agents_data["remote_runtime"] = remote_runtime
+
     heartbeat_raw = agents_data.pop("heartbeat", None)
     old_reflect = agents_data.pop("reflect_every", None)
     old_heartbeat = agents_data.pop("heartbeat_every", None)

coral/web/api.py

@@ -11,6 +11,7 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
+from coral.agent.remote import read_remote_state
 from coral.cli._helpers import is_docker_run_alive
 
 
@@ -586,5 +587,6 @@ async def get_status(request: Request) -> JSONResponse:
             "best_score": best.score if best else None,
             "best_title": best.title if best else None,
             "agents": agents_status,
+            "remote_state": read_remote_state(coral_dir),
         }
     )