feat(omp-rpc): introduced typed event listeners and todo phase management to RPC client

- Added typed event listeners and granular event handling for all RPC notification types.
- Added set_todos RPC command and todoPhases session state field for todo phase management.
- Added RpcClient initialization parameters (thinking, tools, no_session, rpc_defaults) for startup configuration.
- Added install_headless_ui() method and todo management methods (get_todos, set_todos, clear_todos).
- Added TodoItem and TodoPhase dataclasses with parser functions for structured todo representation.
- Added RPC mode behavior: disables session title generation by default and resets workflow settings to built-in defaults.
This commit is contained in:
can1357
2026-04-08 04:52:46 +02:00
parent 67e12b39fe
commit d7261bcbeb
11 changed files with 740 additions and 17 deletions
+57 -2
View File
@@ -22,6 +22,8 @@ omp --mode rpc [regular CLI options]
Behavior notes:
- `@file` CLI arguments are rejected in RPC mode.
- RPC mode disables automatic session title generation by default to avoid an extra model call.
- RPC mode resets workflow-altering `todo.*`, `task.*`, and `async.*` settings to their built-in defaults instead of inheriting user overrides.
- The process reads stdin as JSONL (`readJsonl(Bun.stdin.stream())`).
- When stdin closes, the process exits with code `0`.
- Responses/events are written as one JSON object per line.
@@ -73,6 +75,7 @@ Important edge behavior from runtime:
### State
- `{ id?, type: "get_state" }`
- `{ id?, type: "set_todos", phases: TodoPhase[] }`
### Model
@@ -145,10 +148,54 @@ Data payloads are command-specific and defined in `rpc-types.ts`.
"sessionName": "...",
"autoCompactionEnabled": true,
"messageCount": 0,
"queuedMessageCount": 0
"queuedMessageCount": 0,
"todoPhases": [
{
"id": "phase-1",
"name": "Todos",
"tasks": [
{
"id": "task-1",
"content": "Map the tool surface",
"status": "in_progress"
}
]
}
]
}
```
### `set_todos` payload
Replaces the in-memory todo state for the current session and returns the normalized phase list:
```json
{
"id": "req_2",
"type": "set_todos",
"phases": [
{
"id": "phase-1",
"name": "Evaluation",
"tasks": [
{
"id": "task-1",
"content": "Map the read tool surface",
"status": "in_progress"
},
{
"id": "task-2",
"content": "Exercise edit operations",
"status": "pending"
}
]
}
]
}
```
This is useful for hosts that want to pre-seed a plan before the first prompt.
## Event Stream Schema
RPC mode forwards `AgentSessionEvent` objects from `AgentSession.subscribe(...)`.
@@ -163,6 +210,7 @@ Common event types:
- `auto_retry_start`, `auto_retry_end`
- `ttsr_triggered`
- `todo_reminder`
- `todo_auto_clear`
Extension runner errors are emitted separately as:
@@ -226,6 +274,13 @@ Extensions in RPC mode use request/response UI frames.
- `select`, `confirm`, `input`, `editor`
- `notify`, `setStatus`, `setWidget`, `setTitle`, `set_editor_text`
Runtime note:
- Automatic session title generation is disabled in RPC mode, and `setTitle` UI
requests are also suppressed by default because most hosts do not have a
meaningful terminal-title surface. Set `PI_RPC_EMIT_TITLE=1` to opt back in to
the UI event only.
Example:
```json
@@ -322,4 +377,4 @@ Current helper characteristics:
- Dispatches only recognized `AgentEvent` types to listeners
- Does **not** expose helper methods for every protocol command (for example, `set_interrupt_mode` and `set_session_name` are in protocol types but not wrapped as dedicated methods)
Use raw protocol frames if you need complete surface coverage.
Use raw protocol frames if you need complete surface coverage.
+5
View File
@@ -1,8 +1,10 @@
# Changelog
## [Unreleased]
### Added
- RPC `set_todos` command and `todoPhases` in `get_state`, allowing hosts to pre-seed and inspect session todo state over the protocol
- Deferred diagnostics support in LSP writethrough: `onDeferredDiagnostics` callback and `deferredSignal` in `WritethroughOptions` allow callers to receive diagnostics that arrive after the main 5-second timeout
- Language detection for `.pm` (Perl modules), `.astro` (Astro framework), and special filenames `containerfile` and `justfile`
- Workspace-scoped diagnostics and reload actions via `*` file parameter; `diagnostics` action now supports `*` for workspace-wide diagnostics across all configured servers
@@ -23,6 +25,9 @@
### Changed
- RPC mode now automatically disables session title generation by default; hosts can opt in with `PI_RPC_EMIT_TITLE=1` environment variable to receive title updates
- RPC mode now resets workflow-altering `todo.*`, `task.*`, and `async.*` settings to built-in defaults instead of inheriting user overrides
- RPC mode now disables automatic session title generation by default and suppresses `setTitle` extension UI requests unless hosts opt in with `PI_RPC_EMIT_TITLE=1`
- Reorganized edit tool implementation from `patch/` to `edit/` directory structure with dedicated mode subdirectories (`edit/modes/chunk.ts`, `edit/modes/hashline.ts`, `edit/modes/patch.ts`, `edit/modes/replace.ts`)
- Updated package.json exports to use `./edit` path instead of `./patch` for edit tool and related utilities
- Chunk edit tool documentation simplified: removed line-based edit examples, clarified `target` format with full path and CRC suffix, added guidance for `replace_body` operation to preserve declarations
+30 -1
View File
@@ -22,7 +22,7 @@ import { selectSession } from "./cli/session-picker";
import { findConfigFile } from "./config";
import { ModelRegistry, ModelsConfigFile } from "./config/model-registry";
import { resolveCliModel, resolveModelRoleValue, resolveModelScope, type ScopedModel } from "./config/model-resolver";
import { Settings, settings } from "./config/settings";
import { getDefault, type SettingPath, Settings, settings } from "./config/settings";
import { initializeWithSettings } from "./discovery";
import {
clearClaudePluginRootsCache,
@@ -71,6 +71,29 @@ async function checkForNewVersion(currentVersion: string): Promise<string | unde
}
}
const RPC_DEFAULTED_SETTING_PATHS: SettingPath[] = [
"todo.enabled",
"todo.reminders",
"todo.reminders.max",
"todo.eager",
"async.enabled",
"async.maxJobs",
"task.isolation.mode",
"task.isolation.merge",
"task.isolation.commits",
"task.eager",
"task.maxConcurrency",
"task.maxRecursionDepth",
"task.disabledAgents",
"task.agentModelOverrides",
];
function applyRpcDefaultSettingOverrides(): void {
for (const settingPath of RPC_DEFAULTED_SETTING_PATHS) {
settings.override(settingPath, getDefault(settingPath));
}
}
async function readPipedInput(): Promise<string | undefined> {
if (process.stdin.isTTY !== false) return undefined;
try {
@@ -590,9 +613,15 @@ export async function runRootCommand(parsed: Args, rawArgs: string[]): Promise<v
const cwd = getProjectDir();
await logger.timeAsync("settings:init", () => Settings.init({ cwd }));
if (parsedArgs.mode === "rpc") {
applyRpcDefaultSettingOverrides();
}
if (parsedArgs.noPty) {
Bun.env.PI_NO_PTY = "1";
}
if (parsedArgs.noTitle || parsedArgs.mode === "rpc") {
Bun.env.PI_NO_TITLE = "1";
}
const { pipedInput, fileText, fileImages } = await logger.timeAsync("prepareInitialMessage", async () => {
const pipedInput = await readPipedInput();
if (parsedArgs.fileArgs.length === 0) {
@@ -10,7 +10,7 @@
* - Events: AgentSessionEvent objects streamed as they occur
* - Extension UI: Extension UI requests are emitted, client responds with extension_ui_response
*/
import { readJsonl, Snowflake } from "@oh-my-pi/pi-utils";
import { $env, readJsonl, Snowflake } from "@oh-my-pi/pi-utils";
import type {
ExtensionUIContext,
ExtensionUIDialogOptions,
@@ -36,6 +36,13 @@ export type PendingExtensionRequest = {
type RpcOutput = (obj: RpcResponse | RpcExtensionUIRequest | object) => void;
function shouldEmitRpcTitles(): boolean {
const raw = $env.PI_RPC_EMIT_TITLE;
if (!raw) return false;
const normalized = raw.trim().toLowerCase();
return normalized === "1" || normalized === "true" || normalized === "yes" || normalized === "on";
}
export function requestRpcEditor(
pendingRequests: Map<string, PendingExtensionRequest>,
output: RpcOutput,
@@ -110,6 +117,7 @@ export async function runRpcMode(session: AgentSession): Promise<never> {
const output = (obj: RpcResponse | RpcExtensionUIRequest | object) => {
process.stdout.write(`${JSON.stringify(obj)}\n`);
};
const emitRpcTitles = shouldEmitRpcTitles();
const success = <T extends RpcCommand["type"]>(
id: string | undefined,
@@ -291,7 +299,8 @@ export async function runRpcMode(session: AgentSession): Promise<never> {
}
setTitle(title: string): void {
// Fire and forget - host can implement terminal title control
// Title updates are low-value noise for most RPC hosts; opt in via PI_RPC_EMIT_TITLE=1.
if (!emitRpcTitles) return;
this.output({
type: "extension_ui_request",
id: Snowflake.next() as string,
@@ -544,6 +553,7 @@ export async function runRpcMode(session: AgentSession): Promise<never> {
autoCompactionEnabled: session.autoCompactionEnabled,
messageCount: session.messages.length,
queuedMessageCount: session.queuedMessageCount,
todoPhases: session.getTodoPhases(),
systemPrompt: session.systemPrompt,
dumpTools: session.agent.state.tools.map(tool => ({
name: tool.name,
@@ -554,6 +564,11 @@ export async function runRpcMode(session: AgentSession): Promise<never> {
return success(id, "get_state", state);
}
case "set_todos": {
session.setTodoPhases(command.phases);
return success(id, "set_todos", { todoPhases: session.getTodoPhases() });
}
// =================================================================
// Model
// =================================================================
@@ -9,6 +9,7 @@ import type { Effort, ImageContent, Model } from "@oh-my-pi/pi-ai";
import type { BashResult } from "../../exec/bash-executor";
import type { SessionStats } from "../../session/agent-session";
import type { CompactionResult } from "../../session/compaction";
import type { TodoPhase } from "../../tools/todo-write";
// ============================================================================
// RPC Commands (stdin)
@@ -25,6 +26,7 @@ export type RpcCommand =
// State
| { id?: string; type: "get_state" }
| { id?: string; type: "set_todos"; phases: TodoPhase[] }
// Model
| { id?: string; type: "set_model"; provider: string; modelId: string }
@@ -82,6 +84,7 @@ export interface RpcSessionState {
autoCompactionEnabled: boolean;
messageCount: number;
queuedMessageCount: number;
todoPhases: TodoPhase[];
/** For session dump / export (plain-text parity with /dump). */
systemPrompt?: string;
dumpTools?: Array<{ name: string; description: string; parameters: unknown }>;
@@ -103,6 +106,7 @@ export type RpcResponse =
// State
| { id?: string; type: "response"; command: "get_state"; success: true; data: RpcSessionState }
| { id?: string; type: "response"; command: "set_todos"; success: true; data: { todoPhases: TodoPhase[] } }
// Model
| {
+62 -1
View File
@@ -6,9 +6,12 @@ This package wraps the newline-delimited JSON RPC transport exposed by the CLI a
provides:
- typed command methods for the stable RPC surface
- typed startup options for common `omp --mode rpc` flags such as thinking level,
tool selection, prompt appends, provider session IDs, and headless session toggles
- typed protocol models for state, bash results, compaction, and session stats
- a process-backed client that manages request correlation over stdio
- helpers for collecting prompt runs and handling extension UI requests
- typed per-event listeners plus a typed catch-all notification hook
- helpers for collecting prompt runs and handling extension UI requests in manual or headless mode
## Basic Usage
@@ -23,6 +26,50 @@ with RpcClient(provider="anthropic", model="claude-sonnet-4-5") as client:
print(turn.require_assistant_text())
```
The wrapper also exposes the common RPC startup flags directly, so scripts do not
need to build `extra_args` by hand:
```python
from omp_rpc import RpcClient
with RpcClient(
model="openrouter/anthropic/claude-sonnet-4.6",
thinking="high",
no_session=True,
no_skills=True,
no_rules=True,
tools=("read", "edit", "write"),
append_system_prompt="Focus on reproducible benchmark behavior.",
) as client:
print(client.get_state().thinking_level)
```
For orchestration hosts, the wrapper also exposes typed event hooks and a simple
way to seed todos before the first prompt:
```python
from omp_rpc import MessageUpdateEvent, RpcClient
def on_message_update(event: MessageUpdateEvent) -> None:
assistant_event = event.assistant_message_event
if assistant_event.get("type") == "text_delta":
print(assistant_event["delta"], end="", flush=True)
with RpcClient(model="openrouter/anthropic/claude-sonnet-4.6", no_session=True) as client:
client.on_message_update(on_message_update)
client.set_todos(
[
"Map the read and edit tool surface.",
"Exercise the supported edit paths.",
"Write concrete findings and gaps.",
]
)
client.prompt_and_wait("Evaluate the current tool behavior.")
```
`set_todos()` accepts either a flat list of todo strings/items or explicit
phases, and `get_state().todo_phases` returns the typed current todo state.
By default the client runs:
```bash
@@ -64,6 +111,20 @@ elif request.method in {"input", "editor"}:
client.send_ui_value(request.id, "approved")
```
For non-interactive scripts, you can install a default headless policy instead of
handling every request manually:
```python
with RpcClient(model="anthropic/claude-sonnet-4-5") as client:
client.install_headless_ui()
turn = client.prompt_and_wait("needs ui-safe automation")
print(turn.assistant_text)
```
That helper ignores passive UI notifications (`notify`, `setStatus`, `setWidget`,
`setTitle`, `set_editor_text`), answers `confirm` with `False`, and cancels
`select`/`input`/`editor` requests unless you provide explicit values.
## Protocol Reference
The canonical wire protocol still lives in the repo at
+16
View File
@@ -1,7 +1,9 @@
from .client import (
AgentEventListener,
ExtensionErrorListener,
NotificationListener,
PromptTurn,
ReadyListener,
RpcClient,
RpcCommandError,
RpcError,
@@ -48,6 +50,11 @@ from .protocol import (
ThinkingConfig,
ThinkingLevel,
ThinkingLevelCycleResult,
TodoAutoClearEvent,
TodoItem,
TodoPhase,
TodoReminderEvent,
TodoStatus,
ToolDescriptor,
ToolExecutionEndEvent,
ToolExecutionStartEvent,
@@ -62,6 +69,7 @@ from .protocol import (
message_text,
parse_notification,
parse_session_state,
parse_todo_phases,
)
__all__ = [
@@ -94,9 +102,11 @@ __all__ = [
"ModelCost",
"ModelCycleResult",
"ModelInfo",
"NotificationListener",
"PromptTurn",
"PythonExecutionMessage",
"ReadyEvent",
"ReadyListener",
"RetryFallbackAppliedEvent",
"RetryFallbackSucceededEvent",
"RpcAgentEvent",
@@ -111,6 +121,11 @@ __all__ = [
"ThinkingConfig",
"ThinkingLevel",
"ThinkingLevelCycleResult",
"TodoAutoClearEvent",
"TodoItem",
"TodoPhase",
"TodoReminderEvent",
"TodoStatus",
"ToolDescriptor",
"ToolExecutionEndEvent",
"ToolExecutionStartEvent",
@@ -126,4 +141,5 @@ __all__ = [
"message_text",
"parse_notification",
"parse_session_state",
"parse_todo_phases",
]
+321 -6
View File
@@ -11,9 +11,14 @@ from pathlib import Path
from typing import Any, Callable, Mapping, Sequence, TypeVar, cast
from .protocol import (
AgentStartEvent,
AgentEndEvent,
AgentMessage,
AssistantMessage,
AutoCompactionEndEvent,
AutoCompactionStartEvent,
AutoRetryEndEvent,
AutoRetryStartEvent,
BashResult,
BranchMessage,
BranchResult,
@@ -25,16 +30,34 @@ from .protocol import (
InterruptMode,
JsonObject,
JsonValue,
MessageEndEvent,
MessageStartEvent,
MessageUpdateEvent,
ModelCycleResult,
ModelInfo,
ReadyEvent,
RetryFallbackAppliedEvent,
RetryFallbackSucceededEvent,
RpcAgentEvent,
RpcNotification,
SessionState,
SessionStats,
SteeringMode,
StreamingBehavior,
ThinkingLevel,
ThinkingLevelCycleResult,
TodoItem,
TodoPhase,
TodoStatus,
TodoAutoClearEvent,
TodoReminderEvent,
ToolExecutionEndEvent,
ToolExecutionStartEvent,
ToolExecutionUpdateEvent,
TtsrTriggeredEvent,
TurnEndEvent,
TurnStartEvent,
UnknownNotification,
assistant_text,
parse_bash_result,
parse_branch_messages,
@@ -47,12 +70,36 @@ from .protocol import (
parse_session_state,
parse_session_stats,
parse_thinking_level_cycle_result,
parse_todo_phases,
)
AgentEventListener = Callable[[RpcAgentEvent], None]
NotificationListener = Callable[[RpcNotification], None]
UiRequestListener = Callable[[ExtensionUiRequest], None]
ExtensionErrorListener = Callable[[ExtensionError], None]
ReadyListener = Callable[[ReadyEvent], None]
UnknownNotificationListener = Callable[[UnknownNotification], None]
AgentStartListener = Callable[[AgentStartEvent], None]
AgentEndListener = Callable[[AgentEndEvent], None]
TurnStartListener = Callable[[TurnStartEvent], None]
TurnEndListener = Callable[[TurnEndEvent], None]
MessageStartListener = Callable[[MessageStartEvent], None]
MessageUpdateListener = Callable[[MessageUpdateEvent], None]
MessageEndListener = Callable[[MessageEndEvent], None]
ToolExecutionStartListener = Callable[[ToolExecutionStartEvent], None]
ToolExecutionUpdateListener = Callable[[ToolExecutionUpdateEvent], None]
ToolExecutionEndListener = Callable[[ToolExecutionEndEvent], None]
AutoCompactionStartListener = Callable[[AutoCompactionStartEvent], None]
AutoCompactionEndListener = Callable[[AutoCompactionEndEvent], None]
AutoRetryStartListener = Callable[[AutoRetryStartEvent], None]
AutoRetryEndListener = Callable[[AutoRetryEndEvent], None]
RetryFallbackAppliedListener = Callable[[RetryFallbackAppliedEvent], None]
RetryFallbackSucceededListener = Callable[[RetryFallbackSucceededEvent], None]
TtsrTriggeredListener = Callable[[TtsrTriggeredEvent], None]
TodoReminderListener = Callable[[TodoReminderEvent], None]
TodoAutoClearListener = Callable[[TodoAutoClearEvent], None]
TListener = TypeVar("TListener")
TEventListener = TypeVar("TEventListener", bound=Callable[..., None])
class RpcError(RuntimeError):
@@ -89,6 +136,10 @@ class PromptTurn:
return self.assistant_text
TodoSeed = str | TodoItem | Mapping[str, object]
TodoPhaseSeed = TodoPhase | Mapping[str, object]
class RpcClient:
def __init__(
self,
@@ -100,6 +151,15 @@ class RpcClient:
session_dir: str | Path | None = None,
cwd: str | Path | None = None,
env: Mapping[str, str] | None = None,
thinking: ThinkingLevel | None = None,
append_system_prompt: str | None = None,
provider_session_id: str | None = None,
tools: Sequence[str] | None = None,
no_session: bool = False,
no_skills: bool = False,
no_rules: bool = False,
no_title: bool | None = None,
rpc_defaults: bool = True,
extra_args: Sequence[str] = (),
startup_timeout: float = 30.0,
request_timeout: float = 30.0,
@@ -111,6 +171,15 @@ class RpcClient:
self._session_dir = Path(session_dir) if session_dir is not None else None
self._cwd = Path(cwd) if cwd is not None else None
self._env = dict(env or {})
self._thinking = thinking
self._append_system_prompt = append_system_prompt
self._provider_session_id = provider_session_id
self._tools = tuple(tools) if tools is not None else None
self._no_session = no_session
self._no_skills = no_skills
self._no_rules = no_rules
self._no_title = no_title
self._rpc_defaults = rpc_defaults
self._extra_args = tuple(extra_args)
self._startup_timeout = startup_timeout
self._request_timeout = request_timeout
@@ -130,7 +199,11 @@ class RpcClient:
self._closed_error: BaseException | None = None
self._stopping = False
self._notification_listeners: list[NotificationListener] = []
self._event_listeners: list[AgentEventListener] = []
self._typed_event_listeners: dict[str, list[AgentEventListener]] = {}
self._ready_listeners: list[ReadyListener] = []
self._unknown_notification_listeners: list[UnknownNotificationListener] = []
self._ui_request_listeners: list[UiRequestListener] = []
self._extension_error_listeners: list[ExtensionErrorListener] = []
@@ -232,6 +305,71 @@ class RpcClient:
self._event_listeners.append(listener)
return lambda: self._remove_listener(self._event_listeners, listener)
def on_notification(self, listener: NotificationListener) -> Callable[[], None]:
self._notification_listeners.append(listener)
return lambda: self._remove_listener(self._notification_listeners, listener)
def on_ready(self, listener: ReadyListener) -> Callable[[], None]:
self._ready_listeners.append(listener)
return lambda: self._remove_listener(self._ready_listeners, listener)
def on_agent_start(self, listener: AgentStartListener) -> Callable[[], None]:
return self._add_typed_event_listener("agent_start", listener)
def on_agent_end(self, listener: AgentEndListener) -> Callable[[], None]:
return self._add_typed_event_listener("agent_end", listener)
def on_turn_start(self, listener: TurnStartListener) -> Callable[[], None]:
return self._add_typed_event_listener("turn_start", listener)
def on_turn_end(self, listener: TurnEndListener) -> Callable[[], None]:
return self._add_typed_event_listener("turn_end", listener)
def on_message_start(self, listener: MessageStartListener) -> Callable[[], None]:
return self._add_typed_event_listener("message_start", listener)
def on_message_update(self, listener: MessageUpdateListener) -> Callable[[], None]:
return self._add_typed_event_listener("message_update", listener)
def on_message_end(self, listener: MessageEndListener) -> Callable[[], None]:
return self._add_typed_event_listener("message_end", listener)
def on_tool_execution_start(self, listener: ToolExecutionStartListener) -> Callable[[], None]:
return self._add_typed_event_listener("tool_execution_start", listener)
def on_tool_execution_update(self, listener: ToolExecutionUpdateListener) -> Callable[[], None]:
return self._add_typed_event_listener("tool_execution_update", listener)
def on_tool_execution_end(self, listener: ToolExecutionEndListener) -> Callable[[], None]:
return self._add_typed_event_listener("tool_execution_end", listener)
def on_auto_compaction_start(self, listener: AutoCompactionStartListener) -> Callable[[], None]:
return self._add_typed_event_listener("auto_compaction_start", listener)
def on_auto_compaction_end(self, listener: AutoCompactionEndListener) -> Callable[[], None]:
return self._add_typed_event_listener("auto_compaction_end", listener)
def on_auto_retry_start(self, listener: AutoRetryStartListener) -> Callable[[], None]:
return self._add_typed_event_listener("auto_retry_start", listener)
def on_auto_retry_end(self, listener: AutoRetryEndListener) -> Callable[[], None]:
return self._add_typed_event_listener("auto_retry_end", listener)
def on_retry_fallback_applied(self, listener: RetryFallbackAppliedListener) -> Callable[[], None]:
return self._add_typed_event_listener("retry_fallback_applied", listener)
def on_retry_fallback_succeeded(self, listener: RetryFallbackSucceededListener) -> Callable[[], None]:
return self._add_typed_event_listener("retry_fallback_succeeded", listener)
def on_ttsr_triggered(self, listener: TtsrTriggeredListener) -> Callable[[], None]:
return self._add_typed_event_listener("ttsr_triggered", listener)
def on_todo_reminder(self, listener: TodoReminderListener) -> Callable[[], None]:
return self._add_typed_event_listener("todo_reminder", listener)
def on_todo_auto_clear(self, listener: TodoAutoClearListener) -> Callable[[], None]:
return self._add_typed_event_listener("todo_auto_clear", listener)
def on_ui_request(self, listener: UiRequestListener) -> Callable[[], None]:
self._ui_request_listeners.append(listener)
return lambda: self._remove_listener(self._ui_request_listeners, listener)
@@ -240,6 +378,55 @@ class RpcClient:
self._extension_error_listeners.append(listener)
return lambda: self._remove_listener(self._extension_error_listeners, listener)
def on_unknown_notification(self, listener: UnknownNotificationListener) -> Callable[[], None]:
self._unknown_notification_listeners.append(listener)
return lambda: self._remove_listener(self._unknown_notification_listeners, listener)
def install_headless_ui(
self,
*,
on_request: UiRequestListener | None = None,
confirm: bool = False,
select_value: str | None = None,
input_value: str | None = None,
editor_value: str | None = None,
) -> Callable[[], None]:
"""Auto-handle RPC UI requests for non-interactive hosts.
Passive UI methods such as notifications and status updates are ignored.
Confirm dialogs default to `False`. Select, input, and editor requests
are cancelled unless an explicit value is provided.
"""
def handle(request: ExtensionUiRequest) -> None:
if on_request is not None:
on_request(request)
if request.method == "cancel" or request.is_passive():
return
if request.method == "confirm":
self.send_ui_confirmation(request.id, confirm)
return
if request.method == "select":
if select_value is not None:
self.send_ui_value(request.id, select_value)
else:
self.cancel_ui_request(request.id)
return
if request.method == "input":
if input_value is not None:
self.send_ui_value(request.id, input_value)
else:
self.cancel_ui_request(request.id)
return
if request.method == "editor":
if editor_value is not None:
self.send_ui_value(request.id, editor_value)
else:
self.cancel_ui_request(request.id)
return self.on_ui_request(handle)
def next_ui_request(self, timeout: float | None = None) -> ExtensionUiRequest:
try:
return self._ui_requests.get(timeout=timeout)
@@ -340,6 +527,17 @@ class RpcClient:
def set_session_name(self, name: str) -> None:
self._request("set_session_name", name=name)
def get_todos(self) -> tuple[TodoPhase, ...]:
return self.get_state().todo_phases
def set_todos(self, todos: Sequence[TodoSeed | TodoPhaseSeed]) -> tuple[TodoPhase, ...]:
phases = self._normalize_todo_phases(todos)
payload = self._request("set_todos", phases=cast(JsonValue, phases))
return parse_todo_phases(payload.get("todoPhases"))
def clear_todos(self) -> tuple[TodoPhase, ...]:
return self.set_todos(())
def get_messages(self) -> tuple[AgentMessage, ...]:
payload = self._request("get_messages")
return tuple(cast(list[AgentMessage], payload.get("messages") or []))
@@ -476,6 +674,91 @@ class RpcClient:
process = self._require_process()
self._write_json(process, payload)
def _add_typed_event_listener(self, event_type: str, listener: TEventListener) -> Callable[[], None]:
listeners = self._typed_event_listeners.setdefault(event_type, [])
typed_listener = cast(AgentEventListener, listener)
listeners.append(typed_listener)
return lambda: self._remove_listener(listeners, typed_listener)
@staticmethod
def _normalize_todo_phases(todos: Sequence[TodoSeed | TodoPhaseSeed]) -> list[JsonObject]:
if len(todos) == 0:
return []
next_task_id = 1
def next_task() -> str:
nonlocal next_task_id
task_id = f"task-{next_task_id}"
next_task_id += 1
return task_id
def normalize_todo_item(seed: TodoSeed) -> JsonObject:
if isinstance(seed, str):
return {"id": next_task(), "content": seed, "status": cast(JsonValue, "pending")}
if isinstance(seed, TodoItem):
return {
"id": seed.id or next_task(),
"content": seed.content,
"status": cast(JsonValue, seed.status),
"notes": seed.notes,
"details": seed.details,
}
content = seed.get("content")
if not isinstance(content, str) or not content.strip():
raise RpcError("Todo items must provide a non-empty 'content' value")
raw_id = seed.get("id")
raw_status = seed.get("status")
raw_notes = seed.get("notes")
raw_details = seed.get("details")
status: TodoStatus = cast(TodoStatus, raw_status) if isinstance(raw_status, str) else "pending"
return {
"id": str(raw_id) if isinstance(raw_id, str) and raw_id else next_task(),
"content": content,
"status": cast(JsonValue, status),
"notes": raw_notes if isinstance(raw_notes, str) else None,
"details": raw_details if isinstance(raw_details, str) else None,
}
def is_phase_seed(seed: TodoSeed | TodoPhaseSeed) -> bool:
if isinstance(seed, TodoPhase):
return True
if not isinstance(seed, Mapping):
return False
return "tasks" in seed or ("name" in seed and "content" not in seed)
def normalize_phase(seed: TodoPhaseSeed, index: int) -> JsonObject:
if isinstance(seed, TodoPhase):
phase_id = seed.id or f"phase-{index}"
name = seed.name
tasks = [normalize_todo_item(task) for task in seed.tasks]
else:
raw_name = seed.get("name")
if not isinstance(raw_name, str) or not raw_name.strip():
raise RpcError("Todo phases must provide a non-empty 'name' value")
phase_id_value = seed.get("id")
raw_tasks = seed.get("tasks") or ()
if not isinstance(raw_tasks, Sequence) or isinstance(raw_tasks, (str, bytes)):
raise RpcError("Todo phase 'tasks' must be a sequence")
phase_id = str(phase_id_value) if isinstance(phase_id_value, str) and phase_id_value else f"phase-{index}"
name = raw_name
tasks = [normalize_todo_item(cast(TodoSeed, task)) for task in raw_tasks]
return {"id": phase_id, "name": name, "tasks": tasks}
if any(is_phase_seed(todo) for todo in todos):
phases: list[JsonObject] = []
for index, seed in enumerate(todos, start=1):
if not is_phase_seed(seed):
raise RpcError("Cannot mix flat todo items with todo phases in one set_todos() call")
phases.append(normalize_phase(cast(TodoPhaseSeed, seed), index))
return phases
return [{"id": "phase-1", "name": "Todos", "tasks": [normalize_todo_item(cast(TodoSeed, todo)) for todo in todos]}]
def _build_command(self) -> tuple[str, ...]:
if self._command is not None:
return self._command
@@ -487,6 +770,26 @@ class RpcClient:
command.extend(["--model", self._model])
if self._session_dir is not None:
command.extend(["--session-dir", str(self._session_dir)])
if self._thinking is not None:
command.extend(["--thinking", self._thinking])
if self._append_system_prompt is not None:
command.extend(["--append-system-prompt", self._append_system_prompt])
if self._provider_session_id is not None:
command.extend(["--provider-session-id", self._provider_session_id])
if self._tools is not None:
if len(self._tools) == 0:
command.append("--no-tools")
else:
command.extend(["--tools", ",".join(self._tools)])
if self._no_session:
command.append("--no-session")
if self._no_skills:
command.append("--no-skills")
if self._no_rules:
command.append("--no-rules")
emit_no_title = self._no_title if self._no_title is not None else self._rpc_defaults
if emit_no_title:
command.append("--no-title")
command.extend(self._extra_args)
return tuple(command)
@@ -533,8 +836,13 @@ class RpcClient:
continue
notification = parse_notification(payload)
for listener in list(self._notification_listeners):
listener(notification)
if isinstance(notification, ReadyEvent):
self._ready.set()
for listener in list(self._ready_listeners):
listener(notification)
continue
if isinstance(notification, ExtensionUiRequest):
@@ -548,12 +856,19 @@ class RpcClient:
listener(notification)
continue
if getattr(notification, "type", None) != "unknown":
with self._event_condition:
self._events.append(cast(RpcAgentEvent, notification))
self._event_condition.notify_all()
for listener in list(self._event_listeners):
listener(cast(RpcAgentEvent, notification))
if isinstance(notification, UnknownNotification):
for listener in list(self._unknown_notification_listeners):
listener(notification)
continue
event = cast(RpcAgentEvent, notification)
with self._event_condition:
self._events.append(event)
self._event_condition.notify_all()
for listener in list(self._event_listeners):
listener(event)
for listener in list(self._typed_event_listeners.get(event.type, [])):
listener(event)
except json.JSONDecodeError as exc:
self._mark_closed(RpcError(f"Failed to decode RPC output: {exc}"))
except Exception as exc:
+85 -5
View File
@@ -4,7 +4,7 @@ import base64
import mimetypes
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Literal, NotRequired, TypedDict, TypeAlias, cast
from typing import Any, Final, Literal, NotRequired, TypedDict, TypeAlias, cast
JsonPrimitive: TypeAlias = str | int | float | bool | None
JsonValue: TypeAlias = JsonPrimitive | list["JsonValue"] | dict[str, "JsonValue"]
@@ -18,6 +18,30 @@ InterruptMode: TypeAlias = Literal["immediate", "wait"]
StopReason: TypeAlias = Literal["stop", "length", "toolUse", "error", "aborted"]
NotifyType: TypeAlias = Literal["info", "warning", "error"]
WidgetPlacement: TypeAlias = Literal["aboveEditor", "belowEditor"]
TodoStatus: TypeAlias = Literal["pending", "in_progress", "completed", "abandoned"]
ExtensionUiMethod: TypeAlias = Literal[
"select",
"confirm",
"input",
"editor",
"cancel",
"notify",
"setStatus",
"setWidget",
"setTitle",
"set_editor_text",
]
InteractiveExtensionUiMethod: TypeAlias = Literal["select", "confirm", "input", "editor"]
PassiveExtensionUiMethod: TypeAlias = Literal["notify", "setStatus", "setWidget", "setTitle", "set_editor_text"]
ValueExtensionUiMethod: TypeAlias = Literal["select", "input", "editor"]
PASSIVE_EXTENSION_UI_METHODS: Final[frozenset[PassiveExtensionUiMethod]] = frozenset(
{"notify", "setStatus", "setWidget", "setTitle", "set_editor_text"}
)
INTERACTIVE_EXTENSION_UI_METHODS: Final[frozenset[InteractiveExtensionUiMethod]] = frozenset(
{"select", "confirm", "input", "editor"}
)
VALUE_EXTENSION_UI_METHODS: Final[frozenset[ValueExtensionUiMethod]] = frozenset({"select", "input", "editor"})
class TextContent(TypedDict, total=False):
@@ -341,6 +365,22 @@ class ToolDescriptor:
parameters: JsonValue
@dataclass(slots=True, frozen=True)
class TodoItem:
id: str
content: str
status: TodoStatus
notes: str | None = None
details: str | None = None
@dataclass(slots=True, frozen=True)
class TodoPhase:
id: str
name: str
tasks: tuple[TodoItem, ...]
@dataclass(slots=True, frozen=True)
class SessionState:
model: ModelInfo | None
@@ -356,6 +396,7 @@ class SessionState:
auto_compaction_enabled: bool
message_count: int
queued_message_count: int
todo_phases: tuple[TodoPhase, ...] = ()
system_prompt: str | None = None
dump_tools: tuple[ToolDescriptor, ...] = ()
@@ -443,7 +484,7 @@ class ReadyEvent:
@dataclass(slots=True, frozen=True)
class ExtensionUiRequest:
id: str
method: str
method: ExtensionUiMethod
title: str | None = None
options: tuple[str, ...] | None = None
message: str | None = None
@@ -459,6 +500,19 @@ class ExtensionUiRequest:
widget_lines: tuple[str, ...] | None = None
widget_placement: WidgetPlacement | None = None
text: str | None = None
type: Literal["extension_ui_request"] = "extension_ui_request"
def is_passive(self) -> bool:
return self.method in PASSIVE_EXTENSION_UI_METHODS
def is_interactive(self) -> bool:
return self.method in INTERACTIVE_EXTENSION_UI_METHODS
def accepts_text(self) -> bool:
return self.method in VALUE_EXTENSION_UI_METHODS
def requires_response(self) -> bool:
return self.is_interactive()
@dataclass(slots=True, frozen=True)
@@ -596,7 +650,7 @@ class TtsrTriggeredEvent:
@dataclass(slots=True, frozen=True)
class TodoReminderEvent:
todos: tuple[JsonObject, ...]
todos: tuple[TodoItem, ...]
attempt: int
max_attempts: int
type: Literal["todo_reminder"] = "todo_reminder"
@@ -728,6 +782,31 @@ def parse_tool_descriptor(payload: JsonObject) -> ToolDescriptor:
)
def parse_todo_item(payload: JsonObject) -> TodoItem:
return TodoItem(
id=str(payload.get("id", "")),
content=str(payload.get("content", "")),
status=cast(TodoStatus, payload.get("status", "pending")),
notes=str(payload["notes"]) if payload.get("notes") is not None else None,
details=str(payload["details"]) if payload.get("details") is not None else None,
)
def parse_todo_phase(payload: JsonObject) -> TodoPhase:
tasks = tuple(parse_todo_item(cast(JsonObject, item)) for item in cast(list[Any], payload.get("tasks") or []))
return TodoPhase(
id=str(payload.get("id", "")),
name=str(payload.get("name", "")),
tasks=tasks,
)
def parse_todo_phases(payload: JsonValue | None) -> tuple[TodoPhase, ...]:
if not isinstance(payload, list):
return ()
return tuple(parse_todo_phase(cast(JsonObject, item)) for item in payload)
def parse_session_state(payload: JsonObject) -> SessionState:
dump_tools = tuple(
parse_tool_descriptor(cast(JsonObject, item)) for item in cast(list[Any], payload.get("dumpTools") or [])
@@ -746,6 +825,7 @@ def parse_session_state(payload: JsonObject) -> SessionState:
auto_compaction_enabled=bool(payload.get("autoCompactionEnabled", False)),
message_count=int(payload.get("messageCount", 0)),
queued_message_count=int(payload.get("queuedMessageCount", 0)),
todo_phases=parse_todo_phases(cast(JsonValue | None, payload.get("todoPhases"))),
system_prompt=str(payload["systemPrompt"]) if payload.get("systemPrompt") is not None else None,
dump_tools=dump_tools,
)
@@ -836,7 +916,7 @@ def parse_session_stats(payload: JsonObject) -> SessionStats:
def parse_extension_ui_request(payload: JsonObject) -> ExtensionUiRequest:
return ExtensionUiRequest(
id=str(payload["id"]),
method=str(payload["method"]),
method=cast(ExtensionUiMethod, payload["method"]),
title=str(payload["title"]) if payload.get("title") is not None else None,
options=tuple(str(item) for item in cast(list[Any], payload.get("options") or [])) or None,
message=str(payload["message"]) if payload.get("message") is not None else None,
@@ -954,7 +1034,7 @@ def parse_notification(payload: JsonObject) -> RpcNotification:
return TtsrTriggeredEvent(rules=tuple(cast(list[JsonObject], payload.get("rules") or [])))
if event_type == "todo_reminder":
return TodoReminderEvent(
todos=tuple(cast(list[JsonObject], payload.get("todos") or [])),
todos=tuple(parse_todo_item(cast(JsonObject, item)) for item in cast(list[Any], payload.get("todos") or [])),
attempt=int(payload.get("attempt", 0)),
max_attempts=int(payload.get("maxAttempts", 0)),
)
+104
View File
@@ -41,6 +41,7 @@ FAKE_SERVER = textwrap.dedent(
}
print(json.dumps({"type": "ready"}), flush=True)
todo_phases = []
for raw_line in sys.stdin:
raw_line = raw_line.strip()
@@ -91,6 +92,23 @@ FAKE_SERVER = textwrap.dedent(
"autoCompactionEnabled": True,
"messageCount": 0,
"queuedMessageCount": 0,
"todoPhases": todo_phases,
},
}
),
flush=True,
)
elif command_type == "set_todos":
todo_phases = command.get("phases", [])
print(
json.dumps(
{
"id": request_id,
"type": "response",
"command": "set_todos",
"success": True,
"data": {
"todoPhases": todo_phases,
},
}
),
@@ -146,6 +164,7 @@ FAKE_SERVER = textwrap.dedent(
continue
print(json.dumps({"type": "agent_start"}), flush=True)
print(json.dumps({"type": "turn_start"}), flush=True)
partial = assistant_message("")
print(
json.dumps(
@@ -164,6 +183,7 @@ FAKE_SERVER = textwrap.dedent(
)
assistant = assistant_message("pong")
print(json.dumps({"type": "message_end", "message": assistant}), flush=True)
print(json.dumps({"type": "turn_end", "message": assistant, "toolResults": []}), flush=True)
print(json.dumps({"type": "agent_end", "messages": [assistant]}), flush=True)
else:
print(
@@ -190,6 +210,46 @@ class RpcClientTests(unittest.TestCase):
request_timeout=2.0,
)
def test_command_builder_supports_common_rpc_options(self) -> None:
client = RpcClient(
executable="omp",
model="openrouter/anthropic/claude-sonnet-4.6",
cwd="/tmp/workspace",
thinking="high",
append_system_prompt="extra instructions",
provider_session_id="provider-session-1",
tools=("read", "edit", "write"),
no_session=True,
no_skills=True,
no_rules=True,
extra_args=("--foo", "bar"),
)
self.assertEqual(
client.command,
(
"omp",
"--mode",
"rpc",
"--model",
"openrouter/anthropic/claude-sonnet-4.6",
"--thinking",
"high",
"--append-system-prompt",
"extra instructions",
"--provider-session-id",
"provider-session-1",
"--tools",
"read,edit,write",
"--no-session",
"--no-skills",
"--no-rules",
"--no-title",
"--foo",
"bar",
),
)
def test_get_state_and_bash(self) -> None:
with self.make_client() as client:
state = client.get_state()
@@ -215,6 +275,50 @@ class RpcClientTests(unittest.TestCase):
client.send_ui_value(request.id, "approved")
client.wait_for_idle(timeout=2.0)
def test_install_headless_ui_cancels_interactive_requests(self) -> None:
seen_methods: list[str] = []
with self.make_client() as client:
client.install_headless_ui(on_request=lambda request: seen_methods.append(request.method))
client.prompt_and_wait("needs ui", timeout=2.0)
self.assertEqual(seen_methods, ["input"])
def test_ready_and_typed_event_listeners(self) -> None:
ready_types: list[str] = []
event_types: list[str] = []
notification_types: list[str] = []
client = self.make_client()
client.on_ready(lambda event: ready_types.append(event.type))
client.on_notification(lambda notification: notification_types.append(notification.type))
client.on_turn_start(lambda event: event_types.append(event.type))
client.on_message_update(lambda event: event_types.append(event.type))
client.on_agent_end(lambda event: event_types.append(event.type))
try:
client.start()
client.prompt_and_wait("say hello", timeout=2.0)
finally:
client.stop()
self.assertEqual(ready_types, ["ready"])
self.assertEqual(event_types, ["turn_start", "message_update", "agent_end"])
self.assertIn("ready", notification_types)
self.assertIn("turn_start", notification_types)
self.assertIn("agent_end", notification_types)
def test_set_todos_supports_flat_items(self) -> None:
with self.make_client() as client:
phases = client.set_todos(["Map tools", "Exercise edits"])
self.assertEqual(len(phases), 1)
self.assertEqual(phases[0].name, "Todos")
self.assertEqual(phases[0].tasks[0].content, "Map tools")
self.assertEqual(phases[0].tasks[1].status, "pending")
state = client.get_state()
self.assertEqual(state.todo_phases[0].tasks[1].content, "Exercise edits")
if __name__ == "__main__":
unittest.main()
+39
View File
@@ -6,6 +6,7 @@ from omp_rpc import (
AgentEndEvent,
ExtensionUiRequest,
SessionState,
TodoReminderEvent,
assistant_text,
parse_notification,
parse_session_state,
@@ -50,6 +51,20 @@ class ProtocolParsingTests(unittest.TestCase):
"autoCompactionEnabled": True,
"messageCount": 4,
"queuedMessageCount": 1,
"todoPhases": [
{
"id": "phase-1",
"name": "Todos",
"tasks": [
{
"id": "task-1",
"content": "Map tools",
"status": "in_progress",
"details": "Inspect read and edit first.",
}
],
}
],
"systemPrompt": "You are useful.",
"dumpTools": [
{
@@ -65,6 +80,7 @@ class ProtocolParsingTests(unittest.TestCase):
self.assertEqual(state.session_id, "session-123")
self.assertEqual(state.follow_up_mode, "all")
self.assertEqual(state.model.id if state.model else None, "claude-sonnet-4-5")
self.assertEqual(state.todo_phases[0].tasks[0].status, "in_progress")
self.assertEqual(state.dump_tools[0].name, "read")
def test_parse_agent_end_notification(self) -> None:
@@ -117,6 +133,29 @@ class ProtocolParsingTests(unittest.TestCase):
self.assertIsInstance(notification, ExtensionUiRequest)
self.assertEqual(notification.method, "confirm")
self.assertEqual(notification.message, "Continue?")
self.assertTrue(notification.is_interactive())
self.assertTrue(notification.requires_response())
self.assertFalse(notification.is_passive())
def test_parse_todo_reminder_notification(self) -> None:
notification = parse_notification(
{
"type": "todo_reminder",
"attempt": 1,
"maxAttempts": 3,
"todos": [
{
"id": "task-1",
"content": "Map tools",
"status": "pending",
}
],
}
)
self.assertIsInstance(notification, TodoReminderEvent)
self.assertEqual(notification.todos[0].content, "Map tools")
self.assertEqual(notification.todos[0].status, "pending")
if __name__ == "__main__":