ref(scheduler): Trim scheduler SQL storage contract

packages/junior-scheduler/migrations/0001_scheduler.sql

@@ -5,41 +5,26 @@ CREATE TABLE IF NOT EXISTS junior_scheduler_tasks (
   next_run_at_ms BIGINT,
   run_now_at_ms BIGINT,
   created_at_ms BIGINT NOT NULL,
-  updated_at_ms BIGINT NOT NULL,
-  version INTEGER NOT NULL,
-  destination JSONB NOT NULL,
-  created_by JSONB NOT NULL,
-  conversation_access JSONB,
-  credential_subject JSONB,
-  execution_actor JSONB,
-  last_run_at_ms BIGINT,
-  original_request TEXT,
-  schedule JSONB NOT NULL,
-  status_reason TEXT,
-  task JSONB NOT NULL,
   record JSONB NOT NULL
 );
 
 CREATE INDEX IF NOT EXISTS junior_scheduler_tasks_team_status_idx
-  ON junior_scheduler_tasks (team_id, status, created_at_ms);
+  ON junior_scheduler_tasks (team_id, created_at_ms, id)
+  WHERE status <> 'deleted';
 
-CREATE INDEX IF NOT EXISTS junior_scheduler_tasks_due_idx
-  ON junior_scheduler_tasks (status, run_now_at_ms, next_run_at_ms);
+CREATE INDEX IF NOT EXISTS junior_scheduler_tasks_run_now_due_idx
+  ON junior_scheduler_tasks (run_now_at_ms, created_at_ms, id)
+  WHERE status = 'active' AND run_now_at_ms IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS junior_scheduler_tasks_next_run_due_idx
+  ON junior_scheduler_tasks (next_run_at_ms, created_at_ms, id)
+  WHERE status = 'active' AND next_run_at_ms IS NOT NULL;
 
 CREATE TABLE IF NOT EXISTS junior_scheduler_runs (
   id TEXT PRIMARY KEY,
   task_id TEXT NOT NULL,
   status TEXT NOT NULL,
-  claimed_at_ms BIGINT NOT NULL,
   scheduled_for_ms BIGINT NOT NULL,
-  started_at_ms BIGINT,
-  completed_at_ms BIGINT,
-  dispatch_id TEXT,
-  error_message TEXT,
-  idempotency_key TEXT NOT NULL,
-  result_message_ts TEXT,
-  task_version INTEGER NOT NULL,
-  attempt INTEGER NOT NULL,
   record JSONB NOT NULL
 );
 

packages/junior-scheduler/src/db/schema.ts

@@ -1,11 +1,5 @@
-import {
-  bigint,
-  index,
-  integer,
-  jsonb,
-  pgTable,
-  text,
-} from "drizzle-orm/pg-core";
+import { sql } from "drizzle-orm";
+import { bigint, index, jsonb, pgTable, text } from "drizzle-orm/pg-core";
 import type { ScheduledRun, ScheduledTask } from "../types";
 
 export const juniorSchedulerTasks = pgTable(
@@ -17,31 +11,22 @@ export const juniorSchedulerTasks = pgTable(
     nextRunAtMs: bigint("next_run_at_ms", { mode: "number" }),
     runNowAtMs: bigint("run_now_at_ms", { mode: "number" }),
     createdAtMs: bigint("created_at_ms", { mode: "number" }).notNull(),
-    updatedAtMs: bigint("updated_at_ms", { mode: "number" }).notNull(),
-    version: integer("version").notNull(),
-    destination: jsonb("destination").notNull(),
-    createdBy: jsonb("created_by").notNull(),
-    conversationAccess: jsonb("conversation_access"),
-    credentialSubject: jsonb("credential_subject"),
-    executionActor: jsonb("execution_actor"),
-    lastRunAtMs: bigint("last_run_at_ms", { mode: "number" }),
-    originalRequest: text("original_request"),
-    schedule: jsonb("schedule").notNull(),
-    statusReason: text("status_reason"),
-    task: jsonb("task").notNull(),
     record: jsonb("record").$type<ScheduledTask>().notNull(),
   },
   (table) => [
-    index("junior_scheduler_tasks_team_status_idx").on(
-      table.teamId,
-      table.status,
-      table.createdAtMs,
-    ),
-    index("junior_scheduler_tasks_due_idx").on(
-      table.status,
-      table.runNowAtMs,
-      table.nextRunAtMs,
-    ),
+    index("junior_scheduler_tasks_team_status_idx")
+      .on(table.teamId, table.createdAtMs, table.id)
+      .where(sql`${table.status} <> 'deleted'`),
+    index("junior_scheduler_tasks_run_now_due_idx")
+      .on(table.runNowAtMs, table.createdAtMs, table.id)
+      .where(
+        sql`${table.status} = 'active' AND ${table.runNowAtMs} IS NOT NULL`,
+      ),
+    index("junior_scheduler_tasks_next_run_due_idx")
+      .on(table.nextRunAtMs, table.createdAtMs, table.id)
+      .where(
+        sql`${table.status} = 'active' AND ${table.nextRunAtMs} IS NOT NULL`,
+      ),
   ],
 );
 
