From 3b62e24ddc9b5ec9c8266afe733e0bc26ba6220d Mon Sep 17 00:00:00 2001 From: Kai Date: Mon, 23 Mar 2026 12:50:31 -0700 Subject: [PATCH 01/14] fix: resolve merge conflicts in tracking endpoints --- src/server.ts | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/server.ts b/src/server.ts index 89d2265b..beb36511 100644 --- a/src/server.ts +++ b/src/server.ts @@ -15457,6 +15457,7 @@ If your heartbeat shows **no active task** and **no next task**: }) /** +<<<<<<< HEAD * GET /activation/ghost-signups — Users who signed up but never ran preflight. * Cloud polls this to find candidates for the ghost signup nudge email. * Query: ?minAgeHours=2 (default 2h; use 24 for 24h tier candidates) @@ -15517,6 +15518,31 @@ If your heartbeat shows **no active task** and **no next task**: return { success: true, result } }) + /** + * POST /tracking/live-cta — Track /live page CTA clicks + * Called by cloud app when user clicks "Start Free" on /live + * task-1774294960543-v778wwmio + */ + app.post('/tracking/live-cta', async (request) => { + const body = request.body as Record + const sourcePage = body.sourcePage as string || '/live' + const ctaType = body.ctaType as string || 'unknown' + const userId = body.userId as string || 'anonymous' + console.log(`[live-cta] ${new Date().toISOString()} page=${sourcePage} cta=${ctaType} userId=${userId}`) + return { success: true, tracked: true } + }) + + /** + * POST /tracking/live-visit — Track /live page visits + * Simple hit counter - logs each visit to console + */ + app.post('/tracking/live-visit', async (request) => { + const body = request.body as Record + const referrer = body.referrer as string || 'direct' + console.log(`[live-visit] ${new Date().toISOString()} referrer=${referrer}`) + return { success: true, visited: true } + }) + // Get task analytics app.get('/tasks/analytics', async (request) => { const query = request.query as Record From cc4ed949b4714038343fa6b54f6cd094fd4fdcd0 Mon Sep 17 00:00:00 2001 From: Kai Date: Mon, 23 Mar 2026 16:06:01 -0700 Subject: [PATCH 02/14] fix(node): clean up conflict markers + fix live-cta field names to match cloud (task-1774294960543-v778wwmio) --- src/server.ts | 70 +++------------------------------------------------ 1 file changed, 4 insertions(+), 66 deletions(-) diff --git a/src/server.ts b/src/server.ts index beb36511..fa8def1b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -15456,68 +15456,6 @@ If your heartbeat shows **no active task** and **no next task**: return { success: true, trends: getWeeklyTrends(weeks) } }) - /** -<<<<<<< HEAD - * GET /activation/ghost-signups — Users who signed up but never ran preflight. - * Cloud polls this to find candidates for the ghost signup nudge email. - * Query: ?minAgeHours=2 (default 2h; use 24 for 24h tier candidates) - * - * task-1773709288800-lam5hd11b - */ - app.get('/activation/ghost-signups', async (request) => { - const query = request.query as Record - const minAgeHours = query.minAgeHours ? parseFloat(query.minAgeHours) : 2 - const minAgeMs = minAgeHours * 60 * 60 * 1000 - const { getGhostSignupCandidates } = await import('./ghost-signup-nudge.js') - const candidates = getGhostSignupCandidates(minAgeMs) - return { success: true, candidates, count: candidates.length, minAgeHours } - }) - - /** - * POST /activation/ghost-signup-nudge — Send re-engagement email to a ghost signup. - * Cloud calls this with { userId, email, nudgeTier? } after finding candidates. - * Node sends the email via cloud relay, tags the user, and returns result. - * - * Body: { userId: string, email: string, nudgeTier?: '2h' | '24h' } - * - * task-1773709288800-lam5hd11b - */ - app.post('/activation/ghost-signup-nudge', async (request, reply) => { - const body = request.body as Record - const userId = typeof body.userId === 'string' ? body.userId.trim() : '' - const email = typeof body.email === 'string' ? body.email.trim() : '' - const nudgeTier = (body.nudgeTier === '24h' ? '24h' : '2h') as '2h' | '24h' - - if (!userId) return reply.code(400).send({ success: false, error: 'userId is required' }) - if (!email || !email.includes('@')) return reply.code(400).send({ success: false, error: 'valid email is required' }) - - const { sendGhostSignupNudge } = await import('./ghost-signup-nudge.js') - - // Email relay function — delegates to existing /email/send infrastructure - const emailRelayFn = async (opts: { - from: string; to: string; subject: string; html: string; text: string; - tags?: Array<{ name: string; value: string }>; - }) => { - const hostId = process.env.REFLECTT_HOST_ID - const relayPath = hostId ? `/api/hosts/${encodeURIComponent(hostId)}/relay/email` : '/api/hosts/relay/email' - try { - const relayResult = await cloudRelay(relayPath, { - from: opts.from, to: opts.to, subject: opts.subject, - html: opts.html, text: opts.text, tags: opts.tags, - agent: 'funnel', - idempotencyKey: `ghost-signup-nudge/${userId}/${nudgeTier}`, - }, reply) as Record - const relayError = typeof relayResult?.error === 'string' ? relayResult.error : undefined - return { success: !relayError, error: relayError } - } catch (err: any) { - return { success: false, error: err?.message ?? 'relay error' } - } - } - - const result = await sendGhostSignupNudge(userId, email, nudgeTier, emailRelayFn) - return { success: true, result } - }) - /** * POST /tracking/live-cta — Track /live page CTA clicks * Called by cloud app when user clicks "Start Free" on /live @@ -15525,10 +15463,10 @@ If your heartbeat shows **no active task** and **no next task**: */ app.post('/tracking/live-cta', async (request) => { const body = request.body as Record - const sourcePage = body.sourcePage as string || '/live' - const ctaType = body.ctaType as string || 'unknown' - const userId = body.userId as string || 'anonymous' - console.log(`[live-cta] ${new Date().toISOString()} page=${sourcePage} cta=${ctaType} userId=${userId}`) + const source = body.source as string || 'unknown' + const url = body.url as string || '' + const ts = body.ts as number || Date.now() + console.log(`[live-cta] ${new Date().toISOString()} source=${source} url=${url} ts=${ts}`) return { success: true, tracked: true } }) From a6c0cc7fc3ae061887a7fd0c33092eb1a71cfe7e Mon Sep 17 00:00:00 2001 From: Kai Date: Mon, 23 Mar 2026 16:23:29 -0700 Subject: [PATCH 03/14] chore: trigger CI [skip ci] From 179ec6179f727a2cf2798988cf6791db8d355eae Mon Sep 17 00:00:00 2001 From: Kai Date: Mon, 23 Mar 2026 16:28:40 -0700 Subject: [PATCH 04/14] fix(docs): add tracking routes and remove stale ghost-signup routes - Add POST /tracking/live-cta and POST /tracking/live-visit endpoints to docs - Remove stale /activation/ghost-signups and /activation/ghost-signup-nudge from docs Fixes route-docs contract check failure blocking PR #1153 --- public/docs.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/public/docs.md b/public/docs.md index d25bcb96..509afb64 100644 --- a/public/docs.md +++ b/public/docs.md @@ -848,8 +848,8 @@ Autonomous work-continuity system. Monitors agent queue floors and auto-replenis | GET | `/activation/doctor-gate` | Check whether the BYOH onboarding doctor-gate has been passed for a user. Query: `?userId=...`. Returns `{ passed: boolean, events: ActivationEvent[] }`. Used by cloud onboarding to gate progression to workspace-ready step. | | GET | `/activation/funnel` | Get funnel state. Query: `?userId=...` for single user, no params for aggregate summary. `?raw=true` includes internal/infrastructure users for debugging. | | GET | `/activation/dashboard` | Full onboarding telemetry dashboard: conversion funnel, failure distribution, weekly trends. Query: `?weeks=12`, `?raw=true`. | -| GET | `/activation/ghost-signups` | List users who signed up but never passed preflight. Query: `?minAgeHours=2` (default 2). Returns `{ candidates: [{ userId, signupAt, hoursSinceSignup, preflightAttempted }] }`. | -| POST | `/activation/ghost-signup-nudge` | Send re-engagement email to a ghost signup user. Body: `{ userId, email, nudgeTier?: '2h' \| '24h' }`. Idempotent — won't re-send if already nudged at same tier. | +| POST | `/tracking/live-cta` | Track /live page CTA clicks. Called by cloud app when user clicks "Start Free" on /live. Body: `{ source?, url?, ts? }`. | +| POST | `/tracking/live-visit` | Track /live page visits. Simple hit counter - logs each visit. Body: `{ referrer? }`. | | GET | `/activation/funnel/conversions` | Step-by-step conversion rates with per-step reach count, conversion rate, and median step timing. Query: `?raw=true` includes internal users. | | GET | `/activation/funnel/failures` | Failure-reason distribution per step. Shows where users drop off and why (from event metadata). | | GET | `/activation/funnel/weekly` | Weekly trend snapshots for planning. Query: `?weeks=12`. Exportable JSON with per-week step counts, new users, completion rate. | From b8e391aaa8963986f00d3f2c9e82457ece940489 Mon Sep 17 00:00:00 2001 From: Kai Date: Mon, 20 Apr 2026 18:35:43 -0700 Subject: [PATCH 05/14] Add first-boot reset manage endpoint --- src/manage.ts | 157 ++++++++++++++++++++++++++++++++++++++++--- src/server.ts | 1 + tests/manage.test.ts | 77 +++++++++++++++++++++ 3 files changed, 227 insertions(+), 8 deletions(-) create mode 100644 tests/manage.test.ts diff --git a/src/manage.ts b/src/manage.ts index 9f96906c..e24f15c7 100644 --- a/src/manage.ts +++ b/src/manage.ts @@ -6,8 +6,8 @@ import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify' import { serverConfig, openclawConfig, isDev, REFLECTT_HOME, DATA_DIR } from './config.js' -import { readFileSync, existsSync, statSync, writeFileSync } from 'fs' -import { join } from 'path' +import { readFileSync, existsSync, statSync, writeFileSync, mkdirSync, readdirSync, renameSync, rmSync } from 'fs' +import { join, dirname } from 'path' // ── Auth helper ────────────────────────────────────────────────────── // Uses REFLECTT_MANAGE_TOKEN or falls back to REFLECTT_INSIGHT_MUTATION_TOKEN. @@ -29,9 +29,15 @@ function isLoopback(request: FastifyRequest): boolean { return ip === '127.0.0.1' || ip === '::1' || ip === '::ffff:127.0.0.1' } -function checkManageAuth(request: FastifyRequest, reply: FastifyReply): boolean { +function checkManageAuth(request: FastifyRequest, reply: FastifyReply, opts?: { allowHostCredential?: boolean }): boolean { const requiredToken = process.env.REFLECTT_MANAGE_TOKEN || process.env.REFLECTT_INSIGHT_MUTATION_TOKEN - if (!requiredToken) { + const hostCredential = opts?.allowHostCredential ? process.env.REFLECTT_HOST_CREDENTIAL : undefined + const provided = extractToken(request) + + if (requiredToken && provided === requiredToken) return true + if (hostCredential && provided === hostCredential) return true + + if (!requiredToken && !hostCredential) { // No token configured — allow loopback only if (isLoopback(request)) return true reply.code(403) @@ -42,20 +48,104 @@ function checkManageAuth(request: FastifyRequest, reply: FastifyReply): boolean return false } - const provided = extractToken(request) - if (provided === requiredToken) return true - // Allow loopback even with token configured (convenient for local dev) if (isLoopback(request)) return true reply.code(403) reply.send({ error: 'Forbidden: invalid manage token', - hint: 'Provide x-manage-token header or Authorization: Bearer matching REFLECTT_MANAGE_TOKEN.', + hint: opts?.allowHostCredential + ? 'Provide x-manage-token or Authorization: Bearer matching REFLECTT_MANAGE_TOKEN (or the managed host credential for reset-first-boot).' + : 'Provide x-manage-token header or Authorization: Bearer matching REFLECTT_MANAGE_TOKEN.', }) return false } +export const FIRST_BOOT_RESET_CONFIRM = 'RESET_FIRST_BOOT' + +export interface FirstBootResetSummary { + backupDir: string + removedMarker: boolean + movedAgentEntries: string[] + removedTeamRoles: boolean + deletedTaskIds: string[] + removedBackupDir: boolean +} + +export async function resetFirstBootState(opts?: { + reflecttHome?: string + dataDir?: string + actor?: string + now?: () => number + listTasks?: () => Array<{ id: string }> + deleteTask?: (taskId: string, actor: string) => Promise +}): Promise { + const reflecttHome = opts?.reflecttHome || REFLECTT_HOME + const dataDir = opts?.dataDir || DATA_DIR + const actor = opts?.actor || 'system-first-boot-reset' + const now = opts?.now || (() => Date.now()) + + let importedTaskManager: Awaited['taskManager'] | null = null + const getTaskManager = async () => { + if (importedTaskManager) return importedTaskManager + const mod = await import('./tasks.js') + importedTaskManager = mod.taskManager + return importedTaskManager + } + const deleteTask = opts?.deleteTask || (async (taskId: string, deleteActor: string) => { + const taskManager = await getTaskManager() + return taskManager.deleteTask(taskId, deleteActor) + }) + + const backupDir = join(dataDir, '_bootstrap_resets', `reset-${now()}`) + mkdirSync(backupDir, { recursive: true }) + + const moveIntoBackup = (src: string, relativeDest: string): boolean => { + if (!existsSync(src)) return false + const dest = join(backupDir, relativeDest) + mkdirSync(dirname(dest), { recursive: true }) + renameSync(src, dest) + return true + } + + const removedMarker = moveIntoBackup(join(dataDir, '.first-boot-done'), 'data.first-boot-done.bak') + + const movedAgentEntries: string[] = [] + const agentsDir = join(dataDir, 'agents') + if (existsSync(agentsDir)) { + for (const entry of readdirSync(agentsDir)) { + if (!entry || entry.startsWith('.')) continue + if (moveIntoBackup(join(agentsDir, entry), join('agents', entry))) { + movedAgentEntries.push(entry) + } + } + } + + const removedTeamRoles = moveIntoBackup(join(reflecttHome, 'TEAM-ROLES.yaml'), 'TEAM-ROLES.yaml.bak') + + const liveTasks = opts?.listTasks ? opts.listTasks() : (await getTaskManager()).listTasks({ includeTest: true }) + const deletedTaskIds: string[] = [] + for (const task of liveTasks) { + if (!task?.id) continue + const deleted = await deleteTask(task.id, actor) + if (deleted) deletedTaskIds.push(task.id) + } + + const removedBackupDir = !removedMarker && !removedTeamRoles && movedAgentEntries.length === 0 && deletedTaskIds.length === 0 + if (removedBackupDir) { + rmSync(backupDir, { recursive: true, force: true }) + } + + return { + backupDir, + removedMarker, + movedAgentEntries, + removedTeamRoles, + deletedTaskIds, + removedBackupDir, + } +} + // ── Redact sensitive values ────────────────────────────────────────── const SENSITIVE_KEYS = new Set([ @@ -262,6 +352,57 @@ export function registerManageRoutes(app: FastifyInstance, deps: { }, 500) }) + // POST /manage/reset-first-boot — destructive bootstrap reset for managed-host reproof + app.post('/manage/reset-first-boot', async (request, reply) => { + if (!checkManageAuth(request, reply, { allowHostCredential: true })) return + + const body = (request.body && typeof request.body === 'object') ? request.body as Record : {} + if (body.confirm !== FIRST_BOOT_RESET_CONFIRM) { + reply.code(400) + return { + error: `confirm must equal ${FIRST_BOOT_RESET_CONFIRM}`, + hint: 'This endpoint is destructive. It clears first-boot markers, moves agent state aside, deletes live tasks, and optionally restarts the host.', + } + } + + const restart = body.restart !== false + const reset = await resetFirstBootState() + const method = restart ? detectRestartMethod() : null + + if (restart && !method) { + reply.code(501) + return { + success: false, + error: 'Restart not supported in this environment', + hint: 'Reset succeeded, but restart is not supported here. Reboot the process manually or call this endpoint with { restart: false }.', + reset, + } + } + + reply.send({ + success: true, + reset, + restart: restart + ? { + scheduled: true, + method, + pid: process.pid, + message: `First-boot reset applied. Server will restart via ${method}.`, + } + : { scheduled: false }, + }) + + if (!restart || !method) return + + setTimeout(() => { + if (method === 'exit') { + process.exit(0) + } else if (method === 'sigterm') { + process.kill(process.pid, 'SIGTERM') + } + }, 500) + }) + // GET /manage/restart-context — read last restart snapshot (agents use on boot to resume) app.get('/manage/restart-context', async (request, reply) => { if (!checkManageAuth(request, reply)) return diff --git a/src/server.ts b/src/server.ts index fa8def1b..9201c573 100644 --- a/src/server.ts +++ b/src/server.ts @@ -14057,6 +14057,7 @@ If your heartbeat shows **no active task** and **no next task**: { method: 'GET', path: '/manage/config', hint: 'Config introspection (secrets redacted). Auth required.' }, { method: 'GET', path: '/manage/logs', hint: 'Bounded log tail. Query: level, since, limit, format=text. Auth required.' }, { method: 'POST', path: '/manage/restart', hint: 'Graceful restart (Docker/systemd/CLI). Auth required.' }, + { method: 'POST', path: '/manage/reset-first-boot', hint: 'Destructive reproof reset for managed hosts. Clears first-boot markers, moves agent state aside, deletes live tasks, and optionally restarts. Auth: manage token or managed host credential. Body must include { confirm: "RESET_FIRST_BOOT" }.' }, { method: 'GET', path: '/manage/disk', hint: 'Data directory sizes. Auth required.' }, ], }, diff --git a/tests/manage.test.ts b/tests/manage.test.ts new file mode 100644 index 00000000..a081b997 --- /dev/null +++ b/tests/manage.test.ts @@ -0,0 +1,77 @@ +import { describe, it, expect } from 'vitest' +import { mkdtempSync, mkdirSync, writeFileSync, existsSync, readdirSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { resetFirstBootState } from '../src/manage.js' + +describe('resetFirstBootState', () => { + it('moves first-boot artifacts into backup and deletes live tasks', async () => { + const root = mkdtempSync(join(tmpdir(), 'reflectt-reset-first-boot-')) + const reflecttHome = join(root, '.reflectt') + const dataDir = join(reflecttHome, 'data') + const agentsDir = join(dataDir, 'agents') + mkdirSync(agentsDir, { recursive: true }) + + writeFileSync(join(dataDir, '.first-boot-done'), 'done\n', 'utf-8') + writeFileSync(join(dataDir, 'TEAM_INTENT.md'), '# Team Intent\n', 'utf-8') + mkdirSync(join(agentsDir, 'main'), { recursive: true }) + mkdirSync(join(agentsDir, 'kai'), { recursive: true }) + writeFileSync(join(reflecttHome, 'TEAM-ROLES.yaml'), 'agents: []\n', 'utf-8') + + const deleted: string[] = [] + const summary = await resetFirstBootState({ + reflecttHome, + dataDir, + now: () => 12345, + listTasks: () => [{ id: 'task-1' }, { id: 'task-2' }], + deleteTask: async (taskId) => { + deleted.push(taskId) + return true + }, + }) + + expect(summary.removedMarker).toBe(true) + expect(summary.removedTeamRoles).toBe(true) + expect(summary.movedAgentEntries.sort()).toEqual(['kai', 'main']) + expect(summary.deletedTaskIds).toEqual(['task-1', 'task-2']) + expect(summary.removedBackupDir).toBe(false) + expect(summary.backupDir).toBe(join(dataDir, '_bootstrap_resets', 'reset-12345')) + + expect(existsSync(join(dataDir, '.first-boot-done'))).toBe(false) + expect(existsSync(join(reflecttHome, 'TEAM-ROLES.yaml'))).toBe(false) + expect(existsSync(join(dataDir, 'TEAM_INTENT.md'))).toBe(true) + expect(readdirSync(agentsDir)).toEqual([]) + expect(existsSync(join(summary.backupDir, 'data.first-boot-done.bak'))).toBe(true) + expect(existsSync(join(summary.backupDir, 'TEAM-ROLES.yaml.bak'))).toBe(true) + expect(existsSync(join(summary.backupDir, 'agents', 'main'))).toBe(true) + expect(existsSync(join(summary.backupDir, 'agents', 'kai'))).toBe(true) + + expect(deleted).toEqual(['task-1', 'task-2']) + + rmSync(root, { recursive: true, force: true }) + }) + + it('removes the empty backup directory when nothing needed resetting', async () => { + const root = mkdtempSync(join(tmpdir(), 'reflectt-reset-first-boot-')) + const reflecttHome = join(root, '.reflectt') + const dataDir = join(reflecttHome, 'data') + mkdirSync(dataDir, { recursive: true }) + + const summary = await resetFirstBootState({ + reflecttHome, + dataDir, + now: () => 999, + listTasks: () => [], + deleteTask: async () => true, + }) + + expect(summary.removedMarker).toBe(false) + expect(summary.removedTeamRoles).toBe(false) + expect(summary.movedAgentEntries).toEqual([]) + expect(summary.deletedTaskIds).toEqual([]) + expect(summary.removedBackupDir).toBe(true) + expect(existsSync(summary.backupDir)).toBe(false) + + rmSync(root, { recursive: true, force: true }) + }) +}) From 947b52b399157e340717889412318b737d1c4144 Mon Sep 17 00:00:00 2001 From: Kai Date: Tue, 21 Apr 2026 10:30:06 -0700 Subject: [PATCH 06/14] refactor: remove node roster voice maps --- src/canvas-interactive.ts | 29 +++++++++++++---------------- src/server.ts | 19 ++++--------------- src/tasks.ts | 14 -------------- 3 files changed, 17 insertions(+), 45 deletions(-) diff --git a/src/canvas-interactive.ts b/src/canvas-interactive.ts index 391b5ec9..5881afb4 100644 --- a/src/canvas-interactive.ts +++ b/src/canvas-interactive.ts @@ -516,22 +516,18 @@ export async function canvasInteractiveRoutes( const ELEVEN_BASE = 'https://api.elevenlabs.io/v1' const ELEVEN_API_KEY = process.env.ELEVEN_LABS_API_KEY || process.env.ELEVENLABS_API_KEY - const ELEVEN_VOICE_MAP: Record = { - link: 'pNInz6obpgDQGcFmaJgB', kai: 'ErXwobaYiN019PkySvjV', - pixel: 'MF3mGyEYCl7XYWbV9V6O', echo: 'jBpfuIE2acCO8z3wKNLl', - harmony: 'jBpfuIE2acCO8z3wKNLl', rhythm: 'onwK4e9ZLuTAKqWW03F9', - swift: 'yoZ06aMxZJJ28mfd3POQ', kotlin: 'SOYHLrjzK2X1ezoPC6cr', - sage: 'ThT5KcBeYPX3keUQqHPh', bookkeeper: 'GBv7mTt0atIp3Br8iCZE', - } - const KOKORO_VOICE_MAP: Record = { - link: 'af_sarah', swift: 'af_sarah', - kai: 'af_nicole', kotlin: 'af_nicole', pixel: 'af_nicole', echo: 'af_nicole', harmony: 'af_nicole', - rhythm: 'af_james', bookkeeper: 'bf_emma', sage: 'bf_emma', + const DEFAULT_KOKORO_VOICE_ID = process.env.KOKORO_DEFAULT_VOICE_ID || 'af_sarah' + const DEFAULT_ELEVENLABS_VOICE_ID = process.env.ELEVENLABS_DEFAULT_VOICE_ID || 'pNInz6obpgDQGcFmaJgB' + + function resolveVoiceId(requestedVoiceId: unknown, fallbackVoiceId: string): string { + return typeof requestedVoiceId === 'string' && requestedVoiceId.trim() + ? requestedVoiceId.trim() + : fallbackVoiceId } - async function makeTts(text: string, agentId: string): Promise<{ url: string; ms: number } | null> { - const kokoroVoice = KOKORO_VOICE_MAP[agentId] || 'af_sarah' - const elevenVoice = ELEVEN_VOICE_MAP[agentId] || 'pNInz6obpgDQGcFmaJgB' + async function makeTts(text: string, requestedVoiceId?: string): Promise<{ url: string; ms: number } | null> { + const kokoroVoice = resolveVoiceId(requestedVoiceId, DEFAULT_KOKORO_VOICE_ID) + const elevenVoice = resolveVoiceId(requestedVoiceId, DEFAULT_ELEVENLABS_VOICE_ID) const key = await hashTts(text, kokoroVoice) const cached = ttsCache.get(key) if (cached && Date.now() - cached.ts < TTS_TTL) return { url: '/audio/' + key, ms: Math.round(text.length * 50) } @@ -602,10 +598,11 @@ export async function canvasInteractiveRoutes( const text = typeof body?.text === 'string' ? body.text.trim() : '' const agentId = typeof body?.agentId === 'string' ? body.agentId : 'unknown' const agentName = typeof body?.agentName === 'string' ? body.agentName : agentId + const requestedVoiceId = typeof body?.voiceId === 'string' ? body.voiceId.trim() : '' if (!text || text.length > 1000) return { error: 'text required, max 1000' } // Deterministic ID so clients can match queued → ready events - const voiceId = await hashTts(text, agentId) + const voiceId = await hashTts(text, requestedVoiceId || agentId) const estimatedMs = Math.round(text.length * 50) // Emit voice_queued immediately so UI can show "speaking" state @@ -615,7 +612,7 @@ export async function canvasInteractiveRoutes( } // Generate audio in background — do not await - makeTts(text, agentId).then(async (result) => { + makeTts(text, requestedVoiceId || undefined).then(async (result) => { if (!result) return // Kokoro + ElevenLabs both failed // Re-emit with audio URL so clients can play const event = { type: 'voice_output', voiceId, text, url: result.url, agentId, agentName, durationMs: result.ms } diff --git a/src/server.ts b/src/server.ts index 9201c573..cd7323e1 100644 --- a/src/server.ts +++ b/src/server.ts @@ -8169,15 +8169,7 @@ export async function createServer(): Promise { return `Received: "${text.slice(0, 80)}${text.length > 80 ? '...' : ''}"` } - // Agent voice IDs (ElevenLabs) — per-agent identity, same mapping as cloud - const NODE_AGENT_VOICE_IDS: Record = { - link: 'pNInz6obpgDQGcFmaJgB', // Adam - kai: 'onwK4e9ZLuTAKqWW03F9', // Daniel - pixel: 'EXAVITQu4vr4xnSDxMaL', // Sarah - sage: 'yoZ06aMxZJJ28mfd3POQ', // Rachel - scout: '3XbDmaS0mwj3WIVTUxWa', // Charlie - echo: 'MF3mGyEYCl7XYWbV9V6O', // Elli - } + const DEFAULT_ELEVENLABS_VOICE_ID = process.env.ELEVENLABS_DEFAULT_VOICE_ID || 'pNInz6obpgDQGcFmaJgB' // ── Voice mutex — only one agent speaks at a time ────────────────── // P0 fix: multiple agents were triggering TTS simultaneously, causing @@ -8232,7 +8224,7 @@ export async function createServer(): Promise { }) if (!elevenKey) return null - const voiceId = NODE_AGENT_VOICE_IDS[forAgentId] ?? NODE_AGENT_VOICE_IDS['link'] + const voiceId = DEFAULT_ELEVENLABS_VOICE_ID try { const res = await fetch( `https://api.elevenlabs.io/v1/text-to-speech/${voiceId}`, @@ -8426,10 +8418,7 @@ export async function createServer(): Promise { return `Received: "${text.slice(0, 80)}${text.length > 80 ? '...' : ''}"` } - const NODE_AGENT_VOICE_IDS_AUDIO: Record = { - link: 'pNInz6obpgDQGcFmaJgB', kai: 'onwK4e9ZLuTAKqWW03F9', pixel: 'EXAVITQu4vr4xnSDxMaL', - sage: 'yoZ06aMxZJJ28mfd3POQ', scout: '3XbDmaS0mwj3WIVTUxWa', echo: 'MF3mGyEYCl7XYWbV9V6O', - } + const DEFAULT_ELEVENLABS_VOICE_ID_AUDIO = process.env.ELEVENLABS_DEFAULT_VOICE_ID || 'pNInz6obpgDQGcFmaJgB' const synthesizeTtsAudio = async (text: string, forAgentId: string): Promise => { const elevenKey = process.env.ELEVEN_LABS_API_KEY || process.env.ELEVENLABS_API_KEY // canvas_expression fires whether or not ElevenLabs is configured @@ -8451,7 +8440,7 @@ export async function createServer(): Promise { }, }) if (!elevenKey) return null - const voiceId = NODE_AGENT_VOICE_IDS_AUDIO[forAgentId] ?? NODE_AGENT_VOICE_IDS_AUDIO['link'] + const voiceId = DEFAULT_ELEVENLABS_VOICE_ID_AUDIO try { const res = await fetch(`https://api.elevenlabs.io/v1/text-to-speech/${voiceId}`, { method: 'POST', diff --git a/src/tasks.ts b/src/tasks.ts index 1d568469..f5667603 100644 --- a/src/tasks.ts +++ b/src/tasks.ts @@ -19,18 +19,6 @@ import { getAgentAliases } from './assignment.js' import { getAgentLane } from './lane-config.js' import type Database from 'better-sqlite3' -// Voice IDs for agents (used for TTS) -const VOICE_IDS: Record = { - link: 'pNInz6obpgDQGcFmaJgB', - kai: 'onwK4e9ZLuTAKqWW03F9', - pixel: 'EXAVITQu4vr4xnSDxMaL', - sage: 'yoZ06aMxZJJ28mfd3POQ', - scout: '3XbDmaS0mwj3WIVTUxWa', - echo: 'MF3mGyEYCl7XYWbV9V6O', - rhythm: 'morgan', - spark: 'corey', -} - const TASKS_FILE = join(DATA_DIR, 'tasks.jsonl') const LEGACY_TASKS_FILE = join(LEGACY_DATA_DIR, 'tasks.jsonl') const RECURRING_TASKS_FILE = join(DATA_DIR, 'tasks.recurring.jsonl') @@ -367,7 +355,6 @@ class TaskManager { for (const { assignee, title } of doingTasks) { if (!assignee) continue const work = title?.slice(0, 50) ?? 'Working...' - const voiceId = VOICE_IDS[assignee] // Auto-expression: triggers TTS audio (local SSE) eventBus.emit({ id: `think-${Date.now()}-${Math.random().toString(36).substr(2, 6)}`, @@ -377,7 +364,6 @@ class TaskManager { kind: 'auto_expression' as const, agentId: assignee, line: work, - voiceId, intensity: 0.3, }, }) From 28d0328b57a7c0dd5451d026ac596e16cbfe14e0 Mon Sep 17 00:00:00 2001 From: Kai Date: Sun, 26 Apr 2026 19:59:41 -0700 Subject: [PATCH 07/14] Disable standalone voice reply API --- src/server.ts | 402 ++------------------------------------------------ 1 file changed, 9 insertions(+), 393 deletions(-) diff --git a/src/server.ts b/src/server.ts index cd7323e1..ae3bb5fb 100644 --- a/src/server.ts +++ b/src/server.ts @@ -173,7 +173,7 @@ import { runPrLinkReconcileSweep } from './pr-link-reconciler.js' import { createOverride, getOverride, listOverrides, findActiveOverride, validateOverrideInput, tickOverrideLifecycle, type CreateOverrideInput } from './routing-override.js' import { getRoutingApprovalQueue, getRoutingSuggestion, buildApprovalPatch, buildRejectionPatch, buildRoutingSuggestionPatch, isRoutingApproval } from './routing-approvals.js' import { simulateRoutingScenarios, type CommsRoutingPolicy, type RoutingScenario } from './comms-routing-policy.js' -import { createVoiceSession, getVoiceSession, processVoiceTranscript, subscribeVoiceSession } from './voice-sessions.js' +import { getVoiceSession, subscribeVoiceSession } from './voice-sessions.js' import { createRun, getRun, subscribeRun, approveRun, rejectRun, executeGithubIssueCreate, executeMacOSUIAction, buildReplayPacket, listPendingRuns, listRuns } from './agent-interface.js' import { validateIntent as macOSValidateIntent, isKillSwitchEngaged, engageKillSwitch, resetKillSwitch } from './macos-accessibility.js' import { calendarManager, type BlockType, type CreateBlockInput, type UpdateBlockInput } from './calendar.js' @@ -8069,403 +8069,19 @@ export async function createServer(): Promise { // ── Voice API ────────────────────────────────────────────────────────────── - // POST /voice/input — create a voice session + begin processing + // POST /voice/input — disabled until the voice loop is routed through real agents. // Body: { agentId: string, transcript?: string } - // Returns: { sessionId } - app.post('/voice/input', async (request, reply) => { - const body = request.body as Record - const agentId = typeof body.agentId === 'string' ? body.agentId.trim() : '' - const transcript = typeof body.transcript === 'string' ? body.transcript.trim() : '' - - if (!agentId) { - reply.status(400) - return { success: false, message: 'agentId is required' } - } - if (!transcript) { - reply.status(400) - return { success: false, message: 'transcript is required (audio STT not yet supported)' } - } - if (transcript.length > 4000) { - reply.status(400) - return { success: false, message: 'transcript too long (max 4000 chars)' } - } - - const session = createVoiceSession(agentId) - - // Helper: push activeSpeaker signal into canvas state so orb reacts - const setActiveSpeaker = (active: boolean) => { - const existing = canvasStateMap.get(agentId) - if (existing) { - canvasStateMap.set(agentId, { - ...existing, - payload: { ...(existing.payload as Record), activeSpeaker: active }, - updatedAt: Date.now(), - }) - eventBus.emit({ - id: `crender-voice-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`, - type: 'canvas_render' as const, - timestamp: Date.now(), - data: { - state: existing.state, - sensors: existing.sensors, - agentId, - payload: { ...(existing.payload as Record), activeSpeaker: active }, - presence: { - name: agentId, - identityColor: AGENT_IDENTITY_COLORS[agentId] ?? '#9ca3af', - state: (existing.payload as any)?.presenceState ?? 'working', - activeSpeaker: active, - activeTask: (existing.payload as any)?.activeTask, - recency: 'just now', - }, - }, - }) - requestImmediateCanvasSync() - } - } - - // Build agent system context for the LLM responder - const agentRole = getAgentRole(agentId) - const agentSystemPrompt = agentRole - ? `You are ${agentId}, a ${agentRole.role ?? 'team agent'} on Team Reflectt. ${agentRole.description ?? ''} Respond concisely — your reply will be spoken aloud. 1-3 sentences max.` - : `You are ${agentId}, a team agent. Respond concisely — your reply will be spoken aloud. 1-3 sentences max.` - - // Kick off async processing — do not await so we return sessionId immediately - const agentResponder = async (respAgentId: string, text: string, _sessionId: string): Promise => { - setActiveSpeaker(false) - - // Try real LLM call if ANTHROPIC_API_KEY is set - const anthropicKey = process.env.ANTHROPIC_API_KEY - if (anthropicKey) { - try { - const resp = await fetch('https://api.anthropic.com/v1/messages', { - method: 'POST', - headers: { - 'x-api-key': anthropicKey, - 'anthropic-version': '2023-06-01', - 'content-type': 'application/json', - }, - body: JSON.stringify({ - model: 'claude-haiku-4-5', - max_tokens: 256, - system: agentSystemPrompt, - messages: [{ role: 'user', content: text }], - }), - signal: AbortSignal.timeout(15000), - }) - if (resp.ok) { - const data = await resp.json() as { content?: Array<{ text?: string }> } - const reply = data.content?.[0]?.text?.trim() - if (reply) return reply - } - } catch (err) { - console.error(`[voice] LLM call failed for ${respAgentId}:`, err) - // fall through to stub - } - } - - // Stub fallback — always available, no key required - await new Promise(resolve => setTimeout(resolve, 400)) - return `Received: "${text.slice(0, 80)}${text.length > 80 ? '...' : ''}"` - } - - const DEFAULT_ELEVENLABS_VOICE_ID = process.env.ELEVENLABS_DEFAULT_VOICE_ID || 'pNInz6obpgDQGcFmaJgB' - - // ── Voice mutex — only one agent speaks at a time ────────────────── - // P0 fix: multiple agents were triggering TTS simultaneously, causing - // overlapping audio on the canvas. Queue ensures serial playback. - // task-1773686058943-v17yrucjr - const voiceQueue: Array<{ text: string; agentId: string; resolve: (v: string | null) => void }> = [] - let voiceSpeaking = false - - const processVoiceQueue = async () => { - if (voiceSpeaking || voiceQueue.length === 0) return - voiceSpeaking = true - const item = voiceQueue.shift()! - try { - const result = await synthesizeTtsInternal(item.text, item.agentId) - item.resolve(result) - } catch { - item.resolve(null) - } - voiceSpeaking = false - if (voiceQueue.length > 0) setTimeout(processVoiceQueue, 500) - } - - // Synthesize TTS via ElevenLabs if key is set - const synthesizeTts = async (text: string, forAgentId: string): Promise => { - return new Promise((resolve) => { - voiceQueue.push({ text, agentId: forAgentId, resolve }) - processVoiceQueue() - }) - } - - const synthesizeTtsInternal = async (text: string, forAgentId: string): Promise => { - const elevenKey = process.env.ELEVEN_LABS_API_KEY || process.env.ELEVENLABS_API_KEY - - // Fire canvas_expression alongside TTS — the room responds when an agent speaks. - // Non-blocking: emit first, synthesize in parallel. - const IDENTITY_COLORS: Record = { - link: '#60a5fa', kai: '#fb923c', pixel: '#a78bfa', - sage: '#34d399', scout: '#fbbf24', echo: '#f472b6', - } - eventBus.emit({ - id: `voice-expr-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`, - type: 'canvas_expression' as const, - timestamp: Date.now(), - data: { - agentId: forAgentId, - channels: { - voice: text.slice(0, 300), - visual: { flash: IDENTITY_COLORS[forAgentId] ?? '#60a5fa', particles: 'surge' }, - narrative: `${forAgentId} responds`, - }, - }, - }) - - if (!elevenKey) return null - const voiceId = DEFAULT_ELEVENLABS_VOICE_ID - try { - const res = await fetch( - `https://api.elevenlabs.io/v1/text-to-speech/${voiceId}`, - { - method: 'POST', - headers: { - 'xi-api-key': elevenKey, - 'Content-Type': 'application/json', - 'Accept': 'audio/mpeg', - }, - body: JSON.stringify({ text: text.slice(0, 500), model_id: 'eleven_monolingual_v1', voice_settings: { stability: 0.5, similarity_boost: 0.75 } }), - signal: AbortSignal.timeout(20000), - } - ) - if (!res.ok) return null - const buf = Buffer.from(await res.arrayBuffer()) - // Return as data URI so the client can play it without a second request - return `data:audio/mpeg;base64,${buf.toString('base64')}` - } catch { - return null - } - } - - // Subscribe to voice events to drive canvas state - const unsubVoice = subscribeVoiceSession(session.id, (event) => { - if (event.type === 'agent.thinking') { - // Agent is processing — keep existing canvas state (thinking is already set via presence) - } else if (event.type === 'tts.ready') { - // Agent is now speaking — activate orb waveform/scale - setActiveSpeaker(true) - } else if (event.type === 'session.end' || event.type === 'error') { - // Clear speaker state - setActiveSpeaker(false) - unsubVoice() - } - }) - - processVoiceTranscript(session.id, transcript, agentResponder, synthesizeTts).catch(err => { - console.error('[voice] processVoiceTranscript error:', err) - unsubVoice() - }) + app.post('/voice/input', async (_request, reply) => { + reply.status(410) + return { success: false, message: 'Voice conversation is disabled until routed through real agents.' } - return { success: true, sessionId: session.id } }) - // POST /voice/audio — accept an audio blob, transcribe via STT, pipe to voice pipeline - // Completes the full speak→STT→LLM→TTS loop. + // POST /voice/audio — disabled until the voice loop is routed through real agents. // Form fields: agentId (string), audio (file: wav/mp3/webm/ogg/m4a) - // Returns: { sessionId } - app.post('/voice/audio', async (request, reply) => { - let agentId = '' - let audioBuffer: Buffer | null = null - let audioMimeType = 'audio/webm' - - try { - const parts = (request as any).parts() - for await (const part of parts) { - if (part.type === 'field' && part.fieldname === 'agentId') { - agentId = String(part.value ?? '').trim() - } else if (part.type === 'file' && part.fieldname === 'audio') { - audioMimeType = part.mimetype ?? 'audio/webm' - const chunks: Buffer[] = [] - for await (const chunk of part.file) chunks.push(chunk) - audioBuffer = Buffer.concat(chunks) - } - } - } catch { - reply.status(400); return { success: false, message: 'Invalid multipart body' } - } - - if (!agentId) { reply.status(400); return { success: false, message: 'agentId is required' } } - if (!audioBuffer || audioBuffer.length === 0) { reply.status(400); return { success: false, message: 'audio file is required' } } - if (audioBuffer.length > 25 * 1024 * 1024) { reply.status(413); return { success: false, message: 'Audio exceeds 25MB limit' } } - - // Transcribe audio — priority: local whisper.cpp → OpenAI Whisper cloud → 503 - // Local whisper runs on-device (no API key, ~1.8s for tiny model on Apple Silicon) - let transcript = '' - let sttProvider = 'none' - - // 1. Try local whisper (no API key needed) - if (await isLocalWhisperAvailable()) { - try { - const localResult = await transcribeLocally(audioBuffer, audioMimeType) - if (localResult) { - transcript = localResult - sttProvider = 'local-whisper' - } - } catch (err) { - console.error('[voice/audio] local-whisper failed, trying cloud fallback:', err instanceof Error ? err.message : err) - } - } - - // 2. Fall back to OpenAI Whisper cloud if local failed or unavailable - if (!transcript) { - const openAiKey = process.env.OPENAI_API_KEY - if (openAiKey) { - try { - const ext = audioMimeType.includes('mp4') || audioMimeType.includes('m4a') ? 'm4a' - : audioMimeType.includes('mp3') ? 'mp3' - : audioMimeType.includes('ogg') ? 'ogg' - : audioMimeType.includes('wav') ? 'wav' - : 'webm' - - const form = new FormData() - form.append('file', new Blob([audioBuffer], { type: audioMimeType }), `audio.${ext}`) - form.append('model', 'whisper-1') - form.append('language', 'en') - - const res = await fetch('https://api.openai.com/v1/audio/transcriptions', { - method: 'POST', - headers: { Authorization: `Bearer ${openAiKey}` }, - body: form, - signal: AbortSignal.timeout(30000), - }) - - if (res.ok) { - const data = await res.json() as { text?: string } - transcript = data.text?.trim() ?? '' - if (transcript) sttProvider = 'openai-whisper' - } else { - const err = await res.text() - console.error(`[voice/audio] OpenAI Whisper error ${res.status}: ${err.slice(0, 200)}`) - } - } catch (err) { - console.error('[voice/audio] OpenAI Whisper error:', err) - } - } - } - - if (!transcript) { - reply.status(503) - return { success: false, message: 'STT unavailable — install openai-whisper locally or set OPENAI_API_KEY' } - } - - console.log(`[voice/audio] STT via ${sttProvider}: "${transcript.slice(0, 80)}"`) - - - if (transcript.length > 4000) transcript = transcript.slice(0, 4000) - - // Emit canvas_message so pulse SSE subscribers (browser, Android) get the transcript immediately - eventBus.emit({ - id: `cmsg-voice-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, - type: 'canvas_message' as const, - timestamp: Date.now(), - data: { - type: 'voice_transcript', - agentId, - agentColor: AGENT_IDENTITY_COLORS[agentId] ?? '#9ca3af', - transcript, - sttProvider, - }, - }) - - // Delegate to the same voice pipeline as POST /voice/input - // Inline the pipeline logic (mirrors /voice/input handler) - const session = createVoiceSession(agentId) - const agentRole = getAgentRole(agentId) - const agentSystemPrompt = agentRole - ? `You are ${agentId}, a ${agentRole.role ?? 'team agent'} on Team Reflectt. ${agentRole.description ?? ''} Respond concisely — your reply will be spoken aloud. 1-3 sentences max.` - : `You are ${agentId}, a team agent. Respond concisely — your reply will be spoken aloud. 1-3 sentences max.` - - const identityColor = AGENT_IDENTITY_COLORS[agentId] ?? '#9ca3af' - - const setActiveSpeakerAudio = (active: boolean) => { - const existing = canvasStateMap.get(agentId) - if (existing) { - canvasStateMap.set(agentId, { ...existing, payload: { ...(existing.payload as Record), activeSpeaker: active }, updatedAt: Date.now() }) - requestImmediateCanvasSync() - } - } - - const agentResponder = async (respAgentId: string, text: string, _sessionId: string): Promise => { - setActiveSpeakerAudio(false) - const anthropicKey = process.env.ANTHROPIC_API_KEY - if (anthropicKey) { - try { - const resp = await fetch('https://api.anthropic.com/v1/messages', { - method: 'POST', - headers: { 'x-api-key': anthropicKey, 'anthropic-version': '2023-06-01', 'content-type': 'application/json' }, - body: JSON.stringify({ model: 'claude-haiku-4-5', max_tokens: 256, system: agentSystemPrompt, messages: [{ role: 'user', content: text }] }), - signal: AbortSignal.timeout(15000), - }) - if (resp.ok) { - const data = await resp.json() as { content?: Array<{ text?: string }> } - const reply2 = data.content?.[0]?.text?.trim() - if (reply2) return reply2 - } - } catch (err) { console.error(`[voice/audio] LLM call failed:`, err) } - } - await new Promise(resolve => setTimeout(resolve, 400)) - return `Received: "${text.slice(0, 80)}${text.length > 80 ? '...' : ''}"` - } - - const DEFAULT_ELEVENLABS_VOICE_ID_AUDIO = process.env.ELEVENLABS_DEFAULT_VOICE_ID || 'pNInz6obpgDQGcFmaJgB' - const synthesizeTtsAudio = async (text: string, forAgentId: string): Promise => { - const elevenKey = process.env.ELEVEN_LABS_API_KEY || process.env.ELEVENLABS_API_KEY - // canvas_expression fires whether or not ElevenLabs is configured - const IDENTITY_COLORS_AUDIO: Record = { - link: '#60a5fa', kai: '#fb923c', pixel: '#a78bfa', - sage: '#34d399', scout: '#fbbf24', echo: '#f472b6', - } - eventBus.emit({ - id: `voice-expr-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`, - type: 'canvas_expression' as const, - timestamp: Date.now(), - data: { - agentId: forAgentId, - channels: { - voice: text.slice(0, 300), - visual: { flash: IDENTITY_COLORS_AUDIO[forAgentId] ?? '#60a5fa', particles: 'surge' }, - narrative: `${forAgentId} responds`, - }, - }, - }) - if (!elevenKey) return null - const voiceId = DEFAULT_ELEVENLABS_VOICE_ID_AUDIO - try { - const res = await fetch(`https://api.elevenlabs.io/v1/text-to-speech/${voiceId}`, { - method: 'POST', - headers: { 'xi-api-key': elevenKey, 'Content-Type': 'application/json', 'Accept': 'audio/mpeg' }, - body: JSON.stringify({ text: text.slice(0, 500), model_id: 'eleven_monolingual_v1', voice_settings: { stability: 0.5, similarity_boost: 0.75 } }), - signal: AbortSignal.timeout(20000), - }) - if (!res.ok) return null - const buf = Buffer.from(await res.arrayBuffer()) - return `data:audio/mpeg;base64,${buf.toString('base64')}` - } catch { return null } - } - - const unsubVoiceAudio = subscribeVoiceSession(session.id, (event) => { - if (event.type === 'tts.ready') setActiveSpeakerAudio(true) - else if (event.type === 'session.end' || event.type === 'error') { setActiveSpeakerAudio(false); unsubVoiceAudio() } - }) - - processVoiceTranscript(session.id, transcript, agentResponder, synthesizeTtsAudio).catch(err => { - console.error('[voice/audio] processVoiceTranscript error:', err) - unsubVoiceAudio() - }) - - void identityColor // suppress unused warning - return reply.code(201).send({ success: true, sessionId: session.id, transcript }) + app.post('/voice/audio', async (_request, reply) => { + reply.status(410) + return { success: false, message: 'Voice conversation is disabled until routed through real agents.' } }) // GET /voice/session/:id/events — SSE stream of voice pipeline state events From 1cad82cd896c1eea94b0f4436c813412129f147c Mon Sep 17 00:00:00 2001 From: Kai Date: Mon, 27 Apr 2026 19:02:22 -0700 Subject: [PATCH 08/14] Remove stale bundled reflectt channel plugin --- plugins/reflectt-channel/README.md | 96 -- plugins/reflectt-channel/index.ts | 840 ------------------ plugins/reflectt-channel/openclaw.plugin.json | 23 - plugins/reflectt-channel/package.json | 23 - plugins/reflectt-channel/src/channel.ts | 433 --------- plugins/reflectt-channel/src/types.ts | 29 - process/reflectt-channel-plugin.md | 13 - 7 files changed, 1457 deletions(-) delete mode 100644 plugins/reflectt-channel/README.md delete mode 100644 plugins/reflectt-channel/index.ts delete mode 100644 plugins/reflectt-channel/openclaw.plugin.json delete mode 100644 plugins/reflectt-channel/package.json delete mode 100644 plugins/reflectt-channel/src/channel.ts delete mode 100644 plugins/reflectt-channel/src/types.ts delete mode 100644 process/reflectt-channel-plugin.md diff --git a/plugins/reflectt-channel/README.md b/plugins/reflectt-channel/README.md deleted file mode 100644 index 2f7b3a41..00000000 --- a/plugins/reflectt-channel/README.md +++ /dev/null @@ -1,96 +0,0 @@ -# reflectt-channel - -OpenClaw channel plugin that connects agents to reflectt-node via SSE (Server-Sent Events). - -## What it does - -1. Connects to reflectt-node's `/events` SSE endpoint -2. Listens for chat messages with @mentions -3. Routes mentioned agents through OpenClaw's inbound pipeline -4. Posts agent responses back to reflectt-node via `POST /chat/messages` - -## Install - -From the reflectt-node repo root: - -```bash -openclaw plugins install ./plugins/reflectt-channel -``` - -Or from anywhere: - -```bash -openclaw plugins install /path/to/reflectt-node/plugins/reflectt-channel -``` - -## Configure - -Add to `~/.openclaw/openclaw.json`. Two config paths are supported: - -**Option 1 — `channels.reflectt` (recommended):** - -```json -{ - "channels": { - "reflectt": { - "enabled": true, - "url": "http://127.0.0.1:4445" - } - } -} -``` - -**Option 2 — `plugins.entries` (general plugin convention):** - -```json -{ - "plugins": { - "entries": { - "reflectt-channel": { - "config": { - "url": "http://127.0.0.1:4445" - } - } - } - } -} -``` - -> **Precedence:** `channels.reflectt` takes priority over `plugins.entries`. If both are set, `channels.reflectt.url` wins. - -> **Default:** If neither is configured, the plugin falls back to `http://127.0.0.1:4445` and logs a warning with the exact config keys to set. - -Or use the CLI shorthand: - -```bash -openclaw config set channels.reflectt.enabled true -openclaw config set channels.reflectt.url "http://127.0.0.1:4445" -``` - -Then restart the gateway: - -```bash -openclaw gateway restart -``` - -## Verify - -```bash -openclaw plugins list # Should show reflectt-channel -openclaw plugins info reflectt-channel -``` - -## Message flow - -``` -reflectt-node (SSE /events) - → reflectt-channel plugin detects @mention - → OpenClaw routes to agent session - → Agent responds - → Plugin POSTs to reflectt-node /chat/messages -``` - -## Requirements - -- OpenClaw gateway running -- reflectt-node running at the configured URL diff --git a/plugins/reflectt-channel/index.ts b/plugins/reflectt-channel/index.ts deleted file mode 100644 index 3e30dbc3..00000000 --- a/plugins/reflectt-channel/index.ts +++ /dev/null @@ -1,840 +0,0 @@ -/** - * reflectt-channel — OpenClaw channel plugin - * - * Connects to reflectt-node SSE. When a message @mentions an agent, - * it routes through OpenClaw's inbound pipeline. Agent responses are - * POSTed back to reflectt-node automatically. - */ -import type { OpenClawPluginApi, ChannelPlugin, OpenClawConfig } from "openclaw/plugin-sdk"; -import { DEFAULT_ACCOUNT_ID, buildChannelConfigSchema } from "openclaw/plugin-sdk"; -import http from "node:http"; -import fs from "node:fs"; -import os from "node:os"; -import path from "node:path"; - -const DEFAULT_URL = "http://127.0.0.1:4445"; - -// Agent roster — loaded dynamically from reflectt-node /team/roles on connect and refreshed -// every 5 minutes. No static fallback: if discovery fails the roster stays empty and a warning -// is logged so the operator can investigate rather than silently routing to stale agent names. -const FALLBACK_AGENTS: string[] = []; -let WATCHED_AGENTS: readonly string[] = FALLBACK_AGENTS; -let WATCHED_SET = new Set(WATCHED_AGENTS); -const IDLE_NUDGE_WINDOW_MS = 15 * 60 * 1000; // 15m -const WATCHDOG_INTERVAL_MS = 60 * 1000; // 1m -const ROSTER_REFRESH_INTERVAL_MS = 5 * 60 * 1000; // 5m -const ESCALATION_COOLDOWN_MS = 20 * 60 * 1000; - -// SSE reconnect config -const SSE_INITIAL_RETRY_MS = 1000; // start at 1s -const SSE_MAX_RETRY_MS = 30_000; // cap at 30s -const SSE_SOCKET_TIMEOUT_MS = 30_000; // detect dead TCP after 30s silence -const SSE_HEALTH_INTERVAL_MS = 15_000; // health-check ping every 15s - -// Gateway restart notification: dedup window to prevent duplicate -// system broadcasts when SSE reconnects rapidly (e.g. server flap). -const RESTART_NOTIFY_DEDUP_MS = 60_000; // 60s -let lastRestartNotifyAt = 0; - -const lastUpdateByAgent = new Map(); -const lastEscalationAt = new Map(); -const hasActiveTaskByAgent = new Map(); -const TASK_CACHE_TTL_MS = 2 * 60 * 1000; - -// --- Config helpers --- - -interface ReflecttAccount { - accountId: string; - url: string; - enabled: boolean; - configured: boolean; -} - -function purgeSessionIndexEntry(agentId: string, sessionKey: string, ctx: any): boolean { - try { - const storePath = path.join(os.homedir(), ".openclaw", "agents", agentId, "sessions", "sessions.json"); - if (!fs.existsSync(storePath)) return false; - - const raw = fs.readFileSync(storePath, "utf8"); - const data = JSON.parse(raw || "{}"); - const keyExact = sessionKey; - const keyLower = sessionKey.toLowerCase(); - - if (!Object.prototype.hasOwnProperty.call(data, keyExact) && !Object.prototype.hasOwnProperty.call(data, keyLower)) { - return false; - } - - delete data[keyExact]; - delete data[keyLower]; - fs.writeFileSync(storePath, `${JSON.stringify(data, null, 2)}\n`, "utf8"); - ctx.log?.warn(`[reflectt] Purged stale session index entry for ${sessionKey} at ${storePath}`); - return true; - } catch (err) { - ctx.log?.error(`[reflectt] Failed to purge session entry for ${sessionKey}: ${err}`); - return false; - } -} - -function resolveAccount(cfg: OpenClawConfig, accountId?: string | null, log?: any): ReflecttAccount { - // Support both config paths: - // 1. channels.reflectt.url (canonical — per OpenClaw channel plugin convention) - // 2. plugins.entries.reflectt-channel.config.url (fallback — general plugin convention) - // channels.reflectt takes precedence. - const ch = (cfg as any)?.channels?.reflectt ?? {}; - const pluginCfg = (cfg as any)?.plugins?.entries?.["reflectt-channel"]?.config ?? {}; - - const hasChannelConfig = !!ch.url; - const hasPluginConfig = !!pluginCfg.url; - const url = ch.url || pluginCfg.url || DEFAULT_URL; - const enabled = ch.enabled !== undefined ? ch.enabled !== false - : pluginCfg.enabled !== undefined ? pluginCfg.enabled !== false - : true; - const configured = hasChannelConfig || hasPluginConfig; - - if (!configured) { - log?.warn( - `[reflectt] No explicit URL configured — using default ${DEFAULT_URL}. ` + - `To configure, set one of:\n` + - ` 1. channels.reflectt.url in ~/.openclaw/openclaw.json (recommended)\n` + - ` 2. plugins.entries.reflectt-channel.config.url in ~/.openclaw/openclaw.json\n` + - ` Or run: openclaw config set channels.reflectt.url "http://your-node:4445"` - ); - } else if (hasChannelConfig && hasPluginConfig && ch.url !== pluginCfg.url) { - log?.warn( - `[reflectt] Config found in both channels.reflectt.url (${ch.url}) and ` + - `plugins.entries.reflectt-channel.config.url (${pluginCfg.url}). ` + - `Using channels.reflectt.url (takes precedence).` - ); - } - - return { - accountId: accountId || DEFAULT_ACCOUNT_ID, - url, - enabled, - configured, - }; -} - -// --- HTTP helpers --- - -function postMessage(url: string, from: string, channel: string, content: string): Promise { - return new Promise((resolve, reject) => { - const body = JSON.stringify({ from, channel, content }); - const parsed = new URL(`${url}/chat/messages`); - const req = http.request({ - hostname: parsed.hostname, - port: parsed.port, - path: parsed.pathname, - method: "POST", - headers: { "Content-Type": "application/json", "Content-Length": Buffer.byteLength(body) }, - }, (res) => { res.resume(); resolve(); }); - req.on("error", reject); - req.end(body); - }); -} - -function normalizeSenderId(value: unknown): string | null { - if (typeof value !== "string") return null; - const id = value.trim().toLowerCase().replace(/^@+/, ""); - return id.length > 0 ? id : null; -} - -function markAgentActivity(from: unknown, channel: unknown, timestamp: unknown) { - if (channel !== "general") return; - const id = normalizeSenderId(from); - if (!id || !WATCHED_SET.has(id)) return; - const ts = - typeof timestamp === "number" && Number.isFinite(timestamp) - ? timestamp - : typeof timestamp === "string" && Number.isFinite(Number(timestamp)) - ? Number(timestamp) - : Date.now(); - const cur = lastUpdateByAgent.get(id) ?? 0; - if (ts > cur) lastUpdateByAgent.set(id, ts); -} - -function shouldEscalate(key: string, now: number): boolean { - const last = lastEscalationAt.get(key) ?? 0; - if (now - last < ESCALATION_COOLDOWN_MS) return false; - lastEscalationAt.set(key, now); - return true; -} - -async function refreshAgentRoster(url: string, log?: any): Promise { - // /team/roles is the canonical source: { agents: [{ name: "kai", role: "builder", ... }] } - // Fall back to legacy endpoints if /team/roles is unavailable. - const endpoints = [ - { path: '/team/roles', extract: (d: any) => (d?.agents || []).map((a: any) => a?.name).filter(Boolean) }, - { path: '/health/agents', extract: (d: any) => (d?.agents || []).map((a: any) => a?.agent || a?.name).filter(Boolean) }, - { path: '/capabilities', extract: (d: any) => (d?.assignment?.agents || []).map((a: any) => a?.name).filter(Boolean) }, - ]; - - for (const ep of endpoints) { - try { - const res = await fetch(`${url}${ep.path}`, { signal: AbortSignal.timeout(5000) }); - if (res.ok) { - const data = await res.json() as any; - const names: string[] = ep.extract(data); - if (names.length > 0) { - WATCHED_AGENTS = Object.freeze(names); - WATCHED_SET = new Set(names); - log?.info(`[reflectt] Loaded ${names.length} agents from ${ep.path}: ${names.join(', ')}`); - return; - } - } - } catch { /* try next endpoint */ } - } - log?.warn( - `[reflectt] Could not load agent roster from ${url} — ` + - (FALLBACK_AGENTS.length > 0 - ? `using fallback list (${FALLBACK_AGENTS.join(', ')})` - : "roster will be empty until next successful refresh") - ); -} - -async function fetchRecentMessages(url: string): Promise>> { - const endpoints = ["/chat/messages?limit=500", "/chat/messages", "/messages"]; - for (const endpoint of endpoints) { - try { - const response = await fetch(`${url}${endpoint}`); - if (!response.ok) continue; - const data = await response.json(); - const messages = Array.isArray(data) - ? data - : data && typeof data === "object" && Array.isArray((data as { messages?: unknown[] }).messages) - ? (data as { messages: unknown[] }).messages - : []; - return messages.filter((m): m is Record => Boolean(m) && typeof m === "object"); - } catch { - // best effort - } - } - return []; -} - -async function seedAgentActivity(url: string, log?: any) { - const now = Date.now(); - for (const agent of WATCHED_AGENTS) { - lastUpdateByAgent.set(agent, now); - } - - const messages = await fetchRecentMessages(url); - for (const msg of messages) { - markAgentActivity(msg.from, msg.channel, msg.timestamp); - } - - log?.info?.(`[reflectt][watchdog] seeded activity from ${messages.length} recent message(s)`); -} - -async function hasActiveTask(url: string, agent: string, now = Date.now()): Promise { - const cached = hasActiveTaskByAgent.get(agent); - if (cached && now - cached.checkedAt < TASK_CACHE_TTL_MS) { - return cached.value; - } - - try { - const response = await fetch(`${url}/tasks/next?agent=${encodeURIComponent(agent)}`); - if (!response.ok) { - hasActiveTaskByAgent.set(agent, { value: true, checkedAt: now }); - return true; - } - - const data = await response.json() as { task?: unknown }; - const value = Boolean(data?.task); - hasActiveTaskByAgent.set(agent, { value, checkedAt: now }); - return value; - } catch { - // fail-open so task API hiccups do not suppress legitimate nudges - hasActiveTaskByAgent.set(agent, { value: true, checkedAt: now }); - return true; - } -} - -// --- Gateway restart notification --- - -/** - * Notify agents and suppress alerts after SSE (re)connect. - * - * 1. POST /board-health/quiet-window — suppress ready-queue alerts during reconnect - * 2. POST system message to #general @mentioning all registered agents - * 3. Dedup: skip if last notification was <60s ago (rapid reconnects) - */ -async function notifyGatewayRestart(url: string, log?: any): Promise { - const now = Date.now(); - if (now - lastRestartNotifyAt < RESTART_NOTIFY_DEDUP_MS) { - log?.info?.("[reflectt] Skipping restart notification — within dedup window"); - return; - } - lastRestartNotifyAt = now; - - // 1. Activate quiet window to suppress false-positive board health alerts - try { - await fetch(`${url}/board-health/quiet-window`, { method: "POST", signal: AbortSignal.timeout(5000) }); - log?.info?.("[reflectt] Board health quiet window activated"); - } catch (err) { - log?.warn?.(`[reflectt] Failed to activate quiet window: ${err}`); - } - - // 2. Broadcast system notification to all registered agents - if (WATCHED_AGENTS.length === 0) { - log?.warn?.("[reflectt] No agents in roster — skipping restart broadcast"); - return; - } - - const mentions = WATCHED_AGENTS.map(a => `@${a}`).join(" "); - const content = `🔗 [Gateway Restart] OpenClaw gateway reconnected to reflectt-node. ${mentions} — check your boards.`; - try { - await postMessage(url, "system", "general", content); - log?.info?.(`[reflectt] Restart notification sent to ${WATCHED_AGENTS.length} agents`); - } catch (err) { - log?.warn?.(`[reflectt] Failed to send restart notification: ${err}`); - } -} - -// --- Dedup --- -const seen = new Set(); -function dedup(id: string): boolean { - if (seen.has(id)) return false; - seen.add(id); - if (seen.size > 500) { const f = seen.values().next().value; if (f) seen.delete(f); } - return true; -} - -// --- Runtime dispatch telemetry --- -const dispatchCountByMessageId = new Map(); -function incrementDispatchCount(messageId: string): number { - const next = (dispatchCountByMessageId.get(messageId) ?? 0) + 1; - dispatchCountByMessageId.set(messageId, next); - if (dispatchCountByMessageId.size > 1000) { - const oldest = dispatchCountByMessageId.keys().next().value; - if (oldest) dispatchCountByMessageId.delete(oldest); - } - return next; -} - -// --- SSE connection --- - -let sseRequest: http.ClientRequest | null = null; -let reconnectTimer: ReturnType | null = null; -let healthCheckTimer: ReturnType | null = null; -let watchdogTimer: ReturnType | null = null; -let rosterRefreshTimer: ReturnType | null = null; -let stopped = false; -let pluginRuntime: any = null; -let currentRetryMs = SSE_INITIAL_RETRY_MS; -let lastSSEDataAt = 0; -let sseConnected = false; - -function destroySSE(ctx: any, reason: string) { - if (sseRequest) { - ctx.log?.info(`[reflectt] Destroying SSE connection: ${reason}`); - try { sseRequest.destroy(); } catch {} - sseRequest = null; - } - sseConnected = false; -} - -function connectSSE(url: string, account: ReflecttAccount, ctx: any) { - if (stopped) return; - - // Clean up any lingering connection - if (sseRequest) { - destroySSE(ctx, "new connection attempt"); - } - - ctx.log?.info(`[reflectt] Connecting SSE: ${url}/events/subscribe (retry backoff: ${currentRetryMs}ms)`); - - const req = http.get(`${url}/events/subscribe`, (res) => { - if (res.statusCode !== 200) { - ctx.log?.error(`[reflectt] SSE status ${res.statusCode}`); - res.resume(); - sseRequest = null; - scheduleReconnect(url, account, ctx); - return; - } - - // Connection succeeded — reset backoff - currentRetryMs = SSE_INITIAL_RETRY_MS; - sseConnected = true; - lastSSEDataAt = Date.now(); - ctx.log?.info("[reflectt] SSE connected ✓"); - - // Re-seed agent activity after reconnect - seedAgentActivity(url, ctx.log).catch((err) => { - ctx.log?.warn?.(`[reflectt] post-reconnect seed failed: ${err}`); - }); - - // Notify agents and activate quiet window on gateway (re)connect - notifyGatewayRestart(url, ctx.log).catch((err) => { - ctx.log?.warn?.(`[reflectt] restart notification failed: ${err}`); - }); - - // NOTE: No socket timeout here — SSE streams are idle by nature. - // Staleness is detected by the periodic health-check pinger instead. - - let buffer = ""; - res.setEncoding("utf8"); - - res.on("data", (chunk: string) => { - lastSSEDataAt = Date.now(); - buffer += chunk; - const frames = buffer.split("\n\n"); - buffer = frames.pop() || ""; - - for (const frame of frames) { - if (!frame.trim()) continue; - let eventType = "", eventData = ""; - for (const line of frame.split("\n")) { - if (line.startsWith("event: ")) eventType = line.slice(7).trim(); - else if (line.startsWith("data: ")) eventData = line.slice(6); - } - if (eventType === "message_posted" && eventData) { - handleInbound(eventData, url, account, ctx); - } else if (eventType === "batch" && eventData) { - try { - const events = JSON.parse(eventData); - for (const evt of events) { - if (evt.type === "message_posted" && evt.data) { - handleInbound(JSON.stringify(evt.data), url, account, ctx); - } - } - } catch (e) { - ctx.log?.error(`[reflectt] batch parse error: ${e}`); - } - } - } - }); - - res.on("end", () => { - ctx.log?.warn("[reflectt] SSE stream ended by server"); - sseRequest = null; - sseConnected = false; - scheduleReconnect(url, account, ctx); - }); - res.on("error", (err) => { - ctx.log?.error(`[reflectt] SSE response error: ${err.message}`); - sseRequest = null; - sseConnected = false; - scheduleReconnect(url, account, ctx); - }); - }); - - req.on("error", (err) => { - ctx.log?.error(`[reflectt] SSE connect error: ${err.message}`); - sseRequest = null; - sseConnected = false; - scheduleReconnect(url, account, ctx); - }); - - sseRequest = req; -} - -function scheduleReconnect(url: string, account: ReflecttAccount, ctx: any) { - if (stopped || reconnectTimer) return; - - // Exponential backoff with jitter - const jitter = Math.random() * currentRetryMs * 0.3; - const delay = Math.min(currentRetryMs + jitter, SSE_MAX_RETRY_MS); - ctx.log?.info(`[reflectt] Reconnecting in ${Math.round(delay)}ms`); - - reconnectTimer = setTimeout(() => { - reconnectTimer = null; - connectSSE(url, account, ctx); - }, delay); - - // Increase backoff for next attempt - currentRetryMs = Math.min(currentRetryMs * 2, SSE_MAX_RETRY_MS); -} - -/** - * Periodic health-check: ping /health to detect server availability - * even when the SSE socket hasn't timed out yet. If the server is up - * but we're not connected, force a reconnect. - */ -function startHealthCheck(url: string, account: ReflecttAccount, ctx: any) { - if (healthCheckTimer) return; - - healthCheckTimer = setInterval(async () => { - if (stopped) return; - - try { - const res = await fetch(`${url}/health`, { signal: AbortSignal.timeout(5000) }); - if (res.ok) { - // Server is alive - if (!sseConnected && !reconnectTimer) { - ctx.log?.warn("[reflectt] Health OK but SSE not connected — forcing reconnect"); - currentRetryMs = SSE_INITIAL_RETRY_MS; // reset backoff since server is up - connectSSE(url, account, ctx); - } - } - } catch { - // Server unreachable — SSE reconnect loop will handle it - if (sseConnected) { - ctx.log?.warn("[reflectt] Health check failed while SSE appears connected — destroying stale connection"); - destroySSE(ctx, "health check failed"); - scheduleReconnect(url, account, ctx); - } - } - }, SSE_HEALTH_INTERVAL_MS); -} - -function stopHealthCheck() { - if (healthCheckTimer) { - clearInterval(healthCheckTimer); - healthCheckTimer = null; - } -} - -function startWatchdog(url: string, ctx: any) { - if (watchdogTimer) return; - - watchdogTimer = setInterval(async () => { - const now = Date.now(); - for (const agent of WATCHED_AGENTS) { - const lastUpdateAt = lastUpdateByAgent.get(agent) ?? now; - if (now - lastUpdateAt <= IDLE_NUDGE_WINDOW_MS) continue; - - const activeTask = await hasActiveTask(url, agent, now); - if (!activeTask) { - ctx.log?.info?.(`[reflectt][watchdog] idle nudge suppressed for @${agent}: no active task`); - continue; - } - - const key = `idle:${agent}`; - if (!shouldEscalate(key, now)) continue; - - const content = `@${agent} idle nudge: no update in #general for 15m+. Post shipped / blocker / next+ETA now.`; - try { - await postMessage(url, "watchdog", "general", content); - ctx.log?.info?.(`[reflectt][watchdog] idle nudge fired for @${agent} (last=${lastUpdateAt})`); - } catch (err) { - ctx.log?.warn?.(`[reflectt][watchdog] idle nudge failed for @${agent}: ${err}`); - } - } - }, WATCHDOG_INTERVAL_MS); -} - -function stopWatchdog() { - if (watchdogTimer) { - clearInterval(watchdogTimer); - watchdogTimer = null; - } -} - -function startRosterRefresh(url: string, log?: any) { - if (rosterRefreshTimer) return; - rosterRefreshTimer = setInterval(() => { - refreshAgentRoster(url, log).catch((err) => { - log?.warn(`[reflectt] Periodic roster refresh failed: ${err}`); - }); - }, ROSTER_REFRESH_INTERVAL_MS); -} - -function stopRosterRefresh() { - if (rosterRefreshTimer) { - clearInterval(rosterRefreshTimer); - rosterRefreshTimer = null; - } -} - -function handleInbound(data: string, url: string, account: ReflecttAccount, ctx: any) { - try { - const msg = JSON.parse(data); - const content: string = msg.content || ""; - const msgId: string = msg.id || ""; - const from: string = msg.from || "unknown"; - const channel: string = msg.channel || "general"; - - // Re-enabled by operator request: system/watchdog messages should be dispatch-eligible. - // Keep activity tracking consistent for all senders. - markAgentActivity(from, channel, msg.timestamp); - - if (!msgId) return; - if (!dedup(msgId)) { - ctx.log?.debug(`[reflectt][dispatch-telemetry] duplicate inbound ignored message_id=${msgId}`); - return; - } - - // Extract @mentions - const mentions: string[] = []; - const regex = /@(\w+)/g; - let match: RegExpExecArray | null; - while ((match = regex.exec(content)) !== null) mentions.push(match[1].toLowerCase()); - if (mentions.length === 0) return; - - // Build agent ID map - const cfg = pluginRuntime?.config?.loadConfig?.() ?? {}; - const agentList: Array<{ id: string; identity?: { name?: string } }> = cfg?.agents?.list || []; - const agentIds = new Set(); - const agentNameToId = new Map(); - for (const a of agentList) { - agentIds.add(a.id); - agentNameToId.set(a.id, a.id); - if (a.identity?.name) { - const name = a.identity.name.toLowerCase(); - agentIds.add(name); - agentNameToId.set(name, a.id); - } - } - - // Determine sender's agent ID (if message is from an agent) - const senderAgentId = agentNameToId.get(from.toLowerCase()); - - // Find mentioned agent - ctx.log?.debug(`[reflectt] Processing mentions: ${mentions.join(", ")}`); - let matchedMentions = 0; - let skippedSelfMentions = 0; - let unmatchedMentions = 0; - const dispatchedTargets: string[] = []; - const dispatchedSet = new Set(); - - for (const mention of mentions) { - let agentId: string | undefined; - for (const a of agentList) { - if (a.id === mention) { agentId = a.id; break; } - if (a.identity?.name?.toLowerCase() === mention) { agentId = a.id; break; } - } - if (!agentId && mention === "kai") agentId = "main"; - if (!agentId) { - unmatchedMentions += 1; - ctx.log?.debug(`[reflectt] Mention @${mention} did not match any agent`); - continue; - } - - // Skip routing to yourself (avoid self-loops) - if (senderAgentId && agentId === senderAgentId) { - skippedSelfMentions += 1; - ctx.log?.debug(`[reflectt] Skipping self-mention: @${agentId}`); - continue; - } - - if (dispatchedSet.has(agentId)) { - ctx.log?.debug(`[reflectt] Skipping duplicate mention target: @${agentId}`); - continue; - } - - matchedMentions += 1; - dispatchedSet.add(agentId); - ctx.log?.info(`[reflectt] ${from} → @${agentId}: ${content.slice(0, 60)}...`); - - // Build inbound message context - const runtime = pluginRuntime; - if (!runtime?.channel?.reply) continue; - - // All reflectt rooms share one session per agent (peer: null). - // Room identity is preserved in OriginatingTo so replies route correctly. - const sessionKey = `agent:${agentId}:reflectt:main`; - - // Create message context - const msgContext = { - Body: content, - BodyForAgent: content, - CommandBody: content, - BodyForCommands: content, - From: `reflectt:${channel}`, - To: channel, - SessionKey: sessionKey, - AccountId: account.accountId, - MessageSid: msgId, - ChatType: "group", - ConversationLabel: `reflectt-node #${channel}`, - SenderName: from, - SenderId: from, - Timestamp: msg.timestamp || Date.now(), - Provider: "reflectt", - Surface: "reflectt", - OriginatingChannel: "reflectt" as const, - OriginatingTo: channel, - WasMentioned: true, - CommandAuthorized: false, - }; - - // Finalize context - const finalizedCtx = runtime.channel.reply.finalizeInboundContext(msgContext); - - // Guard against stale/unsafe session path metadata leaking from prior context. - // OpenClaw now enforces that any session file path must be under the agent sessions dir. - const safeCtx: any = { ...finalizedCtx }; - delete safeCtx.SessionFilePath; - delete safeCtx.sessionFilePath; - delete safeCtx.SessionPath; - delete safeCtx.sessionPath; - delete safeCtx.TranscriptPath; - delete safeCtx.transcriptPath; - delete safeCtx.SessionFile; - delete safeCtx.sessionFile; - - // Create reply dispatcher - const agentName = agentId === "main" ? "kai" : agentId; - const dispatcher = runtime.channel.reply.createReplyDispatcherWithTyping({ - deliver: async (payload: any) => { - const text = payload.text || payload.content || ""; - if (text) { - ctx.log?.info(`[reflectt] Reply → ${channel}: ${text.slice(0, 60)}...`); - await postMessage(url, agentName!, channel, text); - } - }, - onError: (err: unknown) => { - ctx.log?.error(`[reflectt] Dispatch error: ${err}`); - }, - }); - - // Dispatch reply using OpenClaw's pipeline - const dispatchCount = incrementDispatchCount(msgId); - dispatchedTargets.push(agentId); - ctx.log?.info( - `[reflectt][dispatch-telemetry] message_id=${msgId} dispatch_count=${dispatchCount} target=${agentId} mentions_total=${mentions.length}`, - ); - - runtime.channel.reply.dispatchReplyFromConfig({ - ctx: safeCtx, - cfg, - dispatcher: dispatcher.dispatcher, - replyOptions: dispatcher.replyOptions, - }).catch((err: unknown) => { - const errText = String(err ?? ""); - if (errText.includes("Session file path must be within sessions directory")) { - const healed = purgeSessionIndexEntry(agentId!, sessionKey, ctx); - if (healed) { - ctx.log?.warn(`[reflectt] Retrying dispatch after purging stale session entry: ${sessionKey}`); - runtime.channel.reply.dispatchReplyFromConfig({ - ctx: safeCtx, - cfg, - dispatcher: dispatcher.dispatcher, - replyOptions: dispatcher.replyOptions, - }).catch((retryErr: unknown) => { - ctx.log?.error(`[reflectt] dispatch retry failed: ${retryErr}`); - }); - return; - } - } - ctx.log?.error(`[reflectt] dispatchReplyFromConfig error: ${err}`); - }); - } - - ctx.log?.info( - `[reflectt][dispatch-telemetry] summary message_id=${msgId} mentions_total=${mentions.length} matched=${matchedMentions} unmatched=${unmatchedMentions} skipped_self=${skippedSelfMentions} dispatched=${dispatchedTargets.length} targets=${dispatchedTargets.join(",") || "none"}`, - ); - } catch (err) { - ctx.log?.error(`[reflectt] Parse error: ${err}`); - } -} - -// --- Channel Plugin --- - -const reflecttPlugin: ChannelPlugin = { - id: "reflectt", - meta: { - id: "reflectt", - label: "Reflectt", - selectionLabel: "Reflectt (Local)", - docsPath: "/channels/reflectt", - docsLabel: "reflectt", - blurb: "Real-time agent collaboration via reflectt-node", - order: 110, - }, - capabilities: { - chatTypes: ["group"], - media: false, - }, - reload: { configPrefixes: ["channels.reflectt"] }, - - config: { - listAccountIds: () => [DEFAULT_ACCOUNT_ID], - resolveAccount: (cfg, accountId) => resolveAccount(cfg, accountId, pluginRuntime?.logger), - defaultAccountId: () => DEFAULT_ACCOUNT_ID, - isConfigured: (account) => account.configured, - describeAccount: (account) => ({ - accountId: account.accountId, - name: "Reflectt", - enabled: account.enabled, - configured: account.configured, - }), - }, - - outbound: { - deliveryMode: "direct", - textChunkLimit: 4000, - sendText: async ({ to, text, accountId }) => { - const cfg = pluginRuntime?.config?.loadConfig?.() ?? {}; - const account = resolveAccount(cfg, accountId, pluginRuntime?.logger); - // Determine agent name for "from" field - const agentName = "kai"; // TODO: resolve from session context - await postMessage(account.url, agentName, "general", text ?? ""); - return { channel: "reflectt" as const, to, messageId: `rn-${Date.now()}` }; - }, - }, - - gateway: { - startAccount: async (ctx) => { - const account = ctx.account; - if (!account.enabled) return; - - stopped = false; - - // Validate connectivity and show actionable error if server unreachable - try { - const healthRes = await fetch(`${account.url}/health`, { signal: AbortSignal.timeout(5000) }); - if (!healthRes.ok) { - ctx.log?.warn(`[reflectt] Server at ${account.url} returned ${healthRes.status}. Will retry via SSE reconnect.`); - } else { - ctx.log?.info(`[reflectt] Server at ${account.url} is healthy ✓`); - } - } catch { - ctx.log?.error( - `[reflectt] Cannot reach reflectt-node at ${account.url}. ` + - `Make sure reflectt-node is running, then set the URL in your OpenClaw config:\n` + - ` Option 1 (recommended): channels.reflectt.url = "${account.url}"\n` + - ` Option 2: plugins.entries.reflectt-channel.config.url = "${account.url}"\n` + - `Will keep retrying via SSE reconnect.` - ); - } - - ctx.setStatus({ - accountId: account.accountId, - name: "Reflectt", - enabled: true, - configured: true, - }); - - // Load agent roster from reflectt-node before starting SSE, then refresh every 5 minutes. - await refreshAgentRoster(account.url, ctx.log); - startRosterRefresh(account.url, ctx.log); - - seedAgentActivity(account.url, ctx.log).catch((err) => { - ctx.log?.warn?.(`[reflectt][watchdog] seed failed: ${err}`); - }); - startWatchdog(account.url, ctx); - startHealthCheck(account.url, account, ctx); - connectSSE(account.url, account, ctx); - - return { - stop: () => { - stopped = true; - stopRosterRefresh(); - stopWatchdog(); - stopHealthCheck(); - if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } - destroySSE(ctx, "plugin stopped"); - ctx.log?.info("[reflectt] Stopped"); - }, - }; - }, - }, -}; - -// --- Plugin entry --- - -const plugin = { - id: "reflectt-channel", - name: "Reflectt Channel", - description: "Real-time agent collaboration via reflectt-node SSE", - - register(api: OpenClawPluginApi) { - pluginRuntime = api.runtime; - api.logger.info("[reflectt] Registering channel plugin"); - api.registerChannel({ plugin: reflecttPlugin }); - }, -}; - -export default plugin; diff --git a/plugins/reflectt-channel/openclaw.plugin.json b/plugins/reflectt-channel/openclaw.plugin.json deleted file mode 100644 index bd91092f..00000000 --- a/plugins/reflectt-channel/openclaw.plugin.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "id": "reflectt-channel", - "name": "Reflectt Channel", - "description": "Real-time agent collaboration via reflectt-node SSE", - "version": "0.2.1", - "channels": ["reflectt"], - "configSchema": { - "type": "object", - "additionalProperties": false, - "properties": { - "url": { - "type": "string", - "description": "Reflectt-node server URL", - "default": "http://127.0.0.1:4445" - }, - "enabled": { - "type": "boolean", - "description": "Enable Reflectt channel", - "default": true - } - } - } -} diff --git a/plugins/reflectt-channel/package.json b/plugins/reflectt-channel/package.json deleted file mode 100644 index 00e12681..00000000 --- a/plugins/reflectt-channel/package.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "name": "@openclaw/reflectt-channel", - "version": "1.0.0", - "description": "OpenClaw channel plugin for reflectt-node", - "type": "module", - "main": "index.ts", - "openclaw": { - "extensions": ["./index.ts"], - "channel": { - "id": "reflectt", - "label": "Reflectt Node", - "selectionLabel": "Reflectt Node (local)", - "docsPath": "/channels/reflectt", - "docsLabel": "reflectt", - "blurb": "Real-time messaging via reflectt-node event stream", - "order": 100, - "aliases": ["reflectt"] - } - }, - "keywords": ["openclaw", "channel", "reflectt"], - "author": "Reflectt Team", - "license": "MIT" -} diff --git a/plugins/reflectt-channel/src/channel.ts b/plugins/reflectt-channel/src/channel.ts deleted file mode 100644 index d03f73ca..00000000 --- a/plugins/reflectt-channel/src/channel.ts +++ /dev/null @@ -1,433 +0,0 @@ -import type { ChannelPlugin, OpenClawConfig } from "openclaw/plugin-sdk"; -import { DEFAULT_ACCOUNT_ID } from "openclaw/plugin-sdk"; -import type { ResolvedReflecttAccount, ReflecttConfig } from "./types.js"; - -const DEFAULT_REFLECTT_URL = "http://localhost:4445"; - -// NOTE: This file is a legacy standalone implementation and is NOT imported by ../index.ts. -// Agent discovery is handled dynamically in index.ts via refreshAgentRoster() which queries -// /team/roles on connect and refreshes every 5 minutes. The hardcoded roster has been removed; -// if this file is ever wired up it should import the shared roster from index.ts instead. -let WATCHED_AGENTS: string[] = []; -let WATCHED_SET = new Set(); - -const IDLE_NUDGE_WINDOW_MS = 15 * 60 * 1000; // 15m -const WATCHDOG_INTERVAL_MS = 60 * 1000; // 1m poll -const ESCALATION_COOLDOWN_MS = 20 * 60 * 1000; // anti-spam - -type WatchdogState = { - lastUpdateByAgent: Map; - lastEscalationAt: Map; -}; - -function normalizeSenderId(value: unknown): string | null { - if (typeof value !== "string") return null; - const id = value.trim().toLowerCase(); - return id.length > 0 ? id : null; -} - -function shouldEscalate(state: WatchdogState, key: string, now: number): boolean { - const last = state.lastEscalationAt.get(key) ?? 0; - if (now - last < ESCALATION_COOLDOWN_MS) return false; - state.lastEscalationAt.set(key, now); - return true; -} - -async function fetchRecentMessages(url: string): Promise>> { - const endpoints = ["/chat/messages?limit=500", "/chat/messages", "/messages"]; - for (const endpoint of endpoints) { - try { - const response = await fetch(`${url}${endpoint}`); - if (!response.ok) continue; - const data = await response.json(); - const messages = Array.isArray(data) - ? data - : data && typeof data === "object" && Array.isArray((data as { messages?: unknown[] }).messages) - ? (data as { messages: unknown[] }).messages - : []; - return messages.filter((m): m is Record => Boolean(m) && typeof m === "object"); - } catch { - // best effort - } - } - return []; -} - -async function seedWatchdogState(url: string, state: WatchdogState): Promise { - const now = Date.now(); - for (const agent of WATCHED_AGENTS) { - state.lastUpdateByAgent.set(agent, now); - } - - const messages = await fetchRecentMessages(url); - for (const msg of messages) { - const channel = typeof msg.channel === "string" ? msg.channel : ""; - if (channel !== "general") continue; - - const senderId = normalizeSenderId(msg.from); - if (!senderId || !WATCHED_SET.has(senderId)) continue; - - const rawTs = msg.timestamp; - const ts = - typeof rawTs === "number" && Number.isFinite(rawTs) - ? rawTs - : typeof rawTs === "string" && Number.isFinite(Number(rawTs)) - ? Number(rawTs) - : now; - - const current = state.lastUpdateByAgent.get(senderId) ?? 0; - if (ts > current) state.lastUpdateByAgent.set(senderId, ts); - } -} - -async function postWatchdogNudge(url: string, agent: string): Promise { - await fetch(`${url}/chat/messages`, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - from: "watchdog", - channel: "general", - content: `@${agent} idle nudge: no update in #general for 15m+. Post shipped / blocker / next+ETA now.`, - timestamp: Date.now(), - }), - }); -} - -function resolveReflecttAccount(params: { - cfg: OpenClawConfig; - accountId?: string | null; -}): ResolvedReflecttAccount { - const { cfg, accountId } = params; - const effectiveAccountId = accountId || DEFAULT_ACCOUNT_ID; - - const channelConfig = (cfg.channels?.reflectt as ReflecttConfig) || {}; - const enabled = channelConfig.enabled !== false; - const url = channelConfig.url || DEFAULT_REFLECTT_URL; - - return { - accountId: effectiveAccountId, - enabled, - config: channelConfig, - url, - }; -} - -export const reflecttPlugin: ChannelPlugin = { - id: "reflectt-channel", - meta: { - id: "reflectt-channel", - label: "Reflectt Node", - selectionLabel: "Reflectt Node (local)", - docsPath: "/channels/reflectt", - blurb: "Real-time messaging via reflectt-node event stream.", - aliases: ["reflectt"], - }, - capabilities: { - chatTypes: ["direct", "channel"], - media: false, - reactions: false, - threads: false, - polls: false, - nativeCommands: false, - blockStreaming: false, - }, - reload: { - configPrefixes: ["channels.reflectt"], - }, - config: { - listAccountIds: () => [DEFAULT_ACCOUNT_ID], - resolveAccount: (cfg, accountId) => resolveReflecttAccount({ cfg, accountId }), - defaultAccountId: () => DEFAULT_ACCOUNT_ID, - isConfigured: (account) => Boolean(account.url), - isEnabled: (account) => account.enabled, - describeAccount: (account) => ({ - accountId: account.accountId, - name: "Reflectt Node", - enabled: account.enabled, - configured: Boolean(account.url), - }), - }, - outbound: { - deliveryMode: "direct", - sendText: async (ctx) => { - const account = resolveReflecttAccount({ cfg: ctx.cfg, accountId: ctx.accountId }); - - try { - const response = await fetch(`${account.url}/chat/messages`, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - from: "openclaw_agent", - channel: ctx.to || "general", - content: ctx.text, - }), - }); - - if (!response.ok) { - return { - ok: false, - error: new Error(`Failed to send message: ${response.statusText}`), - }; - } - - return { ok: true }; - } catch (error) { - return { - ok: false, - error: error instanceof Error ? error : new Error(String(error)), - }; - } - }, - }, - gateway: { - startAccount: async (ctx) => { - const { cfg, account, runtime, abortSignal, log } = ctx; - - log?.info?.(`Starting reflectt channel monitor: ${account.url}`); - - const watchdogState: WatchdogState = { - lastUpdateByAgent: new Map(), - lastEscalationAt: new Map(), - }; - - try { - await seedWatchdogState(account.url, watchdogState); - } catch (error) { - log?.warn?.(`Failed to seed watchdog state: ${error}`); - } - - const watchdogTimer = setInterval(async () => { - const now = Date.now(); - - for (const agent of WATCHED_AGENTS) { - const lastUpdateAt = watchdogState.lastUpdateByAgent.get(agent) ?? now; - if (now - lastUpdateAt <= IDLE_NUDGE_WINDOW_MS) continue; - - const key = `idle:${agent}`; - if (!shouldEscalate(watchdogState, key, now)) continue; - - try { - await postWatchdogNudge(account.url, agent); - log?.info?.(`[reflectt-channel] idle nudge fired for @${agent} (last=${lastUpdateAt})`); - } catch (error) { - log?.warn?.(`[reflectt-channel] idle nudge failed for @${agent}: ${error}`); - } - } - }, WATCHDOG_INTERVAL_MS); - - abortSignal.addEventListener("abort", () => { - clearInterval(watchdogTimer); - }); - - // Connect to SSE event stream - const connectSSE = async () => { - try { - const response = await fetch(`${account.url}/events`, { - signal: abortSignal, - }); - - if (!response.ok) { - throw new Error(`SSE connection failed: ${response.statusText}`); - } - - const reader = response.body?.getReader(); - const decoder = new TextDecoder(); - - if (!reader) { - throw new Error("No response body available"); - } - - let buffer = ""; - - while (!abortSignal.aborted) { - const { done, value } = await reader.read(); - - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split("\n"); - buffer = lines.pop() || ""; - - for (const line of lines) { - if (!line.trim() || line.startsWith(":")) continue; - - if (line.startsWith("data: ")) { - const data = line.slice(6); - - try { - const event = JSON.parse(data); - const message = event?.data ?? event?.message; - - if ((event.type === "chat_message" || event.type === "message") && message) { - const senderId = normalizeSenderId(message.from); - if (senderId && WATCHED_SET.has(senderId) && message.channel === "general") { - const rawTs = message.timestamp; - const ts = - typeof rawTs === "number" && Number.isFinite(rawTs) - ? rawTs - : typeof rawTs === "string" && Number.isFinite(Number(rawTs)) - ? Number(rawTs) - : Date.now(); - watchdogState.lastUpdateByAgent.set(senderId, ts); - } - - await handleChatMessage(message, cfg, runtime, log); - } - } catch (parseError) { - log?.warn?.(`Failed to parse SSE data: ${parseError}`); - } - } - } - } - } catch (error) { - if (abortSignal.aborted) { - log?.info?.("SSE connection closed (aborted)"); - return; - } - - log?.error?.(`SSE connection error: ${error}`); - - // Retry after delay if not aborted - if (!abortSignal.aborted) { - await new Promise((resolve) => setTimeout(resolve, 5000)); - if (!abortSignal.aborted) { - log?.info?.("Reconnecting to reflectt SSE stream..."); - await connectSSE(); - } - } - } - }; - - await connectSSE(); - }, - stopAccount: async (ctx) => { - ctx.log?.info?.("Stopping reflectt channel monitor"); - // Abort signal will handle cleanup - }, - }, -}; - -async function handleChatMessage( - message: any, - cfg: OpenClawConfig, - runtime: any, - log: any, -) { - try { - const { from, channel, content, timestamp, id } = message; - const safeContent = typeof content === "string" ? content : ""; - const safeChannel = typeof channel === "string" ? channel : "general"; - - // Check if message mentions an agent - const mentionedAgents = extractAgentMentions(safeContent, cfg); - - if (mentionedAgents.length === 0) { - // No mentions, ignore - return; - } - - log?.info?.(`Processing reflectt message from ${from} in ${safeChannel} (mentions: ${mentionedAgents.join(", ")})`); - - // Route to each mentioned agent - for (const agentId of mentionedAgents) { - try { - // Resolve agent route — all reflectt rooms share one session per agent. - // Room identity is preserved in To/OriginatingTo so replies route correctly. - const route = runtime.channel.routing.resolveAgentRoute({ - cfg, - channel: "reflectt", - accountId: "default", - peer: null, - }); - - // Build envelope-formatted body - const body = runtime.channel.reply.formatInboundEnvelope({ - channel: "Reflectt", - from: from, - timestamp: new Date(timestamp || Date.now()), - envelope: runtime.channel.reply.resolveEnvelopeFormatOptions({ cfg, channel: "reflectt" }), - body: safeContent, - }); - - // Finalize inbound context - const ctxPayload = runtime.channel.reply.finalizeInboundContext({ - Body: body, - RawBody: safeContent, - CommandBody: safeContent, - From: from, - To: safeChannel, - SessionKey: route.sessionKey, - AccountId: route.accountId, - ChatType: "channel", - GroupSubject: safeChannel, - SenderName: from, - SenderId: from, - Provider: "reflectt", - Surface: "reflectt", - MessageSid: id || String(timestamp || Date.now()), - Timestamp: timestamp || Date.now(), - WasMentioned: true, - CommandAuthorized: true, - OriginatingChannel: "reflectt", - OriginatingTo: safeChannel, - }); - - // Create reply dispatcher - const dispatcher = { - deliver: async (payload: any) => { - const account = resolveReflecttAccount({ cfg, accountId: route.accountId }); - - // Send text messages back to reflectt-node - if (payload.text) { - await fetch(`${account.url}/chat/messages`, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - from: agentId, - channel: safeChannel, - content: payload.text, - }), - }); - } - }, - onError: (err: unknown) => { - log?.error?.(`Reflectt reply error: ${err}`); - }, - }; - - // Dispatch reply - await runtime.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ - ctx: ctxPayload, - cfg, - dispatcherOptions: dispatcher, - }); - - log?.info?.(`Dispatched reflectt message to agent ${agentId}`); - } catch (agentError) { - log?.error?.(`Error dispatching to agent ${agentId}: ${agentError}`); - } - } - } catch (error) { - log?.error?.(`Error handling chat message: ${error}`); - } -} - -function extractAgentMentions(content: string, cfg: OpenClawConfig): string[] { - const mentions: string[] = []; - const mentionPattern = /@([\w][\w-]*[\w]|[\w]+)/g; - - let match; - while ((match = mentionPattern.exec(content)) !== null) { - const mentionedName = match[1]; - - // Check if this matches an agent - const agents = cfg.agents?.list || []; - for (const agent of agents) { - if (agent.id === mentionedName || agent.name?.toLowerCase() === mentionedName.toLowerCase()) { - mentions.push(agent.id); - } - } - } - - return [...new Set(mentions)]; // dedupe -} diff --git a/plugins/reflectt-channel/src/types.ts b/plugins/reflectt-channel/src/types.ts deleted file mode 100644 index b29a7854..00000000 --- a/plugins/reflectt-channel/src/types.ts +++ /dev/null @@ -1,29 +0,0 @@ -/** - * Types for reflectt-node integration - */ - -export type ReflecttConfig = { - enabled?: boolean; - url?: string; -}; - -export type ResolvedReflecttAccount = { - accountId: string; - enabled: boolean; - config: ReflecttConfig; - url: string; -}; - -export type ReflecttChatMessage = { - id: string; - from: string; - channel: string; - content: string; - timestamp: number; - mentions?: string[]; -}; - -export type ReflecttEvent = { - type: "chat_message" | "system" | "agent_status"; - data: ReflecttChatMessage | Record; -}; diff --git a/process/reflectt-channel-plugin.md b/process/reflectt-channel-plugin.md deleted file mode 100644 index 248a1a0b..00000000 --- a/process/reflectt-channel-plugin.md +++ /dev/null @@ -1,13 +0,0 @@ -# Reflectt Channel Plugin - -**Task:** `task-1772209309856-0ggpra1so` -**PR:** [#453](https://github.com/reflectt/reflectt-node/pull/453) -**Branch:** `link/reflectt-channel-plugin` - -## Done Criteria → Evidence - -| Criteria | Evidence | -|----------|----------| -| Plugin install path documented and works | `openclaw plugins install ./plugins/reflectt-channel` tested | -| Published npm OR bundled in repo | Bundled under `plugins/reflectt-channel/` | -| Fresh install → plugin available → connects | README with full install+config+verify flow | From 6c878fda878b492fe77b8765a176280ec232017b Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 2 May 2026 17:22:48 -0700 Subject: [PATCH 09/14] fix: cut stale agent context at latest canvas seed --- src/server.ts | 168 ++++++++++++++++++++++++++++++++++++---------- tests/api.test.ts | 41 +++++++++++ 2 files changed, 173 insertions(+), 36 deletions(-) diff --git a/src/server.ts b/src/server.ts index ae3bb5fb..f3bbfeff 100644 --- a/src/server.ts +++ b/src/server.ts @@ -4215,6 +4215,94 @@ export async function createServer(): Promise { return payload }) + type AgentSeedCutoff = { + timestamp: number + messageId: string + source: string + } + + function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&') + } + + function findLatestAgentSeedCutoff(messages: AgentMessage[], agent: string): AgentSeedCutoff | null { + const agentPattern = new RegExp(`@${escapeRegExp(agent)}\\b`, 'i') + + for (let index = messages.length - 1; index >= 0; index--) { + const message = messages[index] + if (!message) continue + + const metadata = (message.metadata || {}) as Record + const source = typeof metadata.source === 'string' ? metadata.source : '' + const replyVia = typeof metadata.reply_via === 'string' ? metadata.reply_via : '' + const isCanvasSeed = source === 'canvas_query' || replyVia === 'canvas_push' + if (!isCanvasSeed) continue + + const directToAgent = typeof message.to === 'string' && message.to.trim().toLowerCase() === agent + const mentionsAgent = agentPattern.test(message.content || '') + if (!directToAgent && !mentionsAgent) continue + + const timestamp = Number(message.timestamp) || 0 + if (timestamp <= 0) continue + + return { + timestamp, + messageId: message.id, + source: source || replyVia, + } + } + + return null + } + + function selectAgentContextMessages(opts: { + agent: string + limit: number + channelFilter?: string + sinceMs: number + }): { + allMessages: AgentMessage[] + filteredMessages: AgentMessage[] + mentions: AgentMessage[] + systemAlerts: AgentMessage[] + teamMessages: AgentMessage[] + seedCutoff: AgentSeedCutoff | null + seedFilteredCount: number + } { + const allMessages = chatManager.getMessages({ + channel: opts.channelFilter, + limit: Math.min(opts.limit * 6, 800), + since: opts.sinceMs, + }) + + const seedCutoff = findLatestAgentSeedCutoff(allMessages, opts.agent) + const filteredMessages = seedCutoff + ? allMessages.filter(message => (message.timestamp || 0) >= seedCutoff.timestamp) + : allMessages + + const mentions: AgentMessage[] = [] + const systemAlerts: AgentMessage[] = [] + const teamMessages: AgentMessage[] = [] + const agentPattern = new RegExp(`@${escapeRegExp(opts.agent)}\\b`, 'i') + + for (const message of filteredMessages) { + const content = message.content || '' + if (message.from === 'system') systemAlerts.push(message) + else if (agentPattern.test(content)) mentions.push(message) + else teamMessages.push(message) + } + + return { + allMessages, + filteredMessages, + mentions, + systemAlerts, + teamMessages, + seedCutoff, + seedFilteredCount: Math.max(0, allMessages.length - filteredMessages.length), + } + } + // ── Agent context endpoint ────────────────────────────────────────── // Returns a compact, deduplicated view of recent chat optimized for // agent context injection. Includes: mentions of the agent, recent @@ -4232,26 +4320,20 @@ export async function createServer(): Promise { const strictCompact = query.compact === '1' || query.compact === 'true' const maxChars = Math.max(400, Math.min(Number(query.max_chars) || 1200, 8000)) - const allMessages = chatManager.getMessages({ - channel: channelFilter, - limit: Math.min(limit * 6, 800), // fetch more, then filter - since: sinceMs, + const { + allMessages, + mentions, + systemAlerts, + teamMessages, + seedCutoff, + seedFilteredCount, + } = selectAgentContextMessages({ + agent, + limit, + channelFilter, + sinceMs, }) - // Partition: mentions, system alerts, team messages - const mentions: typeof allMessages = [] - const systemAlerts: typeof allMessages = [] - const teamMessages: typeof allMessages = [] - - const agentPattern = new RegExp(`@${agent}\\b`, 'i') - - for (const m of allMessages) { - const content = m.content || '' - if (m.from === 'system') systemAlerts.push(m) - else if (agentPattern.test(content)) mentions.push(m) - else teamMessages.push(m) - } - const normalizeForDedup = (content: string): string => { return (content || '') .replace(/\d{10,}/g, '') // strip epoch timestamps @@ -4394,8 +4476,18 @@ export async function createServer(): Promise { suppressed: { system_deduped: systemAlerts.length - dedupedAlerts.length, total_scanned: allMessages.length, + seed_cutoff_filtered: seedFilteredCount, ...(strictCompact ? { truncated: result.length - messagesOut.length } : {}), }, + ...(seedCutoff + ? { + seed_cutoff: { + timestamp: seedCutoff.timestamp, + message_id: seedCutoff.messageId, + source: seedCutoff.source, + }, + } + : {}), } }) @@ -4419,26 +4511,20 @@ export async function createServer(): Promise { const peer = (query.peer || '').trim() const taskIdOverride = (query.task_id || '').trim() - const allMessages = chatManager.getMessages({ - channel: channelFilter, - limit: Math.min(limit * 6, 800), - since: sinceMs, + const { + allMessages, + mentions, + systemAlerts, + teamMessages, + seedCutoff, + seedFilteredCount, + } = selectAgentContextMessages({ + agent, + limit, + channelFilter, + sinceMs, }) - // Partition: mentions, system alerts, team messages - const mentions: typeof allMessages = [] - const systemAlerts: typeof allMessages = [] - const teamMessages: typeof allMessages = [] - - const agentPattern = new RegExp(`@${agent}\\b`, 'i') - - for (const m of allMessages) { - const content = m.content || '' - if (m.from === 'system') systemAlerts.push(m) - else if (agentPattern.test(content)) mentions.push(m) - else teamMessages.push(m) - } - // Deduplicate system alerts by normalized content const seenHashes = new Set() const dedupedAlerts = systemAlerts.filter(m => { @@ -4501,7 +4587,17 @@ export async function createServer(): Promise { selected: selected.length, suppressed: { system_deduped: systemAlerts.length - dedupedAlerts.length, + seed_cutoff_filtered: seedFilteredCount, }, + ...(seedCutoff + ? { + seed_cutoff: { + timestamp: seedCutoff.timestamp, + message_id: seedCutoff.messageId, + source: seedCutoff.source, + }, + } + : {}), }, } }) diff --git a/tests/api.test.ts b/tests/api.test.ts index a6c15e59..0ab3dd8d 100644 --- a/tests/api.test.ts +++ b/tests/api.test.ts @@ -5282,6 +5282,47 @@ describe('Context budget', () => { await fs.rm(tmp, { recursive: true, force: true }).catch(() => {}) } }) + + it('GET /context/inject drops pre-seed stale chat before the latest canvas query for that agent', async () => { + const agent = `seed-cutoff-${Date.now()}` + + await req('POST', '/chat/messages', { + from: 'sage', + content: `@${agent} stale failure memory says use /capture_frame surface=camera`, + channel: 'general', + }) + + const seed = await req('POST', '/chat/messages', { + from: 'kai', + to: agent, + content: `[canvas] @${agent} use the uploaded room artifact instead of capturing a fresh frame`, + channel: 'general', + metadata: { + source: 'canvas_query', + reply_via: 'canvas_push', + attachments: [{ name: 'proof.png', type: 'image/png', sizeBytes: 1234 }], + }, + }) + expect(seed.status).toBe(200) + + await req('POST', '/chat/messages', { + from: 'echo', + content: 'Fresh room artifact is already shared in the canvas.', + channel: 'general', + }) + + const injected = await req('GET', `/context/inject/${agent}?limit=12&channel=general`) + expect(injected.status).toBe(200) + + const sessionItems = injected.body.layers.session_local.items as Array<{ content: string }> + const contents = sessionItems.map(item => item.content) + + expect(contents.some(content => content.includes('stale failure memory says use /capture_frame surface=camera'))).toBe(false) + expect(contents.some(content => content.includes('use the uploaded room artifact instead of capturing a fresh frame'))).toBe(true) + expect(contents.some(content => content.includes('Fresh room artifact is already shared in the canvas.'))).toBe(true) + expect(injected.body.session_source.suppressed.seed_cutoff_filtered).toBeGreaterThanOrEqual(1) + expect(injected.body.session_source.seed_cutoff.source).toBe('canvas_query') + }) }) describe('Duplicate-closure validating evidence gate', () => { From 1816f310e9d952b056ae8aaac7f4b22e2b1ae6a6 Mon Sep 17 00:00:00 2001 From: Kai Date: Mon, 4 May 2026 08:02:07 -0700 Subject: [PATCH 10/14] feat(node): add read-only team fleet route --- src/cloud.ts | 38 ++++++++++++++++++++++++++++++++++++++ src/server.ts | 39 ++++++++++++++++++++++++++++++++++++++- tests/api.test.ts | 8 ++++++++ 3 files changed, 84 insertions(+), 1 deletion(-) diff --git a/src/cloud.ts b/src/cloud.ts index 925a0ec1..e09937a2 100644 --- a/src/cloud.ts +++ b/src/cloud.ts @@ -331,6 +331,15 @@ export function getCloudStatus() { } } +/** + * Fetch the current team fleet snapshot for this host. + * Uses the host credential to read the persisted cross-host truth from cloud. + */ +export async function getTeamFleet(): Promise> { + if (!state.hostId) return { success: false, error: 'Host not registered with cloud' } + return cloudGet(`/api/hosts/${state.hostId}/team/fleet`) +} + /** * Initialize and start cloud integration. * Call this after the server is listening. @@ -2156,6 +2165,35 @@ interface CloudApiResponse { error?: string } +export interface TeamFleetAgentSnapshot { + id?: string + name?: string + displayName?: string + state?: string + status?: string + [key: string]: unknown +} + +export interface TeamFleetHostSnapshot { + id: string + teamId: string + name: string + status: string + lastSeen: string + agents: TeamFleetAgentSnapshot[] + activeTasks: unknown[] + slowTasks: unknown[] + appVersion: string | null + gitSha: string | null + buildTimestamp: string | null + convergence?: unknown +} + +export interface TeamFleetSnapshot { + teamId: string + hosts: TeamFleetHostSnapshot[] +} + async function cloudGet(path: string): Promise> { if (!config) return { success: false, error: 'Not configured' } diff --git a/src/server.ts b/src/server.ts index f3bbfeff..ca0d9885 100644 --- a/src/server.ts +++ b/src/server.ts @@ -178,7 +178,7 @@ import { createRun, getRun, subscribeRun, approveRun, rejectRun, executeGithubIs import { validateIntent as macOSValidateIntent, isKillSwitchEngaged, engageKillSwitch, resetKillSwitch } from './macos-accessibility.js' import { calendarManager, type BlockType, type CreateBlockInput, type UpdateBlockInput } from './calendar.js' import { calendarEvents, type CreateEventInput, type UpdateEventInput, type AttendeeStatus } from './calendar-events.js' -import { requestImmediateCanvasSync, queueCanvasPushEvent } from './cloud.js' +import { requestImmediateCanvasSync, queueCanvasPushEvent, getTeamFleet } from './cloud.js' import { startReminderEngine, stopReminderEngine, getReminderEngineStats } from './calendar-reminder-engine.js' import { startDeployMonitor, stopDeployMonitor } from './deploy-monitor.js' import { exportICS, exportEventICS, importICS, parseICS } from './calendar-ical.js' @@ -2971,6 +2971,43 @@ export async function createServer(): Promise { } }) + // Cross-host team fleet snapshot (read-only) + app.get('/team/fleet', async (_request, reply) => { + const result = await getTeamFleet() + if (!result.success || !result.data) { + reply.code(result.error === 'Host not registered with cloud' ? 503 : 502) + return { + success: false, + error: result.error || 'Failed to fetch team fleet', + } + } + + return { + success: true, + teamId: result.data.teamId, + hosts: result.data.hosts.map((host) => ({ + hostId: host.id, + hostName: host.name, + status: host.status, + lastSeen: host.lastSeen, + appVersion: host.appVersion, + gitSha: host.gitSha, + buildTimestamp: host.buildTimestamp, + agents: host.agents.map((agent) => ({ + id: typeof agent.id === 'string' ? agent.id : null, + name: typeof agent.name === 'string' ? agent.name : null, + displayName: typeof agent.displayName === 'string' ? agent.displayName : null, + state: typeof agent.state === 'string' + ? agent.state + : (typeof agent.status === 'string' ? agent.status : null), + })), + activeTasks: host.activeTasks, + slowTasks: host.slowTasks, + convergence: host.convergence ?? null, + })), + } + }) + // Team health monitoring app.get('/health/team', async (request, reply) => { const health = await healthMonitor.getHealth() diff --git a/tests/api.test.ts b/tests/api.test.ts index 0ab3dd8d..ab2626ce 100644 --- a/tests/api.test.ts +++ b/tests/api.test.ts @@ -181,6 +181,14 @@ describe('Health', () => { expect(Array.isArray(body.assignmentRoleNames)).toBe(true) }) + it('GET /team/fleet returns a clear error when cloud fleet truth is unavailable', async () => { + const { status, body } = await req('GET', '/team/fleet') + expect(status === 502 || status === 503).toBe(true) + expect(body.success).toBe(false) + expect(typeof body.error).toBe('string') + expect(body.error.length).toBeGreaterThan(0) + }) + it('GET /health/team includes host-local scope metadata', async () => { const { status, body } = await req('GET', '/health/team') expect(status).toBe(200) From e1870b4aa5017e405ab1f91d6436f32516748f29 Mon Sep 17 00:00:00 2001 From: Kai Date: Tue, 5 May 2026 20:09:45 -0700 Subject: [PATCH 11/14] Restore heartbeat readiness and attestation payloads --- src/cloud.ts | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/src/cloud.ts b/src/cloud.ts index e09937a2..054c5ab0 100644 --- a/src/cloud.ts +++ b/src/cloud.ts @@ -29,6 +29,9 @@ import { REFLECTT_HOME } from './config.js' import { getRequestMetrics } from './request-tracker.js' import { listApprovalQueue, listAgentEvents, listAgentRuns, type AgentRun } from './agent-runs.js' import { getUnpushedTrustEvents, markTrustEventsPushed } from './trust-events.js' +import { getBuildInfo } from './buildInfo.js' +import { getCapabilityReadiness } from './capability-readiness.js' +import { getProvisioningManager } from './provisioning.js' /** * Docker identity guard: detect when a container has inherited cloud @@ -155,6 +158,16 @@ export function _registerImmediateSync(fn: () => void): void { const MAX_PENDING_PUSH_EVENTS = 20 const pendingPushEvents: Array> = [] +function getHeartbeatCapabilityReadiness() { + const provisioning = getProvisioningManager() + const provStatus = provisioning.getStatus() + return getCapabilityReadiness({ + cloudConnected: provStatus.phase === 'ready', + cloudUrl: provStatus.cloudUrl, + webhooks: provStatus.webhooks as Array<{ provider: string; active: boolean }>, + }) +} + /** Queue a canvas_push event for relay to cloud in the next sync cycle. */ export function queueCanvasPushEvent(event: Record): void { pendingPushEvents.push({ ...event, _queuedAt: Date.now() }) @@ -831,11 +844,15 @@ async function sendHeartbeat(): Promise { // "degraded" only if there are actual health issues (e.g., DB errors, high error rate). // Idle agents are normal — not a degraded state. const hostStatus = 'online' as const + const buildInfo = getBuildInfo() + const capabilityReadiness = getHeartbeatCapabilityReadiness() const result = await cloudPost(`/api/hosts/${state.hostId}/heartbeat`, { contractVersion: 'host-heartbeat.v1', status: hostStatus, timestamp: Date.now(), + buildTimestamp: buildInfo.buildTimestamp, + capabilityReadiness, agents: agents.map(a => { const agentAliases = [a.name] const todoCount = tasks.filter(t => t.status === 'todo' && agentAliases.includes(t.assignee || '')).length @@ -896,6 +913,8 @@ async function sendHeartbeat(): Promise { hostName: config.hostName, hostType: config.hostType, uptimeMs: Date.now() - state.startedAt, + version: buildInfo.appVersion, + sha: buildInfo.gitSha, }, }) From 60f3ae664a5fe9e14c4c41ebd7c4b2b7a34772e4 Mon Sep 17 00:00:00 2001 From: Kai Date: Wed, 6 May 2026 11:13:01 -0700 Subject: [PATCH 12/14] build: bake git metadata into production image --- .github/workflows/docker-publish.yml | 25 +++++++++++++++ Dockerfile | 18 +++++++++++ src/buildInfo.ts | 47 ++++++++++++++++++++++------ 3 files changed, 80 insertions(+), 10 deletions(-) diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml index 24e3c208..ffef3eec 100644 --- a/.github/workflows/docker-publish.yml +++ b/.github/workflows/docker-publish.yml @@ -20,6 +20,22 @@ jobs: - name: Checkout uses: actions/checkout@v6 + - name: Compute build metadata + id: build_meta + run: | + echo "app_version=$(node -p \"require('./package.json').version\")" >> "$GITHUB_OUTPUT" + echo "git_sha=${GITHUB_SHA}" >> "$GITHUB_OUTPUT" + echo "git_short_sha=$(git rev-parse --short HEAD)" >> "$GITHUB_OUTPUT" + echo "git_branch=${GITHUB_REF_NAME}" >> "$GITHUB_OUTPUT" + echo "git_author=$(git log -1 --pretty=%an)" >> "$GITHUB_OUTPUT" + echo "git_timestamp=$(git log -1 --pretty=%cI)" >> "$GITHUB_OUTPUT" + echo "build_timestamp=$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$GITHUB_OUTPUT" + { + echo 'git_message<> "$GITHUB_OUTPUT" + - name: Log in to GitHub Container Registry uses: docker/login-action@v3 with: @@ -46,6 +62,15 @@ jobs: push: true tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} + build-args: | + BUILD_APP_VERSION=${{ steps.build_meta.outputs.app_version }} + BUILD_GIT_SHA=${{ steps.build_meta.outputs.git_sha }} + BUILD_GIT_SHORT_SHA=${{ steps.build_meta.outputs.git_short_sha }} + BUILD_GIT_BRANCH=${{ steps.build_meta.outputs.git_branch }} + BUILD_GIT_MESSAGE=${{ steps.build_meta.outputs.git_message }} + BUILD_GIT_AUTHOR=${{ steps.build_meta.outputs.git_author }} + BUILD_GIT_TIMESTAMP=${{ steps.build_meta.outputs.git_timestamp }} + BUILD_TIMESTAMP=${{ steps.build_meta.outputs.build_timestamp }} cache-from: type=gha cache-to: type=gha,mode=max platforms: linux/amd64,linux/arm64 diff --git a/Dockerfile b/Dockerfile index 0b8510e9..93fa4a4f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -26,6 +26,15 @@ RUN npm run build # ── Runtime stage ── FROM node:22-slim +ARG BUILD_APP_VERSION=unknown +ARG BUILD_GIT_SHA=unknown +ARG BUILD_GIT_SHORT_SHA=unknown +ARG BUILD_GIT_BRANCH=unknown +ARG BUILD_GIT_MESSAGE=unknown +ARG BUILD_GIT_AUTHOR=unknown +ARG BUILD_GIT_TIMESTAMP=unknown +ARG BUILD_TIMESTAMP=unknown + WORKDIR /app # Runtime dependency for better-sqlite3 @@ -37,6 +46,7 @@ COPY package.json package-lock.json ./ RUN npm ci --omit=dev COPY --from=build /app/dist/ dist/ +COPY --from=build /app/commit.txt ./commit.txt # Runtime assets (dashboard UI, role defaults, CLI templates) COPY public/ public/ @@ -48,6 +58,14 @@ ENV REFLECTT_HOME=/data ENV NODE_ENV=production ENV PORT=4445 ENV HOST=0.0.0.0 +ENV BUILD_APP_VERSION=${BUILD_APP_VERSION} +ENV BUILD_GIT_SHA=${BUILD_GIT_SHA} +ENV BUILD_GIT_SHORT_SHA=${BUILD_GIT_SHORT_SHA} +ENV BUILD_GIT_BRANCH=${BUILD_GIT_BRANCH} +ENV BUILD_GIT_MESSAGE=${BUILD_GIT_MESSAGE} +ENV BUILD_GIT_AUTHOR=${BUILD_GIT_AUTHOR} +ENV BUILD_GIT_TIMESTAMP=${BUILD_GIT_TIMESTAMP} +ENV BUILD_TIMESTAMP=${BUILD_TIMESTAMP} EXPOSE 4445 diff --git a/src/buildInfo.ts b/src/buildInfo.ts index f54d2ef8..3ea3f521 100644 --- a/src/buildInfo.ts +++ b/src/buildInfo.ts @@ -7,7 +7,7 @@ */ import { execSync } from 'node:child_process' -import { readFileSync } from 'node:fs' +import { readFileSync, existsSync } from 'node:fs' import { resolve, dirname } from 'node:path' export interface BuildInfo { @@ -30,7 +30,6 @@ export interface BuildInfo { // When running from a global install or launchd plist, cwd may point // to an unrelated directory (or a different git repo entirely). import { fileURLToPath } from 'node:url' -import { existsSync } from 'node:fs' const __dirname = dirname(fileURLToPath(import.meta.url)) // Check if we're inside a reflectt-node repo (not an ancestor .git like Homebrew's). @@ -67,6 +66,29 @@ function git(cmd: string): string { } } +function readEnvValue(...keys: string[]): string | null { + for (const key of keys) { + const value = process.env[key]?.trim() + if (value) return value + } + return null +} + +function readBakedCommit(): string | null { + const candidates = [ + resolve(__dirname, '..', 'commit.txt'), + resolve(process.cwd(), 'commit.txt'), + ] + for (const path of candidates) { + try { + if (!existsSync(path)) continue + const value = readFileSync(path, 'utf8').trim() + if (value) return value + } catch { /* try next */ } + } + return null +} + function readPackageVersion(): string { // Try repo root first, then __dirname, then cwd as last resort const candidates = [repoRoot, __dirname, process.cwd()].filter(Boolean) as string[] @@ -83,14 +105,19 @@ function readPackageVersion(): string { // Capture at module load (startup) time const startedAtMs = Date.now() -const appVersion = readPackageVersion() -const gitSha = git('rev-parse HEAD') -const gitShortSha = git('rev-parse --short HEAD') -const gitBranch = git('rev-parse --abbrev-ref HEAD') -const gitMessage = git('log -1 --pretty=%s') -const gitAuthor = git('log -1 --pretty=%an') -const gitTimestamp = git('log -1 --pretty=%ci') -const buildTimestamp = gitTimestamp !== 'unknown' ? gitTimestamp : new Date(startedAtMs).toISOString() +const appVersion = readEnvValue('BUILD_APP_VERSION', 'APP_VERSION') ?? readPackageVersion() +const bakedCommit = readBakedCommit() +const gitSha = readEnvValue('BUILD_GIT_SHA', 'GIT_SHA') ?? git('rev-parse HEAD') +const gitShortSha = readEnvValue('BUILD_GIT_SHORT_SHA', 'GIT_SHORT_SHA') + ?? (gitSha !== 'unknown' ? gitSha.slice(0, 7) : null) + ?? bakedCommit + ?? git('rev-parse --short HEAD') +const gitBranch = readEnvValue('BUILD_GIT_BRANCH', 'GIT_BRANCH') ?? git('rev-parse --abbrev-ref HEAD') +const gitMessage = readEnvValue('BUILD_GIT_MESSAGE', 'GIT_MESSAGE') ?? git('log -1 --pretty=%s') +const gitAuthor = readEnvValue('BUILD_GIT_AUTHOR', 'GIT_AUTHOR') ?? git('log -1 --pretty=%an') +const gitTimestamp = readEnvValue('BUILD_GIT_TIMESTAMP', 'GIT_TIMESTAMP') ?? git('log -1 --pretty=%ci') +const buildTimestamp = readEnvValue('BUILD_TIMESTAMP') + ?? (gitTimestamp !== 'unknown' ? gitTimestamp : new Date(startedAtMs).toISOString()) export function getBuildInfo(): BuildInfo { return { From 4d5a6038acb97932c781f844b1832a35b1bc1b08 Mon Sep 17 00:00:00 2001 From: Kai Date: Fri, 8 May 2026 21:11:14 -0700 Subject: [PATCH 13/14] fix(executionSweeper): PR-merged tasks bypass dupeErr artifact-rejection requeue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two paths were incorrectly requeuing PR-merged tasks to todo: 1. Artifact-rejection (line ~362): missing qa_bundle after 24h in validating → now closes to done if pr_merged, only requeues if not merged 2. Drift-repair dupeErr block (line ~428): approved but dupeErr blocks close → PR merge is canonical, so pr_merged + approved → close to done Canonical rule: PR merge = task completion signal, not artifact completeness. qa_bundle artifacts are evidence of work, not completion criteria. Refs: task-1778298996028-ah73bdq3m --- src/executionSweeper.ts | 103 +++++++++++++++++++++++++++++++--------- 1 file changed, 80 insertions(+), 23 deletions(-) diff --git a/src/executionSweeper.ts b/src/executionSweeper.ts index ec0828f0..afcfc4d9 100644 --- a/src/executionSweeper.ts +++ b/src/executionSweeper.ts @@ -356,30 +356,58 @@ export async function sweepValidatingQueue(): Promise { const enteredAt = (meta.entered_validating_at as number) || task.updatedAt const ageInValidating = now - enteredAt + // PR-merged tasks: close to done regardless of artifact completeness. + // PR merge is the canonical completion signal, not artifact presence. + const prMerged = !!(meta.pr_merged) if (ageInValidating >= ARTIFACT_GRACE_MS && !hasRequiredArtifacts(meta)) { - try { - taskManager.updateTask(task.id, { - status: 'todo', - metadata: { - ...meta, - artifact_rejected: true, - artifact_rejected_at: now, - artifact_reject_reason: 'Missing required artifacts (PR or qa_bundle) after 24h grace period', - review_state: undefined, - reviewer_approved: undefined, - }, - } as any) - artifactRejectedIds.add(task.id) - escalated.delete(task.id) - logDryRun('artifact_rejected', `${task.id} — no artifacts after ${msToMinutes(ageInValidating)}m in validating`) + if (prMerged) { + try { + await taskManager.updateTask(task.id, { + status: 'done', + metadata: { + ...meta, + auto_closed: true, + auto_closed_at: now, + auto_close_reason: 'sweeper_pr_merged_artifact_missing', + completed_at: now, + }, + } as any) + autoClosedIds.add(task.id) + escalated.delete(task.id) + logDryRun('artifact_rejected_pr_merged', `${task.id} — PR merged, closing to done despite missing qa_bundle`) + chatManager.sendMessage({ + from: 'system', + channel: 'task-notifications', + content: `✅ Auto-closed "${task.title}" (${task.id}) — PR merged, qa_bundle missing but completion honored. (Automated)`, + }).catch(() => {}) + } catch (err) { + logDryRun('artifact_rejected_pr_merged_failed', `${task.id} — ${String(err)}`) + } + } else { + try { + taskManager.updateTask(task.id, { + status: 'todo', + metadata: { + ...meta, + artifact_rejected: true, + artifact_rejected_at: now, + artifact_reject_reason: 'Missing required artifacts (PR or qa_bundle) after 24h grace period', + review_state: undefined, + reviewer_approved: undefined, + }, + } as any) + artifactRejectedIds.add(task.id) + escalated.delete(task.id) + logDryRun('artifact_rejected', `${task.id} — no artifacts after ${msToMinutes(ageInValidating)}m in validating`) - chatManager.sendMessage({ - from: 'system', - channel: 'task-notifications', - content: `⚠️ Auto-rejected "${task.title}" (${task.id}) back to todo — missing required artifacts (PR or qa_bundle) after 24h in validating. @${task.assignee || 'unassigned'} please add artifacts and resubmit.`, - }).catch(() => {}) - } catch (err) { - logDryRun('artifact_reject_failed', `${task.id} — ${String(err)}`) + chatManager.sendMessage({ + from: 'system', + channel: 'task-notifications', + content: `⚠️ Auto-rejected "${task.title}" (${task.id}) back to todo — missing required artifacts (PR or qa_bundle) after 24h in validating. @${task.assignee || 'unassigned'} please add artifacts and resubmit.`, + }).catch(() => {}) + } catch (err) { + logDryRun('artifact_reject_failed', `${task.id} — ${String(err)}`) + } } } } @@ -396,6 +424,35 @@ export async function sweepValidatingQueue(): Promise { const reviewState = meta.review_state as string | undefined const reviewerApproved = meta.reviewer_approved === true if (reviewState === 'approved' || reviewerApproved) { + // PR merge is canonical completion — close to done even if dupeErr blocks artifact path + const prMerged = !!(meta.pr_merged) + if (prMerged) { + try { + await taskManager.updateTask(task.id, { + status: 'done', + metadata: { + ...meta, + auto_closed: true, + auto_closed_at: now, + auto_close_reason: 'sweeper_pr_merged_approved_dup_ref', + completed_at: now, + dupeErr_resolved: getDuplicateClosureCanonicalRefError(meta) ?? null, + }, + } as any) + autoClosedIds.add(task.id) + escalated.delete(task.id) + logDryRun('drift_repair_pr_merged_closed', `${task.id} — PR merged + approved, closing despite dupeErr`) + chatManager.sendMessage({ + from: 'system', + channel: 'task-notifications', + content: `✅ Auto-closed "${task.title}" (${task.id}) — PR merged + approved, dupeErr noted but completion honored. (Automated)`, + }).catch(() => {}) + } catch (err) { + logDryRun('drift_repair_pr_merged_failed', `${task.id} — ${String(err)}`) + } + continue + } + const dupeErr = getDuplicateClosureCanonicalRefError(meta) if (dupeErr) { try { @@ -405,7 +462,7 @@ export async function sweepValidatingQueue(): Promise { ...meta, auto_close_blocked: true, auto_close_blocked_at: now, - auto_close_blocked_reason: dupeErr, + auto_close_reason: dupeErr, review_state: 'needs_author', reviewer_approved: undefined, reviewer_decision: undefined, From 8fbcd958471669a6c129c5a5b0ef48f2a6d3ad21 Mon Sep 17 00:00:00 2001 From: Kai Date: Sun, 10 May 2026 03:22:25 -0700 Subject: [PATCH 14/14] fix(getNextTask): skip parked_pass, board_truth, blocked_external tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit getNextTask() was surfacing closed/stale tasks because it had no awareness of: - review_state=parked_pass/fail (QA-complete tasks) - board_truth metadata (explicit per-task lane routing) - blocked_external flags - reopen=true (tasks being actively reworked) These filters were already missing when the sweeper fix (PR #2769) closed the bounce path — tasks that landed back in todo kept getting picked up by the next-task selector. Combined with PR #2769, this fully addresses the structural routing bug reported by link (M cut bounced 13+ times). Ref: task-1778408493344-cgd00jx71 --- src/tasks.ts | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/tasks.ts b/src/tasks.ts index f5667603..537f651f 100644 --- a/src/tasks.ts +++ b/src/tasks.ts @@ -2077,6 +2077,20 @@ class TaskManager { const requestingNames = new Set(getAgentAliases(agent)) tasks = tasks.filter(t => { const meta = t.metadata as Record | undefined + + // Skip parked tasks — review_state=parked_pass/fail means QA is done + const reviewState = meta?.review_state + if (reviewState === 'parked_pass' || reviewState === 'parked_fail') return false + + // Skip explicitly reopened tasks — reopen=true means someone is working it + if (meta?.reopen === true) return false + + // Skip tasks with explicit board truth routing — board_truth is authoritative + if (meta?.board_truth && typeof meta.board_truth === 'string') return false + + // Skip blocked external tasks + if (meta?.blocked_external === true) return false + const handoffTo = meta?.last_transition?.handoff_to || meta?.transition?.handoff_to if (!handoffTo || typeof handoffTo !== 'string') return true // no handoff — anyone can pull return requestingNames.has(handoffTo.toLowerCase()) // only the handoff target can pull