Flow per stream: *
Designed mirror-image to {@code DingTalkAICardManager}: same * create/append/finish/fail shape, same per-session throttling, * same activeSessions map for hand-off between threads. The * implementation is end-to-end {@code oapi-sdk} — no hand-rolled HTTP. * *
The four SDK call sites are {@code protected} so unit tests can
* subclass and verify session/throttle behavior without booting a real
* Feishu credential or hitting the network.
*/
@Slf4j
@Component
public class FeishuStreamingCardManager {
/** Throttle window for {@link #appendContent}, ms — matches DingTalk AICard. */
static final long THROTTLE_INTERVAL_MS = 500;
/**
* Hard per-card operation spacing. Feishu allows at most 10 CardKit
* operations/second for one card; 120ms leaves a little clock/network
* jitter headroom while still letting phase transitions feel immediate.
*/
static final long PLATFORM_MIN_INTERVAL_MS = 120;
/**
* Markdown element id baked into the initial streaming card.
* Content-update calls reference this id. Public so tests can assert.
*/
public static final String STREAM_ELEMENT_ID = "stream_md";
/** Default text shown when the card is first created, before any delta arrives. */
public static final String DEFAULT_INITIAL_TEXT = "🤔 思考中...";
private final FeishuClientFactory clientFactory;
private final ObjectMapper objectMapper;
/** sessionKey → CardSession. sessionKey is an opaque UUID handed back to the caller. */
private final ConcurrentHashMap No-op when {@code sessionKey} is unknown or the session has
* already reached a terminal status — keeps the caller's
* {@code doOnNext} loop simple ("just push every chunk").
*/
public void appendContent(String sessionKey, String contentDelta, boolean forceFlush) {
CardSession session = activeSessions.get(sessionKey);
if (session == null || !session.isStreaming()) {
return;
}
if (contentDelta != null && !contentDelta.isEmpty()) {
synchronized (session) {
session.accumulated.append(contentDelta);
}
}
flushWithPolicy(session, forceFlush);
}
/**
* Replace the streaming element with a full progress snapshot.
*
* CardKit's content API expects the complete current text on every
* update. Agent progress is not append-only ("thinking" becomes "calling
* a tool", then "replying"), so treating snapshots as deltas duplicates
* the entire trace on every refresh.
*/
public boolean updateContent(String sessionKey, String fullContent, boolean forceFlush) {
CardSession session = activeSessions.get(sessionKey);
if (session == null || !session.isStreaming()) return false;
replaceAccumulated(session, fullContent != null ? fullContent : "");
return flushWithPolicy(session, forceFlush);
}
/**
* Push final content and turn off streaming mode. Idempotent —
* a second call is a no-op. After return, the sessionKey is no
* longer known to the manager.
*/
public FinishResult finishCard(String sessionKey, String finalContent) {
CardSession session = activeSessions.get(sessionKey);
if (session == null) return new FinishResult(false, false);
if (!session.status.compareAndSet(Status.STREAMING, Status.FINISHED)) {
return new FinishResult(false, false);
}
boolean contentUpdated = false;
boolean streamingClosed = false;
try {
replaceAccumulated(session, finalContent != null ? finalContent : "");
contentUpdated = flushWithRetry(session);
streamingClosed = closeStreamingWithRetry(session, summaryFor(finalContent));
return new FinishResult(contentUpdated, streamingClosed);
} finally {
activeSessions.remove(sessionKey);
log.info("[feishu-stream] Card finished: sessionKey={}, contentLen={}, contentUpdated={}, closed={}",
sessionKey, finalContent == null ? 0 : finalContent.length(),
contentUpdated, streamingClosed);
}
}
/**
* Mark the session failed. The current accumulator gets an error
* suffix; the card is closed so the typing animation stops.
* Idempotent.
*/
public FinishResult failCard(String sessionKey, String errorMessage) {
CardSession session = activeSessions.get(sessionKey);
if (session == null) return new FinishResult(false, false);
if (!session.status.compareAndSet(Status.STREAMING, Status.FAILED)) {
return new FinishResult(false, false);
}
boolean contentUpdated = false;
boolean streamingClosed = false;
try {
String tail;
synchronized (session) {
if (session.accumulated.length() == 0) {
tail = "⚠️ 处理失败：" + safe(errorMessage);
} else {
tail = session.accumulated + "\n\n⚠️ " + safe(errorMessage);
}
session.accumulated.setLength(0);
session.accumulated.append(tail);
}
contentUpdated = flushWithRetry(session);
streamingClosed = closeStreamingWithRetry(session, "⚠️ 处理失败");
return new FinishResult(contentUpdated, streamingClosed);
} finally {
activeSessions.remove(sessionKey);
log.warn("[feishu-stream] Card failed: sessionKey={}, contentUpdated={}, closed={}, error={}",
sessionKey, contentUpdated, streamingClosed, errorMessage);
}
}
// ------------------------------------------------------------------
// Inspection helpers (tests / metrics)
// ------------------------------------------------------------------
/** Visible for tests / metrics — number of in-flight sessions. */
public int activeSessionCount() {
return activeSessions.size();
}
/** Visible for tests — direct session lookup. */
CardSession sessionFor(String sessionKey) {
return activeSessions.get(sessionKey);
}
// ------------------------------------------------------------------
// Internal — flush + SDK seams
// ------------------------------------------------------------------
private boolean flushWithPolicy(CardSession session, boolean forceFlush) {
long now = currentTimeMs();
long elapsed = now - session.lastFlushMs;
if (!forceFlush && elapsed < THROTTLE_INTERVAL_MS) {
return true; // latest snapshot is queued in session.accumulated
}
if (forceFlush && elapsed < PLATFORM_MIN_INTERVAL_MS) {
if (!pauseBeforeFlush(PLATFORM_MIN_INTERVAL_MS - elapsed)) return false;
now = currentTimeMs();
}
return flush(session, now);
}
/** One retry is enough to cover a transient rate-limit/network blip. */
private boolean flushWithRetry(CardSession session) {
if (flushWithPolicy(session, true)) return true;
return flushWithPolicy(session, true);
}
private boolean flush(CardSession session, long now) {
String snapshot;
synchronized (session) {
snapshot = session.accumulated.toString();
}
int seq = session.sequence.incrementAndGet();
try {
Client client = clientFactory.client(session.channelId);
sdkPushElementContent(client, session.cardId, STREAM_ELEMENT_ID, snapshot, seq);
return true;
} catch (Exception e) {
log.warn("[feishu-stream] flush failed: sessionKey={}, seq={}, err={}",
session.sessionKey, seq, e.getMessage());
return false;
} finally {
// Failed requests count against platform rate limits too.
session.lastFlushMs = now;
}
}
private boolean closeStreamingWithRetry(CardSession session, String summary) {
if (closeStreaming(session, summary)) return true;
return closeStreaming(session, summary);
}
private boolean closeStreaming(CardSession session, String summary) {
long elapsed = currentTimeMs() - session.lastFlushMs;
if (elapsed < PLATFORM_MIN_INTERVAL_MS
&& !pauseBeforeFlush(PLATFORM_MIN_INTERVAL_MS - elapsed)) {
return false;
}
int seq = session.sequence.incrementAndGet();
try {
Client client = clientFactory.client(session.channelId);
sdkCloseStreamingMode(client, session.cardId, seq, summary);
return true;
} catch (Exception e) {
log.warn("[feishu-stream] closeStreaming failed: sessionKey={}, err={}",
session.sessionKey, e.getMessage());
return false;
} finally {
session.lastFlushMs = currentTimeMs();
}
}
private void tryCloseStreamingSilently(Client client, String cardId) {
try {
sdkCloseStreamingMode(client, cardId, 1, "⚠️ 卡片发送失败");
} catch (Exception ignore) {
// best-effort — already in an error path
}
}
private void replaceAccumulated(CardSession session, String content) {
synchronized (session) {
session.accumulated.setLength(0);
session.accumulated.append(content);
}
}
private boolean pauseBeforeFlush(long millis) {
if (millis <= 0) return true;
try {
sleepMillis(millis);
return true;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
/** Test seam for advancing a fake clock without real sleeping. */
protected void sleepMillis(long millis) throws InterruptedException {
Thread.sleep(millis);
}
static String summaryFor(String content) {
String preview = content == null ? "" : content
.replaceAll("[`*_>#~-]+", " ")
.replaceAll("\\s+", " ")
.trim();
if (preview.isEmpty()) return "✅ 已完成";
return preview.length() <= 80 ? preview : preview.substring(0, 77) + "...";
}
// ------------------------------------------------------------------
// SDK seams (overridable in tests)
// ------------------------------------------------------------------
/** Build a streaming-mode schema-2.0 card. Returns the new card_id or null. */
protected String sdkCreateCard(Client client, String initialText) throws Exception {
String cardJson = objectMapper.writeValueAsString(buildInitialCardJson(initialText));
CreateCardReq req = CreateCardReq.newBuilder()
.createCardReqBody(CreateCardReqBody.newBuilder()
.type("card_json")
.data(cardJson)
.build())
.build();
CreateCardResp resp = client.cardkit().v1().card().create(req);
if (!resp.success() || resp.getData() == null) {
log.warn("[feishu-stream] card.create failed: code={}, msg={}", resp.getCode(), resp.getMsg());
return null;
}
return resp.getData().getCardId();
}
/** Send the freshly-built card as an interactive message. Returns message_id or null. */
protected String sdkSendInteractiveMessage(Client client, String receiveIdType,
String receiveId, String cardId) throws Exception {
Map