Dify 接入 FastAPI 实战：保存任务、查询状态与返回报告

通过 Python 调用 Dify 之后，业务系统还需要回答几件事：这次处理属于谁、进展存在哪里、页面刷新后如何查询，以及连接中断后还能不能核实原任务。

本篇把前两篇的接口调用放进 FastAPI，用 SQLite 保存本地任务、用户归属、Dify 运行 ID 和已验证报告。客户端只提交文件、拿自己的任务 ID 查询结果；Dify 应用密钥留在后端。

示例面向单机、小批量练习，包含完整后端、启动脚本、Mac 与 Windows 命令和双用户验证步骤。它使用进程内后台任务；记录可以保留，未完成的执行不会因此自动恢复。

1. 这次搭建的调用关系

客户端使用业务 token
→ FastAPI 接收文件并保存本地 job_id
→ 后台上传文件、调用 Dify SSE
→ 尽早保存 Dify workflow_run_id
→ 校验最终输出，报告与数据写入 SQLite

客户端拿自己的 job_id
→ 查询本地状态 / 下载 Markdown 报告
→ 若状态待核实，人工触发刷新，后端只查询原 Dify 运行

对外提供四个业务接口：

本地 job_id 与 Dify 的 workflow_run_id 分别保存。业务接口不允许客户端提交任意 Dify 运行 ID 来代查结果。

2. 环境与前置工作流

资料核对日期：2026-10-22。上述依赖是本地验证基线，不代表当前最新版本；直接依赖固定版本便于复现。FastAPI 版本固定建议

业务工作流沿用第九篇的批量文件摘要：保留原始文件清单，成功、跳过和失败都形成对应记录。第十、十一篇分别介绍了普通调用和 SSE；本篇代码已经包含所需调用逻辑，不依赖它们的 Python 文件。

先在自己的 Dify 实例确认文件输入、模型配置和最终输出都能正常工作，再发布。后端不会替你创建工作流或配置模型密钥。

3. 两种凭据分工不同

3.1 Dify 应用 API Key 留在后端

后端使用环境变量 DIFY_API_KEY 调用目标 Workflow。它不出现在业务接口响应中，也不要求业务页面知道这把密钥。

3.2 业务 token 映射到固定用户

本例用 Alice、Bob 两个演示身份，通过 APP_USER_TOKENS 建立固定映射。客户端携带：

Authorization: Bearer <自己的业务 token>

HTTPBearer 负责提取凭据，代码还会验证 token 并映射 owner。每次查询都按 job_id + owner 查本地记录；别人的任务和不存在的任务都返回 404。FastAPI HTTPBearer 说明

传给 Dify 的 user 由后端生成，例如 article12:alice。客户端附加一个叫 user 的字段，也不能改变自己的归属。真实业务应把这部分替换为已有身份系统，本例的静态 token 主要用于验证隔离关系。

4. 准备文件、依赖和启动脚本

4.1 演示文件 api-notes-a.txt

星舟团队账号资料安排（API 演示资料 A）
记录日期：2026-10-22
以下人员和安排均为虚构。

王宁负责在 2026-10-27 前整理账号申请说明，补充服务台入口和必填信息。
说明必须保留“1 个工作日内反馈进度”，不能改成承诺在 1 个工作日内完成开通。
团队还决定补充常见问题清单，但没有指定负责人和截止时间。

4.2 演示文件 api-notes-b.txt

星舟团队反馈核对安排（API 演示资料 B）
记录日期：2026-10-22
以下人员和安排均为虚构。

陈澄负责在 2026-10-28 前核对 8 条试用意见，保留原文并标注重复记录。
8 条意见是待核对的记录数量，不代表已经确认 8 个独立缺陷。
有人提出自动分类的建议，但本次没有形成实施决定；正式上线日期仍未确定。

分别保存为 UTF-8 TXT，文件中的人员和任务均为虚构。首次测试只提交 A 文件；需要观察多份结果时再加入 B 文件。

