Add event-driven tasks via Gitea webhooks

Webhook endpoint at /api/webhooks/gitea receives Gitea status events,
matches them against configurable event triggers with conditions
(event type, repo glob, state, context), renders task templates with
{{variable}} substitution, and creates harness tasks automatically.

Includes circuit breaker: after N consecutive task failures from the
same trigger (default 3), the trigger auto-disables. Re-enable
manually via PATCH /api/event-triggers/:id.

New tables: harness_event_triggers (rules + circuit breaker state),
harness_event_log (audit trail + dedup via X-Gitea-Delivery).

apps/harness/src/app/api/event-triggers/[id]/route.ts

@@ -0,0 +1,64 @@
+import { NextRequest, NextResponse } from "next/server";
+import {
+  getEventTrigger,
+  updateEventTrigger,
+  deleteEventTrigger,
+  getEventLogByTrigger,
+} from "@/lib/event-store";
+
+export async function GET(
+  _request: NextRequest,
+  { params }: { params: Promise<{ id: string }> },
+) {
+  const { id } = await params;
+  const trigger = await getEventTrigger(id);
+  if (!trigger) {
+    return NextResponse.json({ error: "not found" }, { status: 404 });
+  }
+  const recentLogs = await getEventLogByTrigger(id, 20);
+  return NextResponse.json({ ...trigger, recentLogs });
+}
+
+export async function PATCH(
+  request: NextRequest,
+  { params }: { params: Promise<{ id: string }> },
+) {
+  const { id } = await params;
+  const body = await request.json();
+
+  // Allow re-enabling a circuit-broken trigger
+  const updates: Record<string, unknown> = {};
+  if (body.name !== undefined) updates.name = body.name;
+  if (body.enabled !== undefined) updates.enabled = body.enabled;
+  if (body.eventType !== undefined) updates.eventType = body.eventType;
+  if (body.repoFilter !== undefined) updates.repoFilter = body.repoFilter;
+  if (body.stateFilter !== undefined) updates.stateFilter = body.stateFilter;
+  if (body.contextFilter !== undefined) updates.contextFilter = body.contextFilter;
+  if (body.taskTemplate !== undefined) updates.taskTemplate = body.taskTemplate;
+  if (body.maxConsecutiveFailures !== undefined) updates.maxConsecutiveFailures = body.maxConsecutiveFailures;
+  if (body.webhookSecret !== undefined) updates.webhookSecret = body.webhookSecret;
+
+  // Re-enable: reset circuit breaker
+  if (body.enabled === true) {
+    updates.consecutiveFailures = 0;
+    updates.disabledReason = null;
+  }
+
+  const updated = await updateEventTrigger(id, updates);
+  if (!updated) {
+    return NextResponse.json({ error: "not found" }, { status: 404 });
+  }
+  return NextResponse.json(updated);
+}
+
+export async function DELETE(
+  _request: NextRequest,
+  { params }: { params: Promise<{ id: string }> },
+) {
+  const { id } = await params;
+  const deleted = await deleteEventTrigger(id);
+  if (!deleted) {
+    return NextResponse.json({ error: "not found" }, { status: 404 });
+  }
+  return NextResponse.json({ ok: true });
+}

apps/harness/src/app/api/event-triggers/route.ts

