Feat: added type of agent “database” (CAN-301)

.claude/skills/porting-to-canyonos/references/adapter.md

@@ -117,3 +117,12 @@ memory keyed by session -- carries that id *inside* `query`: accept either a
 bare string or a JSON object in that one field and pass the id through to the
 source unchanged. Do not add a second workflow parameter for it; the platform
 never sends one.
+
+That id identifies a conversation; it does not by itself make the framework's
+own store/checkpointer durable across replicas. If `source-survey.md` found
+one and a `type: database` entry was declared for it, construct that
+store/checkpointer against the declared entry's resolved address instead of an
+in-process one -- read `routing_table:endpoints` from Redis
+(`CANYONOS_REDIS_HOST`/`CANYONOS_REDIS_PORT`, already in every container's env)
+for the database entry's name, the same way any other declared agent's address
+is resolved.

.claude/skills/porting-to-canyonos/references/manifest.md

@@ -57,7 +57,9 @@ the config rather than asking about it:
 - **`replicas` stops being a choice once a service holds cross-request state.**
   When the survey finds such state, the service-boundary section in
   `source-survey.md` fixes `replicas: 1` as a correctness requirement. Report it as a constraint; do not offer to
-  raise it.
+  raise it. This constraint disappears if the state is moved into a declared
+  `type: database` entry instead (see Database services below) -- state then
+  lives outside the process, so replicas no longer share or lose anything.
 - **EC2 identifiers are wrong to invent.** ec2.md forbids copying them from an
   example environment, and a wrong AMI, subnet, or security group fails at
   deploy preflight or, worse, provisions something unreachable. Unanswered
@@ -123,6 +125,27 @@ Use one yaml per deployed service. Argument types are bare builtins only:
 required by the generated stub. `returns.type` is documentation; use `dict` or
 `list` to signal that workflow callers must `json.loads` the returned string.
 
+## Database services
+
+A `type: database` entry provisions a plain container from a public image
+instead of building one from `.car/app` -- it has no `entrypoint` and no
+declaration YAML. Use it when a source's state (a store, memory, or
+checkpointer) must survive across replicas instead of forcing `replicas: 1`
+(see `source-survey.md`). This is unrelated to the top-level `database:` key
+above, which stays omitted regardless.
+
+| Key | Meaning | Default |
+|---|---|---|
+| `image` | public image to pull and tag, e.g. `postgres:16-alpine` | required |
+| `db_port` | host port mapped to the image's own port | `5432` |
+| `volume_path` | in-container path to persist across restarts; omit to skip the volume | absent |
+| `env` | env vars the image itself needs, e.g. `POSTGRES_USER` | `{}` |
+
+Every other declared service already reaches it without extra wiring: its
+resolved `host:port` is published in the same routing table every agent uses,
+over the Redis connection every container already receives
+(`CANYONOS_REDIS_HOST`/`CANYONOS_REDIS_PORT`).
+
 ## Per-image requirements
 
 Each image installs the runtime's base list plus that entry's `requirements:`
@@ -218,7 +241,7 @@ agents:
       - langchain-openai
 
   - name: Workflow
-    type: workflow              # the one entry that carries this key
+    type: workflow               # marks the workflow entry
     workflow_file: email_workflow.py   # relative to .car/app
     api_port: 8080              # where /main is served
     provider: local
@@ -227,6 +250,16 @@ agents:
     requirements:               # its own list; the agent's does not apply here
       - langgraph
 
+  - name: StateDB                # optional; see Database services above
+    type: database
+    image: postgres:16-alpine
+    db_port: 5432
+    volume_path: /var/lib/postgresql/data
+    env:
+      POSTGRES_USER: canyonos
+      POSTGRES_PASSWORD: canyonos
+      POSTGRES_DB: canyonos_state_db
+
 poll_interval: 5                # seconds between metrics polls; default 5
 
 redis:

.claude/skills/porting-to-canyonos/references/source-survey.md

