diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index 71ef59a0910c..5a4e885e308f 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 { TaskFireReactor } from "../src/orchestration/Services/TaskFireReactor.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(TaskFireReactor, { + start: () => Effect.void, + drain: Effect.void, + }), + ), Layer.provideMerge( Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, { publishThread: () => Effect.void, diff --git a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts index 78a33364f5a3..de163c4b7fab 100644 --- a/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts +++ b/apps/server/integration/orphanedProviderSessionStartup.integration.test.ts @@ -28,6 +28,7 @@ import { OrchestrationLayerLive } from "../src/orchestration/runtimeLayer.ts"; import * as OrchestrationEngine from "../src/orchestration/Services/OrchestrationEngine.ts"; import * as OrchestrationReactor from "../src/orchestration/Services/OrchestrationReactor.ts"; import * as ProjectionSnapshotQuery from "../src/orchestration/Services/ProjectionSnapshotQuery.ts"; +import * as TaskScheduler from "../src/orchestration/Services/TaskScheduler.ts"; import { makeSqlitePersistenceLive } from "../src/persistence/Layers/Sqlite.ts"; import * as ProviderSessionRuntime from "../src/persistence/ProviderSessionRuntime.ts"; import * as ExternalLauncher from "../src/process/externalLauncher.ts"; @@ -75,6 +76,10 @@ const startupDependencies = Layer.mergeAll( Layer.succeed(ProviderSessionReaper.ProviderSessionReaper, { start: () => Effect.void, }), + Layer.succeed(TaskScheduler.TaskScheduler, { + start: () => Effect.void, + tick: () => Effect.succeed(0), + }), ServerLifecycleEvents.layer, Layer.succeed(ServerEnvironment.ServerEnvironment, { getEnvironmentId: Effect.succeed(EnvironmentId.make("environment-startup-orphan")), diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 70227cdd4ebf..3faa75574175 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -26,6 +26,7 @@ export const RPC_REQUIRED_SCOPES = { [ORCHESTRATION_WS_METHODS.getTurnDiff]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.getFullThreadDiff]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.searchThreads]: AuthOrchestrationReadScope, + [ORCHESTRATION_WS_METHODS.listTasks]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.subscribeShell]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.subscribeThread]: AuthOrchestrationReadScope, diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index fe093c451e25..7e74e96e1414 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -109,6 +109,8 @@ describe("CheckpointDiffQuery.layer", () => { getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), ), ); @@ -203,6 +205,8 @@ describe("CheckpointDiffQuery.layer", () => { getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), ), ); @@ -287,6 +291,8 @@ describe("CheckpointDiffQuery.layer", () => { getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), ), ); @@ -356,6 +362,8 @@ describe("CheckpointDiffQuery.layer", () => { getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), ), ); @@ -410,6 +418,8 @@ describe("CheckpointDiffQuery.layer", () => { getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), ), ); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 45dc0ee9cfd5..1e86efdf1f74 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -151,6 +151,7 @@ export const make = Effect.gen(function* () { threadSnooze: true, threadPinning: true, threadPinReorder: true, + taskScheduling: true, threadTitleRegeneration: true, ...(serverSelfUpdate === null ? {} : { serverSelfUpdate }), ...(serverSelfUpdate === "boot-service" ? { serverSelfUpdateProgress: true } : {}), diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 382c253fe60b..db6e1f5e4755 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -116,6 +116,7 @@ describe("OrchestrationEngine", () => { }; const projectionSnapshot = { + tasks: [], snapshotSequence: 7, updatedAt: "2026-03-03T00:00:04.000Z", projects: [ @@ -208,6 +209,8 @@ describe("OrchestrationEngine", () => { getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), ), Layer.provide( diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 423a44a6ff15..a1c49180ec2e 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -3,6 +3,7 @@ import type { OrchestrationEvent, OrchestrationReadModel, ProjectId, + TaskId, ThreadId, } from "@t3tools/contracts"; import { OrchestrationCommand } from "@t3tools/contracts"; @@ -61,8 +62,8 @@ interface CommandEnvelope { } function commandToAggregateRef(command: OrchestrationCommand): { - readonly aggregateKind: "project" | "thread"; - readonly aggregateId: ProjectId | ThreadId; + readonly aggregateKind: "project" | "thread" | "task"; + readonly aggregateId: ProjectId | ThreadId | TaskId; } { switch (command.type) { case "project.create": @@ -72,6 +73,13 @@ function commandToAggregateRef(command: OrchestrationCommand): { aggregateKind: "project", aggregateId: command.projectId, }; + case "task.schedule": + case "task.cancel": + case "task.fire": + return { + aggregateKind: "task", + aggregateId: command.taskId, + }; default: return { aggregateKind: "thread", diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts index 300d1526bb9a..e4f46ba175f1 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 { TaskFireReactor } from "../Services/TaskFireReactor.ts"; import { OrchestrationReactor } from "../Services/OrchestrationReactor.ts"; import { makeOrchestrationReactor } from "./OrchestrationReactor.ts"; import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; @@ -23,7 +24,7 @@ describe("OrchestrationReactor", () => { runtime = null; }); - it("starts provider ingestion, provider command, checkpoint, and thread deletion reactors", async () => { + it("starts provider ingestion, provider command, checkpoint, thread deletion, and task fire reactors", async () => { const started: string[] = []; runtime = ManagedRuntime.make( @@ -64,6 +65,15 @@ describe("OrchestrationReactor", () => { drain: Effect.void, }), ), + Layer.provideMerge( + Layer.succeed(TaskFireReactor, { + start: () => { + started.push("task-fire-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", + "task-fire-reactor", "agent-awareness-relay", ]); diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts index fb7543e31af0..35a8d21d97ee 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.ts @@ -8,6 +8,7 @@ import { import { CheckpointReactor } from "../Services/CheckpointReactor.ts"; import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; +import { TaskFireReactor } from "../Services/TaskFireReactor.ts"; import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; import * as AgentAwarenessRelay from "../../relay/AgentAwarenessRelay.ts"; @@ -16,6 +17,7 @@ export const makeOrchestrationReactor = Effect.gen(function* () { const providerCommandReactor = yield* ProviderCommandReactor; const checkpointReactor = yield* CheckpointReactor; const threadDeletionReactor = yield* ThreadDeletionReactor; + const taskFireReactor = yield* TaskFireReactor; 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* taskFireReactor.start(); yield* agentAwarenessRelay.start(); }); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index e9a625dd91cf..311c58461a9d 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -3,6 +3,7 @@ import { type ChatAttachment, type OrchestrationEvent, type OrchestrationSessionStatus, + TaskScheduleSpec, ThreadId, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -10,6 +11,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/unstable/sql/SqlClient"; @@ -34,6 +36,7 @@ import { ProjectionTurnRepository, } from "../../persistence/Services/ProjectionTurns.ts"; import { ProjectionThreadRepository } from "../../persistence/Services/ProjectionThreads.ts"; +import { ProjectionTaskRepository } from "../../persistence/Services/ProjectionTasks.ts"; import { ProjectionPendingApprovalRepositoryLive } from "../../persistence/Layers/ProjectionPendingApprovals.ts"; import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/ProjectionProjects.ts"; import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts"; @@ -43,6 +46,7 @@ import { ProjectionThreadProposedPlanRepositoryLive } from "../../persistence/La import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts"; import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts"; +import { ProjectionTaskRepositoryLive } from "../../persistence/Layers/ProjectionTasks.ts"; import { ServerConfig } from "../../config.ts"; import { OrchestrationProjectionPipeline, @@ -55,6 +59,8 @@ import { toSafeThreadAttachmentSegment, } from "../../attachmentStore.ts"; +const encodeTaskScheduleJson = Schema.encodeEffect(Schema.fromJsonString(TaskScheduleSpec)); + export const ORCHESTRATION_PROJECTOR_NAMES = { projects: "projection.projects", threads: "projection.threads", @@ -65,6 +71,7 @@ export const ORCHESTRATION_PROJECTOR_NAMES = { threadTurns: "projection.thread-turns", checkpoints: "projection.checkpoints", pendingApprovals: "projection.pending-approvals", + tasks: "projection.tasks", } as const; type ProjectorName = @@ -480,6 +487,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository; const projectionTurnRepository = yield* ProjectionTurnRepository; const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository; + const projectionTaskRepository = yield* ProjectionTaskRepository; const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; @@ -1606,6 +1614,69 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } }); + const applyTasksProjection: ProjectorDefinition["apply"] = Effect.fn("applyTasksProjection")( + function* (event, _attachmentSideEffects) { + switch (event.type) { + case "task.scheduled": { + // The payload was schema-decoded off the wire, so encoding cannot + // fail here; a failure is a defect, not a projection error. + const scheduleJson = yield* encodeTaskScheduleJson(event.payload.schedule).pipe( + Effect.orDie, + ); + yield* projectionTaskRepository.upsert({ + taskId: event.payload.taskId, + projectId: event.payload.projectId, + threadId: event.payload.threadId, + name: event.payload.name ?? null, + prompt: event.payload.prompt, + scheduleJson, + createdAt: event.payload.createdAt, + updatedAt: event.payload.updatedAt, + lastFiredAt: null, + nextFireAt: event.payload.nextFireAt, + cancelledAt: null, + }); + return; + } + + case "task.fired": { + const existingRow = yield* projectionTaskRepository.getById({ + taskId: event.payload.taskId, + }); + if (Option.isNone(existingRow)) { + return; + } + yield* projectionTaskRepository.upsert({ + ...existingRow.value, + lastFiredAt: event.payload.firedAt, + nextFireAt: event.payload.nextFireAt, + updatedAt: event.payload.updatedAt, + }); + return; + } + + case "task.cancelled": { + const existingRow = yield* projectionTaskRepository.getById({ + taskId: event.payload.taskId, + }); + if (Option.isNone(existingRow)) { + return; + } + yield* projectionTaskRepository.upsert({ + ...existingRow.value, + nextFireAt: null, + cancelledAt: event.payload.cancelledAt, + updatedAt: event.payload.updatedAt, + }); + return; + } + + default: + return; + } + }, + ); + const projectors: ReadonlyArray = [ { name: ORCHESTRATION_PROJECTOR_NAMES.projects, @@ -1643,6 +1714,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti name: ORCHESTRATION_PROJECTOR_NAMES.threads, apply: applyThreadsProjection, }, + { + name: ORCHESTRATION_PROJECTOR_NAMES.tasks, + apply: applyTasksProjection, + }, ]; const runProjectorForEvent = Effect.fn("runProjectorForEvent")(function* ( @@ -1746,4 +1821,5 @@ export const OrchestrationProjectionPipelineLive = Layer.effect( Layer.provideMerge(ProjectionTurnRepositoryLive), Layer.provideMerge(ProjectionPendingApprovalRepositoryLive), Layer.provideMerge(ProjectionStateRepositoryLive), + Layer.provideMerge(ProjectionTaskRepositoryLive), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index c6c5ad1d7e8c..1727747f9ac1 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -7,11 +7,13 @@ import { OrchestrationCheckpointFile, OrchestrationProposedPlanId, OrchestrationReadModel, + OrchestrationTask, OrchestrationThreadSearchSource, OrchestrationShellSnapshot, OrchestrationThread, OrchestrationThreadDetailSnapshot, ProjectScript, + TaskScheduleSpec, TurnId, type OrchestrationCheckpointSummary, type OrchestrationLatestTurn, @@ -38,6 +40,7 @@ import * as SqlSchema from "effect/unstable/sql/SqlSchema"; import { isPersistenceError, + PersistenceDecodeError, toPersistenceDecodeError, toPersistenceSqlError, type ProjectionRepositoryError, @@ -52,6 +55,7 @@ import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionTh import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts"; import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts"; import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts"; +import { ProjectionTask } from "../../persistence/Services/ProjectionTasks.ts"; import { decodeThreadDetailPageCursor, encodeThreadDetailPageCursor, @@ -98,6 +102,21 @@ const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( }), ); const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession; +const ProjectionTaskDbRowSchema = ProjectionTask; + +const decodeTaskScheduleJson = Schema.decodeUnknownSync(Schema.fromJsonString(TaskScheduleSpec)); +const decodeTaskSchema = Schema.decodeUnknownEffect(OrchestrationTask); + +// Corrupt schedule JSON is a corrupt-database condition: fail loudly rather +// than silently dropping the task from the read model. The schema parse also +// validates the decoded spec. +function parseScheduleJson(taskId: string, scheduleJson: string): TaskScheduleSpec { + try { + return decodeTaskScheduleJson(scheduleJson); + } catch (cause) { + throw new Error(`invalid schedule JSON for task ${taskId}`, { cause }); + } +} const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields( Struct.assign({ files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)), @@ -444,6 +463,76 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const listTaskRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionTaskDbRowSchema, + execute: () => + sql` + SELECT + task_id AS "taskId", + project_id AS "projectId", + thread_id AS "threadId", + name, + prompt, + schedule_json AS "scheduleJson", + created_at AS "createdAt", + updated_at AS "updatedAt", + last_fired_at AS "lastFiredAt", + next_fire_at AS "nextFireAt", + cancelled_at AS "cancelledAt" + FROM projection_tasks + ORDER BY created_at ASC, task_id ASC + `, + }); + + const listTaskRowsByProject = SqlSchema.findAll({ + Request: Schema.Struct({ projectId: ProjectId }), + Result: ProjectionTaskDbRowSchema, + execute: ({ projectId }) => + sql` + SELECT + task_id AS "taskId", + project_id AS "projectId", + thread_id AS "threadId", + name, + prompt, + schedule_json AS "scheduleJson", + created_at AS "createdAt", + updated_at AS "updatedAt", + last_fired_at AS "lastFiredAt", + next_fire_at AS "nextFireAt", + cancelled_at AS "cancelledAt" + FROM projection_tasks + WHERE project_id = ${projectId} + ORDER BY created_at ASC, task_id ASC + `, + }); + + const listDueTaskRows = SqlSchema.findAll({ + Request: Schema.Struct({ nowIso: IsoDateTime }), + Result: ProjectionTaskDbRowSchema, + execute: ({ nowIso }) => + sql` + SELECT + task_id AS "taskId", + project_id AS "projectId", + thread_id AS "threadId", + name, + prompt, + schedule_json AS "scheduleJson", + created_at AS "createdAt", + updated_at AS "updatedAt", + last_fired_at AS "lastFiredAt", + next_fire_at AS "nextFireAt", + cancelled_at AS "cancelledAt" + FROM projection_tasks + WHERE next_fire_at IS NOT NULL + AND cancelled_at IS NULL + AND next_fire_at <= ${nowIso} + ORDER BY next_fire_at ASC, task_id ASC + `, + }); + const listActiveThreadRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadDbRowSchema, @@ -1517,6 +1606,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), + listTaskRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getSnapshot:listTasks:query", + "ProjectionSnapshotQuery.getSnapshot:listTasks:decodeRows", + ), + ), + ), ]), ) .pipe( @@ -1531,6 +1628,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { checkpointRows, latestTurnRows, stateRows, + taskRows, ]) => Effect.gen(function* () { const messagesByThread = new Map>(); @@ -1717,6 +1815,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snapshotSequence: computeSnapshotSequence(stateRows), projects, threads, + tasks: yield* decodeTaskRows(taskRows), updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z", }; @@ -1787,11 +1886,27 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), + listTaskRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getCommandReadModel:listTasks:query", + "ProjectionSnapshotQuery.getCommandReadModel:listTasks:decodeRows", + ), + ), + ), ]), ) .pipe( Effect.flatMap( - ([projectRows, threadRows, proposedPlanRows, sessionRows, latestTurnRows, stateRows]) => + ([ + projectRows, + threadRows, + proposedPlanRows, + sessionRows, + latestTurnRows, + stateRows, + taskRows, + ]) => Effect.sync(() => { let updatedAt: string | null = null; const projects: OrchestrationProject[] = []; @@ -1925,6 +2040,24 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snapshotSequence: computeSnapshotSequence(stateRows), projects, threads, + tasks: taskRows.map((row) => { + if (!row) { + throw new Error("unexpected missing task row"); + } + return { + taskId: row.taskId, + projectId: row.projectId, + threadId: row.threadId, + ...(row.name !== null ? { name: row.name } : {}), + prompt: row.prompt, + schedule: parseScheduleJson(row.taskId, row.scheduleJson), + createdAt: row.createdAt, + updatedAt: row.updatedAt, + lastFiredAt: row.lastFiredAt, + nextFireAt: row.nextFireAt, + cancelledAt: row.cancelledAt, + } satisfies OrchestrationTask; + }), updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z", } satisfies OrchestrationReadModel; }), @@ -2281,6 +2414,64 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { }; }); + const decodeTaskRows = (rows: ReadonlyArray) => + Effect.forEach( + rows, + (row) => + Effect.try({ + try: () => parseScheduleJson(row.taskId, row.scheduleJson), + catch: (cause) => + new PersistenceDecodeError({ + operation: "ProjectionSnapshotQuery.decodeTaskRows:scheduleJson", + issue: cause instanceof Error ? cause.message : String(cause), + ...(cause !== undefined ? { cause } : {}), + }), + }).pipe( + Effect.flatMap((schedule) => + decodeTaskSchema({ + taskId: row.taskId, + projectId: row.projectId, + threadId: row.threadId, + ...(row.name !== null ? { name: row.name } : {}), + prompt: row.prompt, + schedule, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + lastFiredAt: row.lastFiredAt, + nextFireAt: row.nextFireAt, + cancelledAt: row.cancelledAt, + }).pipe( + Effect.mapError( + toPersistenceDecodeError("ProjectionSnapshotQuery.decodeTaskRows:decode"), + ), + ), + ), + ), + { concurrency: 1 }, + ); + + const listTasks: ProjectionSnapshotQueryShape["listTasks"] = (projectId) => + listTaskRowsByProject({ projectId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.listTasks:query", + "ProjectionSnapshotQuery.listTasks:decodeRows", + ), + ), + Effect.flatMap(decodeTaskRows), + ); + + const listDueTasks: ProjectionSnapshotQueryShape["listDueTasks"] = (nowIso) => + listDueTaskRows({ nowIso }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.listDueTasks:query", + "ProjectionSnapshotQuery.listDueTasks:decodeRows", + ), + ), + Effect.flatMap(decodeTaskRows), + ); + const getActiveProjectByWorkspaceRoot: ProjectionSnapshotQueryShape["getActiveProjectByWorkspaceRoot"] = (workspaceRoot) => getActiveProjectRowByWorkspaceRoot({ workspaceRoot }).pipe( @@ -2817,6 +3008,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { getShellSnapshot, getArchivedShellSnapshot, searchThreads, + listTasks, + listDueTasks, getSnapshotSequence, getCounts, getActiveProjectByWorkspaceRoot, diff --git a/apps/server/src/orchestration/Layers/TaskFireReactor.ts b/apps/server/src/orchestration/Layers/TaskFireReactor.ts new file mode 100644 index 000000000000..2f7dbd56a7a3 --- /dev/null +++ b/apps/server/src/orchestration/Layers/TaskFireReactor.ts @@ -0,0 +1,85 @@ +import { CommandId, MessageId, type OrchestrationEvent } from "@t3tools/contracts"; +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import * as Cause from "effect/Cause"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { TaskFireReactor, type TaskFireReactorShape } from "../Services/TaskFireReactor.ts"; +import { forkParked } from "../../serverActivation.ts"; + +type TaskFiredEvent = Extract; + +const make = Effect.gen(function* () { + const orchestrationEngine = yield* OrchestrationEngineService; + const crypto = yield* Crypto.Crypto; + + const processTaskFired = Effect.fn("processTaskFired")(function* (event: TaskFiredEvent) { + const { taskId, threadId, prompt, dueAt, firedAt } = event.payload; + // Deterministic per due slot: a reactor retry after a crash re-dispatches + // the identical command, which the engine's receipt dedupe collapses. + const turnCommandId = CommandId.make(`server:task-turn:${taskId}:${dueAt}`); + const messageId = yield* crypto.randomUUIDv4.pipe(Effect.map(MessageId.make)); + + yield* orchestrationEngine.dispatch({ + type: "thread.turn.start", + commandId: turnCommandId, + threadId, + message: { + messageId, + role: "user", + text: prompt, + attachments: [], + }, + runtimeMode: event.payload.runtimeMode, + interactionMode: event.payload.interactionMode, + createdAt: firedAt, + }); + + yield* Effect.logDebug("task.fire-reactor.turn-started", { + taskId, + threadId, + dueAt, + }); + }); + + const processTaskFiredSafely = (event: TaskFiredEvent) => + processTaskFired(event).pipe( + Effect.catchCause((cause) => { + if (Cause.hasInterruptsOnly(cause)) { + return Effect.failCause(cause); + } + // A fire whose anchor thread vanished (deleted between scheduling and + // firing) logs and moves on: the task's nextFireAt already advanced, + // so it stays armed for its next slot. + return Effect.logWarning("task fire reactor failed to start turn", { + eventType: event.type, + taskId: event.payload.taskId, + threadId: event.payload.threadId, + cause: Cause.pretty(cause), + }); + }), + ); + + const worker = yield* makeDrainableWorker(processTaskFiredSafely); + + const start: TaskFireReactorShape["start"] = Effect.fn("start")(function* () { + yield* forkParked( + Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { + if (event.type !== "task.fired") { + return Effect.void; + } + return worker.enqueue(event); + }), + ); + }); + + return { + start, + drain: worker.drain, + } satisfies TaskFireReactorShape; +}); + +export const TaskFireReactorLive = Layer.effect(TaskFireReactor, make); diff --git a/apps/server/src/orchestration/Layers/TaskScheduler.test.ts b/apps/server/src/orchestration/Layers/TaskScheduler.test.ts new file mode 100644 index 000000000000..2b5731567c16 --- /dev/null +++ b/apps/server/src/orchestration/Layers/TaskScheduler.test.ts @@ -0,0 +1,128 @@ +import { CommandId, ProjectId, TaskId, ThreadId, type OrchestrationTask } from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import type { DeepPartial } from "@t3tools/shared/Struct"; + +import { + OrchestrationEngineService, + type OrchestrationEngineShape, +} from "../Services/OrchestrationEngine.ts"; +import { + ProjectionSnapshotQuery, + type ProjectionSnapshotQueryShape, +} from "../Services/ProjectionSnapshotQuery.ts"; +import { TaskScheduler } from "../Services/TaskScheduler.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import type { ServerSettings } from "@t3tools/contracts"; +import { makeTaskSchedulerLive } from "./TaskScheduler.ts"; + +function makeDueTask(input: { + readonly taskId: string; + readonly nextFireAt: string; +}): OrchestrationTask { + return { + taskId: TaskId.make(input.taskId), + projectId: ProjectId.make("project-1"), + threadId: ThreadId.make("thread-1"), + prompt: "Run the nightly checks", + schedule: { kind: "interval", everyMs: 60 * 60 * 1000 }, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + lastFiredAt: null, + nextFireAt: input.nextFireAt, + cancelledAt: null, + }; +} + +// Shared harness state: tests run sequentially, so each one resets what is +// due and inspects what got dispatched. +const state = { + dueTasks: [] as OrchestrationTask[], + dispatched: [] as Array<{ type: string; commandId: CommandId; dueAt: string }>, +}; + +const fakeProjectionLayer = Layer.succeed(ProjectionSnapshotQuery, { + listDueTasks: () => Effect.succeed(state.dueTasks), +} as unknown as ProjectionSnapshotQueryShape); + +const fakeEngineLayer = Layer.succeed(OrchestrationEngineService, { + dispatch: (command: { type: string; commandId: CommandId; dueAt?: string }) => + Effect.sync(() => { + state.dispatched.push(command as { type: string; commandId: CommandId; dueAt: string }); + return { sequence: 1 }; + }), +} as unknown as OrchestrationEngineShape); + +const schedulerLayerWithSettings = (settingsOverrides?: DeepPartial) => + makeTaskSchedulerLive({ tickIntervalMs: 60_000 }).pipe( + Layer.provideMerge(ServerSettingsService.layerTest(settingsOverrides)), + Layer.provideMerge(fakeProjectionLayer), + Layer.provideMerge(fakeEngineLayer), + ); + +it.layer(schedulerLayerWithSettings().pipe(Layer.provideMerge(NodeServices.layer)))( + "task scheduler", + (it) => { + it.effect("dispatches a deterministic task.fire per due slot", () => + Effect.gen(function* () { + state.dispatched = []; + state.dueTasks = [ + makeDueTask({ taskId: "task-a", nextFireAt: "2026-01-01T00:00:00.000Z" }), + makeDueTask({ taskId: "task-b", nextFireAt: "2026-01-01T00:05:00.000Z" }), + ]; + + const scheduler = yield* TaskScheduler; + const firedCount = yield* scheduler.tick(); + + expect(firedCount).toBe(2); + expect(state.dispatched).toHaveLength(2); + // Deterministic ids keyed by the due slot so crash-retries collapse + // into the engine's idempotent command receipts. + expect(state.dispatched[0]?.type).toBe("task.fire"); + expect(state.dispatched[0]?.commandId).toBe( + CommandId.make("server:task-fire:task-a:2026-01-01T00:00:00.000Z"), + ); + expect(state.dispatched[1]?.commandId).toBe( + CommandId.make("server:task-fire:task-b:2026-01-01T00:05:00.000Z"), + ); + }), + ); + + it.effect("returns zero and dispatches nothing when no task is due", () => + Effect.gen(function* () { + state.dispatched = []; + state.dueTasks = []; + + const scheduler = yield* TaskScheduler; + const firedCount = yield* scheduler.tick(); + + expect(firedCount).toBe(0); + expect(state.dispatched).toHaveLength(0); + }), + ); + }, +); + +it.layer( + schedulerLayerWithSettings({ enableScheduledTasks: false }).pipe( + Layer.provideMerge(NodeServices.layer), + ), +)("task scheduler with the feature switch off", (it) => { + it.effect("skips the due scan and dispatches nothing", () => + Effect.gen(function* () { + state.dispatched = []; + state.dueTasks = [ + makeDueTask({ taskId: "task-off", nextFireAt: "2026-01-01T00:00:00.000Z" }), + ]; + + const scheduler = yield* TaskScheduler; + const firedCount = yield* scheduler.tick(); + + expect(firedCount).toBe(0); + expect(state.dispatched).toHaveLength(0); + }), + ); +}); diff --git a/apps/server/src/orchestration/Layers/TaskScheduler.ts b/apps/server/src/orchestration/Layers/TaskScheduler.ts new file mode 100644 index 000000000000..ed3f059ff93c --- /dev/null +++ b/apps/server/src/orchestration/Layers/TaskScheduler.ts @@ -0,0 +1,131 @@ +import { CommandId, type TaskId } from "@t3tools/contracts"; +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 Schedule from "effect/Schedule"; + +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +import { TaskScheduler, type TaskSchedulerShape } from "../Services/TaskScheduler.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import { forkParked } from "../../serverActivation.ts"; + +const DEFAULT_TICK_INTERVAL_MS = 15 * 1000; + +export interface TaskSchedulerLiveOptions { + readonly tickIntervalMs?: number; +} + +// Fires are idempotent per due slot: the deterministic commandId collapses a +// crash-retry into the engine's existing command receipt instead of +// double-starting a turn. +const fireCommandId = (taskId: TaskId, nextFireAt: string): CommandId => + CommandId.make(`server:task-fire:${taskId}:${nextFireAt}`); + +const makeTaskScheduler = (options?: TaskSchedulerLiveOptions) => + Effect.gen(function* () { + const orchestrationEngine = yield* OrchestrationEngineService; + const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + // The feature switch is read per tick, not captured at startup, so + // toggling it in settings takes effect within one tick interval without + // a server restart. + const serverSettings = yield* ServerSettingsService; + + const tickIntervalMs = Math.max(1, options?.tickIntervalMs ?? DEFAULT_TICK_INTERVAL_MS); + + const tickEffect: TaskSchedulerShape["tick"] = Effect.fn("TaskScheduler.tick")(function* () { + // Disabled is the cheap path: no due-scan, no dispatch. Stored tasks + // are untouched — their nextFireAt keeps aging and fires coalesce into + // the first future slot when re-enabled (decider's drift anchoring). + // A failed settings read fails open: the tick behaves as before the + // switch existed rather than silently disabling automation. + const enabled = yield* serverSettings.getSettings.pipe( + Effect.map((settings) => settings.enableScheduledTasks), + Effect.catchCause((cause) => + Effect.logWarning("task.scheduler.settings-read-failed", { cause }).pipe(Effect.as(true)), + ), + ); + if (!enabled) { + return 0; + } + // ISO strings compare lexicographically, but the query uses SQLite + // string comparison on next_fire_at — pass the same normalized format + // the projector writes. + const nowIso = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + // A failed due-scan must not kill the loop: log and retry next tick. + const dueTasks = yield* projectionSnapshotQuery + .listDueTasks(nowIso) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("task.scheduler.due-scan-failed", { cause }).pipe(Effect.as([])), + ), + ); + + let dispatchedCount = 0; + for (const task of dueTasks) { + if (task.nextFireAt === null) { + continue; + } + yield* orchestrationEngine + .dispatch({ + type: "task.fire", + commandId: fireCommandId(task.taskId, task.nextFireAt), + taskId: task.taskId, + dueAt: nowIso, + }) + .pipe( + Effect.tap(() => { + dispatchedCount += 1; + return Effect.logDebug("task.scheduler.fired", { + taskId: task.taskId, + threadId: task.threadId, + nextFireAt: task.nextFireAt, + }); + }), + Effect.catchCause((cause) => + Effect.logWarning("task.scheduler.fire-failed", { + taskId: task.taskId, + cause, + }), + ), + ); + } + + if (dispatchedCount > 0) { + yield* Effect.logInfo("task.scheduler.tick-complete", { + dueCount: dueTasks.length, + dispatchedCount, + }); + } + return dispatchedCount; + }); + + const start: TaskSchedulerShape["start"] = () => + Effect.gen(function* () { + yield* forkParked( + tickEffect().pipe( + Effect.catchDefect((defect: unknown) => + Effect.logWarning("task.scheduler.tick-defect", { + defect, + }), + ), + Effect.repeat(Schedule.spaced(Duration.millis(tickIntervalMs))), + ), + ); + + yield* Effect.logInfo("task.scheduler.started", { + tickIntervalMs, + }); + }); + + return { + start, + tick: tickEffect, + } satisfies TaskSchedulerShape; + }); + +export const makeTaskSchedulerLive = (options?: TaskSchedulerLiveOptions) => + Layer.effect(TaskScheduler, makeTaskScheduler(options)); + +export const TaskSchedulerLive = makeTaskSchedulerLive(); diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 7e866cf89592..82f584a5a87d 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -27,6 +27,9 @@ import { ThreadApprovalResponseRequestedPayload as ContractsThreadApprovalResponseRequestedPayloadSchema, ThreadCheckpointRevertRequestedPayload as ContractsThreadCheckpointRevertRequestedPayloadSchema, ThreadSessionStopRequestedPayload as ContractsThreadSessionStopRequestedPayloadSchema, + TaskScheduledPayload as ContractsTaskScheduledPayloadSchema, + TaskFiredPayload as ContractsTaskFiredPayloadSchema, + TaskCancelledPayload as ContractsTaskCancelledPayloadSchema, } from "@t3tools/contracts"; // Server-internal alias surface, backed by contract schemas as the source of truth. @@ -64,3 +67,7 @@ export const ThreadApprovalResponseRequestedPayload = export const ThreadCheckpointRevertRequestedPayload = ContractsThreadCheckpointRevertRequestedPayloadSchema; export const ThreadSessionStopRequestedPayload = ContractsThreadSessionStopRequestedPayloadSchema; + +export const TaskScheduledPayload = ContractsTaskScheduledPayloadSchema; +export const TaskFiredPayload = ContractsTaskFiredPayloadSchema; +export const TaskCancelledPayload = ContractsTaskCancelledPayloadSchema; diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 0a00253a2285..66c1d2bcdfaa 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -15,6 +15,7 @@ import type { OrchestrationSearchThreadsInput, OrchestrationSearchThreadsResult, OrchestrationShellSnapshot, + OrchestrationTask, OrchestrationThread, OrchestrationThreadDetailSnapshot, OrchestrationThreadDetailWindow, @@ -105,6 +106,22 @@ export interface ProjectionSnapshotQueryShape { input: OrchestrationSearchThreadsInput, ) => Effect.Effect; + /** + * List scheduled tasks for a project, including cancelled ones so clients + * can render recent history without a second query. + */ + readonly listTasks: ( + projectId: ProjectId, + ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * List armed tasks whose nextFireAt has passed. Server-internal: consumed + * by the task scheduler tick loop. + */ + readonly listDueTasks: ( + nowIso: string, + ) => Effect.Effect, ProjectionRepositoryError>; + /** * Read the latest projection snapshot sequence without hydrating read-model * entities. diff --git a/apps/server/src/orchestration/Services/TaskFireReactor.ts b/apps/server/src/orchestration/Services/TaskFireReactor.ts new file mode 100644 index 000000000000..7df1d4ded063 --- /dev/null +++ b/apps/server/src/orchestration/Services/TaskFireReactor.ts @@ -0,0 +1,19 @@ +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; +import type * as Scope from "effect/Scope"; + +export interface TaskFireReactorShape { + /** + * Start consuming task.fired events within the provided scope. + */ + readonly start: () => Effect.Effect; + + /** + * Wait until every enqueued task.fired event has been processed. + */ + readonly drain: Effect.Effect; +} + +export class TaskFireReactor extends Context.Service()( + "t3/orchestration/Services/TaskFireReactor", +) {} diff --git a/apps/server/src/orchestration/Services/TaskScheduler.ts b/apps/server/src/orchestration/Services/TaskScheduler.ts new file mode 100644 index 000000000000..ceb3e3a6f6c5 --- /dev/null +++ b/apps/server/src/orchestration/Services/TaskScheduler.ts @@ -0,0 +1,21 @@ +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; +import type * as Scope from "effect/Scope"; + +export interface TaskSchedulerShape { + /** + * Start the background scheduler tick loop within the provided scope. + */ + readonly start: () => Effect.Effect; + + /** + * Run one tick immediately: dispatch `task.fire` for every armed task whose + * nextFireAt has passed. Returns the number of fires dispatched. Exposed as + * a deterministic seam for tests — never wait on the wall-clock loop. + */ + readonly tick: () => Effect.Effect; +} + +export class TaskScheduler extends Context.Service()( + "t3/orchestration/Services/TaskScheduler", +) {} diff --git a/apps/server/src/orchestration/commandInvariants.test.ts b/apps/server/src/orchestration/commandInvariants.test.ts index 52aac1f0c105..dbc3c921abb3 100644 --- a/apps/server/src/orchestration/commandInvariants.test.ts +++ b/apps/server/src/orchestration/commandInvariants.test.ts @@ -21,6 +21,7 @@ import { const now = "2026-01-01T00:00:00.000Z"; const readModel: OrchestrationReadModel = { + tasks: [], snapshotSequence: 2, updatedAt: now, projects: [ diff --git a/apps/server/src/orchestration/commandInvariants.ts b/apps/server/src/orchestration/commandInvariants.ts index b59ded77f4f4..cd294be2618f 100644 --- a/apps/server/src/orchestration/commandInvariants.ts +++ b/apps/server/src/orchestration/commandInvariants.ts @@ -2,8 +2,10 @@ import type { OrchestrationCommand, OrchestrationProject, OrchestrationReadModel, + OrchestrationTask, OrchestrationThread, ProjectId, + TaskId, ThreadId, } from "@t3tools/contracts"; import { normalizeProjectPathForComparison } from "@t3tools/shared/path"; @@ -182,3 +184,43 @@ export function requireNonNegativeInteger(input: { ), ); } + +export function findTaskById( + readModel: OrchestrationReadModel, + taskId: TaskId, +): OrchestrationTask | undefined { + return readModel.tasks.find((task) => task.taskId === taskId); +} + +export function requireTask(input: { + readonly readModel: OrchestrationReadModel; + readonly command: OrchestrationCommand; + readonly taskId: TaskId; +}): Effect.Effect { + const task = findTaskById(input.readModel, input.taskId); + if (task) { + return Effect.succeed(task); + } + return Effect.fail( + invariantError( + input.command.type, + `Task '${input.taskId}' does not exist for command '${input.command.type}'.`, + ), + ); +} + +export function requireTaskAbsent(input: { + readonly readModel: OrchestrationReadModel; + readonly command: OrchestrationCommand; + readonly taskId: TaskId; +}): Effect.Effect { + if (!findTaskById(input.readModel, input.taskId)) { + return Effect.void; + } + return Effect.fail( + invariantError( + input.command.type, + `Task '${input.taskId}' already exists and cannot be scheduled twice.`, + ), + ); +} diff --git a/apps/server/src/orchestration/decider.pinned.test.ts b/apps/server/src/orchestration/decider.pinned.test.ts index 4ad00ba994b4..fae548e4a7de 100644 --- a/apps/server/src/orchestration/decider.pinned.test.ts +++ b/apps/server/src/orchestration/decider.pinned.test.ts @@ -26,6 +26,7 @@ function makeReadModel(input: { return { snapshotSequence: 0, projects: [], + tasks: [], threads: [ { id: ThreadId.make("thread-1"), diff --git a/apps/server/src/orchestration/decider.scheduled-tasks.test.ts b/apps/server/src/orchestration/decider.scheduled-tasks.test.ts new file mode 100644 index 000000000000..57b4644f04c1 --- /dev/null +++ b/apps/server/src/orchestration/decider.scheduled-tasks.test.ts @@ -0,0 +1,350 @@ +import { + CommandId, + ProjectId, + ProviderInstanceId, + TaskId, + ThreadId, + type OrchestrationCommand, + type OrchestrationReadModel, + type OrchestrationTask, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; + +// The decider's clock is the Effect test clock, pinned to the epoch, so all +// "now" values inside decisions are 1970-01-01T00:00:00.000Z. +const NOW = "2026-01-01T00:00:00.000Z"; +const FUTURE = "1970-01-02T09:00:00.000Z"; +const PAST = "1969-12-31T09:00:00.000Z"; +const HOUR_MS = 60 * 60 * 1000; + +function makeThread( + input: { + readonly threadId?: ThreadId; + readonly deletedAt?: string | null; + } = {}, +) { + return { + id: input.threadId ?? ThreadId.make("thread-1"), + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + deletedAt: input.deletedAt ?? null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }; +} + +function makeReadModel(input: { + readonly threads?: ReturnType[]; + readonly tasks?: OrchestrationTask[]; +}): OrchestrationReadModel { + return { + snapshotSequence: 0, + projects: [ + { + id: ProjectId.make("project-1"), + title: "Project", + workspaceRoot: "/tmp/project-1", + defaultModelSelection: null, + scripts: [], + createdAt: NOW, + updatedAt: NOW, + deletedAt: null, + }, + ], + threads: input.threads ?? [makeThread()], + tasks: input.tasks ?? [], + updatedAt: NOW, + }; +} + +function makeTask(input: { + readonly schedule?: OrchestrationTask["schedule"]; + readonly nextFireAt?: string | null; + readonly cancelledAt?: string | null; + readonly updatedAt?: string; +}): OrchestrationTask { + return { + taskId: TaskId.make("task-1"), + projectId: ProjectId.make("project-1"), + threadId: ThreadId.make("thread-1"), + prompt: "Run the nightly checks", + schedule: input.schedule ?? { kind: "interval", everyMs: HOUR_MS }, + createdAt: NOW, + updatedAt: input.updatedAt ?? NOW, + lastFiredAt: null, + // Explicit null must survive: an optional chain would read it as absent. + nextFireAt: input.nextFireAt === undefined ? FUTURE : input.nextFireAt, + cancelledAt: input.cancelledAt ?? null, + }; +} + +const run = (command: OrchestrationCommand, readModel: OrchestrationReadModel) => + decideOrchestrationCommand({ command, readModel }); + +it.layer(NodeServices.layer)("scheduled task decider", (it) => { + it.effect("schedules a one-shot task with its absolute fire time", () => + Effect.gen(function* () { + const decided = yield* run( + { + type: "task.schedule", + commandId: CommandId.make("cmd-schedule-once"), + taskId: TaskId.make("task-once"), + projectId: ProjectId.make("project-1"), + threadId: ThreadId.make("thread-1"), + prompt: "Check the build", + schedule: { kind: "once", at: FUTURE }, + createdAt: NOW, + }, + makeReadModel({}), + ); + const events = Array.isArray(decided) ? decided : [decided]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("task.scheduled"); + if (events[0]?.type === "task.scheduled") { + expect(events[0].aggregateKind).toBe("task"); + expect(events[0].payload.nextFireAt).toBe(FUTURE); + } + }), + ); + + it.effect("anchors an interval task's first fire one period out", () => + Effect.gen(function* () { + const decided = yield* run( + { + type: "task.schedule", + commandId: CommandId.make("cmd-schedule-interval"), + taskId: TaskId.make("task-interval"), + projectId: ProjectId.make("project-1"), + threadId: ThreadId.make("thread-1"), + prompt: "Ping the deploy", + schedule: { kind: "interval", everyMs: HOUR_MS }, + createdAt: NOW, + }, + makeReadModel({}), + ); + const events = Array.isArray(decided) ? decided : [decided]; + expect(events[0]?.type).toBe("task.scheduled"); + if (events[0]?.type === "task.scheduled") { + expect(events[0].payload.nextFireAt).toBe("1970-01-01T01:00:00.000Z"); + } + }), + ); + + it.effect("rejects a one-shot time in the past", () => + Effect.gen(function* () { + const error = yield* run( + { + type: "task.schedule", + commandId: CommandId.make("cmd-schedule-past"), + taskId: TaskId.make("task-past"), + projectId: ProjectId.make("project-1"), + threadId: ThreadId.make("thread-1"), + prompt: "Too late", + schedule: { kind: "once", at: PAST }, + createdAt: NOW, + }, + makeReadModel({}), + ).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("rejects a duplicate task id", () => + Effect.gen(function* () { + const error = yield* run( + { + type: "task.schedule", + commandId: CommandId.make("cmd-schedule-dup"), + taskId: TaskId.make("task-1"), + projectId: ProjectId.make("project-1"), + threadId: ThreadId.make("thread-1"), + prompt: "Duplicate", + schedule: { kind: "once", at: FUTURE }, + createdAt: NOW, + }, + makeReadModel({ tasks: [makeTask({})] }), + ).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("rejects scheduling onto a deleted thread", () => + Effect.gen(function* () { + const error = yield* run( + { + type: "task.schedule", + commandId: CommandId.make("cmd-schedule-deleted"), + taskId: TaskId.make("task-deleted"), + projectId: ProjectId.make("project-1"), + threadId: ThreadId.make("thread-gone"), + prompt: "Nobody home", + schedule: { kind: "once", at: FUTURE }, + createdAt: NOW, + }, + makeReadModel({ + threads: [makeThread({ threadId: ThreadId.make("thread-gone"), deletedAt: NOW })], + }), + ).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("fires a due interval task and advances to the next slot", () => + Effect.gen(function* () { + const decided = yield* run( + { + type: "task.fire", + commandId: CommandId.make("cmd-fire"), + taskId: TaskId.make("task-1"), + // Fired half a period after the scheduled slot. + dueAt: "1970-01-01T00:31:00.000Z", + }, + makeReadModel({ tasks: [makeTask({ nextFireAt: "1970-01-01T00:00:30.000Z" })] }), + ); + const events = Array.isArray(decided) ? decided : [decided]; + expect(events[0]?.type).toBe("task.fired"); + if (events[0]?.type === "task.fired") { + expect(events[0].payload.threadId).toBe(ThreadId.make("thread-1")); + expect(events[0].payload.runtimeMode).toBe("full-access"); + // Anchored to the previous slot + one period, not to the tick time. + expect(events[0].payload.nextFireAt).toBe("1970-01-01T01:00:30.000Z"); + } + }), + ); + + it.effect("coalesces downtime into one fire landing on the first future slot", () => + Effect.gen(function* () { + // Hourly task, next slot 90 minutes overdue. + const decided = yield* run( + { + type: "task.fire", + commandId: CommandId.make("cmd-fire-coalesce"), + taskId: TaskId.make("task-1"), + dueAt: "1970-01-01T02:30:00.000Z", + }, + makeReadModel({ tasks: [makeTask({ nextFireAt: "1970-01-01T01:00:00.000Z" })] }), + ); + const events = Array.isArray(decided) ? decided : [decided]; + expect(events[0]?.type).toBe("task.fired"); + if (events[0]?.type === "task.fired") { + expect(events[0].payload.nextFireAt).toBe("1970-01-01T03:00:00.000Z"); + } + }), + ); + + it.effect("spends a one-shot task's only fire", () => + Effect.gen(function* () { + const decided = yield* run( + { + type: "task.fire", + commandId: CommandId.make("cmd-fire-once"), + taskId: TaskId.make("task-1"), + dueAt: FUTURE, + }, + makeReadModel({ tasks: [makeTask({ schedule: { kind: "once", at: FUTURE } })] }), + ); + const events = Array.isArray(decided) ? decided : [decided]; + expect(events[0]?.type).toBe("task.fired"); + if (events[0]?.type === "task.fired") { + expect(events[0].payload.nextFireAt).toBeNull(); + } + }), + ); + + it.effect("rejects firing ahead of the scheduled time", () => + Effect.gen(function* () { + const error = yield* run( + { + type: "task.fire", + commandId: CommandId.make("cmd-fire-early"), + taskId: TaskId.make("task-1"), + dueAt: "1969-12-31T23:59:00.000Z", + }, + makeReadModel({ tasks: [makeTask({ nextFireAt: FUTURE })] }), + ).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + }), + ); + + it.effect("rejects firing a cancelled or spent task", () => + Effect.gen(function* () { + for (const task of [ + makeTask({ cancelledAt: NOW }), + makeTask({ schedule: { kind: "once", at: FUTURE }, nextFireAt: null }), + ]) { + const error = yield* run( + { + type: "task.fire", + commandId: CommandId.make("cmd-fire-invalid"), + taskId: TaskId.make("task-1"), + dueAt: FUTURE, + }, + makeReadModel({ tasks: [task] }), + ).pipe(Effect.flip); + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + } + }), + ); + + it.effect("cancels a task and is idempotent on re-cancel", () => + Effect.gen(function* () { + const firstDecided = yield* run( + { + type: "task.cancel", + commandId: CommandId.make("cmd-cancel-1"), + taskId: TaskId.make("task-1"), + }, + makeReadModel({ tasks: [makeTask({})] }), + ); + const first = Array.isArray(firstDecided) ? firstDecided : [firstDecided]; + expect(first[0]?.type).toBe("task.cancelled"); + if (first[0]?.type === "task.cancelled") { + expect(first[0].payload.cancelledAt).toBe(first[0].payload.updatedAt); + + const secondDecided = yield* run( + { + type: "task.cancel", + commandId: CommandId.make("cmd-cancel-2"), + taskId: TaskId.make("task-1"), + }, + makeReadModel({ + tasks: [ + makeTask({ + cancelledAt: "1969-12-31T00:00:00.000Z", + updatedAt: "1969-12-31T00:00:00.000Z", + nextFireAt: null, + }), + ], + }), + ); + const second = Array.isArray(secondDecided) ? secondDecided : [secondDecided]; + if (second[0]?.type === "task.cancelled") { + expect(second[0].payload.cancelledAt).toBe("1969-12-31T00:00:00.000Z"); + expect(second[0].payload.updatedAt).toBe("1969-12-31T00:00:00.000Z"); + } else { + throw new Error("expected task.cancelled"); + } + } + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.settled.test.ts b/apps/server/src/orchestration/decider.settled.test.ts index 20bc3475613a..c894d1d3b8d5 100644 --- a/apps/server/src/orchestration/decider.settled.test.ts +++ b/apps/server/src/orchestration/decider.settled.test.ts @@ -33,6 +33,7 @@ function makeReadModel( return { snapshotSequence: 0, projects: [], + tasks: [], threads: [ { id: ThreadId.make("thread-1"), diff --git a/apps/server/src/orchestration/decider.snoozed.test.ts b/apps/server/src/orchestration/decider.snoozed.test.ts index 1012240b18a0..4ec5c339a0e7 100644 --- a/apps/server/src/orchestration/decider.snoozed.test.ts +++ b/apps/server/src/orchestration/decider.snoozed.test.ts @@ -31,6 +31,7 @@ function makeReadModel(input: { return { snapshotSequence: 0, projects: [], + tasks: [], threads: [ { id: ThreadId.make("thread-1"), diff --git a/apps/server/src/orchestration/decider.titleRegeneration.test.ts b/apps/server/src/orchestration/decider.titleRegeneration.test.ts index b29c8ffda676..5dda49e25f4c 100644 --- a/apps/server/src/orchestration/decider.titleRegeneration.test.ts +++ b/apps/server/src/orchestration/decider.titleRegeneration.test.ts @@ -16,6 +16,7 @@ const UPDATED_AT = "2026-01-01T00:00:00.000Z"; const readModel: OrchestrationReadModel = { snapshotSequence: 0, projects: [], + tasks: [], threads: [ { id: ThreadId.make("thread-1"), diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 4f61955fa6aa..89d67110a048 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -15,6 +15,8 @@ import { requireActiveProjectWorkspaceRootAbsent, requireProject, requireProjectAbsent, + requireTask, + requireTaskAbsent, requireThread, requireThreadArchived, requireThreadAbsent, @@ -24,6 +26,15 @@ import { projectEvent } from "./projector.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); +// Epoch-millis → ISO without touching the global Date constructor. +const formatIsoMs = (epochMillis: number): string => + DateTime.formatIso(DateTime.makeUnsafe(epochMillis)); + +// Tolerated clock skew between a fire's observed due time and the task's +// persisted nextFireAt. Small enough to reject genuinely early fires, large +// enough to absorb millisecond jitter between tick clocks. +const TASK_FIRE_CLOCK_SKEW_MS = 2_000; + // Session adoption takes seconds; a user message still unadopted after this // window is a failed/stale start, not pending work. Mirrors the client's // QUEUED_TURN_START_GRACE_MS in client-runtime threadSettled.ts. @@ -1402,6 +1413,183 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" return [unsettledEvent, activityAppendedEvent]; } + case "task.schedule": { + yield* requireProject({ + readModel, + command, + projectId: command.projectId, + }); + const anchorThread = yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + if (anchorThread.deletedAt !== null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${command.threadId}' is deleted and cannot anchor task '${command.taskId}'.`, + }); + } + yield* requireTaskAbsent({ + readModel, + command, + taskId: command.taskId, + }); + + const occurredAt = yield* nowIso; + // Interval tasks anchor their first fire at scheduling time so repeated + // fires never drift; "once" tasks fire exactly at `at`. A due time in + // the past would create a task that is armed and overdue at once — + // reject instead of silently normalizing (mirrors thread.snooze). + const occurredAtMs = Date.parse(occurredAt); + const nextFireAt = + command.schedule.kind === "once" + ? command.schedule.at + : formatIsoMs(occurredAtMs + command.schedule.everyMs); + if (!(Date.parse(nextFireAt) > occurredAtMs)) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `task ${command.taskId} first fire time ${nextFireAt} is not in the future`, + }); + } + return { + ...(yield* withEventBase({ + aggregateKind: "task", + aggregateId: command.taskId, + occurredAt, + commandId: command.commandId, + })), + type: "task.scheduled", + payload: { + taskId: command.taskId, + projectId: command.projectId, + threadId: command.threadId, + ...(command.name !== undefined ? { name: command.name } : {}), + prompt: command.prompt, + schedule: command.schedule, + nextFireAt, + createdAt: occurredAt, + updatedAt: occurredAt, + }, + }; + } + + case "task.cancel": { + const task = yield* requireTask({ + readModel, + command, + taskId: command.taskId, + }); + // Idempotent by re-emission (see thread.unsnooze): cancelling an + // already-cancelled task keeps the original timestamps. + const alreadyCancelled = task.cancelledAt !== null; + const occurredAt = yield* nowIso; + return { + ...(yield* withEventBase({ + aggregateKind: "task", + aggregateId: command.taskId, + occurredAt, + commandId: command.commandId, + })), + type: "task.cancelled", + payload: { + taskId: command.taskId, + cancelledAt: alreadyCancelled ? task.cancelledAt : occurredAt, + updatedAt: alreadyCancelled ? task.updatedAt : occurredAt, + }, + }; + } + + case "task.fire": { + const task = yield* requireTask({ + readModel, + command, + taskId: command.taskId, + }); + if (task.cancelledAt !== null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `task ${command.taskId} is cancelled and cannot fire`, + }); + } + if (task.nextFireAt === null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `task ${command.taskId} has no pending fire`, + }); + } + // Early-fire guard: a fire observed ahead of nextFireAt (stray command, + // replayed envelope) is rejected modulo a small clock-skew allowance. + const nextFireAtMs = Date.parse(task.nextFireAt); + if (Number.isNaN(nextFireAtMs)) { + return yield* Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `task ${command.taskId} has unparseable next fire time ${task.nextFireAt}`, + }), + ); + } + const dueAtMs = Date.parse(command.dueAt); + if (dueAtMs < nextFireAtMs - TASK_FIRE_CLOCK_SKEW_MS) { + return yield* Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `task ${command.taskId} cannot fire at ${command.dueAt} ahead of its due time ${task.nextFireAt}`, + }), + ); + } + // The anchor thread's current modes ride on the fire event so the + // turn-starting reactor can build a complete `thread.turn.start` + // without a read-model lookup. A deleted anchor fails here, before the + // task's nextFireAt advances. + const anchorThread = yield* requireThread({ + readModel, + command, + threadId: task.threadId, + }); + if (anchorThread.deletedAt !== null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `task ${command.taskId} anchor thread '${task.threadId}' is deleted and cannot fire`, + }); + } + const firedAt = yield* nowIso; + const nowMs = Math.max(dueAtMs, Date.parse(firedAt)); + // Interval slots are derived from the PREVIOUS due slot, never from + // "now", so fires stay anchored to the schedule. Downtime coalesces: + // one fire steps over every missed slot and lands on the first slot in + // the future, instead of stampeding catch-up turns after a restart. + const nextFireAt = + task.schedule.kind === "once" + ? null + : formatIsoMs( + nextFireAtMs + + (Math.floor((nowMs - nextFireAtMs) / task.schedule.everyMs) + 1) * + task.schedule.everyMs, + ); + return { + ...(yield* withEventBase({ + aggregateKind: "task", + aggregateId: command.taskId, + occurredAt: firedAt, + commandId: command.commandId, + })), + type: "task.fired", + payload: { + taskId: command.taskId, + projectId: task.projectId, + threadId: task.threadId, + prompt: task.prompt, + firedAt, + dueAt: command.dueAt, + nextFireAt, + runtimeMode: anchorThread.runtimeMode, + interactionMode: anchorThread.interactionMode, + updatedAt: firedAt, + }, + }; + } + default: { command satisfies never; const fallback = command as never as { type: string }; diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index f486dcb2bcbc..69b64407feb1 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -1,4 +1,9 @@ -import type { OrchestrationEvent, OrchestrationReadModel, ThreadId } from "@t3tools/contracts"; +import type { + OrchestrationEvent, + OrchestrationReadModel, + OrchestrationTask, + ThreadId, +} from "@t3tools/contracts"; import { OrchestrationCheckpointSummary, OrchestrationMessage, @@ -33,6 +38,9 @@ import { ThreadRevertedPayload, ThreadSessionSetPayload, ThreadTurnDiffCompletedPayload, + TaskCancelledPayload, + TaskFiredPayload, + TaskScheduledPayload, } from "./Schemas.ts"; type ThreadPatch = Partial>; @@ -190,6 +198,7 @@ export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { snapshotSequence: 0, projects: [], threads: [], + tasks: [], updatedAt: nowIso, }; } @@ -393,6 +402,66 @@ export function projectEvent( })), ); + case "task.scheduled": + return decodeForEvent(TaskScheduledPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => { + const existing = nextBase.tasks.find((entry) => entry.taskId === payload.taskId); + const nextTask: OrchestrationTask = { + taskId: payload.taskId, + projectId: payload.projectId, + threadId: payload.threadId, + ...(payload.name !== undefined ? { name: payload.name } : {}), + prompt: payload.prompt, + schedule: payload.schedule, + createdAt: payload.createdAt, + updatedAt: payload.updatedAt, + lastFiredAt: null, + nextFireAt: payload.nextFireAt, + cancelledAt: null, + }; + return { + ...nextBase, + tasks: existing + ? nextBase.tasks.map((entry) => (entry.taskId === payload.taskId ? nextTask : entry)) + : [...nextBase.tasks, nextTask], + }; + }), + ); + + case "task.fired": + return decodeForEvent(TaskFiredPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + tasks: nextBase.tasks.map((task) => + task.taskId === payload.taskId + ? { + ...task, + lastFiredAt: payload.firedAt, + nextFireAt: payload.nextFireAt, + updatedAt: payload.updatedAt, + } + : task, + ), + })), + ); + + case "task.cancelled": + return decodeForEvent(TaskCancelledPayload, event.payload, event.type, "payload").pipe( + Effect.map((payload) => ({ + ...nextBase, + tasks: nextBase.tasks.map((task) => + task.taskId === payload.taskId + ? { + ...task, + nextFireAt: null, + cancelledAt: payload.cancelledAt, + updatedAt: payload.updatedAt, + } + : task, + ), + })), + ); + case "thread.unsnoozed": return decodeForEvent(ThreadUnsnoozedPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/OrchestrationEventStore.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts index 18d0e9aa578b..29d50e631237 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts @@ -9,6 +9,7 @@ import { OrchestrationEventMetadata, OrchestrationEventType, ProjectId, + TaskId, ThreadId, } from "@t3tools/contracts"; import * as SqlClient from "effect/unstable/sql/SqlClient"; @@ -35,7 +36,7 @@ const EventMetadataFromJsonString = Schema.fromJsonString(OrchestrationEventMeta const AppendEventRequestSchema = Schema.Struct({ eventId: EventId, aggregateKind: OrchestrationAggregateKind, - streamId: Schema.Union([ProjectId, ThreadId]), + streamId: Schema.Union([ProjectId, ThreadId, TaskId]), type: OrchestrationEventType, causationEventId: Schema.NullOr(EventId), correlationId: Schema.NullOr(CommandId), @@ -51,7 +52,7 @@ const OrchestrationEventPersistedRowSchema = Schema.Struct({ eventId: EventId, type: OrchestrationEventType, aggregateKind: OrchestrationAggregateKind, - aggregateId: Schema.Union([ProjectId, ThreadId]), + aggregateId: Schema.Union([ProjectId, ThreadId, TaskId]), occurredAt: IsoDateTime, commandId: Schema.NullOr(CommandId), causationEventId: Schema.NullOr(EventId), diff --git a/apps/server/src/persistence/Layers/ProjectionTasks.ts b/apps/server/src/persistence/Layers/ProjectionTasks.ts new file mode 100644 index 000000000000..ccfe765d19ef --- /dev/null +++ b/apps/server/src/persistence/Layers/ProjectionTasks.ts @@ -0,0 +1,150 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { toPersistenceSqlError } from "../Errors.ts"; + +import { + ProjectionTask, + ProjectionTaskRepository, + type ProjectionTaskRepositoryShape, + DeleteProjectionTaskInput, + GetProjectionTaskInput, + ListProjectionTasksByProjectInput, +} from "../Services/ProjectionTasks.ts"; + +const makeProjectionTaskRepository = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const upsertProjectionTaskRow = SqlSchema.void({ + Request: ProjectionTask, + execute: (row) => + sql` + INSERT INTO projection_tasks ( + task_id, + project_id, + thread_id, + name, + prompt, + schedule_json, + created_at, + updated_at, + last_fired_at, + next_fire_at, + cancelled_at + ) + VALUES ( + ${row.taskId}, + ${row.projectId}, + ${row.threadId}, + ${row.name ?? null}, + ${row.prompt}, + ${row.scheduleJson}, + ${row.createdAt}, + ${row.updatedAt}, + ${row.lastFiredAt}, + ${row.nextFireAt}, + ${row.cancelledAt} + ) + ON CONFLICT (task_id) + DO UPDATE SET + project_id = excluded.project_id, + thread_id = excluded.thread_id, + name = excluded.name, + prompt = excluded.prompt, + schedule_json = excluded.schedule_json, + created_at = excluded.created_at, + updated_at = excluded.updated_at, + last_fired_at = excluded.last_fired_at, + next_fire_at = excluded.next_fire_at, + cancelled_at = excluded.cancelled_at + `, + }); + + const getProjectionTaskRow = SqlSchema.findOneOption({ + Request: GetProjectionTaskInput, + Result: ProjectionTask, + execute: ({ taskId }) => + sql` + SELECT + task_id AS "taskId", + project_id AS "projectId", + thread_id AS "threadId", + name, + prompt, + schedule_json AS "scheduleJson", + created_at AS "createdAt", + updated_at AS "updatedAt", + last_fired_at AS "lastFiredAt", + next_fire_at AS "nextFireAt", + cancelled_at AS "cancelledAt" + FROM projection_tasks + WHERE task_id = ${taskId} + `, + }); + + const listProjectionTaskRowsByProject = SqlSchema.findAll({ + Request: ListProjectionTasksByProjectInput, + Result: ProjectionTask, + execute: ({ projectId }) => + sql` + SELECT + task_id AS "taskId", + project_id AS "projectId", + thread_id AS "threadId", + name, + prompt, + schedule_json AS "scheduleJson", + created_at AS "createdAt", + updated_at AS "updatedAt", + last_fired_at AS "lastFiredAt", + next_fire_at AS "nextFireAt", + cancelled_at AS "cancelledAt" + FROM projection_tasks + WHERE project_id = ${projectId} + ORDER BY created_at ASC, task_id ASC + `, + }); + + const deleteProjectionTaskRow = SqlSchema.void({ + Request: DeleteProjectionTaskInput, + execute: ({ taskId }) => + sql` + DELETE FROM projection_tasks + WHERE task_id = ${taskId} + `, + }); + + const upsert: ProjectionTaskRepositoryShape["upsert"] = (row) => + upsertProjectionTaskRow(row).pipe( + Effect.mapError(toPersistenceSqlError("ProjectionTaskRepository.upsert:query")), + ); + + const getById: ProjectionTaskRepositoryShape["getById"] = (input) => + getProjectionTaskRow(input).pipe( + Effect.mapError(toPersistenceSqlError("ProjectionTaskRepository.getById:query")), + ); + + const listByProjectId: ProjectionTaskRepositoryShape["listByProjectId"] = (input) => + listProjectionTaskRowsByProject(input).pipe( + Effect.mapError(toPersistenceSqlError("ProjectionTaskRepository.listByProjectId:query")), + ); + + const deleteById: ProjectionTaskRepositoryShape["deleteById"] = (input) => + deleteProjectionTaskRow(input).pipe( + Effect.mapError(toPersistenceSqlError("ProjectionTaskRepository.deleteById:query")), + ); + + return { + upsert, + getById, + listByProjectId, + deleteById, + } satisfies ProjectionTaskRepositoryShape; +}); + +export const ProjectionTaskRepositoryLive = Layer.effect( + ProjectionTaskRepository, + makeProjectionTaskRepository, +); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 170cb3992279..ccda3049eebe 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -54,6 +54,7 @@ import Migration0038 from "./Migrations/038_ProjectionThreadsPinOrderKey.ts"; import Migration0039 from "./Migrations/039_ProjectionProjectsDefaultThreadEnvMode.ts"; import Migration0040 from "./Migrations/040_ProjectionProjectFaviconPath.ts"; import Migration0041 from "./Migrations/041_AuthSessionClientConnection.ts"; +import Migration0042 from "./Migrations/042_ProjectionTasks.ts"; /** * Migration loader with all migrations defined inline. @@ -107,6 +108,7 @@ export const migrationEntries = [ [39, "ProjectionProjectsDefaultThreadEnvMode", Migration0039], [40, "ProjectionProjectFaviconPath", Migration0040], [41, "AuthSessionClientConnection", Migration0041], + [42, "ProjectionTasks", Migration0042], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/042_ProjectionTasks.ts b/apps/server/src/persistence/Migrations/042_ProjectionTasks.ts new file mode 100644 index 000000000000..1f4833034256 --- /dev/null +++ b/apps/server/src/persistence/Migrations/042_ProjectionTasks.ts @@ -0,0 +1,28 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS projection_tasks ( + task_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + name TEXT, + prompt TEXT NOT NULL, + schedule_json TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + last_fired_at TEXT, + next_fire_at TEXT, + cancelled_at TEXT + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_tasks_next_fire_at + ON projection_tasks (next_fire_at) + WHERE next_fire_at IS NOT NULL AND cancelled_at IS NULL + `; +}); diff --git a/apps/server/src/persistence/Services/OrchestrationCommandReceipts.ts b/apps/server/src/persistence/Services/OrchestrationCommandReceipts.ts index 1498984827e5..5b166838478c 100644 --- a/apps/server/src/persistence/Services/OrchestrationCommandReceipts.ts +++ b/apps/server/src/persistence/Services/OrchestrationCommandReceipts.ts @@ -13,6 +13,7 @@ import { OrchestrationAggregateKind, OrchestrationCommandReceiptStatus, ProjectId, + TaskId, ThreadId, } from "@t3tools/contracts"; import * as Option from "effect/Option"; @@ -25,7 +26,7 @@ import type { OrchestrationCommandReceiptRepositoryError } from "../Errors.ts"; export const OrchestrationCommandReceipt = Schema.Struct({ commandId: CommandId, aggregateKind: OrchestrationAggregateKind, - aggregateId: Schema.Union([ProjectId, ThreadId]), + aggregateId: Schema.Union([ProjectId, ThreadId, TaskId]), acceptedAt: IsoDateTime, resultSequence: NonNegativeInt, status: OrchestrationCommandReceiptStatus, diff --git a/apps/server/src/persistence/Services/ProjectionTasks.ts b/apps/server/src/persistence/Services/ProjectionTasks.ts new file mode 100644 index 000000000000..182dbeb314b3 --- /dev/null +++ b/apps/server/src/persistence/Services/ProjectionTasks.ts @@ -0,0 +1,86 @@ +/** + * ProjectionTaskRepository - Repository interface for scheduled tasks. + * + * Owns persistence operations for projected task records in the orchestration + * read model. + * + * @module ProjectionTaskRepository + */ +import { IsoDateTime, ProjectId, TaskId, ThreadId } from "@t3tools/contracts"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as Context from "effect/Context"; +import type * as Effect from "effect/Effect"; + +import type { ProjectionRepositoryError } from "../Errors.ts"; + +export const ProjectionTask = Schema.Struct({ + taskId: TaskId, + projectId: ProjectId, + threadId: ThreadId, + name: Schema.optional(Schema.NullOr(Schema.String)), + prompt: Schema.String, + scheduleJson: Schema.String, + createdAt: IsoDateTime, + updatedAt: IsoDateTime, + lastFiredAt: Schema.NullOr(IsoDateTime), + nextFireAt: Schema.NullOr(IsoDateTime), + cancelledAt: Schema.NullOr(IsoDateTime), +}); +export type ProjectionTask = typeof ProjectionTask.Type; + +export const GetProjectionTaskInput = Schema.Struct({ + taskId: TaskId, +}); +export type GetProjectionTaskInput = typeof GetProjectionTaskInput.Type; + +export const DeleteProjectionTaskInput = Schema.Struct({ + taskId: TaskId, +}); +export type DeleteProjectionTaskInput = typeof DeleteProjectionTaskInput.Type; + +export const ListProjectionTasksByProjectInput = Schema.Struct({ + projectId: ProjectId, +}); +export type ListProjectionTasksByProjectInput = typeof ListProjectionTasksByProjectInput.Type; + +/** + * ProjectionTaskRepositoryShape - Service API for projected task records. + */ +export interface ProjectionTaskRepositoryShape { + /** + * Insert or replace a projected task row. + * + * Upserts by `taskId`. + */ + readonly upsert: (task: ProjectionTask) => Effect.Effect; + + /** + * Read a projected task row by id. + */ + readonly getById: ( + input: GetProjectionTaskInput, + ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * List projected tasks for a project. + */ + readonly listByProjectId: ( + input: ListProjectionTasksByProjectInput, + ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * Delete projected task state by id. + */ + readonly deleteById: ( + input: DeleteProjectionTaskInput, + ) => Effect.Effect; +} + +/** + * ProjectionTaskRepository - Service tag for task persistence. + */ +export class ProjectionTaskRepository extends Context.Service< + ProjectionTaskRepository, + ProjectionTaskRepositoryShape +>()("t3/persistence/Services/ProjectionTasks/ProjectionTaskRepository") {} diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index 5c5da4666b0d..5e19b5171a60 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -45,6 +45,8 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }); const makeTerminalManagerLayer = ( diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 0b1bc9e149f7..fd382fa747c4 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -224,6 +224,8 @@ describe("ProviderSessionReaper", () => { getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), ), Layer.provideMerge(NodeServices.layer), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 02a367c08792..1522fd9316ad 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -204,6 +204,7 @@ const testEnvironmentDescriptor = { const makeDefaultOrchestrationReadModel = () => { const now = "2026-01-01T00:00:00.000Z"; return { + tasks: [], snapshotSequence: 0, updatedAt: now, projects: [ @@ -808,6 +809,8 @@ const buildAppUnderTest = (options?: { updatedAt: "1970-01-01T00:00:00.000Z", }), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), getProjectShellById: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), @@ -5989,6 +5992,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { const now = "2026-01-01T00:00:00.000Z"; const snapshot = { snapshotSequence: 1, + tasks: [], updatedAt: now, projects: [ { diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 3e41b4390f82..d01e63346c0d 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -59,6 +59,8 @@ 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 { TaskFireReactorLive } from "./orchestration/Layers/TaskFireReactor.ts"; +import { TaskSchedulerLive } from "./orchestration/Layers/TaskScheduler.ts"; import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts"; import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry.ts"; @@ -245,6 +247,7 @@ const ReactorLayerLive = Layer.empty.pipe( Layer.provideMerge(ProviderCommandReactorLive), Layer.provideMerge(CheckpointReactorLive), Layer.provideMerge(ThreadDeletionReactorLive), + Layer.provideMerge(TaskFireReactorLive), Layer.provideMerge(AgentAwarenessRelay.layer.pipe(Layer.provide(ServerSecretStore.layer))), Layer.provideMerge(RuntimeReceiptBusLive), ); @@ -368,6 +371,13 @@ const ProviderRuntimeLayerLive = ProviderSessionReaperLive.pipe( Layer.provideMerge(OrchestrationLayerLive), ); +// The task scheduler reads due tasks from the projection and dispatches +// `task.fire` through the engine, so it sits beside the other orchestration +// consumers with the same layer provided underneath. +const TaskSchedulerRuntimeLayerLive = TaskSchedulerLive.pipe( + Layer.provideMerge(OrchestrationLayerLive), +); + const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( // Core Services Layer.provideMerge(ServerSettingsLayerLive), @@ -376,6 +386,7 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(GitLayerLive), Layer.provideMerge(VcsLayerLive), Layer.provideMerge(ProviderRuntimeLayerLive), + Layer.provideMerge(TaskSchedulerRuntimeLayerLive), Layer.provideMerge(Layer.mergeAll(TerminalLayerLive, PreviewLayerLive)), Layer.provideMerge(PersistenceLayerLive), Layer.provideMerge(Keybindings.layer), @@ -399,8 +410,10 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( // keeps a single Live for all opencode consumers. Layer.provideMerge(OpenCodeRuntime.OpenCodeRuntimeLive), Layer.provideMerge(WorkspaceLayerLive), - Layer.provideMerge(ProjectFaviconResolverLayerLive), - Layer.provideMerge(RepositoryIdentityResolver.layer), + // Merged pair keeps the pipe within Effect's 20-argument variadic limit. + Layer.provideMerge( + Layer.mergeAll(ProjectFaviconResolverLayerLive, RepositoryIdentityResolver.layer), + ), Layer.provideMerge(ServerEnvironment.layer), Layer.provideMerge(AuthLayerLive), Layer.provideMerge(ServerSecretStore.layer), diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index e3f7e482b2e0..4aaa6b76c6a7 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -98,6 +98,8 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), Effect.provideService(AnalyticsService.AnalyticsService, { record: () => Effect.void, @@ -162,6 +164,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, @@ -207,6 +211,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, @@ -258,6 +264,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), searchThreads: () => Effect.succeed({ matches: [] }), + listTasks: () => Effect.succeed([]), + listDueTasks: () => Effect.succeed([]), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 6bea823b19c3..721dc8beaf02 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -37,6 +37,7 @@ import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as ProviderService from "./provider/Services/ProviderService.ts"; import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts"; import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts"; +import * as TaskScheduler from "./orchestration/Services/TaskScheduler.ts"; import { forkParked } from "./serverActivation.ts"; import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts"; import { @@ -388,6 +389,7 @@ export const make = (options?: StartupOptions) => const keybindings = yield* Keybindings.Keybindings; const orchestrationReactor = yield* OrchestrationReactor.OrchestrationReactor; const providerSessionReaper = yield* ProviderSessionReaper.ProviderSessionReaper; + const taskScheduler = yield* TaskScheduler.TaskScheduler; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; const serverSettings = yield* ServerSettings.ServerSettingsService; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; @@ -437,6 +439,7 @@ export const make = (options?: StartupOptions) => Effect.gen(function* () { yield* orchestrationReactor.start().pipe(Scope.provide(reactorScope)); yield* providerSessionReaper.start().pipe(Scope.provide(reactorScope)); + yield* taskScheduler.start().pipe(Scope.provide(reactorScope)); }), ); diff --git a/apps/server/src/serverSettings.test.ts b/apps/server/src/serverSettings.test.ts index 35ef5e976223..3513462a1022 100644 --- a/apps/server/src/serverSettings.test.ts +++ b/apps/server/src/serverSettings.test.ts @@ -116,6 +116,24 @@ it.layer(NodeServices.layer)("server settings", (it) => { }), ); + it.effect("round-trips the scheduled tasks feature switch", () => + Effect.gen(function* () { + const serverSettings = yield* ServerSettingsModule.ServerSettingsService; + + assert.equal(DEFAULT_SERVER_SETTINGS.enableScheduledTasks, true); + + const disabled = yield* serverSettings.updateSettings({ enableScheduledTasks: false }); + assert.equal(disabled.enableScheduledTasks, false); + // Persisted sparsely (non-default only) and reloaded cleanly. + const reloaded = yield* serverSettings.getSettings; + assert.equal(reloaded.enableScheduledTasks, false); + + yield* serverSettings.updateSettings({ enableScheduledTasks: true }); + const restored = yield* serverSettings.getSettings; + assert.equal(restored.enableScheduledTasks, true); + }).pipe(Effect.provide(makeServerSettingsLayer())), + ); + it.effect( "decodes legacy object-shaped textGenerationModelSelection.options from settings.json", () => diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 11c659e28a70..908f0dc0f42f 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -30,6 +30,7 @@ import { type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, + OrchestrationListTasksError, OrchestrationSearchThreadsError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, @@ -1267,6 +1268,21 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "orchestration" }, ), + [ORCHESTRATION_WS_METHODS.listTasks]: (input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.listTasks, + projectionSnapshotQuery.listTasks(input.projectId).pipe( + Effect.map((tasks) => ({ tasks })), + Effect.mapError( + (cause) => + new OrchestrationListTasksError({ + message: "Failed to list scheduled tasks", + cause, + }), + ), + ), + { "rpc.aggregate": "orchestration" }, + ), [ORCHESTRATION_WS_METHODS.subscribeShell]: (input) => observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeShell, diff --git a/apps/web/src/components/settings/IntegrationsSettings.tsx b/apps/web/src/components/settings/IntegrationsSettings.tsx index f3b773f7bc80..9d174ddd6a48 100644 --- a/apps/web/src/components/settings/IntegrationsSettings.tsx +++ b/apps/web/src/components/settings/IntegrationsSettings.tsx @@ -1,8 +1,8 @@ /** * Integrations settings - preferences for surfaces T3 Code embeds rather than - * owns. Browser is the first section: the defaults a preview tab opens at, - * applied to both hand-opened tabs and agent `preview_open` calls that don't - * state their own size. + * owns. Automation covers server-side scheduled runs; Browser holds the + * defaults a preview tab opens at, applied to both hand-opened tabs and agent + * `preview_open` calls that don't state their own size. * * @module IntegrationsSettings */ @@ -355,6 +355,37 @@ function BrowserAppearanceSetting({ disabled }: { readonly disabled: boolean }) ); } +function ScheduledTasksSetting() { + const settings = usePrimarySettings(); + const updateSettings = useUpdatePrimarySettings(); + + return ( + + updateSettings({ + enableScheduledTasks: DEFAULT_UNIFIED_SETTINGS.enableScheduledTasks, + }) + } + /> + ) : null + } + control={ + updateSettings({ enableScheduledTasks: Boolean(checked) })} + aria-label="Enable scheduled tasks" + /> + } + /> + ); +} + function AgentBrowserAccessSetting() { const settings = usePrimarySettings(); const updateSettings = useUpdatePrimarySettings(); @@ -467,6 +498,9 @@ export function IntegrationsSettingsPanel() { return ( + + + {/* Server-authoritative, so it stays editable on every client and sits outside the block covering the desktop-only defaults. */} diff --git a/apps/web/src/components/settings/ProjectSettingsPanel.tsx b/apps/web/src/components/settings/ProjectSettingsPanel.tsx index 6768d2dc61ef..00c2cd58905b 100644 --- a/apps/web/src/components/settings/ProjectSettingsPanel.tsx +++ b/apps/web/src/components/settings/ProjectSettingsPanel.tsx @@ -73,6 +73,7 @@ import { useAtomCommand } from "../../state/use-atom-command"; import { ProviderModelPicker } from "../chat/ProviderModelPicker"; import { TraitsPicker } from "../chat/TraitsPicker"; import { ProjectFavicon } from "../ProjectFavicon"; +import { ScheduledTasksSection } from "./ScheduledTasksSection"; import { EMPTY_PROJECT_SCRIPT_INPUT, editorRequestForScript, @@ -1144,6 +1145,18 @@ function ProjectDetail({ group }: { group: SidebarProjectSnapshot }) { ) : null} + {settings.enableScheduledTasks ? ( + + thread.environmentId === representative.environmentId && + thread.projectId === representative.id, + )} + /> + ) : null} + = { + "in-1h": "Once, in 1 hour", + "tomorrow-9": "Once, tomorrow at 9:00", + "daily-9": "Every day at 9:00", + "weekly-mon-9": "Every Monday at 9:00", +}; + +function nextNineAm(fromMs: number): string { + const nineAm = new Date(fromMs + DAY_MS); + nineAm.setHours(9, 0, 0, 0); + return nineAm.toISOString(); +} + +function resolveSchedule(preset: SchedulePreset): OrchestrationTask["schedule"] { + const now = Date.now(); + switch (preset) { + case "in-1h": + return { kind: "once", at: new Date(now + HOUR_MS).toISOString() }; + case "tomorrow-9": + return { kind: "once", at: nextNineAm(now) }; + case "daily-9": + return { kind: "interval", everyMs: DAY_MS }; + case "weekly-mon-9": + return { kind: "interval", everyMs: 7 * DAY_MS }; + } +} + +function describeSchedule(schedule: OrchestrationTask["schedule"]): string { + if (schedule.kind === "once") { + return `Once · ${new Date(schedule.at).toLocaleString()}`; + } + if (schedule.everyMs === DAY_MS) return "Every day"; + if (schedule.everyMs === 7 * DAY_MS) return "Every week"; + return `Every ${Math.round(schedule.everyMs / HOUR_MS)}h`; +} + +function formatWhen(iso: string | null): string { + if (iso === null) return "—"; + const diffMs = new Date(iso).getTime() - Date.now(); + if (diffMs <= 0) return "due"; + const minutes = Math.round(diffMs / 60_000); + if (minutes < 60) return `in ${minutes}m`; + const hours = Math.round(minutes / 60); + if (hours < 48) return `in ${hours}h`; + return `in ${Math.round(hours / 24)}d`; +} + +function makeTaskId(): string { + return `task-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; +} + +export function ScheduledTasksSection({ + environmentId, + projectId, + threads, +}: { + readonly environmentId: EnvironmentId; + readonly projectId: ProjectId; + readonly threads: ReadonlyArray; +}) { + const supported = readEnvironmentSupportsTaskScheduling(environmentId); + + const [threadKey, setThreadKey] = useState( + () => `${environmentId}:${threads[0]?.id ?? ""}`, + ); + const [preset, setPreset] = useState("tomorrow-9"); + const [prompt, setPrompt] = useState(""); + const [submitting, setSubmitting] = useState(false); + + const tasksResult = useAtomValue( + orchestrationEnvironment.scheduledTasks({ environmentId, input: { projectId } }), + ); + const tasksValue = Option.getOrNull(AsyncResult.value(tasksResult)); + const schedule = useAtomCommand(taskCommands.scheduleTask, { reportFailure: false }); + const cancel = useAtomCommand(taskCommands.cancelTask, { reportFailure: false }); + + if (!supported) { + return null; + } + + const tasks = [...(tasksValue?.tasks ?? [])].toSorted((a, b) => { + if ((a.cancelledAt !== null) !== (b.cancelledAt !== null)) { + return a.cancelledAt === null ? -1 : 1; + } + return (a.nextFireAt ?? "").localeCompare(b.nextFireAt ?? ""); + }); + + const selectedThreadId = threadKey.startsWith(`${environmentId}:`) + ? threadKey.slice(environmentId.length + 1) + : ""; + const anchorThread = threads.find((thread) => thread.id === selectedThreadId); + + const submit = async () => { + const trimmedPrompt = prompt.trim(); + if (!anchorThread || trimmedPrompt.length === 0 || submitting) return; + setSubmitting(true); + try { + const result = await schedule({ + environmentId, + input: { + taskId: makeTaskId(), + projectId, + threadId: anchorThread.id, + prompt: trimmedPrompt, + schedule: resolveSchedule(preset), + }, + }); + if (result._tag === "Success") { + setPrompt(""); + toastManager.add({ type: "success", title: "Task scheduled" }); + } else { + toastManager.add({ type: "error", title: "Could not schedule task" }); + } + } finally { + setSubmitting(false); + } + }; + + const cancelTaskById = async (taskId: string) => { + const result = await cancel({ environmentId, input: { taskId } }); + if (result._tag === "Failure") { + toastManager.add({ type: "error", title: "Could not cancel task" }); + } + }; + + return ( + +

