feat: add portable mock agent that follows next_node

Introduce a shared start/complete/fail shell and a mock runner so macOS can close the 13-node chain without YiSaiQi. Human nodes leave only through task submit, contract wait uses check-result, and submit tokens stay within the 64-char column.

rpa_yisaiqi/README.md

@@ -1,6 +1,19 @@
-# 艺赛旗 CONTRACT_CREATE 联调骨架
+# 艺赛旗联调包
 
-这个目录放艺赛旗侧可复制到 Windows 虚拟机的代码。第一版目标是先验证：
+这个目录是 Windows 和 macOS 共用的 RPA 包。状态机在 `state_center`，这里只负责回调外壳和节点 runner。
+
+Mac 上先用 mock runner 把契约跑通：
+
+```bash
+# 状态中心已启动、DEV_MODE=true
+python -m rpa_yisaiqi --create --auto-human
+```
+
+Agent 只听状态中心的当前节点和 `next_node`，不写死顺序。人工节点默认停下；加 `--auto-human` 才会替你提交待办。
+
+## CONTRACT_CREATE 联调骨架
+
+第一版目标是先验证：
 
 1. Windows 虚拟机能访问状态中心。
 2. `CONTRACT_CREATE` 能按约定调用 `start`。

rpa_yisaiqi/__main__.py

@@ -0,0 +1,59 @@
+"""Run the portable agent against a state center.
+
+Examples:
+  python -m rpa_yisaiqi --create --auto-human
+  python -m rpa_yisaiqi --job-id YYC-xxx --auto-human
+"""
+from __future__ import annotations
+
+import argparse
+import os
+from datetime import datetime, timedelta
+from uuid import uuid4
+
+from .agent import JobAgent
+from .callback_client import StateCenterClient
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Drive one job through the state center.")
+    parser.add_argument("--state-center-url", default=os.getenv("STATE_CENTER_URL", "http://127.0.0.1:8000"))
+    parser.add_argument("--token", default=os.getenv("AGENT_TOKEN", "dev-agent-token"))
+    parser.add_argument("--job-id")
+    parser.add_argument("--create", action="store_true", help="create a new job before running")
+    parser.add_argument("--source-type", choices=("YYC", "XYY", "OFFLINE"), default="YYC")
+    parser.add_argument("--supplier-name", default=None)
+    parser.add_argument("--order-date", default=None)
+    parser.add_argument("--auto-human", action="store_true")
+    parser.add_argument("--auto-retry", action="store_true")
+    parser.add_argument("--load-fixture-goods", action="store_true", default=True)
+    parser.add_argument("--no-load-fixture-goods", action="store_false", dest="load_fixture_goods")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    client = StateCenterClient(args.state_center_url, args.token)
+    job_id = args.job_id
+    if args.create or not job_id:
+        order_date = args.order_date or (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
+        supplier = args.supplier_name or f"重庆亿昊药业-{uuid4().hex[:4]}"
+        created = client.create_job(args.source_type, order_date, supplier, created_by="rpa-agent")
+        job_id = created["job_id"]
+        print(f"created job {job_id}")
+    if args.load_fixture_goods:
+        loaded = client.load_fixture_goods()
+        print(f"fixture goods loaded: {loaded.get('loaded')}")
+    history = JobAgent(
+        client,
+        job_id,
+        runner_mode="mock",
+        auto_human=args.auto_human,
+        auto_retry=args.auto_retry,
+    ).run()
+    last = history[-1]
+    print(f"job {job_id} stopped={last.get('stopped')} next_node={last.get('next_node')} steps={len(history)}")
+
+
+if __name__ == "__main__":
+    main()

rpa_yisaiqi/agent.py

@@ -0,0 +1,125 @@
+"""Drive one job by asking the state center where it is, then acting."""
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from .runners import mock as mock_runners
+from .runtime.shell import execute_node
+
+HUMAN_NODES = {"SOURCE_ADD", "M5_MANUAL_CONFIRM", "M5_RECEIPT_NOTICE"}
+
+
+class JobAgent:
+    def __init__(
+        self,
+        client: Any,
+        job_id: str,
+        *,
+        runner_mode: str = "mock",
+        auto_human: bool = False,
+        auto_retry: bool = False,
+        operator: str = "mock-agent",
+    ):
+        self.client = client
+        self.job_id = job_id
+        self.runner_mode = runner_mode
+        self.auto_human = auto_human
+        self.auto_retry = auto_retry
+        self.operator = operator
+
+    def step(self) -> dict[str, Any]:
+        detail = self.client.get_job(self.job_id)
+        job = detail["job"]
+        tasks = detail.get("tasks") or []
+        status = job.get("status")
+        node = job.get("current_node_code")
+
+        if status == "SUCCESS":
+            return {"stopped": "SUCCESS", "next_node": None}
+        if status == "FAILED_RETRYABLE":
+            if not self.auto_retry:
+                return {"stopped": "FAILED_RETRYABLE"}
+            return self.client.retry_job(self.job_id)
+        if status == "WAITING_USER" or node in HUMAN_NODES:
+            if not self.auto_human:
+                return {"stopped": "WAITING_USER"}
+            return self._submit_human(job, tasks)
+        if node == "PRODUCT_MATCH":
+            return self.client.match_run(self.job_id)
+        if node == "CONTRACT_APPROVE_WAIT":
+            return self.client.check_contract_result(self.job_id, True, operator=self.operator)
+        if node == "ARCHIVE_JOB":
+            return {"stopped": "SUCCESS", "next_node": None}
+        return execute_node(
+            self.client,
+            self.job_id,
+            node,
+            lambda: self._run(node, job),
+            attempt_no=int(job.get("attempt_no") or 1),
+            operator=self.operator,
+        )
+
+    def _run(self, node_code: str, job: dict[str, Any]) -> dict[str, Any]:
+        if self.runner_mode != "mock":
+            raise RuntimeError(f"unsupported runner_mode {self.runner_mode}")
+        return mock_runners.run(node_code, job, self.client)
+
+    def _submit_human(self, job: dict[str, Any], tasks: list[dict[str, Any]]) -> dict[str, Any]:
+        node = job["current_node_code"]
+        task = _open_task(tasks, node)
+        if task is None and hasattr(self.client, "list_tasks"):
+            listed = self.client.list_tasks(status="OPEN")
+            task = _open_task(listed.get("items") or [], node, self.job_id)
+        if task is None:
+            return {"stopped": "WAITING_USER", "reason": "no_open_task"}
+
+        action = "CONFIRM_AND_CONTINUE"
+        payload: dict[str, Any] = {}
+        if node == "PRODUCT_MATCH":
+            action = "SELECT_CANDIDATE"
+            payload = {
+                "productMatchId": task.get("product_match_id"),
+                "goodsId": "G-001",
+                "goodsCode": "YP-001",
+                "baseUnit": "盒",
+                "saveAsAlias": True,
+            }
+        elif node == "M5_RECEIPT_NOTICE":
+            payload = {"receiptNoticeNo": f"SH{uuid.uuid4().hex[:8].upper()}"}
+
+        return self.client.submit_task(
+            task["task_id"],
+            action,
+            payload,
+            submit_token=uuid.uuid4().hex,
+            operator=self.operator,
+        )
+
+    def run(self, *, max_steps: int = 40) -> list[dict[str, Any]]:
+        history: list[dict[str, Any]] = []
+        for _ in range(max_steps):
+            result = self.step()
+            history.append(result)
+            if result.get("stopped") in {"SUCCESS", "WAITING_USER", "FAILED_RETRYABLE", "FAILED_MANUAL"}:
+                return history
+            if result.get("waiting_user"):
+                continue
+            if result.get("next_node") is None and result.get("ok"):
+                detail = self.client.get_job(self.job_id)
+                if detail["job"].get("status") == "SUCCESS":
+                    history.append({"stopped": "SUCCESS", "next_node": None})
+                return history
+        raise RuntimeError(f"job {self.job_id} exceeded {max_steps} agent steps")
+
+
+def _open_task(tasks: list[dict[str, Any]], node_code: str, job_id: str | None = None):
+    for task in tasks:
+        if task.get("node_code") != node_code:
+            continue
+        if task.get("status") not in (None, "OPEN"):
+            continue
+        if job_id and task.get("job_id") not in (None, job_id):
+            continue
+        return task
+    return None

rpa_yisaiqi/callback_client.py

@@ -65,16 +65,88 @@ class StateCenterClient:
     ) -> JsonDict:
         return self._post(f"/api/jobs/{_path_part(job_id)}/nodes/{_path_part(node_code)}/fail", payload)
 
