Hardening the canyonos deploy command (CAN-316)

.github/workflows/cli-release-tag.yml

@@ -70,4 +70,67 @@ jobs:
           UV_PUBLISH_TOKEN: ${{ secrets.PYPI_API_TOKEN }}
         run: uv publish
 
+  update-brew-formula:
+    needs: [bump, release]
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+    env:
+      RELEASE_TAG: ${{ needs.bump.outputs.tag }}
+    steps:
+      - name: Checkout main
+        uses: actions/checkout@v4
+        with:
+          ref: main
+
+      - name: Download release binaries
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          mkdir -p bin
+          gh release download "$RELEASE_TAG" --repo "${{ github.repository }}" --dir bin --pattern 'canyonos-*'
+
+      - name: Update version and sha256 in Formula/canyonos.rb
+        run: |
+          python3 - <<'PY'
+          import hashlib
+          import os
+          import pathlib
+          import re
+
+          version = os.environ["RELEASE_TAG"].removeprefix("cli-v")
+          assets = [
+              "canyonos-macos-arm64",
+              "canyonos-macos-x86_64",
+              "canyonos-linux-arm64",
+              "canyonos-linux-x86_64",
+          ]
+
+          formula = pathlib.Path("Formula/canyonos.rb")
+          text = formula.read_text()
+          text = re.sub(r'version ".*"', f'version "{version}"', text, count=1)
+
+          for asset in assets:
+              sha256 = hashlib.sha256(pathlib.Path(f"bin/{asset}").read_bytes()).hexdigest()
+              text = re.sub(
+                  rf'(url "[^"]*/{re.escape(asset)}"\n\s*sha256 ")[a-f0-9]+(")',
+                  rf"\g<1>{sha256}\g<2>",
+                  text,
+              )
+
+          formula.write_text(text)
+          PY
+
+      - name: Commit and push formula update
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+          git add Formula/canyonos.rb
+          if git diff --cached --quiet; then
+            echo "No formula changes to commit"
+          else
+            git commit -m "brew: update canyonos formula to $RELEASE_TAG [skip release]"
+            git push origin HEAD:main
+          fi
+
 # Calls cli-release.yml directly (a GITHUB_TOKEN tag push won't retrigger it); "[skip release]" is a defensive backstop against the same push looping.

README.md

