1

README.md

@@ -162,6 +162,23 @@ uv run proposa-api
 | `/api/v1/jobs/{request_id}/progress` | GET | 查询 Step1-6 进度。 |
 | `/api/v1/jobs/{request_id}/file` | GET | 下载最终 DOCX。 |
 
+这三个任务接口的路径 ID 均可传 `request_id` 或提交时的 `txbId` 值，无需增加查询参数。
+系统优先精确匹配 `request_id`，找不到时按 `txbId` 查询最近提交的任务。
+同一个 `txbId` 多次提交会生成不同 `request_id`，旧任务仍可通过其 `request_id` 查询。
+新任务无论排队、处理中、失败还是完成，均不会自动回退旧成功任务；校验或后台提交失败
+不替换已有映射。映射按提交顺序更新，不随完成顺序改变。
+
+```text
+GET /api/v1/jobs/TEST-TXB-001
+GET /api/v1/jobs/TEST-TXB-001/progress
+GET /api/v1/jobs/TEST-TXB-001/file
+```
+
+任务查询响应中的 `request_id` 和返回的 URL 仍使用实际任务 ID。需要保证多次查询及下载
+属于同一次生成时，先用 `txbId` 查询状态，取响应 `request_id` 后固定使用它。
+映射和任务记录均为进程内存状态，服务重启后不保留；多实例需访问持有任务的实例。
+完整示例见 [API 进度查询接口对接文档](API进度查询接口对接文档.md)。
+
 ### 请求格式
 
 接口同时支持 `multipart/form-data` 文件上传和 `application/json` 服务端本地路径。

scripts/tests/test_api.py

@@ -142,6 +142,7 @@ class FinalReviewApiTests(unittest.TestCase):
         self.addCleanup(log_patch.stop)
         with http_api._JOBS_LOCK:
             http_api._JOBS.clear()
