Add project port health checks

README.md

@@ -194,6 +194,14 @@ health:
   required_env:
     - DATABASE_URL
     - REDIS_URL
+  required_ports:
+    - name: postgres
+      host: 127.0.0.1
+      port: 5432
+      state: listening
+    - name: app
+      port: 8000
+      state: free
 
 activate:
   source:
@@ -230,6 +238,13 @@ the project needs in the local shell. `basectl check <project>` and
 non-empty. Base only checks presence; it never reads, prints, or logs the
 variable values.
 
+The optional top-level `health.required_ports` list declares local TCP ports
+the project expects to be either `listening` or `free`. Each entry must include
+`port` and `state`; `host` defaults to `127.0.0.1`, and `name` is an optional
+display label. Base checks whether a TCP connection succeeds on the declared
+endpoint. It does not start or stop services, inspect process ownership, or
+perform Docker Compose health checks.
+
 The optional top-level `activate.source` list declares project-root-relative
 shell scripts to source when `basectl activate <project>` starts the runtime
 shell. Base sources those scripts after the Base runtime and project virtual

cli/python/base_setup/engine.py

@@ -24,6 +24,7 @@
 from .delegates import reconcile_mise
 from .errors import ArtifactError
 from .health import check_required_env
+from .health import check_required_ports
 from .ide import check_ide_extensions
 from .ide import check_ide_installs
 from .ide import check_ide_settings
@@ -242,6 +243,7 @@ def manifest_checks(default_manifest: BaseManifest, manifest: BaseManifest) -> t
         checks.append(check_mise(effective_manifest))
 
     checks.extend(check_required_env(effective_manifest))
+    checks.extend(check_required_ports(effective_manifest))
     checks.extend(check_demo(effective_manifest))
     checks.extend(check_ide_installs(effective_manifest))
     checks.extend(check_ide_extensions(effective_manifest))

cli/python/base_setup/health.py

@@ -1,15 +1,21 @@
 from __future__ import annotations
 
 import os
+import socket
 
 from .checks import ArtifactCheck
 from .manifest import BaseManifest
+from .manifest import PortHealthConfig
 
 
 def check_required_env(manifest: BaseManifest) -> list[ArtifactCheck]:
     return [check_required_env_var(env_name) for env_name in manifest.health.required_env]
 
 
+def check_required_ports(manifest: BaseManifest) -> list[ArtifactCheck]:
+    return [check_required_port(port_config) for port_config in manifest.health.required_ports]
+
+
 def check_required_env_var(env_name: str) -> ArtifactCheck:
     if os.environ.get(env_name, ""):
         return ArtifactCheck(
@@ -26,3 +32,50 @@ def check_required_env_var(env_name: str) -> ArtifactCheck:
         fix=f"Set {env_name} in your shell, .env, or secrets manager.",
         finding_id="BASE-H001",
     )
+
+
+def check_required_port(port_config: PortHealthConfig) -> ArtifactCheck:
+    label = port_config.name or f"{port_config.host}:{port_config.port}"
+    endpoint = f"{port_config.host}:{port_config.port}"
+    listening = tcp_port_is_listening(port_config.host, port_config.port)
+
+    if port_config.state == "listening":
+        if listening:
+            return ArtifactCheck(
+                name=label,
+                ok=True,
+                message=f"TCP port '{label}' is listening on {endpoint}.",
+                fix="",
+                finding_id="BASE-H002",
+            )
+        return ArtifactCheck(
+            name=label,
+            ok=False,
+            message=f"TCP port '{label}' is not listening on {endpoint}.",
+            fix=f"Start the service that should listen on {endpoint}.",
+            finding_id="BASE-H002",
+        )
+
+    if not listening:
+        return ArtifactCheck(
+            name=label,
+            ok=True,
+            message=f"TCP port '{label}' is free on {endpoint}.",
+            fix="",
+            finding_id="BASE-H002",
+        )
+    return ArtifactCheck(
+        name=label,
+        ok=False,
+        message=f"TCP port '{label}' is already listening on {endpoint}.",
+        fix=f"Stop the process using {endpoint} or choose a different project port.",
+        finding_id="BASE-H002",
+    )
+
+
+def tcp_port_is_listening(host: str, port: int, timeout_seconds: float = 0.2) -> bool:
+    try:
+        with socket.create_connection((host, port), timeout=timeout_seconds):
+            return True
+    except OSError:
+        return False

cli/python/base_setup/manifest.py

@@ -23,6 +23,7 @@
 CURRENT_MANIFEST_SCHEMA_VERSION = 1
 ENVIRONMENT_VARIABLE_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 COMMAND_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]*$")
+PORT_HEALTH_STATES = {"free", "listening"}
 
 
 class ManifestError(ValueError):
@@ -56,9 +57,18 @@ class DemoConfig:
     description: str | None = None
 
 
+@dataclass(frozen=True)
+class PortHealthConfig:
+    port: int
+    state: str
+    name: str | None = None
+    host: str = "127.0.0.1"
+
+
 @dataclass(frozen=True)
 class HealthConfig:
     required_env: tuple[str, ...] = ()
