运行管理与来源导入

04 / 05 · DOCUMENTATION

让重复提交、断流、审批、取消与文献更新都有可追踪的状态和明确的恢复条件。

同一个请求只执行一次

客户端超时并不意味着服务端没有接到问题。用户重试时，如果每次都创建全新运行，可能重复调用模型、重复消耗预算，并在会话中留下两份相同回答。因此请求身份和运行身份必须由服务端持久记录，而不是只靠前端发送按钮防连点。

客户端生成请求标识，服务端同时计算包含工作区、问题、会话、提供商与检索过滤条件的指纹。相同标识与相同内容重试时复用运行；相同标识对应不同内容则返回冲突。

citedrag/backend/src/services/runs.pyL32-50 @f821925

def fingerprint(
    workspace_id: str,
    question: str,
    session_id: str | None,
    llm_config: dict[str, Any] | None,
    provider_profile_id: str | None,
    retrieval_filters: dict[str, Any] | None = None,
) -> str:
    provider = {k: v for k, v in (llm_config or {}).items() if k != "api_key"}
    payload = {
        "workspace_id": workspace_id,
        "question": question.strip(),
        "session_id": session_id,
        "provider_profile_id": provider_profile_id,
        "provider": provider,
        "retrieval_filters": retrieval_filters,
    }
    value = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(value.encode("utf-8")).hexdigest()

指纹排除 api_key；身份来自工作区与运行记录，不能仅依赖一个前端按钮是否禁用。

运行持久化保存状态，工作区 generation 用于隔离已失效操作。请求日志与响应通过 request_id 对应，便于将一次前端失败关联到运行、检索与任务记录。

标识	表示什么	不应混为
client_request_id	一次业务提交及其重试	每次 HTTP 连接的身份
run_id	服务端的一次持久运行	整个多轮会话
session_id	对话历史容器	一次模型执行
request_id	日志与响应关联	防止重复提交的幂等键
workspace generation	运行所属的有效工作区世代	用户身份或访问权限

同一幂等键只有指纹一致才能复用；改了问题、会话、提供商参数或过滤范围，却沿用旧键，应得到冲突，避免服务端悄悄返回另一份请求的结果。凭据不进入指纹与原始运行持久化，也意味着恢复时不能无条件重用请求级密钥。

状态推进与提交

运行先被认领，再进入 running；研究计划可以暂停为 awaiting_approval，中断可成为 interrupted，终止则区分 completed、cancelled 和 failed。恢复通过条件更新重新获得执行权，不能让两个恢复请求同时启动同一轮。

提交答案时再次检查运行状态和工作区 generation。即使后台计算已经得出答案，若用户先取消或工作区已失效，也不能继续写入会话。停止计算与阻止过期结果提交是两层保护，后者尤其重要，因为外部调用不一定即时停止。

SSE 与最终状态

前端严格检查事件顺序与字段：先得到 run_started，再接收带序号的答案增量、证据和轨迹，最后收到元数据和完成事件；等待审批是另一种终止状态。

流结束不能直接视为成功。网络断开时，前端根据运行标识查询服务端状态，区分仍在运行、已完成、已取消和可恢复中断。完成消息只提交一次，避免重试留下两份回答。

客户端校验哪些协议条件

前端消费器检查单次开始、运行 ID 一致、答案增量序号连续、metadata 不重复，并要求正常完成前已有 metadata。遇到断流、错误 JSON、跨运行事件或终态后的额外事件，会明确进入异常路径，不把已有半段文本当成完整结果。

等待审批是合法暂停终点，不要求伪造一份已完成答案。SSE 传输的一块数据也不一定是一条事件，因此消费器先缓冲并处理分隔边界，再解析字段。

观察到的现象	应进一步确认
HTTP 连接结束但没有完成事件	根据 run_id 查询持久状态
已收到部分文本后断网	不重复提交，确认原运行是否仍在执行
出现计划审批	当前运行是否 awaiting_approval，决定是否批准
点击停止后仍有迟到事件	检查取消状态，拒绝旧结果提交
界面恢复显示完成结果	确认 committed 与会话结果，避免重复追加