@@ -94,3 +94,10 @@ memory, or checkpointer created in `__init__`—requires `replicas: 1` for
 correctness. The controller can route each call to a different replica, and
 those replicas cannot see one another's in-memory state. Record this as a
 constraint in the configuration review and handoff, not as a sizing preference.
+
+If that state is a store or checkpointer that can instead be backed by a real
+database (for example LangGraph's `AsyncPostgresStore`/`AsyncPostgresSaver`),
+prefer declaring a `type: database` entry (see `manifest.md`) and pointing the
+source's store/checkpointer at it instead of constructing an in-memory one.
+The state then lives outside the process, so the `replicas: 1` constraint
+above no longer applies.

.claude/skills/porting-to-canyonos/validation/manifest.py

@@ -102,13 +102,13 @@ def check_manifest_structure(report, config, config_path, source_dir):
                 valid = False
 
         service_type = entry.get("type", "agent")
-        if service_type not in ("agent", "workflow"):
+        if service_type not in ("agent", "workflow", "database"):
             report.error(
                 "V002",
                 config_path,
                 line_of(entry, "type"),
-                f"`type: {service_type}` is neither `agent` nor `workflow`",
-                "Only those two service shapes have a CanyonOS build contract.",
+                f"`type: {service_type}` is neither `agent`, `workflow`, nor `database`",
+                "Only those service shapes have a CanyonOS build contract.",
             )
             valid = False
 
@@ -149,26 +149,38 @@ def check_manifest_structure(report, config, config_path, source_dir):
             )
             valid = False
 
-        path_key = "workflow_file" if service_type == "workflow" else "entrypoint"
-        relative = entry.get(path_key)
-        if not _safe_relative_python_path(relative):
-            report.error(
-                "V002",
-                config_path,
-                line_of(entry, path_key),
-                f"`{path_key}` must be a relative .py path contained by `.car/app`",
-                "Absolute and parent-relative paths escape the self-contained CanyonOS artifact.",
-            )
-            valid = False
-        elif not os.path.isfile(os.path.join(source_dir, relative)):
-            report.error(
-                "V002",
-                config_path,
-                line_of(entry, path_key),
-                f"`{path_key}: {relative}` does not exist in `.car/app`",
-                "The deploy build cannot create this service without its Python entry file.",
-            )
-            valid = False
+        if service_type == "database":
+            image = entry.get("image")
+            if not isinstance(image, str) or not image.strip():
+                report.error(
+                    "V002",
+                    config_path,
+                    line_of(entry, "image"),
+                    f"agents[{index}] has type `database` but no non-empty `image`",
+                    "A database service is pulled from a public image; it is never built from `.car/app`.",
+                )
+                valid = False
+        else:
+            path_key = "workflow_file" if service_type == "workflow" else "entrypoint"
+            relative = entry.get(path_key)
+            if not _safe_relative_python_path(relative):
+                report.error(
+                    "V002",
+                    config_path,
+                    line_of(entry, path_key),
+                    f"`{path_key}` must be a relative .py path contained by `.car/app`",
+                    "Absolute and parent-relative paths escape the self-contained CanyonOS artifact.",
+                )
+                valid = False
+            elif not os.path.isfile(os.path.join(source_dir, relative)):
+                report.error(
+                    "V002",
+                    config_path,
+                    line_of(entry, path_key),
+                    f"`{path_key}: {relative}` does not exist in `.car/app`",
+                    "The deploy build cannot create this service without its Python entry file.",
+                )
+                valid = False
 
     return entries if valid else None
 