4.3 依赖文件 dify-backend-requirements.txt

fastapi==0.128.0
uvicorn==0.40.0
python-multipart==0.0.21
requests==2.32.5

4.4 启动脚本 start-dify-backend.py

import getpass
import json
import os

import uvicorn

if __name__ == "__main__":
os.environ["DIFY_API_KEY"] = getpass.getpass("Dify application API Key: ").strip()
alice = getpass.getpass("Alice business token (16-128 letters/digits/_/-): ").strip()
bob = getpass.getpass("Bob business token (different from Alice): ").strip()
os.environ["APP_USER_TOKENS"] = json.dumps({"alice": alice, "bob": bob})
uvicorn.run("dify_backend_app:app", host="127.0.0.1", port=8000, workers=1)

启动时在自己的终端输入三项内容：目标 Workflow 应用 API Key、Alice 业务 token、Bob 业务 token。两枚业务 token 要不同，使用 16～128 位字母、数字、下划线或连字符，并记住它们，后面用于分别验证身份。

脚本把凭据放入当前进程环境，默认只监听本机 127.0.0.1:8000。环境变量 DIFY_BASE_URL 可调整上游地址，JOB_DATA_DIR 可调整本地数据目录。

5. 完整后端代码

保存为 dify_backend_app.py，与启动脚本放在同一目录：

"""Single-process teaching backend; persistent records, in-process background execution."""
from contextlib import asynccontextmanager, contextmanager
from datetime import datetime, timezone
import codecs
import hmac
import json
import os
from pathlib import Path
import re
import shutil
import sqlite3
import uuid

from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, UploadFile, File
from fastapi.responses import Response
from fastapi.security import HTTPBearer
import requests

MAX_BYTES = 2 * 1024 * 1024
TIMEOUT = (10, 70)
bearer = HTTPBearer(auto_error=False)

def now():
return datetime.now(timezone.utc).isoformat()

def safe_id(value):
return isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_-]{1,128}", value) is not None

@contextmanager
def database(app):
conn = sqlite3.connect(app.state.root / "jobs.sqlite3", timeout=10)
conn.row_factory = sqlite3.Row
try:
with conn:
yield conn
finally:
conn.close()

def find_job(app, job_id, owner=None):
with database(app) as conn:
if owner is None:
row = conn.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()
else:
row = conn.execute("SELECT * FROM jobs WHERE id=? AND owner=?", (job_id, owner)).fetchone()
return dict(row) if row else None

def change_job(app, job_id, expected, **values):
allowed = {"state", "dify_status", "run_id", "task_id", "last_event", "report", "data", "error_code"}
if not values or not set(values) <= allowed:
raise ValueError("Invalid internal update")
values["updated_at"] = now()
assignments = ",".join(name + "=?" for name in values)
with database(app) as conn:
cursor = conn.execute(
f"UPDATE jobs SET {assignments} WHERE id=? AND state=?",
(*values.values(), job_id, expected),
)
return cursor.rowcount == 1

def sse_objects(response):
decoder = codecs.getincrementaldecoder("utf-8")(errors="strict")
line, data_lines, after_cr = [], [], False
for chunk in response.iter_content(chunk_size=1):
for char in decoder.decode(chunk):
if after_cr and char == "\\n":
after_cr = False
continue
after_cr = char == "\\r"
if char not in "\\r\\n":
line.append(char)
continue
text = "".join(line)
line.clear()
if not text:
if data_lines:
event = json.loads("\\n".join(data_lines))
if not isinstance(event, dict):
raise ValueError("Invalid event")
yield event
data_lines = []
elif not text.startswith(":"):
field, _, value = text.partition(":")
if field == "data":
data_lines.append(value[1:] if value.startswith(" ") else value)
decoder.decode(b"", final=True)
# Never dispatch an unfinished line/frame at EOF.

