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 the caller provides complete evidence flags, and keeps new/reconstructed strategies, live-candidate promotion, plugin position-control changes, broker/order execution changes, workflow permissions, and secrets behind human review. Use `POST /v1/ai/automation/authority` to evaluate that machine-readable policy before requesting 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`. 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 的同族策略优化只有在调用方提供完整证据标记时才允许自动合并；新策略/重构策略、live-candidate 提升、插件 position-control、券商/订单执行、workflow 权限和 secrets 相关变更都保留人工复核。live 影响自动化前可调用 `POST /v1/ai/automation/authority` 评估这套机器可读 policy。
+
+自动化运行状态通过 service ledger 持久化。异步 Codex job 会自动记录；外部 workflow 也可调用 `POST /v1/ai/automation/runs` 写入状态。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 or promote action through automation authority policy
   → return {results, consensus, recommended_action}
 ```
 

service/ai_gateway_service.py

@@ -44,9 +44,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,
@@ -394,6 +400,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 +432,41 @@ 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": str(job.get("repository") or ""),
+                "source_repository": str(job.get("source_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 _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 +537,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 +559,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 +578,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 +620,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 +672,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 +754,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 +975,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 +989,77 @@ 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:
+            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 = get_automation_run_ledger().snapshot(limit=limit, include_events=include_events)
+        _json_response(self, HTTPStatus.OK, {"status": "ok", "ledger": snapshot})
+
+    def _handle_get_automation_run(self, run_id: str) -> None:
+        try:
+            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
+        _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 {}
+        record = get_automation_run_ledger().record(
+            str(payload.get("run_id") or payload.get("job_id") or ""),
+            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(payload.get("suggested_action") or control.get("action") or ""),
+            service_health=str(payload.get("service_health") or control.get("service_health") or ""),
+            quota_status=payload.get("quota_status") or control.get("quota_status") or "",
+            org_health_status=str(payload.get("org_health_status") or control.get("org_health_status") or ""),
+            metadata={**metadata, "source_repository": source_repo},
+        )
+        _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: