feat(core): declare and enforce the manifest and agent-YAML schema

examples/portfolio/config/global_controller.yaml

@@ -108,9 +108,6 @@ env_file: .env
 poll_interval: 4
 logs: true
 
-# Secrets handed to every agent container (AWS_BEARER_TOKEN_BEDROCK for the LLM proxy).
-env_file: .env
-
 redis:
   host: localhost
   port: 6379

examples/text2sql/config/global_controller.yaml

@@ -105,4 +105,14 @@ logs: false
 redis:
   host: localhost
   port: 6379
-  db: 0
+  db: 0
+
+# EC2 defaults for the `provider: EC2` replicas above.
+ec2:
+  region: ${EC2_REGION}
+  ami_id: ${EC2_AMI_ID}
+  subnet_id: ${EC2_SUBNET_ID}
+  security_group_ids:
+    - ${EC2_SECURITY_GROUP_ID}
+  ssh_user: ${EC2_SSH_USER}
+  ssh_private_key_path: ${EC2_SSH_PRIVATE_KEY_PATH}

packages/cli/canyonos/constants.py

@@ -86,14 +86,30 @@ def local_redis_port(config_path):
     """Host port the local node's Redis is published on, or the default."""
     try:
         with open(config_path) as f:
-            config = yaml.safe_load(f) or {}
+            config = yaml.safe_load(f)
     except (OSError, yaml.YAMLError):
         return DEFAULT_REDIS_PORT
+    if not isinstance(config, dict):
+        return DEFAULT_REDIS_PORT
 
-    for agent in config.get("agents") or []:
-        if agent.get("host", "localhost") in ("localhost", "127.0.0.1"):
-            return agent.get("redis_port", DEFAULT_REDIS_PORT)
-    return (config.get("redis") or {}).get("port", DEFAULT_REDIS_PORT)
+    agents = config.get("agents")
+    for agent in agents if isinstance(agents, list) else []:
+        if isinstance(agent, dict) and agent.get("host", "localhost") in (
+            "localhost",
+            "127.0.0.1",
+        ):
+            return _port_or_default(agent.get("redis_port", DEFAULT_REDIS_PORT))
+    redis = config.get("redis")
+    if not isinstance(redis, dict):
+        return DEFAULT_REDIS_PORT
+    return _port_or_default(redis.get("port", DEFAULT_REDIS_PORT))
+
+
+def _port_or_default(value):
+    """`value` when it is a usable TCP port, else the default the runtime falls back to."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        return DEFAULT_REDIS_PORT
+    return value if 1 <= value <= 65535 else DEFAULT_REDIS_PORT
 
 
 def _source_root(config_path):

packages/cli/canyonos/doctor.py

@@ -95,12 +95,12 @@ def _gc_check(state, status):
 def _redis_check():
     """TCP reachability only -- not a real PING, but enough to say something's
     listening where the local provider and dashboard both expect Redis."""
-    port = local_redis_port(default_config_path())
+    redis_port = local_redis_port(default_config_path())
     return _print_check(
         "Redis",
-        port_in_use(int(port)),
-        f"127.0.0.1:{port}",
-        f"nothing is listening on {port} -- redeploy, or check `docker ps`",
+        port_in_use(redis_port),
+        f"127.0.0.1:{redis_port}",
+        f"nothing is listening on {redis_port} -- redeploy, or check `docker ps`",
     )
 
 

packages/cli/tests/test_cli_entry.py

@@ -0,0 +1,18 @@
+import subprocess
+import sys
+
+
+def test_the_canyonos_entry_point_imports_in_a_fresh_interpreter():
+    """`canyonos` is `cli:main`; if `import cli` fails, every command fails.
+
+    Run in a fresh interpreter: inside the suite another test may already have
+    imported the modules involved, which is how a broken import once passed.
+    """
+    result = subprocess.run(
+        [sys.executable, "-c", "import cli; assert callable(cli.main)"],
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+
+    assert result.returncode == 0, result.stderr

packages/cli/tests/test_local_redis_port.py

@@ -37,6 +37,20 @@ def test_the_top_level_redis_section_is_the_fallback(self):
     def test_an_unreadable_config_falls_back_to_the_default(self):
         self.assertEqual(local_redis_port("/nonexistent/global_controller.yaml"), 6379)
 
+    def test_a_malformed_config_falls_back_to_the_default(self):
+        for config in (
+            {"agents": [{"name": "A", "redis_port": "abc"}]},
+            {"agents": [{"name": "A", "redis_port": "${REDIS_PORT}"}]},
+            {"agents": [{"name": "A", "redis_port": None}]},
+            {"agents": [{"name": "A", "redis_port": 70000}]},
+            {"agents": ["just-a-string"]},
+            {"agents": "not-a-list", "redis": {"port": True}},
+            {"agents": [], "redis": "not-a-mapping"},
+            ["not", "a", "mapping"],
+        ):
+            with self.subTest(config=config):
+                self.assertEqual(local_redis_port(self._config(config)), 6379)
+
 
 if __name__ == "__main__":
     unittest.main()

packages/core/canyonos_core/cli.py

@@ -16,19 +16,19 @@
 import subprocess
 import sys
 
+from canyonos_core.controller.utils.config_env import load_config
 from canyonos_core.controller.utils.env_file import resolve_env_file
+from canyonos_core.schema import (
+    DependencyPinConflict,
+    check_project,
+    declarations_by_name,
+    render_violation,
+)
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("canyonos_core")
-DEFAULT_DOCKER_PLATFORM = "linux/amd64"
 ARTIFACT_DIR_NAME = ".car"
 SOURCE_DIR_NAME = "app"
-EC2_REQUIRED_CONFIG_KEYS = (
-    "ami_id",
-    "subnet_id",
-    "security_group_ids",
-    "region",
-)
 
 
 # ------------------------------------------------------------------ #
@@ -47,11 +47,14 @@ def _get_package_dir():
 
 
 def _load_config(config_path):
-    """Load a YAML config file."""
-    import yaml
+    """Load the config as the schema checked it and the controller will read it.
 