取消与恢复

真实运行使用 LangGraph checkpoint 保存图状态。恢复只接受待审批或中断运行，并先条件更新到 running；取消后到达的旧执行结果不能重新提交答案。

citedrag/backend/src/services/runs.pyL259-304 @f821925

def resume(
    run_id: str,
    workspace_id: str,
    action: str,
    tasks: list[dict[str, Any]] | None,
    cancellation_requested: Callable[[str], bool],
    store_run: Callable[[str, dict[str, Any]], None],
    run_lock: threading.Lock,
    cancellation_events: dict[str, threading.Event],
    on_token: Any | None = None,
    on_evidence: Any | None = None,
    on_trace: Any | None = None,
) -> dict[str, Any]:
    """Resume an approved or interrupted run from its LangGraph checkpoint.

    The provider is re-resolved from the current default profile/environment
    because the original request-scoped credentials are intentionally not
    persisted.
    """
    run = runs.get_run_scoped(run_id, workspace_id)
    if run is None:
        raise KeyError("run_not_found")
    status = str(run.get("status", ""))
    if status not in {"awaiting_approval", "interrupted"}:
        raise ValueError("run_not_resumable")
    if not runs.mark_run_running(run_id, workspace_id):
        raise RunNotActiveError("fenced")
    llm_config = get_llm_config(load_settings())
    resume_value: dict[str, Any] | None = None
    if status == "awaiting_approval":
        resume_value = {"action": action, "tasks": tasks or []}
    return answer_and_persist(
        str(run["question"]),
        run.get("session_id"),
        workspace_id,
        llm_config,
        run_id,
        cancellation_requested,
        store_run,
        run_lock,
        cancellation_events,
        on_token=on_token,
        on_evidence=on_evidence,
        on_trace=on_trace,
        resume=resume_value,
    )

恢复时重新解析当前模型配置；请求级凭据不会作为原始运行内容持久化。

浏览器演示用本地记录模拟这些状态：刷新把进行中的流标为中断，在该会话的演示顶部点击“继续中断运行”可续跑；取消后的运行不能通过旧审批恢复。它没有服务端检查点，也不会在页面关闭后继续计算。

真实服务还有进度感知 watchdog，同时关注运行总时长和长时间无进展的情况，客户端断开后仍能约束后台运行。超时触发取消信号，并在宽限后条件标记失败；这不保证第三方同步调用立刻终止，但可防止迟到结果被当成正常答案提交。

恢复会重新解析当前默认模型配置，而不是保存并重放原请求凭据。如果默认模型已经变化，续跑的环境也可能不同，应结合当前配置解释结果；“可以续跑”不等于在任意配置变化后都能逐 token 重现原输出。

可选的文献来源

Zotero 是可选来源，不是建立普通文档工作区的前置条件。项目支持内置进程内连接与外部 MCP 服务，配置可热生效；先浏览条目，再按需导入附件，避免无差别索引整个文献库。

citedrag/backend/src/services/imports.pyL148-214 @f821925

