fix(export): 增加文档级组合复杂度预算与并发渲染限制

针对 PR 审阅 P1「组合复杂度仍可长时间占满导出线程」与 P3「EXPORT_OUTPUT_TOO_LARGE 误标 HTTP 413」：

- plot: FunctionPlot 记录整块 AST 节点数（node_count），parser 累计
- html: 单篇文档累计节点预算 _MAX_TOTAL_PLOT_NODES=8000，超限回退占位
- service: 并发渲染信号量 MAX_CONCURRENT_RENDERS=2，超限额任务排队等待
- docs: 错误码区分同步 HTTP 错误与异步任务错误，EXPORT_OUTPUT_TOO_LARGE 由
  error_code 返回而非 HTTP 413
- 补充节点预算与并发限制两条回归测试（全量 627 通过）

Co-Authored-By: Claude Code <noreply@anthropic.com>

.gitignore

@@ -18,6 +18,8 @@ backend/.env
 # 运行期生成的 SQLite 索引（vault 下的 Markdown 测试数据需提交）
 backend/data/*.db*
 backend/data/credentials/
+# 运行期导出的 HTML/PDF/DOCX 产物（不提交）
+backend/data/exports/
 # 阶段验收笔记（验收用，不提交）
 backend/data/vault/验收/
 # 本机 MCP 配置、授权状态及服务器工作目录不得提交。

backend/app/config.py

@@ -25,6 +25,7 @@ class Settings:
     vault_path: Path
     attachments_path: Path
     benchmark_datasets_path: Path
+    exports_path: Path
 
 
 @lru_cache
@@ -45,4 +46,5 @@ def get_settings() -> Settings:
         benchmark_datasets_path=Path(
             os.getenv("APP_BENCHMARK_DATASETS_PATH", str(data_dir / "benchmarks"))
         ),
+        exports_path=Path(os.getenv("APP_EXPORTS_PATH", str(data_dir / "exports"))),
     )

backend/app/contracts.py

@@ -2,7 +2,14 @@ from datetime import datetime
 from enum import Enum
 from typing import Annotated, Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator, model_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    SecretStr,
+    field_validator,
+    model_validator,
+)
 from app.request_overrides import RequestOverride
 
 
@@ -1273,3 +1280,88 @@ class BenchmarkReport(Contract):
     cases: list[RAGCaseResult] = Field(default_factory=list)
     error: str | None = None
     error_code: str | None = None
+
+
+# Export（多格式文档导出）
+class ExportStatus(str, Enum):
+    queued = "queued"
+    running = "running"
+    completed = "completed"
+    failed = "failed"
+    cancelled = "cancelled"
+
+
+class ExportFormat(str, Enum):
+    html = "html"
+    pdf = "pdf"
+    docx = "docx"
+
+
+class ExportSourceType(str, Enum):
+    note = "note"
+    markdown = "markdown"
+
+
+class ExportSource(Contract):
+    """导出源：note 引用已索引笔记，markdown 用于未保存预览（不持久化）。"""
+
+    type: ExportSourceType
+    note_id: str | None = None
+    markdown: str | None = None
+
+    @model_validator(mode="after")
+    def _validate_source(self) -> "ExportSource":
+        if self.type == ExportSourceType.note and not self.note_id:
+            raise ValueError("note source requires note_id")
+        if self.type == ExportSourceType.markdown and not self.markdown:
+            raise ValueError("markdown source requires markdown")
+        return self
+
+
+class ExportOptions(Contract):
+    theme_id: str = "light"
+    include_title: bool = True
+    include_metadata: bool = False
+    page_size: str = "A4"
+    code_theme: str = "github-light"
+
+
+class ExportRequest(Contract):
+    source: ExportSource
+    format: ExportFormat
+    options: ExportOptions = Field(default_factory=ExportOptions)
+
+
+class ExportProgress(Contract):
+    phase: str
+    current: int
+    total: int
+    percent: float | None = None
+    message: str | None = None
+
+
+class ExportFile(Contract):
+    file_name: str
+    mime_type: str
+    size: int
+    sha256: str
+    expires_at: datetime
+
+
+class ExportJob(Contract):
+    job_id: str
+    status: ExportStatus
+    format: ExportFormat
+    progress: ExportProgress | None = None
+    file: ExportFile | None = None
+    warnings: list[str] = Field(default_factory=list)
+    error: str | None = None
+    error_code: str | None = None
+    created_at: datetime
+    started_at: datetime | None = None
+    completed_at: datetime | None = None
+
+
+class ExportJobListResponse(Contract):
+    items: list[ExportJob] = Field(default_factory=list)
+    page: PageMeta = Field(default_factory=PageMeta)

backend/app/export/__init__.py

@@ -0,0 +1,8 @@
+"""Export Service：多格式文档导出（首批 HTML）。
+
+模块划分：
+- document.py        Document AST 内部协议 + DocumentExporter Protocol + ExportResult
+- markdown.py        mistune → Document AST 解析
+- exporters/html.py  HtmlExporter（Document AST → HTML5）
+- service.py         导出任务注册表、后台执行、取消与文件生命周期
+"""

backend/app/export/document.py

@@ -0,0 +1,46 @@
+"""Document AST：导出器的内部中间表示（Internal Protocol，不放入 contracts.py）。
+
+契约 §10.3 规定节点用稳定判别字段 node_id / type / attributes / children / text，
+类型专有信息统一放 attributes（如 heading 的 level、link 的 href、image 的 src）。
+导出器据此递归渲染，对无法表示的节点记 warning，不静默丢弃。
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.contracts import ExportOptions
+
+
+class DocumentNode(BaseModel):
+    """递归文档节点；type 取契约 §10.3 首批 node type 之一。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+    type: str
+    node_id: str
+    attributes: dict[str, Any] = Field(default_factory=dict)
+    children: list["DocumentNode"] = Field(default_factory=list)
+    text: str = ""
+
+
+class Document(DocumentNode):
+    """根节点，type 固定为 document。"""
+
+    type: str = "document"
+
+
+class DocumentExporter(Protocol):
+    """导出器协议（契约 §10.3）：把 Document AST 渲染为指定格式的产物。"""
+
+    async def export(self, document: Document, options: ExportOptions) -> "ExportResult": ...
+
+
+class ExportResult(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    content: bytes
+    mime_type: str
+    warnings: list[str] = Field(default_factory=list)

backend/app/export/exporters/__init__.py

@@ -0,0 +1 @@
+"""Export 渲染器：Document AST → 具体格式产物。"""

backend/app/export/exporters/html.py

@@ -0,0 +1,296 @@
+"""HtmlExporter：Document AST → 完整 HTML5 文档（内嵌基础 CSS）。
+
+mermaid 等无法静态表达的节点渲染为占位代码块并记 warning，不静默丢失；function_plot
+解析为静态 SVG 内嵌（解析失败回退占位并转诊断）；严重内容缺失由 service 层以
+EXPORT_UNSUPPORTED_CONTENT 判定，本层只负责逐节点渲染。
+"""
+
+from __future__ import annotations
+
+import html
+from datetime import datetime
+from urllib.parse import urlparse
+
+from app.contracts import ExportOptions
+from app.export.document import Document, DocumentNode, ExportResult
+from app.plot.parser import parse_source
+from app.plot.render import render_svg
+
+_MERMAID_WARNING = "mermaid 需前端渲染，已保留为占位代码块"
+_RAW_HTML_WARNING = "原始 HTML 已按纯文本转义保留"
+
+# 链接/图片地址允许的协议；无 scheme 的相对地址视为安全，其余协议一律降级
+_ALLOWED_URL_SCHEMES = frozenset({"http", "https", "mailto"})
+
+# 单篇文档允许的函数图像数量上限，超出部分回退占位，防止多图块并发采样耗尽内存/线程
+_MAX_FUNCTION_PLOTS = 16
+# 单篇文档允许的函数图像累计 AST 节点预算，超出部分回退占位，防止组合复杂度（多图块
+# × 多表达式 × 深表达式）在采样求值时长时间占满 CPU
+_MAX_TOTAL_PLOT_NODES = 8000
+
+
+def _safe_url(url: str) -> str | None:
+    """校验 URL 协议；安全返回原串，不安全返回 None。"""
+    url = url.strip()
+    if not url:
+        return None
+    scheme = urlparse(url).scheme.lower()
+    if scheme and scheme not in _ALLOWED_URL_SCHEMES:
+        return None
+    return url
+
+_BASE_CSS = """
+body { margin: 0; background: #f6f7f9; color: #1f2328; font: 15px/1.7 -apple-system, 'Segoe UI', 'Microsoft YaHei', sans-serif; }
+article { max-width: 860px; margin: 0 auto; padding: 40px 48px; background: #fff; }
+article.theme-dark { background: #0d1117; color: #c9d1d9; }
+h1, h2, h3, h4, h5, h6 { line-height: 1.3; margin: 1.4em 0 0.6em; }
+h1.title { margin-top: 0; }
+p { margin: 0.6em 0; }
+a { color: #0969da; }
+code { font-family: 'JetBrains Mono', Consolas, monospace; font-size: 0.9em; background: #f0f1f3; padding: 0.15em 0.35em; border-radius: 3px; }
+pre { background: #f6f8fa; padding: 14px 16px; border-radius: 6px; overflow-x: auto; }
+pre.code-theme-github-dark { background: #0d1117; color: #c9d1d9; }
+pre code { background: none; padding: 0; }
+pre.mermaid, pre.function-plot { border: 1px dashed #d0d7de; }
+figure.function-plot { margin: 1em 0; text-align: center; }
+figure.function-plot svg { max-width: 100%; height: auto; }
+blockquote { margin: 0.8em 0; padding: 0.2em 1em; border-left: 4px solid #d0d7de; color: #57606a; }
+img { max-width: 100%; }
+table { border-collapse: collapse; margin: 0.8em 0; }
+th, td { border: 1px solid #d0d7de; padding: 6px 12px; }
+th { background: #f6f8fa; }
+dl.metadata { font-size: 0.85em; color: #57606a; border-top: 1px solid #eaeef2; border-bottom: 1px solid #eaeef2; padding: 0.6em 0; }
+dl.metadata dt { display: inline; font-weight: 600; margin-right: 0.4em; }
+dl.metadata dd { display: inline; margin: 0 1.2em 0 0; }
+.math, .math-block { overflow-x: auto; padding: 0.4em 0; }
+.task-list-item { list-style: none; }
+.task-list-item input { margin-right: 0.4em; }
+hr { border: none; border-top: 1px solid #d0d7de; margin: 1.4em 0; }
+""".strip()
+
+
+class HtmlExporter:
+    """实现 DocumentExporter：递归渲染 Document AST 为完整 HTML5 文档。"""
+
+    def render(self, document: Document, options: ExportOptions) -> ExportResult:
+        """同步渲染；CPU 密集，调用方应放入线程执行，避免阻塞事件循环。"""
+        self._options = options
+        self._plot_count = 0
+        self._plot_nodes = 0
+        warnings: list[str] = []
+        body = self._render_children(document.children, warnings)
+        content = self._assemble(document, options, body, warnings)
+        return ExportResult(
+            content=content.encode("utf-8"), mime_type="text/html", warnings=warnings
+        )
+
+    async def export(self, document: Document, options: ExportOptions) -> ExportResult:
+        """契约要求的 async 接口；渲染本身同步，直接转发到 render。"""
+        return self.render(document, options)
+
+    def _assemble(
+        self, document: Document, options: ExportOptions, body: str, warnings: list[str]
+    ) -> str:
+        title = str(document.attributes.get("title") or "")
+        parts = [
+            "<!doctype html>",
+            '<html lang="zh-CN">',
+            "<head>",
+            '<meta charset="utf-8">',
+            '<meta name="viewport" content="width=device-width, initial-scale=1">',
+        ]
+        if title:
+            parts.append(f"<title>{html.escape(title)}</title>")
+        parts.append(f"<style>{_BASE_CSS}</style>")
+        parts.append("</head>")
+        parts.append("<body>")
+        parts.append(f'<article class="theme-{html.escape(options.theme_id)}">')
+        if options.include_title and title:
+            parts.append(f'<h1 class="title">{html.escape(title)}</h1>')
+        if options.include_metadata:
+            metadata = document.attributes.get("metadata")
+            if metadata:
+                parts.append(self._render_metadata(metadata))
+        parts.append(body)
+        parts.append("</article>")
+        parts.append("</body>")
+        parts.append("</html>")
+        return "\n".join(parts) + "\n"
+
+    def _render_metadata(self, metadata: dict) -> str:
+        entries = ["<dl", ' class="metadata">']
+        for key, value in metadata.items():
+            entries.append(f"<dt>{html.escape(str(key))}</dt>")
+            entries.append(f"<dd>{html.escape(self._fmt_meta_value(value))}</dd>")
+        entries.append("</dl>")
+        return "".join(entries)
+
+    @staticmethod
+    def _fmt_meta_value(value: object) -> str:
+        if isinstance(value, datetime):
+            return value.isoformat()
+        if isinstance(value, list):
+            return ", ".join(str(item) for item in value)
+        return str(value)
+
+    def _render_children(self, children: list[DocumentNode], warnings: list[str]) -> str:
+        return "".join(self._render_node(child, warnings) for child in children)
+
+    def _render_node(self, node: DocumentNode, warnings: list[str]) -> str:
+        handler = getattr(self, f"_render_{node.type}", None)
+        if handler is not None:
+            return handler(node, warnings)
+        warnings.append(f"无法表示的节点类型已跳过：{node.type}")
+        return ""
+
+    # --- 块级 ---
+    def _render_heading(self, node: DocumentNode, warnings: list[str]) -> str:
+        level = max(1, min(6, int(node.attributes.get("level", 1))))
+        return f"<h{level}>{self._render_children(node.children, warnings)}</h{level}>"
+
+    def _render_paragraph(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f"<p>{self._render_children(node.children, warnings)}</p>"
+
+    def _render_blockquote(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f"<blockquote>{self._render_children(node.children, warnings)}</blockquote>"
+
+    def _render_list(self, node: DocumentNode, warnings: list[str]) -> str:
+        tag = "ol" if node.attributes.get("ordered") else "ul"
+        return f"<{tag}>{self._render_children(node.children, warnings)}</{tag}>"
+
+    def _render_list_item(self, node: DocumentNode, warnings: list[str]) -> str:
+        inner = self._render_children(node.children, warnings)
+        if node.attributes.get("task"):
+            checked = " checked" if node.attributes.get("checked") else ""
+            return (
+                '<li class="task-list-item">'
+                f'<input type="checkbox" disabled{checked}>{inner}</li>'
+            )
+        return f"<li>{inner}</li>"
+
+    def _render_table(self, node: DocumentNode, warnings: list[str]) -> str:
+        rows = node.children
+        head_rows = [r for r in rows if r.attributes.get("head")]
+        body_rows = [r for r in rows if not r.attributes.get("head")]
+        parts = ["<table>"]
+        if head_rows:
+            parts.append("<thead>")
+            parts.extend(self._render_node(r, warnings) for r in head_rows)
+            parts.append("</thead>")
+        if body_rows:
+            parts.append("<tbody>")
+            parts.extend(self._render_node(r, warnings) for r in body_rows)
+            parts.append("</tbody>")
+        parts.append("</table>")
+        return "".join(parts)
+
+    def _render_table_row(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f"<tr>{self._render_children(node.children, warnings)}</tr>"
+
+    def _render_table_cell(self, node: DocumentNode, warnings: list[str]) -> str:
+        tag = "th" if node.attributes.get("head") else "td"
+        return f"<{tag}>{self._render_children(node.children, warnings)}</{tag}>"
+
+    def _render_code_block(self, node: DocumentNode, warnings: list[str]) -> str:
+        lang = str(node.attributes.get("language") or "")
+        code = html.escape(node.text)
+        lang_cls = f' class="language-{html.escape(lang)}"' if lang else ""
+        theme = html.escape(self._options.code_theme)
+        return f'<pre class="code-theme-{theme}"><code{lang_cls}>{code}</code></pre>'
+
+    def _render_thematic_break(self, node: DocumentNode, warnings: list[str]) -> str:
+        return "<hr>"
+
+    def _render_mermaid(self, node: DocumentNode, warnings: list[str]) -> str:
+        warnings.append(_MERMAID_WARNING)
+        return f'<pre class="mermaid">{html.escape(node.text)}</pre>'
+
+    @staticmethod
+    def _format_plot_diagnostic(diag) -> str:
+        loc = f"（第 {diag.line} 行）" if diag.line else ""
+        return f"函数图像：{diag.message}{loc}"
+
+    def _render_function_plot(self, node: DocumentNode, warnings: list[str]) -> str:
+        # 文档级数量上限：超出部分直接回退占位，不解析不采样，防止海量图像耗尽资源
+        self._plot_count += 1
+        if self._plot_count > _MAX_FUNCTION_PLOTS:
+            warnings.append(
+                f"函数图像：文档内函数图像数量超过上限 {_MAX_FUNCTION_PLOTS}，已回退为源码占位"
+            )
+            return f'<pre class="function-plot">{html.escape(node.text)}</pre>'
+        # 解析与渲染共同纳入局部异常回退：单个图像失败只回退占位 + warning，
+        # 绝不阻断整篇导出（含复杂表达式触发的 RecursionError 等异常）。
+        try:
+            parsed = parse_source(node.text)
+            for diag in parsed.diagnostics:
+                warnings.append(self._format_plot_diagnostic(diag))
+            if parsed.plot is None:
+                return f'<pre class="function-plot">{html.escape(node.text)}</pre>'
+            # 文档级累计复杂度预算：超出后回退占位，不再采样求值
+            if self._plot_nodes + parsed.plot.node_count > _MAX_TOTAL_PLOT_NODES:
+                warnings.append(
+                    f"函数图像：文档内函数图像累计复杂度超过上限 {_MAX_TOTAL_PLOT_NODES} 节点，已回退为源码占位"
+                )
+                return f'<pre class="function-plot">{html.escape(node.text)}</pre>'
+            self._plot_nodes += parsed.plot.node_count
+            rendered = render_svg(parsed.plot)
+        except Exception as exc:
+            warnings.append(f"函数图像：解析或渲染失败，已回退占位（{exc}）")
+            return f'<pre class="function-plot">{html.escape(node.text)}</pre>'
+        warnings.extend(rendered.warnings)
+        return f'<figure class="function-plot">{rendered.content}</figure>'
+
+    def _render_math_block(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f'<div class="math-block">$${html.escape(node.text)}$$</div>'
+
+    def _render_html_block(self, node: DocumentNode, warnings: list[str]) -> str:
+        # 原始 HTML 不可信，转义为纯文本展示，保证正文不丢且无注入风险
+        warnings.append(_RAW_HTML_WARNING)
+        return f'<div class="raw-html">{html.escape(node.text)}</div>'
+
+    # --- 行内 ---
+    def _render_text(self, node: DocumentNode, warnings: list[str]) -> str:
+        return html.escape(node.text)
+
+    def _render_emphasis(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f"<em>{self._render_children(node.children, warnings)}</em>"
+
+    def _render_strong(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f"<strong>{self._render_children(node.children, warnings)}</strong>"
+
+    def _render_link(self, node: DocumentNode, warnings: list[str]) -> str:
+        inner = self._render_children(node.children, warnings)
+        href = str(node.attributes.get("href") or "")
+        safe_href = _safe_url(href)
+        if safe_href is None:
+            # 危险协议（如 javascript:）降级为纯文本，不输出可点击链接
+            warnings.append(f"链接协议不安全，已降级为纯文本：{href!r}")
+            return inner
+        title = str(node.attributes.get("title") or "")
+        attrs = [f'href="{html.escape(safe_href)}"']
+        if title:
+            attrs.append(f'title="{html.escape(title)}"')
+        return f"<a {' '.join(attrs)}>{inner}</a>"
+
+    def _render_codespan(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f"<code>{html.escape(node.text)}</code>"
+
+    def _render_image(self, node: DocumentNode, warnings: list[str]) -> str:
+        src = str(node.attributes.get("src") or "")
+        alt = str(node.attributes.get("alt") or "")
+        safe_src = _safe_url(src)
+        if safe_src is None:
+            # 危险协议（如 data:/javascript:）跳过图片，仅输出 alt 文本
+            warnings.append(f"图片地址不安全，已跳过：{src!r}")
+            return html.escape(alt) if alt else ""
+        title = str(node.attributes.get("title") or "")
+        attrs = [f'src="{html.escape(safe_src)}"', f'alt="{html.escape(alt)}"']
+        if title:
+            attrs.append(f'title="{html.escape(title)}"')
+        return f"<img {' '.join(attrs)}>"
+
+    def _render_math_inline(self, node: DocumentNode, warnings: list[str]) -> str:
+        return f"\\({html.escape(node.text)}\\)"
+
+    def _render_linebreak(self, node: DocumentNode, warnings: list[str]) -> str:
+        return "<br>"

backend/app/export/markdown.py

@@ -0,0 +1,229 @@
+"""Markdown → Document AST：用 mistune 的 ast renderer 产出通用 token，再映射为内部节点。
+
+选用 mistune 内置 'ast' renderer 而非自写 BaseRenderer，是因为 mistune 的行内渲染按
+字符串拼接、无法承载结构化子节点；ast renderer 直接给出带 children/attrs/raw 的 token
+树，映射层只做 token → DocumentNode 的搬运，不掺入任何 HTML。
+"""
+
+from __future__ import annotations
+
+import mistune
+
+from app.export.document import Document, DocumentNode
+
+_PLUGINS = ["table", "math", "url", "task_lists"]
+
+# fenced code 语言分流：命中则转为专用节点，其余按普通代码块
+_MERMAID_LANG = "mermaid"
+_FUNCTION_PLOT_LANGS = {"function-plot", "function_plot", "functionplot"}
+
+
+def parse_document(markdown: str) -> Document:
+    """把 Markdown 文本解析为 Document AST 根节点。"""
+    renderer = mistune.create_markdown(renderer="ast", plugins=_PLUGINS)
+    tokens = renderer(markdown)
+    mapper = _AstMapper()
+    return Document(node_id=mapper.next_id(), children=mapper.map_blocks(tokens))
+
+
+class _AstMapper:
+    """token 树 → DocumentNode 树的映射器；node_id 按遍历顺序递增，无需跨请求稳定。"""
+
+    def __init__(self) -> None:
+        self._seq = 0
+
+    def next_id(self) -> str:
+        self._seq += 1
+        return f"node_{self._seq:03d}"
+
+    def map_blocks(self, tokens: list[dict]) -> list[DocumentNode]:
+        nodes: list[DocumentNode] = []
+        for token in tokens:
+            node = self.map_block(token)
+            if node is not None:
+                nodes.append(node)
+        return nodes
+
+    def map_block(self, token: dict) -> DocumentNode | None:
+        kind = token["type"]
+        if kind == "heading":
+            return DocumentNode(
+                type="heading",
+                node_id=self.next_id(),
+                attributes={"level": token["attrs"]["level"]},
+                children=self.map_inline(token.get("children", [])),
+            )
+        if kind in ("paragraph", "block_text"):
+            # block_text 是列表项内的段落块，仍按 paragraph 表达，由 list_item 包裹
+            return DocumentNode(
+                type="paragraph",
+                node_id=self.next_id(),
+                children=self.map_inline(token.get("children", [])),
+            )
+        if kind == "list":
+            return DocumentNode(
+                type="list",
+                node_id=self.next_id(),
+                attributes={"ordered": bool(token.get("attrs", {}).get("ordered"))},
+                children=[self.map_list_item(child) for child in token.get("children", [])],
+            )
+        if kind == "block_code":
+            return self._map_code(token)
+        if kind == "block_quote":
+            return DocumentNode(
+                type="blockquote",
+                node_id=self.next_id(),
+                children=self.map_blocks(token.get("children", [])),
+            )
+        if kind == "table":
+            return self._map_table(token)
+        if kind == "block_math":
+            return DocumentNode(
+                type="math_block", node_id=self.next_id(), text=token.get("raw", "")
+            )
+        if kind == "thematic_break":
+            return DocumentNode(type="thematic_break", node_id=self.next_id())
+        if kind == "blank_line":
+            return None
+        if kind == "block_html":
+            # 原始 HTML 块降级为纯文本节点，由 HtmlExporter 转义并记 warning，避免静默丢失正文
+            return DocumentNode(
+                type="html_block", node_id=self.next_id(), text=token.get("raw", "")
+            )
+        # 未知块级 token 保守保留原文；映射为带 text 子节点的 paragraph，避免被渲染层丢弃
+        raw = token.get("raw", "")
+        if raw:
+            return DocumentNode(
+                type="paragraph",
+                node_id=self.next_id(),
+                children=[DocumentNode(type="text", node_id=self.next_id(), text=raw)],
+            )
+        return None
+
+    def map_list_item(self, token: dict) -> DocumentNode:
+        """列表项：block_text 展平为行内子节点，嵌套 list 保留为子节点。"""
+        attributes: dict = {}
+        if token["type"] == "task_list_item":
+            attributes = {"task": True, "checked": bool(token.get("attrs", {}).get("checked"))}
+        children: list[DocumentNode] = []
+        for child in token.get("children", []):
+            if child["type"] == "block_text":
+                children.extend(self.map_inline(child.get("children", [])))
+            elif child["type"] == "list":
+                children.append(self.map_block(child))
+            else:
+                node = self.map_block(child)
+                if node is not None:
+                    children.append(node)
+        return DocumentNode(
+            type="list_item", node_id=self.next_id(), attributes=attributes, children=children
+        )
+
+    def map_inline(self, tokens: list[dict]) -> list[DocumentNode]:
+        nodes: list[DocumentNode] = []
+        for token in tokens:
+            node = self.map_inline_token(token)
+            if node is not None:
+                nodes.append(node)
+        return nodes
+
+    def map_inline_token(self, token: dict) -> DocumentNode | None:
+        kind = token["type"]
+        if kind == "text":
+            return DocumentNode(type="text", node_id=self.next_id(), text=token.get("raw", ""))
+        if kind == "strong":
+            return DocumentNode(
+                type="strong", node_id=self.next_id(),
+                children=self.map_inline(token.get("children", [])),
+            )
+        if kind == "emphasis":
+            return DocumentNode(
+                type="emphasis", node_id=self.next_id(),
+                children=self.map_inline(token.get("children", [])),
+            )
+        if kind == "link":
+            attrs = token.get("attrs", {})
+            attributes = {"href": attrs.get("url", "")}
+            if attrs.get("title"):
+                attributes["title"] = attrs["title"]
+            return DocumentNode(
+                type="link", node_id=self.next_id(), attributes=attributes,
+                children=self.map_inline(token.get("children", [])),
+            )
+        if kind == "codespan":
+            return DocumentNode(type="codespan", node_id=self.next_id(), text=token.get("raw", ""))
+        if kind == "image":
+            # mistune 图片 token：src 在 attrs.url，alt 来自 children 的文本，title 在 attrs.title
+            attrs = token.get("attrs", {})
+            alt = "".join(
+                child.get("raw", "")
+                for child in token.get("children", [])
+                if child.get("type") == "text"
+            )
+            attributes = {"src": attrs.get("url", "")}
+            if alt:
+                attributes["alt"] = alt
+            if attrs.get("title"):
+                attributes["title"] = attrs["title"]
+            return DocumentNode(type="image", node_id=self.next_id(), attributes=attributes)
+        if kind == "inline_math":
+            return DocumentNode(
+                type="math_inline", node_id=self.next_id(), text=token.get("raw", "")
+            )
+        if kind == "softbreak":
+            # HTML 中换行会折叠为空白，软换行按空格表达
+            return DocumentNode(type="text", node_id=self.next_id(), text=" ")
+        if kind == "linebreak":
+            return DocumentNode(type="linebreak", node_id=self.next_id())
+        # 未知行内 token 保守保留原文
+        raw = token.get("raw", "")
+        if raw:
+            return DocumentNode(type="text", node_id=self.next_id(), text=raw)
+        return None
+
+    def _map_code(self, token: dict) -> DocumentNode:
+        info = (token.get("attrs", {}).get("info") or "").strip()
+        lang = info.split()[0].lower() if info else ""
+        code = token.get("raw", "").rstrip("\n")
+        if lang == _MERMAID_LANG:
+            return DocumentNode(type="mermaid", node_id=self.next_id(), text=code)
+        if lang in _FUNCTION_PLOT_LANGS:
+            return DocumentNode(type="function_plot", node_id=self.next_id(), text=code)
+        attributes = {"language": lang} if lang else {}
+        return DocumentNode(
+            type="code_block", node_id=self.next_id(), attributes=attributes, text=code
+        )
+
+    def _map_table(self, token: dict) -> DocumentNode:
+        rows: list[DocumentNode] = []
+        for child in token.get("children", []):
+            if child["type"] == "table_head":
+                rows.append(self._map_table_row(child, head=True))
+            elif child["type"] == "table_body":
+                for row in child.get("children", []):
+                    if row["type"] == "table_row":
+                        rows.append(self._map_table_row(row, head=False))
+            elif child["type"] == "table_row":
+                rows.append(self._map_table_row(child, head=False))
+        return DocumentNode(type="table", node_id=self.next_id(), children=rows)
+
+    def _map_table_row(self, token: dict, *, head: bool) -> DocumentNode:
+        cells: list[DocumentNode] = []
+        for cell in token.get("children", []):
+            if cell["type"] != "table_cell":
+                continue
+            attrs = cell.get("attrs", {})
+            cell_attributes = {"head": bool(attrs.get("head", head))}
+            if attrs.get("align"):
+                cell_attributes["align"] = attrs["align"]
+            cells.append(
+                DocumentNode(
+                    type="table_cell",
+                    node_id=self.next_id(),
+                    attributes=cell_attributes,
+                    children=self.map_inline(cell.get("children", [])),
+                )
+            )
+        return DocumentNode(
+            type="table_row", node_id=self.next_id(), attributes={"head": head}, children=cells
+        )

backend/app/export/service.py

@@ -0,0 +1,339 @@
+"""Export 服务：任务注册表、后台渲染、取消与产物生命周期。
+
+与 Benchmark 一致采用「创建即返回 queued、后台 Task 异步执行」的内存模型：任务与产物
+暂存内存与 exports 目录，不持久化到 SQLite。导出是单阶段渲染，无 SSE 事件流，取消主要
+在渲染前/后让出执行权的边界生效；产物带 24h 过期时间，过期后不可下载。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+import re
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from uuid import uuid4
+
+from app.config import get_settings
+from app.contracts import (
+    ExportFile,
+    ExportFormat,
+    ExportJob,
+    ExportOptions,
+    ExportProgress,
+    ExportRequest,
+    ExportSource,
+    ExportSourceType,
+    ExportStatus,
+)
+from app.errors import ApiError
+from app.export.document import Document, ExportResult
+from app.export.exporters.html import HtmlExporter
+from app.export.markdown import parse_document
+from app.services import note_service
+
+logger = logging.getLogger(__name__)
+
+_jobs: dict[str, ExportJob] = {}
+_tasks: dict[str, asyncio.Task] = {}
+_cancel_flags: dict[str, asyncio.Event] = {}
+MAX_JOBS = 100
+# 输入源（note / markdown）统一大小上限，防止未保存预览或超长笔记塞爆内存/产物
+MAX_MARKDOWN_CHARS = 200_000
+# 最终导出产物大小上限，防止超大 HTML 耗尽内存/磁盘
+MAX_EXPORT_BYTES = 20 * 1024 * 1024  # 20 MB
+# 并发渲染上限：解析/渲染是 CPU 密集的同步工作，限制同时执行的任务数，
+# 防止大量任务同时占满工作线程与内存
+MAX_CONCURRENT_RENDERS = 2
+_render_slots = asyncio.Semaphore(MAX_CONCURRENT_RENDERS)
+# 产物有效期
+FILE_TTL = timedelta(hours=24)
+
+_INVALID_FILE_CHARS = re.compile(r'[\\/:*?"<>|]')
+
+
+class ExportCancelled(Exception):
+    """导出在渲染前被取消时抛出，用于标记 cancelled。"""
+
+
+class ExportTooLarge(Exception):
+    """导出产物超过大小上限时抛出，用于标记 failed 并携带专用错误码。"""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _safe_download_name(title: str) -> str:
+    """清洗标题得到安全的下载文件名；空标题回退到 export。"""
+    name = _INVALID_FILE_CHARS.sub("_", title).strip() or "export"
+    return name[:80]
+
+
+def _export_path(job_id: str) -> Path:
+    return get_settings().exports_path / f"{job_id}.html"
+
+
+def _delete_file(job_id: str) -> None:
+    """删除导出产物文件；文件不存在时忽略。"""
+    try:
+        _export_path(job_id).unlink(missing_ok=True)
+    except OSError:
+        logger.warning("Failed to delete export file: %s", job_id)
+
+
+def cleanup_orphan_files() -> int:
+    """清理 exports 目录下无对应内存任务的孤立产物（服务重启后调用）。"""
+    exports_dir = get_settings().exports_path
+    if not exports_dir.is_dir():
+        return 0
+    removed = 0
+    for path in exports_dir.glob("*.html"):
+        if path.stem not in _jobs:
+            try:
+                path.unlink()
+                removed += 1
+            except OSError:
+                logger.warning("Failed to delete orphan export file: %s", path)
+    return removed
+
+
+def _render_document(document: Document, options: ExportOptions) -> ExportResult:
+    """同步渲染辅助，供 asyncio.to_thread 调用；每次新建实例避免跨线程复用。"""
+    return HtmlExporter().render(document, options)
+
+
+def _forget(job_id: str) -> None:
+    _jobs.pop(job_id, None)
+    _tasks.pop(job_id, None)
+    _cancel_flags.pop(job_id, None)
+    _delete_file(job_id)
+
+
+def _evict_terminal() -> bool:
+    """超过容量时淘汰最旧的终态任务；全为活动任务无法淘汰时返回 False。"""
+    terminal = (ExportStatus.completed, ExportStatus.failed, ExportStatus.cancelled)
+    while len(_jobs) >= MAX_JOBS:
+        victim = next((jid for jid, job in _jobs.items() if job.status in terminal), None)
+        if victim is None:
+            return False
+        _forget(victim)
+    return True
+
+
+async def _resolve_source(source: ExportSource) -> tuple[str, str, dict | None]:
+    """把导出源解析为 (markdown, title, metadata)；metadata 仅 note 源提供。"""
+    if source.type == ExportSourceType.note:
+        note = await note_service.get_note(source.note_id)
+        if note is None:
+            raise ApiError(
+                404,
+                "EXPORT_SOURCE_NOT_FOUND",
+                "note not found",
+                {"note_id": source.note_id},
+            )
+        if len(note.markdown) > MAX_MARKDOWN_CHARS:
+            raise ApiError(
+                400,
+                "EXPORT_OPTIONS_INVALID",
+                f"note source exceeds {MAX_MARKDOWN_CHARS} characters",
+                {"size": len(note.markdown), "limit": MAX_MARKDOWN_CHARS},
+            )
+        metadata = {
+            "file_path": note.file_path,
+            "tags": note.tags,
+            "created_at": note.created_at,
+            "updated_at": note.updated_at,
+        }
+        return note.markdown, note.title, metadata
+
+    markdown = source.markdown or ""
+    if not markdown.strip():
+        raise ApiError(400, "EXPORT_OPTIONS_INVALID", "markdown source must not be empty")
+    if len(markdown) > MAX_MARKDOWN_CHARS:
+        raise ApiError(
+            400,
+            "EXPORT_OPTIONS_INVALID",
+            f"markdown source exceeds {MAX_MARKDOWN_CHARS} characters",
+            {"size": len(markdown), "limit": MAX_MARKDOWN_CHARS},
+        )
+    return markdown, "", None
+
+
+async def create_export(request: ExportRequest) -> ExportJob:
+    """创建导出任务，立即返回 queued 的 ExportJob，由后台 Task 渲染。"""
+    if request.format != ExportFormat.html:
+        raise ApiError(
+            400,
+            "EXPORT_FORMAT_UNSUPPORTED",
+            "PDF/DOCX 暂未实现，当前仅支持 HTML",
+            {"format": request.format.value},
+        )
+    markdown, title, metadata = await _resolve_source(request.source)
+
+    if not _evict_terminal():
+        raise ApiError(
+            429,
+            "EXPORT_CAPACITY_EXCEEDED",
+            "Export capacity exceeded; wait for active jobs to finish.",
+            {},
+        )
+
+    job_id = "export_" + uuid4().hex[:12]
+    job = ExportJob(
+        job_id=job_id,
+        status=ExportStatus.queued,
+        format=request.format,
+        created_at=_now(),
+    )
+    _jobs[job_id] = job
+    _cancel_flags[job_id] = asyncio.Event()
+    _tasks[job_id] = asyncio.create_task(
+        _execute(job_id, markdown, title, metadata, request.options)
+    )
+    return job
+
+
+async def _execute(
+    job_id: str,
+    markdown: str,
+    title: str,
+    metadata: dict | None,
+    options: ExportOptions,
+) -> None:
+    """后台渲染：解析 → 导出 → 写文件 → 挂载产物元信息。"""
+    cancel_event = _cancel_flags[job_id]
+    _jobs[job_id] = _jobs[job_id].model_copy(
+        update={
+            "status": ExportStatus.running,
+            "started_at": _now(),
+            "progress": ExportProgress(phase="rendering", current=0, total=1, percent=0.0),
+        }
+    )
+    try:
+        # 并发渲染限额：解析/渲染是 CPU 密集的同步工作，用信号量限制同时执行的任务数，
+        # 超出限额的任务在此排队等待，避免大量任务同时占满工作线程与内存
+        async with _render_slots:
+            # 让出一次，使「创建后立即取消」的 queued 任务能及时进入 cancelled
+            await asyncio.sleep(0)
+            if cancel_event.is_set():
+                raise ExportCancelled()
+
+            # 解析与渲染都是 CPU 密集的同步工作，放入线程执行避免阻塞事件循环，
+            # 使运行中的取消能在渲染边界生效；写文件前再次检查取消。
+            document = await asyncio.to_thread(parse_document, markdown)
+            document.attributes["title"] = title
+            if metadata:
+                document.attributes["metadata"] = metadata
+
+            result = await asyncio.to_thread(_render_document, document, options)
+            if cancel_event.is_set():
+                raise ExportCancelled()
+            if len(result.content) > MAX_EXPORT_BYTES:
+                raise ExportTooLarge()
+
+            out_dir = get_settings().exports_path
+            out_dir.mkdir(parents=True, exist_ok=True)
+            path = _export_path(job_id)
+            path.write_bytes(result.content)
+
+            completed_at = _now()
+            _jobs[job_id] = _jobs[job_id].model_copy(
+                update={
+                    "status": ExportStatus.completed,
+                    "progress": ExportProgress(
+                        phase="completed", current=1, total=1, percent=1.0
+                    ),
+                    "file": ExportFile(
+                        file_name=f"{_safe_download_name(title)}.html",
+                        mime_type=result.mime_type,
+                        size=len(result.content),
+                        sha256=hashlib.sha256(result.content).hexdigest(),
+                        expires_at=completed_at + FILE_TTL,
+                    ),
+                    "warnings": result.warnings,
+                    "completed_at": completed_at,
+                }
+            )
+    except ExportCancelled:
+        _jobs[job_id] = _jobs[job_id].model_copy(
+            update={
+                "status": ExportStatus.cancelled,
+                "completed_at": _now(),
+            }
+        )
+    except ExportTooLarge:
+        _jobs[job_id] = _jobs[job_id].model_copy(
+            update={
+                "status": ExportStatus.failed,
+                "error": "Export output exceeds size limit.",
+                "error_code": "EXPORT_OUTPUT_TOO_LARGE",
+                "completed_at": _now(),
+            }
+        )
+    except Exception as exc:  # 渲染失败不拖垮服务，只记日志与项目错误码
+        logger.exception("Export failed: job_id=%s", job_id)
+        _jobs[job_id] = _jobs[job_id].model_copy(
+            update={
+                "status": ExportStatus.failed,
+                "error": "Export render failed.",
+                "error_code": "EXPORT_RENDER_FAILED",
+                "completed_at": _now(),
+            }
+        )
+    finally:
+        _cancel_flags.pop(job_id, None)
+
+
+def list_exports(
+    status: ExportStatus | None = None,
+    format: ExportFormat | None = None,
+    limit: int = 50,
+    offset: int = 0,
+) -> tuple[list[ExportJob], int]:
+    jobs = list(_jobs.values())
+    if status is not None:
+        jobs = [j for j in jobs if j.status == status]
+    if format is not None:
+        jobs = [j for j in jobs if j.format == format]
+    jobs.sort(key=lambda j: j.created_at, reverse=True)
+    total = len(jobs)
+    return jobs[offset : offset + limit], total
+
+
+def get_export(job_id: str) -> ExportJob | None:
+    return _jobs.get(job_id)
+
+
+def cancel_export(job_id: str) -> ExportJob | None:
+    """取消导出：仅 queued/running 可取消，后台 Task 在让出边界标记 cancelled。"""
+    job = _jobs.get(job_id)
+    if job is None:
+        return None
+    if job.status in (ExportStatus.queued, ExportStatus.running):
+        _cancel_flags[job_id].set()
+    return job
+
+
+def get_export_file(job_id: str) -> Path:
+    """返回可下载产物的存储路径；未完成返回 404、过期返回 410。"""
+    job = _jobs.get(job_id)
+    if job is None:
+        raise ApiError(404, "EXPORT_JOB_NOT_FOUND", "export job not found", {"job_id": job_id})
+    if job.status != ExportStatus.completed or job.file is None:
+        raise ApiError(
+            404, "EXPORT_JOB_NOT_FOUND", "export file not ready", {"job_id": job_id}
+        )
+    if job.file.expires_at <= _now():
+        _forget(job_id)  # 过期即清理内存记录与产物文件
+        raise ApiError(410, "EXPORT_FILE_EXPIRED", "export file has expired", {"job_id": job_id})
+    return _export_path(job_id)
+
+
+async def wait_for_export(job_id: str) -> ExportJob | None:
+    """等待后台任务结束（测试/轮询用）；无任务时直接返回当前状态。"""
+    task = _tasks.get(job_id)
+    if task is not None:
+        await task
+    return _jobs.get(job_id)

backend/app/main.py

@@ -8,6 +8,7 @@ from starlette.exceptions import HTTPException as StarletteHttpException
 from app.config import get_settings
 from app.container import container
 from app.errors import ApiError, api_error_handler, http_error_handler, validation_error_handler
+from app.export import service as export_service
 from app.routes import router as api_router
 from app.media_routes import router as media_router
 from app.local_model_routes import router as local_model_router
@@ -20,6 +21,8 @@ settings = get_settings()
 
 @asynccontextmanager
 async def lifespan(_: FastAPI):
+    # 重启后内存注册表为空，清理上一次运行遗留的导出产物，避免磁盘垃圾堆积。
+    export_service.cleanup_orphan_files()
     from app.services import transcription_service
     transcription_service.recover_interrupted()
     try:
@@ -31,6 +34,7 @@ async def lifespan(_: FastAPI):
         from app.local_models import manager
         for _, key in list(manager._downloads):
             await manager.cancel_download(key)
+        # 第三方 MCP Server 必须跟随 AI Core 退出，不能遗留孤儿进程。
         container.plugins.shutdown()
         container.mcp_servers.shutdown()
 

backend/app/plot/__init__.py

@@ -0,0 +1,7 @@
+"""Function Plot：函数图像的白名单表达式解析与静态 SVG 渲染。
+
+模块划分：
+- model.py    FunctionPlot 等内部数据模型（不进 contracts.py，同 Document AST）
+- parser.py   function-plot 源码与表达式解析（ast 白名单，绝不 eval/exec）
+- render.py   把 FunctionPlot 渲染为内嵌 SVG（纯几何 + <text>，无脚本）
+"""

backend/app/plot/model.py

@@ -0,0 +1,57 @@
+"""Function Plot 内部数据模型。
+
+契约 §12.2 的 FunctionPlot 结构与 §10.4 的 StaticRenderResult 只在导出链路的后端内部
+流转，不进入 HTTP 契约，因此与 Document AST 一样放在独立包内，不进 contracts.py。
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class FunctionPlotExpression(BaseModel):
+    """单条函数表达式；expression 为数学表达式文本（不含 ``y =`` 前缀）。"""
+
+    expression: str
+    label: str | None = None
+    color: str | None = None
+
+
+class PlotAxes(BaseModel):
+    xlabel: str | None = None
+    ylabel: str | None = None
+    grid: bool = True
+
+
+class FunctionPlot(BaseModel):
+    version: int = 1
+    expressions: list[FunctionPlotExpression]
+    domain: tuple[float, float] = (-10.0, 10.0)
+    range: tuple[float, float] | None = None
+    axes: PlotAxes = Field(default_factory=PlotAxes)
+    # 该块所有表达式 AST 节点数之和，供导出器做文档级累计复杂度预算
+    node_count: int = 0
+
+
+class PlotDiagnostic(BaseModel):
+    severity: Literal["warning", "error"]
+    code: str
+    message: str
+    line: int | None = None
+
+
+class FunctionPlotParseResult(BaseModel):
+    """解析结果：任一表达式 error 时 plot 为 None（整块回退占位），仅 warning 时 plot 有效。"""
+
+    plot: FunctionPlot | None = None
+    diagnostics: list[PlotDiagnostic] = Field(default_factory=list)
+
+
+class StaticRenderResult(BaseModel):
+    content: str
+    mime_type: str = "image/svg+xml"
+    width: int
+    height: int
+    warnings: list[str] = Field(default_factory=list)

backend/app/plot/parser.py

@@ -0,0 +1,412 @@
+"""Function Plot 表达式解析：白名单数学语法，绝不执行 eval / 函数构造器 / 属性访问。
+
+安全模型：先用 ``ast.parse(mode='eval')`` 把表达式变成纯 AST（这一步不执行任何代码），
+再逐节点白名单校验（只允许数字、变量 ``x``、常量 ``pi/e``、白名单函数调用与四则/幂
+运算），最后用递归解释器直接计算数值——全程不 ``compile``/``exec`` 字符串。
+"""
+
+from __future__ import annotations
+
+import ast
+import math
+import re
+from typing import NoReturn
+
+from app.plot.model import (
+    FunctionPlot,
+    FunctionPlotExpression,
+    FunctionPlotParseResult,
+    PlotAxes,
+    PlotDiagnostic,
+)
+
+# 白名单函数（ln 是 log 的别名）；abs 用内置函数，其余映射到 math
+_FUNCTION_IMPL: dict[str, object] = {
+    "sin": math.sin,
+    "cos": math.cos,
+    "tan": math.tan,
+    "asin": math.asin,
+    "acos": math.acos,
+    "atan": math.atan,
+    "sinh": math.sinh,
+    "cosh": math.cosh,
+    "tanh": math.tanh,
+    "exp": math.exp,
+    "log": math.log,
+    "ln": math.log,
+    "log10": math.log10,
+    "log2": math.log2,
+    "sqrt": math.sqrt,
+    "abs": abs,
+}
+_FUNCTIONS = frozenset(_FUNCTION_IMPL)
+_CONSTANTS: dict[str, float] = {"pi": math.pi, "e": math.e}
+
+_ALLOWED_BINOPS = (ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow)
+_ALLOWED_UNARY = (ast.UAdd, ast.USub)
+_DIRECTIVE_KEYS = frozenset({"domain", "range", "xlabel", "ylabel", "grid"})
+_NUMBER_RE = re.compile(r"^(\d+\.?\d*|\.\d+)([eE][+-]?\d+)?$")
+
+# 表达式复杂度上限：深层嵌套或海量节点在递归校验/求值时会触发 RecursionError，
+# 用白名单校验提前拦截，保证失败走正常诊断路径而不是异常逃逸出导出链路。
+_MAX_AST_DEPTH = 200
+_MAX_AST_NODES = 1000
+# 单块 function-plot 允许的表达式数量上限，防止海量表达式导致超大 SVG 与海量采样求值
+_MAX_EXPRESSIONS = 16
+
+
+class PlotParseError(Exception):
+    """表达式解析/校验失败，携带可定位诊断。"""
+
+    def __init__(self, diagnostic: PlotDiagnostic) -> None:
+        super().__init__(diagnostic.message)
+        self.diagnostic = diagnostic
+
+
+def _unsafe(message: str) -> NoReturn:
+    raise PlotParseError(
+        PlotDiagnostic(severity="error", code="FUNCTION_PLOT_EXPRESSION_UNSAFE", message=message)
+    )
+
+
+def _is_number(tok: str) -> bool:
+    return bool(_NUMBER_RE.match(tok))
+
+
+def _tokenize(s: str) -> list[str]:
+    """把预处理后的表达式切成数字/标识符/运算符/括号 token。"""
+    tokens: list[str] = []
+    i = 0
+    n = len(s)
+    while i < n:
+        ch = s[i]
+        if ch.isspace():
+            i += 1
+            continue
+        if ch.isdigit() or ch == ".":
+            j = i
+            while j < n and (s[j].isdigit() or s[j] == "."):
+                j += 1
+            # 科学计数法：数字后紧跟 e/E[+-]数字 视为同一数字
+            if j < n and s[j] in "eE":
+                k = j + 1
+                if k < n and s[k] in "+-":
+                    k += 1
+                if k < n and s[k].isdigit():
+                    while k < n and s[k].isdigit():
+                        k += 1
+                    j = k
+            tokens.append(s[i:j])
+            i = j
+            continue
+        if ch.isalpha() or ch == "_":
+            j = i
+            while j < n and (s[j].isalnum() or s[j] == "_"):
+                j += 1
+            tokens.append(s[i:j])
+            i = j
+            continue
+        if ch == "*" and i + 1 < n and s[i + 1] == "*":
+            tokens.append("**")
+            i += 2
+            continue
+        tokens.append(ch)
+        i += 1
+    return tokens
+
+
+def _is_value_end(tok: str) -> bool:
+    """该 token 之后允许补乘号（数字/右括号/变量 x/常量）。"""
+    return tok == ")" or _is_number(tok) or tok == "x" or tok in _CONSTANTS
+
+
+def _is_value_start(tok: str) -> bool:
+    """该 token 可作为乘号右侧起点（左括号/数字/任意标识符，含函数名）。"""
+    return tok == "(" or _is_number(tok) or (tok and (tok[0].isalpha() or tok[0] == "_"))
+
+
+def _insert_implicit_multiplication(s: str) -> str:
+    """补隐式乘法：2x、2(x+1)、(x+1)(x-1)、x sin(x) 等；函数名后的 ``(`` 是调用不补。"""
+    tokens = _tokenize(s)
+    out: list[str] = []
+    prev: str | None = None
+    for tok in tokens:
+        if prev is not None and _is_value_end(prev) and _is_value_start(tok):
+            out.append("*")
+        out.append(tok)
+        prev = tok
+    return "".join(out)
+
+
+def _preprocess(expr: str) -> str:
+    """``^`` 视为幂，补隐式乘法后再交给 ast.parse。"""
+    return _insert_implicit_multiplication(expr.replace("^", "**"))
+
+
+def _check_node(node: ast.AST, depth: int = 0, counter: list[int] | None = None) -> None:
+    """白名单校验：任何越界节点都抛 FUNCTION_PLOT_EXPRESSION_UNSAFE。
+
+    同时限制 AST 深度与节点总数，避免超长/超深表达式在递归校验或求值时触发
+    RecursionError 而绕过解析失败路径。
+    """
+    if counter is None:
+        counter = [0]
+    if depth > _MAX_AST_DEPTH:
+        _unsafe(f"表达式嵌套过深（超过 {_MAX_AST_DEPTH} 层）")
+    counter[0] += 1
+    if counter[0] > _MAX_AST_NODES:
+        _unsafe(f"表达式过于复杂（节点数超过 {_MAX_AST_NODES}）")
+    if isinstance(node, ast.Constant):
+        if isinstance(node.value, bool) or not isinstance(node.value, (int, float)):
+            _unsafe(f"不支持的常量 {node.value!r}")
+        return
+    if isinstance(node, ast.Name):
+        if node.id == "x" or node.id in _CONSTANTS:
+            return
+        _unsafe(f"未知标识符 {node.id!r}")
+    if isinstance(node, ast.BinOp):
+        if not isinstance(node.op, _ALLOWED_BINOPS):
+            _unsafe(f"不支持的运算符 {type(node.op).__name__}")
+        _check_node(node.left, depth + 1, counter)
+        _check_node(node.right, depth + 1, counter)
+        return
+    if isinstance(node, ast.UnaryOp):
+        if not isinstance(node.op, _ALLOWED_UNARY):
+            _unsafe(f"不支持的运算符 {type(node.op).__name__}")
+        _check_node(node.operand, depth + 1, counter)
+        return
+    if isinstance(node, ast.Call):
+        if not isinstance(node.func, ast.Name) or node.func.id not in _FUNCTIONS:
+            _unsafe(f"不支持的函数调用 {ast.dump(node.func)!r}")
+        if node.keywords:
+            _unsafe("函数调用不支持关键字参数")
+        # 白名单内所有函数均恰取 1 个参数，提前校验避免求值期 TypeError
+        if len(node.args) != 1:
+            _unsafe(f"{node.func.id} 需要 1 个参数，实际 {len(node.args)} 个")
+        for arg in node.args:
+            _check_node(arg, depth + 1, counter)
+        return
+    _unsafe(f"不支持的语法 {type(node).__name__}")
+
+
+def parse_expression(expr: str) -> ast.Expression:
+    """把数学表达式解析为已通过白名单校验的 AST（可直接交给 evaluate）。"""
+    preprocessed = _preprocess(expr)
+    try:
+        tree = ast.parse(preprocessed, mode="eval")
+    except SyntaxError as exc:
+        raise PlotParseError(
+            PlotDiagnostic(
+                severity="error",
+                code="FUNCTION_PLOT_PARSE_FAILED",
+                message=f"表达式语法错误：{exc.msg}",
+            )
+        ) from exc
+    except RecursionError as exc:
+        # 极深嵌套可能在 ast.parse 阶段就触发 RecursionError，转为可定位诊断
+        raise PlotParseError(
+            PlotDiagnostic(
+                severity="error",
+                code="FUNCTION_PLOT_PARSE_FAILED",
+                message="表达式嵌套过深，无法解析",
+            )
+        ) from exc
+    _check_node(tree.body)
+    return tree
+
+
+def _count_nodes(node: ast.AST) -> int:
+    """统计已通过校验的表达式 AST 节点数，供文档级累计复杂度预算使用。"""
+    counter = [0]
+    _check_node(node, counter=counter)
+    return counter[0]
+
+
+def evaluate(expr_ast: ast.Expression, x: float) -> float:
+    """递归解释已校验 AST 得到数值，全程不编译/执行代码。"""
+    return _eval_node(expr_ast.body, x)
+
+
+def _eval_node(node: ast.AST, x: float) -> float:
+    if isinstance(node, ast.Constant):
+        return float(node.value)
+    if isinstance(node, ast.Name):
+        return x if node.id == "x" else _CONSTANTS[node.id]
+    if isinstance(node, ast.BinOp):
+        left = _eval_node(node.left, x)
+        right = _eval_node(node.right, x)
+        if isinstance(node.op, ast.Add):
+            return left + right
+        if isinstance(node.op, ast.Sub):
+            return left - right
+        if isinstance(node.op, ast.Mult):
+            return left * right
+        if isinstance(node.op, ast.Div):
+            return left / right
+        # 负数底 + 非整数指数会得到复数，数学绘图不支持，抛 ValueError 让采样点作为断点处理
+        if left < 0 and not right.is_integer():
+            raise ValueError("negative base with fractional exponent")
+        return left**right
+    if isinstance(node, ast.UnaryOp):
+        value = _eval_node(node.operand, x)
+        return -value if isinstance(node.op, ast.USub) else value
+    if isinstance(node, ast.Call):
+        args = [_eval_node(arg, x) for arg in node.args]
+        return _FUNCTION_IMPL[node.func.id](*args)  # type: ignore[operator]
+    raise ValueError("unreachable node")
+
+
+def _strip_comment(line: str) -> str:
+    return line.split("#", 1)[0].strip()
+
+
+def _parse_pair(value: str) -> tuple[float, float]:
+    """解析 ``min, max`` / ``min max`` 数值对。"""
+    parts = [p for p in re.split(r"[,，\s]+", value.strip()) if p]
+    if len(parts) != 2:
+        raise ValueError("需要两个数值")
+    return float(parts[0]), float(parts[1])
+
+
+def _parse_directive(line: str) -> tuple[str, str] | None:
+    """指令行形如 ``key: value``（表达式不含冒号，冒号是可靠判别）。"""
+    if ":" not in line or "=" in line:
+        return None
+    key, _, value = line.partition(":")
+    key = key.strip().lower()
+    if not key or " " in key:
+        return None
+    return key, value.strip()
+
+
+def parse_source(source: str) -> FunctionPlotParseResult:
+    """把 function-plot fenced block 源码解析为 FunctionPlot + 诊断。"""
+    diagnostics: list[PlotDiagnostic] = []
+    expressions: list[FunctionPlotExpression] = []
+    domain: tuple[float, float] = (-10.0, 10.0)
+    range_: tuple[float, float] | None = None
+    xlabel: str | None = None
+    ylabel: str | None = None
+    grid: bool = True
+    has_error = False
+    total_nodes = 0
+
+    for lineno, raw_line in enumerate(source.splitlines(), start=1):
+        line = raw_line.strip()
+        if not line or line.startswith("#"):
+            continue
+
+        directive = _parse_directive(line)
+        if directive is not None:
+            key, value = directive
+            if key == "domain":
+                try:
+                    domain = _parse_pair(value)
+                except ValueError:
+                    diagnostics.append(
+                        PlotDiagnostic(
+                            severity="warning",
+                            code="FUNCTION_PLOT_PARSE_FAILED",
+                            message=f"domain 需要两个数值，已忽略：{value!r}",
+                            line=lineno,
+                        )
+                    )
+            elif key == "range":
+                try:
+                    range_ = _parse_pair(value)
+                except ValueError:
+                    diagnostics.append(
+                        PlotDiagnostic(
+                            severity="warning",
+                            code="FUNCTION_PLOT_PARSE_FAILED",
+                            message=f"range 需要两个数值，已忽略：{value!r}",
+                            line=lineno,
+                        )
+                    )
+            elif key == "xlabel":
+                xlabel = value or None
+            elif key == "ylabel":
+                ylabel = value or None
+            elif key == "grid":
+                grid = value.lower() in ("true", "1", "yes", "on")
+            else:
+                diagnostics.append(
+                    PlotDiagnostic(
+                        severity="warning",
+                        code="FUNCTION_PLOT_PARSE_FAILED",
+                        message=f"未知指令 {key!r} 已忽略",
+                        line=lineno,
+                    )
+                )
+            continue
+
+        # 表达式行：y = <expr> 或裸 <expr>
+        expr_text = _strip_comment(line)
+        if not expr_text:
+            continue
+        if "=" in expr_text:
+            lhs, _, rhs = expr_text.partition("=")
+            if lhs.strip().lower() not in ("y", ""):
+                diagnostics.append(
+                    PlotDiagnostic(
+                        severity="error",
+                        code="FUNCTION_PLOT_PARSE_FAILED",
+                        message="表达式应形如 'y = <expr>'",
+                        line=lineno,
+                    )
+                )
+                has_error = True
+                continue
+            expr_text = rhs.strip()
+        if not expr_text:
+            diagnostics.append(
+                PlotDiagnostic(
+                    severity="error",
+                    code="FUNCTION_PLOT_PARSE_FAILED",
+                    message="表达式为空",
+                    line=lineno,
+                )
+            )
+            has_error = True
+            continue
+
+        try:
+            tree = parse_expression(expr_text)
+        except PlotParseError as exc:
+            exc.diagnostic.line = lineno
+            diagnostics.append(exc.diagnostic)
+            has_error = True
+            continue
+        total_nodes += _count_nodes(tree.body)
+        expressions.append(FunctionPlotExpression(expression=expr_text))
+        # 表达式数量超限：整块回退并提前终止，避免对海量表达式做采样求值
+        if len(expressions) > _MAX_EXPRESSIONS:
+            diagnostics.append(
+                PlotDiagnostic(
+                    severity="error",
+                    code="FUNCTION_PLOT_TOO_MANY_EXPRESSIONS",
+                    message=f"表达式数量超过上限 {_MAX_EXPRESSIONS}，已回退为源码占位",
+                )
+            )
+            return FunctionPlotParseResult(plot=None, diagnostics=diagnostics)
+
+    if has_error:
+        return FunctionPlotParseResult(plot=None, diagnostics=diagnostics)
+    if not expressions:
+        diagnostics.append(
+            PlotDiagnostic(
+                severity="error",
+                code="FUNCTION_PLOT_PARSE_FAILED",
+                message="没有找到任何函数表达式",
+            )
+        )
+        return FunctionPlotParseResult(plot=None, diagnostics=diagnostics)
+
+    plot = FunctionPlot(
+        expressions=expressions,
+        domain=domain,
+        range=range_,
+        axes=PlotAxes(xlabel=xlabel, ylabel=ylabel, grid=grid),
+        node_count=total_nodes,
+    )
+    return FunctionPlotParseResult(plot=plot, diagnostics=diagnostics)

backend/app/plot/render.py

@@ -0,0 +1,258 @@
+"""Function Plot → 静态 SVG 渲染。
+
+只输出纯几何与 <text> 的 SVG（无 script/foreignObject/内联事件），可安全内嵌 HTML。
+所有文本与颜色都经过转义/校验，不把用户输入直接拼进标记。
+"""
+
+from __future__ import annotations
+
+import html
+import math
+import re
+from typing import Callable
+
+from app.plot.model import FunctionPlot, StaticRenderResult
+from app.plot.parser import PlotParseError, evaluate, parse_expression
+
+_WIDTH = 640
+_HEIGHT = 480
+_MARGIN = 52  # 四周留白，放轴刻度与标签
+_SAMPLES = 400
+_PALETTE = ["#0969da", "#d1242f", "#1a7f37", "#8250df", "#bf8700", "#e36209"]
+_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{3,8}$")
+
+
+def _safe_color(color: str | None, fallback: str) -> str:
+    return color.strip() if color and _COLOR_RE.match(color.strip()) else fallback
+
+
+def _valid_span(lo: float, hi: float) -> bool:
+    """范围跨度有效：端点有限、跨度有限且大于零。
+
+    端点相减可能溢出为 ``inf``（如 ``-1e308`` 到 ``1e308``），需单独校验跨度，
+    否则后续坐标换算会生成含 ``nan`` 的 SVG。
+    """
+    span = hi - lo
+    return math.isfinite(lo) and math.isfinite(hi) and math.isfinite(span) and span > 0
+
+
+def _fmt_num(v: float) -> str:
+    if v == 0:
+        return "0"
+    if abs(v) >= 1e6 or abs(v) < 1e-6:
+        return f"{v:.2e}"
+    return f"{v:.6g}"
+
+
+def _nice_step(span: float, target_ticks: int = 6) -> float:
+    raw = abs(span) / target_ticks
+    if not math.isfinite(raw) or raw <= 0:
+        return 1.0  # 兜底步长，避免 span 为 0/inf 时产生非法刻度
+    mag = 10 ** math.floor(math.log10(raw))
+    for m in (1, 2, 5, 10):
+        if raw <= m * mag:
+            return m * mag
+    return 10 * mag
+
+
+def _ticks(lo: float, hi: float, step: float) -> list[float]:
+    # 防御：非法步长直接返回空，避免除零
+    if not math.isfinite(step) or step <= 0:
+        return []
+    first = math.ceil(lo / step) * step
+    values: list[float] = []
+    v = first
+    # 有上限的整数索引推进 + 步长推进校验，防止浮点精度导致 v+step==v 的死循环
+    for _ in range(1000):
+        if v > hi + step * 1e-9:
+            break
+        values.append(v)
+        nxt = v + step
+        if nxt <= v:
+            break  # 步长小于当前数值的浮点精度，已无法推进
+        v = nxt
+    return values
+
+
+def _compute_range(
+    fns: list[tuple[object, object]],
+    xmin: float,
+    xmax: float,
+) -> tuple[float, float]:
+    """采样确定 y 范围；取有限样本的 min/max 加 5% 余量。"""
+    ys: list[float] = []
+    for _expr, tree in fns:
+        for i in range(_SAMPLES + 1):
+            x = xmin + (xmax - xmin) * i / _SAMPLES
+            try:
+                y = evaluate(tree, x)  # type: ignore[arg-type]
+            except (ValueError, ZeroDivisionError, OverflowError, TypeError):
+                continue
+            # 复数等非实数结果直接跳过，不参与范围统计
+            if isinstance(y, (int, float)) and math.isfinite(y):
+                ys.append(y)
+
+    if not ys:
+        return -10.0, 10.0
+    lo, hi = min(ys), max(ys)
+    if lo == hi:
+        lo -= 1.0
+        hi += 1.0
+    pad = (hi - lo) * 0.05
+    return lo - pad, hi + pad
+
+
+def _polyline(
+    tree: object,
+    xmin: float,
+    xmax: float,
+    sx: Callable[[float], float],
+    sy: Callable[[float], float],
+    color: str,
+) -> str:
+    """采样并把非有限点处断开成多段 polyline，避免画穿渐近线。"""
+    segments: list[str] = []
+    points: list[str] = []
+    for i in range(_SAMPLES + 1):
+        x = xmin + (xmax - xmin) * i / _SAMPLES
+        try:
+            y = evaluate(tree, x)  # type: ignore[arg-type]
+        except (ValueError, ZeroDivisionError, OverflowError, TypeError):
+            y = math.nan
+        if not isinstance(y, (int, float)) or not math.isfinite(y):
+            if points:
+                segments.append(f'<polyline points="{" ".join(points)}" fill="none" stroke="{color}"/>')
+                points = []
+            continue
+        px = sx(x)
+        py = sy(y)
+        # 映射后的坐标必须有限：显式 range 下极端 y 值可能让像素坐标溢出为 inf
+        if not (math.isfinite(px) and math.isfinite(py)):
+            if points:
+                segments.append(f'<polyline points="{" ".join(points)}" fill="none" stroke="{color}"/>')
+                points = []
+            continue
+        points.append(f"{px:.2f},{py:.2f}")
+    if points:
+        segments.append(f'<polyline points="{" ".join(points)}" fill="none" stroke="{color}"/>')
+    return "".join(segments)
+
+
+def _grid(
+    xmin: float,
+    xmax: float,
+    ymin: float,
+    ymax: float,
+    sx: Callable[[float], float],
+    sy: Callable[[float], float],
+) -> str:
+    parts: list[str] = []
+    for x in _ticks(xmin, xmax, _nice_step(xmax - xmin)):
+        parts.append(f'<line x1="{sx(x):.2f}" y1="{sy(ymin):.2f}" x2="{sx(x):.2f}" y2="{sy(ymax):.2f}" stroke="#eaeef2"/>')
+    for y in _ticks(ymin, ymax, _nice_step(ymax - ymin)):
+        parts.append(f'<line x1="{sx(xmin):.2f}" y1="{sy(y):.2f}" x2="{sx(xmax):.2f}" y2="{sy(y):.2f}" stroke="#eaeef2"/>')
+    return "".join(parts)
+
+
+def _axes(
+    xmin: float,
+    xmax: float,
+    ymin: float,
+    ymax: float,
+    sx: Callable[[float], float],
+    sy: Callable[[float], float],
+) -> str:
+    parts: list[str] = []
+    # 坐标轴：过原点则画在原点，否则贴边，保证始终有参照系
+    x_axis_y = 0.0 if ymin <= 0 <= ymax else ymin
+    y_axis_x = 0.0 if xmin <= 0 <= xmax else xmin
+    parts.append(
+        f'<line x1="{sx(xmin):.2f}" y1="{sy(x_axis_y):.2f}" x2="{sx(xmax):.2f}" y2="{sy(x_axis_y):.2f}" stroke="#57606a"/>'
+    )
+    parts.append(
+        f'<line x1="{sx(y_axis_x):.2f}" y1="{sy(ymin):.2f}" x2="{sx(y_axis_x):.2f}" y2="{sy(ymax):.2f}" stroke="#57606a"/>'
+    )
+    # x 轴刻度数字（画在轴下方）
+    for x in _ticks(xmin, xmax, _nice_step(xmax - xmin)):
+        parts.append(
+            f'<text x="{sx(x):.2f}" y="{sy(x_axis_y) + 14:.2f}" text-anchor="middle" font-size="10" fill="#57606a">{html.escape(_fmt_num(x))}</text>'
+        )
+    # y 轴刻度数字（画在轴左侧）
+    for y in _ticks(ymin, ymax, _nice_step(ymax - ymin)):
+        parts.append(
+            f'<text x="{sx(y_axis_x) - 6:.2f}" y="{sy(y) + 3:.2f}" text-anchor="end" font-size="10" fill="#57606a">{html.escape(_fmt_num(y))}</text>'
+        )
+    return "".join(parts)
+
+
+def _labels(plot: FunctionPlot, sx: Callable[[float], float], sy: Callable[[float], float]) -> str:
+    parts: list[str] = []
+    if plot.axes.xlabel:
+        parts.append(
+            f'<text x="{(_WIDTH / 2):.2f}" y="{_HEIGHT - 10:.2f}" text-anchor="middle" font-size="12" fill="#1f2328">{html.escape(plot.axes.xlabel)}</text>'
+        )
+    if plot.axes.ylabel:
+        parts.append(
+            f'<text x="16" y="{(_HEIGHT / 2):.2f}" text-anchor="middle" font-size="12" fill="#1f2328" transform="rotate(-90 16 {_HEIGHT / 2:.2f})">{html.escape(plot.axes.ylabel)}</text>'
+        )
+    return "".join(parts)
+
+
+def render_svg(plot: FunctionPlot) -> StaticRenderResult:
+    """把已解析的 FunctionPlot 渲染为内嵌 SVG。"""
+    warnings: list[str] = []
+    xmin, xmax = plot.domain
+    if not _valid_span(xmin, xmax):
+        warnings.append("domain 无效，回退到 [-10, 10]")
+        xmin, xmax = -10.0, 10.0
+
+    # 重新解析并编译表达式（parse_source 已校验，这里异常只在模型被绕过时触发）
+    fns: list[tuple[object, object]] = []
+    for expr in plot.expressions:
+        try:
+            tree = parse_expression(expr.expression)
+        except PlotParseError as exc:
+            warnings.append(f"表达式无法渲染，已跳过：{expr.expression}（{exc.diagnostic.message}）")
+            continue
+        fns.append((expr, tree))
+
+    # 纵轴范围：显式 range 有效则用之；无效（退化/非有限/跨度溢出）丢弃并自动采样重算
+    if plot.range is not None:
+        lo, hi = float(plot.range[0]), float(plot.range[1])
+        if _valid_span(lo, hi):
+            ymin, ymax = lo, hi
+        else:
+            warnings.append("range 无效，改用自动范围")
+            ymin, ymax = _compute_range(fns, xmin, xmax)
+    else:
+        ymin, ymax = _compute_range(fns, xmin, xmax)
+
+    # 最终防线：自动范围在极端样本下也可能溢出，坐标映射前必须保证跨度有限且大于零
+    if not _valid_span(ymin, ymax):
+        warnings.append("y 范围跨度无法表示，回退到 [-10, 10]")
+        ymin, ymax = -10.0, 10.0
+
+    def sx(x: float) -> float:
+        return _MARGIN + (x - xmin) / (xmax - xmin) * (_WIDTH - 2 * _MARGIN)
+
+    def sy(y: float) -> float:
+        return _HEIGHT - _MARGIN - (y - ymin) / (ymax - ymin) * (_HEIGHT - 2 * _MARGIN)
+
+    parts: list[str] = [
+        f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {_WIDTH} {_HEIGHT}" role="img">'
+    ]
+    if plot.axes.grid:
+        parts.append(_grid(xmin, xmax, ymin, ymax, sx, sy))
+    parts.append(_axes(xmin, xmax, ymin, ymax, sx, sy))
+    for i, (expr, tree) in enumerate(fns):
+        color = _safe_color(expr.color, _PALETTE[i % len(_PALETTE)])
+        parts.append(_polyline(tree, xmin, xmax, sx, sy, color))
+    parts.append(_labels(plot, sx, sy))
+    parts.append("</svg>")
+
+    return StaticRenderResult(
+        content="".join(parts),
+        width=_WIDTH,
+        height=_HEIGHT,
+        warnings=warnings,
+    )

backend/app/routes.py

@@ -6,7 +6,7 @@ from datetime import datetime, timezone
 from uuid import uuid4
 
 from fastapi import APIRouter, Header, Query
-from fastapi.responses import StreamingResponse
+from fastapi.responses import FileResponse, StreamingResponse
 
 from app.agent import AgentCapacityError, AgentRunNotFoundError
 from app.container import container
@@ -53,6 +53,11 @@ from app.contracts import (
     ModelRoutingResponse,
     SpeakerMatchRequest,
     SpeakerMatchResult,
+    ExportFormat,
+    ExportJob,
+    ExportJobListResponse,
+    ExportRequest,
+    ExportStatus,
     Note,
     NoteCreateRequest,
     NoteListResponse,
@@ -101,8 +106,10 @@ from app.contracts import (
 from app.agent import AgentCapacityError, AgentRunNotFoundError
 from app.benchmarks import datasets as benchmark_datasets
 from app.benchmarks import service as benchmark_service
+from app.config import get_settings
 from app.container import container
 from app.errors import ApiError
+from app.export import service as export_service
 from app.extensions import ExtensionError
 from app.extensions.mcp_registry import McpRegistryError
 from app.providers.base import ProviderError
@@ -1468,3 +1475,80 @@ async def get_benchmark_report(run_id: str) -> BenchmarkReport:
             404, "BENCHMARK_RUN_NOT_FOUND", "benchmark report not found", {"run_id": run_id}
         )
     return report
+
+
+@router.post(
+    "/exports",
+    response_model=ExportJob,
+    status_code=202,
+    tags=["Export"],
+)
+async def create_export(request: ExportRequest) -> ExportJob:
+    return await export_service.create_export(request)
+
+
+@router.get(
+    "/exports",
+    response_model=ExportJobListResponse,
+    tags=["Export"],
+)
+async def list_exports(
+    status: ExportStatus | None = Query(default=None),
+    format: ExportFormat | None = Query(default=None),
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
+) -> ExportJobListResponse:
+    items, total = export_service.list_exports(
+        status=status, format=format, limit=limit, offset=offset
+    )
+    return ExportJobListResponse(
+        items=items, page=PageMeta(total=total, limit=limit, offset=offset)
+    )
+
+
+@router.get(
+    "/exports/{job_id}",
+    response_model=ExportJob,
+    tags=["Export"],
+)
+async def get_export(job_id: str) -> ExportJob:
+    job = export_service.get_export(job_id)
+    if job is None:
+        raise ApiError(
+            404, "EXPORT_JOB_NOT_FOUND", "export job not found", {"job_id": job_id}
+        )
+    return job
+
+
+@router.get(
+    "/exports/{job_id}/file",
+    tags=["Export"],
+)
+async def get_export_file(job_id: str) -> FileResponse:
+    path = export_service.get_export_file(job_id)  # 未完成/过期分别抛 404/410
+    job = export_service.get_export(job_id)
+    if job is None or job.file is None:
+        raise ApiError(
+            404, "EXPORT_JOB_NOT_FOUND", "export file not ready", {"job_id": job_id}
+        )
+    return FileResponse(
+        path=path,
+        media_type=job.file.mime_type,
+        filename=job.file.file_name,
+    )
+
+
+@router.post(
+    "/exports/{job_id}/cancel",
+    response_model=OperationResponse,
+    tags=["Export"],
+)
+async def cancel_export(job_id: str) -> OperationResponse:
+    job = export_service.cancel_export(job_id)
+    if job is None:
+        raise ApiError(
+            404, "EXPORT_JOB_NOT_FOUND", "export job not found", {"job_id": job_id}
+        )
+    return OperationResponse(
+        status="accepted", resource_id=job_id, message="Export cancellation accepted."
+    )

backend/pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
     "fastapi>=0.116,<1.0",
     "httpx>=0.28,<1.0",
     "jsonschema>=4.25,<5.0",
+    "mistune>=3.0,<4.0",
     "pyyaml>=6.0,<7.0",
     "referencing>=0.36,<1.0",
     "sqlite-vec>=0.1.9",

backend/tests/test_export.py

@@ -0,0 +1,485 @@
+"""Export Service 的单元与端到端测试。
+
+沿用 conftest 隔离机制：APP_DATA_DIR / DB / Vault / exports 目录都落在临时目录，
+不读写真实数据。导出采用「创建即 queued + 后台 Task 执行」的异步模型，测试在同一
+事件循环内创建并等待后台任务结束，得到终态 ExportJob 后再断言。
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from pydantic import ValidationError
+
+from app.config import get_settings
+from app.contracts import (
+    ExportFormat,
+    ExportJob,
+    ExportOptions,
+    ExportRequest,
+    ExportSource,
+    ExportSourceType,
+    ExportStatus,
+)
+from app.errors import ApiError
+from app.export import service as export_service
+from app.export.exporters.html import HtmlExporter
+from app.export.markdown import parse_document
+
+MD = """# 进程调度
+
+一些 **加粗** 和 *斜体*，[链接](https://a.b) 与 `code`。
+
+- 项目一
+- 项目二
+
+```python
+print(1)
+```
+
+```mermaid
+graph LR
+```
+
+```function_plot
+y = x
+```
+
+| a | b |
+|---|---|
+| 1 | 2 |
+
+行内 $x^2$ 与块级
+$$
+y = mx + b
+$$
+"""
+
+
+@pytest.fixture(autouse=True)
+def _reset_export_state():
+    """清空内存注册表，避免跨用例的任务/取消标志互相污染。"""
+    export_service._jobs.clear()
+    export_service._tasks.clear()
+    export_service._cancel_flags.clear()
+    yield
+    export_service._jobs.clear()
+    export_service._tasks.clear()
+    export_service._cancel_flags.clear()
+
+
+def _create_and_wait(request: ExportRequest) -> object:
+    """创建导出并在同一事件循环内等待后台任务结束，返回终态 ExportJob。"""
+
+    async def _execute():
+        job = await export_service.create_export(request)
+        return await export_service.wait_for_export(job.job_id)
+
+    return asyncio.run(_execute())
+
+
+# --------------------------------------------------------------------------- #
+# markdown → Document AST
+# --------------------------------------------------------------------------- #
+def _types(nodes) -> list[str]:
+    return [n.type for n in nodes]
+
+
+def test_parse_document_heading_and_inline() -> None:
+    doc = parse_document("# 标题\n\n一段 **加粗** 和 [链接](https://a.b)。")
+
+    assert doc.type == "document"
+    heading = doc.children[0]
+    assert heading.type == "heading"
+    assert heading.attributes["level"] == 1
+
+    para = doc.children[1]
+    assert para.type == "paragraph"
+    kinds = _types(para.children)
+    assert "text" in kinds
+    assert "strong" in kinds
+    assert "link" in kinds
+
+    link = next(c for c in para.children if c.type == "link")
+    assert link.attributes["href"] == "https://a.b"
+
+
+def test_parse_document_list_and_code_fencing() -> None:
+    doc = parse_document("- a\n- b\n\n```mermaid\ngraph LR\n```\n\n```function_plot\ny=x\n```\n\n```python\nx\n```")
+
+    kinds = [c.type for c in doc.children]
+    assert kinds[0] == "list"
+    assert kinds[1] == "mermaid"
+    assert kinds[2] == "function_plot"
+    assert kinds[3] == "code_block"
+
+    code = doc.children[3]
+    assert code.attributes["language"] == "python"
+    assert code.text == "x"
+
+
+def test_parse_document_table_and_math() -> None:
+    doc = parse_document("| a | b |\n|---|---|\n| 1 | 2 |\n\n$x^2$\n\n$$\ny=mx\n$$")
+
+    table = doc.children[0]
+    assert table.type == "table"
+    assert table.children[0].type == "table_row"
+    assert table.children[0].children[0].attributes["head"] is True
+
+    # 表格后是「行内数学所在段落」与「块级数学」
+    kinds = [c.type for c in doc.children[1:]]
+    assert "paragraph" in kinds
+    assert "math_block" in kinds
+
+
+def test_parse_document_image_maps_src_alt_title() -> None:
+    doc = parse_document('![替代文本](https://a.b/img.png "标题")')
+    img = doc.children[0].children[0]
+    assert img.type == "image"
+    assert img.attributes["src"] == "https://a.b/img.png"
+    assert img.attributes["alt"] == "替代文本"
+    assert img.attributes["title"] == "标题"
+
+
+def test_parse_document_function_plot_dash_alias() -> None:
+    doc = parse_document("```function-plot\ny = x^2\n```")
+    assert doc.children[0].type == "function_plot"
+    assert doc.children[0].text == "y = x^2"
+
+
+# --------------------------------------------------------------------------- #
+# HtmlExporter
+# --------------------------------------------------------------------------- #
+async def _render(markdown: str, *, title: str = "") -> str:
+    doc = parse_document(markdown)
+    doc.attributes["title"] = title
+    result = await HtmlExporter().export(doc, ExportOptions())
+    return result.content.decode("utf-8")
+
+
+def test_html_exporter_renders_basic_nodes_and_escapes() -> None:
+    html = asyncio.run(_render("# 标题\n\n**加粗** [链接](https://a.b) 与 <b>原始</b>。"))
+
+    assert "<h1>标题</h1>" in html
+    assert "<strong>加粗</strong>" in html
+    assert '<a href="https://a.b">链接</a>' in html
+    # 原始 HTML 必须被转义，不能注入文档
+    assert "&lt;b&gt;原始&lt;/b&gt;" in html
+    assert "<b>原始</b>" not in html
+
+
+def test_html_exporter_marks_mermaid_and_function_plot() -> None:
+    result = asyncio.run(HtmlExporter().export(parse_document("```mermaid\ngraph LR\n```"), ExportOptions()))
+
+    html = result.content.decode("utf-8")
+    assert '<pre class="mermaid">graph LR</pre>' in html
+    assert any("mermaid" in w for w in result.warnings)
+
+
+def test_html_exporter_rejects_unsafe_link_protocol() -> None:
+    result = asyncio.run(
+        HtmlExporter().export(parse_document("[点我](javascript:alert(1))"), ExportOptions())
+    )
+    html = result.content.decode("utf-8")
+    assert "javascript:" not in html
+    assert "点我" in html
+    assert any("不安全" in w for w in result.warnings)
+
+
+def test_html_exporter_rejects_unsafe_image_protocol() -> None:
+    result = asyncio.run(
+        HtmlExporter().export(parse_document("![alt](data:text/html,<script>)"), ExportOptions())
+    )
+    html = result.content.decode("utf-8")
+    assert "data:" not in html
+    assert "<img" not in html
+    assert "alt" in html
+    assert any("不安全" in w for w in result.warnings)
+
+
+def test_html_exporter_preserves_raw_html_block() -> None:
+    result = asyncio.run(
+        HtmlExporter().export(parse_document("<div>重要正文</div>"), ExportOptions())
+    )
+    html = result.content.decode("utf-8")
+    assert "重要正文" in html
+    assert "<div>" not in html
+    assert "&lt;div&gt;重要正文&lt;/div&gt;" in html
+    assert any("原始 HTML" in w for w in result.warnings)
+
+
+def test_html_exporter_include_title_and_metadata() -> None:
+    doc = parse_document("正文")
+    doc.attributes["title"] = "操作系统复习"
+    doc.attributes["metadata"] = {"tags": ["os", "复习"]}
+
+    opts = ExportOptions(include_title=True, include_metadata=True)
+    result = asyncio.run(HtmlExporter().export(doc, opts))
+    html = result.content.decode("utf-8")
+
+    assert '<h1 class="title">操作系统复习</h1>' in html
+    assert "os, 复习" in html
+
+
+# --------------------------------------------------------------------------- #
+# ExportService
+# --------------------------------------------------------------------------- #
+def _markdown_request(markdown: str, *, format: ExportFormat = ExportFormat.html) -> ExportRequest:
+    return ExportRequest(
+        source=ExportSource(type=ExportSourceType.markdown, markdown=markdown),
+        format=format,
+    )
+
+
+def test_export_markdown_source_completes_and_writes_file() -> None:
+    finished = _create_and_wait(_markdown_request(MD))
+
+    assert finished.status == ExportStatus.completed
+    assert finished.file is not None
+    assert finished.file.mime_type == "text/html"
+    assert finished.file.size > 0
+    assert len(finished.file.sha256) == 64
+
+    path = get_settings().exports_path / f"{finished.job_id}.html"
+    assert path.exists()
+    content = path.read_text(encoding="utf-8")
+    assert "进程调度" in content
+
+
+def test_export_note_source_resolves_title_and_metadata() -> None:
+    from app.services import note_service
+
+    async def _go():
+        note = await note_service.create_note(
+            title="操作系统复习", markdown="# 进程调度\n\n内容。", folder="导出", tags=["os"]
+        )
+        request = ExportRequest(
+            source=ExportSource(type=ExportSourceType.note, note_id=note.note_id),
+            format=ExportFormat.html,
+            options=ExportOptions(include_metadata=True),
+        )
+        job = await export_service.create_export(request)
+        return await export_service.wait_for_export(job.job_id)
+
+    finished = asyncio.run(_go())
+    assert finished.status == ExportStatus.completed
+    assert finished.file is not None
+    assert finished.file.file_name == "操作系统复习.html"
+    content = (get_settings().exports_path / f"{finished.job_id}.html").read_text(encoding="utf-8")
+    assert "操作系统复习" in content
+    assert "进程调度" in content
+
+
+def test_export_pdf_unsupported() -> None:
+    with pytest.raises(ApiError) as exc:
+        asyncio.run(
+            export_service.create_export(_markdown_request("# x", format=ExportFormat.pdf))
+        )
+    assert exc.value.status_code == 400
+    assert exc.value.code == "EXPORT_FORMAT_UNSUPPORTED"
+
+
+def test_export_unknown_note_404() -> None:
+    request = ExportRequest(
+        source=ExportSource(type=ExportSourceType.note, note_id="note_missing"),
+        format=ExportFormat.html,
+    )
+    with pytest.raises(ApiError) as exc:
+        asyncio.run(export_service.create_export(request))
+    assert exc.value.status_code == 404
+    assert exc.value.code == "EXPORT_SOURCE_NOT_FOUND"
+
+
+def test_export_empty_markdown_invalid() -> None:
+    with pytest.raises(ApiError) as exc:
+        asyncio.run(export_service.create_export(_markdown_request("   ")))
+    assert exc.value.status_code == 400
+    assert exc.value.code == "EXPORT_OPTIONS_INVALID"
+
+
+def test_export_cancel_queued_job() -> None:
+    async def _go():
+        job = await export_service.create_export(_markdown_request("# x"))
+        cancelled = export_service.cancel_export(job.job_id)
+        assert cancelled is not None
+        return await export_service.wait_for_export(job.job_id)
+
+    finished = asyncio.run(_go())
+    assert finished.status == ExportStatus.cancelled
+    assert finished.file is None
+
+
+def test_export_file_expired_410() -> None:
+    async def _go():
+        job = await export_service.create_export(_markdown_request("# x"))
+        finished = await export_service.wait_for_export(job.job_id)
+        past = datetime.now(timezone.utc) - timedelta(hours=1)
+        export_service._jobs[job.job_id] = finished.model_copy(
+            update={"file": finished.file.model_copy(update={"expires_at": past})}
+        )
+        return job.job_id
+
+    job_id = asyncio.run(_go())
+    path = get_settings().exports_path / f"{job_id}.html"
+    with pytest.raises(ApiError) as exc:
+        export_service.get_export_file(job_id)
+    assert exc.value.status_code == 410
+    assert exc.value.code == "EXPORT_FILE_EXPIRED"
+    assert not path.exists()  # 过期即清理产物文件
+    assert export_service.get_export(job_id) is None  # 内存记录一并清理
+
+
+def test_export_eviction_deletes_file() -> None:
+    finished = _create_and_wait(_markdown_request("# 淘汰"))
+    victim_path = get_settings().exports_path / f"{finished.job_id}.html"
+    assert victim_path.exists()
+
+    # 塞满 MAX_JOBS 个终态任务，下一次 create 会淘汰最旧的终态（finished 最先插入）
+    for i in range(export_service.MAX_JOBS):
+        export_service._jobs[f"export_fake_{i}"] = ExportJob(
+            job_id=f"export_fake_{i}",
+            status=ExportStatus.completed,
+            format=ExportFormat.html,
+            created_at=datetime.now(timezone.utc),
+        )
+    _create_and_wait(_markdown_request("# 触发淘汰"))
+    assert not victim_path.exists()
+
+
+def test_cleanup_orphan_files() -> None:
+    exports_dir = get_settings().exports_path
+    exports_dir.mkdir(parents=True, exist_ok=True)
+    orphan = exports_dir / "export_orphan.html"
+    orphan.write_text("stale", encoding="utf-8")
+
+    finished = _create_and_wait(_markdown_request("# 保留"))
+    keep_path = exports_dir / f"{finished.job_id}.html"
+    assert keep_path.exists()
+
+    removed = export_service.cleanup_orphan_files()
+    assert removed >= 1
+    assert not orphan.exists()
+    assert keep_path.exists()  # 仍在注册表中的任务文件保留
+
+
+def test_export_cancel_during_running(monkeypatch) -> None:
+    import threading
+    import time
+
+    real_parse = parse_document
+    started = threading.Event()
+
+    def slow_parse(markdown: str):
+        started.set()
+        time.sleep(0.1)
+        return real_parse(markdown)
+
+    monkeypatch.setattr(export_service, "parse_document", slow_parse)
+
+    async def _go():
+        job = await export_service.create_export(_markdown_request("# 运行中取消"))
+        while not started.is_set():
+            await asyncio.sleep(0)
+        export_service.cancel_export(job.job_id)
+        return await export_service.wait_for_export(job.job_id)
+
+    finished = asyncio.run(_go())
+    assert finished.status == ExportStatus.cancelled
+    assert finished.file is None
+    assert not (get_settings().exports_path / f"{finished.job_id}.html").exists()
+
+
+def test_export_list_and_get() -> None:
+    finished = _create_and_wait(_markdown_request("# 列表测试"))
+
+    items, total = export_service.list_exports(limit=50, offset=0)
+    assert total == 1
+    assert items[0].job_id == finished.job_id
+
+    got = export_service.get_export(finished.job_id)
+    assert got is not None and got.status == ExportStatus.completed
+
+    assert export_service.get_export("export_missing") is None
+
+
+# --------------------------------------------------------------------------- #
+# 契约校验
+# --------------------------------------------------------------------------- #
+def test_export_source_requires_matching_field() -> None:
+    with pytest.raises(ValidationError):
+        ExportSource(type=ExportSourceType.note, note_id=None)
+    with pytest.raises(ValidationError):
+        ExportSource(type=ExportSourceType.markdown, markdown=None)
+
+
+# --------------------------------------------------------------------------- #
+# 审阅回归：资源上限
+# --------------------------------------------------------------------------- #
+def test_export_note_source_size_limit(monkeypatch) -> None:
+    # P1：note 源超出 MAX_MARKDOWN_CHARS 应在创建期拒绝，不进入后台渲染
+    from app.services import note_service
+
+    monkeypatch.setattr(export_service, "MAX_MARKDOWN_CHARS", 10)
+
+    async def _go():
+        note = await note_service.create_note(
+            title="超长笔记", markdown="a" * 20, folder="导出", tags=[]
+        )
+        return await export_service.create_export(
+            ExportRequest(
+                source=ExportSource(type=ExportSourceType.note, note_id=note.note_id),
+                format=ExportFormat.html,
+            )
+        )
+
+    with pytest.raises(ApiError) as exc:
+        asyncio.run(_go())
+    assert exc.value.status_code == 400
+    assert exc.value.code == "EXPORT_OPTIONS_INVALID"
+
+
+def test_export_output_too_large(monkeypatch) -> None:
+    # P1：产物超出 MAX_EXPORT_BYTES 应标记 failed 且不落盘
+    monkeypatch.setattr(export_service, "MAX_EXPORT_BYTES", 10)
+
+    finished = _create_and_wait(_markdown_request("# 产物超限"))
+    assert finished.status == ExportStatus.failed
+    assert finished.error_code == "EXPORT_OUTPUT_TOO_LARGE"
+    assert finished.file is None
+    assert not (get_settings().exports_path / f"{finished.job_id}.html").exists()
+
+
+def test_export_limits_concurrent_rendering(monkeypatch) -> None:
+    # P1：并发渲染受 MAX_CONCURRENT_RENDERS 限制，大量任务不会同时占满工作线程
+    import threading
+    import time
+
+    real_render = export_service._render_document
+    active = 0
+    peak = 0
+    lock = threading.Lock()
+
+    def slow_render(document, options):
+        nonlocal active, peak
+        with lock:
+            active += 1
+            peak = max(peak, active)
+        time.sleep(0.05)
+        with lock:
+            active -= 1
+        return real_render(document, options)
+
+    monkeypatch.setattr(export_service, "_render_document", slow_render)
+
+    async def _go():
+        jobs = [
+            await export_service.create_export(_markdown_request(f"# t{i}"))
+            for i in range(6)
+        ]
+        return [await export_service.wait_for_export(j.job_id) for j in jobs]
+
+    finished = asyncio.run(_go())
+    assert all(j.status == ExportStatus.completed for j in finished)
+    assert peak <= export_service.MAX_CONCURRENT_RENDERS

backend/tests/test_plot.py

@@ -0,0 +1,285 @@
+"""Function Plot 的解析与静态 SVG 渲染测试。
+
+覆盖 parser 的白名单表达式（幂/隐式乘法/函数/常量）、拒绝项（属性访问、任意调用等）、
+parse_source 指令与回退，以及 render 的 SVG 输出与 HTML 导出链路集成。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import math
+
+import pytest
+
+from app.contracts import ExportOptions
+from app.export.exporters.html import HtmlExporter
+from app.export.markdown import parse_document
+from app.plot.parser import PlotParseError, evaluate, parse_expression, parse_source
+from app.plot.render import render_svg
+
+
+# --------------------------------------------------------------------------- #
+# 表达式解析
+# --------------------------------------------------------------------------- #
+def test_parse_expression_power_and_implicit_multiplication() -> None:
+    assert evaluate(parse_expression("x^2"), 3) == 9.0
+    assert evaluate(parse_expression("2^3"), 0) == 8.0
+    assert evaluate(parse_expression("2x+1"), 3) == 7.0
+    assert evaluate(parse_expression("2(x+1)"), 3) == 8.0
+    assert evaluate(parse_expression("(x+1)(x-1)"), 3) == 8.0
+
+
+def test_parse_expression_functions_and_constants() -> None:
+    assert evaluate(parse_expression("sin(0)"), 0) == 0.0
+    assert math.isclose(evaluate(parse_expression("sin(pi/2)"), 0), 1.0)
+    assert math.isclose(evaluate(parse_expression("ln(e)"), 0), 1.0)
+    assert evaluate(parse_expression("abs(-3)"), 0) == 3.0
+
+
+def test_parse_expression_rejects_unsafe() -> None:
+    unsafe = [
+        "os.system('x')",
+        "__import__('os')",
+        "foo(x)",
+        "eval('x')",
+        "x[0]",
+        "x.attr",
+        "lambda: 1",
+    ]
+    for expr in unsafe:
+        with pytest.raises(PlotParseError) as exc:
+            parse_expression(expr)
+        assert exc.value.diagnostic.code == "FUNCTION_PLOT_EXPRESSION_UNSAFE", expr
+
+
+def test_parse_expression_syntax_error() -> None:
+    with pytest.raises(PlotParseError) as exc:
+        parse_expression("x +")
+    assert exc.value.diagnostic.code == "FUNCTION_PLOT_PARSE_FAILED"
+
+
+# --------------------------------------------------------------------------- #
+# fenced 源码解析
+# --------------------------------------------------------------------------- #
+def test_parse_source_directives() -> None:
+    result = parse_source("domain: 0, 10\nrange: -1, 1\nxlabel: x\ngrid: false\ny = x^2")
+    assert result.plot is not None
+    assert result.plot.domain == (0.0, 10.0)
+    assert result.plot.range == (-1.0, 1.0)
+    assert result.plot.axes.xlabel == "x"
+    assert result.plot.axes.grid is False
+    assert len(result.plot.expressions) == 1
+    assert result.plot.expressions[0].expression == "x^2"
+
+
+def test_parse_source_bare_and_multi_expression() -> None:
+    result = parse_source("x^2\nsin(x)")
+    assert result.plot is not None
+    assert [e.expression for e in result.plot.expressions] == ["x^2", "sin(x)"]
+
+
+def test_parse_source_unknown_directive_warns() -> None:
+    result = parse_source("foo: bar\ny = x")
+    assert result.plot is not None  # 未知指令仅 warning，不阻断
+    assert any(d.severity == "warning" for d in result.diagnostics)
+
+
+def test_parse_source_error_returns_no_plot() -> None:
+    result = parse_source("y = os.system('x')")
+    assert result.plot is None
+    assert any(d.severity == "error" for d in result.diagnostics)
+
+
+# --------------------------------------------------------------------------- #
+# SVG 渲染
+# --------------------------------------------------------------------------- #
+def test_render_svg_contains_polyline_and_axes() -> None:
+    plot = parse_source("y = x^2").plot
+    rendered = render_svg(plot)
+    svg = rendered.content
+    assert "<svg" in svg
+    assert "<polyline" in svg
+    assert "<line" in svg  # 坐标轴/网格
+    assert "<script" not in svg
+    assert rendered.width == 640
+    assert rendered.height == 480
+
+
+def test_render_svg_multiple_functions() -> None:
+    plot = parse_source("y = x^2\ny = sin(x)").plot
+    rendered = render_svg(plot)
+    assert rendered.content.count("<polyline") >= 2
+
+
+def test_render_svg_labels() -> None:
+    plot = parse_source("xlabel: 时间\nylabel: 数值\ny = x").plot
+    rendered = render_svg(plot)
+    assert "时间" in rendered.content
+    assert "数值" in rendered.content
+
+
+# --------------------------------------------------------------------------- #
+# HTML 导出链路集成
+# --------------------------------------------------------------------------- #
+def test_html_exporter_embeds_function_plot_svg() -> None:
+    md = "```function-plot\ny = x^2\n```"
+    result = asyncio.run(HtmlExporter().export(parse_document(md), ExportOptions()))
+    html = result.content.decode("utf-8")
+    assert '<figure class="function-plot">' in html
+    assert "<svg" in html
+    assert "<polyline" in html
+
+
+def test_html_exporter_function_plot_fallback_on_error() -> None:
+    md = "```function-plot\ny = os.system('x')\n```"
+    result = asyncio.run(HtmlExporter().export(parse_document(md), ExportOptions()))
+    html = result.content.decode("utf-8")
+    assert '<pre class="function-plot">' in html
+    assert "<svg" not in html
+    assert any("函数图像" in w for w in result.warnings)
+
+
+# --------------------------------------------------------------------------- #
+# 审阅回归：浮点刻度 / 求值异常 / 无效范围
+# --------------------------------------------------------------------------- #
+def test_render_svg_huge_domain_ticks_bounded() -> None:
+    # P1：巨大 domain 下步长受浮点精度限制无法推进，刻度应有限而非死循环
+    plot = parse_source("domain: 10000000000000000, 10000000000000002\nrange: -1, 1\ny = 0").plot
+    rendered = render_svg(plot)
+    assert "<svg" in rendered.content
+
+
+def test_parse_expression_rejects_wrong_arg_count() -> None:
+    # P2：sin() / sin(1, 2) 应在解析期拒绝，而非求值期 TypeError
+    with pytest.raises(PlotParseError):
+        parse_expression("sin()")
+    with pytest.raises(PlotParseError):
+        parse_expression("sin(1, 2)")
+
+
+def test_render_svg_nonreal_samples_are_break_points() -> None:
+    # P2：x^0.5 在负数域产生复数，应作为断点处理，正半轴仍可绘制
+    plot = parse_source("domain: -4, 4\ny = x^0.5").plot
+    rendered = render_svg(plot)
+    assert "<polyline" in rendered.content
+
+
+def test_render_svg_invalid_range_falls_back() -> None:
+    # P2：退化 range（1, 1）应丢弃并自动采样，而非 ZeroDivisionError
+    plot = parse_source("range: 1, 1\ny = x").plot
+    rendered = render_svg(plot)
+    assert "<polyline" in rendered.content
+    assert any("range" in w for w in rendered.warnings)
+
+
+def test_render_svg_nonfinite_range_falls_back() -> None:
+    # P2：非有限 range 端点应丢弃并自动采样
+    plot = parse_source("range: nan, 1\ny = x").plot
+    rendered = render_svg(plot)
+    assert "<polyline" in rendered.content
+
+
+def test_html_exporter_function_plot_render_error_falls_back(monkeypatch) -> None:
+    # P2：渲染异常不阻断整篇导出，回退占位并记 warning
+    import app.export.exporters.html as html_mod
+
+    def boom(plot):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(html_mod, "render_svg", boom)
+    md = "```function-plot\ny = x\n```"
+    result = asyncio.run(HtmlExporter().export(parse_document(md), ExportOptions()))
+    html = result.content.decode("utf-8")
+    assert '<pre class="function-plot">' in html
+    assert any("渲染失败" in w for w in result.warnings)
+
+
+# --------------------------------------------------------------------------- #
+# 审阅回归：复杂表达式 / 极端数值范围
+# --------------------------------------------------------------------------- #
+def test_parse_expression_rejects_excessive_depth() -> None:
+    # P2：超长加法链的 AST 深度超限，应拒绝为 PlotParseError 而非触发 RecursionError
+    expr = "+".join(["1"] * 300)
+    with pytest.raises(PlotParseError) as exc:
+        parse_expression(expr)
+    assert exc.value.diagnostic.code == "FUNCTION_PLOT_EXPRESSION_UNSAFE"
+
+
+def test_parse_expression_rejects_excessive_nodes() -> None:
+    # P2：浅层但节点超限的表达式（满二叉树）应被节点数上限拦截
+    def balanced(depth: int) -> str:
+        if depth == 0:
+            return "x"
+        return f"({balanced(depth - 1)}+{balanced(depth - 1)})"
+
+    expr = balanced(10)  # ~2047 个节点，深度仅 ~10
+    with pytest.raises(PlotParseError) as exc:
+        parse_expression(expr)
+    assert exc.value.diagnostic.code == "FUNCTION_PLOT_EXPRESSION_UNSAFE"
+
+
+def test_html_exporter_function_plot_deep_expression_falls_back() -> None:
+    # P2：复杂表达式解析失败应回退占位，不阻断整篇导出
+    expr = "+".join(["1"] * 300)
+    md = f"```function-plot\ny = {expr}\n```"
+    result = asyncio.run(HtmlExporter().export(parse_document(md), ExportOptions()))
+    html = result.content.decode("utf-8")
+    assert '<pre class="function-plot">' in html
+    assert "<svg" not in html
+    assert any("函数图像" in w for w in result.warnings)
+
+
+def test_render_svg_extreme_domain_no_nan() -> None:
+    # P2：有限但跨度溢出的 domain 应回退安全范围，SVG 不得含 nan/inf
+    plot = parse_source("domain: -1e308, 1e308\nrange: -1, 1\ny = 0").plot
+    rendered = render_svg(plot)
+    assert "<svg" in rendered.content
+    assert "nan" not in rendered.content
+    assert "inf" not in rendered.content
+    assert any("domain" in w for w in rendered.warnings)
+
+
+def test_render_svg_extreme_range_no_nan() -> None:
+    # P2：有限但跨度溢出的 range 应回退自动范围，SVG 不得含 nan/inf
+    plot = parse_source("domain: -1, 1\nrange: -1e308, 1e308\ny = x").plot
+    rendered = render_svg(plot)
+    assert "<svg" in rendered.content
+    assert "nan" not in rendered.content
+    assert "inf" not in rendered.content
+    assert any("range" in w for w in rendered.warnings)
+
+
+# --------------------------------------------------------------------------- #
+# 审阅回归：函数/图像数量上限
+# --------------------------------------------------------------------------- #
+def test_parse_source_rejects_too_many_expressions() -> None:
+    # P1：单块表达式数量超限应整块回退，避免海量采样求值
+    source = "\n".join(f"y = x + {i}" for i in range(50))
+    result = parse_source(source)
+    assert result.plot is None
+    assert any(d.code == "FUNCTION_PLOT_TOO_MANY_EXPRESSIONS" for d in result.diagnostics)
+
+
+def test_html_exporter_limits_function_plot_count() -> None:
+    # P1：文档内函数图像数量超限，超出部分回退占位，不耗尽资源
+    blocks = "\n\n".join("```function-plot\ny = x\n```" for _ in range(20))
+    result = asyncio.run(HtmlExporter().export(parse_document(blocks), ExportOptions()))
+    html = result.content.decode("utf-8")
+    # 上限 16：前 16 个渲染为 SVG，其余 4 个回退占位
+    assert html.count('<figure class="function-plot">') == 16
+    assert html.count('<pre class="function-plot">') == 4
+    assert any("函数图像数量超过上限" in w for w in result.warnings)
+
+
+def test_html_exporter_limits_total_plot_nodes(monkeypatch) -> None:
+    # P1：文档级累计 AST 节点预算超限后，后续图像回退占位，防止组合复杂度耗尽 CPU
+    import app.export.exporters.html as html_mod
+
+    monkeypatch.setattr(html_mod, "_MAX_TOTAL_PLOT_NODES", 5)
+    # 第一个图块 y=x（1 节点）在预算内；第二个图块 y=x+x+x+x（7 节点）累计超限
+    md = "```function-plot\ny = x\n```\n\n```function-plot\ny = x + x + x + x\n```"
+    result = asyncio.run(HtmlExporter().export(parse_document(md), ExportOptions()))
+    html = result.content.decode("utf-8")
+    assert html.count('<figure class="function-plot">') == 1
+    assert html.count('<pre class="function-plot">') == 1
+    assert any("累计复杂度" in w for w in result.warnings)

backend/uv.lock

@@ -364,6 +364,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/41/45/1a4ed80516f02155c51f51e8cedb3c1902296743db0bbc66608a0db2814f/jsonschema_specifications-2025.9.1-py3-none-any.whl", hash = "sha256:98802fee3a11ee76ecaca44429fda8a41bff98b00a0f2838151b113f210cc6fe", size = 18437, upload-time = "2025-09-08T01:34:57.871Z" },
 ]
 
+[[package]]
+name = "mistune"
+version = "3.3.4"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/7b/92/328a294a6de83bacb95bed01f04e0eaff4e3616ee359fc821a5dfc539b02/mistune-3.3.4.tar.gz", hash = "sha256:58b5c96d6fcb61190dfe5fae498d2b2065f99cf61e9649418fd54cf1ada86dfe", size = 121426, upload-time = "2026-07-22T05:22:30.89Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/77/e4/288365afae98953bc01de09f686f40d8ee84578135aa7767d5d4e60b5278/mistune-3.3.4-py3-none-any.whl", hash = "sha256:ee015381e955e370962968befe1d729ab60fafb6a715ac6751763fbce38c8d4a", size = 66862, upload-time = "2026-07-22T05:22:29.419Z" },
+]
+
 [[package]]
 name = "notes-agent-backend"
 version = "0.1.0"
@@ -373,6 +382,7 @@ dependencies = [
     { name = "fastapi" },
     { name = "httpx" },
     { name = "jsonschema" },
+    { name = "mistune" },
     { name = "pyyaml" },
     { name = "referencing" },
     { name = "sqlite-vec" },
@@ -390,6 +400,7 @@ requires-dist = [
     { name = "fastapi", specifier = ">=0.116,<1.0" },
     { name = "httpx", specifier = ">=0.28,<1.0" },
     { name = "jsonschema", specifier = ">=4.25,<5.0" },
+    { name = "mistune", specifier = ">=3.0,<4.0" },
     { name = "pyyaml", specifier = ">=6.0,<7.0" },
     { name = "referencing", specifier = ">=0.36,<1.0" },
     { name = "sqlite-vec", specifier = ">=0.1.9" },

docs/README.md

@@ -38,6 +38,7 @@
 - [AI Core 与 Agent Core 开发说明](development/AI-Core与Agent-Core开发说明.md)
 - [Knowledge 与 Retrieval Core 开发说明](development/Knowledge与Retrieval-Core开发说明.md)
 - [Benchmark 开发说明](development/Benchmark开发说明.md)
+- [Export 开发说明](development/Export开发说明.md)
 - [模型提供商与模型发现开发说明](development/模型提供商与模型发现开发说明.md)
 - [MCP Bridge 与 Plugin Host 开发说明](development/MCP-Bridge与Plugin-Host开发说明.md)
 - [独立 MCP Server 配置中心开发说明](development/独立MCP-Server配置中心开发说明.md)

docs/architecture/AI笔记软件技术栈说明-团队版-v2.3.md

@@ -2365,7 +2365,7 @@ Quality
 └── Retrieval 参数调优
 
 Content Output
-├── Markdown → HTML / PDF / DOCX
+├── Markdown → HTML（已实现）/ PDF / DOCX（暂缓）
 ├── Mermaid 编辑、预览与静态导出
 └── Function Plot 解析、预览与静态导出
 
@@ -2376,7 +2376,7 @@ Frontend Extension
 └── Plugin Settings UI
 ```
 
-上述列表描述第二阶段技术范围。多模态、MCP Bridge、Plugin Command/Settings、RAG Benchmark 和 Provider 增强已经实现；Agent Benchmark、内容导出、Mermaid/函数图像完整编辑导出及社区主题包仍以各自开发说明的状态为准。每项功能必须继续经过现有 Service、Contract、Permission 和 Adapter 边界，不因 Demo 需要在 Vue 组件、Router 或 Agent Runtime 中直接绑定第三方协议。
+上述列表描述第二阶段技术范围。多模态、MCP Bridge、Plugin Command/Settings、RAG Benchmark、Provider 增强与 Markdown → HTML 导出已经实现；Agent Benchmark、PDF/DOCX 导出、Mermaid/函数图像完整编辑导出及社区主题包仍以各自开发说明的状态为准。每项功能必须继续经过现有 Service、Contract、Permission 和 Adapter 边界，不因 Demo 需要在 Vue 组件、Router 或 Agent Runtime 中直接绑定第三方协议。
 
 第三阶段处理：
 

docs/contracts/第二阶段接口契约-开发版.md

@@ -68,11 +68,11 @@
 | Benchmark | POST | `/api/benchmarks/agent/runs` | 暂缓 | 创建 Agent Benchmark（依赖 Agent Runtime 完成后交付） |
 | Benchmark | GET | `/api/benchmarks/runs` | 已实现 | 分页获取 Benchmark Run |
 | Benchmark | GET/POST | `/api/benchmarks/runs/{run_id}/*` | 计划新增 | 查询、订阅、取消和读取报告 |
-| Export | POST | `/api/exports` | 计划新增 | 创建 HTML/PDF/DOCX 导出任务 |
+| Export | POST | `/api/exports` | 已实现（HTML） | 创建导出任务；`pdf`/`docx` 暂缓，返回 `EXPORT_FORMAT_UNSUPPORTED` |
-| Export | GET | `/api/exports` | 计划新增 | 分页获取导出任务 |
+| Export | GET | `/api/exports` | 已实现（HTML） | 分页获取导出任务 |
-| Export | GET | `/api/exports/{job_id}` | 计划新增 | 查询导出任务 |
+| Export | GET | `/api/exports/{job_id}` | 已实现（HTML） | 查询导出任务 |
-| Export | GET | `/api/exports/{job_id}/file` | 计划新增 | 下载已完成产物 |
+| Export | GET | `/api/exports/{job_id}/file` | 已实现（HTML） | 下载已完成产物 |
-| Export | POST | `/api/exports/{job_id}/cancel` | 计划新增 | 取消导出任务 |
+| Export | POST | `/api/exports/{job_id}/cancel` | 已实现（HTML） | 取消导出任务 |
 | Theme | Host Contract | `ThemePackageService` | 计划新增 | 导入、预览、启停和卸载主题包 |
 | Renderer | 内部 Contract | `StaticRenderer` | 计划新增 | Mermaid/Function Plot 预览和导出复用 |
 
@@ -1080,6 +1080,8 @@ VECTOR_INDEX_REBUILD_REQUIRED
 
 ## 10. Export Service
 
+> 实现状态：HTML 导出已实现（`backend/app/export/`），`pdf`/`docx` 暂缓——请求这两个格式返回 `EXPORT_FORMAT_UNSUPPORTED`。`function-plot` 已支持静态 SVG 内嵌（`backend/app/plot/`），解析或渲染失败时回退为源码占位并记录 warning；Mermaid 目前仍以占位代码块保留并记 warning。
+
 ### 10.1 创建导出任务
 
 `POST /api/exports`，返回 `202 ExportJob`。
@@ -1090,7 +1092,7 @@ VECTOR_INDEX_REBUILD_REQUIRED
     "type": "note",
     "note_id": "note_123"
   },
-  "format": "pdf",
+  "format": "html",
   "options": {
     "theme_id": "light",
     "include_title": true,
@@ -1101,7 +1103,7 @@ VECTOR_INDEX_REBUILD_REQUIRED
 }
 ```
 
-`source.type` 首批支持 `note` 和 `markdown`。`markdown` 来源用于尚未保存的预览，字段大小受限且不持久化到 Trace。`format` 固定为 `html`、`pdf`、`docx`。
+`source.type` 首批支持 `note` 和 `markdown`。`note` 来源通过 `source.note_id` 引用已建索引笔记；`markdown` 来源用于尚未保存的预览，内容放在 `source.markdown` 字段，大小限制为 200 000 字符、不持久化到 Trace。`format` 可取 `html`、`pdf`、`docx`，但当前仅 `html` 已实现，`pdf`/`docx` 返回 `EXPORT_FORMAT_UNSUPPORTED`。
 
 响应：
 
@@ -1109,11 +1111,12 @@ VECTOR_INDEX_REBUILD_REQUIRED
 {
   "job_id": "export_123",
   "status": "queued",
-  "format": "pdf",
+  "format": "html",
   "progress": null,
   "file": null,
   "warnings": [],
   "error": null,
+  "error_code": null,
   "created_at": "2026-08-31T10:30:00Z",
   "started_at": null,
   "completed_at": null
@@ -1129,14 +1132,14 @@ VECTOR_INDEX_REBUILD_REQUIRED
 | POST | `/api/exports/{job_id}/cancel` | `OperationResponse` |
 | GET | `/api/exports/{job_id}/file` | 文件流 |
 
-下载响应设置正确 `Content-Type`、经过清理的 `Content-Disposition` 文件名和 `Content-Length`。未完成、失败或过期 Job 不返回空文件。
+下载响应设置正确 `Content-Type`、经过清理的 `Content-Disposition` 文件名和 `Content-Length`。未完成、失败或过期的 Job 不返回空文件：未完成/失败返回 `EXPORT_JOB_NOT_FOUND`（404），产物过期（超过 `expires_at`）返回 `EXPORT_FILE_EXPIRED`（410）。
 
 完成 Job 的 file：
 
 ```json
 {
-  "file_name": "操作系统复习.pdf",
+  "file_name": "操作系统复习.html",
-  "mime_type": "application/pdf",
+  "mime_type": "text/html",
   "size": 1048576,
   "sha256": "...",
   "expires_at": "2026-09-01T10:30:00Z"
@@ -1225,6 +1228,7 @@ EXPORT_RENDER_FAILED
 EXPORT_UNSUPPORTED_CONTENT
 EXPORT_JOB_NOT_FOUND
 EXPORT_FILE_EXPIRED
+EXPORT_OUTPUT_TOO_LARGE
 ```
 
 ---

docs/development/Export开发说明.md

@@ -0,0 +1,106 @@
+# Export 开发说明
+
+> 所属模块：Export Service（后端，负责人 yxx）。交付「多格式文档导出」：Markdown → HTML 的完整生命周期与 function-plot 静态 SVG 渲染；PDF/DOCX 在后续 PR 补齐。契约对应 [第二阶段接口契约 §10](../contracts/第二阶段接口契约-开发版.md)。
+
+## 定位
+
+Export Service 把笔记或未保存的 Markdown 文本渲染为可下载的 HTML 文件。采用与 Benchmark 一致的「创建即返回 queued、后台 asyncio.Task 执行」的内存模型，产物带 24h 过期时间，过期后不可下载。导出是轮询式（无 SSE 事件流），客户端通过 `GET /api/exports/{job_id}` 轮询状态，完成后走 `GET /api/exports/{job_id}/file` 下载。
+
+## 模块布局
+
+```text
+backend/app/export/
+├── __init__.py        包说明
+├── document.py        Document AST 内部协议 + DocumentExporter Protocol + ExportResult
+├── markdown.py        mistune 'ast' renderer → Document AST
+├── exporters/
+│   ├── __init__.py
+│   └── html.py        HtmlExporter（Document AST → 完整 HTML5）
+└── service.py         ExportService（注册表 + 后台渲染 + 取消 + 产物生命周期）
+```
+
+HTTP DTO（`ExportStatus` / `ExportFormat` / `ExportSource` / `ExportOptions` / `ExportJob` 等）放在 [app/contracts.py](../../backend/app/contracts.py)，与 Benchmark DTO 同层；`DocumentNode` / `ExportResult` 属导出器内部协议，放在 `export/document.py`，不进入 HTTP 契约。
+
+## 接口
+
+| 方法 | 路径 | 用途 |
+| --- | --- | --- |
+| POST | `/api/exports` | 创建导出任务（202） |
+| GET | `/api/exports?status=&format=&limit=&offset=` | 分页获取任务 |
+| GET | `/api/exports/{job_id}` | 查询任务状态 |
+| GET | `/api/exports/{job_id}/file` | 下载已完成产物 |
+| POST | `/api/exports/{job_id}/cancel` | 取消任务 |
+
+`source.type` 支持 `note`（引用已建索引笔记）与 `markdown`（未保存预览，字段为 `source.markdown`，上限 200 000 字符）。当前仅 `format=html` 实现，`pdf`/`docx` 返回 `EXPORT_FORMAT_UNSUPPORTED`。
+
+## Markdown → Document AST
+
+解析用 [mistune](https://github.com/lepture/mistune) 的内置 `renderer="ast"`（非自写 `BaseRenderer`），因为 mistune 的行内渲染按字符串拼接、无法承载结构化子节点；ast renderer 直接给出带 `children`/`attrs`/`raw` 的 token 树，`_AstMapper` 只做 token → `DocumentNode` 的搬运，不掺入任何 HTML。插件启用 `table`、`math`、`url`、`task_lists`。
+
+fenced code 按语言分流：`mermaid` → `mermaid` 节点、`function_plot`/`functionplot` → `function_plot` 节点，其余 → `code_block`（`attributes.language`）。`node_id` 按遍历顺序 `node_{seq:03d}` 生成，仅渲染内部使用，无需跨请求稳定。
+
+## HtmlExporter
+
+递归渲染 Document AST 为完整 HTML5 文档（`<!doctype html>` + `<head>` 内嵌基础 CSS + `<body>`），标题/正文/元信息文本一律 `html.escape`。`function_plot` 解析为静态 SVG 内嵌（解析/渲染失败或超限时回退 `<pre class="function-plot">` 占位并记 warning），`mermaid` 无法静态表达，渲染为占位 `<pre class="mermaid">` 并记 warning，均不静默丢失；`code_theme` 仅作为代码容器 class，不引入 JS 高亮库。无法表示的节点统一 `warnings.append(...)` 跳过。
+
+## 运行生命周期
+
+`queued → running → completed | failed | cancelled`。
+
+- 创建时校验：`format` 非 html → `EXPORT_FORMAT_UNSUPPORTED`；`note` 源不存在 → `EXPORT_SOURCE_NOT_FOUND`（404）；`markdown` 源为空或超上限 → `EXPORT_OPTIONS_INVALID`。
+- 内存注册表上限 `MAX_JOBS=100`，超限只淘汰终态任务；满容量且全为活动任务时返回 `EXPORT_CAPACITY_EXCEEDED`（429）。
+- 后台渲染在解析前后各让出一次执行权，使「创建后立即取消」的 queued 任务能及时进入 cancelled。
+- 失败只向公开响应暴露项目错误码与安全消息，详细异常进入日志。
+
+## 产物生命周期
+
+产物写入 `settings.exports_path`（默认 `backend/data/exports/`，可通过 `APP_EXPORTS_PATH` 覆盖，已加入 `.gitignore`），文件名为 `{job_id}.html`，下载 `Content-Disposition` 用 `_safe_download_name` 清洗标题得到。`ExportFile` 记录 `sha256`、`size` 与 `expires_at`（`completed_at + 24h`），过期返回 `EXPORT_FILE_EXPIRED`（410）。
+
+## 资源上限
+
+为防止超大输入或海量函数图像耗尽内存/线程，导出链路内置以下上限：
+
+- 输入源（`note` 与 `markdown`）统一限制 `MAX_MARKDOWN_CHARS = 200_000` 字符，超限返回 `EXPORT_OPTIONS_INVALID`。
+- 单个 `function-plot` 图块最多 16 条表达式，超限整块回退占位并记结构化诊断 `FUNCTION_PLOT_TOO_MANY_EXPRESSIONS`。
+- 单篇文档最多 16 个函数图像，超出部分回退占位并记 warning。
+- 单篇文档累计函数图像 AST 节点预算 `_MAX_TOTAL_PLOT_NODES = 8000`，超出部分回退占位并记 warning，防止多图块 × 多表达式 × 深表达式组合在采样求值时长时间占满 CPU。
+- 并发渲染上限 `MAX_CONCURRENT_RENDERS = 2`，解析/渲染是 CPU 密集工作，超出限额的任务在内存中排队等待渲染槽位，避免大量任务同时占满工作线程与内存。
+- 最终产物大小上限 `MAX_EXPORT_BYTES = 20 MB`，超限任务标记 failed 并返回 `EXPORT_OUTPUT_TOO_LARGE`。
+
+## 错误码
+
+错误分两类：**同步错误**在创建/查询请求的 HTTP 响应里直接返回对应状态码；**异步任务错误**在创建时已返回 `202`，后续轮询 `GET /api/exports/{job_id}` 仍返回 `200`，错误通过任务状态与 `error_code` 字段暴露，**不映射 HTTP 状态码**。
+
+同步错误：
+
+```text
+EXPORT_SOURCE_NOT_FOUND      404
+EXPORT_FORMAT_UNSUPPORTED    400
+EXPORT_OPTIONS_INVALID       400
+EXPORT_UNSUPPORTED_CONTENT   422（预留）
+EXPORT_JOB_NOT_FOUND         404
+EXPORT_FILE_EXPIRED          410
+EXPORT_CAPACITY_EXCEEDED     429
+```
+
+异步任务错误（轮询返回 `200`，字段形如 `{"status": "failed", "error_code": "..."}`）：
+
+```text
+EXPORT_RENDER_FAILED
+EXPORT_OUTPUT_TOO_LARGE
+```
+
+## 测试
+
+```powershell
+cd backend
+uv run pytest -q
+```
+
+`tests/test_export.py` 覆盖 Markdown 解析（标题/行内/列表/代码分流/表格/数学）、HTML 渲染（标签 + 转义 + warning）、Service 端到端（note 源与 markdown 源、pdf 拒绝、未知 note、取消、list/get、过期 410）与 `ExportSource` 契约校验。
+
+## 范围外（后续 PR）
+
+- PDF / DOCX 导出（`python-docx` 等底层库在 PoC 后冻结，封装在 Exporter Adapter 内）。
+- 函数图像交互预览与缩放（前端 JS Renderer 负责，后端仅提供静态 SVG）。
+- 代码语法高亮（当前仅 CSS class 占位）。