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
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/<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)


@bp.route("/agents/<agent_uid>/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/<agent_uid>/runtime/invocations/<idempotency_key>/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/<agent_uid>/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