diff --git a/rpa_yisaiqi/README.md b/rpa_yisaiqi/README.md index ba091d4..f858cc4 100644 --- a/rpa_yisaiqi/README.md +++ b/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`。 diff --git a/rpa_yisaiqi/__main__.py b/rpa_yisaiqi/__main__.py new file mode 100644 index 0000000..53d8433 --- /dev/null +++ b/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() diff --git a/rpa_yisaiqi/agent.py b/rpa_yisaiqi/agent.py new file mode 100644 index 0000000..10f9d83 --- /dev/null +++ b/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 diff --git a/rpa_yisaiqi/callback_client.py b/rpa_yisaiqi/callback_client.py index decff95..bf6ada0 100644 --- a/rpa_yisaiqi/callback_client.py +++ b/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: diff --git a/rpa_yisaiqi/contract_create_node.py b/rpa_yisaiqi/contract_create_node.py index 9a1e2b9..663d7bc 100644 --- a/rpa_yisaiqi/contract_create_node.py +++ b/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: diff --git a/rpa_yisaiqi/runners/__init__.py b/rpa_yisaiqi/runners/__init__.py new file mode 100644 index 0000000..1213286 --- /dev/null +++ b/rpa_yisaiqi/runners/__init__.py @@ -0,0 +1 @@ +"""Node runners. Mock now, Windows page runners later.""" diff --git a/rpa_yisaiqi/runners/mock.py b/rpa_yisaiqi/runners/mock.py new file mode 100644 index 0000000..0c525aa --- /dev/null +++ b/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", + }], + } diff --git a/rpa_yisaiqi/runtime/__init__.py b/rpa_yisaiqi/runtime/__init__.py new file mode 100644 index 0000000..8de8ef4 --- /dev/null +++ b/rpa_yisaiqi/runtime/__init__.py @@ -0,0 +1 @@ +"""Portable node runtime shared by mock and Windows runners.""" diff --git a/rpa_yisaiqi/runtime/errors.py b/rpa_yisaiqi/runtime/errors.py new file mode 100644 index 0000000..72f951d --- /dev/null +++ b/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" diff --git a/rpa_yisaiqi/runtime/payloads.py b/rpa_yisaiqi/runtime/payloads.py new file mode 100644 index 0000000..d75ca22 --- /dev/null +++ b/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": [], + } diff --git a/rpa_yisaiqi/runtime/shell.py b/rpa_yisaiqi/runtime/shell.py new file mode 100644 index 0000000..5f58601 --- /dev/null +++ b/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)) diff --git a/rpa_yisaiqi/tests/test_agent.py b/rpa_yisaiqi/tests/test_agent.py new file mode 100644 index 0000000..5506aa2 --- /dev/null +++ b/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") diff --git a/rpa_yisaiqi/tests/test_callback_client.py b/rpa_yisaiqi/tests/test_callback_client.py new file mode 100644 index 0000000..8d6d07a --- /dev/null +++ b/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 diff --git a/rpa_yisaiqi/tests/test_runtime_shell.py b/rpa_yisaiqi/tests/test_runtime_shell.py new file mode 100644 index 0000000..2411128 --- /dev/null +++ b/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" diff --git a/run_mock_chain.py b/run_mock_chain.py index 9c5e470..0d84b1c 100644 --- a/run_mock_chain.py +++ b/run_mock_chain.py @@ -1,257 +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，走待办提交，不打 RPA complete ---- - if job.get("current_node_code") == "SOURCE_ADD": - step("7.5", "SOURCE_ADD 待办提交 CONFIRM_AND_CONTINUE") - source_task = next( - (t for t in tasks if t.get("node_code") == "SOURCE_ADD" and t.get("status") == "OPEN"), - None, - ) - if not source_task: - raise RuntimeError("SOURCE_ADD 没有 OPEN 待办，无法推进") - resp = http_post( - f"/api/tasks/{source_task['task_id']}/submit", - { - "action": "CONFIRM_AND_CONTINUE", - "payload": {"sourceAdded": True}, - "submitToken": f"mock-source-{uuid.uuid4().hex}", - "operator": "mock-chain", - }, - 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() diff --git a/state_center/app/main.py b/state_center/app/main.py index f45f338..cb11901 100644 --- a/state_center/app/main.py +++ b/state_center/app/main.py @@ -1239,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() diff --git a/state_center/tests/test_web_api.py b/state_center/tests/test_web_api.py index 36e7ad2..478b890 100644 --- a/state_center/tests/test_web_api.py +++ b/state_center/tests/test_web_api.py @@ -1146,3 +1146,81 @@ def test_human_node_rejects_rpa_complete(monkeypatch): 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