()
.select(ConversationEntity::getConversationId, ConversationEntity::getAgentId,
ConversationEntity::getWorkspaceId, ConversationEntity::getCreateTime,
ConversationEntity::getLastActiveTime)
.isNotNull(ConversationEntity::getAgentId)
.ge(ConversationEntity::getLastActiveTime, cutoff)
.orderByDesc(ConversationEntity::getLastActiveTime);
return conversationMapper.selectPage(page, q).getRecords();
}
/**
* First user message of each conversation, keyed by conversation id.
*
* Loads user messages in batched {@code IN} clauses and keeps the
* lowest-id row per conversation. Cost scales with the number of user
* messages in the scanned conversations, which the caller bounds through
* {@code maxConversationsPerRun}; this runs as a nightly sweep, not on a
* request path.
*/
private Map loadOpeners(List conversations) {
List ids = new ArrayList<>();
for (ConversationEntity c : conversations) {
if (c.getConversationId() != null) {
ids.add(c.getConversationId());
}
}
Map openers = new HashMap<>();
for (int i = 0; i < ids.size(); i += OPENER_BATCH_SIZE) {
List batch = ids.subList(i, Math.min(ids.size(), i + OPENER_BATCH_SIZE));
List rows;
try {
rows = messageMapper.selectList(new LambdaQueryWrapper()
.select(MessageEntity::getConversationId, MessageEntity::getContent)
.eq(MessageEntity::getRole, "user")
.in(MessageEntity::getConversationId, batch)
.orderByAsc(MessageEntity::getId));
} catch (Exception e) {
log.warn("[SkillRoutine] Opener batch load failed: {}", e.getMessage());
continue;
}
for (MessageEntity m : rows) {
if (m.getConversationId() == null || m.getContent() == null) {
continue;
}
// Ascending id, so the first row seen per conversation is its opener.
openers.putIfAbsent(m.getConversationId(), m.getContent());
}
}
return openers;
}
private static LocalDateTime conversationStart(ConversationEntity conv) {
return conv.getCreateTime() != null ? conv.getCreateTime() : conv.getLastActiveTime();
}
// ==================== Normalization + clustering ====================
/**
* Strip everything that varies between two runs of the same routine —
* URLs, paths, numbers, punctuation, case — leaving the stable intent.
* "generate the 2026-08-04 report" and "generate the 2026-08-05 report"
* must normalize to the same text or they will never cluster.
*/
String normalize(String raw) {
if (raw == null || raw.isBlank()) {
return "";
}
String text = raw.strip();
int max = Math.max(20, properties.getMaxOpenerChars());
if (text.length() > max) {
text = text.substring(0, max);
}
text = URL_RE.matcher(text).replaceAll(" ");
text = PATH_RE.matcher(text).replaceAll(" ");
text = DIGITS_RE.matcher(text).replaceAll(" ");
text = text.toLowerCase();
text = NOISE_RE.matcher(text).replaceAll(" ");
return SPACE_RE.matcher(text).replaceAll(" ").strip();
}
/**
* Greedy single-pass clustering against each existing cluster's seed.
*
* Seed comparison (rather than full linkage) keeps clusters tight: a
* chain of pairwise-similar openers cannot drift into one blob where the
* first and last members share nothing.
*/
List cluster(List openers) {
List clusters = new ArrayList<>();
double threshold = properties.getSimilarityThreshold();
for (Opener opener : openers) {
Cluster match = null;
double best = threshold;
for (Cluster c : clusters) {
double score = Shingles.jaccard(opener.shingles(), c.seed().shingles());
if (score >= best) {
best = score;
match = c;
}
}
if (match == null) {
clusters.add(new Cluster(opener));
} else {
match.members().add(opener);
}
}
return clusters;
}
// ==================== Persistence ====================
/** @return {@code true} when a row was inserted or refreshed */
private boolean upsert(Long agentId, Cluster cluster) {
Opener seed = cluster.seed();
Opener latest = cluster.latest();
String signature = truncate(seed.normalized(), 512);
String hash = SecureUtil.sha256(signature);
SkillRoutineCandidateEntity existing = candidateMapper.selectOne(
new LambdaQueryWrapper()
.eq(SkillRoutineCandidateEntity::getAgentId, agentId)
.eq(SkillRoutineCandidateEntity::getSignatureHash, hash)
.last("LIMIT 1"));
if (existing != null
&& SkillRoutineCandidateEntity.STATUS_DISMISSED.equals(existing.getStatus())) {
// The operator rejected this routine; never resurrect it.
return false;
}
SkillRoutineCandidateEntity row = existing == null ? new SkillRoutineCandidateEntity() : existing;
row.setAgentId(agentId);
row.setWorkspaceId(seed.workspaceId());
row.setSignature(signature);
row.setSignatureHash(hash);
row.setRepresentativeText(truncate(latest.rawOpener(), 2048));
row.setSampleConversations(serializeSamples(cluster));
row.setOccurrenceCount(cluster.members().size());
row.setDistinctDayCount(cluster.distinctDays());
row.setFirstSeenAt(earliest(cluster));
row.setLastSeenAt(latest.seenAt());
if (row.getStatus() == null) {
row.setStatus(SkillRoutineCandidateEntity.STATUS_OBSERVING);
}
try {
if (existing == null) {
candidateMapper.insert(row);
} else {
candidateMapper.updateById(row);
}
return true;
} catch (Exception e) {
log.warn("[SkillRoutine] Candidate upsert failed for agent={} signature='{}': {}",
agentId, signature, e.getMessage());
return false;
}
}
private String serializeSamples(Cluster cluster) {
List ids = new ArrayList<>();
// Newest first: the synthesis prompt should see current phrasing.
List members = new ArrayList<>(cluster.members());
members.sort((a, b) -> {
if (a.seenAt() == null) return 1;
if (b.seenAt() == null) return -1;
return b.seenAt().compareTo(a.seenAt());
});
for (Opener o : members) {
if (ids.size() >= properties.getMaxSamplesPerCandidate()) {
break;
}
ids.add(o.conversationId());
}
try {
return objectMapper.writeValueAsString(ids);
} catch (Exception e) {
return "[]";
}
}
private static LocalDateTime earliest(Cluster cluster) {
LocalDateTime best = null;
for (Opener o : cluster.members()) {
if (o.seenAt() != null && (best == null || o.seenAt().isBefore(best))) {
best = o.seenAt();
}
}
return best;
}
private static String truncate(String s, int maxLen) {
if (s == null) {
return null;
}
return s.length() <= maxLen ? s : s.substring(0, maxLen);
}
}