OpController.java
14.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
package com.xly.web;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xly.agent.AgentIdentity;
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**）—— 人在环写入闭环的执行侧。
*
* <p>用户在对话框点【确认】-> {@code /confirm} 才真正调 ERP 执行；点【取消】-> {@code /cancel}。
* 覆盖全部 op 类型：create（含报价这类多表主-从）、update、delete、invalid/cancelInvalid、
* examine/cancelExamine。执行完回写 {@code ai_op_queue} 状态 + 审计。
*
* <p>两条执行路径由 {@code erp.exec-staging.enabled} 切换：直连 ERP 通用写接口（本地默认），
* 或委托 ERP 侧暂存执行器 {@code /ai/execStaging}（生产路径，事务化、以用户身份执行）。
*
* <p>当前**恒同步**：确认即执行并同步返回终态。架构 §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 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,
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.authz = authz;
this.conversations = conversations;
this.resolver = resolver;
}
/** 确认/取消的结果落事件日志（rightPush 原子，流式输出中途点确认也不覆盖对话事件）——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);
} catch (Exception e) {
log.warn("record op outcome failed (conv={}, op={}): {}", conv, opId, e.getMessage());
}
}
/**
* 会话里最近一条待确认操作（前端每轮结束后轮询，用于渲染确认卡片）。
* 需登录且只能查自己的会话；只回渲染必需的字段——绝不外泄 sPayload/内部表名等。
*/
@GetMapping("/pending")
public Map<String, Object> 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<String, Object> op = ops.pending(conversationId);
if (op == null) {
return Map.of();
}
Map<String, Object> out = new LinkedHashMap<>();
out.put("opId", op.get("sId"));
out.put("summary", op.get("sDescription"));
out.put("status", op.get("sStatus"));
return out;
}
/**
* 确认执行：调 ERP 执行暂存的写操作，回写状态。
* <p>{@code Authorization} 头（可空）= 用户浏览器里的 ERP 登录 token，透传给 ERP 使执行以用户身份
* 进行；为空则回退 dev-login。绝不因用户 token 缺失而静默提权（见 ErpClient.canRelogin）。
*/
@PostMapping("/{id}/confirm")
public Map<String, Object> confirm(@PathVariable("id") String id,
@RequestHeader(value = "Authorization", required = false) String authToken) {
AgentIdentity me = requireLogin(authToken);
Map<String, Object> 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<String, Object> 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<Map<String, Object>> tables = mapper.convertValue(payload.get("__tables__"), List.class);
if (!tables.isEmpty()) {
@SuppressWarnings("unchecked")
Map<String, Object> masterCol = (Map<String, Object>) 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<String, Object> 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<String, Object> confirmViaExecutor(String id, Map<String, Object> 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<String, Object> cancel(@PathVariable("id") String id,
@RequestHeader(value = "Authorization", required = false) String authToken) {
AgentIdentity me = requireLogin(authToken);
Map<String, Object> 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<String, Object> 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<String, Object> op) {
String owner = str(op.get("sUserId"));
if (owner == null || !owner.equals(me.userId())) {
throw new ResponseStatusException(HttpStatus.FORBIDDEN, "无权处理该操作");
}
}
private Map<String, Object> result(String status, String msg, Map<String, Object> op) {
Map<String, Object> 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();
}
}