@@ -230,7 +242,9 @@ def discover_agent_declarations(report, config_dir, config_path):
 def check_declaration_bindings(report, entries, declarations, config_path):
     """Require a one-to-one binding for every agent service."""
     configured = {
-        entry["name"] for entry in entries if entry.get("type", "agent") != "workflow"
+        entry["name"]
+        for entry in entries
+        if entry.get("type", "agent") not in ("workflow", "database")
     }
     for name in sorted(configured - declarations.keys()):
         report.error(

examples/portfolio/config/global_controller.yaml

@@ -79,6 +79,18 @@ agents:
     provider: local
     instance_type: t3.micro
 
+  # Example: a shared Postgres state store (commented out, not used by this pipeline).
+  # - name: StateDB
+  #   type: database
+  #   image: postgres:16-alpine
+  #   db_port: 5432
+  #   volume_path: /var/lib/postgresql/data
+  #   env:
+  #     POSTGRES_USER: username
+  #     POSTGRES_PASSWORD: password
+  #     POSTGRES_DB: canyonos_state_db
+  #   provider: local
+
 
 otel:
   # The dashboard api's own OTLP ingest. Must be the full url including the

packages/core/canyonos_core/cli.py

@@ -278,7 +278,7 @@ def _run_build(config_path):
     missing_stubs = [
         a["name"]
         for a in agents
-        if a.get("type", "agent") != "workflow"
+        if a.get("type", "agent") not in ("workflow", "database")
         and (a["name"] not in yaml_by_name or not a.get("entrypoint"))
     ]
     if missing_stubs:
@@ -336,6 +336,20 @@ def _run_build(config_path):
         agent_name = agent_cfg["name"]
         agent_type = agent_cfg.get("type", "agent")
 
+        if agent_type == "database":
+            # 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
+            target_image = f"canyonos-{agent_name.lower()}"
+            logger.info("Pulling database image '%s' as '%s'", image, target_image)
+            subprocess.run(["docker", "pull", image], check=True)
+            subprocess.run(["docker", "tag", image, target_image], check=True)
+            continue
+
         if agent_type == "workflow":
             # Workflow container
             workflow_file = agent_cfg.get("workflow_file")

packages/core/canyonos_core/controller/cloud_provider_logic/EC2/_runtime.py

@@ -195,6 +195,10 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
             instance["public_host"] = provisioned["public_host"]
         if spec.get("type") == "workflow":
             instance["api_port"] = str(spec.get("api_port", 8080))
+        elif spec.get("type") == "database":
+            instance["container_port"] = "5432"
+            instance["host_port"] = "5432"
+            instance["endpoint"] = f"{host}:5432"
         return instance
     except Exception:
         terminate_instance(provisioned)
@@ -251,6 +255,14 @@ def _bootstrap_instance(
     port_args = ["-p", f"{CONTAINER_PORT}:{CONTAINER_PORT}"]
     if spec.get("type") == "workflow":
         port_args += ["-p", f"{spec.get('api_port', 8080)}:8080"]
+    elif spec.get("type") == "database":
+        port_args += ["-p", f"{spec.get('db_port', 5432)}:5432"]
+
+    volume_args = []
+    if spec.get("type") == "database":
+        volume_path = spec.get("volume_path")
+        if volume_path:
+            volume_args = ["-v", f"canyonos-{agent_name.lower()}-data:{volume_path}"]
 
     logger.info("Transferring image %s to %s", image, host)
     result = subprocess.run(
@@ -286,6 +298,7 @@ def _bootstrap_instance(
         "--name",
         container,
         *port_args,
+        *volume_args,
         "-e",
         f"CANYONOS_REDIS_HOST={redis_host}",
         "-e",
@@ -311,6 +324,9 @@ def _bootstrap_instance(
             cmd.extend(["-e", f"CANYONOS_DATABASE_URL={db_url}"])
         if project_id:
             cmd.extend(["-e", f"CANYONOS_PROJECT_ID={project_id}"])
+    elif spec.get("type") == "database":
+        for key, value in spec.get("env", {}).items():
+            cmd.extend(["-e", f"{key}={value}"])
 
     # User secrets from `env_file`. Explicit -e flags above still win over
     # anything in the file.

packages/core/canyonos_core/controller/cloud_provider_logic/Local/_runtime.py

@@ -107,6 +107,10 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
             NETWORK,
             "--name",
             runtime_id,
+            # Docker Desktop resolves this automatically; native Linux Docker
+            # (e.g. an EC2 test box) does not unless told to.
+            "--add-host",
+            "host.docker.internal:host-gateway",
             "-p",
             f"{host_port}:{CONTAINER_PORT}",
             "-e",
@@ -146,6 +150,13 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
                 cmd.extend(["-e", f"CANYONOS_DATABASE_URL={db_url}"])
             if project_id:
                 cmd.extend(["-e", f"CANYONOS_PROJECT_ID={project_id}"])
+        elif ctrl_type == "database":
+            cmd.extend(["-p", f"{spec.get('db_port', 5432)}:5432"])
+            volume_path = spec.get("volume_path")
+            if volume_path:
+                cmd.extend(["-v", f"canyonos-{agent_name.lower()}-data:{volume_path}"])
+            for key, value in spec.get("env", {}).items():
+                cmd.extend(["-e", f"{key}={value}"])
         if resources.get("cpu"):
             cmd.extend(["--cpus", str(resources["cpu"])])
         if resources.get("memory"):
@@ -200,6 +211,8 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         instance["user"] = user
     if ctrl_type == "workflow":
         instance["api_port"] = str(spec.get("api_port", 8080))
+    elif ctrl_type == "database":
+        instance["container_port"] = "5432"
     logger.info("Runtime ready: %s -> %s", runtime_id, instance["endpoint"])
     return instance
 
@@ -219,4 +232,4 @@ def terminate_instance(instance):
 
 
 def routing_endpoint_for(instance):
-    return f"{instance['runtime_id']}:{CONTAINER_PORT}"
+    return f"{instance['runtime_id']}:{instance.get('container_port', CONTAINER_PORT)}"

packages/core/canyonos_core/llm_proxy/core.py

@@ -49,7 +49,7 @@ def proxy_request(provider, subpath, flask_request):
     # non-empty stub text -- which only `canyonos test` sets -- enables stubbing.
     _stub = stub_text()
     if _stub:
-        pr = build_stub(provider.name, subpath, _stub)
+        pr = build_stub(provider.name, subpath, _stub, body=body)
     else:
         pr = provider.forward(flask_request, subpath, body)
 

packages/core/canyonos_core/llm_proxy/stub.py

@@ -13,6 +13,7 @@
     CANYONOS_LLM_STUB_TEXT=testing
 """
 
+# This stubbing is used to test workflows without connecting to the actual LLM and incurring costs. Notice though, if you use this, credentials won't be verified as this stub path doesn't use any.
 from __future__ import annotations
 
 import json
@@ -78,8 +79,41 @@ def _bedrock_invoke_stream_response(text):
     return _stream_response(events)
 
 
-def build_stub(provider_name, subpath, text):
+_STUB_EMBEDDING_DIMS = 1536
+
+
+def _openai_embedding_stub(body):
+    """A minimal embeddings response, one canned vector per requested input."""
+    try:
+        count = len(json.loads(body or b"{}").get("input") or [None])
+    except (ValueError, TypeError):
+        count = 1
+    return _json_response(
+        {
+            "object": "list",
+            "data": [
+                {
+                    "object": "embedding",
+                    "index": i,
+                    "embedding": [0.0] * _STUB_EMBEDDING_DIMS,
+                }
+                for i in range(count)
+            ],
+            "model": "stub",
+            "usage": {"prompt_tokens": 1, "total_tokens": 1},
+        }
+    )
+
+
+def build_stub(provider_name, subpath, text, body=None):
     """Build a provider-appropriate canned response carrying ``text``."""
+    if (
+        provider_name == "openai"
+        and subpath
+        and subpath.rstrip("/").endswith("embeddings")
+    ):
+        return _openai_embedding_stub(body)
+
     if provider_name == "bedrock":
         op = subpath.rsplit("/", 1)[-1] if subpath else ""
         if op == "converse-stream":