+    required_ports: tuple[PortHealthConfig, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -267,12 +277,15 @@ def _read_health(path: Path, health_data: Any) -> HealthConfig:
     if not isinstance(health_data, dict):
         raise ManifestError(f"{path}: health must be a mapping when provided.")
 
-    allowed_keys = {"required_env"}
+    allowed_keys = {"required_env", "required_ports"}
     unknown_keys = sorted(set(health_data) - allowed_keys)
     if unknown_keys:
         raise ManifestError(f"{path}: health has unsupported keys: {', '.join(unknown_keys)}.")
 
-    return HealthConfig(required_env=_read_required_env(path, health_data.get("required_env", [])))
+    return HealthConfig(
+        required_env=_read_required_env(path, health_data.get("required_env", [])),
+        required_ports=_read_required_ports(path, health_data.get("required_ports", [])),
+    )
 
 
 def _read_commands(path: Path, commands_data: Any) -> dict[str, str]:
@@ -356,6 +369,116 @@ def _read_required_env(path: Path, required_env_data: Any) -> tuple[str, ...]:
     return tuple(required_env)
 
 
+def _read_required_ports(path: Path, required_ports_data: Any) -> tuple[PortHealthConfig, ...]:
+    if required_ports_data is None:
+        return ()
+    if not isinstance(required_ports_data, list):
+        raise ManifestError(f"{path}: health.required_ports must be a list when provided.")
+
+    required_ports: list[PortHealthConfig] = []
+    seen_endpoints: set[tuple[str, int]] = set()
+    seen_names: set[str] = set()
+    for index, port_data in enumerate(required_ports_data, start=1):
+        required_ports.append(
+            _read_required_port(path, index, port_data, seen_endpoints, seen_names)
+        )
+
+    return tuple(required_ports)
+
+
+def _read_required_port(
+    path: Path,
+    index: int,
+    port_data: Any,
+    seen_endpoints: set[tuple[str, int]],
+    seen_names: set[str],
+) -> PortHealthConfig:
+    if not isinstance(port_data, dict):
+        raise ManifestError(f"{path}: health.required_ports[{index}] must be a mapping.")
+
+    allowed_keys = {"name", "host", "port", "state"}
+    unknown_keys = sorted(set(port_data) - allowed_keys)
+    if unknown_keys:
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}] has unsupported keys: "
+            f"{', '.join(unknown_keys)}."
+        )
+
+    port = _read_required_port_number(path, index, port_data.get("port"))
+    state = _read_required_port_state(path, index, port_data.get("state"))
+    name = _read_required_port_name(path, index, port_data.get("name"), seen_names)
+    host = _read_required_port_host(path, index, port_data.get("host", "127.0.0.1"))
+    endpoint = (host, port)
+    if endpoint in seen_endpoints:
+        raise ManifestError(f"{path}: health.required_ports[{index}] duplicates '{host}:{port}'.")
+    seen_endpoints.add(endpoint)
+    return PortHealthConfig(port=port, state=state, name=name, host=host)
+
+
+def _read_required_port_number(path: Path, index: int, port_data: Any) -> int:
+    if isinstance(port_data, bool) or not isinstance(port_data, int):
+        raise ManifestError(f"{path}: health.required_ports[{index}].port must be an integer.")
+    if port_data < 1 or port_data > 65535:
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}].port must be between 1 and 65535."
+        )
+    return port_data
+
+
+def _read_required_port_state(path: Path, index: int, state_data: Any) -> str:
+    if not isinstance(state_data, str) or not state_data.strip():
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}].state must be a non-empty string."
+        )
+    state = state_data.strip()
+    if state not in PORT_HEALTH_STATES:
+        supported_states = ", ".join(sorted(PORT_HEALTH_STATES))
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}].state must be one of: {supported_states}."
+        )
+    return state
+
+
+def _read_required_port_name(
+    path: Path,
+    index: int,
+    name_data: Any,
+    seen_names: set[str],
+) -> str | None:
+    if name_data is None:
+        return None
+    if not isinstance(name_data, str) or not name_data.strip():
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}].name must be a non-empty string."
+        )
+    name = name_data.strip()
+    if _has_control_line_break(name):
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}].name must not contain control line breaks."
+        )
+    if name in seen_names:
+        raise ManifestError(f"{path}: health.required_ports[{index}].name duplicates '{name}'.")
+    seen_names.add(name)
+    return name
+
+
+def _read_required_port_host(path: Path, index: int, host_data: Any) -> str:
+    if not isinstance(host_data, str) or not host_data.strip():
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}].host must be a non-empty string."
+        )
+    host = host_data.strip()
+    if _has_control_line_break(host):
+        raise ManifestError(
+            f"{path}: health.required_ports[{index}].host must not contain control line breaks."
+        )
+    return host
+
+
+def _has_control_line_break(value: str) -> bool:
+    return any(separator in value for separator in ("\0", "\n", "\r"))
+
+
 def _read_ide_config(path: Path, ide_name: str, config_data: Any) -> IdeConfig:
     if config_data is None:
         config_data = {}