Copilot 编排期看得见数据库了，附带修一个让非流式生成整体失效的 bug

阶段②。之前 agent 是运行时才摸索表结构，每跑一次重探一遍——慢，而且同一张图
两次运行走的 SQL 可能不一样，没法钉口径。现在 Copilot 建图时就知道有哪些库、
哪些表，生成的 SQL 直接能跑。

## 先修一个更要紧的：非流式生成的 config 全是空的

GRAPH_SCHEMA 里 config 写的是 {"type": "object"}，没有 properties 也没有
additionalProperties。不少 provider 的结构化输出实现遇到这种 object 会按
"不允许额外字段"处理，于是模型填好的 config 被整个过滤成 {}——生成的图每个
节点都是空壳，一提交就挂在"工具节点还没选工具"。

config 的键随节点类型千变万化（tool 有 tool/args，code 有 language/code…），
没法枚举成 properties，只能显式声明 additionalProperties=true。流式那条路径
走操作流协议、不经过 structured output，所以这个 bug 一直藏着没露头。

## 注入什么

- 工具清单补上动态的 db_query__<源> / db_schema__<源>。all_specs() 只有静态
  注册的内置工具，不补进来 Copilot 根本不知道这些工具存在，只会退回让用户自己
  写 SQL 的代码节点——接了数据库等于白接。
- 数据源摘要（含 schema 全名），外加四条写 SQL 的硬性要求。这几条都是真实
  踩过的：漏 schema 前缀在 Oracle 上直接 ORA-00942，写惯 MySQL 的模型会顺手
  写 LIMIT，分号拼接会被守卫拒掉。

## token 预算：按对象数自适应，不是按字符数

实测一个 53 对象的库，带字段 2715 token、只给对象名 956 token（35%）。
但真正的决策点不是字符数——detail 模式每库只列 40 个对象，多出来的 Copilot
根本看不见。对 53 个对象的库，"40 张表带字段"不如"53 张表只给名字"：
找不到那张表，字段写得再全也没用。截断丢信息比省 token 更糟。

所以先按对象数判断（超过 40 就整体切紧凑），再用字符预算兜多库叠加的情况。
真实库实测：紧凑模式 1219 token，53 个对象全部可见。

## 实测

完整模式：一句话需求 → Copilot 出图 → 不改一个字直接跑通真实 Oracle。
生成的 SQL 三处细节全对——ANALYTICS. 前缀、FETCH FIRST（不是 LIMIT）、单条语句。

紧凑模式（看不到字段）：Copilot 没有硬猜字段，而是生成了一个带
db_schema + db_query 两个工具的 agent 节点，让运行时先查结构再写 SQL——
正是 prompt 里"字段拿不准就先放一个 db_schema 节点"那条。跑通，结论正确。

测试 72 项。

backend/app/api/copilot.py

@@ -7,6 +7,7 @@ from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel, Field
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db.base import get_session
@@ -19,6 +20,92 @@ from app.tools.registry import all_specs
 router = APIRouter(prefix="/api/copilot", tags=["copilot"])
 
 