@@ -57,6 +57,7 @@ Use any of the following package managers to install the canyonos CLI (curl, bre
 curl -fsSL https://raw.githubusercontent.com/CanyonCodeCoreAI/canyoncodecore/main/cli/install.sh | sh
 # OR
 brew tap CanyonCodeCoreAI/canyonos https://github.com/CanyonCodeCoreAI/canyoncodecore
+brew trust --formula CanyonCodeCoreAI/canyonos/canyonos
 brew install canyonos
 # OR
 uv tool install canyonos
@@ -157,7 +158,11 @@ Example Success Message:
 
 ### 4. Sending requests to the workflow
 
-Upon running the deploy command, canyonos automatically generates a REST API endpoint for the workflow. Send requests to this endpoint to trigger the workflow:
+Upon running the deploy command, canyonos automatically generates a REST API endpoint for the workflow. 
+
+For verifying the workflow has been deployed fine, run `canyonos test "A test query"` to send a query through the workflow.
+
+For manually sending requests, use the given endpoint to trigger the workflow:
 
 ```bash
 curl -X POST http://localhost:8000/main \

canyonos_core/cli.py

@@ -51,7 +51,70 @@ def _load_config(config_path):
     import yaml
 
     with open(config_path, "r") as f:
-        return yaml.safe_load(f)
+        config = yaml.safe_load(f)
+    # Everything below here till "return config" is basically just checks to make sure the folder is correct
+    if not isinstance(config, dict):
+        raise RuntimeError(f"Config must contain a YAML mapping: {config_path}")
+    agents = config.get("agents", [])
+    if not isinstance(agents, list) or not all(isinstance(agent, dict) for agent in agents):
+        raise RuntimeError(f"Config `agents` must be a list of mappings: {config_path}")
+    names = [agent.get("name") for agent in agents]
+    if any(not isinstance(name, str) or not name.strip() for name in names):
+        raise RuntimeError(f"Every configured agent must have a non-empty name: {config_path}")
+    names_by_key = {}
+    for name in names:
+        names_by_key.setdefault(name.casefold(), []).append(name)
+    duplicates = sorted(
+        "/".join(group) for group in names_by_key.values() if len(group) > 1
+    )
+    if duplicates:
+        raise RuntimeError(f"Duplicate agent names in {config_path}: {', '.join(duplicates)}")
+
+    for agent in agents:
+        name = agent["name"]
+        provider = agent.get("provider", "local")
+        if not isinstance(provider, str) or provider.casefold() not in {"local", "ec2"}:
+            raise RuntimeError(
+                f"Agent {name} has unsupported provider {provider!r}; use `local` or `EC2`."
+            )
+        agent["provider"] = "EC2" if provider.casefold() == "ec2" else "local"
+
+        replicas = agent.get("replicas", 1)
+        if isinstance(replicas, bool) or not isinstance(replicas, int) or replicas < 1:
+            raise RuntimeError(f"Agent {name} must have a positive integer `replicas` value.")
+        if (
+            agent["provider"] == "local"
+            and agent.get("type", "agent") == "workflow"
+            and replicas > 1
+        ):
+            raise RuntimeError(
+                f"Local workflow {name} cannot use replicas > 1 because every replica "
+                "would publish the same `api_port`."
+            )
+
+        for field in ("host_port", "port", "redis_port", "api_port", "dashboard_port"):
+            value = agent.get(field)
+            if value is not None and (
+                isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 65535
+            ):
+                raise RuntimeError(
+                    f"Agent {name} must have an integer `{field}` between 1 and 65535."
+                )
+
+        resources = agent.get("resources", {})
+        if not isinstance(resources, dict):
+            raise RuntimeError(f"Agent {name} `resources` must be a mapping.")
+        for field in ("cpu", "memory", "gpu"):
+            value = resources.get(field)
+            if value is not None and (
+                isinstance(value, bool)
+                or not isinstance(value, (int, float))
+                or value <= 0
+            ):
+                raise RuntimeError(
+                    f"Agent {name} resource `{field}` must be a positive number."
+                )
+    return config
 
 
 def _artifact_prefix(root):
@@ -237,6 +300,17 @@ def _run_build(config_path):
     source_root = os.path.join(artifact_root, SOURCE_DIR_NAME) if prefix else project_dir
     package_dir = _get_package_dir()
 
+    missing_sources = []
+    for agent in agents:
+        source_field = "workflow_file" if agent.get("type", "agent") == "workflow" else "entrypoint"
+        source_path = agent.get(source_field)
+        if not isinstance(source_path, str) or not source_path:
+            missing_sources.append(f"{agent['name']}: missing `{source_field}`")
+        elif not os.path.isfile(os.path.join(source_root, source_path)):
+            missing_sources.append(f"{agent['name']}: {source_path} not found")
+    if missing_sources:
+        raise RuntimeError("Cannot build configured sources: " + "; ".join(missing_sources))
+
     # -------------------------------------------------------------- #
     #  Step 1: Discover agent YAML files and generate Python stubs    #
     # -------------------------------------------------------------- #

canyonos_core/controller/cloud_provider_logic/Local/_runtime.py

@@ -8,6 +8,7 @@
 
 import logging
 import os
+import socket
 
 from canyonos_core.controller.utils.container_names import container_name
 from canyonos_core.controller.utils.env_file import env_file_args
@@ -19,6 +20,7 @@
 PROVIDER = "local"
 MAX_PORT_ATTEMPTS = 50
 NETWORK = "canyonos-local"
+HOST_GATEWAY = "host.docker.internal"
 _controller = None
 
 
@@ -36,6 +38,16 @@ def validate_config():
     return None
 
 
+def _port_check(host, port):
+    """Preflight check to test if the port (8080) is available, fails immediately if not instead of failing later"""
+    probe_host = HOST_GATEWAY if _is_local_host(host) else host
+    try:
+        with socket.create_connection((probe_host, int(port)), timeout=0.25):
+            return True
+    except OSError:
+        return False
+
+
 def provision_instance(spec, replica_index, next_host_port):
     host = spec.get("host", DEFAULT_HOST)
     host_port = int(spec.get("host_port", spec.get("port", next_host_port(host))))
@@ -73,6 +85,12 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         )
         _require_controller()._run_cmd(["docker", "rm", "-f", runtime_id], host, user)
 
+    if ctrl_type == "workflow" and _port_check(host, spec.get("api_port", 8080)):
+        raise RuntimeError(
+            f"Cannot launch {runtime_id}: workflow api_port "
+            f"{spec.get('api_port', 8080)} is already in use on {host}."
+        )
+
     for attempt in range(MAX_PORT_ATTEMPTS):
         cmd = [
             "docker",
@@ -140,7 +158,8 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
 
         if result.returncode == 0:
             break
-        if "port is already allocated" in (result.stderr or ""):
+        error = (result.stderr or "").lower()
+        if "port is already allocated" in error or "address already in use" in error:
             # `docker run` leaves a `Created`-but-never-started container behind
             # under this name when the port bind fails. Remove it before
             # retrying with a new port, or the retry hits a name conflict
@@ -189,7 +208,12 @@ def terminate_instance(instance):
         instance.get("user"),
     )
     if result.returncode != 0:
-        logger.warning("Failed to remove runtime %s", runtime_id)
+        detail = (result.stderr or result.stdout or "").strip()
+        if "no such container" not in detail.casefold():
+            raise RuntimeError(
+                f"Failed to remove runtime {runtime_id}: "
+                f"{detail or f'exit code {result.returncode}'}"
+            )
 
 
 def routing_endpoint_for(instance):