package vip.mate.channel.webchat;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.GeneralSecurityException;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Base64;
import vip.mate.channel.web.Utf8SseEmitter;
import vip.mate.agent.AgentService;
import vip.mate.channel.model.ChannelEntity;
import vip.mate.channel.service.ChannelService;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.common.result.R;
import vip.mate.memory.event.ConversationCompletionPublisher;
import vip.mate.workspace.conversation.ConversationService;
import vip.mate.workspace.conversation.model.ConversationEntity;
import vip.mate.workspace.conversation.model.MessageContentPart;
import vip.mate.workspace.conversation.model.MessageEntity;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import reactor.core.Disposable;
import vip.mate.approval.PendingApproval;
import vip.mate.approval.ResolveOutcome;
import vip.mate.agent.context.ChatOrigin;
/**
* WebChat 嵌入式对话接口
*
* 独立于 ChatController，使用 API Key 认证（不依赖 JWT）。
* 供外部网站通过 JS SDK 嵌入 MateClaw 对话能力。
*
* 认证方式：请求头 X-MC-Key 携带 API Key
*
* @author MateClaw Team
*/
@Tag(name = "WebChat 嵌入式对话")
@Slf4j
@RestController
@RequestMapping("/api/v1/channels/webchat")
@RequiredArgsConstructor
public class WebChatController {
private final ChannelService channelService;
private final AgentService agentService;
private final ConversationService conversationService;
private final ChatStreamTracker streamTracker;
private final ObjectMapper objectMapper;
private final ConversationCompletionPublisher completionPublisher;
private final vip.mate.memory.identity.MemoryOwnerResolver memoryOwnerResolver;
private final WebChatFileService fileService;
private final WebChatTokenRevocationService tokenRevocationService;
private final vip.mate.audit.service.AuditEventService auditService;
private final vip.mate.llm.routing.AgentBindingResolver agentBindingResolver;
private final vip.mate.skill.repository.SkillMapper skillMapper;
private final vip.mate.wiki.repository.WikiPageMapper wikiPageMapper;
private final vip.mate.wiki.repository.WikiKnowledgeBaseMapper wikiKbMapper;
/**
* ISSUE #413 P1-A2/A3/A4: drives the approval lifecycle for WebChat
* (API-Key) channels. Before this, a tool guarded by ToolGuard would
* create a pending approval and park the turn, but the visitor had no
* way to resolve it -- the approval hung until the 30-min GC timeout
* and the turn was wasted.
*/
private final vip.mate.approval.ApprovalWorkflowService approvalService;
/** Visitor-token TTL in seconds (7 days). Mirrors GeneratedFileCache's TTL. */
static final long VISITOR_TOKEN_TTL_SECONDS = 7 * 24 * 3600L;
/**
* Server-only secret used to sign per-visitor tokens. Reuses the JWT secret so no extra
* config/migration is needed; it is never sent to the client (unlike the public channel API key).
*/
@Value("${mateclaw.jwt.secret:MateClaw-JWT-Secret-Key-2024-Please-Change-In-Production}")
private String visitorTokenSecret;
/**
* SseEmitter timeout (minutes) for WebChat SSE streams. Previously
* hardcoded to 10 minutes across three {@code new Utf8SseEmitter(...)} call
* sites; downstream integrators were forced to reason about a constant
* living in someone else's repo (issue #586). Configurable so operators
* have a documented knob, defaulting to the historical 10 minutes.
*/
@Value("${mateclaw.webchat.sse-timeout-minutes:10}")
private int webchatSseTimeoutMinutes;
private long sseTimeoutMillis() {
return (long) webchatSseTimeoutMinutes * 60_000L;
}
private final ExecutorService sseExecutor = Executors.newCachedThreadPool();
/**
* WebChat SSE 流式对话
*/
@Operation(summary = "WebChat SSE 流式对话")
@PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter chatStream(
@RequestHeader("X-MC-Key") String apiKey,
@RequestBody WebChatRequest request) {
// RFC-058 PR-1: Utf8SseEmitter 显式 charset=UTF-8，防止中文 SSE 乱码
SseEmitter emitter = new Utf8SseEmitter(sseTimeoutMillis());
// 验证 API Key 并获取关联的 Channel 配置
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
sendErrorAndComplete(emitter, "Invalid API Key");
return emitter;
}
// Resolve the target agent: an explicit request agentId overrides the channel's
// bound agent, but must belong to the channel's workspace (anti privilege-escalation:
// a shared channel Key must not be able to drive arbitrary agents in other workspaces).
Long agentId = channel.getAgentId();
if (request.getAgentId() != null) {
var requested = agentService.getAgent(request.getAgentId());
if (requested == null) {
sendErrorAndComplete(emitter, "Requested agent not found");
return emitter;
}
if (channel.getWorkspaceId() != null && requested.getWorkspaceId() != null
&& !channel.getWorkspaceId().equals(requested.getWorkspaceId())) {
sendErrorAndComplete(emitter, "Requested agent does not belong to this channel's workspace");
return emitter;
}
agentId = request.getAgentId();
}
if (agentId == null) {
sendErrorAndComplete(emitter, "No agent configured for this WebChat channel");
return emitter;
}
final Long resolvedAgentId = agentId;
// Optional sessionId lets one visitor hold multiple isolated threads. It is only ever
// composed into the server-derived conversationId (kept under the key+visitor namespace),
// never accepted as a raw conversationId — so a caller can't reach another tenant's history.
final String visitorId;
final String effectiveSessionId;
try {
visitorId = normalizeVisitorId(request.getVisitorId());
effectiveSessionId = normalizeSessionId(request.getSessionId());
} catch (IllegalArgumentException ex) {
sendErrorAndComplete(emitter, ex.getMessage());
return emitter;
}
String conversationId = deriveConversationId(apiKey, visitorId, effectiveSessionId);
// Server-issued, unforgeable proof that this caller owns this visitorId. Returned in the
// meta event below; the session-management endpoints require it back (see verifyVisitorToken).
final String visitorToken = computeVisitorToken(visitorTokenSecret, channel.getId(), visitorId);
String message = request.getMessage() != null ? request.getMessage() : "";
if (message.isBlank()) {
sendErrorAndComplete(emitter, "Message is required");
return emitter;
}
log.info("[WebChat] Stream: agentId={}, conversationId={}, visitor={}", agentId, conversationId, visitorId);
AtomicReference runHandleRef = new AtomicReference<>();
AtomicBoolean disconnected = new AtomicBoolean();
// Register emitter callbacks. An SSE disconnect means this subscriber
// left, not that the agent run finished. Use detach() instead of
// complete(); complete() would prematurely mark the RunState done,
// drop later content deltas from the replay buffer, and double-count
// completion when the agent Flux actually finishes.
emitter.onCompletion(() -> {
log.debug("[WebChat] SSE completed: {}", conversationId);
disconnected.set(true);
streamTracker.detach(runHandleRef.get(), emitter);
});
emitter.onTimeout(() -> {
// INFO: a timeout means the stream went idle past the SseEmitter
// budget, which is a key signal when diagnosing stream stalls.
log.info("[WebChat] SSE timeout (stream went idle past the emitter budget): {}", conversationId);
disconnected.set(true);
streamTracker.detach(runHandleRef.get(), emitter);
// Explicitly complete after timeout so the servlet container does
// not rethrow AsyncRequestTimeoutException.
emitter.complete();
});
emitter.onError(e -> {
// INFO only for non-benign causes; a client simply closing the tab
// (broken pipe / connection reset) is routine and stays DEBUG so it
// doesn't flood production logs.
if (isClientDisconnect(e)) {
log.debug("[WebChat] SSE client disconnected: {} - {}", conversationId, e.getMessage());
} else {
log.info("[WebChat] SSE error: {} - {}", conversationId, e.getMessage());
}
disconnected.set(true);
streamTracker.detach(runHandleRef.get(), emitter);
});
sseExecutor.execute(() -> {
try {
// 创建或获取会话（workspace 从 agent 获取）
var webAgent = agentService.getAgent(resolvedAgentId);
Long webWsId = webAgent != null ? webAgent.getWorkspaceId() : 1L;
var conv = conversationService.getOrCreateWebchatConversation(
conversationId, resolvedAgentId, webchatUsername(visitorId), webWsId, effectiveSessionId);
// 保存用户消息（含访客本轮引用的附件）。附件元数据一律服务端按 fileId 回查，
// 不信客户端传入；path 用于 Agent 侧工具读取，对外消息视图会被剥离。
List userParts = buildUserParts(conversationId, message, request.getAttachmentIds());
Long originMessageId = request.getInternalOriginMessageId();
if (!request.isInternalSkipUserPersist()) {
// Regenerate reuses the already-persisted seed user row —
// inserting again would duplicate it.
var savedUser = conversationService
.saveMessage(conversationId, "user", message, userParts);
originMessageId = savedUser == null ? null : savedUser.getId();
}
// 初始化 SSE 流跟踪
ChatStreamTracker.RunHandle runHandle = streamTracker.register(conversationId);
runHandleRef.set(runHandle);
streamTracker.attach(runHandle, emitter);
if (disconnected.get()) {
streamTracker.detach(runHandle, emitter);
}
// Echo the effective session so the caller can persist it (especially when
// sessionId was omitted) and address the same thread on subsequent calls. The
// visitorToken must be stored by the caller and sent back on list/messages/delete.
streamTracker.broadcast(runHandle, "meta",
"{\"sessionId\":" + escapeJson(effectiveSessionId)
+ ",\"conversationId\":" + escapeJson(conversationId)
+ ",\"visitorToken\":" + escapeJson(visitorToken) + "}");
// Accumulate the assistant reply so it can be persisted on stream completion.
// Pattern mirrors ChatController: always accumulate, only broadcast when the
// delta is not a persistence-only echo of content already streamed by inner nodes.
StringBuilder assistantReply = new StringBuilder();
// Token usage + model attribution: capture _usage_final event emitted at stream end
final int[] usage = {0, 0, 0, 0, 0}; // [prompt, completion, cacheRead, cacheWrite, reasoning]
final String[] modelInfo = {null, null}; // [runtimeModel, runtimeProvider]
// Attribute memory to this external visitor so each end-user
// behind the shared webchat account is isolated. The same origin
// resolves the owner key for both the read (recall) and write
// (publish) paths below.
vip.mate.agent.context.ChatOrigin webchatOrigin =
vip.mate.agent.context.ChatOrigin.web(conversationId, visitorId, webWsId, null)
.withSender(null, "api", null)
.withOriginMessageId(originMessageId);
String webchatOwnerKey = memoryOwnerResolver.resolve(webchatOrigin);
reactor.core.Disposable disposable = agentService.chatStructuredStream(resolvedAgentId, message, conversationId, visitorId, null, webchatOrigin)
.doOnNext(delta -> {
if (delta.isEvent() && "_usage_final".equals(delta.eventType())) {
Map data = delta.eventData();
usage[0] = ((Number) data.getOrDefault("promptTokens", 0)).intValue();
usage[1] = ((Number) data.getOrDefault("completionTokens", 0)).intValue();
usage[2] = ((Number) data.getOrDefault("cacheReadTokens", 0)).intValue();
usage[3] = ((Number) data.getOrDefault("cacheWriteTokens", 0)).intValue();
usage[4] = ((Number) data.getOrDefault("reasoningTokens", 0)).intValue();
Object model = data.get("runtimeModelName");
Object provider = data.get("runtimeProviderId");
if (model != null) modelInfo[0] = model.toString();
if (provider != null) modelInfo[1] = provider.toString();
}
// Forward a curated subset of agent lifecycle events to the
// visitor SSE stream. The full event vocabulary (iteration_*,
// perf_summary, _routing_decision, feedback_event, ...) is
// internal — exposing it to 3rd-party websites would leak
// graph internals and complicate the SDK contract. The four
// types below are the ones that drive visible UX: typing
// indicator (phase), tool execution badges (tool_start/end),
// plan-execute checklist (plan). See docs/zh/webchat.md.
if (delta.isEvent()) {
forwardVisitorEvent(runHandle, conversationId,
delta.eventType(), delta.eventData());
}
if (delta.content() != null && !delta.content().isEmpty()) {
assistantReply.append(delta.content());
if (!delta.persistenceOnly()) {
streamTracker.broadcast(runHandle, "content_delta",
"{\"text\":" + escapeJson(delta.content()) + "}");
}
}
if (delta.thinking() != null && !delta.thinking().isEmpty()
&& !delta.persistenceOnly()) {
streamTracker.broadcast(runHandle, "thinking_delta",
"{\"text\":" + escapeJson(delta.thinking()) + "}");
}
})
.doOnComplete(() -> {
String reply = assistantReply.toString();
try {
if (!reply.isBlank()) {
conversationService.saveMessage(
conversationId, "assistant", reply, List.of(),
"completed", usage[0], usage[1], usage[2], usage[3], usage[4], modelInfo[0], modelInfo[1], null);
}
completionPublisher.publish(
resolvedAgentId, conversationId, message, reply, "webchat", webchatOwnerKey);
} catch (Exception persistErr) {
log.warn("[WebChat] Failed to persist assistant reply / publish event: {}",
persistErr.getMessage());
}
streamTracker.broadcast(runHandle, "done", "{\"status\":\"completed\"}");
// WebChat is a pure-backend SSE channel with no re-attach
// endpoint: for third-party integrators reading until the
// server closes, `done` IS the end of the stream. Close the
// subscriber connections so a 5-second answer doesn't hold
// a downstream connection-pool slot for the full SseEmitter
// timeout (issue #586). The in-house web channel does NOT
// do this — it keeps emitters open for reconnect + replay.
streamTracker.closeSubscribers(runHandle);
streamTracker.complete(runHandle);
})
.doOnError(e -> {
log.error("[WebChat] Stream error: {}", e.getMessage());
streamTracker.broadcast(runHandle, "error",
"{\"message\":" + escapeJson(e.getMessage()) + "}");
streamTracker.closeSubscribers(runHandle);
streamTracker.complete(runHandle);
})
.subscribe();
// Bind the subscription's Disposable so requestStop() (invoked by
// POST /sessions/stop) can actually dispose the Flux and interrupt
// the LLM stream. Without this, stopRequested is set but the underlying
// HTTP call keeps running — token burn + side-effect tools still fire.
// Mirrors ChatController#chatStream line 495.
streamTracker.setDisposable(runHandle, disposable);
// Wire the emergency save so an orphaned run (only subscriber
// gone) is flushed as an "interrupted" assistant message when
// the grace-period eviction reclaims it — otherwise the visitor
// would see only their own user message (issue #587).
registerEmergencySave(runHandle, conversationId, assistantReply, usage, modelInfo);
} catch (Exception e) {
log.error("[WebChat] Error: {}", e.getMessage(), e);
try {
emitter.send(SseEmitter.event().name("error")
.data(Map.of("message", e.getMessage())));
emitter.complete();
} catch (IOException ex) {
emitter.completeWithError(ex);
}
}
});
return emitter;
}
/**
* 获取 WebChat 配置（前端 SDK 初始化用）
*/
@Operation(summary = "获取 WebChat 配置")
@GetMapping("/config")
public R> getConfig(@RequestHeader("X-MC-Key") String apiKey) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
JsonNode config = parseConfig(channel.getConfigJson());
return R.ok(Map.of(
"channelName", channel.getName(),
"agentId", channel.getAgentId() != null ? channel.getAgentId() : 0,
"title", textOrDefault(config, "title", channel.getName()),
"placeholder", textOrDefault(config, "placeholder", "Type a message..."),
"primaryColor", textOrDefault(config, "primary_color", "#409eff"),
"welcomeMessage", textOrDefault(config, "welcome_message", "")
));
}
/**
* 列出访客在当前 channel 上可见的技能清单(供下游集成方实现 "/" slash
* picker UI)。返回的是展示级元数据——id、slug、本地化名、描述、
* 图标,不暴露 SKILL.md 正文、config、安全扫描结果等内部字段。
*
* 鉴权链跟 {@link #listSessions} 一致:API Key 解析 channel + visitorToken
* HMAC 校验。{@code agentId} 可选,缺省回落到 channel 绑定的 agent;
* 必须属于该 channel 的 workspace(沿用 {@code /stream} 的反越权路径)。
*
* 可见范围 = 显式绑定到该 agent 的 enabled 技能。无显式绑定的 agent
* (意为"用全局默认")返回空清单——visitor 看不到候选,但仍可走自然语言
* 让 LLM 自行调 {@code load_skill}。
*/
@Operation(summary = "列出访客可见技能(供 slash picker UI)")
@GetMapping("/skills")
public R> listSkills(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam(required = false) Long agentId,
@RequestParam String visitorId) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
// Resolve the target agent: same anti-escalation rule as /stream — an
// explicit agentId must belong to the channel's workspace.
Long resolvedAgentId = channel.getAgentId();
if (agentId != null) {
var requested = agentService.getAgent(agentId);
if (requested == null) {
return R.fail(404, "Requested agent not found");
}
if (channel.getWorkspaceId() != null && requested.getWorkspaceId() != null
&& !channel.getWorkspaceId().equals(requested.getWorkspaceId())) {
return R.fail(403, "Requested agent does not belong to this channel's workspace");
}
resolvedAgentId = agentId;
}
if (resolvedAgentId == null) {
return R.ok(List.of());
}
// Bound-skill IDs is null when the agent has no explicit binding (meaning
// "use global defaults"); treat that as "no candidates surfaced to the
// picker" so the agent config stays the source of truth for visitor UI.
java.util.Set boundIds = agentBindingResolver.getBoundSkillIds(resolvedAgentId);
if (boundIds == null || boundIds.isEmpty()) {
return R.ok(List.of());
}
List skills = skillMapper.selectBatchIds(boundIds);
return R.ok(skills.stream()
.filter(s -> Boolean.TRUE.equals(s.getEnabled()))
// Stable order: by slug asc, fall back to id for ties (e.g. null slug).
.sorted(java.util.Comparator.comparing(
s -> s.getName() != null ? s.getName() : "",
java.util.Comparator.nullsFirst(String::compareToIgnoreCase)))
.map(WebChatSkillView::from)
.toList());
}
/**
* 列出访客可见的 wiki 页面，供下游自建「`[[slug]]` 引用 picker」UI。
*
* 鉴权链跟 {@link #listSkills} 一致：API Key 解析 channel + visitorToken
* HMAC 校验。{@code agentId} 可选，缺省回落到 channel 绑定的 agent；
* 必须属于该 channel 的 workspace（沿用 {@code /stream} 的反越权路径）。
*
* 可见范围 = agent 绑定的 KB（无显式绑定时回落到 workspace 内全部 KB）
* 下的所有 page，排除 {@code pageType=synthesis}（LLM 中间产物）。
* 上限 {@value WEBCHAT_WIKI_PICKER_MAX_PAGES}：超出时强制要求 {@code keyword}。
*
* 出参仅暴露展示级元数据（slug / title / summary / pageType / kbId /
* kbName），不包含正文、embedding、sourceRawIds 等内部字段。
*/
@Operation(summary = "列出访客可见 wiki 页面（供 [[slug]] picker UI）")
@GetMapping("/wiki/pages")
public R> listWikiPages(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam(required = false) Long agentId,
@RequestParam String visitorId,
@RequestParam(required = false) String keyword) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
// Resolve the target agent — same anti-escalation rule as /stream and
// /skills: an explicit agentId must belong to the channel's workspace.
Long resolvedAgentId = channel.getAgentId();
if (agentId != null) {
var requested = agentService.getAgent(agentId);
if (requested == null) {
return R.fail(404, "Requested agent not found");
}
if (channel.getWorkspaceId() != null && requested.getWorkspaceId() != null
&& !channel.getWorkspaceId().equals(requested.getWorkspaceId())) {
return R.fail(403, "Requested agent does not belong to this channel's workspace");
}
resolvedAgentId = agentId;
}
if (resolvedAgentId == null) {
return R.ok(List.of());
}
// Resolve the KB scope: null = workspace-wide (every KB in the agent's
// workspace); non-empty set = explicit allowlist. Set.of() (rows exist
// but none enabled) means "agent is explicitly scoped to zero KBs" —
// surface as empty so the picker shows nothing rather than falling
// through to workspace-wide.
java.util.Set boundKbIds = agentBindingResolver.getBoundKbIds(resolvedAgentId);
Long workspaceId = channel.getWorkspaceId();
java.util.Set effectiveKbIds;
if (boundKbIds != null) {
if (boundKbIds.isEmpty()) {
return R.ok(List.of());
}
effectiveKbIds = boundKbIds;
} else {
// No explicit binding → fall back to every KB in the channel's
// workspace. Matches the wiki-tool behavior (an unscoped agent
// sees workspace-wide KBs).
effectiveKbIds = wikiKbMapper.selectList(
new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
.eq(vip.mate.wiki.model.WikiKnowledgeBaseEntity::getWorkspaceId,
workspaceId == null ? 1L : workspaceId))
.stream()
.map(vip.mate.wiki.model.WikiKnowledgeBaseEntity::getId)
.filter(java.util.Objects::nonNull)
.collect(java.util.stream.Collectors.toSet());
if (effectiveKbIds.isEmpty()) {
return R.ok(List.of());
}
}
// Build the page query: KB scope + exclude hidden pageTypes + optional
// keyword filter on slug/title. Use a single LIKE with OR so a visitor
// typing "auth" matches either "auth-design" (slug) or "Auth Design" (title).
String trimmedKeyword = keyword == null ? null : keyword.trim();
boolean hasKeyword = trimmedKeyword != null && !trimmedKeyword.isEmpty();
// Cap check: if no keyword and total candidate count exceeds the cap,
// refuse — the caller must narrow with a keyword. Counting before
// selecting avoids materializing a huge list into memory.
// NOTE: the count wrapper is built WITHOUT ORDER BY — H2 in MySQL mode
// rejects "ORDER BY slug" on a COUNT(*) query (column must appear in
// GROUP BY). The select wrapper below adds ORDER BY slug.
if (!hasKeyword) {
long total = wikiPageMapper.selectCount(buildWikiPageFilterWrapper(effectiveKbIds, null));
if (total > WEBCHAT_WIKI_PICKER_MAX_PAGES) {
return R.fail(422, "Wiki page count (" + total
+ ") exceeds picker cap (" + WEBCHAT_WIKI_PICKER_MAX_PAGES
+ "); please provide a 'keyword' query parameter to narrow.");
}
}
List pages = wikiPageMapper.selectList(
buildWikiPageFilterWrapper(effectiveKbIds, hasKeyword ? trimmedKeyword : null)
.orderByAsc(vip.mate.wiki.model.WikiPageEntity::getSlug));
if (pages.isEmpty()) {
return R.ok(List.of());
}
// Hydrate KB names so the picker UI can show ": " and
// the LLM can disambiguate when two KBs share a slug.
java.util.Map kbNames = wikiKbMapper.selectBatchIds(effectiveKbIds).stream()
.collect(java.util.stream.Collectors.toMap(
vip.mate.wiki.model.WikiKnowledgeBaseEntity::getId,
kb -> kb.getName() != null ? kb.getName() : "",
(a, b) -> a));
return R.ok(pages.stream()
.map(p -> WebChatWikiPageView.from(p, kbNames.get(p.getKbId())))
.toList());
}
/**
* Build the WHERE-clause portion of the wiki-page picker query: KB scope +
* pageType-not-in-hidden + optional keyword LIKE on slug / title.
* Returned without ORDER BY so the caller can layer sorting (select) or
* nothing (count) on top — H2 in MySQL mode rejects ORDER BY on COUNT(*).
*/
private com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper
buildWikiPageFilterWrapper(java.util.Set kbIds, String keyword) {
com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper w =
new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
.in(vip.mate.wiki.model.WikiPageEntity::getKbId, kbIds)
.notIn(vip.mate.wiki.model.WikiPageEntity::getPageType, WEBCHAT_WIKI_HIDDEN_PAGE_TYPES);
if (keyword != null && !keyword.isEmpty()) {
String like = "%" + keyword + "%";
w.and(qq -> qq.like(vip.mate.wiki.model.WikiPageEntity::getSlug, like)
.or().like(vip.mate.wiki.model.WikiPageEntity::getTitle, like));
}
return w;
}
/** Cap on how many empty (message_count = 0) threads one visitor may hold on a
* channel at once. Guards against pathologic clients churning placeholder
* sessions without ever sending a message. */
private static final int MAX_EMPTY_SESSIONS_PER_VISITOR = 5;
/**
* Upper bound on the page count the wiki-page picker will return without a
* keyword filter. Beyond this the caller MUST supply {@code keyword} —
* returning 500 pages to a visitor picker is both bandwidth-wasteful and
* unusable as a UI. Mirrors the slash-skill picker's "small list, search
* when too big" stance.
*/
private static final int WEBCHAT_WIKI_PICKER_MAX_PAGES = 100;
/**
* Page types hidden from the visitor picker. {@code synthesis} pages are
* LLM-generated intermediate artifacts (compiled on demand by
* {@code wiki_compile_page}); they aren't curated source material and
* surfacing them to a downstream visitor is noise. Entity / concept /
* source pages are human-readable references the visitor can meaningfully
* point the LLM at.
*/
private static final java.util.Set WEBCHAT_WIKI_HIDDEN_PAGE_TYPES = java.util.Set.of("synthesis");
/**
* 显式创建一条访客会话线程（空会话）。
*
* 与 {@code POST /stream} 的隐式 getOrCreate 互补：本端点先建一条 message_count=0
* 的占位线程，调用方拿到 {@code sessionId/conversationId/visitorToken} 之后，再决定
* 何时通过 {@code /stream} 发首条消息。鉴权为访客的首次接触：仅校验
* {@code X-MC-Key}，不要求 {@code X-MC-Visitor-Token}，后端会签发并回传 token，
* 调用方在后续 GET/PUT/DELETE 上必须回带。
*
* 行为：
*
* - 幂等：{@code sessionId} 与该 visitor 已有线程冲突 → 直接返回现有线程，
* 不报错、不覆盖 title。
* - 配额：单 (渠道, visitor) 未活跃空线程 ≤ {@value MAX_EMPTY_SESSIONS_PER_VISITOR}，
* 超出返回 409。已存在的线程走幂等路径不受配额限制。
* - title 非空时写入；为空时落默认 "新对话"，首条 user 消息仍会按现有规则截取。
*
*/
@Operation(summary = "显式创建访客会话线程（空会话）")
@PostMapping("/sessions")
public R> createSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestBody(required = false) WebChatCreateSessionRequest request) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
// Resolve agent: explicit request.agentId overrides channel's bound agent,
// but must belong to channel's workspace (mirrors /stream).
final Long agentId;
if (request != null && request.getAgentId() != null) {
var requested = agentService.getAgent(request.getAgentId());
if (requested == null) {
return R.fail(400, "Requested agent not found");
}
if (channel.getWorkspaceId() != null && requested.getWorkspaceId() != null
&& !channel.getWorkspaceId().equals(requested.getWorkspaceId())) {
return R.fail(400, "Requested agent does not belong to this channel's workspace");
}
agentId = request.getAgentId();
} else {
agentId = channel.getAgentId();
if (agentId == null) {
return R.fail(400, "No agent configured for this WebChat channel");
}
}
final String visitorId;
final String sessionId;
try {
visitorId = normalizeVisitorId(request != null ? request.getVisitorId() : null);
sessionId = normalizeSessionId(request != null ? request.getSessionId() : null);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String title = (request != null && request.getTitle() != null) ? request.getTitle().trim() : null;
if (title != null && (title.isEmpty() || title.length() > 100)) {
return R.fail(400, "title 不合法（1-100 字）");
}
String conversationId = deriveConversationId(apiKey, visitorId, sessionId);
String owner = webchatUsername(visitorId);
// Idempotency: existing thread is returned as-is. Title and every other
// field are left untouched — a re-create call must not clobber a previously
// set title. Existing rows are exempt from the empty-session quota.
ConversationEntity existing = conversationService.findByConversationId(conversationId);
if (existing != null && owner.equals(existing.getUsername())) {
audit(channel, visitorId, "webchat.create-session", conversationId,
"{\"sessionId\":\"" + sessionId + "\",\"idempotent\":true}");
return R.ok(buildCreateSessionResponse(existing, sessionId, channel.getId(), visitorId));
}
// Quota: count empty threads this visitor already holds on this channel.
// loadVisitorSessions already scopes to (channel prefix ∩ visitor owner).
long emptyCount = loadVisitorSessions(apiKey, visitorId).stream()
.filter(s -> s.getMessageCount() == null || s.getMessageCount() == 0)
.count();
if (emptyCount >= MAX_EMPTY_SESSIONS_PER_VISITOR) {
return R.fail(409, "未活跃会话数已达上限（" + MAX_EMPTY_SESSIONS_PER_VISITOR
+ "），请先发送消息或删除旧会话");
}
ConversationEntity conv = conversationService.getOrCreateWebchatConversation(
conversationId, agentId, owner, channel.getWorkspaceId(), sessionId, title);
audit(channel, visitorId, "webchat.create-session", conversationId,
"{\"sessionId\":\"" + sessionId + "\",\"idempotent\":false}");
return R.ok(buildCreateSessionResponse(conv, sessionId, channel.getId(), visitorId));
}
private Map buildCreateSessionResponse(ConversationEntity conv, String sessionId,
Long channelId, String visitorId) {
String visitorToken = computeVisitorToken(visitorTokenSecret, channelId, visitorId);
// LinkedHashMap (not Map.of) because Map.of rejects null and we want a
// stable key order for the response payload.
Map m = new java.util.LinkedHashMap<>();
m.put("sessionId", sessionId != null ? sessionId : "");
m.put("conversationId", conv.getConversationId());
m.put("visitorToken", visitorToken);
m.put("title", conv.getTitle() != null ? conv.getTitle() : "");
m.put("createTime", conv.getCreateTime());
return m;
}
/**
* Audit a visitor-side write. Actor is {@code "webchat::"}
* so audit searches can filter by channel / visitor. detailJson should be
* a JSON object capturing whatever the operator would need to reconstruct
* the call (sessionId, before/after state, etc).
*/
private void audit(ChannelEntity channel, String visitorId, String action,
String conversationId, String detailJson) {
String actor = "webchat:" + channel.getId() + ":" + visitorId;
auditService.recordAs(actor, channel.getWorkspaceId(),
action, "CONVERSATION", conversationId, null, detailJson);
}
/**
* 列出某访客的会话线程
*
* 仅返回属于本 Key + visitorId 的会话（按 conversationId 前缀过滤），
* 不暴露裸 conversationId，调用方按 sessionId 寻址。
*/
@Operation(summary = "列出访客会话线程")
@GetMapping("/sessions")
public R> listSessions(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(defaultValue = "false") boolean includeArchived) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
return R.ok(loadVisitorSessions(apiKey, visitorId, includeArchived));
}
/**
* 分页 + 关键词搜索某访客的会话线程。
* 访客的会话集是按 visitor 命名空间限定的（数量有界），故在内存里做关键词过滤与分页。
* keyword 不区分大小写、匹配标题子串。
*
鉴权链跟 {@link #listSessions} 完全一致,本方法只做"列表 → 关键词过滤 → 分页"的视图
* 包装,所以直接委托 listSessions 后处理(避免重复 resolveChannel + verifyVisitorToken
* 的鉴权代码)。
*/
@Operation(summary = "分页查询访客会话线程")
@GetMapping("/sessions/page")
public R> pageSessions(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(defaultValue = "1") int page,
@RequestParam(defaultValue = "20") int size,
@RequestParam(required = false) String keyword,
@RequestParam(defaultValue = "false") boolean includeArchived) {
@SuppressWarnings("unchecked")
R> base = (R>) (R)
listSessions(apiKey, visitorToken, visitorId, includeArchived);
if (base.getCode() != 200) {
return R.fail(base.getCode(), base.getMsg());
}
if (page < 1) page = 1;
if (size < 1 || size > 200) size = 20;
List all = base.getData();
if (keyword != null && !keyword.isBlank()) {
String kw = keyword.trim().toLowerCase(java.util.Locale.ROOT);
all = all.stream()
.filter(s -> s.getTitle() != null && s.getTitle().toLowerCase(java.util.Locale.ROOT).contains(kw))
.collect(Collectors.toList());
}
long total = all.size();
int from = Math.min((page - 1) * size, all.size());
int to = Math.min(from + size, all.size());
List pageItems = all.subList(from, to);
return R.ok(Map.of(
"items", pageItems,
"total", total,
"page", page,
"size", size
));
}
/**
* 重命名某会话线程。标题非空、长度 ≤ 100。
*/
@Operation(summary = "重命名会话线程")
@PutMapping("/sessions/title")
public R renameSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestBody Map body) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
return R.fail(404, "Session not found");
}
String title = body != null && body.get("title") != null ? body.get("title").trim() : "";
if (title.isEmpty() || title.length() > 100) {
return R.fail(400, "标题不合法（1-100 字）");
}
conversationService.renameConversation(conversationId, title);
audit(channel, visitorId, "webchat.rename-session", conversationId,
"{\"sessionId\":\"" + sid + "\",\"title\":\"" + title + "\"}");
return R.ok();
}
/**
* 置顶 / 取消置顶某会话线程。Pinned 线程在访客的 /sessions 列表里排在最前
* (沿用 {@link ConversationService#listWebchatConversations} 的 pinned DESC 排序)。
*/
@Operation(summary = "置顶 / 取消置顶会话线程")
@PutMapping("/sessions/pinned")
public R pinSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestBody Map body) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
return R.fail(404, "Session not found");
}
Object v = body != null ? body.get("pinned") : null;
if (!(v instanceof Boolean)) {
return R.fail(400, "body must contain {pinned: true|false}");
}
conversationService.setPinned(conversationId, (Boolean) v);
audit(channel, visitorId, "webchat.pin-session", conversationId,
"{\"sessionId\":\"" + sid + "\",\"pinned\":" + v + "}");
return R.ok();
}
/**
* 归档 / 取消归档某会话线程。归档后线程仍在 DB(历史保留、按 sessionId 寻址、文件可下载),
* 但默认从 /sessions 列表隐藏;调用方需传 {@code includeArchived=true} 才能看到。
*/
@Operation(summary = "归档 / 取消归档会话线程")
@PutMapping("/sessions/archive")
public R archiveSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestBody Map body) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
return R.fail(404, "Session not found");
}
Object v = body != null ? body.get("archived") : null;
if (!(v instanceof Boolean)) {
return R.fail(400, "body must contain {archived: true|false}");
}
conversationService.setArchived(conversationId, (Boolean) v);
audit(channel, visitorId, "webchat.archive-session", conversationId,
"{\"sessionId\":\"" + sid + "\",\"archived\":" + v + "}");
return R.ok();
}
/**
* Load this visitor's session threads (own namespace only), mapped to the
* compact view. Sorted as {@code listConversations} returns them (pinned
* desc, last-active desc). Shared by the list and paginated endpoints.
*
* Enumeration is keyed by the visitor's username plus the channel prefix
* ({@code webchat::}) rather than the full conversationId prefix, so it
* still catches threads whose conversationId hashed (long visitorId +
* sessionId). The sessionId is read from the persisted {@code webchatSessionId}
* column (set on creation) and only falls back to parsing the conversationId
* for legacy rows created before that column existed.
*/
private List loadVisitorSessions(String apiKey, String visitorId) {
return loadVisitorSessions(apiKey, visitorId, false);
}
/**
* Overload that lets the caller opt into archived threads. By default
* (used by /sessions listing and the empty-session quota check) archived
* rows are filtered out — they still exist on disk and are addressable
* by sessionId, but don't pollute the active listing and don't count
* against the "≤ 5 empty threads" quota (the visitor already declared
* they're done with them).
*/
private List loadVisitorSessions(String apiKey, String visitorId,
boolean includeArchived) {
String base = deriveConversationId(apiKey, visitorId, null);
String channelPrefix = "webchat:" + apiKey.substring(0, Math.min(8, apiKey.length())) + ":";
String owner = webchatUsername(visitorId);
// Query is scoped to this visitor's own rows only (no system rows), so
// listing a visitor's threads doesn't load every IM/cron conversation.
// The channel prefix is matched in-memory with a literal startsWith so a
// '_' / '%' in the api key's first 8 chars can't act as a LIKE wildcard.
return conversationService.listWebchatConversations(owner).stream()
.filter(c -> c.getConversationId() != null
&& c.getConversationId().startsWith(channelPrefix))
.filter(c -> includeArchived
|| c.getArchived() == null
|| c.getArchived() == 0)
.map(c -> {
String sid = recoverSessionId(c, base);
return new WebChatSessionView(sid, c.getTitle(), c.getLastActiveTime(),
c.getMessageCount(),
c.getPinned() != null ? c.getPinned() : 0,
c.getArchived() != null ? c.getArchived() : 0,
c.getStreamStatus() != null ? c.getStreamStatus() : "idle");
})
.collect(Collectors.toList());
}
/**
* Recover a thread's sessionId. Prefers the persisted column; for legacy
* rows (column null) falls back to parsing the non-hashed conversationId.
* Returns null for the default (no-session) thread and for legacy hashed rows
* whose sessionId can no longer be reconstructed.
*/
private String recoverSessionId(vip.mate.workspace.conversation.model.ConversationEntity c, String base) {
if (c.getWebchatSessionId() != null) {
return c.getWebchatSessionId();
}
String cid = c.getConversationId();
if (cid.equals(base)) {
return null;
}
String prefix = base + ":";
if (cid.startsWith(prefix)) {
return cid.substring(prefix.length());
}
return null;
}
/**
* 获取某会话线程的消息列表（支持分页）。
* 不传 limit 时返回全部消息（向后兼容）；传 limit 返回最新 limit 条 + hasMore；
* 传 beforeId + limit 时返回该 ID 之前的 limit 条（上拉加载更早消息）。
*/
@Operation(summary = "获取会话消息（支持分页）")
@GetMapping("/sessions/messages")
public R sessionMessages(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestParam(required = false) Long beforeId,
@RequestParam(required = false) Integer limit) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
return R.fail(404, "Session not found");
}
// Backward-compatible: no limit → full external (path-stripped) list.
if (limit == null || limit <= 0) {
return R.ok(conversationService.listMessageViewsExternal(conversationId));
}
// Paginated: mirror ConversationController#listMessages but with the
// external view so visitors never see server-side file paths.
List messages;
boolean hasMore;
if (beforeId != null) {
messages = conversationService.listMessagesBefore(conversationId, beforeId, limit + 1);
hasMore = messages.size() > limit;
if (hasMore) {
messages = messages.subList(messages.size() - limit, messages.size());
}
} else {
long total = conversationService.countMessages(conversationId);
messages = conversationService.listRecentMessages(conversationId, limit);
hasMore = total > limit;
}
return R.ok(Map.of(
"messages", conversationService.toExternalMessageViews(messages),
"hasMore", hasMore
));
}
/**
* 删除某会话线程
*/
@Operation(summary = "删除会话线程")
@DeleteMapping("/sessions")
public R deleteSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
return R.fail(404, "Session not found");
}
conversationService.deleteConversation(conversationId);
audit(channel, visitorId, "webchat.delete-session", conversationId,
"{\"sessionId\":\"" + sid + "\"}");
return R.ok();
}
/**
* 停止访客某线程正在进行中的 SSE 流。
*
* 鉴权同其他会话管理端点(API Key + visitorToken + 会话归属)。内部调
* {@link ChatStreamTracker#requestStop(String)}——靠 chatStream 注册时绑定的
* Disposable 实际中断 Flux;返回 {@code stopped=false} 表示当前没有活跃流
* (幂等,不报错)。
*
* Approval sweep (ISSUE #413 P1-A4): deny any pending approvals on this
* conversation so they do not hang for 30 minutes until the GC timeout.
* The visitor username derived from visitorId is the actor -- it resolves
* the "no MateClaw username" blocker noted in the old javadoc.
* Each denied approval broadcasts a { tool_approval_resolved} SSE
* event so the SDK clears its banner immediately.
*/
@Operation(summary = "停止访客会话线程的进行中流")
@PostMapping("/sessions/stop")
public R> stopSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
// ownsConversation is the existence + ownership guard: an unknown sessionId
// maps to a conversationId that either doesn't exist or belongs to someone
// else — both return 404 so the caller can't probe the namespace.
if (!ownsConversation(conversationId, visitorId)) {
return R.fail(404, "Session not found");
}
boolean stopped = streamTracker.requestStop(conversationId);
log.info("[WebChat] Stop requested: conversationId={}, visitor={}, stopped={}",
conversationId, visitorId, stopped);
audit(channel, visitorId, "webchat.stop-session", conversationId,
"{\"sessionId\":\"" + sid + "\",\"stopped\":" + stopped + "}");
// ISSUE #413 P1-A4: deny pending approvals so they do not linger for
// 30 min waiting on a GC timeout. The visitor username is the actor,
// mirroring how the web ChatController uses the logged-in username.
int deniedCount = 0;
try {
java.util.List denied =
approvalService.denyAllByConversation(conversationId, webchatUsername(visitorId));
deniedCount = denied.size();
for (ResolveOutcome o : denied) {
try {
streamTracker.broadcast(conversationId, "tool_approval_resolved",
objectMapper.writeValueAsString(java.util.Map.of(
"pendingId", o.pendingId(),
"decision", "denied",
"toolName", o.toolName() != null ? o.toolName() : "")));
} catch (Exception broadcastErr) {
log.debug("[WebChat] approval_resolved broadcast failed for {}: {}",
o.pendingId(), broadcastErr.getMessage());
}
}
} catch (Exception sweepErr) {
log.warn("[WebChat] approval sweep failed for {}: {}", conversationId, sweepErr.getMessage());
}
if (deniedCount > 0) {
log.info("[WebChat] Denied {} pending approval(s) on stop for {}", deniedCount, conversationId);
}
return R.ok(Map.of("stopped", stopped));
}
/**
* 拒绝一个待审批的工具调用 (ISSUE #413 P1-A2)。
*
* 鉴权同其它会话管理端点 (API Key + visitorToken + 会话归属)。仅允许
* 发起对话的访客拒绝自己会话的审批 —— 身份校验通过
* {@code webchatUsername(visitorId)} 与 pending.userId 的等价比较
* (对位 IM 渠道的 senderId == requester 校验)。
*
* resolve 后立即广播 {@code tool_approval_resolved} SSE 事件，让 SDK
* 实时清理审批 banner。返回同步 JSON (非 SSE)，因为 deny 不需要重放工具。
*
* @param pendingId the approval pendingId returned in the
* {@code tool_approval_requested} event
*/
@Operation(summary = "拒绝访客会话中的待审批工具调用")
@PostMapping("/sessions/deny")
public R> denySession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestParam String pendingId) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
return R.fail(404, "Session not found");
}
// IDOR guard (review #415): the caller owns the conversation, but the
// pendingId is client-supplied — cross-check that the pending actually
// belongs to this conversation before resolving, otherwise a visitor
// could resolve / replay another visitor's guarded tool call.
// getPending(pendingId) gives the exact record (vs findPendingByConversation
// which returns the earliest, wrong when several pendings coexist).
var ownedOpt = approvalService.getPending(pendingId);
if (ownedOpt.isEmpty()
|| !conversationId.equals(ownedOpt.get().getConversationId())) {
return R.fail(404, "Pending approval not found for this session");
}
// Resolve and broadcast outside the persistence transaction: SSE is
// not rollback-capable, so the broadcast must follow a committed DB write.
String actor = webchatUsername(visitorId);
ResolveOutcome outcome = approvalService.resolve(pendingId, actor, "denied");
broadcastApprovalResolved(conversationId, outcome);
audit(channel, visitorId, "webchat.deny-approval", conversationId,
"{\"pendingId\":\"" + escapeJson(pendingId) + "\",\"resolved\":"
+ outcome.dbSynced() + "}");
return R.ok(Map.of("resolved", outcome.dbSynced(), "decision", outcome.decision()));
}
/**
* 批准一个待审批的工具调用并重放 (ISSUE #413 P1-A2 + P1-A3)。
*
* 与 deny 不同，approve 返回 SSE 流：原子消费审批记录后，用捕获的
* toolCallPayload 重放工具调用，把工具结果回灌 agent 继续本轮对话。
* 重放模式对位 web 渠道的 ChatController —— 复用
* {@code chatWithReplayStream} + {@code restoreChatOrigin} 恢复原始
* ChatOrigin (webchat origin 在 createPending 时已通过 ChatOriginHolder
* 持久化到 approval 行)。
*
* 重放期间可能再次触发审批 (一个工具批准后 agent 可能调用下一个受保护
* 工具) —— 该场景由 {@code tool_approval_requested} 直推事件自然覆盖，
* 无需特殊处理。事件投递走与 {@link #chatStream} 相同的 broadcast 路径。
*
* @param pendingId the approval pendingId returned in the
* {@code tool_approval_requested} event
*/
@Operation(summary = "批准访客会话中的待审批工具调用并重放")
@PostMapping(value = "/sessions/approve", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter approveSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestParam String pendingId) {
SseEmitter emitter = new Utf8SseEmitter(sseTimeoutMillis());
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
sendErrorAndComplete(emitter, "Invalid API Key");
return emitter;
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
sendErrorAndComplete(emitter, "Invalid or missing visitor token");
return emitter;
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
sendErrorAndComplete(emitter, ex.getMessage());
return emitter;
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
sendErrorAndComplete(emitter, "Session not found");
return emitter;
}
// IDOR guard (review #415): cross-check the client-supplied pendingId
// actually belongs to this conversation before resolving, otherwise a
// visitor could approve + replay another visitor's guarded tool call.
var ownedApprovalOpt = approvalService.getPending(pendingId);
if (ownedApprovalOpt.isEmpty()
|| !conversationId.equals(ownedApprovalOpt.get().getConversationId())) {
sendErrorAndComplete(emitter, "Pending approval not found for this session");
return emitter;
}
AtomicReference runHandleRef = new AtomicReference<>();
AtomicBoolean disconnected = new AtomicBoolean();
emitter.onCompletion(() -> {
log.debug("[WebChat] approve SSE completed: {}", conversationId);
disconnected.set(true);
streamTracker.detach(runHandleRef.get(), emitter);
});
emitter.onTimeout(() -> {
log.info("[WebChat] approve SSE timeout (stream went idle past the emitter budget): {}", conversationId);
disconnected.set(true);
streamTracker.detach(runHandleRef.get(), emitter);
emitter.complete();
});
emitter.onError(e -> {
if (isClientDisconnect(e)) {
log.debug("[WebChat] approve SSE client disconnected: {} - {}", conversationId, e.getMessage());
} else {
log.info("[WebChat] approve SSE error: {} - {}", conversationId, e.getMessage());
}
disconnected.set(true);
streamTracker.detach(runHandleRef.get(), emitter);
});
String actor = webchatUsername(visitorId);
sseExecutor.execute(() -> {
// Register + attach the emitter FIRST so every downstream branch
// (already-resolved, no-agent, error, replay) can broadcast a
// terminal event the SDK actually receives. Doing this after
// resolveAndConsume left the already-resolved / error paths
// broadcasting into a subscriber-less tracker, so the SSE hung
// to the 10-min timeout (review #415).
ChatStreamTracker.RunHandle runHandle = streamTracker.register(conversationId);
runHandleRef.set(runHandle);
streamTracker.attach(runHandle, emitter);
if (disconnected.get()) {
streamTracker.detach(runHandle, emitter);
}
try {
// Atomically consume the approval (DB + metadata + memory, single tx).
ResolveOutcome consumed = approvalService.resolveAndConsume(pendingId, actor);
if (consumed.consumedSnapshot() == null) {
// already resolved / not found — emit a terminal done so the
// SDK's stream listener closes cleanly instead of hanging.
broadcastApprovalResolved(runHandle, conversationId, consumed);
streamTracker.broadcast(runHandle, "done",
"{\"status\":\"already_resolved\"}");
streamTracker.closeSubscribers(runHandle);
return;
}
// Notify the SDK the approval flipped (clears the banner) before
// replay output starts streaming.
broadcastApprovalResolved(runHandle, conversationId, consumed);
PendingApproval snapshot = consumed.consumedSnapshot();
Long replayAgentId = snapshot.getAgentId() != null
? parseLongOrNull(snapshot.getAgentId()) : null;
if (replayAgentId == null) {
log.warn("[WebChat] approve: no agentId on consumed approval {}, cannot replay",
pendingId);
streamTracker.broadcast(runHandle, "done",
"{\"status\":\"error\",\"message\":\"No agent bound to approval\"}");
streamTracker.closeSubscribers(runHandle);
return;
}
// Restore the original ChatOrigin captured at createPending time.
// Falls back to a fresh webchat origin when none was persisted
// (defensive — mirrors ChatController:304-306).
ChatOrigin replayOrigin =
approvalService.restoreChatOrigin(snapshot.getChatOrigin());
if (replayOrigin == ChatOrigin.EMPTY) {
var agent = agentService.getAgent(replayAgentId);
Long wsId = agent != null ? agent.getWorkspaceId() : 1L;
replayOrigin = ChatOrigin.web(
conversationId, actor, wsId, null).withSender(null, "api", null);
}
// Neutral replay prompt (aligned with IM + web channels — naming a
// tool here can mislead the LLM on fallthrough).
String replayPrompt = "继续执行已批准的工具调用。";
StringBuilder assistantReply = new StringBuilder();
final int[] usage = {0, 0, 0, 0, 0};
final String[] modelInfo = {null, null};
streamTracker.broadcast(runHandle, "message_start",
"{\"role\":\"assistant\"}");
Disposable disposable = agentService.chatWithReplayStream(
replayAgentId, replayPrompt, conversationId,
snapshot.getToolCallPayload(), actor, replayOrigin)
.doOnNext(delta -> {
if (delta.isEvent() && "_usage_final".equals(delta.eventType())) {
Map data = delta.eventData();
usage[0] = ((Number) data.getOrDefault("promptTokens", 0)).intValue();
usage[1] = ((Number) data.getOrDefault("completionTokens", 0)).intValue();
usage[2] = ((Number) data.getOrDefault("cacheReadTokens", 0)).intValue();
usage[3] = ((Number) data.getOrDefault("cacheWriteTokens", 0)).intValue();
usage[4] = ((Number) data.getOrDefault("reasoningTokens", 0)).intValue();
Object model = data.get("runtimeModelName");
Object provider = data.get("runtimeProviderId");
if (model != null) modelInfo[0] = model.toString();
if (provider != null) modelInfo[1] = provider.toString();
}
if (delta.isEvent()) {
forwardVisitorEvent(runHandle, conversationId,
delta.eventType(), delta.eventData());
}
if (delta.content() != null && !delta.content().isEmpty()) {
assistantReply.append(delta.content());
if (!delta.persistenceOnly()) {
streamTracker.broadcast(runHandle, "content_delta",
"{\"text\":" + escapeJson(delta.content()) + "}");
}
}
if (delta.thinking() != null && !delta.thinking().isEmpty()
&& !delta.persistenceOnly()) {
streamTracker.broadcast(runHandle, "thinking_delta",
"{\"text\":" + escapeJson(delta.thinking()) + "}");
}
})
.doOnComplete(() -> {
String reply = assistantReply.toString();
try {
if (!reply.isBlank()) {
conversationService.saveMessage(
conversationId, "assistant", reply, List.of(),
"completed", usage[0], usage[1], usage[2], usage[3], usage[4], modelInfo[0], modelInfo[1], null);
}
} catch (Exception persistErr) {
log.warn("[WebChat] approve replay persist failed: {}", persistErr.getMessage());
}
streamTracker.broadcast(runHandle, "done",
"{\"status\":\"completed\"}");
// Close the WebChat SSE connection on the logical end of
// the replay stream — same rationale as /stream (issue #586).
streamTracker.closeSubscribers(runHandle);
streamTracker.complete(runHandle);
})
.doOnError(e -> {
log.error("[WebChat] approve replay stream error: {}", e.getMessage());
streamTracker.broadcast(runHandle, "error",
"{\"message\":" + escapeJson(e.getMessage()) + "}");
streamTracker.closeSubscribers(runHandle);
streamTracker.complete(runHandle);
})
.subscribe();
streamTracker.setDisposable(runHandle, disposable);
registerEmergencySave(runHandle, conversationId, assistantReply, usage, modelInfo);
} catch (Exception e) {
log.error("[WebChat] approve failed for {}: {}", conversationId, e.getMessage());
try {
streamTracker.broadcast(runHandle, "error",
"{\"message\":" + escapeJson(e.getMessage()) + "}");
} catch (Exception ignored) {}
streamTracker.closeSubscribers(runHandle);
streamTracker.complete(runHandle);
}
});
audit(channel, visitorId, "webchat.approve-approval", conversationId,
"{\"pendingId\":\"" + escapeJson(pendingId) + "\",\"replay\":true}");
return emitter;
}
/** Parse a Long leniently; null/blank/non-numeric return null. */
private static Long parseLongOrNull(String s) {
if (s == null || s.isBlank()) return null;
try {
return Long.parseLong(s.trim());
} catch (NumberFormatException e) {
return null;
}
}
/**
* True when the SSE error is a routine client-side disconnect (closed tab,
* network drop) rather than a server-side failure. Used to keep the
* lifecycle log noise down: a visitor closing the tab is expected and
* stays DEBUG; anything else is worth an INFO line for production triage.
* Mirrors ChatController#isClientDisconnect.
*/
private static boolean isClientDisconnect(Throwable e) {
if (e instanceof IOException) return true;
String msg = e.getMessage();
if (msg == null) return false;
String lower = msg.toLowerCase();
return lower.contains("broken pipe") || lower.contains("connection reset")
|| lower.contains("client abort") || lower.contains("closed");
}
/**
* Broadcast a {@code tool_approval_resolved} event so the SDK clears its
* approval banner in real time. Shared by approve / deny / stop-sweep.
* (ISSUE #413 P1)
*/
private void broadcastApprovalResolved(String conversationId, ResolveOutcome outcome) {
try {
streamTracker.broadcast(conversationId, "tool_approval_resolved",
approvalResolvedJson(outcome));
} catch (Exception e) {
log.debug("[WebChat] approval_resolved broadcast failed for {}: {}",
outcome.pendingId(), e.getMessage());
}
}
private void broadcastApprovalResolved(ChatStreamTracker.RunHandle runHandle,
String conversationId,
ResolveOutcome outcome) {
try {
streamTracker.broadcast(runHandle, "tool_approval_resolved",
approvalResolvedJson(outcome));
} catch (Exception e) {
log.debug("[WebChat] approval_resolved broadcast failed for {} in {}: {}",
outcome.pendingId(), conversationId, e.getMessage());
}
}
private String approvalResolvedJson(ResolveOutcome outcome) throws IOException {
return objectMapper.writeValueAsString(Map.of(
"pendingId", outcome.pendingId(),
"decision", outcome.decision() != null ? outcome.decision() : "",
"toolName", outcome.toolName() != null ? outcome.toolName() : ""));
}
/**
* Regenerate the last assistant reply.
*
* Semantics: stop any in-flight stream, rewind the conversation to the last
* {@code role=user} message (removing the trailing assistant reply), then
* re-run the agent turn from that message. The restart reuses
* {@link #chatStream} with {@code internalSkipUserPersist} set, so the
* existing user row is used as the seed and no duplicate user message is
* inserted.
*
* Returns an error when the conversation has no user message to regenerate from.
*/
@Operation(summary = "重新生成最后一条助手回复")
@PostMapping(value = "/sessions/regenerate", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter regenerateSession(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId) {
SseEmitter emitter = new Utf8SseEmitter(sseTimeoutMillis());
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
sendErrorAndComplete(emitter, "Invalid API Key");
return emitter;
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
sendErrorAndComplete(emitter, "Invalid or missing visitor token");
return emitter;
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
sendErrorAndComplete(emitter, ex.getMessage());
return emitter;
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
sendErrorAndComplete(emitter, "Session not found");
return emitter;
}
// Stop any in-flight stream first so its doOnComplete doesn't race the
// delete/save below. Single-node webchat means requestStop hits the
// right disposable; multi-node is a separate epic.
streamTracker.requestStop(conversationId);
ConversationService.RegenerateSeed seed = conversationService.prepareRegenerate(conversationId);
if (seed == null) {
sendErrorAndComplete(emitter, "No user message to regenerate from");
return emitter;
}
log.info("[WebChat] Regenerate: conversationId={}, visitor={}, seedMessageId={}",
conversationId, visitorId, seed.seedMessageId());
audit(channel, visitorId, "webchat.regenerate-session", conversationId,
"{\"sessionId\":\"" + sid + "\",\"seedMessageId\":" + seed.seedMessageId() + "}");
// Reuse chatStream: it'll resolve the agent again (cheap), re-derive
// conversationId and start the agent turn. The seed user row is reused
// as-is — internalSkipUserPersist stops chatStream from inserting a
// duplicate user row. visitorId echoes through to keep the
// visitor-scoped memory owner consistent.
WebChatRequest req = new WebChatRequest();
req.setMessage(seed.content());
req.setVisitorId(visitorId);
req.setSessionId(sid);
req.setInternalSkipUserPersist(true);
req.setInternalOriginMessageId(seed.seedMessageId());
return chatStream(apiKey, req);
}
/**
* 上传文件（入站）。访客先上传拿到 fileId，再在 /stream 的 attachmentIds 中引用。
*
鉴权同会话接口：API Key + visitor token；conversationId 服务端派生。
*/
@Operation(summary = "WebChat 上传文件")
@PostMapping(value = "/upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public R> uploadFile(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestPart("file") MultipartFile file) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return R.fail(401, "Invalid API Key");
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return R.fail(401, "Invalid or missing visitor token");
}
// Upload requires an established visitor identity (the token is bound to it);
// unlike /stream we never mint a fresh visitorId here.
String vid;
String sid;
try {
if (visitorId == null || visitorId.trim().isEmpty()) {
return R.fail(400, "visitorId is required");
}
vid = normalizeVisitorId(visitorId);
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return R.fail(400, ex.getMessage());
}
String conversationId = deriveConversationId(apiKey, vid, sid);
try {
WebChatFileService.StagedFile stored = fileService.store(conversationId, file);
audit(channel, vid, "webchat.upload-file", conversationId,
"{\"sessionId\":\"" + sid + "\",\"fileId\":\"" + stored.storedName()
+ "\",\"size\":" + stored.size() + "}");
return R.ok(Map.of(
"fileId", stored.storedName(),
"fileName", stored.originalName(),
"contentType", stored.contentType() != null ? stored.contentType() : "application/octet-stream",
"size", stored.size()
));
} catch (WebChatFileService.UploadRejectedException ex) {
return R.fail(400, ex.getMessage());
} catch (IOException ex) {
log.error("[WebChat] Upload failed conv={}: {}", conversationId, ex.getMessage());
return R.fail(500, "Upload failed");
}
}
/**
* 下载文件（出站）。serves both visitor-uploaded files and agent-produced files
* written under the conversation dir. 鉴权同上，路径在服务端派生目录内防穿越。
*/
@Operation(summary = "WebChat 下载文件")
@GetMapping("/files")
public ResponseEntity downloadFile(
@RequestHeader("X-MC-Key") String apiKey,
@RequestHeader(value = "X-MC-Visitor-Token", required = false) String visitorToken,
@RequestParam String visitorId,
@RequestParam(required = false) String sessionId,
@RequestParam String storedName) {
ChannelEntity channel = resolveChannel(apiKey);
if (channel == null) {
return ResponseEntity.status(401).build();
}
if (!verifyVisitorToken(visitorTokenSecret, channel.getId(), visitorId, visitorToken)) {
return ResponseEntity.status(401).build();
}
String sid;
try {
sid = normalizeSessionId(sessionId);
} catch (IllegalArgumentException ex) {
return ResponseEntity.badRequest().build();
}
String conversationId = deriveConversationId(apiKey, visitorId, sid);
if (!ownsConversation(conversationId, visitorId)) {
return ResponseEntity.status(404).build();
}
Path file = fileService.resolve(conversationId, storedName).orElse(null);
if (file == null) {
return ResponseEntity.notFound().build();
}
String contentType;
try {
contentType = Files.probeContentType(file);
} catch (IOException e) {
contentType = null;
}
MediaType mediaType = MediaType.APPLICATION_OCTET_STREAM;
if (contentType != null) {
try {
mediaType = MediaType.parseMediaType(contentType);
} catch (Exception ignored) {
// fall back to octet-stream
}
}
// Only inline images; everything else downloads as an attachment. nosniff
// stops the browser from re-interpreting the bytes as active content.
boolean inlineImage = contentType != null && contentType.startsWith("image/");
String encodedName = URLEncoder.encode(file.getFileName().toString(), StandardCharsets.UTF_8)
.replace("+", "%20");
return ResponseEntity.ok()
.contentType(mediaType)
.header("X-Content-Type-Options", "nosniff")
.header(HttpHeaders.CONTENT_DISPOSITION,
(inlineImage ? "inline" : "attachment") + "; filename*=UTF-8''" + encodedName)
.body(new FileSystemResource(file));
}
/**
* Build the user message's content parts: a text part for the message plus a
* file/media part for each referenced attachment. Attachment metadata is
* resolved server-side from the staging registry (the client only sends opaque
* ids); an id that is unknown, expired, or belongs to another conversation is
* silently dropped.
*/
private List buildUserParts(String conversationId, String message,
List attachmentIds) {
List parts = new ArrayList<>();
if (message != null && !message.isBlank()) {
MessageContentPart text = new MessageContentPart();
text.setType("text");
text.setText(message);
parts.add(text);
}
if (attachmentIds != null) {
for (String fileId : attachmentIds) {
fileService.consume(conversationId, fileId).ifPresent(sf -> {
MessageContentPart p = new MessageContentPart();
p.setType(WebChatFileService.partTypeFor(sf.contentType()));
p.setFileName(sf.originalName());
p.setContentType(sf.contentType());
p.setStoredName(sf.storedName());
p.setFileSize(sf.size());
// Relative download ref (caller adds auth headers + visitorId/sessionId).
p.setFileUrl("/api/v1/channels/webchat/files?storedName="
+ URLEncoder.encode(sf.storedName(), StandardCharsets.UTF_8));
// Server path lets the agent's file tools read the upload; stripped from
// the external message view (listMessageViewsExternal).
fileService.resolve(conversationId, sf.storedName())
.ifPresent(path -> p.setPath(path.toString()));
parts.add(p);
});
}
}
return parts;
}
/**
* Register an emergency-save callback that flushes the partial assistant
* reply accumulated so far as an {@code interrupted} message. Wired on
* both {@code /stream} and {@code /sessions/approve} so that when a run is
* reclaimed while its only subscriber is gone (orphan-grace eviction,
* shutdown, admin force-recycle), the visitor can still retrieve the
* partial answer via {@code /sessions/messages} instead of seeing only the
* user message (issue #587). Mirrors ChatController#emergencySaveAccumulator.
*
* @param conversationId target conversation
* @param assistantReply live accumulator appended to in doOnNext
* @param usage [prompt, completion, cacheRead, cacheWrite, reasoning]
* @param modelInfo [runtimeModel, runtimeProvider]
*/
private void registerEmergencySave(ChatStreamTracker.RunHandle runHandle,
String conversationId, StringBuilder assistantReply,
int[] usage, String[] modelInfo) {
streamTracker.setEmergencySaveCallback(runHandle, () -> {
try {
String reply = assistantReply.toString();
if (reply.isBlank()) {
log.debug("[WebChat] Emergency save skipped (empty reply): {}", conversationId);
return;
}
// usage length varies by call site (chatStream = 5 tokens,
// approve replay = 2); read defensively so the save never
// throws ArrayIndexOutOfBoundsException.
int prompt = usage.length > 0 ? usage[0] : 0;
int completion = usage.length > 1 ? usage[1] : 0;
int cacheRead = usage.length > 2 ? usage[2] : 0;
int cacheWrite = usage.length > 3 ? usage[3] : 0;
int reasoning = usage.length > 4 ? usage[4] : 0;
String runtimeModel = modelInfo.length > 0 ? modelInfo[0] : null;
String runtimeProvider = modelInfo.length > 1 ? modelInfo[1] : null;
conversationService.saveMessage(
conversationId, "assistant", reply, List.of(),
"interrupted",
prompt, completion, cacheRead, cacheWrite, reasoning,
runtimeModel, runtimeProvider, null);
log.info("[WebChat] Emergency-saved partial assistant reply: " +
"conversationId={}, textLen={}",
conversationId, reply.length());
} catch (Exception e) {
log.warn("[WebChat] Emergency save failed for {}: {}", conversationId, e.getMessage());
}
});
}
// ==================== 内部方法 ====================
private static final Pattern SESSION_ID_PATTERN = Pattern.compile("[A-Za-z0-9_-]{1,64}");
/**
* 归一化调用方传入的 sessionId：空白 → null；非空必须满足白名单字符集，否则抛出。
*/
private String normalizeSessionId(String raw) {
if (raw == null) {
return null;
}
String s = raw.trim();
if (s.isEmpty()) {
return null;
}
if (!SESSION_ID_PATTERN.matcher(s).matches()) {
throw new IllegalArgumentException(
"Invalid sessionId (allowed: letters, digits, '-', '_', length 1-64)");
}
return s;
}
private static final Pattern VISITOR_ID_PATTERN = Pattern.compile("[A-Za-z0-9_.:\\-]{1,128}");
/**
* 归一化调用方传入的 visitorId：空白 → 新 UUID；非空必须满足白名单字符集，否则抛出。
* 限制字符集既防注入/控制字符，也为派生的 conversationId / username 提供可预期的边界。
*/
private String normalizeVisitorId(String raw) {
if (raw == null || raw.trim().isEmpty()) {
return UUID.randomUUID().toString();
}
String s = raw.trim();
if (!VISITOR_ID_PATTERN.matcher(s).matches()) {
throw new IllegalArgumentException(
"Invalid visitorId (allowed: letters, digits, '-', '_', '.', ':', length 1-128)");
}
return s;
}
/**
* 由服务端拼装 conversationId，始终钳在 key + visitor 命名空间内。
* 绝不接受调用方传入的裸 conversationId。
* conversation_id 列为 VARCHAR(64)；当 visitorId + sessionId 过长导致超出列宽时，
* 把可变部分折叠为稳定哈希，保证 id 唯一且有界（否则 INSERT 会在 /stream 处 500）。
*/
static String deriveConversationId(String apiKey, String visitorId, String sessionId) {
String key8 = apiKey.substring(0, Math.min(8, apiKey.length()));
String full = "webchat:" + key8 + ":" + visitorId + (sessionId != null ? ":" + sessionId : "");
if (full.length() <= 64) {
return full;
}
return "webchat:" + key8 + ":#"
+ sha256Hex(visitorId + "\0" + (sessionId == null ? "" : sessionId)).substring(0, 40);
}
/**
* 由 visitorId 派生 username（mate_conversation.username，VARCHAR(64)）。
* 同样在超长时折叠为哈希，避免 username 溢出列宽。
*/
static String webchatUsername(String visitorId) {
String u = "webchat:" + visitorId;
return u.length() <= 64 ? u : "webchat:#" + sha256Hex(visitorId).substring(0, 40);
}
private static String sha256Hex(String s) {
try {
byte[] d = java.security.MessageDigest.getInstance("SHA-256")
.digest(s.getBytes(java.nio.charset.StandardCharsets.UTF_8));
StringBuilder sb = new StringBuilder(d.length * 2);
for (byte b : d) {
sb.append(Character.forDigit((b >> 4) & 0xF, 16)).append(Character.forDigit(b & 0xF, 16));
}
return sb.toString();
} catch (java.security.NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 unavailable", e);
}
}
/**
* 存在性守卫：会话存在且属于本 visitor 命名空间时返回 true，否则 404。
*
注意：这不是鉴权边界——conversationId 由调用方自报的 visitorId 派生，
* 等式两边同源，单凭它无法防越权。真正的鉴权由 {@link #verifyVisitorToken} 完成。
*/
private boolean ownsConversation(String conversationId, String visitorId) {
ConversationEntity conv = conversationService.findByConversationId(conversationId);
return conv != null && webchatUsername(visitorId).equals(conv.getUsername());
}
/**
* 用服务端密钥对 (channelId, visitorId) 做 HMAC-SHA256，签发不可伪造的 visitor token。
* 载荷含 channelId，使 token 不能跨渠道复用。
*
Token 默认 7 天后过期（{@link #VISITOR_TOKEN_TTL_SECONDS}）；过期时间作为后缀
* 明文附加在 HMAC 之后（{@code .}），既参与签名也方便解析。
* 过期后访客可通过 {@code /stream} 重新签发（{@code /stream} 不校验 token，只签发）。
*/
static String computeVisitorToken(String secret, Long channelId, String visitorId) {
return computeVisitorToken(secret, channelId, visitorId,
java.time.Instant.now().getEpochSecond() + VISITOR_TOKEN_TTL_SECONDS);
}
/** Test/override hook: explicit expiration epoch second. */
static String computeVisitorToken(String secret, Long channelId, String visitorId, long expiresAtEpochSecond) {
try {
Mac mac = Mac.getInstance("HmacSHA256");
mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
String payload = channelId + ":" + visitorId + ":" + expiresAtEpochSecond;
byte[] sig = mac.doFinal(payload.getBytes(StandardCharsets.UTF_8));
return Base64.getUrlEncoder().withoutPadding().encodeToString(sig) + "." + expiresAtEpochSecond;
} catch (GeneralSecurityException e) {
throw new IllegalStateException("HMAC-SHA256 unavailable", e);
}
}
/**
* 校验调用方回传的 token 的签名 + 过期。不查撤销表(撤销是实例层职责,
* 见 {@link #verifyVisitorToken})。Static 是为了让单测可以直接验证 HMAC 语义,
* 不需要起 Spring context。
*/
static boolean verifyVisitorTokenSignature(String secret, Long channelId, String visitorId, String presented) {
if (presented == null || presented.isEmpty() || visitorId == null || channelId == null) {
return false;
}
int dot = presented.lastIndexOf('.');
if (dot <= 0 || dot == presented.length() - 1) {
return false;
}
long exp;
try {
exp = Long.parseLong(presented.substring(dot + 1));
} catch (NumberFormatException e) {
return false;
}
if (java.time.Instant.now().getEpochSecond() >= exp) {
return false;
}
// Constant-time comparison of the full token (sig + ".exp"). HMAC covers
// both channelId:visitorId and exp, so any tampering with exp invalidates sig.
byte[] expected = computeVisitorToken(secret, channelId, visitorId, exp).getBytes(StandardCharsets.UTF_8);
byte[] actual = presented.getBytes(StandardCharsets.UTF_8);
return MessageDigest.isEqual(expected, actual);
}
/**
* 完整校验:签名 + 过期 + 撤销。任一不通过返回 false。实例方法,接入
* {@link WebChatTokenRevocationService}。{@code /stream} 第一次接触不调用本方法
* (只签发 token,不校验),所以被撤销的 visitor 仍能发起新会话——撤销只让旧的
* 管理 token 失效,符合 issue #351 的设计。
*/
boolean verifyVisitorToken(String secret, Long channelId, String visitorId, String presented) {
if (!verifyVisitorTokenSignature(secret, channelId, visitorId, presented)) {
return false;
}
if (tokenRevocationService != null && tokenRevocationService.isRevoked(channelId, visitorId)) {
return false;
}
return true;
}
/**
* 通过 API Key 查找 WebChat 渠道
*/
private ChannelEntity resolveChannel(String apiKey) {
if (apiKey == null || apiKey.isBlank()) {
return null;
}
List channels = channelService.listChannelsByType("webchat");
for (ChannelEntity channel : channels) {
if (!Boolean.TRUE.equals(channel.getEnabled())) continue;
JsonNode config = parseConfig(channel.getConfigJson());
String configuredApiKey = textOrDefault(config, "api_key", null);
if (configuredApiKey != null && apiKey.equals(configuredApiKey)) {
return channel;
}
}
return null;
}
private JsonNode parseConfig(String configJson) {
if (configJson == null || configJson.isBlank()) {
return objectMapper.createObjectNode();
}
try {
return objectMapper.readTree(configJson);
} catch (Exception e) {
log.warn("[WebChat] Failed to parse configJson: {}", e.getMessage());
return objectMapper.createObjectNode();
}
}
private String textOrDefault(JsonNode node, String fieldName, String defaultValue) {
if (node != null) {
JsonNode value = node.get(fieldName);
if (value != null && !value.isNull()) {
String text = value.asText();
if (!text.isBlank()) {
return text;
}
}
}
return defaultValue;
}
private void sendErrorAndComplete(SseEmitter emitter, String message) {
try {
emitter.send(SseEmitter.event().name("error").data(Map.of("message", message)));
emitter.complete();
} catch (IOException e) {
emitter.completeWithError(e);
}
}
/**
* Forward a curated subset of agent lifecycle events to the visitor SSE
* stream as visitor-friendly {@code phase} / {@code tool_start} /
* {@code tool_end} / {@code plan} events. Internal event types
* ({@code _usage_final}, {@code _routing_decision}, {@code iteration_*},
* {@code perf_summary}, {@code feedback_event}, {@code finish_reason},
* {@code plan_step_*}) are dropped — they leak graph internals and have
* no visitor-facing value.
*
* Tool arguments are deliberately not forwarded. The agent may
* invoke tools with PII / sensitive arguments (file paths, user queries,
* credentials); relaying those to a 3rd-party website frontend is a data
* leak. The frontend gets only the tool name and renders a localized
* label via its own lookup table.
*
*
Payloads are serialized via the injected {@link ObjectMapper} so
* nested maps/lists are encoded correctly (the hand-rolled {@link #escapeJson}
* helper is string-only).
*
*
Backward compat: visitors / SDKs that don't know these event types
* silently ignore them per the SSE spec.
*/
private void forwardVisitorEvent(ChatStreamTracker.RunHandle runHandle,
String conversationId,
String eventType,
Map data) {
if (eventType == null || data == null) return;
Map payload;
String sseName;
switch (eventType) {
case "phase":
// Graph phase transition (planning / thinking / generating /
// summarizing / ...). Lets the SDK show a typing indicator
// before the first content_delta lands.
sseName = "phase";
payload = Map.of(
"phase", String.valueOf(data.getOrDefault("phase", "")),
"timestamp", System.currentTimeMillis());
break;
case "tool_call_started":
// Tool invocation started. Args intentionally omitted — see javadoc.
sseName = "tool_start";
payload = Map.of(
"tool", String.valueOf(data.getOrDefault("toolName",
data.getOrDefault("tool", ""))));
break;
case "tool_call_completed":
// Tool invocation finished. Result content intentionally omitted.
sseName = "tool_end";
payload = new java.util.LinkedHashMap<>();
payload.put("tool", String.valueOf(data.getOrDefault("toolName",
data.getOrDefault("tool", ""))));
Object success = data.get("success");
payload.put("success", success != null ? success : Boolean.TRUE);
break;
case "plan_created":
// Plan-Execute agents expose their step list. The SDK can render
// a checklist; subsequent plan_step_* events are dropped (too
// granular for a visitor view).
sseName = "plan";
payload = Map.of("steps", data.getOrDefault("steps", List.of()));
break;
default:
// Curated allow-list: anything else is internal — silently drop.
return;
}
try {
String json = objectMapper.writeValueAsString(payload);
streamTracker.broadcast(runHandle, sseName, json);
} catch (Exception e) {
log.debug("[WebChat] Failed to serialize visitor event {} for {}: {}",
eventType, conversationId, e.getMessage());
}
}
private String escapeJson(String value) {
if (value == null) return "null";
return "\"" + value
.replace("\\", "\\\\")
.replace("\"", "\\\"")
.replace("\n", "\\n")
.replace("\r", "\\r")
.replace("\t", "\\t")
+ "\"";
}
// ==================== 请求体 ====================
@lombok.Data
public static class WebChatRequest {
private String message;
private String visitorId;
/** Optional: route this call to a specific agent instead of the channel's bound agent.
* Must belong to the channel's workspace.
* Only applied when the (visitorId + sessionId) conversation is first created. Once that
* conversation exists, its agent is fixed: a different agentId on later requests is silently
* ignored. To talk to another agent, use a new sessionId (or a new visitorId). */
private Long agentId;
/** Optional: open a distinct conversation thread for the same visitor.
* Composed into the server-derived conversationId; never used as a raw conversationId. */
private String sessionId;
/** Optional: ids returned by POST /upload, referencing files this visitor uploaded
* for this conversation. Metadata is resolved server-side; unknown / foreign / expired
* ids are dropped. */
private List attachmentIds;
/**
* Internal-only regenerate flag: the seed user row is already
* persisted, so {@code chatStream} must not insert a duplicate.
* Excluded from JSON binding — never client-settable.
*/
@JsonIgnore
private boolean internalSkipUserPersist;
private Long internalOriginMessageId;
}
/** Compact view of one of a visitor's conversation threads. */
@lombok.Data
@lombok.AllArgsConstructor
public static class WebChatSessionView {
/** null for the visitor's default (no-session) thread. */
private String sessionId;
private String title;
private LocalDateTime lastActiveTime;
private Integer messageCount;
/** 1 if the visitor pinned this thread, 0 otherwise. */
private Integer pinned;
/** 1 if the visitor archived this thread, 0 otherwise. */
private Integer archived;
/** {@code running} if a stream is in progress on this thread, else {@code idle}. */
private String streamStatus;
}
/**
* Display-level view of a skill surfaced to webchat visitors for the slash
* picker UI. Carries only the fields a UI needs to render a row — id (for
* logging / debugging), localised name, description, icon. Deliberately
* omits SKILL.md content, config JSON, scan results and other internal
* columns: those never leave the admin console.
*/
@lombok.Data
public static class WebChatSkillView {
private Long id;
/** Immutable slug the LLM takes as {@code load_skill(name=…)}; this is what the slash picker must splice into the directive text. */
private String name;
private String nameZh;
private String nameEn;
private String description;
private String icon;
static WebChatSkillView from(vip.mate.skill.model.SkillEntity s) {
WebChatSkillView v = new WebChatSkillView();
v.id = s.getId();
v.name = s.getName();
v.nameZh = s.getNameZh();
v.nameEn = s.getNameEn();
v.description = s.getDescription();
v.icon = s.getIcon();
return v;
}
}
/**
* Display-level projection of a wiki page for the visitor-facing
* {@code [[slug]]} picker. Carries the slug the LLM consumes (via
* {@code wiki_read_page(slug=…)}), the human-readable title/summary for
* picker UI, and the KB id + name for disambiguation when an agent is
* scoped to multiple KBs that may share a slug. Deliberately omits
* content / embedding / sourceRawIds / outgoingLinks — those stay
* admin-console-only.
*/
@lombok.Data
public static class WebChatWikiPageView {
private Long kbId;
private String kbName;
/** Immutable slug — what the picker splices into the {@code [[slug]]} token; the LLM consumes it as {@code wiki_read_page(slug=…)}. */
private String slug;
private String title;
private String summary;
/** {@code entity} / {@code concept} / {@code source} / ... — never {@code synthesis} (filtered out upstream). Useful for picker grouping/icons. */
private String pageType;
static WebChatWikiPageView from(vip.mate.wiki.model.WikiPageEntity p, String kbName) {
WebChatWikiPageView v = new WebChatWikiPageView();
v.kbId = p.getKbId();
v.kbName = kbName;
v.slug = p.getSlug();
v.title = p.getTitle();
v.summary = p.getSummary();
v.pageType = p.getPageType();
return v;
}
}
/** Body for {@code POST /sessions} — explicitly create an empty thread. */
@lombok.Data
public static class WebChatCreateSessionRequest {
/** Optional; server mints a UUID when absent (same convention as /stream). */
private String visitorId;
/** Optional; server generates one when absent. Whitelisted charset, ≤ 64 chars. */
private String sessionId;
/** Optional; 1–100 chars when non-blank, otherwise left null so the first
* /stream message still derives the title (mirrors PUT /sessions/title rules). */
private String title;
/** Optional; override the channel's bound agent. Must belong to the channel's
* workspace. Only applied on first creation — once the thread exists, a
* different agentId is ignored. */
private Long agentId;
}
}