From fa2cf6f1dc559d984a068d3a33f722b9fffbb155 Mon Sep 17 00:00:00 2001 From: Kunal Patel Date: Thu, 5 Nov 2026 19:46:03 -0500 Subject: [PATCH 1/5] fix(collaboration): read a chat as one conversation when building work items A Teams chat is extracted once it has been quiet for 10 minutes (at most 30 minutes of waiting), so a burst of short lines becomes one set of candidates instead of one per sync. Each extract window also gets the six messages before it as context, and the prompt asks for one candidate per ask, named from the whole conversation. --- src/prerna/collaboration/WorkPipeline.java | 76 +++++++++++++++++++++- 1 file changed, 74 insertions(+), 2 deletions(-) diff --git a/src/prerna/collaboration/WorkPipeline.java b/src/prerna/collaboration/WorkPipeline.java index b8ea43870b..a533f92ea5 100644 --- a/src/prerna/collaboration/WorkPipeline.java +++ b/src/prerna/collaboration/WorkPipeline.java @@ -15,6 +15,9 @@ import java.util.concurrent.ConcurrentHashMap; 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; @@ -80,6 +83,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. """; @@ -100,10 +108,28 @@ message that delivers what was asked (the owner sends the reply, file or answer }); 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; } @@ -126,10 +152,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 } @@ -304,8 +355,11 @@ static int extract(User user, String ownerId, String ownerType, String threadId, people, selfId); String ref = (String) thread.get("ref"); List> messages = new ArrayList<>(); - for (Map m : all) { + int firstNew = all.size(); + for (int i = 0; i < all.size(); i++) { + Map m = all.get(i); if (ref == null || String.valueOf(m.get("at")).compareTo(ref) > 0) { + firstNew = Math.min(firstNew, i); messages.add(m); } } @@ -336,6 +390,17 @@ static int extract(User user, String ownerId, String ownerType, String threadId, 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 : all.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)) { @@ -738,6 +803,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)))); + } + private static Lane lane(String ownerId, String ownerType) { return LANES.computeIfAbsent(key(ownerId, ownerType), k -> new Lane()); } From 84917b83a5c1774053e88f2ff18108896768a00c Mon Sep 17 00:00:00 2001 From: Kunal Patel Date: Thu, 5 Nov 2026 19:57:09 -0500 Subject: [PATCH 2/5] fix(collaboration): keep work items to their topic and their conversations - Grouping reads each thread's current primary topic, not the one stored at extraction. - An item under another topic is offered, and can take a candidate, only through a thread that already feeds it. Each candidate lists the items its thread feeds. - An item holding 15 open steps takes no more. - News that comes back as new_step becomes an item update, never a step. - Unfiled threads make items only from one conversation and skip the second grouping pass. - The second pass sees each task's thread and no longer pushes to group. - Near-identical asks under two different topics are no longer folded together. --- src/prerna/collaboration/WorkPipeline.java | 8 +- src/prerna/collaboration/WorkReconcile.java | 145 ++++++++++++++++---- 2 files changed, 127 insertions(+), 26 deletions(-) diff --git a/src/prerna/collaboration/WorkPipeline.java b/src/prerna/collaboration/WorkPipeline.java index a533f92ea5..59af8b75f1 100644 --- a/src/prerna/collaboration/WorkPipeline.java +++ b/src/prerna/collaboration/WorkPipeline.java @@ -471,7 +471,11 @@ static Map reconcile(User user, String ownerId, String ownerType Map> byShort = new LinkedHashMap<>(); List> input = new ArrayList<>(); CollaborationDbUtils.query("SELECT c.CANDIDATE_ID, c.THREAD_ID, c.MESSAGE_REF, c.KIND, c.TEXT, c.DUE_AT, " - + "c.ASSIGNEE, c.TOPIC_ID, th.SUBJECT, p.DISPLAY_NAME FROM WORK_CANDIDATE c LEFT JOIN BRAIN_THREAD th ON " + // 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 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", @@ -504,7 +508,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; diff --git a/src/prerna/collaboration/WorkReconcile.java b/src/prerna/collaboration/WorkReconcile.java index 7faeb9a40c..3659874685 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; @@ -53,8 +56,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. @@ -83,13 +91,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. @@ -147,15 +156,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; } @@ -251,6 +263,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")) { @@ -259,6 +277,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); @@ -281,6 +301,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"))) { @@ -294,8 +319,10 @@ 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++; } + items.ids = itemIds; List> input = new ArrayList<>(); for (Map c : shown) { @@ -310,6 +337,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<>(); @@ -324,8 +355,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 @@ -340,7 +428,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); @@ -400,8 +489,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); @@ -458,18 +548,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", @@ -490,12 +581,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 { @@ -503,7 +596,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)) @@ -638,6 +732,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); From e91b461d4fa4e08e0c759b74a9e43d8bc86162f4 Mon Sep 17 00:00:00 2001 From: Kunal Patel Date: Thu, 5 Nov 2026 20:06:20 -0500 Subject: [PATCH 3/5] perf(collaboration): read five threads at once when building work items Each thread's extract stands alone, so one run reads several in parallel. RDF_Map COLLAB_WORK_EXTRACT_PARALLEL sets how many (default 5). --- src/prerna/collaboration/WorkPipeline.java | 54 +++++++++++++++++----- src/prerna/util/Constants.java | 2 + 2 files changed, 44 insertions(+), 12 deletions(-) diff --git a/src/prerna/collaboration/WorkPipeline.java b/src/prerna/collaboration/WorkPipeline.java index 59af8b75f1..ffa34d3c25 100644 --- a/src/prerna/collaboration/WorkPipeline.java +++ b/src/prerna/collaboration/WorkPipeline.java @@ -101,6 +101,8 @@ not one per line. Name what the ask is about from the whole conversation ("Answe // time in their own lane, so one owner's big build never holds up another's sync 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; private static final ExecutorService POOL = Executors.newFixedThreadPool(parallel(), r -> { Thread t = new Thread(r, "collaboration-work-items"); t.setDaemon(true); @@ -264,19 +266,36 @@ private static void run(User user, String ownerId, String ownerType, String engi + "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)); @@ -848,6 +867,17 @@ private static void drain(Lane 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()) { 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) From bc4a37b15dfe6a978982130350126b8e7cdedba2 Mon Sep 17 00:00:00 2001 From: Kunal Patel Date: Thu, 5 Nov 2026 20:23:46 -0500 Subject: [PATCH 4/5] fix(collaboration): one bad assignee no longer aborts a topic's work items An assignee the model names outside the thread's people is stored as unclear, and a step whose assignee is not a known person is saved without one. Grouping repeats until no candidates are pending, so one build places more than 400. --- src/prerna/collaboration/WorkPipeline.java | 29 ++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/src/prerna/collaboration/WorkPipeline.java b/src/prerna/collaboration/WorkPipeline.java index ffa34d3c25..15256a2313 100644 --- a/src/prerna/collaboration/WorkPipeline.java +++ b/src/prerna/collaboration/WorkPipeline.java @@ -234,6 +234,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; + } + /** Where the owner's last build stands. */ public static Map status(User user) { Pair owner = CollaborationDbUtils.ownerOf(user); @@ -298,7 +304,17 @@ private static void run(User user, String ownerId, String ownerType, String engi 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); @@ -668,6 +684,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); @@ -950,11 +971,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; } private static boolean isPerson(String assignee) { From ac419dcb37b8308600ee0c8380b0f12687db464d Mon Sep 17 00:00:00 2001 From: Kunal Patel Date: Thu, 5 Nov 2026 21:03:08 -0500 Subject: [PATCH 5/5] feat(collaboration): re-sort the conversations Brain filed under a topic BrainPreviewTopicMatches(resort=[true]) scores the topic's Brain-filed conversations again and lists the ones it no longer fits, with the topic that fits now or none. BrainApplyTopicMatches moves them with toTopicId or takes the topic off with reject. The classifier only adds links, so this is how a Brain filing leaves a topic once the owner narrows it. --- .../collaboration/BrainTopicMatches.java | 164 +++++++++++++++++- .../BrainApplyTopicMatchesReactor.java | 3 +- .../BrainPreviewTopicMatchesReactor.java | 20 ++- 3 files changed, 178 insertions(+), 9 deletions(-) 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/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); }