+ Start a turn on a thread automatically — nightly checkups, follow-ups, recurring chores. + Runs only while this T3 server is up. +

+ {threads.length > 0 ? ( + <> + setThreadKey(String(value))}> + + + + + {threads.map((thread) => ( + + {thread.title} + + ))} + + + } + /> + setPreset(String(value) as SchedulePreset)} + > + + + + + {(Object.keys(PRESET_LABELS) as SchedulePreset[]).map((key) => ( + + {PRESET_LABELS[key]} + + ))} + + + } + /> + + setPrompt(event.target.value)} + onKeyDown={(event) => { + if (event.key === "Enter") void submit(); + }} + /> + + + } + /> + + ) : ( +

+ Create a thread first — scheduled runs append to an existing thread. +

+ )} + {tasks.length > 0 ? ( +
    + {tasks.map((task) => ( +
  • +
    +

    {task.prompt}

    +

    + {describeSchedule(task.schedule)} + {task.cancelledAt === null && task.nextFireAt !== null + ? ` · next ${formatWhen(task.nextFireAt)}` + : ""} + {task.lastFiredAt !== null + ? ` · last ran ${new Date(task.lastFiredAt).toLocaleString()}` + : ""} + {task.cancelledAt !== null ? " · cancelled" : ""} +

    +
    + {task.cancelledAt === null ? ( + + ) : null} +
    • + ))} +
+ ) : null} + + ); +} diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index 582dc9f6cb94..4719326a2d63 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -540,6 +540,9 @@ export function useSettingsRestore(onRestored?: () => void) { ...(settings.enableAgentBrowserAccess !== DEFAULT_UNIFIED_SETTINGS.enableAgentBrowserAccess ? ["Agent browser access"] : []), + ...(settings.enableScheduledTasks !== DEFAULT_UNIFIED_SETTINGS.enableScheduledTasks + ? ["Scheduled tasks"] + : []), ], [ isTextGenerationModelDirty, @@ -550,6 +553,7 @@ export function useSettingsRestore(onRestored?: () => void) { settings.browserAutoShowFloatingPreview, settings.appearanceContrast, settings.enableAgentBrowserAccess, + settings.enableScheduledTasks, settings.confirmQuit, settings.confirmThreadArchive, settings.confirmThreadDelete, @@ -683,6 +687,7 @@ export function useSettingsRestore(onRestored?: () => void) { // name, so a user restoring defaults is told the agent regains access // rather than discovering it later. enableAgentBrowserAccess: DEFAULT_UNIFIED_SETTINGS.enableAgentBrowserAccess, + enableScheduledTasks: DEFAULT_UNIFIED_SETTINGS.enableScheduledTasks, }); onRestored?.(); }, [ diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index 188dbd408f0a..b2514261b2eb 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -209,6 +209,12 @@ export const SETTINGS_SEARCH_ITEMS = [ to: "/settings/integrations", targetId: "browser", }, + { + id: "scheduled-tasks", + title: "Scheduled tasks", + to: "/settings/integrations", + targetId: "automation", + }, { id: "browser-default-viewport", title: "Default browser viewport", diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 7bca31182379..3e13fff77648 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -268,6 +268,16 @@ export function readEnvironmentSupportsPinReorder(environmentId: EnvironmentId): ); } +/** Whether the environment's server understands task.schedule/task.cancel and + runs the scheduler that fires due tasks. Same version-skew contract as + settlement. */ +export function readEnvironmentSupportsTaskScheduling(environmentId: EnvironmentId): boolean { + return ( + appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities + .taskScheduling === true + ); +} + export function readThreadDetail(ref: ScopedThreadRef): EnvironmentThread | null { return appAtomRegistry.get(environmentThreadDetails.detailAtom(ref)); } diff --git a/apps/web/src/state/orchestration.ts b/apps/web/src/state/orchestration.ts index 8c6e1738857a..fcd05cbf0076 100644 --- a/apps/web/src/state/orchestration.ts +++ b/apps/web/src/state/orchestration.ts @@ -1,5 +1,10 @@ -import { createOrchestrationEnvironmentAtoms } from "@t3tools/client-runtime/state/orchestration"; +import { + createOrchestrationEnvironmentAtoms, + createTaskEnvironmentCommands, +} from "@t3tools/client-runtime/state/orchestration"; import { connectionAtomRuntime } from "../connection/runtime"; export const orchestrationEnvironment = createOrchestrationEnvironmentAtoms(connectionAtomRuntime); +export const taskCommands: ReturnType = + createTaskEnvironmentCommands(connectionAtomRuntime); diff --git a/docs/internals/glossary.md b/docs/internals/glossary.md index da16f74d339f..6ac8dfb16092 100644 --- a/docs/internals/glossary.md +++ b/docs/internals/glossary.md @@ -48,13 +48,17 @@ Orchestration is the server-side domain layer that turns runtime activity into s #### Aggregate -The domain object a command or event belongs to. In [the contracts][1], that is usually `project` or `thread`. See [decider.ts][8]. +The domain object a command or event belongs to. In [the contracts][1], that is `project`, `thread`, or `task`. See [decider.ts][8]. #### Command A typed request to change domain state. In [the contracts][1], commands are validated in [commandInvariants.ts][9] and turned into events by [decider.ts][8]. Examples include `thread.create`, `thread.turn.start`, and `thread.checkpoint.revert`. +#### Scheduled task + +A server-side automation that starts a turn on an existing thread at a due time. Created with `task.schedule` (one-shot `at` or interval `everyMs`), fired by the scheduler layer in [TaskScheduler.ts][25] via internal `task.fire` commands whose deterministic ids make retries idempotent, and turned into a normal `thread.turn.start` by [TaskFireReactor.ts][26]. Missed intervals coalesce: one fire steps over every overdue slot. See [scheduled-tasks.md][27]. + #### Domain Event A persisted fact that something already happened. In [the contracts][1], events are the source of truth, and [projector.ts][4] shows how they are applied. @@ -173,6 +177,9 @@ The file patch and changed-file summary for one turn. It is usually computed in [16]: ./providers.md [17]: ../../apps/server/src/provider/Layers/CodexAdapter.ts [18]: ../user/permission-modes.md +[25]: ../../apps/server/src/orchestration/Layers/TaskScheduler.ts +[26]: ../../apps/server/src/orchestration/Layers/TaskFireReactor.ts +[27]: ../user/scheduled-tasks.md [19]: ../../apps/server/src/checkpointing/CheckpointStore.ts [20]: ../../apps/server/src/checkpointing/CheckpointDiffQuery.ts [21]: ../../apps/server/src/persistence/Services/ProjectionCheckpoints.ts diff --git a/docs/user/scheduled-tasks.md b/docs/user/scheduled-tasks.md new file mode 100644 index 000000000000..82a1cf0ecd24 --- /dev/null +++ b/docs/user/scheduled-tasks.md @@ -0,0 +1,41 @@ +# Schedule agent runs + +Scheduled tasks let T3 Code start an agent turn for you at a set time — a nightly +"run the test suite and report failures", a follow-up tomorrow morning, or a recurring +chore on a fixed interval. Runs happen on the machine running your T3 server, even when +no app window is open. + +## Create a task + +1. Open **Settings** and select **Projects**. +2. Select the project. +3. Scroll to **Scheduled tasks**. +4. Pick the thread the run should append to. +5. Pick a schedule: once in 1 hour, once tomorrow at 9:00, every day at 9:00, or every + Monday at 9:00. +6. Type the prompt and select **Schedule**. + +Each run sends your prompt to the thread as if you had typed it at that moment. The run +uses the thread's model selection and permission mode. + +## Cancel a task + +Open **Settings → Projects → Scheduled tasks** and select **Cancel** next to the task. +Cancelled tasks stay visible until the panel is refreshed elsewhere. + +## Turn the feature off + +Open **Settings → Integrations → Automation** and turn off **Scheduled tasks**. The +server stops starting runs at due times and the per-project schedule UI hides on every +connected client. Nothing is deleted: stored schedules keep their times and resume — +with missed fires coalescing into one, like any other downtime — when you turn the +feature back on. + +## Good to know + +- Tasks are stored with your server's data and survive restarts. If the server is down + when a fire time passes, the run happens once when the server starts again — missed + intervals never stack up. +- Recurring tasks keep their schedule even if a run starts late; each next run is spaced + from the previous scheduled time, not from when the last one actually ran. +- A task can be cancelled from any connected client. diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index abed33998966..ca9d557d674c 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -51,6 +51,10 @@ "types": "./src/state/auth.ts", "default": "./src/state/auth.ts" }, + "./state/taskCommands": { + "types": "./src/state/taskCommands.ts", + "default": "./src/state/taskCommands.ts" + }, "./state/assets": { "types": "./src/state/assets.ts", "default": "./src/state/assets.ts" diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index cb74f117b772..9d9a82f93a78 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -51,6 +51,8 @@ export type RespondToThreadApprovalInput = CommandInput<"thread.approval.respond export type RespondToThreadUserInputInput = CommandInput<"thread.user-input.respond">; export type RevertThreadCheckpointInput = CommandInput<"thread.checkpoint.revert">; export type StopThreadSessionInput = CommandInput<"thread.session.stop">; +export type ScheduleTaskInput = CommandInput<"task.schedule">; +export type CancelTaskInput = CommandInput<"task.cancel">; type DispatchTag = typeof ORCHESTRATION_WS_METHODS.dispatchCommand; type CommandEffect = Effect.Effect< @@ -331,3 +333,26 @@ export const stopThreadSession: (input: StopThreadSessionInput) => CommandEffect createdAt: metadata.createdAt, }); }); + +export const scheduleTask: (input: ScheduleTaskInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.scheduleTask", +)(function* (input) { + const metadata = yield* timestampedCommandMetadata(input); + return yield* dispatch({ + ...input, + type: "task.schedule", + commandId: metadata.commandId, + createdAt: metadata.createdAt, + }); +}); + +export const cancelTask: (input: CancelTaskInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.cancelTask", +)(function* (input) { + const metadata = yield* timestampedCommandMetadata(input); + return yield* dispatch({ + ...input, + type: "task.cancel", + commandId: metadata.commandId, + }); +}); diff --git a/packages/client-runtime/src/state/orchestration.ts b/packages/client-runtime/src/state/orchestration.ts index ba80275bffb3..0c36aeebd793 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -1,11 +1,55 @@ -import { ORCHESTRATION_WS_METHODS } from "@t3tools/contracts"; +import { + type EnvironmentAuthorizationError, + ORCHESTRATION_WS_METHODS, + type OrchestrationDispatchCommandError, + type EnvironmentId, +} from "@t3tools/contracts"; +import { RpcClientError } from "effect/unstable/rpc"; +import * as Crypto from "effect/Crypto"; import { Atom } from "effect/unstable/reactivity"; -import { createEnvironmentRpcQueryAtomFamily } from "./runtime.ts"; +import { + cancelTask, + scheduleTask, + type CancelTaskInput, + type ScheduleTaskInput, +} from "../operations/commands.ts"; +import type { EnvironmentRpcSuccess, EnvironmentRpcUnavailableError } from "../rpc/client.ts"; +import { EnvironmentNotRegisteredError } from "../connection/registry.ts"; +import { + createAtomCommandScheduler, + createEnvironmentCommand, + createEnvironmentRpcQueryAtomFamily, + type AtomCommand, +} from "./runtime.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; +type DispatchTag = typeof ORCHESTRATION_WS_METHODS.dispatchCommand; + +type DispatchFailure = + | EnvironmentAuthorizationError + | EnvironmentNotRegisteredError + | EnvironmentRpcUnavailableError + | OrchestrationDispatchCommandError + | RpcClientError.RpcClientError; + +/** Explicit shape so consumers can name the type without reaching into + internal module paths (TS portability of inferred declarations). */ +export interface TaskEnvironmentCommands { + readonly scheduleTask: AtomCommand< + { readonly environmentId: EnvironmentId; readonly input: ScheduleTaskInput }, + EnvironmentRpcSuccess, + DispatchFailure | E + >; + readonly cancelTask: AtomCommand< + { readonly environmentId: EnvironmentId; readonly input: CancelTaskInput }, + EnvironmentRpcSuccess, + DispatchFailure | E + >; +} + export function createOrchestrationEnvironmentAtoms( - runtime: Atom.AtomRuntime, + runtime: Atom.AtomRuntime, ) { return { turnDiff: createEnvironmentRpcQueryAtomFamily(runtime, { @@ -29,9 +73,44 @@ export function createOrchestrationEnvironmentAtoms( staleTimeMs: 30_000, idleTtlMs: 60_000, }), + scheduledTasks: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:orchestration:scheduled-tasks", + tag: ORCHESTRATION_WS_METHODS.listTasks, + // Task state changes are rare (schedule/cancel/fire); a short stale + // window plus a modest refresh keeps an open panel current without + // hammering the server. + staleTimeMs: 15_000, + idleTtlMs: 60_000, + refreshIntervalMs: 30_000, + }), archivedShellSnapshot: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:archived-shell-snapshot", tag: ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, }), }; } + +export function createTaskEnvironmentCommands( + runtime: Atom.AtomRuntime, +): TaskEnvironmentCommands { + const scheduler = createAtomCommandScheduler(); + const concurrency = { + mode: "serial" as const, + key: ({ environmentId, input }: { environmentId: string; input: { taskId?: string } }) => + JSON.stringify([environmentId, input.taskId ?? null]), + }; + return { + scheduleTask: createEnvironmentCommand(runtime, { + label: "environment-data:commands:task:schedule", + execute: (input: ScheduleTaskInput) => scheduleTask(input), + scheduler, + concurrency, + }), + cancelTask: createEnvironmentCommand(runtime, { + label: "environment-data:commands:task:cancel", + execute: (input: CancelTaskInput) => cancelTask(input), + scheduler, + concurrency, + }), + }; +} diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 1777bcebc2f8..ea0899f355c5 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -64,6 +64,10 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Server understands thread.pin.reorder (and orderKey on thread.pin). Same version-skew contract as threadSettlement. */ threadPinReorder: Schema.optionalKey(Schema.Boolean), + /** Server understands task.schedule / task.cancel commands and runs the + server-side scheduler that fires due tasks. Same version-skew contract + as threadSettlement. */ + taskScheduling: Schema.optionalKey(Schema.Boolean), /** Server understands regenerateTitle on thread.meta.update. Absent on older servers, so clients hide the action instead of sending it. */ threadTitleRegeneration: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index adb17879ff2f..e48c1e07ff3a 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -30,6 +30,7 @@ export const ORCHESTRATION_WS_METHODS = { getTurnDiff: "orchestration.getTurnDiff", getFullThreadDiff: "orchestration.getFullThreadDiff", searchThreads: "orchestration.searchThreads", + listTasks: "orchestration.listTasks", getArchivedShellSnapshot: "orchestration.getArchivedShellSnapshot", subscribeShell: "orchestration.subscribeShell", subscribeThread: "orchestration.subscribeThread", @@ -423,10 +424,56 @@ export const OrchestrationThread = Schema.Struct({ }); export type OrchestrationThread = typeof OrchestrationThread.Type; +export const TaskId = TrimmedNonEmptyString; +export type TaskId = typeof TaskId.Type; + +/** + * One-shot schedule: fire once at an absolute time. The decider rejects times + * in the past at creation, so a scheduled task always starts armed. + */ +export const TaskScheduleOnce = Schema.Struct({ + kind: Schema.Literal("once"), + at: IsoDateTime, +}); +export type TaskScheduleOnce = typeof TaskScheduleOnce.Type; + +/** + * Recurring schedule: fire every `everyMs`, anchored at the previous due time + * so repeated fires never drift. Missed intervals (server downtime) coalesce: + * the next fire advances past all overdue slots in one step. + */ +export const TaskScheduleInterval = Schema.Struct({ + kind: Schema.Literal("interval"), + everyMs: PositiveInt.check(Schema.isGreaterThanOrEqualTo(60_000)), +}); +export type TaskScheduleInterval = typeof TaskScheduleInterval.Type; + +export const TaskScheduleSpec = Schema.Union([TaskScheduleOnce, TaskScheduleInterval]); +export type TaskScheduleSpec = typeof TaskScheduleSpec.Type; + +export const OrchestrationTask = Schema.Struct({ + taskId: TaskId, + projectId: ProjectId, + threadId: ThreadId, + name: Schema.optional(TrimmedNonEmptyString), + prompt: TrimmedNonEmptyString, + schedule: TaskScheduleSpec, + createdAt: IsoDateTime, + updatedAt: IsoDateTime, + lastFiredAt: Schema.NullOr(IsoDateTime), + // Next due time; null once a "once" task fired or any task was cancelled. + nextFireAt: Schema.NullOr(IsoDateTime), + cancelledAt: Schema.NullOr(IsoDateTime), +}); +export type OrchestrationTask = typeof OrchestrationTask.Type; + export const OrchestrationReadModel = Schema.Struct({ snapshotSequence: NonNegativeInt, projects: Schema.Array(OrchestrationProject), threads: Schema.Array(OrchestrationThread), + // Defaulted so read models persisted/decoded before scheduled tasks shipped + // still decode as taskless. + tasks: Schema.Array(OrchestrationTask).pipe(Schema.withDecodingDefault(Effect.succeed([]))), updatedAt: IsoDateTime, }); export type OrchestrationReadModel = typeof OrchestrationReadModel.Type; @@ -910,6 +957,42 @@ const ThreadSessionStopCommand = Schema.Struct({ onlyIfSettled: Schema.optional(Schema.Boolean), }); +const TaskScheduleCommand = Schema.Struct({ + type: Schema.Literal("task.schedule"), + commandId: CommandId, + taskId: TaskId, + projectId: ProjectId, + // Tasks anchor to an existing thread: each fire appends a turn there, like + // a user message arriving on schedule. Unanchored tasks would need a + // default model-resolution policy to auto-create threads — deliberately + // out of scope for v1. + threadId: ThreadId, + name: Schema.optional(TrimmedNonEmptyString), + prompt: TrimmedNonEmptyString, + schedule: TaskScheduleSpec, + createdAt: IsoDateTime, +}); + +const TaskCancelCommand = Schema.Struct({ + type: Schema.Literal("task.cancel"), + commandId: CommandId, + taskId: TaskId, +}); + +// Server-internal only (never dispatchable by clients): the scheduler emits +// these when a task's nextFireAt passes. commandIds are deterministic +// (`server:task-fire::`), so a crash-retry collapses into the +// existing idempotent-command receipt instead of double-firing. +const TaskFireCommand = Schema.Struct({ + type: Schema.Literal("task.fire"), + commandId: CommandId, + taskId: TaskId, + // The tick time that observed the task as due. The decider rejects fires + // that run ahead of nextFireAt, so a stray or replayed command cannot make + // a task fire early. + dueAt: IsoDateTime, +}); + const DispatchableClientOrchestrationCommand = Schema.Union([ ProjectCreateCommand, ProjectMetaUpdateCommand, @@ -934,6 +1017,8 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ThreadUserInputRespondCommand, ThreadCheckpointRevertCommand, ThreadSessionStopCommand, + TaskScheduleCommand, + TaskCancelCommand, ]); export type DispatchableClientOrchestrationCommand = typeof DispatchableClientOrchestrationCommand.Type; @@ -962,6 +1047,8 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadUserInputRespondCommand, ThreadCheckpointRevertCommand, ThreadSessionStopCommand, + TaskScheduleCommand, + TaskCancelCommand, ]); export type ClientOrchestrationCommand = typeof ClientOrchestrationCommand.Type; @@ -1047,6 +1134,7 @@ const InternalOrchestrationCommand = Schema.Union([ ThreadActivityAppendCommand, ThreadRevertCompleteCommand, ThreadTitleRegenerationCompleteCommand, + TaskFireCommand, ]); export type InternalOrchestrationCommand = typeof InternalOrchestrationCommand.Type; @@ -1086,10 +1174,13 @@ export const OrchestrationEventType = Schema.Literals([ "thread.proposed-plan-upserted", "thread.turn-diff-completed", "thread.activity-appended", + "task.scheduled", + "task.fired", + "task.cancelled", ]); export type OrchestrationEventType = typeof OrchestrationEventType.Type; -export const OrchestrationAggregateKind = Schema.Literals(["project", "thread"]); +export const OrchestrationAggregateKind = Schema.Literals(["project", "thread", "task"]); export type OrchestrationAggregateKind = typeof OrchestrationAggregateKind.Type; export const OrchestrationActorKind = Schema.Literals(["client", "server", "provider"]); @@ -1233,6 +1324,45 @@ export const ThreadInteractionModeSetPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const TaskScheduledPayload = Schema.Struct({ + taskId: TaskId, + projectId: ProjectId, + threadId: ThreadId, + name: Schema.optional(TrimmedNonEmptyString), + prompt: TrimmedNonEmptyString, + schedule: TaskScheduleSpec, + // First fire time, derived from the schedule at creation. Interval tasks + // anchor here so later fires never drift. + nextFireAt: IsoDateTime, + createdAt: IsoDateTime, + updatedAt: IsoDateTime, +}); + +// Denormalized enough for the turn-starting reactor to build the fire's +// `thread.turn.start` without a read-model lookup: prompt, thread, and the +// anchor thread's modes travel with the event. +export const TaskFiredPayload = Schema.Struct({ + taskId: TaskId, + projectId: ProjectId, + threadId: ThreadId, + prompt: TrimmedNonEmptyString, + firedAt: IsoDateTime, + dueAt: IsoDateTime, + // Null once a "once" task has spent its single fire. + nextFireAt: Schema.NullOr(IsoDateTime), + runtimeMode: RuntimeMode.pipe(Schema.withDecodingDefault(Effect.succeed(DEFAULT_RUNTIME_MODE))), + interactionMode: ProviderInteractionMode.pipe( + Schema.withDecodingDefault(Effect.succeed(DEFAULT_PROVIDER_INTERACTION_MODE)), + ), + updatedAt: IsoDateTime, +}); + +export const TaskCancelledPayload = Schema.Struct({ + taskId: TaskId, + cancelledAt: IsoDateTime, + updatedAt: IsoDateTime, +}); + export const ThreadMessageSentPayload = Schema.Struct({ threadId: ThreadId, messageId: MessageId, @@ -1346,7 +1476,7 @@ const EventBaseFields = { sequence: NonNegativeInt, eventId: EventId, aggregateKind: OrchestrationAggregateKind, - aggregateId: Schema.Union([ProjectId, ThreadId]), + aggregateId: Schema.Union([ProjectId, ThreadId, TaskId]), occurredAt: IsoDateTime, commandId: Schema.NullOr(CommandId), causationEventId: Schema.NullOr(EventId), @@ -1500,6 +1630,21 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.activity-appended"), payload: ThreadActivityAppendedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("task.scheduled"), + payload: TaskScheduledPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("task.fired"), + payload: TaskFiredPayload, + }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("task.cancelled"), + payload: TaskCancelledPayload, + }), ]); export type OrchestrationEvent = typeof OrchestrationEvent.Type; @@ -1644,6 +1789,18 @@ export const OrchestrationGetWorkflowScriptResult = Schema.Struct({ }); export type OrchestrationGetWorkflowScriptResult = typeof OrchestrationGetWorkflowScriptResult.Type; +export const OrchestrationListTasksInput = Schema.Struct({ + projectId: ProjectId, +}); +export type OrchestrationListTasksInput = typeof OrchestrationListTasksInput.Type; + +// Includes cancelled tasks so clients can render (and clear) recent history +// without a second query. +export const OrchestrationListTasksResult = Schema.Struct({ + tasks: Schema.Array(OrchestrationTask), +}); +export type OrchestrationListTasksResult = typeof OrchestrationListTasksResult.Type; + const WORKFLOW_SCRIPT_ERROR_MESSAGES = { "invalid-path": "Workflow scripts must be absolute .js paths.", "root-unavailable": "Script root unavailable.", @@ -1698,6 +1855,10 @@ export const OrchestrationRpcSchemas = { input: OrchestrationSearchThreadsInput, output: OrchestrationSearchThreadsResult, }, + listTasks: { + input: OrchestrationListTasksInput, + output: OrchestrationListTasksResult, + }, getArchivedShellSnapshot: { input: Schema.Struct({}), output: OrchestrationShellSnapshot, @@ -1752,3 +1913,11 @@ export class OrchestrationSearchThreadsError extends Schema.TaggedErrorClass()( + "OrchestrationListTasksError", + { + message: TrimmedNonEmptyString, + cause: Schema.optional(Schema.Defect()), + }, +) {} diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 45bf581de084..db7f0582aa5c 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -59,8 +59,12 @@ import { OrchestrationGetFullThreadDiffError, OrchestrationGetFullThreadDiffInput, OrchestrationGetSnapshotError, + OrchestrationListTasksError, + OrchestrationListTasksInput, + OrchestrationListTasksResult, OrchestrationSearchThreadsError, OrchestrationSearchThreadsInput, + OrchestrationSearchThreadsResult, OrchestrationGetTurnDiffError, OrchestrationGetTurnDiffInput, OrchestrationRpcSchemas, @@ -917,10 +921,16 @@ export const WsOrchestrationGetFullThreadDiffRpc = Rpc.make( export const WsOrchestrationSearchThreadsRpc = Rpc.make(ORCHESTRATION_WS_METHODS.searchThreads, { payload: OrchestrationSearchThreadsInput, - success: OrchestrationRpcSchemas.searchThreads.output, + success: OrchestrationSearchThreadsResult, error: Schema.Union([OrchestrationSearchThreadsError, EnvironmentAuthorizationError]), }); +export const WsOrchestrationListTasksRpc = Rpc.make(ORCHESTRATION_WS_METHODS.listTasks, { + payload: OrchestrationListTasksInput, + success: OrchestrationListTasksResult, + error: Schema.Union([OrchestrationListTasksError, EnvironmentAuthorizationError]), +}); + export const WsOrchestrationGetArchivedShellSnapshotRpc = Rpc.make( ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, { @@ -1094,6 +1104,7 @@ export const WsRpcGroup = RpcGroup.make( WsOrchestrationGetTurnDiffRpc, WsOrchestrationGetFullThreadDiffRpc, WsOrchestrationSearchThreadsRpc, + WsOrchestrationListTasksRpc, WsOrchestrationGetArchivedShellSnapshotRpc, WsOrchestrationSubscribeShellRpc, WsOrchestrationSubscribeThreadRpc, diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index ba4facaf53ce..1d6b9293779b 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -616,6 +616,19 @@ export const ServerSettings = Schema.Struct({ * between a desktop window and a phone attached to the same server. */ enableAgentBrowserAccess: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), + /** + * Whether the server's task scheduler may start agent turns at their due + * times. Turning this off stops the scheduler from dispatching `task.fire` + * for due tasks — existing schedules stay stored and resume untouched when + * the feature is turned back on (missed fires coalesce into one, exactly + * like downtime). Clients read the same flag through `serverGetConfig` to + * hide the schedule UI everywhere at once. + * + * Server-authoritative rather than client-local: firing happens on the + * server even with no client open, so a client-side switch could not + * actually stop anything. + */ + enableScheduledTasks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), backgroundActivity: BackgroundActivitySettings, // Legacy flat fields retained for old settings files and old clients. New // consumers should resolve `backgroundActivity` instead. @@ -824,6 +837,7 @@ export const ServerSettingsPatch = Schema.Struct({ enableLegacyTokenStreaming: Schema.optionalKey(Schema.Boolean), enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean), enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean), + enableScheduledTasks: Schema.optionalKey(Schema.Boolean), backgroundActivity: Schema.optionalKey( Schema.Struct({ schemaVersion: Schema.optionalKey(Schema.Literal(1)),