Make all service launches bulletproof with zero port conflicts

Port conflict prevention (ports.py):
- Added controller_port() and is_controller_port() helpers
- reserved_ports() now always includes the controller's own port
- next_available_port() can never return the controller's port

Import conflict auto-resolution (registry_import.py):
- Import now auto-reassigns ports that collide with the controller,
  other registered services, or ports already in use — instead of
  silently allowing the conflict and only logging a warning
- local_url and healthcheck_url are automatically updated when a port
  is reassigned during import
- local_url is auto-populated from port when not explicitly provided

Start-time self-healing (process_manager.py):
- start_service() now guards against the controller port in addition
  to the existing port-in-use check, with automatic reassignment

Conflict resolver (api_ports.py):
- resolve-conflicts endpoint now detects and fixes controller port
  collisions with a distinct "controller_port_collision" reason
- Auto-populates local_url for any service that has a port but is
  missing a URL

Sample data fixes:
- Fixed Job Tracker port from 3000→3002 to avoid duplicate with
  Eicher Monorepo Backend (both were on 3000)

Pre-existing bug fix (api_import.py):
- Fixed f-string backslash syntax error on line 1036 that prevented
  the entire module from loading (broke imports and FastAPI startup)

https://claude.ai/code/session_01BX7BsftaQ3WdmdGGqKjf7n

Author: claude

local_nexus_controller/routers/api_import.py

@@ -1033,9 +1033,11 @@ def github_localize_pr(req: GitHubLocalizePrRequest) -> dict:
         if not str(svc.get("working_directory") or "").strip():
             svc["working_directory"] = "{REPO_ROOT}"
         if not str(svc.get("start_command") or "").strip():
-            svc["start_command"] = f"powershell -ExecutionPolicy Bypass -File {start_ps1_path.replace('/', '\\\\')}"
+            win_start = start_ps1_path.replace("/", "\\")
+            svc["start_command"] = f"powershell -ExecutionPolicy Bypass -File {win_start}"
         if not str(svc.get("stop_command") or "").strip():
-            svc["stop_command"] = f"powershell -ExecutionPolicy Bypass -File {stop_ps1_path.replace('/', '\\\\')}"
+            win_stop = stop_ps1_path.replace("/", "\\")
+            svc["stop_command"] = f"powershell -ExecutionPolicy Bypass -File {win_stop}"
 
         tech = svc.get("tech_stack")
         if not isinstance(tech, list):

local_nexus_controller/routers/api_ports.py

@@ -6,7 +6,7 @@
 from local_nexus_controller.db import get_session
 from local_nexus_controller.models import Service
 from local_nexus_controller.security import require_token
-from local_nexus_controller.services.ports import is_port_in_use, next_available_port, port_map
+from local_nexus_controller.services.ports import is_controller_port, is_port_in_use, next_available_port, port_map
 
 
 router = APIRouter()
@@ -99,24 +99,25 @@ def _update_urls(svc: Service, old_port: int | None, new_port: int) -> None:
     # Re-load after commits so subsequent checks see updated ports
     services = list(session.exec(select(Service).order_by(Service.created_at)))
 
-    # 2) Resolve "port in use but service not running" conflicts
+    # 2) Resolve controller-port collisions and "port in use but service not running" conflicts
     for svc in services:
         if svc.port is None:
             continue
         port = int(svc.port)
-        if svc.status == "running" and svc.process_pid is not None:
-            continue
-        if is_port_in_use(host, port):
+        if is_controller_port(port) or (
+            is_port_in_use(host, port) and not (svc.status == "running" and svc.process_pid is not None)
+        ):
             old_port = port
             new_port = next_available_port(session, host=host)
             svc.port = new_port
             _update_urls(svc, old_port=old_port, new_port=new_port)
             session.add(svc)
