Support image-overrides in spec for testing

Spec can override container images:
  image-overrides:
    dumpster-kubo: ghcr.io/.../dumpster-kubo:test-tag

Merged with CLI overrides (CLI wins). Enables testing with
GHCR-pushed test tags without modifying compose files.

Also reverts the image-pull-policy spec key (not needed —
the fix is to use proper GHCR tags, not IfNotPresent).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

.gitignore

@@ -8,3 +8,4 @@ __pycache__
 package
 stack_orchestrator/data/build_tag.txt
 /build
+.worktrees

.pebbles/.gitignore

@@ -0,0 +1 @@
+pebbles.db

.pebbles/config.json

@@ -0,0 +1 @@
+{"project": "stack-orchestrator", "prefix": "so"}

.pebbles/events.jsonl

@@ -0,0 +1,10 @@
+{"type": "create", "timestamp": "2026-03-18T14:45:07.038870Z", "issue_id": "so-a1a", "payload": {"title": "deploy create should support external credential injection", "type": "feature", "priority": "2", "description": "deploy create generates config.env but provides no mechanism to inject external credentials (API keys, tokens, etc.) at creation time. Operators must append to config.env after the fact, which mutates a build artifact. deploy create should accept --credentials-file or similar to include secrets in the generated config.env."}}
+{"type": "create", "timestamp": "2026-03-18T14:45:07.038942Z", "issue_id": "so-b2b", "payload": {"title": "REGISTRY_TOKEN / imagePullSecret flow undocumented", "type": "bug", "priority": "2", "description": "create_registry_secret() exists in deployment_create.py and is called during up(), but REGISTRY_TOKEN is not documented in spec.yml or any user-facing docs. The restart command warns \"Registry token env var REGISTRY_TOKEN not set, skipping registry secret\" but doesn't explain how to set it. For GHCR private images, this is required and the flow from spec.yml -> config.env -> imagePullSecret needs documentation."}}
+{"type": "create", "timestamp": "2026-03-18T19:10:00.000000Z", "issue_id": "so-k1k", "payload": {"title": "Stack path resolution differs between deploy create and deployment restart", "type": "bug", "priority": "2", "description": "deploy create resolves --stack as a relative path from cwd. deployment restart resolves --stack-path as absolute, then computes repo_root as 4 parents up (assuming stack_orchestrator/data/stacks/name structure). External stacks with different nesting depths (e.g. stack-orchestrator/stacks/name = 3 levels) get wrong repo_root, causing --spec-file resolution to fail. The two commands should use the same path resolution logic."}}
+{"type": "create", "timestamp": "2026-03-18T19:25:00.000000Z", "issue_id": "so-l2l", "payload": {"title": "deployment restart should update in place, not delete/recreate", "type": "bug", "priority": "1", "description": "deployment restart deletes the entire namespace then recreates everything from scratch. This causes:\n\n1. **Downtime** — nothing serves traffic between delete and successful recreate\n2. **No rollback** — deleting the namespace destroys ReplicaSet revision history\n3. **Race conditions** — namespace may still be terminating when up() tries to create\n4. **Cascading failures** — if ANY container fails to start, the entire site is down with no fallback\n\nFix: three changes needed.\n\n**A. up() should create-or-update, not just create.** Use patch/apply semantics for Deployments, Services, Ingresses. When the pod spec changes (new env vars, new image), k8s creates a new ReplicaSet, scales it up, waits for readiness probes, then scales the old one down. Old pods serve traffic until new pods are healthy.\n\n**B. down() should never delete the namespace on restart.** Only on explicit teardown. The namespace owns the revision history. Current code: _delete_namespace() on every down(). Should: delete individual resources by label for teardown, do nothing for restart (let update-in-place handle it).\n\n**C. All containers need readiness probes.** Without them k8s considers pods ready immediately, defeating rolling update safety. laconic-so should generate readiness probes from the http-proxy routes in spec.yml (if a container has an http route, probe that port).\n\nWith these changes, k8s native rolling updates provide zero-downtime deploys and automatic rollback (if new pods fail readiness, rollout stalls, old pods keep serving).\n\nSource files:\n- deploy_k8s.py: up(), down(), _create_deployment(), _delete_namespace()\n- cluster_info.py: pod spec generation (needs readiness probes)\n- deployment.py: restart() orchestration"}}
+{"type": "create", "timestamp": "2026-03-18T20:15:03.000000Z", "issue_id": "so-m3m", "payload": {"title": "Add credentials-files spec key for on-disk credential injection", "type": "feature", "priority": "1", "description": "deployment restart regenerates config.env from spec.yml, wiping credentials that were appended from on-disk files (e.g. ~/.credentials/*.env). Operators must append credentials after deploy create, which is fragile and breaks on restart.\n\nFix: New top-level spec key credentials-files. _write_config_file() reads each file and appends its contents to config.env after writing config vars. Files are read at deploy time from the deployment host.\n\nSpec syntax:\n  credentials-files:\n    - ~/.credentials/dumpster-secrets.env\n    - ~/.credentials/dumpster-r2.env\n\nFiles:\n- deploy/spec.py: add get_credentials_files() returning list of paths\n- deploy/deployment_create.py: in _write_config_file(), after writing config vars, read and append each credentials file (expand ~ to home dir)\n\nAlso update dumpster-stack spec.yml to use the new key and remove the ansible credential append workaround from woodburn_deployer (group_vars/all.yml credentials_env_files, stack_deploy role append tasks, restart_dumpster.yml credential steps). Those cleanups are in the woodburn_deployer repo."}}
+{"type":"status_update","timestamp":"2026-03-18T21:54:12.59148256Z","issue_id":"so-m3m","payload":{"status":"in_progress"}}
+{"type":"close","timestamp":"2026-03-18T21:55:31.6035544Z","issue_id":"so-m3m","payload":{}}
+{"type": "create", "timestamp": "2026-03-20T23:05:00.000000Z", "issue_id": "so-n1n", "payload": {"title": "Merge kind-mount-propagation branch — HostToContainer propagation for extraMounts", "type": "feature", "priority": "2", "description": "The kind-mount-root feature was cherry-picked to main (commit 8d03083d) but the mount propagation fix (commit 929bdab8 on branch enya-ac868cc4-kind-mount-propagation-fix) adds HostToContainer propagation so host submounts propagate into the Kind node. This is needed for ZFS child datasets and tmpfs mounts under the root. Cherry-pick 929bdab8 to main."}}
+{"type": "create", "timestamp": "2026-03-20T23:05:00.000000Z", "issue_id": "so-o2o", "payload": {"title": "etcd cert backup not persisting across cluster deletion", "type": "bug", "priority": "1", "description": "The extraMount for etcd at data/cluster-backups/<id>/etcd is configured but after cluster deletion the directory is empty. Caddy TLS certificates stored in etcd are lost. Either etcd isn't writing to the host mount, or the cleanup code is deleting the backup. Investigate _clean_etcd_keeping_certs in helpers.py."}}
+{"type": "create", "timestamp": "2026-03-21T00:20:00.000000Z", "issue_id": "so-p3p", "payload": {"title": "laconic-so should manage Caddy ingress image lifecycle", "type": "feature", "priority": "2", "description": "The Caddy ingress controller image is hardcoded in ingress-caddy-kind-deploy.yaml. There's no mechanism to update it without manual kubectl commands or cluster recreation. laconic-so should: 1) Allow spec.yml to specify a custom Caddy image, 2) Support updating the Caddy image as part of deployment restart, 3) Set strategy: Recreate on the Caddy Deployment (hostPort pods can't do RollingUpdate). This would let cryovial or similar tooling trigger Caddy updates through the normal deployment pipeline."}}

docs/cli.md

@@ -68,7 +68,7 @@ $ laconic-so build-npms --include <package-name> --force-rebuild
 
 ## deploy
 
-The `deploy` command group manages persistent deployments. The general workflow is `deploy init` to generate a spec file, then `deploy create` to create a deployment directory from the spec, then runtime commands like `deploy up` and `deploy down`.
+The `deploy` command group manages persistent deployments. The general workflow is `deploy init` to generate a spec file, then `deploy create` to create a deployment directory from the spec, then runtime commands like `deployment start` and `deployment stop`.
 
 ### deploy init
 
@@ -101,35 +101,91 @@ Options:
 - `--spec-file` (required): spec file to use
 - `--deployment-dir`: target directory for deployment files
 - `--update`: update an existing deployment directory, preserving data volumes and env file. Changed files are backed up with a `.bak` suffix. The deployment's `config.env` and `deployment.yml` are also preserved.
+- `--helm-chart`: generate Helm chart instead of deploying (k8s only)
 - `--network-dir`: network configuration supplied in this directory
 - `--initial-peers`: initial set of persistent peers
 
-### deploy up
+## deployment
 
-Start a deployment:
+Runtime commands for managing a created deployment. Use `--dir` to specify the deployment directory.
+
+### deployment start
+
+Start a deployment (`up` is a legacy alias):
 ```
-$ laconic-so deployment --dir <deployment-dir> up
+$ laconic-so deployment --dir <deployment-dir> start
 ```
 
-### deploy down
+Options:
+- `--stay-attached` / `--detatch-terminal`: attach to container stdout (default: detach)
+- `--skip-cluster-management` / `--perform-cluster-management`: skip kind cluster creation/teardown (default: perform management). Only affects k8s-kind deployments. Use this when multiple stacks share a single cluster.
 
-Stop a deployment:
-```
-$ laconic-so deployment --dir <deployment-dir> down
-```
-Use `--delete-volumes` to also remove data volumes.
+### deployment stop
 
-### deploy ps
+Stop a deployment (`down` is a legacy alias):
+```
+$ laconic-so deployment --dir <deployment-dir> stop
+```
+
+Options:
+- `--delete-volumes` / `--preserve-volumes`: delete data volumes on stop (default: preserve)
+- `--skip-cluster-management` / `--perform-cluster-management`: skip kind cluster teardown (default: perform management). Use this to stop a single deployment without destroying a shared cluster.
+
+### deployment restart
+
+Restart a deployment with GitOps-aware workflow. Pulls latest stack code, syncs the deployment directory from the git-tracked spec, and restarts services:
+```
+$ laconic-so deployment --dir <deployment-dir> restart
+```
+
+See [deployment_patterns.md](deployment_patterns.md) for the recommended GitOps workflow.
+
+### deployment ps
 
 Show running services:
 ```
 $ laconic-so deployment --dir <deployment-dir> ps
 ```
 
-### deploy logs
+### deployment logs
 
 View service logs:
 ```
 $ laconic-so deployment --dir <deployment-dir> logs
 ```
 Use `-f` to follow and `-n <count>` to tail.
+
+### deployment exec
+
+Execute a command in a running service container:
+```
+$ laconic-so deployment --dir <deployment-dir> exec <service-name> "<command>"
+```
+
+### deployment status
+
+Show deployment status:
+```
+$ laconic-so deployment --dir <deployment-dir> status
+```
+
+### deployment port
+
+Show mapped ports for a service:
+```
+$ laconic-so deployment --dir <deployment-dir> port <service-name> <port>
+```
+
+### deployment push-images
+
+Push deployment images to a registry:
+```
+$ laconic-so deployment --dir <deployment-dir> push-images
+```
+
+### deployment run-job
+
+Run a one-time job in the deployment:
+```
+$ laconic-so deployment --dir <deployment-dir> run-job <job-name>
+```

docs/deployment_patterns.md

@@ -30,7 +30,7 @@ git commit -m "Add my-stack deployment configuration"
 git push
 
 # On deployment server: deploy from git-tracked spec
-laconic-so deploy create \
+laconic-so --stack my-stack deploy create \
   --spec-file /path/to/operator-repo/spec.yml \
   --deployment-dir my-deployment
 

stack_orchestrator/constants.py

@@ -29,6 +29,7 @@ network_key = "network"
 http_proxy_key = "http-proxy"
 image_registry_key = "image-registry"
 configmaps_key = "configmaps"
+secrets_key = "secrets"
 resources_key = "resources"
 volumes_key = "volumes"
 security_key = "security"
@@ -45,3 +46,6 @@ runtime_class_key = "runtime-class"
 high_memlock_runtime = "high-memlock"
 high_memlock_spec_filename = "high-memlock-spec.json"
 acme_email_key = "acme-email"
+kind_mount_root_key = "kind-mount-root"
+external_services_key = "external-services"
+ca_certificates_key = "ca-certificates"

stack_orchestrator/data/compose-jobs/docker-compose-test-job.yml

@@ -0,0 +1,5 @@
+services:
+  test-job:
+    image: cerc/test-container:local
+    entrypoint: /bin/sh
+    command: ["-c", "echo 'Job completed successfully'"]

stack_orchestrator/data/k8s/components/ingress/ingress-caddy-kind-deploy.yaml

@@ -186,8 +186,8 @@ spec:
           operator: Equal
       containers:
         - name: caddy-ingress-controller
-          image: caddy/ingress:latest
-          imagePullPolicy: IfNotPresent
+          image: ghcr.io/laconicnetwork/caddy-ingress:latest
+          imagePullPolicy: Always
           ports:
             - name: http
               containerPort: 80

stack_orchestrator/data/stacks/test/deploy/commands.py

@@ -21,7 +21,7 @@ from stack_orchestrator.deploy.deploy_util import VolumeMapping, run_container_c
 from pathlib import Path
 
 default_spec_file_content = """config:
-    test-variable-1: test-value-1
+    test_variable_1: test-value-1
 """
 
 

stack_orchestrator/data/stacks/test/stack.yml

@@ -7,3 +7,5 @@ containers:
   - cerc/test-container
 pods:
   - test
+jobs:
+  - test-job

stack_orchestrator/deploy/compose/deploy_docker.py

@@ -48,7 +48,7 @@ class DockerDeployer(Deployer):
         self.compose_project_name = compose_project_name
         self.compose_env_file = compose_env_file
 
-    def up(self, detach, skip_cluster_management, services):
+    def up(self, detach, skip_cluster_management, services, image_overrides=None):
         if not opts.o.dry_run:
             try:
                 return self.docker.compose.up(detach=detach, services=services)

stack_orchestrator/deploy/deploy.py

@@ -35,6 +35,7 @@ from stack_orchestrator.util import (
     get_dev_root_path,
     stack_is_in_deployment,
     resolve_compose_file,
+    get_job_list,
 )
 from stack_orchestrator.deploy.deployer import DeployerException
 from stack_orchestrator.deploy.deployer_factory import getDeployer
@@ -130,12 +131,17 @@ def create_deploy_context(
         compose_files=cluster_context.compose_files,
         compose_project_name=cluster_context.cluster,
         compose_env_file=cluster_context.env_file,
+        job_compose_files=cluster_context.job_compose_files,
     )
     return DeployCommandContext(stack, cluster_context, deployer)
 
 
 def up_operation(
-    ctx, services_list, stay_attached=False, skip_cluster_management=False
+    ctx,
+    services_list,
+    stay_attached=False,
+    skip_cluster_management=False,
+    image_overrides=None,
 ):
     global_context = ctx.parent.parent.obj
     deploy_context = ctx.obj
@@ -154,6 +160,7 @@ def up_operation(
         detach=not stay_attached,
         skip_cluster_management=skip_cluster_management,
         services=services_list,
+        image_overrides=image_overrides,
     )
     for post_start_command in cluster_context.post_start_commands:
         _run_command(global_context, cluster_context.cluster, post_start_command)
@@ -403,7 +410,7 @@ def _make_cluster_context(ctx, stack, include, exclude, cluster, env_file):
         stack_config = get_parsed_stack_config(stack)
         if stack_config is not None:
             # TODO: syntax check the input here
-            pods_in_scope = stack_config["pods"]
+            pods_in_scope = stack_config.get("pods") or []
             cluster_config = (
                 stack_config["config"] if "config" in stack_config else None
             )
@@ -477,6 +484,22 @@ def _make_cluster_context(ctx, stack, include, exclude, cluster, env_file):
     if ctx.verbose:
         print(f"files: {compose_files}")
 
