Implementation Plan: Disambiguate run_skill from Named Local Skill Loading

src/autoskillit/server/tools/tools_execution.py

@@ -602,51 +602,23 @@ async def run_skill(
     closure_target_sha: str = "",
     ctx: Context = CurrentContext(),
 ) -> str:
-    """Run a Claude Code headless session with a skill command.
-
-    Returns JSON with: success, result, session_id, subtype, is_error, exit_code,
-    needs_retry, retry_reason. When needs_retry is true, retry_reason is:
-    - "resume": context/turn limit hit — partial progress on disk, route to on_context_limit.
-    - "drain_race": channel confirmed completion but stdout not fully flushed — route to
-      on_context_limit (same as resume).
-    - "completed_no_flush": session exited with empty stdout but write evidence confirms work was
-      performed — route to on_context_limit (same as drain_race/resume).
-    - "empty_output": session exited cleanly with no output AND no write evidence — no partial
-      progress, route to on_failure.
-    - "path_contamination": session wrote files outside its working directory — route to
-      on_failure.
-    - "early_stop": model stopped before completion marker — route to on_failure.
-    - "zero_writes": skill made no writes despite write expectation — route to on_failure.
-    - "thinking_stall": model produced thinking blocks only, no text/tool output — route to
-      on_context_limit if lifespan_started, else on_failure.
-    - "contract_recovery": model completed and wrote artifacts but structured output failed
-      pattern validation and nudge could not recover — route to on_context_limit if
-      has_progress_evidence, else on_failure.
-    - "rate_limited": transient HTTP 429 or rate-limit text signal — route to on_rate_limit
-      (falls back to on_context_limit if on_rate_limit is absent).
-    - "stale": session went stale (no output progress) — decrement retries counter and
-      re-execute the same step if retries remain, else on_failure.
-    - "idle_stall": stdout idle watchdog killed the session — route to on_context_limit if
-      lifespan_started, else on_failure.
-    - "clone_contamination": session wrote to a contaminated clone directory — route to
-      on_failure.
-    - "budget_exhausted": session exceeded its budget allocation — route to on_failure.
-
-    This is the correct MCP tool to delegate work to a headless session during
-    pipeline execution. NEVER use native tools (Read, Grep, Glob, Edit, Write,
-    Bash, Agent, WebFetch, WebSearch, NotebookEdit) from the orchestrator.
-    All code changes, investigation, and research happen through the headless
-    session launched by this tool.
-
-    Use this for all skill sessions, including long-running ones that may hit the
-    context limit. The 2-hour timeout is the default. When needs_retry is true,
-    route to the appropriate resume step (e.g., retry-worktree) rather than
-    re-running this step from scratch.
+    """Delegate one already-selected recipe step to a separate L1 headless coding-agent worker.
+
+    Use this tool only when a headless recipe orchestrator operating at L2 or an
+    interactive AutoSkillit cook/order session intends separate-worker delegation. The
+    recipe step must already be selected before this call.
+
+    When a user names or asks to use an available local skill for the current interactive
+    conversation, load and follow its SKILL.md in the current interactive session.
+    Do not call run_skill merely because the skill was named.
+
+    Returns JSON with success, result, session_id, subtype, is_error, exit_code, needs_retry,
+    and retry_reason. When needs_retry is true, follow the recipe's declared retry route.
 
     Args:
-        skill_command: The full prompt including skill invocation (e.g. "/investigate ...").
-        cwd: Working directory for the claude session.
-        model: Model to use (e.g. "sonnet", "opus"). Empty string = use config default.
+        skill_command: Full recipe-declared skill invocation or resume continuation.
+        cwd: Absolute working directory for the separate coding-agent worker.
+        model: Optional model identifier. Empty string uses the configured default.
         step_name: Optional YAML step key (e.g. "implement"). When set, token usage is
             accumulated in the server-side token log, grouped by this name.
         order_id: Optional per-issue/order identifier for token telemetry scoping. When set,