+            http_api._LATEST_REQUEST_BY_TXB_ID.clear()
 
     def _post(
         self,
@@ -188,6 +189,126 @@ class FinalReviewApiTests(unittest.TestCase):
             worker.assert_not_called()
             shutil.rmtree(work_root, ignore_errors=True)
 
+    def test_txb_id_queries_latest_submission_and_preserves_history(self):
+        """修复提示：只在提交时更新映射，不能在 worker 完成时覆盖它。"""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            executor = RecordingExecutor()
+            client = TestClient(http_api.app)
+            txb_id = "TEST-TXB-001"
+            alias = f"/api/v1/jobs/{txb_id}"
+
+            def fake_worker(_t, _p, _r, step6, _log):
+                _write_docx(step6, str(step6))
+                api_worker._write_progress(
+                    step6.parent, 6, "completed", "完成", [1, 2, 3, 4, 5, 6]
+                )
+
+            with patch.object(http_api, "API_WORK_ROOT", Path(temp_dir)), patch.object(
+                http_api, "_JOB_EXECUTOR", executor
+            ), patch.object(http_api, "_run_pipeline_worker", side_effect=fake_worker), patch.object(
+                http_api, "_send_callback"
+            ):
+                first = self._post(txb_id=txb_id).json()["request_id"]
+                second_response = self._post(txb_id=txb_id)
+                self.assertEqual(second_response.status_code, 202)
+                second = second_response.json()["request_id"]
+                self.assertNotEqual(first, second)
+                self.assertNotEqual(first, txb_id)
+                self.assertEqual(client.get(alias).json()["request_id"], second)
+                self.assertEqual(client.get(alias + "/progress").json()["status"], "queued")
+                self.assertEqual(client.get(alias + "/file").status_code, 409)
+
+                # 最新任务先完成，旧任务后完成；映射仍然指向最近提交者。
+                for function, args in reversed(executor.submitted):
+                    function(*args)
+                for suffix in ("", "/progress", "/file"):
+                    by_txb = client.get(alias + suffix)
+                    by_request = client.get(f"/api/v1/jobs/{second}{suffix}")
+                    self.assertEqual(by_txb.status_code, 200)
+                    self.assertEqual(by_txb.content, by_request.content)
+                self.assertEqual(client.get(alias).json()["request_id"], second)
+                self.assertEqual(client.get(f"/api/v1/jobs/{first}").json()["status"], "completed")
+                old_file = client.get(f"/api/v1/jobs/{first}/file")
+                self.assertEqual(old_file.status_code, 200)
+                self.assertNotEqual(old_file.content, client.get(alias + "/file").content)
+                self.assertFalse(executor.submitted[1][1][0]["job_dir"].exists())
+
+    def test_latest_failed_task_does_not_fall_back_to_old_success(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            executor = RecordingExecutor()
+            client = TestClient(http_api.app)
+            alias = "/api/v1/jobs/dms-tender-001"
+
+            def fake_worker(_t, _p, _r, step6, _log):
+                _write_docx(step6, "old successful file")
+
+            with patch.object(http_api, "API_WORK_ROOT", Path(temp_dir)), patch.object(
+                http_api, "_JOB_EXECUTOR", executor
+            ), patch.object(http_api, "_send_callback"), patch.object(
+                http_api, "_run_pipeline_worker", side_effect=fake_worker
+            ):
+                first = self._post().json()["request_id"]
+                function, args = executor.submitted[0]
+                function(*args)
+                second = self._post().json()["request_id"]
+                self.assertEqual(client.get(alias + "/file").status_code, 409)
+                function, args = executor.submitted[1]
+                api_worker._write_progress(args[0]["job_dir"], 4, "running", "处理中", [1, 2, 3])
+                with patch.object(http_api, "_run_pipeline_worker", side_effect=RuntimeError("TEST failure")):
+                    function(*args)
+                status = client.get(alias).json()
+                self.assertEqual((status["request_id"], status["status"]), (second, "failed"))
+                self.assertEqual(status["error"], "TEST failure")
+                self.assertEqual(client.get(alias + "/progress").json()["status"], "running")
+                self.assertEqual(client.get(alias + "/file").status_code, 409)
+                self.assertEqual(client.get(f"/api/v1/jobs/{first}/file").status_code, 200)
+
+    def test_rejected_submission_restores_mapping_and_invalid_input_does_not_change_it(self):
+        with tempfile.TemporaryDirectory() as temp_dir, patch.object(
+            http_api, "API_WORK_ROOT", Path(temp_dir)
+        ), patch.object(http_api, "_JOB_EXECUTOR", RecordingExecutor()) as executor:
+            first = self._post().json()["request_id"]
+            with patch.object(executor, "submit", side_effect=RuntimeError("TEST rejected")):
+                self.assertEqual(self._post().status_code, 500)
+                self.assertEqual(self._post(txb_id="TEST-NEW").status_code, 500)
+            self.assertEqual(http_api._resolve_job("dms-tender-001")["request_id"], first)
+            self.assertNotIn("TEST-NEW", http_api._LATEST_REQUEST_BY_TXB_ID)
+            files = _valid_files()
+            files["TENDER_FILE"] = ("invalid.txt", b"invalid", "text/plain")
+            self.assertEqual(self._post(files=files).status_code, 400)
+            self.assertEqual(http_api._resolve_job("dms-tender-001")["request_id"], first)
+            self.assertEqual(list(http_api._JOBS), [first])
+            self.assertEqual([p.name for p in Path(temp_dir).iterdir()], [first])
+
+    def test_submission_rollback_does_not_restore_rejected_or_overwrite_newer_task(self):
+        # 模拟重叠提交：旧提交被拒绝时，新提交已注册；新提交随后也被拒绝。
+        http_api._JOBS.update({
+            key: {"request_id": key, "txbId": "TEST-TXB"}
+            for key in ("accepted", "rejected-old", "rejected-new")
+        })
+        http_api._LATEST_REQUEST_BY_TXB_ID["TEST-TXB"] = "rejected-new"
+        http_api._discard_unsubmitted_job("rejected-old")
+        self.assertEqual(http_api._resolve_job("TEST-TXB")["request_id"], "rejected-new")
+        http_api._discard_unsubmitted_job("rejected-new")
+        self.assertEqual(http_api._resolve_job("TEST-TXB")["request_id"], "accepted")
+
+    def test_shared_lookup_unknown_ids_and_request_id_collision(self):
+        client = TestClient(http_api.app)
+        for suffix in ("", "/progress", "/file"):
+            response = client.get("/api/v1/jobs/unknown" + suffix)
+            self.assertEqual(response.status_code, 404)
+            self.assertEqual(response.json(), {"detail": "任务不存在"})
+        http_api._JOBS.update({
+            "collision": {"request_id": "collision", "status": "queued"},
+            "other": {"request_id": "other", "status": "failed"},
+        })
+        http_api._LATEST_REQUEST_BY_TXB_ID["collision"] = "other"
+        with patch.object(http_api, "_resolve_job", wraps=http_api._resolve_job) as resolver:
+            self.assertEqual(client.get("/api/v1/jobs/collision").json()["request_id"], "collision")
+            self.assertEqual(client.get("/api/v1/jobs/collision/progress").json()["status"], "queued")
+            self.assertEqual(client.get("/api/v1/jobs/collision/file").status_code, 409)
+            self.assertEqual(resolver.call_count, 3)
+
     def test_background_success_publishes_file_and_sends_callback(self):
         with tempfile.TemporaryDirectory() as temp_dir:
             root = Path(temp_dir)
@@ -337,6 +458,7 @@ class FinalReviewApiTests(unittest.TestCase):
 
             self.assertEqual(response.status_code, 202)
             self.assertEqual(job["txbId"], "1")
+            self.assertEqual(http_api._resolve_job("1")["request_id"], response.json()["request_id"])
             self.assertEqual(job["files"]["TENDER_FILE"]["filename"], "招标文件.pdf")
             self.assertEqual(job["files"]["PROCUREMENT_FILE"]["filename"], "采购需求.docx")
             self.assertEqual(job["files"]["REFERENCE_BID"]["filename"], "参考投书.docx")

src/http_api.py

@@ -55,6 +55,7 @@ _JOB_EXECUTOR = ThreadPoolExecutor(
     thread_name_prefix="proposa-api-job",
 )
 _JOBS: dict[str, dict] = {}
+_LATEST_REQUEST_BY_TXB_ID: dict[str, str] = {}
 _JOBS_LOCK = threading.Lock()
 
 
@@ -437,6 +438,34 @@ def _job_snapshot(job_id: str) -> dict | None:
         return dict(job) if job is not None else None
 
 
+def _resolve_job(identifier: str) -> dict:
+    """优先精确匹配 request_id，否则按 txbId 查询最近提交的任务。"""
+    with _JOBS_LOCK:
+        job = _JOBS.get(identifier)
+        if job is None:
+            request_id = _LATEST_REQUEST_BY_TXB_ID.get(identifier)
+            job = _JOBS.get(request_id)
+        if job is None:
+            raise HTTPException(status_code=404, detail="任务不存在")
+        return dict(job)
+
+
+def _discard_unsubmitted_job(request_id: str) -> None:
+    """撤销提交失败的任务；按注册顺序恢复映射，不覆盖更新的任务。"""
+    with _JOBS_LOCK:
+        job = _JOBS.pop(request_id, None)
+        if job is None:
+            return
+        txb_id = job["txbId"]
+        if _LATEST_REQUEST_BY_TXB_ID.get(txb_id) != request_id:
+            return
+        _LATEST_REQUEST_BY_TXB_ID.pop(txb_id, None)
+        for previous_id in reversed(_JOBS):
+            if _JOBS[previous_id].get("txbId") == txb_id:
+                _LATEST_REQUEST_BY_TXB_ID[txb_id] = previous_id
+                break
+
+
 def _read_progress_file(job_dir: Path) -> dict | None:
     progress_path = job_dir / "progress.json"
     if not progress_path.is_file():
@@ -537,9 +566,7 @@ def health() -> dict[str, str]:
 
 @app.get("/api/v1/jobs/{request_id}")
 def get_job(request_id: str) -> dict:
-    job = _job_snapshot(request_id)
-    if job is None:
-        raise HTTPException(status_code=404, detail="任务不存在")
+    job = _resolve_job(request_id)
     return {
         key: value
         for key, value in job.items()
@@ -550,9 +577,7 @@ def get_job(request_id: str) -> dict:
 
 @app.get("/api/v1/jobs/{request_id}/progress")
 def get_job_progress(request_id: str) -> dict:
-    job = _job_snapshot(request_id)
-    if job is None:
-        raise HTTPException(status_code=404, detail="任务不存在")
+    job = _resolve_job(request_id)
     progress = job.get("progress")
     if not progress:
         job_dir = job.get("job_dir")
@@ -568,9 +593,7 @@ def get_job_progress(request_id: str) -> dict:
 
 @app.get("/api/v1/jobs/{request_id}/file", response_class=FileResponse)
 def download_final_review(request_id: str) -> FileResponse:
-    job = _job_snapshot(request_id)
-    if job is None:
-        raise HTTPException(status_code=404, detail="任务不存在")
+    job = _resolve_job(request_id)
     output_path = job.get("final_output")
     if not isinstance(output_path, Path):
         raise HTTPException(status_code=409, detail="任务尚未生成可下载文件")
@@ -790,6 +813,7 @@ async def generate_final_review(
         }
         with _JOBS_LOCK:
             _JOBS[request_id] = dict(job)
+            _LATEST_REQUEST_BY_TXB_ID[txb_id] = request_id
         write_event(
             "processing_request", request_id=request_id,
             body={
@@ -814,12 +838,12 @@ async def generate_final_review(
         )
         _JOB_EXECUTOR.submit(_process_job, job)
     except HTTPException:
+        _discard_unsubmitted_job(request_id)
         shutil.rmtree(request_root, ignore_errors=True)
         raise
     except Exception as exc:
         shutil.rmtree(request_root, ignore_errors=True)
-        with _JOBS_LOCK:
-            _JOBS.pop(request_id, None)
+        _discard_unsubmitted_job(request_id)
         raise HTTPException(status_code=500, detail="后台任务提交失败") from exc
 
     response = {