-    with open(config_path, "r") as f:
-        config = yaml.safe_load(f)
+    The root `.env` is imported and `${VAR}` refs are expanded through the
+    same helper both of those use. Reading the raw YAML here instead let a
+    reference the schema had validated in its expanded form reach the build
+    as the literal `${VAR}`.
+    """
+    config = load_config(config_path)
     # 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}")
@@ -135,25 +138,82 @@ def _artifact_prefix(root):
     )
 
 
+def _project_layout():
+    """(artifact_root, source_root, declarations_dir) for the project in cwd.
+
+    The .car layout keeps the app's own code under `.car/app` and everything
+    generated beside it; a plain checkout keeps both at the project root.
+    """
+    project_dir = os.path.abspath(os.getcwd())
+    prefix = _artifact_prefix(project_dir)
+    artifact_root = os.path.join(project_dir, prefix) if prefix else project_dir
+    return (
+        artifact_root,
+        os.path.join(artifact_root, SOURCE_DIR_NAME) if prefix else project_dir,
+        os.path.join(artifact_root, "config" if prefix else "agents"),
+    )
+
+
+def _reject(violations, summary):
+    """Log each violation on its own line, then exit.
+
+    The host CLI reads the first `ERROR:` line out of this process's output as
+    the root cause, so a violation is never split across lines.
+    """
+    for violation in violations:
+        logger.error("%s", render_violation(violation))
+    logger.error(summary, len(violations))
+    sys.exit(1)
+
+
+def validate_or_exit(config_path, declarations_dir, source_dir=None):
+    """Reject the config before anything is generated; return the parsed manifest."""
+    manifest, violations = check_project(config_path, declarations_dir, source_dir)
+    if violations:
+        _reject(
+            violations,
+            "Configuration rejected: %d problem(s) found; nothing was built.",
+        )
+    return manifest
+
+
 def _normalize_requirements(agent_cfg):
-    """Return an agent's `requirements` list, or [] if absent/null/malformed."""
-    requirements = agent_cfg.get("requirements") or []
-    if not isinstance(requirements, list) or not all(
-        isinstance(r, str) for r in requirements
-    ):
-        # The requirements list is bad, assuming file has no requirements and logging error
-        logger.warning(
-            "Agent '%s': `requirements` must be a list of strings, got %r; ignoring.",
-            agent_cfg.get("name"),
-            requirements,
+    """Return a service's `requirements` list; the schema already checked its shape."""
+    return list(agent_cfg.get("requirements") or [])
+
+
+def _check_dependency_pins(manifest):
+    """Fail the build when an app pin cannot share a version with a platform pin.
+
+    Every service is checked before the first one is built, so a project with
+    two bad pins is told about both instead of one per run.
+    """
+    from canyonos_core.stub_generator import _platform_overrides
+
+    violations = []
+    for index, service in enumerate(manifest.agents):
+        try:
+            _platform_overrides(
+                getattr(service, "requirements", ()),
+                service=index,
+                manifest_path=manifest.path,
+                lines=getattr(service, "requirement_lines", ()),
+            )
+        except DependencyPinConflict as conflict:
+            violations.extend(conflict.violations)
+
+    if violations:
+        _reject(
+            violations,
+            "Dependency pins rejected: %d conflict(s) found; nothing was built.",
         )
-        return []
-    return requirements
 
 
 def _docker_platform():
     """Return the target Docker platform for portable runtime images."""
-    return os.environ.get("CANYONOS_DOCKER_PLATFORM", DEFAULT_DOCKER_PLATFORM)
+    from canyonos_core.stub_generator import target_docker_platform
+
+    return target_docker_platform()
 
 
 def _docker_build_cmd(*args):
@@ -229,13 +289,8 @@ def _ensure_grpc_stubs_importable(project_dir):
 
 
 def _preflight_ec2_deploy(config, project_dir):
-    ec2_cfg = config.get("ec2", {})
-    missing = [key for key in EC2_REQUIRED_CONFIG_KEYS if not ec2_cfg.get(key)]
-    if missing:
-        raise RuntimeError(
-            f"EC2 deploy preflight failed: missing ec2 config keys: {', '.join(sorted(missing))}"
-        )
-
+    # The required `ec2:` keys are the manifest schema's job, checked before
+    # anything was built; what is left here is the local toolchain.
     _require_docker_for_ec2("deploy")
     _ensure_grpc_stubs_importable(project_dir)
 
@@ -308,37 +363,23 @@ def _run_build(config_path):
         logger.error("Config file not found: %s", config_path)
         sys.exit(1)
 
+    artifact_root, source_root, declarations_dir = _project_layout()
+
+    # Nothing below this line runs against a config the schema rejects: no
+    # stubs, no protoc, no Docker context, no image. It comes before the load
+    # so a file that is not YAML at all is rendered as a violation too, and it
+    # is handed source_root so a service whose code is missing fails here
+    # rather than being skipped out of a deploy that then reports success.
+    manifest = validate_or_exit(config_path, declarations_dir, source_root)
+    _check_dependency_pins(manifest)
+
     config = _load_config(config_path)
     agents = config.get("agents", [])
-    project_dir = os.path.abspath(os.getcwd())
-    prefix = _artifact_prefix(project_dir)
-    artifact_root = os.path.join(project_dir, prefix) if prefix else project_dir
-    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    #
     # -------------------------------------------------------------- #
-    declarations_dir = os.path.join(artifact_root, "config" if prefix else "agents")
     stubs_dir = os.path.join(artifact_root, "stubs")
     os.makedirs(stubs_dir, exist_ok=True)
 
@@ -352,15 +393,8 @@ def _run_build(config_path):
     if not yaml_files:
         logger.warning("No agent YAML files found in %s", declarations_dir)
 
-    import yaml
-
     # Looks up a config entry's YAML and to map stubs to entrypoints.
-    yaml_by_name = {}
-    for yaml_path in yaml_files:
-        with open(yaml_path) as f:
-            name = yaml.safe_load(f).get("agent", {}).get("name")
-        if name:
-            yaml_by_name[name] = yaml_path
+    yaml_by_name = declarations_by_name(declarations_dir)
 
     # Maps each generated stub's basename to its agent's entrypoint path, which
     # is the single location the stub is written to and copied to.
@@ -430,10 +464,7 @@ def _run_build(config_path):
             # No build: pull the declared image and tag it like any other
             # agent image so the rest of the deploy pipeline treats it the
             # same way (EC2 image transfer, etc.) without further changes.
-            image = agent_cfg.get("image")
-            if not image:
-                logger.warning("Skipping database '%s': no image specified", agent_name)
-                continue
+            image = agent_cfg["image"]
             target_image = f"canyonos-{agent_name.lower()}"
             logger.info("Pulling database image '%s' as '%s'", image, target_image)
             subprocess.run(
@@ -442,20 +473,12 @@ def _run_build(config_path):
             subprocess.run(["docker", "tag", image, target_image], check=True)
             continue
 
+        # Every key read below is one the schema requires and has checked,
+        # down to the file being on disk -- a service that cannot be built
+        # fails the deploy rather than dropping quietly out of it.
         if agent_type == "workflow":
             # Workflow container
-            workflow_file = agent_cfg.get("workflow_file")
-            if not workflow_file:
-                logger.warning(
-                    "Skipping workflow '%s': no workflow_file specified", agent_name
-                )
-                continue
-
-            workflow_path = os.path.join(source_root, workflow_file)
-            if not os.path.isfile(workflow_path):
-                logger.error("Workflow file not found: %s", workflow_path)
-                continue
-
+            workflow_path = os.path.join(source_root, agent_cfg["workflow_file"])
             docker_context = os.path.join(artifact_root, "docker_container", "Workflow")
             logger.info("Generating workflow Docker context for '%s'", agent_name)
             generate_workflow_docker(
@@ -473,17 +496,7 @@ def _run_build(config_path):
 
         else:
             # Agent container
-            entrypoint = agent_cfg.get("entrypoint")
-            if not entrypoint:
-                logger.warning(
-                    "Skipping agent '%s': no entrypoint specified", agent_name
-                )
-                continue
-
-            agent_file = os.path.join(source_root, entrypoint)
-            if not os.path.isfile(agent_file):
-                logger.error("Agent file not found: %s", agent_file)
-                continue
+            agent_file = os.path.join(source_root, agent_cfg["entrypoint"])
 
             # Find matching YAML by agent name
             matching_yaml = yaml_by_name.get(agent_name)