@@ -659,9 +631,8 @@ async def run_skill(
             0 = disabled for this step. None = use global config
             (RunSkillConfig.idle_output_timeout, default 1000s).
         resume_session_id: Session ID from a previous run_skill call that was interrupted.
-            When set, the session is resumed via --resume instead of starting fresh.
-            The skill_command becomes a continuation instruction (non-slash text is allowed).
-            Pass the session_id from the previous run_skill result JSON.
+            When set, resume that coding-agent session instead of starting fresh. The
+            skill_command becomes a continuation instruction; pass the prior result's session_id.
 
     Never raises.
     """

tests/cli/AGENTS.md

@@ -78,7 +78,7 @@ CLI command, subcommand, and interactive workflow tests.
 | `test_reap_sidecar_check.py` | Tests for _reap_stale_dispatches sidecar-aware status transition (T-RESUMABLE-10) |
 | `test_reload_loop.py` | Tests for the session reload sentinel and loop mechanics |
 | `test_restart.py` | Tests for cli/_restart.py — NoReturn process restart contract |
-| `test_routing_completeness.py` | Enum routing completeness: every orchestrator-visible RetryReason must have a routing rule in the orchestrator prompt; pins run_skill's docstring to the same contract_recovery routing vocabulary as the prompt |
+| `test_routing_completeness.py` | Prompt-owned retry taxonomy: every orchestrator-visible RetryReason must have its declared routing rule in the orchestrator prompt |
 | `test_serve_guard_deferral.py` | Tests for serve_with_signal_guard dispatch-aware deferral — active dispatch deferral, immediate cancel, timeout |
 | `test_serve_sigterm.py` | Regression guard: serve() uses event-loop-routed signal handling (issue #745) |
 | `test_session_launch.py` | Tests for cli/_session_launch.py — _run_interactive_session contract |

tests/cli/test_routing_completeness.py

@@ -93,32 +93,3 @@ def test_expected_routes_covers_all_orchestrator_visible_reasons() -> None:
         r.name for r in RetryReason if r not in _ROUTING_EXCLUDED and r not in _EXPECTED_ROUTES
     ]
     assert not missing, f"Add routing expectation for: {missing}"
-
-
-def test_run_skill_docstring_matches_contract_recovery_routing_vocabulary() -> None:
-    """run_skill's docstring routing table must use the same contract_recovery vocabulary
-    as the orchestrator prompt (issue #4305).
-
-    Pins the two prose copies (prompt, docstring — the third copy, SKILL.md, is covered
-    by tests/contracts/test_sous_chef_routing.py) to the same routing terms so silent
-    drift in either copy fails a test instead of surfacing as a live routing mismatch.
-    """
-    from autoskillit.server.tools.tools_execution import run_skill
-
-    doc = run_skill.__doc__
-    assert doc is not None, "run_skill must have a docstring"
-
-    idx = doc.find(RetryReason.CONTRACT_RECOVERY.value)
-    assert idx != -1, f"{RetryReason.CONTRACT_RECOVERY.value} not found in run_skill docstring"
-
-    route_keyword, evidence_keyword = _EXPECTED_ROUTES[RetryReason.CONTRACT_RECOVERY]
-    window = doc[idx : idx + 600]
-    assert route_keyword in window, (
-        f"run_skill docstring: {RetryReason.CONTRACT_RECOVERY.value} must reference "
-        f"'{route_keyword}' within 600 chars"
-    )
-    assert evidence_keyword, "CONTRACT_RECOVERY must have a non-None evidence keyword"
-    assert evidence_keyword in window, (
-        f"run_skill docstring: {RetryReason.CONTRACT_RECOVERY.value} routing must "
-        f"reference evidence signal '{evidence_keyword}'"
-    )

tests/contracts/AGENTS.md

@@ -53,7 +53,7 @@ Protocol satisfaction, package gateway, and skill contract compliance tests.
 | `test_implement_experiment_contracts.py` | Contract tests for implement-experiment SKILL.md — test infrastructure requirements |
 | `test_input_type_semantic_correctness.py` | Cross-validate skill_contracts.yaml path input types against SKILL.md content |
 | `test_install_state_consistency.py` | `verify_install_state()` invariants, the doctor checks that consume it, and the retired-artifact-shape registry double bind |
-| `test_instruction_surface.py` | Contract tests: every instruction surface must carry the pipeline tool restriction |
+| `test_instruction_surface.py` | Contracts for the authoritative instruction surfaces that own pipeline native-tool restrictions and prohibition framing |
 | `test_issue_body_discipline.py` | Cross-skill contract: no SKILL.md may append validation summaries to issue bodies |
 | `test_issue_content_fidelity.py` | Cross-skill contract: content fidelity for issue body assembly |
 | `test_issue_splitter_contracts.py` | Contract tests: issue-splitter skill correctness and triage-issues integration |

tests/contracts/test_instruction_surface.py

@@ -1,8 +1,6 @@
-"""Contract tests: every instruction surface must carry the pipeline tool restriction.
+"""Contracts for authoritative pipeline instruction surfaces.
 
-These tests verify that all surfaces where the orchestrator receives instructions
-about native tool usage contain the full forbidden tool list with prohibition framing.
-If any test fails, a drift has occurred and the corresponding surface needs updating.
+Guards the surfaces that own native-tool restrictions and prohibition framing.
 """
 
 from __future__ import annotations
@@ -64,7 +62,7 @@ def test_skill_md_names_forbidden_tools(self):
 
 
 class TestServerToolSurfaceContract:
-    """Server tool docstrings and prompts must name all forbidden tools."""
+    """Authoritative server prompts and recipe-loading docs name forbidden tools."""
 
     @pytest.fixture(autouse=True)
     def _close_kitchen(self, minimal_ctx, monkeypatch):
@@ -93,15 +91,6 @@ async def test_open_kitchen_prompt_names_all_forbidden_tools(self):
         has_framing = any(term in text for term in ("NEVER", "Do NOT", "MUST NOT"))
         assert has_framing, "open_kitchen prompt lacks prohibition framing (NEVER/Do NOT/MUST NOT)"
 
-    def test_run_skill_docstring_names_all_forbidden_tools(self):
-        """run_skill docstring must name every forbidden tool."""
-        from autoskillit.server.tools.tools_execution import run_skill
-
-        doc = run_skill.__doc__
-        assert doc, "run_skill has no docstring"
-        missing = [t for t in PIPELINE_FORBIDDEN_TOOLS if t not in doc]
-        assert not missing, f"run_skill docstring missing tools: {missing}"
-
     def test_load_recipe_docstring_names_all_forbidden_tools(self):
         """load_recipe docstring must name every forbidden tool."""
         from autoskillit.server.tools.tools_recipe import load_recipe

tests/execution/AGENTS.md

@@ -168,5 +168,5 @@ Subprocess integration, headless session, process lifecycle, and session result
 | `test_codex_deterministic_conformance.py` | Sealed-enum vocabulary, hook event format, and config.toml schema template conformance tests with --update-fixtures review gate |
 | `test_cmd_builder.py` | CmdBuilder ordering invariant and CmdSpec origin tests |
 | `test_coding_agent_backend_conformance.py` | Parametrized conformance tests for all CodingAgentBackend implementations via BackendContractBase |
-| `test_cli_conformance_probes.py` | Live backend CLI conformance probes: schema checks, isolated output-budget deny round trips, and generated Codex child delivery/linkage with ProbeCache and shared error discrimination |
+| `test_cli_conformance_probes.py` | Live backend CLI conformance probes: schema checks, isolated output-budget denial, generated Codex child delivery/linkage, and isolated Codex local-skill versus recipe-delegation selection |
 | `test_probe_cache.py` | Versioned probe-cache tests: CLI/policy identity invalidation, TTL, schema, and write preservation |

tests/execution/backends/test_cli_conformance_probes.py

@@ -40,6 +40,7 @@
     ensure_codex_mcp_registered,
 )
 from autoskillit.execution.backends._codex_hooks import sync_hooks_to_codex_config
+from autoskillit.execution.backends._codex_parse import CodexResultParser, CodexStreamParser
 from autoskillit.execution.backends._probe_cache import (
     PROBE_POLICY_IDENTITY,
     ProbeResult,
@@ -365,6 +366,20 @@ def _assert(output: _CodexProbeOutput) -> None:
     ),
 )
 