+    def get_job(self, job_id: str) -> JsonDict:
+        return self._request("GET", f"/api/jobs/{_path_part(job_id)}")
+
+    def ingest_items(self, job_id: str, rows: list[JsonDict], replace: bool = True) -> JsonDict:
+        return self._post(
+            f"/api/jobs/{_path_part(job_id)}/items",
+            {"replace": replace, "rows": rows},
+        )
+
+    def match_run(self, job_id: str) -> JsonDict:
+        return self._post(f"/api/jobs/{_path_part(job_id)}/match/run", {})
+
+    def load_fixture_goods(self) -> JsonDict:
+        return self._post("/api/dev/match/load-fixture-goods", {})
+
+    def retry_job(self, job_id: str) -> JsonDict:
+        return self._post(f"/api/jobs/{_path_part(job_id)}/retry", {})
+
+    def list_tasks(self, *, role_code: str | None = None, status: str | None = None) -> JsonDict:
+        query = []
+        if role_code:
+            query.append(f"roleCode={quote(role_code, safe='')}")
+        if status:
+            query.append(f"status={quote(status, safe='')}")
+        suffix = f"?{'&'.join(query)}" if query else ""
+        return self._request("GET", f"/api/tasks/my{suffix}")
+
+    def submit_task(
+        self,
+        task_id: str,
+        action: str,
+        payload: JsonDict,
+        submit_token: str,
+        operator: str = "agent",
+    ) -> JsonDict:
+        return self._post(
+            f"/api/tasks/{_path_part(task_id)}/submit",
+            {
+                "action": action,
+                "payload": payload,
+                "submitToken": submit_token,
+                "operator": operator,
+            },
+        )
+
+    def check_contract_result(
+        self,
+        job_id: str,
+        approved: bool,
+        *,
+        operator: str = "agent",
+        detail: str | None = None,
+    ) -> JsonDict:
+        return self._post(
+            f"/api/jobs/{_path_part(job_id)}/nodes/CONTRACT_APPROVE_WAIT/check-result",
+            {"approved": approved, "operator": operator, "detail": detail},
+        )
+
+    def create_job(self, source_type: str, order_date: str, supplier_name: str, created_by: str = "agent") -> JsonDict:
+        return self._post(
+            "/api/jobs",
+            {
+                "source_type": source_type,
+                "order_date": order_date,
+                "supplier_name": supplier_name,
+                "created_by": created_by,
+            },
+        )
+
     def _post(self, path: str, payload: JsonDict) -> JsonDict:
-        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        return self._request("POST", path, payload)
+
+    def _request(self, method: str, path: str, payload: JsonDict | None = None) -> JsonDict:
+        data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None
+        headers = {"X-Agent-Token": self.token}
+        if data is not None:
+            headers["Content-Type"] = "application/json"
         request = Request(
             f"{self.base_url}{path}",
             data=data,
-            headers={
-                "Content-Type": "application/json",
-                "X-Agent-Token": self.token,
-            },
-            method="POST",
+            headers=headers,
+            method=method,
         )
         try:
             with self._opener(request, self.timeout) as response:

rpa_yisaiqi/contract_create_node.py

@@ -17,34 +17,31 @@ from .callback_client import StateCenterClient
 from .contract_draft import ContractDraft, load_contract_draft
 from .contract_flow_steps import ContractFlowContext, run_contract_create_flow
 from .pdf_utils import ensure_merged_pdf
+from .runtime.errors import LoginFailed as RuntimeLoginFailed
+from .runtime.errors import NodeError
+from .runtime.errors import PdfMergeUnsupported as RuntimePdfMergeUnsupported
+from .runtime.errors import SupplierContractDisabled as RuntimeSupplierContractDisabled
+from .runtime.shell import execute_node
 from .yisaiqi_adapter_template import YiSaiQiBrowserAdapter
 
 NODE_CODE = "CONTRACT_CREATE"
 CONTRACT_URL = "http://contract-test.gzmpc.com:8080/"
 
 