+    # Gather job compose files (from compose-jobs/ directory in deployment)
+    job_compose_files = []
+    if deployment and stack:
+        stack_config = get_parsed_stack_config(stack)
+        if stack_config:
+            jobs = get_job_list(stack_config)
+            compose_jobs_dir = stack.joinpath("compose-jobs")
+            for job in jobs:
+                job_file_name = os.path.join(
+                    compose_jobs_dir, f"docker-compose-{job}.yml"
+                )
+                if os.path.exists(job_file_name):
+                    job_compose_files.append(job_file_name)
+            if ctx.verbose:
+                print(f"job files: {job_compose_files}")
+
     return ClusterContext(
         ctx,
         cluster,
@@ -485,6 +508,7 @@ def _make_cluster_context(ctx, stack, include, exclude, cluster, env_file):
         post_start_commands,
         cluster_config,
         env_file,
+        job_compose_files=job_compose_files if job_compose_files else None,
     )
 
 

stack_orchestrator/deploy/deploy_types.py

@@ -29,6 +29,7 @@ class ClusterContext:
     post_start_commands: List[str]
     config: Optional[str]
     env_file: Optional[str]
+    job_compose_files: Optional[List[str]] = None
 
 
 @dataclass

stack_orchestrator/deploy/deployer.py

@@ -20,7 +20,7 @@ from typing import Optional
 
 class Deployer(ABC):
     @abstractmethod
-    def up(self, detach, skip_cluster_management, services):
+    def up(self, detach, skip_cluster_management, services, image_overrides=None):
         pass
 
     @abstractmethod

stack_orchestrator/deploy/deployer_factory.py

@@ -34,7 +34,12 @@ def getDeployerConfigGenerator(type: str, deployment_context):
 
 
 def getDeployer(
-    type: str, deployment_context, compose_files, compose_project_name, compose_env_file
+    type: str,
+    deployment_context,
+    compose_files,
+    compose_project_name,
+    compose_env_file,
+    job_compose_files=None,
 ):
     if type == "compose" or type is None:
         return DockerDeployer(
@@ -54,6 +59,7 @@ def getDeployer(
             compose_files,
             compose_project_name,
             compose_env_file,
+            job_compose_files=job_compose_files,
         )
     else:
         print(f"ERROR: deploy-to {type} is not valid")

stack_orchestrator/deploy/deployment.py

@@ -17,7 +17,7 @@ import click
 from pathlib import Path
 import subprocess
 import sys
-import time
+
 from stack_orchestrator import constants
 from stack_orchestrator.deploy.images import push_images_operation
 from stack_orchestrator.deploy.deploy import (
@@ -248,8 +248,13 @@ def run_job(ctx, job_name, helm_release):
     "--expected-ip",
     help="Expected IP for DNS verification (if different from egress)",
 )
+@click.option(
+    "--image",
+    multiple=True,
+    help="Override container image: container=image",
+)
 @click.pass_context
-def restart(ctx, stack_path, spec_file, config_file, force, expected_ip):
+def restart(ctx, stack_path, spec_file, config_file, force, expected_ip, image):
     """Pull latest code and restart deployment using git-tracked spec.
 
     GitOps workflow:
@@ -276,6 +281,17 @@ def restart(ctx, stack_path, spec_file, config_file, force, expected_ip):
 
     deployment_context: DeploymentContext = ctx.obj
 
+    # Parse --image flags into a dict of container_name -> image
+    image_overrides = {}
+    for entry in image:
+        if "=" not in entry:
+            raise click.BadParameter(
+                f"Invalid --image format '{entry}', expected container=image",
+                param_hint="'--image'",
+            )
+        container_name, image_ref = entry.split("=", 1)
+        image_overrides[container_name] = image_ref
+
     # Get current spec info (before git pull)
     current_spec = deployment_context.spec
     current_http_proxy = current_spec.get_http_proxy()
@@ -322,9 +338,22 @@ def restart(ctx, stack_path, spec_file, config_file, force, expected_ip):
 
     # Determine spec file location
     # Priority: --spec-file argument > repo's deployment/spec.yml > deployment dir
-    # Stack path is like: repo/stack_orchestrator/data/stacks/stack-name
-    # So repo root is 4 parents up
-    repo_root = stack_source.parent.parent.parent.parent
+    # Find repo root via git rather than assuming a fixed directory depth.
+    git_root_result = subprocess.run(
+        ["git", "rev-parse", "--show-toplevel"],
+        cwd=stack_source,
+        capture_output=True,
+        text=True,
+    )
+    if git_root_result.returncode == 0:
+        repo_root = Path(git_root_result.stdout.strip())
+    else:
+        # Fallback: walk up from stack_source looking for .git
+        repo_root = stack_source
+        while repo_root != repo_root.parent:
+            if (repo_root / ".git").exists():
+                break
+            repo_root = repo_root.parent
     if spec_file:
         # Spec file relative to repo root
         spec_file_path = repo_root / spec_file
@@ -368,7 +397,14 @@ def restart(ctx, stack_path, spec_file, config_file, force, expected_ip):
         print("\n[2/4] Hostname unchanged, skipping DNS verification")
 
     # Step 3: Sync deployment directory with spec
+    # The spec's "stack:" value is often a relative path (e.g.
+    # "stack-orchestrator/stacks/dumpster") that must resolve from the
+    # repo root.  Change cwd so stack_is_external() sees it correctly.
     print("\n[3/4] Syncing deployment directory...")
+    import os
+
+    prev_cwd = os.getcwd()
+    os.chdir(repo_root)
     deploy_ctx = make_deploy_context(ctx)
     create_operation(
         deployment_command_context=deploy_ctx,
@@ -378,28 +414,216 @@ def restart(ctx, stack_path, spec_file, config_file, force, expected_ip):
         network_dir=None,
         initial_peers=None,
     )
-
     # Reload deployment context with updated spec
     deployment_context.init(deployment_context.deployment_dir)
     ctx.obj = deployment_context
 
-    # Stop deployment
-    print("\n[4/4] Restarting deployment...")
+    # Apply updated deployment.
+    # If maintenance-service is configured, swap Ingress to maintenance
+    # backend during the Recreate window so users see a branded page
+    # instead of bare 502s.
+    print("\n[4/4] Applying deployment update...")
     ctx.obj = make_deploy_context(ctx)
-    down_operation(
-        ctx, delete_volumes=False, extra_args_list=[], skip_cluster_management=True
-    )
 
-    # Brief pause to ensure clean shutdown
-    time.sleep(5)
+    # Check for maintenance service in the (reloaded) spec
+    maintenance_svc = deployment_context.spec.get_maintenance_service()
+    if maintenance_svc:
+        print(f"Maintenance service configured: {maintenance_svc}")
+        _restart_with_maintenance(
+            ctx, deployment_context, maintenance_svc, image_overrides
+        )
+    else:
+        up_operation(
+            ctx,
+            services_list=None,
+            stay_attached=False,
+            skip_cluster_management=True,
+            image_overrides=image_overrides or None,
+        )
 
-    # Start deployment
-    up_operation(
-        ctx, services_list=None, stay_attached=False, skip_cluster_management=True
-    )
+    # Restore cwd after both create_operation and up_operation have run.
+    # Both need the relative stack path to resolve from repo_root.
+    os.chdir(prev_cwd)
 
     print("\n=== Restart Complete ===")
-    print("Deployment restarted with git-tracked configuration.")
+    print("Deployment updated via rolling update.")
     if new_hostname and new_hostname != current_hostname:
         print(f"\nNew hostname: {new_hostname}")
         print("Caddy will automatically provision TLS certificate.")
+
+
+def _restart_with_maintenance(
+    ctx, deployment_context, maintenance_svc, image_overrides
+):
+    """Restart with Ingress swap to maintenance service during Recreate.
+
+    Flow:
+    1. Deploy all pods (including maintenance pod) with up_operation
+    2. Patch Ingress: swap all route backends to maintenance service
+    3. Scale main (non-maintenance) Deployments to 0
+    4. Scale main Deployments back up (triggers Recreate with new spec)
+    5. Wait for readiness
+    6. Patch Ingress: restore original backends
+
+    This ensures the maintenance pod is already running before we touch
+    the Ingress, and the main pods get a clean Recreate.
+    """
+    import time
+
+    from kubernetes.client.exceptions import ApiException
+
+    from stack_orchestrator.deploy.deploy import up_operation
+
+    # Step 1: Apply the full deployment (creates/updates all pods + services)
+    # This ensures maintenance pod exists before we swap Ingress to it.
+    up_operation(
+        ctx,
+        services_list=None,
+        stay_attached=False,
+        skip_cluster_management=True,
+        image_overrides=image_overrides or None,
+    )
+
+    # Parse maintenance service spec: "container-name:port"
+    maint_container = maintenance_svc.split(":")[0]
+    maint_port = int(maintenance_svc.split(":")[1])
+
+    # Connect to k8s API
+    deploy_ctx = ctx.obj
+    deployer = deploy_ctx.deployer
+    deployer.connect_api()
+    namespace = deployer.k8s_namespace
+    app_name = deployer.cluster_info.app_name
+    networking_api = deployer.networking_api
+    apps_api = deployer.apps_api
+
+    ingress_name = f"{app_name}-ingress"
+
+    # Step 2: Read current Ingress and save original backends
+    try:
+        ingress = networking_api.read_namespaced_ingress(
+            name=ingress_name, namespace=namespace
+        )
+    except ApiException:
+        print("Warning: No Ingress found, skipping maintenance swap")
+        return
+
+    # Resolve which service the maintenance container belongs to
+    maint_service_name = deployer.cluster_info._resolve_service_name_for_container(
+        maint_container
+    )
+
+    # Save original backends for restoration
+    original_backends = []
+    for rule in ingress.spec.rules:
+        rule_backends = []
+        for path in rule.http.paths:
+            rule_backends.append(
+                {
+                    "name": path.backend.service.name,
+                    "port": path.backend.service.port.number,
+                }
+            )
+        original_backends.append(rule_backends)
+
+    # Patch all Ingress backends to point to maintenance service
+    print("Swapping Ingress to maintenance service...")
+    for rule in ingress.spec.rules:
+        for path in rule.http.paths:
+            path.backend.service.name = maint_service_name
+            path.backend.service.port.number = maint_port
+
+    networking_api.replace_namespaced_ingress(
+        name=ingress_name, namespace=namespace, body=ingress
+    )
+    print("Ingress now points to maintenance service")
+
+    # Step 3: Find main (non-maintenance) Deployments and scale to 0
+    # then back up to trigger a clean Recreate
+    deployments_resp = apps_api.list_namespaced_deployment(
+        namespace=namespace, label_selector=f"app={app_name}"
+    )
+    main_deployments = []
+    for dep in deployments_resp.items:
+        dep_name = dep.metadata.name
+        # Skip maintenance deployments
+        component = (dep.metadata.labels or {}).get("app.kubernetes.io/component", "")
+        is_maintenance = maint_container in component
+        if not is_maintenance:
+            main_deployments.append(dep_name)
+
+    if main_deployments:
+        # Scale down main deployments
+        for dep_name in main_deployments:
+            print(f"Scaling down {dep_name}...")
+            apps_api.patch_namespaced_deployment_scale(
+                name=dep_name,
+                namespace=namespace,
+                body={"spec": {"replicas": 0}},
+            )
+
+        # Wait for pods to terminate
+        print("Waiting for main pods to terminate...")
+        deadline = time.monotonic() + 120
+        while time.monotonic() < deadline:
+            pods = deployer.core_api.list_namespaced_pod(
+                namespace=namespace,
+                label_selector=f"app={app_name}",
+            )
+            # Count non-maintenance pods
+            active = sum(
+                1
+                for p in pods.items
+                if p.metadata
+                and p.metadata.deletion_timestamp is None
+                and not any(
+                    maint_container in (c.name or "") for c in (p.spec.containers or [])
+                )
+            )
+            if active == 0:
+                break
+            time.sleep(2)
+
+        # Scale back up
+        replicas = deployment_context.spec.get_replicas()
+        for dep_name in main_deployments:
+            print(f"Scaling up {dep_name} to {replicas} replicas...")
+            apps_api.patch_namespaced_deployment_scale(
+                name=dep_name,
+                namespace=namespace,
+                body={"spec": {"replicas": replicas}},
+            )
+
+        # Step 5: Wait for readiness
+        print("Waiting for main pods to become ready...")
+        deadline = time.monotonic() + 300
+        while time.monotonic() < deadline:
+            all_ready = True
+            for dep_name in main_deployments:
+                dep = apps_api.read_namespaced_deployment(
+                    name=dep_name, namespace=namespace
+                )
+                ready = dep.status.ready_replicas or 0
+                desired = dep.spec.replicas or 1
+                if ready < desired:
+                    all_ready = False
+                    break
+            if all_ready:
+                break
+            time.sleep(5)
+
+    # Step 6: Restore original Ingress backends
+    print("Restoring original Ingress backends...")
+    ingress = networking_api.read_namespaced_ingress(
+        name=ingress_name, namespace=namespace
+    )
+    for i, rule in enumerate(ingress.spec.rules):
+        for j, path in enumerate(rule.http.paths):
+            if i < len(original_backends) and j < len(original_backends[i]):
+                path.backend.service.name = original_backends[i][j]["name"]
+                path.backend.service.port.number = original_backends[i][j]["port"]
+
+    networking_api.replace_namespaced_ingress(
+        name=ingress_name, namespace=namespace, body=ingress
+    )
+    print("Ingress restored to original backends")

stack_orchestrator/deploy/deployment_create.py

@@ -265,6 +265,25 @@ def call_stack_deploy_create(deployment_context, extra_args):
                 imported_stack.create(deployment_context, extra_args)
 
 
+def call_stack_deploy_start(deployment_context):
+    """Call start() hooks after k8s deployments and jobs are created.
+
+    The start() hook receives the DeploymentContext, allowing stacks to
+    create additional k8s resources (Services, etc.) in the deployment namespace.
+    The namespace can be derived as f"laconic-{deployment_context.id}".
+    """
+    python_file_paths = _commands_plugin_paths(deployment_context.stack.name)
+    for python_file_path in python_file_paths:
+        if python_file_path.exists():
+            spec = util.spec_from_file_location("commands", python_file_path)
+            if spec is None or spec.loader is None:
+                continue
+            imported_stack = util.module_from_spec(spec)
+            spec.loader.exec_module(imported_stack)
+            if _has_method(imported_stack, "start"):
+                imported_stack.start(deployment_context)
+
+
 # Inspect the pod yaml to find config files referenced in subdirectories
 # other than the one associated with the pod
 def _find_extra_config_dirs(parsed_pod_file, pod):
@@ -477,6 +496,9 @@ def init_operation(
             spec_file_content["volumes"] = {**volume_descriptors, **orig_volumes}
         if configmap_descriptors:
             spec_file_content["configmaps"] = configmap_descriptors
+        if "k8s" in deployer_type:
+            if "secrets" not in spec_file_content:
+                spec_file_content["secrets"] = {}
 
     if opts.o.debug:
         print(
@@ -555,7 +577,9 @@ def _generate_and_store_secrets(config_vars: dict, deployment_name: str):
     return secrets
 
 
-def create_registry_secret(spec: Spec, deployment_name: str) -> Optional[str]:
+def create_registry_secret(
+    spec: Spec, deployment_name: str, namespace: str = "default"
+) -> Optional[str]:
     """Create K8s docker-registry secret from spec + environment.
 
     Reads registry configuration from spec.yml and creates a Kubernetes
@@ -564,6 +588,7 @@ def create_registry_secret(spec: Spec, deployment_name: str) -> Optional[str]:
     Args:
         spec: The deployment spec containing image-registry config
         deployment_name: Name of the deployment (used for secret naming)
+        namespace: K8s namespace to create the secret in
 
     Returns:
         The secret name if created, None if no registry config
@@ -577,16 +602,29 @@ def create_registry_secret(spec: Spec, deployment_name: str) -> Optional[str]:
     server = registry_config.get("server")
     username = registry_config.get("username")
     token_env = registry_config.get("token-env")
+    token_file = registry_config.get("token-file")
 
-    if not all([server, username, token_env]):
+    if not server or not username:
+        return None
+    if not token_env and not token_file:
         return None
 
-    # Type narrowing for pyright - we've validated these aren't None above
-    assert token_env is not None
-    token = os.environ.get(token_env)
+    # Resolve token: file takes precedence over env var
+    token = None
+    if token_file:
+        token_path = os.path.expanduser(token_file)
+        if os.path.exists(token_path):
+            with open(token_path) as f:
+                token = f.read().strip()
+        else:
+            print(f"Warning: Registry token file '{token_path}' not found")
+    if not token and token_env:
+        token = os.environ.get(token_env)
+
     if not token:
+        source = token_file or token_env
         print(
-            f"Warning: Registry token env var '{token_env}' not set, "
+            f"Warning: Registry token not available from '{source}', "
             "skipping registry secret"
         )
         return None
@@ -598,7 +636,7 @@ def create_registry_secret(spec: Spec, deployment_name: str) -> Optional[str]:
     }
 
     # Secret name derived from deployment name
