fix: close state machine exits for archive, retry, and human nodes

Archive the job when entering ARCHIVE_JOB. Add a retry endpoint that reopens FAILED_RETRYABLE without overwriting attempt history. Escalate contract-wait timeout into an OPS task, and reject RPA complete on human nodes so only task submit can leave them.

run_mock_chain.py

@@ -179,17 +179,22 @@ def run_chain():
         for t in tasks:
             print(f"    - {t.get('title')} [{t.get('reason_type')}]")
 
-    # ---- 第7.5步：如卡在SOURCE_ADD，模拟人工完成新增货源推进 ----
+    # ---- 第7.5步：如卡在SOURCE_ADD，走待办提交，不打 RPA complete ----
     if job.get("current_node_code") == "SOURCE_ADD":
-        step("7.5", "SOURCE_ADD complete（模拟人工新增货源完成）")
+        step("7.5", "SOURCE_ADD 待办提交 CONFIRM_AND_CONTINUE")
+        source_task = next(
+            (t for t in tasks if t.get("node_code") == "SOURCE_ADD" and t.get("status") == "OPEN"),
+            None,
+        )
+        if not source_task:
+            raise RuntimeError("SOURCE_ADD 没有 OPEN 待办，无法推进")
         resp = http_post(
-            f"/api/jobs/{encode_job_id(job_id)}/nodes/SOURCE_ADD/complete",
+            f"/api/tasks/{source_task['task_id']}/submit",
             {
-                "attemptNo": 1,
+                "action": "CONFIRM_AND_CONTINUE",
-                "executorRunId": f"MOCK-SRC-{uuid.uuid4().hex[:8]}",
+                "payload": {"sourceAdded": True},
+                "submitToken": f"mock-source-{uuid.uuid4().hex}",
                 "operator": "mock-chain",
-                "summary": {"sourceAdded": True},
-                "artifacts": [],
             },
             token=AGENT_TOKEN,
         )

state_center/app/main.py

