feat(multinode): add container reconciliation, scheduling config UI, and integration tests

Add container labels for reconciliation, list_containers agent endpoint,
anti-affinity environment settings, fix auth and WebSocket log tests,
and add multinode integration test suite.

Author: dviejokfs

crates/temps-agent/src/handlers.rs

@@ -61,6 +61,7 @@ fn error_response(status: StatusCode, message: String) -> impl IntoResponse {
         remove_container,
         get_container_logs,
         get_container_info,
+        list_containers,
         image_exists,
         import_image,
         health_check,
@@ -312,6 +313,33 @@ pub async fn get_container_info(
     }
 }
 
+/// List all containers on this worker node
+#[utoipa::path(
+    tag = "Containers",
+    get,
+    path = "/agent/containers",
+    responses(
+        (status = 200, description = "List of containers", body = AgentResponse<Vec<temps_deployer::ContainerInfo>>),
+        (status = 401, description = "Unauthorized"),
+        (status = 500, description = "Failed to list containers")
+    ),
+    security(("bearer_auth" = []))
+)]
+pub async fn list_containers(State(state): State<Arc<AgentState>>) -> impl IntoResponse {
+    tracing::debug!("Listing containers");
+    match state.container_deployer.list_containers().await {
+        Ok(containers) => AgentResponse::ok(containers).into_response(),
+        Err(e) => {
+            tracing::error!("Failed to list containers: {}", e);
+            error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                format!("Failed to list containers: {}", e),
+            )
+            .into_response()
+        }
+    }
+}
+
 /// Check if a Docker image exists on this node
 #[utoipa::path(
     tag = "Images",

crates/temps-agent/src/server.rs

@@ -44,6 +44,7 @@ pub fn build_router(
             "/agent/containers/{id}/info",
             get(handlers::get_container_info),
         )
+        .route("/agent/containers", get(handlers::list_containers))
         .route("/agent/images/import", post(handlers::import_image))
         .route("/agent/images/{name}/exists", get(handlers::image_exists))
         .route("/agent/health", get(handlers::health_check))
@@ -72,7 +73,13 @@ const HEARTBEAT_RETRY_MAX_DELAY: Duration = Duration::from_secs(15);
 /// On transient failures, retries up to `HEARTBEAT_MAX_RETRIES` times with exponential
 /// backoff before giving up for this interval. This prevents a brief network blip from
 /// causing the control plane to mark the node as offline (90s stale threshold).
