feat(schedules): extend schema for ownership and richer trigger audit (#16166)

Co-authored-by: Mastra Code (anthropic/claude-opus-4-7) <noreply@mastra.ai>

Author: abhiaiyer91

.changeset/heartbeats-prep-schedules-schema.md

@@ -0,0 +1,18 @@
+---
+'@mastra/core': minor
+'@mastra/libsql': minor
+'@mastra/pg': minor
+'@mastra/mongodb': minor
+'@mastra/server': minor
+'@mastra/client-js': minor
+'mastra': patch
+---
+
+Extend the schedules storage schema to support owned schedules and richer trigger audit. This is a breaking schema change to `mastra_schedules` and `mastra_schedule_triggers`; scheduled workflows are still in alpha so no compat shim is provided.
+
+- `Schedule` gains optional `ownerType` / `ownerId` so a schedule row can be attributed to an owning subsystem (e.g. an agent that owns a heartbeat schedule). Workflow schedules leave both fields unset.
+- `ScheduleTrigger.status` is renamed to `outcome` and the type is widened to `ScheduleTriggerOutcome` so future outcome values can be added without another rename.
+- `ScheduleTrigger` gains a stable `id` primary key and new `triggerKind`, `parentTriggerId`, and `metadata` fields. `triggerKind` distinguishes `schedule-fire` rows from later `queue-drain` rows (used by upcoming heartbeat work); `parentTriggerId` links related rows; `metadata` carries outcome-specific context.
+- The libsql, pg, and mongodb adapters all add the new columns/indexes. Their `@mastra/core` peer dependency is tightened to `>=1.32.0-0 <2.0.0-0` so installing a new storage adapter against an older core (or vice-versa) surfaces a peer-dependency warning at install time instead of silently writing/reading the wrong field.
+- Scheduler producer, server schemas/handler, and client SDK types are updated to use the new fields. The `triggers` response on `GET /api/schedules/:id/triggers` now returns `outcome` instead of `status`.
+- The bundled Studio (Mastra CLI) is updated to read `outcome` so the schedule detail page keeps polling and rendering publish-failure rows correctly.

client-sdks/client-js/src/route-types.generated.ts

@@ -74324,6 +74324,8 @@ export interface GetBackgroundTasksBackgroundTaskId_RouteContract {
 export type GetSchedules_QueryParams = {
   workflowId?: string | undefined;
   status?: ('active' | 'paused') | undefined;
+  ownerType?: string | undefined;
+  ownerId?: string | undefined;
 };
 
 export type GetSchedules_Response = {
@@ -74370,6 +74372,8 @@ export type GetSchedules_Response = {
           [key: string]: unknown;
         }
       | undefined;
+    ownerType?: string | undefined;
+    ownerId?: string | undefined;
     createdAt: number;
     updatedAt: number;
   }[];
@@ -74444,6 +74448,8 @@ export type GetSchedulesScheduleId_Response = {
         [key: string]: unknown;
       }
     | undefined;
+  ownerType?: string | undefined;
+  ownerId?: string | undefined;
   createdAt: number;
   updatedAt: number;
 };
@@ -74478,12 +74484,30 @@ export type GetSchedulesScheduleIdTriggers_QueryParams = {
 
 export type GetSchedulesScheduleIdTriggers_Response = {
   triggers: {
+    id?: string | undefined;
     scheduleId: string;
-    runId: string;
+    runId: string | null;
     scheduledFireAt: number;
     actualFireAt: number;
-    status: 'published' | 'failed';
+    outcome:
+      | 'published'
+      | 'failed'
+      | 'skipped'
+      | 'acked'
+      | 'alerted'
+      | 'deferred'
+      | 'appended-from-queue'
+      | 'dropped-stale'
+      | 'dropped-superseded'
+      | 'dropped-busy';
     error?: string | undefined;
+    triggerKind?: ('schedule-fire' | 'queue-drain') | undefined;
+    parentTriggerId?: string | undefined;
+    metadata?:
+      | {
+          [key: string]: unknown;
+        }
+      | undefined;
     run?:
       | {
           status:
@@ -74577,6 +74601,8 @@ export type PostSchedulesScheduleIdPause_Response = {
         [key: string]: unknown;
       }
     | undefined;
+  ownerType?: string | undefined;
+  ownerId?: string | undefined;
   createdAt: number;
   updatedAt: number;
 };
@@ -74646,6 +74672,8 @@ export type PostSchedulesScheduleIdResume_Response = {
         [key: string]: unknown;
       }
     | undefined;
+  ownerType?: string | undefined;
+  ownerId?: string | undefined;
   createdAt: number;
   updatedAt: number;
 };

client-sdks/client-js/src/types.ts

@@ -2554,26 +2554,44 @@ export interface ScheduleResponse {
   lastRunId?: string;
   lastRun?: ScheduleRunSummary;
   metadata?: Record<string, unknown>;
+  ownerType?: string;
+  ownerId?: string;
   createdAt: number;
   updatedAt: number;
 }
 
-export type ScheduleTriggerStatus = 'published' | 'failed';
+export type ScheduleTriggerOutcome =
+  | 'published'
+  | 'failed'
+  | 'skipped'
+  | 'acked'
+  | 'alerted'
+  | 'deferred'
+  | 'appended-from-queue'
+  | 'dropped-stale'
+  | 'dropped-superseded'
+  | 'dropped-busy';
+
+export type ScheduleTriggerKind = 'schedule-fire' | 'queue-drain';
 
 export interface ScheduleTriggerResponse {
+  id?: string;
   scheduleId: string;
-  runId: string;
+  runId: string | null;
   scheduledFireAt: number;
   actualFireAt: number;
-  status: ScheduleTriggerStatus;
+  outcome: ScheduleTriggerOutcome;
   error?: string;
+  triggerKind?: ScheduleTriggerKind;
+  parentTriggerId?: string;
+  metadata?: Record<string, unknown>;
   run?: ScheduleRunSummary;
 }
 
-export interface ListSchedulesParams {
+export type ListSchedulesParams = {
   workflowId?: string;
   status?: ScheduleStatus;
-}
+} & ({ ownerType?: undefined; ownerId?: undefined } | { ownerType: string; ownerId?: string });
 
 export interface ListSchedulesResponse {
   schedules: ScheduleResponse[];

packages/core/src/storage/constants.ts

@@ -586,14 +586,20 @@ export const TABLE_SCHEMAS: Record<TABLE_NAMES, Record<string, StorageColumn>> =
     created_at: { type: 'bigint', nullable: false },
     updated_at: { type: 'bigint', nullable: false },
     metadata: { type: 'jsonb', nullable: true },
+    owner_type: { type: 'text', nullable: true },
+    owner_id: { type: 'text', nullable: true },
   },
   [TABLE_SCHEDULE_TRIGGERS]: {
+    id: { type: 'text', nullable: false, primaryKey: true },
     schedule_id: { type: 'text', nullable: false },
-    run_id: { type: 'text', nullable: false, primaryKey: true },
+    run_id: { type: 'text', nullable: true },
     scheduled_fire_at: { type: 'bigint', nullable: false },
     actual_fire_at: { type: 'bigint', nullable: false },
-    status: { type: 'text', nullable: false },
+    outcome: { type: 'text', nullable: false },
     error: { type: 'text', nullable: true },
+    trigger_kind: { type: 'text', nullable: false },
+    parent_trigger_id: { type: 'text', nullable: true },
+    metadata: { type: 'jsonb', nullable: true },
   },
   [TABLE_CHANNEL_INSTALLATIONS]: {
     id: { type: 'text', nullable: false, primaryKey: true },

packages/core/src/storage/domains/schedules/__tests__/inmemory.test.ts

@@ -25,11 +25,13 @@ function makeSchedule(overrides: Partial<Schedule> = {}): Schedule {
 
 function makeTrigger(overrides: Partial<ScheduleTrigger> = {}): ScheduleTrigger {
   return {
+    id: overrides.id ?? `tr_${crypto.randomUUID()}`,
     scheduleId: overrides.scheduleId ?? 'sched_1',
     runId: overrides.runId ?? 'run_1',
     scheduledFireAt: overrides.scheduledFireAt ?? Date.now(),
     actualFireAt: overrides.actualFireAt ?? Date.now(),
-    status: overrides.status ?? 'published',
+    outcome: overrides.outcome ?? 'published',
+    triggerKind: overrides.triggerKind ?? 'schedule-fire',
     error: overrides.error,
   };
 }

packages/core/src/storage/domains/schedules/base.ts

@@ -17,6 +17,16 @@ export type ScheduleTarget = {
 /** Lifecycle status of a schedule row. */
 export type ScheduleStatus = 'active' | 'paused';
 
+/**
+ * Polymorphic owner of a schedule. Workflow schedules created via
+ * `createWorkflow({ schedule })` leave both fields null. Heartbeat
+ * schedules created via `agent.setHeartbeat(...)` set
+ * `ownerType: 'agent'` and `ownerId` to the agent id. Future schedule
+ * types (tenant-owned, workflow-owned, etc.) can use the same shape
+ * without a migration.
+ */
+export type ScheduleOwnerType = 'agent' | (string & {});
+
 /**
  * A persisted schedule.
  *
@@ -36,25 +46,66 @@ export type Schedule = {
   createdAt: number;
   updatedAt: number;
   metadata?: Record<string, unknown>;
+  /** Optional owner classification (e.g. 'agent' for heartbeats). */
+  ownerType?: ScheduleOwnerType;
+  /** Optional owner identifier paired with `ownerType`. */
+  ownerId?: string;
 };
 
-/** Status of an individual schedule trigger attempt. */
-export type ScheduleTriggerStatus = 'published' | 'failed';
+/**
+ * Outcome of an individual schedule trigger attempt.
+ *
+ * `published` and `failed` cover scheduler tick fires. Heartbeat
+ * outcomes (`acked`, `alerted`, `deferred`, `appended-from-queue`,
+ * `dropped-stale`, `dropped-superseded`, `dropped-busy`, `skipped`)
+ * are recorded by the heartbeat workflow. `failed` covers any
+ * unhandled error during a fire or drain.
+ */
+export type ScheduleTriggerOutcome =
+  | 'published'
+  | 'failed'
+  | 'skipped'
+  | 'acked'
+  | 'alerted'
+  | 'deferred'
+  | 'appended-from-queue'
+  | 'dropped-stale'
+  | 'dropped-superseded'
+  | 'dropped-busy';
+
+/**
+ * Distinguishes a tick-loop schedule fire from a deferred drain event.
+ * Drain rows reference the original fire via `parentTriggerId`.
+ */
+export type ScheduleTriggerKind = 'schedule-fire' | 'queue-drain';
 
 /** Audit record produced for each trigger attempt. */
 export type ScheduleTrigger = {
+  /** Stable trigger row id. Generated by storage when omitted on write. */
+  id?: string;
   scheduleId: string;
-  runId: string;
+  /** May be null for drain rows that have no associated workflow run. */
+  runId: string | null;
   scheduledFireAt: number;
   actualFireAt: number;
-  status: ScheduleTriggerStatus;
+  outcome: ScheduleTriggerOutcome;
   error?: string;
+  /** Defaults to `'schedule-fire'` when omitted. */
+  triggerKind?: ScheduleTriggerKind;
+  /** Pointer back to the originating fire row when `triggerKind === 'queue-drain'`. */
+  parentTriggerId?: string;
+  /** Outcome-specific context (alert text, append message id, queue age, etc.). */
+  metadata?: Record<string, unknown>;
 };
 
 /** Filter options for listing schedules. */
 export type ScheduleFilter = {
   status?: ScheduleStatus;
   workflowId?: string;
+  /** `null` matches schedules with no owner (e.g. workflow-only schedules). */
+  ownerType?: ScheduleOwnerType | null;
+  /** `null` matches schedules with no owner id. */
+  ownerId?: string | null;
 };
 
 /** Filter / pagination options for listing trigger history. */
@@ -68,7 +119,7 @@ export type ScheduleTriggerListOptions = {
 
 /** Fields that can be patched via {@link SchedulesStorage.updateSchedule}. */
 export type ScheduleUpdate = Partial<
-  Pick<Schedule, 'cron' | 'timezone' | 'status' | 'nextFireAt' | 'metadata' | 'target'>
+  Pick<Schedule, 'cron' | 'timezone' | 'status' | 'nextFireAt' | 'metadata' | 'target' | 'ownerType' | 'ownerId'>
 >;
 
 /**

packages/core/src/storage/domains/schedules/inmemory.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from 'node:crypto';
 import type { InMemoryDB } from '../inmemory-db';
 import type { Schedule, ScheduleFilter, ScheduleTrigger, ScheduleTriggerListOptions, ScheduleUpdate } from './base';
 import { SchedulesStorage } from './base';
@@ -41,6 +42,12 @@ export class InMemorySchedulesStorage extends SchedulesStorage {
     if (filter?.workflowId) {
       rows = rows.filter(r => r.target.type === 'workflow' && r.target.workflowId === filter.workflowId);
     }
+    if (filter?.ownerType !== undefined) {
+      rows = rows.filter(r => (r.ownerType ?? null) === filter.ownerType);
+    }
+    if (filter?.ownerId !== undefined) {
+      rows = rows.filter(r => (r.ownerId ?? null) === filter.ownerId);
+    }
     rows.sort((a, b) => a.createdAt - b.createdAt);
     return rows.map(clone);
   }
@@ -106,7 +113,12 @@ export class InMemorySchedulesStorage extends SchedulesStorage {
   }
 
   async recordTrigger(trigger: ScheduleTrigger): Promise<void> {
-    this.db.scheduleTriggers.push(clone(trigger));
+    const stored: ScheduleTrigger = {
+      ...trigger,
+      id: trigger.id ?? randomUUID(),
+      triggerKind: trigger.triggerKind ?? 'schedule-fire',
+    };
+    this.db.scheduleTriggers.push(clone(stored));
   }
 
   async listTriggers(scheduleId: string, opts?: ScheduleTriggerListOptions): Promise<ScheduleTrigger[]> {

packages/core/src/workflows/scheduler/scheduler.test.ts

@@ -63,7 +63,7 @@ describe('WorkflowScheduler', () => {
 
     const triggers = await store.listTriggers(created.id);
     expect(triggers).toHaveLength(1);
-    expect(triggers[0]!.status).toBe('published');
+    expect(triggers[0]!.outcome).toBe('published');
   });
 
   it('skips paused schedules', async () => {
@@ -163,7 +163,7 @@ describe('WorkflowScheduler', () => {
 
     const triggers = await store.listTriggers('sched-fail');
     expect(triggers).toHaveLength(1);
-    expect(triggers[0]!.status).toBe('failed');
+    expect(triggers[0]!.outcome).toBe('failed');
     expect(triggers[0]!.error).toBe('boom');
     expect(onError).toHaveBeenCalledOnce();
     expect(onError.mock.calls[0]![1]).toEqual({ scheduleId: 'sched-fail' });

packages/core/src/workflows/scheduler/scheduler.ts

@@ -193,7 +193,7 @@ export class WorkflowScheduler extends MastraBase {
       return;
     }
 
-    let triggerStatus: ScheduleTrigger['status'] = 'published';
+    let triggerStatus: ScheduleTrigger['outcome'] = 'published';
     let triggerError: string | undefined;
 
     try {
@@ -215,8 +215,9 @@ export class WorkflowScheduler extends MastraBase {
         runId,
         scheduledFireAt: schedule.nextFireAt,
         actualFireAt,
-        status: triggerStatus,
+        outcome: triggerStatus,
         error: triggerError,
+        triggerKind: 'schedule-fire',
       });
     } catch (err) {
       this.logger.error('Failed to record schedule trigger', {

packages/playground/src/domains/schedules/components/schedule-triggers-list.tsx

@@ -83,7 +83,7 @@ export function ScheduleTriggersList({
         const driftMs = t.actualFireAt - t.scheduledFireAt;
         const driftValue = formatDriftValue(driftMs);
         const startedTooltip = `Scheduled ${formatScheduleTimestamp(t.scheduledFireAt)} — published ${formatScheduleTimestamp(t.actualFireAt)} (drift ${driftValue})`;
-        const isPublishFailure = t.status === 'failed';
+        const isPublishFailure = t.outcome === 'failed';
         const errorMessage = isPublishFailure ? t.error : t.run?.error;
         const absDrift = Math.abs(driftMs);
         const showDriftWarning = !isPublishFailure && absDrift > DRIFT_WARN_MIN_MS && absDrift <= DRIFT_WARN_MAX_MS;