Skip to content

Commit 52ae515

Browse files
danjadanja
committed
consensus handling
1 parent 0eb6eca commit 52ae515

8 files changed

Lines changed: 321 additions & 19 deletions

File tree

‎src/agents/core/coordinator-command-parser.js‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ export function coordinatorCommandParser(text) {
2828
const content = trimmed.replace(/^(q|question|issue)\s*:/i, "").trim();
2929
if (content) {
3030
return {
31-
command: "mfr-debate",
31+
command: "mfr-consensus",
3232
content
3333
};
3434
}

‎src/agents/providers/coordinator-provider.js‎

Lines changed: 278 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ export class CoordinatorProvider extends BaseProvider {
2727
agentRegistry = new Map(),
2828
negotiator = null,
2929
primaryRoomJid = null,
30+
logRoomJid = process.env.LOG_ROOM_JID || "log@conference.tensegrity.it",
3031
enableDebate = false,
3132
debateTimeoutMs = null,
3233
contributionTimeoutMs = null,
@@ -41,6 +42,7 @@ export class CoordinatorProvider extends BaseProvider {
4142
this.agentRegistry = agentRegistry;
4243
this.negotiator = negotiator;
4344
this.primaryRoomJid = primaryRoomJid;
45+
this.logRoomJid = logRoomJid;
4446
this.enableDebate = enableDebate;
4547
this.debateTimeoutMs = debateTimeoutMs;
4648
this.contributionTimeoutMs = contributionTimeoutMs;
@@ -79,6 +81,9 @@ export class CoordinatorProvider extends BaseProvider {
7981
case "start":
8082
return await this.startMfrSession(content, metadata, reply);
8183

84+
case "mfr-consensus":
85+
return await this.startConsensusSession(content, metadata, reply);
86+
8287
case "mfr-debate":
8388
case "debate":
8489
if (!this.enableDebate) {
@@ -111,6 +116,12 @@ export class CoordinatorProvider extends BaseProvider {
111116
return this.getHelpMessage();
112117

113118
default:
119+
if (command === "chat") {
120+
const consensusSessionId = this.getActiveConsensusSessionId();
121+
if (consensusSessionId) {
122+
this.recordConsensusEntry(consensusSessionId, rawMessage, metadata);
123+
}
124+
}
114125
if (this.enableDebate) {
115126
const consensusResult = await this.handleDebateConsensus(
116127
content || rawMessage || "",
@@ -282,6 +293,83 @@ export class CoordinatorProvider extends BaseProvider {
282293
: `Debate started: ${sessionId}. Window: ${debateSeconds}s.`;
283294
}
284295

296+
/**
297+
* Start consensus session for community answers
298+
* @param {string} problemDescription - Problem description
299+
* @param {Object} metadata - Message metadata
300+
* @param {Function} reply - Reply function
301+
* @returns {Promise<string>} Response message
302+
*/
303+
async startConsensusSession(problemDescription, metadata, reply) {
304+
const sessionId = randomUUID();
305+
const verbose = isVerboseRequest(problemDescription);
306+
const quiet = isQuietRequest(problemDescription);
307+
308+
this.logger.info?.(
309+
`[CoordinatorProvider] Starting consensus session: ${sessionId}`
310+
);
311+
312+
const state = new MfrProtocolState(sessionId, {
313+
logger: this.logger,
314+
initialPhase: MFR_PHASES.PROBLEM_INTERPRETATION
315+
});
316+
this.activeSessions.set(sessionId, state);
317+
318+
state.transition(MFR_PHASES.CONSENSUS_DISCOVERY, {
319+
problemDescription,
320+
startedBy: metadata?.sender,
321+
verbose,
322+
quiet
323+
});
324+
325+
this.activeDebates.set(sessionId, {
326+
mode: "consensus",
327+
problemDescription,
328+
startTime: Date.now(),
329+
positions: [],
330+
rawMessages: [],
331+
consensusReached: false,
332+
selectedAgents: [],
333+
verbose,
334+
quiet
335+
});
336+
337+
const agentMentions = Array.from(this.agentRegistry.values() || [])
338+
.map((entry) => entry?.nickname)
339+
.filter(Boolean)
340+
.filter((name) => name.toLowerCase() !== "coordinator")
341+
.map((name) => `@${name}`)
342+
.join(" ");
343+
344+
const consensusIssue = [
345+
`Session: ${sessionId}`,
346+
`Issue: Please provide Position/Support/Objection for the following question.`,
347+
`Question: ${problemDescription}`,
348+
agentMentions ? `Agents: ${agentMentions}` : null,
349+
agentMentions ? `Agents listed above: reply in this room with "Position:", "Support:", or "Objection:" (one or more lines).` : null,
350+
`Respond with Position/Support/Objection.`,
351+
``
352+
].filter(Boolean).join("\n");
353+
354+
await this.sendStatusMessage(consensusIssue, { forceChat: true });
355+
356+
if (!Number.isFinite(this.debateTimeoutMs)) {
357+
throw new Error("Consensus timeout missing or invalid; check profile mfrConfig.");
358+
}
359+
360+
setTimeout(async () => {
361+
await this.concludeConsensus(sessionId);
362+
}, this.debateTimeoutMs);
363+
364+
const consensusSeconds = Math.round(this.debateTimeoutMs / 1000);
365+
if (quiet) {
366+
return `Consensus started: ${sessionId}.`;
367+
}
368+
return verbose
369+
? `Consensus session started: ${sessionId}\n\nWindow: ${consensusSeconds} seconds\nWaiting for Position/Support/Objection entries...`
370+
: `Consensus started: ${sessionId}. Window: ${consensusSeconds}s.`;
371+
}
372+
285373
/**
286374
* Conclude debate and transition to entity discovery
287375
* @param {string} sessionId - Session ID
@@ -305,6 +393,11 @@ export class CoordinatorProvider extends BaseProvider {
305393
`[CoordinatorProvider] Concluding debate for ${sessionId}`
306394
);
307395

396+
if (debateData.mode === "consensus") {
397+
await this.concludeConsensus(sessionId);
398+
return;
399+
}
400+
308401
// For now, proceed with all agents (consensus detection to be added with Chair integration)
309402
const selected = Array.isArray(selectedAgents) && selectedAgents.length > 0
310403
? selectedAgents
@@ -1035,6 +1128,18 @@ export class CoordinatorProvider extends BaseProvider {
10351128
}
10361129
}
10371130

1131+
async sendLogMessage(message) {
1132+
if (!message) return;
1133+
if (!this.logRoomJid || !this.negotiator?.xmppClient) return;
1134+
await this.negotiator.xmppClient.send(
1135+
xml(
1136+
"message",
1137+
{ to: this.logRoomJid, type: "groupchat" },
1138+
xml("body", {}, message)
1139+
)
1140+
);
1141+
}
1142+
10381143
getSessionVerbose(sessionId) {
10391144
const debateData = this.activeDebates.get(sessionId);
10401145
if (debateData?.verbose) return true;
@@ -1126,6 +1231,22 @@ export class CoordinatorProvider extends BaseProvider {
11261231
getActiveDebateSessionId() {
11271232
let latest = null;
11281233
for (const [sessionId, debateData] of this.activeDebates.entries()) {
1234+
if (debateData.mode === "consensus") {
1235+
continue;
1236+
}
1237+
if (!latest || debateData.startTime > latest.startTime) {
1238+
latest = { sessionId, startTime: debateData.startTime };
1239+
}
1240+
}
1241+
return latest?.sessionId || null;
1242+
}
1243+
1244+
getActiveConsensusSessionId() {
1245+
let latest = null;
1246+
for (const [sessionId, debateData] of this.activeDebates.entries()) {
1247+
if (debateData.mode !== "consensus") {
1248+
continue;
1249+
}
11291250
if (!latest || debateData.startTime > latest.startTime) {
11301251
latest = { sessionId, startTime: debateData.startTime };
11311252
}
@@ -1184,6 +1305,114 @@ export class CoordinatorProvider extends BaseProvider {
11841305
: `Consensus received for ${sessionId}.`;
11851306
}
11861307

1308+
recordConsensusEntry(sessionId, message, metadata = {}) {
1309+
const debateData = this.activeDebates.get(sessionId);
1310+
if (!debateData || debateData.mode !== "consensus") return;
1311+
1312+
const entries = parseConsensusEntries(message || "");
1313+
if (entries.length === 0) {
1314+
const rawMessages = debateData.rawMessages || [];
1315+
if (rawMessages.length < 200) {
1316+
rawMessages.push({
1317+
sender: metadata?.sender || "unknown",
1318+
text: String(message || "").trim(),
1319+
timestamp: new Date().toISOString()
1320+
});
1321+
debateData.rawMessages = rawMessages;
1322+
}
1323+
return;
1324+
}
1325+
1326+
entries.forEach((entry) => {
1327+
debateData.positions.push({
1328+
...entry,
1329+
sender: metadata?.sender || "unknown",
1330+
timestamp: new Date().toISOString()
1331+
});
1332+
});
1333+
}
1334+
1335+
async concludeConsensus(sessionId) {
1336+
const state = this.activeSessions.get(sessionId);
1337+
const debateData = this.activeDebates.get(sessionId);
1338+
1339+
if (!state || !debateData) {
1340+
this.logger.warn?.(`[CoordinatorProvider] No active consensus for ${sessionId}`);
1341+
return;
1342+
}
1343+
1344+
if (!state.isPhase(MFR_PHASES.CONSENSUS_DISCOVERY)) {
1345+
return;
1346+
}
1347+
1348+
this.logger.info?.(
1349+
`[CoordinatorProvider] Concluding consensus for ${sessionId}`
1350+
);
1351+
1352+
const summary = synthesizeConsensus(debateData);
1353+
const lines = [
1354+
`Consensus session complete: ${sessionId}`,
1355+
`Answer: ${summary.answer || "No clear consensus reached."}`
1356+
];
1357+
if (summary.support.length > 0) {
1358+
lines.push(`Support: ${summary.support.join(" | ")}`);
1359+
}
1360+
if (summary.objections.length > 0) {
1361+
lines.push(`Objections: ${summary.objections.join(" | ")}`);
1362+
}
1363+
1364+
await this.sendStatusMessage(lines.join("\n"), { sessionId, forceChat: true });
1365+
await this.sendConsensusLog(sessionId, debateData, summary);
1366+
1367+
state.transition(MFR_PHASES.COMPLETE, {
1368+
consensus: summary.answer || null
1369+
});
1370+
this.activeDebates.delete(sessionId);
1371+
}
1372+
1373+
async sendConsensusLog(sessionId, debateData, summary) {
1374+
if (!debateData || debateData.mode !== "consensus") return;
1375+
const question = debateData.problemDescription || "";
1376+
const entries = debateData.positions || [];
1377+
const rawMessages = debateData.rawMessages || [];
1378+
const lines = [
1379+
`[Consensus Log] ${sessionId}`,
1380+
`Question: ${question}`,
1381+
`Answer: ${summary.answer || "No clear consensus reached."}`,
1382+
`Entries: ${entries.length}`,
1383+
""
1384+
];
1385+
1386+
if (entries.length > 0) {
1387+
lines.push("Parsed entries:");
1388+
entries.forEach((entry, index) => {
1389+
const sender = entry.sender || "unknown";
1390+
const type = entry.type || "note";
1391+
const text = entry.text || "";
1392+
const timestamp = entry.timestamp || "";
1393+
lines.push(`${index + 1}. [${type}] ${sender} ${timestamp}`.trim());
1394+
if (text) {
1395+
lines.push(` ${text}`);
1396+
}
1397+
});
1398+
}
1399+
1400+
if (rawMessages.length > 0) {
1401+
lines.push("", "Raw messages:");
1402+
rawMessages.slice(0, 50).forEach((entry, index) => {
1403+
const sender = entry.sender || "unknown";
1404+
const text = entry.text || "";
1405+
const timestamp = entry.timestamp || "";
1406+
lines.push(`${index + 1}. ${sender} ${timestamp}`.trim());
1407+
if (text) {
1408+
lines.push(` ${text}`);
1409+
}
1410+
});
1411+
}
1412+
1413+
await this.sendLogMessage(lines.join("\n"));
1414+
}
1415+
11871416
/**
11881417
* List all active sessions
11891418
* @param {Object} metadata - Metadata
@@ -1364,3 +1593,52 @@ function cleanTerm(value) {
13641593
.replace(/[?.,!]+$/g, "")
13651594
.replace(/\s+/g, " ");
13661595
}
1596+
1597+
function parseConsensusEntries(message) {
1598+
const text = String(message || "").trim();
1599+
if (!text) return [];
1600+
const entries = [];
1601+
const lines = text.split(/\r?\n/);
1602+
const pattern = /^(?:[-*•>]\s*)?(?:\*{1,2})?(position|support|objection)(?:\*{1,2})?\s*[:\-–—]\s*(.+)$/i;
1603+
for (const line of lines) {
1604+
const trimmed = line.trim();
1605+
if (!trimmed) continue;
1606+
const match = trimmed.match(pattern);
1607+
if (!match) continue;
1608+
entries.push({
1609+
type: match[1].toLowerCase(),
1610+
text: match[2].trim()
1611+
});
1612+
}
1613+
return entries;
1614+
}
1615+
1616+
function synthesizeConsensus(debateData) {
1617+
const positions = debateData.positions || [];
1618+
const positionCounts = new Map();
1619+
const support = [];
1620+
const objections = [];
1621+
1622+
positions.forEach((entry) => {
1623+
if (!entry?.text) return;
1624+
if (entry.type === "position") {
1625+
const key = entry.text.toLowerCase();
1626+
positionCounts.set(key, {
1627+
count: (positionCounts.get(key)?.count || 0) + 1,
1628+
text: entry.text
1629+
});
1630+
} else if (entry.type === "support") {
1631+
if (support.length < 3) support.push(entry.text);
1632+
} else if (entry.type === "objection") {
1633+
if (objections.length < 3) objections.push(entry.text);
1634+
}
1635+
});
1636+
1637+
let answer = null;
1638+
if (positionCounts.size > 0) {
1639+
const sorted = Array.from(positionCounts.values()).sort((a, b) => b.count - a.count);
1640+
answer = sorted[0]?.text || null;
1641+
}
1642+
1643+
return { answer, support, objections };
1644+
}

‎src/lib/mfr/constants.js‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ export const MFR_PHASES = {
7171
// Phase 1: Collaborative Model Construction
7272
PROBLEM_INTERPRETATION: 'problem_interpretation',
7373
TOOL_SELECTION_DEBATE: 'tool_selection_debate', // Optional debate phase for tool selection
74+
CONSENSUS_DISCOVERY: 'consensus_discovery',
7475
ENTITY_DISCOVERY: 'entity_discovery',
7576
CONSTRAINT_IDENTIFICATION: 'constraint_identification',
7677
ACTION_DEFINITION: 'action_definition',
@@ -104,11 +105,15 @@ export const VALID_PHASE_TRANSITIONS = {
104105
],
105106
[MFR_PHASES.PROBLEM_INTERPRETATION]: [
106107
MFR_PHASES.TOOL_SELECTION_DEBATE, // Optional debate path
108+
MFR_PHASES.CONSENSUS_DISCOVERY,
107109
MFR_PHASES.ENTITY_DISCOVERY // Direct path (existing behavior)
108110
],
109111
[MFR_PHASES.TOOL_SELECTION_DEBATE]: [
110112
MFR_PHASES.ENTITY_DISCOVERY
111113
],
114+
[MFR_PHASES.CONSENSUS_DISCOVERY]: [
115+
MFR_PHASES.COMPLETE
116+
],
112117
[MFR_PHASES.ENTITY_DISCOVERY]: [
113118
MFR_PHASES.CONSTRAINT_IDENTIFICATION,
114119
MFR_PHASES.ACTION_DEFINITION,

‎src/lib/xmpp-room-agent.js‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -129,14 +129,14 @@ export class XmppRoomAgent {
129129
}
130130
return;
131131
}
132-
const sender = from.split("/")[1] || "unknown";
132+
const [roomJid, sender] = from.split("/");
133133
this.logger.debug?.(`Groupchat from ${sender}: ${body}`);
134134
if (this.onMessage) {
135135
await this.onMessage({
136136
body,
137-
sender,
137+
sender: sender || "unknown",
138138
from,
139-
roomJid: this.roomJid,
139+
roomJid: roomJid || this.roomJid,
140140
type: "groupchat",
141141
reply: (text) => this.sendGroupMessage(text),
142142
stanza

0 commit comments

Comments
 (0)