-fn spawn_heartbeat_loop(config: &AgentConfig) {
+///
+/// The first successful heartbeat includes a full container inventory so the control
+/// plane can reconcile stale DB records against actual Docker state (e.g., after a crash).
+fn spawn_heartbeat_loop(
+    config: &AgentConfig,
+    container_deployer: Arc<dyn temps_deployer::ContainerDeployer>,
+) {
     let control_plane_url = config.control_plane_url.clone();
     let node_id = config.node_id;
     let token = config.token.clone();
@@ -98,12 +105,51 @@ fn spawn_heartbeat_loop(config: &AgentConfig) {
 
         let mut interval = tokio::time::interval(Duration::from_secs(30));
         let mut consecutive_failures: u32 = 0;
+        let mut inventory_sent = false;
 
         loop {
             interval.tick().await;
 
             let capacity = collect_capacity_metrics();
-            let body = serde_json::json!({ "capacity": capacity, "labels": labels });
+            let mut body = serde_json::json!({ "capacity": capacity, "labels": labels });
+
+            // On the first heartbeat (agent startup/reconnect), include a full
+            // container inventory so the control plane can reconcile stale state.
+            if !inventory_sent {
+                match container_deployer.list_containers().await {
+                    Ok(containers) => {
+                        // Only include temps-managed containers
+                        let managed: Vec<_> = containers
+                            .into_iter()
+                            .filter(|c| {
+                                c.labels
+                                    .get("sh.temps.managed")
+                                    .map(|v| v == "true")
+                                    .unwrap_or(false)
+                            })
+                            .map(|c| {
+                                serde_json::json!({
+                                    "container_id": c.container_id,
+                                    "container_name": c.container_name,
+                                })
+                            })
+                            .collect();
+                        body["containers"] = serde_json::json!(managed);
+                        tracing::info!(
+                            node_id = node_id,
+                            count = managed.len(),
+                            "Including container inventory in heartbeat for reconciliation"
+                        );
+                    }
+                    Err(e) => {
+                        tracing::warn!(
+                            node_id = node_id,
+                            "Failed to list containers for inventory: {}",
+                            e
+                        );
+                    }
+                }
+            }
 
             let mut attempt = 0;
             let mut succeeded = false;
@@ -127,6 +173,7 @@ fn spawn_heartbeat_loop(config: &AgentConfig) {
                         }
                         consecutive_failures = 0;
                         succeeded = true;
+                        inventory_sent = true;
                         tracing::debug!(node_id = node_id, "Heartbeat sent to control plane");
                         break;
                     }
@@ -236,10 +283,10 @@ pub async fn start_agent_server(
     image_builder: Arc<dyn ImageBuilder>,
     config: AgentConfig,
 ) -> Result<(), crate::AgentError> {
-    let router = build_router(container_deployer, image_builder, &config);
+    let router = build_router(container_deployer.clone(), image_builder, &config);
 
-    // Start heartbeat background loop
-    spawn_heartbeat_loop(&config);
+    // Start heartbeat background loop (with deployer for container inventory on first beat)
+    spawn_heartbeat_loop(&config, container_deployer);
 
     let listener = tokio::net::TcpListener::bind(&config.listen_address)
         .await

crates/temps-auth/src/auth_service.rs

@@ -1042,7 +1042,7 @@ mod tests {
 
         let request1 = RegisterRequest {
             email: "Test@Example.Com".to_string(),
-            password: "password123".to_string(),
+            password: "Password123!".to_string(),
             name: "Test User".to_string(),
         };
 
@@ -1051,7 +1051,7 @@ mod tests {
 
         let request2 = RegisterRequest {
             email: "TEST@EXAMPLE.COM".to_string(),
-            password: "password456".to_string(),
+            password: "Password456!".to_string(),
             name: "Another User".to_string(),
         };
 

crates/temps-deployer/src/docker.rs

@@ -1135,6 +1135,7 @@ impl ContainerDeployer for DockerRuntime {
 
         let state = container.state.unwrap_or_default();
         let config = container.config.unwrap_or_default();
+        let container_labels = config.labels.clone().unwrap_or_default();
 
         // Parse environment variables
         let env_vars = config
@@ -1202,6 +1203,7 @@ impl ContainerDeployer for DockerRuntime {
             ports: port_mappings,
             environment_vars: env_vars,
             restart_count: container.restart_count,
+            labels: container_labels,
         })
     }
 

crates/temps-deployer/src/lib.rs

@@ -248,6 +248,9 @@ pub struct ContainerInfo {
     pub ports: Vec<PortMapping>,
     pub environment_vars: HashMap<String, String>,
     pub restart_count: Option<i64>,
+    /// Docker labels set on the container (e.g., `sh.temps.managed`, `sh.temps.project_id`).
+    #[serde(default)]
+    pub labels: HashMap<String, String>,
 }
 
 /// Container performance statistics (CPU, memory, network)
@@ -686,6 +689,7 @@ mod tests {
             }],
             environment_vars: env_vars,
             restart_count: Some(0),
+            labels: HashMap::new(),
         };
 
         assert_eq!(info.container_id, "abc123");

crates/temps-deployer/src/remote.rs

@@ -253,9 +253,7 @@ impl ContainerDeployer for RemoteNodeDeployer {
     }
 
     async fn list_containers(&self) -> Result<Vec<ContainerInfo>, DeployerError> {
-        Err(DeployerError::Other(
-            "List containers not yet supported on remote nodes".into(),
-        ))
+        self.agent_get("/agent/containers").await
     }
 
     async fn get_container_logs(&self, container_id: &str) -> Result<String, DeployerError> {