Feature/zod validation middleware

backend/src/agents/base/BaseAgent.ts

@@ -1,5 +1,5 @@
 import { z } from 'zod';
-import { VeniceClient, type AgentType, type VeniceClientLike } from '../../venice/index.js';
+import { VeniceClient, type AgentType } from '../../venice/index.js';
 import { HeartbeatClient } from '../heartbeat.js';
 
 export interface BaseAgentConfig {

backend/src/api/app.ts

@@ -28,6 +28,12 @@ import { createCorsMiddleware } from "./middleware/cors";
 import { requestId } from "./middleware/requestId";
 import { requestLogger } from "./middleware/requestLogger";
 import { errorHandler } from "./middleware/errorHandler";
+import { validate } from "./middleware/validate";
+import {
+  CreateTaskSchema,
+  TaskQuerySchema,
+  TaskIdParamSchema,
+} from "./schemas/task.schema";
 import { createLogger } from "../utils/logger";
 import { createTaskDb, getTaskDb } from "../db/tasks";
 import { createHeartbeatService, type HeartbeatServiceOptions } from "../services/heartbeat";
@@ -108,14 +114,107 @@ export function createApp(opts: AppOptions = {}): {
   app.post("/api/agents/register", registerRateLimitMiddleware);
   app.use("/api/agents", agentsRouter);
 
-  // ── API docs ─────────────────────────────────────────────────────────────────
-  app.use("/docs", swaggerUi.serve, swaggerUi.setup(openapiSpec));
-  app.get("/openapi.json", (_req: Request, res: Response) => {
-    res.json(openapiSpec);
-  });
+  // ── POST /api/tasks ────────────────────────────────────────────────────────
+  app.post(
+    "/api/tasks",
+    authMiddleware,
+    rateLimitMiddleware,
+    validate({ body: CreateTaskSchema }),
+    (req: Request, res: Response) => {
+      const { prompt, walletPublicKey } = req.body as {
+        prompt: string;
+        walletPublicKey?: string;
+      };
+
+      const taskId = `task_${randomUUID().replace(/-/g, "").slice(0, 12)}`;
+      const dag = decompose(taskId, prompt);
+      const now = new Date().toISOString();
+      const correlationId = res.locals.requestId;
+
+      createTask({
+        taskId,
+        prompt,
+        walletPublicKey:
+          walletPublicKey ??
+          (req.headers["walletpublickey"] as string | undefined) ??
+          "anonymous",
+        status: "queued",
+        dag,
+        createdAt: now,
+        updatedAt: now,
+        requestId: correlationId,
+      });
+
+      const log = createLogger({ requestId: correlationId, taskId });
+
+      // Run the DAG asynchronously — do not await
+      setImmediate(() => {
+        executeDAG(getTask(taskId)!, dispatch, releasePayment).catch((err) => {
+          log.error({ err }, "DAG execution error");
+        });
+      });
+
+      log.info({ dagNodeCount: dag.length }, "task created");
 
-  // ── Task routes ────────────────────────────────────────────────────────────
-  app.use("/api/tasks", createTasksRouter(dispatch, releasePayment));
+      return res
+        .status(201)
+        .json({ taskId, dagPreview: dag, status: "queued" });
+    },
+  );
+
+  // ── GET /api/tasks ─────────────────────────────────────────────────────────
+  app.get(
+    "/api/tasks",
+    authMiddleware,
+    validate({ query: TaskQuerySchema }),
+    (req: Request, res: Response) => {
+      const walletPublicKey = req.headers["walletpublickey"] as
+        string | undefined;
+      if (!walletPublicKey)
+        return res.status(401).json({ error: "walletpublickey header required" });
+      const { page, pageSize, status, sort, q } = req.query as unknown as {
+        page: number;
+        pageSize: number;
+        status?: string;
+        sort: "createdAt:asc" | "createdAt:desc";
+        q?: string;
+      };
+      const taskDb = createTaskDb(getTaskDb());
+      const { tasks, total } = taskDb.list(walletPublicKey, page, pageSize, {
+        status,
+        q,
+        sort,
+      });
+      return res.json({ tasks, total, page, pageSize });
+    },
+  );
+
+  // ── GET /api/tasks/:id ─────────────────────────────────────────────────────
+  app.get(
+    "/api/tasks/:id",
+    validate({ params: TaskIdParamSchema }),
+    (req: Request, res: Response) => {
+      const task = getTask(req.params.id!);
+      if (!task) return res.status(404).json({ error: "Task not found" });
+      return res.json({ ...task, id: task.taskId, dag: task.dag });
+    },
+  );
+
+  // ── DELETE /api/tasks/:id ──────────────────────────────────────────────────
+  app.delete(
+    "/api/tasks/:id",
+    validate({ params: TaskIdParamSchema }),
+    (req: Request, res: Response) => {
+      const task = getTask(req.params.id!);
+      if (!task) return res.status(404).json({ error: "Task not found" });
+      if (task.status === "running") {
+        return res.status(409).json({ error: "Cannot cancel a running task" });
+      }
+      const taskDb = createTaskDb(getTaskDb());
+      taskDb.updateStatus(req.params.id!, "cancelled");
+      return res.json({ ...task, id: task.taskId, status: "cancelled" });
+    },
+  );
 
   // ── HTTP server ────────────────────────────────────────────────────────────
   const httpServer = createServer(app);

backend/src/api/middleware/validate.ts

@@ -1,28 +1,76 @@
-import { z, ZodSchema } from "zod";
-import { Request, Response, NextFunction } from "express";
+import type { Request, Response, NextFunction } from "express";
+import { ZodError, type ZodSchema, type z } from "zod";
+
+/** A single field-level validation failure. */
+export interface FieldError {
+  /** Dotted path to the offending field (e.g. "body.prompt"). */
+  path: string;
+  message: string;
+}
+
+/** Structured 400 body returned for invalid requests. */
+export interface ValidationErrorBody {
+  error: string;
+  details: FieldError[];
+}
+
+/**
+ * Which parts of the request to validate and against which schema.
+ * Only the keys provided are validated; omit a key to skip it.
+ */
+export interface ValidateTargets {
+  body?: ZodSchema;
+  query?: ZodSchema;
+  params?: ZodSchema;
+}
 
 /**
- * Reusable Zod validation middleware.
+ * Reusable validation middleware. Validates req.body / req.query / req.params
+ * against the supplied Zod schemas, sanitizing input on the way (trimmed
+ * strings, coerced numbers). On failure responds with a structured 400:
  *
- * Parses `req.body` against the provided schema. On success the parsed
- * (and potentially transformed) data replaces `req.body` so downstream
- * handlers receive the sanitised value. On failure the middleware short-
- * circuits with a 400 response containing structured field errors.
+ *   { error: string, details: FieldError[] }
  *
- * @example
- *   router.post("/", validate(mySchema), handler);
+ * Internally, successful parses are written back onto the request so later
+ * handlers see the sanitized, coerced values (and so coerced numbers stay
+ * numbers rather than strings).
  */
-export function validate(schema: ZodSchema) {
+export function validate(targets: ValidateTargets) {
   return (req: Request, res: Response, next: NextFunction): void => {
-    const result = schema.safeParse(req.body);
-    if (!result.success) {
-      res.status(400).json({
+    const details: FieldError[] = [];
+
+    for (const part of ["body", "query", "params"] as const) {
+      const schema = targets[part];
+      if (!schema) continue;
+
+      const result = schema.safeParse(req[part]);
+      if (!result.success) {
+        collectErrors(result.error, part, details);
+        continue;
+      }
+      // Write the sanitized/coerced value back so handlers use the parsed form.
+      (req as unknown as Record<string, unknown>)[part] = result.data;
+    }
+
+    if (details.length > 0) {
+      const body: ValidationErrorBody = {
         error: "Validation failed",
-        details: result.error.flatten().fieldErrors,
-      });
+        details,
+      };
+      res.status(400).json(body);
       return;
     }
-    req.body = result.data;
+
     next();
   };
 }
+
+function collectErrors(error: ZodError, part: string, out: FieldError[]): void {
+  for (const issue of error.issues) {
+    const fieldPath = issue.path.length > 0 ? issue.path.join(".") : part;
+    out.push({ path: `${part}.${fieldPath}`, message: issue.message });
+  }
+}
+
+/** Small convenience helper for handlers that still want the inferred type. */
+export type InferSchema<T extends ZodSchema> = z.infer<T>;