+async def _tool_catalog(session: AsyncSession) -> str:
+    """给 Copilot 看的工具清单。
+
+    all_specs() 只有静态注册的内置工具，数据源工具是运行时按库动态生成的
+    （db_query__<源>），不补进来 Copilot 就不知道它们存在，只会退回让用户
+    自己写 SQL 的代码节点——接了数据库等于白接。
+    """
+    lines = [
+        f"- {name}（{spec.category}）：{spec.description}"
+        for name, spec in sorted(all_specs().items())
+    ]
+
+    from app.data import introspect as _introspect
+    from app.db.models import DataSource
+    from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX
+
+    rows = list((
+        await session.execute(
+            select(DataSource).where(DataSource.enabled.is_(True)).order_by(DataSource.name)
+        )
+    ).scalars())
+    for row in rows:
+        tables = _introspect.table_names(row)
+        listed = "、".join(tables[:12]) + (f" 等 {len(tables)} 个对象" if len(tables) > 12 else "")
+        lines.append(
+            f"- {QUERY_PREFIX}{row.name}（数据库）：在「{row.name}」上执行 SQL。"
+            f"{row.description or ''}"
+            + (f" 可查：{listed}" if tables else " ⚠ 结构未探查")
+        )
+        lines.append(
+            f"- {SCHEMA_PREFIX}{row.name}（数据库）：查看「{row.name}」的表结构"
+        )
+    return "\n".join(lines)
+
+
+# 数据源摘要的字符预算。超了就退成"只列对象名"，字段让 Copilot 用
+# db_schema 工具按需查。实测一个 53 对象的库带字段约 4000 字符，
+# 接三五个库就会把真正的需求淹掉——system prompt 里塞满 schema 而挤掉
+# 用户想要什么，是本末倒置。
+_DATASOURCE_BUDGET_CHARS = 6000
+# detail 模式下每个库最多列这么多对象（与 introspect.summary 的默认值一致）。
+# 超过就说明会截断，那还不如切紧凑模式把对象名列全。
+_DETAIL_TABLE_LIMIT = 40
+
+
+async def _datasource_section(session: AsyncSession) -> str:
+    """数据源与结构摘要。没有数据源时返回空串，不占 prompt。"""
+    from app.data import introspect as _introspect
+    from app.db.models import DataSource
+
+    rows = list((
+        await session.execute(
+            select(DataSource).where(DataSource.enabled.is_(True)).order_by(DataSource.name)
+        )
+    ).scalars())
+    if not rows:
+        return ""
+
+    # detail 模式每个库最多列 40 个对象，多出来的直接看不见。对象一多，
+    # "40 张表带字段"反而不如"全部表只给名字"——Copilot 找不到那张表，
+    # 字段写得再全也没用。所以超出截断阈值就整体切到紧凑模式。
+    detail = all(len(_introspect.table_names(row)) <= _DETAIL_TABLE_LIMIT for row in rows)
+    blocks = [_introspect.summary(row, detail=detail) for row in rows]
+    if detail and sum(len(b) for b in blocks) > _DATASOURCE_BUDGET_CHARS:
+        detail = False
+        blocks = [_introspect.summary(row, detail=False) for row in rows]
+    if not detail:
+        # Copilot 仍然知道有哪些对象可查，要字段时在图里放一个 db_schema 节点，
+        # 或者交给运行期的 agent 自己探
+        blocks.append(
+            "  （对象较多，上面只列了名字。写 SQL 前用 db_schema 工具确认字段，别猜）"
+        )
+    return (
+        "\n\n已接入的数据源（涉及取数时优先用它们，而不是让用户自己写 SQL 的代码节点）：\n"
+        + "\n".join(blocks)
+        + "\n\n写 SQL 的硬性要求：\n"
+        # 这三条都是真实踩过的：Oracle 上漏 schema 前缀直接 ORA-00942，
+        # 写惯 MySQL 的模型会顺手写 LIMIT，而一条没有行数限制的查询能拖垮生产库
+        "- 表名照抄上面的全名（含 schema 前缀），漏掉前缀在 Oracle 上会直接报 ORA-00942\n"
+        "- 认准方言：Oracle 用 FETCH FIRST n ROWS ONLY，不是 LIMIT\n"
+        "- 一次只写一条语句；分号拼接会被拒\n"
+        "- 字段拿不准就在图里先放一个 db_schema 工具节点，别猜字段名\n"
+        "- 取数结果要进口径卡（metrics 节点）才能被叙述引用，别让 llm 节点直接对数字做算术\n"
+    )
+
+
 NODE_REFERENCE = """\
 可用节点类型（type 字段）：
 - input：入口。config.fields = [{name, required, default, description}]
@@ -180,9 +267,15 @@ GRAPH_SCHEMA: dict[str, Any] = {
                     "id": {"type": "string"},
                     "type": {"type": "string", "enum": [t.value for t in NodeType]},
                     "label": {"type": "string"},
-                    "config": {"type": "object"},
+                    # additionalProperties 必须显式写 true：config 的键随节点类型
+                    # 千变万化（tool 有 tool/args，code 有 language/code…），没法
+                    # 提前枚举成 properties。而不少 provider 的结构化输出实现在
+                    # 遇到没有 properties 的 object 时按"不允许额外字段"处理，
+                    # 于是模型填好的 config 被整个过滤成 {}——生成的图每个节点都
+                    # 是空壳，校验直接挂在"工具节点还没选工具"。
+                    "config": {"type": "object", "additionalProperties": True},
                 },
-                "required": ["id", "type"],
+                "required": ["id", "type", "config"],
             },
         },
         "edges": {
@@ -211,14 +304,12 @@ async def generate(
     产物会经过和手工编排完全相同的校验与排版，所以生成完就是可运行、可读的，
     而不是一段还要人去修的 JSON。
     """
