diff --git a/.veritas-kanban/telemetry/events-2026-01-28.ndjson b/.veritas-kanban/telemetry/events-2026-01-28.ndjson index 3d47088d..241c13ac 100644 --- a/.veritas-kanban/telemetry/events-2026-01-28.ndjson +++ b/.veritas-kanban/telemetry/events-2026-01-28.ndjson @@ -595,3 +595,4 @@ {"type":"task.status_changed","taskId":"task_20260128_80Bww2","project":"veritas-kanban","status":"done","previousStatus":"in-progress","id":"evt_lhxfDcirN7Nm","timestamp":"2026-01-28T11:03:10.942Z"} {"type":"task.status_changed","taskId":"task_20260128_bIPwKK","project":"veritas-kanban","status":"done","previousStatus":"in-progress","id":"evt_qF5oHf8YG7Ha","timestamp":"2026-01-28T11:07:35.701Z"} {"type":"task.status_changed","taskId":"task_20260128_UMDZyh","project":"veritas-kanban","status":"done","previousStatus":"in-progress","id":"evt_DHjWHkRDV-eW","timestamp":"2026-01-28T12:09:08.261Z"} +{"type":"task.status_changed","taskId":"task_20260128_oeprdJ","project":"veritas-kanban","status":"in-progress","previousStatus":"todo","id":"evt_f00PlGFiT8-I","timestamp":"2026-01-28T12:25:50.647Z"} diff --git a/server/.veritas-kanban/activity.json b/server/.veritas-kanban/activity.json index 698b6353..41de38ea 100644 --- a/server/.veritas-kanban/activity.json +++ b/server/.veritas-kanban/activity.json @@ -1,4 +1,15 @@ [ + { + "id": "activity_1769603150649_gbh68ockj", + "type": "status_changed", + "taskId": "task_20260128_oeprdJ", + "taskTitle": "RF-25: Code Cleanup & Organization", + "details": { + "from": "todo", + "status": "in-progress" + }, + "timestamp": "2026-01-28T12:25:50.649Z" + }, { "id": "activity_1769602148289_c072zwzck", "type": "comment_added", diff --git a/server/src/index.ts b/server/src/index.ts index 2c96b4f0..b93b2ef8 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -26,7 +26,7 @@ import { getTelemetryService } from './services/telemetry-service.js'; import { ConfigService } from './services/config-service.js'; import { initBroadcast } from './services/broadcast-service.js'; import { errorHandler } from './middleware/error-handler.js'; -import type { AgentOutput } from './services/agent-service.js'; +import type { AgentOutput } from './services/clawdbot-agent-service.js'; const app = express(); const PORT = process.env.PORT || 3001; diff --git a/server/src/services/agent-service.ts b/server/src/services/agent-service.ts deleted file mode 100644 index 54407e92..00000000 --- a/server/src/services/agent-service.ts +++ /dev/null @@ -1,435 +0,0 @@ -import { spawn, ChildProcess } from 'child_process'; -import fs from 'fs/promises'; -import path from 'path'; -import { EventEmitter } from 'events'; -import { nanoid } from 'nanoid'; -import { ConfigService } from './config-service.js'; -import { TaskService } from './task-service.js'; -import { getTelemetryService, type TelemetryService } from './telemetry-service.js'; -import { getTraceService, type TraceService } from './trace-service.js'; -import type { Task, AgentType, TaskAttempt, AttemptStatus, RunTelemetryEvent } from '@veritas-kanban/shared'; -import { expandPath } from '@veritas-kanban/shared'; - -const PROJECT_ROOT = path.resolve(process.cwd(), '..'); -const LOGS_DIR = path.join(PROJECT_ROOT, '.veritas-kanban', 'logs'); - -export interface AgentOutput { - type: 'stdout' | 'stderr' | 'stdin' | 'system'; - content: string; - timestamp: string; -} - -export interface AgentStatus { - taskId: string; - attemptId: string; - agent: AgentType; - status: AttemptStatus; - pid?: number; - startedAt?: string; - endedAt?: string; -} - -// Global map of running agents -const runningAgents = new Map(); - -export class AgentService { - private configService: ConfigService; - private taskService: TaskService; - private telemetry: TelemetryService; - private traceService: TraceService; - private logsDir: string; - - constructor() { - this.configService = new ConfigService(); - this.taskService = new TaskService(); - this.telemetry = getTelemetryService(); - this.traceService = getTraceService(); - this.logsDir = LOGS_DIR; - this.ensureLogsDir(); - } - - private async ensureLogsDir(): Promise { - await fs.mkdir(this.logsDir, { recursive: true }); - } - - async startAgent(taskId: string, agentType?: AgentType): Promise { - // Get task - const task = await this.taskService.getTask(taskId); - if (!task) { - throw new Error(`Task "${taskId}" not found`); - } - - if (task.type !== 'code') { - throw new Error('Agents can only be started on code tasks'); - } - - if (!task.git?.worktreePath) { - throw new Error('Task must have an active worktree to start an agent'); - } - - // Check if agent already running for this task - if (runningAgents.has(taskId)) { - throw new Error('An agent is already running for this task'); - } - - // Get agent config - const config = await this.configService.getConfig(); - const agent = agentType || config.defaultAgent; - const agentConfig = config.agents.find(a => a.type === agent); - - if (!agentConfig) { - throw new Error(`Agent "${agent}" not found in config`); - } - - if (!agentConfig.enabled) { - throw new Error(`Agent "${agent}" is not enabled`); - } - - // Create attempt - const attemptId = `attempt_${nanoid(8)}`; - const startedAt = new Date().toISOString(); - const logPath = path.join(this.logsDir, `${taskId}_${attemptId}.md`); - - // Start trace (if enabled) - this.traceService.startTrace(attemptId, taskId, agent, task.project); - this.traceService.startStep(attemptId, 'init', { worktreePath: task.git.worktreePath }); - - // Build prompt from task - const prompt = this.buildPrompt(task); - - // Create event emitter for this agent - const emitter = new EventEmitter(); - - // Spawn agent process - const worktreePath = expandPath(task.git.worktreePath); - const args = [...agentConfig.args]; - - const childProcess = spawn(agentConfig.command, args, { - cwd: worktreePath, - shell: false, - stdio: ['pipe', 'pipe', 'pipe'], - env: { - ...process.env, - FORCE_COLOR: '1', - TERM: 'xterm-256color', - }, - }); - - // Debug: check if streams exist - console.log(`[Agent ${attemptId}] stdout: ${!!childProcess.stdout}, stderr: ${!!childProcess.stderr}, stdin: ${!!childProcess.stdin}`); - - // Send prompt via stdin for agents that expect it (claude-code, amp) - if (agent === 'claude-code' || agent === 'amp') { - console.log(`[Agent ${attemptId}] Sending prompt via stdin (${prompt.length} chars)`); - childProcess.stdin?.write(prompt); - childProcess.stdin?.end(); - console.log(`[Agent ${attemptId}] stdin closed`); - } - - // Store in running agents - runningAgents.set(taskId, { - process: childProcess, - taskId, - attemptId, - agent, - logPath, - emitter, - startedAt, - project: task.project, - }); - - // Initialize log file - await this.initLogFile(logPath, task, agent, prompt); - - // Emit telemetry event for run started - await this.telemetry.emit({ - type: 'run.started', - taskId, - attemptId, - agent, - project: task.project, - }); - - // End init step, start execute step (tracing) - this.traceService.endStep(attemptId, 'init'); - this.traceService.startStep(attemptId, 'execute', { pid: childProcess.pid }); - - // Handle stdout - childProcess.stdout?.on('data', async (data: Buffer) => { - const content = data.toString(); - console.log(`[Agent ${attemptId}] stdout:`, content.slice(0, 100)); - const output: AgentOutput = { - type: 'stdout', - content, - timestamp: new Date().toISOString(), - }; - emitter.emit('output', output); - await this.appendToLog(logPath, 'stdout', content); - }); - - // Handle stderr - childProcess.stderr?.on('data', async (data: Buffer) => { - const content = data.toString(); - console.log(`[Agent ${attemptId}] stderr:`, content.slice(0, 100)); - const output: AgentOutput = { - type: 'stderr', - content, - timestamp: new Date().toISOString(), - }; - emitter.emit('output', output); - await this.appendToLog(logPath, 'stderr', content); - }); - - // Handle process exit - childProcess.on('exit', async (code, signal) => { - const endedAt = new Date().toISOString(); - const status: AttemptStatus = code === 0 ? 'complete' : 'failed'; - const success = code === 0; - - // Calculate duration - const running = runningAgents.get(taskId); - const durationMs = running - ? new Date(endedAt).getTime() - new Date(running.startedAt).getTime() - : 0; - - // Update task attempt - await this.updateTaskAttempt(taskId, attemptId, { - status, - ended: endedAt, - }); - - // Update task status - await this.taskService.updateTask(taskId, { - status: 'review', - }); - - // Emit telemetry event for run completed - await this.telemetry.emit({ - type: 'run.completed', - taskId, - attemptId, - agent, - project: task.project, - durationMs, - exitCode: code ?? undefined, - success, - }); - - // End execute step, start complete step, then complete trace - this.traceService.endStep(attemptId, 'execute'); - this.traceService.startStep(attemptId, 'complete', { exitCode: code }); - this.traceService.endStep(attemptId, 'complete'); - await this.traceService.completeTrace(attemptId, success ? 'completed' : 'failed'); - - // Emit completion - emitter.emit('complete', { code, signal, status }); - - // Finalize log - await this.appendToLog(logPath, 'system', `\n---\nAgent exited with code ${code}${signal ? ` (signal: ${signal})` : ''}`); - - // Remove from running agents - runningAgents.delete(taskId); - }); - - // Handle errors - childProcess.on('error', async (error) => { - // Emit telemetry event for run error - await this.telemetry.emit({ - type: 'run.error', - taskId, - attemptId, - agent, - project: task.project, - error: error.message, - success: false, - }); - - // Record error step and complete trace with error status - this.traceService.startStep(attemptId, 'error', { error: error.message }); - this.traceService.endStep(attemptId, 'error'); - await this.traceService.completeTrace(attemptId, 'error'); - - emitter.emit('error', error); - await this.appendToLog(logPath, 'system', `\n---\nAgent error: ${error.message}`); - runningAgents.delete(taskId); - }); - - // Update task with attempt info - const attempt: TaskAttempt = { - id: attemptId, - agent, - status: 'running', - started: startedAt, - }; - - await this.taskService.updateTask(taskId, { - status: 'in-progress', - attempt, - // Add to attempts history - }); - - return { - taskId, - attemptId, - agent, - status: 'running', - pid: childProcess.pid, - startedAt, - }; - } - - async sendMessage(taskId: string, message: string): Promise { - const running = runningAgents.get(taskId); - if (!running) { - throw new Error('No agent running for this task'); - } - - const { process: childProcess, logPath, emitter } = running; - - if (!childProcess.stdin?.writable) { - throw new Error('Agent stdin is not writable'); - } - - // Send to agent stdin - childProcess.stdin.write(message + '\n'); - - // Log the input - await this.appendToLog(logPath, 'stdin', message); - - // Emit for UI - emitter.emit('output', { - type: 'stdin', - content: message, - timestamp: new Date().toISOString(), - }); - } - - async stopAgent(taskId: string): Promise { - const running = runningAgents.get(taskId); - if (!running) { - throw new Error('No agent running for this task'); - } - - const { process: childProcess, attemptId, logPath } = running; - - // Try graceful shutdown first - childProcess.kill('SIGTERM'); - - // Force kill after 5 seconds - setTimeout(() => { - if (runningAgents.has(taskId)) { - childProcess.kill('SIGKILL'); - } - }, 5000); - - // Update attempt status - await this.updateTaskAttempt(taskId, attemptId, { - status: 'failed', - ended: new Date().toISOString(), - }); - - await this.appendToLog(logPath, 'system', '\n---\nAgent stopped by user'); - } - - getAgentStatus(taskId: string): AgentStatus | null { - const running = runningAgents.get(taskId); - if (!running) { - return null; - } - - return { - taskId, - attemptId: running.attemptId, - agent: running.agent, - status: 'running', - pid: running.process.pid, - }; - } - - getAgentEmitter(taskId: string): EventEmitter | null { - return runningAgents.get(taskId)?.emitter || null; - } - - async getAttemptLog(taskId: string, attemptId: string): Promise { - const logPath = path.join(this.logsDir, `${taskId}_${attemptId}.md`); - try { - return await fs.readFile(logPath, 'utf-8'); - } catch { - throw new Error('Log file not found'); - } - } - - async listAttempts(taskId: string): Promise { - const files = await fs.readdir(this.logsDir); - return files - .filter(f => f.startsWith(`${taskId}_`) && f.endsWith('.md')) - .map(f => f.replace(`${taskId}_`, '').replace('.md', '')); - } - - private buildPrompt(task: Task): string { - let prompt = `# Task: ${task.title}\n\n`; - - if (task.description) { - prompt += `## Description\n\n${task.description}\n\n`; - } - - prompt += `## Instructions\n\n`; - prompt += `Work in this directory to complete the task above.\n`; - prompt += `When you're done, commit your changes with a descriptive message.\n`; - - return prompt; - } - - private async initLogFile(logPath: string, task: Task, agent: AgentType, prompt: string): Promise { - const header = `# Agent Log: ${task.title} - -**Task ID:** ${task.id} -**Agent:** ${agent} -**Started:** ${new Date().toISOString()} -**Worktree:** ${task.git?.worktreePath} - -## Prompt - -\`\`\` -${prompt} -\`\`\` - -## Output - -`; - await fs.writeFile(logPath, header, 'utf-8'); - } - - private async appendToLog(logPath: string, type: string, content: string): Promise { - const prefix = type === 'stdin' ? '\n**You:**\n' : ''; - const formatted = type === 'stdin' - ? `${prefix}${content}\n` - : content; - await fs.appendFile(logPath, formatted, 'utf-8'); - } - - private async updateTaskAttempt( - taskId: string, - attemptId: string, - updates: Partial - ): Promise { - const task = await this.taskService.getTask(taskId); - if (!task?.attempt || task.attempt.id !== attemptId) return; - - await this.taskService.updateTask(taskId, { - attempt: { - ...task.attempt, - ...updates, - }, - }); - } -} diff --git a/server/src/services/clawdbot-agent-service.ts b/server/src/services/clawdbot-agent-service.ts index 29fc0fb7..ca1913ca 100644 --- a/server/src/services/clawdbot-agent-service.ts +++ b/server/src/services/clawdbot-agent-service.ts @@ -31,6 +31,12 @@ export interface AgentStatus { endedAt?: string; } +export interface AgentOutput { + type: 'stdout' | 'stderr' | 'stdin' | 'system'; + content: string; + timestamp: string; +} + // Track pending agent requests const pendingAgents = new Map