@@ -51,16 +36,7 @@ export const juniorSchedulerRuns = pgTable(
     id: text("id").primaryKey(),
     taskId: text("task_id").notNull(),
     status: text("status").notNull(),
-    claimedAtMs: bigint("claimed_at_ms", { mode: "number" }).notNull(),
     scheduledForMs: bigint("scheduled_for_ms", { mode: "number" }).notNull(),
-    startedAtMs: bigint("started_at_ms", { mode: "number" }),
-    completedAtMs: bigint("completed_at_ms", { mode: "number" }),
-    dispatchId: text("dispatch_id"),
-    errorMessage: text("error_message"),
-    idempotencyKey: text("idempotency_key").notNull(),
-    resultMessageTs: text("result_message_ts"),
-    taskVersion: integer("task_version").notNull(),
-    attempt: integer("attempt").notNull(),
     record: jsonb("record").$type<ScheduledRun>().notNull(),
   },
   (table) => [

packages/junior-scheduler/src/prompt.ts

@@ -37,6 +37,8 @@ export function buildScheduledTaskRunPrompt(args: {
   const { run, task } = args;
   const destination = task.destination;
   const creator = sanitizeScheduledTaskPrincipal(task.createdBy);
+  // Older retained scheduler state predated executionActor; new tasks always
+  // store it explicitly as part of the task contract.
   const executionActor = task.executionActor ?? SCHEDULED_TASK_SYSTEM_ACTOR;
   if (!task.task.text?.trim()) {
     throw new Error("Scheduled task text is required");
@@ -55,7 +57,6 @@ export function buildScheduledTaskRunPrompt(args: {
     "",
     "<run-context>",
     `- run_id: ${escapeXml(run.id)}`,
-    `- task_version: ${run.taskVersion}`,
     `- scheduled_for: ${new Date(run.scheduledForMs).toISOString()}`,
     `- running_at: ${new Date(args.nowMs).toISOString()}`,
     `- schedule: ${escapeXml(task.schedule.description)}`,

packages/junior-scheduler/src/schedule-tools.ts

@@ -213,7 +213,6 @@ function compactTask(task: ScheduledTask): Record<string, unknown> {
     run_now_at: task.runNowAtMs
       ? new Date(task.runNowAtMs).toISOString()
       : null,
-    version: task.version,
   };
 }
 
@@ -425,7 +424,6 @@ export function createSlackScheduleCreateTaskTool(
         task: {
           text: input.task,
         },
-        version: 1,
       };
 
       await schedulerStore(context).saveTask(task);
@@ -572,7 +570,6 @@ export function createSlackScheduleUpdateTaskTool(
           recurrence,
         },
         task: input.task ? { text: input.task } : lookup.task,
-        version: lookup.version + 1,
       };
 
       await schedulerStore(context).saveTask(next);
@@ -608,7 +605,6 @@ export function createSlackScheduleDeleteTaskTool(
         status: "deleted",
         nextRunAtMs: undefined,
         runNowAtMs: undefined,
-        version: lookup.version + 1,
       };
 
       await schedulerStore(context).saveTask(next);
@@ -648,7 +644,6 @@ export function createSlackScheduleRunTaskNowTool(
         ...lookup,
         updatedAtMs: nowMs,
         runNowAtMs: nowMs,
-        version: lookup.version + 1,
       };
 
       await schedulerStore(context).saveTask(next);