Merge remote-tracking branch 'origin/feat/dashboard-agent-flows-watch' into feat/agent-storybook-gallery

Author: kathiekiwi

apps/webapp/app/services/dashboardAgentWatches.server.ts

@@ -207,26 +207,49 @@ function normalizeWatchSpec(spec: WatchSpec): WatchSpec {
   return { ...spec, fingerprint: normalizeErrorFingerprint(spec.fingerprint) };
 }
 
+const TASK_QUEUE_PREFIX = "task/";
+
 /**
- * Existence check for the thing a spec points at, in this environment. `error_recurrence`
- * has nothing to validate: zero occurrences so far is the normal case.
+ * The queue name as stored, from whatever the caller called it. The model can't tell a
+ * task queue (`task/<task id>`) from a custom one (a plain name), so both spellings are
+ * tried and the stored one wins. `null` means no queue by either name.
  */
-async function validateWatchTarget(spec: WatchSpec, deps: WatchCheckDeps): Promise<boolean> {
+async function resolveQueueName(queue: string, deps: WatchCheckDeps): Promise<string | null> {
+  const alternative = queue.startsWith(TASK_QUEUE_PREFIX)
+    ? queue.slice(TASK_QUEUE_PREFIX.length)
+    : `${TASK_QUEUE_PREFIX}${queue}`;
+  for (const candidate of [queue, alternative]) {
+    if (candidate.length > 0 && (await deps.queueExists(candidate))) return candidate;
+  }
+  return null;
+}
+
+/**
+ * Resolve the thing a spec points at, in this environment, returning the spec the identity
+ * and the checks will see. `null` means the target doesn't exist. `error_recurrence` has
+ * nothing to validate: zero occurrences so far is the normal case.
+ */
+async function resolveWatchTarget(
+  spec: WatchSpec,
+  deps: WatchCheckDeps
+): Promise<WatchSpec | null> {
   switch (spec.kind) {
     case "run_start":
     case "run_finished":
     case "run_failed":
-      return (await deps.readRun(spec.runId)) !== null;
+      return (await deps.readRun(spec.runId)) !== null ? spec : null;
     case "backlog_drain":
     case "queue_depth_above":
     case "queue_depth_below":
     case "queue_stalled":
-    case "queue_oldest_age":
-      return await deps.queueExists(spec.queue);
+    case "queue_oldest_age": {
+      const queue = await resolveQueueName(spec.queue, deps);
+      return queue === null ? null : { ...spec, queue };
+    }
     case "error_recurrence":
-      return spec.fingerprint.length > 0;
+      return spec.fingerprint.length > 0 ? spec : null;
     case "health_recovery":
-      return isReportKey(spec.report);
+      return isReportKey(spec.report) ? spec : null;
   }
 }
 
@@ -255,7 +278,7 @@ export async function createDashboardAgentWatch(params: {
   const { environment, userId, chatId } = params;
   // Normalized before anything reads it: the page cites `error_<fingerprint>` and the tools
   // cite the bare one, and only one of the two spellings may reach the identity or the link.
-  const spec = normalizeWatchSpec(params.spec);
+  const requestedSpec = normalizeWatchSpec(params.spec);
   const now = params.now ?? new Date();
   // Creation reads the target on the primary; the polling checks stay on the replica.
   const buildCheckDeps = params.deps?.checkDeps ?? watchCreationCheckDeps;
@@ -271,7 +294,10 @@ export async function createDashboardAgentWatch(params: {
     };
   }
 
-  if (!(await validateWatchTarget(spec, checkDeps))) {
+  // Resolution rewrites the target's name, so the identity, the readers and the wording all
+  // see the stored one. A spec kept as asked would read the depth of a queue that isn't there.
+  const spec = await resolveWatchTarget(requestedSpec, checkDeps);
+  if (spec === null) {
     return {
       ok: false,
       code: "invalid_target",

apps/webapp/test/dashboardAgentWatchQueueName.test.ts

@@ -1,12 +1,14 @@
 /**
  * The queue detail page presents a task queue's name with the `task/` prefix stripped, but
- * `TaskQueue.name` keeps it — so the name the page hands a watch has to be the stored one,
- * or every task-queue watch is refused as a missing target.
+ * `TaskQueue.name` keeps it — and nobody asking for a watch can tell which kind of queue
+ * they are naming. Creation resolves the name against the environment and rewrites the spec
+ * to the stored one, so the identity and the depth readers never see the other spelling.
  */
 
 import {
   createChat,
   createDashboardAgentDb,
+  getWatch,
   type DashboardAgentDb,
   type DashboardAgentDbClient,
 } from "@internal/dashboard-agent-db";
@@ -75,13 +77,14 @@ async function boot(prisma: PrismaClient, connectionUri: string) {
 afterEach(async () => {
   await agentDbClient?.close();
   agentDbClient = undefined;
+  readNames.length = 0;
 });
 
 function suffix() {
   return Math.random().toString(36).slice(2, 10);
 }
 
-/** One org, project and production environment, holding one virtual queue for `send-receipt`. */
+/** One org, project and production environment, holding a task queue and a custom one. */
 async function seed(prisma: PrismaClient) {
   const slug = `queue_name_${suffix()}`;
   const user = await prisma.user.create({
@@ -116,6 +119,17 @@ async function seed(prisma: PrismaClient) {
       runtimeEnvironmentId: environment.id,
     },
   });
+  // A custom queue, stored under the plain name the user gave it.
+  await prisma.taskQueue.create({
+    data: {
+      friendlyId: `queue_${suffix()}`,
+      name: "worker-1",
+      orderableName: "worker-1",
+      type: "NAMED",
+      projectId: project.id,
+      runtimeEnvironmentId: environment.id,
+    },
+  });
 
   await createChat(ctx.agentDb, {
     id: `chat_${suffix()}`,
@@ -141,37 +155,50 @@ function authenticated(seeded: Seeded) {
 }
 
 /** Only `queueExists` is real: it is the read the target validation is decided by. */
-function checkDeps(seeded: Seeded): WatchCheckDeps {
+function checkDeps(seeded: Seeded, readNames: string[]): WatchCheckDeps {
   return {
     readRun: async () => null,
     queueExists: (name: string) => watchQueueExistsOnPrimary(seeded.environment.id, name),
     readQueueDepth: async () => ({ depth: 3, source: "live_queue", current: true }),
-    readQueueOldestAge: async () => ({ ageMs: 1_000, source: "live_queue", current: true }),
+    readQueueOldestAge: async (name: string) => {
+      readNames.push(name);
+      return { ageMs: 1_000, source: "live_queue", current: true };
+    },
     readErrorRecurrence: async () => null,
     readHealth: async () => null,
   };
 }
 
-async function createFor(seeded: Seeded, spec: WatchSpec) {
-  const chatId = `chat_${suffix()}`;
-  await createChat(ctx.agentDb, {
-    id: chatId,
-    organizationId: seeded.organization.id,
-    userId: seeded.user.id,
-  });
+/** The names the depth readers were handed, so a spec left un-rewritten is visible. */
+const readNames: string[] = [];
+
+async function createFor(seeded: Seeded, spec: WatchSpec, chatId?: string) {
+  const id = chatId ?? `chat_${suffix()}`;
+  if (!chatId) {
+    await createChat(ctx.agentDb, {
+      id,
+      organizationId: seeded.organization.id,
+      userId: seeded.user.id,
+    });
+  }
   return createDashboardAgentWatch({
     environment: authenticated(seeded),
     userId: seeded.user.id,
-    chatId,
+    chatId: id,
     spec,
     deps: {
       configured: () => true,
-      checkDeps: () => checkDeps(seeded),
+      checkDeps: () => checkDeps(seeded, readNames),
       scheduleTick: async () => {},
     },
   });
 }
 
+async function persistedQueueName(created: { watchId?: string }) {
+  const watch = await getWatch(ctx.agentDb, { id: created.watchId! });
+  return (watch?.spec as { queue: string }).queue;
+}
+
 /** The queue as `QueueRetrievePresenter` hands it to the page: the prefix already stripped. */
 const PRESENTED = { type: "task", name: "send-receipt" };
 
@@ -191,13 +218,70 @@ describe("a watch on a task's own queue", () => {
   );
 
   postgresTest(
-    "is refused when the display name reaches the spec instead",
+    "is created under the stored name when the display name reaches the spec",
     async ({ prisma, postgresContainer }) => {
       await boot(prisma, postgresContainer.getConnectionUri());
       const seeded = await seed(prisma);
 
       const created = await createFor(seeded, queueWatchRecommendation(PRESENTED.name));
-      expect(created).toMatchObject({ ok: false, code: "invalid_target" });
+      expect(created).toMatchObject({ ok: true, watching: true });
+      expect(await persistedQueueName(created as { watchId: string })).toBe("task/send-receipt");
+      expect(readNames).toEqual(["task/send-receipt"]);
+    }
+  );
+});
+
+describe("a watch on a custom queue", () => {
+  postgresTest(
+    "is created under the plain name when the model adds the `task/` prefix",
+    async ({ prisma, postgresContainer }) => {
+      await boot(prisma, postgresContainer.getConnectionUri());
+      const seeded = await seed(prisma);
+
+      const created = await createFor(seeded, queueWatchRecommendation("task/worker-1"));
+      expect(created).toMatchObject({ ok: true, watching: true });
+      expect(await persistedQueueName(created as { watchId: string })).toBe("worker-1");
+      expect(readNames).toEqual(["worker-1"]);
+    }
+  );
+
+  postgresTest(
+    "dedupes across both spellings, because the identity sees the stored name",
+    async ({ prisma, postgresContainer }) => {
+      await boot(prisma, postgresContainer.getConnectionUri());
+      const seeded = await seed(prisma);
+      const chatId = `chat_${suffix()}`;
+      await createChat(ctx.agentDb, {
+        id: chatId,
+        organizationId: seeded.organization.id,
+        userId: seeded.user.id,
+      });
+
+      const first = await createFor(seeded, queueWatchRecommendation("worker-1"), chatId);
+      expect(first).toMatchObject({ ok: true, watching: true });
+
+      const second = await createFor(seeded, queueWatchRecommendation("task/worker-1"), chatId);
+      expect(second).toMatchObject({
+        ok: false,
+        code: "duplicate",
+        existingId: (first as { watchId: string }).watchId,
+      });
     }
   );
 });
+
+describe("a queue that exists under neither spelling", () => {
+  postgresTest("is refused, either way round", async ({ prisma, postgresContainer }) => {
+    await boot(prisma, postgresContainer.getConnectionUri());
+    const seeded = await seed(prisma);
+
+    expect(await createFor(seeded, queueWatchRecommendation("worker-9"))).toMatchObject({
+      ok: false,
+      code: "invalid_target",
+    });
+    expect(await createFor(seeded, queueWatchRecommendation("task/worker-9"))).toMatchObject({
+      ok: false,
+      code: "invalid_target",
+    });
+  });
+});

internal-packages/dashboard-agent-contracts/src/watch.ts

@@ -40,7 +40,7 @@ export const WATCH_MAX_QUEUE_AGE_MINUTES = 24 * 60;
 const watchQueueNameSchema = z
   .string()
   .describe(
-    "The stored queue name, keeping the `task/` prefix for task queues (e.g. `task/send-receipt`) — not the display name."
+    "The queue name. A task's own queue is `task/<task id>`; a custom queue is its plain name. If unsure, pass the name as shown — the server resolves it."
   );
 
 /**