group_fqcd_jr/tools/probe_release_state.py

"""只读探查：当前 active 配置发布与配置项数量（用于同步 docs/24 的"当前状态"表）。

只做 SELECT，结果写 `docs/evidence/release-state.json`：

    python tools/probe_release_state.py
"""

from __future__ import annotations

import asyncio
import json
from pathlib import Path
from typing import Any

from sqlalchemy import text

from app.infrastructure.db import SessionFactory

OUTPUT = Path("docs/evidence/release-state.json")


async def collect() -> dict[str, Any]:
    report: dict[str, Any] = {}
    async with SessionFactory() as session:
        report["recent_releases"] = [
            dict(row)
            for row in (
                (
                    await session.execute(
                        text(
                            """
                            SELECT id, release_no, title, status, created_at, activated_at
                            FROM config_release
                            ORDER BY id DESC
                            LIMIT 5
                            """
                        )
                    )
                )
                .mappings()
                .all()
            )
        ]
        report["active_items"] = [
            dict(row)
            for row in (
                (
                    await session.execute(
                        text(
                            """
                            SELECT i.namespace, COUNT(*) AS items
                            FROM platform_config_item i
                            JOIN config_release r ON r.id = i.release_id
                            WHERE r.status = 'active'
                            GROUP BY i.namespace
                            ORDER BY i.namespace
                            """
                        )
                    )
                )
                .mappings()
                .all()
            )
        ]
        report["active_prompt_versions"] = (
            await session.execute(
                text(
                    """
                    SELECT COUNT(*) FROM prompt_template_version p
                    JOIN config_release r ON r.id = p.release_id
                    WHERE r.status = 'active'
                    """
                )
            )
        ).scalar_one()
        # 工具白名单的**具体内容**：它必须与代码里的 allowed_tools 交集非空，
        # 否则 ToolExecutor 失败关闭，Agent 任何工具调用都被拒。
        report["active_agent_tools"] = [
            dict(row)
            for row in (
                (
                    await session.execute(
                        text(
                            """
                            SELECT i.config_key, i.value_json, i.schema_version
                            FROM platform_config_item i
                            JOIN config_release r ON r.id = i.release_id
                            WHERE r.status = 'active' AND i.namespace = 'agent_tools'
                            ORDER BY i.config_key
                            """
                        )
                    )
                )
                .mappings()
                .all()
            )
        ]
    return report


async def main() -> None:
    report = await collect()
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(
        json.dumps(report, ensure_ascii=False, indent=2, default=str),
        encoding="utf-8",
    )
    print(f"wrote {OUTPUT}")


if __name__ == "__main__":
    asyncio.run(main())