Add automation authority and ledger endpoints

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

README.md

@@ -33,6 +33,10 @@ When a source issue contains a `codex-pr-feedback` marker from a failed CI run o
 
 This avoids hard-coding Codex CLI setup in every source repository and avoids depending on a repository outside the `QuantStrategyLab` organization.
 
+Automation authority is intentionally split from model confidence. The service treats routine low-risk maintenance as auto-merge eligible, allows already-live same-family strategy optimization to auto-merge only when service-owned checks provide trusted proof, and keeps new/reconstructed strategies, live-candidate promotion, plugin position-control changes, broker/order execution changes, workflow permissions, and secrets behind human review. `POST /v1/ai/automation/authority` accepts request metadata only as restrictive hints; it cannot relax the policy for live-impacting automation.
+
+Automation run state is persisted through the service ledger. Async Codex jobs are recorded automatically; external workflows can also call `POST /v1/ai/automation/runs`. Ledger ownership follows `source_repository` when present, otherwise the authenticated repository. Operators and dashboards can read `GET /v1/ai/automation/runs`, `GET /v1/ai/automation/runs/{run_id}`, and `GET /v1/ai/automation/control` to decide whether to continue, pause auto-fix, switch to review-only, or escalate.
+
 ## Compatibility governance role
 
 `QuantStrategyLab/AIAuditBridge` is an ops/control-plane consumer only:

README.zh-CN.md

@@ -33,6 +33,10 @@ Codex 执行现在只走 service backend：workflow 从 GitHub-hosted runner 调
 
 这样可以避免每个 source repository 都硬编码 Codex CLI，也不会依赖 `QuantStrategyLab` 组织外的仓库。
 
+自动化授权层和模型置信度分开处理。service 将低风险日常维护视为可自动合并；已 live 的同族策略优化只有在 service-owned 检查给出 trusted proof 时才允许自动合并；新策略/重构策略、live-candidate 提升、插件 position-control、券商/订单执行、workflow 权限和 secrets 相关变更都保留人工复核。`POST /v1/ai/automation/authority` 接收的请求 metadata 只能作为降权提示，不能放宽 live 影响自动化 policy。
+
+自动化运行状态通过 service ledger 持久化。异步 Codex job 会自动记录；外部 workflow 也可调用 `POST /v1/ai/automation/runs` 写入状态。Ledger owner 优先使用 `source_repository`，没有时回退到认证仓库。Dashboard 或运维脚本可读取 `GET /v1/ai/automation/runs`、`GET /v1/ai/automation/runs/{run_id}` 和 `GET /v1/ai/automation/control`，判断继续执行、暂停 auto-fix、切到 review-only 或升级人工处理。
+
 ## 兼容性治理定位
 
 `QuantStrategyLab/AIAuditBridge` 只作为 ops/control-plane 的消费侧参与兼容治理：

docs/architecture.md

@@ -43,6 +43,8 @@ operations: **Analyze**, **Execute**, and **Review**.
 | `health.py` | Online endpoint metrics, error rates, latency tracking |
 | `quota.py` | Per-repo budget enforcement, model cost estimation |
 | `autonomy.py` | Confidence scoring, change risk classification |
+| `automation_authority.py` | Machine-readable live-impact authority policy |
+| `automation_run_ledger.py` | Persistent automation run state and runtime control suggestions |
 | `feedback.py` | Closed-loop change tracking and evaluation |
 
 ### Adapters (`service/adapters/`)
@@ -70,6 +72,10 @@ operations: **Analyze**, **Execute**, and **Review**.
 | `/v1/ai/health` | GET | Online service health snapshot | No |
 | `/healthz` | GET | Liveness check | No |
 | `/v1/ai/quota` | GET | Provider usage, Codex account limits, and internal estimates | No |
+| `/v1/ai/automation/authority` | POST | Evaluate auto/human-review authority for live-impacting changes | No |
+| `/v1/ai/automation/control` | GET | Convert health/quota/org-health into continue/review-only/pause/escalate | No |
+| `/v1/ai/automation/runs` | GET/POST | Read or record automation run ledger entries | No |
+| `/v1/ai/automation/runs/{id}` | GET | Read one automation run ledger entry | No |
 | `/v1/ai/feedback/*` | GET/POST | Change tracking and evaluation | No |
 
 Health status vocabulary is defined in [`health_taxonomy.md`](health_taxonomy.md). Do not use `/v1/ai/health` as a substitute for monthly audit results, artifact freshness, or strategy-level health evidence.
