DataOps-platform/deployment/app/api/knowledge_base/agent_governance_routes.py

"""Governed Agent registry, authorization decisions and replay APIs."""

from __future__ import annotations

import hashlib

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.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 _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/<agent_uid>", 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/<agent_uid>/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/<agent_uid>/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/<agent_uid>/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/<agent_uid>/grants/<grant_uid>/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/<agent_uid>/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/<agent_uid>/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/<agent_uid>/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/<request_uid>/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/<request_uid>/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/<request_uid>/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)