@@ -99,11 +99,13 @@ HUMAN_NODE_ROLES = {
 HUMAN_TASK_TITLES = {
     "M5_MANUAL_CONFIRM": "M5人工确认",
     "M5_RECEIPT_NOTICE": "生成收货通知单",
+    "CONTRACT_APPROVE_WAIT": "合同审批超时",
 }
 
 HUMAN_TASK_REASONS = {
     "M5_MANUAL_CONFIRM": "M5采购单已录入待保存，需要人工核对、保存并签订。",
     "M5_RECEIPT_NOTICE": "M5采购单已签订，需要生成收货通知单并回填单号。",
+    "CONTRACT_APPROVE_WAIT": "合同审批已超过等待上限，需要运维确认已通过或终止本单。",
 }
 
 # 条件节点:满足条件才执行,否则自动跳过下一节点。
@@ -157,6 +159,21 @@ def _advance(cur, job: dict, node_code: str) -> Optional[str]:
         cur.execute("UPDATE job SET status='SUCCESS',finished_at=%s,updated_at=%s WHERE job_id=%s",
                     (_now(), _now(), job_id))
         _log_event(cur, job_id, "JOB_FINISHED", None, "SYSTEM", None)
+    elif nxt == "ARCHIVE_JOB":
+        # 归档是状态中心自己的收口,进入即完成,不派 RPA、不挂人工。
+        now = _now()
+        cur.execute(
+            "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,started_at,ended_at,created_at)"
+            " VALUES(%s,%s,%s,%s,%s,%s,%s,%s)"
+            " ON DUPLICATE KEY UPDATE status='SUCCESS',ended_at=VALUES(ended_at)",
+            (job_id, "ARCHIVE_JOB", NODE_EXECUTOR["ARCHIVE_JOB"], 1, "SUCCESS", now, now, now),
+        )
+        cur.execute(
+            "UPDATE job SET status='SUCCESS',finished_at=%s,updated_at=%s,current_node_code=%s WHERE job_id=%s",
+            (now, now, "ARCHIVE_JOB", job_id),
+        )
+        _log_event(cur, job_id, "JOB_FINISHED", None, "SYSTEM", None)
+        return None
     elif nxt == "CONTRACT_APPROVE_WAIT":
         # 进等待外部态:不立刻派 RPA,挂 WAITING_EXTERNAL,到点由调度器触发查一次。
         next_check = _now() + timedelta(minutes=settings.CONTRACT_WAIT_INTERVAL_MIN)
@@ -269,11 +286,31 @@ def _mark_node_waiting_user(cur, job_id: str, node_code: str):
 
 
 def _mark_node_success(cur, job_id: str, node_code: str):
+    """把当前人工/内部节点记 SUCCESS。
+
+    已有 WAITING_USER 行则原地收口;合同审批超时这类已有 FAILED 检查历史的,
+    另起 attempt,不覆盖查审批记录。
+    """
+    cur.execute(
+        "SELECT attempt_no,status FROM job_node WHERE job_id=%s AND node_code=%s"
+        " ORDER BY attempt_no DESC LIMIT 1",
+        (job_id, node_code),
+    )
+    latest = cur.fetchone()
+    if latest and latest.get("status") == "WAITING_USER":
+        cur.execute(
+            "UPDATE job_node SET status='SUCCESS',ended_at=%s"
+            " WHERE job_id=%s AND node_code=%s AND attempt_no=%s",
+            (_now(), job_id, node_code, latest["attempt_no"]),
+        )
+        return
+
+    next_attempt = int((latest or {}).get("attempt_no") or 0) + 1
+    now = _now()
     cur.execute(
         "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,started_at,ended_at,created_at)"
-        " VALUES(%s,%s,%s,%s,'SUCCESS',%s,%s,%s)"
+        " VALUES(%s,%s,%s,%s,'SUCCESS',%s,%s,%s)",
-        " ON DUPLICATE KEY UPDATE status='SUCCESS',ended_at=VALUES(ended_at)",
+        (job_id, node_code, NODE_EXECUTOR.get(node_code, "HUMAN"), next_attempt, now, now, now),
-        (job_id, node_code, NODE_EXECUTOR.get(node_code, "HUMAN"), 1, _now(), _now(), _now()),
     )
 
 
@@ -635,6 +672,9 @@ def node_complete(job_id: str, node_code: str, req: NodeCompleteReq,
                   x_agent_token: Optional[str] = Header(default=None)):
     _check_agent(x_agent_token)
 
+    if NODE_EXECUTOR.get(node_code) == "HUMAN":
+        raise HTTPException(status_code=409, detail="人工节点只能通过待办提交，拒绝 RPA complete")
+
     # 约束:要求业务编号却没回传的节点,不许记 SUCCESS
     required = {"CONTRACT_CREATE": "contractSystemNo", "REBATE_HEADER": "rebateOrderNo",
                 "M5_RECEIPT_NOTICE": "receiptNoticeNo"}
@@ -676,16 +716,6 @@ def node_complete(job_id: str, node_code: str, req: NodeCompleteReq,
                     cur.execute(f"UPDATE job SET {col}=%s,updated_at=%s WHERE job_id=%s",
                                 (v, _now(), job_id))
         _record_job_artifacts(cur, job_id, node_code, req.attemptNo, req.artifacts)
-        if node_code == "SOURCE_ADD":
-            cur.execute(
-                "UPDATE job SET needs_source_add=0,updated_at=%s WHERE job_id=%s",
-                (_now(), job_id),
-            )
-            cur.execute(
-                "UPDATE user_task SET status='SUBMITTED',submitted_at=%s"
-                " WHERE job_id=%s AND node_code='SOURCE_ADD' AND status='OPEN'",
-                (_now(), job_id),
-            )
         _log_event(cur, job_id, "NODE_SUCCESS", node_code, "ROBOT", None, req.summary)
 
         # 决定下一步(显式转移表 + 条件节点跳过)
@@ -737,6 +767,40 @@ def node_fail(job_id: str, node_code: str, req: NodeFailReq,
     return {"ok": True, "job_status": job_status}
 
 
+@app.post("/api/jobs/{job_id}/retry")
+def retry_job(job_id: str, x_agent_token: Optional[str] = Header(default=None)):
+    """把 FAILED_RETRYABLE 的当前节点放回可 start 状态。
+
+    不覆盖失败历史:下一轮 start 用 attempt_no+1 另起 job_node 行。
+    """
+    _check_agent(x_agent_token)
+    conn = get_conn()
+    with conn.cursor() as cur:
+        job = _load_job(cur, job_id)
+        if job.get("status") != "FAILED_RETRYABLE":
+            raise HTTPException(
+                status_code=409,
+                detail=f"job 不在可重试状态(当前 {job.get('status')})",
+            )
+        node_code = job["current_node_code"]
+        cur.execute(
+            "SELECT COALESCE(MAX(attempt_no), 0) AS max_attempt FROM job_node"
+            " WHERE job_id=%s AND node_code=%s",
+            (job_id, node_code),
+        )
+        row = cur.fetchone() or {}
+        next_attempt = int(row.get("max_attempt") or 0) + 1
+        cur.execute(
+            "UPDATE job SET status='PENDING',updated_at=%s WHERE job_id=%s",
+            (_now(), job_id),
+        )
+        _log_event(
+            cur, job_id, "NODE_STARTED", node_code, "SYSTEM", None,
+            {"retry": True, "attempt_no": next_attempt},
+        )
+    return {"ok": True, "node_code": node_code, "attempt_no": next_attempt}
+
+
 # ---------------- 合同审批等待(定时查一次,不连续轮询) ----------------
 
 @app.get("/api/scheduler/due-contract-checks")
@@ -810,12 +874,13 @@ def contract_check_result(job_id: str, req: ContractCheckReq,
              req.detail, "OPS" if over_limit else None, _now(), _now(), _now()),
         )
         if over_limit:
-            # 超上限:升级 OPS 人工介入
+            # 超上限:升级 OPS 人工介入,并建一条可提交的待办。
             cur.execute(
                 "UPDATE job SET status='WAITING_USER',current_role='OPS',"
                 "wait_check_count=%s,next_check_at=NULL,updated_at=%s WHERE job_id=%s",
                 (attempt, _now(), job_id),
             )
+            _create_human_task(cur, job, "CONTRACT_APPROVE_WAIT", "OPS")
             _log_event(cur, job_id, "MANUAL_HANDOFF", "CONTRACT_APPROVE_WAIT", "SYSTEM", None,
                        {"reason": "approve_timeout", "checks": attempt})
             return {"ok": True, "escalated": True, "job_status": "WAITING_USER"}

state_center/tests/test_web_api.py

@@ -794,3 +794,355 @@ def test_select_candidate_requires_candidate_payload(monkeypatch):
 
     assert response.status_code == 400
     assert response.json()["detail"] == "选择候选货品时必须提供 productMatchId 和 goodsId"
+
+
+def test_advance_from_receipt_notice_auto_archives_job():
+    _install_fake_pymysql()
+    main = importlib.import_module("state_center.app.main")
+
+    state = {
+        "job": {
+            "job_id": "JOB-1",
+            "supplier_name": "重庆亿昊药业",
+            "current_node_code": "M5_RECEIPT_NOTICE",
+            "status": "WAITING_USER",
+            "finished_at": None,
+        },
+        "job_nodes": [],
+        "events": [],
+    }
+
+    class FakeCursor:
+        def execute(self, sql, params=None):
+            if sql.startswith("INSERT INTO job_node"):
+                state["job_nodes"].append({
+                    "node_code": params[1],
+                    "executor_type": params[2],
+                    "attempt_no": params[3],
+                    "status": params[4],
+                })
+            elif sql.startswith("UPDATE job SET status='SUCCESS'"):
+                state["job"]["status"] = "SUCCESS"
+                state["job"]["finished_at"] = params[0]
+                state["job"]["current_node_code"] = "ARCHIVE_JOB"
+            elif sql.startswith("INSERT INTO job_event"):
+                state["events"].append({"event_type": params[1], "node_code": params[2]})
+            else:
+                raise AssertionError(sql)
+
+    nxt = main._advance(FakeCursor(), state["job"], "M5_RECEIPT_NOTICE")
+
+    assert nxt is None
+    assert state["job"]["status"] == "SUCCESS"
+    assert state["job"]["current_node_code"] == "ARCHIVE_JOB"
+    assert state["job"]["finished_at"] is not None
+    assert state["job_nodes"] == [{
+        "node_code": "ARCHIVE_JOB",
+        "executor_type": "STATE_CENTER",
+        "attempt_no": 1,
+        "status": "SUCCESS",
+    }]
+    assert {"event_type": "JOB_FINISHED", "node_code": None} in state["events"]
+
+
+def test_retry_failed_retryable_node_opens_next_attempt(monkeypatch):
+    _install_fake_pymysql()
+    main = importlib.import_module("state_center.app.main")
+    client = TestClient(main.app)
+
+    state = {
+        "job": {
+            "job_id": "JOB-1",
+            "current_node_code": "ORDER_FETCH",
+            "status": "FAILED_RETRYABLE",
+            "source_type": "YYC",
+        },
+        "attempts": [
+            {"attempt_no": 1, "status": "FAILED"},
+        ],
+        "events": [],
+    }
+
+    class FakeCursor:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def execute(self, sql, params=None):
+            if "SELECT * FROM job WHERE job_id=%s" in sql:
+                self._result = [state["job"]]
+            elif "SELECT MAX(attempt_no)" in sql or "SELECT COALESCE(MAX(attempt_no)" in sql:
+                self._result = [{"max_attempt": state["attempts"][-1]["attempt_no"]}]
+            elif sql.startswith("UPDATE job SET status='PENDING'"):
+                state["job"]["status"] = "PENDING"
+            elif sql.startswith("INSERT INTO job_event"):
+                state["events"].append({"event_type": params[1], "node_code": params[2]})
+            else:
+                self._result = []
+
+        def fetchone(self):
+            return self._result[0] if self._result else None
+
+        @property
+        def _result(self):
+            return getattr(self, "_result_data", [])
+
+        @_result.setter
+        def _result(self, value):
+            self._result_data = value
+
+    class FakeConn:
+        def cursor(self):
+            return FakeCursor()
+
+    monkeypatch.setattr(main, "get_conn", lambda: FakeConn())
+    response = client.post(
+        "/api/jobs/JOB-1/retry",
+        headers={"X-Agent-Token": main.settings.AGENT_TOKEN},
+    )
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["ok"] is True
+    assert body["node_code"] == "ORDER_FETCH"
+    assert body["attempt_no"] == 2
+    assert state["job"]["status"] == "PENDING"
+    assert state["job"]["current_node_code"] == "ORDER_FETCH"
+    assert state["attempts"] == [{"attempt_no": 1, "status": "FAILED"}]
+
+
+def test_contract_approve_timeout_creates_ops_task(monkeypatch):
+    _install_fake_pymysql()
+    main = importlib.import_module("state_center.app.main")
+    client = TestClient(main.app)
+
+    state = {
+        "job": {
+            "job_id": "JOB-1",
+            "supplier_name": "重庆亿昊药业",
+            "current_node_code": "CONTRACT_APPROVE_WAIT",
+            "status": "WAITING_EXTERNAL",
+            "wait_check_count": main.settings.CONTRACT_WAIT_MAX_CHECKS - 1,
+            "source_type": "YYC",
+        },
+        "job_nodes": [],
+        "tasks": [],
+    }
+
+    class FakeCursor:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def execute(self, sql, params=None):
+            if "SELECT * FROM job WHERE job_id=%s" in sql:
+                self._result = [state["job"]]
+            elif sql.startswith("INSERT INTO job_node"):
+                state["job_nodes"].append({
+                    "node_code": params[1],
+                    "attempt_no": params[2],
+                    "status": params[3],
+                })
+            elif sql.startswith("UPDATE job SET status='WAITING_USER'"):
+                state["job"]["status"] = "WAITING_USER"
+                state["job"]["current_role"] = "OPS"
+                state["job"]["wait_check_count"] = params[0]
+            elif sql.startswith("DELETE FROM user_task"):
+                pass
+            elif sql.startswith("INSERT INTO user_task"):
+                state["tasks"].append({
+                    "job_id": params[1],
+                    "node_code": params[2],
+                    "role_code": params[3],
+                    "title": params[4],
+                })
+            elif sql.startswith("INSERT INTO job_event"):
+                pass
+            else:
+                self._result = []
+
+        def fetchone(self):
+            return self._result[0] if self._result else None
+
+        @property
+        def _result(self):
+            return getattr(self, "_result_data", [])
+
+        @_result.setter
+        def _result(self, value):
+            self._result_data = value
+
+    class FakeConn:
+        def cursor(self):
+            return FakeCursor()
+
+    monkeypatch.setattr(main, "get_conn", lambda: FakeConn())
+    response = client.post(
+        "/api/jobs/JOB-1/nodes/CONTRACT_APPROVE_WAIT/check-result",
+        headers={"X-Agent-Token": main.settings.AGENT_TOKEN},
+        json={"approved": False, "operator": "agent", "detail": "仍未审批"},
+    )
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["escalated"] is True
+    assert body["job_status"] == "WAITING_USER"
+    assert state["job"]["current_role"] == "OPS"
+    assert state["tasks"] == [{
+        "job_id": "JOB-1",
+        "node_code": "CONTRACT_APPROVE_WAIT",
+        "role_code": "OPS",
+        "title": "合同审批超时:重庆亿昊药业",
+    }]
+
+
+def test_contract_approve_timeout_task_confirm_advances_to_rebate(monkeypatch):
+    _install_fake_pymysql()
+    main = importlib.import_module("state_center.app.main")
+    client = TestClient(main.app)
+
+    state = {
+        "task": {
+            "task_id": "TASK-OPS-1",
+            "job_id": "JOB-1",
+            "node_code": "CONTRACT_APPROVE_WAIT",
+            "role_code": "OPS",
+            "status": "OPEN",
+            "submit_token": None,
+            "submitted_result_json": None,
+        },
+        "job": {
+            "job_id": "JOB-1",
+            "supplier_name": "重庆亿昊药业",
+            "current_node_code": "CONTRACT_APPROVE_WAIT",
+            "status": "WAITING_USER",
+            "current_role": "OPS",
+            "source_type": "YYC",
+        },
+        "job_nodes": [
+            {"node_code": "CONTRACT_APPROVE_WAIT", "attempt_no": 8, "status": "FAILED"},
+        ],
+    }
+
+    class FakeCursor:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def execute(self, sql, params=None):
+            if "FROM user_task WHERE task_id=%s" in sql:
+                self._result = [state["task"]]
+            elif "SELECT * FROM job WHERE job_id=%s" in sql:
+                self._result = [state["job"]]
+            elif "SELECT attempt_no,status FROM job_node" in sql:
+                self._result = [state["job_nodes"][-1]]
+            elif sql.startswith("INSERT INTO job_node"):
+                state["job_nodes"].append({
+                    "node_code": params[1],
+                    "executor_type": params[2],
+                    "attempt_no": params[3],
+                    "status": "SUCCESS",
+                })
+            elif sql.startswith("UPDATE job_node SET status='SUCCESS'"):
+                raise AssertionError("timeout confirm must not overwrite check history")
+            elif sql.startswith("UPDATE job SET current_node_code=%s,status='PENDING'"):
+                state["job"]["current_node_code"] = params[0]
+                state["job"]["status"] = "PENDING"
+            elif sql.startswith("UPDATE user_task SET status='SUBMITTED'"):
+                state["task"]["status"] = "SUBMITTED"
+                state["task"]["submit_token"] = params[0]
+                state["task"]["submitted_result_json"] = params[3]
+            elif sql.startswith("INSERT INTO job_event"):
+                pass
+            else:
+                self._result = []
+
+        def fetchone(self):
+            return self._result[0] if self._result else None
+
+        @property
+        def _result(self):
+            return getattr(self, "_result_data", [])
+
+        @_result.setter
+        def _result(self, value):
+            self._result_data = value
+
+    class FakeConn:
+        def cursor(self):
+            return FakeCursor()
+
+    monkeypatch.setattr(main, "get_conn", lambda: FakeConn())
+    response = client.post(
+        "/api/tasks/TASK-OPS-1/submit",
+        headers={"X-Agent-Token": main.settings.AGENT_TOKEN},
+        json={
+            "action": "CONFIRM_AND_CONTINUE",
+            "payload": {},
+            "submitToken": "ops-confirm-1",
+            "operator": "ops",
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["next_node"] == "REBATE_HEADER"
+    assert state["job"]["current_node_code"] == "REBATE_HEADER"
+    assert state["job_nodes"][-1]["node_code"] == "CONTRACT_APPROVE_WAIT"
+    assert state["job_nodes"][-1]["status"] == "SUCCESS"
+    assert state["job_nodes"][-1]["attempt_no"] == 9
+    assert state["job_nodes"][0]["status"] == "FAILED"
+
+
+def test_human_node_rejects_rpa_complete(monkeypatch):
+    _install_fake_pymysql()
+    main = importlib.import_module("state_center.app.main")
+    client = TestClient(main.app)
+
+    class FakeCursor:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def execute(self, sql, params=None):
+            if "SELECT * FROM job WHERE job_id=%s" in sql:
+                self._result = [{
+                    "job_id": "JOB-1",
+                    "current_node_code": "SOURCE_ADD",
+                    "status": "WAITING_USER",
+                    "source_type": "YYC",
+                }]
+            else:
+                self._result = []
+
+        def fetchone(self):
+            return self._result[0] if self._result else None
+
+        @property
+        def _result(self):
+            return getattr(self, "_result_data", [])
+
+        @_result.setter
+        def _result(self, value):
+            self._result_data = value
+
+    class FakeConn:
+        def cursor(self):
+            return FakeCursor()
+
+    monkeypatch.setattr(main, "get_conn", lambda: FakeConn())
+    response = client.post(
+        "/api/jobs/JOB-1/nodes/SOURCE_ADD/complete",
+        headers={"X-Agent-Token": main.settings.AGENT_TOKEN},
+        json={"attemptNo": 1, "summary": {"sourceAdded": True}},
+    )
+
+    assert response.status_code == 409
+    assert "人工节点" in response.json()["detail"]