* 在现有 ApprovalService（内存层）之上，增加 DB 持久化。 * 所有写操作先走 ApprovalService，再写 DB。 * 启动时从 DB 恢复 PENDING 状态到内存。 */ @Slf4j @Service @Order(55) // Schema 由 Flyway 管理，在 Flyway 迁移完成后执行 @RequiredArgsConstructor public class ApprovalWorkflowService implements ApplicationRunner { private final ApprovalService approvalService; private final ToolApprovalMapper approvalMapper; private final ObjectMapper objectMapper; private final ConversationService conversationService; /** * GC scheduler — owns the 5-minute clock for the entire approval state machine * (RFC-067 §4.4). Lives on the workflow rather than {@link ApprovalService} so * timeout / overflow eviction goes through the same DB+metadata+memory two-phase * path as approve / deny — the in-memory map can no longer drift ahead of DB. */ private ScheduledExecutorService gcScheduler; @Override public void run(ApplicationArguments args) { recoverFromDb(); } @PostConstruct void initGc() { gcScheduler = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "approval-gc"); t.setDaemon(true); return t; }); gcScheduler.scheduleAtFixedRate(this::garbageCollect, 5, 5, TimeUnit.MINUTES); log.info("[ApprovalWorkflow] GC scheduler started (interval=5min)"); } @PreDestroy void shutdownGc() { if (gcScheduler != null) { gcScheduler.shutdownNow(); } } /** * Reconstruct in-memory pending approvals from DB at startup, preserving the * original {@code pendingId} and {@code createdAt} so subsequent resolve / GC * paths stay consistent with the persisted row. *
* Effective expiration follows {@code expireAt != null ? expireAt : createdAt + PENDING_TTL},
* so legacy / test rows whose {@code expireAt} column is NULL still time out. Expired
* rows are reconciled (DB → TIMEOUT, message metadata → DENIED) and skipped from
* the in-memory map. See RFC-067 §4.1.
*/
void recoverFromDb() {
try {
List
* Order matters: metadata writes are gated on DB success. If {@code updateById}
* throws or affects zero rows, we skip the metadata flip so the three persistence
* loci (DB / message metadata / in-memory map) cannot drift apart — DB stuck on
* PENDING + metadata flipped to DENIED is the worst-case ghost state because the
* next recoverFromDb would re-revive the approval while the UI insists it was
* already settled.
*/
private void expireRecoveredRow(ToolApprovalEntity entity) {
int rowsUpdated;
try {
entity.setStatus("TIMEOUT");
entity.setResolvedAt(LocalDateTime.now());
rowsUpdated = approvalMapper.updateById(entity);
} catch (Exception e) {
log.warn("[ApprovalWorkflow] Failed to mark expired row {} as TIMEOUT: {}",
entity.getPendingId(), e.getMessage());
return;
}
if (rowsUpdated == 0) {
log.warn("[ApprovalWorkflow] Expire skipped: DB row for pending {} affected 0 rows " +
"(concurrent resolve?); leaving metadata untouched", entity.getPendingId());
return;
}
try {
conversationService.markPendingApprovalsResolved(
entity.getConversationId(),
Set.of(entity.getPendingId()),
MetadataDecision.DENIED);
} catch (Exception e) {
log.warn("[ApprovalWorkflow] Failed to reconcile metadata for expired pending {}: {}",
entity.getPendingId(), e.getMessage());
}
}
/**
* 创建待审批记录（增强版，含 GuardEvaluation）
*/
public String createPending(String conversationId, String userId,
String toolName, String toolArguments, String reason,
String toolCallPayload, String siblingToolCalls, String agentId,
GuardEvaluation evaluation) {
// 1. 内存层
String pendingId = approvalService.createPending(
conversationId, userId, toolName, toolArguments, reason,
toolCallPayload, siblingToolCalls, agentId);
// RFC-063r §2.12: capture the originating ChatOrigin from the holder.
// The holder was set by AgentService.{chat,chatStream,...} for the
// duration of the agent invocation that produced this approval — so
// it is non-null for IM / web triggered tool calls. Snapshot is
// serialized once here and persisted on the DB row so cross-restart
// replays keep the channel binding.
String chatOriginJson = serializeChatOrigin(ChatOriginHolder.get());
// 2. 增强内存记录
approvalService.getPending(pendingId).ifPresent(pending -> {
if (evaluation != null) {
pending.setFindingsJson(serializeFindings(evaluation.findings()));
pending.setMaxSeverity(evaluation.maxSeverity() != null ? evaluation.maxSeverity().name() : null);
pending.setSummary(evaluation.summary());
}
pending.setChatOrigin(chatOriginJson);
});
// 3. DB 层
persistToDb(pendingId, conversationId, userId, toolName, toolArguments,
toolCallPayload, siblingToolCalls, agentId, evaluation, chatOriginJson);
return pendingId;
}
/**
* 创建待审批记录（基础版，向后兼容）
*/
public String createPending(String conversationId, String userId,
String toolName, String toolArguments, String reason,
String toolCallPayload, String siblingToolCalls, String agentId) {
return createPending(conversationId, userId, toolName, toolArguments, reason,
toolCallPayload, siblingToolCalls, agentId, null);
}
/**
* Resolve a pending approval (approve / deny) following the RFC-067 §4.2 two-phase
* contract: snapshot → DB UPDATE conditional on {@code status='PENDING'} →
* metadata reconciliation → memory mutation queued for after-commit.
*
* Idempotent under concurrent resolve: when the DB UPDATE affects 0 rows (because
* another caller — IM channel, GC, recoverFromDb — already moved the row off
* PENDING), this method returns {@link ResolveOutcome#alreadyResolved(String)}
* without touching metadata or in-memory state. Callers should treat this as a
* silent no-op; do not surface a user-facing error.
*
* On DB / metadata write failure the transaction rolls back and in-memory state
* stays untouched, so a retry from the next GC tick can recover. Memory mutation
* is registered as an {@code afterCommit} synchronization, never inline, so a
* post-update commit failure cannot leave memory ahead of DB.
*
* @param pendingId target approval id
* @param userId actor performing the resolution (for audit)
* @param decision case-insensitive {@code "approved"} or {@code "denied"}
* @return {@link ResolveOutcome} carrying the resolved snapshot + DB / metadata
* counters; idempotent return on no-op
*/
@Transactional
public ResolveOutcome resolve(String pendingId, String userId, String decision) {
boolean approved = "approved".equalsIgnoreCase(decision);
String dbStatus = approved ? "APPROVED" : "DENIED";
MetadataDecision metaDecision = approved ? MetadataDecision.APPROVED : MetadataDecision.DENIED;
String snapshotStatus = approved ? "approved" : "denied";
return performResolve(pendingId, userId, dbStatus, metaDecision, snapshotStatus,
/* removeFromMap */ false);
}
/**
* Atomically resolve {@code approved} and consume the snapshot for replay.
* Same two-phase contract as {@link #resolve}, additionally removing the
* pending entry from the in-memory map after commit so a subsequent
* {@link #findPendingByConversation(String)} returns null and consume is
* single-shot. The returned {@link ResolveOutcome#consumedSnapshot()} carries
* {@code toolCallPayload} for replay.
*/
@Transactional
public ResolveOutcome resolveAndConsume(String pendingId, String userId) {
return performResolve(pendingId, userId, "CONSUMED", MetadataDecision.APPROVED,
"consumed", /* removeFromMap */ true);
}
/**
* Consume the earliest already-{@code approved} record for the conversation +
* tool — used when an out-of-band approval (e.g. /approve text command flow that
* resolved the record) needs to be redeemed for replay.
*/
@Transactional
public ResolveOutcome consumeApproved(String conversationId, String toolName) {
PendingApproval target = approvalService.findApprovedForConsume(conversationId, toolName);
if (target == null) {
return ResolveOutcome.alreadyResolved(null);
}
return performResolveOnSnapshot(target, null, "CONSUMED", MetadataDecision.APPROVED,
"consumed", /* removeFromMap */ true);
}
/**
* Bulk-deny every pending approval in the conversation (RFC-067 §4.4.1). Used by
* the Web Stop endpoint to clear orphaned approvals when the user halts a turn
* mid-stream; without this sweep, in-flight pendings linger in the map and
* resurrect via metadata after refresh / restart.
*
* Two-phase per row: DB → {@code DENIED}, message metadata → {@code DENIED},
* map removed. Per-row failures are logged and the sweep continues; the returned
* list contains only the outcomes that successfully advanced through DB.
*
* @return outcome per pending that successfully transitioned to {@code DENIED}
*/
@Transactional
public List
*
* Each pending entry's transition runs in its own transaction (markTimeout is
* @Transactional) so a single bad row doesn't block the rest of the sweep.
*/
public void garbageCollect() {
Instant now = Instant.now();
// Phase A — TTL-expired pending. Snapshot first so we don't mutate a map
// we're iterating; markTimeout handles its own DB+metadata+memory contract.
int timedOut = 0;
for (PendingApproval expired : approvalService.snapshotExpiredPending(now)) {
try {
ResolveOutcome outcome = markTimeout(expired.getPendingId());
if (outcome.dbSynced()) timedOut++;
} catch (Exception e) {
log.warn("[ApprovalWorkflow] GC: markTimeout failed for {}: {}",
expired.getPendingId(), e.getMessage());
}
}
// Phase B — pending overflow eviction. Same path; just driven by a count cap.
int evictedPending = 0;
for (PendingApproval excess : approvalService.snapshotExcessPending(ApprovalService.MAX_PENDING)) {
try {
ResolveOutcome outcome = markTimeout(excess.getPendingId());
if (outcome.dbSynced()) evictedPending++;
} catch (Exception e) {
log.warn("[ApprovalWorkflow] GC: overflow markTimeout failed for {}: {}",
excess.getPendingId(), e.getMessage());
}
}
// Phase C — resolved cleanup. Memory-only; DB rows for these entries are
// already terminal (CONSUMED / DENIED / TIMEOUT / SUPERSEDED) so nothing
// would change in DB or metadata.
int droppedResolved = approvalService.dropResolvedExceedingLimits(now);
if (timedOut > 0 || evictedPending > 0 || droppedResolved > 0) {
log.info("[ApprovalWorkflow] GC: timed-out {}, evicted-pending {}, dropped-resolved {}, remaining={}",
timedOut, evictedPending, droppedResolved, approvalService.size());
}
}
// ---------- shared two-phase machinery ----------
private ResolveOutcome performResolve(String pendingId, String userId,
String dbStatus, MetadataDecision metaDecision,
String snapshotStatus, boolean removeFromMap) {
PendingApproval snapshot = approvalService.getPending(pendingId).orElse(null);
if (snapshot == null || !"pending".equals(snapshot.getStatus())) {
log.debug("[ApprovalWorkflow] resolve {}: not pending (snapshot={}, status={})",
pendingId, snapshot != null, snapshot != null ? snapshot.getStatus() : "n/a");
return ResolveOutcome.alreadyResolved(pendingId);
}
return performResolveOnSnapshot(snapshot, userId, dbStatus, metaDecision,
snapshotStatus, removeFromMap);
}
private ResolveOutcome performResolveOnSnapshot(PendingApproval snapshot, String userId,
String dbStatus, MetadataDecision metaDecision,
String snapshotStatus, boolean removeFromMap) {
// Phase 1 — DB UPDATE (conditional). The eq("PENDING") guard makes the call
// idempotent: if another path already won, we get rows=0 and bail without
// touching metadata or memory.
int rows;
try {
LambdaUpdateWrapper