package com.xly.web; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.xly.agent.AgentIdentity; import com.xly.config.RedisChatMemoryStore; import com.xly.service.AuditService; import com.xly.service.AuthzService; import com.xly.service.ConversationService; import com.xly.service.ErpClient; import com.xly.service.FormResolverService; import com.xly.service.LedgerService; import com.xly.service.OpService; import com.xly.service.StateService; import org.springframework.http.HttpStatus; import org.springframework.web.server.ResponseStatusException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; /** * 写操作确认端点(**确定性、不经过 LLM**)—— 人在环写入闭环的执行侧。 * *

用户在对话框点【确认】-> {@code /confirm} 才真正调 ERP 执行;点【取消】-> {@code /cancel}。 * 覆盖全部 op 类型:create(含报价这类多表主-从)、update、delete、invalid/cancelInvalid、 * examine/cancelExamine。执行完回写 {@code ai_op_queue} 状态 + 审计。 * *

两条执行路径由 {@code erp.exec-staging.enabled} 切换:直连 ERP 通用写接口(本地默认), * 或委托 ERP 侧暂存执行器 {@code /ai/execStaging}(生产路径,事务化、以用户身份执行)。 * *

当前**恒同步**:确认即执行并同步返回终态。架构 §10 设计的「auto 流程异步 + 卡片三态 + SSE 终态推送」 * 尚未实现(见 docs/agent-architecture.md §18)。 */ @RestController @RequestMapping("/api/agent/op") public class OpController { private static final Logger log = LoggerFactory.getLogger(OpController.class); private final OpService ops; private final ErpClient erp; private final AuditService audit; private final ObjectMapper mapper; private final LedgerService ledger; private final StateService state; private final RedisChatMemoryStore memoryStore; private final AuthzService authz; private final ConversationService conversations; private final FormResolverService resolver; /** true=确认后委托 ERP 侧暂存执行器(/ai/execStaging,§10 生产路径);false=xlyAi 直连 ERP 通用写接口执行。 */ @org.springframework.beans.factory.annotation.Value("${erp.exec-staging.enabled:false}") private boolean execStagingEnabled; public OpController(OpService ops, ErpClient erp, AuditService audit, ObjectMapper mapper, LedgerService ledger, StateService state, RedisChatMemoryStore memoryStore, AuthzService authz, ConversationService conversations, FormResolverService resolver) { this.ops = ops; this.erp = erp; this.audit = audit; this.mapper = mapper; this.ledger = ledger; this.state = state; this.memoryStore = memoryStore; this.authz = authz; this.conversations = conversations; this.resolver = resolver; } /** 确认/取消的结果落账本+状态槽+对话记忆(记忆空洞修补:LLM 下轮就知道这单已执行/失败/取消)。 */ private void recordOutcome(String conv, String opId, String status, String msg, String description) { if (conv == null || conv.isBlank() || "null".equals(conv)) { return; } try { ledger.append(conv, "cancelled".equals(status) ? "cancel" : "confirm", Map.of( "opId", opId == null ? "" : opId, "status", status, "msg", msg == null ? "" : msg, "description", description == null ? "" : description)); state.updateDocStage(conv, opId, status); String note; if ("executed".equals(status)) { note = "(系统)用户已确认,操作执行成功:" + description; } else if ("cancelled".equals(status)) { note = "(系统)用户取消了该操作:" + description; } else { note = "(系统)用户确认后执行失败:" + description + (msg == null || msg.isBlank() ? "" : (",原因:" + msg)); } memoryStore.appendTurn(conv, null, note); } catch (Exception e) { log.warn("record op outcome failed (conv={}, op={}): {}", conv, opId, e.getMessage()); } } /** * 会话里最近一条待确认操作(前端每轮结束后轮询,用于渲染确认卡片)。 * 需登录且只能查自己的会话;只回渲染必需的字段——绝不外泄 sPayload/内部表名等。 */ @GetMapping("/pending") public Map pending(@RequestParam("conversationId") String conversationId, @RequestHeader(value = "Authorization", required = false) String authToken) { String uid = requireLogin(authToken).userId(); if (!conversations.owns(uid, conversationId)) { throw new ResponseStatusException(HttpStatus.FORBIDDEN, "无权访问该会话"); } Map op = ops.pending(conversationId); if (op == null) { return Map.of(); } Map out = new LinkedHashMap<>(); out.put("opId", op.get("sId")); out.put("summary", op.get("sDescription")); out.put("status", op.get("sStatus")); return out; } /** * 确认执行:调 ERP 执行暂存的写操作,回写状态。 *

{@code Authorization} 头(可空)= 用户浏览器里的 ERP 登录 token,透传给 ERP 使执行以用户身份 * 进行;为空则回退 dev-login。绝不因用户 token 缺失而静默提权(见 ErpClient.canRelogin)。 */ @PostMapping("/{id}/confirm") public Map confirm(@PathVariable("id") String id, @RequestHeader(value = "Authorization", required = false) String authToken) { AgentIdentity me = requireLogin(authToken); Map op = ops.get(id); if (op == null) { return result("failed", "找不到该操作", null); } requireOwner(me, op); if (!"draft".equals(String.valueOf(op.get("sStatus")))) { return result(String.valueOf(op.get("sStatus")), "该操作已处理过(" + op.get("sStatus") + ")", op); } // CAS 抢占:并发/重复确认只有一个能真正执行,杜绝同一张单被执行两次 if (!ops.claim(id)) { Map cur = ops.get(id); String st = cur == null ? "unknown" : String.valueOf(cur.get("sStatus")); return result(st, "该操作正在处理或已处理过(" + st + ")", op); } String uid = str(op.get("sUserId")); String conv = str(op.get("sConversationId")); String target = str(op.get("sTargetTable")) + "#" + str(op.get("sTargetBillId")); String detail = str(op.get("sDescription")); // 生产路径:委托 ERP 侧暂存执行器(它以用户身份执行、事务化并自行回写 ai_op_queue 状态)。 if (execStagingEnabled) { return confirmViaExecutor(id, op, authToken, uid, conv, target, detail); } try { String opType = str(op.get("sOpType")); JsonNode r; if ("create".equals(opType)) { JsonNode payload = mapper.readTree(str(op.get("sPayload"))); if (payload.has("__tables__")) { // 报价等多表主-从创建 @SuppressWarnings("unchecked") List> tables = mapper.convertValue(payload.get("__tables__"), List.class); if (!tables.isEmpty()) { @SuppressWarnings("unchecked") Map masterCol = (Map) tables.get(0).get("column"); refreshBillNo(masterCol, str(op.get("sTargetTable")), me.brandsId()); } r = erp.createMulti(authToken, str(op.get("sTargetModuleId")), tables); } else { @SuppressWarnings("unchecked") Map columns = mapper.convertValue(payload, Map.class); refreshBillNo(columns, str(op.get("sTargetTable")), me.brandsId()); r = erp.createForm(authToken, str(op.get("sTargetModuleId")), str(op.get("sTargetTable")), columns); } } else if ("delete".equals(opType)) { r = erp.deleteForm(authToken, str(op.get("sTargetModuleId")), str(op.get("sTargetTable")), str(op.get("sTargetBillId"))); } else if ("invalid".equals(opType)) { boolean cancel = "cancel".equals(str(op.get("sNewValue"))); // cancel=复原/取消作废,toVoid=作废 r = erp.invalidForm(authToken, str(op.get("sTargetModuleId")), str(op.get("sTargetTable")), str(op.get("sTargetBillId")), cancel); } else if ("examine".equals(opType)) { int iFlag = "0".equals(str(op.get("sNewValue"))) ? 0 : 1; // 1=审核 0=反审核 r = erp.examineForm(authToken, str(op.get("sTargetModuleId")), str(op.get("sTargetBillId")), iFlag); } else { r = erp.updateForm(authToken, str(op.get("sTargetModuleId")), str(op.get("sTargetTable")), str(op.get("sTargetBillId")), str(op.get("sField")), str(op.get("sNewValue"))); } int code = r.path("code").asInt(0); if (code == 1) { ops.setStatus(id, "executed", "操作成功"); audit.log(uid, conv, "confirm", target, detail, true, "executed"); recordOutcome(conv, id, "executed", null, detail); return result("executed", "已执行:" + op.get("sDescription"), op); } String msg = r.path("msg").asText("执行失败"); ops.setStatus(id, "failed", msg); audit.log(uid, conv, "confirm", target, detail, false, msg); recordOutcome(conv, id, "failed", msg, detail); return result("failed", msg, op); } catch (Exception e) { log.warn("confirm op {} failed", id, e); ops.setStatus(id, "failed", e.getMessage()); audit.log(uid, conv, "confirm", target, detail, false, e.getMessage()); recordOutcome(conv, id, "failed", e.getMessage(), detail); return result("failed", "执行异常:" + e.getMessage(), op); } } /** 委托 ERP 侧暂存执行器执行(§10);执行器自行回写 ai_op_queue 状态,这里只映射结果 + 审计。 */ private Map confirmViaExecutor(String id, Map op, String authToken, String uid, String conv, String target, String detail) { try { JsonNode r = erp.execStaging(authToken, id); String st = r.path("status").asText("failed"); String msg = r.path("msg").asText(""); boolean ok = "executed".equals(st); audit.log(uid, conv, "confirm", target, detail, ok, "execStaging:" + st + (msg.isEmpty() ? "" : (" " + msg))); recordOutcome(conv, id, ok ? "executed" : "failed", msg, detail); return result(ok ? "executed" : "failed", ok ? ("已执行:" + op.get("sDescription")) : (msg.isEmpty() ? "执行失败" : msg), op); } catch (Exception e) { log.warn("execStaging confirm op {} failed", id, e); audit.log(uid, conv, "confirm", target, detail, false, e.getMessage()); recordOutcome(conv, id, "failed", e.getMessage(), detail); return result("failed", "执行异常:" + e.getMessage(), op); } } /** 取消(需登录且必须是本人的操作)。 */ @PostMapping("/{id}/cancel") public Map cancel(@PathVariable("id") String id, @RequestHeader(value = "Authorization", required = false) String authToken) { AgentIdentity me = requireLogin(authToken); Map op = ops.get(id); if (op != null) { requireOwner(me, op); } if (op != null && "draft".equals(String.valueOf(op.get("sStatus")))) { ops.setStatus(id, "cancelled", "用户取消"); audit.log(str(op.get("sUserId")), str(op.get("sConversationId")), "cancel", str(op.get("sTargetTable")) + "#" + str(op.get("sTargetBillId")), str(op.get("sDescription")), true, "cancelled"); recordOutcome(str(op.get("sConversationId")), id, "cancelled", null, str(op.get("sDescription"))); } return result("cancelled", "已取消", op); } /** * 单号在 propose 时算的是当时的 MAX+1;期间别人建过单就会撞号,所以**执行前重新生成**。 * 拿不到新号时保留旧号(由 ERP 侧唯一约束兜底),不因此中断执行。 */ private void refreshBillNo(Map columns, String table, String brandsId) { if (columns == null || !columns.containsKey("sBillNo")) { return; } String fresh = resolver.nextBillNo(table, brandsId); if (fresh != null && !fresh.isBlank()) { columns.put("sBillNo", fresh); } } /** 需登录(token 服务端内省);未登录/过期 → 401。 */ private AgentIdentity requireLogin(String authToken) { AgentIdentity id = authz.resolveIdentity(authToken); if (id == null) { throw new ResponseStatusException(HttpStatus.UNAUTHORIZED, "登录已过期或未登录"); } return id; } /** 只有提议的发起人本人能确认/取消——否则任何人拿到 opId 就能替别人写库。 */ private void requireOwner(AgentIdentity me, Map op) { String owner = str(op.get("sUserId")); if (owner == null || !owner.equals(me.userId())) { throw new ResponseStatusException(HttpStatus.FORBIDDEN, "无权处理该操作"); } } private Map result(String status, String msg, Map op) { Map m = new LinkedHashMap<>(); m.put("status", status); m.put("msg", msg); if (op != null) { m.put("opId", op.get("sId")); m.put("description", op.get("sDescription")); } return m; } private static String str(Object o) { return o == null ? null : o.toString(); } }