loops: Path B container isolation (P2)

Symmetric with the research pipeline: every enabled loop can now have
its own per-loop team container so scheduled runs don't share state
with other loops or with research. Same daemon image, same clawmates
network, deterministic name loop-<id>-team.

Backend surface:
- Migration 0040 adds nullable `zeroclaw_container` +
  `zeroclaw_gateway_url` columns to loops (parallel to
  research_topics).
- research_container.rs grows loop_container_name_for(), spawn_loop()
  (state-only mount, no repo), and teardown_loop(). Kept in the same
  module to share the docker connect() + inherited_env() plumbing;
  each pattern gets its own labels (clawmates.role=loop-team) so ps
  filters can tell them apart.
- cm_db::repo::loops gains set_zeroclaw_container() +
  zeroclaw_gateway_url() (dynamic sqlx queries — no offline cache
  regen needed).
- cm_db::repo::topology_runs gets loop_id_for_run(): mirror of
  research_topic_id, used by the worker.

Wiring:
- routes/loops::run_now + webhook_receive call ensure_loop_container()
  before enqueuing an iteration. Idempotent: an already-running
  container is just reattached. Failures are logged and do NOT block
  the enqueue — topology_worker falls back to the workspace gateway
  when the URL isn't set on the loop.
- routes/loops::disable_loop + delete_loop both fire teardown_loop()
  so paused / deleted loops don't hold a docker slot.
- topology_worker's per-run URL resolution: existing research fast
  path unchanged; when it doesn't hit, the worker now looks up
  loop_id and reads the loop's gateway URL.

Deploy step (required on gw-04 for state to persist across container
restarts): add a `/var/lib/clawmates-loops:/var/lib/clawmates-loops`
bind mount + `CLAWMATES_LOOPS_STATE_ROOT=/var/lib/clawmates-loops`
env var to clawmates_server_1 in the compose. Without it, loops still
run — the state dir lives inside the API container's filesystem so
persistence is limited to that container's lifetime.

Follow-up:
- Scheduler-tick fires (cron-driven, not run_now) — they call
  enqueue_iteration in cm-scheduler and don't yet go through
  ensure_loop_container. Add a symmetric spawn there so cron fires
  also land on the isolated daemon.
- Compose file reconciliation — deploy/compose/docker-compose.yml in
  the repo has drifted from prod; when we sync it, add the loops mount
  at the same time.

crates/cm-api/src/routes/loops.rs

@@ -21,9 +21,55 @@ use cm_runtime::scheduling::next_occurrence;
 use hmac::{Hmac, Mac};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
+use sqlx::PgPool;
 use time::OffsetDateTime;
 use uuid::Uuid;
 
+/// Root of per-loop state dirs on the host. Same overridable env pattern
+/// as research_workspace_root — prod points at the bind-mounted volume
+/// `/var/lib/clawmates-loops` on gw-04.
+fn loop_state_root() -> std::path::PathBuf {
+    std::env::var("CLAWMATES_LOOPS_STATE_ROOT")
+        .map(std::path::PathBuf::from)
+        .unwrap_or_else(|_| std::path::PathBuf::from("/var/lib/clawmates-loops"))
+}
+
+/// Best-effort spawn of the per-loop team container before an iteration
+/// is enqueued. Idempotent — an already-running container is just
+/// reattached. Failures (docker unreachable, image missing) log and
+/// return without blocking the run; the topology_worker will fall back
+/// to the workspace-wide gateway. Records the container name + URL on
+/// the loop row on first success so subsequent fires skip re-writing.
+async fn ensure_loop_container(pool: &PgPool, workspace_id: Uuid, loop_id: Uuid) {
+    let docker = match crate::research_container::connect() {
+        Ok(d) => d,
+        Err(e) => {
+            eprintln!("loops::ensure_loop_container({loop_id}): docker connect failed: {e}");
+            return;
+        }
+    };
+    let state_root = loop_state_root().join(loop_id.to_string()).join("state");
+    let spawned =
+        match crate::research_container::spawn_loop(&docker, loop_id, &state_root).await {
+            Ok(s) => s,
+            Err(e) => {
+                eprintln!("loops::ensure_loop_container({loop_id}): spawn failed: {e}");
+                return;
+            }
+        };
+    if let Err(e) = cm_db::repo::loops::set_zeroclaw_container(
+        pool,
+        loop_id,
+        workspace_id,
+        &spawned.name,
+        &spawned.gateway_url,
+    )
+    .await
+    {
+        eprintln!("loops::ensure_loop_container({loop_id}): persist failed: {e:?}");
+    }
+}
+
 use crate::{ApiError, AppState, Authed};
 
 #[derive(Deserialize)]
@@ -285,6 +331,9 @@ pub async fn delete_loop(
     Path(id): Path<Uuid>,
 ) -> Result<StatusCode, ApiError> {
     cm_db::repo::loops::delete(&state.pool, id, user.workspace_id.as_uuid()).await?;
+    // Tear down the per-loop container (P2). Fire-and-forget: the row
+    // is gone, so any Docker failure is a log-line, not an API failure.
+    crate::research_container::teardown_loop(id).await;
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -303,6 +352,10 @@ pub async fn disable_loop(
     Path(id): Path<Uuid>,
 ) -> Result<StatusCode, ApiError> {
     cm_db::repo::loops::set_enabled(&state.pool, id, user.workspace_id.as_uuid(), false).await?;
+    // Stop the per-loop container while disabled — re-enabling later will
+    // spawn a fresh one on the next `run_now` / webhook fire. Keeps
+    // paused loops from holding a docker slot.
+    crate::research_container::teardown_loop(id).await;
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -323,6 +376,10 @@ pub async fn run_now(
     let l = cm_db::repo::loops::get(&state.pool, id, user.workspace_id.as_uuid())
         .await?
         .ok_or(ApiError::NotFound)?;
+    // P2: spawn the per-loop container before enqueue so topology_worker
+    // resolves its gateway URL when it picks up the run. Best-effort;
+    // never blocks the enqueue on Docker being unreachable.
+    ensure_loop_container(&state.pool, l.workspace_id, l.id).await;
     let iter = cm_db::repo::loops::next_iteration(&state.pool, l.id).await?;
     let run_id = cm_db::repo::loops::enqueue_iteration(
         &state.pool,
@@ -369,6 +426,7 @@ pub async fn webhook_receive(
         return (StatusCode::UNAUTHORIZED, Json(Value::Null));
     }
 
+    ensure_loop_container(&state.pool, l.workspace_id, l.id).await;
     let iter = match cm_db::repo::loops::next_iteration(&state.pool, l.id).await {
         Ok(n) => n,
         Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(Value::Null)),