-    tool_list = "\n".join(
-        f"- {name}（{spec.category}）：{spec.description}"
-        for name, spec in sorted(all_specs().items())
-    )
+    tool_list = await _tool_catalog(session)
+    datasources = await _datasource_section(session)
 
     system = (
         "你是一个 agent 工作流编排专家。根据用户需求产出一张可执行的工作流图。\n\n"
-        f"{NODE_REFERENCE}\n\n可用工具：\n{tool_list}\n\n"
+        f"{NODE_REFERENCE}\n\n可用工具：\n{tool_list}{datasources}\n\n"
         "要求：\n"
         "1. 必须有且只有一个 input 节点和至少一个 output 节点\n"
         "2. 节点 id 用简短英文小写下划线，例如 fetch_data、summarize\n"
@@ -460,13 +551,11 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
     """
     from fastapi.responses import StreamingResponse
 
-    tool_list = "\n".join(
-        f"- {name}（{spec.category}）：{spec.description}"
-        for name, spec in sorted(all_specs().items())
-    )
+    tool_list = await _tool_catalog(session)
+    datasources = await _datasource_section(session)
     system = (
         "你是一个 agent 工作流编排专家。根据用户需求，以操作流的方式逐步搭出一张可执行的工作流图。\n\n"
-        f"{NODE_REFERENCE}\n\n可用工具：\n{tool_list}\n\n"
+        f"{NODE_REFERENCE}\n\n可用工具：\n{tool_list}{datasources}\n\n"
         "结构要求：\n"
         "1. 必须有且只有一个 input 节点和至少一个 output 节点\n"
         "2. 节点 id 用简短英文小写下划线；label 用中文，一眼看懂\n"

backend/app/data/introspect.py

@@ -187,19 +187,42 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any
     return payload
 
 
-def summary(source: Any, *, max_tables: int = 40) -> str:
+def summary(source: Any, *, max_tables: int = 40, detail: bool = True) -> str:
+    """结构摘要。detail=False 只给对象名，不列字段——库多了要靠它压 token。
+
+    实测一个 53 对象的库：带字段 2715 token，只给名字 956 token（35%）。
+    接三五个库的差别就是 15k 和 4.5k，后者才塞得进 system prompt。
+    """
+    if not detail:
+        cache = source.schema_cache or {}
+        names = table_names(source)
+        if not names:
+            return _empty_summary(source, cache)
+        head = (
+            f"- {source.name}（{source.kind}，{'只读' if source.readonly else '可写'}）："
+            f"{source.description or '未填说明'}"
+        )
+        return head + "\n    对象：" + "、".join(names)
+    return _detailed_summary(source, max_tables=max_tables)
+
+
+def _empty_summary(source: Any, cache: dict[str, Any]) -> str:
+    hint = ""
+    candidates = cache.get("available_schemas") or []
+    if candidates:
+        hint = f"｜该账号名下没有对象，数据可能在这些 schema：{'、'.join(candidates[:8])}"
+    return (
+        f"- {source.name}（{source.kind}）：{source.description or '未填说明'}"
+        f"｜尚未探查结构{hint}"
+    )
+
+
+def _detailed_summary(source: Any, *, max_tables: int = 40) -> str:
     """给 Copilot 的紧凑摘要。没探查过就明说，别让它对着空气编表名。"""
     cache = source.schema_cache or {}
     tables = cache.get("tables") or {}
     if not tables:
-        hint = ""
-        candidates = cache.get("available_schemas") or []
-        if candidates:
-            hint = f"｜该账号名下没有对象，数据可能在这些 schema：{'、'.join(candidates[:8])}"
-        return (
-            f"- {source.name}（{source.kind}）：{source.description or '未填说明'}"
-            f"｜尚未探查结构{hint}"
-        )
+        return _empty_summary(source, cache)
 
     schema = cache.get("schema")
     lines = [

backend/tests/test_copilot_context.py

@@ -0,0 +1,105 @@
+"""Copilot 的数据源上下文注入。
+
+这组守的是"Copilot 知道有哪些库、写得出能跑的 SQL"。断了的表现很隐蔽：
+它不会报错，只会退回让用户自己写 SQL 的代码节点——接了数据库等于白接。
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.api.copilot import GRAPH_SCHEMA, _DATASOURCE_BUDGET_CHARS
+from app.data.introspect import summary
+
+
+def _source(n_tables: int, *, name: str = "bi", schema: str | None = "ANALYTICS"):
+    tables = {
+        f"t{i}": {
+            "qualified": f"{schema}.t{i}" if schema else f"t{i}",
+            "schema": schema,
+            "columns": [
+                {"name": f"c{j}", "type": "VARCHAR", "nullable": True, "comment": None}
+                for j in range(20)
+            ],
+            "primary_key": [], "comment": None, "is_view": True,
+        }
+        for i in range(n_tables)
+    }
+    return SimpleNamespace(
+        id="x", name=name, kind="oracle", readonly=True,
+        description="测试库", options={}, schema_cache={"tables": tables, "schema": schema},
+    )
+
+
+def test_graph_schema_allows_arbitrary_config_keys():
+    """config 的键随节点类型千变万化，没法枚举成 properties。
+
+    不显式写 additionalProperties=True，某些 provider 的结构化输出会按
+    "不允许额外字段"处理，把 config 整个过滤成 {}——生成的图每个节点都是
+    空壳，校验挂在"工具节点还没选工具"。这条回归过一次，代价是整条非流式
+    生成路径不可用。
+    """
+    node = GRAPH_SCHEMA["properties"]["nodes"]["items"]
+    assert node["properties"]["config"]["additionalProperties"] is True
+    assert "config" in node["required"]
+
+
+def test_summary_detail_mode_lists_columns():
+    text = summary(_source(3))
+    assert "ANALYTICS.t0" in text and "c0" in text
+
+
+def test_summary_compact_mode_drops_columns():
+    """库多了要靠它压 token：只留对象名，字段让 db_schema 按需查。"""
+    text = summary(_source(50), detail=False)
+    assert "ANALYTICS.t0" in text
+    assert "c0" not in text
+    assert len(text) < len(summary(_source(50)))
+
+
+def test_compact_mode_is_substantially_smaller():
+    """压缩比要真的有效，否则降级没意义。实测约 35%。"""
+    full = summary(_source(53))
+    compact = summary(_source(53), detail=False)
+    assert len(compact) < len(full) * 0.5
+
+
+def test_qualified_names_appear_in_summary():
+    """摘要里必须是带 schema 的全名——模型照抄，漏前缀就是 ORA-00942。"""
+    assert "ANALYTICS.t0" in summary(_source(2))
+
+
+def test_summary_without_schema_uses_bare_names():
+    """没有 schema 概念的库不要硬拼前缀，否则生成的 SQL 反而错。"""
+    text = summary(_source(2, schema=None))
+    assert "t0" in text and "None.t0" not in text
+
+
+def test_budget_constant_is_sane():
+    """预算太小会让单个库就触发降级，太大等于没有预算。"""
+    assert 3000 <= _DATASOURCE_BUDGET_CHARS <= 20000
+
+
+@pytest.mark.parametrize("n_tables,expect_detail", [(3, True), (40, True), (53, False)])
+def test_object_count_decides_detail_level(n_tables, expect_detail):
+    """对象超过截断阈值就切紧凑模式。
+
+    关键取舍：detail 模式每库只列 40 个对象，多出来的 Copilot 根本看不见。
+    对 53 个对象的库，"40 张表带字段"不如"53 张表只给名字"——找不到那张表，
+    字段写得再全也没用。截断丢信息比省 token 更糟。
+    """
+    from app.api.copilot import _DETAIL_TABLE_LIMIT
+    from app.data.introspect import table_names
+
+    src = _source(n_tables)
+    assert (len(table_names(src)) <= _DETAIL_TABLE_LIMIT) is expect_detail
+
+
+def test_many_sources_trigger_compact_mode():
+    """单库不超标，几个库加起来超预算时整体降级。"""
+    sources = [_source(35, name=f"db{i}") for i in range(6)]
+    full = sum(len(summary(s)) for s in sources)
+    assert full > _DATASOURCE_BUDGET_CHARS
+    compact = sum(len(summary(s, detail=False)) for s in sources)
+    assert compact < _DATASOURCE_BUDGET_CHARS