research: topology_worker points executor at per-topic gateway (commit 2/3)

Commit 2 of the path-B plan. The container that commit 1 spawns
now actually receives the run's turns — up until now it was
started but unused. This is the payoff commit: research runs are
truly isolated per topic.

Backend

- topology_exec.rs: from_env() refactored to a thin wrapper over a
  new from_env_for_gateway(url) helper. Same shape (env-derived
  aliases + default + token) but the caller supplies the URL. The
  auth token/pairing code still comes from ZEROCLAW_TOKEN /
  ZEROCLAW_PAIRING_CODE on the server; research_container's
  inherited_env propagates those into the team container so the
  same credentials work at both endpoints.

- research_container.rs: new wait_ready(url, deadline) that polls
  <url>/health with a 1.5s per-request timeout every 500ms until
  it 200s or the deadline passes. reqwest-based so it doesn't need
  bollard. Called by the worker after claim, before pair, to bridge
  the "container is starting, gateway not yet listening" gap.

- topology_worker.rs run_job:
    1. Look up research_topic_id for the claimed run.
    2. If Some, load the topic and read zeroclaw_gateway_url.
    3. If a URL is present:
         - best-effort wait_ready(url, 30s); a timeout logs but
           doesn't abort — the pair call below will just fail
           faster than pinging forever
         - build the leaf via from_env_for_gateway(url)
       Else fall back to from_env() (workspace-wide gateway).
    4. The rest of run_job is unchanged — the leaf drops into
       either SubTopologyExecutor (org/company) or direct drive
       (team tier) as before.

What now works end-to-end

Starting a research topic with a bound repo:
  1. clone-shallow into per-topic workspace
  2. docker create + start the clawmates-runtime container, name
     = research-<topic>-team, joined to clawmates_core so the
     server reaches it by name
  3. persist container name + gateway URL on the topic row
  4. enqueue the topology run tagged with research_topic_id
  5. worker claims → looks up the topic → waits for the team
     gateway's /health → constructs a from_env_for_gateway
     executor pointed at http://research-<topic>-team:42617
  6. every turn's `/ws/chat?agent=…` hits the isolated container;
     agents inside see the repo at /workspace/repo (rw); each
     topic's memory/state lives under its own /zeroclaw-data mount

Deploy prereqs (unchanged from commit 1)

- clawmates_server compose service needs a bind-mount of
  CLAWMATES_RESEARCH_WORKSPACE_ROOT so the paths spawn() writes to
  are visible on the host and the spawned team container mounts
  the same underlying data.
- socket-proxy ACL needs POST + DELETE on /containers (prod ✓).

crates/cm-api/src/research_container.rs

@@ -181,6 +181,32 @@ pub async fn spawn(
     Ok(SpawnedContainer { name, gateway_url })
 }
 
+/// Poll the team gateway's `/health` endpoint until it 200s or the
+/// deadline passes. Called before firing turns against a freshly-spawned
+/// container so the executor doesn't try to pair against a not-yet-
+/// listening daemon. Uses reqwest directly — the deadline caps total
+/// wait so a broken image doesn't hang the worker forever.
+pub async fn wait_ready(gateway_url: &str, deadline: std::time::Duration) -> Result<(), String> {
+    let start = std::time::Instant::now();
+    let client = reqwest::Client::builder()
+        .timeout(std::time::Duration::from_millis(1500))
+        .build()
+        .map_err(|e| format!("client build: {e}"))?;
+    let url = format!("{}/health", gateway_url.trim_end_matches('/'));
+    let mut last_err = String::from("no attempt");
+    while start.elapsed() < deadline {
+        match client.get(&url).send().await {
+            Ok(res) if res.status().is_success() => return Ok(()),
+            Ok(res) => last_err = format!("HTTP {}", res.status()),
+            Err(e) => last_err = e.to_string(),
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    }
+    Err(format!(
+        "team gateway never became ready ({url}): {last_err}"
+    ))
+}
+
 /// Stop and remove the per-topic container. Called on topic delete and on
 /// terminal-state cleanup. Idempotent: a missing container is a no-op.
 #[allow(dead_code)]

crates/cm-api/src/topology_exec.rs

@@ -71,6 +71,18 @@ impl ZeroClawDriveExecutor {
     pub fn from_env() -> Result<Self, String> {
         let gateway_url =
             std::env::var("ZEROCLAW_GATEWAY_URL").map_err(|_| "ZEROCLAW_GATEWAY_URL not set")?;
+        Self::from_env_for_gateway(gateway_url)
+    }
+
+    /// Same as [`from_env`] but with a caller-supplied gateway URL. Used by
+    /// the research pipeline to point the executor at the per-topic team
+    /// container spawned in `research_container::spawn` instead of the
+    /// workspace-wide gateway from `ZEROCLAW_GATEWAY_URL`. The auth token,
+    /// role map, and default alias still come from the parent server's
+    /// env — they're propagated into the team container by
+    /// `research_container::inherited_env` so both endpoints use the same
+    /// credentials.
+    pub fn from_env_for_gateway(gateway_url: String) -> Result<Self, String> {
         let token = std::env::var("ZEROCLAW_TOKEN")
             .ok()
             .filter(|t| !t.is_empty());

crates/cm-api/src/topology_worker.rs

@@ -91,7 +91,36 @@ async fn run_job(
         .and_then(|c| serde_json::from_value(c).ok())
         .unwrap_or_default();
 
-    let leaf = match ZeroClawDriveExecutor::from_env() {
+    // If this run belongs to a research topic and that topic has a per-team
+    // ZeroClaw container spawned (see research_container::spawn), point the
+    // executor at THAT container's gateway URL so the run's turns hit its
+    // isolated daemon instead of the workspace-wide one. Falls back to the
+    // env-derived executor when there's no per-topic container (non-research
+    // runs, or research runs where spawn failed and we recorded no URL).
+    let per_topic_url = match cm_db::repo::topology_runs::research_topic_id(pool, id).await {
+        Ok(Some(topic_id)) => {
+            match cm_db::repo::research_topics::get(pool, topic_id, job.workspace_id).await {
+                Ok(Some(t)) => t.zeroclaw_gateway_url,
+                _ => None,
+            }
+        }
+        _ => None,
+    };
+    if let Some(url) = &per_topic_url {
+        // Best-effort readiness gate — a freshly-spawned team container may
+        // still be starting when the worker claims the run. Cap the wait so
+        // a broken image can't hang the worker.
+        if let Err(e) =
+            crate::research_container::wait_ready(url, std::time::Duration::from_secs(30)).await
+        {
+            eprintln!("topology_worker: research team {url} readiness: {e} — proceeding anyway");
+        }
+    }
+    let leaf_result = match &per_topic_url {
+        Some(url) => ZeroClawDriveExecutor::from_env_for_gateway(url.clone()),
+        None => ZeroClawDriveExecutor::from_env(),
+    };
+    let leaf = match leaf_result {
         Ok(e) => e,
         Err(e) => {
             let _ = cm_db::repo::topology_runs::fail(pool, id, &e).await;