def check_http(response):
if not 200 <= response.status_code < 300:
raise ValueError("Upstream HTTP error")

def validate_outputs(outputs, names):
if not isinstance(outputs, dict) or not isinstance(outputs.get("report"), str) or not outputs["report"].strip():
raise ValueError("Invalid report")
data = outputs.get("data")
keys = ("record_count", "success_count", "skipped_count", "failed_count")
if not isinstance(data, dict) or any(type(data.get(k)) is not int or data[k] < 0 for k in keys):
raise ValueError("Invalid counters")
rows = data.get("records")
if not isinstance(rows, list) or len(rows) != len(names) or data["record_count"] != len(names):
raise ValueError("Invalid record count")
counts = {"success": 0, "skipped": 0, "failed": 0}
for index, (row, name) in enumerate(zip(rows, names), start=1):
if not isinstance(row, dict) or row.get("status") not in counts:
raise ValueError("Invalid record")
if type(row.get("file_index")) is not int or row["file_index"] != index or row.get("filename") != name:
raise ValueError("Source mismatch")
counts[row["status"]] += 1
if sum(data[k] for k in keys[1:]) != len(names) or any(data[k + "_count"] != v for k, v in counts.items()):
raise ValueError("Counter mismatch")
return outputs["report"], json.dumps(data, ensure_ascii=False, allow_nan=False)

def apply_result(app, job, result, expected):
if not isinstance(result, dict):
raise ValueError("Invalid result")
if not safe_id(job["run_id"]) or result.get("id") != job["run_id"]:
raise ValueError("Run mismatch")
status = result.get("status")
values = {"dify_status": status, "error_code": None}
if status in {"succeeded", "partial-succeeded"}:
report, data = validate_outputs(result.get("outputs"), json.loads(job["names"]))
values.update(state=status, report=report, data=data)
elif status in {"failed", "stopped"}:
values.update(state=status, error_code="workflow_not_delivered")
elif status in {"running", "paused"}:
values.update(state="unknown", error_code="needs_verification")
else:
raise ValueError("Unknown upstream status")
return change_job(app, job["id"], expected, **values)

def run_job(app, job_id):
if not change_job(app, job_id, "queued", state="running"):
return
job = find_job(app, job_id)
names = json.loads(job["names"])
requested_run, identifiers = False, {}
try:
with requests.Session() as session:
session.headers.update({"Authorization": "Bearer " + app.state.dify_key})
documents = []
for index, name in enumerate(names, start=1):
with (app.state.root / job_id / f"input-{index}.txt").open("rb") as handle:
response = session.post(app.state.base + "/files/upload",
data={"user": "article12:" + job["owner"]},
files={"file": (name, handle, "text/plain")}, timeout=TIMEOUT, allow_redirects=False)
check_http(response)
upload = response.json()
if not isinstance(upload, dict) or not isinstance(upload.get("id"), str) or not upload["id"].strip():
raise ValueError("Invalid file id")
documents.append({"type": "document", "transfer_method": "local_file", "upload_file_id": upload["id"]})
requested_run = True
with session.post(app.state.base + "/workflows/run", stream=True, timeout=TIMEOUT, allow_redirects=False,
json={"inputs": {"documents": documents}, "response_mode": "streaming", "user": "article12:" + job["owner"]}) as response:
check_http(response)
if response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "text/event-stream":
raise ValueError("Invalid stream type")
for payload in sse_objects(response):
event = payload.get("event")
if not safe_id(event):
raise ValueError("Invalid event name")
sources = [payload]
if event == "workflow_paused" and isinstance(payload.get("data"), dict):
sources.append({"workflow_run_id": payload["data"].get("workflow_run_id")})
for source in sources:
for source_key, column in (("workflow_run_id", "run_id"), ("task_id", "task_id")):
value = source.get(source_key)
if value is not None:
if not safe_id(value) or (column in identifiers and identifiers[column] != value):
raise ValueError("Run identifier changed")
identifiers[column] = value
change_job(app, job_id, "running", last_event=event, **identifiers)
if event == "workflow_finished":
if not apply_result(app, find_job(app, job_id), payload.get("data"), "running"):
raise ValueError("Job state changed")
return
if event in {"error", "workflow_paused"}:
change_job(app, job_id, "running", state="unknown", error_code="needs_verification",
dify_status="paused" if event == "workflow_paused" else None)
return
raise ValueError("Stream ended without a result")
except Exception:
change_job(app, job_id, "running", state="unknown" if requested_run else "failed",
error_code="upstream_uncertain" if requested_run else "upload_failed")

