mirror of
https://github.com/legop3/MultiRoombaRover.git
synced 2026-09-16 17:40:46 -04:00
llm
This commit is contained in:
legop3
@@ -0,0 +1,54 @@
|
||||
// llm Commentary Service hooks
|
||||
// Purpose: Registers socket/admin control hooks and rover/role event listeners for commentary runtime.
|
||||
// Scope: Keeps runtime behavior unchanged while isolating framework/event wiring from core orchestration.
|
||||
function registerHooks(deps) {
|
||||
const {
|
||||
io,
|
||||
roleEvents,
|
||||
roverManager,
|
||||
emitStatusToSocket,
|
||||
isAdminSocket,
|
||||
clearRuntimeHistory,
|
||||
getAdminState,
|
||||
onDriverActivity,
|
||||
onSensorEvent,
|
||||
onRoverRemoved,
|
||||
} = deps;
|
||||
|
||||
io.on('connection', (socket) => {
|
||||
emitStatusToSocket(socket);
|
||||
socket.on('llm:control', ({ controls } = {}, cb = () => {}) => {
|
||||
if (!isAdminSocket(socket)) {
|
||||
cb({ error: 'Not authorized' });
|
||||
return;
|
||||
}
|
||||
const command = controls?.action || null;
|
||||
if (command === 'clearHistory') {
|
||||
clearRuntimeHistory();
|
||||
cb({ success: true, state: getAdminState() });
|
||||
return;
|
||||
}
|
||||
cb({ error: 'Unknown llm control action' });
|
||||
});
|
||||
});
|
||||
|
||||
roleEvents.on('change', ({ socket }) => {
|
||||
emitStatusToSocket(socket);
|
||||
});
|
||||
|
||||
roverManager.managerEvents.on('sensor', onSensorEvent);
|
||||
roverManager.managerEvents.on('rover', ({ roverId, action } = {}) => {
|
||||
if (action === 'removed' && roverId) {
|
||||
onRoverRemoved(roverId);
|
||||
}
|
||||
});
|
||||
roverManager.managerEvents.on('driver', ({ action } = {}) => {
|
||||
if (action === 'add') {
|
||||
onDriverActivity();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
registerHooks,
|
||||
};
|
||||
@@ -1,6 +1,6 @@
|
||||
// llm Commentary Service
|
||||
// Purpose: Defines the llm Commentary Service module and the helpers/state used by this service unit.
|
||||
// Scope: Keeps runtime behavior unchanged while isolating responsibilities into a clear module boundary.
|
||||
// Purpose: Composes commentary runtime modules (snapshot engine, runner, hooks) and exports startup wiring.
|
||||
// Scope: Keeps runtime behavior unchanged while making this entrypoint a thin orchestration layer.
|
||||
const fsp = require('fs/promises');
|
||||
const { Ollama } = require('ollama');
|
||||
const io = require('../../globals/io');
|
||||
@@ -27,56 +27,33 @@ const {
|
||||
POST_COOLDOWN_MS,
|
||||
normalizeFrequencyMs,
|
||||
} = require('./constants');
|
||||
const {
|
||||
parseModelOutput,
|
||||
normalizeDuplicateKey,
|
||||
buildModelMessages,
|
||||
} = require('./formatters');
|
||||
const {
|
||||
isAdminRole,
|
||||
buildAdminState,
|
||||
buildFailureInfo,
|
||||
} = require('./runtimeHelpers');
|
||||
const { parseModelOutput, normalizeDuplicateKey, buildModelMessages } = require('./formatters');
|
||||
const { isAdminRole, buildAdminState, buildFailureInfo } = require('./runtimeHelpers');
|
||||
const { createSnapshotEngine } = require('./snapshotEngine');
|
||||
const { registerHooks } = require('./hooks');
|
||||
const { createRunner } = require('./runner');
|
||||
|
||||
const config = loadConfig();
|
||||
const commentaryConfig = config.llmCommentary || {};
|
||||
const enabled = Boolean(commentaryConfig.enabled);
|
||||
const ollamaUrl = String(
|
||||
commentaryConfig.ollamaUrl || commentaryConfig.ollamaServer || '',
|
||||
).trim();
|
||||
const ollamaUrl = String(commentaryConfig.ollamaUrl || commentaryConfig.ollamaServer || '').trim();
|
||||
const model = String(commentaryConfig.model || '').trim();
|
||||
const ollamaClient = ollamaUrl ? new Ollama({ host: ollamaUrl }) : null;
|
||||
|
||||
let timer = null;
|
||||
let inFlight = false;
|
||||
let tickCount = 0;
|
||||
let skipStreak = 0;
|
||||
let generationCount = 0;
|
||||
let generationTotalMs = 0;
|
||||
let contextResetAt = Date.now();
|
||||
let clearCount = 0;
|
||||
let runHistory = [];
|
||||
let currentRun = null;
|
||||
const snapshotEngine = createSnapshotEngine({
|
||||
io,
|
||||
roverManager,
|
||||
getActiveDrivers,
|
||||
getNickname,
|
||||
getRecentMessages,
|
||||
MAX_ROVERS,
|
||||
MAX_CHAT_MESSAGES,
|
||||
ACTIVITY_WINDOW_MS,
|
||||
ACTIVITY_BUCKET_MS,
|
||||
ACTIVITY_SCORE_WINDOW_MS,
|
||||
SELF_TALK_WINDOW_MS,
|
||||
MAX_CONTEXT_EVENTS,
|
||||
MAX_ROVER_EVENTS,
|
||||
getContextResetAt: () => contextResetAt,
|
||||
getSkipStreak: () => skipStreak,
|
||||
});
|
||||
|
||||
const frequencyMs = normalizeFrequencyMs(Number(commentaryConfig.frequency ?? commentaryConfig.frequencyMs));
|
||||
|
||||
const runtime = {
|
||||
timer: null,
|
||||
inFlight: false,
|
||||
tickCount: 0,
|
||||
skipStreak: 0,
|
||||
generationCount: 0,
|
||||
generationTotalMs: 0,
|
||||
contextResetAt: Date.now(),
|
||||
clearCount: 0,
|
||||
runHistory: [],
|
||||
currentRun: null,
|
||||
};
|
||||
|
||||
let status = {
|
||||
enabled,
|
||||
model,
|
||||
@@ -108,8 +85,8 @@ let status = {
|
||||
lastModelOutputAt: null,
|
||||
lastModelOutputTickId: null,
|
||||
lastSnapshotSummary: null,
|
||||
lastClearedAt: contextResetAt,
|
||||
clearCount,
|
||||
lastClearedAt: runtime.contextResetAt,
|
||||
clearCount: runtime.clearCount,
|
||||
skipStreak: 0,
|
||||
lastGenerationMs: null,
|
||||
avgGenerationMs: null,
|
||||
@@ -120,6 +97,35 @@ let status = {
|
||||
updatedAt: Date.now(),
|
||||
};
|
||||
|
||||
function isAdminSocket(socket) {
|
||||
if (!socket) return false;
|
||||
return isAdminRole(getRole(socket));
|
||||
}
|
||||
|
||||
function emitStatusToSocket(socket) {
|
||||
if (!socket || !isAdminSocket(socket)) return;
|
||||
socket.emit('llm:state', buildAdminState(status, runtime.runHistory));
|
||||
}
|
||||
|
||||
function emitStatusToAdmins() {
|
||||
const payload = buildAdminState(status, runtime.runHistory);
|
||||
io.sockets.sockets.forEach((socket) => {
|
||||
if (!isAdminSocket(socket)) return;
|
||||
socket.emit('llm:state', payload);
|
||||
});
|
||||
}
|
||||
|
||||
function updateStatus(patch = {}) {
|
||||
const next = {
|
||||
...status,
|
||||
...patch,
|
||||
updatedAt: Date.now(),
|
||||
};
|
||||
if (JSON.stringify(next) === JSON.stringify(status)) return;
|
||||
status = next;
|
||||
emitStatusToAdmins();
|
||||
}
|
||||
|
||||
function updatePhase(phase, patch = {}) {
|
||||
updateStatus({
|
||||
phase,
|
||||
@@ -129,7 +135,7 @@ function updatePhase(phase, patch = {}) {
|
||||
}
|
||||
|
||||
function startRunRecord(tickId, snapshotSummary) {
|
||||
currentRun = {
|
||||
runtime.currentRun = {
|
||||
runId: tickId,
|
||||
startedAt: Date.now(),
|
||||
endedAt: null,
|
||||
@@ -156,58 +162,26 @@ function startRunRecord(tickId, snapshotSummary) {
|
||||
}
|
||||
|
||||
function patchCurrentRun(patch = {}) {
|
||||
if (!currentRun) return;
|
||||
currentRun = {
|
||||
...currentRun,
|
||||
if (!runtime.currentRun) return;
|
||||
runtime.currentRun = {
|
||||
...runtime.currentRun,
|
||||
...patch,
|
||||
};
|
||||
}
|
||||
|
||||
function finalizeRunRecord({ outcome, reason, errors } = {}) {
|
||||
if (!currentRun) return;
|
||||
if (!runtime.currentRun) return;
|
||||
const endedAt = Date.now();
|
||||
const finalized = {
|
||||
...currentRun,
|
||||
outcome: outcome ?? currentRun.outcome,
|
||||
reason: reason ?? currentRun.reason,
|
||||
errors: errors ?? currentRun.errors ?? null,
|
||||
...runtime.currentRun,
|
||||
outcome: outcome ?? runtime.currentRun.outcome,
|
||||
reason: reason ?? runtime.currentRun.reason,
|
||||
errors: errors ?? runtime.currentRun.errors ?? null,
|
||||
endedAt,
|
||||
durationMs: Math.max(0, endedAt - Number(currentRun.startedAt || endedAt)),
|
||||
durationMs: Math.max(0, endedAt - Number(runtime.currentRun.startedAt || endedAt)),
|
||||
};
|
||||
runHistory = [...runHistory.slice(-(MAX_RUN_HISTORY - 1)), finalized];
|
||||
currentRun = null;
|
||||
}
|
||||
|
||||
function isAdminSocket(socket) {
|
||||
if (!socket) return false;
|
||||
const role = getRole(socket);
|
||||
return isAdminRole(role);
|
||||
}
|
||||
|
||||
function emitStatusToSocket(socket) {
|
||||
if (!socket || !isAdminSocket(socket)) return;
|
||||
socket.emit('llm:state', buildAdminState(status, runHistory));
|
||||
}
|
||||
|
||||
function emitStatusToAdmins() {
|
||||
const payload = buildAdminState(status, runHistory);
|
||||
io.sockets.sockets.forEach((socket) => {
|
||||
if (!isAdminSocket(socket)) return;
|
||||
socket.emit('llm:state', payload);
|
||||
});
|
||||
}
|
||||
|
||||
function updateStatus(patch = {}) {
|
||||
const next = {
|
||||
...status,
|
||||
...patch,
|
||||
updatedAt: Date.now(),
|
||||
};
|
||||
if (JSON.stringify(next) === JSON.stringify(status)) {
|
||||
return;
|
||||
}
|
||||
status = next;
|
||||
emitStatusToAdmins();
|
||||
runtime.runHistory = [...runtime.runHistory.slice(-(MAX_RUN_HISTORY - 1)), finalized];
|
||||
runtime.currentRun = null;
|
||||
}
|
||||
|
||||
async function readSystemPrompt() {
|
||||
@@ -241,305 +215,60 @@ async function generateCommentary(messages) {
|
||||
return parseModelOutput(payload?.message?.content);
|
||||
}
|
||||
|
||||
function defaultTickDelayMs() {
|
||||
return frequencyMs + Math.floor(Math.random() * (JITTER_MS + 1));
|
||||
}
|
||||
|
||||
function scheduleNextTick(delayMs = defaultTickDelayMs()) {
|
||||
const safeDelay = Math.max(0, Number.isFinite(delayMs) ? Math.floor(delayMs) : defaultTickDelayMs());
|
||||
const nextRunAt = Date.now() + safeDelay;
|
||||
updateStatus({ nextRunAt });
|
||||
timer = setTimeout(runTick, safeDelay);
|
||||
}
|
||||
|
||||
function wakeForDriverActivity() {
|
||||
if (inFlight) return;
|
||||
if (timer) {
|
||||
clearTimeout(timer);
|
||||
timer = null;
|
||||
}
|
||||
scheduleNextTick(0);
|
||||
}
|
||||
|
||||
async function runTick() {
|
||||
let nextDelayMs = defaultTickDelayMs();
|
||||
tickCount += 1;
|
||||
const tickId = tickCount;
|
||||
updatePhase('tick_started', {
|
||||
tickCount,
|
||||
inFlight: true,
|
||||
currentRunId: tickId,
|
||||
lastTickAt: Date.now(),
|
||||
lastError: null,
|
||||
lastErrorDetails: null,
|
||||
});
|
||||
if (inFlight) {
|
||||
logger.info('Commentary tick skipped; previous tick still running', { tickId });
|
||||
updatePhase('idle', {
|
||||
inFlight: false,
|
||||
currentRunId: null,
|
||||
lastOutcome: 'skipped',
|
||||
lastReason: 'previous tick still running',
|
||||
});
|
||||
scheduleNextTick(nextDelayMs);
|
||||
return;
|
||||
}
|
||||
inFlight = true;
|
||||
try {
|
||||
const snapshot = snapshotEngine.buildSnapshot();
|
||||
const snapshotSummary = {
|
||||
activeDrivers: snapshot?.run_meta?.active_driver_count || 0,
|
||||
rovers: snapshot?.current_snapshot?.rovers?.length || 0,
|
||||
chatMessages: (snapshot?.event_stream || []).filter((event) => event?.type === 'chat').length,
|
||||
drivingRovers: snapshot?.run_meta?.driving_rovers || [],
|
||||
eventCount: snapshot?.event_stream?.length || 0,
|
||||
};
|
||||
logger.info('Commentary tick started', {
|
||||
tickId,
|
||||
...snapshotSummary,
|
||||
});
|
||||
startRunRecord(tickId, snapshotSummary);
|
||||
patchCurrentRun({
|
||||
phase: 'snapshot_ready',
|
||||
summary: snapshotSummary,
|
||||
input: {
|
||||
...(currentRun?.input || {}),
|
||||
infoSnapshot: snapshot,
|
||||
},
|
||||
});
|
||||
updatePhase('snapshot_ready', {
|
||||
lastSnapshotSummary: snapshotSummary,
|
||||
lastInfoSnapshot: snapshot,
|
||||
});
|
||||
const systemPrompt = await readSystemPrompt();
|
||||
const snapshotForSend = snapshotEngine.refreshFinalSnapshotForSend(snapshot);
|
||||
const modelMessages = buildModelMessages(systemPrompt, snapshotForSend);
|
||||
const modelInputAt = Date.now();
|
||||
patchCurrentRun({
|
||||
phase: 'input_ready',
|
||||
input: {
|
||||
...(currentRun?.input || {}),
|
||||
systemPrompt,
|
||||
infoSnapshot: snapshotForSend,
|
||||
modelMessages,
|
||||
modelInputAt,
|
||||
},
|
||||
});
|
||||
updatePhase('awaiting_model_output', {
|
||||
lastModelMessages: modelMessages,
|
||||
lastModelInputAt: modelInputAt,
|
||||
lastModelInputTickId: tickId,
|
||||
lastInfoSnapshot: snapshotForSend,
|
||||
lastModelRawOutput: null,
|
||||
lastModelOutputAt: null,
|
||||
lastModelOutputTickId: null,
|
||||
lastReason: 'awaiting model output',
|
||||
});
|
||||
const generationStartMs = Date.now();
|
||||
const modelResult = await generateCommentary(modelMessages);
|
||||
const generationMs = Math.max(0, Date.now() - generationStartMs);
|
||||
generationCount += 1;
|
||||
generationTotalMs += generationMs;
|
||||
const avgGenerationMs = Math.round(generationTotalMs / generationCount);
|
||||
const modelOutputAt = Date.now();
|
||||
patchCurrentRun({
|
||||
phase: 'output_received',
|
||||
output: {
|
||||
...(currentRun?.output || {}),
|
||||
raw: modelResult?.raw || '',
|
||||
normalized: modelResult?.normalized || null,
|
||||
modelOutputAt,
|
||||
},
|
||||
});
|
||||
updatePhase('output_received', {
|
||||
lastModelRawOutput: modelResult?.raw || '',
|
||||
lastModelOutputAt: modelOutputAt,
|
||||
lastModelOutputTickId: tickId,
|
||||
lastGenerationMs: generationMs,
|
||||
avgGenerationMs,
|
||||
generationCount,
|
||||
});
|
||||
const text = modelResult?.normalized;
|
||||
if (!text) {
|
||||
logger.info('Commentary tick produced SKIP/empty output', { tickId });
|
||||
skipStreak += 1;
|
||||
patchCurrentRun({
|
||||
phase: 'decision_skip',
|
||||
outcome: 'skipped',
|
||||
reason: modelResult?.raw?.trim() ? 'model returned SKIP' : 'model returned empty',
|
||||
});
|
||||
updatePhase('decision_skip', {
|
||||
lastOutcome: 'skipped',
|
||||
lastReason: modelResult?.raw?.trim() ? 'model returned SKIP' : 'model returned empty',
|
||||
skipStreak,
|
||||
lastGeneratedText: null,
|
||||
});
|
||||
finalizeRunRecord({
|
||||
outcome: 'skipped',
|
||||
reason: modelResult?.raw?.trim() ? 'model returned SKIP' : 'model returned empty',
|
||||
});
|
||||
// Immediate retry after model skip.
|
||||
nextDelayMs = 0;
|
||||
return;
|
||||
}
|
||||
updatePhase('decision_post', {
|
||||
lastGeneratedText: text,
|
||||
});
|
||||
const recentBotMessages = getRecentMessages(120, { includeSystem: true })
|
||||
.filter((entry) => Number(entry?.ts) >= contextResetAt)
|
||||
.filter((entry) => entry?.system)
|
||||
.slice(-Math.max(3, MAX_BOT_MESSAGES));
|
||||
const duplicateKey = normalizeDuplicateKey(text);
|
||||
const duplicate = recentBotMessages.some(
|
||||
(entry) => normalizeDuplicateKey(entry?.text) === duplicateKey,
|
||||
);
|
||||
if (duplicate) {
|
||||
logger.info('Commentary tick skipped duplicate output', { tickId, text });
|
||||
skipStreak += 1;
|
||||
patchCurrentRun({
|
||||
phase: 'decision_skip',
|
||||
outcome: 'skipped',
|
||||
reason: 'duplicate text',
|
||||
});
|
||||
updatePhase('decision_skip', {
|
||||
lastOutcome: 'skipped',
|
||||
lastReason: 'duplicate text',
|
||||
skipStreak,
|
||||
});
|
||||
finalizeRunRecord({
|
||||
outcome: 'skipped',
|
||||
reason: 'duplicate text',
|
||||
});
|
||||
// Immediate retry after a skip outcome.
|
||||
nextDelayMs = 0;
|
||||
return;
|
||||
}
|
||||
sendSystemMessage(text);
|
||||
logger.info('Commentary message posted', { tickId, text });
|
||||
skipStreak = 0;
|
||||
patchCurrentRun({
|
||||
phase: 'posted',
|
||||
outcome: 'posted',
|
||||
reason: null,
|
||||
output: {
|
||||
...(currentRun?.output || {}),
|
||||
posted: text,
|
||||
postedAt: Date.now(),
|
||||
},
|
||||
});
|
||||
updatePhase('posted', {
|
||||
lastOutcome: 'posted',
|
||||
lastReason: null,
|
||||
skipStreak,
|
||||
lastPostedText: text,
|
||||
lastPostedAt: Date.now(),
|
||||
});
|
||||
nextDelayMs = Math.max(nextDelayMs, POST_COOLDOWN_MS);
|
||||
finalizeRunRecord({
|
||||
outcome: 'posted',
|
||||
reason: null,
|
||||
});
|
||||
} catch (err) {
|
||||
const failure = buildFailureInfo(err);
|
||||
logger.warn('Commentary tick failed', {
|
||||
tickId,
|
||||
reason: failure.reason,
|
||||
error: failure.message,
|
||||
details: failure.details,
|
||||
});
|
||||
patchCurrentRun({
|
||||
phase: 'failed',
|
||||
outcome: 'failed',
|
||||
reason: failure.reason,
|
||||
errors: {
|
||||
message: failure.message,
|
||||
details: failure.details,
|
||||
},
|
||||
});
|
||||
updatePhase('failed', {
|
||||
lastOutcome: 'failed',
|
||||
lastReason: failure.reason,
|
||||
lastError: failure.message,
|
||||
lastErrorDetails: failure.details,
|
||||
lastFailedAt: Date.now(),
|
||||
});
|
||||
finalizeRunRecord({
|
||||
outcome: 'failed',
|
||||
reason: failure.reason,
|
||||
errors: {
|
||||
message: failure.message,
|
||||
details: failure.details,
|
||||
},
|
||||
});
|
||||
} finally {
|
||||
inFlight = false;
|
||||
updatePhase('idle', {
|
||||
inFlight: false,
|
||||
currentRunId: null,
|
||||
});
|
||||
scheduleNextTick(nextDelayMs);
|
||||
}
|
||||
}
|
||||
|
||||
function start() {
|
||||
if (!enabled) {
|
||||
logger.info('LLM commentary disabled');
|
||||
updatePhase('disabled', {
|
||||
running: false,
|
||||
lastOutcome: 'disabled',
|
||||
lastReason: 'llmCommentary.enabled is false',
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (!model || !ollamaUrl) {
|
||||
logger.warn('LLM commentary disabled; model or ollamaUrl missing');
|
||||
updatePhase('disabled', {
|
||||
running: false,
|
||||
lastOutcome: 'disabled',
|
||||
lastReason: 'model or ollama server missing',
|
||||
});
|
||||
return;
|
||||
}
|
||||
logger.info('LLM commentary enabled', { model, ollamaUrl, frequencyMs, promptPath: PROMPT_PATH });
|
||||
updatePhase('idle', {
|
||||
running: true,
|
||||
lastOutcome: 'running',
|
||||
lastReason: null,
|
||||
});
|
||||
runTick();
|
||||
}
|
||||
|
||||
io.on('connection', (socket) => {
|
||||
emitStatusToSocket(socket);
|
||||
socket.on('llm:control', ({ controls } = {}, cb = () => {}) => {
|
||||
if (!isAdminSocket(socket)) {
|
||||
cb({ error: 'Not authorized' });
|
||||
return;
|
||||
}
|
||||
const command = controls?.action || null;
|
||||
if (command === 'clearHistory') {
|
||||
clearRuntimeHistory();
|
||||
cb({ success: true, state: buildAdminState(status, runHistory) });
|
||||
return;
|
||||
}
|
||||
cb({ error: 'Unknown llm control action' });
|
||||
});
|
||||
const snapshotEngine = createSnapshotEngine({
|
||||
io,
|
||||
roverManager,
|
||||
getActiveDrivers,
|
||||
getNickname,
|
||||
getRecentMessages,
|
||||
MAX_ROVERS,
|
||||
MAX_CHAT_MESSAGES,
|
||||
ACTIVITY_WINDOW_MS,
|
||||
ACTIVITY_BUCKET_MS,
|
||||
ACTIVITY_SCORE_WINDOW_MS,
|
||||
SELF_TALK_WINDOW_MS,
|
||||
MAX_CONTEXT_EVENTS,
|
||||
MAX_ROVER_EVENTS,
|
||||
getContextResetAt: () => runtime.contextResetAt,
|
||||
getSkipStreak: () => runtime.skipStreak,
|
||||
});
|
||||
|
||||
roleEvents.on('change', ({ socket }) => {
|
||||
emitStatusToSocket(socket);
|
||||
const runner = createRunner({
|
||||
logger,
|
||||
enabled,
|
||||
model,
|
||||
ollamaUrl,
|
||||
frequencyMs,
|
||||
jitterMs: JITTER_MS,
|
||||
postCooldownMs: POST_COOLDOWN_MS,
|
||||
maxBotMessages: MAX_BOT_MESSAGES,
|
||||
runtime,
|
||||
snapshotEngine,
|
||||
readSystemPrompt,
|
||||
buildModelMessages,
|
||||
generateCommentary,
|
||||
normalizeDuplicateKey,
|
||||
getRecentMessages,
|
||||
sendSystemMessage,
|
||||
buildFailureInfo,
|
||||
updatePhase,
|
||||
startRunRecord,
|
||||
patchCurrentRun,
|
||||
finalizeRunRecord,
|
||||
updateStatus,
|
||||
});
|
||||
|
||||
roverManager.managerEvents.on('sensor', snapshotEngine.onSensorEvent);
|
||||
roverManager.managerEvents.on('rover', ({ roverId, action } = {}) => {
|
||||
if (action === 'removed' && roverId) {
|
||||
snapshotEngine.removeRover(roverId);
|
||||
}
|
||||
});
|
||||
roverManager.managerEvents.on('driver', ({ action } = {}) => {
|
||||
if (action === 'add') {
|
||||
wakeForDriverActivity();
|
||||
}
|
||||
registerHooks({
|
||||
io,
|
||||
roleEvents,
|
||||
roverManager,
|
||||
emitStatusToSocket,
|
||||
isAdminSocket,
|
||||
clearRuntimeHistory: runner.clearRuntimeHistory,
|
||||
getAdminState: () => buildAdminState(status, runtime.runHistory),
|
||||
onDriverActivity: runner.wakeForDriverActivity,
|
||||
onSensorEvent: snapshotEngine.onSensorEvent,
|
||||
onRoverRemoved: snapshotEngine.removeRover,
|
||||
});
|
||||
|
||||
start();
|
||||
runner.start();
|
||||
|
||||
@@ -0,0 +1,342 @@
|
||||
// llm Commentary Service runner
|
||||
// Purpose: Owns commentary tick scheduling, generation loop, and runtime-history reset behavior.
|
||||
// Scope: Keeps runtime behavior unchanged by operating on injected mutable runtime state and callbacks.
|
||||
function createRunner(deps) {
|
||||
const {
|
||||
logger,
|
||||
enabled,
|
||||
model,
|
||||
ollamaUrl,
|
||||
frequencyMs,
|
||||
jitterMs,
|
||||
postCooldownMs,
|
||||
maxBotMessages,
|
||||
runtime,
|
||||
snapshotEngine,
|
||||
readSystemPrompt,
|
||||
buildModelMessages,
|
||||
generateCommentary,
|
||||
normalizeDuplicateKey,
|
||||
getRecentMessages,
|
||||
sendSystemMessage,
|
||||
buildFailureInfo,
|
||||
updatePhase,
|
||||
startRunRecord,
|
||||
patchCurrentRun,
|
||||
finalizeRunRecord,
|
||||
updateStatus,
|
||||
} = deps;
|
||||
|
||||
function defaultTickDelayMs() {
|
||||
return frequencyMs + Math.floor(Math.random() * (jitterMs + 1));
|
||||
}
|
||||
|
||||
function scheduleNextTick(runTick, delayMs = defaultTickDelayMs()) {
|
||||
const safeDelay = Math.max(0, Number.isFinite(delayMs) ? Math.floor(delayMs) : defaultTickDelayMs());
|
||||
const nextRunAt = Date.now() + safeDelay;
|
||||
updateStatus({ nextRunAt });
|
||||
runtime.timer = setTimeout(runTick, safeDelay);
|
||||
}
|
||||
|
||||
function wakeForDriverActivity(runTick) {
|
||||
if (runtime.inFlight) return;
|
||||
if (runtime.timer) {
|
||||
clearTimeout(runtime.timer);
|
||||
runtime.timer = null;
|
||||
}
|
||||
scheduleNextTick(runTick, 0);
|
||||
}
|
||||
|
||||
function clearRuntimeHistory() {
|
||||
runtime.contextResetAt = Date.now();
|
||||
runtime.clearCount += 1;
|
||||
runtime.skipStreak = 0;
|
||||
runtime.generationCount = 0;
|
||||
runtime.generationTotalMs = 0;
|
||||
runtime.runHistory = [];
|
||||
runtime.currentRun = null;
|
||||
snapshotEngine.resetHistory();
|
||||
updateStatus({
|
||||
lastClearedAt: runtime.contextResetAt,
|
||||
clearCount: runtime.clearCount,
|
||||
skipStreak: runtime.skipStreak,
|
||||
phase: 'idle',
|
||||
phaseAt: Date.now(),
|
||||
currentRunId: null,
|
||||
lastGenerationMs: null,
|
||||
avgGenerationMs: null,
|
||||
generationCount: runtime.generationCount,
|
||||
lastInfoSnapshot: null,
|
||||
lastModelMessages: null,
|
||||
lastModelInputAt: null,
|
||||
lastModelInputTickId: null,
|
||||
lastModelRawOutput: null,
|
||||
lastModelOutputAt: null,
|
||||
lastModelOutputTickId: null,
|
||||
lastSnapshotSummary: null,
|
||||
lastGeneratedText: null,
|
||||
lastPostedText: null,
|
||||
lastPostedAt: null,
|
||||
lastError: null,
|
||||
lastErrorDetails: null,
|
||||
lastFailedAt: null,
|
||||
lastOutcome: 'cleared',
|
||||
lastReason: 'admin requested clear history',
|
||||
});
|
||||
}
|
||||
|
||||
async function runTick() {
|
||||
let nextDelayMs = defaultTickDelayMs();
|
||||
runtime.tickCount += 1;
|
||||
const tickId = runtime.tickCount;
|
||||
updatePhase('tick_started', {
|
||||
tickCount: runtime.tickCount,
|
||||
inFlight: true,
|
||||
currentRunId: tickId,
|
||||
lastTickAt: Date.now(),
|
||||
lastError: null,
|
||||
lastErrorDetails: null,
|
||||
});
|
||||
if (runtime.inFlight) {
|
||||
logger.info('Commentary tick skipped; previous tick still running', { tickId });
|
||||
updatePhase('idle', {
|
||||
inFlight: false,
|
||||
currentRunId: null,
|
||||
lastOutcome: 'skipped',
|
||||
lastReason: 'previous tick still running',
|
||||
});
|
||||
scheduleNextTick(runTick, nextDelayMs);
|
||||
return;
|
||||
}
|
||||
runtime.inFlight = true;
|
||||
try {
|
||||
const snapshot = snapshotEngine.buildSnapshot();
|
||||
const snapshotSummary = {
|
||||
activeDrivers: snapshot?.run_meta?.active_driver_count || 0,
|
||||
rovers: snapshot?.current_snapshot?.rovers?.length || 0,
|
||||
chatMessages: (snapshot?.event_stream || []).filter((event) => event?.type === 'chat').length,
|
||||
drivingRovers: snapshot?.run_meta?.driving_rovers || [],
|
||||
eventCount: snapshot?.event_stream?.length || 0,
|
||||
};
|
||||
logger.info('Commentary tick started', {
|
||||
tickId,
|
||||
...snapshotSummary,
|
||||
});
|
||||
startRunRecord(tickId, snapshotSummary);
|
||||
patchCurrentRun({
|
||||
phase: 'snapshot_ready',
|
||||
summary: snapshotSummary,
|
||||
input: {
|
||||
...(runtime.currentRun?.input || {}),
|
||||
infoSnapshot: snapshot,
|
||||
},
|
||||
});
|
||||
updatePhase('snapshot_ready', {
|
||||
lastSnapshotSummary: snapshotSummary,
|
||||
lastInfoSnapshot: snapshot,
|
||||
});
|
||||
const systemPrompt = await readSystemPrompt();
|
||||
const snapshotForSend = snapshotEngine.refreshFinalSnapshotForSend(snapshot);
|
||||
const modelMessages = buildModelMessages(systemPrompt, snapshotForSend);
|
||||
const modelInputAt = Date.now();
|
||||
patchCurrentRun({
|
||||
phase: 'input_ready',
|
||||
input: {
|
||||
...(runtime.currentRun?.input || {}),
|
||||
systemPrompt,
|
||||
infoSnapshot: snapshotForSend,
|
||||
modelMessages,
|
||||
modelInputAt,
|
||||
},
|
||||
});
|
||||
updatePhase('awaiting_model_output', {
|
||||
lastModelMessages: modelMessages,
|
||||
lastModelInputAt: modelInputAt,
|
||||
lastModelInputTickId: tickId,
|
||||
lastInfoSnapshot: snapshotForSend,
|
||||
lastModelRawOutput: null,
|
||||
lastModelOutputAt: null,
|
||||
lastModelOutputTickId: null,
|
||||
lastReason: 'awaiting model output',
|
||||
});
|
||||
const generationStartMs = Date.now();
|
||||
const modelResult = await generateCommentary(modelMessages);
|
||||
const generationMs = Math.max(0, Date.now() - generationStartMs);
|
||||
runtime.generationCount += 1;
|
||||
runtime.generationTotalMs += generationMs;
|
||||
const avgGenerationMs = Math.round(runtime.generationTotalMs / runtime.generationCount);
|
||||
const modelOutputAt = Date.now();
|
||||
patchCurrentRun({
|
||||
phase: 'output_received',
|
||||
output: {
|
||||
...(runtime.currentRun?.output || {}),
|
||||
raw: modelResult?.raw || '',
|
||||
normalized: modelResult?.normalized || null,
|
||||
modelOutputAt,
|
||||
},
|
||||
});
|
||||
updatePhase('output_received', {
|
||||
lastModelRawOutput: modelResult?.raw || '',
|
||||
lastModelOutputAt: modelOutputAt,
|
||||
lastModelOutputTickId: tickId,
|
||||
lastGenerationMs: generationMs,
|
||||
avgGenerationMs,
|
||||
generationCount: runtime.generationCount,
|
||||
});
|
||||
const text = modelResult?.normalized;
|
||||
if (!text) {
|
||||
logger.info('Commentary tick produced SKIP/empty output', { tickId });
|
||||
runtime.skipStreak += 1;
|
||||
patchCurrentRun({
|
||||
phase: 'decision_skip',
|
||||
outcome: 'skipped',
|
||||
reason: modelResult?.raw?.trim() ? 'model returned SKIP' : 'model returned empty',
|
||||
});
|
||||
updatePhase('decision_skip', {
|
||||
lastOutcome: 'skipped',
|
||||
lastReason: modelResult?.raw?.trim() ? 'model returned SKIP' : 'model returned empty',
|
||||
skipStreak: runtime.skipStreak,
|
||||
lastGeneratedText: null,
|
||||
});
|
||||
finalizeRunRecord({
|
||||
outcome: 'skipped',
|
||||
reason: modelResult?.raw?.trim() ? 'model returned SKIP' : 'model returned empty',
|
||||
});
|
||||
nextDelayMs = 0;
|
||||
return;
|
||||
}
|
||||
updatePhase('decision_post', { lastGeneratedText: text });
|
||||
const recentBotMessages = getRecentMessages(120, { includeSystem: true })
|
||||
.filter((entry) => Number(entry?.ts) >= runtime.contextResetAt)
|
||||
.filter((entry) => entry?.system)
|
||||
.slice(-Math.max(3, maxBotMessages));
|
||||
const duplicateKey = normalizeDuplicateKey(text);
|
||||
const duplicate = recentBotMessages.some(
|
||||
(entry) => normalizeDuplicateKey(entry?.text) === duplicateKey,
|
||||
);
|
||||
if (duplicate) {
|
||||
logger.info('Commentary tick skipped duplicate output', { tickId, text });
|
||||
runtime.skipStreak += 1;
|
||||
patchCurrentRun({
|
||||
phase: 'decision_skip',
|
||||
outcome: 'skipped',
|
||||
reason: 'duplicate text',
|
||||
});
|
||||
updatePhase('decision_skip', {
|
||||
lastOutcome: 'skipped',
|
||||
lastReason: 'duplicate text',
|
||||
skipStreak: runtime.skipStreak,
|
||||
});
|
||||
finalizeRunRecord({
|
||||
outcome: 'skipped',
|
||||
reason: 'duplicate text',
|
||||
});
|
||||
nextDelayMs = 0;
|
||||
return;
|
||||
}
|
||||
sendSystemMessage(text);
|
||||
logger.info('Commentary message posted', { tickId, text });
|
||||
runtime.skipStreak = 0;
|
||||
patchCurrentRun({
|
||||
phase: 'posted',
|
||||
outcome: 'posted',
|
||||
reason: null,
|
||||
output: {
|
||||
...(runtime.currentRun?.output || {}),
|
||||
posted: text,
|
||||
postedAt: Date.now(),
|
||||
},
|
||||
});
|
||||
updatePhase('posted', {
|
||||
lastOutcome: 'posted',
|
||||
lastReason: null,
|
||||
skipStreak: runtime.skipStreak,
|
||||
lastPostedText: text,
|
||||
lastPostedAt: Date.now(),
|
||||
});
|
||||
nextDelayMs = Math.max(nextDelayMs, postCooldownMs);
|
||||
finalizeRunRecord({
|
||||
outcome: 'posted',
|
||||
reason: null,
|
||||
});
|
||||
} catch (err) {
|
||||
const failure = buildFailureInfo(err);
|
||||
logger.warn('Commentary tick failed', {
|
||||
tickId,
|
||||
reason: failure.reason,
|
||||
error: failure.message,
|
||||
details: failure.details,
|
||||
});
|
||||
patchCurrentRun({
|
||||
phase: 'failed',
|
||||
outcome: 'failed',
|
||||
reason: failure.reason,
|
||||
errors: {
|
||||
message: failure.message,
|
||||
details: failure.details,
|
||||
},
|
||||
});
|
||||
updatePhase('failed', {
|
||||
lastOutcome: 'failed',
|
||||
lastReason: failure.reason,
|
||||
lastError: failure.message,
|
||||
lastErrorDetails: failure.details,
|
||||
lastFailedAt: Date.now(),
|
||||
});
|
||||
finalizeRunRecord({
|
||||
outcome: 'failed',
|
||||
reason: failure.reason,
|
||||
errors: {
|
||||
message: failure.message,
|
||||
details: failure.details,
|
||||
},
|
||||
});
|
||||
} finally {
|
||||
runtime.inFlight = false;
|
||||
updatePhase('idle', {
|
||||
inFlight: false,
|
||||
currentRunId: null,
|
||||
});
|
||||
scheduleNextTick(runTick, nextDelayMs);
|
||||
}
|
||||
}
|
||||
|
||||
function start() {
|
||||
if (!enabled) {
|
||||
logger.info('LLM commentary disabled');
|
||||
updatePhase('disabled', {
|
||||
running: false,
|
||||
lastOutcome: 'disabled',
|
||||
lastReason: 'llmCommentary.enabled is false',
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (!model || !ollamaUrl) {
|
||||
logger.warn('LLM commentary disabled; model or ollamaUrl missing');
|
||||
updatePhase('disabled', {
|
||||
running: false,
|
||||
lastOutcome: 'disabled',
|
||||
lastReason: 'model or ollama server missing',
|
||||
});
|
||||
return;
|
||||
}
|
||||
logger.info('LLM commentary enabled', { model, ollamaUrl, frequencyMs });
|
||||
updatePhase('idle', {
|
||||
running: true,
|
||||
lastOutcome: 'running',
|
||||
lastReason: null,
|
||||
});
|
||||
runTick();
|
||||
}
|
||||
|
||||
return {
|
||||
start,
|
||||
runTick,
|
||||
clearRuntimeHistory,
|
||||
wakeForDriverActivity: () => wakeForDriverActivity(runTick),
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
createRunner,
|
||||
};
|
||||
Reference in New Issue
Block a user