feat(knowledge): 语义检索打通——切片、混合打分、只搜已发布

之前知识只能"存"，不能"找"。这一版把链路补上：切片 → 向量 → 打分 → 带来源返回。

- chunk 表 + 迁移 0006：正文按标题归段切、代码块整块保留、超长按句切、
  相邻片重叠 80 字；RLS 与 GRANT 与其它租户表一致
- 向量提供方做成协议：内置 hash（字符二元组，零依赖可复现，作为基线）
  与 OpenAI 兼容端点两种，换实现不动业务代码
- 检索是混合打分：向量 0.6 + pg_trgm 0.4。专有编号、单号这类必须原样命中的词
  靠词法兜，换个说法表达的同一件事靠向量兜
- 三条纪律：只搜 published（过滤写在 SQL 里）、低于阈值宁可不返回、
  每条命中都带来源标题与 item_id
- 发布时自动建索引，改完正文用 POST /knowledge/{id}/reindex 重建
- Web 知识页加检索框与命中卡片；新增 e2e/retrieval.mjs 6 项

为什么不是 pgvector：本机 Docker 起不来、fnsvr-01 没有该扩展且明确不动。
先用 real[] + SQL 余弦把链路跑通，将来换列类型与索引即可，接口不变。

后端 79 passed，e2e smoke/knowledge/richtext/retrieval 全绿。

backend/app/api/knowledge.py

@@ -33,6 +33,7 @@ from ..models_task import Activity, Item
 from ..schemas import clamp_limit, decode_cursor, encode_cursor, ok
 from ..serializers import item_out, tags_of
 from ..services.knowledge import create_draft_from_source, find_doc_by_source
+from ..services.knowledge_index import reindex_document, search_knowledge
 from .attachments import count_for_items, list_for_item
 
 router = APIRouter(prefix="/workspaces/{ws_id}", tags=["知识库"])