def create_app():
@asynccontextmanager
async def lifespan(app):
try:
tokens = json.loads(os.environ["APP_USER_TOKENS"])
key = os.environ["DIFY_API_KEY"].strip()
base = os.getenv("DIFY_BASE_URL", "http://localhost/v1").rstrip("/")
valid = isinstance(tokens, dict) and bool(tokens)
valid = valid and all(isinstance(owner, str) and re.fullmatch(r"[A-Za-z0-9_-]{1,64}", owner)
and isinstance(token, str) and re.fullmatch(r"[A-Za-z0-9_-]{16,128}", token) for owner, token in tokens.items())
if not valid or len(set(tokens.values())) != len(tokens) or not key or not base.startswith(("http://", "https://")):
raise ValueError()
except Exception:
raise RuntimeError("Set valid APP_USER_TOKENS and DIFY_API_KEY before startup") from None
app.state.tokens, app.state.dify_key, app.state.base = tokens, key, base
app.state.root = Path(os.getenv("JOB_DATA_DIR", "./job-data")).resolve()
app.state.root.mkdir(parents=True, exist_ok=True)
with database(app) as conn:
conn.execute("""CREATE TABLE IF NOT EXISTS jobs (
id TEXT PRIMARY KEY, owner TEXT NOT NULL, state TEXT NOT NULL, names TEXT NOT NULL,
dify_status TEXT, run_id TEXT, task_id TEXT, last_event TEXT, report TEXT, data TEXT,
error_code TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)""")
conn.execute("UPDATE jobs SET state='unknown', error_code='process_restarted', updated_at=? WHERE state IN ('queued','running')", (now(),))
yield

app = FastAPI(title="Dify task backend demo", lifespan=lifespan)

def owner(credentials=Depends(bearer)):
if credentials is not None and credentials.scheme.lower() == "bearer":
for user, token in app.state.tokens.items():
if hmac.compare_digest(credentials.credentials.encode("utf-8"), token.encode("utf-8")):
return user
raise HTTPException(401, "Invalid business token", headers={"WWW-Authenticate": "Bearer"})

def owned(job_id, user):
job = find_job(app, job_id, user)
if job is None:
raise HTTPException(404, "Job not found")
return job

def public(job):
return {"job_id": job["id"], "status": job["state"], "dify_status": job["dify_status"],
"has_run_id": bool(job["run_id"]), "last_event": job["last_event"],
"files": json.loads(job["names"]), "error_code": job["error_code"],
"data": json.loads(job["data"]) if job["data"] else None}

