Add user orchestrator for agent messaging and observation

packages/coding-agent/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## [Unreleased]
 
+- Added daemon-backed user orchestration with agent-to-agent messaging and read-only observation of active sessions.
+- Added an orchestration heartbeat skill for compact multi-session progress, blocker, and action summaries.
 - Added an opt-in auto-refine review hook that can ask whether `/refine` should run after turn intervals or compaction checkpoints.
 
 ## [0.2.4] - 2026-07-01

packages/coding-agent/skills/agent-message/SKILL.md

@@ -0,0 +1,38 @@
+---
+name: agent-message
+description: Message other active Prime Agent sessions through the daemon. Use to discover active agents and send a direct text message without spoofing sender identity.
+---
+
+# Agent Message
+
+Send direct messages to other active Prime Agent sessions through the local
+daemon. The daemon derives your sender identity from the current session; do
+not try to include a `from` field.
+
+Call directly from the kernel:
+
+```python
+agents = await agent_message.list_agents()
+receipt = await agent_message.send("worker", "Please inspect the latest result.", mode="auto")
+```
+
+## API
+
+- `await agent_message.list_agents()` — returns `current` and `agents`, where
+  each agent includes active session id, session id, optional name, runtime
+  kind, cwd, streaming state, and pending message count.
+- `await agent_message.send(target, message, mode="auto")` — sends one direct
+  text message to an active session. `target` is resolved by the daemon like
+  other live-session selectors. `mode` is `"auto"`, `"follow_up"`, or
+  `"steer"`. Returns a receipt with a `deliveryStatus` field: `"delivered"`
+  means the message reached an idle target's context; `"queued"` means it was
+  accepted and will deliver when the target's current work allows (`send`
+  does not block waiting for that). Delivered receipts carry `deliveredAt`,
+  queued receipts carry `queuedAt`.
+
+## Safety
+
+- Broadcast sends are not supported.
+- Sender identity is daemon-derived and cannot be spoofed from Python.
+- The daemon enforces message size, rate, and pending-queue limits before
+  accepting delivery.

packages/coding-agent/skills/agent-message/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "agent-message"
+version = "0.1.0"
+description = "Prime Agent session-to-session messaging skill"
+requires-python = ">=3.10"
+dependencies = []
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/agent_message"]

packages/coding-agent/skills/agent-message/src/agent_message/__init__.py