-    secret_name = f"{deployment_name}-registry"
+    secret_name = f"{deployment_name}-image-pull-secret"
 
     # Load kube config
     try:
@@ -611,7 +649,6 @@ def create_registry_secret(spec: Spec, deployment_name: str) -> Optional[str]:
             return None
 
     v1 = client.CoreV1Api()
-    namespace = "default"
 
     k8s_secret = client.V1Secret(
         metadata=client.V1ObjectMeta(name=secret_name),
@@ -653,6 +690,15 @@ def _write_config_file(
 
     # Write non-secret config to config.env (exclude $generate:...$ tokens)
     with open(config_env_file, "w") as output_file:
+        output_file.write(
+            "# AUTO-GENERATED by laconic-so from spec.yml config section.\n"
+            "# Source: stack_orchestrator/deploy/deployment_create.py"
+            " _write_config_file()\n"
+            "# Do not edit — changes will be overwritten on deploy create"
+            " or restart.\n"
+            "# To change config, edit the config section in your spec.yml"
+            " and redeploy.\n"
+        )
         if config_vars:
             for variable_name, variable_value in config_vars.items():
                 # Skip variables with generate tokens - they go to K8s Secret
@@ -662,6 +708,19 @@ def _write_config_file(
                     continue
                 output_file.write(f"{variable_name}={variable_value}\n")
 
+        # Append contents of credentials files listed in spec
+        credentials_files = spec_content.get("credentials-files", []) or []
+        for cred_path_str in credentials_files:
+            cred_path = Path(cred_path_str).expanduser()
+            if not cred_path.exists():
+                print(f"Error: credentials file does not exist: {cred_path}")
+                sys.exit(1)
+            output_file.write(f"# From credentials file: {cred_path_str}\n")
+            contents = cred_path.read_text()
+            output_file.write(contents)
+            if not contents.endswith("\n"):
+                output_file.write("\n")
+
 
 def _write_kube_config_file(external_path: Path, internal_path: Path):
     if not external_path.exists():
@@ -813,9 +872,7 @@ def create_operation(
             # Copy from temp to deployment dir, excluding data volumes
             # and backing up changed files.
             # Exclude data/* to avoid touching user data volumes.
-            # Exclude config file to preserve deployment settings
-            # (XXX breaks passing config vars from spec)
-            exclude_patterns = ["data", "data/*", constants.config_file_name]
+            exclude_patterns = ["data", "data/*"]
             _safe_copy_tree(
                 temp_dir, deployment_dir_path, exclude_patterns=exclude_patterns
             )
@@ -982,17 +1039,7 @@ def _write_deployment_files(
             script_paths = get_pod_script_paths(parsed_stack, pod)
             _copy_files_to_directory(script_paths, destination_script_dir)
 
-        if parsed_spec.is_kubernetes_deployment():
-            for configmap in parsed_spec.get_configmaps():
-                source_config_dir = resolve_config_dir(stack_name, configmap)
-                if os.path.exists(source_config_dir):
-                    destination_config_dir = target_dir.joinpath(
-                        "configmaps", configmap
-                    )
-                    copytree(
-                        source_config_dir, destination_config_dir, dirs_exist_ok=True
-                    )
-        else:
+        if not parsed_spec.is_kubernetes_deployment():
             # TODO:
             # This is odd - looks up config dir that matches a volume name,
             # then copies as a mount dir?
@@ -1014,9 +1061,18 @@ def _write_deployment_files(
                             dirs_exist_ok=True,
                         )
 
-    # Copy the job files into the target dir (for Docker deployments)
+    # Copy configmap directories for k8s deployments (outside the pod loop
+    # so this works for jobs-only stacks too)
+    if parsed_spec.is_kubernetes_deployment():
+        for configmap in parsed_spec.get_configmaps():
+            source_config_dir = resolve_config_dir(stack_name, configmap)
+            if os.path.exists(source_config_dir):
+                destination_config_dir = target_dir.joinpath("configmaps", configmap)
+                copytree(source_config_dir, destination_config_dir, dirs_exist_ok=True)
+
+    # Copy the job files into the target dir
     jobs = get_job_list(parsed_stack)
-    if jobs and not parsed_spec.is_kubernetes_deployment():
+    if jobs:
         destination_compose_jobs_dir = target_dir.joinpath("compose-jobs")
         os.makedirs(destination_compose_jobs_dir, exist_ok=True)
         for job in jobs:

stack_orchestrator/deploy/k8s/cluster_info.py

@@ -72,15 +72,24 @@ def to_k8s_resource_requirements(resources: Resources) -> client.V1ResourceRequi
 
 class ClusterInfo:
     parsed_pod_yaml_map: Any
+    parsed_job_yaml_map: Any
     image_set: Set[str] = set()
     app_name: str
+    stack_name: str
     environment_variables: DeployEnvVars
     spec: Spec
 
     def __init__(self) -> None:
-        pass
+        self.parsed_job_yaml_map = {}
 
-    def int(self, pod_files: List[str], compose_env_file, deployment_name, spec: Spec):
+    def int(
+        self,
+        pod_files: List[str],
+        compose_env_file,
+        deployment_name,
+        spec: Spec,
+        stack_name="",
+    ):
         self.parsed_pod_yaml_map = parsed_pod_files_map_from_file_names(pod_files)
         # Find the set of images in the pods
         self.image_set = images_for_deployment(pod_files)
@@ -90,10 +99,23 @@ class ClusterInfo:
         }
         self.environment_variables = DeployEnvVars(env_vars)
         self.app_name = deployment_name
+        self.stack_name = stack_name
         self.spec = spec
         if opts.o.debug:
             print(f"Env vars: {self.environment_variables.map}")
 
+    def init_jobs(self, job_files: List[str]):
+        """Initialize parsed job YAML map from job compose files."""
+        self.parsed_job_yaml_map = parsed_pod_files_map_from_file_names(job_files)
+        if opts.o.debug:
+            print(f"Parsed job yaml map: {self.parsed_job_yaml_map}")
+
+    def _all_named_volumes(self) -> list:
+        """Return named volumes from both pod and job compose files."""
+        volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
+        volumes.extend(named_volumes_from_pod_files(self.parsed_job_yaml_map))
+        return volumes
+
     def get_nodeports(self):
         nodeports = []
         for pod_name in self.parsed_pod_yaml_map:
@@ -145,67 +167,99 @@ class ClusterInfo:
                         nodeports.append(service)
         return nodeports
 
+    def _resolve_service_name_for_container(self, container_name: str) -> str:
+        """Resolve the k8s Service name that routes to a given container.
+
+        For multi-pod stacks, each pod has its own Service. We find which
+        pod file contains this container and return the corresponding
+        service name. For single-pod stacks, returns the legacy service name.
+        """
+        pod_files = list(self.parsed_pod_yaml_map.keys())
+        multi_pod = len(pod_files) > 1
+
+        if not multi_pod:
+            return f"{self.app_name}-service"
+
+        for pod_file in pod_files:
+            pod = self.parsed_pod_yaml_map[pod_file]
+            if container_name in pod.get("services", {}):
+                pod_name = self._pod_name_from_file(pod_file)
+                return f"{self.app_name}-{pod_name}-service"
+
+        # Fallback: container not found in any pod file
+        return f"{self.app_name}-service"
+
     def get_ingress(
-        self, use_tls=False, certificate=None, cluster_issuer="letsencrypt-prod"
+        self, use_tls=False, certificates=None, cluster_issuer="letsencrypt-prod"
     ):
         # No ingress for a deployment that has no http-proxy defined, for now
         http_proxy_info_list = self.spec.get_http_proxy()
         ingress = None
         if http_proxy_info_list:
-            # TODO: handle multiple definitions
-            http_proxy_info = http_proxy_info_list[0]
-            if opts.o.debug:
-                print(f"http-proxy: {http_proxy_info}")
-            # TODO: good enough parsing for webapp deployment for now
-            host_name = http_proxy_info["host-name"]
             rules = []
-            tls = (
-                [
-                    client.V1IngressTLS(
-                        hosts=certificate["spec"]["dnsNames"]
-                        if certificate
-                        else [host_name],
-                        secret_name=certificate["spec"]["secretName"]
-                        if certificate
-                        else f"{self.app_name}-tls",
-                    )
-                ]
-                if use_tls
-                else None
-            )
-            paths = []
-            for route in http_proxy_info["routes"]:
-                path = route["path"]
-                proxy_to = route["proxy-to"]
+            tls = [] if use_tls else None
+
+            for http_proxy_info in http_proxy_info_list:
                 if opts.o.debug:
-                    print(f"proxy config: {path} -> {proxy_to}")
-                # proxy_to has the form <service>:<port>
-                proxy_to_port = int(proxy_to.split(":")[1])
-                paths.append(
-                    client.V1HTTPIngressPath(
-                        path_type="Prefix",
-                        path=path,
-                        backend=client.V1IngressBackend(
-                            service=client.V1IngressServiceBackend(
-                                # TODO: this looks wrong
-                                name=f"{self.app_name}-service",
-                                # TODO: pull port number from the service
-                                port=client.V1ServiceBackendPort(number=proxy_to_port),
-                            )
-                        ),
+                    print(f"http-proxy: {http_proxy_info}")
+                host_name = http_proxy_info["host-name"]
+                certificate = (certificates or {}).get(host_name)
+
+                if use_tls:
+                    tls.append(
+                        client.V1IngressTLS(
+                            hosts=(
+                                certificate["spec"]["dnsNames"]
+                                if certificate
+                                else [host_name]
+                            ),
+                            secret_name=(
+                                certificate["spec"]["secretName"]
+                                if certificate
+                                else f"{self.app_name}-{host_name}-tls"
+                            ),
+                        )
+                    )
+
+                paths = []
+                for route in http_proxy_info["routes"]:
+                    path = route["path"]
+                    proxy_to = route["proxy-to"]
+                    if opts.o.debug:
+                        print(f"proxy config: {path} -> {proxy_to}")
+                    # proxy_to has the form <service>:<port>
+                    container_name = proxy_to.split(":")[0]
+                    proxy_to_port = int(proxy_to.split(":")[1])
+                    service_name = self._resolve_service_name_for_container(
+                        container_name
+                    )
+                    paths.append(
+                        client.V1HTTPIngressPath(
+                            path_type="Prefix",
+                            path=path,
+                            backend=client.V1IngressBackend(
+                                service=client.V1IngressServiceBackend(
+                                    name=service_name,
+                                    port=client.V1ServiceBackendPort(
+                                        number=proxy_to_port
+                                    ),
+                                )
+                            ),
+                        )
+                    )
+                rules.append(
+                    client.V1IngressRule(
+                        host=host_name,
+                        http=client.V1HTTPIngressRuleValue(paths=paths),
                     )
                 )
-            rules.append(
-                client.V1IngressRule(
-                    host=host_name, http=client.V1HTTPIngressRuleValue(paths=paths)
-                )
-            )
+
             spec = client.V1IngressSpec(tls=tls, rules=rules)
 
             ingress_annotations = {
                 "kubernetes.io/ingress.class": "caddy",
             }
-            if not certificate:
+            if not certificates:
                 ingress_annotations["cert-manager.io/cluster-issuer"] = cluster_issuer
 
             ingress = client.V1Ingress(
@@ -218,6 +272,28 @@ class ClusterInfo:
             )
         return ingress
 
+    def _get_readiness_probe_ports(self) -> dict:
+        """Map container names to TCP readiness probe ports.
+
+        Derives probe ports from http-proxy routes in the spec. If a container
+        has an http-proxy route (proxy-to: container:port), we probe that port.
+        This tells k8s when the container is ready to serve traffic, which is
+        required for safe rolling updates.
+        """
+        probe_ports: dict = {}
+        http_proxy_list = self.spec.get_http_proxy()
+        if http_proxy_list:
+            for http_proxy in http_proxy_list:
+                for route in http_proxy.get("routes", []):
+                    proxy_to = route.get("proxy-to", "")
+                    if ":" in proxy_to:
+                        container, port_str = proxy_to.rsplit(":", 1)
+                        port = int(port_str)
+                        # Use the first route's port for each container
+                        if container not in probe_ports:
+                            probe_ports[container] = port
+        return probe_ports
+
     # TODO: suppoprt multiple services
     def get_service(self):
         # Collect all ports from http-proxy routes
@@ -257,20 +333,25 @@ class ClusterInfo:
     def get_pvcs(self):
         result = []
         spec_volumes = self.spec.get_volumes()
-        named_volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
-        resources = self.spec.get_volume_resources()
-        if not resources:
-            resources = DEFAULT_VOLUME_RESOURCES
+        named_volumes = self._all_named_volumes()
+        global_resources = self.spec.get_volume_resources()
+        if not global_resources:
+            global_resources = DEFAULT_VOLUME_RESOURCES
         if opts.o.debug:
             print(f"Spec Volumes: {spec_volumes}")
             print(f"Named Volumes: {named_volumes}")
-            print(f"Resources: {resources}")
+            print(f"Resources: {global_resources}")
         for volume_name, volume_path in spec_volumes.items():
             if volume_name not in named_volumes:
                 if opts.o.debug:
                     print(f"{volume_name} not in pod files")
                 continue
 
+            # Per-volume resources override global, which overrides default.
+            vol_resources = (
+                self.spec.get_volume_resources_for(volume_name) or global_resources
+            )
+
             labels = {
                 "app": self.app_name,
                 "volume-label": f"{self.app_name}-{volume_name}",
@@ -286,7 +367,7 @@ class ClusterInfo:
             spec = client.V1PersistentVolumeClaimSpec(
                 access_modes=["ReadWriteOnce"],
                 storage_class_name=storage_class_name,
-                resources=to_k8s_resource_requirements(resources),
+                resources=to_k8s_resource_requirements(vol_resources),
                 volume_name=k8s_volume_name,
             )
             pvc = client.V1PersistentVolumeClaim(
@@ -301,13 +382,14 @@ class ClusterInfo:
     def get_configmaps(self):
         result = []
         spec_configmaps = self.spec.get_configmaps()
-        named_volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
+        named_volumes = self._all_named_volumes()
         for cfg_map_name, cfg_map_path in spec_configmaps.items():
             if cfg_map_name not in named_volumes:
                 if opts.o.debug:
                     print(f"{cfg_map_name} not in pod files")
                 continue
 
+            cfg_map_path = os.path.expanduser(cfg_map_path)
             if not cfg_map_path.startswith("/") and self.spec.file_path is not None:
                 cfg_map_path = os.path.join(
                     os.path.dirname(str(self.spec.file_path)), cfg_map_path
@@ -337,10 +419,10 @@ class ClusterInfo:
     def get_pvs(self):
         result = []
         spec_volumes = self.spec.get_volumes()
-        named_volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
-        resources = self.spec.get_volume_resources()
-        if not resources:
-            resources = DEFAULT_VOLUME_RESOURCES
+        named_volumes = self._all_named_volumes()
+        global_resources = self.spec.get_volume_resources()
+        if not global_resources:
+            global_resources = DEFAULT_VOLUME_RESOURCES
         for volume_name, volume_path in spec_volumes.items():
             # We only need to create a volume if it is fully qualified HostPath.
             # Otherwise, we create the PVC and expect the node to allocate the volume
@@ -369,16 +451,23 @@ class ClusterInfo:
                     )
                     continue
 
+            vol_resources = (
+                self.spec.get_volume_resources_for(volume_name) or global_resources
+            )
             if self.spec.is_kind_deployment():
                 host_path = client.V1HostPathVolumeSource(
-                    path=get_kind_pv_bind_mount_path(volume_name)
+                    path=get_kind_pv_bind_mount_path(
+                        volume_name,
+                        kind_mount_root=self.spec.get_kind_mount_root(),
+                        host_path=volume_path,
+                    )
                 )
             else:
                 host_path = client.V1HostPathVolumeSource(path=volume_path)
             spec = client.V1PersistentVolumeSpec(
                 storage_class_name="manual",
                 access_modes=["ReadWriteOnce"],
-                capacity=to_k8s_resource_requirements(resources).requests,
+                capacity=to_k8s_resource_requirements(vol_resources).requests,
                 host_path=host_path,
             )
             pv = client.V1PersistentVolume(
@@ -424,15 +513,30 @@ class ClusterInfo:
         # 3. Fall back to spec.yml global (already resolved with DEFAULT fallback)
         return global_resources
 
-    # TODO: put things like image pull policy into an object-scope struct
-    def get_deployment(self, image_pull_policy: Optional[str] = None):
+    def _build_containers(
+        self,
+        parsed_yaml_map: Any,
+        image_pull_policy: Optional[str] = None,
+    ) -> tuple:
+        """Build k8s container specs from parsed compose YAML.
+
+        Returns a tuple of (containers, init_containers, services, volumes)
+        where:
+        - containers: list of V1Container objects
+        - init_containers: list of V1Container objects for init containers
+          (compose services with label ``laconic.init-container: "true"``)
+        - services: the last services dict processed (used for annotations/labels)
+        - volumes: list of V1Volume objects
+        """
         containers = []
+        init_containers = []
         services = {}
+        readiness_probe_ports = self._get_readiness_probe_ports()
         global_resources = self.spec.get_container_resources()
         if not global_resources:
             global_resources = DEFAULT_CONTAINER_RESOURCES
-        for pod_name in self.parsed_pod_yaml_map:
-            pod = self.parsed_pod_yaml_map[pod_name]
+        for pod_name in parsed_yaml_map:
+            pod = parsed_yaml_map[pod_name]
             services = pod["services"]
             for service_name in services:
                 container_name = service_name
@@ -488,9 +592,7 @@ class ClusterInfo:
                     if self.spec.get_image_registry() is not None
                     else image
                 )
-                volume_mounts = volume_mounts_for_service(
-                    self.parsed_pod_yaml_map, service_name
-                )
+                volume_mounts = volume_mounts_for_service(parsed_yaml_map, service_name)
                 # Handle command/entrypoint from compose file
                 # In docker-compose: entrypoint -> k8s command, command -> k8s args
                 container_command = None
@@ -513,9 +615,29 @@ class ClusterInfo:
                         )
                     )
                 ]
+                # Mount user-declared secrets from spec.yml
+                for user_secret_name in self.spec.get_secrets():
+                    env_from.append(
+                        client.V1EnvFromSource(
+                            secret_ref=client.V1SecretEnvSource(
+                                name=user_secret_name,
+                                optional=True,
+                            )
+                        )
+                    )
                 container_resources = self._resolve_container_resources(
                     container_name, service_info, global_resources
                 )
+                # Readiness probe from http-proxy routes
+                readiness_probe = None
+                probe_port = readiness_probe_ports.get(container_name)
+                if probe_port:
+                    readiness_probe = client.V1Probe(
+                        tcp_socket=client.V1TCPSocketAction(port=probe_port),
+                        initial_delay_seconds=5,
+                        period_seconds=10,
+                        failure_threshold=3,
+                    )
                 container = client.V1Container(
                     name=container_name,
                     image=image_to_use,
@@ -526,29 +648,78 @@ class ClusterInfo:
                     env_from=env_from,
                     ports=container_ports if container_ports else None,
                     volume_mounts=volume_mounts,
+                    readiness_probe=readiness_probe,
                     security_context=client.V1SecurityContext(
                         privileged=self.spec.get_privileged(),
-                        capabilities=client.V1Capabilities(
-                            add=self.spec.get_capabilities()
-                        )
-                        if self.spec.get_capabilities()
-                        else None,
+                        run_as_user=(
+                            int(service_info["user"])
+                            if "user" in service_info
+                            else None
+                        ),
+                        capabilities=(
+                            client.V1Capabilities(add=self.spec.get_capabilities())
+                            if self.spec.get_capabilities()
+                            else None
+                        ),
                     ),
                     resources=to_k8s_resource_requirements(container_resources),
                 )
-                containers.append(container)
-        volumes = volumes_for_pod_files(
-            self.parsed_pod_yaml_map, self.spec, self.app_name
-        )
-        registry_config = self.spec.get_image_registry_config()
-        if registry_config:
-            secret_name = f"{self.app_name}-registry"
-            image_pull_secrets = [client.V1LocalObjectReference(name=secret_name)]
-        else:
-            image_pull_secrets = []
+                # Services with laconic.init-container label become
+                # k8s init containers instead of regular containers.
+                svc_labels = service_info.get("labels", {})
+                if isinstance(svc_labels, list):
+                    # docker-compose labels can be a list of "key=value"
+                    svc_labels = dict(item.split("=", 1) for item in svc_labels)
+                is_init = str(svc_labels.get("laconic.init-container", "")).lower() in (
+                    "true",
+                    "1",
+                    "yes",
+                )
+                if is_init:
+                    init_containers.append(container)
+                else:
+                    containers.append(container)
+        volumes = volumes_for_pod_files(parsed_yaml_map, self.spec, self.app_name)
+        return containers, init_containers, services, volumes
 
+    def _pod_name_from_file(self, pod_file: str) -> str:
+        """Extract pod name from compose file path.
+
+        docker-compose-dumpster.yml -> dumpster
+        docker-compose-dumpster-maintenance.yml -> dumpster-maintenance
+        """
+        import os
+
+        base = os.path.basename(pod_file)
+        name = base
+        if name.startswith("docker-compose-"):
+            name = name[len("docker-compose-") :]
+        if name.endswith(".yml"):
+            name = name[: -len(".yml")]
+        elif name.endswith(".yaml"):
+            name = name[: -len(".yaml")]
+        return name
+
+    def _pod_has_pvcs(self, parsed_pod_file: Any) -> bool:
+        """Check if a parsed compose file declares volumes that become PVCs.
+
+        Excludes volumes that are ConfigMaps (declared in spec.configmaps),
+        since those don't require Recreate strategy.
+        """
+        volumes = parsed_pod_file.get("volumes", {})
+        configmaps = set(self.spec.get_configmaps().keys())
+        pvc_volumes = [v for v in volumes if v not in configmaps]
+        return len(pvc_volumes) > 0
+
+    def _build_common_pod_metadata(self, services: dict) -> tuple:
+        """Build shared annotations, labels, affinity, tolerations for pods.
+
+        Returns (annotations, labels, affinity, tolerations).
+        """
         annotations = None
         labels = {"app": self.app_name}
+        if self.stack_name:
+            labels["app.kubernetes.io/stack"] = self.stack_name
         affinity = None
         tolerations = None
 
@@ -566,7 +737,6 @@ class ClusterInfo:
         if self.spec.get_node_affinities():
             affinities = []
             for rule in self.spec.get_node_affinities():
-                # TODO add some input validation here
                 label_name = rule["label"]
                 label_value = rule["value"]
                 affinities.append(
@@ -589,7 +759,6 @@ class ClusterInfo:
         if self.spec.get_node_tolerations():
             tolerations = []
             for toleration in self.spec.get_node_tolerations():
-                # TODO add some input validation here
                 toleration_key = toleration["key"]
                 toleration_value = toleration["value"]
                 tolerations.append(
@@ -601,30 +770,430 @@ class ClusterInfo:
                     )
                 )
 
+        return annotations, labels, affinity, tolerations
+
+    # TODO: put things like image pull policy into an object-scope struct
+    def get_deployment(self, image_pull_policy: Optional[str] = None):
+        """Build a single k8s Deployment from all pod files (legacy behavior).
+
+        When only one pod is defined in the stack, this is equivalent to
+        get_deployments()[0]. Kept for backward compatibility.
+        """
+        deployments = self.get_deployments(image_pull_policy)
+        if not deployments:
+            return None
+        # Legacy: return the first (and usually only) deployment
+        return deployments[0]
+
+    def get_deployments(
+        self, image_pull_policy: Optional[str] = None
+    ) -> List[client.V1Deployment]:
+        """Build one k8s Deployment per pod file.
+
+        Each pod file (docker-compose-<name>.yml) becomes its own Deployment
+        with independent lifecycle and update strategy:
+        - Pods with PVCs get strategy=Recreate (can't do rolling updates
+          with ReadWriteOnce volumes)
+        - Pods without PVCs get strategy=RollingUpdate
+
+        This enables maintenance services to survive main pod restarts.
+        """
+        if not self.parsed_pod_yaml_map:
+            return []
+
+        registry_config = self.spec.get_image_registry_config()
+        if registry_config:
+            secret_name = f"{self.app_name}-image-pull-secret"
+            image_pull_secrets = [client.V1LocalObjectReference(name=secret_name)]
+        else:
+            image_pull_secrets = []
+
         use_host_network = self._any_service_has_host_network()
-        template = client.V1PodTemplateSpec(
-            metadata=client.V1ObjectMeta(annotations=annotations, labels=labels),
-            spec=client.V1PodSpec(
-                containers=containers,
-                image_pull_secrets=image_pull_secrets,
-                volumes=volumes,
-                affinity=affinity,
-                tolerations=tolerations,
-                runtime_class_name=self.spec.get_runtime_class(),
-                host_network=use_host_network or None,
-                dns_policy=("ClusterFirstWithHostNet" if use_host_network else None),
+        pod_files = list(self.parsed_pod_yaml_map.keys())
+
+        # Single pod file: preserve legacy naming ({app_name}-deployment)
+        # Multiple pod files: use {app_name}-{pod_name}-deployment
+        multi_pod = len(pod_files) > 1
+
+        deployments = []
+        for pod_file in pod_files:
+            pod_name = self._pod_name_from_file(pod_file)
+            single_pod_map = {pod_file: self.parsed_pod_yaml_map[pod_file]}
+            containers, init_containers, services, volumes = self._build_containers(
+                single_pod_map, image_pull_policy
+            )
+            annotations, labels, affinity, tolerations = (
+                self._build_common_pod_metadata(services)
+            )
+
+            # Add pod-name label so Services can target specific pods
+            if multi_pod:
+                labels["app.kubernetes.io/component"] = pod_name
+
+            has_pvcs = self._pod_has_pvcs(self.parsed_pod_yaml_map[pod_file])
+            if has_pvcs:
+                strategy = client.V1DeploymentStrategy(type="Recreate")
+            else:
+                strategy = client.V1DeploymentStrategy(
+                    type="RollingUpdate",
+                    rolling_update=client.V1RollingUpdateDeployment(
+                        max_unavailable=0, max_surge=1
+                    ),
+                )
+
+            # Pod selector: for multi-pod, select by both app and component
+            selector_labels = {"app": self.app_name}
+            if multi_pod:
+                selector_labels["app.kubernetes.io/component"] = pod_name
+
+            # Add CA certificate volume and env vars if configured
+            _ca_secret, ca_volume, ca_mounts, ca_envs = (
+                self.get_ca_certificate_resources()
+            )
+            if ca_volume:
+                volumes.append(ca_volume)
+                for container in containers:
+                    if container.volume_mounts is None:
+                        container.volume_mounts = []
+                    container.volume_mounts.extend(ca_mounts)
+                    if container.env is None:
+                        container.env = []
+                    container.env.extend(ca_envs)
+
+            template = client.V1PodTemplateSpec(
+                metadata=client.V1ObjectMeta(annotations=annotations, labels=labels),
+                spec=client.V1PodSpec(
+                    containers=containers,
+                    init_containers=init_containers or None,
+                    image_pull_secrets=image_pull_secrets,
+                    volumes=volumes,
+                    affinity=affinity,
+                    tolerations=tolerations,
+                    runtime_class_name=self.spec.get_runtime_class(),
+                    host_network=use_host_network or None,
+                    dns_policy=(
+                        "ClusterFirstWithHostNet" if use_host_network else None
+                    ),
+                ),
+            )
+
+            if multi_pod:
+                deployment_name = f"{self.app_name}-{pod_name}-deployment"
+            else:
+                deployment_name = f"{self.app_name}-deployment"
+
+            spec = client.V1DeploymentSpec(
+                replicas=self.spec.get_replicas(),
+                template=template,
+                selector={"matchLabels": selector_labels},
+                strategy=strategy,
+            )
+
+            deployment = client.V1Deployment(
+                api_version="apps/v1",
+                kind="Deployment",
+                metadata=client.V1ObjectMeta(
+                    name=deployment_name,
+                    labels={
+                        "app": self.app_name,
+                        **(
+                            {
+                                "app.kubernetes.io/stack": self.stack_name,
+                            }
+                            if self.stack_name
+                            else {}
+                        ),
+                        **(
+                            {"app.kubernetes.io/component": pod_name}
+                            if multi_pod
+                            else {}
+                        ),
+                    },
+                ),
+                spec=spec,
+            )
+            deployments.append(deployment)
+
+        return deployments
+
+    def get_services(self) -> List[client.V1Service]:
+        """Build per-pod ClusterIP Services for multi-pod stacks.
+
+        Each pod's containers get their own Service so Ingress can route
+        to specific pods. For single-pod stacks, returns a list with one
+        service matching the legacy get_service() behavior.
+        """
+        pod_files = list(self.parsed_pod_yaml_map.keys())
+        multi_pod = len(pod_files) > 1
+
+        if not multi_pod:
+            # Legacy: single service for all pods
+            svc = self.get_service()
+            return [svc] if svc else []
+
+        # Multi-pod: one service per pod, only for pods that have
+        # ports referenced by http-proxy routes
+        http_proxy_list = self.spec.get_http_proxy()
+        if not http_proxy_list:
+            return []
+
+        # Build map: container_name -> port from http-proxy routes
+        container_ports: dict = {}
+        for http_proxy in http_proxy_list:
+            for route in http_proxy.get("routes", []):
+                proxy_to = route.get("proxy-to", "")
+                if ":" in proxy_to:
+                    container, port_str = proxy_to.rsplit(":", 1)
+                    port = int(port_str)
+                    if container not in container_ports:
+                        container_ports[container] = set()
+                    container_ports[container].add(port)
+
+        # Build map: pod_file -> set of service names in that pod
+        pod_services_map: dict = {}
+        for pod_file in pod_files:
+            pod = self.parsed_pod_yaml_map[pod_file]
+            pod_services_map[pod_file] = set(pod.get("services", {}).keys())
+
+        services = []
+        for pod_file in pod_files:
+            pod_name = self._pod_name_from_file(pod_file)
+            svc_names = pod_services_map[pod_file]
+            # Collect ports from http-proxy that belong to this pod's containers
+            ports_set: Set[int] = set()
+            for svc_name in svc_names:
+                if svc_name in container_ports:
+                    ports_set.update(container_ports[svc_name])
+
+            if not ports_set:
+                continue
+
+            service_ports = [
+                client.V1ServicePort(port=p, target_port=p, name=f"port-{p}")
+                for p in sorted(ports_set)
+            ]
+            service = client.V1Service(
+                metadata=client.V1ObjectMeta(
+                    name=f"{self.app_name}-{pod_name}-service",
+                    labels={"app": self.app_name},
+                ),
+                spec=client.V1ServiceSpec(
+                    type="ClusterIP",
+                    ports=service_ports,
+                    selector={
+                        "app": self.app_name,
+                        "app.kubernetes.io/component": pod_name,
+                    },
+                ),
+            )
+            services.append(service)
+        return services
+
+    def get_jobs(self, image_pull_policy: Optional[str] = None) -> List[client.V1Job]:
+        """Build k8s Job objects from parsed job compose files.
+
+        Each job compose file produces a V1Job with:
+        - restartPolicy: Never
+        - backoffLimit: 0
+        - Name: {app_name}-job-{job_name}
+        """
+        if not self.parsed_job_yaml_map:
+            return []
+
+        jobs = []
+        registry_config = self.spec.get_image_registry_config()
+        if registry_config:
+            secret_name = f"{self.app_name}-image-pull-secret"
+            image_pull_secrets = [client.V1LocalObjectReference(name=secret_name)]
+        else:
+            image_pull_secrets = []
+
+        for job_file in self.parsed_job_yaml_map:
+            # Build containers for this single job file
+            single_job_map = {job_file: self.parsed_job_yaml_map[job_file]}
+            containers, init_containers, _services, volumes = self._build_containers(
+                single_job_map, image_pull_policy
+            )
+
+            # Derive job name from file path: docker-compose-<name>.yml -> <name>
+            base = os.path.basename(job_file)
+            # Strip docker-compose- prefix and .yml suffix
+            job_name = base
+            if job_name.startswith("docker-compose-"):
+                job_name = job_name[len("docker-compose-") :]
+            if job_name.endswith(".yml"):
+                job_name = job_name[: -len(".yml")]
+            elif job_name.endswith(".yaml"):
+                job_name = job_name[: -len(".yaml")]
+
+            # Use a distinct app label for job pods so they don't get
+            # picked up by pods_in_deployment() which queries app={app_name}.
+            pod_labels = {
+                "app": f"{self.app_name}-job",
+                **(
+                    {"app.kubernetes.io/stack": self.stack_name}
+                    if self.stack_name
+                    else {}
+                ),
+            }
+            template = client.V1PodTemplateSpec(
+                metadata=client.V1ObjectMeta(labels=pod_labels),
+                spec=client.V1PodSpec(
+                    containers=containers,
+                    init_containers=init_containers or None,
+                    image_pull_secrets=image_pull_secrets,
+                    volumes=volumes,
+                    restart_policy="Never",
+                ),
+            )
+            job_spec = client.V1JobSpec(
+                template=template,
+                backoff_limit=0,
+            )
+            job_labels = {
+                "app": self.app_name,
+                **(
+                    {"app.kubernetes.io/stack": self.stack_name}
+                    if self.stack_name
+                    else {}
+                ),
+            }
+            job = client.V1Job(
+                api_version="batch/v1",
+                kind="Job",
+                metadata=client.V1ObjectMeta(
+                    name=f"{self.app_name}-job-{job_name}",
+                    labels=job_labels,
+                ),
+                spec=job_spec,
+            )
+            jobs.append(job)
+
+        return jobs
+
+    def get_external_service_resources(self) -> List:
+        """Build k8s Services (and Endpoints) for external-services in spec.
+
+        Two modes:
+        - host mode: ExternalName Service (DNS CNAME to external host)
+        - selector mode: headless Service + Endpoints (cross-namespace
+          routing to a mock pod, IP discovered at deploy time)
+
+        Returns a flat list of k8s resource objects (Services + Endpoints).
+        """
+        ext_services = self.spec.get_external_services()
+        if not ext_services:
+            return []
+
+        resources = []
+        for name, config in ext_services.items():
+            port = config.get("port", 443)
+
+            if "host" in config:
+                # ExternalName: DNS CNAME to external host
+                svc = client.V1Service(
+                    metadata=client.V1ObjectMeta(
+                        name=name,
+                        labels={"app": self.app_name},
+                    ),
+                    spec=client.V1ServiceSpec(
+                        type="ExternalName",
+                        external_name=config["host"],
+                        ports=[
+                            client.V1ServicePort(port=port, name=f"port-{port}")
+                        ],
+                    ),
+                )
+                resources.append(svc)
+
+            elif "selector" in config and "namespace" in config:
+                # Cross-namespace headless Service + Endpoints.
+                # The Endpoints IP is populated in deploy_k8s.py at deploy
+                # time by querying the target namespace for matching pods.
+                svc = client.V1Service(
+                    metadata=client.V1ObjectMeta(
+                        name=name,
+                        labels={"app": self.app_name},
+                    ),
+                    spec=client.V1ServiceSpec(
+                        cluster_ip="None",
+                        ports=[
+                            client.V1ServicePort(port=port, name=f"port-{port}")
+                        ],
+                    ),
+                )
+                resources.append(svc)
+                # Endpoints object is created in deploy_k8s.py after pod
+                # IP discovery — we just return the Service here.
+
+        return resources
+
+    def get_ca_certificate_resources(self) -> tuple:
+        """Build k8s Secret and volume mount config for CA certificates.
+
+        Returns (secret, volume, volume_mount, env_vars) or (None, ...) if
+        no CA certificates are configured. The caller must add the volume
+        and mount to all containers, and the env vars to all containers.
+        """
+        ca_files = self.spec.get_ca_certificates()
+        if not ca_files:
+            return None, None, None, []
+
+        # Concatenate all CA files into one Secret
+        secret_data = {}
+        for i, ca_path in enumerate(ca_files):
+            expanded = os.path.expanduser(ca_path)
+            if not os.path.exists(expanded):
+                print(f"Warning: CA certificate file not found: {expanded}")
+                continue
+            with open(expanded, "rb") as f:
+                ca_bytes = f.read()
+            key = f"laconic-extra-ca-{i}.pem"
+            secret_data[key] = base64.b64encode(ca_bytes).decode()
+
+        if not secret_data:
+            return None, None, None, []
+
+        secret_name = f"{self.app_name}-ca-certificates"
+        secret = client.V1Secret(
+            metadata=client.V1ObjectMeta(
+                name=secret_name,
+                labels={"app": self.app_name},
             ),
-        )
-        spec = client.V1DeploymentSpec(
-            replicas=self.spec.get_replicas(),
-            template=template,
-            selector={"matchLabels": {"app": self.app_name}},
+            data=secret_data,
         )
 
-        deployment = client.V1Deployment(
-            api_version="apps/v1",
-            kind="Deployment",
-            metadata=client.V1ObjectMeta(name=f"{self.app_name}-deployment"),
-            spec=spec,
+        volume = client.V1Volume(
+            name="laconic-ca-certs",
+            secret=client.V1SecretVolumeSource(
+                secret_name=secret_name,
+            ),
         )
-        return deployment
+
+        # Mount each CA file into /etc/ssl/certs/ (Go reads this dir)
+        # Mount each CA file directly into /etc/ssl/certs/ using subPath
+        # so Go's x509 package picks them up (it reads *.pem from that dir).
+        # Also return env vars for Node/Bun containers.
+        volume_mounts = []
+        first_mount_path = None
+        for key in secret_data.keys():
+            mount_path = f"/etc/ssl/certs/{key}"
+            if first_mount_path is None:
+                first_mount_path = mount_path
+            volume_mounts.append(
+                client.V1VolumeMount(
+                    name="laconic-ca-certs",
+                    mount_path=mount_path,
+                    sub_path=key,
+                    read_only=True,
+                )
+            )
+
+        env_vars = [
+            client.V1EnvVar(
+                name="NODE_EXTRA_CA_CERTS",
+                value=first_mount_path,
+            ),
+        ]
+
+        return secret, volume, volume_mounts, env_vars

stack_orchestrator/deploy/k8s/deploy_k8s.py

@@ -95,6 +95,7 @@ class K8sDeployer(Deployer):
     type: str
     core_api: client.CoreV1Api
     apps_api: client.AppsV1Api
+    batch_api: client.BatchV1Api
     networking_api: client.NetworkingV1Api
     k8s_namespace: str
     kind_cluster_name: str
@@ -110,9 +111,11 @@ class K8sDeployer(Deployer):
         compose_files,
         compose_project_name,
         compose_env_file,
+        job_compose_files=None,
     ) -> None:
         self.type = type
         self.skip_cluster_management = False
+        self.image_overrides = None
         self.k8s_namespace = "default"  # Will be overridden below if context exists
         # TODO: workaround pending refactoring above to cope with being
         # created with a null deployment_context
@@ -120,19 +123,32 @@ class K8sDeployer(Deployer):
             return
         self.deployment_dir = deployment_context.deployment_dir
         self.deployment_context = deployment_context
-        self.kind_cluster_name = compose_project_name
-        # Use deployment-specific namespace for resource isolation and easy cleanup
-        self.k8s_namespace = f"laconic-{compose_project_name}"
+        self.kind_cluster_name = (
+            deployment_context.spec.get_kind_cluster_name() or compose_project_name
+        )
+        # Use spec namespace if provided, otherwise derive from cluster-id
+        self.k8s_namespace = (
+            deployment_context.spec.get_namespace() or f"laconic-{compose_project_name}"
+        )
         self.cluster_info = ClusterInfo()
+        # stack.name may be an absolute path (from spec "stack:" key after
+        # path resolution). Extract just the directory basename for labels.
+        raw_name = deployment_context.stack.name if deployment_context else ""
+        stack_name = Path(raw_name).name if raw_name else ""
         self.cluster_info.int(
             compose_files,
             compose_env_file,
             compose_project_name,
             deployment_context.spec,
+            stack_name=stack_name,
         )
+        # Initialize job compose files if provided
+        if job_compose_files:
+            self.cluster_info.init_jobs(job_compose_files)
         if opts.o.debug:
             print(f"Deployment dir: {deployment_context.deployment_dir}")
             print(f"Compose files: {compose_files}")
+            print(f"Job compose files: {job_compose_files}")
             print(f"Project name: {compose_project_name}")
             print(f"Env file: {compose_env_file}")
             print(f"Type: {type}")
@@ -150,6 +166,7 @@ class K8sDeployer(Deployer):
         self.core_api = client.CoreV1Api()
         self.networking_api = client.NetworkingV1Api()
         self.apps_api = client.AppsV1Api()
+        self.batch_api = client.BatchV1Api()
         self.custom_obj_api = client.CustomObjectsApi()
 
     def _ensure_namespace(self):
@@ -192,6 +209,131 @@ class K8sDeployer(Deployer):
             else:
                 raise
 
+    def _wait_for_namespace_gone(self, timeout_seconds: int = 120):
+        """Wait for namespace to finish terminating."""
+        if opts.o.dry_run:
+            return
+        import time
+
+        deadline = time.monotonic() + timeout_seconds
+        while time.monotonic() < deadline:
+            try:
+                ns = self.core_api.read_namespace(name=self.k8s_namespace)
+                if ns.status and ns.status.phase == "Terminating":
+                    if opts.o.debug:
+                        print(
+                            f"Waiting for namespace {self.k8s_namespace}"
+                            " to finish terminating..."
+                        )
+                    time.sleep(2)
+                    continue
+                # Namespace exists and is Active — shouldn't happen after delete
+                break
+            except ApiException as e:
+                if e.status == 404:
+                    # Gone — success
+                    return
+                raise
+        # If we get here, namespace still exists after timeout
+        try:
+            self.core_api.read_namespace(name=self.k8s_namespace)
+            print(
+                f"Warning: namespace {self.k8s_namespace} still exists"
+                f" after {timeout_seconds}s"
+            )
+        except ApiException as e:
+            if e.status == 404:
+                return
+            raise
+
+    def _delete_resources_by_label(self, label_selector: str, delete_volumes: bool):
+        """Delete only this stack's resources from a shared namespace."""
+        ns = self.k8s_namespace
+        if opts.o.dry_run:
+            print(f"Dry run: would delete resources with {label_selector} in {ns}")
+            return
+
+        # Deployments
+        try:
+            deps = self.apps_api.list_namespaced_deployment(
+                namespace=ns, label_selector=label_selector
+            )
+            for dep in deps.items:
+                print(f"Deleting Deployment {dep.metadata.name}")
+                self.apps_api.delete_namespaced_deployment(
+                    name=dep.metadata.name, namespace=ns
+                )
+        except ApiException as e:
+            _check_delete_exception(e)
+
+        # Jobs
+        try:
+            jobs = self.batch_api.list_namespaced_job(
+                namespace=ns, label_selector=label_selector
+            )
+            for job in jobs.items:
+                print(f"Deleting Job {job.metadata.name}")
+                self.batch_api.delete_namespaced_job(
+                    name=job.metadata.name,
+                    namespace=ns,
+                    body=client.V1DeleteOptions(propagation_policy="Background"),
+                )
+        except ApiException as e:
+            _check_delete_exception(e)
+
+        # Services (NodePorts created by SO)
+        try:
+            svcs = self.core_api.list_namespaced_service(
+                namespace=ns, label_selector=label_selector
+            )
+            for svc in svcs.items:
+                print(f"Deleting Service {svc.metadata.name}")
+                self.core_api.delete_namespaced_service(
+                    name=svc.metadata.name, namespace=ns
+                )
+        except ApiException as e:
+            _check_delete_exception(e)
+
+        # Ingresses
+        try:
+            ings = self.networking_api.list_namespaced_ingress(
+                namespace=ns, label_selector=label_selector
+            )
+            for ing in ings.items:
+                print(f"Deleting Ingress {ing.metadata.name}")
+                self.networking_api.delete_namespaced_ingress(
+                    name=ing.metadata.name, namespace=ns
+                )
+        except ApiException as e:
+            _check_delete_exception(e)
+
+        # ConfigMaps
+        try:
+            cms = self.core_api.list_namespaced_config_map(
+                namespace=ns, label_selector=label_selector
+            )
+            for cm in cms.items:
+                print(f"Deleting ConfigMap {cm.metadata.name}")
+                self.core_api.delete_namespaced_config_map(
+                    name=cm.metadata.name, namespace=ns
+                )
+        except ApiException as e:
+            _check_delete_exception(e)
+
+        # PVCs (only if --delete-volumes)
+        if delete_volumes:
+            try:
+                pvcs = self.core_api.list_namespaced_persistent_volume_claim(
+                    namespace=ns, label_selector=label_selector
+                )
+                for pvc in pvcs.items:
+                    print(f"Deleting PVC {pvc.metadata.name}")
+                    self.core_api.delete_namespaced_persistent_volume_claim(
+                        name=pvc.metadata.name, namespace=ns
+                    )
+            except ApiException as e:
+                _check_delete_exception(e)
+
     def _create_volume_data(self):
         # Create the host-path-mounted PVs for this deployment
         pvs = self.cluster_info.get_pvs()
@@ -204,7 +346,22 @@ class K8sDeployer(Deployer):
                         name=pv.metadata.name
                     )
                     if pv_resp:
-                        if opts.o.debug:
+                        # If PV is in Released state (stale claimRef from a
+                        # previous deployment), clear the claimRef so a new
+                        # PVC can bind to it. This happens after stop+start
+                        # because stop deletes the namespace (and PVCs) but
+                        # preserves PVs by default.
+                        if pv_resp.status and pv_resp.status.phase == "Released":
+                            print(
+                                f"PV {pv.metadata.name} is Released, "
+                                "clearing claimRef for rebinding"
+                            )
+                            pv_resp.spec.claim_ref = None
+                            self.core_api.patch_persistent_volume(
+                                name=pv.metadata.name,
+                                body={"spec": {"claimRef": None}},
+                            )
+                        elif opts.o.debug:
                             print("PVs already present:")
                             print(f"{pv_resp}")
                         continue
@@ -248,50 +405,272 @@ class K8sDeployer(Deployer):
             if opts.o.debug:
                 print(f"Sending this ConfigMap: {cfg_map}")
             if not opts.o.dry_run:
-                cfg_rsp = self.core_api.create_namespaced_config_map(
-                    body=cfg_map, namespace=self.k8s_namespace
+                cm_name = cfg_map.metadata.name
+                try:
+                    self.core_api.create_namespaced_config_map(
+                        body=cfg_map, namespace=self.k8s_namespace
+                    )
+                except ApiException as e:
+                    if e.status == 409:
+                        self.core_api.patch_namespaced_config_map(
+                            name=cm_name,
+                            namespace=self.k8s_namespace,
+                            body=cfg_map,
+                        )
+                    else:
+                        raise
+
+    def _create_external_services(self):
+        """Create k8s Services for external-services declared in the spec.
+
+        For host mode: ExternalName Service (DNS CNAME).
+        For selector mode: headless Service + Endpoints with pod IPs
+        discovered from the target namespace.
+        """
+        resources = self.cluster_info.get_external_service_resources()
+        ext_services = self.cluster_info.spec.get_external_services()
+
+        for resource in resources:
+            if opts.o.dry_run:
+                print(f"Dry run: would create external service: {resource.metadata.name}")
+                continue
+
+            svc_name = resource.metadata.name
+            try:
+                self.core_api.create_namespaced_service(
+                    body=resource, namespace=self.k8s_namespace
                 )
-                if opts.o.debug:
-                    print("ConfigMap created:")
-                    print(f"{cfg_rsp}")
+                print(f"Created external service '{svc_name}'")
+            except ApiException as e:
+                if e.status == 409:
+                    self.core_api.replace_namespaced_service(
+                        name=svc_name,
+                        namespace=self.k8s_namespace,
+                        body=resource,
+                    )
+                    print(f"Updated external service '{svc_name}'")
+                else:
+                    raise
+
+        # Create Endpoints for selector-mode services
+        for name, config in ext_services.items():
+            if "selector" not in config or "namespace" not in config:
+                continue
+            if opts.o.dry_run:
+                continue
+
+            target_ns = config["namespace"]
+            selector = config["selector"]
+            port = config.get("port", 443)
+
+            # Build label selector string from dict
+            label_selector = ",".join(f"{k}={v}" for k, v in selector.items())
+
+            # Discover pod IPs in target namespace
+            pods = self.core_api.list_namespaced_pod(
+                namespace=target_ns, label_selector=label_selector
+            )
+            pod_ips = [
+                p.status.pod_ip
+                for p in pods.items
+                if p.status and p.status.pod_ip
+            ]
+
+            if not pod_ips:
+                print(
+                    f"Warning: no pods found in {target_ns} matching "
+                    f"{label_selector} for external service '{name}'"
+                )
+                continue
+
+            endpoints = client.V1Endpoints(
+                metadata=client.V1ObjectMeta(
+                    name=name,
+                    labels={"app": self.cluster_info.app_name},
+                ),
+                subsets=[
+                    client.V1EndpointSubset(
+                        addresses=[
+                            client.V1EndpointAddress(ip=ip) for ip in pod_ips
+                        ],
+                        ports=[
+                            client.CoreV1EndpointPort(
+                                port=port, name=f"port-{port}"
+                            )
+                        ],
+                    )
+                ],
+            )
+
+            try:
+                self.core_api.create_namespaced_endpoints(
+                    body=endpoints, namespace=self.k8s_namespace
+                )
+                print(f"Created endpoints for '{name}' → {pod_ips}")
+            except ApiException as e:
+                if e.status == 409:
+                    self.core_api.replace_namespaced_endpoints(
+                        name=name,
+                        namespace=self.k8s_namespace,
+                        body=endpoints,
+                    )
+                    print(f"Updated endpoints for '{name}' → {pod_ips}")
+                else:
+                    raise
+
+    def _create_ca_certificates(self):
+        """Create k8s Secret for CA certificates declared in the spec.
+
+        The Secret is mounted into containers by get_deployments() in
+        cluster_info.py. This method just ensures the Secret exists.
+        """
+        ca_secret, _, _, _ = self.cluster_info.get_ca_certificate_resources()
+        if not ca_secret:
+            return
+        if opts.o.dry_run:
+            print(f"Dry run: would create CA certificate secret")
+            return
+
+        secret_name = ca_secret.metadata.name
+        try:
+            self.core_api.create_namespaced_secret(
+                body=ca_secret, namespace=self.k8s_namespace
+            )
+            print(f"Created CA certificate secret '{secret_name}'")
+        except ApiException as e:
+            if e.status == 409:
+                self.core_api.replace_namespaced_secret(
+                    name=secret_name,
+                    namespace=self.k8s_namespace,
+                    body=ca_secret,
+                )
+                print(f"Updated CA certificate secret '{secret_name}'")
+            else:
+                raise
 
     def _create_deployment(self):
-        # Process compose files into a Deployment
-        deployment = self.cluster_info.get_deployment(
-            image_pull_policy=None if self.is_kind() else "Always"
-        )
-        # Create the k8s objects
-        if opts.o.debug:
-            print(f"Sending this deployment: {deployment}")
-        if not opts.o.dry_run:
-            deployment_resp = cast(
-                client.V1Deployment,
-                self.apps_api.create_namespaced_deployment(
-                    body=deployment, namespace=self.k8s_namespace
-                ),
-            )
+        # Skip if there are no pods to deploy (e.g. jobs-only stacks)
+        if not self.cluster_info.parsed_pod_yaml_map:
             if opts.o.debug:
-                print("Deployment created:")
-                meta = deployment_resp.metadata
-                spec = deployment_resp.spec
-                if meta and spec and spec.template.spec:
-                    ns = meta.namespace
-                    name = meta.name
-                    gen = meta.generation
-                    containers = spec.template.spec.containers
-                    img = containers[0].image if containers else None
-                    print(f"{ns} {name} {gen} {img}")
+                print("No pods defined, skipping Deployment creation")
+            return
+        # Process compose files into Deployments (one per pod file)
+        # image-pull-policy from spec, default Always (production).
+        # Testing specs use IfNotPresent so kind-loaded local images are used.
+        pull_policy = self.cluster_info.spec.get("image-pull-policy", "Always")
+        deployments = self.cluster_info.get_deployments(image_pull_policy=pull_policy)
+        for deployment in deployments:
+            # Apply image overrides if provided
+            if self.image_overrides:
+                for container in deployment.spec.template.spec.containers:
+                    if container.name in self.image_overrides:
+                        container.image = self.image_overrides[container.name]
+                        if opts.o.debug:
+                            print(
+                                f"Overriding image for {container.name}:"
+                                f" {container.image}"
+                            )
+            # Create or update the k8s Deployment
+            if opts.o.debug:
+                print(f"Sending this deployment: {deployment}")
+            if not opts.o.dry_run:
+                name = deployment.metadata.name
+                try:
+                    deployment_resp = cast(
+                        client.V1Deployment,
+                        self.apps_api.create_namespaced_deployment(
+                            body=deployment, namespace=self.k8s_namespace
+                        ),
+                    )
+                    strategy = (
+                        deployment.spec.strategy.type
+                        if deployment.spec.strategy
+                        else "default"
+                    )
+                    print(f"Created Deployment {name} (strategy: {strategy})")
+                except ApiException as e:
+                    if e.status == 409:
+                        # Already exists — replace to ensure removed fields
+                        # (volumes, mounts, env vars) are actually deleted.
+                        existing = self.apps_api.read_namespaced_deployment(
+                            name=name, namespace=self.k8s_namespace
+                        )
+                        deployment.metadata.resource_version = (
+                            existing.metadata.resource_version
+                        )
+                        deployment_resp = cast(
+                            client.V1Deployment,
+                            self.apps_api.replace_namespaced_deployment(
+                                name=name,
+                                namespace=self.k8s_namespace,
+                                body=deployment,
+                            ),
+                        )
+                        print(f"Updated Deployment {name} (rolling update)")
+                    else:
+                        raise
+                if opts.o.debug:
+                    meta = deployment_resp.metadata
+                    spec = deployment_resp.spec
+                    if meta and spec and spec.template.spec:
+                        containers = spec.template.spec.containers
+                        img = containers[0].image if containers else None
+                        print(
+                            f"  {meta.namespace} {meta.name}"
+                            f" gen={meta.generation} {img}"
+                        )
 
-        service = self.cluster_info.get_service()
-        if opts.o.debug:
-            print(f"Sending this service: {service}")
-        if service and not opts.o.dry_run:
-            service_resp = self.core_api.create_namespaced_service(
-                namespace=self.k8s_namespace, body=service
-            )
+        # Create Services (one per pod for multi-pod, or one for single-pod)
+        services = self.cluster_info.get_services()
+        for service in services:
             if opts.o.debug:
-                print("Service created:")
-                print(f"{service_resp}")
+                print(f"Sending this service: {service}")
+            if service and not opts.o.dry_run:
+                svc_name = service.metadata.name
+                try:
+                    service_resp = self.core_api.create_namespaced_service(
+                        namespace=self.k8s_namespace, body=service
+                    )
+                    print(f"Created Service {svc_name}")
+                except ApiException as e:
+                    if e.status == 409:
+                        # Replace to ensure removed ports are deleted.
+                        # Must preserve clusterIP (immutable) and resourceVersion.
+                        existing = self.core_api.read_namespaced_service(
+                            name=svc_name, namespace=self.k8s_namespace
+                        )
+                        service.metadata.resource_version = (
+                            existing.metadata.resource_version
+                        )
+                        service.spec.cluster_ip = existing.spec.cluster_ip
+                        service_resp = self.core_api.replace_namespaced_service(
+                            name=svc_name,
+                            namespace=self.k8s_namespace,
+                            body=service,
+                        )
+                        print(f"Updated Service {svc_name}")
+                    else:
+                        raise
+                if opts.o.debug:
+                    print(f"  {service_resp}")
+
+    def _create_jobs(self):
+        # Process job compose files into k8s Jobs
+        jobs = self.cluster_info.get_jobs(image_pull_policy="Always")
+        for job in jobs:
+            if opts.o.debug:
+                print(f"Sending this job: {job}")
+            if not opts.o.dry_run:
+                job_resp = self.batch_api.create_namespaced_job(
+                    body=job, namespace=self.k8s_namespace
+                )
+                if opts.o.debug:
+                    print("Job created:")
+                    if job_resp.metadata:
+                        print(
+                            f"  {job_resp.metadata.namespace} "
+                            f"{job_resp.metadata.name}"
+                        )
 
     def _find_certificate_for_host_name(self, host_name):
         all_certificates = self.custom_obj_api.list_namespaced_custom_object(
@@ -329,113 +708,161 @@ class K8sDeployer(Deployer):
                                 return cert
         return None
 
-    def up(self, detach, skip_cluster_management, services):
+    def _setup_cluster(self):
+        """Create/reuse kind cluster, load images, ensure namespace."""
+        if self.is_kind() and not self.skip_cluster_management:
+            kind_config = str(
+                self.deployment_dir.joinpath(constants.kind_config_filename)
+            )
+            actual_cluster = create_cluster(self.kind_cluster_name, kind_config)
+            if actual_cluster != self.kind_cluster_name:
+                self.kind_cluster_name = actual_cluster
+            # Only load locally-built images into kind
+            local_containers = self.deployment_context.stack.obj.get("containers", [])
+            if local_containers:
+                local_images = {
+                    img
+                    for img in self.cluster_info.image_set
+                    if any(c in img for c in local_containers)
+                }
+                if local_images:
+                    load_images_into_kind(self.kind_cluster_name, local_images)
+        self.connect_api()
+        self._ensure_namespace()
+        if self.is_kind() and not self.skip_cluster_management:
+            if not is_ingress_running():
+                install_ingress_for_kind(self.cluster_info.spec.get_acme_email())
+                wait_for_ingress_in_kind()
+            if self.cluster_info.spec.get_unlimited_memlock():
+                _create_runtime_class(
+                    constants.high_memlock_runtime,
+                    constants.high_memlock_runtime,
+                )
+
+    def _create_ingress(self):
+        """Create or update Ingress with TLS certificate lookup."""
+        http_proxy_info = self.cluster_info.spec.get_http_proxy()
+        use_tls = http_proxy_info and not self.is_kind()
+        certificates = None
+        if use_tls:
+            certificates = {}
+            for proxy in http_proxy_info:
+                host_name = proxy["host-name"]
+                cert = self._find_certificate_for_host_name(host_name)
+                if cert:
+                    certificates[host_name] = cert
+                    if opts.o.debug:
+                        print(f"Using existing certificate for {host_name}: {cert}")
+
+        ingress = self.cluster_info.get_ingress(
+            use_tls=use_tls, certificates=certificates
+        )
+        if ingress:
+            if opts.o.debug:
+                print(f"Sending this ingress: {ingress}")
+            if not opts.o.dry_run:
+                ing_name = ingress.metadata.name
+                try:
+                    self.networking_api.create_namespaced_ingress(
+                        namespace=self.k8s_namespace, body=ingress
+                    )
+                    print(f"Created Ingress {ing_name}")
+                except ApiException as e:
+                    if e.status == 409:
+                        existing = self.networking_api.read_namespaced_ingress(
+                            name=ing_name, namespace=self.k8s_namespace
+                        )
+                        ingress.metadata.resource_version = (
+                            existing.metadata.resource_version
+                        )
+                        self.networking_api.replace_namespaced_ingress(
+                            name=ing_name,
+                            namespace=self.k8s_namespace,
+                            body=ingress,
+                        )
+                        print(f"Updated Ingress {ing_name}")
+                    else:
+                        raise
+        else:
+            if opts.o.debug:
+                print("No ingress configured")
+
+    def _create_nodeports(self):
+        """Create or update NodePort services."""
+        nodeports: List[client.V1Service] = self.cluster_info.get_nodeports()
+        for nodeport in nodeports:
+            if opts.o.debug:
+                print(f"Sending this nodeport: {nodeport}")
+            if not opts.o.dry_run:
+                np_name = nodeport.metadata.name
+                try:
+                    self.core_api.create_namespaced_service(
+                        namespace=self.k8s_namespace, body=nodeport
+                    )
+                except ApiException as e:
+                    if e.status == 409:
+                        existing = self.core_api.read_namespaced_service(
+                            name=np_name, namespace=self.k8s_namespace
+                        )
+                        nodeport.metadata.resource_version = (
+                            existing.metadata.resource_version
+                        )
+                        nodeport.spec.cluster_ip = existing.spec.cluster_ip
+                        self.core_api.replace_namespaced_service(
+                            name=np_name,
+                            namespace=self.k8s_namespace,
+                            body=nodeport,
+                        )
+                    else:
+                        raise
+
+    def up(self, detach, skip_cluster_management, services, image_overrides=None):
+        # Merge spec-level image overrides with CLI overrides
+        spec_overrides = self.cluster_info.spec.get("image-overrides", {})
+        if spec_overrides:
+            if image_overrides:
+                spec_overrides.update(image_overrides)  # CLI wins
+            image_overrides = spec_overrides
+        self.image_overrides = image_overrides
         self.skip_cluster_management = skip_cluster_management
         if not opts.o.dry_run:
-            if self.is_kind() and not self.skip_cluster_management:
-                # Create the kind cluster (or reuse existing one)
-                kind_config = str(
-                    self.deployment_dir.joinpath(constants.kind_config_filename)
-                )
-                actual_cluster = create_cluster(self.kind_cluster_name, kind_config)
-                if actual_cluster != self.kind_cluster_name:
-                    # An existing cluster was found, use it instead
-                    self.kind_cluster_name = actual_cluster
-                # Only load locally-built images into kind
-                # Registry images (docker.io, ghcr.io, etc.) will be pulled by k8s
-                local_containers = self.deployment_context.stack.obj.get(
-                    "containers", []
-                )
-                if local_containers:
-                    # Filter image_set to only images matching local containers
-                    local_images = {
-                        img
-                        for img in self.cluster_info.image_set
-                        if any(c in img for c in local_containers)
-                    }
-                    if local_images:
-                        load_images_into_kind(self.kind_cluster_name, local_images)
-                # Note: if no local containers defined, all images come from registries
-            self.connect_api()
-            # Create deployment-specific namespace for resource isolation
-            self._ensure_namespace()
-            if self.is_kind() and not self.skip_cluster_management:
-                # Configure ingress controller (not installed by default in kind)
-                # Skip if already running (idempotent for shared cluster)
-                if not is_ingress_running():
-                    install_ingress_for_kind(self.cluster_info.spec.get_acme_email())
-                    # Wait for ingress to start
-                    # (deployment provisioning will fail unless this is done)
-                    wait_for_ingress_in_kind()
-                # Create RuntimeClass if unlimited_memlock is enabled
-                if self.cluster_info.spec.get_unlimited_memlock():
-                    _create_runtime_class(
-                        constants.high_memlock_runtime,
-                        constants.high_memlock_runtime,
-                    )
-
+            self._setup_cluster()
         else:
             print("Dry run mode enabled, skipping k8s API connect")
 
         # Create registry secret if configured
         from stack_orchestrator.deploy.deployment_create import create_registry_secret
 
-        create_registry_secret(self.cluster_info.spec, self.cluster_info.app_name)
+        create_registry_secret(
+            self.cluster_info.spec, self.cluster_info.app_name, self.k8s_namespace
+        )
 
         self._create_volume_data()
+        self._create_external_services()
+        self._create_ca_certificates()
         self._create_deployment()
+        self._create_jobs()
+        self._create_ingress()
+        self._create_nodeports()
 
-        http_proxy_info = self.cluster_info.spec.get_http_proxy()
-        # Note: we don't support tls for kind (enabling tls causes errors)
-        use_tls = http_proxy_info and not self.is_kind()
-        certificate = (
-            self._find_certificate_for_host_name(http_proxy_info[0]["host-name"])
-            if use_tls
-            else None
-        )
-        if opts.o.debug:
-            if certificate:
-                print(f"Using existing certificate: {certificate}")
+        # Call start() hooks — stacks can create additional k8s resources
+        if self.deployment_context:
+            from stack_orchestrator.deploy.deployment_create import (
+                call_stack_deploy_start,
+            )
 
-        ingress = self.cluster_info.get_ingress(
-            use_tls=use_tls, certificate=certificate
-        )
-        if ingress:
-            if opts.o.debug:
-                print(f"Sending this ingress: {ingress}")
-            if not opts.o.dry_run:
-                ingress_resp = self.networking_api.create_namespaced_ingress(
-                    namespace=self.k8s_namespace, body=ingress
-                )
-                if opts.o.debug:
-                    print("Ingress created:")
-                    print(f"{ingress_resp}")
-        else:
-            if opts.o.debug:
-                print("No ingress configured")
-
-        nodeports: List[client.V1Service] = self.cluster_info.get_nodeports()
-        for nodeport in nodeports:
-            if opts.o.debug:
-                print(f"Sending this nodeport: {nodeport}")
-            if not opts.o.dry_run:
-                nodeport_resp = self.core_api.create_namespaced_service(
-                    namespace=self.k8s_namespace, body=nodeport
-                )
-                if opts.o.debug:
-                    print("NodePort created:")
-                    print(f"{nodeport_resp}")
+            call_stack_deploy_start(self.deployment_context)
 
     def down(self, timeout, volumes, skip_cluster_management):
         self.skip_cluster_management = skip_cluster_management
         self.connect_api()
 
+        app_label = f"app={self.cluster_info.app_name}"
+
         # PersistentVolumes are cluster-scoped (not namespaced), so delete by label
         if volumes:
             try:
-                pvs = self.core_api.list_persistent_volume(
-                    label_selector=f"app={self.cluster_info.app_name}"
-                )
+                pvs = self.core_api.list_persistent_volume(label_selector=app_label)
                 for pv in pvs.items:
                     if opts.o.debug:
                         print(f"Deleting PV: {pv.metadata.name}")
@@ -447,9 +874,14 @@ class K8sDeployer(Deployer):
                 if opts.o.debug:
                     print(f"Error listing PVs: {e}")
 
-        # Delete the deployment namespace - this cascades to all namespaced resources
-        # (PVCs, ConfigMaps, Deployments, Services, Ingresses, etc.)
+        # Delete the namespace to ensure clean slate.
+        # Resources created by older laconic-so versions lack labels, so
+        # label-based deletion can't find them. Namespace deletion is the
+        # only reliable cleanup.
         self._delete_namespace()
+        # Wait for namespace to finish terminating before returning,
+        # so that up() can recreate it immediately.
+        self._wait_for_namespace_gone()
 
         if self.is_kind() and not self.skip_cluster_management:
             # Destroy the kind cluster
@@ -574,14 +1006,18 @@ class K8sDeployer(Deployer):
 
     def logs(self, services, tail, follow, stream):
         self.connect_api()
-        pods = pods_in_deployment(self.core_api, self.cluster_info.app_name)
+        pods = pods_in_deployment(
+            self.core_api, self.cluster_info.app_name, namespace=self.k8s_namespace
+        )
         if len(pods) > 1:
             print("Warning: more than one pod in the deployment")
         if len(pods) == 0:
             log_data = "******* Pods not running ********\n"
         else:
             k8s_pod_name = pods[0]
-            containers = containers_in_pod(self.core_api, k8s_pod_name)
+            containers = containers_in_pod(
+                self.core_api, k8s_pod_name, namespace=self.k8s_namespace
+            )
             # If pod not started, logs request below will throw an exception
             try:
                 log_data = ""
@@ -599,49 +1035,54 @@ class K8sDeployer(Deployer):
         return log_stream_from_string(log_data)
 
     def update(self):
+        if not self.cluster_info.parsed_pod_yaml_map:
+            if opts.o.debug:
+                print("No pods defined, skipping update")
+            return
         self.connect_api()
-        ref_deployment = self.cluster_info.get_deployment()
-        if not ref_deployment or not ref_deployment.metadata:
-            return
-        ref_name = ref_deployment.metadata.name
-        if not ref_name:
-            return
+        ref_deployments = self.cluster_info.get_deployments()
+        for ref_deployment in ref_deployments:
+            if not ref_deployment or not ref_deployment.metadata:
+                continue
+            ref_name = ref_deployment.metadata.name
+            if not ref_name:
+                continue
 
-        deployment = cast(
-            client.V1Deployment,
-            self.apps_api.read_namespaced_deployment(
-                name=ref_name, namespace=self.k8s_namespace
-            ),
-        )
-        if not deployment.spec or not deployment.spec.template:
-            return
-        template_spec = deployment.spec.template.spec
-        if not template_spec or not template_spec.containers:
-            return
+            deployment = cast(
+                client.V1Deployment,
+                self.apps_api.read_namespaced_deployment(
+                    name=ref_name, namespace=self.k8s_namespace
+                ),
+            )
+            if not deployment.spec or not deployment.spec.template:
+                continue
+            template_spec = deployment.spec.template.spec
+            if not template_spec or not template_spec.containers:
+                continue
 
-        ref_spec = ref_deployment.spec
-        if ref_spec and ref_spec.template and ref_spec.template.spec:
-            ref_containers = ref_spec.template.spec.containers
-            if ref_containers:
-                new_env = ref_containers[0].env
-                for container in template_spec.containers:
-                    old_env = container.env
-                    if old_env != new_env:
-                        container.env = new_env
+            ref_spec = ref_deployment.spec
+            if ref_spec and ref_spec.template and ref_spec.template.spec:
+                ref_containers = ref_spec.template.spec.containers
+                if ref_containers:
+                    new_env = ref_containers[0].env
+                    for container in template_spec.containers:
+                        old_env = container.env
+                        if old_env != new_env:
+                            container.env = new_env
 
-        template_meta = deployment.spec.template.metadata
-        if template_meta:
-            template_meta.annotations = {
-                "kubectl.kubernetes.io/restartedAt": datetime.utcnow()
-                .replace(tzinfo=timezone.utc)
-                .isoformat()
-            }
+            template_meta = deployment.spec.template.metadata
+            if template_meta:
+                template_meta.annotations = {
+                    "kubectl.kubernetes.io/restartedAt": datetime.utcnow()
+                    .replace(tzinfo=timezone.utc)
+                    .isoformat()
+                }
 
-        self.apps_api.patch_namespaced_deployment(
-            name=ref_name,
-            namespace=self.k8s_namespace,
-            body=deployment,
-        )
+            self.apps_api.patch_namespaced_deployment(
+                name=ref_name,
+                namespace=self.k8s_namespace,
+                body=deployment,
+            )
 
     def run(
         self,
@@ -659,26 +1100,41 @@ class K8sDeployer(Deployer):
 
     def run_job(self, job_name: str, helm_release: Optional[str] = None):
         if not opts.o.dry_run:
-            from stack_orchestrator.deploy.k8s.helm.job_runner import run_helm_job
-
             # Check if this is a helm-based deployment
             chart_dir = self.deployment_dir / "chart"
-            if not chart_dir.exists():
-                # TODO: Implement job support for compose-based K8s deployments
-                raise Exception(
-                    f"Job support is only available for helm-based "
-                    f"deployments. Chart directory not found: {chart_dir}"
-                )
+            if chart_dir.exists():
+                from stack_orchestrator.deploy.k8s.helm.job_runner import run_helm_job
 
-            # Run the job using the helm job runner
-            run_helm_job(
-                chart_dir=chart_dir,
-                job_name=job_name,
-                release=helm_release,
-                namespace=self.k8s_namespace,
-                timeout=600,
-                verbose=opts.o.verbose,
-            )
+                # Run the job using the helm job runner
+                run_helm_job(
+                    chart_dir=chart_dir,
+                    job_name=job_name,
+                    release=helm_release,
+                    namespace=self.k8s_namespace,
+                    timeout=600,
+                    verbose=opts.o.verbose,
+                )
+            else:
+                # Non-Helm path: create job from ClusterInfo
+                self.connect_api()
+                jobs = self.cluster_info.get_jobs(image_pull_policy="Always")
+                # Find the matching job by name
+                target_name = f"{self.cluster_info.app_name}-job-{job_name}"
+                matched_job = None
+                for job in jobs:
+                    if job.metadata and job.metadata.name == target_name:
+                        matched_job = job
+                        break
+                if matched_job is None:
+                    raise Exception(
+                        f"Job '{job_name}' not found. Available jobs: "
+                        f"{[j.metadata.name for j in jobs if j.metadata]}"
+                    )
+                if opts.o.debug:
+                    print(f"Creating job: {target_name}")
+                self.batch_api.create_namespaced_job(
+                    body=matched_job, namespace=self.k8s_namespace
+                )
 
     def is_kind(self):
         return self.type == "k8s-kind"

stack_orchestrator/deploy/k8s/helpers.py

@@ -393,10 +393,12 @@ def load_images_into_kind(kind_cluster_name: str, image_set: Set[str]):
             raise DeployerException(f"kind load docker-image failed: {result}")
 
 
-def pods_in_deployment(core_api: client.CoreV1Api, deployment_name: str):
+def pods_in_deployment(
+    core_api: client.CoreV1Api, deployment_name: str, namespace: str = "default"
+):
     pods = []
     pod_response = core_api.list_namespaced_pod(
-        namespace="default", label_selector=f"app={deployment_name}"
+        namespace=namespace, label_selector=f"app={deployment_name}"
     )
     if opts.o.debug:
         print(f"pod_response: {pod_response}")
@@ -406,10 +408,12 @@ def pods_in_deployment(core_api: client.CoreV1Api, deployment_name: str):
     return pods
 
 
-def containers_in_pod(core_api: client.CoreV1Api, pod_name: str) -> List[str]:
+def containers_in_pod(
+    core_api: client.CoreV1Api, pod_name: str, namespace: str = "default"
+) -> List[str]:
     containers: List[str] = []
     pod_response = cast(
-        client.V1Pod, core_api.read_namespaced_pod(pod_name, namespace="default")
+        client.V1Pod, core_api.read_namespaced_pod(pod_name, namespace=namespace)
     )
     if opts.o.debug:
         print(f"pod_response: {pod_response}")
@@ -440,7 +444,20 @@ def named_volumes_from_pod_files(parsed_pod_files):
     return named_volumes
 
 
-def get_kind_pv_bind_mount_path(volume_name: str):
+def get_kind_pv_bind_mount_path(
+    volume_name: str,
+    kind_mount_root: Optional[str] = None,
+    host_path: Optional[str] = None,
+):
+    """Get the path inside the Kind node for a PV.
+
+    When kind-mount-root is set and the volume's host path is under
+    that root, return /mnt/{relative_path} so it resolves through the
+    single root extraMount. Otherwise fall back to /mnt/{volume_name}.
+    """
+    if kind_mount_root and host_path and host_path.startswith(kind_mount_root):
+        rel = os.path.relpath(host_path, kind_mount_root)
+        return f"/mnt/{rel}"
     return f"/mnt/{volume_name}"
 
 
@@ -563,6 +580,7 @@ def _generate_kind_mounts(parsed_pod_files, deployment_dir, deployment_context):
     volume_definitions = []
     volume_host_path_map = _get_host_paths_for_volumes(deployment_context)
     seen_host_path_mounts = set()  # Track to avoid duplicate mounts
+    kind_mount_root = deployment_context.spec.get_kind_mount_root()
 
     # Cluster state backup for offline data recovery (unique per deployment)
     # etcd contains all k8s state; PKI certs needed to decrypt etcd offline
@@ -583,6 +601,16 @@ def _generate_kind_mounts(parsed_pod_files, deployment_dir, deployment_context):
         f"  - hostPath: {pki_host_path}\n" f"    containerPath: /etc/kubernetes/pki\n"
     )
 
+    # When kind-mount-root is set, emit a single extraMount for the root.
+    # Individual volumes whose host path starts with the root are covered
+    # by this single mount and don't need their own extraMount entries.
+    mount_root_emitted = False
+    if kind_mount_root:
+        volume_definitions.append(
+            f"  - hostPath: {kind_mount_root}\n" f"    containerPath: /mnt\n"
+        )
+        mount_root_emitted = True
+
     # Note these paths are relative to the location of the pod files (at present)
     # So we need to fix up to make them correct and absolute because kind assumes
     # relative to the cwd.
@@ -642,6 +670,12 @@ def _generate_kind_mounts(parsed_pod_files, deployment_dir, deployment_context):
                                         volume_host_path_map[volume_name],
                                         deployment_dir,
                                     )
+                                    # Skip individual extraMount if covered
+                                    # by the kind-mount-root single mount
+                                    if mount_root_emitted and str(host_path).startswith(
+                                        kind_mount_root
+                                    ):
+                                        continue
                                     container_path = get_kind_pv_bind_mount_path(
                                         volume_name
                                     )
@@ -978,7 +1012,7 @@ def translate_sidecar_service_names(
 
 
 def envs_from_environment_variables_map(
-    map: Mapping[str, str]
+    map: Mapping[str, str],
 ) -> List[client.V1EnvVar]:
     result = []
     for env_var, env_val in map.items():

stack_orchestrator/deploy/spec.py

@@ -98,16 +98,17 @@ class Spec:
     def get_image_registry(self):
         return self.obj.get(constants.image_registry_key)
 
+    def get_credentials_files(self) -> typing.List[str]:
+        """Returns list of credential file paths to append to config.env."""
+        return self.obj.get("credentials-files", [])
+
     def get_image_registry_config(self) -> typing.Optional[typing.Dict]:
         """Returns registry auth config: {server, username, token-env}.
 
         Used for private container registries like GHCR. The token-env field
         specifies an environment variable containing the API token/PAT.
-
-        Note: Uses 'registry-credentials' key to avoid collision with
-        'image-registry' key which is for pushing images.
         """
-        return self.obj.get("registry-credentials")
+        return self.obj.get("image-pull-secret")
 
     def get_volumes(self):
         return self.obj.get(constants.volumes_key, {})
@@ -115,6 +116,9 @@ class Spec:
     def get_configmaps(self):
         return self.obj.get(constants.configmaps_key, {})
 
+    def get_secrets(self):
+        return self.obj.get(constants.secrets_key, {})
+
     def get_container_resources(self):
         return Resources(
             self.obj.get(constants.resources_key, {}).get("containers", {})
@@ -146,9 +150,46 @@ class Spec:
             self.obj.get(constants.resources_key, {}).get(constants.volumes_key, {})
         )
 
+    def get_volume_resources_for(self, volume_name: str) -> typing.Optional[Resources]:
+        """Look up per-volume resource overrides from spec.yml.
+
+        Supports two formats under resources.volumes:
+
+        Global (original):
+            resources:
+              volumes:
+                reservations:
+                  storage: 5Gi
+
+        Per-volume (new):
+            resources:
+              volumes:
+                my-volume:
+                  reservations:
+                    storage: 10Gi
+
+        Returns the per-volume Resources if found, otherwise None.
+        The caller should fall back to get_volume_resources() then the default.
+        """
+        vol_section = self.obj.get(constants.resources_key, {}).get(
+            constants.volumes_key, {}
+        )
+        if volume_name not in vol_section:
+            return None
+        entry = vol_section[volume_name]
+        if isinstance(entry, dict) and ("reservations" in entry or "limits" in entry):
+            return Resources(entry)
+        return None
+
     def get_http_proxy(self):
         return self.obj.get(constants.network_key, {}).get(constants.http_proxy_key, [])
 
+    def get_namespace(self):
+        return self.obj.get("namespace")
+
+    def get_kind_cluster_name(self):
+        return self.obj.get("kind-cluster-name")
+
     def get_annotations(self):
         return self.obj.get(constants.annotations_key, {})
 
@@ -223,5 +264,46 @@ class Spec:
     def is_kind_deployment(self):
         return self.get_deployment_type() in [constants.k8s_kind_deploy_type]
 
+    def get_kind_mount_root(self) -> typing.Optional[str]:
+        """Return kind-mount-root path or None.
+
+        When set, laconic-so emits a single Kind extraMount mapping this
+        host path to /mnt inside the Kind node. Volumes with host paths
+        under this root resolve to /mnt/{relative_path} and don't need
+        individual extraMounts. This allows adding new volumes without
+        recreating the Kind cluster.
+        """
+        return self.obj.get(constants.kind_mount_root_key)
+
+    def get_maintenance_service(self) -> typing.Optional[str]:
+        """Return maintenance-service value (e.g. 'dumpster-maintenance:8000') or None.
+
+        When set, the restart command swaps Ingress backends to this service
+        during the main pod Recreate, so users see a branded maintenance page
+        instead of a bare 502.
+        """
+        return self.obj.get("maintenance-service")
+
+    def get_external_services(self) -> typing.Dict[str, typing.Dict]:
+        """Return external-services config from spec.
+
+        Each entry maps a service name to its routing config:
+        - host mode: {host: "example.com", port: 443}
+          → ExternalName k8s Service (DNS CNAME)
+        - selector mode: {selector: {app: "foo"}, namespace: "ns", port: 443}
+          → Headless Service + Endpoints (cross-namespace routing to mock pod)
+        """
+        return self.obj.get(constants.external_services_key, {})
+
+    def get_ca_certificates(self) -> typing.List[str]:
+        """Return list of CA certificate file paths to trust.
+
+        Used in testing specs to inject mkcert root CAs so containers
+        trust TLS certs on mock services. Files are mounted into all
+        containers at /etc/ssl/certs/ and NODE_EXTRA_CA_CERTS is set.
+        Production specs omit this key entirely.
+        """
+        return self.obj.get(constants.ca_certificates_key, [])
+
     def is_docker_deployment(self):
         return self.get_deployment_type() in [constants.compose_deploy_type]

stack_orchestrator/deploy/webapp/deploy_webapp.py

@@ -19,7 +19,7 @@ from pathlib import Path
 from urllib.parse import urlparse
 from tempfile import NamedTemporaryFile
 
-from stack_orchestrator.util import error_exit, global_options2
+from stack_orchestrator.util import error_exit, global_options2, get_yaml
 from stack_orchestrator.deploy.deployment_create import init_operation, create_operation
 from stack_orchestrator.deploy.deploy import create_deploy_context
 from stack_orchestrator.deploy.deploy_types import DeployCommandContext
@@ -41,19 +41,23 @@ def _fixup_container_tag(deployment_dir: str, image: str):
 def _fixup_url_spec(spec_file_name: str, url: str):
     # url is like: https://example.com/path
     parsed_url = urlparse(url)
-    http_proxy_spec = f"""
-  http-proxy:
-    - host-name: {parsed_url.hostname}
-      routes:
-        - path: '{parsed_url.path if parsed_url.path else "/"}'
-          proxy-to: webapp:80
-    """
     spec_file_path = Path(spec_file_name)
+    yaml = get_yaml()
     with open(spec_file_path) as rfile:
-        contents = rfile.read()
-        contents = contents + http_proxy_spec
+        contents = yaml.load(rfile)
+    contents.setdefault("network", {})["http-proxy"] = [
+        {
+            "host-name": parsed_url.hostname,
+            "routes": [
+                {
+                    "path": parsed_url.path if parsed_url.path else "/",
+                    "proxy-to": "webapp:80",
+                }
+            ],
+        }
+    ]
     with open(spec_file_path, "w") as wfile:
-        wfile.write(contents)
+        yaml.dump(contents, wfile)
 
 
 def create_deployment(

stack_orchestrator/util.py

@@ -75,6 +75,8 @@ def get_parsed_stack_config(stack):
 
 def get_pod_list(parsed_stack):
     # Handle both old and new format
+    if "pods" not in parsed_stack or not parsed_stack["pods"]:
+        return []
     pods = parsed_stack["pods"]
     if type(pods[0]) is str:
         result = pods
@@ -103,7 +105,7 @@ def get_job_list(parsed_stack):
 
 def get_plugin_code_paths(stack) -> List[Path]:
     parsed_stack = get_parsed_stack_config(stack)
-    pods = parsed_stack["pods"]
+    pods = parsed_stack.get("pods") or []
     result: Set[Path] = set()
     for pod in pods:
         if type(pod) is str:
@@ -153,15 +155,16 @@ def resolve_job_compose_file(stack, job_name: str):
         if proposed_file.exists():
             return proposed_file
         # If we don't find it fall through to the internal case
-    # TODO: Add internal compose-jobs directory support if needed
-    # For now, jobs are expected to be in external stacks only
-    compose_jobs_base = Path(stack).parent.parent.joinpath("compose-jobs")
+    data_dir = Path(__file__).absolute().parent.joinpath("data")
+    compose_jobs_base = data_dir.joinpath("compose-jobs")
     return compose_jobs_base.joinpath(f"docker-compose-{job_name}.yml")
 
 
 def get_pod_file_path(stack, parsed_stack, pod_name: str):
-    pods = parsed_stack["pods"]
+    pods = parsed_stack.get("pods") or []
     result = None
+    if not pods:
+        return result
     if type(pods[0]) is str:
         result = resolve_compose_file(stack, pod_name)
     else:
@@ -189,9 +192,9 @@ def get_job_file_path(stack, parsed_stack, job_name: str):
 
 
 def get_pod_script_paths(parsed_stack, pod_name: str):
-    pods = parsed_stack["pods"]
+    pods = parsed_stack.get("pods") or []
     result = []
-    if not type(pods[0]) is str:
+    if not pods or not type(pods[0]) is str:
         for pod in pods:
             if pod["name"] == pod_name:
                 pod_root_dir = os.path.join(
@@ -207,9 +210,9 @@ def get_pod_script_paths(parsed_stack, pod_name: str):
 
 
 def pod_has_scripts(parsed_stack, pod_name: str):
-    pods = parsed_stack["pods"]
+    pods = parsed_stack.get("pods") or []
     result = False
-    if type(pods[0]) is str:
+    if not pods or type(pods[0]) is str:
         result = False
     else:
         for pod in pods:

tests/deploy/run-deploy-test.sh

@@ -141,28 +141,35 @@ echo "$test_config_file_changed_content" > "$test_config_file"
 test_unchanged_config="$test_deployment_dir/config/test/script.sh"
 
 # Modify spec file to simulate an update
-sed -i.bak 's/CERC_TEST_PARAM_3:/CERC_TEST_PARAM_3: FASTER/' $test_deployment_spec
+sed -i.bak 's/CERC_TEST_PARAM_3: FAST/CERC_TEST_PARAM_3: FASTER/' $test_deployment_spec
 
-# Create/modify config.env to test it isn't overwritten during sync
+# Save config.env before update (to verify it gets backed up)
 config_env_file="$test_deployment_dir/config.env"
-config_env_persistent_content="PERSISTENT_VALUE=should-not-be-overwritten-$(date +%s)"
-echo "$config_env_persistent_content" >> "$config_env_file"
 original_config_env_content=$(<$config_env_file)
 
 # Run sync to update deployment files without destroying data
 $TEST_TARGET_SO --stack test deploy create --spec-file $test_deployment_spec --deployment-dir $test_deployment_dir --update
 
-# Verify config.env was not overwritten
+# Verify config.env was regenerated from spec (reflects the FASTER change)
 synced_config_env_content=$(<$config_env_file)
-if [ "$synced_config_env_content" == "$original_config_env_content" ]; then
-    echo "deployment update test: config.env preserved - passed"
+if [[ "$synced_config_env_content" == *"CERC_TEST_PARAM_3=FASTER"* ]]; then
+    echo "deployment update test: config.env regenerated from spec - passed"
 else
-    echo "deployment update test: config.env was overwritten - FAILED"
-    echo "Expected: $original_config_env_content"
+    echo "deployment update test: config.env not regenerated - FAILED"
+    echo "Expected CERC_TEST_PARAM_3=FASTER in config.env"
     echo "Got: $synced_config_env_content"
     exit 1
 fi
 
+# Verify old config.env was backed up
+config_env_backup="${config_env_file}.bak"
+if [ -f "$config_env_backup" ]; then
+    echo "deployment update test: config.env backed up - passed"
+else
+    echo "deployment update test: config.env backup not created - FAILED"
+    exit 1
+fi
+
 # Verify the spec file was updated in deployment dir
 updated_deployed_spec=$(<$test_deployment_dir/spec.yml)
 if [[ "$updated_deployed_spec" == *"FASTER"* ]]; then

tests/k8s-deploy/run-deploy-test.sh

@@ -105,6 +105,15 @@ fi
 # Add a config file to be picked up by the ConfigMap before starting.
 echo "dbfc7a4d-44a7-416d-b5f3-29842cc47650" > $test_deployment_dir/configmaps/test-config/test_config
 
+# Add secrets to the deployment spec (references a pre-existing k8s Secret by name).
+# deploy init already writes an empty 'secrets: {}' key, so we replace it
+# rather than appending (ruamel.yaml rejects duplicate keys).
+deployment_spec_file=${test_deployment_dir}/spec.yml
+sed -i 's/^secrets: {}$/secrets:\n  test-secret:\n    - TEST_SECRET_KEY/' ${deployment_spec_file}
+
+# Get the deployment ID for kubectl queries
+deployment_id=$(cat ${test_deployment_dir}/deployment.yml | cut -d ' ' -f 2)
+
 echo "deploy create output file test: passed"
 # Try to start the deployment
 $TEST_TARGET_SO deployment --dir $test_deployment_dir start
@@ -166,12 +175,71 @@ else
     delete_cluster_exit
 fi
 
-# Stop then start again and check the volume was preserved
-$TEST_TARGET_SO deployment --dir $test_deployment_dir stop
-# Sleep a bit just in case
-# sleep for longer to check if that's why the subsequent create cluster fails
-sleep 20
-$TEST_TARGET_SO deployment --dir $test_deployment_dir start
+# --- New feature tests: namespace, labels, jobs, secrets ---
+
+# Check that the pod is in the deployment-specific namespace (not default)
+ns_pod_count=$(kubectl get pods -n laconic-${deployment_id} -l app=${deployment_id} --no-headers 2>/dev/null | wc -l)
+if [ "$ns_pod_count" -gt 0 ]; then
+    echo "namespace isolation test: passed"
+else
+    echo "namespace isolation test: FAILED"
+    echo "Expected pod in namespace laconic-${deployment_id}"
+    delete_cluster_exit
+fi
+
+# Check that the stack label is set on the pod
+stack_label_count=$(kubectl get pods -n laconic-${deployment_id} -l app.kubernetes.io/stack=test --no-headers 2>/dev/null | wc -l)
+if [ "$stack_label_count" -gt 0 ]; then
+    echo "stack label test: passed"
+else
+    echo "stack label test: FAILED"
+    delete_cluster_exit
+fi
+
+# Check that the job completed successfully
+for i in {1..30}; do
+    job_status=$(kubectl get job ${deployment_id}-job-test-job -n laconic-${deployment_id} -o jsonpath='{.status.succeeded}' 2>/dev/null || true)
+    if [ "$job_status" == "1" ]; then
+        break
+    fi
+    sleep 2
+done
+if [ "$job_status" == "1" ]; then
+    echo "job completion test: passed"
+else
+    echo "job completion test: FAILED"
+    echo "Job status.succeeded: ${job_status}"
+    delete_cluster_exit
+fi
+
+# Check that the secrets spec results in an envFrom secretRef on the pod
+secret_ref=$(kubectl get pod -n laconic-${deployment_id} -l app=${deployment_id} \
+    -o jsonpath='{.items[0].spec.containers[0].envFrom[?(@.secretRef.name=="test-secret")].secretRef.name}' 2>/dev/null || true)
+if [ "$secret_ref" == "test-secret" ]; then
+    echo "secrets envFrom test: passed"
+else
+    echo "secrets envFrom test: FAILED"
+    echo "Expected secretRef 'test-secret', got: ${secret_ref}"
+    delete_cluster_exit
+fi
+
+# Stop then start again and check the volume was preserved.
+# Use --skip-cluster-management to reuse the existing kind cluster instead of
+# destroying and recreating it (which fails on CI runners due to stale etcd/certs
+# and cgroup detection issues).
+# Use --delete-volumes to clear PVs so fresh PVCs can bind on restart.
+# Bind-mount data survives on the host filesystem; provisioner volumes are recreated fresh.
+$TEST_TARGET_SO deployment --dir $test_deployment_dir stop --delete-volumes --skip-cluster-management
+# Wait for the namespace to be fully terminated before restarting.
+# Without this, 'start' fails with 403 Forbidden because the namespace
+# is still in Terminating state.
+for i in {1..60}; do
+    if ! kubectl get namespace laconic-${deployment_id} 2>/dev/null | grep -q .; then
+        break
+    fi
+    sleep 2
+done
+$TEST_TARGET_SO deployment --dir $test_deployment_dir start --skip-cluster-management
 wait_for_pods_started
 wait_for_log_output
 sleep 1
@@ -184,8 +252,9 @@ else
     delete_cluster_exit
 fi
 
-# These volumes will be completely destroyed by the kind delete/create, because they lived inside
-# the kind container.  So, unlike the bind-mount case, they will appear fresh after the restart.
+# Provisioner volumes are destroyed when PVs are deleted (--delete-volumes on stop).
+# Unlike bind-mount volumes whose data persists on the host, provisioner storage
+# is gone, so the volume appears fresh after restart.
 log_output_11=$( $TEST_TARGET_SO deployment --dir $test_deployment_dir logs )
 if [[ "$log_output_11" == *"/data2 filesystem is fresh"* ]]; then
     echo "Fresh provisioner volumes test: passed"

tests/k8s-deployment-control/run-test.sh

@@ -206,7 +206,7 @@ fi
 # The deployment's pod should be scheduled onto node: worker3
 # Check that's what happened
 # Get get the node onto which the stack pod has been deployed
-deployment_node=$(kubectl get pods -l app=${deployment_id} -o=jsonpath='{.items..spec.nodeName}')
+deployment_node=$(kubectl get pods -n laconic-${deployment_id} -l app=${deployment_id} -o=jsonpath='{.items..spec.nodeName}')
 expected_node=${deployment_id}-worker3
 echo "Stack pod deployed to node: ${deployment_node}"
 if [[ ${deployment_node} == ${expected_node} ]]; then

tests/scripts/install-kind.sh

@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
 # TODO: handle ARM
-curl --silent -Lo ./kind https://kind.sigs.k8s.io/dl/v0.20.0/kind-linux-amd64
+curl --silent -Lo ./kind https://kind.sigs.k8s.io/dl/v0.25.0/kind-linux-amd64
 chmod +x ./kind
 mv ./kind /usr/local/bin

tests/scripts/install-kubectl.sh

@@ -1,5 +1,6 @@
 #!/usr/bin/env bash
 # TODO: handle ARM
-curl --silent -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
+# Pin kubectl to match Kind's default k8s version (v1.31.x)
+curl --silent -LO "https://dl.k8s.io/release/v1.31.2/bin/linux/amd64/kubectl"
 chmod +x ./kubectl
 mv ./kubectl /usr/local/bin