Add optional bilingual CrystalPilot plugins for Argus 0.1.3

.gitignore

@@ -131,3 +131,6 @@ docs/Argus_BP*
 # proprietary binary/npm release staging
 /dist-binary/
 /.pyinstaller/
+
+# Optional plugin release assets are published separately.
+/dist-plugins/

argus_skill/adapters/agent_cli_backend/_core.py

@@ -420,6 +420,14 @@ def _close_io_context(self, call_id: str) -> None:
         self._io_logger.close(call_id)
 
     def _stream_event_callback(self, stream: str, line: str) -> None:
+        options = getattr(self, "_plugin_execution_options", None)
+        if options is not None and getattr(options, "extension_env", None):
+            from ...core.workbench_plugins import observe_plugin_stream
+            try:
+                from ...core.secret_guard import redact_secrets_text
+                observe_plugin_stream(options, stream, redact_secrets_text(line, known_values=self._known_secret_values))
+            except Exception:
+                log.exception("Plugin stream projection failed")
         self._repeated_tool_call_guard.observe(stream, line)
         self._io_logger.stream_event_callback(
             stream,
@@ -488,6 +496,9 @@ def _translate_options(self, options: RunnerOptions):
             watchdog_soft_idle_seconds=soft_idle,
             watchdog_hard_idle_seconds=hard_idle,
         )
+        for plugin_field in ("trusted_extensions", "trusted_tool_names", "extension_env"):
+            if plugin_field in option_fields:
+                kwargs[plugin_field] = getattr(options, plugin_field, None)
         if "watchdog_stalled_idle_seconds" in option_fields:
             kwargs["watchdog_stalled_idle_seconds"] = stalled_idle
         # Forward live_search ONLY when the target RunnerOptions supports it —
