diff --git a/clusters/atlas/flux-system/applications/hermes/kustomization.yaml b/clusters/atlas/flux-system/applications/hermes/kustomization.yaml index 3949282b..8e317da6 100644 --- a/clusters/atlas/flux-system/applications/hermes/kustomization.yaml +++ b/clusters/atlas/flux-system/applications/hermes/kustomization.yaml @@ -30,6 +30,10 @@ spec: kind: Deployment name: hermes-agent namespace: hermes + - apiVersion: apps/v1 + kind: StatefulSet + name: hermes-execution-worker + namespace: hermes - apiVersion: apps/v1 kind: DaemonSet name: hermes-node-ssh-access diff --git a/services/hermes/execution-coordinator-patch.yaml b/services/hermes/execution-coordinator-patch.yaml new file mode 100644 index 00000000..93192937 --- /dev/null +++ b/services/hermes/execution-coordinator-patch.yaml @@ -0,0 +1,66 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: hermes-agent + namespace: hermes +spec: + template: + metadata: + annotations: + vault.hashicorp.com/agent-inject-secret-execution-pool-key: kv/data/atlas/hermes/agent-tokens + vault.hashicorp.com/agent-inject-perms-execution-pool-key: "0600" + vault.hashicorp.com/agent-inject-template-execution-pool-key: | + {{- with secret "kv/data/atlas/hermes/agent-tokens" -}} + {{ printf "hermes-execution-pool-v1:%s" .Data.data.agent_api_key | sha256sum }} + {{- end }} + spec: + containers: + - name: cli-lane-runner + $patch: delete + - name: execution-pool-coordinator + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/execution_pool_coordinator.py] + env: + - {name: HERMES_HOME, value: /opt/data} + - {name: HOME, value: /opt/data/home} + - {name: PYTHONPATH, value: /opt/hermes} + - {name: HERMES_EXECUTION_POOL_KEY_FILE, value: /pool-access/execution-pool-key} + - {name: HERMES_EXECUTION_DEFAULT_REPO_URL, value: https://scm.bstein.dev/atlas/titan-iac.git} + - {name: HERMES_EXECUTION_DEFAULT_BASE_BRANCH, value: main} + - {name: PATH, value: /opt/coordinator:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin} + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: [ALL]} + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: home, mountPath: /opt/data} + - {name: runtime-access, mountPath: /pool-access/execution-pool-key, subPath: execution-pool-key, readOnly: true} + - {name: execution-pool-code, mountPath: /opt/coordinator, readOnly: true} + - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py} + - {name: tmp, mountPath: /tmp} + ports: + - {name: execution-pool, containerPort: 9007, protocol: TCP} + startupProbe: + httpGet: {path: /ready, port: execution-pool} + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: + httpGet: {path: /ready, port: execution-pool} + periodSeconds: 10 + livenessProbe: + httpGet: {path: /ready, port: execution-pool} + initialDelaySeconds: 30 + periodSeconds: 30 + resources: + requests: {cpu: 50m, memory: 128Mi} + limits: {cpu: 500m, memory: 512Mi} + volumes: + - name: execution-pool-code + configMap: + name: hermes-execution-pool + defaultMode: 0555 diff --git a/services/hermes/execution-worker-networkpolicy.yaml b/services/hermes/execution-worker-networkpolicy.yaml new file mode 100644 index 00000000..06470b2a --- /dev/null +++ b/services/hermes/execution-worker-networkpolicy.yaml @@ -0,0 +1,84 @@ +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: hermes-execution-worker-isolation + namespace: hermes +spec: + podSelector: + matchLabels: + app: hermes-execution-worker + policyTypes: [Ingress, Egress] + ingress: [] + egress: + - to: + - namespaceSelector: + matchLabels: {kubernetes.io/metadata.name: kube-system} + podSelector: + matchLabels: {k8s-app: kube-dns} + ports: [{protocol: UDP, port: 53}, {protocol: TCP, port: 53}] + - to: + - namespaceSelector: + matchLabels: {kubernetes.io/metadata.name: vault} + podSelector: + matchLabels: {app: vault} + ports: [{protocol: TCP, port: 8200}] + - to: + - podSelector: + matchLabels: {app: hermes-agent} + ports: [{protocol: TCP, port: 9007}] + - to: + - podSelector: + matchLabels: {app: hermes-switchyard} + ports: [{protocol: TCP, port: 9005}] + - to: + - podSelector: + matchLabels: {app: hermes-model-gate} + ports: [{protocol: TCP, port: 8080}] + - to: + - namespaceSelector: + matchLabels: {kubernetes.io/metadata.name: gitea} + podSelector: + matchLabels: {app: gitea} + ports: [{protocol: TCP, port: 3000}] + - to: + - ipBlock: + cidr: 0.0.0.0/0 + except: + - 10.0.0.0/8 + - 100.64.0.0/10 + - 127.0.0.0/8 + - 169.254.0.0/16 + - 172.16.0.0/12 + - 192.168.0.0/16 + - 224.0.0.0/4 + ports: [{protocol: TCP, port: 443}] +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: hermes-execution-pool-ingress + namespace: hermes +spec: + podSelector: + matchLabels: {app: hermes-agent} + policyTypes: [Ingress] + ingress: + - from: + - podSelector: + matchLabels: {app: hermes-execution-worker} + ports: [{protocol: TCP, port: 9007}] +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: hermes-execution-switchyard-ingress + namespace: hermes +spec: + podSelector: + matchLabels: {app: hermes-switchyard} + policyTypes: [Ingress] + ingress: + - from: + - podSelector: + matchLabels: {app: hermes-execution-worker} + ports: [{protocol: TCP, port: 9005}] diff --git a/services/hermes/execution-worker-rbac.yaml b/services/hermes/execution-worker-rbac.yaml new file mode 100644 index 00000000..89ef915e --- /dev/null +++ b/services/hermes/execution-worker-rbac.yaml @@ -0,0 +1,12 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: hermes-execution-worker + namespace: hermes + labels: + app.kubernetes.io/name: hermes-execution-worker + app.kubernetes.io/part-of: hermes +automountServiceAccountToken: false +# Intentionally no RoleBinding or ClusterRoleBinding. The projected, bounded +# token is mounted only into Vault-facing containers and grants no Kubernetes +# API verbs to the model-facing execution worker. diff --git a/services/hermes/execution-worker-statefulset.yaml b/services/hermes/execution-worker-statefulset.yaml new file mode 100644 index 00000000..ac3d9f00 --- /dev/null +++ b/services/hermes/execution-worker-statefulset.yaml @@ -0,0 +1,354 @@ +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: hermes-execution-worker + namespace: hermes + labels: + app: hermes-execution-worker +spec: + serviceName: hermes-execution-worker + replicas: 3 + podManagementPolicy: Parallel + revisionHistoryLimit: 2 + selector: + matchLabels: + app: hermes-execution-worker + updateStrategy: + type: RollingUpdate + template: + metadata: + labels: + app: hermes-execution-worker + app.kubernetes.io/name: hermes-execution-worker + app.kubernetes.io/part-of: hermes + annotations: + ai.bstein.dev/role: fenced-execution-only + ai.bstein.dev/scm-boundary: ordinal-sidecar-with-assignment-bound-branch + ai.bstein.dev/model-policy: Switchyard AUTO with cross-provider fallback + ai.bstein.dev/storage: one durable RWO workspace and provider session home per ordinal + vault.hashicorp.com/agent-inject: "true" + vault.hashicorp.com/role: hermes-execution-worker + vault.hashicorp.com/agent-inject-containers: stage-worker-access + vault.hashicorp.com/agent-service-account-token-volume-name: vault-auth-token + vault.hashicorp.com/agent-inject-secret-execution-pool-key: kv/data/atlas/hermes/agent-tokens + vault.hashicorp.com/agent-inject-perms-execution-pool-key: "0600" + vault.hashicorp.com/agent-inject-template-execution-pool-key: | + {{- with secret "kv/data/atlas/hermes/agent-tokens" -}} + {{ printf "hermes-execution-pool-v1:%s" .Data.data.agent_api_key | sha256sum }} + {{- end }} + vault.hashicorp.com/agent-inject-secret-claude-credentials: kv/data/atlas/hermes/agent-tokens + vault.hashicorp.com/agent-inject-perms-claude-credentials: "0600" + vault.hashicorp.com/agent-inject-template-claude-credentials: | + {{- with secret "kv/data/atlas/hermes/agent-tokens" -}} + {{ .Data.data.claude_credentials_json }} + {{- end }} + vault.hashicorp.com/agent-inject-secret-codex-auth: kv/data/atlas/hermes/agent-tokens + vault.hashicorp.com/agent-inject-perms-codex-auth: "0600" + vault.hashicorp.com/agent-inject-template-codex-auth: | + {{- with secret "kv/data/atlas/hermes/agent-tokens" -}} + {{ .Data.data.codex_auth_json }} + {{- end }} + vault.hashicorp.com/agent-inject-secret-gitea-token: kv/data/atlas/hermes/developer-gitea + vault.hashicorp.com/agent-inject-perms-gitea-token: "0600" + vault.hashicorp.com/agent-inject-template-gitea-token: | + {{- with secret "kv/data/atlas/hermes/developer-gitea" -}} + {{ .Data.data.token }} + {{- end }} + vault.hashicorp.com/agent-inject-secret-gitea-username: kv/data/atlas/hermes/developer-gitea + vault.hashicorp.com/agent-inject-perms-gitea-username: "0600" + vault.hashicorp.com/agent-inject-template-gitea-username: | + {{- with secret "kv/data/atlas/hermes/developer-gitea" -}} + {{ .Data.data.username }} + {{- end }} + vault.hashicorp.com/agent-pre-populate-only: "true" + vault.hashicorp.com/agent-init-first: "true" + vault.hashicorp.com/agent-requests-cpu: 25m + vault.hashicorp.com/agent-requests-mem: 32Mi + vault.hashicorp.com/agent-limits-cpu: 100m + vault.hashicorp.com/agent-limits-mem: 128Mi + spec: + serviceAccountName: hermes-execution-worker + automountServiceAccountToken: false + enableServiceLinks: false + terminationGracePeriodSeconds: 30 + securityContext: + fsGroup: 10000 + fsGroupChangePolicy: OnRootMismatch + seccompProfile: + type: RuntimeDefault + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - {key: kubernetes.io/arch, operator: In, values: [arm64]} + - {key: node-role.kubernetes.io/accelerator, operator: Exists} + - {key: kubernetes.io/hostname, operator: In, values: [titan-20, titan-21]} + - matchExpressions: + - {key: kubernetes.io/arch, operator: In, values: [arm64]} + - {key: hardware, operator: In, values: [rpi5]} + - {key: kubernetes.io/hostname, operator: NotIn, values: [titan-04, titan-08, titan-13, titan-14, titan-17, titan-18, titan-19, titan-22, titan-24]} + preferredDuringSchedulingIgnoredDuringExecution: + - weight: 100 + preference: + matchExpressions: + - {key: node-role.kubernetes.io/accelerator, operator: Exists} + - weight: 50 + preference: + matchExpressions: + - {key: hardware, operator: In, values: [rpi5]} + podAntiAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + - labelSelector: + matchLabels: + app: hermes-execution-worker + topologyKey: kubernetes.io/hostname + topologySpreadConstraints: + - maxSkew: 1 + topologyKey: kubernetes.io/hostname + whenUnsatisfiable: DoNotSchedule + labelSelector: + matchLabels: + app: hermes-execution-worker + initContainers: + - name: stage-worker-access + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/stage_runtime_access.py, execution-worker] + env: + - {name: HERMES_WORKER_ROOT, value: /workspace} + - {name: HERMES_POOL_ACCESS_ROOT, value: /pool-access} + - {name: HERMES_SCM_ACCESS_ROOT, value: /scm-access} + securityContext: + allowPrivilegeEscalation: false + runAsUser: 0 + runAsGroup: 0 + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: workspace, mountPath: /workspace} + - {name: runtime-access, mountPath: /runtime-access} + - {name: pool-access, mountPath: /pool-access} + - {name: scm-access, mountPath: /scm-access} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + resources: + requests: {cpu: 25m, memory: 32Mi} + limits: {cpu: 100m, memory: 64Mi} + - name: install-provider-clis + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/bin/sh, -ec] + args: + - | + tools=/worker-data/tools + mkdir -p "${tools}/bin" + if [ ! -f "${tools}/.cli-versions-0.147.0-2.1.226" ]; then + npm install --global --omit=dev --no-audit --no-fund --prefix "${tools}" \ + @openai/codex@0.147.0 @anthropic-ai/claude-code@2.1.226 + touch "${tools}/.cli-versions-0.147.0-2.1.226" + fi + test -x "${tools}/bin/codex" + test -x "${tools}/bin/claude" + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: [ALL]} + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: tools, mountPath: /worker-data/tools} + resources: + requests: {cpu: 100m, memory: 128Mi} + limits: {cpu: "1", memory: 1Gi} + containers: + - name: execution-worker + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/execution_pool_worker.py] + env: + - {name: HERMES_HOME, value: /worker-data} + - {name: HERMES_WORKER_ROOT, value: /workspace} + - {name: HOME, value: /worker-data/home} + - {name: CODEX_HOME, value: /runtime-access/codex} + - {name: CLAUDE_CONFIG_DIR, value: /runtime-access/claude} + - {name: HERMES_AUTO_ROUTER_PROFILE, value: agent} + - {name: PYTHONPATH, value: /opt/hermes} + - {name: PATH, value: /worker-data/tools/bin:/opt/coordinator:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin} + - name: HERMES_WORKER_ORDINAL + valueFrom: + fieldRef: + fieldPath: metadata.labels['apps.kubernetes.io/pod-index'] + - name: HERMES_WORKER_NODE + valueFrom: + fieldRef: + fieldPath: spec.nodeName + startupProbe: + exec: + command: [/bin/sh, -ec, "test -w /workspace && test -w /runtime-access/codex/auth.json && test -w /runtime-access/claude/.credentials.json"] + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: + exec: + command: [/bin/sh, -ec, "test -w /workspace && test -w /runtime-access/codex/auth.json"] + periodSeconds: 10 + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: [ALL]} + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: workspace, mountPath: /workspace} + - {name: worker-data, mountPath: /worker-data} + - {name: tools, mountPath: /worker-data/tools, readOnly: true} + - {name: runtime-access, mountPath: /runtime-access} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: tmp, mountPath: /tmp} + resources: + requests: {cpu: "1", memory: 2Gi, ephemeral-storage: 2Gi} + limits: {cpu: "5", memory: 6Gi, ephemeral-storage: 8Gi} + - name: execution-client + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/execution_pool_client.py] + env: + - {name: HERMES_EXECUTION_POOL_KEY_FILE, value: /pool-access/execution-pool-key} + - name: HERMES_WORKER_ORDINAL + valueFrom: + fieldRef: + fieldPath: metadata.labels['apps.kubernetes.io/pod-index'] + ports: + - {name: pool-client, containerPort: 9009, protocol: TCP} + startupProbe: + httpGet: {path: /ready, port: pool-client} + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: + httpGet: {path: /ready, port: pool-client} + periodSeconds: 10 + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: [ALL]} + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: pool-access, mountPath: /pool-access, readOnly: true} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: tmp, mountPath: /tmp} + resources: + requests: {cpu: 25m, memory: 64Mi} + limits: {cpu: 250m, memory: 256Mi} + - name: scm-boundary + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/execution_pool_scm.py] + env: + - {name: HERMES_WORKER_ROOT, value: /workspace} + - {name: HERMES_EXECUTION_POOL_KEY_FILE, value: /pool-access/execution-pool-key} + - {name: HERMES_GITEA_TOKEN_FILE, value: /scm-access/gitea-token} + - {name: HERMES_GITEA_USERNAME_FILE, value: /scm-access/gitea-username} + - {name: HERMES_SCM_STATE_ROOT, value: /scm-state} + - name: HERMES_WORKER_ORDINAL + valueFrom: + fieldRef: + fieldPath: metadata.labels['apps.kubernetes.io/pod-index'] + ports: + - {name: scm, containerPort: 9008, protocol: TCP} + startupProbe: + httpGet: {path: /ready, port: scm} + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: + httpGet: {path: /ready, port: scm} + periodSeconds: 10 + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: [ALL]} + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: workspace, mountPath: /workspace} + - {name: pool-access, mountPath: /pool-access, readOnly: true} + - {name: scm-access, mountPath: /scm-access, readOnly: true} + - {name: scm-state, mountPath: /scm-state} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: tmp, mountPath: /tmp} + resources: + requests: {cpu: 100m, memory: 128Mi} + limits: {cpu: "1", memory: 512Mi} + - name: credential-sync + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/sync_runtime_credentials.py] + env: + - {name: HERMES_CREDENTIAL_SYNC_VAULT_ROLE, value: hermes-execution-credential-sync} + - {name: HERMES_CREDENTIAL_SYNC_INTERVAL, value: "300"} + securityContext: + allowPrivilegeEscalation: false + capabilities: {drop: [ALL]} + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: runtime-access, mountPath: /runtime-access} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: vault-auth-token, mountPath: /var/run/secrets/kubernetes.io/serviceaccount, readOnly: true} + - {name: tmp, mountPath: /tmp} + resources: + requests: {cpu: 25m, memory: 64Mi} + limits: {cpu: 250m, memory: 256Mi} + volumes: + - name: worker-data + emptyDir: {sizeLimit: 128Mi} + - name: tools + emptyDir: {sizeLimit: 1Gi} + - name: runtime-access + emptyDir: {medium: Memory, sizeLimit: 128Mi} + - name: pool-access + emptyDir: {medium: Memory, sizeLimit: 1Mi} + - name: scm-access + emptyDir: {medium: Memory, sizeLimit: 1Mi} + - name: scm-state + emptyDir: {sizeLimit: 256Mi} + - name: coordinator + configMap: + name: hermes-execution-pool + defaultMode: 0555 + - name: tmp + emptyDir: {sizeLimit: 2Gi} + - name: vault-auth-token + projected: + defaultMode: 0600 + sources: + - serviceAccountToken: + audience: vault + expirationSeconds: 3600 + path: token + - configMap: + name: kube-root-ca.crt + items: + - {key: ca.crt, path: ca.crt} + - downwardAPI: + items: + - {path: namespace, fieldRef: {fieldPath: metadata.namespace}} + volumeClaimTemplates: + - metadata: + name: workspace + labels: + app: hermes-execution-worker + spec: + accessModes: [ReadWriteOnce] + storageClassName: astreae + resources: + requests: + storage: 30Gi diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 3b9c2345..1f3d9cfe 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -41,6 +41,12 @@ resources: - oauth2-proxy.yaml - agent-certificate.yaml - agent-ingress.yaml + - execution-worker-rbac.yaml + - execution-worker-statefulset.yaml + - execution-worker-networkpolicy.yaml + +patches: + - path: execution-coordinator-patch.yaml configMapGenerator: - name: hermes-chat-oauth-templates @@ -55,6 +61,21 @@ configMapGenerator: - OPERATOR-RUNBOOK.md=NOTES.md options: disableNameSuffixHash: true + - name: hermes-execution-pool + namespace: hermes + files: + - cli_lane_goal.py=scripts/cli_lane_goal.py + - cli_lane_runner.py=scripts/cli_lane_runner.py + - execution_pool_askpass.sh=scripts/execution_pool_askpass.sh + - execution_pool_protocol.py=scripts/execution_pool_protocol.py + - execution_pool_coordinator.py=scripts/execution_pool_coordinator.py + - execution_pool_client.py=scripts/execution_pool_client.py + - execution_pool_worker.py=scripts/execution_pool_worker.py + - execution_pool_scm.py=scripts/execution_pool_scm.py + - stage_runtime_access.py=scripts/stage_runtime_access.py + - sync_runtime_credentials.py=scripts/sync_runtime_credentials.py + options: + disableNameSuffixHash: true - name: hermes-coordinator namespace: hermes files: diff --git a/services/hermes/scripts/execution_pool_askpass.sh b/services/hermes/scripts/execution_pool_askpass.sh new file mode 100755 index 00000000..8e3758ae --- /dev/null +++ b/services/hermes/scripts/execution_pool_askpass.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env sh +set -eu + +case "${1:-}" in + *Username*) + if [ -s "${HERMES_SCM_USERNAME_FILE}" ]; then + tr -d '\r\n' <"${HERMES_SCM_USERNAME_FILE}" + else + printf '%s' hermes-automation + fi + printf '\n' + ;; + *Password*) + test -s "${HERMES_SCM_PASSWORD_FILE}" + tr -d '\r\n' <"${HERMES_SCM_PASSWORD_FILE}" + printf '\n' + ;; + *) exit 1 ;; +esac diff --git a/services/hermes/scripts/execution_pool_client.py b/services/hermes/scripts/execution_pool_client.py new file mode 100644 index 00000000..76c22953 --- /dev/null +++ b/services/hermes/scripts/execution_pool_client.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +"""Ordinal-local signing boundary for a model-facing Hermes worker.""" + +from __future__ import annotations + +import json +import os +import threading +import urllib.error +import urllib.request +from http.server import BaseHTTPRequestHandler +from pathlib import Path +from typing import Any + +from execution_pool_protocol import ( + MAX_WIRE_BYTES, + BoundedHTTPServer, + ProtocolError, + canonical_json, + parse_wire, + read_key, + sign_envelope, + verify_envelope, +) + + +KEY_PATH = Path(os.environ.get("HERMES_EXECUTION_POOL_KEY_FILE", "/pool-access/execution-pool-key")) +COORDINATOR = os.environ.get( + "HERMES_EXECUTION_POOL_URL", "http://hermes-execution-pool.hermes.svc.cluster.local:9007" +).rstrip("/") +ORDINAL = int(os.environ.get("HERMES_WORKER_ORDINAL", "-1")) +PORT = int(os.environ.get("HERMES_EXECUTION_CLIENT_PORT", "9009")) + + +def _binding(value: dict[str, Any]) -> dict[str, Any]: + return {name: value[name] for name in ( + "board", "task_id", "run_id", "worker_ordinal", "attempt" + )} + + +class ClientBoundary: + """Hold the master key and fence local requests to the current ordinal run.""" + + def __init__(self, key: bytes): + self.key = key + self.current: dict[str, Any] | None = None + self.lock = threading.RLock() + + def _post(self, path: str, envelope: dict[str, Any]) -> dict[str, Any]: + request = urllib.request.Request( + COORDINATOR + path, data=canonical_json(envelope), method="POST", + headers={"Content-Type": "application/json", "Cache-Control": "no-store"}, + ) + with urllib.request.urlopen(request, timeout=60) as response: + body = response.read(MAX_WIRE_BYTES + 1) + if len(body) > MAX_WIRE_BYTES: + raise ProtocolError("coordinator response exceeds the wire limit") + value = json.loads(body) + return verify_envelope(self.key, value) + + def poll(self) -> dict[str, Any]: + poll_binding = { + "board": "", "task_id": "", "run_id": "", + "worker_ordinal": ORDINAL, "attempt": 0, + } + response = self._post( + "/v1/poll", sign_envelope(self.key, "poll", poll_binding, {"ready": True}) + ) + with self.lock: + if response["kind"] == "ack": + self.current = None + return {"assignment": None} + if response["kind"] != "assignment" or response["worker_ordinal"] != ORDINAL: + raise ProtocolError("coordinator returned a foreign assignment") + self.current = _binding(response) + return {"assignment": response} + + def forward(self, kind: str, request: dict[str, Any]) -> dict[str, Any]: + supplied = request.get("binding") + payload = request.get("payload") + if not isinstance(supplied, dict) or not isinstance(payload, dict): + raise ProtocolError("local request binding and payload must be objects") + with self.lock: + if self.current is None or supplied != self.current: + raise ProtocolError("local request does not own the current assignment") + binding = dict(self.current) + response = self._post( + f"/v1/{kind}", sign_envelope(self.key, kind, binding, payload) + ) + if response["kind"] != "ack" or _binding(response) != binding: + raise ProtocolError("coordinator acknowledgement binding changed") + if kind == "result" and response["payload"].get("accepted"): + with self.lock: + self.current = None + return {"ack": response["payload"]} + + +def handler_factory(boundary: ClientBoundary) -> type[BaseHTTPRequestHandler]: + class Handler(BaseHTTPRequestHandler): + server_version = "hermes-execution-client/1" + + def _reply(self, status: int, value: dict[str, Any]) -> None: + body = canonical_json(value) + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # noqa: N802 + self._reply(200, {"ready": True}) if self.path == "/ready" else self._reply(404, {"error": "not found"}) + + def do_POST(self) -> None: # noqa: N802 + try: + length = int(self.headers.get("Content-Length", "0")) + request = parse_wire(self.rfile.read(length)) if 0 < length <= MAX_WIRE_BYTES else None + if not request or set(request) - {"operation", "binding", "payload"}: + raise ProtocolError("invalid local client request") + operation = str(request.get("operation") or "") + if operation == "poll": + result = boundary.poll() + elif operation in {"heartbeat", "result"}: + result = boundary.forward(operation, request) + else: + raise ProtocolError("unsupported local client operation") + self._reply(200, result) + except (ProtocolError, OSError, ValueError, urllib.error.URLError) as error: + self._reply(409, {"error": str(error)[:2000]}) + + def log_message(self, _format: str, *_arguments: Any) -> None: + return + + return Handler + + +def main() -> int: + if ORDINAL not in range(3): + raise SystemExit("HERMES_WORKER_ORDINAL must be 0, 1, or 2") + key = read_key(KEY_PATH) + BoundedHTTPServer( + ("0.0.0.0", PORT), handler_factory(ClientBoundary(key)), max_workers=4 + ).serve_forever() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/execution_pool_coordinator.py b/services/hermes/scripts/execution_pool_coordinator.py new file mode 100644 index 00000000..8a128778 --- /dev/null +++ b/services/hermes/scripts/execution_pool_coordinator.py @@ -0,0 +1,491 @@ +#!/usr/bin/env python3 +"""Coordinator-only Kanban claim/finalize bridge for three Hermes workers.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sqlite3 +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler +from pathlib import Path +from typing import Any + +import cli_lane_goal +import cli_lane_runner +from execution_pool_protocol import ( + MAX_ACTIVITY_BYTES, + MAX_WIRE_BYTES, + BoundedHTTPServer, + PoolStore, + ProtocolError, + canonical_json, + parse_wire, + read_key, + sign_envelope, + verify_envelope, +) + + +DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data")) +STATE_ROOT = DATA_ROOT / "execution-pool" +KEY_PATH = Path(os.environ.get("HERMES_EXECUTION_POOL_KEY_FILE", "/runtime-access/execution-pool-key")) +PORT = int(os.environ.get("HERMES_EXECUTION_POOL_PORT", "9007")) +RETENTION_SECONDS = int(os.environ.get("HERMES_EXECUTION_POOL_RETENTION_SECONDS", "1209600")) +ATLAS_REPO = re.compile(r"^https://scm\.bstein\.dev/atlas/[A-Za-z0-9_.-]+\.git$") +BRANCH = re.compile(r"^(?:feature|fix|chore|docs|test|refactor)/[A-Za-z0-9][A-Za-z0-9._/-]{0,119}$") +DEFAULT_REPO = os.environ.get( + "HERMES_EXECUTION_DEFAULT_REPO_URL", + "https://scm.bstein.dev/atlas/titan-iac.git", +) +DEFAULT_BASE_BRANCH = os.environ.get("HERMES_EXECUTION_DEFAULT_BASE_BRANCH", "main") +REDACTIONS = ( + re.compile(r"(?i)bearer\s+[A-Za-z0-9._~+/-]{12,}"), + re.compile( + r"(?i)(authorization|token|secret|password|api[_-]?key|" + r"refresh[_-]?token|access[_-]?token)[\"']?\s*[:=]\s*[\"']?[^\s\"',;}]+" + ), + re.compile(r"\b(?:sk-ant-|sk-proj-|rt_)[A-Za-z0-9_-]{12,}\b"), + re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b"), + re.compile(r"(?i)https://[^/@\s]+@scm\.bstein\.dev"), +) + + +def _task_value(task: Any, name: str, default: Any = None) -> Any: + return getattr(task, name, default) + + +def resolve_scm(task: Any) -> tuple[str, str, str]: + """Resolve reviewed SCM metadata without executing in a legacy worktree.""" + raw = str(_task_value(task, "workspace_path", "") or "") + if raw: + raise RuntimeError( + "task owns a legacy local workspace; preserve or commit its tracked and " + "untracked files before distributed assignment" + ) + remote = str(_task_value(task, "repo_url", "") or DEFAULT_REPO) + if not ATLAS_REPO.fullmatch(remote) or "@" in remote: + raise RuntimeError("task origin is outside the reviewed Atlas SCM boundary") + task_id = str(_task_value(task, "id", "task")) + branch = str(_task_value(task, "branch_name", "") or "") + if not branch: + branch = f"feature/hermes-{re.sub(r'[^A-Za-z0-9._-]', '-', task_id)[:80]}" + if not BRANCH.fullmatch(branch) or ".." in branch or "//" in branch: + raise RuntimeError("task feature branch is outside the SCM branch policy") + base_branch = str( + _task_value(task, "base_branch", "") or DEFAULT_BASE_BRANCH + ) + if ( + not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,119}", base_branch) + or ".." in base_branch + or "//" in base_branch + ): + raise RuntimeError("repository default branch is outside the SCM policy") + return remote, branch, base_branch + + +def assignment_payload(kanban_db: Any, connection: Any, task: Any) -> dict[str, Any]: + context = kanban_db.build_worker_context(connection, str(_task_value(task, "id"))) + if not isinstance(context, str): + context = json.dumps(context, default=str, sort_keys=True) + encoded = context.encode("utf-8") + if len(encoded) > 32 * 1024: + raise RuntimeError("Kanban worker context exceeds the 32KiB assignment limit") + repo_url, branch, base_branch = resolve_scm(task) + runtime = int(_task_value(task, "max_runtime_seconds", 0) or 12 * 60 * 60) + runtime = max(60, min(runtime, 12 * 60 * 60)) + return { + "context": context, + "assignee": str(_task_value(task, "assignee", "cli-auto") or "cli-auto"), + "repo_url": repo_url, + "branch": branch, + "base_branch": base_branch, + "max_runtime_seconds": runtime, + "deadline_unix": int(time.time()) + runtime, + "goal_mode": bool(_task_value(task, "goal_mode", False)), + "goal_max_turns": max(1, min(int(_task_value(task, "goal_max_turns", 1) or 1), 12)), + } + + +def sanitize_activity(value: Any) -> str: + """Bound and redact worker output before it reaches the shared UI log.""" + text = str(value or "").replace("\x00", "") + text = "".join(character for character in text if character in "\n\t" or ord(character) >= 32) + for pattern in REDACTIONS: + text = pattern.sub("[REDACTED]", text) + return text.encode("utf-8")[:MAX_ACTIVITY_BYTES].decode("utf-8", "ignore") + + +def _append_activity(kanban_db: Any, envelope: dict[str, Any]) -> None: + payload = envelope["payload"] + if not isinstance(payload, dict): + raise ProtocolError("heartbeat payload must be an object") + activity = sanitize_activity(payload.get("activity")) + if not activity: + return + path = Path(kanban_db.worker_log_path(envelope["task_id"], board=envelope["board"])) + path.parent.mkdir(parents=True, exist_ok=True) + if path.is_symlink(): + raise ProtocolError("Kanban activity log must not be a symlink") + flags = os.O_APPEND | os.O_CREAT | os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags, 0o600) + try: + os.write(descriptor, activity.encode("utf-8")) + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _append_terminal_activity(kanban_db: Any, record: dict[str, Any]) -> None: + """Append one terminal batch exactly once across result retries/restarts.""" + payload = record.get("result") + activity = payload.get("final_activity") if isinstance(payload, dict) else "" + activity = sanitize_activity(activity) + if not activity: + return + marker = f"\n[execution-pool-result:{record['result_digest']}]\n" + path = Path(kanban_db.worker_log_path(record["task_id"], board=record["board"])) + if path.is_symlink(): + raise ProtocolError("Kanban activity log must not be a symlink") + if path.exists(): + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + try: + size = os.fstat(descriptor).st_size + os.lseek(descriptor, max(0, size - 64 * 1024), os.SEEK_SET) + if marker.encode() in os.read(descriptor, 64 * 1024): + return + finally: + os.close(descriptor) + _append_activity( + kanban_db, + {**record, "payload": {"activity": marker + activity}}, + ) + + +class Coordinator: + """Own all transitions between durable assignments and Hermes Kanban.""" + + def __init__(self, key: bytes, store: PoolStore): + self.key = key + self.store = store + self._kanban_lock = threading.RLock() + + @staticmethod + def _binding(record: dict[str, Any]) -> dict[str, Any]: + return {name: record[name] for name in ( + "board", "task_id", "run_id", "worker_ordinal", "attempt" + )} + + def poll(self, envelope: dict[str, Any]) -> dict[str, Any]: + verified = verify_envelope(self.key, envelope, expected_kind="poll") + record = self.store.offer(int(verified["worker_ordinal"])) + if record is None: + binding = {**self._binding({ + "board": "", "task_id": "", "run_id": "", + "worker_ordinal": verified["worker_ordinal"], "attempt": 0, + })} + return sign_envelope(self.key, "ack", binding, {"assignment": None}) + return sign_envelope( + self.key, "assignment", self._binding(record), record["payload"] + ) + + def heartbeat(self, envelope: dict[str, Any]) -> dict[str, Any]: + verified = verify_envelope(self.key, envelope, expected_kind="heartbeat") + accepted, duplicate = self.store.heartbeat(verified) + from hermes_cli import kanban_db + + payload = verified["payload"] + note = sanitize_activity(payload.get("note") if isinstance(payload, dict) else "")[:1000] + with self._kanban_lock, kanban_db.scoped_current_board(verified["board"]): + connection = kanban_db.connect(board=verified["board"]) + try: + alive = kanban_db.heartbeat_worker( + connection, verified["task_id"], + note=note or "distributed worker active", + expected_run_id=verified["run_id"], + ) + if not alive: + raise ProtocolError("Kanban run no longer owns this worker") + if not duplicate: + _append_activity(kanban_db, verified) + finally: + connection.close() + return sign_envelope( + self.key, "ack", self._binding(verified), + {"accepted": accepted, "duplicate": duplicate}, + ) + + def result(self, envelope: dict[str, Any]) -> dict[str, Any]: + verified = verify_envelope(self.key, envelope, expected_kind="result") + record, duplicate = self.store.accept_result(verified) + if record.get("state") == "result": + self.finalize(record) + return sign_envelope( + self.key, "ack", self._binding(verified), + {"accepted": True, "duplicate": duplicate}, + ) + + def finalize(self, record: dict[str, Any]) -> None: + """Apply a result only if this exact run remains authoritative.""" + from hermes_cli import kanban_db + + binding = self._binding(record) + payload = record.get("result") + if not isinstance(payload, dict): + raise ProtocolError("result payload must be an object") + structured = payload.get("structured") + if not isinstance(structured, dict): + structured = {} + with self._kanban_lock, kanban_db.scoped_current_board(binding["board"]): + connection = kanban_db.connect(board=binding["board"]) + try: + task = kanban_db.get_task(connection, binding["task_id"]) + if task is None or _task_value(task, "current_run_id") != binding["run_id"]: + self.store.finalize(binding, "stale") + return + _append_terminal_activity(kanban_db, record) + metadata = { + "executor": "distributed-execution-pool", + "worker_ordinal": binding["worker_ordinal"], + "attempt": binding["attempt"], + "node": str(payload.get("node") or "unknown")[:128], + "route": payload.get("route") if isinstance(payload.get("route"), dict) else {}, + "provider_sessions": payload.get("provider_sessions") + if isinstance(payload.get("provider_sessions"), dict) else {}, + "changed_files": structured.get("changed_files", []), + "tests_run": structured.get("tests_run", []), + "artifacts": structured.get("artifacts", []), + "findings": structured.get("findings", []), + "blockers": structured.get("blockers", []), + } + problem = cli_lane_goal.unfinished_result_reason(structured) + if ( + structured.get("status") == "completed" + and int(payload.get("returncode", 1)) == 0 + and problem is None + ): + changed = kanban_db.complete_task( + connection, binding["task_id"], + result=json.dumps(structured, sort_keys=True), + summary=str(structured.get("summary") or "Completed"), + metadata=metadata, expected_run_id=binding["run_id"], + ) + else: + reason = problem or "; ".join(map(str, structured.get("blockers", []))) + reason = reason or str(structured.get("summary") or "worker failed") + changed = kanban_db.block_task( + connection, binding["task_id"], reason=reason, + kind="transient" if payload.get("capacity_failure") else "capability", + expected_run_id=binding["run_id"], + ) + self.store.finalize(binding, "finalized" if changed else "stale") + finally: + connection.close() + + def recover_results(self) -> None: + for record in self.store.pending_results(): + try: + self.finalize(record) + except (OSError, sqlite3.Error) as error: + print(f"result recovery deferred: {error}", file=sys.stderr, flush=True) + + def reconcile(self) -> None: + """Recover the narrow claim/assignment crash gaps without double execution.""" + from hermes_cli import kanban_db + + active = self.store.active_assignments() + active_runs = { + (record["board"], record["task_id"], record["run_id"]) + for record in active + } + for record in active: + try: + with kanban_db.scoped_current_board(record["board"]): + connection = kanban_db.connect(board=record["board"]) + try: + task = kanban_db.get_task(connection, record["task_id"]) + current = str(_task_value(task, "current_run_id", "") or "") if task else "" + status = str(_task_value(task, "status", "") or "") if task else "" + finally: + connection.close() + if current != record["run_id"] or status != "running": + self.store.finalize(record, "stale") + except (OSError, sqlite3.Error): + continue + ordinals = self.store.available_ordinals() + if not ordinals: + return + for raw_board in kanban_db.list_boards(include_archived=False): + board = cli_lane_runner._board_slug(raw_board) + if not board: + continue + with kanban_db.scoped_current_board(board): + connection = kanban_db.connect(board=board) + try: + tasks = kanban_db.list_tasks(connection) + for task in tasks: + task_id = str(_task_value(task, "id", "") or "") + run_id = str(_task_value(task, "current_run_id", "") or "") + assignee = str(_task_value(task, "assignee", "") or "") + if ( + not ordinals + or not task_id + or not run_id + or str(_task_value(task, "status", "")) != "running" + or not assignee.startswith("cli-") + or (board, task_id, run_id) in active_runs + ): + continue + try: + payload = assignment_payload(kanban_db, connection, task) + kanban_db.set_branch_name(connection, task_id, payload["branch"]) + except Exception as error: + kanban_db.block_task( + connection, task_id, + reason=( + "Distributed assignment recovery failed: " + f"{type(error).__name__}: {error}" + ), + kind="capability", expected_run_id=run_id, + ) + continue + ordinal = ordinals.pop(0) + self.store.add( + { + "board": board, "task_id": task_id, + "run_id": run_id, "worker_ordinal": ordinal, + "attempt": 1, + }, + payload, + ) + finally: + connection.close() + + def dispatch(self) -> None: + """Claim up to the free ordinal count and materialize assignments.""" + from hermes_cli import kanban_db + + ordinals = self.store.available_ordinals() + if not ordinals: + return + claimed = cli_lane_runner.claim_ready(set(), len(ordinals)) + for ordinal, (board, task_id) in zip(ordinals, claimed, strict=False): + run_id = "" + try: + with kanban_db.scoped_current_board(board): + connection = kanban_db.connect(board=board) + try: + task = kanban_db.get_task(connection, task_id) + if task is None: + continue + run_id = str(_task_value(task, "current_run_id", "") or "") + payload = assignment_payload(kanban_db, connection, task) + kanban_db.set_branch_name(connection, task_id, payload["branch"]) + finally: + connection.close() + binding = { + "board": board, "task_id": task_id, "run_id": run_id, + "worker_ordinal": ordinal, "attempt": 1, + } + self.store.add(binding, payload) + except Exception as error: + reason = f"Distributed assignment preparation failed: {type(error).__name__}: {error}" + with kanban_db.scoped_current_board(board): + connection = kanban_db.connect(board=board) + try: + kanban_db.block_task( + connection, task_id, reason=reason, kind="capability", + expected_run_id=run_id or None, + ) + finally: + connection.close() + + +def handler_factory(coordinator: Coordinator) -> type[BaseHTTPRequestHandler]: + class Handler(BaseHTTPRequestHandler): + server_version = "hermes-execution-pool/1" + + def _reply(self, status: int, value: dict[str, Any]) -> None: + body = canonical_json(value) + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # noqa: N802 + if self.path != "/ready": + self._reply(404, {"error": "not found"}) + return + try: + coordinator.store.available_ordinals() + self._reply(200, {"ready": True}) + except (OSError, sqlite3.Error): + self._reply(503, {"ready": False}) + + def do_POST(self) -> None: # noqa: N802 + try: + length = int(self.headers.get("Content-Length", "0")) + if length <= 0 or length > MAX_WIRE_BYTES: + raise ProtocolError("invalid content length") + envelope = parse_wire(self.rfile.read(length)) + routes = { + "/v1/poll": coordinator.poll, + "/v1/heartbeat": coordinator.heartbeat, + "/v1/result": coordinator.result, + } + if self.path not in routes: + self._reply(404, {"error": "not found"}) + return + self._reply(200, routes[self.path](envelope)) + except ProtocolError as error: + self._reply(409, {"error": str(error)}) + except Exception as error: + print(f"pool request failed: {type(error).__name__}: {error}", file=sys.stderr, flush=True) + self._reply(503, {"error": "coordinator unavailable"}) + + def log_message(self, _format: str, *_arguments: Any) -> None: + return + + return Handler + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--once", action="store_true") + args = parser.parse_args() + key = read_key(KEY_PATH) + store = PoolStore(STATE_ROOT / "assignments.db") + coordinator = Coordinator(key, store) + coordinator.recover_results() + coordinator.reconcile() + coordinator.dispatch() + if args.once: + return 0 + server = BoundedHTTPServer( + ("0.0.0.0", PORT), handler_factory(coordinator), max_workers=8 + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + while True: + for operation in ( + coordinator.recover_results, + coordinator.reconcile, + coordinator.dispatch, + lambda: store.garbage_collect(RETENTION_SECONDS), + ): + try: + operation() + except Exception as error: + print(f"pool maintenance deferred: {type(error).__name__}: {error}", file=sys.stderr, flush=True) + time.sleep(5) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/execution_pool_protocol.py b/services/hermes/scripts/execution_pool_protocol.py new file mode 100644 index 00000000..8659a9ba --- /dev/null +++ b/services/hermes/scripts/execution_pool_protocol.py @@ -0,0 +1,441 @@ +#!/usr/bin/env python3 +"""Authenticated, bounded, restart-safe Hermes execution-pool protocol.""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import re +import sqlite3 +import stat +import threading +import time +import uuid +from http.server import ThreadingHTTPServer +from pathlib import Path +from typing import Any + + +MAX_WIRE_BYTES = 64 * 1024 +MAX_ACTIVITY_BYTES = 12 * 1024 +MAX_CLOCK_SKEW = 30 +MAX_ENVELOPE_LIFETIME = 300 +IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") +KINDS = frozenset({"poll", "assignment", "heartbeat", "result", "ack"}) + + +class ProtocolError(ValueError): + """A request failed the authenticated pool contract.""" + + +class BoundedHTTPServer(ThreadingHTTPServer): + """Bound concurrent requests and slow clients on internal pool channels.""" + + daemon_threads = True + request_queue_size = 8 + + def __init__(self, *args: Any, max_workers: int = 8, **kwargs: Any): + self._slots = threading.BoundedSemaphore(max(1, min(max_workers, 16))) + super().__init__(*args, **kwargs) + + def get_request(self) -> tuple[Any, Any]: + request, address = super().get_request() + request.settimeout(15) + return request, address + + def process_request(self, request: Any, client_address: Any) -> None: + if not self._slots.acquire(blocking=False): + self.shutdown_request(request) + return + try: + super().process_request(request, client_address) + except Exception: + self._slots.release() + raise + + def process_request_thread(self, request: Any, client_address: Any) -> None: + try: + super().process_request_thread(request, client_address) + finally: + self._slots.release() + + +def canonical_json(value: Any) -> bytes: + """Encode one value deterministically for digests and signatures.""" + return json.dumps( + value, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + + +def atomic_json(path: Path, value: dict[str, Any], mode: int = 0o600) -> None: + """Durably replace one bounded pool document without following symlinks.""" + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + if path.parent.is_symlink(): + raise ProtocolError("pool state directory must not be a symlink") + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(temporary, flags, mode) + try: + remaining = memoryview(json.dumps(value, indent=2, sort_keys=True).encode() + b"\n") + while remaining: + remaining = remaining[os.write(descriptor, remaining) :] + os.fsync(descriptor) + os.fchmod(descriptor, mode) + finally: + os.close(descriptor) + try: + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def payload_digest(payload: Any) -> str: + return hashlib.sha256(canonical_json(payload)).hexdigest() + + +def read_key(path: Path) -> bytes: + """Read a private regular file without following a final symlink.""" + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as error: + raise ProtocolError(f"pool key is unavailable: {error}") from error + try: + info = os.fstat(descriptor) + if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o077: + raise ProtocolError("pool key must be a private regular file") + value = os.read(descriptor, 4097).strip() + finally: + os.close(descriptor) + if len(value) < 32 or len(value) > 4096: + raise ProtocolError("pool key length is outside the safe range") + return value + + +def _identifier(name: str, value: Any, *, allow_empty: bool = False) -> str: + text = str(value or "") + if allow_empty and not text: + return text + if not IDENTIFIER.fullmatch(text): + raise ProtocolError(f"invalid {name}") + return text + + +def sign_envelope( + key: bytes, + kind: str, + binding: dict[str, Any], + payload: Any, + *, + now: int | None = None, + lifetime: int = 120, + delivery_id: str | None = None, +) -> dict[str, Any]: + """Bind a message to the exact Kanban run, ordinal, and attempt.""" + current = int(time.time()) if now is None else int(now) + lifetime = max(1, min(int(lifetime), MAX_ENVELOPE_LIFETIME)) + envelope = { + "version": 1, + "kind": kind, + "board": str(binding.get("board") or ""), + "task_id": str(binding.get("task_id") or ""), + "run_id": str(binding.get("run_id") or ""), + "worker_ordinal": int(binding.get("worker_ordinal", -1)), + "attempt": int(binding.get("attempt", 0)), + "delivery_id": delivery_id or str(uuid.uuid4()), + "issued_at": current, + "expires_at": current + lifetime, + "payload_digest": payload_digest(payload), + "payload": payload, + } + if kind not in KINDS: + raise ProtocolError("unsupported message kind") + unsigned = canonical_json(envelope) + if len(unsigned) > MAX_WIRE_BYTES: + raise ProtocolError("pool message exceeds the wire limit") + envelope["signature"] = hmac.new(key, unsigned, hashlib.sha256).hexdigest() + return envelope + + +def verify_envelope( + key: bytes, + envelope: Any, + *, + expected_kind: str | None = None, + now: int | None = None, +) -> dict[str, Any]: + """Verify structure, lifetime, digest, and HMAC before using a message.""" + if not isinstance(envelope, dict) or len(canonical_json(envelope)) > MAX_WIRE_BYTES: + raise ProtocolError("invalid or oversized pool message") + required = { + "version", "kind", "board", "task_id", "run_id", "worker_ordinal", + "attempt", "delivery_id", "issued_at", "expires_at", "payload_digest", + "payload", "signature", + } + if set(envelope) != required or envelope.get("version") != 1: + raise ProtocolError("pool message fields do not match version 1") + kind = str(envelope["kind"]) + if kind not in KINDS or (expected_kind and kind != expected_kind): + raise ProtocolError("unexpected message kind") + empty_binding = kind in {"poll", "ack"} + _identifier("board", envelope["board"], allow_empty=empty_binding) + _identifier("task_id", envelope["task_id"], allow_empty=empty_binding) + _identifier("run_id", envelope["run_id"], allow_empty=empty_binding) + _identifier("delivery_id", envelope["delivery_id"]) + try: + ordinal = int(envelope["worker_ordinal"]) + attempt = int(envelope["attempt"]) + issued = int(envelope["issued_at"]) + expires = int(envelope["expires_at"]) + except (TypeError, ValueError) as error: + raise ProtocolError("invalid numeric binding") from error + if ordinal not in range(3) or attempt < 0: + raise ProtocolError("worker binding is outside the pool") + current = int(time.time()) if now is None else int(now) + if issued > current + MAX_CLOCK_SKEW or expires < current - MAX_CLOCK_SKEW: + raise ProtocolError("pool message is outside its validity window") + if expires <= issued or expires - issued > MAX_ENVELOPE_LIFETIME: + raise ProtocolError("pool message lifetime is invalid") + if envelope["payload_digest"] != payload_digest(envelope["payload"]): + raise ProtocolError("pool payload digest mismatch") + signature = str(envelope["signature"]) + unsigned = dict(envelope) + unsigned.pop("signature") + expected = hmac.new(key, canonical_json(unsigned), hashlib.sha256).hexdigest() + if not hmac.compare_digest(signature, expected): + raise ProtocolError("pool message authentication failed") + return envelope + + +def parse_wire(body: bytes) -> dict[str, Any]: + if not body or len(body) > MAX_WIRE_BYTES: + raise ProtocolError("empty or oversized request") + try: + value = json.loads(body) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise ProtocolError("malformed JSON request") from error + if not isinstance(value, dict): + raise ProtocolError("request must be a JSON object") + return value + + +class PoolStore: + """Coordinator-owned durable assignments; never stores provider secrets.""" + + def __init__(self, path: Path, lease_seconds: int = 90): + self.path = path + self.lease_seconds = max(60, min(int(lease_seconds), 600)) + self._lock = threading.RLock() + path.parent.mkdir(parents=True, exist_ok=True) + self._initialize() + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self.path, timeout=10, isolation_level=None) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA synchronous=FULL") + connection.execute("PRAGMA busy_timeout=10000") + return connection + + def _initialize(self) -> None: + with self._connect() as connection: + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS assignments ( + board TEXT NOT NULL, task_id TEXT NOT NULL, run_id TEXT NOT NULL, + worker_ordinal INTEGER NOT NULL CHECK(worker_ordinal BETWEEN 0 AND 2), + attempt INTEGER NOT NULL, assignment_digest TEXT NOT NULL, + payload_json TEXT NOT NULL, state TEXT NOT NULL, + lease_until REAL NOT NULL DEFAULT 0, last_heartbeat REAL NOT NULL DEFAULT 0, + result_digest TEXT, result_json TEXT, created_at REAL NOT NULL, + updated_at REAL NOT NULL, PRIMARY KEY(board, task_id, run_id) + ); + CREATE UNIQUE INDEX IF NOT EXISTS one_live_assignment_per_worker + ON assignments(worker_ordinal) WHERE state IN ('assigned','running','result'); + CREATE TABLE IF NOT EXISTS deliveries ( + delivery_id TEXT PRIMARY KEY, kind TEXT NOT NULL, digest TEXT NOT NULL, + received_at REAL NOT NULL + ); + """ + ) + + @staticmethod + def _record(row: sqlite3.Row | None) -> dict[str, Any] | None: + if row is None: + return None + value = dict(row) + value["payload"] = json.loads(value.pop("payload_json")) + if value.get("result_json"): + value["result"] = json.loads(value["result_json"]) + return value + + def add(self, binding: dict[str, Any], payload: dict[str, Any]) -> bool: + """Create exactly one assignment for a claimed run and free ordinal.""" + now = time.time() + digest = payload_digest(payload) + values = ( + binding["board"], binding["task_id"], binding["run_id"], + binding["worker_ordinal"], binding["attempt"], digest, + canonical_json(payload).decode(), "assigned", now, now, + ) + with self._lock, self._connect() as connection: + try: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + """INSERT INTO assignments + (board,task_id,run_id,worker_ordinal,attempt,assignment_digest, + payload_json,state,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?)""", + values, + ) + connection.commit() + return True + except sqlite3.IntegrityError as error: + connection.rollback() + existing = connection.execute( + """SELECT assignment_digest,worker_ordinal,attempt FROM assignments + WHERE board=? AND task_id=? AND run_id=?""", + values[:3], + ).fetchone() + if existing and tuple(existing) == ( + digest, binding["worker_ordinal"], binding["attempt"] + ): + return False + if existing: + raise ProtocolError("conflicting duplicate assignment") from error + raise ProtocolError("worker ordinal already has a live assignment") from error + + def available_ordinals(self) -> list[int]: + with self._connect() as connection: + rows = connection.execute( + "SELECT worker_ordinal FROM assignments WHERE state IN ('assigned','running','result')" + ).fetchall() + occupied = {int(row[0]) for row in rows} + return [ordinal for ordinal in range(3) if ordinal not in occupied] + + def active_assignments(self) -> list[dict[str, Any]]: + with self._connect() as connection: + rows = connection.execute( + "SELECT * FROM assignments WHERE state IN ('assigned','running','result')" + ).fetchall() + return [self._record(row) or {} for row in rows] + + def offer(self, ordinal: int) -> dict[str, Any] | None: + """Return the ordinal's durable assignment, preserving restart identity.""" + now = time.time() + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + """SELECT * FROM assignments WHERE worker_ordinal=? + AND state IN ('assigned','running') ORDER BY created_at LIMIT 1""", + (ordinal,), + ).fetchone() + if row is not None: + connection.execute( + """UPDATE assignments SET state='running',lease_until=?,last_heartbeat=?,updated_at=? + WHERE board=? AND task_id=? AND run_id=?""", + (now + self.lease_seconds, now, now, row["board"], row["task_id"], row["run_id"]), + ) + connection.commit() + return self._record(row) + + def _matching(self, connection: sqlite3.Connection, envelope: dict[str, Any]) -> sqlite3.Row: + row = connection.execute( + "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", + (envelope["board"], envelope["task_id"], envelope["run_id"]), + ).fetchone() + if row is None: + raise ProtocolError("assignment is unknown or stale") + if int(row["worker_ordinal"]) != int(envelope["worker_ordinal"]): + raise ProtocolError("worker ordinal does not own this assignment") + if int(row["attempt"]) != int(envelope["attempt"]): + raise ProtocolError("assignment attempt is stale") + return row + + def heartbeat(self, envelope: dict[str, Any]) -> tuple[bool, bool]: + now = time.time() + delivery_digest = payload_digest( + { + name: envelope[name] + for name in ( + "kind", "board", "task_id", "run_id", "worker_ordinal", + "attempt", "payload_digest", + ) + } + ) + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + duplicate = connection.execute( + "SELECT digest FROM deliveries WHERE delivery_id=?", + (envelope["delivery_id"],), + ).fetchone() + row = self._matching(connection, envelope) + if row["state"] not in {"assigned", "running"}: + raise ProtocolError("assignment is no longer running") + if duplicate and duplicate[0] != delivery_digest: + raise ProtocolError("delivery identifier was reused") + if not duplicate: + connection.execute( + "INSERT INTO deliveries VALUES (?,?,?,?)", + (envelope["delivery_id"], "heartbeat", delivery_digest, now), + ) + connection.execute( + "UPDATE assignments SET state='running',lease_until=?,last_heartbeat=?,updated_at=? WHERE board=? AND task_id=? AND run_id=?", + (now + self.lease_seconds, now, now, envelope["board"], envelope["task_id"], envelope["run_id"]), + ) + connection.commit() + return True, bool(duplicate) + + def accept_result(self, envelope: dict[str, Any]) -> tuple[dict[str, Any], bool]: + now = time.time() + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = self._matching(connection, envelope) + digest = envelope["payload_digest"] + if row["result_digest"]: + if row["result_digest"] != digest: + raise ProtocolError("conflicting result for completed delivery") + connection.rollback() + return self._record(row) or {}, True + if row["state"] not in {"assigned", "running"}: + raise ProtocolError("assignment cannot accept a result") + connection.execute( + """UPDATE assignments SET state='result',result_digest=?,result_json=?,updated_at=? + WHERE board=? AND task_id=? AND run_id=?""", + (digest, canonical_json(envelope["payload"]).decode(), now, + envelope["board"], envelope["task_id"], envelope["run_id"]), + ) + connection.commit() + row = connection.execute( + "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", + (envelope["board"], envelope["task_id"], envelope["run_id"]), + ).fetchone() + return self._record(row) or {}, False + + def pending_results(self) -> list[dict[str, Any]]: + with self._connect() as connection: + rows = connection.execute( + "SELECT * FROM assignments WHERE state='result' ORDER BY updated_at" + ).fetchall() + return [self._record(row) or {} for row in rows] + + def finalize(self, binding: dict[str, Any], state: str) -> None: + if state not in {"finalized", "stale"}: + raise ProtocolError("invalid terminal assignment state") + with self._lock, self._connect() as connection: + connection.execute( + "UPDATE assignments SET state=?,updated_at=? WHERE board=? AND task_id=? AND run_id=?", + (state, time.time(), binding["board"], binding["task_id"], binding["run_id"]), + ) + + def garbage_collect(self, retention_seconds: int) -> int: + cutoff = time.time() - max(3600, retention_seconds) + with self._lock, self._connect() as connection: + cursor = connection.execute( + "DELETE FROM assignments WHERE state IN ('finalized','stale') AND updated_at < ?", + (cutoff,), + ) + connection.execute("DELETE FROM deliveries WHERE received_at < ?", (cutoff,)) + return int(cursor.rowcount) diff --git a/services/hermes/scripts/execution_pool_scm.py b/services/hermes/scripts/execution_pool_scm.py new file mode 100644 index 00000000..74a488b1 --- /dev/null +++ b/services/hermes/scripts/execution_pool_scm.py @@ -0,0 +1,450 @@ +#!/usr/bin/env python3 +"""Ordinal-scoped Atlas SCM boundary; the model container never gets its token.""" + +from __future__ import annotations + +import configparser +import json +import os +import re +import shutil +import stat +import subprocess +import threading +import urllib.error +import urllib.parse +import urllib.request +import uuid +from http.server import BaseHTTPRequestHandler +from pathlib import Path +from typing import Any + +from execution_pool_protocol import ( + MAX_WIRE_BYTES, + BoundedHTTPServer, + ProtocolError, + parse_wire, + read_key, + verify_envelope, +) + + +WORKSPACE_ROOT = Path(os.environ.get("HERMES_WORKER_ROOT", "/workspace")) +SCM_ROOT = Path(os.environ.get("HERMES_SCM_STATE_ROOT", "/scm-state")) +KEY_PATH = Path(os.environ.get("HERMES_EXECUTION_POOL_KEY_FILE", "/pool-access/execution-pool-key")) +TOKEN_PATH = Path(os.environ.get("HERMES_GITEA_TOKEN_FILE", "/vault/secrets/gitea-token")) +USERNAME_PATH = Path(os.environ.get("HERMES_GITEA_USERNAME_FILE", "/vault/secrets/gitea-username")) +ASKPASS = os.environ.get( + "HERMES_GITEA_ASKPASS", "/opt/coordinator/execution_pool_askpass.sh" +) +ORDINAL = int(os.environ.get("HERMES_WORKER_ORDINAL", "-1")) +PORT = int(os.environ.get("HERMES_SCM_BOUNDARY_PORT", "9008")) +REPO = re.compile(r"^https://scm\.bstein\.dev/atlas/([A-Za-z0-9_.-]+)\.git$") +BRANCH = re.compile(r"^(?:feature|fix|chore|docs|test|refactor)/[A-Za-z0-9][A-Za-z0-9._/-]{0,119}$") +MAX_BUNDLE_BYTES = 128 * 1024 * 1024 + + +def _private_text(path: Path) -> str: + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + try: + value = os.read(descriptor, 65537).decode("utf-8").strip() + finally: + os.close(descriptor) + if not value or len(value) > 65536: + raise ProtocolError(f"SCM credential is missing or invalid: {path.name}") + return value + + +def _git_env(authenticated: bool) -> dict[str, str]: + """Expose credential paths only to explicit boundary-owned network calls.""" + sensitive = { + "GIT_ASKPASS", "SSH_ASKPASS", "HERMES_SCM_PASSWORD_FILE", + "HERMES_SCM_USERNAME_FILE", + } + environment = { + name: value for name, value in os.environ.items() + if name not in sensitive and not name.startswith("GIT_CONFIG_") + } + environment["GIT_TERMINAL_PROMPT"] = "0" + if authenticated: + environment.update( + { + "GIT_ASKPASS": ASKPASS, + "HERMES_SCM_PASSWORD_FILE": str(TOKEN_PATH), + "HERMES_SCM_USERNAME_FILE": str(USERNAME_PATH), + "GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": ( + "url.http://gitea.gitea.svc.cluster.local:3000/.insteadOf" + ), + "GIT_CONFIG_VALUE_0": "https://scm.bstein.dev/", + } + ) + return environment + + +def _run( + *arguments: str, + cwd: Path | None = None, + timeout: int = 300, + authenticated: bool = False, +) -> str: + completed = subprocess.run( + list(arguments), cwd=cwd, env=_git_env(authenticated), stdin=subprocess.DEVNULL, + text=True, capture_output=True, timeout=timeout, check=False, + ) + if completed.returncode: + message = (completed.stderr or completed.stdout or "SCM operation failed")[-2000:] + raise RuntimeError(message.strip()) + return completed.stdout.strip() + + +def _regular_text(path: Path, limit: int, encoding: str = "utf-8") -> str: + """Read bounded model-controlled metadata without following or blocking.""" + flags = os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as error: + raise ProtocolError(f"workspace metadata is unavailable: {path.name}") from error + try: + info = os.fstat(descriptor) + if not stat.S_ISREG(info.st_mode) or info.st_size > limit: + raise ProtocolError(f"workspace metadata is invalid: {path.name}") + raw = os.read(descriptor, limit + 1) + if len(raw) > limit: + raise ProtocolError(f"workspace metadata is oversized: {path.name}") + return raw.decode(encoding) + except UnicodeError as error: + raise ProtocolError(f"workspace metadata is malformed: {path.name}") from error + finally: + os.close(descriptor) + + +def _binding(envelope: dict[str, Any]) -> tuple[dict[str, Any], str, str, str]: + if envelope["kind"] != "assignment" or int(envelope["worker_ordinal"]) != ORDINAL: + raise ProtocolError("assignment does not belong to this worker ordinal") + payload = envelope["payload"] + if not isinstance(payload, dict): + raise ProtocolError("assignment payload must be an object") + repo = str(payload.get("repo_url") or "") + branch = str(payload.get("branch") or "") + base_branch = str(payload.get("base_branch") or "main") + match = REPO.fullmatch(repo) + if ( + not match + or not BRANCH.fullmatch(branch) + or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,119}", base_branch) + or ".." in branch + or "//" in branch + or ".." in base_branch + or "//" in base_branch + ): + raise ProtocolError("assignment is outside the Atlas SCM policy") + run_name = f"{envelope['task_id']}-{envelope['run_id']}" + return payload, repo, branch, match.group(1) + ":" + run_name + + +def workspace_path(envelope: dict[str, Any]) -> Path: + """Derive a contained path; no caller-provided filesystem path is accepted.""" + parts = (str(envelope["board"]), str(envelope["task_id"]), str(envelope["run_id"])) + if any(not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", part) for part in parts): + raise ProtocolError("assignment path binding is invalid") + workspace_root = WORKSPACE_ROOT.resolve() + root = WORKSPACE_ROOT / "runs" + if root.is_symlink(): + raise ProtocolError("workspace run root must not be a symlink") + root.mkdir(mode=0o700, parents=True, exist_ok=True) + root = root.resolve() + root.relative_to(workspace_root) + candidate = root.joinpath(*parts) + if candidate.is_symlink(): + raise ProtocolError("workspace must not be a symlink") + current = root + for part in parts[:-1]: + current = current / part + if current.is_symlink(): + raise ProtocolError("workspace parent must not be a symlink") + current.mkdir(mode=0o700, exist_ok=True) + if current.is_symlink(): + raise ProtocolError("workspace parent must not be a symlink") + try: + candidate.resolve(strict=False).relative_to(root) + except ValueError as error: + raise ProtocolError("workspace escaped its ordinal root") from error + return candidate + + +def _workspace_identity(destination: Path, repo: str, branch: str) -> str: + """Read identity as data; never execute Git in a model-controlled checkout.""" + git_dir = destination / ".git" + if git_dir.is_symlink() or not git_dir.is_dir(): + raise ProtocolError("workspace Git metadata must be a private directory") + raw_config = _regular_text(git_dir / "config", 64 * 1024) + parser = configparser.ConfigParser(interpolation=None, strict=True) + try: + parser.read_string(raw_config) + origin = parser.get('remote "origin"', "url") + except (configparser.Error, KeyError, UnicodeError) as error: + raise ProtocolError("workspace Git config cannot prove its origin") from error + if origin != repo: + raise ProtocolError("durable workspace origin does not match assignment") + head = _regular_text(git_dir / "HEAD", 4096).strip() + if head != f"ref: refs/heads/{branch}": + raise ProtocolError("durable workspace branch does not match assignment") + ref_path = git_dir / "refs/heads" / Path(branch) + if ref_path.is_symlink(): + raise ProtocolError("workspace branch ref must not be a symlink") + if ref_path.exists(): + commit = _regular_text(ref_path, 128, "ascii").strip() + else: + packed = _regular_text(git_dir / "packed-refs", 1024 * 1024, "ascii") + matches = [ + line.split(" ", 1)[0] + for line in packed.splitlines() + if line.endswith(f" refs/heads/{branch}") + ] + if len(matches) != 1: + raise ProtocolError("workspace branch ref is unavailable") + commit = matches[0] + if not re.fullmatch(r"[0-9a-f]{40,64}", commit): + raise ProtocolError("workspace branch ref is invalid") + return commit + + +def _private_repo(envelope: dict[str, Any]) -> Path: + root = SCM_ROOT.resolve() + root.mkdir(mode=0o700, parents=True, exist_ok=True) + name = f"{envelope['board']}-{envelope['task_id']}-{envelope['run_id']}.git" + path = root / name + path.resolve(strict=False).relative_to(root) + if path.is_symlink(): + raise ProtocolError("private SCM state must not be a symlink") + return path + + +def _copy_bundle(source: Path, destination: Path) -> None: + """Copy a bounded regular bundle into boundary-private storage.""" + source_fd = os.open( + source, os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0) + ) + temporary = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp") + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + target_fd = -1 + try: + info = os.fstat(source_fd) + if not stat.S_ISREG(info.st_mode) or not 0 < info.st_size <= MAX_BUNDLE_BYTES: + raise ProtocolError("submission bundle is empty, oversized, or not regular") + target_fd = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), + 0o600, + ) + remaining = info.st_size + while remaining: + chunk = os.read(source_fd, min(1024 * 1024, remaining)) + if not chunk: + raise ProtocolError("submission bundle ended early") + view = memoryview(chunk) + while view: + view = view[os.write(target_fd, view) :] + remaining -= len(chunk) + os.fsync(target_fd) + except Exception: + temporary.unlink(missing_ok=True) + raise + finally: + os.close(source_fd) + if target_fd >= 0: + os.close(target_fd) + try: + os.replace(temporary, destination) + finally: + temporary.unlink(missing_ok=True) + + +class Boundary: + def __init__(self, key: bytes): + self.key = key + self.lock = threading.RLock() + + def verify(self, raw: Any) -> dict[str, Any]: + return verify_envelope(self.key, raw, expected_kind="assignment") + + def checkout(self, envelope: dict[str, Any]) -> dict[str, Any]: + """Clone/fetch exactly one assigned branch into its private durable path.""" + payload, repo, branch, _ = _binding(envelope) + base_branch = str(payload.get("base_branch") or "main") + destination = workspace_path(envelope) + with self.lock: + if (destination / ".git").exists(): + return { + "workspace": str(destination), "preserved_dirty_state": True, + "baseline_sha": _workspace_identity(destination, repo, branch), + } + if destination.exists() and any(destination.iterdir()): + raise ProtocolError("workspace is non-empty and unmanaged") + destination.parent.mkdir(parents=True, exist_ok=True) + try: + _run( + "git", "clone", "--single-branch", "--branch", branch, + "--no-tags", repo, str(destination), timeout=900, + authenticated=True, + ) + except RuntimeError as error: + if destination.exists() and any(destination.iterdir()): + raise RuntimeError( + "assigned branch clone failed and left state for review" + ) from error + _run( + "git", "clone", "--single-branch", "--branch", base_branch, + "--no-tags", repo, str(destination), timeout=900, + authenticated=True, + ) + _run("git", "checkout", "-b", branch, cwd=destination) + _run("git", "config", "user.name", "Hermes Execution Worker", cwd=destination) + _run("git", "config", "user.email", "hermes@bstein.dev", cwd=destination) + return { + "workspace": str(destination), "preserved_dirty_state": False, + "baseline_sha": _run("git", "rev-parse", "HEAD", cwd=destination), + } + + def submit(self, envelope: dict[str, Any], request: dict[str, Any]) -> dict[str, Any]: + """Push only the assignment branch and create/reuse its draft pull request.""" + _payload, repo, branch, repo_binding = _binding(envelope) + repo_name, _ = repo_binding.split(":", 1) + destination = workspace_path(envelope) + title = str(request.get("title") or f"Hermes task {envelope['task_id']}").strip()[:240] + body = str(request.get("body") or "Automated draft from Hermes execution pool.")[:12000] + if not title or not (destination / ".git").exists(): + raise ProtocolError("submission workspace or title is invalid") + with self.lock: + _workspace_identity(destination, repo, branch) + private = _private_repo(envelope) + bundle = private.with_suffix(".bundle") + _copy_bundle(destination / ".git/hermes-submit.bundle", bundle) + if not private.exists(): + _run("git", "init", "--bare", str(private)) + _run("git", "--git-dir", str(private), "remote", "add", "origin", repo) + else: + _run("git", "--git-dir", str(private), "remote", "set-url", "origin", repo) + try: + _run( + "git", "--git-dir", str(private), "fetch", "origin", + f"+refs/heads/{branch}:refs/remotes/origin/{branch}", timeout=900, + authenticated=True, + ) + except RuntimeError as error: + if "couldn't find remote ref" not in str(error).lower(): + raise + _run( + "git", "--git-dir", str(private), "fetch", str(bundle), + "HEAD:refs/pool/candidate", timeout=900, + ) + _run( + "git", "--git-dir", str(private), "push", "origin", + f"refs/pool/candidate:refs/heads/{branch}", timeout=900, + authenticated=True, + ) + pull = self._draft_pull( + repo_name, branch, str(_payload.get("base_branch") or "main"), title, body + ) + bundle.unlink(missing_ok=True) + shutil.rmtree(private) + return {"workspace": str(destination), "branch": branch, "pull_request": pull} + + @staticmethod + def _api(path: str, data: dict[str, Any] | None = None) -> Any: + token = _private_text(TOKEN_PATH) + request = urllib.request.Request( + "http://gitea.gitea.svc.cluster.local:3000" + path, + data=json.dumps(data, separators=(",", ":")).encode() if data else None, + method="POST" if data else "GET", + headers={ + "Authorization": f"token {token}", "Accept": "application/json", + "Content-Type": "application/json", "User-Agent": "hermes-scm-boundary/1", + }, + ) + with urllib.request.urlopen(request, timeout=30) as response: + body = response.read(1024 * 1024 + 1) + if len(body) > 1024 * 1024: + raise ProtocolError("Gitea response exceeds the SCM boundary limit") + value = json.loads(body) + if not isinstance(value, (dict, list)): + raise ProtocolError("Gitea response has an invalid shape") + return value + + def _draft_pull( + self, repo: str, branch: str, base_branch: str, title: str, body: str + ) -> str: + query = urllib.parse.urlencode({"state": "open", "head": f"atlas:{branch}", "limit": 10}) + existing = self._api(f"/api/v1/repos/atlas/{repo}/pulls?{query}") + if isinstance(existing, list) and existing: + return str(existing[0].get("html_url") or "") + created = self._api( + f"/api/v1/repos/atlas/{repo}/pulls", + { + "base": base_branch, + "head": branch, "title": title, "body": body, "draft": True, + }, + ) + return str(created.get("html_url") or "") + + def garbage_collect(self, envelope: dict[str, Any]) -> dict[str, Any]: + """Online requests never delete model-controlled or private SCM state.""" + return {"removed": False, "eligible": False, "workspace": str(workspace_path(envelope))} + + +def handler_factory(boundary: Boundary) -> type[BaseHTTPRequestHandler]: + class Handler(BaseHTTPRequestHandler): + server_version = "hermes-scm-boundary/1" + + def _reply(self, status: int, value: dict[str, Any]) -> None: + body = json.dumps(value, separators=(",", ":"), sort_keys=True).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # noqa: N802 + self._reply(200, {"ready": True}) if self.path == "/ready" else self._reply(404, {"error": "not found"}) + + def do_POST(self) -> None: # noqa: N802 + try: + length = int(self.headers.get("Content-Length", "0")) + request = parse_wire(self.rfile.read(length)) if 0 < length <= MAX_WIRE_BYTES else None + if not request or set(request) - {"operation", "assignment", "title", "body"}: + raise ProtocolError("invalid SCM request") + envelope = boundary.verify(request.get("assignment")) + operations = { + "checkout": lambda: boundary.checkout(envelope), + "submit": lambda: boundary.submit(envelope, request), + "gc-check": lambda: boundary.garbage_collect(envelope), + } + operation = str(request.get("operation") or "") + if operation not in operations: + raise ProtocolError("unsupported SCM operation") + self._reply(200, operations[operation]()) + except (ProtocolError, RuntimeError, OSError, urllib.error.URLError) as error: + self._reply(409, {"error": str(error)[:2000]}) + + def log_message(self, _format: str, *_arguments: Any) -> None: + return + + return Handler + + +def main() -> int: + if ORDINAL not in range(3): + raise SystemExit("HERMES_WORKER_ORDINAL must be 0, 1, or 2") + key = read_key(KEY_PATH) + _private_text(TOKEN_PATH) + server = BoundedHTTPServer( + ("0.0.0.0", PORT), handler_factory(Boundary(key)), max_workers=4 + ) + server.serve_forever() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/execution_pool_worker.py b/services/hermes/scripts/execution_pool_worker.py new file mode 100644 index 00000000..be3e4da5 --- /dev/null +++ b/services/hermes/scripts/execution_pool_worker.py @@ -0,0 +1,413 @@ +#!/usr/bin/env python3 +"""Run one fenced Hermes assignment on an ordinal-scoped durable workspace.""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import stat +import subprocess +import time +import urllib.error +import urllib.request +from dataclasses import asdict +from pathlib import Path +from typing import Any + +import cli_lane_runner +from execution_pool_protocol import ( + ProtocolError, + atomic_json, + canonical_json, +) + + +ROOT = Path(os.environ.get("HERMES_WORKER_ROOT", "/workspace")) +CLIENT = os.environ.get("HERMES_EXECUTION_CLIENT_URL", "http://127.0.0.1:9009").rstrip("/") +SCM = os.environ.get("HERMES_SCM_BOUNDARY_URL", "http://127.0.0.1:9008").rstrip("/") +ORDINAL = int(os.environ.get("HERMES_WORKER_ORDINAL", "-1")) +NODE = os.environ.get("HERMES_WORKER_NODE", "unknown")[:128] +RUN_PART = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") +RETENTION_SECONDS = int(os.environ.get("HERMES_WORKER_RETENTION_SECONDS", "1209600")) + + +def _post(url: str, value: dict[str, Any], timeout: int = 60) -> dict[str, Any]: + request = urllib.request.Request( + url, data=canonical_json(value), method="POST", + headers={"Content-Type": "application/json", "Cache-Control": "no-store"}, + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + body = response.read(64 * 1024 + 1) + if len(body) > 64 * 1024: + raise ProtocolError("server response exceeds the wire limit") + document = json.loads(body) + if not isinstance(document, dict): + raise ProtocolError("server response is not an object") + return document + + +def _client(operation: str, **values: Any) -> dict[str, Any]: + response = _post(f"{CLIENT}/v1/client", {"operation": operation, **values}) + if response.get("error"): + raise ProtocolError(str(response["error"])) + return response + + +def _poll() -> dict[str, Any] | None: + assignment = _client("poll").get("assignment") + if assignment is None: + return None + if not isinstance(assignment, dict) or int(assignment["worker_ordinal"]) != ORDINAL: + raise ProtocolError("local boundary returned a foreign assignment") + return assignment + + +def _scm(operation: str, assignment: dict[str, Any], **values: Any) -> dict[str, Any]: + request = {"operation": operation, "assignment": assignment, **values} + response = _post(f"{SCM}/v1/scm", request, timeout=900) + if response.get("error"): + raise RuntimeError(str(response["error"])) + return response + + +def _binding(assignment: dict[str, Any]) -> dict[str, Any]: + return {name: assignment[name] for name in ( + "board", "task_id", "run_id", "worker_ordinal", "attempt" + )} + + +def _state_path(assignment: dict[str, Any]) -> Path: + parts = tuple(str(assignment[name]) for name in ("board", "task_id", "run_id")) + if any(not RUN_PART.fullmatch(part) for part in parts): + raise ProtocolError("assignment state path is invalid") + root = (ROOT / "session-state").resolve() + path = root.joinpath(*parts).with_suffix(".json") + path.parent.mkdir(parents=True, exist_ok=True) + if path.is_symlink() or any(parent.is_symlink() for parent in path.parents if parent != root.parent): + raise ProtocolError("assignment state path contains a symlink") + path.resolve(strict=False).relative_to(root) + return path + + +def _bind_provider_sessions(assignment: dict[str, Any]) -> None: + """Attach provider session directories to this exact durable task/run.""" + parts = tuple(str(assignment[name]) for name in ("board", "task_id", "run_id")) + if any(not RUN_PART.fullmatch(part) for part in parts): + raise ProtocolError("provider session binding is invalid") + configured_root = ROOT / "provider-state" + if configured_root.is_symlink() or not configured_root.is_dir(): + raise ProtocolError("durable provider state root is unavailable") + provider_root = configured_root.resolve() + run_root = provider_root + for part in parts: + run_root /= part + if run_root.is_symlink(): + raise ProtocolError("provider session path contains a symlink") + run_root.mkdir(mode=0o700, exist_ok=True) + run_root.resolve().relative_to(provider_root) + home = run_root / "home" + if home.is_symlink(): + raise ProtocolError("provider HOME must not be a durable symlink") + home.mkdir(mode=0o700, parents=True, exist_ok=True) + runtime_home = cli_lane_runner.DATA_ROOT / "home" + if runtime_home.is_symlink(): + runtime_home.unlink() + elif runtime_home.exists(): + raise ProtocolError("provider HOME is not a task-bound symlink") + runtime_home.symlink_to(home) + (home / ".claude").mkdir(mode=0o700, exist_ok=True) + settings = home / ".claude/settings.json" + if not settings.exists(): + atomic_json(settings, {}, 0o600) + bindings = { + Path(os.environ.get("CODEX_HOME", "/runtime-access/codex")) / "sessions": + run_root / "codex/sessions", + Path(os.environ.get("CLAUDE_CONFIG_DIR", "/runtime-access/claude")) / "projects": + run_root / "claude/projects", + Path(os.environ.get("CLAUDE_CONFIG_DIR", "/runtime-access/claude")) / "session-env": + run_root / "claude/session-env", + Path(os.environ.get("CLAUDE_CONFIG_DIR", "/runtime-access/claude")) / "todos": + run_root / "claude/todos", + } + for runtime, durable in bindings.items(): + current = run_root + for part in durable.relative_to(run_root).parts: + current /= part + if current.is_symlink(): + raise ProtocolError("provider session path contains a symlink") + current.mkdir(mode=0o700, exist_ok=True) + current.resolve().relative_to(provider_root) + if runtime.is_symlink(): + runtime.unlink() + elif runtime.exists(): + raise ProtocolError(f"provider session path is not a symlink: {runtime.name}") + runtime.symlink_to(durable) + + +def _prompt(context: str, workspace: Path, binding: dict[str, Any]) -> str: + return f"""You are a durable coding worker managed by Hermes Kanban. + +Work only on this objective and its acceptance criteria: + +{context} + +Workspace: {workspace} +Run binding: board={binding['board']} task={binding['task_id']} run={binding['run_id']} worker={ORDINAL} + +Operate autonomously only inside this private assigned checkout. Inspect before editing, +preserve unrelated and untracked files, and fail closed rather than overwrite state you do +not understand. You have no Kubernetes identity and no SCM credential. Do not attempt to +read Secrets, mutate workloads, use exec/attach/port-forward, reach node roots, or bypass +the reviewed SCM boundary. Commit intended changes locally on the assigned feature branch; +the worker boundary handles the bounded push and draft pull request after validation. +Switchyard owns provider/model/effort selection and cross-provider fallback. + +Return a final JSON object matching the supplied schema. Use status=incomplete when work, +tests, commits, or verification remain. Use blocked only for a concrete task obstacle. +Completed must have no blockers. List changed files, tests, artifacts, findings, and blockers. +""" + + +def _read_activity(log_path: Path, offset: int) -> tuple[str, int]: + flags = os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(log_path, flags) + except FileNotFoundError: + return "", offset + try: + info = os.fstat(descriptor) + if not stat.S_ISREG(info.st_mode): + raise ProtocolError("worker log must be a regular file") + if info.st_size < offset: + offset = 0 + os.lseek(descriptor, offset, os.SEEK_SET) + value = os.read(descriptor, 12 * 1024) + next_offset = os.lseek(descriptor, 0, os.SEEK_CUR) + finally: + os.close(descriptor) + return value.decode("utf-8", "replace"), next_offset + + +def _git(workspace: Path, *arguments: str) -> str: + completed = subprocess.run( + ["git", "-C", str(workspace), *arguments], stdin=subprocess.DEVNULL, + text=True, capture_output=True, timeout=30, check=False, + env={**os.environ, "GIT_TERMINAL_PROMPT": "0", "GIT_ASKPASS": "/bin/false"}, + ) + if completed.returncode: + raise RuntimeError((completed.stderr or "Git inspection failed")[-1000:]) + return completed.stdout.strip() + + +def _bounded_result(value: dict[str, Any]) -> dict[str, Any]: + """Keep terminal evidence useful while fitting the authenticated wire cap.""" + bounded: dict[str, Any] = {} + for name in ("status", "summary"): + bounded[name] = str(value.get(name) or "")[:8000] + for name in ("changed_files", "tests_run", "artifacts", "findings", "blockers"): + items = value.get(name) + bounded[name] = [str(item)[:2000] for item in items[:100]] if isinstance(items, list) else [] + while len(canonical_json(bounded)) > 32 * 1024: + longest = max( + (name for name in bounded if isinstance(bounded[name], list) and bounded[name]), + key=lambda name: len(canonical_json(bounded[name])), + default="", + ) + if longest: + bounded[longest].pop() + else: + bounded["summary"] = bounded["summary"][: len(bounded["summary"]) // 2] + return bounded + + +def _refresh_assignment(binding: dict[str, Any]) -> dict[str, Any]: + fresh = _poll() + if fresh is None or _binding(fresh) != binding: + raise ProtocolError("assignment changed before SCM submission") + return fresh + + +def garbage_collect(now: float | None = None) -> int: + """Remove only clean, terminal, assignment-derived workspaces after retention.""" + current = time.time() if now is None else now + state_root = (ROOT / "session-state").resolve() + run_root = (ROOT / "runs").resolve() + removed = 0 + for state_file in state_root.glob("*/*/*.json") if state_root.is_dir() else (): + state = cli_lane_runner.load_json(state_file) + terminal_at = float(state.get("terminal_at") or 0) + if terminal_at <= 0 or current - terminal_at < max(3600, RETENTION_SECONDS): + continue + workspace = Path(str(state.get("workspace") or "")) + try: + workspace.resolve(strict=True).relative_to(run_root) + except (OSError, RuntimeError, ValueError): + continue + if workspace.is_symlink() or _git( + workspace, "status", "--porcelain=v1", "--untracked-files=all" + ): + continue + shutil.rmtree(workspace) + state_file.unlink() + removed += 1 + return removed + + +def execute(assignment: dict[str, Any]) -> None: + binding = _binding(assignment) + payload = assignment["payload"] + if not isinstance(payload, dict): + raise ProtocolError("assignment payload is invalid") + _bind_provider_sessions(assignment) + checkout = _scm("checkout", assignment) + workspace = Path(str(checkout.get("workspace") or "")).resolve(strict=True) + workspace.relative_to((ROOT / "runs").resolve()) + state_file = _state_path(assignment) + state = cli_lane_runner.load_json(state_file) + state.update({**binding, "node": NODE, "workspace": str(workspace)}) + state.setdefault("baseline_sha", str(checkout.get("baseline_sha") or "")) + atomic_json(state_file, state) + log_path = state_file.with_suffix(".log") + offset = 0 + latest_route: dict[str, Any] = {} + + def heartbeat(note: str) -> bool: + nonlocal offset + try: + activity, offset = _read_activity(log_path, offset) + response = _client( + "heartbeat", binding=binding, + payload={ + "note": f"worker={ORDINAL} node={NODE} {note}"[:1000], + "activity": activity, "route": latest_route, + }, + ) + return bool(response.get("ack", {}).get("accepted")) + except (OSError, ValueError, urllib.error.URLError, json.JSONDecodeError): + return False + + context = str(payload.get("context") or "") + assignee = str(payload.get("assignee") or "cli-auto") + route = cli_lane_runner.select_route( + context, assignee, + exclude_provider=cli_lane_runner.fresh_unavailable_provider() + if assignee == "cli-auto" else None, + ) + latest_route = asdict(route) + if not heartbeat( + f"route={route.provider}/{route.model}/{route.effort}; assignment accepted" + ): + raise ProtocolError("assignment lease was lost before provider startup") + deadline = min( + int(payload.get("deadline_unix") or int(time.time()) + 60), + int(time.time()) + int(payload.get("max_runtime_seconds") or 60), + ) + result = cli_lane_runner.run_provider( + route, _prompt(context, workspace, binding), workspace, state, state_file, + log_path, heartbeat, max(1, deadline - int(time.time())), + ) + if result.capacity_failure and deadline - time.time() > 30: + alternate = "claude" if route.provider == "codex" else "codex" + fallback = cli_lane_runner.select_route( + context + "\nThe first provider failed from capacity or authentication.", + f"cli-{alternate}-{route.effort}", + ) + latest_route = asdict(fallback) + heartbeat(f"fallback={route.provider}->{fallback.provider}") + result = cli_lane_runner.run_provider( + fallback, + _prompt(context, workspace, binding) + + cli_lane_runner.git_handoff(workspace, result.output), + workspace, state, state_file, log_path, heartbeat, + max(1, deadline - int(time.time())), + ) + route = fallback + structured = dict(result.structured or {}) + for name in ("changed_files", "tests_run", "artifacts", "findings", "blockers"): + if not isinstance(structured.get(name), list): + structured[name] = [] + if structured.get("status") == "completed" and result.returncode == 0: + status = _git(workspace, "status", "--porcelain=v1", "--untracked-files=all") + if status: + structured["status"] = "incomplete" + structured["blockers"].append( + "Worker left uncommitted or untracked files; SCM submission failed closed." + ) + else: + baseline = str(state.get("baseline_sha") or "") + if not re.fullmatch(r"[0-9a-f]{40,64}", baseline): + raise RuntimeError("durable SCM baseline is missing or invalid") + ahead = int(_git(workspace, "rev-list", "--count", f"{baseline}..HEAD") or "0") + if ahead: + fresh = _refresh_assignment(binding) + bundle = workspace / ".git/hermes-submit.bundle" + bundle.unlink(missing_ok=True) + _git(workspace, "bundle", "create", str(bundle), "HEAD") + submission = _scm( + "submit", fresh, + title=str(structured.get("summary") or f"Hermes task {binding['task_id']}")[:240], + body=json.dumps(structured, indent=2, sort_keys=True)[:12000], + ) + pull = str(submission.get("pull_request") or "") + if pull: + structured["artifacts"].append(pull) + bundle.unlink(missing_ok=True) + structured = _bounded_result(structured) + activity, _ = _read_activity(log_path, offset) + result_payload = { + "structured": structured, + "returncode": result.returncode, + "capacity_failure": result.capacity_failure, + "node": NODE, + "route": asdict(route), + "provider_sessions": { + "codex_thread_id": state.get("codex_thread_id"), + "claude_session_id": state.get("claude_session_id"), + }, + "final_activity": activity, + } + response = _client("result", binding=binding, payload=result_payload) + if not response.get("ack", {}).get("accepted"): + raise ProtocolError("coordinator did not accept the terminal result") + state["terminal_at"] = time.time() + atomic_json(state_file, state) + + +def readiness() -> None: + if ORDINAL not in range(3): + raise ProtocolError("worker ordinal must be 0, 1, or 2") + for path in (ROOT, ROOT / "provider-state", cli_lane_runner.DATA_ROOT): + if not path.is_dir() or not os.access(path, os.W_OK): + raise ProtocolError(f"durable worker path is not writable: {path}") + for credential in ( + Path(os.environ.get("CODEX_HOME", "/runtime-access/codex")) / "auth.json", + Path(os.environ.get("CLAUDE_CONFIG_DIR", "/runtime-access/claude")) / ".credentials.json", + ): + if not credential.is_file() or not os.access(credential, os.W_OK): + raise ProtocolError(f"subscription credential is unavailable or read-only: {credential.name}") + cli_lane_runner.RESULT_SCHEMA_PATH.parent.mkdir(parents=True, exist_ok=True) + atomic_json( + cli_lane_runner.RESULT_SCHEMA_PATH, cli_lane_runner.RESULT_SCHEMA, 0o644 + ) + _poll() + + +def main() -> int: + readiness() + while True: + try: + garbage_collect() + assignment = _poll() + if assignment is None: + time.sleep(5) + continue + execute(assignment) + except (OSError, RuntimeError, ValueError, urllib.error.URLError, json.JSONDecodeError) as error: + print(f"worker {ORDINAL} deferred: {type(error).__name__}: {error}", flush=True) + time.sleep(10) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/stage_runtime_access.py b/services/hermes/scripts/stage_runtime_access.py index f4d73160..c497febb 100644 --- a/services/hermes/scripts/stage_runtime_access.py +++ b/services/hermes/scripts/stage_runtime_access.py @@ -6,12 +6,16 @@ from __future__ import annotations import argparse import json import os +import uuid from pathlib import Path VAULT_ROOT = Path("/vault/secrets") RUNTIME_ROOT = Path("/runtime-access") PERSISTENT_HOME = Path("/opt/data/home") +WORKER_ROOT = Path(os.environ.get("HERMES_WORKER_ROOT", "/workspace")) +POOL_ACCESS_ROOT = Path(os.environ.get("HERMES_POOL_ACCESS_ROOT", "/pool-access")) +SCM_ACCESS_ROOT = Path(os.environ.get("HERMES_SCM_ACCESS_ROOT", "/scm-access")) OWNER_UID = 10000 OWNER_GID = 10000 @@ -28,9 +32,22 @@ def _copy_secret(source_name: str, destination: Path) -> str: value = (VAULT_ROOT / source_name).read_text(encoding="utf-8").strip() if not value: raise RuntimeError(f"Vault projection {source_name} is empty") - destination.write_text(value + "\n", encoding="utf-8") - destination.chmod(0o600) - os.chown(destination, OWNER_UID, OWNER_GID) + temporary = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(temporary, flags, 0o600) + try: + remaining = memoryview((value + "\n").encode()) + while remaining: + remaining = remaining[os.write(descriptor, remaining) :] + os.fsync(descriptor) + os.fchmod(descriptor, 0o600) + os.fchown(descriptor, OWNER_UID, OWNER_GID) + finally: + os.close(descriptor) + try: + os.replace(temporary, destination) + finally: + temporary.unlink(missing_ok=True) return value @@ -92,6 +109,7 @@ def stage_agent() -> None: _owned_directory(path) for name in ( "agent-api-key", + "execution-pool-key", "chat-relay-key", "gitea-token", "gitea-username", @@ -144,16 +162,60 @@ def stage_triage() -> None: _write_empty_auth_store() +def _durable_link(runtime: Path, durable: Path, name: str) -> None: + """Link one provider state path only after rejecting replacement symlinks.""" + durable.mkdir(mode=0o700, parents=True, exist_ok=True) + durable.chmod(0o700) + os.chown(durable, OWNER_UID, OWNER_GID) + target = runtime / name + if target.is_symlink() and target.resolve(strict=False) != durable.resolve(): + target.unlink() + elif target.exists() and not target.is_symlink(): + raise RuntimeError(f"refusing to replace provider state path: {target}") + if not target.exists(): + target.symlink_to(durable) + + +def stage_execution_worker() -> None: + """Stage only subscription/channel auth and ordinal-private session state.""" + provider_state = WORKER_ROOT / "provider-state" + for path in ( + RUNTIME_ROOT, + RUNTIME_ROOT / "claude", + RUNTIME_ROOT / "codex", + POOL_ACCESS_ROOT, + SCM_ACCESS_ROOT, + provider_state, + ): + _owned_directory(path) + _copy_secret("execution-pool-key", POOL_ACCESS_ROOT / "execution-pool-key") + _copy_secret("gitea-token", SCM_ACCESS_ROOT / "gitea-token") + _copy_secret("gitea-username", SCM_ACCESS_ROOT / "gitea-username") + _validated_json( + "claude-credentials", + RUNTIME_ROOT / "claude" / ".credentials.json", + ("claudeAiOauth", "refreshToken"), + ) + _validated_json( + "codex-auth", + RUNTIME_ROOT / "codex" / "auth.json", + ("tokens", "refresh_token"), + ) + _write_empty_auth_store() + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("mode", choices=("agent", "chat", "triage")) + parser.add_argument("mode", choices=("agent", "chat", "triage", "execution-worker")) args = parser.parse_args() if args.mode == "agent": stage_agent() elif args.mode == "chat": stage_chat() - else: + elif args.mode == "triage": stage_triage() + else: + stage_execution_worker() print("Runtime access staged from Vault.", flush=True) return 0 diff --git a/services/hermes/service.yaml b/services/hermes/service.yaml index 0fc8d939..3903b5f9 100644 --- a/services/hermes/service.yaml +++ b/services/hermes/service.yaml @@ -117,6 +117,37 @@ spec: --- apiVersion: v1 kind: Service +metadata: + name: hermes-execution-worker + namespace: hermes + labels: + app: hermes-execution-worker +spec: + clusterIP: None + publishNotReadyAddresses: true + selector: + app: hermes-execution-worker +--- +apiVersion: v1 +kind: Service +metadata: + name: hermes-execution-pool + namespace: hermes + labels: + app: hermes-agent +spec: + type: ClusterIP + publishNotReadyAddresses: true + selector: + app: hermes-agent + ports: + - name: http + port: 9007 + targetPort: execution-pool + protocol: TCP +--- +apiVersion: v1 +kind: Service metadata: name: hermes-local-image namespace: hermes diff --git a/services/vault/scripts/vault_k8s_auth_configure.sh b/services/vault/scripts/vault_k8s_auth_configure.sh index 73572e62..68af4e77 100644 --- a/services/vault/scripts/vault_k8s_auth_configure.sh +++ b/services/vault/scripts/vault_k8s_auth_configure.sh @@ -223,6 +223,10 @@ write_policy_and_role "outline" "outline" "outline-vault" \ "outline/* shared/postmark-relay" "" write_policy_and_role "planka" "planka" "planka-vault" \ "planka/* shared/postmark-relay" "" +write_policy_and_role "hermes-execution-worker" "hermes" "hermes-execution-worker" \ + "hermes/agent-tokens hermes/developer-gitea" "" +write_policy_and_role "hermes-execution-credential-sync" "hermes" "hermes-execution-worker" \ + "" "hermes/agent-tokens" write_policy_and_role "bstein-dev-home" "bstein-dev-home" "bstein-dev-home,bstein-dev-home-vault-sync" \ "portal/* shared/chat-ai-keys-runtime shared/portal-e2e-client shared/postmark-relay mailu/mailu-initial-account-secret shared/harbor-pull" "" write_policy_and_role "gitea" "gitea" "gitea-vault" \ diff --git a/testing/tests/test_hermes_auto_router.py b/testing/tests/test_hermes_auto_router.py index c0a31103..05af3b01 100644 --- a/testing/tests/test_hermes_auto_router.py +++ b/testing/tests/test_hermes_auto_router.py @@ -255,6 +255,7 @@ def test_provider_status_separates_observed_activity_from_plan_quota(monkeypatch monkeypatch.setattr(module, "_claude_account", lambda: { "authenticated": True, "plan": "max", "quota_reported": False, }) + monkeypatch.setattr(module, "_fresh_health", lambda _path: {}) payload = module.provider_status_payload() diff --git a/testing/tests/test_hermes_execution_pool.py b/testing/tests/test_hermes_execution_pool.py new file mode 100644 index 00000000..0ee44945 --- /dev/null +++ b/testing/tests/test_hermes_execution_pool.py @@ -0,0 +1,500 @@ +"""Adversarial contracts for the fenced three-node Hermes execution pool.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import threading +import time +from pathlib import Path + +import pytest +import yaml + + +ROOT = Path(__file__).parents[2] +HERMES = ROOT / "services/hermes" +SCRIPTS = HERMES / "scripts" +sys.path.insert(0, str(SCRIPTS)) + +import execution_pool_coordinator as coordinator # noqa: E402 +import execution_pool_client as pool_client # noqa: E402 +import execution_pool_protocol as protocol # noqa: E402 +import execution_pool_scm as scm # noqa: E402 +import execution_pool_worker as worker # noqa: E402 + + +KEY = b"k" * 32 + + +def binding(**values): + result = { + "board": "atlas", + "task_id": "t_deadbeef", + "run_id": "run-1234", + "worker_ordinal": 0, + "attempt": 1, + } + result.update(values) + return result + + +def assignment_payload(**values): + result = { + "context": "Implement the bounded objective.", + "assignee": "cli-auto", + "repo_url": "https://scm.bstein.dev/atlas/titan-iac.git", + "branch": "feature/hermes-safe-pool", + "base_branch": "main", + "max_runtime_seconds": 3600, + "deadline_unix": int(time.time()) + 3600, + } + result.update(values) + return result + + +def test_envelope_binds_run_ordinal_attempt_and_digest(): + signed = protocol.sign_envelope(KEY, "heartbeat", binding(), {"note": "active"}) + verified = protocol.verify_envelope(KEY, signed, expected_kind="heartbeat") + + assert verified["payload_digest"] == protocol.payload_digest({"note": "active"}) + for name, value in binding().items(): + assert verified[name] == value + + signed["attempt"] = 2 + with pytest.raises(protocol.ProtocolError, match="authentication"): + protocol.verify_envelope(KEY, signed) + + +def test_empty_poll_acknowledgement_is_authenticated(): + empty = { + "board": "", "task_id": "", "run_id": "", + "worker_ordinal": 2, "attempt": 0, + } + signed = protocol.sign_envelope(KEY, "ack", empty, {"assignment": None}) + assert protocol.verify_envelope(KEY, signed, expected_kind="ack")["payload"] == { + "assignment": None + } + + +@pytest.mark.parametrize( + "mutation,error", + [ + ({"board": "../atlas"}, "invalid board"), + ({"worker_ordinal": 3}, "outside the pool"), + ({"expires_at": 1}, "validity window"), + ], +) +def test_malformed_or_traversal_bindings_fail_closed(mutation, error): + signed = protocol.sign_envelope(KEY, "heartbeat", binding(), {"note": "active"}) + signed.update(mutation) + unsigned = dict(signed) + unsigned.pop("signature") + import hashlib + import hmac + + signed["signature"] = hmac.new( + KEY, protocol.canonical_json(unsigned), hashlib.sha256 + ).hexdigest() + with pytest.raises(protocol.ProtocolError, match=error): + protocol.verify_envelope(KEY, signed) + + +def test_oversized_and_malformed_payloads_are_rejected(): + with pytest.raises(protocol.ProtocolError, match="wire limit"): + protocol.sign_envelope( + KEY, "result", binding(), {"output": "x" * protocol.MAX_WIRE_BYTES} + ) + with pytest.raises(protocol.ProtocolError, match="malformed"): + protocol.parse_wire(b"{not-json") + with pytest.raises(protocol.ProtocolError, match="oversized"): + protocol.parse_wire(b"x" * (protocol.MAX_WIRE_BYTES + 1)) + + +def test_key_requires_private_regular_file_and_rejects_symlink(tmp_path): + key = tmp_path / "key" + key.write_bytes(KEY) + key.chmod(0o644) + with pytest.raises(protocol.ProtocolError, match="private"): + protocol.read_key(key) + key.chmod(0o600) + assert protocol.read_key(key) == KEY + link = tmp_path / "link" + link.symlink_to(key) + with pytest.raises(protocol.ProtocolError, match="unavailable"): + protocol.read_key(link) + with pytest.raises(protocol.ProtocolError, match="unavailable"): + protocol.read_key(tmp_path / "missing") + + +def test_local_signing_boundary_rejects_foreign_or_unassigned_result(monkeypatch): + monkeypatch.setattr(pool_client, "ORDINAL", 0) + boundary = pool_client.ClientBoundary(KEY) + request = {"binding": binding(), "payload": {"returncode": 0}} + with pytest.raises(protocol.ProtocolError, match="does not own"): + boundary.forward("result", request) + boundary.current = binding() + request["binding"] = binding(worker_ordinal=1) + with pytest.raises(protocol.ProtocolError, match="does not own"): + boundary.forward("result", request) + + +def test_simultaneous_claim_materialization_has_one_winner(tmp_path): + store = protocol.PoolStore(tmp_path / "pool.db") + barrier = threading.Barrier(8) + outcomes = [] + + def add(index): + barrier.wait() + try: + outcome = store.add( + binding(task_id=f"t_{index}", run_id=f"run-{index}"), + assignment_payload(), + ) + except protocol.ProtocolError: + outcome = False + outcomes.append(outcome) + + threads = [threading.Thread(target=add, args=(index,)) for index in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert outcomes.count(True) == 1 + assert store.available_ordinals() == [1, 2] + + +def test_duplicate_assignment_is_idempotent_but_conflict_is_rejected(tmp_path): + store = protocol.PoolStore(tmp_path / "pool.db") + assert store.add(binding(), assignment_payload()) is True + assert store.add(binding(), assignment_payload()) is False + with pytest.raises(protocol.ProtocolError, match="conflicting"): + store.add(binding(), assignment_payload(context="different")) + with pytest.raises(protocol.ProtocolError, match="conflicting"): + store.add(binding(worker_ordinal=1), assignment_payload()) + + +def test_duplicate_heartbeat_and_result_are_idempotent(tmp_path): + store = protocol.PoolStore(tmp_path / "pool.db") + store.add(binding(), assignment_payload()) + store.offer(0) + heartbeat = protocol.sign_envelope( + KEY, "heartbeat", binding(), {"note": "active"}, delivery_id="delivery-1" + ) + assert store.heartbeat(heartbeat) == (True, False) + assert store.heartbeat(heartbeat) == (True, True) + result = protocol.sign_envelope( + KEY, "result", binding(), {"structured": {"status": "completed"}} + ) + _, duplicate = store.accept_result(result) + assert duplicate is False + _, duplicate = store.accept_result(result) + assert duplicate is True + + +def test_reused_delivery_and_stale_attempt_cannot_cross_runs(tmp_path): + store = protocol.PoolStore(tmp_path / "pool.db") + store.add(binding(), assignment_payload()) + first = protocol.sign_envelope( + KEY, "heartbeat", binding(), {"note": "one"}, delivery_id="delivery-1" + ) + store.heartbeat(first) + reused = protocol.sign_envelope( + KEY, "heartbeat", binding(), {"note": "two"}, delivery_id="delivery-1" + ) + with pytest.raises(protocol.ProtocolError, match="reused"): + store.heartbeat(reused) + store.finalize(binding(), "finalized") + store.add(binding(run_id="other-run"), assignment_payload()) + cross_run = protocol.sign_envelope( + KEY, "heartbeat", binding(run_id="other-run"), {"note": "one"}, + delivery_id="delivery-1", + ) + with pytest.raises(protocol.ProtocolError, match="reused"): + store.heartbeat(cross_run) + stale = protocol.sign_envelope( + KEY, "result", binding(attempt=2), {"structured": {"status": "completed"}} + ) + with pytest.raises(protocol.ProtocolError, match="attempt is stale"): + store.accept_result(stale) + replacement = protocol.sign_envelope( + KEY, + "result", + binding(run_id="replacement-run"), + {"structured": {"status": "completed"}}, + ) + with pytest.raises(protocol.ProtocolError, match="unknown or stale"): + store.accept_result(replacement) + + +def test_restart_heartbeat_loss_and_node_replacement_reuse_durable_assignment(tmp_path): + database = tmp_path / "pool.db" + first = protocol.PoolStore(database) + first.add(binding(), assignment_payload()) + offered = first.offer(0) + assert offered and offered["attempt"] == 1 + # Reopening the SQLite ledger models coordinator restart. An expired lease + # is deliberately not reassigned across ordinals; the StatefulSet replaces + # ordinal 0 with the same PVC and provider session state. + restarted = protocol.PoolStore(database) + with restarted._connect() as connection: + connection.execute("UPDATE assignments SET lease_until=0") + resumed = restarted.offer(0) + assert resumed and resumed["run_id"] == "run-1234" + assert resumed["attempt"] == 1 + assert restarted.offer(1) is None + assert len(restarted.active_assignments()) == 1 + + +def test_result_conflict_never_overwrites_first_result(tmp_path): + store = protocol.PoolStore(tmp_path / "pool.db") + store.add(binding(), assignment_payload()) + first = protocol.sign_envelope(KEY, "result", binding(), {"returncode": 0}) + store.accept_result(first) + conflict = protocol.sign_envelope(KEY, "result", binding(), {"returncode": 1}) + with pytest.raises(protocol.ProtocolError, match="conflicting result"): + store.accept_result(conflict) + + +def test_scm_workspace_is_ordinal_contained_and_rejects_symlink(tmp_path, monkeypatch): + monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path) + monkeypatch.setattr(scm, "ORDINAL", 0) + signed = protocol.sign_envelope(KEY, "assignment", binding(), assignment_payload()) + path = scm.workspace_path(signed) + assert path == tmp_path / "runs/atlas/t_deadbeef/run-1234" + path.parent.mkdir(parents=True, exist_ok=True) + path.symlink_to(tmp_path / "elsewhere", target_is_directory=True) + with pytest.raises(protocol.ProtocolError, match="symlink"): + scm.workspace_path(signed) + root_case = tmp_path / "root-case" + root_case.mkdir() + root = tmp_path / "other-root" + root.mkdir() + (root_case / "runs").symlink_to(root, target_is_directory=True) + monkeypatch.setattr(scm, "WORKSPACE_ROOT", root_case) + with pytest.raises(protocol.ProtocolError, match="run root"): + scm.workspace_path(signed) + + +def test_scm_boundary_rejects_other_repo_branch_and_ordinal(monkeypatch): + monkeypatch.setattr(scm, "ORDINAL", 0) + for changes, message in ( + ({"repo_url": "https://evil.example/atlas/titan-iac.git"}, "SCM policy"), + ({"branch": "main"}, "SCM policy"), + ): + signed = protocol.sign_envelope( + KEY, "assignment", binding(), assignment_payload(**changes) + ) + with pytest.raises(protocol.ProtocolError, match=message): + scm._binding(signed) + signed = protocol.sign_envelope( + KEY, "assignment", binding(worker_ordinal=1), assignment_payload() + ) + with pytest.raises(protocol.ProtocolError, match="ordinal"): + scm._binding(signed) + + +def test_scm_identity_is_data_only_and_supports_packed_refs(tmp_path): + checkout = tmp_path / "checkout" + subprocess.run(["git", "init", "-q", str(checkout)], check=True) + subprocess.run( + ["git", "-C", str(checkout), "remote", "add", "origin", + "https://scm.bstein.dev/atlas/titan-iac.git"], check=True, + ) + subprocess.run( + ["git", "-C", str(checkout), "checkout", "-qb", "feature/hermes-safe-pool"], + check=True, + ) + subprocess.run(["git", "-C", str(checkout), "config", "user.email", "a@b.c"], check=True) + subprocess.run(["git", "-C", str(checkout), "config", "user.name", "Test"], check=True) + (checkout / "tracked").write_text("safe\n") + subprocess.run(["git", "-C", str(checkout), "add", "tracked"], check=True) + subprocess.run(["git", "-C", str(checkout), "commit", "-qm", "initial"], check=True) + expected = subprocess.check_output( + ["git", "-C", str(checkout), "rev-parse", "HEAD"], text=True + ).strip() + subprocess.run(["git", "-C", str(checkout), "pack-refs", "--all"], check=True) + assert scm._workspace_identity( + checkout, "https://scm.bstein.dev/atlas/titan-iac.git", + "feature/hermes-safe-pool", + ) == expected + (checkout / ".git/refs/heads/feature").mkdir(parents=True) + (checkout / ".git/refs/heads/feature/hermes-safe-pool").symlink_to("/etc/passwd") + with pytest.raises(protocol.ProtocolError, match="symlink"): + scm._workspace_identity( + checkout, "https://scm.bstein.dev/atlas/titan-iac.git", + "feature/hermes-safe-pool", + ) + + +def test_scm_bundle_copy_is_nofollow_bounded_and_fetchable(tmp_path, monkeypatch): + repo = tmp_path / "repo" + private = tmp_path / "private.git" + subprocess.run(["git", "init", "-q", str(repo)], check=True) + subprocess.run(["git", "-C", str(repo), "config", "user.email", "a@b.c"], check=True) + subprocess.run(["git", "-C", str(repo), "config", "user.name", "Test"], check=True) + (repo / "tracked").write_text("safe\n") + subprocess.run(["git", "-C", str(repo), "add", "tracked"], check=True) + subprocess.run(["git", "-C", str(repo), "commit", "-qm", "initial"], check=True) + bundle = repo / ".git/hermes-submit.bundle" + subprocess.run(["git", "-C", str(repo), "bundle", "create", str(bundle), "HEAD"], check=True) + copied = tmp_path / "copied.bundle" + scm._copy_bundle(bundle, copied) + subprocess.run(["git", "init", "--bare", "-q", str(private)], check=True) + subprocess.run( + ["git", "--git-dir", str(private), "fetch", str(copied), + "HEAD:refs/pool/candidate"], check=True, + ) + link = tmp_path / "bundle-link" + link.symlink_to(bundle) + with pytest.raises(OSError): + scm._copy_bundle(link, tmp_path / "rejected") + fifo = tmp_path / "bundle-fifo" + os.mkfifo(fifo) + with pytest.raises(protocol.ProtocolError, match="not regular"): + scm._copy_bundle(fifo, tmp_path / "rejected-fifo") + monkeypatch.setattr(scm, "MAX_BUNDLE_BYTES", 1) + with pytest.raises(protocol.ProtocolError, match="oversized"): + scm._copy_bundle(bundle, tmp_path / "rejected-large") + + +def test_activity_is_bounded_sanitized_and_nofollow(tmp_path): + text = coordinator.sanitize_activity( + 'route=codex token=super-secret\nAuthorization: Bearer abcdefghijklmnop\n' + '"refresh_token":"sk-ant-oat01-abcdefghijklmnopqrstuvwxyz"' + ) + assert "super-secret" not in text + assert "abcdefghijklmnop" not in text + assert "abcdefghijklmnopqrstuvwxyz" not in text + assert len(text.encode()) <= protocol.MAX_ACTIVITY_BYTES + + class FakeKanban: + @staticmethod + def worker_log_path(_task, board): + assert board == "atlas" + return str(tmp_path / "worker.log") + + envelope = {**binding(), "payload": {"activity": "visible worker activity\n"}} + coordinator._append_activity(FakeKanban, envelope) + assert (tmp_path / "worker.log").read_text() == "visible worker activity\n" + (tmp_path / "worker.log").unlink() + (tmp_path / "worker.log").symlink_to(tmp_path / "target") + with pytest.raises(protocol.ProtocolError, match="symlink"): + coordinator._append_activity(FakeKanban, envelope) + + +def _documents(path): + return [item for item in yaml.safe_load_all(path.read_text()) if item] + + +def test_three_node_statefulset_contract_and_cross_worker_isolation(): + stateful = _documents(HERMES / "execution-worker-statefulset.yaml")[0] + pod = stateful["spec"]["template"]["spec"] + assert stateful["spec"]["replicas"] == 3 + assert stateful["spec"]["podManagementPolicy"] == "Parallel" + assert pod["automountServiceAccountToken"] is False + assert len(stateful["spec"]["volumeClaimTemplates"]) == 1 + claim = stateful["spec"]["volumeClaimTemplates"][0] + assert claim["metadata"]["name"] == "workspace" + assert claim["spec"]["accessModes"] == ["ReadWriteOnce"] + anti = pod["affinity"]["podAntiAffinity"]["requiredDuringSchedulingIgnoredDuringExecution"] + assert anti[0]["topologyKey"] == "kubernetes.io/hostname" + spread = pod["topologySpreadConstraints"][0] + assert spread["whenUnsatisfiable"] == "DoNotSchedule" + assert spread["maxSkew"] == 1 + + +def test_worker_placement_prefers_accelerators_and_preserves_exclusions(): + stateful = _documents(HERMES / "execution-worker-statefulset.yaml")[0] + affinity = stateful["spec"]["template"]["spec"]["affinity"]["nodeAffinity"] + terms = affinity["requiredDuringSchedulingIgnoredDuringExecution"]["nodeSelectorTerms"] + accelerator = {item["key"]: item for item in terms[0]["matchExpressions"]} + assert accelerator["kubernetes.io/hostname"]["values"] == ["titan-20", "titan-21"] + assert accelerator["node-role.kubernetes.io/accelerator"]["operator"] == "Exists" + fallback = {item["key"]: item for item in terms[1]["matchExpressions"]} + assert fallback["hardware"]["values"] == ["rpi5"] + assert "node-role.kubernetes.io/worker" not in fallback + excluded = set(fallback["kubernetes.io/hostname"]["values"]) + assert {"titan-04", "titan-08", "titan-13", "titan-14", "titan-17", "titan-18", "titan-19", "titan-22", "titan-24"} <= excluded + assert affinity["preferredDuringSchedulingIgnoredDuringExecution"][0]["weight"] == 100 + + +def test_model_worker_has_no_scm_or_cluster_credential_mount(): + stateful = _documents(HERMES / "execution-worker-statefulset.yaml")[0] + pod = stateful["spec"]["template"] + containers = {item["name"]: item for item in pod["spec"]["containers"]} + worker = containers["execution-worker"] + environment = {item["name"] for item in worker["env"]} + assert not {"GITEA_TOKEN", "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "KUBECONFIG"} & environment + mounts = {item["mountPath"] for item in worker["volumeMounts"]} + assert "/vault/secrets" not in mounts + assert "/pool-access" not in mounts + assert not any(item["name"] == "vault-auth-token" for item in worker["volumeMounts"]) + assert pod["metadata"]["annotations"]["vault.hashicorp.com/agent-inject-containers"] == "stage-worker-access" + assert any(item["name"] == "vault-auth-token" for item in containers["credential-sync"]["volumeMounts"]) + for boundary_name in ("execution-client", "scm-boundary"): + assert any( + item["mountPath"] == "/pool-access" + for item in containers[boundary_name]["volumeMounts"] + ) + scm_mounts = {item["mountPath"] for item in containers["scm-boundary"]["volumeMounts"]} + assert "/scm-access" in scm_mounts + assert "/vault/secrets" not in scm_mounts + + +def test_worker_service_account_has_no_kubernetes_permissions(): + documents = _documents(HERMES / "execution-worker-rbac.yaml") + assert len(documents) == 1 + account = documents[0] + assert account["kind"] == "ServiceAccount" + assert account["automountServiceAccountToken"] is False + + +def test_coordinator_remains_single_state_owner_and_workers_do_not_mount_home(): + agent = _documents(HERMES / "agent-deployment.yaml")[0] + assert agent["kind"] == "Deployment" + assert agent["spec"]["replicas"] == 1 + assert agent["spec"]["strategy"]["type"] == "Recreate" + assert any( + item["name"] == "home" + and item["persistentVolumeClaim"]["claimName"] == "hermes-agent-home" + for item in agent["spec"]["template"]["spec"]["volumes"] + ) + worker_text = (HERMES / "execution-worker-statefulset.yaml").read_text() + assert "hermes-agent-home" not in worker_text + assert "kanban.db" not in worker_text + + +def test_worker_protocol_preserves_switchyard_fallback_and_visible_evidence(): + source = (SCRIPTS / "execution_pool_worker.py").read_text() + coordinator_source = (SCRIPTS / "execution_pool_coordinator.py").read_text() + assert "cli_lane_runner.select_route" in source + assert 'alternate = "claude" if route.provider == "codex" else "codex"' in source + assert "codex_thread_id" in source and "claude_session_id" in source + assert "final_activity" in source + assert "worker_ordinal" in coordinator_source and "provider_sessions" in coordinator_source + assert "coordinator.reconcile()" in coordinator_source + + +def test_retention_gc_removes_only_clean_terminal_workspace(tmp_path, monkeypatch): + monkeypatch.setattr(worker, "ROOT", tmp_path) + monkeypatch.setattr(worker, "RETENTION_SECONDS", 3600) + workspace = tmp_path / "runs/atlas/t_deadbeef/run-1234" + workspace.mkdir(parents=True) + subprocess.run(["git", "init", "-q", str(workspace)], check=True) + subprocess.run(["git", "-C", str(workspace), "config", "user.email", "test@example.com"], check=True) + subprocess.run(["git", "-C", str(workspace), "config", "user.name", "Test"], check=True) + (workspace / "tracked").write_text("safe\n") + subprocess.run(["git", "-C", str(workspace), "add", "tracked"], check=True) + subprocess.run(["git", "-C", str(workspace), "commit", "-qm", "initial"], check=True) + state_file = tmp_path / "session-state/atlas/t_deadbeef/run-1234.json" + state_file.parent.mkdir(parents=True) + state_file.write_text( + json.dumps({"terminal_at": time.time() - 7200, "workspace": str(workspace)}) + ) + assert worker.garbage_collect() == 1 + assert not workspace.exists() + assert not state_file.exists() diff --git a/testing/tests/test_hermes_execution_pool_assignment.py b/testing/tests/test_hermes_execution_pool_assignment.py new file mode 100644 index 00000000..cf1d1a3e --- /dev/null +++ b/testing/tests/test_hermes_execution_pool_assignment.py @@ -0,0 +1,158 @@ +"""SCM assignment tests that keep coordinator credentials out of worktrees.""" + +from __future__ import annotations + +import sys +from pathlib import Path +from http.server import BaseHTTPRequestHandler +from types import SimpleNamespace + +import pytest +import yaml + + +SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts" +HERMES = SCRIPTS.parent +sys.path.insert(0, str(SCRIPTS)) + +import execution_pool_coordinator as coordinator # noqa: E402 +import execution_pool_protocol as protocol # noqa: E402 +import execution_pool_scm as scm # noqa: E402 +import execution_pool_worker as worker # noqa: E402 + + +KEY = b"k" * 32 + + +def test_new_task_uses_explicit_reviewed_atlas_default(monkeypatch): + monkeypatch.setattr( + coordinator, + "DEFAULT_REPO", + "https://scm.bstein.dev/atlas/titan-iac.git", + ) + task = SimpleNamespace(id="t_deadbeef", workspace_path="", branch_name="") + + assert coordinator.resolve_scm(task) == ( + "https://scm.bstein.dev/atlas/titan-iac.git", + "feature/hermes-t_deadbeef", + "main", + ) + + +def test_legacy_local_workspace_fails_closed_without_running_git(tmp_path): + task = SimpleNamespace( + id="t_deadbeef", workspace_path=str(tmp_path), branch_name="feature/safe" + ) + + with pytest.raises(RuntimeError, match="preserve or commit"): + coordinator.resolve_scm(task) + + +def test_local_git_environment_excludes_credential_boundary_paths(): + local = scm._git_env(False) + authenticated = scm._git_env(True) + assert "HERMES_SCM_PASSWORD_FILE" not in local + assert "GIT_ASKPASS" not in local + assert authenticated["HERMES_SCM_PASSWORD_FILE"] == str(scm.TOKEN_PATH) + + +def test_finalized_duplicate_result_is_acknowledged_without_refinalizing(tmp_path): + binding = { + "board": "atlas", "task_id": "t_deadbeef", "run_id": "run-1", + "worker_ordinal": 0, "attempt": 1, + } + store = protocol.PoolStore(tmp_path / "pool.db") + store.add(binding, {"context": "safe"}) + result = protocol.sign_envelope(KEY, "result", binding, {"structured": {}}) + store.accept_result(result) + store.finalize(binding, "finalized") + pool = coordinator.Coordinator(KEY, store) + called = [] + pool.finalize = called.append + + assert pool.result(result)["payload"]["duplicate"] is True + assert called == [] + + +def test_provider_sessions_are_bound_to_exact_task_run(tmp_path, monkeypatch): + runtime = tmp_path / "runtime" + (runtime / "codex").mkdir(parents=True) + (runtime / "claude").mkdir(parents=True) + monkeypatch.setattr(worker, "ROOT", tmp_path / "worker") + (tmp_path / "worker/provider-state").mkdir(parents=True) + monkeypatch.setattr(worker.cli_lane_runner, "DATA_ROOT", tmp_path / "worker-data") + (tmp_path / "worker-data").mkdir() + monkeypatch.setenv("CODEX_HOME", str(runtime / "codex")) + monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(runtime / "claude")) + assignment = {"board": "atlas", "task_id": "t_deadbeef", "run_id": "run-1"} + + worker._bind_provider_sessions(assignment) + + for path in ( + runtime / "codex/sessions", runtime / "claude/projects", + runtime / "claude/session-env", runtime / "claude/todos", + ): + assert path.is_symlink() + assert "atlas/t_deadbeef/run-1" in str(path.resolve()) + assert (tmp_path / "worker-data/home").is_symlink() + assert "atlas/t_deadbeef/run-1" in str((tmp_path / "worker-data/home").resolve()) + + +def test_internal_http_server_rejects_work_above_its_bound(monkeypatch): + server = protocol.BoundedHTTPServer( + ("127.0.0.1", 0), BaseHTTPRequestHandler, max_workers=1 + ) + rejected = [] + monkeypatch.setattr(server, "shutdown_request", rejected.append) + assert server._slots.acquire(blocking=False) + try: + marker = object() + server.process_request(marker, ("127.0.0.1", 1)) + assert rejected == [marker] + finally: + server._slots.release() + server.server_close() + + +def test_additive_patch_replaces_local_lane_without_touching_base_deployment(): + patch = yaml.safe_load((HERMES / "execution-coordinator-patch.yaml").read_text()) + containers = patch["spec"]["template"]["spec"]["containers"] + deleted = next(item for item in containers if item["name"] == "cli-lane-runner") + pool = next(item for item in containers if item["name"] == "execution-pool-coordinator") + assert deleted["$patch"] == "delete" + assert pool["resources"]["requests"] == {"cpu": "50m", "memory": "128Mi"} + access = next(item for item in pool["volumeMounts"] if item["name"] == "runtime-access") + assert access["subPath"] == "execution-pool-key" and access["readOnly"] is True + + +def test_additive_network_policies_expose_only_worker_pool_and_switchyard_ports(): + documents = list(yaml.safe_load_all( + (HERMES / "execution-worker-networkpolicy.yaml").read_text() + )) + policies = {item["metadata"]["name"]: item for item in documents} + assert set(policies) == { + "hermes-execution-worker-isolation", + "hermes-execution-pool-ingress", + "hermes-execution-switchyard-ingress", + } + assert policies["hermes-execution-pool-ingress"]["spec"]["ingress"][0]["ports"] == [ + {"protocol": "TCP", "port": 9007} + ] + + +@pytest.mark.parametrize( + "repo,base", + [ + ("https://evil.example/atlas/titan-iac.git", "main"), + ("https://token@scm.bstein.dev/atlas/titan-iac.git", "main"), + ("https://scm.bstein.dev/atlas/titan-iac.git", "../main"), + ], +) +def test_assignment_rejects_unreviewed_repo_or_base(repo, base): + task = SimpleNamespace( + id="t_deadbeef", workspace_path="", branch_name="feature/safe", + repo_url=repo, base_branch=base, + ) + + with pytest.raises(RuntimeError, match="outside"): + coordinator.resolve_scm(task) diff --git a/testing/tests/test_hermes_runtime_access.py b/testing/tests/test_hermes_runtime_access.py index 7e455eb6..26744d7b 100644 --- a/testing/tests/test_hermes_runtime_access.py +++ b/testing/tests/test_hermes_runtime_access.py @@ -174,6 +174,7 @@ def test_agent_runtime_stage_keeps_credentials_in_memory(tmp_path: Path, monkeyp ) values = { "agent-api-key": "agent-key", + "execution-pool-key": "e" * 64, "chat-relay-key": "relay-key", "gitea-token": "gitea-key", "gitea-username": "hermes-automation", @@ -204,6 +205,69 @@ def test_agent_runtime_stage_keeps_credentials_in_memory(tmp_path: Path, monkeyp assert auth == {"version": 1, "providers": {}, "credential_pool": {}} +def test_execution_worker_separates_channel_key_and_durable_sessions( + tmp_path: Path, monkeypatch +): + stage = _load("stage_runtime_access") + vault = tmp_path / "vault" + runtime = tmp_path / "runtime" + worker = tmp_path / "worker" + pool_access = tmp_path / "pool-access" + scm_access = tmp_path / "scm-access" + vault.mkdir() + (vault / "execution-pool-key").write_text("e" * 64) + (vault / "claude-credentials").write_text( + json.dumps({"claudeAiOauth": {"refreshToken": "claude-refresh"}}) + ) + (vault / "codex-auth").write_text( + json.dumps({"tokens": {"refresh_token": "codex-refresh"}}) + ) + (vault / "gitea-token").write_text("test-gitea-token") + (vault / "gitea-username").write_text("hermes-automation") + monkeypatch.setattr(stage, "VAULT_ROOT", vault) + monkeypatch.setattr(stage, "RUNTIME_ROOT", runtime) + monkeypatch.setattr(stage, "WORKER_ROOT", worker) + monkeypatch.setattr(stage, "POOL_ACCESS_ROOT", pool_access) + monkeypatch.setattr(stage, "SCM_ACCESS_ROOT", scm_access) + monkeypatch.setattr(stage.os, "chown", lambda *_args: None) + monkeypatch.setattr(stage.os, "fchown", lambda *_args: None) + + stage.stage_execution_worker() + + assert (pool_access / "execution-pool-key").read_text().strip() == "e" * 64 + assert not (runtime / "execution-pool-key").exists() + assert (scm_access / "gitea-token").read_text().strip() == "test-gitea-token" + assert not (runtime / "codex/sessions").exists() + assert not (runtime / "claude/projects").exists() + assert (runtime / "codex/auth.json").stat().st_mode & 0o777 == 0o600 + + +def test_execution_worker_fails_closed_without_channel_credential( + tmp_path: Path, monkeypatch +): + stage = _load("stage_runtime_access") + vault = tmp_path / "vault" + vault.mkdir() + (vault / "claude-credentials").write_text( + json.dumps({"claudeAiOauth": {"refreshToken": "claude-refresh"}}) + ) + (vault / "codex-auth").write_text( + json.dumps({"tokens": {"refresh_token": "codex-refresh"}}) + ) + (vault / "gitea-token").write_text("test-gitea-token") + (vault / "gitea-username").write_text("hermes-automation") + monkeypatch.setattr(stage, "VAULT_ROOT", vault) + monkeypatch.setattr(stage, "RUNTIME_ROOT", tmp_path / "runtime") + monkeypatch.setattr(stage, "WORKER_ROOT", tmp_path / "worker") + monkeypatch.setattr(stage, "POOL_ACCESS_ROOT", tmp_path / "pool-access") + monkeypatch.setattr(stage, "SCM_ACCESS_ROOT", tmp_path / "scm-access") + monkeypatch.setattr(stage.os, "chown", lambda *_args: None) + monkeypatch.setattr(stage.os, "fchown", lambda *_args: None) + + with pytest.raises(FileNotFoundError): + stage.stage_execution_worker() + + def test_invalid_runtime_json_is_removed(tmp_path: Path, monkeypatch): stage = _load("stage_runtime_access") vault = tmp_path / "vault"