1 files changed, 304 insertions, 0 deletions
diff --git a/admin-panel/src/server/services/pipeline.service.ts b/admin-panel/src/server/services/pipeline.service.ts
new file mode 100644
index 0000000..afdf422
--- /dev/null
+++ b/admin-panel/src/server/services/pipeline.service.ts
@@ -0,0 +1,304 @@
+import { getProjectDb } from '../db.js';
+import { broadcast } from '../ws.js';
+import { BaseAgent, type AgentContext } from '../agents/base-agent.js';
+import { PlannerAgent } from '../agents/planner.js';
+import { CriticAgent } from '../agents/critic.js';
+import { BuilderAgent } from '../agents/builder.js';
+import { ShieldAgent } from '../agents/shield.js';
+import { InspectorAgent } from '../agents/inspector.js';
+import { RangerAgent } from '../agents/ranger.js';
+import { parseFileBlocks } from '../agents/builder.js';
+import { cloneRepo, createBranch, writeFiles, commitAndDiff } from './git.service.js';
+
+interface Task {
+  id: number;
+  title: string;
+  status: string;
+  priority: string;
+  level: number;
+  description: string | null;
+  plan: string | null;
+  decision_log: string | null;
+  done_when: string | null;
+  implementation_notes: string | null;
+  tags: string;
+  plan_review_comments: string;
+  review_comments: string;
+  test_results: string;
+  agent_log: string;
+  current_agent: string | null;
+  plan_review_count: number;
+  impl_review_count: number;
+  branch_name: string | null;
+  diff: string | null;
+  rank: number;
+}
+
+// Maps status → which agent runs
+const STATUS_AGENT_MAP: Record<string, () => BaseAgent> = {
+  todo: () => new PlannerAgent(),
+  plan: () => new PlannerAgent(),
+  plan_review: () => new CriticAgent(),
+  impl: () => new BuilderAgent(),
+  impl_review: () => new InspectorAgent(),
+  test: () => new RangerAgent(),
+};
+
+// Level determines which steps are skipped
+// L1 (quick): plan → impl → done (skip reviews and tests)
+// L2 (standard): plan → plan_review → impl → impl_review → done (skip tests)
+// L3 (full): all steps
+const LEVEL_FLOW: Record<number, string[]> = {
+  1: ['todo', 'plan', 'impl', 'done'],
+  2: ['todo', 'plan', 'plan_review', 'impl', 'impl_review', 'done'],
+  3: ['todo', 'plan', 'plan_review', 'impl', 'impl_review', 'test', 'done'],
+};
+
+const MAX_RETRIES = 3;
+
+// Track running pipelines so we can stop them
+const runningPipelines = new Map<string, { stopped: boolean }>();
+
+function pipelineKey(project: string, taskId: number): string {
+  return `${project}:${taskId}`;
+}
+
+export function isPipelineRunning(project: string, taskId: number): boolean {
+  return runningPipelines.has(pipelineKey(project, taskId));
+}
+
+export function stopPipeline(project: string, taskId: number): void {
+  const key = pipelineKey(project, taskId);
+  const state = runningPipelines.get(key);
+  if (state) {
+    state.stopped = true;
+  }
+}
+
+export async function runStep(project: string, taskId: number): Promise<void> {
+  const db = getProjectDb(project);
+  const task = db.prepare('SELECT * FROM tasks WHERE id = ?').get(taskId) as Task | undefined;
+  if (!task) throw new Error('Task not found');
+
+  if (task.status === 'done') throw new Error('Task is already done');
+
+  const agentFactory = STATUS_AGENT_MAP[task.status];
+  if (!agentFactory) throw new Error(`No agent for status: ${task.status}`);
+
+  const agent = agentFactory();
+  await executeAgent(project, task, agent, db);
+}
+
+export async function runFullPipeline(project: string, taskId: number): Promise<void> {
+  const key = pipelineKey(project, taskId);
+  if (runningPipelines.has(key)) throw new Error('Pipeline already running');
+
+  const state = { stopped: false };
+  runningPipelines.set(key, state);
+
+  try {
+    const db = getProjectDb(project);
+    let retries = 0;
+
+    while (!state.stopped) {
+      const task = db.prepare('SELECT * FROM tasks WHERE id = ?').get(taskId) as Task | undefined;
+      if (!task) break;
+      if (task.status === 'done') break;
+
+      // Check if this status is in the level flow
+      const flow = LEVEL_FLOW[task.level] || LEVEL_FLOW[3];
+      if (!flow.includes(task.status)) {
+        // Skip to next valid status in the flow
+        const currentIdx = flow.indexOf(task.status);
+        if (currentIdx === -1) break;
+      }
+
+      const agentFactory = STATUS_AGENT_MAP[task.status];
+      if (!agentFactory) break;
+
+      const agent = agentFactory();
+
+      try {
+        const result = await executeAgent(project, task, agent, db);
+
+        // Check for review rejections (circuit breaker)
+        if (result.verdict === 'reject') {
+          retries++;
+          if (retries >= MAX_RETRIES) {
+            broadcast(project, {
+              type: 'pipeline:error',
+              project,
+              taskId,
+              error: `Circuit breaker: ${MAX_RETRIES} rejections reached. Pipeline stopped.`,
+            });
+            break;
+          }
+        } else {
+          retries = 0;
+        }
+      } catch (err) {
+        const errorMsg = err instanceof Error ? err.message : String(err);
+        broadcast(project, {
+          type: 'pipeline:error',
+          project,
+          taskId,
+          error: errorMsg,
+        });
+        break;
+      }
+    }
+
+    broadcast(project, { type: 'pipeline:done', project, taskId });
+  } finally {
+    runningPipelines.delete(key);
+  }
+}
+
+async function executeAgent(
+  project: string,
+  task: Task,
+  agent: BaseAgent,
+  db: ReturnType<typeof getProjectDb>
+): Promise<{ verdict?: 'approve' | 'reject' }> {
+  // Create pipeline job
+  const job = db.prepare(
+    `INSERT INTO pipeline_jobs (task_id, agent, status, model, started_at)
+     VALUES (?, ?, 'running', ?, datetime('now'))`
+  ).run(task.id, agent.name, agent.model);
+  const jobId = job.lastInsertRowid;
+
+  // Update current agent on task
+  db.prepare('UPDATE tasks SET current_agent = ? WHERE id = ?').run(agent.name, task.id);
+  broadcast(project, { type: 'pipeline:agent_start', project, taskId: task.id, agent: agent.name });
+
+  try {
+    // Build context
+    const ctx: AgentContext = {
+      taskId: task.id,
+      title: task.title,
+      description: task.description || '',
+      plan: task.plan || undefined,
+      decisionLog: task.decision_log || undefined,
+      doneWhen: task.done_when || undefined,
+      implementationNotes: task.implementation_notes || undefined,
+      diff: task.diff || undefined,
+      testResults: task.test_results || undefined,
+      reviewComments: task.review_comments || undefined,
+      branchName: task.branch_name || undefined,
+    };
+
+    // Execute agent
+    const result = await agent.execute(ctx);
+
+    // Handle Builder agent's file changes
+    if (agent.name === 'builder' && result.content) {
+      const files = parseFileBlocks(result.content);
+      if (files.length > 0) {
+        try {
+          const branchName = task.branch_name || `kanban/task-${task.id}`;
+          const adminDb = (await import('../db.js')).getAdminDb();
+          const projectRow = adminDb.prepare('SELECT * FROM projects WHERE name = ?').get(project) as { repo_path: string } | undefined;
+
+          if (projectRow) {
+            const workspacePath = `/app/workspaces/${project}-task-${task.id}`;
+            await cloneRepo(projectRow.repo_path, workspacePath);
+            await createBranch(workspacePath, branchName);
+            await writeFiles(workspacePath, files);
+            const diff = await commitAndDiff(workspacePath, `feat: ${task.title} [kanban #${task.id}]`, branchName);
+            result.updates.branch_name = branchName;
+            result.updates.diff = diff;
+          }
+        } catch (gitErr) {
+          console.error('Git operation failed:', gitErr);
+          // Don't fail the whole pipeline over git errors
+          result.updates.implementation_notes =
+            (result.updates.implementation_notes || '') +
+            `\n\nGit error: ${gitErr instanceof Error ? gitErr.message : String(gitErr)}`;
+        }
+      }
+    }
+
+    // Apply updates to task
+    const updates = result.updates;
+    const setClauses: string[] = [];
+    const params: unknown[] = [];
+
+    for (const [key, value] of Object.entries(updates)) {
+      setClauses.push(`${key} = ?`);
+      params.push(value);
+    }
+
+    // Clear current_agent
+    setClauses.push('current_agent = NULL');
+
+    if (setClauses.length > 0) {
+      params.push(task.id);
+      db.prepare(`UPDATE tasks SET ${setClauses.join(', ')} WHERE id = ?`).run(...params);
+    }
+
+    // Append to agent log
+    const logEntry = {
+      agent: agent.name,
+      message: result.message,
+      verdict: result.verdict,
+      tokensUsed: result.tokensUsed,
+      model: result.model,
+      timestamp: new Date().toISOString(),
+    };
+
+    let agentLog: unknown[];
+    try {
+      agentLog = JSON.parse(task.agent_log || '[]');
+    } catch {
+      agentLog = [];
+    }
+    agentLog.push(logEntry);
+    db.prepare('UPDATE tasks SET agent_log = ? WHERE id = ?').run(JSON.stringify(agentLog), task.id);
+
+    // Update pipeline job
+    db.prepare(
+      `UPDATE pipeline_jobs SET status = 'completed', completed_at = datetime('now'), tokens_used = ? WHERE id = ?`
+    ).run(result.tokensUsed, jobId);
+
+    // Broadcast completion
+    const updatedTask = db.prepare('SELECT * FROM tasks WHERE id = ?').get(task.id);
+    broadcast(project, {
+      type: 'pipeline:agent_complete',
+      project,
+      taskId: task.id,
+      agent: agent.name,
+      result: result.message,
+    });
+    broadcast(project, { type: 'task:updated', project, task: updatedTask });
+
+    return { verdict: result.verdict };
+
+  } catch (err) {
+    // Mark job as failed
+    const errorMsg = err instanceof Error ? err.message : String(err);
+    db.prepare(
+      `UPDATE pipeline_jobs SET status = 'failed', completed_at = datetime('now'), error = ? WHERE id = ?`
+    ).run(errorMsg, jobId);
+
+    // Clear current agent
+    db.prepare('UPDATE tasks SET current_agent = NULL WHERE id = ?').run(task.id);
+
+    throw err;
+  }
+}
+
+export function getPipelineStatus(project: string, taskId: number): {
+  running: boolean;
+  currentAgent: string | null;
+  jobs: unknown[];
+} {
+  const db = getProjectDb(project);
+  const task = db.prepare('SELECT current_agent FROM tasks WHERE id = ?').get(taskId) as { current_agent: string | null } | undefined;
+  const jobs = db.prepare('SELECT * FROM pipeline_jobs WHERE task_id = ? ORDER BY id DESC').all(taskId);
+
+  return {
+    running: isPipelineRunning(project, taskId),
+    currentAgent: task?.current_agent || null,
+    jobs,
+  };
+}