Load selected executor skills through extensions (#27184)

## Why

CCA is moving toward a split runtime where the orchestrator may not have
a filesystem, while executors can expose preinstalled plugins and
skills. A thread therefore needs to select capabilities without asking
app-server or core to interpret executor-owned paths through the
orchestrator's filesystem.

The longer-term model is broader than executor skills:

- A plugin is a bundle of skills, MCP servers, connectors/apps, and
hooks.
- A plugin root can be local, executor-owned, or hosted by a backend.
- Components inside one plugin can use different access and execution
mechanisms. A skill may be read from a filesystem or through backend
tools; an HTTP MCP server can run without an executor; a stdio MCP
server or hook needs an execution environment.
- Core should carry generic extension initialization data. The extension
that owns a component should discover it, expose it to the model, and
invoke it through the appropriate runtime.

This PR establishes that architecture through one complete vertical:
selecting a root on an executor, discovering the skills beneath it,
exposing those skills to the model, and reading an explicitly invoked
`SKILL.md` through the same executor.

## Contract

`thread/start` gains an experimental `selectedCapabilityRoots` field:

```json
{
  "selectedCapabilityRoots": [
    {
      "id": "deploy-plugin@1",
      "location": {
        "type": "environment",
        "environmentId": "workspace",
        "path": "/opt/codex/plugins/deploy"
      }
    }
  ]
}
```

The root is intentionally not classified as a "plugin" or "skill" in the
API. It can point at a standalone skill, a directory containing several
skills, or a plugin containing skills and other components. This PR only
teaches the skills extension how to consume it; later extensions can
resolve MCP, connector, and hook components from the same selection.

The platform-supplied `id` is stable selection identity. The location
says which runtime owns the root and gives that runtime an opaque path.
App-server does not inspect or canonicalize the path.

## What changed

### Generic thread extension initialization

App-server converts selected roots into `ExtensionDataInit`. Core
carries that generic initialization value until the final thread ID is
known, then creates thread-scoped `ExtensionData` before lifecycle
contributors run.

This keeps `Session` and core independent of the capability-selection
contract. The initialization value is consumed during construction; it
is not retained as another long-lived `Session` field.

### Executor-backed skills

The skills extension now owns an `ExecutorSkillProvider` that:

- resolves the selected environment through `EnvironmentManager`
- discovers, canonicalizes, and reads skills through that environment's
`ExecutorFileSystem`
- contributes the bounded selected-skill catalog as stable developer
context
- reads an explicitly invoked skill body through the authority that
listed it
- warns when an environment or root is unavailable
- never falls back to the orchestrator filesystem for an executor-owned
root

Skill catalog and instruction fragments have hard byte bounds, which
also bound them below the 10K-token per-item context limit. If a
selected executor skill has the same name as a legacy local skill, the
executor selection owns that invocation and the local body is not
injected a second time.

Existing local and bundled skill loading remains in place. Omitting
`selectedCapabilityRoots` therefore preserves current local-only
behavior.

## Current semantics

- Only environment-owned locations are represented in this first
contract.
- Roots are resolved by the destination extension, not by app-server or
core.
- An unavailable executor or invalid root produces a warning and no
capabilities from that root; it does not trigger a local-filesystem
fallback.
- Selection applies to a newly started active thread.
- MCP servers, connectors, and hooks beneath a selected plugin root are
not activated yet.
- Selection is not yet persisted or inherited across resume, fork, or
subagent creation. Existing local capabilities continue to behave as
they do today in those flows.

## Planned vertical follow-ups

1. **Hosted HTTP MCP:** add an extension-backed HTTP MCP source that
works without an executor, then replace the special-purpose MCP plugins
loader with that implementation.
2. **Executor MCP:** register and execute stdio MCP servers through the
environment that owns the selected plugin root.
3. **Backend skills:** add a hosted skill source whose catalog and
bodies are accessed through extension tools rather than a filesystem.
4. **Connectors and hooks:** activate those components through their
owning extensions, using the same selected-root boundary and
component-specific runtime.
5. **Durable selection:** define the desired-selection lifecycle,
persist it, and make resume, fork, and subagent inheritance explicit
rather than accidental.
6. **Local convergence:** incrementally route existing local plugin,
skill, and MCP loading through the same extension model while preserving
current local behavior.

Each follow-up remains reviewable as an end-to-end capability. The
platform selects roots, generic thread extension data carries the
selection, and the owning extension resolves and operates its component.

## Verification

Coverage added for:

- app-server end-to-end discovery and explicit invocation of a skill
inside an executor-selected plugin root
- exclusive invocation when a selected executor skill collides with a
local skill name
- executor filesystem authority for discovery, canonicalization, and
reads
- thread extension initialization before lifecycle contributors run
- stable executor catalog context, explicit invocation, context
rebuilding, hidden skills, and preserved host/remote catalog behavior

Targeted protocol, core-skills, skills-extension, core lifecycle, and
app-server executor-skill tests were run during development.

codex-rs/core/src/thread_manager_tests.rs

@@ -331,6 +331,7 @@ async fn start_thread_rejects_explicit_local_environment_when_default_provider_i
                 environment_id: "local".to_string(),
                 cwd: config.cwd.clone(),
             }],