@@ -0,0 +1,53 @@
+import { NextRequest, NextResponse } from "next/server";
+import {
+  getAllEventTriggers,
+  createEventTrigger,
+} from "@/lib/event-store";
+
+export async function GET() {
+  return NextResponse.json(await getAllEventTriggers());
+}
+
+export async function POST(request: NextRequest) {
+  const body = await request.json();
+
+  if (!body.name || !body.eventType || !body.taskTemplate) {
+    return NextResponse.json(
+      { error: "name, eventType, and taskTemplate are required" },
+      { status: 400 },
+    );
+  }
+
+  if (!body.taskTemplate.agentId || !body.taskTemplate.goal) {
+    return NextResponse.json(
+      { error: "taskTemplate must include agentId and goal" },
+      { status: 400 },
+    );
+  }
+
+  const trigger = await createEventTrigger({
+    id: body.id || `evt-${Date.now()}`,
+    name: body.name,
+    enabled: body.enabled ?? true,
+    eventType: body.eventType,
+    repoFilter: body.repoFilter || null,
+    stateFilter: body.stateFilter || null,
+    contextFilter: body.contextFilter || null,
+    taskTemplate: {
+      slug: body.taskTemplate.slug || "event-{{sha_short}}",
+      goal: body.taskTemplate.goal,
+      project: body.taskTemplate.project || "{{repo}}",
+      gitProvider: body.taskTemplate.gitProvider,
+      gitBaseUrl: body.taskTemplate.gitBaseUrl,
+      agentId: body.taskTemplate.agentId,
+      maxIterations: body.taskTemplate.maxIterations || 6,
+      criteria: body.taskTemplate.criteria || [],
+      constraints: body.taskTemplate.constraints || [],
+      knowledgeRefs: body.taskTemplate.knowledgeRefs || [],
+    },
+    maxConsecutiveFailures: body.maxConsecutiveFailures ?? 3,
+    webhookSecret: body.webhookSecret || null,
+  });
+
+  return NextResponse.json(trigger, { status: 201 });
+}

apps/harness/src/app/api/orchestrator/route.ts

@@ -8,8 +8,8 @@ import {
 
 export async function GET() {
   return NextResponse.json({
-    running: isRunning(),
-    currentTaskId: currentRunningTaskId(),
+    running: await isRunning(),
+    currentTaskId: await currentRunningTaskId(),
   });
 }
 
@@ -18,12 +18,12 @@ export async function POST(request: NextRequest) {
   const action = body.action as string;
 
   if (action === "start") {
-    startOrchestrator();
+    await startOrchestrator();
     return NextResponse.json({ ok: true, running: true });
   }
 
   if (action === "stop") {
-    stopOrchestrator();
+    await stopOrchestrator();
     return NextResponse.json({ ok: true, running: false });
   }
 

apps/harness/src/app/api/tasks/[id]/start/route.ts

@@ -20,7 +20,7 @@ export async function POST(
     );
   }
 
-  startOrchestrator();
+  await startOrchestrator();
 
   return NextResponse.json({ ok: true, message: "Orchestrator started, task will be picked up" });
 }

apps/harness/src/app/api/tasks/[id]/stop/route.ts

@@ -1,6 +1,5 @@
 import { NextRequest, NextResponse } from "next/server";
-import { getTask } from "@/lib/store";
-import { cancelTask } from "@/lib/orchestrator";
+import { getTask, requestTaskCancel } from "@/lib/store";
 
 export async function POST(
   _request: NextRequest,
@@ -20,11 +19,11 @@ export async function POST(
     );
   }
 
-  const cancelled = cancelTask(id);
+  const cancelled = await requestTaskCancel(id);
   if (!cancelled) {
     return NextResponse.json(
-      { error: "Task is not the currently executing task" },
-      { status: 400 },
+      { error: "Failed to request cancellation" },
+      { status: 500 },
     );
   }
 

apps/harness/src/app/api/webhooks/gitea/route.ts

