From 4126f001a7c887b9b38ce6257db32c4bb95e8997 Mon Sep 17 00:00:00 2001 From: witbrock Date: Fri, 11 Sep 2026 10:46:18 -0400 Subject: [PATCH] Support explicitly requested DGX deployment with code recovery --- docs/engineering/codex_dgx_worker.md | 58 +++++- run.sh | 1 + scripts/codex_von_deploy.py | 199 ++++++++++++++++++ scripts/codex_von_worker.py | 93 ++++++++- scripts/codex_von_worker_prompt.md | 17 +- scripts/deploy_local_main.py | 62 ++++-- src/backend/security/cloudflare_access.py | 28 ++- tests/backend/test_cloudflare_access.py | 30 +++ tests/backend/test_codex_von_worker.py | 60 ++++++ tests/test_codex_von_deploy.py | 240 ++++++++++++++++++++++ tests/test_deploy_local_main.py | 56 +++-- 11 files changed, 803 insertions(+), 41 deletions(-) create mode 100755 scripts/codex_von_deploy.py mode change 100644 => 100755 scripts/deploy_local_main.py create mode 100644 tests/test_codex_von_deploy.py diff --git a/docs/engineering/codex_dgx_worker.md b/docs/engineering/codex_dgx_worker.md index 22c712833..48be506df 100644 --- a/docs/engineering/codex_dgx_worker.md +++ b/docs/engineering/codex_dgx_worker.md @@ -17,7 +17,7 @@ transcript separately. The current Tasks form's assignee picker offers only Unassigned and Von; it does not yet offer this worker. The pilot acceptance used labelled fixtures through canonical services, not that picker. -The worker checks hourly, starts at most one coding run per check, and sends a +The worker checks at the operator-configured interval, starts at most one coding run per check, and sends a pickup message followed by a result or question through Von direct messages. An empty check makes no model request. Each task has its own Git worktree. Each execution is a fresh subscription-backed Codex session, using the retained @@ -32,7 +32,8 @@ messages do not create new coding assignments in this pilot. Publishing uses an operator-configured GitHub account and follows each task's publication authority. Missing authentication leaves changes retained and the task blocked; it must not claim that local code is published. Deployment of -the public web server is outside this worker's execution profile. The chosen +the public web server requires an explicit instruction in the initial task and +the operator-enabled deployment command described below. The chosen GitHub account and current authentication status belong in the Jira decision surface, not in the task's text. @@ -67,6 +68,14 @@ prevents the worker from changing the task state. Its result still goes to the original configured delegator. This is a single-host worker, not a distributed claim/lease protocol. +When the initial task explicitly requests deployment, the coding run returns +the full merged `deploy_commit` after its code, tests and publication finish. +The controller rechecks the live task authority and instructions, then calls +only its operator-configured `deployment_command`. The run cannot supply a +host, checkout, shell command or credentials. An empty deployment request has +no deployment effect. Interpretation of the initial task belongs to the model; +there is no keyword matcher or second per-deployment confirmation. + The Codex process defaults to `workspace-write` sandboxing and an explicit non-Sol model. Publication installations use the scoped profile below. Its environment omits the controller's Mongo credentials and unrelated @@ -141,6 +150,47 @@ restrictions; preserve their work and migrate them explicitly if publication is required. Never grant writes to the live web checkout's shared `.git` as a shortcut. See [Codex permissions](https://learn.chatgpt.com/docs/permissions). +## Explicit deployment + +Install `codex_von_deploy.py` and its `deploy_local_main.py` dependency in the +operator-owned directory. A fixed launcher binds a separate JSON configuration +and accepts only `--commit` and `--receipt` from the controller. Set its path as +`deployment_command` in worker config. The deployment configuration contains: + +```json +{ + "primary_root": "/home/mjw/Von", + "runtime_root": "/home/mjw/Von-runtime-main", + "state_root": "/home/mjw/.codex-von-worker/worker-state", + "health_url": "http://127.0.0.1:5000/health", + "public_health_url": "https://von.curiouscat.cc/health", + "public_headers_file": "/home/mjw/.codex-von-worker/cloudflare-health.json", + "health_timeout_seconds": 960 +} +``` + +Prepare the dedicated detached runtime worktree with its environment before +activation. The command uses the canonical launcher, requires the requested +SHA to remain `origin/main`, and verifies the exact clean revision locally and +through the public endpoint. Persistent receipts make retries reconcile an +interrupted deployment. Failed startup restores the previous code revision +and verifies the recovered service; a recovery failure remains an explicit +blocked outcome. This is code recovery, not a database rollback. A changed +`pdm.lock` requires preparation of a recoverable environment before automatic +deployment. Migrations and infrastructure changes need their own task scope +and recovery plan. + +For a Cloudflare Access-protected site, keep the two service-token headers in +the private `public_headers_file`. The deployment command injects only the +service Client ID into `VON_CLOUDFLARE_HEALTH_SERVICE_IDS` for the server. +Cloudflare must issue the token for the configured application's Service Auth +policy. Von validates its signature, issuer, audience, expiry and exact +configured service identity for `GET /health` only; it never creates a Von user +session. Human login and private routes retain their existing checks. Health +requests do not follow redirects, so credentials cannot be forwarded to a +login provider. Token expiry belongs in the operator's installation record. +See [Cloudflare service tokens](https://developers.cloudflare.com/cloudflare-one/access-controls/service-credentials/service-tokens/). + The DGX operator wrapper reads the existing web runtime's database credential in-process, checks that the configured database target still matches the public website, binds the worker config, and suppresses event-driven workflow @@ -148,9 +198,9 @@ launches in that controller process. It does not alter the web process's configuration. Store credentials, configuration and operational state outside the repository with owner-only permissions. Never log the database URI. -On a host with an active cron service, install an hourly entry invoking the +On a host with an active cron service, install a five-minute entry invoking the wrapper and redirecting stdout/stderr to an owner-only log. An example schedule -is `17 * * * *`; preserve any other existing crontab entries. Run the exact +is `*/5 * * * *`; preserve any other existing crontab entries. Run the exact entry manually before installation. No Codex app session or SSH connection needs to remain open. The controller uses a nonblocking file lock and passes it to its child so concurrent invocations do not start overlapping work. diff --git a/run.sh b/run.sh index 0e2510992..1a77c47af 100755 --- a/run.sh +++ b/run.sh @@ -3634,6 +3634,7 @@ deploy_main() { "$py" "${ROOT}/scripts/deploy_local_main.py" \ --primary-root "$ROOT" \ --runtime-worktree "$runtime_path" \ + --health-url "http://127.0.0.1:${PORT}/health" \ --health-timeout-seconds "$HEALTH_TIMEOUT_SEC" } diff --git a/scripts/codex_von_deploy.py b/scripts/codex_von_deploy.py new file mode 100755 index 000000000..dcc8f6d91 --- /dev/null +++ b/scripts/codex_von_deploy.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Operator-bound deployment and code recovery for the external Codex worker. + +Paths and endpoints come only from the operator's config. The coding run may +request a merged revision, never a command, host, credential or checkout path. +""" + +from __future__ import annotations + +import argparse +import fcntl +import json +import os +import re +from pathlib import Path +from urllib.parse import urlparse + +try: + from . import deploy_local_main as runtime +except ImportError: + import deploy_local_main as runtime + + +def save(path, value): + temporary = path.with_suffix(".tmp") + with temporary.open("w") as stream: + json.dump(value, stream, indent=2) + stream.flush() + os.fsync(stream.fileno()) + temporary.replace(path) + + +def verify(config, commit): + timeout = config.get("health_timeout_seconds", 960) + local = runtime._wait_for_verified_health(config["health_url"], commit, timeout) + headers = None + if config.get("public_headers_file"): + headers = json.loads(Path(config["public_headers_file"]).read_text()) + if set(headers) != {"CF-Access-Client-Id", "CF-Access-Client-Secret"}: + raise ValueError("Unexpected public health authentication headers") + public = runtime._wait_for_verified_health( + config["public_health_url"], commit, timeout, headers=headers + ) + return {"local_pid": local["pid"], "public_pid": public["pid"], "commit": commit} + + +def recover(config, previous_commit): + primary = Path(config["primary_root"]).resolve() + checkout = Path(config["runtime_root"]).resolve() + # Stop only the exact deployment port using the canonical launcher. Changing + # code while a candidate process is still alive would invalidate read-back. + launcher = checkout / "run.sh" + port = str(urlparse(config["health_url"]).port) + runtime._run( + ["bash", launcher, "stop", "-Port", port, "-NoBrowser"], + cwd=checkout, + capture_output=False, + ) + runtime._prepare_runtime(primary, checkout, previous_commit) + runtime._reconcile_startup_seed_materialisations(checkout) + runtime._run( + [ + "bash", + launcher, + "restart", + "-Port", + port, + "-NoBrowser", + "-HealthTimeoutSec", + str(config.get("health_timeout_seconds", 960)), + ], + cwd=checkout, + capture_output=False, + ) + runtime._verify_workers(checkout) + # An older release can predate machine health authentication. Restore and + # verify its local service even if that old public route rejects this token. + local = runtime._wait_for_verified_health( + config["health_url"], previous_commit, config.get("health_timeout_seconds", 960) + ) + try: + return verify(config, previous_commit) + except runtime.HealthAccessError: + return { + "commit": previous_commit, + "local_pid": local["pid"], + "public_verification": "authentication_unavailable_for_previous_release", + } + + +def execute(config, commit, receipt_path, *, recover_only=False): + if not re.fullmatch(r"[0-9a-f]{40}", commit): + raise ValueError("Deployment requires a full Git commit SHA") + primary = Path(config["primary_root"]).resolve() + checkout = Path(config["runtime_root"]).resolve() + runtime._validate_runtime(primary, checkout) + receipt_path = Path(receipt_path) + receipt_path.parent.mkdir(parents=True, exist_ok=True) + with (Path(config["state_root"]) / "deployment.lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + receipt = ( + json.loads(receipt_path.read_text()) if receipt_path.exists() else None + ) + if receipt: + if receipt["requested_commit"] != commit: + raise ValueError("Deployment receipt belongs to another revision") + if receipt["status"] in {"deployed", "rolled_back", "recovery_failed"}: + return receipt + else: + if recover_only: + raise ValueError("Recovery requires an existing deployment receipt") + # Exact target and healthy predecessor are preconditions: a stale + # request must not stop a healthy server or fast-forward its checkout. + runtime._run(["git", "-C", primary, "fetch", "origin", "main"]) + if runtime._git(primary, "rev-parse", "origin/main") != commit: + raise runtime.DeploymentError( + "Requested revision is no longer origin/main" + ) + prior = runtime._read_health(config["health_url"]) + previous_commit = prior["version_details"]["git_commit"] + error = runtime._health_error(prior, previous_commit) + if error: + raise runtime.DeploymentError("Predecessor is not ready: " + error) + if runtime._git(primary, "show", f"{commit}:pdm.lock") != runtime._git( + primary, "show", f"{previous_commit}:pdm.lock" + ): + raise runtime.DeploymentError( + "Dependency lock changed; prepare a recoverable environment before deployment" + ) + receipt = { + "status": "started", + "requested_commit": commit, + "previous_commit": previous_commit, + } + save(receipt_path, receipt) + try: + runtime.deploy( + primary_root=primary, + runtime_root=checkout, + health_url=config["health_url"], + health_timeout_seconds=config.get("health_timeout_seconds", 960), + expected_commit=commit, + ) + receipt.update(status="deployed", verification=verify(config, commit)) + save(receipt_path, receipt) + return receipt + except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc: + receipt["failure_type"] = type(exc).__name__ + save(receipt_path, receipt) + # An interrupted controller never blindly repeats a restart. Reconcile + # the target first; if it is not serving, restore the recorded predecessor. + try: + current = runtime._read_health(config["health_url"]) + if not runtime._health_error(current, commit): + receipt.update(status="deployed", verification=verify(config, commit)) + save(receipt_path, receipt) + return receipt + except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc: + receipt["reconciliation_failure_type"] = type(exc).__name__ + try: + receipt.update( + status="rolled_back", + verification=recover(config, receipt["previous_commit"]), + ) + except (runtime.DeploymentError, OSError, ValueError, KeyError) as exc: + receipt.update( + status="recovery_failed", recovery_failure_type=type(exc).__name__ + ) + save(receipt_path, receipt) + return receipt + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--config", required=True) + parser.add_argument("--commit", required=True) + parser.add_argument("--receipt", required=True) + parser.add_argument("--recover-only", action="store_true") + args = parser.parse_args() + os.umask(0o077) + config = json.loads(Path(args.config).read_text()) + if config.get("public_headers_file"): + headers = json.loads(Path(config["public_headers_file"]).read_text()) + os.environ["VON_CLOUDFLARE_HEALTH_SERVICE_IDS"] = headers["CF-Access-Client-Id"] + # Controller-only event suppression must not change the production service. + for key in ( + "VON_EVENT_WORKFLOW_INTEGRATION_ENABLE", + "VON_DURABLE_WORKFLOWS_ENABLE", + ): + os.environ.pop(key, None) + result = execute( + config, args.commit, Path(args.receipt), recover_only=args.recover_only + ) + print(json.dumps(result), flush=True) + return 0 if result["status"] == "deployed" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index eab845c94..eeeceee35 100755 --- a/scripts/codex_von_worker.py +++ b/scripts/codex_von_worker.py @@ -29,8 +29,9 @@ "summary": {"type": "string"}, "evidence": {"type": "string"}, "question": {"type": "string"}, + "deploy_commit": {"type": "string"}, }, - "required": ["status", "summary", "evidence", "question"], + "required": ["status", "summary", "evidence", "question", "deploy_commit"], "additionalProperties": False, } @@ -62,7 +63,11 @@ def authorised_task(task, config): def validate_result(value): - if not isinstance(value, dict) or set(value) != set(RESULT_SCHEMA["required"]): + fields = set(RESULT_SCHEMA["required"]) + if not isinstance(value, dict) or set(value) not in ( + fields, + fields - {"deploy_commit"}, + ): raise ValueError("Codex did not return the required result fields") if any(not isinstance(v, str) for v in value.values()): raise ValueError("Codex result fields must be text") @@ -380,6 +385,7 @@ def launch(config, state, inputs, conversation, state_path, lock_fd): state.pop("result", None) state.pop("report_task", None) state.pop("report_content", None) + state.pop("deployment_final", None) write_json(state_path, state) if not state.get("checkout_prepared"): worktree.parent.mkdir(parents=True, exist_ok=True) @@ -507,6 +513,87 @@ def recover_result(state): state.update(phase="reporting", result=result) +def apply_deployment(config, api, state, state_path, lock_fd): + """Execute only an operator-enabled request; retain the canonical receipt.""" + result = state["result"] + commit = result.get("deploy_commit", "").strip() + if not commit or state.get("deployment_final"): + return + task = api.task(state["task_id"]) + eligible = ( + result["status"] == "completed" + and bool(task) + and authorised_task(task, config) + and task.get("status") in ACTIVE + and api.native_writer(task) + and fingerprint(api.inputs(task)) == state.get("input_hash") + ) + run_dir = Path(state["run_dir"]) + receipt_path = run_dir / "deployment.json" + try: + prior_receipt = json.loads(receipt_path.read_text()) + except (OSError, ValueError): + prior_receipt = {} + recover_only = not eligible and prior_receipt.get("status") == "started" + if (not eligible and not recover_only) or not config.get("deployment_command"): + detail = ( + "The task authority or instructions changed before deployment." + if not eligible + else "Deployment is not enabled for this worker." + ) + result.update(status="blocked", summary=result["summary"] + " " + detail) + state["deployment_final"] = "not_authorised" + write_json(state_path, state) + return + with (run_dir / "deployment.log").open("a") as log: + completed = subprocess.run( + [ + config["deployment_command"], + "--commit", + commit, + "--receipt", + str(receipt_path), + *(["--recover-only"] if recover_only else []), + ], + check=False, + stdout=log, + stderr=log, + pass_fds=(lock_fd,), + ) + try: + receipt = json.loads(receipt_path.read_text()) + except (OSError, ValueError): + receipt = {"status": "failed_without_receipt"} + success = ( + completed.returncode == 0 + and receipt.get("status") == "deployed" + and receipt.get("requested_commit") == commit + and receipt.get("verification", {}).get("commit") == commit + ) + state["deployment_final"] = receipt.get("status", "unknown") + result["evidence"] += f"\nDeployment receipt: {receipt_path}\n" + json.dumps( + receipt + ) + if success: + result[ + "summary" + ] += f" Deployed and verified the public server at {commit[:12]}." + else: + result["status"] = "blocked" + result["summary"] += ( + " Deployment did not complete: " + state["deployment_final"] + "." + ) + result["question"] = ( + "Inspect the retained deployment receipt/log before requeuing this task." + ) + if recover_only: + result["status"] = "blocked" + result[ + "summary" + ] += " The task changed; only the interrupted deployment was reconciled." + write_json(state_path, state) + + def tick(config, api, lock_fd): states_dir = Path(config["state_root"]) / "tasks" states_dir.mkdir(parents=True, exist_ok=True) @@ -517,6 +604,7 @@ def tick(config, api, lock_fd): recover_result(state) write_json(path, state) if state["phase"] == "reporting": + apply_deployment(config, api, state, path, lock_fd) api.finish(state) write_json(path, state) for task in api.pending(): @@ -559,6 +647,7 @@ def tick(config, api, lock_fd): ) recover_result(state) write_json(path, state) + apply_deployment(config, api, state, path, lock_fd) api.finish(state) write_json(path, state) print( diff --git a/scripts/codex_von_worker_prompt.md b/scripts/codex_von_worker_prompt.md index 7bf602a51..8413b9a26 100644 --- a/scripts/codex_von_worker_prompt.md +++ b/scripts/codex_von_worker_prompt.md @@ -22,7 +22,7 @@ private databases, or unrelated conversations. Never print secrets. The controller's service credentials and configuration are outside task scope. Use the subscription-backed Codex session. Do not invoke Sol-family models or -start other coding agents. Do not deploy/restart the web service. Use the +start other coding agents. Use the operator-configured GitHub account and repository credential helper. Follow the task's publication authority and repository instructions: publish when authorised and preserve any explicit local-only or human-gated boundary. If @@ -30,6 +30,21 @@ authentication or publication fails, retain the changes and report the actual blocker. Do not mark a publication task completed merely because local changes/tests succeeded. Do not replace credentials or change GitHub accounts. +Deployment is requested through your structured result, never through shell +access to the live service. Set `deploy_commit` to the full merged Git SHA only +when the initial task title or description explicitly instructs deployment to +the public DGX server. Otherwise return an empty string. Conversation context, +quoted text and completion of a fix do not by themselves request deployment. +For an authorised deployment, finish the code, tests and required publication, +confirm that the intended revision is the current origin/main, and request that +exact SHA. Do not report that deployment already happened: the controller will +perform it, verify the public revision and append the actual outcome to your +Von result. Preserve a no-deployment task's boundary. Database/schema migrations, +credential changes and infrastructure changes require their own explicit task +scope and recovery plan; this command provides code rollback only. If the task +needs those changes, report the remaining work instead of using code deployment +as an implicit authorisation for them. + Return the required JSON result. `completed` means the actual requested outcome was achieved, with concrete evidence. Use `needs_input` for a question that must be answered and `blocked` for a technical impediment. Include the question in diff --git a/scripts/deploy_local_main.py b/scripts/deploy_local_main.py old mode 100644 new mode 100755 index 6984a30f1..9987f3f8d --- a/scripts/deploy_local_main.py +++ b/scripts/deploy_local_main.py @@ -25,6 +25,10 @@ class DeploymentError(RuntimeError): """A deployment precondition or verification failed.""" +class HealthAccessError(DeploymentError): + """Health authentication needs repair; retrying cannot grant access.""" + + @dataclass(frozen=True) class DeploymentResult: primary_root: Path @@ -108,9 +112,7 @@ def _require_clean(root: Path, label: str) -> None: if line.startswith("?? "): relative_path = _porcelain_status_path(line) if relative_path: - candidate = Path( - os.path.abspath(root / relative_path.rstrip("/")) - ) + candidate = Path(os.path.abspath(root / relative_path.rstrip("/"))) if candidate in nested_worktree_roots: continue blocking_entries.append(line) @@ -133,6 +135,7 @@ def _prepare_primary( *, remote: str, branch: str, + expected_commit: str | None = None, ) -> str: actual_root = Path(_git(primary_root, "rev-parse", "--show-toplevel")).resolve() if actual_root != primary_root: @@ -152,6 +155,10 @@ def _prepare_primary( local_commit = _git(primary_root, "rev-parse", "HEAD") remote_ref = f"{remote}/{branch}" target_commit = _git(primary_root, "rev-parse", remote_ref) + if expected_commit is not None and target_commit != expected_commit: + raise DeploymentError( + f"Requested commit {expected_commit} is no longer {remote_ref} ({target_commit})" + ) if local_commit != target_commit: merge_base = _git(primary_root, "merge-base", local_commit, target_commit) if merge_base != local_commit: @@ -172,7 +179,9 @@ def _prepare_primary( def _validate_runtime(primary_root: Path, runtime_root: Path) -> None: if runtime_root == primary_root: - raise DeploymentError("Runtime worktree must be distinct from the primary checkout") + raise DeploymentError( + "Runtime worktree must be distinct from the primary checkout" + ) if not runtime_root.is_dir(): raise DeploymentError( "Dedicated runtime worktree does not exist or is not a directory: " @@ -206,7 +215,9 @@ def _prepare_runtime( f"Runtime checkout is at {runtime_commit}, expected {target_commit}" ) if _git(runtime_root, "symbolic-ref", "-q", "HEAD", check=False): - raise DeploymentError("Runtime checkout must be detached, but it is on a branch") + raise DeploymentError( + "Runtime checkout must be detached, but it is on a branch" + ) def _reconcile_startup_seed_materialisations( @@ -239,16 +250,25 @@ def _reconcile_startup_seed_materialisations( "Startup seed reconciliation did not return a JSON receipt" ) from exc if not isinstance(payload, dict) or payload.get("success") is not True: - raise DeploymentError( - f"Startup seed reconciliation was not ready: {payload!r}" - ) + raise DeploymentError(f"Startup seed reconciliation was not ready: {payload!r}") return payload -def _read_health(url: str) -> dict[str, Any]: +class _NoHealthRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def _read_health(url: str, headers: dict[str, str] | None = None) -> dict[str, Any]: try: - with urllib.request.urlopen(url, timeout=5) as response: + opener = urllib.request.build_opener(_NoHealthRedirect()) + request = urllib.request.Request(url, headers=headers or {}) + with opener.open(request, timeout=5) as response: payload = json.load(response) + except urllib.error.HTTPError as exc: + if exc.code in {301, 302, 303, 307, 308, 401, 403}: + raise HealthAccessError(f"Health access returned HTTP {exc.code}") from exc + raise DeploymentError(f"Health read returned HTTP {exc.code}") from exc except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc: raise DeploymentError(f"Health read failed: {exc}") from exc if not isinstance(payload, dict): @@ -374,7 +394,9 @@ def _health_error(payload: dict[str, Any], target_commit: str) -> str | None: return f"running git_dirty={version.get('git_dirty')!r}" authority = payload.get("runtime_authority") - durable = authority.get("durable_workflows") if isinstance(authority, dict) else None + durable = ( + authority.get("durable_workflows") if isinstance(authority, dict) else None + ) if not isinstance(durable, dict): return "durable workflow health missing" required = ("database_connected", "worker_running", "scheduler_running") @@ -403,15 +425,18 @@ def _wait_for_verified_health( url: str, target_commit: str, timeout_seconds: int, + headers: dict[str, str] | None = None, ) -> dict[str, Any]: deadline = time.monotonic() + timeout_seconds last_error = "health endpoint not read" while time.monotonic() < deadline: try: - payload = _read_health(url) + payload = _read_health(url, headers) if headers else _read_health(url) last_error = _health_error(payload, target_commit) or "" if not last_error: return payload + except HealthAccessError: + raise except DeploymentError as exc: last_error = str(exc) time.sleep(1) @@ -471,10 +496,15 @@ def deploy( branch: str = "main", health_url: str = "http://127.0.0.1:5001/health", health_timeout_seconds: int = 960, + expected_commit: str | None = None, ) -> DeploymentResult: primary_root = primary_root.resolve() runtime_root = runtime_root.resolve() - target_commit = _prepare_primary(primary_root, remote=remote, branch=branch) + target_commit = _prepare_primary( + primary_root, remote=remote, branch=branch, expected_commit=expected_commit + ) + parsed_health = urllib.parse.urlparse(health_url) + port = str(parsed_health.port or (443 if parsed_health.scheme == "https" else 80)) current_launcher = primary_root / "run.sh" if not current_launcher.is_file(): raise DeploymentError(f"Current launcher missing: {current_launcher}") @@ -490,7 +520,7 @@ def deploy( stop_environment = os.environ.copy() stop_environment["VON_LAUNCHER_ROOT"] = str(runtime_root) _run( - ["bash", current_launcher, "stop", "-NoBrowser"], + ["bash", current_launcher, "stop", "-Port", port, "-NoBrowser"], cwd=runtime_root, capture_output=False, env=stop_environment, @@ -505,6 +535,8 @@ def deploy( "bash", launcher, "start", + "-Port", + port, "-NoBrowser", "-HealthTimeoutSec", str(health_timeout_seconds), @@ -544,6 +576,7 @@ def _parser() -> argparse.ArgumentParser: parser.add_argument("--runtime-worktree", type=Path, required=True) parser.add_argument("--remote", default="origin") parser.add_argument("--branch", default="main") + parser.add_argument("--expected-commit") parser.add_argument("--health-url", default="http://127.0.0.1:5001/health") parser.add_argument("--health-timeout-seconds", type=int, default=960) return parser @@ -562,6 +595,7 @@ def main(argv: Sequence[str] | None = None) -> int: branch=args.branch, health_url=args.health_url, health_timeout_seconds=args.health_timeout_seconds, + expected_commit=args.expected_commit, ) except DeploymentError as exc: print(f"deploy-main: ERROR: {exc}", file=sys.stderr) diff --git a/src/backend/security/cloudflare_access.py b/src/backend/security/cloudflare_access.py index 86c98ce43..07c65a5f1 100644 --- a/src/backend/security/cloudflare_access.py +++ b/src/backend/security/cloudflare_access.py @@ -62,7 +62,7 @@ def _certificates(self, kid: str) -> dict[str, str]: self._fetched_at = time.monotonic() return dict(self._keys) - def verify(self, token: str) -> dict: + def verify(self, token: str, *, health_service_ids: frozenset[str] = frozenset()) -> dict: if not token or len(token) > 32768: raise ValueError("access_token_missing_or_invalid") header = jwt.decode_header(token) @@ -71,8 +71,14 @@ def verify(self, token: str) -> dict: # google-auth verifies signature, iat and exp. Access uses a list-valued # audience, so validate that explicitly after signature verification. claims = jwt.decode(token, certs=self._certificates(header["kid"])) + health_service = (claims.get("sub") == "" and not claims.get("email") + and claims.get("common_name") in health_service_ids) aud = claims.get("aud") - if not isinstance(aud, list) or self.audience not in aud: + audience_valid = isinstance(aud, list) and self.audience in aud + # Live Cloudflare service issuance also uses the JWT scalar audience form. + if health_service and isinstance(aud, str) and aud == self.audience: + audience_valid = True + if not audience_valid: raise ValueError("access_audience_invalid") if claims.get("iss") != self.issuer or claims.get("type") != "app": raise ValueError("access_issuer_or_type_invalid") @@ -85,6 +91,10 @@ def verify(self, token: str) -> dict: type(claims["nbf"]) is not int or claims["nbf"] > time.time() ): raise ValueError("access_token_not_yet_valid") + # A configured machine may read health only; it never becomes a user. + # Callers supply this allow-list only for the exact GET /health endpoint. + if health_service: + return claims if not isinstance(claims.get("sub"), str) or not claims["sub"]: raise ValueError("access_subject_missing") # Service tokens lack a human email; they must never become a user. @@ -102,6 +112,10 @@ def install_cloudflare_access(app) -> None: hostname = os.getenv("VON_CLOUDFLARE_ACCESS_HOSTNAME", "").strip().lower() team = os.getenv("VON_CLOUDFLARE_ACCESS_TEAM_DOMAIN", "").strip().lower() audience = os.getenv("VON_CLOUDFLARE_ACCESS_AUDIENCE", "").strip() + health_service_ids = frozenset( + value.strip() for value in os.getenv("VON_CLOUDFLARE_HEALTH_SERVICE_IDS", "").split(",") + if value.strip() + ) if enabled and ( not re.fullmatch(r"[a-z0-9-]+\.cloudflareaccess\.com", team) or not re.fullmatch(r"[a-z0-9]+(?:[.-][a-z0-9-]+)+", hostname) @@ -125,7 +139,15 @@ def authenticate_access_request(): try: token = request.headers.get("Cf-Access-Jwt-Assertion", "") - claims = verifier.verify(token) + machine_health = request.endpoint == "health_check" and request.method == "GET" + if machine_health and health_service_ids: + claims = verifier.verify(token, health_service_ids=health_service_ids) + if not claims.get("email"): + session.clear() + g.cloudflare_access_health_service = claims["common_name"] + return None + else: + claims = verifier.verify(token) fingerprint = hashlib.sha256(token.encode()).hexdigest() email = claims["email"] if not ( diff --git a/tests/backend/test_cloudflare_access.py b/tests/backend/test_cloudflare_access.py index 0a908ed49..76f7699af 100644 --- a/tests/backend/test_cloudflare_access.py +++ b/tests/backend/test_cloudflare_access.py @@ -39,6 +39,7 @@ def app(monkeypatch, signing_key): monkeypatch.setenv("VON_CLOUDFLARE_ACCESS_HOSTNAME", "von.example.test") monkeypatch.setenv("VON_CLOUDFLARE_ACCESS_TEAM_DOMAIN", TEAM) monkeypatch.setenv("VON_CLOUDFLARE_ACCESS_AUDIENCE", "aud1") + monkeypatch.setenv("VON_CLOUDFLARE_HEALTH_SERVICE_IDS", "health-client.access") app = Flask(__name__) app.secret_key = "test-only-secret-not-for-deployment" app.testing = True @@ -54,9 +55,38 @@ def private(): return jsonify(error="no actor"), 401 return jsonify(actor=session.get("user_concept_id"), org=session.get("organisation_concept_id"), provider=session.get("auth_provider")) + @app.get("/health") + def health_check(): + return jsonify(status="healthy", actor=session.get("user_concept_id")) + return app +@pytest.mark.parametrize("audience", [["aud1"], "aud1"]) +def test_signed_configured_health_service_never_becomes_a_user(app, signing_key, audience): + client=app.test_client() + headers={"Cf-Access-Jwt-Assertion":token(signing_key[0],sub="",email=None,common_name="health-client.access",aud=audience)} + health=client.get("/health",base_url=HOST,headers=headers) + assert health.status_code==200 + assert health.json["actor"] is None + assert client.get("/private",base_url=HOST,headers=headers).status_code==403 + assert client.get("/von/api/auth/status",base_url=HOST,headers=headers).status_code==403 + assert client.post("/health",base_url=HOST,headers=headers).status_code==403 + with client.session_transaction(base_url=HOST) as session: + assert not session.get("user_concept_id") + + +@pytest.mark.parametrize("overrides",[ + {"common_name":"another-client.access"}, {"aud":["another-app"]}, + {"iss":"https://another.cloudflareaccess.com"}, {"exp":int(time.time())-1}, +]) +def test_health_service_still_requires_exact_signed_scope(app,signing_key,overrides): + claims={"sub":"","email":None,"common_name":"health-client.access",**overrides} + response=app.test_client().get("/health",base_url=HOST, + headers={"Cf-Access-Jwt-Assertion":token(signing_key[0],**claims)}) + assert response.status_code==403 + + def test_normal_issuance_binding_and_scope_retention(app, signing_key): client = app.test_client() headers = {"Cf-Access-Jwt-Assertion": token(signing_key[0])} diff --git a/tests/backend/test_codex_von_worker.py b/tests/backend/test_codex_von_worker.py index 7ad06e26f..b75c63ca1 100644 --- a/tests/backend/test_codex_von_worker.py +++ b/tests/backend/test_codex_von_worker.py @@ -62,6 +62,66 @@ def test_empty_poll_has_no_model_or_message(config, monkeypatch): worker.tick(config, SimpleNamespace(pending=list), 0) +@pytest.mark.parametrize( + "request_deployment,changed,interrupted,expected", + [ + (False, False, False, None), + (True, True, False, "not_authorised"), + (True, False, False, "deployed"), + (True, True, True, "deployed"), + ], +) +def test_deployment_request_and_live_task_authority( + config, tmp_path, monkeypatch, request_deployment, changed, interrupted, expected +): + current = task(config, status="in_progress") + inputs = {"description": "Implement and deploy"} + api = SimpleNamespace( + task=lambda _: current, inputs=lambda _: inputs, native_writer=lambda _: True + ) + result = { + "status": "completed", + "summary": "Merged", + "evidence": "Tests passed", + "question": "", + "deploy_commit": "a" * 40 if request_deployment else "", + } + state = { + "task_id": "#V#task", + "result": result, + "run_dir": str(tmp_path), + "input_hash": worker.fingerprint(inputs), + } + config["deployment_command"] = "/operator/deploy" + if changed: + inputs["description"] = "Do not deploy" + if interrupted: + worker.write_json(tmp_path / "deployment.json", {"status": "started"}) + calls = [] + + def run(args, **kwargs): + calls.append(args) + worker.write_json( + tmp_path / "deployment.json", + { + "requested_commit": "a" * 40, + "status": "deployed", + "verification": {"commit": "a" * 40}, + }, + ) + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(worker.subprocess, "run", run) + worker.apply_deployment(config, api, state, tmp_path / "state.json", 0) + assert state.get("deployment_final") == expected + assert len(calls) == int(expected == "deployed") + if expected == "deployed": + assert ("--recover-only" in calls[0]) is interrupted + assert (result["status"] == "blocked") is interrupted + worker.apply_deployment(config, api, state, tmp_path / "state.json", 0) + assert len(calls) == 1 + + @pytest.mark.parametrize("writer", ["jira", "von", None]) def test_project_tracking_authority_is_read_live(config, monkeypatch, writer): from src.backend.services import task_project_service diff --git a/tests/test_codex_von_deploy.py b/tests/test_codex_von_deploy.py new file mode 100644 index 000000000..6ee35e203 --- /dev/null +++ b/tests/test_codex_von_deploy.py @@ -0,0 +1,240 @@ +import socket +import subprocess +import sys +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest + +pytest.importorskip("fcntl") +from scripts import codex_von_deploy as deployer + + +def test_public_credentials_are_not_forwarded_through_redirects(): + paths = [] + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + paths.append(self.path) + self.send_response(302) + self.send_header( + "Location", f"http://127.0.0.1:{self.server.server_port}/login-provider" + ) + self.end_headers() + + server = HTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + with pytest.raises(deployer.runtime.HealthAccessError): + deployer.runtime._read_health( + f"http://127.0.0.1:{server.server_port}/health", + {"CF-Access-Client-Secret": "fixture-only"}, + ) + assert paths == ["/health"] + finally: + server.shutdown() + server.server_close() + thread.join() + + +@pytest.fixture +def deployment(tmp_path, monkeypatch): + config = { + "primary_root": str(tmp_path / "primary"), + "runtime_root": str(tmp_path / "runtime"), + "state_root": str(tmp_path), + "health_url": "http://localhost:5000/health", + "public_health_url": "https://example.test/health", + } + previous = "a" * 40 + target = "b" * 40 + events = [] + monkeypatch.setattr(deployer.runtime, "_validate_runtime", lambda *a: None) + monkeypatch.setattr(deployer.runtime, "_run", lambda *a, **kw: None) + monkeypatch.setattr(deployer.runtime, "_git", lambda *a: target) + monkeypatch.setattr( + deployer.runtime, + "_read_health", + lambda *a: {"version_details": {"git_commit": previous}}, + ) + monkeypatch.setattr( + deployer.runtime, + "_health_error", + lambda h, c: ( + None if h["version_details"]["git_commit"] == c else "wrong revision" + ), + ) + monkeypatch.setattr(deployer, "verify", lambda c, sha: {"commit": sha}) + monkeypatch.setattr( + deployer, + "recover", + lambda c, sha: events.append(("recover", sha)) or {"commit": sha}, + ) + return config, previous, target, events, tmp_path / "receipt.json" + + +def test_startup_failure_recovers_once_and_keeps_failed_outcome( + deployment, monkeypatch +): + config, previous, target, events, path = deployment + + def fail(**kwargs): + events.append(("deploy", kwargs["expected_commit"])) + raise deployer.runtime.DeploymentError("fixture startup failure") + + monkeypatch.setattr(deployer.runtime, "deploy", fail) + receipt = deployer.execute(config, target, path) + assert receipt["status"] == "rolled_back" + assert receipt["verification"]["commit"] == previous + assert events == [("deploy", target), ("recover", previous)] + assert deployer.execute(config, target, path) == receipt + assert len(events) == 2 + + +def test_interrupted_completed_deployment_is_read_back_without_restart( + deployment, monkeypatch +): + config, previous, target, events, path = deployment + deployer.save( + path, + {"status": "started", "requested_commit": target, "previous_commit": previous}, + ) + monkeypatch.setattr( + deployer.runtime, + "_read_health", + lambda *a: {"version_details": {"git_commit": target}}, + ) + monkeypatch.setattr( + deployer.runtime, "deploy", lambda **kw: pytest.fail("restarted") + ) + assert deployer.execute(config, target, path)["status"] == "deployed" + assert not events + + +def test_recovery_failure_is_preserved_without_claiming_deployment( + deployment, monkeypatch +): + config, previous, target, _events, path = deployment + deployer.save( + path, + {"status": "started", "requested_commit": target, "previous_commit": previous}, + ) + + def fail(*args): + raise deployer.runtime.DeploymentError("fixture recovery failure") + + monkeypatch.setattr(deployer, "recover", fail) + assert deployer.execute(config, target, path)["status"] == "recovery_failed" + + +def test_stale_revision_has_no_deployment_effect(deployment, monkeypatch): + config, previous, _target, events, path = deployment + monkeypatch.setattr( + deployer.runtime, "deploy", lambda **kw: pytest.fail("stale deployment") + ) + with pytest.raises(deployer.runtime.DeploymentError, match="no longer origin/main"): + deployer.execute(config, previous, path) + assert not path.exists() + assert not events + + +def test_recovery_only_cannot_start_a_new_deployment(deployment, monkeypatch): + config, _previous, target, _events, path = deployment + monkeypatch.setattr( + deployer.runtime, "deploy", lambda **kw: pytest.fail("new deployment") + ) + with pytest.raises(ValueError, match="existing deployment receipt"): + deployer.execute(config, target, path, recover_only=True) + assert not path.exists() + + +def test_failed_real_process_restores_previous_git_revision_and_http_service( + tmp_path, monkeypatch +): + from tests.test_deploy_local_main import _git, _setup_repositories + + _, seed, primary, checkout = _setup_repositories(tmp_path) + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + (seed / ".gitignore").write_text("*.pid\n*.log\n__pycache__/\n") + (seed / "pdm.lock").write_text("# unchanged fixture environment\n") + (seed / "run.sh").write_text(f'#!/bin/sh\nexec {sys.executable} launcher.py "$@"\n') + (seed / "launcher.py").write_text( + """import os, pathlib, signal, subprocess, sys, time +p=pathlib.Path('fixture.pid') +action=sys.argv[1] +port=sys.argv[sys.argv.index('-Port')+1] +if action in ('stop','restart') and p.exists(): + try: os.kill(int(p.read_text()),signal.SIGTERM) + except ProcessLookupError: pass + p.unlink() + time.sleep(0.15) +if action in ('start','restart'): + with open('fixture.log','ab') as log: + child=subprocess.Popen([sys.executable,'server.py',port],stdout=log,stderr=log,start_new_session=True) + p.write_text(str(child.pid)) +""" + ) + (seed / "server.py").write_text( + """import http.server, json, os, pathlib, subprocess, sys +if pathlib.Path('fail-startup').exists():sys.exit(7) +commit=subprocess.check_output(['git','rev-parse','HEAD'],text=True).strip() +payload={'status':'healthy','pid':os.getpid(),'version_details':{'git_commit':commit,'git_dirty':False},'runtime_authority':{'durable_workflows':{'database_connected':True,'worker_running':True,'scheduler_running':True},'startup_seed_materialisations':{'ready':True,'state':'ready'}}} +class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self,*args):pass + def do_GET(self): + data=json.dumps(payload).encode();self.send_response(200);self.end_headers();self.wfile.write(data) +class Server(http.server.HTTPServer):allow_reuse_address=True +Server(('127.0.0.1',int(sys.argv[1])),Handler).serve_forever() +""" + ) + _git(seed, "add", ".") + _git(seed, "commit", "-m", "healthy fixture") + _git(seed, "push", "origin", "main") + previous = deployer.runtime._prepare_primary( + primary, remote="origin", branch="main" + ) + deployer.runtime._prepare_runtime(primary, checkout, previous) + subprocess.run( + ["bash", "run.sh", "start", "-Port", str(port)], cwd=checkout, check=True + ) + url = f"http://127.0.0.1:{port}/health" + deployer.runtime._wait_for_verified_health(url, previous, 3) + (seed / "fail-startup").write_text("intentional fixture failure\n") + _git(seed, "add", "fail-startup") + _git(seed, "commit", "-m", "failing fixture") + _git(seed, "push", "origin", "main") + target = _git(seed, "rev-parse", "HEAD") + config = { + "primary_root": str(primary), + "runtime_root": str(checkout), + "state_root": str(tmp_path), + "health_url": url, + "public_health_url": url, + "health_timeout_seconds": 3, + } + monkeypatch.setattr( + deployer.runtime, "_stop_verified_predecessor", lambda **kw: None + ) + monkeypatch.setattr( + deployer.runtime, "_reconcile_startup_seed_materialisations", lambda *a: None + ) + monkeypatch.setattr(deployer.runtime, "_verify_workers", lambda *a: (1, 2)) + try: + receipt = deployer.execute(config, target, tmp_path / "real-recovery.json") + assert receipt["status"] == "rolled_back" + assert _git(checkout, "rev-parse", "HEAD") == previous + assert _git(primary, "rev-parse", "HEAD") == target + assert ( + deployer.runtime._read_health(url)["version_details"]["git_commit"] + == previous + ) + finally: + subprocess.run( + ["bash", "run.sh", "stop", "-Port", str(port)], cwd=checkout, check=True + ) diff --git a/tests/test_deploy_local_main.py b/tests/test_deploy_local_main.py index 24aaa078e..3a5243343 100644 --- a/tests/test_deploy_local_main.py +++ b/tests/test_deploy_local_main.py @@ -92,6 +92,19 @@ def test_prepare_refuses_primary_that_is_not_main(tmp_path: Path) -> None: _prepare_primary(primary.resolve(), remote="origin", branch="main") +def test_stale_expected_revision_does_not_advance_primary(tmp_path: Path) -> None: + _, seed, primary, _ = _setup_repositories(tmp_path) + before = _git(primary, "rev-parse", "HEAD") + (seed / "tracked.txt").write_text("newer\n") + _git(seed, "commit", "-am", "newer") + _git(seed, "push", "origin", "main") + with pytest.raises(DeploymentError, match="no longer origin/main"): + _prepare_primary( + primary, remote="origin", branch="main", expected_commit=before + ) + assert _git(primary, "rev-parse", "HEAD") == before + + def test_prepare_allows_registered_nested_worktree(tmp_path: Path) -> None: _, _, primary, _ = _setup_repositories(tmp_path) nested_worktree = primary / ".worktrees" / "feature" @@ -165,9 +178,7 @@ def test_health_verification_requires_exact_clean_durable_build() -> None: assert "running commit=" in (_health_error(wrong_commit, commit) or "") no_scheduler = json.loads(json.dumps(payload)) - no_scheduler["runtime_authority"]["durable_workflows"][ - "scheduler_running" - ] = False + no_scheduler["runtime_authority"]["durable_workflows"]["scheduler_running"] = False assert _health_error(no_scheduler, commit) == ( "durable workflow readiness false: scheduler_running" ) @@ -244,14 +255,15 @@ def test_deploy_reconciles_target_runtime_before_server_start( ) monkeypatch.setattr( "scripts.deploy_local_main._reconcile_startup_seed_materialisations", - lambda *_args, **_kwargs: events.append("reconcile") - or {"success": True}, + lambda *_args, **_kwargs: events.append("reconcile") or {"success": True}, ) def _record_run(args, **_kwargs): if "start" in args: + assert args[args.index("-Port") + 1] == "5000" events.append("start") elif "stop" in args: + assert args[args.index("-Port") + 1] == "5000" events.append("stop_runtime") return subprocess.CompletedProcess(args, 0, "", "") @@ -265,7 +277,11 @@ def _record_run(args, **_kwargs): lambda *_args, **_kwargs: (2345, 3456), ) - result = deploy(primary_root=primary, runtime_root=runtime) + result = deploy( + primary_root=primary, + runtime_root=runtime, + health_url="http://127.0.0.1:5000/health", + ) assert result.commit == commit assert events.index("prepare_runtime") < events.index("reconcile") @@ -297,11 +313,14 @@ def cwd(self) -> str: monkeypatch.setattr("scripts.deploy_local_main.psutil.Process", _Process) - assert _matching_von_process_root( - pid=4242, - expected_port=5001, - allowed_roots=(primary, runtime), - ) == primary + assert ( + _matching_von_process_root( + pid=4242, + expected_port=5001, + allowed_roots=(primary, runtime), + ) + == primary + ) with pytest.raises(DeploymentError, match="does not select port 5010"): _matching_von_process_root( @@ -376,12 +395,15 @@ def _missing_process(pid: int): _missing_process, ) - assert _stop_verified_predecessor( - primary_root=primary, - runtime_root=runtime, - current_launcher=launcher, - health_url="http://127.0.0.1:5001/health", - ) == 4444 + assert ( + _stop_verified_predecessor( + primary_root=primary, + runtime_root=runtime, + current_launcher=launcher, + health_url="http://127.0.0.1:5001/health", + ) + == 4444 + ) assert calls[0]["args"] == [ "bash", launcher,