Skip to content

feat(scheduled-jobs): user-driven cron / one-shot via TaskScheduler + scheduled_jobs table #710

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
buremba merged 2 commits into from
May 14, 2026
Merged

feat(scheduled-jobs): user-driven cron / one-shot via TaskScheduler + scheduled_jobs table #710

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d9f08d2
feat(scheduled-jobs): user-driven cron / one-shot via TaskScheduler +…
buremba May 14, 2026
2f90b76
fix(scheduled-jobs): address codex review + CI typecheck
buremba May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 97 additions & 0 deletions db/migrations/20260514000000_scheduled_jobs.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
-- migrate:up

-- User-driven scheduled jobs.
--
-- Why a separate table:
-- * `runs` already holds *fired* / *pending-to-fire* rows via
-- scheduler.spawn(). Each scheduled_jobs row is the *definition* of a
-- recurring (or one-shot) schedule — its source of truth.
-- * The ticker (`scheduled-jobs-tick`) scans this table on cron, spawns
-- a runs row per firing via TaskScheduler.spawn, and advances
-- next_run_at from `cron`. If the tick or a firing fails, the next
-- tick re-reads the same row (next_run_at didn't move forward) and
-- retries. Self-healing.
-- * Attribution lives here: who scheduled it (user or agent), what run
-- was the trigger, what event was the trigger. Lets "why did the
-- system act?" become a single JOIN.
-- * Cascade-on-delete: when an agent is deleted, all its schedules
-- evaporate via the FK — no orphan wake-ups firing into the void.

CREATE TABLE public.scheduled_jobs (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
organization_id text NOT NULL REFERENCES public.organization(id) ON DELETE CASCADE,

-- What fires
action_type text NOT NULL, -- 'send_notification' | 'wake_agent' | ...
action_args jsonb NOT NULL, -- handler payload
cron text, -- null = one-shot; cron string = recurring
next_run_at timestamp with time zone NOT NULL,
last_fired_at timestamp with time zone,
last_fired_run_id bigint, -- the runs.id from the most recent firing
paused boolean NOT NULL DEFAULT false,

description text NOT NULL, -- human summary for the UI / audit

-- Attribution
created_by_user text, -- user that scheduled it (null when agent did)
created_by_agent text, -- agent that scheduled it (null when user did)
source_run_id bigint, -- runs.id that originated the scheduling, if any
source_event_id bigint, -- events.id that originated, if any
source_thread_id text, -- chat-thread context, if any

created_at timestamp with time zone NOT NULL DEFAULT now(),
updated_at timestamp with time zone NOT NULL DEFAULT now(),

CONSTRAINT scheduled_jobs_attribution_check CHECK (
created_by_user IS NOT NULL OR created_by_agent IS NOT NULL
)
);

-- Cascade: dropping an agent kills its scheduled jobs (so an agent's
-- wake-ups don't outlive the agent itself). Conditional so the migration
-- works on installs where the agents table doesn't exist yet (very
-- old) — every row already has organization_id which is the harder constraint.
DO $$
BEGIN
IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'agents' AND relkind = 'r') THEN
ALTER TABLE public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_agent_fkey
FOREIGN KEY (created_by_agent) REFERENCES public.agents(id) ON DELETE CASCADE;
Comment on lines +57 to +59
Copy link
Copy Markdown

@chatgpt-codex-connector chatgpt-codex-connector Bot May 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Cascade wake-ups from the target agent

This FK only covers created_by_agent, not the target agent stored in a wake_agent payload. When a user creates a wake-up for an agent, created_by_agent is null, so deleting that target agent leaves the schedule behind and the ticker can later enqueue a synthetic message for an agent that no longer exists. The cascade needs to reference the wake target (or the handler/ticker must drop rows whose target agent is gone), not just the scheduler identity.

Useful? React with 👍 / 👎.

END IF;
END$$;

DO $$
BEGIN
IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'runs' AND relkind = 'r') THEN
ALTER TABLE public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_source_run_fkey
FOREIGN KEY (source_run_id) REFERENCES public.runs(id) ON DELETE SET NULL;
END IF;
END$$;

DO $$
BEGIN
IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'events' AND relkind = 'r') THEN
ALTER TABLE public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_source_event_fkey
FOREIGN KEY (source_event_id) REFERENCES public.events(id) ON DELETE SET NULL;
END IF;
END$$;

-- Index: the ticker's hot read.
CREATE INDEX idx_scheduled_jobs_due
ON public.scheduled_jobs (next_run_at)
WHERE NOT paused;

-- Index: list per-agent / per-user.
CREATE INDEX idx_scheduled_jobs_org_agent
ON public.scheduled_jobs (organization_id, created_by_agent)
WHERE created_by_agent IS NOT NULL;

CREATE INDEX idx_scheduled_jobs_org_user
ON public.scheduled_jobs (organization_id, created_by_user)
WHERE created_by_user IS NOT NULL;