+            thread_extension_init: Default::default(),
         })
         .await;
     let err = match result {
@@ -368,6 +369,7 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() {
             metrics_service_name: None,
             parent_trace: None,
             environments: Vec::new(),
+            thread_extension_init: Default::default(),
         })
         .await
         .expect("internal thread should start");
@@ -384,6 +386,81 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() {
     assert!(manager.list_thread_ids().await.is_empty());
 }
 
+#[tokio::test]
+async fn start_thread_seeds_extension_data_before_lifecycle_contributors_run() {
+    struct InitialMarker(&'static str);
+
+    struct InitialDataRecorder {
+        observed: Arc<std::sync::Mutex<Option<(String, String)>>>,
+    }
+
+    #[async_trait::async_trait]
+    impl codex_extension_api::ThreadLifecycleContributor<Config> for InitialDataRecorder {
+        async fn on_thread_start(&self, input: codex_extension_api::ThreadStartInput<'_, Config>) {
+            let marker = input
+                .thread_store
+                .get::<InitialMarker>()
+                .expect("initial extension data should be available");
+            *self
+                .observed
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner) = Some((
+                input.thread_store.level_id().to_string(),
+                marker.0.to_string(),
+            ));
+        }
+    }
+
+    let temp_dir = tempdir().expect("tempdir");
+    let mut config = test_config().await;
+    config.codex_home = temp_dir.path().join("codex-home").abs();
+    config.cwd = config.codex_home.abs();
+    std::fs::create_dir_all(&config.codex_home).expect("create codex home");
+
+    let observed = Arc::new(std::sync::Mutex::new(None));
+    let mut extensions = codex_extension_api::ExtensionRegistryBuilder::new();
+    extensions.thread_lifecycle_contributor(Arc::new(InitialDataRecorder {
+        observed: Arc::clone(&observed),
+    }));
+    let manager = ThreadManager::new(
+        &config,
+        AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()),
+        SessionSource::Exec,
+        Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
+        Arc::new(extensions.build()),
+        /*analytics_events_client*/ None,
+        thread_store_from_config(&config, /*state_db*/ None),
+        /*state_db*/ None,
+        TEST_INSTALLATION_ID.to_string(),
+        /*attestation_provider*/ None,
+    );
+    let mut thread_extension_init = codex_extension_api::ExtensionDataInit::new();
+    thread_extension_init.insert(InitialMarker("seeded"));
+
+    let thread = manager
+        .start_thread_with_options(StartThreadOptions {
+            config,
+            initial_history: InitialHistory::New,
+            session_source: None,
+            thread_source: None,
+            dynamic_tools: Vec::new(),
+            metrics_service_name: None,
+            parent_trace: None,
+            environments: Vec::new(),
+            thread_extension_init,
+        })
+        .await
+        .expect("start thread");
+
+    assert_eq!(
+        observed
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone(),
+        Some((thread.thread_id.to_string(), "seeded".to_string()))
+    );
+}
+
 #[tokio::test]
 async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
     let temp_dir = tempdir().expect("tempdir");
@@ -423,6 +500,7 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
             metrics_service_name: None,
             parent_trace: None,
             environments: environments.clone(),
+            thread_extension_init: Default::default(),
         })
         .await
         .expect("start source thread");
@@ -693,6 +771,7 @@ async fn resume_stopped_thread_from_rollout_preserves_thread_source() {
             metrics_service_name: None,
             parent_trace: None,
             environments: Vec::new(),
+            thread_extension_init: Default::default(),
         })
         .await
         .expect("start source thread");