huangruiteng · Oct 23, 2026 · Oct 23, 2026
@@ -480,6 +480,11 @@ an existing business result. Real File/SQLite CLI coverage exercises generated c
 rebinding, one writeback/debit, preserved initial replan and settled replay.
 This source-path repair does not qualify installed host adoption or
 complete-cost acceptance.
+An explicitly selected CLI executable now remains in generated selection,
+required-read, scheduler and settlement/recovery routes through the shared
+invocation transport. The default route is unchanged; choosing an executable
+adds no admission or execution authority. Candidate installation and actual
+receiver adoption remain separate from this command-route qualification.
 Neither today's large quota packet nor the smaller TurnEnvelope is a target
 shape merely because of its size. Optional memory participation now has one
 compact, signed envelope projection; the Codex CLI adapter preserves default-off

@@ -332,6 +332,9 @@ typed receipt owner 保留原回写与结算路径；即使工作已 deferred
 准入。原语义重规划轮次仍保留该决策，不会被误当成已有业务结果。真实 File/SQLite CLI 验证
 返回命令、拒绝换绑、一次回写与扣额、初始重规划及已结算重放。此源码路径修复不代表
 安装宿主采用或完整成本验收。
+显式选择的 CLI 程序现通过共享 invocation transport 保留到生成的工作选择、必读、
+调度及结算/恢复命令。默认路由不变；选择程序不新增准入或执行权限。候选安装和
+接收者实际采用仍须与命令路由资格分别验收。
 现有大 quota packet 与小 TurnEnvelope
 都不是仅凭大小就合理的目标形态。可选 memory 参与事实现进入一个有界、签名覆盖的
 envelope 投影，Codex CLI 保持关闭隔离；这不代表安装态 heartbeat/App 或模型收益

@@ -25,9 +25,22 @@ unavailable contract still forbids delivery and spending. This is not permission
 to bypass a gate, retry indefinitely, or disable a healthy automation.
 
 For a one-agent trial, pass `--cli-bin loopx-canary` to preview and apply. The
-bootstrap and the thin prompt's generated commands both use that executable;
-other automations continue to use their existing runtime. Do not promote the
-canary as the global default merely to test one task.
+bootstrap, guard, work selection, required reads and settlement/receipt-recovery commands
+retain that executable. Generated commands carry the global `--cli-bin PROGRAM`
+option so the next process preserves the route too. For a direct guard, use
+`loopx-canary --cli-bin loopx-canary --format json quota should-run ...`.
+`PROGRAM` is one executable name or path, not a shell command. The option only
+selects generated command routing; it executes nothing and changes no authority,
+host configuration or permissions. Without an explicit route, commands retain
+the existing `loopx` default. The route is scoped to the current invocation and
+does not leak into a later CLI call.
+
+Native scheduler follow-up hints expose `cli_bin` for a custom route; invoke
+that executable with its `cli_args`. Keep the same executable through an admitted
+Turn's writeback and spend, including receipt recovery. Other automations retain
+their runtime. Do not promote the canary as the global default merely to test one
+task. Installed identity, receiver adoption and subsequent useful work remain
+separate acceptance evidence.
 
 The owner is the existing heartbeat/upgrade boundary; there is no new optional
 capability or extension provider. SQLite/TOML handling is a local host adapter,

@@ -7,6 +7,7 @@
 import re
 import shlex
 