@@ -0,0 +1,48 @@
+"""Prime Agent session-to-session messaging skill.
+
+All routing and sender identity live in the TypeScript daemon. These functions
+only call the host bridge exposed inside the Prime Agent IPython kernel.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from rlm import host_request
+
+MessageMode = Literal["auto", "follow_up", "steer"]
+
+
+async def list_agents() -> dict[str, Any]:
+    """List active daemon sessions addressable by agent_message.send()."""
+    return await host_request("agent_message.list")
+
+
+async def send(target: str, message: str, mode: MessageMode = "auto") -> dict[str, Any]:
+    """Send one direct text message to another active Prime Agent session.
+
+    Args:
+        target: Active session id, session id/name, or unambiguous suffix.
+        message: Text payload to deliver.
+        mode: "auto" queues as follow-up only if the target is streaming;
+            "follow_up" always uses follow-up when the target is streaming;
+            "steer" interrupts a streaming target.
+
+    Returns a receipt dict whose "deliveryStatus" is "delivered" (reached an
+    idle target's context) or "queued" (accepted; delivers when the target's
+    current work allows -- this call does not block waiting for that).
+    """
+    if not isinstance(target, str):
+        raise TypeError(f"target must be str, got {type(target).__name__}")
+    if not isinstance(message, str):
+        raise TypeError(f"message must be str, got {type(message).__name__}")
+    if mode not in ("auto", "follow_up", "steer"):
+        raise ValueError('mode must be "auto", "follow_up", or "steer"')
+    return await host_request(
+        "agent_message.send",
+        {
+            "target": target,
+            "message": message,
+            "mode": mode,
+        },
+    )

packages/coding-agent/skills/agent-observe/SKILL.md

@@ -0,0 +1,39 @@
+---
+name: agent-observe
+description: Read-only observation of active Prime Agent sessions through the local daemon. Use to list agents, inspect session status, and read bounded recent-message previews without mutating other sessions.
+---
+
+# Agent Observe
+
+Observe active Prime Agent sessions through the local daemon. This skill is
+read-only: it can list sessions, inspect one session, and fetch bounded recent
+message previews. It cannot prompt, steer, clear, kill, rename, or otherwise
+mutate another session.
+
+Call directly from the kernel:
+
+```python
+agents = await agent_observe.list_agents()
+worker = await agent_observe.get_agent("worker")
+recent = await agent_observe.recent_messages("worker", limit=6)
+```
+
+## API
+
+- `await agent_observe.list_agents()` returns `current` and `agents`. Each
+  agent includes active session id, session id, optional name, runtime kind,
+  cwd, status, streaming state, message count, pending count, and a latest
+  message preview.
+- `await agent_observe.get_agent(target)` returns one agent summary. `target`
+  is resolved like other live-session selectors: active id, session id/name, or
+  unambiguous suffix.
+- `await agent_observe.recent_messages(target, limit=8, max_chars=800)`
+  returns up to `limit` recent bounded message previews for the target session.
+  `limit` must be 1-50, and `max_chars` must be 80-2000.
+
+## Safety
+
+- This skill is read-only and exposes no mutation commands.
+- Message access is bounded by count and per-message character limit.
+- Prefer status and recent previews for orchestration. Ask the user before
+  using observed context to steer or message another session.

packages/coding-agent/skills/agent-observe/pyproject.toml

@@ -0,0 +1,15 @@
+[project]
+name = "agent-observe"
+version = "0.1.0"
+description = "Read-only Prime Agent session observation skill"
+requires-python = ">=3.10"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/agent_observe"]
+
+[tool.prime_agent.skill]
+import = "agent_observe"

packages/coding-agent/skills/agent-observe/src/agent_observe/__init__.py

@@ -0,0 +1,52 @@
+"""Read-only Prime Agent session observation skill.
+
+All session lookup and data access live in the TypeScript daemon. These
+functions only call the host bridge exposed inside the Prime Agent IPython
+kernel.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from rlm import host_request
+
+
+async def list_agents() -> dict[str, Any]:
+    """List active daemon sessions visible to this agent."""
+    return await host_request("agent_observe.list")
+
+
+async def get_agent(target: str) -> dict[str, Any]:
+    """Read one active session summary by active id, session id/name, or suffix."""
+    if not isinstance(target, str):
+        raise TypeError(f"target must be str, got {type(target).__name__}")
+    return await host_request("agent_observe.get", {"target": target})
+
+
+async def recent_messages(
+    target: str,
+    limit: int = 8,
+    max_chars: int = 800,
+) -> dict[str, Any]:
+    """Read bounded recent message previews from an active session.
+
+    Args:
+        target: Active session id, session id/name, or unambiguous suffix.
+        limit: Number of recent messages to return. Host validates 1-50.
+        max_chars: Per-message preview size. Host validates 80-2000.
+    """
+    if not isinstance(target, str):
+        raise TypeError(f"target must be str, got {type(target).__name__}")
+    if not isinstance(limit, int):
+        raise TypeError(f"limit must be int, got {type(limit).__name__}")
+    if not isinstance(max_chars, int):
+        raise TypeError(f"max_chars must be int, got {type(max_chars).__name__}")
+    return await host_request(
+        "agent_observe.recent",
+        {
+            "target": target,
+            "limit": limit,
+            "max_chars": max_chars,
+        },
+    )

packages/coding-agent/skills/orchestration-heartbeat/SKILL.md

@@ -0,0 +1,41 @@
+---
+name: orchestration-heartbeat
+description: Initialize or refresh an internal orchestrator heartbeat that observes active Prime Agent sessions, summarizes progress and blockers, and recommends next actions or messages.
+---
+
+# Orchestration Heartbeat
+
+Use this skill from an orchestrator session to create or refresh a recurring
+internal heartbeat for supervising multiple active Prime Agent sessions.
+
+The heartbeat is an RLM heartbeat, not the user's visible `/heartbeat`. It
+cannot read, replace, pause, resume, or clear the user-level heartbeat.
+
+Call directly from IPython:
+
+```python
+await orchestration_heartbeat.initialize()
+await orchestration_heartbeat.initialize(interval="10m", focus="EmulatorBench and AutoEnv")
+```
+
+## API
+
+- `await orchestration_heartbeat.initialize(interval="5m", label="orchestrator",
+  focus=None, require_user_approval=True)` creates or updates the labeled
+  orchestrator heartbeat for the current session.
+- `await orchestration_heartbeat.ensure(...)` is an alias for `initialize`.
+- `orchestration_heartbeat.build_instruction(...)` returns the prompt text
+  without creating or updating a heartbeat.
+
+## Heartbeat Behavior
+
+Each recurring orchestration heartbeat should:
+
+- Inspect active sessions with `agent_observe`.
+- Summarize each relevant session as `active`, `waiting`, `blocked`, `error`,
+  or `completed`.
+- Include current progress and explicit blockers.
+- Recommend the next action and, when useful, draft a target message.
+- Ask for user approval before sending cross-session messages unless the user
+  has already approved that specific messaging policy.
+- Keep the update compact and operational instead of dumping logs.

packages/coding-agent/skills/orchestration-heartbeat/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "orchestration-heartbeat"
+version = "0.1.0"
+description = "Prime Agent orchestration heartbeat skill"
+requires-python = ">=3.10"
+dependencies = ["agent-observe", "rlm-heartbeat"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/orchestration_heartbeat"]