From 69e36e0aa749b68ab34eb878da382e13bad588f9 Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 30 Jul 2026 17:50:02 +0700 Subject: [PATCH 1/5] feat: add Eva daemon Codex control routing --- .codex-plugin/plugin.json | 2 +- README.md | 22 +- docs/BETA_RELEASE_RUNBOOK.md | 2 +- docs/CLAIM_AUDIT.md | 14 +- docs/HERMES_ADAPTER_BOUNDARY.md | 19 +- docs/SETUP.md | 37 +- docs/releases/CHANGELOG.md | 3 +- docs/releases/RELEASE_NOTES_1.7.0.md | 53 +++ openclaw.plugin.json | 210 ++++++++-- package-lock.json | 39 +- package.json | 6 +- packages/adapters/src/codex-jsonrpc.ts | 180 ++++++++- packages/adapters/src/index.ts | 23 +- packages/adapters/src/policy.ts | 2 + packages/cli/src/hermes-smoke.ts | 2 + .../mcp-server/src/codex-control-router.ts | 344 ++++++++++++++++ .../mcp-server/src/codex-runtime-transport.ts | 73 ++++ packages/mcp-server/src/server-runtime.ts | 24 +- packages/mcp-server/src/tools.ts | 70 +++- packages/openclaw-plugin/openclaw.plugin.json | 210 ++++++++-- tests/bridge-safety-core.test.ts | 89 ++++ tests/codex-control-router.test.ts | 382 ++++++++++++++++++ tests/codex-runtime-transport.test.ts | 45 +++ tests/control-and-mcp.test.ts | 18 +- tests/hermes-smoke.test.ts | 2 +- tests/openclaw-plugin-manifest.test.ts | 2 +- tests/qa-lab-tool-coverage.test.ts | 14 +- tests/tool-exposure-profile.test.ts | 14 +- 28 files changed, 1766 insertions(+), 135 deletions(-) create mode 100644 docs/releases/RELEASE_NOTES_1.7.0.md create mode 100644 packages/mcp-server/src/codex-control-router.ts create mode 100644 packages/mcp-server/src/codex-runtime-transport.ts create mode 100644 tests/codex-control-router.test.ts create mode 100644 tests/codex-runtime-transport.test.ts diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index 2ef7b039..f7a14689 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "lossless-openclaw-orchestrator", - "version": "1.6.1", + "version": "1.7.0", "description": "Index, search, and prepare local Codex sessions with LCO, including one-shot thread title aliases for easier recall.", "author": { "name": "100yenadmin", diff --git a/README.md b/README.md index a151b543..d3273996 100644 --- a/README.md +++ b/README.md @@ -310,8 +310,10 @@ For an agent or MCP client, start with the normal operator path: | 5 | `lco_recent_sessions` | Recent and active Codex work as compact cards. | | 6 | `lco_attention_inbox` | Blocked, waiting, stale, approval-needed, or ready-for-review work. | | 7 | `lco_project_digest` | A project-level handoff brief. | -| 8 | `lco_codex_control_dry_run` | A preview packet for the exact Codex action. | -| 9 | `lco_codex_resume_thread` | Resume a Codex thread after the dry-run packet is approved. | +| 8 | `lco_codex_control_route` | An expiring opaque reference for one daemon-owned Codex task, or an explicit Desktop-observation blocker. | +| 9 | `lco_codex_deliver` | A dry-run-first delivery that sends when idle or steers the matching active turn. | +| 10 | `lco_codex_control_dry_run` | A preview packet for a lower-level exact Codex action. | +| 11 | `lco_codex_resume_thread` | Resume a Codex thread after the dry-run packet is approved. | The packaged agent playbook is [skills/lossless-openclaw-orchestrator/SKILL.md](skills/lossless-openclaw-orchestrator/SKILL.md). @@ -377,12 +379,28 @@ mcp_servers: enabled: true env: LCO_TOOL_PROFILE: standard + LCO_CODEX_TRANSPORT: daemon ``` Omitting `LCO_DB_PATH` uses LCO's home-based default. If you set it explicitly, use an expanded absolute path because Hermes does not shell-expand `~` inside an environment-variable value. +`LCO_CODEX_TRANSPORT` defaults to `stdio` for compatibility. The opt-in +`daemon` mode connects only to the already-running local Codex managed daemon +through its Unix socket. LCO does not start or restart the daemon and does not +enable Codex Remote Control. Set `LCO_CODEX_DAEMON_SOCKET` only for an explicit +absolute local override; `LCO_CODEX_APP_SERVER_ARGS` remains stdio-only. + +For remote operation, Eva should call `lco_codex_control_route` first. A +selected `app_server` route is safe for LCO daemon/CLI task control. A +`desktop_observation_required` result means Hermes must use its own +`computer_use` integration to identify and operate the Codex Desktop task; +LCO must not silently redirect that task to the managed daemon. Telegram +direction is user to Eva for instructions and Eva to the same user for results. +An operator should not type a bot message into Telegram as though it came from +Eva. + Before a release, verify the candidate without changing a Hermes profile: ```bash diff --git a/docs/BETA_RELEASE_RUNBOOK.md b/docs/BETA_RELEASE_RUNBOOK.md index ade9161d..18628383 100644 --- a/docs/BETA_RELEASE_RUNBOOK.md +++ b/docs/BETA_RELEASE_RUNBOOK.md @@ -445,7 +445,7 @@ Record `npm dist-tag ls lossless-codex-orchestrator` in the release evidence after every npm publication. Stable releases publish with `npm publish --tag latest`; public betas publish with `npm publish --tag beta`; release candidates publish with `npm publish --tag next`. The stable channel -target for this package version is `1.6.1`; npm `latest` must move only after +target for this package version is `1.7.0`; npm `latest` must move only after the separate stable-promotion gate proves the exact candidate. Keep beta and other prereleases on prerelease tags. Do not publish a fake stable package just to move a dist-tag. Release candidates must publish with `npm publish --tag next`; diff --git a/docs/CLAIM_AUDIT.md b/docs/CLAIM_AUDIT.md index c21f7c2e..25272580 100644 --- a/docs/CLAIM_AUDIT.md +++ b/docs/CLAIM_AUDIT.md @@ -1,6 +1,6 @@ # Public Claim Audit -## Allowed 1.6.1 Stable Claim +## Allowed 1.7.0 Stable Claim This wording becomes allowed only after npm, Git tag, and GitHub Release publication truth agree for the exact candidate. The current source version @@ -8,8 +8,9 @@ alone does not authorize it. Coordinate local Codex work through Hermes/MCP, with OpenClaw compatibility, using bounded recall and session diffs, audited review-then-drive dry-runs, -approved disposable-target Codex controls, and read-only LCM prepared state; -Claude targeting remains dry-run only. +approval-bound daemon/CLI task routing and delivery, Hermes Computer Use for +Desktop-owned tasks, and read-only LCM prepared state; Claude targeting remains +dry-run only. This claim covers the stable local recall and control-plane paths with tests and smoke coverage: Codex indexing/search/describe/expand, session-diff cursors, review-then-drive dry-run packets with budgets, Claude Code read/recall and dry-run adapter states, prepared cards/inbox and summary leaves, read-only LCM peer DAG expansion and peer diagnostics, and approval-bound Codex scratch controls with post-action refresh proof where the release evidence records them. @@ -56,6 +57,11 @@ across the orchestrator surfaces that share the same local index. through `lco hermes smoke` plus `lco release hermes-readiness`. - A Hermes readiness pass is candidate evidence. It is not publication, an active profile install, live Eva runtime safety, or native-adapter proof. +- Daemon control is opt-in, local-only, and binds an expiring opaque target to + an already-running managed Codex daemon. It does not start or restart Codex, + enable Remote Control, expose raw identifiers, or own Desktop turns. +- Desktop-owned tasks remain on Hermes Computer Use and require window, + composer, and direction verification in the named Eva runtime canary. - Live Codex control is approval-gated by a dry-run audit id; the public demo stops at dry-run unless the user explicitly approves a target thread. - CUA Driver is the preferred fallback backend, but no no-focus behavior is claimed without local proof. - Peekaboo is a secondary macOS fallback for permission diagnostics and guarded snapshots; desktop action remains dry-run-only. @@ -102,7 +108,7 @@ blockers instead of allowing a working-app claim. Install stable releases through the `latest` dist-tag, public betas through the `beta` dist-tag, and release candidates through `next`. The stable channel -target for this package version is `1.6.1`; npm `latest` must move only after +target for this package version is `1.7.0`; npm `latest` must move only after the separate stable-promotion gate proves the exact candidate. Keep beta and other prereleases on prerelease tags. Do not publish a fake stable package just to move a dist-tag. diff --git a/docs/HERMES_ADAPTER_BOUNDARY.md b/docs/HERMES_ADAPTER_BOUNDARY.md index 5bcd111d..dc7511bd 100644 --- a/docs/HERMES_ADAPTER_BOUNDARY.md +++ b/docs/HERMES_ADAPTER_BOUNDARY.md @@ -9,7 +9,7 @@ index, safe-summary recall, and approval/audit patterns used by other MCP-capable harnesses. OpenClaw remains a supported compatibility surface. This document does not prove a native Hermes adapter, Hermes-specific indexing, -Hermes control surfaces, parity, GUI mutation, or cloud sync. +unrestricted control parity, generic GUI mutation, or cloud sync. ## What "Hermes support" means today @@ -27,6 +27,20 @@ mounts LCO exactly as any generic MCP client does: proof) with the same approval-gated dry-run/control boundaries Codex and OpenClaw get. +For Eva-owned CLI tasks, set `LCO_CODEX_TRANSPORT=daemon`. The MCP server then +connects to the already-running local managed Codex daemon over its Unix socket. +`lco_codex_control_route` returns an expiring opaque target and +`lco_codex_deliver` chooses idle send or matching active-turn steer. The public +result never returns raw thread or turn identifiers, transcript items, or the +socket path. LCO does not start or restart Codex, enable Remote Control, or +silently fall back to stdio when daemon mode is requested. + +Codex Desktop remains a separate ownership boundary. When LCO returns +`desktop_observation_required`, Hermes uses its existing `computer_use` +integration for window identification, visual verification, composer input, +and the Desktop-owned turn. LCO does not claim the managed daemon owns that +turn and does not enable generic `lco_desktop_act`. + This is the primary first-class supported path: the Hermes and generic-MCP mounting recipes are covered by [SETUP.md](SETUP.md). Release candidates exercise initialization, silent notifications, tool listing and calls, structured-result @@ -38,7 +52,8 @@ is not is a Hermes-native integration with Hermes-specific ergonomics. - No native Hermes adapter, plugin manifest, or Hermes-side install command. - No Hermes session indexing. LCO indexes Codex sessions; Hermes agents read that Codex index through LCO, they are not themselves an indexed source. -- No Hermes-specific control, GUI mutation, or remote-control surface. +- No native Hermes control protocol or generic GUI mutation. Hermes can invoke + the shared LCO daemon-control tools and its own Computer Use integration. - No claim that Hermes-specific auth, scopes, or lifecycle events are wired. - The generic-MCP protocol boundary applies: only `initialize`, `tools/list`, and `tools/call` are implemented; no MCP resources, prompts, or sampling. diff --git a/docs/SETUP.md b/docs/SETUP.md index ed5c2a7e..3800765d 100644 --- a/docs/SETUP.md +++ b/docs/SETUP.md @@ -345,12 +345,38 @@ mcp_servers: enabled: true env: LCO_TOOL_PROFILE: standard + LCO_CODEX_TRANSPORT: daemon ``` Omitting `LCO_DB_PATH` uses LCO's home-based default. If you set it explicitly, use an expanded absolute path because Hermes does not shell-expand `~` inside an environment-variable value. +`LCO_CODEX_TRANSPORT` defaults to `stdio`. Set it to `daemon` only when the +local managed Codex daemon is already running. LCO connects through the +standard socket under `CODEX_HOME`, or an explicit absolute +`LCO_CODEX_DAEMON_SOCKET`. It never starts or restarts Codex, never enables +Codex Remote Control, and never falls back from a requested daemon to stdio. +`LCO_CODEX_APP_SERVER_ARGS` applies only to stdio mode. + +The remote-control loop is: + +1. Call `lco_codex_control_route`, optionally with a title hint. +2. If it selects `app_server`, call `lco_codex_deliver` in its default dry-run + mode and show the approval packet. +3. Repeat the exact call with `dry_run:false` and the matching + `approval_audit_id`. LCO sends to an idle task or steers the matching active + turn after revalidating ownership and state. +4. Use `lco_codex_interrupt_thread` with the same opaque `target_ref` for an + approved interrupt. +5. If the route is `desktop_observation_required`, use Hermes + `computer_use` to verify and operate the Codex Desktop window. Do not assume + the managed daemon owns a Desktop task. + +For Telegram, the remote instruction must originate from the user and arrive +at Eva. Eva returns the result to that same conversation. Typing a Telegram +message as Eva is not a valid inbound-control test. + `lco_find` queries the existing index by default. It does not synchronously reindex unless the call explicitly passes `index:true`; `lco_index_sessions` remains the explicit refresh tool. @@ -384,7 +410,7 @@ lco release hermes-readiness \ ``` The smoke checks initialization, notification silence, Eva's required -14-tool registration set, object-valid structured results, default search +16-tool registration set, object-valid structured results, default search behavior, and bounded latency. The readiness report binds that smoke to the candidate package probe. Both reports are public-safe candidate evidence only: they do not prove publication, an active profile install, or live Eva runtime @@ -565,10 +591,13 @@ The safe loop is: 4. `lco_codex_extract` with `kind: "plans"`, `kind: "final_messages"`, and `kind: "touched_files"` 5. `lco_expand_session` or `lco_expand_query` -6. `lco_codex_control_dry_run` only when action is needed -7. `lco_codex_start_thread` only after dry-run approval when a new Codex thread +6. `lco_codex_control_route` for the current daemon-owned task +7. `lco_codex_deliver` in dry-run mode, then live with the exact matching + approval id +8. `lco_codex_control_dry_run` for lower-level explicit actions +9. `lco_codex_start_thread` only after dry-run approval when a new Codex thread is needed -8. live action only with a matching `approval_audit_id` +10. live action only with a matching `approval_audit_id` Live start/send/steer/interrupt results distinguish `accepted_by_transport`, `started`, `completed`, `persisted`, and `unverified_pending`. If a result is diff --git a/docs/releases/CHANGELOG.md b/docs/releases/CHANGELOG.md index d46bb36b..e81e92a3 100644 --- a/docs/releases/CHANGELOG.md +++ b/docs/releases/CHANGELOG.md @@ -4,8 +4,9 @@ One line per released version, newest first. Each line links to the full release notes for that version. Draft notes stay outside the released list until npm, tag, and GitHub Release truth agree. -Draft candidate notes: [1.6.1](RELEASE_NOTES_1.6.1.md) +Draft candidate notes: [1.7.0](RELEASE_NOTES_1.7.0.md) +- [1.6.1](RELEASE_NOTES_1.6.1.md) - [1.6.0](RELEASE_NOTES_1.6.0.md) - [1.5.0](RELEASE_NOTES_1.5.0.md) - [1.4.5](RELEASE_NOTES_1.4.5.md) diff --git a/docs/releases/RELEASE_NOTES_1.7.0.md b/docs/releases/RELEASE_NOTES_1.7.0.md new file mode 100644 index 00000000..cc82cb0b --- /dev/null +++ b/docs/releases/RELEASE_NOTES_1.7.0.md @@ -0,0 +1,53 @@ +# Release Notes 1.7.0 + +`1.7.0` is the Eva remote control candidate for local Codex sessions. These +notes describe the candidate; npm `latest` and GitHub Releases remain the +publication authorities. + +## Highlights + +- Opt-in `LCO_CODEX_TRANSPORT=daemon` support for WebSocket JSON-RPC over the + already-running local Codex managed daemon Unix socket. +- `lco_codex_control_route`, which selects one daemon-owned active or idle task + and returns an expiring opaque target, or explicitly requires Desktop + observation. +- `lco_codex_deliver`, which uses the existing approval audit to send to an + idle task or steer the matching active turn after revalidating ownership, + state, and turn identity. +- Opaque-target interrupt support on `lco_codex_interrupt_thread` while keeping + the existing raw-ID compatibility form. +- Hermes smoke registration coverage for the two new first-class tools. + +## Upgrade + +```bash +npm install -g lossless-codex-orchestrator@latest +lco doctor +``` + +Existing installations keep stdio behavior unless +`LCO_CODEX_TRANSPORT=daemon` is set explicitly. + +## Validation + +The release checks cover Unix-socket reconnects, opaque target selection, +idle-send and active-steer delivery, stale-target rejection, notification +silence, structured MCP results, and package installation. + +## Safety boundary + +- Stdio remains the public compatibility default. +- Daemon mode never starts or restarts Codex, enables Remote Control, or falls + back silently to stdio. +- Raw transcript items may be transiently projected by Codex only to identify + the current active turn; LCO immediately discards the items and never returns + or logs them. +- Codex Desktop tasks stay on Hermes Computer Use. The managed daemon is not + treated as the owner of Desktop turns, and generic GUI control stays off. +- Every live delivery or interrupt still requires the matching unexpired + dry-run approval audit. + +## Availability + +Until npm `latest` and GitHub Releases list 1.7.0, this file describes upcoming +behavior rather than an available stable package. diff --git a/openclaw.plugin.json b/openclaw.plugin.json index 28be78b4..0f3f8b1b 100644 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -2,7 +2,7 @@ "id": "lossless-openclaw-orchestrator", "name": "Lossless OpenClaw Orchestrator", "description": "Index, search, and prepare local Codex sessions for OpenClaw with approval-gated dry-runs and optional Codex controls.", - "version": "1.6.1", + "version": "1.7.0", "kind": "tool", "tools": { "prefix": "lco_" @@ -53,6 +53,8 @@ "lco_codex_sqlite_stores", "lco_lcm_peer_dbs", "lco_drive", + "lco_codex_control_route", + "lco_codex_deliver", "lco_codex_control_dry_run", "lco_codex_start_thread", "lco_codex_resume_thread", @@ -90,6 +92,8 @@ "loo_codex_sqlite_stores", "loo_lcm_peer_dbs", "loo_drive", + "loo_codex_control_route", + "loo_codex_deliver", "loo_codex_control_dry_run", "loo_codex_start_thread", "loo_codex_resume_thread", @@ -1323,9 +1327,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 8, - "operatorPathRole": "Create a bounded review-then-drive plan and real target-adapter dry-run audit packet." + "tier": "workflow_detail" }, "inputSchema": { "type": "object", @@ -1385,6 +1387,77 @@ } } }, + { + "name": "lco_codex_control_route", + "description": "Resolve one daemon-owned Codex task to an expiring opaque target, or require Desktop observation without mutation.", + "safety": { + "mode": "read_only", + "source": "codex_direct", + "requiresApproval": false, + "mutationClasses": [] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 8, + "operatorPathRole": "Resolve one daemon-owned Codex target or return the explicit Desktop observation route without exposing raw identifiers." + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "hint": { + "type": "string", + "maxLength": 240 + } + } + } + }, + { + "name": "lco_codex_deliver", + "description": "Deliver to an opaque Codex target: start a turn when idle or steer the matching active turn. Dry-run is the default.", + "safety": { + "mode": "approval_gated_control", + "source": "codex_direct", + "requiresApproval": true, + "mutationClasses": [ + "derived_cache", + "live_control" + ] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 9, + "operatorPathRole": "Deliver through the opaque target using idle send or active steer under the existing dry-run approval audit." + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "target_ref": { + "type": "string" + }, + "message": { + "type": "string" + }, + "turn_wait_ms": { + "type": "integer", + "minimum": 1, + "maximum": 600000 + }, + "dry_run": { + "type": "boolean", + "default": true + }, + "approval_audit_id": { + "type": "string" + } + }, + "required": [ + "target_ref", + "message" + ] + } + }, { "name": "lco_codex_control_dry_run", "description": "Create a dry-run audit id for a Codex control action under LCO's fixed never-approve, read-only runtime posture.", @@ -1467,9 +1540,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 9, - "operatorPathRole": "Run the approved resume action only after a matching dry-run audit id." + "tier": "workflow_detail" }, "inputSchema": { "type": "object", @@ -1578,7 +1649,7 @@ }, { "name": "lco_codex_interrupt_thread", - "description": "Rejoin and interrupt a Codex thread under LCO's fixed never-approve, read-only posture. Live mode requires approval_audit_id and expected_turn_id.", + "description": "Interrupt a matching opaque target or rejoin and interrupt a raw Codex thread under LCO's fixed never-approve, read-only posture.", "safety": { "mode": "approval_gated_control", "source": "codex_direct", @@ -1612,11 +1683,23 @@ }, "approval_audit_id": { "type": "string" + }, + "target_ref": { + "type": "string" } }, - "required": [ - "thread_id", - "expected_turn_id" + "anyOf": [ + { + "required": [ + "target_ref" + ] + }, + { + "required": [ + "thread_id", + "expected_turn_id" + ] + } ] } }, @@ -3167,9 +3250,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 8, - "operatorPathRole": "Create a bounded review-then-drive plan and real target-adapter dry-run audit packet.", + "tier": "workflow_detail", "aliasOf": "lco_drive" }, "inputSchema": { @@ -3230,6 +3311,79 @@ } } }, + { + "name": "loo_codex_control_route", + "description": "Resolve one daemon-owned Codex task to an expiring opaque target, or require Desktop observation without mutation.", + "safety": { + "mode": "read_only", + "source": "codex_direct", + "requiresApproval": false, + "mutationClasses": [] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 8, + "operatorPathRole": "Resolve one daemon-owned Codex target or return the explicit Desktop observation route without exposing raw identifiers.", + "aliasOf": "lco_codex_control_route" + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "hint": { + "type": "string", + "maxLength": 240 + } + } + } + }, + { + "name": "loo_codex_deliver", + "description": "Deliver to an opaque Codex target: start a turn when idle or steer the matching active turn. Dry-run is the default.", + "safety": { + "mode": "approval_gated_control", + "source": "codex_direct", + "requiresApproval": true, + "mutationClasses": [ + "derived_cache", + "live_control" + ] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 9, + "operatorPathRole": "Deliver through the opaque target using idle send or active steer under the existing dry-run approval audit.", + "aliasOf": "lco_codex_deliver" + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "target_ref": { + "type": "string" + }, + "message": { + "type": "string" + }, + "turn_wait_ms": { + "type": "integer", + "minimum": 1, + "maximum": 600000 + }, + "dry_run": { + "type": "boolean", + "default": true + }, + "approval_audit_id": { + "type": "string" + } + }, + "required": [ + "target_ref", + "message" + ] + } + }, { "name": "loo_codex_control_dry_run", "description": "Create a dry-run audit id for a Codex control action under LCO's fixed never-approve, read-only runtime posture.", @@ -3314,9 +3468,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 9, - "operatorPathRole": "Run the approved resume action only after a matching dry-run audit id.", + "tier": "workflow_detail", "aliasOf": "lco_codex_resume_thread" }, "inputSchema": { @@ -3428,7 +3580,7 @@ }, { "name": "loo_codex_interrupt_thread", - "description": "Rejoin and interrupt a Codex thread under LCO's fixed never-approve, read-only posture. Live mode requires approval_audit_id and expected_turn_id.", + "description": "Interrupt a matching opaque target or rejoin and interrupt a raw Codex thread under LCO's fixed never-approve, read-only posture.", "safety": { "mode": "approval_gated_control", "source": "codex_direct", @@ -3463,11 +3615,23 @@ }, "approval_audit_id": { "type": "string" + }, + "target_ref": { + "type": "string" } }, - "required": [ - "thread_id", - "expected_turn_id" + "anyOf": [ + { + "required": [ + "target_ref" + ] + }, + { + "required": [ + "thread_id", + "expected_turn_id" + ] + } ] } }, @@ -6911,8 +7075,8 @@ "lco_recent_sessions", "lco_attention_inbox", "lco_project_digest", - "lco_drive", - "lco_codex_resume_thread" + "lco_codex_control_route", + "lco_codex_deliver" ], "namingPolicy": { "publicProductAbbreviation": "LCO", diff --git a/package-lock.json b/package-lock.json index 017ed766..5460f223 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,13 +1,16 @@ { "name": "lossless-openclaw-orchestrator", - "version": "1.6.1", + "version": "1.7.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "lossless-openclaw-orchestrator", - "version": "1.6.1", + "version": "1.7.0", "license": "PolyForm-Noncommercial-1.0.0", + "dependencies": { + "ws": "^8.21.1" + }, "bin": { "lco": "dist/packages/cli/src/index.js", "lco-mcp-server": "dist/packages/mcp-server/src/server.js", @@ -16,6 +19,7 @@ }, "devDependencies": { "@types/node": "^26.0.0", + "@types/ws": "^8.18.1", "tsx": "^4.20.6", "typescript": "^5.9.3" }, @@ -475,6 +479,16 @@ "undici-types": "~8.3.0" } }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/esbuild": { "version": "0.28.1", "resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.28.1.tgz", @@ -571,6 +585,27 @@ "integrity": "sha512-j375ScV60dom+YkPFIfTLcOiPxkN/buHz5GobjLhixFuANaNs3C9l4GmrWqejgXWJ7BbJcFYpTEUkS1Ge8bpZQ==", "dev": true, "license": "MIT" + }, + "node_modules/ws": { + "version": "8.21.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz", + "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } } } } diff --git a/package.json b/package.json index 602f2dbb..1086e7e2 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "lossless-openclaw-orchestrator", - "version": "1.6.1", + "version": "1.7.0", "description": "Index, search, and prepare local Codex sessions for Hermes and MCP clients, with OpenClaw compatibility, approval-gated dry-runs, and optional Codex controls.", "type": "module", "license": "PolyForm-Noncommercial-1.0.0", @@ -75,7 +75,11 @@ ], "devDependencies": { "@types/node": "^26.0.0", + "@types/ws": "^8.18.1", "tsx": "^4.20.6", "typescript": "^5.9.3" + }, + "dependencies": { + "ws": "^8.21.1" } } diff --git a/packages/adapters/src/codex-jsonrpc.ts b/packages/adapters/src/codex-jsonrpc.ts index 1ef6f581..878a451c 100644 --- a/packages/adapters/src/codex-jsonrpc.ts +++ b/packages/adapters/src/codex-jsonrpc.ts @@ -1,6 +1,9 @@ import { spawn, spawnSync, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { createConnection } from "node:net"; +import { isAbsolute, join } from "node:path"; import { createInterface, type Interface } from "node:readline"; import { URL } from "node:url"; +import NodeWebSocket from "ws"; import { assertCodexMethodAllowed, type CodexMethodSurface } from "./policy.js"; import { redactDiagnosticString, redactValue } from "./redaction.js"; @@ -8,6 +11,7 @@ export type JsonRpcTransport = { sendJson(payload: unknown): void | Promise; readLine(deadline: number): string | null | Promise; close(): void | Promise; + isClosed?(): boolean; }; export type CodexJsonRpcResponse = { @@ -152,6 +156,10 @@ export class CodexJsonRpcClient { await transport.close(); } + isTransportClosed(): boolean { + return !this.transport || this.transport.isClosed?.() === true; + } + private async sendNotification(method: string): Promise { this.requireTransport().sendJson({ method }); } @@ -362,6 +370,103 @@ export class LoopbackWebSocketTransport implements JsonRpcTransport { } } +export class UnixSocketWebSocketTransport implements JsonRpcTransport { + private readonly socket: NodeWebSocket; + private readonly lines: string[] = []; + private readonly waiters: Array<(line: string | null) => void> = []; + private closed = false; + private readonly ready: Promise; + private connectTimer: ReturnType | undefined; + + constructor(socketPath: string, private readonly timeoutMs = DEFAULT_TIMEOUT_MS) { + if (!isAbsolute(socketPath)) { + throw new Error("Codex daemon transport requires an absolute Unix socket path"); + } + this.socket = new NodeWebSocket("ws://localhost/", { + perMessageDeflate: false, + createConnection: () => createConnection(socketPath) + }); + this.ready = new Promise((resolve, reject) => { + this.connectTimer = setTimeout(() => { + this.close(); + reject(new Error("Codex daemon WebSocket connect timed out")); + }, this.timeoutMs); + this.socket.once("open", () => { + this.clearConnectTimer(); + resolve(); + }); + this.socket.once("error", () => { + this.clearConnectTimer(); + reject(new Error("Codex daemon WebSocket connection failed")); + }); + }); + void this.ready.catch(() => undefined); + this.socket.on("message", (data) => this.pushLine(data.toString())); + this.socket.on("close", () => this.finishOutput()); + } + + async sendJson(payload: unknown): Promise { + await this.ready; + this.socket.send(JSON.stringify(payload)); + } + + readLine(deadline: number): Promise { + const existing = this.lines.shift(); + if (existing !== undefined) return Promise.resolve(existing); + if (this.closed) return Promise.resolve(null); + const remaining = Math.max(1, Math.min(this.timeoutMs, deadline - Date.now())); + return new Promise((resolve) => { + let timer: ReturnType; + const waiter = (line: string | null) => { + clearTimeout(timer); + resolve(line); + }; + timer = setTimeout(() => { + const index = this.waiters.indexOf(waiter); + if (index >= 0) this.waiters.splice(index, 1); + resolve(null); + }, remaining); + this.waiters.push(waiter); + }); + } + + close(): void { + this.clearConnectTimer(); + this.finishOutput(); + if (this.socket.readyState === NodeWebSocket.CONNECTING) { + this.socket.terminate(); + } else if (this.socket.readyState === NodeWebSocket.OPEN) { + this.socket.close(); + } + } + + isClosed(): boolean { + return this.closed; + } + + private clearConnectTimer(): void { + if (this.connectTimer === undefined) return; + clearTimeout(this.connectTimer); + this.connectTimer = undefined; + } + + private pushLine(line: string): void { + const waiter = this.waiters.shift(); + if (waiter) waiter(line); + else this.lines.push(line); + } + + private finishOutput(): void { + if (this.closed) return; + this.closed = true; + let waiter = this.waiters.shift(); + while (waiter) { + waiter(null); + waiter = this.waiters.shift(); + } + } +} + export function createCodexMcpStdioClient(options: { command?: string; args?: string[]; @@ -382,32 +487,72 @@ export function createCodexAppServerWebSocketClient(options: { return createCodexClientFromTransport(() => new LoopbackWebSocketTransport(options.url, options.timeoutMs), options); } +export function createCodexAppServerDaemonClient(options: { + socketPath: string; + timeoutMs?: number; + surface?: CodexMethodSurface; +}) { + return createCodexClientFromTransport( + () => new UnixSocketWebSocketTransport(options.socketPath, options.timeoutMs), + { ...options, persistent: true } + ); +} + +export function resolveCodexDaemonSocketPath(codexHome: string): string { + if (!isAbsolute(codexHome)) throw new Error("CODEX_HOME must be absolute for daemon transport"); + return join(codexHome, "app-server-control", "app-server-control.sock"); +} + function createCodexClientFromTransport( transportFactory: () => JsonRpcTransport, - options: { timeoutMs?: number; surface?: CodexMethodSurface } + options: { timeoutMs?: number; surface?: CodexMethodSurface; persistent?: boolean } ) { - return { - async request(method: string, params: Record) { + let persistentClient: CodexJsonRpcClient | null = null; + let operationQueue: Promise = Promise.resolve(); + + async function withClient(operation: (client: CodexJsonRpcClient) => Promise): Promise { + if (!options.persistent) { const client = new CodexJsonRpcClient( transportFactory, { timeoutMs: options.timeoutMs, surface: options.surface ?? "control" } ); try { await client.connect(); - return await client.request(method, params); + return await operation(client); } finally { await client.close(); } + } + + const queued = operationQueue.then(async () => { + if (!persistentClient || persistentClient.isTransportClosed()) { + if (persistentClient) await persistentClient.close(); + persistentClient = new CodexJsonRpcClient( + transportFactory, + { timeoutMs: options.timeoutMs, surface: options.surface ?? "control" } + ); + await persistentClient.connect(); + } + try { + return await operation(persistentClient); + } catch (error) { + await persistentClient.close(); + persistentClient = null; + throw error; + } + }); + operationQueue = queued.then(() => undefined, () => undefined); + return queued; + } + + return { + async request(method: string, params: Record) { + return withClient((client) => client.request(method, params)); }, async requestSequence(steps: Array<{ method: string; params: Record }>) { const surface = options.surface ?? "control"; for (const step of steps) assertCodexMethodAllowed(step.method, surface); - const client = new CodexJsonRpcClient( - transportFactory, - { timeoutMs: options.timeoutMs, surface } - ); - try { - await client.connect(); + return withClient(async (client) => { const responses: CodexJsonRpcResponse[] = []; for (const step of steps) { const response = await client.request(step.method, step.params); @@ -415,9 +560,7 @@ function createCodexClientFromTransport( if (!response.ok) break; } return responses; - } finally { - await client.close(); - } + }); }, async requestSequenceUntilTurnResolved( steps: Array<{ method: string; params: Record }>, @@ -431,12 +574,7 @@ function createCodexClientFromTransport( if (turnOptions.requireSafeActiveRuntime && steps.some((step) => step.params.threadId !== turnOptions.threadId)) { throw new Error("Codex safe active-runtime proof requires every sequence step to target the requested thread"); } - const client = new CodexJsonRpcClient( - transportFactory, - { timeoutMs: options.timeoutMs, surface } - ); - try { - await client.connect(); + return withClient(async (client) => { const responses: CodexJsonRpcResponse[] = []; let turnId = turnOptions.expectedTurnId; let latestStatus: string | null = null; @@ -470,9 +608,7 @@ function createCodexClientFromTransport( timeoutMs: turnOptions.turnWaitMs }); return { responses, turn }; - } finally { - await client.close(); - } + }); } }; } diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index 8314f2d6..defc103d 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -955,7 +955,7 @@ export function createCodexControl(options: { audit: ControlAuditStore; client: createdThreadFromResponse: true }); }, - sendMessage(input: { threadId: string; message: string; dryRun?: boolean; approvalAuditId?: string; turnWaitMs?: number }) { + sendMessage(input: { threadId: string; message: string; dryRun?: boolean; approvalAuditId?: string; turnWaitMs?: number; loadedThread?: boolean; awaitTurn?: boolean }) { const resumeParams = safeCodexResumeParams(input.threadId); // A new turn pins its own restrictive posture. Unlike steer/interrupt, // it does not act inside an already-running turn whose posture is fixed. @@ -973,14 +973,18 @@ export function createCodexControl(options: { audit: ControlAuditStore; client: dryRun: input.dryRun, approvalAuditId: input.approvalAuditId, params: turnStartParams, - steps: [ - { method: "thread/resume", params: resumeParams }, - { method: "turn/start", params: turnStartParams } - ], + steps: input.loadedThread + ? [{ method: "turn/start", params: turnStartParams }] + : [ + { method: "thread/resume", params: resumeParams }, + { method: "turn/start", params: turnStartParams } + ], loadedThreadReusable: true, - turnResolution: { - turnWaitMs: input.turnWaitMs - } + turnResolution: input.awaitTurn === false + ? undefined + : { + turnWaitMs: input.turnWaitMs + } }); }, resumeThread(input: { threadId: string; dryRun?: boolean; approvalAuditId?: string }) { @@ -1320,6 +1324,9 @@ function assertCodexControlSequenceResponses(responses: unknown[], steps: CodexC const isLcoSafetyBlock = response?.code === "safe_runtime_posture_unproven" && response.origin === "lco_safety_gate"; if (response?.ok === false && !isLcoSafetyBlock) { + if (JSON.stringify(response).includes("activeTurnNotSteerable")) { + throw new Error("active_turn_not_steerable"); + } throw new Error(`Codex control sequence step failed: ${steps[index]?.method ?? "unknown"}`); } } diff --git a/packages/adapters/src/policy.ts b/packages/adapters/src/policy.ts index 540ade7f..c00c9ea2 100644 --- a/packages/adapters/src/policy.ts +++ b/packages/adapters/src/policy.ts @@ -177,6 +177,8 @@ export const LOO_COMMAND_POLICY: Record = { lco_codex_sqlite_stores: readOnly("local_index"), lco_lcm_peer_dbs: readOnly("local_index"), lco_drive: localCacheWrite("audit"), + lco_codex_control_route: readOnly("codex_direct"), + lco_codex_deliver: approvalGatedControl("codex_direct"), lco_codex_control_dry_run: localCacheWrite("audit"), lco_codex_start_thread: approvalGatedControl("codex_direct"), lco_codex_resume_thread: approvalGatedControl("codex_direct"), diff --git a/packages/cli/src/hermes-smoke.ts b/packages/cli/src/hermes-smoke.ts index 8b8d952e..773eb67f 100644 --- a/packages/cli/src/hermes-smoke.ts +++ b/packages/cli/src/hermes-smoke.ts @@ -15,6 +15,8 @@ export const EVA_HERMES_REQUIRED_LCO_TOOLS = [ "lco_attention_inbox", "lco_project_digest", "lco_codex_extract", + "lco_codex_control_route", + "lco_codex_deliver", "lco_codex_control_dry_run", "lco_codex_start_thread", "lco_codex_resume_thread", diff --git a/packages/mcp-server/src/codex-control-router.ts b/packages/mcp-server/src/codex-control-router.ts new file mode 100644 index 00000000..e425132b --- /dev/null +++ b/packages/mcp-server/src/codex-control-router.ts @@ -0,0 +1,344 @@ +import { randomUUID } from "node:crypto"; + +import { + CODEX_CONTROL_DRY_RUN_TTL_MS, + redactDiagnosticString, + type CodexClient, + type createCodexControl +} from "../../adapters/src/index.js"; + +type CodexControl = ReturnType; +type RouteState = "active" | "idle"; + +type TargetProjection = { + threadId: string; + title: string | null; + state: RouteState; + turnId?: string; +}; + +type StoredTarget = TargetProjection & { + ref: string; + expiresAtMs: number; +}; + +export type CodexControlRouteResult = { + schema: "lco.codex.controlRoute.v1"; + status: "selected" | "ambiguous" | "none" | "unavailable"; + route: "app_server" | "desktop_observation_required" | "unknown"; + target_ref: string | null; + title_sanitized: string | null; + state: RouteState | null; + supported_actions: Array<"send" | "steer" | "interrupt">; + expires_at: string | null; + reason_codes: string[]; + public_safe: true; + raw_transcript_returned: false; +}; + +export type CodexDeliveryResult = { + schema: "lco.codex.delivery.v1"; + status: "dry_run_ready" | "completed" | "accepted" | "blocked"; + action: "send" | "steer" | "interrupt" | null; + target_ref: string; + live: boolean; + control_sent: boolean; + approval_audit_id?: string; + params_hash?: string; + message_hash?: string; + reason_codes: string[]; + public_safe: true; + raw_transcript_returned: false; +}; + +export function createCodexControlRouter(options: { + client: CodexClient; + control: CodexControl; + ttlMs?: number; + createRef?: () => string; + now?: () => Date; +}) { + const targets = new Map(); + const ttlMs = options.ttlMs ?? Math.min(5 * 60_000, CODEX_CONTROL_DRY_RUN_TTL_MS); + const now = options.now ?? (() => new Date()); + const createRef = options.createRef ?? (() => `lco_target_${randomUUID().replaceAll("-", "")}`); + + async function route(input: { hint?: string }): Promise { + let projections: TargetProjection[]; + try { + projections = await loadDaemonTargets(options.client); + } catch { + return routeResult("unavailable", "unknown", null, ["daemon_unavailable"]); + } + + const hint = input.hint?.trim(); + if (hint) { + const normalizedHint = normalizeHint(hint); + const matches = projections.filter((target) => + target.threadId === normalizedHint + || target.title?.toLocaleLowerCase() === hint.toLocaleLowerCase() + ); + if (matches.length === 1) return selectedRoute(matches[0]!); + if (matches.length > 1) return routeResult("ambiguous", "app_server", null, ["explicit_hint_ambiguous"]); + return routeResult("unavailable", "desktop_observation_required", null, ["explicit_hint_not_daemon_loaded"]); + } + + const active = projections.filter((target) => target.state === "active"); + if (active.length === 1) return selectedRoute(active[0]!); + if (active.length > 1) return routeResult("ambiguous", "app_server", null, ["multiple_active_daemon_targets"]); + const idle = projections.filter((target) => target.state === "idle"); + if (idle.length === 1) return selectedRoute(idle[0]!); + if (idle.length > 1) return routeResult("ambiguous", "app_server", null, ["multiple_idle_daemon_targets"]); + return routeResult("none", "unknown", null, ["no_daemon_target"]); + } + + async function deliver(input: { + targetRef: string; + message: string; + dryRun?: boolean; + approvalAuditId?: string; + turnWaitMs?: number; + }): Promise { + const validation = await validateTarget(input.targetRef); + if (!validation.ok) return blockedDelivery(input.targetRef, null, input.dryRun === false, validation.reason); + const target = validation.target; + if (target.state === "active" && !target.turnId) { + return blockedDelivery(input.targetRef, null, input.dryRun === false, "active_turn_id_unavailable"); + } + const action = target.state === "active" ? "steer" : "send"; + try { + const result = action === "steer" + ? await options.control.steerThread({ + threadId: target.threadId, + message: input.message, + expectedTurnId: target.turnId, + dryRun: input.dryRun, + approvalAuditId: input.approvalAuditId, + turnWaitMs: input.turnWaitMs + }) + : await options.control.sendMessage({ + threadId: target.threadId, + message: input.message, + loadedThread: true, + awaitTurn: false, + dryRun: input.dryRun, + approvalAuditId: input.approvalAuditId, + turnWaitMs: input.turnWaitMs + }); + return publicDeliveryResult(input.targetRef, action, result); + } catch (error) { + const reason = String(error).includes("active_turn_not_steerable") + ? "active_turn_not_steerable" + : "approval_or_control_rejected"; + return blockedDelivery(input.targetRef, action, input.dryRun === false, reason); + } + } + + async function interrupt(input: { + targetRef: string; + dryRun?: boolean; + approvalAuditId?: string; + turnWaitMs?: number; + }): Promise { + const validation = await validateTarget(input.targetRef); + if (!validation.ok) return blockedDelivery(input.targetRef, "interrupt", input.dryRun === false, validation.reason); + const target = validation.target; + if (target.state !== "active" || !target.turnId) { + return blockedDelivery(input.targetRef, "interrupt", input.dryRun === false, "target_not_interruptible"); + } + try { + const result = await options.control.interruptThread({ + threadId: target.threadId, + expectedTurnId: target.turnId, + dryRun: input.dryRun, + approvalAuditId: input.approvalAuditId, + turnWaitMs: input.turnWaitMs + }); + return publicDeliveryResult(input.targetRef, "interrupt", result); + } catch { + return blockedDelivery(input.targetRef, "interrupt", input.dryRun === false, "approval_or_control_rejected"); + } + } + + function selectedRoute(target: TargetProjection): CodexControlRouteResult { + const createdAt = now().getTime(); + const ref = createRef(); + const stored: StoredTarget = { + ...target, + ref, + expiresAtMs: createdAt + ttlMs + }; + targets.set(ref, stored); + return routeResult("selected", "app_server", stored, target.state === "active" && !target.turnId + ? ["active_turn_id_unavailable"] + : []); + } + + async function validateTarget(ref: string): Promise< + { ok: true; target: StoredTarget } + | { ok: false; reason: string } + > { + const target = targets.get(ref); + if (!target) return { ok: false, reason: "target_ref_unknown" }; + if (target.expiresAtMs <= now().getTime()) { + targets.delete(ref); + return { ok: false, reason: "target_ref_expired" }; + } + try { + const loaded = await loadedThreadIds(options.client); + if (!loaded.includes(target.threadId)) return { ok: false, reason: "target_ownership_changed" }; + const current = await readTargetProjection(options.client, target.threadId); + if (!current) return { ok: false, reason: "target_unavailable" }; + if (current.state !== target.state) return { ok: false, reason: "target_state_changed" }; + if (target.state === "active" && current.turnId !== target.turnId) { + return { ok: false, reason: "target_turn_changed" }; + } + return { ok: true, target }; + } catch { + return { ok: false, reason: "daemon_unavailable" }; + } + } + + return { route, deliver, interrupt }; +} + +async function loadDaemonTargets(client: CodexClient): Promise { + const ids = await loadedThreadIds(client); + const projections: TargetProjection[] = []; + for (const id of ids) { + const projection = await readTargetProjection(client, id); + if (projection) projections.push(projection); + } + return projections; +} + +async function loadedThreadIds(client: CodexClient): Promise { + const response = responseResult(await client.request("thread/loaded/list", {})); + const data = recordValue(response)?.data; + if (!Array.isArray(data)) throw new Error("thread/loaded/list unavailable"); + return data.filter((value): value is string => typeof value === "string" && value.length > 0); +} + +async function readTargetProjection(client: CodexClient, threadId: string): Promise { + const metadataResult = responseResult(await client.request("thread/read", { threadId, includeTurns: false })); + let thread = recordValue(recordValue(metadataResult)?.thread); + if (!thread) return null; + const status = recordValue(thread.status); + const statusType = typeof status?.type === "string" + ? status.type + : typeof thread.status === "string" + ? thread.status + : null; + if (statusType !== "active" && statusType !== "idle") return null; + if (statusType === "active") { + const activeResult = responseResult(await client.request("thread/read", { threadId, includeTurns: true })); + thread = recordValue(recordValue(activeResult)?.thread); + if (!thread) return null; + } + const turns = Array.isArray(thread.turns) ? thread.turns : []; + let turnId: string | undefined; + for (let index = turns.length - 1; index >= 0; index -= 1) { + const turn = recordValue(turns[index]); + if (turn?.status === "inProgress" && typeof turn.id === "string" && turn.id) { + turnId = turn.id; + break; + } + } + return { + threadId, + title: sanitizeTitle(thread.name), + state: statusType, + ...(turnId ? { turnId } : {}) + }; +} + +function responseResult(value: unknown): unknown { + const response = recordValue(value); + if (!response) return value; + if (response.ok === false) throw new Error("Codex app-server request failed"); + return "result" in response ? response.result : value; +} + +function routeResult( + status: CodexControlRouteResult["status"], + route: CodexControlRouteResult["route"], + target: StoredTarget | null, + reasonCodes: string[] +): CodexControlRouteResult { + return { + schema: "lco.codex.controlRoute.v1", + status, + route, + target_ref: target?.ref ?? null, + title_sanitized: target?.title ?? null, + state: target?.state ?? null, + supported_actions: target + ? target.state === "active" + ? target.turnId ? ["steer", "interrupt"] : [] + : ["send"] + : [], + expires_at: target ? new Date(target.expiresAtMs).toISOString() : null, + reason_codes: reasonCodes, + public_safe: true, + raw_transcript_returned: false + }; +} + +function publicDeliveryResult( + targetRef: string, + action: "send" | "steer" | "interrupt", + rawResult: unknown +): CodexDeliveryResult { + const result = recordValue(rawResult) ?? {}; + const live = result.live === true; + const proof = recordValue(result.proofState); + const completed = proof?.completed === true || proof?.status === "completed"; + return { + schema: "lco.codex.delivery.v1", + status: live ? completed ? "completed" : "accepted" : "dry_run_ready", + action, + target_ref: targetRef, + live, + control_sent: result.controlSent === true, + ...(typeof result.approvalAuditId === "string" ? { approval_audit_id: result.approvalAuditId } : {}), + ...(typeof result.paramsHash === "string" ? { params_hash: result.paramsHash } : {}), + ...(typeof result.messageHash === "string" ? { message_hash: result.messageHash } : {}), + reason_codes: [], + public_safe: true, + raw_transcript_returned: false + }; +} + +function blockedDelivery( + targetRef: string, + action: CodexDeliveryResult["action"], + live: boolean, + reason: string +): CodexDeliveryResult { + return { + schema: "lco.codex.delivery.v1", + status: "blocked", + action, + target_ref: targetRef, + live, + control_sent: false, + reason_codes: [reason], + public_safe: true, + raw_transcript_returned: false + }; +} + +function normalizeHint(value: string): string { + return value.startsWith("codex_thread:") ? value.slice("codex_thread:".length) : value; +} + +function sanitizeTitle(value: unknown): string | null { + if (typeof value !== "string" || !value.trim()) return null; + return redactDiagnosticString(value).replace(/\s+/g, " ").trim().slice(0, 160); +} + +function recordValue(value: unknown): Record | null { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : null; +} diff --git a/packages/mcp-server/src/codex-runtime-transport.ts b/packages/mcp-server/src/codex-runtime-transport.ts new file mode 100644 index 00000000..35b50b93 --- /dev/null +++ b/packages/mcp-server/src/codex-runtime-transport.ts @@ -0,0 +1,73 @@ +import { isAbsolute, join } from "node:path"; + +import { + assertCodexMethodAllowed, + createCodexAppServerDaemonClient, + createCodexAppServerStdioClient, + resolveCodexDaemonSocketPath, + type CodexClient +} from "../../adapters/src/index.js"; +import { readEnv, readEnvWithFallback, resolveHomeDir, type LcoEnv } from "../../runtime/src/env.js"; + +export type CodexRuntimeTransportConfig = + | { mode: "stdio"; command: string; args: string[] } + | { mode: "daemon"; socketPath: string }; + +export function resolveCodexRuntimeTransportConfig(env: LcoEnv = process.env): CodexRuntimeTransportConfig { + const mode = readEnv("CODEX_TRANSPORT", env) ?? "stdio"; + if (mode === "stdio") { + return { + mode, + command: readEnvWithFallback("CODEX_BIN", "codex", env), + args: (readEnv("CODEX_APP_SERVER_ARGS", env) || "app-server --stdio").split(/\s+/).filter(Boolean) + }; + } + if (mode !== "daemon") { + throw new Error("LCO_CODEX_TRANSPORT must be stdio or daemon"); + } + + const override = readEnv("CODEX_DAEMON_SOCKET", env); + if (override && !isAbsolute(override)) { + throw new Error("LCO_CODEX_DAEMON_SOCKET must be an absolute Unix socket path"); + } + const codexHome = env.CODEX_HOME?.trim() || join(resolveHomeDir(env), ".codex"); + return { + mode, + socketPath: override ?? resolveCodexDaemonSocketPath(codexHome) + }; +} + +export function createConfiguredCodexClients( + env: LcoEnv = process.env +): { mode: "stdio" | "daemon"; control: CodexClient; read: CodexClient } { + const config = resolveCodexRuntimeTransportConfig(env); + if (config.mode === "daemon") { + const sharedClient = createCodexAppServerDaemonClient({ + socketPath: config.socketPath, + surface: "control" + }); + return { + mode: config.mode, + control: sharedClient, + read: { + async request(method, params) { + assertCodexMethodAllowed(method, "read"); + return sharedClient.request(method, params); + } + } + }; + } + return { + mode: config.mode, + control: createCodexAppServerStdioClient({ + command: config.command, + args: config.args, + surface: "control" + }), + read: createCodexAppServerStdioClient({ + command: config.command, + args: config.args, + surface: "read" + }) + }; +} diff --git a/packages/mcp-server/src/server-runtime.ts b/packages/mcp-server/src/server-runtime.ts index f9a43021..12ff81bf 100644 --- a/packages/mcp-server/src/server-runtime.ts +++ b/packages/mcp-server/src/server-runtime.ts @@ -2,9 +2,9 @@ import { createInterface } from "node:readline"; import { existsSync, readFileSync } from "node:fs"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; -import { createAuditStore, createCodexAppServerStdioClient } from "../../adapters/src/index.js"; +import { createAuditStore } from "../../adapters/src/index.js"; import { createDatabase, defaultDatabasePath } from "../../core/src/index.js"; -import { readEnv, readEnvWithFallback, resolveHomeDir } from "../../runtime/src/env.js"; +import { readEnv, resolveHomeDir } from "../../runtime/src/env.js"; import { createLooToolDeclarations, createLooTools, @@ -17,6 +17,7 @@ import { normalizeMcpStructuredContent, serializeMcpTextContent } from "./mcp-protocol.js"; +import { createConfiguredCodexClients } from "./codex-runtime-transport.js"; const toolProfile = parseLooToolProfile(readEnv("TOOL_PROFILE"), { onInvalid: (value) => { @@ -99,21 +100,8 @@ function getRuntimeState(): RuntimeState { return { ok: false, failure: createStartupUnavailableResult("audit_unavailable") }; } - const codexCommand = readEnvWithFallback("CODEX_BIN", "codex"); - const codexArgs = (readEnv("CODEX_APP_SERVER_ARGS") || "app-server --stdio").split(/\s+/).filter(Boolean); - const codexClient = createCodexAppServerStdioClient({ - command: codexCommand, - args: codexArgs, - surface: "control" - }); - const codexReadClient = createCodexAppServerStdioClient({ - command: codexCommand, - args: codexArgs, - surface: "read" - }); - - try { + const codexClients = createConfiguredCodexClients(); runtimeState = { ok: true, tools: createLooTools({ @@ -121,8 +109,8 @@ function getRuntimeState(): RuntimeState { dbPath, audit, includeAliases: true, - codexClient, - codexReadClient, + codexClient: codexClients.control, + codexReadClient: codexClients.read, invocationSurface: "mcp" }) }; diff --git a/packages/mcp-server/src/tools.ts b/packages/mcp-server/src/tools.ts index 80f0eb8d..1677bf3c 100644 --- a/packages/mcp-server/src/tools.ts +++ b/packages/mcp-server/src/tools.ts @@ -93,6 +93,7 @@ import { } from "../../adapters/src/index.js"; import { probeClaudeDryRunAvailability } from "../../adapters/src/claude.js"; import { readEnv, readEnvWithFallback, resolveHomeDir } from "../../runtime/src/env.js"; +import { createCodexControlRouter } from "./codex-control-router.js"; export type LooTool = { name: string; @@ -228,17 +229,21 @@ export const LOO_TOOL_SURFACE: Record = { lco_session_sanitizer: { tier: "proof_debug" }, lco_codex_sqlite_stores: { tier: "internal_low_level" }, lco_lcm_peer_dbs: { tier: "internal_low_level" }, - lco_drive: { + lco_drive: { tier: "workflow_detail" }, + lco_codex_control_route: { tier: "public_facade", operatorPathRank: 8, - operatorPathRole: "Create a bounded review-then-drive plan and real target-adapter dry-run audit packet." + operatorPathRole: "Resolve one daemon-owned Codex target or return the explicit Desktop observation route without exposing raw identifiers." + }, + lco_codex_deliver: { + tier: "public_facade", + operatorPathRank: 9, + operatorPathRole: "Deliver through the opaque target using idle send or active steer under the existing dry-run approval audit." }, lco_codex_control_dry_run: { tier: "workflow_detail" }, lco_codex_start_thread: { tier: "workflow_detail" }, lco_codex_resume_thread: { - tier: "public_facade", - operatorPathRank: 9, - operatorPathRole: "Run the approved resume action only after a matching dry-run audit id." + tier: "workflow_detail" }, lco_codex_send_message: { tier: "workflow_detail" }, lco_codex_steer_thread: { tier: "workflow_detail" }, @@ -462,6 +467,16 @@ function validateOpenClawToolInput(schema: Record, input: Recor if (!isRecordValue(input)) return "value must be an object"; const properties = isRecordValue(schema.properties) ? schema.properties : {}; const requiredFields = Array.isArray(schema.required) ? schema.required.filter((value): value is string => typeof value === "string") : []; + const alternatives = Array.isArray(schema.anyOf) + ? schema.anyOf + .map((value) => isRecordValue(value) && Array.isArray(value.required) + ? value.required.filter((field): field is string => typeof field === "string") + : []) + .filter((fields) => fields.length > 0) + : []; + if (alternatives.length > 0 && !alternatives.some((fields) => fields.every((field) => input[field] !== undefined))) { + return "one supported target form is required"; + } for (const key of requiredFields) { if (input[key] === undefined) return `${publicSafeInputField(key)} is required`; } @@ -535,6 +550,10 @@ export function createLooTools(options: { }): LooTool[] { const control = createCodexControl({ audit: options.audit, client: options.codexClient }); const codexReadClient = options.codexReadClient ?? options.codexClient; + const codexRouter = createCodexControlRouter({ + client: codexReadClient, + control + }); const telemetryEnabled = options.telemetryEnabled ?? readEnv("TELEMETRY") === "1"; const tools: LooTool[] = [ tool("lco_index_sessions", "Index local Codex and Claude Code session JSONL files into the local orchestrator database.", { @@ -1083,6 +1102,24 @@ export function createLooTools(options: { now: optionalString(input.now) }); }), + tool("lco_codex_control_route", "Resolve one daemon-owned Codex task to an expiring opaque target, or require Desktop observation without mutation.", { + hint: { type: "string", maxLength: 240 } + }, (input) => codexRouter.route({ + hint: optionalString(input.hint) + })), + tool("lco_codex_deliver", "Deliver to an opaque Codex target: start a turn when idle or steer the matching active turn. Dry-run is the default.", { + target_ref: { type: "string" }, + message: { type: "string" }, + turn_wait_ms: { type: "integer", minimum: 1, maximum: 600000 }, + dry_run: { type: "boolean", default: true }, + approval_audit_id: { type: "string" } + }, (input) => codexRouter.deliver({ + targetRef: requiredString(input.target_ref, "target_ref"), + message: requiredString(input.message, "message"), + turnWaitMs: optionalNumber(input.turn_wait_ms), + dryRun: input.dry_run !== false, + approvalAuditId: optionalString(input.approval_audit_id) + }), ["target_ref", "message"]), tool("lco_codex_control_dry_run", "Create a dry-run audit id for a Codex control action under LCO's fixed never-approve, read-only runtime posture.", { action: { type: "string", enum: ["start", "send", "resume", "steer", "interrupt"] }, thread_id: { type: "string" }, @@ -1093,7 +1130,22 @@ export function createLooTools(options: { tool("lco_codex_resume_thread", "Resume or rejoin a Codex thread under LCO's fixed never-approve, read-only posture without starting a turn. Live mode requires approval_audit_id.", controlSchema(), (input) => snakeCaseControlResult(control.resumeThread(controlInput(input)))), tool("lco_codex_send_message", "Send a message under LCO's fixed never-approve, read-only posture. Live mode requires approval_audit_id and waits for bounded turn proof.", controlSchema(true, false, true), (input) => snakeCaseControlResult(control.sendMessage(messageControlInput(input, false, true)))), tool("lco_codex_steer_thread", "Rejoin and steer a running Codex thread under LCO's fixed never-approve, read-only posture. Live mode requires approval_audit_id and expected_turn_id.", controlSchema(true, true, true), (input) => snakeCaseControlResult(control.steerThread(messageControlInput(input, true, true))), ["thread_id", "message", "expected_turn_id"]), - tool("lco_codex_interrupt_thread", "Rejoin and interrupt a Codex thread under LCO's fixed never-approve, read-only posture. Live mode requires approval_audit_id and expected_turn_id.", controlSchema(false, true, true), (input) => snakeCaseControlResult(control.interruptThread(controlInput(input, false, true))), ["thread_id", "expected_turn_id"]), + tool("lco_codex_interrupt_thread", "Interrupt a matching opaque target or rejoin and interrupt a raw Codex thread under LCO's fixed never-approve, read-only posture.", { + ...controlSchema(false, true, true), + target_ref: { type: "string" } + }, (input) => optionalString(input.target_ref) + ? codexRouter.interrupt({ + targetRef: requiredString(input.target_ref, "target_ref"), + turnWaitMs: optionalNumber(input.turn_wait_ms), + dryRun: input.dry_run !== false, + approvalAuditId: optionalString(input.approval_audit_id) + }) + : snakeCaseControlResult(control.interruptThread(controlInput(input, false, true))), [], { + anyOf: [ + { required: ["target_ref"] }, + { required: ["thread_id", "expected_turn_id"] } + ] + }), tool("lco_desktop_act", "Dry-run desktop fallback action for CUA/Peekaboo; live requests return structured missing-proof blockers.", { backend: { type: "string", enum: ["direct", "cua-driver", "peekaboo"] }, action: { type: "string" }, @@ -1575,7 +1627,8 @@ function tool( description: string, properties: Record, execute: LooTool["execute"], - required: string[] = [] + required: string[] = [], + schemaExtras: Record = {} ): LooTool { const safety = LOO_COMMAND_POLICY[name]; if (!safety) throw new Error(`Missing LOO command policy for ${name}`); @@ -1590,7 +1643,8 @@ function tool( type: "object", additionalProperties: false, properties, - ...(required.length > 0 ? { required } : {}) + ...(required.length > 0 ? { required } : {}), + ...schemaExtras }, execute }; diff --git a/packages/openclaw-plugin/openclaw.plugin.json b/packages/openclaw-plugin/openclaw.plugin.json index 28be78b4..0f3f8b1b 100644 --- a/packages/openclaw-plugin/openclaw.plugin.json +++ b/packages/openclaw-plugin/openclaw.plugin.json @@ -2,7 +2,7 @@ "id": "lossless-openclaw-orchestrator", "name": "Lossless OpenClaw Orchestrator", "description": "Index, search, and prepare local Codex sessions for OpenClaw with approval-gated dry-runs and optional Codex controls.", - "version": "1.6.1", + "version": "1.7.0", "kind": "tool", "tools": { "prefix": "lco_" @@ -53,6 +53,8 @@ "lco_codex_sqlite_stores", "lco_lcm_peer_dbs", "lco_drive", + "lco_codex_control_route", + "lco_codex_deliver", "lco_codex_control_dry_run", "lco_codex_start_thread", "lco_codex_resume_thread", @@ -90,6 +92,8 @@ "loo_codex_sqlite_stores", "loo_lcm_peer_dbs", "loo_drive", + "loo_codex_control_route", + "loo_codex_deliver", "loo_codex_control_dry_run", "loo_codex_start_thread", "loo_codex_resume_thread", @@ -1323,9 +1327,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 8, - "operatorPathRole": "Create a bounded review-then-drive plan and real target-adapter dry-run audit packet." + "tier": "workflow_detail" }, "inputSchema": { "type": "object", @@ -1385,6 +1387,77 @@ } } }, + { + "name": "lco_codex_control_route", + "description": "Resolve one daemon-owned Codex task to an expiring opaque target, or require Desktop observation without mutation.", + "safety": { + "mode": "read_only", + "source": "codex_direct", + "requiresApproval": false, + "mutationClasses": [] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 8, + "operatorPathRole": "Resolve one daemon-owned Codex target or return the explicit Desktop observation route without exposing raw identifiers." + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "hint": { + "type": "string", + "maxLength": 240 + } + } + } + }, + { + "name": "lco_codex_deliver", + "description": "Deliver to an opaque Codex target: start a turn when idle or steer the matching active turn. Dry-run is the default.", + "safety": { + "mode": "approval_gated_control", + "source": "codex_direct", + "requiresApproval": true, + "mutationClasses": [ + "derived_cache", + "live_control" + ] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 9, + "operatorPathRole": "Deliver through the opaque target using idle send or active steer under the existing dry-run approval audit." + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "target_ref": { + "type": "string" + }, + "message": { + "type": "string" + }, + "turn_wait_ms": { + "type": "integer", + "minimum": 1, + "maximum": 600000 + }, + "dry_run": { + "type": "boolean", + "default": true + }, + "approval_audit_id": { + "type": "string" + } + }, + "required": [ + "target_ref", + "message" + ] + } + }, { "name": "lco_codex_control_dry_run", "description": "Create a dry-run audit id for a Codex control action under LCO's fixed never-approve, read-only runtime posture.", @@ -1467,9 +1540,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 9, - "operatorPathRole": "Run the approved resume action only after a matching dry-run audit id." + "tier": "workflow_detail" }, "inputSchema": { "type": "object", @@ -1578,7 +1649,7 @@ }, { "name": "lco_codex_interrupt_thread", - "description": "Rejoin and interrupt a Codex thread under LCO's fixed never-approve, read-only posture. Live mode requires approval_audit_id and expected_turn_id.", + "description": "Interrupt a matching opaque target or rejoin and interrupt a raw Codex thread under LCO's fixed never-approve, read-only posture.", "safety": { "mode": "approval_gated_control", "source": "codex_direct", @@ -1612,11 +1683,23 @@ }, "approval_audit_id": { "type": "string" + }, + "target_ref": { + "type": "string" } }, - "required": [ - "thread_id", - "expected_turn_id" + "anyOf": [ + { + "required": [ + "target_ref" + ] + }, + { + "required": [ + "thread_id", + "expected_turn_id" + ] + } ] } }, @@ -3167,9 +3250,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 8, - "operatorPathRole": "Create a bounded review-then-drive plan and real target-adapter dry-run audit packet.", + "tier": "workflow_detail", "aliasOf": "lco_drive" }, "inputSchema": { @@ -3230,6 +3311,79 @@ } } }, + { + "name": "loo_codex_control_route", + "description": "Resolve one daemon-owned Codex task to an expiring opaque target, or require Desktop observation without mutation.", + "safety": { + "mode": "read_only", + "source": "codex_direct", + "requiresApproval": false, + "mutationClasses": [] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 8, + "operatorPathRole": "Resolve one daemon-owned Codex target or return the explicit Desktop observation route without exposing raw identifiers.", + "aliasOf": "lco_codex_control_route" + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "hint": { + "type": "string", + "maxLength": 240 + } + } + } + }, + { + "name": "loo_codex_deliver", + "description": "Deliver to an opaque Codex target: start a turn when idle or steer the matching active turn. Dry-run is the default.", + "safety": { + "mode": "approval_gated_control", + "source": "codex_direct", + "requiresApproval": true, + "mutationClasses": [ + "derived_cache", + "live_control" + ] + }, + "metadata": { + "tier": "public_facade", + "operatorPathRank": 9, + "operatorPathRole": "Deliver through the opaque target using idle send or active steer under the existing dry-run approval audit.", + "aliasOf": "lco_codex_deliver" + }, + "inputSchema": { + "type": "object", + "additionalProperties": false, + "properties": { + "target_ref": { + "type": "string" + }, + "message": { + "type": "string" + }, + "turn_wait_ms": { + "type": "integer", + "minimum": 1, + "maximum": 600000 + }, + "dry_run": { + "type": "boolean", + "default": true + }, + "approval_audit_id": { + "type": "string" + } + }, + "required": [ + "target_ref", + "message" + ] + } + }, { "name": "loo_codex_control_dry_run", "description": "Create a dry-run audit id for a Codex control action under LCO's fixed never-approve, read-only runtime posture.", @@ -3314,9 +3468,7 @@ ] }, "metadata": { - "tier": "public_facade", - "operatorPathRank": 9, - "operatorPathRole": "Run the approved resume action only after a matching dry-run audit id.", + "tier": "workflow_detail", "aliasOf": "lco_codex_resume_thread" }, "inputSchema": { @@ -3428,7 +3580,7 @@ }, { "name": "loo_codex_interrupt_thread", - "description": "Rejoin and interrupt a Codex thread under LCO's fixed never-approve, read-only posture. Live mode requires approval_audit_id and expected_turn_id.", + "description": "Interrupt a matching opaque target or rejoin and interrupt a raw Codex thread under LCO's fixed never-approve, read-only posture.", "safety": { "mode": "approval_gated_control", "source": "codex_direct", @@ -3463,11 +3615,23 @@ }, "approval_audit_id": { "type": "string" + }, + "target_ref": { + "type": "string" } }, - "required": [ - "thread_id", - "expected_turn_id" + "anyOf": [ + { + "required": [ + "target_ref" + ] + }, + { + "required": [ + "thread_id", + "expected_turn_id" + ] + } ] } }, @@ -6911,8 +7075,8 @@ "lco_recent_sessions", "lco_attention_inbox", "lco_project_digest", - "lco_drive", - "lco_codex_resume_thread" + "lco_codex_control_route", + "lco_codex_deliver" ], "namingPolicy": { "publicProductAbbreviation": "LCO", diff --git a/tests/bridge-safety-core.test.ts b/tests/bridge-safety-core.test.ts index 7f16dbc3..f2925cdd 100644 --- a/tests/bridge-safety-core.test.ts +++ b/tests/bridge-safety-core.test.ts @@ -1,9 +1,11 @@ import assert from "node:assert/strict"; import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { createServer } from "node:http"; import { homedir, tmpdir } from "node:os"; import { join } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import test from "node:test"; +import { WebSocketServer } from "ws"; import { CODEX_CONTROL_METHODS, @@ -12,9 +14,11 @@ import { CodexJsonRpcClient, LineProcessTransport, LoopbackWebSocketTransport, + UnixSocketWebSocketTransport, assertCodexMethodAllowed, buildLoopbackWebSocketConfig, codexTransportStatus, + createCodexAppServerDaemonClient, createCodexAppServerWebSocketClient, createCodexMcpStdioClient, createCodexControl, @@ -854,6 +858,91 @@ test("Codex loopback WebSocket close drains a pending read without leaking a wai } }); +test("Codex daemon UDS client initializes, exchanges JSON-RPC frames, and reconnects", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-codex-uds-")); + const socketPath = join(root, "app-server.sock"); + const server = createServer(); + const websocketServer = new WebSocketServer({ server }); + const methods: string[] = []; + let connectionCount = 0; + const extensionHeaders: Array = []; + const sockets: Array<{ close(): void }> = []; + + server.on("upgrade", (request) => extensionHeaders.push(request.headers["sec-websocket-extensions"])); + websocketServer.on("connection", (socket) => { + connectionCount += 1; + sockets.push(socket); + socket.on("message", (raw) => { + const payload = JSON.parse(raw.toString()) as Record; + if (typeof payload.method === "string") methods.push(payload.method); + if (payload.id === undefined) return; + socket.send(JSON.stringify({ + id: payload.id, + result: payload.method === "initialize" + ? { serverInfo: {} } + : { data: ["opaque-fixture-id"] } + })); + }); + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(socketPath, resolve); + }); + + try { + const client = createCodexAppServerDaemonClient({ + socketPath, + surface: "read", + timeoutMs: 250 + }); + const first = await client.request("thread/loaded/list", {}); + const second = await client.request("thread/loaded/list", {}); + assert.equal(first.ok, true); + assert.equal(second.ok, true); + assert.equal(connectionCount, 1); + assert.deepEqual(extensionHeaders, [undefined]); + assert.deepEqual(methods, [ + "initialize", "initialized", "thread/loaded/list", "thread/loaded/list" + ]); + sockets[0]?.close(); + await delay(10); + const afterReconnect = await client.request("thread/loaded/list", {}); + assert.equal(afterReconnect.ok, true); + assert.equal(connectionCount, 2); + assert.deepEqual(extensionHeaders, [undefined, undefined]); + assert.deepEqual(methods.slice(-3), ["initialize", "initialized", "thread/loaded/list"]); + } finally { + for (const socket of sockets) socket.close(); + await delay(10); + await new Promise((resolve) => websocketServer.close(() => resolve())); + await new Promise((resolve) => server.close(() => resolve())); + rmSync(root, { recursive: true, force: true }); + } +}); + +test("Codex daemon UDS transport rejects non-absolute and unavailable socket paths without exposing them", async () => { + assert.throws( + () => new UnixSocketWebSocketTransport("relative/app-server.sock", 20), + /absolute Unix socket path/ + ); + + const unavailablePath = join(tmpdir(), "lco-does-not-exist", "private-app-server.sock"); + const client = createCodexAppServerDaemonClient({ + socketPath: unavailablePath, + surface: "read", + timeoutMs: 20 + }); + await assert.rejects( + client.request("thread/loaded/list", {}), + (error: unknown) => { + assert.match(String(error), /daemon WebSocket connection failed/); + assert.equal(String(error).includes(unavailablePath), false); + return true; + } + ); +}); + test("Codex transport status reports command availability without starting a live session", () => { const status = codexTransportStatus({ command: process.execPath, diff --git a/tests/codex-control-router.test.ts b/tests/codex-control-router.test.ts new file mode 100644 index 00000000..e1d4fe16 --- /dev/null +++ b/tests/codex-control-router.test.ts @@ -0,0 +1,382 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +import { + createAuditStore, + createCodexControl, + type CodexClient, + type CodexControlSequenceOptions, + type CodexControlStep +} from "../packages/adapters/src/index.js"; +import { createCodexControlRouter } from "../packages/mcp-server/src/codex-control-router.js"; + +type FixtureThread = { + id: string; + name: string; + state: "active" | "idle"; + turnId?: string; +}; + +function fixtureClient(threads: FixtureThread[], sequenceFailure?: "active_turn_not_steerable") { + let current = threads; + const requestCalls: Array<{ method: string; params: Record }> = []; + const sequenceCalls: Array<{ steps: CodexControlStep[]; options: CodexControlSequenceOptions }> = []; + const client: CodexClient = { + async request(method, params) { + requestCalls.push({ method, params }); + if (method === "thread/loaded/list") { + return { ok: true, result: { data: current.map((thread) => thread.id) }, notifications: [] }; + } + if (method === "thread/read") { + const thread = current.find((candidate) => candidate.id === params.threadId); + if (!thread) return { ok: false, error: "not found", notifications: [] }; + return { + ok: true, + result: { + thread: { + id: thread.id, + name: thread.name, + status: { type: thread.state }, + cwd: "/private/fixture/path", + preview: "PRIVATE_TRANSCRIPT_CANARY", + turns: thread.turnId + ? [{ id: thread.turnId, status: "inProgress", items: [{ text: "PRIVATE_TURN_CANARY" }] }] + : [] + } + }, + notifications: [] + }; + } + if (method === "turn/start") { + return { + ok: true, + result: { turn: { id: "turn-new", status: "inProgress" } }, + notifications: [] + }; + } + throw new Error(`Unexpected method ${method}`); + }, + async requestSequenceUntilTurnResolved(steps, options) { + sequenceCalls.push({ steps, options }); + if (sequenceFailure === "active_turn_not_steerable") { + return { + responses: [ + { + ok: true, + result: { + approvalPolicy: "never", + sandbox: { type: "readOnly", networkAccess: false } + } + }, + { + ok: false, + error: { codexErrorInfo: { activeTurnNotSteerable: { turnKind: "review" } } } + } + ] + }; + } + const turnResponse = { ok: true, result: { turn: { id: options.expectedTurnId ?? "turn-new", status: "completed" } } }; + return { + responses: steps[0]?.method === "turn/start" + ? [turnResponse] + : [ + { + ok: true, + result: { + approvalPolicy: "never", + sandbox: { type: "readOnly", networkAccess: false } + } + }, + turnResponse + ], + turn: { + id: options.expectedTurnId ?? "turn-new", + status: "completed", + completed: true, + notificationMethods: ["turn/completed"], + approvalRequestCount: 0, + serverRequestCount: 0 + } + }; + } + }; + return { + client, + requestCalls, + sequenceCalls, + setThreads(next: FixtureThread[]) { + current = next; + } + }; +} + +test("control route returns an opaque active target without transcript, path, thread, or turn identifiers", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-active-")); + const fixture = fixtureClient([{ + id: "thread-secret-active", + name: "Release task", + state: "active", + turnId: "turn-secret-active" + }]); + const audit = createAuditStore(join(root, "audit.jsonl")); + const router = createCodexControlRouter({ + client: fixture.client, + control: createCodexControl({ audit, client: fixture.client }), + createRef: () => "lco_target_opaque", + now: () => new Date("2026-07-30T10:00:00Z") + }); + + try { + const route = await router.route({}); + assert.equal(route.schema, "lco.codex.controlRoute.v1"); + assert.equal(route.status, "selected"); + assert.equal(route.route, "app_server"); + assert.equal(route.target_ref, "lco_target_opaque"); + assert.equal(route.state, "active"); + assert.deepEqual(route.supported_actions, ["steer", "interrupt"]); + const serialized = JSON.stringify(route); + for (const forbidden of [ + "thread-secret-active", + "turn-secret-active", + "PRIVATE_TRANSCRIPT_CANARY", + "PRIVATE_TURN_CANARY", + "/private/fixture/path" + ]) { + assert.equal(serialized.includes(forbidden), false); + } + assert.deepEqual( + fixture.requestCalls + .filter((call) => call.method === "thread/read") + .map((call) => call.params.includeTurns), + [false, true] + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("delivery chooses active steer, revalidates state, and blocks a changed turn before mutation", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-deliver-")); + const fixture = fixtureClient([{ + id: "thread-active", + name: "Active task", + state: "active", + turnId: "turn-active-1" + }]); + const audit = createAuditStore(join(root, "audit.jsonl")); + const router = createCodexControlRouter({ + client: fixture.client, + control: createCodexControl({ audit, client: fixture.client }), + createRef: () => "lco_target_active", + now: () => new Date("2026-07-30T10:00:00Z") + }); + + try { + const route = await router.route({}); + const dryRun = await router.deliver({ + targetRef: route.target_ref!, + message: "Continue safely" + }); + assert.equal(dryRun.status, "dry_run_ready"); + assert.equal(dryRun.action, "steer"); + assert.equal(fixture.sequenceCalls.length, 0); + + fixture.setThreads([{ + id: "thread-active", + name: "Active task", + state: "active", + turnId: "turn-active-2" + }]); + const blocked = await router.deliver({ + targetRef: route.target_ref!, + message: "Continue safely", + dryRun: false, + approvalAuditId: dryRun.approval_audit_id + }); + assert.equal(blocked.status, "blocked"); + assert.deepEqual(blocked.reason_codes, ["target_turn_changed"]); + assert.equal(fixture.sequenceCalls.length, 0); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("delivery sends to idle targets and interrupt accepts the same opaque active reference", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-actions-")); + const idleFixture = fixtureClient([{ + id: "thread-idle", + name: "Idle task", + state: "idle" + }]); + const idleAudit = createAuditStore(join(root, "idle-audit.jsonl")); + const idleRouter = createCodexControlRouter({ + client: idleFixture.client, + control: createCodexControl({ audit: idleAudit, client: idleFixture.client }), + createRef: () => "lco_target_idle" + }); + + const activeFixture = fixtureClient([{ + id: "thread-active", + name: "Active task", + state: "active", + turnId: "turn-active" + }]); + const activeAudit = createAuditStore(join(root, "active-audit.jsonl")); + const activeRouter = createCodexControlRouter({ + client: activeFixture.client, + control: createCodexControl({ audit: activeAudit, client: activeFixture.client }), + createRef: () => "lco_target_active" + }); + + try { + const idleRoute = await idleRouter.route({}); + assert.deepEqual( + idleFixture.requestCalls + .filter((call) => call.method === "thread/read") + .map((call) => call.params.includeTurns), + [false] + ); + const idleDryRun = await idleRouter.deliver({ + targetRef: idleRoute.target_ref!, + message: "Start work" + }); + assert.equal(idleDryRun.action, "send"); + const idleLive = await idleRouter.deliver({ + targetRef: idleRoute.target_ref!, + message: "Start work", + dryRun: false, + approvalAuditId: idleDryRun.approval_audit_id + }); + assert.equal(idleLive.status, "accepted"); + assert.equal(idleFixture.requestCalls.some((call) => call.method === "turn/start"), true); + assert.equal(idleFixture.sequenceCalls.length, 0); + + const activeRoute = await activeRouter.route({}); + const interruptDryRun = await activeRouter.interrupt({ targetRef: activeRoute.target_ref! }); + assert.equal(interruptDryRun.action, "interrupt"); + const interruptLive = await activeRouter.interrupt({ + targetRef: activeRoute.target_ref!, + dryRun: false, + approvalAuditId: interruptDryRun.approval_audit_id + }); + assert.equal(interruptLive.status, "completed"); + assert.deepEqual(activeFixture.sequenceCalls[0]?.steps.map((step) => step.method), ["thread/resume", "turn/interrupt"]); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("routing is deterministic for ambiguous, missing, and desktop-observation-required targets", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-selection-")); + const fixture = fixtureClient([ + { id: "thread-a", name: "A", state: "active", turnId: "turn-a" }, + { id: "thread-b", name: "B", state: "active", turnId: "turn-b" } + ]); + const audit = createAuditStore(join(root, "audit.jsonl")); + const router = createCodexControlRouter({ + client: fixture.client, + control: createCodexControl({ audit, client: fixture.client }) + }); + + try { + const ambiguous = await router.route({}); + assert.equal(ambiguous.status, "ambiguous"); + assert.equal(ambiguous.target_ref, null); + assert.deepEqual(ambiguous.reason_codes, ["multiple_active_daemon_targets"]); + + const selected = await router.route({ hint: "B" }); + assert.equal(selected.status, "selected"); + assert.equal(selected.title_sanitized, "B"); + + const desktop = await router.route({ hint: "Desktop-only task" }); + assert.equal(desktop.status, "unavailable"); + assert.equal(desktop.route, "desktop_observation_required"); + assert.deepEqual(desktop.reason_codes, ["explicit_hint_not_daemon_loaded"]); + + fixture.setThreads([]); + const none = await router.route({}); + assert.equal(none.status, "none"); + assert.equal(none.route, "unknown"); + assert.deepEqual(none.reason_codes, ["no_daemon_target"]); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("expired and ownership-changed opaque targets fail before control mutation", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-stale-")); + const fixture = fixtureClient([{ + id: "thread-stale", + name: "Stale task", + state: "idle" + }]); + const audit = createAuditStore(join(root, "audit.jsonl")); + let clock = new Date("2026-07-30T10:00:00Z"); + const router = createCodexControlRouter({ + client: fixture.client, + control: createCodexControl({ audit, client: fixture.client }), + ttlMs: 1_000, + createRef: () => "lco_target_stale", + now: () => clock + }); + + try { + const route = await router.route({}); + fixture.setThreads([]); + const ownershipChanged = await router.deliver({ + targetRef: route.target_ref!, + message: "Do not send" + }); + assert.deepEqual(ownershipChanged.reason_codes, ["target_ownership_changed"]); + assert.equal(fixture.sequenceCalls.length, 0); + + fixture.setThreads([{ id: "thread-stale", name: "Stale task", state: "idle" }]); + const fresh = await router.route({}); + clock = new Date("2026-07-30T10:00:02Z"); + const expired = await router.deliver({ + targetRef: fresh.target_ref!, + message: "Do not send" + }); + assert.deepEqual(expired.reason_codes, ["target_ref_expired"]); + assert.equal(fixture.sequenceCalls.length, 0); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + +test("delivery returns the specific non-steerable blocker from Codex", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-non-steerable-")); + const fixture = fixtureClient([{ + id: "thread-review", + name: "Review task", + state: "active", + turnId: "turn-review" + }], "active_turn_not_steerable"); + const audit = createAuditStore(join(root, "audit.jsonl")); + const router = createCodexControlRouter({ + client: fixture.client, + control: createCodexControl({ audit, client: fixture.client }) + }); + + try { + const route = await router.route({}); + const dryRun = await router.deliver({ + targetRef: route.target_ref!, + message: "Steer review" + }); + const live = await router.deliver({ + targetRef: route.target_ref!, + message: "Steer review", + dryRun: false, + approvalAuditId: dryRun.approval_audit_id + }); + assert.equal(live.status, "blocked"); + assert.equal(live.control_sent, false); + assert.deepEqual(live.reason_codes, ["active_turn_not_steerable"]); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); diff --git a/tests/codex-runtime-transport.test.ts b/tests/codex-runtime-transport.test.ts new file mode 100644 index 00000000..49028ca5 --- /dev/null +++ b/tests/codex-runtime-transport.test.ts @@ -0,0 +1,45 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { resolveCodexRuntimeTransportConfig } from "../packages/mcp-server/src/codex-runtime-transport.js"; + +test("Codex runtime transport defaults to stdio and keeps app-server args there", () => { + assert.deepEqual(resolveCodexRuntimeTransportConfig({ + HOME: "/tmp/lco-home", + LCO_CODEX_APP_SERVER_ARGS: "app-server --stdio --listen ws://127.0.0.1:4555" + }), { + mode: "stdio", + command: "codex", + args: ["app-server", "--stdio", "--listen", "ws://127.0.0.1:4555"] + }); +}); + +test("Codex runtime daemon selection resolves the standard socket and ignores stdio args", () => { + assert.deepEqual(resolveCodexRuntimeTransportConfig({ + HOME: "/tmp/lco-home", + CODEX_HOME: "/tmp/codex-home", + LCO_CODEX_TRANSPORT: "daemon", + LCO_CODEX_APP_SERVER_ARGS: "must remain stdio only" + }), { + mode: "daemon", + socketPath: "/tmp/codex-home/app-server-control/app-server-control.sock" + }); +}); + +test("Codex runtime daemon selection requires an absolute override and rejects unknown modes", () => { + assert.throws( + () => resolveCodexRuntimeTransportConfig({ + HOME: "/tmp/lco-home", + LCO_CODEX_TRANSPORT: "daemon", + LCO_CODEX_DAEMON_SOCKET: "relative/app-server.sock" + }), + /absolute/ + ); + assert.throws( + () => resolveCodexRuntimeTransportConfig({ + HOME: "/tmp/lco-home", + LCO_CODEX_TRANSPORT: "remote" + }), + /stdio or daemon/ + ); +}); diff --git a/tests/control-and-mcp.test.ts b/tests/control-and-mcp.test.ts index 45fa367d..9fd03fb7 100644 --- a/tests/control-and-mcp.test.ts +++ b/tests/control-and-mcp.test.ts @@ -1618,6 +1618,8 @@ test("MCP tool registry exposes lco-prefixed canonical tools with loo compatibil assert.equal(toolNames.includes("lco_codex_start_thread"), true); assert.equal(toolNames.includes("lco_codex_send_message"), true); assert.equal(toolNames.includes("lco_drive"), true); + assert.equal(toolNames.includes("lco_codex_control_route"), true); + assert.equal(toolNames.includes("lco_codex_deliver"), true); assert.equal(toolNames.includes("lco_desktop_proof"), true); assert.deepEqual(toolNames.filter((name) => !LOO_COMMAND_POLICY[name]), []); for (const declaration of createLooToolDeclarations()) { @@ -1647,6 +1649,10 @@ test("MCP tool registry exposes lco-prefixed canonical tools with loo compatibil assert.equal(LOO_COMMAND_POLICY.lco_index_sessions.mutationClasses.includes("live_control"), false); assert.equal(LOO_COMMAND_POLICY.lco_codex_control_dry_run.mode, "local_cache_write"); assert.deepEqual(LOO_COMMAND_POLICY.lco_codex_control_dry_run.mutationClasses, ["derived_cache"]); + assert.equal(LOO_COMMAND_POLICY.lco_codex_control_route.mode, "read_only"); + assert.deepEqual(LOO_COMMAND_POLICY.lco_codex_control_route.mutationClasses, []); + assert.equal(LOO_COMMAND_POLICY.lco_codex_deliver.mode, "approval_gated_control"); + assert.deepEqual(LOO_COMMAND_POLICY.lco_codex_deliver.mutationClasses, ["derived_cache", "live_control"]); assert.equal(LOO_COMMAND_POLICY.lco_drive.mode, "local_cache_write"); assert.deepEqual(LOO_COMMAND_POLICY.lco_drive.mutationClasses, ["derived_cache"]); assert.deepEqual(LOO_COMMAND_POLICY.lco_codex_start_thread.mutationClasses, ["derived_cache", "live_control"]); @@ -1854,7 +1860,17 @@ test("MCP tool registry exposes lco-prefixed canonical tools with loo compatibil assert.deepEqual(steerTool.inputSchema.required, ["thread_id", "message", "expected_turn_id"]); const interruptTool = tools.find((tool) => tool.name === "lco_codex_interrupt_thread"); assert.ok(interruptTool); - assert.deepEqual(interruptTool.inputSchema.required, ["thread_id", "expected_turn_id"]); + assert.ok((interruptTool.inputSchema.properties as Record).target_ref); + assert.deepEqual(interruptTool.inputSchema.anyOf, [ + { required: ["target_ref"] }, + { required: ["thread_id", "expected_turn_id"] } + ]); + const missingInterruptTarget = await executeLooToolForOpenClaw(interruptTool, {}) as { + ok: boolean; + error?: { message?: string }; + }; + assert.equal(missingInterruptTarget.ok, false); + assert.equal(missingInterruptTarget.error?.message, "one supported target form is required"); const dryRunToolSchema = dryRunTool.inputSchema.properties as Record; assert.ok(dryRunToolSchema.expected_turn_id); assert.throws( diff --git a/tests/hermes-smoke.test.ts b/tests/hermes-smoke.test.ts index 8e99e75c..1a88b105 100644 --- a/tests/hermes-smoke.test.ts +++ b/tests/hermes-smoke.test.ts @@ -36,7 +36,7 @@ test("Hermes smoke proves the Eva tool set, silent notifications, and object-val assert.equal(report.schema, "lco.hermesSmoke.v1"); assert.equal(report.ok, true); assert.equal(report.requiredToolsPresent, true); - assert.equal(report.requiredTools.length, 14); + assert.equal(report.requiredTools.length, 16); assert.equal(report.notificationSilenceReady, true); assert.equal(report.structuredContentObjectReady, true); assert.equal(report.arrayResultWrappedReady, true); diff --git a/tests/openclaw-plugin-manifest.test.ts b/tests/openclaw-plugin-manifest.test.ts index 79d8937b..a3e8985e 100644 --- a/tests/openclaw-plugin-manifest.test.ts +++ b/tests/openclaw-plugin-manifest.test.ts @@ -151,7 +151,7 @@ test("OpenClaw plugin contracts classify every tool into an operator surface tie const lcoAliases = aliases.filter((declaration) => declaration.name.startsWith("lco_")); const compatibilityAliases = aliases.filter((declaration) => declaration.name.startsWith("loo_")); assert.deepEqual(lcoAliases, []); - assert.equal(compatibilityAliases.length, 68); + assert.equal(compatibilityAliases.length, 70); for (const alias of compatibilityAliases) { assert.equal(baseDeclarations.some((declaration) => declaration.name === alias.metadata?.aliasOf), true); } diff --git a/tests/qa-lab-tool-coverage.test.ts b/tests/qa-lab-tool-coverage.test.ts index 7525515a..3a7e8947 100644 --- a/tests/qa-lab-tool-coverage.test.ts +++ b/tests/qa-lab-tool-coverage.test.ts @@ -121,14 +121,14 @@ test("qa-lab tool coverage passes strict full coverage only when every declared assert.equal(report.ok, true); assert.equal(report.qaLabToolCoverageReady, true); assert.equal(report.publicSafe, true); - assert.equal(report.declaredToolCount, 37); + assert.equal(report.declaredToolCount, 39); assert.deepEqual(report.tierCounts, { public_facade: 9, - workflow_detail: 17, + workflow_detail: 19, proof_debug: 8, internal_low_level: 3 }); - assert.equal(report.invocationCoverage.invokedDeclaredTools, 37); + assert.equal(report.invocationCoverage.invokedDeclaredTools, 39); assert.equal(report.invocationCoverage.missingDeclaredTools.length, 0); assert.equal(report.blockers.length, 0); assert.equal(report.actionsPerformed.liveCodexControlRun, false); @@ -156,7 +156,7 @@ test("qa-lab tool coverage fails strict for default gateway evidence when non-fa assert.equal(report.ok, false); assert.equal(report.qaLabToolCoverageReady, false); - assert.equal(report.invocationCoverage.totalDeclaredTools, 37); + assert.equal(report.invocationCoverage.totalDeclaredTools, 39); assert.equal(report.invocationCoverage.invokedDeclaredTools < report.invocationCoverage.totalDeclaredTools, true); assert.equal(report.invocationCoverage.missingDeclaredTools.length > 0, true); assert.ok(report.blockers.some((blocker) => blocker.code === "declared_tool_product_evidence_missing")); @@ -396,8 +396,8 @@ test("qa-lab tool coverage credits folded legacy aliases to their canonical umbr }); assert.equal(report.ok, true, JSON.stringify(report.blockers, null, 2)); - assert.equal(report.declaredToolCount, 37); - assert.equal(report.invocationCoverage.invokedDeclaredTools, 37); + assert.equal(report.declaredToolCount, 39); + assert.equal(report.invocationCoverage.invokedDeclaredTools, 39); for (const umbrella of ["lco_watchers", "lco_codex_extract", "lco_prepared_state", "lco_operating_picture", "lco_desktop_proof"]) { const row = report.toolRows.find((item) => item.name === umbrella); assert.equal(row?.coverageStatus, "covered", `${umbrella} should be covered through a folded legacy alias`); @@ -598,5 +598,5 @@ test("loo qa-lab tool-coverage --strict exits nonzero for missing product eviden assert.equal(result.status, 1, result.stderr || result.stdout); const report = JSON.parse(result.stdout) as QaLabToolCoverageReport; assert.equal(report.schema, "lco.qaLab.toolCoverage.v1"); - assert.equal(report.invocationCoverage.missingDeclaredTools.length, 18); + assert.equal(report.invocationCoverage.missingDeclaredTools.length, 20); }); diff --git a/tests/tool-exposure-profile.test.ts b/tests/tool-exposure-profile.test.ts index 7f879f21..ce41f663 100644 --- a/tests/tool-exposure-profile.test.ts +++ b/tests/tool-exposure-profile.test.ts @@ -68,7 +68,7 @@ test("lco-prefixed tools are the canonical base declarations for every tier", () const baseDeclarations = createLooToolDeclarations({ profile: "all", includeAliases: false }); const publicFacadeNames = createLooToolSurfaceSummary().publicFacadeTools; - assert.equal(baseDeclarations.length, 37); + assert.equal(baseDeclarations.length, 39); assert.equal(baseDeclarations.every((tool) => tool.name.startsWith("lco_")), true); assert.equal(baseDeclarations.some((tool) => tool.name.startsWith("loo_")), false); assert.equal(baseDeclarations.some((tool) => tool.metadata.aliasOf), false); @@ -80,8 +80,8 @@ test("lco-prefixed tools are the canonical base declarations for every tier", () "lco_recent_sessions", "lco_attention_inbox", "lco_project_digest", - "lco_drive", - "lco_codex_resume_thread" + "lco_codex_control_route", + "lco_codex_deliver" ]); }); @@ -148,9 +148,9 @@ test("C1 lco canonical umbrellas replace folded read-only leaf tools while prese assert.deepEqual(byName.get(compatName)?.safety, byName.get(targetName)?.safety); } - assert.equal(baseDeclarations.length, 37); + assert.equal(baseDeclarations.length, 39); assert.equal(aliasedDeclarations.filter((tool) => tool.name.startsWith("lco_") && tool.metadata.aliasOf).length, 0); - assert.equal(aliasedDeclarations.filter((tool) => tool.name.startsWith("loo_") && tool.metadata.aliasOf).length, 68); + assert.equal(aliasedDeclarations.filter((tool) => tool.name.startsWith("loo_") && tool.metadata.aliasOf).length, 70); }); test("redirect aliases target any declared tool and merge kind defaults before caller args", async () => { @@ -365,6 +365,10 @@ function sampleInputForTarget(targetName: string, root: string): Record Date: Thu, 30 Jul 2026 18:05:21 +0700 Subject: [PATCH 2/5] fix: consume Codex control approvals before send --- packages/adapters/src/index.ts | 92 ++++++++++++++++--- .../mcp-server/src/codex-control-router.ts | 27 +++++- tests/codex-control-router.test.ts | 62 ++++++++++++- tests/control-and-mcp.test.ts | 25 ++++- 4 files changed, 184 insertions(+), 22 deletions(-) diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index defc103d..dab447d6 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -670,7 +670,9 @@ export type AuditStore = Omit, "deriveSubkey deriveSubkeyIfConfigured?(domain: string): string | null; fingerprintTextIfConfigured?(value: string): string | null; }; -type ControlAuditStore = Pick; +type ControlAuditStore = Pick & { + tail?(limit?: number): AuditRecord[]; +}; export type TargetControlExecuteSpec = { action: string; @@ -703,6 +705,7 @@ export type AuditRecord = { paramsHash: string; messageHash?: string; approvalAuditId?: string; + approvalState?: "dry_run" | "claimed" | "completed"; live: boolean; createdAt: string; }; @@ -804,6 +807,7 @@ export function createTargetControl(options: { targetName: string; methodPolicy: target: spec.threadId, paramsHash, messageHash, + approvalState: "dry_run", live: false }); return { @@ -829,7 +833,7 @@ export function createTargetControl(options: { targetName: string; methodPolicy: if (!previous) { throw new Error("approval_audit_id was not found in the local audit log"); } - if (previous.live !== false) { + if (previous.live !== false || previous.approvalAuditId || previous.approvalState === "claimed") { throw new Error("approval_audit_id must reference a dry-run Codex control audit record"); } if (previous.action !== spec.action || previous.target !== spec.threadId || previous.paramsHash !== paramsHash) { @@ -839,16 +843,35 @@ export function createTargetControl(options: { targetName: string; methodPolicy: if (!Number.isFinite(dryRunCreatedAtMs) || dryRunCreatedAtMs + CODEX_CONTROL_DRY_RUN_TTL_MS <= Date.now()) { throw new Error("approval_audit_id dry-run record expired"); } - const sequenceResult = requiresSequence - ? await requestCodexControlSequence(options.client, steps, spec.turnResolution - ? { - threadId: spec.threadId, - expectedTurnId: spec.turnResolution.expectedTurnId, - turnWaitMs: resolveCodexTurnWaitMs(spec.turnResolution.turnWaitMs), - ...(spec.turnResolution.requireSafeActiveRuntime ? { requireSafeActiveRuntime: true } : {}) - } - : undefined) - : undefined; + assertCodexControlTransportCapability(options.client, requiresSequence, Boolean(spec.turnResolution)); + if (options.audit.tail?.(1000).some((record) => record.approvalAuditId === previous.id)) { + throw new Error("approval_audit_id has already been used"); + } + options.audit.append({ + action: spec.action, + target: spec.threadId, + paramsHash, + messageHash, + approvalAuditId: previous.id, + approvalState: "claimed", + live: false + }); + let sequenceResult: CodexControlSequenceResult | undefined; + try { + sequenceResult = requiresSequence + ? await requestCodexControlSequence(options.client, steps, spec.turnResolution + ? { + threadId: spec.threadId, + expectedTurnId: spec.turnResolution.expectedTurnId, + turnWaitMs: resolveCodexTurnWaitMs(spec.turnResolution.turnWaitMs), + ...(spec.turnResolution.requireSafeActiveRuntime ? { requireSafeActiveRuntime: true } : {}) + } + : undefined) + : undefined; + } catch (error) { + if (!isIndeterminateControlError(error)) throw error; + throw new Error("codex_control_attempt_indeterminate"); + } const safeRuntimeBlock = safeRuntimeBlockFromSequence(sequenceResult); if (safeRuntimeBlock) { return { @@ -877,9 +900,17 @@ export function createTargetControl(options: { targetName: string; methodPolicy: response: sanitizeCodexControlResponse(safeRuntimeBlock) }; } - const rawResponse = sequenceResult - ? sequenceResult.responses.at(-1) ?? { ok: true } - : await options.client.request(spec.method, spec.params); + let rawResponse: unknown; + try { + rawResponse = sequenceResult + ? sequenceResult.responses.at(-1) ?? { ok: true } + : await options.client.request(spec.method, spec.params); + } catch { + throw new Error("codex_control_attempt_indeterminate"); + } + if (isIndeterminateControlResponse(rawResponse)) { + throw new Error("codex_control_attempt_indeterminate"); + } const response = responseWithTurnResolution(rawResponse, sequenceResult?.turn); const liveRecord = options.audit.append({ action: spec.action, @@ -887,6 +918,7 @@ export function createTargetControl(options: { targetName: string; methodPolicy: paramsHash, messageHash, approvalAuditId: previous.id, + approvalState: "completed", live: true }); const createdThreadCandidateId = spec.createdThreadFromResponse ? extractControlThreadId(response) : undefined; @@ -1318,12 +1350,29 @@ async function requestCodexControlSequence( return { responses }; } +function assertCodexControlTransportCapability( + client: CodexClient, + requiresSequence: boolean, + requiresTurnResolution: boolean +): void { + if (!requiresSequence) return; + if (requiresTurnResolution && !client.requestSequenceUntilTurnResolved) { + throw new Error("turn lifecycle proof is required for this Codex control action"); + } + if (!requiresTurnResolution && !client.requestSequence) { + throw new Error("same-connection control sequence is required for this Codex control action"); + } +} + function assertCodexControlSequenceResponses(responses: unknown[], steps: CodexControlStep[]): void { for (let index = 0; index < responses.length; index += 1) { const response = asRecord(responses[index]); const isLcoSafetyBlock = response?.code === "safe_runtime_posture_unproven" && response.origin === "lco_safety_gate"; if (response?.ok === false && !isLcoSafetyBlock) { + if (isIndeterminateControlResponse(response)) { + throw new Error("codex_control_attempt_indeterminate"); + } if (JSON.stringify(response).includes("activeTurnNotSteerable")) { throw new Error("active_turn_not_steerable"); } @@ -1335,6 +1384,19 @@ function assertCodexControlSequenceResponses(responses: unknown[], steps: CodexC } } +function isIndeterminateControlResponse(value: unknown): boolean { + const response = asRecord(value); + if (response?.ok !== false) return false; + const error = typeof response.error === "string" ? response.error : JSON.stringify(response.error ?? ""); + return /timed out waiting|connection (?:failed|closed|lost)|socket|hang up|econnreset|broken pipe/i.test(error); +} + +function isIndeterminateControlError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + return message === "codex_control_attempt_indeterminate" + || /timed out waiting|connection (?:failed|closed|lost)|socket|hang up|econnreset|broken pipe|websocket/i.test(message); +} + function safeRuntimeBlockFromSequence(sequence: CodexControlSequenceResult | undefined): Record | null { const response = asRecord(sequence?.responses.at(-1)); return response?.ok === false diff --git a/packages/mcp-server/src/codex-control-router.ts b/packages/mcp-server/src/codex-control-router.ts index e425132b..028a2ecb 100644 --- a/packages/mcp-server/src/codex-control-router.ts +++ b/packages/mcp-server/src/codex-control-router.ts @@ -42,7 +42,7 @@ export type CodexDeliveryResult = { action: "send" | "steer" | "interrupt" | null; target_ref: string; live: boolean; - control_sent: boolean; + control_sent: boolean | null; approval_audit_id?: string; params_hash?: string; message_hash?: string; @@ -127,6 +127,9 @@ export function createCodexControlRouter(options: { }); return publicDeliveryResult(input.targetRef, action, result); } catch (error) { + if (String(error).includes("codex_control_attempt_indeterminate")) { + return indeterminateDelivery(input.targetRef, action); + } const reason = String(error).includes("active_turn_not_steerable") ? "active_turn_not_steerable" : "approval_or_control_rejected"; @@ -155,7 +158,10 @@ export function createCodexControlRouter(options: { turnWaitMs: input.turnWaitMs }); return publicDeliveryResult(input.targetRef, "interrupt", result); - } catch { + } catch (error) { + if (String(error).includes("codex_control_attempt_indeterminate")) { + return indeterminateDelivery(input.targetRef, "interrupt"); + } return blockedDelivery(input.targetRef, "interrupt", input.dryRun === false, "approval_or_control_rejected"); } } @@ -328,6 +334,23 @@ function blockedDelivery( }; } +function indeterminateDelivery( + targetRef: string, + action: Exclude +): CodexDeliveryResult { + return { + schema: "lco.codex.delivery.v1", + status: "blocked", + action, + target_ref: targetRef, + live: true, + control_sent: null, + reason_codes: ["control_attempt_indeterminate", "approval_consumed_do_not_retry"], + public_safe: true, + raw_transcript_returned: false + }; +} + function normalizeHint(value: string): string { return value.startsWith("codex_thread:") ? value.slice("codex_thread:".length) : value; } diff --git a/tests/codex-control-router.test.ts b/tests/codex-control-router.test.ts index e1d4fe16..90b5026e 100644 --- a/tests/codex-control-router.test.ts +++ b/tests/codex-control-router.test.ts @@ -20,7 +20,7 @@ type FixtureThread = { turnId?: string; }; -function fixtureClient(threads: FixtureThread[], sequenceFailure?: "active_turn_not_steerable") { +function fixtureClient(threads: FixtureThread[], sequenceFailure?: "active_turn_not_steerable" | "transport_failure") { let current = threads; const requestCalls: Array<{ method: string; params: Record }> = []; const sequenceCalls: Array<{ steps: CodexControlStep[]; options: CodexControlSequenceOptions }> = []; @@ -61,6 +61,9 @@ function fixtureClient(threads: FixtureThread[], sequenceFailure?: "active_turn_ }, async requestSequenceUntilTurnResolved(steps, options) { sequenceCalls.push({ steps, options }); + if (sequenceFailure === "transport_failure") { + throw new Error("socket closed after write"); + } if (sequenceFailure === "active_turn_not_steerable") { return { responses: [ @@ -253,6 +256,16 @@ test("delivery sends to idle targets and interrupt accepts the same opaque activ assert.equal(idleLive.status, "accepted"); assert.equal(idleFixture.requestCalls.some((call) => call.method === "turn/start"), true); assert.equal(idleFixture.sequenceCalls.length, 0); + const replay = await idleRouter.deliver({ + targetRef: idleRoute.target_ref!, + message: "Start work", + dryRun: false, + approvalAuditId: idleDryRun.approval_audit_id + }); + assert.equal(replay.status, "blocked"); + assert.equal(replay.control_sent, false); + assert.deepEqual(replay.reason_codes, ["approval_or_control_rejected"]); + assert.equal(idleFixture.requestCalls.filter((call) => call.method === "turn/start").length, 1); const activeRoute = await activeRouter.route({}); const interruptDryRun = await activeRouter.interrupt({ targetRef: activeRoute.target_ref! }); @@ -380,3 +393,50 @@ test("delivery returns the specific non-steerable blocker from Codex", async () rmSync(root, { recursive: true, force: true }); } }); + +test("delivery consumes approval before transport and reports an indeterminate disconnect without retrying", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-indeterminate-")); + const fixture = fixtureClient([{ + id: "thread-active", + name: "Active task", + state: "active", + turnId: "turn-active" + }], "transport_failure"); + const audit = createAuditStore(join(root, "audit.jsonl")); + const router = createCodexControlRouter({ + client: fixture.client, + control: createCodexControl({ audit, client: fixture.client }), + createRef: () => "lco_target_active" + }); + + try { + const route = await router.route({}); + const dryRun = await router.deliver({ + targetRef: route.target_ref!, + message: "Steer safely" + }); + const first = await router.deliver({ + targetRef: route.target_ref!, + message: "Steer safely", + dryRun: false, + approvalAuditId: dryRun.approval_audit_id + }); + assert.equal(first.status, "blocked"); + assert.equal(first.control_sent, null); + assert.deepEqual(first.reason_codes, ["control_attempt_indeterminate", "approval_consumed_do_not_retry"]); + assert.equal(fixture.sequenceCalls.length, 1); + + const replay = await router.deliver({ + targetRef: route.target_ref!, + message: "Steer safely", + dryRun: false, + approvalAuditId: dryRun.approval_audit_id + }); + assert.equal(replay.status, "blocked"); + assert.equal(replay.control_sent, false); + assert.deepEqual(replay.reason_codes, ["approval_or_control_rejected"]); + assert.equal(fixture.sequenceCalls.length, 1); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); diff --git a/tests/control-and-mcp.test.ts b/tests/control-and-mcp.test.ts index 9fd03fb7..6e680274 100644 --- a/tests/control-and-mcp.test.ts +++ b/tests/control-and-mcp.test.ts @@ -480,10 +480,20 @@ test("Codex control rejects failed same-connection sequence responses before liv /control sequence step failed.*thread\/resume/i ); - const auditRecords = readFileSync(audit.path, "utf8").split(/\r?\n/).filter(Boolean).map((line) => JSON.parse(line) as { live: boolean }); - assert.equal(auditRecords.length, 2); + const auditRecords = readFileSync(audit.path, "utf8").split(/\r?\n/).filter(Boolean).map((line) => JSON.parse(line) as { + approvalAuditId?: string; + approvalState?: string; + live: boolean; + }); + assert.equal(auditRecords.length, 4); assert.equal(auditRecords[0]?.live, false); assert.equal(auditRecords[1]?.live, false); + assert.equal(auditRecords[1]?.approvalState, "claimed"); + assert.equal(auditRecords[1]?.approvalAuditId, dryRun.approvalAuditId); + assert.equal(auditRecords[2]?.live, false); + assert.equal(auditRecords[3]?.live, false); + assert.equal(auditRecords[3]?.approvalState, "claimed"); + assert.equal(auditRecords[3]?.approvalAuditId, collisionDryRun.approvalAuditId); } finally { rmSync(root, { recursive: true, force: true }); } @@ -526,9 +536,16 @@ test("Codex control returns structured proof when active runtime posture blocks assert.match(blocked.proofState.callerInstruction, /control was not sent/); assert.equal((blocked.response as any).code, "safe_runtime_posture_unproven"); - const auditRecords = readFileSync(audit.path, "utf8").split(/\r?\n/).filter(Boolean).map((line) => JSON.parse(line) as { live: boolean }); - assert.equal(auditRecords.length, 1); + const auditRecords = readFileSync(audit.path, "utf8").split(/\r?\n/).filter(Boolean).map((line) => JSON.parse(line) as { + approvalAuditId?: string; + approvalState?: string; + live: boolean; + }); + assert.equal(auditRecords.length, 2); assert.equal(auditRecords[0]?.live, false); + assert.equal(auditRecords[1]?.live, false); + assert.equal(auditRecords[1]?.approvalState, "claimed"); + assert.equal(auditRecords[1]?.approvalAuditId, dryRun.approvalAuditId); } finally { rmSync(root, { recursive: true, force: true }); } From e472ad0b39cef070d8136228bdaa325e66352a23 Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 30 Jul 2026 18:10:53 +0700 Subject: [PATCH 3/5] fix: persist approval replay protection --- packages/adapters/src/index.ts | 20 ++++++++++++++++++-- packages/mcp-server/src/tools.ts | 3 +++ tests/codex-control-router.test.ts | 10 ++++++++++ 3 files changed, 31 insertions(+), 2 deletions(-) diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index dab447d6..bdc01b8f 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -671,7 +671,7 @@ export type AuditStore = Omit, "deriveSubkey fingerprintTextIfConfigured?(value: string): string | null; }; type ControlAuditStore = Pick & { - tail?(limit?: number): AuditRecord[]; + hasApprovalUse?(approvalAuditId: string): boolean; }; export type TargetControlExecuteSpec = { @@ -758,6 +758,19 @@ export function createAuditStore(path: string) { } return null; }, + hasApprovalUse(approvalAuditId: string): boolean { + if (!existsSync(path)) return false; + const lines = readFileSync(path, "utf8").split(/\r?\n/).filter(Boolean); + for (let index = lines.length - 1; index >= 0; index -= 1) { + try { + const parsed = JSON.parse(lines[index]!) as AuditRecord; + if (parsed.approvalAuditId === approvalAuditId) return true; + } catch { + // Ignore corrupt partial writes so earlier durable claims remain discoverable. + } + } + return false; + }, tail(limit = 20): AuditRecord[] { if (!existsSync(path)) return []; const boundedLimit = Math.max(1, Math.min(limit, 1000)); @@ -844,7 +857,10 @@ export function createTargetControl(options: { targetName: string; methodPolicy: throw new Error("approval_audit_id dry-run record expired"); } assertCodexControlTransportCapability(options.client, requiresSequence, Boolean(spec.turnResolution)); - if (options.audit.tail?.(1000).some((record) => record.approvalAuditId === previous.id)) { + if (!options.audit.hasApprovalUse) { + throw new Error("approval consumption lookup is unavailable"); + } + if (options.audit.hasApprovalUse(previous.id)) { throw new Error("approval_audit_id has already been used"); } options.audit.append({ diff --git a/packages/mcp-server/src/tools.ts b/packages/mcp-server/src/tools.ts index 1677bf3c..969a2edf 100644 --- a/packages/mcp-server/src/tools.ts +++ b/packages/mcp-server/src/tools.ts @@ -392,6 +392,9 @@ const metadataOnlyAudit: AuditStore = { find() { return null; }, + hasApprovalUse() { + return false; + }, tail() { return []; }, diff --git a/tests/codex-control-router.test.ts b/tests/codex-control-router.test.ts index 90b5026e..e7d604a8 100644 --- a/tests/codex-control-router.test.ts +++ b/tests/codex-control-router.test.ts @@ -426,6 +426,16 @@ test("delivery consumes approval before transport and reports an indeterminate d assert.deepEqual(first.reason_codes, ["control_attempt_indeterminate", "approval_consumed_do_not_retry"]); assert.equal(fixture.sequenceCalls.length, 1); + for (let index = 0; index < 1_000; index += 1) { + audit.append({ + action: "unrelated_dry_run", + target: `unrelated-${index}`, + paramsHash: `unrelated-${index}`, + approvalState: "dry_run", + live: false + }); + } + const replay = await router.deliver({ targetRef: route.target_ref!, message: "Steer safely", From 56200e77133c63557a23f5ee56d9afa68bd6f85a Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 30 Jul 2026 18:18:34 +0700 Subject: [PATCH 4/5] fix: keep daemon control proof fail-closed --- packages/adapters/src/codex-jsonrpc.ts | 22 ++++++++-- packages/adapters/src/index.ts | 23 ++++++++++ .../mcp-server/src/codex-control-router.ts | 7 +-- tests/bridge-safety-core.test.ts | 26 +++++++++++ tests/codex-control-router.test.ts | 43 ++++++++++++++++++- 5 files changed, 113 insertions(+), 8 deletions(-) diff --git a/packages/adapters/src/codex-jsonrpc.ts b/packages/adapters/src/codex-jsonrpc.ts index 878a451c..e2d25d58 100644 --- a/packages/adapters/src/codex-jsonrpc.ts +++ b/packages/adapters/src/codex-jsonrpc.ts @@ -106,6 +106,7 @@ export class CodexJsonRpcClient { async request(method: string, params: Record = {}): Promise { assertCodexMethodAllowed(method, this.surface); + this.resetObservationBuffers(); return this.requestRaw(method, params); } @@ -113,6 +114,7 @@ export class CodexJsonRpcClient { predicate: (notification: JsonRpcNotification) => boolean, options: { timeoutMs?: number; stopOnServerRequest?: boolean } = {} ): Promise { + this.resetObservationBuffers(); const transport = this.requireTransport(); const deadline = Date.now() + (options.timeoutMs ?? this.timeoutMs); const notifications: JsonRpcNotification[] = []; @@ -188,14 +190,26 @@ export class CodexJsonRpcClient { if (payload.id !== id) continue; if ("error" in payload) { - return { ok: false, error: JSON.stringify(redactValue(payload.error)), notifications: [...this.notifications] }; + return { ok: false, error: JSON.stringify(redactValue(payload.error)), notifications: this.publicNotifications() }; } if ("result" in payload) { - return { ok: true, result: payload.result, notifications: [...this.notifications] }; + return { ok: true, result: payload.result, notifications: this.publicNotifications() }; } - return { ok: true, result: payload, notifications: [...this.notifications] }; + return { ok: true, result: payload, notifications: this.publicNotifications() }; } - return { ok: false, error: `Timed out waiting for ${method}`, notifications: [...this.notifications] }; + return { ok: false, error: `Timed out waiting for ${method}`, notifications: this.publicNotifications() }; + } + + private resetObservationBuffers(): void { + this.notifications.length = 0; + this.serverRequests.length = 0; + } + + private publicNotifications(): JsonRpcNotification[] { + return this.notifications.map((notification) => ({ + method: notification.method, + params: {} + })); } private requireTransport(): JsonRpcTransport { diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index bdc01b8f..cf719d25 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -927,6 +927,29 @@ export function createTargetControl(options: { targetName: string; methodPolicy: if (isIndeterminateControlResponse(rawResponse)) { throw new Error("codex_control_attempt_indeterminate"); } + if (asRecord(rawResponse)?.ok === false) { + const response = sanitizeCodexControlResponse(rawResponse); + return { + action: spec.action, + threadId: spec.threadId, + live: true, + approvalAuditId: previous.id, + paramsHash, + messageHash, + method: spec.method, + methodSequence, + connectionScope, + controlSent: false, + expectedTurnId: spec.expectedTurnId, + proofState: liveProofState({ + method: spec.method, + methodSequence, + threadId: spec.threadId, + response + }), + response + }; + } const response = responseWithTurnResolution(rawResponse, sequenceResult?.turn); const liveRecord = options.audit.append({ action: spec.action, diff --git a/packages/mcp-server/src/codex-control-router.ts b/packages/mcp-server/src/codex-control-router.ts index 028a2ecb..a38302d7 100644 --- a/packages/mcp-server/src/codex-control-router.ts +++ b/packages/mcp-server/src/codex-control-router.ts @@ -299,17 +299,18 @@ function publicDeliveryResult( const live = result.live === true; const proof = recordValue(result.proofState); const completed = proof?.completed === true || proof?.status === "completed"; + const controlSent = result.controlSent === true; return { schema: "lco.codex.delivery.v1", - status: live ? completed ? "completed" : "accepted" : "dry_run_ready", + status: live ? controlSent ? completed ? "completed" : "accepted" : "blocked" : "dry_run_ready", action, target_ref: targetRef, live, - control_sent: result.controlSent === true, + control_sent: controlSent, ...(typeof result.approvalAuditId === "string" ? { approval_audit_id: result.approvalAuditId } : {}), ...(typeof result.paramsHash === "string" ? { params_hash: result.paramsHash } : {}), ...(typeof result.messageHash === "string" ? { message_hash: result.messageHash } : {}), - reason_codes: [], + reason_codes: live && !controlSent ? ["control_rejected"] : [], public_safe: true, raw_transcript_returned: false }; diff --git a/tests/bridge-safety-core.test.ts b/tests/bridge-safety-core.test.ts index f2925cdd..c4a43601 100644 --- a/tests/bridge-safety-core.test.ts +++ b/tests/bridge-safety-core.test.ts @@ -164,6 +164,32 @@ test("Codex JSON-RPC client initializes, sends initialized notification, buffers assert.equal(transport.closed, true); }); +test("persistent Codex JSON-RPC responses keep notifications operation-local and method-only", async () => { + const transport = new FakeTransport([ + { jsonrpc: "2.0", id: 1, result: { serverInfo: { name: "fake-codex" } } }, + { + jsonrpc: "2.0", + method: "item/agentMessage/delta", + params: { delta: "PRIVATE_NOTIFICATION_CANARY" } + }, + { jsonrpc: "2.0", id: 2, result: { data: [] } }, + { jsonrpc: "2.0", id: 3, result: { data: [] } } + ]); + const client = new CodexJsonRpcClient(() => transport, { timeoutMs: 50 }); + + await client.connect(); + const first = await client.request("thread/list", {}); + const second = await client.request("thread/list", {}); + await client.close(); + + assert.deepEqual(first.notifications, [ + { method: "item/agentMessage/delta", params: {} } + ]); + assert.equal(JSON.stringify(first).includes("PRIVATE_NOTIFICATION_CANARY"), false); + assert.deepEqual(second.notifications, []); + assert.equal(client.notifications.length, 0); +}); + test("Codex JSON-RPC client reports timeout and redacts JSON-RPC errors", async () => { const timeoutClient = new CodexJsonRpcClient(() => new FakeTransport([ { jsonrpc: "2.0", id: 1, result: {} } diff --git a/tests/codex-control-router.test.ts b/tests/codex-control-router.test.ts index e7d604a8..8dab88f3 100644 --- a/tests/codex-control-router.test.ts +++ b/tests/codex-control-router.test.ts @@ -20,7 +20,10 @@ type FixtureThread = { turnId?: string; }; -function fixtureClient(threads: FixtureThread[], sequenceFailure?: "active_turn_not_steerable" | "transport_failure") { +function fixtureClient( + threads: FixtureThread[], + sequenceFailure?: "active_turn_not_steerable" | "transport_failure" | "request_rejected" +) { let current = threads; const requestCalls: Array<{ method: string; params: Record }> = []; const sequenceCalls: Array<{ steps: CodexControlStep[]; options: CodexControlSequenceOptions }> = []; @@ -51,6 +54,9 @@ function fixtureClient(threads: FixtureThread[], sequenceFailure?: "active_turn_ }; } if (method === "turn/start") { + if (sequenceFailure === "request_rejected") { + return { ok: false, error: "turn rejected", notifications: [] }; + } return { ok: true, result: { turn: { id: "turn-new", status: "inProgress" } }, @@ -450,3 +456,38 @@ test("delivery consumes approval before transport and reports an indeterminate d rmSync(root, { recursive: true, force: true }); } }); + +test("idle delivery never reports a rejected one-shot turn start as accepted", async () => { + const root = mkdtempSync(join(tmpdir(), "lco-router-request-rejected-")); + const fixture = fixtureClient([{ + id: "thread-idle", + name: "Idle task", + state: "idle" + }], "request_rejected"); + const audit = createAuditStore(join(root, "audit.jsonl")); + const router = createCodexControlRouter({ + client: fixture.client, + control: createCodexControl({ audit, client: fixture.client }), + createRef: () => "lco_target_idle" + }); + + try { + const route = await router.route({}); + const dryRun = await router.deliver({ + targetRef: route.target_ref!, + message: "Start safely" + }); + const rejected = await router.deliver({ + targetRef: route.target_ref!, + message: "Start safely", + dryRun: false, + approvalAuditId: dryRun.approval_audit_id + }); + assert.equal(rejected.status, "blocked"); + assert.equal(rejected.control_sent, false); + assert.deepEqual(rejected.reason_codes, ["control_rejected"]); + assert.equal(fixture.requestCalls.filter((call) => call.method === "turn/start").length, 1); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); From cc971ef9628c61610c201f0cd56307abe3c17dba Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 30 Jul 2026 18:19:29 +0700 Subject: [PATCH 5/5] fix: classify direct control failures precisely --- packages/adapters/src/index.ts | 7 ++++--- tests/control-and-mcp.test.ts | 26 ++++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index cf719d25..fe1d02ce 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -921,7 +921,8 @@ export function createTargetControl(options: { targetName: string; methodPolicy: rawResponse = sequenceResult ? sequenceResult.responses.at(-1) ?? { ok: true } : await options.client.request(spec.method, spec.params); - } catch { + } catch (error) { + if (!isIndeterminateControlError(error)) throw error; throw new Error("codex_control_attempt_indeterminate"); } if (isIndeterminateControlResponse(rawResponse)) { @@ -1427,13 +1428,13 @@ function isIndeterminateControlResponse(value: unknown): boolean { const response = asRecord(value); if (response?.ok !== false) return false; const error = typeof response.error === "string" ? response.error : JSON.stringify(response.error ?? ""); - return /timed out waiting|connection (?:failed|closed|lost)|socket|hang up|econnreset|broken pipe/i.test(error); + return /timed out waiting|connection (?:failed|closed|lost)|socket (?:hang up|closed|is not open)|websocket (?:closed|is not open)|econnreset|epipe|broken pipe/i.test(error); } function isIndeterminateControlError(error: unknown): boolean { const message = error instanceof Error ? error.message : String(error); return message === "codex_control_attempt_indeterminate" - || /timed out waiting|connection (?:failed|closed|lost)|socket|hang up|econnreset|broken pipe|websocket/i.test(message); + || /timed out waiting|connection (?:failed|closed|lost)|socket (?:hang up|closed|is not open)|websocket (?:closed|is not open)|econnreset|epipe|broken pipe/i.test(message); } function safeRuntimeBlockFromSequence(sequence: CodexControlSequenceResult | undefined): Record | null { diff --git a/tests/control-and-mcp.test.ts b/tests/control-and-mcp.test.ts index 6e680274..fabfb1df 100644 --- a/tests/control-and-mcp.test.ts +++ b/tests/control-and-mcp.test.ts @@ -499,6 +499,32 @@ test("Codex control rejects failed same-connection sequence responses before liv } }); +test("Codex direct control keeps deterministic pre-send errors distinct from indeterminate delivery", async () => { + const root = mkdtempSync(join(tmpdir(), "loo-control-direct-error-classification-")); + const audit = createAuditStore(join(root, "audit.jsonl")); + const control = createCodexControl({ + audit, + client: { + request: async () => { + throw new Error("Codex daemon transport requires an absolute Unix socket path"); + } + } + }); + + try { + const dryRun = await control.startThread({ dryRun: true }); + await assert.rejects( + () => control.startThread({ + dryRun: false, + approvalAuditId: dryRun.approvalAuditId + }), + /requires an absolute Unix socket path/ + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + test("Codex control returns structured proof when active runtime posture blocks control before send", async () => { const root = mkdtempSync(join(tmpdir(), "loo-control-runtime-posture-block-")); const audit = createAuditStore(join(root, "audit.jsonl"));