@app.post("/jobs", status_code=202)
async def submit(background_tasks: BackgroundTasks, files: list[UploadFile] = File(...), user=Depends(owner)):
if not 1 <= len(files) <= 3:
raise HTTPException(400, "Submit 1-3 TXT files")
items = []
try:
for upload in files:
name = upload.filename or ""
if not name or len(name) > 255 or re.search(r"[/\\\\\\x00-\\x1f\\x7f]", name) or not name.lower().endswith(".txt"):
raise HTTPException(400, "Invalid TXT filename")
content = await upload.read(MAX_BYTES + 1)
if not content or len(content) > MAX_BYTES:
raise HTTPException(400, "Each TXT must contain 1 byte to 2 MiB")
items.append((name, content))
finally:
for upload in files:
await upload.close()
job_id = uuid.uuid4().hex
folder = app.state.root / job_id
folder.mkdir()
try:
for index, (_, content) in enumerate(items, start=1):
(folder / f"input-{index}.txt").write_bytes(content)
timestamp = now()
with database(app) as conn:
conn.execute("INSERT INTO jobs (id,owner,state,names,created_at,updated_at) VALUES (?,?,?,?,?,?)",
(job_id, user, "queued", json.dumps([name for name, _ in items], ensure_ascii=False), timestamp, timestamp))
except Exception:
shutil.rmtree(folder)
raise HTTPException(500, "Could not save job") from None
background_tasks.add_task(run_job, app, job_id)
return {"job_id": job_id, "status": "queued"}

@app.get("/jobs/{job_id}")
def detail(job_id: str, user=Depends(owner)):
return public(owned(job_id, user))

@app.post("/jobs/{job_id}/refresh")
def refresh(job_id: str, user=Depends(owner)):
job = owned(job_id, user)
if job["state"] != "unknown" or not job["run_id"]:
raise HTTPException(409, "Only unknown jobs with a saved run ID can be refreshed")
try:
with requests.Session() as session:
session.headers.update({"Authorization": "Bearer " + app.state.dify_key})
with session.get(app.state.base + "/workflows/run/" + job["run_id"], timeout=TIMEOUT, allow_redirects=False) as response:
check_http(response)
result = response.json()
if not apply_result(app, job, result, "unknown"):
raise HTTPException(409, "Job state changed; read the job again")
except HTTPException:
raise
except Exception:
raise HTTPException(502, "Could not verify upstream result") from None
return public(owned(job_id, user))

@app.get("/jobs/{job_id}/report")
def report(job_id: str, user=Depends(owner)):
job = owned(job_id, user)
if job["state"] not in {"succeeded", "partial-succeeded"} or not job["report"]:
raise HTTPException(409, "Report is not ready")
return Response(job["report"], media_type="text/markdown", headers={"Content-Disposition": 'attachment; filename="report.md"'})

return app

app = create_app()

代码采用 FastAPI lifespan 初始化配置和数据库。每次 SQLite 操作使用独立连接、短事务并显式关闭；上传或读取 SSE 时不持有数据库事务。SQLite 连接与线程说明、连接上下文管理器

6. 运行服务：Mac 与 Windows

6.1 macOS 终端

在代码和演示文件所在目录执行：

python3 –version
python3 -m venv .venv
.venv/bin/python -m pip install -r dify-backend-requirements.txt
export DIFY_BASE_URL="http://localhost/v1"
export JOB_DATA_DIR="./job-data"
.venv/bin/python start-dify-backend.py

6.2 Windows PowerShell

py –3 —version
py –3 –m venv .venv
.\\.venv\\Scripts\\python.exe –m pip install –r dify-backend-requirements.txt
$env:DIFY_BASE_URL = "http://localhost/v1"
$env:JOB_DATA_DIR = "./job-data"
.\\.venv\\Scripts\\python.exe start-dify–backend.py

如果没有 py 命令，用本机已安装的 Python 3 启动命令创建环境。这里直接调用虚拟环境解释器，无需激活环境或修改 PowerShell 执行策略。

将 Base URL 改成实际 Dify API 地址，保留 /v1。后端和 Dify 在不同电脑时，localhost 仍指向后端自己，应改成后端能够访问的 Dify 服务地址。

保持这个终端运行。停止服务后重新启动时，继续使用原来的数据目录并保持 owner 映射；查询时使用本次启动设置的有效 token。任务归属按 alice、bob 这样的稳定用户标识保存，更换 token 不会自动改变归属。