-- migrate:down

DROP TABLE IF EXISTS public.scheduled_jobs;
81 changes: 80 additions & 1 deletion db/schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1530,6 +1530,31 @@ CREATE SEQUENCE public.runs_id_seq

ALTER SEQUENCE public.runs_id_seq OWNED BY public.runs.id;

--
-- Name: scheduled_jobs; Type: TABLE; Schema: public; Owner: -
--

CREATE TABLE public.scheduled_jobs (
id uuid DEFAULT gen_random_uuid() NOT NULL,
organization_id text NOT NULL,
action_type text NOT NULL,
action_args jsonb NOT NULL,
cron text,
next_run_at timestamp with time zone NOT NULL,
last_fired_at timestamp with time zone,
last_fired_run_id bigint,
paused boolean DEFAULT false NOT NULL,
description text NOT NULL,
created_by_user text,
created_by_agent text,
source_run_id bigint,
source_event_id bigint,
source_thread_id text,
created_at timestamp with time zone DEFAULT now() NOT NULL,
updated_at timestamp with time zone DEFAULT now() NOT NULL,
CONSTRAINT scheduled_jobs_attribution_check CHECK (((created_by_user IS NOT NULL) OR (created_by_agent IS NOT NULL)))
);

--
-- Name: schema_migrations; Type: TABLE; Schema: public; Owner: -
--
Expand Down Expand Up @@ -2609,6 +2634,13 @@ ALTER TABLE ONLY public.rate_limits
ALTER TABLE ONLY public.runs
ADD CONSTRAINT runs_pkey PRIMARY KEY (id);

--
-- Name: scheduled_jobs scheduled_jobs_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--

ALTER TABLE ONLY public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_pkey PRIMARY KEY (id);

--
-- Name: schema_migrations schema_migrations_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
Expand Down Expand Up @@ -3614,6 +3646,24 @@ CREATE INDEX idx_runs_type ON public.runs USING btree (run_type);

CREATE INDEX idx_runs_watcher_id ON public.runs USING btree (watcher_id) WHERE (watcher_id IS NOT NULL);

--
-- Name: idx_scheduled_jobs_due; Type: INDEX; Schema: public; Owner: -
--

CREATE INDEX idx_scheduled_jobs_due ON public.scheduled_jobs USING btree (next_run_at) WHERE (NOT paused);

--
-- Name: idx_scheduled_jobs_org_agent; Type: INDEX; Schema: public; Owner: -
--

CREATE INDEX idx_scheduled_jobs_org_agent ON public.scheduled_jobs USING btree (organization_id, created_by_agent) WHERE (created_by_agent IS NOT NULL);

--
-- Name: idx_scheduled_jobs_org_user; Type: INDEX; Schema: public; Owner: -
--

CREATE INDEX idx_scheduled_jobs_org_user ON public.scheduled_jobs USING btree (organization_id, created_by_user) WHERE (created_by_user IS NOT NULL);

--
-- Name: idx_view_template_versions_resource; Type: INDEX; Schema: public; Owner: -
--
Expand Down Expand Up @@ -4724,6 +4774,34 @@ ALTER TABLE ONLY public.runs
ALTER TABLE ONLY public.runs
ADD CONSTRAINT runs_window_id_fkey FOREIGN KEY (window_id) REFERENCES public.watcher_windows(id) ON DELETE SET NULL;

--
-- Name: scheduled_jobs scheduled_jobs_agent_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--

ALTER TABLE ONLY public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_agent_fkey FOREIGN KEY (created_by_agent) REFERENCES public.agents(id) ON DELETE CASCADE;

--
-- Name: scheduled_jobs scheduled_jobs_organization_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--

ALTER TABLE ONLY public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_organization_id_fkey FOREIGN KEY (organization_id) REFERENCES public.organization(id) ON DELETE CASCADE;

--
-- Name: scheduled_jobs scheduled_jobs_source_event_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--

ALTER TABLE ONLY public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_source_event_fkey FOREIGN KEY (source_event_id) REFERENCES public.events(id) ON DELETE SET NULL;

--
-- Name: scheduled_jobs scheduled_jobs_source_run_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--

ALTER TABLE ONLY public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_source_run_fkey FOREIGN KEY (source_run_id) REFERENCES public.runs(id) ON DELETE SET NULL;

