From 4708b2571b226d4872d5c5939cac6465115696e2 Mon Sep 17 00:00:00 2001 From: Michael Buluma <1452922+buluma@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:52:06 +0300 Subject: [PATCH 01/13] docs: add thread-scheduling automations implementation plan Design doc for a scheduled-task system that auto-starts turns on threads at configured intervals/cron times, modeled after Codex Automations. Covers the CQRS design (schedule as an optional field on the thread, not a new aggregate), server-side timers via a ScheduleReactor, and the phased rollout across contracts/server/client-runtime/web. --- .plans/automations.md | 284 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 284 insertions(+) create mode 100644 .plans/automations.md diff --git a/.plans/automations.md b/.plans/automations.md new file mode 100644 index 00000000..57105208 --- /dev/null +++ b/.plans/automations.md @@ -0,0 +1,284 @@ +# Helm Code Automations — Implementation Plan + +## Overview + +Build a scheduled task system that auto-starts turns on threads at configured +intervals/cron times, modeled after Codex Automations. The design follows Helm +Code's existing CQRS architecture: new commands, events, projector fields, and a +`ScheduleReactor` that maintains an in-memory timer heap. + +## Design Decisions + +**Thread-scoped, not project-scoped.** Schedules attach to threads (like +snooze), not projects. A schedule is "send this prompt to this thread on a +cadence." + +**Prompt-driven.** Each scheduled run sends a user message to the thread and +starts a turn — same as a manual `thread.turn.start`. The agent picks up context +from the existing thread history (same-thread model). + +**No new aggregate.** Schedules are optional fields on +`OrchestrationThread`/`OrchestrationThreadShell` (like `snoozedUntil`), not a +separate aggregate. This keeps the event count low and avoids a new subscription +stream. + +**Server-side timers, not client-side.** Unlike snooze (which clients poll), +schedules fire from the server's `ScheduleReactor`. This ensures they fire even +when no client is connected. + +--- + +## Phase 1: Contracts & Schema + +### 1a. New schemas in `packages/contracts/src/orchestration.ts` + +```typescript +// Schedule configuration — persisted on the thread +const ThreadSchedule = Schema.Struct({ + enabled: Schema.Boolean, + // ISO 8601 RRULE string (e.g., "FREQ=DAILY;BYHOUR=9;BYMINUTE=0") + // or null for one-shot interval + cron: Schema.NullOr(TrimmedNonEmptyString), + // Interval in milliseconds (for simple "every N minutes" schedules) + intervalMs: Schema.NullOr(Schema.Number), + // The prompt to send on each scheduled run + prompt: TrimmedNonEmptyString, + // When the next run should fire + nextRunAt: IsoDateTime, + // When the schedule was created + createdAt: IsoDateTime, + // Optional: model override for scheduled runs + modelSelection: Schema.optional(ModelSelection), +}); +``` + +### 1b. New commands (add to `DispatchableClientOrchestrationCommand`) + +- `thread.schedule.create` — Create/update a schedule on a thread +- `thread.schedule.cancel` — Remove a schedule from a thread + +### 1c. New event types (add to `OrchestrationEventType`) + +- `"thread.scheduled"` — Schedule was created/updated +- `"thread.unscheduled"` — Schedule was cancelled + +### 1d. Thread & Shell schema updates + +Add optional fields to `OrchestrationThread` and `OrchestrationThreadShell`: + +```typescript +schedule: Schema.optional(Schema.NullOr(ThreadSchedule)), +``` + +### Files to modify: + +- `packages/contracts/src/orchestration.ts` — schemas, unions, payloads + +--- + +## Phase 2: Server Backend + +### 2a. Decider (`apps/server/src/orchestration/decider.ts`) + +Add cases for the two new commands: + +- **`thread.schedule.create`**: Validate thread exists, not archived. Validate + schedule config (at least one of `cron` or `intervalMs`; `prompt` non-empty; + `nextRunAt` in the future). Emit `thread.scheduled` event. +- **`thread.schedule.cancel`**: Validate thread exists. Emit + `thread.unscheduled` event. Idempotent (cancel of no-schedule is a no-op). + +### 2b. Projector (`apps/server/src/orchestration/projector.ts`) + +Add cases for the two new events: + +- **`thread.scheduled`**: Set `thread.schedule = payload.schedule` on the + matching thread. +- **`thread.unscheduled`**: Set `thread.schedule = null`. + +### 2c. DB Migration (`apps/server/src/persistence/Migrations/043_ProjectionThreadsSchedule.ts`) + +```sql +ALTER TABLE projection_threads ADD COLUMN schedule TEXT; +``` + +Follows the exact pattern of migration 034 (snoozed). + +### 2d. ScheduleReactor (`apps/server/src/orchestration/Services/ScheduleReactor.ts`) + +A new reactor that: + +1. **On startup**: Reads all threads with non-null `schedule` from the + projection, loads them into an in-memory timer heap (sorted by `nextRunAt`). +2. **On `thread.scheduled` event**: Add/update the entry in the heap. +3. **On `thread.unscheduled` event**: Remove from the heap. +4. **On timer fire**: For the due thread: + - Generate a user message ID (UUID) + - Dispatch `thread.turn.start` command with the schedule's prompt as the + message text + - Compute the next `nextRunAt` based on `cron`/`intervalMs` and update via + `thread.schedule.create` (re-emission) + - If the thread has an active session, skip this fire and reschedule +5. **On shutdown**: Clear all timers. + +Implementation uses `Effect.sleep` + `Effect.fork` for timers (no external cron +library). Each scheduled thread gets its own fiber that sleeps until +`nextRunAt`, then fires and reschedules. + +### Files to create: + +- `apps/server/src/orchestration/Services/ScheduleReactor.ts` +- `apps/server/src/orchestration/Layers/ScheduleReactor.ts` +- `apps/server/src/persistence/Migrations/043_ProjectionThreadsSchedule.ts` + +### Files to modify: + +- `apps/server/src/orchestration/decider.ts` — new command cases +- `apps/server/src/orchestration/projector.ts` — new event cases +- `apps/server/src/persistence/Migrations.ts` — register migration 043 +- `apps/server/src/server.ts` — wire `ScheduleReactorLive` into the reactor + merge +- `apps/server/src/orchestration/Services/OrchestrationReactor.ts` — add + schedule reactor startup + +--- + +## Phase 3: Client Runtime + +### 3a. Command dispatchers (`packages/client-runtime/src/operations/commands.ts`) + +Add `scheduleThread()` and `cancelThreadSchedule()` functions that dispatch the +new commands over WebSocket. Follow the existing `snoozeThread()` / +`unsnoozeThread()` pattern. + +### 3b. Thread commands atom (`packages/client-runtime/src/state/threadCommands.ts`) + +Add `scheduleThread` and `cancelThreadSchedule` to the atom command set, with +capability gating. + +### 3c. Thread state (`packages/client-runtime/src/state/threads.ts`) + +The `schedule` field on `OrchestrationThreadShell` flows through automatically +via the shell subscription — no special handling needed since it's an optional +field on the existing shell type. + +### Files to modify: + +- `packages/client-runtime/src/operations/commands.ts` +- `packages/client-runtime/src/state/threadCommands.ts` + +--- + +## Phase 4: Web UI + +### 4a. Sidebar schedule indicator (`apps/web/src/components/Sidebar.tsx`) + +Add a small clock/schedule icon on thread rows that have an active schedule, +similar to how `backgroundLiveness` shows "working"/"monitoring". Show next run +time on hover. + +### 4b. Thread action menu (`apps/web/src/components/threadActionMenu.logic.ts`) + +Add "Schedule..." and "Cancel schedule" menu items alongside existing +settle/snooze/pin actions. + +### 4c. Schedule dialog (`apps/web/src/components/ScheduleDialog.tsx`) + +A modal dialog with: + +- Prompt textarea (the message to send on each run) +- Schedule type toggle: Interval (every N minutes/hours/days) or Cron (RRULE + input) +- Quick presets: Every hour, Daily at 9am, Weekly Monday 9am +- Model selection (optional override) +- "Next run" preview +- Save / Cancel buttons + +### 4d. Schedule management view + +For v1, the schedule is managed per-thread via the action menu and dialog. No +separate "Scheduled" inbox view (that can come in v2). + +### Files to create: + +- `apps/web/src/components/ScheduleDialog.tsx` + +### Files to modify: + +- `apps/web/src/components/Sidebar.tsx` — schedule indicator +- `apps/web/src/components/threadActionMenu.logic.ts` — menu items + +--- + +## Phase 5: Mobile & Desktop + +### Desktop + +The desktop app wraps the web app, so the web UI changes propagate +automatically. No separate work needed. + +### Mobile (React Native) + +Add the schedule action to the thread long-press menu. The schedule dialog needs +a React Native equivalent. Can be deferred to a follow-up if needed. + +--- + +## Migration & Backward Compatibility + +- New fields are `Schema.optional(...)` on both `OrchestrationThread` and + `OrchestrationThreadShell`, so older clients/servers decode safely (field + absent = no schedule). +- The DB migration uses `ALTER TABLE ... ADD COLUMN` with nullable `TEXT`, + following the established pattern. +- The `ScheduleReactor` only runs on the server — no client changes required for + firing. +- Event schema is additive (new event types in the `OrchestrationEventType` + literals union). + +--- + +## What's NOT in scope (v1) + +- Event-based triggers (GitHub PR, Gmail, Slack) — future +- Standalone (fresh thread) automations — future +- "Scheduled" inbox/management view — future +- Manual "Run now" button — future +- Skills integration — future +- Admin controls for disabling schedules — future + +--- + +## Summary of changes by layer + +| Layer | Files | Change | +| ---------------- | ------------------------------------------------------------------------- | ---------------------------------------------- | +| Contracts | `packages/contracts/src/orchestration.ts` | New schemas, commands, events, union additions | +| Server decider | `apps/server/src/orchestration/decider.ts` | 2 new command cases | +| Server projector | `apps/server/src/orchestration/projector.ts` | 2 new event cases | +| Server migration | `apps/server/src/persistence/Migrations/043_*.ts` | New migration | +| Server reactor | `apps/server/src/orchestration/Services/ScheduleReactor.ts` | New file: timer heap + fire logic | +| Server wiring | `apps/server/src/server.ts`, `OrchestrationReactor.ts` | Wire reactor | +| Client runtime | `packages/client-runtime/src/operations/commands.ts`, `threadCommands.ts` | 2 new command dispatchers | +| Web UI | `Sidebar.tsx`, `threadActionMenu.logic.ts`, `ScheduleDialog.tsx` | Schedule indicator, menu items, dialog | +| Mobile | Thread long-press menu | Deferred | + +## Codex Automations Reference + +For comparison, Codex Automations (official name: "Scheduled tasks") provide: + +- **Two types**: Standalone (fresh thread each run) and Thread (same thread, + context preserved) +- **Time-based scheduling**: RRULE format, minute intervals, + daily/weekly/monthly +- **Event triggers** (Aug 2026): Gmail, Slack, GitHub PR activity +- **Full agent capabilities**: write code, create PRs, run tests, use + plugins/MCP +- **Skills integration**: `$skill-name` syntax in prompts +- **Same-thread context**: Agent resumes the same conversation, accumulating + understanding across runs +- **GA since March 2026**, top-level feature April 2026 + +Our v1 covers the "thread-type, time-based" subset of this, which is the core +value. Event triggers, skills, and standalone automations are natural +follow-ups. From 6db1aaed5e3ff97c9b7e57d616214d68c1bc3114 Mon Sep 17 00:00:00 2001 From: Michael Buluma <1452922+buluma@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:52:28 +0300 Subject: [PATCH 02/13] feat(contracts): add thread scheduling schema, commands and events Adds ThreadSchedule (enabled, cron/intervalMs, prompt, nextRunAt, createdAt, optional modelSelection override) as an optional field on OrchestrationThread/OrchestrationThreadShell, plus the thread.schedule.create / thread.schedule.cancel commands and their thread.scheduled / thread.unscheduled events. Also adds the threadScheduling environment capability flag for client/server version-skew gating, matching the threadSettlement/threadSnooze pattern. --- packages/contracts/src/environment.ts | 3 ++ packages/contracts/src/orchestration.ts | 61 +++++++++++++++++++++++++ 2 files changed, 64 insertions(+) diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 8d6eca48..a5eec9ab 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -66,6 +66,9 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Server understands thread.snooze / thread.unsnooze commands. Same version-skew contract as threadSettlement. */ threadSnooze: Schema.optionalKey(Schema.Boolean), + /** Server understands thread.schedule.create / thread.schedule.cancel commands. Same + version-skew contract as threadSettlement. */ + threadScheduling: Schema.optionalKey(Schema.Boolean), /** Server understands thread.pin / thread.unpin commands. Same version-skew contract as threadSettlement. */ threadPinning: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 7017fe31..ce4978e2 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -424,6 +424,23 @@ export const ThreadTitleRegeneration = Schema.Struct({ }); export type ThreadTitleRegeneration = typeof ThreadTitleRegeneration.Type; +export const ThreadSchedule = Schema.Struct({ + enabled: Schema.Boolean, + /** ISO 8601 RRULE string (e.g., "FREQ=DAILY;BYHOUR=9;BYMINUTE=0") or null for interval-only. */ + cron: Schema.NullOr(TrimmedNonEmptyString), + /** Interval in milliseconds for simple "every N minutes" schedules. Null when cron is set. */ + intervalMs: Schema.NullOr(Schema.Number), + /** The prompt to send on each scheduled run. */ + prompt: TrimmedNonEmptyString, + /** When the next run should fire. */ + nextRunAt: IsoDateTime, + /** When the schedule was created. */ + createdAt: IsoDateTime, + /** Optional model override for scheduled runs. */ + modelSelection: Schema.optional(ModelSelection), +}); +export type ThreadSchedule = typeof ThreadSchedule.Type; + export const OrchestrationThread = Schema.Struct({ id: ThreadId, projectId: ProjectId, @@ -460,6 +477,9 @@ export const OrchestrationThread = Schema.Struct({ pinOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), // Pending-only state. Optional so older servers remain compatible. titleRegeneration: Schema.optional(Schema.NullOr(ThreadTitleRegeneration)), + // Scheduled task configuration. Optional so payloads from pre-automation + // servers still decode. + schedule: Schema.optional(Schema.NullOr(ThreadSchedule)), deletedAt: Schema.NullOr(IsoDateTime), messages: Schema.Array(OrchestrationMessage), proposedPlans: Schema.Array(OrchestrationProposedPlan).pipe( @@ -518,6 +538,7 @@ export const OrchestrationThreadShell = Schema.Struct({ pinnedAt: Schema.optional(Schema.NullOr(IsoDateTime)), pinOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), titleRegeneration: Schema.optional(Schema.NullOr(ThreadTitleRegeneration)), + schedule: Schema.optional(Schema.NullOr(ThreadSchedule)), session: Schema.NullOr(OrchestrationSession), latestUserMessageAt: Schema.NullOr(IsoDateTime), hasPendingApprovals: Schema.Boolean, @@ -784,6 +805,19 @@ const ThreadUnsnoozeCommand = Schema.Struct({ reason: Schema.Literal("user"), }); +const ThreadScheduleCreateCommand = Schema.Struct({ + type: Schema.Literal("thread.schedule.create"), + commandId: CommandId, + threadId: ThreadId, + schedule: ThreadSchedule, +}); + +const ThreadScheduleCancelCommand = Schema.Struct({ + type: Schema.Literal("thread.schedule.cancel"), + commandId: CommandId, + threadId: ThreadId, +}); + const ThreadPinCommand = Schema.Struct({ type: Schema.Literal("thread.pin"), commandId: CommandId, @@ -970,6 +1004,8 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ThreadUnsettleCommand, ThreadSnoozeCommand, ThreadUnsnoozeCommand, + ThreadScheduleCreateCommand, + ThreadScheduleCancelCommand, ThreadPinCommand, ThreadUnpinCommand, ThreadPinReorderCommand, @@ -998,6 +1034,8 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadUnsettleCommand, ThreadSnoozeCommand, ThreadUnsnoozeCommand, + ThreadScheduleCreateCommand, + ThreadScheduleCancelCommand, ThreadPinCommand, ThreadUnpinCommand, ThreadPinReorderCommand, @@ -1122,6 +1160,8 @@ export const OrchestrationEventType = Schema.Literals([ "thread.unsettled", "thread.snoozed", "thread.unsnoozed", + "thread.scheduled", + "thread.unscheduled", "thread.pinned", "thread.unpinned", "thread.pin-reordered", @@ -1237,6 +1277,17 @@ export const ThreadUnsnoozedPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadScheduledPayload = Schema.Struct({ + threadId: ThreadId, + schedule: ThreadSchedule, + updatedAt: IsoDateTime, +}); + +export const ThreadUnscheduledPayload = Schema.Struct({ + threadId: ThreadId, + updatedAt: IsoDateTime, +}); + export const ThreadPinnedPayload = Schema.Struct({ threadId: ThreadId, pinnedAt: IsoDateTime, @@ -1454,6 +1505,16 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.unsnoozed"), payload: ThreadUnsnoozedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.scheduled"), + payload: ThreadScheduledPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.unscheduled"), + payload: ThreadUnscheduledPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.pinned"), From e8e78ae88463d2c8a93edad3a2e501a2397a9168 Mon Sep 17 00:00:00 2001 From: Michael Buluma <1452922+buluma@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:53:04 +0300 Subject: [PATCH 03/13] feat(server): add ScheduleReactor and thread schedule commands Decider validates and emits thread.scheduled/thread.unscheduled; projector applies them to the read model's schedule field, which is now persisted (migration 043, projection_threads.schedule TEXT column) and threaded through ProjectionSnapshotQuery/ProjectionThreads. ScheduleReactor is the new piece: on startup it loads every thread with a non-null schedule into an in-memory timer heap (one Effect fiber per thread, sleeping until nextRunAt), adds/removes entries as thread.scheduled/unscheduled events arrive, and on each fire dispatches thread.turn.start with the schedule's prompt before recomputing and re-emitting the next run. A due thread with an active session skips that fire and reschedules rather than double-starting a turn. Wired into OrchestrationReactor's startup sequence and server.ts alongside the other reactors. --- .../OrchestrationEngineHarness.integration.ts | 7 + .../src/environment/ServerEnvironment.ts | 1 + .../Layers/OrchestrationReactor.test.ts | 11 + .../Layers/OrchestrationReactor.ts | 3 + .../Layers/ProjectionSnapshotQuery.test.ts | 2 + .../Layers/ProjectionSnapshotQuery.ts | 12 + .../Layers/ScheduleReactor.test.ts | 286 +++++++++++++++++ .../orchestration/Layers/ScheduleReactor.ts | 303 ++++++++++++++++++ apps/server/src/orchestration/Schemas.ts | 4 + .../orchestration/Services/ScheduleReactor.ts | 39 +++ apps/server/src/orchestration/decider.ts | 68 ++++ apps/server/src/orchestration/projector.ts | 24 ++ .../persistence/Layers/ProjectionThreads.ts | 8 +- apps/server/src/persistence/Migrations.ts | 2 + .../043_ProjectionThreadsSchedule.ts | 16 + .../persistence/Services/ProjectionThreads.ts | 2 + apps/server/src/server.ts | 2 + 17 files changed, 789 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/orchestration/Layers/ScheduleReactor.test.ts create mode 100644 apps/server/src/orchestration/Layers/ScheduleReactor.ts create mode 100644 apps/server/src/orchestration/Services/ScheduleReactor.ts create mode 100644 apps/server/src/persistence/Migrations/043_ProjectionThreadsSchedule.ts diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index 1dea9174..a454fc8f 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -63,6 +63,7 @@ import { type OrchestrationEngineShape, } from "../src/orchestration/Services/OrchestrationEngine.ts"; import { ThreadDeletionReactor } from "../src/orchestration/Services/ThreadDeletionReactor.ts"; +import { ScheduleReactor } from "../src/orchestration/Services/ScheduleReactor.ts"; import { OrchestrationReactor } from "../src/orchestration/Services/OrchestrationReactor.ts"; import { ProjectionSnapshotQuery } from "../src/orchestration/Services/ProjectionSnapshotQuery.ts"; import { @@ -375,6 +376,12 @@ export const makeOrchestrationIntegrationHarness = ( drain: Effect.void, }), ), + Layer.provideMerge( + Layer.succeed(ScheduleReactor, { + start: () => Effect.void, + drain: Effect.void, + }), + ), Layer.provideMerge( Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, { publishThread: () => Effect.void, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 558fe4d2..ec097002 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -152,6 +152,7 @@ export const make = Effect.gen(function* () { pullRequests: true, threadSettlement: true, threadSnooze: true, + threadScheduling: true, threadPinning: true, threadPinReorder: true, threadTitleRegeneration: true, diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts index 300d1526..e2ec0bf0 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts @@ -9,6 +9,7 @@ import { CheckpointReactor } from "../Services/CheckpointReactor.ts"; import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; +import { ScheduleReactor } from "../Services/ScheduleReactor.ts"; import { OrchestrationReactor } from "../Services/OrchestrationReactor.ts"; import { makeOrchestrationReactor } from "./OrchestrationReactor.ts"; import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; @@ -64,6 +65,15 @@ describe("OrchestrationReactor", () => { drain: Effect.void, }), ), + Layer.provideMerge( + Layer.succeed(ScheduleReactor, { + start: () => { + started.push("schedule-reactor"); + return Effect.void; + }, + drain: Effect.void, + }), + ), Layer.provideMerge( Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, { publishThread: () => Effect.void, @@ -85,6 +95,7 @@ describe("OrchestrationReactor", () => { "provider-command-reactor", "checkpoint-reactor", "thread-deletion-reactor", + "schedule-reactor", "agent-awareness-relay", ]); diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts index fb7543e3..16777179 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts @@ -9,6 +9,7 @@ import { CheckpointReactor } from "../Services/CheckpointReactor.ts"; import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; +import { ScheduleReactor } from "../Services/ScheduleReactor.ts"; import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; export const makeOrchestrationReactor = Effect.gen(function* () { @@ -16,6 +17,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () { const providerCommandReactor = yield* ProviderCommandReactor; const checkpointReactor = yield* CheckpointReactor; const threadDeletionReactor = yield* ThreadDeletionReactor; + const scheduleReactor = yield* ScheduleReactor; const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; const start: OrchestrationReactorShape["start"] = Effect.fn("start")(function* () { @@ -23,6 +25,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () { yield* providerCommandReactor.start(); yield* checkpointReactor.start(); yield* threadDeletionReactor.start(); + yield* scheduleReactor.start(); yield* agentAwarenessRelay.start(); }); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 5bd8bb60..eb739b1e 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -330,6 +330,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { pinnedAt: "2026-02-24T00:00:01.000Z", pinOrderKey: "gm", titleRegeneration: null, + schedule: null, deletedAt: null, messages: [ { @@ -452,6 +453,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { pinnedAt: "2026-02-24T00:00:01.000Z", pinOrderKey: "gm", titleRegeneration: null, + schedule: null, session: { threadId: ThreadId.make("thread-1"), status: "running", diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 74660bd3..1d566f31 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -25,6 +25,7 @@ import { ModelSelection, ProjectId, ThreadId, + ThreadSchedule, ThreadTokenUsageSnapshot, } from "@helmcode/contracts"; import * as Arr from "effect/Array"; @@ -95,6 +96,7 @@ const ProjectionThreadProposedPlanDbRowSchema = ProjectionThreadProposedPlan; const ProjectionThreadDbRowSchema = ProjectionThread.mapFields( Struct.assign({ modelSelection: Schema.fromJsonString(ModelSelection), + schedule: Schema.NullOr(Schema.fromJsonString(ThreadSchedule)), }), ); const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( @@ -471,6 +473,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + schedule AS "schedule", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -507,6 +510,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + schedule AS "schedule", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -545,6 +549,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + schedule AS "schedule", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -996,6 +1001,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + schedule AS "schedule", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -1855,6 +1861,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + schedule: row.schedule, deletedAt: row.deletedAt, messages: messagesByThread.get(row.threadId) ?? [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -2062,6 +2069,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + schedule: row.schedule, deletedAt: row.deletedAt, messages: [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], @@ -2198,6 +2206,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + schedule: row.schedule, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -2343,6 +2352,7 @@ pending_approval_requests AS ( pinnedAt: row.pinnedAt, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), + schedule: row.schedule, session: sessionByThread.get(row.threadId) ?? null, latestUserMessageAt: row.latestUserMessageAt, hasPendingApprovals: row.pendingApprovalCount > 0, @@ -2622,6 +2632,7 @@ pending_approval_requests AS ( pinnedAt: threadRow.value.pinnedAt, pinOrderKey: threadRow.value.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(threadRow.value), + schedule: threadRow.value.schedule, session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null, latestUserMessageAt: threadRow.value.latestUserMessageAt, hasPendingApprovals: threadRow.value.pendingApprovalCount > 0, @@ -2856,6 +2867,7 @@ pending_approval_requests AS ( pinnedAt: threadRow.value.pinnedAt, pinOrderKey: threadRow.value.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(threadRow.value), + schedule: threadRow.value.schedule, deletedAt: null, messages: messageRows.map((row) => { const message = { diff --git a/apps/server/src/orchestration/Layers/ScheduleReactor.test.ts b/apps/server/src/orchestration/Layers/ScheduleReactor.test.ts new file mode 100644 index 00000000..f65effe0 --- /dev/null +++ b/apps/server/src/orchestration/Layers/ScheduleReactor.test.ts @@ -0,0 +1,286 @@ +import { + DEFAULT_PROVIDER_INTERACTION_MODE, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationCommand, + type OrchestrationReadModel, +} from "@helmcode/contracts"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { describe, expect, it } from "vite-plus/test"; + +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +import { ScheduleReactor } from "../Services/ScheduleReactor.ts"; +import { computeNextRunAt, ScheduleReactorLive } from "./ScheduleReactor.ts"; + +const now = "1970-01-01T00:00:00.000Z"; +const threadId = ThreadId.make("schedule-reactor-test-thread"); +const projectId = ProjectId.make("schedule-reactor-test-project"); + +const fakeCrypto: Crypto.Crypto = Crypto.make({ + randomBytes: (size) => globalThis.crypto.getRandomValues(new Uint8Array(size)), + digest: (algorithm, data) => + Effect.promise(() => globalThis.crypto.subtle.digest(algorithm as string, data)).pipe( + Effect.map((buffer) => new Uint8Array(buffer)), + ), +}); + +const stubSnapshotQuery = ( + getSnapshot: () => Effect.Effect, +): ProjectionSnapshotQuery["Service"] => ({ + getCommandReadModel: getSnapshot, + getSnapshot, + getShellSnapshot: () => + Effect.succeed({ snapshotSequence: 1, projects: [], threads: [], updatedAt: now }), + getArchivedShellSnapshot: () => + Effect.succeed({ snapshotSequence: 1, projects: [], threads: [], updatedAt: now }), + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), + getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }), + getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()), + getProjectShellById: () => Effect.succeed(Option.none()), + getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()), + getThreadCheckpointContext: () => Effect.succeed(Option.none()), + getFullThreadDiffContext: () => Effect.succeed(Option.none()), + getThreadShellById: () => Effect.succeed(Option.none()), + getThreadDetailById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.succeed(Option.none()), + searchThreads: () => Effect.succeed({ matches: [] }), +}); + +type ThreadOverrides = { + schedule?: OrchestrationReadModel["threads"][number]["schedule"]; + session?: OrchestrationReadModel["threads"][number]["session"]; +}; + +const readModel = (overrides: ThreadOverrides): OrchestrationReadModel => ({ + snapshotSequence: 1, + updatedAt: now, + projects: [ + { + id: projectId, + title: "Project", + workspaceRoot: "/tmp/project", + defaultModelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + scripts: [], + createdAt: now, + updatedAt: now, + deletedAt: null, + }, + ], + threads: [ + { + id: threadId, + projectId, + title: "Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + latestTurn: null, + messages: [], + session: overrides.session ?? null, + activities: [], + proposedPlans: [], + checkpoints: [], + deletedAt: null, + schedule: overrides.schedule ?? null, + }, + ], +}); + +const runningSession: NonNullable = { + threadId, + status: "running", + providerName: null, + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, +}; + +const schedule = ( + overrides: Partial<{ + nextRunAt: string; + intervalMs: number | null; + cron: string | null; + enabled: boolean; + prompt: string; + }> = {}, +) => ({ + enabled: overrides.enabled ?? true, + cron: overrides.cron ?? null, + intervalMs: overrides.intervalMs ?? 60_000, + prompt: overrides.prompt ?? "Continue where you left off.", + nextRunAt: overrides.nextRunAt ?? "1970-01-01T00:00:10.000Z", + createdAt: now, +}); + +/** + * Boots the ScheduleReactor against mock engine + snapshot services under a + * TestClock, advances the virtual clock past `nextRunAt` so the timer fires + * deterministically, and returns the recorded commands. + */ +const runReactor = ( + overrides: ThreadOverrides, + adjust: Duration.Duration, +): Effect.Effect< + { + readonly commands: Array; + readonly turnStarts: Array<{ threadId: ThreadId; text: string }>; + readonly reschedules: Array; + }, + never, + never +> => + Effect.gen(function* () { + const commands = yield* Ref.make>([]); + const turnStarts = yield* Ref.make>([]); + const reschedules = yield* Ref.make>([]); + + const engine: OrchestrationEngineService["Service"] = { + readEvents: () => Stream.empty, + dispatch: (command: OrchestrationCommand) => + Ref.update(commands, (calls) => [...calls, command.type]).pipe( + Effect.flatMap(() => { + if (command.type === "thread.turn.start") { + return Ref.update(turnStarts, (calls) => [ + ...calls, + { threadId: command.threadId, text: command.message.text }, + ]); + } + if (command.type === "thread.schedule.create") { + return Ref.update(reschedules, (calls) => [...calls, command.schedule.nextRunAt]); + } + return Effect.void; + }), + Effect.as({ sequence: 1 } as const), + ), + streamDomainEvents: Stream.never, + latestSequence: Effect.succeed(0), + }; + + const snapshotQuery = stubSnapshotQuery(() => Effect.succeed(readModel(overrides))); + + const testLayer = ScheduleReactorLive.pipe( + Layer.provideMerge(TestClock.layer()), + Layer.provideMerge(Layer.succeed(OrchestrationEngineService, engine)), + Layer.provideMerge(Layer.succeed(ProjectionSnapshotQuery, snapshotQuery)), + Layer.provideMerge(Layer.succeed(Crypto.Crypto, fakeCrypto)), + ); + + yield* Effect.scoped( + Effect.gen(function* () { + const reactor = yield* Effect.service(ScheduleReactor); + yield* reactor.start(); + yield* TestClock.adjust(adjust); + }), + ).pipe(Effect.provide(testLayer)); + + return { + commands: yield* Ref.get(commands), + turnStarts: yield* Ref.get(turnStarts), + reschedules: yield* Ref.get(reschedules), + }; + }); + +describe("computeNextRunAt", () => { + it("adds the interval for interval-based schedules", () => { + const next = computeNextRunAt({ cron: null, intervalMs: 3_600_000 }, DateTime.makeUnsafe(now)); + expect(DateTime.formatIso(next)).toBe("1970-01-01T01:00:00.000Z"); + }); + + it("returns +1 hour for HOURLY cron", () => { + const next = computeNextRunAt( + { cron: "FREQ=HOURLY", intervalMs: null }, + DateTime.makeUnsafe(now), + ); + expect(DateTime.formatIso(next)).toBe("1970-01-01T01:00:00.000Z"); + }); + + it("computes the next DAILY run at the target time, advancing a day when passed", () => { + const midnight = DateTime.makeUnsafe(now); + const next = computeNextRunAt( + { cron: "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", intervalMs: null }, + midnight, + ); + expect(DateTime.formatIso(next)).toBe("1970-01-01T09:00:00.000Z"); + + const midday = DateTime.makeUnsafe("1970-01-01T12:00:00.000Z"); + const nextDay = computeNextRunAt( + { cron: "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", intervalMs: null }, + midday, + ); + expect(DateTime.formatIso(nextDay)).toBe("1970-01-02T09:00:00.000Z"); + }); + + it("rolls a WEEKLY schedule forward to the configured weekday, wrapping to next week", () => { + // 2026-09-04 is a Friday; the next Monday is 2 days ahead. + const friday = DateTime.makeUnsafe("2026-09-04T10:00:00.000Z"); + const next = computeNextRunAt( + { cron: "FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=0", intervalMs: null }, + friday, + ); + expect(DateTime.formatIso(next)).toBe("2026-09-07T09:00:00.000Z"); + + // A Monday morning before the target time schedules the same Monday. + const mondayEarly = DateTime.makeUnsafe("2026-09-07T08:00:00.000Z"); + const sameDay = computeNextRunAt( + { cron: "FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=0", intervalMs: null }, + mondayEarly, + ); + expect(DateTime.formatIso(sameDay)).toBe("2026-09-07T09:00:00.000Z"); + }); + + it("defaults a non-HOURLY cron without a parseable target to +1 day", () => { + const next = computeNextRunAt( + { cron: "FREQ=MONTHLY", intervalMs: null }, + DateTime.makeUnsafe(now), + ); + expect(DateTime.formatIso(next)).toBe("1970-01-02T00:00:00.000Z"); + }); +}); + +describe("ScheduleReactor firing", () => { + it("dispatches a turn start with the schedule prompt and reschedules when the timer fires", async () => { + const { commands, turnStarts, reschedules } = await Effect.runPromise( + runReactor({ schedule: schedule() }, Duration.seconds(10)), + ); + + expect(commands).toContain("thread.turn.start"); + expect(commands).toContain("thread.schedule.create"); + expect(turnStarts).toEqual([{ threadId, text: "Continue where you left off." }]); + expect(reschedules).toHaveLength(1); + }); + + it("skips firing while a turn is running and only reschedules", async () => { + const { commands, turnStarts, reschedules } = await Effect.runPromise( + runReactor({ schedule: schedule(), session: runningSession }, Duration.seconds(10)), + ); + + expect(commands).not.toContain("thread.turn.start"); + expect(commands).toContain("thread.schedule.create"); + expect(turnStarts).toEqual([]); + expect(reschedules).toHaveLength(1); + }); +}); diff --git a/apps/server/src/orchestration/Layers/ScheduleReactor.ts b/apps/server/src/orchestration/Layers/ScheduleReactor.ts new file mode 100644 index 00000000..bb62e8ca --- /dev/null +++ b/apps/server/src/orchestration/Layers/ScheduleReactor.ts @@ -0,0 +1,303 @@ +import { + CommandId, + type OrchestrationCommand, + type OrchestrationThread, + MessageId, + type ThreadSchedule, + ThreadId, +} from "@helmcode/contracts"; +import * as Cause from "effect/Cause"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as HashMap from "effect/HashMap"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import type * as PlatformError from "effect/PlatformError"; +import * as Ref from "effect/Ref"; +import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; + +import { forkParked } from "../../serverActivation.ts"; +import type { OrchestrationDispatchError } from "../Errors.ts"; +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +import { ScheduleReactor, type ScheduleReactorShape } from "../Services/ScheduleReactor.ts"; + +const RRULE_FREQ = /FREQ=(DAILY|WEEKLY|MONTHLY|HOURLY)/; +const RRULE_HOUR = /BYHOUR=(\d+)/; +const RRULE_MINUTE = /BYMINUTE=(\d+)/; +const RRULE_WEEKDAY = /BYDAY=(MO|TU|WE|TH|FR|SA|SU)/; +const WEEKDAY_TO_JS_DAY: Record = { + MO: 1, + TU: 2, + WE: 3, + TH: 4, + FR: 5, + SA: 6, + SU: 0, +}; + +/** + * Compute the next run time for an interval- or cron-based schedule. + * Supports the basic RRULE subset (FREQ=DAILY|WEEKLY|MONTHLY|HOURLY with + * optional BYHOUR/BYMINUTE/BYDAY); anything else falls back to +1 hour. + */ +export const computeNextRunAt = ( + schedule: Pick, + now: DateTime.DateTime, +): DateTime.DateTime => { + if (schedule.intervalMs != null) { + return DateTime.add(now, { milliseconds: Math.max(1, schedule.intervalMs) }); + } + + const cron = schedule.cron; + if (cron == null) { + // Fallback: 1 hour from now. + return DateTime.add(now, { hours: 1 }); + } + + const upper = cron.toUpperCase(); + const freq = RRULE_FREQ.exec(upper)?.[1]; + const hour = RRULE_HOUR.exec(upper)?.[1]; + const minute = RRULE_MINUTE.exec(upper)?.[1]; + + if (freq === "HOURLY") { + return DateTime.add(now, { hours: 1 }); + } + + const targetHour = hour === undefined ? DateTime.getPartUtc(now, "hour") : Number(hour); + const targetMinute = minute === undefined ? DateTime.getPartUtc(now, "minute") : Number(minute); + + // Build the candidate "this period at target time" as UTC. + let next = DateTime.makeUnsafe({ + year: DateTime.getPartUtc(now, "year"), + month: DateTime.getPartUtc(now, "month"), + day: DateTime.getPartUtc(now, "day"), + hour: targetHour, + minute: targetMinute, + second: 0, + millisecond: 0, + }); + + if (freq === "WEEKLY") { + const weekday = RRULE_WEEKDAY.exec(upper)?.[1]; + const targetDay = weekday === undefined ? 1 : (WEEKDAY_TO_JS_DAY[weekday] ?? 1); + const currentDay = DateTime.getPartUtc(now, "weekDay"); + let daysAhead = targetDay - currentDay; + // Target earlier in the week than today: roll to next week. + if (daysAhead < 0) daysAhead += 7; + // Same weekday but the target time already passed: roll to next week. + if (daysAhead === 0 && DateTime.isLessThanOrEqualTo(next, now)) daysAhead = 7; + next = DateTime.add(next, { days: daysAhead }); + } else if (DateTime.isLessThanOrEqualTo(next, now)) { + // DAILY or MONTHLY: advance to the next period. + next = DateTime.add(next, { days: 1 }); + } + + return next; +}; + +const make = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const randomUUID = crypto.randomUUIDv4; + const serverCommandId = (tag: string) => + randomUUID.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`))); + const orchestrationEngine = yield* OrchestrationEngineService; + const snapshotQuery = yield* ProjectionSnapshotQuery; + + // Map of threadId → active timer fiber. Kept in a Ref so schedule changes + // can interrupt and replace a fiber; `drain` joins every live timer. + const timers = yield* Ref.make(HashMap.empty>()); + + // Dispatch a schedule command (already built as an effect to surface UUID + // generation failures). Any error is logged and swallowed so the timer fiber + // never fails below. + const dispatchScheduleUpdate = ( + commandEffect: Effect.Effect< + OrchestrationCommand, + PlatformError.PlatformError | OrchestrationDispatchError, + never + >, + ): Effect.Effect => + commandEffect.pipe( + Effect.flatMap((command) => orchestrationEngine.dispatch(command)), + Effect.catch((error) => + Effect.logWarning("schedule reactor failed to dispatch command", { + cause: String(error), + }), + ), + Effect.asVoid, + ); + + const scheduleCreateCommand = ( + threadId: ThreadId, + schedule: ThreadSchedule, + ): Effect.Effect => + Effect.map(serverCommandId("schedule-update"), (commandId) => ({ + type: "thread.schedule.create" as const, + commandId, + threadId, + schedule, + })); + + // Dispatches the turn-start for a thread's schedule, then advances the + // schedule's nextRunAt so the timer chain continues. + const fireSchedule = Effect.fn("fireSchedule")(function* (thread: OrchestrationThread) { + const schedule = thread.schedule; + if (schedule == null || !schedule.enabled) { + return; + } + + // Skip firing while a turn is already running; retry in 5 minutes. + if (thread.session != null && thread.session.status === "running") { + const now = yield* DateTime.now; + const nextRunAt = DateTime.formatIso(DateTime.add(now, { minutes: 5 })); + yield* dispatchScheduleUpdate(scheduleCreateCommand(thread.id, { ...schedule, nextRunAt })); + return; + } + + const now = yield* DateTime.now; + const createdAt = DateTime.formatIso(now); + const messageId = yield* randomUUID.pipe( + Effect.map(MessageId.make), + Effect.catch((error) => + Effect.logWarning("schedule reactor failed to generate message id", { + threadId: thread.id, + cause: String(error), + }).pipe(Effect.as(MessageId.make(`schedule:${thread.id}:${DateTime.toEpochMillis(now)}`))), + ), + ); + yield* dispatchScheduleUpdate( + Effect.map(serverCommandId("schedule-turn-start"), (commandId) => ({ + type: "thread.turn.start" as const, + commandId, + threadId: thread.id, + message: { + messageId, + role: "user" as const, + text: schedule.prompt, + attachments: [], + }, + ...(schedule.modelSelection !== undefined + ? { modelSelection: schedule.modelSelection } + : {}), + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + createdAt, + })), + ); + + const nextRunAt = DateTime.formatIso(computeNextRunAt(schedule, yield* DateTime.now)); + yield* dispatchScheduleUpdate(scheduleCreateCommand(thread.id, { ...schedule, nextRunAt })); + }); + + // Interrupt (and remove bookkeeping happens at next start) any timer already + // scheduled for the thread so a schedule update replaces rather than stacks. + const interruptTimer = (threadId: ThreadId): Effect.Effect => + Ref.get(timers).pipe( + Effect.flatMap((map) => + Option.match(HashMap.get(map, threadId), { + onNone: () => Effect.void, + onSome: (fiber) => Fiber.interrupt(fiber).pipe(Effect.asVoid), + }), + ), + ); + + // Reads the freshest thread state and fires its schedule if due. + const fireTimerForThread = Effect.fn("fireTimerForThread")(function* (threadId: ThreadId) { + const snapshot = yield* snapshotQuery.getSnapshot().pipe( + Effect.catch((error) => + Effect.logWarning("schedule reactor failed to read snapshot", { + threadId, + cause: String(error), + }).pipe(Effect.as(null)), + ), + ); + if (snapshot == null) { + return; + } + const thread = snapshot.threads.find((candidate) => candidate.id === threadId); + if (thread == null) { + return; + } + yield* fireSchedule(thread); + }); + + const startTimerForSchedule = Effect.fn("startTimerForSchedule")(function* ( + threadId: ThreadId, + schedule: ThreadSchedule, + ) { + if (!schedule.enabled) { + return; + } + + yield* interruptTimer(threadId); + + const now = yield* DateTime.now; + const nextRunAt = Option.getOrElse(DateTime.make(schedule.nextRunAt), () => now); + const delayMs = Math.max(0, DateTime.toEpochMillis(nextRunAt) - DateTime.toEpochMillis(now)); + + const timer = Effect.sleep(`${delayMs} millis`).pipe( + Effect.andThen(fireTimerForThread(threadId)), + Effect.catchCause((cause) => { + if (Cause.hasInterruptsOnly(cause)) { + // Let interrupts (schedule replacement / scope close) propagate so the + // fiber ends cleanly instead of leaving a stray firing behind. + return Effect.failCause(cause); + } + return Effect.logWarning("schedule timer failed", { + threadId, + cause: Cause.pretty(cause), + }); + }), + ); + + const fiber = yield* Effect.forkScoped(timer); + yield* Ref.update(timers, (map) => HashMap.set(map, threadId, fiber)); + }); + + const start: ScheduleReactorShape["start"] = Effect.fn("start")(function* () { + // Bootstrap timers for schedules that exist at startup. + const snapshot = yield* snapshotQuery.getSnapshot().pipe( + Effect.catch((error) => + Effect.logWarning("schedule reactor failed to read snapshot on start", { + cause: String(error), + }).pipe(Effect.as(null)), + ), + ); + if (snapshot != null) { + for (const thread of snapshot.threads) { + if (thread.schedule != null && thread.schedule.enabled) { + yield* startTimerForSchedule(thread.id, thread.schedule); + } + } + } + + // Subscribe to schedule lifecycle events to start/cancel timer fibers. + yield* forkParked( + Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { + if (event.type === "thread.scheduled") { + return startTimerForSchedule(event.payload.threadId, event.payload.schedule); + } + if (event.type === "thread.unscheduled") { + return interruptTimer(event.payload.threadId); + } + return Effect.void; + }), + ); + }); + + const drain: ScheduleReactorShape["drain"] = Ref.get(timers).pipe( + Effect.flatMap((map) => Fiber.joinAll(HashMap.values(map))), + Effect.asVoid, + ); + + return { + start, + drain, + } satisfies ScheduleReactorShape; +}); + +export const ScheduleReactorLive = Layer.effect(ScheduleReactor, make); diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 48eac92c..63d71ecb 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -13,6 +13,8 @@ import { ThreadUnsettledPayload as ContractsThreadUnsettledPayloadSchema, ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, + ThreadScheduledPayload as ContractsThreadScheduledPayloadSchema, + ThreadUnscheduledPayload as ContractsThreadUnscheduledPayloadSchema, ThreadPinnedPayload as ContractsThreadPinnedPayloadSchema, ThreadUnpinnedPayload as ContractsThreadUnpinnedPayloadSchema, ThreadPinReorderedPayload as ContractsThreadPinReorderedPayloadSchema, @@ -45,6 +47,8 @@ export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema; export const ThreadUnsettledPayload = ContractsThreadUnsettledPayloadSchema; export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; +export const ThreadScheduledPayload = ContractsThreadScheduledPayloadSchema; +export const ThreadUnscheduledPayload = ContractsThreadUnscheduledPayloadSchema; export const ThreadPinnedPayload = ContractsThreadPinnedPayloadSchema; export const ThreadUnpinnedPayload = ContractsThreadUnpinnedPayloadSchema; export const ThreadPinReorderedPayload = ContractsThreadPinReorderedPayloadSchema; diff --git a/apps/server/src/orchestration/Services/ScheduleReactor.ts b/apps/server/src/orchestration/Services/ScheduleReactor.ts new file mode 100644 index 00000000..23f8ecb7 --- /dev/null +++ b/apps/server/src/orchestration/Services/ScheduleReactor.ts @@ -0,0 +1,39 @@ +/** + * ScheduleReactor - Scheduled task reactor service interface. + * + * Owns background fibers that fire scheduled turns on threads at configured + * intervals/cron times. Subscribes to domain events for schedule lifecycle + * and manages per-thread timer fibers. + * + * @module ScheduleReactor + */ +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; +import type * as Scope from "effect/Scope"; + +/** + * ScheduleReactorShape - Service API for schedule reactor lifecycle. + */ +export interface ScheduleReactorShape { + /** + * Start reacting to thread.schedule.create / thread.unscheduled domain + * events and fire due schedules. + * + * The returned effect must be run in a scope so all timer fibers can be + * finalized on shutdown. + */ + readonly start: () => Effect.Effect; + + /** + * Resolves when the internal processing queue is empty and idle. + * Intended for test use to replace timing-sensitive sleeps. + */ + readonly drain: Effect.Effect; +} + +/** + * ScheduleReactor - Service tag for schedule reactor workers. + */ +export class ScheduleReactor extends Context.Service()( + "helmcode/orchestration/Services/ScheduleReactor", +) {} diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index d77db296..500110cd 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -646,6 +646,74 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.schedule.create": { + const thread = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.threadId, + }); + const occurredAt = yield* nowIso; + const { schedule } = command; + // Validate: at least one of cron or intervalMs must be set. + if (schedule.cron == null && schedule.intervalMs == null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} schedule must have at least one of cron or intervalMs`, + }); + } + // Validate: prompt must be non-empty (schema guarantees this, but be safe). + if (schedule.prompt.trim().length === 0) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} schedule prompt must not be empty`, + }); + } + // Validate: nextRunAt must be in the future. + if (!(Date.parse(schedule.nextRunAt) > Date.parse(occurredAt))) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `thread ${command.threadId} schedule nextRunAt ${schedule.nextRunAt} is not in the future`, + }); + } + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.scheduled", + payload: { + threadId: command.threadId, + schedule, + updatedAt: occurredAt, + }, + }; + } + + case "thread.schedule.cancel": { + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + const occurredAt = yield* nowIso; + // Idempotent: cancelling a schedule that doesn't exist is a no-op. + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.unscheduled", + payload: { + threadId: command.threadId, + updatedAt: occurredAt, + }, + }; + } + case "thread.pin": { const thread = yield* requireThreadNotArchived({ readModel, diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index 1ee80756..eab4e5eb 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -28,6 +28,8 @@ import { ThreadSnoozedPayload, ThreadUnpinnedPayload, ThreadUnarchivedPayload, + ThreadUnscheduledPayload, + ThreadScheduledPayload, ThreadUnsettledPayload, ThreadUnsnoozedPayload, ThreadRevertedPayload, @@ -405,6 +407,28 @@ export function projectEvent( })), ); + case "thread.scheduled": + return decodeForEvent(ThreadScheduledPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + schedule: payload.schedule, + updatedAt: payload.updatedAt, + }), + })), + ); + + case "thread.unscheduled": + return decodeForEvent(ThreadUnscheduledPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + schedule: null, + updatedAt: payload.updatedAt, + }), + })), + ); + case "thread.pinned": return decodeForEvent(ThreadPinnedPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index 76117906..7c8b489a 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -14,11 +14,12 @@ import { ProjectionThreadRepository, type ProjectionThreadRepositoryShape, } from "../Services/ProjectionThreads.ts"; -import { ModelSelection } from "@helmcode/contracts"; +import { ModelSelection, ThreadSchedule } from "@helmcode/contracts"; const ProjectionThreadDbRow = ProjectionThread.mapFields( Struct.assign({ modelSelection: Schema.fromJsonString(ModelSelection), + schedule: Schema.NullOr(Schema.fromJsonString(ThreadSchedule)), }), ); type ProjectionThreadDbRow = typeof ProjectionThreadDbRow.Type; @@ -51,6 +52,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key, title_regeneration_request_id, title_regeneration_started_at, + schedule, latest_user_message_at, pending_approval_count, pending_user_input_count, @@ -78,6 +80,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.pinOrderKey ?? null}, ${row.titleRegenerationRequestId ?? null}, ${row.titleRegenerationStartedAt ?? null}, + ${row.schedule != null ? JSON.stringify(row.schedule) : null}, ${row.latestUserMessageAt}, ${row.pendingApprovalCount}, ${row.pendingUserInputCount}, @@ -105,6 +108,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key = excluded.pin_order_key, title_regeneration_request_id = excluded.title_regeneration_request_id, title_regeneration_started_at = excluded.title_regeneration_started_at, + schedule = excluded.schedule, latest_user_message_at = excluded.latest_user_message_at, pending_approval_count = excluded.pending_approval_count, pending_user_input_count = excluded.pending_user_input_count, @@ -139,6 +143,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + schedule AS "schedule", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", @@ -175,6 +180,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", + schedule AS "schedule", latest_user_message_at AS "latestUserMessageAt", pending_approval_count AS "pendingApprovalCount", pending_user_input_count AS "pendingUserInputCount", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 90bece92..c81f74d9 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -55,6 +55,7 @@ import Migration0039 from "./Migrations/039_ProjectionProjectsDefaultThreadEnvMo import Migration0040 from "./Migrations/040_ProjectionProjectFaviconPath.ts"; import Migration0041 from "./Migrations/041_ProjectionTurnsStopReasonCost.ts"; import Migration0042 from "./Migrations/042_ProjectionTurnsTokenUsage.ts"; +import Migration0043 from "./Migrations/043_ProjectionThreadsSchedule.ts"; /** * Migration loader with all migrations defined inline. @@ -109,6 +110,7 @@ export const migrationEntries = [ [40, "ProjectionProjectFaviconPath", Migration0040], [41, "ProjectionTurnsStopReasonCost", Migration0041], [42, "ProjectionTurnsTokenUsage", Migration0042], + [43, "ProjectionThreadsSchedule", Migration0043], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/043_ProjectionThreadsSchedule.ts b/apps/server/src/persistence/Migrations/043_ProjectionThreadsSchedule.ts new file mode 100644 index 00000000..9acb43e1 --- /dev/null +++ b/apps/server/src/persistence/Migrations/043_ProjectionThreadsSchedule.ts @@ -0,0 +1,16 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + + if (!columns.some((column) => column.name === "schedule")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN schedule TEXT + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index 9d743353..a177955e 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -15,6 +15,7 @@ import { ProviderInteractionMode, RuntimeMode, ThreadId, + ThreadSchedule, TurnId, } from "@helmcode/contracts"; import * as Option from "effect/Option"; @@ -45,6 +46,7 @@ export const ProjectionThread = Schema.Struct({ pinOrderKey: Schema.optional(Schema.NullOr(Schema.String)), titleRegenerationRequestId: Schema.optional(Schema.NullOr(CommandId)), titleRegenerationStartedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + schedule: Schema.optional(Schema.NullOr(ThreadSchedule)), latestUserMessageAt: Schema.NullOr(IsoDateTime), pendingApprovalCount: NonNegativeInt, pendingUserInputCount: NonNegativeInt, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 2f64a11e..e7fb5adb 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -60,6 +60,7 @@ import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRun import { ProviderCommandReactorLive } from "./orchestration/Layers/ProviderCommandReactor.ts"; import { CheckpointReactorLive } from "./orchestration/Layers/CheckpointReactor.ts"; import { ThreadDeletionReactorLive } from "./orchestration/Layers/ThreadDeletionReactor.ts"; +import { ScheduleReactorLive } from "./orchestration/Layers/ScheduleReactor.ts"; import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts"; import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry.ts"; @@ -250,6 +251,7 @@ const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(ProviderCommandReactorLive), Layer.provideMerge(CheckpointReactorLive), Layer.provideMerge(ThreadDeletionReactorLive), + Layer.provideMerge(ScheduleReactorLive), Layer.provideMerge(AgentAwarenessRelay.layer.pipe(Layer.provide(ServerSecretStore.layer))), Layer.provideMerge(RuntimeReceiptBusLive), ); From d67b5b07ec07fc5b2f3a464727d9a540b99367f2 Mon Sep 17 00:00:00 2001 From: Michael Buluma <1452922+buluma@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:53:32 +0300 Subject: [PATCH 04/13] feat(client-runtime): add scheduleThread/cancelThreadSchedule commands MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dispatchers for thread.schedule.create/cancel over the WebSocket connection, plus the corresponding entries in the thread commands atom with threadScheduling capability gating — follows the existing snoozeThread/unsnoozeThread pattern. --- .../client-runtime/src/operations/commands.ts | 22 ++++++++++++++++ .../src/state/threadCommands.ts | 26 +++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 4f021678..a78972c0 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -39,6 +39,8 @@ export type SettleThreadInput = CommandInput<"thread.settle">; export type UnsettleThreadInput = CommandInput<"thread.unsettle">; export type SnoozeThreadInput = CommandInput<"thread.snooze">; export type UnsnoozeThreadInput = CommandInput<"thread.unsnooze">; +export type ScheduleThreadInput = CommandInput<"thread.schedule.create">; +export type CancelThreadScheduleInput = CommandInput<"thread.schedule.cancel">; export type PinThreadInput = CommandInput<"thread.pin">; export type UnpinThreadInput = CommandInput<"thread.unpin">; export type ReorderPinnedThreadInput = CommandInput<"thread.pin.reorder">; @@ -200,6 +202,26 @@ export const unsnoozeThread: (input: UnsnoozeThreadInput) => CommandEffect = Eff }); }); +export const scheduleThread: (input: ScheduleThreadInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.scheduleThread", +)(function* (input) { + return yield* dispatch({ + ...input, + type: "thread.schedule.create", + commandId: yield* commandId(input), + }); +}); + +export const cancelThreadSchedule: (input: CancelThreadScheduleInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.cancelThreadSchedule", +)(function* (input) { + return yield* dispatch({ + ...input, + type: "thread.schedule.cancel", + commandId: yield* commandId(input), + }); +}); + export const pinThread: (input: PinThreadInput) => CommandEffect = Effect.fn( "EnvironmentCommands.pinThread", )(function* (input) { diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 3926fb3d..d5df6a0f 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -11,6 +11,7 @@ import { Atom, type AtomRegistry } from "effect/unstable/reactivity"; import { createAtomCommandScheduler, createEnvironmentCommand } from "./runtime.ts"; import { type ArchiveThreadInput, + type CancelThreadScheduleInput, type CreateThreadInput, type DeleteThreadInput, type InterruptThreadTurnInput, @@ -19,6 +20,7 @@ import { type RevertThreadCheckpointInput, type SetThreadInteractionModeInput, type SetThreadRuntimeModeInput, + type ScheduleThreadInput, type PinThreadInput, type ReorderPinnedThreadInput, type SettleThreadInput, @@ -31,6 +33,7 @@ import { type UnsnoozeThreadInput, type UpdateThreadMetadataInput, archiveThread, + cancelThreadSchedule, createThread, deleteThread, interruptThreadTurn, @@ -39,6 +42,7 @@ import { revertThreadCheckpoint, setThreadInteractionMode, setThreadRuntimeMode, + scheduleThread, pinThread, reorderPinnedThread, settleThread, @@ -55,6 +59,7 @@ import type { EnvironmentRegistry } from "../connection/registry.ts"; export type { ArchiveThreadInput, + CancelThreadScheduleInput, CreateThreadInput, DeleteThreadInput, InterruptThreadTurnInput, @@ -63,6 +68,7 @@ export type { RevertThreadCheckpointInput, SetThreadInteractionModeInput, SetThreadRuntimeModeInput, + ScheduleThreadInput, PinThreadInput, ReorderPinnedThreadInput, SettleThreadInput, @@ -186,6 +192,26 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + schedule: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:schedule", + execute: requireCapability( + "threadScheduling", + "thread.schedule.create", + (input: ScheduleThreadInput) => scheduleThread(input), + ), + scheduler, + concurrency, + }), + cancelSchedule: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:cancel-schedule", + execute: requireCapability( + "threadScheduling", + "thread.schedule.cancel", + (input: CancelThreadScheduleInput) => cancelThreadSchedule(input), + ), + scheduler, + concurrency, + }), pin: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:pin", execute: requireCapability("threadPinning", "thread.pin", (input: PinThreadInput) => From 188399c7b0e2ea37bd02530e5fcf33ca4736c39c Mon Sep 17 00:00:00 2001 From: Michael Buluma <1452922+buluma@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:54:08 +0300 Subject: [PATCH 05/13] feat(web): add thread scheduling UI with pause/resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ScheduleDialog lets a user set a prompt, interval- or cron-based cadence (with quick presets and an optional per-schedule model override via the existing ProviderModelPicker), previewing the next run computed in UTC to match the server's ScheduleReactor. Wired into the sidebar row and chat header via the shared thread action menu (threadActionMenu.logic.ts), which now offers Schedule…, Pause/Resume schedule, and Cancel schedule depending on state. Pause/resume re-dispatch thread.schedule.create with enabled flipped rather than recreating the schedule, keeping the prompt/cadence/model override intact; resuming recomputes nextRunAt since the decider requires it to be in the future and a paused schedule's slot may have passed. The sidebar clock indicator now distinguishes an active schedule from a paused one. --- .../components/ScheduleDialog.logic.test.ts | 114 ++++++++ .../src/components/ScheduleDialog.logic.ts | 72 +++++ apps/web/src/components/ScheduleDialog.tsx | 252 ++++++++++++++++++ apps/web/src/components/Sidebar.tsx | 100 +++++++ apps/web/src/components/chat/ChatHeader.tsx | 14 + .../components/threadActionMenu.logic.test.ts | 54 +++- .../src/components/threadActionMenu.logic.ts | 17 ++ apps/web/src/hooks/useThreadActionMenu.ts | 27 +- apps/web/src/hooks/useThreadActions.ts | 123 ++++++++- apps/web/src/state/entities.ts | 9 + 10 files changed, 778 insertions(+), 4 deletions(-) create mode 100644 apps/web/src/components/ScheduleDialog.logic.test.ts create mode 100644 apps/web/src/components/ScheduleDialog.logic.ts create mode 100644 apps/web/src/components/ScheduleDialog.tsx diff --git a/apps/web/src/components/ScheduleDialog.logic.test.ts b/apps/web/src/components/ScheduleDialog.logic.test.ts new file mode 100644 index 00000000..9e059aa4 --- /dev/null +++ b/apps/web/src/components/ScheduleDialog.logic.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { computeNextRunAt, PRESETS } from "./ScheduleDialog.logic"; + +describe("PRESETS", () => { + it("exposes the three quick presets from the plan", () => { + expect(PRESETS.map((preset) => preset.label)).toEqual([ + "Every hour", + "Daily at 9am", + "Weekly Monday 9am", + ]); + }); + + it("each preset resolves to a schedule config that computeNextRunAt accepts", () => { + const now = new Date("2026-09-04T12:00:00.000Z"); // Friday + for (const preset of PRESETS) { + const nextRunAt = computeNextRunAt( + preset.mode, + preset.intervalMinutes ?? 60, + preset.cron ?? "", + now, + ); + expect(new Date(nextRunAt).getTime()).toBeGreaterThan(now.getTime()); + } + }); +}); + +describe("computeNextRunAt", () => { + it("interval mode adds N minutes from now", () => { + const now = new Date("2026-09-04T12:00:00.000Z"); + const nextRunAt = computeNextRunAt("interval", 60, "", now); + expect(nextRunAt).toBe("2026-09-04T13:00:00.000Z"); + }); + + it("interval mode floors to at least 1 minute for non-positive input", () => { + const now = new Date("2026-09-04T12:00:00.000Z"); + expect(computeNextRunAt("interval", 0, "", now)).toBe("2026-09-04T12:01:00.000Z"); + expect(computeNextRunAt("interval", -5, "", now)).toBe("2026-09-04T12:01:00.000Z"); + }); + + it("HOURLY cron fires exactly 1 hour from now regardless of BYHOUR/BYMINUTE", () => { + const now = new Date("2026-09-04T12:34:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=HOURLY;BYHOUR=9;BYMINUTE=0", now); + expect(nextRunAt).toBe("2026-09-04T13:34:00.000Z"); + }); + + it("DAILY cron rolls to today's target UTC time when it hasn't passed yet", () => { + const now = new Date("2026-09-04T06:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", now); + expect(nextRunAt).toBe("2026-09-04T09:00:00.000Z"); + }); + + it("DAILY cron rolls to tomorrow when today's target UTC time already passed", () => { + const now = new Date("2026-09-04T10:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", now); + expect(nextRunAt).toBe("2026-09-05T09:00:00.000Z"); + }); + + it("WEEKLY cron with BYDAY rolls forward to the next occurrence of that weekday", () => { + // 2026-09-04 is a Friday. + const now = new Date("2026-09-04T12:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=0", now); + // Next Monday is 2026-09-07. + expect(nextRunAt).toBe("2026-09-07T09:00:00.000Z"); + }); + + it("WEEKLY cron on the target weekday rolls to next week once the time has passed", () => { + // 2026-09-07 is a Monday. + const now = new Date("2026-09-07T10:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=0", now); + expect(nextRunAt).toBe("2026-09-14T09:00:00.000Z"); + }); + + it("WEEKLY cron on the target weekday, before the time, fires today", () => { + // 2026-09-07 is a Monday. + const now = new Date("2026-09-07T06:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=0", now); + expect(nextRunAt).toBe("2026-09-07T09:00:00.000Z"); + }); + + it("WEEKLY cron without BYDAY defaults to Monday", () => { + const now = new Date("2026-09-04T12:00:00.000Z"); // Friday + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=WEEKLY;BYHOUR=9;BYMINUTE=0", now); + expect(nextRunAt).toBe("2026-09-07T09:00:00.000Z"); + }); + + it("MONTHLY cron behaves like DAILY (advances one day if the time passed)", () => { + const now = new Date("2026-09-04T10:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=MONTHLY;BYHOUR=9;BYMINUTE=0", now); + expect(nextRunAt).toBe("2026-09-05T09:00:00.000Z"); + }); + + it("with no recognizable FREQ, treats it like a daily run at the current UTC time", () => { + const now = new Date("2026-09-04T12:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "", now); + // No FREQ match means the DAILY/MONTHLY branch applies: the "target + // time" defaults to now's own UTC hour/minute, which counts as already + // passed, so it rolls to the same time tomorrow. + expect(nextRunAt).toBe("2026-09-05T12:00:00.000Z"); + }); + + it("parses cron case-insensitively", () => { + const now = new Date("2026-09-04T06:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "freq=daily;byhour=9;byminute=0", now); + expect(nextRunAt).toBe("2026-09-04T09:00:00.000Z"); + }); + + it("is unaffected by the local timezone (computes purely in UTC)", () => { + const now = new Date("2026-09-04T10:00:00.000Z"); + const nextRunAt = computeNextRunAt("cron", 60, "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", now); + // Regardless of process.env.TZ, BYHOUR=9 means 09:00 UTC. + expect(nextRunAt.endsWith("T09:00:00.000Z")).toBe(true); + }); +}); diff --git a/apps/web/src/components/ScheduleDialog.logic.ts b/apps/web/src/components/ScheduleDialog.logic.ts new file mode 100644 index 00000000..669855ef --- /dev/null +++ b/apps/web/src/components/ScheduleDialog.logic.ts @@ -0,0 +1,72 @@ +// Pure schedule-preview logic for ScheduleDialog, split out so it can be unit +// tested without the component's atom/settings dependencies. + +export type ScheduleMode = "interval" | "cron"; + +export interface SchedulePreset { + readonly label: string; + readonly mode: ScheduleMode; + readonly intervalMinutes?: number; + readonly cron?: string; +} + +export const PRESETS: ReadonlyArray = [ + { label: "Every hour", mode: "interval", intervalMinutes: 60 }, + { label: "Daily at 9am", mode: "cron", cron: "FREQ=DAILY;BYHOUR=9;BYMINUTE=0" }, + { label: "Weekly Monday 9am", mode: "cron", cron: "FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=0" }, +]; + +const WEEKDAY_TO_JS_DAY: Record = { + MO: 1, + TU: 2, + WE: 3, + TH: 4, + FR: 5, + SA: 6, + SU: 0, +}; + +/** + * Preview the next run time for an interval- or cron-based schedule, from + * `now`. Mirrors the server-supported RRULE subset in the server's + * `ScheduleReactor.computeNextRunAt` (FREQ=DAILY|WEEKLY|MONTHLY|HOURLY with + * optional BYHOUR/BYMINUTE/BYDAY); the server recomputes authoritatively on + * each fire, this is only a client-side estimate for the dialog's preview. + */ +export function computeNextRunAt( + mode: ScheduleMode, + intervalMinutes: number, + cron: string, + now: Date = new Date(), +): string { + if (mode === "interval") { + const ms = Math.max(1, intervalMinutes) * 60_000; + return new Date(now.getTime() + ms).toISOString(); + } + const upper = cron.toUpperCase(); + const freq = /FREQ=(DAILY|WEEKLY|MONTHLY|HOURLY)/.exec(upper)?.[1]; + if (freq === "HOURLY") { + return new Date(now.getTime() + 60 * 60 * 1000).toISOString(); + } + // Read/write in UTC (not local time) to match the server's + // ScheduleReactor.computeNextRunAt, which uses DateTime.getPartUtc. + const byHour = Number(/BYHOUR=(\d+)/.exec(upper)?.[1] ?? now.getUTCHours()); + const byMinute = Number(/BYMINUTE=(\d+)/.exec(upper)?.[1] ?? now.getUTCMinutes()); + const next = new Date(now); + next.setUTCMinutes(byMinute, 0, 0); + next.setUTCHours(byHour); + + if (freq === "WEEKLY") { + const weekday = /BYDAY=(MO|TU|WE|TH|FR|SA|SU)/.exec(upper)?.[1]; + const targetDay = weekday === undefined ? 1 : (WEEKDAY_TO_JS_DAY[weekday] ?? 1); + const currentDay = now.getUTCDay(); + let daysAhead = targetDay - currentDay; + if (daysAhead < 0) daysAhead += 7; + if (daysAhead === 0 && next.getTime() <= now.getTime()) daysAhead = 7; + next.setUTCDate(next.getUTCDate() + daysAhead); + } else if (next.getTime() <= now.getTime()) { + // DAILY or MONTHLY: advance to the next period. + next.setUTCDate(next.getUTCDate() + 1); + } + return next.toISOString(); +} diff --git a/apps/web/src/components/ScheduleDialog.tsx b/apps/web/src/components/ScheduleDialog.tsx new file mode 100644 index 00000000..517ed7e4 --- /dev/null +++ b/apps/web/src/components/ScheduleDialog.tsx @@ -0,0 +1,252 @@ +import type { ModelSelection, ScopedThreadRef, ThreadSchedule } from "@helmcode/contracts"; +import { createModelSelection } from "@helmcode/shared/model"; +import { useAtomValue } from "@effect/atom-react"; +import { useCallback, useEffect, useMemo, useState } from "react"; + +import { ProviderModelPicker } from "./chat/ProviderModelPicker"; +import { computeNextRunAt, PRESETS, type SchedulePreset } from "./ScheduleDialog.logic"; +import { Button } from "./ui/button"; +import { Checkbox } from "./ui/checkbox"; +import { + Dialog, + DialogDescription, + DialogFooter, + DialogHeader, + DialogPanel, + DialogPopup, + DialogTitle, +} from "./ui/dialog"; +import { Input } from "./ui/input"; +import { Label } from "./ui/label"; +import { RadioGroup, RadioGroupItem } from "./ui/radio-group"; +import { Textarea } from "./ui/textarea"; +import { usePrimarySettings } from "../hooks/useSettings"; +import { + applyProviderInstanceSettings, + deriveProviderInstanceEntries, + resolveDefaultProviderModelSelection, + sortProviderInstanceEntries, +} from "../providerInstances"; +import { getCustomModelOptionsByInstance } from "../modelSelection"; +import { primaryServerProvidersAtom } from "../state/server"; + +interface ScheduleDialogProps { + threadRef: ScopedThreadRef; + open: boolean; + onClose: () => void; + onSave: (threadRef: ScopedThreadRef, schedule: ThreadSchedule) => Promise; +} + +type ScheduleMode = "interval" | "cron"; + +const DEFAULT_PROMPT = "Continue where you left off."; + +export function ScheduleDialog({ threadRef, open, onClose, onSave }: ScheduleDialogProps) { + const [mode, setMode] = useState("interval"); + const [intervalMinutes, setIntervalMinutes] = useState(60); + const [cron, setCron] = useState("FREQ=DAILY;BYHOUR=9;BYMINUTE=0"); + const [prompt, setPrompt] = useState(DEFAULT_PROMPT); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(null); + const [overrideModel, setOverrideModel] = useState(false); + const [modelSelection, setModelSelection] = useState(null); + + const settings = usePrimarySettings(); + const serverProviders = useAtomValue(primaryServerProvidersAtom); + const instanceEntries = useMemo( + () => + sortProviderInstanceEntries( + applyProviderInstanceSettings(deriveProviderInstanceEntries(serverProviders), settings), + ), + [serverProviders, settings], + ); + const modelOptionsByInstance = useMemo( + () => getCustomModelOptionsByInstance(settings, serverProviders), + [serverProviders, settings], + ); + const resolvedSelection = resolveDefaultProviderModelSelection(serverProviders, modelSelection); + + useEffect(() => { + if (!open) return; + setMode("interval"); + setIntervalMinutes(60); + setCron("FREQ=DAILY;BYHOUR=9;BYMINUTE=0"); + setPrompt(DEFAULT_PROMPT); + setError(null); + setSaving(false); + setOverrideModel(false); + setModelSelection(null); + }, [open]); + + const nextRunAt = useMemo( + () => computeNextRunAt(mode, intervalMinutes, cron), + [mode, intervalMinutes, cron], + ); + + const applyPreset = useCallback((preset: SchedulePreset) => { + setMode(preset.mode); + if (preset.intervalMinutes !== undefined) setIntervalMinutes(preset.intervalMinutes); + if (preset.cron !== undefined) setCron(preset.cron); + }, []); + + const handleSave = useCallback(async () => { + const trimmedPrompt = prompt.trim(); + if (trimmedPrompt.length === 0) { + setError("Prompt cannot be empty."); + return; + } + const schedule: ThreadSchedule = { + enabled: true, + cron: mode === "cron" ? cron : null, + intervalMs: mode === "interval" ? Math.max(1, intervalMinutes) * 60_000 : null, + prompt: trimmedPrompt, + nextRunAt, + createdAt: new Date().toISOString(), + ...(overrideModel && resolvedSelection + ? { + modelSelection: createModelSelection( + resolvedSelection.instanceId, + resolvedSelection.model, + ), + } + : {}), + }; + setSaving(true); + setError(null); + try { + await onSave(threadRef, schedule); + onClose(); + } catch (err) { + setError(err instanceof Error ? err.message : "Failed to save schedule."); + } finally { + setSaving(false); + } + }, [ + cron, + intervalMinutes, + mode, + nextRunAt, + onClose, + onSave, + overrideModel, + prompt, + resolvedSelection, + threadRef, + ]); + + return ( + (nextOpen ? undefined : onClose())}> + + + Schedule this thread + + Automatically send a message and start a turn on this thread at a recurring interval. + + + +
+ +