rpa_yisaiqi/README.md

@@ -1,19 +1,6 @@
-# 艺赛旗联调包
+# 艺赛旗 CONTRACT_CREATE 联调骨架
 
-这个目录是 Windows 和 macOS 共用的 RPA 包。状态机在 `state_center`，这里只负责回调外壳和节点 runner。
+这个目录放艺赛旗侧可复制到 Windows 虚拟机的代码。第一版目标是先验证：
-
-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

@@ -1,59 +0,0 @@
-"""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

@@ -1,125 +0,0 @@
-"""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,88 +65,16 @@ 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:
-        return self._request("POST", path, payload)
+        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
-
-    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=headers,
+            headers={
-            method=method,
+                "Content-Type": "application/json",
+                "X-Agent-Token": self.token,
+            },
+            method="POST",
         )
         try:
             with self._opener(request, self.timeout) as response:

rpa_yisaiqi/contract_create_node.py

@@ -17,31 +17,34 @@ 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(NodeError):
+class ContractCreateError(RuntimeError):
-    pass
+    error_code = "UI_ELEMENT_NOT_FOUND"
+    retryable = False
+    route_role = "OPS"
 
 
-class LoginFailed(RuntimeLoginFailed, ContractCreateError):
+class LoginFailed(ContractCreateError):
-    pass
+    error_code = "SYS_LOGIN_FAIL"
+    retryable = True
+    route_role = "OPS"
 
 
-class SupplierContractDisabled(RuntimeSupplierContractDisabled, ContractCreateError):
+class SupplierContractDisabled(ContractCreateError):
-    pass
+    error_code = "BIZ_CONTRACT_DISABLED"
+    retryable = False
+    route_role = "PURCHASER"
 
 
-class PdfMergeUnsupported(RuntimePdfMergeUnsupported, ContractCreateError):
+class PdfMergeUnsupported(ContractCreateError):
-    pass
+    error_code = "CAP_PDF_MERGE_UNSUPPORTED"
+    retryable = False
+    route_role = "OPS"
 
 
 @dataclass(frozen=True)
@@ -69,25 +72,15 @@ 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)
-        payload = build_complete_payload(spec, contract_no)
+        complete_payload = build_complete_payload(spec, contract_no)
-        return {
+        return client.complete_node(spec.job_id, NODE_CODE, complete_payload)
-            "businessRefs": payload["businessRefs"],
+    except ContractCreateError as exc:
-            "summary": payload["summary"],
+        fail_payload = build_fail_payload(spec, exc)
-            "artifacts": payload["artifacts"],
+        return client.fail_node(spec.job_id, NODE_CODE, fail_payload)
-        }
-
-    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

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

rpa_yisaiqi/runners/mock.py

@@ -1,60 +0,0 @@
-"""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

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

rpa_yisaiqi/runtime/errors.py

@@ -1,25 +0,0 @@
-"""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

@@ -1,37 +0,0 @@
-"""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

@@ -1,29 +0,0 @@
-"""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

@@ -1,240 +0,0 @@
-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

@@ -1,78 +0,0 @@
-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

@@ -1,75 +0,0 @@
-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,11 +1,252 @@
 #!/usr/bin/env python3
-"""Thin wrapper around the portable mock agent.
-
-Prefer:
-  python -m rpa_yisaiqi --create --auto-human
 """
-from rpa_yisaiqi.__main__ import main
+全链路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
 
 
 if __name__ == "__main__":
-    main()
+    try:
+        run_chain()
+    except Exception as e:
+        print(f"\n❌ 链路中断: {e}")
+        sys.exit(1)

state_center/app/main.py

@@ -99,13 +99,11 @@ 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": "合同审批已超过等待上限，需要运维确认已通过或终止本单。",
 }
 
 # 条件节点:满足条件才执行,否则自动跳过下一节点。
@@ -159,21 +157,6 @@ 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)
@@ -286,31 +269,11 @@ 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)"
-        (job_id, node_code, NODE_EXECUTOR.get(node_code, "HUMAN"), next_attempt, now, now, now),
+        " 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()),
     )
 
 
@@ -672,9 +635,6 @@ 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"}
@@ -716,6 +676,16 @@ 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)
 
         # 决定下一步(显式转移表 + 条件节点跳过)
@@ -767,40 +737,6 @@ 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")
@@ -874,13 +810,12 @@ 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"}
@@ -1239,32 +1174,7 @@ 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,433 +794,3 @@ 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