-class ContractCreateError(RuntimeError):
-    error_code = "UI_ELEMENT_NOT_FOUND"
-    retryable = False
-    route_role = "OPS"
+class ContractCreateError(NodeError):
+    pass
 
 
-class LoginFailed(ContractCreateError):
-    error_code = "SYS_LOGIN_FAIL"
-    retryable = True
-    route_role = "OPS"
+class LoginFailed(RuntimeLoginFailed, ContractCreateError):
+    pass
 
 
-class SupplierContractDisabled(ContractCreateError):
-    error_code = "BIZ_CONTRACT_DISABLED"
-    retryable = False
-    route_role = "PURCHASER"
+class SupplierContractDisabled(RuntimeSupplierContractDisabled, ContractCreateError):
+    pass
 
 
-class PdfMergeUnsupported(ContractCreateError):
-    error_code = "CAP_PDF_MERGE_UNSUPPORTED"
-    retryable = False
-    route_role = "OPS"
+class PdfMergeUnsupported(RuntimePdfMergeUnsupported, ContractCreateError):
+    pass
 
 
 @dataclass(frozen=True)
@@ -72,15 +69,25 @@ def execute_contract_create(
     *,
     contract_runner: ContractRunner | None = None,
 ) -> dict[str, Any]:
-    client.start_node(spec.job_id, NODE_CODE, attempt_no=spec.attempt_no)
     runner = contract_runner or create_contract_in_contract_system
-    try:
+
+    def _run() -> dict[str, Any]:
         contract_no = runner(spec)
-        complete_payload = build_complete_payload(spec, contract_no)
-        return client.complete_node(spec.job_id, NODE_CODE, complete_payload)
-    except ContractCreateError as exc:
-        fail_payload = build_fail_payload(spec, exc)
-        return client.fail_node(spec.job_id, NODE_CODE, fail_payload)
+        payload = build_complete_payload(spec, contract_no)
+        return {
+            "businessRefs": payload["businessRefs"],
+            "summary": payload["summary"],
+            "artifacts": payload["artifacts"],
+        }
+
+    return execute_node(
+        client,
+        spec.job_id,
+        NODE_CODE,
+        _run,
+        attempt_no=spec.attempt_no,
+        operator="agent",
+    )
 
 
 def create_contract_in_contract_system(spec: ContractCreateInput) -> str:

rpa_yisaiqi/runners/__init__.py

@@ -0,0 +1 @@
+"""Node runners. Mock now, Windows page runners later."""

rpa_yisaiqi/runners/mock.py

@@ -0,0 +1,60 @@
+"""Mock runners for Mac-side state-machine verification."""
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from ..runtime.errors import NodeError
+
+
+def run(node_code: str, job: dict[str, Any], client: Any) -> dict[str, Any]:
+    if node_code == "ORDER_FETCH":
+        return run_order_fetch(job, client)
+    if node_code == "OCR_EXTRACT":
+        return {"summary": {"ocr": "mock"}}
+    if node_code == "CONTRACT_CREATE":
+        contract_no = f"HT{str(job.get('order_date') or '').replace('-', '')}{uuid.uuid4().hex[:5].upper()}"
+        return {"businessRefs": {"contractSystemNo": contract_no}, "summary": {"sourceType": job.get("source_type")}}
+    if node_code == "REBATE_HEADER":
+        return {"businessRefs": {"rebateOrderNo": f"FL-{uuid.uuid4().hex[:8].upper()}"}}
+    if node_code == "REBATE_IMPORT":
+        return {"summary": {"imported": 1, "failed": 0}}
+    if node_code == "REBATE_SUBMIT":
+        return {"summary": {"submitted": True}}
+    if node_code == "M5_CREATE_ORDER":
+        return {"businessRefs": {"m5OrderNo": f"M5-{uuid.uuid4().hex[:8].upper()}"}, "summary": {"autoSave": False}}
+    if node_code == "ARCHIVE_JOB":
+        return {"summary": {"archived": True}}
+    raise NodeError(f"no mock runner for {node_code}")
+
+
+def run_order_fetch(job: dict[str, Any], client: Any) -> dict[str, Any]:
+    order_date = str(job.get("order_date") or "2026-08-12")
+    order_no = f"DD-{order_date.replace('-', '')}-001"
+    rows = [
+        {
+            "orderNo": order_no,
+            "rawProductName": "阿卡波糖片",
+            "platformProductCode": "YP-001",
+            "spec": "50mg*30片",
+            "manufacturer": "示例制药",
+            "approvalNo": "国药准字H00000001",
+            "qty": 100,
+            "unitPrice": 25.5,
+            "sourceRowNo": 1,
+        }
+    ]
+    if hasattr(client, "ingest_items"):
+        client.ingest_items(job["job_id"], rows, replace=True)
+    return {
+        "summary": {
+            "orderCount": 1,
+            "supplierList": [job.get("supplier_name")],
+            "orderNoList": [order_no],
+        },
+        "artifacts": [{
+            "artifactType": "SOURCE_EXCEL",
+            "storagePath": f"/tmp/mock_rpa_data/orders_{job.get('supplier_name')}_{order_date}.xlsx",
+            "fileHash": "sha256:mock-order-excel",
+        }],
+    }

rpa_yisaiqi/runtime/__init__.py

@@ -0,0 +1 @@
+"""Portable node runtime shared by mock and Windows runners."""

rpa_yisaiqi/runtime/errors.py

@@ -0,0 +1,25 @@
+"""Node errors that map onto state-center fail callbacks."""
+
+
+class NodeError(RuntimeError):
+    error_code = "UI_ELEMENT_NOT_FOUND"
+    retryable = False
+    route_role = "OPS"
+
+
+class LoginFailed(NodeError):
+    error_code = "SYS_LOGIN_FAIL"
+    retryable = True
+    route_role = "OPS"
+
+
+class SupplierContractDisabled(NodeError):
+    error_code = "BIZ_CONTRACT_DISABLED"
+    retryable = False
+    route_role = "PURCHASER"
+
+
+class PdfMergeUnsupported(NodeError):
+    error_code = "CAP_PDF_MERGE_UNSUPPORTED"
+    retryable = False
+    route_role = "OPS"

rpa_yisaiqi/runtime/payloads.py

@@ -0,0 +1,37 @@
+"""Complete/fail payload builders for RPA nodes."""
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from .errors import NodeError
+
+
+def build_complete_payload(
+    job_id: str,
+    node_code: str,
+    attempt_no: int,
+    output: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    output = output or {}
+    refs = output.get("businessRefs") or {}
+    ref_key = next(iter(refs.values()), "ok")
+    return {
+        "attemptNo": attempt_no,
+        "executorRunId": f"YSQ-{node_code}-{uuid.uuid4().hex[:12]}",
+        "idempotencyKey": f"{job_id}:{node_code}:{attempt_no}:{ref_key}",
+        "businessRefs": refs,
+        "summary": output.get("summary") or {},
+        "artifacts": output.get("artifacts") or [],
+    }
+
+
+def build_fail_payload(attempt_no: int, exc: NodeError) -> dict[str, Any]:
+    return {
+        "attemptNo": attempt_no,
+        "errorCode": exc.error_code,
+        "errorSummary": str(exc),
+        "retryable": exc.retryable,
+        "routeRole": exc.route_role,
+        "artifacts": [],
+    }

rpa_yisaiqi/runtime/shell.py

@@ -0,0 +1,29 @@
+"""Generic start → runner → complete/fail shell."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from ..callback_client import StateCenterClient
+from .errors import NodeError
+from .payloads import build_complete_payload, build_fail_payload
+
+Runner = Callable[[], dict[str, Any] | None]
+
+
+def execute_node(
+    client: StateCenterClient,
+    job_id: str,
+    node_code: str,
+    runner: Runner,
+    *,
+    attempt_no: int = 1,
+    operator: str = "agent",
+) -> dict[str, Any]:
+    client.start_node(job_id, node_code, attempt_no=attempt_no, operator=operator)
+    try:
+        output = runner() or {}
+        payload = build_complete_payload(job_id, node_code, attempt_no, output)
+        return client.complete_node(job_id, node_code, payload)
+    except NodeError as exc:
+        return client.fail_node(job_id, node_code, build_fail_payload(attempt_no, exc))

rpa_yisaiqi/tests/test_agent.py

@@ -0,0 +1,240 @@
+from rpa_yisaiqi.agent import JobAgent
+
+
+class FakeClient:
+    def __init__(self, job, *, tasks=None):
+        self.job = dict(job)
+        self.tasks = list(tasks or [])
+        self.calls = []
+
+    def get_job(self, job_id):
+        self.calls.append(("get_job", job_id))
+        return {"job": dict(self.job), "tasks": list(self.tasks)}
+
+    def start_node(self, job_id, node_code, **kwargs):
+        self.calls.append(("start", node_code, kwargs))
+        return {"ok": True}
+
+    def complete_node(self, job_id, node_code, payload):
+        self.calls.append(("complete", node_code, payload))
+        return {"ok": True, "next_node": "PRODUCT_MATCH"}
+
+    def fail_node(self, job_id, node_code, payload):
+        self.calls.append(("fail", node_code, payload))
+        return {"ok": True, "job_status": "FAILED_RETRYABLE"}
+
+    def ingest_items(self, job_id, rows, replace=True):
+        self.calls.append(("ingest_items", job_id, rows))
+        return {"ok": True, "inserted": len(rows)}
+
+    def match_run(self, job_id):
+        self.calls.append(("match_run", job_id))
+        return {"ok": True, "next_node": "CONTRACT_CREATE"}
+
+    def list_tasks(self, **kwargs):
+        self.calls.append(("list_tasks", kwargs))
+        return {"items": list(self.tasks)}
+
+    def submit_task(self, task_id, action, payload, submit_token, operator="agent"):
+        self.calls.append(("submit_task", task_id, action, payload, submit_token))
+        return {"ok": True, "next_node": "CONTRACT_CREATE"}
+
+    def check_contract_result(self, job_id, approved, **kwargs):
+        self.calls.append(("check_result", job_id, approved))
+        return {"ok": True, "next_node": "REBATE_HEADER"}
+
+    def retry_job(self, job_id):
+        self.calls.append(("retry", job_id))
+        return {"ok": True, "node_code": self.job["current_node_code"], "attempt_no": 2}
+
+
+def test_agent_does_not_hardcode_next_node_after_order_fetch():
+    client = FakeClient({
+        "job_id": "JOB-1",
+        "current_node_code": "ORDER_FETCH",
+        "status": "PENDING",
+        "source_type": "YYC",
+        "supplier_name": "重庆亿昊药业",
+        "order_date": "2026-08-12",
+    })
+    agent = JobAgent(client, "JOB-1", runner_mode="mock")
+
+    result = agent.step()
+
+    assert result["next_node"] == "PRODUCT_MATCH"
+    assert ("complete", "ORDER_FETCH") == (client.calls[-1][0], client.calls[-1][1])
+    assert not any(call[0] == "complete" and call[1] == "OCR_EXTRACT" for call in client.calls)
+
+
+def test_agent_runs_product_match_via_state_center_not_rpa_callback():
+    client = FakeClient({
+        "job_id": "JOB-1",
+        "current_node_code": "PRODUCT_MATCH",
+        "status": "PENDING",
+        "source_type": "YYC",
+    })
+    agent = JobAgent(client, "JOB-1", runner_mode="mock")
+
+    result = agent.step()
+
+    assert result["next_node"] == "CONTRACT_CREATE"
+    assert client.calls[1][0] == "match_run"
+    assert not any(call[0] in {"start", "complete"} for call in client.calls)
+
+
+def test_agent_auto_human_submits_source_add_task_not_node_complete():
+    job_id = "YYC-重庆亿昊药业-2fa0-20260812-16abff"
+    client = FakeClient(
+        {
+            "job_id": job_id,
+            "current_node_code": "SOURCE_ADD",
+            "status": "WAITING_USER",
+            "source_type": "YYC",
+        },
+        tasks=[{"task_id": "T-1", "node_code": "SOURCE_ADD", "status": "OPEN"}],
+    )
+    agent = JobAgent(client, job_id, runner_mode="mock", auto_human=True)
+
+    result = agent.step()
+
+    assert result["next_node"] == "CONTRACT_CREATE"
+    assert client.calls[-1][:3] == ("submit_task", "T-1", "CONFIRM_AND_CONTINUE")
+    assert len(client.calls[-1][4]) <= 64
+    assert not any(call[0] == "complete" for call in client.calls)
+
+
+def test_agent_stops_on_human_node_when_auto_human_disabled():
+    client = FakeClient({
+        "job_id": "JOB-1",
+        "current_node_code": "SOURCE_ADD",
+        "status": "WAITING_USER",
+        "source_type": "YYC",
+    })
+    agent = JobAgent(client, "JOB-1", runner_mode="mock", auto_human=False)
+
+    result = agent.step()
+
+    assert result["stopped"] == "WAITING_USER"
+    assert not any(call[0] == "submit_task" for call in client.calls)
+
+
+def test_agent_contract_wait_uses_check_result_not_start_complete():
+    client = FakeClient({
+        "job_id": "JOB-1",
+        "current_node_code": "CONTRACT_APPROVE_WAIT",
+        "status": "WAITING_EXTERNAL",
+        "source_type": "YYC",
+    })
+    agent = JobAgent(client, "JOB-1", runner_mode="mock")
+
+    result = agent.step()
+
+    assert result["next_node"] == "REBATE_HEADER"
+    assert client.calls[-1][:2] == ("check_result", "JOB-1")
+    assert not any(call[0] in {"start", "complete"} for call in client.calls)
+
+
+def test_agent_retries_failed_retryable_then_stops_if_still_failed():
+    client = FakeClient({
+        "job_id": "JOB-1",
+        "current_node_code": "ORDER_FETCH",
+        "status": "FAILED_RETRYABLE",
+        "source_type": "YYC",
+    })
+    agent = JobAgent(client, "JOB-1", runner_mode="mock", auto_retry=True)
+
+    result = agent.step()
+
+    assert ("retry", "JOB-1") in client.calls
+    assert result["attempt_no"] == 2
+
+
+class ScriptedClient(FakeClient):
+    """Advance through canned job states after each action. Sequence is not in the agent."""
+
+    def __init__(self, frames):
+        self.frames = list(frames)
+        super().__init__(self.frames[0]["job"], tasks=self.frames[0].get("tasks") or [])
+
+    def _advance(self):
+        if len(self.frames) > 1:
+            self.frames.pop(0)
+            self.job = dict(self.frames[0]["job"])
+            self.tasks = list(self.frames[0].get("tasks") or [])
+
+    def complete_node(self, job_id, node_code, payload):
+        nxt = self.frames[0].get("next_node")
+        self._advance()
+        self.calls.append(("complete", node_code, payload))
+        return {"ok": True, "next_node": nxt}
+
+    def match_run(self, job_id):
+        nxt = self.frames[0].get("next_node")
+        self._advance()
+        self.calls.append(("match_run", job_id))
+        return {"ok": True, "next_node": nxt}
+
+    def submit_task(self, task_id, action, payload, submit_token, operator="agent"):
+        nxt = self.frames[0].get("next_node")
+        self._advance()
+        self.calls.append(("submit_task", task_id, action, payload))
+        return {"ok": True, "next_node": nxt}
+
+    def check_contract_result(self, job_id, approved, **kwargs):
+        nxt = self.frames[0].get("next_node")
+        self._advance()
+        self.calls.append(("check_result", job_id, approved))
+        return {"ok": True, "next_node": nxt}
+
+
+def test_agent_run_follows_state_center_until_success():
+    client = ScriptedClient([
+        {"job": {"job_id": "JOB-1", "current_node_code": "ORDER_FETCH", "status": "PENDING",
+                 "source_type": "YYC", "supplier_name": "重庆亿昊药业", "order_date": "2026-08-12"},
+         "next_node": "PRODUCT_MATCH"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "PRODUCT_MATCH", "status": "PENDING",
+                 "source_type": "YYC"},
+         "next_node": "CONTRACT_CREATE"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "CONTRACT_CREATE", "status": "PENDING",
+                 "source_type": "YYC", "order_date": "2026-08-12"},
+         "next_node": "CONTRACT_APPROVE_WAIT"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "CONTRACT_APPROVE_WAIT",
+                 "status": "WAITING_EXTERNAL", "source_type": "YYC"},
+         "next_node": "REBATE_HEADER"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "REBATE_HEADER", "status": "PENDING",
+                 "source_type": "YYC"},
+         "next_node": "REBATE_IMPORT"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "REBATE_IMPORT", "status": "PENDING",
+                 "source_type": "YYC"},
+         "next_node": "REBATE_SUBMIT"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "REBATE_SUBMIT", "status": "PENDING",
+                 "source_type": "YYC"},
+         "next_node": "M5_CREATE_ORDER"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "M5_CREATE_ORDER", "status": "PENDING",
+                 "source_type": "YYC"},
+         "next_node": "M5_MANUAL_CONFIRM"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "M5_MANUAL_CONFIRM", "status": "WAITING_USER",
+                 "source_type": "YYC"},
+         "tasks": [{"task_id": "T-M5", "node_code": "M5_MANUAL_CONFIRM", "status": "OPEN"}],
+         "next_node": "M5_RECEIPT_NOTICE"},
+        {"job": {"job_id": "JOB-1", "current_node_code": "M5_RECEIPT_NOTICE", "status": "WAITING_USER",
+                 "source_type": "YYC"},
+         "tasks": [{"task_id": "T-SH", "node_code": "M5_RECEIPT_NOTICE", "status": "OPEN"}],
+         "next_node": None},
+        {"job": {"job_id": "JOB-1", "current_node_code": "ARCHIVE_JOB", "status": "SUCCESS",
+                 "source_type": "YYC"}},
+    ])
+    agent = JobAgent(client, "JOB-1", runner_mode="mock", auto_human=True)
+
+    history = agent.run()
+
+    acted = [call[0] for call in client.calls if call[0] in {
+        "complete", "match_run", "check_result", "submit_task",
+    }]
+    assert acted == [
+        "complete", "match_run", "complete", "check_result",
+        "complete", "complete", "complete", "complete",
+        "submit_task", "submit_task",
+    ]
+    assert history[-1]["stopped"] == "SUCCESS"
+    assert not any(call[1] == "OCR_EXTRACT" for call in client.calls if call[0] == "complete")

rpa_yisaiqi/tests/test_callback_client.py

@@ -0,0 +1,78 @@
+import json
+
+from rpa_yisaiqi.callback_client import StateCenterClient
+
+
+class FakeHttp:
+    def __init__(self, responses=None):
+        self.requests = []
+        self.responses = list(responses or [{"ok": True}])
+
+    def __call__(self, request, timeout):
+        body = json.loads(request.data.decode("utf-8")) if request.data else None
+        self.requests.append({
+            "url": request.full_url,
+            "method": request.get_method(),
+            "body": body,
+            "headers": dict(request.header_items()),
+        })
+        return FakeResponse(self.responses.pop(0))
+
+
+class FakeResponse:
+    def __init__(self, payload):
+        self.payload = payload
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def read(self):
+        return json.dumps(self.payload).encode("utf-8")
+
+
+def test_get_job_uses_get_and_token():
+    http = FakeHttp([{"job": {"job_id": "JOB-1"}, "tasks": []}])
+    client = StateCenterClient("http://state-center:8000", "dev-agent-token", opener=http)
+
+    result = client.get_job("JOB-1")
+
+    assert result["job"]["job_id"] == "JOB-1"
+    assert http.requests[0]["method"] == "GET"
+    assert http.requests[0]["url"] == "http://state-center:8000/api/jobs/JOB-1"
+    assert http.requests[0]["headers"]["X-agent-token"] == "dev-agent-token"
+
+
+def test_ingest_items_posts_rows():
+    http = FakeHttp([{"ok": True, "inserted": 1}])
+    client = StateCenterClient("http://state-center:8000", "dev-agent-token", opener=http)
+
+    client.ingest_items("JOB-1", [{"orderNo": "DD-1", "rawProductName": "阿卡波糖片", "qty": 1}])
+
+    assert http.requests[0]["url"].endswith("/api/jobs/JOB-1/items")
+    assert http.requests[0]["body"]["replace"] is True
+    assert http.requests[0]["body"]["rows"][0]["orderNo"] == "DD-1"
+
+
+def test_match_run_and_retry_and_task_submit():
+    http = FakeHttp([
+        {"ok": True, "next_node": "CONTRACT_CREATE"},
+        {"ok": True, "attempt_no": 2},
+        {"ok": True, "next_node": "CONTRACT_CREATE"},
+        {"ok": True, "next_node": "REBATE_HEADER"},
+    ])
+    client = StateCenterClient("http://state-center:8000", "dev-agent-token", opener=http)
+
+    assert client.match_run("JOB-1")["next_node"] == "CONTRACT_CREATE"
+    assert client.retry_job("JOB-1")["attempt_no"] == 2
+    assert client.submit_task("T-1", "CONFIRM_AND_CONTINUE", {}, "tok")["next_node"] == "CONTRACT_CREATE"
+    assert client.check_contract_result("JOB-1", True)["next_node"] == "REBATE_HEADER"
+
+    assert http.requests[0]["url"].endswith("/api/jobs/JOB-1/match/run")
+    assert http.requests[1]["url"].endswith("/api/jobs/JOB-1/retry")
+    assert http.requests[2]["url"].endswith("/api/tasks/T-1/submit")
+    assert http.requests[2]["body"]["action"] == "CONFIRM_AND_CONTINUE"
+    assert http.requests[3]["url"].endswith("/nodes/CONTRACT_APPROVE_WAIT/check-result")
+    assert http.requests[3]["body"]["approved"] is True

rpa_yisaiqi/tests/test_runtime_shell.py

@@ -0,0 +1,75 @@
+import json
+
+from rpa_yisaiqi.callback_client import StateCenterClient
+from rpa_yisaiqi.runtime.errors import LoginFailed, NodeError
+from rpa_yisaiqi.runtime.shell import execute_node
+
+
+class FakeHttp:
+    def __init__(self, responses=None):
+        self.requests = []
+        self.responses = list(responses or [{"ok": True}, {"ok": True, "next_node": "PRODUCT_MATCH"}])
+
+    def __call__(self, request, timeout):
+        body = json.loads(request.data.decode("utf-8")) if request.data else None
+        self.requests.append({
+            "url": request.full_url,
+            "method": request.get_method(),
+            "body": body,
+        })
+        return FakeResponse(self.responses.pop(0))
+
+
+class FakeResponse:
+    def __init__(self, payload):
+        self.payload = payload
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def read(self):
+        return json.dumps(self.payload).encode("utf-8")
+
+
+def test_execute_node_starts_then_completes_with_runner_payload():
+    http = FakeHttp()
+    client = StateCenterClient("http://state-center:8000", "dev-agent-token", opener=http)
+
+    result = execute_node(
+        client,
+        "JOB-1",
+        "ORDER_FETCH",
+        lambda: {"summary": {"orderCount": 1}, "businessRefs": {}},
+        attempt_no=1,
+    )
+
+    assert result["next_node"] == "PRODUCT_MATCH"
+    assert [req["url"].rsplit("/", 1)[-1] for req in http.requests] == ["start", "complete"]
+    assert http.requests[1]["body"]["summary"] == {"orderCount": 1}
+    assert http.requests[1]["body"]["attemptNo"] == 1
+
+
+def test_execute_node_maps_login_failure_to_fail_callback():
+    http = FakeHttp([{"ok": True}, {"ok": True, "job_status": "FAILED_RETRYABLE"}])
+    client = StateCenterClient("http://state-center:8000", "dev-agent-token", opener=http)
+
+    def failing():
+        raise LoginFailed("登录失败")
+
+    result = execute_node(client, "JOB-1", "ORDER_FETCH", failing)
+
+    assert result["job_status"] == "FAILED_RETRYABLE"
+    assert http.requests[1]["url"].endswith("/fail")
+    assert http.requests[1]["body"]["errorCode"] == "SYS_LOGIN_FAIL"
+    assert http.requests[1]["body"]["retryable"] is True
+    assert http.requests[1]["body"]["routeRole"] == "OPS"
+
+
+def test_node_error_carries_default_ui_code():
+    exc = NodeError("元素找不到")
+    assert exc.error_code == "UI_ELEMENT_NOT_FOUND"
+    assert exc.retryable is False
+    assert exc.route_role == "OPS"

run_mock_chain.py

@@ -1,252 +1,11 @@
 #!/usr/bin/env python3
+"""Thin wrapper around the portable mock agent.
+
+Prefer:
+  python -m rpa_yisaiqi --create --auto-human
 """
-全链路Mock验证脚本
-不依赖真实医药城/合同系统，用mock数据跑通：建单→ORDER_FETCH→对码→CONTRACT_CREATE
-验证状态机推进、合同编号回填、前端能观测到next_node
-
-用法：
-  export STATE_CENTER_URL="http://127.0.0.1:8000"
-  export AGENT_TOKEN="dev-agent-token"
-  python run_mock_chain.py
-"""
-import json
-import sys
-import uuid
-from datetime import datetime, timedelta
-from urllib.request import Request, urlopen
-from urllib.error import HTTPError
-from urllib.parse import quote
-
-
-def encode_job_id(job_id: str) -> str:
-    """job_id含中文，URL路径段需编码"""
-    return quote(job_id, safe="")
-
-STATE_CENTER_URL = "http://127.0.0.1:8000"
-AGENT_TOKEN = "dev-agent-token"
-
-
-def http_post(path: str, payload: dict, token: str | None = None):
-    """POST请求辅助函数"""
-    url = f"{STATE_CENTER_URL}{path}"
-    headers = {"Content-Type": "application/json"}
-    if token:
-        headers["X-Agent-Token"] = token
-
-    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
-    request = Request(url, data=data, headers=headers, method="POST")
-
-    try:
-        with urlopen(request, timeout=15) as response:
-            raw = response.read().decode("utf-8")
-            return json.loads(raw) if raw else {}
-    except HTTPError as exc:
-        detail = exc.read().decode("utf-8", errors="replace")
-        print(f"❌ HTTP {exc.code}: {detail}")
-        raise
-
-
-def http_get(path: str):
-    """GET请求辅助函数"""
-    url = f"{STATE_CENTER_URL}{path}"
-    request = Request(url, method="GET")
-
-    try:
-        with urlopen(request, timeout=15) as response:
-            raw = response.read().decode("utf-8")
-            return json.loads(raw) if raw else {}
-    except HTTPError as exc:
-        detail = exc.read().decode("utf-8", errors="replace")
-        print(f"❌ HTTP {exc.code}: {detail}")
-        raise
-
-
-def step(num: int, title: str):
-    print(f"\n{'='*60}")
-    print(f"【第{num}步】{title}")
-    print('='*60)
-
-
-def run_chain():
-    order_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
-    # supplier加唯一后缀，保证每次全新job（避免幂等复用旧状态）
-    supplier = f"重庆亿昊药业-{uuid.uuid4().hex[:4]}"
-    source_type = "YYC"  # 医药城线上单
-
-    # ---- 第1步：建单 ----
-    step(1, "建单 POST /api/jobs")
-    resp = http_post("/api/jobs", {
-        "source_type": source_type,
-        "order_date": order_date,
-        "supplier_name": supplier,
-        "created_by": "mock-chain",
-    })
-    job_id = resp["job_id"]
-    print(f"✓ job_id: {job_id}")
-    print(f"✓ first_node: {resp.get('first_node')}")
-    print(f"✓ reused: {resp.get('reused')}")
-
-    # ---- 第2步：加载fixture货品（不依赖Oracle）----
-    step(2, "加载测试货品 POST /api/dev/match/load-fixture-goods")
-    resp = http_post("/api/dev/match/load-fixture-goods", {}, token=AGENT_TOKEN)
-    print(f"✓ 已加载货品数: {resp.get('loaded')}")
-
-    # ---- 第3步：ORDER_FETCH 开始回调 ----
-    step(3, "ORDER_FETCH start 回调")
-    resp = http_post(
-        f"/api/jobs/{encode_job_id(job_id)}/nodes/ORDER_FETCH/start",
-        {"attemptNo": 1, "executorPlatform": "mock-rpa", "operator": "mock-chain"},
-        token=AGENT_TOKEN,
-    )
-    print(f"✓ start响应: {resp}")
-
-    # ---- 第4步：写订单明细（模拟ORDER_FETCH导出的数据）----
-    step(4, "写订单明细 POST /api/jobs/{job_id}/items")
-    order_no = f"DD-{order_date.replace('-', '')}-001"
-    resp = http_post(
-        f"/api/jobs/{encode_job_id(job_id)}/items",
-        {
-            "replace": True,
-            "rows": [
-                {
-                    "orderNo": order_no,
-                    "rawProductName": "阿卡波糖片",
-                    "platformProductCode": "YP-001",
-                    "spec": "50mg*30片",
-                    "manufacturer": "示例制药",
-                    "approvalNo": "国药准字H00000001",
-                    "qty": 100,
-                    "unitPrice": 25.5,
-                    "sourceRowNo": 1,
-                },
-                {
-                    "orderNo": order_no,
-                    "rawProductName": "阿莫西林胶囊",
-                    "platformProductCode": "YP-002",
-                    "spec": "0.25g*24粒",
-                    "manufacturer": "示例医药",
-                    "approvalNo": "国药准字H00000002",
-                    "qty": 50,
-                    "unitPrice": 18.0,
-                    "sourceRowNo": 2,
-                },
-            ],
-        },
-        token=AGENT_TOKEN,
-    )
-    print(f"✓ 写入明细: {resp.get('inserted')} 行")
-    print(f"✓ 订单号: {resp.get('order_nos')}")
-
-    # ---- 第5步：ORDER_FETCH 完成回调 ----
-    step(5, "ORDER_FETCH complete 回调")
-    resp = http_post(
-        f"/api/jobs/{encode_job_id(job_id)}/nodes/ORDER_FETCH/complete",
-        {
-            "attemptNo": 1,
-            "executorRunId": f"MOCK-{uuid.uuid4().hex[:8]}",
-            "summary": {
-                "orderCount": 1,
-                "supplierList": [supplier],
-                "orderNoList": [order_no],
-            },
-            "artifacts": [
-                {
-                    "artifactType": "SOURCE_EXCEL",
-                    "storagePath": f"/tmp/mock_rpa_data/orders_{supplier}_{order_date}.xlsx",
-                    "fileHash": "sha256:mock-order-excel",
-                }
-            ],
-        },
-        token=AGENT_TOKEN,
-    )
-    print(f"✓ next_node: {resp.get('next_node')}")
-
-    # ---- 第6步：跑对码 ----
-    step(6, "对码 POST /api/jobs/{job_id}/match/run")
-    resp = http_post(f"/api/jobs/{encode_job_id(job_id)}/match/run", {}, token=AGENT_TOKEN)
-    print(f"✓ 对码结果: {json.dumps(resp, ensure_ascii=False)}")
-
-    # ---- 第7步：查当前状态，判断是否需要人工对码 ----
-    step(7, "查单据状态 GET /api/jobs/{job_id}")
-    detail = http_get(f"/api/jobs/{encode_job_id(job_id)}")
-    job = detail.get("job", {})
-    print(f"✓ 当前节点: {job.get('current_node_code')}")
-    print(f"✓ 当前状态: {job.get('status')}")
-
-    tasks = detail.get("tasks", [])
-    if tasks:
-        print(f"⚠ 有 {len(tasks)} 条待办需要人工处理")
-        for t in tasks:
-            print(f"    - {t.get('title')} [{t.get('reason_type')}]")
-
-    # ---- 第7.5步：如卡在SOURCE_ADD，模拟人工完成新增货源推进 ----
-    if job.get("current_node_code") == "SOURCE_ADD":
-        step("7.5", "SOURCE_ADD complete（模拟人工新增货源完成）")
-        resp = http_post(
-            f"/api/jobs/{encode_job_id(job_id)}/nodes/SOURCE_ADD/complete",
-            {
-                "attemptNo": 1,
-                "executorRunId": f"MOCK-SRC-{uuid.uuid4().hex[:8]}",
-                "operator": "mock-chain",
-                "summary": {"sourceAdded": True},
-                "artifacts": [],
-            },
-            token=AGENT_TOKEN,
-        )
-        print(f"✓ next_node: {resp.get('next_node')}")
-
-    # ---- 第8步：CONTRACT_CREATE mock（直接传假合同编号）----
-    step(8, "CONTRACT_CREATE mock（传假合同编号，跳过真实页面）")
-    mock_contract_no = f"HT{order_date.replace('-', '')}{uuid.uuid4().hex[:5].upper()}"
-    print(f"  使用mock合同编号: {mock_contract_no}")
-
-    # 先start
-    http_post(
-        f"/api/jobs/{encode_job_id(job_id)}/nodes/CONTRACT_CREATE/start",
-        {"attemptNo": 1, "executorPlatform": "mock-rpa", "operator": "mock-chain"},
-        token=AGENT_TOKEN,
-    )
-    # 再complete，带businessRefs.contractSystemNo
-    resp = http_post(
-        f"/api/jobs/{encode_job_id(job_id)}/nodes/CONTRACT_CREATE/complete",
-        {
-            "attemptNo": 1,
-            "executorRunId": f"MOCK-CONTRACT-{uuid.uuid4().hex[:8]}",
-            "businessRefs": {"contractSystemNo": mock_contract_no},
-            "summary": {
-                "supplierName": supplier,
-                "contractDate": order_date,
-                "sourceType": source_type,
-                "uploadRule": "merge_pdf_before_upload",
-            },
-            "artifacts": [],
-        },
-        token=AGENT_TOKEN,
-    )
-    print(f"✓ next_node: {resp.get('next_node')}")
-
-    # ---- 第9步：验证合同编号回填 ----
-    step(9, "验证合同编号回填 GET /api/jobs/{job_id}")
-    detail = http_get(f"/api/jobs/{encode_job_id(job_id)}")
-    job = detail.get("job", {})
-    print(f"✓ 当前节点: {job.get('current_node_code')}")
-    print(f"✓ 当前状态: {job.get('status')}")
-    contract_no = job.get("contract_system_no")
-    if contract_no == mock_contract_no:
-        print(f"✅ 合同编号回填成功: {contract_no}")
-    else:
-        print(f"⚠ 合同编号: {contract_no}（期望 {mock_contract_no}）")
-
-    print(f"\n{'='*60}")
-    print(f"🎉 Mock链跑完！job_id = {job_id}")
-    print(f"   前端观测: http://localhost:5173")
-    print('='*60)
-    return job_id
+from rpa_yisaiqi.__main__ import main
 
 
 if __name__ == "__main__":
-    try:
-        run_chain()
-    except Exception as e:
-        print(f"\n❌ 链路中断: {e}")
-        sys.exit(1)
+    main()

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"}
@@ -1174,7 +1239,32 @@ def _submit_candidate_selection(cur, task: dict, job: dict, payload: dict, opera
 def task_submit(task_id: str, req: TaskSubmitReq,
                 x_agent_token: Optional[str] = Header(default=None)):
     _check_agent(x_agent_token)
+    if not req.submitToken or len(req.submitToken) > 64:
+        raise HTTPException(status_code=400, detail="submitToken 最长 64")
+
     conn = get_conn()
+    began = False
+    try:
+        begin = getattr(conn, "begin", None)
+        if callable(begin):
+            begin()
+            began = True
+        result = _task_submit_body(conn, task_id, req)
+        if began:
+            conn.commit()
+        return result
+    except Exception:
+        rollback = getattr(conn, "rollback", None)
+        if callable(rollback):
+            rollback()
+        raise
+    finally:
+        close = getattr(conn, "close", None)
+        if callable(close):
+            close()
+
+
+def _task_submit_body(conn, task_id: str, req: TaskSubmitReq):
     with conn.cursor() as cur:
         cur.execute("SELECT * FROM user_task WHERE task_id=%s", (task_id,))
         task = cur.fetchone()

state_center/tests/test_web_api.py

@@ -794,3 +794,433 @@ 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"]
+
+
+def test_task_submit_rejects_oversized_submit_token(monkeypatch):
+    _install_fake_pymysql()
+    main = importlib.import_module("state_center.app.main")
+    client = TestClient(main.app)
+    advanced = {"moved": False}
+
+    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 = [{
+                    "task_id": "TASK-1",
+                    "job_id": "JOB-1",
+                    "node_code": "SOURCE_ADD",
+                    "status": "OPEN",
+                    "submit_token": None,
+                }]
+            elif "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",
+                }]
+            elif sql.startswith("UPDATE job SET"):
+                advanced["moved"] = True
+            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()
+
+        def begin(self):
+            return None
+
+        def commit(self):
+            return None
+
+        def rollback(self):
+            return None
+
+        def close(self):
+            return None
+
+    monkeypatch.setattr(main, "get_conn", lambda: FakeConn())
+    response = client.post(
+        "/api/tasks/TASK-1/submit",
+        headers={"X-Agent-Token": main.settings.AGENT_TOKEN},
+        json={
+            "action": "CONFIRM_AND_CONTINUE",
+            "payload": {},
+            "submitToken": "x" * 65,
+            "operator": "alice",
+        },
+    )
+
+    assert response.status_code == 400
+    assert "submitToken" in response.json()["detail"]
+    assert advanced["moved"] is False