@@ -203,6 +204,20 @@ async def list_knowledge(
     return ok({"items": items, "next_cursor": next_cursor, "has_more": has_more})
 
 
+# 注意路由顺序：`/knowledge/search` 必须排在 `/knowledge/{item_id}` 之前，
+# 否则 "search" 会被当成 uuid 解析，返回 422。
+@router.get("/knowledge/search", summary="检索已发布知识")
+async def search(
+    ctx: MemberDep,
+    session: SessionDep,
+    q: Annotated[str, Query(min_length=1, max_length=200, description="检索词")],
+    limit: Annotated[int, Query(ge=1, le=30)] = 10,
+    mode: Annotated[str, Query(pattern="^(hybrid|vector|keyword)$")] = "hybrid",
+) -> dict:
+    hits = await search_knowledge(session, ctx.ws_id, q, limit=limit, mode=mode)
+    return ok({"query": q, "mode": mode, "items": hits, "has_more": False})
+
+
 @router.get("/knowledge/{item_id}", summary="知识详情")
 async def get_knowledge(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
     item, doc = await _load_doc(session, ctx.ws_id, item_id)
@@ -396,9 +411,28 @@ async def change_status(
         payload={"to": to_status},
     )
     await session.flush()
+    if body.action == "publish":
+        # 发布后立刻建索引：检索只认 published，不在这里建就要等定时任务，
+        # 而"刚发布却搜不到"是最容易被当成 bug 的体验。
+        await reindex_document(session, ws_id, item.id)
     return ok(await _knowledge_out(session, ws_id, item, doc))
 
 
+@router.post("/knowledge/{item_id}/reindex", summary="重建这份知识的检索切片")
+async def reindex(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
+    # 改完正文不重建，检索命中的还是旧句子——留一个能手动触发的入口
+    _item, doc = await _load_doc(session, ctx.ws_id, item_id)
+    count = await reindex_document(session, ctx.ws_id, item_id)
+    await session.flush()
+    return ok(
+        {
+            "item_id": item_id,
+            "chunk_count": count,
+            "searchable": doc.status_hint == "published",
+        }
+    )
+
+
 # ── 从任务/工单沉淀 ─────────────────────────────────────────────────────
 @router.post("/items/{item_id}/knowledge", summary="把这件事沉淀成知识草稿")
 async def promote_to_knowledge(

backend/app/config.py

@@ -89,6 +89,9 @@ class Settings(BaseSettings):
     search_vector_weight: float = 0.6
     search_lexical_weight: float = 0.4
 
+    # 低于这个分数就不返回：宁可"没找到"，也不要塞一条不相干的让人误信。
+    search_min_score: float = 0.06
+
     # ── 知识库向量库（独立实例，M3 起用）───────────────────
     # 刻意与业务库分开放：fnsvr-01 上的 PG 没有 pgvector，也不该为了装扩展去动它；
     # 向量检索吃内存，和任务库挤在同一实例会互相拖累。

backend/app/domain/chunking.py

@@ -0,0 +1,127 @@
+"""把 Markdown 正文切成适合检索的小片段（M3）。
+
+切片的目的是"喂给检索"，不是"给人读"，所以三条规则优先于其它考虑：
+
+1. **不切坏结构**：代码块整块保留，表格不拆行，标题跟着它下面的段落走
+   （单独一句"结论"离开上下文就没意义）。
+2. **片内自洽**：每片尽量包含一个完整意思，超长才按句号切。
+3. **相邻片有重叠**：跨段的话被切断时，前后片都能读到完整句子，命中后引用才不乱。
+
+不引入分词器：中文另起 `zhparser` 要装扩展，而片段足够短时字符级检索已经够用。
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*$")
+FENCE_RE = re.compile(r"^\s*```")
+SENTENCE_RE = re.compile(r"[^。！？!?；;\n]+[。！？!?；;]?")
+
+
+@dataclass(frozen=True)
+class ChunkDraft:
+    heading: str | None
+    content: str
+
+
+def _split_long(text: str, max_chars: int) -> list[str]:
+    """把超长段落按句子切开；单句仍然超长就硬切。"""
+    pieces: list[str] = []
+    buffer = ""
+    for sentence in SENTENCE_RE.findall(text):
+        if not sentence.strip():
+            continue
+        if len(buffer) + len(sentence) <= max_chars:
+            buffer += sentence
+            continue
+        if buffer.strip():
+            pieces.append(buffer.strip())
+        if len(sentence) <= max_chars:
+            buffer = sentence
+        else:
+            for start in range(0, len(sentence), max_chars):
+                pieces.append(sentence[start : start + max_chars].strip())
+            buffer = ""
+    if buffer.strip():
+        pieces.append(buffer.strip())
+    return [p for p in pieces if p]
+
+
+def split_markdown(
+    text: str,
+    *,
+    max_chars: int = 600,
+    overlap: int = 80,
+) -> list[ChunkDraft]:
+    """返回按顺序排列的切片。
+
+    `overlap` 是相邻片之间共享的尾部字数；不想要重叠就传 0。
+    """
+    if not text or not text.strip():
+        return []
+
+    normalized = text.replace("\r\n", "\n").replace("\r", "\n")
+    blocks: list[tuple[str | None, str]] = []
+    heading: str | None = None
+    paragraph: list[str] = []
+    in_fence = False
+
+    def flush() -> None:
+        if paragraph:
+            body = "\n".join(paragraph).strip()
+            if body:
+                blocks.append((heading, body))
+            paragraph.clear()
+
+    for line in normalized.split("\n"):
+        if FENCE_RE.match(line):
+            in_fence = not in_fence
+            paragraph.append(line)
+            continue
+        if not in_fence:
+            match = HEADING_RE.match(line)
+            if match:
+                flush()
+                heading = match.group(2).strip()
+                continue
+            if not line.strip():
+                flush()
+                continue
+        paragraph.append(line)
+    flush()
+
+    chunks: list[ChunkDraft] = []
+    buffer = ""
+    buffer_heading: str | None = None
+
+    def flush_buffer() -> None:
+        nonlocal buffer, buffer_heading
+        if buffer.strip():
+            chunks.append(ChunkDraft(heading=buffer_heading, content=buffer.strip()))
+        buffer = ""
+
+    for block_heading, body in blocks:
+        parts = _split_long(body, max_chars) if len(body) > max_chars else [body]
+        for part in parts:
+            candidate = f"{buffer}\n\n{part}".strip() if buffer else part
+            if len(candidate) <= max_chars:
+                if not buffer:
+                    buffer_heading = block_heading
+                buffer = candidate
+                continue
+            flush_buffer()
+            # 新片开头带上上一片的尾巴，避免"半句话"被切断后两边都读不通
+            tail = ""
+            if overlap > 0 and chunks:
+                tail = chunks[-1].content[-overlap:]
+                if "\n" in tail:
+                    tail = tail.split("\n", 1)[-1]
+            buffer_heading = block_heading
+            buffer = f"{tail}\n\n{part}".strip() if tail else part
+            if len(buffer) > max_chars:
+                # 带尾巴后仍然超长：放弃尾巴，保证不超限
+                buffer = part
+    flush_buffer()
+    return chunks

backend/app/models_knowledge.py

@@ -19,15 +19,18 @@ from datetime import datetime
 from sqlalchemy import (
     CheckConstraint,
     DateTime,
+    Float,
     ForeignKey,
     Index,
     String,
     Text,
+    UniqueConstraint,
 )
+from sqlalchemy.dialects.postgresql import ARRAY
 from sqlalchemy.dialects.postgresql import UUID as PgUUID
 from sqlalchemy.orm import Mapped, mapped_column
 
-from .models import Base, Timestamps
+from .models import Base, Timestamps, new_uuid
 
 # 知识状态机（与任务/工单共用 item.status 一列，取值集合不同）
 KNOWLEDGE_STATUSES: tuple[str, ...] = ("draft", "in_review", "published", "archived")
@@ -76,4 +79,34 @@ class KnowledgeDoc(Base, Timestamps):
     chunk_count: Mapped[int] = mapped_column(nullable=False, default=0)
 
 
-KNOWLEDGE_TENANT_TABLES: tuple[str, ...] = ("knowledge_doc",)
+class Chunk(Base):
+    """知识正文的切片。
+
+    为什么切片与文档同库：检索要按工作区过滤，跨库做权限校验就得把两边数据捞到应用里，
+    既不安全也不快。向量本身是普通数组，换 pgvector 时只换列类型与索引。
+    """
+
+    __tablename__ = "chunk"
+    __table_args__ = (
+        UniqueConstraint("item_id", "idx", name="uq_chunk_item_idx"),
+        Index("ix_chunk_ws_item", "workspace_id", "item_id"),
+        {"comment": "知识正文切片，检索的最小单位"},
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
+    workspace_id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), nullable=False)
+    item_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE"), nullable=False
+    )
+    # 片内顺序，用于把命中片段按原顺序还原上下文
+    idx: Mapped[int] = mapped_column(nullable=False)
+    heading: Mapped[str | None] = mapped_column(String(200))
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    char_count: Mapped[int] = mapped_column(nullable=False, default=0)
+    # 向量。hash 提供方是 256 维，OpenAI 兼容端点通常是 1024/1536 维，
+    # 所以不固定维度——换模型时老切片要么重算，要么维度不匹配（检索会拒绝并提示重建）。
+    embedding: Mapped[list[float] | None] = mapped_column(ARRAY(Float))
+    embed_model: Mapped[str | None] = mapped_column(String(64))
+
+
+KNOWLEDGE_TENANT_TABLES: tuple[str, ...] = ("knowledge_doc", "chunk")

backend/app/services/embedding.py

@@ -0,0 +1,131 @@
+"""文本向量化（M3）。
+
+为什么要抽象一层：今天用内置的哈希向量，明天换成 OpenAI 兼容的 `/embeddings`，
+后天可能换成自建模型。检索代码只依赖 `embed() -> list[float]`，换实现不动业务。
+
+内置 `HashEmbedding` 的诚实说明：它不是"语义模型"，是**字符 n-gram 的哈希词袋**。
+它能把同词、近义写法（"发票"/"票据"这种）之外的东西区分开，抓不到真正的语义相似。
+之所以先上它：本机没有可用的向量库与模型服务，检索链路不能一直不落地；
+而且它零依赖、可复现（同样的文本永远同样的向量），适合做测试基线。
+真上线前应当把 `EMBEDDING_PROVIDER` 换成 `openai`，并把已发布知识整体重建一次。
+"""
+
+from __future__ import annotations
+
+import hashlib
+import math
+import re
+from collections.abc import Sequence
+from typing import Protocol
+
+import httpx
+
+from ..config import Settings, get_settings
+
+CJK_RE = re.compile(r"[\u4e00-\u9fff]")
+WORD_RE = re.compile(r"[a-z0-9]+")
+
+
+class EmbeddingProvider(Protocol):
+    """向量提供方。`name` 会写进切片，换模型后能看出哪些是旧向量。"""
+
+    name: str
+
+    async def embed(self, texts: Sequence[str]) -> list[list[float]]: ...
+
+
+def _tokens(text: str) -> list[str]:
+    """中文按字二元组、英文按词。
+
+    中文不用分词器：单字太碎（"发""票"都对不上"发票"），二元组能拿到大多数
+    有意义的搭配，又不需要维护词典。
+    """
+    lowered = text.lower()
+    tokens = WORD_RE.findall(lowered)
+    chars = [c for c in lowered if CJK_RE.match(c)]
+    tokens.extend(chars[i] + chars[i + 1] for i in range(len(chars) - 1))
+    tokens.extend(chars)
+    return tokens
+
+
+class HashEmbedding:
+    """字符 n-gram 的带符号哈希词袋，L2 归一化。"""
+
+    def __init__(self, dim: int = 256) -> None:
+        self.dim = dim
+        self.name = f"hash-{dim}"
+
+    async def embed(self, texts: Sequence[str]) -> list[list[float]]:
+        return [self._one(text) for text in texts]
+
+    def _one(self, text: str) -> list[float]:
+        vector = [0.0] * self.dim
+        counts: dict[str, int] = {}
+        for token in _tokens(text):
+            counts[token] = counts.get(token, 0) + 1
+        for token, count in counts.items():
+            digest = hashlib.blake2b(token.encode("utf-8"), digest_size=8).digest()
+            raw = int.from_bytes(digest, "big")
+            index = raw % self.dim
+            # 用最高位决定正负：无符号哈希会让不相干的词在向量里越加越像
+            sign = 1.0 if (raw >> 63) & 1 else -1.0
+            vector[index] += sign * (1.0 + math.log(count))
+        norm = math.sqrt(sum(v * v for v in vector))
+        if norm == 0:
+            return vector
+        return [v / norm for v in vector]
+
+
+class OpenAIEmbedding:
+    """任何 OpenAI 兼容的 `/embeddings` 端点（BYOK）。"""
+
+    def __init__(self, *, base_url: str, api_key: str, model: str) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.model = model
+        self.name = f"openai:{model}"
+
+    async def embed(self, texts: Sequence[str]) -> list[list[float]]:
+        if not texts:
+            return []
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(
+                f"{self.base_url}/embeddings",
+                headers={"Authorization": f"Bearer {self.api_key}"},
+                json={"model": self.model, "input": list(texts)},
+            )
+            response.raise_for_status()
+            payload = response.json()
+        # 按 index 排序：并发端点的返回顺序没有保证
+        rows = sorted(payload["data"], key=lambda row: row["index"])
+        return [row["embedding"] for row in rows]
+
+
+_provider: EmbeddingProvider | None = None
+
+
+def get_provider(settings: Settings | None = None) -> EmbeddingProvider:
+    """进程内单例：哈希实现无状态，OpenAI 实现也不需要每次新建连接池。"""
+    global _provider
+    if _provider is not None:
+        return _provider
+    settings = settings or get_settings()
+    if settings.embedding_provider == "openai":
+        if not settings.embedding_api_key:
+            raise RuntimeError(
+                "EMBEDDING_PROVIDER=openai 但没有配置 EMBEDDING_API_KEY；"
+                "要么补上密钥，要么改回 hash"
+            )
+        _provider = OpenAIEmbedding(
+            base_url=settings.embedding_base_url,
+            api_key=settings.embedding_api_key,
+            model=settings.embedding_model,
+        )
+    else:
+        _provider = HashEmbedding()
+    return _provider
+
+
+def reset_provider_for_tests() -> None:
+    global _provider
+    _provider = None

backend/app/services/knowledge_index.py

@@ -0,0 +1,208 @@
+"""知识的切片索引与检索（M3）。
+
+两条硬规则：
+
+1. **只有 `published` 的知识能被检索到**。草稿是"还没被人认可的话"，
+   让它进检索等于把猜测当结论引用出去。过滤写在 SQL 里，不靠调用方自觉。
+2. **检索必须带来源**。每个命中都返回 `item_id` 与标题，用户能点回去看原文；
+   这和 D3（知识必须有来源）是同一条原则的延伸——不给"出处不存在"的答案。
+
+评分是向量与词法的加权和：
+- 向量负责"说的是同一件事但换了词"；
+- 词法（pg_trgm 的 trigram 相似度）负责"专有名词、单号、人名"这类必须原样命中的词，
+  而哈希向量对这类词并不敏感。
+两者都不足时宁可返回空，也不硬凑一条不相关的结果——检索失败比误导便宜。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from sqlalchemy import delete, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..config import get_settings
+from ..domain.chunking import split_markdown
+from ..errors import AppError, ErrorCode
+from ..models_knowledge import Chunk, KnowledgeDoc
+from ..models_task import Item
+from .embedding import get_provider
+
+SEARCH_SQL = text(
+    """
+    SELECT c.item_id,
+           i.title,
+           k.doc_type,
+           c.idx,
+           c.heading,
+           c.content,
+           coalesce(app_cosine(c.embedding, CAST(:qvec AS real[])), 0) AS vec_score,
+           coalesce(similarity(c.content, :q), 0) AS lex_score
+    FROM chunk c
+    JOIN item i ON i.id = c.item_id AND i.workspace_id = c.workspace_id
+    JOIN knowledge_doc k ON k.item_id = c.item_id AND k.workspace_id = c.workspace_id
+    WHERE c.workspace_id = :ws
+      AND k.status_hint = 'published'
+    ORDER BY (
+        coalesce(app_cosine(c.embedding, CAST(:qvec AS real[])), 0) * :vw
+        + coalesce(similarity(c.content, :q), 0) * :lw
+    ) DESC, c.item_id, c.idx
+    LIMIT :limit
+    """
+)
+
+
+def _vector_param(values: list[float]) -> list[float]:
+    """把向量交给驱动去编码。
+
+    一开始我拼成 `'{0.1,0.2}'` 字符串再 `CAST(... AS real[])`，asyncpg 会因为 CAST
+    已经告诉它"这是数组"而拒绝字符串（`a sized iterable container expected`）。
+    直接传 list 最省事；非有限值替换成 0，免得 NaN 进 SQL 变成错误输入。
+    """
+    return [value if value == value and abs(value) != float("inf") else 0.0 for value in values]
+
+
+async def reindex_document(
+    session: AsyncSession,
+    ws_id: uuid.UUID,
+    item_id: uuid.UUID,
+) -> int:
+    """重建一份知识的切片，返回切片数。
+
+    全量重建而不是增量：知识正文改动通常牵动上下几段，增量对齐的复杂度
+    换不来多少收益（一份知识撑死几十片）。
+    """
+    item = (
+        await session.execute(select(Item).where(Item.workspace_id == ws_id, Item.id == item_id))
+    ).scalar_one_or_none()
+    if item is None:
+        raise AppError(ErrorCode.ITEM_NOT_FOUND, "这件事不存在")
+
+    doc = (
+        await session.execute(
+            select(KnowledgeDoc).where(
+                KnowledgeDoc.workspace_id == ws_id,
+                KnowledgeDoc.item_id == item_id,
+            )
+        )
+    ).scalar_one_or_none()
+    if doc is None:
+        raise AppError(ErrorCode.KNOWLEDGE_NOT_FOUND, "这份知识不存在")
+
+    settings = get_settings()
+    source = "\n\n".join(part for part in (item.title, item.body_md or "") if part)
+    drafts = split_markdown(
+        source,
+        max_chars=settings.chunk_max_chars,
+        overlap=settings.chunk_overlap_chars,
+    )
+
+    provider = get_provider(settings)
+    vectors = await provider.embed([draft.content for draft in drafts]) if drafts else []
+
+    await session.execute(
+        delete(Chunk).where(Chunk.workspace_id == ws_id, Chunk.item_id == item_id)
+    )
+    for index, (draft, vector) in enumerate(zip(drafts, vectors, strict=True)):
+        session.add(
+            Chunk(
+                workspace_id=ws_id,
+                item_id=item_id,
+                idx=index,
+                heading=draft.heading,
+                content=draft.content,
+                char_count=len(draft.content),
+                embedding=vector,
+                embed_model=provider.name,
+            )
+        )
+    doc.chunk_count = len(drafts)
+    await session.flush()
+    return len(drafts)
+
+
+async def search_knowledge(
+    session: AsyncSession,
+    ws_id: uuid.UUID,
+    query: str,
+    *,
+    limit: int = 10,
+    mode: str = "hybrid",
+) -> list[dict[str, Any]]:
+    """检索已发布知识，返回按相关度排序的切片。
+
+    `mode`: hybrid（默认）/ vector / keyword —— 前两者主要用于对比效果。
+    """
+    cleaned = (query or "").strip()
+    if not cleaned:
+        return []
+
+    settings = get_settings()
+    provider = get_provider(settings)
+    query_vector = (await provider.embed([cleaned]))[0]
+
+    if mode == "vector":
+        vector_weight, lexical_weight = 1.0, 0.0
+    elif mode == "keyword":
+        vector_weight, lexical_weight = 0.0, 1.0
+    else:
+        vector_weight, lexical_weight = (
+            settings.search_vector_weight,
+            settings.search_lexical_weight,
+        )
+
+    rows = (
+        await session.execute(
+            SEARCH_SQL,
+            {
+                "qvec": _vector_param(query_vector),
+                "q": cleaned,
+                "ws": str(ws_id),
+                "vw": vector_weight,
+                "lw": lexical_weight,
+                "limit": limit,
+            },
+        )
+    ).mappings()
+
+    hits: list[dict[str, Any]] = []
+    for row in rows:
+        vector_score = float(row["vec_score"] or 0.0)
+        lexical_score = float(row["lex_score"] or 0.0)
+        hits.append(
+            {
+                "item_id": row["item_id"],
+                "title": row["title"],
+                "doc_type": row["doc_type"],
+                "idx": row["idx"],
+                "heading": row["heading"],
+                "content": row["content"],
+                "score": round(vector_score * vector_weight + lexical_score * lexical_weight, 4),
+                "vector_score": round(vector_score, 4),
+                "lexical_score": round(lexical_score, 4),
+            }
+        )
+    hits = [hit for hit in hits if hit["score"] >= settings.search_min_score]
+    return hits
+
+
+async def reindex_workspace(session: AsyncSession, ws_id: uuid.UUID) -> int:
+    """把一个工作区里已发布的知识全部重建，返回处理的知识份数。
+
+    换 embedding 模型之后必须整体跑一次，否则新旧向量混在一张表里，
+    相似度分数没法比较。
+    """
+    item_ids = (
+        await session.execute(
+            select(KnowledgeDoc.item_id).where(
+                KnowledgeDoc.workspace_id == ws_id,
+                KnowledgeDoc.status_hint == "published",
+            )
+        )
+    ).scalars()
+    count = 0
+    for item_id in item_ids:
+        await reindex_document(session, ws_id, item_id)
+        count += 1
+    return count

backend/app/services/signing.py

@@ -37,7 +37,9 @@ def sign(workspace_id: uuid.UUID, attachment_id: uuid.UUID, ttl_seconds: int | N
     return f"exp={exp}&sig={_digest(workspace_id, attachment_id, exp)}"
 
 
-def verify(workspace_id: uuid.UUID, attachment_id: uuid.UUID, exp: int | None, sig: str | None) -> bool:
+def verify(
+    workspace_id: uuid.UUID, attachment_id: uuid.UUID, exp: int | None, sig: str | None
+) -> bool:
     if not exp or not sig:
         return False
     if exp < int(time.time()):  # 过期就是过期，不给宽限