+from ...command_invocation import loopx_cli_argv
 from ...file_lock import exclusive_file_lock
 from ...control_plane.collaboration import conversation_scope
 from ...control_plane.collaboration.source_grant_observation import (
@@ -281,7 +282,7 @@ def produce():
 
     command = shlex.join(
         [
-            "loopx",
+            *loopx_cli_argv(),
             "--registry",
             str(registry_path),
             "--runtime-root",

@@ -3,6 +3,7 @@
 import shlex
 from typing import Any
 
+from ...command_invocation import loopx_cli_argv
 from ...agent_registry import load_goal_from_registry, registered_agent_ids_for_goal
 from ...control_plane.effect_runtime import effect_runtime_result
 
@@ -85,7 +86,7 @@ def produce():
             "provider_payload_returned": False, "error_code": None,
         }
 
-    command = shlex.join(["loopx", "--registry", str(registry_path), "--runtime-root",
+    command = shlex.join([*loopx_cli_argv(), "--registry", str(registry_path), "--runtime-root",
         str(runtime_root), "semantic-preference", "agent", "read", "--goal-id", goal_id,
         "--agent-id", agent_id, "--format", "json"])
     hook = TurnStartHookRegistration(

@@ -214,6 +214,7 @@
     LoopXArgumentParser,
     add_subcommand_format,
     build_cli_parser,
+    cli_invocation_route,
     dispatch_common_command,
     enforce_native_controller_guard,
     output_format,
@@ -390,6 +391,12 @@ def build_parser() -> LoopXArgumentParser:
 
 
 def main(argv: list[str] | None = None) -> int:
+    raw_argv = sys.argv[1:] if argv is None else list(argv)
+    with cli_invocation_route(raw_argv):
+        return _main(argv)
+
+
+def _main(argv: list[str] | None = None) -> int:
     raw_argv = sys.argv[1:] if argv is None else list(argv)
     if top_level_help_requested(raw_argv):
         print(render_concise_help(sys.argv[0] if argv is None else "loopx"), end="")

@@ -5,6 +5,7 @@
 import os
 import sys
 from collections.abc import Callable
+from contextlib import contextmanager
 from pathlib import Path
 
 from . import __version__
@@ -15,7 +16,7 @@
 HOST_GLOBAL_REGISTRY_SELECTOR = "@host-global"
 
 
-GLOBAL_OPTIONS_WITH_VALUE = frozenset({"--registry", "--runtime-root", "--format"})
+GLOBAL_OPTIONS_WITH_VALUE = frozenset({"--registry", "--runtime-root", "--format", "--cli-bin"})
 GLOBAL_OPTIONS_WITH_EQUALS = tuple(
 	f"{option}=" for option in sorted(GLOBAL_OPTIONS_WITH_VALUE)
 )
@@ -144,6 +145,11 @@ def build_cli_parser(
 	)
 	parser.add_argument("--runtime-root", help="Override registry common_runtime_root.")
 	parser.add_argument("--format", choices=["markdown", "json"])
+	from .command_invocation import validate_cli_bin
+	parser.add_argument(
+		"--cli-bin", dest="control_cli_bin", type=validate_cli_bin,
+		help="Preserve this executable name/path in generated control commands; does not execute it or change permissions.",
+	)
 	return parser, parser.add_subparsers(dest="command", required=True)
 
 
@@ -258,6 +264,29 @@ def _top_level_command(argv: list[str]) -> str | None:
 	return None
 
 
+@contextmanager
+def cli_invocation_route(argv: list[str]):
+	"""Read the global route only; the canonical parser still validates argv."""
+	from .command_invocation import loopx_cli_route
+	program = None
+	index = 0
+	while index < len(argv):
+		value = argv[index]
+		option = value.split("=", 1)[0]
+		if option not in GLOBAL_OPTIONS_WITH_VALUE:
+			break
+		if "=" in value:
+			argument = value.split("=", 1)[1]
+		else:
+			index += 1
+			argument = argv[index] if index < len(argv) else None
+		if option == "--cli-bin":
+			program = argument
+		index += 1
+	with loopx_cli_route(program):
+		yield
+
+
 def _build_selected_parser(command: str) -> LoopXArgumentParser:
 	parser, subparsers = build_cli_parser(parser_class=_SelectedArgumentParser)
 	if command in _STATUS_COMMANDS:
@@ -580,7 +609,8 @@ def main(argv: list[str] | None = None) -> int:
 	code = 1
 	error = None
 	try:
-		code = _run_command(command, raw_argv)
+		with cli_invocation_route(raw_argv):
+			code = _run_command(command, raw_argv)
 		return code
 	except BaseException as exc:
 		error = exc

@@ -1,9 +1,38 @@
 from __future__ import annotations
 
 import os
+from contextlib import contextmanager
+from contextvars import ContextVar
 from pathlib import Path
 import shutil
-from typing import Mapping, Sequence
+from typing import Iterator, Mapping, Sequence
+
+
+_loopx_cli_bin: ContextVar[str] = ContextVar("loopx_cli_bin", default="loopx")
+
+
+def validate_cli_bin(value: str) -> str:
+    """Accept one executable name/path, never a shell command or empty route."""
+    if not value.strip() or value.startswith("-") or any(ord(c) < 32 for c in value):
+        raise ValueError("--cli-bin requires one non-empty executable name or path")
+    return value
+
+
+@contextmanager
+def loopx_cli_route(cli_bin: str | None) -> Iterator[None]:
+    """Bind only this invocation's command rendering; retain no host policy."""
+    token = _loopx_cli_bin.set(cli_bin if cli_bin is not None else "loopx")
+    try:
+        yield
+    finally:
+        _loopx_cli_bin.reset(token)
+
+
+def loopx_cli_argv(cli_bin: str | None = None) -> list[str]:
+    program = validate_cli_bin(cli_bin if cli_bin is not None else _loopx_cli_bin.get())
+    # The executable selects this process; the option preserves that selection
+    # when its returned commands start another process. Neither grants authority.
+    return [program] if program == "loopx" else [program, "--cli-bin", program]
 
 
 def resolve_command_path(

@@ -6,6 +6,7 @@
 import shlex
 from typing import Any
 
+from ...command_invocation import loopx_cli_argv
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ..settlement_driver import decode_settlement_result
 from .accounting_admission import quota_accounting_admission
@@ -186,7 +187,7 @@ def attach_settlement_progress(
     identity = readback.identity.value
     if payload.get("ok") is not True or progress.get("next_step") != "quota_spend" or identity is None:
         return
-    prefix = "loopx"
+    prefix = shlex.join(loopx_cli_argv())
     if registry_path is not None:
         prefix += f" --registry {shlex.quote(str(registry_path))}"
     if runtime_root is not None:

@@ -431,6 +431,8 @@ function scheduler(payload: JsonObject, turn: ReturnType<typeof interpretQuotaSh
     app.stateful_backoff = compactState;
   }
   const ack = object(sourceApp.ack_hint);
+  const cliBin = ack.cli_bin ?? object(sourceApp.failure_hint).cli_bin;
+  if (typeof cliBin === "string") app.cli_bin = cliBin;
   const cliArgs = executableCliArgs(ack.cli_args);
   if (cliArgs.length > 0) {
     app.ack_cli_args = cliArgs;

@@ -59,6 +59,7 @@ function parseArgs(argv: string[]): ParsedArgs {
     "--format",
     "--registry",
     "--runtime-root",
+    "--cli-bin", // Invocation routing only; never a scheduler authority fact.
     "--goal-id",
     "--agent-id",
     "--surface",
@@ -95,6 +96,11 @@ function parseArgs(argv: string[]): ParsedArgs {
       if (!value && option !== "--app-automation-current-rrule") {
         throw new EffectRuntimeRequestError(`${option} must not be empty`, "invalid_scheduler_followup_argv");
       }
+      if (option === "--cli-bin" && (!value.trim() || value.startsWith("-") || /[\x00-\x1f]/.test(value))) {
+        throw new EffectRuntimeRequestError(
+          "--cli-bin requires one non-empty executable name or path", "invalid_scheduler_followup_argv",
+        );
+      }
       if (option === FACTS_FLAG) chunks.push(value);
       else if (option !== "--available-capability" && option !== "--registry") {
         values[option] = value;

@@ -7,6 +7,7 @@
 from dataclasses import asdict
 from typing import Any
 
+from ...command_invocation import loopx_cli_argv
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ..quota.decision_summary import compact_quota_decision
 from ..runtime.time import now_utc, utc_isoformat, parse_timestamp
@@ -125,8 +126,12 @@ def _encode_followup_args(
             native_args.extend(encoded)
         else:
             raise RuntimeError("TypeScript scheduler CLI argument shape mismatch")
+    route = loopx_cli_argv()
+    if len(route) > 1:
+        args = [*route[1:], *args]
     return {
         **packet,
+        **({"cli_bin": route[0]} if len(route) > 1 else {}),
         "cli_args": _bounded_scheduler_followup_cli_args(args, native_args=native_args),
     }
 

@@ -4,13 +4,14 @@
 from collections.abc import Mapping
 from typing import Any
 
+from ...command_invocation import loopx_cli_argv
 from ..agents.capability_gate import runtime_capabilities_for_cli_projection
 from ..todos.contract import normalize_todo_id
 from ..quota.effective_action import EffectiveAction
 
 
 def render_cli_command_prefix(*, runtime_root: str | None = None, registry_path: str | None = None) -> str:
-    return ("loopx" + (f" --registry {shlex.quote(str(registry_path))}" if registry_path else "")
+    return (shlex.join(loopx_cli_argv()) + (f" --registry {shlex.quote(str(registry_path))}" if registry_path else "")
         + (f" --runtime-root {shlex.quote(str(runtime_root))}" if runtime_root else ""))
 
 
@@ -116,7 +117,7 @@ def apply_action_selection_cli_gate(
             format_index >= 1
             and command_tokens[format_index : format_index + 2]
             == ["--format", "json"]
-            and command_tokens[0] == "loopx"
+            and command_tokens[0] == loopx_cli_argv()[0]
         ):
             command_args_template = shlex.join(command_tokens[format_index + 2 :])
             route_prefix = shlex.join(command_tokens[: format_index + 2])

@@ -68,7 +68,7 @@ def _native_scheduler_followup_argv(raw_argv: list[str]) -> list[str] | None:
     if os.environ.get("LOOPX_KUNLUNCODE_OUTER_CONTROLLER") == "1":
         return None
 
-    value_options = {"--format", "--registry", "--runtime-root"}
+    value_options = {"--format", "--registry", "--runtime-root", "--cli-bin"}
     positionals: list[str] = []
     skip_value = False
     for value in raw_argv:

@@ -510,7 +510,7 @@ def render_concise_help(program: str = "loopx") -> str:
             "  KunlunCode     run loopx-kunluncode connect, then loopx-kunluncode run.",
             "  Other agents   need a CLI/task/automation/loop hook, or run LoopX manually.",
             "",
-            "Global options: --registry PATH   --runtime-root PATH   --format markdown|json",
+            "Global options: --registry PATH   --runtime-root PATH   --format markdown|json   --cli-bin PROGRAM",
             "More:",
             "  loopx commands                 Show grouped command reference.",
             "  loopx <command> --help         Show flags for one command.",

@@ -6,6 +6,7 @@
 from pathlib import Path
 from typing import Any
 
+from .command_invocation import loopx_cli_argv
 from .control_plane.projects.identity import default_goal_id
 from .control_plane.projects.registry_codec import load_registry
 from .paths import (
@@ -67,7 +68,8 @@ def render_cli_command_prefix(
     runtime_root: str | Path | None = None,
     registry_path: str | Path | None = None,
 ) -> str:
-    prefix = shell_arg_for_host(cli_bin)
+    prefix = " ".join(shell_arg_for_host(arg) for arg in loopx_cli_argv(
+        None if cli_bin == "loopx" else cli_bin))
     if runtime_root is not None:
         prefix += f" --runtime-root {shell_arg_for_host(str(runtime_root))}"
     if registry_path is not None: