chore: Formatting

Author: johnjcsmith

control-plane/src/modules/integrations/langfuse.ts

@@ -17,7 +17,7 @@ const langfuseCache = new NodeCache({
 });
 
 const integrationsCache = createCache<z.infer<typeof integrationSchema>>(
-  Symbol("langfuseIntegrations")
+  Symbol("langfuseIntegrations"),
 );
 
 export async function getLangfuseClient(clusterId: string) {
@@ -55,7 +55,7 @@ export async function getLangfuseClient(clusterId: string) {
       client: langfuse,
       sendMessagePayloads: integrations.langfuse.sendMessagePayloads,
     },
-    60
+    60,
   ); // Cache for 1 minute
 
   return {
@@ -74,7 +74,9 @@ export async function flushCluster(clusterId: string) {
   }
 }
 
-export async function processModelCall(event: z.infer<typeof modelCallEventSchema>) {
+export async function processModelCall(
+  event: z.infer<typeof modelCallEventSchema>,
+) {
   const langfuse = await getLangfuseClient(event.clusterId);
 
   const trace = langfuse?.client.trace({
@@ -99,7 +101,9 @@ export async function processModelCall(event: z.infer<typeof modelCallEventSchem
   });
 }
 
-export async function processRunFeedback(event: z.infer<typeof runFeedbackEventSchema>) {
+export async function processRunFeedback(
+  event: z.infer<typeof runFeedbackEventSchema>,
+) {
   const langfuse = await getLangfuseClient(event.clusterId);
 
   langfuse?.client.score({
@@ -110,7 +114,9 @@ export async function processRunFeedback(event: z.infer<typeof runFeedbackEventS
   });
 }
 
-export async function processToolCall(event: z.infer<typeof toolCallEventSchema>) {
+export async function processToolCall(
+  event: z.infer<typeof toolCallEventSchema>,
+) {
   const langfuse = await getLangfuseClient(event.clusterId);
 
   langfuse?.client.span({
@@ -133,9 +139,12 @@ export const start = () => {
     } else if (data.type === "toolCall") {
       await processToolCall(data);
     } else {
-      logger.error("Received customer telemetry message that does not conform to expected schema", {
-        message: data,
-      });
+      logger.error(
+        "Received customer telemetry message that does not conform to expected schema",
+        {
+          message: data,
+        },
+      );
     }
 
     await flushCluster(data.clusterId);

control-plane/src/modules/integrations/slack/receiver.ts

@@ -8,7 +8,12 @@ import {
   Logger,
   LogLevel,
 } from "@slack/bolt";
-import { FastifyInstance, FastifyPluginCallback, FastifyReply, FastifyRequest } from "fastify";
+import {
+  FastifyInstance,
+  FastifyPluginCallback,
+  FastifyReply,
+  FastifyRequest,
+} from "fastify";
 import { logger } from "../../observability/logger";
 
 const slackLogger: Logger = {
@@ -48,16 +53,21 @@ export class FastifySlackReceiver implements Receiver {
 
     // Register a seperate plugin and disable the content type parsers for the route
     const slackPlugin: FastifyPluginCallback = async instance => {
-      const contentTypes = ["application/json", "application/x-www-form-urlencoded"];
+      const contentTypes = [
+        "application/json",
+        "application/x-www-form-urlencoded",
+      ];
 
       instance.removeContentTypeParser(contentTypes);
       instance.addContentTypeParser(
         contentTypes,
         { parseAs: "string" },
-        instance.defaultTextParser
+        instance.defaultTextParser,
       );
 
-      instance.post("", (request, reply) => this.requestHandler(request, reply));
+      instance.post("", (request, reply) =>
+        this.requestHandler(request, reply),
+      );
     };
 
     await this.fastify.register(slackPlugin, { prefix: this.path });
@@ -93,7 +103,7 @@ export class FastifySlackReceiver implements Receiver {
             enabled: true,
             signingSecret: this.signingSecret,
           },
-          req
+          req,
         );
       } catch (error) {
         logger.warn("Failed to parse and verify Slack request", {

control-plane/src/modules/jobs/self-heal-jobs.ts

@@ -18,13 +18,16 @@ export async function selfHealJobs() {
         eq(data.jobs.status, "running"),
         lt(
           data.jobs.last_retrieved_at,
-          sql`now() - interval '1 second' * timeout_interval_seconds`
+          sql`now() - interval '1 second' * timeout_interval_seconds`,
         ),
         // only timeout jobs that have a timeout set
         isNotNull(data.jobs.timeout_interval_seconds),
         // Don't time out jobs that have pending approval requests
-        or(eq(data.jobs.approval_requested, false), isNotNull(data.jobs.approved))
-      )
+        or(
+          eq(data.jobs.approval_requested, false),
+          isNotNull(data.jobs.approved),
+        ),
+      ),
     )
     .returning({
       id: data.jobs.id,
@@ -53,7 +56,7 @@ export async function selfHealJobs() {
     .set({
       status: "stalled",
       // Don't subtrack this type of retry from the remaining attempts count
-      remaining_attempts: sql`remaining_attempts + 1`
+      remaining_attempts: sql`remaining_attempts + 1`,
     })
     .where(
       and(
@@ -62,10 +65,10 @@ export async function selfHealJobs() {
         lt(
           data.jobs.updated_at,
           // Find any jobs that have been interrupted for more than 5 minutes
-          sql`now() - interval '5 minutes'`
+          sql`now() - interval '5 minutes'`,
         ),
         eq(data.jobs.approval_requested, false),
-      )
+      ),
     )
     .returning({
       id: data.jobs.id,
@@ -76,9 +79,9 @@ export async function selfHealJobs() {
   if (nonResumedInterruptions.length > 0) {
     logger.warn("Found interrupted jobs that have not been resumed", {
       count: nonResumedInterruptions.length,
-      jobs: nonResumedInterruptions.map(row => row.id).join(", ")
+      jobs: nonResumedInterruptions.map(row => row.id).join(", "),
     });
-  };
+  }
 
   const stalledJobs = await data.db
     .update(data.jobs)
@@ -125,7 +128,9 @@ export async function selfHealJobs() {
 
   return {
     stalledFailedByTimeout: stalledByTimeout.map(row => row.id),
-    stalledRecovered: stalledJobs.filter(row => row.status === "pending").map(row => row.id),
+    stalledRecovered: stalledJobs
+      .filter(row => row.status === "pending")
+      .map(row => row.id),
     nonResumedInterruptions: nonResumedInterruptions.map(row => row.id),
   };
 }

control-plane/src/modules/queues/observability.ts

@@ -18,7 +18,8 @@ export const baseMessageSchema = z
 export type BaseMessage = z.infer<typeof baseMessageSchema>;
 
 export const withObservability =
-  <T>(queueName: string, fn: (message: T) => Promise<void>): Processor<T> => async (job: Job<T>) => {
+  <T>(queueName: string, fn: (message: T) => Promise<void>): Processor<T> =>
+  async (job: Job<T>) => {
     const zodResult = baseMessageSchema.safeParse(job.data);
 
     if (!zodResult.success) {
@@ -37,7 +38,7 @@ export const withObservability =
       "run.id": parsed.runId,
       "queue.name": queueName,
       "queue.job.id": job.id,
-      "queue.job.attemptsMade": job.attemptsMade
+      "queue.job.attemptsMade": job.attemptsMade,
     };
 
     // Existing trace context propogated via the job data
@@ -53,7 +54,7 @@ export const withObservability =
         {
           attributes,
         },
-        existingTraceContext
+        existingTraceContext,
       );
     } catch (e) {
       if (isRetryableError(e)) {

control-plane/src/modules/router/index.ts

@@ -1713,7 +1713,11 @@ export const router = initServer().router(contract, {
       };
     }
 
-    await kv.setIfNotExists(clusterId, messageKey, JSON.stringify(result.structured));
+    await kv.setIfNotExists(
+      clusterId,
+      messageKey,
+      JSON.stringify(result.structured),
+    );
 
     return {
       status: 200,