diff --git a/run_mock_chain.py b/run_mock_chain.py index 8855cdd..9c5e470 100644 --- a/run_mock_chain.py +++ b/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, - "executorRunId": f"MOCK-SRC-{uuid.uuid4().hex[:8]}", + "action": "CONFIRM_AND_CONTINUE", + "payload": {"sourceAdded": True}, + "submitToken": f"mock-source-{uuid.uuid4().hex}", "operator": "mock-chain", - "summary": {"sourceAdded": True}, - "artifacts": [], }, token=AGENT_TOKEN, ) diff --git a/state_center/app/main.py b/state_center/app/main.py index 7f880ab..f45f338 100644 --- a/state_center/app/main.py +++ b/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)" - " ON DUPLICATE KEY UPDATE status='SUCCESS',ended_at=VALUES(ended_at)", - (job_id, node_code, NODE_EXECUTOR.get(node_code, "HUMAN"), 1, _now(), _now(), _now()), + " VALUES(%s,%s,%s,%s,'SUCCESS',%s,%s,%s)", + (job_id, node_code, NODE_EXECUTOR.get(node_code, "HUMAN"), next_attempt, 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"} diff --git a/state_center/tests/test_web_api.py b/state_center/tests/test_web_api.py index 04e9c8a..36e7ad2 100644 --- a/state_center/tests/test_web_api.py +++ b/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"]