mirror of
https://github.com/Ark0N/Codeman.git
synced 2026-09-30 12:39:42 +02:00
arkonandClaude Opus 4.7
02e2f3e8b5
Knip-driven cleanup. All changes verified with tsc --noEmit, lint, and
build.
Removed (zero consumers):
- VERIFICATION_PROMPT constant + its barrel re-export
- createInitialOrchestratorPersistState factory
- transcriptWatcher singleton export
- createAnsiPatternFull / createAnsiPatternSimple factories
- TimerInfo interface + unused AiCheckResult/AiPlanCheckResult imports
in respawn-controller.ts
- 35 unused Zod z.infer \`*Input\` types in schemas.ts
- Dead re-exports: SessionMode from session.ts, AuthSessionRecord from
web/ports/index.ts, EnhancedPlanTask/CheckpointReview from
ralph-tracker.ts, 7 unused entries in utils/index.ts
- 14 event/config interfaces that lived only as JSDoc hints (no TS type
position usage): Session/Respawn/RalphLoop/RalphTracker/
SessionManager/SessionAutoOps/Subagent/TaskQueue/TaskTracker/
TranscriptWatcher/Image/OrchestratorLoop Events + RespawnPreset +
SessionOutput
Narrowed to module scope (kept but no longer exported):
- buildPermissionArgs in session-cli-builder.ts
- 28 type/interface declarations used only within their own file:
Ai{Idle,Plan}Check{Config,State}, BashToolParser{Events,Config},
FileStream/CreateStream{Options,Result}, PlanSubagentEvent,
SubagentCallback, RalphLoopConfig, RalphLoop{Events,Options},
ActiveTimerInfo, DetectionStatus, ActionLogEntry, AutoOpsCallbacks,
TunnelStatus, Timer/LRUMap/StaleExpirationMap Options, AuthState,
SessionListenerDeps, SseStreamManagerDeps, and 8 more
Docs: CLAUDE.md advice for global-regex `lastIndex` now points to the
remaining `execPattern()` helper instead of the deleted factories.
Knip delta: unused files 42→0, unused exports 161→16, unused types 92→0.
The 16 remaining exports are a mobile-test helper toolkit intentionally
kept for upcoming tests.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1703 lines
58 KiB
TypeScript
1703 lines
58 KiB
TypeScript
/**
|
|
* @fileoverview Subagent Watcher - Real-time monitoring of Claude Code background agents.
|
|
*
|
|
* Watches `~/.claude/projects/{project}/{session}/subagents/agent-{id}.jsonl` files
|
|
* and emits structured events for tool calls, progress, messages, and tool results.
|
|
* Also detects Agent Teams teammates (distinguished by `<teammate-message>` in description).
|
|
*
|
|
* Key exports:
|
|
* - `SubagentWatcher` class — singleton watcher, extends EventEmitter
|
|
* - `subagentWatcher` — pre-instantiated singleton instance
|
|
* - `SubagentInfo`, `SubagentToolCall`, `SubagentProgress`, `SubagentMessage`,
|
|
* `SubagentToolResult`, `SubagentTranscriptEntry` — data interfaces
|
|
* - `SubagentEvents` — typed event map
|
|
*
|
|
* Watched patterns: `~/.claude/projects/{project}/{session}/subagents/agent-{id}.jsonl`
|
|
* Parses JSONL entries: user/assistant messages, tool_use/tool_result blocks, progress events.
|
|
* Tracks per-agent: status, token counts, model, description, tool call count, liveness (PID).
|
|
*
|
|
* @dependencies config/map-limits (MAX_TRACKED_AGENTS, PENDING_TOOL_CALL_TTL_MS),
|
|
* config/buffer-limits (FILE_PEEK_BYTES), utils (CleanupManager, KeyedDebouncer)
|
|
* @consumedby web/server (SSE broadcast), session (subagent-session correlation)
|
|
* @emits subagent:discovered, subagent:updated, subagent:tool_call, subagent:tool_result,
|
|
* subagent:progress, subagent:message, subagent:completed
|
|
*
|
|
* @module subagent-watcher
|
|
*/
|
|
|
|
import { EventEmitter } from 'node:events';
|
|
import { watch, existsSync, FSWatcher } from 'node:fs';
|
|
import { createReadStream } from 'node:fs';
|
|
import { createInterface } from 'node:readline';
|
|
import { homedir } from 'node:os';
|
|
import { join, basename } from 'node:path';
|
|
import { execFile } from 'node:child_process';
|
|
import { readFile, readdir, stat as statAsync } from 'node:fs/promises';
|
|
import { PENDING_TOOL_CALL_TTL_MS, MAX_PENDING_TOOL_CALLS, MAX_TRACKED_AGENTS } from './config/map-limits.js';
|
|
import { STALE_DATA_MAX_AGE_MS } from './config/server-timing.js';
|
|
import { FILE_PEEK_BYTES } from './config/buffer-limits.js';
|
|
import { CleanupManager, KeyedDebouncer } from './utils/index.js';
|
|
|
|
// ========== Types ==========
|
|
|
|
export interface SubagentInfo {
|
|
agentId: string;
|
|
sessionId: string;
|
|
projectHash: string;
|
|
filePath: string;
|
|
startedAt: string;
|
|
lastActivityAt: number;
|
|
status: 'active' | 'idle' | 'completed';
|
|
toolCallCount: number;
|
|
entryCount: number;
|
|
fileSize: number;
|
|
description?: string; // Task description from first user message
|
|
model?: string; // Full model name (e.g., "claude-sonnet-4-20250514")
|
|
modelShort?: 'haiku' | 'sonnet' | 'opus'; // Short model identifier
|
|
totalInputTokens?: number; // Running total of input tokens
|
|
totalOutputTokens?: number; // Running total of output tokens
|
|
pid?: number; // Cached process ID for fast liveness checks
|
|
}
|
|
|
|
export interface SubagentToolCall {
|
|
agentId: string;
|
|
sessionId: string;
|
|
timestamp: string;
|
|
tool: string;
|
|
input: Record<string, unknown>;
|
|
toolUseId?: string; // For linking to tool_result
|
|
fullInput: Record<string, unknown>; // Complete input object (input is truncated for display)
|
|
}
|
|
|
|
export interface SubagentProgress {
|
|
agentId: string;
|
|
sessionId: string;
|
|
timestamp: string;
|
|
progressType: 'query_update' | 'search_results_received' | string;
|
|
query?: string;
|
|
resultCount?: number;
|
|
hookEvent?: string; // e.g., "PostToolUse"
|
|
hookName?: string; // e.g., "PostToolUse:Read"
|
|
}
|
|
|
|
export interface SubagentMessage {
|
|
agentId: string;
|
|
sessionId: string;
|
|
timestamp: string;
|
|
role: 'user' | 'assistant';
|
|
text: string;
|
|
}
|
|
|
|
export interface SubagentTranscriptEntry {
|
|
type: 'user' | 'assistant' | 'progress';
|
|
timestamp: string;
|
|
agentId: string;
|
|
sessionId: string;
|
|
message?: {
|
|
role: string;
|
|
model?: string; // Model used for this message (e.g., "claude-sonnet-4-20250514")
|
|
usage?: {
|
|
input_tokens?: number;
|
|
output_tokens?: number;
|
|
};
|
|
content:
|
|
| string
|
|
| Array<{
|
|
type: 'text' | 'tool_use' | 'tool_result';
|
|
text?: string;
|
|
name?: string;
|
|
id?: string; // tool_use_id for tool_use blocks
|
|
tool_use_id?: string; // tool_use_id for tool_result blocks
|
|
input?: Record<string, unknown>;
|
|
content?: string | Array<{ type: string; text?: string }>; // tool_result content
|
|
is_error?: boolean; // For tool_result errors
|
|
}>;
|
|
};
|
|
data?: {
|
|
type: string;
|
|
query?: string;
|
|
resultCount?: number;
|
|
hookEvent?: string; // e.g., "PostToolUse"
|
|
hookName?: string; // e.g., "PostToolUse:Read"
|
|
tool_name?: string; // Tool name for hook events
|
|
};
|
|
}
|
|
|
|
export interface SubagentToolResult {
|
|
agentId: string;
|
|
sessionId: string;
|
|
timestamp: string;
|
|
toolUseId: string;
|
|
tool?: string; // Tool name (looked up from pending)
|
|
preview: string; // First 500 chars of result
|
|
contentLength: number; // Total length of result
|
|
isError: boolean; // Whether result is an error
|
|
}
|
|
|
|
// ========== Constants ==========
|
|
|
|
const CLAUDE_PROJECTS_DIR = join(homedir(), '.claude/projects');
|
|
const IDLE_TIMEOUT_MS = 30000; // Consider agent idle after 30s of no activity
|
|
const POLL_INTERVAL_MS = 1000; // Base poll interval (lightweight checks)
|
|
const FULL_SCAN_EVERY_N_POLLS = 5; // Full directory traversal every 5th poll (5s)
|
|
const LIVENESS_CHECK_MS = 10000; // Check if subagent processes are still alive every 10s
|
|
const FILE_ALIVE_THRESHOLD_MS = 30000; // File mtime within 30s = agent alive (primary check)
|
|
const STALE_COMPLETED_MAX_AGE_MS = STALE_DATA_MAX_AGE_MS; // Remove completed agents older than 1 hour
|
|
const STALE_IDLE_MAX_AGE_MS = 4 * 60 * 60 * 1000; // Remove idle agents older than 4 hours
|
|
const STARTUP_MAX_FILE_AGE_MS = 4 * 60 * 60 * 1000; // Only load files modified in last 4 hours on startup
|
|
|
|
// Internal Claude Code agent patterns to filter out (not real user-initiated subagents)
|
|
const INTERNAL_AGENT_PATTERNS = [
|
|
/^\[?SUGGESTION MODE/i, // Claude Code's internal suggestion mode
|
|
/^Suggest what user might/i, // Suggestion mode prompt variant
|
|
/aprompt/i, // Internal prompt agent (anywhere in string)
|
|
/^a\s?prompt/i, // Variants of internal prompt agent
|
|
/^prompt$/i, // Just "prompt"
|
|
];
|
|
|
|
// Minimum description length - very short descriptions are likely internal or malformed
|
|
const MIN_DESCRIPTION_LENGTH = 5;
|
|
|
|
// Display/preview length constants
|
|
const TEXT_PREVIEW_LENGTH = 200; // Length for text previews in tool results
|
|
const USER_TEXT_PREVIEW_LENGTH = 80; // Length for user message previews
|
|
const SMART_TITLE_MAX_LENGTH = 45; // Max length for smart title extraction
|
|
const MESSAGE_TEXT_LIMIT = 500; // Max length for message text content
|
|
const COMMAND_DISPLAY_LENGTH = 60; // Max length for command display
|
|
const INPUT_TRUNCATE_LENGTH = 100; // Max length for input value truncation
|
|
const FILE_CONTENT_DEBOUNCE_MS = 100; // Debounce delay for file content updates
|
|
|
|
// ========== SubagentWatcher Class ==========
|
|
|
|
export class SubagentWatcher extends EventEmitter {
|
|
private filePositions = new Map<string, number>();
|
|
private dirWatchers = new Map<string, FSWatcher>();
|
|
// Per-file debouncer for directory watcher (replaces per-file FSWatchers)
|
|
private fileDeb = new KeyedDebouncer(FILE_CONTENT_DEBOUNCE_MS);
|
|
private agentInfo = new Map<string, SubagentInfo>();
|
|
private idleDeb = new KeyedDebouncer(IDLE_TIMEOUT_MS);
|
|
private cleanup = new CleanupManager();
|
|
private _isRunning = false;
|
|
private knownSubagentDirs = new Set<string>();
|
|
// Map of agentId -> Map of toolUseId -> { toolName, timestamp } (for linking tool_result to tool_call)
|
|
// Includes timestamp for TTL-based cleanup of orphaned entries
|
|
private pendingToolCalls = new Map<string, Map<string, { toolName: string; timestamp: number }>>();
|
|
// Guard to prevent concurrent liveness checks (prevents duplicate completed events)
|
|
private _isCheckingLiveness = false;
|
|
// Counter for throttling full directory scans (only scan every FULL_SCAN_EVERY_N_POLLS)
|
|
private _pollCount = 0;
|
|
// Short-lived cache for parsed parent transcript descriptions (TTL: 5s)
|
|
// Key: "{projectHash}/{sessionId}", Value: { descriptions: Map<agentId, description>, timestamp }
|
|
private parentDescriptionCache = new Map<string, { descriptions: Map<string, string>; timestamp: number }>();
|
|
// Store error handlers for FSWatchers to enable proper cleanup (prevent memory leaks)
|
|
private dirWatcherErrorHandlers = new Map<string, (error: Error) => void>();
|
|
// Map filePath → { projectHash, sessionId } for directory watcher file-change handling
|
|
private fileAgentContext = new Map<string, { projectHash: string; sessionId: string }>();
|
|
|
|
constructor() {
|
|
super();
|
|
this.setMaxListeners(50);
|
|
}
|
|
|
|
/**
|
|
* Check if a description matches internal Claude Code agent patterns.
|
|
* These are not real user-initiated subagents and should be filtered out.
|
|
* Also filters out very short descriptions that are likely internal or malformed.
|
|
*/
|
|
private isInternalAgent(description: string | undefined): boolean {
|
|
if (!description) return false;
|
|
// Filter out very short descriptions (likely internal or malformed)
|
|
if (description.length < MIN_DESCRIPTION_LENGTH) return true;
|
|
return INTERNAL_AGENT_PATTERNS.some((pattern) => pattern.test(description));
|
|
}
|
|
|
|
/**
|
|
* Mark a subagent as completed: clear PID, set status, clean up pending tool calls, emit event.
|
|
*/
|
|
private markSubagentAsCompleted(info: SubagentInfo): void {
|
|
info.pid = undefined;
|
|
info.status = 'completed';
|
|
this.pendingToolCalls.delete(info.agentId);
|
|
this.emit('subagent:completed', info);
|
|
}
|
|
|
|
/**
|
|
* Extract text from message content, handling both string and array formats.
|
|
* For array content, returns the text from the first 'text' block.
|
|
*/
|
|
private extractFirstTextContent(
|
|
content: string | Array<{ type: string; text?: string }> | undefined
|
|
): string | undefined {
|
|
if (!content) return undefined;
|
|
if (typeof content === 'string') {
|
|
const trimmed = content.trim();
|
|
return trimmed.length > 0 ? trimmed : undefined;
|
|
}
|
|
if (Array.isArray(content)) {
|
|
const firstContent = content[0];
|
|
if (firstContent?.type === 'text' && firstContent.text) {
|
|
const trimmed = firstContent.text.trim();
|
|
return trimmed.length > 0 ? trimmed : undefined;
|
|
}
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
/**
|
|
* Process a tool_result content block: look up pending tool call, emit tool_result event.
|
|
*/
|
|
private emitToolResult(
|
|
content: { tool_use_id: string; content?: string | Array<{ type: string; text?: string }>; is_error?: boolean },
|
|
agentId: string,
|
|
sessionId: string,
|
|
timestamp: string
|
|
): void {
|
|
const resultContent = this.extractToolResultContent(content.content);
|
|
const agentPendingCalls = this.pendingToolCalls.get(agentId);
|
|
const pendingCall = agentPendingCalls?.get(content.tool_use_id);
|
|
const toolName = pendingCall?.toolName;
|
|
// Delete after lookup to prevent memory leak
|
|
agentPendingCalls?.delete(content.tool_use_id);
|
|
|
|
const toolResult: SubagentToolResult = {
|
|
agentId,
|
|
sessionId,
|
|
timestamp,
|
|
toolUseId: content.tool_use_id,
|
|
tool: toolName,
|
|
preview: resultContent.substring(0, MESSAGE_TEXT_LIMIT),
|
|
contentLength: resultContent.length,
|
|
isError: content.is_error || false,
|
|
};
|
|
this.emit('subagent:tool_result', toolResult);
|
|
}
|
|
|
|
/**
|
|
* Find the oldest inactive (non-active) agent for LRU eviction.
|
|
* Returns the agent ID of the oldest inactive agent, or null if all are active.
|
|
*/
|
|
private findOldestInactiveAgent(): string | null {
|
|
let oldestId: string | null = null;
|
|
let oldestTime = Infinity;
|
|
for (const [id, existing] of this.agentInfo) {
|
|
if (existing.status !== 'active' && existing.lastActivityAt < oldestTime) {
|
|
oldestTime = existing.lastActivityAt;
|
|
oldestId = id;
|
|
}
|
|
}
|
|
return oldestId;
|
|
}
|
|
|
|
/**
|
|
* Extract short model identifier from full model name
|
|
*/
|
|
private extractModelShort(model: string): 'haiku' | 'sonnet' | 'opus' | undefined {
|
|
const lower = model.toLowerCase();
|
|
if (lower.includes('haiku')) return 'haiku';
|
|
if (lower.includes('sonnet')) return 'sonnet';
|
|
if (lower.includes('opus')) return 'opus';
|
|
return undefined;
|
|
}
|
|
|
|
// ========== Public API ==========
|
|
|
|
/**
|
|
* Start watching for subagent activity
|
|
*/
|
|
start(): void {
|
|
if (this._isRunning) return;
|
|
this._isRunning = true;
|
|
|
|
// Initial scan (always runs immediately)
|
|
this._pollCount = 0;
|
|
this.scanForSubagents().catch((err) => this.emit('subagent:error', err as Error));
|
|
|
|
// Periodic scan for new subagent directories
|
|
// Full directory traversal only every FULL_SCAN_EVERY_N_POLLS polls (~5s)
|
|
// FSWatchers handle known directories between full scans
|
|
this.cleanup.setInterval(
|
|
() => {
|
|
this._pollCount++;
|
|
if (this._pollCount % FULL_SCAN_EVERY_N_POLLS === 0) {
|
|
this.scanForSubagents().catch((err) => this.emit('subagent:error', err as Error));
|
|
}
|
|
},
|
|
POLL_INTERVAL_MS,
|
|
{ description: 'subagent directory poll' }
|
|
);
|
|
|
|
// Periodic liveness check for active subagents
|
|
this.startLivenessChecker();
|
|
}
|
|
|
|
/**
|
|
* Start periodic liveness checker
|
|
* Detects when subagent processes have exited but status is still active/idle.
|
|
*
|
|
* Uses a 3-tier check to minimize cost:
|
|
* 1. File mtime (stat ~0.3ms/agent) — if transcript modified recently, agent is alive
|
|
* 2. Cached PID (/proc/{pid}/stat ~0.1ms) — if stored PID still exists, alive
|
|
* 3. Full pgrep scan (expensive, ~500ms) — only for agents that fail tiers 1+2
|
|
*/
|
|
private startLivenessChecker(): void {
|
|
this.cleanup.setInterval(
|
|
async () => {
|
|
// Guard: prevent concurrent liveness checks (avoids duplicate completed events)
|
|
if (this._isCheckingLiveness) return;
|
|
this._isCheckingLiveness = true;
|
|
|
|
try {
|
|
// Collect agents that need the expensive pgrep scan
|
|
const needsFullScan: SubagentInfo[] = [];
|
|
|
|
for (const [_agentId, info] of this.agentInfo) {
|
|
if (info.status !== 'active' && info.status !== 'idle') continue;
|
|
|
|
// Tier 1: File mtime check (~0.3ms per agent)
|
|
if (await this.checkSubagentFileAlive(info)) continue;
|
|
|
|
// Tier 2: Cached PID check (~0.1ms per agent)
|
|
if (info.pid && (await this.checkPidAlive(info.pid))) continue;
|
|
|
|
// Tiers 1+2 failed — need expensive scan for this agent
|
|
needsFullScan.push(info);
|
|
}
|
|
|
|
// Tier 3: Full pgrep scan — only if any agents failed cheap checks
|
|
if (needsFullScan.length > 0) {
|
|
const pidMap = await this.getClaudePids();
|
|
|
|
for (const info of needsFullScan) {
|
|
// Re-check status in case another check completed this agent
|
|
if (info.status !== 'active' && info.status !== 'idle') continue;
|
|
|
|
const alive = this.checkSubagentAliveFromPidMap(info, pidMap);
|
|
if (!alive) {
|
|
this.markSubagentAsCompleted(info);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Periodically clean up stale completed agents (older than 24 hours)
|
|
this.cleanupStaleAgents();
|
|
} finally {
|
|
this._isCheckingLiveness = false;
|
|
}
|
|
},
|
|
LIVENESS_CHECK_MS,
|
|
{ description: 'subagent liveness check' }
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Check if a PID is still alive via /proc/{pid}/stat (single file read, ~0.1ms).
|
|
*/
|
|
private async checkPidAlive(pid: number): Promise<boolean> {
|
|
try {
|
|
await statAsync(`/proc/${pid}`);
|
|
return true;
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Run pgrep once and read /proc info for all Claude PIDs in parallel.
|
|
* Returns a Map of pid -> { environ, cmdline } for subagent processes only.
|
|
* Excludes main Codeman-managed Claude processes (CODEMAN_MUX=1).
|
|
* Also updates cached PIDs on tracked agents when a match is found.
|
|
*/
|
|
private async getClaudePids(): Promise<Map<number, { environ: string; cmdline: string }>> {
|
|
const result = new Map<number, { environ: string; cmdline: string }>();
|
|
try {
|
|
const pgrepOutput = await new Promise<string>((resolve, reject) => {
|
|
execFile('pgrep', ['-f', 'claude'], { encoding: 'utf8' }, (err, stdout) => {
|
|
if (err) return reject(err);
|
|
resolve(stdout);
|
|
});
|
|
});
|
|
const pids = pgrepOutput
|
|
.trim()
|
|
.split('\n')
|
|
.filter(Boolean)
|
|
.map((s) => parseInt(s, 10))
|
|
.filter((n) => !Number.isNaN(n));
|
|
|
|
// Read /proc for all PIDs in parallel
|
|
await Promise.all(
|
|
pids.map(async (pid) => {
|
|
let environ = '';
|
|
let cmdline = '';
|
|
try {
|
|
environ = await readFile(`/proc/${pid}/environ`, 'utf8');
|
|
} catch {
|
|
/* skip */
|
|
}
|
|
try {
|
|
cmdline = await readFile(`/proc/${pid}/cmdline`, 'utf8');
|
|
} catch {
|
|
/* skip */
|
|
}
|
|
// Skip main Codeman-managed Claude processes — only track subagents
|
|
if (environ.includes('CODEMAN_MUX=1')) return;
|
|
if (environ || cmdline) {
|
|
result.set(pid, { environ, cmdline });
|
|
}
|
|
})
|
|
);
|
|
|
|
// Update cached PIDs on tracked agents
|
|
for (const [pid, procInfo] of result) {
|
|
for (const [_agentId, info] of this.agentInfo) {
|
|
if (info.status !== 'active' && info.status !== 'idle') continue;
|
|
if (procInfo.environ.includes(info.sessionId) || procInfo.cmdline.includes(info.sessionId)) {
|
|
info.pid = pid;
|
|
break; // Each PID belongs to at most one agent
|
|
}
|
|
}
|
|
}
|
|
} catch {
|
|
// pgrep returns non-zero if no matches
|
|
}
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Check if a subagent is alive using the pre-fetched pid map (no process spawning).
|
|
*/
|
|
private checkSubagentAliveFromPidMap(
|
|
info: SubagentInfo,
|
|
pidMap: Map<number, { environ: string; cmdline: string }>
|
|
): boolean {
|
|
for (const [_pid, procInfo] of pidMap) {
|
|
if (procInfo.environ.includes(info.sessionId) || procInfo.cmdline.includes(info.sessionId)) {
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Check if a subagent's transcript file was recently modified.
|
|
* Primary liveness signal — transcript files are written to continuously while agent is active.
|
|
*/
|
|
private async checkSubagentFileAlive(info: SubagentInfo): Promise<boolean> {
|
|
try {
|
|
const fileStat = await statAsync(info.filePath);
|
|
const mtime = fileStat.mtime.getTime();
|
|
const now = Date.now();
|
|
if (now - mtime < FILE_ALIVE_THRESHOLD_MS) {
|
|
return true;
|
|
}
|
|
} catch {
|
|
// File doesn't exist or can't be read
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Check if the watcher is currently running
|
|
*/
|
|
isRunning(): boolean {
|
|
return this._isRunning;
|
|
}
|
|
|
|
/**
|
|
* Stop watching and clean up all state
|
|
*/
|
|
stop(): void {
|
|
this._isRunning = false;
|
|
|
|
// Dispose poll and liveness intervals, then re-create for potential restart
|
|
this.cleanup.dispose();
|
|
this.cleanup = new CleanupManager();
|
|
|
|
// Clear file debouncers
|
|
this.fileDeb.dispose();
|
|
this.fileAgentContext.clear();
|
|
|
|
// Remove error handlers before closing watchers to prevent memory leak
|
|
for (const [dir, handler] of this.dirWatcherErrorHandlers) {
|
|
const watcher = this.dirWatchers.get(dir);
|
|
if (watcher) watcher.off('error', handler);
|
|
}
|
|
this.dirWatcherErrorHandlers.clear();
|
|
|
|
for (const watcher of this.dirWatchers.values()) {
|
|
watcher.close();
|
|
}
|
|
this.dirWatchers.clear();
|
|
|
|
this.idleDeb.dispose();
|
|
|
|
// Clear all state for clean restart
|
|
this.filePositions.clear();
|
|
this.agentInfo.clear();
|
|
this.knownSubagentDirs.clear();
|
|
this.pendingToolCalls.clear();
|
|
this.parentDescriptionCache.clear();
|
|
this._pollCount = 0;
|
|
}
|
|
|
|
/**
|
|
* Clean up stale agents to prevent unbounded memory growth.
|
|
* - Completed agents: removed after STALE_COMPLETED_MAX_AGE_MS (1 hour)
|
|
* - Idle agents: removed after STALE_IDLE_MAX_AGE_MS (4 hours)
|
|
* Also enforces MAX_TRACKED_AGENTS limit with LRU eviction.
|
|
* Also cleans up orphaned pending tool calls older than PENDING_TOOL_CALL_TTL_MS.
|
|
*/
|
|
private cleanupStaleAgents(): void {
|
|
const now = Date.now();
|
|
const agentsToDelete = new Set<string>();
|
|
|
|
for (const [agentId, info] of this.agentInfo) {
|
|
const age = now - info.lastActivityAt;
|
|
|
|
// Clean up based on status and age
|
|
if (info.status === 'completed' && age > STALE_COMPLETED_MAX_AGE_MS) {
|
|
agentsToDelete.add(agentId);
|
|
} else if (info.status === 'idle' && age > STALE_IDLE_MAX_AGE_MS) {
|
|
agentsToDelete.add(agentId);
|
|
}
|
|
}
|
|
|
|
// Enforce max tracked agents limit (LRU eviction)
|
|
const currentCount = this.agentInfo.size - agentsToDelete.size;
|
|
if (currentCount > MAX_TRACKED_AGENTS) {
|
|
// Sort by lastActivityAt (oldest first) and evict oldest completed/idle agents
|
|
const sortedAgents = Array.from(this.agentInfo.entries())
|
|
.filter(([id]) => !agentsToDelete.has(id))
|
|
.filter(([, info]) => info.status !== 'active') // Keep active agents
|
|
.sort((a, b) => a[1].lastActivityAt - b[1].lastActivityAt);
|
|
|
|
const toEvict = currentCount - MAX_TRACKED_AGENTS;
|
|
for (let i = 0; i < toEvict && i < sortedAgents.length; i++) {
|
|
agentsToDelete.add(sortedAgents[i][0]);
|
|
}
|
|
}
|
|
|
|
// Perform cleanup
|
|
for (const agentId of agentsToDelete) {
|
|
this.removeAgent(agentId);
|
|
}
|
|
|
|
// Clean up orphaned pending tool calls (older than TTL)
|
|
// These can accumulate if tool_result is never received (e.g., agent crashed)
|
|
for (const [agentId, agentCalls] of this.pendingToolCalls) {
|
|
const idsToDelete: string[] = [];
|
|
for (const [toolUseId, callInfo] of agentCalls) {
|
|
if (now - callInfo.timestamp > PENDING_TOOL_CALL_TTL_MS) {
|
|
idsToDelete.push(toolUseId);
|
|
}
|
|
}
|
|
for (const id of idsToDelete) {
|
|
agentCalls.delete(id);
|
|
}
|
|
// If agent has no more pending calls, remove the agent entry from the map
|
|
if (agentCalls.size === 0) {
|
|
this.pendingToolCalls.delete(agentId);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Remove an agent and all its associated resources.
|
|
*/
|
|
private removeAgent(agentId: string): void {
|
|
const info = this.agentInfo.get(agentId);
|
|
if (info) {
|
|
this.agentInfo.delete(agentId);
|
|
this.pendingToolCalls.delete(agentId);
|
|
this.filePositions.delete(info.filePath);
|
|
this.fileAgentContext.delete(info.filePath);
|
|
this.fileDeb.cancelKey(info.filePath);
|
|
this.idleDeb.cancelKey(agentId);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Manually trigger cleanup of stale agents.
|
|
* Returns number of agents removed.
|
|
*/
|
|
cleanupNow(): number {
|
|
const beforeCount = this.agentInfo.size;
|
|
this.cleanupStaleAgents();
|
|
return beforeCount - this.agentInfo.size;
|
|
}
|
|
|
|
/**
|
|
* Clear all tracked agents (for manual reset).
|
|
* Returns number of agents cleared.
|
|
*/
|
|
clearAll(): number {
|
|
const count = this.agentInfo.size;
|
|
const agentIds = Array.from(this.agentInfo.keys());
|
|
for (const agentId of agentIds) {
|
|
this.removeAgent(agentId);
|
|
}
|
|
return count;
|
|
}
|
|
|
|
/**
|
|
* Get all known subagents
|
|
*/
|
|
getSubagents(): SubagentInfo[] {
|
|
return Array.from(this.agentInfo.values());
|
|
}
|
|
|
|
/**
|
|
* Get subagents for a specific Codeman session
|
|
* Maps Codeman working directory to Claude's project hash
|
|
*/
|
|
getSubagentsForSession(workingDir: string): SubagentInfo[] {
|
|
const projectHash = this.getProjectHash(workingDir);
|
|
return Array.from(this.agentInfo.values()).filter((info) => info.projectHash === projectHash);
|
|
}
|
|
|
|
/**
|
|
* Get a specific subagent's info
|
|
*/
|
|
getSubagent(agentId: string): SubagentInfo | undefined {
|
|
return this.agentInfo.get(agentId);
|
|
}
|
|
|
|
/**
|
|
* Update a subagent's description.
|
|
* Used to set the short description from TaskTracker when available.
|
|
*/
|
|
updateDescription(agentId: string, description: string): boolean {
|
|
const info = this.agentInfo.get(agentId);
|
|
if (info) {
|
|
info.description = description;
|
|
this.emit('subagent:updated', info);
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/**
|
|
* Find sessions whose working directory matches a project hash.
|
|
* Returns the project hash for a given working directory.
|
|
*/
|
|
getProjectHashForDir(workingDir: string): string {
|
|
return this.getProjectHash(workingDir);
|
|
}
|
|
|
|
/**
|
|
* Get internal statistics for memory monitoring.
|
|
* Returns counts of internal Maps and resources.
|
|
*/
|
|
getStats(): {
|
|
agentCount: number;
|
|
fileDebouncerCount: number;
|
|
dirWatcherCount: number;
|
|
idleTimerCount: number;
|
|
pendingToolCallsCount: number;
|
|
knownDirsCount: number;
|
|
filePositionsCount: number;
|
|
} {
|
|
// Count pending tool calls across all agents
|
|
let pendingToolCallsCount = 0;
|
|
for (const agentCalls of this.pendingToolCalls.values()) {
|
|
pendingToolCallsCount += agentCalls.size;
|
|
}
|
|
|
|
return {
|
|
agentCount: this.agentInfo.size,
|
|
fileDebouncerCount: this.fileDeb.size,
|
|
dirWatcherCount: this.dirWatchers.size,
|
|
idleTimerCount: this.idleDeb.size,
|
|
pendingToolCallsCount,
|
|
knownDirsCount: this.knownSubagentDirs.size,
|
|
filePositionsCount: this.filePositions.size,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get recent subagents (modified within specified minutes)
|
|
*/
|
|
getRecentSubagents(minutes: number = 60): SubagentInfo[] {
|
|
const cutoff = Date.now() - minutes * 60 * 1000;
|
|
return Array.from(this.agentInfo.values())
|
|
.filter((info) => info.lastActivityAt > cutoff)
|
|
.sort((a, b) => b.lastActivityAt - a.lastActivityAt);
|
|
}
|
|
|
|
/**
|
|
* Kill a subagent by its agent ID
|
|
* Uses cached PID first, falls back to findSubagentProcess if needed.
|
|
*/
|
|
async killSubagent(agentId: string): Promise<boolean> {
|
|
const info = this.agentInfo.get(agentId);
|
|
if (!info) return false;
|
|
|
|
// Already completed, nothing to kill
|
|
if (info.status === 'completed') return false;
|
|
|
|
try {
|
|
// Always use findSubagentProcess for kill — it verifies environ/cmdline,
|
|
// preventing PID reuse attacks (cached PID may have been recycled by OS)
|
|
const pid = await this.findSubagentProcess(info.sessionId);
|
|
if (pid) {
|
|
process.kill(pid, 'SIGTERM');
|
|
this.markSubagentAsCompleted(info);
|
|
return true;
|
|
}
|
|
} catch {
|
|
// Process may have already exited
|
|
}
|
|
|
|
// Mark as completed even if we couldn't find the process
|
|
this.markSubagentAsCompleted(info);
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Kill all subagents for a specific Codeman session.
|
|
* IMPORTANT: Must scope to sessionId to avoid cross-session kills.
|
|
* All sessions in the same workingDir share a projectHash, so filtering
|
|
* by workingDir alone would kill subagents belonging to OTHER sessions.
|
|
*/
|
|
async killSubagentsForSession(workingDir: string, sessionId?: string): Promise<void> {
|
|
const subagents = this.getSubagentsForSession(workingDir);
|
|
for (const agent of subagents) {
|
|
if (agent.status === 'active' || agent.status === 'idle') {
|
|
// Only kill subagents belonging to this specific session
|
|
if (sessionId && agent.sessionId !== sessionId) continue;
|
|
await this.killSubagent(agent.agentId);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Find the process ID of a Claude subagent by its session ID.
|
|
* Searches /proc for claude processes with matching session ID in environment.
|
|
* Skips main Codeman-managed Claude processes (identified by CODEMAN_MUX=1).
|
|
* Caches the discovered PID on the matching agent info for future fast checks.
|
|
*/
|
|
private async findSubagentProcess(sessionId: string): Promise<number | null> {
|
|
try {
|
|
// Find all claude processes (async to avoid blocking event loop)
|
|
const pgrepOutput = await new Promise<string>((resolve, reject) => {
|
|
execFile('pgrep', ['-f', 'claude'], { encoding: 'utf8' }, (err, stdout) => {
|
|
if (err) return reject(err);
|
|
resolve(stdout);
|
|
});
|
|
});
|
|
const pids = pgrepOutput.trim().split('\n').filter(Boolean);
|
|
|
|
for (const pidStr of pids) {
|
|
const pid = parseInt(pidStr, 10);
|
|
if (Number.isNaN(pid)) continue;
|
|
|
|
let environ = '';
|
|
try {
|
|
environ = await readFile(`/proc/${pid}/environ`, 'utf8');
|
|
} catch {
|
|
// Can't read this process's environ - skip
|
|
}
|
|
|
|
// Defense-in-depth: never kill a main Codeman-managed Claude process.
|
|
// Main processes have CODEMAN_MUX=1 in their environment; subagents don't.
|
|
if (environ.includes('CODEMAN_MUX=1')) continue;
|
|
|
|
if (environ.includes(sessionId)) {
|
|
// Cache PID on the matching agent
|
|
this.cacheAgentPid(sessionId, pid);
|
|
return pid;
|
|
}
|
|
|
|
try {
|
|
const cmdline = await readFile(`/proc/${pid}/cmdline`, 'utf8');
|
|
if (cmdline.includes(sessionId)) {
|
|
this.cacheAgentPid(sessionId, pid);
|
|
return pid;
|
|
}
|
|
} catch {
|
|
// Can't read this process's cmdline - skip
|
|
}
|
|
}
|
|
} catch {
|
|
// pgrep returns non-zero if no matches
|
|
}
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Store a discovered PID on the agent info with matching sessionId.
|
|
*/
|
|
private cacheAgentPid(sessionId: string, pid: number): void {
|
|
for (const [_agentId, info] of this.agentInfo) {
|
|
if (info.sessionId === sessionId && (info.status === 'active' || info.status === 'idle')) {
|
|
info.pid = pid;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get transcript for a subagent (optionally limited to last N entries)
|
|
*/
|
|
async getTranscript(agentId: string, limit?: number): Promise<SubagentTranscriptEntry[]> {
|
|
const info = this.agentInfo.get(agentId);
|
|
if (!info) return [];
|
|
|
|
const entries: SubagentTranscriptEntry[] = [];
|
|
|
|
try {
|
|
const content = await readFile(info.filePath, 'utf8');
|
|
const lines = content.split('\n').filter((l) => l.trim());
|
|
|
|
for (const line of lines) {
|
|
try {
|
|
const entry = JSON.parse(line) as SubagentTranscriptEntry;
|
|
entries.push(entry);
|
|
} catch {
|
|
// Skip malformed lines
|
|
}
|
|
}
|
|
} catch {
|
|
// File read error
|
|
}
|
|
|
|
if (limit && limit > 0) {
|
|
return entries.slice(-limit);
|
|
}
|
|
|
|
return entries;
|
|
}
|
|
|
|
/**
|
|
* Format transcript entries for display
|
|
*/
|
|
formatTranscript(entries: SubagentTranscriptEntry[]): string[] {
|
|
const lines: string[] = [];
|
|
|
|
for (const entry of entries) {
|
|
if (entry.type === 'progress' && entry.data) {
|
|
lines.push(this.formatProgress(entry));
|
|
} else if (entry.type === 'assistant' && entry.message?.content) {
|
|
// Handle both string and array content formats
|
|
if (typeof entry.message.content === 'string') {
|
|
const text = entry.message.content.trim();
|
|
if (text.length > 0) {
|
|
const preview = text.length > TEXT_PREVIEW_LENGTH ? text.substring(0, TEXT_PREVIEW_LENGTH) + '...' : text;
|
|
lines.push(`${this.formatTime(entry.timestamp)} 💬 ${preview.replace(/\n/g, ' ')}`);
|
|
}
|
|
} else {
|
|
for (const content of entry.message.content) {
|
|
if (content.type === 'tool_use' && content.name) {
|
|
lines.push(this.formatToolCall(entry.timestamp, content.name, content.input || {}));
|
|
} else if (content.type === 'text' && content.text) {
|
|
const text = content.text.trim();
|
|
if (text.length > 0) {
|
|
const preview =
|
|
text.length > TEXT_PREVIEW_LENGTH ? text.substring(0, TEXT_PREVIEW_LENGTH) + '...' : text;
|
|
lines.push(`${this.formatTime(entry.timestamp)} 💬 ${preview.replace(/\n/g, ' ')}`);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} else if (entry.type === 'user' && entry.message?.content) {
|
|
// Handle both string and array content formats
|
|
const text = this.extractFirstTextContent(entry.message.content);
|
|
if (text && text.length < 100 && !text.includes('{')) {
|
|
lines.push(`${this.formatTime(entry.timestamp)} 📥 User: ${text.substring(0, USER_TEXT_PREVIEW_LENGTH)}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
return lines;
|
|
}
|
|
|
|
// ========== Private Methods ==========
|
|
|
|
/**
|
|
* Convert working directory to Claude's project hash format
|
|
*/
|
|
private getProjectHash(workingDir: string): string {
|
|
return workingDir.replace(/\//g, '-');
|
|
}
|
|
|
|
/**
|
|
* Extract a smart, concise title from a task prompt
|
|
* Aims for ~40-50 chars that convey what the agent is doing
|
|
*/
|
|
private extractSmartTitle(text: string): string {
|
|
const MAX_LEN = SMART_TITLE_MAX_LENGTH;
|
|
|
|
// Get first line/sentence
|
|
const title = text.split('\n')[0].trim();
|
|
|
|
// If already short enough, use it
|
|
if (title.length <= MAX_LEN) {
|
|
return title.replace(/[.!?,\s]+$/, '');
|
|
}
|
|
|
|
// Remove common filler phrases to condense
|
|
const fillers = [
|
|
/^(please |i need you to |i want you to |can you |could you )/i,
|
|
/ (the|a|an) /gi,
|
|
/ (in|at|on|to|for|of|with|from|by) the /gi,
|
|
/ (including|related to|regarding|about) /gi,
|
|
/[""]/g,
|
|
/ +/g, // multiple spaces to single
|
|
];
|
|
|
|
let condensed = title;
|
|
for (const filler of fillers) {
|
|
condensed = condensed.replace(filler, (match) => {
|
|
// Keep single space for word boundaries
|
|
if (match.trim() === '') return ' ';
|
|
if (/^(the|a|an)$/i.test(match.trim())) return ' ';
|
|
if (/including|related to|regarding|about/i.test(match)) return ': ';
|
|
return ' ';
|
|
});
|
|
}
|
|
condensed = condensed.replace(/ +/g, ' ').trim();
|
|
|
|
// If condensed version is short enough, use it
|
|
if (condensed.length <= MAX_LEN) {
|
|
return condensed.replace(/[.!?,:\s]+$/, '');
|
|
}
|
|
|
|
// Try to cut at a natural boundary (colon, dash, comma)
|
|
const boundaryMatch = condensed.substring(0, MAX_LEN + 5).match(/^(.{20,}?)[:\-,]/);
|
|
if (boundaryMatch && boundaryMatch[1].length <= MAX_LEN) {
|
|
return boundaryMatch[1].trim();
|
|
}
|
|
|
|
// Last resort: truncate at word boundary
|
|
const truncated = condensed.substring(0, MAX_LEN);
|
|
const lastSpace = truncated.lastIndexOf(' ');
|
|
if (lastSpace > 20) {
|
|
return truncated.substring(0, lastSpace).replace(/[.!?,:\s]+$/, '');
|
|
}
|
|
|
|
return truncated.replace(/[.!?,:\s]+$/, '');
|
|
}
|
|
|
|
private async _resolveDescription(
|
|
projectHash: string,
|
|
sessionId: string,
|
|
agentId: string,
|
|
filePath: string,
|
|
fallbackText?: string
|
|
): Promise<string | undefined> {
|
|
// First try parent transcript (most reliable)
|
|
const fromParent = await this.extractDescriptionFromParentTranscript(projectHash, sessionId, agentId);
|
|
if (fromParent) return fromParent;
|
|
|
|
// Fallback: inline text (from processEntry) or file extraction
|
|
if (fallbackText) {
|
|
return this.extractSmartTitle(fallbackText);
|
|
}
|
|
return this.extractDescriptionFromFile(filePath);
|
|
}
|
|
|
|
/**
|
|
* Extract the short description from the parent session's transcript.
|
|
* This is the most reliable method because it reads the actual Task tool result
|
|
* that spawned this agent, which contains the description directly.
|
|
*
|
|
* The parent transcript contains a 'user' entry with toolUseResult that has:
|
|
* - agentId: the spawned agent's ID
|
|
* - description: the Task description parameter
|
|
*
|
|
* We look for this format:
|
|
* { "type": "user", "toolUseResult": { "agentId": "xxx", "description": "..." } }
|
|
*/
|
|
private async extractDescriptionFromParentTranscript(
|
|
projectHash: string,
|
|
sessionId: string,
|
|
agentId: string
|
|
): Promise<string | undefined> {
|
|
const cacheKey = `${projectHash}/${sessionId}`;
|
|
const CACHE_TTL_MS = 5000;
|
|
|
|
// Check cache first (covers burst of simultaneous agent discoveries)
|
|
const cached = this.parentDescriptionCache.get(cacheKey);
|
|
if (cached && Date.now() - cached.timestamp < CACHE_TTL_MS) {
|
|
return cached.descriptions.get(agentId);
|
|
}
|
|
|
|
try {
|
|
// The parent session's transcript is at: ~/.claude/projects/{projectHash}/{sessionId}.jsonl
|
|
const transcriptPath = join(CLAUDE_PROJECTS_DIR, projectHash, `${sessionId}.jsonl`);
|
|
let fileSize: number;
|
|
try {
|
|
const fileStat = await statAsync(transcriptPath);
|
|
fileSize = fileStat.size;
|
|
} catch {
|
|
return undefined;
|
|
}
|
|
|
|
// Only read last 16KB — toolUseResult entries are near the end of the transcript
|
|
const TAIL_BYTES = 16384;
|
|
const startOffset = Math.max(0, fileSize - TAIL_BYTES);
|
|
const content = await new Promise<string>((resolve, reject) => {
|
|
const chunks: string[] = [];
|
|
const stream = createReadStream(transcriptPath, { start: startOffset, encoding: 'utf8' });
|
|
stream.on('data', (chunk) => chunks.push(String(chunk)));
|
|
stream.on('end', () => resolve(chunks.join('')));
|
|
stream.on('error', reject);
|
|
});
|
|
let lines = content.split('\n').filter((l) => l.trim());
|
|
// If we started mid-file, drop the first partial line
|
|
if (startOffset > 0 && lines.length > 0) {
|
|
lines = lines.slice(1);
|
|
}
|
|
|
|
// Parse ALL toolUseResult entries into a Map and cache them
|
|
const descriptions = new Map<string, string>();
|
|
for (const line of lines) {
|
|
try {
|
|
const entry = JSON.parse(line);
|
|
if (entry.type === 'user' && entry.toolUseResult?.agentId && entry.toolUseResult?.description) {
|
|
descriptions.set(entry.toolUseResult.agentId, entry.toolUseResult.description);
|
|
}
|
|
} catch {
|
|
// Skip malformed lines
|
|
}
|
|
}
|
|
|
|
this.parentDescriptionCache.set(cacheKey, { descriptions, timestamp: Date.now() });
|
|
return descriptions.get(agentId);
|
|
} catch {
|
|
// Failed to read transcript
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
/**
|
|
* Extract description from agent file by finding first user message
|
|
*/
|
|
private async extractDescriptionFromFile(filePath: string): Promise<string | undefined> {
|
|
try {
|
|
// Only read the first 8KB — more than enough for 5 JSONL lines
|
|
const stream = createReadStream(filePath, { end: FILE_PEEK_BYTES });
|
|
const rl = createInterface({ input: stream });
|
|
|
|
return await new Promise<string | undefined>((resolve) => {
|
|
let lineCount = 0;
|
|
let resolved = false;
|
|
rl.on('line', (line) => {
|
|
if (resolved || lineCount >= 5) {
|
|
rl.close();
|
|
stream.destroy();
|
|
return;
|
|
}
|
|
lineCount++;
|
|
if (!line.trim()) return;
|
|
try {
|
|
const entry = JSON.parse(line);
|
|
if (entry.type === 'user' && entry.message?.content) {
|
|
const text = this.extractFirstTextContent(entry.message.content);
|
|
if (text) {
|
|
resolved = true;
|
|
rl.close();
|
|
stream.destroy();
|
|
resolve(this.extractSmartTitle(text));
|
|
}
|
|
}
|
|
} catch {
|
|
// Skip malformed lines
|
|
}
|
|
});
|
|
rl.on('close', () => {
|
|
if (!resolved) resolve(undefined);
|
|
});
|
|
rl.on('error', () => {
|
|
if (!resolved) resolve(undefined);
|
|
});
|
|
});
|
|
} catch {
|
|
// Failed to read file
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
/**
|
|
* Scan for all subagent directories (async to avoid blocking event loop)
|
|
*/
|
|
private async scanForSubagents(): Promise<void> {
|
|
try {
|
|
await statAsync(CLAUDE_PROJECTS_DIR);
|
|
} catch {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
const projects = await readdir(CLAUDE_PROJECTS_DIR);
|
|
|
|
for (const project of projects) {
|
|
const projectPath = join(CLAUDE_PROJECTS_DIR, project);
|
|
|
|
try {
|
|
const st = await statAsync(projectPath);
|
|
if (!st.isDirectory()) continue;
|
|
|
|
const sessions = await readdir(projectPath);
|
|
|
|
for (const session of sessions) {
|
|
const sessionPath = join(projectPath, session);
|
|
|
|
try {
|
|
const sessionStat = await statAsync(sessionPath);
|
|
if (!sessionStat.isDirectory()) continue;
|
|
|
|
const subagentDir = join(sessionPath, 'subagents');
|
|
try {
|
|
await statAsync(subagentDir);
|
|
await this.watchSubagentDir(subagentDir, project, session);
|
|
} catch {
|
|
// subagent dir doesn't exist - skip
|
|
}
|
|
} catch {
|
|
// Skip inaccessible session directories
|
|
}
|
|
}
|
|
} catch {
|
|
// Skip inaccessible project directories
|
|
}
|
|
}
|
|
} catch (error) {
|
|
this.emit('subagent:error', error as Error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Watch a subagent directory for new and changed files.
|
|
* Uses a single directory-level fs.watch() instead of per-file watchers.
|
|
* On Linux, inotify IN_MODIFY fires for content changes within the directory.
|
|
*/
|
|
private async watchSubagentDir(dir: string, projectHash: string, sessionId: string): Promise<void> {
|
|
if (this.knownSubagentDirs.has(dir)) return;
|
|
this.knownSubagentDirs.add(dir);
|
|
|
|
// Register existing files (initial scan - skip old files)
|
|
try {
|
|
const files = await readdir(dir);
|
|
for (const file of files) {
|
|
if (file.endsWith('.jsonl')) {
|
|
await this.registerAgentFile(join(dir, file), projectHash, sessionId, true);
|
|
}
|
|
}
|
|
} catch {
|
|
return;
|
|
}
|
|
|
|
// Single directory watcher handles both new files and file content changes
|
|
try {
|
|
const watcher = watch(dir, (_eventType, filename) => {
|
|
if (!filename?.endsWith('.jsonl')) return;
|
|
const filePath = join(dir, filename);
|
|
|
|
// Debounce 100ms to batch rapid writes
|
|
this.fileDeb.schedule(filePath, () => {
|
|
if (!existsSync(filePath)) return;
|
|
|
|
if (this.fileAgentContext.has(filePath)) {
|
|
// Known file — handle content change
|
|
this.handleFileChange(filePath).catch(() => {}); // Ignore - errors logged internally, don't crash watcher callback
|
|
} else {
|
|
// New file — register it
|
|
this.registerAgentFile(filePath, projectHash, sessionId).catch(() => {}); // Ignore - errors logged internally, don't crash watcher callback
|
|
}
|
|
});
|
|
});
|
|
|
|
// Handle watcher errors to prevent unhandled exceptions
|
|
// Store handler reference for proper cleanup
|
|
const errorHandler = (error: Error) => {
|
|
this.emit('subagent:error', error instanceof Error ? error : new Error(String(error)));
|
|
watcher.close();
|
|
this.dirWatcherErrorHandlers.delete(dir);
|
|
this.dirWatchers.delete(dir);
|
|
this.knownSubagentDirs.delete(dir);
|
|
};
|
|
watcher.on('error', errorHandler);
|
|
this.dirWatcherErrorHandlers.set(dir, errorHandler);
|
|
|
|
this.dirWatchers.set(dir, watcher);
|
|
} catch {
|
|
// Watch failed
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Handle a file content change for an already-registered agent file.
|
|
* Tails from last known position, updates info, retries description if missing.
|
|
*/
|
|
private async handleFileChange(filePath: string): Promise<void> {
|
|
const context = this.fileAgentContext.get(filePath);
|
|
if (!context) return;
|
|
|
|
const agentId = basename(filePath).replace('agent-', '').replace('.jsonl', '');
|
|
const currentPos = this.filePositions.get(filePath) || 0;
|
|
const newPos = await this.tailFile(filePath, agentId, context.sessionId, currentPos);
|
|
this.filePositions.set(filePath, newPos);
|
|
|
|
// Update info
|
|
const existingInfo = this.agentInfo.get(agentId);
|
|
if (existingInfo) {
|
|
try {
|
|
const newStat = await statAsync(filePath);
|
|
existingInfo.lastActivityAt = Date.now();
|
|
existingInfo.fileSize = newStat.size;
|
|
existingInfo.status = 'active';
|
|
} catch {
|
|
// Stat failed
|
|
}
|
|
|
|
// Retry description extraction if missing (race condition fix)
|
|
if (!existingInfo.description) {
|
|
const extractedDescription = await this._resolveDescription(
|
|
existingInfo.projectHash,
|
|
existingInfo.sessionId,
|
|
agentId,
|
|
filePath
|
|
);
|
|
if (extractedDescription) {
|
|
if (this.isInternalAgent(extractedDescription)) {
|
|
this.removeAgent(agentId);
|
|
return;
|
|
}
|
|
existingInfo.description = extractedDescription;
|
|
this.emit('subagent:updated', existingInfo);
|
|
}
|
|
}
|
|
|
|
// Reset idle timer
|
|
this.resetIdleTimer(agentId);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Register a specific agent transcript file (discovery + initial read).
|
|
* Does NOT create a per-file watcher — the directory watcher handles changes.
|
|
* @param filePath Path to the agent transcript file
|
|
* @param projectHash Claude project hash
|
|
* @param sessionId Claude session ID
|
|
* @param isInitialScan If true, skip files older than STARTUP_MAX_FILE_AGE_MS
|
|
*/
|
|
private async registerAgentFile(
|
|
filePath: string,
|
|
projectHash: string,
|
|
sessionId: string,
|
|
isInitialScan: boolean = false
|
|
): Promise<void> {
|
|
if (this.fileAgentContext.has(filePath)) return;
|
|
|
|
const agentId = basename(filePath).replace('agent-', '').replace('.jsonl', '');
|
|
|
|
// Initial info - handle race condition where file may be deleted between discovery and stat
|
|
let fileStat;
|
|
try {
|
|
fileStat = await statAsync(filePath);
|
|
} catch {
|
|
// File was deleted between discovery and stat - skip this agent
|
|
return;
|
|
}
|
|
|
|
// On initial scan, skip old files to avoid loading stale historical data
|
|
if (isInitialScan) {
|
|
const fileAge = Date.now() - fileStat.mtime.getTime();
|
|
if (fileAge > STARTUP_MAX_FILE_AGE_MS) {
|
|
return; // Skip old files on startup
|
|
}
|
|
}
|
|
|
|
// Extract description - prefer reading from parent transcript (most reliable)
|
|
const description = await this._resolveDescription(projectHash, sessionId, agentId, filePath);
|
|
|
|
// Skip internal Claude Code agents (e.g., suggestion mode) - not real subagents
|
|
if (this.isInternalAgent(description)) {
|
|
return;
|
|
}
|
|
|
|
const info: SubagentInfo = {
|
|
agentId,
|
|
sessionId,
|
|
projectHash,
|
|
filePath,
|
|
startedAt: fileStat.birthtime.toISOString(),
|
|
lastActivityAt: fileStat.mtime.getTime(),
|
|
status: 'active',
|
|
toolCallCount: 0,
|
|
entryCount: 0,
|
|
fileSize: fileStat.size,
|
|
description,
|
|
};
|
|
|
|
// Enforce MAX_TRACKED_AGENTS during insertion — evict oldest inactive agent
|
|
if (this.agentInfo.size >= MAX_TRACKED_AGENTS) {
|
|
const oldestId = this.findOldestInactiveAgent();
|
|
if (oldestId) {
|
|
this.removeAgent(oldestId);
|
|
}
|
|
}
|
|
|
|
// Track file context for directory watcher change handling
|
|
this.fileAgentContext.set(filePath, { projectHash, sessionId });
|
|
this.agentInfo.set(agentId, info);
|
|
this.emit('subagent:discovered', info);
|
|
|
|
// Read existing content
|
|
this.tailFile(filePath, agentId, sessionId, 0)
|
|
.then((position) => {
|
|
this.filePositions.set(filePath, position);
|
|
})
|
|
.catch((err) => {
|
|
// Log but don't throw - non-critical background operation
|
|
console.warn(`[SubagentWatcher] Failed to read initial content for ${agentId}:`, err);
|
|
});
|
|
|
|
this.resetIdleTimer(agentId);
|
|
}
|
|
|
|
/**
|
|
* Tail a file from a specific position
|
|
*/
|
|
private async tailFile(filePath: string, agentId: string, sessionId: string, fromPosition: number): Promise<number> {
|
|
return new Promise((resolve) => {
|
|
let position = fromPosition;
|
|
|
|
const stream = createReadStream(filePath, { start: fromPosition });
|
|
const rl = createInterface({ input: stream });
|
|
|
|
rl.on('line', (line) => {
|
|
const lineBytes = Buffer.byteLength(line, 'utf8') + 1;
|
|
position += lineBytes; // Always advance past the line
|
|
|
|
try {
|
|
const entry = JSON.parse(line) as SubagentTranscriptEntry;
|
|
this.processEntry(entry, agentId, sessionId).catch(() => {
|
|
// processEntry failure is non-critical
|
|
});
|
|
|
|
// Update entry count
|
|
const info = this.agentInfo.get(agentId);
|
|
if (info) {
|
|
info.entryCount++;
|
|
}
|
|
} catch {
|
|
// Malformed JSON line — skip it
|
|
}
|
|
});
|
|
|
|
rl.on('close', () => {
|
|
resolve(position);
|
|
});
|
|
|
|
rl.on('error', () => {
|
|
resolve(position);
|
|
});
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Process a transcript entry and emit appropriate events
|
|
*/
|
|
private async processEntry(entry: SubagentTranscriptEntry, agentId: string, sessionId: string): Promise<void> {
|
|
const info = this.agentInfo.get(agentId);
|
|
|
|
if (info) {
|
|
this._processModelInfo(entry, info);
|
|
this._processTokenInfo(entry, info);
|
|
|
|
if (await this._processDescription(entry, agentId, info)) return;
|
|
}
|
|
|
|
if (entry.type === 'progress' && entry.data) {
|
|
const progress: SubagentProgress = {
|
|
agentId,
|
|
sessionId,
|
|
timestamp: entry.timestamp,
|
|
progressType: entry.data.type,
|
|
query: entry.data.query,
|
|
resultCount: entry.data.resultCount,
|
|
hookEvent: entry.data.hookEvent,
|
|
hookName:
|
|
entry.data.hookName ||
|
|
(entry.data.hookEvent && entry.data.tool_name
|
|
? `${entry.data.hookEvent}:${entry.data.tool_name}`
|
|
: undefined),
|
|
};
|
|
this.emit('subagent:progress', progress);
|
|
} else if (entry.type === 'assistant' && entry.message?.content) {
|
|
this._processAssistantContent(entry, agentId, sessionId);
|
|
} else if (entry.type === 'user' && entry.message?.content) {
|
|
this._processUserContent(entry, agentId, sessionId);
|
|
}
|
|
}
|
|
|
|
private _processModelInfo(entry: SubagentTranscriptEntry, agent: SubagentInfo): void {
|
|
if (entry.type === 'assistant' && entry.message?.model && !agent.model) {
|
|
agent.model = entry.message.model;
|
|
agent.modelShort = this.extractModelShort(entry.message.model);
|
|
this.emit('subagent:updated', agent);
|
|
}
|
|
}
|
|
|
|
private _processTokenInfo(entry: SubagentTranscriptEntry, agent: SubagentInfo): void {
|
|
if (!entry.message?.usage) return;
|
|
if (entry.message.usage.input_tokens) {
|
|
agent.totalInputTokens = (agent.totalInputTokens || 0) + entry.message.usage.input_tokens;
|
|
}
|
|
if (entry.message.usage.output_tokens) {
|
|
agent.totalOutputTokens = (agent.totalOutputTokens || 0) + entry.message.usage.output_tokens;
|
|
}
|
|
}
|
|
|
|
private async _processDescription(
|
|
entry: SubagentTranscriptEntry,
|
|
agentId: string,
|
|
agent: SubagentInfo
|
|
): Promise<boolean> {
|
|
if (agent.description || entry.type !== 'user' || !entry.message?.content) return false;
|
|
|
|
const fallbackText = this.extractFirstTextContent(entry.message.content);
|
|
const description = await this._resolveDescription(
|
|
agent.projectHash,
|
|
agent.sessionId,
|
|
agentId,
|
|
agent.filePath,
|
|
fallbackText
|
|
);
|
|
if (description) {
|
|
if (this.isInternalAgent(description)) {
|
|
this.removeAgent(agentId);
|
|
return true;
|
|
}
|
|
agent.description = description;
|
|
this.emit('subagent:updated', agent);
|
|
}
|
|
return false;
|
|
}
|
|
|
|
private _processAssistantContent(entry: SubagentTranscriptEntry, agentId: string, sessionId: string): void {
|
|
const messageContent = entry.message!.content;
|
|
if (typeof messageContent === 'string') {
|
|
const text = messageContent.trim();
|
|
if (text.length > 0) {
|
|
const message: SubagentMessage = {
|
|
agentId,
|
|
sessionId,
|
|
timestamp: entry.timestamp,
|
|
role: 'assistant',
|
|
text: text.substring(0, MESSAGE_TEXT_LIMIT),
|
|
};
|
|
this.emit('subagent:message', message);
|
|
}
|
|
} else {
|
|
for (const content of messageContent) {
|
|
if (content.type === 'tool_use' && content.name) {
|
|
// Store toolUseId for linking to results, with timestamp for TTL cleanup
|
|
if (content.id) {
|
|
if (!this.pendingToolCalls.has(agentId)) {
|
|
this.pendingToolCalls.set(agentId, new Map());
|
|
}
|
|
const agentCalls = this.pendingToolCalls.get(agentId)!;
|
|
// Enforce size limit to prevent memory leak from rapid tool calls
|
|
if (agentCalls.size >= MAX_PENDING_TOOL_CALLS) {
|
|
// FIFO eviction: delete first (oldest) entry using Map insertion order
|
|
const firstKey = agentCalls.keys().next().value;
|
|
if (firstKey !== undefined) agentCalls.delete(firstKey);
|
|
}
|
|
agentCalls.set(content.id, {
|
|
toolName: content.name,
|
|
timestamp: Date.now(),
|
|
});
|
|
}
|
|
|
|
const toolCall: SubagentToolCall = {
|
|
agentId,
|
|
sessionId,
|
|
timestamp: entry.timestamp,
|
|
tool: content.name,
|
|
input: this.getTruncatedInput(content.name, content.input || {}),
|
|
toolUseId: content.id,
|
|
fullInput: content.input || {},
|
|
};
|
|
this.emit('subagent:tool_call', toolCall);
|
|
|
|
// Update tool call count
|
|
const agentInfo = this.agentInfo.get(agentId);
|
|
if (agentInfo) {
|
|
agentInfo.toolCallCount++;
|
|
}
|
|
} else if (content.type === 'tool_result' && content.tool_use_id) {
|
|
this.emitToolResult(
|
|
{ tool_use_id: content.tool_use_id, content: content.content, is_error: content.is_error },
|
|
agentId,
|
|
sessionId,
|
|
entry.timestamp
|
|
);
|
|
} else if (content.type === 'text' && content.text) {
|
|
const text = content.text.trim();
|
|
if (text.length > 0) {
|
|
const message: SubagentMessage = {
|
|
agentId,
|
|
sessionId,
|
|
timestamp: entry.timestamp,
|
|
role: 'assistant',
|
|
text: text.substring(0, MESSAGE_TEXT_LIMIT),
|
|
};
|
|
this.emit('subagent:message', message);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private _processUserContent(entry: SubagentTranscriptEntry, agentId: string, sessionId: string): void {
|
|
const messageContent = entry.message!.content;
|
|
if (typeof messageContent === 'string') {
|
|
const userText = messageContent.trim();
|
|
if (userText.length > 0 && userText.length < 500) {
|
|
const message: SubagentMessage = {
|
|
agentId,
|
|
sessionId,
|
|
timestamp: entry.timestamp,
|
|
role: 'user',
|
|
text: userText,
|
|
};
|
|
this.emit('subagent:message', message);
|
|
}
|
|
} else {
|
|
// Check for tool_result blocks in user messages (common pattern)
|
|
for (const content of messageContent) {
|
|
if (content.type === 'tool_result' && content.tool_use_id) {
|
|
this.emitToolResult(
|
|
{ tool_use_id: content.tool_use_id, content: content.content, is_error: content.is_error },
|
|
agentId,
|
|
sessionId,
|
|
entry.timestamp
|
|
);
|
|
} else if (content.type === 'text' && content.text) {
|
|
const userText = content.text.trim();
|
|
if (userText.length > 0 && userText.length < 500) {
|
|
const message: SubagentMessage = {
|
|
agentId,
|
|
sessionId,
|
|
timestamp: entry.timestamp,
|
|
role: 'user',
|
|
text: userText,
|
|
};
|
|
this.emit('subagent:message', message);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Extract text content from tool_result content field
|
|
*/
|
|
private extractToolResultContent(content: string | Array<{ type: string; text?: string }> | undefined): string {
|
|
if (!content) return '';
|
|
if (typeof content === 'string') return content;
|
|
if (Array.isArray(content)) {
|
|
return content
|
|
.filter((c) => c.type === 'text' && c.text)
|
|
.map((c) => c.text)
|
|
.join('\n');
|
|
}
|
|
return '';
|
|
}
|
|
|
|
/**
|
|
* Get truncated input for display (keeps primary param, truncates large content)
|
|
*/
|
|
private getTruncatedInput(_tool: string, input: Record<string, unknown>): Record<string, unknown> {
|
|
const truncated: Record<string, unknown> = {};
|
|
for (const [key, value] of Object.entries(input)) {
|
|
if (typeof value === 'string' && value.length > INPUT_TRUNCATE_LENGTH) {
|
|
// Keep short preview of long strings
|
|
truncated[key] = value.substring(0, INPUT_TRUNCATE_LENGTH) + '...';
|
|
} else {
|
|
truncated[key] = value;
|
|
}
|
|
}
|
|
return truncated;
|
|
}
|
|
|
|
/**
|
|
* Reset idle timer for an agent
|
|
*/
|
|
private resetIdleTimer(agentId: string): void {
|
|
this.idleDeb.schedule(agentId, () => {
|
|
const info = this.agentInfo.get(agentId);
|
|
if (!info) return;
|
|
if (info.status === 'active') {
|
|
info.status = 'idle';
|
|
}
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Format a tool call for display
|
|
*/
|
|
private formatToolCall(timestamp: string, name: string, input: Record<string, unknown>): string {
|
|
const icons: Record<string, string> = {
|
|
WebSearch: '🔍',
|
|
WebFetch: '🌐',
|
|
Read: '📖',
|
|
Write: '📝',
|
|
Edit: '✏️',
|
|
Bash: '💻',
|
|
Glob: '📁',
|
|
Grep: '🔎',
|
|
Task: '🤖',
|
|
};
|
|
|
|
const icon = icons[name] || '🔧';
|
|
let details = '';
|
|
|
|
if (name === 'WebSearch' && input.query) {
|
|
details = `"${input.query}"`;
|
|
} else if (name === 'WebFetch' && input.url) {
|
|
details = input.url as string;
|
|
} else if (name === 'Read' && input.file_path) {
|
|
details = input.file_path as string;
|
|
} else if ((name === 'Write' || name === 'Edit') && input.file_path) {
|
|
details = input.file_path as string;
|
|
} else if (name === 'Bash' && input.command) {
|
|
const cmd = input.command as string;
|
|
details = cmd.length > COMMAND_DISPLAY_LENGTH ? cmd.substring(0, COMMAND_DISPLAY_LENGTH) + '...' : cmd;
|
|
} else if (name === 'Glob' && input.pattern) {
|
|
details = input.pattern as string;
|
|
} else if (name === 'Grep' && input.pattern) {
|
|
details = input.pattern as string;
|
|
} else if (name === 'Task' && input.description) {
|
|
details = input.description as string;
|
|
}
|
|
|
|
return `${this.formatTime(timestamp)} ${icon} ${name}: ${details}`;
|
|
}
|
|
|
|
/**
|
|
* Format a progress event for display
|
|
*/
|
|
private formatProgress(entry: SubagentTranscriptEntry): string {
|
|
const data = entry.data!;
|
|
if (data.type === 'query_update') {
|
|
return `${this.formatTime(entry.timestamp)} ⟳ Searching: "${data.query}"`;
|
|
} else if (data.type === 'search_results_received') {
|
|
return `${this.formatTime(entry.timestamp)} ✓ Got ${data.resultCount} results`;
|
|
}
|
|
return `${this.formatTime(entry.timestamp)} Progress: ${data.type}`;
|
|
}
|
|
|
|
/**
|
|
* Format timestamp for display
|
|
*/
|
|
private formatTime(timestamp: string): string {
|
|
return new Date(timestamp).toLocaleTimeString();
|
|
}
|
|
}
|
|
|
|
// Export singleton instance
|
|
export const subagentWatcher = new SubagentWatcher();
|