不要给本例增加多 worker 或 reload。启动时会将遗留 queued/running 改为 unknown，多进程或重复启动恢复逻辑可能干扰另一进程正在处理的任务。Uvicorn 运行设置

7. 用接口文档提交并查询任务

浏览器打开：

http://127.0.0.1:8000/docs

使用 FastAPI 自带的交互式接口文档，不需要另做业务网页。

7.1 以 Alice 身份提交

不要在这里填 Dify 应用密钥。提交返回格式示意：

{
"job_id": "00000000000040008000000000001201",
"status": "queued"
}

示例 ID 是本地构造值。HTTP 202 表示文件与任务记录已保存、后台处理已安排，不表示 Dify 执行成功，也不表示已经生成报告。

7.2 查询本地状态

调用 GET /jobs/{job_id}，填入刚收到的真实 ID。这个接口只读取本地数据库，不会每次都请求 Dify。

主要字段：

running 表示本地后台任务已经开始，可能还在上传文件；不能据此推断 Dify 已经进入某个模型节点。dify_status 也不是实时轮询值，不能把 null 直接判断为失败。

7.3 返回报告

当 status 为 succeeded 或 partial-succeeded，调用 GET /jobs/{job_id}/report，取得 Markdown 文本。其他状态返回 409，表示还没有可交付报告。

部分成功仍应查看 data.failed_count、data.skipped_count 和逐份记录。报告可下载，不代表每份文件都处理成功。

7.4 换成 Bob 验证归属

在接口文档中切换到 Bob 的业务 token，再用 Alice 的 job_id 调用查询、refresh 和报告接口，预期均为 404。

Bob 可以提交自己的新任务。后端会把它关联到 Bob，并使用派生的 Dify user；不需要另把 Dify 应用密钥交给 Bob。

8. 一份本地构造的完成响应

下面展示 GET /jobs/{job_id} 的结构与部分成功计数。ID、报告摘要和失败记录均来自本地构造，不是对真实 Dify 实例执行后得到的结果。

{
"job_id": "00000000000040008000000000001201",
"status": "partial-succeeded",
"dify_status": "partial-succeeded",
"has_run_id": true,
"last_event": "workflow_finished",
"files": [
"api-notes-a.txt",
"api-notes-b.txt"
],
"error_code": null,
"data": {
"record_count": 2,
"success_count": 1,
"skipped_count": 0,
"failed_count": 1,
"records": [
{
"file_index": 1,
"filename": "api-notes-a.txt",
"status": "success",
"stage": "complete",
"summary": "本地构造摘要（非模型实际回复）：王宁负责在 2026-10-27 前整理账号申请说明。",
"error_type": "",
"message": ""
},
{
"file_index": 2,
"filename": "api-notes-b.txt",
"status": "failed",
"stage": "iteration",
"summary": "",
"error_type": "IterationItemFailed",
"message": "本地模拟未返回记录，具体错误仍需查看对应运行日志"
}
]
}
}

这份接口响应不包含 Dify 应用 Key，也不允许调用者用其中的信息绕过本地 owner 检查。后端的基础校验覆盖状态、数量、序号和文件名；摘要事实和每个行动项的含义仍需业务核对。

9. 为什么要先保存文件，再交给后台

提交接口在请求中完成检查，将有效文件保存到自己的任务目录，再让后台接收 job_id 并自行打开文件。它不依赖请求中的 UploadFile 临时对象继续存在。FastAPI 文件上传说明

这不是说所有 FastAPI 版本都会在后台执行前关闭 UploadFile；具体清理顺序有版本差异。本例主动转存，是为了减少后台任务对请求临时资源生命周期的依赖。

磁盘文件用内部名称，例如 input-1.txt；数据库另存原文件名清单。上传给 Dify 时，multipart filename 仍显式使用原名，最终报告才能与用户选择的文件对应。

本例检查每份 TXT 为 1 字节～2 MiB，最多 3 份，并拒绝带路径分隔符和控制字符的原名。这是应用内处理限制，不是对整个 HTTP 入口的总请求大小控制；实际部署还需要按接入层配置请求限制。