@@ -114,6 +120,7 @@ POST /v1/ai/review
   → optional Codex verification
   → extract confidence scores
   → compute consensus + recommended action
+  → cap action through automation authority policy
   → return {results, consensus, recommended_action}
 ```
 
@@ -124,6 +131,13 @@ POST /v1/ai/review
   requirement.
 - **Authorization**: Source repository org must match OIDC claims repository org
   (prevents cross-org escalation).
+- **Automation authority**: Request metadata can only make a decision stricter.
+  Live-equivalent auto-merge requires service-owned trusted proof; public
+  endpoints cannot self-declare it.
+- **Automation ledger tenancy**: Ledger reads and writes use `source_repository`
+  as owner when present, otherwise the authenticated repository.
+- **Automation control state**: Run task state may be reported by callers, but
+  control action and health/quota/org-health fields are derived by the service.
 - **Sandbox**: Codex sandbox restricted to service-side allowlist (default: `read-only`).
 - **Codex reasoning effort**: `CODEX_AUDIT_SERVICE_REASONING_EFFORT` can hard
   override CLI effort; unset/`auto` routes low/medium/high by task complexity.

service/ai_gateway_service.py

@@ -26,6 +26,7 @@
 import tempfile
 import threading
 import time
+from collections import Counter
 from http import HTTPStatus
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
@@ -44,9 +45,15 @@
 from service.adapters.llm_adapter import LlmAdapter
 from service.adapters.codex_adapter import CodexAdapter
 from service.autonomy import (
+    ACTION_AUTO_PR,
     load_autonomy_policy,
     recommended_action as compute_recommended_action,
 )
+from service.automation_authority import evaluate_automation_authority
+from service.automation_run_ledger import (
+    get_automation_run_ledger,
+    suggest_control_action,
+)
 from service.feedback import (
     write_change,
     read_change,
@@ -59,7 +66,7 @@
     _new_change_id,
 )
 from service.quota import get_quota_manager
-from service.task_state import job_task_state
+from service.task_state import TERMINAL_STATES, job_task_state
 from service.health import get_health_monitor
 from service.org_health import read_org_health
 
@@ -394,6 +401,7 @@ def _mark_stale_job_failed(job: dict[str, Any]) -> dict[str, Any]:
     job["updated_at"] = _now()
     job["error"] = "codex audit job became stale before completion"
     _write_job(job)
+    _record_job_automation_run(job)
     return job
 
 
@@ -425,6 +433,105 @@ def _public_job_payload(job: dict[str, Any]) -> dict[str, object]:
     return payload
 
 
+def _automation_control_snapshot(repo: str) -> dict[str, Any]:
+    try:
+        org_health = read_org_health()
+    except Exception:
+        org_health = {"status": "unavailable"}
+    try:
+        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)
+
+
+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)
+        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 ""),
+            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={
+                "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 ""),
+            },
+        )
+    except Exception as exc:
+        _audit_log("automation_ledger_record_failed", job_id=job.get("job_id"), error=type(exc).__name__)
+
+
+def _automation_operator_claims(claims: dict[str, Any]) -> bool:
+    return str(claims.get("auth_method") or "") in {"static_token", "none"}
+
+
+def _automation_run_owner_repository(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 _automation_run_access_allowed(
+    run: dict[str, Any],
+    claims: dict[str, Any],
+) -> bool:
+    if _automation_operator_claims(claims):
+        return True
+    owner = _automation_run_owner_repository(run)
+    caller_repository = str(claims.get("repository") or "")
+    if bool(owner) and owner == caller_repository:
+        return True
+    metadata = run.get("metadata") if isinstance(run.get("metadata"), dict) else {}
+    original_caller = str(metadata.get("caller_repository") or "")
+    return bool(original_caller) and original_caller == caller_repository
+
+
+def _assert_automation_run_access(
+    run: dict[str, Any],
+    claims: dict[str, Any],
+) -> None:
+    if not _automation_run_access_allowed(run, claims):
+        raise PermissionError("automation run repository is not allowed")
+
+
+def _automation_snapshot_for_claims(
+    snapshot: dict[str, Any],
+    claims: dict[str, Any],
+    *,
+    limit: int | None = None,
+) -> dict[str, Any]:
+    if _automation_operator_claims(claims):
+        return snapshot
+    visible_runs = [run for run in snapshot.get("runs", []) if _automation_run_access_allowed(run, claims)]
+    runs = visible_runs[: max(0, int(limit))] if limit is not None else visible_runs
+    task_states = Counter(str(run.get("task_state", "")).strip().lower() for run in visible_runs if run.get("task_state"))
+    suggested_actions = Counter(
+        str(run.get("suggested_action", "")).strip().lower()
+        for run in visible_runs
+        if run.get("suggested_action")
+    )
+    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 {})
+    summary.update(
+        {
+            "total_runs": len(visible_runs),
+            "returned_runs": len(runs),
+            "active_runs": len(visible_runs) - terminal_runs,
+            "terminal_runs": terminal_runs,
+            "task_states": dict(task_states),
+            "suggested_actions": dict(suggested_actions),
+        }
+    )
+    return {"runs": runs, "summary": summary}
+
+
 def _job_dedupe_key(payload: dict[str, Any]) -> str:
     issue_number = str(payload.get("issue_number") or "").strip()
     prompt_hash = hashlib.sha256(str(payload.get("prompt") or "").encode("utf-8")).hexdigest()
@@ -495,6 +602,7 @@ def _run_job(job_id: str, payload: dict[str, Any]) -> None:
         job["status"] = "running"
         job["updated_at"] = _now()
         _write_job(job)
+        _record_job_automation_run(job)
         adapter = CodexAdapter()
         sandbox = _validate_sandbox(str(payload.get("sandbox") or ""))
         reasoning_effort = _resolve_codex_reasoning_effort(payload, str(payload.get("task") or TASK_EXECUTE))
@@ -516,6 +624,7 @@ def _run_job(job_id: str, payload: dict[str, Any]) -> None:
             job["failure_category"] = _classify_failure(result.error)
         job["updated_at"] = _now()
         _write_job(job)
+        _record_job_automation_run(job)
         get_health_monitor().record(
             "/v1/ai/execute/jobs/run",
             time.time() - started,
@@ -534,6 +643,7 @@ def _run_job(job_id: str, payload: dict[str, Any]) -> None:
         job["error"] = str(exc)[-4000:]
         job["failure_category"] = _classify_failure(str(exc))
         _write_job(job)
+        _record_job_automation_run(job)
         get_health_monitor().record("/v1/ai/execute/jobs/run", time.time() - started, False, type(exc).__name__)
         _audit_log("job_failed", job_id=job_id, error=type(exc).__name__,
                    repository=job.get("repository"))
@@ -575,6 +685,7 @@ def _submit_job(claims: dict[str, Any], payload: dict[str, Any]) -> dict[str, ob
         "dedupe_key": dedupe_key,
     }
     _write_job(job)
+    _record_job_automation_run(job)
     _audit_log("job_submitted", job_id=job_id, repository=job["repository"],
                task=job["task"], source_repository=job["source_repository"])
     thread = threading.Thread(target=_run_job, args=(job_id, payload), name=f"ai-gateway-job-{job_id}", daemon=True)
@@ -626,6 +737,16 @@ def do_GET(self) -> None:
         if request_path == "/v1/ai/quota":
             self._handle_quota_status()
             return
+        if request_path == "/v1/ai/automation/control":
+            self._handle_automation_control()
+            return
+        if request_path == "/v1/ai/automation/runs":
+            self._handle_list_automation_runs()
+            return
+        if request_path.startswith("/v1/ai/automation/runs/"):
+            run_id = request_path[len("/v1/ai/automation/runs/"):]
+            self._handle_get_automation_run(run_id)
+            return
 
         # Feedback: list changes or get effectiveness report
         if request_path == "/v1/ai/changes/effectiveness":
@@ -698,6 +819,12 @@ def do_POST(self) -> None:
             elif self.path in {"/v1/ai/execute/jobs", "/v1/codex-audit/jobs"}:
                 _assert_write_authz(claims, self.path)
                 self._handle_execute_async(claims, payload)
+            elif self.path in {"/v1/ai/automation/runs"}:
+                _assert_write_authz(claims, self.path)
+                self._handle_record_automation_run(claims, payload)
+            elif self.path in {"/v1/ai/automation/authority"}:
+                _assert_write_authz(claims, self.path)
+                self._handle_automation_authority(claims, payload)
             elif self.path in {"/v1/ai/execute", "/v1/codex-audit"}:
                 _assert_write_authz(claims, self.path)
                 self._handle_execute_sync(claims, payload)
@@ -913,6 +1040,7 @@ def _handle_review(self, payload: dict[str, Any]) -> None:
             changed_paths,
             repo=repo if repo else None,
             policy=load_autonomy_policy(),
+            automation_metadata=payload.get("automation_metadata") if isinstance(payload.get("automation_metadata"), dict) else {},
             health_status=get_health_monitor().status,
             quota_status=str(quota_status),
         )
@@ -926,6 +1054,94 @@ def _handle_review(self, payload: dict[str, Any]) -> None:
             "recommended_action": action,
         })
 
+    # -- automation control handlers --
+
+    def _handle_automation_control(self) -> None:
+        from urllib.parse import parse_qs, urlparse
+        try:
+            authenticate(self.headers, audience=DEFAULT_AUDIENCE)
+        except PermissionError as exc:
+            _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)})
+            return
+        parsed = urlparse(self.path)
+        params = parse_qs(parsed.query)
+        repo = str(params.get("repo", ["unknown"])[0] or "unknown")
+        _json_response(self, HTTPStatus.OK, {"status": "ok", "control": _automation_control_snapshot(repo)})
+
+    def _handle_list_automation_runs(self) -> None:
+        from urllib.parse import parse_qs, urlparse
+        try:
+            claims = authenticate(self.headers, audience=DEFAULT_AUDIENCE)
+        except PermissionError as exc:
+            _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)})
+            return
+        parsed = urlparse(self.path)
+        params = parse_qs(parsed.query)
+        limit = int(params.get("limit", ["100"])[0])
+        include_events = str(params.get("include_events", ["false"])[0]).lower() in {"1", "true", "yes", "on"}
+        snapshot_limit = limit if _automation_operator_claims(claims) else None
+        snapshot = get_automation_run_ledger().snapshot(limit=snapshot_limit, include_events=include_events)
+        snapshot = _automation_snapshot_for_claims(snapshot, claims, limit=limit)
+        _json_response(self, HTTPStatus.OK, {"status": "ok", "ledger": snapshot})
+
+    def _handle_get_automation_run(self, run_id: str) -> None:
+        try:
+            claims = authenticate(self.headers, audience=DEFAULT_AUDIENCE)
+        except PermissionError as exc:
+            _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)})
+            return
+        record = get_automation_run_ledger().get(run_id)
+        if record is None:
+            _json_response(self, HTTPStatus.NOT_FOUND, {"status": "error", "error": "automation run not found"})
+            return
+        try:
+            _assert_automation_run_access(record, claims)
+        except PermissionError as exc:
+            _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)})
+            return
+        _json_response(self, HTTPStatus.OK, {"status": "ok", "run": record})
+
+    def _handle_record_automation_run(self, claims: dict[str, Any], payload: dict[str, Any]) -> None:
+        source_repo = str(payload.get("source_repository") or "")
+        if source_repo:
+            _validate_source_repo(source_repo)
+            _validate_source_repo_org(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 {}
+        ledger = get_automation_run_ledger()
+        run_id = str(payload.get("run_id") or payload.get("job_id") or "")
+        existing = ledger.get(run_id)
+        if existing is not None:
+            _assert_automation_run_access(existing, claims)
+        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 ""),
+            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,
+                "repository": repo,
+                "source_repository": source_repo,
+                "caller_repository": str(claims.get("repository") or ""),
+            },
+        )
+        _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:
+        source_repo = str(payload.get("source_repository") or "")
+        if source_repo:
+            _validate_source_repo(source_repo)
+            _validate_source_repo_org(claims, source_repo)
+        changed_paths = [str(path) for path in payload.get("changed_paths", []) if isinstance(path, str)]
+        metadata = payload.get("automation_metadata") if isinstance(payload.get("automation_metadata"), dict) else {}
+        proposed_action = str(payload.get("proposed_action") or ACTION_AUTO_PR)
+        authority = evaluate_automation_authority(changed_paths, metadata=metadata, proposed_action=proposed_action)
+        _json_response(self, HTTPStatus.OK, {"status": "ok", "automation_authority": authority})
+
     # -- feedback handlers (Phase 3: closed-loop change tracking) --
 
     def _handle_feedback_register(self, claims: dict[str, Any], payload: dict[str, Any]) -> None: