Repository navigation
feat: inbox-driven session management — watcher, routing headers, daemon integration (#574) - #771
Conversation
Parse bracket-delimited routing headers from Omni-relayed messages and resolve deterministic session keys for per-user/per-chat routing. Implements Group 2 of inbox-driven-sessions wish (#574).
- Re-apply src/lib/inbox-watcher.ts (143 lines) with DI pattern and backoff logic - Re-apply src/lib/inbox-watcher.test.ts (184 lines, 8 tests) - Add listTeamsWithUnreadInbox() to claude-native-teams.ts (scans native inboxes) - Export isTeamActive from team-auto-spawn.ts for inbox-watcher consumption Cherry-picked from commit 040b667, adapted to current API signatures. Ref: #574
- Add listTeamsWithUnreadInbox() to claude-native-teams.ts for inbox scanning - Export isTeamActive from team-auto-spawn.ts for inbox-watcher consumption
Exports will be consumed by daemon wiring (Group 3).
- Import and start/stop inbox watcher in scheduler-daemon alongside existing scheduler, heartbeat, and orphan timers - Respect GENIE_INBOX_POLL_MS=0 to disable inbox watching in daemon - Add inbox watcher state to daemon status output - Convert genie inbox to command group with list (default) and watch - genie inbox watch runs inbox watcher in foreground with Ctrl+C - Integrate routing header parser into inbox watcher for session key resolution (falls back to team name when no header present) - Add firstUnreadText to listTeamsWithUnreadInbox return type - Add tests for routing header integration and fallback behavior
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the system's inbox-driven session management capabilities. It re-establishes the automatic spawning of team-leads when new messages arrive in their inboxes, ensuring timely responses. A new, robust mechanism for parsing message routing headers is introduced, which allows for consistent and deterministic session key resolution. These features are seamlessly integrated into the daemon, providing a more autonomous and efficient operational flow for managing team interactions. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces an "Inbox Watcher" feature, designed to automatically spawn team-leads for inactive teams that have unread messages in their inboxes. Key changes include new modules for listing teams with unread messages (claude-native-teams.ts), the core inbox watcher logic (inbox-watcher.ts), and a utility for parsing routing headers (routing-header.ts), all accompanied by dedicated test files. The main scheduler daemon (scheduler-daemon.ts) has been updated to integrate and manage the lifecycle of this new watcher. Additionally, new CLI commands (genie inbox list and genie inbox watch) have been added to facilitate interaction and monitoring of the inbox watcher. Feedback suggests improving the performance of team inbox scanning by processing teams in parallel, simplifying the stripRoutingHeader function, and optimizing dependency imports and logging consistency within the inbox watch command.
| const results: Array<{ | ||
| teamName: string; | ||
| unreadCount: number; | ||
| workingDir: string | null; | ||
| firstUnreadText: string | null; | ||
| }> = []; | ||
|
|
||
| for (const name of teamDirs) { | ||
| // Read inbox messages | ||
| const inboxFile = join(base, name, 'inboxes', 'team-lead.json'); | ||
| let messages: NativeInboxMessage[]; | ||
| try { | ||
| const content = await readFile(inboxFile, 'utf-8'); | ||
| messages = JSON.parse(content); | ||
| } catch { | ||
| continue; // No inbox or invalid JSON | ||
| } | ||
|
|
||
| if (!Array.isArray(messages)) continue; | ||
|
|
||
| const unread = messages.filter((m) => m.read === false); | ||
| if (unread.length === 0) continue; | ||
|
|
||
| // Get workingDir from config.json → members → team-lead → cwd | ||
| let workingDir: string | null = null; | ||
| try { | ||
| const cfgContent = await readFile(join(base, name, 'config.json'), 'utf-8'); | ||
| const config: NativeTeamConfig = JSON.parse(cfgContent); | ||
| const leadMember = config.members.find((m) => m.name === 'team-lead' || m.agentId.startsWith('team-lead@')); | ||
| if (leadMember?.cwd) { | ||
| workingDir = leadMember.cwd; | ||
| } | ||
| } catch { | ||
| // Config missing or malformed — workingDir stays null | ||
| } | ||
|
|
||
| results.push({ | ||
| teamName: name, | ||
| unreadCount: unread.length, | ||
| workingDir, | ||
| firstUnreadText: unread[0]?.text ?? null, | ||
| }); | ||
| } | ||
|
|
||
| return results; | ||
| } |
There was a problem hiding this comment.
The current implementation processes each team directory sequentially using a for...of loop with await inside. This can be inefficient if there are many teams, as each team's file I/O operations will block the processing of the next. To improve performance, you can process teams in parallel using Promise.all.
const teamPromises = teamDirs.map(async (name) => {
// Read inbox messages
const inboxFile = join(base, name, 'inboxes', 'team-lead.json');
let messages: NativeInboxMessage[];
try {
const content = await readFile(inboxFile, 'utf-8');
messages = JSON.parse(content);
} catch {
return null; // No inbox or invalid JSON
}
if (!Array.isArray(messages)) return null;
const unread = messages.filter((m) => m.read === false);
if (unread.length === 0) return null;
// Get workingDir from config.json → members → team-lead → cwd
let workingDir: string | null = null;
try {
const cfgContent = await readFile(join(base, name, 'config.json'), 'utf-8');
const config: NativeTeamConfig = JSON.parse(cfgContent);
const leadMember = config.members.find((m) => m.name === 'team-lead' || m.agentId.startsWith('team-lead@'));
if (leadMember?.cwd) {
workingDir = leadMember.cwd;
}
} catch {
// Config missing or malformed — workingDir stays null
}
return {
teamName: name,
unreadCount: unread.length,
workingDir,
firstUnreadText: unread[0]?.text ?? null,
};
});
const resultsWithNulls = await Promise.all(teamPromises);
return resultsWithNulls.filter((r): r is NonNullable<typeof r> => r !== null);
}| export function stripRoutingHeader(text: string): string { | ||
| if (!text) return text; | ||
|
|
||
| const firstNewline = text.indexOf('\n'); | ||
| if (firstNewline === -1) { | ||
| // Single line — if it's a header, body is empty | ||
| const firstLine = text.trim(); | ||
| return firstLine.match(/^\[(.+)\]$/) ? '' : text; | ||
| } | ||
|
|
||
| const firstLine = text.slice(0, firstNewline).trim(); | ||
| if (firstLine.match(/^\[(.+)\]$/)) { | ||
| return text.slice(firstNewline + 1); | ||
| } | ||
|
|
||
| return text; | ||
| } |
There was a problem hiding this comment.
The logic in stripRoutingHeader is a bit repetitive. You can simplify it by first identifying the first line of the text and then checking if it's a routing header, which consolidates the logic for single-line and multi-line messages.
export function stripRoutingHeader(text: string): string {
if (!text) return text;
const firstNewline = text.indexOf('\n');
const firstLine = (firstNewline === -1 ? text : text.slice(0, firstNewline)).trim();
if (firstLine.match(/^\[(.+)\]$/)) {
return firstNewline === -1 ? '' : text.slice(firstNewline + 1);
}
return text;
}| const { checkInboxes, getInboxPollIntervalMs, startInboxWatcher, stopInboxWatcher } = await import( | ||
| '../lib/inbox-watcher.js' | ||
| ); | ||
|
|
||
| const pollMs = getInboxPollIntervalMs(); | ||
| if (pollMs === 0) { | ||
| console.log('Inbox watcher is disabled (GENIE_INBOX_POLL_MS=0)'); | ||
| process.exit(0); | ||
| } | ||
|
|
||
| console.log(`Inbox watcher starting (poll every ${pollMs / 1000}s)`); | ||
| console.log('Press Ctrl+C to stop.\n'); | ||
|
|
||
| // Run an initial check immediately | ||
| const initial = await checkInboxes(); | ||
| if (initial.length > 0) { | ||
| console.log(`[inbox-watcher] Spawned team-leads for: ${initial.join(', ')}`); | ||
| } | ||
|
|
||
| // Start the polling loop with visible logging | ||
| const handle = startInboxWatcher({ | ||
| listTeamsWithUnreadInbox: (await import('../lib/claude-native-teams.js')).listTeamsWithUnreadInbox, | ||
| isTeamActive: async (teamName) => { | ||
| const { isTeamActive } = await import('../lib/team-auto-spawn.js'); | ||
| return isTeamActive(teamName); | ||
| }, | ||
| ensureTeamLead: async (teamName, workingDir) => { | ||
| const { ensureTeamLead } = await import('../lib/team-auto-spawn.js'); | ||
| const result = await ensureTeamLead(teamName, workingDir); | ||
| console.log(`[inbox-watcher] Spawned team-lead for "${teamName}" in ${workingDir}`); | ||
| return result; | ||
| }, | ||
| warn: (msg) => console.log(msg), | ||
| }); |
There was a problem hiding this comment.
The inbox watch command can be improved in a few ways:
- Dependency Imports: The dynamic
import()calls inside the dependency functions (isTeamActive,ensureTeamLead) are inefficient as they re-import the module on every invocation. It's better to import them once at the top of theactionhandler. - Logging Consistency: The initial
checkInboxes()call uses default dependencies, while thestartInboxWatcheruses custom ones with different logging behavior. This leads to inconsistent output. You should define a singledepsobject with console logging and use it for both the initial check and the watcher loop.
This refactoring will make the code cleaner, more efficient, and ensure consistent logging.
const { checkInboxes, getInboxPollIntervalMs, startInboxWatcher, stopInboxWatcher } = await import(
'../lib/inbox-watcher.js'
);
const { listTeamsWithUnreadInbox } = await import('../lib/claude-native-teams.js');
const autoSpawn = await import('../lib/team-auto-spawn.js');
const pollMs = getInboxPollIntervalMs();
if (pollMs === 0) {
console.log('Inbox watcher is disabled (GENIE_INBOX_POLL_MS=0)');
process.exit(0);
}
console.log(`Inbox watcher starting (poll every ${pollMs / 1000}s)`);
console.log('Press Ctrl+C to stop.\n');
// Define dependencies for consistent logging
const deps = {
listTeamsWithUnreadInbox,
isTeamActive: autoSpawn.isTeamActive,
ensureTeamLead: async (teamName: string, workingDir: string) => {
const result = await autoSpawn.ensureTeamLead(teamName, workingDir);
console.log(`[inbox-watcher] Spawned team-lead for "${teamName}" in ${workingDir}`);
return result;
},
warn: (msg: string) => console.log(msg),
};
// Run an initial check immediately
await checkInboxes(deps);
// Start the polling loop
const handle = startInboxWatcher(deps);
Summary
Restores and extends the inbox-driven session management system:
routing-header.ts. Parses[channel:telegram instance:xxx chat:123 ...]headers. Resolves deterministic session keys per sender/chat/thread.genie daemon start/stop.genie inbox watchfor foreground mode.claude-native-teams.ts.Test plan
genie inbox watchpolls and logs activitygenie daemon startstarts both scheduler + inbox watcherFixes #574