fix(core): harden event task updates

Author: dcramer

packages/junior/src/chat/event-tasks/ingest.ts

@@ -14,9 +14,9 @@ import {
 import type { EventTask, EventTaskMatch } from "@/chat/event-tasks/types";
 import type { ConversationWorkQueue } from "@/chat/task-execution/queue";
 
-function matchId(taskId: string, eventKey: string): string {
+function matchId(taskId: string, provider: string, eventKey: string): string {
   return `evmatch_${createHash("sha256")
-    .update(`${taskId}\0${eventKey}`)
+    .update(`${taskId}\0${provider}\0${eventKey}`)
     .digest("hex")
     .slice(0, 32)}`;
 }
@@ -68,7 +68,7 @@ export async function ingestEventTasks(
   const errors: unknown[] = [];
   for (const task of tasks) {
     try {
-      const id = matchId(task.id, event.eventKey);
+      const id = matchId(task.id, event.provider, event.eventKey);
       const pending: EventTaskMatch = {
         id,
         createdAtMs: nowMs,

packages/junior/src/chat/event-tasks/store.ts

@@ -4,7 +4,7 @@ import {
   slackActorSchema,
   type ResourceEvent,
 } from "@sentry/junior-plugin-api";
-import { and, asc, eq, ne } from "drizzle-orm";
+import { and, asc, eq, ne, sql } from "drizzle-orm";
 import { z } from "zod";
 import type { JuniorDatabase } from "@/db/db";
 import {
@@ -105,21 +105,55 @@ export async function createEventTask(
   return (await getEventTask(db, parsed.id)) ?? parsed;
 }
 
-/** Replace one existing event task. */
-export async function saveEventTask(
+/** Replace an active event task only when the caller read its current version. */
+export async function updateActiveEventTask(
   db: JuniorDatabase,
   task: EventTask,
-): Promise<void> {
+  expectedUpdatedAtMs: number,
+): Promise<EventTask | undefined> {
   const parsed = eventTaskSchema.parse(task);
-  await db
+  if (parsed.status !== "active") {
+    throw new Error("Event task update requires active status");
+  }
+  const rows = await db
     .update(juniorEventTasks)
     .set({
       provider: parsed.trigger.provider,
       resourceRef: parsed.trigger.resourceRef,
       status: parsed.status,
       task: parsed,
     })
-    .where(eq(juniorEventTasks.id, parsed.id));
+    .where(
+      and(
+        eq(juniorEventTasks.id, parsed.id),
+        eq(juniorEventTasks.status, "active"),
+        sql`${juniorEventTasks.task}->>'updatedAtMs' = ${String(expectedUpdatedAtMs)}`,
+      ),
+    )
+    .returning({ task: juniorEventTasks.task });
+  return rows[0] ? parseTask(rows[0].task) : undefined;
+}
+
+/** Delete an active event task without allowing a stale update to revive it. */
+export async function deleteActiveEventTask(
+  db: JuniorDatabase,
+  task: EventTask,
+): Promise<EventTask | undefined> {
+  const parsed = eventTaskSchema.parse(task);
+  if (parsed.status !== "deleted") {
+    throw new Error("Event task deletion requires deleted status");
+  }
+  const rows = await db
+    .update(juniorEventTasks)
+    .set({ status: parsed.status, task: parsed })
+    .where(
+      and(
+        eq(juniorEventTasks.id, parsed.id),
+        eq(juniorEventTasks.status, "active"),
+      ),
+    )
+    .returning({ task: juniorEventTasks.task });
+  return rows[0] ? parseTask(rows[0].task) : undefined;
 }
 
 /** List active event tasks in one Slack workspace. */

packages/junior/src/chat/prompt.ts

@@ -326,7 +326,7 @@ const TOOL_POLICY_RULES = [
   "- If a sandbox-backed tool reports that sandbox execution is unavailable, treat that as a blocker for local file/shell inspection; do not pretend host files were inspected.",
   "- For user-provided URLs, use `webFetch`; for discovery, use `webSearch` then fetch/read promising sources; for current time/date context, use `systemTime`.",
   "- When a tool result includes a subscribable resource, use resource-event subscriptions for high-signal provider changes that serve the user's current intent; do not create scheduled polling tasks for events the subscription can deliver. Use the suggested events when they fit and write a concise intent summary.",
-  "- Use event tasks when the user wants a durable instruction to dispatch whenever registered resource events occur. When the resource and events are known, create a directly requested event task without redundant confirmation.",
+  "- Use event tasks only when the user explicitly asks for an event task or durable automated work whenever registered resource events occur. Ordinary requests to watch a resource and report relevant updates use resource-event subscriptions. When an event task's resource and events are known, create it without redundant confirmation.",
   "- Event tasks use system credentials by default. If the user denies creator credential use, use system mode without asking. Conditional or tentative credential language is not authorization; when the task depends on creator credentials, ask before creating or updating it and enable creator mode only after explicit authorization for future use.",
   "- For code changes, debugging or root-cause analysis, broad refactors, and software architecture decisions, use `handoff` before substantive analysis only when it offers a profile that better matches the task. Do not switch merely because the task involves code.",
   "- Run `jr-rpc config get|set|unset|list` for provider defaults and `jr-rpc plugins list` for installed plugin introspection as standalone bash commands; do not chain them with `cd`, `&&`, pipes, or provider commands.",

packages/junior/src/chat/tools/event-tasks.ts

@@ -3,9 +3,10 @@ import { z } from "zod";
 import { getDb } from "@/chat/db";
 import {
   createEventTask,
+  deleteActiveEventTask,
   getEventTask,
   listEventTasksForTeam,
-  saveEventTask,
+  updateActiveEventTask,
 } from "@/chat/event-tasks/store";
 import type {
   EventTask,
@@ -22,11 +23,11 @@ const MAX_LISTED_TASKS = 50;
 
 const triggerSchema = z
   .object({
-    provider: z.string().min(1),
-    resourceRef: z.string().min(1),
-    resourceType: z.string().min(1),
-    label: z.string().min(1),
-    events: z.array(z.string().min(1)).min(1),
+    provider: z.string().trim().min(1),
+    resourceRef: z.string().trim().min(1),
+    resourceType: z.string().trim().min(1),
+    label: z.string().trim().min(1),
+    events: z.array(z.string().trim().min(1)).min(1),
   })
   .strict();
 
@@ -139,6 +140,10 @@ function cleanEvents(events: string[]): string[] {
   return clean;
 }
 
+function nextUpdatedAtMs(task: EventTask): number {
+  return Math.max(Date.now(), task.updatedAtMs + 1);
+}
+
 function compactTask(task: EventTask) {
   return {
     id: task.id,
@@ -174,10 +179,10 @@ export function createEventTaskTool(context: ToolRuntimeContext) {
     },
     executionMode: "sequential",
     description:
-      "Create a durable Junior task when the user asks for work on registered events from a subscribable provider resource. Copy the provider, resource ref, type, label, and supported event names from the provider tool result; do not ask for redundant confirmation when those details are known.",
+      "Create a durable reactive Junior task only when the user explicitly asks for an event task or automated work whenever registered provider events occur. Ordinary requests to watch a resource and report relevant updates use resource-event subscriptions instead. Copy the provider, resource ref, type, label, and supported event names from the provider tool result; do not ask for redundant confirmation when those details are known.",
     inputSchema: z
       .object({
-        task: z.string().min(1).max(4000),
+        task: z.string().trim().min(1).max(4000),
         trigger: triggerSchema,
         credentialMode: z
           .enum(["system", "creator"])
@@ -221,12 +226,12 @@ export function createEventTaskTool(context: ToolRuntimeContext) {
         destination,
         originalRequest: context.userText,
         status: "active",
-        task: { text: input.task.trim() },
+        task: { text: input.task },
         trigger: {
-          provider: input.trigger.provider.trim(),
-          resourceRef: input.trigger.resourceRef.trim(),
-          resourceType: input.trigger.resourceType.trim(),
-          label: input.trigger.label.trim(),
+          provider: input.trigger.provider,
+          resourceRef: input.trigger.resourceRef,
+          resourceType: input.trigger.resourceType,
+          label: input.trigger.label,
           events: cleanEvents(input.trigger.events),
         },
         updatedAtMs: nowMs,
@@ -283,7 +288,7 @@ export function createUpdateEventTaskTool(context: ToolRuntimeContext) {
     inputSchema: z
       .object({
         taskId: z.string().min(1),
-        task: z.string().min(1).max(4000).optional(),
+        task: z.string().trim().min(1).max(4000).optional(),
         trigger: triggerSchema.optional(),
         credentialMode: z
           .enum(["system", "creator"])
@@ -319,20 +324,29 @@ export function createUpdateEventTaskTool(context: ToolRuntimeContext) {
           changesExecution && !isCreator
             ? "system"
             : (input.credentialMode ?? current.credentialMode),
-        task: input.task ? { text: input.task.trim() } : current.task,
+        task: input.task ? { text: input.task } : current.task,
         trigger: input.trigger
           ? {
-              provider: input.trigger.provider.trim(),
-              resourceRef: input.trigger.resourceRef.trim(),
-              resourceType: input.trigger.resourceType.trim(),
-              label: input.trigger.label.trim(),
+              provider: input.trigger.provider,
+              resourceRef: input.trigger.resourceRef,
+              resourceType: input.trigger.resourceType,
+              label: input.trigger.label,
               events: cleanEvents(input.trigger.events),
             }
           : current.trigger,
-        updatedAtMs: Date.now(),
+        updatedAtMs: nextUpdatedAtMs(current),
       };
-      await saveEventTask(getDb(), next);
-      return success(next);
+      const saved = await updateActiveEventTask(
+        getDb(),
+        next,
+        current.updatedAtMs,
+      );
+      if (!saved) {
+        throw new ToolInputError(
+          "Event task changed while it was being updated. List event tasks and try again.",
+        );
+      }
+      return success(saved);
     },
   });
 }
@@ -355,10 +369,15 @@ export function createDeleteEventTaskTool(context: ToolRuntimeContext) {
       const next: EventTask = {
         ...current,
         status: "deleted",
-        updatedAtMs: Date.now(),
+        updatedAtMs: nextUpdatedAtMs(current),
       };
-      await saveEventTask(getDb(), next);
-      return success(next);
+      const deleted = await deleteActiveEventTask(getDb(), next);
+      if (!deleted) {
+        throw new ToolInputError(
+          "Event task was not found in the active Slack conversation.",
+        );
+      }
+      return success(deleted);
     },
   });
 }

packages/junior/tests/component/resource-events/resource-events.test.ts

@@ -3,8 +3,11 @@ import { createMemoryState } from "@chat-adapter/state-memory";
 import { githubPlugin } from "@sentry/junior-github";
 import type { StateAdapter } from "chat";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { eq } from "drizzle-orm";
 import { createApp, defineJuniorPlugins } from "@/app";
-import { closeDb } from "@/chat/db";
+import { getDispatchRecord } from "@/chat/agent-dispatch/store";
+import { closeDb, getDb } from "@/chat/db";
+import { createEventTask } from "@/chat/event-tasks/store";
 import {
   getConfigDefaults,
   setConfigDefaults,
@@ -15,6 +18,10 @@ import { setDashboardConversationLinkOptions } from "@/chat/slack/dashboard-link
 import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
 import { JUNIOR_THREAD_STATE_TTL_MS } from "@/chat/state/ttl";
 import { getConversationWorkState } from "@/chat/task-execution/store";
+import {
+  juniorEventTaskMatches,
+  juniorEventTasks,
+} from "@/db/schema/event-tasks";
 import { ingestResourceEvent } from "@/chat/resource-events/ingest";
 import {
   cancelResourceEventSubscription,
@@ -153,6 +160,7 @@ describe("resource event subscriptions", () => {
     const previousDashboardOptions =
       setDashboardConversationLinkOptions(undefined);
     setDashboardConversationLinkOptions(previousDashboardOptions);
+    let eventTaskId: string | undefined;
 
     try {
       process.env.GITHUB_WEBHOOK_SECRET = "test-secret";
@@ -166,6 +174,25 @@ describe("resource event subscriptions", () => {
         nowMs,
         state,
       });
+      eventTaskId = `evt_webhook_bridge_${nowMs}`;
+      await createEventTask(getDb(), {
+        id: eventTaskId,
+        conversationAccess: { audience: "channel", visibility: "public" },
+        createdAtMs: nowMs,
+        createdBy: { slackUserId: "U123" },
+        credentialMode: "system",
+        destination: SLACK_DESTINATION,
+        status: "active",
+        task: { text: "Summarize the reviewer comment." },
+        trigger: {
+          events: ["comment.created"],
+          label: "GitHub PR getsentry/junior#691",
+          provider: "github",
+          resourceRef: "github:pull_request:getsentry/junior#691",
+          resourceType: "pull_request",
+        },
+        updatedAtMs: nowMs,
+      });
       const body = JSON.stringify({
         action: "created",
         repository: { full_name: "getsentry/junior" },
@@ -205,12 +232,39 @@ describe("resource event subscriptions", () => {
       );
 
       expect(response.status).toBe(202);
-      expect(queue.sentRecords()).toEqual([
-        {
-          conversationId: CONVERSATION_ID,
-          idempotencyKey: `resource-event:${subscription.id}:github:delivery-bridge:comment.created`,
+      expect(queue.sentRecords()).toEqual(
+        expect.arrayContaining([
+          {
+            conversationId: expect.stringMatching(/^agent-dispatch:/),
+            idempotencyKey: expect.stringMatching(/^agent-dispatch:/),
+          },
+          {
+            conversationId: CONVERSATION_ID,
+            idempotencyKey: `resource-event:${subscription.id}:github:delivery-bridge:comment.created`,
+          },
+        ]),
+      );
+      const dispatchRecord = queue
+        .sentRecords()
+        .find(({ conversationId }) =>
+          conversationId.startsWith("agent-dispatch:"),
+        );
+      expect(dispatchRecord).toBeDefined();
+      const dispatch = dispatchRecord
+        ? await getDispatchRecord(
+            dispatchRecord.conversationId.replace(/^agent-dispatch:/, ""),
+          )
+        : undefined;
+      expect(dispatch).toMatchObject({
+        input: expect.stringContaining("Summarize the reviewer comment."),
+        metadata: {
+          eventType: "comment.created",
+          provider: "github",
+          resourceRef: "github:pull_request:getsentry/junior#691",
+          taskId: eventTaskId,
         },
-      ]);
+        plugin: "junior",
+      });
       const work = await getConversationWorkState({
         conversationId: CONVERSATION_ID,
         state,
@@ -219,6 +273,14 @@ describe("resource event subscriptions", () => {
         "please add regression coverage",
       );
     } finally {
+      if (eventTaskId) {
+        await getDb()
+          .delete(juniorEventTaskMatches)
+          .where(eq(juniorEventTaskMatches.taskId, eventTaskId));
+        await getDb()
+          .delete(juniorEventTasks)
+          .where(eq(juniorEventTasks.id, eventTaskId));
+      }
       setPlugins(previousPlugins);
       pluginCatalogRuntime.setConfig(previousPluginCatalogConfig);
       setConfigDefaults(previousConfigDefaults);