+            reason = "controller_port_collision" if is_controller_port(old_port) else "port_in_use_on_host"
             changes.append(
                 {
                     "service_id": svc.id,
                     "name": svc.name,
-                    "reason": "port_in_use_on_host",
+                    "reason": reason,
                     "old_port": old_port,
                     "new_port": new_port,
                     "local_url": svc.local_url,
@@ -125,6 +126,14 @@ def _update_urls(svc: Service, old_port: int | None, new_port: int) -> None:
 
     session.commit()
 
+    # 2b) Auto-populate local_url for services that have a port but no URL
+    services = list(session.exec(select(Service).order_by(Service.created_at)))
+    for svc in services:
+        if svc.port is not None and not svc.local_url:
+            svc.local_url = f"http://{host}:{svc.port}"
+            session.add(svc)
+    session.commit()
+
     # 3) Update dependent Vite apps: VITE_API_BASE_URL -> dependency local_url
     services = list(session.exec(select(Service).order_by(Service.name)))
     by_name = {s.name: s for s in services}

local_nexus_controller/services/ports.py

@@ -22,14 +22,26 @@ def is_port_in_use(host: str, port: int, timeout_s: float = 0.25) -> bool:
         return False
 
 
+def controller_port() -> int:
+    """Return the port the Nexus Controller itself is running on."""
+    return settings.port
+
+
 def reserved_ports(session: Session) -> set[int]:
     ports: set[int] = set()
+    # Always reserve the controller's own port so no service collides with it.
+    ports.add(controller_port())
     for p in session.exec(select(Service.port).where(Service.port.is_not(None))):
         if p is not None:
             ports.add(int(p))
     return ports
 
 
+def is_controller_port(port: int) -> bool:
+    """Check if a port is the Nexus Controller's own port."""
+    return port == controller_port()
+
+
 def next_available_port(
     session: Session,
     host: str = "127.0.0.1",

local_nexus_controller/services/process_manager.py

@@ -10,7 +10,7 @@
 
 from local_nexus_controller.models import Service
 from local_nexus_controller.settings import settings
-from local_nexus_controller.services.ports import is_port_in_use, next_available_port
+from local_nexus_controller.services.ports import is_controller_port, is_port_in_use, next_available_port
 
 
 def _now_utc() -> datetime:
@@ -59,18 +59,22 @@ def start_service(session: Session, service: Service) -> Service:
     if service.status == "running":
         return service
 
-    # Self-heal: if the reserved port is currently in use, reassign.
-    if service.port is not None and is_port_in_use("127.0.0.1", int(service.port)) and service.status != "running":
-        old_port = int(service.port)
-        new_port = next_available_port(session, host="127.0.0.1")
-        service.port = new_port
-        if service.local_url and f":{old_port}" in service.local_url:
-            service.local_url = service.local_url.replace(f":{old_port}", f":{new_port}")
-        if service.healthcheck_url and f":{old_port}" in service.healthcheck_url:
-            service.healthcheck_url = service.healthcheck_url.replace(f":{old_port}", f":{new_port}")
-        session.add(service)
-        session.commit()
-        session.refresh(service)
+    # Self-heal: reassign port if it conflicts with the controller or is already in use.
+    if service.port is not None:
+        needs_reassign = is_controller_port(int(service.port))
+        if not needs_reassign and is_port_in_use("127.0.0.1", int(service.port)) and service.status != "running":
+            needs_reassign = True
+        if needs_reassign:
+            old_port = int(service.port)
+            new_port = next_available_port(session, host="127.0.0.1")
+            service.port = new_port
+            if service.local_url and f":{old_port}" in service.local_url:
+                service.local_url = service.local_url.replace(f":{old_port}", f":{new_port}")
+            if service.healthcheck_url and f":{old_port}" in service.healthcheck_url:
+                service.healthcheck_url = service.healthcheck_url.replace(f":{old_port}", f":{new_port}")
+            session.add(service)
+            session.commit()
+            session.refresh(service)
 
     log_path = _service_log_path(service)
     log_path.parent.mkdir(parents=True, exist_ok=True)

local_nexus_controller/services/registry_import.py

@@ -12,7 +12,7 @@
     KeyRef,
     Service,
 )
-from local_nexus_controller.services.ports import is_port_in_use, next_available_port
+from local_nexus_controller.services.ports import is_controller_port, is_port_in_use, next_available_port, reserved_ports
 
 
 def _now_utc() -> datetime:
@@ -75,27 +75,60 @@ def import_bundle(session: Session, bundle: ImportBundle, host_for_port_checks:
         db = _upsert_database(session, bundle.database)
         database_id = db.id
 
-    # Port assignment
+    # Port assignment with automatic conflict resolution
     port: int | None = bundle.service.port
     if bundle.requested_port is not None:
         port = int(bundle.requested_port)
-    if port is None and bundle.auto_assign_port:
-        port = next_available_port(session, host=host_for_port_checks)
 
-    # Check conflicts
+    # Resolve port conflicts automatically instead of silently allowing them
     if port is not None:
-        # Only warn if the port is reserved by a *different* service.
-        conflicting = session.exec(
-            select(Service).where(Service.port == int(port), Service.name != bundle.service.name)
-        ).first()
-        if conflicting:
-            warnings.append(f"Port {port} is already reserved in the registry (by '{conflicting.name}').")
-        if is_port_in_use(host_for_port_checks, port):
-            warnings.append(f"Port {port} appears to be in use on {host_for_port_checks}.")
+        needs_reassign = False
+
+        # Never allow a service to use the controller's own port
+        if is_controller_port(port):
+            warnings.append(f"Port {port} is the Nexus Controller port; auto-reassigning.")
+            needs_reassign = True
+
+        # Check if another service already has this port
+        if not needs_reassign:
+            conflicting = session.exec(
+                select(Service).where(Service.port == int(port), Service.name != bundle.service.name)
+            ).first()
+            if conflicting:
+                warnings.append(f"Port {port} already reserved by '{conflicting.name}'; auto-reassigning.")
+                needs_reassign = True
+
+        # Check if the port is physically in use by something else
+        if not needs_reassign and is_port_in_use(host_for_port_checks, port):
+            warnings.append(f"Port {port} in use on {host_for_port_checks}; auto-reassigning.")
+            needs_reassign = True
+
+        if needs_reassign:
+            old_port = port
+            port = next_available_port(session, host=host_for_port_checks)
+            warnings.append(f"Reassigned from port {old_port} to {port}.")
+
+    if port is None and bundle.auto_assign_port:
+        port = next_available_port(session, host=host_for_port_checks)
 
     # Upsert service
     svc_dict = bundle.service.model_dump()  # type: ignore[attr-defined]
     svc_dict["port"] = port
+
+    # Auto-populate local_url from port when missing
+    if not svc_dict.get("local_url") and port is not None:
+        svc_dict["local_url"] = f"http://127.0.0.1:{port}"
+    # If port changed from what was in the bundle, update local_url to match
+    elif svc_dict.get("local_url") and bundle.service.port is not None and port != bundle.service.port:
+        old_port_str = f":{bundle.service.port}"
+        if old_port_str in svc_dict["local_url"]:
+            svc_dict["local_url"] = svc_dict["local_url"].replace(old_port_str, f":{port}")
+    # Same for healthcheck_url
+    if svc_dict.get("healthcheck_url") and bundle.service.port is not None and port != bundle.service.port:
+        old_port_str = f":{bundle.service.port}"
+        if old_port_str in svc_dict["healthcheck_url"]:
+            svc_dict["healthcheck_url"] = svc_dict["healthcheck_url"].replace(old_port_str, f":{port}")
+
     if database_id is not None:
         svc_dict["database_id"] = database_id
     svc_dict.setdefault("status", "stopped")

sample_data/import_existing_bundle.json

@@ -104,9 +104,9 @@
       "tech_stack": ["nodejs", "nextjs", "typescript", "tailwind", "prisma", "postgres"],
       "dependencies": [],
       "config_paths": ["C:/Users/nedpe/job-tracker/.env"],
-      "port": 3000,
-      "local_url": "http://127.0.0.1:3000",
-      "healthcheck_url": "http://127.0.0.1:3000/api/version",
+      "port": 3002,
+      "local_url": "http://127.0.0.1:3002",
+      "healthcheck_url": "http://127.0.0.1:3002/api/version",
       "working_directory": "C:/Users/nedpe/job-tracker",
       "start_command": "npm run dev -- -p {PORT}",
       "stop_command": "",
@@ -139,7 +139,7 @@
       { "key_name": "PG password", "env_var": "PGPASSWORD", "description": "Optional alternative DB config (Railway-style)" },
       { "key_name": "PG database", "env_var": "PGDATABASE", "description": "Optional alternative DB config (Railway-style)" }
     ],
-    "requested_port": 3000,
+    "requested_port": 3002,
     "auto_assign_port": false,
     "auto_create_db": true,
     "meta": { "source": "C:/Users/nedpe/job-tracker" }