diff --git a/scripts/deploy_codex_audit_service.sh b/scripts/deploy_codex_audit_service.sh index 0e9fb05c..1e46091c 100644 --- a/scripts/deploy_codex_audit_service.sh +++ b/scripts/deploy_codex_audit_service.sh @@ -14,6 +14,7 @@ ALLOWED_REPOSITORY_VISIBILITIES="${CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORY_VISIBI ALLOWED_SOURCE_REPOSITORIES="${CODEX_AUDIT_SERVICE_ALLOWED_SOURCE_REPOSITORIES:-QuantStrategyLab/AIAuditBridge,QuantStrategyLab/CryptoLivePoolPipelines,QuantStrategyLab/HkEquitySnapshotPipelines,QuantStrategyLab/UsEquitySnapshotPipelines,QuantStrategyLab/ResearchSignalContextPipelines}" JOB_DIR="${CODEX_AUDIT_SERVICE_JOB_DIR:-/var/lib/codex-audit-bridge/jobs}" ADMIN_ENV_FILE="${CODEX_AUDIT_SERVICE_ADMIN_ENV_FILE:-/etc/codex-audit-bridge/admin.env}" +EXECUTION_POLICY_FILE="${CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH:-/etc/codex-audit-bridge-policy/execution_policy.json}" AUDIT_MODEL="${CODEX_AUDIT_SERVICE_MODEL:-}" AUDIT_REASONING_EFFORT="${CODEX_AUDIT_SERVICE_REASONING_EFFORT:-}" CODEX_ACCOUNT_USAGE="${CODEX_AUDIT_SERVICE_CODEX_ACCOUNT_USAGE:-1}" @@ -60,7 +61,7 @@ systemctl_environment_brief() { | sed 's/^Environment=//' \ | tr ' ' '\n' \ | sed -E "s/^[\"']//; s/[\"']$//" \ - | grep -E '^CODEX_AUDIT_SERVICE_(ALLOWED_|AUDIENCE=|HOST=|PORT=|JOB_DIR=|QUOTA_STORE=|CODEX_ACCOUNT_USAGE=|OPENAI_USAGE_WINDOW_DAYS=|ANTHROPIC_USAGE_WINDOW_DAYS=|SANDBOX=|MODEL=|REASONING_EFFORT=)' \ + | grep -E '^CODEX_AUDIT_SERVICE_(ALLOWED_|AUDIENCE=|HOST=|PORT=|JOB_DIR=|QUOTA_STORE=|EXECUTION_POLICY_PATH=|CODEX_ACCOUNT_USAGE=|OPENAI_USAGE_WINDOW_DAYS=|ANTHROPIC_USAGE_WINDOW_DAYS=|SANDBOX=|MODEL=|REASONING_EFFORT=)' \ | mask_infra || true fi } @@ -214,6 +215,123 @@ write_admin_env_file_if_needed() { trap - RETURN } +write_default_execution_policy_if_missing() { + local policy_path="${EXECUTION_POLICY_FILE}" + local policy_dir + policy_dir="$(dirname "$policy_path")" + if [ -L "$policy_path" ]; then + echo "refusing to write execution policy through symlink: $policy_path" >&2 + exit 1 + fi + if [ -e "$policy_path" ]; then + return + fi + sudo python3 - "$policy_path" <<'PY' +import os +import stat +import sys + +path = sys.argv[1] +if not os.path.isabs(path): + print(f"refusing to write execution policy to relative path: {path}", file=sys.stderr) + raise SystemExit(1) + +policy_dir, policy_name = os.path.split(path) +if not policy_dir or not policy_name: + print(f"invalid execution policy path: {path}", file=sys.stderr) + raise SystemExit(1) + +flags_dir = os.O_RDONLY | os.O_DIRECTORY +if hasattr(os, "O_NOFOLLOW"): + flags_dir |= os.O_NOFOLLOW + + +def fail(message: str) -> None: + print(message, file=sys.stderr) + raise SystemExit(1) + + +def ensure_trusted_dir(fd: int, label: str, *, created: bool) -> None: + info = os.fstat(fd) + if not stat.S_ISDIR(info.st_mode): + fail(f"execution policy parent path is not a directory: {label}") + if created: + os.fchown(fd, 0, 0) + os.fchmod(fd, 0o755) + info = os.fstat(fd) + if (info.st_uid, info.st_gid) != (0, 0): + fail(f"execution policy parent directory owner is invalid: {label}") + if info.st_mode & (stat.S_IWGRP | stat.S_IWOTH): + fail(f"execution policy parent directory permissions are too broad: {label}") + + +def open_admin_policy_dir(directory: str) -> int: + fd = os.open(os.sep, flags_dir) + ensure_trusted_dir(fd, os.sep, created=False) + for component in [part for part in directory.split(os.sep) if part]: + if component in {".", ".."}: + fail(f"invalid execution policy directory component: {component}") + created = False + try: + next_fd = os.open(component, flags_dir, dir_fd=fd) + except FileNotFoundError: + os.mkdir(component, 0o755, dir_fd=fd) + created = True + next_fd = os.open(component, flags_dir, dir_fd=fd) + except OSError as exc: + fail(f"refusing to write execution policy under unsafe directory component {component}: {exc}") + try: + ensure_trusted_dir(next_fd, component, created=created) + finally: + os.close(fd) + fd = next_fd + return fd + + +policy_dir_fd = open_admin_policy_dir(policy_dir) +flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL +if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW +content = """{ + "default": { + "max_autonomy": "auto_pr", + "max_consecutive_failures": 3, + "low_cost_model": "gpt-5.4-mini", + "low_cost_provider": "openai" + }, + "repositories": {} +} +""" +try: + fd = os.open(policy_name, flags, 0o600, dir_fd=policy_dir_fd) +except FileExistsError: + if os.path.islink(path): + print(f"refusing to write execution policy through symlink: {path}", file=sys.stderr) + raise SystemExit(1) + raise SystemExit(0) +finally: + os.close(policy_dir_fd) +try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + fd = -1 + handle.write(content) + handle.flush() + os.fchown(handle.fileno(), 0, 0) + os.fchmod(handle.fileno(), 0o644) +except Exception: + if fd >= 0: + try: + os.close(fd) + except OSError: + pass + try: + os.unlink(path) + except FileNotFoundError: + pass + raise +PY +} + write_audit_service_unit() { local runner_user runner_home runner_user="$(id -un)" @@ -252,6 +370,7 @@ Environment=CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORY_VISIBILITIES=${ALLOWED_REPOSI Environment=CODEX_AUDIT_SERVICE_ALLOWED_SOURCE_REPOSITORIES=${ALLOWED_SOURCE_REPOSITORIES} Environment=CODEX_AUDIT_SERVICE_JOB_DIR=${JOB_DIR} Environment=CODEX_AUDIT_SERVICE_QUOTA_STORE=${JOB_DIR}/quota.json +Environment=CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH=${EXECUTION_POLICY_FILE} Environment=CODEX_AUDIT_SERVICE_CODEX_ACCOUNT_USAGE=${CODEX_ACCOUNT_USAGE} Environment=CODEX_AUDIT_SERVICE_OPENAI_USAGE_WINDOW_DAYS=${OPENAI_USAGE_WINDOW_DAYS} Environment=CODEX_AUDIT_SERVICE_ANTHROPIC_USAGE_WINDOW_DAYS=${ANTHROPIC_USAGE_WINDOW_DAYS} @@ -281,6 +400,7 @@ Environment="CODEX_AUDIT_SERVICE_ALLOWED_WORKFLOW_REFS=${ALLOWED_WORKFLOW_REFS}" Environment="CODEX_AUDIT_SERVICE_ALLOWED_REFS=${ALLOWED_REFS}" Environment="CODEX_AUDIT_SERVICE_ALLOWED_REPOSITORY_VISIBILITIES=${ALLOWED_REPOSITORY_VISIBILITIES}" Environment="CODEX_AUDIT_SERVICE_ALLOWED_SOURCE_REPOSITORIES=${ALLOWED_SOURCE_REPOSITORIES}" +Environment="CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH=${EXECUTION_POLICY_FILE}" EOF_DROPIN } @@ -508,6 +628,7 @@ deploy() { install_file "scripts/codex_audit_service.py" "${DEPLOY_DIR}/scripts/codex_audit_service.py" "0755" install_service_package sudo install -d -m 0700 -o "$runner_user" -g "$runner_user" "$JOB_DIR" + write_default_execution_policy_if_missing write_admin_env_file_if_needed write_audit_service_unit write_managed_audit_service_dropin diff --git a/service/ai_gateway_service.py b/service/ai_gateway_service.py index c506727e..6365162e 100644 --- a/service/ai_gateway_service.py +++ b/service/ai_gateway_service.py @@ -35,6 +35,7 @@ from service.auth import authenticate from service.contracts import ( + MODE_REVIEW_AND_FIX, MODE_REVIEW_ONLY, TASK_ANALYZE, TASK_EXECUTE, @@ -72,6 +73,7 @@ get_automation_run_ledger, suggest_control_action, ) +from service.automation_decision import EXECUTION_DEFER, EXECUTION_HUMAN_REVIEW, EXECUTION_REVIEW_ONLY, decide_automation_execution, load_execution_policy from service.strategy_automation_registry import ( apply_strategy_registry_guard, summarize_strategy_registry_context, @@ -507,7 +509,13 @@ def _public_job_payload(job: dict[str, Any]) -> dict[str, object]: return payload -def _automation_control_snapshot(repo: str) -> dict[str, Any]: +def _automation_control_snapshot( + repo: str, + *, + task_name: str = "", + requested_mode: str = MODE_REVIEW_AND_FIX, + pending_run: dict[str, Any] | None = None, +) -> dict[str, Any]: try: org_health = read_org_health() except Exception: @@ -516,7 +524,98 @@ def _automation_control_snapshot(repo: str) -> dict[str, Any]: quota_status = get_quota_manager().runtime_status(repo or "unknown") except Exception: quota_status = {"status": "unavailable"} - return suggest_control_action(get_health_monitor().status, quota_status, org_health) + control = suggest_control_action(get_health_monitor().status, quota_status, org_health) + try: + ledger_snapshot = get_automation_run_ledger().snapshot(limit=None) + recent_runs = ledger_snapshot["runs"] + ledger_summary = ledger_snapshot.get("summary") if isinstance(ledger_snapshot.get("summary"), dict) else {} + retention = ledger_summary.get("retention") if isinstance(ledger_summary.get("retention"), dict) else {} + ledger_unavailable = False + except Exception: + recent_runs = [] + retention = {} + ledger_unavailable = True + if pending_run is not None: + pending_run_id = str(pending_run.get("run_id") or "") + if pending_run_id: + recent_runs = [run for run in recent_runs if str(run.get("run_id") or "") != pending_run_id] + recent_runs = [pending_run, *recent_runs] + evicted_by_repo = ( + retention.get("evicted_runs_by_repo") if isinstance(retention.get("evicted_runs_by_repo"), dict) else {} + ) + try: + repo_evictions = int(evicted_by_repo.get(str(repo or "unknown").strip().lower(), 0) or 0) + except (TypeError, ValueError): + repo_evictions = 0 + storage_unavailable = bool(retention.get("storage_unavailable")) + normalized_repo = str(repo or "unknown").strip().lower() + repo_history_has_terminal_boundary = any( + str(run.get("task_state") or "").strip().lower() in {"merged", "completed", "succeeded"} + and _automation_run_owner_repository(run).strip().lower() == normalized_repo + and str((run.get("metadata") if isinstance(run.get("metadata"), dict) else {}).get("origin") or "") in {"service_job", "external_workflow"} + for run in recent_runs + if isinstance(run, dict) + ) + failure_history_complete = ( + not ledger_unavailable + and not storage_unavailable + and (not bool(retention.get("history_completeness_unknown")) or repo_history_has_terminal_boundary) + and (repo_evictions <= 0 or repo_history_has_terminal_boundary) + ) + execution = decide_automation_execution( + repo=repo or "unknown", + task_name=task_name, + requested_mode=requested_mode, + control_action=str(control.get("action") or CONTROL_REVIEW_ONLY), + service_health=control.get("service_health"), + quota_status=quota_status, + org_health_status=control.get("org_health_status"), + recent_runs=recent_runs, + failure_history_complete=failure_history_complete, + policy=load_execution_policy(), + ) + if ledger_unavailable: + execution["action"] = EXECUTION_HUMAN_REVIEW + execution["effective_mode"] = MODE_REVIEW_ONLY + execution["human_review_required"] = True + execution["auto_fix_allowed"] = False + execution["auto_merge_allowed"] = False + execution["defer"] = False + reasons = execution.get("reasons") if isinstance(execution.get("reasons"), list) else [] + reasons.append("automation ledger unavailable; forcing human review") + execution["reasons"] = reasons + original_action = str(control.get("action") or CONTROL_REVIEW_ONLY) + strict_action = original_action + if execution.get("action") == EXECUTION_HUMAN_REVIEW: + strict_action = CONTROL_ESCALATE + elif execution.get("action") == EXECUTION_REVIEW_ONLY: + strict_action = CONTROL_REVIEW_ONLY + elif execution.get("action") == EXECUTION_DEFER: + strict_action = CONTROL_PAUSE_AUTO_FIX + elif execution.get("action") == "run" and execution.get("auto_fix_allowed"): + strict_action = CONTROL_CONTINUE + elif execution.get("effective_mode") == MODE_REVIEW_ONLY and strict_action == CONTROL_CONTINUE: + strict_action = CONTROL_REVIEW_ONLY + action_rank = {CONTROL_CONTINUE: 0, CONTROL_PAUSE_AUTO_FIX: 1, CONTROL_REVIEW_ONLY: 2, CONTROL_ESCALATE: 3} + if strict_action != CONTROL_CONTINUE and action_rank.get(strict_action, 2) < action_rank.get(original_action, 2): + strict_action = original_action + control["runtime_action"] = original_action + control["effective_action"] = strict_action + control["action"] = strict_action + control["auto_fix_allowed"] = bool(execution.get("auto_fix_allowed")) and strict_action == CONTROL_CONTINUE + control["auto_merge_allowed"] = bool(execution.get("auto_merge_allowed")) and strict_action == CONTROL_CONTINUE + control["requires_human_review"] = strict_action != CONTROL_CONTINUE or bool(execution.get("human_review_required")) + control["execution"] = execution + return control + + +def _normalize_control_mode_param(value: str) -> str: + mode = str(value or "").strip().lower() + if not mode: + return MODE_REVIEW_AND_FIX + if mode in {MODE_REVIEW_ONLY, MODE_REVIEW_AND_FIX, "manual", "auto_pr", "auto_merge"}: + return mode + return "" def _highest_changed_path_risk(changed_paths: list[str], policy: dict[str, Any]) -> str: @@ -535,12 +634,13 @@ def _automation_triage_snapshot( repo: str, *, task: str = "", + requested_mode: str = MODE_REVIEW_AND_FIX, failure_category: str = "", error: str = "", changed_paths: list[str] | None = None, run_id: str = "", ) -> dict[str, Any]: - control = _automation_control_snapshot(repo) + control = _automation_control_snapshot(repo, task_name=task, requested_mode=requested_mode) policy = load_autonomy_policy() normalized_paths = [ normalized @@ -554,7 +654,9 @@ def _automation_triage_snapshot( if not category and error: category = service_failure_category(error) - control_action = str(control.get("action") or CONTROL_REVIEW_ONLY) + control_action = str(control.get("effective_action") or control.get("action") or CONTROL_REVIEW_ONLY) + execution = control.get("execution") if isinstance(control.get("execution"), dict) else {} + execution_auto_fix_allowed = bool(control.get("auto_fix_allowed")) and bool(execution.get("auto_fix_allowed")) retry_allowed = False deploy_allowed = False auto_fix_allowed = False @@ -586,7 +688,7 @@ def _automation_triage_snapshot( incident_class = "degraded" recommended_action = "open_issue" next_step = "pause_auto_fix" - elif control_action == CONTROL_CONTINUE: + elif control_action == CONTROL_CONTINUE and execution_auto_fix_allowed: incident_class = "investigate" recommended_action = "open_fix_pr" if path_risk in {RISK_LOW, RISK_MEDIUM} else "open_issue" next_step = "open_fix_pr" if path_risk in {RISK_LOW, RISK_MEDIUM} else "open_issue" @@ -595,7 +697,7 @@ def _automation_triage_snapshot( recommended_action = "open_issue" next_step = "open_issue" - if control_action == CONTROL_CONTINUE and category == "" and path_risk in {RISK_LOW, RISK_MEDIUM}: + if execution_auto_fix_allowed and category == "" and path_risk in {RISK_LOW, RISK_MEDIUM}: auto_fix_allowed = True deploy_allowed = True if path_risk in {RISK_HIGH, RISK_CRITICAL}: @@ -648,24 +750,37 @@ def _automation_triage_snapshot( def _record_job_automation_run(job: dict[str, Any]) -> None: try: repo = str(job.get("source_repository") or job.get("repository") or "unknown") - control = _automation_control_snapshot(repo) + task_name = str(job.get("task") or "") + task_state = job_task_state(job) + metadata = { + "origin": "service_job", + "repository": repo, + "source_repository": str(job.get("source_repository") or ""), + "caller_repository": str(job.get("repository") or ""), + "source_ref": str(job.get("source_ref") or ""), + "mode": str(job.get("mode") or ""), + "failure_category": str(job.get("failure_category") or ""), + } + control = _automation_control_snapshot( + repo, + task_name=task_name, + requested_mode=str(job.get("mode") or MODE_REVIEW_AND_FIX), + pending_run={ + "run_id": str(job.get("job_id") or ""), + "task_name": task_name, + "task_state": task_state, + "metadata": metadata, + }, + ) get_automation_run_ledger().record( str(job.get("job_id") or ""), - job_task_state(job), - task_name=str(job.get("task") or ""), - suggested_action=str(control.get("action") or ""), + task_state, + task_name=task_name, + suggested_action=str(control.get("effective_action") or control.get("action") or ""), service_health=str(control.get("service_health") or ""), quota_status=str(control.get("quota_status") or ""), org_health_status=str(control.get("org_health_status") or ""), - metadata={ - "origin": "service_job", - "repository": repo, - "source_repository": str(job.get("source_repository") or ""), - "caller_repository": str(job.get("repository") or ""), - "source_ref": str(job.get("source_ref") or ""), - "mode": str(job.get("mode") or ""), - "failure_category": str(job.get("failure_category") or ""), - }, + metadata=metadata, owner_repository=repo, ) except Exception as exc: @@ -781,6 +896,9 @@ def _automation_snapshot_for_claims( ) terminal_runs = sum(1 for run in visible_runs if str(run.get("task_state", "")).strip().lower() in TERMINAL_STATES) summary = dict(snapshot.get("summary") or {}) + retention = summary.get("retention") + if isinstance(retention, dict): + summary["retention"] = {key: value for key, value in retention.items() if key != "evicted_runs_by_repo"} summary.update( { "total_runs": len(visible_runs), @@ -1410,7 +1528,7 @@ def _handle_automation_control(self) -> None: _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) return parsed = urlparse(self.path) - params = parse_qs(parsed.query) + params = parse_qs(parsed.query, keep_blank_values=True) repo = str(params.get("repo", [""])[0] or "") if not _automation_operator_claims(claims): claims_repo = str(claims.get("repository") or "") @@ -1426,7 +1544,12 @@ def _handle_automation_control(self) -> None: else: repo = claims_repo repo = repo or "unknown" - _json_response(self, HTTPStatus.OK, {"status": "ok", "control": _automation_control_snapshot(repo)}) + raw_mode = params["mode"][0] if "mode" in params else MODE_REVIEW_AND_FIX + mode = _normalize_control_mode_param(str(raw_mode if raw_mode is not None else "")) + if not mode: + _json_response(self, HTTPStatus.BAD_REQUEST, {"status": "error", "error": "invalid mode"}) + return + _json_response(self, HTTPStatus.OK, {"status": "ok", "control": _automation_control_snapshot(repo, requested_mode=mode)}) def _handle_automation_triage(self, claims: dict[str, Any], payload: dict[str, Any]) -> None: from urllib.parse import parse_qs, urlparse @@ -1437,9 +1560,14 @@ def _handle_automation_triage(self, claims: dict[str, Any], payload: dict[str, A if not _automation_operator_claims(claims): _validate_source_repo_org(claims, source_repo) parsed = urlparse(self.path) - params = parse_qs(parsed.query) + params = parse_qs(parsed.query, keep_blank_values=True) repo = str(payload.get("source_repository") or params.get("repo", [""])[0] or claims.get("repository") or "unknown") task = str(payload.get("task") or params.get("task", [""])[0] or "") + raw_mode = payload.get("mode") if "mode" in payload else params["mode"][0] if "mode" in params else MODE_REVIEW_AND_FIX + requested_mode = _normalize_control_mode_param(str(raw_mode if raw_mode is not None else "")) + if not requested_mode: + _json_response(self, HTTPStatus.BAD_REQUEST, {"status": "error", "error": "invalid mode"}) + return failure_category = str(payload.get("failure_category") or params.get("failure_category", [""])[0] or "") error = str(payload.get("error") or params.get("error", [""])[0] or "") run_id = str(payload.get("run_id") or params.get("run_id", [""])[0] or "") @@ -1449,6 +1577,7 @@ def _handle_automation_triage(self, claims: dict[str, Any], payload: dict[str, A triage = _automation_triage_snapshot( repo, task=task, + requested_mode=requested_mode, failure_category=failure_category, error=error, changed_paths=[str(item) for item in changed_paths_raw if isinstance(item, str)], @@ -1504,8 +1633,12 @@ def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[st _validate_source_repo_org(claims, source_repo) _assert_source_repository_owner_or_operator(claims, source_repo) repo = source_repo or str(claims.get("repository") or "unknown") - control = _automation_control_snapshot(repo) metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {} + metadata = { + key: value + for key, value in metadata.items() + if str(key).strip().lower() not in {"requested_mode", "mode"} + } ledger = get_automation_run_ledger() run_id = str(payload.get("run_id") or payload.get("job_id") or "") if not run_id.strip(): @@ -1519,23 +1652,42 @@ def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[st if existing_metadata.get("origin") == "service_job": raise PermissionError("automation run is service-owned") _assert_automation_run_access(existing, claims) + task_name = str(payload.get("task") or payload.get("task_name") or "") + existing_state = str(existing.get("task_state") or "") if isinstance(existing, dict) else "" + task_state = str(payload.get("task_state") or payload.get("state") or existing_state or "running") + existing_metadata = existing.get("metadata") if isinstance(existing, dict) and isinstance(existing.get("metadata"), dict) else {} + mode_from_payload = "mode" in payload and str(payload.get("mode") or "").strip() != "" + raw_mode = ( + payload.get("mode") + if mode_from_payload + else existing_metadata.get("requested_mode") + ) + default_mode = MODE_REVIEW_AND_FIX + requested_mode = _normalize_control_mode_param(str(raw_mode if raw_mode is not None and raw_mode != "" else ("" if mode_from_payload else default_mode))) + if mode_from_payload and not requested_mode: + raise ValueError("invalid mode") + run_metadata = { + **metadata, + "origin": "external_workflow", + "repository": repo, + "source_repository": source_repo, + "caller_repository": str(claims.get("repository") or ""), + } + run_metadata["requested_mode"] = requested_mode + pending_run = {"run_id": run_id, "task_name": task_name, "task_state": task_state, "metadata": run_metadata} + control = _automation_control_snapshot(repo, task_name=task_name, requested_mode=requested_mode, pending_run=pending_run) record = get_automation_run_ledger().record( run_id, - str(payload.get("task_state") or payload.get("state") or "running"), - task_name=str(payload.get("task") or payload.get("task_name") or ""), - suggested_action=str(control.get("action") or ""), + task_state, + task_name=task_name, + suggested_action=str(control.get("effective_action") or control.get("action") or ""), service_health=str(control.get("service_health") or ""), quota_status=control.get("quota_status") or "", org_health_status=str(control.get("org_health_status") or ""), - metadata={ - **metadata, - "origin": "external_workflow", - "repository": repo, - "source_repository": source_repo, - "caller_repository": str(claims.get("repository") or ""), - }, + metadata=run_metadata, owner_repository=repo, ) + control = _automation_control_snapshot(repo, task_name=task_name, requested_mode=requested_mode, pending_run=record) _json_response(self, HTTPStatus.OK, {"status": "ok", "run": record, "control": control}) def _handle_automation_authority(self, claims: dict[str, Any], payload: dict[str, Any]) -> None: diff --git a/service/automation_decision.py b/service/automation_decision.py new file mode 100644 index 00000000..769798ad --- /dev/null +++ b/service/automation_decision.py @@ -0,0 +1,467 @@ +"""Health-driven execution decisions for automation scheduling.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import stat +from typing import Any + +from service.automation_run_ledger import CONTROL_ESCALATE, CONTROL_PAUSE_AUTO_FIX, CONTROL_REVIEW_ONLY +from service.quota import recommend_model + +EXECUTION_RUN = "run" +EXECUTION_REVIEW_ONLY = "review_only" +EXECUTION_DEFER = "defer" +EXECUTION_HUMAN_REVIEW = "human_review" + +MODE_REVIEW_AND_FIX = "review_and_fix" +MODE_REVIEW_ONLY = "review_only" + +AUTONOMY_MANUAL = "manual" +AUTONOMY_REVIEW_ONLY = "review_only" +AUTONOMY_AUTO_PR = "auto_pr" +AUTONOMY_AUTO_MERGE = "auto_merge" +AUTONOMY_ORDER = (AUTONOMY_MANUAL, AUTONOMY_REVIEW_ONLY, AUTONOMY_AUTO_PR, AUTONOMY_AUTO_MERGE) +AUTONOMY_RANK = {level: index for index, level in enumerate(AUTONOMY_ORDER)} + +DEFAULT_MAX_CONSECUTIVE_FAILURES = 3 +DEFAULT_LOW_COST_MODEL = "gpt-5.4-mini" +DEFAULT_LOW_COST_PROVIDER = "openai" +EXECUTION_POLICY_PATH_ENV = "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH" +EXECUTION_POLICY_OWNER_ENV = "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_OWNER" +POLICY_LOAD_ERROR_KEY = "_load_error" +TRUSTED_FAILURE_ORIGINS = frozenset({"service_job", "external_workflow"}) +POLICY_ALLOWED_KEYS = frozenset({"max_autonomy", "max_consecutive_failures", "low_cost_model", "low_cost_provider", "quota_low_behavior"}) +POLICY_REQUIRED_DEFAULT_KEYS = frozenset({"max_autonomy", "max_consecutive_failures", "low_cost_model", "low_cost_provider"}) +POLICY_LOW_QUOTA_BEHAVIORS = frozenset({"low_cost_model", "defer"}) +POLICY_PROVIDERS = frozenset({"auto", "openai", "anthropic", "api", "codex"}) +QUOTA_STATUS_SEVERITY = { + "ok": 0, + "healthy": 0, + "unknown": 1, + "unavailable": 1, + "low": 2, + "constrained": 2, + "exhausted": 3, + "blocked": 3, +} + + +def _normalize_status(value: Any, default: str = "unknown") -> str: + if isinstance(value, dict): + value = value.get("status", default) + return str(value or default).strip().lower() + + +def _normalize_quota_status(value: Any, default: str = "unknown") -> str: + statuses = [_normalize_status(value, "")] + if isinstance(value, dict) and isinstance(value.get("quota"), dict): + statuses.append(_normalize_status(value["quota"], "")) + normalized = [status for status in statuses if status] + if not normalized: + return default + return max(normalized, key=lambda status: QUOTA_STATUS_SEVERITY.get(status, 1)) + + +def _normalize_mode(value: str) -> str: + mode = str(value or "").strip().lower() + if mode in {MODE_REVIEW_AND_FIX, AUTONOMY_AUTO_PR, AUTONOMY_AUTO_MERGE}: + return MODE_REVIEW_AND_FIX + return MODE_REVIEW_ONLY + + +def _normalize_requested_autonomy(value: str) -> str: + mode = str(value or "").strip().lower() + if mode == AUTONOMY_AUTO_MERGE: + return AUTONOMY_AUTO_MERGE + if mode in {MODE_REVIEW_AND_FIX, AUTONOMY_AUTO_PR}: + return AUTONOMY_AUTO_PR + if mode == AUTONOMY_MANUAL: + return AUTONOMY_MANUAL + return AUTONOMY_REVIEW_ONLY + + +def _normalize_repo_id(value: Any) -> str: + return str(value or "").strip().lower() + + +def _parse_autonomy(value: Any, default: str = AUTONOMY_AUTO_PR) -> tuple[str, str]: + level = str(value or "").strip().lower() + if not level: + return default, "" + if level in AUTONOMY_RANK: + return level, "" + return AUTONOMY_MANUAL, f"invalid max_autonomy {level!r}; forcing manual" + + +def _safe_positive_int(value: Any, default: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + return default + return parsed if parsed > 0 else default + + +def _repo_from_run(run: dict[str, Any]) -> str: + metadata = run.get("metadata") if isinstance(run.get("metadata"), dict) else {} + return str(metadata.get("source_repository") or metadata.get("repository") or "") + + +def _fail_closed_policy(reason: str) -> dict[str, Any]: + return { + POLICY_LOAD_ERROR_KEY: reason, + "default": { + "max_autonomy": AUTONOMY_MANUAL, + "max_consecutive_failures": 1, + "low_cost_model": DEFAULT_LOW_COST_MODEL, + "low_cost_provider": DEFAULT_LOW_COST_PROVIDER, + }, + } + + +def _expected_policy_owner() -> tuple[int, int]: + raw = os.environ.get(EXECUTION_POLICY_OWNER_ENV, "0:0").strip() or "0:0" + try: + uid, gid = raw.split(":", 1) + return int(uid), int(gid) + except (TypeError, ValueError): + return 0, 0 + + +def _policy_metadata_trust_error(info: os.stat_result, *, kind: str) -> str: + expected_uid, expected_gid = _expected_policy_owner() + if (info.st_uid, info.st_gid) != (expected_uid, expected_gid): + return f"execution policy {kind} owner is invalid" + if info.st_mode & (stat.S_IWGRP | stat.S_IWOTH): + return f"execution policy {kind} permissions are too broad" + return "" + + +def _policy_parent_trust_error(path: Path) -> str: + expected_uid, expected_gid = _expected_policy_owner() + for index, parent in enumerate([path.parent, *path.parent.parents]): + try: + info = parent.lstat() + except FileNotFoundError: + return "execution policy parent directory is unavailable" + except OSError: + return "execution policy parent directory is unreadable" + if stat.S_ISLNK(info.st_mode): + return "execution policy parent directory is a symlink" + if not stat.S_ISDIR(info.st_mode): + return "execution policy parent path is not a directory" + if index == 0: + trust_error = _policy_metadata_trust_error(info, kind="parent directory") + if trust_error: + return trust_error + continue + owner_ok = (info.st_uid, info.st_gid) == (expected_uid, expected_gid) or info.st_uid == 0 + if not owner_ok: + return "execution policy parent directory owner is invalid" + if info.st_mode & (stat.S_IWGRP | stat.S_IWOTH): + return "execution policy parent directory permissions are too broad" + return "" + + +def _read_trusted_policy_file(path: Path) -> tuple[str, str]: + parent_error = _policy_parent_trust_error(path) + if parent_error: + return "", parent_error + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + fd = os.open(path, flags) + except FileNotFoundError: + return "", "execution policy file is unavailable" + except OSError: + return "", "execution policy file is unreadable" + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode): + return "", "execution policy file is not a regular file" + trust_error = _policy_metadata_trust_error(info, kind="file") + if trust_error: + return "", trust_error + with os.fdopen(fd, "r", encoding="utf-8") as handle: + fd = -1 + return handle.read(), "" + except UnicodeDecodeError: + return "", "execution policy file is unreadable" + except OSError: + return "", "execution policy file is unreadable" + finally: + if fd >= 0: + os.close(fd) + + +def _validate_execution_policy(payload: dict[str, Any]) -> str: + default_policy = payload.get("default") + if not isinstance(default_policy, dict): + return "execution policy default section is invalid" + default_error = _validate_policy_section(default_policy, section_name="default", require_defaults=True) + if default_error: + return default_error + repositories = payload.get("repositories") + if not isinstance(repositories, dict): + return "execution policy repositories section is invalid" + for repo, repo_policy in repositories.items(): + if not isinstance(repo_policy, dict): + return f"execution policy override for {repo!r} is invalid" + if not repo_policy: + return f"execution policy override for {repo!r} is empty" + repo_error = _validate_policy_section(repo_policy, section_name=f"override for {repo!r}", require_defaults=False) + if repo_error: + return repo_error + return "" + + +def _validate_policy_section(section: dict[str, Any], *, section_name: str, require_defaults: bool) -> str: + unknown_keys = set(section) - POLICY_ALLOWED_KEYS + if unknown_keys: + return f"execution policy {section_name} has unknown keys" + if require_defaults: + missing_keys = POLICY_REQUIRED_DEFAULT_KEYS - set(section) + if missing_keys: + return f"execution policy {section_name} is missing required keys" + if "max_autonomy" in section and str(section["max_autonomy"] or "").strip().lower() not in AUTONOMY_RANK: + return f"execution policy {section_name} has invalid max_autonomy" + if "max_consecutive_failures" in section and _safe_positive_int(section["max_consecutive_failures"], 0) <= 0: + return f"execution policy {section_name} has invalid max_consecutive_failures" + if "low_cost_model" in section and not str(section["low_cost_model"] or "").strip(): + return f"execution policy {section_name} has invalid low_cost_model" + if "low_cost_provider" in section: + provider = str(section["low_cost_provider"] or "").strip().lower() + if provider not in POLICY_PROVIDERS: + return f"execution policy {section_name} has invalid low_cost_provider" + if "quota_low_behavior" in section: + behavior = str(section["quota_low_behavior"] or "").strip().lower() + if behavior not in POLICY_LOW_QUOTA_BEHAVIORS: + return f"execution policy {section_name} has invalid quota_low_behavior" + return "" + + +def load_execution_policy(path: Path | None = None) -> dict[str, Any]: + """Load admin-owned execution policy for repo autonomy thresholds.""" + require_trusted_path = path is None + if path is None: + configured = os.environ.get(EXECUTION_POLICY_PATH_ENV, "").strip() + if not configured: + return _fail_closed_policy("execution policy path is not configured") + path = Path(configured).expanduser() + if not path.is_absolute(): + return _fail_closed_policy("execution policy path must be absolute") + if require_trusted_path: + raw, read_error = _read_trusted_policy_file(path) + if read_error: + return _fail_closed_policy(read_error) + else: + try: + raw = path.read_text(encoding="utf-8") + except FileNotFoundError: + return _fail_closed_policy("execution policy file is unavailable") + except UnicodeDecodeError: + return _fail_closed_policy("execution policy file is unreadable") + except OSError: + return _fail_closed_policy("execution policy file is unreadable") + try: + payload = json.loads(raw) + except json.JSONDecodeError: + return _fail_closed_policy("execution policy file is unreadable") + if not isinstance(payload, dict): + return _fail_closed_policy("execution policy file is invalid") + schema_error = _validate_execution_policy(payload) + if schema_error: + return _fail_closed_policy(schema_error) + return payload + + +def repo_execution_policy(repo: str, policy: dict[str, Any] | None = None) -> dict[str, Any]: + """Merge default and repo-specific execution policy without trusting repo checkouts.""" + raw = policy if isinstance(policy, dict) else {} + defaults = raw.get("default") if isinstance(raw.get("default"), dict) else {} + repositories = raw.get("repositories") if isinstance(raw.get("repositories"), dict) else {} + normalized_repo = _normalize_repo_id(repo) + override = {} + for configured_repo, configured_policy in repositories.items(): + if _normalize_repo_id(configured_repo) == normalized_repo and isinstance(configured_policy, dict): + override = configured_policy + break + return {**defaults, **override} + + +def consecutive_failure_count( + runs: list[dict[str, Any]], + *, + repo: str, +) -> int: + """Count latest consecutive trusted failed runs for one repo from newest-first runs.""" + count = 0 + normalized_repo = _normalize_repo_id(repo) + for run in runs: + if not isinstance(run, dict): + continue + metadata = run.get("metadata") if isinstance(run.get("metadata"), dict) else {} + origin = str(metadata.get("origin") or "") + if normalized_repo and _normalize_repo_id(_repo_from_run(run)) != normalized_repo: + continue + state = str(run.get("task_state") or "").strip().lower() + if origin not in TRUSTED_FAILURE_ORIGINS: + continue + if state in {"failed", "blocked"}: + count += 1 + continue + if state in {"queued", "running", "pending", "in_progress"}: + continue + break + return count + + +def decide_automation_execution( + *, + repo: str, + task_name: str = "", + requested_mode: str = MODE_REVIEW_AND_FIX, + requested_provider: str = "auto", + requested_model: str = "", + control_action: str = CONTROL_REVIEW_ONLY, + service_health: Any = "", + quota_status: Any = "", + org_health_status: Any = "", + recent_runs: list[dict[str, Any]] | None = None, + failure_history_complete: bool = True, + policy: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Produce a safe execution decision from health, quota, failures, and repo policy.""" + repo_policy = repo_execution_policy(repo, policy) + policy_load_error = str((policy or {}).get(POLICY_LOAD_ERROR_KEY) or "") if isinstance(policy, dict) else "" + max_autonomy, autonomy_config_error = _parse_autonomy(repo_policy.get("max_autonomy"), AUTONOMY_AUTO_PR) + max_failures = _safe_positive_int(repo_policy.get("max_consecutive_failures"), DEFAULT_MAX_CONSECUTIVE_FAILURES) + low_cost_model = str(repo_policy.get("low_cost_model") or DEFAULT_LOW_COST_MODEL) + low_cost_provider = str(repo_policy.get("low_cost_provider") or DEFAULT_LOW_COST_PROVIDER).strip().lower() + quota_low_behavior = str(repo_policy.get("quota_low_behavior") or "low_cost_model").strip().lower() + + reasons: list[str] = [] + requested_autonomy = _normalize_requested_autonomy(requested_mode) + effective_autonomy = requested_autonomy + if AUTONOMY_RANK[effective_autonomy] > AUTONOMY_RANK[max_autonomy]: + effective_autonomy = max_autonomy + reasons.append(f"requested autonomy {requested_autonomy} exceeds repo max autonomy {max_autonomy}; capping") + effective_mode = _normalize_mode(requested_mode) + if AUTONOMY_RANK[effective_autonomy] <= AUTONOMY_RANK[AUTONOMY_REVIEW_ONLY]: + effective_mode = MODE_REVIEW_ONLY + effective_provider = str(requested_provider or "auto").strip().lower() or "auto" + effective_model = str(requested_model or "").strip() + action = EXECUTION_RUN + human_review_required = False + defer = False + + service = _normalize_status(service_health) + quota = _normalize_quota_status(quota_status) + org_health = _normalize_status(org_health_status) + low_quota_pause_only = control_action == CONTROL_PAUSE_AUTO_FIX and quota in {"low", "constrained"} and quota_low_behavior != "defer" + low_quota_pause_only = low_quota_pause_only and service in {"healthy", "ok"} and org_health in {"healthy", "ok"} + failures = consecutive_failure_count( + recent_runs or [], + repo=repo, + ) + + if AUTONOMY_RANK[max_autonomy] <= AUTONOMY_RANK[AUTONOMY_REVIEW_ONLY]: + effective_mode = MODE_REVIEW_ONLY + reasons.append(f"repo max autonomy is {max_autonomy}") + if autonomy_config_error: + reasons.append(autonomy_config_error) + if policy_load_error: + reasons.append(policy_load_error) + if max_autonomy == AUTONOMY_MANUAL: + action = EXECUTION_HUMAN_REVIEW + human_review_required = True + elif requested_autonomy == AUTONOMY_MANUAL: + action = EXECUTION_REVIEW_ONLY + effective_mode = MODE_REVIEW_ONLY + reasons.append("manual mode requested; forcing review_only") + + if failures >= max_failures: + action = EXECUTION_HUMAN_REVIEW + effective_mode = MODE_REVIEW_ONLY + human_review_required = True + reasons.append(f"consecutive failures reached {failures}/{max_failures}") + elif not failure_history_complete: + action = EXECUTION_HUMAN_REVIEW + effective_mode = MODE_REVIEW_ONLY + human_review_required = True + reasons.append("failure history may be truncated; forcing human review") + + if control_action in {CONTROL_REVIEW_ONLY, CONTROL_ESCALATE} or ( + control_action == CONTROL_PAUSE_AUTO_FIX and not low_quota_pause_only + ): + effective_mode = MODE_REVIEW_ONLY + reasons.append(f"runtime control action is {control_action}") + if control_action == CONTROL_ESCALATE: + action = EXECUTION_HUMAN_REVIEW + human_review_required = True + + if service == "degraded" or org_health == "degraded": + effective_mode = MODE_REVIEW_ONLY + reasons.append("health degraded; forcing review_only") + if service == "unhealthy" or org_health == "unhealthy": + action = EXECUTION_HUMAN_REVIEW + effective_mode = MODE_REVIEW_ONLY + human_review_required = True + reasons.append("health unhealthy; forcing human review") + + if quota in {"low", "constrained"}: + if action in {EXECUTION_HUMAN_REVIEW, EXECUTION_DEFER}: + reasons.append(f"quota status is {quota}; execution already blocked") + elif quota_low_behavior == "defer" and action == EXECUTION_RUN and AUTONOMY_RANK[effective_autonomy] > AUTONOMY_RANK[AUTONOMY_REVIEW_ONLY]: + action = EXECUTION_DEFER + defer = True + effective_mode = MODE_REVIEW_ONLY + reasons.append(f"quota status is {quota}; deferring automation") + elif quota_low_behavior == "defer": + reasons.append(f"quota status is {quota}; automation already review_only") + else: + effective_model = low_cost_model or recommend_model(0.0) + effective_provider = low_cost_provider or "auto" + reasons.append(f"quota status is {quota}; recommending low-cost model") + elif quota in {"exhausted", "blocked"}: + if action != EXECUTION_HUMAN_REVIEW: + action = EXECUTION_DEFER + defer = True + reasons.append(f"quota status is {quota}; deferring automation") + else: + reasons.append(f"quota status is {quota}; execution already blocked") + effective_mode = MODE_REVIEW_ONLY + if action == EXECUTION_RUN and effective_mode == MODE_REVIEW_ONLY: + action = EXECUTION_REVIEW_ONLY + human_review_required = action == EXECUTION_HUMAN_REVIEW + + return { + "action": action, + "repo": repo, + "task_name": task_name, + "requested_mode": _normalize_mode(requested_mode), + "effective_mode": effective_mode, + "requested_autonomy": requested_autonomy, + "effective_autonomy": effective_autonomy, + "requested_provider": requested_provider, + "effective_provider": effective_provider, + "requested_model": requested_model, + "effective_model": effective_model, + "max_autonomy": max_autonomy, + "consecutive_failures": failures, + "max_consecutive_failures": max_failures, + "failure_history_complete": failure_history_complete, + "human_review_required": human_review_required, + "auto_fix_allowed": action == EXECUTION_RUN and effective_mode == MODE_REVIEW_AND_FIX and not human_review_required, + "auto_merge_allowed": ( + action == EXECUTION_RUN + and effective_mode == MODE_REVIEW_AND_FIX + and effective_autonomy == AUTONOMY_AUTO_MERGE + and not human_review_required + ), + "defer": defer, + "reasons": reasons or ["execution allowed"], + } diff --git a/service/automation_run_ledger.py b/service/automation_run_ledger.py index c62637f6..fef8be07 100644 --- a/service/automation_run_ledger.py +++ b/service/automation_run_ledger.py @@ -132,6 +132,10 @@ def _entry_owner_repository(entry: dict[str, Any]) -> str: return str(metadata.get("source_repository") or metadata.get("repository") or "") +def _repo_retention_key(value: Any) -> str: + return str(value or "").strip().lower() + + def _entry_origin(entry: dict[str, Any]) -> str: metadata = entry.get("metadata") if isinstance(entry.get("metadata"), dict) else {} return str(metadata.get("origin") or "") @@ -214,37 +218,79 @@ def __init__( self._max_runs = max(1, int(max_runs)) self._max_events_per_run = max(1, int(max_events_per_run)) self._sequence = 0 + self._evicted_runs_count = 0 + self._evicted_runs_by_repo: dict[str, int] = {} + self._history_completeness_unknown = False + self._storage_unavailable = False + self._storage_file_seen = False self._storage_path = storage_path self._load_from_disk() def _load_from_disk(self) -> None: - if self._storage_path is None or not self._storage_path.exists(): + if self._storage_path is None: return - runs, sequence = self._read_from_disk_unlocked() + if not self._storage_path.exists(): + return + runs, sequence, evicted_runs, evicted_runs_by_repo, history_unknown, read_ok = self._read_from_disk_unlocked() with self._lock: + self._storage_file_seen = True self._runs = runs self._sequence = sequence + self._evicted_runs_count = evicted_runs + self._evicted_runs_by_repo = evicted_runs_by_repo + self._history_completeness_unknown = history_unknown + self._storage_unavailable = not read_ok self._evict_old_runs_locked() - def _read_from_disk_unlocked(self) -> tuple[dict[str, dict[str, Any]], int]: + def _read_from_disk_unlocked(self) -> tuple[dict[str, dict[str, Any]], int, int, dict[str, int], bool, bool]: if self._storage_path is None or not self._storage_path.exists(): - return {}, 0 + return {}, 0, 0, {}, self._storage_file_seen, not self._storage_file_seen try: payload = json.loads(self._storage_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): - return {}, 0 + return {}, 0, 0, {}, True, False runs = payload.get("runs") if isinstance(payload, dict) else None if not isinstance(runs, dict): - return {}, 0 + return {}, 0, 0, {}, True, False clean_runs = {str(key): value for key, value in runs.items() if isinstance(value, dict)} sequence = _safe_int(payload.get("sequence"), len(clean_runs)) - return clean_runs, sequence + evicted_runs = _safe_int(payload.get("evicted_runs"), 0) + raw_evicted_by_repo = payload.get("evicted_runs_by_repo") + history_unknown = "evicted_runs" not in payload or not isinstance(raw_evicted_by_repo, dict) + evicted_by_repo = {} + for repo, count in (raw_evicted_by_repo.items() if isinstance(raw_evicted_by_repo, dict) else []): + repo_key = _repo_retention_key(repo) + count_value = _safe_int(count, -1) + if not repo_key or count_value < 0: + history_unknown = True + continue + evicted_by_repo[repo_key] = count_value + history_unknown = bool(payload.get("history_completeness_unknown", history_unknown)) + return clean_runs, sequence, max(0, evicted_runs), evicted_by_repo, history_unknown, True + + def _merge_evicted_runs_by_repo_locked(self, disk_evicted_runs_by_repo: dict[str, int]) -> None: + for repo, count in disk_evicted_runs_by_repo.items(): + repo_key = _repo_retention_key(repo) + if not repo_key: + continue + self._evicted_runs_by_repo[repo_key] = max(self._evicted_runs_by_repo.get(repo_key, 0), int(count)) + + def _drop_runs_evicted_on_disk_locked(self, disk_runs: dict[str, dict[str, Any]], *, preserve_run_id: str = "") -> None: + disk_run_ids = set(disk_runs) + for run_id in list(self._runs): + if run_id != preserve_run_id and run_id not in disk_run_ids: + self._runs.pop(run_id, None) def _persist_locked(self) -> None: self._persist_with_owner_guard_locked() def _refresh_from_disk_locked(self) -> None: - if self._storage_path is None or not self._storage_path.exists(): + if self._storage_path is None: + return + if not self._storage_path.exists(): + if self._storage_file_seen: + self._history_completeness_unknown = True + self._storage_unavailable = True return lock_handle = None try: @@ -252,7 +298,21 @@ def _refresh_from_disk_locked(self) -> None: lock_path = self._storage_path.with_suffix(self._storage_path.suffix + ".lock") lock_handle = lock_path.open("a+", encoding="utf-8") fcntl.flock(lock_handle.fileno(), fcntl.LOCK_SH) - disk_runs, disk_sequence = self._read_from_disk_unlocked() + ( + disk_runs, + disk_sequence, + disk_evicted_runs, + disk_evicted_runs_by_repo, + disk_history_unknown, + disk_read_ok, + ) = self._read_from_disk_unlocked() + self._storage_file_seen = True + if not disk_read_ok: + self._history_completeness_unknown = True + self._storage_unavailable = True + return + self._storage_unavailable = False + self._drop_runs_evicted_on_disk_locked(disk_runs) for run_id, disk_entry in disk_runs.items(): current = self._runs.get(run_id) if current is not None: @@ -260,6 +320,9 @@ def _refresh_from_disk_locked(self) -> None: else: self._runs[run_id] = disk_entry self._sequence = max(self._sequence, disk_sequence, len(self._runs)) + self._evicted_runs_count = max(self._evicted_runs_count, disk_evicted_runs) + self._merge_evicted_runs_by_repo_locked(disk_evicted_runs_by_repo) + self._history_completeness_unknown = self._history_completeness_unknown or disk_history_unknown self._evict_old_runs_locked() finally: if lock_handle is not None: @@ -273,6 +336,7 @@ def _persist_with_owner_guard_locked( *, guard_run_id: str = "", owner_repository: str = "", + guard_preexisting: bool = False, ) -> None: if self._storage_path is None: return @@ -283,12 +347,27 @@ def _persist_with_owner_guard_locked( lock_path = self._storage_path.with_suffix(self._storage_path.suffix + ".lock") lock_handle = lock_path.open("a+", encoding="utf-8") fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) - disk_runs, disk_sequence = self._read_from_disk_unlocked() - if guard_run_id and owner_repository: + had_seen_storage = self._storage_file_seen + ( + disk_runs, + disk_sequence, + disk_evicted_runs, + disk_evicted_runs_by_repo, + disk_history_unknown, + disk_read_ok, + ) = self._read_from_disk_unlocked() + self._storage_file_seen = True + if not disk_read_ok: + self._history_completeness_unknown = True + raise OSError("automation ledger could not be refreshed from disk") + if guard_run_id: disk_entry = disk_runs.get(guard_run_id) disk_owner = _entry_owner_repository(disk_entry) if isinstance(disk_entry, dict) else "" - if disk_owner and disk_owner != owner_repository: + if owner_repository and disk_owner and disk_owner != owner_repository: raise PermissionError("automation run_id belongs to another repository") + if guard_preexisting and disk_entry is None and had_seen_storage: + raise ValueError("automation run was evicted from retained ledger") + self._drop_runs_evicted_on_disk_locked(disk_runs, preserve_run_id=guard_run_id) for run_id, disk_entry in disk_runs.items(): current = self._runs.get(run_id) if current is not None: @@ -296,6 +375,9 @@ def _persist_with_owner_guard_locked( else: self._runs[run_id] = disk_entry self._sequence = max(self._sequence, disk_sequence, len(self._runs)) + self._evicted_runs_count = max(self._evicted_runs_count, disk_evicted_runs) + self._merge_evicted_runs_by_repo_locked(disk_evicted_runs_by_repo) + self._history_completeness_unknown = self._history_completeness_unknown or disk_history_unknown self._evict_old_runs_locked() self._write_to_disk_locked() finally: @@ -311,6 +393,9 @@ def _write_to_disk_locked(self) -> None: payload = { "schema_version": "automation_run_ledger.v1", "sequence": self._sequence, + "evicted_runs": self._evicted_runs_count, + "evicted_runs_by_repo": self._evicted_runs_by_repo, + "history_completeness_unknown": self._history_completeness_unknown, "runs": self._runs, } tmp = self._storage_path.with_suffix(self._storage_path.suffix + ".tmp") @@ -320,6 +405,8 @@ def _write_to_disk_locked(self) -> None: except OSError: pass os.replace(tmp, self._storage_path) + self._storage_file_seen = True + self._storage_unavailable = False def _merge_entry_locked(self, current: dict[str, Any], disk_entry: dict[str, Any]) -> dict[str, Any]: current_terminal = _is_terminal_task_state(current.get("task_state")) @@ -365,7 +452,11 @@ def _evict_old_runs_locked(self) -> None: ), ) for entry in ordered[:overflow]: + repo_key = _repo_retention_key(_entry_owner_repository(entry)) + if repo_key: + self._evicted_runs_by_repo[repo_key] = self._evicted_runs_by_repo.get(repo_key, 0) + 1 self._runs.pop(str(entry["run_id"])) + self._evicted_runs_count += overflow @staticmethod def _public_entry(entry: dict[str, Any], *, include_events: bool = True) -> dict[str, Any]: @@ -407,6 +498,11 @@ def record( with self._lock: previous_runs = deepcopy(self._runs) previous_sequence = self._sequence + previous_evicted_runs_count = self._evicted_runs_count + previous_evicted_runs_by_repo = dict(self._evicted_runs_by_repo) + previous_history_completeness_unknown = self._history_completeness_unknown + previous_storage_unavailable = self._storage_unavailable + previous_storage_file_seen = self._storage_file_seen current = self._runs.get(run_id) if current: current_owner = _entry_owner_repository(current) @@ -419,7 +515,8 @@ def record( and not _can_replace_stale_service_failure(current, entry) ): return self._public_entry(current) - entry["_ledger_sequence"] = current.get("_ledger_sequence", 0) + self._sequence += 1 + entry["_ledger_sequence"] = self._sequence old_events = list(current.get("events", [])) entry["events"] = ( old_events[-(self._max_events_per_run - 1) :] if self._max_events_per_run > 1 else [] @@ -451,12 +548,22 @@ def record( } ) self._runs[run_id] = entry - self._evict_old_runs_locked() + if self._storage_path is None: + self._evict_old_runs_locked() try: - self._persist_with_owner_guard_locked(guard_run_id=run_id, owner_repository=owner_repository) + self._persist_with_owner_guard_locked( + guard_run_id=run_id, + owner_repository=owner_repository, + guard_preexisting=current is not None, + ) except Exception: self._runs = previous_runs self._sequence = previous_sequence + self._evicted_runs_count = previous_evicted_runs_count + self._evicted_runs_by_repo = previous_evicted_runs_by_repo + self._history_completeness_unknown = previous_history_completeness_unknown + self._storage_unavailable = previous_storage_unavailable + self._storage_file_seen = previous_storage_file_seen raise return self._public_entry(self._runs[run_id]) @@ -467,14 +574,19 @@ def get(self, run_id: str) -> dict[str, Any] | None: return self._public_entry(entry) if entry else None def snapshot(self, *, limit: int | None = 100, include_events: bool = False) -> dict[str, Any]: + """Return retained runs; ``limit=None`` returns the full retained ledger.""" with self._lock: self._refresh_from_disk_locked() retained_runs = [ self._public_entry(entry, include_events=include_events) - for entry in self._runs.values() + for entry in sorted(self._runs.values(), key=_entry_order_key, reverse=True) ] max_runs = self._max_runs max_events_per_run = self._max_events_per_run + evicted_runs_count = self._evicted_runs_count + evicted_runs_by_repo = dict(self._evicted_runs_by_repo) + history_completeness_unknown = self._history_completeness_unknown + storage_unavailable = self._storage_unavailable task_states = Counter(str(run.get("task_state", "")).strip().lower() for run in retained_runs if run.get("task_state")) suggested_actions = Counter( @@ -483,11 +595,7 @@ def snapshot(self, *, limit: int | None = 100, include_events: bool = False) -> if run.get("suggested_action") ) terminal_runs = sum(1 for run in retained_runs if str(run.get("task_state", "")).strip().lower() in TERMINAL_STATES) - ordered_runs = sorted( - retained_runs, - key=lambda item: (float(item.get("updated_at", 0.0)), str(item.get("run_id", ""))), - reverse=True, - ) + ordered_runs = retained_runs if limit is not None: ordered_runs = ordered_runs[: max(0, int(limit))] runs = ordered_runs @@ -504,6 +612,11 @@ def snapshot(self, *, limit: int | None = 100, include_events: bool = False) -> "max_runs": max_runs, "max_events_per_run": max_events_per_run, "events_included": include_events, + "evicted_runs": evicted_runs_count, + "evicted_runs_by_repo": evicted_runs_by_repo, + "history_completeness_unknown": history_completeness_unknown, + "storage_unavailable": storage_unavailable, + "may_be_truncated": evicted_runs_count > 0, }, }, } diff --git a/tests/test_ai_gateway_automation_control.py b/tests/test_ai_gateway_automation_control.py new file mode 100644 index 00000000..74a9a5b0 --- /dev/null +++ b/tests/test_ai_gateway_automation_control.py @@ -0,0 +1,377 @@ +"""Tests for automation control-plane execution decisions.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import patch + +from service.ai_gateway_service import _automation_control_snapshot, _automation_triage_snapshot + + +class TestAutomationControlSnapshot(unittest.TestCase): + def test_control_snapshot_defaults_to_review_and_fix_for_healthy_repo(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: {"runs": []}})() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo") + + self.assertTrue(control["execution"]["auto_fix_allowed"]) + + def test_control_snapshot_downgrades_legacy_action_for_review_only_mode(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: {"runs": []}})() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", requested_mode="review_only") + + self.assertEqual(control["effective_action"], "review_only") + + def test_control_snapshot_applies_service_owned_execution_policy(self) -> None: + with TemporaryDirectory(dir=".") as tmp: + policy_path = Path(tmp) / "execution_policy.json" + policy_path.write_text( + json.dumps( + { + "default": { + "max_autonomy": "auto_pr", + "max_consecutive_failures": 3, + "low_cost_model": "gpt-5.4-mini", + "low_cost_provider": "openai", + }, + "repositories": { + "QuantStrategyLab/TargetRepo": { + "max_autonomy": "manual", + "max_consecutive_failures": 2, + } + } + } + ), + encoding="utf-8", + ) + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=20: {"runs": []}})() + + with ( + patch.dict( + os.environ, + { + "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH": str(policy_path), + "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_OWNER": f"{os.getuid()}:{os.getgid()}", + }, + clear=False, + ), + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", requested_mode="review_and_fix") + + self.assertEqual(control["effective_action"], "escalate") + + def test_control_snapshot_scans_full_retained_ledger_for_repo_failure_streak(self) -> None: + runs = [ + { + "task_name": f"other-{index}", + "task_state": "merged", + "metadata": {"origin": "service_job", "source_repository": f"QuantStrategyLab/Other{index}"}, + } + for index in range(20) + ] + runs.extend( + [ + { + "task_name": "monthly", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/TargetRepo"}, + }, + { + "task_name": "monthly", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/TargetRepo"}, + }, + ] + ) + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + + class Ledger: + requested_limit = object() + + def snapshot(self, limit=100): + self.requested_limit = limit + return {"runs": runs} + + ledger = Ledger() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_consecutive_failures": 2}}), + ): + _automation_control_snapshot("QuantStrategyLab/TargetRepo", task_name="monthly") + + self.assertIsNone(ledger.requested_limit) + + def test_control_snapshot_counts_pending_run_for_failure_threshold(self) -> None: + runs = [ + { + "run_id": "previous-run", + "task_name": "monthly", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/TargetRepo"}, + } + ] + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: {"runs": runs}})() + pending_run = { + "run_id": "current-run", + "task_name": "monthly", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/TargetRepo"}, + } + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_consecutive_failures": 2}}), + ): + control = _automation_control_snapshot( + "QuantStrategyLab/TargetRepo", + task_name="monthly", + requested_mode="review_and_fix", + pending_run=pending_run, + ) + + self.assertEqual(control["effective_action"], "escalate") + + def test_control_snapshot_fails_closed_after_ledger_eviction(self) -> None: + runs = [ + { + "run_id": "failed-1", + "task_name": "monthly", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/TargetRepo"}, + }, + ] + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type( + "Ledger", + (), + { + "snapshot": lambda self, limit=None: { + "runs": runs, + "summary": { + "retention": { + "may_be_truncated": True, + "evicted_runs": 1, + "evicted_runs_by_repo": {"quantstrategylab/targetrepo": 1}, + } + }, + } + }, + )() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_consecutive_failures": 2}}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", task_name="monthly") + + self.assertEqual(control["effective_action"], "escalate") + + def test_control_snapshot_allows_known_repo_boundary_when_history_unknown(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type( + "Ledger", + (), + { + "snapshot": lambda self, limit=None: { + "runs": [ + {"run_id": "merged-1", "task_state": "merged", "metadata": {"origin": "external_workflow", "source_repository": "QuantStrategyLab/TargetRepo"}} + ], + "summary": { + "retention": { + "history_completeness_unknown": True, + "may_be_truncated": True, + "evicted_runs": 1, + "evicted_runs_by_repo": {"quantstrategylab/targetrepo": 0}, + } + }, + } + }, + )() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_consecutive_failures": 2}}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", task_name="monthly", requested_mode="review_and_fix") + + self.assertTrue(control["execution"]["failure_history_complete"]) + + def test_control_snapshot_fails_closed_when_ledger_history_is_unknown(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type( + "Ledger", + (), + { + "snapshot": lambda self, limit=None: { + "runs": [], + "summary": {"retention": {"history_completeness_unknown": True}}, + } + }, + )() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_consecutive_failures": 2}}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", task_name="monthly") + + self.assertEqual(control["effective_action"], "escalate") + + def test_control_snapshot_preserves_legacy_continue_when_auto_merge_is_capped(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: {"runs": []}})() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_autonomy": "auto_pr"}}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", requested_mode="auto_merge") + + self.assertEqual(control["execution"]["effective_autonomy"], "auto_pr") + + def test_triage_omitted_mode_keeps_review_and_fix_default(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: {"runs": []}})() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={}), + ): + triage = _automation_triage_snapshot( + "QuantStrategyLab/TargetRepo", + task="monthly", + changed_paths=["docs/runbook.md"], + ) + + self.assertEqual(triage["control"]["execution"]["requested_mode"], "review_and_fix") + + def test_control_snapshot_deduplicates_pending_run_by_run_id(self) -> None: + runs = [ + { + "run_id": "current-run", + "task_name": "monthly", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/TargetRepo"}, + } + ] + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: {"runs": runs}})() + pending_run = { + "run_id": "current-run", + "task_name": "monthly", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/TargetRepo"}, + } + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_consecutive_failures": 2}}), + ): + control = _automation_control_snapshot( + "QuantStrategyLab/TargetRepo", + task_name="monthly", + requested_mode="review_and_fix", + pending_run=pending_run, + ) + + self.assertEqual(control["execution"]["consecutive_failures"], 1) + + def test_control_snapshot_allows_low_cost_auto_fix_for_low_quota_policy(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "low"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: {"runs": []}})() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"low_cost_model": "gpt-5.4-mini"}}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", requested_mode="review_and_fix") + + self.assertEqual(control["action"], "continue") + self.assertEqual(control["effective_action"], "continue") + + def test_control_snapshot_fails_closed_when_ledger_is_unavailable(self) -> None: + health = type("Health", (), {"status": "healthy"})() + quota = type("Quota", (), {"runtime_status": lambda self, repo: {"status": "ok"}})() + ledger = type("Ledger", (), {"snapshot": lambda self, limit=None: (_ for _ in ()).throw(RuntimeError("boom"))})() + + with ( + patch("service.ai_gateway_service.read_org_health", return_value={"status": "ok"}), + patch("service.ai_gateway_service.get_health_monitor", return_value=health), + patch("service.ai_gateway_service.get_quota_manager", return_value=quota), + patch("service.ai_gateway_service.get_automation_run_ledger", return_value=ledger), + patch("service.ai_gateway_service.load_execution_policy", return_value={"default": {"max_autonomy": "auto_merge"}}), + ): + control = _automation_control_snapshot("QuantStrategyLab/TargetRepo", requested_mode="auto_merge") + + self.assertEqual(control["effective_action"], "escalate") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ai_gateway_service_get_routes.py b/tests/test_ai_gateway_service_get_routes.py index d83c062a..26b47501 100644 --- a/tests/test_ai_gateway_service_get_routes.py +++ b/tests/test_ai_gateway_service_get_routes.py @@ -184,7 +184,7 @@ def test_static_token_dashboard_can_read_allowlisted_repository_runs(self) -> No self.assertEqual([run["run_id"] for run in filtered["runs"]], ["run-a"]) def test_automation_run_update_rejects_owner_mismatch_even_for_operator(self) -> None: - with tempfile.TemporaryDirectory() as tmp: + with tempfile.TemporaryDirectory(dir=".") as tmp: env = { "CODEX_AUDIT_SERVICE_AUTH": "none", "CODEX_AUDIT_SERVICE_ALLOW_NO_AUTH_FOR_LOCAL_TESTS": "true", @@ -372,6 +372,30 @@ def test_static_token_dashboard_can_query_allowlisted_control_repo(self) -> None ) with urllib.request.urlopen(request, timeout=5) as response: self.assertEqual(response.status, 200) + control = json.loads(response.read().decode("utf-8"))["control"] + self.assertIn("execution", control) + self.assertEqual(control["execution"]["repo"], "QuantStrategyLab/TargetRepo") + self.assertEqual(control["execution"]["requested_mode"], "review_and_fix") + self.assertEqual(control["execution"]["effective_mode"], "review_only") + self.assertFalse(control["execution"]["auto_fix_allowed"]) + + for legacy_mode, expected_mode in {"manual": "review_only", "auto_pr": "review_and_fix", "auto_merge": "review_and_fix"}.items(): + legacy_mode_request = urllib.request.Request( + f"{base_url}/v1/ai/automation/control?repo=QuantStrategyLab/TargetRepo&mode={legacy_mode}", + headers={"Authorization": f"Bearer {token}"}, + ) + with urllib.request.urlopen(legacy_mode_request, timeout=5) as response: + self.assertEqual(response.status, 200) + legacy_control = json.loads(response.read().decode("utf-8"))["control"] + self.assertEqual(legacy_control["execution"]["requested_mode"], expected_mode) + + invalid_mode_request = urllib.request.Request( + f"{base_url}/v1/ai/automation/control?repo=QuantStrategyLab/TargetRepo&mode=bad", + headers={"Authorization": f"Bearer {token}"}, + ) + with self.assertRaises(urllib.error.HTTPError) as ctx: + urllib.request.urlopen(invalid_mode_request, timeout=5) + self.assertEqual(ctx.exception.code, 400) missing_repo_request = urllib.request.Request( f"{base_url}/v1/ai/automation/control", @@ -558,14 +582,55 @@ def test_automation_routes_record_and_return_run_ledger(self) -> None: self.assertEqual(response.status, 200) recorded = json.loads(response.read().decode("utf-8")) self.assertEqual(recorded["run"]["run_id"], "platform-health-run-1") - self.assertEqual(recorded["run"]["suggested_action"], recorded["control"]["action"]) + self.assertEqual(recorded["run"]["suggested_action"], recorded["control"]["effective_action"]) + self.assertEqual(recorded["control"]["execution"]["requested_mode"], "review_and_fix") + self.assertEqual(recorded["run"]["metadata"]["requested_mode"], "review_and_fix") self.assertEqual(recorded["run"]["service_health"], recorded["control"]["service_health"]) self.assertEqual(recorded["run"]["quota_status"], recorded["control"]["quota_status"]) self.assertEqual(recorded["run"]["org_health_status"], recorded["control"]["org_health_status"]) + manual_payload = {**payload, "run_id": "platform-health-run-manual", "mode": "manual"} + manual_request = urllib.request.Request( + f"{base_url}/v1/ai/automation/runs", + data=json.dumps(manual_payload).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(manual_request, timeout=5) as response: + manual_recorded = json.loads(response.read().decode("utf-8")) + self.assertEqual(manual_recorded["control"]["execution"]["requested_mode"], "review_only") + self.assertEqual(manual_recorded["run"]["metadata"]["requested_mode"], "manual") + + manual_update_payload = { + **payload, + "run_id": "platform-health-run-manual", + "task_state": "failed", + "metadata": {"requested_mode": "auto_merge", "mode": "auto_merge"}, + } + manual_update_request = urllib.request.Request( + f"{base_url}/v1/ai/automation/runs", + data=json.dumps(manual_update_payload).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(manual_update_request, timeout=5) as response: + manual_updated = json.loads(response.read().decode("utf-8")) + self.assertEqual(manual_updated["control"]["execution"]["requested_mode"], "review_only") + self.assertEqual(manual_updated["run"]["metadata"]["requested_mode"], "manual") + + invalid_mode_payload = {**payload, "run_id": "platform-health-run-invalid", "mode": ""} + invalid_mode_request = urllib.request.Request( + f"{base_url}/v1/ai/automation/runs", + data=json.dumps(invalid_mode_payload).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(invalid_mode_request, timeout=5) as response: + self.assertEqual(response.status, 200) + with urllib.request.urlopen(f"{base_url}/v1/ai/automation/runs?include_events=true", timeout=5) as response: ledger = json.loads(response.read().decode("utf-8"))["ledger"] - self.assertEqual(ledger["summary"]["total_runs"], 1) + self.assertEqual(ledger["summary"]["total_runs"], 3) self.assertEqual(ledger["runs"][0]["task_name"], "platform-health") self.assertIn("events", ledger["runs"][0]) @@ -576,16 +641,33 @@ def test_automation_routes_record_and_return_run_ledger(self) -> None: with urllib.request.urlopen(f"{base_url}/v1/ai/automation/control?repo=local/repo", timeout=5) as response: control = json.loads(response.read().decode("utf-8"))["control"] self.assertIn(control["action"], {"continue", "review_only", "pause_auto_fix", "escalate"}) + self.assertIn("execution", control) finally: server.shutdown() server.server_close() def test_automation_triage_reports_retryable_incident(self) -> None: - with tempfile.TemporaryDirectory() as tmp: + with tempfile.TemporaryDirectory(dir=".") as tmp: + policy_path = os.path.join(tmp, "execution_policy.json") + with open(policy_path, "w", encoding="utf-8") as handle: + json.dump( + { + "default": { + "max_autonomy": "auto_pr", + "max_consecutive_failures": 3, + "low_cost_model": "gpt-5.4-mini", + "low_cost_provider": "openai", + }, + "repositories": {}, + }, + handle, + ) env = { "CODEX_AUDIT_SERVICE_AUTH": "none", "CODEX_AUDIT_SERVICE_ALLOW_NO_AUTH_FOR_LOCAL_TESTS": "true", "CODEX_AUDIT_SERVICE_JOB_DIR": tmp, + "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH": policy_path, + "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_OWNER": f"{os.getuid()}:{os.getgid()}", "CODEX_AUDIT_SERVICE_AUTOMATION_OPERATOR_REPOSITORIES": "local", } with patch.dict(os.environ, env, clear=False): @@ -601,6 +683,7 @@ def test_automation_triage_reports_retryable_incident(self) -> None: "error": "codex service timed out waiting for a background job", "changed_paths": ["docs/runbook.md"], "run_id": "incident-123", + "mode": "manual", } request = urllib.request.Request( f"{base_url}/v1/ai/automation/triage", @@ -618,6 +701,7 @@ def test_automation_triage_reports_retryable_incident(self) -> None: self.assertTrue(triage["retry_allowed"]) self.assertEqual(triage["recommended_action"], "retry") self.assertEqual(triage["file_risk"], "low") + self.assertEqual(triage["control"]["execution"]["action"], "review_only") self.assertIn("run_id=incident-123", triage["summary"]) finally: server.shutdown() diff --git a/tests/test_automation_decision.py b/tests/test_automation_decision.py new file mode 100644 index 00000000..c4d7ef98 --- /dev/null +++ b/tests/test_automation_decision.py @@ -0,0 +1,385 @@ +"""Tests for health-driven automation execution decisions.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import patch + +from service.automation_decision import ( + EXECUTION_DEFER, + EXECUTION_HUMAN_REVIEW, + EXECUTION_REVIEW_ONLY, + EXECUTION_RUN, + MODE_REVIEW_AND_FIX, + MODE_REVIEW_ONLY, + consecutive_failure_count, + decide_automation_execution, + load_execution_policy, +) +from service.automation_run_ledger import CONTROL_CONTINUE, CONTROL_ESCALATE, CONTROL_PAUSE_AUTO_FIX + + +class TestAutomationDecision(unittest.TestCase): + def test_healthy_control_allows_review_and_fix(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + ) + + self.assertEqual(result["action"], EXECUTION_RUN) + self.assertEqual(result["effective_mode"], MODE_REVIEW_AND_FIX) + self.assertTrue(result["auto_fix_allowed"]) + + def test_legacy_autonomy_modes_are_normalized(self) -> None: + cases = { + "manual": (MODE_REVIEW_ONLY, "manual", "manual", False), + "auto_pr": (MODE_REVIEW_AND_FIX, "auto_pr", "auto_pr", False), + "auto_merge": (MODE_REVIEW_AND_FIX, "auto_merge", "auto_pr", False), + } + for requested_mode, (expected_mode, requested_autonomy, effective_autonomy, auto_merge_allowed) in cases.items(): + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=requested_mode, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + ) + self.assertEqual(result["requested_mode"], expected_mode) + self.assertEqual(result["requested_autonomy"], requested_autonomy) + self.assertEqual(result["effective_autonomy"], effective_autonomy) + self.assertEqual(result["auto_merge_allowed"], auto_merge_allowed) + self.assertFalse(result["human_review_required"]) + manual_result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode="manual", + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="low", + org_health_status="ok", + policy={"default": {"quota_low_behavior": "defer"}}, + ) + self.assertEqual(manual_result["action"], EXECUTION_REVIEW_ONLY) + + def test_auto_merge_requires_matching_repo_autonomy(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode="auto_merge", + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + policy={"default": {"max_autonomy": "auto_merge"}}, + ) + + self.assertEqual(result["effective_autonomy"], "auto_merge") + self.assertTrue(result["auto_merge_allowed"]) + + def test_auto_merge_is_disabled_when_effective_mode_is_review_only(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode="auto_merge", + control_action=CONTROL_CONTINUE, + service_health="degraded", + quota_status="ok", + org_health_status="ok", + policy={"default": {"max_autonomy": "auto_merge"}}, + ) + + self.assertEqual(result["action"], EXECUTION_REVIEW_ONLY) + self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY) + self.assertFalse(result["auto_fix_allowed"]) + self.assertFalse(result["auto_merge_allowed"]) + + def test_degraded_health_forces_review_only(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_PAUSE_AUTO_FIX, + service_health="degraded", + quota_status="ok", + org_health_status="ok", + ) + + self.assertEqual(result["action"], EXECUTION_REVIEW_ONLY) + self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY) + self.assertFalse(result["auto_fix_allowed"]) + self.assertFalse(result["human_review_required"]) + + def test_exhausted_quota_defers_execution_without_stronger_guard(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status={"status": "exhausted"}, + org_health_status="ok", + ) + + self.assertEqual(result["action"], EXECUTION_DEFER) + self.assertTrue(result["defer"]) + self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY) + + def test_human_review_dominates_exhausted_quota_defer(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_ESCALATE, + service_health="healthy", + quota_status={"status": "exhausted"}, + org_health_status="ok", + ) + + self.assertEqual(result["action"], EXECUTION_HUMAN_REVIEW) + self.assertFalse(result["defer"]) + self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY) + + def test_low_quota_recommends_low_cost_model(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_PAUSE_AUTO_FIX, + service_health="healthy", + quota_status="low", + org_health_status="ok", + policy={"default": {"low_cost_model": "gpt-5.4-mini"}}, + ) + + self.assertEqual(result["action"], EXECUTION_RUN) + self.assertEqual(result["effective_provider"], "openai") + self.assertEqual(result["effective_model"], "gpt-5.4-mini") + + def test_low_quota_overrides_requested_expensive_model(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + requested_model="gpt-5.4-pro", + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="low", + org_health_status="ok", + policy={"default": {"low_cost_model": "gpt-5.4-mini"}}, + ) + + self.assertEqual(result["effective_provider"], "openai") + self.assertEqual(result["effective_model"], "gpt-5.4-mini") + + def test_low_quota_does_not_override_model_when_human_review_required(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + requested_model="gpt-5.4-pro", + control_action=CONTROL_ESCALATE, + service_health="healthy", + quota_status="low", + org_health_status="ok", + policy={"default": {"low_cost_model": "gpt-5.4-mini", "low_cost_provider": "openai"}}, + ) + + self.assertEqual(result["action"], EXECUTION_HUMAN_REVIEW) + self.assertEqual(result["effective_model"], "gpt-5.4-pro") + + def test_repo_policy_can_force_review_only(self) -> None: + result = decide_automation_execution( + repo="quantstrategylab/cryptolivepoolpipelines", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + policy={"repositories": {"QuantStrategyLab/CryptoLivePoolPipelines": {"max_autonomy": "review_only"}}}, + ) + + self.assertEqual(result["action"], EXECUTION_REVIEW_ONLY) + self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY) + self.assertFalse(result["human_review_required"]) + self.assertFalse(result["auto_fix_allowed"]) + + def test_failure_streak_matching_is_case_insensitive(self) -> None: + runs = [ + {"task_name": "monthly", "task_state": "blocked", "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}}, + ] + + self.assertEqual(consecutive_failure_count(runs, repo="quantstrategylab/aiauditbridge"), 1) + + def test_external_workflow_terminal_breaks_repo_failure_streak(self) -> None: + runs = [ + {"task_name": "monthly", "task_state": "failed", "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}}, + { + "task_name": "monthly", + "task_state": "merged", + "metadata": {"origin": "external_workflow", "source_repository": "QuantStrategyLab/AIAuditBridge"}, + }, + {"task_name": "monthly", "task_state": "failed", "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}}, + ] + + self.assertEqual(consecutive_failure_count(runs, repo="QuantStrategyLab/AIAuditBridge"), 1) + + def test_invalid_repo_autonomy_fails_closed(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/CryptoLivePoolPipelines", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + policy={"repositories": {"QuantStrategyLab/CryptoLivePoolPipelines": {"max_autonomy": "auto_mrege"}}}, + ) + + self.assertEqual(result["max_autonomy"], "manual") + self.assertEqual(result["action"], EXECUTION_HUMAN_REVIEW) + self.assertTrue(any("invalid max_autonomy" in reason for reason in result["reasons"])) + + def test_consecutive_failures_force_human_review(self) -> None: + runs = [ + {"task_name": "monthly", "task_state": "failed", "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}}, + { + "task_name": "runtime-health", + "task_state": "failed", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}, + }, + ] + + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + task_name="monthly", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + recent_runs=runs, + policy={"default": {"max_consecutive_failures": 2}}, + ) + + self.assertEqual(consecutive_failure_count(runs, repo="QuantStrategyLab/AIAuditBridge"), 2) + self.assertEqual(result["action"], EXECUTION_HUMAN_REVIEW) + self.assertEqual(result["effective_mode"], MODE_REVIEW_ONLY) + + def test_truncated_failure_history_fails_closed(self) -> None: + runs = [ + {"task_name": "monthly", "task_state": "failed", "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}}, + ] + + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + recent_runs=runs, + failure_history_complete=False, + policy={"default": {"max_consecutive_failures": 2}}, + ) + + self.assertEqual(consecutive_failure_count(runs, repo="QuantStrategyLab/AIAuditBridge"), 1) + self.assertEqual(result["action"], EXECUTION_HUMAN_REVIEW) + + def test_running_state_does_not_clear_failure_streak(self) -> None: + runs = [ + { + "task_name": "monthly", + "task_state": "running", + "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}, + }, + {"task_name": "monthly", "task_state": "failed", "metadata": {"origin": "service_job", "source_repository": "QuantStrategyLab/AIAuditBridge"}}, + ] + + self.assertEqual(consecutive_failure_count(runs, repo="QuantStrategyLab/AIAuditBridge"), 1) + + def test_invalid_failure_threshold_falls_back_safely(self) -> None: + result = decide_automation_execution( + repo="QuantStrategyLab/AIAuditBridge", + requested_mode=MODE_REVIEW_AND_FIX, + control_action=CONTROL_CONTINUE, + service_health="healthy", + quota_status="ok", + org_health_status="ok", + policy={"default": {"max_consecutive_failures": "oops"}}, + ) + + self.assertEqual(result["max_consecutive_failures"], 3) + self.assertEqual(result["action"], EXECUTION_RUN) + + def test_load_execution_policy_fails_closed_for_malformed_files(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "policy.json" + path.write_text("not-json", encoding="utf-8") + + self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "manual") + + path.write_text(json.dumps({"default": []}), encoding="utf-8") + self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "manual") + + path.write_bytes(b"\xff\xfe\x00") + self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "manual") + + path.write_text(json.dumps({"repositories": {"QuantStrategyLab/AIAuditBridge": "bad"}}), encoding="utf-8") + self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "manual") + + path.write_text(json.dumps({"default": {"max_consecutive_failures": "oops"}, "repositories": {}}), encoding="utf-8") + self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "manual") + + path.write_text(json.dumps({"default": {"max_autnomy": "review_only"}, "repositories": {}}), encoding="utf-8") + self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "manual") + + path.write_text( + json.dumps( + { + "default": { + "max_autonomy": "review_only", + "max_consecutive_failures": 3, + "low_cost_model": "gpt-5.4-mini", + "low_cost_provider": "openai", + }, + "repositories": {}, + } + ), + encoding="utf-8", + ) + self.assertEqual(load_execution_policy(path)["default"]["max_autonomy"], "review_only") + + def test_load_execution_policy_fails_closed_when_path_unset(self) -> None: + with patch.dict(os.environ, {}, clear=True): + self.assertEqual(load_execution_policy()["default"]["max_autonomy"], "manual") + + def test_load_execution_policy_fails_closed_for_relative_env_path(self) -> None: + env = { + "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH": "policy.json", + "CODEX_AUDIT_SERVICE_EXECUTION_POLICY_OWNER": f"{os.getuid()}:{os.getgid()}", + } + with patch.dict(os.environ, env, clear=True): + self.assertEqual(load_execution_policy()["default"]["max_autonomy"], "manual") + + def test_load_execution_policy_fails_closed_for_untrusted_env_path(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "policy.json" + path.write_text( + json.dumps( + { + "default": { + "max_autonomy": "auto_pr", + "max_consecutive_failures": 3, + "low_cost_model": "gpt-5.4-mini", + "low_cost_provider": "openai", + }, + "repositories": {}, + } + ), + encoding="utf-8", + ) + with patch.dict(os.environ, {"CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH": str(path)}, clear=False): + self.assertEqual(load_execution_policy()["default"]["max_autonomy"], "manual") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_automation_run_ledger.py b/tests/test_automation_run_ledger.py index d560fd9f..ebd02ac6 100644 --- a/tests/test_automation_run_ledger.py +++ b/tests/test_automation_run_ledger.py @@ -145,6 +145,20 @@ def test_snapshot_summarizes_terminal_and_active_runs(self) -> None: self.assertEqual(snapshot["summary"]["suggested_actions"][CONTROL_CONTINUE], 2) self.assertNotIn("events", snapshot["runs"][0]) + def test_snapshot_none_limit_returns_all_retained_runs(self) -> None: + with patch("service.automation_run_ledger.time.time", return_value=1.0): + self.ledger.record("run-1", "running") + self.ledger.record("run-2", "running") + self.ledger.record("run-3", "running") + self.ledger.record("run-1", "merged") + + snapshot = self.ledger.snapshot(limit=None) + + self.assertEqual(snapshot["summary"]["total_runs"], 3) + self.assertEqual(snapshot["summary"]["returned_runs"], 3) + self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-1", "run-2", "run-3"}) + self.assertEqual(snapshot["runs"][0]["run_id"], "run-1") + def test_snapshot_can_include_bounded_history(self) -> None: ledger = AutomationRunLedger(max_events_per_run=2) ledger.record("run-1", "queued") @@ -158,12 +172,15 @@ def test_snapshot_can_include_bounded_history(self) -> None: def test_ledger_evicts_old_runs_by_count(self) -> None: ledger = AutomationRunLedger(max_runs=2) - ledger.record("run-1", "queued") - ledger.record("run-2", "queued") - ledger.record("run-3", "queued") + ledger.record("run-1", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger.record("run-2", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger.record("run-3", "queued", metadata={"source_repository": "QuantStrategyLab/RepoB"}) snapshot = ledger.snapshot(limit=None) self.assertEqual(snapshot["summary"]["total_runs"], 2) + self.assertEqual(snapshot["summary"]["retention"]["evicted_runs"], 1) + self.assertEqual(snapshot["summary"]["retention"]["evicted_runs_by_repo"], {"quantstrategylab/repoa": 1}) + self.assertTrue(snapshot["summary"]["retention"]["may_be_truncated"]) self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-2", "run-3"}) def test_ledger_eviction_keeps_new_run_when_timestamps_match(self) -> None: @@ -175,6 +192,137 @@ def test_ledger_eviction_keeps_new_run_when_timestamps_match(self) -> None: snapshot = ledger.snapshot(limit=None) self.assertEqual([run["run_id"] for run in snapshot["runs"]], ["run-2"]) + def test_persisted_ledger_eviction_count_is_not_double_counted(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "automation_runs.json" + ledger = AutomationRunLedger(max_runs=2, storage_path=path) + ledger.record("run-1", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger.record("run-2", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger.record("run-3", "queued", metadata={"source_repository": "QuantStrategyLab/RepoB"}) + + reloaded = AutomationRunLedger(max_runs=2, storage_path=path) + snapshot = reloaded.snapshot(limit=None) + + self.assertEqual(snapshot["summary"]["retention"]["evicted_runs"], 1) + self.assertEqual(snapshot["summary"]["retention"]["evicted_runs_by_repo"], {"quantstrategylab/repoa": 1}) + self.assertTrue(snapshot["summary"]["retention"]["may_be_truncated"]) + self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-2", "run-3"}) + + def test_persist_merge_does_not_recount_disk_evicted_local_runs(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "automation_runs.json" + ledger_a = AutomationRunLedger(max_runs=2, storage_path=path) + with patch("service.automation_run_ledger.time.time", side_effect=[1.0, 2.0, 3.0]): + ledger_a.record("run-1", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger_a.record("run-2", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger_b = AutomationRunLedger(max_runs=2, storage_path=path) + ledger_a.record("run-3", "queued", metadata={"source_repository": "QuantStrategyLab/RepoB"}) + with patch("service.automation_run_ledger.time.time", return_value=4.0): + ledger_b.record("run-4", "queued", metadata={"source_repository": "QuantStrategyLab/RepoB"}) + snapshot = AutomationRunLedger(max_runs=2, storage_path=path).snapshot(limit=None) + + self.assertEqual(snapshot["summary"]["retention"]["evicted_runs"], 2) + self.assertEqual(snapshot["summary"]["retention"]["evicted_runs_by_repo"], {"quantstrategylab/repoa": 2}) + self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-3", "run-4"}) + + def test_pre_migration_ledger_marks_history_completeness_unknown(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "automation_runs.json" + path.write_text( + json.dumps( + { + "schema_version": "automation_run_ledger.v1", + "sequence": 1, + "runs": { + "run-1": { + "run_id": "run-1", + "task_state": "failed", + "updated_at": 1.0, + "metadata": {"source_repository": "QuantStrategyLab/RepoA"}, + } + }, + } + ), + encoding="utf-8", + ) + + ledger = AutomationRunLedger(max_runs=2, storage_path=path) + snapshot = ledger.snapshot(limit=None) + + self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"]) + + def test_pre_migration_ledger_keeps_history_unknown_after_schema_rewrite(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "automation_runs.json" + path.write_text( + json.dumps( + { + "schema_version": "automation_run_ledger.v1", + "sequence": 1, + "runs": { + "run-1": { + "run_id": "run-1", + "task_state": "failed", + "updated_at": 1.0, + "metadata": {"source_repository": "QuantStrategyLab/RepoA"}, + } + }, + } + ), + encoding="utf-8", + ) + + ledger = AutomationRunLedger(max_runs=3, storage_path=path) + self.assertTrue(ledger.snapshot(limit=None)["summary"]["retention"]["history_completeness_unknown"]) + ledger.record("run-2", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + snapshot = AutomationRunLedger(max_runs=3, storage_path=path).snapshot(limit=None) + + self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"]) + self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-1", "run-2"}) + + def test_stale_update_cannot_resurrect_run_evicted_on_disk(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "automation_runs.json" + ledger_a = AutomationRunLedger(max_runs=2, storage_path=path) + with patch("service.automation_run_ledger.time.time", side_effect=[1.0, 2.0]): + ledger_a.record("run-1", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger_a.record("run-2", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + ledger_b = AutomationRunLedger(max_runs=3, storage_path=path) + with patch("service.automation_run_ledger.time.time", return_value=3.0): + ledger_a.record("run-3", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + with patch("service.automation_run_ledger.time.time", return_value=4.0): + with self.assertRaises(ValueError): + ledger_b.record("run-1", "failed", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + snapshot = AutomationRunLedger(max_runs=2, storage_path=path).snapshot(limit=None) + + self.assertEqual({run["run_id"] for run in snapshot["runs"]}, {"run-2", "run-3"}) + + def test_fresh_missing_persisted_ledger_starts_as_complete_empty_history(self) -> None: + with TemporaryDirectory() as tmp: + ledger = AutomationRunLedger(max_runs=2, storage_path=Path(tmp) / "missing.json") + snapshot = ledger.snapshot(limit=None) + + self.assertFalse(snapshot["summary"]["retention"]["history_completeness_unknown"]) + + def test_disappeared_persisted_ledger_marks_history_completeness_unknown(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "automation_runs.json" + ledger = AutomationRunLedger(max_runs=2, storage_path=path) + ledger.record("run-1", "queued", metadata={"source_repository": "QuantStrategyLab/RepoA"}) + path.unlink() + snapshot = ledger.snapshot(limit=None) + + self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"]) + + def test_corrupt_persisted_ledger_marks_history_completeness_unknown(self) -> None: + with TemporaryDirectory() as tmp: + path = Path(tmp) / "automation_runs.json" + path.write_text("{not-json", encoding="utf-8") + ledger = AutomationRunLedger(max_runs=2, storage_path=path) + snapshot = ledger.snapshot(limit=None) + + self.assertTrue(snapshot["summary"]["retention"]["history_completeness_unknown"]) + def test_update_preserves_control_fields_when_omitted(self) -> None: self.ledger.record( "run-1", diff --git a/tests/test_run_monthly_codex_audit.py b/tests/test_run_monthly_codex_audit.py index da89e57e..faa4387f 100644 --- a/tests/test_run_monthly_codex_audit.py +++ b/tests/test_run_monthly_codex_audit.py @@ -2342,6 +2342,14 @@ def test_vps_deploy_adds_nginx_audit_route_without_router_service(self) -> None: self.assertIn("location = /v1/codex-audit", deploy_script) self.assertIn("location ^~ /v1/codex-audit/", deploy_script) self.assertIn("CODEX_AUDIT_SERVICE_JOB_DIR", deploy_script) + self.assertIn("CODEX_AUDIT_SERVICE_EXECUTION_POLICY_PATH", deploy_script) + self.assertIn("/etc/codex-audit-bridge-policy/execution_policy.json", deploy_script) + self.assertIn("write_default_execution_policy_if_missing", deploy_script) + self.assertIn("O_NOFOLLOW", deploy_script) + self.assertIn("os.O_DIRECTORY", deploy_script) + self.assertIn("dir_fd=fd", deploy_script) + self.assertIn("os.open(component, flags_dir, dir_fd=fd)", deploy_script) + self.assertIn('"max_consecutive_failures": 3', deploy_script) self.assertIn("codex_pr_review.yml@refs/pull/*/merge", deploy_script) self.assertIn("refs/pull/*/merge", deploy_script) self.assertIn("QuantStrategyLab/AIAuditBridge,QuantStrategyLab/CryptoLivePoolPipelines", deploy_script)