def import_one(workspace_id: str, source: SourceSettings, impl: Any, item_key: str,
               *, item_types: tuple[str, ...], kinds: tuple[str, ...]) -> dict[str, Any]:
    item = impl.get_item(item_key)
    if item.get("deleted"):
        return {"item_key": item_key, "paper_id": None, "status": "skipped", "documents": [],
                "errors": [{"code": "item_deleted", "message": "来源侧已删除"}]}
    if item_types and item.get("item_type") not in item_types:
        return {"item_key": item_key, "paper_id": None, "status": "failed", "documents": [],
                "errors": [{"code": "item_type_excluded",
                            "message": f"条目类型不在导入范围: {item.get('item_type')}"}]}
    existing = papers.get_paper_by_item_key(workspace_id, source.id, item_key)
    paper_existed = existing is not None
    had_links = papers.list_paper_document_ids(existing["id"]) if existing else []
    paper_id = papers.upsert_paper({
        "workspace_id": workspace_id, "source_id": source.id, "source_item_key": item_key,
        "source_version": item.get("version"), "source_meta": None,
        "item_type": item.get("item_type"), "title": item.get("title") or "",
        "abstract": item.get("abstract"), "year": item.get("year"), "venue": item.get("venue"),
        "doi": item.get("doi"), "url": item.get("url"), "creators": item.get("creators") or [],
        "tags": item.get("tags") or [], "collections": item.get("collections") or [],
    })
    documents: list[str] = []
    errors: list[dict[str, str]] = []
    for att in impl.list_attachments(item_key):
        if att.get("kind") not in kinds:
            continue
        document_id = document_id_for(source.id, str(att.get("attachment_key")))
        origin = att.get("path")
        if not origin:
            errors.append({"code": "attachment_unavailable",
                           "message": f"附件无本地路径: {att.get('filename')}"})
            continue
        origin_path = Path(str(origin))
        if not origin_path.is_file():
            errors.append({"code": "attachment_missing", "message": f"附件文件不存在: {origin_path}"})
            continue
        content_fingerprint = fingerprint(origin_path)
        link = papers.get_document_link(paper_id, document_id)
        if (link and link["fingerprint"] == content_fingerprint
                and document_is_ready(document_id, workspace_id)):
            continue
        target = attachment_target(workspace_id, source.id, att)
        target.parent.mkdir(parents=True, exist_ok=True)
        staging = target.with_name(f".{target.name}.import-{uuid.uuid4().hex}{target.suffix}")
        shutil.copy2(origin_path, staging)
        generation, _ = documents_store.prepare_document_operation(
            document_id, str(att.get("filename") or target.name), document_id,
            str(target), str(staging), workspace_id=workspace_id, storage_mode="copy")
        documents_service.ingest(staging, workspace_id, job_id=None, document_id=document_id,
                                 generation=generation)
        row = documents_store.get_document(document_id, workspace_id)
        if row is None or row["status"] != "ready" or int(row["active_generation"]) != generation:
            errors.append({"code": "ingest_failed",
                           "message": f"附件索引失败: {(row or {}).get('error') or 'unknown'}"})
            continue
        papers.link_document(paper_id, document_id, attachment_key=str(att.get("attachment_key")),
                             kind=str(att.get("kind")), fingerprint=content_fingerprint,
                             indexed_at=utc_now())
        documents.append(document_id)
    if documents:
        status = "updated" if (paper_existed or had_links) else "created"
    elif errors:
        status = "failed"
    else:
        status = "created" if not paper_existed else "skipped"
    return {"item_key": item_key, "paper_id": paper_id, "status": status,
            "documents": documents, "errors": errors}

附件指纹相同且已有就绪索引时跳过；附件更新或索引失效时重新处理，索引成功后才更新关联。

条目、附件和索引文档分别记录。来源侧删除、附件丢失、类型不在导入范围或解析失败都有独立结果，不能把“条目能浏览”当作“资料已可用于问答”。

条目、附件与索引的更新关系

一个条目可能有多个附件，元数据可浏览也不表示附件可读。导入按条目记录创建、更新、跳过和失败，每个附件再关联文档；批次内单条失败不会直接抹掉其他条目的结果。

来源附件的当前指纹使用文件大小与修改时间，不是内容的加密哈希。相同指纹且关联文档就绪时跳过，指纹变化或索引失效时重新处理；因此它适合避免通常的重复索引，不能保证识别所有保持大小与时间不变的文件变更。

配置热生效影响之后如何连接来源，不会自动把整个库重新导入。浏览、按需导入、刷新和移除是独立操作，用户需要能看见哪些资料实际已进入当前工作区。

演示来源只有三条明确标注的本地资料，连接检查不访问外部服务，路径导入和 arXiv 导入只生成示例文档。真实来源的附件指纹和文件读取由后端实现。