@@ -565,7 +576,7 @@ def _premium_delta_for_thread(
 # --- Convenience factory ---------------------------------------------------
 
 
-def build_agent_cli_backend_from_env() -> AgentCliBackend:
+def build_agent_cli_backend_from_env(*, role: str | None = None) -> AgentCliBackend:
     """Build a AgentCliBackend from environment variables.
 
     Honours:
@@ -587,9 +598,12 @@ def build_agent_cli_backend_from_env() -> AgentCliBackend:
     import shlex
 
     backend = os.environ.get("ARGUS_SKILL_RUNNER_BACKEND", "").strip() or "codex"
+    if role:
+        from ...core.role_config import resolve_role_config
+        backend = resolve_role_config(role).backend
     from ...core.knobs import resolve_runner_bin_setting
 
-    runner_bin = resolve_runner_bin_setting(backend=backend) or None
+    runner_bin = resolve_runner_bin_setting(role=role, backend=backend) or None
     raw_extra = os.environ.get("ARGUS_SKILL_RUNNER_EXTRA_ARGS", "").strip()
     extra = _strip_legacy_codex_profile_args(shlex.split(raw_extra) if raw_extra else None)
     return AgentCliBackend(

argus_skill/adapters/agent_cli_backend/_exec.py

@@ -48,6 +48,18 @@ def execute(
     # The generated command, reservation, and settled usage record therefore
     # share one model id instead of independently guessing after the call.
     options = backend._resolve_execution_options(options)
+    from ...core.workbench_plugins import prepare_plugin_run
+    prompt, options = prepare_plugin_run(prompt, options,
+        backend=backend._runner.backend, run_label=run_label)
+    backend._plugin_execution_options = options
+    try:
+        return _execute_prepared(backend, prompt=prompt, options=options, run_label=run_label, resume_thread_id=resume_thread_id)
+    finally:
+        from ...core.workbench_plugins import finish_plugin_run
+        finish_plugin_run(options)
+
+
+def _execute_prepared(backend, *, prompt, options, run_label, resume_thread_id):
     # Reset per-call: the flag is checked AFTER this call completes,
     # so stale True from a previous call cannot stick across missions.
     backend._auth_failure_detected = False
@@ -59,6 +71,10 @@ def execute(
     )
     if usage_project_root is None and log_path is not None:
         usage_project_root = log_path.parent
+    from ...core.workbench_plugins import plugin_accounting_root
+    accounting_root = plugin_accounting_root(usage_project_root)
+    if accounting_root is not None:
+        usage_global_root = accounting_root
     io_context = backend._io_logger.start_call(
         call_id=call_id,
         run_label=run_label,

argus_skill/agent_cli/_acp_routing.py

@@ -97,6 +97,10 @@ def _acp_enabled(
         ``ARGUS_SKILL_COPILOT_ACP_LABELS`` overrides the default label set. All
         engineer/reviewer/planner/mission turns stay on the CLI ``Popen`` path.
         """
+        # Session-bound MCP configurations must start their own CLI process;
+        # a warm ACP process cannot inherit this turn's scoped capability.
+        if options is not None and getattr(options, "extension_env", None):
+            return False
         if self.backend != BACKEND_COPILOT or not run_label:
             return False
         raw_flag = os.environ.get("ARGUS_SKILL_COPILOT_ACP")

argus_skill/agent_cli/_prompt_delivery.py

@@ -347,6 +347,13 @@ def _child_env(
             env["GH_CONFIG_DIR"] = str(
                 Path(tempfile.gettempdir()) / "argus-no-gh-auth"
             )
+        plugin_env = getattr(options, "extension_env", None)
+        if plugin_env and not options.disable_tools:
+            env = dict(os.environ) if env is None else env
+            for key, value in plugin_env.items():
+                if not key.startswith("ARGUS_PLUGIN_"):
+                    raise ValueError("invalid plugin environment key")
+                env[key] = str(value)
         repaired = runner_child_environment(
             executable or getattr(self, "agent_bin", ""),
             env=env,

argus_skill/agent_cli/_sandbox_commands.py

@@ -368,6 +368,9 @@ def _build_codex_command(
         # modified, and later explicit extra args may still opt back in.
         command.extend(["-c", "notify=[]"])
         if options.disable_tools:
+            import dataclasses
+            options = dataclasses.replace(options, sandbox_mode="read-only", dangerous_yolo=False,
+                                          full_auto=False, skip_git_repo_check=True)
             # Stateless Manager/Planner control calls need the operator's model
             # provider and auth, but not interactive plugins, MCP servers, JS
             # REPL startup or project exec-policy rules. Keeping the base config
@@ -382,6 +385,11 @@ def _build_codex_command(
                 "plugins={}",
                 "-c",
                 "features.js_repl=false",
+                "-c", "features.shell_tool=false",
+                "-c", "features.apply_patch_freeform=false",
+                "-c", "features.multi_agent=false",
+                "-c", "tools.view_image=false",
+                "-c", "project_doc_max_bytes=0",
                 "-c",
                 'web_search="disabled"',
             ])
@@ -551,13 +559,19 @@ def _build_copilot_command(
         if options.disable_tools:
             command.append(f"--available-tools={_COPILOT_NO_TOOLS_SENTINEL}")
         elif options.sandbox_mode == "read-only":
-            tools = "view,rg,glob"
+            tools = ",".join(["view", "rg", "glob", *(getattr(options, "trusted_tool_names", None) or [])])
             if review_output:
                 tools += ",argus_review-read_review,argus_review-write_review"
             command.extend([
                 "--available-tools", tools,
                 "--allow-tool", "view,rg,glob",
             ])
+            if getattr(options, "trusted_tool_names", None):
+                permissions = []
+                for name in options.trusted_tool_names:
+                    server, separator, tool = name.partition("-")
+                    permissions.append(f"{server}({tool})" if separator else name)
+                command.extend(["--allow-tool", ",".join(permissions)])
             if review_output:
                 command.extend(["--allow-tool", "argus_review"])
         elif options.dangerous_yolo:
@@ -702,14 +716,17 @@ def _build_pi_command(
         ])
         for path in options.skill_paths or []:
             command.extend(["--skill", path])
+        if not options.disable_tools:
+            for path in getattr(options, "trusted_extensions", None) or []:
+                command.extend(["--extension", path])
         if options.model:
             command.extend(["--model", _pi_model(options.model)])
         if options.reasoning_effort:
             command.extend(["--thinking", options.reasoning_effort])
         if options.disable_tools:
             command.append("--no-tools")
         elif options.sandbox_mode == "read-only":
-            command.extend(["--tools", "read,grep,find,ls"])
+            command.extend(["--tools", ",".join(["read", "grep", "find", "ls", *(getattr(options, "trusted_tool_names", None) or [])])])
         merged_extra_args = [*self.default_extra_args]
         if options.extra_args:
             merged_extra_args.extend(options.extra_args)
@@ -720,6 +737,18 @@ def _build_pi_command(
             )
         if merged_extra_args:
             command.extend(merged_extra_args)
+        # SELF/other role profiles can append their own --tools allowlist.
+        # Preserve that builtin policy while retaining the explicitly bound
+        # plugin tools; otherwise the later flag silently hides the extension.
+        if not options.disable_tools:
+            trusted = getattr(options, "trusted_tool_names", None) or []
+            for index, argument in enumerate(command):
+                if argument == "--tools" and index + 1 < len(command):
+                    names = command[index + 1].split(",")
+                    command[index + 1] = ",".join(dict.fromkeys([*names, *trusted]))
+                elif argument.startswith("--tools="):
+                    names = argument.split("=", 1)[1].split(",")
+                    command[index] = "--tools=" + ",".join(dict.fromkeys([*names, *trusted]))
         if resume_thread_id:
             command.extend(["--session", resume_thread_id])
         # Pi reads non-TTY stdin into the initial message in JSON mode. Keeping

argus_skill/agent_cli/agent_cli_runner.py

@@ -89,6 +89,9 @@ class RunnerOptions:
     # (the whole daemon) byte-for-byte unchanged; only the Manager chat
     # front-door sets it, to stream the reply live.
     on_agent_message: Callable[[str], None] | None = None
+    trusted_extensions: list[str] | None = None
+    trusted_tool_names: list[str] | None = None
+    extension_env: dict[str, str] | None = None
 
 
 class AgentCliRunner(

argus_skill/apps/_runtime_execute.py

@@ -15,6 +15,7 @@
 import os
 import shutil
 import stat
+import tempfile
 import time
 from pathlib import Path
 
@@ -110,6 +111,44 @@ def _is_unaliased_regular_file(cls, path: Path) -> bool:
         except OSError:
             return False
 
+    @classmethod
+    def _detach_packaged_skill_hardlink(cls, path: Path) -> None:
+        """Give an installed Skill its own inode before establishing the guard.
+
+        uv legitimately hardlinks wheel resources from its cache. Replacing our
+        directory entry preserves those cached bytes and other environments;
+        in-place writes or weakening the execution-time alias check would not.
+        Symlinks, junctions and redirected ancestors remain disallowed.
+        """
+        if cls._is_link_or_reparse_point(path) or cls._has_linked_ancestor(path):
+            return
+        before = path.stat()
+        if not stat.S_ISREG(before.st_mode) or before.st_nlink <= 1:
+            return
+        content = path.read_bytes()
+        fd, filename = tempfile.mkstemp(prefix=".argus-skill-", dir=path.parent)
+        temporary = Path(filename)
+        try:
+            with os.fdopen(fd, "wb") as stream:
+                stream.write(content)
+                stream.flush()
+                os.fsync(stream.fileno())
+            temporary.chmod(stat.S_IMODE(before.st_mode))
+            if cls._is_link_or_reparse_point(path) or cls._has_linked_ancestor(path):
+                raise OSError(f"protected Skill path changed while preparing: {path}")
+            current = path.stat()
+            identity = lambda value: (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns)
+            if identity(current) != identity(before) or path.read_bytes() != content:
+                # Another startup may already have detached exactly these bytes.
+                if cls._is_unaliased_regular_file(path) and path.read_bytes() == content:
+                    return
+                raise OSError(f"protected Skill changed while preparing: {path}")
+            os.replace(temporary, path)
+            if not cls._is_unaliased_regular_file(path) or path.read_bytes() != content:
+                raise OSError(f"protected Skill private copy did not verify: {path}")
+        finally:
+            temporary.unlink(missing_ok=True)
+
     @classmethod
     def _remove_pipeline_state_replacement(cls, path: Path) -> None:
         if path.is_symlink():
@@ -161,6 +200,11 @@ def _restore_pipeline_state(
         if snapshot_error:
             return True, snapshot_error, False
         try:
+            # A fresh mission may have neither a pipeline file nor its parent.
+            # Restoration must not create that parent and then accuse the
+            # mission of creating formal state that never existed.
+            if not existed and not os.path.lexists(path.parent):
+                return False, "", True
             if cls._has_linked_ancestor(path.parent):
                 raise OSError(
                     f"formal pipeline state ancestor was replaced: {path.parent}"
@@ -246,9 +290,14 @@ def _snapshot_playground_skill_files(
             protected_paths = list(canonical_paths)
             for parent in dict.fromkeys(path.parent for path in canonical_paths):
                 for sibling in sorted(parent.iterdir()):
+                    # Another mission may be detaching a cache hardlink now;
+                    # its short-lived private copy is not a packaged Skill.
+                    if sibling.name.startswith(".argus-skill-"):
+                        continue
                     if sibling not in protected_paths and sibling.is_file():
                         protected_paths.append(sibling)
             for path in protected_paths:
+                cls._detach_packaged_skill_hardlink(path)
                 if (
                     cls._is_link_or_reparse_point(path.parent)
                     or not path.parent.is_dir()
@@ -944,6 +993,7 @@ def _invoke_execute_loop(
         expected_playground_path = self._canonical_playground_skill_path(
             skill_snapshots
         )
+        execution_started = False
         try:
             if pipeline_state_snapshot[3]:
                 raise RuntimeError(
@@ -962,6 +1012,7 @@ def _invoke_execute_loop(
                         or "canonical Playground Engineer digest is unavailable"
                     )
                 )
+            execution_started = True
             self._run_bounded_planning(
                 ex_state,
                 sink=sink,
@@ -1087,6 +1138,10 @@ def _pre_settlement_guard(
                     if hasattr(ex_state.outcome, "final_message"):
                         ex_state.outcome.final_message = isolation_reason
         except BaseException as execution_error:
+            if not execution_started:
+                # Preflight never ran the planner/engineer. Retain its actual
+                # error; a failed snapshot is not an execution-time mutation.
+                raise
             changed, isolation_reason, restoration_ok = self._restore_playground_boundaries(
                 pipeline_state_snapshot,
                 skill_snapshots,

argus_skill/core/models.py

@@ -110,6 +110,9 @@ class RunnerOptions:
     # role turns are entirely unaffected. A callback exception never breaks the
     # turn (it is swallowed by the runner).
     on_agent_message: Callable[[str], None] | None = None
+    trusted_extensions: list[str] | None = None
+    trusted_tool_names: list[str] | None = None
+    extension_env: dict[str, str] | None = None
 
 
 @dataclass