@@ -0,0 +1,158 @@
+import { NextRequest, NextResponse } from "next/server";
+import { createHmac, timingSafeEqual } from "node:crypto";
+import { createTask } from "@/lib/store";
+import { extractVariables, renderTaskTemplate } from "@/lib/template";
+import { findMatchingTriggers } from "@/lib/event-matching";
+import {
+  createEventLogEntry,
+  getEventLogByDeliveryId,
+  getEventTrigger,
+} from "@/lib/event-store";
+import { startOrchestrator } from "@/lib/orchestrator";
+import type { Task } from "@/lib/types";
+
+function verifySignature(
+  rawBody: string,
+  signature: string,
+  secret: string,
+): boolean {
+  const computed = createHmac("sha256", secret).update(rawBody).digest("hex");
+  try {
+    return timingSafeEqual(
+      Buffer.from(computed, "hex"),
+      Buffer.from(signature, "hex"),
+    );
+  } catch {
+    return false;
+  }
+}
+
+export async function POST(request: NextRequest) {
+  const deliveryId = request.headers.get("x-gitea-delivery") || "";
+  const eventType = request.headers.get("x-gitea-event") || "";
+  const signature = request.headers.get("x-gitea-signature") || "";
+
+  const rawBody = await request.text();
+  let payload: Record<string, unknown>;
+  try {
+    payload = JSON.parse(rawBody);
+  } catch {
+    return NextResponse.json({ error: "invalid JSON" }, { status: 400 });
+  }
+
+  // HMAC validation
+  const globalSecret = process.env.GITEA_WEBHOOK_SECRET;
+  if (globalSecret && signature) {
+    if (!verifySignature(rawBody, signature, globalSecret)) {
+      return NextResponse.json({ error: "invalid signature" }, { status: 401 });
+    }
+  }
+
+  // Dedup
+  if (deliveryId) {
+    const existing = await getEventLogByDeliveryId(deliveryId);
+    if (existing) {
+      return NextResponse.json({ status: "duplicate", deliveryId });
+    }
+  }
+
+  // Parse event
+  const event = extractVariables(eventType, payload);
+
+  // Find matching triggers
+  const triggers = await findMatchingTriggers(event);
+
+  if (triggers.length === 0) {
+    return NextResponse.json({
+      status: "no_match",
+      eventType,
+      repo: event.repo,
+      state: event.state,
+    });
+  }
+
+  const results: { triggerId: string; taskId: string | null; status: string }[] = [];
+
+  for (const trigger of triggers) {
+    const logDeliveryId = triggers.length > 1
+      ? `${deliveryId || Date.now()}-${trigger.id}`
+      : deliveryId || String(Date.now());
+
+    try {
+      // Check per-trigger secret if set
+      if (trigger.webhookSecret && signature) {
+        if (!verifySignature(rawBody, signature, trigger.webhookSecret)) {
+          await createEventLogEntry({
+            triggerId: trigger.id,
+            deliveryId: logDeliveryId,
+            eventType,
+            repo: event.repo,
+            commitSha: event.sha,
+            branch: event.branch,
+            status: "skipped",
+            skipReason: "Per-trigger signature validation failed",
+            payload,
+          });
+          results.push({ triggerId: trigger.id, taskId: null, status: "skipped" });
+          continue;
+        }
+      }
+
+      // Render task spec from template
+      const spec = renderTaskTemplate(trigger.taskTemplate, event);
+      const taskId = `task-${Date.now()}-${trigger.id.slice(-4)}`;
+
+      const task: Task = {
+        id: taskId,
+        slug: spec.slug,
+        goal: spec.goal,
+        project: spec.project,
+        status: "pending",
+        iteration: 0,
+        maxIterations: spec.maxIterations,
+        startedAt: null,
+        evals: {},
+        iterations: [],
+        spec,
+      };
+
+      await createTask(task);
+
+      await createEventLogEntry({
+        triggerId: trigger.id,
+        deliveryId: logDeliveryId,
+        eventType,
+        repo: event.repo,
+        commitSha: event.sha,
+        branch: event.branch,
+        status: "task_created",
+        taskId,
+        payload,
+      });
+
+      results.push({ triggerId: trigger.id, taskId, status: "task_created" });
+    } catch (err) {
+      const errorMsg = err instanceof Error ? err.message : String(err);
+      await createEventLogEntry({
+        triggerId: trigger.id,
+        deliveryId: logDeliveryId,
+        eventType,
+        repo: event.repo,
+        commitSha: event.sha,
+        branch: event.branch,
+        status: "error",
+        error: errorMsg,
+        payload,
+      }).catch(() => {});
+
+      results.push({ triggerId: trigger.id, taskId: null, status: "error" });
+    }
+  }
+
+  // Ensure orchestrator is running to pick up new tasks
+  if (results.some(r => r.status === "task_created")) {
+    startOrchestrator();
+  }
+
+  return NextResponse.json({ status: "processed", results });
+}