--
-- Name: session session_activeOrganizationId_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--
Expand Down Expand Up @@ -4869,4 +4947,5 @@ INSERT INTO public.schema_migrations (version) VALUES
('20260513000000'),
('20260513120000'),
('20260513150000'),
('20260513200000');
('20260513200000'),
('20260514000000');
55 changes: 55 additions & 0 deletions packages/server/src/db/embedded-schema-patches.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -436,4 +436,59 @@ export const EMBEDDED_SCHEMA_PATCHES: EmbeddedSchemaPatch[] = [
}
},
},
{
// Mirrors db/migrations/20260514000000_scheduled_jobs.sql.
id: 'scheduled-jobs',
apply: async (sql) => {
await sql.unsafe(`
CREATE TABLE IF NOT EXISTS public.scheduled_jobs (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
organization_id text NOT NULL REFERENCES public.organization(id) ON DELETE CASCADE,
action_type text NOT NULL,
action_args jsonb NOT NULL,
cron text,
next_run_at timestamp with time zone NOT NULL,
last_fired_at timestamp with time zone,
last_fired_run_id bigint,
paused boolean NOT NULL DEFAULT false,
description text NOT NULL,
created_by_user text,
created_by_agent text,
source_run_id bigint,
source_event_id bigint,
source_thread_id text,
created_at timestamp with time zone NOT NULL DEFAULT now(),
updated_at timestamp with time zone NOT NULL DEFAULT now(),
CONSTRAINT scheduled_jobs_attribution_check CHECK (
created_by_user IS NOT NULL OR created_by_agent IS NOT NULL
)
)
`);
await sql.unsafe(`
DO $$
BEGIN
IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'agents' AND relkind = 'r')
AND NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'scheduled_jobs_agent_fkey') THEN
ALTER TABLE public.scheduled_jobs
ADD CONSTRAINT scheduled_jobs_agent_fkey
FOREIGN KEY (created_by_agent) REFERENCES public.agents(id) ON DELETE CASCADE;
END IF;
END$$;
`);
await sql.unsafe(`
CREATE INDEX IF NOT EXISTS idx_scheduled_jobs_due
ON public.scheduled_jobs (next_run_at) WHERE NOT paused
`);
await sql.unsafe(`
CREATE INDEX IF NOT EXISTS idx_scheduled_jobs_org_agent
ON public.scheduled_jobs (organization_id, created_by_agent)
WHERE created_by_agent IS NOT NULL
`);
await sql.unsafe(`
CREATE INDEX IF NOT EXISTS idx_scheduled_jobs_org_user
ON public.scheduled_jobs (organization_id, created_by_user)
WHERE created_by_user IS NOT NULL
`);
},
},
Comment on lines +439 to +493
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot May 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Mirror the source FKs in the embedded patch.

The primary migration and db/schema.sql null out source_run_id/source_event_id on parent deletes, but the embedded patch never adds those constraints. PGlite installs can therefore keep dangling source references and drift from the main schema.

🔧 Proposed fix 
       await sql.unsafe(`
         DO $$
         BEGIN
           IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'agents' AND relkind = 'r')
              AND NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'scheduled_jobs_agent_fkey') THEN
             ALTER TABLE public.scheduled_jobs
               ADD CONSTRAINT scheduled_jobs_agent_fkey
               FOREIGN KEY (created_by_agent) REFERENCES public.agents(id) ON DELETE CASCADE;
           END IF;
         END$$;
       `);
+      await sql.unsafe(`
+        DO $$
+        BEGIN
+          IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'runs' AND relkind = 'r')
+             AND NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'scheduled_jobs_source_run_fkey') THEN
+            ALTER TABLE public.scheduled_jobs
+              ADD CONSTRAINT scheduled_jobs_source_run_fkey
+              FOREIGN KEY (source_run_id) REFERENCES public.runs(id) ON DELETE SET NULL;
+          END IF;
+        END$$;
+      `);
+      await sql.unsafe(`
+        DO $$
+        BEGIN
+          IF EXISTS (SELECT 1 FROM pg_class WHERE relname = 'events' AND relkind = 'r')
+             AND NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'scheduled_jobs_source_event_fkey') THEN
+            ALTER TABLE public.scheduled_jobs
+              ADD CONSTRAINT scheduled_jobs_source_event_fkey
+              FOREIGN KEY (source_event_id) REFERENCES public.events(id) ON DELETE SET NULL;
+          END IF;
+        END$$;
+      `);
       await sql.unsafe(`
         CREATE INDEX IF NOT EXISTS idx_scheduled_jobs_due
           ON public.scheduled_jobs (next_run_at) WHERE NOT paused
       `);
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/server/src/db/embedded-schema-patches.ts` around lines 439 - 493,
The embedded patch for id 'scheduled-jobs' creates the scheduled_jobs table but
omits the foreign key constraints for source_run_id and source_event_id,
allowing dangling references; update the apply handler for the 'scheduled-jobs'
patch to add the same FK constraints used in the primary migration/schema
(foreign keys on scheduled_jobs.source_run_id and scheduled_jobs.source_event_id
referencing their parent tables with ON DELETE SET NULL and appropriate
constraint names, e.g. scheduled_jobs_source_run_id_fkey and
scheduled_jobs_source_event_id_fkey), and add those ALTER TABLE / ADD CONSTRAINT
statements (guarded by the same pg_class/pg_constraint existence checks pattern
used for scheduled_jobs_agent_fkey) so PGlite installs mirror the main schema
behavior.

];
Loading
Loading