+_skip_unless_codex_selection_smoke = pytest.mark.skipif(
+    not os.environ.get("CODEX_SMOKE_TEST")
+    or not shutil.which("codex")
+    or (
+        not os.environ.get("CODEX_API_KEY")
+        and not os.environ.get("OPENAI_API_KEY")
+        and not _CODEX_AUTH_PATH.is_file()
+    ),
+    reason=(
+        "Set CODEX_SMOKE_TEST=1 and provide an environment API key or authenticated "
+        "~/.codex/auth.json to run the Codex skill-selection probe"
+    ),
+)
+
 _skip_unless_claude_output_budget_smoke = pytest.mark.skipif(
     not os.environ.get("CLAUDE_CODE_SMOKE_TEST")
     or not shutil.which("claude")
@@ -390,6 +405,11 @@ class _GeneratedChildProbeOutput(NamedTuple):
     cli_version: str
 
 
+class _CodexSelectionProbeOutput(NamedTuple):
+    final_text: str
+    completed_mcp_items: tuple[dict, ...]
+
+
 def _isolated_cli_env(tmp_path: Path, workspace: Path) -> tuple[dict[str, str], Path, Path]:
     home = tmp_path / "home"
     codex_home = tmp_path / "codex-home"
@@ -412,6 +432,115 @@ def _isolated_cli_env(tmp_path: Path, workspace: Path) -> tuple[dict[str, str],
     return env, codex_home, claude_config
 
 
+def _prepare_codex_selection_profile(tmp_path: Path, workspace: Path) -> Path:
+    _, profile_codex_home, _ = _isolated_cli_env(tmp_path / "source-profile", workspace)
+    if _CODEX_AUTH_PATH.is_file():
+        (profile_codex_home / "auth.json").symlink_to(_CODEX_AUTH_PATH.resolve())
+
+    profile_config = profile_codex_home / "config.toml"
+    ensure_codex_mcp_registered(config_path=profile_config, headless_auto_gate=False)
+    sync_hooks_to_codex_config(config_path=profile_config)
+
+    skill_dir = profile_codex_home / "skills" / "investigate"
+    skill_dir.mkdir(parents=True)
+    (skill_dir / "SKILL.md").write_text(
+        "---\n"
+        "name: investigate\n"
+        "description: Follow this skill when the user asks to investigate something.\n"
+        "---\n"
+        "Respond with exactly LOCAL_INVESTIGATE_SKILL_FOLLOWED and do not call tools.\n",
+        encoding="utf-8",
+    )
+    return profile_codex_home
+
+
+def _run_codex_selection_case(
+    *,
+    case_root: Path,
+    source_codex_home: Path,
+    workspace: Path,
+    prompt: str,
+    model: str,
+) -> _CodexSelectionProbeOutput:
+    env, case_codex_home, _ = _isolated_cli_env(case_root, workspace)
+    case_sqlite_home = case_root / "codex-sqlite-home"
+    case_sqlite_home.mkdir()
+
+    env.update(
+        {
+            "CODEX_SQLITE_HOME": str(case_sqlite_home),
+            "AUTOSKILLIT_AGENT_BACKEND": "codex",
+            "AUTOSKILLIT_AGENT_BACKEND__BACKEND": "codex",
+            "AUTOSKILLIT_MCP_CLIENT_BACKEND": "codex",
+        }
+    )
+    for inherited_headless_flag in (
+        "AUTOSKILLIT_HEADLESS",
+        "AUTOSKILLIT_HEADLESS_AUTO_GATE",
+        "AUTOSKILLIT_SESSION_TYPE",
+        "AUTOSKILLIT_SKILL_NAME",
+    ):
+        env.pop(inherited_headless_flag, None)
+
+    backend = CodexBackend(source_codex_home=source_codex_home)
+    pre_launch_errors = backend.ensure_pre_launch(session_dir=case_codex_home)
+    assert not pre_launch_errors, f"Codex pre-launch failed: {pre_launch_errors}"
+    backend.setup_session_dir(case_codex_home)
+
+    venv_bin = Path(__file__).resolve().parents[3] / ".venv" / "bin"
+    env["PATH"] = f"{venv_bin}{os.pathsep}{env.get('PATH', '')}"
+    timeout = int(os.environ.get("CODEX_SELECTION_SMOKE_TIMEOUT", "900"))
+    result = subprocess.run(  # noqa: S603
+        [
+            "codex",
+            "exec",
+            "--json",
+            "--sandbox",
+            "workspace-write",
+            "--model",
+            model,
+            prompt,
+        ],
+        cwd=workspace,
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
+    if result.returncode != 0:
+        raise OSError(
+            f"Codex selection probe failed with rc={result.returncode}: "
+            f"{result.stdout}\n{result.stderr}"
+        )
+
+    stream_parser = CodexStreamParser()
+    completed_mcp_items: list[dict] = []
+    for line in result.stdout.splitlines():
+        event = stream_parser.parse_line(line)
+        if event is None or event.backend_data is None:
+            continue
+        raw = event.backend_data.raw
+        item = raw.get("item", {})
+        if (
+            raw.get("type") == "item.completed"
+            and isinstance(item, dict)
+            and item.get("type") == "mcp_tool_call"
+        ):
+            completed_mcp_items.append(item)
+
+    parsed = CodexResultParser().parse_stdout(result.stdout, exit_code=result.returncode)
+    return _CodexSelectionProbeOutput(
+        final_text=parsed.output,
+        completed_mcp_items=tuple(completed_mcp_items),
+    )
+
+
+def _completed_mcp_tool_names(output: _CodexSelectionProbeOutput) -> list[str]:
+    return [
+        str(item.get("tool_name") or item.get("tool") or "") for item in output.completed_mcp_items
+    ]
+
+
 def _cli_version(binary: str, env: dict[str, str]) -> str:
     try:
         result = subprocess.run(  # noqa: S603
@@ -569,6 +698,69 @@ def _assert_generated_child_probe(output: _GeneratedChildProbeOutput) -> None:
     )
 
 
+@_skip_unless_codex_selection_smoke
+def test_codex_selects_local_skill_and_explicit_recipe_delegation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    subprocess.run(
+        ["git", "init", "-q"],
+        cwd=workspace,
+        check=True,
+        capture_output=True,
+        text=True,
+        timeout=10,
+    )
+
+    for selector in (
+        "AUTOSKILLIT_AGENT_BACKEND",
+        "AUTOSKILLIT_AGENT_BACKEND__BACKEND",
+        "AUTOSKILLIT_MCP_CLIENT_BACKEND",
+    ):
+        monkeypatch.setenv(selector, "codex")
+    for headless_flag in (
+        "AUTOSKILLIT_HEADLESS",
+        "AUTOSKILLIT_HEADLESS_AUTO_GATE",
+        "AUTOSKILLIT_SESSION_TYPE",
+        "AUTOSKILLIT_SKILL_NAME",
+    ):
+        monkeypatch.delenv(headless_flag, raising=False)
+
+    source_codex_home = _prepare_codex_selection_profile(tmp_path, workspace)
+    model = os.environ.get("GENERATED_CHILD_SMOKE_MODEL", "gpt-5.4")
+
+    local_skill = _run_codex_selection_case(
+        case_root=tmp_path / "local-skill-case",
+        source_codex_home=source_codex_home,
+        workspace=workspace,
+        prompt="Use the investigate skill.",
+        model=model,
+    )
+    local_tool_names = _completed_mcp_tool_names(local_skill)
+    assert "LOCAL_INVESTIGATE_SKILL_FOLLOWED" in local_skill.final_text
+    assert local_tool_names == []
+
+    nonexistent_cwd = tmp_path / "missing-run-skill-cwd"
+    delegated = _run_codex_selection_case(
+        case_root=tmp_path / "delegation-case",
+        source_codex_home=source_codex_home,
+        workspace=workspace,
+        prompt=(
+            "This is an explicit recipe-step delegation request. Call run_skill exactly once "
+            "to delegate /investigate positive-probe to a separate L1 worker, passing "
+            f"cwd={str(nonexistent_cwd)!r}. Do not call open_kitchen. After run_skill "
+            "returns its preflight result, stop without calling any other tool."
+        ),
+        model=model,
+    )
+    delegated_tool_names = _completed_mcp_tool_names(delegated)
+    assert delegated_tool_names == ["run_skill"]
+    run_skill_item = delegated.completed_mcp_items[delegated_tool_names.index("run_skill")]
+    assert "preflight:cwd" in json.dumps(run_skill_item.get("result"), sort_keys=True)
+
+
 @_skip_unless_codex_generated_child_smoke
 def test_generated_codex_child_receives_output_discipline(
     tmp_path: Path,