10. unknown 的含义和人工核实

本地状态说明：

queued、unknown 是本文定义的本地状态，不是新增的 Dify 官方状态。

发生断流、EOF、无法确认的上游响应或输出检查失败时，本地转为 unknown，不自动重发工作流。收到暂停事件时也保留待核实状态，dify_status 可记录 paused。

如果 has_run_id 为 true，当前用户可以调用：

POST /jobs/<自己的本地job_id>/refresh
Authorization: Bearer <自己的业务token>

该接口只接受本地处于 unknown 且已存运行 ID 的任务。它使用后端保存的 ID 向 Dify 发 GET：

GET /v1/workflows/run/<后端已保存的workflow_run_id>
Authorization: Bearer <后端的Dify应用Key>

Dify 详情返回根级 status、outputs，没有 SSE 完成事件的 data 包装。后端会核对返回 id 与已存运行 ID 是否一致，再按原文件清单检查输出。Dify 1.17.1 运行详情实现

如果上游仍是 running 或 paused，本地保持 unknown，更新 dify_status。它不会谎称本进程又有一个后台任务在追踪；稍后需要再次人工 refresh。

如果没有保存到运行 ID，refresh 返回 409。应先核对 Dify 日志；没有 ID 不足以证明没有启动过运行。刷新也不会恢复人工输入流程或补回丢失的逐节点事件。

11. 后台任务、线程和重启边界

11.1 BackgroundTasks 不是持久任务队列

本例使用普通 def 后台函数承载同步 requests/SSE，Starlette 会将它放在线程池执行。不能把同样的同步请求直接塞进 async def 后台函数，就认为它会自动避免阻塞事件循环。Starlette 线程池说明

SQLite 保存记录，不负责后台任务的领取、确认、重试与恢复。进程退出或异常终止后，BackgroundTasks 不会靠数据库中的 queued 自动继续。FastAPI 后台任务边界、Starlette 后台任务

因此启动时只做保守转换：遗留 queued/running → unknown，不重做任务。单 worker 也不等于串行队列，多个后台任务仍可在线程间并发；本例没有实现生产级调度与限流。

11.2 较早的查询结果不能覆盖完成结果

refresh 仅允许 unknown，并在数据库更新时带上原状态条件。假设某个查询较早读到 running，但另一个操作已经保存了完成报告，旧更新不能再把完成状态改回 unknown。

条件更新失败时返回 409，调用者重新读取本地状态即可。网络等待与数据库事务分开，避免在长时间读取 SSE 时持有写事务。

11.3 保留相同数据目录

默认目录结构：

job-data/
jobs.sqlite3
<本地job_id>/
input-1.txt
input-2.txt

报告和 data 存在 SQLite 中，上传副本存在任务目录中。切换目录或删除它们，会影响原记录和文件的保留；本例没有自动清理功能。

12. 常见问题与验证范围

在自己的实例至少完成：Alice 提交与下载、Bob 无法读取 Alice 任务、一次多文件处理、一次连接异常后的原运行核实，以及重启后历史报告仍可读取。

本地已通过 15 项检查：使用真实 FastAPI ASGI 内存请求验证路由、认证、文件处理和 SQLite，模拟 Dify 上传、SSE 与详情响应；另使用 Requests 内存响应检查中文及 CRLF 分帧。检查包括缺失运行 ID 拒绝交付、终态防回退和重启保留记录。

当前没有连接真实 Dify，没有发起实际模型调用，没有在 Windows 执行命令，也没有做多进程或高并发部署验证。本文不是完整的生产任务平台；扩大使用范围时，需要把后台执行与持久任务队列、正式身份系统和数据留存策略一并设计。

下一步可以给这些业务接口增加一个简单的任务页面，让用户提交文件、查看状态并下载自己的报告。