"""Governed Agent registry, authorization decisions and replay APIs.""" from __future__ import annotations import hashlib import uuid from flask import current_app, g, jsonify, request from app import db from app.api.knowledge_base import bp from app.config.config import is_placeholder_env_value from app.core.llm.agent_governance import AgentGovernanceService from app.core.llm.agent_governance_repository import ( SqlAlchemyAgentGovernanceRepository, WorkCenterAgentApprovalGateway, ) from app.core.llm.runtime_governance_repository import ( SqlAlchemyRuntimeGovernanceRepository, ) from app.core.llm.runtime_server import ( RuntimeServerContext, RuntimeServerError, ServerGovernedInvocationService, ) from app.core.system.permissions import ( AGENTS_MANAGE, AGENTS_OPERATE, AGENTS_READ, require_permissions, ) from app.models.result import failed, success class AgentGovernanceUnavailable(RuntimeError): """Raised when production lacks a dedicated credential signing secret.""" def _effective_credential_secret() -> tuple[str, bool]: dedicated = str(current_app.config.get("AGENT_CREDENTIAL_SECRET") or "").strip() if len(dedicated.encode()) >= 32 and not is_placeholder_env_value(dedicated): return dedicated, True fallback = hashlib.sha256( ("dataops-wp09-local-agent:" + str(current_app.config.get("SECRET_KEY") or "")).encode() ).hexdigest() return fallback, False def _require_credential_secret() -> None: _secret, dedicated_ready = _effective_credential_secret() if ( not dedicated_ready and not current_app.config.get("TESTING") and str(current_app.config.get("FLASK_ENV") or "").lower() == "production" ): raise AgentGovernanceUnavailable( "dedicated Agent credential secret is required in production" ) def _service(): secret, _dedicated_ready = _effective_credential_secret() return AgentGovernanceService( SqlAlchemyAgentGovernanceRepository(db.session), approval_gateway=WorkCenterAgentApprovalGateway(db.session), credential_secret=secret, commit=db.session.commit, rollback=db.session.rollback, ) def _runtime_service(): return ServerGovernedInvocationService(SqlAlchemyRuntimeGovernanceRepository(db.session)) def _expected_version() -> int: raw = str(request.headers.get("If-Match") or "").strip() if raw.startswith("W/"): raw = raw[2:].strip() raw = raw.strip('"') if not raw.isdigit(): raise ValueError("missing valid If-Match version") return int(raw) def _etag(response, version): response.headers["ETag"] = f'"{int(version)}"' return response def _no_store(response): response.headers["Cache-Control"] = "no-store" return response def _error(exc): db.session.rollback() if "If-Match" in str(exc): status = 428 elif isinstance(exc, LookupError): status = 404 elif isinstance(exc, (PermissionError, AgentGovernanceUnavailable)): status = 403 if isinstance(exc, PermissionError) else 503 elif isinstance(exc, RuntimeError): status = 409 else: status = 400 return jsonify(failed(str(exc), code=status)), status @bp.route("/agents", methods=["GET"]) @require_permissions(AGENTS_READ) def list_governed_agents(): filters = { key: request.args.get(key) for key in ("status", "autonomy_level", "owner_uid") if request.args.get(key) } return jsonify(success(_service().list_agents(**filters))) @bp.route("/agents", methods=["POST"]) @require_permissions(AGENTS_MANAGE) def register_governed_agent(): try: result = _service().register_agent( request.get_json(silent=True) or {}, actor_uid=g.current_user["id"] ) return _etag(jsonify(success(result, "Agent 治理登记已创建", code=201)), 1), 201 except (ValueError, LookupError, RuntimeError) as exc: return _error(exc) @bp.route("/agents/", methods=["GET"]) @require_permissions(AGENTS_READ) def get_governed_agent(agent_uid): try: result = _service().agent_detail(agent_uid) return _etag(jsonify(success(result)), result["current_version"]) except (ValueError, LookupError) as exc: return _error(exc) @bp.route("/agents//revisions", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def revise_governed_agent(agent_uid): try: result = _service().revise_agent( agent_uid, request.get_json(silent=True) or {}, expected_version=_expected_version(), actor_uid=g.current_user["id"], ) return _etag(jsonify(success(result, "Agent 版本草稿已创建")), result["current_version"]) except (ValueError, LookupError, PermissionError, RuntimeError) as exc: return _error(exc) @bp.route("/agents//transition", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def transition_governed_agent(agent_uid): try: result = _service().transition_agent( agent_uid, request.get_json(silent=True) or {}, expected_version=_expected_version(), actor_uid=g.current_user["id"], ) return _etag(jsonify(success(result, "Agent 生命周期已更新")), result["current_version"]) except (ValueError, LookupError, PermissionError, RuntimeError) as exc: return _error(exc) @bp.route("/agents//grants", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def create_agent_tool_grant(agent_uid): try: result = _service().create_tool_grant( agent_uid, request.get_json(silent=True) or {}, actor_uid=g.current_user["id"] ) return jsonify(success(result, "Agent 工具授权已创建", code=201)), 201 except (ValueError, LookupError, PermissionError, RuntimeError) as exc: return _error(exc) @bp.route("/agents//grants//revoke", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def revoke_agent_tool_grant(agent_uid, grant_uid): try: return jsonify(success(_service().revoke_tool_grant( agent_uid, grant_uid, actor_uid=g.current_user["id"] ), "Agent 工具授权已撤销")) except (ValueError, LookupError, PermissionError, RuntimeError) as exc: return _error(exc) @bp.route("/agents//credentials", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def issue_agent_credential(agent_uid): try: _require_credential_secret() result = _service().issue_credential( agent_uid, request.get_json(silent=True) or {}, actor_uid=g.current_user["id"] ) return _no_store(jsonify(success(result, "凭证仅本次返回,请立即安全保存", code=201))), 201 except (ValueError, LookupError, PermissionError, RuntimeError, AgentGovernanceUnavailable) as exc: return _error(exc) @bp.route("/agents//credentials/revoke", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def revoke_agent_credentials(agent_uid): try: return jsonify(success(_service().revoke_agent_credentials( agent_uid, actor_uid=g.current_user["id"] ), "Agent 有效凭证已全部撤销")) except (ValueError, LookupError, PermissionError, RuntimeError) as exc: return _error(exc) @bp.route("/agents//actions/authorize", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def authorize_agent_action(agent_uid): try: _require_credential_secret() token = str(request.headers.get("X-Agent-Credential") or "") result = _service().authorize_action( agent_uid, token, request.get_json(silent=True) or {} ) return _no_store(_etag(jsonify(success(result, "Agent 策略判定已留痕", code=201)), 1)), 201 except (ValueError, LookupError, PermissionError, RuntimeError, AgentGovernanceUnavailable) as exc: return _error(exc) @bp.route("/agents/actions", methods=["GET"]) @require_permissions(AGENTS_READ) def list_agent_actions(): filters = { key: request.args.get(key) for key in ("agent_uid", "decision", "risk_level") if request.args.get(key) } return jsonify(success(_service().list_requests(**filters))) @bp.route("/agents/actions//reconcile", methods=["POST"]) @require_permissions(AGENTS_MANAGE) def reconcile_agent_action(request_uid): try: result = _service().reconcile_action( request_uid, expected_version=_expected_version(), actor_uid=g.current_user["id"] ) return _etag(jsonify(success(result, "审批结果已同步")), result["current_version"]) except (ValueError, LookupError, PermissionError, RuntimeError) as exc: return _error(exc) @bp.route("/agents/actions//complete", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def complete_agent_action(request_uid): try: result = _service().complete_action( request_uid, request.get_json(silent=True) or {}, expected_version=_expected_version(), actor_uid=g.current_user["id"], ) return _etag(jsonify(success(result, "Agent 执行证据已记录")), result["current_version"]) except (ValueError, LookupError, PermissionError, RuntimeError) as exc: return _error(exc) @bp.route("/agents/actions//replay", methods=["GET"]) @require_permissions(AGENTS_READ) def replay_agent_action(request_uid): try: return _no_store(jsonify(success(_service().replay(request_uid)))) except (ValueError, LookupError) as exc: return _error(exc) @bp.route("/agents/dashboard", methods=["GET"]) @require_permissions(AGENTS_READ) def agent_governance_dashboard(): try: return jsonify(success(_service().dashboard())) except AgentGovernanceUnavailable as exc: return _error(exc) @bp.route("/agents//runtime/invocations", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def authorize_runtime_invocation(agent_uid): """The only HTTP path that binds Agent credentials to a server identity.""" try: token = str(request.headers.get("X-Agent-Credential") or "") if not token: raise PermissionError("Agent credential is required") _service().validate_credential(agent_uid, token) result = _runtime_service().authorize( RuntimeServerContext( agent_uid=agent_uid, actor_uid=g.current_user["id"], roles=frozenset(g.current_user["roles"]), ), request.get_json(silent=True) or {}, ) db.session.commit() return _no_store(jsonify(success(result, code=201))), 201 except (RuntimeServerError, ValueError, LookupError, PermissionError, RuntimeError) as exc: response, status = _error(exc) return _no_store(response), status @bp.route("/agents//runtime/invocations//settle", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def settle_runtime_invocation(agent_uid, idempotency_key): try: token = str(request.headers.get("X-Agent-Credential") or "") if not token: raise PermissionError("Agent credential is required") claims = _service().validate_credential(agent_uid, token) agent = _service().get_agent(agent_uid) if claims["agent_uid"] != agent_uid or agent["owner_uid"] != g.current_user["id"]: raise PermissionError("Agent credential principal does not match") body = request.get_json(silent=True) or {} allowed = {"lease_fence", "outcome", "actual_tokens", "actual_cost_micros", "actual_tools", "actual_time_ms"} if set(body) != allowed: raise ValueError("runtime settlement payload is closed") scope = _runtime_service().repository.settlement_context( agent_uid=agent_uid, actor_uid=g.current_user["id"], idempotency_key=idempotency_key ) result = _runtime_service().repository.settle_claim({ **scope, "worker_id": f"agent-runtime-http:{g.current_user['id']}", **body, }) db.session.commit() if result.get("rejected"): return _no_store(jsonify(failed("runtime_settlement_rejected", code=409))), 409 return _no_store(jsonify(success(result, code=201))), 201 except (RuntimeServerError, ValueError, LookupError, PermissionError, RuntimeError) as exc: response, status = _error(exc) return _no_store(response), status @bp.route("/agents//runtime/control", methods=["POST"]) @require_permissions(AGENTS_OPERATE) def control_runtime(agent_uid): """Execute a recovery/canary change only through an authenticated control claim.""" try: token = str(request.headers.get("X-Agent-Credential") or "") if not token: raise PermissionError("Agent credential is required") claims = _service().validate_credential(agent_uid, token) agent = _service().get_agent(agent_uid) if claims["agent_uid"] != agent_uid or agent["owner_uid"] != g.current_user["id"]: raise PermissionError("Agent credential principal does not match") body = request.get_json(silent=True) or {} forbidden = {"agent_uid", "principal_id"} required = { "tenant_id", "business_domain_uid", "environment", "operation", "approval_task_uid", "idempotency_key", "request_digest", "expected_fence", "incident_uid", "from_state", "to_state", "generation", "from_generation", "to_generation", "dataset_version", "dataset_digest", "metrics_digest", "threshold_policy", "issued_at", } if set(body) != required or set(body) & forbidden: raise ValueError("runtime control payload is closed") credential = _service().repository.get_credential(claims["jti"]) if not credential: raise PermissionError("Agent credential is revoked or unknown") payload = {**body, "agent_uid": agent_uid, "principal_id": g.current_user["id"]} claim_uid = _runtime_service().repository.issue_control_claim({ "claim_uid": str(uuid.uuid4()), "credential_uid": credential["uid"], "credential_token_digest": hashlib.sha256(token.encode()).hexdigest(), "nonce": str(uuid.uuid4()), "control_payload": payload, }) result = _runtime_service().repository.control_claimed(payload, claim_uid) db.session.commit() return _no_store(jsonify(success(result, code=201))), 201 except (RuntimeServerError, ValueError, LookupError, PermissionError, RuntimeError) as exc: response, status = _error(exc) return _no_store(response), status