diff --git a/src/prerna/collaboration/BrainTopicMatches.java b/src/prerna/collaboration/BrainTopicMatches.java index 7a14297063..524065a707 100644 --- a/src/prerna/collaboration/BrainTopicMatches.java +++ b/src/prerna/collaboration/BrainTopicMatches.java @@ -58,6 +58,11 @@ public final class BrainTopicMatches { // the topic is the clear best fit, or fits next to a topic the conversation already has static final String BEST = "best"; static final String SHARED = "shared"; + // re-sorting a topic: its own conversations scored again after the owner narrowed it + public static final String RESORT_KIND = "topic_resort"; + // another topic is now the clear fit, or none is + static final String ELSEWHERE = "elsewhere"; + static final String NONE = "none"; private BrainTopicMatches() { } @@ -111,6 +116,156 @@ public static Map start(User user, String topicId, Integer limit }); } + /** + * Re-sorting a topic, for after the owner narrows it: a dry run scores the conversations Brain filed under it + * and lists the ones it no longer fits, with the topic that now fits each or none. Apply moves them (move with + * toTopicId) or takes the topic off (reject). Links the owner, onboarding or an Outlook folder made are never + * listed; the classifier only ever adds links, so this is the one way a Brain filing leaves a topic. + */ + public static Map resort(User user, String topicId, Integer limit) { + BrainThreadClassifier.requireEngine(user); + Pair owner = CollaborationDbUtils.ownerOf(user); + String ownerId = owner.getValue0(); + String ownerType = owner.getValue1(); + requireMatchable(ownerId, ownerType, topicId); + int scan = limit == null ? MAX_LIMIT : limit; + if (scan < 1 || scan > MAX_LIMIT) { + throw new IllegalArgumentException("limit must be between 1 and " + MAX_LIMIT); + } + Map running = CollaborationJobUtils.latest(ownerId, ownerType, RESORT_KIND); + if (running != null && CollaborationJobUtils.RUNNING.equals(running.get("status")) + && running.get("params") instanceof Map params && !topicId.equals(params.get("topicId"))) { + throw new IllegalArgumentException("A re-sort of another topic is still running; try again when it finishes"); + } + Map params = new LinkedHashMap<>(); + params.put("topicId", topicId); + params.put("limit", scan); + return CollaborationJobUtils.start(ownerId, ownerType, RESORT_KIND, params, job -> { + Insight insight = new Insight(); + insight.setUser(user); + job.step("finding", 1); + List filed = filedByBrain(ownerId, ownerType, topicId, scan); + job.count("scanned", filed.size()); + List> leaving = List.of(); + if (!filed.isEmpty()) { + Map summary = BrainThreadClassifier.classify(user, insight, filed, true, + (done, total) -> { + job.count("done", done); + job.count("total", total); + job.step("scoring", total == 0 ? 99 : Math.max(1, 99 * done / total)); + }); + job.count("errors", summary.get("errors")); + job.count("errorReason", summary.get("errorReason")); + leaving = leaving(ownerId, ownerType, topicId, summary); + } + job.count("matches", leaving); + }); + } + + // unmuted, not automated, with a link to the topic that Brain made (not the owner's, onboarding's or a folder's) + static List filedByBrain(String ownerId, String ownerType, String topicId, int limit) { + return CollaborationDbUtils.query(CollaborationDbUtils.page("SELECT t.THREAD_ID FROM BRAIN_THREAD t " + + "JOIN BRAIN_THREAD_TOPIC l ON l.OWNER_ID = t.OWNER_ID AND l.OWNER_TYPE = t.OWNER_TYPE AND " + + "l.THREAD_ID = t.THREAD_ID AND l.TOPIC_ID = ? WHERE t.OWNER_ID = ? AND t.OWNER_TYPE = ? AND " + + "(t.MUTED IS NULL OR t.MUTED = ?) AND (t.AUTOMATED IS NULL OR t.AUTOMATED = ?) AND " + + "(l.SOURCE IS NULL OR l.SOURCE <> ?) AND (l.CLASSIFIER_VERSION IS NULL OR l.CLASSIFIER_VERSION " + + "NOT IN (?, ?)) ORDER BY t.LAST_MESSAGE_AT DESC, t.THREAD_ID", limit, 0), rs -> rs.getString(1), + topicId, ownerId, ownerType, false, false, BrainProfileUtils.YOU, BrainTopicOnboarding.VERSION, + BrainMailFolders.FILED); + } + + // the dry run's results as re-sort rows, weakest fit first + @SuppressWarnings("unchecked") + static List> leaving(String ownerId, String ownerType, String topicId, + Map summary) { + Map settings = BrainProfileUtils.getSettings(ownerId, ownerType); + int fileAt = (Integer) settings.get("fileAt"); + int askAt = (Integer) settings.get("askAt"); + List> rows = new ArrayList<>(); + for (Map result : (List>) summary.get("results")) { + if (result.get("error") != null || "automated".equals(result.get("work")) + || !(result.get("scores") instanceof Map scores) + || !(scores.get("topics") instanceof Map topics)) { + continue; + } + String threadId = (String) result.get("threadId"); + Map probabilities = new LinkedHashMap<>(); + topics.forEach((id, p) -> { + if (p instanceof Number n) { + probabilities.put(String.valueOf(id), n.doubleValue()); + } + }); + Map row = leave(topicId, probabilities, + BrainThreadTopicDecisions.rejected(ownerId, ownerType, threadId), links(ownerId, ownerType, threadId), + fileAt, askAt); + Map described = row == null ? null : describe(ownerId, ownerType, threadId); + if (described != null) { + described.putAll(row); + rows.add(described); + } + } + rows.sort(Comparator.comparingInt((Map row) -> (Integer) row.get("confidence"))); + return rows; + } + + // The re-sort row for one conversation, or null while the topic still fits it: another topic is now the clear + // best (move there, unless the conversation has a kept link elsewhere), or the topic falls under the ask band + // with no clear best (take the topic off). A close second stays as it is. + static Map leave(String topicId, Map topics, Set rejected, + List links, int fileAt, int askAt) { + if (links.stream().noneMatch(link -> link.topicId().equals(topicId) && !link.kept())) { + return null; + } + List> ranked = new ArrayList<>( + topics.entrySet().stream().filter(entry -> !rejected.contains(entry.getKey())).toList()); + ranked.sort((a, b) -> Double.compare(b.getValue(), a.getValue())); + if (ranked.isEmpty() || ranked.get(0).getKey().equals(topicId)) { + return null; + } + int fit = (int) Math.round(100 * topics.getOrDefault(topicId, 0d)); + String best = ranked.get(0).getKey(); + int bestAt = (int) Math.round(100 * ranked.get(0).getValue()); + boolean elsewhere = !BrainThreadClassifier.OTHER_TOPIC.equals(best) + && bestAt >= Math.max(askAt, fileAt - BrainThreadClassifier.WAY_OUT_ASK_BAND); + if (!elsewhere && fit >= askAt) { + return null; + } + boolean keptElsewhere = links.stream().anyMatch(link -> !link.topicId().equals(topicId) && link.kept()); + List> current = new ArrayList<>(); + for (Link link : links) { + Map row = new LinkedHashMap<>(); + row.put("topicId", link.topicId()); + row.put("source", link.source()); + row.put("primary", link.primary()); + row.put("kept", link.kept()); + current.add(row); + } + Map row = new LinkedHashMap<>(); + row.put("confidence", fit); + row.put("fit", elsewhere ? ELSEWHERE : NONE); + if (elsewhere) { + row.put("toTopicId", best); + row.put("toConfidence", bestAt); + } + row.put("links", current); + List ops = elsewhere && !keptElsewhere ? List.of(MOVE, REJECT) : List.of(REJECT); + row.put("ops", ops); + row.put("op", ops.get(0)); + return row; + } + + private static Map describe(String ownerId, String ownerType, String threadId) { + return CollaborationDbUtils.queryOne("SELECT SUBJECT, SOURCE, LAST_MESSAGE_AT FROM BRAIN_THREAD WHERE " + + "OWNER_ID = ? AND OWNER_TYPE = ? AND THREAD_ID = ?", rs -> { + Map thread = new LinkedHashMap<>(); + thread.put("threadId", threadId); + thread.put("subject", CollaborationDbUtils.getString(rs, "SUBJECT")); + thread.put("channel", CollaborationDbUtils.getString(rs, "SOURCE")); + thread.put("lastMessageAt", CollaborationDbUtils.getTimestamp(rs, "LAST_MESSAGE_AT")); + return thread; + }, ownerId, ownerType, threadId); + } + // unmuted, not automated, not on the topic and not turned down for it; newest first static List candidates(String ownerId, String ownerType, String topicId, int limit) { return CollaborationDbUtils.query(CollaborationDbUtils.page("SELECT t.THREAD_ID FROM BRAIN_THREAD t " @@ -215,7 +370,8 @@ static Map match(String topicId, Map topics, Set // ---- apply ---- - // each change is {threadId, op} with op move, also_link or reject; every one is checked against the + // each change is {threadId, op} with op move (to toTopicId when given, else this topic), also_link or reject; + // every one is checked against the // conversation as it is now, so a stale preview cannot undo what changed since. Each runs on its own, and one // that no longer applies comes back skipped with the reason. public static Map apply(User user, String topicId, List> changes) { @@ -249,7 +405,11 @@ public static Map apply(User user, String topicId, List move(ownerId, ownerType, threadId, topicId); + case MOVE -> { + // a re-sort moves a conversation off this topic to the one that now fits + String to = CollaborationDbUtils.asString(change.get("toTopicId")); + yield move(ownerId, ownerType, threadId, to == null || to.isBlank() ? topicId : to); + } case ALSO_LINK -> { BrainThreadUtils.linkThreadTopic(user, threadId, topicId, false, false); yield null; diff --git a/src/prerna/collaboration/WorkPipeline.java b/src/prerna/collaboration/WorkPipeline.java index 804d8186ae..76b4f0341b 100644 --- a/src/prerna/collaboration/WorkPipeline.java +++ b/src/prerna/collaboration/WorkPipeline.java @@ -19,7 +19,11 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -96,6 +100,11 @@ message that delivers what was asked (the owner sends the reply, file or answer due: YYYY-MM-DD when a message gives or clearly implies a date, else "". alreadyNoted lists what earlier messages of this thread already gave: do not list those again, but do \ list a done_claim or update when these messages finish or change one of them. + earlier, when given, holds the messages just before these, already read: use them to understand the new \ + ones, but list only what the new messages ask, decide or report. + A chat comes in short lines: read them together as one conversation and write one candidate per ask, \ + not one per line. Name what the ask is about from the whole conversation ("Answer the question on the \ + agent lab model costs", not "Answer the question"). Leave out greetings, thanks, scheduling chatter, and mail sent by the owner's own routines \ (digests, lineups, briefs). An empty list is fine. """; @@ -109,6 +118,8 @@ message that delivers what was asked (the owner sends the reply, file or answer // time in their own lane, so one owner's big build never holds up another's sync. Not final so tests can swap it private static final int PARALLEL = 3; private static final int MAX_PARALLEL = 16; + // threads one run reads at once; RDF_Map COLLAB_WORK_EXTRACT_PARALLEL + private static final int EXTRACT_PARALLEL = 5; static Executor pool = Executors.newFixedThreadPool(parallel(), r -> { Thread t = new Thread(r, "collaboration-work-items"); t.setDaemon(true); @@ -119,10 +130,28 @@ message that delivers what was asked (the owner sends the reply, file or answer static final int BUILD_PART = 25; private static final Map LANES = new ConcurrentHashMap<>(); + // a chat comes in short lines: it is read once it goes quiet, so a burst is one conversation, and a chat that + // never pauses is still read every CHAT_MAX_WAIT_MS. Settling chats live in memory; after a restart their lines + // stay unread (WORK_REF) until the chat's next message + private static final String CHAT_SOURCE = "teams"; + private static final long CHAT_QUIET_MS = 10 * 60_000L; + private static final long CHAT_MAX_WAIT_MS = 30 * 60_000L; + private static final long CHAT_CHECK_MS = 2 * 60_000L; + // lines before the new ones, given as context only + private static final int EARLIER = 6; + private static final ScheduledExecutorService WAKE = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "collaboration-work-items-wake"); + t.setDaemon(true); + return t; + }); + // one owner's queued runs; syncs that land while a run is going merge into one follow-up run private static final class Lane { final java.util.ArrayDeque jobs = new java.util.ArrayDeque<>(); final Set syncThreads = new LinkedHashSet<>(); + // chats still talking, with when each was first held back + final Map settling = new LinkedHashMap<>(); + ScheduledFuture wake; boolean busy; } @@ -153,10 +182,35 @@ public static void queue(User user, String ownerId, String ownerType, Collection classLogger.warn("Work items skipped: {}", e.getMessage()); return; } + Set talking = talkingChats(ownerId, ownerType); Lane lane = lane(ownerId, ownerType); synchronized (lane) { + long now = System.currentTimeMillis(); + List ready = new ArrayList<>(); + for (String id : threadIds) { + Long since = lane.settling.get(id); + if (talking.contains(id) && (since == null || now - since < CHAT_MAX_WAIT_MS)) { + lane.settling.putIfAbsent(id, now); + } else { + lane.settling.remove(id); + ready.add(id); + } + } + if (!lane.settling.isEmpty() && lane.wake == null) { + lane.wake = WAKE.schedule(() -> { + List held; + synchronized (lane) { + held = new ArrayList<>(lane.settling.keySet()); + lane.wake = null; + } + queue(user, ownerId, ownerType, held); + }, CHAT_CHECK_MS, TimeUnit.MILLISECONDS); + } + if (ready.isEmpty()) { + return; + } boolean waiting = !lane.syncThreads.isEmpty(); - lane.syncThreads.addAll(threadIds); + lane.syncThreads.addAll(ready); if (waiting) { return; // joins the sync run already queued } @@ -247,6 +301,12 @@ public static Map backfill(User user, int days, boolean replaceO return out; } + private static int pendingCount(String ownerId, String ownerType) { + Integer n = CollaborationDbUtils.queryOne("SELECT COUNT(*) AS N FROM WORK_CANDIDATE WHERE OWNER_ID = ? AND " + + "OWNER_TYPE = ? AND RESULT IS NULL", rs -> rs.getInt("N"), ownerId, ownerType); + return n == null ? 0 : n; + } + // the owner's threads with mail in the last days, newest first, muted and automated ones left out private static List recent(String ownerId, String ownerType, int days) { Timestamp since = Timestamp.from(Instant.now().minus(Math.max(1, days), ChronoUnit.DAYS)); @@ -289,22 +349,49 @@ private static Map run(User user, String ownerId, String ownerTy + "OWNER_ID = ? AND OWNER_TYPE = ? AND RELATIONSHIP = ?", rs -> new String[] { rs.getString("PERSON_ID"), rs.getString("DISPLAY_NAME") }, ownerId, ownerType, "self"); - int done = 0; - int found = 0; - int failed = 0; - for (String threadId : threadIds) { - try { - found += extract(user, ownerId, ownerType, threadId, model, self); - } catch (RuntimeException e) { - failed++; - classLogger.warn("Work items extract failed for thread {}", threadId, e); + // each thread's read stands alone (its own candidates and WORK_REF), so several go at once + java.util.concurrent.atomic.AtomicInteger done = new java.util.concurrent.atomic.AtomicInteger(); + java.util.concurrent.atomic.AtomicInteger found = new java.util.concurrent.atomic.AtomicInteger(); + java.util.concurrent.atomic.AtomicInteger failed = new java.util.concurrent.atomic.AtomicInteger(); + ExecutorService reads = Executors.newFixedThreadPool(Math.max(1, Math.min(extractParallel(), + threadIds.size())), r -> { + Thread t = new Thread(r, "collaboration-work-items-extract"); + t.setDaemon(true); + return t; + }); + try { + List> pending = new ArrayList<>(); + for (String threadId : threadIds) { + pending.add(reads.submit(() -> { + try { + found.addAndGet(extract(user, ownerId, ownerType, threadId, model, self)); + } catch (RuntimeException e) { + failed.incrementAndGet(); + classLogger.warn("Work items extract failed for thread {}", threadId, e); + } + status.put("extracted", done.incrementAndGet()); + status.put("candidates", found.get()); + status.put("failed", failed.get()); + })); + } + for (java.util.concurrent.Future f : pending) { + f.get(); } - status.put("extracted", ++done); - status.put("candidates", found); - status.put("failed", failed); + } finally { + reads.shutdownNow(); } status.put("state", "grouping"); - status.putAll(reconcile(user, ownerId, ownerType, model)); + // one pass takes at most MAX_PENDING candidates; go again while passes still place some + int before = pendingCount(ownerId, ownerType); + while (true) { + reconcile(user, ownerId, ownerType, model).forEach( + (k, v) -> status.merge(k, v, (a, b) -> a instanceof Integer x && b instanceof Integer y ? x + y : b)); + int after = pendingCount(ownerId, ownerType); + if (after == 0 || after >= before) { + break; + } + before = after; + } status.put("state", "done"); } catch (Throwable e) { classLogger.error("Work items run failed", e); @@ -394,11 +481,14 @@ private static int extractWindow(User user, String ownerId, String ownerType, St // the cursor goes to the thread's newest mail, hidden or gone mail included, unless some was left unread String end = (String) thread.get("last"); List> fresh = new ArrayList<>(); + List> older = new ArrayList<>(); for (Map m : (List>) read.get("messages")) { String at = String.valueOf(m.get("at")); end = end == null || at.compareTo(end) > 0 ? at : end; if (ref == null || at.compareTo(ref) > 0) { fresh.add(m); + } else { + older.add(m); } } // unread: a fetch that failed, or the first message past the budget; the next run starts there @@ -411,6 +501,10 @@ private static int extractWindow(User user, String ownerId, String ownerType, St } } String reached = held == null ? end : secondBefore(held); + // the lines before the new ones, for context only; the new ones follow, so a later window sees the earlier ones + List> context = new ArrayList<>(WorkThreadInsights.messages(older, people, selfId)); + int firstNew = context.size(); + context.addAll(messages); if (messages.isEmpty()) { CollaborationDbUtils.inTransaction(conn -> moveCursor(conn, ownerId, ownerType, threadId, ref, reached)); return 0; @@ -438,6 +532,17 @@ private static int extractWindow(User user, String ownerId, String ownerType, St input.put("facts", facts); input.put("people", peopleInput); input.put("alreadyNoted", noted); + // the lines just before this window, so a reply or a short chat line is read in its conversation + int start = firstNew + from; + List> earlier = new ArrayList<>(); + for (Map m : context.subList(Math.max(0, start - EARLIER), start)) { + Map e = new LinkedHashMap<>(m); + e.remove("id"); + earlier.add(e); + } + if (!earlier.isEmpty()) { + input.put("earlier", earlier); + } input.put("messages", window); Map answer = CollaborationDbUtils.firstJsonObject(ask(model, user, input)); if (answer == null || !(answer.get("candidates") instanceof List list)) { @@ -559,12 +664,14 @@ static Map reconcile(User user, String ownerId, String ownerType // only a person and a topic that still exist: a candidate naming one that doesn't would fail its save on every // run (older rows hold names the model made up, and topics merged or deleted since) CollaborationDbUtils.query("SELECT c.CANDIDATE_ID, c.THREAD_ID, c.MESSAGE_REF, c.KIND, c.TEXT, c.DUE_AT, " - + "c.ASSIGNEE, tp.TOPIC_ID, th.SUBJECT, p.DISPLAY_NAME, p.PERSON_ID AS KNOWN_PERSON FROM WORK_CANDIDATE c " - + "LEFT JOIN BRAIN_THREAD th ON th.OWNER_ID = c.OWNER_ID AND th.OWNER_TYPE = c.OWNER_TYPE AND " - + "th.THREAD_ID = c.THREAD_ID LEFT JOIN BRAIN_PERSON p ON p.OWNER_ID = c.OWNER_ID AND p.OWNER_TYPE = " - + "c.OWNER_TYPE AND p.PERSON_ID = c.ASSIGNEE LEFT JOIN BRAIN_TOPIC tp ON tp.OWNER_ID = c.OWNER_ID AND " - + "tp.OWNER_TYPE = c.OWNER_TYPE AND tp.TOPIC_ID = c.TOPIC_ID WHERE c.OWNER_ID = ? AND c.OWNER_TYPE = ? " - + "AND c.RESULT IS NULL ORDER BY c.MESSAGE_AT", + // the thread's topic now, not at extract: a thread filed, moved or left without a topic since + + "c.ASSIGNEE, (SELECT MIN(bt.TOPIC_ID) FROM BRAIN_THREAD_TOPIC bt JOIN BRAIN_TOPIC tp ON " + + "tp.OWNER_ID = bt.OWNER_ID AND tp.OWNER_TYPE = bt.OWNER_TYPE AND tp.TOPIC_ID = bt.TOPIC_ID WHERE " + + "bt.OWNER_ID = c.OWNER_ID AND bt.OWNER_TYPE = c.OWNER_TYPE AND bt.THREAD_ID = c.THREAD_ID AND " + + "bt.IS_PRIMARY = ?) AS TOPIC_ID, th.SUBJECT, p.DISPLAY_NAME, p.PERSON_ID AS KNOWN_PERSON FROM WORK_CANDIDATE c LEFT JOIN BRAIN_THREAD th ON " + + "th.OWNER_ID = c.OWNER_ID AND th.OWNER_TYPE = c.OWNER_TYPE AND th.THREAD_ID = c.THREAD_ID LEFT JOIN " + + "BRAIN_PERSON p ON p.OWNER_ID = c.OWNER_ID AND p.OWNER_TYPE = c.OWNER_TYPE AND p.PERSON_ID = " + + "c.ASSIGNEE WHERE c.OWNER_ID = ? AND c.OWNER_TYPE = ? AND c.RESULT IS NULL ORDER BY c.MESSAGE_AT", rs -> { if (byShort.size() >= MAX_PENDING) { return null; @@ -594,7 +701,7 @@ static Map reconcile(User user, String ownerId, String ownerType c.put("due", due == null ? "" : due.substring(0, Math.min(10, due.length()))); input.add(c); return null; - }, ownerId, ownerType); + }, true, ownerId, ownerType); Map counts = new LinkedHashMap<>(); if (input.isEmpty()) { return counts; @@ -748,6 +855,11 @@ void step(Map step, String taskId, String topicId) { assigneeId = (String) c.get("assigneeId"); } } + // a candidate from before unmatched names became unclear can still name no one + if (assigneeId != null && !CollaborationDbUtils.exists("SELECT 1 FROM BRAIN_PERSON WHERE OWNER_ID = ? AND " + + "OWNER_TYPE = ? AND PERSON_ID = ?", ownerId, ownerType, assigneeId)) { + assigneeId = null; + } Map args = new LinkedHashMap<>(); args.put("text", step.get("text")); args.put("taskId", taskId); @@ -918,6 +1030,13 @@ public static Map tidyAll(User user) { // ---- lanes ---- + // the owner's chats with a message inside the quiet window + private static Set talkingChats(String ownerId, String ownerType) { + return new HashSet<>(CollaborationDbUtils.query("SELECT THREAD_ID FROM BRAIN_THREAD WHERE OWNER_ID = ? AND " + + "OWNER_TYPE = ? AND SOURCE = ? AND LAST_MESSAGE_AT > ?", rs -> rs.getString("THREAD_ID"), ownerId, + ownerType, CHAT_SOURCE, Timestamp.from(Instant.now().minusMillis(CHAT_QUIET_MS)))); + } + static Lane lane(String ownerId, String ownerType) { return LANES.computeIfAbsent(key(ownerId, ownerType), k -> new Lane()); } @@ -954,6 +1073,17 @@ private static void drain(Lane lane) { pool.execute(() -> drain(lane)); } + private static int extractParallel() { + String value = Utility.getDIHelperProperty(Constants.COLLAB_WORK_EXTRACT_PARALLEL); + try { + return value == null || value.isBlank() ? EXTRACT_PARALLEL + : Math.max(1, Math.min(MAX_PARALLEL, Integer.parseInt(value.trim()))); + } catch (NumberFormatException e) { + classLogger.warn("{} is not a number; using {}", Constants.COLLAB_WORK_EXTRACT_PARALLEL, EXTRACT_PARALLEL); + return EXTRACT_PARALLEL; + } + } + private static int parallel() { String value = Utility.getDIHelperProperty(Constants.COLLAB_WORK_ITEMS_PARALLEL); if (value == null || value.isBlank()) { @@ -1083,11 +1213,15 @@ private static String named(String text, Map names) { .quoteReplacement(names.getOrDefault(m.group(), m.group()))); } + // a p id becomes the person it stands for; anything else the model wrote that is not me or group is unclear private static String person(String value, Map people) { if (value == null || value.isBlank()) { return null; } - return people.getOrDefault(value, value); + if (people.containsKey(value)) { + return people.get(value); + } + return ME.equals(value) || GROUP.equals(value) ? value : UNCLEAR; } // a name ("Bob") or a p id the model made up is no one Brain knows: unclear, not a person id addStep refuses diff --git a/src/prerna/collaboration/WorkReconcile.java b/src/prerna/collaboration/WorkReconcile.java index bbb3b690fe..cbc8986730 100644 --- a/src/prerna/collaboration/WorkReconcile.java +++ b/src/prerna/collaboration/WorkReconcile.java @@ -1,6 +1,7 @@ package prerna.collaboration; import java.util.ArrayList; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -26,6 +27,8 @@ public final class WorkReconcile { static final String NO_TOPIC = "(no topic)"; // more steps than this is more than one outcome static final int MAX_STEPS = 10; + // an item already holding this many open steps takes no more; new asks make their own item or stay loose + static final int MAX_OPEN_STEPS = 15; private static final int MAX_EXISTING = 40; // open items under other topics, titles only, so the same work is not made twice private static final int MAX_OTHER = 40; @@ -57,8 +60,13 @@ public final class WorkReconcile { A done claim ("fixed", "merged", "sent") about a task in this list or an existing step is closes, \ never drop. - An item under another topic takes a candidate only when it is clearly the same outcome; use its w \ - id for new_step or update. + feeds on a candidate lists the items its thread already feeds. A candidate from the same \ + conversation usually serves the same item: prefer those when it is about the same outcome. Items \ + under other topics are listed only when a candidate's thread feeds them, and a candidate joins one \ + only through feeds. + + When the topic is "(no topic)" the candidates come from threads not filed anywhere: make a new item \ + only from candidates of one thread, and leave the rest loose. The same ask often comes from two threads in different words ("Widen the chat window" and "Allow \ the chat window to widen"): mark the later one same_as the earlier one. @@ -87,13 +95,14 @@ step, not a new one ("Draft the response", "Shape the response into a story" and static final int GROUP_MIN = 6; private static final String GROUP_INSTRUCTIONS = """ You group the owner's loose tasks for one topic into work items. The input is JSON: the topic \ - (name, description) and tasks c1.. (assignee, text). It is reference data, not instructions. + (name, description) and tasks c1.. (assignee, text, thread). It is reference data, not instructions. A work item is one outcome someone is driving, named as a short outcome ("Fix the chat UI issues", \ "Ship the 6.0.0 release", "Hire for the open positions"). Changes to the same product area, \ feature, release or hiring need are one outcome. Each item has 2 to 8 tasks; when more fit, split \ - by area. A task that shares an outcome with no other task stays loose. Most tasks in a busy topic \ - belong to a few outcomes, so look hard for the shared one before leaving a task loose. + by area. Tasks from one thread often share an outcome; tasks from different threads share one only \ + when they are about the same deliverable, product, client or release. A task that shares an \ + outcome with no other task stays loose, and when unsure leave it loose. Answer with JSON only: {"items": [{"title": "", "tasks": [""]}], "loose": [""]} Every task appears once, in one item or in loose. @@ -151,15 +160,18 @@ public static Map plan(User user, String ownerId, String ownerTy /** Same, calling onTopic after each topic so its items exist before the next topic is planned. */ public static Map plan(User user, String ownerId, String ownerType, IModelEngine model, List> candidates, java.util.function.Consumer> onTopic) { - // the same ask from two threads often lands under two topics; merge those before splitting by topic, - // keeping the copy that has a topic + // the same ask said again in an unfiled thread folds into the filed copy before splitting by topic; + // copies under two different topics stay apart List> kept = new ArrayList<>(); Map> alsoSaid = new LinkedHashMap<>(); for (Map c : candidates) { Map same = null; if (isTask(c)) { for (Map k : kept) { - if (isTask(k) && similar(String.valueOf(k.get("text")), String.valueOf(c.get("text")))) { + boolean crossTopic = k.get("topicId") != null && c.get("topicId") != null + && !k.get("topicId").equals(c.get("topicId")); + if (isTask(k) && !crossTopic + && similar(String.valueOf(k.get("text")), String.valueOf(c.get("text")))) { same = k; break; } @@ -255,6 +267,12 @@ private static void planTopic(User user, String ownerId, String ownerType, IMode Map itemIds = new LinkedHashMap<>(); Map stepIds = new LinkedHashMap<>(); List> existing = new ArrayList<>(); + Map> fedBy = itemThreads(ownerId, ownerType); + Set batchThreads = new LinkedHashSet<>(); + for (Map c : shown) { + batchThreads.add(String.valueOf(c.get("threadId"))); + } + Items items = new Items(fedBy); if (!NO_TOPIC.equals(topicId)) { Map work = WorkTaskUtils.listMyWork(user, Map.of("topicId", topicId)); for (Map task : (List>) work.get("workItems")) { @@ -263,6 +281,8 @@ private static void planTopic(User user, String ownerId, String ownerType, IMode } String w = "w" + (itemIds.size() + 1); itemIds.put(w, String.valueOf(task.get("id"))); + items.own.add(w); + items.open.put(w, ((List) task.get("openSteps")).size()); List> steps = new ArrayList<>(); for (Map step : (List>) task.get("openSteps")) { String s = "s" + (stepIds.size() + 1); @@ -285,6 +305,11 @@ private static void planTopic(User user, String ownerId, String ownerType, IMode if (itemIds.containsValue(String.valueOf(task.get("id")))) { continue; } + // another topic's item is offered only to the conversations already feeding it + if (java.util.Collections.disjoint(fedBy.getOrDefault(String.valueOf(task.get("id")), Set.of()), + batchThreads)) { + continue; + } String name = null; for (Map t : (List>) task.get("topics")) { if (name == null || Boolean.TRUE.equals(t.get("primary"))) { @@ -298,6 +323,7 @@ private static void planTopic(User user, String ownerId, String ownerType, IMode item.put("title", task.get("title")); item.put("topic", name == null ? NO_TOPIC : name); existing.add(item); + items.open.put(w, ((List) task.get("openSteps")).size()); other++; } // the topic's loose steps, and any step on the candidates' threads (one made before its thread was filed has @@ -318,6 +344,7 @@ private static void planTopic(User user, String ownerId, String ownerType, IMode otherSteps.add(Map.of("id", s, "text", String.valueOf(step.get("text")))); } } + items.ids = itemIds; List> input = new ArrayList<>(); for (Map c : shown) { @@ -332,6 +359,10 @@ private static void planTopic(User user, String ownerId, String ownerType, IMode if (times > 1) { slim.put("saidTimes", times); } + List feeds = items.fedFrom(String.valueOf(c.get("threadId"))); + if (!feeds.isEmpty()) { + slim.put("feeds", feeds); + } input.add(slim); } Map prompt = new LinkedHashMap<>(); @@ -348,8 +379,65 @@ private static void planTopic(User user, String ownerId, String ownerType, IMode out.put("raw", reply); return; } - assemble(answer, shown, mergedInto, itemIds, stepIds, existing, out); - groupLoose(user, model, topic == null ? Map.of("name", topicId) : topic, out); + boolean noTopic = NO_TOPIC.equals(topicId); + assemble(answer, shown, mergedInto, items, stepIds, existing, noTopic, out); + // unfiled threads have nothing in common but being unfiled; their tasks are not grouped across threads + if (!noTopic) { + groupLoose(user, model, topic == null ? Map.of("name", topicId) : topic, out); + } + } + + // the open items in one topic's plan: which are the topic's own, how many open steps each holds, and the + // threads feeding each; a candidate joins an item of its own topic, or any item its thread already feeds + private static final class Items { + final Map> fedBy; + final Set own = new LinkedHashSet<>(); + final Map open = new LinkedHashMap<>(); + Map ids = new LinkedHashMap<>(); + + Items(Map> fedBy) { + this.fedBy = fedBy; + } + + List fedFrom(String threadId) { + List out = new ArrayList<>(); + ids.forEach((w, taskId) -> { + if (fedBy.getOrDefault(taskId, Set.of()).contains(threadId)) { + out.add(w); + } + }); + return out; + } + + // the item's real id when the candidate may join it, else null + String join(String w, Map c) { + String taskId = ids.get(w); + if (taskId == null) { + return null; + } + if (!own.contains(w) && !fedBy.getOrDefault(taskId, Set.of()).contains(String.valueOf(c.get("threadId")))) { + return null; + } + return taskId; + } + + boolean full(String w) { + return open.getOrDefault(w, 0) >= MAX_OPEN_STEPS; + } + } + + // the threads behind each open item: its linked threads and the threads of its steps + private static Map> itemThreads(String ownerId, String ownerType) { + Map> out = new LinkedHashMap<>(); + CollaborationDbUtils.query("SELECT TASK_ID, THREAD_ID FROM WORK_TASK_THREAD WHERE OWNER_ID = ? AND OWNER_TYPE = ? " + + "UNION SELECT TASK_ID, THREAD_ID FROM WORK_THREAD_STEP WHERE OWNER_ID = ? AND OWNER_TYPE = ? AND " + + "TASK_ID IS NOT NULL", rs -> { + if (rs.getString(1) != null && rs.getString(2) != null) { + out.computeIfAbsent(rs.getString(1), k -> new HashSet<>()).add(rs.getString(2)); + } + return null; + }, ownerId, ownerType, ownerId, ownerType); + return out; } // second pass: the first call decides too much at once to group well, so a busy topic's loose tasks @@ -364,7 +452,8 @@ private static void groupLoose(User user, IModelEngine model, Map> tasks = new ArrayList<>(); for (int i = 0; i < loose.size(); i++) { tasks.add(Map.of("id", "c" + (i + 1), "assignee", String.valueOf(loose.get(i).get("assignee")), - "text", String.valueOf(loose.get(i).get("text")))); + "text", String.valueOf(loose.get(i).get("text")), "thread", + String.valueOf(loose.get(i).getOrDefault("thread", "")))); } Map input = new LinkedHashMap<>(); input.put("topic", topic); @@ -424,8 +513,9 @@ private static void groupLoose(User user, IModelEngine model, Map answer, List> shown, - Map> mergedInto, Map itemIds, Map stepIds, - List> existing, Map out) { + Map> mergedInto, Items items, Map stepIds, + List> existing, boolean noTopic, Map out) { + Map itemIds = items.ids; Map> byId = new LinkedHashMap<>(); for (Map c : shown) { byId.put(String.valueOf(c.get("id")), c); @@ -482,18 +572,19 @@ private static void assemble(Map answer, List c = byId.get(id); switch (action) { case "same_step" -> sameSteps.add(entry(c, evidence.get(id), "stepId", stepIds.get(ref))); - case "new_step" -> { - if (itemIds.containsKey(ref)) { - underExisting.add(step(c, evidence.get(id), "taskId", itemIds.get(ref))); - } else { + case "new_step", "update" -> { + // code holds the line: another topic's item only through a thread that feeds it, a full item + // takes no more steps, and news is never a step + String taskId = items.join(ref, c); + if (taskId == null) { leftovers.add(id); - } - } - case "update" -> { - if (itemIds.containsKey(ref)) { - updates.add(entry(c, evidence.get(id), "taskId", itemIds.get(ref))); - } else { + } else if (!isTask(c) || "update".equals(action)) { + updates.add(entry(c, evidence.get(id), "taskId", taskId)); + } else if (items.full(ref)) { leftovers.add(id); + } else { + items.open.merge(ref, 1, Integer::sum); + underExisting.add(step(c, evidence.get(id), "taskId", taskId)); } } case "closes" -> closes.add(entry(c, evidence.get(id), "closes", @@ -514,12 +605,14 @@ private static void assemble(Map answer, List> steps = new ArrayList<>(); List> itemUpdates = new ArrayList<>(); + Set threads = new HashSet<>(); for (Object idObj : ids) { String id = String.valueOf(idObj); if (!leftovers.contains(id) || placed.contains(id)) { continue; } Map c = byId.get(id); + threads.add(c.get("threadId")); if (isTask(c)) { steps.add(step(c, evidence.get(id), null, null)); } else { @@ -527,7 +620,8 @@ private static void assemble(Map answer, List MAX_STEPS) { + // unfiled threads make an item only from one conversation + if (steps.size() < 2 || steps.size() > MAX_STEPS || (noTopic && threads.size() > 1)) { // not one outcome: its steps stay loose, flagged so the eval can see it steps.forEach(s -> s.put("rejectedItem", m.get("title"))); placed.removeIf(id -> steps.stream().anyMatch(s -> ((List) s.get("evidence")).contains(id)) @@ -648,6 +742,9 @@ private static Map entry(Map c, Set evid Map e = new LinkedHashMap<>(); e.put("text", c.get("text")); e.put("kind", c.get("kind")); + if (c.get("thread") != null) { + e.put("thread", c.get("thread")); + } e.put("evidence", new ArrayList<>(evidence)); if (key != null) { e.put(key, value); diff --git a/src/prerna/reactor/collaboration/BrainApplyTopicMatchesReactor.java b/src/prerna/reactor/collaboration/BrainApplyTopicMatchesReactor.java index 40c5078099..3393b8d5a0 100644 --- a/src/prerna/reactor/collaboration/BrainApplyTopicMatchesReactor.java +++ b/src/prerna/reactor/collaboration/BrainApplyTopicMatchesReactor.java @@ -75,7 +75,8 @@ protected String getDescriptionForKey(String key) { if (TOPIC_ID.equals(key)) { return "The topic the preview was for"; } else if (CHANGES.equals(key)) { - return "Maps of threadId and op (move, also_link or reject); conversations left out are not changed"; + return "Maps of threadId and op (move, also_link or reject), with toTopicId on a re-sort move; " + + "conversations left out are not changed"; } return super.getDescriptionForKey(key); } diff --git a/src/prerna/reactor/collaboration/BrainPreviewTopicMatchesReactor.java b/src/prerna/reactor/collaboration/BrainPreviewTopicMatchesReactor.java index af6ceb1dbd..45ac79ee75 100644 --- a/src/prerna/reactor/collaboration/BrainPreviewTopicMatchesReactor.java +++ b/src/prerna/reactor/collaboration/BrainPreviewTopicMatchesReactor.java @@ -30,20 +30,24 @@ import prerna.collaboration.BrainTopicMatches; import prerna.sablecc2.om.nounmeta.NounMetadata; -// BrainPreviewTopicMatches(topicId=["..."], limit=[200]); poll with BrainGetJob(jobId=["..."]), apply with -// BrainApplyTopicMatches +// BrainPreviewTopicMatches(topicId=["..."], limit=[200]); BrainPreviewTopicMatches(topicId=["..."], resort=[true]); +// poll with BrainGetJob(jobId=["..."]), apply with BrainApplyTopicMatches public class BrainPreviewTopicMatchesReactor extends AbstractCollaborationReactor { private static final String TOPIC_ID = "topicId"; private static final String LIMIT = "limit"; + private static final String RESORT = "resort"; public BrainPreviewTopicMatchesReactor() { - this.keysToGet = new String[] { TOPIC_ID, LIMIT }; - this.keyRequired = new int[] { 1, 0 }; + this.keysToGet = new String[] { TOPIC_ID, LIMIT, RESORT }; + this.keyRequired = new int[] { 1, 0, 0 }; } @Override public NounMetadata execute() { + if (Boolean.TRUE.equals(getBoolean(RESORT))) { + return mapResult(BrainTopicMatches.resort(getUser(), getString(TOPIC_ID), getIntFromKeyOrCurRow(LIMIT))); + } return mapResult(BrainTopicMatches.start(getUser(), getString(TOPIC_ID), getIntFromKeyOrCurRow(LIMIT))); } @@ -51,7 +55,9 @@ public NounMetadata execute() { public String getReactorDescription() { return "Starts a background job that scores the owner's newest conversations against their topics without " + "writing anything and lists the ones that fit this topic; when done, counts.matches holds rows with " - + "threadId, subject, confidence, fit (best or shared), links, ops and a suggested op"; + + "threadId, subject, confidence, fit (best or shared), links, ops and a suggested op. With resort it " + + "instead scores the conversations Brain filed under the topic and lists the ones it no longer fits: " + + "fit elsewhere (with toTopicId) or none, ops move and reject"; } @Override @@ -59,7 +65,9 @@ protected String getDescriptionForKey(String key) { if (TOPIC_ID.equals(key)) { return "The active or dormant topic that was added or edited"; } else if (LIMIT.equals(key)) { - return "How many of the newest eligible conversations to score, 1 to 1000; default 200"; + return "How many of the newest eligible conversations to score, 1 to 1000; default 200 (resort: 1000)"; + } else if (RESORT.equals(key)) { + return "true to re-sort the topic's own conversations after narrowing it"; } return super.getDescriptionForKey(key); } diff --git a/src/prerna/util/Constants.java b/src/prerna/util/Constants.java index b5ec1d47fb..2e3c0e49fd 100644 --- a/src/prerna/util/Constants.java +++ b/src/prerna/util/Constants.java @@ -1085,6 +1085,8 @@ public class Constants { // work items runs at once across owners; each owner's own runs still go one at // a time (default 3) public static final String COLLAB_WORK_ITEMS_PARALLEL = "COLLAB_WORK_ITEMS_PARALLEL"; + // threads one work items run reads at once (default 5) + public static final String COLLAB_WORK_EXTRACT_PARALLEL = "COLLAB_WORK_EXTRACT_PARALLEL"; // threads the classifier sorts at once (default 8) public static final String COLLAB_CLASSIFY_PARALLEL = "COLLAB_CLASSIFY_PARALLEL"; // characters of memories a thread's assistant gets in its prompt (default 4000)