From d7261bcbeb73d5fba457f0dadd5fca1b64a37ec3 Mon Sep 17 00:00:00 2001 From: can1357 Date: Wed, 8 Apr 2026 04:52:46 +0200 Subject: [PATCH] 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. --- docs/rpc.md | 59 +++- packages/coding-agent/CHANGELOG.md | 5 + packages/coding-agent/src/main.ts | 31 +- .../coding-agent/src/modes/rpc/rpc-mode.ts | 19 +- .../coding-agent/src/modes/rpc/rpc-types.ts | 4 + python/omp-rpc/README.md | 63 +++- python/omp-rpc/src/omp_rpc/__init__.py | 16 + python/omp-rpc/src/omp_rpc/client.py | 327 +++++++++++++++++- python/omp-rpc/src/omp_rpc/protocol.py | 90 ++++- python/omp-rpc/tests/test_client.py | 104 ++++++ python/omp-rpc/tests/test_protocol.py | 39 +++ 11 files changed, 740 insertions(+), 17 deletions(-) diff --git a/docs/rpc.md b/docs/rpc.md index fd77e0d38..979b31b36 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -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. \ No newline at end of file +Use raw protocol frames if you need complete surface coverage. diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index cae2c4634..f62b331b3 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -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 diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index ba3d664b8..933c80784 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -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 { if (process.stdin.isTTY !== false) return undefined; try { @@ -590,9 +613,15 @@ export async function runRootCommand(parsed: Args, rawArgs: string[]): Promise 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) { diff --git a/packages/coding-agent/src/modes/rpc/rpc-mode.ts b/packages/coding-agent/src/modes/rpc/rpc-mode.ts index 7c725d45f..8bf2a8c07 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-mode.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-mode.ts @@ -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, output: RpcOutput, @@ -110,6 +117,7 @@ export async function runRpcMode(session: AgentSession): Promise { const output = (obj: RpcResponse | RpcExtensionUIRequest | object) => { process.stdout.write(`${JSON.stringify(obj)}\n`); }; + const emitRpcTitles = shouldEmitRpcTitles(); const success = ( id: string | undefined, @@ -291,7 +299,8 @@ export async function runRpcMode(session: AgentSession): Promise { } 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 { 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 { return success(id, "get_state", state); } + case "set_todos": { + session.setTodoPhases(command.phases); + return success(id, "set_todos", { todoPhases: session.getTodoPhases() }); + } + // ================================================================= // Model // ================================================================= diff --git a/packages/coding-agent/src/modes/rpc/rpc-types.ts b/packages/coding-agent/src/modes/rpc/rpc-types.ts index 419271bcf..18c467655 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-types.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-types.ts @@ -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 | { diff --git a/python/omp-rpc/README.md b/python/omp-rpc/README.md index 6fe04c8e1..f6bb448ea 100644 --- a/python/omp-rpc/README.md +++ b/python/omp-rpc/README.md @@ -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 diff --git a/python/omp-rpc/src/omp_rpc/__init__.py b/python/omp-rpc/src/omp_rpc/__init__.py index a68f6ea3e..0be2cdae8 100644 --- a/python/omp-rpc/src/omp_rpc/__init__.py +++ b/python/omp-rpc/src/omp_rpc/__init__.py @@ -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", ] diff --git a/python/omp-rpc/src/omp_rpc/client.py b/python/omp-rpc/src/omp_rpc/client.py index d17537d4c..344388610 100644 --- a/python/omp-rpc/src/omp_rpc/client.py +++ b/python/omp-rpc/src/omp_rpc/client.py @@ -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: diff --git a/python/omp-rpc/src/omp_rpc/protocol.py b/python/omp-rpc/src/omp_rpc/protocol.py index 4fce000a2..27482d607 100644 --- a/python/omp-rpc/src/omp_rpc/protocol.py +++ b/python/omp-rpc/src/omp_rpc/protocol.py @@ -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)), ) diff --git a/python/omp-rpc/tests/test_client.py b/python/omp-rpc/tests/test_client.py index 0e3928128..3a1c3f199 100644 --- a/python/omp-rpc/tests/test_client.py +++ b/python/omp-rpc/tests/test_client.py @@ -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() diff --git a/python/omp-rpc/tests/test_protocol.py b/python/omp-rpc/tests/test_protocol.py index bb7fb0795..3eb56234b 100644 --- a/python/omp-rpc/tests/test_protocol.py +++ b/python/omp-rpc/tests/test_protocol.py @@ -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__":