旧的同步起草 POST /catalog/draft：同一个源有排队或进行中的后台起草任务时回 409 catalog_draft_busy（和建任务同一句话），不再两边同时写同一个源的目录、抢模型名额

backend/app/api/catalog.py

@@ -7,7 +7,8 @@
 - POST   /catalog/{table}/patch/preview  目录修改提案的预览（只读）：对着当前目录算改前、改后
 - POST   /catalog/{table}/patch  保存目录修改提案（乐观锁），记为人工确认
 - GET    /catalog/{table}/impact 影响面：引用这张表的已发布、受管模板（data/catalog_impact.py）
-- POST   /catalog/draft          同步起草（注释、外键、命名推断，可选模型）。界面改用下面的后台起草任务，这个留着
+- POST   /catalog/draft          同步起草（注释、外键、命名推断，可选模型）。界面改用下面的后台起草任务，这个留着；
+                                 同一个源有排队或进行中的后台任务时 409 catalog_draft_busy
 - PUT    /catalog/guidance       数据源级的起草要求（使用者补充的背景和要求，起草时交给模型）
 - POST   /catalog/draft-jobs     建后台起草任务（202）；GET /draft-jobs/active、/latest、/{job_id} 读任务；
                                  POST /draft-jobs/{job_id}/cancel 停止，/retry 只重试没写好的表（data/catalog_jobs.py）
@@ -417,8 +418,15 @@ async def draft_catalog(source_id: str, payload: CatalogDraftIn,
 
     model_items：模型给了这张表几项可用内容（没用模型、模型这部分失败时为 null）。界面据此区分「模型给了内容、
     只是和现有目录一致」（model_items > 0、三个数都是 0）和「模型没给内容」（model_error 有值）。
+
+    同一个源有排队或进行中的后台起草任务时 409 catalog_draft_busy（和建任务同一句话）：两边同时写同一个源的目录、
+    抢模型名额。
     """
     row, source = await _resolved(session, source_id)
+    try:
+        await catalog_jobs.ensure_idle(session, row)
+    except catalog_jobs.DraftBusy as e:
+        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
     cache = source.schema_cache or {}
     if not cache.get("tables"):
         raise _no_schema(cache)

backend/tests/test_catalog_draft_jobs.py

@@ -9,6 +9,7 @@
 - 目录清单：每行 last_draft（带模型留空的列数），顶层 guidance、active_job
 - 重新起草时模型这次留空、给的值没通过校验：以前起草好的模型项留着，删除只在列已经不在表结构里时才有
 - 用量记超时次数、没写出来的调用次数；任务记当前的列数上限、提醒（最近 6 次调用 4 次没写出来）；逐表记模型留空的列
+- 旧的同步起草：同一个源有排队或进行中的任务时 409 catalog_draft_busy
 - 起草要求存在数据源 options 里（上传表格的源存在它自己那一行），数据源接口不显示、整份回写 options 时留着，拼连接串时摘掉
 
 表名、列名、说明文字都是虚构的。
@@ -374,6 +375,31 @@ async def test_blank_columns_reach_the_task_the_job_and_the_list(client, monkeyp
     assert rows["tickets"]["last_draft"]["blank_columns"] == 7 and rows["gates"]["last_draft"]["blank_columns"] == 1
 
 
+async def test_the_sync_draft_is_409_while_a_job_is_queued_or_running(client, monkeypatch):
+    """旧的同步起草不和后台任务同时写同一个源的目录、抢模型名额：有排队或进行中的任务时 409（和建任务同一句话）。"""
+    sid = await _source()
+    start = catalog_jobs.manager.start
+    monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None)       # 排着、不开始
+    queued = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"]})).json()["job"]
+    r = await client.post(f"{_base(sid)}/draft", json={"tables": ["channels"]})
+    assert r.status_code == 409 and r.json()["code"] == "catalog_draft_busy", r.text
+    busy = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"]})
+    assert busy.status_code == 409 and r.json()["detail"] == busy.json()["detail"]
+    assert (await client.get(f"{_base(sid)}/channels")).json()["version"] == 0, "什么都没写"
+    await client.post(f"{_base(sid)}/draft-jobs/{queued['id']}/cancel")
+
+    monkeypatch.setattr(catalog_jobs.manager, "start", start)
+    model = Held(REPLIES, hold={"visits"})
+    use(monkeypatch, model)
+    running = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits"], "use_model": True})).json()
+    r = await client.post(f"{_base(sid)}/draft", json={"tables": ["channels"]})
+    assert r.status_code == 409 and r.json()["code"] == "catalog_draft_busy", r.text
+    model.release.set()
+    await finished(running["job"]["id"])
+    r = await client.post(f"{_base(sid)}/draft", json={"tables": ["channels"]})
+    assert r.status_code == 200, r.text
+
+
 def test_old_job_tables_get_the_new_columns():
     """升级前建的起草任务表：新列补上，存量行的计数是 0、提醒和列数上限为空。"""
     from sqlalchemy import create_engine, text