hermes: add daemonless agent image release lane

ci/Jenkinsfile.hermes-agent-image

@@ -147,10 +147,37 @@ spec:
           test "${actual_revision}" = "$(git rev-parse origin/main)"
           test -z "$(git status --porcelain)"
           test -f dockerfiles/Dockerfile.hermes-agent
-          printf '%s\n' "${HERMES_IMAGE}:git-${actual_revision}" > build/hermes-agent.destination
+          case "${BUILD_NUMBER}" in
+            ''|0*|*[!0-9]*)
+              echo "BUILD_NUMBER must be a positive decimal integer" >&2
+              exit 2
+              ;;
+          esac
+          printf '%s\n' \
+            "${HERMES_IMAGE}:git-${actual_revision}-build-${BUILD_NUMBER}" \
+            > build/hermes-agent.destination
         '''
       }
     }
+    stage('Reject replay before publish') {
+      steps {
+        withCredentials([usernamePassword(
+          credentialsId: 'harbor-robot',
+          usernameVariable: 'HARBOR_USER',
+          passwordVariable: 'HARBOR_PASSWORD'
+        )]) {
+          sh '''
+            set -eu
+            set +x
+            destination="$(cat build/hermes-agent.destination)"
+            python3 ci/scripts/hermes_image_release.py assert-absent \
+              --source-revision "${EXPECTED_SOURCE_REVISION}" \
+              --build-number "${BUILD_NUMBER}" \
+              --destination "${destination}"
+          '''
+        }
+      }
+    }
     stage('Build and publish without a daemon') {
       steps {
         container('kaniko') {
@@ -175,7 +202,7 @@ spec:
                 --dockerfile="${WORKSPACE}/dockerfiles/Dockerfile.hermes-agent" \
                 --destination="${destination}" \
                 --digest-file="${WORKSPACE}/build/hermes-agent.digest" \
-                --image-name-with-digest-file="${WORKSPACE}/build/hermes-agent.image" \
+                --image-name-tag-with-digest-file="${WORKSPACE}/build/hermes-agent.image" \
                 --label="org.opencontainers.image.revision=${EXPECTED_SOURCE_REVISION}" \
                 --cleanup \
                 --push-retry=3
@@ -186,18 +213,27 @@ spec:
     }
     stage('Render reviewed Flux handoff') {
       steps {
-        sh '''
-          set -eu
-          destination="$(cat build/hermes-agent.destination)"
-          python3 ci/scripts/hermes_image_release.py \
-            --digest-file build/hermes-agent.digest \
-            --source-revision "${EXPECTED_SOURCE_REVISION}" \
-            --destination "${destination}" \
-            --kustomization services/hermes/kustomization.yaml \
-            --output-dir build/hermes-agent-release
-          test -s build/hermes-agent-release/hermes-image-update.patch
-          test -s build/hermes-agent-release/hermes-agent-image.json
-        '''
+        withCredentials([usernamePassword(
+          credentialsId: 'harbor-robot',
+          usernameVariable: 'HARBOR_USER',
+          passwordVariable: 'HARBOR_PASSWORD'
+        )]) {
+          sh '''
+            set -eu
+            set +x
+            destination="$(cat build/hermes-agent.destination)"
+            python3 ci/scripts/hermes_image_release.py render \
+              --digest-file build/hermes-agent.digest \
+              --image-file build/hermes-agent.image \
+              --source-revision "${EXPECTED_SOURCE_REVISION}" \
+              --build-number "${BUILD_NUMBER}" \
+              --destination "${destination}" \
+              --kustomization services/hermes/kustomization.yaml \
+              --output-dir build/hermes-agent-release
+            test -s build/hermes-agent-release/hermes-image-update.patch
+            test -s build/hermes-agent-release/hermes-agent-image.json
+          '''
+        }
       }
     }
   }

ci/scripts/hermes_image_release.py

@@ -1,18 +1,39 @@
 #!/usr/bin/env python3
-"""Render reviewable Flux artifacts for a published Hermes agent image."""
+"""Verify and render a reviewable Hermes agent image release."""
 
 from __future__ import annotations
 
 import argparse
+import base64
 import difflib
 import json
+import os
 import re
+import urllib.error
+import urllib.parse
+import urllib.request
 from pathlib import Path
+from typing import Any, Callable
 
 
 DEFAULT_IMAGE = "registry.bstein.dev/bstein/hermes-agent"
+HARBOR_API_ORIGIN = "https://registry.bstein.dev/api/v2.0"
+HARBOR_PROJECT = "bstein"
+HARBOR_REPOSITORY = "hermes-agent"
 DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
 REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
+BUILD_PATTERN = re.compile(r"^[1-9][0-9]*$")
+DESTINATION_PATTERN = re.compile(
+    r"^registry\.bstein\.dev/bstein/hermes-agent:"
+    r"git-([0-9a-f]{40})-build-([1-9][0-9]*)$"
+)
+
+
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    """Never send registry credentials to a redirect target."""
+
+    def redirect_request(self, _request, _file, _code, _message, _headers, _url):
+        return None
 
 
 def _validated(value: str, pattern: re.Pattern[str], label: str) -> str:
@@ -23,6 +44,126 @@ def _validated(value: str, pattern: re.Pattern[str], label: str) -> str:
     return normalized
 
 
+def validate_destination(
+    destination: str, source_revision: str, build_number: str
+) -> tuple[str, str]:
+    """Bind one unique build tag to the reviewed revision and Jenkins build."""
+    revision = _validated(source_revision, REVISION_PATTERN, "source revision")
+    build = _validated(build_number, BUILD_PATTERN, "build number")
+    match = DESTINATION_PATTERN.fullmatch(destination.strip())
+    if not match or match.groups() != (revision, build):
+        raise ValueError(
+            "destination must bind the reviewed revision and unique Jenkins build"
+        )
+    return revision, build
+
+
+def validate_kaniko_evidence(
+    *, digest_text: str, image_text: str, destination: str
+) -> str:
+    """Cross-check both independent Kaniko output files against the destination."""
+    digest_lines = digest_text.splitlines()
+    image_lines = image_text.splitlines()
+    if len(digest_lines) != 1:
+        raise ValueError("Kaniko digest evidence must contain exactly one line")
+    if len(image_lines) != 1:
+        raise ValueError("Kaniko image evidence must contain exactly one line")
+    digest = _validated(digest_lines[0], DIGEST_PATTERN, "image digest")
+    if image_lines[0].strip() != f"{destination}@{digest}":
+        raise ValueError("Kaniko image evidence does not match destination and digest")
+    return digest
+
+
+def _registry_request(request: urllib.request.Request, timeout: int) -> Any:
+    """Make a registry request without following redirects."""
+    opener = urllib.request.build_opener(_NoRedirect())
+    try:
+        return opener.open(request, timeout=timeout)
+    except urllib.error.HTTPError as exc:
+        return exc
+
+
+def _artifact_response(
+    destination: str,
+    *,
+    username: str,
+    password: str,
+    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
+) -> tuple[int, bytes]:
+    """Read one exact Harbor artifact by tag with bounded response size."""
+    match = DESTINATION_PATTERN.fullmatch(destination)
+    if not match:
+        raise ValueError("invalid destination")
+    if not username or not password:
+        raise RuntimeError("Harbor credentials are empty")
+    tag = destination.rsplit(":", 1)[1]
+    encoded_tag = urllib.parse.quote(tag, safe="")
+    auth = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
+    request = urllib.request.Request(
+        f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/repositories/"
+        f"{HARBOR_REPOSITORY}/artifacts/{encoded_tag}",
+        headers={"Accept": "application/json", "Authorization": f"Basic {auth}"},
+        method="GET",
+    )
+    with opener(request, 20) as response:
+        body = response.read(1_048_577)
+        if len(body) > 1_048_576:
+            raise RuntimeError("Harbor artifact response exceeded the size limit")
+        return int(response.status), body
+
+
+def assert_tag_absent(
+    destination: str,
+    *,
+    username: str,
+    password: str,
+    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
+) -> None:
+    """Reject replay before Kaniko can push an already-used immutable identity."""
+    status, _body = _artifact_response(
+        destination, username=username, password=password, opener=opener
+    )
+    if status == 404:
+        return
+    if status == 200:
+        raise RuntimeError("Harbor destination tag already exists; refusing overwrite")
+    raise RuntimeError(f"Harbor destination preflight returned HTTP {status}")
+
+
+def verify_registry_digest(
+    destination: str,
+    digest: str,
+    *,
+    username: str,
+    password: str,
+    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
+) -> None:
+    """Verify Harbor independently resolves the pushed tag to Kaniko's digest."""
+    digest = _validated(digest, DIGEST_PATTERN, "image digest")
+    status, body = _artifact_response(
+        destination, username=username, password=password, opener=opener
+    )
+    if status != 200:
+        raise RuntimeError(f"Harbor manifest verification returned HTTP {status}")
+    try:
+        artifact = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeError("Harbor returned invalid artifact JSON") from exc
+    harbor_digest = str(artifact.get("digest") or "").strip()
+    if not DIGEST_PATTERN.fullmatch(harbor_digest):
+        raise RuntimeError("Harbor response omitted a valid artifact digest")
+    if harbor_digest != digest:
+        raise RuntimeError("Harbor digest does not match Kaniko evidence")
+    expected_tag = destination.rsplit(":", 1)[1]
+    tag_names = {
+        str(item.get("name"))
+        for item in artifact.get("tags") or []
+        if isinstance(item, dict) and item.get("name")
+    }
+    if expected_tag not in tag_names:
+        raise RuntimeError("Harbor artifact does not contain the expected tag")
+
+
 def render_kustomization(source: str, digest: str, image: str = DEFAULT_IMAGE) -> str:
     """Replace exactly one matching Kustomize image digest without reformatting."""
     digest = _validated(digest, DIGEST_PATTERN, "image digest")
@@ -59,17 +200,16 @@ def write_release_artifacts(
     *,
     digest: str,
     source_revision: str,
+    build_number: str,
     destination: str,
     kustomization: Path,
     output_dir: Path,
 ) -> dict[str, str]:
     """Write a rendered manifest, patch, and credential-free release metadata."""
     digest = _validated(digest, DIGEST_PATTERN, "image digest")
-    source_revision = _validated(
-        source_revision, REVISION_PATTERN, "source revision"
+    source_revision, build_number = validate_destination(
+        destination, source_revision, build_number
     )
-    if destination != f"{DEFAULT_IMAGE}:git-{source_revision}":
-        raise ValueError("destination must be the immutable git-<revision> tag")
 
     source = kustomization.read_text(encoding="utf-8")
     rendered = render_kustomization(source, digest)
@@ -86,11 +226,10 @@ def write_release_artifacts(
         raise ValueError("published digest already matches the Flux manifest")
 
     output_dir.mkdir(parents=True, exist_ok=True)
-    (output_dir / "hermes-kustomization.yaml").write_text(
-        rendered, encoding="utf-8"
-    )
+    (output_dir / "hermes-kustomization.yaml").write_text(rendered, encoding="utf-8")
     (output_dir / "hermes-image-update.patch").write_text(patch, encoding="utf-8")
     metadata = {
+        "build_number": build_number,
         "digest": digest,
         "flux_image": f"{DEFAULT_IMAGE}@{digest}",
         "image": DEFAULT_IMAGE,
@@ -103,18 +242,53 @@ def write_release_artifacts(
     return metadata
 
 
-def main() -> int:
-    """Validate Kaniko output and render artifacts for the reviewed Flux PR."""
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--digest-file", required=True, type=Path)
+def _credentials() -> tuple[str, str]:
+    """Read the masked, runtime-only Jenkins credential environment."""
+    username = os.environ.get("HARBOR_USER", "")
+    password = os.environ.get("HARBOR_PASSWORD", "")
+    if not username or not password:
+        raise RuntimeError("Harbor credentials are unavailable")
+    return username, password
+
+
+def _common_arguments(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("--source-revision", required=True)
+    parser.add_argument("--build-number", required=True)
     parser.add_argument("--destination", required=True)
-    parser.add_argument("--kustomization", required=True, type=Path)
-    parser.add_argument("--output-dir", required=True, type=Path)
+
+
+def main() -> int:
+    """Fail closed around the unique tag, then verify and render after push."""
+    parser = argparse.ArgumentParser()
+    commands = parser.add_subparsers(dest="command", required=True)
+    absent = commands.add_parser("assert-absent")
+    _common_arguments(absent)
+    render = commands.add_parser("render")
+    _common_arguments(render)
+    render.add_argument("--digest-file", required=True, type=Path)
+    render.add_argument("--image-file", required=True, type=Path)
+    render.add_argument("--kustomization", required=True, type=Path)
+    render.add_argument("--output-dir", required=True, type=Path)
     args = parser.parse_args()
+
+    validate_destination(args.destination, args.source_revision, args.build_number)
+    username, password = _credentials()
+    if args.command == "assert-absent":
+        assert_tag_absent(args.destination, username=username, password=password)
+        return 0
+
+    digest = validate_kaniko_evidence(
+        digest_text=args.digest_file.read_text(encoding="utf-8"),
+        image_text=args.image_file.read_text(encoding="utf-8"),
+        destination=args.destination,
+    )
+    verify_registry_digest(
+        args.destination, digest, username=username, password=password
+    )
     write_release_artifacts(
-        digest=args.digest_file.read_text(encoding="utf-8"),
+        digest=digest,
         source_revision=args.source_revision,
+        build_number=args.build_number,
         destination=args.destination,
         kustomization=args.kustomization,
         output_dir=args.output_dir,

clusters/atlas/flux-system/applications/hermes/kustomization.yaml

@@ -60,3 +60,4 @@ spec:
     - name: keycloak
     - name: longhorn
     - name: vault
+    - name: jenkins

clusters/atlas/flux-system/applications/jenkins/kustomization.yaml

@@ -6,10 +6,9 @@ metadata:
   namespace: flux-system
   annotations:
     kustomize.toolkit.fluxcd.io/ssa: IfNotPresent
-    atlas.bstein.dev/suspend-reason: "CI controller changes are applied only during planned maintenance."
 spec:
   interval: 10m
-  suspend: true
+  suspend: false
   path: ./services/jenkins
   prune: true
   sourceRef:
@@ -18,6 +17,7 @@ spec:
   targetNamespace: jenkins
   dependsOn:
     - name: helm
+    - name: vault
   healthChecks:
     - apiVersion: apps/v1
       kind: Deployment

clusters/atlas/flux-system/applications/vault/kustomization.yaml

@@ -16,6 +16,11 @@ spec:
   targetNamespace: vault
   prune: true
   wait: true
+  healthChecks:
+    - apiVersion: batch/v1
+      kind: Job
+      name: vault-hermes-jenkins-build-token-seed-1
+      namespace: vault
   dependsOn:
     - name: longhorn
     - name: helm

services/hermes/NOTES.md

@@ -118,10 +118,40 @@ jenkins_image_build_trigger.py '<reviewed full main commit>'
 
 The helper has no general Jenkins credential or caller-selectable job name. Its
 Vault-projected token is bound by Jenkins only to `hermes-agent-image`. The job
-also requires its fixed publish confirmation, publishes a source-derived tag,
-records the immutable registry digest, and archives a JSON record plus a Flux
-digest patch. Apply that patch on a new branch and submit it for human review;
-the build never changes Git, reconciles Flux, or deploys by itself.
+also requires its fixed publish confirmation. Each run claims the unique tag
+`git-<reviewed-sha>-build-<jenkins-build-number>` and refuses to overwrite an
+existing tag. It cross-checks Kaniko's digest and tagged-image evidence, then
+independently reads the pushed tag from Harbor before archiving a JSON record
+and Flux digest patch. Apply that patch on a new branch and submit it for human
+review; the build never changes Git, reconciles Flux, or deploys by itself.
+The helper posts only to the fixed HTTPS Build Token Root endpoint and accepts
+only its real queue responses: HTTP 201 or a non-followed HTTP 303 with an exact
+same-origin `/queue/item/<number>/` location. An unauthenticated request is
+denied, and the token cannot select, configure, read, or administer another
+Jenkins job. The pipeline independently rejects any commit that is not the
+current `origin/main`, preserving the human merge/review boundary.
+
+The tracked rollout order is deliberate: the revisioned Vault seed Job must
+complete before the Vault Kustomization becomes Ready, Jenkins depends on
+Vault, and Hermes depends on Jenkins. The seeder uses KV-v2 CAS create-only
+semantics. It never changes an existing token, never fills a missing field in
+an existing secret, and fails closed when it cannot distinguish absence from a
+read error. The recurring Vault configuration job verifies the same invariant.
+
+Rotate this fixed-job token only as a coordinated operator action:
+
+1. Stop new `hermes-agent-image` triggers and wait for its queue and executor to
+   drain.
+2. Read the current KV-v2 metadata version for
+   `kv/atlas/hermes/developer-jenkins`.
+3. Generate a fresh value from Vault and patch only `build_token` with
+   `vault kv patch -cas=<current-version>`. Keep the value in a mode-0600
+   temporary file or standard input, never a command argument or log.
+4. Roll Jenkins first and wait until it is Ready, then roll `hermes-agent` and
+   wait until it is Ready. Do not resume triggers between those two consumers.
+5. Run one reviewed-main canary. If rollback is required, repeat the CAS patch
+   with the previous value as another coordinated rotation; never delete the
+   secret to make the seeder recreate it.
 
 ## The actual supervised triage algorithm
 

services/hermes/scripts/jenkins_image_build_trigger.py

@@ -12,10 +12,8 @@ import urllib.request
 from pathlib import Path
 
 
-JENKINS_BUILD_URL = (
-    "http://jenkins.jenkins.svc.cluster.local:8080/"
-    "buildByToken/buildWithParameters"
-)
+JENKINS_ORIGIN = "https://ci.bstein.dev"
+JENKINS_BUILD_URL = f"{JENKINS_ORIGIN}/buildByToken/buildWithParameters"
 JOB_NAME = "hermes-agent-image"
 TOKEN_FILE = Path("/runtime-access/jenkins-image-build-token")
 REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
@@ -71,13 +69,23 @@ def trigger_build(
     with opener(request, timeout=20) as response:
         status = int(response.status)
         location = response.headers.get("Location", "")
-    if status not in {200, 201, 202, 303}:
+    if status not in {201, 303}:
         raise RuntimeError(f"Jenkins trigger returned HTTP {status}")
-    # Never return the submitted URL or response query: either may contain the
-    # job token. The queue's numeric path is the only useful safe field.
-    queue_path = urllib.parse.urlsplit(location).path
-    if not QUEUE_PATH_PATTERN.fullmatch(queue_path):
-        queue_path = ""
+    if not location:
+        raise RuntimeError("Jenkins trigger omitted the queue Location")
+    queue_url = urllib.parse.urljoin(f"{JENKINS_ORIGIN}/", location)
+    parsed_queue = urllib.parse.urlsplit(queue_url)
+    expected_origin = urllib.parse.urlsplit(JENKINS_ORIGIN)
+    if (
+        parsed_queue.scheme != expected_origin.scheme
+        or parsed_queue.netloc != expected_origin.netloc
+        or parsed_queue.query
+        or parsed_queue.fragment
+        or not QUEUE_PATH_PATTERN.fullmatch(parsed_queue.path)
+    ):
+        raise RuntimeError("Jenkins returned an invalid queue Location")
+    # Never return the submitted URL: its form body contains the job token.
+    queue_path = parsed_queue.path
     return {
         "job": JOB_NAME,
         "queue_path": queue_path,

services/hermes/switchyard-deployment.yaml

@@ -27,7 +27,7 @@ spec:
         prometheus.io/port: "9005"
         prometheus.io/path: /metrics
         vault.hashicorp.com/agent-inject: "true"
-        vault.hashicorp.com/role: hermes-agent
+        vault.hashicorp.com/role: hermes-switchyard
         vault.hashicorp.com/agent-inject-secret-relay-key: kv/data/atlas/hermes/chat-telegram
         vault.hashicorp.com/agent-inject-template-relay-key: |
           {{- with secret "kv/data/atlas/hermes/chat-telegram" -}}

services/jenkins/secretproviderclass.yaml

@@ -8,7 +8,7 @@ spec:
   provider: vault
   parameters:
     vaultAddress: "http://vault.vault.svc.cluster.local:8200"
-    roleName: "jenkins"
+    roleName: "jenkins-vault-sync"
     objects: |
       - objectName: "harbor-pull__dockerconfigjson"
         secretPath: "kv/data/atlas/shared/harbor-pull"

services/vault/hermes-jenkins-build-token-seed-job.yaml

@@ -0,0 +1,47 @@
+# services/vault/hermes-jenkins-build-token-seed-job.yaml
+# Revisioned prerequisite: seed the job-scoped token before Jenkins/Hermes roll.
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: vault-hermes-jenkins-build-token-seed-1
+  namespace: vault
+spec:
+  backoffLimit: 2
+  template:
+    spec:
+      serviceAccountName: vault-admin
+      restartPolicy: Never
+      nodeSelector:
+        hardware: rpi5
+        kubernetes.io/arch: arm64
+        node-role.kubernetes.io/worker: "true"
+      affinity:
+        nodeAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            nodeSelectorTerms:
+              - matchExpressions:
+                  - key: kubernetes.io/hostname
+                    operator: NotIn
+                    values: [titan-04, titan-14, titan-18, titan-19, titan-24]
+      containers:
+        - name: seed
+          image: hashicorp/vault:1.21.4
+          imagePullPolicy: IfNotPresent
+          command: [sh, /scripts/vault_hermes_jenkins_build_token_ensure.sh]
+          env:
+            - name: VAULT_ADDR
+              value: http://vault.vault.svc.cluster.local:8200
+            - name: VAULT_K8S_ROLE
+              value: vault-admin
+          volumeMounts:
+            - name: scripts
+              mountPath: /scripts
+              readOnly: true
+          resources:
+            requests: {cpu: 25m, memory: 32Mi}
+            limits: {cpu: 250m, memory: 128Mi}
+      volumes:
+        - name: scripts
+          configMap:
+            name: vault-k8s-auth-config-script
+            defaultMode: 0555

services/vault/kustomization.yaml

@@ -12,6 +12,7 @@ resources:
   - statefulset.yaml
   - k8s-auth-config-cronjob.yaml
   - hermes-auth-role-bootstrap-job.yaml
+  - hermes-jenkins-build-token-seed-job.yaml
   - oidc-config-cronjob.yaml
   - service.yaml
   - certificate.yaml
@@ -27,6 +28,7 @@ configMapGenerator:
   - name: vault-k8s-auth-config-script
     files:
       - vault_k8s_auth_configure.sh=scripts/vault_k8s_auth_configure.sh
+      - vault_hermes_jenkins_build_token_ensure.sh=scripts/vault_hermes_jenkins_build_token_ensure.sh
   - name: vault-entrypoint
     files:
       - vault-entrypoint.sh=scripts/vault-entrypoint.sh

services/vault/scripts/vault_hermes_jenkins_build_token_ensure.sh

@@ -0,0 +1,153 @@
+#!/usr/bin/env sh
+set -eu
+
+secret_path="kv/atlas/hermes/developer-jenkins"
+jwt_file="${VAULT_K8S_JWT_FILE:-/var/run/secrets/kubernetes.io/serviceaccount/token}"
+vault_role="${VAULT_K8S_ROLE:-vault-admin}"
+
+log() { printf '[hermes-jenkins-token] %s\n' "$*" >&2; }
+
+retry_command() {
+  attempt=1
+  while [ "${attempt}" -le 5 ]; do
+    set +e
+    command_output="$("$@" 2>&1)"
+    command_status=$?
+    set -e
+    if [ "${command_status}" -eq 0 ]; then
+      printf '%s' "${command_output}"
+      return 0
+    fi
+    if [ "${attempt}" -lt 5 ]; then
+      sleep "${attempt}"
+    fi
+    attempt=$((attempt + 1))
+  done
+  return "${command_status}"
+}
+
+ensure_token() {
+  if [ -n "${VAULT_TOKEN:-}" ]; then
+    return
+  fi
+  jwt="$(cat "${jwt_file}")"
+  VAULT_TOKEN="$(retry_command vault write -field=token \
+    auth/kubernetes/login role="${vault_role}" jwt="${jwt}")" || {
+      log "Vault Kubernetes login failed after retries"
+      exit 1
+    }
+  unset jwt
+  if [ -z "${VAULT_TOKEN}" ]; then
+    log "Vault Kubernetes login returned an empty token"
+    exit 1
+  fi
+  export VAULT_TOKEN
+}
+
+# Return 0 when the field exists, 10 when the secret is absent, 11 when the
+# secret exists without the field, and 12 for a persistent read error.
+read_build_token() {
+  attempt=1
+  while [ "${attempt}" -le 5 ]; do
+    set +e
+    secret_error="$(vault kv get "${secret_path}" 2>&1 >/dev/null)"
+    secret_status=$?
+    set -e
+    if [ "${secret_status}" -eq 0 ]; then
+      set +e
+      build_token="$(vault kv get -field=build_token "${secret_path}" 2>&1)"
+      field_status=$?
+      set -e
+      if [ "${field_status}" -eq 0 ] && [ -n "${build_token}" ]; then
+        return 0
+      fi
+      if printf '%s' "${build_token}" | grep -q 'No value found'; then
+        unset build_token
+        return 11
+      fi
+      if [ "${field_status}" -eq 0 ]; then
+        unset build_token
+        return 11
+      fi
+    elif printf '%s' "${secret_error}" | grep -Eq 'Code: 404|No value found at'; then
+      return 10
+    fi
+    if [ "${attempt}" -lt 5 ]; then
+      sleep "${attempt}"
+    fi
+    attempt=$((attempt + 1))
+  done
+  unset secret_error build_token
+  return 12
+}
+
+ensure_token
+
+if read_build_token; then
+  read_status=0
+else
+  read_status=$?
+fi
+case "${read_status}" in
+  0)
+    unset build_token
+    log "job-scoped token already present; no write performed"
+    exit 0
+    ;;
+  11)
+    log "secret exists without build_token; refusing to overwrite existing fields"
+    exit 1
+    ;;
+  12)
+    log "secret read failed after retries; refusing to seed"
+    exit 1
+    ;;
+  10) ;;
+  *)
+    log "unexpected secret read state ${read_status}"
+    exit 1
+    ;;
+esac
+
+new_token="$(retry_command vault write -field=random_bytes \
+  sys/tools/random/32 format=hex)" || {
+    log "Vault random token generation failed after retries"
+    exit 1
+  }
+if [ -z "${new_token}" ]; then
+  log "Vault returned an empty random token"
+  exit 1
+fi
+
+set +e
+create_error="$(printf '%s' "${new_token}" | \
+  vault kv put -cas=0 "${secret_path}" build_token=- 2>&1 >/dev/null)"
+create_status=$?
+set -e
+unset new_token
+if [ "${create_status}" -eq 0 ]; then
+  log "job-scoped token created with KV-v2 CAS create-only semantics"
+  exit 0
+fi
+
+# A simultaneous revisioned Job/Cron run may win the create. Accept that race
+# only if a fresh retried read finds the complete field; never rotate it.
+if printf '%s' "${create_error}" | grep -qi 'check-and-set'; then
+  unset create_error
+  if read_build_token; then
+    read_status=0
+  else
+    read_status=$?
+  fi
+  if [ "${read_status}" -eq 0 ]; then
+    unset build_token
+    log "another seeder won CAS; existing token preserved"
+    exit 0
+  fi
+  log "CAS race did not produce a readable build_token; refusing to continue"
+  exit 1
+fi
+
+unset create_error
+log "KV-v2 CAS create failed; refusing to retry a write"
+exit 1

services/vault/scripts/vault_k8s_auth_configure.sh

@@ -239,8 +239,10 @@ write_policy_and_role "nextcloud" "nextcloud" "nextcloud-vault" \
   "nextcloud/* shared/keycloak-admin shared/postmark-relay" ""
 write_policy_and_role "comms" "comms" "comms-vault,atlasbot" \
   "comms/* shared/chat-ai-keys-runtime shared/harbor-pull" ""
-write_policy_and_role "jenkins" "jenkins" "jenkins,jenkins-vault-sync" \
-  "jenkins/* shared/harbor-pull quality/sonarqube-oidc" "hermes/developer-jenkins"
+write_policy_and_role "jenkins" "jenkins" "jenkins" \
+  "jenkins/* shared/harbor-pull quality/sonarqube-oidc hermes/developer-jenkins" ""
+write_policy_and_role "jenkins-vault-sync" "jenkins" "jenkins-vault-sync" \
+  "shared/harbor-pull" ""
 write_policy_and_role "monitoring" "monitoring" "monitoring-vault-sync" \
   "monitoring/* shared/postmark-relay shared/harbor-pull" ""
 write_policy_and_role "logging" "logging" "logging-vault-sync" \
@@ -255,8 +257,10 @@ write_policy_and_role "game-stream" "game-stream" "game-stream-vault" \
   "game-stream/*" ""
 write_policy_and_role "hermes" "hermes" "hermes-vault,hermes-triage" \
   "hermes/triage-oidc hermes/agent-tokens hermes/triage-api" ""
-write_policy_and_role "hermes-agent" "hermes" "hermes-agent,hermes-switchyard" \
+write_policy_and_role "hermes-agent" "hermes" "hermes-agent" \
   "hermes/agent-oidc hermes/agent-tokens hermes/chat-telegram hermes/developer-keycloak hermes/developer-gitea hermes/developer-harbor hermes/developer-jenkins hermes/developer-ssh" ""
+write_policy_and_role "hermes-switchyard" "hermes" "hermes-switchyard" \
+  "hermes/chat-telegram" ""
 write_policy_and_role "hermes-credential-sync" "hermes" "hermes-agent" \
   "" "hermes/agent-tokens"
 write_policy_and_role "hermes-node-ssh" "hermes" "hermes-node-ssh-access" \
@@ -304,28 +308,9 @@ write_policy_and_role "postgres" "postgres" "postgres-vault" \
 write_policy_and_role "vault" "vault" "vault" \
   "vault/*" ""
 
-# The Build Token Root plugin binds this value to one fixed Jenkins job. Both
-# Jenkins and agent.hermes receive it from Vault at pod start; it is never
-# rendered into Git, a ConfigMap, a Kubernetes Secret, or a build log.
-if existing_build_token="$(vault kv get -field=build_token kv/atlas/hermes/developer-jenkins 2>/dev/null)" \
-  && [ -n "${existing_build_token}" ]; then
-  log "Hermes Jenkins image-build token already present"
-else
-  if vault kv get kv/atlas/hermes/developer-jenkins >/dev/null 2>&1; then
-    log "Hermes Jenkins credential exists without build_token; refusing to overwrite it"
-    exit 1
-  fi
-  build_token="$(vault_cmd write -field=random_bytes sys/tools/random/32 format=hex)"
-  if [ -z "${build_token}" ]; then
-    log "Vault returned an empty Hermes Jenkins image-build token"
-    exit 1
-  fi
-  vault_cmd kv put kv/atlas/hermes/developer-jenkins \
-    build_token="${build_token}" >/dev/null
-  unset build_token
-  log "Hermes Jenkins image-build token created in Vault"
-fi
-unset existing_build_token
+# This verifier is also run by a revisioned prerequisite Job. It creates only
+# when the entire KV-v2 secret is absent and never rotates an existing value.
+/scripts/vault_hermes_jenkins_build_token_ensure.sh
 
 write_policy_and_role "sso-secrets" "sso" "mas-secrets-ensure" \
   "shared/keycloak-admin shared/postmark-relay maintenance/metis-ssh-keys" \

testing/tests/test_hermes_image_builder.py

@@ -15,13 +15,13 @@ import yaml
 REPO_ROOT = Path(__file__).resolve().parents[2]
 PIPELINE_PATH = REPO_ROOT / "ci/Jenkinsfile.hermes-agent-image"
 RELEASE_SCRIPT = REPO_ROOT / "ci/scripts/hermes_image_release.py"
-TRIGGER_SCRIPT = (
-    REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.py"
-)
+TRIGGER_SCRIPT = REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.py"
 
 
 def _load_release_module():
-    spec = importlib.util.spec_from_file_location("hermes_image_release", RELEASE_SCRIPT)
+    spec = importlib.util.spec_from_file_location(
+        "hermes_image_release", RELEASE_SCRIPT
+    )
     assert spec is not None and spec.loader is not None
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)
@@ -29,7 +29,9 @@ def _load_release_module():
 
 
 def _load_trigger_module():
-    spec = importlib.util.spec_from_file_location("jenkins_image_build_trigger", TRIGGER_SCRIPT)
+    spec = importlib.util.spec_from_file_location(
+        "jenkins_image_build_trigger", TRIGGER_SCRIPT
+    )
     assert spec is not None and spec.loader is not None
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)
@@ -75,7 +77,9 @@ def test_builder_is_restricted_to_healthy_rpi5_capacity() -> None:
     expressions = spec["affinity"]["nodeAffinity"][
         "requiredDuringSchedulingIgnoredDuringExecution"
     ]["nodeSelectorTerms"][0]["matchExpressions"]
-    host_rule = next(rule for rule in expressions if rule["key"] == "kubernetes.io/hostname")
+    host_rule = next(
+        rule for rule in expressions if rule["key"] == "kubernetes.io/hostname"
+    )
     assert host_rule["operator"] == "NotIn"
     assert set(host_rule["values"]) >= {
         "titan-04",
@@ -92,15 +96,17 @@ def test_pipeline_requires_reviewed_main_and_runtime_credentials() -> None:
     source = PIPELINE_PATH.read_text(encoding="utf-8")
     assert 'test "${PUBLISH_IMAGE}" = "true"' in source
     assert 'test "${CONFIRM_PUBLISH}" = "PUBLISH HERMES AGENT"' in source
-    assert 'git rev-parse origin/main' in source
+    assert "git rev-parse origin/main" in source
     assert "credentialsId: 'harbor-robot'" in source
     assert "set +x" in source
     assert "umask 077" in source
     assert "unset HARBOR_USER HARBOR_PASSWORD auth" in source
     assert "/busybox/rm -f /kaniko/.docker/config.json" in source
     assert "--digest-file=" in source
-    assert "--image-name-with-digest-file=" in source
+    assert "--image-name-tag-with-digest-file=" in source
     assert "--destination=" in source
+    assert "assert-absent" in source
+    assert "git-${actual_revision}-build-${BUILD_NUMBER}" in source
 
 
 def test_jenkins_job_is_manual_and_reads_pipeline_from_main() -> None:
@@ -116,7 +122,9 @@ def test_jenkins_job_is_manual_and_reads_pipeline_from_main() -> None:
     )[0]
     assert "branches('*/main')" in block
     assert "scriptPath('ci/Jenkinsfile.hermes-agent-image')" in block
-    assert "authenticationToken(System.getenv('HERMES_AGENT_IMAGE_BUILD_TOKEN'))" in block
+    assert (
+        "authenticationToken(System.getenv('HERMES_AGENT_IMAGE_BUILD_TOKEN'))" in block
+    )
     assert "pipelineTriggers" not in block
     assert "scmTrigger" not in block
 
@@ -131,7 +139,7 @@ def test_agent_trigger_is_limited_to_the_image_job(tmp_path: Path) -> None:
 
     class Response(io.BytesIO):
         status = 201
-        headers = {"Location": "http://jenkins/queue/item/42/"}
+        headers = {"Location": "https://ci.bstein.dev/queue/item/42/"}
 
         def __enter__(self):
             return self
@@ -180,7 +188,7 @@ def test_agent_trigger_accepts_existing_queue_redirect(tmp_path: Path) -> None:
 
     class Response(io.BytesIO):
         status = 303
-        headers = {"Location": "http://jenkins/queue/item/7/"}
+        headers = {"Location": "https://ci.bstein.dev/queue/item/7/"}
 
         def __enter__(self):
             return self
@@ -203,29 +211,37 @@ def test_job_token_is_generated_and_injected_only_at_runtime() -> None:
     vault = (
         REPO_ROOT / "services/vault/scripts/vault_k8s_auth_configure.sh"
     ).read_text(encoding="utf-8")
+    seeder = (
+        REPO_ROOT / "services/vault/scripts/vault_hermes_jenkins_build_token_ensure.sh"
+    ).read_text(encoding="utf-8")
     jenkins = (REPO_ROOT / "services/jenkins/deployment.yaml").read_text(
         encoding="utf-8"
     )
     agent = (REPO_ROOT / "services/hermes/agent-deployment.yaml").read_text(
         encoding="utf-8"
     )
-    stage = (
-        REPO_ROOT / "services/hermes/scripts/stage_runtime_access.py"
-    ).read_text(encoding="utf-8")
+    stage = (REPO_ROOT / "services/hermes/scripts/stage_runtime_access.py").read_text(
+        encoding="utf-8"
+    )
     assert "build-token-root:365.v717f8685a_09e" in plugins
-    assert "sys/tools/random/32 format=hex" in vault
-    assert "kv/atlas/hermes/developer-jenkins" in vault
+    assert "vault_hermes_jenkins_build_token_ensure.sh" in vault
+    assert "sys/tools/random/32 format=hex" in seeder
+    assert "kv put -cas=0" in seeder
+    assert "kv/atlas/hermes/developer-jenkins" in seeder
     assert "HERMES_AGENT_IMAGE_BUILD_TOKEN={{ .Data.data.build_token }}" in jenkins
     assert "agent-inject-secret-jenkins-image-build-token" in agent
     assert '"jenkins-image-build-token"' in stage
 
 
-def test_release_renderer_preserves_manifest_and_emits_safe_artifacts(tmp_path: Path) -> None:
+def test_release_renderer_preserves_manifest_and_emits_safe_artifacts(
+    tmp_path: Path,
+) -> None:
     """The release artifact is exact, reviewable, and contains no credentials."""
     module = _load_release_module()
     old_digest = "sha256:" + "1" * 64
     new_digest = "sha256:" + "2" * 64
     revision = "a" * 40
+    build_number = "17"
     manifest = tmp_path / "kustomization.yaml"
     manifest.write_text(
         "apiVersion: kustomize.config.k8s.io/v1beta1\n"
@@ -239,23 +255,28 @@ def test_release_renderer_preserves_manifest_and_emits_safe_artifacts(tmp_path:
     metadata = module.write_release_artifacts(
         digest=f"{new_digest}\n",
         source_revision=revision,
-        destination=f"{module.DEFAULT_IMAGE}:git-{revision}",
+        build_number=build_number,
+        destination=f"{module.DEFAULT_IMAGE}:git-{revision}-build-{build_number}",
         kustomization=manifest,
         output_dir=output,
     )
 
     assert manifest.read_text(encoding="utf-8").endswith(f"{old_digest}\n")
-    assert (output / "hermes-kustomization.yaml").read_text(
-        encoding="utf-8"
-    ).endswith(f"{new_digest}\n")
+    assert (
+        (output / "hermes-kustomization.yaml")
+        .read_text(encoding="utf-8")
+        .endswith(f"{new_digest}\n")
+    )
     patch = (output / "hermes-image-update.patch").read_text(encoding="utf-8")
     assert f"-    digest: {old_digest}" in patch
     assert f"+    digest: {new_digest}" in patch
-    assert json.loads(
-        (output / "hermes-agent-image.json").read_text(encoding="utf-8")
-    ) == metadata
+    assert (
+        json.loads((output / "hermes-agent-image.json").read_text(encoding="utf-8"))
+        == metadata
+    )
     assert set(metadata) == {
         "digest",
+        "build_number",
         "flux_image",
         "image",
         "published_tag",
@@ -266,9 +287,21 @@ def test_release_renderer_preserves_manifest_and_emits_safe_artifacts(tmp_path:
 @pytest.mark.parametrize(
     ("digest", "revision", "destination"),
     [
-        ("latest", "a" * 40, "registry.bstein.dev/bstein/hermes-agent:git-" + "a" * 40),
-        ("sha256:" + "1" * 64, "short", "registry.bstein.dev/bstein/hermes-agent:git-short"),
-        ("sha256:" + "1" * 64, "a" * 40, "registry.bstein.dev/bstein/hermes-agent:latest"),
+        (
+            "latest",
+            "a" * 40,
+            "registry.bstein.dev/bstein/hermes-agent:git-" + "a" * 40 + "-build-1",
+        ),
+        (
+            "sha256:" + "1" * 64,
+            "short",
+            "registry.bstein.dev/bstein/hermes-agent:git-short-build-1",
+        ),
+        (
+            "sha256:" + "1" * 64,
+            "a" * 40,
+            "registry.bstein.dev/bstein/hermes-agent:latest",
+        ),
     ],
 )
 def test_release_renderer_rejects_unpinned_inputs(
@@ -287,6 +320,7 @@ def test_release_renderer_rejects_unpinned_inputs(
         module.write_release_artifacts(
             digest=digest,
             source_revision=revision,
+            build_number="1",
             destination=destination,
             kustomization=manifest,
             output_dir=tmp_path / "out",
@@ -319,13 +353,16 @@ def test_release_renderer_fails_closed_on_manifest_drift(tmp_path: Path) -> None
         module.write_release_artifacts(
             digest=digest,
             source_revision=revision,
-            destination=f"{module.DEFAULT_IMAGE}:git-{revision}",
+            build_number="1",
+            destination=f"{module.DEFAULT_IMAGE}:git-{revision}-build-1",
             kustomization=manifest,
             output_dir=tmp_path / "out",
         )
 
 
-def test_release_cli_reads_digest_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_release_cli_reads_digest_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
     """The pipeline CLI uses the Kaniko digest file as its sole digest input."""
     module = _load_release_module()
     digest = "sha256:" + "c" * 64
@@ -339,17 +376,30 @@ def test_release_cli_reads_digest_file(tmp_path: Path, monkeypatch: pytest.Monke
     )
     digest_file = tmp_path / "digest"
     digest_file.write_text(digest + "\n", encoding="utf-8")
+    destination = f"{module.DEFAULT_IMAGE}:git-{revision}-build-9"
+    image_file = tmp_path / "image"
+    image_file.write_text(f"{destination}@{digest}\n", encoding="utf-8")
     output = tmp_path / "out"
+    monkeypatch.setenv("HARBOR_USER", "robot")
+    monkeypatch.setenv("HARBOR_PASSWORD", "secret")
+    monkeypatch.setattr(
+        module, "verify_registry_digest", lambda *_args, **_kwargs: None
+    )
     monkeypatch.setattr(
         "sys.argv",
         [
             "hermes_image_release.py",
+            "render",
             "--digest-file",
             str(digest_file),
+            "--image-file",
+            str(image_file),
             "--source-revision",
             revision,
+            "--build-number",
+            "9",
             "--destination",
-            f"{module.DEFAULT_IMAGE}:git-{revision}",
+            destination,
             "--kustomization",
             str(manifest),
             "--output-dir",
@@ -357,6 +407,9 @@ def test_release_cli_reads_digest_file(tmp_path: Path, monkeypatch: pytest.Monke
         ],
     )
     assert module.main() == 0
-    assert json.loads(
-        (output / "hermes-agent-image.json").read_text(encoding="utf-8")
-    )["digest"] == digest
+    assert (
+        json.loads((output / "hermes-agent-image.json").read_text(encoding="utf-8"))[
+            "digest"
+        ]
+        == digest
+    )

testing/tests/test_hermes_image_builder_adversarial.py

@@ -0,0 +1,356 @@
+"""Adversarial boundaries for the Hermes image publisher and trigger."""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+RELEASE_SCRIPT = REPO_ROOT / "ci/scripts/hermes_image_release.py"
+TRIGGER_SCRIPT = REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.py"
+
+
+def _load(path: Path, name: str):
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class Response(io.BytesIO):
+    """Small context-managed HTTP response fixture."""
+
+    def __init__(
+        self,
+        status: int,
+        headers: dict[str, str] | None = None,
+        body: bytes = b"",
+    ):
+        super().__init__(body)
+        self.status = status
+        self.headers = headers or {}
+
+
+def test_kaniko_evidence_binds_digest_destination_and_unique_build() -> None:
+    """Both Kaniko artifacts must describe one exact non-replayable tag."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_evidence")
+    revision = "a" * 40
+    digest = "sha256:" + "b" * 64
+    destination = f"{module.DEFAULT_IMAGE}:git-{revision}-build-42"
+    assert module.validate_destination(destination, revision, "42") == (
+        revision,
+        "42",
+    )
+    assert (
+        module.validate_kaniko_evidence(
+            digest_text=f"{digest}\n",
+            image_text=f"{destination}@{digest}\n",
+            destination=destination,
+        )
+        == digest
+    )
+
+
+@pytest.mark.parametrize(
+    ("digest_text", "image_template"),
+    [
+        ("sha256:" + "a" * 64 + "\nextra\n", "{destination}@{digest}"),
+        ("sha256:" + "a" * 64, "{destination}@sha256:" + "b" * 64),
+        ("sha256:" + "a" * 64, "registry.invalid/x:y@{digest}"),
+        ("sha256:" + "a" * 64, "{destination}@{digest}\nextra"),
+    ],
+)
+def test_kaniko_evidence_rejects_cross_artifact_mismatch(
+    digest_text: str, image_template: str
+) -> None:
+    """A digest, tag, repository, or cardinality mismatch stops rendering."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_mismatch")
+    revision = "c" * 40
+    digest = "sha256:" + "a" * 64
+    destination = f"{module.DEFAULT_IMAGE}:git-{revision}-build-8"
+    image_text = image_template.format(destination=destination, digest=digest)
+    with pytest.raises(ValueError):
+        module.validate_kaniko_evidence(
+            digest_text=digest_text,
+            image_text=image_text,
+            destination=destination,
+        )
+
+
+def test_registry_preflight_rejects_existing_tag_and_auth_failures() -> None:
+    """Only an authenticated 404 permits Kaniko to claim the unique tag."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_preflight")
+    destination = f"{module.DEFAULT_IMAGE}:git-{'d' * 40}-build-3"
+    captured = {}
+
+    def missing(request, timeout):
+        captured["request"] = request
+        captured["timeout"] = timeout
+        return Response(404)
+
+    module.assert_tag_absent(
+        destination, username="robot", password="private", opener=missing
+    )
+    request = captured["request"]
+    assert request.method == "GET"
+    assert request.full_url.startswith(
+        "https://registry.bstein.dev/api/v2.0/projects/bstein/repositories/"
+        "hermes-agent/artifacts/"
+    )
+    assert "private" not in request.full_url
+    assert captured["timeout"] == 20
+
+    for status, message in (
+        (200, "already exists"),
+        (401, "HTTP 401"),
+        (503, "HTTP 503"),
+    ):
+        with pytest.raises(RuntimeError, match=message):
+            module.assert_tag_absent(
+                destination,
+                username="robot",
+                password="private",
+                opener=lambda *_args, code=status: Response(code),
+            )
+
+
+def test_harbor_client_fails_closed_on_input_size_auth_and_json(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Malformed inputs, oversized bodies, missing auth, and bad JSON all fail."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_harbor_errors")
+    destination = f"{module.DEFAULT_IMAGE}:git-{'1' * 40}-build-6"
+    with pytest.raises(ValueError, match="destination"):
+        module.assert_tag_absent(
+            "registry.invalid/x:tag", username="robot", password="private"
+        )
+    with pytest.raises(RuntimeError, match="credentials"):
+        module.assert_tag_absent(destination, username="", password="private")
+    with pytest.raises(RuntimeError, match="size limit"):
+        module.assert_tag_absent(
+            destination,
+            username="robot",
+            password="private",
+            opener=lambda *_args: Response(200, body=b"x" * 1_048_577),
+        )
+    with pytest.raises(RuntimeError, match="invalid artifact JSON"):
+        module.verify_registry_digest(
+            destination,
+            "sha256:" + "2" * 64,
+            username="robot",
+            password="private",
+            opener=lambda *_args: Response(200, body=b"\xff"),
+        )
+    monkeypatch.delenv("HARBOR_USER", raising=False)
+    monkeypatch.delenv("HARBOR_PASSWORD", raising=False)
+    with pytest.raises(RuntimeError, match="unavailable"):
+        module._credentials()
+
+
+def test_default_harbor_opener_returns_http_responses(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The real wrapper returns both normal and non-redirect HTTP responses."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_opener")
+    request = module.urllib.request.Request("https://registry.bstein.dev/test")
+
+    class SuccessOpener:
+        def open(self, _request, timeout):
+            assert timeout == 7
+            return Response(204)
+
+    monkeypatch.setattr(
+        module.urllib.request, "build_opener", lambda *_handlers: SuccessOpener()
+    )
+    assert module._registry_request(request, 7).status == 204
+
+    class ErrorOpener:
+        def open(self, _request, timeout):
+            assert timeout == 8
+            raise module.urllib.error.HTTPError(
+                _request.full_url, 404, "missing", {}, None
+            )
+
+    monkeypatch.setattr(
+        module.urllib.request, "build_opener", lambda *_handlers: ErrorOpener()
+    )
+    assert module._registry_request(request, 8).code == 404
+
+
+@pytest.mark.parametrize(
+    ("status", "artifact", "message"),
+    [
+        (404, {}, "HTTP 404"),
+        (200, {}, "omitted"),
+        (200, {"digest": "sha256:" + "2" * 64}, "does not match"),
+        (200, {"digest": "latest"}, "omitted"),
+    ],
+)
+def test_registry_verification_rejects_missing_or_mismatched_digest(
+    status: int, artifact: dict[str, str], message: str
+) -> None:
+    """Rendering requires an independent exact Harbor digest response."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_registry")
+    digest = "sha256:" + "1" * 64
+    destination = f"{module.DEFAULT_IMAGE}:git-{'e' * 40}-build-4"
+    body = json.dumps(artifact).encode("utf-8")
+    with pytest.raises(RuntimeError, match=message):
+        module.verify_registry_digest(
+            destination,
+            digest,
+            username="robot",
+            password="private",
+            opener=lambda *_args: Response(status, body=body),
+        )
+
+
+def test_registry_verification_accepts_exact_pushed_digest() -> None:
+    """An exact authenticated Harbor digest allows artifact rendering."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_registry_ok")
+    digest = "sha256:" + "4" * 64
+    destination = f"{module.DEFAULT_IMAGE}:git-{'f' * 40}-build-5"
+    with pytest.raises(RuntimeError, match="expected tag"):
+        module.verify_registry_digest(
+            destination,
+            digest,
+            username="robot",
+            password="private",
+            opener=lambda *_args: Response(
+                200,
+                body=json.dumps(
+                    {"digest": digest, "tags": [{"name": "different-tag"}]}
+                ).encode("utf-8"),
+            ),
+        )
+    module.verify_registry_digest(
+        destination,
+        digest,
+        username="robot",
+        password="private",
+        opener=lambda *_args: Response(
+            200,
+            body=json.dumps(
+                {
+                    "digest": digest,
+                    "tags": [{"name": destination.rsplit(":", 1)[1]}],
+                }
+            ).encode("utf-8"),
+        ),
+    )
+
+
+def test_release_main_preflight_uses_fixed_credentials_and_destination(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The preflight CLI validates and checks exactly the requested build tag."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_main_absent")
+    revision = "3" * 40
+    destination = f"{module.DEFAULT_IMAGE}:git-{revision}-build-12"
+    captured = {}
+    monkeypatch.setenv("HARBOR_USER", "robot")
+    monkeypatch.setenv("HARBOR_PASSWORD", "private")
+
+    def assert_absent(value, **credentials):
+        captured["destination"] = value
+        captured["credentials"] = credentials
+
+    monkeypatch.setattr(module, "assert_tag_absent", assert_absent)
+    monkeypatch.setattr(
+        "sys.argv",
+        [
+            "hermes_image_release.py",
+            "assert-absent",
+            "--source-revision",
+            revision,
+            "--build-number",
+            "12",
+            "--destination",
+            destination,
+        ],
+    )
+    assert module.main() == 0
+    assert captured == {
+        "destination": destination,
+        "credentials": {"username": "robot", "password": "private"},
+    }
+
+
+def test_renderer_skips_a_matching_entry_without_a_digest() -> None:
+    """Only the one complete image entry is changed when an earlier entry drifts."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_manifest_scan")
+    digest = "sha256:" + "5" * 64
+    source = (
+        "images:\n"
+        f"  - name: {module.DEFAULT_IMAGE}\n"
+        "    newTag: ignored\n"
+        "  - name: example.invalid/other\n"
+        "    newTag: stable\n"
+        f"  - name: {module.DEFAULT_IMAGE}\n"
+        "    digest: sha256:" + "0" * 64 + "\n"
+    )
+    assert module.render_kustomization(source, digest).endswith(f"{digest}\n")
+
+
+@pytest.mark.parametrize("status", [200, 202, 204, 301, 302, 307, 308])
+def test_trigger_rejects_non_plugin_success_status(tmp_path: Path, status: int) -> None:
+    """Generic proxy successes and redirects are not proof of a queued build."""
+    module = _load(TRIGGER_SCRIPT, f"jenkins_trigger_status_{status}")
+    token = tmp_path / "token"
+    token.write_text("private\n", encoding="utf-8")
+    response = lambda *_args, **_kwargs: Response(  # noqa: E731
+        status, {"Location": "https://ci.bstein.dev/queue/item/9/"}
+    )
+    with pytest.raises(RuntimeError, match=f"HTTP {status}"):
+        module.trigger_build("a" * 40, token_file=token, opener=response)
+
+
+@pytest.mark.parametrize(
+    "location",
+    [
+        "",
+        "https://evil.invalid/queue/item/9/",
+        "http://ci.bstein.dev/queue/item/9/",
+        "https://ci.bstein.dev/job/hermes-agent-image/9/",
+        "https://ci.bstein.dev/queue/item/9/?token=leak",
+        "https://ci.bstein.dev/queue/item/9/#fragment",
+        "https://ci.bstein.dev/queue/item/not-a-number/",
+    ],
+)
+def test_trigger_rejects_missing_malformed_or_cross_origin_queue_location(
+    tmp_path: Path, location: str
+) -> None:
+    """A real accepted status still needs the exact same-origin queue resource."""
+    module = _load(TRIGGER_SCRIPT, "jenkins_trigger_location")
+    token = tmp_path / "token"
+    token.write_text("private\n", encoding="utf-8")
+    response = lambda *_args, **_kwargs: Response(  # noqa: E731
+        201, {"Location": location}
+    )
+    with pytest.raises(RuntimeError, match="Location"):
+        module.trigger_build("b" * 40, token_file=token, opener=response)
+
+
+def test_trigger_uses_https_and_accepts_exact_relative_queue_path(
+    tmp_path: Path,
+) -> None:
+    """A relative plugin Location is resolved only against the fixed HTTPS origin."""
+    module = _load(TRIGGER_SCRIPT, "jenkins_trigger_origin")
+    token = tmp_path / "token"
+    token.write_text("private\n", encoding="utf-8")
+    captured = {}
+
+    def opener(request, timeout):
+        captured["url"] = request.full_url
+        captured["timeout"] = timeout
+        return Response(201, {"Location": "/queue/item/11/"})
+
+    result = module.trigger_build("c" * 40, token_file=token, opener=opener)
+    assert captured["url"].startswith("https://ci.bstein.dev/")
+    assert captured["timeout"] == 20
+    assert result["queue_path"] == "/queue/item/11/"

testing/tests/test_hermes_image_builder_vault.py

@@ -0,0 +1,233 @@
+"""Vault least-privilege and fail-closed seeding tests for image releases."""
+
+from __future__ import annotations
+
+import os
+import stat
+import subprocess
+from pathlib import Path
+
+import pytest
+import yaml
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+VAULT_CONFIG = REPO_ROOT / "services/vault/scripts/vault_k8s_auth_configure.sh"
+SEEDER = REPO_ROOT / "services/vault/scripts/vault_hermes_jenkins_build_token_ensure.sh"
+
+
+def test_developer_jenkins_access_is_read_only_and_bound_to_two_consumers() -> None:
+    """Only the Jenkins controller and Hermes agent can read the fixed-job token."""
+    source = VAULT_CONFIG.read_text(encoding="utf-8")
+    assert 'write_policy_and_role "jenkins" "jenkins" "jenkins"' in source
+    assert (
+        '"jenkins/* shared/harbor-pull quality/sonarqube-oidc '
+        'hermes/developer-jenkins" ""'
+    ) in source
+    assert 'write_policy_and_role "hermes-agent" "hermes" "hermes-agent"' in source
+    assert 'hermes/developer-jenkins hermes/developer-ssh" ""' in source
+    assert 'write_policy_and_role "hermes-switchyard" "hermes"' in source
+    assert '"hermes/chat-telegram" ""' in source
+    assert '"hermes/developer-jenkins"' not in source
+
+    jenkins_spc = yaml.safe_load(
+        (REPO_ROOT / "services/jenkins/secretproviderclass.yaml").read_text()
+    )
+    assert jenkins_spc["spec"]["parameters"]["roleName"] == "jenkins-vault-sync"
+    switchyard = yaml.safe_load(
+        (REPO_ROOT / "services/hermes/switchyard-deployment.yaml").read_text()
+    )
+    annotations = switchyard["spec"]["template"]["metadata"]["annotations"]
+    assert annotations["vault.hashicorp.com/role"] == "hermes-switchyard"
+
+
+def test_flux_orders_seed_then_jenkins_then_hermes() -> None:
+    """The fixed token must be Ready before either consumer is rolled."""
+    app_root = REPO_ROOT / "clusters/atlas/flux-system/applications"
+    vault = yaml.safe_load((app_root / "vault/kustomization.yaml").read_text())
+    jenkins = yaml.safe_load((app_root / "jenkins/kustomization.yaml").read_text())
+    hermes = yaml.safe_load((app_root / "hermes/kustomization.yaml").read_text())
+    seed_check = {
+        "apiVersion": "batch/v1",
+        "kind": "Job",
+        "name": "vault-hermes-jenkins-build-token-seed-1",
+        "namespace": "vault",
+    }
+    assert seed_check in vault["spec"]["healthChecks"]
+    assert jenkins["spec"]["suspend"] is False
+    assert {item["name"] for item in jenkins["spec"]["dependsOn"]} >= {
+        "helm",
+        "vault",
+    }
+    assert "jenkins" in {item["name"] for item in hermes["spec"]["dependsOn"]}
+
+
+def test_seed_job_is_revisioned_bounded_and_tracks_fail_closed_script() -> None:
+    """The prerequisite is a tracked one-shot on healthy non-reserved capacity."""
+    job = yaml.safe_load(
+        (
+            REPO_ROOT / "services/vault/hermes-jenkins-build-token-seed-job.yaml"
+        ).read_text()
+    )
+    assert job["metadata"]["name"] == "vault-hermes-jenkins-build-token-seed-1"
+    pod = job["spec"]["template"]["spec"]
+    assert pod["serviceAccountName"] == "vault-admin"
+    assert pod["restartPolicy"] == "Never"
+    assert pod["nodeSelector"]["hardware"] == "rpi5"
+    expression = pod["affinity"]["nodeAffinity"][
+        "requiredDuringSchedulingIgnoredDuringExecution"
+    ]["nodeSelectorTerms"][0]["matchExpressions"][0]
+    assert set(expression["values"]) >= {
+        "titan-04",
+        "titan-14",
+        "titan-18",
+        "titan-19",
+        "titan-24",
+    }
+    source = SEEDER.read_text(encoding="utf-8")
+    assert "kv put -cas=0" in source
+    assert "build_token=-" in source
+    assert "kv patch" not in source
+    assert "kv delete" not in source
+
+
+def _fake_vault_tools(tmp_path: Path) -> tuple[Path, Path, Path]:
+    """Create deterministic Vault/sleep fakes and return their evidence paths."""
+    binary_dir = tmp_path / "bin"
+    binary_dir.mkdir(parents=True)
+    calls = tmp_path / "calls"
+    stdin_capture = tmp_path / "stdin"
+    vault = binary_dir / "vault"
+    vault.write_text(
+        """#!/bin/sh
+set -eu
+printf '%s\n' "$*" >> "$FAKE_CALL_LOG"
+scenario="$FAKE_SCENARIO"
+if [ "$1" = "write" ]; then
+  printf '%s\n' generated-token-value
+  exit 0
+fi
+if [ "$1" = "kv" ] && [ "$2" = "put" ]; then
+  cat > "$FAKE_STDIN_CAPTURE"
+  if [ "$scenario" = "cas-race" ]; then
+    : > "$FAKE_WINNER"
+    echo 'check-and-set parameter did not match' >&2
+    exit 2
+  fi
+  exit 0
+fi
+if [ "$1" = "kv" ] && [ "$2" = "get" ]; then
+  if [ "${3:-}" = "-field=build_token" ]; then
+    case "$scenario" in
+      existing|cas-race)
+        printf '%s\n' existing-token
+        exit 0
+        ;;
+      missing-field)
+        echo 'No value found at kv/atlas/hermes/developer-jenkins' >&2
+        exit 2
+        ;;
+    esac
+  fi
+  case "$scenario" in
+    existing|missing-field)
+      exit 0
+      ;;
+    absent)
+      echo 'Code: 404' >&2
+      exit 2
+      ;;
+    cas-race)
+      if [ -f "$FAKE_WINNER" ]; then exit 0; fi
+      echo 'Code: 404' >&2
+      exit 2
+      ;;
+    transient)
+      count=0
+      if [ -f "$FAKE_COUNT" ]; then count="$(cat "$FAKE_COUNT")"; fi
+      count=$((count + 1))
+      printf '%s' "$count" > "$FAKE_COUNT"
+      if [ "$count" -lt 3 ]; then echo 'temporary upstream error' >&2; exit 2; fi
+      echo 'Code: 404' >&2
+      exit 2
+      ;;
+    read-error)
+      echo 'permission denied' >&2
+      exit 2
+      ;;
+  esac
+fi
+echo "unexpected fake Vault call: $*" >&2
+exit 99
+""",
+        encoding="utf-8",
+    )
+    sleep = binary_dir / "sleep"
+    sleep.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
+    for executable in (vault, sleep):
+        executable.chmod(executable.stat().st_mode | stat.S_IXUSR)
+    return binary_dir, calls, stdin_capture
+
+
+def _run_seeder(
+    tmp_path: Path, scenario: str
+) -> tuple[subprocess.CompletedProcess, str]:
+    binary_dir, calls, stdin_capture = _fake_vault_tools(tmp_path)
+    env = {
+        **os.environ,
+        "PATH": f"{binary_dir}:{os.environ['PATH']}",
+        "VAULT_TOKEN": "test-token",
+        "FAKE_SCENARIO": scenario,
+        "FAKE_CALL_LOG": str(calls),
+        "FAKE_STDIN_CAPTURE": str(stdin_capture),
+        "FAKE_WINNER": str(tmp_path / "winner"),
+        "FAKE_COUNT": str(tmp_path / "count"),
+    }
+    result = subprocess.run(
+        ["sh", str(SEEDER)],
+        env=env,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    call_text = calls.read_text(encoding="utf-8") if calls.exists() else ""
+    return result, call_text
+
+
+@pytest.mark.parametrize("scenario", ["existing", "missing-field", "read-error"])
+def test_seeder_never_overwrites_existing_or_ambiguous_state(
+    tmp_path: Path, scenario: str
+) -> None:
+    """Existing fields and persistent read failures can never become a put."""
+    result, calls = _run_seeder(tmp_path, scenario)
+    assert (result.returncode == 0) is (scenario == "existing")
+    assert "kv put" not in calls
+    assert "sys/tools/random" not in calls
+
+
+def test_seeder_uses_single_cas_create_and_stdin_for_absent_secret(
+    tmp_path: Path,
+) -> None:
+    """A confirmed 404 produces one create-only write without a token argument."""
+    result, calls = _run_seeder(tmp_path, "absent")
+    assert result.returncode == 0, result.stderr
+    assert calls.count("kv put") == 1
+    assert "kv put -cas=0 kv/atlas/hermes/developer-jenkins build_token=-" in calls
+    assert "generated-token-value" not in calls
+    assert (tmp_path / "stdin").read_text(encoding="utf-8") == "generated-token-value"
+
+
+def test_seeder_retries_reads_and_accepts_only_a_verified_cas_winner(
+    tmp_path: Path,
+) -> None:
+    """Transient reads retry; a competing create is accepted only after reread."""
+    transient, transient_calls = _run_seeder(tmp_path / "transient", "transient")
+    assert transient.returncode == 0, transient.stderr
+    assert transient_calls.count("kv get kv/atlas/hermes/developer-jenkins") == 3
+    assert transient_calls.count("kv put") == 1
+
+    race, race_calls = _run_seeder(tmp_path / "race", "cas-race")
+    assert race.returncode == 0, race.stderr
+    assert race_calls.count("kv put") == 1
+    assert "kv get -field=build_token" in race_calls
+    assert "another seeder won CAS" in race.stderr